From b150841d0f3c72c5a16132e740a6dde16c1a544e Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Thu, 24 Sep 2026 17:27:38 +0800 Subject: [PATCH] fix(quota): release rejected monitor poll reservations Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../coordination/todo_monitor_poll.ts | 20 +++++-- loopx/control_plane/quota/monitor_poll.py | 20 +++++-- .../quota/monitor_poll_commit.ts | 52 ++++++++++++++++++- .../control_plane/test_leased_monitor_poll.py | 31 +++++++++++ .../quota_monitor_poll_commit.test.ts | 46 ++++++++++++++++ 5 files changed, 159 insertions(+), 10 deletions(-) diff --git a/loopx/control_plane/coordination/todo_monitor_poll.ts b/loopx/control_plane/coordination/todo_monitor_poll.ts index fea14c2645..83ece163ff 100644 --- a/loopx/control_plane/coordination/todo_monitor_poll.ts +++ b/loopx/control_plane/coordination/todo_monitor_poll.ts @@ -35,6 +35,14 @@ export interface CoordinationMonitorPollInput { now?: Date; } +/** The request identity used by both business receipts and no-effect replies. */ +export function monitorPollRequestHash(input: Pick): string { + return canonicalAuthoritySha256({goal_id: input.goal_id, observation: input.observation, + intent: input.intent, actor_agent_id: input.actor_agent_id, dry_run: input.dry_run, + ...(input.lease_proof ? {lease_proof: input.lease_proof} : {})}); +} + function failure(reason_code: string, reason: string): JsonObject & {schema_version: typeof COORDINATION_MONITOR_POLL_RESULT_SCHEMA} { return {schema_version: COORDINATION_MONITOR_POLL_RESULT_SCHEMA, status: "failed", changed: false, reason_code, reason}; } @@ -170,9 +178,7 @@ export async function executeCoordinationMonitorPoll(store: AuthorityStore, try { input = normalize(raw); } catch (error) { return failure("invalid_monitor_poll_request", String(error)); } // Original wire identity, before any normalization/default route inference. - const hash = canonicalAuthoritySha256({goal_id: input.goal_id, observation: input.observation, - intent: input.intent, actor_agent_id: input.actor_agent_id, dry_run: input.dry_run, - ...(input.lease_proof ? {lease_proof: input.lease_proof} : {})}); + const hash = monitorPollRequestHash(input); const receipt = monitorReceipt(input, hash); const previous = await receipt.read(store); if (previous) return previous; @@ -183,7 +189,13 @@ export async function executeCoordinationMonitorPoll(store: AuthorityStore, if (head.status !== "loaded") return {schema_version: COORDINATION_MONITOR_POLL_RESULT_SCHEMA, ...head}; let plan: ReturnType; try { plan = planWriteback(input, head.head); } - catch (error) { return failure("monitor_poll_rejected", error instanceof Error ? error.message : String(error)); } + catch (error) { + // Receipt lookup succeeded and planning failed before commit. Only this + // boundary can certify no effect; outages and commit failures cannot. + return {...failure("monitor_poll_rejected", error instanceof Error ? error.message : String(error)), + no_effect: {schema_version: "monitor_poll_no_effect_v0", goal_id: input.goal_id, + operation_id: input.operation_id, request_sha256: hash}}; + } if (!await authoritySourcesCurrent()) return failure(AUTHORITY_SOURCE_CHANGED.code, AUTHORITY_SOURCE_CHANGED.reason); if (input.dry_run) return {schema_version: COORDINATION_MONITOR_POLL_RESULT_SCHEMA, status: "planned", changed: true, writeback: plan.writeback, provider_revision: head.provider_revision}; diff --git a/loopx/control_plane/quota/monitor_poll.py b/loopx/control_plane/quota/monitor_poll.py index fa136205ff..4ac85ba6f4 100644 --- a/loopx/control_plane/quota/monitor_poll.py +++ b/loopx/control_plane/quota/monitor_poll.py @@ -815,11 +815,21 @@ def transact() -> tuple[dict[str, Any], dict[str, Any]]: raise TypeError("TypeScript monitor-poll preflight omitted provider plan") if registry_path is None: raise ValueError("monitor todo writeback requires registry_path") - provider_receipt = _provider_writeback( - plan, - registry_path=registry_path, - runtime_root=runtime_root, - ) + from ..coordination.local_authority import LocalCoordinationAuthorityUnavailable + + try: + provider_receipt = _provider_writeback( + plan, + registry_path=registry_path, + runtime_root=runtime_root, + ) + except LocalCoordinationAuthorityUnavailable as exc: + # Transport the owner's typed negative evidence. TypeScript alone + # decides whether it releases the exact pending reservation. An + # outage/ambiguous commit carries no no-effect proof and is retained. + if execute and exc.payload.get("no_effect") is not None: + _native_result(_request(phase="provider_rejected", provider_receipt=exc.payload, **common)) + raise status_warning = None if execute: after_status, status_warning = _reload_status_after_monitor_writeback( diff --git a/loopx/control_plane/quota/monitor_poll_commit.ts b/loopx/control_plane/quota/monitor_poll_commit.ts index ed8c5d17cd..721a3c3292 100644 --- a/loopx/control_plane/quota/monitor_poll_commit.ts +++ b/loopx/control_plane/quota/monitor_poll_commit.ts @@ -1,4 +1,5 @@ import {decodeTaskLeaseProof, type TaskLeaseProof} from "../coordination/task_lease_proof.ts"; +import {monitorPollRequestHash} from "../coordination/todo_monitor_poll.ts"; import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts"; import { AgentScopeFrontierAction } from "../agents/agent_scope_frontier.generated.ts"; import { createHash } from "node:crypto"; @@ -41,7 +42,7 @@ const MONITOR_TARGET_SCHEMA = "quota_monitor_target_v0"; const MONITOR_TODO_PROVIDER_PLAN_SCHEMA = "monitor_poll_todo_provider_plan_v0"; const LEASED_MONITOR_TODO_PROVIDER_PLAN_SCHEMA = "monitor_poll_todo_provider_plan_v1"; const MONITOR_TODO_WRITEBACK_SCHEMA = "monitor_poll_todo_writeback_v0"; -const MONITOR_PHASES = ["event", "preflight", "commit"] as const; +const MONITOR_PHASES = ["event", "preflight", "commit", "provider_rejected"] as const; const MONITOR_SOURCES = ["heartbeat", "controller", "adapter", "visible-goal"] as const; const EXTERNAL_MONITOR_POLICIES = new Set([ "material_transition_only", @@ -59,6 +60,7 @@ type MonitorSource = (typeof MONITOR_SOURCES)[number]; type MonitorStatus = | "preview" | "provider_required" + | "aborted" | "written" | "replayed" | "repaired" @@ -1890,11 +1892,38 @@ function effectConflict( ); } +function validateNoEffect(receipt: JsonObject | null, plan: MonitorProviderPlan): void { + const proof = receipt?.no_effect as JsonObject | undefined; + const observation = Object.fromEntries([ + "todo_id", "target_key", "result_hash", "material_change", "generated_at", + "cadence", "next_due_at", "reason_summary", + ].map(key => [key, plan[key]])); + const intent = Object.fromEntries([ + "next_agent_todo", "next_action_kind", "next_task_repository", "next_required_capabilities", + "next_continuation_policy", "next_target_key", "next_claimed_by", "next_user_todo", "next_user_task_class", + ].map(key => [key, plan[key]])); + const hash = monitorPollRequestHash({goal_id: plan.goal_id, actor_agent_id: plan.agent_id, + dry_run: !plan.execute, observation, intent, lease_proof: plan.lease_proof}); + if (receipt?.schema_version !== "loopx_coordination_monitor_poll_result_v0" || + receipt.status !== "failed" || receipt.changed !== false || + receipt.reason_code !== "monitor_poll_rejected" || + !["file_v0", "sqlite_v0", "postgresql_v0"].includes(String(receipt.source_authority)) || + receipt.decision_read_from_provider !== true || receipt.legacy_fallback_used !== false || + proof?.schema_version !== "monitor_poll_no_effect_v0" || + proof.goal_id !== plan.goal_id || proof.operation_id !== plan.monitor_effect_id || proof.request_sha256 !== hash) { + throw new EffectRuntimeRequestError("provider rejection does not prove this pending Monitor request had no effect", + "monitor_poll_no_effect_unproven"); + } +} + export async function evaluateQuotaMonitorPollCommit( value: unknown, ): Promise { const request = requestObject(value); const fingerprint = requestDigest(request); + if (request.phase === "provider_rejected" && !request.execute) { + throw new EffectRuntimeRequestError("provider rejection recovery requires execute"); + } if (request.phase === "event") { const record = buildRecord(request, admission(request)); return result( @@ -2003,6 +2032,9 @@ export async function evaluateQuotaMonitorPollCommit( ); } if (existing.status !== "provider_pending") { + if (request.phase === "provider_rejected") { + throw new EffectRuntimeRequestError("cannot abort a prepared or committed Monitor effect"); + } return await replayDurableReceipt( request, fingerprint, @@ -2012,6 +2044,24 @@ export async function evaluateQuotaMonitorPollCommit( } } + if (request.phase === "provider_rejected") { + if (!existing || existing.status !== "provider_pending") { + throw new EffectRuntimeRequestError("provider rejection recovery requires an exact pending Monitor receipt"); + } + validateNoEffect(request.provider_receipt, providerPlanObject(existing.provider_plan)); + const bytes = await readOptionalBytes(indexPath); + if (!pendingIndexHistoryIntact(existing, bytes) || + matchingIndexRecord(indexRecords(bytes?.toString("utf8") ?? null), request.effect_id)) { + throw new EffectRuntimeRequestError("cannot abort Monitor effect with conflicting index history"); + } + // Remove only this proven uncommitted reservation under the effect lock. + // Any timeout, unknown outcome, changed request or committed effect keeps + // its original recovery fence and never reaches this branch. + await rm(receiptPath); + return result(request, fingerprint, "aborted", null, {ok: false, appended: false}, + "provider rejected before commit; pending Monitor reservation released"); + } + // A pending v1 receipt preserves the decision that admitted this effect. // Validate that historical basis, never the post-business-commit projection. // It only authorizes settlement; the provider still fences any new mutation. diff --git a/tests/control_plane/test_leased_monitor_poll.py b/tests/control_plane/test_leased_monitor_poll.py index 057727caed..f54a361e6e 100644 --- a/tests/control_plane/test_leased_monitor_poll.py +++ b/tests/control_plane/test_leased_monitor_poll.py @@ -20,6 +20,37 @@ PROOF = {"idempotency_key": "monitor-execution", "expected_version": 3} +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_rejected_missing_proof_can_retry_same_turn_with_valid_lease(tmp_path, monkeypatch, provider): + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, runtime, _state, monitor = _canonical(tmp_path, provider=provider, lease=LEASE) + turn = ["--turn-instance-id", "monitor-rejected-proof", + "--available-capability", "network", "--available-capability", "external_evidence_poll"] + guard = run_json_cli("quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--runtime-profile", "generic_cli", *turn, registry_path=registry, runtime_root=runtime) + assert guard["selected_todo"]["todo_id"] == monitor["todo_id"] + before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID, include_leases=True) + args = arguments(monitor) + start = args.index("--task-lease-idempotency-key") + missing_proof = args[:start] + args[start + 4:] + code, rejected = run_json_cli_result(*missing_proof, *turn, registry_path=registry, runtime_root=runtime) + assert code != 0 + assert "lease proof" in json.dumps(rejected) + assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID, include_leases=True) == before + pending = runtime / "goals" / GOAL_ID / "runs" / ".transactions" / "quota-monitor-poll" + assert not list(pending.glob("*.json")), "definitively rejected provider request must not reserve the Turn effect" + result = run_json_cli(*args, *turn, registry_path=registry, runtime_root=runtime) + assert result["ok"] is True + assert result["todo_writeback"]["lease_proof"] == PROOF + assert run_json_cli(*args, *turn, registry_path=registry, runtime_root=runtime)["replayed"] is True + after = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID, include_leases=True) + assert after["leases"] == before["leases"] + assert len(after["todos"]) == len(before["todos"]) + 1 + records = [json.loads(line) for line in (runtime / "goals" / GOAL_ID / "runs" / "index.jsonl").read_text().splitlines()] + assert sum(row.get("classification") == "quota_monitor_poll" for row in records) == 1 + assert all(row.get("classification") != "quota_slot_spend" for row in records) + + def arguments(monitor): return ["quota", "monitor-poll", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--runtime-profile", "generic_cli", "--todo-id", monitor["todo_id"], diff --git a/tests/control_plane_ts/quota_monitor_poll_commit.test.ts b/tests/control_plane_ts/quota_monitor_poll_commit.test.ts index 3851680566..abe305bb84 100644 --- a/tests/control_plane_ts/quota_monitor_poll_commit.test.ts +++ b/tests/control_plane_ts/quota_monitor_poll_commit.test.ts @@ -13,6 +13,7 @@ import { } from "../../loopx/control_plane/quota/monitor_poll_commit.ts"; import { EffectRuntimeRequestError } from "../../loopx/control_plane/effect_runtime_errors.ts"; import { resolveTestPython } from "../../scripts/test-python.mjs"; +import { monitorPollRequestHash } from "../../loopx/control_plane/coordination/todo_monitor_poll.ts"; const goalId = "monitor-native-goal"; @@ -967,6 +968,51 @@ test("leased Monitor pending settlement binds the original proof and rejects mis assert.equal(changed.status, "conflict"); }); +test("only an exact provider no-effect rejection releases a pending Monitor reservation", async t => { + const runtime = await tempRuntime(t); + const params = request({phase: "preflight", runtime_root: runtime, execute: true, + effect_id: "lease-retry-after-rejection", + observation: observation({todo_id: "todo_public_monitor", result_hash: "observed-a"})}); + const first = await evaluateQuotaMonitorPollCommit(params); + const plan = first.provider_plan!; + const receiptPath = join(runtime, "goals", goalId, "runs", ".transactions", "quota-monitor-poll", + `${createHash("sha256").update(String(params.effect_id)).digest("hex").slice(0, 24)}.json`); + const pendingBytes = await readFile(receiptPath, "utf8"); + const rejection = {schema_version: "loopx_coordination_monitor_poll_result_v0", status: "failed", + changed: false, reason_code: "monitor_poll_rejected", source_authority: "file_v0", + decision_read_from_provider: true, legacy_fallback_used: false, + no_effect: {schema_version: "monitor_poll_no_effect_v0", goal_id: goalId, + operation_id: params.effect_id, + request_sha256: monitorPollRequestHash({goal_id: goalId, + actor_agent_id: plan.agent_id as string, dry_run: false, + observation: {todo_id: plan.todo_id, target_key: plan.target_key, result_hash: plan.result_hash, + material_change: plan.material_change, generated_at: plan.generated_at, cadence: plan.cadence, + next_due_at: plan.next_due_at, reason_summary: plan.reason_summary}, + intent: {next_agent_todo: plan.next_agent_todo, next_action_kind: plan.next_action_kind, + next_task_repository: plan.next_task_repository, + next_required_capabilities: plan.next_required_capabilities, + next_continuation_policy: plan.next_continuation_policy, next_target_key: plan.next_target_key, + next_claimed_by: plan.next_claimed_by, next_user_todo: plan.next_user_todo, + next_user_task_class: plan.next_user_task_class}})}}; + const release = {...params, phase: "provider_rejected", provider_receipt: rejection}; + for (const unproven of [null, {...rejection, no_effect: null}, + {...rejection, no_effect: {...rejection.no_effect, request_sha256: "wrong"}}, + {...rejection, source_authority: "legacy"}, {...rejection, reason_code: "provider_timeout"}]) { + await assert.rejects(evaluateQuotaMonitorPollCommit({...release, provider_receipt: unproven})); + assert.equal(await readFile(receiptPath, "utf8"), pendingBytes); + } + const aborted = await evaluateQuotaMonitorPollCommit(release); + assert.equal(aborted.status, "aborted"); + await assert.rejects(readFile(receiptPath), {code: "ENOENT"}); + const proof = {idempotency_key: "monitor-execution", expected_version: 3}; + const corrected = await evaluateQuotaMonitorPollCommit({...params, + schema_version: QUOTA_LEASED_MONITOR_POLL_COMMIT_REQUEST_SCHEMA, + observation: observation({todo_id: "todo_public_monitor", result_hash: "observed-a", lease_proof: proof})}); + assert.equal(corrected.status, "provider_required"); + assert.deepEqual(corrected.provider_plan?.lease_proof, proof); + assert.equal((await evaluateQuotaMonitorPollCommit(release)).status, "conflict"); +}); + test("pending admission is scoped, mandatory in v1, and preserves bounded v0 recovery", async t => { const runtime = await tempRuntime(t); const effect = "pending-admission-compatibility";