feat(issue-fix): materialize grouped PR monitors (#2725)

* feat(issue-fix): materialize grouped PR monitors

* fix(issue-fix): keep monitor command packet compact

* test(issue-fix): compare monitor due instants

* fix(issue-fix): define grouped monitor ownership

Author: huangruiteng

docs/capabilities/issue-fix/protocols/issue-fix-workflow-contract-v0.md

@@ -125,7 +125,17 @@ open PRs, merge, publish, or run destructive git without an explicit gate.
    `continuous_monitor` for each nonempty bucket, upsert/remove PR membership
    as state changes, and complete empty buckets. Never create one monitor per
    PR. Material PR work remains a one-shot advancement todo, and reviewer
-   notifications remain one PR per message.
+   notifications remain one PR per message. `pr-lifecycle
+   --execute-transition --goal-id <goal> --claimed-by <agent>` performs this
+   reconciliation through the generic todo API; `--monitor-cadence` controls
+   the schedule and defaults to `30m`. A quiet replay with unchanged bucket
+   membership is idempotent. The monitor poll lane, rather than repeated PR
+   lifecycle execution, owns later cadence advancement. The creating issue-fix
+   agent remains the monitor's `claimed_by` owner across turns; another peer
+   cannot update, retire, reopen, or poll that monitor without explicit Todo
+   lifecycle authority.
+   With a public PR URL, `--execute-transition` fetches compact public metadata
+   automatically unless `--metadata-json` supplies a deterministic fixture.
 11. **Gate handling:** surface concrete gates instead of silently blocking. Safe
    metadata-only triage, public-code search, and focused smoke drafting may
    continue when those gates do not cover the selected action.

docs/capabilities/issue-fix/state-kernel-domain-state-case-study.zh-CN.md

@@ -226,7 +226,13 @@ metadata；CI pending 只继续 monitor，CI failure 和未处理的 changes req
 
 多个同类 PR 不需要一 PR 一 monitor。`issue_fix_pr_grouped_monitor_projection_v1` 按仓库和
 lifecycle bucket 聚合，Kernel 只调度非空 bucket 的一个 monitor；真正的 patch 仍是独立的一次性
-advancement todo。
+advancement todo。`pr-lifecycle --execute-transition --goal-id <goal> --claimed-by <agent>`
+通过通用 Todo API 完成这笔物化事务；`--monitor-cadence` 控制调度周期，缺省为 `30m`。
+相同 bucket membership 的安静重放保持幂等，后续 cadence 推进由 monitor poll lane 负责，而不是
+反复执行 PR lifecycle transition。创建该 monitor 的 issue-fix Agent 会跨 turn 保持为
+`claimed_by` owner；其他 peer 未获得显式 Todo lifecycle authority 时不能更新、retire、重开或
+poll 该 monitor。给出公开 PR URL 时，`--execute-transition` 会自动读取紧凑的
+公开 metadata；只有确定性 fixture 才需要显式传 `--metadata-json`。
 
 ### 6. Terminal closeout 与继续运行
 

examples/bootstrap-command-pack-smoke.py

@@ -231,6 +231,8 @@ def test_goal_text_invocation_plans_ranked_todos_before_activation() -> None:
         assert "--url <github-pr-url>" in pr_lifecycle_template
         assert "--goal-id" in pr_lifecycle_template
         assert str(payload["goal_id"]) in pr_lifecycle_template
+        assert "--claimed-by codex-test-agent" in pr_lifecycle_template
+        assert "--execute-transition" in pr_lifecycle_template
         assert "issue_fix_reviewer_request_template" in commands
         reviewer_request_template = str(commands["issue_fix_reviewer_request_template"])
         assert "issue-fix reviewer-request" in reviewer_request_template

examples/issue-fix-pr-lifecycle-smoke.py

@@ -196,7 +196,7 @@ def main() -> int:
     )
     assert blocked_pending["transition"]["material_change"] is False
     assert blocked_pending["grouped_monitor_projection"]["target_key"] == (
-        "github-pr-state-checks-pending"
+        "github-pr-state-huangruiteng--loopx-checks-pending"
     )
 
     stale = build_issue_fix_pr_lifecycle_monitor_packet(
@@ -239,7 +239,7 @@ def main() -> int:
         quiet["grouped_monitor_projection"]
         | {
             "state_bucket": "review_required",
-            "target_key": "github-pr-state-review-required",
+            "target_key": "github-pr-state-huangruiteng--loopx-review-required",
             "action_kind": "issue_fix_pr_state_review_required_monitor",
             "member_key": "huangruiteng/loopx#1715",
             "member_operation": "upsert",
@@ -275,7 +275,7 @@ def main() -> int:
     assert_packet_shape(approved)
     assert approved["grouped_monitor_projection"]["state_bucket"] == "ready_to_merge"
     assert approved["grouped_monitor_projection"]["target_key"] == (
-        "github-pr-state-ready-to-merge"
+        "github-pr-state-huangruiteng--loopx-ready-to-merge"
     )
     for alias in ("#1700", "issue_1700", "issues/1700", "issue 1700"):
         alias_packet = build_issue_fix_pr_lifecycle_monitor_packet(

examples/issue-fix-workflow-plan-smoke.py

@@ -124,6 +124,10 @@ def assert_workflow_shape(payload: dict[str, Any]) -> None:
     assert feasibility["writes_loopx_todo"] is False
     post_pr = payload["post_pr_lifecycle_monitor_plan"]
     assert post_pr["schema_version"] == "issue_fix_post_pr_lifecycle_monitor_plan_v1"
+    assert "--goal-id <goal-id>" in post_pr["command_preview"]
+    assert "--claimed-by <agent-id>" in post_pr["command_preview"]
+    assert "--execute-transition" in post_pr["command_preview"]
+    assert "--fetch-metadata" not in post_pr["command_preview"]
     assert post_pr["creates_per_pr_continuous_monitor_todo"] is False, post_pr
     assert post_pr["monitor_scope"] == "lifecycle_state_bucket", post_pr
     assert post_pr["materializes_nonempty_buckets_only"] is True, post_pr

loopx/bootstrap_command_pack.py

@@ -775,10 +775,6 @@ def build_loopx_bootstrap_command_pack(
     mutation_confirmation_required = bool(inspection.get("mutation_confirmation_required"))
     normalized_goal_text = " ".join(goal_text.split()) if goal_text else None
     explicit_goal_start = bool(normalized_goal_text)
-    issue_fix_commands = build_issue_fix_goal_command_templates(
-        cli_bin=cli_bin,
-        goal_id=resolved_goal_id,
-    )
     issue_fix_hint_commands = build_issue_fix_goal_command_templates(
         cli_bin=cli_bin,
         goal_id="<goal-id>",
@@ -811,6 +807,11 @@ def build_loopx_bootstrap_command_pack(
         available_capabilities=available_capabilities,
     )
     selected_agent_id = host_loop_activation.get("agent_id")
+    issue_fix_commands = build_issue_fix_goal_command_templates(
+        cli_bin=cli_bin,
+        goal_id=resolved_goal_id,
+        agent_id=str(selected_agent_id) if selected_agent_id else "<agent-id>",
+    )
     activation_allowed = bool(host_loop_activation.get("activation_allowed"))
     activation_commands = host_loop_activation.get("commands")
     activation_commands = activation_commands if isinstance(activation_commands, dict) else {}

loopx/capabilities/catalog.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 from .registry import CapabilityRegistry
+from .issue_fix.workflow_plan import build_issue_fix_pr_lifecycle_command
 
 from ..extensions.runtime import extension_catalog_entries
 
@@ -255,9 +256,14 @@
                 "write_boundary": "verified configured secondary sends plus compact receipt or stale-queue state writeback; no per-PR continuous monitor, arbitrary comment, push, merge, or publish",
             },
             {
-                "command": "loopx issue-fix pr-lifecycle --url <github-pr-url> --goal-id <goal-id> --format json",
-                "purpose": "Project public PR lifecycle state into a successor, monitor continuation, user gate, or no-follow-up transition.",
-                "write_boundary": "writes compact project-local domain state when goal or ledger context is provided; no external comment, PR creation, merge, raw logs, or body/comment capture",
+                "command": build_issue_fix_pr_lifecycle_command(
+                    cli_bin="loopx",
+                    goal_id="<goal-id>",
+                    agent_id="<agent-id>",
+                    project="<repo>",
+                ),
+                "purpose": "Project public PR lifecycle state and reconcile its grouped monitor, successor, user gate, or no-follow-up transition.",
+                "write_boundary": "reads compact public PR metadata and writes compact project-local domain state plus generic LoopX todos; no external comment, PR creation, merge, raw logs, or body/comment capture",
             },
             {
                 "command": "loopx issue-fix outcome --goal-id <goal-id> --repo <owner/repo> --issue-ref <issue-ref> --pr-ref <pr-ref> --format json",

loopx/capabilities/issue_fix/cli.py

@@ -64,6 +64,10 @@
     validate_issue_fix_pr_lifecycle_monitor_packet,
 )
 from .pr_lifecycle_rollout import append_pr_merge_rollout_event
+from .pr_monitor_materialization import (
+    DEFAULT_ISSUE_FIX_MONITOR_CADENCE,
+    materialize_issue_fix_grouped_monitors,
+)
 from .provider_hooks import IssueFixReviewerProviderHooksFactory
 from . import pr_gate_reconcile_cli
 from .reviewer_cli import (
@@ -620,8 +624,9 @@ def register_issue_fix_commands(
         "--execute-transition",
         action="store_true",
         help=(
-            "Write the correction transition into the existing LoopX todo state. "
-            "Requires --goal-id, --claimed-by, --maintainer-correction-json, and a registry."
+            "Write the lifecycle transition and reconcile grouped monitors in "
+            "the existing LoopX todo state. Requires --goal-id, --claimed-by, "
+            "and a registry."
         ),
     )
     pr_lifecycle_parser.add_argument(
@@ -632,6 +637,14 @@ def register_issue_fix_commands(
             "by the generated concrete user gate."
         ),
     )
+    pr_lifecycle_parser.add_argument(
+        "--monitor-cadence",
+        default=DEFAULT_ISSUE_FIX_MONITOR_CADENCE,
+        help=(
+            "Cadence for grouped PR lifecycle continuous monitors materialized "
+            "by --execute-transition (default: 30m)."
+        ),
+    )
     pr_gate_reconcile_cli.register_pr_gate_reconciliation_command(issue_fix_sub)
     outcome_parser = issue_fix_sub.add_parser(
         "outcome",
@@ -1231,10 +1244,20 @@ def handle_issue_fix_command(
                 raise ValueError(
                     "--fetch-metadata cannot be combined with --metadata-json"
                 )
-            if args.execute_transition and not args.maintainer_correction_json:
+            fetch_metadata = bool(
+                args.fetch_metadata
+                or (args.execute_transition and args.url and not args.metadata_json)
+            )
+            if args.execute_transition and args.no_write_domain_state:
                 raise ValueError(
-                    "--execute-transition requires --maintainer-correction-json"
+                    "--execute-transition cannot be combined with --no-write-domain-state"
                 )
+            if args.execute_transition and registry_path is None:
+                raise ValueError("--execute-transition requires a LoopX registry")
+            if args.execute_transition and not args.goal_id:
+                raise ValueError("--execute-transition requires --goal-id")
+            if args.execute_transition and not args.claimed_by:
+                raise ValueError("--execute-transition requires --claimed-by")
             payload = build_issue_fix_pr_lifecycle_monitor_packet(
                 repo=args.repo,
                 pr_ref=args.pr_ref,
@@ -1243,7 +1266,7 @@ def handle_issue_fix_command(
                 provider_payload=_load_json_object(args.metadata_json)
                 if args.metadata_json
                 else None,
-                fetch_metadata=args.fetch_metadata,
+                fetch_metadata=fetch_metadata,
                 fetch_timeout_seconds=args.fetch_timeout_seconds,
                 maintainer_correction_input=(
                     _load_json_object(args.maintainer_correction_json)
@@ -1288,14 +1311,23 @@ def handle_issue_fix_command(
                     )
             transition = payload.get("transition")
             if args.execute_transition:
-                if registry_path is None:
-                    raise ValueError("--execute-transition requires a LoopX registry")
-                if not args.goal_id:
-                    raise ValueError("--execute-transition requires --goal-id")
-                if not args.claimed_by:
-                    raise ValueError("--execute-transition requires --claimed-by")
                 if not isinstance(transition, dict):
                     raise ValueError("PR lifecycle transition is missing")
+                grouped_monitor_writeback = materialize_issue_fix_grouped_monitors(
+                    registry_path=registry_path,
+                    goal_id=args.goal_id,
+                    project=Path(args.project).expanduser(),
+                    ledger_path=ledger_path,
+                    claimed_by=args.claimed_by,
+                    cadence=args.monitor_cadence,
+                    generated_at=str(payload.get("generated_at") or generated_at),
+                )
+                payload["grouped_monitor_writeback"] = grouped_monitor_writeback
+                grouped_monitor_projection = payload.get("grouped_monitor_projection")
+                if isinstance(grouped_monitor_projection, dict):
+                    grouped_monitor_projection["todo_write_performed"] = bool(
+                        grouped_monitor_writeback.get("write_performed")
+                    )
                 decision = str(transition.get("decision") or "")
                 if decision in {"runnable_successor", "user_gate"}:
                     role = str(transition.get("role") or "agent")
@@ -1348,6 +1380,26 @@ def handle_issue_fix_command(
                         ),
                         "path_recorded": False,
                     }
+                grouped_write_performed = bool(
+                    grouped_monitor_writeback.get("write_performed")
+                )
+                transition_write_performed = bool(payload.get("todo_write_performed"))
+                payload["todo_write_performed"] = bool(
+                    grouped_write_performed or transition_write_performed
+                )
+                if grouped_write_performed and not transition_write_performed:
+                    payload["todo_write"] = {
+                        "schema_version": "issue_fix_pr_lifecycle_todo_write_v1",
+                        "write_performed": True,
+                        "grouped_monitor_write_performed": True,
+                        "transition_write_performed": False,
+                        "path_recorded": False,
+                    }
+                writeback_contract = payload.get("writeback_contract")
+                if isinstance(writeback_contract, dict):
+                    writeback_contract["todo_write_performed"] = bool(
+                        payload["todo_write_performed"]
+                    )
                 validation = validate_issue_fix_pr_lifecycle_monitor_packet(payload)
                 payload["validation"] = validation
                 payload["ok"] = bool(validation["ok"])

loopx/capabilities/issue_fix/pr_lifecycle.py

@@ -570,6 +570,19 @@ def _observation_fingerprint(observation: Mapping[str, Any]) -> str:
     return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]
 
 
+def issue_fix_grouped_monitor_identity(
+    *, repository: str, state_bucket: str
+) -> tuple[str, str]:
+    """Return stable repository-scoped monitor target and action identities."""
+
+    repository_key = repository.casefold().replace("/", "--")
+    bucket_key = state_bucket.replace("_", "-")
+    return (
+        f"github-pr-state-{repository_key}-{bucket_key}",
+        f"issue_fix_pr_state_{state_bucket}_monitor",
+    )
+
+
 def _grouped_monitor_projection(
     observation: Mapping[str, Any], transition: Mapping[str, Any]
 ) -> dict[str, Any]:
@@ -608,17 +621,18 @@ def _grouped_monitor_projection(
 
     terminal = state_bucket == "terminal"
     member_key = f"{observation.get('repo')}#{observation.get('number')}"
-    target_key = (
-        None if terminal else f"github-pr-state-{state_bucket.replace('_', '-')}"
+    repository = str(observation.get("repo") or "unknown")
+    target_key, action_kind = issue_fix_grouped_monitor_identity(
+        repository=repository,
+        state_bucket=state_bucket,
     )
     return {
         "schema_version": ISSUE_FIX_PR_GROUPED_MONITOR_PROJECTION_SCHEMA_VERSION,
         "scope": "repository_pr_lifecycle_state",
+        "repository": repository,
         "state_bucket": state_bucket,
-        "target_key": target_key,
-        "action_kind": (
-            None if terminal else f"issue_fix_pr_state_{state_bucket}_monitor"
-        ),
+        "target_key": None if terminal else target_key,
+        "action_kind": None if terminal else action_kind,
         "member_key": member_key,
         "member_operation": "remove" if terminal else "upsert",
         "materialize_nonempty_bucket_monitor": not terminal,
@@ -988,7 +1002,9 @@ def build_issue_fix_pr_lifecycle_monitor_packet(
         "first_screen": {
             "waiting_on": transition["role"],
             "user_action_required": transition["role"] == "user",
-            "agent_can_continue": transition["decision"] == "runnable_successor",
+            "agent_can_continue": transition["decision"] != "user_gate",
+            "current_pr_actionable": transition["decision"] == "runnable_successor",
+            "immediate_repoll_required": False,
             "next_safe_action": transition["reason"],
         },
         "writeback_contract": {