From d55f55f7ccd302a0fd0e07479f436702f5310233 Mon Sep 17 00:00:00 2001
From: lulu feizhu <279467036+FeiZhuLulu@users.noreply.github.com>
Date: Tue, 6 Oct 2026 07:24:14 +0000
Subject: [PATCH 1/7] =?UTF-8?q?=E5=A2=9E=E5=BC=BA=E5=8F=AF=E8=A7=82?=
 =?UTF-8?q?=E6=B5=8B=E6=80=A7=E4=B8=8E=E5=BD=92=E5=9B=A0=EF=BC=9Asession?=
 =?UTF-8?q?=5Frecreated=E3=80=81=E5=8F=8C=E8=BD=A8=E7=9C=8B=E9=97=A8?=
 =?UTF-8?q?=E7=8B=97=E3=80=81=E8=B7=AF=E5=BE=84=E5=BD=92=E5=9B=A0=20(PR-3)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- session/load 失败静默回退改走 pending_warnings + session_recreated 转录事件（H-05）
- WORKER_ACTIVITY_TYPES 白名单：仅 worker 输出事件重置活跃钟；新增 turn_timeout_sec 硬上限（H-16）
- SKIP_DIR_NAMES 补 .agent-bridge/.codex/.claude；快照跳过 junction；逃逸路径进 files_outside_cwd（H-06/E2/E3）
- transcript.forget_session 挂到会话淘汰，清缓存+删转录（H-02）
- stdout readline 超限转译为带数值的 RuntimeError（H-10）
---
 src/agent_bridge/adapters/acp.py         | 16 +++++
 src/agent_bridge/adapters/antigravity.py | 17 ++++-
 src/agent_bridge/adapters/codex.py       | 17 ++++-
 src/agent_bridge/config.py               |  4 ++
 src/agent_bridge/models.py               |  4 ++
 src/agent_bridge/registry.py             | 74 +++++++++++++++-----
 src/agent_bridge/transcript.py           | 39 ++++++++++-
 src/agent_bridge/workspace.py            | 87 ++++++++++++++++++------
 tests/echo_agent.py                      |  4 ++
 tests/test_acp_echo.py                   | 50 ++++++++++++++
 tests/test_registry.py                   | 50 +++++++++++++-
 tests/test_transcript.py                 | 36 ++++++++++
 tests/test_workspace.py                  | 70 +++++++++++++++++--
 13 files changed, 418 insertions(+), 50 deletions(-)

diff --git a/src/agent_bridge/adapters/acp.py b/src/agent_bridge/adapters/acp.py
index 89dfad6..d25a42b 100644
--- a/src/agent_bridge/adapters/acp.py
+++ b/src/agent_bridge/adapters/acp.py
@@ -1647,6 +1647,22 @@ 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)))
+                # 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.
+                live.pending_warnings.append(
+                    f"session/load failed ({type(exc).__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(exc)),
+                    },
+                    self.home,
+                )
             else:
                 self._remember_config_options(live, revived)
                 await self._sync_selection(live, session)
diff --git a/src/agent_bridge/adapters/antigravity.py b/src/agent_bridge/adapters/antigravity.py
index 365ff4d..8763a77 100644
--- a/src/agent_bridge/adapters/antigravity.py
+++ b/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()
diff --git a/src/agent_bridge/adapters/codex.py b/src/agent_bridge/adapters/codex.py
index 5dc7e26..5987657 100644
--- a/src/agent_bridge/adapters/codex.py
+++ b/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()
diff --git a/src/agent_bridge/config.py b/src/agent_bridge/config.py
index e756384..ed952bd 100644
--- a/src/agent_bridge/config.py
+++ b/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"
 
 
diff --git a/src/agent_bridge/models.py b/src/agent_bridge/models.py
index d261fd6..98e6691 100644
--- a/src/agent_bridge/models.py
+++ b/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
diff --git a/src/agent_bridge/registry.py b/src/agent_bridge/registry.py
index ca10342..98da548 100644
--- a/src/agent_bridge/registry.py
+++ b/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,
diff --git a/src/agent_bridge/transcript.py b/src/agent_bridge/transcript.py
index dfd3b1d..d15ad2e 100644
--- a/src/agent_bridge/transcript.py
+++ b/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
diff --git a/src/agent_bridge/workspace.py b/src/agent_bridge/workspace.py
index 3bf1c59..86a092e 100644
--- a/src/agent_bridge/workspace.py
+++ b/src/agent_bridge/workspace.py
@@ -33,6 +33,12 @@
     ".parcel-cache",
     ".svelte-kit",
     ".terraform",
+    # Bridge/worker bookkeeping dirs. A worker CLI keeps its own state under
+    # these and churns them every turn; counting that as project changes
+    # misattributes Bridge-internal writes to the worker (H-06).
+    ".agent-bridge",
+    ".codex",
+    ".claude",
 }
 
 _PATH_KEYS = (
@@ -64,7 +70,11 @@ def snapshot_workspace(cwd: str | Path) -> dict[str, tuple[int, int]]:
                 for entry in it:
                     try:
                         if entry.is_dir(follow_symlinks=False):
-                            if entry.name not in SKIP_DIR_NAMES:
+                            # Windows junctions pass is_dir(follow_symlinks=False)
+                            # but point outside cwd; traversing them attributes
+                            # foreign files as workspace writes (E2).
+                            is_junction = getattr(entry, "is_junction", lambda: False)()
+                            if entry.name not in SKIP_DIR_NAMES and not is_junction:
                                 stack.append(entry.path)
                         elif entry.is_file(follow_symlinks=False):
                             st = entry.stat(follow_symlinks=False)
@@ -90,48 +100,81 @@ def changed_since(cwd: str | Path, before: dict[str, tuple[int, int]]) -> list[s
 
 
 def normalize_changed_paths(cwd: str | Path, paths: list[str]) -> list[str]:
+    inside, _outside = classify_changed_paths(cwd, paths)
+    return inside
+
+
+def classify_changed_paths(
+    cwd: str | Path, paths: list[str]
+) -> tuple[list[str], list[str]]:
+    """Split reported paths into (inside-cwd rel, outside-cwd abs).
+
+    A path that resolves outside the workspace is reported as its resolved
+    absolute form in the second list — never folded into an inside path the
+    way ``../x`` used to collapse to ``x`` (E3).
+    """
     root = Path(cwd).resolve()
-    seen: set[str] = set()
-    ordered: list[str] = []
+    seen_in: set[str] = set()
+    seen_out: set[str] = set()
+    inside: list[str] = []
+    outside: list[str] = []
     for raw in paths:
         if not raw or not isinstance(raw, str):
             continue
-        rel = _relative_to_cwd(raw, root)
-        if not rel or _ignored_rel(rel):
-            continue
-        if rel not in seen:
-            seen.add(rel)
-            ordered.append(rel)
-    return ordered
+        rel, out = _classify_path(raw, root)
+        if rel and not _ignored_rel(rel) and rel not in seen_in:
+            seen_in.add(rel)
+            inside.append(rel)
+        elif out and out not in seen_out:
+            seen_out.add(out)
+            outside.append(out)
+    return inside, outside
 
 
 def merge_files_changed(
     cwd: str | Path,
     reported: list[str],
     before: dict[str, tuple[int, int]],
-) -> list[str]:
-    return normalize_changed_paths(cwd, [*reported, *changed_since(cwd, before)])
+) -> tuple[list[str], list[str]]:
+    """(inside-cwd rel paths, outside-cwd abs paths)."""
+    inside, outside = classify_changed_paths(cwd, reported)
+    disk = changed_since(cwd, before)
+    merged = inside + [rel for rel in disk if rel not in set(inside)]
+    return merged, outside
+
 
 
-def _relative_to_cwd(raw: str, root: Path) -> str | None:
+def _classify_path(raw: str, root: Path) -> tuple[str | None, str | None]:
+    """(inside rel | None, outside absolute | None). Both None = unusable."""
     text = raw.strip()
     if text.startswith("file://"):
         text = text[7:]
         if text.startswith("/") and len(text) >= 3 and text[2] == ":":
             text = text[1:]
     path = Path(text)
-    rel: str | None
     try:
         resolved = path.resolve() if path.is_absolute() else (root / path).resolve()
-        rel = resolved.relative_to(root).as_posix()
-    except (OSError, ValueError):
-        rel = path.as_posix().lstrip("./") if not path.is_absolute() else None
+    except (OSError, ValueError, RuntimeError):
+        resolved = None
+    if resolved is not None:
+        try:
+            rel = resolved.relative_to(root).as_posix()
+        except ValueError:
+            return None, resolved.as_posix()
+        if rel in {".", "..", ""}:
+            return None, None
+        return rel, None
+    # Unresolvable: keep prior lenient behavior for relative strings, but a
+    # leading ../ can never be proven inside — report it as outside instead
+    # of folding it into the workspace (E3).
+    if not path.is_absolute():
+        rel = path.as_posix().lstrip("./")
         if rel in {".", "..", ""}:
-            return None
-        return rel
-    if rel in {".", "..", ""}:
-        return None
-    return rel
+            return None, None
+        if rel.startswith("../") or rel == "..":
+            return None, path.as_posix()
+        return rel, None
+    return None, path.as_posix()
 
 
 def _root_prefix_len(root: str) -> int:
diff --git a/tests/echo_agent.py b/tests/echo_agent.py
index 7ce1c88..6e9191a 100644
--- a/tests/echo_agent.py
+++ b/tests/echo_agent.py
@@ -65,6 +65,10 @@ async def new_session(self, cwd: str, mcp_servers=None, **kwargs: Any) -> NewSes
     async def load_session(self, cwd: str, session_id: str, mcp_servers=None, **kwargs: Any) -> None:
         if os.environ.get("BRIDGE_ECHO_FAILURE") == "load":
             raise RequestError.auth_required()
+        if os.environ.get("BRIDGE_ECHO_FAILURE") == "load_internal":
+            # Generic failure (not "unavailable"): Bridge should fall back to
+            # session/new and mark the switch visibly (H-05).
+            raise RequestError(-32603, "native session data corrupt")
         self._session_id = session_id
         self._ready = True
         return None
diff --git a/tests/test_acp_echo.py b/tests/test_acp_echo.py
index c744703..08116cc 100644
--- a/tests/test_acp_echo.py
+++ b/tests/test_acp_echo.py
@@ -222,3 +222,53 @@ def test_legacy_task_has_no_failure_evidence():
         "message": "old", "status": "failed", "error": "Connection closed",
     })
     assert task.failure is None
+
+
+@pytest.mark.asyncio
+async def test_acp_load_failure_surfaces_session_recreated(bridge_home, tmp_path, monkeypatch):
+    """H-05: session/load -> session/new fallback must be visible to the
+    coordinator (task warning) and auditable later (transcript event)."""
+    echo = Path(__file__).resolve().parent / "echo_agent.py"
+    work = tmp_path / "work"
+    work.mkdir()
+    cfg = AgentConfig(
+        name="echo",
+        protocol="acp",
+        command=[sys.executable, str(echo)],
+        revivable=True,
+        idle_unload_sec=0,
+    )
+    adapter = AcpAdapter(cfg, bridge_home)
+    session = Session(session_id="sess_reload", agent="echo", cwd=str(work.resolve()))
+    first = Task(
+        task_id="task_reload_1",
+        session_id=session.session_id,
+        agent="echo",
+        message="one",
+        cwd=str(work.resolve()),
+    )
+    try:
+        r1 = await adapter.run_turn(session, first)
+        assert r1.stop_reason == "end_turn"
+        assert session.native_session_id
+        await adapter.shutdown(session)
+        # Make the worker's session/load fail (generic, non-unavailable) on
+        # the next spawn so Bridge takes the session/new fallback.
+        cfg.env["BRIDGE_ECHO_FAILURE"] = "load_internal"
+        second = Task(
+            task_id="task_reload_2",
+            session_id=session.session_id,
+            agent="echo",
+            message="two",
+            cwd=str(work.resolve()),
+        )
+        r2 = await adapter.run_turn(session, second)
+        assert r2.stop_reason == "end_turn"
+        assert any("context is lost" in w for w in r2.warnings)
+        from agent_bridge.transcript import read_events
+
+        types = {event["type"] for event in read_events(session.session_id, bridge_home)}
+        assert "session_recreated" in types
+    finally:
+        cfg.env.pop("BRIDGE_ECHO_FAILURE", None)
+        await adapter.shutdown(session)
diff --git a/tests/test_registry.py b/tests/test_registry.py
index ec011eb..0478af4 100644
--- a/tests/test_registry.py
+++ b/tests/test_registry.py
@@ -1215,7 +1215,7 @@ async def counter() -> None:
 @pytest.mark.asyncio
 async def test_files_changed_is_capped(bridge_home, tmp_path, monkeypatch):
     paths = [f"gen/{i:04d}.txt" for i in range(500)]
-    monkeypatch.setattr("agent_bridge.registry.merge_files_changed", lambda *args, **kwargs: list(paths))
+    monkeypatch.setattr("agent_bridge.registry.merge_files_changed", lambda *args, **kwargs: (list(paths), []))
     work = tmp_path / "work"
     work.mkdir()
     registry = Registry.create(bridge_home)
@@ -1781,3 +1781,51 @@ async def slow_cancel(self, session):
         assert cancel_finished.is_set()
     finally:
         await registry.stop()
+
+
+@pytest.mark.asyncio
+async def test_turn_timeout_sec_caps_chatty_turn(bridge_home, tmp_path, monkeypatch):
+    """H-16: a worker emitting worker-type events forever still hits the cap."""
+    import asyncio as _asyncio
+
+    async def chatty_turn(self, session, task):
+        from agent_bridge.transcript import append_event
+
+        for _ in range(30):
+            append_event(session.session_id, "thought_chunk", {"text": "..."}, self.home)
+            await _asyncio.sleep(0.1)
+        return TurnResult(text="done")
+
+    monkeypatch.setattr(FakeAdapter, "run_turn", chatty_turn)
+    registry = Registry.create(bridge_home)
+    registry.config.agents["fake"].turn_timeout_sec = 1
+    await registry.start()
+    try:
+        dispatched = await registry.dispatch_task("fake", "chatty", cwd=str(tmp_path))
+        waited = await registry.wait_task(dispatched["task_id"], timeout_sec=10)
+        assert waited["status"] == "failed"
+        assert waited["stop_reason"] == "timed_out"
+        assert "turn_timeout_sec" in (waited["error"] or "")
+    finally:
+        await registry.stop()
+
+
+@pytest.mark.asyncio
+async def test_silent_turn_still_stalls(bridge_home, tmp_path, monkeypatch):
+    """Silence semantics: with no worker events at all, stall_timeout fires."""
+
+    async def silent_turn(self, session, task):
+        await asyncio.sleep(10)
+        return TurnResult(text="done")
+
+    monkeypatch.setattr(FakeAdapter, "run_turn", silent_turn)
+    registry = Registry.create(bridge_home)
+    registry.config.agents["fake"].stall_timeout_sec = 1
+    await registry.start()
+    try:
+        dispatched = await registry.dispatch_task("fake", "quiet", cwd=str(tmp_path))
+        waited = await registry.wait_task(dispatched["task_id"], timeout_sec=10)
+        assert waited["status"] == "failed"
+        assert waited["stop_reason"] == "stalled"
+    finally:
+        await registry.stop()
diff --git a/tests/test_transcript.py b/tests/test_transcript.py
index d25d9b4..a065c03 100644
--- a/tests/test_transcript.py
+++ b/tests/test_transcript.py
@@ -159,3 +159,39 @@ def test_parse_cache_is_bounded(bridge_home):
         flush_session(session_id, bridge_home)
         read_events(session_id, bridge_home)
     assert len(transcript._parse_cache) <= 4
+
+
+def test_only_worker_events_reset_the_activity_clock(bridge_home, monkeypatch):
+    """H-16: bridge-side bookkeeping must not count as worker aliveness."""
+    import time as _time
+
+    path = transcript_path("sess_gate", bridge_home)
+    with transcript._buffers_lock:
+        transcript._activity[path] = _time.monotonic() - 10.0
+    append_event("sess_gate", "prompt_sent", {"text": "hi"}, bridge_home)
+    append_event("sess_gate", "permission", {"decision": "selected"}, bridge_home)
+    append_event("sess_gate", "turn_end", {"stop_reason": "x"}, bridge_home)
+    silence = worker_silence_sec("sess_gate", bridge_home)
+    assert silence is not None and silence >= 9.0
+    append_event("sess_gate", "message_chunk", {"text": "worker speaks"}, bridge_home)
+    silence = worker_silence_sec("sess_gate", bridge_home)
+    assert silence is not None and silence < 2.0
+    append_event("sess_gate", "thought_chunk", {"text": "..."}, bridge_home)
+    assert (worker_silence_sec("sess_gate", bridge_home) or 0) < 2.0
+    forget_worker_activity("sess_gate", bridge_home)
+
+
+def test_forget_session_drops_caches_and_transcript_file(bridge_home):
+    """H-02: session prune must delete the transcript, not just the result."""
+    append_event("sess_forget", "message_chunk", {"text": "one"}, bridge_home)
+    flush_session("sess_forget", bridge_home)
+    path = transcript_path("sess_forget", bridge_home)
+    assert path.is_file()
+    mark_worker_activity("sess_forget", bridge_home)
+    read_events("sess_forget", bridge_home)
+    transcript.forget_session("sess_forget", bridge_home)
+    assert not path.exists()
+    assert worker_silence_sec("sess_forget", bridge_home) is None
+    with transcript._buffers_lock:
+        assert path not in transcript._parse_cache
+        assert path not in transcript._buffers
diff --git a/tests/test_workspace.py b/tests/test_workspace.py
index 7e6fdf3..4816f0c 100644
--- a/tests/test_workspace.py
+++ b/tests/test_workspace.py
@@ -1,6 +1,8 @@
 import os
 from pathlib import Path
 
+import pytest
+
 from agent_bridge.workspace import (
     _root_prefix_len,
     collect_update_paths,
@@ -16,8 +18,9 @@ def test_snapshot_sees_new_file_and_ignores_sessions(tmp_path: Path):
     sessions = tmp_path / ".sessions"
     sessions.mkdir()
     (sessions / "log.jsonl").write_text("{}\n", encoding="utf-8")
-    changed = merge_files_changed(tmp_path, [], before)
+    changed, outside = merge_files_changed(tmp_path, [], before)
     assert changed == ["smoke.txt"]
+    assert outside == []
 
 
 def test_merges_protocol_paths_as_cwd_relative(tmp_path: Path):
@@ -25,15 +28,19 @@ def test_merges_protocol_paths_as_cwd_relative(tmp_path: Path):
     target = tmp_path / "src" / "app.py"
     target.parent.mkdir()
     target.write_text("print(1)\n", encoding="utf-8")
-    changed = merge_files_changed(tmp_path, [str(target)], before)
+    changed, outside = merge_files_changed(tmp_path, [str(target)], before)
     assert changed == ["src/app.py"]
+    assert outside == []
 
 
 def test_drops_cwd_dot_and_root_paths(tmp_path: Path):
     before = snapshot_workspace(tmp_path)
     (tmp_path / "ok.txt").write_text("x\n", encoding="utf-8")
-    changed = merge_files_changed(tmp_path, [".", str(tmp_path), str(tmp_path / "ok.txt")], before)
+    changed, outside = merge_files_changed(
+        tmp_path, [".", str(tmp_path), str(tmp_path / "ok.txt")], before
+    )
     assert changed == ["ok.txt"]
+    assert outside == []
 
 
 def test_snapshot_skips_build_dirs_and_tracks_changes(tmp_path: Path):
@@ -56,11 +63,11 @@ def test_snapshot_skips_build_dirs_and_tracks_changes(tmp_path: Path):
     # A same-length rewrite can keep both size and mtime_ns on a coarse
     # filesystem clock, so the payload length has to change.
     (tmp_path / "src" / "a.py").write_text("print(22)\n", encoding="utf-8")
-    assert merge_files_changed(tmp_path, [], before) == ["src/a.py"]
+    assert merge_files_changed(tmp_path, [], before)[0] == ["src/a.py"]
 
     (nested / "b.txt").unlink()
     (tmp_path / "src" / "c.py").write_text("print(3)\n", encoding="utf-8")
-    changed = set(merge_files_changed(tmp_path, [], before))
+    changed = set(merge_files_changed(tmp_path, [], before)[0])
     assert "nested/deep/b.txt" in changed
     assert "src/c.py" in changed
     assert "src/a.py" in changed
@@ -92,3 +99,56 @@ def test_collect_nested_tool_paths():
     )
     assert "README.md" in found
     assert "src/main.py" in found
+
+
+def test_classify_splits_inside_and_outside_paths(tmp_path: Path):
+    from agent_bridge.workspace import classify_changed_paths
+
+    inside, outside = classify_changed_paths(
+        tmp_path,
+        [
+            "src/ok.py",
+            "../escape.txt",
+            str(tmp_path.parent / "sibling.py"),
+            str(tmp_path / "in.txt"),
+        ],
+    )
+    assert inside == ["src/ok.py", "in.txt"]
+    assert len(outside) == 2
+    assert not any(".." in p for p in inside)
+    # E3 regression: ../escape.txt must NOT fold to escape.txt inside cwd.
+    assert "escape.txt" not in inside
+
+
+def test_dotdot_absolute_outside_goes_to_outside_list(tmp_path: Path):
+    inside, outside = merge_files_changed(
+        tmp_path, [os.path.join("..", "..", "far_away.txt")], {}
+    )
+    assert inside == []
+    assert outside and outside[0].endswith("far_away.txt")
+
+
+def test_bridge_dirs_are_skipped_in_snapshot(tmp_path: Path):
+    for d in (".agent-bridge", ".codex", ".claude"):
+        sub = tmp_path / d
+        sub.mkdir()
+        (sub / "state.json").write_text("{}\n", encoding="utf-8")
+    (tmp_path / "real.py").write_text("x\n", encoding="utf-8")
+    assert set(snapshot_workspace(tmp_path)) == {"real.py"}
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows junction repro")
+def test_junction_inside_cwd_is_not_traversed(tmp_path: Path):
+    import subprocess
+
+    target = tmp_path.parent / "junction_target"
+    target.mkdir(exist_ok=True)
+    (target / "foreign.txt").write_text("out\n", encoding="utf-8")
+    link = tmp_path / "link"
+    subprocess.run(
+        ["cmd", "/c", "mklink", "/J", str(link), str(target)],
+        check=True,
+        capture_output=True,
+    )
+    snap = snapshot_workspace(tmp_path)
+    assert "link/foreign.txt" not in snap

From 3fa74494fc05c721c4da2f063c8241127dca4e36 Mon Sep 17 00:00:00 2001
From: lulu feizhu <279467036+FeiZhuLulu@users.noreply.github.com>
Date: Tue, 6 Oct 2026 07:43:38 +0000
Subject: [PATCH 2/7] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=A4=8D=E5=AE=A1?=
 =?UTF-8?q?=E5=8F=91=E7=8E=B0=E7=9A=84=E4=B8=89=E5=A4=84=E7=BC=BA=E9=99=B7?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- _classify_path 未解析回退：先判 ../ 再 lstrip——原顺序 lstrip 会先吃掉 ../，逃逸路径被折回 cwd 内
- session_recreated 事件与 pending_warning 移到 session/new 成功之后；fallback 失败即任务失败，不再虚报已重建
- _ignored_rel 拆分 reported/snapshot 两级：磁盘快照仍跳过 .codex/.claude（churn 控制），worker 显式报告的该目录路径作为有意项目配置修改予以保留
---
 src/agent_bridge/adapters/acp.py | 37 ++++++++++++++++++--------------
 src/agent_bridge/workspace.py    | 23 ++++++++++++++------
 2 files changed, 37 insertions(+), 23 deletions(-)

diff --git a/src/agent_bridge/adapters/acp.py b/src/agent_bridge/adapters/acp.py
index d25a42b..f085bd3 100644
--- a/src/agent_bridge/adapters/acp.py
+++ b/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,22 +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)))
-                # 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.
-                live.pending_warnings.append(
-                    f"session/load failed ({type(exc).__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(exc)),
-                    },
-                    self.home,
-                )
+                load_failure = exc
             else:
                 self._remember_config_options(live, revived)
                 await self._sync_selection(live, session)
@@ -1672,6 +1658,25 @@ async def ensure_session(self, session: Session) -> None:
             "session/new",
             session,
         )
+        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,
+            )
         session.native_session_id = created.session_id
         self._remember_config_options(live, created)
         if self.agent.name == "grok":
diff --git a/src/agent_bridge/workspace.py b/src/agent_bridge/workspace.py
index 86a092e..5426bcf 100644
--- a/src/agent_bridge/workspace.py
+++ b/src/agent_bridge/workspace.py
@@ -122,7 +122,7 @@ def classify_changed_paths(
         if not raw or not isinstance(raw, str):
             continue
         rel, out = _classify_path(raw, root)
-        if rel and not _ignored_rel(rel) and rel not in seen_in:
+        if rel and not _ignored_rel(rel, reported=True) and rel not in seen_in:
             seen_in.add(rel)
             inside.append(rel)
         elif out and out not in seen_out:
@@ -166,13 +166,15 @@ def _classify_path(raw: str, root: Path) -> tuple[str | None, str | None]:
         return rel, None
     # Unresolvable: keep prior lenient behavior for relative strings, but a
     # leading ../ can never be proven inside — report it as outside instead
-    # of folding it into the workspace (E3).
+    # of folding it into the workspace (E3). Check the traversal BEFORE
+    # lstrip, which would otherwise eat the leading ../ as well.
     if not path.is_absolute():
-        rel = path.as_posix().lstrip("./")
+        posix = path.as_posix()
+        if posix.startswith("../") or posix == "..":
+            return None, posix
+        rel = posix.lstrip("./")
         if rel in {".", "..", ""}:
             return None, None
-        if rel.startswith("../") or rel == "..":
-            return None, path.as_posix()
         return rel, None
     return None, path.as_posix()
 
@@ -181,8 +183,15 @@ def _root_prefix_len(root: str) -> int:
     return len(root.rstrip(os.sep)) + 1
 
 
-def _ignored_rel(rel: str) -> bool:
-    return any(part in SKIP_DIR_NAMES for part in rel.split("/")[:-1])
+# Reported paths skip less than the disk snapshot: .codex/.claude churn is
+# filtered from the snapshot walker, but a path a worker *explicitly reports*
+# there is an intentional project-config edit and must surface (review).
+_REPORTED_SKIP = SKIP_DIR_NAMES - {".codex", ".claude"}
+
+
+def _ignored_rel(rel: str, *, reported: bool = False) -> bool:
+    names = _REPORTED_SKIP if reported else SKIP_DIR_NAMES
+    return any(part in names for part in rel.split("/")[:-1])
 
 
 def collect_update_paths(obj: object, into: set[str]) -> None:

From d752f1842f5f0aa772c016aa3682a3a79e81847f Mon Sep 17 00:00:00 2001
From: lulu feizhu <279467036+FeiZhuLulu@users.noreply.github.com>
Date: Tue, 6 Oct 2026 07:48:47 +0000
Subject: [PATCH 3/7] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=A4=8D=E5=AE=A1?=
 =?UTF-8?q?=E7=AC=AC=E4=BA=8C=E5=A4=84=E7=BC=BA=E9=99=B7=EF=BC=88=E7=AC=AC?=
 =?UTF-8?q?=E4=BA=8C=E8=BD=AE=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- _classify_path ../回退：返回 normpath(root/path) 的词法绝对路径——files_outside_cwd 契约是绝对路径，原样返回 '../x' 违约
- ensure_session：session/new 成功后立即落地 native_session_id，再做 append_event 等可失败簿记——转录写失败不再把已拒绝的旧 id 留在会话上
---
 src/agent_bridge/adapters/acp.py | 4 +++-
 src/agent_bridge/workspace.py    | 4 +++-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/src/agent_bridge/adapters/acp.py b/src/agent_bridge/adapters/acp.py
index f085bd3..32e7cbd 100644
--- a/src/agent_bridge/adapters/acp.py
+++ b/src/agent_bridge/adapters/acp.py
@@ -1658,6 +1658,9 @@ 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
@@ -1677,7 +1680,6 @@ async def ensure_session(self, session: Session) -> None:
                 },
                 self.home,
             )
-        session.native_session_id = created.session_id
         self._remember_config_options(live, created)
         if self.agent.name == "grok":
             # Grok /new applies the _meta reasoningEffort but always lands on
diff --git a/src/agent_bridge/workspace.py b/src/agent_bridge/workspace.py
index 5426bcf..8a90913 100644
--- a/src/agent_bridge/workspace.py
+++ b/src/agent_bridge/workspace.py
@@ -171,7 +171,9 @@ def _classify_path(raw: str, root: Path) -> tuple[str | None, str | None]:
     if not path.is_absolute():
         posix = path.as_posix()
         if posix.startswith("../") or posix == "..":
-            return None, posix
+            # files_outside_cwd is an absolute-path field: return the lexical
+            # absolute form (root-joined, normalized, no symlink resolution).
+            return None, Path(os.path.normpath(root / path)).as_posix()
         rel = posix.lstrip("./")
         if rel in {".", "..", ""}:
             return None, None

From a6406d196030038bec395bdad521938a6e3a53e9 Mon Sep 17 00:00:00 2001
From: lulu feizhu <279467036+FeiZhuLulu@users.noreply.github.com>
Date: Tue, 6 Oct 2026 07:52:42 +0000
Subject: [PATCH 4/7] =?UTF-8?q?=E9=A1=B9=E7=9B=AE=E5=86=85=20.codex/.claud?=
 =?UTF-8?q?e=20=E6=94=B9=E5=8A=A8=E6=81=A2=E5=A4=8D=E4=B8=8A=E6=8A=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

项目 cwd 内的 .codex/.claude 是合法项目配置（如 .codex/config.toml），快照跳过会让 worker 的真实配置修改从 files_changed 消失；worker 自身状态目录位于其 home、不在 cwd 内，不受影响。SKIP_DIR_NAMES 只保留 .agent-bridge（bridge 簿记目录），随之回收 _REPORTED_SKIP 拆分
---
 src/agent_bridge/workspace.py | 24 +++++++++---------------
 tests/test_workspace.py       |  6 +++++-
 2 files changed, 14 insertions(+), 16 deletions(-)

diff --git a/src/agent_bridge/workspace.py b/src/agent_bridge/workspace.py
index 8a90913..692e900 100644
--- a/src/agent_bridge/workspace.py
+++ b/src/agent_bridge/workspace.py
@@ -33,12 +33,13 @@
     ".parcel-cache",
     ".svelte-kit",
     ".terraform",
-    # Bridge/worker bookkeeping dirs. A worker CLI keeps its own state under
-    # these and churns them every turn; counting that as project changes
-    # misattributes Bridge-internal writes to the worker (H-06).
+    # Bridge bookkeeping dir. A nested bridge home under the project churns
+    # every turn; counting that as project changes misattributes Bridge-
+    # internal writes to the worker (H-06). `.codex`/`.claude` inside a
+    # project are NOT skipped: they hold legitimate project config (e.g.
+    # .codex/config.toml) whose edits must be reported; each worker's own
+    # state dirs live under its home, outside the workspace cwd.
     ".agent-bridge",
-    ".codex",
-    ".claude",
 }
 
 _PATH_KEYS = (
@@ -122,7 +123,7 @@ def classify_changed_paths(
         if not raw or not isinstance(raw, str):
             continue
         rel, out = _classify_path(raw, root)
-        if rel and not _ignored_rel(rel, reported=True) and rel not in seen_in:
+        if rel and not _ignored_rel(rel) and rel not in seen_in:
             seen_in.add(rel)
             inside.append(rel)
         elif out and out not in seen_out:
@@ -185,15 +186,8 @@ def _root_prefix_len(root: str) -> int:
     return len(root.rstrip(os.sep)) + 1
 
 
-# Reported paths skip less than the disk snapshot: .codex/.claude churn is
-# filtered from the snapshot walker, but a path a worker *explicitly reports*
-# there is an intentional project-config edit and must surface (review).
-_REPORTED_SKIP = SKIP_DIR_NAMES - {".codex", ".claude"}
-
-
-def _ignored_rel(rel: str, *, reported: bool = False) -> bool:
-    names = _REPORTED_SKIP if reported else SKIP_DIR_NAMES
-    return any(part in names for part in rel.split("/")[:-1])
+def _ignored_rel(rel: str) -> bool:
+    return any(part in SKIP_DIR_NAMES for part in rel.split("/")[:-1])
 
 
 def collect_update_paths(obj: object, into: set[str]) -> None:
diff --git a/tests/test_workspace.py b/tests/test_workspace.py
index 4816f0c..4f65502 100644
--- a/tests/test_workspace.py
+++ b/tests/test_workspace.py
@@ -134,7 +134,11 @@ def test_bridge_dirs_are_skipped_in_snapshot(tmp_path: Path):
         sub.mkdir()
         (sub / "state.json").write_text("{}\n", encoding="utf-8")
     (tmp_path / "real.py").write_text("x\n", encoding="utf-8")
-    assert set(snapshot_workspace(tmp_path)) == {"real.py"}
+    found = set(snapshot_workspace(tmp_path))
+    # .agent-bridge churn is bridge bookkeeping; .codex/.claude inside the
+    # project hold legitimate config (e.g. .codex/config.toml) and must be
+    # reported (review: worker state dirs live under home, outside cwd).
+    assert found == {"real.py", ".codex/state.json", ".claude/state.json"}
 
 
 @pytest.mark.skipif(os.name != "nt", reason="Windows junction repro")

From ac51e17065708e9ae799a87a30d02eb5748a5c61 Mon Sep 17 00:00:00 2001
From: lulu feizhu <279467036+FeiZhuLulu@users.noreply.github.com>
Date: Tue, 6 Oct 2026 07:56:38 +0000
Subject: [PATCH 5/7] =?UTF-8?q?junction=20=E6=A3=80=E6=B5=8B=E5=85=BC?=
 =?UTF-8?q?=E5=AE=B9=20Python=203.11?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

DirEntry.is_junction 是 3.12 新增；3.11 上回落恒 False 导致 junction 被遍历（CI windows-3.11 挂）。改用 st_file_attributes & FILE_ATTRIBUTE_REPARSE_POINT——跨版本可用的重解析点检测
---
 src/agent_bridge/workspace.py | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/src/agent_bridge/workspace.py b/src/agent_bridge/workspace.py
index 692e900..6912e0d 100644
--- a/src/agent_bridge/workspace.py
+++ b/src/agent_bridge/workspace.py
@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import os
+import stat
 from pathlib import Path
 
 SKIP_DIR_NAMES = {
@@ -73,9 +74,16 @@ def snapshot_workspace(cwd: str | Path) -> dict[str, tuple[int, int]]:
                         if entry.is_dir(follow_symlinks=False):
                             # Windows junctions pass is_dir(follow_symlinks=False)
                             # but point outside cwd; traversing them attributes
-                            # foreign files as workspace writes (E2).
+                            # foreign files as workspace writes (E2). The
+                            # DirEntry.is_junction method needs Python 3.12;
+                            # the reparse-point attribute works everywhere.
                             is_junction = getattr(entry, "is_junction", lambda: False)()
-                            if entry.name not in SKIP_DIR_NAMES and not is_junction:
+                            st = entry.stat(follow_symlinks=False)
+                            is_reparse = bool(
+                                getattr(st, "st_file_attributes", 0)
+                                & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
+                            )
+                            if entry.name not in SKIP_DIR_NAMES and not is_junction and not is_reparse:
                                 stack.append(entry.path)
                         elif entry.is_file(follow_symlinks=False):
                             st = entry.stat(follow_symlinks=False)

From 08ab3013f84817861a4c6abc097140e06a13d90c Mon Sep 17 00:00:00 2001
From: lulu feizhu <279467036+FeiZhuLulu@users.noreply.github.com>
Date: Tue, 6 Oct 2026 07:59:20 +0000
Subject: [PATCH 6/7] =?UTF-8?q?junction=20=E5=88=A4=E6=8D=AE=E6=94=B6?=
 =?UTF-8?q?=E7=AA=84=E4=B8=BA=E6=8C=82=E8=BD=BD=E7=82=B9=20tag?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

REPARSE_POINT 属性会连 OneDrive 云目录等非 junction 一起跳过，丢失其下真实改动。改用 os.path.ismount——仅匹配 IO_REPARSE_TAG_MOUNT_POINT（junction/卷挂载），云重解析目录照常遍历；全版本可用
---
 src/agent_bridge/workspace.py | 18 +++++++-----------
 1 file changed, 7 insertions(+), 11 deletions(-)

diff --git a/src/agent_bridge/workspace.py b/src/agent_bridge/workspace.py
index 6912e0d..5b7a18a 100644
--- a/src/agent_bridge/workspace.py
+++ b/src/agent_bridge/workspace.py
@@ -7,7 +7,6 @@
 from __future__ import annotations
 
 import os
-import stat
 from pathlib import Path
 
 SKIP_DIR_NAMES = {
@@ -74,16 +73,13 @@ def snapshot_workspace(cwd: str | Path) -> dict[str, tuple[int, int]]:
                         if entry.is_dir(follow_symlinks=False):
                             # Windows junctions pass is_dir(follow_symlinks=False)
                             # but point outside cwd; traversing them attributes
-                            # foreign files as workspace writes (E2). The
-                            # DirEntry.is_junction method needs Python 3.12;
-                            # the reparse-point attribute works everywhere.
-                            is_junction = getattr(entry, "is_junction", lambda: False)()
-                            st = entry.stat(follow_symlinks=False)
-                            is_reparse = bool(
-                                getattr(st, "st_file_attributes", 0)
-                                & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
-                            )
-                            if entry.name not in SKIP_DIR_NAMES and not is_junction and not is_reparse:
+                            # foreign files as workspace writes (E2).
+                            # DirEntry.is_junction needs Python 3.12, so fall
+                            # back to ismount: it matches the mount-point
+                            # reparse tag only, not cloud (OneDrive) reparse
+                            # dirs that legitimately belong to the workspace.
+                            is_junction = getattr(entry, "is_junction", lambda: False)() or os.path.ismount(entry.path)
+                            if entry.name not in SKIP_DIR_NAMES and not is_junction:
                                 stack.append(entry.path)
                         elif entry.is_file(follow_symlinks=False):
                             st = entry.stat(follow_symlinks=False)

From d2b536ab08bf9ce52f71ea0d27bc864336e81fa0 Mon Sep 17 00:00:00 2001
From: lulu feizhu <279467036+FeiZhuLulu@users.noreply.github.com>
Date: Tue, 6 Oct 2026 08:02:19 +0000
Subject: [PATCH 7/7] =?UTF-8?q?junction=20=E5=88=A4=E6=8D=AE=E6=94=B9?=
 =?UTF-8?q?=E4=B8=BA=20realpath=20=E7=88=B6=E7=9B=AE=E5=BD=95=E6=AF=94?=
 =?UTF-8?q?=E5=AF=B9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

os.path.ismount 对目录 junction 返回 False（只认卷挂载），且 POSIX 上误伤真实挂载目录。改用 DirEntry.is_junction（3.12+）或 realpath(entry) vs realpath(dirname)/name 比对：junction 解析到别处，普通目录/云重解析目录/POSIX 挂载解析为自身、照常遍历
---
 src/agent_bridge/workspace.py | 20 ++++++++++++++------
 1 file changed, 14 insertions(+), 6 deletions(-)

diff --git a/src/agent_bridge/workspace.py b/src/agent_bridge/workspace.py
index 5b7a18a..809f9d2 100644
--- a/src/agent_bridge/workspace.py
+++ b/src/agent_bridge/workspace.py
@@ -71,14 +71,22 @@ def snapshot_workspace(cwd: str | Path) -> dict[str, tuple[int, int]]:
                 for entry in it:
                     try:
                         if entry.is_dir(follow_symlinks=False):
-                            # Windows junctions pass is_dir(follow_symlinks=False)
+                            # Junctions/mounts pass is_dir(follow_symlinks=False)
                             # but point outside cwd; traversing them attributes
                             # foreign files as workspace writes (E2).
-                            # DirEntry.is_junction needs Python 3.12, so fall
-                            # back to ismount: it matches the mount-point
-                            # reparse tag only, not cloud (OneDrive) reparse
-                            # dirs that legitimately belong to the workspace.
-                            is_junction = getattr(entry, "is_junction", lambda: False)() or os.path.ismount(entry.path)
+                            # DirEntry.is_junction needs Python 3.12; elsewhere
+                            # compare realpath against the canonicalized parent:
+                            # a junction resolves elsewhere while normal dirs,
+                            # cloud (OneDrive) reparse dirs and POSIX mounts
+                            # resolve to themselves and stay traversable.
+                            if hasattr(entry, "is_junction"):
+                                is_junction = entry.is_junction()
+                            else:
+                                expected = os.path.join(
+                                    os.path.realpath(os.path.dirname(entry.path)),
+                                    entry.name,
+                                )
+                                is_junction = os.path.realpath(entry.path) != expected
                             if entry.name not in SKIP_DIR_NAMES and not is_junction:
                                 stack.append(entry.path)
                         elif entry.is_file(follow_symlinks=False):