fix: harden isolation, elision, and concurrency per adversarial arc review

A 16-agent adversarial review (4 dimensions, every finding refuted-or-
confirmed against live code) confirmed 11 findings; all fixed except one
deliberate deferral (exclusive gate held across approval waits — needs
the approval-split refactor; recorded in tasks/todo.md).

- DATA LOSS (high): a child that committed its work left a clean
  worktree, so cleanup removed it and orphaned the commits. Creation now
  records a base-SHA sidecar (next to the worktree, never inside it);
  commits ahead of base count as changes and force retention, with
  unknown provenance failing closed to retention.
- FALSE ISOLATION (high): foreground shell inherited the process cwd and
  relative file-tool paths resolved against it, so isolated children
  mutated the original repo. Host exec (protocol, local, ssh, ACP
  fallback) gained a cwd argument; foreground shell passes the runtime
  work dir, and write/replace/read resolve relative paths against it
  while preserving the relative-escape error contract.
- REGRESSION (high): safe mode now disables the read-only-command prompt
  elision — users who disabled auto-approval keep every checkpoint.
- REGRESSION (high): untrusted-project hook stripping now publishes a
  session notification (web/ACP visible), not just a shell log line.
- MCP readOnlyHint annotations enable supports_parallel via property;
  worktree add/remove serializes per repo; CHANGELOG documents the
  same-step serialization and stderr-diagnostics behavior changes.

Author: elkaix

CHANGELOG.md

@@ -15,6 +15,7 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- **Adversarial branch review hardening.** A multi-agent review pass confirmed and fixed: committed-but-clean isolation worktrees are now retained (commits ahead of the creation base count as changes — they were previously orphaned on cleanup); foreground shell commands and relative-path file edits now resolve against the agent's work dir, so worktree isolation actually binds them (host exec gained a `cwd` argument); safe mode now also disables the read-only-command prompt elision; MCP tools annotated `readOnlyHint` run in parallel in the same-step gate; worktree add/remove serializes per repo. Behavior notes: same-step tool calls without `supports_parallel` now serialize deterministically (previously fully concurrent), and text-mode error diagnostics moved to stderr — capture `2>&1` or use `--output-format stream-json` if you scraped stdout.
 - **The agent now knows its own permission posture.** A live permissions-state reminder renders the enforced profile, safe-mode/yolo/auto flags, mutation/network allowances, session-approved actions, and the shell gate's command-shaping rules — re-emitted exactly when the posture changes (/yolo, /auto, /trust, new approvals) instead of the model discovering policy through denied tool calls.
 - **Edits recover from whitespace and smart-punctuation drift.** StrReplaceFile no longer hard-fails with "old string not found" when the only mismatch is trailing whitespace, indentation, or smart quotes/dashes: a graduated line-window ladder relocates the edit, replaces the actual file slice (preserving CRLF endings), and names the relaxation it used in the tool message. Multiple fuzzy hits without replace_all still error, so ambiguity is never silently resolved.
 - **`isolation="worktree"` is now enforced for background write agents.** Previously it only recorded intent, so parallel coders shared one working tree and could clobber each other. A write-profile child now runs in its own git worktree of HEAD; its final report names the worktree path with a diff summary so changes merge deliberately, clean worktrees are removed, non-git roots fail with an actionable error, and read-profile children ignore the request.

packages/pythinker-host/src/pythinker_host/__init__.py

@@ -220,14 +220,18 @@ async def mkdir(
         """Create a directory at the given path."""
         ...
 
-    async def exec(self, *args: str, env: Mapping[str, str] | None = None) -> HostProcess:
+    async def exec(
+        self, *args: str, env: Mapping[str, str] | None = None, cwd: str | None = None
+    ) -> HostProcess:
         """
         Execute a command with arguments and return the running process.
 
         Args:
             *args: Command and its arguments.
             env: Environment variables for the subprocess. If None, inherits
                  from the parent process.
+            cwd: Working directory for the subprocess. If None, inherits the
+                 backend's current working directory (process cwd locally).
         """
         ...
 
@@ -347,8 +351,10 @@ async def mkdir(path: StrOrHostPath, parents: bool = False, exist_ok: bool = Fal
     return await get_current_host().mkdir(path, parents=parents, exist_ok=exist_ok)
 
 
-async def exec(*args: str, env: Mapping[str, str] | None = None) -> HostProcess:
-    return await get_current_host().exec(*args, env=env)
+async def exec(
+    *args: str, env: Mapping[str, str] | None = None, cwd: str | None = None
+) -> HostProcess:
+    return await get_current_host().exec(*args, env=env, cwd=cwd)
 
 
 from pythinker_host._current import current_host as current_host  # noqa: E402

packages/pythinker-host/src/pythinker_host/local.py

@@ -190,7 +190,9 @@ async def mkdir(
         local_path = path.unsafe_to_local_path() if isinstance(path, HostPath) else Path(path)
         await asyncio.to_thread(local_path.mkdir, parents=parents, exist_ok=exist_ok)
 
-    async def exec(self, *args: str, env: Mapping[str, str] | None = None) -> HostProcess:
+    async def exec(
+        self, *args: str, env: Mapping[str, str] | None = None, cwd: str | None = None
+    ) -> HostProcess:
         if not args:
             raise ValueError("At least one argument (the program to execute) is required.")
 
@@ -208,6 +210,7 @@ async def exec(self, *args: str, env: Mapping[str, str] | None = None) -> HostPr
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
             env=env,
+            cwd=cwd,
             **process_options,
         )
         return self.Process(process)

packages/pythinker-host/src/pythinker_host/ssh.py

@@ -303,7 +303,9 @@ async def mkdir(
                 raise FileExistsError(f"{path} already exists")
             await self._sftp.mkdir(str(path))
 
-    async def exec(self, *args: str, env: Mapping[str, str] | None = None) -> HostProcess:
+    async def exec(
+        self, *args: str, env: Mapping[str, str] | None = None, cwd: str | None = None
+    ) -> HostProcess:
         if not args:
             raise ValueError("At least one argument (the program to execute) is required.")
         command = " ".join(shlex.quote(arg) for arg in args)
@@ -313,8 +315,9 @@ async def exec(self, *args: str, env: Mapping[str, str] | None = None) -> HostPr
         #   cwd before running the command.
         #
         # This is intentionally strict: if cwd doesn't exist, the command fails.
-        if self._cwd:
-            command = f"cd {shlex.quote(self._cwd)} && {command}"
+        effective_cwd = cwd or self._cwd
+        if effective_cwd:
+            command = f"cd {shlex.quote(effective_cwd)} && {command}"
         process = await self._connection.create_process(command, encoding=None, env=env)
         return self.Process(process)
 

src/pythinker_code/acp/host.py

@@ -296,8 +296,10 @@ async def mkdir(
     ) -> None:
         await self._fallback.mkdir(path, parents=parents, exist_ok=exist_ok)
 
-    async def exec(self, *args: str, env: Mapping[str, str] | None = None) -> HostProcess:
-        return await self._fallback.exec(*args, env=env)
+    async def exec(
+        self, *args: str, env: Mapping[str, str] | None = None, cwd: str | None = None
+    ) -> HostProcess:
+        return await self._fallback.exec(*args, env=env, cwd=cwd)
 
     def _abs_path(self, path: StrOrHostPath) -> str:
         host_path = path if isinstance(path, HostPath) else HostPath(path)

src/pythinker_code/app.py

@@ -388,6 +388,30 @@ async def create(
         from pythinker_code.hooks.engine import HookEngine
 
         hook_engine = HookEngine(config.hooks, cwd=str(session.work_dir))
+        if config.disabled_project_hooks:
+            # The load-time logger.warning only reaches shell users; publish a
+            # notification so web/ACP frontends also learn why their project
+            # hooks did not run and how to enable them.
+            from pythinker_code.notifications.models import NotificationEvent
+
+            runtime.notifications.publish(
+                NotificationEvent(
+                    id=f"project-hooks-disabled:{session.id}",
+                    category="system",
+                    type="project_hooks_disabled",
+                    source_kind="config",
+                    source_id="project_trust",
+                    title="Project hooks disabled (untrusted project)",
+                    body=(
+                        "Hooks defined in "
+                        + ", ".join(config.disabled_project_hooks)
+                        + " are disabled until you trust this project. Run /trust to "
+                        "enable them (takes effect on /reload or next start)."
+                    ),
+                    severity="warning",
+                    dedupe_key=f"project-hooks-disabled:{session.id}",
+                )
+            )
         soul.set_hook_engine(hook_engine)
         runtime.hook_engine = hook_engine
 

src/pythinker_code/config.py

@@ -379,6 +379,8 @@ def _read_toml(path: Path) -> dict[str, Any]:
     local_file: Path | None = None
     project_dict: dict[str, Any] = {}
     local_dict: dict[str, Any] = {}
+    project_trusted = True
+    stripped_hook_files: list[str] = []
 
     if project_root is not None:
         from pythinker_code.project_trust import is_project_trusted
@@ -406,6 +408,7 @@ def _read_toml(path: Path) -> dict[str, Any]:
                 local_dict = {}
             for scope_dict, scope_file in ((project_dict, project_file), (local_dict, local_file)):
                 if scope_dict.pop("hooks", None) is not None:
+                    stripped_hook_files.append(str(scope_file))
                     logger.warning(
                         "Project hooks in {file} are disabled until the project is "
                         "trusted; run /trust to enable them",
@@ -444,6 +447,8 @@ def _read_toml(path: Path) -> dict[str, Any]:
         raise ConfigError("Invalid configuration:\n" + "\n".join(enriched)) from exc
 
     # ── METADATA ──────────────────────────────────────────────────────────
+    if project_root is not None and not project_trusted:
+        config.disabled_project_hooks = stripped_hook_files
     config.is_from_default_location = True
     config.source_file = user_file
     if user_file.exists():
@@ -1103,6 +1108,15 @@ class Config(BaseModel):
     mcp: MCPConfig = Field(default_factory=MCPConfig, description="MCP configuration")
     tui: TUIConfig = Field(default_factory=TUIConfig, description="TUI rendering configuration")
     hooks: list[HookDef] = Field(default_factory=list, description="Hook definitions")  # pyright: ignore[reportUnknownVariableType]
+    disabled_project_hooks: list[str] = Field(
+        default_factory=list,
+        exclude=True,
+        description=(
+            "Config files whose project-scope hooks were stripped because the "
+            "project is untrusted. Populated at load; surfaced as a "
+            "notification so non-shell frontends see it too."
+        ),
+    )
     merge_all_available_skills: bool = Field(
         default=True,
         description=(

src/pythinker_code/soul/toolset.py

@@ -1296,6 +1296,16 @@ def mcp_server_name(self) -> str:
         """Name of the MCP server this tool belongs to."""
         return self._mcp_server_name
 
+    @property
+    def supports_parallel(self) -> bool:
+        """Honor the MCP readOnlyHint annotation in the same-step gate.
+
+        Read-only server tools (doc/resource lookups) may overlap instead of
+        serializing; anything unannotated stays exclusive (safe default).
+        """
+        annotations = getattr(self._mcp_tool, "annotations", None)
+        return bool(getattr(annotations, "readOnlyHint", False))
+
     async def __call__(self, *args: Any, **kwargs: Any) -> ToolReturnValue:
         # Call-time re-check of the list-time filter: defense in depth for
         # tool maps shared across agents (e.g. runtime.mcp_tools handed to

src/pythinker_code/subagents/worktree.py

@@ -9,11 +9,21 @@
 from __future__ import annotations
 
 import asyncio
+from collections import defaultdict
 from pathlib import Path
 
 from pythinker_code.utils.logging import logger
 
 _GIT_TIMEOUT_S = 30.0
+# Serialize worktree add/remove per repo: git's internal locking is reliable
+# on current versions, but concurrent isolated agents should not depend on it.
+_REPO_LOCKS: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
+
+
+def _base_sha_file(worktree: Path) -> Path:
+    # Sidecar NEXT TO the worktree, never inside it — an untracked file inside
+    # would make every clean worktree look dirty.
+    return worktree.parent / f"{worktree.name}.base-sha"
 
 
 class WorktreeError(Exception):
@@ -58,21 +68,58 @@ async def create_agent_worktree(repo_dir: Path, dest: Path) -> None:
         # Resume of an isolated agent reuses its existing worktree.
         return
     dest.parent.mkdir(parents=True, exist_ok=True)
-    code, _, stderr = await _git(["worktree", "add", "--detach", str(dest), "HEAD"], repo_dir)
-    if code != 0:
-        first_line = stderr.splitlines()[0] if stderr else "unknown git error"
-        raise WorktreeError(f"could not create isolation worktree at {dest}: {first_line}")
+    async with _REPO_LOCKS[str(repo_dir)]:
+        code, _, stderr = await _git(["worktree", "add", "--detach", str(dest), "HEAD"], repo_dir)
+        if code != 0:
+            first_line = stderr.splitlines()[0] if stderr else "unknown git error"
+            raise WorktreeError(f"could not create isolation worktree at {dest}: {first_line}")
+    # Record the creation base so committed-but-clean child work is detected
+    # later; commits ahead of this SHA must never be silently removed.
+    code, base_sha, _ = await _git(["rev-parse", "HEAD"], dest)
+    if code == 0 and base_sha:
+        _base_sha_file(dest).write_text(base_sha + "\n", encoding="utf-8")
 
 
 async def worktree_change_summary(worktree: Path) -> str:
-    """Short human summary of changes in *worktree*; empty string when clean."""
+    """Short human summary of changes in *worktree*; empty string when clean.
+
+    "Changes" includes commits the child made on its detached HEAD: a child
+    that commits its work leaves a clean working tree, and `worktree remove`
+    would orphan those commits as dangling objects.
+    """
+    parts: list[str] = []
+    commits_ahead = await _commits_ahead_of_base(worktree)
+    if commits_ahead:
+        parts.append(f"{commits_ahead} commit(s) ahead of the creation base")
     code, status, _ = await _git(["status", "--porcelain"], worktree)
-    if code != 0 or not status:
-        return ""
-    _, diff_stat, _ = await _git(["diff", "--stat", "HEAD"], worktree)
-    untracked = sum(1 for line in status.splitlines() if line.startswith("??"))
-    parts = [part for part in (diff_stat, f"{untracked} untracked file(s)" if untracked else "")]
-    return "\n".join(part for part in parts if part)
+    if code == 0 and status:
+        _, diff_stat, _ = await _git(["diff", "--stat", "HEAD"], worktree)
+        if diff_stat:
+            parts.append(diff_stat)
+        untracked = sum(1 for line in status.splitlines() if line.startswith("??"))
+        if untracked:
+            parts.append(f"{untracked} untracked file(s)")
+    return "\n".join(parts)
+
+
+async def _commits_ahead_of_base(worktree: Path) -> int:
+    """Commits on the worktree's detached HEAD since creation.
+
+    Missing or unreadable sidecar fails CLOSED (pretend one commit exists)
+    when HEAD cannot be compared — losing work is the only unacceptable
+    outcome, so unknown provenance means retain.
+    """
+    sidecar = _base_sha_file(worktree)
+    try:
+        base_sha = sidecar.read_text(encoding="utf-8").strip()
+    except OSError:
+        base_sha = ""
+    if not base_sha:
+        return 1  # unknown provenance — retain
+    code, count, _ = await _git(["rev-list", "--count", f"{base_sha}..HEAD"], worktree)
+    if code != 0:
+        return 1
+    return int(count or 0)
 
 
 async def cleanup_agent_worktree(repo_dir: Path, worktree: Path, *, has_changes: bool) -> str:
@@ -84,10 +131,12 @@ async def cleanup_agent_worktree(repo_dir: Path, worktree: Path, *, has_changes:
     """
     if has_changes:
         return "retained"
-    code, _, stderr = await _git(["worktree", "remove", str(worktree)], repo_dir)
+    async with _REPO_LOCKS[str(repo_dir)]:
+        code, _, stderr = await _git(["worktree", "remove", str(worktree)], repo_dir)
     if code != 0:
         logger.warning(
             "Could not remove clean isolation worktree {wt}: {err}", wt=worktree, err=stderr
         )
         return "retained"
+    _base_sha_file(worktree).unlink(missing_ok=True)
     return "removed"

src/pythinker_code/tools/file/read.py

@@ -111,7 +111,13 @@ async def __call__(self, params: Params) -> ToolReturnValue:
 
         try:
             raw = HostPath(params.path).expanduser()
-            p = raw.canonical()
+            # Relative tool paths resolve against the runtime work dir
+            # (override-aware), NOT the process cwd — an isolated child's
+            # relative write must land in its worktree. `raw` keeps the
+            # original form: the workspace-escape rule for relative paths
+            # checks (and reports) what the caller actually passed.
+            base_joined = raw if raw.is_absolute() else self._work_dir.joinpath(str(raw))
+            p = base_joined.canonical()
 
             # Resolve the real (symlink-followed) path for security checks only.
             # os.path.realpath follows symlinks at every component including the leaf,