DeepSeek Harness: StreamChunk Protocol and Error Handling
Last updated: 2026-08-31
Streaming output is the core Agent interaction experience — users don't want to wait 30 seconds for a complete reply, they want to watch the Agent think word by word. The StreamChunk protocol is DSH's unified abstraction for streaming output, and error handling ensures graceful degradation when exceptions occur.
💡 Tip: The hard part of streaming output isn't "sending" — it's "what to do when errors happen." The StreamChunk protocol treats errors as a type of chunk, letting consumers handle normal and abnormal cases uniformly.
📋 Prerequisites: Completed 24-llm-adapter.md, understand LLM adapters
1. What You'll Learn
- Streaming output protocol: StreamChunk
- Chunk types: text/tool_call/tool_result
- Error recovery mechanisms
- Interruption and cancellation
- Retry strategies
- Logging and observability
2. StreamChunk Protocol Details
(1) ▶ Example 1
TYPESCRIPT
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
}
(2) ▶ Example 2
graph LR
START[Stream start] --> TEXT[TextChunk × N]
TEXT --> TC[ToolCallChunk]
TC --> TR[ToolResultChunk]
TR --> TEXT2[TextChunk × N]
TEXT2 --> DONE[DoneChunk]
A typical Agent conversation flow:
- LLM outputs text → TextChunk
- LLM decides to call a tool → ToolCallChunk
- Tool execution completes → ToolResultChunk
- LLM continues output → TextChunk
- Stream ends → DoneChunk
▶ Example 3
TYPESCRIPT
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
}
}
3. Chunk Types in Detail
(1) TextChunk
Text content fragments, incrementally output:
TYPESCRIPT
// LLM outputs "Hello, world!"
// May produce multiple TextChunks:
// chunk 1: { type: 'text', content: 'Hello' }
// chunk 2: { type: 'text', content: ', ' }
// chunk 3: { type: 'text', content: 'world' }
// chunk 4: { type: 'text', content: '!' }
Consumers should concatenate all TextChunks rather than displaying them individually.
(2) ToolCallChunk
LLM requests a tool call:
TYPESCRIPT
{
type: 'tool_call',
id: 'call_abc123',
name: 'file_edit',
arguments: '{"action":"read","path":"src/index.ts"}',
index: 0
}
Note: arguments is a JSON string that needs parsing.
(3) ToolResultChunk
Results after tool execution:
TYPESCRIPT
{
type: 'tool_result',
id: 'result_xyz789',
toolCallId: 'call_abc123',
result: { content: 'export const name = ...' },
isError: false
}
isError: true indicates tool execution failure; result contains error information.
(4) Multiple Tool Calls
A single conversation may call multiple tools:
TEXT
📖 Display only
TextChunk: "Let me check two files"
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: "The contents of both files above..."
DoneChunk: { reason: "stop" }
4. Error Recovery Mechanism
(1) Error Classification
TYPESCRIPT
// Non-recoverable error (e.g., invalid API key)
{
type: 'error',
error: new Error('Invalid API key'),
recoverable: false
}
// Recoverable error (e.g., rate limit)
{
type: 'error',
error: new Error('Rate limit exceeded'),
recoverable: true,
retryAfter: 5000
}
// Recoverable error (e.g., network jitter)
{
type: 'error',
error: new Error('Connection timeout'),
recoverable: true,
retryAfter: 2000
}
(2) Auto-Recovery Flow
graph TD
ERR[ErrorChunk] --> CHECK{recoverable?}
CHECK -->|No| FAIL[Terminate stream + DoneChunk<br/>reason: error]
CHECK -->|Yes| WAIT[Wait retryAfter]
WAIT --> RETRY[Retry request]
RETRY --> SUCCESS{Success?}
SUCCESS -->|Yes| CONTINUE[Continue stream]
SUCCESS -->|No| ERR2[Another ErrorChunk]
ERR2 --> CHECK2{Retries exhausted?}
CHECK2 -->|No| WAIT
CHECK2 -->|Yes| FAIL
(3) Manual Recovery
TYPESCRIPT
for await (const chunk of stream) {
if (chunk.type === 'error') {
if (chunk.recoverable) {
ctx.logger.warn(`Stream error, recoverable: ${chunk.error.message}`)
// Framework auto-retries
} else {
ctx.logger.error(`Stream error, non-recoverable: ${chunk.error.message}`)
break
}
}
}
(4) Tool Error Handling
When a tool execution fails, the framework returns the error as a ToolResultChunk to the LLM:
TYPESCRIPT
{
type: 'tool_result',
id: 'result_1',
toolCallId: 'call_1',
result: { error: true, message: 'Permission denied: /etc/passwd' },
isError: true
}
The LLM can then choose to:
- Retry with a different path
- Inform the user of insufficient permissions
- Complete the task another way
5. Interruption and Cancellation
(1) User Cancellation
When a user clicks the "Stop" button in the Web UI, the stream is cancelled:
TYPESCRIPT
const controller = new AbortController()
const stream = ctx.llm.stream(request, { signal: controller.signal })
// User cancels
controller.abort()
// Stream produces DoneChunk
// { type: 'done', reason: 'cancel' }
(2) Timeout Cancellation
TYPESCRIPT
const controller = new AbortController()
const timer = setTimeout(() => controller.abort(), 60000)
try {
for await (const chunk of ctx.llm.stream(request, { signal: controller.signal })) {
// Process chunk
}
} finally {
clearTimeout(timer)
}
(3) Conditional Cancellation
TYPESCRIPT
let tokenCount = 0
for await (const chunk of stream) {
if (chunk.type === 'text') {
tokenCount += chunk.content.length
if (tokenCount > 10000) {
controller.abort()
break
}
}
}
(4) Cancellation Cleanup
After cancellation, the framework ensures:
- Stops receiving new chunks
- Already-received chunks are processed normally
- Produces
DoneChunk { reason: 'cancel' } - Releases network connections
6. Retry Strategies
(1) Built-in Retry Strategies
| Error Type | Retries | Backoff Strategy |
|---|---|---|
| Rate limit (429) | 3 | Exponential backoff |
| Network timeout | 2 | Fixed interval |
| Server error (5xx) | 2 | Exponential backoff |
| Auth error (401) | 0 | No retry |
| Request error (400) | 0 | No retry |
(2) Configuring Retries
TYPESCRIPT
export const Config = Schema.object({
maxRetries: Schema.number().default(3).description('Maximum retry attempts'),
retryDelay: Schema.number().default(1000).description('Initial retry delay (ms)'),
retryMultiplier: Schema.number().default(2).description('Delay multiplier for exponential backoff')
})
(3) Custom Retry Logic
TYPESCRIPT
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 // Re-enter while loop
}
// Process normal chunks
}
return // Success
} catch (error) {
attempt++
if (attempt > maxRetries) throw error
}
}
}
7. Logging and Observability
(1) Stream Logging
TYPESCRIPT
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}`)
})
(2) Performance Metrics
TYPESCRIPT
interface StreamMetrics {
ttfb: number // Time to First Byte
totalDuration: number // Total duration
chunkCount: number // Chunk count
toolCallCount: number // Tool call count
retryCount: number // Retry count
tokenUsage: TokenUsage // Token usage
}
(3) Structured Logging
TYPESCRIPT
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
})
❓ FAQ
Q After stream cancellation, are already-called tools rolled back?
A No. Tool execution takes effect immediately (e.g., files already created). Cancellation only stops subsequent LLM output.
Q Does the stream always end after an ErrorChunk?
A Not necessarily. If the error is recoverable, the framework retries and the stream may continue. Only non-recoverable errors or exhausted retries produce a DoneChunk to end the stream.
Q How to calculate TTFB for streaming output?
A Record the time the first chunk arrives:
typescript const start = Date.now() for await (const chunk of stream) { if (chunk.type === 'text') { const ttfb = Date.now() - start ctx.logger.info(`TTFB: ${ttfb}ms`) break } } Q What is the index in multiple ToolCallChunks for?
A index identifies the order of parallel tool calls. A single LLM response may contain multiple tool_calls; index increments from 0.
Q Can streaming and complete output be supported simultaneously?
A Yes. Adapters implement both
complete() and stream() methods. Callers choose as needed.Q How to mock streaming output for testing?
A
typescript async function* mockStream(): AsyncIterable<StreamChunk> { yield { type: 'text', content: 'Hello' } yield { type: 'text', content: ', world!' } yield { type: 'done', reason: 'stop' } } 📖 Summary
- StreamChunk five types: text, tool_call, tool_result, error, done
- Stream lifecycle: text output → tool call → tool result → continue output → done
- Errors are classified as recoverable and non-recoverable; recoverable errors auto-retry
- Cancellation via AbortController produces
DoneChunk { reason: 'cancel' } - Retry strategies vary by error type; exponential backoff is the default
- Logging and observability through lifecycle events and structured logging
📝 Exercises
1. ⭐ Basic： Write a consumer that iterates over a StreamChunk stream, counting text/tool_call/tool_result/error/done chunks separately. Test with a mock stream.
2. ⭐⭐ Intermediate： Implement a stream consumer with a timeout — if no chunk arrives within 30 seconds, automatically cancel the stream. Use AbortController for cancellation.
3. ⭐⭐⭐ Challenge： Implement a custom retry wrapper around ctx.llm.stream() that auto-retries on recoverable errors (exponential backoff, max 3 attempts). Log each retry's delay and result. Test with a mock LLM adapter that alternates between failure and success.
