fix: distinguish billing refusals across workflow calls

src/tin_lite/ai_answers_activities.py

@@ -24,7 +24,7 @@
     call_outcome,
     measure,
 )
-from tin_lite.billing_contracts import BillingError
+from tin_lite.billing_contracts import BillingError, billing_refusal
 from tin_lite.usage_capture import external_usage_scope
 
 OPERATION = "ai.answers"
@@ -94,8 +94,8 @@ async def once(self, name, *, reserve_usd: Decimal, call):
                         maximum_usd=str(reserve_usd),
                     ):
                         result = await call_outcome(call)
-                except BillingError:
-                    result = {"status": "failed", "reason": "spending_limit"}
+                except BillingError as exc:
+                    result = {"status": "failed", **billing_refusal(exc)}
                 except Exception:  # noqa: BLE001 - a paid attempt's outcome is now unknown.
                     activity.logger.warning("AI answers request outcome is unknown.")
                     result = {"status": "unknown", "reason": "provider_result_unavailable"}

src/tin_lite/billing_contracts.py

@@ -22,6 +22,12 @@ def diagnostic(self):
         return {"code": self.code, "message": str(self)}
 
 
+def billing_refusal(error: BillingError) -> dict[str, str]:
+    if error.code in {"spending_limit", "run_limit", "project_limit", "budget_exhausted"}:
+        return {"reason": "spending_limit"}
+    return {"reason": "billing_refused", "billing_error_code": error.code}
+
+
 class ProjectSpendingPolicy(BaseModel):
     model_config = ConfigDict(extra="forbid")
     per_run_nanos: Money

src/tin_lite/organic_audit.py

@@ -401,6 +401,9 @@ def grounded_preparation(policy_version: str) -> bool:
     "provider_result_unavailable": "The provider request outcome could not be confirmed.",
     "unconfirmed_previous_request": "An earlier request was not confirmed and was not repeated.",
     "spending_limit": "The spending limit prevented this request.",
+    "billing_refused": (
+        "A billing check refused this request; its saved call contains the error code."
+    ),
     "model_not_configured": "The model provider was not configured.",
     "classification_exceeded_evidence_budget": (
         "The scored result exceeded the evidence size limit."

src/tin_lite/organic_audit_activities.py

@@ -233,10 +233,10 @@ async def _paid(
                     with external_usage_scope(self.db, conn, run_id, stage, maximum_usd=amount):
                         result = {"status": "completed", "value": await call()}
                 except Exception as exc:
-                    from tin_lite.billing_contracts import BillingError
+                    from tin_lite.billing_contracts import BillingError, billing_refusal
 
                     if isinstance(exc, BillingError):
-                        result = {"status": "unavailable", "reason": "spending_limit"}
+                        result = {"status": "unavailable", **billing_refusal(exc)}
                     # Preserve only a safe product fact, not raw provider errors.
                     # Leave crawl intent incomplete so a later attempt can recover it.
                     elif recover:

src/tin_lite/paid_ads_monitor_activities.py

@@ -156,10 +156,10 @@ async def _paid(self, run_id: str, stage: str, request: dict, amount: str, call,
                         value = await call()
                     result = {"status": "completed", "value": value}
                 except Exception as exc:
-                    from tin_lite.billing_contracts import BillingError
+                    from tin_lite.billing_contracts import BillingError, billing_refusal
 
                     result = (
-                        {"status": "unavailable", "reason": "spending_limit"}
+                        {"status": "unavailable", **billing_refusal(exc)}
                         if isinstance(exc, BillingError)
                         else {"status": "unknown", "reason": "provider_result_unavailable"}
                     )
@@ -211,9 +211,26 @@ async def call():
             needs="model",
         )
         if result["status"] != "completed":
-            await self._save(run_id, "failure", {"code": "model_unavailable", "stage": step})
+            reason = result.get("reason")
+            code = (
+                reason if reason in {"billing_refused", "spending_limit"} else "model_unavailable"
+            )
+            await self._save(
+                run_id,
+                "failure",
+                {
+                    "code": code,
+                    "stage": step,
+                    "billing_error_code": result.get("billing_error_code"),
+                },
+            )
+            message = {
+                "billing_refused": "A billing check refused a model call",
+                "spending_limit": "The spending limit stopped a model call",
+                "model_unavailable": "A model result was unavailable",
+            }[code]
             raise ApplicationError(
-                "A model result was unavailable; no replacement was purchased.",
+                f"{message}; no replacement was purchased.",
                 non_retryable=True,
             )
         if result["value"].get("unusable"):
@@ -392,6 +409,23 @@ async def read(self, run_id: str) -> None:
                 gaql = builder()
             result = await self._ads(run_id, f"read:{name}", "search", {"query": gaql})
             if result.get("status") != "completed" or result.get("error"):
+                if name == "campaign_7d" and result.get("reason") in {
+                    "billing_refused",
+                    "spending_limit",
+                }:
+                    await self._save(
+                        run_id,
+                        "failure",
+                        {
+                            "code": result["reason"],
+                            "stage": f"read:{name}",
+                            "billing_error_code": result.get("billing_error_code"),
+                        },
+                    )
+                    raise ApplicationError(
+                        "A billing check refused the required campaign read. Nothing was changed.",
+                        non_retryable=True,
+                    )
                 unavailable.append(name)
                 raw[name] = []
             else:
@@ -665,6 +699,15 @@ async def failure(self, run_id: str) -> None:
             return
         failure = await self._result(run_id, "failure") or {}
         message = {
+            "billing_refused": (
+                "The Google Ads check stopped because a billing check refused a request "
+                f"({failure.get('billing_error_code')}). Automatic changes already applied stand; "
+                "nothing else was changed."
+            ),
+            "spending_limit": (
+                "The Google Ads check stopped because a request reached the spending limit. "
+                "Automatic changes already applied stand; nothing else was changed."
+            ),
             "model_unavailable": (
                 "The Google Ads check stopped because a model result could not be confirmed. "
                 "Automatic changes already applied stand; nothing else was changed."

src/tin_lite/payment_recovery_activities.py

@@ -581,8 +581,8 @@ async def _model(self, run_id: str, user: str):
                             conn, execution_key=key, result={"parsed": None, "reason": exc.code}
                         )
                         raise ApplicationError(
-                            "Tin credits or project spending limits do not allow drafting the "
-                            "emails; nothing was sent.",
+                            f"A billing check refused email drafting ({exc.code}); "
+                            "nothing was sent.",
                             non_retryable=True,
                         ) from None
                     result = {"parsed": None, "reason": "provider_result_unavailable"}

tests/test_workflow_billing_refusals.py

@@ -0,0 +1,121 @@
+"""Paid refusals keep their cause without buying the same request again."""
+
+import json
+from decimal import Decimal
+from unittest.mock import AsyncMock
+
+import pytest
+from temporalio.exceptions import ApplicationError
+from test_ai_answers_activity import FakeRunDatabase
+from test_organic_audit import activities_fixture
+from test_paid_ads_monitor_activities import fixture as monitor_fixture
+from test_payment_recovery import setup as payment_fixture
+
+from tin_lite.ai_answers_activities import ReceiptLedger
+from tin_lite.billing_contracts import BillingError
+from tin_lite.organic_audit_ai import AuditValidationError
+
+
+@pytest.mark.parametrize("workflow", ["audit", "monitor", "ai_answers"])
+@pytest.mark.parametrize(
+    "code",
+    [
+        "spending_limit",
+        "run_limit",
+        "project_limit",
+        "budget_exhausted",
+        "unpriced_model",
+        "unmetered_operation",
+        "operation_already_attempted",
+        "usage_pending",
+        "insufficient_funds",
+        "spending_stopped",
+    ],
+)
+async def test_refusal_classification_and_replay(workflow, code):
+    call = AsyncMock(side_effect=BillingError(code, "Private diagnostic must not escape"))
+    if workflow == "audit":
+        work, db, *_ = await activities_fixture()
+
+        async def invoke():
+            return await work._paid(str(db.run.id), "answer:test", {}, "0.01", call)
+    elif workflow == "monitor":
+        work, db, *_ = monitor_fixture()
+        await work.prepare(str(db.run.id))
+
+        async def invoke():
+            return await work._paid(str(db.run.id), "model:test", {}, "0.01", call, needs="model")
+    else:
+        db = FakeRunDatabase()
+        work = ReceiptLedger(db, run_id=db.run.id, stage="test")
+
+        async def invoke():
+            return await work.once("request", reserve_usd=Decimal("0.01"), call=call)
+
+    result = await invoke()
+    assert await invoke() == result
+    assert call.await_count == 1
+    is_limit = code in {"spending_limit", "run_limit", "project_limit", "budget_exhausted"}
+    assert result["reason"] == ("spending_limit" if is_limit else "billing_refused")
+    if not is_limit:
+        assert result["billing_error_code"] == code
+    assert "Private diagnostic" not in json.dumps(result)
+
+
+@pytest.mark.parametrize("code", ["run_limit", "project_limit", "unpriced_model", "usage_pending"])
+async def test_monitor_projected_failure_keeps_the_classification(code):
+    work, db, *_ = monitor_fixture()
+    run_id = str(db.run.id)
+    await work.prepare(run_id)
+    work.router.generate.side_effect = BillingError(code, "Private diagnostic must not escape")
+    for _ in range(2):
+        with pytest.raises(ApplicationError):
+            await work._model(run_id, "brief", "system", "user", {}, 100, "low")
+    assert work.router.generate.await_count == 1
+    await work.failure(run_id)
+    message = db.project_failure.await_args.kwargs["error_message"]
+    assert ("spending limit" in message) is (code in {"run_limit", "project_limit"})
+    if code not in {"run_limit", "project_limit"}:
+        assert code in message and "could not be confirmed" not in message
+    assert "Private diagnostic" not in message
+
+
+def test_audit_billing_gap_has_an_honest_explanation():
+    gap = AuditValidationError("billing_refused")
+    assert "billing check refused" in str(gap) and "spending limit" not in str(gap)
+
+
+@pytest.mark.parametrize("code", ["spending_stopped", "unmetered_operation", "run_limit"])
+async def test_required_ads_read_projects_billing_cause_without_retrying(code):
+    work, db, _, integrations, _ = monitor_fixture()
+    run_id = str(db.run.id)
+    await work.prepare(run_id)
+    integrations.google_ads_call = AsyncMock(side_effect=BillingError(code, "Private diagnostic"))
+    for _ in range(2):
+        with pytest.raises(ApplicationError) as caught:
+            await work.read(run_id)
+        assert caught.value.non_retryable is True
+    assert integrations.google_ads_call.await_count == 1
+    assert integrations.mutations == []
+    assert work.router.generate.await_count == 0
+    await work.failure(run_id)
+    message = db.project_failure.await_args.kwargs["error_message"]
+    assert ("spending limit" in message) is (code == "run_limit")
+    if code != "run_limit":
+        assert code in message
+    assert "model call" not in message
+    assert "Private diagnostic" not in message
+
+
+@pytest.mark.parametrize(
+    "code", ["unpriced_model", "unmetered_operation", "operation_already_attempted"]
+)
+async def test_payment_drafting_refusal_does_not_claim_a_credit_limit(monkeypatch, code):
+    fixture = payment_fixture(monkeypatch)
+    fixture.router.generate = AsyncMock(side_effect=BillingError(code, "Private diagnostic"))
+    with pytest.raises(ApplicationError) as caught:
+        await fixture.activities._model(fixture.run_id, "draft request")
+    assert code in str(caught.value) and "spending limits" not in str(caught.value)
+    assert "Private diagnostic" not in str(caught.value)
+    assert await fixture.activities._model(fixture.run_id, "draft request") is None
+    assert fixture.router.generate.await_count == 1