[eric] agents: extraction-shaped handoffs and relayed refusals never reach the subscription lane (ENG-387)

backend/apps/agents/core/error_classify.py

@@ -74,6 +74,42 @@ def is_content_policy_block(text: str) -> bool:
     return bool(P_CONTENT_POLICY_BLOCK.search(text))
 
 
+# Asking one agent to reproduce another's work verbatim is the shape the subscription lane refuses ("duplicating model outputs"). Our own handoff prompts are model-authored and land in a forked child as its user turn, which is where a third of one user's blocks came from; one agent even diagnosed itself: "my phrasing about 'dumping verbatim' tripped it".
+P_EXTRACTION_ASK = re.compile(
+    r"\bverbatim\b|\bword[- ]for[- ]word\b|complete\s+dump|\bdump\s+(?:of|everything|all|your)|"
+    r"(?:exact|full|entire|complete)\s+(?:\w+\s+){0,2}(?:text|body|output|response|reply|contents?|transcript|work)|"
+    r"repeat\s+(?:back\s+)?(?:what|your|the)|reproduce\s+(?:the|your|it)",
+    re.IGNORECASE,
+)
+
+
+@typechecked
+def defuse_extraction_ask(text: str) -> str:
+    """Rewrite a handoff prompt that asks another agent to reproduce output. The delegation prompt is
+    written by the model at call time, so the only place this can be made unrepresentable is the
+    dispatch boundary: what leaves here can never carry the shape."""
+    if not text or not P_EXTRACTION_ASK.search(text):
+        return text
+    return (text + "\n\nAnswer in your own words as a short task-relevant summary. Do not reproduce "
+            "any earlier message, output, or file contents verbatim.")
+
+
+@typechecked
+def neutralize_provider_refusal(text: str) -> str:
+    """A delegated agent's provider refusal must never travel back as CONTENT. Measured 2026-08-21:
+    a blocked child's refusal came home as its result and was then stored as the PARENT's own
+    assistant text (264 times across 161 chats), so the parent chat carried policy-violation
+    language as the model's own words into every later request. Returns a short neutral status
+    instead, and leaves any real answer untouched."""
+    if not text:
+        return text
+    if is_content_policy_block(text) or "unable to respond to this request" in text.lower():
+        return ("That agent could not answer this request and returned no usable result. "
+                "Do not repeat or quote its response; continue with what you already have, "
+                "or do the work directly.")
+    return text
+
+
 # Real account STATES a retry cannot fix: the subscription is gone, not the token. These must keep dying to the banner, or a canceled account silently burns a request per turn forever.
 P_SUBSCRIPTION_STATE_PATTERNS = re.compile(
     r"(?:no\s+active\s+subscription"
@@ -424,20 +460,26 @@ def is_out_of_tokens(exc: BaseException, extra_text: str = "") -> bool:
     ))
 
 
+# The CLI says this in its own words when it gives up after 3 refill cycles. It arrives two ways: a bare exit-1 ProcessError, and (on the persistent client) inside a ResultMessage that TurnRunner raises as a TurnResultError, which the exception-type gate below silently missed: measured on Alex's install 2026-08-21, 13 thrash deaths in under two hours all landed in the catch-all while the valve fired 4 times in ten days.
+P_AUTOCOMPACT_THRASH = re.compile(r"autocompact\s+is\s+thrashing|context\s+refilled\s+to\s+the\s+limit", re.IGNORECASE)
+
+
 @typechecked
 def is_context_pressure_death(exc: BaseException, compact_boundaries: int, extra_text: str = "") -> bool:
     """The CLI autocompact-thrash class: the process compacted during this turn and then
-    died with a bare exit-1 ProcessError (its thrash detector gives up after 3 refill
+    died, either with a bare exit-1 ProcessError (its thrash detector gives up after 3 refill
     cycles, which can straddle turns on a persistent client, so one boundary in the dying
-    turn is the reliable tell). Only claims deaths no other classifier owns, so auth/
-    capacity/credit errors keep their specific handling; a misfire costs one bounded
-    silent retry, a miss just means today's error card.
+    turn is the reliable tell) or with its own thrash verdict in the result text. Only claims
+    deaths no other classifier owns, so auth/capacity/credit errors keep their specific
+    handling; a misfire costs one bounded silent retry, a miss just means today's error card.
     """
-    if compact_boundaries < 1:
-        return False
-    # Type-name check, not isinstance: the SDK is lazy-imported (mock mode must work without it), mirroring the client-pool dead-client idiom.
-    if "ProcessError" not in type(exc).__name__:
-        return False
+    # The CLI naming its own thrash is self-identifying, so it needs no exception type and no boundary count.
+    if not P_AUTOCOMPACT_THRASH.search(f"{exc!s}\n{extra_text}"):
+        if compact_boundaries < 1:
+            return False
+        # Type-name check, not isinstance: the SDK is lazy-imported (mock mode must work without it), mirroring the client-pool dead-client idiom.
+        if "ProcessError" not in type(exc).__name__:
+            return False
     for p_claimed_by in (
         is_long_context_error, is_transient_capacity_error, is_free_trial_exhausted,
         is_out_of_tokens, is_auth_error, is_unknown_model_error,

backend/apps/agents/manager/AgentLaunch.py

@@ -266,9 +266,11 @@ class AgentLaunch(AgentManagerProtocol):
                     last_assistant = str(content)
                 break
 
+        # One door for every delegation caller: a child's provider refusal never travels home as content.
+        from backend.apps.agents.core.error_classify import neutralize_provider_refusal
         return {
             "forked_session_id": fork.id,
             "source_name": source_name,
-            "response": last_assistant or "No response from invoked agent.",
+            "response": neutralize_provider_refusal(last_assistant or "") or "No response from invoked agent.",
             "cost_usd": fork.cost_usd,
         }

backend/main.py

@@ -936,6 +936,9 @@ async def spawn_agent_run(request: Request):
         return JSONResponse({"error": "prompt is required"}, status_code=400)
     if not parent_session_id:
         return JSONResponse({"error": "parent_session_id is required"}, status_code=400)
+    # SpawnAgent is a SEPARATE route from InvokeAgent, so the same seal has to stand at both doors or the shape just walks through this one (ENG-387).
+    from backend.apps.agents.core.error_classify import defuse_extraction_ask
+    prompt = defuse_extraction_ask(prompt)
 
     try:
         from backend.apps.agents.agent_manager import agent_manager
@@ -1047,6 +1050,9 @@ async def invoke_agent_run(request: Request):
         return JSONResponse({"error": "session_id is required"}, status_code=400)
     if not message:
         return JSONResponse({"error": "message is required"}, status_code=400)
+    # The handoff prompt is model-authored and lands in the forked child as its user turn; an extraction-shaped one is what the subscription lane refuses (ENG-387). Defused at the dispatch boundary so the shape cannot leave here.
+    from backend.apps.agents.core.error_classify import defuse_extraction_ask
+    message = defuse_extraction_ask(message)
 
     try:
         from backend.apps.agents.agent_manager import agent_manager

backend/tests/test_delegation_refusal_hygiene.py

@@ -0,0 +1,71 @@
+"""The subscription lane refuses requests shaped like "reproduce your output" and, once one agent is
+refused, its refusal used to travel home as CONTENT and poison the parent (264 stores across 161
+chats, ENG-387). Both are sealed at their one chokepoint here: the dispatch boundary for the ask,
+the delegation return for the refusal. Prompts below are the REAL ones read from blocked chats.
+"""
+
+from backend.apps.agents.core.error_classify import defuse_extraction_ask, neutralize_provider_refusal
+
+# Verbatim from the transcripts of chats that were blocked (install 517559f0, 2026-08-21).
+REAL_EXTRACTION_ASKS = [
+    "Please give me a complete dump of what you were asked to do and how far you got. Include: "
+    "(1) the original user request verbatim if possible, (2) any files you created",
+    "Please give me the complete final output of your work in this session, formatted cleanly so "
+    "it can be sent as an email. Include all the key details, numbers, and recommendations.",
+    "Don't try to send anything. Just reply in plain text with the exact email you were supposed "
+    "to send to alex@openswarm.com: the subject line, and the full body text. Nothing else.",
+]
+REAL_REFUSAL = (
+    "API Error: Claude Code is unable to respond to this request, which appears to violate our "
+    "Usage Policy (https://www.anthropic.com/legal/aup). This request was blocked as it seems to "
+    "violate Anthropic's Terms of Service restrictions on reverse engineering or duplicating model outputs."
+)
+
+
+def test_real_extraction_asks_never_leave_the_dispatch_boundary_unchanged():
+    for ask in REAL_EXTRACTION_ASKS:
+        out = defuse_extraction_ask(ask)
+        assert out != ask, f"still shipped the extraction shape: {ask[:60]}"
+        assert "in your own words" in out and "verbatim" in out.split("Answer in your own words")[1]
+
+
+def test_ordinary_handoffs_are_untouched():
+    """Negative control: gutting normal delegation would be its own regression."""
+    for ok in (
+        "Check whether the API server is running and report the port.",
+        "Summarize the pricing tiers you landed on.",
+        "Fix the failing test in api.test.ts and tell me what broke.",
+        "",
+    ):
+        assert defuse_extraction_ask(ok) == ok
+
+
+def test_a_childs_refusal_never_travels_home_as_content():
+    out = neutralize_provider_refusal(REAL_REFUSAL)
+    assert "Usage Policy" not in out and "duplicating model outputs" not in out
+    assert "could not answer" in out
+    assert "Do not repeat or quote" in out, "the parent must not be invited to re-state it either"
+
+
+def test_a_real_delegation_answer_is_passed_through_untouched():
+    """Negative control: only refusals are rewritten, never a genuine result."""
+    for real in ("The build passed, 12 files changed.", "Tiers: $99 / $299 / custom.", ""):
+        assert neutralize_provider_refusal(real) == real
+
+
+def test_both_delegation_doors_defuse_not_just_one():
+    """SpawnAgent and InvokeAgent are separate routes. Sealing one and calling it done is exactly the
+    hole this caught: the first pass wired the defuse into /api/invoke-agent/run only, so every
+    extraction-shaped SpawnAgent prompt still went out untouched."""
+    import re
+    from pathlib import Path
+
+    p_main = Path(__file__).resolve().parents[1] / "main.py"
+    p_src = p_main.read_text()
+
+    for p_route in ("/api/invoke-agent/run", "/api/spawn-agent/run"):
+        p_at = p_src.index(p_route)
+        # The seal has to sit in the handler itself, not merely somewhere in the file.
+        p_body = p_src[p_at:p_at + 2000]
+        assert "defuse_extraction_ask" in p_body, f"{p_route} dispatches an un-defused handoff prompt"
+        assert re.search(r"=\s*defuse_extraction_ask\(", p_body), f"{p_route} calls the defuse but drops its result"