fix(collaboration): surface unfinished delegation requests every Turn

loopx/capabilities/manager_context/README.md

@@ -137,8 +137,33 @@ Messages outside the shared inbox's source-context bounds remain intact in
 ordinary Chat, but cannot be handed off as inline context; use a scoped artifact
 for larger material. Image content is not transferred by this text ingress record.
 
-The existing worker turn-start hook exposes only a bounded pending count and
-required read command, without copying private content into status projections.
+The worker turn-start hook supplies a compact **unfinished delegation agenda** in
+its private `work_context.sources[].content.unfinished_delegations` on every
+managed Turn. It includes incoming requests after read/adoption, sent peer work
+awaiting a result, conclusions awaiting original-route delivery, and peer results
+awaiting explicit requester consumption. The same read model is returned by
+`manager-inbox read`. Discovery/status hook receipts remain content-free; the
+private agenda contains current-purpose excerpts and explicit links, not source
+transcripts, result bodies or receiver rationale.
+
+Each item names its request, receiving/source Agent where recorded, current
+revision, related Todo ids and the remaining receipt step. Counts are request
+records (`count_unit=request`), not distinct tasks: follow-up requests can link
+to the same existing Todo. A completed Todo is
+not a returned answer; a queued conclusion is not delivered; delivery is not
+requester adoption or independent acceptance. A current cancellation is omitted
+only after its active publication reservation is gone. Failed reads and scan or
+prompt overflow remain explicit rather than becoming an empty queue.
+
+The automatic agenda is a partial read (`read_complete=false`): it cannot fulfill
+the full Inbox read or replace its requirements. Before acting, read the current
+brief and independently assess its authority, evidence and constraints. The
+agenda never changes priority, wakes an Agent, claims work, acknowledges a
+request, publishes a result or grants effects. It is collected by the existing
+LoopX quota/heartbeat Turn-start path; arbitrary host Turns that do not invoke
+that path are outside this integration. Frontend/Lark conversations retain their
+existing request and return views; no new screen or setting is introduced.
+
 Read and record a decision through the installed CLI:
 
 ```sh
@@ -436,8 +461,11 @@ ordered by request id; restart without a cursor to find new requests sorted
 before the last position. Peer results retain their separate consumption flow
 and are not paginated by this cursor. `status --offset/--limit` remains separate.
 
-After `acknowledge`, the request remains in the turn-start hook until the worker
-publishes a conclusion. The worker uses `link` for canonical Todo/evidence lineage
+After `acknowledge`, the request remains visible until the answer obligation is
+resolved. `pending.items` retains its existing receipt-only semantics: it stops
+including a request once its conclusion exists. The separate unfinished agenda
+keeps undelivered conclusions visible, and keeps peer work until the requester
+explicitly consumes the current result. The worker uses `link` for canonical Todo/evidence lineage
 and `report` to publish the answer intended for the original audience:
 
 ```sh

loopx/capabilities/manager_context/__init__.py

@@ -216,8 +216,16 @@ def turn_start_hook(
         TURN_START_HOOK_RESULT_SCHEMA_VERSION,
     )
 
+    snapshot = None
+
     def produce():
+        nonlocal snapshot
         try:
+            from ...control_plane.collaboration.turn_context_observation import read_delegation_turn_context
+
+            snapshot = read_delegation_turn_context(runtime_root, registry_path, goal_id, agent_id,
+                read_command=command)
+            agenda = snapshot["unfinished_delegations"]
             with collaboration_goal_scope(
                 registry_path,
                 goal_id=goal_id,
@@ -230,10 +238,12 @@ def produce():
                     scope=goal_scope,
                 )
             count = (
-                len(inbox["items"])
-                + len(inbox.get("peer_returns", {}).get("items", []))
+                max(len(inbox["items"]), agenda["incoming_count"])
+                + max(len(inbox.get("peer_returns", {}).get("items", [])), agenda["outgoing_count"])
                 + len(inbox.get("operation_handoffs", []))
             )
+            incomplete = not agenda["scan_complete"]
+            count += int(incomplete)
             status, error = ("observed" if count else "empty"), None
         except (OSError, ValueError):
             count, status, error = 0, "unavailable", "manager_context_unreadable"
@@ -274,6 +284,7 @@ def produce():
         requested_read_scope=("owner_private_context_inbox",),
         requested_write_scope=(),
         producer=produce,
+        context_reader=lambda: snapshot,
         required_read={
             "kind": "operator_inbox",
             "command": command,

loopx/capabilities/manager_context/roundtrip.py

@@ -37,37 +37,15 @@
     result_paths,
     inspect_result_identity,
     prior_result_delivered,
+    DELIVERY_STATUSES as DELIVERY_STATUSES,
+    DELIVERY_ERRORS as DELIVERY_ERRORS,
+    reply_status as reply_status,
 )
 from ...control_plane.content_digest import BARE_SHA256_PATTERN
 from ...control_plane.collaboration.source_chat_observation import original_delivery_facts
 from ...chat import ChatMessagePayloadConflictError
 
-DELIVERY_STATUSES = {
-    "admitted",
-    "queued",
-    "retry_pending",
-    "verification_required",
-    "delivered",
-    "superseded",
-    "explicit_unverified",
-}
 EXACT_DELIVERY_ADMISSION_SECONDS = 60
-DELIVERY_ERRORS = {
-    "provider_delivery_unverified",
-    "provider_locator_unavailable",
-    "provider_verifier_unavailable",
-    "provider_verification_unavailable",
-    "provider_delivery_intent_conflict",
-    "provider_message_missing",
-    "provider_delivery_mismatch",
-    "return_authorization_unavailable",
-    "original_route_unavailable",
-    "initial_delivery_receipt_unavailable",
-    "original_route_or_return_delivery_unavailable",
-    "delivery_state_unreadable",
-    "return_transport_unavailable",
-    "manager_return_payload_conflict",
-}
 
 class ReturnResolutionBlocked(ValueError, RuntimeError):
     """A return cannot be resolved; ``reason`` is the provider-neutral code.
@@ -298,45 +276,6 @@ def report(
     }
 
 
-def reply_status(root, row):
-    result = []
-    for path in result_paths(_root(root) / "replies" / row["request_id"]):
-        phase = "decision" if path.stem == "decision" or path.stem.startswith("decision-") else "conclusion"
-        state_path = path.with_name(path.stem + ".delivery.json")
-        try:
-            reply = _read(path)
-            state = _read(state_path) if state_path.exists() else {}
-        except (OSError, ValueError):
-            result.append({
-                "phase": phase, "status": "explicit_unverified",
-                **({"result_key": path.stem} if path.stem != phase else {}),
-                "created_at": None, "delivered_at": None,
-                "error": "delivery_state_unreadable",
-            })
-            continue
-        status = state.get("status", "queued")
-        error = state.get("error")
-        if status not in DELIVERY_STATUSES:
-            status, error = "explicit_unverified", "delivery_state_unreadable"
-        elif error is not None and error not in DELIVERY_ERRORS:
-            status, error = "explicit_unverified", "delivery_state_unreadable"
-        delivered_at = state.get("delivered_at")
-        if not isinstance(delivered_at, str) or len(delivered_at) > 80:
-            delivered_at = None
-        item = {
-            "phase": phase,
-            **({"result_key": path.stem} if path.stem != phase else {}),
-            "status": status,
-            "created_at": reply.get("created_at"),
-            "delivered_at": delivered_at,
-            "error": error,
-        }
-        if state.get("verification") == "reconciled_after_restart":
-            item["verification"] = "reconciled_after_restart"
-        result.append(item)
-    return result
-
-
 def project_chat_return_deliveries(root, session_id, messages):
     """Attach public-safe delivery readback to returned Chat transcript rows."""
 

loopx/control_plane/collaboration/inbox.py

@@ -692,6 +692,71 @@ def iter_result_paths(folder: Path):
             yield from result_paths(request_folder)
 
 
+DELIVERY_STATUSES = {
+    "admitted",
+    "queued",
+    "retry_pending",
+    "verification_required",
+    "delivered",
+    "superseded",
+    "explicit_unverified",
+}
+DELIVERY_ERRORS = {
+    "provider_delivery_unverified",
+    "provider_locator_unavailable",
+    "provider_verifier_unavailable",
+    "provider_verification_unavailable",
+    "provider_delivery_intent_conflict",
+    "provider_message_missing",
+    "provider_delivery_mismatch",
+    "return_authorization_unavailable",
+    "original_route_unavailable",
+    "initial_delivery_receipt_unavailable",
+    "original_route_or_return_delivery_unavailable",
+    "delivery_state_unreadable",
+    "return_transport_unavailable",
+    "manager_return_payload_conflict",
+}
+
+def reply_status(root, row):
+    result = []
+    for path in result_paths(_root(root) / "replies" / row["request_id"]):
+        phase = "decision" if path.stem == "decision" or path.stem.startswith("decision-") else "conclusion"
+        state_path = path.with_name(path.stem + ".delivery.json")
+        try:
+            reply = _read(path)
+            state = _read(state_path) if state_path.exists() else {}
+        except (OSError, ValueError):
+            result.append({
+                "phase": phase, "status": "explicit_unverified",
+                **({"result_key": path.stem} if path.stem != phase else {}),
+                "created_at": None, "delivered_at": None,
+                "error": "delivery_state_unreadable",
+            })
+            continue
+        status = state.get("status", "queued")
+        error = state.get("error")
+        if status not in DELIVERY_STATUSES:
+            status, error = "explicit_unverified", "delivery_state_unreadable"
+        elif error is not None and error not in DELIVERY_ERRORS:
+            status, error = "explicit_unverified", "delivery_state_unreadable"
+        delivered_at = state.get("delivered_at")
+        if not isinstance(delivered_at, str) or len(delivered_at) > 80:
+            delivered_at = None
+        item = {
+            "phase": phase,
+            **({"result_key": path.stem} if path.stem != phase else {}),
+            "status": status,
+            "created_at": reply.get("created_at"),
+            "delivered_at": delivered_at,
+            "error": error,
+        }
+        if state.get("verification") == "reconciled_after_restart":
+            item["verification"] = "reconciled_after_restart"
+        result.append(item)
+    return result
+
+
 def prior_result_delivered(path, reply):
     """Ordered transport: a newer update cannot race an uncertain earlier send."""
     from ..effect_runtime import effect_runtime_result

loopx/control_plane/collaboration/peers.py

@@ -227,6 +227,21 @@ def request(
         }
 
 
+def _return_consumed(path, row, reply, *, agent_id):
+    """One readback identity for peer consumption and the per-Turn agenda."""
+    consumed = path.with_name(path.stem + ".consumed.json")
+    if not consumed.exists():
+        return False
+    value = _read(consumed)
+    expected = {"request_id": row["request_id"], "goal_id": row["goal_id"],
+        "agent_id": agent_id,
+        **({"result_key": path.stem} if "result_key" in reply else {}),
+        **({"goal_ref": row["goal_ref"]} if "goal_ref" in row else {})}
+    if any(value.get(key) != item for key, item in expected.items()):
+        raise ValueError("peer consumption receipt scope mismatch")
+    return True
+
+
 def _collect_returns(root, goal_id, agent_id, *, scope=None):
     items = []
     deliveries = []
@@ -296,24 +311,7 @@ def _collect_returns(root, goal_id, agent_id, *, scope=None):
                 )
                 if lifecycle.get("kind") == "omit":
                     continue
-            consumed = path.with_name(path.stem + ".consumed.json")
-            if consumed.exists():
-                value = _read(consumed)
-                if any(
-                    value.get(k) != v
-                    for k, v in {
-                        "request_id": row["request_id"],
-                        **({"result_key": path.stem} if "result_key" in reply else {}),
-                        "goal_id": goal_id,
-                        "agent_id": agent_id,
-                        **(
-                            {"goal_ref": row["goal_ref"]}
-                            if "goal_ref" in row
-                            else {}
-                        ),
-                    }.items()
-                ):
-                    raise ValueError("peer consumption receipt scope mismatch")
+            if _return_consumed(path, row, reply, agent_id=agent_id):
                 continue
             items.append(
                 {
@@ -673,6 +671,10 @@ def read_inbox(
                     record=return_deliveries[0][1],
                 )
             _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"]
     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. "