[eric] agents: the recap-free policy retry arms on ANY fresh-session recap, not just post-compaction; the packaged drill caught the gate firing never (ENG-358)

backend/apps/agents/manager/run/handle_run_error.py

@@ -204,7 +204,9 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
         # re-blocks, which is the "hit a snag" flicker Alex reported, and a compaction recap that
         # re-sends each turn bricks the whole chat. One honest card; a recap-bearing session gets
         # one silent retry WITHOUT the recap, which is the only self-heal that can work.
-        if getattr(session, "compacted_through_msg_id", None) and not getattr(session, "policy_retry_used", False):
+        # Gate on "a recap could have been sent", not on the compaction mark: every fresh-session
+        # spawn with history carries the recap, and the drill proved the block fires there too.
+        if len(session.messages) > 1 and not getattr(session, "policy_retry_used", False):
             session.policy_retry_used = True
             session.suppress_recap_once = True
             session.pending_continuation = True

backend/tests/test_content_policy_block.py

@@ -41,3 +41,34 @@ def test_policy_fields_default_off():
     s = AgentSession(name="t", model="sonnet")
     assert s.policy_retry_used is False
     assert s.suppress_recap_once is False
+
+
+def test_policy_block_arms_one_recap_free_retry(monkeypatch):
+    """The drill-found gate bug: a fresh-session recap block must arm the silent retry even when
+    compacted_through_msg_id is still None (the recap is sent before any compaction mark exists)."""
+    import asyncio
+    from backend.apps.agents.manager.run.handle_run_error import handle_run_error
+    from backend.apps.agents.manager.streaming.state import TurnState
+    s = AgentSession(name="t", model="opus-5")
+    s.messages.append(Message(role="user", content="do the thing"))
+    s.messages.append(Message(role="tool_call", content={"tool": "Read", "input": {}}))
+    s.messages.append(Message(role="tool_result", content={"text": "x" * 300}))
+    assert s.compacted_through_msg_id is None
+    asyncio.run(handle_run_error(Exception(TOS_TEXT), s, "sid-pol", TurnState(), []))
+    assert s.policy_retry_used is True
+    assert s.suppress_recap_once is True
+    assert s.pending_continuation is True
+    assert not any(m.role == "system" for m in s.messages), "retry turn must be silent, no card yet"
+
+
+def test_second_policy_block_renders_the_terminal_card():
+    import asyncio
+    from backend.apps.agents.manager.run.handle_run_error import handle_run_error
+    from backend.apps.agents.manager.streaming.state import TurnState
+    s = AgentSession(name="t", model="opus-5")
+    s.messages.append(Message(role="user", content="do the thing"))
+    s.messages.append(Message(role="assistant", content="working"))
+    s.policy_retry_used = True
+    asyncio.run(handle_run_error(Exception(TOS_TEXT), s, "sid-pol2", TurnState(), []))
+    cards = [m for m in s.messages if m.role == "system"]
+    assert len(cards) == 1 and "declined this request" in str(cards[0].content)