diff --git a/ORCHESTRATION.md b/ORCHESTRATION.md
index 81c34c6..3554677 100644
--- a/ORCHESTRATION.md
+++ b/ORCHESTRATION.md
@@ -62,7 +62,7 @@ In `auto`/`eager`, tell the user after the fact. In `manual`, their explicit req
    - Devin CLI: advertised ids (`devin models list`; level is in the id, e.g. `swe-1-7-medium`). Unknown id fails; `effort` warns and is ignored. Mode forced to `bypass`. Revive via `session/load` (replays into `get_transcript`; `get_result` stays clean).
    - ZCode: `provider\model` or a unique bare id. Thought: `off`→`low`, `medium`→`high`. Unknown or ambiguous model fails; unmappable effort warns. Mode `yolo`. Revive via `session/resume`.
    - MiniMax Code: `provider/model` or `provider/model#variant`. `thinkingEffort` (`off`→`default`). Unknown model fails; unmappable effort warns. Permission `bypassPermissions`. Revive via `session/resume`.
-3. Loop `wait_task` until terminal. A timeout is **not** failure — call it again. `silent_for_sec` is time since the last output. Silence past `stall_timeout_sec` (default 1800, per worker, 0 disables) ends `failed` / `stalled`: raise that worker's limit or resume the `session_id` with a narrower task. Stall counts only worker output — a chatty turn never stalls; `turn_timeout_sec` (per worker, 0 = off) caps a turn's wall-clock and ends `failed` / `timed_out`. Terminal `wait_task` carries `persisted`; `false` means `state.json` was not written — the result stays valid but may not survive a restart. `files_changed` lists only paths inside the task `cwd`; outside paths go to `files_outside_cwd` (absolute, capped alike, `files_outside_cwd_total`). A `session/load`-failed warning means the worker started a fresh native session — re-supply context next message. Size `timeout_sec` under the host MCP tool timeout:
+3. Loop `wait_task` until terminal. A timeout is **not** failure — call it again. `silent_for_sec` is time since the last output. Silence past `stall_timeout_sec` (default 1800, per worker, 0 disables) ends `failed` / `stalled`: raise that worker's limit or resume the `session_id` with a narrower task. Stall counts only worker output — a chatty turn never stalls; `turn_timeout_sec` (per worker, 0 = off) caps a turn's wall-clock and ends `failed` / `timed_out`. `dispatch_task` and every terminal task answer (`wait_task` / `check_task` / `get_result` / `cancel_task`) carry `persisted`; `false` means `state.json` was not written — the answer stays valid but may not survive a restart. `files_changed` lists only paths inside the task `cwd`; outside paths go to `files_outside_cwd` (absolute, capped alike, `files_outside_cwd_total`). A `session/load`-failed warning means the worker started a fresh native session — re-supply context next message. Size `timeout_sec` under the host MCP tool timeout:
    - Codex: `tool_timeout_sec` 600; default 180 is fine.
    - Cursor: host ~45–60 s; pass ~30 and loop.
    - Kimi Code: configure `toolTimeoutMs` 600000; otherwise ~45 s polls.
diff --git a/ORCHESTRATION.zh-CN.md b/ORCHESTRATION.zh-CN.md
index 0565185..177f31f 100644
--- a/ORCHESTRATION.zh-CN.md
+++ b/ORCHESTRATION.zh-CN.md
@@ -72,7 +72,7 @@ Claude 的 `quota.status` 只汇总通用的 `5h` / `weekly` 额度。派发前
 1. 先调 `list_agents`，选一个可用的 Worker。看返回里的 `coordinator.mode` / `coordinator.instructions` / `coordinator.dispatch_enabled`（用户偏好优先于第二步的默认分工），以及 `env.proxy` / `env.warnings`。`dispatch_enabled` 为 false 时立刻停止派发。直连网络上 proxy 为 null 是正常的；若机器需要代理而 Worker 报网络/连接错误，去改代理配置（仓库 `agents.toml` 或 `~/.agent-bridge/agents.toml` 里的 `[env.proxy]`），不要对同一次失败反复重试。
 2. `dispatch_task` 的 `cwd` 必须是 **本次对话的项目目录**（绝对路径）。也就是你被启动时所在的文件夹——和用户自己 `cd` 再运行 `grok` / `kimi` / `agy` / `opencode` / `claude` 是同一个地方。Worker 的会话历史按 cwd 存放，换目录就是对方 UI 里的另一场对话，之后也更难看、更难续。**不要**把 Agent Bridge 的安装目录当作 cwd，除非用户正在改 Bridge 本身。不要自造临时目录。`message` 必须自洽：背景、绝对路径、验收标准、不要做什么。model 和 effort：没有特别理由就都不传，用 Worker 默认值——Antigravity 默认的 `gemini-3.7-flash` 就够了（要换才用 `agy models` 里的 slug）。确要指定时：Grok 的 `model` 是 `grok models` 里的 slug（用账号目录里有的，不要编），`effort` 为 `off` / `low` / `medium` / `high` / `max`（`off` → Grok `none`，`max` → Grok `xhigh`）。Grok 的 `/new` 总会落在活动默认模型上（目前是 grok-4.6 xhigh）；Bridge 在会话建立后再 `session/setModel`。Kimi 的 `model` 取会话自己声明的那几个 slug（`kimi-code/k3`、`kimi-code/k3-256k`、`kimi-code/kimi-for-coding` 等），`effort` 还是那五个词，由 Bridge 映射到该模型声明的思考档位——k3 只声明 `low` / `high` / `max`，所以 `medium` 落到 `high`，`off` 落到 `low`。传了会话没声明的 slug 会让这一轮失败，错误里会列出真正可用的；`effort` 映射不上则只出 warning，不算失败。OpenCode 的 `model` 是会话声明的 `provider/model`（官方 OpenCode Zen / Go，或用 `opencode auth` 以 API key 接上的其它提供商——没有产品级登录）。`effort` 还是那五个词，映射到该模型的 variant（常见是 `default|low|medium|high`，`max` 通常落到 `high`）。未声明的 slug 会让这一轮失败；没有 effort 选项或映射不上只出 warning。`get_result.observed_model` / `observed_effort` 是 Bridge 映射后成功设上的最后值（例如 `max` → `high`），不是实时采样器转储。同一会话换模型会重新设 effort，因为 OpenCode 会把 variant 重置成新模型的默认。Bridge 用 `session/resume` 复活 OpenCode，不用 `session/load`。Claude Code 的 `model` 取会话声明的 slug（`sonnet` / `opus` / `haiku` 或完整 id），`effort` 还是那五个词（`off` → `default`，`max` → `xhigh`，除非会话列出了 `max`）。未声明的 slug 会让这一轮失败；没有 effort 选项或映射不上只出 warning。Bridge 在 `session/new` 后把模式切到 `bypassPermissions`，并用 `session/resume` 复活（`session/load` 会回放整段历史）。
    Cursor 的 `model` 必须是 `cursor-agent --list-models` 当前列出的完整 ID；Bridge 会校验并固定首次启动，再把完整 ID 映射到 Cursor 会话声明的 ACP `model` 和参数选项。同一个 `session_id` 可以切换模型和变体；若所选模型声明了思考档位，单独传入的 `effort` 会覆盖 ID 中编码的档位。`observed_model` 是 Cursor 确认映射结果后的请求 ID，`observed_effort` 是 Cursor 确认的思考档位；两者都不是实时采样器。Codex CLI 的 `model` 是 Codex 自己的 slug（如 `gpt-5.6-sol`），`effort` 为 `off` / `low` / `medium` / `high` / `max`（`off` → `none`）；默认 `--approve-for-me`，提示词走 stdin，用 `exec resume` 续会话；JSONL 建立前的启动失败会进入 `get_result.error`。Devin CLI 的 `model` 取会话声明的模型 id（`devin models list`；档位就在 id 里，如 `swe-1-7-medium`、`claude-opus-5-high`），未声明的 id 会让这一轮失败；`effort` 会被忽略并出 warning。Bridge 在 `session/new` 后把模式切到 `bypass`；续会话走 `session/load`，旧历史会被回放进 `get_transcript`（新一轮的 `get_result` 不受影响）。ZCode 的 `model` 是 `provider\model`，或在只对应一个选项时用裸模型 id（如 `GLM-5.3`）；`effort` 映射到该模型的 thought 档（常见 `low|high|max`，`off` → `low`，`medium` → `high`）。未声明或有歧义的模型会让这一轮失败；映射不上只出 warning。Bridge 把模式切到 `yolo`，并用 `session/resume` 复活。MiniMax Code 的 `model` 是 `provider/model` 或 `provider/model#variant`；`effort` 映射到 `thinkingEffort`（`off` → `default`）。未声明的模型会让这一轮失败；映射不上只出 warning。Bridge 把 `permissionMode` 切到 `bypassPermissions`，并用 `session/resume` 复活。DSH 的 `model` 是 `provider/model` 或模型 id，`effort` 为 `off` / `low` / `high` / `max`。新版 `dsh --profile acp` 通过 `session/set_config_option` 在同一会话内切换——传了会话没声明的 model 会让这一轮失败，`effort` 映射不上只出 warning；旧版 `dsh-acp-demo` 走启动环境变量，改 model/effort 会重启进程（进程内不能切换），失败后换模型同样靠下次 `dispatch_task` 重启进程。
-3. 循环调用 `wait_task` 直到 `status` 为终态。超时不是失败，再调一次即可。`wait_task` / `check_task` 还会返回 `silent_for_sec`（Worker 最近一次输出到现在的秒数）。Worker 持续静默超过 `stall_timeout_sec`（默认 1800，可在 `agents.toml` 按 Worker 设置，0 关闭）时 Bridge 会取消这一轮，任务以 `status=failed`、`stop_reason="stalled"` 结束。静默判定只看 Worker 输出——一直在产生事件的回合永远不会 stall。`turn_timeout_sec`（按 Worker 设置，默认 0 关闭）则给整轮设墙钟上限，超时以 `failed` / `timed_out` 结束。终态 `wait_task` 还带 `persisted`；为 `false` 表示 `state.json` 没写成功——结果仍然有效，但 Bridge 重启后可能丢失。`files_changed` 只列任务 `cwd` 内的路径；Worker 上报的 cwd 外路径进 `files_outside_cwd`（绝对路径，与 `files_changed` 同样截断，总数见 `files_outside_cwd_total`）。任务 warnings 里说 `session/load` 失败、先前上下文已丢失，表示 Worker 开了一个新的原生会话——下一条消息要重新带上上下文。长时间静默的构建和挂死的 Worker 看起来一样：如果那一步确实需要那么久，就调高该 Worker 的上限；否则在同一个 `session_id` 上派一个更小的任务。`timeout_sec` 必须低于宿主的 MCP 工具超时：Codex 配置 `tool_timeout_sec` 600，默认 180 即可；Cursor 宿主大约 45–60 秒，传约 30 并循环；Kimi Code 把 `toolTimeoutMs` 配到 600000，未配置则约 45 秒短轮询；ZCode 把 `timeoutMs` 配到 600000，未配置则约 15–20 秒短轮询；Grok Build 官方默认 `tool_timeout_sec` 是 6000，建议显式写成 600，拿不准或调用被掐掉时用约 30–45 秒短轮询；Claude Code 在 `.mcp.json` 里把每服务器 `timeout` 配成 600000（毫秒），CLI 默认很长，桌面端历史上大约 60 秒会掐掉，拿不准时用约 45 秒短轮询。等待期间可以并行做别的事。
+3. 循环调用 `wait_task` 直到 `status` 为终态。超时不是失败，再调一次即可。`wait_task` / `check_task` 还会返回 `silent_for_sec`（Worker 最近一次输出到现在的秒数）。Worker 持续静默超过 `stall_timeout_sec`（默认 1800，可在 `agents.toml` 按 Worker 设置，0 关闭）时 Bridge 会取消这一轮，任务以 `status=failed`、`stop_reason="stalled"` 结束。静默判定只看 Worker 输出——一直在产生事件的回合永远不会 stall。`turn_timeout_sec`（按 Worker 设置，默认 0 关闭）则给整轮设墙钟上限，超时以 `failed` / `timed_out` 结束。`dispatch_task` 和终态任务响应（`wait_task` / `check_task` / `get_result` / `cancel_task`）都带 `persisted`；为 `false` 表示 `state.json` 没写成功——结果仍然有效，但 Bridge 重启后可能丢失。`files_changed` 只列任务 `cwd` 内的路径；Worker 上报的 cwd 外路径进 `files_outside_cwd`（绝对路径，与 `files_changed` 同样截断，总数见 `files_outside_cwd_total`）。任务 warnings 里说 `session/load` 失败、先前上下文已丢失，表示 Worker 开了一个新的原生会话——下一条消息要重新带上上下文。长时间静默的构建和挂死的 Worker 看起来一样：如果那一步确实需要那么久，就调高该 Worker 的上限；否则在同一个 `session_id` 上派一个更小的任务。`timeout_sec` 必须低于宿主的 MCP 工具超时：Codex 配置 `tool_timeout_sec` 600，默认 180 即可；Cursor 宿主大约 45–60 秒，传约 30 并循环；Kimi Code 把 `toolTimeoutMs` 配到 600000，未配置则约 45 秒短轮询；ZCode 把 `timeoutMs` 配到 600000，未配置则约 15–20 秒短轮询；Grok Build 官方默认 `tool_timeout_sec` 是 6000，建议显式写成 600，拿不准或调用被掐掉时用约 30–45 秒短轮询；Claude Code 在 `.mcp.json` 里把每服务器 `timeout` 配成 600000（毫秒），CLI 默认很长，桌面端历史上大约 60 秒会掐掉，拿不准时用约 45 秒短轮询。等待期间可以并行做别的事。
 4. 调 `get_result`；若 `has_more` 为 true，就继续用 `cursor=next_cursor` 读取并拼接所有分页。然后自己看 `git status` / `git diff`。该编的、该测的你来跑。不要只信 Worker 的自我汇报。Grok 的模型和 effort 以 `get_result.model` / `get_result.observed_model`（以及对应的 effort 字段）为准。`observed_model` 来自 Grok `events.jsonl` 的 `turn_started.model_id`。不要因为 Worker 说「I am Grok 4.6」或引用了 `You are Grok 4.6` 就判定切模型成败——那句横幅是 `/new` 写进系统提示的，`session/setModel` 换采样器时不会改它。Kimi 从不把失败的回合报成失败：它的 ACP 宿主会把失败回合映射成空文本的 `end_turn`，于是配额或服务端报错看起来和干净的空转一模一样。Kimi 交回空回合就要当可疑对待，去看 `get_result.warnings`——Bridge 每轮结束都会读 Kimi 的 `wire.jsonl`，把真实原因放在那里，旁边就是 `observed_model` / `observed_effort`。
 5. 验收不通过时，用同一个 `session_id` 再 `dispatch_task`，并列出具体问题。最多跟进三轮。之后你自己改，并告诉用户。
 6. 结束后总结 diff、剩余风险、用了哪个 Worker。不再需要时调用 `end_session`。
diff --git a/skills/agent-bridge/SKILL.md b/skills/agent-bridge/SKILL.md
index e553ca1..440fb3c 100644
--- a/skills/agent-bridge/SKILL.md
+++ b/skills/agent-bridge/SKILL.md
@@ -45,7 +45,7 @@ Examples: "fix the README typo" → yourself. "Add a None check at line 120" →
 ## The dispatch loop
 
 1. `dispatch_task` with `cwd` = **this conversation's project folder** (absolute) — never the Agent Bridge install path, never a temp dir. The `message` must be self-contained: background, absolute paths, acceptance criteria, things not to do. Leave `model`/`effort` unset unless you have a reason; slugs and effort mappings per worker are documented in ORCHESTRATION.md in the Agent-Bridge repo.
-2. Loop `wait_task` until terminal. A timeout is **not** failure — call it again. `silent_for_sec` in the payload is the time since the worker's last output; a turn silent past `stall_timeout_sec` (default 1800) ends `failed` / `stop_reason="stalled"` — raise that worker's limit in `agents.toml` if the step was legitimately long. Stall counts only worker output; `turn_timeout_sec` (default 0 = off) instead caps a turn's wall-clock and ends `failed` / `timed_out`. A terminal `wait_task` carries `persisted`; `false` means `state.json` could not be written — the result is still valid but may not survive a restart. `files_changed` lists only paths inside the task `cwd`; outside-cwd paths land in `files_outside_cwd` (absolute, capped the same way, `files_outside_cwd_total`). A warning that `session/load` failed and prior context is lost means a fresh native session — re-supply context next message. Size `timeout_sec` under the host MCP tool timeout:
+2. Loop `wait_task` until terminal. A timeout is **not** failure — call it again. `silent_for_sec` in the payload is the time since the worker's last output; a turn silent past `stall_timeout_sec` (default 1800) ends `failed` / `stop_reason="stalled"` — raise that worker's limit in `agents.toml` if the step was legitimately long. Stall counts only worker output; `turn_timeout_sec` (default 0 = off) instead caps a turn's wall-clock and ends `failed` / `timed_out`. `dispatch_task` and every terminal task answer (`wait_task` / `check_task` / `get_result` / `cancel_task`) carry `persisted`; `false` means `state.json` could not be written — the answer is still valid but may not survive a restart. `files_changed` lists only paths inside the task `cwd`; outside-cwd paths land in `files_outside_cwd` (absolute, capped the same way, `files_outside_cwd_total`). A warning that `session/load` failed and prior context is lost means a fresh native session — re-supply context next message. Size `timeout_sec` under the host MCP tool timeout:
    - Codex: `tool_timeout_sec` 600; default 180 is fine.
    - Cursor: host ~45–60 s; pass ~30 and loop.
    - Kimi Code: configure `toolTimeoutMs` 600000; otherwise ~45 s polls.
diff --git a/src/agent_bridge/processes.py b/src/agent_bridge/processes.py
index dbfbe25..c6b887e 100644
--- a/src/agent_bridge/processes.py
+++ b/src/agent_bridge/processes.py
@@ -2,11 +2,13 @@
 
 import asyncio
 import contextlib
+import json
 import logging
 import os
 import shutil
 import signal
 import sys
+import time
 from collections.abc import Callable, Iterable, Mapping, Sequence
 from pathlib import Path
 from typing import Any
@@ -14,7 +16,7 @@
 import psutil
 
 from agent_bridge.paths import pids_path
-from agent_bridge.persist import atomic_write_json, read_json
+from agent_bridge.persist import FileLockTimeout, atomic_write_json, file_lock
 
 log = logging.getLogger(__name__)
 
@@ -194,12 +196,66 @@ async def reap_subprocess(proc: Any | None, timeout: float = 5.0) -> None:
         await asyncio.wait_for(proc.wait(), timeout=2)
 
 
-def record_pid(home, session_id: str, pid: int, create_time: float | None, image_name: str | None) -> None:
-    path = pids_path(home)
+PIDS_WRITE_ATTEMPTS = 5
+PIDS_WRITE_RETRY_BASE_SEC = 0.05
+PIDS_LOCK_TIMEOUT_SEC = 5.0
+
+
+def _read_pids_table(path: Path) -> dict[str, Any]:
+    """Read pids.json under the caller's lock.
+
+    Missing or corrupt JSON reads as empty — atomic replacement can never
+    leave a partial file, so overwrite is safe. An OSError (locked-away
+    reads on Windows) is NOT an empty table: it propagates so the caller
+    retries or gives up instead of wiping siblings' registrations.
+    """
     try:
-        table = read_json(path, {})
-        if not isinstance(table, dict):
-            table = {}
+        raw = path.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return {}
+    try:
+        table = json.loads(raw)
+    except json.JSONDecodeError:
+        return {}
+    return table if isinstance(table, dict) else {}
+
+
+def _update_pids(home, mutate: Callable[[dict[str, Any]], dict[str, Any] | None]) -> None:
+    """Locked read-modify-write of pids.json.
+
+    ``mutate`` gets a copy of the current table; returning a changed dict
+    writes it back atomically, returning None writes nothing. PermissionError
+    (read or os.replace under a racing sibling) retries with the same
+    backoff as state.json writes; lock timeout or other OSError warns and
+    gives up — never raises, never writes a partial view.
+    """
+    path = pids_path(home)
+    lock_path = path.with_name(f"{path.name}.lock")
+    delay = PIDS_WRITE_RETRY_BASE_SEC
+    for attempt in range(PIDS_WRITE_ATTEMPTS):
+        try:
+            with file_lock(lock_path, timeout_sec=PIDS_LOCK_TIMEOUT_SEC):
+                current = _read_pids_table(path)
+                updated = mutate(dict(current))
+                if updated is not None and updated != current:
+                    atomic_write_json(path, updated)
+            return
+        except FileLockTimeout as exc:
+            log.warning("could not update %s: %s", path, exc)
+            return
+        except PermissionError:
+            if attempt + 1 == PIDS_WRITE_ATTEMPTS:
+                log.warning("could not update %s after retries", path, exc_info=True)
+                return
+            time.sleep(delay)
+            delay *= 2
+        except OSError:
+            log.warning("could not update %s", path, exc_info=True)
+            return
+
+
+def record_pid(home, session_id: str, pid: int, create_time: float | None, image_name: str | None) -> None:
+    def mutate(table: dict[str, Any]) -> dict[str, Any]:
         table[session_id] = {
             "pid": pid,
             "create_time": create_time,
@@ -209,20 +265,18 @@ def record_pid(home, session_id: str, pid: int, create_time: float | None, image
             "owner_pid": os.getpid(),
             "owner_create_time": process_create_time(os.getpid()),
         }
-        atomic_write_json(path, table)
-    except OSError:
-        log.warning("could not record pid for session %s", session_id, exc_info=True)
+        return table
+
+    _update_pids(home, mutate)
 
 
 def drop_pid(home, session_id: str) -> None:
-    path = pids_path(home)
-    try:
-        table = read_json(path, {})
-        if isinstance(table, dict) and session_id in table:
-            table.pop(session_id, None)
-            atomic_write_json(path, table)
-    except OSError:
-        log.warning("could not drop pid for session %s", session_id, exc_info=True)
+    def mutate(table: dict[str, Any]) -> dict[str, Any] | None:
+        if session_id not in table:
+            return None
+        return {key: value for key, value in table.items() if key != session_id}
+
+    _update_pids(home, mutate)
 
 
 def owner_alive(pid: int | None, create_time: float | None) -> bool:
@@ -260,17 +314,24 @@ def reap_orphans(home) -> list[int]:
     records survive the rewrite.
     """
     path = pids_path(home)
-    table = read_json(path, {})
-    if not isinstance(table, dict):
-        table = {}
+    lock_path = path.with_name(f"{path.name}.lock")
+    # Snapshot under the lock; killing takes seconds and must not hold it.
+    try:
+        with file_lock(lock_path, timeout_sec=PIDS_LOCK_TIMEOUT_SEC):
+            table = _read_pids_table(path)
+    except OSError as exc:
+        # An unreadable table is not an empty one: reap nothing this boot.
+        log.warning("could not read pid table %s; skipping orphan reap: %s", path, exc)
+        return []
     killed: list[int] = []
-    kept: dict[str, Any] = {}
+    processed: dict[str, Any] = {}
     for session_id, info in table.items():
         if not isinstance(info, dict):
+            processed[session_id] = info
             continue
         if _owner_alive(info):
-            kept[session_id] = info
             continue
+        processed[session_id] = info
         pid = info.get("pid")
         create_time = info.get("create_time")
         image_name = info.get("image_name")
@@ -305,13 +366,20 @@ def reap_orphans(home) -> list[int]:
                 pid,
                 session_id,
             )
-    if table != kept:
-        try:
-            atomic_write_json(path, kept)
-        except OSError:
-            # A transient Windows file lock (antivirus, racing sibling boot)
-            # must not abort server startup; the next boot retries.
-            log.warning("could not rewrite pid table %s", path, exc_info=True)
+
+    def mutate(current: dict[str, Any]) -> dict[str, Any] | None:
+        # Only remove records this pass actually processed and only when the
+        # row still matches the snapshot — registrations a sibling added or
+        # rewrote while we killed processes stay.
+        out = dict(current)
+        removed = False
+        for session_id, info in processed.items():
+            if out.get(session_id) == info:
+                out.pop(session_id)
+                removed = True
+        return out if removed else None
+
+    _update_pids(home, mutate)
     return killed
 
 
diff --git a/src/agent_bridge/registry.py b/src/agent_bridge/registry.py
index ab328e3..3a3ed69 100644
--- a/src/agent_bridge/registry.py
+++ b/src/agent_bridge/registry.py
@@ -736,6 +736,7 @@ async def dispatch_task(
                     "effort": task.effort,
                     "request_id": request_id,
                     "reused": True,
+                    "persisted": await self.flush_state(),
                 }
             if session_id:
                 session = self.sessions.get(session_id)
@@ -792,19 +793,30 @@ async def dispatch_task(
             self._cancel_idle(session.session_id)
             self._prune()
             self.save()
+            # The task row and request_id binding must reach disk before the
+            # worker starts: a crash between them would otherwise replay the
+            # dispatch as a duplicate. Failure degrades to persisted=False,
+            # never a rejected dispatch.
             log.info(
                 "task_dispatched task_id=%s session_id=%s agent=%s",
                 task.task_id,
                 session.session_id,
                 agent,
             )
-            self._bg[task.task_id] = asyncio.create_task(self._run_task(task.task_id), name=f"task-{task.task_id}")
+            try:
+                persisted = await self.flush_state()
+            finally:
+                # Even when the caller cancels mid-flush, the registered task
+                # must still start — a request_id retry would otherwise reuse
+                # a row that never runs.
+                self._bg[task.task_id] = asyncio.create_task(self._run_task(task.task_id), name=f"task-{task.task_id}")
         return {
             "task_id": task.task_id,
             "session_id": session.session_id,
             "agent": agent,
             "model": session.model,
             "effort": session.effort,
+            "persisted": persisted,
             **({"request_id": request_id, "reused": False} if request_id is not None else {}),
         }
 
@@ -1355,6 +1367,33 @@ def _task_snapshot(self, task: Task, include_result: bool = False) -> dict:
         payload["stall_timeout_sec"] = cfg.stall_timeout_sec if cfg is not None else None
         return payload
 
+    async def _persisted(self, task: Task) -> bool | None:
+        """Persistence verdict for terminal answers; None while non-terminal.
+
+        A disk-fallback row belongs to another instance and is already on
+        disk by definition, so it answers True without a local flush.
+        """
+        if task.status not in TERMINAL_STATUSES:
+            return None
+        if task.task_id not in self.tasks:
+            return True
+        done = self._done.get(task.task_id)
+        if done is None or not done.is_set():
+            # _run_task marks the status terminal in-memory, then collects
+            # final files_changed and flushes the transcript before its
+            # closing save. Done set means that save has been issued (it
+            # runs synchronously right after set()); before that, issue the
+            # save here so a read cannot answer persisted=True while disk
+            # still says running.
+            self.save()
+        return await self.flush_state()
+
+    async def _with_persisted(self, task: Task, payload: dict) -> dict:
+        persisted = await self._persisted(task)
+        if persisted is not None:
+            payload["persisted"] = persisted
+        return payload
+
     async def wait_task(self, task_id: str, timeout_sec: float = DEFAULT_WAIT_SEC) -> dict:
         task = self._require_task(task_id)
         event = self._done.setdefault(task_id, asyncio.Event())
@@ -1363,21 +1402,19 @@ async def wait_task(self, task_id: str, timeout_sec: float = DEFAULT_WAIT_SEC) -
                 await asyncio.wait_for(event.wait(), timeout=timeout_sec)
             except TimeoutError:
                 return {"timed_out": True, **self._task_snapshot(self.tasks[task_id])}
-        persisted = False
-        if task.status in TERMINAL_STATUSES:
-            # A terminal answer must also be on disk, not merely queued
-            # behind the background flusher.
-            persisted = await self.flush_state()
+        # A terminal answer must also be on disk, not merely queued
+        # behind the background flusher.
         return {
             "timed_out": False,
-            "persisted": persisted,
+            "persisted": (await self._persisted(task)) is True,
             **self._task_snapshot(self.tasks[task_id], include_result=True),
         }
 
-    def check_task(self, task_id: str) -> dict:
-        return self._task_snapshot(self._require_task(task_id, allow_disk=True))
+    async def check_task(self, task_id: str) -> dict:
+        task = self._require_task(task_id, allow_disk=True)
+        return await self._with_persisted(task, self._task_snapshot(task))
 
-    def get_result(
+    async def get_result(
         self,
         task_id: str,
         cursor: int = 0,
@@ -1419,7 +1456,7 @@ def get_result(
                 ),
             }
         )
-        return payload
+        return await self._with_persisted(task, payload)
 
     def get_transcript(self, session_id: str, offset: int = 0, limit: int = 50, kinds: list[str] | None = None) -> dict:
         if session_id not in self.sessions and not transcript_path(session_id, self.home).is_file():
@@ -1432,7 +1469,7 @@ async def cancel_task(self, task_id: str) -> dict:
             raise RuntimeError(NESTED_CANCEL_ERROR)
         task = self._require_task(task_id)
         if task.status in TERMINAL_STATUSES:
-            return self._task_snapshot(task)
+            return await self._with_persisted(task, self._task_snapshot(task))
         self._cancel_requested.add(task_id)
         session = self.sessions[task.session_id]
         adapter = self._adapters.get(session.session_id)
@@ -1457,7 +1494,8 @@ async def cancel_task(self, task_id: str) -> dict:
             self._bg.pop(task_id, None)
             self._done[task_id].set()
             self.save()
-        return self._task_snapshot(self.tasks[task_id])
+        task = self.tasks[task_id]
+        return await self._with_persisted(task, self._task_snapshot(task))
 
     def list_sessions(self, active_only: bool = False) -> list[dict]:
         rows = []
diff --git a/src/agent_bridge/server.py b/src/agent_bridge/server.py
index e12e2ce..754c6fc 100644
--- a/src/agent_bridge/server.py
+++ b/src/agent_bridge/server.py
@@ -156,7 +156,7 @@ async def wait_task(ctx: Context, task_id: str, timeout_sec: float = DEFAULT_WAI
 async def check_task(ctx: Context, task_id: str) -> dict[str, Any]:
     """Non-blocking status, elapsed time, and recent activity for a task. files_changed_state is pending until the final workspace diff, collected when the list/count are authoritative, or unavailable if collection could not run. files_changed is capped at 200 paths; files_changed_total carries the real count. silent_for_sec is the time since the worker's last output; Bridge fails the task with stop_reason "stalled" once it passes stall_timeout_sec."""
     try:
-        return {"ok": True, **_registry(ctx).check_task(task_id)}
+        return {"ok": True, **await _registry(ctx).check_task(task_id)}
     except Exception as exc:
         return _error(exc)
 
@@ -172,7 +172,7 @@ async def get_result(
     try:
         return {
             "ok": True,
-            **_registry(ctx).get_result(task_id, cursor=cursor, max_chars=max_chars),
+            **await _registry(ctx).get_result(task_id, cursor=cursor, max_chars=max_chars),
         }
     except Exception as exc:
         return _error(exc)
diff --git a/tests/test_acp_echo.py b/tests/test_acp_echo.py
index 08116cc..c536f3c 100644
--- a/tests/test_acp_echo.py
+++ b/tests/test_acp_echo.py
@@ -95,8 +95,8 @@ async def test_acp_failures_survive_tools_and_restart(
         assert failure["prompt_may_have_been_sent"] is possibly_sent
         assert failure["kind"] == kind
         assert failure["exit_code"] == exit_code
-        assert registry.check_task(task_id)["failure"] == failure
-        assert registry.get_result(task_id)["failure"] == failure
+        assert (await registry.check_task(task_id))["failure"] == failure
+        assert (await registry.get_result(task_id))["failure"] == failure
         if possibly_sent:
             assert marker.read_text(encoding="utf-8") == "executed\n"
         else:
@@ -108,8 +108,8 @@ async def test_acp_failures_survive_tools_and_restart(
     restored.config.agents["echo"] = cfg
     await restored.start()
     try:
-        assert restored.get_result(task_id)["failure"] == failure
-        assert restored.check_task(task_id)["status"] == "failed"
+        assert (await restored.get_result(task_id))["failure"] == failure
+        assert (await restored.check_task(task_id))["status"] == "failed"
     finally:
         await restored.stop()
 
@@ -139,7 +139,7 @@ async def test_cancel_of_cancel_ignoring_worker_ends_cancelled(bridge_home, tmp_
         assert cancelled["failure"] is None
         waited = await registry.wait_task(dispatched["task_id"], timeout_sec=10)
         assert waited["status"] == "cancelled"
-        assert registry.check_task(dispatched["task_id"])["failure"] is None
+        assert (await registry.check_task(dispatched["task_id"]))["failure"] is None
         session = registry.sessions[dispatched["session_id"]]
         assert session.pid is None
         assert session.session_id not in read_json(pids_path(bridge_home), {})
diff --git a/tests/test_processes.py b/tests/test_processes.py
index c66f606..dac4f08 100644
--- a/tests/test_processes.py
+++ b/tests/test_processes.py
@@ -414,6 +414,112 @@ def boom(*_args, **_kwargs):
     drop_pid(tmp_path, "sess_keep")
 
 
+def test_record_pid_concurrent_processes_merge(tmp_path: Path):
+    # Three sibling instances race to register 30 workers each; a locked
+    # read-modify-write must leave all 90 rows, not the last writer's 30.
+    root = Path(__file__).resolve().parents[1]
+    env = {**os.environ, "PYTHONPATH": str(root / "src")}
+    script = (
+        "import os, sys\n"
+        "from pathlib import Path\n"
+        "from agent_bridge.processes import record_pid\n"
+        "home, tag = Path(sys.argv[1]), sys.argv[2]\n"
+        "for i in range(30):\n"
+        "    record_pid(home, f'{tag}-sess-{i}', os.getpid(), None, None)\n"
+    )
+    procs = [
+        subprocess.Popen([sys.executable, "-c", script, str(tmp_path), tag], env=env)
+        for tag in ("a", "b", "c")
+    ]
+    try:
+        for proc in procs:
+            assert proc.wait(timeout=30) == 0
+    finally:
+        for proc in procs:
+            if proc.poll() is None:
+                proc.kill()
+    table = read_json(pids_path(tmp_path), {})
+    assert len(table) == 90
+    for tag in ("a", "b", "c"):
+        for i in range(30):
+            assert f"{tag}-sess-{i}" in table
+
+
+def test_record_pid_read_failure_never_wipes_table(tmp_path: Path, monkeypatch, caplog):
+    # An unreadable pids.json is not an empty one: the update must fail
+    # instead of writing back only the caller's own row.
+    record_pid(tmp_path, "sess_a", os.getpid(), 1.0, "python.exe")
+    record_pid(tmp_path, "sess_b", os.getpid(), 1.0, "python.exe")
+    before = read_json(pids_path(tmp_path), {})
+
+    def denied(_path):
+        raise PermissionError("locked away")
+
+    monkeypatch.setattr("agent_bridge.processes._read_pids_table", denied)
+    monkeypatch.setattr("agent_bridge.processes.PIDS_WRITE_RETRY_BASE_SEC", 0.001)
+    with caplog.at_level("WARNING"):
+        record_pid(tmp_path, "sess_c", os.getpid(), 1.0, "python.exe")
+
+    assert read_json(pids_path(tmp_path), {}) == before
+    assert "sess_c" not in before
+    assert any("could not update" in record.getMessage() for record in caplog.records)
+
+
+def test_record_pid_retries_transient_permission_error(tmp_path: Path, monkeypatch):
+    from agent_bridge import processes
+
+    record_pid(tmp_path, "sess_a", os.getpid(), 1.0, "python.exe")
+    record_pid(tmp_path, "sess_b", os.getpid(), 1.0, "python.exe")
+    real_read = processes._read_pids_table
+    calls = 0
+
+    def flaky(path):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise PermissionError("transient")
+        return real_read(path)
+
+    monkeypatch.setattr("agent_bridge.processes._read_pids_table", flaky)
+    record_pid(tmp_path, "sess_c", os.getpid(), 1.0, "python.exe")
+
+    assert calls == 2
+    table = read_json(pids_path(tmp_path), {})
+    assert {"sess_a", "sess_b", "sess_c"} <= set(table)
+
+
+def test_reap_orphans_keeps_sibling_records_added_mid_sweep(tmp_path: Path, monkeypatch):
+    # The orphan's pid is unused, so the sweep drops the record without
+    # killing anything; a sibling record registered while the sweep runs
+    # must survive the rewrite.
+    atomic_write_json(
+        pids_path(tmp_path),
+        {
+            "sess_orphan": {
+                "pid": 2_000_000_001,
+                "create_time": 1.0,
+                "image_name": "ghost.exe",
+            },
+        },
+    )
+    added = False
+    real_process = psutil.Process
+
+    def add_sibling_then_process(pid):
+        nonlocal added
+        if not added:
+            added = True
+            record_pid(tmp_path, "sess_sibling", os.getpid(), 1.0, "python.exe")
+        return real_process(pid)
+
+    monkeypatch.setattr("agent_bridge.processes.psutil.Process", add_sibling_then_process)
+    killed = reap_orphans(tmp_path)
+    assert killed == []
+    table = read_json(pids_path(tmp_path), {})
+    assert "sess_orphan" not in table
+    assert "sess_sibling" in table
+
+
 @pytest.mark.asyncio
 async def test_interrupt_escalates_at_once_when_signal_fails(monkeypatch):
     async def never_wait():
diff --git a/tests/test_registry.py b/tests/test_registry.py
index 2781d9b..0bc624c 100644
--- a/tests/test_registry.py
+++ b/tests/test_registry.py
@@ -9,6 +9,7 @@
 import sys
 import threading
 import time
+import uuid
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
@@ -408,10 +409,10 @@ async def test_check_task_reads_sibling_task_from_state(bridge_home, monkeypatch
     await registry.start()
     try:
         assert "task_remote" not in registry.tasks
-        snap = registry.check_task("task_remote")
+        snap = await registry.check_task("task_remote")
         assert snap["status"] == "completed"
         assert snap["from_state"] is True
-        result = registry.get_result("task_remote")
+        result = await registry.get_result("task_remote")
         assert result["result_text"] == "remote result"
         with pytest.raises(KeyError, match="unknown task"):
             await registry.wait_task("task_remote", timeout_sec=0.1)
@@ -443,7 +444,7 @@ async def test_check_task_from_state_reports_null_silence(bridge_home, monkeypat
     registry = Registry.create(bridge_home)
     await registry.start()
     try:
-        snap = registry.check_task("task_remote_run")
+        snap = await registry.check_task("task_remote_run")
         assert snap["from_state"] is True
         assert snap["silent_for_sec"] is None
     finally:
@@ -484,7 +485,7 @@ async def big_turn(self, session, task):
     again = Registry.create(bridge_home)
     await again.start()
     try:
-        result = again.get_result(dispatched["task_id"])
+        result = await again.get_result(dispatched["task_id"])
         assert result["result_text"] == big
         assert len(result["result_text"]) == 20000
     finally:
@@ -694,6 +695,214 @@ def fail_always(own):
         await registry.start()
 
 
+@pytest.mark.asyncio
+async def test_check_task_terminal_waits_for_slow_state_write(
+    bridge_home, tmp_path, monkeypatch
+):
+    """check_task must not answer terminal before the terminal row is on disk."""
+    work = tmp_path / "work"
+    work.mkdir()
+    registry = Registry.create(bridge_home)
+    await registry.start()
+    original_write = registry._write_state
+    try:
+
+        def slow_write(own):
+            time.sleep(0.3)
+            original_write(own)
+
+        monkeypatch.setattr(registry, "_write_state", slow_write)
+        dispatched = await registry.dispatch_task("fake", "tiny", cwd=str(work.resolve()))
+        terminal = {status.value for status in TERMINAL_STATUSES}
+        checked = {}
+        for _ in range(200):
+            checked = await registry.check_task(dispatched["task_id"])
+            if checked["status"] in terminal:
+                break
+            await asyncio.sleep(0.05)
+        assert checked["status"] == "completed"
+        assert checked["persisted"] is True
+        rows = read_json(state_path(bridge_home), {}).get("tasks", [])
+        row = next(r for r in rows if r["task_id"] == dispatched["task_id"])
+        assert row["status"] == "completed"
+    finally:
+        monkeypatch.setattr(registry, "_write_state", original_write)
+        await registry.stop()
+
+
+@pytest.mark.asyncio
+async def test_dispatch_returns_after_task_row_and_request_binding_on_disk(
+    bridge_home, tmp_path, monkeypatch
+):
+    """A dispatch answer implies the task row and request_id binding landed;
+    a coordinator crash before that would otherwise replay a duplicate."""
+    work = tmp_path / "work"
+    work.mkdir()
+    registry = Registry.create(bridge_home)
+    await registry.start()
+    original_write = registry._write_state
+    try:
+
+        def slow_write(own):
+            time.sleep(0.3)
+            original_write(own)
+
+        monkeypatch.setattr(registry, "_write_state", slow_write)
+        request_id = str(uuid.uuid4())
+        dispatched = await registry.dispatch_task(
+            "fake", "tiny", cwd=str(work.resolve()), request_id=request_id
+        )
+        assert dispatched["persisted"] is True
+        payload = read_json(state_path(bridge_home), {})
+        assert {r["task_id"] for r in payload.get("tasks", [])} == {dispatched["task_id"]}
+        bindings = {r["request_id"]: r for r in payload.get("requests", [])}
+        assert bindings[request_id]["task_id"] == dispatched["task_id"]
+        waited = await registry.wait_task(dispatched["task_id"], timeout_sec=10)
+        assert waited["persisted"] is True
+    finally:
+        monkeypatch.setattr(registry, "_write_state", original_write)
+        await registry.stop()
+
+
+@pytest.mark.asyncio
+async def test_check_task_reports_unpersisted_when_writes_fail(
+    bridge_home, tmp_path, monkeypatch
+):
+    """A broken disk degrades check_task to persisted=False; the in-memory
+    result stays readable."""
+    work = tmp_path / "work"
+    work.mkdir()
+    registry = Registry.create(bridge_home)
+    await registry.start()
+    original_write = registry._write_state
+    try:
+
+        def fail_always(own):
+            raise PermissionError(5, "synthetic sharing/access failure")
+
+        monkeypatch.setattr(registry, "_write_state", fail_always)
+        dispatched = await registry.dispatch_task("fake", "tiny", cwd=str(work.resolve()))
+        assert dispatched["persisted"] is False
+        waited = await registry.wait_task(dispatched["task_id"], timeout_sec=10)
+        assert waited["status"] == "completed"
+        checked = await registry.check_task(dispatched["task_id"])
+        assert checked["status"] == "completed"
+        assert checked["persisted"] is False
+        result = await registry.get_result(dispatched["task_id"])
+        assert result["persisted"] is False
+        assert result["result_text"]
+    finally:
+        monkeypatch.setattr(registry, "_write_state", original_write)
+        await registry.stop()
+
+
+@pytest.mark.asyncio
+async def test_cancel_task_reports_persisted_on_terminal(bridge_home, tmp_path):
+    """cancel_task on an already-terminal task returns persisted=True."""
+    work = tmp_path / "work"
+    work.mkdir()
+    registry = Registry.create(bridge_home)
+    await registry.start()
+    try:
+        dispatched = await registry.dispatch_task("fake", "tiny", cwd=str(work.resolve()))
+        waited = await registry.wait_task(dispatched["task_id"], timeout_sec=5)
+        assert waited["status"] == "completed"
+        cancelled = await registry.cancel_task(dispatched["task_id"])
+        assert cancelled["status"] == "completed"
+        assert cancelled["persisted"] is True
+    finally:
+        await registry.stop()
+
+
+@pytest.mark.asyncio
+async def test_check_task_persists_terminal_row_during_finalize(
+    bridge_home, tmp_path, monkeypatch
+):
+    """task.status flips terminal before _run_task's finally issues the final
+    save (files_changed collection sits in between). A check_task inside that
+    window must land the terminal row itself instead of answering
+    persisted=True for a disk row that still says running."""
+    work = tmp_path / "work"
+    work.mkdir()
+    registry = Registry.create(bridge_home)
+    await registry.start()
+    try:
+        from agent_bridge import workspace
+
+        real_merge = workspace.merge_files_changed
+
+        def slow_merge(*args, **kwargs):
+            time.sleep(1.0)
+            return real_merge(*args, **kwargs)
+
+        # The failure path reaches the pending-files_changed merge in finally
+        # *after* status is already terminal — a deterministic widen window.
+        async def boom(_self, _session, _task):
+            raise RuntimeError("synthetic turn failure")
+
+        monkeypatch.setattr("agent_bridge.registry.merge_files_changed", slow_merge)
+        monkeypatch.setattr(FakeAdapter, "run_turn", boom)
+        dispatched = await registry.dispatch_task("fake", "tiny", cwd=str(work.resolve()))
+        terminal = {status.value for status in TERMINAL_STATUSES}
+        checked = {}
+        for _ in range(200):
+            checked = await registry.check_task(dispatched["task_id"])
+            if checked["status"] in terminal:
+                break
+            await asyncio.sleep(0.05)
+        assert checked["status"] == "failed"
+        assert checked["persisted"] is True
+        rows = read_json(state_path(bridge_home), {}).get("tasks", [])
+        row = next(r for r in rows if r["task_id"] == dispatched["task_id"])
+        assert row["status"] == "failed"
+    finally:
+        await registry.stop()
+
+
+@pytest.mark.asyncio
+async def test_dispatch_cancelled_during_flush_still_runs_task(
+    bridge_home, tmp_path, monkeypatch
+):
+    """If the MCP dispatch call is cancelled while awaiting the initial
+    flush, the already-registered task must still start — otherwise a
+    request_id retry would reuse a row that never runs."""
+    work = tmp_path / "work"
+    work.mkdir()
+    registry = Registry.create(bridge_home)
+    await registry.start()
+    try:
+        request_id = str(uuid.uuid4())
+        real_flush = registry.flush_state
+        gate = asyncio.Event()
+
+        async def hung_flush():
+            await gate.wait()
+            return await real_flush()
+
+        monkeypatch.setattr(registry, "flush_state", hung_flush)
+        dispatching = asyncio.create_task(
+            registry.dispatch_task(
+                "fake", "tiny", cwd=str(work.resolve()), request_id=request_id
+            )
+        )
+        await asyncio.sleep(0.05)  # let dispatch register the task and reach the flush
+        dispatching.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await dispatching
+        monkeypatch.setattr(registry, "flush_state", real_flush)
+
+        (task_id,) = [tid for tid in registry.tasks if registry.tasks[tid].agent == "fake"]
+        waited = await registry.wait_task(task_id, timeout_sec=10)
+        assert waited["status"] == "completed"
+        retry = await registry.dispatch_task(
+            "fake", "tiny", cwd=str(work.resolve()), request_id=request_id
+        )
+        assert retry["reused"] is True
+        assert retry["task_id"] == task_id
+    finally:
+        await registry.stop()
+
+
 def test_state_write_retries_bounded_permission_errors(bridge_home, monkeypatch):
     registry = Registry.create(bridge_home)
     original_write = atomic_write_json
@@ -913,7 +1122,7 @@ async def write_then_ok(self, session, task):
         assert waited["model"] == "gemini-3.7-flash"
         assert waited["effort"] == "low"
         assert waited["observed_model"] is None
-        result = registry.get_result(dispatched["task_id"])
+        result = await registry.get_result(dispatched["task_id"])
         assert result["model"] == "gemini-3.7-flash"
         assert "observed_model" in result
         assert result["result_text"] == result_text
@@ -950,7 +1159,7 @@ async def long_turn(self, session, task):
         cursor = 0
         parts: list[str] = []
         while True:
-            page = registry.get_result(
+            page = await registry.get_result(
                 dispatched["task_id"], cursor=cursor, max_chars=17_000
             )
             assert page["result_complete"] is True
@@ -1020,7 +1229,7 @@ async def empty_turn(self, session, task):
         assert waited["observed_model"] == "kimi-code/k3-256k"
         assert waited["observed_effort"] == "high"
         assert any("402 membership" in w for w in waited["warnings"])
-        result = registry.get_result(dispatched["task_id"])
+        result = await registry.get_result(dispatched["task_id"])
         assert "end_turn with empty text" in result["hint"]
     finally:
         await registry.stop()
@@ -1084,7 +1293,7 @@ async def applied_turn(self, session, task):
         assert waited["effort"] == "max"
         assert waited["observed_model"] == "opencode/x-preview-f-free"
         assert waited["observed_effort"] == "high"
-        result = registry.get_result(dispatched["task_id"])
+        result = await registry.get_result(dispatched["task_id"])
         assert "last values Bridge successfully set" in result["hint"]
     finally:
         await registry.stop()
@@ -1124,7 +1333,7 @@ async def applied_turn(self, session, task):
         assert waited["effort"] == "max"
         assert waited["observed_model"] == "sonnet"
         assert waited["observed_effort"] == "xhigh"
-        result = registry.get_result(dispatched["task_id"])
+        result = await registry.get_result(dispatched["task_id"])
         assert "Claude Code observed_model/effort" in result["hint"]
         assert "last values Bridge successfully set" in result["hint"]
     finally:
@@ -1507,9 +1716,9 @@ async def test_files_changed_is_capped(bridge_home, tmp_path, monkeypatch):
         assert waited["files_changed"] == paths[:200]
         assert waited["files_changed_total"] == 500
         assert waited["files_changed_truncated"] is True
-        result = registry.get_result(dispatched["task_id"])
+        result = await registry.get_result(dispatched["task_id"])
         assert "first 200 of 500" in result["hint"]
-        checked = registry.check_task(dispatched["task_id"])
+        checked = await registry.check_task(dispatched["task_id"])
         assert checked["files_changed_total"] == 500
         assert checked["files_changed_truncated"] is True
     finally:
@@ -1553,7 +1762,7 @@ async def write_then_wait(self, session, task):
         dispatched = await registry.dispatch_task("fake", "write", cwd=str(work.resolve()))
         await asyncio.wait_for(wrote.wait(), timeout=5)
 
-        running = registry.check_task(dispatched["task_id"])
+        running = await registry.check_task(dispatched["task_id"])
         assert running["status"] == "running"
         assert running["files_changed"] == []
         assert running["files_changed_total"] == 0
@@ -1904,7 +2113,7 @@ async def test_check_task_reports_silent_for_sec(bridge_home, tmp_path, monkeypa
     try:
         dispatched = await registry.dispatch_task("fake", "slow", cwd=str(work.resolve()))
         await asyncio.sleep(0.5)
-        checked = registry.check_task(dispatched["task_id"])
+        checked = await registry.check_task(dispatched["task_id"])
         assert isinstance(checked["silent_for_sec"], int)
         assert 0 <= checked["silent_for_sec"] <= 2
         assert checked["stall_timeout_sec"] == 1800