fix: show founders what they can fix and keep Tin's internals for operators

AGENTS.md

@@ -145,6 +145,13 @@ outside marketing, or were written without ever being run. Full rules:
 - Limits are runaway guards set well above normal use, not estimates that stop ordinary runs.
 - A failure names its cause. Never wrap Tin's own refusal in a generic message; keep the
   reason on the run and its receipt, bounded and without provider text or secrets.
+- Show the founder a cause they can act on, not Tin's internals. `failures.py` sorts every
+  failure: needs you (reconnect, credits or limits, input), temporary, Tin's own sentence,
+  or internal (a plain line for the founder, the raw text in `error_detail` for operators).
+  Raise founder-fixable failures as their own types (`IntegrationAuthorizationError`,
+  `WorkflowInputError`, `BillingError`) and use `billing_refusal` for billing wording; not
+  every billing refusal is a spending limit. Failure activities take `failure_input` and
+  project with `failure_projection`; HTTP and MCP bodies use `public_error_text`.
 - Fixture tests prove the contract; a real run proves the workflow. Before calling a workflow
   change done, run it once for real where a provider, sandbox or outside site is involved.
 

migrations/063_run_failure_kind.sql

@@ -0,0 +1,7 @@
+-- What kind of failure stopped a run, kept apart from its founder-safe error_message.
+-- error_detail is the raw, scrubbed cause for operators; it is never returned by the API or MCP.
+ALTER TABLE workflow_runs
+    ADD COLUMN error_kind text
+        CHECK (error_kind IN ('needs_you', 'transient', 'explained', 'internal')),
+    ADD COLUMN error_action text CHECK (length(error_action) <= 40),
+    ADD COLUMN error_detail text CHECK (length(error_detail) <= 2000);

src/tin_lite/activities.py

@@ -75,6 +75,14 @@
     parse_selected_shortlist,
     render_email_template,
 )
+from tin_lite.failures import (
+    billing_refusal,
+    classify,
+    failure_payload,
+    payload_failure,
+    receipt_failure,
+)
+from tin_lite.failures import detail as failure_detail
 from tin_lite.integrations import (
     GOOGLE_WORKSPACE_PROVIDER,
     GitHubFileChange,
@@ -191,32 +199,60 @@ def restarted_recently() -> bool:
     return process_uptime_seconds() < RESTART_WINDOW_SECONDS
 
 
-def failure_message(
+def failure_record(
     payload: dict | None,
     exc: BaseException | None = None,
     *,
     default: str = "workflow failed",
     restarted: bool | None = None,
-) -> str:
-    """The failure text a run row stores: the first reason the payload or exception carries."""
-    message = None
-    if isinstance(payload, dict):
+) -> dict[str, str | None]:
+    """What a run row stores for a failure: founder-safe message, kind, action and operator
+    detail, as `project_failure` keyword arguments.
+
+    A payload from `workflows.failure_fields` is already classified. An older payload's
+    first reason key is kept as it was. An exception is classified here (failures.py), so
+    library and driver text stays in the detail.
+    """
+    message = kind = action = detail = None
+    carried = payload_failure(payload)
+    if carried is not None:
+        message, kind, action = carried.message, carried.kind, carried.action
+        raw = payload.get("detail") if isinstance(payload, dict) else None
+        detail = raw if isinstance(raw, str) else None
+    elif isinstance(payload, dict):
         for key in _FAILURE_MESSAGE_KEYS:
             value = payload.get(key)
             if isinstance(value, str) and value.strip():
                 message = value.strip()
                 break
     if message is None and exc is not None:
-        text = str(exc).strip()
-        message = f"{type(exc).__name__}: {text}" if text else type(exc).__name__
+        failure = classify(exc)
+        message, kind, action = failure.message, failure.kind, failure.action
+        detail = failure_detail(exc)
     if message is None:
         message = default
     message = scrub_secrets(message)
     if restarted is None:
         restarted = restarted_recently()
     if restarted and not message.endswith(RESTART_SUFFIX):
         message = message[: FAILURE_MESSAGE_LIMIT - len(RESTART_SUFFIX)] + RESTART_SUFFIX
-    return message[:FAILURE_MESSAGE_LIMIT]
+    return {
+        "error_message": message[:FAILURE_MESSAGE_LIMIT],
+        "error_kind": kind,
+        "error_action": action,
+        "error_detail": scrub_secrets(detail) if detail else None,
+    }
+
+
+def failure_message(
+    payload: dict | None,
+    exc: BaseException | None = None,
+    *,
+    default: str = "workflow failed",
+    restarted: bool | None = None,
+) -> str:
+    """The founder-safe failure text a run row stores."""
+    return failure_record(payload, exc, default=default, restarted=restarted)["error_message"]
 
 
 def transient_failure(message: str | None, *, restarted: bool = False) -> bool:
@@ -406,7 +442,9 @@ async def dispatch_scheduled_workflow(self, payload: dict[str, str]) -> dict[str
                 )
                 return {}
             except (ValueError, KeyError, IntegrationError) as exc:
-                await pause_for_issue(self, configured, str(exc))
+                from tin_lite.code_schedules import schedule_issue
+
+                await pause_for_issue(self, configured, schedule_issue(exc))
                 return {}
         try:
             run, created = await self._db.create_run(
@@ -425,9 +463,17 @@ async def dispatch_scheduled_workflow(self, payload: dict[str, str]) -> dict[str
                 **selection,
             )
         except BillingError as exc:
-            from tin_lite.code_schedules import pause_for_issue
+            from tin_lite.code_schedules import pause_for_issue, schedule_issue
 
-            await pause_for_issue(self, configured, str(exc))
+            if billing_refusal(exc).kind == "transient":
+                # An earlier paid step is still settling: skip this occurrence only.
+                await self._db.advance_project_workflow_schedule(
+                    project_workflow_id=configured.id,
+                    next_run_at=next_run_after(schedule, scheduled_for),
+                    expected_settings_revision=configured.settings_revision,
+                )
+                return {}
+            await pause_for_issue(self, configured, schedule_issue(exc))
             return {}
         except ScheduledWorkflowSkip:
             await self._db.advance_project_workflow_schedule(
@@ -905,11 +951,21 @@ async def _fail_run(
     ) -> str:
         """Store the failure with its reason, then start one retry if the cause was transient."""
         restarted = restarted_recently()
-        if message is None:
-            message = failure_message(payload, exc, default=default, restarted=restarted)
-        await self._db.project_failure(run_id=run_id, error_message=message)
-        await self._auto_retry_after_transient_failure(run_id, message=message, restarted=restarted)
-        return message
+        record = (
+            failure_record(payload, exc, default=default, restarted=restarted)
+            if message is None
+            else {"error_message": message}
+        )
+        await self._db.project_failure(run_id=run_id, **record)
+        # The raw detail still decides a silent retry (a 502 or a reset connection), exactly
+        # as the message did before failures were classified; the kind does not widen it.
+        detail = record.get("error_detail")
+        await self._auto_retry_after_transient_failure(
+            run_id,
+            message=record["error_message"] + (f" {detail}" if detail else ""),
+            restarted=restarted,
+        )
+        return record["error_message"]
 
     async def _auto_retry_eligible(self, run: WorkflowRun) -> bool:
         """Only unattended starts get a silent second attempt: schedules and onboarding setup."""
@@ -4673,20 +4729,37 @@ async def project_codex_procedure_result(self, run_id_text: str) -> None:
 
     @activity.defn(name="project_codex_procedure_failure")
     async def project_codex_procedure_failure(self, payload: dict[str, str]) -> None:
+        await self.project_procedure_failure(payload)
+
+    async def project_procedure_failure(
+        self, payload: dict[str, str], *, prefer_payload: bool = False
+    ) -> None:
+        """`prefer_payload`: a code workflow already chose Tin's sentence for its failed
+        receipt; otherwise the first failed procedure receipt names the cause."""
         run_id = UUID(payload["run_id"])
         run = await self._db.get_run(run_id)
-        reason = failure_message(payload, default="Codex procedure failed", restarted=False)
+        failed_payload = payload
         for step in (
-            "procedure_sandbox_create",
-            "procedure_artifact_persist",
-            "procedure_canonical_commit",
-            "procedure_projection",
+            ()
+            if prefer_payload
+            else (
+                "procedure_sandbox_create",
+                "procedure_artifact_persist",
+                "procedure_canonical_commit",
+                "procedure_projection",
+            )
         ):
             receipt = await self._db.get_effect(f"{run_id}:{step}")
             if receipt is not None and receipt.status == "failed" and receipt.error_message:
-                reason = receipt.error_message
+                # The receipt keeps the raw text for operators; the founder reads Tin's words.
+                stored = receipt_failure(receipt.error_message)
+                if stored is not None:
+                    failed_payload = {
+                        **failure_payload(stored, receipt.error_message),
+                        "run_id": str(run_id),
+                    }
                 break
-        reason = await self._fail_run(run_id, {"reason": reason})
+        reason = await self._fail_run(run_id, failed_payload, default="Codex procedure failed")
         await self._db.finalize_pending_test_identity(
             run_id=run_id,
             status="failed",
@@ -5904,7 +5977,8 @@ def _require_active_lease(run) -> None:
 
 
 def _safe_failure(exc: BaseException) -> str:
-    """Name the exception with its text, so a failed receipt says what actually went wrong."""
+    """Name the exception with its text, so a failed receipt says what actually went wrong.
+    Receipts are operator records; founder text comes from failures.receipt_failure."""
     text = scrub_secrets(" ".join(str(exc).split()))
     return f"{type(exc).__name__}: {text or 'operation failed'}"[:FAILURE_MESSAGE_LIMIT]
 

src/tin_lite/ai_answers_activities.py

@@ -25,6 +25,7 @@
     measure,
 )
 from tin_lite.billing_contracts import BillingError
+from tin_lite.failures import is_spending_limit
 from tin_lite.usage_capture import external_usage_scope
 
 OPERATION = "ai.answers"
@@ -94,8 +95,14 @@ 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:
+                    # Refused before the request was bought: a project or run limit reads as
+                    # the spending limit; any other refusal keeps its code, not that claim.
+                    result = {
+                        "status": "failed",
+                        "reason": "spending_limit" if is_spending_limit(exc) else "billing_refused",
+                        "billing_code": exc.code,
+                    }
                 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"}