fix(cli): preserve original conversation Goal instance selectors

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")):

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. |

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"]