From 47fe0f121bb3cb01c8d58fbead533019f440afba Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 17:58:07 +0800 Subject: [PATCH 1/2] fix(quota): keep outcome review identity after work refinement Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/quota/live_decision.py | 1 + loopx/control_plane/quota/settlement.py | 6 + loopx/control_plane/quota/settlement_cli.py | 20 +++ .../quota/settlement_readback.ts | 11 ++ loopx/control_plane/quota/should_run.py | 2 + .../control_plane/quota/should_run_prepare.py | 115 +++++++++--- .../work_items/replan_semantics.ts | 30 ++++ .../work_items/semantic_replan_writeback.py | 25 +++ loopx/quota.py | 2 + loopx/state_refresh.py | 1 + .../test_quota_settlement_cli.py | 163 ++++++++++++++++++ .../quota_settlement_readback.test.ts | 14 ++ .../control_plane_ts/replan_semantics.test.ts | 46 +++++ 13 files changed, 409 insertions(+), 27 deletions(-) diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index c149bc11b6..8518f03ad5 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/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)), } diff --git a/loopx/control_plane/quota/settlement.py b/loopx/control_plane/quota/settlement.py index dbbc496b0f..4b438c521a 100644 --- a/loopx/control_plane/quota/settlement.py +++ b/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 diff --git a/loopx/control_plane/quota/settlement_cli.py b/loopx/control_plane/quota/settlement_cli.py index a6ab19c5cf..bf3e2cd458 100644 --- a/loopx/control_plane/quota/settlement_cli.py +++ b/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 diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index 7cd43d8e5f..d83d3a7d54 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/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, diff --git a/loopx/control_plane/quota/should_run.py b/loopx/control_plane/quota/should_run.py index 21ebc67e26..8de194bc8a 100644 --- a/loopx/control_plane/quota/should_run.py +++ b/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( diff --git a/loopx/control_plane/quota/should_run_prepare.py b/loopx/control_plane/quota/should_run_prepare.py index 2b5355ef5e..3b4fd8036e 100644 --- a/loopx/control_plane/quota/should_run_prepare.py +++ b/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 = ( diff --git a/loopx/control_plane/work_items/replan_semantics.ts b/loopx/control_plane/work_items/replan_semantics.ts index b95b0b1817..e8a62227fe 100644 --- a/loopx/control_plane/work_items/replan_semantics.ts +++ b/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,24 @@ 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 && + 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)}; } diff --git a/loopx/control_plane/work_items/semantic_replan_writeback.py b/loopx/control_plane/work_items/semantic_replan_writeback.py index 8ea408326c..31ed43607f 100644 --- a/loopx/control_plane/work_items/semantic_replan_writeback.py +++ b/loopx/control_plane/work_items/semantic_replan_writeback.py @@ -32,6 +32,8 @@ project_todo_lifecycle_settlement_reentry, ) from .progress_observation import guarded_replan_transition_delta, semantic_delta_from_writeback +from .lifecycle_recovery import attach_receipt_replan_context +from .replan_semantics_codec import project_replan_semantics from ..effect_runtime import effect_runtime_request_scope from .repair_delta import ( build_repair_delta_contract, @@ -228,6 +230,7 @@ def qualify_replan_writeback( capability_evidence: CapabilityReplanEvidence | None = None, guard_capability: Mapping[str, Any] | None = None, receipt_bound_guard_recorded_at: str | None = None, + receipt_bound_replan_review: Mapping[str, Any] | None = None, replan_source_route: Mapping[str, Any] | None = None, replan_explore_reader: Callable[[str, str | None], Mapping[str, Any] | None] | None = None, ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: @@ -371,6 +374,24 @@ def qualify_replan_writeback( context.get("replan_transition_candidates", []) if isinstance(candidate.get("obligation"), dict) and candidate["obligation"].get("obligation_id") == guard_semantic_replan_obligation_id), None) + if guard_scoped and receipt_bound_replan_review is not None: + obligation = project_replan_semantics({ + "operation": "receipt_bound_obligation", "guard_scoped": True, + "selected_obligation_id": guard_semantic_replan_obligation_id, + "agent_id": safe_agent_id, "replay_phase": "open", + "current_obligation": obligation, + "receipt_review": receipt_bound_replan_review, + "transition_candidates": context.get("replan_transition_candidates", []), + "stride_strategy": context["stride_strategy"], + })["obligation"] + if obligation and not obligation.get("replan_context") and not obligation.get("settlement_action_packet"): + obligation = attach_receipt_replan_context( + obligation, None, goal_id=goal_id, agent_id=safe_agent_id, + source_status=status_payload, stride_strategy=context["stride_strategy"], + explore_reader=replan_explore_reader, + goal_acceptance_contract=(agent_todos or {}).get("goal_acceptance_contract"), + todo_items=agent_todo_source_items, + ) if not obligation: # A validated Todo transition can discharge the read-model obligation # before refresh. Preserve that evidence in this run so periodic review @@ -426,6 +447,7 @@ def enforce_open_replan_writeback( guard_scoped: bool = False, guard_semantic_replan_obligation_id: str | None = None, receipt_bound_guard_recorded_at: str | None = None, + receipt_bound_replan_review: Mapping[str, Any] | None = None, todo_fields: dict[str, Any] | None = None, external_progress_review: Mapping[str, Any] | None = None, capability_evidence: CapabilityReplanEvidence | None = None, @@ -461,6 +483,7 @@ def enforce_open_replan_writeback( capability_evidence=capability_evidence, guard_capability=guard_capability, receipt_bound_guard_recorded_at=receipt_bound_guard_recorded_at, + receipt_bound_replan_review=receipt_bound_replan_review, replan_source_route=replan_source_route, replan_explore_reader=replan_explore_reader, ) if not obligation: @@ -508,6 +531,7 @@ def qualify_refresh_replan_writeback( settlement_guard_scoped: bool, settlement_guard_semantic_replan_obligation_id: str | None, receipt_bound_guard_recorded_at: str | None = None, + receipt_bound_replan_review: Mapping[str, Any] | None = None, newest_first_runs: list[dict[str, Any]], state_text: str, goal_id: str, @@ -593,6 +617,7 @@ def qualify_refresh_replan_writeback( settlement_guard_semantic_replan_obligation_id ), receipt_bound_guard_recorded_at=receipt_bound_guard_recorded_at, + receipt_bound_replan_review=receipt_bound_replan_review, replan_source_route=replan_source_route, replan_explore_reader=replan_explore_reader, ) if semantic_delta: diff --git a/loopx/quota.py b/loopx/quota.py index beb943eff6..ed4b56de23 100644 --- a/loopx/quota.py +++ b/loopx/quota.py @@ -930,6 +930,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, @@ -965,6 +966,7 @@ def read_explore_context(read_goal_id: str, read_agent_id: str | None): receipt_bound_replan_guard_scoped=receipt_bound_replan_guard_scoped, receipt_bound_successor_replan=receipt_bound_successor_replan, receipt_bound_guard_recorded_at=receipt_bound_guard_recorded_at, + receipt_bound_replan_review=receipt_bound_replan_review, turn_instance_id=turn_instance_id, runtime_root=runtime_root, workspace_path=workspace_path, diff --git a/loopx/state_refresh.py b/loopx/state_refresh.py index ecfba0ec0d..abf4d99dd9 100644 --- a/loopx/state_refresh.py +++ b/loopx/state_refresh.py @@ -1120,6 +1120,7 @@ def read_explore_context(read_goal_id: str, read_agent_id: str | None): settlement_replan_guard.get("scope") == "turn_guard" ), receipt_bound_guard_recorded_at=(settlement_readback.heartbeat_receipt or {}).get("recorded_at") if settlement_readback else None, + receipt_bound_replan_review=settlement_replan_guard.get("selected_review_source"), settlement_guard_semantic_replan_obligation_id=( settlement_replan_guard.get("selected_obligation_id") if isinstance( diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index c9b9d84991..523f0d985a 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -7207,6 +7207,169 @@ def test_runtime_capability_reentry_preserves_receipt_bound_autonomous_replan( assert _heartbeat_receipt_count(runtime, turn_instance_id) == 1 +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("stride", [False, True]) +def test_replan_reentry_after_outcome_work_refinement_keeps_original_turn( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, stride: bool, +) -> None: + """Refreshing cumulative requirements must not rebind an admitted review.""" + if provider == "sqlite": + from canonical_authority_fixture import isolate_sqlite_runtime + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry_path = _write_fixture(tmp_path) + state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" + state.write_text(state.read_text().replace("status: active-read-only", "status: active")) + registry = json.loads(registry_path.read_text()) + registry["goals"][0]["status"] = "active" + registry["goals"][0]["coordination"]["registered_agents"].append("another-agent") + registry_path.write_text(json.dumps(registry)) + _initialize_fixture_authority(project, runtime, provider=provider) + + def call(*args: str, expected_code: int = 0) -> dict[str, Any]: + code, result = _run_cli(registry_path, runtime, *args) + assert code == expected_code, result + return result + + def refine(document: dict[str, Any], operation: str) -> dict[str, Any]: + current = call("goal-acceptance", "inspect", "--goal-id", GOAL_ID) + path = tmp_path / f"{operation}.json" + path.write_text(json.dumps(document), encoding="utf-8") + return call("goal-acceptance", "refine", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--document", str(path), "--operation-id", operation, + "--expected-provider-revision", current["provider_revision"], "--execute") + + if stride: + call("configure-goal", "--goal-id", GOAL_ID, + "--execution-stride-strategy", "evidence_conditioned_v1", "--execute") + refine({"outcome_id": "delivery", "owner_agent_id": AGENT_ID, + "objective": "Deliver the requested artifact", + "requirements": [{"id": "original", "description": "Retain the original evidence"}], + "todo_ids": [TODO_ID], + "source": {"ref": "message:original", "summary": "The owner requested the artifact"}}, + "declare-outcome") + call("todo", "complete", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--todo-id", TODO_ID, "--no-follow-up", "--evidence", "artifact:initial-delivery") + turn = "turn-replan-outcome-refinement" + guard_args = ("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--turn-instance-id", turn) + first = call(*guard_args) + assert first["decision"] == "autonomous_replan_required", first + identity = first["heartbeat_receipt"]["settlement_identity"] + assert identity["binding_kind"] == "autonomous_replan" + receipt = _heartbeat_receipt_events(runtime, turn)[0] + + successor = call("todo", "add", "--goal-id", GOAL_ID, "--role", "agent", + "--text", "Read back the original owner's artifact when available", + "--claimed-by", AGENT_ID, "--action-kind", "validate", + "--resume-when", "resume_at:2999-01-01T00:00:00Z", + "--operation-id", "deferred-result-readback")["todo_id"] + updated = refine({"outcome_id": "delivery", "todo_ids": [successor]}, "link-existing-work") + expected = updated["goal_acceptance_contract"]["outcome_requirements"] + replay = call(*guard_args) + assert replay["heartbeat_receipt"]["settlement_identity"] == identity + assert replay["selected_todo"] is None + context = replay["interaction_contract"]["agent_channel"]["work_context"] + acceptance = next(row["content"] for row in context["sources"] + if row.get("source") == "goal_acceptance") + assert acceptance["goal_acceptance_contract"]["outcome_requirements"] == expected + assert _heartbeat_receipt_count(runtime, turn) == 1 + assert receipt["details"]["semantic_replan_review_kind"] == "goal_outcome_uncovered" + assert receipt["details"]["semantic_replan_review_stop_condition"] == first["autonomous_replan_obligation"]["stop_condition"] + assert replay["autonomous_replan_obligation"]["stop_condition"] == first["autonomous_replan_obligation"]["stop_condition"] + assert "semantic_replan_obligation" not in receipt["details"] + assert receipt["boundary"]["raw_task_text_recorded"] is False + + # A further owner requirement must be read in the original review, while + # the durable receipt and waiting work retain their exact identities. + updated = refine({"outcome_id": "delivery", + "requirements": [{"id": "recovery", "description": "Validate recovery without losing the original artifact"}], + "source": {"ref": "message:recovery", "summary": "The owner added recovery acceptance"}}, + "add-recovery-requirement") + expected = updated["goal_acceptance_contract"]["outcome_requirements"] + fresh = call(*guard_args) + assert fresh["heartbeat_receipt"]["settlement_identity"] == identity + current_context = fresh["autonomous_replan_obligation"]["replan_context"] + assert current_context["context_id"] != first["autonomous_replan_obligation"]["replan_context"]["context_id"] + sources = fresh["interaction_contract"]["agent_channel"]["work_context"]["sources"] + acceptance = next(row["content"] for row in sources if row.get("source") == "goal_acceptance") + assert acceptance["goal_acceptance_contract"]["outcome_requirements"] == expected + original = expected["outcomes"][0] + assert {row["id"] for row in original["requirements"]} == {"original", "recovery"} + assert original["sources"][0]["ref"] == "message:original" + assert _heartbeat_receipt_events(runtime, turn) == [receipt] + + conflict = call(*guard_args, "--todo-id", successor, expected_code=1) + assert conflict["ok"] is False + assert _heartbeat_receipt_events(runtime, turn) == [receipt] + # Wrong scopes cannot borrow this receipt to write back or debit. + spend = ["quota", "spend-slot", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--replan-obligation-id", identity["replan_obligation_id"], "--turn-instance-id", turn, + "--slots", "1", "--source", "heartbeat", "--execute"] + for flag, wrong in [("--goal-id", "another-goal"), ("--agent-id", "another-agent"), + ("--turn-instance-id", "another-turn"), + ("--replan-obligation-id", "replan-2222222222222222")]: + invalid = list(spend) + invalid[invalid.index(flag) + 1] = wrong + call(*invalid, expected_code=1) + assert _spend_run_count(runtime) == 0 + + from tests.control_plane.test_outcome_replan_composition_cli import _review_packet + + control = fresh["replan_action_packet"].get("stride_control") + decision = tmp_path / "current-review.json" + decision.write_text(json.dumps(_review_packet(control))) + command = fresh["interaction_contract"]["cli_channel"]["next_cli_actions"][0] + for old, new in [("", "advanced"), + ("", "review-current-artifact"), ("", "retain-original-delivery"), + ("", "recovery-check"), ("", "evidence:current-artifact")]: + command = command.replace(old, new) + refresh = (*_projected_cli_args(command, turn_instance_id=turn), + "--agent-vision-json", str(decision), "--no-global-sync", "--suppress-external-sinks") + empty = call("refresh-state", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--replan-obligation-id", identity["replan_obligation_id"], "--turn-instance-id", turn, + "--classification", "bounded_replan_progress", "--delivery-batch-scale", "single_surface", + "--delivery-outcome", "outcome_progress", "--agent-vision-json", str(decision), + "--no-global-sync", "--suppress-external-sinks", expected_code=1) + assert empty["appended"] is False + assert _spend_run_count(runtime) == 0 + written = call(*refresh) + assert written["autonomous_replan_ack"]["semantic_delta"]["accepted"] is True + owed = _projected_cli_args(written["settlement_owed"]["command"], turn_instance_id=turn) + call(*owed) + call(*owed) + assert _spend_run_count(runtime) == 1 + settled = call(*guard_args) + assert settled["effective_action"] == "heartbeat_settled_skip" + assert settled["heartbeat_receipt"]["settlement_identity"] == identity + assert _heartbeat_receipt_events(runtime, turn) == [receipt] + after = call("goal-acceptance", "inspect", "--goal-id", GOAL_ID) + assert after["goal_acceptance_contract"]["outcome_requirements"] == expected + waiting = call("todo", "list", "--goal-id", GOAL_ID, "--todo-id", successor)["todo"] + assert waiting["status"] == "open" + assert waiting["resume_when"] == "resume_at:2999-01-01T00:00:00Z" + + +def test_receipt_review_does_not_persist_private_or_lossy_stop_clauses() -> None: + from argparse import Namespace + from loopx.control_plane.quota.settlement_cli import quota_rollout_details + + obligation = {"schema_version": "autonomous_replan_obligation_v0", "required": True, + "agent_id": AGENT_ID, "obligation_id": "replan-1111111111111111", + "triggers": [{"kind": "goal_outcome_uncovered"}]} + for stop, retained in [("Stop at owner-only decisions.", True), + ("Stop if Authorization: example is required.", False), + ("Stop before\nowner-only decisions.", False), ("s" * 241, False)]: + details = quota_rollout_details({"autonomous_replan_obligation": {**obligation, "stop_condition": stop}, + "replan_action_packet": {"obligation_id": obligation["obligation_id"]}}, + Namespace(quota_command="should-run", agent_id=AGENT_ID), todo_id=None) + assert ("semantic_replan_review_kind" in details) is retained + if retained: + assert details["semantic_replan_review_stop_condition"] == stop + else: + assert "semantic_replan_review_stop_condition" not in details + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_peer_refresh_rejects_implicit_canonical_workspace_before_writeback( tmp_path: Path, diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts index ae400ee336..eeabe3dc20 100644 --- a/tests/control_plane_ts/quota_settlement_readback.test.ts +++ b/tests/control_plane_ts/quota_settlement_readback.test.ts @@ -87,6 +87,20 @@ test("capability guard survives scalar receipt transport and rejects missing sco assert.throws(() => projectSemanticReplanGuard({...details, semantic_replan_frontier_revision: undefined}), /capability guard is malformed/); }); +test("recorded review source retains its stop clause under the exact selected guard", () => { + const details = {semantic_replan_obligation_id: "replan-0000000000000001", + semantic_replan_review_kind: "goal_outcome_uncovered", + semantic_replan_review_stop_condition: "Stop at owner-only decisions."}; + assert.deepEqual(projectSemanticReplanGuard(details).selected_review_source, { + kind: details.semantic_replan_review_kind, stop_condition: details.semantic_replan_review_stop_condition}); + for (const patch of [{semantic_replan_obligation_id: ""}, + {semantic_replan_obligation_id: undefined}, {semantic_replan_review_kind: undefined}, + {semantic_replan_review_stop_condition: undefined}, {semantic_replan_review_stop_condition: "s".repeat(241)}, + {semantic_replan_capability_id: "explore"}]) { + assert.throws(() => projectSemanticReplanGuard({...details, ...patch})); + } +}); + test("capability retirement verifies its exact receipt and never erases a committed debit", async () => { const obligation = "replan-0000000000000001"; const guard = {schema_version: "semantic_replan_capability_guard_v0", capability_id: "explore", diff --git a/tests/control_plane_ts/replan_semantics.test.ts b/tests/control_plane_ts/replan_semantics.test.ts index 3f934ce3d7..0eb85867f4 100644 --- a/tests/control_plane_ts/replan_semantics.test.ts +++ b/tests/control_plane_ts/replan_semantics.test.ts @@ -124,6 +124,52 @@ test("verified original writeback projects only its outstanding debit even after current_obligation: null, transition_candidates: []}), /malformed/); }); +test("only a plain outcome review has a scalar receipt source", () => { + const source = {schema_version: "autonomous_replan_obligation_v0", required: true, + obligation_id: "replan-1111111111111111", agent_id: "reviewer", + stop_condition: "Stop at owner-only decisions.", + triggers: [{kind: "goal_outcome_uncovered", text: "Private Goal prose stays outside telemetry"}], + replan_context: {context_id: "old-acceptance-basis"}}; + const request = {operation: "receipt_review_source", agent_id: "reviewer", obligation: source}; + assert.deepEqual(projectReplanSemantics(request), {review_source: { + kind: "goal_outcome_uncovered", stop_condition: source.stop_condition}}); + for (const patch of [{required: false}, {schema_version: "unknown"}, {agent_id: "another-agent"}, + {triggers: []}, {triggers: [{kind: "periodic_review_due"}]}, + {triggers: [{kind: "goal_outcome_uncovered"}, {kind: "external_progress_review_drift"}]}, + {satisfying_semantic_outcomes: ["new_runnable_successor"]}, + {progress_baseline: {}}, {progress_window: []}, + {triggers: [{kind: "goal_outcome_uncovered", progress_baseline: {}}]}, + {triggers: [{kind: "goal_outcome_uncovered", progress_window: []}]}, + {stop_condition: null}, {stop_condition: "s".repeat(241)}, + {resolution_mode: "todo_lifecycle_settlement"}, + {capability_guard: {schema_version: "semantic_replan_capability_guard_v0"}}]) { + assert.deepEqual(projectReplanSemantics({...request, obligation: {...source, ...patch}}), {review_source: null}); + } +}); + +test("a recorded outcome review survives work-link refinement without cached context or discharge", () => { + const id = "replan-1111111111111111"; + const request = {operation: "receipt_bound_obligation", guard_scoped: true, + replay_phase: "open", selected_obligation_id: id, current_obligation: null, + transition_candidates: [], agent_id: "reviewer", receipt_review: { + kind: "goal_outcome_uncovered", stop_condition: "Stop at owner-only decisions."}}; + const recovered = projectReplanSemantics(request).obligation as JsonObject; + assert.equal(recovered.obligation_id, id); + assert.equal(recovered.agent_id, "reviewer"); + assert.equal(recovered.stop_condition, request.receipt_review.stop_condition); + assert.deepEqual(recovered.triggers, [{kind: "goal_outcome_uncovered"}]); + assert.equal(recovered.selection_binding, "heartbeat_receipt"); + assert.equal(recovered.replan_context, undefined); + assert.equal(recovered.settlement_action_packet, undefined); + assert.equal(projectReplanSemantics({operation: "qualify", obligation: recovered}).accepted, false); + for (const invalid of [{guard_scoped: false}, {agent_id: null}, + {receipt_review: null}, {receipt_review: {kind: "periodic_review_due"}}, + {receipt_review: {kind: "goal_outcome_uncovered"}}]) { + assert.throws(() => projectReplanSemantics({...request, ...invalid}), /identity conflicts/); + } + assert.equal(projectReplanSemantics({...request, selected_obligation_id: null}).obligation, null); +}); + test("obligation source governs both authoring projection and semantic discharge", () => { const projection = projectReplanSemantics({operation: "requirements", obligation}); assert.match(String(projection.cli_semantic_args), /--agent-vision-json/); From d0bfe2f40651cd279d426d5c895bfec59da85485 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 18:29:28 +0800 Subject: [PATCH 2/2] fix(quota): require a complete receipt review source Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/work_items/replan_semantics.ts | 1 + tests/control_plane/test_quota_settlement_cli.py | 2 +- tests/control_plane_ts/replan_semantics.test.ts | 5 +++-- 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/loopx/control_plane/work_items/replan_semantics.ts b/loopx/control_plane/work_items/replan_semantics.ts index e8a62227fe..6ebf44aaaf 100644 --- a/loopx/control_plane/work_items/replan_semantics.ts +++ b/loopx/control_plane/work_items/replan_semantics.ts @@ -429,6 +429,7 @@ export function projectReplanSemantics(value: unknown): JsonObject { // 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") && diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 523f0d985a..9a5eacf70c 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -7356,7 +7356,7 @@ def test_receipt_review_does_not_persist_private_or_lossy_stop_clauses() -> None obligation = {"schema_version": "autonomous_replan_obligation_v0", "required": True, "agent_id": AGENT_ID, "obligation_id": "replan-1111111111111111", - "triggers": [{"kind": "goal_outcome_uncovered"}]} + "trigger_count": 1, "triggers": [{"kind": "goal_outcome_uncovered"}]} for stop, retained in [("Stop at owner-only decisions.", True), ("Stop if Authorization: example is required.", False), ("Stop before\nowner-only decisions.", False), ("s" * 241, False)]: diff --git a/tests/control_plane_ts/replan_semantics.test.ts b/tests/control_plane_ts/replan_semantics.test.ts index 0eb85867f4..7183df3877 100644 --- a/tests/control_plane_ts/replan_semantics.test.ts +++ b/tests/control_plane_ts/replan_semantics.test.ts @@ -126,7 +126,7 @@ test("verified original writeback projects only its outstanding debit even after test("only a plain outcome review has a scalar receipt source", () => { const source = {schema_version: "autonomous_replan_obligation_v0", required: true, - obligation_id: "replan-1111111111111111", agent_id: "reviewer", + obligation_id: "replan-1111111111111111", agent_id: "reviewer", trigger_count: 1, stop_condition: "Stop at owner-only decisions.", triggers: [{kind: "goal_outcome_uncovered", text: "Private Goal prose stays outside telemetry"}], replan_context: {context_id: "old-acceptance-basis"}}; @@ -134,7 +134,8 @@ test("only a plain outcome review has a scalar receipt source", () => { assert.deepEqual(projectReplanSemantics(request), {review_source: { kind: "goal_outcome_uncovered", stop_condition: source.stop_condition}}); for (const patch of [{required: false}, {schema_version: "unknown"}, {agent_id: "another-agent"}, - {triggers: []}, {triggers: [{kind: "periodic_review_due"}]}, + {triggers: []}, {trigger_count: 2}, {trigger_count: undefined}, + {triggers: [{kind: "periodic_review_due"}]}, {triggers: [{kind: "goal_outcome_uncovered"}, {kind: "external_progress_review_drift"}]}, {satisfying_semantic_outcomes: ["new_runnable_successor"]}, {progress_baseline: {}}, {progress_window: []},