fix(usage): frame the bound Codex header on LF instead of one 64 KiB read

The Codex reader took a fixed 65,536-byte chunk, split it at the first
newline and parsed the result. A `session_meta` line is not a short id
record -- Codex's recorder writes `base_instructions` and the dynamic tool
list into it, and observed first lines already reach ~50 KiB -- so a
legitimate header above that chunk truncated the JSON, threw before any
cursor was built, and left the session with `quota_cycle` but never
`codex_turn`. Retrying re-read the same truncated head, so the loss was
permanent.

Read the opening record by LF framing under its own bounded budget
(`HEADER_BUDGET`, 2 MiB) in `HEADER_CHUNK` reads, parse only what the
identity check needs, and give the two non-bindable outcomes a name:
an unfinished header returns without advancing the cursor so the next scan
reads it whole, and a header past the budget fails as
`usage_session_header_too_large` instead of as an identity mismatch.

Validation: the sources test adds a 72 KiB header case (binds, keeps the
identity mismatch rejection and recovers timing), an unfinished-header case
that recovers on the next scan, and an over-budget case that names the
reason; `tests/test_usage_goal.py` drives the same long header through the
real quota observer -> detached TS chain and asserts both measurements
arrive. Reverting to the fixed chunk fails all four.

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>

Author: huangruiteng

loopx/control_plane/runtime/usage_statistics_codex.ts

@@ -1,20 +1,64 @@
 /** Read only timing envelopes from one explicitly bound session, in bounded chunks. */
 import { open } from "node:fs/promises";
+import type { FileHandle } from "node:fs/promises";
 import type { GoalObservation, Host } from "./usage_statistics_goal_contract.ts";
 import { object } from "./usage_statistics_contract.ts";
 export type CodexCursor = { offset: number; inode: string; since: number; seen: number; skipping?: boolean; open?: { id: string; start: number; confirmed: number } };
 const BUDGET = 1024 * 1024;
+// The first line is not a short id record: Codex's recorder writes
+// `base_instructions` and the dynamic tool list into `session_meta`, and the
+// first lines observed in this project's Codex homes reach ~50 KiB. Frame the
+// line by LF under its own bounded budget instead of assuming one fixed 64 KiB
+// chunk holds it, which truncated the JSON and stranded the whole session.
+const HEADER_CHUNK = 65536;
+const HEADER_BUDGET = 2 * 1024 * 1024;
+type HeaderLine = { line: string } | { error: "incomplete" | "too_large" };
 function timestamp(value: unknown): number { return typeof value === "string" ? Date.parse(value) : NaN; }
+/**
+ * Read the opening record whole, framed by LF, up to a fixed budget.
+ *
+ * `incomplete` means the writer has not finished the record yet and the caller
+ * should read it again later; `too_large` means the identity cannot be verified
+ * at all, which is named rather than reported as a mismatch.
+ */
+async function readHeaderLine(file: FileHandle, size: number): Promise<HeaderLine> {
+  const parts: Buffer[] = [];
+  let offset = 0;
+  while (offset < size && offset < HEADER_BUDGET) {
+    const take = Math.min(HEADER_CHUNK, size - offset, HEADER_BUDGET - offset);
+    const buffer = Buffer.alloc(take);
+    const read = await file.read(buffer, 0, take, offset);
+    if (read.bytesRead <= 0) break;
+    const chunk = buffer.subarray(0, read.bytesRead);
+    offset += read.bytesRead;
+    const newline = chunk.indexOf(10);
+    parts.push(newline < 0 ? chunk : chunk.subarray(0, newline));
+    if (newline >= 0) return { line: Buffer.concat(parts).toString("utf8") };
+  }
+  return offset >= HEADER_BUDGET ? { error: "too_large" } : { error: "incomplete" };
+}
 export async function readCodexTiming(path: string, thread: string, previous: CodexCursor | undefined, now: number, key: string, host: Host) {
   const file = await open(path, "r");
   try {
     const stat = await file.stat();
-    const header = Buffer.alloc(65536);
-    const head = await file.read(header, 0, header.length, 0);
-    const first = header.subarray(0, head.bytesRead).toString().split("\n")[0];
-    const meta = JSON.parse(first);
-    if (meta.type !== "session_meta" || (meta.payload?.id ?? meta.payload?.session_id) !== thread) throw new Error("usage_session_identity_mismatch");
     const inode = `${stat.dev}:${stat.ino}`;
+    const header = await readHeaderLine(file, stat.size);
+    if ("error" in header) {
+      if (header.error === "incomplete") {
+        // Nothing may be emitted before the identity is verified, and nothing is
+        // lost: the record is simply not written yet, so read it again later.
+        const cursor: CodexCursor = previous?.inode === inode && previous.offset <= stat.size
+          ? { ...structuredClone(previous), seen: now }
+          : { offset: 0, inode, since: now, seen: now };
+        return { cursor, observations: [] as GoalObservation[] };
+      }
+      throw new Error(`usage_session_header_too_large:${stat.size}`);
+    }
+    let meta: unknown;
+    try { meta = JSON.parse(header.line); } catch { throw new Error("usage_session_header_unparsable"); }
+    const record = object(meta) ? meta : {};
+    const payload = object(record.payload) ? record.payload : {};
+    if (record.type !== "session_meta" || (payload.id ?? payload.session_id) !== thread) throw new Error("usage_session_identity_mismatch");
     const reusable = previous?.inode === inode && previous.offset <= stat.size;
     const cursor: CodexCursor = reusable ? structuredClone(previous!) : { offset: Math.max(0, stat.size - BUDGET), inode, since: now, seen: now };
     cursor.seen = now;

tests/control_plane_ts/usage_statistics_sources.test.ts

@@ -85,6 +85,47 @@ test("opt-out prevents opening bound transcript files; disable erases cycle curs
   await assert.rejects(readFile(path+".cycles"),/ENOENT/);
 });
 
+test("a session_meta header larger than one chunk still binds and recovers timing", async t => {
+  const path=join(await fixture(t),"long-header.jsonl");
+  // Codex's recorder writes base_instructions and the tool list into this line;
+  // 72 KiB already exceeds the old fixed 64 KiB read and is well inside the
+  // framing budget, so it must bind exactly like a short header.
+  const header=JSON.stringify({type:"session_meta",payload:{id:"thread",base_instructions:{text:"i".repeat(72000)},dynamic_tools:[{name:"shell"}]}});
+  assert.ok(header.length>65536,"fixture header must exceed the fixed chunk this reader used");
+  await writeFile(path,header+"\n"+event("task_started",at-1000,{turn_id:"first",started_at:new Date(at-1000).toISOString()}));
+  let result=await readCodexTiming(path,"thread",undefined,at,key,"codex_app");
+  assert.equal(result.observations.length,0);
+  await assert.rejects(readCodexTiming(path,"other-thread",undefined,at,key,"codex_app"),/identity_mismatch/);
+  await appendFile(path,event("task_complete",at+60000,{turn_id:"first",started_at:new Date(at-1000).toISOString(),completed_at:new Date(at+60000).toISOString()}));
+  result=await readCodexTiming(path,"thread",result.cursor,at+60001,key,"codex_app");
+  assert.equal(result.observations.length,1);
+  assert.equal(result.observations[0].end-result.observations[0].start,60000);
+  assert.equal(result.observations[0].measurement,"codex_turn");
+});
+
+test("an unfinished header waits for the rest of the line instead of failing the session", async t => {
+  const path=join(await fixture(t),"partial-header.jsonl");
+  const header=JSON.stringify({type:"session_meta",payload:{id:"thread",base_instructions:{text:"j".repeat(72000)}}});
+  await writeFile(path,header.slice(0,70000));
+  const waiting=await readCodexTiming(path,"thread",undefined,at,key,"codex_app");
+  assert.equal(waiting.observations.length,0);
+  assert.equal(waiting.cursor.offset,0,"an incomplete header must not advance the cursor past unread bytes");
+  await writeFile(path,header+"\n"+event("task_started",at-1000,{turn_id:"first",started_at:new Date(at-1000).toISOString()}));
+  const started=await readCodexTiming(path,"thread",waiting.cursor,at,key,"codex_app");
+  await appendFile(path,event("task_complete",at+30000,{turn_id:"first",completed_at:new Date(at+30000).toISOString()}));
+  const done=await readCodexTiming(path,"thread",started.cursor,at+30001,key,"codex_app");
+  assert.equal(done.observations.length,1);
+  // A start observed before the first scan is clamped to the cursor's `since`,
+  // so the span reaches back only to the first read that saw this session.
+  assert.equal(done.observations[0].end-done.observations[0].start,30000);
+});
+
+test("a header above the framing budget is named instead of reported as a mismatch", async t => {
+  const path=join(await fixture(t),"oversized-header.jsonl");
+  await writeFile(path,JSON.stringify({type:"session_meta",payload:{id:"thread",base_instructions:{text:"k".repeat(2*1024*1024+16)}}})+"\n");
+  await assert.rejects(readCodexTiming(path,"thread",undefined,at,key,"codex_app"),/usage_session_header_too_large/);
+});
+
 test("oversized transcript content cannot strand later terminal timing", async t => {
   const path=join(await fixture(t),"large.jsonl");
   await writeFile(path,JSON.stringify({type:"session_meta",payload:{id:"thread"}})+"\n");

tests/test_usage_goal.py

@@ -110,10 +110,19 @@ def test_metadata_cannot_redirect_timing_read_outside_selected_home(tmp_path, mo
     assert usage_goal._bound_codex_session(registry, "goal", "agent") is None
 
 
-def test_real_detached_cycle_and_bound_codex_event_reach_shared_ts_aggregator(tmp_path, monkeypatch):
+def _bound_cycle_through_detached_ts(tmp_path, monkeypatch, *, header_characters: int = 0):
+    """Drive the real quota observer -> detached TS chain over one bound session.
+
+    `header_characters` widens the session_meta line the way Codex's recorder
+    does with base_instructions, so the same entry point covers a header that
+    does not fit in one fixed read.
+    """
     from datetime import datetime, timezone
     from pathlib import Path
     registry, rollout = _bound_fixture(tmp_path, monkeypatch)
+    if header_characters:
+        rollout.write_text(json.dumps({"type": "session_meta", "payload": {
+            "id": "thread-a", "base_instructions": {"text": "i" * header_characters}}}) + "\n")
     monkeypatch.setattr(usage_ping, "DEFAULT_RUNTIME_ROOT", tmp_path)
     for name in ("CI", "DO_NOT_TRACK", "LOOPX_USAGE_PING", "LOOPX_USAGE_POLICY"):
         monkeypatch.delenv(name, raising=False)
@@ -150,6 +159,11 @@ def publish(phase):
     goals = Path(str(usage_ping.state_path()) + ".goals")
     wait_for(lambda: len(json.loads(goals.read_text())["goals"]) == 2)
     preview = usage_ping.control("status")["goal_preview"]
+    return preview, cycle_path, goals, publish
+
+
+def test_real_detached_cycle_and_bound_codex_event_reach_shared_ts_aggregator(tmp_path, monkeypatch):
+    preview, cycle_path, goals, publish = _bound_cycle_through_detached_ts(tmp_path, monkeypatch)
     assert {row["measurement"] for row in preview["counters"]} == {"quota_cycle", "codex_turn"}
     assert all(row["host"] == "codex_app" for row in preview["counters"])
     assert "PRIVATE CONTENT" not in cycle_path.read_text() + goals.read_text()
@@ -158,3 +172,18 @@ def publish(phase):
     monkeypatch.setattr(usage_ping, "_detach", lambda *a, **k: pytest.fail("disabled observer spawned"))
     publish("start")
     assert not cycle_path.exists() and not goals.exists()
+
+
+def test_bound_session_with_a_long_metadata_header_still_reaches_ts(tmp_path, monkeypatch):
+    """Codex records base_instructions in session_meta; a long header must bind."""
+    characters = 72_000
+    header = json.dumps({"type": "session_meta", "payload": {
+        "id": "thread-a", "base_instructions": {"text": "i" * characters}}}) + "\n"
+    assert len(header) > 65536, "fixture header must exceed one fixed read"
+    preview, _, _, publish = _bound_cycle_through_detached_ts(
+        tmp_path, monkeypatch, header_characters=characters)
+    assert {row["measurement"] for row in preview["counters"]} == {"quota_cycle", "codex_turn"}
+    assert all(row["host"] == "codex_app" for row in preview["counters"])
+    usage_ping.control("disable")
+    monkeypatch.setattr(usage_ping, "_detach", lambda *a, **k: pytest.fail("disabled observer spawned"))
+    publish("start")