DeepSeek Harness: StreamChunk プロトコルとエラー処理

type StreamChunk =
  | TextChunk
  | ToolCallChunk
  | ToolResultChunk
  | ErrorChunk
  | DoneChunk

interface TextChunk {
  type: 'text'
  content: string
}

interface ToolCallChunk {
  type: 'tool_call'
  id: string
  name: string
  arguments: string
  index: number
}

interface ToolResultChunk {
  type: 'tool_result'
  id: string
  toolCallId: string
  result: any
  isError: boolean
}

interface ErrorChunk {
  type: 'error'
  error: Error
  recoverable: boolean
  retryAfter?: number
}

interface DoneChunk {
  type: 'done'
  reason: 'stop' | 'tool_use' | 'length' | 'cancel' | 'error'
  usage?: TokenUsage
}

interface TokenUsage {
  promptTokens: number
  completionTokens: number
  totalTokens: number
}

const stream = ctx.llm.stream({
  messages: [{ role: 'user', content: 'list project files' }],
  tools: availableTools
})

for await (const chunk of stream) {
  switch (chunk.type) {
    case 'text':
      process.stdout.write(chunk.content)
      break

    case 'tool_call':
      console.log(`\n🔧 Tool call: ${chunk.name}`)
      console.log(`   Arguments: ${chunk.arguments}`)
      break

    case 'tool_result':
      if (chunk.isError) {
        console.log(`   ❌ Tool error: ${chunk.result}`)
      } else {
        console.log(`   ✅ Tool result: ${JSON.stringify(chunk.result)}`)
      }
      break

    case 'error':
      console.error(`\n⚠️ Error: ${chunk.error.message}`)
      if (chunk.recoverable) {
        console.log(`   Will retry in ${chunk.retryAfter}ms`)
      }
      break

    case 'done':
      console.log(`\n✅ Done (${chunk.reason})`)
      if (chunk.usage) {
        console.log(`   Token usage: ${chunk.usage.totalTokens}`)
      }
      break
  }
}

// LLM が "Hello, world!" を出力
// 複数の TextChunk に分割される可能性：
// chunk 1：{ type：'text', content：'Hello' }
// chunk 2：{ type：'text', content：', ' }
// chunk 3：{ type：'text', content：'world' }
// chunk 4：{ type：'text', content：'!' }

{
  type: 'tool_call',
  id: 'call_abc123',
  name: 'file_edit',
  arguments: '{"action":"read","path":"src/index.ts"}',
  index: 0
}

{
  type: 'tool_result',
  id: 'result_xyz789',
  toolCallId: 'call_abc123',
  result: { content: 'export const name = ...' },
  isError: false
}

TextChunk："2つのファイルを確認します"
ToolCallChunk：{ name："file_edit", id："call_1", index：0 }
ToolCallChunk：{ name："file_edit", id："call_2", index：1 }
ToolResultChunk：{ toolCallId："call_1", result：... }
ToolResultChunk：{ toolCallId："call_2", result：... }
TextChunk："上記2ファイルの内容..."
DoneChunk：{ reason："stop" }

// リカバリ不可エラー（例：無効な API キー）
{
  type: 'error',
  error: new Error('Invalid API key'),
  recoverable: false
}

// リカバリ可能エラー（例：レート制限）
{
  type: 'error',
  error: new Error('Rate limit exceeded'),
  recoverable: true,
  retryAfter: 5000
}

// リカバリ可能エラー（例：ネットワークジッタ）
{
  type: 'error',
  error: new Error('Connection timeout'),
  recoverable: true,
  retryAfter: 2000
}

for await (const chunk of stream) {
  if (chunk.type === 'error') {
    if (chunk.recoverable) {
      ctx.logger.warn(`Stream error, recoverable: ${chunk.error.message}`)
      // フレームワークが自動リトライ
    } else {
      ctx.logger.error(`Stream error, non-recoverable: ${chunk.error.message}`)
      break
    }
  }
}

{
  type: 'tool_result',
  id: 'result_1',
  toolCallId: 'call_1',
  result: { error: true, message: 'Permission denied: /etc/passwd' },
  isError: true
}

const controller = new AbortController()

const stream = ctx.llm.stream(request, { signal: controller.signal })

// ユーザーがキャンセル
controller.abort()

// ストリームが DoneChunk を生成
// { type：'done', reason：'cancel' }

const controller = new AbortController()
const timer = setTimeout(() => controller.abort(), 60000)

try {
  for await (const chunk of ctx.llm.stream(request, { signal: controller.signal })) {
    // チャンクを処理
  }
} finally {
  clearTimeout(timer)
}

let tokenCount = 0

for await (const chunk of stream) {
  if (chunk.type === 'text') {
    tokenCount += chunk.content.length
    if (tokenCount > 10000) {
      controller.abort()
      break
    }
  }
}

export const Config = Schema.object({
  maxRetries: Schema.number().default(3).description('最大リトライ回数'),
  retryDelay: Schema.number().default(1000).description('初期リトライ遅延（ms）'),
  retryMultiplier: Schema.number().default(2).description('指数バックオフの遅延乗数')
})

async function streamWithRetry(
  ctx: Context,
  request: LLMRequest,
  maxRetries = 3
): Promise<void> {
  let attempt = 0
  
  while (attempt <= maxRetries) {
    try {
      for await (const chunk of ctx.llm.stream(request)) {
        if (chunk.type === 'error' && chunk.recoverable) {
          attempt++
          const delay = Math.min(1000 * Math.pow(2, attempt), 30000)
          ctx.logger.warn(`retry ${attempt}/${maxRetries} in ${delay}ms`)
          await sleep(delay)
          break
        }
      }
      return
    } catch (error) {
      attempt++
      if (attempt > maxRetries) throw error
    }
  }
}

ctx.on('llm/stream/start', (request) => {
  ctx.logger.info(`stream started：model=${request.model}`)
})

ctx.on('llm/stream/chunk', (chunk) => {
  ctx.logger.debug(`chunk：type=${chunk.type}`)
})

ctx.on('llm/stream/end', (done) => {
  ctx.logger.info(`stream ended：reason=${done.reason}, tokens=${done.usage?.totalTokens}`)
})

interface StreamMetrics {
  ttfb: number
  totalDuration: number
  chunkCount: number
  toolCallCount: number
  retryCount: number
  tokenUsage: TokenUsage
}

ctx.logger.info('llm_request', {
  model: request.model,
  messageCount: request.messages.length,
  toolCount: request.tools?.length || 0,
  stream: true
})

ctx.logger.info('llm_response', {
  model: response.model,
  duration: Date.now() - startTime,
  tokens: response.usage?.totalTokens
})