[codex] preserve active background delegations during idle GC

docs/chat-chain-changes/2026-07-28-agent-bridge-background-delegation-gc.md

@@ -0,0 +1,11 @@
+---
+date: 2026-07-28
+pr_or_commit: pending
+feature: Agent Bridge background delegation idle GC
+impact: Agent Bridge idle-session cleanup now preserves sessions that still own running or finalizing background delegations.
+---
+
+The worker checks durable async delegation state before destroying a session
+that has exceeded the 30-minute idle timeout. If that registry is temporarily
+unavailable, it falls back to the worker's background task telemetry. Sessions
+without active background work continue to be reclaimed normally.

packages/server/src/services/hermes/agent-bridge/python/bridge_pool.py

@@ -1731,6 +1731,33 @@ def _background_delegation_ids_for_session(session_id: str) -> list[str]:
         except Exception:
             return []
 
+    def has_active_background_for_session(self, session_id: str) -> bool:
+        """Return whether a session still owns running background work."""
+        try:
+            from tools.async_delegation import list_async_delegations
+
+            for record in list_async_delegations():
+                if str(record.get("status") or "").lower() not in {"running", "finalizing"}:
+                    continue
+                if any(
+                    str(record.get(key) or "") == session_id
+                    for key in ("session_key", "origin_ui_session_id", "parent_session_id")
+                ):
+                    return True
+            return False
+        except Exception:
+            # The durable registry is authoritative when available. Fall back
+            # to worker telemetry so a transient registry failure cannot let
+            # idle GC interrupt a task the worker still sees as active.
+            with self._lock:
+                session = self._sessions.get(session_id)
+                if session is None:
+                    return False
+                return any(
+                    str(task.get("status") or "").lower() in {"running", "finalizing"}
+                    for task in session.background_tasks.values()
+                )
+
     @staticmethod
     def _interrupt_background_for_session(session_id: str, reason: str) -> int:
         try:

packages/server/src/services/hermes/agent-bridge/python/bridge_server.py

@@ -648,6 +648,8 @@ def _gc_idle_sessions(self) -> None:
                 if not s.running and now - s.last_used_at > self.IDLE_TIMEOUT_SECONDS
             ]
         for sid in idle_ids:
+            if self.pool.has_active_background_for_session(sid):
+                continue
             self.pool.destroy(sid)
 
     def serve_forever(self) -> None:

tests/server/agent-bridge-python-concurrency.test.ts

@@ -626,6 +626,62 @@ assert polled["sessions"][0]["events"][-1]["status"] == "interrupted"
 `)
   })
 
+  it('keeps idle sessions alive while durable background delegations are active', () => {
+    runPython(String.raw`
+${harness}
+
+async_module = types.ModuleType("tools.async_delegation")
+async_module.list_async_delegations = lambda: [
+    {
+        "delegation_id": "deleg-running",
+        "status": "running",
+        "parent_session_id": "session-running",
+    },
+    {
+        "delegation_id": "deleg-finalizing",
+        "status": "finalizing",
+        "origin_ui_session_id": "session-finalizing",
+    },
+]
+async_module.interrupt_for_session = lambda **kwargs: 0
+sys.modules["tools.async_delegation"] = async_module
+
+server = bridge.BridgeServer("tcp://127.0.0.1:1")
+server.GC_INTERVAL_SECONDS = 0
+old = time.time() - server.IDLE_TIMEOUT_SECONDS - 1
+for session_id in ("session-running", "session-finalizing", "session-idle"):
+    session = bridge.AgentSession(session_id=session_id, agent=object())
+    session.last_used_at = old
+    server.pool._sessions[session_id] = session
+
+server._gc_idle_sessions()
+
+assert set(server.pool._sessions) == {"session-running", "session-finalizing"}
+`)
+  })
+
+  it('falls back to worker task state when checking idle background work', () => {
+    runPython(String.raw`
+${harness}
+
+async_module = types.ModuleType("tools.async_delegation")
+async_module.list_async_delegations = lambda: (_ for _ in ()).throw(RuntimeError("registry unavailable"))
+async_module.interrupt_for_session = lambda **kwargs: 0
+sys.modules["tools.async_delegation"] = async_module
+
+server = bridge.BridgeServer("tcp://127.0.0.1:1")
+server.GC_INTERVAL_SECONDS = 0
+session = bridge.AgentSession(session_id="session-1", agent=object())
+session.last_used_at = time.time() - server.IDLE_TIMEOUT_SECONDS - 1
+session.background_tasks["child-1"] = {"status": "running"}
+server.pool._sessions[session.session_id] = session
+
+server._gc_idle_sessions()
+
+assert "session-1" in server.pool._sessions
+`)
+  })
+
   it('hot-switches a loaded idle session model without recreating the session', () => {
     runPython(String.raw`
 ${harness}