fix(eve): terminate turns on cancellation and session failure

.changeset/eve-terminate-cancelled-turns.md

@@ -0,0 +1,9 @@
+---
+"evlog": patch
+---
+
+Emit a wide event for eve turns that end without `turn.completed` or `turn.failed`.
+
+A turn cancelled by eve produced no wide event, and its logger, accumulator and session slot stayed in memory for good: LRU eviction skips any session that still has an active turn, so nothing ever reclaimed them. `turn.cancelled` now closes the turn on its own terminal path — status `499`, `eve.phase: 'cancelled'`, level `info`, because cancellation is not a failure in eve's model.
+
+`session.failed` and `session.completed` flush any turn still open for that session and drop its carried-over context, which also ends the indefinite retention of snapshots for finished sessions.

packages/evlog/src/eve/index.ts

@@ -14,6 +14,9 @@ import {
 
 const DEFAULT_MAX_SESSIONS = 256
 
+/** Client-closed-request status used for turns eve cancelled before a terminal outcome. */
+const CANCELLED_STATUS = 499
+
 /** Options for {@link defineEvlogHook}. */
 export interface EvlogEveOptions extends BaseEvlogOptions {
   /** Passed to {@link initLogger} on the first hook invocation. */
@@ -305,6 +308,7 @@ export function useLogger(ctx?: EveTurnSessionContext): AuditableLogger {
 interface EveGlobalState {
   turnStates: Map<string, TurnState>
   activeTurnBySession: Map<string, string>
+  sessionTurnIds: Map<string, Set<string>>
   sessionSnapshots: Map<string, Record<string, unknown>>
   sessionPendingActions: Map<string, Map<string, PendingAction>>
   sessionApprovals: Map<string, EveApprovalPending[]>
@@ -323,6 +327,7 @@ function getEveGlobalState(): EveGlobalState {
     host[EVE_GLOBAL_STATE] = {
       turnStates: new Map(),
       activeTurnBySession: new Map(),
+      sessionTurnIds: new Map(),
       sessionSnapshots: new Map(),
       sessionPendingActions: new Map(),
       sessionApprovals: new Map(),
@@ -342,6 +347,15 @@ function activeTurnBySession(): Map<string, string> {
   return getEveGlobalState().activeTurnBySession
 }
 
+function sessionTurnIds(): Map<string, Set<string>> {
+  return getEveGlobalState().sessionTurnIds
+}
+
+/** Turn ids still open for a session, as a snapshot safe to iterate while finishing. */
+function openTurnIds(sessionId: string): string[] {
+  return [...(sessionTurnIds().get(sessionId) ?? [])]
+}
+
 function sessionSnapshots(): Map<string, Record<string, unknown>> {
   return getEveGlobalState().sessionSnapshots
 }
@@ -371,6 +385,7 @@ function clearSessionState(sessionId: string): void {
   sessionRollups().delete(sessionId)
   sessionPendingActions().delete(sessionId)
   sessionApprovals().delete(sessionId)
+  sessionTurnIds().delete(sessionId)
 }
 
 function evictStaleSessions(): void {
@@ -409,10 +424,12 @@ function derivePhase(
   accumulator: TurnAccumulator,
   httpStatus: number,
 ): string | undefined {
+  const eve = ctx.eve as { cancelled?: boolean } | undefined
+  if (eve?.cancelled) return 'cancelled'
   const approval = ctx.approval as { status?: string } | undefined
   if (approval?.status === 'rejected') return 'rejected'
   if (approval?.status === 'pending' || accumulator.pausedForInput) return 'awaiting-approval'
-  if (httpStatus >= 400) return 'failed'
+  if (httpStatus >= 400 && httpStatus !== CANCELLED_STATUS) return 'failed'
   return undefined
 }
 
@@ -564,6 +581,9 @@ function getOrCreateTurnState(
 
   turnStates().set(key, state)
   activeTurnBySession().set(sessionId, turnId)
+  const open = sessionTurnIds().get(sessionId) ?? new Set<string>()
+  open.add(turnId)
+  sessionTurnIds().set(sessionId, open)
   return state
 }
 
@@ -601,10 +621,38 @@ async function finishTurn(
     if (activeTurnBySession().get(sessionId) === turnId) {
       activeTurnBySession().delete(sessionId)
     }
+    const open = sessionTurnIds().get(sessionId)
+    open?.delete(turnId)
+    if (open?.size === 0) sessionTurnIds().delete(sessionId)
     pruneEmptySessionMaps(sessionId)
   }
 }
 
+/**
+ * Emit every turn still open for a session. eve ends a session with
+ * `session.completed` / `session.failed`; a turn left open at that point never
+ * received its own terminal event, so without this it would neither be emitted
+ * nor released.
+ *
+ * Each turn is finished independently: a user `keep` callback that throws
+ * rejects that turn's `finish`, and must not take the remaining turns with it.
+ */
+async function finishOpenTurns(
+  sessionId: string,
+  opts: { status?: number; error?: Error },
+  decorate?: (state: TurnState) => void,
+): Promise<void> {
+  for (const turnId of openTurnIds(sessionId)) {
+    try {
+      const state = getTurnState(sessionId, turnId)
+      if (state && decorate) decorate(state)
+      await finishTurn(sessionId, turnId, opts)
+    } catch (err) {
+      console.error('[evlog] eve hook handler failed:', err)
+    }
+  }
+}
+
 function runSafe(fn: () => void | Promise<void>): void {
   void (async () => {
     try {
@@ -821,6 +869,48 @@ export function defineEvlogHook(options: EvlogEveOptions = {}): HookDefinition {
         }
       },
 
+      async 'turn.cancelled'(event, ctx) {
+        try {
+          const state = getTurnState(ctx.session.id, event.data.turnId)
+          state?.logger.set({ eve: { cancelled: true } })
+          await finishTurn(ctx.session.id, event.data.turnId, { status: CANCELLED_STATUS })
+        } catch (err) {
+          console.error('[evlog] eve hook handler failed:', err)
+        }
+      },
+
+      async 'session.completed'(_event, ctx) {
+        try {
+          await finishOpenTurns(ctx.session.id, { status: 200 })
+        } catch (err) {
+          console.error('[evlog] eve hook handler failed:', err)
+        } finally {
+          clearSessionState(ctx.session.id)
+        }
+      },
+
+      async 'session.failed'(event, ctx) {
+        try {
+          const error = new Error(event.data.message)
+          error.name = event.data.code
+          await finishOpenTurns(ctx.session.id, { error, status: 500 }, (state) => {
+            state.logger.set({
+              eve: {
+                failure: {
+                  code: event.data.code,
+                  message: event.data.message,
+                  ...(event.data.details ? { details: event.data.details } : {}),
+                },
+              },
+            })
+          })
+        } catch (err) {
+          console.error('[evlog] eve hook handler failed:', err)
+        } finally {
+          clearSessionState(ctx.session.id)
+        }
+      },
+
       async 'turn.failed'(event, ctx) {
         try {
           const state = getTurnState(ctx.session.id, event.data.turnId)
@@ -855,6 +945,7 @@ export function resetEvlogEveForTests(): void {
   clearAsyncLocalStorage(turnLoggerStorage)
   turnStates().clear()
   activeTurnBySession().clear()
+  sessionTurnIds().clear()
   sessionSnapshots().clear()
   sessionPendingActions().clear()
   sessionApprovals().clear()

packages/evlog/test/eve.test.ts

@@ -43,6 +43,7 @@ async function runTurn(
   options: {
     turnId?: string
     fail?: boolean
+    cancel?: boolean
     steps?: number
     toolResults?: Array<{
       toolName: string
@@ -181,7 +182,12 @@ async function runTurn(
     }, ctx)
   }
 
-  if (options.fail) {
+  if (options.cancel) {
+    await events['turn.cancelled']!({
+      type: 'turn.cancelled',
+      data: { sequence: 99, turnId },
+    }, ctx)
+  } else if (options.fail) {
     await events['turn.failed']!({
       type: 'turn.failed',
       data: {
@@ -301,6 +307,129 @@ describe('evlog/eve', () => {
     })
   })
 
+  it('emits a cancelled turn as a non-error wide event', async () => {
+    const spies = createPipelineSpies()
+    const hook = defineEvlogHook({ drain: spies.drain })
+
+    await runTurn(hook, { cancel: true })
+
+    await waitForDrainCalls(spies.drain)
+    const event = findEventViaDrain(spies.drain, () => true)
+    expect(event?.status).toBe(499)
+    expect(event?.level).toBe('info')
+    expect(event?.eve).toMatchObject({ phase: 'cancelled', cancelled: true })
+  })
+
+  it('releases turn state after a cancelled turn', async () => {
+    const spies = createPipelineSpies()
+    const hook = defineEvlogHook({ drain: spies.drain })
+
+    await runTurn(hook, { cancel: true })
+
+    expect(() => useLogger(toolContext())).toThrow(/could not find a logger/)
+
+    await runTurn(hook, { turnId: TURN_ID_1 })
+    await waitForDrainCalls(spies.drain, 2)
+    expect(findEventViaDrain(spies.drain, e => e.path?.includes(TURN_ID_1))).toBeDefined()
+  })
+
+  it('flushes an in-flight turn when the session fails without turn.failed', async () => {
+    const spies = createPipelineSpies()
+    const hook = defineEvlogHook({ drain: spies.drain })
+    const ctx = hookContext()
+
+    hook.events!['turn.started']!({
+      type: 'turn.started',
+      data: { sequence: 0, turnId: TURN_ID },
+    }, ctx)
+
+    await hook.events!['session.failed']!({
+      type: 'session.failed',
+      data: { code: 'SESSION_ERROR', message: 'session exploded', sessionId: SESSION_ID },
+    }, ctx)
+
+    await waitForDrainCalls(spies.drain)
+    const event = findEventViaDrain(spies.drain, () => true)
+    expect(event?.status).toBe(500)
+    expect(event?.level).toBe('error')
+    expect(event?.eve).toMatchObject({
+      failure: { code: 'SESSION_ERROR', message: 'session exploded' },
+    })
+    expect(() => useLogger(toolContext())).toThrow(/could not find a logger/)
+  })
+
+  it('drops session context once the session completes', async () => {
+    const spies = createPipelineSpies()
+    const hook = defineEvlogHook({ drain: spies.drain })
+    const ctx = hookContext()
+
+    hook.events!['turn.started']!({
+      type: 'turn.started',
+      data: { sequence: 0, turnId: TURN_ID },
+    }, ctx)
+    useLogger(toolContext()).set({ customer: { slug: 'acme' } })
+
+    await hook.events!['session.completed']!({ type: 'session.completed' } as never, ctx)
+
+    await waitForDrainCalls(spies.drain)
+    const openTurn = findEventViaDrain(spies.drain, e => e.path?.includes(TURN_ID))
+    expect(openTurn?.status).toBe(200)
+    expect(() => useLogger(toolContext())).toThrow(/could not find a logger/)
+
+    hook.events!['turn.started']!({
+      type: 'turn.started',
+      data: { sequence: 0, turnId: TURN_ID_1 },
+    }, ctx)
+    await hook.events!['turn.completed']!({
+      type: 'turn.completed',
+      data: { sequence: 1, turnId: TURN_ID_1 },
+    }, ctx)
+
+    await waitForDrainCalls(spies.drain, 2)
+    const secondTurn = findEventViaDrain(spies.drain, e => e.path?.includes(TURN_ID_1))
+    expect(secondTurn?.customer).toBeUndefined()
+  })
+
+  it('finishes the remaining turns and clears session state when one turn fails to finish', async () => {
+    const spies = createPipelineSpies()
+    const hook = defineEvlogHook({
+      drain: spies.drain,
+      keep: (tail) => {
+        if (tail.path?.endsWith(TURN_ID)) throw new Error('keep exploded')
+      },
+    })
+    const ctx = hookContext()
+
+    hook.events!['turn.started']!({
+      type: 'turn.started',
+      data: { sequence: 0, turnId: TURN_ID },
+    }, ctx)
+    useLogger(toolContext()).set({ customer: { slug: 'acme' } })
+    hook.events!['turn.started']!({
+      type: 'turn.started',
+      data: { sequence: 1, turnId: TURN_ID_1 },
+    }, ctx)
+
+    await hook.events!['session.completed']!({ type: 'session.completed' } as never, ctx)
+
+    await waitForDrainCalls(spies.drain)
+    expect(findEventViaDrain(spies.drain, e => e.path?.endsWith(TURN_ID))).toBeUndefined()
+    expect(findEventViaDrain(spies.drain, e => e.path?.endsWith(TURN_ID_1))).toBeDefined()
+
+    hook.events!['turn.started']!({
+      type: 'turn.started',
+      data: { sequence: 2, turnId: 'turn_2' },
+    }, ctx)
+    await hook.events!['turn.completed']!({
+      type: 'turn.completed',
+      data: { sequence: 3, turnId: 'turn_2' },
+    }, ctx)
+
+    await waitForDrainCalls(spies.drain, 2)
+    const thirdTurn = findEventViaDrain(spies.drain, e => e.path?.endsWith('turn_2'))
+    expect(thirdTurn?.customer).toBeUndefined()
+  })
+
   it('does not throw when an internal handler fails', async () => {
     const hook = defineEvlogHook({
       enrich: () => {