增强可观测性与归因：session_recreated、双轨看门狗、路径归因 (PR-3)

src/agent_bridge/adapters/acp.py

@@ -1629,6 +1629,7 @@ async def ensure_session(self, session: Session) -> None:
                 return
         live = await self._spawn(session)
         native = session.native_session_id
+        load_failure: Exception | None = None
         if native and self.can_revive():
             use_resume = self.agent.name in _RESUME_AGENTS or (
                 self.agent.name == "dsh" and live.dsh_native_acp
@@ -1647,6 +1648,7 @@ async def ensure_session(self, session: Session) -> None:
                 raise
             except Exception as exc:
                 log.warning("session/load failed for %s; creating a new session: %s", session.session_id, _redact_diagnostic(str(exc)))
+                load_failure = exc
             else:
                 self._remember_config_options(live, revived)
                 await self._sync_selection(live, session)
@@ -1656,7 +1658,28 @@ async def ensure_session(self, session: Session) -> None:
             "session/new",
             session,
         )
+        # Adopt the new id before any fallible bookkeeping: a transcript write
+        # failure must not leave the session pointing at the rejected id.
         session.native_session_id = created.session_id
+        if load_failure is not None:
+            # H-05: the coordinator must see that its session lost context.
+            # pending_warnings lands on the next turn's task.warnings; the
+            # transcript event keeps the switch auditable after the fact. Both
+            # are emitted only once session/new has actually succeeded — a
+            # failed fallback is a failed task, not a recreation.
+            live.pending_warnings.append(
+                f"session/load failed ({type(load_failure).__name__}); started a new worker "
+                "session — prior conversation context is lost"
+            )
+            append_event(
+                session.session_id,
+                "session_recreated",
+                {
+                    "old_native_session_id": native,
+                    "reason": _redact_diagnostic(str(load_failure)),
+                },
+                self.home,
+            )
         self._remember_config_options(live, created)
         if self.agent.name == "grok":
             # Grok /new applies the _meta reasoningEffort but always lands on

src/agent_bridge/adapters/antigravity.py

@@ -288,7 +288,13 @@ async def _drain_stderr(self, proc: asyncio.subprocess.Process, session_id: str)
                 if text:
                     tail = f"{tail}\n{text}"[-STDERR_TAIL_LIMIT:]
         except (ValueError, OSError):
-            log.warning("stderr drain aborted for %s", session_id, exc_info=True)
+            log.warning(
+                "stderr drain aborted for %s (line over %s MiB or stream closed); "
+                "later stderr evidence is lost",
+                session_id,
+                STDIO_LIMIT // (1024 * 1024),
+                exc_info=True,
+            )
             return tail
 
     async def _write_prompt(self, proc: asyncio.subprocess.Process, message: str) -> None:
@@ -345,7 +351,14 @@ async def run_turn(self, session: Session, task: Task) -> TurnResult:
         try:
             assert proc.stdout is not None
             while True:
-                line = await proc.stdout.readline()
+                try:
+                    line = await proc.stdout.readline()
+                except ValueError as exc:
+                    raise RuntimeError(
+                        f"antigravity emitted a stdout line longer than "
+                        f"{STDIO_LIMIT // (1024 * 1024)} MiB (asyncio stream limit); "
+                        "the turn cannot continue"
+                    ) from exc
                 if not line:
                     break
                 raw = line.decode("utf-8", errors="replace").rstrip()

src/agent_bridge/adapters/codex.py

@@ -76,7 +76,13 @@ async def _drain_stderr(self, proc: asyncio.subprocess.Process, session_id: str)
                 if text:
                     tail = f"{tail}\n{text}"[-STDERR_TAIL_LIMIT:]
         except (ValueError, OSError):
-            log.warning("stderr drain aborted for %s", session_id, exc_info=True)
+            log.warning(
+                "stderr drain aborted for %s (line over %s MiB or stream closed); "
+                "later stderr evidence is lost",
+                session_id,
+                STDIO_LIMIT // (1024 * 1024),
+                exc_info=True,
+            )
             return tail
 
     async def _write_prompt(self, proc: asyncio.subprocess.Process, message: str) -> None:
@@ -137,7 +143,14 @@ async def run_turn(self, session: Session, task: Task) -> TurnResult:
             await self._write_prompt(proc, task.message)
             assert proc.stdout is not None
             while True:
-                line = await proc.stdout.readline()
+                try:
+                    line = await proc.stdout.readline()
+                except ValueError as exc:
+                    raise RuntimeError(
+                        f"codex emitted a stdout line longer than "
+                        f"{STDIO_LIMIT // (1024 * 1024)} MiB (asyncio stream limit); "
+                        "the turn cannot continue"
+                    ) from exc
                 if not line:
                     break
                 raw = line.decode("utf-8", errors="replace").rstrip()

src/agent_bridge/config.py

@@ -116,6 +116,10 @@ class AgentConfig(BaseModel):
     revivable: bool = False
     idle_unload_sec: int = 0
     stall_timeout_sec: int = Field(default=1800, ge=0)
+    # Hard cap on one turn's wall-clock length (0 = disabled). Catches a
+    # worker that stays chatty forever without ever going silent —
+    # stall_timeout_sec alone cannot see that case (H-16).
+    turn_timeout_sec: int = Field(default=0, ge=0)
     print_timeout: str = "120m"
 
 

src/agent_bridge/models.py

@@ -134,6 +134,10 @@ class Task(BaseModel):
     files_changed_total: int = 0
     files_changed_truncated: bool = False
     files_changed_state: FilesChangedState = FilesChangedState.pending
+    # Paths the worker REPORTED that resolve outside task.cwd. Kept separate
+    # so an escape is never silently folded into an in-workspace path (E3).
+    files_outside_cwd: list[str] = Field(default_factory=list)
+    files_outside_cwd_total: int = 0
     usage: dict[str, Any] = Field(default_factory=dict)
     error: str | None = None
     failure: TaskFailure | None = None

src/agent_bridge/registry.py

@@ -51,6 +51,7 @@
     append_event,
     flush_pending,
     flush_session,
+    forget_session,
     forget_worker_activity,
     mark_worker_activity,
     page_events,
@@ -694,13 +695,15 @@ async def _run_task(self, task_id: str) -> None:
         try:
             mark_worker_activity(session.session_id, self.home)
             before = await asyncio.to_thread(snapshot_workspace, task.cwd)
-            limit = self.config.get(session.agent).stall_timeout_sec
+            agent_cfg = self.config.get(session.agent)
+            limit = agent_cfg.stall_timeout_sec
+            turn_cap = agent_cfg.turn_timeout_sec
             watch = (
                 asyncio.create_task(
-                    self._stall_watch(task, session, adapter, limit),
+                    self._stall_watch(task, session, adapter, limit, turn_cap),
                     name=f"stall-{task_id}",
                 )
-                if limit > 0
+                if limit > 0 or turn_cap > 0
                 else None
             )
             result = await adapter.run_turn(session, task)
@@ -717,10 +720,10 @@ async def _run_task(self, task_id: str) -> None:
                     f"full result persistence failed: {type(exc).__name__}: {exc}"
                 )
                 log.exception("could not persist full result for task %s", task.task_id)
-            full_changed = await asyncio.to_thread(
+            inside_changed, outside_changed = await asyncio.to_thread(
                 merge_files_changed, task.cwd, result.files_changed, before
             )
-            self._set_files_changed(task, full_changed)
+            self._set_files_changed(task, inside_changed, outside_changed)
             task.usage = result.usage
             if session.agent == "grok":
                 observed = await asyncio.to_thread(observe_grok_session, session.cwd, session.native_session_id)
@@ -788,10 +791,10 @@ async def _run_task(self, task_id: str) -> None:
                     task.files_changed_state = FilesChangedState.unavailable
                 else:
                     try:
-                        full_changed = await asyncio.to_thread(
+                        inside_changed, outside_changed = await asyncio.to_thread(
                             merge_files_changed, task.cwd, [], before
                         )
-                        self._set_files_changed(task, full_changed)
+                        self._set_files_changed(task, inside_changed, outside_changed)
                     except Exception as exc:
                         task.files_changed_state = FilesChangedState.unavailable
                         task.warnings.append(
@@ -835,10 +838,13 @@ async def _run_task(self, task_id: str) -> None:
             self._schedule_idle(session.session_id)
 
     @staticmethod
-    def _set_files_changed(task: Task, paths: list[str]) -> None:
+    def _set_files_changed(task: Task, paths: list[str], outside: list[str] | None = None) -> None:
         task.files_changed_total = len(paths)
         task.files_changed = paths[:FILES_CHANGED_MAX]
         task.files_changed_truncated = len(paths) > FILES_CHANGED_MAX
+        outside = outside or []
+        task.files_outside_cwd_total = len(outside)
+        task.files_outside_cwd = outside[:FILES_CHANGED_MAX]
         task.files_changed_state = FilesChangedState.collected
 
     async def _stall_watch(
@@ -847,30 +853,54 @@ async def _stall_watch(
         session: Session,
         adapter: Adapter,
         limit: int,
+        turn_cap: int = 0,
     ) -> None:
+        """Dual watchdog: ``limit`` bounds worker silence, ``turn_cap`` bounds
+        total turn wall-clock. Silence only counts worker-originated transcript
+        events, so a heartbeat-spamming worker still needs the cap (H-16)."""
+        started = time.monotonic()
         while True:
             silence = worker_silence_sec(session.session_id, self.home) or 0.0
-            remaining = limit - silence
-            if remaining <= 0:
+            elapsed = time.monotonic() - started
+            if limit > 0 and silence >= limit:
+                reason = "stalled"
                 break
-            await asyncio.sleep(min(remaining, STALL_POLL_SEC))
+            if turn_cap > 0 and elapsed >= turn_cap:
+                reason = "timed_out"
+                break
+            waits: list[float] = [STALL_POLL_SEC]
+            if limit > 0:
+                waits.append(limit - silence)
+            if turn_cap > 0:
+                waits.append(turn_cap - elapsed)
+            await asyncio.sleep(max(min(waits), 0.05))
         if task.status in TERMINAL_STATUSES:
             return
         log.warning(
-            "task %s stalled: no worker output for %ss; cancelling the turn",
+            "task %s %s: cancelling the turn (silence=%.0fs, elapsed=%.0fs)",
             task.task_id,
-            limit,
+            reason,
+            silence,
+            elapsed,
         )
         task.status = TaskStatus.failed
-        task.stop_reason = "stalled"
+        task.stop_reason = reason
         task.error = (
             f"worker produced no output for {limit}s (stall_timeout_sec); "
             "Bridge cancelled the turn"
+            if reason == "stalled"
+            else f"turn exceeded turn_timeout_sec={turn_cap}s; Bridge cancelled the turn"
         )
         append_event(
             session.session_id,
             "error",
-            {"error": task.error, "stalled": True, "stall_timeout_sec": limit},
+            {
+                "error": task.error,
+                "stalled": reason == "stalled",
+                "timed_out": reason == "timed_out",
+                "stall_timeout_sec": limit,
+                "turn_timeout_sec": turn_cap,
+            },
             self.home,
         )
         # The turn normally returns (and _run_task tears this watch down)
@@ -920,7 +950,9 @@ def _prune_sessions(self) -> None:
         for session in drop:
             session_id = session.session_id
             self.sessions.pop(session_id, None)
-            forget_worker_activity(session_id, self.home)
+            # Result files die with their tasks; the per-session transcript
+            # would otherwise grow unbounded on disk (H-02).
+            forget_session(session_id, self.home)
             self._cancel_idle(session_id)
             for task in [item for item in self.tasks.values() if item.session_id == session_id]:
                 self._drop_task(task.task_id)
@@ -1048,11 +1080,17 @@ def _result_hint(task: Task, prefix: str) -> str:
             )
         if task.stop_reason == "stalled":
             hint += (
-                " The worker went silent for stall_timeout_sec and Bridge cancelled the turn; "
+                " The worker produced no output for stall_timeout_sec and Bridge cancelled the turn; "
                 "read get_transcript for its last activity, then either dispatch a narrower "
                 f"task on the same session_id or raise [agents.{task.agent}] stall_timeout_sec "
                 "if that step was legitimately long."
             )
+        if task.stop_reason == "timed_out":
+            hint += (
+                " The turn hit turn_timeout_sec and Bridge cancelled it; "
+                f"raise [agents.{task.agent}] turn_timeout_sec if turns legitimately "
+                "need more wall-clock time."
+            )
         return hint
 
     def _task_snapshot(self, task: Task, include_result: bool = False) -> dict:
@@ -1070,6 +1108,8 @@ def _task_snapshot(self, task: Task, include_result: bool = False) -> dict:
             "files_changed_total": task.files_changed_total,
             "files_changed_truncated": task.files_changed_truncated,
             "files_changed_state": task.files_changed_state.value,
+            "files_outside_cwd": task.files_outside_cwd,
+            "files_outside_cwd_total": task.files_outside_cwd_total,
             "model": task.model,
             "effort": task.effort,
             "observed_model": task.observed_model,

src/agent_bridge/transcript.py

@@ -56,6 +56,21 @@ def _pending_text(path: Path) -> str:
         return "".join(buffer.lines) if buffer else ""
 
 
+# Event types that represent bytes arriving FROM the worker (session/update
+# notifications, raw JSONL, streamed chunks). Everything else in a transcript
+# is written by Bridge itself and never proves the worker is alive.
+WORKER_ACTIVITY_TYPES = frozenset(
+    {
+        "message_chunk",
+        "thought_chunk",
+        "tool_call",
+        "tool_call_update",
+        "plan",
+        "raw",
+    }
+)
+
+
 def append_event(session_id: str, event_type: str, data: dict[str, Any] | None = None, home: Path | None = None) -> TranscriptEvent:
     event = TranscriptEvent(type=event_type, data=data or {})
     path = transcript_path(session_id, home)
@@ -73,7 +88,11 @@ def append_event(session_id: str, event_type: str, data: dict[str, Any] | None =
             and now - buffer.first_pending_at >= BUFFER_MAX_AGE_SEC
         )
         terminal = event_type in {"turn_end", "error"}
-        _activity[path] = now
+        # Only worker-originated events reset the stall clock. Bridge-side
+        # bookkeeping (prompt_sent, turn_end, error, permission, ...) written
+        # into the same transcript must not count as "the worker is alive".
+        if event_type in WORKER_ACTIVITY_TYPES:
+            _activity[path] = now
         if buffer.size >= BUFFER_BYTE_LIMIT or aged or terminal:
             _flush_locked(path, buffer)
         if terminal and not buffer.lines:
@@ -102,6 +121,24 @@ def forget_worker_activity(session_id: str, home: Path | None = None) -> None:
         _activity.pop(path, None)
 
 
+def forget_session(session_id: str, home: Path | None = None) -> None:
+    """Drop every cached trace of a pruned session and delete its transcript.
+
+    Result files die with their task; the per-session transcript would
+    otherwise live forever (H-02). Any unflushed buffered lines are dropped
+    with the file — the session is being deleted, not archived.
+    """
+    path = transcript_path(session_id, home)
+    with _buffers_lock:
+        _activity.pop(path, None)
+        _buffers.pop(path, None)
+        _parse_cache.pop(path, None)
+    try:
+        path.unlink(missing_ok=True)
+    except OSError:
+        log.warning("could not remove pruned transcript %s", path)
+
+
 def flush_pending(home: Path | None = None) -> None:
     """Flush buffered transcript events, normally during Bridge shutdown."""
     root = home / "transcripts" if home is not None else None