A State Machine That Catches Malformed AI Event Streams Before They Reach Your Agent

This article is the fourth installment of the "Developing a Coding Agent from Scratch" series. In the previous article, we implemented a generic EventStream channel: the Provider calls push() to write events, and the Agent reads events using for await...of.

However, EventStream only transports data and does not know whether the order of events is correct. For example, the following object fully conforms to the TypeScript type of StreamEvent:

{ type: "text_delta", contentIndex: 0, delta: "你好" }

If it is not preceded by start and text_start, it is still an illegal event according to the protocol. TypeScript can only check the fields of a single object; it cannot remember what happened previously in this stream. Therefore, we also need a validator with memory, i.e., a State Machine.

After completion, events will pass through sequence validation before entering EventStream:

Provider -> EventStream.push(event) -> StreamEventValidator.accept(event) -> Agent

Legal events continue into the queue; illegal events throw a StreamSequenceError and deliver the same error to the event producer, the async iterator, and result().

States of the Event Stream

First, let's look at a legal text response:

start
  -> text_start(0) 
  -> text_delta(0, "你")
  -> text_delta(0, "好")
  -> text_end(0, "你好")
  -> done

The entire response has three phases:

stateDiagram-v2
    [*] --> idle
    idle --> streaming: start
    streaming --> streaming: content block events
    streaming --> terminal: done or error
    terminal --> [*]

• idle: start has not been received yet.
• streaming: has started, can receive content blocks or terminal events.
• terminal: done or error has been received, no more events can be received.

In addition to the overall stream phase, we also need to track the currently assembling content block. An AssistantMessage can contain text, thinking processes, and tool calls:

content[0] -> text
content[1] -> thinking
content[2] -> tool_call

Therefore, the state machine must also ensure:

1. At most one unfinished content block exists at any time.
2. contentIndex starts from 0 and increments consecutively.
3. Deltas must belong to the currently active block.
4. The complete content carried by end must equal the accumulated result of all deltas.
5. A new index can only start after a block is fully completed.

Defining States and Sequence Errors

Create validation.ts in the packages/ai/src/utils directory, first importing the common event and message content types:

import type { AssistantContent, StreamEvent } from "../types.ts";

type StreamPhase = "idle" | "streaming" | "terminal";

Then define the active content block:

type ActiveBlock =
	| { kind: "text"; contentIndex: number; value: string }
	| { kind: "thinking"; contentIndex: number; value: string }
	| {
			kind: "tool_call";
			contentIndex: number;
			id: string;
			name: string;
			argumentsJson: string;
			sawArgumentsDelta: boolean;
	  };

Text and thinking blocks only need to save the accumulated string. Tool calls, besides the index, also need to save the tool's id, name, and the JSON string arriving in segments.

sawArgumentsDelta cannot be replaced directly with argumentsJson.length > 0. Because the Provider might actually push an empty string delta; "no delta received" and "an empty delta received" are two different protocol states.

Next, define the internal state of the entire stream:

interface SequenceState {
	phase: StreamPhase;
	nextContentIndex: number;
	activeBlock?: ActiveBlock;
	completedContent: AssistantContent[];
}

The four fields here represent:

Field	Purpose
phase	Whether it is currently before start, in streaming, or terminated
nextContentIndex	The index that the next content block must use
activeBlock	The content block currently being accumulated
completedContent	The message content that has been fully completed, used to verify the final message

A plain Error can only tell us "something went wrong". When debugging a Provider, we also need to know which event failed at which phase. Therefore, define an identifiable error type:

export class StreamSequenceError extends Error {
	readonly eventType: StreamEvent["type"];
	readonly phase: StreamPhase;

	constructor(eventType: StreamEvent["type"], phase: StreamPhase, detail: string) {
		super(`Invalid stream event "${eventType}" in phase "${phase}": ${detail}`);
		this.name = "StreamSequenceError";
		this.eventType = eventType;
		this.phase = phase;
	}
}

export interface StreamEventValidator {
	accept(event: StreamEvent): void;
}

accept() has no return value. It returns directly when the event is legal, and throws synchronously when illegal. This allows EventStream.push() to complete validation before handing the event to consumers.

Add two helper functions to unify error formatting and check the index of new content blocks:

function reject(event: StreamEvent, state: SequenceState, detail: string): never {
	throw new StreamSequenceError(event.type, state.phase, detail);
}

function assertStartIndex(event: StreamEvent, state: SequenceState, received: number): void {
	if (received !== state.nextContentIndex) {
		reject(event, state, `expected contentIndex ${state.nextContentIndex}, received ${received}`);
	}
}

The return type of reject() is never, indicating that the function will definitely terminate the current control flow. TypeScript can therefore continue narrowing union types after calling reject().

Validating JSON Data

The common type for tool parameters is Record<string, unknown>, but not all JavaScript values can be written to JSON. For example, undefined, functions, Date, Map, and Infinity are not legal JSON data.

First, define the recursive type that JSON can express:

type JsonValue = null | boolean | number | string | JsonValue[] | { [key: string]: JsonValue };

Then add runtime checks:

function isJsonValue(value: unknown): value is JsonValue {
	if (value === null || typeof value === "string" || typeof value === "boolean") {
		return true;
	}
	if (typeof value === "number") {
		return Number.isFinite(value);
	}
	if (Array.isArray(value)) {
		return value.every(isJsonValue);
	}
	if (typeof value === "object") {
		const prototype = Object.getPrototypeOf(value);
		if (prototype !== Object.prototype && prototype !== null) {
			return false;
		}
		return Object.values(value as Record<string, unknown>).every(isJsonValue);
	}
	return false;
}

function isJsonObject(value: unknown): value is Record<string, JsonValue> {
	return value !== null && typeof value === "object" && !Array.isArray(value) && isJsonValue(value);
}

isJsonValue() recursively checks arrays and plain objects. Numbers additionally use Number.isFinite() to exclude NaN, Infinity, and -Infinity, which JSON cannot accurately represent.

Tool parameters must also have an object as the root, so isJsonObject() excludes null and arrays. This checks "is it a legal JSON object", not whether { path: 123 } conforms to a specific tool's parameter schema; specific tool parameter validation belongs to the subsequent Task 03c.

Tool parameters are strings in deltas and objects in the end event. JSON.parse() creates new objects each time, so === cannot be used to compare references; deep comparison by structure and value is needed:

function jsonDeepEqual(left: unknown, right: unknown): boolean {
	if (Object.is(left, right)) {
		return true;
	}
	if (Array.isArray(left) || Array.isArray(right)) {
		return (
			Array.isArray(left) &&
			Array.isArray(right) &&
			left.length === right.length &&
			left.every((value, index) => jsonDeepEqual(value, right[index]))
		);
	}
	if (left === null || right === null || typeof left !== "object" || typeof right !== "object") {
		return false;
	}
	const leftRecord = left as Record<string, unknown>;
	const rightRecord = right as Record<string, unknown>;
	const leftKeys = Object.keys(leftRecord).sort();
	const rightKeys = Object.keys(rightRecord).sort();
	return (
		leftKeys.length === rightKeys.length &&
		leftKeys.every((key, index) => key === rightKeys[index] && jsonDeepEqual(leftRecord[key], rightRecord[key]))
	);
}

Object keys are sorted first, so { path: "A", line: 1 } and { line: 1, path: "A" } are considered equal.

Creating the Validator

Next, implement createStreamEventValidator(). It is not a global singleton; each call creates an independent state:

export function createStreamEventValidator(): StreamEventValidator {
	const state: SequenceState = {
		phase: "idle",
		nextContentIndex: 0,
		completedContent: [],
	};

	return {
		accept(event) {
			if (state.phase === "idle") {
				if (event.type !== "start") {
					reject(event, state, 'expected "start" as the first event');
				}
				state.phase = "streaming";
				return;
			}

			if (state.phase === "terminal") {
				reject(event, state, "no events are allowed after a terminal event");
			}
			if (event.type === "start") {
				reject(event, state, '"start" may appear only once');
			}

			switch (event.type) {
				// Various event branches will be added sequentially below
			}
		},
	};
}

The state in the closure belongs only to this one validator. When two concurrent model requests call the factory function separately, they will not share phase, index, or content.

Before entering the switch, three global rules are handled:

• The first event must be start.
• start can only appear once.
• After entering terminal, no more events are allowed.

The following code is all added inside this switch (event.type).

Implementing Text and Thinking Content Blocks

Text and thinking have the same structure, consisting of start, several deltas, and end, so they can be handled together.

First, implement the start event:

case "text_start":
case "thinking_start": {
	if (state.activeBlock) {
		reject(event, state, `cannot start a new block while ${state.activeBlock.kind} is active`);
	}
	assertStartIndex(event, state, event.contentIndex);
	state.activeBlock = {
		kind: event.type === "text_start" ? "text" : "thinking",
		contentIndex: event.contentIndex,
		value: "",
	};
	return;
}

A new block can only start when there is no active block and contentIndex equals nextContentIndex. The index is not incremented immediately here because this block has not ended yet.

Then implement delta:

case "text_delta":
case "thinking_delta": {
	const expectedKind = event.type === "text_delta" ? "text" : "thinking";
	const active = state.activeBlock;
	if (!active || active.kind === "tool_call" || active.kind !== expectedKind) {
		reject(event, state, `expected active ${expectedKind} block`);
	}
	if (event.contentIndex !== active.contentIndex) {
		reject(event, state, `expected contentIndex ${active.contentIndex}, received ${event.contentIndex}`);
	}
	active.value += event.delta;
	return;
}

For example, if the current active block is text_start(0), then thinking_delta(0) and text_delta(1) will both be rejected. Only when the type and index match simultaneously will the delta be appended to active.value.

Finally, handle end:

case "text_end":
case "thinking_end": {
	const expectedKind = event.type === "text_end" ? "text" : "thinking";
	const active = state.activeBlock;
	if (!active || active.kind === "tool_call" || active.kind !== expectedKind) {
		reject(event, state, `expected active ${expectedKind} block`);
	}
	if (event.contentIndex !== active.contentIndex) {
		reject(event, state, `expected contentIndex ${active.contentIndex}, received ${event.contentIndex}`);
	}
	if (event.content !== active.value) {
		reject(event, state, "end content does not match accumulated deltas");
	}
	state.completedContent.push(
		expectedKind === "text"
			? { type: "text", text: active.value }
			: { type: "thinking", thinking: active.value },
	);
	state.activeBlock = undefined;
	state.nextContentIndex += 1;
	return;
}

The order of modifications here is important: first check the type, index, and complete content; only after all pass do we write to completedContent, clear the active block, and increment the index. This way, an illegal end does not pollute the state.

Implementing Tool Call Content Blocks

Tool calls also have start, delta, and end, but their deltas are JSON string fragments. For example:

Segment 1: {"pa
Segment 2: th":"README
Segment 3: .md"}

The first two segments are not complete JSON, so JSON.parse() cannot be called on each delta individually. The correct approach is to concatenate the strings first and parse only once when tool_call_end arrives.

First, record the tool identity:

case "tool_call_start": {
	if (state.activeBlock) {
		reject(event, state, `cannot start a new block while ${state.activeBlock.kind} is active`);
	}
	assertStartIndex(event, state, event.contentIndex);
	state.activeBlock = {
		kind: "tool_call",
		contentIndex: event.contentIndex,
		id: event.id,
		name: event.name,
		argumentsJson: "",
		sawArgumentsDelta: false,
	};
	return;
}

When receiving parameter fragments, only type and index checks and string accumulation are performed:

case "tool_call_delta": {
	const active = state.activeBlock;
	if (!active || active.kind !== "tool_call") {
		reject(event, state, "expected active tool_call block");
	}
	if (event.contentIndex !== active.contentIndex) {
		reject(event, state, `expected contentIndex ${active.contentIndex}, received ${event.contentIndex}`);
	}
	active.sawArgumentsDelta = true;
	active.argumentsJson += event.argumentsDelta;
	return;
}

Only at end do we parse and verify the complete result:

case "tool_call_end": {
	const active = state.activeBlock;
	if (!active || active.kind !== "tool_call") {
		reject(event, state, "expected active tool_call block");
	}
	if (event.contentIndex !== active.contentIndex) {
		reject(event, state, `expected contentIndex ${active.contentIndex}, received ${event.contentIndex}`);
	}

	let parsed: Record<string, JsonValue>;
	if (active.sawArgumentsDelta) {
		let candidate: unknown;
		try {
			candidate = JSON.parse(active.argumentsJson);
		} catch {
			reject(event, state, "tool arguments must be valid JSON");
		}
		if (!isJsonObject(candidate)) {
			reject(event, state, "tool arguments must have a JSON object root");
		}
		parsed = candidate;
	} else {
		if (!isJsonObject(event.toolCall.arguments)) {
			reject(event, state, "tool arguments must have a JSON object root");
		}
		parsed = event.toolCall.arguments;
	}

	if (event.toolCall.id !== active.id) {
		reject(event, state, `tool id must remain "${active.id}"`);
	}
	if (event.toolCall.name !== active.name) {
		reject(event, state, `tool name must remain "${active.name}"`);
	}
	if (
		!isJsonObject(event.toolCall.arguments) ||
		(active.sawArgumentsDelta && !jsonDeepEqual(parsed, event.toolCall.arguments))
	) {
		reject(event, state, "final tool arguments must match accumulated arguments");
	}

	state.completedContent.push({
		type: "tool_call",
		id: active.id,
		name: active.name,
		arguments: parsed,
	});
	state.activeBlock = undefined;
	state.nextContentIndex += 1;
	return;
}

This section completes four layers of checks:

1. end must belong to the current tool block.
2. Parameters must be complete JSON, and the root node must be an object.
3. The tool's id and name cannot change from start to end.
4. The toolCall.arguments in end must match the object assembled from all deltas.

If tool_call_delta was never received, the parameter object carried by the end event is used directly. This allows tool calls with no parameters or where the Provider gives parameters all at once to pass as well.

Implementing done and error

Terminal events not only mean "no subsequent events" but also carry the final AssistantMessage. The state machine needs to confirm that the final message is indeed composed of the preceding events.

The rules for normal completion and abnormal interruption are slightly different:

• done indicates normal completion and cannot leave an unfinished content block.
• error may occur after any delta and can retain already generated text or thinking fragments.
• A half-finished tool JSON cannot become an executable ToolCallContent, so the unfinished tool block must be discarded on error.

First, add before jsonDeepEqual() and before the factory function:

function getExpectedTerminalContent(state: SequenceState): AssistantContent[] {
	const content = [...state.completedContent];
	const active = state.activeBlock;
	if (active?.kind === "text") {
		content.push({ type: "text", text: active.value });
	}
	if (active?.kind === "thinking") {
		content.push({ type: "thinking", thinking: active.value });
	}
	return content;
}

Note that completedContent is copied first here, without directly modifying the state. This function is only responsible for calculating what content an error message should carry.

Then add two terminal branches in the switch:

case "done": {
	if (state.activeBlock) {
		reject(event, state, "done cannot terminate an active block");
	}
	if (event.reason !== event.message.stopReason) {
		reject(event, state, "event reason must match message.stopReason");
	}
	if (!jsonDeepEqual(event.message.content, state.completedContent)) {
		reject(event, state, "done message content must match completed stream content");
	}
	state.phase = "terminal";
	return;
}
case "error": {
	if (event.reason !== event.message.stopReason) {
		reject(event, state, "event reason must match message.stopReason");
	}
	if (!jsonDeepEqual(event.message.content, getExpectedTerminalContent(state))) {
		reject(event, state, "error message content must match safe partial stream content");
	}
	state.phase = "terminal";
	return;
}
default:
	reject(event, state, "event is not implemented at this step");

reason and message.stopReason are two expression positions for the same fact and must be consistent. The content also cannot just trust what the terminal event declares itself; it must be deeply compared with the content the state machine has already observed.

Connecting to EventStream

Now the validator can independently receive StreamEvent. The next step is to make the real EventStream.push() call it automatically.

Open packages/ai/src/utils/event-stream.ts and add at the top of the file:

import type { AssistantMessage, StreamEvent } from "../types.ts";
import { createStreamEventValidator } from "./validation.ts";

Do not modify the generic EventStream<TEvent, TResult> already implemented in the previous article. After the class definition, add a concrete type and factory function for AssistantMessage:

export type AssistantMessageEventStream = EventStream<StreamEvent, AssistantMessage>;

export function createAssistantMessageEventStream(): AssistantMessageEventStream {
	const validator = createStreamEventValidator();

	return new EventStream<StreamEvent, AssistantMessage>({
		validate(event) {
			validator.accept(event);
		},
		isTerminal(event) {
			return event.type === "done" || event.type === "error";
		},
		getResult(event) {
			if (event.type === "done" || event.type === "error") {
				return event.message;
			}
			throw new Error("Expected terminal stream event");
		},
	});
}

This factory connects three things:

EventStream Option	Specific Behavior
validate(event)	Hands each event to the validator unique to this stream
isTerminal(event)	Identifies done and error as protocol terminal events
getResult(event)	Extracts the final AssistantMessage from the terminal event

createStreamEventValidator() must be called inside the factory function here. If the validator were placed at the module top level, multiple model requests would share state: after the first stream enters terminal, the start of the second stream would also be incorrectly rejected.

The previous article's EventStream.push() already includes error propagation logic:

try {
	this.options.validate(event);
	// Determine terminal event and extract final result
} catch (cause) {
	const error = normalizeError(cause);
	this.fail(error);
	throw error;
}

So after the validator throws, there is no need to add another error channel. push() will throw synchronously, while fail() will reject the waiting consumers and result().

Maintaining the Public Entry Point

Finally, open packages/ai/src/index.ts and add the public exports:

export type { AssistantMessageEventStream, EventStreamOptions } from "./utils/event-stream.ts";
export { createAssistantMessageEventStream, EventStream } from "./utils/event-stream.ts";
export type { StreamEventValidator } from "./utils/validation.ts";
export { createStreamEventValidator, StreamSequenceError } from "./utils/validation.ts";

What is publicly exposed:

• createAssistantMessageEventStream(): The main entry point for Providers to create message event streams.
• AssistantMessageEventStream: The concrete event stream type.
• createStreamEventValidator(): Factory for when the sequence validator needs to be used standalone.
• StreamSequenceError: An identifiable protocol sequence error for callers.
• StreamEventValidator: The minimal interface of the validator.

SequenceState, ActiveBlock, JsonValue, and the deep comparison function are all just implementation details and should not be exported from the package entry point. This way, when the internal structure of the state machine is adjusted later, code using @di-code/ai will not be broken.

Formatting and Building

In PowerShell, navigate to the project root directory:

Set-Location D:\pi\di-code

Use the project's already installed fixed version of Biome to format these three production files:

npx --no-install biome check --write packages\ai\src\index.ts packages\ai\src\utils\event-stream.ts packages\ai\src\utils\validation.ts

Then execute static checks and the @di-code/ai build:

npm run check
npm run build --workspace @di-code/ai

npm run check should have no warnings or errors, and the build should successfully generate packages/ai/dist. These two commands only confirm formatting, types, and build results; testing of event sequence behavior is not covered in this article.

Summary

Up to this point, we have added a layer of stateful protocol validation in front of the generic EventStream. What it solves is not "does the event object have these fields", but "can this event appear at this moment".

The entire data flow is now:

Provider generates unified StreamEvent
    -> EventStream.push()
    -> StreamEventValidator.accept()
    -> Legal events enter the queue or directly wake the waiter
    -> Agent consumes using for await...of
    -> The AssistantMessage carried by done/error is returned by result()

The state machine remembers the phase of the entire stream, the current active block, the next content index, and the completed content, so it can reject errors such as missing start, block type mismatch, index jumps, incomplete tool parameters, inconsistent final messages, and pushing events after termination.

This step still only validates the unified event protocol and JSON data boundaries. Whether tool parameters conform to a specific TypeBox schema will be handled in the next article.

This chapter's git branch address: eventstream-validation

If you are also interested in Agent development, feel free to like, bookmark, and follow. Column: Developing a Coding Agent from Scratch - Dongfang Xiaoyue's Column - Juejin