Merge remote-tracking branch 'origin/master' into worktree-dynamic-workflows
Master's reconstructable-requests overhaul (#179) meets the workflow tool: - subagent-inprocess structured-output nudge becomes a system-prompt section plus logged context (the injected-request waterfall shape is gone upstream) - snapshot fixtures re-recorded on the merged tree so every request/header carries the workflow tool; authored error-finish/cancel headers patched to the merged tool list and system text - architecture.md condensed back under its word ceiling; module graph regenerated
This commit is contained in:
@@ -56,9 +56,11 @@ forever:
|
||||
drain steering
|
||||
assembly = systemPrompt.assemble({agent}) ⟵ renderPrompt(assembly) IS the full prompt
|
||||
await serial agent/pre-step ⟵ surface mutation (compaction) outside the step
|
||||
session('step/start')
|
||||
request = waterfall agent/request
|
||||
stream llm.stream(request) → session('assistant/chunk')
|
||||
boundary = session.deriveMessages() ⟵ reconstruction boundary: same sync frame,
|
||||
session('step/start') strictly before step/start
|
||||
config = waterfall agent/request ⟵ frozen seed; return a replacement to switch
|
||||
session('request/header'[-delta]) ⟵ the header event this request owes the log
|
||||
stream llm.stream(freeze({header..., messages: boundary})) → session('assistant/chunk')
|
||||
message = waterfall agent/step-result
|
||||
session('assistant/message')
|
||||
each tool-call: session('tool/call')
|
||||
|
||||
@@ -8,10 +8,13 @@
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import type { FinishReason, GenerateOptions, Message } from '@deepseek-ai/dsh-llm'
|
||||
import { BlockAssembler, HarnessError } from '@deepseek-ai/dsh-llm'
|
||||
import type { FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
|
||||
import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
|
||||
import { canonicalHeader } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
|
||||
import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
|
||||
import type { TransmissionLog } from './request-log.ts'
|
||||
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
@@ -155,10 +158,13 @@ export interface LoopHandle {
|
||||
* assembly = ctx.systemPrompt.assemble({agent}) ⟵ waterfall system-prompt/assemble; renderPrompt
|
||||
* (persona section + {{variables}}) IS the full prompt
|
||||
* await ctx.serial('agent/pre-step') ⟵ surface mutation (compaction) OUTSIDE the step
|
||||
* session('step/start') ⟵ durable step boundary (no agent/* mirror)
|
||||
* req = {model, system, tools, messages: session.deriveMessages(), signal}
|
||||
* req = waterfall agent/request ⟵ hooks/model-switch
|
||||
* stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks)
|
||||
* boundary = session.deriveMessages() ⟵ the reconstruction boundary: snapshot in the
|
||||
* session('step/start') same sync frame, strictly before step/start
|
||||
* config = waterfall agent/request(config) ⟵ frozen seed; a returned replacement switches
|
||||
* session('request/header'|'request/header-delta') ⟵ the header event this request owes the
|
||||
* log (initial/resume anchor, delta, fallback)
|
||||
* req = freeze({header..., messages: boundary, sessionId, signal})
|
||||
* stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks, frozen req)
|
||||
* session('assistant/chunk')
|
||||
* msg = waterfall agent/step-result ⟵ BEFORE the log append, so the
|
||||
* session('assistant/message' {content, usage?}) session records what actually ran
|
||||
@@ -181,6 +187,13 @@ export interface LoopHandle {
|
||||
* ```
|
||||
*/
|
||||
export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle): Promise<void> {
|
||||
// Per-instance transmission bookkeeping: whether THIS loop instance has
|
||||
// anchored the log's header fold yet (its first request logs a
|
||||
// 'initial'/'resume' request/header snapshot). Everything else the request
|
||||
// needs is read from the session log itself — the loop holds no
|
||||
// conversation state (the reconstructability RFC).
|
||||
const transmission = createTransmissionLog()
|
||||
|
||||
const { session } = agent
|
||||
|
||||
while (!handle.isDisposed()) {
|
||||
@@ -236,7 +249,7 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
|
||||
// turn number is actually last in the log — a stale counter would collide.
|
||||
const turn = lastTurnNumber(session) + 1
|
||||
try {
|
||||
await runTurn(ctx, agent, handle, turn)
|
||||
await runTurn(ctx, agent, handle, turn, transmission)
|
||||
} catch (error: unknown) {
|
||||
// Backstop: runTurn rethrows only a PRE-turn throw (the invariant guard
|
||||
// before turn/start) — no turn/start was appended, so no turn is open and
|
||||
@@ -271,7 +284,9 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
|
||||
}
|
||||
}
|
||||
|
||||
async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle, turn: number): Promise<void> {
|
||||
async function runTurn(
|
||||
ctx: Context, agent: ReactLoopAgent, handle: LoopHandle, turn: number, transmission: TransmissionLog,
|
||||
): Promise<void> {
|
||||
const { session } = agent
|
||||
|
||||
// --- Pre-turn. A throw here (the invariant guard) is owed NO turn/end —
|
||||
@@ -474,6 +489,17 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
break
|
||||
}
|
||||
|
||||
// The reconstruction boundary (the reconstructability RFC): the request's
|
||||
// messages are snapshotted HERE, in the same synchronous frame as the
|
||||
// step/start append directly below — so the snapshot is exactly the
|
||||
// derivation over the log prefix strictly before step/start's seq.
|
||||
// Anything appended later — by a step/start session/event listener, an
|
||||
// agent/request-window inject(), any concurrent task — lands after the
|
||||
// boundary and joins the NEXT request. An external reconstructor
|
||||
// recovers these exact messages by folding the surface over
|
||||
// events[0..stepStartSeq).
|
||||
const boundaryMessages = session.deriveMessages()
|
||||
|
||||
// Mark the step open BEFORE the append: Session.append pushes the event
|
||||
// to the log before notifying session/event listeners, so a THROWING
|
||||
// step/start listener leaves step/start in the log. Setting stepOpen first
|
||||
@@ -495,7 +521,7 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
|
||||
let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
|
||||
try {
|
||||
stepOutcome = await runStep(ctx, agent, turn, step, assembly, fullSystemPrompt, abort.signal)
|
||||
stepOutcome = await runStep(ctx, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
|
||||
} catch (error: unknown) {
|
||||
stepOutcome = { error: toError(error) }
|
||||
} finally {
|
||||
@@ -644,11 +670,12 @@ function drainSteering(agent: ReactLoopAgent, turn: number): boolean {
|
||||
return messages.length > 0
|
||||
}
|
||||
|
||||
/** One step: derive request from the (already pre-step-mutated) surface →
|
||||
* stream model → record → execute tools. The caller assembles the system prompt
|
||||
* and fires the `agent/pre-step` seam BEFORE opening the step, then passes the
|
||||
* resulting `assembly`/`system` here, so the surface this step derives from
|
||||
* already reflects any compaction. */
|
||||
/** One step: build the request from the boundary snapshot + the step's
|
||||
* header → log the header event the request owes → stream model → record →
|
||||
* execute tools. The caller assembles the system prompt, fires the
|
||||
* `agent/pre-step` seam, snapshots the derivation, and opens the step BEFORE
|
||||
* calling this, so `boundaryMessages` is exactly the surface prefix at
|
||||
* step/start and already reflects any compaction. */
|
||||
async function runStep(
|
||||
ctx: Context,
|
||||
agent: ReactLoopAgent,
|
||||
@@ -656,23 +683,63 @@ async function runStep(
|
||||
step: number,
|
||||
assembly: PromptAssembly,
|
||||
system: string,
|
||||
boundaryMessages: Message[],
|
||||
transmission: TransmissionLog,
|
||||
signal: AbortSignal,
|
||||
): Promise<{ hadToolCalls: boolean; finish: FinishReason }> {
|
||||
const { session, options } = agent
|
||||
|
||||
let request: GenerateOptions = {
|
||||
model: options.model ?? '',
|
||||
messages: session.deriveMessages(),
|
||||
...system ? { system } : {},
|
||||
...assembly.tools.length > 0 ? { tools: assembly.tools } : {},
|
||||
sessionId: session.id,
|
||||
signal,
|
||||
}
|
||||
request = await ctx.waterfall('agent/request', agent, turn, step, request, () => Promise.resolve(request))
|
||||
if (!request.model) {
|
||||
// Seed the call config: the first request of THIS loop instance seeds from
|
||||
// current AgentOptions — explicit options always win over the logged
|
||||
// baseline, which is what keeps fork model-overrides and resume-time
|
||||
// reconfiguration correct. Later steps seed from the log's folded header,
|
||||
// which by then is exactly what this instance last logged.
|
||||
// One deep-cloned, frozen seed serves BOTH the listener chain and the
|
||||
// no-listener fallback: structuredClone decouples it from the session's
|
||||
// cached header fold (a raw reference would let a delegating listener
|
||||
// mutate the fold in place and silently skip the delta log), and the freeze
|
||||
// makes in-place shaping unrepresentable — a switch is a RETURNED
|
||||
// replacement, which the header event below records.
|
||||
const seedConfig: LlmCallConfig = deepFreeze(structuredClone(transmission.loggedHeader
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- loggedHeader ⟹ a snapshot is in the log
|
||||
? session.requestHeader()!.config
|
||||
: { model: options.model ?? '' }))
|
||||
|
||||
// Shape the call config: listeners return a replacement to switch model or
|
||||
// sampling (the seed is frozen — content shaping is not expressible here;
|
||||
// model-visible content flows through the log channels). The header event
|
||||
// below records whatever the request ACTUALLY uses, so a listener's switch
|
||||
// is a logged, reconstructable fact, never silent drift.
|
||||
const config = await ctx.waterfall('agent/request', agent, turn, step, seedConfig, () => Promise.resolve(seedConfig))
|
||||
if (!config.model) {
|
||||
throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
|
||||
}
|
||||
|
||||
// The request header (the log's request/header* vocabulary): canonical form,
|
||||
// recorded before dispatch so the log always explains the request.
|
||||
const header = canonicalHeader({
|
||||
config,
|
||||
...system ? { system } : {},
|
||||
...assembly.tools.length > 0 ? { tools: assembly.tools } : {},
|
||||
})
|
||||
recordRequestHeader(session, transmission, header)
|
||||
|
||||
// Build and freeze: the request is a pure function of (boundary snapshot,
|
||||
// logged header) — llm/stream listeners and adapters read it, mutation
|
||||
// throws. sessionId + frozen is the loop-built marker the dev invariant
|
||||
// keys on.
|
||||
const request: GenerateOptions = deepFreeze({
|
||||
model: header.config.model,
|
||||
messages: boundaryMessages,
|
||||
...header.system !== undefined ? { system: header.system } : {},
|
||||
...header.tools !== undefined ? { tools: header.tools } : {},
|
||||
...header.config.temperature !== undefined ? { temperature: header.config.temperature } : {},
|
||||
...header.config.maxTokens !== undefined ? { maxTokens: header.config.maxTokens } : {},
|
||||
...header.config.stop !== undefined ? { stop: header.config.stop } : {},
|
||||
sessionId: session.id,
|
||||
signal,
|
||||
})
|
||||
|
||||
// --- Model call (streaming-first; raw chunks are the replay record) ---
|
||||
const assembler = new BlockAssembler()
|
||||
const chunkSeqs: number[] = []
|
||||
|
||||
68
packages/core/agent-loop/src/request-log.ts
Normal file
68
packages/core/agent-loop/src/request-log.ts
Normal file
@@ -0,0 +1,68 @@
|
||||
/**
|
||||
* Per-loop-instance transmission bookkeeping for the reconstructability
|
||||
* contract: which header event to append before a request so the session log
|
||||
* always explains the request (the reconstructability RFC). The loop is
|
||||
* otherwise transmission-stateless — the comparison baseline is the log's own
|
||||
* folded header (`Session.requestHeader()`), so resume and fork need no
|
||||
* special path: a fresh loop instance simply logs a `'resume'` snapshot on
|
||||
* its first request and deltas from there.
|
||||
*
|
||||
* @module dsh-agent-loop/request-log
|
||||
*/
|
||||
|
||||
import { diffHeader, headerEquals, applyHeaderDelta } from '@deepseek-ai/dsh-session'
|
||||
import type { EpochHeader, Session } from '@deepseek-ai/dsh-session'
|
||||
|
||||
/** Per-loop-instance bookkeeping: whether THIS instance has logged a header yet. */
|
||||
export interface TransmissionLog {
|
||||
/** True once this loop instance appended its anchoring `request/header` snapshot. */
|
||||
loggedHeader: boolean
|
||||
}
|
||||
|
||||
/** Fresh bookkeeping for a newly-started loop instance. */
|
||||
export function createTransmissionLog(): TransmissionLog {
|
||||
return { loggedHeader: false }
|
||||
}
|
||||
|
||||
/**
|
||||
* Append whatever header event this request owes the log, so folding the log
|
||||
* reproduces the header the request was built under. Exactly one of four
|
||||
* things happens:
|
||||
*
|
||||
* 1. This loop instance has not logged a header yet → a full `request/header`
|
||||
* snapshot anchors the fold: reason `'initial'` when the log has no header
|
||||
* events at all (a new conversation), `'resume'` when it does (process
|
||||
* restart, fork seed — the boundary itself is a recorded fact, so the
|
||||
* snapshot is appended even when nothing changed).
|
||||
* 2. The header equals the folded baseline → nothing; the log already
|
||||
* explains this request.
|
||||
* 3. It differs and the delta round-trips (`applyHeaderDelta` on the baseline
|
||||
* reproduces the header exactly) → a `request/header-delta`.
|
||||
* 4. It differs and the delta encoding cannot express the change (a pure tool
|
||||
* reordering) → a full snapshot with reason `'fallback'`; deltas are an
|
||||
* encoding optimization, never a correctness dependency.
|
||||
*
|
||||
* @param session - the session whose log explains the request.
|
||||
* @param state - this loop instance's bookkeeping (mutated on first log).
|
||||
* @param header - the canonical header the request will ACTUALLY use
|
||||
* (post-`agent/request`).
|
||||
*/
|
||||
export function recordRequestHeader(session: Session, state: TransmissionLog, header: EpochHeader): void {
|
||||
if (!state.loggedHeader) {
|
||||
session.append('request/header', { header, reason: session.requestHeader() === undefined ? 'initial' : 'resume' })
|
||||
state.loggedHeader = true
|
||||
return
|
||||
}
|
||||
// This instance logged a snapshot, so the fold is necessarily defined.
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
|
||||
const baseline = session.requestHeader()!
|
||||
if (headerEquals(baseline, header)) return
|
||||
const delta = diffHeader(baseline, header)
|
||||
/* v8 ignore next -- headerEquals false ⟹ diffHeader defined: both compare the same three parts */
|
||||
if (delta === undefined) return
|
||||
if (headerEquals(applyHeaderDelta(baseline, delta), header)) {
|
||||
session.append('request/header-delta', delta)
|
||||
} else {
|
||||
session.append('request/header', { header, reason: 'fallback' })
|
||||
}
|
||||
}
|
||||
@@ -230,9 +230,8 @@ describe('agent loop', () => {
|
||||
assembly.variables['model'] = 'mock'
|
||||
return next()
|
||||
})
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
|
||||
options.model = 'mock'
|
||||
return next()
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => {
|
||||
return { ...config, model: 'mock' }
|
||||
})
|
||||
const agent = ctx.agentLoop.create(AgentId('a-late-model'), {})
|
||||
|
||||
@@ -428,20 +427,27 @@ describe('agent loop', () => {
|
||||
expect(agent.session.events.some(e => e.type === 'tool/result')).toBe(true)
|
||||
})
|
||||
|
||||
it('agent/request waterfall can rewrite the request (model-switch pattern)', async () => {
|
||||
it('agent/request waterfall switches models by returning a replacement config; the switch is logged', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.llm.registerAdapter(['other-model'], adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
|
||||
options.model = 'other-model'
|
||||
return next()
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => {
|
||||
// The seed is frozen — config is not a mutable per-call knob; a switch
|
||||
// is proposed by returning a replacement, and the loop logs it.
|
||||
expect(Object.isFrozen(config)).toBe(true)
|
||||
expect(() => { (config as { model: string }).model = 'other-model' }).toThrow(TypeError)
|
||||
return { ...config, model: 'other-model' }
|
||||
})
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(adapter.requests[0]!.model).toBe('other-model')
|
||||
// The header event records what the request ACTUALLY used — the switch is
|
||||
// a reconstructable fact, not silent drift.
|
||||
const headerEvent = agent.session.events.find(e => e.type === 'request/header')
|
||||
expect(headerEvent?.type === 'request/header' && headerEvent.data.header.config.model).toBe('other-model')
|
||||
})
|
||||
|
||||
it('agent/pre-step fires once per step before the step is opened', async () => {
|
||||
|
||||
104
packages/core/agent-loop/tests/request-cache.e2e.ts
Normal file
104
packages/core/agent-loop/tests/request-cache.e2e.ts
Normal file
@@ -0,0 +1,104 @@
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { AgentId, type Agent } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
||||
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
|
||||
|
||||
/**
|
||||
* With-key proof that log-derived requests translate into REAL provider cache
|
||||
* hits: a multi-step tool turn (plus a follow-up turn) against the live
|
||||
* DeepSeek API must report `cacheReadTokens > 0` on every request after the
|
||||
* first — the adapter maps the provider's `prompt_cache_hit_tokens`, and the
|
||||
* per-step usage recorded on `assistant/message` events is the production
|
||||
* observable for cache behavior (the reconstructability RFC's measurement
|
||||
* layer: prefix stability is corollary #1). Mocks prove the requests are
|
||||
* append-extensions; only the real API proves those bytes actually hit the
|
||||
* provider cache. Key-gated — skips entirely without $DEEPSEEK_API_KEY.
|
||||
*/
|
||||
|
||||
// Long enough that the shared request prefix comfortably spans the provider's
|
||||
// cache-block granularity (64 tokens) from the very first request.
|
||||
const SYSTEM = 'You are a terse coding assistant used in an automated cache test. '
|
||||
+ 'Always follow instructions literally and exactly. When the user asks you to look '
|
||||
+ 'something up, call the lookup tool with the requested key and wait for its result '
|
||||
+ 'before answering. Never invent a value the tool has not returned. After the tool '
|
||||
+ 'returns, answer with a single short sentence that repeats the returned value '
|
||||
+ 'verbatim. Do not add explanations, do not use markdown, do not ask follow-up '
|
||||
+ 'questions. If the user asks anything else, answer in one short sentence.'
|
||||
|
||||
let ctx: Context | undefined
|
||||
|
||||
afterEach(async () => {
|
||||
await ctx?.fiber.dispose()
|
||||
ctx = undefined
|
||||
})
|
||||
|
||||
async function loopHarness(): Promise<Context> {
|
||||
const created = new Context()
|
||||
await created.plugin(LlmService)
|
||||
await created.plugin(SessionStore)
|
||||
await created.plugin(SystemPrompt, { persona: SYSTEM })
|
||||
await created.plugin(ToolRegistry)
|
||||
await created.plugin(AgentRegistry)
|
||||
await created.plugin(AgentLoop, { agents: [] })
|
||||
await created.plugin(LlmDeepSeek, { models: ['deepseek-v4-flash'] })
|
||||
created.tools.register(defineTool({
|
||||
name: 'lookup',
|
||||
description: 'Look up the stored value for a key.',
|
||||
parameters: { key: { type: 'string', description: 'The key to look up.' } },
|
||||
async execute(args) {
|
||||
return [{ type: 'text', text: `value(${String(args.key)}) = azure-falcon-42` }]
|
||||
},
|
||||
}))
|
||||
return created
|
||||
}
|
||||
|
||||
function waitForIdle(context: Context, agent: Agent): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const dispose = context.on('agent/status', (subject, status) => {
|
||||
if (subject === agent && status === 'idle') {
|
||||
dispose()
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
describe.skipIf(!process.env.DEEPSEEK_API_KEY)('log-derived request cache hits (real API)', () => {
|
||||
it('every request after the first hits the provider prefix cache', async () => {
|
||||
ctx = await loopHarness()
|
||||
const agent = ctx.agentLoop.create(AgentId('cache-e2e'), { model: 'deepseek-v4-flash' })
|
||||
|
||||
// Turn 1: forces a tool call → at least two steps (two model requests).
|
||||
agent.send([{ type: 'text', text: 'Look up the key "deploy-color" with the lookup tool and tell me the value.' }])
|
||||
await waitForIdle(ctx, agent)
|
||||
// Turn 2: a follow-up over the same (longer) prefix.
|
||||
agent.send([{ type: 'text', text: 'Thanks. Repeat that value one more time.' }])
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const usages = [...agent.session.events]
|
||||
.filter(e => e.type === 'assistant/message')
|
||||
.map(e => e.data.usage)
|
||||
expect(usages.length).toBeGreaterThanOrEqual(3) // 2 steps in turn 1 + ≥1 in turn 2
|
||||
for (const usage of usages) expect(usage).toBeDefined()
|
||||
|
||||
// The first request has nothing to hit; every later one shares its
|
||||
// predecessor as a byte-identical prefix, so the provider must report
|
||||
// cached prompt tokens (prompt_cache_hit_tokens → cacheReadTokens).
|
||||
for (const usage of usages.slice(1)) {
|
||||
expect(usage!.cacheReadTokens ?? 0).toBeGreaterThan(0)
|
||||
}
|
||||
|
||||
// World-verification of the conversation itself: the tool value made it
|
||||
// through the loop into the final answer.
|
||||
const finalText = agent.session.deriveMessages().at(-1)!.content
|
||||
.filter(block => block.type === 'text')
|
||||
.map(block => block.text)
|
||||
.join('')
|
||||
expect(finalText).toContain('azure-falcon-42')
|
||||
}, 180_000)
|
||||
})
|
||||
85
packages/core/agent-loop/tests/request-log.spec.ts
Normal file
85
packages/core/agent-loop/tests/request-log.spec.ts
Normal file
@@ -0,0 +1,85 @@
|
||||
/**
|
||||
* recordRequestHeader unit tests: exactly one of four things per request —
|
||||
* an 'initial' snapshot (log has no header yet), a 'resume' snapshot (fresh
|
||||
* loop instance over a log that has one), nothing (header unchanged), a
|
||||
* round-tripping delta, or a 'fallback' snapshot when the delta encoding
|
||||
* cannot express the change (pure tool reordering).
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Session, SessionId, canonicalHeader } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import type { ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
import { createTransmissionLog, recordRequestHeader } from '../src/request-log.ts'
|
||||
|
||||
function tool(name: string, description = 'd'): ToolSchema {
|
||||
return { name, description, parameters: { type: 'object' } }
|
||||
}
|
||||
|
||||
function openSession(id: string): Session {
|
||||
const session = new Session(SessionId(id))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
return session
|
||||
}
|
||||
|
||||
function headerEvents(session: Session): SessionEvent[] {
|
||||
return session.events.filter(e => e.type === 'request/header' || e.type === 'request/header-delta')
|
||||
}
|
||||
|
||||
describe('recordRequestHeader', () => {
|
||||
it("anchors a new conversation with an 'initial' snapshot, then logs nothing while unchanged", () => {
|
||||
const session = openSession('rl-initial')
|
||||
const state = createTransmissionLog()
|
||||
const header = canonicalHeader({ config: { model: 'm' }, system: 's', tools: [tool('t')] })
|
||||
|
||||
recordRequestHeader(session, state, header)
|
||||
const [first] = headerEvents(session)
|
||||
expect(first?.type === 'request/header' && first.data.reason).toBe('initial')
|
||||
|
||||
recordRequestHeader(session, state, header)
|
||||
expect(headerEvents(session)).toHaveLength(1)
|
||||
})
|
||||
|
||||
it("anchors a fresh loop instance over an anchored log with a 'resume' snapshot, even unchanged", () => {
|
||||
const session = openSession('rl-resume')
|
||||
const header = canonicalHeader({ config: { model: 'm' }, system: 's' })
|
||||
recordRequestHeader(session, createTransmissionLog(), header)
|
||||
|
||||
// A second instance (process restart / fork): the boundary itself is a
|
||||
// recorded fact — snapshot appended even though the header is identical.
|
||||
recordRequestHeader(session, createTransmissionLog(), header)
|
||||
const events = headerEvents(session)
|
||||
expect(events).toHaveLength(2)
|
||||
expect(events[1]?.type === 'request/header' && events[1].data.reason).toBe('resume')
|
||||
})
|
||||
|
||||
it('logs a round-tripping delta for a mid-run change, and the fold reproduces the header', () => {
|
||||
const session = openSession('rl-delta')
|
||||
const state = createTransmissionLog()
|
||||
const first = canonicalHeader({ config: { model: 'm' }, system: 'a\nb', tools: [tool('t')] })
|
||||
recordRequestHeader(session, state, first)
|
||||
|
||||
const second = canonicalHeader({ config: { model: 'm' }, system: 'a\nc', tools: [tool('t'), tool('u')] })
|
||||
recordRequestHeader(session, state, second)
|
||||
const events = headerEvents(session)
|
||||
expect(events).toHaveLength(2)
|
||||
expect(events[1]?.type).toBe('request/header-delta')
|
||||
expect(session.requestHeader()).toEqual(second)
|
||||
})
|
||||
|
||||
it("records a change the delta cannot express (pure reordering) as a 'fallback' snapshot", () => {
|
||||
const session = openSession('rl-fallback')
|
||||
const state = createTransmissionLog()
|
||||
const first = canonicalHeader({ config: { model: 'm' }, tools: [tool('a'), tool('b')] })
|
||||
recordRequestHeader(session, state, first)
|
||||
|
||||
const reordered = canonicalHeader({ config: { model: 'm' }, tools: [tool('b'), tool('a')] })
|
||||
recordRequestHeader(session, state, reordered)
|
||||
const events = headerEvents(session)
|
||||
expect(events).toHaveLength(2)
|
||||
expect(events[1]?.type === 'request/header' && events[1].data.reason).toBe('fallback')
|
||||
// The fold still lands on the exact header — deltas are an encoding
|
||||
// optimization, never a correctness dependency.
|
||||
expect(session.requestHeader()).toEqual(reordered)
|
||||
})
|
||||
})
|
||||
313
packages/core/agent-loop/tests/request-reconstruction.spec.ts
Normal file
313
packages/core/agent-loop/tests/request-reconstruction.spec.ts
Normal file
@@ -0,0 +1,313 @@
|
||||
/**
|
||||
* Loop-level reconstructability: every request the loop sends is a pure
|
||||
* function of the session log — messages are the derivation at the step/start
|
||||
* boundary, the header is the fold of request/header* events — and every
|
||||
* request is an append-extension of its predecessor unless a logged event
|
||||
* (compaction replace, header change) explains the difference. The requests
|
||||
* recorded by the mock adapter are the observable; the offline-rebuild test
|
||||
* at the bottom is the theorem stated end-to-end.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateOptions } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { Session, SessionId, foldRequestHeader } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
|
||||
|
||||
async function harness(adapter: MockAdapter, persona = 'stable base') {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt, { persona })
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
return ctx
|
||||
}
|
||||
|
||||
function waitForIdle(ctx: Context, agent: ReactLoopAgent): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const dispose = ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent && status === 'idle') {
|
||||
dispose()
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
function send(agent: ReactLoopAgent, text: string) {
|
||||
agent.send([{ type: 'text', text }])
|
||||
}
|
||||
|
||||
/** Assert `previous` is a strict value-prefix of `current`. */
|
||||
function expectPrefixExtension(previous: GenerateOptions, current: GenerateOptions) {
|
||||
expect(current.messages.length).toBeGreaterThan(previous.messages.length)
|
||||
expect(current.messages.slice(0, previous.messages.length)).toEqual([...previous.messages])
|
||||
expect(current.system).toEqual(previous.system)
|
||||
expect(current.tools).toEqual(previous.tools)
|
||||
}
|
||||
|
||||
function registerEcho(ctx: Context) {
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'echo',
|
||||
description: 'echo back',
|
||||
parameters: { text: { type: 'string' } },
|
||||
async execute(args) {
|
||||
return [{ type: 'text', text: `echo: ${String(args.text)}` }]
|
||||
},
|
||||
}))
|
||||
}
|
||||
|
||||
describe('request stability across the loop', () => {
|
||||
it('each step request within a turn append-extends the previous, frozen end to end', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', { text: 'one' }, 'first'),
|
||||
toolCallResponse('c2', 'echo', { text: 'two' }, 'second'),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
registerEcho(ctx)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(3)
|
||||
expectPrefixExtension(adapter.requests[0]!, adapter.requests[1]!)
|
||||
expectPrefixExtension(adapter.requests[1]!, adapter.requests[2]!)
|
||||
for (const request of adapter.requests) {
|
||||
expect(Object.isFrozen(request)).toBe(true)
|
||||
expect(Object.isFrozen(request.messages)).toBe(true)
|
||||
}
|
||||
// One anchoring header snapshot; no further header events (nothing changed).
|
||||
const headerEvents = agent.session.events.filter(e => e.type === 'request/header' || e.type === 'request/header-delta')
|
||||
expect(headerEvents).toHaveLength(1)
|
||||
expect(headerEvents[0]?.type === 'request/header' && headerEvents[0].data.reason).toBe('initial')
|
||||
})
|
||||
|
||||
it('a later turn append-extends the previous turn (one conversation, one log)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
expectPrefixExtension(adapter.requests[0]!, adapter.requests[1]!)
|
||||
})
|
||||
|
||||
it('a compaction replace rewrites the resend, and the log explains it', async () => {
|
||||
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// A pre-step listener compacts turn 1's history before turn 2's step —
|
||||
// the sanctioned surface rewrite, landing OUTSIDE the step.
|
||||
const preStep = ctx.on('agent/pre-step', () => {
|
||||
preStep()
|
||||
const session = agent.session
|
||||
const nodes = session.surface.nodes
|
||||
session.append('context/message', {
|
||||
content: [{ type: 'text', text: '[summary of turn 1]' }],
|
||||
source: { kind: 'plugin', plugin: 'test-compact' },
|
||||
}, {
|
||||
surfaceOp: { op: 'replace', start: nodes[0]!.seq, end: nodes[1]!.seq },
|
||||
sourceEventSeqs: [nodes[0]!.seq, nodes[1]!.seq],
|
||||
})
|
||||
})
|
||||
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const second = adapter.requests[1]!
|
||||
// The rewritten history: summary replaces turn 1's user+assistant pair.
|
||||
expect(second.messages[0]!.content.some(b => b.type === 'text' && b.text.includes('[summary of turn 1]'))).toBe(true)
|
||||
// No header event beyond the anchor: the replace is itself in the log.
|
||||
expect(agent.session.events.filter(e => e.type === 'request/header')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('a real system-prompt change is a logged header delta; a stable prompt logs nothing', async () => {
|
||||
const adapter = new MockAdapter([textResponse('one'), textResponse('two'), textResponse('three')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
// Identical assembly re-rendered per step is NOT a change.
|
||||
expect(agent.session.events.filter(e => e.type === 'request/header-delta')).toHaveLength(0)
|
||||
|
||||
ctx.systemPrompt.section({ name: 'extra', order: 2, text: 'new guidance' })
|
||||
send(agent, 'third')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const deltas = agent.session.events.filter(e => e.type === 'request/header-delta')
|
||||
expect(deltas).toHaveLength(1)
|
||||
expect(adapter.requests[2]!.system).toContain('new guidance')
|
||||
// History is preserved across the change — only the header moved.
|
||||
expect(adapter.requests[2]!.messages.length).toBeGreaterThan(adapter.requests[1]!.messages.length)
|
||||
})
|
||||
|
||||
it('an inject() during the agent/request waterfall joins the NEXT request (the step/start boundary)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let injected = false
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, _config, next) => {
|
||||
if (!injected) {
|
||||
injected = true
|
||||
agent.inject([{ type: 'text', text: '[late context]' }], { source: { kind: 'plugin', plugin: 'test' } })
|
||||
}
|
||||
return next()
|
||||
})
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
const first = adapter.requests[0]!
|
||||
// The inject landed in the log after the boundary: not in THIS request…
|
||||
expect(first.messages.some(m => m.content.some(b => b.type === 'text' && b.text.includes('[late context]')))).toBe(false)
|
||||
expect(agent.session.events.some(e => e.type === 'context/message')).toBe(true)
|
||||
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
// …but in the next one, at its logged position.
|
||||
const second = adapter.requests[1]!
|
||||
expect(second.messages.some(m => m.content.some(b => b.type === 'text' && b.text.includes('[late context]')))).toBe(true)
|
||||
})
|
||||
|
||||
it('a mutation attempt on the frozen request content throws into the step (loud, not silent)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('one')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
|
||||
ctx.on('llm/stream', (options, next) => {
|
||||
// The historical failure mode this design kills: a listener rewriting
|
||||
// request content in place. The freeze turns it into a loud error.
|
||||
options.messages.push({ role: 'user', content: [{ type: 'text', text: 'sneaky' }] })
|
||||
return next()
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(errors).toHaveLength(1)
|
||||
expect(errors[0]!.message).toMatch(/not extensible|frozen|read only|readonly/i)
|
||||
})
|
||||
|
||||
it('a fresh loop instance over a seeded log anchors with a resume snapshot and stays cache-aligned', async () => {
|
||||
const adapter = new MockAdapter([textResponse('one')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('gen1'), { model: 'mock' })
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// Second generation: a new agent whose session is seeded with the first
|
||||
// one's full log (the resume/fork path).
|
||||
const adapter2 = new MockAdapter([textResponse('two')])
|
||||
const ctx2 = await harness(adapter2)
|
||||
const handle = ctx2.agents.create({
|
||||
agentId: AgentId('gen2'),
|
||||
sessionId: SessionId('gen2-session'),
|
||||
seed: [...agent.session.events],
|
||||
agentOptions: { model: 'mock' },
|
||||
})
|
||||
const agent2 = handle.agent as ReactLoopAgent
|
||||
send(agent2, 'second')
|
||||
await waitForIdle(ctx2, agent2)
|
||||
|
||||
const snapshots = agent2.session.events.filter(e => e.type === 'request/header')
|
||||
expect(snapshots).toHaveLength(2)
|
||||
expect(snapshots[1]?.type === 'request/header' && snapshots[1].data.reason).toBe('resume')
|
||||
// Identical header across the restart: byte-identical continuation.
|
||||
expect(adapter2.requests[0]!.system).toEqual(adapter.requests[0]!.system)
|
||||
expectPrefixExtension(adapter.requests[0]!, adapter2.requests[0]!)
|
||||
})
|
||||
|
||||
it('a delegating listener cannot mutate the seed through next() — the fold stays log-true', async () => {
|
||||
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, _config, next) => {
|
||||
const config = await next()
|
||||
// next() resolves the SAME frozen seed — in-place shaping after
|
||||
// delegation is unrepresentable, so a "mutate what next() returned"
|
||||
// listener cannot desync the log from the request (nor reach the
|
||||
// session's cached header fold, which is deep-cloned away and itself
|
||||
// frozen).
|
||||
expect(Object.isFrozen(config)).toBe(true)
|
||||
expect(() => { (config as { temperature?: number }).temperature = 0.9 }).toThrow(TypeError)
|
||||
return config
|
||||
})
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// No delta was logged (nothing really changed), and the session's own
|
||||
// fold is immutable state.
|
||||
expect(agent.session.events.filter(e => e.type === 'request/header-delta')).toHaveLength(0)
|
||||
expect(Object.isFrozen(agent.session.requestHeader())).toBe(true)
|
||||
expect(adapter.requests[1]!.temperature).toBeUndefined()
|
||||
})
|
||||
|
||||
it('THEOREM: every request rebuilds byte-equal from the session log alone', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', { text: 'one' }, 'calling'),
|
||||
textResponse('done'),
|
||||
textResponse('after change'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
registerEcho(ctx)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
ctx.systemPrompt.section({ name: 'extra', order: 2, text: 'now with guidance' })
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => ({ ...config, temperature: 0.5, maxTokens: 99, stop: ['<END>'] }))
|
||||
send(agent, 'again')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(3)
|
||||
const events = agent.session.events
|
||||
const stepStarts = events.filter(e => e.type === 'step/start')
|
||||
expect(stepStarts).toHaveLength(3)
|
||||
|
||||
adapter.requests.forEach((request, index) => {
|
||||
const stepStart = stepStarts[index]!
|
||||
// Messages: the derivation over the log prefix strictly before this
|
||||
// step's step/start — rebuilt here through a completely fresh Session.
|
||||
const rebuilt = new Session(SessionId(`rebuild-${index}`), structuredClone(events.slice(0, stepStart.seq)))
|
||||
expect(structuredClone(request.messages)).toEqual(rebuilt.deriveMessages())
|
||||
|
||||
// Header: the fold of request/header* events up to this step's dispatch
|
||||
// (its header event sits between step/start and the first chunk).
|
||||
const firstChunk = events.find(e => e.type === 'assistant/chunk' && e.seq > stepStart.seq)!
|
||||
const header = foldRequestHeader(events.slice(0, firstChunk.seq))!
|
||||
expect(request.model).toBe(header.config.model)
|
||||
expect(request.system).toEqual(header.system)
|
||||
expect(structuredClone(request.tools ?? [])).toEqual(structuredClone(header.tools ?? []))
|
||||
expect(request.temperature).toBe(header.config.temperature)
|
||||
expect(request.maxTokens).toBe(header.config.maxTokens)
|
||||
expect(request.stop).toEqual(header.config.stop)
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -399,9 +399,8 @@ describe('MEDIUM: misc registry and config fixes', () => {
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), {}) // no model — router plugin decides
|
||||
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
|
||||
options.model = 'mock'
|
||||
return next()
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => {
|
||||
return { ...config, model: 'mock' }
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
|
||||
@@ -44,7 +44,7 @@ Turn and step boundaries are NOT mirrored as `agent/*` emits: a consumer that ne
|
||||
- `agent/session-start` (emit) — fired once before the first turn; a listener seeds context via `agent.inject()` (it cannot veto startup).
|
||||
- `agent/prompt-submit` — decide what happens to one drained queued message before it becomes a `user/message`: `PromptDecision` = `allow` (optionally rewriting the prompt `content` or attaching `additionalContext`) or `block` (drop it; a batch whose every prompt is blocked opens a zero-step turn that ends `rejected`). Maps onto Claude Code's `UserPromptSubmit`.
|
||||
- `agent/pre-step` (serial) — mutate the session surface before the step opens and history is derived (compaction). Fires after `turn/start` and before `step/start`, so a listener's appended events land outside the step.
|
||||
- `agent/request` — mutate `GenerateOptions` before the model call (hooks, model switching, tool filtering)
|
||||
- `agent/request` — shape the call config before the model call: a frozen `LlmCallConfig` seed in, a replacement out (model switching, sampling overrides). Content is not shapeable here — every request is a pure function of the session log ([reconstructability RFC](../../../docs/rfc/implemented/architecture/2026-07-05-reconstructable-requests.md)); the loop logs whatever config the request actually uses as a `request/header*` event
|
||||
- `agent/step-result` — post-process the assembled assistant message before tool dispatch (validates what the log records)
|
||||
- `agent/turn-continuation` — override the continue/stop decision via `ContinuationDecision` = `{action:'stop'}` or `{action:'continue', reason?}` (a `continue` `reason` is recorded as next-step steering in the same turn — the typed `/goal` pattern). Force-continue `/loop`, force-stop budget guard.
|
||||
|
||||
|
||||
@@ -44,7 +44,7 @@
|
||||
*/
|
||||
|
||||
import type { Branded } from '@deepseek-ai/dsh-brand'
|
||||
import type { ContentBlock, GenerateOptions, Message, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock, LlmCallConfig, Message, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import type {} from '@deepseek-ai/dsh-system-prompt'
|
||||
|
||||
/** Identifies one live agent in the registry. */
|
||||
@@ -345,18 +345,28 @@ declare module 'cordis' {
|
||||
*/
|
||||
'agent/prompt-submit'(agent: Agent, content: ContentBlock[], source: MessageSource, next: () => Promise<PromptDecision>): Promise<PromptDecision>
|
||||
/**
|
||||
* Waterfall: mutate the fully-assembled {@link GenerateOptions} before the
|
||||
* model call (hooks, model switching, tool filtering, …). Call `next()` to
|
||||
* delegate, or return without it to short-circuit. For surface mutation that
|
||||
* must precede history derivation (compaction), use {@link agent/pre-step}
|
||||
* instead — by the time this fires, `options.messages` is already derived.
|
||||
* Waterfall: shape the step's call configuration — model switching,
|
||||
* sampling overrides — by returning a replacement {@link LlmCallConfig}
|
||||
* (the frozen seed is the config the loop would otherwise use). Config is
|
||||
* ALL a listener shapes here: every request is a pure function of the
|
||||
* session log (the reconstructability RFC), so model-visible content
|
||||
* flows through the log channels — `inject()`, steering, prompt-submit
|
||||
* `additionalContext`, prompt sections via `system-prompt/assemble` —
|
||||
* never through request mutation, and the loop records whatever config
|
||||
* the request actually uses as a `request/header*` event before dispatch.
|
||||
* The step's messages are already snapshotted when this fires (the
|
||||
* `step/start` boundary): an `inject()` from a listener here lands in the
|
||||
* log but joins the NEXT request. For surface mutation that must precede
|
||||
* the snapshot (compaction), use {@link agent/pre-step}. Call `next()` to
|
||||
* delegate, or return an {@link LlmCallConfig} without it to
|
||||
* short-circuit.
|
||||
* @param agent - the agent making the model call.
|
||||
* @param turn - the open turn number.
|
||||
* @param step - the step whose request this is.
|
||||
* @param options - the assembled request; listeners return a transformed copy.
|
||||
* @param config - the config the loop would use (frozen); return a replacement to switch.
|
||||
* @mode waterfall
|
||||
*/
|
||||
'agent/request'(agent: Agent, turn: number, step: number, options: GenerateOptions, next: () => Promise<GenerateOptions>): Promise<GenerateOptions>
|
||||
'agent/request'(agent: Agent, turn: number, step: number, config: LlmCallConfig, next: () => Promise<LlmCallConfig>): Promise<LlmCallConfig>
|
||||
/**
|
||||
* Waterfall: post-process the assembled assistant {@link Message} before
|
||||
* tool dispatch (validation, content rewriting, …).
|
||||
|
||||
@@ -35,8 +35,9 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall
|
||||
Plain class (not a Cordis Service). Create via `ctx.sessions.create()`.
|
||||
|
||||
- `session.append(type, data, opts?): SessionEvent` — synchronous, never blocks on I/O. **Throws** if `data` is not losslessly JSON-serializable (BigInt, function, symbol, undefined, non-finite number, circular ref, or an exotic object like Map/Set/Date) — the event log is the durable source of truth, so this invariant is enforced at the source (exported as `isJsonValue` for backends to reuse on their replay/fork entry points). A third parameter `opts: SurfaceIntent` carries surface metadata: `surfaceOp` controls how the event enters the surface linked list, and `sourceEventSeqs` records provenance (the seq numbers of events this one derives from). It is **required** for the five `SurfaceEventType` events (every message-producing event must declare how it joins the surface) and rejected by the compiler for non-surface types. The marker requirement is enforced two ways: the typed overload makes `opts` mandatory when `type` is a specific `SurfaceEventType` literal, AND `append` **throws** at runtime if a surface-eligible event arrives with no `surfaceOp` — covering the case where `type` widens to the `SessionEventType` union (a caller iterating raw events, where the conditional overload collapses to optional) so a marker-less message event can never silently land in the log and vanish from `deriveMessages()`.
|
||||
- `session.deriveMessages(): Message[]` — derive the LLM message history by walking the surface linked list (skipping non-surface events like chunks and boundaries; a `replace` shadows the nodes it covers). The surface is the single source of derived history — there is no raw-log fallback.
|
||||
- `session.surface: SurfaceManager` — the derived surface, lazily rebuilt from `surfaceOp` markers in the log. Processes only new events (delta) on each access — the log is append-only, so prior events never change.
|
||||
- `session.deriveMessages(): Message[]` — the LLM message history, CACHED: each surface node is projected exactly once, when first seen (O(new nodes) per call; a surface rewrite rebuilds via `surface.replaceGeneration`). Returns a fresh array snapshot per call over SHARED, deep-frozen `Message` objects — cloned once off the log at projection time, so a consumer can never mutate logged data (mutation throws). The surface is the single source of derived history — there is no raw-log fallback.
|
||||
- `session.deriveEventMessage(event): Message | null` — the per-event projection `deriveMessages()` folds: one event's derived message (an unfrozen clone), or `null` when it produces none (a non-surface event, or an empty-content `assistant/message` hosting only usage). External reconstructors and the dev invariant fold the same function over a log prefix's surface, so no two paths can disagree about what a request's messages were (the reconstructability RFC).
|
||||
- `session.surface: SurfaceManager` — the derived surface, lazily rebuilt from `surfaceOp` markers in the log. Processes only new events (delta) on each access — the log is append-only, so prior events never change. `surface.replaceGeneration` is the rewrite signal: bumped by every folded `replace` and by `invalidate()`, never reset, so an incremental consumer comparing generations cannot be fooled.
|
||||
- `session.events`, `session.seq`, `session.id`
|
||||
- `session.header: SessionHeader` — immutable creation metadata (`version`, `id`, `createdAt`, optional `cwd`/`parentSession`/`seedLength`). Kept out of the event log (a storage concern, not replayable state); a minimal header (stamped with the current `SESSION_FORMAT_VERSION`) is synthesized for bare `Session` construction.
|
||||
|
||||
@@ -47,6 +48,10 @@ Plain class (not a Cordis Service). Create via `ctx.sessions.create()`.
|
||||
- `SurfaceNode` — `{ seq: number; prev: number | null; next: number | null }`, one node in the surface linked list.
|
||||
- `isSurfaceEvent(event)` / `isSurfaceEligibleType(type)` — the first narrows a `SessionEvent` to a fully-formed surface node (type is surface-eligible AND `surfaceOp` present); the second is the type-only check (is this one of the five `SurfaceEventType` values?), used to detect a surface-eligible event MISSING its marker — e.g. when validating a seed/load log.
|
||||
|
||||
### Request-header reconstruction (`request-header.ts`)
|
||||
|
||||
The `request/header` (full `EpochHeader` snapshot with a `RequestHeaderReason`) and `request/header-delta` (system line-trim / name-keyed tools delta / whole config) events make the request envelope logged session state, so every conversation request is a pure function of the log. The pure trio reconstructs it: `foldRequestHeader(events)` folds a log (or any prefix) into the header in force; `diffHeader(prev, next)` encodes a change (undefined when equal); `applyHeaderDelta(prev, delta)` replays one. Writer contract: every logged delta is round-trip-verified (`apply(prev, delta)` deep-equals the new header) with a `'fallback'` snapshot when the encoding cannot express the change (a pure tool reordering), so folding never needs error recovery on a well-formed log. `canonicalHeader` pins the one representation of absence (empty system/tools ≡ absent fields).
|
||||
|
||||
### Session event vocabulary (`types.ts`)
|
||||
|
||||
The append-only log's event types, enumerated member by member — payloads, surface badges, provenance — in the generated [persistence log event catalog](../../../docs/persistence-catalog/log-events.md). Token usage rides on `assistant/message.usage`; an operational error's step is on `turn/end.reason` for `kind: 'error'`.
|
||||
|
||||
@@ -8,11 +8,13 @@
|
||||
|
||||
import { Context, Service } from 'cordis'
|
||||
import { isAbsolute } from 'node:path'
|
||||
import { deepFreeze } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock, Message, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import { SESSION_FORMAT_VERSION, SessionId } from './types.ts'
|
||||
import type { CreateSessionOptions, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts'
|
||||
import type { CreateSessionOptions, EpochHeader, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts'
|
||||
import { isJsonValue } from './json.ts'
|
||||
import { SurfaceManager, isSurfaceEligibleType } from './surface.ts'
|
||||
import { foldRequestHeader } from './request-header.ts'
|
||||
|
||||
export * from './types.ts'
|
||||
export { isJsonValue } from './json.ts'
|
||||
@@ -21,6 +23,7 @@ export { interruptedTurnClosers } from './repair.ts'
|
||||
export type { SurfaceNode } from './surface.ts'
|
||||
export { isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
|
||||
export { isToolPairingBalanced } from './tool-pairing.ts'
|
||||
export { applyHeaderDelta, canonicalHeader, diffHeader, foldRequestHeader, headerEquals } from './request-header.ts'
|
||||
|
||||
declare module 'cordis' {
|
||||
interface Context {
|
||||
@@ -234,53 +237,93 @@ export class Session {
|
||||
return event
|
||||
}
|
||||
|
||||
/** Cached fold of the request-header events — see {@link requestHeader}. */
|
||||
private headerFold: EpochHeader | undefined
|
||||
/** Log position (events consumed) the header fold has reached. */
|
||||
private headerFoldSeq = 0
|
||||
|
||||
/**
|
||||
* The {@link EpochHeader} in force after the log's last header event — the
|
||||
* header the NEXT request will be compared against — or undefined before
|
||||
* the first `request/header` snapshot. The live, incrementally-maintained
|
||||
* form of `foldRequestHeader(session.events)`: each header event is folded
|
||||
* once, when first seen, so a per-step read costs O(new events).
|
||||
* @returns the folded header, or undefined when no header event exists yet.
|
||||
*/
|
||||
requestHeader(): EpochHeader | undefined {
|
||||
if (this.headerFoldSeq < this.log.length) {
|
||||
// Frozen on update: the fold is session state exposed by reference — a
|
||||
// consumer mutating it in place (instead of building a replacement)
|
||||
// would desync every later comparison against the log, so mutation
|
||||
// throws instead.
|
||||
this.headerFold = deepFreeze(foldRequestHeader(this.log.slice(this.headerFoldSeq), this.headerFold))
|
||||
this.headerFoldSeq = this.log.length
|
||||
}
|
||||
return this.headerFold
|
||||
}
|
||||
|
||||
/** The derived-message cache: frozen projections, extended per unseen node. */
|
||||
private derived: Message[] = []
|
||||
/** Surface position (nodes projected) the cache has reached. */
|
||||
private derivedNodes = 0
|
||||
/** {@link SurfaceManager.replaceGeneration} the cache was built under. */
|
||||
private derivedGeneration = 0
|
||||
|
||||
/**
|
||||
* Derive the LLM message history by walking the session surface — the linked
|
||||
* list of message-producing events maintained by `surfaceOp` markers. The
|
||||
* surface is the single source of derived history: every message-producing
|
||||
* append records its `surfaceOp`, so a raw event with no marker (a chunk, a
|
||||
* turn boundary) is correctly absent, and a compaction `replace` deletes the
|
||||
* shadowed nodes from the derivation.
|
||||
* shadowed nodes from the derivation. The projection rules are
|
||||
* {@link deriveEventMessage}, folded per node.
|
||||
*
|
||||
* - `user/message` → user message
|
||||
* - `assistant/message` → assistant message (chunks are skipped — they are
|
||||
* replay/UI data; the assembled message is authoritative for history). An
|
||||
* EMPTY-content assistant/message is skipped: a max-tokens step cut off with
|
||||
* no content still records an assistant/message to host its `usage`, but a
|
||||
* content-less assistant turn must not enter the provider transcript.
|
||||
* - `tool/result` → user message carrying a tool-result block
|
||||
* - `context/message` / `steering/message` → tagged synthetic user messages
|
||||
* at their chronological position
|
||||
*
|
||||
* The returned `content` is **deep-cloned** off the logged events: the loop
|
||||
* hands these messages into the mutable `agent/request` waterfall and on to
|
||||
* adapters, where mutating the request is sanctioned — but the session log
|
||||
* is append-only by contract. Cloning at this boundary keeps in-flight
|
||||
* mutation from reaching back and rewriting history (which would silently
|
||||
* break replay equivalence). Cost is one structured clone per step,
|
||||
* negligible next to a model call.
|
||||
* CACHED: each surface node is projected exactly once, when first seen — a
|
||||
* call costs O(new nodes), and a surface rewrite (a `replace`;
|
||||
* {@link SurfaceManager.replaceGeneration}) rebuilds. The returned array is
|
||||
* a fresh snapshot per call (later appends never grow an array a caller
|
||||
* already holds); the `Message` objects in it are SHARED and **deep-frozen**
|
||||
* — cloned once off the log at projection time, so consumers can never
|
||||
* mutate logged data, and mutation attempts throw instead of silently
|
||||
* diverging replay from history.
|
||||
* @returns a fresh array of the shared, frozen derived history.
|
||||
*/
|
||||
deriveMessages(): Message[] {
|
||||
const messages: Message[] = []
|
||||
for (const node of this.surface.nodes) {
|
||||
const nodes = this.surface.nodes
|
||||
const generation = this.surface.replaceGeneration
|
||||
if (generation !== this.derivedGeneration) {
|
||||
this.derived = []
|
||||
this.derivedNodes = 0
|
||||
this.derivedGeneration = generation
|
||||
}
|
||||
for (const node of nodes.slice(this.derivedNodes)) {
|
||||
// Surface nodes are built from this.log — node.seq is always a valid
|
||||
// index by construction. The non-null assertion expresses that invariant.
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
|
||||
const msg = this._deriveOneMessage(this.log[node.seq]!)
|
||||
const msg = this.deriveEventMessage(this.log[node.seq]!)
|
||||
// A surface node is one of the five message-producing types, but an
|
||||
// empty-content assistant/message (a max-tokens step that hosts only
|
||||
// usage) derives to null and must not enter the transcript.
|
||||
if (msg) messages.push(msg)
|
||||
if (msg) this.derived.push(deepFreeze(msg))
|
||||
}
|
||||
return messages
|
||||
this.derivedNodes = nodes.length
|
||||
return [...this.derived]
|
||||
}
|
||||
|
||||
/**
|
||||
* Derive a single LLM message from one surface event, or null if it produces
|
||||
* no message (an empty-content assistant/message that exists only to host
|
||||
* usage).
|
||||
* Project a single event into the LLM message it derives to, or null when
|
||||
* it produces none — a non-surface event (chunk, boundary, log-only record)
|
||||
* or an empty-content assistant/message (which exists only to host usage).
|
||||
* The per-node pure function {@link deriveMessages} folds over the surface;
|
||||
* an external reconstructor (or the dev invariant) folds the same function
|
||||
* over a log prefix's surface to rebuild the exact messages any request was
|
||||
* built from (the reconstructability RFC). The returned `content` is
|
||||
* deep-cloned off the logged event: the log is append-only by contract, so
|
||||
* no live reference to logged data leaves this boundary.
|
||||
* @param event - the event to project.
|
||||
* @returns the derived message, or null when the event produces none.
|
||||
*/
|
||||
private _deriveOneMessage(event: SessionEvent): Message | null {
|
||||
deriveEventMessage(event: SessionEvent): Message | null {
|
||||
// Intentionally non-exhaustive: only message-producing events derive
|
||||
// history; turn/step boundaries, chunks, usage, and errors are
|
||||
// trace/replay data.
|
||||
@@ -311,8 +354,9 @@ export class Session {
|
||||
const { content, source } = event.data
|
||||
return { role: 'user', content: renderTagged('steering', structuredClone(content), source) }
|
||||
}
|
||||
/* v8 ignore next 2 -- unreachable: only surface nodes (the 5 message-producing types) reach here */
|
||||
default:
|
||||
// A non-surface event (boundary, chunk, log-only record) projects to
|
||||
// no message. Merge-extensible union: no assertNever here.
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
191
packages/core/session/src/request-header.ts
Normal file
191
packages/core/session/src/request-header.ts
Normal file
@@ -0,0 +1,191 @@
|
||||
/**
|
||||
* Request-header reconstruction utilities: the pure fold/diff/apply trio over
|
||||
* the `request/header` / `request/header-delta` session events. Anyone
|
||||
* holding a session log reconstructs the {@link EpochHeader} any request was
|
||||
* built under by folding these events in log order; the loop uses the same
|
||||
* functions to decide whether a step's header changed and to encode the
|
||||
* change. Deltas are an encoding optimization with a safety valve — the
|
||||
* writer round-trip-verifies every delta before appending and falls back to
|
||||
* a full snapshot when the encoding cannot express the change — so folding
|
||||
* never needs error recovery on a well-formed log.
|
||||
*
|
||||
* @module dsh-session/request-header
|
||||
*/
|
||||
|
||||
import { callConfigEquals } from '@deepseek-ai/dsh-llm'
|
||||
import type { LlmCallConfig, ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
import type { EpochHeader, SessionEvent, SystemDelta, ToolsDelta } from './types.ts'
|
||||
|
||||
/**
|
||||
* Normalize a header to canonical form: an empty system prompt and an empty
|
||||
* tool list become ABSENT fields, matching how requests are built (both
|
||||
* request-build spreads skip empty values). Diff, fold, and comparison all
|
||||
* operate on canonical headers, so "no system prompt" has exactly one
|
||||
* representation.
|
||||
* @param header - the header to normalize (not mutated).
|
||||
* @returns the canonical header.
|
||||
*/
|
||||
export function canonicalHeader(header: EpochHeader): EpochHeader {
|
||||
return {
|
||||
config: header.config,
|
||||
...header.system !== undefined && header.system.length > 0 ? { system: header.system } : {},
|
||||
...header.tools !== undefined && header.tools.length > 0 ? { tools: header.tools } : {},
|
||||
}
|
||||
}
|
||||
|
||||
/** Split a canonical (possibly absent) system prompt into lines; absence is zero lines. */
|
||||
function systemLines(system: string | undefined): string[] {
|
||||
return system === undefined ? [] : system.split('\n')
|
||||
}
|
||||
|
||||
/** Join lines back into a canonical system value; zero lines is absence. */
|
||||
function joinSystem(lines: string[]): string | undefined {
|
||||
return lines.length === 0 ? undefined : lines.join('\n')
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute the line-level {@link SystemDelta} between two canonical system
|
||||
* prompts: trim the common prefix and (non-overlapping) common suffix, and
|
||||
* carry the replacement lines between them. Deterministic and library-free;
|
||||
* with nothing shared it degenerates to a full replacement.
|
||||
*/
|
||||
function diffSystem(prev: string | undefined, next: string | undefined): SystemDelta {
|
||||
const a = systemLines(prev)
|
||||
const b = systemLines(next)
|
||||
let keepStart = 0
|
||||
while (keepStart < a.length && keepStart < b.length && a[keepStart] === b[keepStart]) keepStart += 1
|
||||
let keepEnd = 0
|
||||
while (
|
||||
keepEnd < a.length - keepStart &&
|
||||
keepEnd < b.length - keepStart &&
|
||||
a[a.length - 1 - keepEnd] === b[b.length - 1 - keepEnd]
|
||||
) keepEnd += 1
|
||||
return { keepStart, keepEnd, insert: b.slice(keepStart, b.length - keepEnd) }
|
||||
}
|
||||
|
||||
/** Apply a {@link SystemDelta} to a canonical system prompt. */
|
||||
function applySystem(prev: string | undefined, delta: SystemDelta): string | undefined {
|
||||
const a = systemLines(prev)
|
||||
return joinSystem([...a.slice(0, delta.keepStart), ...delta.insert, ...a.slice(a.length - delta.keepEnd)])
|
||||
}
|
||||
|
||||
/** Canonical JSON equality for tool schemas — sound because schemas are
|
||||
* JSON-serializable by construction and both sides come from the same
|
||||
* assembly path, so key insertion order matches when the values do. */
|
||||
function sameSchema(a: ToolSchema, b: ToolSchema): boolean {
|
||||
return JSON.stringify(a) === JSON.stringify(b)
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute the name-keyed {@link ToolsDelta} between two canonical tool lists.
|
||||
* A pure reordering produces an empty delta — the writer's round-trip guard
|
||||
* catches that case and records a snapshot instead.
|
||||
*/
|
||||
function diffTools(prev: readonly ToolSchema[], next: readonly ToolSchema[]): ToolsDelta {
|
||||
const prevByName = new Map(prev.map(tool => [tool.name, tool]))
|
||||
const nextNames = new Set(next.map(tool => tool.name))
|
||||
return {
|
||||
added: next.filter(tool => !prevByName.has(tool.name)),
|
||||
removed: prev.filter(tool => !nextNames.has(tool.name)).map(tool => tool.name),
|
||||
changed: next.filter((tool) => {
|
||||
const before = prevByName.get(tool.name)
|
||||
return before !== undefined && !sameSchema(before, tool)
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/** Apply a {@link ToolsDelta} to a canonical tool list: drop removed, replace changed in place, append added. */
|
||||
function applyTools(prev: readonly ToolSchema[], delta: ToolsDelta): ToolSchema[] {
|
||||
const removed = new Set(delta.removed)
|
||||
const changedByName = new Map(delta.changed.map(tool => [tool.name, tool]))
|
||||
const kept = prev
|
||||
.filter(tool => !removed.has(tool.name))
|
||||
.map(tool => changedByName.get(tool.name) ?? tool)
|
||||
return [...kept, ...delta.added]
|
||||
}
|
||||
|
||||
/**
|
||||
* Field-wise equality over canonical headers — the cheap comparison the
|
||||
* writer's round-trip guard runs (`applyHeaderDelta(prev, delta)` must equal
|
||||
* the intended header) and the loop runs to skip logging an unchanged header.
|
||||
* Tools compare per-schema IN ORDER (canonical JSON), so a pure reordering is
|
||||
* correctly unequal.
|
||||
* @param a - one canonical header.
|
||||
* @param b - the other.
|
||||
* @returns whether config, system, and tools (in order) all match.
|
||||
*/
|
||||
export function headerEquals(a: EpochHeader, b: EpochHeader): boolean {
|
||||
if (!callConfigEquals(a.config, b.config) || a.system !== b.system) return false
|
||||
const at = a.tools ?? []
|
||||
const bt = b.tools ?? []
|
||||
return at.length === bt.length && at.every((tool, i) => sameSchema(tool, bt[i] as ToolSchema))
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute the `request/header-delta` payload between two canonical headers,
|
||||
* or undefined when they are equal. The caller MUST round-trip the result
|
||||
* ({@link applyHeaderDelta} on `prev` deep-equals `next`) before logging it —
|
||||
* the encoding cannot express every change (a pure tool reordering) — and
|
||||
* fall back to a full `request/header` snapshot when the check fails.
|
||||
* @param prev - the folded header the log currently implies.
|
||||
* @param next - the header the next request will actually use.
|
||||
* @returns the delta payload, or undefined when nothing changed.
|
||||
*/
|
||||
export function diffHeader(
|
||||
prev: EpochHeader, next: EpochHeader,
|
||||
): { system?: SystemDelta; tools?: ToolsDelta; config?: LlmCallConfig } | undefined {
|
||||
const delta: { system?: SystemDelta; tools?: ToolsDelta; config?: LlmCallConfig } = {}
|
||||
if (prev.system !== next.system) delta.system = diffSystem(prev.system, next.system)
|
||||
const prevTools = prev.tools ?? []
|
||||
const nextTools = next.tools ?? []
|
||||
if (JSON.stringify(prevTools) !== JSON.stringify(nextTools)) delta.tools = diffTools(prevTools, nextTools)
|
||||
if (!callConfigEquals(prev.config, next.config)) delta.config = next.config
|
||||
return Object.keys(delta).length > 0 ? delta : undefined
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply a `request/header-delta` payload to a canonical header, producing the
|
||||
* canonical header it encodes. Total for well-formed logs (the writer only
|
||||
* appends round-trip-verified deltas).
|
||||
* @param prev - the folded header before the delta.
|
||||
* @param delta - the logged delta payload.
|
||||
* @returns the canonical header after the delta.
|
||||
*/
|
||||
export function applyHeaderDelta(
|
||||
prev: EpochHeader, delta: { system?: SystemDelta; tools?: ToolsDelta; config?: LlmCallConfig },
|
||||
): EpochHeader {
|
||||
const system = delta.system !== undefined ? applySystem(prev.system, delta.system) : prev.system
|
||||
const tools = delta.tools !== undefined ? applyTools(prev.tools ?? [], delta.tools) : prev.tools
|
||||
return canonicalHeader({
|
||||
config: delta.config ?? prev.config,
|
||||
...system !== undefined ? { system } : {},
|
||||
...tools !== undefined ? { tools } : {},
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Fold the header events of a log (or any prefix of one) into the
|
||||
* {@link EpochHeader} in force after the last of them: each
|
||||
* `request/header` snapshot replaces the state, each `request/header-delta`
|
||||
* amends it. The pure, offline form of reconstruction — external tooling and
|
||||
* the dev invariant both use it; the live session tracks the same fold
|
||||
* incrementally.
|
||||
* @param events - session events in log order (non-header events are skipped).
|
||||
* @param from - a previously folded state to continue from (the live session's
|
||||
* incremental cursor); omit to fold from nothing.
|
||||
* @returns the folded header, or undefined when no header event exists yet.
|
||||
*/
|
||||
export function foldRequestHeader(events: readonly SessionEvent[], from?: EpochHeader): EpochHeader | undefined {
|
||||
let state: EpochHeader | undefined = from
|
||||
for (const event of events) {
|
||||
if (event.type === 'request/header') {
|
||||
state = canonicalHeader(event.data.header)
|
||||
} else if (event.type === 'request/header-delta') {
|
||||
if (state === undefined) {
|
||||
throw new Error(`request/header-delta at seq ${event.seq} before any request/header snapshot: corrupt log`)
|
||||
}
|
||||
state = applyHeaderDelta(state, event.data)
|
||||
}
|
||||
}
|
||||
return state
|
||||
}
|
||||
@@ -72,6 +72,9 @@ export class SurfaceManager {
|
||||
/** The last processed seq. -1 forces a full rebuild on first access. */
|
||||
private _lastProcessedSeq = -1
|
||||
|
||||
/** Rewrite generation — see {@link replaceGeneration}. */
|
||||
private _replaceGeneration = 0
|
||||
|
||||
constructor(private log: readonly SessionEvent[]) {}
|
||||
|
||||
/**
|
||||
@@ -83,6 +86,23 @@ export class SurfaceManager {
|
||||
this._lastProcessedSeq = -1
|
||||
this._nodes = []
|
||||
this._nodeBySeq.clear()
|
||||
// A wholesale rebuild is a rewrite: bump the generation so incremental
|
||||
// consumers (the session's derived-message cache) discard their view.
|
||||
this._replaceGeneration += 1
|
||||
}
|
||||
|
||||
/**
|
||||
* The surface's rewrite generation: bumped by every folded `replace` op and
|
||||
* by {@link invalidate}. A replace is the ONE operation that rewrites the
|
||||
* surface non-monotonically, so an incremental consumer of {@link nodes}
|
||||
* (the session's derived-message cache) compares this between visits — an
|
||||
* unchanged generation guarantees every node it has not seen is a pure tail
|
||||
* append; a changed one means its view must rebuild. Monotonic: it never
|
||||
* moves backwards, so comparisons cannot be fooled by a re-fold.
|
||||
*/
|
||||
get replaceGeneration(): number {
|
||||
if (this._lastProcessedSeq < this.log.length - 1) this._processDelta()
|
||||
return this._replaceGeneration
|
||||
}
|
||||
|
||||
/** The surface nodes in linked-list order (head to tail). */
|
||||
@@ -155,5 +175,6 @@ export class SurfaceManager {
|
||||
if (nextNode) nextNode.prev = newSeq
|
||||
this._nodes.splice(startIdx, 0, newNode)
|
||||
this._nodeBySeq.set(newSeq, newNode)
|
||||
this._replaceGeneration += 1
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { Branded } from '@deepseek-ai/dsh-brand'
|
||||
import type { CallId, ContentBlock, MessageSource, StreamChunk, TokenUsage } from '@deepseek-ai/dsh-llm'
|
||||
import type { CallId, ContentBlock, LlmCallConfig, MessageSource, StreamChunk, TokenUsage, ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
/** Identifies one session in the store (and its persistence artifacts). */
|
||||
export type SessionId = Branded<'SessionId'>
|
||||
@@ -176,6 +176,67 @@ export interface TodoItem {
|
||||
status: 'pending' | 'in_progress' | 'completed'
|
||||
}
|
||||
|
||||
/**
|
||||
* The request header: everything about an LLM request besides its message
|
||||
* content — the call configuration plus the rendered system prompt and tool
|
||||
* schemas. Logged session state (the reconstructability RFC): a
|
||||
* {@link SessionEventMap} `request/header` snapshot installs one, a
|
||||
* `request/header-delta` amends it, and folding those events over the log
|
||||
* (`foldRequestHeader`) reconstructs the header any request was built under.
|
||||
* Canonical form: an empty system prompt and an empty tool list are ABSENT
|
||||
* fields, matching how requests are built.
|
||||
*/
|
||||
export interface EpochHeader {
|
||||
/** The conversation's call configuration (model + sampling scalars). */
|
||||
config: LlmCallConfig
|
||||
/** Rendered system prompt text; absent for a system-less request. */
|
||||
system?: string
|
||||
/** Assembled tool schemas; absent for a tool-less request. */
|
||||
tools?: ToolSchema[]
|
||||
}
|
||||
|
||||
/**
|
||||
* Why a `request/header` snapshot was appended: `'initial'` — the log's first
|
||||
* header (a new conversation); `'resume'` — a loop instance's first request
|
||||
* over a log that already has header events (process restart, fork seed);
|
||||
* `'fallback'` — a mid-run change the delta encoding could not round-trip
|
||||
* (e.g. a pure tool reordering), recorded whole instead.
|
||||
*/
|
||||
export type RequestHeaderReason = 'initial' | 'resume' | 'fallback'
|
||||
|
||||
/**
|
||||
* Line-level edit of the system prompt: keep the first `keepStart` and last
|
||||
* `keepEnd` lines of the previous text, with `insert` replacing everything
|
||||
* between. Computed as a common-prefix/common-suffix trim — deterministic,
|
||||
* library-free, degenerating to a full replacement when nothing is shared.
|
||||
* Absence is encoded as zero lines (the canonical form has no empty-string
|
||||
* system), so a transition to or from "no system prompt" round-trips.
|
||||
*/
|
||||
export interface SystemDelta {
|
||||
/** Lines kept from the start of the previous system prompt. */
|
||||
keepStart: number
|
||||
/** Lines kept from the end of the previous system prompt. */
|
||||
keepEnd: number
|
||||
/** Lines replacing everything between the kept edges. */
|
||||
insert: string[]
|
||||
}
|
||||
|
||||
/**
|
||||
* Tool-set edit keyed by tool name (names are unique — the registry rejects
|
||||
* duplicates): `removed` names drop, `changed` schemas replace their
|
||||
* predecessor in place, `added` schemas append at the end. A change this
|
||||
* encoding cannot express (a pure reordering) fails the writer's round-trip
|
||||
* guard and is recorded as a `'fallback'` snapshot instead.
|
||||
*/
|
||||
export interface ToolsDelta {
|
||||
/** Schemas appended to the end of the tool list. */
|
||||
added: ToolSchema[]
|
||||
/** Names of schemas dropped from the tool list. */
|
||||
removed: string[]
|
||||
/** Schemas replacing the same-named predecessor in place. */
|
||||
changed: ToolSchema[]
|
||||
}
|
||||
|
||||
/**
|
||||
* The session event vocabulary — the append-only source of truth for an
|
||||
* agent's whole interaction history. The LLM message history is *derived*
|
||||
@@ -274,6 +335,30 @@ export interface SessionEventMap {
|
||||
* cordis-catalog row.
|
||||
*/
|
||||
'todo/write': { todos: TodoItem[] }
|
||||
/**
|
||||
* Full snapshot of the {@link EpochHeader} the NEXT request is built under,
|
||||
* with the {@link RequestHeaderReason} it was recorded whole. Appended by
|
||||
* the loop inside the step, before dispatch, on a loop instance's first
|
||||
* request-building step (`'initial'`/`'resume'`) or when a delta failed its
|
||||
* round-trip guard (`'fallback'`); always records what the request actually
|
||||
* used, post-`agent/request`. Anchors the header fold: reconstruction reads
|
||||
* the latest snapshot and applies the deltas after it. NOT a
|
||||
* {@link SurfaceEventType}: it produces no LLM message — it is the request
|
||||
* envelope, logged so every request is a pure function of the session log
|
||||
* (the reconstructability RFC).
|
||||
*/
|
||||
'request/header': { header: EpochHeader; reason: RequestHeaderReason }
|
||||
/**
|
||||
* Amendment to the folded {@link EpochHeader}: at least one of a
|
||||
* {@link SystemDelta}, a {@link ToolsDelta}, or a whole replacement
|
||||
* {@link LlmCallConfig} (four scalars — not worth diffing). Appended by the
|
||||
* loop inside the step, before dispatch, when the header for this request
|
||||
* differs from the fold of the log so far; the writer verifies
|
||||
* `applyHeaderDelta(previous, delta)` reproduces the new header exactly and
|
||||
* falls back to a `'fallback'` `request/header` snapshot when it cannot, so
|
||||
* a logged delta ALWAYS round-trips. NOT a {@link SurfaceEventType}.
|
||||
*/
|
||||
'request/header-delta': { system?: SystemDelta; tools?: ToolsDelta; config?: LlmCallConfig }
|
||||
}
|
||||
|
||||
export type SessionEventType = keyof SessionEventMap
|
||||
|
||||
112
packages/core/session/tests/derived-cache.spec.ts
Normal file
112
packages/core/session/tests/derived-cache.spec.ts
Normal file
@@ -0,0 +1,112 @@
|
||||
/**
|
||||
* Derived-message cache tests: the session projects each surface node exactly
|
||||
* once (O(new nodes) per call), rebuilds on a surface rewrite (replace /
|
||||
* invalidate — the replaceGeneration signal), returns a fresh array snapshot
|
||||
* per call over shared frozen messages, and stays deep-equal to a from-scratch
|
||||
* replay derivation at every step — the incremental==scratch property the
|
||||
* reconstructability RFC's invariant enforces in dev at request time.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Session, SessionId } from '@deepseek-ai/dsh-session'
|
||||
|
||||
function userText(session: Session, text: string): void {
|
||||
session.append('user/message', { content: [{ type: 'text', text }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
}
|
||||
|
||||
/** From-scratch oracle: replay the log into a fresh session and derive. */
|
||||
function scratch(session: Session): unknown {
|
||||
return new Session(SessionId(`${session.id}-scratch-${session.seq}`), [...session.events]).deriveMessages()
|
||||
}
|
||||
|
||||
describe('derived-message cache', () => {
|
||||
it('stays deep-equal to a from-scratch replay derivation as the log grows', () => {
|
||||
const session = new Session(SessionId('cache-grow'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
userText(session, 'one')
|
||||
expect(session.deriveMessages()).toEqual(scratch(session))
|
||||
userText(session, 'two')
|
||||
session.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'reply' }] }, { surfaceOp: 'append' })
|
||||
expect(session.deriveMessages()).toEqual(scratch(session))
|
||||
// An empty-content assistant/message (usage host) projects to nothing.
|
||||
session.append('assistant/message', { turn: 1, step: 2, content: [], usage: { inputTokens: 1, outputTokens: 0 } }, { surfaceOp: 'append' })
|
||||
expect(session.deriveMessages()).toEqual(scratch(session))
|
||||
})
|
||||
|
||||
it('rebuilds on a surface replace and still matches scratch', () => {
|
||||
const session = new Session(SessionId('cache-replace'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
userText(session, 'one')
|
||||
userText(session, 'two')
|
||||
const beforeReplace = session.deriveMessages()
|
||||
expect(beforeReplace).toHaveLength(2)
|
||||
|
||||
const nodes = session.surface.nodes
|
||||
session.append('context/message', {
|
||||
content: [{ type: 'text', text: 'summary' }], source: { kind: 'plugin', plugin: 'compact' },
|
||||
}, { surfaceOp: { op: 'replace', start: nodes[0]!.seq, end: nodes[1]!.seq }, sourceEventSeqs: [nodes[0]!.seq, nodes[1]!.seq] })
|
||||
|
||||
expect(session.deriveMessages()).toHaveLength(1)
|
||||
expect(session.deriveMessages()).toEqual(scratch(session))
|
||||
// The array a caller took before the replace is untouched.
|
||||
expect(beforeReplace).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('returns a fresh array per call: later appends never grow a held snapshot', () => {
|
||||
const session = new Session(SessionId('cache-snapshot'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
userText(session, 'one')
|
||||
const first = session.deriveMessages()
|
||||
userText(session, 'two')
|
||||
const second = session.deriveMessages()
|
||||
expect(first).toHaveLength(1)
|
||||
expect(second).toHaveLength(2)
|
||||
// Shared projection objects: the same frozen message instance, once ever.
|
||||
expect(second[0]).toBe(first[0])
|
||||
expect(Object.isFrozen(first[0])).toBe(true)
|
||||
})
|
||||
|
||||
it('rebuilds after surface.invalidate() (the generation covers wholesale rebuilds too)', () => {
|
||||
const session = new Session(SessionId('cache-invalidate'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
userText(session, 'one')
|
||||
const before = session.deriveMessages()
|
||||
session.surface.invalidate()
|
||||
const after = session.deriveMessages()
|
||||
expect(after).toEqual(before)
|
||||
// A rebuild re-projects: fresh objects, same values.
|
||||
expect(after[0]).not.toBe(before[0])
|
||||
})
|
||||
})
|
||||
|
||||
describe('Session.deriveEventMessage — the per-event projection', () => {
|
||||
it('projects one appended event exactly as the full derivation projects its node', () => {
|
||||
const session = new Session(SessionId('per-event'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
// The fold path (deriveMessages) and the per-event path share the
|
||||
// projection, so an external reconstructor cannot disagree with the cache.
|
||||
expect(session.deriveEventMessage(event)).toEqual(session.deriveMessages().at(-1))
|
||||
})
|
||||
|
||||
it('clones content off the log: the projection never aliases the logged event', () => {
|
||||
const session = new Session(SessionId('per-event-clone'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
const event = session.append('user/message', { content: [{ type: 'text', text: 'orig' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
const message = session.deriveEventMessage(event)!
|
||||
expect(message.content).not.toBe(event.data.content)
|
||||
// deriveEventMessage returns an unfrozen clone (the cache freezes ITS
|
||||
// copies); mutating it must not reach the log.
|
||||
;(message.content[0] as { text: string }).text = 'mutated'
|
||||
expect(session.deriveMessages().at(-1)!.content).toEqual([{ type: 'text', text: 'orig' }])
|
||||
})
|
||||
|
||||
it('projects null for events that produce no message (boundaries, empty assistant)', () => {
|
||||
const session = new Session(SessionId('per-event-null'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
const boundary = session.append('step/start', { turn: 1, step: 1 })
|
||||
expect(session.deriveEventMessage(boundary)).toBeNull()
|
||||
const empty = session.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: 'append' })
|
||||
expect(session.deriveEventMessage(empty)).toBeNull()
|
||||
})
|
||||
})
|
||||
@@ -109,15 +109,17 @@ describe('Session properties', () => {
|
||||
))
|
||||
})
|
||||
|
||||
it('every derived message has a known role and decoupled content', () => {
|
||||
it('every derived message has a known role and is frozen (append-only contract)', () => {
|
||||
fc.assert(fc.property(logArb, (events) => {
|
||||
const session = build(events)
|
||||
const messages = session.deriveMessages()
|
||||
const before = structuredClone(session.events)
|
||||
for (const m of messages) {
|
||||
expect(['user', 'assistant', 'system']).toContain(m.role)
|
||||
// Mutating derived content must not touch the log (append-only).
|
||||
m.content.push({ type: 'text', text: 'mutation' })
|
||||
// Derived messages are frozen shared projections: mutation THROWS
|
||||
// (strict mode) instead of relying on per-call clones for isolation.
|
||||
expect(Object.isFrozen(m)).toBe(true)
|
||||
expect(() => { m.content.push({ type: 'text', text: 'mutation' }) }).toThrow(TypeError)
|
||||
}
|
||||
expect(session.events).toEqual(before)
|
||||
}))
|
||||
|
||||
140
packages/core/session/tests/request-header.spec.ts
Normal file
140
packages/core/session/tests/request-header.spec.ts
Normal file
@@ -0,0 +1,140 @@
|
||||
/**
|
||||
* Request-header utility tests: canonical form, the system line-diff
|
||||
* (prefix/suffix trim), the name-keyed tools delta, config replacement, the
|
||||
* round-trip contract (including the reorder case the encoding cannot
|
||||
* express), and the log fold. These pin the reconstruction algebra: for every
|
||||
* logged delta, apply(prev, delta) === next, and folding a log prefix yields
|
||||
* the header its next request was built under.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Session, SessionId, applyHeaderDelta, canonicalHeader, diffHeader, foldRequestHeader } from '@deepseek-ai/dsh-session'
|
||||
import type { EpochHeader, SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import type { ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
const CONFIG = { model: 'm' }
|
||||
|
||||
function tool(name: string, description = 'd'): ToolSchema {
|
||||
return { name, description, parameters: { type: 'object' } }
|
||||
}
|
||||
|
||||
/** Round-trip helper: diff must reproduce `next` from `prev` exactly. */
|
||||
function roundTrip(prev: EpochHeader, next: EpochHeader): ReturnType<typeof diffHeader> {
|
||||
const delta = diffHeader(prev, next)
|
||||
if (delta !== undefined) {
|
||||
expect(applyHeaderDelta(prev, delta)).toEqual(canonicalHeader(next))
|
||||
}
|
||||
return delta
|
||||
}
|
||||
|
||||
describe('canonicalHeader', () => {
|
||||
it('normalizes empty system and empty tools to absent fields', () => {
|
||||
expect(canonicalHeader({ config: CONFIG, system: '', tools: [] })).toEqual({ config: CONFIG })
|
||||
const full = canonicalHeader({ config: CONFIG, system: 's', tools: [tool('a')] })
|
||||
expect(full.system).toBe('s')
|
||||
expect(full.tools).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe('diffHeader / applyHeaderDelta', () => {
|
||||
it('returns undefined for equal headers', () => {
|
||||
const header = canonicalHeader({ config: CONFIG, system: 'a\nb', tools: [tool('t')] })
|
||||
expect(diffHeader(header, header)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('encodes a mid-prompt line change as a prefix/suffix trim', () => {
|
||||
const prev = canonicalHeader({ config: CONFIG, system: 'keep1\nold\nkeep2\nkeep3' })
|
||||
const next = canonicalHeader({ config: CONFIG, system: 'keep1\nnew A\nnew B\nkeep2\nkeep3' })
|
||||
const delta = roundTrip(prev, next)
|
||||
expect(delta?.system).toEqual({ keepStart: 1, keepEnd: 2, insert: ['new A', 'new B'] })
|
||||
expect(delta?.tools).toBeUndefined()
|
||||
expect(delta?.config).toBeUndefined()
|
||||
})
|
||||
|
||||
it('degenerates to a full replacement when nothing is shared, and round-trips absence transitions', () => {
|
||||
const none = canonicalHeader({ config: CONFIG })
|
||||
const some = canonicalHeader({ config: CONFIG, system: 'x\ny' })
|
||||
const gained = roundTrip(none, some)
|
||||
expect(gained?.system).toEqual({ keepStart: 0, keepEnd: 0, insert: ['x', 'y'] })
|
||||
const lost = roundTrip(some, none)
|
||||
expect(lost?.system).toEqual({ keepStart: 0, keepEnd: 0, insert: [] })
|
||||
})
|
||||
|
||||
it('does not double-count overlapping prefix and suffix (repeated lines)', () => {
|
||||
const prev = canonicalHeader({ config: CONFIG, system: 'a\na' })
|
||||
const next = canonicalHeader({ config: CONFIG, system: 'a\na\na' })
|
||||
roundTrip(prev, next)
|
||||
})
|
||||
|
||||
it('encodes tool addition, removal, and in-place schema change by name', () => {
|
||||
const prev = canonicalHeader({ config: CONFIG, tools: [tool('keep'), tool('drop'), tool('edit', 'before')] })
|
||||
const next = canonicalHeader({ config: CONFIG, tools: [tool('keep'), tool('edit', 'after'), tool('new')] })
|
||||
const delta = roundTrip(prev, next)
|
||||
expect(delta?.tools?.added.map(t => t.name)).toEqual(['new'])
|
||||
expect(delta?.tools?.removed).toEqual(['drop'])
|
||||
expect(delta?.tools?.changed.map(t => t.name)).toEqual(['edit'])
|
||||
})
|
||||
|
||||
it('round-trips a tool set gained from a tool-less header and lost back to one', () => {
|
||||
const none = canonicalHeader({ config: CONFIG })
|
||||
const some = canonicalHeader({ config: CONFIG, tools: [tool('t')] })
|
||||
const gained = roundTrip(none, some)
|
||||
expect(gained?.tools?.added.map(t => t.name)).toEqual(['t'])
|
||||
const lost = roundTrip(some, none)
|
||||
expect(lost?.tools?.removed).toEqual(['t'])
|
||||
})
|
||||
|
||||
it('cannot express a pure reordering — the writer detects it via the round-trip check', () => {
|
||||
const prev = canonicalHeader({ config: CONFIG, tools: [tool('a'), tool('b')] })
|
||||
const next = canonicalHeader({ config: CONFIG, tools: [tool('b'), tool('a')] })
|
||||
const delta = diffHeader(prev, next)
|
||||
// A delta IS produced (the lists differ)…
|
||||
expect(delta).toBeDefined()
|
||||
// …but applying it cannot reproduce the new order — exactly the case the
|
||||
// writer's guard turns into a 'fallback' snapshot.
|
||||
expect(applyHeaderDelta(prev, delta!)).not.toEqual(next)
|
||||
})
|
||||
|
||||
it('replaces the config whole and leaves untouched parts alone', () => {
|
||||
const prev = canonicalHeader({ config: { model: 'm' }, system: 's', tools: [tool('t')] })
|
||||
const next = canonicalHeader({ config: { model: 'm2', temperature: 0.1 }, system: 's', tools: [tool('t')] })
|
||||
const delta = roundTrip(prev, next)
|
||||
expect(delta).toEqual({ config: { model: 'm2', temperature: 0.1 } })
|
||||
})
|
||||
})
|
||||
|
||||
describe('foldRequestHeader', () => {
|
||||
function headerEvents(session: Session): readonly SessionEvent[] {
|
||||
return session.events
|
||||
}
|
||||
|
||||
it('returns undefined on a log with no header events', () => {
|
||||
const session = new Session(SessionId('fold-none'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
expect(foldRequestHeader(headerEvents(session))).toBeUndefined()
|
||||
})
|
||||
|
||||
it('folds snapshot then deltas into the header in force, skipping unrelated events', () => {
|
||||
const session = new Session(SessionId('fold'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
const first = canonicalHeader({ config: { model: 'm' }, system: 'a\nb', tools: [tool('t')] })
|
||||
session.append('request/header', { header: first, reason: 'initial' })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
|
||||
const second = canonicalHeader({ config: { model: 'm' }, system: 'a\nc', tools: [tool('t')] })
|
||||
session.append('request/header-delta', diffHeader(first, second)!)
|
||||
expect(foldRequestHeader(headerEvents(session))).toEqual(second)
|
||||
|
||||
// A later snapshot replaces the state wholesale (the 'resume'/'fallback' anchor).
|
||||
const third = canonicalHeader({ config: { model: 'other' } })
|
||||
session.append('request/header', { header: third, reason: 'resume' })
|
||||
expect(foldRequestHeader(headerEvents(session))).toEqual(third)
|
||||
})
|
||||
|
||||
it('throws on a delta before any snapshot (corrupt log)', () => {
|
||||
const session = new Session(SessionId('fold-corrupt'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('request/header-delta', { config: { model: 'x' } })
|
||||
expect(() => foldRequestHeader(headerEvents(session))).toThrow(/before any request\/header snapshot/)
|
||||
})
|
||||
})
|
||||
@@ -78,19 +78,25 @@ describe('Session', () => {
|
||||
}, { surfaceOp: 'append' })
|
||||
const before = structuredClone(session.events)
|
||||
|
||||
// A request middleware / adapter mutates the messages it was handed.
|
||||
// A misbehaving consumer tries to mutate the messages it was handed.
|
||||
// Derived messages are frozen shared projections (cloned once off the
|
||||
// log, then deep-frozen): every mutation attempt THROWS in strict mode —
|
||||
// isolation by unrepresentability, not by per-call cloning.
|
||||
const messages = session.deriveMessages()
|
||||
const userBlock = messages[0]!.content[0]!
|
||||
if (userBlock.type === 'text') userBlock.text = 'HACKED'
|
||||
expect(() => { if (userBlock.type === 'text') userBlock.text = 'HACKED' }).toThrow(TypeError)
|
||||
const toolBlock = messages[1]!.content[0]!
|
||||
if (toolBlock.type === 'tool-result') {
|
||||
toolBlock.content.push({ type: 'text', text: 'injected' })
|
||||
}
|
||||
messages[0]!.content.push({ type: 'text', text: 'extra' })
|
||||
expect(() => {
|
||||
if (toolBlock.type === 'tool-result') toolBlock.content.push({ type: 'text', text: 'injected' })
|
||||
}).toThrow(TypeError)
|
||||
expect(() => { messages[0]!.content.push({ type: 'text', text: 'extra' }) }).toThrow(TypeError)
|
||||
// The returned ARRAY is the caller's own snapshot, though — reordering it
|
||||
// is the caller's business and never reaches the cache or the log.
|
||||
messages.reverse()
|
||||
|
||||
// The log is unchanged: deep-equal to the snapshot taken before mutation.
|
||||
expect(session.events).toEqual(before)
|
||||
// And a fresh derivation still reflects the original content.
|
||||
// And a fresh derivation still reflects the original content and order.
|
||||
expect(session.deriveMessages()[0]!.content).toEqual([{ type: 'text', text: 'original' }])
|
||||
})
|
||||
|
||||
|
||||
@@ -334,3 +334,26 @@ describe('surface type guards', () => {
|
||||
expect(isSurfaceEvent(markerless)).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
describe('SurfaceManager.replaceGeneration', () => {
|
||||
it('folds the pending log delta on access and counts replaces and invalidations', () => {
|
||||
const s = new Session(SessionId('gen'))
|
||||
s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
s.append('user/message', { content: [{ type: 'text', text: 'one' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
s.append('user/message', { content: [{ type: 'text', text: 'two' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
// Read the generation FIRST — before nodes — so the getter itself folds
|
||||
// the pending delta rather than piggybacking on a nodes read.
|
||||
expect(s.surface.replaceGeneration).toBe(0)
|
||||
|
||||
const nodes = s.surface.nodes
|
||||
s.append('context/message', {
|
||||
content: [{ type: 'text', text: 'summary' }], source: { kind: 'plugin', plugin: 'compact' },
|
||||
}, { surfaceOp: { op: 'replace', start: nodes[0]!.seq, end: nodes[1]!.seq }, sourceEventSeqs: [nodes[0]!.seq, nodes[1]!.seq] })
|
||||
expect(s.surface.replaceGeneration).toBe(1)
|
||||
|
||||
// invalidate() is a rewrite too: the generation moves forward (and the
|
||||
// refold re-counts the replace), never backwards.
|
||||
s.surface.invalidate()
|
||||
expect(s.surface.replaceGeneration).toBeGreaterThan(1)
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user