feat(agent-loop): config-driven session resume via RESUME_SESSION_ID
A config agent with `resumeSessionId` set continues a persisted session
instead of starting a fresh `${id}-session-<uuid>`. The id is sourced from
an env var in cordis.yml, so the coding-agent demo can resume a prior
conversation without code changes. The resume is deferred until the
`sessionPersistence` backend loads (via ctx.inject) and is contained: a
missing/unreadable id logs a warning and starts no agent. Adds a real-API
resume e2e proving cross-process continuity through the JSONL backend.
This commit is contained in:
@@ -22,6 +22,16 @@ Type a coding task. The agent's only tools are `bash` (+ `bash_output` / `bash_k
|
||||
…
|
||||
```
|
||||
|
||||
### Resuming a prior session
|
||||
|
||||
Each run starts a fresh session by default (its event log lands under `./.sessions/`). To **continue** a previous conversation, set `RESUME_SESSION_ID` to that session's id — the `main` agent then rehydrates the persisted log instead of starting fresh, so the model sees the earlier turns as history:
|
||||
|
||||
```sh
|
||||
RESUME_SESSION_ID=<prior-session-id> pnpm run demo:coding
|
||||
```
|
||||
|
||||
The id is wired through `cordis.yml` (`resumeSessionId: !!js process.env.RESUME_SESSION_ID`); unset, the agent starts a new session. A missing/unreadable id is non-fatal — it logs a warning and starts no `main` agent.
|
||||
|
||||
## What each plugin demonstrates
|
||||
|
||||
| Entry | Demonstrates |
|
||||
@@ -36,5 +46,6 @@ Type a coding task. The agent's only tools are `bash` (+ `bash_output` / `bash_k
|
||||
|
||||
- `tests/full-loop.e2e.ts` — the canary: real model runs `echo e2e-ok` through the real bash tool; asserts `tool/call`/`tool/result` session events and the final answer.
|
||||
- `tests/coding-task.e2e.ts` — the swebench-style smoke: a temp dir holds `add.js` (with `a - b` where `a + b` belongs) and a failing `add.test.js`; the agent must fix the bug and verify. The test re-runs `node add.test.js` ITSELF and inspects the files — agent claims are not trusted.
|
||||
- `tests/resume.e2e.ts` — durable continuity across processes: run 1 tells the real model a secret code and persists the turn to a temp JSONL root, then the whole context is disposed; run 2 is a fresh context over the same root that RESUMES the session id and asks the model to recall the code. The recall can only come from the rehydrated log.
|
||||
|
||||
Both self-skip without `DEEPSEEK_API_KEY`.
|
||||
@@ -61,6 +61,9 @@
|
||||
agents:
|
||||
- id: main
|
||||
model: deepseek-v4-flash
|
||||
# Set RESUME_SESSION_ID to continue a prior persisted session (the ids
|
||||
# live under ./.sessions); unset starts a fresh session each run.
|
||||
resumeSessionId: !!js process.env.RESUME_SESSION_ID
|
||||
systemPrompt: |
|
||||
You are coding-agent, a CLI coding assistant.
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ import AgentLoop, { LoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { LocalBashExecutor } from '@deepseek-ai/dsh-bash-local'
|
||||
import * as ToolBash from '@deepseek-ai/dsh-tool-bash'
|
||||
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
|
||||
/**
|
||||
* Shared harness for the coding-agent e2e suites: the full plugin stack
|
||||
@@ -20,7 +21,7 @@ export const SYSTEM_PROMPT = 'You are a coding agent. Your only tool is bash; '
|
||||
+ 'do file operations with cat/grep/heredocs, check [exit code: N] markers, '
|
||||
+ 'and report results briefly.'
|
||||
|
||||
export async function codingHarness(workdir: string): Promise<Context> {
|
||||
export async function codingHarness(workdir: string, persistenceRoot?: string): Promise<Context> {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
@@ -31,6 +32,10 @@ export async function codingHarness(workdir: string): Promise<Context> {
|
||||
await ctx.plugin(LlmDeepSeek, { models: ['deepseek-v4-flash'] })
|
||||
await ctx.plugin(LocalBashExecutor, { cwd: workdir, timeoutMs: 30_000 })
|
||||
await ctx.plugin(ToolBash)
|
||||
// Durable JSONL persistence is opt-in: only the resume e2e needs it, and the
|
||||
// other suites stay file-free. Loaded last so a resume's deferred
|
||||
// `ctx.inject(['sessionPersistence'])` resolves once this is present.
|
||||
if (persistenceRoot !== undefined) await ctx.plugin(SessionPersistenceJsonl, { root: persistenceRoot })
|
||||
return ctx
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import type { Context } from 'cordis'
|
||||
import type { LoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { codingHarness, finalText, SYSTEM_PROMPT, waitForIdle } from './harness.ts'
|
||||
|
||||
/**
|
||||
* Proves durable conversation continuity end-to-end: run 1 tells the REAL model
|
||||
* a fact and persists the turn to JSONL; run 2 is a fresh harness (new Context,
|
||||
* same `.sessions` root) that RESUMES the persisted session id and asks the
|
||||
* model to recall the fact. The recall can only come from the rehydrated event
|
||||
* log — a fresh session would have no idea. Key-gated like the other e2es.
|
||||
*/
|
||||
|
||||
const SECRET = 'plum-galaxy-1791'
|
||||
const SESSION_ID = 'resume-e2e-session'
|
||||
|
||||
let ctx: Context | undefined
|
||||
let root: string | undefined
|
||||
|
||||
afterEach(async () => {
|
||||
// Dispose even on failure/retry: agent-loop teardown stops the loop and the
|
||||
// JSONL backend flushes; then drop the on-disk session log.
|
||||
await ctx?.fiber.dispose()
|
||||
ctx = undefined
|
||||
if (root !== undefined) await rm(root, { recursive: true, force: true })
|
||||
root = undefined
|
||||
})
|
||||
|
||||
describe.skipIf(!process.env.DEEPSEEK_API_KEY)('resume: continue a persisted session across processes', () => {
|
||||
it('recalls a fact stored in a prior, separately-disposed session', async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'dsh-resume-e2e-'))
|
||||
|
||||
// Run 1: a fresh agent on a KNOWN session id learns a secret, then we
|
||||
// dispose the whole context (simulating process exit) so only the JSONL
|
||||
// log on disk survives.
|
||||
ctx = await codingHarness(process.cwd(), root)
|
||||
const first = ctx.agents.create({
|
||||
agentId: 'resume-1',
|
||||
sessionId: SESSION_ID,
|
||||
agentOptions: { model: 'deepseek-v4-flash', systemPrompt: SYSTEM_PROMPT },
|
||||
}) as LoopAgent
|
||||
first.send([{ type: 'text', text: `Remember this code for later: ${SECRET}. Just acknowledge it.` }])
|
||||
await waitForIdle(ctx, first)
|
||||
await ctx.fiber.dispose()
|
||||
ctx = undefined
|
||||
|
||||
// Run 2: a brand-new context over the SAME root resumes the persisted
|
||||
// session. The loaded event log seeds the live session, so the model sees
|
||||
// run 1's exchange as conversation history.
|
||||
ctx = await codingHarness(process.cwd(), root)
|
||||
const resumed = await ctx.agents.resume({
|
||||
agentId: 'resume-2',
|
||||
resumeSessionId: SESSION_ID,
|
||||
agentOptions: { model: 'deepseek-v4-flash', systemPrompt: SYSTEM_PROMPT },
|
||||
}) as LoopAgent
|
||||
expect(resumed.session.id).toBe(SESSION_ID)
|
||||
// The prior user turn is in the rehydrated log before the model is asked.
|
||||
expect(JSON.stringify(resumed.session.deriveMessages())).toContain(SECRET)
|
||||
|
||||
resumed.send([{ type: 'text', text: 'What was the code I asked you to remember? Reply with just the code.' }])
|
||||
await waitForIdle(ctx, resumed)
|
||||
|
||||
// The model recalls it — only possible from the resumed history.
|
||||
expect(finalText([...resumed.session.events])).toContain(SECRET)
|
||||
}, 180_000)
|
||||
})
|
||||
@@ -17,7 +17,7 @@ import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session } from '@deepseek-ai/dsh-session'
|
||||
import type {} from '@deepseek-ai/dsh-system-prompt'
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
import type {} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { LoopAgent } from './agent.ts'
|
||||
|
||||
export { LoopAgent } from './agent.ts'
|
||||
@@ -32,7 +32,19 @@ declare module 'cordis' {
|
||||
|
||||
export interface Config {
|
||||
/** Agents created from configuration at startup. */
|
||||
agents: (AgentOptions & { id: string })[]
|
||||
agents: (AgentOptions & {
|
||||
id: string
|
||||
/**
|
||||
* If set, the config agent RESUMES this persisted session id instead of
|
||||
* starting a fresh `${id}-session-<uuid>`. Sourced from an env var in
|
||||
* cordis.yml (`resumeSessionId: !!js process.env.RESUME_SESSION_ID`), so a
|
||||
* demo can continue a prior conversation without code changes. Requires a
|
||||
* `dsh-session-persistence` backend; the resume is deferred until that
|
||||
* service is available (via `ctx.inject`) and the loaded session's events
|
||||
* seed the live session so history continues.
|
||||
*/
|
||||
resumeSessionId?: string
|
||||
})[]
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -53,6 +65,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
id: z.string().required(),
|
||||
model: z.string(),
|
||||
systemPrompt: z.string(),
|
||||
resumeSessionId: z.string(),
|
||||
})).default([]),
|
||||
})
|
||||
|
||||
@@ -61,8 +74,27 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
// Provide the agent-creation factory to the registry (effect-scoped: the
|
||||
// slot is cleared on dispose).
|
||||
ctx.effect(() => this.ctx.agents.setFactory(this), 'agentLoop.setFactory()')
|
||||
for (const { id, ...options } of config.agents) {
|
||||
this.create(id, options)
|
||||
for (const { id, resumeSessionId, ...options } of config.agents) {
|
||||
if (resumeSessionId !== undefined && resumeSessionId !== '') {
|
||||
// Resume a prior session instead of starting fresh. resume() needs
|
||||
// `ctx.sessionPersistence`, which may load AFTER this plugin (cordis.yml
|
||||
// lists the backend later). `ctx.inject(['sessionPersistence'], cb)`
|
||||
// runs `cb` with a child ctx once the service exists; the child reads
|
||||
// the persistence and hands it to resumeWith (which uses this.ctx — the
|
||||
// parent — for sessions/registry, all in AgentLoop's static inject). A
|
||||
// failed resume is contained + logged: startup must not crash.
|
||||
ctx.effect(() => {
|
||||
const fiber = this.ctx.inject(['sessionPersistence'], (childCtx: Context) => {
|
||||
void this.resumeWith(childCtx.sessionPersistence, { agentId: id, resumeSessionId, agentOptions: options })
|
||||
.catch((error: unknown) => {
|
||||
this.ctx.logger.warn(`agent "${id}": config-driven resume of "${resumeSessionId}" failed: ${String(error)}`)
|
||||
})
|
||||
})
|
||||
return () => void fiber.dispose()
|
||||
}, `agentLoop.resume(${id})`)
|
||||
} else {
|
||||
this.create(id, options)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -120,7 +152,6 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* by the time this runs the service exists.
|
||||
*/
|
||||
async resume(options: ResumeAgentOptions): Promise<Agent> {
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
const persistence = this.ctx.sessionPersistence
|
||||
// `sessionPersistence` is declaration-merged onto Context as non-optional,
|
||||
// but the service is only present when a backend plugin is loaded — and
|
||||
@@ -130,6 +161,20 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
if (persistence === undefined) {
|
||||
throw new Error('cannot resume: session persistence is not configured (load a dsh-session-persistence backend)')
|
||||
}
|
||||
return this.resumeWith(persistence, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Resume against an EXPLICIT persistence handle. Factored out of {@link resume}
|
||||
* so the config-driven path can pass the handle it obtained from a
|
||||
* `ctx.inject(['sessionPersistence'], …)` child context: `this.ctx` (the
|
||||
* service's own fiber) did not inject `sessionPersistence`, so reading it
|
||||
* there from inside the inject child trips the cordis inject guard. The
|
||||
* sessions store + registry are still read through `this.ctx` (both are in
|
||||
* AgentLoop's static inject, so they resolve fine).
|
||||
*/
|
||||
private async resumeWith(persistence: SessionPersistence, options: ResumeAgentOptions): Promise<Agent> {
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
const { meta, events } = await persistence.load(SessionId(options.resumeSessionId))
|
||||
// Re-check the agent id AFTER the await: the pre-load check above can go
|
||||
// stale while load() is pending (a concurrent resume/create may register the
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
@@ -62,4 +62,76 @@ describe('config-driven session id', () => {
|
||||
await waitForIdle(ctx2, a2)
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('config-driven resumeSessionId continues a persisted session (env-var resume)', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-resume-'))
|
||||
dirs.push(root)
|
||||
|
||||
// Run 1: a programmatically-created agent on a KNOWN session id persists a
|
||||
// completed turn, so run 2 has a concrete id to resume.
|
||||
const ctx1 = new Context()
|
||||
await ctx1.plugin(LlmService)
|
||||
await ctx1.plugin(SessionStore)
|
||||
await ctx1.plugin(SystemPrompt)
|
||||
await ctx1.plugin(ToolRegistry)
|
||||
await ctx1.plugin(AgentRegistry)
|
||||
await ctx1.plugin(AgentLoop, { agents: [] })
|
||||
await ctx1.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx1.llm.registerAdapter(['mock'], new MockAdapter([textResponse('first')]))
|
||||
const a1 = ctx1.agents.create({ agentId: 'main', sessionId: 'sticky-1' }) as LoopAgent
|
||||
a1.send([{ type: 'text', text: 'remember me' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx1, a1)
|
||||
await ctx1.fiber.dispose()
|
||||
|
||||
// Run 2: a CONFIG agent with resumeSessionId continues that session. The
|
||||
// resume is deferred until sessionPersistence loads (ctx.inject), so wait
|
||||
// for the agent to appear, then assert it is on the resumed id with history.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(LlmService)
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SystemPrompt)
|
||||
await ctx2.plugin(ToolRegistry)
|
||||
await ctx2.plugin(AgentRegistry)
|
||||
await ctx2.plugin(AgentLoop, { agents: [{ id: 'main', model: 'mock', systemPrompt: '', resumeSessionId: 'sticky-1' }] })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx2.llm.registerAdapter(['mock'], new MockAdapter([textResponse('second')]))
|
||||
|
||||
// The deferred resume runs on a microtask after the backend is available.
|
||||
let resumed: LoopAgent | undefined
|
||||
for (let i = 0; i < 50 && !resumed; i++) {
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
resumed = ctx2.agents.get('main') as LoopAgent | undefined
|
||||
}
|
||||
expect(resumed).toBeDefined()
|
||||
// The live session id IS the resumed id (NOT a fresh ${id}-session-<uuid>),
|
||||
// and the prior turn's user message is in the derived history.
|
||||
expect(resumed!.session.id).toBe('sticky-1')
|
||||
const derived = resumed!.session.deriveMessages()
|
||||
expect(JSON.stringify(derived)).toContain('remember me')
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('config-driven resume of a missing session is contained: logs a warning, no agent, no crash', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-resume-miss-'))
|
||||
dirs.push(root)
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt)
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(AgentLoop, { agents: [{ id: 'main', model: 'mock', systemPrompt: '', resumeSessionId: 'does-not-exist' }] })
|
||||
const warn = vi.spyOn((ctx.agentLoop as unknown as { ctx: { logger: { warn: (...a: unknown[]) => void } } }).ctx.logger, 'warn')
|
||||
.mockImplementation(() => undefined)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('x')]))
|
||||
|
||||
// The deferred resume fails (no such session on disk). It must be contained:
|
||||
// a warning is logged, no 'main' agent is registered, and the app stays up.
|
||||
await new Promise(r => setTimeout(r, 200))
|
||||
expect(ctx.agents.get('main')).toBeUndefined()
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('config-driven resume of "does-not-exist" failed'))
|
||||
warn.mockRestore()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user