fix(workflow): close review and snapshot gaps

2026-08-10 19:25:21 +08:00
parent 8de6df19d9
commit 4eb0a52840
20 changed files with 300 additions and 350 deletions
--- a/packages/workflow/tool-workflow/src/index.ts
+++ b/packages/workflow/tool-workflow/src/index.ts
@@ -17,8 +17,7 @@ import type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools'
 import type { ContentBlock } from '@deepseek-ai/dsh-llm'
 import type { JsonValue, Session, SessionEventMap } from '@deepseek-ai/dsh-session'
 import type {
-  WorkflowAgentEndInfo, WorkflowAgentInfo, WorkflowResult, WorkflowRun,
-  WorkflowRunId, WorkflowRunInfo, WorkflowStopReason,
+  WorkflowResult, WorkflowRun, WorkflowRunId, WorkflowStopReason,
 } from '@deepseek-ai/dsh-workflow'
 import type {
  ToolWorkflowAgentEndData, ToolWorkflowAgentStartData,
@@ -45,14 +44,10 @@ export const Config: z<Config> = z.object({

 type ResolvedConfig = Required<Config>

-type BufferedWorkflowEvent =
-  | { readonly kind: 'agent-start'; readonly info: WorkflowRunInfo; readonly agent: WorkflowAgentInfo }
-  | { readonly kind: 'agent-end'; readonly info: WorkflowRunInfo; readonly agent: WorkflowAgentEndInfo }
-
 interface WorkflowRecorder {
-  bind(run: WorkflowRun): void
-  finish(stopReason: WorkflowStopReason): void
-  dispose(): void
+  start(session: Session, run: WorkflowRun): void
+  finish(runId: WorkflowRunId, stopReason: WorkflowStopReason): void
+  abandon(runId: WorkflowRunId): void
 }

 interface ToolWorkflowRecordEventMap {
@@ -72,84 +67,66 @@ function renderRecordingError(error: unknown): string {
 }

 /**
- * Project one top-level workflow run into its parent Session without letting
- * recording failure affect tool execution. Listeners are installed before
- * `start()` so even a synchronous provider cannot outrun the recorder.
+ * Project active top-level workflow runs into their parent Sessions without
+ * letting recording failure affect tool execution.
 */
-function createWorkflowRecorder(ctx: Context, session: Session): WorkflowRecorder {
-  let runId: WorkflowRunId | undefined
-  let enabled = true
-  const buffered: BufferedWorkflowEvent[] = []
-  // These four package-owned events are all log-only. Narrowing the generic
-  // append face here lets TypeScript discharge Session.append's conditional
-  // surface-options tuple once for the complete closed event set.
-  const appendRecord = session.append.bind(session) as <Type extends keyof ToolWorkflowRecordEventMap>(
-    type: Type,
-    data: SessionEventMap[Type],
-  ) => void
-
+function createWorkflowRecorder(ctx: Context): WorkflowRecorder {
+  const active = new Map<WorkflowRunId, Session>()
  const append = <Type extends keyof ToolWorkflowRecordEventMap>(
+    session: Session,
    type: Type,
    data: SessionEventMap[Type],
-  ): void => {
-    if (!enabled) return
+  ): boolean => {
+    // These four package-owned events are all log-only. Narrowing the generic
+    // append face here discharges Session.append's conditional options tuple.
+    const appendRecord = session.append.bind(session) as <Event extends keyof ToolWorkflowRecordEventMap>(
+      event: Event,
+      value: SessionEventMap[Event],
+    ) => void
    try {
      appendRecord(type, data)
+      return true
    } catch (error: unknown) {
-      enabled = false
      ctx.logger.warn(`tool-workflow: disabled durable record after ${type} append failed: ${renderRecordingError(error)}`)
+      return false
    }
  }

-  const record = (event: BufferedWorkflowEvent): void => {
-    if (runId === undefined) {
-      buffered.push(event)
-      return
+  ctx.on('workflow/agent-start', (info, agent) => {
+    const session = active.get(info.id)
+    if (session === undefined) return
+    const data: ToolWorkflowAgentStartData = {
+      runId: info.id,
+      seq: agent.seq,
+      label: agent.label,
+      ...agent.phase === undefined ? {} : { phase: agent.phase },
+      childId: agent.childId,
    }
-    if (event.info.id !== runId) return
-    if (event.kind === 'agent-start') {
-      const data: ToolWorkflowAgentStartData = {
-        runId,
-        seq: event.agent.seq,
-        label: event.agent.label,
-        ...event.agent.phase === undefined ? {} : { phase: event.agent.phase },
-        childId: event.agent.childId,
-      }
-      append('tool-workflow/agent-start', data)
-      return
-    }
-    const data: ToolWorkflowAgentEndData = {
-      runId,
-      seq: event.agent.seq,
-      outcome: event.agent.outcome,
-    }
-    append('tool-workflow/agent-end', data)
-  }
-
-  const disposeStart = ctx.on('workflow/agent-start', (info, agent) => {
-    record({ kind: 'agent-start', info, agent })
+    if (!append(session, 'tool-workflow/agent-start', data)) active.delete(info.id)
  })
-  const disposeEnd = ctx.on('workflow/agent-end', (info, agent) => {
-    record({ kind: 'agent-end', info, agent })
+  ctx.on('workflow/agent-end', (info, agent) => {
+    const session = active.get(info.id)
+    if (session === undefined) return
+    const data: ToolWorkflowAgentEndData = {
+      runId: info.id,
+      seq: agent.seq,
+      outcome: agent.outcome,
+    }
+    if (!append(session, 'tool-workflow/agent-end', data)) active.delete(info.id)
  })

  return {
-    bind(run) {
-      runId = run.id
-      append('tool-workflow/run-start', { runId, name: run.meta.name })
-      for (const event of buffered) record(event)
-      buffered.length = 0
+    start(session, run) {
+      if (append(session, 'tool-workflow/run-start', { runId: run.id, name: run.meta.name })) {
+        active.set(run.id, session)
+      }
    },
-    finish(stopReason) {
-      /* v8 ignore next -- execute binds every returned run before result settlement can call finish. */
-      if (runId === undefined) return
-      append('tool-workflow/run-end', { runId, stopReason })
-    },
-    dispose() {
-      disposeStart()
-      disposeEnd()
-      buffered.length = 0
+    finish(runId, stopReason) {
+      const session = active.get(runId)
+      if (session !== undefined) append(session, 'tool-workflow/run-end', { runId, stopReason })
+      active.delete(runId)
    },
+    abandon: (runId) => { active.delete(runId) },
  }
 }

@@ -229,6 +206,7 @@ export function apply(ctx: Context, config: Config): void {
  // schemastery (the exported Config schema) has already filled the defaulted
  // fields; the assertion records that resolution, not a hidden fallback.
  const { toolName, maxResultChars } = config as ResolvedConfig
+  const recorder = createWorkflowRecorder(ctx)
  // Usage policy ships with the tool (the master convention: tool guidance
  // lives in tool plugins as prompt sections, not in the deployment persona).
  ctx.systemPrompt.section({
@@ -303,23 +281,15 @@ export function apply(ctx: Context, config: Config): void {
      // Meta/body validation failures (META_INVALID/SCRIPT_PARSE) throw
      // synchronously here and become isError results via the registry — the
      // model sees the violation list and can correct the call.
-      const recorder = exec.parent === undefined
-        ? createWorkflowRecorder(ctx, parent.session)
-        : undefined
-      let run: WorkflowRun
-      try {
-        run = ctx.workflows.start({
-          script: args.script,
-          meta: args.meta,
-          ...args.args !== undefined ? { args: args.args } : {},
-          parent,
-          signal: exec.signal,
-        })
-      } catch (error: unknown) {
-        recorder?.dispose()
-        throw error
-      }
-      recorder?.bind(run)
+      const run = ctx.workflows.start({
+        script: args.script,
+        meta: args.meta,
+        ...args.args !== undefined ? { args: args.args } : {},
+        parent,
+        signal: exec.signal,
+      })
+      const recordsRun = exec.parent === undefined
+      if (recordsRun) recorder.start(parent.session, run)

      // Bridge the tool's abort signal to the run: if the parent step is aborted while the
      // script is in flight, cancel the whole run. The signal also enters the engine directly, but
@@ -348,9 +318,9 @@ export function apply(ctx: Context, config: Config): void {
          // synthesize cancelled member endings while reaching quiescence.
          await run.dispose()
          /* v8 ignore next -- WorkflowRun.result never rejects by contract, so result is assigned before finally. */
-          if (result !== undefined) recorder?.finish(result.stopReason)
+          if (recordsRun && result !== undefined) recorder.finish(run.id, result.stopReason)
        } finally {
-          recorder?.dispose()
+          if (recordsRun) recorder.abandon(run.id)
        }
      }
    },
--- a/packages/workflow/tool-workflow/src/invariant.ts
+++ b/packages/workflow/tool-workflow/src/invariant.ts
@@ -19,12 +19,9 @@ interface RunTrace {

 type WorkflowTrace = Map<string, RunTrace>

-/** Clone the independent fold before validating one candidate append. */
-function cloneTrace(source: WorkflowTrace): WorkflowTrace {
-  return new Map([...source].map(([runId, run]) => [runId, {
-    ended: run.ended,
-    members: new Map(run.members),
-  }]))
+/** Whether this package owns the candidate Session event. */
+function isWorkflowRecordEvent(event: SessionEvent): boolean {
+  return event.type.startsWith('tool-workflow/')
 }

 /** Require a durable opaque identity to be a non-empty string. */
@@ -50,6 +47,23 @@ function recordOf(event: SessionEvent, fail: InvariantFailure): Record<string, u
  return data as Record<string, unknown>
 }

+/** Copy only the run one candidate can mutate; other committed states stay shared. */
+function cloneTraceForEvent(
+  source: WorkflowTrace,
+  event: SessionEvent,
+  fail: InvariantFailure,
+): WorkflowTrace {
+  const trace = new Map(source)
+  if (event.type === 'tool-workflow/run-start') return trace
+  const data = recordOf(event, fail)
+  const runId = stringId(data.runId, `${event.type} runId`, fail)
+  const run = source.get(runId)
+  if (run !== undefined) {
+    trace.set(runId, { ended: run.ended, members: new Map(run.members) })
+  }
+  return trace
+}
+
 /** Require the named run to exist and remain open. */
 function openRun(trace: WorkflowTrace, runId: string, eventType: string, fail: InvariantFailure): RunTrace {
  const run = trace.get(runId)
@@ -60,7 +74,6 @@ function openRun(trace: WorkflowTrace, runId: string, eventType: string, fail: I

 /** Advance the workflow-record fold with one relevant Session event. */
 function applyEvent(trace: WorkflowTrace, event: SessionEvent, fail: InvariantFailure): void {
-  if (!event.type.startsWith('tool-workflow/')) return
  const data = recordOf(event, fail)
  const runId = stringId(data.runId, `${event.type} runId`, fail)

@@ -107,6 +120,7 @@ function applyEvent(trace: WorkflowTrace, event: SessionEvent, fail: InvariantFa
        fail(`tool-workflow/run-end leaves member seq ${openMembers.join(', ')} open in run ${runId}`)
      }
      run.ended = true
+      run.members.clear()
      return
    }
    default:
@@ -126,23 +140,23 @@ const install: InvariantInstaller = Object.assign((ctx: Context, fail: Invariant

  const seed = (session: Session): WorkflowTrace => {
    const trace: WorkflowTrace = new Map()
-    for (const event of session.events) applyChecked(trace, event, fail)
+    for (const event of session.events.filter(isWorkflowRecordEvent)) applyChecked(trace, event, fail)
    traces.set(session, trace)
    return trace
  }
-  /* v8 ignore next -- session/event always follows list() or session/created seeding. */
-  const traceFor = (session: Session): WorkflowTrace => traces.get(session) ?? seed(session)
-
-  for (const session of ctx.sessions.list()) seed(session)
+  ctx.sessions.list().forEach(seed)
  ctx.on('session/created', (session) => { seed(session) }, { global: true })
  ctx.on('internal/dispatch', (_mode, eventName, args) => {
    if (eventName !== 'session/event') return
    const [session, event] = args as [Session, SessionEvent]
-    const trace = cloneTrace(traceFor(session))
+    if (!isWorkflowRecordEvent(event)) return
+    // session/event dispatch follows list() or session/created seeding.
+    const trace = cloneTraceForEvent(traces.get(session) as WorkflowTrace, event, fail)
    applyChecked(trace, event, fail)
    staged.set(event, { session, trace })
  }, { global: true })
  ctx.on('session/event', (session, event) => {
+    if (!isWorkflowRecordEvent(event)) return
    const candidate = staged.get(event)
    /* v8 ignore next 2 -- internal/dispatch stages the exact session/event callback arguments. */
    if (candidate === undefined || candidate.session !== session) {
--- a/packages/workflow/tool-workflow/tests/tool-workflow.spec.ts
+++ b/packages/workflow/tool-workflow/tests/tool-workflow.spec.ts
@@ -27,7 +27,6 @@ class StubEngine extends WorkflowService {
  settle!: (result: WorkflowResult) => void
  readonly settlements = new Map<WorkflowRunIdType, (result: WorkflowResult) => void>()
  startError: Error | undefined
-  emitMemberDuringStart = false

  start(request: WorkflowStartRequest): WorkflowRun {
    if (this.startError) throw this.startError
@@ -35,12 +34,6 @@ class StubEngine extends WorkflowService {
    const id = WorkflowRunId(`run-${this.requests.length}`)
    const result = new Promise<WorkflowResult>((resolve) => { this.settle = resolve })
    this.settlements.set(id, this.settle)
-    if (this.emitMemberDuringStart) {
-      const info = { id, meta: request.meta }
-      const member = { seq: 1, label: 'synchronous', childId: SessionId('sync-child') }
-      this.emitWorkflowEvent('workflow/agent-start', info, member)
-      this.emitWorkflowEvent('workflow/agent-end', info, { ...member, outcome: 'completed' })
-    }
    request.signal?.addEventListener('abort', () => {
      this.settle({ value: null, stopReason: 'cancelled', error: 'signal', agentsStarted: 0 })
    }, { once: true })
@@ -205,23 +198,6 @@ describe('dsh-tool-workflow', () => {
      ])
  })

-  it('buffers synchronous member events until start returns the run identity', async () => {
-    const { ctx, engine, parent, session } = await setup()
-    engine.emitMemberDuringStart = true
-    const pending = execute(ctx, { script: SCRIPT, meta: META }, { agent: parent })
-    await vi.waitFor(() => { expect(engine.requests).toHaveLength(1) })
-    engine.settleRun(WorkflowRunId('run-1'), {
-      value: null, stopReason: 'completed', agentsStarted: 1,
-    })
-    expect((await pending).isError).toBe(false)
-    expect(session.events.map(event => event.type)).toEqual([
-      'tool-workflow/run-start',
-      'tool-workflow/agent-start',
-      'tool-workflow/agent-end',
-      'tool-workflow/run-end',
-    ])
-  })
-
  it('does not record nested transport executions', async () => {
    const { ctx, engine, parent, session } = await setup()
    const pending = execute(ctx, { script: SCRIPT, meta: META }, {