[issue-3064][slice-4/6] Worktree re-attach + gateway-session reuse...

orchestrator/concurrent_executor.py

@@ -491,6 +491,7 @@ def _agent_free(*, action: str, role: str, payload: Any = None) -> None:
         supervisor = JobSupervisor(
             overseer_alert=self._emit_supervision_alert,
             agent_failed=self._handle_propose_arm_exhaustion,
+            on_exhausted=self._teardown_exhausted_session,
         )
 
         loop = OrchestratorEventLoop(
@@ -936,6 +937,50 @@ def _handle_propose_arm_exhaustion(
                 error=str(exc),
             )
 
+    def _teardown_exhausted_session(self, *, role: str, action: str, dedupe_key: str) -> None:
+        """Release a role's reused gateway session on streak exhaustion (#3064 slice-4).
+
+        Wired to :class:`JobSupervisor`'s ``on_exhausted`` hook (the
+        ``_exhausted`` transition). The exhausted event arm will spawn no
+        further events, so the long-lived orchestrator-mode session keyed by
+        the role's stable base ``container_id`` is torn down here rather than
+        lingering to pipeline cleanup. Reaches :meth:`KubernetesSpawner.
+        _teardown_session` via the teardown closure attached to ``spawn_fn``
+        (the spawner is not held directly). Best-effort — a teardown failure
+        must never wedge the supervision path.
+        """
+        teardown = getattr(self.spawn_fn, "teardown_event_session", None)
+        if teardown is None:
+            # Pod-mode / test spawn_fn without the event-mode teardown surface:
+            # there is no long-lived per-role session to release.
+            return
+        try:
+            agent_role = role if isinstance(role, AgentRole) else AgentRole(role)
+        except ValueError:
+            logger.warning(
+                "Unknown role on streak-exhaustion teardown; skipping",
+                pipeline_id=self.pipeline.id,
+                role=role,
+                dedupe_key=dedupe_key,
+            )
+            return
+        try:
+            teardown(agent_role)
+            logger.info(
+                "Tore down reused gateway session on streak exhaustion",
+                pipeline_id=self.pipeline.id,
+                role=agent_role.value,
+                action=action,
+                dedupe_key=dedupe_key,
+            )
+        except Exception as exc:  # noqa: BLE001 — never wedge the loop on teardown
+            logger.warning(
+                "Failed to tear down reused gateway session on streak exhaustion",
+                pipeline_id=self.pipeline.id,
+                role=agent_role.value,
+                error=str(exc),
+            )
+
     def _abort_phase(self, error: str, recent_failures: int) -> dict[str, Any]:
         """Abort the phase due to multiple simultaneous failures."""
         emit_event(

orchestrator/event_loop.py

@@ -256,10 +256,20 @@ def __init__(
         clock: Callable[[], float] = time.monotonic,
         overseer_alert: Callable[..., Any] | None = None,
         agent_failed: Callable[..., Any] | None = None,
+        on_exhausted: Callable[..., Any] | None = None,
     ) -> None:
         self.clock = clock
         self._overseer_alert = overseer_alert
         self._agent_failed = agent_failed
+        # #3064 slice-4: fired once when a dedupe key crosses into the
+        # exhausted set (the ``_exhausted`` transition). The orchestrator
+        # wires this to tear down the role's reused gateway session — an
+        # exhausted arm spawns no further events, so its long-lived
+        # orchestrator-mode session is released here rather than lingering to
+        # pipeline cleanup. Called as ``on_exhausted(role=, action=,
+        # dedupe_key=)``; best-effort (a teardown error never wedges
+        # supervision).
+        self._on_exhausted = on_exhausted
         # Per-dedupe-key streaks — each key gets a fresh budget
         # The counter resets when the dedupe key changes, giving a fresh
         # budget for each distinct event.
@@ -353,6 +363,23 @@ def record_abort(self, dedupe_key: str, action: str, role: str) -> None:
             self._alerted_10[dedupe_key] = True
             self._exhausted.add(dedupe_key)
             self._emit_alert(dedupe_key, streak, action, role)
+            # #3064 slice-4: release the role's reused orchestrator-mode gateway
+            # session — the exhausted arm spawns no further events, so the
+            # long-lived session is torn down at this transition (any later
+            # spawn for the role simply re-registers on a cache miss). Fires for
+            # every action (a stuck reviewer arm is just as dead as a producer);
+            # best-effort so a teardown failure never wedges supervision.
+            if self._on_exhausted is not None:
+                try:
+                    self._on_exhausted(role=role, action=action, dedupe_key=dedupe_key)
+                except Exception:  # noqa: BLE001 — teardown is best-effort
+                    logger.warning(
+                        "JobSupervisor: on_exhausted teardown failed for key=%s "
+                        "(action=%s, role=%s)",
+                        dedupe_key,
+                        action,
+                        role,
+                    )
             # #2806 relocated for orchestrator mode: a *producer* propose arm
             # that exhausts its budget engages the existing AGENT_FAILED path.
             # Reviewer arms (ack/nack) are not producer failures.