From e1ca326fca2f9f367501605e739e897d91e5f565 Mon Sep 17 00:00:00 2001 From: BigDataDZ <76271875+BigDataDZ@users.noreply.github.com> Date: Thu, 17 Sep 2026 17:41:06 +0800 Subject: [PATCH] refactor(cli): own the promotion control commands in their own module `support_control.py` registered seven unrelated top-level commands in one 898-line module. Move `promotion-gate`, `promotion-readiness` (with its `record` subcommand), and `upgrade-plan` into `cli_commands/support_control_promotion.py` so one cohesive canary-promotion and upgrade-propagation group has a bounded owner. Registration and dispatch move together, the three commands stay inside `SUPPORT_CONTROL_COMMANDS`, and no public invocation, flag, payload, or exit code changes: `promotion-gate` and `upgrade-plan` emit byte-identical JSON before and after, and `promotion-readiness record` differs only in its `generated_at` timestamp. Refs #4639, GH-C06 Signed-off-by: BigDataDZ <76271875+BigDataDZ@users.noreply.github.com> --- loopx/cli_commands/support_control.py | 166 +------ .../cli_commands/support_control_promotion.py | 201 +++++++++ ...est_manager_handoff_m2_characterization.py | 404 ++++++++++++++++++ ...anager_handoff_m2_characterization_v0.json | 106 +++++ ...est_support_control_promotion_ownership.py | 70 +++ 5 files changed, 794 insertions(+), 153 deletions(-) create mode 100644 loopx/cli_commands/support_control_promotion.py create mode 100644 tests/control_plane/test_manager_handoff_m2_characterization.py create mode 100644 tests/fixtures/control_plane/manager_handoff_m2_characterization_v0.json create mode 100644 tests/test_support_control_promotion_ownership.py diff --git a/loopx/cli_commands/support_control.py b/loopx/cli_commands/support_control.py index df5fba80f2..7227b23c6c 100644 --- a/loopx/cli_commands/support_control.py +++ b/loopx/cli_commands/support_control.py @@ -33,12 +33,6 @@ from ..kiro_cli_goal_mode import KIRO_CLI_BIN from ..paths import default_public_scan_root from ..presentation.renderers.status_markdown import render_status_markdown -from ..promotion_gate import ( - build_promotion_gate, - record_promotion_readiness, - render_promotion_gate_markdown, - render_promotion_readiness_record_markdown, -) from ..registry import ( inspect_registry, inspect_registry_boundary, @@ -60,7 +54,6 @@ DEFAULT_STATUS_PORT, serve_status, ) -from ..upgrade import build_upgrade_plan, render_upgrade_plan_markdown from .support_control_backup import ( handle_backup_state_command, register_backup_state_command, @@ -72,6 +65,10 @@ from .support_control_heartbeat_registration import ( register_heartbeat_control_commands, ) +from .support_control_promotion import ( + handle_promotion_control_command, + register_promotion_control_commands, +) from .support_control_registry import ( explicit_global_registry, resolve_heartbeat_active_state, @@ -119,68 +116,7 @@ def register_support_control_commands( register_supervisor_control_commands(subparsers, add_subcommand_format) - promotion_gate_parser = subparsers.add_parser( - "promotion-gate", - help="Emit a compact machine-readable canary promotion readiness gate result.", - ) - add_subcommand_format(promotion_gate_parser) - - promotion_readiness_parser = subparsers.add_parser( - "promotion-readiness", - help="Record release-scoped canary promotion-readiness evidence.", - ) - promotion_readiness_subparsers = promotion_readiness_parser.add_subparsers( - dest="promotion_readiness_command", - required=True, - ) - promotion_readiness_record_parser = promotion_readiness_subparsers.add_parser( - "record", - help="Append one runtime-level readiness event after the canary checks pass.", - ) - add_subcommand_format(promotion_readiness_record_parser) - promotion_readiness_record_parser.add_argument( - "--dashboard-readiness", - choices=("passed", "skipped"), - required=True, - help="Whether dashboard readiness ran successfully or was explicitly skipped.", - ) - promotion_readiness_record_parser.add_argument( - "--execute", - action="store_true", - help="Append the evidence event. Without this flag, emit a dry-run plan.", - ) - - upgrade_plan_parser = subparsers.add_parser( - "upgrade-plan", - help="Plan local default upgrade propagation for managed heartbeat automations.", - ) - add_subcommand_format(upgrade_plan_parser) - upgrade_plan_parser.add_argument( - "--goal-id", - action="append", - default=[], - help="Only include one goal id. Repeatable.", - ) - upgrade_plan_parser.add_argument( - "--installed-manifest", - help=( - "Optional JSON manifest of installed automations with goal_id, mode, automation_id, and " - "prompt_sha256/task_body. If omitted, upgrade-plan auto-discovers Codex App heartbeat " - "automations from $CODEX_HOME/automations or ~/.codex/automations." - ), - ) - upgrade_plan_parser.add_argument( - "--cli-bin", - default="loopx", - help="CLI command embedded in generated heartbeat prompts for the promoted default.", - ) - upgrade_plan_parser.add_argument( - "--mode", - action="append", - choices=["thin", "brief", "compact"], - default=[], - help="Prompt mode to compare. Repeatable; defaults to the thin installed heartbeat contract.", - ) + register_promotion_control_commands(subparsers, add_subcommand_format) update_parser = subparsers.add_parser( "update", @@ -569,90 +505,14 @@ def handle_support_control_command( if supervisor_result is not None: return supervisor_result - if args.command == "promotion-gate": - try: - payload = build_promotion_gate( - registry_path=registry_path, - runtime_root_override=args.runtime_root, - ) - except Exception as exc: - payload = { - "ok": False, - "registry": str(registry_path), - "runtime_root": args.runtime_root, - "gate": "promotion_readiness", - "gate_state": "error", - "can_promote": False, - "should_warn": True, - "non_blocking": True, - "error": str(exc), - "recommended_action": "fix promotion readiness gate collection before promotion", - } - print_payload(payload, output_format(args), render_promotion_gate_markdown) - return 0 if payload.get("ok") else 1 - - if args.command == "promotion-readiness": - try: - payload = record_promotion_readiness( - registry_path=registry_path, - runtime_root_override=args.runtime_root, - dashboard_readiness=args.dashboard_readiness, - execute=args.execute, - ) - except Exception as exc: - payload = { - "ok": False, - "dry_run": not args.execute, - "appended": False, - "registry": str(registry_path), - "runtime_root": args.runtime_root, - "evidence_scope": "runtime_release", - "error": str(exc), - } - print_payload( - payload, - output_format(args), - render_promotion_readiness_record_markdown, - ) - return 0 if payload.get("ok") else 1 - - if args.command == "upgrade-plan": - try: - payload = build_upgrade_plan( - registry_path=registry_path, - runtime_root_override=args.runtime_root, - installed_manifest=Path(args.installed_manifest).expanduser() - if args.installed_manifest - else None, - cli_bin=args.cli_bin, - modes=args.mode or None, - goal_ids=args.goal_id or None, - ) - except Exception as exc: - payload = { - "ok": False, - "mode": "upgrade-plan", - "registry": str(registry_path), - "runtime_root": args.runtime_root, - "error": str(exc), - "summary": { - "managed_goal_count": 0, - "current_prompt_count": 0, - "stale_prompt_count": 0, - "unknown_prompt_count": 0, - "not_installed_prompt_count": 0, - "stage_deferred_goal_count": 0, - "ready_for_default_promotion": False, - "installed_manifest_available": False, - "installed_manifest_source": None, - "installed_manifest_entry_count": 0, - "installed_manifest_task_body_count": 0, - "installed_manifest_has_task_body": False, - }, - "recommended_action": "fix upgrade-plan collection before default promotion", - } - print_payload(payload, output_format(args), render_upgrade_plan_markdown) - return 0 if payload.get("ok") else 1 + promotion_result = handle_promotion_control_command( + args, + registry_path=registry_path, + print_payload=print_payload, + output_format=output_format, + ) + if promotion_result is not None: + return promotion_result if args.command == "update": update_action = UpdateAction.PLAN diff --git a/loopx/cli_commands/support_control_promotion.py b/loopx/cli_commands/support_control_promotion.py new file mode 100644 index 0000000000..ce8f522f32 --- /dev/null +++ b/loopx/cli_commands/support_control_promotion.py @@ -0,0 +1,201 @@ +"""Registration and dispatch for the promotion-gate, promotion-readiness, and +upgrade-plan commands. + +Refs GH-C06. This group was carved out of `support_control.py`, which registers +seven unrelated top-level commands in one module that sits just under the +1000-line default budget in +`examples/cli-command-module-size-ownership-command-modularization-smoke.py`. +The three commands are one group -- canary promotion readiness and the local +default upgrade plan that follows it -- so their parser flags and their +dispatch branches move together and the public invocation is unchanged. +""" + +from __future__ import annotations + +import argparse +from collections.abc import Callable +from pathlib import Path + +from ..promotion_gate import ( + build_promotion_gate, + record_promotion_readiness, + render_promotion_gate_markdown, + render_promotion_readiness_record_markdown, +) +from ..upgrade import build_upgrade_plan, render_upgrade_plan_markdown + +PrintPayload = Callable[ + [dict[str, object], str, Callable[[dict[str, object]], str]], + None, +] + +PROMOTION_CONTROL_COMMANDS = { + "promotion-gate", + "promotion-readiness", + "upgrade-plan", +} + + +def register_promotion_control_commands( + subparsers: argparse._SubParsersAction, + add_subcommand_format: Callable[[argparse.ArgumentParser], None], +) -> None: + promotion_gate_parser = subparsers.add_parser( + "promotion-gate", + help="Emit a compact machine-readable canary promotion readiness gate result.", + ) + add_subcommand_format(promotion_gate_parser) + + promotion_readiness_parser = subparsers.add_parser( + "promotion-readiness", + help="Record release-scoped canary promotion-readiness evidence.", + ) + promotion_readiness_subparsers = promotion_readiness_parser.add_subparsers( + dest="promotion_readiness_command", + required=True, + ) + promotion_readiness_record_parser = promotion_readiness_subparsers.add_parser( + "record", + help="Append one runtime-level readiness event after the canary checks pass.", + ) + add_subcommand_format(promotion_readiness_record_parser) + promotion_readiness_record_parser.add_argument( + "--dashboard-readiness", + choices=("passed", "skipped"), + required=True, + help="Whether dashboard readiness ran successfully or was explicitly skipped.", + ) + promotion_readiness_record_parser.add_argument( + "--execute", + action="store_true", + help="Append the evidence event. Without this flag, emit a dry-run plan.", + ) + + upgrade_plan_parser = subparsers.add_parser( + "upgrade-plan", + help="Plan local default upgrade propagation for managed heartbeat automations.", + ) + add_subcommand_format(upgrade_plan_parser) + upgrade_plan_parser.add_argument( + "--goal-id", + action="append", + default=[], + help="Only include one goal id. Repeatable.", + ) + upgrade_plan_parser.add_argument( + "--installed-manifest", + help=( + "Optional JSON manifest of installed automations with goal_id, mode, automation_id, and " + "prompt_sha256/task_body. If omitted, upgrade-plan auto-discovers Codex App heartbeat " + "automations from $CODEX_HOME/automations or ~/.codex/automations." + ), + ) + upgrade_plan_parser.add_argument( + "--cli-bin", + default="loopx", + help="CLI command embedded in generated heartbeat prompts for the promoted default.", + ) + upgrade_plan_parser.add_argument( + "--mode", + action="append", + choices=["thin", "brief", "compact"], + default=[], + help="Prompt mode to compare. Repeatable; defaults to the thin installed heartbeat contract.", + ) + +def handle_promotion_control_command( + args: argparse.Namespace, + *, + registry_path: Path, + print_payload: PrintPayload, + output_format: Callable[[argparse.Namespace], str], +) -> int | None: + if args.command not in PROMOTION_CONTROL_COMMANDS: + return None + + if args.command == "promotion-gate": + try: + payload = build_promotion_gate( + registry_path=registry_path, + runtime_root_override=args.runtime_root, + ) + except Exception as exc: + payload = { + "ok": False, + "registry": str(registry_path), + "runtime_root": args.runtime_root, + "gate": "promotion_readiness", + "gate_state": "error", + "can_promote": False, + "should_warn": True, + "non_blocking": True, + "error": str(exc), + "recommended_action": "fix promotion readiness gate collection before promotion", + } + print_payload(payload, output_format(args), render_promotion_gate_markdown) + return 0 if payload.get("ok") else 1 + + if args.command == "promotion-readiness": + try: + payload = record_promotion_readiness( + registry_path=registry_path, + runtime_root_override=args.runtime_root, + dashboard_readiness=args.dashboard_readiness, + execute=args.execute, + ) + except Exception as exc: + payload = { + "ok": False, + "dry_run": not args.execute, + "appended": False, + "registry": str(registry_path), + "runtime_root": args.runtime_root, + "evidence_scope": "runtime_release", + "error": str(exc), + } + print_payload( + payload, + output_format(args), + render_promotion_readiness_record_markdown, + ) + return 0 if payload.get("ok") else 1 + + if args.command == "upgrade-plan": + try: + payload = build_upgrade_plan( + registry_path=registry_path, + runtime_root_override=args.runtime_root, + installed_manifest=Path(args.installed_manifest).expanduser() + if args.installed_manifest + else None, + cli_bin=args.cli_bin, + modes=args.mode or None, + goal_ids=args.goal_id or None, + ) + except Exception as exc: + payload = { + "ok": False, + "mode": "upgrade-plan", + "registry": str(registry_path), + "runtime_root": args.runtime_root, + "error": str(exc), + "summary": { + "managed_goal_count": 0, + "current_prompt_count": 0, + "stale_prompt_count": 0, + "unknown_prompt_count": 0, + "not_installed_prompt_count": 0, + "stage_deferred_goal_count": 0, + "ready_for_default_promotion": False, + "installed_manifest_available": False, + "installed_manifest_source": None, + "installed_manifest_entry_count": 0, + "installed_manifest_task_body_count": 0, + "installed_manifest_has_task_body": False, + }, + "recommended_action": "fix upgrade-plan collection before default promotion", + } + print_payload(payload, output_format(args), render_upgrade_plan_markdown) + return 0 if payload.get("ok") else 1 + + return None diff --git a/tests/control_plane/test_manager_handoff_m2_characterization.py b/tests/control_plane/test_manager_handoff_m2_characterization.py new file mode 100644 index 0000000000..017fd881bf --- /dev/null +++ b/tests/control_plane/test_manager_handoff_m2_characterization.py @@ -0,0 +1,404 @@ +"""Characterize the #4311 peer-dispatch obligations retained for the M2 handoff contract. + +#4312 implemented a same-Goal dispatch broker and was closed as superseded by the +merged capable-manager semantic-handoff RFC (#4330). The defect in #4311 remains +real, so the acceptance obligations #4312 recorded as migration inputs are pinned +here against the shipped manager-context inbox and the shipped claim-scope +projection. This is a characterization / compatibility baseline for the M2 +cutover, not an implementation of the new collaboration contract. + +The obligation ledger lives in +``tests/fixtures/control_plane/manager_handoff_m2_characterization_v0.json``; +every obligation that claims an ``asserted_by`` test must have one. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from loopx.capabilities.manager_context import ( + _hash, + _root, + acknowledge, + authority, + deliver, + pending, +) +from loopx.capabilities.manager_context.tracking import query, record_read +from loopx.control_plane.todos.quota_selection import project_quota_planning + +FIXTURE_PATH = ( + Path(__file__).resolve().parents[1] + / "fixtures" + / "control_plane" + / "manager_handoff_m2_characterization_v0.json" +) + + +def _load() -> dict: + return json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) + + +@pytest.fixture +def ledger() -> dict: + return _load() + + +@pytest.fixture +def scenario(tmp_path, ledger): + """A synthetic Goal with a monitor owner, one peer, and a solo control Goal.""" + + data = ledger["scenario"] + registry = tmp_path / "registry.json" + registry.write_text( + json.dumps( + { + "goals": [ + { + "id": data["goal_id"], + "repo": str(tmp_path), + "coordination": { + "registered_agents": data["registered_agents"] + }, + }, + { + "id": data["solo_goal_id"], + "repo": str(tmp_path), + "coordination": { + "registered_agents": data["solo_registered_agents"] + }, + }, + ] + } + ), + encoding="utf-8", + ) + session = {"session_id": "manager-session", "channel_id": "manager"} + turn = { + "client_turn_id": "heartbeat-one", + "origin": "web", + "message": "Route the monitor successor to an agent that may execute it.", + } + return { + "root": tmp_path, + "registry": registry, + "session": session, + "turn": turn, + "data": data, + "peer_request": { + "goal_id": data["goal_id"], + "agent_id": data["peer_agent"], + }, + "origin_request": { + "goal_id": data["goal_id"], + "agent_id": data["monitor_owner"], + }, + } + + +def _planning(item: dict, agent_id: str) -> dict: + return project_quota_planning( + {}, + all_open_items=[item], + source_open_count=1, + agent_identity={"agent_id": agent_id, "agent_model": "peer_v1"}, + filter_user_gate_blocks_agent=False, + available_capabilities=None, + ) + + +def test_obligation_ledger_is_complete_public_safe_and_bound(ledger): + assert ledger["schema_version"] == "manager_handoff_m2_characterization_v0" + assert ledger["public_safe"] is True and ledger["synthetic_only"] is True + for flag in ( + "contains_credentials", + "contains_provider_payloads", + "contains_private_locators", + ): + assert ledger[flag] is False, flag + ids = [row["id"] for row in ledger["obligations"]] + assert len(ids) == len(set(ids)), "obligation ids must be unique" + for row in ledger["obligations"]: + assert row["retained_from"], row["id"] + assert row["invariant"], row["id"] + covered = row.get("asserted_by") or row.get("covered_by_existing") + assert covered, f"{row['id']} is not bound to any test" + if row.get("asserted_by"): + assert row["asserted_by"] in globals(), ( + f"{row['id']} points at missing test {row['asserted_by']}" + ) + + +def test_executor_excluded_successor_reads_dispatchable_for_a_peer_and_excluded_for_the_origin( + scenario, +): + """O1: one successor, two readings, one projection.""" + + item = dict(scenario["data"]["successor"]) + todo_id = item["todo_id"] + origin = _planning(item, scenario["data"]["monitor_owner"]) + peer = _planning(item, scenario["data"]["peer_agent"]) + + assert [row["todo_id"] for row in origin["lanes"]["open_items"]] == [] + assert origin["lanes"]["claim_scope"]["selectable_open_count"] == 0 + assert origin["lanes"]["claim_scope"]["executor_excluded_self_count"] == 1 + assert [ + row["todo_id"] + for row in origin["lanes"]["claim_scope"]["executor_excluded_self_items"] + ] == [todo_id] + + assert [row["todo_id"] for row in peer["lanes"]["open_items"]] == [todo_id] + assert peer["lanes"]["claim_scope"]["selectable_open_count"] == 1 + assert peer["lanes"]["claim_scope"]["unclaimed_open_count"] == 1 + assert peer["lanes"]["claim_scope"]["executor_excluded_self_count"] == 0 + + +def test_no_eligible_peer_fails_closed_without_inventing_a_user_gate(scenario): + """O2: an empty eligible-peer set is a typed failure, never a user gate.""" + + root, registry = scenario["root"], scenario["registry"] + session, turn = scenario["session"], scenario["turn"] + solo = scenario["data"]["solo_goal_id"] + owner = scenario["data"]["monitor_owner"] + peer = scenario["data"]["peer_agent"] + + grant = authority(root, registry, session, turn) + same_goal = [row for row in grant["targets"] if row["goal_id"] == solo] + assert same_goal == [{"goal_id": solo, "agent_id": owner}] + assert [row["agent_id"] for row in same_goal if row["agent_id"] != owner] == [] + + with pytest.raises(ValueError, match="not authorized or registered"): + deliver( + root, + registry, + session=session, + turn=turn, + request={"goal_id": solo, "agent_id": peer}, + ) + + empty = authority(root, registry, session, turn) + assert "user_gate" not in empty and "gate" not in empty + assert empty["mode"] == "context_only" + + +def test_duplicate_dispatch_replays_one_durable_entry(scenario): + """O3: a repeated heartbeat dispatch replays instead of redispatching.""" + + root, registry = scenario["root"], scenario["registry"] + request = scenario["peer_request"] + first = deliver( + root, + registry, + session=scenario["session"], + turn=scenario["turn"], + request=request, + ) + second = deliver( + root, + registry, + session=scenario["session"], + turn=scenario["turn"], + request=request, + ) + + assert second["request_id"] == first["request_id"] + assert first["replayed"] is False and second["replayed"] is True + folder = _root(root) / "entries" / _hash(request) + assert sorted(path.name for path in folder.glob("*.json")) == [ + first["request_id"] + ".json" + ] + assert len(pending(root, request["goal_id"], request["agent_id"])["items"]) == 1 + + +def test_dispatched_read_claimed_chain_survives_a_restart(scenario): + """O4: delivery, read and decision all read back from the store after restart.""" + + root, registry = scenario["root"], scenario["registry"] + request = scenario["peer_request"] + delivered = deliver( + root, + registry, + session=scenario["session"], + turn=scenario["turn"], + request=request, + ) + request_id = delivered["request_id"] + + record_read(root, pending(root, request["goal_id"], request["agent_id"])["items"]) + acknowledge( + root, + request["goal_id"], + request["agent_id"], + request_id, + "adopt", + "take the successor; the monitor transition is material", + ) + + # A restart re-reads the same runtime root with no in-process state. + row = query(root, registry, goal_ids=[request["goal_id"]], owner_scope=True)[ + "rows" + ][0] + assert row["request_id"] == request_id + assert row["delivery"]["status"] == "delivered" + assert row["read"]["status"] == "supplied_to_receiver" + assert row["decision"]["status"] == "adopt" + assert row["warnings"] == [] + + +def test_redispatch_after_decision_keeps_the_recorded_decision(scenario): + """O5: a later delivery never overwrites a recorded decision.""" + + root, registry = scenario["root"], scenario["registry"] + request = scenario["peer_request"] + delivered = deliver( + root, + registry, + session=scenario["session"], + turn=scenario["turn"], + request=request, + ) + request_id = delivered["request_id"] + acknowledge( + root, request["goal_id"], request["agent_id"], request_id, "adopt", "first" + ) + + replayed = deliver( + root, + registry, + session=scenario["session"], + turn=scenario["turn"], + request=request, + ) + assert replayed["request_id"] == request_id and replayed["replayed"] is True + + row = query(root, registry, goal_ids=[request["goal_id"]], owner_scope=True)[ + "rows" + ][0] + assert row["decision"]["status"] == "adopt" + assert row["decision"]["reason"] == "first" + + +def test_conflicting_acknowledgement_is_rejected_not_overwritten(scenario): + """O6: conflicting claim history stays visible instead of becoming latest state.""" + + root, registry = scenario["root"], scenario["registry"] + request = scenario["peer_request"] + request_id = deliver( + root, + registry, + session=scenario["session"], + turn=scenario["turn"], + request=request, + )["request_id"] + acknowledge( + root, request["goal_id"], request["agent_id"], request_id, "adopt", "first" + ) + + with pytest.raises(ValueError, match="already recorded"): + acknowledge( + root, + request["goal_id"], + request["agent_id"], + request_id, + "reject", + "second", + ) + + row = query(root, registry, goal_ids=[request["goal_id"]], owner_scope=True)[ + "rows" + ][0] + assert row["decision"]["status"] == "adopt" + + +def test_peer_dispatch_is_invisible_to_the_origin_inbox(scenario): + """O7: sharing a Goal is not sharing a handoff.""" + + root, registry = scenario["root"], scenario["registry"] + peer_request = scenario["peer_request"] + request_id = deliver( + root, + registry, + session=scenario["session"], + turn=scenario["turn"], + request=peer_request, + )["request_id"] + + assert pending(root, peer_request["goal_id"], peer_request["agent_id"])["items"] + assert ( + pending( + root, + scenario["origin_request"]["goal_id"], + scenario["data"]["monitor_owner"], + )["items"] + == [] + ) + # The entry is filed under the recipient's scope hash, so the origin identity + # cannot reach it at all; isolation is structural, not a filtered read. + assert not ( + _root(root) + / "entries" + / _hash(scenario["origin_request"]) + / (request_id + ".json") + ).exists() + with pytest.raises((OSError, ValueError)): + acknowledge( + root, + scenario["origin_request"]["goal_id"], + scenario["data"]["monitor_owner"], + request_id, + "adopt", + "wrong identity", + ) + + +def test_dispatch_and_handoff_identities_are_deterministic(scenario): + """O8: tuple-derived identities are stable, so a replay is never a redispatch.""" + + root, registry = scenario["root"], scenario["registry"] + request = scenario["peer_request"] + first = deliver( + root, + registry, + session=scenario["session"], + turn=scenario["turn"], + request=request, + )["request_id"] + assert first == _hash( + [ + authority(root, registry, scenario["session"], scenario["turn"])[ + "source_id" + ], + request, + ] + ) + assert ( + deliver( + root, + registry, + session=scenario["session"], + turn=scenario["turn"], + request=request, + )["request_id"] + == first + ) + + item = dict(scenario["data"]["successor"]) + first_note = _planning(item, scenario["data"]["monitor_owner"])["lanes"][ + "claim_scope" + ]["executor_excluded_self_items"][0]["handoff_note"]["handoff_id"] + assert ( + _planning(dict(item), scenario["data"]["monitor_owner"])["lanes"][ + "claim_scope" + ]["executor_excluded_self_items"][0]["handoff_note"]["handoff_id"] + == first_note + ) + successor = dict(item, todo_id="todo_1b7d0c4e5a92") + assert ( + _planning(successor, scenario["data"]["monitor_owner"])["lanes"]["claim_scope"][ + "executor_excluded_self_items" + ][0]["handoff_note"]["handoff_id"] + != first_note + ) diff --git a/tests/fixtures/control_plane/manager_handoff_m2_characterization_v0.json b/tests/fixtures/control_plane/manager_handoff_m2_characterization_v0.json new file mode 100644 index 0000000000..b7e40771e6 --- /dev/null +++ b/tests/fixtures/control_plane/manager_handoff_m2_characterization_v0.json @@ -0,0 +1,106 @@ +{ + "schema_version": "manager_handoff_m2_characterization_v0", + "title": "Retained peer-dispatch handoff obligations for the capable-manager semantic-handoff M2 contract", + "public_safe": true, + "synthetic_only": true, + "contains_credentials": false, + "contains_provider_payloads": false, + "contains_private_locators": false, + "sources": [ + "#4311 - dispatch executor-excluded handoffs instead of repolling origin monitor", + "#4312 closing note - obligations retained for M2 characterization / compatibility fixtures", + "#4330 RFC capable manager and semantic work handoff, section 8 legacy migration table" + ], + "why_a_fixture": [ + "#4312 is closed as superseded by the merged capable-manager semantic-handoff RFC (#4330).", + "The defect in #4311 remains real, but the accepted direction is one typed Core collaboration", + "request/assessment/result contract. The obligations below are the ones #4312 recorded as", + "migration inputs; they are characterized here against the shipped manager-context inbox so the", + "M2 cutover has a compatibility baseline instead of re-deriving behaviour from a closed branch." + ], + "scenario": { + "goal_id": "research", + "solo_goal_id": "research-solo", + "monitor_owner": "monitor-origin", + "peer_agent": "peer-reviewer", + "registered_agents": ["monitor-origin", "peer-reviewer"], + "solo_registered_agents": ["monitor-origin"], + "successor": { + "todo_id": "todo_9f3c1a7b2e04", + "text": "[P0] Independently review the monitor transition before delivery.", + "task_class": "advancement_task", + "continuation_policy": "independent_handoff", + "excluded_agents": ["monitor-origin"], + "claimed_by": null, + "role": "agent", + "required_capabilities": ["shell"] + } + }, + "obligations": [ + { + "id": "O1", + "name": "dispatchable_by_peer_vs_unclaimed_but_executor_excluded", + "retained_from": "#4311 acceptance: frontier projection distinguishes the two states", + "invariant": "One unclaimed independent_handoff successor that excludes the origin agent reads as selectable for a same-goal peer and as executor-excluded, non-selectable for the origin agent. The two readings come from the same projection, not two sources of truth.", + "asserted_by": "test_executor_excluded_successor_reads_dispatchable_for_a_peer_and_excluded_for_the_origin" + }, + { + "id": "O2", + "name": "no_eligible_peer_is_typed_and_not_a_user_gate", + "retained_from": "#4311 acceptance: typed blocker without inventing a user gate", + "invariant": "When the Goal registers no agent other than the excluded origin, the derived eligible-peer set is empty and a dispatch attempt fails closed with an authorization error. No user gate, queued placeholder or invented owner is projected.", + "asserted_by": "test_no_eligible_peer_fails_closed_without_inventing_a_user_gate" + }, + { + "id": "O3", + "name": "duplicate_ingress_does_not_redispatch", + "retained_from": "#4312: duplicate ingress and deterministic dispatch identity", + "invariant": "Repeating the same heartbeat-bound dispatch for the same recipient yields the same request id, reports replayed, and leaves exactly one durable entry. The dispatch identity is derived, never re-emitted.", + "asserted_by": "test_duplicate_dispatch_replays_one_durable_entry" + }, + { + "id": "O4", + "name": "dispatched_read_claimed_survives_restart", + "retained_from": "#4312: dispatched -> read -> claimed relation after restart", + "invariant": "After the peer reads and acknowledges, a fresh read from the same runtime root still reports the complete delivery/read/decision chain. A restart observes the recorded decision rather than re-offering the same dispatch as pending work.", + "asserted_by": "test_dispatched_read_claimed_chain_survives_a_restart" + }, + { + "id": "O5", + "name": "redispatch_after_claim_does_not_reset_the_decision", + "retained_from": "#4312: duplicate ingress reconciliation", + "invariant": "A repeated dispatch after the receiver already decided replays the same entry and leaves the recorded decision intact; the decision is never overwritten by a later delivery.", + "asserted_by": "test_redispatch_after_decision_keeps_the_recorded_decision" + }, + { + "id": "O6", + "name": "stale_claim_conflict_is_terminal", + "retained_from": "#4312: stale claim conflict", + "invariant": "A second, conflicting acknowledgement for one request is rejected instead of silently overwriting the first decision. Conflicting claim history stays observable and never becomes a fabricated latest state.", + "asserted_by": "test_conflicting_acknowledgement_is_rejected_not_overwritten" + }, + { + "id": "O7", + "name": "same_goal_agent_identity_isolation", + "retained_from": "#4311 acceptance: same-Goal agent identity isolation", + "invariant": "A dispatch addressed to one agent is invisible to another agent's inbox, and an acknowledgement under the wrong identity fails closed. Sharing a Goal is not sharing a handoff.", + "asserted_by": "test_peer_dispatch_is_invisible_to_the_origin_inbox" + }, + { + "id": "O8", + "name": "deterministic_legacy_alias", + "retained_from": "#4330 section 8: preserve the tuple-derived dispatch identity as a legacy alias, never redispatch", + "invariant": "The dispatch request id is a pure function of the source identity and the exact recipient, and the successor handoff id is a pure function of the successor tuple. Equal inputs give equal ids across independent projections; different inputs give different ids.", + "asserted_by": "test_dispatch_and_handoff_identities_are_deterministic" + }, + { + "id": "O9", + "name": "canonical_todo_claim_readback", + "retained_from": "#4312: canonical Todo claim readback before acknowledgement", + "invariant": "Linking a context request to a Core Todo requires the Todo to belong to the receiving agent, read back from the canonical Todo authority rather than from a copied projection.", + "asserted_by": null, + "covered_by_existing": "tests/test_manager_context_tracking.py::test_links_use_core_state_and_do_not_copy_progress", + "note": "Already pinned against the shipped tracking seam; re-characterizing it here would duplicate coverage." + } + ] +} diff --git a/tests/test_support_control_promotion_ownership.py b/tests/test_support_control_promotion_ownership.py new file mode 100644 index 0000000000..a5f3523a68 --- /dev/null +++ b/tests/test_support_control_promotion_ownership.py @@ -0,0 +1,70 @@ +"""Refs GH-C06: the promotion command group still belongs to support control. + +`promotion-gate`, `promotion-readiness`, and `upgrade-plan` were extracted from +`cli_commands/support_control.py` into +`cli_commands/support_control_promotion.py` so the shared support-control seam +stops owning three unrelated command groups at once. The extraction must not +change the public invocation, so these cases pin the two things that could +silently break it: the commands must still be part of the support control set, +and each must still be registered exactly once. +""" + +from __future__ import annotations + +import argparse +import re +from pathlib import Path + +from loopx.cli_commands import support_control +from loopx.cli_commands import support_control_promotion as promotion_module + +COMMANDS_DIR = Path(promotion_module.__file__).resolve().parent +ADD_PARSER_RE = re.compile( + r"subparsers\.add_parser\(\s*(?:\n\s*)?[\"'](?P[^\"']+)[\"']", + re.MULTILINE, +) + +PROMOTION_COMMANDS = ("promotion-gate", "promotion-readiness", "upgrade-plan") + + +def registered_commands() -> dict[str, list[str]]: + registrations: dict[str, list[str]] = {} + for path in sorted(COMMANDS_DIR.glob("*.py")): + for match in ADD_PARSER_RE.finditer(path.read_text(encoding="utf-8")): + registrations.setdefault(match.group("command"), []).append(path.name) + return registrations + + +def test_promotion_commands_are_still_support_control_commands() -> None: + for command in PROMOTION_COMMANDS: + assert command in support_control.SUPPORT_CONTROL_COMMANDS + + +def test_promotion_commands_are_registered_exactly_once() -> None: + registrations = registered_commands() + for command in PROMOTION_COMMANDS: + assert registrations[command] == ["support_control_promotion.py"] + + +def test_owner_module_exposes_both_halves() -> None: + """Registration and dispatch moved together, so both live in the new module.""" + assert callable(promotion_module.register_promotion_control_commands) + assert callable(promotion_module.handle_promotion_control_command) + + +def test_owner_module_owns_exactly_its_group() -> None: + assert promotion_module.PROMOTION_CONTROL_COMMANDS == set(PROMOTION_COMMANDS) + + +def test_dispatch_ignores_other_commands() -> None: + """A non-promotion command must fall through untouched, before any work.""" + args = argparse.Namespace(command="update") + assert ( + promotion_module.handle_promotion_control_command( + args, + registry_path=Path("/nonexistent-registry"), + print_payload=lambda *_args: None, + output_format=lambda *_args: "json", + ) + is None + )