fix(plan-mode): flush pending mode only at the in-turn agent/step seam

Prompt admission runs before any turn opens, so the agent/prompt-submit
flush listener could only ever append plan/mode outside an open turn —
invariant-vetoed and retried at agent/step when the session invariant is
mounted, silently out-of-turn in the durable log when it is not. agent/step
fires inside the open turn before every request derivation (including
turn 1 step 1), so it is the sole flush point and no behavior is lost: a
pending idle switch lands as the first step's in-turn plan/mode. Tests pin
the new contract: admission never appends; the first step boundary flushes.

packages/plan/plan-mode/src/index.ts

@@ -157,24 +157,13 @@ export class PlanModeService extends Service {
     this.section = resolveConfig(config).section
     let disposed = false
 
-    // Boundary flushes use loop interception seams, not post-commit
-    // `session/event` observation. Flush after next(): a selection arriving
-    // while a downstream async listener awaits must still shape the request
-    // this boundary precedes. Failures are contained so policy cannot block a
-    // prompt or turn; a failed append remains pending for a later boundary.
-    const flushAfter = async <T>(agent: Agent, next: () => Promise<T>): Promise<T> => {
-      const decision = await next()
-      if (!disposed) {
-        try {
-          this.onBoundary(agent)
-        } catch (error) {
-          ctx.logger.warn('dsh-plan-mode: boundary flush failed: %o', error)
-        }
-      }
-      return decision
-    }
-    ctx.on('agent/prompt-submit', (agent, _content, _source, _signal, next) =>
-      flushAfter(agent, next), { prepend: true })
+    // The boundary flush uses the loop's `agent/step` interception seam, not
+    // post-commit `session/event` observation. `agent/step` runs inside the
+    // open turn before every request derivation (including turn 1 step 1), so
+    // it is the sole flush point: prompt admission happens pre-turn, where a
+    // `plan/mode` append would land outside any open turn. Failures are
+    // contained so policy cannot block a turn; a failed append remains
+    // pending for a later boundary.
     ctx.on('agent/step', (agent) => {
       if (disposed) return
       try {

packages/plan/plan-mode/tests/plan-mode.spec.ts

@@ -177,53 +177,53 @@ describe('ctx.planMode: get/set', () => {
 })
 
 describe('the boundary flush', () => {
-  it('flushes the pending intent as a plan/mode at turn/start', async () => {
+  it('does not flush at prompt admission — the seam is pre-turn, so the first step boundary lands it', async () => {
     const ctx = await setup()
     const agent = await agentWithSession(ctx)
     ctx.planMode.set(agent, true)
+    // Prompt admission runs before any turn opens; a plan/mode appended there
+    // would sit outside the turn. The pending intent survives admission and
+    // the in-turn agent/step boundary flushes it before the request derives.
     await boundary(ctx, agent, 'turn/start')
+    expect(agent.session.events.some(event => event.type === 'plan/mode')).toBe(false)
+    expect(ctx.planMode.get(agent)).toEqual({ active: false, pending: true })
+    await boundary(ctx, agent, 'step/end')
     expect(foldPlanMode(agent.session.events)).toBe(true)
     expect(ctx.planMode.get(agent)).toEqual({ active: true })
   })
 
-  it('flushes a set() that arrives while a downstream listener is still awaiting (post-next ordering)', async () => {
-    const ctx = await setup()
+  it('skips the flush after the plugin fiber is disposed (a captured wrapper must not write into a dead service)', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SystemPrompt)
+    await ctx.plugin(ToolRegistry)
+    const fiber = await ctx.plugin(PlanModeService, PLAN_CONFIG)
     const agent = await agentWithSession(ctx)
-    // A downstream async listener (the shipped hooks listeners' shape): the
-    // selection lands DURING its await — after this boundary began, before it
-    // returns. The prepended flush runs after next(), so the plan/mode still
-    // precedes the request this boundary gates.
-    ctx.on('agent/prompt-submit', async (_agent, _content, _source, _signal, next) => {
-      await new Promise(resolve => setTimeout(resolve, 5))
-      ctx.planMode.set(agent, true)
-      return next()
-    })
-    await agentEvents(ctx, agent).waterfall(
-      'agent/prompt-submit', [{ type: 'text', text: 'probe' }], { kind: 'user' },
-      new AbortController().signal, () => Promise.resolve({ kind: 'allow' }),
-    )
-    expect(foldPlanMode(agent.session.events)).toBe(true)
-    expect(ctx.planMode.get(agent)).toEqual({ active: true })
+    ctx.planMode.set(agent, true)
+    // A listener captured in the same dispatch snapshot keeps the plan-mode
+    // callback alive across the unload; the resumed wrapper must not append
+    // through the disposed service. Registered prepended AFTER the plugin so
+    // it runs before plan-mode's own prepended flush.
+    ctx.on('agent/step', async () => {
+      await fiber.dispose()
+    }, { prepend: true })
+    await agentEvents(ctx, agent).serial('agent/step', 1, 1, new AbortController().signal)
+    expect(agent.session.events.some(event => event.type === 'plan/mode')).toBe(false)
   })
 
-  it('skips the flush after the plugin fiber is disposed (a captured wrapper must not write into a dead service)', async () => {
+  it('skips the step-seam flush after the plugin fiber is disposed (a captured listener must not write into a dead service)', async () => {
     const ctx = new Context()
     await ctx.plugin(SystemPrompt)
     await ctx.plugin(ToolRegistry)
     const fiber = await ctx.plugin(PlanModeService, PLAN_CONFIG)
     const agent = await agentWithSession(ctx)
     ctx.planMode.set(agent, true)
-    // A downstream listener captured before disposal keeps the waterfall
-    // continuation alive across the unload; the resumed wrapper must not
-    // append through the disposed service.
-    ctx.on('agent/prompt-submit', async (_agent, _content, _source, _signal, next) => {
+    // Serial dispatch captures its listener list up front; prepending after
+    // the plugin puts this listener ahead of the plugin's own prepended one,
+    // so the plugin's captured callback still runs after the disposal below.
+    ctx.on('agent/step', async () => {
       await fiber.dispose()
-      return next()
-    })
-    await agentEvents(ctx, agent).waterfall(
-      'agent/prompt-submit', [{ type: 'text', text: 'probe' }], { kind: 'user' },
-      new AbortController().signal, () => Promise.resolve({ kind: 'allow' }),
-    )
+    }, { prepend: true })
+    await agentEvents(ctx, agent).serial('agent/step', 1, 1, new AbortController().signal)
     expect(agent.session.events.some(event => event.type === 'plan/mode')).toBe(false)
   })
 
@@ -259,7 +259,7 @@ describe('the boundary flush', () => {
     const agent = await agentWithSession(ctx)
     header(agent.session)
     ctx.planMode.set(agent, true)
-    await boundary(ctx, agent, 'turn/start')
+    await boundary(ctx, agent, 'step/end')
     expect(noticeTexts(agent.session)).toEqual(['The user switched this session to plan mode.'])
     await boundary(ctx, agent, 'step/end')
     expect(noticeTexts(agent.session)).toEqual(['The user switched this session to plan mode.'])
@@ -313,7 +313,7 @@ describe('the boundary flush', () => {
     expect(ctx.planMode.get(agent).pending).toBeUndefined()
   })
 
-  it('contains an append failure on the prompt-submit seam the same way', async () => {
+  it('prompt admission never appends, so a broken backend surfaces only at the step boundary', async () => {
     const ctx = await setup()
     const warn = vi.fn()
     ctx.logger.warn = warn as never
@@ -325,6 +325,8 @@ describe('the boundary flush', () => {
       return (original as (...args: unknown[]) => unknown)(type, ...rest)
     }) as unknown) as typeof agent.session.append
     await boundary(ctx, agent, 'turn/start')
+    expect(warn).not.toHaveBeenCalled()
+    await boundary(ctx, agent, 'step/end')
     expect(warn).toHaveBeenCalledOnce()
     expect(ctx.planMode.get(agent)).toEqual({ active: false, pending: true })
   })
@@ -541,7 +543,7 @@ describe('/plan', () => {
       .toEqual({ kind: 'success', text: 'Plan mode entry cancelled.' })
     expect(ctx.planMode.get(entering)).toEqual({ active: false, pending: false })
     expect(enteringSteer).not.toHaveBeenCalled()
-    await boundary(ctx, entering, 'turn/start')
+    await boundary(ctx, entering, 'step/end')
     expect(ctx.planMode.get(entering)).toEqual({ active: false })
     expect(entering.session.events.some(event => event.type === 'plan/mode')).toBe(false)
 
@@ -554,7 +556,7 @@ describe('/plan', () => {
     expect(await ctx.commands.execute(active, '/plan off', signal))
       .toEqual({ kind: 'success', text: 'Leaving plan mode (applies from the next step).' })
     expect(activeSteer).not.toHaveBeenCalled()
-    await boundary(ctx, active, 'turn/start')
+    await boundary(ctx, active, 'step/end')
     expect(ctx.planMode.get(active)).toEqual({ active: false })
   })