From aa623b6e7ac5c83fa1f2ea4fe80e124b2d5d66bd Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Sun, 9 Aug 2026 15:48:07 +0800 Subject: [PATCH] refactor(events): add stable conversation correlation ids --- packages/compact/command-compact/src/index.ts | 2 +- packages/compact/compact-basic/package.json | 2 + packages/compact/compact-basic/src/index.ts | 9 +- packages/compact/compact-basic/src/region.ts | 40 ++++++-- packages/compact/compact-basic/tsconfig.json | 3 + packages/compact/compact/package.json | 8 ++ packages/compact/compact/src/brand.ts | 13 +++ packages/compact/compact/src/checkpoint.ts | 25 +++++ packages/compact/compact/src/index.ts | 7 +- packages/compact/compact/src/invariant.ts | 93 ++++++++++++++++++- packages/compact/compact/src/types.ts | 12 ++- packages/compact/compact/tsconfig.json | 6 ++ packages/compact/compact/tsdown.config.ts | 13 +++ packages/core/tools/src/code-mode.ts | 7 +- packages/core/tools/src/index.ts | 9 ++ packages/core/tools/src/invariant.ts | 28 ++++++ packages/interaction/commands/src/index.ts | 4 +- packages/llm/llm-retry/package.json | 6 ++ packages/llm/llm-retry/src/brand.ts | 13 +++ packages/llm/llm-retry/src/index.ts | 36 +++---- packages/llm/llm-retry/src/invariant.ts | 37 +++++++- packages/llm/llm-retry/src/types.ts | 12 +++ packages/llm/llm-retry/tsconfig.json | 3 + 23 files changed, 347 insertions(+), 41 deletions(-) create mode 100644 packages/compact/compact/src/brand.ts create mode 100644 packages/compact/compact/tsdown.config.ts create mode 100644 packages/llm/llm-retry/src/brand.ts diff --git a/packages/compact/command-compact/src/index.ts b/packages/compact/command-compact/src/index.ts index 4ac171a689..b3f80d87fd 100644 --- a/packages/compact/command-compact/src/index.ts +++ b/packages/compact/command-compact/src/index.ts @@ -63,7 +63,7 @@ async function executeCompact( return { kind: 'error', text: USAGE } } try { - const result = await ctx.compact.compactNow(invocation.agent, invocation.signal) + const result = await ctx.compact.compactNow(invocation.agent, invocation.signal, invocation.commandId) if (result === null) return { kind: 'success', text: 'No compactable history yet.' } return { kind: 'success', diff --git a/packages/compact/compact-basic/package.json b/packages/compact/compact-basic/package.json index 2b0624e06e..3fffee14a7 100644 --- a/packages/compact/compact-basic/package.json +++ b/packages/compact/compact-basic/package.json @@ -27,6 +27,7 @@ "peerDependencies": { "@deepseek-ai/dsh-agent": "^0.0.1", "@deepseek-ai/dsh-compact": "^0.0.1", + "@deepseek-ai/dsh-commands": "^0.0.1", "@deepseek-ai/dsh-invariants": "^0.0.1", "@deepseek-ai/dsh-llm": "^0.0.1", "@deepseek-ai/dsh-session": "^0.0.1", @@ -49,6 +50,7 @@ "@deepseek-ai/dsh-agent-loop": "workspace:^", "@deepseek-ai/dsh-agent-loop-testkit": "workspace:^", "@deepseek-ai/dsh-compact": "workspace:^", + "@deepseek-ai/dsh-commands": "workspace:^", "@deepseek-ai/dsh-invariants": "workspace:^", "@deepseek-ai/dsh-llm": "workspace:^", "@deepseek-ai/dsh-llm-retry": "workspace:^", diff --git a/packages/compact/compact-basic/src/index.ts b/packages/compact/compact-basic/src/index.ts index 27b59d0451..d3c710bef3 100644 --- a/packages/compact/compact-basic/src/index.ts +++ b/packages/compact/compact-basic/src/index.ts @@ -13,6 +13,7 @@ import type { Session } from '@deepseek-ai/dsh-session' import { CONTEXT_WINDOW_EXCEEDED_CODE, assertNever } from '@deepseek-ai/dsh-llm' import type { LlmCallConfig } from '@deepseek-ai/dsh-llm' import type { Agent, PreStepDecision } from '@deepseek-ai/dsh-agent' +import type { CommandId } from '@deepseek-ai/dsh-commands/brand' // Type-only: makes the optional sibling service available to `ctx.get()`. import type {} from '@deepseek-ai/dsh-compact-tool-result-prune' import { @@ -361,9 +362,14 @@ export class BasicCompactService extends CompactService { * resolve only after its standalone marker pair is durably checkpointed. * @param agent - idle agent whose next-turn admission this call reserves. * @param signal - cancellation scoped to this compaction request. + * @param sourceCommandId - initiating command identity for presentation correlation. * @returns the committed result, or `null` when no safe useful range exists. */ - override compactNow(agent: Agent, signal: AbortSignal): Promise { + override compactNow( + agent: Agent, + signal: AbortSignal, + sourceCommandId?: CommandId, + ): Promise { signal.throwIfAborted() try { return agent.runMaintenance(async (agentSignal) => { @@ -385,6 +391,7 @@ export class BasicCompactService extends CompactService { { owner: null, stability: 'selected-span', + ...sourceCommandId === undefined ? {} : { sourceCommandId }, flush: async () => { await this.ctx.sessions.flush(agent.session) }, diff --git a/packages/compact/compact-basic/src/region.ts b/packages/compact/compact-basic/src/region.ts index b220074e02..ae637d03cc 100644 --- a/packages/compact/compact-basic/src/region.ts +++ b/packages/compact/compact-basic/src/region.ts @@ -5,14 +5,17 @@ * @module @deepseek-ai/dsh-compact-basic/region */ +import { randomUUID } from 'node:crypto' import { isDeepStrictEqual } from 'node:util' import { - COMPACT_CHECKPOINT_SOURCE, + CompactionId, ManualCompactionError, + compactCheckpointSource, toolPairingBalancedAfter, toolPairingBalancedBefore, } from '@deepseek-ai/dsh-compact' import type { CompactionResult } from '@deepseek-ai/dsh-compact' +import type { CommandId } from '@deepseek-ai/dsh-commands/brand' import { createUserMessage, errorChain } from '@deepseek-ai/dsh-llm' import type { Message, UserMessage } from '@deepseek-ai/dsh-llm' import type { TokenMeasurement, TokenMeterService } from '@deepseek-ai/dsh-token-meter' @@ -54,6 +57,8 @@ interface CompactionTransactionOptions { readonly stability: 'whole-surface' | 'selected-span' /** Optional durability checkpoint after a successfully closed bracket. */ readonly flush?: () => Promise + /** Manual command that initiated this transaction, when present. */ + readonly sourceCommandId?: CommandId } interface CompactionEntryState { @@ -175,7 +180,13 @@ export async function compactSurfaceRegion( owner = entryState.openTurn } - const startEvent = session.append('compact/start', { turn: owner }) + const compactionId = CompactionId(randomUUID()) + const lifecycle = { + compactionId, + ...options.sourceCommandId === undefined ? {} : { sourceCommandId: options.sourceCommandId }, + turn: owner, + } + const startEvent = session.append('compact/start', lifecycle) const assertStable: StabilityCheck = options.stability === 'whole-surface' ? assertWholeSurfaceUnchanged : assertSelectedSpanStable @@ -188,13 +199,20 @@ export async function compactSurfaceRegion( try { const prepared = prepareCompaction(dependencies, session, selection) - const summarized = await summarizeCompaction(dependencies, prepared, agent, signal) + const summarized = await summarizeCompaction( + dependencies, + prepared, + agent, + compactionId, + options.sourceCommandId, + signal, + ) if (options.owner === null) signal?.throwIfAborted() assertStable(dependencies, session, summarized) stage = 'commit' const pending = commitCompactionBody(session, startEvent, summarized) closing = true - const endEvent = session.append('compact/end', { turn: owner }) + const endEvent = session.append('compact/end', lifecycle) closed = true result = completeCompaction(pending, endEvent) } catch (error: unknown) { @@ -202,7 +220,7 @@ export async function compactSurfaceRegion( if (!closing) { closing = true try { - session.append('compact/end', { turn: owner, error: errorChain(error) }) + session.append('compact/end', { ...lifecycle, error: errorChain(error) }) closed = true } catch (closeError: unknown) { failure = { error: closeError, stage: 'commit' } @@ -343,12 +361,14 @@ async function summarizeCompaction( dependencies: RegionDependencies, prepared: PreparedCompaction, agent: Agent, + compactionId: CompactionResult['compactionId'], + sourceCommandId: CommandId | undefined, signal?: AbortSignal, ): Promise { const summaryResult = await dependencies.summarize(prepared.input, agent, signal) const checkpointMessage = createUserMessage({ content: frameSummary(summaryResult.summary), - source: COMPACT_CHECKPOINT_SOURCE, + source: compactCheckpointSource(compactionId, sourceCommandId), }) const framedSummaryTokenCount = dependencies.meter.estimateMessage(checkpointMessage) if (framedSummaryTokenCount >= prepared.shadowedTokenCount) { @@ -425,6 +445,10 @@ function commitCompactionBody( ? { rawOutput: summarized.rawOutput, llmStreamCall: true as const } : summarized.rawOutput === undefined ? {} : { rawOutput: summarized.rawOutput } const summaryEvent = session.append('compact/summary', { + compactionId: startEvent.data.compactionId, + ...startEvent.data.sourceCommandId === undefined + ? {} + : { sourceCommandId: startEvent.data.sourceCommandId }, summary, ...callProvenance, shadowedRange: { start, end }, @@ -440,6 +464,10 @@ function commitCompactionBody( sourceEventSeqs: [startEvent.seq, summaryEvent.seq, ...shadowedSeqs], }) return { + compactionId: startEvent.data.compactionId, + ...startEvent.data.sourceCommandId === undefined + ? {} + : { sourceCommandId: startEvent.data.sourceCommandId }, startSeq: startEvent.seq, summarySeq: summaryEvent.seq, summary, diff --git a/packages/compact/compact-basic/tsconfig.json b/packages/compact/compact-basic/tsconfig.json index bd1a440119..f90011f04d 100644 --- a/packages/compact/compact-basic/tsconfig.json +++ b/packages/compact/compact-basic/tsconfig.json @@ -27,6 +27,9 @@ { "path": "../../core/agent" }, + { + "path": "../../interaction/commands" + }, { "path": "../compact" }, diff --git a/packages/compact/compact/package.json b/packages/compact/compact/package.json index 100a3f56f8..bb4150ee32 100644 --- a/packages/compact/compact/package.json +++ b/packages/compact/compact/package.json @@ -19,6 +19,10 @@ "types": "./lib/types/checkpoint.d.ts", "default": "./lib/types/checkpoint.js" }, + "./brand": { + "types": "./lib/types/brand.d.ts", + "default": "./lib/types/brand.js" + }, "./src/*": "./src/*", "./package.json": "./package.json" }, @@ -30,12 +34,16 @@ ], "license": "BSD-3-Clause", "peerDependencies": { + "@deepseek-ai/dsh-brand": "^0.0.1", + "@deepseek-ai/dsh-commands": "^0.0.1", "@deepseek-ai/dsh-invariants": "^0.0.1", "@deepseek-ai/dsh-llm": "^0.0.1", "@deepseek-ai/dsh-session": "^0.0.1", "cordis": "^4.0.0-rc.7" }, "devDependencies": { + "@deepseek-ai/dsh-brand": "workspace:^", + "@deepseek-ai/dsh-commands": "workspace:^", "@deepseek-ai/dsh-invariants": "workspace:^", "@deepseek-ai/dsh-llm": "workspace:^", "@deepseek-ai/dsh-session": "workspace:^", diff --git a/packages/compact/compact/src/brand.ts b/packages/compact/compact/src/brand.ts new file mode 100644 index 0000000000..5b6de03e61 --- /dev/null +++ b/packages/compact/compact/src/brand.ts @@ -0,0 +1,13 @@ +import type { Branded } from '@deepseek-ai/dsh-brand' + +/** Stable identity shared by one compact start/summary/checkpoint/end transaction. */ +export type CompactionId = Branded<'CompactionId'> + +/** + * Brand an implementation-minted compaction identity. + * @param id - opaque transaction identity. + * @returns the same string, branded; no validation is performed. + */ +export function CompactionId(id: string): CompactionId { + return id as CompactionId +} diff --git a/packages/compact/compact/src/checkpoint.ts b/packages/compact/compact/src/checkpoint.ts index 9d8b98e4d6..9908fe4ac0 100644 --- a/packages/compact/compact/src/checkpoint.ts +++ b/packages/compact/compact/src/checkpoint.ts @@ -13,10 +13,35 @@ */ import type { MessageSource } from '@deepseek-ai/dsh-llm/message' +import type { CommandId } from '@deepseek-ai/dsh-commands/brand' +import type { CompactionId } from './brand.ts' /** Canonical source for the replacement user message produced by every compaction backend. */ export const COMPACT_CHECKPOINT_SOURCE = Object.freeze({ kind: 'plugin', plugin: 'compact' } as const) +/** Message provenance carried by a concrete compaction checkpoint. */ +export type CompactCheckpointSource = typeof COMPACT_CHECKPOINT_SOURCE & { + readonly compactionId: CompactionId + readonly sourceCommandId?: CommandId +} + +/** + * Create checkpoint provenance correlated with one compaction transaction. + * @param compactionId - owning compaction identity. + * @param sourceCommandId - initiating manual command, when present. + * @returns immutable checkpoint source. + */ +export function compactCheckpointSource( + compactionId: CompactionId, + sourceCommandId?: CommandId, +): CompactCheckpointSource { + return Object.freeze({ + ...COMPACT_CHECKPOINT_SOURCE, + compactionId, + ...sourceCommandId === undefined ? {} : { sourceCommandId }, + }) +} + /** * Test whether a persisted message source identifies a compaction checkpoint. * @param source - source restored from a surface user message. diff --git a/packages/compact/compact/src/index.ts b/packages/compact/compact/src/index.ts index 7b0bd25091..0085321333 100644 --- a/packages/compact/compact/src/index.ts +++ b/packages/compact/compact/src/index.ts @@ -9,14 +9,17 @@ import { Context, Service } from 'cordis' import type { Session } from '@deepseek-ai/dsh-session' +import type { CommandId } from '@deepseek-ai/dsh-commands/brand' import type { CompactionResult } from './types.ts' export type { CompactionResult } from './types.ts' +export { CompactionId } from './brand.ts' export { toolPairingBalancedAfter, toolPairingBalancedBefore } from './tool-pairing.ts' // The checkpoint source and its predicate are declared on the cordis-free // `./checkpoint` leaf so client and wire programs can name them without this // root's Context merge; the root stays the host-side entry point for both. -export { COMPACT_CHECKPOINT_SOURCE, isCompactCheckpointSource } from './checkpoint.ts' +export { COMPACT_CHECKPOINT_SOURCE, compactCheckpointSource, isCompactCheckpointSource } from './checkpoint.ts' +export type { CompactCheckpointSource } from './checkpoint.ts' /** Why automatic policy is asking a backend to consider compaction. */ export type CompactionTrigger = 'pressure' | 'context-overflow' @@ -126,6 +129,7 @@ export abstract class CompactService extends Service { * * @param agent - idle agent whose durable history should be compacted. * @param signal - cancellation scoped to this compaction request. + * @param sourceCommandId - initiating command identity for a manual compaction. * @returns the compaction result, or `null` when no safe useful range exists. * @throws {@link ManualCompactionError} for expected busy, agent-cancellation, * changed-span, summarization/shrink, commit-stage, or persistence failures; @@ -135,6 +139,7 @@ export abstract class CompactService extends Service { abstract compactNow( agent: ManualCompactAgentContext, signal: AbortSignal, + sourceCommandId?: CommandId, ): Promise /** diff --git a/packages/compact/compact/src/invariant.ts b/packages/compact/compact/src/invariant.ts index 221d5aae09..adac557db5 100644 --- a/packages/compact/compact/src/invariant.ts +++ b/packages/compact/compact/src/invariant.ts @@ -1,8 +1,12 @@ /** Package-owned compaction log-stream invariants. @module @deepseek-ai/dsh-compact/invariant */ import type { Context } from 'cordis' +import { isReplacementSurfaceEvent } from '@deepseek-ai/dsh-session' import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants' +import type { CompactionId } from './brand.ts' +import { isCompactCheckpointSource } from './checkpoint.ts' +import type { CompactCheckpointSource } from './checkpoint.ts' import type {} from './types.ts' const PACKAGE_NAME = '@deepseek-ai/dsh-compact' @@ -13,6 +17,8 @@ export const name = 'compact-invariant' export const inject = ['invariants'] interface CompactionTrace { + compactionId: CompactionId + sourceCommandId: string | undefined startSeq: number turn: number | null summarized: boolean @@ -24,11 +30,48 @@ interface SessionTrace { } type CompactionTransition = - | { kind: 'start'; startSeq: number; turn: number | null } - | { kind: 'summary'; startSeq: number; turn: number | null } + | { kind: 'start'; compactionId: CompactionId; sourceCommandId: string | undefined; startSeq: number; turn: number | null } + | { kind: 'summary'; compactionId: CompactionId; sourceCommandId: string | undefined; startSeq: number; turn: number | null } | { kind: 'end' } | { kind: 'end-seed' } +/** Require a durable opaque identity to be a non-empty string. */ +function validateId(value: unknown, label: string, fail: InvariantFailure): asserts value is string { + if (typeof value !== 'string' || value.length === 0) fail(`${label} must be a non-empty string`) +} + +/** Keep the optional initiating command identity stable across one transaction. */ +function validateSourceCommandId( + eventType: string, + value: unknown, + expected: string | undefined, + fail: InvariantFailure, +): void { + if (value !== undefined) validateId(value, `${eventType} sourceCommandId`, fail) + if (value !== expected) { + fail(`${eventType} sourceCommandId ${String(value)} does not match compact/start sourceCommandId ${String(expected)}`) + } +} + +/** Validate one replacement checkpoint against its open compaction transaction. */ +function validateCheckpoint( + trace: SessionTrace, + event: SessionEvent<'user/message'>, + fail: InvariantFailure, +): void { + const source = event.data.source as typeof event.data.source & Partial + validateId(source.compactionId, 'compaction checkpoint compactionId', fail) + if (source.sourceCommandId !== undefined) { + validateId(source.sourceCommandId, 'compaction checkpoint sourceCommandId', fail) + } + const open = trace.compaction + if (open === undefined) fail('compaction checkpoint has no matching compact/start') + if (source.compactionId !== open.compactionId) { + fail(`compaction checkpoint id ${source.compactionId} does not match compact/start id ${open.compactionId}`) + } + validateSourceCommandId('compaction checkpoint', source.sourceCommandId, open.sourceCommandId, fail) +} + /** Compaction starts still unmatched when a later seed boundary made them stale. */ function inheritedOrphanStartSeqs( events: readonly SessionEvent[], @@ -99,20 +142,44 @@ function validateCompactionEvent( fail: InvariantFailure, ): CompactionTransition | undefined { if (event.type === 'session/end-seed') return { kind: 'end-seed' } + if (event.type === 'user/message' + && isReplacementSurfaceEvent(event) + && isCompactCheckpointSource(event.data.source)) { + validateCheckpoint(trace, event, fail) + return undefined + } if (event.type !== 'compact/start' && event.type !== 'compact/summary' && event.type !== 'compact/end') { return undefined } const open = trace.compaction if (event.type === 'compact/start') { + validateId(event.data.compactionId, 'compact/start compactionId', fail) + if (event.data.sourceCommandId !== undefined) { + validateId(event.data.sourceCommandId, 'compact/start sourceCommandId', fail) + } if (open !== undefined) { const owner = open.turn === null ? 'standalone compaction' : `turn ${open.turn}` fail(`compact/start while ${owner} is still compacting`) } validateOwner(event.data.turn, trace.openTurn, event.type, fail) - return { kind: 'start', startSeq: event.seq, turn: event.data.turn } + return { + kind: 'start', + compactionId: event.data.compactionId, + sourceCommandId: event.data.sourceCommandId, + startSeq: event.seq, + turn: event.data.turn, + } } if (event.type === 'compact/summary') { + validateId(event.data.compactionId, 'compact/summary compactionId', fail) + if (event.data.sourceCommandId !== undefined) { + validateId(event.data.sourceCommandId, 'compact/summary sourceCommandId', fail) + } if (open === undefined) fail('compact/summary has no matching compact/start') + if (event.data.compactionId !== open.compactionId) { + fail(`compact/summary id ${event.data.compactionId} does not match compact/start id ${open.compactionId}`) + } + validateSourceCommandId('compact/summary', event.data.sourceCommandId, open.sourceCommandId, fail) validateOwner(open.turn, trace.openTurn, event.type, fail) if (open.summarized) fail('compact/summary repeated within one compaction') const seqs = event.data.shadowedSeqs @@ -123,9 +190,23 @@ function validateCompactionEvent( if (!Number.isSafeInteger(event.data.shadowedTokenCount) || event.data.shadowedTokenCount < 0) { fail('compact/summary shadowedTokenCount must be a non-negative safe integer') } - return { kind: 'summary', startSeq: open.startSeq, turn: open.turn } + return { + kind: 'summary', + compactionId: open.compactionId, + sourceCommandId: open.sourceCommandId, + startSeq: open.startSeq, + turn: open.turn, + } + } + validateId(event.data.compactionId, 'compact/end compactionId', fail) + if (event.data.sourceCommandId !== undefined) { + validateId(event.data.sourceCommandId, 'compact/end sourceCommandId', fail) } if (open === undefined) fail('compact/end has no matching compact/start') + if (event.data.compactionId !== open.compactionId) { + fail(`compact/end id ${event.data.compactionId} does not match compact/start id ${open.compactionId}`) + } + validateSourceCommandId('compact/end', event.data.sourceCommandId, open.sourceCommandId, fail) if (event.data.turn !== open.turn) { fail(`compact/end owner ${String(event.data.turn)} does not match compact/start owner ${String(open.turn)}`) } @@ -142,6 +223,8 @@ function applyCompactionTransition( ): CompactionTrace | undefined { if (transition.kind === 'start') { return { + compactionId: transition.compactionId, + sourceCommandId: transition.sourceCommandId, startSeq: transition.startSeq, turn: transition.turn, summarized: false, @@ -149,6 +232,8 @@ function applyCompactionTransition( } if (transition.kind === 'summary') { return { + compactionId: transition.compactionId, + sourceCommandId: transition.sourceCommandId, startSeq: transition.startSeq, turn: transition.turn, summarized: true, diff --git a/packages/compact/compact/src/types.ts b/packages/compact/compact/src/types.ts index 4429e8b291..c321be426e 100644 --- a/packages/compact/compact/src/types.ts +++ b/packages/compact/compact/src/types.ts @@ -8,6 +8,8 @@ */ import type { ContentBlock, TokenUsage } from '@deepseek-ai/dsh-llm' +import type { CommandId } from '@deepseek-ai/dsh-commands/brand' +import type { CompactionId } from './brand.ts' declare module '@deepseek-ai/dsh-session' { interface SessionEventMap { @@ -16,7 +18,7 @@ declare module '@deepseek-ai/dsh-session' { * `compact/end`. A numbered owner is strictly enclosed by that open turn; * `null` identifies a standalone manual transaction between turns. */ - 'compact/start': { turn: number | null } + 'compact/start': { compactionId: CompactionId; sourceCommandId?: CommandId; turn: number | null } /** * Completed summary, its inputs, and its model call facts — log-only, no surfaceOp. * The summary content is in `data.summary`; the actual surface replacement @@ -27,6 +29,8 @@ declare module '@deepseek-ai/dsh-session' { * before it (`compact/prune` documents the shared protocol). */ 'compact/summary': { + compactionId: CompactionId + sourceCommandId?: CommandId summary: ContentBlock[] shadowedRange: { start: number; end: number } shadowedSeqs: number[] @@ -62,7 +66,7 @@ declare module '@deepseek-ai/dsh-session' { * Marks the end of a compaction — log-only, releases the lock. Its owner * matches `compact/start`; `error` records an unsuccessful attempt. */ - 'compact/end': { turn: number | null; error?: string } + 'compact/end': { compactionId: CompactionId; sourceCommandId?: CommandId; turn: number | null; error?: string } /** * Shadow price of one model-free prune replacement — log-only, no * surfaceOp. The shared shadow-price protocol: a surface `replace` event @@ -85,6 +89,10 @@ declare module '@deepseek-ai/dsh-session' { /** Result of a successful compaction operation. */ export interface CompactionResult { + /** Stable identity shared by this compaction's complete durable lifecycle. */ + compactionId: CompactionId + /** Human command that initiated this compaction, when it was manual. */ + sourceCommandId?: CommandId /** The seq of the appended `compact/start` event. */ startSeq: number /** The seq of the appended `compact/summary` event. */ diff --git a/packages/compact/compact/tsconfig.json b/packages/compact/compact/tsconfig.json index 673ee51547..a231786c32 100644 --- a/packages/compact/compact/tsconfig.json +++ b/packages/compact/compact/tsconfig.json @@ -8,6 +8,9 @@ "src" ], "references": [ + { + "path": "../../util/brand" + }, { "path": "../../../vendor/cosmokit" }, @@ -17,6 +20,9 @@ { "path": "../../llm/llm" }, + { + "path": "../../interaction/commands" + }, { "path": "../../core/session" }, diff --git a/packages/compact/compact/tsdown.config.ts b/packages/compact/compact/tsdown.config.ts new file mode 100644 index 0000000000..6284421125 --- /dev/null +++ b/packages/compact/compact/tsdown.config.ts @@ -0,0 +1,13 @@ +import { defineConfig } from 'tsdown' + +/** Builds each published entry as a self-contained file admitted by the package whitelist. */ +export default defineConfig([ + { + entry: ['lib/types/index.js'], outDir: 'lib', format: ['esm'], platform: 'node', target: 'es2024', + fixedExtension: false, outputOptions: { codeSplitting: false }, dts: false, clean: false, + }, + { + entry: ['lib/types/invariant.js'], outDir: 'lib', format: ['esm'], platform: 'node', target: 'es2024', + fixedExtension: false, outputOptions: { codeSplitting: false }, dts: false, clean: false, + }, +]) diff --git a/packages/core/tools/src/code-mode.ts b/packages/core/tools/src/code-mode.ts index 2ec5a1bb25..63c5b2cb9b 100644 --- a/packages/core/tools/src/code-mode.ts +++ b/packages/core/tools/src/code-mode.ts @@ -30,7 +30,7 @@ declare module '@deepseek-ai/dsh-session' { * with `tool/code-dispatch` by `subCallId` (timing = the two events' * `time` fields). */ - 'tool/code-dispatch-start': { parentCallId: CallId; subCallId: CallId; name: string; arguments: unknown } + 'tool/code-dispatch-start': { rootCallId: CallId; parentCallId: CallId; subCallId: CallId; name: string; arguments: unknown } /** * One bridged sub-dispatch SETTLING: the pairing ids (matching the * `tool/code-dispatch-start` with the same `subCallId`), the tool `name` @@ -46,7 +46,7 @@ declare module '@deepseek-ai/dsh-session' { * before returning), so its execution-enclosure relation holds by * construction. */ - 'tool/code-dispatch': { parentCallId: CallId; subCallId: CallId; name: string; arguments: unknown; isError: boolean; content: ContentBlock[] } + 'tool/code-dispatch': { rootCallId: CallId; parentCallId: CallId; subCallId: CallId; name: string; arguments: unknown; isError: boolean; content: ContentBlock[] } } } @@ -502,6 +502,7 @@ export function createRunCodeTool(registry: ToolRegistry, options: RunCodeBridge const subCallId = CallId(`${String(exec.callId)}:code:${n}`) const input = { callId: subCallId, + rootCallId: exec.rootCallId, name, arguments: normalized.dispatched, ...exec.agent ? { agent: exec.agent } : {}, @@ -539,6 +540,7 @@ export function createRunCodeTool(registry: ToolRegistry, options: RunCodeBridge content: result.content, }) agent.session.append('tool/code-dispatch', { + rootCallId: exec.rootCallId, parentCallId: exec.callId, subCallId, name, @@ -563,6 +565,7 @@ export function createRunCodeTool(registry: ToolRegistry, options: RunCodeBridge }, async start(): Promise { exec.agent?.session.append('tool/code-dispatch-start', { + rootCallId: exec.rootCallId, parentCallId: exec.callId, subCallId, name, diff --git a/packages/core/tools/src/index.ts b/packages/core/tools/src/index.ts index 7c2346e723..6338e54501 100644 --- a/packages/core/tools/src/index.ts +++ b/packages/core/tools/src/index.ts @@ -297,6 +297,11 @@ export type ToolExecutionToken = symbol & { readonly [toolExecutionTokenBrand]: */ export interface ToolExecutionInput { readonly callId: CallId + /** + * Root model-requested call owning this execution tree. Callers omit it for + * a root execution; nested dispatchers propagate the enclosing value. + */ + readonly rootCallId?: CallId readonly name: string /** Losslessly JSON-serializable parsed arguments (tools validate their own schema). */ readonly arguments: unknown @@ -352,6 +357,8 @@ export interface CodeDispatchLog { * observers run. */ export interface ToolExecution extends ToolExecutionInput { + /** Root model-requested call, resolved for every root and nested execution. */ + readonly rootCallId: CallId /** Registry-assigned identity shared with nested calls only as their opaque `parent` token. */ readonly token: ToolExecutionToken } @@ -1123,6 +1130,7 @@ export class ToolRegistry extends Service { const deferredContexts: UserMessage[] = [] const token = createExecutionToken() const callId = exec.callId + const rootCallId = exec.rootCallId ?? callId const name = exec.name const agent = exec.agent const parent = exec.parent @@ -1133,6 +1141,7 @@ export class ToolRegistry extends Service { const base = { token, callId, + rootCallId, name, signal, ...agent !== undefined ? { agent } : {}, diff --git a/packages/core/tools/src/invariant.ts b/packages/core/tools/src/invariant.ts index 78666f5890..a0d9487857 100644 --- a/packages/core/tools/src/invariant.ts +++ b/packages/core/tools/src/invariant.ts @@ -33,9 +33,34 @@ function validateResult( const install: InvariantInstaller = Object.assign((ctx: Context, fail: InvariantFailure) => { const stages = new WeakMap() const openTurns = new WeakMap() + const dispatchRoots = new WeakMap>() + const validateDispatch = (session: Session, event: SessionEvent): void => { + if (event.type !== 'tool/code-dispatch-start' && event.type !== 'tool/code-dispatch') return + const root = String(event.data.rootCallId) + const parent = String(event.data.parentCallId) + const child = String(event.data.subCallId) + if (root.length === 0 || parent.length === 0 || child.length === 0) { + fail(`${event.type} must carry non-empty rootCallId, parentCallId, and subCallId`) + return + } + const roots = dispatchRoots.get(session) + const known = roots?.get(child) + if (known !== undefined && known !== root) fail(`${event.type} changed rootCallId for subCallId ${child}`) + if (parent !== root && roots?.get(parent) !== root) { + fail(`${event.type} parentCallId ${parent} does not belong to rootCallId ${root}`) + } + } + const commitDispatch = (session: Session, event: SessionEvent): void => { + if (event.type !== 'tool/code-dispatch-start' && event.type !== 'tool/code-dispatch') return + const roots = dispatchRoots.get(session) as Map + roots.set(String(event.data.subCallId), String(event.data.rootCallId)) + } const seed = (session: Session): number | null => { let openTurn: number | null = null + dispatchRoots.set(session, new Map()) for (const event of session.events) { + validateDispatch(session, event) + commitDispatch(session, event) if (event.type === 'turn/start') openTurn = event.data.turn else if (event.type === 'turn/end') openTurn = null else if ((event.type === 'tool/code-dispatch-start' || event.type === 'tool/code-dispatch') @@ -51,12 +76,15 @@ const install: InvariantInstaller = Object.assign((ctx: Context, fail: Invariant for (const session of ctx.sessions.list()) seed(session) ctx.on('session/created', (session) => { seed(session) }, { global: true }) ctx.on('session/event', (session, event) => { + validateDispatch(session, event) + commitDispatch(session, event) if (event.type === 'turn/start') openTurns.set(session, event.data.turn) else if (event.type === 'turn/end') openTurns.set(session, null) }, { global: true }) ctx.on('internal/dispatch', (_mode, eventName, args) => { if (eventName === 'session/event') { const [session, event] = args as [Session, SessionEvent] + validateDispatch(session, event) if ((event.type === 'tool/code-dispatch-start' || event.type === 'tool/code-dispatch') && openTurnFor(session) === null) { fail(`${event.type} appended outside any open turn`) diff --git a/packages/interaction/commands/src/index.ts b/packages/interaction/commands/src/index.ts index 94c4861e66..780569c159 100644 --- a/packages/interaction/commands/src/index.ts +++ b/packages/interaction/commands/src/index.ts @@ -37,6 +37,8 @@ export interface CommandInputDescriptor { /** Invocation passed to one registered command handler. */ export interface CommandInvocation { + /** Pairing id already written to this invocation's `command/run` event. */ + readonly commandId: CommandId /** Exact agent whose human-facing surface received the command. */ readonly agent: Agent /** Exact text following the registered command name, including separator whitespace. */ @@ -389,7 +391,7 @@ export class CommandService extends Service { ...command.definition.recordInput === false ? {} : { args: parsed.rawInput }, source: { kind: 'user' }, }) - const invocation = Object.freeze({ agent, rawInput: parsed.rawInput, signal }) + const invocation = Object.freeze({ commandId, agent, rawInput: parsed.rawInput, signal }) let result: CommandResult try { const output = command.definition.handler(invocation) diff --git a/packages/llm/llm-retry/package.json b/packages/llm/llm-retry/package.json index d3b6afebba..1685d0e552 100644 --- a/packages/llm/llm-retry/package.json +++ b/packages/llm/llm-retry/package.json @@ -19,6 +19,10 @@ "types": "./lib/types/types.d.ts", "default": "./lib/types/types.js" }, + "./brand": { + "types": "./lib/types/brand.d.ts", + "default": "./lib/types/brand.js" + }, "./package.json": "./package.json" }, "files": [ @@ -29,6 +33,7 @@ ], "license": "BSD-3-Clause", "peerDependencies": { + "@deepseek-ai/dsh-brand": "^0.0.1", "@deepseek-ai/dsh-agent": "^0.0.1", "@deepseek-ai/dsh-invariants": "^0.0.1", "@deepseek-ai/dsh-llm": "^0.0.1", @@ -40,6 +45,7 @@ "schemastery": "^3.18.0" }, "devDependencies": { + "@deepseek-ai/dsh-brand": "workspace:^", "@cordisjs/plugin-include": "workspace:^", "@cordisjs/plugin-loader": "workspace:^", "@deepseek-ai/dsh-agent": "workspace:^", diff --git a/packages/llm/llm-retry/src/brand.ts b/packages/llm/llm-retry/src/brand.ts new file mode 100644 index 0000000000..41682cc9d1 --- /dev/null +++ b/packages/llm/llm-retry/src/brand.ts @@ -0,0 +1,13 @@ +import type { Branded } from '@deepseek-ai/dsh-brand' + +/** Stable identity shared by every attempt in one request-step retry chain. */ +export type RetryId = Branded<'RetryId'> + +/** + * Brand an implementation-minted retry-chain identity. + * @param id - opaque retry identity. + * @returns the same string, branded; no validation is performed. + */ +export function RetryId(id: string): RetryId { + return id as RetryId +} diff --git a/packages/llm/llm-retry/src/index.ts b/packages/llm/llm-retry/src/index.ts index 82b441fb8f..dcba1e2443 100644 --- a/packages/llm/llm-retry/src/index.ts +++ b/packages/llm/llm-retry/src/index.ts @@ -5,39 +5,26 @@ * @module @deepseek-ai/dsh-llm-retry */ +import { randomUUID } from 'node:crypto' import type { Context, Events } from 'cordis' import z from 'schemastery' import type { Agent, RequestErrorAction } from '@deepseek-ai/dsh-agent' import type { LlmFailure, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm' import type { SessionEvent } from '@deepseek-ai/dsh-session' +import { RetryId } from './brand.ts' +import type { LlmRetryEventData, LlmRetryStartedEventData } from './types.ts' declare module '@deepseek-ai/dsh-session' { interface SessionEventMap { /** Durable, non-surface record of one provider-routed retry scheduled after a failed request attempt. */ - 'llm/retry': { - turn: number - step: number - provider: string - mode: 'normal' - policyKey: string - retry: number - maxRetries: number - delayMs: number - failure: LlmFailure - } | { - turn: number - step: number - provider: string - mode: 'always' - policyKey: string - retry: number - delayMs: number - failure: LlmFailure - } + 'llm/retry': LlmRetryEventData + /** Durable transition written after a retry wait succeeds and before the next request attempt starts. */ + 'llm/retry-started': LlmRetryStartedEventData } } -export type { LlmRetryEventData } from './types.ts' +export type { LlmRetryEventData, LlmRetryStartedEventData } from './types.ts' +export { RetryId } from './brand.ts' export const name = 'llm-retry' export const inject = ['agents'] @@ -139,6 +126,7 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna policy: ResolvedRetryPolicy, policyKey: string, retry: number, + retryId: RetryId, delayMs: number, signal: AbortSignal, ): Promise { @@ -146,6 +134,7 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna if (fusedSignal.aborted) return const eventData = policy.mode === 'normal' ? { + retryId, turn, step, provider, @@ -157,6 +146,7 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna failure, } : { + retryId, turn, step, provider, @@ -168,6 +158,7 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna } agent.session.append('llm/retry', eventData) if (!await cancellableDelay(delayMs, fusedSignal)) return + agent.session.append('llm/retry-started', { retryId, turn, step, retry }) return { kind: 'retry' } } @@ -207,6 +198,7 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna const previousRetry = priorPolicyRetry?.data.retry ?? 0 if (policy.mode === 'normal' && previousRetry >= policy.maxRetries) return next() const retry = previousRetry + 1 + const retryId = priorPolicyRetry?.data.retryId ?? RetryId(randomUUID()) let delayMs: number if (failure.providerRetryAfterMs !== undefined && Number.isFinite(failure.providerRetryAfterMs) @@ -221,7 +213,7 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna delayMs = localDelay(policy, retry, random) } - return backoff(agent, turn, step, failure, provider, policy, policyKey, retry, delayMs, signal) + return backoff(agent, turn, step, failure, provider, policy, policyKey, retry, retryId, delayMs, signal) } const disposeListener = ctx.on('agent/request-error', ( diff --git a/packages/llm/llm-retry/src/invariant.ts b/packages/llm/llm-retry/src/invariant.ts index d324c012f2..1680873c0c 100644 --- a/packages/llm/llm-retry/src/invariant.ts +++ b/packages/llm/llm-retry/src/invariant.ts @@ -47,7 +47,10 @@ function validateRetry( event: SessionEvent<'llm/retry'>, fail: InvariantFailure, ): void { - const { turn, step, provider, mode, policyKey, retry, delayMs } = event.data + const { retryId, turn, step, provider, mode, policyKey, retry, delayMs } = event.data + if (typeof retryId !== 'string' || retryId.length === 0) { + fail('llm/retry retryId must be a non-empty string') + } const failure: unknown = event.data.failure validateFailure(failure, fail) if (!Number.isSafeInteger(retry) || retry < 1) { @@ -110,12 +113,43 @@ function validateRetry( if (retry !== expectedRetry) { fail(`llm/retry retry ${retry} must equal provider policy retry ${expectedRetry}`) } + if (priorPolicyRetry !== undefined && priorPolicyRetry.data.retryId !== retryId) { + fail('llm/retry must preserve retryId across one provider-policy chain') + } + if (priorPolicyRetry === undefined && history.some(prior => + (prior.type === 'llm/retry' || prior.type === 'llm/retry-started') + && prior.data.retryId === retryId)) { + fail(`llm/retry retryId ${JSON.stringify(retryId)} is already owned by another chain`) + } +} + +/** Validate one wait-complete transition against its scheduled attempt. */ +function validateStarted( + history: readonly SessionEvent[], + event: SessionEvent<'llm/retry-started'>, + fail: InvariantFailure, +): void { + const { retryId, turn, step, retry } = event.data + if (typeof retryId !== 'string' || retryId.length === 0) { + fail('llm/retry-started retryId must be a non-empty string') + } + const scheduled = history.findLast((prior): prior is SessionEvent<'llm/retry'> => + prior.type === 'llm/retry' && prior.data.retryId === retryId && prior.data.retry === retry) + if (scheduled === undefined) fail('llm/retry-started pairs no prior scheduled attempt') + if (scheduled.data.turn !== turn || scheduled.data.step !== step) { + fail('llm/retry-started turn/step must match its scheduled attempt') + } + if (history.some(prior => prior.type === 'llm/retry-started' + && prior.data.retryId === retryId && prior.data.retry === retry)) { + fail('llm/retry-started repeats one scheduled attempt') + } } /** Validate every retry record already present in one loaded session. */ function validateSession(session: Session, fail: InvariantFailure): void { for (const [index, event] of session.events.entries()) { if (event.type === 'llm/retry') validateRetry(session.events.slice(0, index), event, fail) + else if (event.type === 'llm/retry-started') validateStarted(session.events.slice(0, index), event, fail) } } @@ -127,6 +161,7 @@ const install: InvariantInstaller = Object.assign((ctx: Context, fail: Invariant if (eventName !== 'session/event') return const [session, event] = args as [Session, SessionEvent] if (event.type === 'llm/retry') validateRetry(session.events, event, fail) + else if (event.type === 'llm/retry-started') validateStarted(session.events, event, fail) }, { global: true }) }, { inject: ['sessions'] }) diff --git a/packages/llm/llm-retry/src/types.ts b/packages/llm/llm-retry/src/types.ts index f59aef4495..12820889dd 100644 --- a/packages/llm/llm-retry/src/types.ts +++ b/packages/llm/llm-retry/src/types.ts @@ -1,8 +1,10 @@ import type { LlmFailure } from '@deepseek-ai/dsh-llm/types' +import type { RetryId } from './brand.ts' /** Durable payload recorded before one provider-routed model-request retry wait. */ export type LlmRetryEventData = | { + retryId: RetryId turn: number step: number provider: string @@ -13,7 +15,9 @@ export type LlmRetryEventData = delayMs: number failure: LlmFailure } + | { + retryId: RetryId turn: number step: number provider: string @@ -23,3 +27,11 @@ export type LlmRetryEventData = delayMs: number failure: LlmFailure } + +/** Durable transition recorded after one retry delay completes. */ +export interface LlmRetryStartedEventData { + retryId: RetryId + turn: number + step: number + retry: number +} diff --git a/packages/llm/llm-retry/tsconfig.json b/packages/llm/llm-retry/tsconfig.json index 48c858951a..41e9ca65c5 100644 --- a/packages/llm/llm-retry/tsconfig.json +++ b/packages/llm/llm-retry/tsconfig.json @@ -8,6 +8,9 @@ "src" ], "references": [ + { + "path": "../../util/brand" + }, { "path": "../../../vendor/cosmokit" },