feat(context): prioritize related outcomes through existing capture reads

docs/reference/required-work-context.md

@@ -210,6 +210,24 @@ retains its complete body and declaration. Todo-less reads and the explicit
 reduces repeated binding observations without treating a shorter context as
 acceptance, authority or a measured model benefit.
 
+For enabled native quota with selected work, the same acceptance snapshot may
+include `outcome_context` navigation. `suggested_outcome_ids` prioritizes work
+connected through registered Todo relations, shared acceptance/source/evidence
+references, or failed/stale observations. An Agent owning two outcomes does not
+by itself connect their work. `coverage_complete` describes coverage of the
+registered relationship graph, not completeness of model understanding or proof
+that other outcomes are irrelevant. Missing relationships, unbound work or global
+acceptance widen suggestions and expose `limitations`.
+
+This is advisory reading order. It neither filters outcomes nor adds a full-read
+obligation, admission rule, approval or completion proof. The complete contract
+stays in the existing capture; use generic `quota context-read --select` to open
+needed sections, with `--list` where the installed reader supports discovery.
+No outcome-specific reader or pagination protocol is introduced. The suggestion
+is bound to the exact acceptance digest and provider revision; host oracle
+readback that changes the source drops that stale advice. Disabled delivery,
+explicit inventory inspection and signed TurnEnvelope retain their prior shapes.
+
 For an exact Todo, the existing command now returns one complete record:
 
 ```sh

loopx/cli_commands/quota.py

@@ -654,6 +654,10 @@ def handle_quota_command(
                 turn_instance_id=heartbeat_turn_id,
                 interaction_projection_hooks=interaction_projection_hooks,
                 turn_start_hook_dispatch=turn_start_hook_dispatch,
+                context_navigation=(
+                    native_context_activation(args)["enabled"]
+                    and not bool(getattr(args, "turn_envelope", False))
+                ),
                 goal_ref=goal_ref,
             )
             _attach_turn_start_hook_dispatch(payload, turn_start_hook_dispatch)

loopx/control_plane/goals/acceptance.py

@@ -195,6 +195,7 @@ def inspect_goal_acceptance(
     agent_id: str | None = None,
     todo_id: str | None = None,
     include_work_details: bool = True,
+    context_agent_id: str | None = None,
     delivery_workspace: Mapping[str, Any] | None = None,
     validation_workspace_path: Path | None = None,
 ) -> dict[str, Any]:
@@ -205,13 +206,20 @@ def inspect_goal_acceptance(
         {**route, "registered_agents": registered_agent_ids_for_goal(goal),
          **({"agent_id": agent_id} if agent_id is not None else {}),
          **({"todo_id": todo_id} if todo_id is not None else {}),
+         **({"context_agent_id": context_agent_id} if context_agent_id is not None else {}),
          **({"include_work_details": False} if not include_work_details else {})},
     )
+    from copy import deepcopy
+    captured_contract = deepcopy(basis["goal_acceptance_contract"]) if "outcome_context" in basis else None
     basis["goal_acceptance_contract"] = observe_goal_acceptance_oracles(
         registry_path=registry_path, goal_id=goal_id,
         contract=basis["goal_acceptance_contract"], criteria=(basis.get("contract") or {}).get("criteria", []),
         task_repository=(basis.get("todo") or {}).get("task_repository"),
         delivery_workspace=delivery_workspace, validation_workspace_path=validation_workspace_path)
+    if captured_contract is not None and basis["goal_acceptance_contract"] != captured_contract:
+        # The optional navigation binds the exact owner's observation. Do not
+        # reuse it after host-side oracle readback changes those facts.
+        basis.pop("outcome_context", None)
     return basis
 
 

loopx/control_plane/goals/acceptance_authority.ts

@@ -1,4 +1,5 @@
-import {bindGoalOutcomesStateOwner, GoalOutcomeValidationError, goalOutcomeCheckBasis, refineGoalOutcome} from "./outcome_requirements.ts";
+import {bindGoalOutcomesStateOwner, GoalOutcomeValidationError, goalOutcomeCheckBasis, refineGoalOutcome,
+  outcomeContextNavigation} from "./outcome_requirements.ts";
 /** Acceptance effects use the existing canonical head, CAS and operation journal.
  * Only trusted local owner/host adapters may invoke these mutation exports. */
 import {isAbsolute} from "node:path";
@@ -561,6 +562,7 @@ export async function inspectGoalAcceptance(
   includeWorkDetails = true,
   agentId?: string,
   registeredAgents?: unknown,
+  contextAgentId?: string,
 ): Promise<JsonObject> {
   const head = await store.loadAuthority();
   if (head.status !== "loaded") return source(store, {...head});
@@ -582,14 +584,23 @@ export async function inspectGoalAcceptance(
     ...(state ? acceptanceTask(key, todo, state)
       : {todo_id: key, state: "unbound", criterion_ids: [], applicable: false}),
   }));
+  const contract = projectGoalAcceptance(head.head, goalId, registeredAgents);
+  const context = !includeWorkDetails && contextAgentId !== undefined && todoId !== undefined
+    ? outcomeContextNavigation(head.head, goalId, contextAgentId, todoId, contract) : null;
+  if (context && (!Array.isArray(registeredAgents) || !registeredAgents.includes(contextAgentId))) {
+    context.coverage_complete = false;
+    context.limitations = [...context.limitations as string[], "context_agent_unavailable"];
+    context.suggested_outcome_ids = ((contract.outcome_requirements as JsonObject).outcomes as JsonObject[]).map(o => o.id);
+  }
   return source(store, {status: "loaded", provider_revision: head.provider_revision,
     revision: state?.revision ?? null, contract_digest: state?.digest ?? null,
     contract: state ? state.document : null, tasks,
     ...(todoId === undefined || !includeWorkDetails ? {} : {todo: todos.get(todoId) ?? null,
       completion_requirements: acceptanceCompletionRequirements(head.head, goalId, todoId)}),
     ...(agentId === undefined || currentGoalObjective(head.head) === null ? {} : {objective_work_digest: goalObjectiveWorkDigest(head.head, agentId),
       goal_objective_guard: goalObjectiveWorkGuard(head.head, agentId)}),
-    goal_acceptance_contract: projectGoalAcceptance(head.head, goalId, registeredAgents)});
+    ...(context === null ? {} : {outcome_context: {...context, provider_revision: head.provider_revision}}),
+    goal_acceptance_contract: contract});
 }
 
 async function local(value: unknown, kind: "inspect" | "configure" | "verify" | "lifecycle" | "refine" | "objective"): Promise<JsonObject> {
@@ -609,7 +620,8 @@ async function local(value: unknown, kind: "inspect" | "configure" | "verify" |
           request.goal_ref === undefined ? undefined : parseWireExactGoalRef(request.goal_ref, "acceptance request goal_ref"),
           request.include_work_details !== false,
           request.agent_id === undefined ? undefined : requireAuthorityStoreId(request.agent_id, "agent_id"),
-          request.registered_agents);
+          request.registered_agents,
+          request.context_agent_id === undefined ? undefined : requireAuthorityStoreId(request.context_agent_id, "context_agent_id"));
       }
       if (kind === "configure") return configureGoalAcceptance(store, request);
       if (kind === "refine") return refineGoalAcceptanceOutcome(store, request);

loopx/control_plane/goals/outcome_requirements.ts

@@ -7,8 +7,11 @@ import {readGoalAcceptanceLifecycle, sameExactGoalRef, parseWireExactGoalRef, ty
 
 import {BARE_SHA256_PATTERN} from "../content_digest.ts";
 import {goalAcceptanceTodoDigest} from "./acceptance_work_basis.ts";
+import {projectRelations, todoRef} from "../work_items/planning_relations.ts";
+import {normalizeTodoResumeWhen, TODO_RESUME_NORMALIZE_REQUEST_SCHEMA_VERSION} from "../todos/resume_condition.ts";
 
 export const GOAL_OUTCOMES_SCHEMA = "loopx_goal_outcomes_v0";
+export const OUTCOME_CONTEXT_NAVIGATION_SCHEMA = "outcome_context_navigation_v1";
 interface Requirement extends JsonObject {id: string; description: string}
 interface ReviewItem extends JsonObject {id: string; conclusion: "met" | "unmet" | "uncertain"; evidence_refs: string[]; note: string}
 interface OutcomeReview extends JsonObject {basis_digest: string; items: ReviewItem[]}
@@ -230,6 +233,112 @@ export function projectGoalOutcomes(head: JsonObject, goalId: string, acceptance
     execution_authority_granted: false, goal_completion_assessed: false}, acceptance);
 }
 
+/** Suggest reading order from registered relationships, never reading obligations.
+ * Use the complete canonical work snapshot and the existing relation owner;
+ * the bounded planning horizon cannot establish this registered-relation closure.
+ * Unknown relationships widen suggestions, never force a full-file model read.
+ * Ownership alone neither establishes relevance nor proves independence.
+ */
+export function outcomeContextNavigation(head: JsonObject, goalId: string, agentId: string,
+  todoId: string, acceptance: JsonObject): JsonObject | null {
+  const declaration = acceptance.outcome_requirements as JsonObject | undefined;
+  if (!declaration || !Array.isArray(declaration.outcomes)) return null;
+  const outcomes = declaration.outcomes as JsonObject[];
+  const work = indexCoordinationProjectionTodos(head, goalId).todos;
+  const suggested = new Set<string>();
+  const related = new Set([todoId]);
+  const reasons = new Set<string>();
+  if (declaration.current_lifetime !== true) reasons.add("outcome_lifetime_unavailable");
+  if (!work.has(todoId) || !outcomes.some(o => (o.todo_ids as string[]).includes(todoId)))
+    reasons.add("selected_work_unbound");
+  if (Array.isArray(acceptance.criteria) && acceptance.criteria.length &&
+      (acceptance.scope as JsonObject | undefined)?.kind !== "selected_work")
+    reasons.add("goal_wide_acceptance");
+  const adjacency = new Map<string, Set<string>>();
+  const connect = (left: string, right: string) => {
+    for (const [from, to] of [[left, right], [right, left]]) {
+      const neighbors = adjacency.get(from) ?? new Set<string>();
+      neighbors.add(to); adjacency.set(from, neighbors);
+    }
+  };
+  const rows = [...work.values()].map(todo => ({todo_id: String(todo.todo_id),
+    successor_todo_ids: (todo.successor_todo_ids ?? []) as string[],
+    ...Object.fromEntries(["unblocks_todo_id", "superseded_by", "resume_when"]
+      .filter(key => typeof todo[key] === "string" && todo[key] !== "")
+      .map(key => [key, todo[key]]))}));
+  const sharedRefs = new Map<string, string>();
+  for (const relation of projectRelations(rows)) {
+    const target = todoRef(relation.to_ref);
+    if (target !== null) {
+      if (!work.has(target)) reasons.add("work_relation_unavailable");
+      else connect(relation.from_todo_id, target);
+    } else if (relation.relation === "resumes_when") {
+      // Exact external conditions may be shared without naming a Todo.
+      // No substring/prose matching or invented cross-Agent dependency.
+      const normalized = normalizeTodoResumeWhen({
+        schema_version: TODO_RESUME_NORMALIZE_REQUEST_SCHEMA_VERSION, resume_when: relation.to_ref,
+      });
+      if (normalized === null) { reasons.add("work_relation_unavailable"); continue; }
+      const key = `${relation.relation}:${normalized}`;
+      const previous = sharedRefs.get(key);
+      if (previous) connect(previous, relation.from_todo_id);
+      else sharedRefs.set(key, relation.from_todo_id);
+    } else reasons.add("work_relation_unavailable");
+  }
+  const checkScope = acceptance.scope as JsonObject | undefined;
+  if (checkScope?.kind === "selected_work" && Array.isArray(checkScope.todo_ids)) {
+    const ids = checkScope.todo_ids as string[];
+    for (const key of ids) {
+      if (!work.has(key)) reasons.add("acceptance_work_unavailable");
+      else if (ids[0]) connect(ids[0], key);
+    }
+  }
+  const materialWork = new Map<string, string>();
+  for (const outcome of outcomes) {
+    const ids = outcome.todo_ids as string[];
+    if (!Array.isArray(ids) || !ids.length || ids.some(key => !work.has(key))) {
+      reasons.add("outcome_work_unavailable"); continue;
+    }
+    // One outcome may legitimately span several Agents' work. Ownership is
+    // not a substitute for its actual linked tasks or their shared acceptance.
+    for (const key of ids) connect(ids[0], key);
+    // A common declared request or evidence item may carry shared obligations
+    // or opposing interpretations. Expand its actual work without deciding
+    // relevance from descriptions or promoting either review to acceptance.
+    const refs = [...(outcome.sources as JsonObject[]).map(source => String(source.ref)),
+      ...(outcome.requirements as JsonObject[]).flatMap(r =>
+        ((r.agent_review as JsonObject | undefined)?.evidence_refs ?? []) as string[])];
+    for (const ref of refs) {
+      const previous = materialWork.get(ref);
+      if (previous) connect(previous, ids[0]);
+      else materialWork.set(ref, ids[0]);
+    }
+    // A failed/stale current observation cannot be hidden solely by ownership;
+    // its related work and complete evidence also need assessment.
+    if ((outcome.requirements as JsonObject[]).some(r => r.status === "failed" || r.status === "stale"))
+      suggested.add(String(outcome.id));
+    if (suggested.has(String(outcome.id))) for (const key of ids) related.add(key);
+  }
+  const queue = [...related];
+  for (let cursor = 0; cursor < queue.length; cursor++) for (const next of adjacency.get(queue[cursor]) ?? []) {
+    if (!related.has(next)) { related.add(next); queue.push(next); }
+  }
+  for (const outcome of outcomes) {
+    if ((outcome.todo_ids as string[]).some(key => related.has(key))) suggested.add(String(outcome.id));
+  }
+  if (reasons.size) for (const outcome of outcomes) suggested.add(String(outcome.id));
+  return {schema_version: OUTCOME_CONTEXT_NAVIGATION_SCHEMA, goal_id: goalId, agent_id: agentId, todo_id: todoId,
+    outcome_revision: declaration.revision,
+    acceptance_digest: canonicalAuthoritySha256(acceptance),
+    coverage_complete: reasons.size === 0, limitations: [...reasons].sort(),
+    suggested_outcome_ids: outcomes.filter(o => suggested.has(String(o.id))).map(o => o.id),
+    instruction: "Navigation from registered work relationships, not a reading gate or a proof of independence. " +
+      "Start with the suggested outcomes; inspect other criteria, shared constraints and contrary evidence when the decision needs them. " +
+      "All outcomes remain in goal_acceptance_contract. Use the same capture's generic field index and selectors for detail. " +
+      "Incomplete coverage calls for judgment and relevant expansion, not an automatic full-file read. " +
+      "Goal intent, required criteria, permissions, stops and verification obligations still apply."};
+}
+
 /** Re-use after the host observes oracle freshness; raw stored evidence cannot
  * become a current passing result without those observations. No state writes. */
 export function observeOutcomeRequirements(declaration: JsonObject, acceptance: JsonObject,

loopx/control_plane/quota/live_decision.py

@@ -545,6 +545,7 @@ def build_live_quota_should_run_decision(
     interaction_projection_hooks: Sequence[InteractionProjectionHookRegistration]
     | None = None,
     turn_start_hook_dispatch: Mapping[str, Any] | None = None,
+    context_navigation: bool = False,
     workspace_path: Path | None = None,
     goal_ref: Mapping[str, object] | None = None,
     inline_goal_source: bool = False,
@@ -825,5 +826,6 @@ def build_live_quota_should_run_decision(
     from ..work_items.context_readback import attach_work_context
 
     attach_work_context(payload, registry_path=registry_path, runtime_root=runtime_root,
-        hook_dispatch=turn_start_hook_dispatch, inline_goal_source=inline_goal_source)
+        hook_dispatch=turn_start_hook_dispatch, context_navigation=context_navigation,
+        inline_goal_source=inline_goal_source)
     return payload

loopx/control_plane/work_items/context_readback.py

@@ -20,7 +20,7 @@
 from ..todos.summary_item import todo_text_content_revision
 
 
-def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id,
+def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id, agent_id=None,
                     inline_goal_source=False):
     source = read.get("source")
     if source == "goal_state":
@@ -73,7 +73,7 @@ def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id,
         # selected-Todo read owns its body; do not transport it a second time
         # through the bounded acceptance RPC before discarding it here.
         return inspect_goal_acceptance(registry_path=registry_path, goal_id=goal_id,
-            runtime_root=str(runtime_root), todo_id=todo_id, include_work_details=False)
+            runtime_root=str(runtime_root), todo_id=todo_id, context_agent_id=agent_id, include_work_details=False)
     if source == "selected_todo":
         if not todo_id:
             raise ValueError("selected Todo identity is unavailable")
@@ -131,6 +131,7 @@ def _user_context(*, registry_path, runtime_root, goal_id, agent_id):
 
 def attach_work_context(payload: dict[str, Any], *, registry_path: Path,
                         runtime_root: Path, hook_dispatch: dict[str, Any] | None,
+                        context_navigation: bool = False,
                         inline_goal_source: bool = False) -> None:
     packet_selected = payload.get("selected_todo")
     selected_text_snapshot = None
@@ -157,7 +158,7 @@ def attach_work_context(payload: dict[str, Any], *, registry_path: Path,
         try:
             result["content"] = _source_content(read, registry_path=registry_path,
                 runtime_root=runtime_root, goal_id=goal_id, todo_id=plan.get("todo_id"),
-                inline_goal_source=inline_goal_source)
+                agent_id=agent_id if context_navigation else None, inline_goal_source=inline_goal_source)
         except (OSError, ValueError, RuntimeError):
             result["error_code"] = "context_source_unavailable"
         results.append(result)