From 98c3c5ab5f90ac555c6f4705cce3f97c47b15512 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:08:40 +0800 Subject: [PATCH 1/2] fix(chat): retain partial handoff scope in steward evidence reads Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../manager-evidence-and-continuity-v0.md | 10 ++- loopx/capabilities/manager_context/README.md | 13 +++ .../manager_context/inspection.py | 49 ++++++++++- tests/test_manager_context_roundtrip.py | 85 +++++++++++++++++++ 4 files changed, 153 insertions(+), 4 deletions(-) diff --git a/docs/reference/protocols/manager-evidence-and-continuity-v0.md b/docs/reference/protocols/manager-evidence-and-continuity-v0.md index e96a7ca543..b389bcdd2c 100644 --- a/docs/reference/protocols/manager-evidence-and-continuity-v0.md +++ b/docs/reference/protocols/manager-evidence-and-continuity-v0.md @@ -27,8 +27,14 @@ Long-running work is assigned to an exact worker with a durable return route. The initial receipt does not end the exchange: the worker evaluates or executes within its current authority and returns an audience-ready decision/result to the original conversation automatically. Acknowledgement or adoption does not -remove the request from the receiver's hook; a committed conclusion does. The -Chat runtime then owns transport recovery, separately from work completion. +remove the request from the receiver's hook. An explicit remainder on the +latest conclusion retains the original request; a queued return retains its +delivery obligation. The scoped manager handoff reader reuses the native Inbox +agenda and exposes `inbox_state` and `remaining_work` alongside transport +receipts. An older delivered conclusion cannot resolve a newer remainder or +pending reply. Unavailable receipts remain unknown. These are receipt facts, +not independent acceptance that the original question was answered. The Chat +runtime owns transport recovery, separately from work completion. An ordinary private project request interrupted by a host restart after tool work is not safe to replay. The original worker can reserve one separately diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index 31b3871e5f..9ccf9ec8a2 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -792,6 +792,19 @@ the same text and remainder: changing either under a committed result key is a conflict. Valid unmarked legacy results retain their ordinary reply state and unknown fulfillment qualification; unreadable or conflicting result chains stay unavailable. This path does not batch-close old work. + +The manager's scoped `view=handoffs` reader joins each audience-checked page to +that same native Inbox readback. `inbox_state` and `remaining_work` describe the +latest request revision; `agenda` retains its existing continuation or return +delivery obligation. A delivered `return_replies[].phase=conclusion` is only a +transport observation: it may carry a remainder, and a later reply may still be +queued. Never clear a request merely because any historical conclusion was +delivered. An unavailable/conflicting result chain retains a `recover_evidence` +agenda and marks the page unknown. A null agenda means no outstanding obligation +in these receipts, not verified business completion or a substantive answer to +every original question. The read is side-effect-free: it neither consumes a +result nor records a receiver read, schedules work, or retries a send. + Before rolling back to a reader without remainder support, stop new partial publications and reconcile the retained result chains with the compatible reader. Do not rewrite their receipts or treat an older reader's empty queue as diff --git a/loopx/capabilities/manager_context/inspection.py b/loopx/capabilities/manager_context/inspection.py index f6ba3de450..35ec177eda 100644 --- a/loopx/capabilities/manager_context/inspection.py +++ b/loopx/capabilities/manager_context/inspection.py @@ -24,6 +24,9 @@ "For a current-work overview, start with the index's bounded Agent Todo summaries; " "use view=portfolio for source and coverage, and view=todos for needed detail. " "Delivery history answers past results, not live activity. " + "For handoffs, read inbox_state, remaining_work and the existing agenda together with return_replies: " + "a delivered conclusion can be a partial result, and an older delivered result does not settle a newer pending reply. " + "These receipt observations do not verify that the original question was substantively answered. " "Every portfolio row carries its Goal lifecycle readback: reached milestones with " "their evidence refs and the phase (starting/qualifying/waiting_owner/closing/closed), " "or a typed unavailable gap naming why it could not be derived. Use that to state where " @@ -326,6 +329,47 @@ def _lifecycle_phase(readback: Any) -> str | None: return phase if isinstance(phase, str) and phase else None +def handoff_receipt_readbacks(root, registry, rows): + """Join an audience-checked page to the existing Inbox's read-only agenda. + + Transport delivery alone cannot tell whether the latest result retained + work, or whether a newer result still awaits its original return. + """ + from ...chat import redact_local_paths + from ...control_plane.collaboration.inbox import _entry, read_inbox_transaction + from ...control_plane.collaboration.goal_instance_scope import collaboration_goal_scope + + if not rows: + return [] + requests = [] + for row in rows: + with collaboration_goal_scope(registry, goal_id=row["goal_id"], agents=(), + caller_goal_ref=row.get("goal_ref")) as scope: + request = _entry(root, row["goal_id"], row["agent_id"], row["request_id"], scope=scope) + if any(request.get(key) != row.get(key) for key in ("request_id", "goal_id", "agent_id", "source_id", "goal_ref")): + raise ValueError("handoff receipt identity changed") + requests.append(request) + snapshot = read_inbox_transaction(root, { + "operation": "receipt_states", "rows": requests, "include_agenda": True, + }) + + def safe(value): + if isinstance(value, str): + return redact_local_paths(value) + return {key: safe(item) for key, item in value.items()} if isinstance(value, dict) else value + + page = [] + for row, state, agenda in zip(rows, snapshot["items"], snapshot["agenda_items"], strict=True): + if "request_revision" in row and row["request_revision"] != (state.get("readback") or {}).get("request_revision"): + raise ValueError("handoff request revision changed") + item = {**row, "inbox_state": state["kind"], "agenda": safe(agenda), + "warnings": list(dict.fromkeys([*row["warnings"], *state["warnings"]]))} + if state.get("remaining_work"): + item["remaining_work"] = safe(state["remaining_work"]) + page.append(item) + return page + + class ManagerInspection: def __init__( self, @@ -525,9 +569,9 @@ def read(self, tool: str, arguments: Any) -> dict[str, Any]: offset=offset, limit=limit, ) - except (OSError, ValueError, TypeError): + page = handoff_receipt_readbacks(self.runtime_root, self.registry_path, source.pop("rows")) + except (OSError, ValueError, TypeError, KeyError): return {"ok": False, "error": "handoff_query_unavailable_or_invalid"} - page = source.pop("rows") matched = source.pop("matched") elif view == "todos": source = read_manager_goal_details( @@ -614,6 +658,7 @@ def read(self, tool: str, arguments: Any) -> dict[str, Any]: not source["coverage"]["scan_complete"] or not source["coverage"]["legacy_audience_scan_complete"] or bool(source["coverage"]["unreadable"]) + or any(row.get("inbox_state") == "receipt_unavailable" for row in page) ) ), "oversized_rows": oversized, diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index 616a6b32d8..6a1b60c30e 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -266,6 +266,91 @@ def create(external=False, project=False, brief=None, source_store=None): return tmp_path, registry, store, create +@pytest.mark.parametrize("external", [False, True], ids=["owner", "original-audience"]) +def test_handoff_reader_retains_partial_scope_after_delivery_and_later_results(flow, external): + from loopx.capabilities.manager_context.inspection import ManagerInspection, TOOL_NAME + + root, registry, store, create = flow + session, _, receipt = create(external=external) + rid = receipt["request_id"] + acknowledge(root, "research", "worker", rid, "adopt", "Inspect original evidence") + tool = ManagerInspection(context={"goals": [{"goal_id": "research"}]}, + registry_path=registry, runtime_root=root, owner_scope=not external, + channel_id=session["channel_id"], scope_valid=lambda: True, record=lambda _: None) + + def read(): + before = {str(p): p.read_bytes() for p in _root(root).rglob("*.json")} + page = tool.read(TOOL_NAME, {"view": "handoffs", "request_id": rid}) + assert page["ok"] and not page["unknown"] and len(page["rows"]) == 1 + assert {str(p): p.read_bytes() for p in _root(root).rglob("*.json")} == before + return page["rows"][0] + + remaining = {"summary": "Measure the original revision sequence", "resume_when": "The original evidence is available"} + sender = lambda *_: {"ok": True, "reply_verified": True} + report(root, "research", "worker", rid, "conclusion", "An initial analysis is available.", remaining_work=remaining) + assert read()["remaining_work"] == remaining + assert drain(root, registry, store, sender) == 1 + partial = read() + assert partial["return_replies"][0]["status"] == "delivered" + assert partial["inbox_state"] == "awaiting_conclusion" + assert partial["agenda"]["remaining_work"] == remaining + assert partial["agenda"]["step"] == "await_peer_result" + + report(root, "research", "worker", rid, "conclusion", "Original evidence measured.", update_id="measured") + queued = read() + assert "remaining_work" not in queued + assert queued["agenda"]["step"] == "await_return_delivery" + assert drain(root, registry, store, sender) == 1 + complete = read() + assert complete["inbox_state"] == "settled" and complete["agenda"] is None + + # An old delivered conclusion must not erase a later explicit remainder. + report(root, "research", "worker", rid, "conclusion", "A qualification gap remains.", + update_id="qualification", remaining_work=remaining) + assert drain(root, registry, store, sender) == 1 + restarted = ManagerInspection(context=tool.context, registry_path=registry, runtime_root=root, + owner_scope=not external, channel_id=session["channel_id"], scope_valid=lambda: True, record=lambda _: None) + row = restarted.read(TOOL_NAME, {"view": "handoffs", "request_id": rid})["rows"][0] + assert row["remaining_work"] == remaining and row["agenda"]["receipt_state"] == "awaiting_conclusion" + scope_reads = [] + + def revoked_after_read(): + scope_reads.append(True) + return len(scope_reads) == 1 + + restarted.scope_valid = revoked_after_read + assert restarted.read(TOOL_NAME, {"view": "handoffs", "request_id": rid}) == { + "ok": False, "error": "authorization_changed"} + + +def test_handoff_reader_keeps_conflicting_result_scope_unknown_and_audience_bound(flow): + from loopx.capabilities.manager_context.inspection import ManagerInspection, TOOL_NAME + + root, registry, store, create = flow + session, _, receipt = create(external=True) + rid = receipt["request_id"] + acknowledge(root, "research", "worker", rid, "adopt", "Checked") + report(root, "research", "worker", rid, "conclusion", "Original partial result.", + remaining_work={"summary": "Validate", "resume_when": "Evidence available"}) + assert drain(root, registry, store, lambda *_: {"ok": True, "reply_verified": True}) == 1 + path = _root(root) / "replies" / rid / "conclusion.json" + saved = json.loads(path.read_text()) + saved["remaining_work"] = {"summary": "Missing continuation condition"} + path.write_text(json.dumps(saved)) + tool = ManagerInspection(context={"goals": [{"goal_id": "research"}]}, + registry_path=registry, runtime_root=root, owner_scope=False, + channel_id=session["channel_id"], scope_valid=lambda: True, record=lambda _: None) + page = tool.read(TOOL_NAME, {"view": "handoffs", "request_id": rid}) + assert page["unknown"] + row = page["rows"][0] + assert row["inbox_state"] == "receipt_unavailable" + assert row["agenda"]["step"] == "recover_evidence" + assert "conclusion_unreadable_or_conflicting" in row["warnings"] + tool.channel_id = "manager.external.another-owner" + other = tool.read(TOOL_NAME, {"view": "handoffs", "request_id": rid}) + assert other["ok"] and other["rows"] == [] + + @pytest.mark.parametrize("project", [False, True], ids=["steward", "project"]) def test_single_request_returns_to_original_transcript_without_second_model_turn(flow, project): root, registry, store, create = flow From 2ba8d250b7979fe70e421c389a4009f2ddfbac48 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:14:33 +0800 Subject: [PATCH 2/2] test(chat): qualify paging and stale handoff annotations Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../manager_context/inspection.py | 1 + tests/test_manager_context_roundtrip.py | 48 +++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/loopx/capabilities/manager_context/inspection.py b/loopx/capabilities/manager_context/inspection.py index 35ec177eda..3ce5076240 100644 --- a/loopx/capabilities/manager_context/inspection.py +++ b/loopx/capabilities/manager_context/inspection.py @@ -364,6 +364,7 @@ def safe(value): raise ValueError("handoff request revision changed") item = {**row, "inbox_state": state["kind"], "agenda": safe(agenda), "warnings": list(dict.fromkeys([*row["warnings"], *state["warnings"]]))} + item.pop("remaining_work", None) if state.get("remaining_work"): item["remaining_work"] = safe(state["remaining_work"]) page.append(item) diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index 6a1b60c30e..d2895c4bda 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -351,6 +351,54 @@ def test_handoff_reader_keeps_conflicting_result_scope_unknown_and_audience_boun assert other["ok"] and other["rows"] == [] +def test_handoff_scope_is_independent_of_page_size_and_stale_projection_annotations(flow, monkeypatch): + from loopx.capabilities.manager_context.inspection import ManagerInspection, TOOL_NAME + from loopx.capabilities.manager_context import tracking + + root, registry, store, create = flow + session, _, receipt = create() + rid = receipt["request_id"] + acknowledge(root, "research", "worker", rid, "adopt", "Checked") + remaining = {"summary": "Validate", "resume_when": "Evidence available"} + report(root, "research", "worker", rid, "conclusion", "A partial result.", remaining_work=remaining) + assert drain(root, registry, store, None) == 1 + for _ in range(12): + create() + tool = ManagerInspection(context={"goals": [{"goal_id": "research"}]}, + registry_path=registry, runtime_root=root, owner_scope=True, + channel_id=session["channel_id"], scope_valid=lambda: True, record=lambda _: None) + + def pages(limit): + result, offset = {}, 0 + while offset is not None: + page = tool.read(TOOL_NAME, {"view": "handoffs", "limit": limit, "offset": offset}) + assert page["ok"] and not page["unknown"] + result.update({row["request_id"]: row for row in page["rows"]}) + offset = page["next_offset"] + return result + + small, large = pages(3), pages(12) + assert len(small) == len(large) == 13 + assert {key: (row["inbox_state"], row["agenda"]) for key, row in small.items()} == { + key: (row["inbox_state"], row["agenda"]) for key, row in large.items()} + assert small[rid]["remaining_work"] == remaining + report(root, "research", "worker", rid, "conclusion", "Final result.", update_id="verified") + assert drain(root, registry, store, None) == 1 + query = tracking.query + + def stale_annotation(*args, **kwargs): + packet = query(*args, **kwargs) + for row in packet["rows"]: + if row["request_id"] == rid: + row.update({key: small[rid][key] for key in ("inbox_state", "agenda", "remaining_work")}) + return packet + + monkeypatch.setattr(tracking, "query", stale_annotation) + final = tool.read(TOOL_NAME, {"view": "handoffs", "request_id": rid})["rows"][0] + assert final["inbox_state"] == "settled" and final["agenda"] is None + assert "remaining_work" not in final + + @pytest.mark.parametrize("project", [False, True], ids=["steward", "project"]) def test_single_request_returns_to_original_transcript_without_second_model_turn(flow, project): root, registry, store, create = flow