Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(session): add cross-session references

Browse Source
This commit is contained in:
Yichen Jiang
2026-07-21 16:46:48 +08:00
parent 9a5c81f9e5
commit 32d786c439
81 changed files with 2837 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/core/agent-loop/README.md
View File

@@ -48,7 +48,7 @@ Configured agents start automatically. A model call requires both `provider` and
The concrete `Agent` class, its `Inbox`, `runLoop`, and instance-bound publication/start controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy.
Each concrete `send()` materializes content plus resolved source once as a detached, deeply frozen lossless-JSON FIFO item. If claimed, it is the sole ordinary message in its turn; a successor waits for the preceding ordinary turn's checkpoint to settle, while cancellation, disposal, or a pre-start failure may drop it without a turn. Running `steer()` enters the steering FIFO: an open turn records it at the next steering checkpoint before a request or continuation decision, but policy can still stop before another step; steering left after turn close and its checkpoint becomes later queued input unless terminal turn policy, cancellation, or disposal discards it. Open-turn `inject()` uses the same accepted-value boundary but defers in a FIFO while the current step executes assistant tool calls; successful batches place it after all results, and interrupted batches drain it before turn close. Malformed data throws before enqueue or append.
Each concrete `send()` materializes content, resolved source, and attached contexts once as a detached, deeply frozen lossless-JSON FIFO item. If claimed, it is the sole ordinary message in its turn; its contexts are the prompt waterfall's default additional contexts and therefore append only after admission. The waterfall's returned allow is authoritative, so a listener wrapping `next()` preserves downstream `content` and `additionalContexts` unless it intentionally replaces them. A successor waits for the preceding ordinary turn's checkpoint to settle, while cancellation, disposal, a prompt block, or a pre-start failure may drop its contexts with the message. Running `steer()` enters the same record shape in the steering FIFO without dispatching `agent/prompt-submit`: an open turn records the steering message followed by its contexts at the next steering checkpoint before a request or continuation decision, but policy can still stop before another step; steering left after turn close and its checkpoint becomes later queued input with contexts intact unless terminal turn policy, cancellation, or disposal discards it. Open-turn `inject()` uses the same accepted-value boundary but defers in a FIFO while the current step executes assistant tool calls; successful batches place it after all results, and interrupted batches drain it before turn close. Malformed data throws before enqueue or append.
### Loop lifecycle (`loop.ts`)

9
packages/core/agent-loop/src/agent.ts
View File

@@ -207,9 +207,10 @@ export class ReactLoopAgent implements Agent {
*/
private acceptMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
const source = this.resolveSource(options)
const accepted = snapshotJsonValue({ content, source })
const contexts = options?.contexts ?? []
const accepted = snapshotJsonValue({ content, source, contexts })
if (accepted === undefined) {
throw new TypeError('agent message content and source must be losslessly JSON-serializable')
throw new TypeError('agent message content, source, and contexts must be losslessly JSON-serializable')
}
return deepFreeze(accepted)
}
@@ -232,7 +233,7 @@ export class ReactLoopAgent implements Agent {
this.assertNotDisposed()
const accepted = this.acceptMessage(content, options)
this.#inbox.enqueue(accepted)
const info = { source: accepted.source, steering: false } as const
const info = { source: accepted.source, contexts: accepted.contexts, steering: false } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
}
@@ -241,7 +242,7 @@ export class ReactLoopAgent implements Agent {
if (this._status !== 'running') { this.send(content, options); return }
const accepted = this.acceptMessage(content, options)
this.#inbox.steer(accepted)
const info = { source: accepted.source, steering: true } as const
const info = { source: accepted.source, contexts: accepted.contexts, steering: true } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
}

2
packages/core/agent-loop/src/inbox.ts
View File

@@ -7,11 +7,13 @@
*/
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
import type { HookContext } from '@deepseek-ai/dsh-agent'
/** One message waiting in an agent's inbox. */
export interface InboxMessage {
content: ContentBlock[]
source: MessageSource
contexts: HookContext[]
}
/**

10
packages/core/agent-loop/src/loop.ts
View File

@@ -207,6 +207,9 @@ async function runTurn(
const messages = handle.inbox.drainSteering()
for (const message of messages) {
session.append('steering/message', { turn, content: message.content, source: message.source }, { surfaceOp: 'append' })
for (const context of message.contexts) {
session.append('context/message', context, { surfaceOp: 'append' })
}
}
return messages.length > 0
}
@@ -263,7 +266,10 @@ async function runTurn(
// throws) is caught below and the turn still closes.
const promptDecision = await events.waterfall(
'agent/prompt-submit', message.content, message.source,
() => Promise.resolve<PromptDecision>({ kind: 'allow' }),
() => Promise.resolve<PromptDecision>({
kind: 'allow',
...message.contexts.length === 0 ? {} : { additionalContexts: message.contexts },
}),
)
if (promptDecision.kind === 'block') {
session.append('prompt/blocked', { content: message.content, source: message.source, reason: promptDecision.reason })
@@ -502,7 +508,7 @@ async function runTurn(
// A continuation reason becomes next-step steering.
if (decision.action === 'continue' && decision.reason) {
handle.inbox.steer({ content: decision.reason.content, source: decision.reason.source })
handle.inbox.steer({ content: decision.reason.content, source: decision.reason.source, contexts: [] })
}
let shouldContinue = decision.action === 'continue'

49
packages/core/agent-loop/tests/contract-regressions.spec.ts
View File

@@ -4,7 +4,7 @@ import LlmService, { CallId, ContentBlock, MessageSource, ProviderRequestId, Str
import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineTool, type PostToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, { type Agent, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
import AgentRegistry, { type Agent, type ContinuationDecision, type HookContext } from '@deepseek-ai/dsh-agent'
import AgentLoop, { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from '@deepseek-ai/dsh-agent-loop'
import { prepareReactLoopAgent } from '../src/agent.ts'
import * as Invariants from '@deepseek-ai/dsh-invariants'
@@ -781,14 +781,14 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
},
}))
const queuedSources: { source: MessageSource; steering: boolean }[] = []
const queuedSources: { source: MessageSource; contexts: HookContext[]; steering: boolean }[] = []
ctx.on('agent/queued', (_agent, _content, info) => void queuedSources.push(info))
send(agent, 'go') // no explicit source → default {kind:'user'} must be visible
await waitForIdle(ctx, agent)
expect(queuedSources[0]).toEqual({ source: { kind: 'user' }, steering: false })
expect(queuedSources[1]).toEqual({ source: { kind: 'plugin', plugin: 'goal' }, steering: true })
expect(queuedSources[0]).toEqual({ source: { kind: 'user' }, contexts: [], steering: false })
expect(queuedSources[1]).toEqual({ source: { kind: 'plugin', plugin: 'goal' }, contexts: [], steering: true })
// The drain appends the durable steering/message with the caller's source
// intact — the log, not a transient emit, is where consumers read it.
const steeringSources = agent.session.events.flatMap(e => e.type === 'steering/message' ? [e.data.source] : [])
@@ -803,24 +803,39 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
const source = { kind: 'plugin' as const, plugin: 'accepted-source' }
let notifiedContent: ContentBlock[] | undefined
let notifiedSource: MessageSource | undefined
let notifiedContexts: HookContext[] | undefined
ctx.on('agent/queued', (subject, acceptedContent, info) => {
if (subject !== agent || info.steering) return
// Retain the exact notification references: cloning here would test the
// listener's copy rather than the event/inbox ownership boundary.
notifiedContent = acceptedContent
notifiedSource = info.source
notifiedContexts = info.contexts
})
agent.send(content, { source })
const contexts: HookContext[] = [{
content: [{ type: 'text', text: 'accepted-context' }],
source: { kind: 'plugin', plugin: 'context-source' },
meta: { version: 1 },
}]
agent.send(content, { source, contexts })
content[0]!.text = 'caller-mutated-send'
source.plugin = 'caller-mutated-source'
contexts[0]!.content[0] = { type: 'text', text: 'caller-mutated-context' }
await waitForIdle(ctx, agent)
expect(notifiedContent).toEqual([{ type: 'text', text: 'accepted-send' }])
expect(notifiedSource).toEqual({ kind: 'plugin', plugin: 'accepted-source' })
expect(notifiedContexts).toEqual([{
content: [{ type: 'text', text: 'accepted-context' }],
source: { kind: 'plugin', plugin: 'context-source' },
meta: { version: 1 },
}])
expect(Object.isFrozen(notifiedContent)).toBe(true)
expect(Object.isFrozen(notifiedContent?.[0])).toBe(true)
expect(Object.isFrozen(notifiedSource)).toBe(true)
expect(Object.isFrozen(notifiedContexts)).toBe(true)
expect(Object.isFrozen(notifiedContexts?.[0]?.content)).toBe(true)
const recorded = agent.session.events.flatMap(event => event.type === 'user/message' ? [event.data] : [])
expect(recorded).toContainEqual({
content: [{ type: 'text', text: 'accepted-send' }],
@@ -828,7 +843,9 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
})
const request = JSON.stringify(adapter.requests[0]!.messages)
expect(request).toContain('accepted-send')
expect(request).toContain('accepted-context')
expect(request).not.toContain('caller-mutated-send')
expect(request).not.toContain('caller-mutated-context')
})
it('running steer() owns content and source before notification and delivery', async () => {
@@ -849,10 +866,12 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
}))
let notifiedContent: ContentBlock[] | undefined
let notifiedSource: MessageSource | undefined
let notifiedContexts: HookContext[] | undefined
ctx.on('agent/queued', (subject, acceptedContent, info) => {
if (subject !== agent || !info.steering) return
notifiedContent = acceptedContent
notifiedSource = info.source
notifiedContexts = info.contexts
})
agent.send([{ type: 'text', text: 'start' }])
@@ -860,18 +879,28 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
expect(agent.status).toBe('running')
const content = [{ type: 'text' as const, text: 'accepted-steer' }]
const source = { kind: 'plugin' as const, plugin: 'accepted-source' }
agent.steer(content, { source })
const contexts: HookContext[] = [{
content: [{ type: 'text', text: 'accepted-steering-context' }],
source: { kind: 'plugin', plugin: 'steering-context' },
}]
agent.steer(content, { source, contexts })
content[0]!.text = 'caller-mutated-steer'
source.plugin = 'caller-mutated-source'
contexts[0]!.content[0] = { type: 'text', text: 'caller-mutated-steering-context' }
const idle = waitForIdle(ctx, agent)
release.resolve(undefined)
await idle
expect(notifiedContent).toEqual([{ type: 'text', text: 'accepted-steer' }])
expect(notifiedSource).toEqual({ kind: 'plugin', plugin: 'accepted-source' })
expect(notifiedContexts).toEqual([{
content: [{ type: 'text', text: 'accepted-steering-context' }],
source: { kind: 'plugin', plugin: 'steering-context' },
}])
expect(Object.isFrozen(notifiedContent)).toBe(true)
expect(Object.isFrozen(notifiedContent?.[0])).toBe(true)
expect(Object.isFrozen(notifiedSource)).toBe(true)
expect(Object.isFrozen(notifiedContexts)).toBe(true)
const recorded = agent.session.events.flatMap(event => event.type === 'steering/message' ? [event.data] : [])
expect(recorded).toContainEqual({
turn: 1,
@@ -880,7 +909,15 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
})
const request = JSON.stringify(adapter.requests[1]!.messages)
expect(request).toContain('accepted-steer')
expect(request).toContain('accepted-steering-context')
expect(request).not.toContain('caller-mutated-steer')
expect(request).not.toContain('caller-mutated-steering-context')
const steeringIndex = agent.session.events.findIndex(event => event.type === 'steering/message')
const contextIndex = agent.session.events.findIndex(event => event.type === 'context/message'
&& event.data.source.kind === 'plugin' && event.data.source.plugin === 'steering-context')
expect(steeringIndex).toBeGreaterThanOrEqual(0)
expect(contextIndex).toBe(steeringIndex + 1)
})
})

20
packages/core/agent-loop/tests/inbox.spec.ts
View File

@@ -1,6 +1,10 @@
import { describe, expect, it } from 'vitest'
import { Inbox } from '../src/inbox.ts'
function message(text: string) {
return { content: [{ type: 'text' as const, text }], source: { kind: 'user' as const }, contexts: [] }
}
function resolverPair() {
let r!: () => void
const p = new Promise<void>((resolve) => { r = resolve })
@@ -10,8 +14,8 @@ function resolverPair() {
describe('Inbox', () => {
it('dequeues one queued message at a time in FIFO order', () => {
const inbox = new Inbox()
inbox.enqueue({ content: [{ type: 'text', text: 'first' }], source: { kind: 'user' } })
inbox.enqueue({ content: [{ type: 'text', text: 'second' }], source: { kind: 'user' } })
inbox.enqueue(message('first'))
inbox.enqueue(message('second'))
expect(inbox.hasQueued).toBe(true)
expect(inbox.dequeueQueued()?.content[0]).toMatchObject({ text: 'first' })
@@ -23,7 +27,7 @@ describe('Inbox', () => {
it('pushes and drains steering messages separately from queued', () => {
const inbox = new Inbox()
inbox.steer({ content: [{ type: 'text', text: 'steer' }], source: { kind: 'user' } })
inbox.steer(message('steer'))
expect(inbox.hasQueued).toBe(false)
expect(inbox.hasSteering).toBe(true)
@@ -34,7 +38,7 @@ describe('Inbox', () => {
it('waitForQueued returns immediately when a queued message is already present', async () => {
const inbox = new Inbox()
inbox.enqueue({ content: [{ type: 'text', text: 'ready' }], source: { kind: 'user' } })
inbox.enqueue(message('ready'))
const started = Date.now()
await inbox.waitForQueued(new Promise(() => {})) // never-resolving cancel
@@ -45,7 +49,7 @@ describe('Inbox', () => {
const inbox = new Inbox()
const waiter = inbox.waitForQueued(new Promise(() => {})) // never-resolving cancel
// enqueue after starting the wait
setTimeout(() => { inbox.enqueue({ content: [{ type: 'text', text: 'wake' }], source: { kind: 'user' } }) }, 5)
setTimeout(() => { inbox.enqueue(message('wake')) }, 5)
await waiter
})
@@ -69,7 +73,7 @@ describe('Inbox', () => {
r1()
await p1
inbox.enqueue({ content: [{ type: 'text', text: 'hey' }], source: { kind: 'user' } })
inbox.enqueue(message('hey'))
})
it('clears wakeup in finally handler when enqueue resolves', async () => {
@@ -77,7 +81,7 @@ describe('Inbox', () => {
void inbox.waitForQueued(new Promise(() => {})) // never-resolving cancel
// The wakeup is set. Now trigger it via enqueue → wakeup() calls resolve,
// promise resolves, finally clears wakeup because wakeup === resolve.
inbox.enqueue({ content: [{ type: 'text', text: 'wake' }], source: { kind: 'user' } })
inbox.enqueue(message('wake'))
// No explicit await needed — enqueue is synchronous, and the microtask
// (finally) runs. The key coverage hit is finally with wakeup === resolve.
})
@@ -94,6 +98,6 @@ describe('Inbox', () => {
await c1
// The replacement remains registered and is resolved by enqueue.
inbox.enqueue({ content: [{ type: 'text', text: 'hey' }], source: { kind: 'user' } })
inbox.enqueue(message('hey'))
})
})

5
packages/core/agent-loop/tests/interception.spec.ts
View File

@@ -154,7 +154,9 @@ describe('agent/prompt-submit', () => {
const reasons: TurnEndReason[] = []
ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
send(agent, 'do something')
agent.send([{ type: 'text', text: 'do something' }], {
contexts: [{ content: [{ type: 'text', text: 'must be dropped' }], source: { kind: 'plugin', plugin: 'test' } }],
})
await waitForIdle(ctx, agent)
// the model was never called
@@ -164,6 +166,7 @@ describe('agent/prompt-submit', () => {
expect(log.some(e => e.type === 'turn/start')).toBe(true)
expect(log.some(e => e.type === 'turn/end')).toBe(true)
expect(log.some(e => e.type === 'user/message')).toBe(false)
expect(log.some(e => e.type === 'context/message')).toBe(false)
expect(log.some(e => e.type === 'step/start')).toBe(false)
// the veto is recorded durably as a prompt/blocked in the open turn
const blocked = log.find(e => e.type === 'prompt/blocked')