fix(control-plane): align replan writeback agent scope

loopx/control_plane/work_items/semantic_replan_writeback.py

@@ -4,13 +4,21 @@
 
 from typing import Any
 
+from ...agent_registry import registered_agent_ids_for_goal
+from ..agents.agent_scope import agent_scope_item_matches_agent_or_unclaimed
+from ..agents.identity import build_quota_agent_identity
 from ..goals.goal_frontier import (
     build_goal_frontier_projection_context_from_status,
 )
 from ..status.autonomous_replan_projection import (
+    AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS,
     autonomous_replan_obligation_from_runs,
 )
 from ..todos.active_state_todo_parser import parse_active_state_todos
+from ..todos.quota_summary import (
+    select_quota_todo_source_items,
+    select_quota_todo_summary,
+)
 from ..todos.succession_warning import todo_succession_gap_items
 from .progress_observation import semantic_delta_from_writeback
 from .work_lane_context import build_work_lane_context_contract
@@ -19,7 +27,8 @@
 def _obligation_was_created_by_current_completion(
     obligation: dict[str, Any],
     *,
-    agent_todos: dict[str, Any] | None,
+    agent_todo_items: list[dict[str, Any]] | None,
+    agent_id: str,
     completion_todo_id: str | None,
     completion_turn_key: str | None,
 ) -> bool:
@@ -39,7 +48,11 @@ def _obligation_was_created_by_current_completion(
 
     safe_todo_id = str(completion_todo_id or "").strip()
     safe_turn_key = str(completion_turn_key or "").strip()
-    if not safe_todo_id or not safe_turn_key or not isinstance(agent_todos, dict):
+    if (
+        not safe_todo_id
+        or not safe_turn_key
+        or not isinstance(agent_todo_items, list)
+    ):
         return False
     triggers = obligation.get("triggers")
     if not isinstance(triggers, list) or not triggers:
@@ -51,13 +64,13 @@ def _obligation_was_created_by_current_completion(
         for trigger in triggers
     ):
         return False
-    items = agent_todos.get("items")
     return any(
         isinstance(item, dict)
+        and agent_scope_item_matches_agent_or_unclaimed(item, agent_id=agent_id)
         and str(item.get("todo_id") or "").strip() == safe_todo_id
         and item.get("status") == "done"
         and str(item.get("completion_turn_key") or "").strip() == safe_turn_key
-        for item in items or []
+        for item in agent_todo_items
     )
 
 
@@ -84,7 +97,41 @@ def qualify_replan_writeback(
     if not safe_agent_id:
         return None, None
     todo_projection = parse_active_state_todos(state_text, item_limit=None)
-    agent_todos = todo_projection.get("agent_todos")
+    registered_agent_ids = registered_agent_ids_for_goal(registry_goal)
+    agent_identity = (
+        build_quota_agent_identity(registry_goal, agent_id=safe_agent_id)
+        if registered_agent_ids
+        else {
+            "agent_id": safe_agent_id,
+            "registered_agents": [safe_agent_id],
+        }
+    )
+    raw_user_todos = todo_projection.get("user_todos")
+    raw_agent_todos = todo_projection.get("agent_todos")
+    user_todos = select_quota_todo_summary(
+        raw_user_todos,
+        None,
+        agent_identity=agent_identity,
+        filter_user_gate_blocks_agent=True,
+    )
+    agent_todos = select_quota_todo_summary(
+        raw_agent_todos,
+        None,
+        agent_identity=agent_identity,
+    )
+    agent_todo_source_items = select_quota_todo_source_items(
+        raw_agent_todos,
+        None,
+    )
+    agent_todo_completion_items = (
+        [
+            item
+            for item in raw_agent_todos.get("items", [])
+            if isinstance(item, dict)
+        ]
+        if isinstance(raw_agent_todos, dict)
+        else []
+    )
     run_obligation = autonomous_replan_obligation_from_runs(
         newest_first_runs,
         agent_todos=agent_todos,
@@ -112,23 +159,31 @@ def qualify_replan_writeback(
             )
         },
         project_asset=None,
-        user_todo_summary=todo_projection.get("user_todos"),
+        user_todo_summary=user_todos,
         agent_todo_summary=agent_todos,
+        agent_todo_source_items=agent_todo_source_items,
         work_lane_contract=build_work_lane_context_contract(
             {"progress_scope": "primary_goal"},
             agent_todo_summary=agent_todos,
         ),
-        neutral_replan_ack_classifications=set(),
-        registered_agent_ids=[safe_agent_id],
+        neutral_replan_ack_classifications=(
+            AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS
+        ),
+        registered_agent_ids=list(agent_identity["registered_agents"]),
         goal_status=str((registry_goal or {}).get("status") or "active"),
-        agent_profile=None,
+        agent_profile=(
+            agent_identity.get("agent_profile")
+            if isinstance(agent_identity.get("agent_profile"), dict)
+            else None
+        ),
     )
     obligation = context.get("replan_obligation")
     if not obligation:
         return None, None
     if _obligation_was_created_by_current_completion(
         obligation,
-        agent_todos=agent_todos,
+        agent_todo_items=agent_todo_completion_items,
+        agent_id=safe_agent_id,
         completion_todo_id=completion_todo_id,
         completion_turn_key=completion_turn_key,
     ):

tests/control_plane/test_refresh_state_replan_gate.py

@@ -10,16 +10,18 @@
 from loopx.control_plane.status.autonomous_replan_projection import (
     AUTONOMOUS_REPLAN_PERIODIC_RUN_THRESHOLD,
 )
+from loopx.control_plane.work_items.semantic_replan_writeback import (
+    _obligation_was_created_by_current_completion,
+    qualify_replan_writeback,
+)
 from loopx.state_refresh import (
     enforce_open_replan_writeback,
     refresh_state_run,
 )
-from loopx.control_plane.work_items.semantic_replan_writeback import (
-    qualify_replan_writeback,
-)
 
 GOAL_ID = "replan-gate-fixture"
 AGENT_ID = "codex-replan-gate-agent"
+OTHER_AGENT_ID = "codex-replan-gate-other-agent"
 STATE_TEXT = "# Active Goal State\n"
 
 
@@ -278,6 +280,20 @@ def _open_vision_after_prior_ack() -> dict:
     }
 
 
+def _state_with_other_agent_user_gate() -> str:
+    return f"""\
+## User Todo / Owner Review Reading Queue
+
+- [ ] [P0] Review the other agent's independent delivery.
+  <!-- loopx:todo todo_id=todo_other_agent_review status=open task_class=user_action bound_agent={OTHER_AGENT_ID} -->
+
+## Agent Todo
+
+- [ ] [P0] Continue the current agent's bounded implementation.
+  <!-- loopx:todo todo_id=todo_current_agent_slice status=open task_class=advancement_task claimed_by={AGENT_ID} -->
+"""
+
+
 def _rotated_vision_runs() -> list[dict]:
     return [
         _open_vision_after_prior_ack(),
@@ -286,6 +302,71 @@ def _rotated_vision_runs() -> list[dict]:
     ]
 
 
+def test_writeback_scopes_other_agent_user_gate_like_quota() -> None:
+    """A peer's user gate must not hide this agent's vision obligation."""
+
+    obligation, semantic_delta = qualify_replan_writeback(
+        newest_first_runs=[_open_vision_after_prior_ack()],
+        state_text=_state_with_other_agent_user_gate(),
+        agent_id=AGENT_ID,
+        goal_id=GOAL_ID,
+        registry_goal={
+            "id": GOAL_ID,
+            "status": "active",
+            "coordination": {
+                "agent_model": "peer_v1",
+                "registered_agents": [AGENT_ID, OTHER_AGENT_ID],
+            },
+        },
+    )
+
+    assert obligation is not None
+    assert semantic_delta is not None
+    assert semantic_delta["accepted"] is False
+    assert [trigger["kind"] for trigger in obligation["triggers"]] == [
+        "vision_acceptance_gap",
+        "vision_outcome_checkpoint_required",
+    ]
+
+
+def test_current_completion_source_identity_exempts_new_todo_obligation() -> None:
+    obligation, semantic_delta = qualify_replan_writeback(
+        newest_first_runs=[],
+        state_text=_completed_advancement_without_successor_state(),
+        agent_id=AGENT_ID,
+        goal_id=GOAL_ID,
+        completion_todo_id="todo_unsettled_completion",
+        completion_turn_key="turn-unsettled",
+    )
+
+    assert obligation is None
+    assert semantic_delta is None
+
+
+def test_current_completion_exemption_rejects_peer_owned_source_item() -> None:
+    assert not _obligation_was_created_by_current_completion(
+        {
+            "triggers": [
+                {
+                    "kind": "completed_advancement_without_successor",
+                    "todo_id": "todo_peer_completion",
+                }
+            ]
+        },
+        agent_todo_items=[
+            {
+                "todo_id": "todo_peer_completion",
+                "status": "done",
+                "claimed_by": OTHER_AGENT_ID,
+                "completion_turn_key": "turn-peer",
+            }
+        ],
+        agent_id=AGENT_ID,
+        completion_todo_id="todo_peer_completion",
+        completion_turn_key="turn-peer",
+    )
+
+
 def test_rotated_vision_obligation_rejects_first_maintenance_writeback() -> None:
     """#3155: a prior periodic ACK cannot hide a new vision/frontier duty."""
 

tests/control_plane/test_replan_semantic_action_behavior.py

@@ -629,6 +629,83 @@ def create_bound_successor(
     assert successor_reentry["replan_closed"] is True
 
 
+def test_successor_dry_run_shares_quota_agent_scope_for_user_gates(
+    tmp_path: Path,
+) -> None:
+    fixture = _build_fixture(tmp_path / "fixture")
+    other_agent_id = "codex-replan-other-agent"
+    registry = json.loads(fixture.global_registry_path.read_text(encoding="utf-8"))
+    registry["goals"][0]["coordination"]["registered_agents"].append(
+        other_agent_id
+    )
+    registry["goals"][0]["coordination"]["agent_profiles"][other_agent_id] = {
+        "schema_version": "agent_profile_v1",
+        "agent_id": other_agent_id,
+        "profile_role": "independent-review",
+        "scope_summary": "Review an independent bounded delivery.",
+        "default_task_classes": ["user_action"],
+        "vision_requirement": "optional",
+    }
+    registry_text = json.dumps(
+        registry,
+        ensure_ascii=False,
+        indent=2,
+        sort_keys=True,
+    ) + "\n"
+    for registry_path in (
+        fixture.global_registry_path,
+        fixture.project_root / ".loopx" / "registry.json",
+    ):
+        registry_path.write_text(registry_text, encoding="utf-8")
+    state_path = (
+        fixture.project_root
+        / ".codex"
+        / "goals"
+        / "replan-semantic-action-fixture"
+        / "ACTIVE_GOAL_STATE.md"
+    )
+    state_text = state_path.read_text(encoding="utf-8").replace(
+        "## Agent Todo\n",
+        (
+            "## User Todo / Owner Review Reading Queue\n\n"
+            "- [ ] [P0] Review the other agent's independent delivery.\n"
+            "  <!-- loopx:todo todo_id=todo_other_agent_review status=open "
+            "task_class=user_action "
+            "bound_agent=codex-replan-other-agent -->\n\n"
+            "## Agent Todo\n"
+        ),
+    )
+    state_path.write_text(state_text, encoding="utf-8")
+
+    quota = json.loads(
+        _execute_loopx(
+            fixture.quota_guard_command,
+            fixture=fixture,
+            turn_instance_id="agent-scoped-quota-turn",
+        )
+    )
+    obligation_id = quota["autonomous_replan_obligation"]["obligation_id"]
+    before = state_path.read_text(encoding="utf-8")
+    dry_run = json.loads(
+        _execute_loopx(
+            "loopx --format json --registry ignored --runtime-root ignored "
+            "todo add --goal-id replan-semantic-action-fixture "
+            "--role agent --task-class advancement_task "
+            "--action-kind inspect --target-key surface:next-bounded-slice "
+            "--text '[P0] Inspect the next bounded surface' "
+            "--claimed-by codex-replan-semantic-action "
+            f"--replan-obligation-id {obligation_id} --dry-run",
+            fixture=fixture,
+            turn_instance_id="agent-scoped-successor-turn",
+        )
+    )
+
+    assert dry_run["ok"] is True
+    assert dry_run["dry_run"] is True
+    assert dry_run["added"] is True
+    assert state_path.read_text(encoding="utf-8") == before
+
+
 def test_stale_successor_obligation_is_rejected_before_todo_mutation(
     tmp_path: Path,
 ) -> None: