feat(agentctl): harden CURRENT projection and human-section guards

W2-U5: own_row_only/slot_protected policy wiring, normalized projection
rows, idempotent request ids, five-state status vocab, expanded protected
human sections, projections ledger readout, and doc contradiction fixes.

.doc/ops/2026-09-21-agent-worktree-workflow.md

@@ -1,6 +1,6 @@
 # Agent worktree 与 L1 操作规范
 
-**修订**：`agent-governance-2026-09-21-r1`。来源：[AGENTS](../../AGENTS.md)、[规范登记](./2026-09-21-policy-registry.md)。工具为 `scripts/agentctl`（Python 标准库，Linux 本机），协议 v1；实际安装与验证见[实施任务](../archive/ops/2026-09-21-agent-governance-implementation.md)。
+**修订**：`agent-governance-2026-09-21-r1`。来源：[AGENTS](../../AGENTS.md)、[规范登记](./2026-09-21-policy-registry.md)。工具为 `scripts/agentctl`（Python 标准库，macOS/Linux 双平台），协议 v1；实际安装与验证见[实施任务](../archive/ops/2026-09-21-agent-governance-implementation.md)。
 
 本规范约束合作 Agent 的统一操作入口和 Git 提交/引用门禁。任意同 UID shell、修改管理器或跳过 hooks 仍可绕过；不把 L1 宣称为 OS 沙箱。共享服务配置与产品数据库不属于任务素材。
 
@@ -128,6 +128,6 @@ validate 要求干净树、记录精确 HEAD 与命令摘要；必要的实际
 
 清理由目录外的维护者执行，核对实际 Git 映射、dev 祖先、所有证据、tracked/untracked/ignored 文件、未完成 Git 操作、registry、生命周期锁和使用目录的进程。只用普通 remove/-d，拒绝时保留 cleanup_pending 和复查责任，不换成强制删除。
 
-**本机限制**：无法读取某些同 UID 进程 cwd 时，工具不能证明其未使用目录，故保留 cleanup_pending；只有明确识别的 OS 用户 systemd 管理进程有诊断性例外。测试中的 3 个删除正例仅替换 `/proc` 枚举以模拟可见主机，其余 Git/锁/文件条件为真实执行；可见占用、未知不可读、未合入、脏文件与活跃登记的拒绝另有真实/针对性测试。不要把这项限制转成跳过检查的通用开关。
+**本机限制**：无法读取某些同 UID 进程 cwd 时，工具不能证明其未使用目录，故保留 cleanup_pending；明确识别的 OS 用户 launchd（macOS）/systemd（Linux）管理进程有诊断性例外。进程 cwd 可见性走双路径：Darwin 用 `lsof`，Linux 用 `/proc`；测试中的删除正例可替换对应枚举以模拟可见主机，其余 Git/锁/文件条件为真实执行；可见占用、未知不可读、未合入、脏文件与活跃登记的拒绝另有真实/针对性测试。不要把这项限制转成跳过检查的通用开关。
 
 保护任务、旧 detached 工作区和非本次产物不自动清理。用户未要求 push 时仅本地提交/合入；用户已要求时再按授权同步并核对远端引用。

.doc/ops/CURRENT.md

@@ -12,12 +12,13 @@
 
 2. **dev→test 已晋升**：`origin/dev@591ea601`（schema 61 门禁绿）；`origin/test@83c6a12a` + tag `rc-v1.1.0-beta.1`；test 环境 health schema=61、Runner `MiniMax-M3`。[报告](../test-reports/2026-09-22-dev-gate-and-test-promotion-test-report.md)。
 3. **本任务 cleanup=pending**：feature / integrate / promote-test worktrees 待独立维护者按门禁复查后清理；不强删。
-4. **Agent 规范与 L1 门禁**：R01–R10、共享 .locks 已启用；本机 macOS 上租约路径不可用，受测 Linux 主机仍 enforced。
+4. **Agent 规范与 L1 门禁**：R01–R10、共享 .locks 已启用；本机 macOS 租约路径可用（Darwin 进程层已落地），与 Linux 双平台 enforced。
 
 ## 并行认领
 
 | owner | task | branch | worktree | mutex | status |
 |---|---|---|---|---|---|
+| Cursor | current-projection-2026-09-23 | feature/current-projection-2026-09-23 | .worktrees/current-projection-2026-09-23 | current-projection | in_progress |
 | Cursor | agentctl-daemon-2026-09-23 | feature/agentctl-daemon-2026-09-23 | .worktrees/agentctl-daemon-2026-09-23 | agentctl-daemon | in_progress |
 | Cursor | agentctl-handoff-2026-09-23 | feature/agentctl-handoff-2026-09-23 | .worktrees/agentctl-handoff-2026-09-23 | agentctl-handoff | in_progress |
 | Cursor | agentctl-claim-2026-09-23 | feature/agentctl-claim-2026-09-23 | .worktrees/agentctl-claim-2026-09-23 | agentctl-claim | done |

.doc/ops/short-term/2026-09-23-current-projection.md

@@ -0,0 +1,48 @@
+---
+title: "W2-U5：CURRENT 状态输出化（投影硬化）"
+task_id: current-projection-2026-09-23
+status: in_progress
+created: 2026-09-23
+updated: 2026-09-23
+owner: Cursor
+task_type: implementation
+task_tier: single
+policy_revision: "agent-governance-2026-09-21-r1 @ 725e2e4beb9e77718f812089061b1b4575344bff"
+branch: feature/current-projection-2026-09-23
+worktree: .worktrees/current-projection-2026-09-23
+base_dev_sha: "725e2e4beb9e77718f812089061b1b4575344bff"
+write_scope: ["scripts", "configs", ".doc"]
+allowed_environments: []
+mutex: current-projection
+review_state: pending
+integration_state: pending
+deployment: not_required
+cleanup: pending
+cleanup_owner: "独立维护者"
+cleanup_trigger: "合入 local_dev 且无进程占用后复查"
+remote_sync: not_requested
+---
+
+# W2-U5：CURRENT 状态输出化（投影硬化）
+
+## 用户目标与授权
+
+- 用户目标：硬化 `docs project` / CURRENT 投影：策略接线、渲染归一、幂等 request_id、五态词汇、人工区块机器禁写、投影账本只读、清理表头与文档矛盾修正。
+- 授权：W2-U5 实施；不 push；不集成（由父代理合入）；不改 FEATURE-MATRIX 状态值；不改 plan 文件。
+- 禁止：扩大业务写环境；把 Markdown idle 当原子锁。
+
+## 范围
+
+- `scripts/agentctl.py`：`own_row_only` / `slot_protected` 一致性、`docs_project`、`docs projections`、状态词汇
+- `scripts/agentctl_w4.py`：`build_projection_row` 消费、`apply_projection_locally` cells[:2]、cleanup render
+- `configs/agent-governance.json`：扩展 `protected_sections`
+- `.doc/ops/CURRENT.md` / 操作规范矛盾修正
+- 真实并发与负例 unittest
+
+## 验收
+
+1. 双任务并行 `docs project --apply` 两行均在
+2. 同行竞态后到者 `BASE_CHANGED`，无部分写
+3. `@document`+`docs:admin` 改集成槽位仍 `PATH_OUT_OF_SCOPE`
+4. 人工区块（焦点/清理接管/受控约束/活跃入口）机器写拒绝
+5. `docs project` preview→apply→receipts 闭环；`unittest discover` 全绿

configs/agent-governance.json

@@ -29,7 +29,13 @@
   },
   "reserved_record_ids": ["holder", "idle", "Cursor", "Codex", "-"],
   "protected_sections": {
-    ".doc/ops/CURRENT.md": ["## 集成槽位"]
+    ".doc/ops/CURRENT.md": [
+      "## 集成槽位",
+      "## 焦点",
+      "## 清理接管与遗留现场",
+      "## 受控约束",
+      "## 活跃入口"
+    ]
   },
   "heartbeat_ttl_seconds": 300,
   "suspect_ttl_seconds": 600

scripts/agentctl.py

@@ -48,10 +48,20 @@ DEFAULT_POLICY = {
         },
     },
     'reserved_record_ids': ['holder', 'idle', 'Cursor', 'Codex', '-'],
-    'protected_sections': {'.doc/ops/CURRENT.md': ['## 集成槽位']},
+    'protected_sections': {
+        '.doc/ops/CURRENT.md': [
+            '## 集成槽位',
+            '## 焦点',
+            '## 清理接管与遗留现场',
+            '## 受控约束',
+            '## 活跃入口',
+        ],
+    },
     'heartbeat_ttl_seconds': 300,
     'suspect_ttl_seconds': 600,
 }
+PROJECTION_STATUS_VOCAB = frozenset({'planned', 'in_progress', 'blocked', 'done', 'cancelled'})
+PROJECTION_STATUS_FROM_MACHINE = {'queued': 'in_progress', 'idle': 'planned'}
 SHARED_DOCS = set(DEFAULT_POLICY['shared_docs'])
 RESERVED_RECORD_IDS = set(DEFAULT_POLICY['reserved_record_ids'])
 DOC_RULES = dict(DEFAULT_POLICY['doc_rules'])
@@ -178,6 +188,18 @@ def _validate_policy_shape(data) -> dict:
     for key in ('heartbeat_ttl_seconds', 'suspect_ttl_seconds'):
         if key in data and (not isinstance(data[key], int) or data[key] <= 0):
             deny('POLICY_INVALID', f'{key} must be a positive integer')
+    for filename, rules in data['doc_rules'].items():
+        if not isinstance(rules, dict):
+            deny('POLICY_INVALID', 'doc_rules entries must be objects', file=filename)
+        if rules.get('slot_protected'):
+            sections = data['protected_sections'].get(filename)
+            if not isinstance(sections, list) or not sections:
+                deny('POLICY_INVALID',
+                     'slot_protected requires a nonempty protected_sections entry',
+                     file=filename)
+            if not all(isinstance(item, str) and item for item in sections):
+                deny('POLICY_INVALID', 'protected_sections values must be nonempty strings',
+                     file=filename)
     return data
 
 
@@ -1077,7 +1099,8 @@ def docs_apply(repo, args):
             seen.add((filename, record_id))
             if record_id == '@document' and 'docs:admin' not in json.loads(task['capabilities']):
                 deny('CAPABILITY_REQUIRED', 'Document layout changes require docs:admin')
-            if filename.endswith('CURRENT.md') and record_id != args.task and 'docs:admin' not in json.loads(task['capabilities']):
+            if (DOC_RULES.get(filename, {}).get('own_row_only') and record_id != args.task
+                    and 'docs:admin' not in json.loads(task['capabilities'])):
                 deny('PATH_OUT_OF_SCOPE', 'A task can update only its own CURRENT record')
             if record_id in RESERVED_RECORD_IDS:
                 deny('PATH_OUT_OF_SCOPE', 'Use task ids; slot ownership is not a document edit')
@@ -1687,7 +1710,16 @@ def docs_project(repo, args):
             "SELECT id,state,resources,kind FROM runs WHERE task=? AND state IN ('running','queued','recovery_required','busy','suspect','quarantine')",
             (args.task,)).fetchall()]
     queue_state = pending[0]['state'] if pending else 'idle'
-    status = args.status or ('queued' if pending else 'in_progress')
+    if args.status is not None:
+        if args.status not in PROJECTION_STATUS_VOCAB:
+            deny('INVALID_REQUEST', 'Projection status must be planned|in_progress|blocked|done|cancelled',
+                 status=args.status)
+        status = args.status
+    else:
+        machine_status = 'queued' if pending else 'idle'
+        status = PROJECTION_STATUS_FROM_MACHINE.get(machine_status, machine_status)
+        if status not in PROJECTION_STATUS_VOCAB:
+            deny('INVALID_REQUEST', 'Mapped projection status is not in vocabulary', status=status)
     worktree_path = Path(task['worktree'])
     try:
         worktree_cell = str(worktree_path.relative_to(repo.main))
@@ -1714,10 +1746,9 @@ def docs_project(repo, args):
         if args.replacement:
             replacement = args.replacement
         elif filename.endswith('CURRENT.md'):
-            replacement = (
-                f"| {task['owner']} | {task['id']} | {task['branch']} | {worktree_cell} "
-                f"| {args.mutex or '-'} | {status} |"
-            )
+            replacement = w4.build_projection_row(
+                task_id=task['id'], owner=task['owner'], branch=task['branch'],
+                worktree=worktree_cell, mutex=args.mutex or '-', status=status)
         else:
             deny('INVALID_REQUEST', 'Row projection for this file requires --replacement')
         change = w4.build_row_projection_change(
@@ -1741,12 +1772,17 @@ def docs_project(repo, args):
         'queue': pending,
         'runs': active_runs,
         'queue_phase': queue_state,
-        'snapshot_time': time.time(),
         'dev_sha': git(repo.main, 'rev-parse', 'dev'),
         'mode': mode,
+        'status': status,
+        'record_id': record_id,
+        'mutex': args.mutex or '-',
     }
+    request_id = args.request_id or (
+        f"{args.task}-project-{digest(json.dumps(machine, sort_keys=True, default=str))[:12]}")
+    machine['snapshot_time'] = time.time()
     request = {
-        'request_id': args.request_id or (args.task + '-project-1'),
+        'request_id': request_id,
         'changes': [change],
     }
     applied = None
@@ -1770,6 +1806,24 @@ def docs_project(repo, args):
     return 0
 
 
+def docs_projections(repo, args):
+    """Read-only projection ledger readout from the projections table."""
+    require_policy(repo)
+    with repo.db(True) as db:
+        ensure_schema(db)
+        if getattr(args, 'file', None):
+            rows = db.execute(
+                'SELECT file,record_id,task,source,content_hash,updated FROM projections '
+                'WHERE file=? ORDER BY updated, record_id',
+                (args.file,)).fetchall()
+        else:
+            rows = db.execute(
+                'SELECT file,record_id,task,source,content_hash,updated FROM projections '
+                'ORDER BY file, updated, record_id').fetchall()
+    emit({'status': 'ok', 'projections': [dict(row) for row in rows]})
+    return 0
+
+
 def deploy_cmd(repo, args):
     """W4 deploy/recover helpers; never touches shared docker from this entrypoint."""
     task = task_row(repo, args.task) if getattr(args, 'task', None) else None
@@ -2335,6 +2389,7 @@ def parser():
     proj.add_argument('--mode',choices=['row','document'],default='row')
     proj.add_argument('--record');proj.add_argument('--replacement')
     proj.add_argument('--apply',action='store_true',help='Apply projected change via docs apply')
+    dproj=d.add_parser('projections');dproj.add_argument('--file')
     q=sub.add_parser('queue').add_subparsers(dest='queue_action',required=True)
     qe=q.add_parser('enqueue');qe.add_argument('--task',required=True);qe.add_argument('--expected-dev',required=True)
     qe.add_argument('--head');qe.add_argument('--priority',type=int,default=0);qe.add_argument('--request-id')
@@ -2392,6 +2447,8 @@ def main():
     if args.action in write_actions:
         if args.action == 'docs' and getattr(args, 'doc_action', None) == 'snapshot':
             pass  # snapshot is read-only CAS helper
+        elif args.action == 'docs' and getattr(args, 'doc_action', None) == 'projections':
+            pass  # projection ledger readout is read-only
         elif args.action == 'queue' and getattr(args, 'queue_action', None) == 'list':
             pass
         elif args.action == 'claim' and getattr(args, 'claim_action', None) in ('status', 'list'):
@@ -2463,6 +2520,7 @@ def main():
     elif args.action=='worktree-check':worktree_check(repo,args.path)
     elif args.action=='docs':
         if args.doc_action=='snapshot':docs_snapshot(repo,args)
+        elif args.doc_action=='projections':return docs_projections(repo,args)
         elif args.doc_action=='project':return docs_project(repo,args)
         else:return docs_apply(repo,args)
     elif args.action=='queue':

scripts/agentctl_w4.py

@@ -267,8 +267,14 @@ def apply_projection_locally(path: Path, change: dict) -> dict:
         after = replacement
     else:
         lines = before.splitlines()
-        matches = [i for i, line in enumerate(lines)
-                   if record_id and record_id in [c.strip() for c in line.split('|') if c.strip()]]
+        matches = []
+        for i, line in enumerate(lines):
+            if line.lstrip(' ').startswith('|'):
+                cells = [cell.strip() for cell in line.strip().strip('|').split('|')]
+                if record_id and record_id in cells[:2]:
+                    matches.append(i)
+        if len(matches) > 1:
+            raise RuntimeError('AMBIGUOUS_RECORD: record id is not unique in this document')
         current_row = lines[matches[0]] if matches else None
         live_hash = content_digest(current_row) if current_row else None
         if expected != live_hash:
@@ -555,28 +561,27 @@ def cleanup_visibility_report(
 
 
 def project_cleanup_markdown(entries):
-    """Render cleanup visibility projection from structured entries.
+    """Render cleanup visibility markdown from structured entries (report only).
 
-    Ported for W2-U5 projection reuse; structured report remains authoritative.
+    This is a readout helper for operators; it is not a docs-apply write path.
     """
     lines = [
         '# Cleanup visibility projection',
         '',
-        '> Generated by `agentctl docs project --kind cleanup`. '
-        'Not a lock; machine `.locks` is authoritative.',
+        '> Generated cleanup report render. Not a lock; machine `.locks` is authoritative.',
         '',
-        '| path | branch | reason | cleanup_owner |',
+        '| 路径/事项 | 分类与保留原因 | cleanup_owner | 复查触发 |',
         '| --- | --- | --- | --- |',
     ]
     if not entries:
-        lines.append('| — | — | none_pending | — |')
+        lines.append('| — | none_pending | — | — |')
     else:
         for item in entries:
-            path = str(item.get('path') or '').replace('|', '\\|')
-            branch = str(item.get('branch') or '').replace('|', '\\|')
-            reason = str(item.get('reason') or '').replace('|', '\\|')
+            path = str(item.get('path') or item.get('item') or '').replace('|', '\\|')
+            reason = str(item.get('reason') or item.get('category') or '').replace('|', '\\|')
             owner = str(item.get('cleanup_owner') or 'independent-maintainer').replace('|', '\\|')
-            lines.append(f'| {path} | {branch} | {reason} | {owner} |')
+            trigger = str(item.get('trigger') or item.get('复查触发') or '').replace('|', '\\|')
+            lines.append(f'| {path} | {reason} | {owner} | {trigger} |')
     lines.append('')
     return '\n'.join(lines)
 

scripts/tests/test_agentctl.py

@@ -137,6 +137,11 @@ class AgentctlTests(unittest.TestCase):
         out, err = process.communicate(timeout=10)
         self.assertEqual(process.returncode, 0, out + err)
 
+    def _last_json(self, text):
+        lines = [line for line in text.splitlines() if line.strip()]
+        self.assertTrue(lines, text)
+        return json.loads(lines[-1])
+
     def docs_request(self, changes, request_id="request-alpha"):
         path = pathlib.Path(self.temp.name) / (request_id + ".json")
         path.write_text(json.dumps({"request_id": request_id, "changes": changes}), encoding="utf-8")
@@ -1105,6 +1110,227 @@ class AgentctlTests(unittest.TestCase):
         self.assertIn("manager_lock", status)
         self.assertIn("queue", status)
 
+    def _write_current_with_sections(self, task_path, *, claim_rows=None, slot_body=None):
+        """CURRENT fixture with human sections + integration slot for protection tests."""
+        rows = claim_rows or [
+            "| owner | task | branch | worktree | mutex | status |",
+            "| --- | --- | --- | --- | --- | --- |",
+            "| Test Agent | alpha | feature/alpha | .worktrees/alpha | - | planned |",
+            "| Test Agent | beta | feature/beta | .worktrees/beta | - | planned |",
+        ]
+        slot = slot_body or (
+            "| holder | phase | since | scope | queue |\n"
+            "|--------|-------|-------|-------|-------|\n"
+            "| - | idle | 2026-09-23 | test | - |\n"
+        )
+        body = (
+            "# Coordination\n\n"
+            "## 焦点\n\nfocus line\n\n"
+            "## 并行认领\n\n"
+            + "\n".join(rows) + "\n\n"
+            "## 清理接管与遗留现场\n\n"
+            "| 路径/事项 | 分类与保留原因 | cleanup_owner | 复查触发 |\n"
+            "| --- | --- | --- | --- |\n"
+            "| none | none | - | - |\n\n"
+            "## 集成槽位\n\n"
+            + slot + "\n"
+            "## 受控约束\n\n- constraint\n\n"
+            "## 活跃入口\n\n- entry\n"
+        )
+        path = task_path / ".doc/ops/CURRENT.md"
+        path.write_text(body, encoding="utf-8")
+        return path
+
+    def _project_apply_with_retry(self, task_id, cwd, *extra, attempts=40):
+        import time
+        last = None
+        for _ in range(attempts):
+            last = self.cli("docs", "project", "--task", task_id, "--apply", *extra,
+                            cwd=cwd, check=False)
+            if last.returncode == 0:
+                return last
+            err = last.stderr + last.stdout
+            if "LOCK_BUSY" in err:
+                time.sleep(0.05)
+                continue
+            return last
+        return last
+
+    def test_docs_project_parallel_apply_keeps_both_rows(self):
+        import concurrent.futures
+        alpha = self.initialized_task(capabilities=("docs:publish",))
+        beta = self.initialized_task(task="beta", scopes=("src/b", ".doc"),
+                                     capabilities=("docs:publish",))
+        self._write_current_with_sections(alpha)
+        self.git("add", ".doc/ops/CURRENT.md", cwd=alpha)
+        self.git("commit", "-m", "sectioned current", cwd=alpha)
+        # Mirror committed baseline into beta worktree.
+        self.git("checkout", "feature/alpha", "--", ".doc/ops/CURRENT.md", cwd=beta)
+        self.git("add", ".doc/ops/CURRENT.md", cwd=beta)
+        self.git("commit", "-m", "sync current", cwd=beta)
+
+        def run(task_id, cwd, status):
+            return self._project_apply_with_retry(
+                task_id, cwd, "--mutex", task_id, "--status", status)
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+            fut_a = pool.submit(run, "alpha", alpha, "in_progress")
+            fut_b = pool.submit(run, "beta", beta, "blocked")
+            results = [fut_a.result(), fut_b.result()]
+        for result in results:
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+            payload = self._last_json(result.stdout)
+            self.assertEqual(payload["status"], "projected_and_applied")
+        text_a = (alpha / ".doc/ops/CURRENT.md").read_text(encoding="utf-8")
+        text_b = (beta / ".doc/ops/CURRENT.md").read_text(encoding="utf-8")
+        self.assertRegex(text_a, r"\| Test Agent \| alpha \|.*\| in_progress \|")
+        self.assertRegex(text_a, r"\| Test Agent \| beta \|")
+        self.assertRegex(text_b, r"\| Test Agent \| beta \|.*\| blocked \|")
+        self.assertRegex(text_b, r"\| Test Agent \| alpha \|")
+
+    def test_docs_project_same_row_race_base_changed_no_partial(self):
+        import concurrent.futures
+        task = self.initialized_task(capabilities=("docs:publish",))
+        self._write_current_with_sections(task)
+        self.git("add", ".doc/ops/CURRENT.md", cwd=task)
+        self.git("commit", "-m", "sectioned current", cwd=task)
+        snap = json.loads(self.cli(
+            "docs", "snapshot", "--file", ".doc/ops/CURRENT.md",
+            "--record", "alpha", cwd=task).stdout)
+        row = snap["row"]
+        self.assertIsNotNone(row)
+        req_a = self.docs_request(
+            [self.doc_change(
+                ".doc/ops/CURRENT.md", "alpha", row,
+                "| Test Agent | alpha | feature/alpha | .worktrees/alpha | - | in_progress |")],
+            "same-row-a")
+        req_b = self.docs_request(
+            [self.doc_change(
+                ".doc/ops/CURRENT.md", "alpha", row,
+                "| Test Agent | alpha | feature/alpha | .worktrees/alpha | - | done |")],
+            "same-row-b")
+        baseline = (task / ".doc/ops/CURRENT.md").read_bytes()
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+            r1 = pool.submit(lambda: self.cli(
+                "docs", "apply", "--task", "alpha", "--request", req_a,
+                cwd=task, check=False))
+            r2 = pool.submit(lambda: self.cli(
+                "docs", "apply", "--task", "alpha", "--request", req_b,
+                cwd=task, check=False))
+            results = [r1.result(), r2.result()]
+        codes = sorted(r.returncode for r in results)
+        self.assertEqual(codes[0], 0, [(r.returncode, r.stderr) for r in results])
+        self.assertNotEqual(codes[1], 0)
+        loser = next(r for r in results if r.returncode != 0)
+        if "LOCK_BUSY" in loser.stderr:
+            detail = json.loads(loser.stderr)
+            rid = detail.get("request_id")
+            retry_req = req_a if rid == "same-row-a" else req_b
+            loser = self.cli("docs", "apply", "--task", "alpha", "--request", retry_req,
+                             cwd=task, check=False)
+        self.assertNotEqual(loser.returncode, 0, loser.stderr)
+        self.assertIn("BASE_CHANGED", loser.stderr)
+        after = (task / ".doc/ops/CURRENT.md").read_text(encoding="utf-8")
+        self.assertEqual(after.count("| alpha |"), 1)
+        self.assertIn("## 焦点", after)
+        self.assertIn("focus line", after)
+        self.assertNotEqual(after.encode(), baseline)
+        # Exactly one of the two statuses landed.
+        self.assertTrue(("| in_progress |" in after) ^ ("| done |" in after))
+
+    def test_admin_document_cannot_overwrite_integration_slot(self):
+        task = self.initialized_task(capabilities=("docs:publish", "docs:admin"))
+        path = self._write_current_with_sections(task)
+        self.git("add", ".doc/ops/CURRENT.md", cwd=task)
+        self.git("commit", "-m", "sectioned current", cwd=task)
+        original = path.read_text(encoding="utf-8")
+        snap = json.loads(self.cli(
+            "docs", "snapshot", "--file", ".doc/ops/CURRENT.md",
+            "--record", "@document", cwd=task).stdout)
+        mutated = original.replace("| - | idle | 2026-09-23 | test | - |",
+                                   "| alice | merging | 2026-09-23 | stolen | - |")
+        self.assertNotEqual(mutated, original)
+        request = self.docs_request([{
+            "file": ".doc/ops/CURRENT.md",
+            "record_id": "@document",
+            "expected_sha256": snap["sha256"],
+            "replacement": mutated,
+        }], "slot-steal")
+        result = self.cli("docs", "apply", "--task", "alpha", "--request", request,
+                          cwd=task, check=False)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("PATH_OUT_OF_SCOPE", result.stderr)
+        self.assertEqual(path.read_text(encoding="utf-8"), original)
+
+    def test_admin_document_cannot_overwrite_human_sections(self):
+        task = self.initialized_task(capabilities=("docs:publish", "docs:admin"))
+        path = self._write_current_with_sections(task)
+        self.git("add", ".doc/ops/CURRENT.md", cwd=task)
+        self.git("commit", "-m", "sectioned current", cwd=task)
+        original = path.read_text(encoding="utf-8")
+        snap = json.loads(self.cli(
+            "docs", "snapshot", "--file", ".doc/ops/CURRENT.md",
+            "--record", "@document", cwd=task).stdout)
+        for old, new, label in (
+            ("focus line", "machine rewrote focus", "focus"),
+            ("| none | none | - | - |", "| hacked | x | y | z |", "cleanup"),
+            ("- constraint", "- machine constraint", "constraints"),
+            ("- entry", "- machine entry", "entries"),
+        ):
+            with self.subTest(section=label):
+                mutated = original.replace(old, new)
+                self.assertNotEqual(mutated, original)
+                request = self.docs_request([{
+                    "file": ".doc/ops/CURRENT.md",
+                    "record_id": "@document",
+                    "expected_sha256": snap["sha256"],
+                    "replacement": mutated,
+                }], "human-" + label)
+                result = self.cli("docs", "apply", "--task", "alpha", "--request", request,
+                                  cwd=task, check=False)
+                self.assertNotEqual(result.returncode, 0, result.stderr)
+                self.assertIn("PATH_OUT_OF_SCOPE", result.stderr)
+                self.assertEqual(path.read_text(encoding="utf-8"), original)
+
+    def test_docs_project_preview_apply_receipts_and_projections(self):
+        task = self.initialized_task(capabilities=("docs:publish",))
+        self._write_current_with_sections(task)
+        self.git("add", ".doc/ops/CURRENT.md", cwd=task)
+        self.git("commit", "-m", "sectioned current", cwd=task)
+        preview = self._last_json(self.cli(
+            "docs", "project", "--task", "alpha", "--mutex", "m1",
+            "--status", "in_progress", cwd=task).stdout)
+        self.assertEqual(preview["status"], "projected")
+        self.assertIn("request_id", preview["request"])
+        self.assertTrue(preview["request"]["request_id"].startswith("alpha-project-"))
+        replay = self._last_json(self.cli(
+            "docs", "project", "--task", "alpha", "--mutex", "m1",
+            "--status", "in_progress", cwd=task).stdout)
+        self.assertEqual(replay["request"]["request_id"], preview["request"]["request_id"])
+        bad = self.cli("docs", "project", "--task", "alpha", "--status", "queued",
+                       cwd=task, check=False)
+        self.assertNotEqual(bad.returncode, 0)
+        self.assertIn("INVALID_REQUEST", bad.stderr)
+        applied = self._last_json(self.cli(
+            "docs", "project", "--task", "alpha", "--apply", "--mutex", "m1",
+            "--status", "in_progress", cwd=task).stdout)
+        self.assertEqual(applied["status"], "projected_and_applied")
+        text = (task / ".doc/ops/CURRENT.md").read_text(encoding="utf-8")
+        self.assertRegex(text, r"\| Test Agent \| alpha \|.*\| in_progress \|")
+        import sqlite3
+        with sqlite3.connect(self.root / ".locks" / "state.sqlite") as db:
+            receipt = db.execute(
+                "SELECT hash FROM receipts WHERE task=? AND file=?",
+                ("alpha", ".doc/ops/CURRENT.md")).fetchone()
+            self.assertIsNotNone(receipt)
+            self.assertEqual(receipt[0], hashlib.sha256(text.encode()).hexdigest())
+        ledger = self._last_json(self.cli(
+            "docs", "projections", "--file", ".doc/ops/CURRENT.md", cwd=task).stdout)
+        self.assertEqual(ledger["status"], "ok")
+        self.assertTrue(any(p["record_id"] == "alpha" and p["task"] == "alpha"
+                            for p in ledger["projections"]))
+
 
 class ProcessPlatformTests(unittest.TestCase):
     @classmethod

scripts/tests/test_agentctl_w4.py

@@ -154,20 +154,45 @@ class AgentctlW4Tests(unittest.TestCase):
         self.assertEqual(receipt.get("recover_reason"),
                          "operator cleared after fake failure")
 
-    def test_deploy_phase_transition_rejects_illegal_jump(self):
-        deploy = self.w4.deployment_create(
-            self.db, deploy_id="dep-phase", task="phase-task",
-            env="dev", expected_commit="abcdef4", manifest=None, dry_run=True)
-        self.assertEqual(deploy["phase"], "prepared")
-        self.w4.deployment_set_phase(self.db, "dep-phase", "deploying")
-        self.w4.deployment_set_phase(self.db, "dep-phase", "accepting")
-        passed = self.w4.deployment_set_phase(self.db, "dep-phase", "passed")
-        self.assertEqual(passed["phase"], "passed")
+    def test_row_projection_ambiguous_record_refuses(self):
+        doc = (
+            "# Coordination\n\n"
+            "| owner | task | branch | state |\n"
+            "| --- | --- | --- | --- |\n"
+            "| Alice | alpha | feature/alpha | active |\n"
+            "| Bob | alpha | feature/alpha-2 | active |\n"
+        )
+        self._write_current(doc)
+        change = self.w4.build_row_projection_change(
+            file=".doc/ops/CURRENT.md",
+            record_id="alpha",
+            current_row="| Alice | alpha | feature/alpha | active |",
+            replacement="| Alice | alpha | feature/alpha | done |",
+        )
         with self.assertRaises(RuntimeError) as ctx:
-            self.w4.deployment_set_phase(self.db, "dep-phase", "deploying")
-        self.assertIn("illegal transition passed -> deploying", str(ctx.exception))
-        still = self.w4.deployment_get(self.db, "dep-phase")
-        self.assertEqual(still["phase"], "passed")
+            self.w4.apply_projection_locally(self.root, change)
+        self.assertIn("AMBIGUOUS_RECORD", str(ctx.exception))
+
+    def test_project_cleanup_markdown_header_is_report_only(self):
+        rendered = self.w4.project_cleanup_markdown([{
+            "path": ".worktrees/example",
+            "reason": "cleanup=pending",
+            "cleanup_owner": "maintainer",
+            "trigger": "tip ancestor",
+        }])
+        self.assertIn("| 路径/事项 | 分类与保留原因 | cleanup_owner | 复查触发 |", rendered)
+        self.assertNotIn("--kind cleanup", rendered)
+        self.assertIn(".worktrees/example", rendered)
+        self.assertIn("tip ancestor", rendered)
+
+    def test_build_projection_row_canonical(self):
+        row = self.w4.build_projection_row(
+            task_id="alpha", owner="Test Agent", branch="feature/alpha",
+            worktree=".worktrees/alpha", mutex="m1", status="in_progress")
+        self.assertEqual(
+            row,
+            "| Test Agent | alpha | feature/alpha | .worktrees/alpha | m1 | in_progress |")
+
 
 
 class DarwinCwdVisibilityTests(unittest.TestCase):