diff --git a/docs/reference/protocols/quota-blocked-causal-closeout-v0.md b/docs/reference/protocols/quota-blocked-causal-closeout-v0.md index ecb9d4cb33..8f44e880b7 100644 --- a/docs/reference/protocols/quota-blocked-causal-closeout-v0.md +++ b/docs/reference/protocols/quota-blocked-causal-closeout-v0.md @@ -77,6 +77,41 @@ number alone is not a qualified blocked-closeout proof. Register a real monitor or dependency Todo and use `monitor_changed` / `todo_done` for causal closeout. Unsupported PR waits now name this recovery route explicitly. +### Unavailable Monitor observation + +An admitted open Agent `continuous_monitor` on canonical authority may discover +that its observation cannot be obtained. Close the original Goal/Agent/Todo/Turn +with `refresh-state --delivery-outcome outcome_gap --progress-result-class +blocked`, a stable blocker ID, evidence IDs and a normal vision decision. Do not +invent a `monitor-poll` hash, record control metadata as an account observation, +or spend. The existing settlement plan exposes this path. + +The same typed quota owner freezes `quota_monitor_unavailable_v0` in the existing +`blocked_retry` receipt slot, with `observation_available=false` and the Monitor +scope. Exact admission, identity, evidence and durable writeback still gate +`typed_blocked_writeback_no_spend`. This closes only the attempted Turn: the Todo +stays open; `last_checked_at`, `next_due_at`, result hash, material generation, +cadence and expiry remain unchanged. It creates neither a retry clock nor an +advancement wait, material successor, completion or delivery claim. Existing +advancement retry/causal-wait semantics and genuine committed polls retain their +behavior. A committed poll cannot be rewritten as an unavailable attempt. + +Exact replay returns the frozen receipt, with no duplicate writeback or debit. +After closing the original effect, reenter the current recovery host Turn to +select independently eligible work; do not create a third identity or rebind the +old one. Missing capability admission or evidence and wrong bindings still fail. +File and SQLite CLI acceptance covers this recovery, preserved observation +fields, capability restoration, committed-poll rejection and zero debit. +PostgreSQL and live host adoption remain separate qualifications. Before rollback, +reconcile these receipts using a runtime that recognizes the new proof. + +This is a bounded R1/S2/S10 recovery slice in the existing quota owner. The legacy +`isBoundedBlockedRetry` entry point remains for active callers, but now qualifies +Monitor unavailable proofs as well; only the advancement schema projects a +bounded retry. No Python decision owner, new switch, frontend control or Lark +authority is added. App/Chat consume the same CLI settlement guidance; their +packaged surfaces are not qualified by the CLI tests. + ## 中文 已准入的 advancement Turn 可以发现真实依赖，以 @@ -130,3 +165,27 @@ excluded/bound 限制仍拒绝。恢复后，模式要求执行 fencing 时获 `pr_merged` 仍是合法的调度等待条件，但 PR 编号本身不能证明阻塞结算所需的 真实依赖。应登记实际 monitor／依赖 Todo，以 `monitor_changed`／`todo_done` 完成因果结算；错误信息明确给出此恢复路径。 + +### Monitor 观察不可取得 + +canonical authority 上已准入、开放的 Agent `continuous_monitor`，可能在尝试中 +发现无法取得观察。用原 Goal/Agent/Todo/Turn 调用 `refresh-state`，提供 +`outcome_gap`、typed `blocked`、稳定 blocker、证据 ID 与正常 vision 决策，沿既有 +计划关闭本次尝试。不得编造 poll hash、把控制元数据记成账户观察或扣额。 + +同一 TS quota owner 在既有 `blocked_retry` 回执槽冻结 +`quota_monitor_unavailable_v0`、`observation_available=false` 与 Monitor 范围； +精确准入、身份、证据和持久写回仍是无扣额结算门槛。只关闭原 Turn，Todo 仍开放， +检查/到期时钟、hash、material generation、cadence 与 expiry 不变。不生成重试时钟、 +advancement 等待、material successor、完成或投递声明。旧 advancement 重试/因果等待 +与真实 poll 保持原行为；已提交 poll 不能改写为不可观察尝试。 + +精确重放返回冻结回执，不重复写回或扣额；原 effect 关闭后，重新进入当前 recovery +host Turn 可选择独立合法任务，不新造第三身份或重绑旧 Turn。缺准入能力、缺证或错绑 +仍拒绝。File/SQLite 的真实 CLI 验收覆盖恢复、观察字段保留、能力恢复、已 poll 拒绝与 +零扣额；PostgreSQL、本机真实采用及 App/Lark 打包入口仍分阶段验收。降级前用支持新 +证明的运行时核对回执。 + +这是既有 quota owner 的 R1/S2/S10 有界修复，兼容保留活动调用者使用的 +`isBoundedBlockedRetry` 名称；只有 advancement schema 产生有界重试投影。 +不新增 Python 判断源、开关、前端控件或 Lark 权威；CLI 计划提供同源说明。 diff --git a/loopx/control_plane/quota/blocked_wait.ts b/loopx/control_plane/quota/blocked_wait.ts index 15a8587998..d2725c25ca 100644 --- a/loopx/control_plane/quota/blocked_wait.ts +++ b/loopx/control_plane/quota/blocked_wait.ts @@ -11,6 +11,7 @@ import { export const BLOCKED_WAIT_REQUEST_SCHEMA = "loopx_quota_blocked_wait_request_v0"; const CAUSAL_WAIT_SCHEMA = "quota_blocked_causal_wait_v0"; +export const MONITOR_UNAVAILABLE_SCHEMA = "quota_monitor_unavailable_v0"; function reject(message: string): never { throw new EffectRuntimeRequestError(`typed blocked no-spend closeout ${message}`); @@ -79,6 +80,21 @@ function retainedTodo(todo: JsonObject): JsonObject { .map((field) => [field, todo[field]])); } +/** An unavailable attempt is not an observation or a retry clock. The + * surrounding settlement owner verifies the exact guard, typed blocker and + * evidence before committing this frozen, canonical Monitor scope. */ +export function isMonitorUnavailableWait(value: unknown, todoId: string | null): boolean { + const wait = jsonObject(value); + const monitor = jsonObject(wait?.monitor_todo); + return !!wait && wait.schema_version === MONITOR_UNAVAILABLE_SCHEMA && + wait.source === "turn_settlement" && wait.observation_available === false && + !!todoId && wait.todo_id === todoId && monitor?.todo_id === todoId && + monitor.role === "agent" && monitor.task_class === "continuous_monitor" && + monitor.status === "open" && + (monitor.archive_state == null || monitor.archive_state === "active") && + timestamp(wait.observed_at) !== null; +} + /** Preflight belongs to the same TS settlement owner as durable readback. * Python only transports the complete current Todo facts and observation clock. */ export function prepareBlockedWait(value: unknown): JsonObject { @@ -89,10 +105,25 @@ export function prepareBlockedWait(value: unknown): JsonObject { const todos = request.todos.map((value) => requireJsonObject(value, "blocked wait Todo")); const matches = todos.filter((todo) => todo.todo_id === request.todo_id); const todo = matches[0]; - if (matches.length !== 1 || !todo || !["open", "deferred"].includes(String(todo.status)) || - todo.task_class !== "advancement_task") reject("requires the same unfinished advancement Todo"); + if (matches.length !== 1 || !todo || !["open", "deferred"].includes(String(todo.status))) { + reject("requires the same unfinished Todo"); + } const observed = timestamp(request.observed_at); if (observed === null) reject("has an invalid timestamp"); + if (todo.task_class === "continuous_monitor") { + if (todo.role !== "agent" || todo.status !== "open" || + (todo.archive_state != null && todo.archive_state !== "active") || + request.allow_turn_settlement_retry !== true) { + reject("requires an admitted open Monitor attempt on canonical authority"); + } + const fields = ["claimed_by", "last_checked_at", "next_due_at", "result_hash", + "material_change", "material_change_generation", "cadence", "expires_at", "watch_only"]; + return {schema_version: MONITOR_UNAVAILABLE_SCHEMA, source: "turn_settlement", + todo_id: request.todo_id, observed_at: request.observed_at, observation_available: false, + monitor_todo: {...retainedTodo(todo), ...Object.fromEntries( + fields.filter(field => todo[field] !== undefined).map(field => [field, todo[field]]))}}; + } + if (todo.task_class !== "advancement_task") reject("requires the same unfinished advancement Todo"); const resume = todo.resume_when; const condition = jsonObject(todo.resume_condition); if (typeof resume === "string" && /^(?:monitor_changed|todo_done):/.test(resume)) { diff --git a/loopx/control_plane/quota/settlement_phase.ts b/loopx/control_plane/quota/settlement_phase.ts index 5bc1b71a2e..624254049c 100644 --- a/loopx/control_plane/quota/settlement_phase.ts +++ b/loopx/control_plane/quota/settlement_phase.ts @@ -1,10 +1,13 @@ import type { SettlementIdentity } from "../effect_program.ts"; import { jsonObject } from "../runtime_decode.ts"; -import { isCausalBlockedWait } from "./blocked_wait.ts"; +import { isCausalBlockedWait, isMonitorUnavailableWait } from "./blocked_wait.ts"; -/** A blocked Turn needs a bounded retry or a verified canonical causal wait. */ +/** Historical entry point for blocked no-spend qualification: an advancement + * retry/causal wait or a canonical Monitor unavailable attempt. The latter + * creates no retry clock and cannot be projected as an advancement wait. */ export function isBoundedBlockedRetry(value: unknown, todoId: string | null): boolean { if (isCausalBlockedWait(value, todoId)) return true; + if (isMonitorUnavailableWait(value, todoId)) return true; const retry = jsonObject(value); if (!retry || retry.schema_version !== "quota_blocked_retry_v0" || (retry.source !== "todo" && retry.source !== "turn_settlement") || @@ -96,12 +99,14 @@ export interface ReceiptBoundMonitorSettlementState { material_change: boolean; durable_writeback_present: boolean; quota_spend_present: boolean; + /** Exact typed unavailable writeback, verified by the settlement owner. */ + unavailable_attempt_present?: boolean; } export function receiptBoundMonitorPhase( state: ReceiptBoundMonitorSettlementState, ): ReceiptBoundMonitorPhase { - if (!state.poll_present) return "poll_due"; + if (!state.poll_present && !state.unavailable_attempt_present) return "poll_due"; // The committed monitor-poll is the durable no-spend closeout for this // monitor Turn. A material observation may atomically release an independent // successor, but it never upgrades the observe-only monitor into an diff --git a/loopx/control_plane/quota/settlement_plan.ts b/loopx/control_plane/quota/settlement_plan.ts index fbb0a9e257..f903d6da75 100644 --- a/loopx/control_plane/quota/settlement_plan.ts +++ b/loopx/control_plane/quota/settlement_plan.ts @@ -53,7 +53,11 @@ export function turnScopedCliSettlementPlan(params: JsonObject): SettlementPlan : "validation succeeded; " + VISION_MATERIAL_CLOSEOUT_HINT + " Pass a new packet with --agent-vision-json.") + " Route elimination needs evidence; failure alone is not progress. " + - "outcome_gap: blocked + blocker/evidence IDs; continuation checks.", + "outcome_gap: blocked + blocker/evidence IDs; continuation checks. " + + "For an admitted open continuous Monitor whose observation is unavailable, " + + "use this original identity for typed blocked outcome_gap; it settles without " + + "poll or spend and preserves observation clocks/hash/generation. " + + "An already committed Monitor poll needs neither refresh nor spend.", idempotency_key_ref: "$.identity.effect_id", expected_receipt: "durable_writeback_receipt", command_template: writeback, // Autonomous replan already carries this owner in its writeback contract. @@ -62,7 +66,8 @@ export function turnScopedCliSettlementPlan(params: JsonObject): SettlementPlan { kind: "quota_spend", owner: "agent", precondition: "matching durable writeback exists and any declared completion validation " + - "has completed the Todo, or the same Turn has qualified in-flight/replan progress", + "has completed the Todo, or the same Turn has qualified in-flight/replan progress; " + + "skip this step after typed_blocked_writeback_no_spend", idempotency_key_ref: "$.identity.effect_id", expected_receipt: "quota_spend_receipt", command_template: spend, }, diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index 55a6ea80c1..9a06ce894e 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/loopx/control_plane/quota/settlement_readback.ts @@ -56,7 +56,7 @@ import { } from "./heartbeat_receipt_identity.ts"; import { refreshExternalDelivery } from "./refresh_external_delivery.ts"; -import { BLOCKED_WAIT_REQUEST_SCHEMA, prepareBlockedWait, RECEIPT_BOUND_WAIT_REQUEST_SCHEMA, projectReceiptBoundWait } from "./blocked_wait.ts"; +import { BLOCKED_WAIT_REQUEST_SCHEMA, prepareBlockedWait, RECEIPT_BOUND_WAIT_REQUEST_SCHEMA, projectReceiptBoundWait, isMonitorUnavailableWait } from "./blocked_wait.ts"; import {nativeChildReportAdmission} from "../capabilities/native_child_admission.ts"; import { parseQuotaAccountingOwner, @@ -1142,6 +1142,8 @@ function readQuotaSettlementFromRequest( material_change: isMaterialMonitorPoll(monitorPoll), durable_writeback_present: writeback.failure === null, quota_spend_present: spend.failure === null, + unavailable_attempt_present: blockedNoSpend && + isMonitorUnavailableWait(writebackRun?.blocked_retry, identity.todo_id), }); // A committed observation closes its exact Turn independently of whether // the monitor still appears in the current Todo frontier. Reuse the monitor @@ -1157,7 +1159,7 @@ function readQuotaSettlementFromRequest( no_spend_closeout_present: blockedNoSpend, }); - const recovery = request.refresh_retry === null ? null : refreshRecovery( + const refreshDecision = request.refresh_retry === null ? null : refreshRecovery( request.refresh_retry, writebackRun, writeback.failure === null, workspaceCausality?.requirement, writebackRun !== null && snapshot.runs.slice( @@ -1169,6 +1171,14 @@ function readQuotaSettlementFromRequest( (jsonObject(run.agent_vision) !== null || jsonObject(run.vision_checkpoint)?.required === true) ), ); + // A non-material poll need not appear as findWriteback's advancement run. + // Its exact committed observation still fences an incompatible unavailable + // claim; retain the normal workspace/vision supplement paths for real polls. + const recovery = monitorPoll !== null && request.refresh_retry?.delivery_outcome === "outcome_gap" && + isTurnScopedSettlementOutcome(request.refresh_retry.delivery_outcome, + request.refresh_retry.progress_observation, identity.todo_id) + ? {...refreshDecision, decision: "reject", reason: "committed_writeback_payload_conflict"} + : refreshDecision; return { schema_version: QUOTA_SETTLEMENT_READBACK_RESULT_SCHEMA, diff --git a/loopx/control_plane/quota/unsettled_host_turn_recovery.ts b/loopx/control_plane/quota/unsettled_host_turn_recovery.ts index bdeb187fd6..632bd8310f 100644 --- a/loopx/control_plane/quota/unsettled_host_turn_recovery.ts +++ b/loopx/control_plane/quota/unsettled_host_turn_recovery.ts @@ -502,6 +502,11 @@ function recoveryObligation( ? `Recover prior unsettled host Turn for ${bindingId}; re-enter its original ` + "guard, inspect existing effects, then resume and settle verified work under " + "that identity before rerunning the current Turn; do not invent an external wait" + : repair === "monitor_poll" + ? `Recover prior unsettled host Turn for ${bindingId}; record a verified ` + + "original observation with monitor-poll, or use the original identity for " + + "typed blocked writeback if observation is unavailable; never invent a " + + "result hash. Then rerun the current Turn and continue eligible work" : `Recover prior unsettled host Turn for ${bindingId}; use a typed ` + "lifecycle observation, then rerun quota and continue eligible work", repair, diff --git a/loopx/control_plane/work_items/unsettled_host_turn_contract.py b/loopx/control_plane/work_items/unsettled_host_turn_contract.py index 2d7825198d..a8e0e755a7 100644 --- a/loopx/control_plane/work_items/unsettled_host_turn_contract.py +++ b/loopx/control_plane/work_items/unsettled_host_turn_contract.py @@ -82,8 +82,9 @@ def recovery_cli_actions( cadence_arg = f" --cadence {shlex.quote(cadence)}" if cadence else "" return [ ( - "inspect the monitor target and record its exact prior-turn " - "observation; never infer external state from Todo prose, and add " + "If a verified observation was obtained under the original Turn, " + "record it with monitor-poll below. Never infer external state " + "from Todo prose, and add " "--material-change plus a runnable successor only for a real change" ), ( @@ -93,6 +94,23 @@ def recovery_cli_actions( f"{cadence_arg} --turn-instance-id {shlex.quote(prior_turn_id)} " "--execute" ), + ( + "If observation is unavailable, instead close only the original " + "attempt with the typed blocked writeback below: supply stable " + "blocker/evidence IDs and a normal vision decision. It preserves " + "Monitor clocks/hash/generation, creates no poll and spends " + "nothing. An already committed poll needs neither branch." + ), + ( + f"{command_prefix} refresh-state --goal-id {goal_id}{lifecycle_actor_args}" + f" --todo-id {shlex.quote(prior_todo_id)}" + f" --turn-instance-id {shlex.quote(prior_turn_id)}" + " --classification monitor_observation_unavailable" + " --delivery-batch-scale single_surface --delivery-outcome outcome_gap" + " --progress-result-class blocked --progress-blocker-id ''" + " --progress-evidence-id ''" + " --vision-unchanged-reason ''" + ), f"{typed_quota_guard}{current_turn_arg}", ] return [ diff --git a/tests/control_plane/test_monitor_unavailable_closeout_cli.py b/tests/control_plane/test_monitor_unavailable_closeout_cli.py new file mode 100644 index 0000000000..91dea4c8bb --- /dev/null +++ b/tests/control_plane/test_monitor_unavailable_closeout_cli.py @@ -0,0 +1,199 @@ +"""Unavailable observation attempts settle their Turn, never a Monitor poll.""" +from __future__ import annotations + +import json +import shlex +from pathlib import Path + +import pytest + +from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from loopx.cli import main as cli_main +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from test_quota_settlement_cli import ( + AGENT_ID, GOAL_ID, TODO_ID, _classification_count, _spend_run_count, _write_fixture, +) + +MONITOR_ID = "todo_unavailable_monitor" +OBSERVATION_FIELDS = ( + "last_checked_at", "next_due_at", "result_hash", "material_change", + "material_change_generation", "cadence", "expires_at", +) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("scenario", ["unavailable", "capability_restore", "observed"]) +def test_unavailable_monitor_closes_original_turn_without_observation_or_debit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], + provider: str, scenario: str, +) -> None: + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture( + tmp_path, required_capability="network" if scenario == "capability_restore" else None, + ) + state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" + state.write_text(state.read_text().replace( + f"todo_id={TODO_ID} status=open task_class=advancement_task action_kind=validate", + f"todo_id={MONITOR_ID} status=open task_class=continuous_monitor action_kind=observe " + f"claimed_by={AGENT_ID} target_key=synthetic-source watch_only=true cadence=30m " + "last_checked_at=1999-01-01T00%3A00%3A00Z next_due_at=2000-01-01T00%3A00%3A00Z " + f"result_hash={'a' * 64} material_change=false material_change_generation=7", + )) + + def cli(*args: str) -> tuple[int, dict]: + with monkeypatch.context() as context: + context.chdir(project) + rc = cli_main(["--registry", str(registry), "--runtime-root", str(runtime), + "--format", "json", *args]) + return rc, json.loads(capsys.readouterr().out) + + rc, listed = cli("todo", "list", "--goal-id", GOAL_ID) + assert rc == 0, listed + initialize_canonical_authority(runtime, GOAL_ID, build_todo_runtime_shadow_projection( + goal_id=GOAL_ID, todos=listed["todos"], handoff_mode="hard_lease", leases=[], + ), state_path=state, provider=provider) + original_turn = f"unavailable-original-{provider}" + binding = ("--agent-id", AGENT_ID, "--todo-id", MONITOR_ID, + "--turn-instance-id", original_turn) + capabilities = ("--available-capability", "network") if scenario == "capability_restore" else () + if scenario == "capability_restore": + rc, unavailable = cli("quota", "should-run", "--goal-id", GOAL_ID, "--codex-app", *binding) + assert rc != 0 and unavailable["error_code"] == "quota_action_selection_rejected", unavailable + assert _classification_count(runtime, "monitor_observation_unavailable") == 0 + assert _spend_run_count(runtime) == 0 + rc, guard = cli("quota", "should-run", "--goal-id", GOAL_ID, "--codex-app", *binding, *capabilities) + assert rc == 0 and guard["should_run"] is True, guard + assert guard["work_lane_contract"]["obligation"] == "attempt_due_monitor" + assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == MONITOR_ID + plan = guard["interaction_contract"]["cli_channel"]["settlement_plan"] + writeback_step = next(step for step in plan["ordered_steps"] if step["kind"] == "durable_writeback") + assert "observation is unavailable" in writeback_step["precondition"] + rc, lease = cli("task-lease", "acquire", "--goal-id", GOAL_ID, "--todo-id", MONITOR_ID, + "--owner", AGENT_ID, "--idempotency-key", f"unavailable-{provider}") + assert rc == 0 and lease["acquired"] is True, lease + rc, before = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", MONITOR_ID) + assert rc == 0, before + fields = {k: before["todo"].get(k) for k in OBSERVATION_FIELDS} + + # The next real Turn must recover the original identity; it cannot use + # metadata as a result hash or silently move the unfinished receipt to P1. + recovery_turn = f"unavailable-recovery-{provider}" + if scenario != "observed": + rc, recovery = cli("quota", "should-run", "--goal-id", GOAL_ID, "--codex-app", + "--agent-id", AGENT_ID, "--turn-instance-id", recovery_turn, *capabilities) + assert rc == 0 and recovery["effective_action"] == "unsettled_host_turn_recovery", recovery + actions = recovery["interaction_contract"]["cli_channel"]["next_cli_actions"] + command = next(action for action in actions if " refresh-state " in action) + # Consume the actual recovery packet rather than inventing an action. + projected = shlex.split(command) + projected = projected[projected.index("refresh-state"):] + assert projected[projected.index("--todo-id") + 1] == MONITOR_ID + assert projected[projected.index("--turn-instance-id") + 1] == original_turn + + refresh = ("refresh-state", "--goal-id", GOAL_ID, *binding, + "--classification", "monitor_observation_unavailable", + "--delivery-batch-scale", "single_surface", "--delivery-outcome", "outcome_gap", + "--progress-result-class", "blocked", "--progress-blocker-id", "blocker:source-control", + "--progress-evidence-id", "evidence:control-ownership-readback", + "--vision-unchanged-reason", "Source unavailable; monitor acceptance and clocks retained.", + "--no-global-sync", "--suppress-external-sinks") + if scenario == "observed": + rc, poll = cli("quota", "monitor-poll", "--goal-id", GOAL_ID, "--codex-app", *binding, + "--target-key", "synthetic-source", "--result-hash", "synthetic-observed", + "--task-lease-idempotency-key", f"unavailable-{provider}", + "--task-lease-expected-version", str(lease["lease"]["version"]), + "--execute") + assert rc == 0 and poll["appended"] is True, poll + rc, denied = cli(*refresh) + assert rc != 0 and denied["appended"] is False, denied + assert _classification_count(runtime, "monitor_observation_unavailable") == 0 + assert _classification_count(runtime, "quota_monitor_poll") == 1 + assert _spend_run_count(runtime) == 0 + rc, cold = cli("quota", "should-run", "--goal-id", GOAL_ID, "--codex-app", *binding) + assert rc == 0 and cold["interaction_contract"]["mode"] == "heartbeat_settled_skip", cold + return + replacements = { + "": "blocker:source-control", + "": "evidence:control-ownership-readback", + "": "Source unavailable; monitor acceptance and clocks retained.", + } + refresh = (*[replacements.get(arg, arg) for arg in projected], + "--no-global-sync", "--suppress-external-sinks") + without_evidence = list(refresh) + evidence_index = without_evidence.index("--progress-evidence-id") + del without_evidence[evidence_index:evidence_index + 2] + rc, denied = cli(*without_evidence) + assert rc != 0 and denied["appended"] is False, denied + assert _classification_count(runtime, "monitor_observation_unavailable") == 0 + rc, closed = cli(*refresh) + assert rc == 0, closed + wait = closed["blocked_retry"] + assert wait["schema_version"] == "quota_monitor_unavailable_v0" + assert wait["observation_available"] is False and "due_at" not in wait + assert {k: wait["monitor_todo"].get(k) for k in OBSERVATION_FIELDS} == fields + assert closed["settlement_progress"]["state"] == "settled" + assert closed["settlement_progress"]["closeout_kind"] == "typed_blocked_writeback_no_spend" + assert {r["effect_id"] for r in closed["settlement_result"]["receipts"]} == { + guard["heartbeat_receipt"]["settlement_identity"]["effect_id"] + } + rc, replay = cli(*refresh) + assert rc == 0 and replay["idempotent_replay"] is True, replay + assert replay["blocked_retry"] == wait + assert _classification_count(runtime, "monitor_observation_unavailable") == 1 + assert _classification_count(runtime, "quota_monitor_poll") == 0 + assert _spend_run_count(runtime) == 0 + rc, cold = cli("quota", "should-run", "--goal-id", GOAL_ID, "--codex-app", *binding) + assert rc == 0 and cold["should_run"] is False, cold + assert cold["interaction_contract"]["mode"] == "heartbeat_settled_skip" + rc, after = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", MONITOR_ID) + assert rc == 0 and after["todo"]["status"] == "open", after + assert {k: after["todo"].get(k) for k in OBSERVATION_FIELDS} == fields + assert not after["todo"].get("resume_when") + + # A newly executable independent advancement belongs to the same recovery + # host Turn after the original effect is closed, without a third identity. + rc, restored = cli("task-lease", "release", "--goal-id", GOAL_ID, "--todo-id", MONITOR_ID, + "--owner", AGENT_ID, "--idempotency-key", f"unavailable-{provider}", + "--expected-version", str(lease["lease"]["version"])) + assert rc == 0, restored + rc, successor = cli("todo", "add", "--goal-id", GOAL_ID, "--role", "agent", + "--text", "Inspect an independent public source.", "--task-class", "advancement_task", + "--claimed-by", AGENT_ID, "--operation-id", f"independent-{provider}") + assert rc == 0, successor + rc, next_turn = cli("quota", "should-run", "--goal-id", GOAL_ID, "--codex-app", + "--agent-id", AGENT_ID, "--turn-instance-id", recovery_turn, *capabilities) + assert rc == 0 and next_turn["effective_action"] != "unsettled_host_turn_recovery", next_turn + assert next_turn["work_lane_contract"]["lane"] == "advancement_task" + assert _spend_run_count(runtime) == 0 + + +@pytest.mark.parametrize("invalid", ["evidence", "binding"]) +def test_unavailable_closeout_does_not_accept_missing_evidence_or_wrong_identity( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], invalid: str, +) -> None: + project, runtime, registry = _write_fixture(tmp_path) + # This guard binds an advancement. A caller cannot close another Monitor + # or omit the typed evidence required by the existing settlement owner. + def cli(*args: str) -> tuple[int, dict]: + with monkeypatch.context() as context: + context.chdir(project) + rc = cli_main(["--registry", str(registry), "--runtime-root", str(runtime), + "--format", "json", *args]) + return rc, json.loads(capsys.readouterr().out) + + binding = ("--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", "invalid-monitor") + rc, guard = cli("quota", "should-run", "--goal-id", GOAL_ID, "--codex-app", *binding) + assert rc == 0 and guard["should_run"] is True, guard + args = ["refresh-state", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--todo-id", MONITOR_ID if invalid == "binding" else TODO_ID, + "--turn-instance-id", "invalid-monitor", "--classification", "invalid_unavailable", + "--delivery-batch-scale", "single_surface", "--delivery-outcome", "outcome_gap", + "--progress-result-class", "blocked", "--progress-blocker-id", "blocker:source-control", + "--no-global-sync", "--suppress-external-sinks"] + if invalid == "binding": + args += ["--progress-evidence-id", "evidence:control-readback"] + rc, denied = cli(*args) + assert rc != 0 and denied["appended"] is False, denied + assert _classification_count(runtime, "invalid_unavailable") == 0 + assert _spend_run_count(runtime) == 0 diff --git a/tests/control_plane/test_quota_blocked_retry_projection.py b/tests/control_plane/test_quota_blocked_retry_projection.py index 23c5825dce..585d95b260 100644 --- a/tests/control_plane/test_quota_blocked_retry_projection.py +++ b/tests/control_plane/test_quota_blocked_retry_projection.py @@ -94,7 +94,7 @@ def fields(due: str) -> dict: observed_at=observed, ) assert valid["due_at"] == "2026-09-24T14:05:00Z" - with pytest.raises(ValueError, match="same unfinished advancement Todo"): + with pytest.raises(ValueError, match="same unfinished Todo"): require_blocked_retry_wait( fields("2026-09-24T14:05:00Z"), todo_id="todo_other",