fix(codex-app): make turn ownership and recovery durable

src/adapters/backend/herdr-backend.ts

@@ -353,6 +353,7 @@ export class HerdrBackend implements SessionBackend {
   private lastText = '';
   private exited = false;
   private started = false;
+  private actuallyReattached = false;
   private cols = 200;
   private rows = 50;
   private agentProbeFailures = 0;
@@ -467,7 +468,7 @@ export class HerdrBackend implements SessionBackend {
   }
 
   get isReattach(): boolean {
-    return this.opts.isReattach ?? false;
+    return this.actuallyReattached;
   }
 
   spawn(bin: string, args: string[], opts: SpawnOpts): void {
@@ -497,6 +498,7 @@ export class HerdrBackend implements SessionBackend {
 
     const external = this.opts.externalTarget;
     if (external) {
+      this.actuallyReattached = false;
       this.paneId = external.paneId ?? external.target;
     } else {
       // Reuse an existing `botmux` agent ONLY when we're genuinely re-attaching
@@ -507,8 +509,9 @@ export class HerdrBackend implements SessionBackend {
       // row from persisted metadata, and reuse would skip `agent start` so the
       // new command never ran. killSession() now deletes that metadata, but we
       // also gate reuse on isReattach so a stale row can never be adopted.
-      const existing = this.isReattach ? this.getAgent() : undefined;
+      const existing = this.opts.isReattach ? this.getAgent() : undefined;
       if (existing) {
+        this.actuallyReattached = true;
         this.paneId = existing.pane_id;
       } else if (herdrUsesPaneAgentStart()) {
         this.paneId = this.startPaneAgent(bin, args, opts);
@@ -540,25 +543,31 @@ export class HerdrBackend implements SessionBackend {
     //   - Re-attach / external adopt: snapshot the current screen so we only
     //     stream new deltas. Worker.ts explicitly seeds the initial screen
     //     via captureCurrentScreen() in those paths.
-    this.lastText = (this.isReattach || this.opts.externalTarget) ? this.readRecentAnsi() : '';
+    this.lastText = (this.actuallyReattached || this.opts.externalTarget) ? this.readRecentAnsi() : '';
     this.startPolling();
     this.startStatusWatcher();
   }
 
-  write(data: string): void {
-    if (this.exited) return;
+  write(data: string): boolean {
+    if (this.exited) return false;
     const target = this.paneId ?? this.agentName;
-    runHerdr(herdrSessionArgs(this.sessionName, ['pane', 'send-text', target, data]), { timeout: 5000 });
+    return runHerdr(
+      herdrSessionArgs(this.sessionName, ['pane', 'send-text', target, data]),
+      { timeout: 5000 },
+    );
   }
 
-  sendText(text: string): void {
-    this.write(text);
+  sendText(text: string): boolean {
+    return this.write(text);
   }
 
-  sendSpecialKeys(...keys: string[]): void {
-    if (this.exited) return;
+  sendSpecialKeys(...keys: string[]): boolean {
+    if (this.exited) return false;
     const target = this.paneId ?? this.agentName;
-    runHerdr(herdrSessionArgs(this.sessionName, ['pane', 'send-keys', target, ...keys]), { timeout: 5000 });
+    return runHerdr(
+      herdrSessionArgs(this.sessionName, ['pane', 'send-keys', target, ...keys]),
+      { timeout: 5000 },
+    );
   }
 
   pasteText(text: string): void {

src/adapters/backend/pty-backend.ts

@@ -45,8 +45,10 @@ export class PtyBackend implements SessionBackend {
     logger.debug(`[pty] spawned pid=${this.process.pid}`);
   }
 
-  write(data: string): void {
-    this.process?.write(data);
+  write(data: string): boolean {
+    if (!this.process) return false;
+    this.process.write(data);
+    return true;
   }
 
   resize(cols: number, rows: number): void {

src/adapters/backend/sandbox.ts

@@ -883,6 +883,7 @@ export function buildRelayHostEnv(
   const env: NodeJS.ProcessEnv = { ...baseEnv };
   delete env.BOTMUX_SEND_RELAY;
   delete env.BOTMUX_CARD_PREPARED_CONTENT_FILE;
+  delete env.BOTMUX_HOST_RELAY_REQUIRES_CODEX_APP_LEDGER;
   if (preparedContentFile) {
     env.BOTMUX_CARD_LOCAL_LINK_MODE = 'disabled';
     env.BOTMUX_CARD_PREPARED_CONTENT_FILE = preparedContentFile;
@@ -904,7 +905,19 @@ export function startOutboxWatcher(
      *  absent the relay still runs, but carries NO durable origin — a missing
      *  hook must never let the sandbox promote its own origin fields. */
     authorize?: (claim: { capability?: string }) =>
-      | { ok: true; origin: { turnId?: string; dispatchAttempt?: number } }
+      | {
+          ok: true;
+          origin: {
+            turnId?: string;
+            dispatchAttempt?: number;
+            /** The worker matched an unsettled Codex App ledger entry. The
+             * host child must still find that exact entry before any provider
+             * side effect; terminal settlement/revocation between authorize
+             * and re-exec therefore fails closed instead of degrading to an
+             * ordinary mutable-session send. */
+            requiresCodexAppLedger?: boolean;
+          };
+        }
       | { ok: false; error: string };
     cliPath?: string;
   } = {},
@@ -1028,6 +1041,11 @@ export function startOutboxWatcher(
       if (trustedOrigin?.dispatchAttempt !== undefined) {
         requestEnv.BOTMUX_DISPATCH_ATTEMPT = String(trustedOrigin.dispatchAttempt);
       }
+      if (trustedOrigin?.requiresCodexAppLedger) {
+        requestEnv.BOTMUX_HOST_RELAY_REQUIRES_CODEX_APP_LEDGER = '1';
+      } else {
+        delete requestEnv.BOTMUX_HOST_RELAY_REQUIRES_CODEX_APP_LEDGER;
+      }
       const child = spawn(process.execPath, [cli, 'send', ...hostArgs], { env: requestEnv });
       let out = '', err = '';
       child.stdout.on('data', d => { out += d; });

src/adapters/backend/tmux-backend.ts

@@ -370,8 +370,10 @@ export class TmuxBackend implements SessionBackend {
    *  file's cwd field so a recycled PID can't mislead the resolver. */
   cliCwd?: string;
 
-  write(data: string): void {
-    this.process?.write(data);
+  write(data: string): boolean {
+    if (!this.process) return false;
+    this.process.write(data);
+    return true;
   }
 
   /**

src/adapters/backend/tmux-pipe-backend.ts

@@ -247,9 +247,9 @@ export class TmuxPipeBackend implements SessionBackend {
     }
   }
 
-  write(data: string): void {
+  write(data: string): boolean {
     // No PTY to write to — interpret as a literal send-keys.
-    this.sendText(data);
+    return this.sendText(data);
   }
 
   sendText(text: string): boolean {

src/adapters/backend/zellij-backend.ts

@@ -201,20 +201,24 @@ export class ZellijBackend implements SessionBackend {
   // are forwarded verbatim to the focused pane — so every input path collapses
   // to pty.write(), exactly like TmuxBackend.write().
 
-  write(data: string): void {
-    this.process?.write(data);
+  write(data: string): boolean {
+    if (!this.process) return false;
+    this.process.write(data);
+    return true;
   }
 
   /** Literal text, no Enter. */
-  sendText(text: string): void {
-    this.process?.write(text);
+  sendText(text: string): boolean {
+    return this.write(text);
   }
 
   /** Special keys by tmux-style name (Enter, Escape, C-c, M-Enter, …). */
-  sendSpecialKeys(...keys: string[]): void {
+  sendSpecialKeys(...keys: string[]): boolean {
+    if (!this.process) return false;
     for (const key of keys) {
-      this.process?.write(tmuxKeyToBytes(key));
+      this.process.write(tmuxKeyToBytes(key));
     }
+    return true;
   }
 
   /** Bracketed paste: wrap with \e[200~ … \e[201~ so TUIs (CoCo/Ink/Codex)

src/adapters/backend/zellij-observe-backend.ts

@@ -182,19 +182,22 @@ export class ZellijObserveBackend implements ObserveBackend {
 
   // ── Input: targeted `action` calls (focus-neutral, non-invasive) ──
 
-  write(data: string): void {
-    this.writeBytes(data);
+  write(data: string): boolean {
+    return this.writeBytes(data);
   }
 
   /** Literal text via write-chars (preserves UTF-8). */
-  sendText(text: string): void {
-    if (!text) return;
-    this.action(['write-chars', '--pane-id', this.paneId, '--', text]);
+  sendText(text: string): boolean {
+    if (!text) return true;
+    return this.action(['write-chars', '--pane-id', this.paneId, '--', text]) !== null;
   }
 
   /** Special keys by tmux-style name → raw bytes → `action write`. */
-  sendSpecialKeys(...keys: string[]): void {
-    for (const key of keys) this.writeBytes(tmuxKeyToBytes(key));
+  sendSpecialKeys(...keys: string[]): boolean {
+    for (const key of keys) {
+      if (!this.writeBytes(tmuxKeyToBytes(key))) return false;
+    }
+    return true;
   }
 
   /** Bracketed paste — wrap so TUIs detect the boundary (mirrors paste-buffer -p). */
@@ -207,14 +210,15 @@ export class ZellijObserveBackend implements ObserveBackend {
   /** Write arbitrary bytes via `action write <decimal>…` (handles control/escape).
    *  Chunked so a large web-terminal paste can't blow the argv limit; zellij
    *  serialises the writes in arrival order. */
-  private writeBytes(data: string): void {
-    if (!data) return;
+  private writeBytes(data: string): boolean {
+    if (!data) return true;
     const buf = Buffer.from(data, 'utf-8');
     const CHUNK = 512;
     for (let i = 0; i < buf.length; i += CHUNK) {
       const bytes = Array.from(buf.subarray(i, i + CHUNK), b => String(b));
-      this.action(['write', '--pane-id', this.paneId, ...bytes]);
+      if (this.action(['write', '--pane-id', this.paneId, ...bytes]) === null) return false;
     }
+    return true;
   }
 
   /** Resize is a NO-OP in observe mode — the pane size is the user's, and

src/adapters/cli/codex-app.ts

@@ -6,6 +6,14 @@ import type { CliAdapter, PtyHandle } from './types.js';
 import { writeRunnerInput } from './runner-input.js';
 
 function runnerPath(): string {
+  // Source-level worker integration tests execute through tsx and need the
+  // matching source runner rather than a possibly absent/stale ignored dist
+  // tree. Keep the override strictly test-scoped so production launch
+  // resolution remains canonical and cannot be redirected by ambient env.
+  const testOverride = process.env.NODE_ENV === 'test'
+    ? process.env.BOTMUX_TEST_CODEX_APP_RUNNER_PATH
+    : undefined;
+  if (testOverride) return resolve(testOverride);
   const here = dirname(fileURLToPath(import.meta.url));
   const compiledSibling = resolve(here, '..', '..', 'codex-app-runner.js');
   if (existsSync(compiledSibling)) return compiledSibling;

src/adapters/cli/runner-input.ts

@@ -1,4 +1,4 @@
-import type { PtyHandle } from './types.js';
+import type { PtyHandle, RunnerSubmissionDisposition } from './types.js';
 import type { CodexAppTurnInput } from '../../types.js';
 import { delay } from '../../utils/timing.js';
 
@@ -17,9 +17,9 @@ import { delay } from '../../utils/timing.js';
  * webhook whose full MR JSON is embedded — ~16-21KB after base64) that single
  * injection overruns the pane pty's input buffer (N_TTY's ~4KB read buffer):
  * tmux's write blocks until the reader drains, which takes longer than
- * execFileSync's 5s timeout, so the send-keys is killed and the keystroke is
- * silently dropped — yet the old writeInput still reported `submitted: true`,
- * wedging the session "busy" forever. (Compare claude-code, which throttles
+ * execFileSync's 5s timeout, so Botmux can no longer prove whether the
+ * keystroke landed — yet the old writeInput still reported `submitted: true`,
+ * potentially wedging the session "busy" forever. (Compare claude-code, which throttles
  * its send-keys for exactly this reason; codex-app/mira were the only naive
  * single-shot writers.)
  *
@@ -61,11 +61,10 @@ export function chunkAscii(line: string, maxBytes: number): string[] {
 /**
  * Write one control line to a runner adapter's stdin, chunked + throttled.
  *
- * Returns `{ submitted: false }` when any chunk (or the final Enter) fails to
- * write — the tmux backend's send methods return `false` on a dropped-but-pane-
- * alive keystroke, so a genuine drop is now surfaced to the worker (which raises
- * a submit-failure notice + recheck so the user can retry) instead of being
- * swallowed as a false success.
+ * Returns `{ submitted: false }` when any chunk (or the final Enter) cannot be
+ * confirmed. A tmux timeout is ambiguous: bytes may still have reached the
+ * pane. `submissionDisposition` therefore tells the worker whether the new
+ * frame was provably untouched or the runner generation must be fenced.
  *
  * Buffer-hygiene contract (the runner only clears its stdin buffer on a newline,
  * see handleInput in codex-app-runner.ts / mira-runner.ts — a half-written
@@ -74,28 +73,35 @@ export function chunkAscii(line: string, maxBytes: number): string[] {
  *   - Pre-flush: emit one Enter before writing, terminating any partial line a
  *     prior failed write may have left behind (runner discards the fragment as
  *     bad input; an empty buffer just ignores the blank line).
- *   - On a dropped chunk: emit a flush Enter so the partial we just wrote can't
- *     merge with the next message, then report non-submission (submit-failure).
- *   - Submit Enter is retried — a single dropped Enter would otherwise leave a
+ *   - On an unconfirmed chunk: attempt a flush Enter so a partial frame is less
+ *     likely to merge with the next message, then report an ambiguous dirty
+ *     generation; the flush cannot make delivery proof retroactive.
+ *   - Submit Enter is retried — a single unconfirmed Enter could otherwise leave a
  *     COMPLETE but unsubmitted line in the buffer.
  */
 export async function writeRunnerInput(
   pty: PtyHandle,
   markerPrefix: string,
   content: string,
   codexAppInput?: CodexAppTurnInput,
-): Promise<{ submitted: boolean }> {
+): Promise<{ submitted: boolean; submissionDisposition: RunnerSubmissionDisposition }> {
   const line = `${markerPrefix}${encodeRunnerInput(content, codexAppInput)}`;
 
   // Non-tmux fallback (raw PTY): a single write is fine — there's no send-keys
   // process to time out, and the PTY write isn't bounded the same way.
   if (!pty.sendText || !pty.sendSpecialKeys) {
     try {
-      pty.write(line + '\r');
+      if (pty.write(line + '\r') === false) {
+        // SessionBackend.write(false) is a rejection, but its contract does not
+        // prove whether a lower layer accepted a prefix before reporting it.
+        return { submitted: false, submissionDisposition: 'dirty_unknown' };
+      }
     } catch {
-      return { submitted: false };
+      // A throwing PTY write does not prove whether the kernel accepted a
+      // prefix (or the complete line) before surfacing the error.
+      return { submitted: false, submissionDisposition: 'dirty_unknown' };
     }
-    return { submitted: true };
+    return { submitted: true, submissionDisposition: 'submitted' };
   }
 
   const sendText = pty.sendText.bind(pty);
@@ -117,24 +123,51 @@ export async function writeRunnerInput(
   // the user can retry); we never touch the buffer with a half write. (Idempotent
   // on the happy path: the previous message's
   // submit Enter already emptied the buffer, so this enqueues an ignored blank.)
-  if (!sendEnterWithRetry()) return { submitted: false };
+  try {
+    if (!sendEnterWithRetry()) {
+      return { submitted: false, submissionDisposition: 'untouched' };
+    }
+  } catch {
+    // The backend threw while attempting the pre-flush.  No new frame bytes
+    // were intentionally written, but the Enter itself may have landed, so the
+    // runner generation is not proven clean enough for same-generation reuse.
+    return { submitted: false, submissionDisposition: 'dirty_unknown' };
+  }
 
   const chunks = chunkAscii(line, RUNNER_INPUT_CHUNK_BYTES);
   for (let i = 0; i < chunks.length; i++) {
-    if (sendText(chunks[i]) === false) {
+    let chunkWritten: void | boolean;
+    try {
+      chunkWritten = sendText(chunks[i]);
+    } catch {
+      return { submitted: false, submissionDisposition: 'dirty_unknown' };
+    }
+    if (chunkWritten === false) {
       // The chunks already written are a partial control line with no
       // terminating newline. Flush it (with retry) so it's less likely to
       // linger; even if every retry drops, the NEXT call's pre-flush gate above
       // refuses to write onto the dirty buffer, so no corruption-as-success can
       // slip through.
-      sendEnterWithRetry();
-      return { submitted: false };
+      try { sendEnterWithRetry(); } catch { /* disposition remains unknown */ }
+      // A tmux send-keys timeout reports false but cannot prove that the pane
+      // received zero bytes. In particular, a last-chunk timeout followed by a
+      // successful cleanup Enter may have submitted the complete valid frame.
+      // Never cancel attribution on this ambiguous boundary.
+      return { submitted: false, submissionDisposition: 'dirty_unknown' };
     }
     if (i < chunks.length - 1) await delay(RUNNER_INPUT_THROTTLE_MS);
   }
 
-  // Submit (with retry — a single dropped Enter would leave a complete but
+  // Submit (with retry — a single unconfirmed Enter could leave a complete but
   // unsubmitted line in the buffer).
-  if (!sendEnterWithRetry()) return { submitted: false };
-  return { submitted: true };
+  try {
+    if (!sendEnterWithRetry()) {
+      // The complete, valid line may still be buffered.  Retrying a successor
+      // in this generation could submit it against the wrong FIFO head.
+      return { submitted: false, submissionDisposition: 'dirty_unknown' };
+    }
+  } catch {
+    return { submitted: false, submissionDisposition: 'dirty_unknown' };
+  }
+  return { submitted: true, submissionDisposition: 'submitted' };
 }