diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index 83fc2f673f..b621a77a66 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -691,6 +691,24 @@ loopx manager-inbox request --goal-id allocation --agent-id builder \ --brief-file review.json --parent-request-id ``` +To inspect an original source by its saved request identity, including a concluded +request, use the same registered Goal and receiving Agent: + +```sh +loopx manager-inbox read --goal-id allocation --agent-id builder --request-id +``` + +`--request-id` now selects that exact request instead of being silently ignored. +It retains the full current brief, original ancestry and typed receipt state; +reading a conclusion does not reopen its work or repeat its result. The read +records only that request's receipt and does not deliver unrelated peer results. +An unknown identity, another recipient or Goal lifetime fails without falling +back to a pending page. Exact reads cannot use inbox cursors. Omit the option for +the existing paged pending inbox, peer results and operation handoffs. These are +trusted local inspection commands, not new source grants or execution authority. +App/project dispatch, receiver adoption and original-conversation return retain +their separate existing entrypoints and acceptance. + In App and Goal conversations, the existing read tools expose the same bounded local observation without shell access: search `view=agents`, then read `{"view":"agent_route","goal_id":"allocation","agent_id":"reviewer"}` through diff --git a/loopx/cli_commands/manager_inbox.py b/loopx/cli_commands/manager_inbox.py index 5196e9de16..53ae1cbf4d 100644 --- a/loopx/cli_commands/manager_inbox.py +++ b/loopx/cli_commands/manager_inbox.py @@ -69,7 +69,7 @@ def register_manager_inbox(subparsers, add_format): parser.add_argument("--local-delivery-scope", choices=("all_registered", "selected")) parser.add_argument("--sender-id", help="For configuring a new delivery source: its independently verified sender.") parser.add_argument("--execute", action="store_true") - parser.add_argument("--request-id") + parser.add_argument("--request-id", help="For read: inspect this exact current or concluded request instead of the pending inbox.") parser.add_argument( "--phase", choices=("decision", "conclusion"), default="conclusion" ) @@ -237,14 +237,17 @@ def handle_manager_inbox(args, registry_path, runtime_root): ) elif args.manager_inbox_action == "read": from ..control_plane.collaboration.peers import read_inbox + request_id = getattr(args, "request_id", None) result = read_inbox(runtime_root, registry_path, args.goal_id, args.agent_id, - workspace=Path.cwd(), cursor=cursor, operation_cursor=operation_cursor) + workspace=Path.cwd(), cursor=cursor, operation_cursor=operation_cursor, + request_id=request_id) result["instruction"] += " " + collaboration_answer_instruction() - result["followthrough"] += ( - " CLI: record assessment with manager-inbox acknowledge; optionally associate existing work " - "with manager-inbox link. Return audience-ready results with manager-inbox report " - "--phase conclusion --reply-text; use --phase decision for meaningful interim news." - ) + if request_id is None: + result["followthrough"] += ( + " CLI: record assessment with manager-inbox acknowledge; optionally associate existing work " + "with manager-inbox link. Return audience-ready results with manager-inbox report " + "--phase conclusion --reply-text; use --phase decision for meaningful interim news." + ) if any(item.get("request_revision") is not None for item in result.get("items", [])): result["followthrough"] += ( " When request_revision is supplied, acknowledge with --request-revision and --request-digest, " diff --git a/loopx/control_plane/collaboration/inbox.py b/loopx/control_plane/collaboration/inbox.py index d054572a2b..47485b269e 100644 --- a/loopx/control_plane/collaboration/inbox.py +++ b/loopx/control_plane/collaboration/inbox.py @@ -145,6 +145,30 @@ def plan_request_publication(request, *, entry, operation=None): raise ValueError(str(exc)) from exc +def _inbox_item(row: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]: + """Present the existing typed receipt state without changing its lifecycle.""" + item = {**row, "inbox_state": state["kind"]} + item.pop("request_control", None) + if state.get("request_control_view") is not None: + view = state["request_control_view"] + item.update(original_brief=row["brief"], brief=view["brief"], request_control_view=view, + request_revision=view["revision"], request_revision_digest=view["revision_digest"]) + item["source_request"] = {"request_id": row["request_id"], "request_revision": view["revision"], + "request_digest": view["revision_digest"]} + item["recorded_decision"] = state["recorded_decision"] + if state["recorded_decision"] is not None: + item["receiver_decision_recorded"] = True + item["next_action"] = "Review this request's recorded decision and linked work before returning its actual result. A recorded adoption is not a completed task; preserve the current owner and do not reprioritize unrelated work." + if state["warnings"]: + item["warnings"] = state["warnings"] + item["next_action"] = "Receiver receipt readback is unavailable or conflicting; recover the original receipt before continuing this request. Do not repeat the decision, result or execution. Other requests can continue." + elif state["kind"] == "settled": + item["next_action"] = "Inspect the concluded request as historical source. Reading it does not reopen work, repeat its result or grant execution authority." + elif state.get("request_control_view") is not None: + item["next_action"] = state["request_control_view"]["receiver_instruction"] + return item + + def pending( runtime_root: Path, goal_id: str, @@ -152,8 +176,28 @@ def pending( *, cursor: str | None = None, operation_cursor: str | None = None, + request_id: str | None = None, scope: CollaborationGoalScope | None = None, ) -> dict: + if request_id is not None: + if cursor is not None or operation_cursor is not None: + raise ValueError("an exact request read cannot use inbox cursors") + try: + row = _entry(runtime_root, goal_id, agent_id, request_id, scope=scope) + except FileNotFoundError as exc: + raise ValueError("context request not found for this Goal and Agent") from exc + if row.get("schema_version") != (EXACT_ENTRY_SCHEMA if scope is not None and scope.exact else ENTRY_SCHEMA): + raise ValueError("context inbox scope mismatch") + if scope is not None: + from .goal_instance_scope import decide_collaboration_lifecycle + + lifecycle = decide_collaboration_lifecycle(scope, operation="inbox_observe", record=row) + if lifecycle.get("kind") == "omit": + raise ValueError("context request is unavailable in this Goal lifetime") + state = _pending_receipt_states(runtime_root, [row])[0] + return {"ok": True, "request_id": request_id, "items": [_inbox_item(row, state)], + "has_more": False, "next_cursor": None, + "instruction": REQUEST_TRIAGE_INSTRUCTION + " This is an exact source read, including concluded requests; it does not reopen their work."} cursor_scope = _hash( ( [ @@ -206,24 +250,7 @@ def append_batch(): ) if lifecycle.get("kind") == "omit": continue - item = {**row, "inbox_state": state["kind"]} - item.pop("request_control", None) - if state.get("request_control_view") is not None: - view = state["request_control_view"] - item.update(original_brief=row["brief"], brief=view["brief"], request_control_view=view, - request_revision=view["revision"], request_revision_digest=view["revision_digest"]) - item["source_request"] = {"request_id": row["request_id"], "request_revision": view["revision"], - "request_digest": view["revision_digest"]} - item["recorded_decision"] = state["recorded_decision"] - if state["recorded_decision"] is not None: - item["receiver_decision_recorded"] = True - item["next_action"] = "Review this request's recorded decision and linked work before returning its actual result. A recorded adoption is not a completed task; preserve the current owner and do not reprioritize unrelated work." - if state["warnings"]: - item["warnings"] = state["warnings"] - item["next_action"] = "Receiver receipt readback is unavailable or conflicting; recover the original receipt before continuing this request. Do not repeat the decision, result or execution. Other requests can continue." - elif state.get("request_control_view") is not None: - item["next_action"] = state["request_control_view"]["receiver_instruction"] - items.append(item) + items.append(_inbox_item(row, state)) if len(items) == 21: break batch.clear() diff --git a/loopx/control_plane/collaboration/peers.py b/loopx/control_plane/collaboration/peers.py index 40cb067306..03f73c6f5e 100644 --- a/loopx/control_plane/collaboration/peers.py +++ b/loopx/control_plane/collaboration/peers.py @@ -607,6 +607,7 @@ def read_inbox( workspace=None, cursor=None, operation_cursor=None, + request_id=None, caller_goal_ref=None, ): from .inbox import pending @@ -624,14 +625,18 @@ def read_inbox( agent_id, cursor=cursor, operation_cursor=operation_cursor, + request_id=request_id, scope=goal_scope, ) - peer_returns, return_deliveries = _collect_returns( - root, - goal_id, - agent_id, - scope=goal_scope, - ) + if request_id is None: + peer_returns, return_deliveries = _collect_returns( + root, + goal_id, + agent_id, + scope=goal_scope, + ) + else: + peer_returns, return_deliveries = {"items": []}, [] if peer_returns["items"]: result["peer_returns"] = peer_returns else: @@ -673,8 +678,9 @@ def read_inbox( _record_return_reads(root, return_deliveries, scope=goal_scope) from .turn_context_observation import read_delegation_turn_context - result["unfinished_delegations"] = read_delegation_turn_context( - root, registry, goal_id, agent_id)["unfinished_delegations"] + if request_id is None: + result["unfinished_delegations"] = read_delegation_turn_context( + root, registry, goal_id, agent_id)["unfinished_delegations"] result["followthrough"] = ( "Use each request's receiver_followthrough to reconcile it with actual Core work. " "Record an explicit assessment even when continuing other work; a read is not a decision. " @@ -688,6 +694,12 @@ def read_inbox( "Finish the original request with return_result, including evidence and remaining gaps. " "Adoption, file hashes and returned opinions are not independent acceptance or Todo completion." ) + if request_id is not None: + result["followthrough"] = ( + "Inspect this exact request's current source, revisions and receipts. " + "A historical read does not reopen concluded work or deliver other peer results. " + "For unfinished work, follow its existing receiver_followthrough and current authority." + ) return result diff --git a/tests/control_plane/test_inbox_exact_request.py b/tests/control_plane/test_inbox_exact_request.py new file mode 100644 index 0000000000..db99880b6f --- /dev/null +++ b/tests/control_plane/test_inbox_exact_request.py @@ -0,0 +1,148 @@ +"""Named source reads must not substitute or acknowledge another inbox page.""" + +import json +import os +import subprocess +import sys + +import pytest + +from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture +from loopx.control_plane.collaboration.inbox import _root, acknowledge +from loopx.control_plane.collaboration.peers import read_inbox, request, return_result + + +@pytest.fixture(params=["file", "sqlite"]) +def inbox(tmp_path, monkeypatch, request): + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, root, _ = promoted_create_fixture(tmp_path, provider=request.param) + data = json.loads(registry.read_text()) + data["goals"][0]["coordination"]["registered_agents"] += ["reviewer", "third"] + registry.write_text(json.dumps(data)) + return registry, root + + +def brief(purpose): + return { + "schema_version": "collaboration_brief_v0", + "purpose": purpose, + "context": "\n".join(["完整原始来源：保留约束与结论。"] * 300), + "constraints": ["Read only; do not launch work or change its owner."], + "inputs": [], + "acceptance": ["Inspect the exact original request."], + "return_requirement": "Keep the original return route.", + } + + +def create(inbox, operation, *, source="agent-a", target="reviewer"): + registry, root = inbox + content = brief(operation) + receipt = request(root, registry, "goal-a", source, target, operation, content) + return receipt["request_id"], content + + +def conclude(inbox, rid, *, agent="reviewer"): + registry, root = inbox + read_inbox(root, registry, "goal-a", agent) + acknowledge(root, "goal-a", agent, rid, "adopt", "Inspected the original request", registry=registry) + return_result(root, "goal-a", agent, rid, "The original inspection is complete.", registry=registry) + + +def cli(inbox, *extra, agent="reviewer"): + registry, root = inbox + process = subprocess.run( + [os.environ.get("LOOPX_TEST_CLI_PYTHON", sys.executable), "-I", "-m", "loopx.cli", "--registry", str(registry), + "--runtime-root", str(root), "--format", "json", "manager-inbox", "read", + "--goal-id", "goal-a", "--agent-id", agent, *extra], + cwd=registry.parent, text=True, capture_output=True, timeout=30, + ) + return process.returncode, json.loads(process.stdout) + + +def test_exact_cli_recovers_concluded_source_without_reopening_or_reading_pending(inbox): + _, root = inbox + original, content = create(inbox, "original-source") + conclude(inbox, original) + other, _ = create(inbox, "unrelated-pending") + owner = _root(root) + retained = {p: p.read_bytes() for p in owner.rglob("*.json") + if "entries" in p.relative_to(owner).parts + or p.parent.name == "decisions" or p.name == "conclusion.json"} + + code, result = cli(inbox, "--request-id", original) + + assert code == 0, result + assert [item["request_id"] for item in result["items"]] == [original] + item = result["items"][0] + assert item["brief"] == content and item["inbox_state"] == "settled" + assert "does not reopen" in item["next_action"] + assert result["has_more"] is False and result["next_cursor"] is None + assert not (owner / "reads" / f"{other}.json").exists() + assert {p: p.read_bytes() for p in retained} == retained + code, pending = cli(inbox) + assert code == 0 and [item["request_id"] for item in pending["items"]] == [other] + + +@pytest.mark.parametrize("extra,agent", [ + (["--request-id", "f" * 64], "reviewer"), + (["--request-id", "../invalid"], "reviewer"), + (["--request-id", "{original}"], "third"), + (["--request-id", "{original}", "--cursor", "invalid"], "reviewer"), + (["--request-id", "{original}", "--operation-cursor", "invalid"], "reviewer"), +]) +def test_invalid_or_ambiguous_exact_read_never_falls_back_to_pending(inbox, extra, agent): + _, root = inbox + original, _ = create(inbox, "pending-source") + before = {p: p.read_bytes() for p in _root(root).rglob("*.json")} + code, result = cli(inbox, *(value.replace("{original}", original) for value in extra), agent=agent) + assert code == 1 and result["ok"] is False, result + assert "items" not in result + assert {p: p.read_bytes() for p in _root(root).rglob("*.json")} == before + assert not (_root(root) / "reads" / f"{original}.json").exists() + + +def test_exact_read_does_not_deliver_unrelated_peer_results(inbox): + _, root = inbox + outgoing, _ = create(inbox, "awaiting-peer-result") + conclude(inbox, outgoing) + incoming, content = create(inbox, "named-incoming", source="third", target="agent-a") + before = {p: p.read_bytes() for p in _root(root).rglob("*.delivery.json")} + + code, result = cli(inbox, "--request-id", incoming, agent="agent-a") + + assert code == 0, result + assert [item["request_id"] for item in result["items"]] == [incoming] + assert result["items"][0]["brief"] == content + assert "peer_returns" not in result and "operation_handoffs" not in result + assert {p: p.read_bytes() for p in _root(root).rglob("*.delivery.json")} == before + code, pending = cli(inbox, agent="agent-a") + assert code == 0 and pending["peer_returns"]["items"][0]["request_id"] == outgoing + + +def test_exact_read_keeps_source_goal_lifetime(tmp_path, monkeypatch): + from loopx.control_plane.projects.registry_codec import source_session_registry_transaction + + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, root = tmp_path / "registry.json", tmp_path / "runtime" + with source_session_registry_transaction(registry, operation="create_fixture", create=lambda: { + "profile_id": "source_session_v1", + "goals": [{"id": "goal-a", "repo": str(tmp_path), "status": "active", + "goal_instance_id": "ginst_" + "a" * 32, + "coordination": {"registered_agents": ["agent-a", "reviewer"]}}], + "session_bindings": [], "session_receipts": [], "lifetime_receipts": [], + }) as transaction: + transaction.commit(transaction.payload_copy()) + original, content = create((registry, root), "source-lifetime") + code, observed = cli((registry, root), "--request-id", original) + assert code == 0 and observed["items"][0]["brief"] == content + assert observed["items"][0]["goal_ref"] == { + "goal_id": "goal-a", "goal_instance_id": "ginst_" + "a" * 32, + } + with source_session_registry_transaction(registry, operation="replace_fixture") as transaction: + payload = transaction.payload_copy() + payload["goals"][0]["goal_instance_id"] = "ginst_" + "b" * 32 + transaction.commit(payload) + before = {p: p.read_bytes() for p in _root(root).rglob("*.json")} + code, rejected = cli((registry, root), "--request-id", original) + assert code == 1 and rejected["ok"] is False and "items" not in rejected + assert {p: p.read_bytes() for p in _root(root).rglob("*.json")} == before diff --git a/tests/test_collaboration_goal_instance.py b/tests/test_collaboration_goal_instance.py index 7fd483c8f2..fa1c7e8210 100644 --- a/tests/test_collaboration_goal_instance.py +++ b/tests/test_collaboration_goal_instance.py @@ -115,12 +115,18 @@ def _create_source_registry(root: Path) -> Path: def _http_snapshot(root, registry, session_id): import http.client + from types import SimpleNamespace from loopx.chat_server import ChatHTTPServer, ChatRequestHandler server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) server.verbose = False server.registry_path, server.runtime_root = registry, root server.chat_store = ChatSessionStore(root) + # These fixtures script Host liveness; the real HTTP/store projection below + # tests collaboration lifetime and receipts, not exited-process recovery. + server.runtime_controller = SimpleNamespace( + recover_exited_turn=lambda _session_id: None, close=lambda: None, + ) worker = threading.Thread(target=server.serve_forever, daemon=True) worker.start() connection = http.client.HTTPConnection(*server.server_address, timeout=10)