diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md index 993a59655..397639278 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md @@ -410,6 +410,13 @@ Stop or narrow M7 when any kill criterion holds: `work_lane_contract`, and `scheduler_hint`. - focused tests and docs that pin the lens. +The quota closeout adapter now consumes receipt-derived settlement progress +from the TS readback instead of independently treating a spend run as settled. +Normal refresh, replay and spend responses share that projection; executable +commands bind the original actor and route. Receipt repair reuses the existing +idempotent writer. This is a bounded M7.4 adoption with no shared executor or +new authority store; it does not certify terminal Todo or Goal acceptance. + ### What Is Missing - A generic shared executor is deliberately absent. The current adapters share diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md index 92cce4a71..f531e8ba1 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md @@ -292,6 +292,8 @@ M7 只有在至少产生一个下列最终 effect 时才有理由存在： - around 语义编码在 `capability_gate`、`interaction_contract`、`work_lane_contract` 和 `scheduler_hint` 中。 - 聚焦测试和文档固定该视角。 +配额收尾适配现在消费 TS readback 按回执归约的结算进度，不再独立把存在 spend run 当作已结算。正常刷新、重放和扣款响应共享该投影；可执行命令绑定原 Agent 和路由。补回执复用现有幂等 writer。这是有界的 M7.4 接入，没有增加共享 executor 或 authority store，也不证明 Todo 终态或 Goal 验收完成。 + ### 还缺什么 - 通用共享 executor 被有意保留为空。当前 adapter 共享 plan/receipt algebra，却拥有不同的执行边界，因此 M7.3 应以 no-follow-up 关闭，而不是用推测性 framework 填充。 diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 7cb7f305d..23a68b222 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1057,6 +1057,17 @@ or every T3 consumer. Python retains codecs, IO and the documented legacy route prose hint until its remaining writers emit explicit replan flags; no new capability/provider or parallel business authority is introduced. +The quota admission/settlement consumer now resolves explicit Todo selection +from the complete unified Todo reader before display compaction. It removes the +direct Markdown candidate append while preserving the pre-promotion event +adapter. Post-promotion empty or unavailable authority cannot revive display +rows. Settlement progress is reduced from the existing TS receipt chain; +Python renders identity-complete commands and both JSON/Markdown readbacks. +The existing idempotent writer repairs a missing spend receipt without a second +debit. This closes the demonstrated T3 consumer gap, not D1–D3, provider +promotion, or the remaining Python transaction adapters. See the +[operating contract](../../quota-allocation.md#receipt-backed-settlement-progress). + **T4 — collect full-writer retirement after durability cutover.** - The 2026-09-19 command audit retires two already-typed but unconsumed diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 30c7035a1..08efd9141 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -792,6 +792,8 @@ PostgreSQL 读取使用同一个 repeatable-read snapshot，并发提交在下 历史。合法结果 schema、File/NoKV 持久字节、请求身份及版本算法保持兼容。这支持 T3/D1 reader，未完成全部 Todo writer、retention/compaction 或 promotion。 +配额准入与结算消费者现在从统一 Todo reader 读取完整来源，在显示压缩前解析显式 Todo 选择。它删除直接追加 Markdown 候选的路径，保留 promote 前的事件适配；promote 后权威为空或不可读都不能复活展示行。结算进度由现有 TS 回执链归约，Python 负责完整身份命令及 JSON/Markdown 展示。现有幂等 writer 可补齐缺失的 spend 回执而不再次扣款。这关闭已复现的 T3 消费者缺口，不代表 D1–D3、provider promotion 或剩余 Python 事务适配已完成。操作语义见[结算进度契约](../../quota-allocation.md#receipt-backed-settlement-progress)。 + **T4 — durable cutover 后兑现完整 writer 删除。** - 2026-09-19 命令审计退役两条已经 typed、但没有实际消费者的执行面： diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 36d607d49..0d4df49b7 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -129,6 +129,26 @@ an outer caller still owns repeated execution. A completed-Todo threshold, a per-Turn path declaration, and a same-Todo continuation budget are different controls. +### Receipt-backed settlement progress + +Turn-scoped `refresh-state` and `quota spend-slot` expose +`settlement_progress` from the TypeScript receipt readback. The states are +`identity_required`, `writeback_required`, `writeback_receipt_required`, +`spend_required`, `spend_receipt_required`, and `settled`. A durable run without +its matching receipt is incomplete. `settled` certifies this writeback/spend +chain; Todo completion and Goal acceptance retain their separate checks. + +After verified writeback, `settlement_owed.command` carries the original Goal, +Agent, Todo or replan obligation, Turn, registry/runtime route and spend source. +Execute it unchanged. In `spend_receipt_required`, the same idempotent spend +writer restores the receipt without another debit. Refresh and recovery never +spend automatically. JSON and normal/recovery Markdown expose the same step. +Rejected recovery reports observed progress without offering a spend command. +The raw Python refresh API returns `writeback_receipt_required` until its CLI +caller appends the refresh receipt and rereads the chain; it must not offer a +spend command before that point. Older guards without a persisted spend source +retain the existing heartbeat default. + ## Minimal Contract The compact status shape can start with a small object: @@ -494,6 +514,14 @@ hard lane or returns a refreshed portfolio. No settlement plan is exposed before reentry, and a previously bound receipt cannot be retargeted. `recommended_action` retains the human-readable rejection or deferral guidance; the executable recovery command lives in `next_cli_actions` and `agent_channel.primary_action`. +Fresh explicit selection reads the complete Todo source through the same reader +as `todo list`, before display or Agent-lane compaction. Before shared-authority +promotion, that reader retains Markdown plus the existing event overlay. After +promotion, it reads the selected canonical provider, including authoritative +empty results; missing/stale display and provider failure never authorize a +Markdown fallback. The guard does not append a second Markdown candidate list. +Historical receipt-bound recovery remains separate from new work admission. + A single-candidate response keeps the direct execution path and does not add an extra selection round trip. diff --git a/loopx/cli_commands/project_lifecycle_refresh_state.py b/loopx/cli_commands/project_lifecycle_refresh_state.py index 3a6c4ff63..55210a1fc 100644 --- a/loopx/cli_commands/project_lifecycle_refresh_state.py +++ b/loopx/cli_commands/project_lifecycle_refresh_state.py @@ -25,6 +25,7 @@ GOAL_VISION_ADVANCEMENT_POLICY_CHOICES, ) from ..control_plane.quota.settlement import ( + attach_settlement_progress, read_heartbeat_settlement, settlement_result_payload, ) @@ -618,6 +619,9 @@ def handle_refresh_state_command( "exact settlement readback unexpectedly returned not-found" ) settlement_result = settlement_readback.delivery + attach_settlement_progress( + payload, settlement_readback, registry_path=registry_path, runtime_root=runtime_root, + ) payload["settlement_result"] = settlement_result_payload( settlement_result ) diff --git a/loopx/control_plane/quota/effect_program.py b/loopx/control_plane/quota/effect_program.py index 24d2a74b3..488c9a956 100644 --- a/loopx/control_plane/quota/effect_program.py +++ b/loopx/control_plane/quota/effect_program.py @@ -59,6 +59,22 @@ def _quoted_turn_ref(turn_instance_id_ref: str) -> str: return shlex.quote(turn_instance_id_ref) +def _settlement_actor_args(arguments: str, agent_id: str) -> str: + """Identity owns the actor; optional host arguments cannot omit or retarget it.""" + tokens = shlex.split(arguments) + actors = [] + for index, token in enumerate(tokens): + if token == "--agent-id": + actors.append(tokens[index + 1] if index + 1 < len(tokens) else None) + elif token.startswith("--agent-id="): + actors.append(token.partition("=")[2]) + if actors and actors != [agent_id]: + raise ValueError("settlement command actor must match its exact identity") + if actors: + return arguments if arguments[:1].isspace() else f" {arguments}" + return f"{arguments} --agent-id {shlex.quote(agent_id)}" + + def build_codex_app_settlement_plan( *, goal_id: str, @@ -116,6 +132,8 @@ def build_turn_scoped_cli_settlement_plan( turn_instance_id=turn_instance_id, replan_obligation_id=replan_obligation_id, ) + scoped_cli_args = _settlement_actor_args(scoped_cli_args, identity.agent_id) + lifecycle_actor_args = _settlement_actor_args(lifecycle_actor_args, identity.agent_id) quoted_turn = _quoted_turn_ref(turn_instance_id) binding_arg = ( f" --todo-id {shlex.quote(todo_id)}" diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index 25031f96f..8e7d67980 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/loopx/control_plane/quota/live_decision.py @@ -541,6 +541,23 @@ def build_live_quota_should_run_decision( fresh_operator_inbox_read = _fresh_operator_inbox_read_required( turn_start_hook_dispatch ) + if requested_action_todo_id and not receipt_bound_todo_id: + # Candidate discovery and admission use the same provider-first reader. + # Keep the complete snapshot internal; presentation is bounded later. + from ...todos import list_goal_todos + + source = list_goal_todos( + registry_path=registry_path, runtime_root_arg=str(runtime_root), goal_id=goal_id, + ) + todo_fields = {key: source[key] for key in ("user_todos", "agent_todos")} + queue = decision_status_payload.get("attention_queue") or {} + decision_status_payload["attention_queue"] = { + **queue, + "items": [ + {**item, **todo_fields} if item.get("goal_id") == goal_id else item + for item in queue.get("items") or [] + ], + } payload = build_quota_should_run( decision_status_payload, goal_id=goal_id, diff --git a/loopx/control_plane/quota/refresh_external_delivery.py b/loopx/control_plane/quota/refresh_external_delivery.py index fa1e1d197..69cae8c31 100644 --- a/loopx/control_plane/quota/refresh_external_delivery.py +++ b/loopx/control_plane/quota/refresh_external_delivery.py @@ -3,70 +3,7 @@ from typing import Any from ...rollout_event_log import append_rollout_event, build_rollout_event, rollout_event_log_path -from .effect_program import ( - SettlementStepKind, - build_turn_scoped_cli_settlement_plan, -) -from .settlement import ( - QuotaSettlementReadback, - settlement_result_payload, - settlement_step_command, -) - -TURN_SETTLEMENT_OWED_SCHEMA_VERSION = "turn_settlement_owed_v0" - - -def turn_settlement_owed( - readback: QuotaSettlementReadback, - *, - command_prefix: str = "loopx", -) -> dict[str, Any] | None: - """Name the settlement step a committed writeback still owes. - - A turn-scoped writeback appends the accountable run, but the Turn is only - settled once the quota spend for the same effect identity exists. Reporting - the writeback as finished hid that debt until the next wake raised - ``unsettled_host_turn_recovery``, so the owed step is named here together - with the exact command that pays it under the same identity. - """ - - identity = readback.identity.value - if identity is None or readback.spend_run is not None: - return None - owed: dict[str, Any] = { - "schema_version": TURN_SETTLEMENT_OWED_SCHEMA_VERSION, - "kind": "quota_spend", - "effect_id": identity.effect_id, - "goal_id": identity.goal_id, - "agent_id": identity.agent_id, - "todo_id": identity.todo_id, - "replan_obligation_id": identity.replan_obligation_id, - "turn_instance_id": identity.turn_instance_id, - "reason": ( - "the writeback is committed but this Turn still owes its quota spend; " - "spend once for the same settlement identity and do not treat the " - "writeback alone as closeout" - ), - "recovery_does_not_spend": True, - } - if bool(identity.todo_id) == bool(identity.replan_obligation_id): - # A turn-scoped identity carries exactly one work item. Without it the - # debt is still owed, but no exact command can be offered. - return owed - plan = build_turn_scoped_cli_settlement_plan( - goal_id=identity.goal_id, - agent_id=identity.agent_id, - command_prefix=command_prefix, - todo_id=identity.todo_id, - replan_obligation_id=identity.replan_obligation_id, - scoped_cli_args="", - lifecycle_actor_args="", - turn_instance_id=identity.turn_instance_id, - ) - command = settlement_step_command(plan.as_dict(), SettlementStepKind.QUOTA_SPEND) - if command: - owed["command"] = command - return owed +from .settlement import QuotaSettlementReadback, attach_settlement_progress, settlement_result_payload def finish_external_delivery_refresh( @@ -80,13 +17,6 @@ def finish_external_delivery_refresh( raise RuntimeError("TypeScript refresh external delivery result missing or invalid") payload["external_delivery"] = {k: v for k, v in plan.items() if k != "transition"} payload["external_sink_delivery_authorized"] = plan["authorized"] is True - if payload.get("ok") and not dry_run: - owed = turn_settlement_owed(readback) - if owed is not None: - # The committed writeback is not a closeout: name the step and the - # exact command this Turn still owes so the caller cannot mistake - # one for the other. - payload["settlement_owed"] = owed transition = plan.get("transition") if payload.get("ok") and transition and not dry_run: identity = readback.identity.value @@ -131,6 +61,8 @@ def refresh_recovery_payload( "refresh_recovery": recovery, "settlement_identity": identity.as_dict(), "settlement_result": settlement_result_payload(readback.delivery), } + if not dry_run: + attach_settlement_progress(payload, readback, registry_path=registry_path, runtime_root=runtime_root) if decision == "reject": payload["error"] = ( f"{recovery['reason']}: committed writeback is unchanged; " diff --git a/loopx/control_plane/quota/settlement.py b/loopx/control_plane/quota/settlement.py index 7a0d87c45..7f9e67e02 100644 --- a/loopx/control_plane/quota/settlement.py +++ b/loopx/control_plane/quota/settlement.py @@ -3,6 +3,7 @@ from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path +import shlex from typing import Any from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result @@ -119,6 +120,7 @@ def render_refresh_recovery_markdown(payload: dict[str, Any]) -> str | None: f"- reason: `{recovery.get('reason')}`", "- appended: `False` — original writeback preserved; no new delivery or spend.", ] + lines.extend(render_settlement_progress_markdown(payload)) checkpoint = payload.get("vision_checkpoint") or {} if checkpoint: lines.append( @@ -152,6 +154,62 @@ class QuotaSettlementReadback: replay_phase: ReceiptBoundReplayPhase | None refresh_recovery: dict[str, Any] | None = None external_delivery: dict[str, Any] | None = None + progress: dict[str, Any] | None = None + + +def attach_settlement_progress( + payload: dict[str, Any], + readback: QuotaSettlementReadback, + *, + registry_path: Path | None = None, + runtime_root: Path | None = None, +) -> None: + """Render the TS-owned receipt progress without deriving a second settlement rule.""" + progress = readback.progress + if not isinstance(progress, dict) or progress.get("schema_version") != "quota_settlement_progress_v0": + raise RuntimeError("TypeScript quota settlement progress missing or invalid") + payload["settlement_progress"] = dict(progress) + payload.pop("settlement_owed", None) + identity = readback.identity.value + if payload.get("ok") is not True or progress.get("next_step") != "quota_spend" or identity is None: + return + prefix = "loopx" + if registry_path is not None: + prefix += f" --registry {shlex.quote(str(registry_path))}" + if runtime_root is not None: + prefix += f" --runtime-root {shlex.quote(str(runtime_root))}" + plan = build_turn_scoped_cli_settlement_plan( + goal_id=identity.goal_id, + agent_id=identity.agent_id, + todo_id=identity.todo_id, + replan_obligation_id=identity.replan_obligation_id, + turn_instance_id=identity.turn_instance_id, + command_prefix=prefix, + scoped_cli_args="", + lifecycle_actor_args="", + quota_spend_source=progress["quota_spend_source"], + ) + payload["settlement_owed"] = { + **identity.as_dict(), "schema_version": "turn_settlement_owed_v0", + "kind": "quota_spend", "recovery_does_not_spend": True, + "reason": ( + "quota spend is committed but its receipt is missing; retry the same identity to repair it without another debit" + if progress["state"] == "spend_receipt_required" else + "writeback is verified; execute quota spend once for the same settlement identity" + ), + "command": settlement_step_command(plan.as_dict(), SettlementStepKind.QUOTA_SPEND), + } + + +def render_settlement_progress_markdown(payload: dict[str, Any]) -> list[str]: + progress = payload.get("settlement_progress") + if not isinstance(progress, dict): + return [] + lines = [f"- settlement: `{progress.get('state')}`"] + owed = payload.get("settlement_owed") + if isinstance(owed, dict): + lines.extend([f"- settlement_owed: {owed['reason']}", "", "```sh", owed["command"], "```"]) + return lines __all__ = [ @@ -307,6 +365,7 @@ def read_heartbeat_settlement( writeback_run=_optional_readback_record(payload.get("writeback_run")), refresh_recovery=_optional_readback_record(payload.get("refresh_recovery")), external_delivery=_optional_readback_record(payload.get("external_delivery")), + progress=_optional_readback_record(payload.get("progress")), spend_run=_optional_readback_record(payload.get("spend_run")), heartbeat_receipt=_optional_readback_record(payload.get("heartbeat_receipt")), writeback_event=_optional_readback_record(payload.get("writeback_event")), diff --git a/loopx/control_plane/quota/settlement_cli.py b/loopx/control_plane/quota/settlement_cli.py index a616d6eed..2096d20ca 100644 --- a/loopx/control_plane/quota/settlement_cli.py +++ b/loopx/control_plane/quota/settlement_cli.py @@ -21,6 +21,7 @@ upgrade_identityless_heartbeat_receipt, ) from .settlement import ( + attach_settlement_progress, read_heartbeat_settlement, settlement_result_payload, ) @@ -374,6 +375,9 @@ def quota_rollout_details( "quiet_noop_allowed": bool(agent_channel.get("quiet_noop_allowed")), "closeout_required": closeout_required, } + cli_channel = interaction.get("cli_channel") + if isinstance(cli_channel, Mapping) and cli_channel.get("quota_spend_source"): + details["quota_spend_source"] = cli_channel["quota_spend_source"] retained_todo_id = normalize_todo_id(retained_selection.get("retained_todo_id")) if retained_todo_id: retained_bound = retained_disposition == "preserve_retained_todo" @@ -416,6 +420,10 @@ def attach_spend_settlement_result( ) if readback is None: raise RuntimeError("exact settlement readback unexpectedly returned not-found") + attach_settlement_progress( + payload, readback, runtime_root=runtime_root, + registry_path=Path(str(payload["registry"])) if payload.get("registry") else None, + ) identity = readback.identity.value if identity is None: settlement_result = readback.identity diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index 1f135abdf..c59be74f2 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/loopx/control_plane/quota/settlement_readback.ts @@ -78,6 +78,31 @@ interface ResultBundle extends JsonObject { payload: JsonObject; } +type SettlementProgressState = "identity_required" | "writeback_required" | + "writeback_receipt_required" | "spend_required" | "spend_receipt_required" | "settled"; + +/** Receipt verification owns progress; a durable debit alone is not settlement. */ +function settlementProgress( + identity: SettlementResult, writeback: SettlementResult, spend: SettlementResult, + writebackRun: JsonObject | null, spendRun: JsonObject | null, + spendSource: unknown = "heartbeat", +): JsonObject { + const source = spendSource ?? "heartbeat"; + if (source !== "heartbeat" && source !== "visible-goal") { + throw new EffectRuntimeRequestError("settlement spend source is invalid", "malformed_settlement_state"); + } + const state: SettlementProgressState = identity.failure ? "identity_required" + : writeback.failure ? (writebackRun ? "writeback_receipt_required" : "writeback_required") + : spend.failure ? (spendRun ? "spend_receipt_required" : "spend_required") + : "settled"; + return { + schema_version: "quota_settlement_progress_v0", state, + next_step: identity.failure ? "validation" : writeback.failure ? "durable_writeback" + : spend.failure ? "quota_spend" : null, + quota_spend_source: source, + }; +} + function optionalRequestString(value: unknown, label: string): string | null { if (value === null || value === undefined) return null; if (typeof value !== "string") { @@ -667,6 +692,7 @@ function failedReadback( settlement: bundle(downstreamFailure), terminal_closeout: bundle(terminalFailure), terminal_settlement: bundle(downstreamFailure), + progress: settlementProgress(identityResult, downstreamFailure, downstreamFailure, null, null), workspace_causality: null, semantic_replan_guard: null, writeback_run: null, @@ -778,6 +804,8 @@ export async function readQuotaSettlement(value: unknown): Promise { settlement: bundle(settled), terminal_closeout: bundle(terminalCloseout), terminal_settlement: bundle(terminalSettlement), + progress: settlementProgress(identityResult, writeback, spend, writebackRun, spendRun, + receiptDetails.quota_spend_source ?? spendRun?.source), workspace_causality: workspaceCausality, semantic_replan_guard: semanticReplanGuard, writeback_run: writebackRun, diff --git a/loopx/control_plane/quota/should_run_prepare.py b/loopx/control_plane/quota/should_run_prepare.py index 7ba06284f..5104d6334 100644 --- a/loopx/control_plane/quota/should_run_prepare.py +++ b/loopx/control_plane/quota/should_run_prepare.py @@ -92,14 +92,12 @@ select_task_orchestration_authority_items, ) from ..todos.summary_item import compact_todo_summary_item -from ..todos.goal_todo_projection import goal_todo_summaries from ..todos.user_gate import ( open_todo_count as _open_todo_count, ) from ..work_items.capability_monitor_fallback import ( build_capability_gate_with_monitor_fallback, ) -from ..work_items.planning_inventory import quota_runnable_action_candidates from ..work_items.primary_action import protocol_action_text as _protocol_action_text from ..work_items.work_lane import ( lark_inbox_reply_due_work_lane_contract, @@ -443,48 +441,6 @@ def _build_agent_work_lane( return monitor_only, work_lane, task_orchestration -def _authoritative_requested_agent_rows( - *, - registry_goal: Mapping[str, Any] | None, - todo_id: str | None, -) -> list[dict[str, Any]]: - """Resolve one explicitly requested Todo from the Goal's own state rows. - - The presented planning lanes are a display budget. Before this lookup an - owned, open, typed advancement Todo that sat outside them was unreachable by - an explicit ``--todo-id``, so the caller had no legal way to bind its own - quota guard to the row it had just worked on. Eligibility predicates still - run in the builder, so this widens reachability of the lookup, not what may - be selected. - """ - - normalized_todo_id = normalize_todo_id(todo_id) - if not normalized_todo_id or not isinstance(registry_goal, Mapping): - return [] - from ...materials import goal_state_path - - state_path = goal_state_path(dict(registry_goal)) - if state_path is None or not state_path.is_file(): - return [] - try: - projected = goal_todo_summaries( - dict(registry_goal), - state_text=state_path.read_text(encoding="utf-8"), - state_path=state_path, - rollout_events=[], - roles=["agent"], - status="open", - todo_id=normalized_todo_id, - agent_id=None, - limit=None, - ) - except (OSError, ValueError): - # A malformed or unreadable state file must not fail the guard; the - # presented lanes still decide, and the caller keeps the same refusal. - return [] - return [row for row in projected.todos if isinstance(row, dict)] - - def _prepare_quota_should_run_item( status_payload: dict[str, Any], *, @@ -842,29 +798,12 @@ def _prepare_quota_should_run_item( recovery_allowed = False reason = str(projection_gap_repair.get("reason") or reason) boundary_projection_repair = None - # An explicit `--todo-id` names one row the caller already chose, so it is - # resolved against the non-terminal rows this Goal records as the shared - # planning inventory. The bounded suggestion lanes above are a presentation - # budget: seeding the by-id lookup from them alone made an owned, open, - # typed advancement Todo unselectable whenever the display lanes were full, - # which left the caller no legal way to bind its own quota guard to it. - # The builder still applies every eligibility predicate, so this widens - # reachability of the lookup, not what may be selected. + # Resolve exact identity before Agent/display compaction. Live callers supply + # the complete source; pure status callers use only their supplied snapshot. requested_action_candidate = ( build_explicit_advancement_next_action( agent_identity=agent_identity, - agent_todo_items=[ - *quota_runnable_action_candidates( - agent_id=agent_frontier_id or "", - agent_todo_summary=agent_todo_summary, - capability_gate=capability_gate, - ), - *agent_todo_planning_source_items, - *_authoritative_requested_agent_rows( - registry_goal=registry_goal, - todo_id=requested_action_todo_id, - ), - ], + agent_todo_items=agent_todo_planning_source_items, available_capabilities=effective_available_capabilities, todo_id=requested_action_todo_id, selection_binding="pending_action_selection", diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index f4fd645ef..5c715dea7 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -18,6 +18,7 @@ from ..quota.spend_sources import ( build_quota_spend_action, host_goal_turn_reentry_action, + quota_spend_source_for_execution_context, ) from ..scheduler.execution_context import ( APP_HEARTBEAT_SETTLEMENT_RUNTIME_PROFILES, @@ -1355,6 +1356,7 @@ def _build_interaction_cli_channel( selection.apply_action_selection_cli_gate(channel, payload) if settlement_plan is not None and spend_after_selection: channel["settlement_plan"] = settlement_plan + channel["quota_spend_source"] = quota_spend_source_for_execution_context(scheduler_execution_context) if settlement_plan is not None and replan_settlement_contract is not None: channel["replan_settlement_contract"] = replan_settlement_contract if capability_reentry is not None: diff --git a/loopx/presentation/renderers/quota_event_markdown.py b/loopx/presentation/renderers/quota_event_markdown.py index 00fdd0ed8..296b5b704 100644 --- a/loopx/presentation/renderers/quota_event_markdown.py +++ b/loopx/presentation/renderers/quota_event_markdown.py @@ -3,6 +3,7 @@ from typing import Any from ...control_plane.quota.slot_accounting import QUOTA_SLOT_SPENT_CLASSIFICATION +from ...control_plane.quota.settlement import render_settlement_progress_markdown from ..markdown import append_operator_action_markdown, as_dict @@ -85,6 +86,7 @@ def render_quota_slot_preview_markdown(payload: dict[str, Any]) -> str: f"- registry_mutated: `{payload.get('registry_mutated')}`", f"- would_throttle: `{payload.get('would_throttle')}`", ] + lines.extend(render_settlement_progress_markdown(payload)) if payload.get("json_path"): lines.append(f"- json_path: `{payload.get('json_path')}`") if payload.get("index_path"): diff --git a/loopx/state_refresh.py b/loopx/state_refresh.py index 19328183c..273b0cfb2 100644 --- a/loopx/state_refresh.py +++ b/loopx/state_refresh.py @@ -30,9 +30,11 @@ ) from .control_plane.quota.settlement import ( SettlementIdentity, + attach_settlement_progress, read_heartbeat_settlement, render_first_refresh_checkpoint_hint, render_refresh_recovery_markdown, + render_settlement_progress_markdown, ) from .control_plane.quota.settlement_workspace_causality import resolve_settlement_workspace_requirement from .control_plane.quota.codex_session_usage import ( @@ -596,6 +598,7 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: f"- state_updated_at: `{frontmatter.get('updated_at')}`", f"- health_check: `{payload.get('health_check')}`", ] + lines.extend(render_settlement_progress_markdown(payload)) if "external_sink_delivery_authorized" in payload: lines.append( "- external_sink_delivery_authorized: " @@ -1489,6 +1492,15 @@ def refresh_state_run( "raw_artifacts_copied": False, "recommended_action_copied": False, } + if settlement_identity is not None and not dry_run: + committed_readback = read_heartbeat_settlement( + runtime_root, goal_id=safe_goal_id, agent_id=settlement_identity.agent_id, + todo_id=settlement_identity.todo_id, turn_instance_id=settlement_identity.turn_instance_id, + replan_obligation_id=settlement_identity.replan_obligation_id, + ) + if committed_readback is None: + raise RuntimeError("committed refresh settlement readback missing") + attach_settlement_progress(payload, committed_readback, registry_path=registry_path, runtime_root=runtime_root) return finish_external_delivery_refresh( payload, settlement_readback, runtime_root, dry_run=dry_run, ) diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md index 0ec5897c8..659a897ca 100644 --- a/skills/loopx-self-repair/references/repair-patterns.md +++ b/skills/loopx-self-repair/references/repair-patterns.md @@ -211,6 +211,7 @@ teaches a reusable control-plane lesson. | `dashboard_verified_mutation_projection_gap` | A preview-locked dashboard mutation reports a successful shared-state readback, but the initiating control still shows its old value; a second click then says the requested setting already exists. | Exact apply receipt, no-change canonical preview, shared-state readback verification, status projection generation/revision, rendered control state, and refresh outcome. | The data adapter verified the canonical write or no-change state, then the UI discarded that receipt and rebound immediately to a separate stale status projection. | Return the verified configuration through apply and no-change preview callbacks and use it as a drawer-scoped read model for the same Goal; refresh the normal projection independently, surface refresh failure without undoing the verified result, and clear the override when the drawer selection changes. Keep preview state visibly pending rather than presenting it as applied. Cover a deliberately stale status response in the browser smoke. | | `dashboard_open_token_picker_gap` | A bounded dashboard setting asks users to type protocol tokens such as `task_domain`; the placeholder looks like a current value, users cannot discover legal choices, or Goals without tagged work cannot enable a capability whose runtime treats the token filter as optional. | Current Goal Todo index, configured token allowlist, option-to-Todo match counts, canonical empty-filter semantics, preview payload, empty state, and packaged browser behavior. | An intentionally open optional backend vocabulary was exposed as a required product authority boundary; reading only a compact Goal-card Todo slice can also hide valid choices. | Keep the open typed token contract in the owning control-plane boundary, but present it as an optional per-Goal multi-select derived from the authoritative Todo index plus already configured values, using compact Goal Todo rows only as a compatibility fallback. Empty means no token filter while every independent admission boundary remains enforced; a non-empty selection remains a strict allowlist. Show match counts, preserve configured zero-match values, and cover unrestricted, restricted, invalid-token, preview, readback, empty-state, and packaged parity behavior. | | `app_host_heartbeat_identity_gap` | An App host can create recurring automations, but LoopX classifies it as a generic visible CLI loop; the agent may complete one phase and then stop because no host-owned successor wake is activated. Another form builds the right full scheduler packet but drops it at a compact Turn-envelope boundary that still reads a sibling App's legacy field. | Exact App versus CLI host surface, ambient thread id, runtime profile, activation packet, full and compact scheduler projections, scheduler ownership, settled-turn liveness, and terminal no-follow-up evidence. | Host capability existed, but LoopX modeled only the sibling CLI surface or left a downstream transport coupled to that sibling's packet name, so the App automation contract or successor wake did not survive the real execution path. | Add a distinct App host/runtime identity while reusing the provider-neutral `app_automation` cadence and ACK rules end to end. Bind the host's ambient thread id, preserve the packet through Turn compaction without a sibling-host alias, create/update the host automation after Todo writeback, keep settled non-terminal turns active for a fresh successor turn, and stop only on validated terminal no-follow-up. Preserve the CLI host and its native visible Goal path. Never reuse another App's local-store fallback. | +| `quota_source_and_settlement_split` | Explicit selection admits stale display work, or refresh reports a spend command that fails or hides a missing receipt. | Current unified Todo source versus displayed rows; exact guard identity; writeback/debit records plus receipts; unchanged returned CLI execution and next readback. | Admission appends Markdown outside the canonical reader, or an adapter reconstructs settlement from run presence and incomplete command arguments. | Read the complete unified source before display bounds, preserve the legacy event adapter before promotion, and fail closed after promotion. Reduce progress from typed receipt results, render the original actor/source/route, and execute returned commands in disposable real-backend journeys. Restore missing receipts through the existing idempotent writer and verify one debit; do not infer settlement from missing hints. | | `artifact_without_goal_delta` | Repeated individually valid changes leave the requested user outcome or peer handoff unqualified; completion reports count fields, receipts or PRs. | Current user/task acceptance, latest main and related work, actual caller/readback, `problem_context` delivery judgment and remaining dependency. | Work was selected and settled around implementation artifacts rather than an independently useful outcome slice. | Reconcile the accepted goal, consolidate the missing integration/negative/readback work, or justify a prerequisite with its real successor and owner. Update the existing task/vision through its owner when direction changed. Do not repair by minimum LOC/PR quotas, a second task ledger, fabricated follow-ups or stronger prose alone. | | `runtime_shadow_query_clock_drift` | A runtime-shadow bootstrap reports matched, then an immediate read-only inspect reports drift even though no Todo or lease writer ran; a subsequent captured write can also stop with `source_partition_continuity_unproved`. | Expected and observed parity and partition digests, field-level normalized projection diff, resume-condition decision fields, prepared outbox bytes, and source mutation receipts. | `resume_condition.evaluated_at` is recomputed from the reader clock and was included in durable authority identity, so observation time changed on every read and invalidated both parity and writer continuity. | Exclude only the query-clock observation from parity and partition semantic digests in the TS owner and its Python codec mirror. Still verify full prepared bytes/projections and compare readiness, reason, generation, target, and every other Todo/lease fact. Cover clock-only equality plus a real decision-change mismatch, then reproduce bootstrap-to-inspect and captured-write continuity through the real CLI before promotion. | diff --git a/tests/control_plane/test_explicit_todo_id_beyond_presented_lanes.py b/tests/control_plane/test_explicit_todo_id_beyond_presented_lanes.py deleted file mode 100644 index 34d0a84bb..000000000 --- a/tests/control_plane/test_explicit_todo_id_beyond_presented_lanes.py +++ /dev/null @@ -1,118 +0,0 @@ -"""An explicit --todo-id must reach the Goal's own rows, not only displayed lanes.""" - -from __future__ import annotations - -from pathlib import Path - -from loopx.control_plane.agents.agent_lane_recommendation import ( - build_explicit_advancement_next_action, -) -from loopx.control_plane.quota.should_run_prepare import ( - _authoritative_requested_agent_rows, -) -from loopx.control_plane.todos.quota_summary import ( - select_planning_inventory_source_items, -) - -GOAL_ID = "goal-beyond-lanes" -AGENT_ID = "agent-beyond-lanes" -WANTED_TODO_ID = "todo_beyond_lanes" -OTHER_AGENT_ID = "agent-somebody-else" - - -def _row(todo_id: str, *, claimed_by: str = AGENT_ID, status: str = "open") -> str: - return ( - f"- [ ] [P1] row {todo_id}\n" - " " - ) - - -def _fixture(tmp_path: Path, rows: list[str]) -> tuple[dict, Path]: - project = tmp_path / "project" - state = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md" - state.parent.mkdir(parents=True) - state.write_text( - "# Active Goal State\n\n## Agent Todo\n\n" + "\n".join(rows) + "\n", - encoding="utf-8", - ) - goal = { - "id": GOAL_ID, - "status": "active", - "repo": str(project), - "state_file": str(state.relative_to(project)), - "coordination": {"agent_model": "peer_v1", "registered_agents": [AGENT_ID]}, - } - return goal, state - - -def _candidate(rows: list[dict], todo_id: str): - return build_explicit_advancement_next_action( - agent_identity={"agent_id": AGENT_ID}, - agent_todo_items=rows, - available_capabilities=["network", "filesystem_write"], - todo_id=todo_id, - selection_binding="pending_action_selection", - ) - - -def test_a_row_outside_the_presented_lanes_is_reachable_by_id(tmp_path: Path) -> None: - rows = [_row(f"todo_presented_{index}") for index in range(1, 6)] - rows.append(_row(WANTED_TODO_ID)) - goal, _state = _fixture(tmp_path, rows) - # The presented lanes carry a bounded selection, so the row the caller - # worked on is absent from them. - presented_summary = { - "items": [ - {"todo_id": "todo_presented_1", "text": "[P1] row", "status": "open", "task_class": "advancement_task", "claimed_by": AGENT_ID}, - {"todo_id": "todo_presented_2", "text": "[P1] row", "status": "open", "task_class": "advancement_task", "claimed_by": AGENT_ID}, - ] - } - presented = select_planning_inventory_source_items(presented_summary, None) - assert WANTED_TODO_ID not in {item.get("todo_id") for item in presented} - assert _candidate(presented, WANTED_TODO_ID) is None - - authoritative = _authoritative_requested_agent_rows( - registry_goal=goal, - todo_id=WANTED_TODO_ID, - ) - - assert [row["todo_id"] for row in authoritative] == [WANTED_TODO_ID] - assert _candidate(authoritative, WANTED_TODO_ID) is not None - - -def test_unknown_id_or_unreadable_state_resolves_nothing(tmp_path: Path) -> None: - goal, _state = _fixture(tmp_path, [_row(WANTED_TODO_ID)]) - - assert _authoritative_requested_agent_rows(registry_goal=goal, todo_id="todo_unknown") == [] - assert _authoritative_requested_agent_rows(registry_goal=goal, todo_id=None) == [] - assert ( - _authoritative_requested_agent_rows( - registry_goal={"id": GOAL_ID, "repo": str(tmp_path / "missing"), "state_file": "none.md"}, - todo_id=WANTED_TODO_ID, - ) - == [] - ) - - -def test_predicates_still_refuse_another_agents_or_blocked_rows(tmp_path: Path) -> None: - goal, _state = _fixture( - tmp_path, - [ - _row("todo_other_agent", claimed_by=OTHER_AGENT_ID), - _row("todo_blocked_row", status="blocked"), - ], - ) - - other_agent_rows = _authoritative_requested_agent_rows( - registry_goal=goal, - todo_id="todo_other_agent", - ) - blocked_rows = _authoritative_requested_agent_rows( - registry_goal=goal, - todo_id="todo_blocked_row", - ) - - assert _candidate(other_agent_rows, "todo_other_agent") is None - assert _candidate(blocked_rows, "todo_blocked_row") is None diff --git a/tests/control_plane/test_quota_authority_settlement_journey.py b/tests/control_plane/test_quota_authority_settlement_journey.py new file mode 100644 index 000000000..04a9e1799 --- /dev/null +++ b/tests/control_plane/test_quota_authority_settlement_journey.py @@ -0,0 +1,196 @@ +"""Select current work and execute its returned settlement, using real stores/CLI.""" +from __future__ import annotations + +import json +import shlex +from pathlib import Path + +import pytest +import test_quota_settlement_cli as cli +from canonical_authority_fixture import initialize_canonical_authority + +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos +from loopx.control_plane.quota.settlement import read_heartbeat_settlement +from loopx.event_sourced_state import AppendOnlyStateEventStore, TODO_ADDED, make_state_event +from loopx.rollout_event_log import rollout_event_log_path +from loopx.presentation.renderers.quota_event_markdown import render_quota_slot_preview_markdown +from loopx.state_refresh import render_state_refresh_markdown +from loopx.todos import list_goal_todos + + +def _row(todo_id: str, *, status: str = "open", extra: str = "") -> str: + return ( + f"- [{'x' if status == 'done' else ' '}] [P1] Validate {todo_id}.\n" + f" \n" + ) + + +def _source(root: Path, *, provider: str, status: str = "open", extra: str = "", empty: bool = False): + project, runtime, registry = cli._write_fixture(root) + goal = json.loads(registry.read_text())["goals"][0] + path = project / goal["state_file"] + prefix = path.read_text().split("## Agent Todo")[0] + "## Agent Todo\n\n" + rows = "".join(_row(f"todo_ready_{i}") for i in range(35)) + path.write_text(prefix + ("" if empty else rows + _row(cli.TODO_ID, status=status, extra=extra))) + if provider != "legacy": + fields = parse_active_state_todos(path.read_text(), goal=goal, item_limit=None) + initialize_canonical_authority( + runtime, cli.GOAL_ID, + build_todo_runtime_shadow_projection( + goal_id=cli.GOAL_ID, todos=fields["agent_todos"]["items"], handoff_mode="soft_claim", + ), + state_path=path, provider=provider, + ) + return project, runtime, registry, path, prefix + + +def _guard(project: Path, runtime: Path, registry: Path, *, turn_id: str = cli.TURN_ID): + return cli._run_cli( + registry, runtime, "quota", "should-run", "--codex-app", + "--goal-id", cli.GOAL_ID, "--agent-id", cli.AGENT_ID, + "--todo-id", cli.TODO_ID, "--turn-instance-id", turn_id, + "--scan-path", str(project), cwd=project, + ) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("status", ["done", "blocked"]) +def test_stale_markdown_cannot_admit_terminal_or_blocked_work(tmp_path, provider, status): + project, runtime, registry, path, prefix = _source(tmp_path, provider=provider, status=status) + path.write_text(prefix + _row(cli.TODO_ID)) + listed = list_goal_todos(registry_path=registry, goal_id=cli.GOAL_ID, + runtime_root_arg=str(runtime), todo_id=cli.TODO_ID) + assert listed["todo"]["status"] == status + code, guard = _guard(project, runtime, registry) + assert code == 1 + assert guard["decision"] == "skip" + assert guard["heartbeat_receipt"]["status"] == "not_committed" + assert not guard.get("selected_todo") + + +@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) +def test_exact_selection_reaches_work_beyond_display_limits(tmp_path, provider): + project, runtime, registry, path, _ = _source(tmp_path, provider=provider) + if provider != "legacy": + path.unlink() # Display is not a prerequisite for promoted admission. + code, guard = _guard(project, runtime, registry) + assert code == 0, guard + assert guard["decision"] == "run" + assert guard["selected_todo"]["todo_id"] == cli.TODO_ID + assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == cli.TODO_ID + + +@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) +@pytest.mark.parametrize("extra", [f"excluded_agents={cli.AGENT_ID}", "claimed_by=agent-other", + "required_capabilities=production_access"]) +def test_exact_selection_keeps_executor_and_capability_gates(tmp_path, provider, extra): + project, runtime, registry, _, _ = _source( + tmp_path, provider=provider, extra=extra, + ) + code, guard = _guard(project, runtime, registry) + assert code == 1 + assert guard["decision"] == "skip" + assert guard["heartbeat_receipt"]["status"] == "not_committed" + + +@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) +def test_absent_current_identity_cannot_be_selected(tmp_path, provider): + project, runtime, registry, path, prefix = _source(tmp_path, provider=provider, empty=True) + if provider != "legacy": + path.write_text(prefix + _row(cli.TODO_ID)) + code, guard = _guard(project, runtime, registry) + assert code == 1 + assert guard["decision"] == "skip" + assert guard["heartbeat_receipt"]["status"] == "not_committed" + + +def test_failed_canonical_read_cannot_fall_back_to_markdown(tmp_path): + project, runtime, registry, path, _ = _source(tmp_path, provider="file") + (runtime / "authority" / "file-v0").rename(runtime / "unavailable-provider") + assert cli.TODO_ID in path.read_text() + code, guard = _guard(project, runtime, registry) + assert code != 0 + assert not guard.get("selected_todo") + assert cli._heartbeat_receipt_count(runtime, cli.TURN_ID) == 0 + + +def test_legacy_event_overlay_keeps_current_role_over_stale_markdown(tmp_path): + project, runtime, registry, path, _ = _source(tmp_path, provider="legacy") + AppendOnlyStateEventStore(path.with_name("events.jsonl")).append(make_state_event( + event_id="event-current-role", goal_id=cli.GOAL_ID, event_type=TODO_ADDED, + refs={"todo_id": cli.TODO_ID}, + payload={"role": "user", "title": "Review the proposed result", "task_class": "user_action"}, + recorded_at="2026-08-04T00:00:00Z", producer="regression-fixture", + )) + listed = list_goal_todos(registry_path=registry, goal_id=cli.GOAL_ID, + runtime_root_arg=str(runtime), todo_id=cli.TODO_ID) + assert listed["todo"]["role"] == "user" + code, guard = _guard(project, runtime, registry) + assert code == 1 + assert guard["decision"] == "skip" + assert guard["heartbeat_receipt"]["status"] == "not_committed" + + +def _refresh(project: Path, runtime: Path, registry: Path): + return cli._run_cli( + registry, runtime, "refresh-state", "--goal-id", cli.GOAL_ID, + "--classification", "validated_progress", "--delivery-batch-scale", "implementation", + "--delivery-outcome", "outcome_progress", "--agent-id", cli.AGENT_ID, + "--todo-id", cli.TODO_ID, "--turn-instance-id", cli.TURN_ID, + "--no-global-sync", "--suppress-external-sinks", cwd=project, + ) + + +def _execute(command: str, project: Path, runtime: Path, registry: Path): + # Execute the returned argv unchanged. The fixture supplies only CLI output + # format and the same registry/runtime, never missing actor/binding arguments. + return cli._run_cli(registry, runtime, *shlex.split(command)[1:], cwd=project) + + +def test_returned_command_settles_and_repairs_receipts_without_another_debit(tmp_path): + project, runtime, registry = cli._write_fixture(tmp_path) + code, guard = _guard(project, runtime, registry) + assert code == 0, guard + code, refreshed = _refresh(project, runtime, registry) + assert code == 0, refreshed + command = refreshed["settlement_owed"]["command"] + code, spent = _execute(command, project, runtime, registry) + assert code == 0, spent + assert refreshed["settlement_progress"]["state"] == "spend_required" + assert command in render_state_refresh_markdown(refreshed) + argv = shlex.split(command) + assert argv[argv.index("--registry") + 1] == str(registry) + assert argv[argv.index("--runtime-root") + 1] == str(runtime) + assert spent["appended"] is True + assert spent["settlement_progress"]["state"] == "settled" + assert "settlement: `settled`" in render_quota_slot_preview_markdown(spent) + assert cli._spend_run_count(runtime) == 1 + + path = rollout_event_log_path(runtime, cli.GOAL_ID) + events = [json.loads(line) for line in path.read_text().splitlines() if line.strip()] + retained = [event for event in events if event["event_kind"] != "quota_spend"] + assert len(events) - len(retained) == 1 + path.write_text("".join(json.dumps(event) + "\n" for event in retained)) + code, recovery = _refresh(project, runtime, registry) + assert code == 0, recovery + assert recovery["settlement_progress"]["state"] == "spend_receipt_required" + assert recovery["settlement_owed"]["recovery_does_not_spend"] is True + repair_command = recovery["settlement_owed"]["command"] + assert repair_command in render_state_refresh_markdown(recovery) + code, repaired = _execute(repair_command, project, runtime, registry) + assert code == 0, repaired + assert repaired["appended"] is False + assert cli._spend_run_count(runtime) == 1 + readback = read_heartbeat_settlement(runtime, goal_id=cli.GOAL_ID, agent_id=cli.AGENT_ID, + todo_id=cli.TODO_ID, turn_instance_id=cli.TURN_ID) + assert readback.settlement.failure is None + code, replay = _refresh(project, runtime, registry) + assert code == 0, replay + assert replay["settlement_progress"]["state"] == "settled" + assert "settlement_owed" not in replay + code, next_wake = _guard(project, runtime, registry, turn_id="turn-after-repair") + assert code == 0, next_wake + assert next_wake["effective_action"] != "unsettled_host_turn_recovery" + assert cli._spend_run_count(runtime) == 1 diff --git a/tests/control_plane/test_quota_settlement.py b/tests/control_plane/test_quota_settlement.py index 7b609ed9d..28428c40f 100644 --- a/tests/control_plane/test_quota_settlement.py +++ b/tests/control_plane/test_quota_settlement.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import shlex from dataclasses import replace from pathlib import Path from types import SimpleNamespace @@ -49,6 +50,46 @@ TURN_ID = "turn-settlement-1" +@pytest.mark.parametrize("replan", [False, True]) +def test_command_plan_supplies_actor_from_identity(replan): + plan = quota_effect_program.build_turn_scoped_cli_settlement_plan( + goal_id=GOAL_ID, agent_id=AGENT_ID, command_prefix="loopx", + todo_id=None if replan else TODO_ID, + replan_obligation_id="replan-0000000000000001" if replan else None, + turn_instance_id="turn with spaces", scoped_cli_args="", lifecycle_actor_args="", + ) + for step in plan.as_dict()["ordered_steps"]: + if command := step.get("command_template"): + argv = shlex.split(command) + assert argv.count("--agent-id") == 1 + assert argv[argv.index("--agent-id") + 1] == AGENT_ID + assert argv[argv.index("--turn-instance-id") + 1] == "turn with spaces" + + +@pytest.mark.parametrize("arguments", ["--agent-id other", "--agent-id=other", "--agent-id", + f"--agent-id {AGENT_ID} --agent-id {AGENT_ID}"]) +@pytest.mark.parametrize("field", ["scoped_cli_args", "lifecycle_actor_args"]) +def test_command_plan_rejects_ambiguous_or_conflicting_actor(arguments, field): + with pytest.raises(ValueError, match="actor must match"): + quota_effect_program.build_turn_scoped_cli_settlement_plan( + goal_id=GOAL_ID, agent_id=AGENT_ID, command_prefix="loopx", + todo_id=TODO_ID, replan_obligation_id=None, turn_instance_id=TURN_ID, + **{"scoped_cli_args": "", "lifecycle_actor_args": "", field: arguments}, + ) + + +@pytest.mark.parametrize("arguments", [f"--agent-id {AGENT_ID}", f" --agent-id={AGENT_ID}"]) +def test_command_plan_retains_one_matching_actor(arguments): + plan = quota_effect_program.build_turn_scoped_cli_settlement_plan( + goal_id=GOAL_ID, agent_id=AGENT_ID, command_prefix="loopx", + todo_id=TODO_ID, replan_obligation_id=None, turn_instance_id=TURN_ID, + scoped_cli_args=arguments, lifecycle_actor_args=arguments, + ) + argv = shlex.split(settlement_step_command(plan.as_dict(), SettlementStepKind.QUOTA_SPEND)) + assert argv[argv.index("--turn-instance-id") + 1] == TURN_ID + assert sum(arg == "--agent-id" or arg.startswith("--agent-id=") for arg in argv) == 1 + + def _receipt(step: SettlementStepKind, marker: str) -> SettlementReceipt: return SettlementReceipt( step_kind=step, diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index a5f7bfc24..c7cb59132 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -2499,22 +2499,9 @@ def test_visible_goal_refresh_and_spend_preserve_selected_todo_causality( assert refresh["delivery_workspace_causality"]["todo_id"] == TODO_ID assert refresh["delivery_workspace_causality"]["requirement"] == "not_required" - spend_rc, spend = _run_cli( - registry_path, - runtime, - "quota", - "spend-slot", - "--goal-id", - GOAL_ID, - "--slots", - "1", - "--source", - "visible-goal", - "--execute", - *binding, - "--scan-path", - str(project), - ) + command = refresh["settlement_owed"]["command"] + assert "--source visible-goal" in command + spend_rc, spend = _run_cli(registry_path, runtime, *shlex.split(command)[1:]) assert spend_rc == 0, spend assert spend["todo_id"] == TODO_ID assert spend["turn_instance_id"] == turn_instance_id diff --git a/tests/control_plane/test_refresh_external_delivery.py b/tests/control_plane/test_refresh_external_delivery.py index 363960799..3f80e5c25 100644 --- a/tests/control_plane/test_refresh_external_delivery.py +++ b/tests/control_plane/test_refresh_external_delivery.py @@ -64,6 +64,8 @@ def test_dry_run_cannot_pause_or_resume_a_committed_operation(session): assert run(args)["external_sink_delivery_authorized"] is True run([*args, "--suppress-external-sinks"]) denied = run(args, expected=1) + assert denied["settlement_progress"]["state"] == "spend_required" + assert "settlement_owed" not in denied # Rejected recovery is not a spend instruction. key = denied["external_delivery"]["resume_key"] before = journal.read_bytes(), index.read_bytes() run([*args, "--resume-external-sinks", key, "--dry-run"]) diff --git a/tests/control_plane/test_turn_settlement_owed_signal.py b/tests/control_plane/test_turn_settlement_owed_signal.py index 4a3fa321f..bf71d7748 100644 --- a/tests/control_plane/test_turn_settlement_owed_signal.py +++ b/tests/control_plane/test_turn_settlement_owed_signal.py @@ -1,4 +1,4 @@ -"""A committed writeback names the quota spend the Turn still owes.""" +"""A raw refresh cannot authorize spend before its CLI receipt is committed.""" from __future__ import annotations @@ -6,8 +6,6 @@ from pathlib import Path from loopx.control_plane.quota.effect_program import SettlementIdentity -from loopx.control_plane.quota.refresh_external_delivery import turn_settlement_owed -from loopx.control_plane.quota.settlement import read_heartbeat_settlement from loopx.rollout_event_log import rollout_event_log_path from loopx.state_refresh import refresh_state_run @@ -48,74 +46,6 @@ def _append_guard_receipt(runtime_root: Path, identity: SettlementIdentity) -> N ) -def _append_run_index_record(runtime_root: Path, record: dict) -> None: - path = runtime_root / "goals" / GOAL_ID / "runs" / "index.jsonl" - path.parent.mkdir(parents=True, exist_ok=True) - with path.open("a", encoding="utf-8") as handle: - handle.write(json.dumps(record) + "\n") - - -def _readback(runtime_root: Path, *, spent: bool): - identity = SettlementIdentity(GOAL_ID, AGENT_ID, TODO_ID, TURN_ID) - _append_guard_receipt(runtime_root, identity) - _append_run_index_record( - runtime_root, - { - "classification": "state_refreshed", - "delivery_outcome": "outcome_progress", - "goal_id": GOAL_ID, - "agent_id": AGENT_ID, - "todo_id": TODO_ID, - "turn_instance_id": TURN_ID, - }, - ) - if spent: - _append_run_index_record( - runtime_root, - { - "classification": "quota_slot_spent", - "goal_id": GOAL_ID, - "agent_id": AGENT_ID, - "todo_id": TODO_ID, - "turn_instance_id": TURN_ID, - }, - ) - readback = read_heartbeat_settlement( - runtime_root, - goal_id=GOAL_ID, - agent_id=AGENT_ID, - todo_id=TODO_ID, - turn_instance_id=TURN_ID, - ) - assert readback is not None - return identity, readback - - -def test_unspent_turn_names_the_owed_quota_spend(tmp_path: Path) -> None: - identity, readback = _readback(tmp_path / "runtime", spent=False) - - owed = turn_settlement_owed(readback) - - assert owed is not None - assert owed["schema_version"] == "turn_settlement_owed_v0" - assert owed["kind"] == "quota_spend" - assert owed["effect_id"] == identity.effect_id - assert owed["todo_id"] == TODO_ID - assert owed["turn_instance_id"] == TURN_ID - assert owed["recovery_does_not_spend"] is True - assert owed["command"].startswith("loopx quota spend-slot") - assert f"--todo-id {TODO_ID}" in owed["command"] - assert f"--turn-instance-id {TURN_ID}" in owed["command"] - assert "--execute" in owed["command"] - - -def test_spent_turn_owes_nothing(tmp_path: Path) -> None: - _identity, readback = _readback(tmp_path / "runtime", spent=True) - - assert readback.spend_run is not None - assert turn_settlement_owed(readback) is None - - def _fixture(tmp_path: Path) -> tuple[Path, Path, Path]: project = tmp_path / "project" state_path = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md" @@ -147,7 +77,7 @@ def _fixture(tmp_path: Path) -> tuple[Path, Path, Path]: return registry_path, project, tmp_path / "runtime" -def test_refresh_state_result_carries_the_owed_step(tmp_path: Path) -> None: +def test_raw_refresh_requires_receipt_before_offering_spend(tmp_path: Path) -> None: registry_path, project, runtime_root = _fixture(tmp_path) identity = SettlementIdentity(GOAL_ID, AGENT_ID, TODO_ID, TURN_ID) _append_guard_receipt(runtime_root, identity) @@ -172,10 +102,9 @@ def test_refresh_state_result_carries_the_owed_step(tmp_path: Path) -> None: assert result["ok"] is True assert result["appended"] is True - owed = result["settlement_owed"] - assert owed["effect_id"] == identity.effect_id - assert owed["kind"] == "quota_spend" - assert f"--turn-instance-id {TURN_ID}" in owed["command"] + assert result["settlement_progress"]["state"] == "writeback_receipt_required" + assert result["settlement_progress"]["next_step"] == "durable_writeback" + assert "settlement_owed" not in result assert "settlement_owed" not in refresh_state_run( registry_path=registry_path, runtime_root_override=str(runtime_root), diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts index 585b3c966..17ad1ec43 100644 --- a/tests/control_plane_ts/quota_settlement_readback.test.ts +++ b/tests/control_plane_ts/quota_settlement_readback.test.ts @@ -203,6 +203,50 @@ function request(runtimeRoot: string, overrides: Record = {}) { }; } +test("settlement progress requires both effects and exact receipts", async t => { + const cases = [ + { guard: false, state: "identity_required", next: "validation" }, + { state: "writeback_required", next: "durable_writeback" }, + { writeback: true, remove: "refresh_state", state: "writeback_receipt_required", next: "durable_writeback" }, + { writeback: true, state: "spend_required", next: "quota_spend" }, + { writeback: true, spend: true, remove: "quota_spend", state: "spend_receipt_required", next: "quota_spend" }, + { writeback: true, spend: true, state: "settled", next: null }, + ]; + for (const entry of cases) await t.test(entry.state, async () => { + const root = await fixture(entry); + try { + if (entry.remove) { + const path = join(root, "goals", goalId, "rollout-event-log.jsonl"); + const events = (await readFile(path, "utf8")).trim().split("\n").map(line => JSON.parse(line)); + await writeFile(path, events.filter(event => event.event_kind !== entry.remove).map(event => JSON.stringify(event)).join("\n") + "\n"); + } + const result = await readQuotaSettlement(request(root)); + assert.deepEqual(result.progress, { + schema_version: "quota_settlement_progress_v0", state: entry.state, + next_step: entry.next, quota_spend_source: "heartbeat", + }); + } finally { await rm(root, { recursive: true, force: true }); } + }); +}); + +test("settlement progress preserves typed source and rejects malformed sources", async () => { + const root = await fixture({writeback: true}); + try { + const path = join(root, "goals", goalId, "rollout-event-log.jsonl"); + const events = (await readFile(path, "utf8")).trim().split("\n").map(line => JSON.parse(line)); + for (const source of ["visible-goal", "unknown", 42]) { + events[0].details.quota_spend_source = source; + await writeFile(path, events.map(event => JSON.stringify(event)).join("\n") + "\n"); + if (source === "visible-goal") { + const result = await readQuotaSettlement(request(root)); + assert.equal((result.progress as Record).quota_spend_source, source); + } else { + await assert.rejects(readQuotaSettlement(request(root)), /settlement spend source is invalid/); + } + } + } finally { await rm(root, {recursive: true, force: true}); } +}); + test("monitor closeout requires the exact committed effect, not a matching observation row", async t => { const effect = `quota-monitor-poll:${goalId}:${agentId}:${turnId}`; const cases: [string, Record, string][] = [