DeepSeek Harness 会话与事件系统：从事件日志投影模型历史

/**
 * The merge-extensible, append-only source of truth for an agent interaction.
 * Message history is derived from this log. Every event is lossless JSON and
 * sequence numbers stay contiguous, including raw chunks, so persistence can
 * store the canonical log verbatim.
 */
export interface SessionEventMap {

export function 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.
  switch (event.type) {
    // Ordinary prompts and injected context project in user role: the event's
    // model-facing content stays verbatim. Do NOT re-add per-type framing
    // (e.g. `<context>`) here: framing is caller-owned — a producer bakes it
    // into `content`, as agent-instructions does with `<system-reminder>` — or,
    // if reintroduced, must be driven by the event `meta` map and a dedicated
    // renderer, keeping this projection a verbatim pass-through. See the
    // deferred design note in
    // ../../../../.agents/notes/implemented/simplification/2026-07-20-unwrap-injected-content-envelopes.md
    case 'user/message': {
      return event.data
    }
    case 'assistant/message': {
      // Skip an empty-content assistant/message: it exists only to host a
      // max-tokens step's usage and must not inject a content-less assistant
      // turn into the provider transcript.
      if (event.data.message.content.length === 0) return null
      return event.data.message
    }
    case 'tool/result': {
      return event.data.message
    }
    default:
      // A non-surface event (boundary, chunk, log-only record) projects to
      // no message. Merge-extensible union: no assertNever here.
      return null
  }
}

  deriveMessages(): Message[] {
    const surface = this.surface
    const nodes = surface.nodes
    const generation = surface.replaceGeneration
    if (generation !== this.derivedGeneration) {
      this.derived = []
      this.derivedNodes = 0
      this.derivedGeneration = generation
    }
    for (const seq of nodes.slice(this.derivedNodes)) {
      // Surface sequences are built from this.log — seq is always a valid
      // index by construction. The non-null assertion expresses that invariant.
      // oxlint-disable-next-line typescript/no-non-null-assertion
      const msg = this.deriveEventMessage(this.log[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) this.derived.push(msg)
    }
    this.derivedNodes = nodes.length
    return [...this.derived]
  }

/**
 * A3 会话与事件系统实证：追加日志 -> 模型历史投影
 * 运行：node --import tsx packages/core/session/scratch/event-log-demo.ts
 */
import { Context } from '@deepseek-ai/cordis'
import {
  SessionId, SessionStore,
  type Session, type SessionEvent,
} from '@deepseek-ai/dsh-session'
import {
  MessageId, createAssistantMessage, createToolResultMessage, createUserMessage,
} from '@deepseek-ai/dsh-llm'

async function main() {
  const app = new Context()
  await app.plugin(SessionStore)
  const session: Session = app.sessions.create(SessionId('demo'), {})

  // ---- 一轮真实 turn 的事件流（与 agent-loop 驱动器写入顺序一致）----
  session.append('turn/start', { turn: 1 })
  session.append('user/message', createUserMessage({
    content: [{ type: 'text', text: '读一下 package.json 的 name 字段' }],
    source: { kind: 'user' },
  }), { surfaceOp: 'append' })
  session.append('step/start', { turn: 1, step: 1 })
  // 流式 chunk：token 级重放保真，但不投影为消息
  session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: '我来' } })
  session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: '看看' } })
  session.append('tool/call', { turn: 1, step: 1, callId: MessageId('c1'), name: 'fs_read', arguments: '{"path":"package.json"}' })
  session.append('tool/result', {
    turn: 1, step: 1,
    message: createToolResultMessage({
      callId: MessageId('c1'),
      content: [{ type: 'text', text: '{"name":"dsh"}' }],
      isError: false,
    }),
  }, { surfaceOp: 'append' })
  session.append('assistant/message', {
    turn: 1, step: 1,
    message: createAssistantMessage({
      content: [{ type: 'text', text: 'name 字段是 dsh。' }],
    }),
  }, { surfaceOp: 'append' })
  session.append('step/end', { turn: 1, step: 1 })
  session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })

  // ---- 日志全览 ----
  console.log('=== 会话日志（append-only，seq 连续）===')
  for (const event of session.log) {
    const brief = summarize(event)
    console.log(`seq=${String(event.seq).padStart(2)} type=${event.type.padEnd(18)} ${brief}`)
  }

  // ---- 投影 ----
  console.log('\n=== deriveMessages()：模型历史投影 ===')
  const messages = session.deriveMessages()
  for (const m of messages) {
    const text = m.content.map(blockText).join('')
    console.log(`role=${m.role.padEnd(9)} ${text}`)
  }
  console.log(`\n日志 ${session.log.length} 条事件 -> 投影 ${messages.length} 条消息`)

  // ---- 追加一条新输入，增量投影 ----
  console.log('\n=== 追加第二条用户消息后重新投影 ===')
  session.append('user/message', createUserMessage({
    content: [{ type: 'text', text: '再读一下 version 字段' }],
    source: { kind: 'user' },
  }), { surfaceOp: 'append' })
  const messages2 = session.deriveMessages()
  for (const m of messages2) {
    const text = m.content.map(blockText).join('')
    console.log(`role=${m.role.padEnd(9)} ${text}`)
  }
}

function blockText(b: any): string {
  if (b.type === 'tool-result') return b.content?.map(blockText).join('') ?? ''
  return b.text ?? b.callId ?? ''
}

function summarize(event: SessionEvent): string {
  switch (event.type) {
    case 'user/message': return (event.data as any).content?.map(blockText).join('') ?? ''
    case 'assistant/message': return (event.data as any).message?.content?.map(blockText).join('') ?? ''
    case 'tool/call': return `${event.data.name}(${(event.data as any).arguments})`
    case 'tool/result': return `→ ${(event.data as any).message?.content?.map(blockText).join('') ?? ''}`
    case 'assistant/chunk': return (event.data as any).chunk?.text ?? ''
    case 'turn/start': case 'turn/end': case 'step/start': case 'step/end':
      return `turn=${event.data.turn}${'step' in event.data ? ` step=${(event.data as any).step}` : ''}`
    default: return ''
  }
}

main().catch((err) => { console.error(err); process.exit(1) })

=== 会话日志（append-only，seq 连续）===
seq= 0 type=turn/start         turn=1
seq= 1 type=user/message       读一下 package.json 的 name 字段
seq= 2 type=step/start         turn=1 step=1
seq= 3 type=assistant/chunk    我来
seq= 4 type=assistant/chunk    看看
seq= 5 type=tool/call          fs_read({"path":"package.json"})
seq= 6 type=tool/result        → {"name":"dsh"}
seq= 7 type=assistant/message  name 字段是 dsh。
seq= 8 type=step/end           turn=1 step=1
seq= 9 type=turn/end           turn=1

=== deriveMessages()：模型历史投影 ===
role=user      读一下 package.json 的 name 字段
role=user      {"name":"dsh"}
role=assistant name 字段是 dsh。

日志 10 条事件 -> 投影 3 条消息

=== 追加第二条用户消息后重新投影 ===
role=user      读一下 package.json 的 name 字段
role=user      {"name":"dsh"}
role=assistant name 字段是 dsh。
role=user      再读一下 version 字段

/** A tool-result specialization whose model-facing block retains call correlation. */
export interface ToolResultMessage extends Message {
  readonly role: 'user'
  readonly content: [ToolResultBlock]
  readonly source: ToolMessageSource
}

  /** Open one turn before claiming its first proposed step. */
  private async turn(): Promise<boolean> {
    if (this.phase.kind !== 'running') {
      this.throwError(new Error(`agent "${this.id}": turn without driver reservation`))
    }
    const phase = this.phase
    const { signal } = phase.abort
    signal.throwIfAborted()
    const turn = phase.turn + 1
    try {
      this.session.append('turn/start', { turn })
    } catch (error: unknown) {
      this.throwError(error)
    }
    phase.turn = turn
    let turnEnds: TurnEndReason | null = null
    let target: InboxTarget = 'next-turn'
    try {
      while (true) {
        signal.throwIfAborted()
        const step = phase.step + 1
        const decision = await this.preStep(target, { turn, step })
        if (decision.kind === 'reject') {
          turnEnds = { kind: 'blocked' }
          return false
        }
        if (turnEnds && decision.messages.length === 0) break
        // A removed waking message or an enter decision rewritten to empty
        // still owns the initial turn boundary, but it spends no model call.
        if (phase.step === 0 && decision.messages.length === 0) {
          turnEnds = { kind: 'completed' }
          return false
        }
        signal.throwIfAborted()
        this.session.append('step/start', { turn, step })
        phase.step = step
        try {
          for (const message of decision.messages) {
            this.session.append('user/message', message, { surfaceOp: 'append' })
          }
          // max-tokens is sticky: once any step hits the ceiling, later steps
          // that complete normally must not downgrade the turn outcome.
          const stepEnd = await this.step(decision.assembly)
          // max-tokens stays sticky: a later completed step must not
          // downgrade the turn outcome.
          if (turnEnds === null || turnEnds.kind !== 'max-tokens') turnEnds = stepEnd
        } finally {
          this.session.append('step/end', { turn, step })
        }
        signal.throwIfAborted()
        if (turnEnds && this.inbox.nextStep.length === 0) {
          await this.dispatch.serial('agent/turn-stopping', { turn, signal })
          signal.throwIfAborted()
        }
        if (turnEnds && this.inbox.nextStep.length === 0) break
        target = 'next-step'
      }
    } catch (error: unknown) {
      if (signal.aborted) {
        turnEnds = { kind: 'aborted', reason: signal.reason as AgentCancelCause }
        throw error
      }
      // Every failure is structured: an `LlmError` keeps its facts, anything
      // else flattens to `errorChain` text under the `UNKNOWN` code.
      turnEnds = {
        kind: 'error',
        error: error instanceof LlmError
          ? error.failure
          : { message: errorChain(error), code: 'UNKNOWN' },
      }
      this.throwError(error)
    } finally {
      try {
        // oxlint-disable-next-line typescript/no-non-null-assertion -- every exit assigns a turn ending
        this.session.append('turn/end', { turn, reason: turnEnds! })
      } catch (error: unknown) {
        this.throwError(error)
      }
    }
    if (!this.inbox.hasPending) return false
    phase.abort = new AbortController()
    // A fresh controller makes a latch set on the old one stale: the live driver claims the queue itself.
    phase.wakeRequested = false
    phase.step = 0
    return true
  }

/** Append a started call and return the event seq that its result must cite. */
function appendToolCall(session: Session, turn: number, step: number, block: ToolCallBlock): number {
  const event = session.append('tool/call', { turn, step, callId: block.id, name: block.name, arguments: block.arguments })
  return event.seq
}

/** Append a model-ordered result linked to its call event. */
function appendToolResult(
  session: Session,
  turn: number,
  step: number,
  block: ToolCallBlock,
  result: ToolExecutionResult,
  callSeq: number,
): void {
  const message = createToolResultMessage({
    callId: block.id,
    content: result.content,
    isError: result.isError,
  })
  session.append('tool/result', {
    turn, step,
    message,
    ...result.error?.info ? { error: result.error.info } : {},
    // The tool's private presentation payload (e.g. a result-time diff),
    // persisted so a UI bridge reproduces the card on replay.
    ...result.meta !== undefined ? { meta: result.meta } : {},
  }, { surfaceOp: 'append', sourceEventSeqs: [callSeq] })
}