fix(quota): release rejected monitor poll reservations

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<CoordinationMonitorPollInput,
+  "goal_id" | "observation" | "intent" | "actor_agent_id" | "dry_run" | "lease_proof">): 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<typeof planWriteback>;
   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};

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(

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<QuotaMonitorPollCommitResult> {
   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.

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"],

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";