diff --git a/examples/coding-agent/README.md b/examples/coding-agent/README.md index dee1c83b09..44445adc5d 100644 --- a/examples/coding-agent/README.md +++ b/examples/coding-agent/README.md @@ -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= 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`. diff --git a/examples/coding-agent/cordis.yml b/examples/coding-agent/cordis.yml index 47eaad8a9b..07eb04d123 100644 --- a/examples/coding-agent/cordis.yml +++ b/examples/coding-agent/cordis.yml @@ -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. diff --git a/examples/coding-agent/tests/harness.ts b/examples/coding-agent/tests/harness.ts index db89b2c183..ca303691e9 100644 --- a/examples/coding-agent/tests/harness.ts +++ b/examples/coding-agent/tests/harness.ts @@ -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 { +export async function codingHarness(workdir: string, persistenceRoot?: string): Promise { const ctx = new Context() await ctx.plugin(LlmService) await ctx.plugin(SessionStore) @@ -31,6 +32,10 @@ export async function codingHarness(workdir: string): Promise { 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 } diff --git a/examples/coding-agent/tests/resume.e2e.ts b/examples/coding-agent/tests/resume.e2e.ts new file mode 100644 index 0000000000..6fd2dcb567 --- /dev/null +++ b/examples/coding-agent/tests/resume.e2e.ts @@ -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) +}) diff --git a/packages/agent-loop/src/index.ts b/packages/agent-loop/src/index.ts index a1efc78d56..9b40b1f770 100644 --- a/packages/agent-loop/src/index.ts +++ b/packages/agent-loop/src/index.ts @@ -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-`. 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 { - 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 { + 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 diff --git a/packages/agent-loop/tests/config-session-id.spec.ts b/packages/agent-loop/tests/config-session-id.spec.ts index b905d54145..52f0223564 100644 --- a/packages/agent-loop/tests/config-session-id.spec.ts +++ b/packages/agent-loop/tests/config-session-id.spec.ts @@ -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-), + // 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() + }) })