fix(session): synthesize tool results for interrupted tool calls on crash recovery (review #33)
The loop logs the assistant/message (carrying tool-call blocks) BEFORE running the tools, so a crash mid-tool leaves durable tool calls with no matching tool/result. interruptedTurnClosers only added step/end + turn/end, so a resumed session's deriveMessages() replayed a dangling assistant tool-call — which every provider rejects as an invalid transcript on the next request. interruptedTurnClosers now scans the interrupted turn for tool-call blocks without a matching tool/result and synthesizes an error tool/result for each (before the step/end), so the rehydrated history is a valid transcript. Adds a dedicated repair.spec.ts and a shared-contract case proving both backends pair every orphaned call with a result. Docs (ADR 0018, both persistence READMEs, load() JSDoc) updated. Also fixes the echo-agent README session-cleanup path: demo:echo runs from the repo root, so sessions land in <repo-root>/.sessions/_no-cwd/, not examples/echo-agent/.sessions/ (review #33).
This commit is contained in:
@@ -7,11 +7,26 @@
|
||||
* in a long-horizon task (many steps, large tool output), so those events MUST
|
||||
* be preserved — truncating the turn would silently destroy real work. Instead,
|
||||
* on reload the backend CLOSES the orphaned turn by appending the minimal
|
||||
* synthetic boundary events (a `step/end` if a step was still open, then a
|
||||
* `turn/end` carrying the merge-extensible `{ kind: 'interrupted' }` reason).
|
||||
* synthetic boundary events:
|
||||
*
|
||||
* 1. an error `tool/result` for every `tool-call` in the interrupted turn that
|
||||
* never got its matching `tool/result` (so the rehydrated history is a
|
||||
* VALID provider transcript — see below),
|
||||
* 2. a `step/end` if a step was still open, then
|
||||
* 3. a `turn/end` carrying the merge-extensible `{ kind: 'interrupted' }` reason.
|
||||
*
|
||||
* The marker records that the turn was cut short by a crash, not completed by
|
||||
* the model. See ADR 0018.
|
||||
*
|
||||
* Why the synthetic tool results matter: `deriveMessages()` renders the
|
||||
* `tool-call` blocks inside a durable `assistant/message` but only emits a
|
||||
* matching tool-result when a `tool/result` EVENT exists. A crash between the
|
||||
* assistant message and its tool results (the loop runs the tools AFTER logging
|
||||
* the assistant message, so a process killed mid-tool leaves the calls without
|
||||
* results) would otherwise reload a history with a dangling assistant tool-call
|
||||
* — which every provider rejects as an invalid transcript on the next request.
|
||||
* Synthesizing an error result per orphaned call keeps resume safe.
|
||||
*
|
||||
* This module computes those synthetic closers from an event list; the backend
|
||||
* returns them inline from `load` (so the reconstructed session is balanced and
|
||||
* immediately usable) and persists them on the first post-load `append`.
|
||||
@@ -19,6 +34,7 @@
|
||||
* @module @deepseek-ai/dsh-session/repair
|
||||
*/
|
||||
|
||||
import type { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import type { SessionEvent } from './types.ts'
|
||||
|
||||
/**
|
||||
@@ -29,6 +45,11 @@ import type { SessionEvent } from './types.ts'
|
||||
* "future" time). Returns an empty array when the log is already balanced
|
||||
* (ends on a `turn/end`, or is empty) — the common, non-crash case.
|
||||
*
|
||||
* The closers, in order: an error `tool/result` for each unmatched `tool-call`
|
||||
* in the interrupted turn, then a `step/end` if a step is open, then the
|
||||
* `turn/end {interrupted}`. The tool-results come first so a step that issued
|
||||
* tool calls is balanced (every call has a result) before its `step/end`.
|
||||
*
|
||||
* Only the LAST turn can be open: the invariants plugin guarantees a `turn/end`
|
||||
* before any later `turn/start`, so an interior open turn is impossible in a
|
||||
* valid committed log. Likewise at most one step is open within that turn.
|
||||
@@ -36,14 +57,22 @@ import type { SessionEvent } from './types.ts'
|
||||
export function interruptedTurnClosers(events: readonly SessionEvent[]): SessionEvent[] {
|
||||
let openTurn: number | null = null
|
||||
let openStep: number | null = null
|
||||
// Track tool calls vs. their results WITHIN the currently-open turn only: a
|
||||
// call is "pending" until its matching tool/result arrives. Reset at every
|
||||
// turn boundary so a committed earlier turn (already balanced) never leaks a
|
||||
// phantom pending call into the interrupted-turn repair.
|
||||
const pendingCalls = new Map<CallId, { step: number }>()
|
||||
for (const event of events) {
|
||||
switch (event.type) {
|
||||
case 'turn/start':
|
||||
openTurn = event.data.turn
|
||||
openStep = null
|
||||
pendingCalls.clear()
|
||||
break
|
||||
case 'turn/end':
|
||||
openTurn = null
|
||||
openStep = null
|
||||
pendingCalls.clear()
|
||||
break
|
||||
case 'step/start':
|
||||
openStep = event.data.step
|
||||
@@ -51,6 +80,16 @@ export function interruptedTurnClosers(events: readonly SessionEvent[]): Session
|
||||
case 'step/end':
|
||||
openStep = null
|
||||
break
|
||||
case 'assistant/message':
|
||||
// The assistant message carries the tool-call blocks; each is pending
|
||||
// until a tool/result event with the same callId is logged.
|
||||
for (const block of event.data.content) {
|
||||
if (block.type === 'tool-call') pendingCalls.set(block.id, { step: event.data.step })
|
||||
}
|
||||
break
|
||||
case 'tool/result':
|
||||
pendingCalls.delete(event.data.callId)
|
||||
break
|
||||
// Other event types do not move the turn/step boundary cursor.
|
||||
default:
|
||||
break
|
||||
@@ -69,7 +108,27 @@ export function interruptedTurnClosers(events: readonly SessionEvent[]): Session
|
||||
const time = last.time
|
||||
const closers: SessionEvent[] = []
|
||||
|
||||
// Close an open step first — a turn/end while a step is open is an invariant
|
||||
// Synthesize an error tool/result for each tool-call left unanswered by the
|
||||
// crash, so deriveMessages() yields a valid provider transcript on resume (a
|
||||
// dangling assistant tool-call is rejected by every provider). Insertion
|
||||
// order follows the Map (insertion = log order of the assistant messages).
|
||||
for (const [callId, { step }] of pendingCalls) {
|
||||
closers.push({
|
||||
type: 'tool/result',
|
||||
seq: seq++,
|
||||
time,
|
||||
data: {
|
||||
turn: openTurn,
|
||||
step,
|
||||
callId,
|
||||
content: [{ type: 'text', text: 'Tool call interrupted by a crash; no result was recorded.' }],
|
||||
isError: true,
|
||||
error: { name: 'InterruptedError', code: 'interrupted' },
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// Close an open step next — a turn/end while a step is open is an invariant
|
||||
// violation, so the step's boundary must be synthesized before the turn's.
|
||||
if (openStep !== null) {
|
||||
closers.push({ type: 'step/end', seq: seq++, time, data: { turn: openTurn, step: openStep } })
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import { interruptedTurnClosers } from '../src/index.ts'
|
||||
import type { SessionEvent } from '../src/index.ts'
|
||||
|
||||
/**
|
||||
* Unit coverage for the crash-recovery closer synthesis. The persistence
|
||||
* contract exercises it end-to-end through both backends; these tests pin the
|
||||
* pure function's branches directly — especially the synthetic error
|
||||
* `tool/result` for a tool call the crash left unanswered (without it a
|
||||
* resumed session replays a dangling assistant tool-call and the provider
|
||||
* rejects the transcript).
|
||||
*/
|
||||
|
||||
const userTurnStart = (turn: number, seq: number): SessionEvent =>
|
||||
({ type: 'turn/start', seq, time: seq, data: { turn, trigger: { kind: 'message', source: { kind: 'user' } } } })
|
||||
|
||||
describe('interruptedTurnClosers', () => {
|
||||
it('returns nothing for a balanced log (ends on turn/end)', () => {
|
||||
const balanced: SessionEvent[] = [
|
||||
userTurnStart(1, 0),
|
||||
{ type: 'turn/end', seq: 1, time: 1, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
expect(interruptedTurnClosers(balanced)).toEqual([])
|
||||
})
|
||||
|
||||
it('returns nothing for an empty log', () => {
|
||||
expect(interruptedTurnClosers([])).toEqual([])
|
||||
})
|
||||
|
||||
it('closes an open turn with no open step (turn/end {interrupted} only)', () => {
|
||||
const events: SessionEvent[] = [userTurnStart(1, 0)]
|
||||
const closers = interruptedTurnClosers(events)
|
||||
expect(closers.map(e => e.type)).toEqual(['turn/end'])
|
||||
const end = closers[0]!
|
||||
expect(end.seq).toBe(1)
|
||||
expect(end.type === 'turn/end' && end.data.reason).toEqual({ kind: 'interrupted' })
|
||||
})
|
||||
|
||||
it('closes an open step before the turn (step/end then turn/end)', () => {
|
||||
const events: SessionEvent[] = [
|
||||
userTurnStart(1, 0),
|
||||
{ type: 'step/start', seq: 1, time: 1, data: { turn: 1, step: 1 } },
|
||||
]
|
||||
const closers = interruptedTurnClosers(events)
|
||||
expect(closers.map(e => e.type)).toEqual(['step/end', 'turn/end'])
|
||||
expect(closers.map(e => e.seq)).toEqual([2, 3])
|
||||
})
|
||||
|
||||
it('synthesizes an error tool/result for a tool-call the crash left unanswered', () => {
|
||||
// A step issued one tool call (in the assistant message) but crashed before
|
||||
// the tool/result was logged — the classic mid-tool crash.
|
||||
const events: SessionEvent[] = [
|
||||
userTurnStart(2, 0),
|
||||
{ type: 'step/start', seq: 1, time: 1, data: { turn: 2, step: 1 } },
|
||||
{ type: 'assistant/message', seq: 2, time: 2, data: { turn: 2, step: 1, content: [
|
||||
{ type: 'text', text: 'calling a tool' },
|
||||
{ type: 'tool-call', id: CallId('call-1'), name: 'bash', arguments: '{}' },
|
||||
] } },
|
||||
]
|
||||
const closers = interruptedTurnClosers(events)
|
||||
// tool/result (for the orphaned call) → step/end → turn/end, contiguous seqs.
|
||||
expect(closers.map(e => e.type)).toEqual(['tool/result', 'step/end', 'turn/end'])
|
||||
expect(closers.map(e => e.seq)).toEqual([3, 4, 5])
|
||||
const result = closers[0]!
|
||||
expect(result.type === 'tool/result' && result.data).toMatchObject({
|
||||
turn: 2, step: 1, callId: CallId('call-1'), isError: true, error: { code: 'interrupted' },
|
||||
})
|
||||
})
|
||||
|
||||
it('does NOT synthesize a result for a tool-call that already has one', () => {
|
||||
const events: SessionEvent[] = [
|
||||
userTurnStart(2, 0),
|
||||
{ type: 'step/start', seq: 1, time: 1, data: { turn: 2, step: 1 } },
|
||||
{ type: 'assistant/message', seq: 2, time: 2, data: { turn: 2, step: 1, content: [
|
||||
{ type: 'tool-call', id: CallId('call-1'), name: 'bash', arguments: '{}' },
|
||||
] } },
|
||||
{ type: 'tool/result', seq: 3, time: 3, data: { turn: 2, step: 1, callId: CallId('call-1'), content: [{ type: 'text', text: 'ok' }], isError: false } },
|
||||
]
|
||||
// The call is answered, so only the open step + turn need closing.
|
||||
const closers = interruptedTurnClosers(events)
|
||||
expect(closers.map(e => e.type)).toEqual(['step/end', 'turn/end'])
|
||||
})
|
||||
|
||||
it('synthesizes results only for the still-open turn, not a committed earlier turn', () => {
|
||||
// Turn 1 completed with its own tool call+result (balanced). Turn 2 crashed
|
||||
// with an unanswered call. Only turn 2's call must get a synthetic result.
|
||||
const events: SessionEvent[] = [
|
||||
userTurnStart(1, 0),
|
||||
{ type: 'step/start', seq: 1, time: 1, data: { turn: 1, step: 1 } },
|
||||
{ type: 'assistant/message', seq: 2, time: 2, data: { turn: 1, step: 1, content: [
|
||||
{ type: 'tool-call', id: CallId('old-call'), name: 'bash', arguments: '{}' },
|
||||
] } },
|
||||
{ type: 'tool/result', seq: 3, time: 3, data: { turn: 1, step: 1, callId: CallId('old-call'), content: [], isError: false } },
|
||||
{ type: 'step/end', seq: 4, time: 4, data: { turn: 1, step: 1 } },
|
||||
{ type: 'turn/end', seq: 5, time: 5, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
userTurnStart(2, 6),
|
||||
{ type: 'step/start', seq: 7, time: 7, data: { turn: 2, step: 1 } },
|
||||
{ type: 'assistant/message', seq: 8, time: 8, data: { turn: 2, step: 1, content: [
|
||||
{ type: 'tool-call', id: CallId('new-call'), name: 'bash', arguments: '{}' },
|
||||
] } },
|
||||
]
|
||||
const closers = interruptedTurnClosers(events)
|
||||
expect(closers.map(e => e.type)).toEqual(['tool/result', 'step/end', 'turn/end'])
|
||||
const result = closers[0]!
|
||||
expect(result.type === 'tool/result' && result.data.callId).toBe('new-call')
|
||||
})
|
||||
|
||||
it('synthesizes a result for each of multiple unanswered calls, in log order', () => {
|
||||
const events: SessionEvent[] = [
|
||||
userTurnStart(1, 0),
|
||||
{ type: 'step/start', seq: 1, time: 1, data: { turn: 1, step: 1 } },
|
||||
{ type: 'assistant/message', seq: 2, time: 2, data: { turn: 1, step: 1, content: [
|
||||
{ type: 'tool-call', id: CallId('call-a'), name: 'bash', arguments: '{}' },
|
||||
{ type: 'tool-call', id: CallId('call-b'), name: 'bash', arguments: '{}' },
|
||||
] } },
|
||||
// call-a got answered before the crash; call-b did not.
|
||||
{ type: 'tool/result', seq: 3, time: 3, data: { turn: 1, step: 1, callId: CallId('call-a'), content: [], isError: false } },
|
||||
]
|
||||
const closers = interruptedTurnClosers(events)
|
||||
expect(closers.map(e => e.type)).toEqual(['tool/result', 'step/end', 'turn/end'])
|
||||
const result = closers[0]!
|
||||
expect(result.type === 'tool/result' && result.data.callId).toBe('call-b')
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user