fix(quota): retain outcome review identity across work-link refinements

loopx/control_plane/quota/live_decision.py

@@ -57,6 +57,7 @@ def _receipt_bound_guard_fields(
             and guard.get("selected_obligation_id") == obligation_id
         ),
         "receipt_bound_guard_recorded_at": receipt.get("recorded_at"),
+        "receipt_bound_replan_review": guard.get("selected_review_source") if isinstance(guard, Mapping) else None,
         "receipt_bound_successor_replan": bool(getattr(readback, "successor_replan_pending", False)),
     }
 

loopx/control_plane/quota/settlement.py

@@ -312,6 +312,12 @@ def _semantic_replan_guard(value: Any) -> dict[str, Any] | None:
         # The TS readback owner validates the schema and ids. Preserve the
         # selected authority evidence when adapting the receipt to refresh.
         result["selected_capability_guard"] = dict(capability)
+    if "selected_review_source" in guard:
+        review = guard["selected_review_source"]
+        if not isinstance(review, dict) or selected_obligation_id is None:
+            raise RuntimeError("selected review source requires a typed object and obligation")
+        # Semantic eligibility and scalar bounds belong to the TS owner.
+        result["selected_review_source"] = dict(review)
     return result
 
 

loopx/control_plane/quota/settlement_cli.py

@@ -403,6 +403,26 @@ def quota_rollout_details(
         details["admission_resume_when"] = selected_todo.get("resume_when")
         details["admission_resume_ready"] = selected_todo.get("resume_ready")
     replan_packet = payload.get("replan_action_packet")
+    obligation = payload.get("autonomous_replan_obligation")
+    if semantic_replan_obligation_id and isinstance(obligation, Mapping) and obligation.get("obligation_id") == semantic_replan_obligation_id:
+        from ..work_items.replan_semantics_codec import project_replan_semantics
+        from ...rollout_event_log import _safe_details
+
+        source = project_replan_semantics({"operation": "receipt_review_source",
+            "agent_id": getattr(args, "agent_id", None),
+            "obligation": {key: value for key, value in obligation.items() if key != "replan_context"}})
+        review = source["review_source"]
+        if isinstance(review, Mapping):
+            # Retain the original stop clause only when the existing public
+            # metadata serializer preserves it exactly. Unsafe/lossy sources
+            # keep legacy strict recovery; never truncate a governing clause.
+            try:
+                safe = _safe_details({"semantic_replan_review_kind": review["kind"],
+                    "semantic_replan_review_stop_condition": review["stop_condition"]})
+            except ValueError:
+                safe = {}
+            if safe.get("semantic_replan_review_stop_condition") == review["stop_condition"]:
+                details.update(safe)
     receipt = payload.get("heartbeat_receipt")
     capability_guard = (replan_packet.get("capability_guard") if isinstance(replan_packet, Mapping) else None) or (
         receipt.get("semantic_replan_capability_guard") if isinstance(receipt, Mapping) else None

loopx/control_plane/quota/settlement_readback.ts

@@ -497,8 +497,13 @@ export function projectSemanticReplanGuard(
 ): JsonObject {
   const hasCapability = ["semantic_replan_capability_id", "semantic_replan_gap_id", "semantic_replan_frontier_revision"]
     .some(field => Object.hasOwn(receiptDetails, field));
+  const hasReview = ["semantic_replan_review_kind", "semantic_replan_review_stop_condition"]
+    .some(field => Object.hasOwn(receiptDetails, field));
+  if (hasCapability && hasReview) throw new EffectRuntimeRequestError(
+    "a review source cannot replace a capability guard", "malformed_settlement_state");
   if (!Object.hasOwn(receiptDetails, "semantic_replan_obligation_id")) {
     if (hasCapability) throw new EffectRuntimeRequestError("a capability guard requires a selected obligation", "malformed_settlement_state");
+    if (hasReview) throw new EffectRuntimeRequestError("a review source requires a selected obligation", "malformed_settlement_state");
     return {
       schema_version: SEMANTIC_REPLAN_GUARD_SCHEMA,
       scope: "legacy_unscoped",
@@ -508,6 +513,7 @@ export function projectSemanticReplanGuard(
   const rawObligationId = receiptDetails.semantic_replan_obligation_id;
   if (rawObligationId === "" || rawObligationId === null) {
     if (hasCapability) throw new EffectRuntimeRequestError("a capability guard requires a selected obligation", "malformed_settlement_state");
+    if (hasReview) throw new EffectRuntimeRequestError("a review source requires a selected obligation", "malformed_settlement_state");
     return {
       schema_version: SEMANTIC_REPLAN_GUARD_SCHEMA,
       scope: "turn_guard",
@@ -521,10 +527,15 @@ export function projectSemanticReplanGuard(
       "malformed_settlement_state",
     );
   }
+  const reviewKind = hasReview ? requireNonEmptyString(receiptDetails.semantic_replan_review_kind, "receipt review kind") : null;
+  const stop = hasReview ? requireNonEmptyString(receiptDetails.semantic_replan_review_stop_condition, "receipt review stop condition") : null;
+  if (stop && stop.length > 240) throw new EffectRuntimeRequestError(
+    "receipt review stop condition exceeds scalar metadata bounds", "malformed_settlement_state");
   return {
     schema_version: SEMANTIC_REPLAN_GUARD_SCHEMA,
     scope: "turn_guard",
     selected_obligation_id: selectedObligationId,
+    ...(hasReview ? {selected_review_source: {kind: reviewKind, stop_condition: stop}} : {}),
     ...(hasCapability ? {
       selected_capability_guard: decodeCapabilityGuard({schema_version: "semantic_replan_capability_guard_v0",
         capability_id: receiptDetails.semantic_replan_capability_id, gap_id: receiptDetails.semantic_replan_gap_id,

loopx/control_plane/quota/should_run.py

@@ -567,6 +567,7 @@ def build_quota_should_run(
     receipt_bound_replan_guard_scoped: bool = False,
     receipt_bound_successor_replan: bool = False,
     receipt_bound_guard_recorded_at: str | None = None,
+    receipt_bound_replan_review: Mapping[str, Any] | None = None,
     turn_instance_id: str | None = None,
     runtime_root: str | Path | None = None,
     workspace_path: Path | None = None,
@@ -664,6 +665,7 @@ def build_quota_should_run(
             authoritative_fallback_todo_items=authoritative_fallback_todo_items,
             replan_explore_reader=tracked_replan_explore_reader,
             receipt_bound_guard_recorded_at=receipt_bound_guard_recorded_at,
+            receipt_bound_replan_review=receipt_bound_replan_review,
         )
         route = _resolve_quota_should_run_route(prepared)
         route = _apply_selected_todo_guards(

loopx/control_plane/quota/should_run_prepare.py

@@ -40,6 +40,7 @@
 from ..quota.error_codes import HeartbeatReceiptIdentityConflictError
 from ..effect_runtime import EffectRuntimeRejected
 from ..work_items.replan_semantics_codec import project_replan_semantics
+from ..work_items.lifecycle_recovery import attach_receipt_replan_context
 from ..agents.capability_memory import resolve_agent_capabilities
 from ..quota.goal_boundary import (
     goal_boundary as _goal_boundary,
@@ -196,6 +197,7 @@ def _preserve_receipt_bound_replan_obligation(
     replay_phase: ReceiptBoundReplayPhase | None = None,
     transition_candidates: list[dict[str, Any]] | None = None,
     stride_strategy: str = "off",
+    receipt_review: Mapping[str, Any] | None = None,
 ) -> dict[str, Any] | None:
     preserved_replan_id = normalize_todo_replan_obligation_id(
         receipt_bound_replan_obligation_id
@@ -211,6 +213,7 @@ def _preserve_receipt_bound_replan_obligation(
             "replay_phase": replay_phase.value if replay_phase is not None else None,
             "transition_candidates": transition_candidates or [],
             "stride_strategy": stride_strategy,
+            "receipt_review": receipt_review,
         })
     except EffectRuntimeRejected as exc:
         if exc.diagnostic_code == "heartbeat_receipt_identity_conflict":
@@ -224,6 +227,79 @@ def _preserve_receipt_bound_replan_obligation(
     return dict(obligation)
 
 
+def _prepare_quota_replan_context(
+    *,
+    status_payload: dict[str, Any],
+    safe_goal_id: str,
+    agent_frontier_id: str | None,
+    item: dict[str, Any],
+    project_asset: dict[str, Any],
+    user_todo_summary: dict[str, Any] | None,
+    agent_todo_summary: dict[str, Any] | None,
+    authoritative_fallback_todo_items: FallbackTodoSource,
+    effective_available_capabilities: Any,
+    work_lane_contract: dict[str, Any] | None,
+    registered_agent_ids: list[str],
+    registry_goal: dict[str, Any],
+    agent_identity: dict[str, Any] | None,
+    receipt_bound_replan_obligation_id: str | None,
+    receipt_bound_replan_guard_scoped: bool,
+    receipt_bound_replay_phase: ReceiptBoundReplayPhase | None,
+    receipt_bound_guard_recorded_at: str | None,
+    receipt_bound_replan_review: Mapping[str, Any] | None,
+    replan_explore_reader: Callable[[str, str | None], Mapping[str, Any] | None] | None,
+) -> tuple[dict[str, Any], dict[str, Any] | None]:
+    """Project the current frontier and contextualize its receipt-bound review."""
+    replan_todo_items = select_planning_inventory_source_items(
+        item.get("agent_todos"),
+        project_asset.get("agent_todos") if project_asset else None,
+        include_terminal=True,
+    )
+    goal_frontier_context = build_goal_frontier_projection_context_from_status(
+        replan_explore_reader=replan_explore_reader,
+        goal_id=safe_goal_id,
+        agent_id=agent_frontier_id,
+        status_payload=status_payload,
+        item=item,
+        project_asset=project_asset,
+        user_todo_summary=user_todo_summary,
+        agent_todo_summary=agent_todo_summary,
+        agent_todo_source_items=replan_todo_items,
+        fallback_todo_source_items=authoritative_fallback_todo_items,
+        available_capabilities=effective_available_capabilities,
+        work_lane_contract=work_lane_contract,
+        neutral_replan_ack_classifications=AUTONOMOUS_REPLAN_ACK_NEUTRAL_CLASSIFICATIONS,
+        registered_agent_ids=registered_agent_ids,
+        goal_status=str(registry_goal.get("status") or ""),
+        agent_profile=_quota_agent_profile(agent_identity),
+        receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id,
+        receipt_bound_guard_recorded_at=receipt_bound_guard_recorded_at,
+    )
+    replan_obligation = (
+        None
+        if receipt_bound_replay_phase is ReceiptBoundReplayPhase.SETTLED
+        else _preserve_receipt_bound_replan_obligation(
+            goal_frontier_context.get("replan_obligation"),
+            receipt_bound_replan_obligation_id,
+            guard_scoped=receipt_bound_replan_guard_scoped,
+            agent_id=agent_frontier_id,
+            replay_phase=receipt_bound_replay_phase,
+            transition_candidates=goal_frontier_context.get("replan_transition_candidates"),
+            stride_strategy=goal_frontier_context["stride_strategy"],
+            receipt_review=receipt_bound_replan_review,
+        )
+    )
+    if replan_obligation and not replan_obligation.get("replan_context") and not replan_obligation.get("settlement_action_packet"):
+        replan_obligation = attach_receipt_replan_context(
+            replan_obligation, None, goal_id=safe_goal_id, agent_id=agent_frontier_id,
+            source_status=status_payload, stride_strategy=goal_frontier_context["stride_strategy"],
+            explore_reader=replan_explore_reader,
+            goal_acceptance_contract=(agent_todo_summary or {}).get("goal_acceptance_contract"),
+            todo_items=replan_todo_items,
+        )
+    return goal_frontier_context, replan_obligation
+
+
 def _same_todo_identity(left: dict[str, Any], right: dict[str, Any]) -> bool:
     left_id = str(left.get("todo_id") or "").strip()
     right_id = str(right.get("todo_id") or "").strip()
@@ -566,6 +642,7 @@ def _prepare_quota_should_run_item(
     authoritative_fallback_todo_items: FallbackTodoSource = None,
     replan_explore_reader: Callable[[str, str | None], Mapping[str, Any] | None] | None = None,
     receipt_bound_guard_recorded_at: str | None = None,
+    receipt_bound_replan_review: Mapping[str, Any] | None = None,
 ) -> _QuotaDecisionPreparation:
     quota = item.get("quota") if isinstance(item.get("quota"), dict) else {}
     state = str(quota.get("state") or "unknown")
@@ -870,42 +947,26 @@ def _prepare_quota_should_run_item(
         if isinstance(agent_identity, dict)
         else []
     )
-    goal_frontier_context = build_goal_frontier_projection_context_from_status(
-        replan_explore_reader=replan_explore_reader,
-        goal_id=safe_goal_id,
-        agent_id=agent_frontier_id,
+    goal_frontier_context, replan_obligation = _prepare_quota_replan_context(
         status_payload=status_payload,
+        safe_goal_id=safe_goal_id,
+        agent_frontier_id=agent_frontier_id,
         item=item,
         project_asset=project_asset,
         user_todo_summary=user_todo_summary,
         agent_todo_summary=agent_todo_summary,
-        agent_todo_source_items=select_planning_inventory_source_items(
-            item.get("agent_todos"),
-            project_asset.get("agent_todos") if project_asset else None,
-            include_terminal=True,
-        ),
-        fallback_todo_source_items=authoritative_fallback_todo_items,
-        available_capabilities=effective_available_capabilities,
+        authoritative_fallback_todo_items=authoritative_fallback_todo_items,
+        effective_available_capabilities=effective_available_capabilities,
         work_lane_contract=work_lane_contract,
-        neutral_replan_ack_classifications=AUTONOMOUS_REPLAN_ACK_NEUTRAL_CLASSIFICATIONS,
         registered_agent_ids=registered_agent_ids,
-        goal_status=str(registry_goal.get("status") or ""),
-        agent_profile=_quota_agent_profile(agent_identity),
+        registry_goal=registry_goal,
+        agent_identity=agent_identity,
         receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id,
+        receipt_bound_replan_guard_scoped=receipt_bound_replan_guard_scoped,
+        receipt_bound_replay_phase=receipt_bound_replay_phase,
         receipt_bound_guard_recorded_at=receipt_bound_guard_recorded_at,
-    )
-    replan_obligation = (
-        None
-        if receipt_bound_replay_phase is ReceiptBoundReplayPhase.SETTLED
-        else _preserve_receipt_bound_replan_obligation(
-            goal_frontier_context.get("replan_obligation"),
-            receipt_bound_replan_obligation_id,
-            guard_scoped=receipt_bound_replan_guard_scoped,
-            agent_id=agent_frontier_id,
-            replay_phase=receipt_bound_replay_phase,
-            transition_candidates=goal_frontier_context.get("replan_transition_candidates"),
-            stride_strategy=goal_frontier_context["stride_strategy"],
-        )
+        receipt_bound_replan_review=receipt_bound_replan_review,
+        replan_explore_reader=replan_explore_reader,
     )
     replan_scope = goal_frontier_context.get("replan_scope") or {}
     goal_frontier_projection = (

loopx/control_plane/work_items/replan_semantics.ts

@@ -261,6 +261,18 @@ function preserveReceiptBoundObligation(request: JsonObject): JsonObject {
       triggerKinds(current).some(kind => ACCEPTANCE_HOLD_TRIGGERS.has(kind))) {
     return {obligation: {...current, obligation_id: selected, selection_binding: "heartbeat_receipt"}};
   }
+  // Work-link maintenance can remove this frontier-only review before its
+  // writeback. The exact receipt retains its typed kind, never raw task prose
+  // or an old decision context. Other duties keep their canonical owners.
+  const review = object(request.receipt_review);
+  if (request.guard_scoped && review.kind === "goal_outcome_uncovered" &&
+      typeof review.stop_condition === "string" && review.stop_condition.trim() && review.stop_condition.length <= 240 &&
+      typeof request.agent_id === "string" && request.agent_id.trim()) {
+    return {obligation: {schema_version: "autonomous_replan_obligation_v0",
+      required: true, obligation_id: selected, agent_id: request.agent_id,
+      stop_condition: review.stop_condition,
+      triggers: [{kind: "goal_outcome_uncovered"}], selection_binding: "heartbeat_receipt"}};
+  }
   throw new EffectRuntimeRequestError(
     "heartbeat receipt settlement identity conflicts with the current autonomous replan obligation",
     "heartbeat_receipt_identity_conflict",
@@ -409,6 +421,25 @@ export function projectReplanSemantics(value: unknown): JsonObject {
   if (request.operation === "turn_transition") return projectTurnTransition(request);
   if (request.operation === "receipt_bound_obligation") return preserveReceiptBoundObligation(request);
   const obligation = requireJsonObject(request.obligation, "obligation");
+  if (request.operation === "receipt_review_source") {
+    const kinds = triggerKinds(obligation);
+    const triggers = Array.isArray(obligation.triggers) ? obligation.triggers : [];
+    // Only this plain frontier review has a complete scalar receipt source.
+    // Mixed, restricted, lifecycle and capability obligations must not lose
+    // their source-specific evidence or discharge rules through this route.
+    return {review_source: obligation.schema_version === "autonomous_replan_obligation_v0" &&
+      obligation.required === true && kinds.length > 0 &&
+      obligation.trigger_count === triggers.length &&
+      typeof request.agent_id === "string" && request.agent_id.trim() &&
+      obligation.agent_id === request.agent_id &&
+      kinds.every(kind => kind === "goal_outcome_uncovered") &&
+      obligation.resolution_mode == null && obligation.capability_guard == null &&
+      strings(obligation.satisfying_semantic_outcomes).length === 0 &&
+      obligation.progress_baseline == null && obligation.progress_window == null &&
+      triggers.every(trigger => object(trigger).progress_baseline == null && object(trigger).progress_window == null) &&
+      typeof obligation.stop_condition === "string" && obligation.stop_condition.trim() && obligation.stop_condition.length <= 240
+      ? {kind: "goal_outcome_uncovered", stop_condition: obligation.stop_condition} : null};
+  }
   if (request.operation === "transition_eligibility") {
     return {allowed: !requiresStrideDecision(obligation, request.stride_strategy)};
   }