feat(agent): create/resume factory seam
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.
This commit is contained in:
@@ -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-<uuid>`, 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-<uuid>` (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
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<Config> = 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-<uuid>`, 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-<uuid>`, 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<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
|
||||
// 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
|
||||
}
|
||||
}
|
||||
|
||||
241
packages/agent-loop/tests/resume.spec.ts
Normal file
241
packages/agent-loop/tests/resume.spec.ts
Normal file
@@ -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<void> {
|
||||
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()
|
||||
})
|
||||
})
|
||||
@@ -11,6 +11,7 @@
|
||||
{ "path": "../../vendor/schemastery" },
|
||||
{ "path": "../llm" },
|
||||
{ "path": "../session" },
|
||||
{ "path": "../session-persistence" },
|
||||
{ "path": "../system-prompt" },
|
||||
{ "path": "../tools" },
|
||||
{ "path": "../agent" }
|
||||
|
||||
@@ -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<Agent>` — 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`
|
||||
|
||||
|
||||
@@ -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>
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<string, Agent>()
|
||||
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<Agent> {
|
||||
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`
|
||||
|
||||
@@ -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/)
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user