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

Merge origin/master into project instruction files

Browse Source
This commit is contained in:
Yichen Jiang
2026-07-06 15:44:42 +08:00
parent 901315d8ab cbc9c26366
commit a3b0da0255
118 changed files with 7460 additions and 6773 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
packages/compact/compact-basic/README.md
View File

@@ -1,6 +1,6 @@
# @deepseek-ai/dsh-compact-basic
The **basic compaction backend**: a `BasicCompactService` implementing the `@deepseek-ai/dsh-compact` seam with a chars-per-token heuristic (the `charsPerToken` config, default 4), token-budget retention, and summarization routed through the agent request pipeline.
The **basic compaction backend**: a `BasicCompactService` implementing the `@deepseek-ai/dsh-compact` seam with a chars-per-token heuristic (the `charsPerToken` config, default 4), token-budget retention, and summarization as a direct one-shot `ctx.llm.stream()` call (interceptable at `llm/stream`).
This is the implementation tier of the compaction capability — see the [interface package](../compact/README.md) for the seam and the [capability-seam RFC](../../../docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md) for the design.
@@ -11,13 +11,13 @@ The abstract contract states only WHAT compaction does; this backend owns every
- **Token estimation** — `estimateContentTokens()`: chars divided by the `charsPerToken` config (default 4) with per-block structural overhead (`text`/`reasoning` = `ceil(len/charsPerToken) + 4`, `tool-call` from name + arguments, `tool-result` recursive, unknown blocks via JSON length).
- **Retention policy** — `compactIfNeeded()` walks the surface nodes tail→head summing per-node token estimates, and retains the smallest tail-run of WHOLE units (a closed step, or a single no-step node such as a pre-step `user/message` or inter-step `steering/message`) whose total reaches `retainTokens`; everything older is compacted. Retention is **turn-agnostic** — turn boundaries play no role, so a single runaway turn that alone exceeds the window compacts its OWN early closed steps rather than being retained verbatim (the failure mode that motivated dropping turn-protection: a tool-heavy turn must stay compactable or the harness dies exactly when compaction is needed). The only structural guard is **tool-pairing balance**: the compacted region's edges are balanced cuts on the surface (no unanswered tool-call crosses either edge), so it never splits a step's `assistant/message` tool-calls from their `tool/result`s. When the only compactable content left is an un-splittable open tail step, it declines (returns `null`) and retries once an older step closes. **Single-unit overflow is out of scope, by design**: if one retained unit (a single closed step, or a large pasted `user/message`) ALONE exceeds the budget, compaction cannot help and the call may go out over-budget — bounding an individual unit's size is a separate concern. `compactRegion()` enforces tool-pairing balance strictly, throwing on a boundary that would split a step. `dsh-session` exports `isToolPairingBalanced` for the check.
- **Dynamic convergence** — no static summary-length config pretends to bound what the model will write. If framing/estimator/system overhead leaves the compacted surface above threshold, `compactIfNeeded()` re-compacts the head checkpoint up to `compactionRetries` extra times; if it still cannot get below threshold, it throws. A summary whose estimated stored size is not smaller than the shadowed content fails closed before it mutates the surface.
- **Summarization** — `summarize()`: a `GenerateOptions` request assembled via `BlockAssembler` with a fixed system prompt that asks for a structured checkpoint (Primary Request and Intent · Key Technical Concepts · Files and Code · Errors and Fixes · Pending Tasks · Current Work · Next Step · Critical Context), every section mandatory, exact paths/commands/identifiers preserved. The request runs through the `agent/request` waterfall before `ctx.llm.stream()`, so router agents that choose the concrete model there also route compaction summaries. `maxTokens` is the provider-side generation cap; only text blocks from the model's reply are kept before the checkpoint is stored (reasoning is dropped so private chain-of-thought never leaks into the durable summary, and a stray `tool-call` is dropped so the synthesized `user/message` summary cannot land an orphaned call with no matching `tool-result`). The compacted region is flattened to a plain-text transcript first: text and reasoning contribute their text, and every non-text block (tool-call, tool-result, plugin-added types) contributes a type-tagged placeholder (`[tool-call: name(args)]`, `[tool-result: …]`, …) so the summarizer is told what existed rather than silently dropping it.
- **Summarization** — `summarize()`: a `GenerateOptions` request assembled via `BlockAssembler` with a fixed system prompt that asks for a structured checkpoint (Primary Request and Intent · Key Technical Concepts · Files and Code · Errors and Fixes · Pending Tasks · Current Work · Next Step · Critical Context), every section mandatory, exact paths/commands/identifiers preserved. The request is a direct one-shot `ctx.llm.stream()` call — NOT a loop step, so it does not run `agent/request` (that seam shapes the loop's conversation requests); the model comes from `summarizationModel` falling back to the agent's own, and per-call routing happens at `llm/stream` like any other direct call. `maxTokens` is the provider-side generation cap; only text blocks from the model's reply are kept before the checkpoint is stored (reasoning is dropped so private chain-of-thought never leaks into the durable summary, and a stray `tool-call` is dropped so the synthesized `user/message` summary cannot land an orphaned call with no matching `tool-result`). The compacted region is flattened to a plain-text transcript first: text and reasoning contribute their text, and every non-text block (tool-call, tool-result, plugin-added types) contributes a type-tagged placeholder (`[tool-call: name(args)]`, `[tool-result: …]`, …) so the summarizer is told what existed rather than silently dropping it.
- **Checkpoint framing** — the raw summary is not landed directly. `compactRegion()` wraps it in a checkpoint preamble (so a resuming model reads it as a checkpoint, not a fresh user request, and builds on the captured context rather than restating it) plus `<compacted-summary>…</compacted-summary>` tags. Because region compaction can be invoked manually, a surface may hold several checkpoints, so the framing does not claim everything after it is recent or verbatim. The tags make a prior checkpoint detectable in the transcript on the next compaction cycle: the summarization prompt then instructs the model to merge it in place (preserve still-true facts, drop stale ones) rather than re-summarize it verbatim — a cheap incremental merge that needs no extra log/event machinery. The unframed summary stays on the `compact/summary` provenance event.
- **Surface mutation** — `compactRegion()` appends the `compact/start` → `compact/summary` → `compact/end` log records and the single `user/message` replace node carrying the framed summary (see the interface README).
- **Auto-compaction** — an `agent/pre-step` listener delegates to `compactIfNeeded()` before every step (not just a turn's first — a tool-heavy turn grows the surface mid-turn, so a runaway turn still compacts, and per-step firing is the only moment to rescue it before overflow). `agent/pre-step` is a serial (awaited, in-order) surface-mutation checkpoint that fires after `turn/start` and BEFORE the step opens (`step/start`) and its request history is derived, so compaction mutates the surface — with its log-only `compact/*` records landing cleanly outside any step — and the loop derives once from the result: no double-derive, and the listener cannot see (or need to rewrite) an already-assembled `messages` array. The listener owns no threshold logic of its own (the single token-pressure check lives in `compactIfNeeded()`); because Cordis `serial` bails early on non-void return values, the listener returns `void` and does not use the dispatcher's bail channel as a veto surface.
- **Failure handling** — the `compact/start … compact/end` bracket is a log-recorded lock: it makes a crash mid-summarization a detectable orphan (a `compact/start` with no `compact/end`), records provenance, and prevents a concurrent compaction. Two failure paths: a **crash** (the loop dies mid-summarization) leaves a dangling `compact/start` that is inert — `compact/*` events are log-only, the surface replacement never landed, so the full history derives fine and generic turn-repair closes the turn; a **recoverable** failure (summarization throws but the loop survives) appends `compact/end` with its `error` field set, leaving the surface untouched so the call proceeds with full history. Core session repair stays compaction-agnostic by design — it never learns about `compact/*`.
`estimateContentTokens()` and `summarize()` are overridable hooks: a tokenizer-based or template-based backend can subclass `BasicCompactService` and override just those, reusing the retention walk and surface plumbing.
`estimateContentTokens()` and `summarize()` are overridable hooks: a tokenizer-based or template-based backend can subclass `BasicCompactService` and override just those, reusing the retention walk and surface plumbing. `summarize()` returns the summary blocks together with the call envelope it actually used (`{ summary, model, maxTokens? }`) — the caller logs that envelope on the `compact/summary` provenance event, so an overriding backend reports its own envelope honestly.
## Config (`BasicCompactConfig`)

52
packages/compact/compact-basic/src/index.ts
View File

@@ -9,9 +9,10 @@
* to the next balanced tool-pairing boundary so a compacted region never
* splits a step's tool-call/result pair (an open tail step is never crossed —
* compaction declines and retries once it closes).
* - **Summarization** — `ctx.llm.stream()` assembled via `BlockAssembler`
* (the single model-call surface; same path the loop uses) with a fixed
* condense-the-history system prompt routed through `agent/request`.
* - **Summarization** — a direct one-shot `ctx.llm.stream()` call assembled
* via `BlockAssembler` with a fixed condense-the-history system prompt;
* NOT a loop step, so `agent/request` never fires — interception happens
* at `llm/stream` like any other direct call.
* - **Surface mutation** — a single `user/message` replace node carries the
* summary; `compact/*` events are log-only lock + provenance records.
* - **Auto-compaction** — an `agent/pre-step` listener delegates to
@@ -182,9 +183,9 @@ export class BasicCompactService extends CompactService {
// log-only `compact/*` records and the replacement node cleanly outside a
// step, so a crash mid-compaction leaves an inert orphan the turn-repair
// closes — never a half-open step.
ctx.on('agent/pre-step', async (agent: Agent, turn: number, step: number, fullSystemPrompt: string, signal: AbortSignal) => {
ctx.on('agent/pre-step', async (agent: Agent, _turn: number, _step: number, fullSystemPrompt: string, signal: AbortSignal) => {
try {
const result = await this.compactIfNeeded(agent, turn, step, fullSystemPrompt, signal)
const result = await this.compactIfNeeded(agent, fullSystemPrompt, signal)
if (result) {
const after = this.estimateTokens(agent.session.deriveMessages(), fullSystemPrompt)
ctx.logger.info(
@@ -271,9 +272,13 @@ export class BasicCompactService extends CompactService {
}
/**
* Summarize conversation text into content blocks via `agent/request` plus
* `ctx.llm.stream()` assembled through a `BlockAssembler` (the single
* model-call surface).
* Summarize conversation text into content blocks via `ctx.llm.stream()`
* assembled through a `BlockAssembler`. A direct one-shot model call, NOT a
* loop step: it does not run the `agent/request` waterfall (that seam shapes
* the loop's conversation requests); per-call
* interception happens at `llm/stream` like any other direct call. The model
* comes from `BasicCompactConfig.summarizationModel`, falling back to the
* agent's own model.
* Override in a subclass for a template or remote summarizer.
*
* Honors the adapter failure contract: an adapter may report a model failure
@@ -283,8 +288,15 @@ export class BasicCompactService extends CompactService {
*
* Forwards `signal` into `GenerateOptions.signal` so an abort/dispose tears
* down the in-flight summarization rather than orphaning the model call.
*
* Returns the summary blocks TOGETHER with the call envelope it actually
* used (`model`, `maxTokens`) — the caller logs the envelope on the
* `compact/summary` provenance event, so an overriding subclass (template
* or remote summarizer) reports its own envelope honestly.
*/
async summarize(text: string, agent: Agent, turn: number, step: number, signal?: AbortSignal): Promise<ContentBlock[]> {
async summarize(
text: string, agent: Agent, signal?: AbortSignal,
): Promise<{ summary: ContentBlock[]; model: string; maxTokens?: number }> {
const assembler = new BlockAssembler()
const options: GenerateOptions = {
model: this.config.summarizationModel || agent.options.model || '',
@@ -299,11 +311,10 @@ export class BasicCompactService extends CompactService {
// exactOptionalPropertyTypes: only set `signal` when present — assigning
// `undefined` to an optional `signal?: AbortSignal` is a type error.
if (signal) options.signal = signal
const request = await this.ctx.waterfall('agent/request', agent, turn, step, options, () => Promise.resolve(options))
if (!request.model) {
throw new Error('no model available for summarization: set BasicCompactConfig.summarizationModel, AgentOptions.model, or supply one via the agent/request waterfall')
if (!options.model) {
throw new Error('no model available for summarization: set BasicCompactConfig.summarizationModel or AgentOptions.model')
}
for await (const chunk of this.ctx.llm.stream(request)) {
for await (const chunk of this.ctx.llm.stream(options)) {
assembler.push(chunk)
}
@@ -315,7 +326,10 @@ export class BasicCompactService extends CompactService {
throw new Error('summarization produced no text summary content')
}
return summary
// config.maxTokens is required and validated positive, so this backend's
// envelope always carries the cap; the return type's optionality exists
// for overriding subclasses whose summarizer has none.
return { summary, model: options.model, maxTokens: this.config.maxTokens }
}
// ---- Core API (implements the abstract contract) ----
@@ -348,8 +362,6 @@ export class BasicCompactService extends CompactService {
*/
override async compactIfNeeded(
agent: Agent,
turn: number,
step: number,
fullSystemPrompt: string,
signal: AbortSignal,
): Promise<CompactionResult | null> {
@@ -368,7 +380,7 @@ export class BasicCompactService extends CompactService {
break
}
result = await this.compactRegion(session, range.start, range.end, agent, turn, step, signal)
result = await this.compactRegion(session, range.start, range.end, agent, signal)
}
const totalTokens = this.estimateTokens(session.deriveMessages(), fullSystemPrompt)
@@ -385,8 +397,6 @@ export class BasicCompactService extends CompactService {
start: number,
end: number,
agent: Agent,
turn: number,
step: number,
signal?: AbortSignal,
): Promise<CompactionResult> {
// Resolve the range by surface POSITION, not numeric seq interval. A prior
@@ -450,7 +460,7 @@ export class BasicCompactService extends CompactService {
try {
// --- Extract text and summarize ---
const text = this._extractText(session, shadowedSeqs)
const summary = await this.summarize(text, agent, turn, step, signal)
const { summary, model, maxTokens } = await this.summarize(text, agent, signal)
// Estimate token count of the shadowed content for provenance.
let shadowedTokenCount = 0
@@ -472,6 +482,8 @@ export class BasicCompactService extends CompactService {
shadowedRange: { start, end },
shadowedSeqs,
shadowedTokenCount,
model,
...maxTokens !== undefined ? { maxTokens } : {},
})
// --- Surface replacement ---

32
packages/compact/compact-basic/tests/compact-basic.spec.ts
View File

@@ -58,13 +58,13 @@ class TestCompactService extends BasicCompactService {
return blocks.length * 10
}
override async summarize(text: string, agent: Agent): Promise<ContentBlock[]> {
override async summarize(text: string, agent: Agent): Promise<{ summary: ContentBlock[]; model: string; maxTokens?: number }> {
const model = this.config.summarizationModel || agent.options.model || ''
this.summarizeCalls.push({ text, model })
if (this.summarizeError) throw this.summarizeError
const summary = this.mockSummaryQueue.shift() ?? this.mockSummary
this.summaryOutputs.add(summary)
return summary
return { summary, model }
}
}
@@ -402,6 +402,9 @@ describe('BasicCompactService.compactRegion', () => {
expect(startEvent).toBeDefined()
expect(summaryEvent).toBeDefined()
expect(endEvent).toBeDefined()
// The provenance record carries the summarize call's envelope, so "which
// model wrote this summary" is answerable from the log alone.
expect(summaryEvent?.type === 'compact/summary' && summaryEvent.data.model).toBe('test-model')
// compact/* events are log-only — no surfaceOp (type system enforces this).
const startRaw = startEvent as unknown as { surfaceOp?: unknown }
@@ -980,7 +983,7 @@ function compactIfNeeded(
model: string,
signal: AbortSignal,
) {
return svc.compactIfNeeded(stubAgent(session, model), 1, 1, fullSystemPrompt, signal)
return svc.compactIfNeeded(stubAgent(session, model), fullSystemPrompt, signal)
}
function compactRegion(
@@ -991,11 +994,11 @@ function compactRegion(
model: string,
signal?: AbortSignal,
) {
return svc.compactRegion(session, start, end, stubAgent(session, model), 1, 1, signal)
return svc.compactRegion(session, start, end, stubAgent(session, model), signal)
}
function summarize(svc: BasicCompactService, text: string, model: string) {
return svc.summarize(text, stubAgent(new Session(SessionId('summary')), model), 1, 1)
return svc.summarize(text, stubAgent(new Session(SessionId('summary')), model))
}
describe('BasicCompactService.summarize (real ctx.llm.stream)', () => {
@@ -1003,8 +1006,12 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => {
const { ctx, adapter } = await ctxWithModel('SUMMARY TEXT')
const svc = new BasicCompactService(ctx, cfg({ auto: false, maxTokens: 512 }))
const summary = await summarize(svc, 'User: hi\n\nAssistant: hello', 'test-model')
const { summary, model, maxTokens } = await summarize(svc, 'User: hi\n\nAssistant: hello', 'test-model')
expect(summary).toEqual([{ type: 'text', text: 'SUMMARY TEXT' }])
// The returned envelope reports what the call actually used — the caller
// logs it on compact/summary (the reconstructability RFC).
expect(model).toBe('test-model')
expect(maxTokens).toBe(512)
// The fixed system prompt and maxTokens flow through.
expect(adapter.lastOptions!.system).toContain('compaction engine')
expect(adapter.lastOptions!.system).toContain('## Next Step')
@@ -1035,7 +1042,7 @@ describe('BasicCompactService.summarize (real ctx.llm.stream)', () => {
])
const svc = new BasicCompactService(ctx, cfg({ auto: false }))
const summary = await summarize(svc, 'User: hi', 'test-model')
const { summary } = await summarize(svc, 'User: hi', 'test-model')
expect(summary).toEqual([{ type: 'text', text: 'PUBLIC SUMMARY' }])
})
@@ -1231,15 +1238,20 @@ describe('BasicCompactService auto-compaction (agent/pre-step listener)', () =>
expect(session.events.some(e => e.type === 'compact/start')).toBe(false)
})
it('routes summarization through agent/request so router agents can choose the model', async () => {
it('summarization is interceptable at llm/stream (model routing for direct calls)', async () => {
const { ctx, adapter } = await ctxWithModel('ROUTED SUMMARY', 'routed-model')
ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
// The summarize call is a direct one-shot model call, not a loop step: it
// does not run agent/request (that seam shapes the loop's conversation
// requests). llm/stream is its interception surface, and a hand-built
// request is not frozen, so mutate-then-next model routing works — the
// adapter resolves AFTER the waterfall, so the rewrite picks the adapter.
ctx.on('llm/stream', (options, next) => {
options.model = 'routed-model'
return next()
})
void new BasicCompactService(ctx, cfg({ contextWindow: 200, thresholdRatio: 0.5, retainTokens: 20 }))
const session = multiTurnSession(5, 1)
const agent = stubAgent(session)
const agent = stubAgent(session, 'agent-model')
await ctx.serial('agent/pre-step', agent, 1, 1, '', SIGNAL)

4
packages/compact/compact-basic/tests/compact-loop-repro.spec.ts
View File

@@ -41,8 +41,8 @@ class ReproCompactService extends BasicCompactService {
return blocks.length * TOKENS_PER_BLOCK
}
override async summarize(): Promise<ContentBlock[]> {
return [{ type: 'text', text: 'CHECKPOINT SUMMARY' }]
override async summarize(): Promise<{ summary: ContentBlock[]; model: string }> {
return { summary: [{ type: 'text', text: 'CHECKPOINT SUMMARY' }], model: 'stub' }
}
}

4
packages/compact/compact/README.md
View File

@@ -18,8 +18,8 @@ Both methods are **abstract** — the backend owns the entire strategy (token es
| Member | Semantics |
|---|---|
| `compactIfNeeded(agent, turn, step, fullSystemPrompt, signal)` | Estimate the surface-derived history size; if over the backend's threshold, compact an older range via `compactRegion`, keeping recent context intact. Returns the `CompactionResult`, or `null` if nothing needed compacting. All parameters required — the loop's `agent/pre-step` checkpoint supplies the agent, lifecycle context, assembled `fullSystemPrompt`, and turn `signal`; router-aware summarizers can use the agent lifecycle context to route their own model call through `agent/request`. |
| `compactRegion(session, start, end, agent, turn, step, signal?)` | Forcibly summarize surface nodes `[start, end]` (inclusive seqs) into a single replacement node. **Throws** if a compaction is already in progress, if `start`/`end` aren't surface nodes, or if `start` is positioned after `end` on the surface. The range is a SURFACE-POSITION span, not a numeric seq interval — after a prior replace lands a fresh high-seq summary node at the shadowed range's position, surface order no longer tracks seq order. |
| `compactIfNeeded(agent, fullSystemPrompt, signal)` | Estimate the surface-derived history size; if over the backend's threshold, compact an older range via `compactRegion`, keeping recent context intact. Returns the `CompactionResult`, or `null` if nothing needed compacting. All parameters required — the loop's `agent/pre-step` checkpoint supplies the agent, assembled `fullSystemPrompt`, and turn `signal`. A backend's summarization request is a direct `ctx.llm.stream()` call (not a loop step), so per-call interception happens at `llm/stream`. |
| `compactRegion(session, start, end, agent, signal?)` | Forcibly summarize surface nodes `[start, end]` (inclusive seqs) into a single replacement node. **Throws** if a compaction is already in progress, if `start`/`end` aren't surface nodes, or if `start` is positioned after `end` on the surface. The range is a SURFACE-POSITION span, not a numeric seq interval — after a prior replace lands a fresh high-seq summary node at the shadowed range's position, surface order no longer tracks seq order. |
`compactIfNeeded` takes a required `signal`; `compactRegion`'s is optional. A backend that summarizes via `ctx.llm.stream()` **must** forward it into the call's `GenerateOptions.signal`, so an abort or fiber dispose tears down the in-flight summarization instead of leaving an orphaned model call running past the cancellation. The session being compacted comes from the agent context; the turn that the `compact/*` events belong to is recoverable from the log (the currently-open turn), so the backend stamps it from the log rather than trusting a caller-supplied value.

8
packages/compact/compact/src/index.ts
View File

@@ -91,8 +91,6 @@ export abstract class CompactService extends Service {
* over-budget. Bounding an individual unit's size is a separate concern.
*
* @param agent - agent context owning the session surface and model options.
* @param turn - turn number of the pre-step checkpoint.
* @param step - step number about to start.
* @param fullSystemPrompt - assembled system prompt, counted toward the estimate.
* @param signal - cancellation signal. A backend summarizing via
* `ctx.llm.stream()` MUST forward this into the call's `GenerateOptions.signal`
@@ -102,8 +100,6 @@ export abstract class CompactService extends Service {
*/
abstract compactIfNeeded(
agent: CompactAgentContext,
turn: number,
step: number,
fullSystemPrompt: string,
signal: AbortSignal,
): Promise<CompactionResult | null>
@@ -129,8 +125,6 @@ export abstract class CompactService extends Service {
* @param start - inclusive seq of the first surface node to compact.
* @param end - inclusive seq of the last surface node to compact.
* @param agent - agent context used by router-aware summarizers.
* @param turn - lifecycle turn forwarded to request-routing seams.
* @param step - lifecycle step forwarded to request-routing seams.
* @param signal - optional cancellation signal. A backend that summarizes via
* `ctx.llm.stream()` MUST forward this into the call's `GenerateOptions.signal`
* so an abort/dispose tears down the in-flight summarization rather than
@@ -148,8 +142,6 @@ export abstract class CompactService extends Service {
start: number,
end: number,
agent: CompactAgentContext,
turn: number,
step: number,
signal?: AbortSignal,
): Promise<CompactionResult>
}

9
packages/compact/compact/src/types.ts
View File

@@ -32,6 +32,15 @@ declare module '@deepseek-ai/dsh-session' {
shadowedRange: { start: number; end: number }
shadowedSeqs: number[]
shadowedTokenCount: number
/**
* The model that wrote the summary — the summarize call's envelope,
* reported by the backend that made the call, logged so the one-shot
* request is reconstructable from log + code and "which model wrote
* this summary" has a durable answer (the reconstructability RFC).
*/
model: string
/** The generation cap the summarize call sent, when one applied. */
maxTokens?: number
}
/** Marks the end of a compaction — log-only, releases the lock. `error` set if summarization failed. */
'compact/end': { turn: number; error?: string }

13
packages/compact/compact/tests/compact.spec.ts
View File

@@ -17,8 +17,6 @@ class StubCompactService extends CompactService {
override async compactIfNeeded(
_agent: CompactAgentContext,
_turn: number,
_step: number,
_fullSystemPrompt: string,
signal: AbortSignal,
): Promise<CompactionResult | null> {
@@ -31,8 +29,6 @@ class StubCompactService extends CompactService {
start: number,
end: number,
_agent: CompactAgentContext,
_turn: number,
_step: number,
signal?: AbortSignal,
): Promise<CompactionResult> {
this.lastSignal = signal
@@ -43,6 +39,7 @@ class StubCompactService extends CompactService {
shadowedRange: { start, end },
shadowedSeqs: [],
shadowedTokenCount: 0,
model: 'stub',
})
const endEvent = session.append('compact/end', { turn: 0 })
return {
@@ -81,7 +78,7 @@ describe('CompactService seam', () => {
const ctx = new Context()
const svc = new StubCompactService(ctx)
const session = new Session(SessionId('s'))
expect(await svc.compactIfNeeded(stubAgent(session), 1, 1, '', new AbortController().signal)).toBeNull()
expect(await svc.compactIfNeeded(stubAgent(session), '', new AbortController().signal)).toBeNull()
})
it('compact/* events merge into SessionEventMap and are log-only', async () => {
@@ -89,7 +86,7 @@ describe('CompactService seam', () => {
const svc = new StubCompactService(ctx)
const session = new Session(SessionId('s'))
const result = await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'), 1, 1)
const result = await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'))
const startEvent = session.events.find(e => e.type === 'compact/start')
expect(startEvent).toBeDefined()
@@ -107,10 +104,10 @@ describe('CompactService seam', () => {
const session = new Session(SessionId('s'))
const controller = new AbortController()
await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'), 1, 1, controller.signal)
await svc.compactRegion(session, 0, 0, stubAgent(session, 'm'), controller.signal)
expect(svc.lastSignal).toBe(controller.signal)
await svc.compactIfNeeded(stubAgent(session), 1, 1, '', controller.signal)
await svc.compactIfNeeded(stubAgent(session), '', controller.signal)
expect(svc.lastSignal).toBe(controller.signal)
})
})

10
packages/core/agent-core/tests/agent-core.spec.ts
View File

@@ -41,9 +41,8 @@ function waitForMainIdle(ctx: Context): Promise<void> {
})
}
function firstText(message: Message | undefined): string | undefined {
const block = message?.content[0]
return block?.type === 'text' ? block.text : undefined
function messageText(message: Message | undefined): string {
return message?.content.map(block => block.type === 'text' ? block.text : '').join('\n') ?? ''
}
describe('dsh-agent-core bundle', () => {
@@ -110,8 +109,9 @@ describe('dsh-agent-core bundle', () => {
agent.send([{ type: 'text', text: 'hi' }])
await waitForMainIdle(ctx)
expect(adapter.requests[0]?.messages[0]?.role).toBe('user')
expect(firstText(adapter.requests[0]?.messages[0])).toContain('bundled project rule')
const sentText = adapter.requests[0]?.messages.map(messageText).join('\n')
expect(sentText).toContain('hi')
expect(sentText).toContain('bundled project rule')
expect(adapter.requests[0]?.system).toContain('You are an AI agent powered by the DeepSeek Harness SDK.')
expect(adapter.requests[0]?.system).not.toContain('bundled project rule')
await handle.dispose()

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

@@ -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')

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

@@ -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
View 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' })
}
}

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

@@ -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
View 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
View 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
View 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)
})
})
})

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

@@ -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')

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

@@ -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.

26
packages/core/agent/src/types.ts
View File

@@ -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, …).

9
packages/core/session/README.md
View File

@@ -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'`.

102
packages/core/session/src/index.ts
View File

@@ -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
View 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
}

21
packages/core/session/src/surface.ts
View File

@@ -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
}
}

87
packages/core/session/src/types.ts
View File

@@ -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
View 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()
})
})

8
packages/core/session/tests/properties.spec.ts
View File

@@ -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
View 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/)
})
})

20
packages/core/session/tests/session.spec.ts
View File

@@ -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' }])
})

23
packages/core/session/tests/surface.spec.ts
View File

@@ -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)
})
})

4
packages/llm/llm/README.md
View File

@@ -29,6 +29,10 @@ Messages are arrays of typed content blocks: `text`, `reasoning`, `tool-call`, `
Streaming is a raw chunk protocol (`block-start`, `text-delta`, `reasoning-delta`, `tool-call-delta`, `block-end`, `usage`, `finish`). `BlockAssembler` is the single shared implementation that assembles chunks into blocks/messages.
### Call configuration (`call-config.ts`)
`LlmCallConfig` is the model + sampling scalars of one conversation's requests (`model`, `temperature`, `maxTokens`, `stop` — each mapping 1:1 onto the same-named `GenerateOptions` field). It is per-conversation state recorded in the session log as part of the request header (see the dsh-session `request/header` events), never a silently-adjustable per-call knob: the `agent/request` waterfall proposes a replacement and the loop logs a real change. `callConfigEquals(a, b)` is the field-wise real-change detector; `deepFreeze(value)` is the ownership helper the loop applies to every built request before dispatch (`llm/stream` listeners and adapters read, never rewrite).
### App attribution (`attribution.ts`)
Every product adapter must identify the application on every provider HTTP request - attribution is part of the adapter contract, not an adapter-local nicety. `attributionHeaders(identity?)` builds the standard `User-Agent` header (`product/version (+url)`, from `userAgent()`) for every request. The default `APP_IDENTITY` carries only static public product facts (its version is read from this package's manifest); a white-label deployment passes its own `AppIdentity`, and omission falls back to the default - nothing can suppress attribution. OpenRouter-specific app attribution headers are intentionally not supported by this contract. An adapter proves compliance with a wire-level test: a mock server asserting the received header (or, for a library-backed adapter, that the library's header hook delivers the same value). Policy and rationale: [Mandatory `User-Agent` attribution](../../../docs/rfc/implemented/architecture/2026-06-21-mandatory-app-attribution-headers.md).

69
packages/llm/llm/src/call-config.ts Normal file
View File

@@ -0,0 +1,69 @@
/**
* The call configuration of a conversation and its comparison/freeze
* utilities. `LlmCallConfig` is the non-content third of the request header
* (see `EpochHeader` in dsh-session): everything about a request besides its
* message content that can undermine provider KV-cache reuse — `model`
* selects the cache namespace outright, and the sampling scalars are treated
* the same way out of caution. It is per-conversation state recorded in the
* session log (the reconstructability RFC), never a silently-drifting
* per-call knob: the `agent/request` waterfall proposes a replacement, and
* the loop logs a real change as a `request/header-delta` event.
*
* @module dsh-llm/call-config
*/
/**
* Model + sampling scalars of one conversation's requests. Every field maps
* 1:1 onto the same-named `GenerateOptions` field; the loop builds requests
* from the logged header rather than accepting these per call.
*/
export interface LlmCallConfig {
model: string
temperature?: number
maxTokens?: number
stop?: string[]
}
/**
* Field-wise equality over {@link LlmCallConfig} — the comparison a caller
* runs to decide whether a proposed configuration is a real change (worth a
* logged header delta) or the held one restated.
* @param a - one configuration.
* @param b - the other.
* @returns whether every field (including the `stop` list, element-wise) matches.
*/
export function callConfigEquals(a: LlmCallConfig, b: LlmCallConfig): boolean {
if (a.model !== b.model || a.temperature !== b.temperature || a.maxTokens !== b.maxTokens) return false
if (a.stop === undefined || b.stop === undefined) return a.stop === b.stop
return a.stop.length === b.stop.length && a.stop.every((s, i) => s === b.stop?.[i])
}
/**
* Deep-freeze a value in place so any later mutation throws (ESM code runs in
* strict mode), and return it. The loop freezes every request it builds
* before dispatch — `llm/stream` listeners and adapters read the request,
* never rewrite it, so the wire bytes cannot silently desync from what the
* session log reconstructs. Guards against cycles with a WeakSet: loop-built
* requests hold `structuredClone`d JSON-validated session data, but the
* helper accepts arbitrarily constructed values. One exemption: an
* `AbortSignal` is never entered or frozen — it is the request's live
* cancellation channel, and freezing one breaks `AbortController.abort()`
* outright (Node stores the aborted flag as an own property of the signal).
* @param value - the value to freeze in place.
* @returns the same value, frozen.
*/
export function deepFreeze<T>(value: T): T {
const seen = new WeakSet<object>()
const walk = (node: unknown): void => {
if (node === null || typeof node !== 'object') return
if (node instanceof AbortSignal) return
if (seen.has(node)) return
seen.add(node)
Object.freeze(node)
for (const key of Object.keys(node)) {
walk((node as Record<string, unknown>)[key])
}
}
walk(value)
return value
}

10
packages/llm/llm/src/index.ts
View File

@@ -16,6 +16,8 @@ export * from './never.ts'
export * from './error.ts'
export * from './types.ts'
export { BlockAssembler } from './assembler.ts'
export { callConfigEquals, deepFreeze } from './call-config.ts'
export type { LlmCallConfig } from './call-config.ts'
declare module 'cordis' {
interface Context {
@@ -24,10 +26,14 @@ declare module 'cordis' {
interface Events {
/**
* Waterfall around every streaming model call (retry, caching, routing).
* Waterfall around every streaming model call (retry, replay, routing).
* Bound to the {@link LlmService}; call `next()` to reach the resolved
* adapter's stream, or yield your own chunks to short-circuit.
* @param options - the full request; listeners may rewrite it before delegating.
* @param options - the full request. A LOOP-built request arrives
* deep-frozen (mutation throws): its content is a pure function of the
* session log (the reconstructability RFC), so listeners read it, never
* rewrite it. A hand-built one-shot (compaction summarize) is the
* caller's own object and stays mutable here.
* @mode waterfall
*/
'llm/stream'(this: LlmService, options: GenerateOptions, next: () => AsyncIterable<StreamChunk>): AsyncIterable<StreamChunk>

56
packages/llm/llm/tests/call-config.spec.ts Normal file
View File

@@ -0,0 +1,56 @@
/**
* call-config unit tests: field-wise LlmCallConfig equality (the real-change
* detector behind logged header deltas) and the deepFreeze ownership helper
* the loop applies to every built request.
*/
import { describe, expect, it } from 'vitest'
import { callConfigEquals, deepFreeze } from '../src/call-config.ts'
describe('callConfigEquals', () => {
it('compares every field, including the stop list element-wise', () => {
expect(callConfigEquals({ model: 'm' }, { model: 'm' })).toBe(true)
expect(callConfigEquals({ model: 'm' }, { model: 'x' })).toBe(false)
expect(callConfigEquals({ model: 'm', temperature: 0.5 }, { model: 'm' })).toBe(false)
expect(callConfigEquals({ model: 'm', maxTokens: 1 }, { model: 'm', maxTokens: 2 })).toBe(false)
expect(callConfigEquals({ model: 'm', stop: ['a'] }, { model: 'm' })).toBe(false)
expect(callConfigEquals({ model: 'm', stop: ['a'] }, { model: 'm', stop: ['a', 'b'] })).toBe(false)
expect(callConfigEquals({ model: 'm', stop: ['a'] }, { model: 'm', stop: ['b'] })).toBe(false)
expect(callConfigEquals({ model: 'm', stop: ['a', 'b'] }, { model: 'm', stop: ['a', 'b'] })).toBe(true)
})
})
describe('deepFreeze', () => {
it('freezes nested structure in place and returns the same reference', () => {
const value = { a: { b: [1, { c: 'x' }] } }
const frozen = deepFreeze(value)
expect(frozen).toBe(value)
expect(Object.isFrozen(value)).toBe(true)
expect(Object.isFrozen(value.a)).toBe(true)
expect(Object.isFrozen(value.a.b)).toBe(true)
expect(Object.isFrozen(value.a.b[1])).toBe(true)
// ESM runs in strict mode: mutation throws rather than silently failing.
expect(() => { (value.a.b[1] as { c: string }).c = 'y' }).toThrow(TypeError)
})
it('never freezes an AbortSignal: the live cancellation channel keeps working', () => {
const controller = new AbortController()
const request = deepFreeze({ model: 'm', signal: controller.signal })
expect(Object.isFrozen(request)).toBe(true)
expect(Object.isFrozen(controller.signal)).toBe(false)
let fired = false
controller.signal.addEventListener('abort', () => { fired = true }, { once: true })
controller.abort('stop')
expect(fired).toBe(true)
expect(controller.signal.aborted).toBe(true)
})
it('passes primitives through and terminates on cycles', () => {
expect(deepFreeze(42)).toBe(42)
expect(deepFreeze(null)).toBeNull()
const cyclic = { self: undefined as unknown }
cyclic.self = cyclic
deepFreeze(cyclic)
expect(Object.isFrozen(cyclic)).toBe(true)
})
})

8
packages/prompt/project-instructions/README.md
View File

@@ -4,15 +4,15 @@ Project instruction file loader for the harness. It discovers the configured per
## Behavior
The plugin listens on the `agent/request` waterfall and reads instruction file content through the `ctx.fs` provider seam. It uses `ctx.fs.lstat` before `ctx.fs.resolve` so repository-owned instruction symlinks are skipped rather than followed across trust boundaries. It deliberately does not declare `fs` as a static dependency: `agent-core` can load the plugin in providerless app trees, and the plugin simply does nothing until a filesystem provider is present at request/tool time. For each request it derives the workspace from `agent.session.header.cwd`; if the session has no cwd, it falls back to `process.cwd()` for single-session local/stdio runs. It then finds the project root by walking upward until it sees `.git` as either a directory or a file, considers the ancestor chain from project root to cwd, and loads at most one instruction file per directory by checking `instructionFileCandidates` in order. With the default candidates, `AGENTS.md` wins and `CLAUDE.md` is a compatibility fallback.
The plugin listens on the `agent/pre-step` checkpoint and reads instruction file content through the `ctx.fs` provider seam before the loop snapshots `deriveMessages()` for the next model request. It uses `ctx.fs.lstat` before `ctx.fs.resolve` so repository-owned instruction symlinks are skipped rather than followed across trust boundaries. It deliberately does not declare `fs` as a static dependency: `agent-core` can load the plugin in providerless app trees, and the plugin simply does nothing until a filesystem provider is present at request/tool time. For each request it derives the workspace from `agent.session.header.cwd`; if the session has no cwd, it falls back to `process.cwd()` for single-session local/stdio runs. It then finds the project root by walking upward until it sees `.git` as either a directory or a file, considers the ancestor chain from project root to cwd, and loads at most one instruction file per directory by checking `instructionFileCandidates` in order. With the default candidates, `AGENTS.md` wins and `CLAUDE.md` is a compatibility fallback.
The plugin also listens on `tools/post-execute` for successful structured filesystem touches from the first-party `read`, `write`, and `edit` tools. When one of those tools touches a descendant of the session cwd, the plugin checks the directories between the session cwd and the touched file for instruction files that are not already visible in session context, then attaches them as `additionalContext` so the loop records a durable `context/message` for the next model request. This intentionally follows file-tool touches, not shell `cd`: `dsh-bash-local` uses fresh shells per call, and parsing arbitrary shell commands for reached paths would be brittle.
User-global instructions live at `$DSH_HOME/AGENTS.md`; `$DSH_HOME` defaults to `~/.dsh`. A configured `~`, `~/...`, or Windows-style `~\...` prefix is expanded against the operating-system home directory before resolution. The user-global file name is harness-level and is not affected by `instructionFileCandidates`, which only controls per-directory project and nested discovery. The user-global file renders before project files, so deeper project files appear later in the context and can override broader guidance.
Baseline files are inserted as a synthetic user-role workspace-context message, not as provider system text and not as persisted session events. Nested files discovered after structured file tools run are inserted through the existing `context/message` path, so they persist with the session and resume like other plugin-provided context. Nested duplicate suppression is derived from the visible session surface plus a short pending window before the loop records `additionalContext`; if compaction removes a nested context message from the surface, a later structured file touch may re-load it so the next model request still sees the applicable guidance. The rendered envelope states that these files are workspace-provided guidance, lower authority than system/developer/direct user instructions, and must not override safety, permission, or secret-handling rules.
Baseline files are inserted through `agent.inject()` as durable `context/message` entries before the request boundary, not as provider system text and not by mutating the frozen request. Nested files discovered after structured file tools run use the same `context/message` path via `additionalContext`, so both baseline and nested guidance persist with the session and resume like other plugin-provided context. Duplicate suppression is derived from the visible session surface plus, for nested tool-time loads, a short pending window before the loop records `additionalContext`; if compaction removes an instruction context message from the surface, a later pre-step or structured file touch may re-load it so the next model request still sees the applicable guidance. The rendered envelope states that these files are workspace-provided guidance, lower authority than system/developer/direct user instructions, and must not override safety, permission, or secret-handling rules.
The baseline hook currently runs for every `agent/request`, including maintenance model calls such as compaction summarization. `GenerateOptions` does not yet carry a request-kind marker, so the plugin cannot distinguish user-facing turns from summarization without brittle prompt sniffing. A future request marker should let prompt-context plugins opt out of maintenance calls deliberately.
Because baseline loading runs on `agent/pre-step`, it only targets agent conversation requests. One-shot maintenance model calls such as compaction summarization do not pass through this checkpoint.
## Config
@@ -31,7 +31,7 @@ export interface Config {
The renderer keeps full text until the configured byte budget is exceeded. When it must trim, it preserves more-specific files first, drops whole less-specific files before truncating a more-specific file, and emits an HTML comment naming omitted and truncated files with byte counts.
Discovery re-walks the applicable ancestor chain on every request so newly created baseline files are noticed. File content is cached by normalized absolute path plus the provider's opaque file version and size; a changed signature causes a re-read. The discovery pass carries the file signature forward to the read pass, so a cache hit does not stat the same instruction file twice in one request. Nested instruction paths are de-duplicated from recorded session context rather than from the content cache, so cache eviction or repeated reads do not duplicate still-visible durable context.
Discovery re-walks the applicable ancestor chain on every pre-step so newly created baseline files are noticed. File content is cached by normalized absolute path plus the provider's opaque file version and size; a changed signature causes a re-read. The discovery pass carries the file signature forward to the read pass, so a cache hit does not stat the same instruction file twice in one request. Instruction paths are de-duplicated from visible recorded session context rather than from the content cache, so cache eviction or repeated reads do not duplicate still-visible durable context.
## Non-goals

27
packages/prompt/project-instructions/src/index.ts
View File

@@ -10,7 +10,6 @@ import { lstat, readFile, stat } from 'node:fs/promises'
import { dirname, isAbsolute, join, relative, resolve } from 'node:path'
import type { Context } from 'cordis'
import z from 'schemastery'
import type { GenerateOptions, Message } from '@deepseek-ai/dsh-llm'
import type { Agent, HookContext } from '@deepseek-ai/dsh-agent'
import type { FileSystem, FsTarget } from '@deepseek-ai/dsh-fs'
import { DEFAULT_DSH_HOME_DISPLAY, defaultDshHome, resolveDshHome } from '@deepseek-ai/dsh-paths'
@@ -481,10 +480,6 @@ export function renderProjectInstructions(files: LoadedInstructionFile[], option
return { text, omitted, truncated }
}
function workspaceContextMessage(text: string): Message {
return { role: 'user', content: [{ type: 'text', text }] }
}
function workspaceContextHook(text: string): HookContext {
return { content: [{ type: 'text', text }], source: PLUGIN_SOURCE }
}
@@ -531,7 +526,7 @@ function instructionDisplayPathsFromContextContent(content: readonly { type: str
return paths
}
function loadedNestedInstructionDisplayPaths(agent: Agent, pendingDisplayPaths: Set<string>): Set<string> {
function visibleInstructionDisplayPaths(agent: Agent): { visible: Set<string>; logged: Set<string> } {
const visibleSeqs = new Set(agent.session.surface.nodes.map(node => node.seq))
const visible = new Set<string>()
const logged = new Set<string>()
@@ -543,6 +538,11 @@ function loadedNestedInstructionDisplayPaths(agent: Agent, pendingDisplayPaths:
if (visibleSeqs.has(seq)) visible.add(displayPath)
}
}
return { visible, logged }
}
function loadedNestedInstructionDisplayPaths(agent: Agent, pendingDisplayPaths: Set<string>): Set<string> {
const { visible, logged } = visibleInstructionDisplayPaths(agent)
// The loop records returned additionalContext shortly after this plugin
// returns it. Once the durable log contains that marker anywhere, clear the
// temporary pending bit; load decisions still use visible surface state so
@@ -591,10 +591,10 @@ export function apply(ctx: Context, config: Config): void {
const resolved = resolveConfig(config)
const cache: InstructionContentCache = new Map()
const pendingNestedDisplayPaths = new WeakMap<object, Set<string>>()
ctx.on('agent/request', async (agent: Agent, _turn: number, _step: number, request: GenerateOptions, next) => {
if (resolved.baselineMaxBytes <= 0 || !Number.isFinite(resolved.baselineMaxBytes)) return next()
ctx.on('agent/pre-step', async (agent: Agent) => {
if (resolved.baselineMaxBytes <= 0 || !Number.isFinite(resolved.baselineMaxBytes)) return
const fileSystem = ctx.get('fs')
if (fileSystem === undefined) return next()
if (fileSystem === undefined) return
/* v8 ignore next -- stdio compatibility fallback; tests avoid process.chdir() because cwd is process-global. */
const cwd = agent.session.header.cwd ?? process.cwd()
const instructions = await loadBaselineInstructions({
@@ -605,10 +605,11 @@ export function apply(ctx: Context, config: Config): void {
instructionFileCandidates: resolved.instructionFileCandidates,
cache,
}, fileSystem)
if (instructions !== undefined) {
request.messages = [workspaceContextMessage(instructions.text), ...request.messages]
}
return next()
if (instructions === undefined) return
const visibleDisplayPaths = visibleInstructionDisplayPaths(agent).visible
const baselineDisplayPaths = instructionDisplayPathsFromText(instructions.text)
if (baselineDisplayPaths.length > 0 && baselineDisplayPaths.every(path => visibleDisplayPaths.has(path))) return
agent.inject(workspaceContextHook(instructions.text).content, { source: PLUGIN_SOURCE })
})
ctx.on('tools/post-execute', async (exec: ToolExecution, result: ToolExecutionResult, next): Promise<PostToolDecision> => {
const downstream = await next()

226
packages/prompt/project-instructions/tests/project-instructions.spec.ts
View File

@@ -5,7 +5,7 @@ import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import Loader from '@cordisjs/plugin-loader'
import * as projectInstructions from '@deepseek-ai/dsh-project-instructions'
import { CallId, type GenerateOptions } from '@deepseek-ai/dsh-llm'
import { CallId } from '@deepseek-ai/dsh-llm'
import { Session, SessionId, SESSION_FORMAT_VERSION } from '@deepseek-ai/dsh-session'
import type { Agent, HookContext } from '@deepseek-ai/dsh-agent'
import { AgentId } from '@deepseek-ai/dsh-agent'
@@ -122,17 +122,14 @@ function stubAgent(cwd?: string): Agent {
status: 'idle',
send() {},
steer() {},
inject() {},
inject(content, options) {
session.append('context/message', { content, source: options?.source ?? { kind: 'user' } }, { surfaceOp: 'append' })
},
cancel() {},
whenIdle: () => Promise.resolve(),
}
}
function firstText(message: GenerateOptions['messages'][number] | undefined): string | undefined {
const block = message?.content[0]
return block?.type === 'text' ? block.text : undefined
}
function blocksText(blocks: { type: string; text?: string }[] | undefined): string {
return blocks?.map(block => block.type === 'text' ? block.text ?? '' : '').join('\n') ?? ''
}
@@ -146,6 +143,18 @@ function appendAdditionalContext(agent: Agent, result: { additionalContext?: Hoo
}, { surfaceOp: 'append' }).seq
}
async function runBaselinePreStep(ctx: Context, agent: Agent): Promise<void> {
await ctx.serial('agent/pre-step', agent, 1, 1, '', AbortSignal.timeout(1000))
}
function derivedText(agent: Agent): string {
return blocksText(agent.session.deriveMessages()[0]?.content)
}
function expectNoDerivedMessages(agent: Agent): void {
expect(agent.session.deriveMessages()).toEqual([])
}
describe('project instruction discovery', () => {
it('loads user-global first, then root-to-cwd project instructions using the default candidate order', async () => {
const root = await tempRepo()
@@ -275,14 +284,11 @@ describe('project instruction discovery', () => {
await symlink(join(outside, 'secret.txt'), join(root, 'AGENTS.md'))
const ctx = new Context()
await mountProjectInstructions(ctx, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(result.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }])
expectNoDerivedMessages(agent)
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -631,18 +637,15 @@ describe('project instruction request injection', () => {
expect('inject' in projectInstructions).toBe(false)
})
it('leaves requests unchanged when no filesystem provider is present', async () => {
it('does not inject baseline context when no filesystem provider is present', async () => {
const ctx = new Context()
try {
await ctx.plugin(projectInstructions, {})
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const agent = stubAgent('/virtual/repo')
const result = await ctx.waterfall('agent/request', stubAgent('/virtual/repo'), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(result.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }])
expectNoDerivedMessages(agent)
} finally {
await ctx.fiber.dispose()
}
@@ -673,7 +676,7 @@ describe('project instruction request injection', () => {
}
})
it('prepends a synthetic user workspace-context message without mutating the system prompt', async () => {
it('injects baseline workspace context through the session log before step derivation', async () => {
const root = await tempRepo()
const home = await tempRepo()
try {
@@ -681,20 +684,35 @@ describe('project instruction request injection', () => {
await write(join(root, 'AGENTS.md'), 'repo rule')
const ctx = new Context()
await mountProjectInstructions(ctx, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
system: 'real system',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(result.system).toBe('real system')
expect(result.messages).toHaveLength(2)
expect(result.messages[0]?.role).toBe('user')
expect(firstText(result.messages[0])).toContain('<workspace-context source="project-instruction-files">')
expect(firstText(result.messages[0])).toContain('repo rule')
expect(result.messages[1]).toEqual({ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] })
expect(agent.session.deriveMessages()).toHaveLength(1)
expect(derivedText(agent)).toContain('<context source="plugin">')
expect(derivedText(agent)).toContain('<workspace-context source="project-instruction-files">')
expect(derivedText(agent)).toContain('repo rule')
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
}
})
it('does not duplicate still-visible baseline workspace context on later pre-step checks', async () => {
const root = await tempRepo()
const home = await tempRepo()
try {
await mkdir(join(root, '.git'), { recursive: true })
await write(join(root, 'AGENTS.md'), 'repo rule')
const ctx = new Context()
await mountProjectInstructions(ctx, { dshHome: home })
const agent = stubAgent(root)
await runBaselinePreStep(ctx, agent)
await runBaselinePreStep(ctx, agent)
expect(agent.session.events.filter(event => event.type === 'context/message')).toHaveLength(1)
expect(derivedText(agent)).toContain('repo rule')
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -713,15 +731,12 @@ describe('project instruction request injection', () => {
fs.entries.set(join(root, '.git'), { type: 'directory' })
fs.entries.set(join(root, 'AGENTS.md'), { type: 'file', content: 'ctx.fs rule' })
await ctx.plugin(projectInstructions, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(firstText(result.messages[0])).toContain('ctx.fs rule')
expect(firstText(result.messages[0])).not.toContain('node fs rule')
expect(derivedText(agent)).toContain('ctx.fs rule')
expect(derivedText(agent)).not.toContain('node fs rule')
expect(fs.readTargets).toEqual([join(root, 'AGENTS.md')])
} finally {
await rm(root, { recursive: true, force: true })
@@ -739,14 +754,11 @@ describe('project instruction request injection', () => {
fs.entries.set(join(root, '.git'), { type: 'directory' })
fs.entries.set(join(root, 'AGENTS.md'), { type: 'file', content: 'provider-only rule' })
await ctx.plugin(projectInstructions, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(firstText(result.messages[0])).toContain('provider-only rule')
expect(derivedText(agent)).toContain('provider-only rule')
expect(fs.readTargets).toEqual([join(root, 'AGENTS.md')])
} finally {
await ctx.fiber.dispose()
@@ -769,17 +781,14 @@ describe('project instruction request injection', () => {
fs.entries.set(join(home, 'AGENTS.md'), { type: 'file', content: 'ctx global rule' })
fs.entries.set(join(root, 'CLAUDE.md'), { type: 'file', content: 'ctx claude rule' })
await ctx.plugin(projectInstructions, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(firstText(result.messages[0])).toContain('ctx global rule')
expect(firstText(result.messages[0])).toContain('ctx claude rule')
expect(firstText(result.messages[0])).not.toContain('node global rule')
expect(firstText(result.messages[0])).not.toContain('node claude rule')
expect(derivedText(agent)).toContain('ctx global rule')
expect(derivedText(agent)).toContain('ctx claude rule')
expect(derivedText(agent)).not.toContain('node global rule')
expect(derivedText(agent)).not.toContain('node claude rule')
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -798,14 +807,11 @@ describe('project instruction request injection', () => {
fs.entries.set(join(root, '.git'), { type: 'directory' })
fs.entries.set(join(root, 'AGENTS.md'), { type: 'directory' })
await ctx.plugin(projectInstructions, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(result.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }])
expectNoDerivedMessages(agent)
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -825,14 +831,11 @@ describe('project instruction request injection', () => {
fs.entries.set(join(root, 'AGENTS.md'), { type: 'directory' })
fs.lstatTypes.set(join(root, 'AGENTS.md'), 'file')
await ctx.plugin(projectInstructions, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(result.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }])
expectNoDerivedMessages(agent)
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -851,14 +854,11 @@ describe('project instruction request injection', () => {
fs.entries.set(join(root, '.git'), { type: 'directory' })
fs.entries.set(join(root, 'AGENTS.md'), { type: 'file' })
await ctx.plugin(projectInstructions, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(firstText(result.messages[0])).toContain('## AGENTS.md')
expect(derivedText(agent)).toContain('## AGENTS.md')
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -877,14 +877,11 @@ describe('project instruction request injection', () => {
fs.entries.set(join(root, '.git'), { type: 'directory' })
fs.throwOnStat.add(join(root, 'AGENTS.md'))
await ctx.plugin(projectInstructions, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(result.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }])
expectNoDerivedMessages(agent)
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -903,14 +900,11 @@ describe('project instruction request injection', () => {
fs.throwOnStat.add(join(root, '.git'))
fs.entries.set(join(root, 'AGENTS.md'), { type: 'file', content: 'repo rule' })
await ctx.plugin(projectInstructions, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(firstText(result.messages[0])).toContain('repo rule')
expect(derivedText(agent)).toContain('repo rule')
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -928,16 +922,16 @@ describe('project instruction request injection', () => {
await write(join(repoB, 'AGENTS.md'), 'repo B only')
const ctx = new Context()
await mountProjectInstructions(ctx, { dshHome: home })
const requestA: GenerateOptions = { model: 'mock', messages: [{ role: 'user', content: [{ type: 'text', text: 'A' }] }] }
const requestB: GenerateOptions = { model: 'mock', messages: [{ role: 'user', content: [{ type: 'text', text: 'B' }] }] }
const agentA = stubAgent(repoA)
const agentB = stubAgent(repoB)
const resultA = await ctx.waterfall('agent/request', stubAgent(repoA), 1, 1, requestA, async () => requestA)
const resultB = await ctx.waterfall('agent/request', stubAgent(repoB), 1, 1, requestB, async () => requestB)
await runBaselinePreStep(ctx, agentA)
await runBaselinePreStep(ctx, agentB)
expect(firstText(resultA.messages[0])).toContain('repo A only')
expect(firstText(resultA.messages[0])).not.toContain('repo B only')
expect(firstText(resultB.messages[0])).toContain('repo B only')
expect(firstText(resultB.messages[0])).not.toContain('repo A only')
expect(derivedText(agentA)).toContain('repo A only')
expect(derivedText(agentA)).not.toContain('repo B only')
expect(derivedText(agentB)).toContain('repo B only')
expect(derivedText(agentB)).not.toContain('repo A only')
} finally {
await rm(repoA, { recursive: true, force: true })
await rm(repoB, { recursive: true, force: true })
@@ -956,19 +950,19 @@ describe('project instruction request injection', () => {
const ctx = new Context()
await ctx.plugin(LocalFileSystem, { cwd: '/' })
await ctx.plugin(projectInstructions, {})
const request: GenerateOptions = { model: 'mock', messages: [{ role: 'user', content: [{ type: 'text', text: 'prompt' }] }] }
const agent = stubAgent(cwd)
const result = await ctx.waterfall('agent/request', stubAgent(cwd), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(firstText(result.messages[0])).toContain('## AGENTS.md\n\nroot schema default rule')
expect(firstText(result.messages[0])).toContain('## child/AGENTS.md\n\nchild schema default rule')
expect(derivedText(agent)).toContain('## AGENTS.md\n\nroot schema default rule')
expect(derivedText(agent)).toContain('## child/AGENTS.md\n\nchild schema default rule')
await ctx.fiber.dispose()
} finally {
await rm(root, { recursive: true, force: true })
}
})
it('cleans up its agent/request listener when the plugin fiber is disposed', async () => {
it('cleans up its agent/pre-step listener when the plugin fiber is disposed', async () => {
const root = await tempRepo()
const home = await tempRepo()
try {
@@ -977,14 +971,11 @@ describe('project instruction request injection', () => {
const ctx = new Context()
const fiber = await mountProjectInstructions(ctx, { dshHome: home })
await fiber.dispose()
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(result.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }])
expectNoDerivedMessages(agent)
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -999,14 +990,11 @@ describe('project instruction request injection', () => {
await write(join(root, 'AGENTS.md'), 'repo rule')
const ctx = new Context()
await mountProjectInstructions(ctx, { dshHome: home, baselineMaxBytes: 0 })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(result.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }])
expectNoDerivedMessages(agent)
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -1021,14 +1009,11 @@ describe('project instruction request injection', () => {
await write(join(root, 'AGENTS.md'), 'repo rule')
const ctx = new Context()
await mountProjectInstructions(ctx, { dshHome: home, baselineMaxBytes: -1 })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(result.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }])
expectNoDerivedMessages(agent)
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })
@@ -1042,14 +1027,11 @@ describe('project instruction request injection', () => {
await mkdir(join(root, '.git'), { recursive: true })
const ctx = new Context()
await mountProjectInstructions(ctx, { dshHome: home })
const agent = stubAgent(root)
const request: GenerateOptions = {
model: 'mock',
messages: [{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }],
}
const result = await ctx.waterfall('agent/request', stubAgent(root), 1, 1, request, async () => request)
await runBaselinePreStep(ctx, agent)
expect(result.messages).toEqual([{ role: 'user', content: [{ type: 'text', text: 'actual prompt' }] }])
expectNoDerivedMessages(agent)
} finally {
await rm(root, { recursive: true, force: true })
await rm(home, { recursive: true, force: true })

4
packages/support/invariants/README.md
View File

@@ -40,6 +40,10 @@ Agent status (per agent):
- **legal transitions only** — `idle↔running` and `(idle|running)→disposed`. A no-op transition (`setStatus` dedups, so it never fires) and leaving the terminal `disposed` state are violations.
Model requests (on `llm/stream`):
- **a loop-built request is exactly what the log reconstructs** — a frozen request with a live `sessionId` (the loop-built marker; hand-built one-shots like compaction's summarize are unfrozen and skipped) must carry frozen `messages` deep-equal to the derivation over the log prefix strictly before the in-flight step's `step/start` (rebuilt through a FRESH `Session`, so the live cache cannot vouch for itself — and boundary-correct: content logged after `step/start` legitimately belongs to the next request), and every non-content field must equal the fold of the log's `request/header*` events (see [the reconstructability RFC](../../../docs/rfc/implemented/architecture/2026-07-05-reconstructable-requests.md)). Registered with `prepend: true` so a short-circuiting `llm/stream` listener (the replay adapter) cannot silence it; prepend orders it against append-registered listeners only — correctness rests on the seq-bounded rebuild, never listener timing.
On any violation it throws `InvariantError` (`code: 'INVARIANT'`).
## Why runtime, not deep-readonly types

75
packages/support/invariants/src/index.ts
View File

@@ -21,9 +21,10 @@
import type { Context } from 'cordis'
import { HarnessError } from '@deepseek-ai/dsh-llm'
import type { CallId } from '@deepseek-ai/dsh-llm'
import type { CallId, GenerateOptions } from '@deepseek-ai/dsh-llm'
import type { Agent, AgentStatus } from '@deepseek-ai/dsh-agent'
import type { Session, SessionEvent, SurfaceEventType } from '@deepseek-ai/dsh-session'
import { Session, SessionId, foldRequestHeader } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SurfaceEventType } from '@deepseek-ai/dsh-session'
export const name = 'invariants'
export const inject = ['sessions']
@@ -360,4 +361,74 @@ export function apply(ctx: Context, config: Config = {}): void {
checkTransition(lastStatus.get(agent), status)
lastStatus.set(agent, status)
})
// Request-reconstruction cross-check (the reconstructability RFC): a
// loop-built request — frozen envelope + live sessionId is the marker; a
// hand-built one-shot (compaction summarize) is unfrozen and skipped — must
// be EXACTLY what the session log reconstructs:
//
// - messages: the derivation over the log prefix strictly before the
// in-flight step's `step/start` (the reconstruction boundary). Compared
// against a FRESH Session built over that prefix — the same projection
// code with zero shared state, so the live cache under test cannot vouch
// for itself. Boundary-correct by construction: content appended after
// the boundary (an `agent/request`-window inject) is legitimately absent
// from this request, and a current-surface comparison would false-fire.
// - header: every non-content field must equal the fold of the log's
// `request/header*` events — the loop logs the header event BEFORE
// dispatch, so the fold already covers this request.
//
// Registered with `prepend: true` so a short-circuiting llm/stream listener
// (the replay adapter returns its chunks without calling next()) cannot
// silence the check by registering first. Prepend beats APPEND-registered
// listeners only — two prepended listeners have no defined mutual order
// (cordis unshift) — which is fine: correctness rests on the seq-bounded
// fold below, never on listener timing.
ctx.on('llm/stream', (options: GenerateOptions, next) => {
if (options.sessionId === undefined || !Object.isFrozen(options)) return next()
// GenerateOptions types sessionId as Branded<'SessionId'>, which IS
// SessionId (dsh-llm cannot import it without a cycle) — no cast needed.
const session = ctx.sessions.get(options.sessionId)
if (!session) return next()
if (!Object.isFrozen(options.messages)) {
throw new InvariantError('a loop-built request must carry a frozen messages array')
}
const events = session.events
// seq === index (checked above), so the last step/start's seq bounds the
// prefix directly. The in-flight step's step/start is necessarily the
// last one: the loop cannot open another step while this call streams.
let boundary = -1
for (let i = events.length - 1; i >= 0; i -= 1) {
if (events[i]?.type === 'step/start') {
boundary = i
break
}
}
if (boundary === -1) {
throw new InvariantError('a loop-built request with no step/start in its session log')
}
const rebuilt = new Session(SessionId(`${String(session.id)}-invariant-rebuild`), structuredClone(events.slice(0, boundary)))
// JSON equality is sound here: both sides are structuredClones produced by
// the same projection code path, so key insertion order matches when the
// values do.
if (JSON.stringify(options.messages) !== JSON.stringify(rebuilt.deriveMessages())) {
throw new InvariantError(`llm request for session "${String(session.id)}" diverges from the boundary derivation (log-reconstruction desync)`)
}
const header = foldRequestHeader(events)
if (header === undefined) {
throw new InvariantError('a loop-built request with no request/header event in its session log')
}
const headerMatches = options.model === header.config.model
&& options.system === header.system
&& options.temperature === header.config.temperature
&& options.maxTokens === header.config.maxTokens
&& JSON.stringify(options.stop) === JSON.stringify(header.config.stop)
&& JSON.stringify(options.tools ?? []) === JSON.stringify(header.tools ?? [])
if (!headerMatches) {
throw new InvariantError(`llm request for session "${String(session.id)}" diverges from the folded request header`)
}
return next()
}, { prepend: true })
}

110
packages/support/invariants/tests/invariants.spec.ts
View File

@@ -670,3 +670,113 @@ describe('surface invariants', () => {
.toThrow(/cannot carry surfaceOp/)
})
})
describe('request-reconstruction cross-check (llm/stream)', () => {
/** Session with a boundary: one derivable user message, an open step, and the header event the loop would have logged. */
async function requestSetup() {
const { ctx } = await setup({ freeze: false })
const session = ctx.sessions.create(SessionId('req-check'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
const boundary = session.deriveMessages()
session.append('step/start', { turn: 1, step: 1 })
session.append('request/header', { header: { config: { model: 'm' } }, reason: 'initial' })
return { ctx, session, boundary }
}
/** Dispatch the llm/stream waterfall with a stub core, collecting the check's verdict. */
function dispatch(ctx: Context, options: unknown): void {
// The invariants listener runs synchronously at dispatch time (its checks
// precede next()); the stub core just yields nothing.
void ctx.waterfall('llm/stream', options as never, () => (async function* () {})() as never)
}
it('passes a frozen request that equals the boundary derivation + the folded header', async () => {
const { ctx, session, boundary } = await requestSetup()
const options = Object.freeze({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id })
expect(() => { dispatch(ctx, options) }).not.toThrow()
})
it('is boundary-correct: content logged after step/start is legitimately absent from this request', async () => {
const { ctx, session, boundary } = await requestSetup()
// An agent/request-window inject: lands in the log after the boundary,
// belongs to the NEXT request. A current-surface comparison would
// false-fire here; the seq-bounded rebuild must not.
session.append('context/message', { content: [{ type: 'text', text: '[late]' }], source: { kind: 'plugin', plugin: 'x' } }, { surfaceOp: 'append' })
const options = Object.freeze({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id })
expect(() => { dispatch(ctx, options) }).not.toThrow()
})
it('rejects a frozen request whose messages diverge from the boundary derivation', async () => {
const { ctx, session, boundary } = await requestSetup()
const messages = [...boundary, { role: 'user', content: [{ type: 'text', text: 'phantom' }] }]
const options = Object.freeze({ model: 'm', messages: Object.freeze(messages), sessionId: session.id })
expect(() => { dispatch(ctx, options) }).toThrow(/diverges from the boundary derivation/)
})
it('rejects a frozen request whose fields diverge from the folded header', async () => {
const { ctx, session, boundary } = await requestSetup()
const options = Object.freeze({ model: 'other', messages: Object.freeze(boundary), sessionId: session.id })
expect(() => { dispatch(ctx, options) }).toThrow(/diverges from the folded request header/)
})
it('rejects a loop-built request with no header event or no step/start in its log', async () => {
const { ctx } = await setup({ freeze: false })
const session = ctx.sessions.create(SessionId('req-bare'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
const bare = Object.freeze({ model: 'm', messages: Object.freeze([]), sessionId: session.id })
expect(() => { dispatch(ctx, bare) }).toThrow(/no step\/start/)
session.append('step/start', { turn: 1, step: 1 })
expect(() => { dispatch(ctx, bare) }).toThrow(/no request\/header event/)
})
it('rejects a frozen request carrying an unfrozen messages array', async () => {
const { ctx, session, boundary } = await requestSetup()
const options = Object.freeze({ model: 'm', messages: [...boundary], sessionId: session.id })
expect(() => { dispatch(ctx, options) }).toThrow(/frozen messages array/)
})
it('skips hand-built (unfrozen) requests — compaction summarize is out of scope', async () => {
const { ctx, session } = await requestSetup()
// Unfrozen envelope + arbitrary messages: a direct one-shot call.
const options = { model: 'summarizer', messages: [{ role: 'user', content: [{ type: 'text', text: 'summarize!' }] }], sessionId: session.id }
expect(() => { dispatch(ctx, options) }).not.toThrow()
})
it('skips requests without a sessionId or with an unknown session', async () => {
const { ctx } = await requestSetup()
expect(() => { dispatch(ctx, Object.freeze({ model: 'm', messages: Object.freeze([]) })) }).not.toThrow()
expect(() => { dispatch(ctx, Object.freeze({ model: 'm', messages: Object.freeze([]), sessionId: SessionId('ghost') })) }).not.toThrow()
})
})
describe('request cross-check ordering (prepend)', () => {
it('runs ahead of a short-circuiting llm/stream listener registered before it', async () => {
// The replay adapter returns its chunks WITHOUT calling next(), which
// would silence a later-registered check — snapshot compositions load
// replay before the app bundle that loads invariants. The check prepends,
// so it fires ahead of append-registered listeners regardless of load
// order. (Prepend orders it against APPENDED listeners only; correctness
// rests on the seq-bounded rebuild, not on listener timing.)
const ctx = new Context()
await ctx.plugin(SessionStore)
ctx.on('llm/stream', () => (async function* () {})() as never) // short-circuits, no next()
await ctx.plugin(Invariants, { freeze: false })
const session = ctx.sessions.create(SessionId('prepend-check'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
session.append('step/start', { turn: 1, step: 1 })
session.append('request/header', { header: { config: { model: 'm' } }, reason: 'initial' })
const divergent = Object.freeze({
model: 'm',
messages: Object.freeze([{ role: 'user', content: [{ type: 'text', text: 'phantom' }] }]),
sessionId: session.id,
})
expect(() => {
void ctx.waterfall('llm/stream', divergent as never, () => (async function* () {})() as never)
}).toThrow(/diverges from the boundary derivation/)
})
})