diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index 13fd420065..0487f52095 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -11,7 +11,6 @@ import type { AgentStatus, CancelOptions, InboxTarget, - PreStepDecision, RequestErrorAction, } from '@deepseek-ai/dsh-agent' import { Inbox, agentCarrier, agentEvents, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent' @@ -26,10 +25,12 @@ import { } from '@deepseek-ai/dsh-llm' import type { Scope } from '@deepseek-ai/dsh-scope' import { createScope } from '@deepseek-ai/dsh-scope' -import type { EpochHeader, Session, SessionId, TurnEndReason, UserMessage } from '@deepseek-ai/dsh-session' +import type { EpochHeader, RequestContext, Session, SessionId, TurnEndReason, UserMessage } from '@deepseek-ai/dsh-session' import { canonicalHeader, headerEquals } from '@deepseek-ai/dsh-session' -import { renderPrompt } from '@deepseek-ai/dsh-system-prompt' +import { renderContextSnapshot, renderPrompt } from '@deepseek-ai/dsh-system-prompt' +import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt' import type { Context } from 'cordis' +import { RuntimeContextProjection } from './runtime-context.ts' import { executeToolCalls } from './tool-calls.ts' type Phase = @@ -39,6 +40,10 @@ type Phase = type StepEndReason = Extract +type PreparedStep = + | { kind: 'reject' } + | { kind: 'enter'; messages: UserMessage[]; assembly: PromptAssembly } + /** Remove adapter-derived values before plugins propose the next request config. */ function requestProposal(header: EpochHeader): LlmCallConfig { if (header.adapterDefaults === undefined) return header.config @@ -60,6 +65,7 @@ export class ReactLoopAgent implements Agent { /** Whether this loop instance has appended its initial/resume request anchor. */ private requestHeaderLogged = false + private readonly runtimeContext: RuntimeContextProjection constructor( private loopCtx: Context, @@ -75,6 +81,7 @@ export class ReactLoopAgent implements Agent { this.phase = { kind: 'idle', lastTurn } this.scope = createScope(loopCtx, this) this.ctx = this.scope.ctx.extend({ agent: this }) + this.runtimeContext = new RuntimeContextProjection(this.ctx, session) } get status(): AgentStatus { @@ -154,19 +161,25 @@ export class ReactLoopAgent implements Agent { } } - private async preStep(target: InboxTarget, position: { turn: number; step: number }): Promise { + private async preStep(target: InboxTarget, position: { turn: number; step: number }): Promise { if (this.phase.kind !== 'running') throw new Error(`agent "${this.id}": pre-step outside running phase`) const signal = this.phase.abort.signal const claimed = this.inbox.claim(target) for (const message of claimed) { emitAgentEvent(this.loopCtx, this, 'agent/inbox/claimed', { message, turn: position.turn }) } + const assembly = await this.loopCtx.systemPrompt.assemble(assembleContextFor(this, signal)) + signal.throwIfAborted() + const context = this.runtimeContext.project(renderContextSnapshot(assembly)) const decision = await agentEvents(this.loopCtx, this).waterfall( 'agent/pre-step', claimed, { ...position, signal }, - () => Promise.resolve({ kind: 'enter', messages: claimed }), + () => Promise.resolve({ + kind: 'enter', + messages: context === undefined ? claimed : [...claimed, context], + }), ) signal.throwIfAborted() - return decision + return decision.kind === 'reject' ? decision : { ...decision, assembly } } /** Claimed input stays unowned until `turn/start` commits. */ @@ -180,7 +193,7 @@ export class ReactLoopAgent implements Agent { const phase = { kind: 'running' as const, abort, turn: lastTurn, step: 0 } this.setPhase(phase) signal.throwIfAborted() - let decision: PreStepDecision + let decision: PreparedStep try { decision = await this.preStep('next-turn', { turn: phase.turn + 1, step: 1 }) if (decision.kind === 'reject') return false @@ -205,7 +218,7 @@ export class ReactLoopAgent implements Agent { for (const message of decision.messages) { this.session.append('user/message', message, { surfaceOp: 'append' }) } - turnEnds = await this.step() + turnEnds = await this.step(decision.assembly) } finally { this.session.append('step/end', { turn, step }) } @@ -244,17 +257,16 @@ export class ReactLoopAgent implements Agent { return this.inbox.hasPending } - private async step(): Promise { + private async step(assembly: PromptAssembly): Promise { if (this.phase.kind !== 'running') throw new Error(`agent "${this.id}": step outside running phase`) const { turn, step, abort: { signal } } = this.phase signal.throwIfAborted() - const assembly = await this.loopCtx.systemPrompt.assemble(assembleContextFor(this, signal)) - signal.throwIfAborted() const system = renderPrompt(assembly) + const boundaryMessages = this.session.deriveMessages() while (true) { const { request, preparedCall } = await this.buildRequest( - turn, step, assembly.tools, system, this.session.deriveMessages(), signal, + turn, step, assembly.tools, system, boundaryMessages, signal, ) const assembler = new BlockAssembler() const chunkSeqs: number[] = [] @@ -383,6 +395,19 @@ export class ReactLoopAgent implements Agent { } else if (baseline === undefined || !headerEquals(baseline, header)) { this.session.append('request/header', { header, reason: 'change' }) } + + const contextWindow = preparedCall?.context?.contextWindow + const requestContext: RequestContext = { + provider: config.provider, + model: config.model, + ...contextWindow === undefined ? {} : { contextWindow }, + } + const previousContext = session.requestContext() + if (previousContext?.provider !== requestContext.provider + || previousContext.model !== requestContext.model + || previousContext.contextWindow !== requestContext.contextWindow) { + session.append('request/context', requestContext) + } signal.throwIfAborted() const request = markAgentLoopRequest(deepFreeze({ diff --git a/packages/core/agent-loop/src/runtime-context.ts b/packages/core/agent-loop/src/runtime-context.ts new file mode 100644 index 0000000000..26c33a18c5 --- /dev/null +++ b/packages/core/agent-loop/src/runtime-context.ts @@ -0,0 +1,71 @@ +/** + * Durable projection state for dynamic runtime context. + * @module @deepseek-ai/dsh-agent-loop/runtime-context + */ + +import { createUserMessage } from '@deepseek-ai/dsh-llm' +import type { Session, UserMessage } from '@deepseek-ai/dsh-session' +import { isReplacementSurfaceEvent } from '@deepseek-ai/dsh-session' +import type { Context } from 'cordis' + +const SOURCE = '@deepseek-ai/dsh-system-prompt' +const CLEARED = 'Current runtime context: none. Earlier runtime-context snapshots no longer apply.' + +function isOwned(message: UserMessage): boolean { + return message.source.kind === 'plugin' && message.source.plugin === SOURCE +} + +function textOf(message: UserMessage): string | undefined { + const [block] = message.content + return message.content.length === 1 && block?.type === 'text' ? block.text : undefined +} + +/** Tracks the last retained runtime-context snapshot without owning its commit. */ +export class RuntimeContextProjection { + /** `undefined` means no snapshot ever existed; `null` means none is retained. */ + private retained: { seq: number; text: string | undefined } | null | undefined + + /** + * Restore projection state once, then follow authoritative session events. + * @param ctx - agent-scoped event context. + * @param session - session receiving projected messages. + */ + constructor(ctx: Context, session: Session) { + const surface = new Set(session.surface.nodes) + for (let index = session.events.length - 1; index >= 0; index -= 1) { + const event = session.events[index] + if (event?.type !== 'user/message' || !isOwned(event.data)) continue + this.retained ??= null + if (surface.has(event.seq)) { + this.retained = { seq: event.seq, text: textOf(event.data) } + break + } + } + + ctx.on('session/event', (subject, event) => { + if (subject !== session) return + if (event.type === 'user/message' && isOwned(event.data)) { + this.retained = { seq: event.seq, text: textOf(event.data) } + } else if (this.retained + && isReplacementSurfaceEvent(event) + && event.sourceEventSeqs?.includes(this.retained.seq) === true) { + this.retained = null + } + }) + } + + /** + * Create an uncommitted snapshot only when the retained value differs. + * @param current - fully rendered dynamic context. + * @returns a candidate user message, or `undefined` when no update is needed. + */ + project(current: string): UserMessage | undefined { + if (this.retained === undefined && current.length === 0) return + const snapshot = current.length === 0 ? CLEARED : current + if (this.retained?.text === snapshot) return + return createUserMessage({ + content: [{ type: 'text', text: snapshot }], + source: { kind: 'plugin', plugin: SOURCE }, + }) + } +} diff --git a/packages/core/agent-loop/tests/cancel.spec.ts b/packages/core/agent-loop/tests/cancel.spec.ts index da5d638d13..69f265ac5a 100644 --- a/packages/core/agent-loop/tests/cancel.spec.ts +++ b/packages/core/agent-loop/tests/cancel.spec.ts @@ -1,8 +1,8 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm' /** * Tests for the queue-aware `Agent.cancel()` primitive. The default clears - * queued and steering work, while `keepInbox` preserves pending input and - * resumes waking turns after the active turn reaches quiescence. The suite + * queued and steering work, while `keepInbox` preserves pending input for a + * later wake after the active turn reaches quiescence. The suite * covers every landing window plus signal reset and `whenIdle()` quiescence. * @module dsh-agent-loop/tests/cancel */ @@ -284,41 +284,6 @@ describe('Agent.cancel()', () => { expect(adapter.requests).toHaveLength(1) }) - it('cancel({ keepInbox: true }) aborts the active turn and drains the queued tail in FIFO order', async () => { - const adapter = new MockAdapter([ - 'hang', - textResponse('second reply'), - textResponse('third reply'), - ]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('keep-inbox-running'), { provider: 'mock', model: 'mock' }) - const reasons: TurnEndReason[] = [] - const discards: unknown[] = [] - ctx.on('session/event', (session, event) => { - if (session === agent.session && event.type === 'turn/end') reasons.push(event.data.reason) - }) - ctx.on('agent/inbox/discard', (subject, items) => { - if (subject === agent) discards.push(items) - }) - - send(agent, 'active') - await new Promise(resolve => setTimeout(resolve, 30)) - send(agent, 'queued second') - send(agent, 'queued third') - const idle = agent.whenIdle() - agent.cancel({ kind: 'user' }, { keepInbox: true }) - await idle - - expect(discards).toEqual([]) - expect(userTexts(agent)).toEqual(['active', 'queued second', 'queued third']) - expect(reasons).toEqual([ - { kind: 'aborted' }, - { kind: 'completed' }, - { kind: 'completed' }, - ]) - expect(adapter.requests).toHaveLength(3) - }) - it('cancel from an assistant/message observer skips execution but balances replay', async () => { const adapter = new MockAdapter([ toolCallResponse('c1', 'danger', {}), @@ -770,7 +735,7 @@ describe('Agent.cancel()', () => { agent.cancel({ kind: 'user' }) await idle const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end') - if (stage === 'pre-step') { + if (stage === 'pre-step' || stage === 'system-prompt') { expect(turnEnd).toBeUndefined() } else { expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'user' } }) diff --git a/packages/core/agent-loop/tests/contract-regressions.spec.ts b/packages/core/agent-loop/tests/contract-regressions.spec.ts index 28c36a24a8..d4e520d958 100644 --- a/packages/core/agent-loop/tests/contract-regressions.spec.ts +++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts @@ -1122,7 +1122,7 @@ describe('tool result call identity', () => { }) describe('disposal and cancellation during pre-step assembly', () => { - it('disposal during system-prompt assembly closes the started step as disposed', { timeout: 30000 }, async () => { + it('disposal during system-prompt assembly prevents the turn from opening', { timeout: 30000 }, async () => { // Start disposal, then release assembly. Do not await disposal first: it // waits for the blocked driver to exit. const adapter = new MockAdapter(['hang']) @@ -1154,7 +1154,7 @@ describe('disposal and cancellation during pre-step assembly', () => { ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) }) send(agent, 'go') - // Give the loop time to enter the step and reach assemble(). + // Give the loop time to reach pre-step assembly. await new Promise(r => setTimeout(r, 50)) // Release assembly before awaiting disposal because disposal joins the blocked driver. @@ -1165,18 +1165,15 @@ describe('disposal and cancellation during pre-step assembly', () => { await driverDone(agent) unlisten() - // Turn boundaries are durable rows; there is no `agent/*` mirror to assert. const e = [...agent.session.events] - expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1) - expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1) - const turnEnd = e.findLast(x => x.type === 'turn/end') - expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'disposed' } }) - expect(e.filter(x => x.type === 'step/start')).toHaveLength(1) - expect(e.filter(x => x.type === 'step/end')).toHaveLength(1) + expect(e.some(x => x.type === 'turn/start')).toBe(false) + expect(e.some(x => x.type === 'turn/end')).toBe(false) + expect(e.some(x => x.type === 'step/start')).toBe(false) + expect(e.some(x => x.type === 'step/end')).toBe(false) expect(e.some(x => x.type === 'assistant/chunk')).toBe(false) }) - it('cancel during system-prompt assembly closes the started step as aborted', { timeout: 30000 }, async () => { + it('cancel during system-prompt assembly prevents the turn from opening', { timeout: 30000 }, async () => { const adapter = new MockAdapter([textResponse('should not appear')]) let releaseAssemble!: () => void const blocker = new Promise(r => void (releaseAssemble = r)) @@ -1215,16 +1212,14 @@ describe('disposal and cancellation during pre-step assembly', () => { unlisten() const e = [...agent.session.events] - expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1) - expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1) - const turnEnd = e.findLast(x => x.type === 'turn/end') - expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'user' } }) - expect(e.filter(x => x.type === 'step/start')).toHaveLength(1) - expect(e.filter(x => x.type === 'step/end')).toHaveLength(1) + expect(e.some(x => x.type === 'turn/start')).toBe(false) + expect(e.some(x => x.type === 'turn/end')).toBe(false) + expect(e.some(x => x.type === 'step/start')).toBe(false) + expect(e.some(x => x.type === 'step/end')).toBe(false) expect(e.some(x => x.type === 'assistant/chunk')).toBe(false) expect(e.some(x => x.type === 'assistant/message')).toBe(false) expect(adapter.requests).toHaveLength(0) - expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }]) + expect(reasons).toEqual([]) }) it('disposal during pre-step prevents the turn from opening', { timeout: 15000 }, async () => { @@ -1356,14 +1351,10 @@ describe('disposal and cancellation during pre-step assembly', () => { await driverDone(agent) const e = [...agent.session.events] - expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1) - expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1) - // The critical assertions: after disposal, the turn has no assistant - // artifacts — the turn ended disposed before the model was invoked. + expect(e.some(x => x.type === 'turn/start')).toBe(false) + expect(e.some(x => x.type === 'turn/end')).toBe(false) expect(e.some(x => x.type === 'assistant/chunk')).toBe(false) expect(e.some(x => x.type === 'assistant/message')).toBe(false) expect(adapter.requests).toHaveLength(0) - // The durable turn/end reason is the authoritative turn-boundary record - // (turn boundaries have no agent/* mirror). }) }) diff --git a/packages/core/agent-loop/tests/tool-order.spec.ts b/packages/core/agent-loop/tests/tool-order.spec.ts index e8a13f6a80..6b9ac94d4b 100644 --- a/packages/core/agent-loop/tests/tool-order.spec.ts +++ b/packages/core/agent-loop/tests/tool-order.spec.ts @@ -93,8 +93,7 @@ describe('loop-level canonical tool order', () => { expect(Object.isFrozen(adapter.requests[0])).toBe(true) }) - it('fails the turn — no model request — when toolOrder names an unregistered tool', async () => { - // Unknown tool order fails before step or request creation and returns the agent to idle. + it('fails before opening a turn when toolOrder names an unregistered tool', async () => { const adapter = new MockAdapter([textResponse('never sent')]) const ctx = await harness(adapter, ['ghost', TOOL_ORDER_REST]) registerNamed(ctx, 'alpha') @@ -103,12 +102,9 @@ describe('loop-level canonical tool order', () => { await waitForIdle(ctx, agent) expect(adapter.requests).toHaveLength(0) expect(foldRequestHeader(agent.session.events)).toBeUndefined() - const end = agent.session.events.find(e => e.type === 'turn/end') - expect(end?.type === 'turn/end' && end.data.reason).toEqual({ - kind: 'error', - error: 'toolOrder lists unregistered tool "ghost"; known tools: alpha', - }) - expect(agent.session.events.filter(e => e.type === 'step/start')).toHaveLength(1) - expect(agent.session.events.filter(e => e.type === 'step/end')).toHaveLength(1) + expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false) + expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(false) + expect(agent.session.events.some(e => e.type === 'step/start')).toBe(false) + expect(agent.session.events.some(e => e.type === 'step/end')).toBe(false) }) }) diff --git a/packages/core/agent-loop/tests/turn-admission.spec.ts b/packages/core/agent-loop/tests/turn-admission.spec.ts deleted file mode 100644 index e5d4d2ebb3..0000000000 --- a/packages/core/agent-loop/tests/turn-admission.spec.ts +++ /dev/null @@ -1,286 +0,0 @@ -import { describe, expect, it } from 'vitest' -import { Context } from 'cordis' -import AgentRegistry, { type Agent, type InboxItem } from '@deepseek-ai/dsh-agent' -import AgentLoop from '@deepseek-ai/dsh-agent-loop' -import LlmService, { createUserMessage } from '@deepseek-ai/dsh-llm' -import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' -import SystemPrompt from '@deepseek-ai/dsh-system-prompt' -import ToolRegistry from '@deepseek-ai/dsh-tools' -import { MockAdapter, textResponse } from './mock-adapter.ts' - -async function harness(adapter: MockAdapter): Promise { - 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) - return ctx -} - -function prompt(agent: Agent, text: string): void { - agent.followup(createUserMessage({ - content: [{ type: 'text', text }], - source: { kind: 'user' }, - })) -} - -function itemText(item: InboxItem): string { - return item.message.content.flatMap(block => block.type === 'text' ? [block.text] : []).join('') -} - -interface InboxRecording { - readonly events: string[] - readonly enqueued: InboxItem['id'][] - readonly dequeued: InboxItem['id'][] - readonly discarded: InboxItem['id'][] -} - -/** Record the complete inbox lifecycle of one agent for order and identity assertions. */ -function recordInbox(ctx: Context): InboxRecording { - const events: string[] = [] - const enqueued: InboxItem['id'][] = [] - const dequeued: InboxItem['id'][] = [] - const discarded: InboxItem['id'][] = [] - ctx.on('agent/inbox/enqueue', (_agent, item) => { - events.push(`enqueue:${item.placement}:${itemText(item)}`) - enqueued.push(item.id) - }) - ctx.on('agent/inbox/dequeue', (_agent, item) => { - events.push(`dequeue:${itemText(item)}`) - dequeued.push(item.id) - }) - ctx.on('agent/inbox/discard', (_agent, items) => { - events.push(`discard:${items.map(itemText).join(',')}`) - discarded.push(...items.map(item => item.id)) - }) - return { events, enqueued, dequeued, discarded } -} - -/** Text of every ordinary prompt the log admitted, in durable order. */ -function promptTexts(agent: Agent): string[] { - return agent.session.events.flatMap(event => event.type === 'user/message' - ? event.data.content.flatMap(block => block.type === 'text' ? [block.text] : []) - : []) -} - -describe('idle turn admission reservation', () => { - it('holds later waking prompts in the FIFO until release', async () => { - const adapter = new MockAdapter([textResponse('first'), textResponse('second')]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - const inbox = recordInbox(ctx) - - const release = agent.reserveTurnAdmission() - expect(release).toBeDefined() - - prompt(agent, 'first prompt') - prompt(agent, 'second prompt') - expect(agent.acceptsNextStep).toBe(false) - await new Promise((resolve) => { setTimeout(resolve, 5) }) - - expect(agent.status).toBe('idle') - expect(adapter.requests).toHaveLength(0) - expect(agent.session.events).toHaveLength(0) - expect(inbox.events).toEqual([ - 'enqueue:queued:first prompt', - 'enqueue:queued:second prompt', - ]) - - release?.() - await agent.whenIdle() - - expect(promptTexts(agent)).toEqual(['first prompt', 'second prompt']) - expect(agent.session.events.flatMap(event => - event.type === 'turn/start' ? [event.data.turn] : [])).toEqual([1, 2]) - expect(inbox.events).toEqual([ - 'enqueue:queued:first prompt', - 'enqueue:queued:second prompt', - 'dequeue:first prompt', - 'dequeue:second prompt', - ]) - expect(inbox.dequeued).toEqual(inbox.enqueued) - expect(inbox.discarded).toEqual([]) - }) - - it('refuses acquisition when an accepted waking prompt still owns the next turn', async () => { - const adapter = new MockAdapter([textResponse('ok')]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - - prompt(agent, 'accepted first') - expect(agent.status).toBe('idle') - expect(agent.reserveTurnAdmission()).toBeUndefined() - - await agent.whenIdle() - expect(adapter.requests).toHaveLength(1) - }) - - it('refuses acquisition while a turn is running', async () => { - const adapter = new MockAdapter([textResponse('ok')]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - const reserved: unknown[] = [] - ctx.on('agent/step', () => { - reserved.push(agent.reserveTurnAdmission()) - }) - - prompt(agent, 'running') - await agent.whenIdle() - - expect(agent.status).toBe('idle') - expect(reserved).toEqual([undefined]) - }) - - it('refuses a second reservation and releases idempotently', async () => { - const adapter = new MockAdapter([textResponse('ok')]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - - const release = agent.reserveTurnAdmission() - expect(agent.reserveTurnAdmission()).toBeUndefined() - prompt(agent, 'queued behind the reservation') - - release?.() - release?.() - await agent.whenIdle() - - expect(promptTexts(agent)).toEqual(['queued behind the reservation']) - expect(adapter.requests).toHaveLength(1) - const second = agent.reserveTurnAdmission() - expect(second).toBeDefined() - second?.() - }) - - it('ignores a stale release once a later reservation owns the boundary', async () => { - const adapter = new MockAdapter([textResponse('ok')]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - - const stale = agent.reserveTurnAdmission() - stale?.() - const live = agent.reserveTurnAdmission() - prompt(agent, 'held by the live reservation') - stale?.() - await new Promise((resolve) => { setTimeout(resolve, 5) }) - - expect(adapter.requests).toHaveLength(0) - live?.() - await agent.whenIdle() - expect(adapter.requests).toHaveLength(1) - }) - - it('acquires beside quiet queued work and leaves it queued', async () => { - const adapter = new MockAdapter([textResponse('ok')]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - - agent.send(createUserMessage({ - content: [{ type: 'text', text: 'quiet' }], - source: { kind: 'user' }, - }), { - target: 'next-turn', - wakeup: false, - }) - const release = agent.reserveTurnAdmission() - expect(release).toBeDefined() - - release?.() - await agent.whenIdle() - expect(adapter.requests).toHaveLength(0) - }) - - it('makes whenIdle() wait for release without spinning on a settled promise', async () => { - const adapter = new MockAdapter([textResponse('ok')]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - const machine = agent as Agent & { done: Promise } - let backing = machine.done - let reads = 0 - Object.defineProperty(agent, 'done', { - configurable: true, - get(): Promise { - reads += 1 - return backing - }, - set(value: Promise) { - backing = value - }, - }) - - const release = agent.reserveTurnAdmission() - prompt(agent, 'waiting for the reservation') - let settled = false - const idle = agent.whenIdle().then(() => { settled = true }) - for (let tick = 0; tick < 5; tick += 1) { - await new Promise((resolve) => { setTimeout(resolve, 1) }) - } - - expect(settled).toBe(false) - expect(reads).toBeLessThanOrEqual(2) - - release?.() - await idle - expect(settled).toBe(true) - expect(adapter.requests).toHaveLength(1) - }) - - it('resolves whenIdle() after release with nothing queued', async () => { - const adapter = new MockAdapter([]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - - const release = agent.reserveTurnAdmission() - let settled = false - const idle = agent.whenIdle().then(() => { settled = true }) - await new Promise((resolve) => { setTimeout(resolve, 5) }) - expect(settled).toBe(false) - - release?.() - await idle - expect(agent.status).toBe('idle') - }) - - it('lets cancellation discard held prompts and keeps the boundary quiet', async () => { - const adapter = new MockAdapter([]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - const inbox = recordInbox(ctx) - - const release = agent.reserveTurnAdmission() - prompt(agent, 'discarded while held') - agent.cancel({ kind: 'user' }) - - expect(inbox.events).toEqual([ - 'enqueue:queued:discarded while held', - 'discard:discarded while held', - ]) - expect(inbox.discarded).toEqual(inbox.enqueued) - expect(inbox.dequeued).toEqual([]) - - release?.() - await agent.whenIdle() - expect(adapter.requests).toHaveLength(0) - expect(agent.session.events).toHaveLength(0) - }) - - it('disposes the agent without waiting for the reservation to be released', async () => { - const adapter = new MockAdapter([]) - const ctx = await harness(adapter) - const handle = await ctx.agents.create({ - sessionId: SessionId('a1'), - agentOptions: { provider: 'mock', model: 'mock' }, - }) - const { agent } = handle - - const release = agent.reserveTurnAdmission() - prompt(agent, 'discarded by disposal') - await handle.dispose() - - expect(ctx.agents.list()).toEqual([]) - expect(adapter.requests).toHaveLength(0) - release?.() - }) -}) diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index d1bafae56a..13fe7a30a3 100644 --- a/packages/core/session/src/index.ts +++ b/packages/core/session/src/index.ts @@ -13,7 +13,7 @@ import { scopeOf, scopeTarget } from '@deepseek-ai/dsh-scope' import type { Scoped } from '@deepseek-ai/dsh-scope' import type { Message } from '@deepseek-ai/dsh-llm' import { SESSION_FORMAT_VERSION, SessionId } from './types.ts' -import type { CreateSessionOptions, EpochHeader, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts' +import type { CreateSessionOptions, EpochHeader, RequestContext, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts' import { snapshotJsonValue } from './json.ts' import { SurfaceManager } from './surface.ts' import type { SessionSurface } from './surface.ts' @@ -392,7 +392,7 @@ export class Session { readonly firstLiveSeq: number constructor(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader) { - if (seed) { + if (seed !== undefined) { // Validate the seed to the SAME invariants `append` enforces, so a // replay/fork (`ctx.sessions.create(id, { seed })`) cannot construct a // live log that no persistence backend could store: each event's `data` @@ -429,7 +429,7 @@ export class Session { // captures the creation seed: no load-time write. Re-marking is skipped // because a cold session is resumed on first touch, so repeatedly opening // one must not grow its log per open. - if (this.firstLiveSeq > 0 && this.log.at(-1)?.type !== 'session/end-seed') { + if (seed !== undefined && this.log.at(-1)?.type !== 'session/end-seed') { this.append('session/end-seed', {}) } } @@ -566,6 +566,25 @@ export class Session { return this.headerFold } + /** Cached fold of `request/context` events. */ + private contextFold: RequestContext | undefined + private contextFoldSeq = 0 + + /** + * Return the latest resolved route metadata, or `undefined` before the first + * `request/context` event. Each event is folded once. + * @returns the latest immutable route metadata. + */ + requestContext(): RequestContext | undefined { + if (this.contextFoldSeq < this.log.length) { + for (const event of this.log.slice(this.contextFoldSeq)) { + if (event.type === 'request/context') this.contextFold = deepFreeze({ ...event.data }) + } + this.contextFoldSeq = this.log.length + } + return this.contextFold + } + /** The derived-message cache: frozen projections, extended per unseen node. */ private derived: Message[] = [] /** Surface position (nodes projected) the cache has reached. */ diff --git a/packages/core/session/src/invariant.ts b/packages/core/session/src/invariant.ts index c454ad6533..f7e7292bc4 100644 --- a/packages/core/session/src/invariant.ts +++ b/packages/core/session/src/invariant.ts @@ -149,7 +149,8 @@ function validateEvent( break case 'steering/message': case 'todo/write': - case 'request/header': { + case 'request/header': + case 'request/context': { if (trace.openTurn === null) { fail(`${event.type} appended outside any open turn (core execution events must be turn-enclosed)`) } diff --git a/packages/core/session/src/types.ts b/packages/core/session/src/types.ts index d814e612f1..691861e8b4 100644 --- a/packages/core/session/src/types.ts +++ b/packages/core/session/src/types.ts @@ -151,6 +151,16 @@ export interface EpochHeader { tools?: ToolSchema[] } +/** Registration-bound metadata for one resolved model route. */ +export interface RequestContext { + /** Registered provider route the metadata belongs to. */ + provider: string + /** Provider-owned model id the metadata belongs to. */ + model: string + /** Maximum combined request and response context in tokens, when advertised. */ + contextWindow?: number +} + /** * Why a `request/header` snapshot was appended: `'initial'` — the log's first * header (a new conversation); `'resume'` — a loop instance's first request @@ -233,6 +243,11 @@ export interface SessionEventMap { * It is log-only; the latest snapshot reconstructs the request header. */ 'request/header': { header: EpochHeader; reason: RequestHeaderReason } + /** + * Route metadata for the next request, logged only when the route or capacity + * changes. It does not participate in request reconstruction or header equality. + */ + 'request/context': RequestContext /** * Marks the end of a constructor seed. Events before it have smaller seq * values and came from the seed (resume, fork, or replay); this lifecycle diff --git a/packages/core/system-prompt/src/index.ts b/packages/core/system-prompt/src/index.ts index 2e4e5e65b7..cb191bcada 100644 --- a/packages/core/system-prompt/src/index.ts +++ b/packages/core/system-prompt/src/index.ts @@ -1,5 +1,5 @@ /** - * Registry for ordered prompt sections, tool schemas, and prompt variables. + * Registry for ordered system sections, dynamic context, tool schemas, and prompt variables. * * @module @deepseek-ai/dsh-system-prompt */ @@ -17,7 +17,7 @@ declare module 'cordis' { interface Events { /** - * Expert waterfall over the assembled sections, tools, and variables. + * Expert waterfall over the assembled sections, contexts, tools, and variables. * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): scoped listeners * receive only that scope's assemblies. The returned value is authoritative. * A supplied signal controls only this explicit assembly request and must not @@ -65,6 +65,16 @@ export interface PromptSection { readonly text: string | ((context: AssembleContext) => string) } +/** Dynamic model context materialized as a durable user-role snapshot. */ +export interface PromptContext { + /** Unique name — a duplicate registration throws (see {@link SystemPrompt.context}). */ + readonly name: string + /** Contexts are joined in ascending order. */ + readonly order: number + /** Static text or a provider evaluated for each assembly. Empty text contributes nothing. */ + readonly text: string | ((context: AssembleContext) => string) +} + /** One section of an assembly: {@link PromptSection} with its text resolved. */ export interface AssembledSection { /** The contributing section's unique name. */ @@ -73,6 +83,14 @@ export interface AssembledSection { text: string } +/** One resolved dynamic context contribution. */ +export interface AssembledContext { + /** The contributing context's unique name. */ + name: string + /** The resolved text before variable interpolation. */ + text: string +} + /** Tool schemas visible in one assembly and their pre-restriction name set. */ export interface ToolProviderResult { /** The schemas this provider contributes to THIS assembly. */ @@ -82,11 +100,12 @@ export interface ToolProviderResult { } /** - * Merge-extensible assembled prompt. Sections remain uninterpolated until - * {@link renderPrompt}; tools are already in canonical model-facing order. + * Merge-extensible assembled model input. Sections and contexts remain + * uninterpolated until rendered; tools are already in canonical order. */ export interface PromptAssembly { sections: AssembledSection[] + contexts: AssembledContext[] tools: ToolSchema[] variables: Record } @@ -170,14 +189,32 @@ export interface Config { */ export function renderPrompt(assembly: PromptAssembly): string { return assembly.sections - .map(section => interpolate(section, assembly.variables)) + .map(section => interpolate(section, assembly.variables, 'section')) .filter(text => text.length > 0) .join('\n\n') } -/** Interpolate one section's `{{variable}}` references (see {@link renderPrompt}). */ -function interpolate(section: AssembledSection, variables: Record): string { - const text = section.text +/** + * Render the complete dynamic context snapshot. + * @param assembly - the assembly whose contexts and variables to render. + * @returns the current full snapshot, or `''` when no context is active. + */ +export function renderContextSnapshot(assembly: PromptAssembly): string { + const body = assembly.contexts + .map(context => interpolate(context, assembly.variables, 'context')) + .filter(text => text.length > 0) + .join('\n\n') + if (body.length === 0) return '' + return `Current runtime context. This snapshot supersedes earlier runtime-context snapshots.\n\n${body}` +} + +/** Interpolate one section or context and attribute diagnostics to its owner. */ +function interpolate( + input: AssembledSection | AssembledContext, + variables: Record, + kind: 'section' | 'context', +): string { + const text = input.text let result = '' let last = 0 for (let open = text.indexOf('{{'); open >= 0; open = text.indexOf('{{', last)) { @@ -185,7 +222,7 @@ function interpolate(section: AssembledSection, variables: Record= 0) { - throw new Error(`malformed prompt variable reference at "${text.slice(open, open + 16)}…" in section "${section.name}" (references are complete simple {{name}} groups)`) + throw new Error(`malformed prompt variable reference at "${text.slice(open, open + 16)}…" in ${kind} "${input.name}" (references are complete simple {{name}} groups)`) } result += text.slice(last, open + 2) last = open + 2 @@ -194,16 +231,16 @@ function interpolate(section: AssembledSection, variables: Record 0 ? known.join(', ') : '(none)'}`) + throw new Error(`unknown prompt variable "{{${name}}}" in ${kind} "${input.name}"; registered variables: ${known.length > 0 ? known.join(', ') : '(none)'}`) } const value = variables[name] if (value === undefined) { - throw new Error(`prompt variable "{{${name}}}" has no value for this assembly (section "${section.name}")`) + throw new Error(`prompt variable "{{${name}}}" has no value for this assembly (${kind} "${input.name}")`) } result += text.slice(last, open) + value last = open + group[0].length @@ -220,6 +257,7 @@ type VariableProvider = (context: AssembleContext) => string | undefined /** All prompt registrations owned by one global or scoped layer. */ class PromptLayer implements ScopeLayer { readonly sections: NamedEntries + readonly contexts: NamedEntries readonly toolProviders = new AnonymousEntries() readonly variables: NamedEntries @@ -231,6 +269,9 @@ class PromptLayer implements ScopeLayer { this.sections = new NamedEntries(name => new Error(scope === undefined ? `prompt section "${name}" is already registered (for a per-agent override, register through that agent's \`agent.ctx\` instead)` : `prompt section "${name}" is already registered in this scope`)) + this.contexts = new NamedEntries(name => new Error(scope === undefined + ? `prompt context "${name}" is already registered (for a per-agent override, register through that agent's \`agent.ctx\` instead)` + : `prompt context "${name}" is already registered in this scope`)) this.variables = new NamedEntries(name => new Error(scope === undefined ? `prompt variable "${name}" is already registered (for a per-agent value, register through that agent's \`agent.ctx\` instead)` : `prompt variable "${name}" is already registered in this scope`)) @@ -239,6 +280,7 @@ class PromptLayer implements ScopeLayer { /** @returns whether this layer owns no prompt registrations. */ isEmpty(): boolean { return this.sections.isEmpty() + && this.contexts.isEmpty() && this.toolProviders.isEmpty() && this.variables.isEmpty() } @@ -297,6 +339,23 @@ export class SystemPrompt extends Service { ) } + /** + * Register ordered dynamic context in the calling context's scope. Scoped + * entries shadow global entries with the same name. + * @param context - the context contribution to register. + * @returns the exact Cordis effect disposer. + */ + context(context: PromptContext): () => void { + if (!Number.isFinite(context.order)) { + throw new TypeError(`prompt context "${context.name}" order must be a finite number`) + } + return this.layers.effect( + this.ctx, + layer => layer.contexts.insert(context.name, context), + { label: 'systemPrompt.context()' }, + ) + } + /** * Register a tool-schema provider in the calling context's scope. Global and * matching scoped providers both contribute; returning the reserved @@ -352,6 +411,7 @@ export class SystemPrompt extends Service { } // Scoped sections shadow globals before the stable order sort. const sectionByName = this.layers.merge(scope, layer => layer.sections) + const contextByName = this.layers.merge(scope, layer => layer.contexts) // Validate order against pre-restriction names while collecting visible schemas. const providers = [ ...this.layers.global.toolProviders.values(), @@ -377,6 +437,12 @@ export class SystemPrompt extends Service { name: section.name, text: typeof section.text === 'function' ? section.text(context) : section.text, })), + contexts: [...contextByName.values()] + .sort((a, b) => a.order - b.order) + .map(entry => ({ + name: entry.name, + text: typeof entry.text === 'function' ? entry.text(context) : entry.text, + })), tools: orderTools(collected, this.toolOrder, knownNames), variables, } diff --git a/packages/core/system-prompt/src/invariant.ts b/packages/core/system-prompt/src/invariant.ts index e199cc98b4..04dc65e7ad 100644 --- a/packages/core/system-prompt/src/invariant.ts +++ b/packages/core/system-prompt/src/invariant.ts @@ -22,6 +22,14 @@ function validateAssembly(assembly: PromptAssembly, fail: InvariantFailure): voi if (typeof section.text !== 'string') fail(`assembled section ${JSON.stringify(section.name)} text must be a string`) } + const contextNames = new Set() + for (const context of assembly.contexts) { + if (context.name.length === 0) fail('assembled context names must be non-empty') + if (contextNames.has(context.name)) fail(`assembled context name ${JSON.stringify(context.name)} is duplicated`) + contextNames.add(context.name) + if (typeof context.text !== 'string') fail(`assembled context ${JSON.stringify(context.name)} text must be a string`) + } + for (const tool of assembly.tools) { if (tool.name.length === 0) fail('assembled tool names must be non-empty') }