From 0fa8a5615c67803d9c2c193ae570427a1d1aab47 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Mon, 21 Sep 2026 02:58:07 +0800
Subject: [PATCH 1/3] refactor(control-plane): own effective lease inspection
 in TypeScript

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../control_plane/effect_runtime_handlers.ts  |   2 +
 loopx/control_plane/work_items/task_lease.py  | 126 +++-----------
 .../work_items/task_lease_acquire.ts          |   2 +-
 .../work_items/task_lease_acquire_adapter.py  |  62 ++++++-
 .../work_items/task_lease_eligibility.ts      |  23 ++-
 .../work_items/task_lease_inspection.ts       | 113 +++++++++++++
 .../test_canonical_lease_inspection.py        | 124 +++++++++++++-
 .../task_lease_inspection.test.ts             | 158 ++++++++++++++++++
 8 files changed, 504 insertions(+), 106 deletions(-)
 create mode 100644 loopx/control_plane/work_items/task_lease_inspection.ts
 create mode 100644 tests/control_plane_ts/task_lease_inspection.test.ts

diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 8f9d036dd5..5f8ead534a 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -1,3 +1,4 @@
+import {inspectTaskLease} from "./work_items/task_lease_inspection.ts";
 import {evaluateTodoPriority} from "./todos/priority.ts";
 import {evaluateUserCompletion} from "./todos/user_completion.ts";
 import {projectTodoSuccession, projectTodoClosure} from "./todos/succession.ts";
@@ -506,6 +507,7 @@ export function createEffectRuntimeHandlers(
     ["task_lease.owner_eligibility", evaluateTaskLeaseOwnerEligibility],
     ["task_lease.acquire.decide", evaluateTaskLeaseAcquireDecision],
     ["task_lease.acquire.native", executeTaskLeaseAcquire],
+    ["task_lease.inspect.native", inspectTaskLease],
     ["task_lease.lifecycle.decide", evaluateTaskLeaseLifecycleDecision],
     ["task_lease.lifecycle.native", executeTaskLeaseLifecycle],
     ["coordination.runtime_shadow.bootstrap", bootstrapCoordinationRuntimeShadow],
diff --git a/loopx/control_plane/work_items/task_lease.py b/loopx/control_plane/work_items/task_lease.py
index e8a24bd4a2..bc585f8a16 100644
--- a/loopx/control_plane/work_items/task_lease.py
+++ b/loopx/control_plane/work_items/task_lease.py
@@ -17,13 +17,8 @@
 from ..runtime.time import now_utc as runtime_now_utc
 from ..runtime.time import parse_timestamp, utc_isoformat
 from ..coordination.authority_core import (
-    CoordinationSnapshot,
-    DecisionOutcome,
-    LeaseOwnerEligibilityCommand,
-    decide,
     write_scopes_overlap as core_write_scopes_overlap,
 )
-from ..coordination.local_snapshot import todo_snapshot_from_mapping
 from ..todos.contract import (
     normalize_required_write_scopes,
     normalize_todo_claimed_by,
@@ -45,7 +40,7 @@
     lease_acquire_ttl_seconds as lease_acquire_ttl_seconds,
     lease_epoch as lease_epoch,
     lease_version as lease_version,
-    read_lease,
+    read_lease as read_lease,
     require_expected_version as require_expected_version,
     write_lease as write_lease,
 )
@@ -605,36 +600,26 @@ def task_lease_owner_constraint(
     owner: Any,
     registered_agents: list[str] | None = None,
 ) -> dict[str, Any]:
+    from ..effect_runtime import effect_runtime_result
+
     normalized_owner = normalize_todo_claimed_by(owner)
-    effective_registered_agents = (
-        tuple(registered_agents)
-        if registered_agents is not None
-        else ((normalized_owner,) if normalized_owner else ())
-    )
-    plan = decide(
-        CoordinationSnapshot(
-            registered_agents=effective_registered_agents,
-            todo=todo_snapshot_from_mapping(todo),
-        ),
-        LeaseOwnerEligibilityCommand(owner=normalized_owner),
-    )
-    if plan.outcome is DecisionOutcome.APPLY:
-        return {"effective": True}
-    result: dict[str, Any] = {"effective": False, "reason": plan.code}
-    if plan.code == "todo_not_open":
-        todo_status = str((todo or {}).get("status") or "").strip().lower()
-        result["todo_status"] = todo_status or "unknown"
-    elif plan.code == "owner_not_registered":
-        result["registered_agents"] = list(registered_agents or [])
-    elif plan.code == "owner_excluded_from_todo":
-        result["excluded_agents"] = normalize_todo_excluded_agents(
-            (todo or {}).get("excluded_agents")
-        )
-    elif plan.code == "owner_conflicts_with_claim":
-        result["claimed_by"] = normalize_todo_claimed_by(
-            (todo or {}).get("claimed_by")
-        )
-    return result
+    registered = registered_agents if registered_agents is not None else ([normalized_owner] if normalized_owner else [])
+    result = effect_runtime_result("task_lease.owner_eligibility", {
+        "todo": None if todo is None else {
+            "status": str(todo.get("status") or "").strip().lower(),
+            "claimed_by": normalize_todo_claimed_by(todo.get("claimed_by")),
+            "excluded_agents": normalize_todo_excluded_agents(todo.get("excluded_agents")),
+        },
+        "owner": normalized_owner, "registered_agents": registered,
+    })
+    if not isinstance(result, dict) or result.get("schema_version") != "task_lease_owner_eligibility_v0":
+        raise RuntimeError("typed lease owner constraint is missing; update the runtime")
+    constraint = result.get("constraint")
+    if not isinstance(constraint, dict) or not isinstance(constraint.get("effective"), bool):
+        raise RuntimeError("typed lease owner constraint shape mismatch")
+    if constraint["effective"] is False and not isinstance(constraint.get("reason"), str):
+        raise RuntimeError("typed lease owner rejection omitted its reason")
+    return constraint
 
 
 def require_registered_task_lease_owner(
@@ -918,70 +903,9 @@ def inspect_task_lease(
     goal_id: str,
     todo_id: str,
 ) -> dict[str, Any]:
-    goal_id = normalize_goal_id(goal_id)
-    todo_id = normalize_lease_todo_id(todo_id)
-    from ..coordination.local_authority import read_canonical_todos_if_promoted
-    from ..todos.handoff_mode import normalize_handoff_mode
-
-    # A promoted read cannot combine canonical Todo facts with stale local
-    # lease files or display frontmatter. Absence is an authoritative result.
-    canonical = read_canonical_todos_if_promoted(
-        runtime_root=runtime_root, goal_id=goal_id, include_leases=True,
+    from .task_lease_acquire_adapter import inspect_native_task_lease
+
+    return inspect_native_task_lease(
+        registry_path=registry_path, runtime_root=runtime_root,
+        goal_id=normalize_goal_id(goal_id), todo_id=normalize_lease_todo_id(todo_id),
     )
-    source_fields: dict[str, Any] = {}
-    if canonical is not None:
-        if "handoff_mode" not in canonical:
-            raise TaskLeaseError("canonical lease snapshot omitted handoff mode; update the runtime",
-                                 code="local_authority_snapshot_incomplete")
-        lease_path = None
-        lease = next((row for row in canonical["leases"] if row.get("todo_id") == todo_id), None)
-        todo = next((row for row in canonical["todos"] if row.get("todo_id") == todo_id), None)
-        handoff_mode = normalize_handoff_mode(canonical.get("handoff_mode"))
-        source_fields = {
-            "source_authority": canonical["source_authority"],
-            "provider_revision": canonical["provider_revision"],
-            "legacy_fallback_used": False,
-        }
-    else:
-        lease_path = task_lease_path(runtime_root=runtime_root, goal_id=goal_id, todo_id=todo_id)
-        lease = read_lease(lease_path)
-        handoff_mode = _optional_handoff_mode(registry_path, goal_id)
-    active = lease_is_active(lease)
-    executor_constraint: dict[str, Any] | None = None
-    if active and lease:
-        try:
-            if canonical is None:
-                todo = task_lease_todo_projection(
-                    registry_path=registry_path,
-                    goal_id=goal_id,
-                    todo_id=todo_id,
-                )
-        except TaskLeaseError as exc:
-            active = False
-            executor_constraint = {
-                "effective": False,
-                "reason": exc.code,
-            }
-        else:
-            executor_constraint = task_lease_owner_constraint(
-                todo,
-                owner=lease.get("owner"),
-                registered_agents=registered_agent_ids_from_registry(registry_path, goal_id),
-            )
-            if executor_constraint.get("effective") is not True:
-                active = False
-            else:
-                executor_constraint = None
-    return {
-        "ok": True,
-        "schema_version": TASK_LEASE_SCHEMA_VERSION,
-        "action": "inspect",
-        "goal_id": goal_id,
-        "todo_id": todo_id,
-        "active": active,
-        "lease": lease,
-        "lease_path": str(lease_path) if lease_path is not None else None,
-        **source_fields,
-        **({"handoff_mode": handoff_mode} if handoff_mode else {}),
-        **({"executor_constraint": executor_constraint} if executor_constraint else {}),
-    }
diff --git a/loopx/control_plane/work_items/task_lease_acquire.ts b/loopx/control_plane/work_items/task_lease_acquire.ts
index 6205d5b923..8a2b998040 100644
--- a/loopx/control_plane/work_items/task_lease_acquire.ts
+++ b/loopx/control_plane/work_items/task_lease_acquire.ts
@@ -384,7 +384,7 @@ export function decodeTaskLeaseAuthority(value: unknown): AuthorityFacts {
   };
 }
 
-function normalizeHandoffMode(value: unknown): string {
+export function normalizeHandoffMode(value: unknown): string {
   const mode = compact(value) || "legacy";
   if (!new Set(["legacy", "soft_claim", "hard_lease"]).has(mode)) {
     throw new TaskLeaseAcquireError(
diff --git a/loopx/control_plane/work_items/task_lease_acquire_adapter.py b/loopx/control_plane/work_items/task_lease_acquire_adapter.py
index 5af98e8307..53f191f046 100644
--- a/loopx/control_plane/work_items/task_lease_acquire_adapter.py
+++ b/loopx/control_plane/work_items/task_lease_acquire_adapter.py
@@ -203,6 +203,7 @@ def _task_lease_authority_projection(
     todo_id: str,
     goal: dict[str, Any] | None,
     state_file: Path | None,
+    project_todos: bool = True,
 ) -> tuple[Any, list[Any], list[dict[str, Any]], dict[str, Any] | None]:
     from ...todos import list_goal_todos
     from ..goals.active_state_metadata import parse_state_frontmatter
@@ -215,6 +216,8 @@ def _task_lease_authority_projection(
         handoff_mode = parse_state_frontmatter(
             state_file.read_text(encoding="utf-8")
         ).get("handoff_mode")
+    if not project_todos:
+        return handoff_mode, _raw_registered_agent_candidates(goal), [], None
     try:
         projection = list_goal_todos(
             registry_path=registry_path,
@@ -258,6 +261,7 @@ def _task_lease_authority_snapshot_attempt(
     registry_path: Path,
     goal_id: str,
     todo_id: str,
+    project_todos: bool = True,
 ) -> dict[str, Any] | None:
     registry_receipt_before = _authority_source_receipt("registry", registry_path)
     registry = load_registry(registry_path)
@@ -279,6 +283,7 @@ def _task_lease_authority_snapshot_attempt(
             todo_id=todo_id,
             goal=goal,
             state_file=state_file,
+            project_todos=project_todos,
         )
     )
 
@@ -309,14 +314,16 @@ def task_lease_acquire_authority_facts(
     registry_path: Path,
     goal_id: str,
     todo_id: str,
+    project_todos: bool = True,
 ) -> dict[str, Any]:
-    """Project a source-stable, decision-free acquire input snapshot."""
+    """Project source-stable facts; inspection can defer Todo parsing to TS demand."""
 
     for _attempt in range(TASK_LEASE_AUTHORITY_SNAPSHOT_ATTEMPTS):
         facts = _task_lease_authority_snapshot_attempt(
             registry_path=registry_path,
             goal_id=goal_id,
             todo_id=todo_id,
+            project_todos=project_todos,
         )
         if facts is not None:
             return facts
@@ -827,3 +834,56 @@ def execute_native_task_lease_lifecycle(
         # fence read it from the nested native payload before redacting it.
         return result
     raise RuntimeError("native task-lease lifecycle exhausted source-CAS retries")
+
+
+def inspect_native_task_lease(
+    *, registry_path: Path, runtime_root: Path, goal_id: str, todo_id: str,
+) -> dict[str, Any]:
+    """Project source facts; the typed reader owns lease interpretation."""
+    from ..coordination.local_authority import (
+        LOCAL_AUTHORITY_SOURCES, LocalCoordinationAuthorityUnavailable,
+        local_authority_is_promoted,
+    )
+    from ..effect_runtime import effect_runtime_result
+
+    for attempt in range(TASK_LEASE_AUTHORITY_SNAPSHOT_ATTEMPTS):
+        canonical = local_authority_is_promoted(runtime_root=runtime_root, goal_id=goal_id)
+        # TS alone decides whether the retained lease needs current Todo facts.
+        # Inactive legacy inspection must not parse the full work document.
+        authority = _canonical_lease_authority_facts(registry_path, goal_id) if canonical else task_lease_acquire_authority_facts(
+            registry_path=registry_path, goal_id=goal_id, todo_id=todo_id, project_todos=False,
+        )
+        request = {
+            "schema_version": "loopx_task_lease_inspect_request_v0",
+            "source": "canonical" if canonical else "legacy",
+            "phase": "effective_lease" if canonical else "lease_record",
+            "runtime_root": str(runtime_root.resolve()), "goal_id": goal_id,
+            "todo_id": todo_id, "authority": authority,
+        }
+        result = effect_runtime_result("task_lease.inspect.native", request)
+        if isinstance(result, dict) and result.get("todo_projection_required") is True:
+            if canonical or result.get("ok") is not True or result.get("action") != "inspect":
+                raise RuntimeError("native lease inspection requested an invalid source projection")
+            request["phase"] = "effective_lease"
+            request["authority"] = task_lease_acquire_authority_facts(
+                registry_path=registry_path, goal_id=goal_id, todo_id=todo_id,
+            )
+            # Re-read the lease and fence: neither the record nor its expiry is
+            # assumed unchanged while Python prepares the Todo projection.
+            result = effect_runtime_result("task_lease.inspect.native", request)
+        if not isinstance(result, dict) or result.get("schema_version") != TASK_LEASE_SCHEMA_VERSION or result.get("action") != "inspect" or not isinstance(result.get("ok"), bool):
+            raise RuntimeError("native lease inspection result shape mismatch")
+        if result.get("error_code") == "authority_source_changed" and attempt + 1 < TASK_LEASE_AUTHORITY_SNAPSHOT_ATTEMPTS:
+            continue
+        if result.get("ok") is not True:
+            code = str(result.get("error_code") or "task_lease_inspection_unavailable")
+            exception = LocalCoordinationAuthorityUnavailable if canonical and code != "corrupt_lease" else TaskLeaseError
+            raise exception(str(result.get("error") or "lease inspection unavailable"), code=code, payload=result)
+        if not isinstance(result.get("active"), bool) or (canonical and (
+            result.get("source_authority") not in LOCAL_AUTHORITY_SOURCES
+            or not isinstance(result.get("provider_revision"), str)
+            or result.get("legacy_fallback_used") is not False
+        )):
+            raise RuntimeError("native lease inspection omitted source evidence")
+        return result
+    raise RuntimeError("native lease inspection exhausted source retries")
diff --git a/loopx/control_plane/work_items/task_lease_eligibility.ts b/loopx/control_plane/work_items/task_lease_eligibility.ts
index 93abde6b5f..16ac440cf5 100644
--- a/loopx/control_plane/work_items/task_lease_eligibility.ts
+++ b/loopx/control_plane/work_items/task_lease_eligibility.ts
@@ -23,6 +23,24 @@ export function leaseOwnerRejection(todo: LeaseEligibilityTodo | null | undefine
   return null;
 }
 
+/** Diagnostics share the exact rejection precedence used by mutation admission. */
+export function leaseOwnerConstraint(todo: LeaseEligibilityTodo | null | undefined,
+  owner: string | null, registered: readonly string[]) {
+  const reason = leaseOwnerRejection(todo, owner, registered);
+  switch (reason) {
+    case null: return {effective: true as const};
+    case "todo_not_open":
+      return {effective: false as const, reason, todo_status: todo?.status || "unknown"};
+    case "owner_not_registered":
+      return {effective: false as const, reason, registered_agents: [...registered]};
+    case "owner_excluded_from_todo":
+      return {effective: false as const, reason, excluded_agents: [...(todo?.excluded_agents ?? [])]};
+    case "owner_conflicts_with_claim":
+      return {effective: false as const, reason, claimed_by: todo?.claimed_by ?? null};
+    default: return {effective: false as const, reason};
+  }
+}
+
 function strings(value: unknown, label: string): string[] {
   if (!Array.isArray(value) || value.some(item => typeof item !== "string")) {
     throw new EffectRuntimeRequestError(`${label} must be an array of strings`);
@@ -47,8 +65,9 @@ export function evaluateTaskLeaseOwnerEligibility(value: unknown) {
     todo = {status: raw.status, claimed_by: nullableString(raw.claimed_by, "todo.claimed_by"),
       excluded_agents: strings(raw.excluded_agents, "todo.excluded_agents")};
   }
-  const code = leaseOwnerRejection(todo, nullableString(input.owner, "owner"),
+  const constraint = leaseOwnerConstraint(todo, nullableString(input.owner, "owner"),
     strings(input.registered_agents, "registered_agents"));
   return {schema_version: "task_lease_owner_eligibility_v0",
-    outcome: code === null ? "apply" : "rejected", code: code ?? "lease_owner_allowed"};
+    outcome: constraint.effective ? "apply" : "rejected",
+    code: constraint.effective ? "lease_owner_allowed" : constraint.reason, constraint};
 }
diff --git a/loopx/control_plane/work_items/task_lease_inspection.ts b/loopx/control_plane/work_items/task_lease_inspection.ts
new file mode 100644
index 0000000000..ea03ac84f9
--- /dev/null
+++ b/loopx/control_plane/work_items/task_lease_inspection.ts
@@ -0,0 +1,113 @@
+/** Read-only lease inspection: one source, one clock, shared execution eligibility. */
+import {isAbsolute, join} from "node:path";
+import type {JsonObject} from "../effect_program.ts";
+import {requireJsonObject} from "../runtime_decode.ts";
+import {canonicalAuthoritySha256, AuthorityStoreProtocolError} from "../coordination/authority_store_codec.ts";
+import {indexCoordinationProjection, validateCoordinationTodoReadModel} from "../coordination/coordination_projection.ts";
+import {loadLegacyCoordinationWriterFence} from "../coordination/legacy_writer_fence.ts";
+import {canonicalLeaseTodoFact, canonicalTaskLease} from "../coordination/task_lease_state.ts";
+import {openLocalAuthorityStoreHandle, localAuthorityOpenFailure,
+  type LocalAuthorityProviderDependencies} from "../coordination/local_authority_provider.ts";
+import {decodeTaskLeaseAuthority, leaseIsActive, normalizeGoalId, normalizeTodoId, normalizeHandoffMode,
+  readLease, revalidateAuthoritySources, TaskLeaseAcquireError, TASK_LEASE_SCHEMA_VERSION,
+  type LeaseRecord, type TodoFact} from "./task_lease_acquire.ts";
+import {leaseOwnerConstraint} from "./task_lease_eligibility.ts";
+
+export const TASK_LEASE_INSPECT_REQUEST = "loopx_task_lease_inspect_request_v0";
+
+export interface TaskLeaseInspectionDependencies {
+  now?: () => Date;
+  authorityProvider?: LocalAuthorityProviderDependencies;
+}
+
+/** Observation never grants a lease or substitutes for mutation-time proof. */
+export async function inspectTaskLease(value: unknown,
+  dependencies: TaskLeaseInspectionDependencies = {}): Promise<JsonObject> {
+  let evidence: JsonObject = {};
+  let goalId: string | null = null, todoId: string | null = null;
+  try {
+    const input = requireJsonObject(value, "task lease inspection");
+    if (input.schema_version !== TASK_LEASE_INSPECT_REQUEST ||
+        (input.source !== "canonical" && input.source !== "legacy") ||
+        (input.phase !== undefined && input.phase !== "lease_record" && input.phase !== "effective_lease") ||
+        (input.source === "canonical" && input.phase === "lease_record") ||
+        Object.keys(input).some(key => !["schema_version", "source", "goal_id", "todo_id", "runtime_root", "authority", "phase"].includes(key))) {
+      throw new TaskLeaseAcquireError("invalid task lease inspection request", "invalid_inspection_request");
+    }
+    goalId = normalizeGoalId(input.goal_id); todoId = normalizeTodoId(input.todo_id);
+    if (typeof input.runtime_root !== "string" || !isAbsolute(input.runtime_root) ||
+        input.runtime_root.trim() !== input.runtime_root) {
+      throw new TaskLeaseAcquireError("runtime_root must be an absolute path", "invalid_runtime_root");
+    }
+    const root = input.runtime_root, canonical = input.source === "canonical";
+    if (canonical) evidence = {source_authority: null, legacy_fallback_used: false};
+    const beforeFence = await loadLegacyCoordinationWriterFence(root, goalId);
+    if (beforeFence.status === "failed") throw new TaskLeaseAcquireError(beforeFence.reason, beforeFence.reason_code);
+    if ((beforeFence.status === "loaded") !== canonical) {
+      throw new TaskLeaseAcquireError("lease authority route changed; retry inspection", "authority_source_changed");
+    }
+    const rawAuthority = requireJsonObject(input.authority, "authority");
+    // Canonical mode/Todos come only from the provider. Invalid legacy mode is
+    // omitted from this diagnostic response, as in the historical inspect API.
+    const authority = decodeTaskLeaseAuthority({...rawAuthority, handoff_mode: "legacy",
+      ...(canonical ? {todos: [], todo_projection_error: null} : {})});
+    await revalidateAuthoritySources(authority.source_receipts);
+    let lease: LeaseRecord | null, todo: TodoFact | null, mode: string | null, leasePath: string | null;
+    if (canonical) {
+      const {store, sourceAuthority} = await openLocalAuthorityStoreHandle(root, goalId, dependencies.authorityProvider);
+      evidence.source_authority = sourceAuthority;
+      const loaded = await store.loadAuthority();
+      if (loaded.status !== "loaded") {
+        throw new TaskLeaseAcquireError("canonical lease snapshot is unavailable", "local_authority_snapshot_unavailable", {...loaded});
+      }
+      const index = indexCoordinationProjection(loaded.head, goalId);
+      validateCoordinationTodoReadModel(loaded.head, goalId);
+      const rawLease = index.leases.get(todoId);
+      lease = rawLease ? canonicalTaskLease(rawLease, goalId, todoId) : null;
+      todo = canonicalLeaseTodoFact(index.todos.get(todoId));
+      mode = normalizeHandoffMode(loaded.head.handoff_mode);
+      leasePath = null;
+      evidence.provider_revision = loaded.provider_revision;
+    } else {
+      leasePath = join(root, "goals", goalId, "task-leases", `${todoId}.json`);
+      lease = await readLease(leasePath);
+      todo = authority.todos.get(todoId) ?? null;
+      try { mode = normalizeHandoffMode(rawAuthority.handoff_mode); }
+      catch (error) {
+        if (!(error instanceof TaskLeaseAcquireError) || error.code !== "invalid_handoff_mode") throw error;
+        mode = null;
+      }
+    }
+    const at = dependencies.now?.() ?? new Date();
+    if (!Number.isFinite(at.valueOf())) throw new TaskLeaseAcquireError("invalid inspection clock", "invalid_inspection_clock");
+    const timeActive = leaseIsActive(lease, at);
+    const needsProjection = !canonical && input.phase === "lease_record" && timeActive;
+    const constraint = !timeActive || lease === null || needsProjection ? null
+      : authority.todo_projection_error !== null
+        ? {effective: false, reason: authority.todo_projection_error.code}
+        : leaseOwnerConstraint(todo, typeof lease.owner === "string" ? lease.owner : null, authority.registered_agents);
+    // Both registration and route must still describe the source we inspected.
+    // No lock is held and no promise is made about later commits or expiry.
+    await revalidateAuthoritySources(authority.source_receipts);
+    const afterFence = await loadLegacyCoordinationWriterFence(root, goalId);
+    if (canonicalAuthoritySha256(afterFence) !== canonicalAuthoritySha256(beforeFence)) {
+      throw new TaskLeaseAcquireError("lease authority route changed; retry inspection", "authority_source_changed");
+    }
+    if (needsProjection) return {ok: true, schema_version: TASK_LEASE_SCHEMA_VERSION,
+      action: "inspect", todo_projection_required: true};
+    return {ok: true, schema_version: TASK_LEASE_SCHEMA_VERSION, action: "inspect",
+      goal_id: goalId, todo_id: todoId, active: timeActive && constraint?.effective === true,
+      lease, lease_path: leasePath, ...evidence,
+      ...(mode === null ? {} : {handoff_mode: mode}),
+      ...(constraint?.effective === false ? {executor_constraint: constraint} : {})};
+  } catch (error) {
+    const opening = localAuthorityOpenFailure(error);
+    return {ok: false, schema_version: TASK_LEASE_SCHEMA_VERSION, action: "inspect",
+      goal_id: goalId, todo_id: todoId, ...evidence,
+      ...(error instanceof TaskLeaseAcquireError ? error.payload : {}), ...opening,
+      error_code: typeof opening.reason_code === "string" ? opening.reason_code
+        : error instanceof TaskLeaseAcquireError ? error.code
+          : error instanceof AuthorityStoreProtocolError ? "local_authority_snapshot_invalid" : "task_lease_inspection_unavailable",
+      error: error instanceof Error ? error.message : "task lease inspection unavailable"};
+  }
+}
diff --git a/tests/control_plane/test_canonical_lease_inspection.py b/tests/control_plane/test_canonical_lease_inspection.py
index 9cb3ab4bdd..255665b552 100644
--- a/tests/control_plane/test_canonical_lease_inspection.py
+++ b/tests/control_plane/test_canonical_lease_inspection.py
@@ -17,7 +17,7 @@
 TODO = "todo_current"
 
 
-def _fixture(root: Path, provider: str, *, retained: bool = True, excluded: bool = False):
+def _fixture(root: Path, provider: str, *, retained: bool = True, excluded: bool = False, todo_patch=None, lease_patch=None):
     runtime = root / "runtime"
     state = root / "ACTIVE_GOAL_STATE.md"
     state.write_text("---\nhandoff_mode: soft_claim\n---\n# Obsolete display\n")
@@ -32,8 +32,12 @@ def _fixture(root: Path, provider: str, *, retained: bool = True, excluded: bool
     lease = {"schema_version": "task_lease_v0", "goal_id": GOAL, "todo_id": TODO,
              "status": "active", "owner": "agent-a", "idempotency_key": "lease-first",
              "expires_at": "2099-01-01T00:00:00Z", "lease_epoch": 3, "version": 2}
+    todo.update(todo_patch or {})
+    lease.update(lease_patch or {})
     projection = build_todo_runtime_shadow_projection(goal_id=GOAL, todos=[todo],
         leases=[lease] if retained else [], handoff_mode="hard_lease")
+    # Preserve retained provider history even when a legacy capture would omit it.
+    projection["leases"] = [lease] if retained else []
     initialize_canonical_authority(runtime, GOAL, projection, state_path=state, provider=provider)
     obsolete = runtime / "goals" / GOAL / "task-leases" / f"{TODO}.json"
     obsolete.parent.mkdir(parents=True, exist_ok=True)
@@ -98,3 +102,121 @@ def test_unavailable_provider_fails_instead_of_reading_legacy_lease(tmp_path):
         _inspect(registry, runtime)
     assert getattr(error.value, "code", "").startswith("local_authority_")
     assert json.loads(obsolete.read_text())["owner"] == "agent-b"
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_archived_open_record_cannot_make_retained_execution_effective(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, runtime, _, _ = _fixture(tmp_path, provider,
+        todo_patch={"archive_state": "archive", "source_section": "Completed Work Archive"})
+    result = _inspect(registry, runtime)
+    assert result["active"] is False
+    assert result["executor_constraint"] == {"effective": False, "reason": "todo_not_found"}
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_corrupt_active_expiry_is_not_reported_as_an_inactive_lease(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, runtime, _, _ = _fixture(tmp_path, provider, lease_patch={"expires_at": "not-a-date"})
+    from loopx.control_plane.work_items.local_lease_record import TaskLeaseError
+    with pytest.raises((TaskLeaseError, LocalCoordinationAuthorityUnavailable)):
+        _inspect(registry, runtime)
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_registration_change_retries_and_uses_new_eligibility(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, runtime, _, _ = _fixture(tmp_path, provider)
+    from loopx.control_plane import effect_runtime
+    original = effect_runtime.effect_runtime_result
+    calls = []
+
+    def change_registration(method, payload):
+        if method == "task_lease.inspect.native":
+            calls.append(method)
+            if len(calls) == 1:
+                updated = json.loads(registry.read_text())
+                updated["goals"][0]["coordination"]["registered_agents"] = ["agent-b"]
+                registry.write_text(json.dumps(updated))
+        return original(method, payload)
+
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", change_registration)
+    result = _inspect(registry, runtime)
+    assert len(calls) == 2
+    assert result["active"] is False
+    assert result["executor_constraint"] == {
+        "effective": False, "reason": "owner_not_registered", "registered_agents": ["agent-b"],
+    }
+
+
+def test_continuous_source_churn_exhausts_bounded_retry_without_success(tmp_path, monkeypatch):
+    registry, runtime, _, _ = _fixture(tmp_path, "file")
+    from loopx.control_plane import effect_runtime
+    original = effect_runtime.effect_runtime_result
+    calls = []
+
+    def change_source(method, payload):
+        if method == "task_lease.inspect.native":
+            calls.append(method)
+            registry.write_text(registry.read_text() + "\n")
+        return original(method, payload)
+
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", change_source)
+    with pytest.raises(LocalCoordinationAuthorityUnavailable) as error:
+        _inspect(registry, runtime)
+    assert error.value.code == "authority_source_changed"
+    assert len(calls) == 3
+
+
+def test_owner_constraint_requires_the_native_diagnostic_contract(monkeypatch):
+    from loopx.control_plane import effect_runtime
+    from loopx.control_plane.work_items.task_lease import task_lease_owner_constraint
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", lambda *_: {
+        "schema_version": "task_lease_owner_eligibility_v0", "outcome": "apply", "code": "lease_owner_allowed",
+    })
+    with pytest.raises(RuntimeError, match="constraint shape mismatch"):
+        task_lease_owner_constraint({"status": "open"}, owner="agent-a")
+
+
+def test_inactive_legacy_inspection_does_not_project_todos(tmp_path, monkeypatch):
+    registry = tmp_path / "registry.json"
+    state = tmp_path / "state.md"
+    state.write_text("---\nhandoff_mode: hard_lease\n---\n# Synthetic state\n")
+    registry.write_text(json.dumps({"goals": [{"id": GOAL, "repo": str(tmp_path), "state_file": str(state)}]}))
+    import loopx.todos
+    def forbidden_projection(**_):
+        raise AssertionError("inactive inspection must not parse Todo history")
+    monkeypatch.setattr(loopx.todos, "list_goal_todos", forbidden_projection)
+    runtime = tmp_path / "runtime"
+    missing = _inspect(registry, runtime)
+    assert missing["active"] is False and missing["lease"] is None
+    lease_path = runtime / "goals" / GOAL / "task-leases" / f"{TODO}.json"
+    lease_path.parent.mkdir(parents=True)
+    lease_path.write_text(json.dumps({"schema_version": "task_lease_v0", "status": "active", "expires_at": "2000-01-01T00:00:00Z"}))
+    expired = _inspect(registry, runtime)
+    assert expired["active"] is False and expired["handoff_mode"] == "hard_lease"
+
+
+def test_legacy_inspection_reloads_lease_after_demanding_todo_facts(tmp_path, monkeypatch):
+    registry = tmp_path / "registry.json"
+    state = tmp_path / "state.md"
+    state.write_text("---\nhandoff_mode: hard_lease\n---\n# Synthetic state\n")
+    registry.write_text(json.dumps({"goals": [{"id": GOAL, "repo": str(tmp_path), "state_file": str(state),
+        "coordination": {"registered_agents": ["agent-a"]}}]}))
+    runtime = tmp_path / "runtime"
+    lease_path = runtime / "goals" / GOAL / "task-leases" / f"{TODO}.json"
+    lease_path.parent.mkdir(parents=True)
+    lease = {"schema_version": "task_lease_v0", "status": "active", "owner": "agent-a", "expires_at": "2099-01-01T00:00:00Z"}
+    lease_path.write_text(json.dumps(lease))
+    import loopx.todos
+    calls = []
+    def release_during_projection(**_):
+        calls.append(True)
+        lease_path.write_text(json.dumps({**lease, "status": "released"}))
+        return {"todos": [{"todo_id": TODO, "status": "open", "claimed_by": None, "excluded_agents": []}]}
+    monkeypatch.setattr(loopx.todos, "list_goal_todos", release_during_projection)
+    result = _inspect(registry, runtime)
+    assert len(calls) == 1
+    assert result["active"] is False
+    assert result["lease"]["status"] == "released"
+    assert "todo_projection_required" not in result
diff --git a/tests/control_plane_ts/task_lease_inspection.test.ts b/tests/control_plane_ts/task_lease_inspection.test.ts
new file mode 100644
index 0000000000..7d4f74dee3
--- /dev/null
+++ b/tests/control_plane_ts/task_lease_inspection.test.ts
@@ -0,0 +1,158 @@
+/** Inspection is observational; all three real providers share admission facts. */
+import assert from "node:assert/strict";
+import {createHash, randomUUID} from "node:crypto";
+import {writeFileSync, unlinkSync} from "node:fs";
+import {mkdtemp, mkdir, writeFile, rm} from "node:fs/promises";
+import {tmpdir} from "node:os";
+import {join} from "node:path";
+import test from "node:test";
+import {Pool} from "pg";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
+import type {AuthorityStore} from "../../loopx/control_plane/coordination/authority_store.ts";
+import {FileAuthorityStore} from "../../loopx/control_plane/coordination/file_authority_store.ts";
+import {SqliteAuthorityStore} from "../../loopx/control_plane/coordination/sqlite_authority_store.ts";
+import {PostgreSqlAuthorityStore, installPostgreSqlAuthorityStoreSchema} from "../../loopx/control_plane/coordination/postgresql_authority_store.ts";
+import {selectLocalSqliteAuthority, type LocalAuthorityProviderDependencies} from "../../loopx/control_plane/coordination/local_authority_provider.ts";
+import {engageLegacyCoordinationWriterFence, legacyCoordinationWriterFencePath} from "../../loopx/control_plane/coordination/legacy_writer_fence.ts";
+import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/authority_store_codec.ts";
+import {inspectTaskLease, TASK_LEASE_INSPECT_REQUEST} from "../../loopx/control_plane/work_items/task_lease_inspection.ts";
+import {authorityProjectionFixture} from "./authority_projection_fixture.ts";
+import {productionScaleLeaseLifecycleFixture} from "./production_scale_coordination_fixture.ts";
+
+const NOW = new Date("2026-09-13T10:05:00Z");
+const pool = process.env.LOOPX_TEST_POSTGRES_URL ? new Pool({connectionString: process.env.LOOPX_TEST_POSTGRES_URL}) : null;
+const database = pool ? {connect: async () => {
+  const client = await pool.connect();
+  return {query: async (sql: string, values?: readonly unknown[]) => client.query(sql, values ? [...values] : undefined),
+    release: (error?: Error) => client.release(error)};
+}} : null;
+const installed = database ? installPostgreSqlAuthorityStoreSchema(database, `postgresql:${"b".repeat(32)}`) : null;
+test.after(async () => {await pool?.end();});
+
+type Provider = "legacy" | "file" | "sqlite" | "postgresql";
+async function fixture(t: test.TestContext, provider: Provider, schema: "native" | "legacy" = "native") {
+  const root = await mkdtemp(join(tmpdir(), "lease-inspect-")), goal = "inspect-goal";
+  t.after(() => rm(root, {recursive: true, force: true}));
+  const registry = join(root, "registry.json"), registryText = "synthetic registration source";
+  await writeFile(registry, registryText);
+  const fixture = productionScaleLeaseLifecycleFixture(goal, schema);
+  const originalTodos = fixture.projection.todos as JsonObject[], originalLeases = fixture.projection.leases as JsonObject[];
+  const todo = originalTodos.find(row => row.todo_id === fixture.target)!;
+  const lease = originalLeases.find(row => row.todo_id === fixture.target)!;
+  // Independent expectations: the tested implementation never generates these.
+  const cases: {name: string; todo?: JsonObject | null; lease?: JsonObject | null; active: boolean; constraint?: JsonObject}[] = [
+    {name: "allowed", active: true},
+    {name: "absent", lease: null, active: false},
+    {name: "expired", lease: {expires_at: NOW.toISOString()}, active: false},
+    {name: "released", lease: {status: "released"}, active: false},
+    {name: "archived", todo: {archive_state: "archive"}, active: false, constraint: {effective: false, reason: "todo_not_found"}},
+    {name: "orphan", todo: null, active: false, constraint: {effective: false, reason: "todo_not_found"}},
+    {name: "closed", todo: {status: "done", done: true}, active: false, constraint: {effective: false, reason: "todo_not_open", todo_status: "done"}},
+    {name: "excluded", todo: {excluded_agents: ["agent-a"], claimed_by: "agent-b"}, active: false,
+      constraint: {effective: false, reason: "owner_excluded_from_todo", excluded_agents: ["agent-a"]}},
+    {name: "claim", todo: {claimed_by: "agent-b"}, active: false,
+      constraint: {effective: false, reason: "owner_conflicts_with_claim", claimed_by: "agent-b"}},
+    {name: "unregistered", lease: {owner: "agent-retired"}, active: false,
+      constraint: {effective: false, reason: "owner_not_registered", registered_agents: fixture.registered_agents}},
+  ];
+  // Canonical heads prohibit orphan leases; legacy files may retain them.
+  if (provider !== "legacy") cases.splice(cases.findIndex(item => item.name === "orphan"), 1);
+  const todos = [...originalTodos], leases = [...originalLeases];
+  for (const item of cases) {
+    if (item.todo !== null) todos.push({...todo, todo_id: `todo_inspect_${item.name}`, ...item.todo});
+    if (item.lease !== null) leases.push({...lease, todo_id: `todo_inspect_${item.name}`, expires_at: "2026-09-13T10:10:00Z", ...item.lease});
+  }
+  const projection = authorityProjectionFixture(goal, todos, leases, schema, {handoff_mode: "hard_lease"});
+  const authority = {handoff_mode: "hard_lease", registered_agent_candidates: [fixture.registered_agents],
+    todos: todos.filter(row => row.archive_state === "active"), todo_projection_error: null,
+    source_receipts: [{source_id: "registry", path: registry, state: "file", sha256: createHash("sha256").update(registryText).digest("hex")}]};
+  let store: AuthorityStore | null = null, authorityProvider: LocalAuthorityProviderDependencies = {};
+  if (provider === "legacy") {
+    const directory = join(root, "goals", goal, "task-leases"); await mkdir(directory, {recursive: true});
+    for (const row of leases) await writeFile(join(directory, `${row.todo_id}.json`), JSON.stringify(row));
+  } else {
+    if (provider === "sqlite") {
+      assert.equal((await selectLocalSqliteAuthority(root, goal, true)).ok, true);
+      store = new SqliteAuthorityStore(join(root, "authority/sqlite-v0"), goal);
+    } else if (provider === "postgresql") {
+      await installed;
+      const tenant = `inspection-${randomUUID()}`;
+      store = new PostgreSqlAuthorityStore(database!, {tenant_id: tenant, goal_id: goal});
+      t.after(async () => {
+        for (const table of ["authority_receipts", "authority_events", "authority_commits", "authority_heads"]) {
+          await pool!.query(`DELETE FROM loopx_control_plane.${table} WHERE tenant_id=$1`, [tenant]);
+        }
+      });
+      const identity = await store.storeIdentity(); assert.equal(identity.status, "available");
+      if (identity.status !== "available") throw new Error("fixture identity missing");
+      await mkdir(join(root, "authority"), {recursive: true});
+      await writeFile(join(root, "authority", `provider-${createHash("sha256").update(goal).digest("hex")}.json`), JSON.stringify({
+        schema_version: "loopx_local_authority_provider_v0", provider, goal_id: goal, tenant_id: tenant, store_identity: identity.store_identity}));
+      const selected = store; authorityProvider = {openPostgresqlStore: () => selected};
+    } else store = new FileAuthorityStore(join(root, "authority/file-v0"), goal);
+    assert.equal((await store.commitAuthority({expected_provider_revision: null, operation_id: "seed",
+      next_projection: projection, events: [], receipts: []})).status, "applied");
+    const head = await store.loadAuthority(); if (head.status !== "loaded") throw new Error("fixture missing");
+    await writeFile(join(root, "state.md"), "# Synthetic display\n");
+    assert.equal((await engageLegacyCoordinationWriterFence({schema_version: "loopx_legacy_coordination_writer_fence_engage_request_v0",
+      runtime_root: root, goal_id: goal, state_path: join(root, "state.md"), fence: {
+        schema_version: "loopx_legacy_coordination_writer_fence_v0", state: "engaged", goal_id: goal,
+        fence_id: "inspection", source_version: "state:1", source_projection_sha256: canonicalAuthoritySha256(projection),
+        expected_shadow_provider_revision: head.provider_revision}})).status, "applied");
+  }
+  const request = {schema_version: TASK_LEASE_INSPECT_REQUEST, source: provider === "legacy" ? "legacy" : "canonical",
+    runtime_root: root, goal_id: goal, todo_id: "todo_inspect_allowed", authority};
+  return {root, goal, registry, store, request, authorityProvider, cases};
+}
+
+for (const provider of ["legacy", "file", "sqlite", "postgresql"] as const) {
+  for (const schema of provider === "legacy" ? ["legacy"] as const : ["legacy", "native"] as const) {
+    test(`${provider} ${schema} full mixed head reports effective ownership without writes`, {skip: provider === "postgresql" && !pool}, async t => {
+      const {request, store, authorityProvider, cases} = await fixture(t, provider, schema);
+      const before = await store?.loadAuthority();
+      for (const item of cases) {
+        const result = await inspectTaskLease({...request, todo_id: `todo_inspect_${item.name}`}, {now: () => NOW, authorityProvider});
+        assert.equal(result.ok, true, JSON.stringify(result));
+        assert.equal(result.active, item.active, item.name);
+        assert.deepEqual(result.executor_constraint, item.constraint, item.name);
+        if (store && before?.status === "loaded") {
+          assert.equal(result.source_authority, `${provider}_v0`);
+          assert.equal(result.provider_revision, before.provider_revision);
+          assert.equal(result.legacy_fallback_used, false); assert.equal(result.lease_path, null);
+        }
+      }
+      assert.deepEqual(await store?.loadAuthority(), before);
+      if (store) {
+        const stale = {...request, authority: {...request.authority, handoff_mode: "invalid", todos: [],
+          todo_projection_error: {code: "stale_display", message: "unused"}}};
+        assert.equal((await inspectTaskLease(stale, {now: () => NOW, authorityProvider})).active, true);
+      }
+    });
+  }
+}
+
+for (const fault of ["registration", "route", "clock", "field"] as const) {
+  test(`inspection rejects ${fault} change instead of publishing an effective observation`, async t => {
+    const {request, store, root, goal, registry} = await fixture(t, "file");
+    const before = await store!.loadAuthority();
+    const result = await inspectTaskLease(fault === "field" ? {...request, lock_token: "unexpected"} : request, {now: () => {
+      if (fault === "registration") writeFileSync(registry, "changed registration");
+      if (fault === "route") unlinkSync(legacyCoordinationWriterFencePath(root, goal));
+      return fault === "clock" ? new Date(NaN) : NOW;
+    }});
+    assert.equal(result.ok, false);
+    assert.equal(result.error_code, fault === "clock" ? "invalid_inspection_clock" : fault === "field" ? "invalid_inspection_request" : "authority_source_changed");
+    assert.equal(result.active, undefined); assert.deepEqual(await store!.loadAuthority(), before);
+  });
+}
+
+test("legacy inspection preserves projection errors and rejects malformed active expiry", async t => {
+  const {request, root, goal} = await fixture(t, "legacy");
+  const unavailable = await inspectTaskLease({...request, authority: {...request.authority,
+    todo_projection_error: {code: "todo_projection_unavailable", message: "source cannot be projected"}}}, {now: () => NOW});
+  assert.equal(unavailable.active, false);
+  assert.deepEqual(unavailable.executor_constraint, {effective: false, reason: "todo_projection_unavailable"});
+  const lease = {schema_version: "task_lease_v0", status: "active", owner: "agent-a", expires_at: "broken"};
+  await writeFile(join(root, "goals", goal, "task-leases", `${request.todo_id}.json`), JSON.stringify(lease));
+  assert.equal((await inspectTaskLease(request, {now: () => NOW})).error_code, "corrupt_lease");
+});

From 5321a09804a4bef3724425b402ee4f0c56d0ea12 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Mon, 21 Sep 2026 02:58:07 +0800
Subject: [PATCH 2/3] docs(coordination): define effective lease observation
 and migration boundary

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 ...shared-goal-authority-state-provider-v0.md |  7 +++-
 ...-goal-authority-state-provider-v0.zh-CN.md |  5 ++-
 .../typescript-control-plane-migration-v0.md  |  9 ++++-
 ...script-control-plane-migration-v0.zh-CN.md |  6 ++-
 docs/reference/canonical-lease-renew.md       | 38 +++++++++++++++++++
 5 files changed, 60 insertions(+), 5 deletions(-)

diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
index 45f0d1e4f4..03070f701c 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
@@ -2932,7 +2932,12 @@ The T3 lease-inspection reader now binds Todo, lease and handoff mode to one
 provider revision and never reads obsolete local lease files after promotion.
 Its eligibility policy is shared with current acquire/lifecycle rules, including
 claim divergence and exclusion; a read result is not a lease grant or a commit
-receipt. An empty canonical lease set stays empty. This read closure and removal
+receipt. Both source routes now interpret time and eligibility in TS, including
+archived-open retained history and explicit malformed-expiry failure; registration
+and promotion-fence changes are revalidated with bounded retry. Python no longer
+reconstructs the canonical head or diagnostic policy for inspection. See
+[the read contract](../../reference/canonical-lease-renew.md#what-inspection-proves).
+An empty canonical lease set stays empty. This read closure and removal
 of duplicate eligibility rules do not qualify a provider, alter CAS/replay or
 relax D1–D3; permanent Markdown display and the remaining roadmap stay intact.
 The ownership-edit slice now uses the same typed authoring and lifecycle boundary
diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
index 2e7c50c9a1..92a2ca7d65 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
@@ -2330,7 +2330,10 @@ Task graph 的 T3 topology consumer 现共用 inventory/horizon 关系目录，
 T3 lease inspect 已将 Todo、lease 与 handoff mode 绑定到同一 provider revision，
 promotion 后不再读取本地旧 lease 文件；canonical 空租约集合保持为空。资格策略与
 当前 acquire/lifecycle 共用 TS owner，包含 claim 分歧和 exclusion；读取结果不是
-租约授权，也不是 commit receipt。该 reader 闭合和重复规则删除不代表 provider
+租约授权，也不是 commit receipt。两条来源路径的时间／资格解释现收敛到 TS，
+覆盖归档 open 历史与损坏到期时间；注册来源及晋升 fence 变化须重新校验并有界重试。
+Python 不再为检查重建 canonical head 或诊断规则，见[读取合同](../../reference/canonical-lease-renew.md#what-inspection-proves)。
+该 reader 闭合和重复规则删除不代表 provider
 资格化，不改变 CAS/replay 或 D1–D3；永久 Markdown 展示与后续规划继续保留。
 ownership 编辑在 promotion 后现在与现有 update transaction 共用 typed authoring
 和 lifecycle 边界。claim/exclusion 门禁保留，带 lease 的 ownership 重写继续拒绝；
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index 8109e2e67a..d442ef7193 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -825,8 +825,13 @@ boundary, not all graph source delivery or the remaining T1–T4 work.
 Lease inspection now consumes one canonical Todo/lease/handoff-mode revision
 after promotion; an absent canonical lease does not revive a local lease file,
 and provider failure cannot fall back to Markdown. The read reports its provider
-revision without repairing display or changing the lease. Unpromoted inspection
-retains its legacy source contract. The shared `task_lease_eligibility.ts` owner
+revision without repairing display or changing the lease. Both routes now use
+`task_lease_inspection.ts` for time and eligibility interpretation; Python only
+projects source-bound registration/legacy facts and transports the response.
+Diagnostics reuse the TS rejection owner, including archived-Todo eligibility,
+strict active expiry and bounded source-change retry. Unpromoted storage stays
+unchanged; malformed active expiry intentionally becomes a visible error. See
+[inspection semantics](../../reference/canonical-lease-renew.md#what-inspection-proves). The shared `task_lease_eligibility.ts` owner
 also replaces the Python authority-core and three TS owner-eligibility copies
 used by acquire, lifecycle and terminal fencing. Current-lease effectiveness is
 derived inside acquire from the supplied owner/claim/exclusion/registration facts,
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
index dd7e7d7954..9e58aa2b9f 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
@@ -627,7 +627,11 @@ evidence/handoff 的脱敏展示。明确的语义修正：successor 谱系不
 
 Lease inspect 在 promotion 后从同一 canonical revision 读取 Todo、lease 与
 handoff mode；canonical 无租约不复活本地旧文件，provider 失败不回退 Markdown。
-结果携带 provider revision，读取不修复展示、不修改租约；未 promotion 的来源契约保留。
+结果携带 provider revision，读取不修复展示、不修改租约。两条路径现由
+`task_lease_inspection.ts` 统一时间与资格解释，Python 仅投递绑定来源的注册／legacy
+事实和响应；诊断字段复用 TS 拒绝规则。归档 Todo 不产生有效租约，active 到期时间
+损坏明确报错，来源变化有界重试。未晋升存储保持原状，错误语义变更见
+[检查合同](../../reference/canonical-lease-renew.md#what-inspection-proves)。
 `task_lease_eligibility.ts` 同时替代 Python authority core 和三处 TS owner 资格判断，
 供 acquire、lifecycle 与终态 fence 复用。当前租约是否有效由 acquire 内部根据同一输入
 的 owner/claim/exclusion/注册事实推导，不再由旧 `effective` 派生提示覆盖。
diff --git a/docs/reference/canonical-lease-renew.md b/docs/reference/canonical-lease-renew.md
index c84f65fe15..61747685de 100644
--- a/docs/reference/canonical-lease-renew.md
+++ b/docs/reference/canonical-lease-renew.md
@@ -49,6 +49,44 @@ already assigned to the eligible receiver. Without `--transfer-claim`, transfer
 claim. Neither form overrides an exclusion or widens write scopes. Use the actual readback
 versions, not these example numbers.
 
+## What inspection proves
+
+`task-lease inspect` uses one TS read owner for legacy files and selected
+File/SQLite authority. Service-owned PostgreSQL uses the same reader through its
+existing identity-fenced factory; the CLI does not gain a PostgreSQL connection
+or enable a service deployment. Promoted inspection reads Todo, lease and mode
+from one complete provider revision, ignoring stale display and lease files.
+
+`lease.status: active` describes the retained record. The top-level `active`
+means its expiry is strictly after the observation clock **and** its current
+owner is eligible for the active, open Todo. An archived record cannot revive
+execution even when imported history retains `status: open` and a future lease.
+Closed, unregistered, excluded and conflicting-claim owners return `active:
+false` with the existing `executor_constraint` diagnostic. Rejection precedence
+and diagnostic fields share the mutation-admission owner.
+
+An active lease with an invalid expiry now fails with `corrupt_lease`, including
+on the legacy route; it no longer looks like an ordinary inactive lease. Missing,
+expired and released leases retain their previous successful inactive response.
+For legacy storage, TS first checks the retained lease and only asks Python for
+a full Todo projection when it is time-active; the final check re-reads the lease
+after that projection. Inactive inspection does not parse the work history.
+Registration-source receipts and the promotion fence are rechecked after the
+read. Source changes trigger at most three host attempts, then an explicit
+`authority_source_changed` failure. Selected-provider failures never fall back.
+
+This is a read-only observation, not an execution grant or a lock across later
+work. A subsequent registration change, provider commit or expiry can invalidate
+it. Mutations must still prove their current owner/key/version and commit under
+their existing fences. Inspection does not repair display, renew a lease, change
+provider selection or spend quota.
+
+检查由 TS 统一解释两条来源路径：`lease.status` 是保留记录的状态，顶层 `active`
+还要求租约未到期、Todo 活跃且未关闭、owner 仍满足注册、排除与认领约束。
+归档但仍标为 open 的历史记录不再产生有效执行资格；损坏的 active 到期时间明确
+报错，不再伪装成正常失效。检查前后校验注册来源与晋升 fence，最多重试三次；
+结果只代表一次观察，后续写操作仍须校验当前执行证明，不获得新的授权。
+
 ## Atomically hand over claimed work
 
 When the current canonical `hard_lease` Todo and lease both belong to the sender,

From 4fb3501009b4b5723c9fee1421a554b93077d5cc Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Mon, 21 Sep 2026 03:02:59 +0800
Subject: [PATCH 3/3] test(control-plane): fence deferred inspection transport
 and read-only records

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../work_items/task_lease_acquire_adapter.py          |  2 +-
 tests/control_plane_ts/task_lease_inspection.test.ts  | 11 +++++++++--
 2 files changed, 10 insertions(+), 3 deletions(-)

diff --git a/loopx/control_plane/work_items/task_lease_acquire_adapter.py b/loopx/control_plane/work_items/task_lease_acquire_adapter.py
index 53f191f046..cd5a87d5ba 100644
--- a/loopx/control_plane/work_items/task_lease_acquire_adapter.py
+++ b/loopx/control_plane/work_items/task_lease_acquire_adapter.py
@@ -862,7 +862,7 @@ def inspect_native_task_lease(
         }
         result = effect_runtime_result("task_lease.inspect.native", request)
         if isinstance(result, dict) and result.get("todo_projection_required") is True:
-            if canonical or result.get("ok") is not True or result.get("action") != "inspect":
+            if canonical or result.get("schema_version") != TASK_LEASE_SCHEMA_VERSION or result.get("ok") is not True or result.get("action") != "inspect":
                 raise RuntimeError("native lease inspection requested an invalid source projection")
             request["phase"] = "effective_lease"
             request["authority"] = task_lease_acquire_authority_facts(
diff --git a/tests/control_plane_ts/task_lease_inspection.test.ts b/tests/control_plane_ts/task_lease_inspection.test.ts
index 7d4f74dee3..9fd0b24bce 100644
--- a/tests/control_plane_ts/task_lease_inspection.test.ts
+++ b/tests/control_plane_ts/task_lease_inspection.test.ts
@@ -2,7 +2,7 @@
 import assert from "node:assert/strict";
 import {createHash, randomUUID} from "node:crypto";
 import {writeFileSync, unlinkSync} from "node:fs";
-import {mkdtemp, mkdir, writeFile, rm} from "node:fs/promises";
+import {mkdtemp, mkdir, writeFile, readFile, readdir, rm} from "node:fs/promises";
 import {tmpdir} from "node:os";
 import {join} from "node:path";
 import test from "node:test";
@@ -108,8 +108,11 @@ async function fixture(t: test.TestContext, provider: Provider, schema: "native"
 for (const provider of ["legacy", "file", "sqlite", "postgresql"] as const) {
   for (const schema of provider === "legacy" ? ["legacy"] as const : ["legacy", "native"] as const) {
     test(`${provider} ${schema} full mixed head reports effective ownership without writes`, {skip: provider === "postgresql" && !pool}, async t => {
-      const {request, store, authorityProvider, cases} = await fixture(t, provider, schema);
+      const {request, store, authorityProvider, cases, root, goal} = await fixture(t, provider, schema);
       const before = await store?.loadAuthority();
+      const directory = join(root, "goals", goal, "task-leases");
+      const legacyNames = provider === "legacy" ? (await readdir(directory)).sort() : [];
+      const legacyBytes = await Promise.all(legacyNames.map(name => readFile(join(directory, name))));
       for (const item of cases) {
         const result = await inspectTaskLease({...request, todo_id: `todo_inspect_${item.name}`}, {now: () => NOW, authorityProvider});
         assert.equal(result.ok, true, JSON.stringify(result));
@@ -122,6 +125,10 @@ for (const provider of ["legacy", "file", "sqlite", "postgresql"] as const) {
         }
       }
       assert.deepEqual(await store?.loadAuthority(), before);
+      if (provider === "legacy") {
+        assert.deepEqual((await readdir(directory)).sort(), legacyNames);
+        assert.deepEqual(await Promise.all(legacyNames.map(name => readFile(join(directory, name)))), legacyBytes);
+      }
       if (store) {
         const stale = {...request, authority: {...request.authority, handoff_mode: "invalid", todos: [],
           todo_projection_error: {code: "stale_display", message: "unused"}}};