fix(agent-loop): discard an interrupted prefix composition instead of caching it

A cancel/dispose landing inside the first agent/session-prefix waterfall
used to commit the listener chain's return value to the instance cache
before the interruption check dropped the turn; an abort-aware listener's
degraded fallback would then ship on every later request of the instance.
The commit now happens only after the composition survives the
interruption check — the cache only ever holds a fully composed prefix,
and the next turn recomposes under a live signal.

packages/core/agent-loop/src/loop.ts

@@ -497,18 +497,27 @@ async function runTurn(
       // CURRENT request.
       if (transmission.sessionPrefix === undefined) {
         const emptyPrefix: Message[] = deepFreeze([])
-        transmission.sessionPrefix = deepFreeze(structuredClone(await ctx.waterfall(
+        const composed = await ctx.waterfall(
           'agent/session-prefix', agent, emptyPrefix, abort.signal,
           () => Promise.resolve(emptyPrefix),
-        )))
-      }
+        )
 
-      // Interruption landing during prefix composition: mirror the assembly
-      // window above — drop the about-to-start step without running the seam.
-      if (handle.isCancelled() || handle.isDisposed()) {
-        handle.setAbort(undefined)
-        reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
-        break
+        // Interruption landing during prefix composition: mirror the assembly
+        // window above — drop the about-to-start step without running the
+        // seam, and DISCARD the composition instead of caching it. An
+        // abort-aware listener may have returned a degraded fallback under
+        // the firing signal; committing it would ship a prefix no request
+        // ever used (and no header ever logged) on this instance's next real
+        // request. The next turn recomposes under a live signal — the cache
+        // only ever holds a fully composed prefix. The cache-hit path needs
+        // no such check: nothing awaits between the assembly check above and
+        // the pre-step seam.
+        if (handle.isCancelled() || handle.isDisposed()) {
+          handle.setAbort(undefined)
+          reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
+          break
+        }
+        transmission.sessionPrefix = deepFreeze(structuredClone(composed))
       }
 
       // Pre-step surface-mutation checkpoint (compaction), fired OUTSIDE the

packages/core/agent-loop/tests/cancel.spec.ts

@@ -12,7 +12,7 @@
 
 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
-import LlmService from '@deepseek-ai/dsh-llm'
+import LlmService, { type Message } from '@deepseek-ai/dsh-llm'
 import SessionStore, { SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry from '@deepseek-ai/dsh-tools'
@@ -229,6 +229,40 @@ describe('Agent.cancel()', () => {
     expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
   })
 
+  it('a cancel-interrupted prefix composition is discarded: the next send recomposes and ships the fresh prefix (stale-cache guard)', async () => {
+    const adapter = new MockAdapter([textResponse('reply')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+
+    // The first composition is interrupted mid-waterfall and — like an
+    // abort-aware listener bailing on a firing signal — contributes nothing.
+    // Caching that degraded result would silently strip the prefix from every
+    // later request of this instance; the loop must discard it and recompose
+    // on the next send, and the SECOND composition's value must be what the
+    // wire and the header log carry.
+    const opener: Message = { role: 'user', content: [{ type: 'text', text: 'fresh opener' }] }
+    let compositions = 0
+    ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next): Promise<Message[]> => {
+      compositions += 1
+      if (compositions === 1) {
+        agent.cancel('mid-composition')
+        return next()
+      }
+      return [opener, ...await next()]
+    })
+
+    send(agent, 'dropped')
+    await waitForIdle(ctx, agent)
+    send(agent, 'real prompt')
+    await waitForIdle(ctx, agent)
+
+    expect(compositions).toBe(2)
+    expect(adapter.requests).toHaveLength(1)
+    expect(adapter.requests[0]?.messages[0]).toEqual(opener)
+    const headerEvent = agent.session.events.find(e => e.type === 'request/header')
+    expect(headerEvent?.type === 'request/header' && headerEvent.data.header.messagePrefix).toEqual([opener])
+  })
+
   it('cancel from a synchronous turn/start session-event listener drops the step (step-start window)', async () => {
     const adapter = new MockAdapter([textResponse('should not stream')])
     const ctx = await harness(adapter)