fix(chat): retain partial handoff scope in steward evidence reads

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

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

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,48 @@ 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"]]))}
+        item.pop("remaining_work", None)
+        if state.get("remaining_work"):
+            item["remaining_work"] = safe(state["remaining_work"])
+        page.append(item)
+    return page
+
+
 class ManagerInspection:
     def __init__(
         self,
@@ -525,9 +570,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 +659,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,

tests/test_manager_context_roundtrip.py

@@ -266,6 +266,139 @@ 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"] == []
+
+
+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