From 9a4006cb2b3ff26b1da19383b56806bef9903de7 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 15 Jun 2026 21:12:14 +0800 Subject: [PATCH 1/3] feat(agent): create/resume factory seam MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the agent-creation factory seam on ctx.agents (AgentRegistry): setFactory/create/resume plus the AgentFactory interface and CreateAgentOptions/ResumeAgentOptions. AgentLoop implements AgentFactory and registers itself via ctx.agents.setFactory(this), so plugins create/resume agents through the interface without depending on the concrete loop package. - create({ agentId, sessionId, meta?, agentOptions? }) — programmatic create on a caller-supplied session id (e.g. an ACP-generated id). - resume({ agentId, resumeSessionId, agentOptions? }) — load a persisted session via ctx.sessionPersistence (RFC 009) and resume an agent on it; the live session id is the resumed id, turn numbering and derived history continue from the loaded log. sessionPersistence is NOT hard-injected (non-persistent demos still work); resume rejects with a typed error when it is absent. assertAgentIdFree runs before any session is created (and again after the load await) so a duplicate id never leaves an orphaned live session. Adds the runtime dsh-session-persistence dependency to agent-loop. --- docs/adr/0016-session-persistence.md | 2 +- docs/adr/0017-turn-enclosure-invariant.md | 4 +- docs/adr/README.md | 2 +- docs/architecture.md | 4 +- packages/agent-loop/README.md | 7 +- packages/agent-loop/package.json | 2 + packages/agent-loop/src/index.ts | 111 ++++++++-- packages/agent-loop/tests/resume.spec.ts | 241 ++++++++++++++++++++++ packages/agent-loop/tsconfig.json | 1 + packages/agent/README.md | 12 +- packages/agent/src/index.ts | 98 ++++++++- packages/agent/tests/agent.spec.ts | 55 +++++ yarn.lock | 2 + 13 files changed, 516 insertions(+), 25 deletions(-) create mode 100644 packages/agent-loop/tests/resume.spec.ts diff --git a/docs/adr/0016-session-persistence.md b/docs/adr/0016-session-persistence.md index 9be7437841..4ab7ad6e37 100644 --- a/docs/adr/0016-session-persistence.md +++ b/docs/adr/0016-session-persistence.md @@ -21,7 +21,7 @@ Key choices recorded here because they are durable, contested, and surprising: - **Append-only with a single exception.** Committed events — those at or below a flushed `turn/end` — are never rewritten. The loop only flushes at `turn/end`, so a crash can leave a half-written final turn below the last checkpoint; `load` returns events only up to the **last complete `turn/end`**, and the first post-load `append` runs a one-time **truncation-repair** (`ftruncate` + `fsync`) that physically discards only that never-committed crash tail before writing. - **File backend canonical, DB backend a drop-in.** `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — `append` is INSERT (in a transaction asserting the contiguous-seq contract), `load` is SELECT … ORDER BY seq. A future `dsh-session-persistence-sqlite` is a `SessionPersistence` subclass with no interface change (opencode runs this exact shape on SQLite/WAL). - **Metadata is out-of-log.** Format version, cwd, and lineage are storage concerns, not replayable conversation state, so they live in a `SessionMeta` (`SessionHeader & SessionSummary`) owned by `dsh-session` and attached to a `Session` via a new readonly `session.header` — never in `SessionEventMap`, never reaching `deriveMessages()`. The alternative (a merge-extensible `session/meta` event as log line 0) was rejected: an in-log event would ride along with a seeded/forked session for free, but metadata is not replayable state, so the explicit out-of-log header seam is the cleaner cost. -- **`load` returns a resumable event log, not just bytes.** `load(sessionId)` yields the `SessionMeta` plus the committed `SessionEvent[]` (through the last complete `turn/end`), shaped so a caller can reconstruct a live session with the loaded events as seed (so `lastTurnNumber`/`deriveMessages` continue) on the SAME session id. The agent-facing create/resume factory that consumes this is a separate seam (a follow-up on `ctx.agents`); the persistence layer deliberately stops at the `load` primitive and does NOT reach into the loop. The agent-loop does NOT hard-inject `sessionPersistence` (that would pend non-persistent demos forever), so any resume path built on this rejects with a typed error when the backend is absent. +- **Resume is an async factory, not a change to synchronous create.** `ctx.agents.resume({ resumeSessionId })` awaits `ctx.sessionPersistence.load`, recreates the live session with the loaded events (so `lastTurnNumber`/`deriveMessages` continue), and starts a fresh agent on the resumed id (NOT `${agentId}-session`). The agent-loop does NOT hard-inject `sessionPersistence` (that would pend non-persistent demos forever); `resume` rejects with a typed error when it is absent. Format versioning: the header carries a `version`; `load` rejects an unknown version (no v1 migration). Stated honestly: append-only + flush is robust to partial trailing writes (tolerated on load) but not to fsync-less power loss mid-line; a DB/WAL backend is the stronger option later. diff --git a/docs/adr/0017-turn-enclosure-invariant.md b/docs/adr/0017-turn-enclosure-invariant.md index f4b88420ea..06f243bc0f 100644 --- a/docs/adr/0017-turn-enclosure-invariant.md +++ b/docs/adr/0017-turn-enclosure-invariant.md @@ -4,7 +4,7 @@ Status: accepted (2026-06-15) ## Context -A durable session-persistence backend (added in a companion change) uses the **turn** as its crash-recovery boundary: `load` returns events only up to the last complete `turn/end`, and the first post-load `append` truncates whatever follows as a never-committed crash tail. This is safe only if nothing *legitimately* durable can sit after the last `turn/end`. +The durable JSONL backend ([ADR 0016](0016-session-persistence.md)) uses the **turn** as its crash-recovery boundary: `load` returns events only up to the last complete `turn/end`, and the first post-load `append` truncates whatever follows as a never-committed crash tail. This is safe only if nothing *legitimately* durable can sit after the last `turn/end`. That assumption did not hold. Two paths recorded events outside any turn: @@ -29,7 +29,7 @@ The serializability invariant is enforced at the same source boundary (`Session. ## Consequences -The turn is now the *single* durability/replay boundary, so a persistence backend's "last `turn/end` = commit point" rule is complete, not merely sufficient: a backend can discard everything after the last `turn/end` with zero risk of losing between-turn context, because there is no between-turn context. `scanLog` stays simple (no partial-turn boundary walk), and an idle background-task notice survives persist + resume. +The turn is now the *single* durability/replay boundary, so [ADR 0016](0016-session-persistence.md)'s "last `turn/end` = commit point" rule is complete, not merely sufficient: a backend can discard everything after the last `turn/end` with zero risk of losing between-turn context, because there is no between-turn context. `scanLog` stays simple (no partial-turn boundary walk), and an idle background-task notice survives persist + resume. Costs: `agent.inject()` while idle now writes three log lines instead of one, and the derived history gains a turn that carries only injected context (no assistant output) — `deriveMessages()` already derives purely by event type, so this renders identically. The `injection` trigger is a new on-disk vocabulary value; like every `SessionEventMap`/`TurnTriggerMap` addition it is part of the frozen format. Event ordering within a turn changed (`turn/start` now precedes `user/message`), which is observable to anything that asserted the old order — the loop's own tests were the only such consumers. diff --git a/docs/adr/README.md b/docs/adr/README.md index b69ad2018a..c6e94139dd 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -27,5 +27,5 @@ Do NOT write an ADR for: a mechanical or local choice (a variable name, a one-fi | [0013](0013-property-based-testing.md) | Property-based testing for protocol-shaped code | accepted | | [0014](0014-doc-sync-enforcement.md) | Doc-sync enforcement (doc code blocks + event taxonomy) | accepted | | [0015](0015-structured-error-taxonomy.md) | Structured error taxonomy (HarnessError base) | accepted | -| [0016](0016-session-persistence.md) | Session persistence as an abstract service over `SessionEvent` | accepted | +| [0016](0016-session-persistence.md) | Session persistence as an abstract service over the existing `SessionEvent` | accepted | | [0017](0017-turn-enclosure-invariant.md) | Every session event is enclosed in a turn | accepted | diff --git a/docs/architecture.md b/docs/architecture.md index c5e0675a43..85398a9f7c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -48,7 +48,7 @@ Dependency rule: plugins depend on interface packages, never on `dsh-agent-loop` | `ctx.sessionPersistence` | `SessionPersistence` (abstract) | dsh-session-persistence | durable persistence seam: create/append/load/list/update sessions | | `ctx.systemPrompt` | `SystemPrompt` | dsh-system-prompt | ordered sections + tool schemas → `assemble()` | | `ctx.tools` | `ToolRegistry` | dsh-tools | tool definitions; `execute()` through waterfall | -| `ctx.agents` | `AgentRegistry` | dsh-agent | live `Agent` handles | +| `ctx.agents` | `AgentRegistry` | dsh-agent | live `Agent` handles + the create/resume factory seam | | `ctx.agentLoop` | `AgentLoop` | dsh-agent-loop | creates `LoopAgent`s and drives their loops | | `ctx.bash` | `BashExecutor` (abstract) | dsh-bash | bash execution seam: foreground runs + background tasks | @@ -95,7 +95,7 @@ A `Session` is an append-only log of typed `SessionEvent`s — the single source Replay/fork = `ctx.sessions.create(id, { seed: seedEvents })`. Trace/telemetry = listen to `session/event`. -**Durability seam**: `session/event` is a synchronous notification; persistence plugins buffer (write-behind) and drain at the awaited `session/flush` checkpoint the loop fires at every turn end. The durable backend is a real **capability seam**: the abstract `SessionPersistence` service (`dsh-session-persistence`, `ctx.sessionPersistence`) defines create/append/load/list/update over the existing `SessionEvent` (no parallel persisted type), and `dsh-session-persistence-jsonl` is the first implementation — an append-only JSONL log per session with crash-safe atomic writes, truncation-repair of a never-committed crash tail, and a read/replay path. Session metadata (format version, cwd, lineage) travels separately as `SessionMeta`, attached to a `Session` via `session.header`. `ctx.sessionPersistence.load(sessionId)` returns the committed event log so a caller can reconstruct a live session and continue it. A SQLite/WAL backend is a future drop-in `SessionPersistence` subclass (the row shape `(session_id, seq, type, time, data)` maps 1:1 onto `SessionEvent`). +**Durability seam**: `session/event` is a synchronous notification; persistence plugins buffer (write-behind) and drain at the awaited `session/flush` checkpoint the loop fires at every turn end. The durable backend is a real **capability seam**: the abstract `SessionPersistence` service (`dsh-session-persistence`, `ctx.sessionPersistence`) defines create/append/load/list/update over the existing `SessionEvent` (no parallel persisted type), and `dsh-session-persistence-jsonl` is the first implementation — an append-only JSONL log per session with crash-safe atomic writes, truncation-repair of a never-committed crash tail, and a read/replay path. Session metadata (format version, cwd, lineage) travels separately as `SessionMeta`, attached to a `Session` via `session.header`. Resuming a persisted session into a live agent is `ctx.agents.resume({ resumeSessionId })`. A SQLite/WAL backend is a future drop-in `SessionPersistence` subclass (the row shape `(session_id, seq, type, time, data)` maps 1:1 onto `SessionEvent`). ## Prompt assembly (dsh-system-prompt) diff --git a/packages/agent-loop/README.md b/packages/agent-loop/README.md index de5074f32a..1092fb09df 100644 --- a/packages/agent-loop/README.md +++ b/packages/agent-loop/README.md @@ -8,7 +8,12 @@ This is the only package in the harness that contains concrete loop logic. Every ### Public API -- `ctx.agentLoop.create(id: string, options?: AgentOptions): LoopAgent` — create an agent on a fresh per-run session id `${id}-session-`, start its loop, and register it in `ctx.agents`. The per-run uuid avoids colliding with the on-disk log a prior run materialized once a durable persistence backend is loaded; each run is a new session (a deliberate demo simplification — a real resume-or-create policy is a TODO). Disposed with the calling fiber. +- `ctx.agentLoop.create(id: string, options?: AgentOptions): LoopAgent` — config-driven create: an agent on a fresh per-run session id `${id}-session-` (no cwd). Used for `cordis.yml`-configured agents. The per-run uuid avoids colliding with the on-disk log a prior run materialized once a durable persistence backend is loaded; each run is a new session (a deliberate demo simplification — a real resume-or-create policy is a TODO). Disposed with the calling fiber. + +`AgentLoop` also implements the `AgentFactory` seam and registers itself via `ctx.agents.setFactory(this)`, so plugins create/resume agents through `ctx.agents` (the interface): + +- `ctx.agents.create({ agentId, sessionId, meta?, agentOptions? })` — programmatic create on a caller-supplied `sessionId` (e.g. an ACP-generated id), NOT `${id}-session`. +- `ctx.agents.resume({ agentId, resumeSessionId, agentOptions? })` — load a persisted session via `ctx.sessionPersistence` (RFC 009) and resume an agent on it. The live session id is the resumed id; turn numbering and derived history continue from the loaded log. Requires a session-persistence backend (NOT hard-injected — non-persistent demos still work; `resume` rejects with a typed error when persistence is absent). ### Injected services diff --git a/packages/agent-loop/package.json b/packages/agent-loop/package.json index 3482691555..81d588da90 100644 --- a/packages/agent-loop/package.json +++ b/packages/agent-loop/package.json @@ -23,6 +23,7 @@ "@deepseek-ai/dsh-agent": "^0.0.1", "@deepseek-ai/dsh-llm": "^0.0.1", "@deepseek-ai/dsh-session": "^0.0.1", + "@deepseek-ai/dsh-session-persistence": "^0.0.1", "@deepseek-ai/dsh-system-prompt": "^0.0.1", "@deepseek-ai/dsh-tools": "^0.0.1", "cordis": "^4.0.0-rc.6" @@ -35,6 +36,7 @@ "@deepseek-ai/dsh-invariants": "^0.0.1", "@deepseek-ai/dsh-llm": "^0.0.1", "@deepseek-ai/dsh-session": "^0.0.1", + "@deepseek-ai/dsh-session-persistence": "^0.0.1", "@deepseek-ai/dsh-session-persistence-jsonl": "^0.0.1", "@deepseek-ai/dsh-system-prompt": "^0.0.1", "@deepseek-ai/dsh-tools": "^0.0.1", diff --git a/packages/agent-loop/src/index.ts b/packages/agent-loop/src/index.ts index c273592fd0..355c610989 100644 --- a/packages/agent-loop/src/index.ts +++ b/packages/agent-loop/src/index.ts @@ -11,11 +11,13 @@ import { Context, Service } from 'cordis' import { randomUUID } from 'node:crypto' import z from 'schemastery' import { AgentId } from '@deepseek-ai/dsh-agent' -import type { AgentOptions } from '@deepseek-ai/dsh-agent' +import type { Agent, AgentFactory, AgentOptions, CreateAgentOptions, ResumeAgentOptions } from '@deepseek-ai/dsh-agent' import type {} from '@deepseek-ai/dsh-llm' -import type {} from '@deepseek-ai/dsh-session' +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 { LoopAgent } from './agent.ts' export { LoopAgent } from './agent.ts' @@ -35,13 +37,15 @@ export interface Config { /** * The agent-loop plugin (`ctx.agentLoop`): creates {@link LoopAgent}s, runs - * their loops, and registers them in `ctx.agents`. + * their loops, and registers them in `ctx.agents`. Also implements the + * {@link AgentFactory} seam, so plugins create/resume agents through + * `ctx.agents` (the interface) without depending on this concrete package. * * The loop itself is deliberately thin — every behavior beyond "call the * model, run the tools, repeat" belongs to plugins listening on the event * taxonomy declared in @deepseek-ai/dsh-agent. */ -export class AgentLoop extends Service { +export class AgentLoop extends Service implements AgentFactory { static inject = ['agents', 'sessions', 'llm', 'tools', 'systemPrompt'] static Config: z = z.object({ @@ -54,20 +58,23 @@ export class AgentLoop extends Service { constructor(ctx: Context, public config: Config) { super(ctx, 'agentLoop') + // 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) } } /** - * Create an agent, start its loop, and register it. Returns the agent. - * Disposed with the calling fiber. + * Config-driven create: an agent on a FRESH, non-colliding session id per run + * (`${id}-session-`, no cwd). Used for `cordis.yml`-configured agents + * and as the shared core for the programmatic factory {@link createAgent}. * - * The session id is per-run (`${id}-session-`, no fixed name): once a - * durable persistence backend is loaded, a fixed `${id}-session` collides on - * the second run — the backend refuses to re-create an id whose log already - * exists on disk (the SessionId is the identity). A fresh id means each run - * is a new session. + * Why a per-run id, not a fixed `${id}-session`: once a durable persistence + * backend is loaded, a fixed id collides on the second run — the backend + * refuses to re-create an id whose log already exists on disk (the SessionId + * is the identity). A fresh id means each run is a new session. * * TODO(demo): each run starting a brand-new session is fine for demos but is * NOT real conversation continuity. A production config-driven agent needs a @@ -80,14 +87,92 @@ export class AgentLoop extends Service { * fresh; the child is returned as a regular Agent handle. */ create(id: string, options: AgentOptions = {}): LoopAgent { + this.assertAgentIdFree(id) const session = this.ctx.sessions.create(`${id}-session-${randomUUID()}`, { meta: {} }) - const agent = new LoopAgent(this.ctx, AgentId(id), options, session) + return this.start(AgentId(id), options, session) + } + + /** + * Programmatic factory create ({@link AgentFactory}): an agent on a + * caller-supplied `sessionId` (NOT `${id}-session`), with optional session + * metadata (validated `cwd`, lineage). The ACP bridge uses this so the + * client-generated session id becomes the live/persisted session id. + */ + createAgent(options: CreateAgentOptions): Agent { + // Check the agent id BEFORE creating the session: register() would reject a + // duplicate id only AFTER sessions.create(), leaving an orphaned live + // session (and lazy persistence state) that blocks reuse of that id. + this.assertAgentIdFree(options.agentId) + const session = this.ctx.sessions.create(options.sessionId, { meta: options.meta ?? {} }) + return this.start(AgentId(options.agentId), options.agentOptions ?? {}, session) + } + + /** + * Resume an agent on a persisted session ({@link AgentFactory}). Loads the + * session log + metadata via `ctx.sessionPersistence`, reconstructs the live + * session with the loaded events (so `lastTurnNumber`/`deriveMessages` + * continue), and starts a fresh agent on it. The live session id is the + * resumed id, NOT `${agentId}-session`. + * + * Requires `ctx.sessionPersistence`; throws a typed error if it is not + * configured. NOT hard-injected (that would make non-persistent demos pend + * forever) — callers that need resume (ACP) inject `sessionPersistence`, so + * 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 + // AgentLoop deliberately does NOT inject it (that would pend non-persistent + // demos forever). So the runtime value can be undefined; the type cannot. + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + if (persistence === undefined) { + throw new Error('cannot resume: session persistence is not configured (load a dsh-session-persistence backend)') + } + 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 + // same id). Re-checking immediately before sessions.create() keeps the + // "no orphaned session on a duplicate id" guarantee under concurrency. + this.assertAgentIdFree(options.agentId) + // Reconstruct the live session with the FULL persisted header (createdAt, + // cwd, lineage) so resume preserves identity, not just the cwd. The seed + // events make lastTurnNumber/deriveMessages continue; the backend already + // has state (cursor) from the load above, so onCreated is a no-op and the + // seed is not re-persisted. + const session = this.ctx.sessions.create(options.resumeSessionId, { + seed: events, + meta: { + createdAt: meta.createdAt, + ...meta.cwd !== undefined ? { cwd: meta.cwd } : {}, + ...meta.parentSession !== undefined ? { parentSession: meta.parentSession } : {}, + }, + }) + return this.start(AgentId(options.agentId), options.agentOptions ?? {}, session) + } + + /** + * Reject a duplicate agent id BEFORE any session is created, so a failed + * factory call never leaves an orphaned live session (and lazy persistence + * state) behind. `register()` enforces the same uniqueness, but only after + * `sessions.create()` has already run. + */ + private assertAgentIdFree(id: string): void { + if (this.ctx.agents.get(id) !== undefined) { + throw new Error(`agent "${id}" is already registered`) + } + } + + /** Shared: construct a LoopAgent, register it, and start its loop (LIFO). */ + private start(id: AgentId, options: AgentOptions, session: Session): LoopAgent { + const agent = new LoopAgent(this.ctx, id, options, session) // Generator effect: stop and unregister are independent disposables // (LIFO), so a throwing stop() cannot leak the registry entry. this.ctx.effect(function* (this: AgentLoop) { yield this.ctx.agents.register(agent) yield agent.start() - }.bind(this), 'agentLoop.create()') + }.bind(this), 'agentLoop.start()') return agent } } diff --git a/packages/agent-loop/tests/resume.spec.ts b/packages/agent-loop/tests/resume.spec.ts new file mode 100644 index 0000000000..99df7cfa31 --- /dev/null +++ b/packages/agent-loop/tests/resume.spec.ts @@ -0,0 +1,241 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { Context } from 'cordis' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import LlmService from '@deepseek-ai/dsh-llm' +import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session' +import type { SessionEvent } from '@deepseek-ai/dsh-session' +import SystemPrompt from '@deepseek-ai/dsh-system-prompt' +import ToolRegistry from '@deepseek-ai/dsh-tools' +import AgentRegistry from '@deepseek-ai/dsh-agent' +import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl' +import AgentLoop, { LoopAgent } from '@deepseek-ai/dsh-agent-loop' +import { MockAdapter, textResponse } from './mock-adapter.ts' + +const dirs: string[] = [] +afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) }) + +async function persistentHarness(adapter: MockAdapter): Promise<{ ctx: Context; root: string }> { + const root = await mkdtemp(join(tmpdir(), 'dsh-resume-')) + 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: [] }) + await ctx.plugin(SessionPersistenceJsonl, { root }) + ctx.llm.registerAdapter(['mock'], adapter) + return { ctx, root } +} + +function waitForIdle(ctx: Context, agent: LoopAgent): Promise { + return new Promise((resolve) => { + const dispose = ctx.on('agent/status', (subject, status) => { + if (subject === agent && status === 'idle') { dispose(); resolve() } + }) + }) +} + +describe('RFC 009: AgentLoop factory create/resume', () => { + it('createAgent uses the caller-supplied sessionId (not ${id}-session)', async () => { + const adapter = new MockAdapter([textResponse('hi')]) + const { ctx } = await persistentHarness(adapter) + const agent = ctx.agents.create({ agentId: 'a1', sessionId: 'custom-session', meta: { cwd: '/w' } }) + expect(agent.session.id).toBe('custom-session') + expect(agent.session.header.cwd).toBe('/w') + await ctx.fiber.dispose() + }) + + it('createAgent rejects a duplicate agent id BEFORE creating the session (no orphan)', async () => { + const adapter = new MockAdapter([textResponse('hi')]) + const { ctx } = await persistentHarness(adapter) + ctx.agents.create({ agentId: 'dup', sessionId: 'sess-a' }) + // A second create with the SAME agent id but a fresh session id must reject + // up front — and must NOT leave an orphaned 'sess-b' session behind. + expect(() => ctx.agents.create({ agentId: 'dup', sessionId: 'sess-b' })).toThrow(/already registered/) + expect(ctx.sessions.get('sess-b')).toBeUndefined() + await ctx.fiber.dispose() + }) + + it('createAgent works without meta (no cwd)', async () => { + const adapter = new MockAdapter([textResponse('hi')]) + const { ctx } = await persistentHarness(adapter) + const agent = ctx.agents.create({ agentId: 'a-nometa', sessionId: 'nometa-session' }) + expect(agent.session.id).toBe('nometa-session') + expect(agent.session.header.cwd).toBeUndefined() + await ctx.fiber.dispose() + }) + + it('resume of a session with no cwd carries an undefined cwd header', async () => { + // Lifecycle 1: create a no-cwd session and run a turn. + const adapter1 = new MockAdapter([textResponse('a')]) + const { ctx: ctx1, root } = await persistentHarness(adapter1) + const a1 = ctx1.agents.create({ agentId: 'm', sessionId: 'nocwd-sess' }) as LoopAgent + a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } }) + await waitForIdle(ctx1, a1) + await ctx1.fiber.dispose() + + // Lifecycle 2: resume it; the header cwd stays undefined (no-cwd branch). + const adapter2 = new MockAdapter([textResponse('b')]) + 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: [] }) + await ctx2.plugin(SessionPersistenceJsonl, { root }) + ctx2.llm.registerAdapter(['mock'], adapter2) + const a2 = await ctx2.agents.resume({ agentId: 'm', resumeSessionId: 'nocwd-sess' }) as LoopAgent + expect(a2.session.header.cwd).toBeUndefined() + await ctx2.fiber.dispose() + }) + + it('resume of a forked session preserves the parentSession lineage in the header', async () => { + // Lifecycle 1: persist a FORKED session (carries parentSession in its + // header) by creating it with a complete-turn seed — the write path + // materializes the fork (header + seed) on disk. + const seed: SessionEvent[] = [ + { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }, + ] + const adapter1 = new MockAdapter([textResponse('a')]) + const { ctx: ctx1, root } = await persistentHarness(adapter1) + const forked = ctx1.sessions.create('forked-sess', { seed, meta: { cwd: '/w', parentSession: SessionId('parent-sess') } }) + await ctx1.parallel('session/flush', forked) + await ctx1.fiber.dispose() + + // Lifecycle 2: resume it; the parentSession header survives the round-trip + // (exercises resume's parentSession-present branch). + const adapter2 = new MockAdapter([textResponse('b')]) + 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: [] }) + await ctx2.plugin(SessionPersistenceJsonl, { root }) + ctx2.llm.registerAdapter(['mock'], adapter2) + const a2 = await ctx2.agents.resume({ agentId: 'm', resumeSessionId: 'forked-sess' }) as LoopAgent + expect(a2.session.header.parentSession).toBe('parent-sess') + expect(a2.session.header.cwd).toBe('/w') + await ctx2.fiber.dispose() + }) + + it('an idle inject() is flushed durably on its own (survives without explicit flush/dispose)', async () => { + // Lifecycle 1: run a turn, then inject context while idle. The idle inject + // wraps its context/message in a one-shot turn AND checkpoints it (ADR 0017) + // — without an explicit flush or clean dispose, the notice must still reach + // disk, since a crash before the next turn would otherwise lose it. + const adapter1 = new MockAdapter([textResponse('answer')]) + const { ctx: ctx1, root } = await persistentHarness(adapter1) + const a1 = ctx1.agents.create({ agentId: 'm', sessionId: 'inject-sess', meta: { cwd: '/w' } }) as LoopAgent + a1.send([{ 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' } }) + // Let inject()'s fire-and-forget flush settle (NO explicit flush/dispose). + await new Promise(r => setTimeout(r, 30)) + + // 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 (turn-enclosed, not dropped as crash tail)', async () => { + // Lifecycle 1: run a turn, then inject context while idle. The idle inject + // wraps its context/message in a one-shot turn so it is turn-enclosed — + // otherwise scanLog would treat the trailing context as a crash tail and + // drop it on reload (the bug this guards). + const adapter1 = new MockAdapter([textResponse('answer')]) + const { ctx: ctx1, root } = await persistentHarness(adapter1) + const a1 = ctx1.agents.create({ agentId: 'm', sessionId: 'inject-sess', meta: { cwd: '/w' } }) as LoopAgent + a1.send([{ 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 ctx1.parallel('session/flush', a1.session) + await ctx1.fiber.dispose() + + // Lifecycle 2: resume; the injected context is still in the derived history. + const adapter2 = new MockAdapter([textResponse('next')]) + 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: [] }) + await ctx2.plugin(SessionPersistenceJsonl, { root }) + ctx2.llm.registerAdapter(['mock'], adapter2) + const a2 = await ctx2.agents.resume({ agentId: 'm', resumeSessionId: 'inject-sess' }) as LoopAgent + const flat = JSON.stringify(a2.session.deriveMessages()) + expect(flat).toContain('background task 42 finished') + await ctx2.fiber.dispose() + }) + + it('resume reloads a persisted session: history + turn numbering continue, no duplicate seqs', async () => { + // Lifecycle 1: run one full turn, persisting it. + const adapter1 = new MockAdapter([textResponse('first answer')]) + const { ctx: ctx1, root } = await persistentHarness(adapter1) + const a1 = ctx1.agents.create({ agentId: 'main', sessionId: 'sess-resume', meta: { cwd: '/w' } }) as LoopAgent + a1.send([{ type: 'text', text: 'first question' }], { source: { kind: 'user' } }) + await waitForIdle(ctx1, a1) + const events1 = [...a1.session.events] + const seqs1 = events1.map(e => e.seq) + expect(seqs1).toEqual([...seqs1].sort((x, y) => x - y)) // contiguous + await ctx1.fiber.dispose() + + // Lifecycle 2: a brand-new context over the SAME root; resume the session. + const adapter2 = new MockAdapter([textResponse('second answer')]) + 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: [] }) + await ctx2.plugin(SessionPersistenceJsonl, { root }) + ctx2.llm.registerAdapter(['mock'], adapter2) + + const a2 = await ctx2.agents.resume({ agentId: 'main', resumeSessionId: 'sess-resume' }) as LoopAgent + // The resumed session carries the prior history… + expect(a2.session.id).toBe('sess-resume') + expect(a2.session.events.length).toBe(events1.length) + const replay = new Session(SessionId('replay'), events1) + expect(a2.session.deriveMessages()).toEqual(replay.deriveMessages()) + + // …and a new turn continues numbering (turn 2) with contiguous seqs. + a2.send([{ type: 'text', text: 'second question' }], { source: { kind: 'user' } }) + await waitForIdle(ctx2, a2) + const allSeqs = a2.session.events.map(e => e.seq) + expect(allSeqs).toEqual(allSeqs.map((_, i) => i)) // 0..N contiguous, no duplicates + const turnStarts = a2.session.events.filter(e => e.type === 'turn/start') + expect(turnStarts.map(e => e.type === 'turn/start' && e.data.turn)).toEqual([1, 2]) + await ctx2.fiber.dispose() + }) + + it('resume rejects when session persistence is not configured', async () => { + // A harness WITHOUT the persistence plugin. + const adapter = new MockAdapter([textResponse('x')]) + 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: [] }) + ctx.llm.registerAdapter(['mock'], adapter) + await expect(ctx.agents.resume({ agentId: 'm', resumeSessionId: 'nope' })) + .rejects.toThrow(/session persistence is not configured/) + await ctx.fiber.dispose() + }) +}) diff --git a/packages/agent-loop/tsconfig.json b/packages/agent-loop/tsconfig.json index 4fdd8dc398..6751664d5c 100644 --- a/packages/agent-loop/tsconfig.json +++ b/packages/agent-loop/tsconfig.json @@ -11,6 +11,7 @@ { "path": "../../vendor/schemastery" }, { "path": "../llm" }, { "path": "../session" }, + { "path": "../session-persistence" }, { "path": "../system-prompt" }, { "path": "../tools" }, { "path": "../agent" } diff --git a/packages/agent/README.md b/packages/agent/README.md index 57b1b87fc4..5bffc4681b 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -8,10 +8,18 @@ Tracks live agents so UI, hook, and orchestrator plugins can find them without i ### Public API -- `ctx.agents.register(agent: Agent): () => void` Register a live agent. Disposed with the calling fiber. +- `ctx.agents.register(agent: Agent): () => void` — record an **already-constructed** agent. Disposed with the calling fiber. - `ctx.agents.get(id: string): Agent | undefined` - `ctx.agents.list(): Agent[]` +#### Factory seam (creation) + +Agent *creation* is provided by whichever plugin implements `AgentFactory` (phase 1: `dsh-agent-loop`), registered via `setFactory`. This keeps creation on the `dsh-agent` interface so consumers (UI, the ACP bridge) program against `ctx.agents` without depending on the concrete loop package. + +- `ctx.agents.setFactory(factory: AgentFactory): () => void` — register the creation factory (the loop calls this on construction). Throws on a second factory; the slot clears on dispose. +- `ctx.agents.create(options: CreateAgentOptions): Agent` — construct, start, AND register a new agent on a caller-supplied `sessionId` (with optional `meta.cwd`). Distinct from `register` (which only records). Throws if no factory is registered. +- `ctx.agents.resume(options: ResumeAgentOptions): Promise` — load a persisted session (RFC 009) and resume an agent on it. Async; rejects if no factory is registered, or if the factory finds session persistence unconfigured. + ### Events The full `agent/*` event taxonomy is declared via declaration merging in `dsh-agent` (not `dsh-agent-loop`), so plugins depend only on this package. @@ -45,7 +53,7 @@ The handle every plugin programs against: - `agent.send(content, options?)` — queue a message; starts a turn when idle - `agent.steer(content, options?)` — steer a running turn (inject between steps); behaves like `send` when idle -- `agent.inject(content, options?)` — inject in-session context (context/message event); next request sees it. While running it joins the open turn; while idle it is wrapped in a one-shot `injection` turn so every event stays turn-enclosed (ADR 0017) +- `agent.inject(content, options?)` — inject in-session context without triggering a turn (context/message event); next request sees it - `agent.abort(reason?)` — abort the in-flight step - `agent.session`, `agent.status`, `agent.options`, `agent.id` diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 3118e6f9e7..5d43f87c25 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -6,7 +6,8 @@ */ import { Context, Service } from 'cordis' -import type { Agent } from './types.ts' +import type { SessionId } from '@deepseek-ai/dsh-session' +import type { Agent, AgentOptions } from './types.ts' export * from './types.ts' @@ -16,19 +17,110 @@ declare module 'cordis' { } } +/** + * Options for programmatically creating an agent through the registry factory + * ({@link AgentRegistry.create}). The caller supplies the live `sessionId` + * (e.g. an ACP-generated id) and optional session metadata (the validated + * `cwd`, fork lineage); the factory creates the session, the agent, and wires + * them together. + */ +export interface CreateAgentOptions { + /** The agent's id (the registry handle). */ + agentId: string + /** The live session's id (NOT derived from agentId). */ + sessionId: string + /** + * Session creation metadata: validated absolute `cwd` and `parentSession` + * fork lineage. Mirrors the `cwd`/`parentSession` fields of + * {@link CreateSessionOptions.meta} in dsh-session (the internal-only + * `createdAt`, used when reconstructing a persisted session, is deliberately + * excluded — a factory caller never sets it). + */ + meta?: { cwd?: string; parentSession?: SessionId } + /** Per-agent options (model, system prompt). */ + agentOptions?: AgentOptions +} + +/** + * Options for resuming an agent on a persisted session + * ({@link AgentRegistry.resume}). + */ +export interface ResumeAgentOptions { + /** The agent's id (the registry handle). */ + agentId: string + /** The persisted session id to load and resume on. */ + resumeSessionId: string + /** Per-agent options (model, system prompt). */ + agentOptions?: AgentOptions +} + +/** + * The agent-creation factory the loop implementation provides to the registry + * via {@link AgentRegistry.setFactory}. Kept on the `dsh-agent` interface so + * consumers (e.g. the ACP bridge) program against `ctx.agents` without + * depending on the concrete `dsh-agent-loop` package. + */ +export interface AgentFactory { + /** Create, start, and register a new agent on a caller-supplied session id. */ + createAgent(options: CreateAgentOptions): Agent + /** + * Load a persisted session and resume an agent on it. Async because it awaits + * `ctx.sessionPersistence.load`; must be called after that service exists + * (consumers inject `sessionPersistence`). + */ + resume(options: ResumeAgentOptions): Promise +} + /** * Agent registry (`ctx.agents`): tracks live agents so UI, hook, and * orchestrator plugins can find them without depending on the concrete loop - * package. Agent *creation* belongs to whichever plugin implements the Agent - * interface (phase 1: `@deepseek-ai/dsh-agent-loop`). + * package. Agent *creation* is provided by whichever plugin implements the + * {@link AgentFactory} (phase 1: `@deepseek-ai/dsh-agent-loop`), registered via + * {@link setFactory}. */ export class AgentRegistry extends Service { private store = new Map() + private factory: AgentFactory | undefined constructor(ctx: Context) { super(ctx, 'agents') } + /** + * Register the agent-creation factory (the loop calls this on construction, + * effect-scoped). Throws if a factory is already registered. Returns the + * disposer; on dispose the factory slot is cleared. + */ + setFactory(factory: AgentFactory): () => void { + const dispose = this.ctx.effect(() => { + if (this.factory !== undefined) throw new Error('an agent factory is already registered') + this.factory = factory + return () => { this.factory = undefined } + }, 'agents.setFactory()') + return () => void dispose() + } + + /** + * Create, start, and register a new agent through the registered factory. + * Distinct from {@link register} (which records an already-constructed + * agent): this constructs the agent and its session. Throws if no factory is + * registered. + */ + create(options: CreateAgentOptions): Agent { + if (this.factory === undefined) throw new Error('no agent factory registered (load an agent-loop plugin)') + return this.factory.createAgent(options) + } + + /** + * Load a persisted session and resume an agent on it through the registered + * factory. Rejects if no factory is registered; the factory rejects if + * session persistence is not configured. + */ + async resume(options: ResumeAgentOptions): Promise { + if (this.factory === undefined) throw new Error('no agent factory registered (load an agent-loop plugin)') + return this.factory.resume(options) + } + /** * Register a live agent. Throws if an agent with the same id is already * registered. Emits `agent/created` on registration and `agent/disposed` diff --git a/packages/agent/tests/agent.spec.ts b/packages/agent/tests/agent.spec.ts index d83c00a006..e72373e072 100644 --- a/packages/agent/tests/agent.spec.ts +++ b/packages/agent/tests/agent.spec.ts @@ -74,3 +74,58 @@ describe('AgentRegistry', () => { expect(ctx.agents.get('main')).toBeUndefined() }) }) + +describe('AgentRegistry factory seam', () => { + /** A stub AgentFactory that records calls and returns a stub agent. */ + function stubFactory() { + const calls: { create: unknown[]; resume: unknown[] } = { create: [], resume: [] } + const factory: import('@deepseek-ai/dsh-agent').AgentFactory = { + createAgent(options) { calls.create.push(options); return stubAgent(options.agentId) }, + resume(options) { calls.resume.push(options); return Promise.resolve(stubAgent(options.agentId)) }, + } + return { factory, calls } + } + + it('create()/resume() throw when no factory is registered', async () => { + const ctx = new Context() + await ctx.plugin(AgentRegistry) + expect(() => ctx.agents.create({ agentId: 'a', sessionId: 's' })).toThrow(/no agent factory/) + await expect(ctx.agents.resume({ agentId: 'a', resumeSessionId: 's' })).rejects.toThrow(/no agent factory/) + }) + + it('setFactory registers a factory; create/resume delegate to it', async () => { + const ctx = new Context() + await ctx.plugin(AgentRegistry) + const { factory, calls } = stubFactory() + ctx.agents.setFactory(factory) + + const created = ctx.agents.create({ agentId: 'c1', sessionId: 'sess-1', meta: { cwd: '/w' } }) + expect(created.id).toBe('c1') + expect(calls.create).toEqual([{ agentId: 'c1', sessionId: 'sess-1', meta: { cwd: '/w' } }]) + + const resumed = await ctx.agents.resume({ agentId: 'r1', resumeSessionId: 'old-sess' }) + expect(resumed.id).toBe('r1') + expect(calls.resume).toEqual([{ agentId: 'r1', resumeSessionId: 'old-sess' }]) + }) + + it('setFactory rejects a second factory', async () => { + const ctx = new Context() + await ctx.plugin(AgentRegistry) + ctx.agents.setFactory(stubFactory().factory) + expect(() => ctx.agents.setFactory(stubFactory().factory)).toThrow(/already registered/) + }) + + it('disposing the setFactory fiber clears the factory (HMR safety)', async () => { + const ctx = new Context() + await ctx.plugin(AgentRegistry) + let dispose!: () => void + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + dispose = inner.agents.setFactory(stubFactory().factory) + }, { inject: ['agents'] })) + expect(() => ctx.agents.create({ agentId: 'a', sessionId: 's' })).not.toThrow() + void dispose + await fiber.dispose() + // factory slot cleared → create throws again + expect(() => ctx.agents.create({ agentId: 'a2', sessionId: 's2' })).toThrow(/no agent factory/) + }) +}) diff --git a/yarn.lock b/yarn.lock index dbc8bbb5e8..8b2339874e 100644 --- a/yarn.lock +++ b/yarn.lock @@ -561,6 +561,7 @@ __metadata: "@deepseek-ai/dsh-invariants": "npm:^0.0.1" "@deepseek-ai/dsh-llm": "npm:^0.0.1" "@deepseek-ai/dsh-session": "npm:^0.0.1" + "@deepseek-ai/dsh-session-persistence": "npm:^0.0.1" "@deepseek-ai/dsh-session-persistence-jsonl": "npm:^0.0.1" "@deepseek-ai/dsh-system-prompt": "npm:^0.0.1" "@deepseek-ai/dsh-tools": "npm:^0.0.1" @@ -570,6 +571,7 @@ __metadata: "@deepseek-ai/dsh-agent": ^0.0.1 "@deepseek-ai/dsh-llm": ^0.0.1 "@deepseek-ai/dsh-session": ^0.0.1 + "@deepseek-ai/dsh-session-persistence": ^0.0.1 "@deepseek-ai/dsh-system-prompt": ^0.0.1 "@deepseek-ai/dsh-tools": ^0.0.1 cordis: ^4.0.0-rc.6 From 5284ed4806c30cb4661b7242905075a372b7b4e0 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 15 Jun 2026 23:55:40 +0800 Subject: [PATCH 2/3] docs(agent): sync inject wording + resume error wording with code (review #34) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - agent/README: the inject() line said "without triggering a turn", regressing the #32 turn-enclosure model. Restored the running-vs-idle wording (idle inject wraps a one-shot injection turn; ADR 0017) to match the interface JSDoc. - agent-loop resume() JSDoc said "throws a typed error" but the code throws a plain Error (consistent with the sibling assertAgentIdFree throw). Softened to "rejects with a clear error" — no behavior change; plain Error is intentional (no consumer needs a structured code here). --- packages/agent-loop/src/index.ts | 2 +- packages/agent/README.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/agent-loop/src/index.ts b/packages/agent-loop/src/index.ts index 355c610989..a1efc78d56 100644 --- a/packages/agent-loop/src/index.ts +++ b/packages/agent-loop/src/index.ts @@ -114,7 +114,7 @@ export class AgentLoop extends Service implements AgentFactory { * continue), and starts a fresh agent on it. The live session id is the * resumed id, NOT `${agentId}-session`. * - * Requires `ctx.sessionPersistence`; throws a typed error if it is not + * Requires `ctx.sessionPersistence`; rejects with a clear error if it is not * configured. NOT hard-injected (that would make non-persistent demos pend * forever) — callers that need resume (ACP) inject `sessionPersistence`, so * by the time this runs the service exists. diff --git a/packages/agent/README.md b/packages/agent/README.md index 5bffc4681b..73914ea05b 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -53,7 +53,7 @@ The handle every plugin programs against: - `agent.send(content, options?)` — queue a message; starts a turn when idle - `agent.steer(content, options?)` — steer a running turn (inject between steps); behaves like `send` when idle -- `agent.inject(content, options?)` — inject in-session context without triggering a turn (context/message event); next request sees it +- `agent.inject(content, options?)` — inject in-session context (context/message event); the next request sees it. Does not run the model. While a turn is open it joins that turn; while idle it is wrapped in a one-shot `injection` turn so every event stays turn-enclosed (ADR 0017) - `agent.abort(reason?)` — abort the in-flight step - `agent.session`, `agent.status`, `agent.options`, `agent.id` From 0000cdb2c2584ffdf13ca91e947ec0fa737631db Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 16 Jun 2026 22:28:01 +0800 Subject: [PATCH 3/3] 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-`. 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. --- examples/coding-agent/README.md | 11 +++ examples/coding-agent/cordis.yml | 3 + examples/coding-agent/tests/harness.ts | 7 +- examples/coding-agent/tests/resume.e2e.ts | 69 +++++++++++++++++ packages/agent-loop/src/index.ts | 55 ++++++++++++-- .../tests/config-session-id.spec.ts | 74 ++++++++++++++++++- 6 files changed, 212 insertions(+), 7 deletions(-) create mode 100644 examples/coding-agent/tests/resume.e2e.ts 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() + }) })