Skip to content

Commit 508e72c

Browse files
bobbai00claude
andcommitted
refactor(agent-service): extract WebSocket message types into types/ws
Move the inline WS message definitions out of server.ts into a dedicated types/ws module, modeled as discriminated unions rather than one all-optional interface: - client.ts: WsClientRequest = WsClientRequestPrompt | WsClientRequestStopCommand - server.ts: WsServerMessage union (init/step/state/complete/error/headChange) + OperatorResultSummaryWs - index.ts: barrel, re-exported from the types barrel The client->server wire discriminator changes from "message"/"stop" to "prompt"/"command" (stop becomes commandType: "stop"); server->client `type` values are unchanged. server.ts parses WsClientRequest and switches on the new shapes, and the frontend WS sends are updated in lockstep. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012qFkyrpTd5PrkNBPcBeo4Q
1 parent 1c580e5 commit 508e72c

6 files changed

Lines changed: 185 additions & 44 deletions

File tree

agent-service/src/server.ts

Lines changed: 14 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ import type {
4040
ReActStep,
4141
} from "./types/agent";
4242
import { OperatorResultSerializationMode } from "./types/agent";
43+
import type { WsClientRequest, WsServerMessage, WsServerInitMessage, OperatorResultSummaryWs } from "./types/ws";
4344

4445
const agentStore = new Map<string, TexeraAgent>();
4546
let agentCounter = 0;
@@ -410,37 +411,6 @@ const agentsRouter = new Elysia({ prefix: "/agents" })
410411
}
411412
);
412413

413-
interface WsMessage {
414-
type: "message" | "stop";
415-
content?: string;
416-
messageSource?: "chat" | "feedback";
417-
}
418-
419-
interface OperatorResultSummaryWs {
420-
state: string;
421-
inputTuples: number;
422-
outputTuples: number;
423-
inputPortShapes?: { portIndex: number; rows: number; columns: number }[];
424-
outputColumns?: number;
425-
error?: string;
426-
warnings?: string[];
427-
consoleLogCount?: number;
428-
totalRowCount?: number;
429-
sampleRecords?: Record<string, any>[];
430-
resultStatistics?: Record<string, string>;
431-
}
432-
433-
interface WsOutgoingMessage {
434-
type: "step" | "state" | "error" | "complete" | "init" | "headChange";
435-
step?: ReActStep;
436-
state?: string;
437-
error?: string;
438-
steps?: ReActStep[];
439-
headId?: string;
440-
operatorResults?: Record<string, OperatorResultSummaryWs>;
441-
workflowContent?: any;
442-
}
443-
444414
function getOperatorResultSummaries(agent: TexeraAgent): Record<string, OperatorResultSummaryWs> {
445415
const resultState = agent.getWorkflowResultState();
446416
const visible = resultState.getAllVisible();
@@ -464,7 +434,7 @@ function getOperatorResultSummaries(agent: TexeraAgent): Record<string, Operator
464434
return results;
465435
}
466436

467-
function broadcastToAgent(agentId: string, message: WsOutgoingMessage): void {
437+
function broadcastToAgent(agentId: string, message: WsServerMessage): void {
468438
const agent = agentStore.get(agentId);
469439
if (!agent) return;
470440

@@ -504,7 +474,7 @@ export function buildApp() {
504474

505475
agent.addWebsocket(ws);
506476

507-
const initMessage: WsOutgoingMessage = {
477+
const initMessage: WsServerInitMessage = {
508478
type: "init",
509479
state: agent.getState(),
510480
steps: agent.getAllSteps(),
@@ -523,21 +493,23 @@ export function buildApp() {
523493
return;
524494
}
525495

526-
let msg: WsMessage;
496+
let msg: WsClientRequest;
527497
try {
528-
msg = typeof messageData === "string" ? JSON.parse(messageData) : (messageData as WsMessage);
498+
msg = typeof messageData === "string" ? JSON.parse(messageData) : (messageData as WsClientRequest);
529499
} catch {
530500
ws.send(JSON.stringify({ type: "error", error: "Invalid message format" }));
531501
return;
532502
}
533503

534-
if (msg.type === "stop") {
535-
agent.stop();
536-
broadcastToAgent(agentId, { type: "state", state: "STOPPING" });
504+
if (msg.type === "command") {
505+
if (msg.commandType === "stop") {
506+
agent.stop();
507+
broadcastToAgent(agentId, { type: "state", state: "STOPPING" });
508+
}
537509
return;
538510
}
539511

540-
if (msg.type === "message") {
512+
if (msg.type === "prompt") {
541513
if (!msg.content || typeof msg.content !== "string") {
542514
ws.send(JSON.stringify({ type: "error", error: "Message content is required" }));
543515
return;
@@ -630,9 +602,9 @@ function printStartupMessage(app: ReturnType<typeof buildApp>) {
630602
for (const route of wsRoutes) {
631603
console.log(` WS ${route.path}`);
632604
}
633-
console.log(" Send: { type: 'message', content: '...' }");
634-
console.log(" Send: { type: 'stop' }");
635-
console.log(" Recv: { type: 'step' | 'state' | 'complete' | 'error' | 'init', ... }");
605+
console.log(" Send: { type: 'prompt', content: '...' }");
606+
console.log(" Send: { type: 'command', commandType: 'stop' }");
607+
console.log(" Recv: { type: 'step' | 'state' | 'complete' | 'error' | 'init' | 'headChange', ... }");
636608
}
637609

638610
console.log("");

agent-service/src/types/index.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,3 +20,4 @@
2020
export * from "./workflow";
2121
export * from "./execution";
2222
export * from "./agent";
23+
export * from "./ws";
Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
/**
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
// Client -> server WebSocket frames for this service's protocol
21+
// (/agents/:id/react). Modeled as a discriminated union on `type` so each
22+
// request kind carries only its own fields, rather than one interface with
23+
// everything optional.
24+
25+
interface WsClientRequestBase {
26+
type: "prompt" | "command";
27+
}
28+
29+
// A user prompt to run through the agent.
30+
export interface WsClientRequestPrompt extends WsClientRequestBase {
31+
type: "prompt";
32+
content: string;
33+
messageSource?: "chat" | "feedback";
34+
}
35+
36+
// A control command. Today the only command stops the in-flight run; the
37+
// `commandType` discriminator leaves room for additional commands later.
38+
export interface WsClientRequestStopCommand extends WsClientRequestBase {
39+
type: "command";
40+
commandType: "stop";
41+
}
42+
43+
export type WsClientRequest = WsClientRequestPrompt | WsClientRequestStopCommand;
Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
/**
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
// WebSocket frames for this service's own protocol (/agents/:id/react):
21+
// inbound client requests and the outbound server messages it pushes back.
22+
23+
export * from "./client";
24+
export * from "./server";
Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
/**
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
// Server -> client WebSocket frames for this service's protocol
21+
// (/agents/:id/react). Modeled as a discriminated union on `type` so each
22+
// message kind declares exactly the fields it sends.
23+
24+
import type { ReActStep } from "../agent";
25+
import type { WorkflowContent } from "../workflow";
26+
27+
// Wire projection of an operator's execution result, summarized for the client
28+
// (counts instead of full payloads; only a sample of records).
29+
export interface OperatorResultSummaryWs {
30+
state: string;
31+
inputTuples: number;
32+
outputTuples: number;
33+
inputPortShapes?: { portIndex: number; rows: number; columns: number }[];
34+
outputColumns?: number;
35+
error?: string;
36+
warnings?: string[];
37+
consoleLogCount?: number;
38+
totalRowCount?: number;
39+
sampleRecords?: Record<string, unknown>[];
40+
resultStatistics?: Record<string, string>;
41+
}
42+
43+
type OperatorResults = Record<string, OperatorResultSummaryWs>;
44+
45+
interface WsServerMessageBase {
46+
type: "init" | "step" | "state" | "complete" | "error" | "headChange";
47+
}
48+
49+
// Sent once on connect: a snapshot of the agent's current state and steps.
50+
export interface WsServerInitMessage extends WsServerMessageBase {
51+
type: "init";
52+
state: string;
53+
steps: ReActStep[];
54+
headId: string;
55+
operatorResults: OperatorResults;
56+
}
57+
58+
// A single ReAct step streamed as the agent runs. Operator results accompany
59+
// steps that ran tools.
60+
export interface WsServerStepMessage extends WsServerMessageBase {
61+
type: "step";
62+
step: ReActStep;
63+
operatorResults?: OperatorResults;
64+
}
65+
66+
// An agent lifecycle transition (e.g. GENERATING, STOPPING).
67+
export interface WsServerStateMessage extends WsServerMessageBase {
68+
type: "state";
69+
state: string;
70+
}
71+
72+
// Terminal message for a finished run.
73+
export interface WsServerCompleteMessage extends WsServerMessageBase {
74+
type: "complete";
75+
state: string;
76+
operatorResults: OperatorResults;
77+
}
78+
79+
// An error surfaced to the client.
80+
export interface WsServerErrorMessage extends WsServerMessageBase {
81+
type: "error";
82+
error: string;
83+
}
84+
85+
// Emitted after a checkout: the head moved, carrying the full step list and the
86+
// workflow snapshot at the new head.
87+
export interface WsServerHeadChangeMessage extends WsServerMessageBase {
88+
type: "headChange";
89+
headId: string;
90+
steps: ReActStep[];
91+
workflowContent?: WorkflowContent;
92+
operatorResults: OperatorResults;
93+
}
94+
95+
export type WsServerMessage =
96+
| WsServerInitMessage
97+
| WsServerStepMessage
98+
| WsServerStateMessage
99+
| WsServerCompleteMessage
100+
| WsServerErrorMessage
101+
| WsServerHeadChangeMessage;

frontend/src/app/workspace/service/agent/agent.service.ts

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -909,7 +909,7 @@ export class AgentService {
909909
}
910910

911911
const wsMessage = {
912-
type: "message",
912+
type: "prompt",
913913
content: message,
914914
messageSource,
915915
};
@@ -967,7 +967,7 @@ export class AgentService {
967967
if (tracking?.websocket && tracking.websocket.readyState === WebSocket.OPEN) {
968968
// Send stop via WebSocket for immediate effect
969969
try {
970-
tracking.websocket.send(JSON.stringify({ type: "stop" }));
970+
tracking.websocket.send(JSON.stringify({ type: "command", commandType: "stop" }));
971971
} catch (error) {
972972
console.error("Failed to send stop command:", error);
973973
}

0 commit comments

Comments
 (0)