feat(web): add harness event runtime bridge

This commit is contained in:
Wenxin Duan
2026-09-07 16:58:23 +08:00
parent acc5e26b1c
commit 63ea92b770
7 changed files with 812 additions and 0 deletions
+112
View File
@@ -0,0 +1,112 @@
import {
ApiErrorSchema,
type ImageAttachment,
type MessageInput,
type ModelConnectionInput,
type RedactedModelConnection,
} from "@agent-studio/shared";
import type { ClientSkill } from "../store/reducer.js";
type RequestOptions = Omit<RequestInit, "body"> & { body?: unknown };
export class HarnessApiError extends Error {
public constructor(
public readonly code: string,
message: string,
public readonly status: number,
public readonly requestId?: string,
public readonly details: Record<string, unknown> = {},
) {
super(message);
this.name = "HarnessApiError";
}
}
export class HarnessApi {
public constructor(private readonly baseUrl = "") {}
public createSession(): Promise<{ sessionId: string; requestId: string }> {
return this.request("/api/sessions", { method: "POST" });
}
public deleteSession(sessionId: string): Promise<{ deleted: true; requestId: string }> {
return this.request(`/api/sessions/${encodeURIComponent(sessionId)}`, { method: "DELETE" });
}
public getSkills(): Promise<{ skills: ClientSkill[]; requestId: string }> {
return this.request("/api/skills");
}
public sendMessage(sessionId: string, input: MessageInput): Promise<{ runId: string; requestId: string }> {
return this.request(`/api/sessions/${encodeURIComponent(sessionId)}/messages`, { method: "POST", body: input });
}
public cancel(sessionId: string): Promise<{ cancelled: boolean; requestId: string }> {
return this.request(`/api/sessions/${encodeURIComponent(sessionId)}/cancel`, { method: "POST" });
}
public getModelConfig(sessionId: string): Promise<{ modelConfig: RedactedModelConnection; requestId: string }> {
return this.request(`/api/sessions/${encodeURIComponent(sessionId)}/model-config`);
}
public putModelConfig(sessionId: string, input: ModelConnectionInput): Promise<{ modelConfig: RedactedModelConnection; requestId: string }> {
return this.request(`/api/sessions/${encodeURIComponent(sessionId)}/model-config`, { method: "PUT", body: input });
}
public clearCredentials(sessionId: string): Promise<{ cleared: true; requestId: string }> {
return this.request(`/api/sessions/${encodeURIComponent(sessionId)}/model-config/credentials`, { method: "DELETE" });
}
public testModelConfig(sessionId: string): Promise<{ status: "ok" | "deferred"; message: string; requestId: string }> {
return this.request(`/api/sessions/${encodeURIComponent(sessionId)}/model-config/test`, { method: "POST" });
}
public uploadAttachment(sessionId: string, file: File): Promise<{ attachment: ImageAttachment; requestId: string }> {
const form = new FormData();
form.set("file", file, file.name);
return this.request(`/api/sessions/${encodeURIComponent(sessionId)}/attachments`, { method: "POST", body: form });
}
public deleteAttachment(sessionId: string, attachmentId: string): Promise<{ removed: true; requestId: string }> {
return this.request(`/api/sessions/${encodeURIComponent(sessionId)}/attachments/${encodeURIComponent(attachmentId)}`, { method: "DELETE" });
}
public approve(approvalId: string): Promise<{ approval: { approvalId: string; status: string }; requestId: string }> {
return this.request(`/api/approvals/${encodeURIComponent(approvalId)}/approve`, { method: "POST" });
}
public deny(approvalId: string): Promise<{ approval: { approvalId: string; status: string }; requestId: string }> {
return this.request(`/api/approvals/${encodeURIComponent(approvalId)}/deny`, { method: "POST" });
}
public eventsUrl(sessionId: string, afterSequence: number): string {
return `${this.baseUrl}/api/sessions/${encodeURIComponent(sessionId)}/events?afterSequence=${String(afterSequence)}`;
}
private async request<T>(path: string, options: RequestOptions = {}): Promise<T> {
const { body, ...init } = options;
const isForm = body instanceof FormData;
const headers = new Headers(init.headers);
if (!isForm) headers.set("Content-Type", "application/json");
const response = await fetch(`${this.baseUrl}${path}`, {
...init,
headers,
...(body === undefined ? {} : { body: isForm ? body : JSON.stringify(body) }),
});
const payload: unknown = await response.json().catch(() => undefined);
if (!response.ok) {
const parsed = ApiErrorSchema.safeParse(payload);
if (parsed.success) {
throw new HarnessApiError(
parsed.data.error.code,
parsed.data.error.message,
response.status,
parsed.data.error.requestId,
parsed.data.error.details,
);
}
throw new HarnessApiError("HTTP_ERROR", `Request failed with status ${String(response.status)}`, response.status);
}
return payload as T;
}
}
+62
View File
@@ -0,0 +1,62 @@
import type { ThreadMessageLike } from "@assistant-ui/react";
import type { ClientMessage, ClientTool } from "../store/reducer.js";
type JsonPrimitive = string | number | boolean | null;
type JsonValue = JsonPrimitive | JsonObject | JsonValue[];
type JsonObject = { readonly [key: string]: JsonValue };
type AssistantContentPart = Exclude<ThreadMessageLike["content"], string>[number];
export function toThreadMessageLike(message: ClientMessage, tools: ClientTool[] = []): ThreadMessageLike {
if (message.role === "user") {
return {
id: message.id,
role: "user",
content: message.text,
metadata: { custom: { attachmentIds: message.attachmentIds } },
};
}
const content: AssistantContentPart[] = [];
if (message.text.length > 0) content.push({ type: "text", text: message.text });
for (const tool of tools) {
content.push({
type: "tool-call",
toolCallId: tool.toolCallId,
toolName: tool.name,
args: toJsonObject(tool.arguments),
...(tool.result === undefined ? {} : { result: tool.result }),
...(tool.status === "failed" ? { isError: true, result: tool.error ?? "Tool failed" } : {}),
});
}
return {
id: message.id,
role: "assistant",
content,
status: toAssistantStatus(message),
metadata: { custom: message.runId === undefined ? {} : { runId: message.runId } },
};
}
function toAssistantStatus(message: ClientMessage): NonNullable<ThreadMessageLike["status"]> {
switch (message.status) {
case "running":
return { type: "running" };
case "complete":
return { type: "complete", reason: "stop" };
case "cancelled":
return { type: "incomplete", reason: "cancelled" };
case "failed":
return { type: "incomplete", reason: "error", error: message.error ?? "Run failed" };
}
}
function toJsonObject(value: unknown): JsonObject {
if (typeof value !== "object" || value === null || Array.isArray(value)) return {};
try {
return JSON.parse(JSON.stringify(value)) as JsonObject;
} catch {
return {};
}
}
@@ -0,0 +1,54 @@
import { useExternalStoreRuntime, type AppendMessage } from "@assistant-ui/react";
import { useMemo, useSyncExternalStore } from "react";
import type { MessageInput } from "@agent-studio/shared";
import { toThreadMessageLike } from "./message-mapper.js";
import type { HarnessClientState } from "../store/reducer.js";
export type HarnessRuntimeStore = {
getSnapshot(): HarnessClientState;
subscribe(subscriber: () => void): () => void;
sendMessage(input: MessageInput): Promise<unknown>;
cancel(): Promise<unknown>;
};
export function useHarnessRuntime(store: HarnessRuntimeStore) {
const snapshot = useSyncExternalStore(
(subscriber) => store.subscribe(subscriber),
() => store.getSnapshot(),
() => store.getSnapshot(),
);
const messages = useMemo(
() => snapshot.messageOrder.flatMap((id) => {
const message = snapshot.messages[id];
if (message === undefined) return [];
const tools = Object.values(snapshot.tools).filter((tool) => tool.runId === message.runId);
return [toThreadMessageLike(message, tools)];
}),
[snapshot.messageOrder, snapshot.messages, snapshot.tools],
);
const isRunning = Object.values(snapshot.runs).some((run) => run.status === "running");
return useExternalStoreRuntime({
messages,
convertMessage: (message) => message,
isRunning,
onNew: async (message) => {
await store.sendMessage(extractMessageInput(message));
},
onCancel: async () => {
await store.cancel();
},
});
}
export function extractMessageInput(message: AppendMessage): MessageInput {
const text = message.content
.filter((part) => part.type === "text")
.map((part) => part.text)
.join("\n")
.trim();
const attachmentIds = (message.attachments ?? [])
.map((attachment) => attachment.id)
.filter((id) => /^att_[a-zA-Z0-9_-]+$/.test(id));
return { text, attachmentIds: [...new Set(attachmentIds)] };
}
+152
View File
@@ -0,0 +1,152 @@
import { HarnessEventSchema, type MessageInput, type ModelConnectionInput } from "@agent-studio/shared";
import { HarnessApi } from "../api/harness-api.js";
import { createInitialHarnessState, reduceHarnessEvent, type HarnessClientState } from "./reducer.js";
type EventSourceLike = Pick<EventSource, "addEventListener" | "close">;
type EventSourceFactory = (url: string) => EventSourceLike;
const harnessEventTypes = [
"session.started", "session.ended", "user.message", "assistant.delta", "assistant.completed",
"skill.requested", "skill.loaded", "tool.requested", "tool.completed", "tool.failed",
"approval.required", "approval.resolved", "attachment.uploaded", "attachment.rejected",
"model.configured", "model.validation_failed", "run.started", "run.cancelled", "run.completed", "run.failed",
] as const;
export class HarnessClientStore {
#state: HarnessClientState = createInitialHarnessState();
readonly #subscribers = new Set<() => void>();
#eventSource: EventSourceLike | undefined;
#starting: Promise<void> | undefined;
public constructor(
public readonly api = new HarnessApi(),
private readonly eventSourceFactory: EventSourceFactory = (url) => new EventSource(url),
) {}
public readonly getSnapshot = (): HarnessClientState => this.#state;
public readonly subscribe = (subscriber: () => void): (() => void) => {
this.#subscribers.add(subscriber);
return () => this.#subscribers.delete(subscriber);
};
public start(): Promise<void> {
this.#starting ??= this.#start();
return this.#starting;
}
public async dispose(): Promise<void> {
this.#eventSource?.close();
this.#eventSource = undefined;
const sessionId = this.#state.sessionId;
if (sessionId !== undefined) await this.api.deleteSession(sessionId);
this.#setState(createInitialHarnessState());
}
public sendMessage(input: MessageInput): Promise<{ runId: string; requestId: string }> {
return this.api.sendMessage(this.#requireSession(), input);
}
public cancel(): Promise<{ cancelled: boolean; requestId: string }> {
return this.api.cancel(this.#requireSession());
}
public uploadAttachment(file: File) {
return this.api.uploadAttachment(this.#requireSession(), file);
}
public deleteAttachment(attachmentId: string) {
return this.api.deleteAttachment(this.#requireSession(), attachmentId);
}
public getModelConfig() {
return this.api.getModelConfig(this.#requireSession());
}
public putModelConfig(input: ModelConnectionInput) {
return this.api.putModelConfig(this.#requireSession(), input);
}
public clearCredentials() {
return this.api.clearCredentials(this.#requireSession());
}
public testModelConfig() {
return this.api.testModelConfig(this.#requireSession());
}
public approve(approvalId: string) {
return this.api.approve(approvalId);
}
public deny(approvalId: string) {
return this.api.deny(approvalId);
}
async #start(): Promise<void> {
const stateWithoutError = { ...this.#state };
delete stateWithoutError.error;
this.#setState({ ...stateWithoutError, connection: "connecting" });
try {
const [session, skills] = await Promise.all([this.api.createSession(), this.api.getSkills()]);
this.#setState({ ...this.#state, sessionId: session.sessionId, skills: skills.skills });
this.#connect(0);
} catch (error) {
this.#setState({
...this.#state,
connection: "error",
error: error instanceof Error ? error.message : "Unable to start session",
});
this.#starting = undefined;
throw error;
}
}
#connect(afterSequence: number): void {
const sessionId = this.#requireSession();
this.#eventSource?.close();
const source = this.eventSourceFactory(this.api.eventsUrl(sessionId, afterSequence));
this.#eventSource = source;
source.addEventListener("open", () => this.#setState({ ...this.#state, connection: "open" }));
source.addEventListener("error", () => this.#setState({ ...this.#state, connection: "reconnecting" }));
source.addEventListener("stream.reset", () => this.#resetStream());
for (const type of harnessEventTypes) {
source.addEventListener(type, (rawEvent) => {
if (!(rawEvent instanceof MessageEvent)) return;
const parsed = HarnessEventSchema.safeParse(parseJson(rawEvent.data));
if (!parsed.success || parsed.data.sessionId !== sessionId) return;
const state = reduceHarnessEvent(this.#state, parsed.data);
this.#setState(state);
if (state.resetRequired) this.#resetStream();
});
}
}
#resetStream(): void {
const { sessionId, skills } = this.#state;
this.#setState({
...createInitialHarnessState({ ...(sessionId === undefined ? {} : { sessionId }), skills }),
connection: "reconnecting",
});
this.#connect(0);
}
#setState(state: HarnessClientState): void {
this.#state = state;
for (const subscriber of this.#subscribers) subscriber();
}
#requireSession(): string {
if (this.#state.sessionId === undefined) throw new Error("Session has not started");
return this.#state.sessionId;
}
}
function parseJson(value: unknown): unknown {
if (typeof value !== "string") return undefined;
try {
return JSON.parse(value) as unknown;
} catch {
return undefined;
}
}
+280
View File
@@ -0,0 +1,280 @@
import type { HarnessEvent, ImageAttachment } from "@agent-studio/shared";
export type ConnectionStatus = "idle" | "connecting" | "open" | "reconnecting" | "error";
export type MessageStatus = "running" | "complete" | "cancelled" | "failed";
export type ClientMessage = {
id: string;
role: "user" | "assistant";
text: string;
status: MessageStatus;
runId?: string;
attachmentIds: string[];
error?: string;
};
export type ClientRun = {
runId: string;
modelId: string;
status: "running" | "complete" | "cancelled" | "failed";
messageId?: string;
error?: string;
};
export type ClientTool = {
toolCallId: string;
runId?: string;
name: string;
arguments: unknown;
status: "requested" | "completed" | "failed";
result?: unknown;
error?: string;
approvalId?: string;
};
export type ClientApproval = {
approvalId: string;
runId?: string;
toolCallId: string;
toolName: string;
arguments: unknown;
riskLevel: "low" | "medium" | "high";
expiresAt: string;
status: "pending" | "approved" | "denied" | "expired" | "cancelled";
};
export type ClientSkill = {
name: string;
description: string;
source: string;
filePath: string;
diagnostics: Array<{ level: "warning" | "error"; message: string; filePath?: string }>;
activity?: "requested" | "loaded";
};
export type HarnessClientState = {
sessionId?: string;
connection: ConnectionStatus;
lastSequence: number;
resetRequired: boolean;
messages: Record<string, ClientMessage>;
messageOrder: string[];
runs: Record<string, ClientRun>;
tools: Record<string, ClientTool>;
approvals: Record<string, ClientApproval>;
attachments: Record<string, ImageAttachment>;
skills: ClientSkill[];
timeline: HarnessEvent[];
error?: string;
};
export function createInitialHarnessState(
seed: { sessionId?: string; skills?: ClientSkill[] } | undefined = undefined,
): HarnessClientState {
return {
...(seed?.sessionId === undefined ? {} : { sessionId: seed.sessionId }),
connection: "idle",
lastSequence: 0,
resetRequired: false,
messages: {},
messageOrder: [],
runs: {},
tools: {},
approvals: {},
attachments: {},
skills: seed?.skills ?? [],
timeline: [],
};
}
export function reduceHarnessEvent(state: HarnessClientState, event: HarnessEvent): HarnessClientState {
if (event.sequence <= state.lastSequence) return state;
if (state.lastSequence > 0 && event.sequence !== state.lastSequence + 1) {
return { ...state, resetRequired: true, connection: "reconnecting" };
}
let next: HarnessClientState = {
...state,
lastSequence: event.sequence,
timeline: [...state.timeline.slice(-499), event],
};
switch (event.type) {
case "session.started":
return next;
case "session.ended":
return { ...next, connection: "idle" };
case "user.message": {
const message: ClientMessage = {
id: event.payload.messageId,
role: "user",
text: event.payload.text,
status: "complete",
attachmentIds: [...event.payload.attachmentIds],
};
return upsertMessage(next, message);
}
case "run.started": {
if (event.runId === undefined) return next;
return {
...next,
runs: {
...next.runs,
[event.runId]: { runId: event.runId, modelId: event.payload.modelId, status: "running" },
},
};
}
case "assistant.delta": {
const current = next.messages[event.payload.messageId];
const message: ClientMessage = {
id: event.payload.messageId,
role: "assistant",
text: `${current?.text ?? ""}${event.payload.delta}`,
status: "running",
attachmentIds: [],
...(event.runId === undefined ? {} : { runId: event.runId }),
};
return upsertMessage(next, message);
}
case "assistant.completed":
return updateMessage(next, event.payload.messageId, { status: "complete" });
case "run.completed": {
if (event.runId === undefined) return next;
next = updateRun(next, event.runId, { status: "complete", messageId: event.payload.messageId });
return updateMessage(next, event.payload.messageId, { status: "complete" });
}
case "run.cancelled": {
if (event.runId === undefined) return next;
next = updateRun(next, event.runId, { status: "cancelled" });
return updateMessageForRun(next, event.runId, { status: "cancelled" });
}
case "run.failed": {
if (event.runId === undefined) return next;
next = updateRun(next, event.runId, { status: "failed", error: event.payload.message });
return updateMessageForRun(next, event.runId, { status: "failed", error: event.payload.message });
}
case "tool.requested": {
const tool: ClientTool = {
toolCallId: event.payload.toolCallId,
name: event.payload.toolName,
arguments: event.payload.arguments,
status: "requested",
...(event.runId === undefined ? {} : { runId: event.runId }),
};
return { ...next, tools: { ...next.tools, [tool.toolCallId]: tool } };
}
case "tool.completed":
return updateTool(next, event.payload.toolCallId, {
status: "completed",
result: event.payload.result,
});
case "tool.failed":
return updateTool(next, event.payload.toolCallId, {
status: "failed",
error: event.payload.error,
});
case "approval.required": {
const approval: ClientApproval = {
approvalId: event.payload.approvalId,
toolCallId: event.payload.toolCallId,
toolName: event.payload.toolName,
arguments: event.payload.arguments,
riskLevel: event.payload.riskLevel,
expiresAt: event.payload.expiresAt,
status: "pending",
...(event.runId === undefined ? {} : { runId: event.runId }),
};
next = { ...next, approvals: { ...next.approvals, [approval.approvalId]: approval } };
return updateTool(next, approval.toolCallId, { approvalId: approval.approvalId });
}
case "approval.resolved": {
const current = next.approvals[event.payload.approvalId];
if (current === undefined) return next;
return {
...next,
approvals: {
...next.approvals,
[current.approvalId]: { ...current, status: event.payload.status },
},
};
}
case "attachment.uploaded": {
const attachment: ImageAttachment = {
attachmentId: event.payload.attachmentId,
sessionId: event.sessionId,
name: event.payload.name,
mediaType: event.payload.mediaType,
byteSize: event.payload.byteSize,
width: event.payload.width,
height: event.payload.height,
status: "ready",
};
return { ...next, attachments: { ...next.attachments, [attachment.attachmentId]: attachment } };
}
case "attachment.rejected":
case "model.configured":
case "model.validation_failed":
return next;
case "skill.requested":
case "skill.loaded":
return {
...next,
skills: next.skills.map((skill) =>
skill.name === event.payload.name
? { ...skill, activity: event.type === "skill.loaded" ? "loaded" : "requested" }
: skill,
),
};
}
}
function upsertMessage(state: HarnessClientState, message: ClientMessage): HarnessClientState {
const exists = state.messages[message.id] !== undefined;
return {
...state,
messages: { ...state.messages, [message.id]: message },
messageOrder: exists ? state.messageOrder : [...state.messageOrder, message.id],
};
}
function updateMessage(
state: HarnessClientState,
messageId: string,
patch: Partial<ClientMessage>,
): HarnessClientState {
const message = state.messages[messageId];
if (message === undefined) return state;
return { ...state, messages: { ...state.messages, [messageId]: { ...message, ...patch } } };
}
function updateMessageForRun(
state: HarnessClientState,
runId: string,
patch: Partial<ClientMessage>,
): HarnessClientState {
let messageId: string | undefined;
for (let index = state.messageOrder.length - 1; index >= 0; index -= 1) {
const candidate = state.messageOrder[index];
if (candidate !== undefined && state.messages[candidate]?.runId === runId) {
messageId = candidate;
break;
}
}
return messageId === undefined ? state : updateMessage(state, messageId, patch);
}
function updateRun(state: HarnessClientState, runId: string, patch: Partial<ClientRun>): HarnessClientState {
const run = state.runs[runId];
if (run === undefined) return state;
return { ...state, runs: { ...state.runs, [runId]: { ...run, ...patch } } };
}
function updateTool(
state: HarnessClientState,
toolCallId: string,
patch: Partial<ClientTool>,
): HarnessClientState {
const tool = state.tools[toolCallId];
if (tool === undefined) return state;
return { ...state, tools: { ...state.tools, [toolCallId]: { ...tool, ...patch } } };
}