diff --git a/CHANGELOG.md b/CHANGELOG.md index a06257f..e433fec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- **Silent Codex turns keep their mailbox identity (T677).** The existing MCP heartbeat records activity while the exact thread's rollout has an unmatched turn start, even when it has no new writes or mbx calls. Matching completion or turn-end records stop that protection; missing, unreadable or rotated evidence falls back to rollout mtime. Lease token, thread and process-birth fences remain in force. An abandoned open turn can conservatively retain a live holder until recovery. + ## 0.5.47 (2026-10-16) - **Service-hosted OpenCode sessions stay wakeable after MCP eviction (T688).** OpenCode stops an idle session's MCP about hourly, which releases its lease. T659 kept standalone sessions wakeable through their plugin's poll; a session hosted by the shared `opencode serve --service` is now a wake target too while that service process (same pid and birth) lives, and the wake posts to the service with `resume: true` and a verified receipt. A lease held by another session of the agent still blocks it. docs/OPENCODE.md recommends plain `opencode` (one service per machine, every project) for long-lived agents. diff --git a/dist/codex-activity.js b/dist/codex-activity.js index 8c6056b..33d6a47 100644 --- a/dist/codex-activity.js +++ b/dist/codex-activity.js @@ -1,13 +1,66 @@ -import { globSync, lstatSync } from "node:fs"; +import { closeSync, fstatSync, globSync, lstatSync, openSync, readFileSync } from "node:fs"; import { homedir } from "node:os"; import { join } from "node:path"; import { activityKey, parseActivity } from "./identity-availability.js"; const THREAD = /^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/i; -/** Metadata-only activity for the existing MCP heartbeat; never opens a rollout or starts a process. */ +/** Exact-thread rollout activity for the existing MCP heartbeat; never starts a process. */ export class CodexActivity { home; paths = new Map(); + turns = new Map(); constructor(home = process.env.CODEX_HOME || join(homedir(), ".codex")) { this.home = home; } + openTurn(thread, path, stat) { + let fd; + try { + fd = openSync(path, "r"); + const observed = fstatSync(fd), prior = this.turns.get(thread); + if (!observed.isFile() || observed.dev !== stat.dev || observed.ino !== stat.ino + || observed.size !== stat.size || observed.mtimeMs !== stat.mtimeMs) + throw new Error("rollout changed"); + if (prior && (prior.dev !== stat.dev || prior.ino !== stat.ino || stat.size < prior.size + || stat.mtimeMs < prior.mtimeMs || (stat.size === prior.size && stat.mtimeMs !== prior.mtimeMs))) { + this.turns.delete(thread); + return false; // A replaced or rewritten rollout is not continuity evidence for the cached turn. + } + if (prior && prior.size === stat.size && prior.mtimeMs === stat.mtimeMs) + return prior.open; + const turns = new Set(); + // shortcut: changed rollouts are rescanned; use incremental reads if heartbeat parsing becomes costly. + for (const line of readFileSync(fd, "utf8").split("\n")) { + if (!line.trim()) + continue; + const row = JSON.parse(line); + if (row?.type !== "event_msg") + continue; + const payload = row.payload; + if (payload?.turn_id !== undefined && (typeof payload.turn_id !== "string" || !payload.turn_id)) + throw new Error("invalid turn id"); + const id = payload?.turn_id ?? ""; + if (payload?.type === "task_started") + turns.add(id); + else if (["task_complete", "turn_complete", "turn_completed", "turn_end", "turn_aborted"].includes(payload?.type)) { + if (id) + turns.delete(id); + else if (turns.size === 1) + turns.clear(); + } + } + const after = fstatSync(fd); + if (after.size !== stat.size || after.mtimeMs !== stat.mtimeMs) + throw new Error("rollout changed"); + const open = turns.size > 0; + this.turns.set(thread, { dev: stat.dev, ino: stat.ino, size: stat.size, mtimeMs: stat.mtimeMs, open }); + return open; + } + catch { + this.turns.delete(thread); + return false; + } + finally { + if (fd !== undefined) + closeSync(fd); + } + } record(store, leases, holder) { if (!THREAD.test(holder.session_id)) return false; @@ -21,10 +74,11 @@ export class CodexActivity { path = join(this.home, found[0]); this.paths.set(thread, path); } - const stat = lstatSync(path), at = Math.floor(stat.mtimeMs); - if (!stat.isFile() || !Number.isSafeInteger(at) || at < 0 || at > Date.now()) + const stat = lstatSync(path), modified = Math.floor(stat.mtimeMs), now = Date.now(); + if (!stat.isFile() || !Number.isSafeInteger(modified) || modified < 0 || modified > now) return false; - // shortcut: rollout writes show activity, but a silent turn remains invisible; use native thread status when available. + // An unmatched start conservatively protects a silent turn until its end is recorded. + const at = this.openTurn(thread, path, stat) ? now : modified; return leases.withHeld(holder.name, holder.token, () => { const lease = store.db.prepare("SELECT * FROM identity_leases WHERE name=?").get(holder.name); if (lease.cli !== "codex" || lease.session_id !== holder.session_id @@ -39,6 +93,7 @@ export class CodexActivity { } catch { this.paths.delete(thread); + this.turns.delete(thread); return false; // Advisory evidence: missing files, lost leases and unknown process evidence write no activity. } } diff --git a/src/codex-activity.ts b/src/codex-activity.ts index ba76ee5..11c62fd 100644 --- a/src/codex-activity.ts +++ b/src/codex-activity.ts @@ -1,4 +1,4 @@ -import { globSync, lstatSync } from "node:fs"; +import { closeSync, fstatSync, globSync, lstatSync, openSync, readFileSync, type Stats } from "node:fs"; import { homedir } from "node:os"; import { join } from "node:path"; import { activityKey, parseActivity } from "./identity-availability.ts"; @@ -8,12 +8,54 @@ import type { Store } from "./store.ts"; type Holder = Pick & { shared: boolean }; const THREAD = /^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/i; -/** Metadata-only activity for the existing MCP heartbeat; never opens a rollout or starts a process. */ +/** Exact-thread rollout activity for the existing MCP heartbeat; never starts a process. */ export class CodexActivity { private home: string; private paths = new Map(); + private turns = new Map(); constructor(home = process.env.CODEX_HOME || join(homedir(), ".codex")) { this.home = home; } + private openTurn(thread: string, path: string, stat: Stats): boolean { + let fd: number | undefined; + try { + fd = openSync(path, "r"); + const observed = fstatSync(fd), prior = this.turns.get(thread); + if (!observed.isFile() || observed.dev !== stat.dev || observed.ino !== stat.ino + || observed.size !== stat.size || observed.mtimeMs !== stat.mtimeMs) throw new Error("rollout changed"); + if (prior && (prior.dev !== stat.dev || prior.ino !== stat.ino || stat.size < prior.size + || stat.mtimeMs < prior.mtimeMs || (stat.size === prior.size && stat.mtimeMs !== prior.mtimeMs))) { + this.turns.delete(thread); + return false; // A replaced or rewritten rollout is not continuity evidence for the cached turn. + } + if (prior && prior.size === stat.size && prior.mtimeMs === stat.mtimeMs) return prior.open; + const turns = new Set(); + // shortcut: changed rollouts are rescanned; use incremental reads if heartbeat parsing becomes costly. + for (const line of readFileSync(fd, "utf8").split("\n")) { + if (!line.trim()) continue; + const row = JSON.parse(line); + if (row?.type !== "event_msg") continue; + const payload = row.payload; + if (payload?.turn_id !== undefined && (typeof payload.turn_id !== "string" || !payload.turn_id)) throw new Error("invalid turn id"); + const id = payload?.turn_id ?? ""; + if (payload?.type === "task_started") turns.add(id); + else if (["task_complete", "turn_complete", "turn_completed", "turn_end", "turn_aborted"].includes(payload?.type)) { + if (id) turns.delete(id); + else if (turns.size === 1) turns.clear(); + } + } + const after = fstatSync(fd); + if (after.size !== stat.size || after.mtimeMs !== stat.mtimeMs) throw new Error("rollout changed"); + const open = turns.size > 0; + this.turns.set(thread, { dev: stat.dev, ino: stat.ino, size: stat.size, mtimeMs: stat.mtimeMs, open }); + return open; + } catch { + this.turns.delete(thread); + return false; + } finally { + if (fd !== undefined) closeSync(fd); + } + } + record(store: Store, leases: IdentityLeases, holder: Holder): boolean { if (!THREAD.test(holder.session_id)) return false; const thread = holder.session_id.toLowerCase(); @@ -25,9 +67,10 @@ export class CodexActivity { path = join(this.home, found[0]); this.paths.set(thread, path); } - const stat = lstatSync(path), at = Math.floor(stat.mtimeMs); - if (!stat.isFile() || !Number.isSafeInteger(at) || at < 0 || at > Date.now()) return false; - // shortcut: rollout writes show activity, but a silent turn remains invisible; use native thread status when available. + const stat = lstatSync(path), modified = Math.floor(stat.mtimeMs), now = Date.now(); + if (!stat.isFile() || !Number.isSafeInteger(modified) || modified < 0 || modified > now) return false; + // An unmatched start conservatively protects a silent turn until its end is recorded. + const at = this.openTurn(thread, path, stat) ? now : modified; return leases.withHeld(holder.name, holder.token, () => { const lease = store.db.prepare("SELECT * FROM identity_leases WHERE name=?").get(holder.name) as unknown as IdentityLease; if (lease.cli !== "codex" || lease.session_id !== holder.session_id @@ -39,6 +82,7 @@ export class CodexActivity { }); } catch { this.paths.delete(thread); + this.turns.delete(thread); return false; // Advisory evidence: missing files, lost leases and unknown process evidence write no activity. } } diff --git a/test/codex-activity.test.ts b/test/codex-activity.test.ts index 684a016..afe17d3 100644 --- a/test/codex-activity.test.ts +++ b/test/codex-activity.test.ts @@ -1,6 +1,6 @@ import { test, type TestContext } from "node:test"; import assert from "node:assert/strict"; -import { lstatSync, mkdirSync, mkdtempSync, rmSync, symlinkSync, utimesSync, writeFileSync } from "node:fs"; +import { appendFileSync, chmodSync, lstatSync, mkdirSync, mkdtempSync, renameSync, rmSync, symlinkSync, utimesSync, writeFileSync } from "node:fs"; import { join, resolve } from "node:path"; import { tmpdir } from "node:os"; import { CodexActivity } from "../src/codex-activity.ts"; @@ -12,6 +12,7 @@ import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js" import { MbxNode } from "../src/node.ts"; const THREAD = "44444444-4444-4444-8444-444444444444", SIBLING = "55555555-5555-4555-8555-555555555555"; +const event = (type: string, turn_id = "turn") => `${JSON.stringify({ type: "event_msg", payload: { type, turn_id } })}\n`; function fixture(t: TestContext) { const home = mkdtempSync(join(tmpdir(), "mbx-codex-activity-")), store = new Store(home); t.after(() => { store.close(); rmSync(home, { recursive: true, force: true }); }); @@ -21,10 +22,10 @@ function fixture(t: TestContext) { const lease = leases.claim("reader", { pid: 10, start: "birth", keyFp: "aaaa-aaaa-aaaa-aaaa", cli: "codex", sessionId: THREAD }); store.set(activityKey(lease.name), JSON.stringify({ at: claimed, shared: true })); const sampler = new CodexActivity(home), holder = { ...lease, shared: true }; - const rollout = (thread: string, at: number, day = "09") => { + const rollout = (thread: string, at: number, day = "09", body = "not JSON; use metadata fallback") => { const dir = join(home, "sessions", "2026", "10", day); mkdirSync(dir, { recursive: true }); const path = join(dir, `rollout-2026-10-${day}T12-00-00-${thread}.jsonl`); - writeFileSync(path, "not JSON; the rollout body must never be read", { mode: 0o000 }); + writeFileSync(path, body, { mode: body.startsWith("{") ? 0o600 : 0o000 }); utimesSync(path, at / 1000, at / 1000); return path; }; @@ -45,7 +46,67 @@ test("exact-thread rollout activity keeps a writing thread live after >10 min wi assert.equal(f.record(), false, "an unchanged rollout must not synthesize another heartbeat"); }); -test("quiet rollout and missing evidence preserve idle recovery and the silent-turn gap", t => { +test("a silent open turn stays unclaimable after >10 min without mbx calls or rollout writes", t => { + const f = fixture(t), path = f.rollout(THREAD, f.claimed - 1000, "09", event("task_started")); + const writtenAt = lstatSync(path).mtimeMs; + assert.equal(f.record(), true); + const later = f.now + SHARED_IDLE_MS + 1000; + t.mock.method(Date, "now", () => later); + assert.equal(f.record(), true, "the same open turn remains active without a new write"); + assert.equal(f.activity().at, later); + assert.equal(lstatSync(path).mtimeMs, writtenAt); + assert.equal(identityAvailability({ lease: f.lease, activity: f.activity(), evidence: { alive: true, start: "birth" }, now: later }).claimable, false); +}); + +test("a completed turn past the idle threshold remains claimable", t => { + const f = fixture(t); + f.rollout(THREAD, f.claimed - 1000, "09", event("task_started") + event("task_complete")); + assert.equal(f.record(), false); + assert.deepEqual(f.activity(), { at: f.claimed, shared: true }); + assert.equal(identityAvailability({ lease: f.lease, activity: f.activity(), evidence: { alive: true, start: "birth" }, now: f.now }).claimable, true); +}); + +for (const end of ["task_complete", "turn_end", "turn_aborted"]) + test(`${end} closes only its matching open turn and restores quiet recovery`, t => { + const f = fixture(t), path = f.rollout(THREAD, f.claimed - 1000, "09", event("task_started", "working") + event(end, "other")); + assert.equal(f.record(), true, "an unrelated turn end must not close this turn"); + appendFileSync(path, event(end, "working")); + utimesSync(path, f.claimed / 1000, f.claimed / 1000); + const later = f.now + SHARED_IDLE_MS + 1000; + t.mock.method(Date, "now", () => later); + assert.equal(f.record(), false, "a completed quiet turn must not refresh observation time"); + assert.equal(identityAvailability({ lease: f.lease, activity: f.activity(), evidence: { alive: true, start: "birth" }, now: later }).claimable, true); + }); + +for (const change of ["missing", "unreadable", "rotated", "truncated"]) + test(`${change} rollout discards cached open-turn protection and falls back to mtime`, t => { + const f = fixture(t), path = f.rollout(THREAD, f.claimed - 1000, "09", event("task_started")); + assert.equal(f.record(), true); + if (change === "missing") rmSync(path); + else if (change === "unreadable") chmodSync(path, 0o000); + else { + if (change === "rotated") renameSync(path, `${path}.old`); + writeFileSync(path, event("task_started"), { mode: 0o600 }); + if (change === "truncated") writeFileSync(path, ""); + utimesSync(path, (f.claimed - 1000) / 1000, (f.claimed - 1000) / 1000); + } + const later = f.now + SHARED_IDLE_MS + 1000; + t.mock.method(Date, "now", () => later); + assert.equal(f.record(), false); + assert.equal(identityAvailability({ lease: f.lease, activity: f.activity(), evidence: { alive: true, start: "birth" }, now: later }).claimable, true); + }); + +test("non-event payloads and malformed rollouts cannot manufacture an open turn", t => { + const f = fixture(t), path = f.rollout(THREAD, f.claimed - 1000, "09", + JSON.stringify({ type: "response_item", payload: { type: "task_started", turn_id: "fake" } })); + assert.equal(f.record(), false); + writeFileSync(path, `${event("task_started")}not JSON\n`); + utimesSync(path, (f.claimed - 1000) / 1000, (f.claimed - 1000) / 1000); + assert.equal(f.record(), false); + assert.deepEqual(f.activity(), { at: f.claimed, shared: true }); +}); + +test("quiet rollout and missing evidence preserve idle recovery", t => { const f = fixture(t); f.rollout(SIBLING, f.now - 1000); assert.equal(f.record(), false, "a sibling's rollout is not this thread's activity"); @@ -82,7 +143,7 @@ test("future, ambiguous, non-file, missing and provisional evidence writes nothi for (const changed of ["token", "holder_pid", "holder_start", "session_id", "cli", "released_at"] as const) test(`rollout evidence is fenced against a changed lease ${changed}`, t => { - const f = fixture(t); f.rollout(THREAD, f.now - 1000); + const f = fixture(t); f.rollout(THREAD, f.now - 1000, "09", event("task_started")); const value = changed === "holder_pid" ? 20 : changed === "released_at" ? f.now : changed === "session_id" ? SIBLING : "changed"; f.store.db.prepare(`UPDATE identity_leases SET ${changed}=? WHERE name=?`).run(value, f.lease.name); assert.equal(f.record(), false); assert.deepEqual(f.activity(), { at: f.claimed, shared: true }); @@ -102,7 +163,7 @@ test("unknown process evidence writes no activity and performs no synthetic rene assert.equal(lease.token, f.lease.token); assert.equal(lease.heartbeat_at, f.claimed); assert.equal(lease.released_at, null); }); -test("the existing MCP heartbeat records exact-thread mtime without another mbx call", async t => { +for (const open of [false, true]) test(`the existing MCP heartbeat records ${open ? "a silent open turn" : "exact-thread mtime"} without another mbx call`, async t => { const home = mkdtempSync(join(tmpdir(), "mbx-codex-heartbeat-")), node = new MbxNode(home, { host: "alpha" }); const client = new Client({ name: "codex-activity", version: "1" }), preload = join(home, "heartbeat.mjs"); writeFileSync(preload, "const interval=globalThis.setInterval; globalThis.setInterval=(fn,ms,...args)=>interval(fn,ms===60000?50:ms,...args);"); @@ -115,14 +176,19 @@ test("the existing MCP heartbeat records exact-thread mtime without another mbx node.store.set(activityKey(lease.name), JSON.stringify({ at: Date.now() - SHARED_IDLE_MS - 1000, shared: true })); const dir = join(home, "sessions", "2026", "10", "16"); mkdirSync(dir, { recursive: true }); const path = join(dir, `rollout-2026-10-16T12-00-00-${THREAD}.jsonl`); - writeFileSync(path, "unreadable rollout contents", { mode: 0o000 }); + writeFileSync(path, open ? event("task_started") : "unreadable rollout contents", { mode: open ? 0o600 : 0o000 }); + if (open) utimesSync(path, (lease.claimed_at - SHARED_IDLE_MS - 1000) / 1000, (lease.claimed_at - SHARED_IDLE_MS - 1000) / 1000); const at = Math.floor(lstatSync(path).mtimeMs); - assert.ok(at > lease.claimed_at); + assert.equal(at > lease.claimed_at, !open); let activity = parseActivity(node.store.get(activityKey(lease.name))); - for (let i = 0; i < 40 && activity?.at !== at; i++) { + for (let i = 0; i < 40 && !(open ? (activity?.at ?? 0) > lease.claimed_at : activity?.at === at); i++) { await new Promise(resolve => setTimeout(resolve, 50)); activity = parseActivity(node.store.get(activityKey(lease.name))); } - assert.deepEqual(activity, { at, shared: true }); + if (open) assert.ok(activity!.at > lease.claimed_at); + else assert.deepEqual(activity, { at, shared: true }); + const first = activity!.at; await new Promise(resolve => setTimeout(resolve, 200)); - assert.equal(parseActivity(node.store.get(activityKey(lease.name)))?.at, at, "an idle heartbeat does not substitute now"); + if (open) assert.ok(parseActivity(node.store.get(activityKey(lease.name)))!.at > first); + else assert.equal(parseActivity(node.store.get(activityKey(lease.name)))?.at, at, "an idle heartbeat does not substitute now"); + assert.equal(Math.floor(lstatSync(path).mtimeMs), at, "heartbeats do not write the rollout"); });