refactor(events): add stable conversation correlation ids

This commit is contained in:
imccyu
2026-08-09 15:48:07 +08:00
parent dc825be8d8
commit aa623b6e7a
23 changed files with 347 additions and 41 deletions

View File

@@ -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<void> {
exec.agent?.session.append('tool/code-dispatch-start', {
rootCallId: exec.rootCallId,
parentCallId: exec.callId,
subCallId,
name,

View File

@@ -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 } : {},

View File

@@ -33,9 +33,34 @@ function validateResult(
const install: InvariantInstaller = Object.assign((ctx: Context, fail: InvariantFailure) => {
const stages = new WeakMap<object, ToolStage>()
const openTurns = new WeakMap<Session, number | null>()
const dispatchRoots = new WeakMap<Session, Map<string, string>>()
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<string, string>
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`)