fix(session): fence turn-failure writes under the run lease; detect paused turns without the .paused marker (Eve2-F2, Eve2-F3)

CHANGELOG.md

@@ -70,6 +70,8 @@ This section lists what is on `main` and not yet on npm.
 - Structured output: when `maxSteps` runs out while the model is still calling tools, a run with `output` set now makes one last model call with no tools offered ("answer now") and validates its reply - the result carries `object` (still `finishReason: 'max-steps'`) or `outputError` (`'output-invalid'`) instead of neither. The extra call is skipped when another budget limit (tokens, cost, duration) is already spent. See [Structured output](docs/structured-output.md).
 
 ### Fixed
+- A checkpointed session turn that stalled past its run lease and then failed for good no longer stamps its `lastError` (`retryable: false`) on the checkpoint of the process that took the turn over, which bricked that turn forever with `LOUSHO_SESSION_TURN_FAILED`: the session now holds the turn's run lease itself (handed to the run as `HELD_RUN_LEASE`) and `recordFailure()` writes through `RunLease.save`, whose fenced save writes nothing once the lease is lost. (Eve2-F2)
+- `agent.send(msg, { sessionId })` and a fresh `agent.session({ id })` no longer run beside a session turn that waits on an approval when its `<id>.paused` marker was never written (a crash or a send between the executor's awaiting-approval checkpoint and `markPaused()`): `pausedSessionTurn` now also scans the store's `<id>.turn-*` checkpoints for `status: 'awaiting-approval'` when the marker is absent, on stores that implement `list()`. (Eve2-F3)
 - Sessions: turn checkpoint history outside `turnHistoryLimit` leaked again when each turn used a new store object or process (serverless functions, CLI runs, a store per request, restarts): past about 42 turns every turn's history was kept, so storage grew with the square of the conversation. Every commit now deletes the turns that just left the window, whatever earlier commits of the process did; older backlog is still swept oldest first, up to 32 turns per commit. `fileStore(dir).prune()`'s result gains `checkpointHistory`, the number of checkpoint history files it deleted (they were deleted but not counted).
 - The published type declarations type-check with `skipLibCheck: false` under zod 4 as well as zod 3: `createTodoTools()`, `askQuestionTool()`, `createWebFetchTool()` / `webFetchTool`, `FsToolArgs` and `lousho add`'s registry types are spelled out instead of inferred zod-3 schema types (`ZodEffects`, three-argument `ZodObject`). Their tool `input` is typed as a `StandardSchemaV1` of the arguments. `npm run pack-smoke` now checks the packed declarations against zod 3 and zod 4.
 - `lousho dev` reloads an edited CommonJS tool file (`.cjs`, or `.js` without `"type": "module"`) and the CommonJS helpers it requires, and reports a broken one (`failed to reload ..., keeping the previous agent`) instead of logging "reloaded" and serving the cached copy. It also keeps reloading a file an editor replaces on save (write to a temp file, then rename), which on Linux stopped being watched after its first save.

src/session/AgentSession.ts

@@ -11,7 +11,8 @@ import { toMessages, type AgentInput } from '../providers/content';
 import type { ExecutionResult } from '../execution/AgentExecutor';
 import { startAgentRun, type AgentRun } from '../execution/agentRun';
 import { InputQueue } from '../execution/inputQueue';
-import { getCheckpointHistory, isCheckpointCorrupt, pausedApprovalIds, type Checkpoint, type CheckpointError, type CheckpointStore } from '../execution/checkpoint';
+import { getCheckpointHistory, isCheckpointCorrupt, pausedApprovalIds, type Checkpoint, type CheckpointError, type CheckpointListEntry, type CheckpointStore } from '../execution/checkpoint';
+import { acquireRunLease, HELD_RUN_LEASE, type RunLease, type RunLeaseOptions } from '../execution/runLease';
 import type { ApprovalKind } from '../execution/ApprovalGate';
 import { CompactedLLMProviderError, ConfigurationError, SDKError, SessionAwaitingApprovalError } from '../execution/errors';
 import { streamSessionTurn } from './sessionStream';
@@ -273,14 +274,46 @@ function pausedTurnKey(id: string): string {
  * Eve EVE-0: the session's turn that waits on an approval, found through its
  * {@link pausedTurnKey} pointer, or `null`. A pointer whose turn no longer
  * waits (decided, finished or discarded elsewhere) is deleted.
+ * Eve2-F3: without a live pointer, the turn checkpoints themselves are scanned
+ * (see {@link pausedTurnScan}) - `markPaused()` writes the pointer AFTER the
+ * executor saved the paused checkpoint, so a crash between the two writes (or a
+ * send that slipped into the gap) leaves the pause with no pointer, permanently.
  */
 export async function pausedSessionTurn(checkpointStore: CheckpointStore, id: string): Promise<Checkpoint | null> {
   const pointer = await checkpointStore.load(pausedTurnKey(id));
-  if (!pointer) return null;
-  const turn = await checkpointStore.load(pointer.sessionId);
-  if (turn?.status === 'awaiting-approval') return turn;
-  await checkpointStore.delete(pausedTurnKey(id));
-  return null;
+  if (pointer) {
+    const turn = await checkpointStore.load(pointer.sessionId);
+    if (turn?.status === 'awaiting-approval') return turn;
+    await checkpointStore.delete(pausedTurnKey(id));
+  }
+  return pausedTurnScan(checkpointStore, id);
+}
+
+/**
+ * Eve2-F3: the newest `<id>.turn-<n>` checkpoint that still waits on an approval,
+ * or `null` - the fallback when the {@link pausedTurnKey} pointer is missing
+ * (see {@link pausedSessionTurn}). A store that cannot `list()` keeps the
+ * pointer's behavior.
+ */
+async function pausedTurnScan(checkpointStore: CheckpointStore, id: string): Promise<Checkpoint | null> {
+  if (!checkpointStore.list) return null;
+  let listed: CheckpointListEntry[];
+  try {
+    listed = await checkpointStore.list({ status: 'awaiting-approval' });
+  } catch {
+    return null; // A store whose list() is unavailable (e.g. a KV binding without it) keeps the pointer's behavior.
+  }
+  const prefix = `${id}.turn-`;
+  let paused: Checkpoint | null = null;
+  let latest = -1;
+  for (const entry of listed) {
+    if (!entry.sessionId.startsWith(prefix)) continue;
+    const n = Number(entry.sessionId.slice(prefix.length));
+    if (!Number.isInteger(n) || n <= latest) continue;
+    latest = n;
+    paused = entry.checkpoint;
+  }
+  return paused;
 }
 
 /**
@@ -439,6 +472,8 @@ export class AgentSession<TObject = unknown> {
   }
   /** The turn running (or about to run) and the queue its run takes input from (LOU-V9). */
   private running: { inputs: InputQueue; result: Promise<ExecutionResult> } | undefined;
+  /** The run lease this session's turn holds on its checkpoint id, from before the run starts until its `record()` is done (Eve2-F2). */
+  private turnLease: RunLease | undefined;
   private turnStartedAt = 0;
   private transcript: Message[] = [];
   /** Whether `transcript` was read in the queued call running now (each call reads it again, Eve DUR-F3). */
@@ -576,9 +611,15 @@ export class AgentSession<TObject = unknown> {
             await this.beforeTurn(signal);
             const call = { input, metadata: options.metadata, principal: options.principal };
             const messages = toMessages(input);
-            const run = streamRun([...this.transcript, ...messages], signal, this.turnOptions(inputs), call);
-            started(run);
-            return this.record(await this.attempt(() => run.result), messages);
+            const turn = this.turnOptions(inputs);
+            await this.holdRunLease(turn);
+            try {
+              const run = streamRun([...this.transcript, ...messages], signal, turn, call);
+              started(run);
+              return this.record(await this.attempt(() => run.result), messages);
+            } finally {
+              await this.releaseRunLease();
+            }
           },
           inputs,
           (joined) => started(startAgentRun(() => joined))
@@ -899,7 +940,13 @@ export class AgentSession<TObject = unknown> {
   private async turn(call: SessionTurnCall, inputs: InputQueue, signal?: AbortSignal): Promise<ExecutionResult> {
     await this.beforeTurn(signal);
     const input = toMessages(call.input);
-    return this.record(await this.attempt(() => this.run([...this.transcript, ...input], signal, this.turnOptions(inputs), call)), input);
+    const options = this.turnOptions(inputs);
+    await this.holdRunLease(options);
+    try {
+      return this.record(await this.attempt(() => this.run([...this.transcript, ...input], signal, options, call)), input);
+    } finally {
+      await this.releaseRunLease();
+    }
   }
 
   /**
@@ -916,13 +963,44 @@ export class AgentSession<TObject = unknown> {
     }
   }
 
+  /**
+   * Eve2-F2: takes the turn's run lease before the run starts and hands it to the
+   * run (`HELD_RUN_LEASE`, so the executor saves under it instead of taking its
+   * own). The session keeps it until `record()` is done, so `recordFailure()`
+   * can write through `RunLease.save` - fenced: a run that lost its lease (it
+   * stalled past it and another process took the turn over) cannot stamp its
+   * failure on the checkpoint the new holder owns.
+   */
+  private async holdRunLease(options: SessionTurnOptions | undefined): Promise<void> {
+    const { sessionId, checkpointStore } = options ?? {};
+    if (!sessionId || !checkpointStore) return;
+    const lease = await acquireRunLease(checkpointStore, sessionId);
+    if (!lease) return;
+    this.turnLease = lease;
+    (options as SessionTurnOptions & RunLeaseOptions)[HELD_RUN_LEASE] = lease;
+  }
+
+  /** Releases the lease `holdRunLease` took; the turn's own `releaseLease` leaves another holder's alone. */
+  private async releaseRunLease(): Promise<void> {
+    const lease = this.turnLease;
+    this.turnLease = undefined;
+    await lease?.release().catch(() => undefined);
+  }
+
   private async recordFailure(error: unknown): Promise<void> {
     // A turn refused because another process runs it (its run lease) did not run: the checkpoint is that process's.
     if (error instanceof SDKError && error.code === 'LOUSHO_SESSION_BUSY') return;
     const turn = this.turnCheckpoint();
     const checkpoint = turn && (await turn.checkpointStore.load(turn.sessionId));
     if (!turn || !checkpoint || (checkpoint.status ?? 'in-progress') !== 'in-progress') return;
-    await turn.checkpointStore.save(turn.sessionId, { ...checkpoint, lastError: checkpointError(error), attempts: (checkpoint.attempts ?? 0) + 1 });
+    const failed = { ...checkpoint, lastError: checkpointError(error), attempts: (checkpoint.attempts ?? 0) + 1 };
+    // Eve2-F2: under the turn's own lease the write is fenced - it throws LOUSHO_SESSION_BUSY, writing
+    // nothing, when the run lost the lease and the checkpoint now belongs to the run that took it over.
+    if (this.turnLease?.id === turn.sessionId) {
+      await this.turnLease.save(failed);
+      return;
+    }
+    await turn.checkpointStore.save(turn.sessionId, failed);
   }
 
   /**
@@ -1038,7 +1116,13 @@ export class AgentSession<TObject = unknown> {
         'LOUSHO_SESSION_TURN_FAILED'
       );
     }
-    return this.record(await this.attempt(() => this.run([], signal, this.turnOptions())));
+    const options = this.turnOptions();
+    await this.holdRunLease(options);
+    try {
+      return this.record(await this.attempt(() => this.run([], signal, options)));
+    } finally {
+      await this.releaseRunLease();
+    }
   }
 
   /**

src/session/pausedSessionLock.test.ts

@@ -80,6 +80,28 @@ describe('a paused session turn locks the session (Eve EVE-0 follow-up)', () =>
     expect((await durable.send('something else')).text).toBe('fresh');
   });
 
+  it('a missing .paused marker still locks the session: the turn checkpoint itself says it waits (Eve2-F3)', async () => {
+    const store = memoryStore();
+    const model = mockModel([calling('restart'), 'resumed']);
+    const agent = createAgent({ provider: model, tools: [restart({ n: 0 })], store });
+    const paused = await agent.session({ id: 'chat', store: new MemorySessionStore() }).send('restart it');
+    expect(paused.finishReason).toBe('awaiting-approval');
+    // A crash between the executor's awaiting-approval checkpoint and markPaused() leaves no pointer, permanently.
+    await store.checkpoints.delete('chat.paused');
+    expect(await store.checkpoints.load('chat.paused')).toBeNull();
+
+    await expect(agent.send('unrelated work', { sessionId: 'chat' })).rejects.toMatchObject({
+      name: 'SessionAwaitingApprovalError',
+      code: 'LOUSHO_SESSION_AWAITING_APPROVAL',
+      approvalId: paused.approvalId,
+      sessionId: 'chat.turn-0',
+    });
+    const durable = agent.session({ id: 'chat' });
+    expect(await durable.pending()).toMatchObject({ status: 'awaiting-approval', approvalId: paused.approvalId });
+    await expect(durable.send('more work')).rejects.toMatchObject({ code: 'LOUSHO_SESSION_AWAITING_APPROVAL' });
+    expect(model.calls).toHaveLength(1);
+  });
+
   it('once the approval is decided the session takes new turns again', async () => {
     const store = memoryStore();
     const runs = { n: 0 };

src/session/sessionLease.test.ts

@@ -189,6 +189,47 @@ describe.each(stallable)('a run that stalled past its lease, over %s', (_name, b
     const transcript = await b.session({ id: 's1' }).load();
     expect(transcript.map((message) => `${message.role}:${typeof message.content === 'string' ? message.content : ''}`)).toEqual(['user:go', 'assistant:', 'tool:"by B"', 'assistant:done by B']);
   });
+
+  it("a stalled run that fails after losing its lease does not stamp its failure on the new holder's checkpoint (Eve2-F2)", async () => {
+    const { open, stall } = backend();
+    let releaseA!: () => void;
+    const gateA = new Promise<void>((resolve) => (releaseA = resolve));
+    let secondCall!: () => void;
+    const waiting = new Promise<void>((resolve) => (secondCall = resolve));
+    const overflow = () => Object.assign(new Error('400 context_length_exceeded: maximum context length is 128000 tokens'), { status: 400 });
+    const work = defineTool({ name: 'work', description: 'work', input: z.object({}), execute: async () => 'done' });
+    // A's turn runs its tool, then stalls inside the next model call - long enough to lose its lease.
+    const a = createAgent({
+      provider: mockModel([
+        { toolCalls: [{ name: 'work', id: 'call_1' }] },
+        async () => {
+          secondCall();
+          await gateA;
+          return { error: overflow() };
+        },
+      ]),
+      tools: [work],
+      store: open(),
+      retry: false,
+    });
+    const bStore = open();
+    const unavailable = Object.assign(new Error('503 service unavailable'), { status: 503 });
+    const b = createAgent({ provider: mockModel([{ error: unavailable }]), tools: [work], store: bStore, retry: false });
+
+    const first = a.session({ id: 's1' }).send('go');
+    await waiting; // A's turn is checkpointed in-progress and stalled in its second model call.
+    stall('s1.turn-0');
+    // B takes the turn over, and its attempt fails retryably: the checkpoint records it (attempts: 1, retryable).
+    await expect(b.session({ id: 's1' }).resume()).rejects.toThrow('503');
+    releaseA();
+    // A wakes and fails for good - but the checkpoint is B's now, so A's recordFailure must write nothing on it.
+    await expect(first).rejects.toThrow('context_length_exceeded');
+
+    expect(await bStore.checkpoints!.load('s1.turn-0')).toMatchObject({ attempts: 1, lastError: { retryable: true } });
+    // The turn stays resumable (no LOUSHO_SESSION_TURN_FAILED bricking).
+    const c = createAgent({ provider: mockModel(['finished']), tools: [work], store: open(), retry: false });
+    expect((await c.session({ id: 's1' }).resume())?.text).toBe('finished');
+  });
 });
 
 describe('run ids and session checkpoint keys', () => {