refactor(agent): trim obsolete loop surfaces
This commit is contained in:
@@ -8,11 +8,13 @@
|
||||
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { Context } from 'cordis'
|
||||
import { Agent, AgentMessageId, agentCarrier, agentInterruptReasonOf, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent'
|
||||
import { AgentMessageId, agentCarrier, agentInterruptReasonOf, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent'
|
||||
import { createScope } from '@deepseek-ai/dsh-scope'
|
||||
import type { Scope } from '@deepseek-ai/dsh-scope'
|
||||
import type {
|
||||
AgentMessage,
|
||||
Agent,
|
||||
AliasSendOptions,
|
||||
CancelOptions,
|
||||
AgentInterruptReason,
|
||||
AgentOptions,
|
||||
@@ -43,7 +45,7 @@ type StepOutcome =
|
||||
* The concrete {@link Agent}: each `run()` owns one turn and repeats model
|
||||
* steps while tools or steering require another request.
|
||||
*/
|
||||
export class ReactLoopAgent extends Agent {
|
||||
export class ReactLoopAgent implements Agent {
|
||||
/** Prompts awaiting individual turns. */
|
||||
private queued: { message: AgentMessage; wakeup: boolean }[] = []
|
||||
/** Input taken into the session log at step boundaries. */
|
||||
@@ -75,7 +77,6 @@ export class ReactLoopAgent extends Agent {
|
||||
public readonly options: AgentOptions,
|
||||
public readonly session: Session,
|
||||
) {
|
||||
super()
|
||||
this.lastTurn = session.events.findLast(event => event.type === 'turn/start')?.data.turn ?? 0
|
||||
this.scope = createScope(loopCtx, this)
|
||||
this.ctx = this.scope.ctx.extend({ agent: this })
|
||||
@@ -118,6 +119,33 @@ export class ReactLoopAgent extends Agent {
|
||||
return id
|
||||
}
|
||||
|
||||
/** Queue one ordinary prompt turn and wake the driver. */
|
||||
followup(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId {
|
||||
return this.send(content, {
|
||||
target: 'next-turn',
|
||||
wakeup: true,
|
||||
source: options?.source ?? { kind: 'user' },
|
||||
})
|
||||
}
|
||||
|
||||
/** Steer the open turn, falling back to a waking prompt while idle. */
|
||||
steer(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId {
|
||||
return this.send(content, {
|
||||
target: 'next-step',
|
||||
wakeup: true,
|
||||
source: options?.source ?? { kind: 'user' },
|
||||
})
|
||||
}
|
||||
|
||||
/** Append model-facing context without waking the driver. */
|
||||
inject(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId {
|
||||
return this.send(content, {
|
||||
target: 'next-step',
|
||||
wakeup: false,
|
||||
source: options?.source ?? { kind: 'plugin', plugin: '' },
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all pending work and abort the active turn; the first cause wins.
|
||||
* The cause is signal payload for observers and the durable turn/end
|
||||
|
||||
@@ -140,13 +140,9 @@ describe('config-driven session id', () => {
|
||||
first.inject([{ type: 'text', text: 'persist before replacement' }], {
|
||||
source: { kind: 'plugin', plugin: 'test' },
|
||||
})
|
||||
await expect.poll(async () => {
|
||||
try {
|
||||
return JSON.stringify((await ctx.sessionPersistence.inspect(sessionId)).events)
|
||||
} catch {
|
||||
return ''
|
||||
}
|
||||
}).toContain('persist before replacement')
|
||||
await ctx.sessions.flush(first.session)
|
||||
expect(JSON.stringify((await ctx.sessionPersistence.inspect(sessionId)).events))
|
||||
.toContain('persist before replacement')
|
||||
|
||||
const firstDisposal = firstLoop.dispose()
|
||||
await cleanupStarted.promise
|
||||
@@ -190,13 +186,9 @@ describe('config-driven session id', () => {
|
||||
first.inject([{ type: 'text', text: 'persist before cancellation' }], {
|
||||
source: { kind: 'plugin', plugin: 'test' },
|
||||
})
|
||||
await expect.poll(async () => {
|
||||
try {
|
||||
return JSON.stringify((await ctx.sessionPersistence.inspect(sessionId)).events)
|
||||
} catch {
|
||||
return ''
|
||||
}
|
||||
}).toContain('persist before cancellation')
|
||||
await ctx.sessions.flush(first.session)
|
||||
expect(JSON.stringify((await ctx.sessionPersistence.inspect(sessionId)).events))
|
||||
.toContain('persist before cancellation')
|
||||
|
||||
const firstDisposal = firstLoop.dispose()
|
||||
await cleanupStarted.promise
|
||||
|
||||
@@ -436,7 +436,7 @@ describe('disposal leaves the two-state status contract balanced', () => {
|
||||
await fiber.dispose()
|
||||
await driverDone(agent) // must not hang
|
||||
|
||||
await expect.poll(() => ctx.agents.get(SessionId('scoped')) === undefined).toBe(true)
|
||||
expect(ctx.agents.get(SessionId('scoped'))).toBeUndefined()
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -161,7 +161,6 @@ describe('agent/prompt-submit', () => {
|
||||
expect(log.some(e => e.type === 'turn/end')).toBe(false)
|
||||
expect(log.some(e => e.type === 'user/message')).toBe(false)
|
||||
expect(log.some(e => e.type === 'step/start')).toBe(false)
|
||||
expect(log.some(e => e.type === 'prompt/blocked')).toBe(false)
|
||||
expect(reasons).toEqual([])
|
||||
})
|
||||
|
||||
@@ -189,7 +188,6 @@ describe('agent/prompt-submit', () => {
|
||||
expect(userMsgs).toHaveLength(1)
|
||||
expect(userMsgs[0]?.type === 'user/message' && userMsgs[0].data.content).toEqual([{ type: 'text', text: 'safe' }])
|
||||
expect(adapter.requests.length).toBeGreaterThanOrEqual(1)
|
||||
expect(log.filter(e => e.type === 'prompt/blocked')).toHaveLength(0)
|
||||
expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1)
|
||||
expect(reasons).toEqual([{ kind: 'completed' }])
|
||||
})
|
||||
|
||||
@@ -1041,7 +1041,7 @@ describe('agent loop', () => {
|
||||
await fiber.dispose()
|
||||
await driverDone(agent)
|
||||
|
||||
await expect.poll(() => ctx.agents.get(SessionId('scoped')) === undefined).toBe(true)
|
||||
expect(ctx.agents.get(SessionId('scoped'))).toBeUndefined()
|
||||
})
|
||||
|
||||
it('creates agents from config on startup', async () => {
|
||||
|
||||
@@ -146,17 +146,17 @@ export type AgentCancelCause =
|
||||
export type AgentInterruptReason = AgentCancelCause | { readonly kind: 'disposed' }
|
||||
|
||||
/** Public live-agent handle with aliases over the unified delivery primitive. */
|
||||
export abstract class Agent {
|
||||
export interface Agent {
|
||||
/** The single identity shared with {@link session}. */
|
||||
abstract readonly id: SessionId
|
||||
readonly id: SessionId
|
||||
/** The provider route and model this agent's requests use. */
|
||||
abstract readonly options: AgentOptions
|
||||
readonly options: AgentOptions
|
||||
/** The live session this agent drives; its log is the durable source of truth. */
|
||||
abstract readonly session: Session
|
||||
readonly session: Session
|
||||
/** The current lifecycle state, mirrored on every `agent/status` transition. */
|
||||
abstract readonly status: AgentStatus
|
||||
readonly status: AgentStatus
|
||||
/** Agent-scoped context; its contributions are agent-local, unwind on disposal, and reject registration afterward. */
|
||||
abstract readonly ctx: Context
|
||||
readonly ctx: Context
|
||||
|
||||
/**
|
||||
* The unified delivery primitive over the (`target` × `wakeup`) matrix.
|
||||
@@ -175,7 +175,7 @@ export abstract class Agent {
|
||||
* @param options - target queue, wakeup decision, and source.
|
||||
* @returns the accepted message's {@link AgentMessageId}, stable across its `agent/inbox/*` events.
|
||||
*/
|
||||
abstract send(content: ContentBlock[], options: SendOptions): AgentMessageId
|
||||
send(content: ContentBlock[], options: SendOptions): AgentMessageId
|
||||
|
||||
/**
|
||||
* Clear queued and steering work — unless `keepInbox` — and abort the active
|
||||
@@ -186,10 +186,10 @@ export abstract class Agent {
|
||||
* @param cause - the stable caller intent carried by the current turn signal.
|
||||
* @param options - cancellation options; `keepInbox` preserves pending work.
|
||||
*/
|
||||
abstract cancel(cause: AgentCancelCause, options?: CancelOptions): void
|
||||
cancel(cause: AgentCancelCause, options?: CancelOptions): void
|
||||
|
||||
/** Resolve at idle quiescence; disposal waits for driver exit rather than only the status transition. */
|
||||
abstract whenIdle(): Promise<void>
|
||||
whenIdle(): Promise<void>
|
||||
|
||||
/**
|
||||
* Queue an ordinary follow-up turn and wake the driver — the
|
||||
@@ -199,13 +199,7 @@ export abstract class Agent {
|
||||
* @param options - message source.
|
||||
* @returns the accepted message's {@link AgentMessageId}.
|
||||
*/
|
||||
followup(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId {
|
||||
return this.send(content, {
|
||||
target: 'next-turn',
|
||||
wakeup: true,
|
||||
source: options?.source ?? { kind: 'user' },
|
||||
})
|
||||
}
|
||||
followup(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId
|
||||
|
||||
/**
|
||||
* Submit steering into the running turn — the `next-step`/wakeup preset of
|
||||
@@ -218,13 +212,7 @@ export abstract class Agent {
|
||||
* @param options - message source.
|
||||
* @returns the accepted message's {@link AgentMessageId}.
|
||||
*/
|
||||
steer(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId {
|
||||
return this.send(content, {
|
||||
target: 'next-step',
|
||||
wakeup: true,
|
||||
source: options?.source ?? { kind: 'user' },
|
||||
})
|
||||
}
|
||||
steer(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId
|
||||
|
||||
/**
|
||||
* Append model-facing context without running the model — the
|
||||
@@ -236,13 +224,7 @@ export abstract class Agent {
|
||||
* @param options - context source.
|
||||
* @returns the accepted message's {@link AgentMessageId}.
|
||||
*/
|
||||
inject(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId {
|
||||
return this.send(content, {
|
||||
target: 'next-step',
|
||||
wakeup: false,
|
||||
source: options?.source ?? { kind: 'plugin', plugin: '' },
|
||||
})
|
||||
}
|
||||
inject(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId
|
||||
|
||||
/**
|
||||
* Re-open a turn on the current session log without a new prompt — the
|
||||
@@ -251,7 +233,7 @@ export abstract class Agent {
|
||||
* immediately. Repeated calls before the scheduled retry coalesce.
|
||||
* @throws while other agent work is running.
|
||||
*/
|
||||
abstract retry(): void
|
||||
retry(): void
|
||||
}
|
||||
|
||||
declare module 'cordis' {
|
||||
|
||||
@@ -3,60 +3,38 @@ import { Context, Service, symbols } from 'cordis'
|
||||
import type { Events } from 'cordis'
|
||||
import { Session, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import AgentRegistry, {
|
||||
Agent,
|
||||
AgentMessageId,
|
||||
agentEvents,
|
||||
agentInterruptReasonOf,
|
||||
} from '@deepseek-ai/dsh-agent'
|
||||
|
||||
import type {
|
||||
Agent,
|
||||
AgentCancelCause,
|
||||
AgentFactory,
|
||||
CreateAgentOptions,
|
||||
ResumeAgentOptions,
|
||||
SendOptions,
|
||||
} from '@deepseek-ai/dsh-agent'
|
||||
|
||||
function stubAgent(rawId: string, overrides: Partial<Agent> = {}): Agent {
|
||||
const id = SessionId(rawId)
|
||||
// Agent is an abstract class, so its alias methods live on the prototype and
|
||||
// object spread would drop them; build the full literal and merge overrides.
|
||||
return Object.assign(Object.create(Agent.prototype) as Agent, {
|
||||
return {
|
||||
id,
|
||||
options: {},
|
||||
session: new Session(id),
|
||||
status: 'idle',
|
||||
ctx: new Context(),
|
||||
send: () => AgentMessageId('stub'),
|
||||
followup: () => AgentMessageId('stub'),
|
||||
steer: () => AgentMessageId('stub'),
|
||||
inject: () => AgentMessageId('stub'),
|
||||
cancel() {},
|
||||
retry() {},
|
||||
whenIdle() { return Promise.resolve() },
|
||||
...overrides,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
describe('Agent delivery aliases', () => {
|
||||
it('materializes complete SendOptions for every preset', () => {
|
||||
const calls: SendOptions[] = []
|
||||
const agent = stubAgent('aliases', {
|
||||
send(_content, options) {
|
||||
if (options !== undefined) calls.push(options)
|
||||
return AgentMessageId('stub')
|
||||
},
|
||||
})
|
||||
|
||||
agent.followup([])
|
||||
agent.steer([])
|
||||
agent.inject([])
|
||||
|
||||
expect(calls).toEqual([
|
||||
{ target: 'next-turn', wakeup: true, source: { kind: 'user' } },
|
||||
{ target: 'next-step', wakeup: true, source: { kind: 'user' } },
|
||||
{ target: 'next-step', wakeup: false, source: { kind: 'plugin', plugin: '' } },
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
describe('AgentRegistry', () => {
|
||||
it('registers exact entries, emits lifecycle events, and unregisters on owner disposal', async () => {
|
||||
const ctx = new Context()
|
||||
|
||||
@@ -8,13 +8,13 @@
|
||||
*/
|
||||
|
||||
import { callConfigEquals } from '@deepseek-ai/dsh-llm'
|
||||
import type { Message, ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
import type { ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
import type { EpochHeader, SessionEvent } from './types.ts'
|
||||
|
||||
/**
|
||||
* Normalize a header to canonical form: an empty system prompt, an empty tool
|
||||
* list, and an empty session prefix become absent fields, matching how requests
|
||||
* are built. Logging, folding, and comparison use this one representation.
|
||||
* Normalize a header to canonical form: an empty system prompt and empty tool
|
||||
* list become absent fields, matching how requests are built. Logging, folding,
|
||||
* and comparison use this one representation.
|
||||
* @param header - the header to normalize (not mutated).
|
||||
* @returns the canonical header.
|
||||
*/
|
||||
@@ -23,7 +23,6 @@ export function canonicalHeader(header: EpochHeader): EpochHeader {
|
||||
config: header.config,
|
||||
...header.system !== undefined && header.system.length > 0 ? { system: header.system } : {},
|
||||
...header.tools !== undefined && header.tools.length > 0 ? { tools: header.tools } : {},
|
||||
...header.messagePrefix !== undefined && header.messagePrefix.length > 0 ? { messagePrefix: header.messagePrefix } : {},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,21 +31,14 @@ function sameSchema(a: ToolSchema, b: ToolSchema): boolean {
|
||||
return JSON.stringify(a) === JSON.stringify(b)
|
||||
}
|
||||
|
||||
/** Canonical JSON equality over session-prefix arrays; absence equals empty. */
|
||||
function sameMessages(a: readonly Message[] | undefined, b: readonly Message[] | undefined): boolean {
|
||||
return JSON.stringify(a ?? []) === JSON.stringify(b ?? [])
|
||||
}
|
||||
|
||||
/**
|
||||
* Field-wise equality over canonical headers. Tool schemas compare in order;
|
||||
* the session prefix compares as canonical JSON.
|
||||
* Field-wise equality over canonical headers. Tool schemas compare in order.
|
||||
* @param a - one canonical header.
|
||||
* @param b - the other.
|
||||
* @returns whether config, system, tools, and session prefix all match.
|
||||
* @returns whether config, system, and tools all match.
|
||||
*/
|
||||
export function headerEquals(a: EpochHeader, b: EpochHeader): boolean {
|
||||
if (!callConfigEquals(a.config, b.config) || a.system !== b.system) return false
|
||||
if (!sameMessages(a.messagePrefix, b.messagePrefix)) return false
|
||||
const at = a.tools ?? []
|
||||
const bt = b.tools ?? []
|
||||
return at.length === bt.length && at.every((tool, i) => sameSchema(tool, bt[i] as ToolSchema))
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { Branded } from '@deepseek-ai/dsh-brand'
|
||||
import type { AssistantProvenance, CallId, ContentBlock, LlmCallConfig, LlmFailure, Message, MessageSource, StreamChunk, TokenUsage, ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
import type { AssistantProvenance, CallId, ContentBlock, LlmCallConfig, LlmFailure, MessageSource, StreamChunk, TokenUsage, ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
import type { JsonValue } from './json.ts'
|
||||
|
||||
/** Identifies one session in the store (and its persistence artifacts). */
|
||||
@@ -117,11 +117,6 @@ export interface TurnEndReasonMap {
|
||||
disposed: { kind: 'disposed' }
|
||||
/** At least one step reached its output-token ceiling, even if a plugin continued the turn. */
|
||||
'max-tokens': { kind: 'max-tokens' }
|
||||
/**
|
||||
* Policy blocked the turn's claimed prompt before the first step. The
|
||||
* zero-step turn still records a balanced durable boundary and veto reason.
|
||||
*/
|
||||
rejected: { kind: 'rejected'; reason: string }
|
||||
/**
|
||||
* A persistence backend closed a crash-orphaned turn on reload. The loop never
|
||||
* emits this marker, and the events recorded before the crash remain intact.
|
||||
@@ -151,9 +146,9 @@ export interface TodoItem {
|
||||
}
|
||||
|
||||
/**
|
||||
* Logged request state outside derived history: call config, system prompt,
|
||||
* tools, and prefix. The latest full `request/header` snapshot reconstructs it;
|
||||
* canonical empty optional fields are absent.
|
||||
* Logged request state outside derived history: call config, system prompt, and
|
||||
* tools. The latest full `request/header` snapshot reconstructs it; canonical
|
||||
* empty optional fields are absent.
|
||||
*/
|
||||
export interface EpochHeader {
|
||||
/** The conversation's call configuration (provider, model, and sampling scalars). */
|
||||
@@ -162,14 +157,6 @@ export interface EpochHeader {
|
||||
system?: string
|
||||
/** Assembled tool schemas; absent for a tool-less request. */
|
||||
tools?: ToolSchema[]
|
||||
/**
|
||||
* The session prefix: request-only messages sent BEFORE the entire derived
|
||||
* history (the `agent/session-prefix` waterfall's product, composed once
|
||||
* per loop instance and reused for every request it sends). Not session
|
||||
* history — `deriveMessages()` never returns it — so the header is its
|
||||
* only durable record; absent when the instance composed none.
|
||||
*/
|
||||
messagePrefix?: Message[]
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -225,11 +212,6 @@ export interface SessionEventMap {
|
||||
* injection may append this event between turns without running the model.
|
||||
*/
|
||||
'user/message': UserMessageData
|
||||
/**
|
||||
* Durable record of a prompt veto and its reason. It is log-only: the blocked
|
||||
* prompt never enters the model-visible surface, and its turn runs zero steps.
|
||||
*/
|
||||
'prompt/blocked': { content: ContentBlock[]; source: MessageSource; reason: string }
|
||||
/** Raw stream chunk — token-level replay fidelity. */
|
||||
'assistant/chunk': { turn: number; step: number; chunk: StreamChunk }
|
||||
/**
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Session, SessionId, canonicalHeader, foldRequestHeader, headerEquals } from '@deepseek-ai/dsh-session'
|
||||
import type { EpochHeader, SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import type { Message, ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
import type { ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
const CONFIG = { provider: 'mock', model: 'm' }
|
||||
|
||||
@@ -11,33 +11,28 @@ function tool(name: string, description = 'd'): ToolSchema {
|
||||
return { name, description, parameters: { type: 'object' } }
|
||||
}
|
||||
|
||||
function msg(text: string): Message {
|
||||
return { role: 'user', content: [{ type: 'text', text }] }
|
||||
}
|
||||
|
||||
describe('canonicalHeader', () => {
|
||||
it('normalizes empty optional fields to absence and preserves populated fields', () => {
|
||||
expect(canonicalHeader({ config: CONFIG, system: '', tools: [], messagePrefix: [] })).toEqual({ config: CONFIG })
|
||||
const full = canonicalHeader({ config: CONFIG, system: 's', tools: [tool('a')], messagePrefix: [msg('p')] })
|
||||
expect(full).toEqual({ config: CONFIG, system: 's', tools: [tool('a')], messagePrefix: [msg('p')] })
|
||||
expect(canonicalHeader({ config: CONFIG, system: '', tools: [] })).toEqual({ config: CONFIG })
|
||||
const full = canonicalHeader({ config: CONFIG, system: 's', tools: [tool('a')] })
|
||||
expect(full).toEqual({ config: CONFIG, system: 's', tools: [tool('a')] })
|
||||
})
|
||||
})
|
||||
|
||||
describe('headerEquals', () => {
|
||||
const base = canonicalHeader({ config: CONFIG, system: 's', tools: [tool('a')], messagePrefix: [msg('p')] })
|
||||
const base = canonicalHeader({ config: CONFIG, system: 's', tools: [tool('a')] })
|
||||
|
||||
it('compares every canonical field and preserves tool order', () => {
|
||||
expect(headerEquals(base, structuredClone(base))).toBe(true)
|
||||
expect(headerEquals(base, { ...base, config: { provider: 'mock', model: 'other' } })).toBe(false)
|
||||
expect(headerEquals(base, { ...base, system: 'other' })).toBe(false)
|
||||
expect(headerEquals(base, { ...base, messagePrefix: [msg('other')] })).toBe(false)
|
||||
expect(headerEquals(base, { ...base, tools: [] })).toBe(false)
|
||||
expect(headerEquals(base, { ...base, tools: [tool('a', 'changed')] })).toBe(false)
|
||||
expect(headerEquals({ config: CONFIG, tools: [tool('a'), tool('b')] }, { config: CONFIG, tools: [tool('b'), tool('a')] })).toBe(false)
|
||||
})
|
||||
|
||||
it('treats absent and empty prefix/tool arrays as equivalent canonical absence', () => {
|
||||
expect(headerEquals({ config: CONFIG }, { config: CONFIG, tools: [], messagePrefix: [] })).toBe(true)
|
||||
it('treats absent and empty tool arrays as equivalent canonical absence', () => {
|
||||
expect(headerEquals({ config: CONFIG }, { config: CONFIG, tools: [] })).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
Reference in New Issue
Block a user