fix(collaboration): surface peer returns on next turn

loopx/capabilities/manager_context/README.md

@@ -241,6 +241,14 @@ awaiting explicit requester consumption. The same read model is returned by
 private agenda contains current-purpose excerpts and explicit links, not source
 transcripts, result bodies or receiver rationale.
 
+The agenda also carries a bounded `return_attention` projection for sent peer
+requests whose returned result has not been consumed. It contains only the
+request/revision identity, result keys and delivery/consumption state. Native
+context delivery places this projection in the next Turn's opening before long
+work-context sources, while the full result text remains behind the existing
+`manager-inbox read` command. This preserves asynchronous pull semantics and
+does not interrupt an active Turn or treat visibility as consumption.
+
 Outgoing requests to an original project conversation use the same typed address
 reader as explicit peer returns. The agenda retains that conversation recipient
 and its current revision without inventing a Goal/Agent receiver. A changed

loopx/control_plane/collaboration/turn_context.ts

@@ -44,6 +44,7 @@ export function projectDelegationTurnItem(value: unknown): JsonObject | null {
   const purpose = String(control?.brief && (control.brief as JsonObject).purpose
     || request.purpose || "").replace(/\s+/g, " ").trim();
   const todos = Array.isArray(row.todo_ids) ? row.todo_ids : [];
+  const visibleReturns = (step === "consume_peer_result" ? unconsumed : results).slice(-3);
   return {
     request_id: request.request_id, direction,
     ...(request.goal_ref ? {goal_ref: request.goal_ref} : {}),
@@ -57,9 +58,9 @@ export function projectDelegationTurnItem(value: unknown): JsonObject | null {
     ...(receipt.remaining_work ? {remaining_work: receipt.remaining_work} : {}),
     ...(todos.length ? {todo_ids: todos.slice(0, 4), additional_todo_count: Math.max(0, todos.length - 4)} : {}),
     ...(results.length ? {pending_return_count: incomplete.length, unconsumed_return_count: unconsumed.length,
-      returns: results.slice(-3).map(result => ({result_key: result.result_key,
+      returns: visibleReturns.map(result => ({result_key: result.result_key,
         request_revision: result.request_revision, status: result.status, consumed: result.consumed})),
-      additional_return_count: Math.max(0, results.length - 3)} : {}),
+      additional_return_count: results.length - visibleReturns.length} : {}),
   };
 }
 
@@ -75,14 +76,72 @@ function visibleRows(rows: JsonObject[]): JsonObject[] {
   return visible;
 }
 
+const RETURN_ATTENTION_LIMIT = 4;
+
+function peerReturnNeedsAttention(row: JsonObject): boolean {
+  return row.direction === "outgoing" && row.step === "consume_peer_result"
+    && typeof row.unconsumed_return_count === "number" && row.unconsumed_return_count > 0
+    && Array.isArray(row.returns) && row.returns.some(value => {
+      const result = value !== null && typeof value === "object" && !Array.isArray(value)
+        ? value as JsonObject : undefined;
+      return result?.consumed !== true;
+    });
+}
+
+function peerReturnAttention(rows: JsonObject[]): JsonObject {
+  const pending = rows.filter(peerReturnNeedsAttention);
+  const items = pending.slice(0, RETURN_ATTENTION_LIMIT).map(row => {
+    const item: JsonObject = {
+      request_id: row.request_id,
+      request_revision: row.request_revision,
+      step: row.step,
+      unconsumed_return_count: row.unconsumed_return_count,
+      returns: row.returns,
+    };
+    for (const field of ["source_agent_id", "agent_id"] as const) {
+      if (row[field] !== undefined) item[field] = row[field];
+    }
+    return item;
+  });
+  return {count: pending.length, items, omitted_count: pending.length - items.length};
+}
+
+function requirePeerReturnAttention(value: unknown): {count: number; items: JsonObject[]} {
+  const attention = requireJsonObject(value, "delegation return attention");
+  const count = attention.count, omitted = attention.omitted_count;
+  if (typeof count !== "number" || !Number.isSafeInteger(count) || count < 0
+      || typeof omitted !== "number" || !Number.isSafeInteger(omitted) || omitted < 0
+      || !Array.isArray(attention.items) || attention.items.length > RETURN_ATTENTION_LIMIT
+      || count !== attention.items.length + omitted) {
+    throw new EffectRuntimeRequestError("invalid delegation return attention");
+  }
+  const items = attention.items.map(value => {
+    const item = requireJsonObject(value, "delegation return attention item");
+    if (typeof item.request_id !== "string" || !BARE_SHA256_PATTERN.test(item.request_id)
+        || typeof item.request_revision !== "number" || !Number.isSafeInteger(item.request_revision)
+        || item.request_revision < 1 || item.step !== "consume_peer_result"
+        || typeof item.unconsumed_return_count !== "number"
+        || !Number.isSafeInteger(item.unconsumed_return_count) || item.unconsumed_return_count < 1
+        || !Array.isArray(item.returns) || !item.returns.some(value => {
+          const result = requireJsonObject(value, "delegation return attention result");
+          return result.consumed !== true;
+        })) {
+      throw new EffectRuntimeRequestError("invalid delegation return attention item");
+    }
+    return item;
+  });
+  return {count, items};
+}
+
 /** Input is already scope-checked by the existing inbox and result owners. */
 export function projectDelegationTurnItems(params: JsonObject): JsonObject {
   if (!Array.isArray(params.observations) || params.observations.length > 128) {
     throw new EffectRuntimeRequestError("delegation observation batch must contain at most 128 requests");
   }
   const rows = params.observations.map(projectDelegationTurnItem).filter((value): value is JsonObject => value !== null);
   return {items: visibleRows(rows), incoming_count: rows.filter(row => row.direction === "incoming").length,
-    outgoing_count: rows.filter(row => row.direction === "outgoing").length};
+    outgoing_count: rows.filter(row => row.direction === "outgoing").length,
+    return_attention: peerReturnAttention(rows)};
 }
 
 export function projectDelegationTurnContext(params: JsonObject): JsonObject {
@@ -103,9 +162,14 @@ export function projectDelegationTurnContext(params: JsonObject): JsonObject {
   }, 0);
   const incoming = total("incoming_count"), outgoing = total("outgoing_count");
   const visible = visibleRows(rows);
+  const attention = batches.map(batch => requirePeerReturnAttention(batch.return_attention));
+  const attentionCount = attention.reduce((sum, batch) => sum + batch.count, 0);
+  const attentionItems = attention.flatMap(batch => batch.items).slice(0, RETURN_ATTENTION_LIMIT);
   return {ok: true, read_complete: false, unfinished_delegations: {
     count_unit: "request", items: visible, incoming_count: incoming, outgoing_count: outgoing,
     omitted_count: incoming + outgoing - visible.length,
+    return_attention: {count: attentionCount, items: attentionItems,
+      omitted_count: attentionCount - attentionItems.length},
     scan_complete: params.scan_complete === true && params.unavailable_count === 0,
     unavailable_count: params.unavailable_count, observed_at: params.observed_at,
     read_command: params.read_command,

loopx/control_plane/quota/progressive_context.ts

@@ -867,8 +867,7 @@ export function scopedReplanYardstickReversal(decision: JsonObject) {
   return {cue, fields, path: [...REPLAN_CONTEXT_PATH, "explore_basis", "work_yardsticks", index]};
 }
 
-/** A pending Manager Inbox read already delivers these scoped request rows. */
-function managerAgendaOpeningFields(value: unknown, requiredReads: unknown[]): string[] | undefined {
+function requiredManagerAgenda(value: unknown, requiredReads: unknown[]): JsonObject | undefined {
   const {content, ...read} = jsonObject(value) ?? {};
   const source = jsonObject(content), agenda = jsonObject(source?.unfinished_delegations);
   if (read.source !== "turn_start_capability_hook" || read.kind !== "operator_inbox"
@@ -879,12 +878,20 @@ function managerAgendaOpeningFields(value: unknown, requiredReads: unknown[]): s
       || agenda.scan_complete !== true || agenda.unavailable_count !== 0
       || !Array.isArray(agenda.items)
       || !known(agenda, ["count_unit", "items", "incoming_count", "outgoing_count", "omitted_count",
-        "scan_complete", "unavailable_count", "observed_at", "read_command", "instruction"])
+        "return_attention", "scan_complete", "unavailable_count", "observed_at", "read_command", "instruction"])
       || !requiredReads.some(required => isDeepStrictEqual(required, read))) return undefined;
+  return agenda;
+}
+
+/** A pending Manager Inbox read already delivers these scoped request rows. */
+function managerAgendaOpeningFields(value: unknown, requiredReads: unknown[]): string[] | undefined {
+  const agenda = requiredManagerAgenda(value, requiredReads);
+  if (!agenda) return undefined;
   const {incoming_count: incoming, outgoing_count: outgoing, omitted_count: omitted, items} = agenda;
   if (typeof incoming !== "number" || !Number.isSafeInteger(incoming) || incoming < 0
       || typeof outgoing !== "number" || !Number.isSafeInteger(outgoing) || outgoing < 0
       || typeof omitted !== "number" || !Number.isSafeInteger(omitted) || omitted < 0
+      || !Array.isArray(items)
       || !Number.isSafeInteger(incoming + outgoing) || incoming + outgoing !== items.length + omitted
       || Buffer.byteLength(JSON.stringify(items), "utf8") < 2048) return undefined;
   const fields = ["incoming_count", "outgoing_count", "scan_complete"];
@@ -897,6 +904,26 @@ function managerAgendaOpeningFields(value: unknown, requiredReads: unknown[]): s
   return undefined;
 }
 
+/** Keep only returned-result identities in the first page; full Inbox text stays on its owner read. */
+function managerReturnAttention(value: unknown, requiredReads: unknown[]): JsonObject | undefined {
+  const agenda = requiredManagerAgenda(value, requiredReads);
+  const attention = jsonObject(agenda?.return_attention);
+  if (!attention || !known(attention, ["count", "items", "omitted_count"])
+      || typeof attention.count !== "number" || !Number.isSafeInteger(attention.count) || attention.count <= 0
+      || typeof attention.omitted_count !== "number" || !Number.isSafeInteger(attention.omitted_count)
+      || attention.omitted_count < 0 || !Array.isArray(attention.items) || attention.items.length > 4
+      || attention.count !== attention.items.length + attention.omitted_count) return undefined;
+  for (const value of attention.items) {
+    const item = jsonObject(value);
+    if (!item || item.step !== "consume_peer_result" || typeof item.request_id !== "string"
+        || typeof item.unconsumed_return_count !== "number" || item.unconsumed_return_count <= 0
+        || !Array.isArray(item.returns) || !item.returns.some(value => jsonObject(value)?.consumed !== true)) {
+      return undefined;
+    }
+  }
+  return attention;
+}
+
 function initialContextReadGroups(context: JsonObject, decisionFields: readonly string[]): {
   opening: string[]; remaining: string[];
 } {
@@ -969,25 +996,36 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
   if (turnBindingRequired)
     for (const field of ["turn_binding_required", ...admissionFields])
       include("interaction_contract", "cli_channel", field);
-  include("interaction_contract", "agent_channel", "required_reads");
-  const work = ["interaction_contract", "agent_channel", "work_context"];
-  include(...work, "instruction");
-  if (deferredReentry) {
-    // Surface the owner's reopen and same-Turn guard before the long read.
-    for (const field of admissionFields) include("interaction_contract", "cli_channel", field);
-  }
   const channel = object(interaction.agent_channel);
   const requiredInboxReads = context.effective_action === EffectiveAction.NORMAL_RUN && context.normal_delivery_allowed === true
     && channel?.delivery_allowed === true && channel.must_attempt === true && Array.isArray(channel.required_reads)
     ? channel.required_reads : [];
   const sources = object(channel?.work_context)?.sources;
   const sourceRows = Array.isArray(sources) ? sources.map((source, index) => ({source, index})) : [];
+  const attentionSources = sourceRows.filter(({source}) => managerReturnAttention(source, requiredInboxReads));
+  const requiredReadsPath = ["interaction_contract", "agent_channel", "required_reads"];
+  if (!attentionSources.length) include(...requiredReadsPath);
+  const work = ["interaction_contract", "agent_channel", "work_context"];
+  include(...work, "instruction");
+  if (deferredReentry) {
+    // Surface the owner's reopen and same-Turn guard before the long read.
+    for (const field of admissionFields) include("interaction_contract", "cli_channel", field);
+  }
+  // A long Manager Inbox agenda stays behind its required read. Surface only
+  // the bounded returned-result identities before work so paging or context
+  // recovery cannot hide that the requester has a result to consume.
+  const cueStart = pointers.length;
+  if (attentionSources.length) {
+    include(...requiredReadsPath);
+    for (const {index} of attentionSources) {
+      include(...work, "sources", index, "content", "unfinished_delegations", "return_attention");
+    }
+  }
   const selectedTodo = object(context.selected_todo);
   const identityAgent = object(context.agent_identity)?.agent_id;
   const agentId = context.agent_id ?? identityAgent;
   // Deliver the opted-in decision cue before work, while the candidate or
   // test can still change. The template and evidence stay on their owner read.
-  const cueStart = pointers.length;
   if (replanReversal) {
     for (const field of replanReversal.fields) include(...replanReversal.path, field);
   }

tests/control_plane/test_context_first_read.py

@@ -1071,6 +1071,42 @@ def test_initial_read_leaves_large_agenda_rows_to_the_matching_required_read(tmp
     assert json.loads(detail["content"])["selections"][0]["value"] == agenda
 
 
+def test_initial_read_surfaces_bounded_peer_return_attention_before_long_sources(tmp_path, decision):
+    command = "loopx manager-inbox read --goal-id fixture --agent-id author"
+    read = {"source": "turn_start_capability_hook", "kind": "operator_inbox",
+        "capability_id": "manager-context", "hook_id": "manager.context_inbox",
+        "ordering": "before_work", "command": command}
+    attention = {"count": 1, "omitted_count": 0, "items": [{
+        "request_id": "a" * 64, "request_revision": 1, "step": "consume_peer_result",
+        "unconsumed_return_count": 1,
+        "returns": [{"result_key": "conclusion", "request_revision": 1,
+            "status": "delivered", "consumed": False}],
+    }]}
+    items = [{"request_id": f"{index:064x}", "direction": "incoming" if index < 10 else "outgoing",
+        "step": "review_request_work", "purpose": "Keep the original evidence. " * 16}
+        for index in range(20)]
+    agenda = {"count_unit": "request", "incoming_count": 10, "outgoing_count": 11,
+        "items": items, "omitted_count": 1, "return_attention": attention,
+        "scan_complete": True, "unavailable_count": 0, "read_command": command,
+        "instruction": "Read the full current request before acting."}
+    decision.update(should_run=True, normal_delivery_allowed=True, effective_action="normal_run")
+    channel = decision["interaction_contract"]["agent_channel"]
+    channel.update(delivery_allowed=True, must_attempt=True)
+    channel["required_reads"].append(read)
+    sources = channel["work_context"]["sources"]
+    sources.append({**read, "content": {"ok": True, "read_complete": False,
+        "unfinished_delegations": agenda}})
+
+    view = native_context_capture_view(decision, tmp_path)
+    _, values = read_selection(view["first_read_command"])
+    prefix = f"/interaction_contract/agent_channel/work_context/sources/{len(sources) - 1}/content/unfinished_delegations"
+    attention_path = prefix + "/return_attention"
+    assert values[attention_path] == attention
+    assert prefix not in values and prefix + "/items" not in values
+    goal_text = "/interaction_contract/agent_channel/work_context/sources/0/content/text"
+    assert list(values).index(attention_path) < list(values).index(goal_text)
+
+
 def test_unknown_settlement_plan_is_not_compacted(tmp_path, decision):
     plan = {"schema_version": "future", "new_stop": "Stop before every new effect."}
     decision["interaction_contract"]["cli_channel"]["settlement_plan"] = plan