fix(release): preserve cycle observations and stable recovery fixtures

docs/reference/usage-ping.md

@@ -265,6 +265,12 @@ backups of authority providers, or public projections. Normal
 CLI invocation reads only a small local hint; a detached Node process owns
 measurement, locks and network I/O. A first-use/settings operation may wait for
 local Node execution, never for a collector connection.
+Quota-cycle start/spend observations may wait up to two seconds for this local
+statistics lock in the detached worker; ordinary observers retain 250 ms.
+Contention beyond that bound remains unobserved. This does not delay business
+commands, retry network delivery or change quota settlement. Consent and its
+generation are rechecked after acquiring the lock, so queued observations cannot
+restore withdrawn collection.
 Offline `migrate-local-state` commands do not schedule these observations,
 because their previews and rollback receipts bind the machine-state bytes.
 

loopx/control_plane/runtime/usage_statistics.ts

@@ -271,7 +271,9 @@ export async function observe(path: string, ctx: Context, generation: string, co
       catch { diagnosticRequest = Promise.resolve(0); }
     }
     return true;
-  }, 250); // Detached workers may briefly contend with startup; never block the host command.
+  // Fresh cycle markers cannot be recovered by replaying a settled command.
+  // Wait through brief observer contention only in this detached worker.
+  }, cycle ? 2000 : 250);
   if (!allowed) return { sent: false, reason: "blocked" };
   let sent = false;
   const outgoing: Array<readonly [string, Parameters<Post>[1] | null]> = [

tests/control_plane_ts/usage_statistics_sources.test.ts

@@ -1,5 +1,6 @@
 import assert from "node:assert/strict";
 import test from "node:test";
+import { randomUUID } from "node:crypto";
 import { mkdtemp, writeFile, appendFile, readFile, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -8,6 +9,7 @@ import type { CycleObservation } from "../../loopx/control_plane/runtime/usage_s
 import { readCodexTiming } from "../../loopx/control_plane/runtime/usage_statistics_codex.ts";
 import { recordGoalUsage, goalPreview } from "../../loopx/control_plane/runtime/usage_statistics_goals.ts";
 import { configure, inspect, observe } from "../../loopx/control_plane/runtime/usage_statistics.ts";
+import { acquireFileMutationLock, releaseFileMutationLock } from "../../loopx/control_plane/effect_runtime_io.ts";
 const at = Date.parse("2026-09-01T10:00:00Z");
 const key = "a".repeat(64), lane = "b".repeat(64), turn = "c".repeat(64);
 const cycle = (phase: "start" | "spend", time: number, extra = {}): CycleObservation => ({ key, lane, turn, phase, at: time, host: "codex-app", ...extra });
@@ -17,6 +19,69 @@ async function fixture(t: test.TestContext) {
 }
 const event = (type: string, time: number, fields = {}) => JSON.stringify({ type: "event_msg", timestamp: new Date(time).toISOString(), payload: { type, ...fields } }) + "\n";
 
+test("cycle observations survive a short local statistics lock without inventing another cycle", async t => {
+  const path = join(await fixture(t), "usage.json");
+  const ctx = { env: { LOOPX_USAGE_PING_ENDPOINT: "http://127.0.0.1:1/v1/ping" }, version: "1.4.2", python: "3.11", channel: "source", now: new Date(at + 2000) };
+  await configure(path, ctx, "enable");
+  const generation = JSON.parse(await readFile(path, "utf8")).generation;
+  for (const input of [cycle("start", at), cycle("spend", at + 1000)]) {
+    const lock = await acquireFileMutationLock(path);
+    const released = new Promise<void>((resolve, reject) => {
+      setTimeout(() => { releaseFileMutationLock(lock.targetPath, lock.token, null, true).then(() => resolve(), reject); }, 500);
+    });
+    try {
+      await observe(path, ctx, generation, null, async () => 204, undefined, input);
+    } finally {
+      await released;
+    }
+  }
+  const state = JSON.parse(await readFile(path + ".cycles", "utf8"));
+  assert.equal(state.cycles.length, 1);
+  assert.equal(state.cycles[0].start, at);
+  assert.equal(state.cycles[0].end, at + 1000);
+  assert.equal(state.cycles[0].exact, true);
+});
+
+test("cycle observations retain a bounded contention failure and recover only after release", { timeout: 10000 }, async t => {
+  const path = join(await fixture(t), "usage.json");
+  const ctx = { env: { LOOPX_USAGE_PING_ENDPOINT: "http://127.0.0.1:1/v1/ping" }, version: "1.4.2", python: "3.11", channel: "source", now: new Date(at) };
+  await configure(path, ctx, "enable");
+  const generation = JSON.parse(await readFile(path, "utf8")).generation;
+  const lock = await acquireFileMutationLock(path);
+  try {
+    await assert.rejects(observe(path, ctx, generation, null, async () => 204, undefined, cycle("start", at)));
+    await assert.rejects(readFile(path + ".cycles"), /ENOENT/);
+  } finally {
+    await releaseFileMutationLock(lock.targetPath, lock.token, null, true);
+  }
+  await observe(path, ctx, generation, null, async () => 204, undefined, cycle("start", at));
+  const state = JSON.parse(await readFile(path + ".cycles", "utf8"));
+  assert.equal(state.cycles.length, 1);
+  assert.equal(state.cycles[0].start, at);
+  assert.equal(state.cycles[0].end, undefined);
+});
+
+test("queued cycle observations recheck withdrawal and consent generation under the lock", async t => {
+  for (const revoked of [{ consent: "disabled" }, { generation: randomUUID() }]) {
+    const path = join(await fixture(t), "usage.json");
+    const ctx = { env: { LOOPX_USAGE_PING_ENDPOINT: "http://127.0.0.1:1/v1/ping" }, version: "1.4.2", python: "3.11", channel: "source", now: new Date(at) };
+    await configure(path, ctx, "enable");
+    const state = JSON.parse(await readFile(path, "utf8"));
+    const lock = await acquireFileMutationLock(path);
+    let sends = 0;
+    const pending = observe(path, ctx, state.generation, null, async () => { sends++; return 204; }, undefined, cycle("start", at));
+    try {
+      // Simulate the already-authorized settings owner while it holds this lock.
+      await writeFile(path, JSON.stringify({ ...state, ...revoked }));
+    } finally {
+      await releaseFileMutationLock(lock.targetPath, lock.token, null, true);
+    }
+    assert.deepEqual(await pending, { sent: false, reason: "blocked" });
+    assert.equal(sends, 0);
+    await assert.rejects(readFile(path + ".cycles"), /ENOENT/);
+  }
+});
+
 test("numeric provider seconds measure the Turn, not a delayed recorder timestamp", async t => {
   const path=join(await fixture(t),"numeric.jsonl");
   await writeFile(path,JSON.stringify({type:"session_meta",payload:{id:"thread"}})+"\n");

tests/test_state_recovery.py

@@ -261,21 +261,45 @@ def test_offline_worker_timeout_does_not_leave_a_copy_or_stop_shared_runtime(tmp
 def test_real_sqlite_history_limit_is_complete_and_overflow_refuses_publication(tmp_path):
     registry, runtime, state = promoted_create_fixture(tmp_path, provider="sqlite")
     project = state.parents[3]
-    module = Path(__file__).parents[1] / "loopx/control_plane/coordination/local_authority_provider.ts"
-    script = f"import {{openLocalAuthorityStore}} from {json.dumps(module.as_uri())};" + """
+    owner = Path(__file__).parents[1] / "loopx/control_plane/coordination"
+    # Seed the read-capacity fixture in one transaction, retaining owner codecs
+    # and a complete real history audit instead of timing 20,000 FULL fsyncs.
+    script = (
+        f"import {{openLocalAuthorityStore}} from {json.dumps((owner / 'local_authority_provider.ts').as_uri())};"
+        f"import {{SqliteAuthorityStore,commitDigest}} from {json.dumps((owner / 'sqlite_authority_store.ts').as_uri())};"
+        f"import {{authorityStateDelta,authorityStateDigest,isAuthorityStateCheckpoint}} from {json.dumps((owner / 'authority_state_log.ts').as_uri())};"
+        "import {DatabaseSync} from 'node:sqlite';"
+    ) + """
 let input='';for await(const chunk of process.stdin)input+=chunk;const r=JSON.parse(input);
 const store=await openLocalAuthorityStore(r.root,'goal-a');const head=await store.loadAuthority();
-if(head.status!=='loaded')throw new Error(JSON.stringify(head));let revision=head.provider_revision;
-for(let i=r.first;i<=r.last;i++){
- const result=await store.commitAuthority({expected_provider_revision:revision,operation_id:`history-capacity-${i}`,
-   events:[],receipts:[],next_projection:head.head});
- if(result.status!=='applied')throw new Error(JSON.stringify(result));revision=result.provider_revision;
-}
+if(!(store instanceof SqliteAuthorityStore)||head.status!=='loaded'||Number(head.cursor)!==r.first-1)
+ throw new Error('Unexpected isolated fixture head');
+const identity=await store.storeIdentity();if(identity.status!=='available')throw new Error('Missing fixture identity');
+const projection=JSON.stringify(head.head),digest=authorityStateDigest(head.head);
+const delta=JSON.stringify(authorityStateDelta(head.head,head.head));
+const db=new DatabaseSync(store.path);db.exec('PRAGMA foreign_keys=ON; BEGIN IMMEDIATE');
+try{
+ const commit=db.prepare('INSERT INTO commits VALUES (?, ?, ?, ?, ?, ?, ?, ?)');
+ const checkpoint=db.prepare('INSERT INTO checkpoints VALUES (?, ?, ?)');
+ for(let i=r.first;i<=r.last;i++){
+  const cursor=BigInt(i),operation=`history-capacity-${i}`;
+  commit.run(i,operation,commitDigest(identity.store_identity,cursor,operation,head.head,[],[]),digest,digest,delta,'[]','[]');
+  if(isAuthorityStateCheckpoint(cursor))checkpoint.run(i,projection,digest);
+ }
+ db.prepare('UPDATE head SET cursor=? WHERE singleton=1').run(r.last);
+ db.exec('COMMIT');
+}finally{db.close();}
+const audit=await store.verifyAuthorityHistory();
+if(audit.status!=='verified'||audit.commits!==r.last)throw new Error(JSON.stringify(audit));
+const readback=await store.loadAuthority();
+if(readback.status!=='loaded'||Number(readback.cursor)!==r.last)throw new Error('Fixture readback mismatch');
+process.stdout.write(JSON.stringify(audit));
 """
     for first, last, expected in [(2, 20000, "readable"), (20001, 20001, "refused")]:
-        subprocess.run(["node", "--no-warnings", "--experimental-sqlite", "--experimental-strip-types",
+        seeded = subprocess.run(["node", "--no-warnings", "--experimental-sqlite", "--experimental-strip-types",
             "--input-type=module", "-e", script], input=json.dumps({"root": str(runtime), "first": first, "last": last}),
             text=True, capture_output=True, check=True, timeout=120)
+        assert json.loads(seeded.stdout)["commits"] == last
         backup = execute_state_backup_plan(build_state_backup_plan(project=project, runtime_root=runtime,
             registry_path=registry, output_dir=tmp_path / "backups", backup_id=f"history-{last}",
             include_registry_projects=False, include_automations=False, include_skills=False))

tests/test_zcode_host_diagnostics.py

@@ -346,7 +346,8 @@ def test_probe_reclaims_persistent_descendants_before_return(tmp_path, monkeypat
     # startup are not constrained by a deadline for deliberately blocked probes.
     if mode not in {"timeout_default", "exit_closed"}:
         monkeypatch.setattr(diagnostics, "PROBE_TIMEOUT_SECONDS", .5)
-    child_script = f"import time; time.sleep({child_startup_delay!r})\nimport pathlib, sys, os; p=pathlib.Path(sys.argv[1]); p.write_text(str(os.getpid())); n=0\nwhile True:\n n+=1; p.with_suffix('.pulse').write_text(str(n)); time.sleep(.01)"
+    # Publishing the ready PID must imply that the first pulse already exists.
+    child_script = f"import time; time.sleep({child_startup_delay!r})\nimport pathlib, sys, os; p=pathlib.Path(sys.argv[1]); p.with_suffix('.pulse').write_text('0'); p.write_text(str(os.getpid())); n=0\nwhile True:\n n+=1; p.with_suffix('.pulse').write_text(str(n)); time.sleep(.01)"
     path = tmp_path / "child.pid"
     parent = "import subprocess,sys,time,pathlib,os; p=pathlib.Path(sys.argv[1]); c=subprocess.Popen([sys.executable,'-c',sys.argv[2],str(p)],stdout=subprocess.DEVNULL if sys.argv[3]=='exit_closed' else None,stderr=subprocess.DEVNULL if sys.argv[3]=='exit_closed' else None)\nwhile not p.exists(): time.sleep(.01)\nif sys.argv[3]=='output_limit': print('x'*70000,flush=True)\nif sys.argv[3].startswith('exit'): sys.exit(0)\ntime.sleep(30)"
     for _ in range(2):