refactor(agent-loop): rely on eager session persistence

packages/core/agent-loop/README.md

@@ -12,7 +12,7 @@ Creation and resume are one rollback-covered transaction: construct a private se
 
 The caller fiber and the AgentLoop provider are co-owners. `AgentFactory.createAgent(ownerCtx, options)` and `resume(ownerCtx, options)` receive caller ownership explicitly, while the factory keeps its own dependency context for `sessions`/`llm`/`tools`/`systemPrompt`; this lets a caller inject only `agents` without shrinking the new agent's service surface. Caller unload, handle disposal, or provider unload converge on one memoized quiescence boundary. Provider shutdown waits both resource teardown and the public create/resume wrapper that observed deactivation, so no continuation can publish after dependencies disappear.
 
-Each agent and its session share one caller-chosen `SessionId`, assumed globally unique; accidental UUID collisions are outside the supported model. Two concurrent operations with the same id may both prepare, but the final `enter()` calls arbitrate publication and every loser rolls its private resources back. Each detach is bound to the exact entered object, so a stale disposer cannot remove a later same-id replacement. A detach requested during a synchronous creation notification waits for that dispatch to unwind, preserving created/disposed pairing. Teardown runs stop and drain (including outstanding idle-injection flushes) → detach agent → detach session → unwind scope; the id becomes reusable at detach even if private scope cleanup is still finishing. Ordinary non-vetoing `agent/*` notifications go through `agentEvents(ctx, agent)`, per-step assembly goes through `assembleContextFor(agent)`, and turn-end durability checkpoints go through `ctx.sessions.flush(session)`.
+Each agent and its session share one caller-chosen `SessionId`, assumed globally unique; accidental UUID collisions are outside the supported model. Two concurrent operations with the same id may both prepare, but the final `enter()` calls arbitrate publication and every loser rolls its private resources back. Each detach is bound to the exact entered object, so a stale disposer cannot remove a later same-id replacement. A detach requested during a synchronous creation notification waits for that dispatch to unwind, preserving created/disposed pairing. Teardown runs stop and drain → detach agent → detach session → unwind scope; the id becomes reusable at detach even if private scope cleanup is still finishing. Ordinary non-vetoing `agent/*` notifications go through `agentEvents(ctx, agent)`, per-step assembly goes through `assembleContextFor(agent)`, and turn-end durability checkpoints go through `ctx.sessions.flush(session)`.
 
 - `ctx.agentLoop.create(id: SessionId, options?: AgentOptions, meta?: { cwd?: string }): Agent` — synchronous no-setup create under the exact shared agent/session id, disposed with the calling fiber. Declarative config treats `agents[].id` as a stable label and normally mints `${label}-session-<uuid>` before calling this boundary. An app may instead supply a stable exact `sessionId`: first use creates it, while a remount with persistence already present resumes its materialized history. `resumeSessionId` requires and loads an existing persisted id and is mutually exclusive with `sessionId`. This keeps default fresh restarts collision-free without retaining a second live routing identity.
 
@@ -52,7 +52,7 @@ Configured agents start automatically. A model call requires both `provider` and
 
 The concrete `Agent` class, its `Inbox`, `runLoop`, and instance-bound publication/start controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy.
 
-The unified `send()` primitive routes content and source by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO, waking the driver unless `wakeup: false`; admission happens before any turn opens. An allowed prompt and the prompt waterfall's `additionalContexts` enter the outbox as separate messages, then `run()` opens the turn and drains them together. A running `next-step`/wakeup `steer()` enters the same outbox without prompt admission and normally causes another step. A `next-step`/no-wakeup `inject()` waits there only while a turn is open; while idle it appends and flushes a `user/message` immediately without opening a turn or running the model. Every inbox enqueue publishes `agent/inbox/enqueue`; taking it publishes `agent/inbox/dequeue`, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`.
+The unified `send()` primitive routes content and source by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO, waking the driver unless `wakeup: false`; admission happens before any turn opens. An allowed prompt and the prompt waterfall's `additionalContexts` enter the outbox as separate messages, then `run()` opens the turn and drains them together. A running `next-step`/wakeup `steer()` enters the same outbox without prompt admission and normally causes another step. A `next-step`/no-wakeup `inject()` waits there only while a turn is open; while idle it appends a `user/message` immediately without opening a turn or running the model. Persistence owns the resulting eager drain. Every inbox enqueue publishes `agent/inbox/enqueue`; taking it publishes `agent/inbox/dequeue`, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`.
 
 ### Loop lifecycle (`loop.ts`)
 

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

@@ -139,11 +139,6 @@ export class ReactLoopAgent extends Agent {
         return id
       }
       this.session.append('user/message', { content, source }, { surfaceOp: 'append' })
-      const previous = this.done
-      const flush = this.loopCtx.sessions.flush(this.session).catch((error: unknown) => {
-        this.loopCtx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${errorChain(toError(error))}`)
-      })
-      this.done = Promise.all([previous, flush]).then(() => undefined)
       return id
     }
 
@@ -199,19 +194,15 @@ export class ReactLoopAgent extends Agent {
    */
   retry(): void {
     if (this.abort !== undefined) throw new Error(`agent "${this.id}" cannot retry while busy`)
-    const previous = this.done
-    const run = this.loopCtx.agents.withInitiator(this, () => this.run({ kind: 'retry' }))
-    this.done = Promise.all([previous, run]).then(() => undefined)
+    this.done = this.loopCtx.agents.withInitiator(this, () => this.run({ kind: 'retry' }))
   }
 
   /** Resolve at idle quiescence: no run driving and no waking prompt waiting. */
   async whenIdle(): Promise<void> {
-    // `done` is replaced by runs and idle-injection flushes. Re-read after
-    // every settlement so work admitted by a synchronous observer is included.
-    while (true) {
-      const done = this.done
-      await done.catch(() => undefined)
-      if (done === this.done && this.abort === undefined && !this.queued.some(message => message.wakeup)) return
+    // `done` is replaced per activity, so re-reading it follows chained turns;
+    // a run failure still counts as quiescence for the waiter.
+    while (this.abort !== undefined || this.queued.some(message => message.wakeup)) {
+      await this.done.catch(() => undefined)
     }
   }
 
@@ -224,8 +215,7 @@ export class ReactLoopAgent extends Agent {
     emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', inboxMessage(message, false))
     const admission = new AbortController()
     this.abort = admission
-    const previous = this.done
-    const admissionTask = this.loopCtx.agents.withInitiator(this, async () => {
+    this.done = this.loopCtx.agents.withInitiator(this, async () => {
       const signal = admission.signal
       const trigger: TurnTrigger = { kind: 'message', source: message.source }
       let admitted = false
@@ -258,7 +248,6 @@ export class ReactLoopAgent extends Agent {
       }
       await this.run(trigger)
     })
-    this.done = Promise.all([previous, admissionTask]).then(() => undefined)
   }
 
   /** Own one complete turn over input already admitted by {@link kick}, or retry history as-is. */

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

@@ -6,7 +6,7 @@ import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry from '@deepseek-ai/dsh-tools'
 import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
 import AgentLoop from '@deepseek-ai/dsh-agent-loop'
-import { bindReactLoopAgentContext, prepareReactLoopAgent, type ReactLoopAgent } from '../src/agent.ts'
+import { bindReactLoopAgentContext, prepareReactLoopAgent } from '../src/agent.ts'
 import { MockAdapter, textResponse } from './mock-adapter.ts'
 
 function driverDone(agent: Agent): Promise<void> {
@@ -83,30 +83,19 @@ describe('Agent', () => {
     await ctx.fiber.dispose()
   })
 
-  it('idle inject() appends context and flushes without opening a turn', async () => {
+  it('idle inject() appends context without opening a turn or requesting a flush', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
-    const release = Promise.withResolvers<void>()
     let flushes = 0
-    ctx.on('session/flush', async (session) => {
-      if (session !== agent.session) return
-      flushes += 1
-      await release.promise
-    })
+    ctx.on('session/flush', () => { flushes += 1 })
 
     agent.inject([{ type: 'text', text: 'context' }], { source: { kind: 'plugin', plugin: 'p' } })
     expect(agent.session.events.map(event => event.type)).toEqual(['user/message'])
     expect(agent.status).toBe('idle')
     expect(adapter.requests).toHaveLength(0)
-
-    let idle = false
-    const settled = agent.whenIdle().then(() => { idle = true })
-    await Promise.resolve()
-    expect(flushes).toBe(1)
-    expect(idle).toBe(false)
-    release.resolve()
-    await settled
+    await agent.whenIdle()
+    expect(flushes).toBe(0)
   })
 
   it('inject() defaults its source to an empty plugin, never user', async () => {
@@ -119,35 +108,15 @@ describe('Agent', () => {
     await agent.whenIdle()
   })
 
-  it('idle inject() contains a failing flush without inventing an agent turn error', async () => {
-    const adapter = new MockAdapter([textResponse('ok')])
-    const ctx = await harness(adapter)
-    ctx.on('session/flush', () => { throw new Error('disk gone') })
-    const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
-    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
-    const errors: Error[] = []
-    ctx.on('agent/error', (_agent, _turn, _step, error) => { errors.push(error) })
-
-    expect(() => { agent.inject([{ type: 'text', text: 'notice' }], { source: { kind: 'plugin', plugin: 'p' } }) }).not.toThrow()
-    await agent.whenIdle()
-    expect(errors).toEqual([])
-    expect(agent.session.events.map(event => event.type)).toEqual(['user/message'])
-    expect(warn).toHaveBeenCalledWith(expect.stringContaining('flush after idle injection failed'))
-    warn.mockRestore()
-  })
-
-  it('idle inject() does not flush input rejected before append', async () => {
+  it('idle inject() rejects invalid input before append', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
-    let flushes = 0
-    ctx.on('session/flush', () => { flushes += 1 })
 
     expect(() => {
       agent.inject([{ type: 'text', text: 'x', bad: 1n } as never], { source: { kind: 'plugin', plugin: 'p' } })
     }).toThrow(/non-JSON-serializable/)
     expect(agent.session.events).toHaveLength(0)
-    expect(flushes).toBe(0)
   })
 
   it('steer() when idle falls through to send() and starts a turn', async () => {

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

@@ -474,28 +474,6 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
     await ctx2.fiber.dispose()
   })
 
-  it('an idle inject() is flushed durably on its own (survives without explicit flush/dispose)', async () => {
-    // No clean disposal follows, so disk presence proves the idle injection's
-    // own checkpoint ran without a synthetic turn.
-    const adapter1 = new MockAdapter([textResponse('answer')])
-    const { ctx: ctx1, root } = await persistentHarness(adapter1)
-    const a1 = (await ctx1.agents.create({ sessionId: SessionId('inject-sess'), meta: { cwd: '/w' } })).agent
-    a1.followup([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
-    await waitForIdle(ctx1, a1)
-    a1.inject([{ type: 'text', text: 'background task 42 finished' }], { source: { kind: 'plugin', plugin: 'tool-bash' } })
-    await a1.whenIdle()
-
-    // A SEPARATE backend reads the on-disk log — proving the inject persisted
-    // itself, not a later dispose drain.
-    const probe = new Context()
-    await probe.plugin(SessionStore)
-    await probe.plugin(SessionPersistenceJsonl, { root })
-    const loaded = await probe.sessionPersistence.load(SessionId('inject-sess'))
-    expect(JSON.stringify(loaded.events)).toContain('background task 42 finished')
-    await probe.fiber.dispose()
-    await ctx1.fiber.dispose()
-  })
-
   it('an idle inject() survives persist + resume without a synthetic turn', async () => {
     const adapter1 = new MockAdapter([textResponse('answer')])
     const { ctx: ctx1, root } = await persistentHarness(adapter1)

packages/core/agent-loop/tests/scope-lifecycle.spec.ts

@@ -1041,33 +1041,4 @@ describe('agent scope lifecycle', () => {
     await ctx.fiber.dispose()
   })
 
-  it('handle.dispose() awaits an idle-injection flush before unregistering or detaching', async () => {
-    const ctx = await harness()
-    const handle = await ctx.agents.create({
-      sessionId: SessionId('idle-flush-s'),
-      agentOptions: { provider: 'mock', model: 'mock' },
-    })
-    const gate = Promise.withResolvers<undefined>()
-    let flushStarted = false
-    ctx.on('session/flush', (session) => {
-      if (session !== handle.agent.session) return
-      flushStarted = true
-      return gate.promise
-    })
-
-    handle.agent.inject(text('durable idle context'), { source: { kind: 'plugin', plugin: 'test' } })
-    expect(flushStarted).toBe(true)
-
-    let disposed = false
-    const disposal = handle.dispose().then(() => { disposed = true })
-    await new Promise(resolve => setTimeout(resolve, 0))
-    expect(disposed).toBe(false)
-    expect(ctx.agents.get(SessionId('idle-flush-s'))).toBe(handle.agent)
-    expect(ctx.sessions.get(SessionId('idle-flush-s'))).toBe(handle.agent.session)
-
-    gate.resolve(undefined)
-    await disposal
-    expect(ctx.agents.get(SessionId('idle-flush-s'))).toBeUndefined()
-    expect(ctx.sessions.get(SessionId('idle-flush-s'))).toBeUndefined()
-  })
 })