fix(T677): keep silent open Codex turns active

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.

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.
         }
     }

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<IdentityLease, "name" | "token" | "holder_pid" | "holder_start" | "session_id"> & { 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<string, string>();
+  private turns = new Map<string, { dev: number; ino: number; size: number; mtimeMs: number; open: boolean }>();
   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<string>();
+      // 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.
     }
   }