From fa02b46b15e21956c5f5dcf5712f25901c0342fe Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 20 Aug 2026 02:30:41 +0800 Subject: [PATCH 1/2] fix(control-plane): align replan writeback agent scope Signed-off-by: huangruiteng --- .../work_items/semantic_replan_writeback.py | 49 ++++++++++-- .../test_refresh_state_replan_gate.py | 48 +++++++++++- .../test_replan_semantic_action_behavior.py | 77 +++++++++++++++++++ 3 files changed, 166 insertions(+), 8 deletions(-) diff --git a/loopx/control_plane/work_items/semantic_replan_writeback.py b/loopx/control_plane/work_items/semantic_replan_writeback.py index 6b8145ffd..9d35bae38 100644 --- a/loopx/control_plane/work_items/semantic_replan_writeback.py +++ b/loopx/control_plane/work_items/semantic_replan_writeback.py @@ -4,13 +4,20 @@ from typing import Any +from ...agent_registry import registered_agent_ids_for_goal +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 @@ -84,7 +91,32 @@ 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, + ) run_obligation = autonomous_replan_obligation_from_runs( newest_first_runs, agent_todos=agent_todos, @@ -112,16 +144,23 @@ 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: diff --git a/tests/control_plane/test_refresh_state_replan_gate.py b/tests/control_plane/test_refresh_state_replan_gate.py index e5f24d97f..a78f541bf 100644 --- a/tests/control_plane/test_refresh_state_replan_gate.py +++ b/tests/control_plane/test_refresh_state_replan_gate.py @@ -10,16 +10,17 @@ from loopx.control_plane.status.autonomous_replan_projection import ( AUTONOMOUS_REPLAN_PERIODIC_RUN_THRESHOLD, ) +from loopx.control_plane.work_items.semantic_replan_writeback import ( + 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 +279,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. + + +## Agent Todo + +- [ ] [P0] Continue the current agent's bounded implementation. + +""" + + def _rotated_vision_runs() -> list[dict]: return [ _open_vision_after_prior_ack(), @@ -286,6 +301,33 @@ 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_rotated_vision_obligation_rejects_first_maintenance_writeback() -> None: """#3155: a prior periodic ACK cannot hide a new vision/frontier duty.""" diff --git a/tests/control_plane/test_replan_semantic_action_behavior.py b/tests/control_plane/test_replan_semantic_action_behavior.py index 00e53c270..d6d219137 100644 --- a/tests/control_plane/test_replan_semantic_action_behavior.py +++ b/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" + " \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: From 534bdc166ba83a3e61fb603790b22fabc26eb0ea Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Fri, 21 Aug 2026 00:26:09 +0800 Subject: [PATCH 2/2] fix(control-plane): preserve completion identity through replan writeback Signed-off-by: huangruiteng --- .../work_items/semantic_replan_writeback.py | 26 ++++++++++--- .../test_refresh_state_replan_gate.py | 39 +++++++++++++++++++ 2 files changed, 60 insertions(+), 5 deletions(-) diff --git a/loopx/control_plane/work_items/semantic_replan_writeback.py b/loopx/control_plane/work_items/semantic_replan_writeback.py index 9d35bae38..f66d330e9 100644 --- a/loopx/control_plane/work_items/semantic_replan_writeback.py +++ b/loopx/control_plane/work_items/semantic_replan_writeback.py @@ -5,6 +5,7 @@ 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, @@ -26,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: @@ -46,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: @@ -58,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 ) @@ -117,6 +123,15 @@ def qualify_replan_writeback( 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, @@ -167,7 +182,8 @@ def qualify_replan_writeback( 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, ): diff --git a/tests/control_plane/test_refresh_state_replan_gate.py b/tests/control_plane/test_refresh_state_replan_gate.py index a78f541bf..cbf2b5f61 100644 --- a/tests/control_plane/test_refresh_state_replan_gate.py +++ b/tests/control_plane/test_refresh_state_replan_gate.py @@ -11,6 +11,7 @@ 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 ( @@ -328,6 +329,44 @@ def test_writeback_scopes_other_agent_user_gate_like_quota() -> None: ] +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."""