diff --git a/loopx/cli_commands/manager_inbox.py b/loopx/cli_commands/manager_inbox.py index 52376111a3..2652620293 100644 --- a/loopx/cli_commands/manager_inbox.py +++ b/loopx/cli_commands/manager_inbox.py @@ -59,7 +59,7 @@ def register_manager_inbox(subparsers, add_format): parser.add_argument("--brief-file", help="For request: collaboration_brief_v0 JSON file.") parser.add_argument("--parent-request-id", help="For request: an inbox request received by the sender.") parser.add_argument("--goal-id") - parser.add_argument("--goal-instance-id", help="Exact source Goal instance for request control.") + parser.add_argument("--goal-instance-id", help="Exact source Goal instance for request control or original-conversation.") parser.add_argument("--control-operation", choices=("inspect", "revise", "cancel", "recover"), help="For control: inspect, correct, cancel or recover this original request.") parser.add_argument("--control-file", help="For control: bounded JSON operation parameters; no source or effect facts.") @@ -99,6 +99,12 @@ def handle_manager_inbox(args, registry_path, runtime_root): from ..control_plane.collaboration.peers import original_conversation_request if not args.conversation_operation or not args.conversation_file: raise ValueError("--conversation-operation and --conversation-file are required") + caller_goal_ref = None + if args.goal_instance_id is not None: + from ..control_plane.goals.goal_ref_validation import exact_goal_ref + if not args.goal_id: + raise ValueError("--goal-instance-id requires --goal-id") + caller_goal_ref = exact_goal_ref(args.goal_id, args.goal_instance_id) with Path(args.conversation_file).open("rb") as stream: raw = stream.read(128001) if len(raw) > 128000: @@ -107,7 +113,8 @@ def handle_manager_inbox(args, registry_path, runtime_root): if not isinstance(parameters, dict) or any(key in parameters for key in ("operation", "requester", "coordination_runtime_root")): raise ValueError("conversation parameters cannot assert source identity or routing") result = original_conversation_request(runtime_root, registry_path, args.goal_id, args.agent_id, - operation=args.conversation_operation, parameters=parameters, **({"chat_url": args.chat_url} if args.chat_url else {})) + operation=args.conversation_operation, parameters=parameters, caller_goal_ref=caller_goal_ref, + **({"chat_url": args.chat_url} if args.chat_url else {})) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if any(getattr(args, key, None) is not None for key in ("conversation_operation", "conversation_file", "chat_url")): diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md index 0f61442347..46eeebabbf 100644 --- a/skills/loopx-self-repair/references/repair-patterns.md +++ b/skills/loopx-self-repair/references/repair-patterns.md @@ -5,6 +5,7 @@ teaches a reusable control-plane lesson. | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair | | --- | --- | --- | --- | --- | +| `original_conversation_cli_goal_instance_drift` | `manager-inbox original-conversation` accepts an explicit `--goal-instance-id` but returns success for a stale instance, or follows the currently registered generation instead. | The physical CLI argv, selected registry profile, explicit versus omitted instance, current Goal reference, and real HTTP/source/queue/return readback in disposable File/SQLite state. | The CLI branch dropped the selector before the existing collaboration owner; current-instance inference concealed the missing caller reference. An unsupported source profile is a separate boundary, not proof that forwarding alone enables it. | Validate and forward the explicit Goal reference through the existing owner. Preserve omitted-selector compatibility and keep stale-instance, source/grant and audience checks authoritative. Cover real CLI discovery, request and control, unchanged inferred discovery, stale-instance rejection without queued work, and original recipient/result return; do not promote profiles or infer grants to make a live request pass. | | `conversation_inventory_refusal_reason_loss` | A delegation inventory stays partial because an original request is rejected as `initial_delivery_unproved`, but the App shows only a generic unreadable count. | The current authorized source, committed request/route, trusted originating Turn receipt and typed lifecycle refusal; distinguish incomplete scans and unloaded pages. | The IO adapter collapses a typed receipt conflict into a generic exception and omits the row without retaining a diagnosable identity. | Preserve the existing TS refusal and expose bounded diagnostic identities only after source/route agreement is proved and current audience/Goal checks pass. Keep the count partial, brief/work withheld and controls/return rejected; disclose the cause through existing accessible help. Other failures remain unknown. Cover File/SQLite real HTTP, source/scope changes, bounded diagnostics and fresh read after source recovery. Never rewrite live receipts, infer cancellation from another target, remove proof checks or claim the historical conflict is repaired. | | `codex_usage_session_rebinding` | Cumulative Codex token usage moves to a different session after another `session_meta`, or replaying the same snapshot loses its recorded duration after repeated metadata. | The explicitly bound synthetic rollout's ordered session ids, first clock, usage snapshot identity, and before/after run-index bytes and aggregate counters. | The usage reader overwrites identity and start time while scanning, allowing later metadata to reinterpret earlier counters. | Require identified metadata before usage, reject conflicting session ids, and retain the first clock for repeated same-id headers. Preserve streaming and the existing booking lock. Cover File/SQLite refusal without an append and repaired same-snapshot zero-delta recovery; never discover another session or rewrite historical usage. | | `implicit_clock_precedes_state_lock` | Concurrent state mutations reject a healthy request as a backward timestamp even though a fresh retry succeeds. | The exact operation, persisted `updated_at`, and a real-clock reproduction where the first caller pauses before acquiring the lock and a later caller commits first. | Implicit time was sampled before lock acquisition, so request order and commit order diverged. | Sample implicit mutation time inside the existing locked state transition. Preserve explicit backward-time rejection and observation-specific out-of-order rules. Cover every adjacent writer with reversed lock acquisition, unchanged-state negative cases, and the real CLI/file-lock path; do not clamp explicit input, weaken the guard, or add retries. | diff --git a/tests/control_plane/test_original_conversation_peer.py b/tests/control_plane/test_original_conversation_peer.py index 7654b4e90b..91b8d005b4 100644 --- a/tests/control_plane/test_original_conversation_peer.py +++ b/tests/control_plane/test_original_conversation_peer.py @@ -4,6 +4,9 @@ binding/source admission, request/revision/result stores and HTTP are real. """ import json +import subprocess +import sys +import threading from concurrent.futures import ThreadPoolExecutor import pytest @@ -80,6 +83,70 @@ def requester_agenda(f, agent="builder"): return observed["contexts"][0]["content"]["unfinished_delegations"] +def test_original_conversation_cli_keeps_the_source_goal_instance(project_peer, tmp_path): + from loopx.chat_server import ChatHTTPServer, ChatRequestHandler + + f = project_peer + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.chat_store, server.runtime_controller = f["store"], f["runtime"] + server.runtime_root, server.registry_path = f["root"], f["registry"] + server.selected_goal_id, server.verbose = None, False + worker = threading.Thread(target=server.serve_forever, daemon=True) + worker.start() + parameters_file = tmp_path / "original-conversation.json" + + def invoke(parameters, *, operation="discover", instance=INSTANCE_A): + parameters_file.write_text(json.dumps(parameters)) + command = [sys.executable, "-m", "loopx.cli", "--registry", str(f["registry"]), + "--runtime-root", str(f["root"]), "manager-inbox", "original-conversation", + "--goal-id", "delivery", "--agent-id", "builder", + "--conversation-operation", operation, "--conversation-file", str(parameters_file), + "--chat-url", f"http://127.0.0.1:{server.server_port}"] + if instance is not None: + command.extend(["--goal-instance-id", instance]) + result = subprocess.run(command, capture_output=True, text=True, timeout=30) + return result.returncode, json.loads(result.stdout) + + parent = {"parent_request_id": f["body"]["parent_request_id"]} + sid = f["session"]["session_id"] + try: + code, unavailable = invoke(parent) + assert code == 0 and unavailable["ok"], unavailable + assert unavailable["recipients"] == [] + code, inferred = invoke(parent, instance=None) + assert code == 0 and inferred["recipients"] == [] + code, refused = invoke(parent, instance="ginst_" + "b" * 32) + assert code != 0 and not refused["ok"], refused + code, forged = invoke({**parent, "requester": f["body"]["requester"]}) + assert code != 0 and not forged["ok"] + assert f["service"].management_view(sid)["requests"] == [] + assert f["store"].queued_turns(sid) == [] + + allow(f) + code, discovered = invoke(parent) + assert code == 0 and discovered["recipients"][0]["recipient"] == f["body"]["recipient"] + code, received = invoke({k: v for k, v in f["body"].items() if k != "requester"}, + operation="request") + assert code == 0 and received["recipient"]["session_id"] == sid, received + completed = f["runtime"].wait_for_turn(session_id=sid, turn_id=received["turn_id"], timeout_sec=15) + assert completed["status"] == "completed", completed + returned = read_inbox(f["root"], f["registry"], "delivery", "builder", + caller_goal_ref=f["ref"])["peer_returns"]["items"][0] + assert returned["text"] == "Verified original project result." + assert returned["decision"] == "adopt" + code, inspected = invoke({"recipient": f["body"]["recipient"], + "request_id": received["request_id"], "control": "inspect", "parameters": {}}, + operation="control") + assert code == 0 and inspected["request_control"]["revision"] == 1, inspected + assert f["store"].load_session(sid)["goal_id"] is None + packets = [json.loads(line) for line in f["capture"].read_text().splitlines()] + assert sum(packet.get("method") == "turn/start" for packet in packets) == 1 + finally: + server.shutdown() + worker.join(timeout=5) + server.server_close() + + def test_original_project_has_explicit_grant_and_returns_through_requester_inbox(project_peer): f = project_peer sid = f["session"]["session_id"]