feat(memory): land the loop ports without wiring them
The four files under lib/loop/ were written against 0.2.1-rc.2 and have been
verified to import on their own; they are the substrate a replacement loop
needs to take over step(). This commit only puts them in the tree — lib/loop.js
is untouched, so nothing imports them yet and behaviour is unchanged.
Two packaging facts they bring with them:
files `lib/*.js` does not recurse, so lib/loop/ was shipping nothing.
`lib/loop.js` and `lib/loop/*.js` are now listed separately because
neither glob covers the other.
peer runtime-context.js imports @deepseek-ai/dsh-session, which the peer
table never declared. G2b's allowlist covered it, so the static guard
passed over a real declaration gap.
test/loop-ports.test.js pins the port interface rather than its behaviour: the
modules import alone, the export names survive, and the inbox projection keeps
key "inbox" / stateVersion 1. That literal pair is a wire format shared with
stock-loop sessions — a different key silently reads as "no pending input", a
different version makes stateSchema.parse refuse stored rows. The schema's
parse() is also exercised against seven malformed persisted rows, since
replacing a zod schema with a hand-written check is exactly the kind of change
that can be a no-op and still look fine.
This commit is contained in:
140
lib/loop/assistant-stream.js
Normal file
140
lib/loop/assistant-stream.js
Normal file
@ -0,0 +1,140 @@
|
||||
/**
|
||||
* Folds one model attempt into one compact stream plus ordered transient frames.
|
||||
*
|
||||
* Reproduced from `dsh-agent-loop/lib/types/index.js:352-473` (0.2.1-rc.2, MIT)
|
||||
* because `dsh-agent-loop` does not export it, and there is no behaviour to
|
||||
* change here: the framing has to match what `dsh-headless` and the session
|
||||
* title plugin read off `agent/assistant-stream` byte for byte.
|
||||
*
|
||||
* @module dsh-memory/lib/loop/assistant-stream
|
||||
*/
|
||||
|
||||
import { AssistantStreamAccumulator, BlockAssembler, LlmAttemptId } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
export class AssistantStreamAttempt {
|
||||
/** @type {() => number} */
|
||||
nextRevision
|
||||
/** @type {number} */
|
||||
turn
|
||||
/** @type {number} */
|
||||
step
|
||||
/** @type {(frame: object) => void} */
|
||||
emit
|
||||
accumulator = new AssistantStreamAccumulator()
|
||||
assembler = new BlockAssembler()
|
||||
index = 0
|
||||
terminal = false
|
||||
/** Attempt identity unique within this Agent lifecycle. */
|
||||
attemptId
|
||||
|
||||
/** Whether this started attempt has emitted its terminal frame. */
|
||||
get ended() {
|
||||
return this.terminal
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} sessionId identity embedded only in the Agent-lifecycle-local attempt id
|
||||
* @param {number} attempt attached-Session-local attempt counter
|
||||
* @param {() => number} nextRevision allocates the next emitted frame revision
|
||||
* @param {number} turn durable turn owning the request
|
||||
* @param {number} step durable step owning the request
|
||||
* @param {(frame: object) => void} emit agent-scoped notification publisher
|
||||
*/
|
||||
constructor(sessionId, attempt, nextRevision, turn, step, emit) {
|
||||
this.nextRevision = nextRevision
|
||||
this.turn = turn
|
||||
this.step = step
|
||||
this.emit = emit
|
||||
this.attemptId = LlmAttemptId(`${sessionId}:${attempt}`)
|
||||
}
|
||||
|
||||
/** Publish the opening marker before the first delivered chunk. */
|
||||
start() {
|
||||
this.emit({
|
||||
type: 'start',
|
||||
attemptId: this.attemptId,
|
||||
revision: this.nextRevision(),
|
||||
turn: this.turn,
|
||||
step: this.step,
|
||||
})
|
||||
}
|
||||
|
||||
/** Snapshot one chunk once, then feed durable compaction, assembly, and live publication. */
|
||||
push(chunk) {
|
||||
const timed = this.accumulator.push({ time: Date.now(), chunk })
|
||||
this.assembler.push(timed.chunk)
|
||||
this.emit({
|
||||
type: 'chunk',
|
||||
attemptId: this.attemptId,
|
||||
revision: this.nextRevision(),
|
||||
index: this.index++,
|
||||
time: timed.time,
|
||||
chunk: timed.chunk,
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Publish terminal settlement after the matching durable event commits.
|
||||
* @param {string} eventType durable settlement type
|
||||
* @param {() => number} append synchronous durable append returning its committed seq
|
||||
*/
|
||||
settle(eventType, append) {
|
||||
let seq
|
||||
try {
|
||||
seq = append()
|
||||
} catch (error) {
|
||||
this.abandon()
|
||||
throw error
|
||||
}
|
||||
this.terminal = true
|
||||
this.emit({
|
||||
type: 'end',
|
||||
attemptId: this.attemptId,
|
||||
revision: this.nextRevision(),
|
||||
index: this.index,
|
||||
outcome: { kind: 'committed', eventType, seq },
|
||||
})
|
||||
}
|
||||
|
||||
/** Publish abandonment when no durable attempt event can be committed. */
|
||||
abandon() {
|
||||
this.terminal = true
|
||||
this.emit({
|
||||
type: 'end',
|
||||
attemptId: this.attemptId,
|
||||
revision: this.nextRevision(),
|
||||
index: this.index,
|
||||
outcome: { kind: 'abandoned' },
|
||||
})
|
||||
}
|
||||
|
||||
/** Exact compact stream for the final durable event. */
|
||||
get stream() {
|
||||
return [...this.accumulator.snapshot()]
|
||||
}
|
||||
|
||||
/** Canonical completed-message blocks from the same chunks. */
|
||||
blocks() {
|
||||
return this.assembler.blocks()
|
||||
}
|
||||
|
||||
/** Safe visible prefix when cancellation interrupts the attempt. */
|
||||
interruptedBlocks() {
|
||||
return this.assembler.interruptedBlocks()
|
||||
}
|
||||
|
||||
/** Latest adapter-reported usage in the stream. */
|
||||
get usage() {
|
||||
return this.assembler.usage
|
||||
}
|
||||
|
||||
/** Terminal reason, defaulting to stop when the stream omitted one. */
|
||||
get finish() {
|
||||
return this.assembler.finish
|
||||
}
|
||||
|
||||
/** Replay state carried by the terminal finish record. */
|
||||
get replayState() {
|
||||
return this.assembler.replayState
|
||||
}
|
||||
}
|
||||
255
lib/loop/inbox.js
Normal file
255
lib/loop/inbox.js
Normal file
@ -0,0 +1,255 @@
|
||||
/**
|
||||
* Durable agent-inbox projection and its command facade.
|
||||
*
|
||||
* ## Why this file exists at all
|
||||
*
|
||||
* `dsh-agent-loop` keeps `inboxProjectionDefinition` module-private: its
|
||||
* package entry exports only `AgentLoop`, `DEFAULT_MAX_PARALLEL_TOOL_CALLS`,
|
||||
* `CONFIGURED_AGENT_IDENTITIES_KEY` and `turnBoundaryProjectionDefinition`
|
||||
* (dsh-agent-loop/lib/index.js:2023). A replacement loop therefore cannot
|
||||
* import the fold, and re-deriving pending input in memory instead of in a
|
||||
* projection would lose queued messages on `resume` — the pending splices live
|
||||
* in the durable log, so the state has to be rebuilt from it.
|
||||
*
|
||||
* So the fold is reproduced here, field for field, from
|
||||
* `dsh-agent-loop/lib/index.js:26-60` (0.2.1-rc.2, MIT). The projection KEY
|
||||
* must stay `"inbox"` and `stateVersion` must stay 1: a different key would
|
||||
* silently read as "no pending input" for sessions written by the stock loop,
|
||||
* and a different version would make `stateSchema.parse` refuse those rows.
|
||||
*
|
||||
* ★ The one deliberate deviation: the host validates with zod
|
||||
* (`z.array(z.custom())`), which is a dependency this plugin does not carry.
|
||||
* {@link inboxProjectionSchema} provides the same contract the projection
|
||||
* registry actually uses — `parse()` that throws on a malformed persisted row —
|
||||
* without pulling zod in. Nothing else about the fold changed.
|
||||
*
|
||||
* @module dsh-memory/lib/loop/inbox
|
||||
*/
|
||||
|
||||
/**
|
||||
* Validate one persisted inbox state.
|
||||
*
|
||||
* The registry calls `stateSchema.parse` on durable checkpoint rows and on the
|
||||
* value a fold returns, so this must reject anything that is not the two-array
|
||||
* shape. Message contents stay opaque (`z.custom()` in the host): they are
|
||||
* already validated by the session layer when they were added to the surface,
|
||||
* and re-validating them here would make the fold able to reject a log the
|
||||
* session accepted.
|
||||
*/
|
||||
export const inboxProjectionSchema = {
|
||||
/**
|
||||
* @param {unknown} value candidate state
|
||||
* @returns {{'next-turn': Array<object>, 'next-step': Array<object>}} the same object
|
||||
* @throws {Error} when the shape is not two message arrays
|
||||
*/
|
||||
parse(value) {
|
||||
if (value === null || typeof value !== 'object' || Array.isArray(value)) {
|
||||
throw new Error('inbox projection state must be an object')
|
||||
}
|
||||
const nextTurn = value['next-turn']
|
||||
const nextStep = value['next-step']
|
||||
if (!Array.isArray(nextTurn) || !Array.isArray(nextStep)) {
|
||||
throw new Error('inbox projection state must hold "next-turn" and "next-step" message arrays')
|
||||
}
|
||||
return value
|
||||
},
|
||||
}
|
||||
|
||||
/** Standard fold that reconstructs pending input and rejects invalid splice history. */
|
||||
export const inboxProjectionDefinition = {
|
||||
key: 'inbox',
|
||||
stateSchema: inboxProjectionSchema,
|
||||
init: () => ({
|
||||
'next-turn': [],
|
||||
'next-step': [],
|
||||
}),
|
||||
apply(state, event) {
|
||||
if (event.type !== 'agent/inbox/spliced') return state
|
||||
const splice = event.data
|
||||
try {
|
||||
const inbox = state[splice.target]
|
||||
const removedCount = splice.removedCount ?? 0
|
||||
if (
|
||||
!Number.isSafeInteger(splice.start) ||
|
||||
splice.start < 0 ||
|
||||
splice.start > inbox.length ||
|
||||
!Number.isSafeInteger(removedCount) ||
|
||||
removedCount < 0 ||
|
||||
splice.start + removedCount > inbox.length
|
||||
) {
|
||||
throw new Error('invalid inbox splice')
|
||||
}
|
||||
const next = inbox.toSpliced(splice.start, removedCount, ...splice.inserted)
|
||||
const ids = new Set()
|
||||
for (const message of splice.target === 'next-turn' ? [...next, ...state['next-step']] : [...state['next-turn'], ...next]) {
|
||||
if (ids.has(message.id)) throw new Error(`message "${message.id}" is already pending`)
|
||||
ids.add(message.id)
|
||||
}
|
||||
return splice.target === 'next-turn'
|
||||
? { 'next-turn': next, 'next-step': state['next-step'] }
|
||||
: { 'next-turn': state['next-turn'], 'next-step': next }
|
||||
} catch (error) {
|
||||
throw new Error(`invalid persisted inbox splice at session seq ${event.seq}`, { cause: error })
|
||||
}
|
||||
},
|
||||
wire: {
|
||||
viewSchema: inboxProjectionSchema,
|
||||
view: (state) => state,
|
||||
},
|
||||
stateVersion: 1,
|
||||
}
|
||||
|
||||
/**
|
||||
* Driver-owned durable Inbox implementation.
|
||||
*
|
||||
* Every mutation is a durable `agent/inbox/spliced` append; the projection is
|
||||
* the only state holder, so a resumed session sees exactly the pending input
|
||||
* the previous process left behind.
|
||||
*/
|
||||
export class ReactLoopInbox {
|
||||
/**
|
||||
* @param {object} projections `ctx.sessionProjections`
|
||||
* @param {object} session the owning session
|
||||
* @param {object} dispatch agent-scoped dispatcher
|
||||
*/
|
||||
constructor(projections, session, dispatch) {
|
||||
this.projections = projections
|
||||
this.session = session
|
||||
this.dispatch = dispatch
|
||||
}
|
||||
|
||||
/** Prompts awaiting individual turns. */
|
||||
get nextTurn() {
|
||||
return this.current()['next-turn']
|
||||
}
|
||||
|
||||
/** Input awaiting the next step boundary. */
|
||||
get nextStep() {
|
||||
return this.current()['next-step']
|
||||
}
|
||||
|
||||
/** Whether either pending-message list contains work. */
|
||||
get hasPending() {
|
||||
const state = this.current()
|
||||
return state['next-turn'].length > 0 || state['next-step'].length > 0
|
||||
}
|
||||
|
||||
/** Durably cancel all pending input, clearing next-step before next-turn. */
|
||||
clear() {
|
||||
this.splice('next-step', 0, this.nextStep.length, [])
|
||||
this.splice('next-turn', 0, this.nextTurn.length, [])
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove and return the complete batch proposed for one step.
|
||||
* @param {'next-turn'|'next-step'} target whether this boundary also consumes one queued turn
|
||||
* @param {number} turn turn that will own the claimed batch
|
||||
*/
|
||||
claim(target, turn) {
|
||||
const claimed = this.mutate('next-step', 0, this.nextStep.length, [], false)
|
||||
if (target === 'next-turn') claimed.push(...this.mutate('next-turn', 0, 1, [], false))
|
||||
for (const message of claimed) this.dispatch.emit('agent/inbox/claimed', { message, turn })
|
||||
return claimed
|
||||
}
|
||||
|
||||
/**
|
||||
* Append one message to a pending list.
|
||||
* @param {'next-turn'|'next-step'} target pending list to extend
|
||||
* @param {object} message message to append
|
||||
*/
|
||||
append(target, message) {
|
||||
this.splice(target, this.current()[target].length, 0, [message])
|
||||
}
|
||||
|
||||
/**
|
||||
* Prepend one message to a pending list.
|
||||
* @param {'next-turn'|'next-step'} target pending list to extend
|
||||
* @param {object} message message to prepend
|
||||
*/
|
||||
prepend(target, message) {
|
||||
this.splice(target, 0, 0, [message])
|
||||
}
|
||||
|
||||
/**
|
||||
* Replace one pending message in place.
|
||||
* @param {string} messageId identity of the pending message to replace
|
||||
* @param {object} newMessage replacement message
|
||||
* @returns {boolean} whether the message was still pending
|
||||
*/
|
||||
replace(messageId, newMessage) {
|
||||
const location = this.locate(messageId)
|
||||
if (location === undefined) return false
|
||||
this.splice(location.target, location.index, 1, [newMessage])
|
||||
return true
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove one pending message.
|
||||
* @param {string} messageId identity of the pending message to remove
|
||||
* @returns {boolean} whether the message was still pending
|
||||
*/
|
||||
remove(messageId) {
|
||||
const location = this.locate(messageId)
|
||||
if (location === undefined) return false
|
||||
this.splice(location.target, location.index, 1, [])
|
||||
return true
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply standard splice semantics and durably record the normalized result.
|
||||
* @param {'next-turn'|'next-step'} target pending list to mutate
|
||||
* @param {number} start splice position
|
||||
* @param {number} deleteCount maximum number of messages to remove
|
||||
* @param {Array<object>} inserted messages to insert at the resolved position
|
||||
*/
|
||||
splice(target, start, deleteCount, inserted) {
|
||||
return this.mutate(target, start, deleteCount, inserted, true)
|
||||
}
|
||||
|
||||
/** Locate one pending identity across both owned lists. */
|
||||
locate(messageId) {
|
||||
const state = this.current()
|
||||
for (const target of ['next-turn', 'next-step']) {
|
||||
const index = state[target].findIndex((message) => message.id === messageId)
|
||||
if (index >= 0) return { target, index }
|
||||
}
|
||||
}
|
||||
|
||||
/** Read the current durable projection state. */
|
||||
current() {
|
||||
const state = this.projections.stateOf(this.session, 'inbox')
|
||||
if (state === undefined) throw new Error(`agent "${this.session.id}" cannot read inbox state: its projection registration is not active`)
|
||||
return state
|
||||
}
|
||||
|
||||
/** Commit one normalized mutation and publish its live events. */
|
||||
mutate(target, start, deleteCount, inserted, discardRemoved) {
|
||||
const state = this.current()
|
||||
const inbox = state[target]
|
||||
const truncatedStart = Math.trunc(start)
|
||||
const offset = Number.isNaN(truncatedStart) ? 0 : truncatedStart
|
||||
const actualStart = offset < 0 ? Math.max(inbox.length + offset, 0) : Math.min(offset, inbox.length)
|
||||
const truncatedDeleteCount = Math.trunc(deleteCount)
|
||||
const actualDeleteCount = Math.min(Math.max(Number.isNaN(truncatedDeleteCount) ? 0 : truncatedDeleteCount, 0), inbox.length - actualStart)
|
||||
if (actualDeleteCount === 0 && inserted.length === 0) return []
|
||||
const candidate = inbox.toSpliced(actualStart, actualDeleteCount, ...inserted)
|
||||
const ids = new Set()
|
||||
for (const message of target === 'next-turn' ? [...candidate, ...state['next-step']] : [...state['next-turn'], ...candidate]) {
|
||||
if (ids.has(message.id)) throw new Error(`message "${message.id}" is already pending`)
|
||||
ids.add(message.id)
|
||||
}
|
||||
const outcome = discardRemoved && actualDeleteCount > 0 ? 'canceled' : undefined
|
||||
const splice = {
|
||||
target,
|
||||
start: actualStart,
|
||||
...(actualDeleteCount === 0 ? {} : { removedCount: actualDeleteCount }),
|
||||
inserted,
|
||||
...(outcome === undefined ? {} : { outcome }),
|
||||
}
|
||||
const removed = inbox.slice(actualStart, actualStart + actualDeleteCount)
|
||||
const event = this.session.append('agent/inbox/spliced', splice)
|
||||
if (discardRemoved) for (const message of removed) this.dispatch.emit('agent/inbox/discarded', { message })
|
||||
for (const message of event.data.inserted) this.dispatch.emit('agent/inbox/inserted', { message })
|
||||
return removed
|
||||
}
|
||||
}
|
||||
145
lib/loop/runtime-context.js
Normal file
145
lib/loop/runtime-context.js
Normal file
@ -0,0 +1,145 @@
|
||||
/**
|
||||
* The two surface projections the loop owns: the rendered system prompt and
|
||||
* the dynamic runtime-context snapshot.
|
||||
*
|
||||
* ## Why this is a copy and not an import
|
||||
*
|
||||
* `dsh-agent-loop` does not export either class (its entry exports only
|
||||
* `AgentLoop`, the constants and `turnBoundaryProjectionDefinition`,
|
||||
* dsh-agent-loop/lib/index.js:2023). They are nevertheless required: without
|
||||
* `SystemPromptProjection` the prompt would never reach surface node 0, and
|
||||
* without `RuntimeContextProjection` every step would append a duplicate
|
||||
* snapshot. Both are reproduced from
|
||||
* `dsh-agent-loop/lib/types/index.js:211-351` (0.2.1-rc.2, MIT) unchanged.
|
||||
*
|
||||
* ★ `SOURCE = 'runtime-context'` is load-bearing and must not be renamed: the
|
||||
* stock loop also owns snapshots under that `source.kind`, so a session that
|
||||
* switches between the stock loop and this one (a profile patch applied
|
||||
* mid-life) must still recognise the previous process's snapshot as its own.
|
||||
* A different kind would leave two snapshots in the surface, both claiming to
|
||||
* supersede the other.
|
||||
*
|
||||
* @module dsh-memory/lib/loop/runtime-context
|
||||
*/
|
||||
|
||||
import { createSystemMessage, createUserMessage } from '@deepseek-ai/dsh-llm'
|
||||
import { isReplacementSurfaceEvent } from '@deepseek-ai/dsh-session'
|
||||
|
||||
const SOURCE = 'runtime-context'
|
||||
const CLEARED = 'Current runtime context: none. Earlier runtime-context snapshots no longer apply.'
|
||||
|
||||
/** @param {object} message derived message */
|
||||
function isOwned(message) {
|
||||
return message.source.kind === SOURCE
|
||||
}
|
||||
|
||||
/** @param {object} message derived message */
|
||||
function textOf(message) {
|
||||
const [block] = message.content
|
||||
return message.content.length === 1 && block?.type === 'text' ? block.text : undefined
|
||||
}
|
||||
|
||||
/** Committed events from the newest backward; the restore scans stop at the first match. */
|
||||
function eventsNewestFirst(session) {
|
||||
return session.snapshotEvents().toReversed()
|
||||
}
|
||||
|
||||
/**
|
||||
* Decides how a rendered system prompt reaches the surface without owning the
|
||||
* commit. The first prompt, even empty, reserves surface node 0.
|
||||
*/
|
||||
export class SystemPromptProjection {
|
||||
/** @param {object} session */
|
||||
constructor(session) {
|
||||
this.session = session
|
||||
}
|
||||
|
||||
/** The surviving `system/message` nodes in surface order. */
|
||||
systemNodes() {
|
||||
const nodes = []
|
||||
for (const seq of this.session.surface.nodes) {
|
||||
const event = this.session.eventAt(seq)
|
||||
if (event?.type !== 'system/message') continue
|
||||
const text = event.data.message.content.length === 0 ? '' : textOf(event.data.message)
|
||||
nodes.push({ seq, text })
|
||||
}
|
||||
return nodes
|
||||
}
|
||||
|
||||
/**
|
||||
* Reconcile effective text and retained nodes with the prepared route and series.
|
||||
* @param {string} rendered the fully rendered system prompt; `''` when none is active
|
||||
* @param {{inHistory: boolean, startsSeries: boolean}} input route capability and series facts
|
||||
*/
|
||||
project(rendered, input) {
|
||||
const nodes = this.systemNodes()
|
||||
const head = nodes[0]
|
||||
if (head === undefined) return [{ message: createSystemMessage(rendered), intent: { surfaceOp: 'append' } }]
|
||||
const latest = nodes.findLast((node) => node.text !== '') ?? head
|
||||
if (!input.inHistory || input.startsSeries || rendered.length === 0) {
|
||||
const updates = nodes.slice(1).filter((node) => node.text !== '').map((node) => this.replace(node.seq, ''))
|
||||
if (head.text !== rendered) updates.push(this.replace(head.seq, rendered))
|
||||
return updates
|
||||
}
|
||||
if (latest.text === rendered) return []
|
||||
return [{ message: createSystemMessage(rendered), intent: { surfaceOp: 'append' } }]
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {number} seq surface node to replace
|
||||
* @param {string} text replacement text
|
||||
*/
|
||||
replace(seq, text) {
|
||||
return {
|
||||
message: createSystemMessage(text),
|
||||
intent: { surfaceOp: { op: 'replace', startSeq: seq, endSeq: seq }, sourceEventSeqs: [seq] },
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 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. */
|
||||
retained
|
||||
|
||||
/**
|
||||
* Restore projection state once, then follow authoritative session events.
|
||||
* @param {object} ctx agent-scoped event context
|
||||
* @param {object} session session receiving projected messages
|
||||
*/
|
||||
constructor(ctx, session) {
|
||||
const surface = new Set(session.surface.nodes)
|
||||
for (const event of eventsNewestFirst(session)) {
|
||||
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 {string} current fully rendered dynamic context
|
||||
* @param {Array<{name: string, text: string}>} sections named contributions of the snapshot
|
||||
* @returns {object|undefined} a candidate user message, or undefined when no update is needed
|
||||
*/
|
||||
project(current, sections) {
|
||||
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: sections.length === 0 ? { kind: SOURCE } : { kind: SOURCE, form: 'snapshot', sections },
|
||||
})
|
||||
}
|
||||
}
|
||||
254
lib/loop/tool-calls.js
Normal file
254
lib/loop/tool-calls.js
Normal file
@ -0,0 +1,254 @@
|
||||
/**
|
||||
* Schedules one assistant step's tool calls. Exclusive calls form barriers;
|
||||
* parallel calls use a bounded rolling pool and are reclassified before start.
|
||||
* Dispatch may overlap, while policy, results, and result context remain
|
||||
* model-ordered. Abort or an internal scheduler failure stops replenishment
|
||||
* and drains started calls.
|
||||
*
|
||||
* Reproduced from `dsh-agent-loop/lib/types/tool-calls.js:474-708`
|
||||
* (0.2.1-rc.2, MIT); the loop replacement deliberately changes no scheduling
|
||||
* behaviour, so this file is a port rather than a rewrite.
|
||||
*
|
||||
* ★ One substitution: the host reads its parallelism cap from
|
||||
* `ctx.agentLoop.config.maxParallelToolCalls` (dsh-agent-loop/lib/index.js:553),
|
||||
* a service this plugin does not provide (its only consumer is the Web settings
|
||||
* panel, which is out of scope). The caller passes the same value explicitly
|
||||
* through {@link ToolCallSchedulerOptions.maxParallelToolCalls} instead; the
|
||||
* cap is still a live getter, so a dynamic override keeps working.
|
||||
*
|
||||
* @module dsh-memory/lib/loop/tool-calls
|
||||
*/
|
||||
|
||||
import { createToolResultMessage } from '@deepseek-ai/dsh-llm'
|
||||
import { TOOL_ABORTED_BEFORE_DISPATCH, TOOL_RUNTIME_SCHEDULER } from '@deepseek-ai/dsh-tools'
|
||||
|
||||
/**
|
||||
* Throw on a value the compiler proved unreachable.
|
||||
*
|
||||
* Inlined rather than imported from `@deepseek-ai/dsh-util-values`, and that is a
|
||||
* G2b decision, not an oversight: `dsh-util-values` is a *dependency* of
|
||||
* `dsh-agent-loop`, not one of its peers (dsh-agent-loop/package.json). Adding it
|
||||
* here would therefore introduce a runtime dependency this plugin carries itself
|
||||
* — exactly what G2b exists to catch — and it would break H5 (no new deps) for a
|
||||
* three-line helper. The four lines below are the upstream body; the message
|
||||
* prefix stays identical so a log line still names the same call site.
|
||||
*
|
||||
* @param {never} value the value that cannot occur
|
||||
* @param {string} [context] name of the switch that produced it
|
||||
* @throws {Error} always
|
||||
*/
|
||||
function assertNever(value, context) {
|
||||
const rendered = JSON.stringify(value) ?? String(value)
|
||||
throw new Error(`unreachable variant${context ? ` in ${context}` : ''}: ${rendered}`)
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule one assistant step's tool calls by their live concurrency mode.
|
||||
*
|
||||
* Ordinary completion and abort commit started-call results in order. Abort
|
||||
* drains them, records synthetic results for unstarted calls, and returns with
|
||||
* the signal still aborted after accepting started-call context through the
|
||||
* caller-supplied acceptor.
|
||||
*
|
||||
* @param {object} ctx loop context that owns the tool registry and carries the initiating Agent
|
||||
* @param {number} turn current turn number
|
||||
* @param {number} step current step number
|
||||
* @param {Array<object>} toolCalls assistant calls in model order
|
||||
* @param {AbortSignal} signal abort signal shared by the step
|
||||
* @param {(context: object) => void} acceptContext accepts committed result context for the next step
|
||||
* @param {{maxParallelToolCalls: () => number}} options the live parallelism cap
|
||||
* @returns {Promise<{concluded: boolean}>}
|
||||
*/
|
||||
export async function executeToolCalls(ctx, turn, step, toolCalls, signal, acceptContext, options) {
|
||||
const agent = ctx.agents.requireInitiator()
|
||||
const { session } = agent
|
||||
const planned = toolCalls.map((block) => ({
|
||||
block,
|
||||
exec: { callId: block.id, name: block.name, arguments: parseArguments(block.arguments), agent, signal },
|
||||
}))
|
||||
let next = 0
|
||||
let concluded = false
|
||||
while (next < planned.length) {
|
||||
const first = planned[next]
|
||||
const mode = ctx.tools.executionMode(first.exec).kind
|
||||
const outcome = await runGroup(
|
||||
ctx,
|
||||
turn,
|
||||
step,
|
||||
mode === 'parallel' ? planned.slice(next) : [first],
|
||||
mode,
|
||||
signal,
|
||||
acceptContext,
|
||||
options,
|
||||
)
|
||||
next += outcome.consumed
|
||||
concluded ||= outcome.concluded
|
||||
if (outcome.aborted) {
|
||||
for (const call of planned.slice(next)) appendSkippedToolCall(session, turn, step, call.block)
|
||||
return { concluded }
|
||||
}
|
||||
}
|
||||
return { concluded }
|
||||
}
|
||||
|
||||
/** Parse model arguments, preserving invalid JSON as text and mapping empty input to `{}`. */
|
||||
function parseArguments(raw) {
|
||||
try {
|
||||
return raw ? JSON.parse(raw) : {}
|
||||
} catch {
|
||||
return raw
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run one exclusive barrier or parallel pool.
|
||||
*
|
||||
* @param {object} ctx
|
||||
* @param {number} turn
|
||||
* @param {number} step
|
||||
* @param {Array<{block: object, exec: object}>} group
|
||||
* @param {'parallel'|'exclusive'} mode
|
||||
* @param {AbortSignal} signal
|
||||
* @param {(context: object) => void} acceptContext
|
||||
* @param {{maxParallelToolCalls: () => number}} options
|
||||
*/
|
||||
async function runGroup(ctx, turn, step, group, mode, signal, acceptContext, options) {
|
||||
const { session } = ctx.agents.requireInitiator()
|
||||
const maxParallelToolCalls = options.maxParallelToolCalls()
|
||||
const slots = group.map(() => undefined)
|
||||
const callSeqs = group.map(() => undefined)
|
||||
let nextToStart = 0
|
||||
let committed = 0
|
||||
let started = 0
|
||||
let aborted = signal.aborted
|
||||
let concluded = false
|
||||
let schedulerFailure
|
||||
const throwSchedulerFailure = () => {
|
||||
if (schedulerFailure !== undefined) throw schedulerFailure.error
|
||||
}
|
||||
const commitReady = async () => {
|
||||
while (committed < group.length) {
|
||||
const slot = slots[committed]
|
||||
if (slot === undefined) break
|
||||
const call = group[committed]
|
||||
const result = slot.needsPost
|
||||
? await ctx.tools[TOOL_RUNTIME_SCHEDULER].finalize(slot.exec, slot.result)
|
||||
: ctx.tools[TOOL_RUNTIME_SCHEDULER].finish(slot.exec, slot.result)
|
||||
appendToolResult(session, turn, step, call.block, result, callSeqs[committed])
|
||||
for (const context of result.additionalContexts ?? []) acceptContext(context)
|
||||
concluded ||= result.concludesTurn === true
|
||||
committed++
|
||||
}
|
||||
}
|
||||
const inFlight = new Map()
|
||||
const startCall = async (index) => {
|
||||
const call = group[index]
|
||||
callSeqs[index] = appendToolCall(session, turn, step, call.block)
|
||||
started++
|
||||
const prepared = await ctx.tools[TOOL_RUNTIME_SCHEDULER].prepare(call.exec)
|
||||
throwSchedulerFailure()
|
||||
switch (prepared.kind) {
|
||||
case 'dispatch': {
|
||||
const promise = ctx.tools[TOOL_RUNTIME_SCHEDULER].dispatch(prepared.exec).then(
|
||||
(outcome) => {
|
||||
slots[index] = { exec: prepared.exec, result: outcome.result, needsPost: outcome.kind === 'post-result' }
|
||||
return index
|
||||
},
|
||||
(error) => {
|
||||
schedulerFailure ??= { error }
|
||||
return index
|
||||
},
|
||||
)
|
||||
inFlight.set(index, promise)
|
||||
break
|
||||
}
|
||||
case 'post-result':
|
||||
slots[index] = { exec: prepared.exec, result: prepared.result, needsPost: true }
|
||||
break
|
||||
case 'final-result':
|
||||
slots[index] = { exec: prepared.exec, result: prepared.result, needsPost: false }
|
||||
break
|
||||
default:
|
||||
assertNever(prepared, 'tool-call scheduler prepare result')
|
||||
}
|
||||
}
|
||||
const fillPool = async () => {
|
||||
while (!aborted && nextToStart < group.length && inFlight.size < maxParallelToolCalls) {
|
||||
const nextCall = group[nextToStart]
|
||||
if (nextToStart > 0 && mode === 'parallel' && ctx.tools.executionMode(nextCall.exec).kind !== 'parallel') break
|
||||
await startCall(nextToStart)
|
||||
nextToStart++
|
||||
throwSchedulerFailure()
|
||||
await commitReady()
|
||||
throwSchedulerFailure()
|
||||
if (signal.aborted) aborted = true
|
||||
}
|
||||
}
|
||||
try {
|
||||
await fillPool()
|
||||
while (inFlight.size > 0) {
|
||||
const settledIndex = await Promise.race(inFlight.values())
|
||||
inFlight.delete(settledIndex)
|
||||
throwSchedulerFailure()
|
||||
await commitReady()
|
||||
throwSchedulerFailure()
|
||||
if (signal.aborted) aborted = true
|
||||
await fillPool()
|
||||
}
|
||||
} catch (error) {
|
||||
schedulerFailure ??= { error }
|
||||
await Promise.allSettled(inFlight.values())
|
||||
throw schedulerFailure.error
|
||||
}
|
||||
if (aborted) {
|
||||
for (const call of group.slice(started)) appendSkippedToolCall(session, turn, step, call.block)
|
||||
return { consumed: group.length, aborted: true, concluded }
|
||||
}
|
||||
if (committed !== started) throw new Error('tool-call scheduler: uncommitted settled calls')
|
||||
return { consumed: started, aborted: false, concluded }
|
||||
}
|
||||
|
||||
/** Append the durable call/result pair for a model call skipped after cancellation. */
|
||||
function appendSkippedToolCall(session, turn, step, block) {
|
||||
const callSeq = appendToolCall(session, turn, step, block)
|
||||
appendToolResult(
|
||||
session,
|
||||
turn,
|
||||
step,
|
||||
block,
|
||||
{
|
||||
content: [{ type: 'text', text: 'Error: tool call aborted before dispatch' }],
|
||||
isError: true,
|
||||
error: {
|
||||
message: 'tool call aborted before dispatch',
|
||||
info: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
|
||||
},
|
||||
},
|
||||
callSeq,
|
||||
)
|
||||
}
|
||||
|
||||
/** Append a started call and return the event seq that its result must cite. */
|
||||
function appendToolCall(session, turn, step, block) {
|
||||
return session.append('tool/call', { turn, step, callId: block.id, name: block.name, arguments: block.arguments }).seq
|
||||
}
|
||||
|
||||
/** Append a model-ordered result linked to its call event. */
|
||||
function appendToolResult(session, turn, step, block, result, callSeq) {
|
||||
const message = createToolResultMessage({
|
||||
callId: block.id,
|
||||
content: result.content,
|
||||
isError: result.isError,
|
||||
})
|
||||
session.append(
|
||||
'tool/result',
|
||||
{
|
||||
turn,
|
||||
step,
|
||||
message,
|
||||
...(result.error?.info ? { error: result.error.info } : {}),
|
||||
...(result.meta !== undefined ? { meta: result.meta } : {}),
|
||||
},
|
||||
{ surfaceOp: 'append', sourceEventSeqs: [callSeq] },
|
||||
)
|
||||
}
|
||||
@ -11,19 +11,21 @@
|
||||
"files": [
|
||||
"lib/*.js",
|
||||
"lib/loop.js",
|
||||
"lib/loop/*.js",
|
||||
"lib/types/*.d.ts"
|
||||
],
|
||||
"engines": {
|
||||
"node": ">=22.5.0"
|
||||
},
|
||||
"scripts": {
|
||||
"test": "node --test test/store.test.js test/plugin.test.js test/loader.test.js test/phase-b-a.test.js test/phase-b-g.test.js test/phase-b-g-static.test.js test/phase-b-h.test.js test/memory-log.test.js test/loop.test.js",
|
||||
"test": "node --test test/store.test.js test/plugin.test.js test/loader.test.js test/phase-b-a.test.js test/phase-b-g.test.js test/phase-b-g-static.test.js test/phase-b-h.test.js test/memory-log.test.js test/loop.test.js test/loop-ports.test.js",
|
||||
"check": "node scripts/checkpoints.js"
|
||||
},
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/cordis": "^4.0.5",
|
||||
"@deepseek-ai/dsh-agent-loop": "0.2.1-rc.2",
|
||||
"@deepseek-ai/dsh-llm": "0.2.1-rc.2",
|
||||
"@deepseek-ai/dsh-session": "0.2.1-rc.2",
|
||||
"@deepseek-ai/dsh-system-prompt": "0.2.1-rc.2",
|
||||
"@deepseek-ai/dsh-tools": "0.2.1-rc.2"
|
||||
},
|
||||
|
||||
154
test/loop-ports.test.js
Normal file
154
test/loop-ports.test.js
Normal file
@ -0,0 +1,154 @@
|
||||
/**
|
||||
* Contract for the four ported loop modules under `lib/loop/`.
|
||||
*
|
||||
* These files are a faithful port of the stock loop's internals into this
|
||||
* package, so nothing here tests their behaviour — that is `loop.test.js` and
|
||||
* the phase-B suites. What this file pins is the **port interface**, because
|
||||
* every one of these failures is silent at runtime:
|
||||
*
|
||||
* L1 all four modules import on their own (no ordering, no index.js).
|
||||
* L2 the export names a replacement loop will import are present.
|
||||
* L3 `inboxProjectionSchema.parse` actually rejects a malformed persisted
|
||||
* row. This is the ONE thing the port re-implemented instead of copying
|
||||
* (it replaces a zod schema with a hand-written shape check), so it is
|
||||
* exactly the thing that can be a no-op and still look fine.
|
||||
* L4 the inbox projection KEY and `stateVersion` are unchanged. A different
|
||||
* key reads as "no pending input" for sessions written by the stock loop;
|
||||
* a different version makes `stateSchema.parse` refuse those rows. Both
|
||||
* are quiet data loss on `resume`, not a crash.
|
||||
*
|
||||
* @module dsh-memory/test/loop-ports
|
||||
*/
|
||||
|
||||
import assert from 'node:assert/strict'
|
||||
import test from 'node:test'
|
||||
|
||||
// ── L1: each module stands alone ─────────────────────────────────────────
|
||||
//
|
||||
// Imported dynamically rather than at the top of the file so the assertion is
|
||||
// per-module: a single top-level `import` would make one broken port fail the
|
||||
// whole file with "load failed" and hide which of the four broke.
|
||||
const MODULES = {
|
||||
'assistant-stream': '../lib/loop/assistant-stream.js',
|
||||
inbox: '../lib/loop/inbox.js',
|
||||
'runtime-context': '../lib/loop/runtime-context.js',
|
||||
'tool-calls': '../lib/loop/tool-calls.js',
|
||||
}
|
||||
|
||||
test('L1: all four loop ports import successfully on their own', async () => {
|
||||
for (const [name, specifier] of Object.entries(MODULES)) {
|
||||
const mod = await import(specifier)
|
||||
assert.ok(mod !== null && typeof mod === 'object', `${name} must import to a module namespace`)
|
||||
}
|
||||
})
|
||||
|
||||
// ── L2: the export surface a replacement loop depends on ─────────────────
|
||||
//
|
||||
// Named explicitly, not counted: the point is that `executeToolCalls`,
|
||||
// `ReactLoopInbox`, … are still exported under those names, because a rename
|
||||
// here only shows up as `undefined is not a function` at step() time.
|
||||
test('L2: loop ports export the names the replacement loop imports', async () => {
|
||||
const toolCalls = await import(MODULES['tool-calls'])
|
||||
assert.equal(typeof toolCalls.executeToolCalls, 'function', 'tool-calls must export executeToolCalls')
|
||||
|
||||
const assistantStream = await import(MODULES['assistant-stream'])
|
||||
assert.equal(
|
||||
typeof assistantStream.AssistantStreamAttempt,
|
||||
'function',
|
||||
'assistant-stream must export AssistantStreamAttempt',
|
||||
)
|
||||
|
||||
const runtimeContext = await import(MODULES['runtime-context'])
|
||||
assert.equal(
|
||||
typeof runtimeContext.SystemPromptProjection,
|
||||
'function',
|
||||
'runtime-context must export SystemPromptProjection',
|
||||
)
|
||||
assert.equal(
|
||||
typeof runtimeContext.RuntimeContextProjection,
|
||||
'function',
|
||||
'runtime-context must export RuntimeContextProjection',
|
||||
)
|
||||
|
||||
const inbox = await import(MODULES.inbox)
|
||||
assert.equal(typeof inbox.ReactLoopInbox, 'function', 'inbox must export ReactLoopInbox')
|
||||
assert.ok(
|
||||
inbox.inboxProjectionDefinition !== null && typeof inbox.inboxProjectionDefinition === 'object',
|
||||
'inbox must export inboxProjectionDefinition',
|
||||
)
|
||||
assert.ok(
|
||||
inbox.inboxProjectionSchema !== null && typeof inbox.inboxProjectionSchema === 'object',
|
||||
'inbox must export inboxProjectionSchema',
|
||||
)
|
||||
})
|
||||
|
||||
// ── L3: the schema bites ──────────────────────────────────────────────────
|
||||
//
|
||||
// The projection registry calls `stateSchema.parse` on durable checkpoint rows,
|
||||
// so "it has a parse function" proves nothing. These assertions run the parse
|
||||
// against the two shapes that matter: a real state must come back, and a
|
||||
// malformed row must throw — not return undefined, not coerce.
|
||||
test('L3: inboxProjectionSchema.parse accepts a well-formed two-array state', async () => {
|
||||
const { inboxProjectionSchema } = await import(MODULES.inbox)
|
||||
|
||||
const state = {
|
||||
'next-turn': [{ id: 'm1', role: 'user' }],
|
||||
'next-step': [{ id: 'm2', role: 'user' }],
|
||||
}
|
||||
assert.equal(inboxProjectionSchema.parse(state), state, 'parse must return the same object it validated')
|
||||
assert.deepEqual(
|
||||
inboxProjectionSchema.parse({ 'next-turn': [], 'next-step': [] }),
|
||||
{ 'next-turn': [], 'next-step': [] },
|
||||
'the init() shape (both empty) must be accepted',
|
||||
)
|
||||
})
|
||||
|
||||
test('L3: inboxProjectionSchema.parse throws on a malformed persisted row', async () => {
|
||||
const { inboxProjectionSchema } = await import(MODULES.inbox)
|
||||
|
||||
const malformed = [
|
||||
['null', null],
|
||||
['a bare array', []],
|
||||
['a string', 'inbox'],
|
||||
['an object missing "next-step"', { 'next-turn': [] }],
|
||||
['a non-array "next-turn"', { 'next-turn': {}, 'next-step': [] }],
|
||||
['a non-array "next-step"', { 'next-turn': [], 'next-step': 'pending' }],
|
||||
['the stock loop key casing', { nextTurn: [], nextStep: [] }],
|
||||
]
|
||||
for (const [label, value] of malformed) {
|
||||
assert.throws(
|
||||
() => inboxProjectionSchema.parse(value),
|
||||
`parse must reject ${label} — a returning-but-not-throwing parse would accept a corrupt checkpoint`,
|
||||
)
|
||||
}
|
||||
})
|
||||
|
||||
// ── L4: the durable identity of the inbox projection ─────────────────────
|
||||
//
|
||||
// These two literals are a wire format shared with sessions written by the
|
||||
// stock `dsh-agent-loop`. Changing either is silent: the stock loop's pending
|
||||
// splices stop folding in, and the session resumes with an empty inbox.
|
||||
test('L4: inbox projection keeps key "inbox" and stateVersion 1', async () => {
|
||||
const { inboxProjectionDefinition, inboxProjectionSchema } = await import(MODULES.inbox)
|
||||
|
||||
assert.equal(
|
||||
inboxProjectionDefinition.key,
|
||||
'inbox',
|
||||
'the projection key is a wire format shared with stock-loop sessions — changing it loses their pending input',
|
||||
)
|
||||
assert.equal(
|
||||
inboxProjectionDefinition.stateVersion,
|
||||
1,
|
||||
'stateVersion 1 is what existing checkpoint rows carry — a bump makes stateSchema.parse refuse them',
|
||||
)
|
||||
assert.equal(
|
||||
inboxProjectionDefinition.stateSchema,
|
||||
inboxProjectionSchema,
|
||||
'the definition must validate with the exported schema, not a private copy',
|
||||
)
|
||||
assert.deepEqual(
|
||||
inboxProjectionDefinition.init(),
|
||||
{ 'next-turn': [], 'next-step': [] },
|
||||
'init() must produce the shape stateSchema.parse accepts',
|
||||
)
|
||||
})
|
||||
Reference in New Issue
Block a user