AFFAANhandClaude Opus 5
6c39d42874
Adds the insights reader, a pure guardrail evaluator, and the sweep that
applies them to live campaigns.
Three things carry the weight here:
Units. Meta reports spend in major units ("12.34") while budgets and
guardrails are in minor units (1234). normalise_row converts, rounding
half up rather than using round(), which rounds halves to even and can
record an exact half-unit of spend as nothing.
Leads. Meta has no leads field; leads live in the actions array, under
several action types depending on whether the lead came from a Facebook
form or a pixel. Cost per lead is computed from spend and leads over the
same window rather than read from cost_per_action_type, so the two can
never disagree.
Order. The campaign is paused on Meta before the local record changes. A
campaign recorded as paused but still delivering is the outcome this
exists to prevent. One campaign's failure never aborts the sweep, so a
rate limit on the third does not leave the fourth unguarded.
Cost rules are skipped until the first lead or click: no leads yet is not
an infinite cost per lead, and pausing for that would kill every campaign
in its first hour.
Deliberately a deterministic loop rather than a maskanx_cron_jobs entry.
That scheduler runs prompts through an agent, and asking a language model
whether a budget has been exceeded would make an arithmetic guarantee
probabilistic. Follows reconcile.py's lifespan-task pattern instead, and
warns every cycle if Meta is unconfigured — a safety system that cannot
run should be loud.
require_approval_for_budget_increase is enforced at the API layer, where
the budget is actually edited: raising it while a campaign awaits approval
returns 409, since it would change what the approver is reviewing.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
279 lines
8.9 KiB
Python
279 lines
8.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""The enforcement sweep: fetching spend, deciding, and acting on Meta.
|
|
|
|
The properties worth protecting are the failure modes, not the happy path:
|
|
Meta is paused before the local record changes, one campaign's failure does
|
|
not leave the others unguarded, and nothing that is not live is touched.
|
|
"""
|
|
import pytest
|
|
|
|
from adclaw.campaigns.enforce import (
|
|
DEFAULT_INTERVAL_SECONDS,
|
|
enforce_once,
|
|
guardrail_interval_seconds,
|
|
observed_spend,
|
|
)
|
|
from adclaw.campaigns.models import CampaignSpec
|
|
from adclaw.meta.client import MetaError
|
|
|
|
|
|
class FakeRepo:
|
|
def __init__(self, campaigns=()):
|
|
self.items = list(campaigns)
|
|
self.events: list[dict] = []
|
|
# Records the campaign status at the moment each event was written,
|
|
# so a test can prove Meta was called first.
|
|
self.saved_status: list[str] = []
|
|
|
|
async def list_campaigns(self, company_id=None, status=None):
|
|
if status is None:
|
|
return list(self.items)
|
|
return [c for c in self.items if c.status == status]
|
|
|
|
async def update_campaign_with_event(
|
|
self, spec, event_type, actor=None, reason=None, payload=None,
|
|
):
|
|
self.events.append({
|
|
"event_type": event_type,
|
|
"actor": actor,
|
|
"reason": reason,
|
|
"payload": payload,
|
|
})
|
|
self.saved_status.append(spec.status)
|
|
return spec
|
|
|
|
|
|
class FakeMeta:
|
|
def __init__(self, today=None, lifetime=None):
|
|
self.today = today if today is not None else []
|
|
self.lifetime = lifetime if lifetime is not None else []
|
|
self.status_calls: list[tuple[str, str]] = []
|
|
self.insight_calls: list[tuple[str, str]] = []
|
|
self.fail_insights_for: set[str] = set()
|
|
|
|
async def get_insights(self, object_id, *, date_preset=None, **kwargs):
|
|
if object_id in self.fail_insights_for:
|
|
raise MetaError("Rate limited", code=17)
|
|
self.insight_calls.append((object_id, date_preset))
|
|
return self.today if date_preset == "today" else self.lifetime
|
|
|
|
async def update_object_status(self, object_id, status):
|
|
self.status_calls.append((object_id, status))
|
|
|
|
|
|
def _campaign(**overrides) -> CampaignSpec:
|
|
data = {
|
|
"id": "camp_1",
|
|
"name": "Q3 lead gen",
|
|
"status": "live",
|
|
"ad_account_id": "act_1",
|
|
"meta_campaign_id": "meta_camp_1",
|
|
"budget": {"daily_budget": 50000},
|
|
"guardrails": {
|
|
"auto_pause_if_spend_reaches": 45000,
|
|
"stop_loss_enabled": True,
|
|
},
|
|
}
|
|
data.update(overrides)
|
|
return CampaignSpec(**data)
|
|
|
|
|
|
# --- interval ---
|
|
|
|
|
|
def test_interval_defaults_to_fifteen_minutes(monkeypatch):
|
|
monkeypatch.delenv("MASKANX_GUARDRAIL_INTERVAL_SECONDS", raising=False)
|
|
|
|
assert guardrail_interval_seconds() == DEFAULT_INTERVAL_SECONDS
|
|
|
|
|
|
def test_a_nonsense_interval_falls_back_to_the_default(monkeypatch):
|
|
"""A typo must not silently disable a safety system."""
|
|
monkeypatch.setenv("MASKANX_GUARDRAIL_INTERVAL_SECONDS", "every 5 mins")
|
|
|
|
assert guardrail_interval_seconds() == DEFAULT_INTERVAL_SECONDS
|
|
|
|
|
|
def test_zero_disables_enforcement(monkeypatch):
|
|
monkeypatch.setenv("MASKANX_GUARDRAIL_INTERVAL_SECONDS", "0")
|
|
|
|
assert guardrail_interval_seconds() == 0
|
|
|
|
|
|
# --- reading spend ---
|
|
|
|
|
|
async def test_observed_spend_reads_today_and_lifetime_separately():
|
|
meta = FakeMeta(
|
|
today=[{"spend": "460.00", "clicks": "20"}],
|
|
lifetime=[{"spend": "5000.00"}],
|
|
)
|
|
|
|
spend = await observed_spend(meta, _campaign())
|
|
|
|
assert spend["today_spend"] == 46000
|
|
assert spend["lifetime_spend"] == 500000
|
|
assert meta.insight_calls == [
|
|
("meta_camp_1", "today"),
|
|
("meta_camp_1", "maximum"),
|
|
]
|
|
|
|
|
|
async def test_a_campaign_with_no_delivery_reads_as_zero():
|
|
spend = await observed_spend(FakeMeta(), _campaign())
|
|
|
|
assert spend["today_spend"] == 0
|
|
assert spend["lifetime_spend"] == 0
|
|
assert spend["cost_per_lead"] is None
|
|
|
|
|
|
# --- the sweep ---
|
|
|
|
|
|
async def test_a_campaign_over_its_daily_limit_is_paused_on_meta():
|
|
repo = FakeRepo([_campaign()])
|
|
meta = FakeMeta(today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}])
|
|
|
|
result = await enforce_once(repo, meta)
|
|
|
|
assert result.paused == ["camp_1"]
|
|
assert meta.status_calls == [("meta_camp_1", "PAUSED")]
|
|
assert repo.items[0].status == "paused"
|
|
|
|
|
|
async def test_the_event_names_the_rule_and_the_observed_value():
|
|
"""A campaign that stopped overnight has to be explainable."""
|
|
repo = FakeRepo([_campaign()])
|
|
meta = FakeMeta(today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}])
|
|
|
|
await enforce_once(repo, meta)
|
|
|
|
event = repo.events[-1]
|
|
assert event["event_type"] == "campaign.guardrail_pause"
|
|
assert event["actor"] == "guardrail"
|
|
assert event["payload"]["rule"] == "auto_pause_if_spend_reaches"
|
|
assert event["payload"]["observed"] == 46000
|
|
assert event["payload"]["limit"] == 45000
|
|
|
|
|
|
async def test_meta_is_paused_before_the_local_record_is_written():
|
|
"""Recorded-as-paused-but-still-delivering is the outcome to avoid."""
|
|
repo = FakeRepo([_campaign()])
|
|
meta = FakeMeta(today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}])
|
|
|
|
class OrderCheckingMeta(FakeMeta):
|
|
async def update_object_status(self, object_id, status):
|
|
# No event may have been written yet.
|
|
assert repo.events == []
|
|
await FakeMeta.update_object_status(self, object_id, status)
|
|
|
|
meta = OrderCheckingMeta(
|
|
today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}],
|
|
)
|
|
|
|
await enforce_once(repo, meta)
|
|
|
|
assert meta.status_calls == [("meta_camp_1", "PAUSED")]
|
|
assert repo.events
|
|
|
|
|
|
async def test_a_campaign_within_its_limits_is_left_alone():
|
|
repo = FakeRepo([_campaign()])
|
|
meta = FakeMeta(today=[{"spend": "100.00"}], lifetime=[{"spend": "100.00"}])
|
|
|
|
result = await enforce_once(repo, meta)
|
|
|
|
assert result.checked == 1
|
|
assert not result.acted
|
|
assert meta.status_calls == []
|
|
assert repo.events == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"status", ["draft", "pending_approval", "approved", "synced", "paused", "stopped"],
|
|
)
|
|
async def test_only_live_campaigns_are_checked(status):
|
|
"""Nothing else can be spending, and acting would fight the operator."""
|
|
repo = FakeRepo([_campaign(status=status)])
|
|
meta = FakeMeta(today=[{"spend": "9999.00"}], lifetime=[{"spend": "9999.00"}])
|
|
|
|
result = await enforce_once(repo, meta)
|
|
|
|
assert result.checked == 0
|
|
assert meta.status_calls == []
|
|
|
|
|
|
async def test_a_live_campaign_without_a_meta_object_is_skipped():
|
|
repo = FakeRepo([_campaign(meta_campaign_id=None)])
|
|
meta = FakeMeta(today=[{"spend": "9999.00"}])
|
|
|
|
result = await enforce_once(repo, meta)
|
|
|
|
assert result.checked == 0
|
|
assert meta.status_calls == []
|
|
|
|
|
|
async def test_one_campaign_failing_does_not_leave_the_rest_unguarded():
|
|
"""A rate limit on the first must not skip the second."""
|
|
repo = FakeRepo([
|
|
_campaign(id="camp_1", meta_campaign_id="meta_1"),
|
|
_campaign(id="camp_2", meta_campaign_id="meta_2"),
|
|
])
|
|
meta = FakeMeta(today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}])
|
|
meta.fail_insights_for = {"meta_1"}
|
|
|
|
result = await enforce_once(repo, meta)
|
|
|
|
assert result.failed == ["camp_1"]
|
|
assert result.paused == ["camp_2"]
|
|
assert ("meta_2", "PAUSED") in meta.status_calls
|
|
|
|
|
|
async def test_a_failed_pause_on_meta_leaves_the_campaign_live():
|
|
"""Better to keep showing it as live than to lose track of real spend."""
|
|
repo = FakeRepo([_campaign()])
|
|
|
|
class RefusingMeta(FakeMeta):
|
|
async def update_object_status(self, object_id, status):
|
|
raise MetaError("Permission denied", code=200)
|
|
|
|
meta = RefusingMeta(
|
|
today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}],
|
|
)
|
|
|
|
result = await enforce_once(repo, meta)
|
|
|
|
assert result.failed == ["camp_1"]
|
|
assert not result.acted
|
|
assert repo.items[0].status == "live"
|
|
assert repo.events == []
|
|
|
|
|
|
async def test_a_lifetime_breach_stops_the_campaign():
|
|
repo = FakeRepo([
|
|
_campaign(
|
|
guardrails={"max_campaign_spend": 500000, "stop_loss_enabled": True},
|
|
),
|
|
])
|
|
meta = FakeMeta(today=[{"spend": "10.00"}], lifetime=[{"spend": "6000.00"}])
|
|
|
|
result = await enforce_once(repo, meta)
|
|
|
|
assert result.stopped == ["camp_1"]
|
|
assert repo.items[0].status == "stopped"
|
|
assert repo.events[-1]["event_type"] == "campaign.guardrail_stop"
|
|
# Stopping still pauses on Meta: that is what actually halts delivery.
|
|
assert meta.status_calls == [("meta_camp_1", "PAUSED")]
|
|
|
|
|
|
async def test_the_trip_is_recorded_on_the_campaign_for_the_ui():
|
|
repo = FakeRepo([_campaign()])
|
|
meta = FakeMeta(today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}])
|
|
|
|
await enforce_once(repo, meta)
|
|
|
|
trip = repo.items[0].advanced["last_guardrail_trip"]
|
|
assert trip["rule"] == "auto_pause_if_spend_reaches"
|
|
assert trip["observed"] == 46000
|
|
assert trip["at"]
|