From cf19c0b5e4b37eafb24c1ff8f0dee802d2d60ed3 Mon Sep 17 00:00:00 2001
From: hyk <4408344+hhyykk@users.noreply.github.com>
Date: Mon, 5 Oct 2026 11:51:33 +0800
Subject: [PATCH 1/2] fix(work-items): fence reviewed sources through team-plan
 commit

Signed-off-by: hyk <4408344+hhyykk@users.noreply.github.com>
---
 .../work_items/team_plan_adapter.py           |  82 ++++++--
 .../work_items/team_plan_authority.ts         |  71 ++++++-
 .../project_registry_io_manifest_v1.json      |  14 +-
 .../control_plane/test_team_plan_authority.py | 191 +++++++++++++++++-
 tests/control_plane_ts/team_plan.test.ts      |  13 ++
 .../team_plan_commit_probe.ts                 |  22 ++
 tests/test_steward_team_plan_apply.py         |  24 +++
 7 files changed, 381 insertions(+), 36 deletions(-)
 create mode 100644 tests/control_plane_ts/team_plan_commit_probe.ts

diff --git a/loopx/control_plane/work_items/team_plan_adapter.py b/loopx/control_plane/work_items/team_plan_adapter.py
index 0b13b3fba1..d580b165d2 100644
--- a/loopx/control_plane/work_items/team_plan_adapter.py
+++ b/loopx/control_plane/work_items/team_plan_adapter.py
@@ -11,18 +11,25 @@
 import json
 import re
 from collections.abc import Callable, Mapping
+from contextlib import ExitStack
 from pathlib import Path
 from typing import Any
 
 from ...agent_registry import registered_agent_ids_for_goal
 from ...history import load_registry
+from ...file_lock import (
+    LockAcquireTimeoutError, cross_runtime_lock_witness, exclusive_cross_runtime_file_lock,
+)
 from ...registry import find_registry_goal
 from ...state_refresh import now_local
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from ..runtime.public_safety import validate_public_safe_value
 from ..coordination.local_authority import read_canonical_todos_if_promoted
 from ..coordination.local_authority_shadow_adapter import effective_runtime_root
-from ..coordination.legacy_writer_fence import legacy_todo_write_transaction
+from ..coordination.legacy_writer_fence import (
+    legacy_coordination_todo_lock_path, legacy_todo_write_transaction,
+)
+from ..coordination.shadow_management import shadow_maintenance_lock_target
 from ..coordination.runtime_shadow_writer_adapter import (
     begin_todo_runtime_shadow_capture, write_captured_todo_state,
     settle_todo_runtime_shadow_capture,
@@ -136,20 +143,34 @@ def apply_team_plan(
     validate_public_safe_value(dict(proposal), path="team_plan")
     registry_path = Path(registry_path).expanduser()
     runtime = effective_runtime_root(registry_path, None)
-    goal = find_registry_goal(load_registry(registry_path), goal_id)
-    if goal is None:
+    if find_registry_goal(load_registry(registry_path), goal_id) is None:
         raise ValueError("steward team plan proposal names an unknown Goal")
     project, state = resolve_todo_state_path(registry_path=registry_path, goal_id=goal_id)
     request = {
         "goal_id": goal_id, "plan": dict(proposal),
         "actor_agent_id": agent_id,
-        "registered_agents": registered_agent_ids_for_goal(goal),
         "supported_action_kinds": sorted(TODO_ACTION_KIND_ADVANCEMENT_VALUES),
         "observed_at": now_local(),
         "expected_state_fingerprint": expected_state_fingerprint,
-        "intent_basis": steward_team_plan_intent_basis(
-            goal_id=goal_id, goal=goal, registry_path=registry_path, plan=proposal),
     }
+
+    def read_source_request() -> None:
+        # Called under the registry and state locks, including on the legacy
+        # path. A registry rebind while acquiring the guards must not redirect
+        # the operation to an unguarded source or provider.
+        goal = find_registry_goal(load_registry(registry_path), goal_id)
+        if goal is None:
+            raise ValueError("steward team plan proposal names an unknown Goal")
+        _, current_state = resolve_todo_state_path(registry_path=registry_path, goal_id=goal_id)
+        if (effective_runtime_root(registry_path, None).resolve() != runtime.resolve()
+                or current_state.resolve() != state.resolve()):
+            raise TeamPlanCommitError("team_plan_preview_stale", "team plan source binding changed after preview")
+        request.update(
+            registered_agents=registered_agent_ids_for_goal(goal),
+            intent_basis=steward_team_plan_intent_basis(
+                goal_id=goal_id, goal=goal, registry_path=registry_path, plan=proposal),
+        )
+
     identity = effect_runtime_result("work_items.team_plan.identity", request)
     marker = f"<!-- loopx:team-plan:{identity['operation_id'].split(':')[1]} "
 
@@ -163,17 +184,40 @@ def previous_receipt(original: str) -> dict[str, Any] | None:
 
     canonical = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal_id)
     if canonical is not None:
-        # Receipts created before promotion remain immutable historical facts.
-        # Read only; never fall back to a legacy writer for a promoted goal.
-        previous = previous_receipt(state.read_text(encoding="utf-8"))
-        if previous is not None:
-            return dict(effect_runtime_result("work_items.team_plan.plan", {
-                **request, "previous_receipt": previous})["result"])
-        response = effect_runtime_result("work_items.team_plan.commit", {
-            **request, "runtime_root": str(runtime.resolve()),
-            "expected_provider_revision": canonical["provider_revision"],
-            "current_state_fingerprint": read_fingerprint() if read_fingerprint else None,
-        })
+        targets = (
+            shadow_maintenance_lock_target(runtime, goal_id),
+            legacy_coordination_todo_lock_path(runtime_root=runtime, goal_id=goal_id),
+            state, registry_path,
+        )
+        with ExitStack() as guards:
+            for target in targets[:-1]:
+                guards.enter_context(exclusive_cross_runtime_file_lock(target, operation="team_plan_commit"))
+            try:
+                # Registry administration can hold R before T/S. Like promotion,
+                # never wait for R while holding source locks in reverse order.
+                guards.enter_context(exclusive_cross_runtime_file_lock(
+                    registry_path, operation="team_plan_commit", timeout_seconds=0))
+            except LockAcquireTimeoutError:
+                raise TeamPlanCommitError("team_plan_commit_failed",
+                    "team plan registry is busy; retry the original confirmation") from None
+            read_source_request()
+            # Pre-promotion receipts remain historical facts; never fall back
+            # to a legacy writer for a promoted Goal.
+            previous = previous_receipt(state.read_text(encoding="utf-8"))
+            if previous is not None:
+                return dict(effect_runtime_result("work_items.team_plan.plan", {
+                    **request, "previous_receipt": previous})["result"])
+            response = effect_runtime_result("work_items.team_plan.commit", {
+                **request, "runtime_root": str(runtime.resolve()),
+                "expected_provider_revision": canonical["provider_revision"],
+                "current_state_fingerprint": read_fingerprint() if read_fingerprint else None,
+                "source": {
+                    "registry": str(registry_path.resolve()), "state_file": str(state.resolve()),
+                    "registry_sha256": hashlib.sha256(registry_path.read_bytes()).hexdigest(),
+                    "state_sha256": hashlib.sha256(state.read_bytes()).hexdigest(),
+                },
+                "locks": [cross_runtime_lock_witness(target) for target in targets],
+            })
         if response.get("status") not in {"applied", "recovered", "replayed", "no_change"}:
             raise TeamPlanCommitError(str(response.get("reason_code") or "team_plan_commit_failed"),
                 str(response.get("reason") or "team plan commit could not be verified"),
@@ -186,8 +230,10 @@ def previous_receipt(original: str) -> dict[str, Any] | None:
         result["projection_delivery"] = projected.get("projection_delivery")
         return result
 
-    with legacy_todo_write_transaction(registry_path, goal_id, state, agent_id,
+    with exclusive_cross_runtime_file_lock(registry_path, operation="team_plan_commit"), \
+            legacy_todo_write_transaction(registry_path, goal_id, state, agent_id,
             "todo_add", False, runtime_root=runtime):
+        read_source_request()
         original = state.read_text(encoding="utf-8")
         previous = previous_receipt(original)
         existing = [metadata for line in original.splitlines()
diff --git a/loopx/control_plane/work_items/team_plan_authority.ts b/loopx/control_plane/work_items/team_plan_authority.ts
index 90a3cbba66..4957e4e071 100644
--- a/loopx/control_plane/work_items/team_plan_authority.ts
+++ b/loopx/control_plane/work_items/team_plan_authority.ts
@@ -4,16 +4,23 @@ import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import type {JsonObject} from "../effect_program.ts";
 import type {AuthorityStore} from "../coordination/authority_store.ts";
 import {authorityStoreSourceAuthority} from "../coordination/authority_store.ts";
-import {requireJsonObject} from "../runtime_decode.ts";
+import {requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts";
 import {canonicalAuthorityObject} from "../coordination/authority_store_codec.ts";
 import {CoordinationCommandReceipt} from "../coordination/command_receipt.ts";
 import {indexCoordinationProjection, prepareCoordinationProjectionCommit, validateCoordinationTodoReadModel} from "../coordination/coordination_projection.ts";
-import {withCanonicalWriter} from "../coordination/local_authority_write.ts";
-import {openLocalAuthorityStore, localAuthorityOpenFailure} from "../coordination/local_authority_provider.ts";
+import {openLocalAuthorityStore, localAuthorityOpenFailure, requireLocalAuthorityRuntimeRoot} from "../coordination/local_authority_provider.ts";
+import {type AuthoritySourceCheck, uncheckedAuthoritySource} from "../coordination/authority_source.ts";
+import {legacyCoordinationTodoLockPath} from "../coordination/legacy_writer_lock_paths.ts";
+import {requireShadowPrimaryWriteAllowed, shadowMaintenanceLockPath} from "../coordination/shadow_management.ts";
+import {claimFileMutationLock, mutationLockOwner, releaseFileMutationLock,
+  releaseFileMutationLockClaim, type FileMutationLockClaim} from "../effect_runtime_io.ts";
 import {planTeamTransaction, replayTeamTransaction, teamTransactionIdentity, TEAM_TRANSACTION_SCHEMA} from "./team_plan.ts";
-import {isAbsolute} from "node:path";
+import {createHash} from "node:crypto";
+import {readFile} from "node:fs/promises";
+import {isAbsolute, resolve} from "node:path";
 
-export async function commitTeamPlan(store: AuthorityStore, request: JsonObject): Promise<JsonObject> {
+export async function commitTeamPlan(store: AuthorityStore, request: JsonObject,
+  sourceCheck: AuthoritySourceCheck = uncheckedAuthoritySource): Promise<JsonObject> {
   const identity = teamTransactionIdentity(request);
   const receipt = new CoordinationCommandReceipt({result_schema: TEAM_TRANSACTION_SCHEMA, identity,
     failure: (reason_code, reason) => ({schema_version: TEAM_TRANSACTION_SCHEMA, status: "failed", reason_code, reason}),
@@ -27,6 +34,9 @@ export async function commitTeamPlan(store: AuthorityStore, request: JsonObject)
   if (observation.kind === "receipt") return observation.result;
   const head = observation.authority;
   if (head.status !== "loaded") return {...head};
+  if (!await sourceCheck()) {
+    return {status: "failed", reason_code: "team_plan_preview_stale", reason: "team plan sources changed after preview"};
+  }
   // The canonical revision is included at preview. Stale work cannot be
   // admitted merely because its Markdown projection has not caught up yet.
   if (request.expected_provider_revision != null && request.expected_provider_revision !== head.provider_revision) {
@@ -46,17 +56,56 @@ export async function commitTeamPlan(store: AuthorityStore, request: JsonObject)
 
 export async function commitLocalTeamPlan(value: unknown): Promise<JsonObject> {
   const request = requireJsonObject(value, "team plan commit");
-  const root = String(request.runtime_root);
-  if (!isAbsolute(root)) throw new Error("runtime_root must be absolute");
+  const root = requireLocalAuthorityRuntimeRoot(request.runtime_root);
   const identity = teamTransactionIdentity(request);
+  const goal = String(identity.goal_id);
+  const claims: {target: string; token: string; claim: FileMutationLockClaim}[] = [];
+  let adopted = false;
   try {
-    return await withCanonicalWriter(root, String(identity.goal_id), false, async () => {
-      const store = await openLocalAuthorityStore(root, String(identity.goal_id));
-      return {...await commitTeamPlan(store, request), source_authority: authorityStoreSourceAuthority(store),
-        decision_read_from_provider: true, legacy_fallback_used: false};
+    const source = requireJsonObject(request.source, "team plan source");
+    const registry = requireNonEmptyString(source.registry, "registry");
+    const state = requireNonEmptyString(source.state_file, "state_file");
+    if (!isAbsolute(registry) || !isAbsolute(state) || resolve(registry) === resolve(state)) {
+      throw new EffectRuntimeRequestError("team plan source paths must be distinct absolute paths");
+    }
+    const targets = [shadowMaintenanceLockPath(root, goal), legacyCoordinationTodoLockPath(root, goal),
+      state, registry].map(path => resolve(path));
+    if (!Array.isArray(request.locks) || request.locks.length !== targets.length) {
+      throw new EffectRuntimeRequestError("team plan requires the complete source lock handoff");
+    }
+    const witnesses = request.locks.map((raw, i) => {
+      const witness = requireJsonObject(raw, "team plan lock witness");
+      if (witness.target !== targets[i]) throw new EffectRuntimeRequestError("team plan lock target mismatch");
+      return {target: targets[i]!, token: requireNonEmptyString(witness.token, "lock token"), pid: witness.pid};
+    });
+    for (const [i, witness] of witnesses.entries()) {
+      const claim = await claimFileMutationLock(witness.target, witness.token);
+      if (!claim) break;
+      claims.push({...witness, claim});
+      const owner = await mutationLockOwner(witness.target);
+      if (owner?.token !== witness.token || owner.pid !== witness.pid) break;
+      if (i === witnesses.length - 1) adopted = true;
+    }
+    const store = await openLocalAuthorityStore(root, goal);
+    const result = await commitTeamPlan(store, request, async () => {
+      // Exact committed receipts replay before these admission checks. An
+      // expired handoff may read history, but can never authorize a new write.
+      if (!adopted) throw new EffectRuntimeRequestError("team plan source lock handoff expired; retry the original confirmation");
+      await requireShadowPrimaryWriteAllowed(root, goal);
+      const digest = async (path: string) => createHash("sha256").update(await readFile(path)).digest("hex");
+      return await digest(registry) === source.registry_sha256 && await digest(state) === source.state_sha256;
     });
+    return {...result, source_authority: authorityStoreSourceAuthority(store),
+      decision_read_from_provider: true, legacy_fallback_used: false};
   } catch (error) {
     return {status: "failed", reason_code: error instanceof EffectRuntimeRequestError ? error.code : "team_plan_commit_failed",
       reason: error instanceof Error ? error.message : String(error), ...localAuthorityOpenFailure(error)};
+  } finally {
+    // Claims keep the source markers alive even after a requesting Python
+    // process exits. The native effect releases the handed-off critical section.
+    for (const entry of claims.reverse()) {
+      if (adopted) await releaseFileMutationLock(entry.target, entry.token, entry.claim, true);
+      else await releaseFileMutationLockClaim(entry.claim);
+    }
   }
 }
diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json
index 56243372b1..506c63892c 100644
--- a/loopx/semantics/project_registry_io_manifest_v1.json
+++ b/loopx/semantics/project_registry_io_manifest_v1.json
@@ -1759,15 +1759,23 @@
     },
     {
       "site": "loopx/control_plane/work_items/team_plan_adapter.py::<module>.apply_team_plan::codec_read:load_registry#1",
-      "line": 139,
-      "column": 31,
+      "line": 146,
+      "column": 27,
+      "kind": "codec_read",
+      "api": "load_registry",
+      "classification": "codec_api"
+    },
+    {
+      "site": "loopx/control_plane/work_items/team_plan_adapter.py::<module>.apply_team_plan.read_source_request::codec_read:load_registry#1",
+      "line": 161,
+      "column": 35,
       "kind": "codec_read",
       "api": "load_registry",
       "classification": "codec_api"
     },
     {
       "site": "loopx/control_plane/work_items/team_plan_adapter.py::<module>.team_plan_state_fingerprint::codec_read:load_registry#1",
-      "line": 78,
+      "line": 85,
       "column": 31,
       "kind": "codec_read",
       "api": "load_registry",
diff --git a/tests/control_plane/test_team_plan_authority.py b/tests/control_plane/test_team_plan_authority.py
index ecc5c97347..c7ec48c41f 100644
--- a/tests/control_plane/test_team_plan_authority.py
+++ b/tests/control_plane/test_team_plan_authority.py
@@ -1,10 +1,15 @@
 """Actual Chat entrypoint against a fenced FileAuthorityStore and projection IO."""
 import hashlib
 import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import time
 
 import pytest
 
-from canonical_authority_fixture import initialize_canonical_authority
+from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
 from loopx.chat_actions import ChatActionService
 from loopx.chat_action_store import ChatActionStore
 from loopx.control_plane.work_items.governed_transition_proposal import (
@@ -19,8 +24,10 @@
 from loopx.control_plane.todos import provider_projection
 
 
-@pytest.fixture
-def canonical_team(tmp_path):
+@pytest.fixture(params=["file", "sqlite"])
+def canonical_team(tmp_path, request, monkeypatch):
+    if request.param == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
     runtime = tmp_path / "runtime"
     state = tmp_path / "state.md"
     state.write_text('# Goal\n\n## Agent Todo\n\n')
@@ -31,7 +38,7 @@ def canonical_team(tmp_path):
     initialize_canonical_authority(runtime, "goal-a", {"goal_id": "goal-a", "todos": [], "leases": [],
         "todo_read_model": {"schema_version": TODO_DOMAIN_READ_RECORD_SCHEMA_VERSION,
             "contract_fields": list(TODO_DOMAIN_RECORD_FIELDS), "todo_count": 0,
-            "records_sha256": hashlib.sha256(canonical_bytes([])).hexdigest()}}, state_path=state)
+            "records_sha256": hashlib.sha256(canonical_bytes([])).hexdigest()}}, state_path=state, provider=request.param)
     service = ChatActionService(store=ChatActionStore(runtime / "chat/actions"), registry_path=registry)
     plan = {"schema_version": "steward_team_plan_preview_v0", "kind": "steward_team_plan_preview",
         "goal_id": "goal-a", "objective": "Independent lane outcomes", "quota_envelope": {"slots": 2},
@@ -43,6 +50,182 @@ def canonical_team(tmp_path):
     return runtime, state, service, preview
 
 
+@pytest.mark.parametrize("source", ["state", "registry"])
+def test_reviewed_sources_remain_locked_through_native_commit(canonical_team, monkeypatch, source):
+    """A compliant writer cannot replace reviewed facts after fingerprinting."""
+    from loopx.file_lock import exclusive_cross_runtime_file_lock, LockAcquireTimeoutError
+    from loopx.control_plane.work_items import team_plan_adapter
+
+    runtime, state, service, preview = canonical_team
+    target = state if source == "state" else service.registry_path
+    original_effect = team_plan_adapter.effect_runtime_result
+    checked = []
+
+    def inspect(method, request):
+        if method == "work_items.team_plan.commit":
+            with pytest.raises(LockAcquireTimeoutError):
+                with exclusive_cross_runtime_file_lock(target, timeout_seconds=0):
+                    pass
+            checked.append(source)
+        return original_effect(method, request)
+
+    monkeypatch.setattr(team_plan_adapter, "effect_runtime_result", inspect)
+    result = service.apply(preview["proposal_id"])["proposal"]
+    assert result["status"] == "applied", result
+    assert checked == [source]
+    assert len(read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")["todos"]) == 2
+    # Every path releases its source guards, allowing the next ordinary writer.
+    with exclusive_cross_runtime_file_lock(target, timeout_seconds=0):
+        pass
+
+
+@pytest.mark.parametrize("source", ["state", "registry"])
+def test_changed_source_during_handoff_creates_no_todos(canonical_team, monkeypatch, source):
+    """The native owner rejects changed bytes even if a writer bypasses locks."""
+    from loopx.control_plane.work_items import team_plan_adapter
+
+    runtime, state, service, preview = canonical_team
+    target = state if source == "state" else service.registry_path
+    original_effect = team_plan_adapter.effect_runtime_result
+
+    def replace(method, request):
+        if method == "work_items.team_plan.commit":
+            target.write_bytes(target.read_bytes() + b"\n")
+        return original_effect(method, request)
+
+    monkeypatch.setattr(team_plan_adapter, "effect_runtime_result", replace)
+    result = service.apply(preview["proposal_id"])["proposal"]
+    assert result["status"] == "stale", result
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")["todos"] == []
+
+
+def test_busy_registry_releases_source_locks_without_committing(canonical_team):
+    from loopx.file_lock import exclusive_cross_runtime_file_lock
+
+    runtime, state, service, preview = canonical_team
+    with exclusive_cross_runtime_file_lock(service.registry_path):
+        result = service.apply(preview["proposal_id"])["proposal"]
+        assert result["status"] == "failed", result
+        with exclusive_cross_runtime_file_lock(state, timeout_seconds=0):
+            pass
+        assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")["todos"] == []
+    assert service.apply(preview["proposal_id"])["proposal"]["status"] == "applied"
+
+
+@pytest.mark.parametrize("damage", ["missing", "token", "pid", "target"])
+def test_native_commit_requires_valid_source_lock_handoff(canonical_team, monkeypatch, damage):
+    from loopx.control_plane.work_items import team_plan_adapter
+
+    runtime, state, service, preview = canonical_team
+    original_effect = team_plan_adapter.effect_runtime_result
+
+    def tamper(method, request):
+        if method == "work_items.team_plan.commit":
+            if damage == "missing":
+                request.pop("locks")
+            elif damage == "token":
+                request["locks"][0]["token"] = "expired-token"
+            elif damage == "pid":
+                request["locks"][0]["pid"] = -1
+            else:
+                request["locks"][0]["target"] = str(state)
+        return original_effect(method, request)
+
+    monkeypatch.setattr(team_plan_adapter, "effect_runtime_result", tamper)
+    result = service.apply(preview["proposal_id"])["proposal"]
+    assert result["status"] == "failed", result
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")["todos"] == []
+
+
+def test_native_lost_response_replays_after_handoff_expires(canonical_team, monkeypatch):
+    from loopx.control_plane.work_items import team_plan_adapter
+
+    runtime, state, service, preview = canonical_team
+    original_effect = team_plan_adapter.effect_runtime_result
+    captured = []
+
+    def capture(method, request):
+        if method == "work_items.team_plan.commit":
+            captured.append(request)
+        return original_effect(method, request)
+
+    monkeypatch.setattr(team_plan_adapter, "effect_runtime_result", capture)
+    assert service.apply(preview["proposal_id"])["proposal"]["status"] == "applied"
+    before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")
+    state.write_text(state.read_text() + "\nOwner revised the objective.\n")
+    replay = original_effect("work_items.team_plan.commit", captured[0])
+    assert replay["status"] == "replayed", replay
+    assert replay["result"]["action"] == "reused"
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")["provider_revision"] == before["provider_revision"]
+
+
+def test_dead_caller_cannot_release_native_source_claims(canonical_team, tmp_path):
+    """Kill the Python caller after adoption, before the real provider CAS."""
+    from loopx.file_lock import exclusive_mutation_file_lock, LockAcquireTimeoutError
+
+    runtime, state, service, preview = canonical_team
+    repo = Path(__file__).resolve().parents[2]
+    barrier = tmp_path / "caller-exit"
+    barrier.mkdir()
+    authority = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")
+    provider = "sqlite" if authority["source_authority"] == "sqlite_v0" else "file"
+    script = r'''
+import json, os, subprocess, sys, time
+from pathlib import Path
+from loopx.chat_actions import ChatActionService
+from loopx.chat_action_store import ChatActionStore
+from loopx.control_plane.work_items import team_plan_adapter
+registry, runtime, proposal, barrier, provider, probe = sys.argv[1:]
+barrier = Path(barrier)
+original = team_plan_adapter.effect_runtime_result
+def native(method, request):
+    if method != "work_items.team_plan.commit":
+        return original(method, request)
+    (barrier / "request.json").write_text(json.dumps(request))
+    with (barrier / "native.log").open("w") as output:
+        child = subprocess.Popen(["node", "--no-warnings", "--experimental-strip-types", probe, str(barrier), provider],
+                                 stdout=output, stderr=output)
+    deadline = time.monotonic() + 15
+    while not (barrier / "adopted").exists():
+        assert child.poll() is None, (barrier / "native.log").read_text()
+        assert time.monotonic() < deadline, "native adoption timed out"
+        time.sleep(.01)
+    os._exit(0)
+team_plan_adapter.effect_runtime_result = native
+ChatActionService(store=ChatActionStore(Path(runtime) / "chat/actions"), registry_path=Path(registry)).apply(proposal)
+'''
+    owner = subprocess.Popen([sys.executable, "-c", script, str(service.registry_path), str(runtime),
+        preview["proposal_id"], str(barrier), provider, str(repo / "tests/control_plane_ts/team_plan_commit_probe.ts")],
+        cwd=repo, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+        env={**os.environ, "PYTHONPATH": str(repo)})
+    try:
+        stdout, stderr = owner.communicate(timeout=20)
+        assert owner.returncode == 0, stdout + stderr
+        request = json.loads((barrier / "request.json").read_text())
+        for witness in request["locks"]:
+            with pytest.raises(LockAcquireTimeoutError):
+                with exclusive_mutation_file_lock(Path(witness["target"]), timeout_seconds=0):
+                    pytest.fail("dead caller's native claim was stolen")
+        (barrier / "release").touch()
+        deadline = time.monotonic() + 15
+        while not (barrier / "finished").exists():
+            assert time.monotonic() < deadline, (barrier / "native.log").read_text()
+            time.sleep(.01)
+        assert json.loads((barrier / "result.json").read_text())["status"] == "applied"
+        before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")
+        assert len(before["todos"]) == 2
+        # Retry repairs the display after caller exit; no lane is admitted twice.
+        result = service.apply(preview["proposal_id"])["proposal"]
+        assert result["status"] == "applied", result
+        assert state.read_text().count("Same work") == 2
+        assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")["provider_revision"] == before["provider_revision"]
+    finally:
+        (barrier / "release").touch()
+        if owner.poll() is None:
+            owner.kill()
+            owner.communicate()
+
+
 @pytest.mark.parametrize("failure_point", ["write", "confirmation"])
 def test_canonical_chat_commit_replays_after_projection_failure(canonical_team, monkeypatch, failure_point):
     runtime, state, service, preview = canonical_team
diff --git a/tests/control_plane_ts/team_plan.test.ts b/tests/control_plane_ts/team_plan.test.ts
index 58ee6d53d2..a2b4dad6e5 100644
--- a/tests/control_plane_ts/team_plan.test.ts
+++ b/tests/control_plane_ts/team_plan.test.ts
@@ -126,6 +126,19 @@ for (const provider of providers) {
     assert.equal(result.status, "ambiguous");
     assert.deepEqual(await store.loadAuthority(), before);
   });
+  test(`${provider}: source admission fences new writes and never replaces historical replay`, options, async t => {
+    const store = await fixture(t, provider); await seed(store);
+    const before = await store.loadAuthority();
+    const rejected = await commitTeamPlan(store, request(), async () => false);
+    assert.equal(rejected.reason_code, "team_plan_preview_stale");
+    assert.deepEqual(await store.loadAuthority(), before);
+    assert.equal((await commitTeamPlan(store, request(), async () => true)).status, "applied");
+    const committed = await store.loadAuthority();
+    assert.equal((await commitTeamPlan(store, request(), async () => {
+      throw new Error("historical replay must not read current source facts");
+    })).status, "replayed");
+    assert.deepEqual(await store.loadAuthority(), committed);
+  });
 }
 
 test("preview stays inert and rejects unenforced policy claims", () => {
diff --git a/tests/control_plane_ts/team_plan_commit_probe.ts b/tests/control_plane_ts/team_plan_commit_probe.ts
new file mode 100644
index 0000000000..53f87b7032
--- /dev/null
+++ b/tests/control_plane_ts/team_plan_commit_probe.ts
@@ -0,0 +1,22 @@
+/** Stop at the real provider CAS after native source-lock adoption. */
+import {existsSync, readFileSync, writeFileSync} from "node:fs";
+import {join} from "node:path";
+import {FileAuthorityStore} from "../../loopx/control_plane/coordination/file_authority_store.ts";
+import {SqliteAuthorityStore} from "../../loopx/control_plane/coordination/sqlite_authority_store.ts";
+import {commitLocalTeamPlan} from "../../loopx/control_plane/work_items/team_plan_authority.ts";
+
+const [barrier, provider] = process.argv.slice(2) as [string, string];
+const prototype = provider === "sqlite" ? SqliteAuthorityStore.prototype : FileAuthorityStore.prototype;
+const original = prototype.commitAuthority;
+prototype.commitAuthority = async function(commit) {
+  writeFileSync(join(barrier, "adopted"), String(process.pid));
+  const deadline = Date.now() + 20000;
+  while (!existsSync(join(barrier, "release"))) {
+    if (Date.now() >= deadline) throw new Error("team plan test barrier timed out");
+    Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 10);
+  }
+  return await original.call(this as FileAuthorityStore & SqliteAuthorityStore, commit);
+};
+const result = await commitLocalTeamPlan(JSON.parse(readFileSync(join(barrier, "request.json"), "utf8")));
+writeFileSync(join(barrier, "result.json"), JSON.stringify(result));
+writeFileSync(join(barrier, "finished"), "");
diff --git a/tests/test_steward_team_plan_apply.py b/tests/test_steward_team_plan_apply.py
index 7a5ae8884f..910e95817b 100644
--- a/tests/test_steward_team_plan_apply.py
+++ b/tests/test_steward_team_plan_apply.py
@@ -110,6 +110,30 @@ def _basis(registry_path: Path) -> str:
     )
 
 
+def test_legacy_registry_stays_locked_through_plan_and_write(tmp_path, monkeypatch):
+    from loopx.file_lock import exclusive_cross_runtime_file_lock, LockAcquireTimeoutError
+    from loopx.control_plane.work_items import team_plan_adapter
+
+    _project, registry = _fixture(tmp_path)
+    original_effect = team_plan_adapter.effect_runtime_result
+    checked = []
+
+    def inspect(method, request):
+        if method == "work_items.team_plan.plan":
+            with pytest.raises(LockAcquireTimeoutError):
+                with exclusive_cross_runtime_file_lock(registry, timeout_seconds=0):
+                    pass
+            checked.append(True)
+        return original_effect(method, request)
+
+    monkeypatch.setattr(team_plan_adapter, "effect_runtime_result", inspect)
+    result = _settle(registry, _proposal())
+    assert result[0]["status"] == "committed", result
+    assert checked == [True]
+    with exclusive_cross_runtime_file_lock(registry, timeout_seconds=0):
+        pass
+
+
 def _settle(
     registry_path: Path,
     proposal: dict,

From 74db27da707d21dcaca7c82097a47bef152f2be4 Mon Sep 17 00:00:00 2001
From: hyk <4408344+hhyykk@users.noreply.github.com>
Date: Mon, 5 Oct 2026 11:51:33 +0800
Subject: [PATCH 2/2] docs(roadmap): qualify local team-plan source fencing

Signed-off-by: hyk <4408344+hhyykk@users.noreply.github.com>
---
 docs/architecture/rfcs/loopx-overall-roadmap-v0.md       | 2 +-
 docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md
index 02899a4dbc..a9b195c1cd 100644
--- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md
+++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md
@@ -909,7 +909,7 @@ This narrows a local recovery gap, not the full R1/R2 coordination acceptance.
 
 **R1 transaction checkpoint.** Team-plan admission and whole-batch planning now live in `work_items/team_plan.ts`. Confirmation assigns all admitted lanes in one write with a durable operation receipt; identity is proposal + lane, never Todo text. File/SQLite authority uses the existing CAS and receipt owner; legacy Markdown writes the records and immutable receipt together under its existing fence and lock. Exact replay reads historical results even after a receiver changes, completes or deletes work. A precommit failure creates no lane prefix, and pending canonical display delivery requires recovery before Chat reports verified success. The card names partial assignments and gaps; quota/stop remain advisory and an explicit enforcement claim is rejected. Agent-originated settlement binds the same state basis at its journal's first write and re-reads it at settlement; a plan whose basis is missing or moved, or whose every lane is a gap, is a typed failed receipt that creates no Todo and replays unchanged.
 
-This closes the local assignment/retry portion of F4, not R1's collaboration acceptance. Registered receivers are assigned without being impersonated as authors; agent-originated settlement cannot assign another peer without owner confirmation. Assignment does not attest receiver adoption, a lease, execution, dependency consumption or independent acceptance. Do not add a second confirmation to ordinary already-authorized work. Gap resolution requires new explicit intent; replay must not silently extend the confirmed subset. The fingerprint binds current local state and canonical revision, not a full shared Goal-intent transaction. R2/R3/R4 still own executor qualification, receiver adoption/result return and shared intent/authorization; the cross-host Turn lease is not a plan barrier.
+This closes the local assignment/retry portion of F4, not R1's collaboration acceptance. Registered receivers are assigned without being impersonated as authors; agent-originated settlement cannot assign another peer without owner confirmation. Assignment does not attest receiver adoption, a lease, execution, dependency consumption or independent acceptance. Do not add a second confirmation to ordinary already-authorized work. Gap resolution requires new explicit intent; replay must not silently extend the confirmed subset. The fingerprint binds current local state and canonical revision, not a full shared Goal-intent transaction. Local registry and active-state guards now remain held through the typed commit: native token claims survive caller exit, changed source bytes reject new assignments, and expired handoffs can only replay an exact committed receipt. Registry contention returns a retryable failure without creating work; canonical display recovery runs after the source guards release. R2/R3/R4 still own executor qualification, receiver adoption/result return and shared intent/authorization; the cross-host Turn lease is not a plan barrier.
 
 Validation: production Chat apply and real file projection recovery; packaged confirmation card; typed batch cases on FileAuthorityStore and an isolated PostgreSQL server, including equal-text identities, concurrent commit, invalid final lane, lost response and receiver changes. These fixtures do not qualify Lark transport or cross-host worker execution.
 
diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md
index dde2c195e4..94338b621c 100644
--- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md
+++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md
@@ -686,7 +686,7 @@ lifecycle。
 
 **R1 事务检查点。** 团队计划准入与整批规划现在归 `work_items/team_plan.ts`。确认后，全部已准入 lane 与持久操作回执一次提交；身份由 proposal + lane 决定，不再由 Todo 文本决定。File/SQLite 权威复用现有 CAS 与回执 owner；legacy Markdown 在原有 fence 和锁内同时写入任务和不可变回执。同一操作重试只读历史结果，接收者后来修改、完成或删除任务也不会触发重建。提交前失败不会留下部分 lane；canonical 展示投递仍 pending 时，Chat 必须恢复后才能报告验证成功。卡片列出部分分配及缺口；quota/stop 是参考，显式强制声明会被拒绝。Agent 发起的结算在 journal 首次写入时绑定同一状态基线并在结算时重读；基线缺失或已变动、或全部 lane 均为缺口的计划，记录为类型化的失败回执，不创建 Todo，重放结果不变。
 
-这完成 F4 的本地分配/重试部分，不等于 R1 协同验收。注册接收者可以被分配任务，但不会被冒充为作者；Agent 发起的结算未经业主确认不能给另一 peer 分配任务。分配不证明接收者采纳、lease、执行、依赖消费或独立验收。普通已授权工作不应普遍增加第二次确认。解决缺口需要明确的新意图；重放不能静默扩展原确认子集。fingerprint 绑定当前本地状态与 canonical revision，不是完整共享 Goal 意图事务。R2/R3/R4 仍负责执行器资格、接收者采纳/结果返回及共享意图/授权；跨主机 Turn lease 不是计划屏障。
+这完成 F4 的本地分配/重试部分，不等于 R1 协同验收。注册接收者可以被分配任务，但不会被冒充为作者；Agent 发起的结算未经业主确认不能给另一 peer 分配任务。分配不证明接收者采纳、lease、执行、依赖消费或独立验收。普通已授权工作不应普遍增加第二次确认。解决缺口需要明确的新意图；重放不能静默扩展原确认子集。fingerprint 绑定当前本地状态与 canonical revision，不是完整共享 Goal 意图事务。本地 registry 与活动状态锁现在持续保护到类型化提交结束：native token claim 在调用进程退出后仍有效；源字节变化拒绝新分配；失效交接只能重放精确匹配的已提交回执。registry 竞争返回可重试失败，不创建任务；canonical 展示恢复在源锁释放后执行。R2/R3/R4 仍负责执行器资格、接收者采纳/结果返回及共享意图/授权；跨主机 Turn lease 不是计划屏障。
 
 验证覆盖真实 Chat apply、文件权威投递恢复、打包确认卡片，以及 FileAuthorityStore 和隔离 PostgreSQL 上的同文不同身份、并发提交、末条 lane 非法、响应丢失和接收者变更。这些 fixture 不验收 Lark 传输或跨主机 worker 执行。