From 92429e8c4ce3674e4170c83af519e3c902e48d9a Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Mon, 21 Sep 2026 18:19:17 +0800
Subject: [PATCH 1/3] fix(issue-fix): close grouped Monitor execution and
 recovery lifecycle

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/capabilities/issue_fix/cli.py           |   1 +
 .../issue_fix/pr_monitor_materialization.py   | 362 +++++++-----------
 .../issue_fix_monitor_reconciliation.ts       | 134 +++++++
 .../control_plane/effect_runtime_handlers.ts  |   2 +
 loopx/todos.py                                |  15 +-
 5 files changed, 287 insertions(+), 227 deletions(-)
 create mode 100644 loopx/control_plane/capabilities/issue_fix_monitor_reconciliation.ts

diff --git a/loopx/capabilities/issue_fix/cli.py b/loopx/capabilities/issue_fix/cli.py
index 9a11114636..e1d586e45e 100644
--- a/loopx/capabilities/issue_fix/cli.py
+++ b/loopx/capabilities/issue_fix/cli.py
@@ -1315,6 +1315,7 @@ def handle_issue_fix_command(
                     raise ValueError("PR lifecycle transition is missing")
                 grouped_monitor_writeback = materialize_issue_fix_grouped_monitors(
                     registry_path=registry_path,
+                    runtime_root=Path(runtime_root_arg).expanduser() if runtime_root_arg else None,
                     goal_id=args.goal_id,
                     project=Path(args.project).expanduser(),
                     ledger_path=ledger_path,
diff --git a/loopx/capabilities/issue_fix/pr_monitor_materialization.py b/loopx/capabilities/issue_fix/pr_monitor_materialization.py
index 9414ac8364..21ff1bc602 100644
--- a/loopx/capabilities/issue_fix/pr_monitor_materialization.py
+++ b/loopx/capabilities/issue_fix/pr_monitor_materialization.py
@@ -1,34 +1,31 @@
+"""Execute the typed issue-fix Monitor reconciliation plan through public writers.
+
+Each bucket is a separate recoverable mutation. The ledger is an observation,
+not permission to borrow another execution's lease or bypass Todo admission.
+"""
 from __future__ import annotations
 
 import hashlib
 import json
-from collections.abc import Mapping
 from pathlib import Path
 from typing import Any
+from uuid import uuid4
 
-from ...control_plane.scheduler.monitor_todo import (
-    monitor_next_due_at,
-)
+from ...control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from ...control_plane.todos.monitor_metadata import MonitorPollObservation
 from ...control_plane.coordination.local_authority import LOCAL_AUTHORITY_SOURCES
 from ...control_plane.todos.provider_projection import settle_canonical_todo_projection
-from ...control_plane.work_items.task_lease import runtime_root_from_registry
-from ...todos import (
-    add_goal_todo,
-    complete_goal_todo,
-    list_goal_todos,
-    update_goal_todo,
+from ...control_plane.work_items.task_lease import (
+    acquire_task_lease, inspect_task_lease, release_task_lease,
+    runtime_root_from_registry, TaskLeaseError,
 )
+from ...todos import add_goal_todo, complete_goal_todo, list_goal_todos, update_goal_todo
 
-ISSUE_FIX_GROUPED_MONITOR_WRITEBACK_SCHEMA_VERSION = (
-    "issue_fix_grouped_monitor_writeback_v0"
-)
+ISSUE_FIX_GROUPED_MONITOR_WRITEBACK_SCHEMA_VERSION = "issue_fix_grouped_monitor_writeback_v0"
 DEFAULT_ISSUE_FIX_MONITOR_CADENCE = "30m"
 
 
 def _load_lifecycle_rows(path: Path) -> list[dict[str, Any]]:
-    if not path.exists():
-        return []
     rows: list[dict[str, Any]] = []
     for line_number, raw_line in enumerate(
         path.read_text(encoding="utf-8").splitlines(), start=1
@@ -49,224 +46,137 @@ def _load_lifecycle_rows(path: Path) -> list[dict[str, Any]]:
     return rows
 
 
-def _active_grouped_monitors(
-    rows: list[dict[str, Any]],
-) -> dict[str, dict[str, Any]]:
-    grouped: dict[str, dict[str, Any]] = {}
-    for row in rows:
-        projection = row.get("grouped_monitor_projection")
-        if not isinstance(projection, Mapping):
-            continue
-        if projection.get("materialize_nonempty_bucket_monitor") is not True:
-            continue
-        target_key = str(projection.get("target_key") or "").strip()
-        member_key = str(projection.get("member_key") or "").strip()
-        action_kind = str(projection.get("action_kind") or "").strip()
-        state_bucket = str(projection.get("state_bucket") or "").strip()
-        repository = str(projection.get("repository") or "").strip()
-        if (
-            not target_key
-            or not member_key
-            or not action_kind
-            or not state_bucket
-            or not repository
-        ):
-            continue
-        group = grouped.setdefault(
-            target_key,
-            {
-                "target_key": target_key,
-                "action_kind": action_kind,
-                "state_bucket": state_bucket,
-                "repository": repository,
-                "member_keys": set(),
-            },
-        )
-        group["member_keys"].add(member_key)
-    return grouped
-
-
-def _existing_issue_fix_monitors(
-    *, registry_path: Path, goal_id: str, project: Path
-) -> tuple[dict[str, dict[str, Any]], str | None]:
-    payload = list_goal_todos(
-        registry_path=registry_path,
-        goal_id=goal_id,
-        role="agent",
-        project=project,
-    )
-    monitors: dict[str, dict[str, Any]] = {}
-    for item in payload.get("todos") or []:
-        if not isinstance(item, dict):
-            continue
-        target_key = str(item.get("target_key") or "").strip()
-        action_kind = str(item.get("action_kind") or "").strip()
-        if (
-            item.get("task_class") == "continuous_monitor"
-            and target_key.startswith("github-pr-state-")
-            and action_kind.startswith("issue_fix_pr_state_")
-        ):
-            monitors[target_key] = item
-    return monitors, (payload.get("authority_read") or {}).get("source_authority")
-
 
-def _group_fingerprint(member_keys: set[str]) -> str:
-    encoded = json.dumps(sorted(member_keys), separators=(",", ":"))
-    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()[:16]
+def _plan(*, registry_path: Path, goal_id: str, project: Path,
+          rows: list[dict[str, Any]], cadence: str, generated_at: str, runtime_root: Path) -> tuple[dict[str, Any], str | None]:
+    source = list_goal_todos(registry_path=registry_path, goal_id=goal_id, role="agent", project=project,
+                             runtime_root_arg=str(runtime_root))
+    try:
+        plan = effect_runtime_result("capabilities.issue_fix.monitor_reconciliation.plan", {
+            "schema_version": "loopx_issue_fix_monitor_plan_request_v0", "ledger_rows": rows,
+            "todos": source["todos"], "cadence": cadence, "generated_at": generated_at,
+        })
+    except EffectRuntimeRejected as error:
+        raise ValueError(str(error)) from error
+    if not isinstance(plan, dict) or plan.get("schema_version") != "loopx_issue_fix_monitor_plan_result_v0":
+        raise TypeError("TypeScript Monitor reconciliation plan shape mismatch")
+    return plan, (source.get("authority_read") or {}).get("source_authority")
+
+
+def _release_attempt(*, registry_path: Path, runtime_root: Path, goal_id: str,
+                     todo_id: str, owner: str, prefix: str,
+                     proof: dict[str, Any] | None = None) -> None:
+    """Recover cleanup after a commit without renewing or borrowing execution."""
+    current = inspect_task_lease(registry_path=registry_path, runtime_root=runtime_root,
+                                goal_id=goal_id, todo_id=todo_id)
+    if current.get("ok") is not True:
+        raise TaskLeaseError("Monitor cleanup readback failed; retry after inspection",
+            code=str(current.get("error_code") or "monitor_cleanup_unavailable"))
+    lease = current.get("lease") or {}
+    key = str(lease.get("idempotency_key") or "")
+    if (lease.get("status") == "active" and lease.get("owner") == owner and key.startswith(prefix)
+        and (proof is None or (key == proof["task_lease_idempotency_key"]
+             and lease.get("version") == proof["task_lease_expected_version"]))):
+        release_task_lease(registry_path=registry_path, runtime_root=runtime_root,
+            goal_id=goal_id, todo_id=todo_id, owner=owner, idempotency_key=key,
+            expected_version=lease["version"])
 
 
 def materialize_issue_fix_grouped_monitors(
-    *,
-    registry_path: Path,
-    goal_id: str,
-    project: Path,
-    ledger_path: Path,
-    claimed_by: str,
-    cadence: str,
-    generated_at: str,
+    *, registry_path: Path, goal_id: str, project: Path, ledger_path: Path,
+    claimed_by: str, cadence: str, generated_at: str, runtime_root: Path | None = None,
 ) -> dict[str, Any]:
-    """Reconcile issue-fix PR lifecycle buckets into generic monitor todos."""
-
-    groups = _active_grouped_monitors(_load_lifecycle_rows(ledger_path))
-    existing, source_authority = _existing_issue_fix_monitors(
-        registry_path=registry_path,
-        goal_id=goal_id,
-        project=project,
-    )
-    next_due_at = monitor_next_due_at(
-        generated_at=generated_at,
-        cadence=cadence,
-    )
-    if next_due_at is None:
-        raise ValueError("issue-fix grouped monitor cadence must be parseable")
-
+    """Reconcile a complete ledger; preserve creator and execution ownership."""
+    runtime_root = runtime_root_from_registry(registry_path, str(runtime_root) if runtime_root is not None else None)
+    rows = _load_lifecycle_rows(ledger_path)
+    plan, source_authority = _plan(registry_path=registry_path, goal_id=goal_id,
+        project=project, rows=rows, cadence=cadence, generated_at=generated_at, runtime_root=runtime_root)
     writes: list[dict[str, Any]] = []
-    for target_key, group in sorted(groups.items()):
-        member_keys = set(group["member_keys"])
-        result_hash = _group_fingerprint(member_keys)
-        previous = existing.get(target_key)
-        previous_hash = str((previous or {}).get("result_hash") or "")
-        reopening = bool((previous or {}).get("done"))
-        material_change = reopening or previous_hash != result_hash
-        monitor_metadata = {
-            "target_key": target_key,
-            "cadence": cadence,
-            "next_due_at": next_due_at,
-            "last_checked_at": generated_at,
-            "result_hash": result_hash,
-            "consecutive_no_change": "0",
-            "material_change": "true" if material_change else "false",
-            "watch_only": "true",
-        }
-        reason = (
-            f"Issue-fix PR lifecycle bucket {group['state_bucket']} contains "
-            f"{len(member_keys)} active member(s)."
-        )
-        if previous:
-            schedule_complete = bool(
-                str(previous.get("cadence") or "").strip() == cadence
-                and str(previous.get("next_due_at") or "").strip()
-            )
-            if not reopening and not material_change and schedule_complete:
-                writes.append(
-                    {
-                        "operation": "unchanged",
-                        "target_key": target_key,
-                        "write_performed": False,
-                    }
-                )
-                continue
-            result = update_goal_todo(
-                registry_path=registry_path,
-                goal_id=goal_id,
-                todo_id=str(previous["todo_id"]),
-                role="agent",
-                status="open" if reopening else None,
-                reason=reason,
-                # Membership is an observation, not precomputed Todo state.
-                # The writer derives counters/generation against its locked
-                # snapshot, just like quota monitor-poll.
-                monitor_metadata=MonitorPollObservation(
-                    generated_at=generated_at, result_hash=result_hash,
-                    material_change=material_change, target_key=target_key,
-                    cadence=cadence, next_due_at=next_due_at,
-                ),
-                no_followup=False if reopening else None,
-                agent_id=claimed_by,
-                project=project,
-            )
-            writes.append(
-                {
-                    "operation": "update",
-                    "target_key": target_key,
-                    "write_performed": bool(result.get("changed")),
-                }
-            )
+    for step in plan["steps"]:
+        operation, target = step["operation"], step["target_key"]
+        identity = json.dumps([goal_id, step.get("todo_id"), claimed_by, generated_at, cadence, rows],
+                              sort_keys=True, separators=(",", ":"))
+        prefix = "issue-fix-monitor:" + hashlib.sha256(identity.encode()).hexdigest() + ":"
+        if operation == "unchanged":
+            _release_attempt(registry_path=registry_path, runtime_root=runtime_root,
+                goal_id=goal_id, todo_id=step["todo_id"], owner=claimed_by, prefix=prefix)
+            writes.append({"operation": operation, "target_key": target, "write_performed": False})
             continue
-        result = add_goal_todo(
-            registry_path=registry_path,
-            goal_id=goal_id,
-            role="agent",
-            text=(
-                f"[P2] Monitor {group['repository']} issue-fix PR lifecycle "
-                f"bucket {group['state_bucket']} for material changes."
-            ),
-            task_class="continuous_monitor",
-            action_kind=str(group["action_kind"]),
-            claimed_by=claimed_by,
-            monitor_metadata=monitor_metadata,
-            project=project,
-        )
-        writes.append(
-            {
-                "operation": "add",
-                "target_key": target_key,
-                "write_performed": bool(
-                    result.get("added") or result.get("metadata_updated")
-                ),
-            }
-        )
-
-    for target_key, item in sorted(existing.items()):
-        if target_key in groups or item.get("done") is True:
+        if operation == "add":
+            result = add_goal_todo(registry_path=registry_path, goal_id=goal_id, role="agent",
+                text=step["text"], priority="P2", task_class="continuous_monitor",
+                action_kind=step["action_kind"], claimed_by=claimed_by, agent_id=claimed_by,
+                monitor_metadata=step["metadata"], project=project, runtime_root_arg=str(runtime_root))
+            writes.append({"operation": "add", "target_key": target,
+                "write_performed": bool(result.get("added") or result.get("metadata_updated"))})
             continue
-        result = complete_goal_todo(
-            registry_path=registry_path,
-            goal_id=goal_id,
-            todo_id=str(item["todo_id"]),
-            role="agent",
-            evidence=f"Issue-fix PR lifecycle bucket {target_key} is empty.",
-            no_followup=True,
-            claimed_by=str(item.get("claimed_by") or claimed_by),
-            agent_id=claimed_by,
-            project=project,
-        )
-        writes.append(
-            {
-                "operation": "complete",
-                "target_key": target_key,
-                "write_performed": bool(result.get("changed")),
-            }
-        )
-
-    result = {
-        "schema_version": ISSUE_FIX_GROUPED_MONITOR_WRITEBACK_SCHEMA_VERSION,
-        "write_performed": any(item["write_performed"] for item in writes),
-        "path_recorded": False,
-        "active_bucket_count": len(groups),
-        "active_member_count": sum(
-            len(group["member_keys"]) for group in groups.values()
-        ),
-        "next_due_at": next_due_at if groups else None,
-        "writes": writes,
-    }
+        todo_id = step["todo_id"]
+        proof: dict[str, Any] = {}
+        acquired: dict[str, Any] | None = None
+        # Reactivation cannot acquire against completed work. Its existing typed
+        # transaction retires old execution; a subsequent observation acquires anew.
+        if operation != "reactivate":
+            inspected = inspect_task_lease(registry_path=registry_path, runtime_root=runtime_root,
+                goal_id=goal_id, todo_id=todo_id)
+            if inspected.get("ok") is not True:
+                raise TaskLeaseError(str(inspected.get("error") or "Monitor lease inspection failed"),
+                    code=str(inspected.get("error_code") or "monitor_lease_inspection_failed"))
+            if inspected.get("handoff_mode") == "hard_lease" or (
+                inspected.get("lease") is not None and inspected.get("handoff_mode") != "soft_claim"
+            ):
+                # Recover only this actor's exact observation attempt. A retired
+                # attempt gets a fresh key; historical receipts grant no execution.
+                prior = inspected.get("lease") or {}
+                retained_key = str(prior.get("idempotency_key") or "")
+                key = (retained_key if inspected.get("active") is True
+                       and prior.get("owner") == claimed_by and retained_key.startswith(prefix)
+                       else prefix + uuid4().hex)
+                acquired = acquire_task_lease(registry_path=registry_path, runtime_root=runtime_root,
+                    goal_id=goal_id, todo_id=todo_id, owner=claimed_by,
+                    idempotency_key=key, ttl_seconds=60, write_scopes=step["write_scopes"])
+                lease = acquired["lease"]
+                proof = {"task_lease_idempotency_key": key,
+                         "task_lease_expected_version": lease["version"]}
+        try:
+            if acquired is not None:
+                # A competing observation may have committed between planning
+                # and acquisition. Revalidate under our own execution, not the old
+                # display snapshot. The transaction verifies the proof again.
+                fresh, _ = _plan(registry_path=registry_path, goal_id=goal_id, project=project,
+                    rows=rows, cadence=cadence, generated_at=generated_at, runtime_root=runtime_root)
+                matching = [candidate for candidate in fresh["steps"] if candidate["target_key"] == target]
+                if not matching or matching[0]["operation"] == "unchanged":
+                    writes.append({"operation": "unchanged", "target_key": target, "write_performed": False})
+                    continue
+                if matching[0] != step:
+                    raise ValueError("Monitor source changed before execution; retry reconciliation")
+            if operation in {"observe", "reactivate"}:
+                result = update_goal_todo(registry_path=registry_path, goal_id=goal_id,
+                    todo_id=todo_id, role="agent", agent_id=claimed_by, project=project, runtime_root_arg=str(runtime_root),
+                    status="open" if operation == "reactivate" else None,
+                    no_followup=False if operation == "reactivate" else None,
+                    reason=step["reason"], monitor_metadata=MonitorPollObservation(**step["observation"]), **proof)
+                recorded_operation = "update"
+            elif operation == "complete":
+                result = complete_goal_todo(registry_path=registry_path, goal_id=goal_id,
+                    todo_id=todo_id, role="agent", agent_id=claimed_by, project=project, runtime_root_arg=str(runtime_root),
+                    evidence=step["evidence"], no_followup=True, **proof)
+                recorded_operation = "complete"
+            else:
+                raise TypeError(f"unsupported Monitor reconciliation operation: {operation}")
+            if result.get("status") in {"failed", "validation_failed", "rejected", "ambiguous"} or result.get("ok") is False:
+                raise ValueError(f"Monitor {operation} did not commit: {result.get('reason_code') or result.get('error') or result.get('status')}")
+            writes.append({"operation": recorded_operation, "target_key": target,
+                "write_performed": bool(result.get("changed"))})
+        finally:
+            if acquired is not None:
+                _release_attempt(registry_path=registry_path, runtime_root=runtime_root,
+                    goal_id=goal_id, todo_id=todo_id, owner=claimed_by, prefix=prefix, proof=proof)
+    result = {"schema_version": ISSUE_FIX_GROUPED_MONITOR_WRITEBACK_SCHEMA_VERSION,
+        "write_performed": any(item["write_performed"] for item in writes), "path_recorded": False,
+        "active_bucket_count": plan["active_bucket_count"], "active_member_count": plan["active_member_count"],
+        "next_due_at": plan["next_due_at"], "writes": writes}
     if source_authority in LOCAL_AUTHORITY_SOURCES:
-        # An unchanged observation may follow a committed write whose display
-        # delivery failed. Drain the current head without repeating that write.
-        result = settle_canonical_todo_projection(
-            payload={**result, "source_authority": source_authority},
-            registry_path=registry_path,
-            runtime_root=runtime_root_from_registry(registry_path, None),
-            goal_id=goal_id, project=project,
-        )
+        result = settle_canonical_todo_projection(payload={**result, "source_authority": source_authority},
+            registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, project=project)
     return result
diff --git a/loopx/control_plane/capabilities/issue_fix_monitor_reconciliation.ts b/loopx/control_plane/capabilities/issue_fix_monitor_reconciliation.ts
new file mode 100644
index 0000000000..aba210c1ed
--- /dev/null
+++ b/loopx/control_plane/capabilities/issue_fix_monitor_reconciliation.ts
@@ -0,0 +1,134 @@
+/** Issue-fix owns bucket identity; Todo/lease transactions own authorization and
+ * persistence. This complete plan is validated before the adapter starts effects. */
+import {createHash} from "node:crypto";
+import type {JsonObject} from "../effect_program.ts";
+import {requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts";
+import {authorityUnicodeCompare} from "../coordination/authority_store_codec.ts";
+import {normalizeWriteScopes} from "../work_items/task_lease_acquire.ts";
+import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
+import {parseTodoTimestampMicros} from "../runtime_timestamp.ts";
+import {evaluateSchedulerStateTransition, SCHEDULER_STATE_TRANSITION_REQUEST_SCHEMA} from "../scheduler/state_transition_rules.ts";
+
+export const ISSUE_FIX_MONITOR_PLAN_REQUEST = "loopx_issue_fix_monitor_plan_request_v0";
+export const ISSUE_FIX_MONITOR_PLAN_RESULT = "loopx_issue_fix_monitor_plan_result_v0";
+type Group = {target_key: string; action_kind: string; state_bucket: string; repository: string; members: Set<string>};
+type ExistingStep = {target_key: string; todo_id: string};
+export type MonitorReconciliationStep =
+  | {operation: "add"; target_key: string; text: string; action_kind: string; metadata: JsonObject}
+  | (ExistingStep & {operation: "observe" | "reactivate"; write_scopes: string[]; reason: string; observation: JsonObject})
+  | (ExistingStep & {operation: "complete"; write_scopes: string[]; evidence: string})
+  | (ExistingStep & {operation: "unchanged"});
+
+function fail(message: string): never { throw new EffectRuntimeRequestError(message); }
+function text(value: unknown): string { return typeof value === "string" ? value.trim() : ""; }
+function rows(value: unknown, label: string): JsonObject[] {
+  if (!Array.isArray(value)) return fail(`${label} must be a complete array`);
+  return value.map(row => requireJsonObject(row, label));
+}
+function timestamp(value: unknown, label: string): bigint {
+  const parsed = parseTodoTimestampMicros(requireNonEmptyString(value, label));
+  return parsed === null ? fail(`${label} must be an ISO timestamp`) : parsed;
+}
+
+function groupsFromLedger(ledger: JsonObject[]): Map<string, Group> {
+  const groups = new Map<string, Group>();
+  for (const row of ledger) {
+    const projection = requireJsonObject(row.grouped_monitor_projection, "grouped Monitor projection");
+    if (typeof projection.materialize_nonempty_bucket_monitor !== "boolean") {
+      fail("Monitor projection must explicitly declare whether its bucket is active");
+    }
+    if (!projection.materialize_nonempty_bucket_monitor) continue;
+    const target = requireNonEmptyString(projection.target_key, "Monitor target_key");
+    const member = requireNonEmptyString(projection.member_key, "Monitor member_key");
+    const action = requireNonEmptyString(projection.action_kind, "Monitor action_kind");
+    const bucket = requireNonEmptyString(projection.state_bucket, "Monitor state_bucket");
+    const repository = requireNonEmptyString(projection.repository, "Monitor repository");
+    if (!target.startsWith("github-pr-state-") || !action.startsWith("issue_fix_pr_state_")) {
+      fail("issue-fix Monitor ledger contains an unrelated target/action namespace");
+    }
+    const existing = groups.get(target);
+    if (existing && (existing.action_kind !== action || existing.state_bucket !== bucket || existing.repository !== repository)) {
+      fail(`conflicting Monitor bucket identity for ${target}`);
+    }
+    const group = existing ?? {target_key: target, action_kind: action, state_bucket: bucket, repository, members: new Set<string>()};
+    group.members.add(member);
+    groups.set(target, group);
+  }
+  return groups;
+}
+
+export function planIssueFixMonitorReconciliation(value: unknown): JsonObject {
+  const request = requireJsonObject(value, "issue-fix Monitor reconciliation");
+  if (request.schema_version !== ISSUE_FIX_MONITOR_PLAN_REQUEST || Object.keys(request).some(key =>
+      !["schema_version", "generated_at", "cadence", "ledger_rows", "todos"].includes(key))) {
+    fail("Monitor reconciliation schema/fields mismatch");
+  }
+  const generatedAt = requireNonEmptyString(request.generated_at, "generated_at");
+  const observedAt = timestamp(generatedAt, "generated_at");
+  const cadence = requireNonEmptyString(request.cadence, "cadence");
+  const schedule = evaluateSchedulerStateTransition({schema_version: SCHEDULER_STATE_TRANSITION_REQUEST_SCHEMA,
+    operation: "monitor_schedule", generated_at: generatedAt, cadence, explicit_next_due_at: null});
+  if (schedule.operation !== "monitor_schedule" || schedule.next_due_at === null) {
+    return fail("issue-fix grouped monitor cadence must be parseable");
+  }
+  const groups = groupsFromLedger(rows(request.ledger_rows, "ledger_rows"));
+  const existing = new Map<string, JsonObject>();
+  for (const item of rows(request.todos, "todos")) {
+    const target = text(item.target_key);
+    if (item.role !== "agent" || item.task_class !== "continuous_monitor" ||
+        !target.startsWith("github-pr-state-") || !text(item.action_kind).startsWith("issue_fix_pr_state_")) continue;
+    // Archived/superseded rows remain historical; they cannot shadow active work.
+    if (item.archive_state === "archive" || item.superseded_by) continue;
+    if (existing.has(target)) fail(`ambiguous Monitor target ${target}; resolve duplicate active Todo identities before reconciliation`);
+    requireNonEmptyString(item.todo_id, "Monitor todo_id");
+    existing.set(target, item);
+  }
+  const steps: MonitorReconciliationStep[] = [];
+  for (const [target, group] of [...groups].sort(([a], [b]) => authorityUnicodeCompare(a, b))) {
+    // Persisted Python identity uses code-point ordering and ensure_ascii=True.
+    // UTF-16 default sort / unescaped JSON would invent a membership change.
+    const membership = JSON.stringify([...group.members].sort(authorityUnicodeCompare))
+      .replace(/[\u0080-\uffff]/g, char => `\\u${char.charCodeAt(0).toString(16).padStart(4, "0")}`);
+    const resultHash = createHash("sha256").update(membership).digest("hex").slice(0, 16);
+    const previous = existing.get(target);
+    const reopening = previous?.status === "done" || previous?.done === true;
+    const material = reopening || text(previous?.result_hash) !== resultHash;
+    if (!previous) {
+      steps.push({operation: "add", target_key: target, action_kind: group.action_kind,
+        text: `Monitor ${group.repository} issue-fix PR lifecycle bucket ${group.state_bucket} for material changes.`,
+        metadata: {target_key: target, cadence, next_due_at: schedule.next_due_at, last_checked_at: generatedAt,
+          result_hash: resultHash, consecutive_no_change: "0", material_change: "true", watch_only: "true"}});
+      continue;
+    }
+    if (text(previous.action_kind) !== group.action_kind) fail(`Monitor action identity changed for ${target}`);
+    const todoId = String(previous.todo_id);
+    if (!reopening && !material && text(previous.cadence) === cadence && text(previous.next_due_at)) {
+      steps.push({operation: "unchanged", target_key: target, todo_id: todoId});
+    } else {
+      if (text(previous.last_checked_at) && observedAt < timestamp(previous.last_checked_at, "last_checked_at")) {
+        fail(`Monitor observation is older than persisted state for ${target}`);
+      }
+      if (reopening && observedAt <= timestamp(previous.completed_at, "completed_at")) {
+        fail(`Monitor reactivation must follow completion for ${target}`);
+      }
+      steps.push({operation: reopening ? "reactivate" : "observe", target_key: target, todo_id: todoId,
+        write_scopes: normalizeWriteScopes(previous.required_write_scopes),
+        reason: `Issue-fix PR lifecycle bucket ${group.state_bucket} contains ${group.members.size} active member(s).`,
+        observation: {generated_at: generatedAt, result_hash: resultHash, material_change: material,
+          target_key: target, cadence, next_due_at: schedule.next_due_at}});
+    }
+  }
+  for (const [target, previous] of [...existing].sort(([a], [b]) => authorityUnicodeCompare(a, b))) {
+    if (groups.has(target) || previous.status === "done" || previous.done === true) continue;
+    // An older empty ledger is not proof that a newer observed group is empty.
+    if (text(previous.last_checked_at) && observedAt < timestamp(previous.last_checked_at, "last_checked_at")) {
+      fail(`empty Monitor observation is older than persisted state for ${target}`);
+    }
+    steps.push({operation: "complete", target_key: target, todo_id: String(previous.todo_id),
+      write_scopes: normalizeWriteScopes(previous.required_write_scopes),
+      evidence: `Issue-fix PR lifecycle bucket ${target} is empty.`});
+  }
+  return {schema_version: ISSUE_FIX_MONITOR_PLAN_RESULT, steps,
+    active_bucket_count: groups.size, active_member_count: [...groups.values()].reduce((n, group) => n + group.members.size, 0),
+    next_due_at: groups.size ? schedule.next_due_at : null};
+}
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index f537f62c7c..386fad6a8d 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -1,4 +1,5 @@
 import {selectPeriodicReportProgress, selectPeriodicReportApprovalRetry} from "./capabilities/periodic_report_progress.ts";
+import {planIssueFixMonitorReconciliation} from "./capabilities/issue_fix_monitor_reconciliation.ts";
 import {inspectTaskLease} from "./work_items/task_lease_inspection.ts";
 import {evaluateTodoPriority} from "./todos/priority.ts";
 import {evaluateUserCompletion} from "./todos/user_completion.ts";
@@ -562,6 +563,7 @@ export function createEffectRuntimeHandlers(
       compileActionReviewPlan(params.proposal)],
     ["scheduler.monitor_successor.plan", planMonitorSuccessor],
     ["scheduler.monitor_target.select", selectMonitorTodoRequest],
+    ["capabilities.issue_fix.monitor_reconciliation.plan", planIssueFixMonitorReconciliation],
     ["coordination.local_authority_shadow.record", recordLocalAuthorityShadow],
     ["coordination.runtime_shadow.commit_entry", commitLocalAuthorityShadowEntry],
     ["coordination.runtime_shadow.outbox_read", readLocalAuthorityShadow],
diff --git a/loopx/todos.py b/loopx/todos.py
index be487070ba..16e6671dc8 100644
--- a/loopx/todos.py
+++ b/loopx/todos.py
@@ -131,6 +131,7 @@
 from .control_plane.todos.path_resolution import resolve_todo_state_path
 from .control_plane.todos.provider_terminal_lifecycle import provider_first_terminal_lifecycle
 from .control_plane.todos.handoff_mode import (
+    goal_handoff_mode,
     enter_added_todo_ownership_handoff_gate,
     enter_todo_ownership_handoff_gate,
     resolve_todo_completion_handoff,
@@ -1248,7 +1249,7 @@ def update_goal_todo(
     if (update_operation_id is not None or update_expected_provider_revision is not None
         or update_expected_registry_sha256 is not None) or (not claim_only and (
         task_lease_idempotency_key is not None or task_lease_expected_version is not None
-    )):
+    ) and not (monitor_intent["observation"] is not None and status is None)):
         raise ValueError("update operation id and lease proof require a supported promoted update; no legacy write attempted")
     resolved_project, resolved_state_file = resolve_todo_state_path(
         registry_path=registry_path,
@@ -1364,6 +1365,18 @@ def update_goal_todo(
             authority_reason=authority_reason,
             requested_claimed_by=effective_claimed_by,
         )
+        if monitor_intent["observation"] is not None and task_lease_idempotency_key is not None:
+            # Explicit observation proof uses the existing native held fence
+            # under the Markdown writer lock. Closing this guard does not retire
+            # execution; the acquiring caller owns release after observation.
+            handoff_gate_stack.enter_context(hold_task_lease_mutation_fence(
+                registry_path=registry_path, runtime_root=shadow_runtime_root,
+                goal_id=goal_id, todo_id=todo_id, todo=authority_todo, actor_agent_id=effective_agent_id,
+                idempotency_key=task_lease_idempotency_key,
+                expected_version=task_lease_expected_version,
+                require_active_when_key_supplied=True,
+                handoff={"handoff_mode": goal_handoff_mode(original)},
+            ))
         handoff_gate = enter_todo_ownership_handoff_gate(
             handoff_gate_stack,
             state_text=original,

From 30e01765eee9a30cd73b6f6cd9dfc554d78d51a9 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Mon, 21 Sep 2026 18:19:55 +0800
Subject: [PATCH 2/3] test(issue-fix): qualify leased Monitor reconciliation
 across providers

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 ...shared-goal-authority-state-provider-v0.md |   4 +-
 .../typescript-control-plane-migration-v0.md  |  20 +-
 ...script-control-plane-migration-v0.zh-CN.md |  13 +-
 loopx/capabilities/issue_fix/README.md        |  29 +++
 loopx/capabilities/issue_fix/README.zh-CN.md  |  21 +++
 ...sue_fix_grouped_monitor_materialization.py |  11 +-
 .../test_issue_fix_monitor_execution.py       | 173 ++++++++++++++++++
 .../authority_store_conformance.ts            |   2 +
 .../issue_fix_monitor_reconciliation.test.ts  |  87 +++++++++
 ..._fix_monitor_reconciliation_conformance.ts |  86 +++++++++
 .../production_scale_coordination_fixture.ts  |  19 ++
 11 files changed, 455 insertions(+), 10 deletions(-)
 create mode 100644 tests/capabilities/test_issue_fix_monitor_execution.py
 create mode 100644 tests/control_plane_ts/issue_fix_monitor_reconciliation.test.ts
 create mode 100644 tests/control_plane_ts/issue_fix_monitor_reconciliation_conformance.ts

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 66baebe95b..bac0923c6a 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
@@ -3211,7 +3211,7 @@ or moving a helper is not by itself a package exit.
 | A / L1: Monitor configuration (this slice) | Existing `todo update` config enters the TS planner/CAS/receipt; delete Python's duplicate intent field catalog. Separate authoring from observed hashes, times and generations. | Ordinary CLI/API, clear/omission, active lease proof, no-op/replay, failed display delivery, complete fixture and real providers. This does not complete delegated Chat or leased polling. |
 | A / L2: Complete public mutation admission | User completion updates share the TS edit/terminal transaction and reviewed Chat recovery; linked decision consumption/reject/cancel/resume now commit with the source, replacing Python followthrough rules. Continue the actual CLI/Turn/Chat inventory for remaining effect-owned decisions, delegated owner actions and Monitor lifecycle transitions; [caller contract](../../reference/canonical-todo-completion-update.md). | Build on merged T1 owners, not a generic raw patch. Prove permission rejection and exact caller response; remove replaced Python admission and name every remaining unsupported command. |
 | A / L3: Canonical lease lifecycle | Standalone acquire/takeover, atomic claim lease admission and maintenance reuse TS facts/decision/materialization and one provider opening fence. Explicit claimed-work transfer now commits source-authorized Todo ownership and the new lease generation together; canonical request types exclude legacy held-fence fields. Acquire success verifies current execution proof; canonical completion can recover missing display. | Full-head scope conflict, archived/ineffective holders, exact create-CAS retry, stale execution, process loss and real CLI/four-arm rehearsal are covered. [Operation and remaining callers](../../reference/canonical-lease-renew.md). Executor-held external-effect fences remain explicit work; D1–D3/default holds remain. |
-| B / L4: Leased Monitor poll and settlement | Current execution proof now binds CLI intent, observation/generation/independent-successor CAS and historical business receipt. Quota pending admission is frozen before the business write; recovery preserves that decision after lease retirement. | Existing L3 lease lifecycle, real File/SQLite/PostgreSQL, mixed fixtures, process death between business/quota commits, competing renewal and unchanged polling. [Operation and snapshot rehearsal](../../reference/protocols/quota-monitor-observation-receipt-v0.md). Ordinary polls leave leases unchanged and spend no quota; separate authorities stay separate. The retained grouped-Monitor observation/reactivation caller now uses Todo update v4 and the shared Monitor planner, with unchanged-group display recovery. Canonical reactivation now atomically retires retained execution and reopens the observation cycle, sharing typed admission with polling; a fresh execution still needs explicit acquisition. Executor acquisition for grouped reconciliation, wider L2 admission and D1–D3/default remain open. |
+| B / L4: Leased Monitor poll and settlement | Current execution proof now binds CLI intent, observation/generation/independent-successor CAS and historical business receipt. Quota pending admission is frozen before the business write; recovery preserves that decision after lease retirement. | Existing L3 lease lifecycle, real File/SQLite/PostgreSQL, mixed fixtures, process death between business/quota commits, competing renewal and unchanged polling. [Operation and snapshot rehearsal](../../reference/protocols/quota-monitor-observation-receipt-v0.md). Ordinary polls leave leases unchanged and spend no quota; separate authorities stay separate. The retained grouped-Monitor observation/reactivation caller now uses Todo update v4 and the shared Monitor planner, with unchanged-group display recovery. Canonical reactivation now atomically retires retained execution and reopens the observation cycle, sharing typed admission with polling; a fresh execution still needs explicit acquisition. Grouped reconciliation now acquires/revalidates/releases its own bounded execution, recovers interrupted cleanup, and plans the complete bucket set in TS; missing evidence and ambiguous/stale targets reject. This closes that retained caller across legacy/File/SQLite; native/imported mixed fixtures exercise the same effects on real PostgreSQL. Wider L2 admission, external-effect fences and D1–D3/default remain open. |
 | B / L5: Consumer and display closure | Reconcile #4316, audit Turn/quota/Dashboard/Chat source reads, and finish D1 freshness/recovery through the existing projection outbox. | CLI, Lark/Chat and packaged frontend read back their affected interactions; absent/stale display, empty canonical state, pending projection and data beyond UI limits. Delete post-promotion legacy fallbacks with each consumer. |
 | A–C / L6: Local durability qualification | Continue contributor-owned #4224/#4328 on the selected SQLite profile; reuse File/NoKV references and complete 7.2's ledger. | Capacity, real process/crash/restore/upgrade, retained receipts/scans, consumer lag, supported runtimes/OS and the separately authorized >=10-day synthetic soak. Missing measurements remain holds. |
 | A–C / L7: Capture continuity | Reconcile the merged #4315 archive/lease-membership repair; qualify its ladder row/mutant and sustained mixed-writer/event-source matrix rather than reimplementing the closed defect. | Real CLI/File capture, history retained, partial drain unqualified, crash/replay and a new lease after archive/rebootstrap. Keep the legacy migration window provable; T4 cannot be used to skip this row. |
@@ -3233,7 +3233,7 @@ PRs**, conditional on the caller audit finding no additional missing effects:
 | L7 capture plus L8 integrated migration | 1–2 | Mixed-writer continuity, fenced whole-Goal rehearsal, export/rollback and cohort evidence. |
 | L9 default and bounded retirement | 1 | New-Goal onboarding/settings/install choose the qualified profile; remove final obsolete callers. |
 
-The retained-Monitor cycle slice removes a concrete L4 hold, not an entire
+The retained-Monitor cycle and grouped executor closure remove concrete L4 holds, not an entire
 remaining package: the **5–8 PR planning range remains conditional**, rather than
 subtracting one for a lifecycle fix. Actual remaining executor/caller coverage,
 L5 consumers, contributor-owned D2, integrated migration and default onboarding
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index c6d3221637..9fb2ce0df2 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -783,8 +783,24 @@ all T2 commands or authorize whole-Goal promotion.
   must be acquired explicitly. `todo_monitor_cycle.ts` owns shared update/poll
   admission, removing their duplicate actor/lease branches and correcting their
   soft-claim disagreement. No new Python transition owner or RPC is introduced.
-  Other lifecycle callers, executor acquisition for grouped reconciliation,
-  legacy persistence/capture and whole-Goal qualification remain separate.
+  Grouped reconciliation now plans the complete bucket set in
+  `capabilities/issue_fix_monitor_reconciliation.ts`; Python retains ledger IO,
+  public writer calls and display delivery. The caller acquires its own bounded
+  execution for hard-lease observations/stops, rechecks the plan after acquisition,
+  and releases only that attempt. Retry after an observation commit can clean up
+  a surviving lease without repeating the business mutation. Reactivation remains
+  a non-execution transition through the existing TS owner.
+  Missing/malformed ledger evidence, duplicate active targets and older empty
+  observations now reject instead of silently dropping or completing a target.
+  Membership hashes preserve the original Python Unicode/ASCII encoding contract.
+  Explicit runtime-root routing reaches every affected writer and readback.
+  This is a complete issue-fix caller closure, not an atomic transaction over all
+  buckets: earlier committed buckets survive a later failure. Unchanged retries
+  may release their own interrupted execution and drain display. The remaining
+  Python adapter is a real caller, not a removable compatibility wrapper.
+  See the [operator contract](../../../loopx/capabilities/issue_fix/README.md#pr-lifecycle-monitor).
+  Other lifecycle callers, external-effect execution fences, legacy
+  persistence/capture and whole-Goal qualification remain separate.
 - Preserve unchanged polling/reschedule behavior, generation fences,
   material-change successor deduplication and accountable settlement.
   A monitor remains non-executable delivery context; its independent
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 086a9a65b7..8b1428e729 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
@@ -594,8 +594,17 @@ delivery pending；这不代表全部 T2 命令或整 Goal promotion 已完成
   raw patch 权限或轮询引擎。完成后的新观察即使 hash 相同也推进新一代；历史重放
   不会重开当前任务。无变化的分组也能恢复显示，包括带优先级前缀的 native 文本。
   见[观察更新与再激活](../../reference/protocols/quota-monitor-observation-receipt-v0.md#observation-updates-and-reactivation)。
-  保留 execution lease／hard-lease 模式的再激活、其他 lifecycle caller、旧持久化／
-  capture 和整 Goal 资格仍是独立边界。
+  再激活已由既有 TS owner 原子退役旧 execution；分组对账的完整桶集合决策现由
+  `capabilities/issue_fix_monitor_reconciliation.ts` 负责，Python 保留 ledger IO、
+  公开 writer 调用和展示交付。hard-lease 观察／结束先领取自己的有限期 execution，
+  领取后重新核对计划，只释放本次执行。观察提交后进程退出，原样重试可清理残留
+  lease，不重复 Todo 业务写入；再激活本身仍不授予执行权。
+  缺失／损坏 ledger、重复活动 target、旧的空组观察现在明确拒绝；成员 hash 保留
+  Python 原有 Unicode 排序及 ASCII 转义合同。显式 runtime-root 贯穿读取和写回。
+  这是 issue-fix 调用链闭合，不是所有桶的一笔原子事务：后续桶失败不回滚之前已
+  提交的桶。无变化重试可以清理自己的中断 execution 并恢复展示。Python 适配器仍
+  有真实调用方，不能直接删除。其他 lifecycle caller、跨外部 effect 的围栏、旧持久化／
+  capture 和整 Goal 资格仍独立。见[操作合同](../../../loopx/capabilities/issue_fix/README.zh-CN.md#pr-lifecycle-monitor)。
 - 保持 unchanged poll/reschedule、generation fence、material-change successor
   去重和可归属 settlement。Monitor 不是 delivery 执行任务；独立 advancement Todo
   不能被 monitor 自身替代。
diff --git a/loopx/capabilities/issue_fix/README.md b/loopx/capabilities/issue_fix/README.md
index 4a5c2b2d90..02a379cded 100644
--- a/loopx/capabilities/issue_fix/README.md
+++ b/loopx/capabilities/issue_fix/README.md
@@ -1054,6 +1054,35 @@ private material remain explicit gates. Each material transition must yield a
 `runnable_successor`, concrete blocker, or structured no-follow-up; unchanged
 polls remain quiet and do not spend delivery quota.
 
+With `--execute-transition`, one Monitor represents each nonempty repository/state
+bucket. Membership changes advance its observation generation; an empty bucket
+ends it, and a newer nonempty observation reopens the same unarchived Todo. New
+Monitors use explicit `priority=P2`; their text is not a priority argument.
+
+In `hard_lease` mode the reconciler acquires a 60-second execution lease for
+observation/stop and releases that exact execution afterward. It never borrows
+another attempt merely because the Agent ID matches. An identical retry can
+recover its own active acquisition; after a committed observation it cleans up
+its remaining lease without repeating that business write. Expired or released
+attempts require fresh acquisition. Reactivation itself grants no execution.
+The existing `--runtime-root` override applies to reads, leases, writes and
+projection recovery. No new provider or capability is enabled.
+
+Missing ledger files, malformed bucket declarations, duplicate current targets,
+and older empty observations are errors, not evidence that all work has ended.
+Restore the ledger or resolve the duplicate before retrying; do not switch to
+legacy state. Buckets commit independently, so after a later failure read back
+`loopx todo list --goal-id GOAL` using the same registry/runtime root, then retry
+the original observation. `write_performed` describes Todo business writes;
+lease cleanup and current display delivery may still happen on an unchanged
+retry. Pending display is recoverable with `loopx todo project-markdown` and
+never rolls back a successful business commit.
+
+Omit `--execute-transition` to inspect without reconciling Todos. To roll back
+this implementation, retain canonical state, writer fences and receipts; restore
+compatible code instead of reviving old Markdown authority. This does not grant
+publication, merge, additional capabilities or access to private material.
+
 Pass `--issue-ref` when persisting PR lifecycle state. This explicit public-safe
 link lets the outcome read model join the PR to its issue without guessing from
 branch names, titles, or text.
diff --git a/loopx/capabilities/issue_fix/README.zh-CN.md b/loopx/capabilities/issue_fix/README.zh-CN.md
index 9c1f545d2a..cd24a6e32b 100644
--- a/loopx/capabilities/issue_fix/README.zh-CN.md
+++ b/loopx/capabilities/issue_fix/README.zh-CN.md
@@ -927,6 +927,27 @@ review、maintainer correction、mergeability、stale branch 和 terminal status
 transition 必须生成 `runnable_successor`、具体 blocker 或结构化 no-follow-up；
 unchanged poll 保持安静且不消耗 delivery quota。
 
+`--execute-transition` 为每个非空的「仓库／状态」分组维护一个 Monitor。成员变化
+推进观察 generation；空组结束；完成后的新非空观察重新开启同一个未归档 Todo。
+创建时通过显式 `priority=P2` 传入优先级，不再把正文当前缀参数使用。
+
+`hard_lease` 模式的观察／结束先领取 60 秒 execution lease，之后只释放这次执行。
+Agent ID 相同也不能借用另一轮的 lease。原样重试可以恢复自己的活动领取；观察
+提交后进程退出，重试清理残留 lease 而不重复业务写入。到期或已释放的旧执行需要
+重新领取；再激活本身不授予执行权。已有 `--runtime-root` 参数贯穿读取、租约、
+写入和显示恢复，不新增 provider 或 capability 启用方式。
+
+缺失 ledger、损坏的分组声明、重复活动 target、旧的空组观察都会报错，不能当作
+「工作已全部结束」。恢复 ledger 或解决重复项后重试，不回退旧存储。各桶独立提交，
+后面的桶失败不会回滚前面的桶；用同一 registry/runtime root 执行
+`loopx todo list --goal-id GOAL` 读回，再重试原观察。`write_performed` 只表示 Todo
+业务写入；无变化重试仍可能清理 lease 和恢复当前显示。显示 pending 可通过
+`loopx todo project-markdown` 重试，不回滚已提交业务。
+
+不传 `--execute-transition` 即只检查而不对账 Todo。回滚实现时保留 canonical
+state、writer fence 和 receipt，恢复兼容代码，不能复活旧 Markdown authority。
+该过程不授予发布、合并、新 capability 或私有材料访问权限。
+
 持久化 PR lifecycle 时应传入 `--issue-ref`。这个显式、public-safe 的关联让 outcome
 read model 可以把 PR 精确连接到 issue，而不用从分支名、标题或正文中猜测。
 
diff --git a/tests/capabilities/test_issue_fix_grouped_monitor_materialization.py b/tests/capabilities/test_issue_fix_grouped_monitor_materialization.py
index 1d08e2bd18..229ed721af 100644
--- a/tests/capabilities/test_issue_fix_grouped_monitor_materialization.py
+++ b/tests/capabilities/test_issue_fix_grouped_monitor_materialization.py
@@ -74,7 +74,9 @@ def _fixture(
     return project, state, registry
 
 
-def _use_provider(state, registry, provider, monkeypatch):
+def _use_provider(state, registry, provider, monkeypatch, *, handoff_mode="soft_claim"):
+    if handoff_mode == "hard_lease":
+        state.write_text("---\nhandoff_mode: hard_lease\n---\n" + state.read_text())
     if provider == "legacy":
         return
     from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
@@ -82,7 +84,7 @@ def _use_provider(state, registry, provider, monkeypatch):
     isolate_sqlite_runtime(registry.parent, monkeypatch)
     todos = list_goal_todos(registry_path=registry, goal_id=GOAL_ID)["todos"]
     initialize_canonical_authority(registry.parent, GOAL_ID,
-        build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=todos, leases=[], handoff_mode="soft_claim"),
+        build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=todos, leases=[], handoff_mode=handoff_mode),
         state_path=state, provider=provider)
     state.unlink()
 
@@ -126,11 +128,12 @@ def _packet(
 
 
 @pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+@pytest.mark.parametrize("handoff_mode", ["soft_claim", "hard_lease"])
 def test_grouped_monitor_materialization_is_one_per_bucket_and_retires_empty_bucket(
-    tmp_path: Path, monkeypatch, provider,
+    tmp_path: Path, monkeypatch, provider, handoff_mode,
 ) -> None:
     project, state, registry = _fixture(tmp_path)
-    _use_provider(state, registry, provider, monkeypatch)
+    _use_provider(state, registry, provider, monkeypatch, handoff_mode=handoff_mode)
     ledger = tmp_path / "pr-lifecycle.jsonl"
     first = _packet(101)
     second = _packet(102)
diff --git a/tests/capabilities/test_issue_fix_monitor_execution.py b/tests/capabilities/test_issue_fix_monitor_execution.py
new file mode 100644
index 0000000000..1046a63045
--- /dev/null
+++ b/tests/capabilities/test_issue_fix_monitor_execution.py
@@ -0,0 +1,173 @@
+"""The real public reconciler must execute with its own fence on every backend."""
+import json
+
+import pytest
+
+from loopx.capabilities.issue_fix import pr_monitor_materialization as materialization
+from loopx.control_plane.work_items.task_lease import acquire_task_lease, inspect_task_lease
+from loopx.domain_packs.issue_fix import upsert_issue_fix_pr_lifecycle_ledger_jsonl
+from tests.capabilities.test_issue_fix_grouped_monitor_materialization import (
+    _fixture, _use_provider, _packet, _monitor_todos, GOAL_ID, AGENT_ID,
+)
+
+
+def scenario(tmp_path, monkeypatch, provider):
+    project, state, registry = _fixture(tmp_path)
+    _use_provider(state, registry, provider, monkeypatch, handoff_mode="hard_lease")
+    ledger = tmp_path / "lifecycle.jsonl"
+    upsert_issue_fix_pr_lifecycle_ledger_jsonl(ledger, _packet(101))
+    arguments = dict(registry_path=registry, goal_id=GOAL_ID, project=project,
+                     ledger_path=ledger, claimed_by=AGENT_ID, cadence="30m")
+    materialization.materialize_issue_fix_grouped_monitors(**arguments, generated_at="2030-01-01T00:00:00Z")
+    monitor = _monitor_todos(registry, project)[0]
+    upsert_issue_fix_pr_lifecycle_ledger_jsonl(ledger, _packet(102))
+    return arguments, monitor, state
+
+
+@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+def test_hard_lease_membership_observation_releases_only_its_execution(tmp_path, monkeypatch, provider):
+    args, before, state = scenario(tmp_path, monkeypatch, provider)
+    result = materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")
+    assert result["write_performed"] is True
+    after = _monitor_todos(args["registry_path"], args["project"])[0]
+    assert after["material_change_generation"] == int(before.get("material_change_generation") or 0) + 1
+    lease = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
+                              goal_id=GOAL_ID, todo_id=before["todo_id"])["lease"]
+    assert lease["status"] == "released"
+    snapshot = state.read_bytes()
+    assert materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")["write_performed"] is False
+    assert snapshot == state.read_bytes()
+
+
+@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+def test_another_execution_even_for_same_actor_is_not_borrowed(tmp_path, monkeypatch, provider):
+    args, monitor, state = scenario(tmp_path, monkeypatch, provider)
+    acquired = acquire_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
+        goal_id=GOAL_ID, todo_id=monitor["todo_id"], owner=AGENT_ID, idempotency_key="unrelated-execution", ttl_seconds=60)
+    before = state.read_bytes()
+    with pytest.raises(ValueError):
+        materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")
+    assert state.read_bytes() == before
+    assert inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
+        goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"] == acquired["lease"]
+
+
+@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+def test_retry_after_interrupted_acquisition_recovers_its_own_attempt(tmp_path, monkeypatch, provider):
+    args, monitor, _ = scenario(tmp_path, monkeypatch, provider)
+    original = materialization.acquire_task_lease
+    def interrupt(**kwargs):
+        original(**kwargs)
+        raise SystemExit("simulated process exit after durable acquire")
+    monkeypatch.setattr(materialization, "acquire_task_lease", interrupt)
+    with pytest.raises(SystemExit):
+        materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")
+    lease_before = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
+        goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"]
+    monkeypatch.setattr(materialization, "acquire_task_lease", original)
+    assert materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")["write_performed"]
+    lease_after = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
+        goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"]
+    assert lease_after["idempotency_key"] == lease_before["idempotency_key"]
+    assert lease_after["lease_epoch"] == lease_before["lease_epoch"]
+    assert lease_after["status"] == "released"
+
+
+@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+def test_failed_write_cleans_attempt_and_identical_retry_can_acquire_again(tmp_path, monkeypatch, provider):
+    args, monitor, _ = scenario(tmp_path, monkeypatch, provider)
+    original = materialization.update_goal_todo
+    def unavailable(**kwargs):
+        raise OSError("synthetic write failure")
+    monkeypatch.setattr(materialization, "update_goal_todo", unavailable)
+    with pytest.raises(OSError):
+        materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")
+    released = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
+        goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"]
+    assert released["status"] == "released"
+    monkeypatch.setattr(materialization, "update_goal_todo", original)
+    assert materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")["write_performed"]
+
+
+@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+def test_missing_or_stale_ledger_never_means_empty_current_work(tmp_path, monkeypatch, provider):
+    args, monitor, state = scenario(tmp_path, monkeypatch, provider)
+    before = state.read_bytes()
+    args["ledger_path"].unlink()
+    with pytest.raises(FileNotFoundError):
+        materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")
+    assert state.read_bytes() == before
+    args["ledger_path"].write_text("")
+    with pytest.raises(ValueError, match="older"):
+        materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2029-01-01T00:00:00Z")
+    assert state.read_bytes() == before
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_cli_runtime_override_keeps_reconciliation_on_selected_authority(tmp_path, monkeypatch, provider):
+    import contextlib
+    import io
+    from loopx.cli import main
+    from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+    from loopx.todos import list_goal_todos
+    from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+    project, state, registry = _fixture(tmp_path)
+    runtime = tmp_path / "selected-runtime"
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    todos = list_goal_todos(registry_path=registry, goal_id=GOAL_ID)["todos"]
+    initialize_canonical_authority(runtime, GOAL_ID,
+        build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=todos, leases=[], handoff_mode="hard_lease"),
+        state_path=state, provider=provider)
+    state.unlink()
+    metadata = tmp_path / "metadata.json"
+    metadata.write_text(json.dumps({"state": "OPEN", "reviewDecision": "REVIEW_REQUIRED",
+        "mergeStateStatus": "CLEAN", "statusCheckRollup": [{"name": "integration", "status": "IN_PROGRESS"}]}))
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        code = main(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "issue-fix", "pr-lifecycle", "--url", "https://github.com/example/repo/pull/1",
+            "--metadata-json", str(metadata), "--goal-id", GOAL_ID, "--project", str(project),
+            "--claimed-by", AGENT_ID, "--execute-transition", "--generated-at", "2030-01-01T00:00:00Z"])
+    assert code == 0
+    result = json.loads(output.getvalue())["grouped_monitor_writeback"]
+    assert result["write_performed"] is True
+    assert result["source_authority"] == f"{provider}_v0"
+    assert result["projection_delivery"] in {"delivered", "current"}
+    assert not (tmp_path / "authority").exists()
+    assert len(list_goal_todos(registry_path=registry, goal_id=GOAL_ID,
+        runtime_root_arg=str(runtime))["todos"]) == len(todos) + 1
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_real_process_exit_after_observation_recovers_cleanup_without_business_replay(tmp_path, monkeypatch, provider):
+    import subprocess
+    import sys
+    args, monitor, _ = scenario(tmp_path, monkeypatch, provider)
+    script = """
+import json, os, sys
+from pathlib import Path
+from loopx.capabilities.issue_fix import pr_monitor_materialization as m
+args=json.loads(sys.argv[1])
+for k in ('registry_path','project','ledger_path'): args[k]=Path(args[k])
+original=m.update_goal_todo
+def crash(**kwargs):
+    original(**kwargs)
+    os._exit(73)
+m.update_goal_todo=crash
+m.materialize_issue_fix_grouped_monitors(**args,generated_at='2030-01-01T01:00:00Z')
+"""
+    process = subprocess.run([sys.executable, "-c", script, json.dumps(args, default=str)],
+                             capture_output=True, text=True, timeout=45)
+    assert process.returncode == 73, process.stderr
+    before = _monitor_todos(args["registry_path"], args["project"])[0]
+    held = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
+                             goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"]
+    assert held["status"] == "active"
+    retried = materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")
+    assert retried["write_performed"] is False
+    after = _monitor_todos(args["registry_path"], args["project"])[0]
+    assert after == before
+    released = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
+                                 goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"]
+    assert released["idempotency_key"] == held["idempotency_key"]
+    assert released["status"] == "released"
diff --git a/tests/control_plane_ts/authority_store_conformance.ts b/tests/control_plane_ts/authority_store_conformance.ts
index 2456fe2b3d..24b946abde 100644
--- a/tests/control_plane_ts/authority_store_conformance.ts
+++ b/tests/control_plane_ts/authority_store_conformance.ts
@@ -1,4 +1,5 @@
 import {registerPeriodicReportConformance} from "./periodic_report_conformance.ts";
+import {registerIssueFixMonitorReconciliationConformance} from "./issue_fix_monitor_reconciliation_conformance.ts";
 import {registerTodoConsumerScopeConformance} from "./todo_consumer_scope_conformance.ts";
 import {registerProjectionConfirmationConformance} from "./projection_confirmation_conformance.ts";
 import {registerUserCompletionFollowthroughConformance} from "./user_completion_followthrough_conformance.ts";
@@ -261,6 +262,7 @@ export function registerAuthorityStoreConformance(
   registerMonitorConfigurationConformance(providerName, factory);
   registerLeasedMonitorConformance(providerName, factory);
   registerMonitorObservationUpdateConformance(providerName, factory);
+  registerIssueFixMonitorReconciliationConformance(providerName, factory);
   registerCoordinationReceiptConformance(providerName, factory);
   registerAuthoritySourceConformance(providerName, factory);
   registerHandoffModeConformance(providerName, factory);
diff --git a/tests/control_plane_ts/issue_fix_monitor_reconciliation.test.ts b/tests/control_plane_ts/issue_fix_monitor_reconciliation.test.ts
new file mode 100644
index 0000000000..2089198e6d
--- /dev/null
+++ b/tests/control_plane_ts/issue_fix_monitor_reconciliation.test.ts
@@ -0,0 +1,87 @@
+import assert from "node:assert/strict";
+import {test} from "node:test";
+import {planIssueFixMonitorReconciliation as plan, ISSUE_FIX_MONITOR_PLAN_REQUEST} from "../../loopx/control_plane/capabilities/issue_fix_monitor_reconciliation.ts";
+
+const target = "github-pr-state-example--repo-checks-pending";
+const action = "issue_fix_pr_state_checks_pending";
+const ledger = (member = "example/repo#1", fields = {}) => ({grouped_monitor_projection: {
+  materialize_nonempty_bucket_monitor: true, target_key: target, action_kind: action,
+  member_key: member, state_bucket: "checks_pending", repository: "example/repo", ...fields}});
+const todo = (fields = {}) => ({role: "agent", todo_id: "todo_monitor", target_key: target,
+  action_kind: action, task_class: "continuous_monitor", status: "open", archive_state: "active", ...fields});
+const request = (fields = {}) => ({schema_version: ISSUE_FIX_MONITOR_PLAN_REQUEST,
+  generated_at: "2030-01-01T01:00:00Z", cadence: "30m", ledger_rows: [ledger()], todos: [], ...fields});
+const steps = (fields = {}) => plan(request(fields)).steps as Record<string, any>[];
+
+test("one typed plan covers create, unchanged, membership change, complete and reactivation", () => {
+  const created = steps()[0];
+  assert.equal(created.operation, "add");
+  assert.equal(created.text.startsWith("[P"), false, "priority belongs to its explicit argument");
+  assert.equal(created.metadata.watch_only, "true");
+  const current = todo(created.metadata);
+  assert.deepEqual(steps({todos: [current]}), [{operation: "unchanged", target_key: target, todo_id: "todo_monitor"}]);
+  const changed = steps({todos: [current], ledger_rows: [ledger(), ledger("example/repo#2")]})[0];
+  assert.equal(changed.operation, "observe");
+  assert.equal(changed.observation.material_change, true);
+  assert.equal(Object.hasOwn(changed.observation, "material_change_generation"), false);
+  assert.equal(steps({todos: [current], ledger_rows: []})[0].operation, "complete");
+  assert.equal(steps({todos: [todo({...created.metadata, status: "done", completed_at: "2029-12-31T00:00:00Z"})]})[0].operation, "reactivate");
+});
+
+test("membership identity is set-based and retains the original Python digest", () => {
+  const a = plan(request({ledger_rows: [ledger(), ledger("example/repo#2"), ledger()]}));
+  const b = plan(request({ledger_rows: [ledger("example/repo#2"), ledger()]}));
+  assert.deepEqual(a, b);
+  assert.equal(a.active_member_count, 2);
+  // Independently calculated from the documented sorted compact JSON membership.
+  assert.equal((a.steps as Record<string, any>[])[0].metadata.result_hash, "7aa53a7c0b9bddf3");
+});
+
+test("archived and superseded history cannot hide a current target or be reactivated", () => {
+  const historical = todo({todo_id: "todo_history", archive_state: "archive", status: "done"});
+  assert.equal(steps({todos: [historical]})[0].operation, "add");
+  assert.deepEqual(steps({todos: [historical], ledger_rows: []}), []);
+  assert.equal(steps({todos: [todo({superseded_by: "todo_new"})]})[0].operation, "add");
+  assert.equal(steps({todos: [historical, todo()]})[0].operation, "observe");
+});
+
+test("duplicate active identities fail before a write plan can escape", () => {
+  assert.throws(() => steps({todos: [todo(), todo({todo_id: "todo_other"})]}), /ambiguous Monitor target/);
+});
+
+for (const field of ["target_key", "action_kind", "state_bucket", "repository", "member_key"]) {
+  test(`malformed active bucket ${field} cannot be mistaken for an empty group`, () => {
+    assert.throws(() => steps({ledger_rows: [ledger("m", {[field]: ""})]}));
+  });
+}
+
+test("inconsistent bucket identity is rejected, even with distinct members", () => {
+  assert.throws(() => steps({ledger_rows: [ledger(), ledger("m2", {repository: "other/repo"})]}), /conflicting/);
+});
+
+test("an older empty observation cannot retire newer membership", () => {
+  assert.throws(() => steps({ledger_rows: [], todos: [todo({last_checked_at: "2030-01-01T02:00:00Z"})]}), /older/);
+  assert.throws(() => steps({todos: [todo({last_checked_at: "2030-01-01T02:00:00Z"})]}), /older/);
+  assert.throws(() => steps({todos: [todo({status: "done", completed_at: "2030-01-01T01:00:00Z"})]}), /follow completion/);
+});
+
+test("unrelated Todo families and inactive ledger projections stay outside reconciliation", () => {
+  assert.deepEqual(steps({ledger_rows: [], todos: [todo({task_class: "advancement_task"}), todo({role: "user"}),
+    todo({action_kind: "another_monitor"})]}), []);
+  assert.deepEqual(steps({ledger_rows: [ledger("m", {materialize_nonempty_bucket_monitor: false})]}), []);
+});
+
+for (const fields of [{todos: null}, {ledger_rows: null}, {generated_at: "invalid"}, {cadence: "bad"}]) {
+  test(`invalid complete input rejects: ${JSON.stringify(fields)}`, () => assert.throws(() => steps(fields)));
+}
+
+
+test("missing or mistyped bucket declaration is unknown, not evidence of emptiness", () => {
+  assert.throws(() => steps({ledger_rows: [{}]}));
+  assert.throws(() => steps({ledger_rows: [ledger("m", {materialize_nonempty_bucket_monitor: "true"})]}));
+});
+
+test("membership digest preserves Python Unicode ordering and ASCII escapes", () => {
+  const row = steps({ledger_rows: [ledger("repo/é#1"), ledger("repo/\ue000#2"), ledger("repo/😀#3")]})[0];
+  assert.equal(row.metadata.result_hash, "7ca4c6e6d940b704");
+});
diff --git a/tests/control_plane_ts/issue_fix_monitor_reconciliation_conformance.ts b/tests/control_plane_ts/issue_fix_monitor_reconciliation_conformance.ts
new file mode 100644
index 0000000000..e41fac02f7
--- /dev/null
+++ b/tests/control_plane_ts/issue_fix_monitor_reconciliation_conformance.ts
@@ -0,0 +1,86 @@
+/** Same typed capability plan and existing native effects on every real store. */
+import assert from "node:assert/strict";
+import {test} from "node:test";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
+import {planIssueFixMonitorReconciliation, ISSUE_FIX_MONITOR_PLAN_REQUEST} from "../../loopx/control_plane/capabilities/issue_fix_monitor_reconciliation.ts";
+import {executeCoordinationTodoUpdate} from "../../loopx/control_plane/coordination/todo_update.ts";
+import {executeCanonicalTaskLeaseAcquire} from "../../loopx/control_plane/coordination/task_lease_acquire.ts";
+import {executeCanonicalTaskLeaseLifecycle} from "../../loopx/control_plane/coordination/task_lease_lifecycle.ts";
+import {executeCoordinationTodoTerminalLifecycle} from "../../loopx/control_plane/coordination/todo_terminal_lifecycle.ts";
+import {indexCoordinationProjection} from "../../loopx/control_plane/coordination/coordination_projection.ts";
+import {productionScaleGroupedMonitorFixture} from "./production_scale_coordination_fixture.ts";
+import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts";
+
+export function registerIssueFixMonitorReconciliationConformance(provider: string, factory: AuthorityStoreConformanceFactory) {
+  for (const schema of ["native", "legacy"] as const) {
+    test(`${provider}: grouped Monitor plan executes observation, stop and next cycle in the full ${schema} graph`, async t => {
+      const {store} = await factory(t);
+      const f = productionScaleGroupedMonitorFixture("grouped-monitor", schema);
+      assert.equal((await store.commitAuthority({operation_id: "seed", expected_provider_revision: null,
+        events: [], receipts: [], next_projection: f.projection})).status, "applied");
+      const loaded = async () => {
+        const result = await store.loadAuthority();
+        assert.equal(result.status, "loaded");
+        if (result.status !== "loaded") throw new Error("missing fixture authority");
+        return result;
+      };
+      const before = (await loaded()).head;
+      const ledger = [{grouped_monitor_projection: {materialize_nonempty_bucket_monitor: true,
+        target_key: f.targetKey, member_key: "example/repo#1", action_kind: "issue_fix_pr_state_checks_pending",
+        state_bucket: "checks_pending", repository: "example/repo"}}];
+      const plan = async (empty = false, now = f.now) => planIssueFixMonitorReconciliation({
+        schema_version: ISSUE_FIX_MONITOR_PLAN_REQUEST, ledger_rows: empty ? [] : ledger,
+        todos: [...indexCoordinationProjection((await loaded()).head, "grouped-monitor").todos.values()],
+        cadence: "30m", generated_at: now.toISOString()}).steps as JsonObject[];
+      const acquire = (key: string) => executeCanonicalTaskLeaseAcquire(store, {
+        goal_id: "grouped-monitor", todo_id: f.target, owner: f.actor, idempotency_key: key,
+        expected_version: null, ttl_seconds: 60, write_scopes: [], registered_agents: f.registered_agents, now: f.now});
+      const first = (await plan())[0];
+      assert.equal(first.operation, "observe");
+      const acquired = await acquire("group-observation");
+      assert.equal(acquired.status, "applied", JSON.stringify(acquired));
+      const lease = acquired.lease as JsonObject;
+      const update = {goal_id: "grouped-monitor", todo_id: f.target, expected_role: "agent", actor_agent_id: f.actor,
+        registered_agents: f.registered_agents, operation_id: "group-observation", patch: {}, clear_fields: [],
+        planning_intent: {reason: first.reason}, monitor_observation: first.observation as {generated_at: string; result_hash: string; material_change: boolean},
+        lease_idempotency_key: String(lease.idempotency_key), lease_expected_version: Number(lease.version),
+        dry_run: false, now: f.now};
+      assert.equal((await executeCoordinationTodoUpdate(store, update)).status, "applied");
+      assert.equal((await executeCoordinationTodoUpdate(store, update)).status, "replayed");
+      assert.equal((await plan())[0].operation, "unchanged");
+      assert.equal((await executeCanonicalTaskLeaseLifecycle(store, {
+        goal_id: "grouped-monitor", todo_id: f.target, operation: "release", owner: f.actor,
+        idempotency_key: String(lease.idempotency_key), expected_version: Number(lease.version), ttl_seconds: null,
+        new_owner: null, new_idempotency_key: null, registered_agents: f.registered_agents, now: f.now,
+      })).status, "applied");
+      const completionLease = await acquire("group-stop");
+      assert.equal(completionLease.status, "applied");
+      const proof = completionLease.lease as JsonObject;
+      const stop = (await plan(true))[0];
+      assert.equal(stop.operation, "complete");
+      const terminal = {goal_id: "grouped-monitor", todo_id: f.target, expected_role: "agent" as const,
+        command: "complete" as const, actor_agent_id: f.actor, registered_agents: f.registered_agents,
+        lifecycle_grants: [], authority_reason: null, decision_outcome: null, operation_id: "group-stop",
+        lease_idempotency_key: String(proof.idempotency_key), lease_expected_version: Number(proof.version),
+        allow_user_gate_auto_acquire: false, requested_no_followup: true, requested_completion_turn_key: null,
+        requested_completion_identity_source: null, linked_successor_todo_ids: [], successor_intents: [],
+        note: null, evidence: String(stop.evidence), reason: null, clear_claim: false,
+        validation_declaration: null, validation_receipt: null, completion_policy_request: null, dry_run: false, now: f.now};
+      assert.equal((await executeCoordinationTodoTerminalLifecycle(store, terminal)).status, "applied");
+      assert.equal((await executeCoordinationTodoTerminalLifecycle(store, terminal)).status, "replayed");
+      const nextTime = new Date(f.now.valueOf() + 1000);
+      const reopen = (await plan(false, nextTime))[0];
+      assert.equal(reopen.operation, "reactivate");
+      assert.equal((await executeCoordinationTodoUpdate(store, {...update, operation_id: "group-reopen",
+        lease_idempotency_key: null, lease_expected_version: null, planning_intent: {status: "open", no_followup: false},
+        monitor_observation: reopen.observation as typeof update.monitor_observation, now: nextTime})).status, "applied");
+      const after = (await loaded()).head;
+      const target = indexCoordinationProjection(after, "grouped-monitor").todos.get(f.target)!;
+      assert.equal(target.status, "open");
+      assert.equal(target.material_change_generation, 6);
+      assert.deepEqual((after.todos as JsonObject[]).filter(row => row.todo_id !== f.target),
+        (before.todos as JsonObject[]).filter(row => row.todo_id !== f.target));
+      assert.deepEqual((after.leases as JsonObject[]).filter(row => row.todo_id !== f.target), before.leases);
+    });
+  }
+}
diff --git a/tests/control_plane_ts/production_scale_coordination_fixture.ts b/tests/control_plane_ts/production_scale_coordination_fixture.ts
index 70122a88fa..f3cbc68d0f 100644
--- a/tests/control_plane_ts/production_scale_coordination_fixture.ts
+++ b/tests/control_plane_ts/production_scale_coordination_fixture.ts
@@ -631,3 +631,22 @@ export function productionScaleConsumerScopeFixture(goalId: string, schema: Auth
     [...fixture.projection.todos as Record<string, unknown>[], ...extra],
     fixture.projection.leases as Record<string, unknown>[], schema, {handoff_mode: "legacy"})};
 }
+
+/** Capability-owned bucket inside a full mixed graph, including a historical
+ * same-target row that must never shadow the current Monitor. */
+export function productionScaleGroupedMonitorFixture(goalId: string,
+  schema: AuthorityProjectionSchema = "native") {
+  const fixture = productionScaleCoordinationFixture(goalId, schema);
+  const targetKey = "github-pr-state-example--repo-checks-pending";
+  const monitor = {schema_version: "todo_domain_record_v0", todo_id: "todo_grouped_monitor",
+    role: "agent", status: "open", done: false, archive_state: "active", text: "Watch the pending PR bucket",
+    task_class: "continuous_monitor", action_kind: "issue_fix_pr_state_checks_pending", target_key: targetKey,
+    claimed_by: "agent-a", cadence: "30m", watch_only: "true", last_checked_at: "2026-09-01T00:00:00Z",
+    result_hash: "previous-membership", material_change_generation: 4, required_write_scopes: []};
+  const projection = authorityProjectionFixture(goalId, [...fixture.projection.todos as Record<string, unknown>[],
+    monitor, {...monitor, todo_id: "todo_grouped_history", status: "done", done: true, archive_state: "archive"}],
+    fixture.projection.leases as Record<string, unknown>[], schema,
+    {source_authority: "synthetic_production_scale_fixture", handoff_mode: "hard_lease"});
+  return {projection, target: monitor.todo_id, targetKey, actor: "agent-a", registered_agents: fixture.registered_agents,
+    now: new Date("2026-09-01T01:00:00Z")};
+}

From 186570880352d0766317fe7bd504aa5f43b04dfe Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 10:17:29 +0800
Subject: [PATCH 3/3] fix(issue-fix): identify a Monitor attempt by its
 subject, not its clock

A retry of an interrupted Monitor execution is a fresh invocation: it observes a
later time and may see a re-ordered ledger. The lease-attempt prefix was derived
from the observation timestamp and the raw ledger rows, so such a retry minted a
different key, refused to recognize the lease it already held, and collided with
it (`todo_lease_conflict`) until the TTL expired - the opposite of the recovery
this change set claims.

Identify the attempt by the reconciliation subject the actor executes against
(Goal, Monitor Todo, watched bucket, actor). A key minted for another subject or
by another actor still never matches, and the post-acquire typed re-plan still
validates the decision: the observation clock and the schedule it derives are
excluded from that comparison, while a changed operation, membership digest,
material-change verdict or write scope still stops the stale step. The
revalidated step is the one written, so a recovered execution records the time
it actually observed.

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../issue_fix/pr_monitor_materialization.py   | 53 ++++++++++++++--
 .../test_issue_fix_monitor_execution.py       | 63 ++++++++++++++++++-
 2 files changed, 110 insertions(+), 6 deletions(-)

diff --git a/loopx/capabilities/issue_fix/pr_monitor_materialization.py b/loopx/capabilities/issue_fix/pr_monitor_materialization.py
index 21ff1bc602..e83dd93f19 100644
--- a/loopx/capabilities/issue_fix/pr_monitor_materialization.py
+++ b/loopx/capabilities/issue_fix/pr_monitor_materialization.py
@@ -63,6 +63,47 @@ def _plan(*, registry_path: Path, goal_id: str, project: Path,
     return plan, (source.get("authority_read") or {}).get("source_authority")
 
 
+def _attempt_prefix(*, goal_id: str, todo_id: str, owner: str,
+                    target_key: str) -> str:
+    """The stable identity of one recoverable Monitor execution attempt.
+
+    A retry of the same bounded work is a new invocation: it observes a later
+    time and may see a re-ordered ledger, so neither the observation clock nor
+    the raw rows may identify the attempt. What identifies it is the
+    reconciliation subject this actor executes against - the Goal, the Monitor
+    Todo and the bucket it watches - which is what lets an execution that
+    stopped after acquiring its own lease recognize that lease again instead of
+    colliding with it. A key minted for a different subject, or by another
+    actor, never matches this prefix.
+    """
+
+    identity = json.dumps(
+        [goal_id, todo_id, owner, target_key],
+        separators=(",", ":"),
+    )
+    return "issue-fix-monitor:" + hashlib.sha256(identity.encode()).hexdigest() + ":"
+
+
+def _decision_signature(step: dict[str, Any]) -> tuple[Any, ...]:
+    """The business decision a planned step carries, without the observation clock.
+
+    ``generated_at`` and the schedule it derives are properties of one
+    observation, not of the decision: a retry that observes the same bounded
+    work later plans the same operation and must be allowed to execute it. A
+    changed operation, membership digest, material-change verdict or write scope
+    still stops the stale step, and the revalidated step is the one written, so
+    the durable observation carries the time this execution actually observed.
+    """
+
+    observation = step.get("observation")
+    if isinstance(observation, dict):
+        return (step.get("operation"), step.get("todo_id"),
+                tuple(step.get("write_scopes") or ()), observation.get("target_key"),
+                observation.get("result_hash"), observation.get("material_change"))
+    return (step.get("operation"), step.get("todo_id"),
+            tuple(step.get("write_scopes") or ()), step.get("evidence"))
+
+
 def _release_attempt(*, registry_path: Path, runtime_root: Path, goal_id: str,
                      todo_id: str, owner: str, prefix: str,
                      proof: dict[str, Any] | None = None) -> None:
@@ -94,9 +135,8 @@ def materialize_issue_fix_grouped_monitors(
     writes: list[dict[str, Any]] = []
     for step in plan["steps"]:
         operation, target = step["operation"], step["target_key"]
-        identity = json.dumps([goal_id, step.get("todo_id"), claimed_by, generated_at, cadence, rows],
-                              sort_keys=True, separators=(",", ":"))
-        prefix = "issue-fix-monitor:" + hashlib.sha256(identity.encode()).hexdigest() + ":"
+        prefix = _attempt_prefix(goal_id=goal_id, todo_id=str(step.get("todo_id") or ""),
+            owner=claimed_by, target_key=target)
         if operation == "unchanged":
             _release_attempt(registry_path=registry_path, runtime_root=runtime_root,
                 goal_id=goal_id, todo_id=step["todo_id"], owner=claimed_by, prefix=prefix)
@@ -148,8 +188,13 @@ def materialize_issue_fix_grouped_monitors(
                 if not matching or matching[0]["operation"] == "unchanged":
                     writes.append({"operation": "unchanged", "target_key": target, "write_performed": False})
                     continue
-                if matching[0] != step:
+                if _decision_signature(matching[0]) != _decision_signature(step):
                     raise ValueError("Monitor source changed before execution; retry reconciliation")
+                # Execute the decision revalidated under our own execution: the
+                # recovered step carries this invocation's observation, while the
+                # stable attempt prefix keeps this execution's lease its own.
+                step = matching[0]
+                operation = step["operation"]
             if operation in {"observe", "reactivate"}:
                 result = update_goal_todo(registry_path=registry_path, goal_id=goal_id,
                     todo_id=todo_id, role="agent", agent_id=claimed_by, project=project, runtime_root_arg=str(runtime_root),
diff --git a/tests/capabilities/test_issue_fix_monitor_execution.py b/tests/capabilities/test_issue_fix_monitor_execution.py
index 1046a63045..8102f42c5e 100644
--- a/tests/capabilities/test_issue_fix_monitor_execution.py
+++ b/tests/capabilities/test_issue_fix_monitor_execution.py
@@ -65,7 +65,34 @@ def interrupt(**kwargs):
     lease_before = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
         goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"]
     monkeypatch.setattr(materialization, "acquire_task_lease", original)
-    assert materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")["write_performed"]
+    # A real retry is a new invocation: the observation time moves on while the
+    # intended work is unchanged, so the recovery must not depend on it.
+    assert materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:01Z")["write_performed"]
+    lease_after = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
+        goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"]
+    assert lease_after["idempotency_key"] == lease_before["idempotency_key"]
+    assert lease_after["lease_epoch"] == lease_before["lease_epoch"]
+    assert lease_after["status"] == "released"
+
+
+@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+def test_retry_recovers_its_own_attempt_despite_reordered_ledger(tmp_path, monkeypatch, provider):
+    """The attempt identity is the reconciliation subject, not the row order."""
+    args, monitor, _ = scenario(tmp_path, monkeypatch, provider)
+    original = materialization.acquire_task_lease
+    def interrupt(**kwargs):
+        original(**kwargs)
+        raise SystemExit("simulated process exit after durable acquire")
+    monkeypatch.setattr(materialization, "acquire_task_lease", interrupt)
+    with pytest.raises(SystemExit):
+        materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")
+    lease_before = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
+        goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"]
+    monkeypatch.setattr(materialization, "acquire_task_lease", original)
+    lines = args["ledger_path"].read_text(encoding="utf-8").splitlines()
+    assert len(lines) > 1
+    args["ledger_path"].write_text("\n".join(reversed(lines)) + "\n", encoding="utf-8")
+    assert materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:02Z")["write_performed"]
     lease_after = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
         goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"]
     assert lease_after["idempotency_key"] == lease_before["idempotency_key"]
@@ -73,6 +100,37 @@ def interrupt(**kwargs):
     assert lease_after["status"] == "released"
 
 
+def test_attempt_prefix_identifies_the_subject_not_the_observation():
+    """Only the reconciliation subject identifies a recoverable attempt."""
+    subject = dict(goal_id=GOAL_ID, todo_id="todo_1", owner=AGENT_ID, target_key="github-pr-state-x")
+    prefix = materialization._attempt_prefix(**subject)
+    assert prefix == materialization._attempt_prefix(**subject)
+    assert prefix.startswith("issue-fix-monitor:")
+    for field, other in (("goal_id", "other-goal"), ("todo_id", "todo_2"),
+                         ("owner", "other-agent"), ("target_key", "github-pr-state-y")):
+        assert materialization._attempt_prefix(**(subject | {field: other})) != prefix
+
+
+def test_attempt_decision_ignores_only_the_observation_clock():
+    """A later observation may repeat the decision; a changed one may not."""
+    step = {"operation": "observe", "todo_id": "todo_1", "write_scopes": [],
+        "observation": {"target_key": "github-pr-state-x", "result_hash": "digest",
+            "material_change": True, "generated_at": "2030-01-01T00:00:00Z",
+            "next_due_at": "2030-01-01T00:30:00Z"}}
+    later = json.loads(json.dumps(step))
+    later["observation"].update(generated_at="2030-01-01T01:00:00Z", next_due_at="2030-01-01T01:30:00Z")
+    assert materialization._decision_signature(step) == materialization._decision_signature(later)
+    for mutate in (
+        lambda candidate: candidate.update(operation="complete"),
+        lambda candidate: candidate["observation"].update(result_hash="other-digest"),
+        lambda candidate: candidate["observation"].update(material_change=False),
+        lambda candidate: candidate.update(write_scopes=["tasks.write"]),
+    ):
+        changed = json.loads(json.dumps(step))
+        mutate(changed)
+        assert materialization._decision_signature(changed) != materialization._decision_signature(step)
+
+
 @pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
 def test_failed_write_cleans_attempt_and_identical_retry_can_acquire_again(tmp_path, monkeypatch, provider):
     args, monitor, _ = scenario(tmp_path, monkeypatch, provider)
@@ -163,7 +221,8 @@ def crash(**kwargs):
     held = inspect_task_lease(registry_path=args["registry_path"], runtime_root=tmp_path,
                              goal_id=GOAL_ID, todo_id=monitor["todo_id"])["lease"]
     assert held["status"] == "active"
-    retried = materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:00Z")
+    # The retry is a fresh invocation with its own observation time.
+    retried = materialization.materialize_issue_fix_grouped_monitors(**args, generated_at="2030-01-01T01:00:01Z")
     assert retried["write_performed"] is False
     after = _monitor_todos(args["registry_path"], args["project"])[0]
     assert after == before