Copilot commented on code in PR #5751: URL: https://github.com/apache/texera/pull/5751#discussion_r3487592867
########## agent-service/src/types/ws/client.ts: ########## @@ -0,0 +1,42 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +// Client -> server WebSocket frames for this service's protocol + +import type { CustomUnionType } from "../util"; + +/** + * Send a prompt to agent to start the ReAct loop + */ +export interface WsClientPromptCommand extends Readonly<{ + content: string; + messageSource?: "chat" | "feedback"; +}> {} + +/** + * Stop the agent's current ReAct loop + */ +export interface WsClientStopCommand extends Readonly<{}> {} + +export type WsClientCommandTypeMap = { + prompt: WsClientPromptCommand; + stop: WsClientStopCommand; +}; + +export type WsClientCommand = CustomUnionType<WsClientCommandTypeMap>; Review Comment: The PR description/linked issue says the client stop request should be a discriminated "command" frame (e.g. { type: "command", commandType: "stop" }), but the extracted WS types define stop as its own discriminator (type: "stop") and there is no commandType. This mismatch can confuse protocol consumers and makes it unclear what the canonical wire format is. ########## agent-service/src/types/ws/server.ts: ########## @@ -0,0 +1,94 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +// Server -> client WebSocket frames for this service's protocol + +import type { AgentState, ReActStep } from "../agent"; +import type { WorkflowContent } from "../workflow"; +import type { CustomUnionType } from "../util"; + +/** + * Wire projection of one operator's execution result, summarized for the + * client: counts and a small record sample instead of full payloads. + */ +export interface OperatorResultSummaryWs { + state: string; + inputTuples: number; + outputTuples: number; + inputPortShapes?: { portIndex: number; rows: number; columns: number }[]; + outputColumns?: number; + error?: string; + warnings?: string[]; + consoleLogCount?: number; + totalRowCount?: number; + sampleRecords?: Record<string, unknown>[]; + resultStatistics?: Record<string, string>; +} + +/** + * Full state pushed once when a client connects: the agent's current lifecycle + * state, the complete step list, and the HEAD pointer. Operator results are not + * included — they are pulled on demand via `GET /operator-results`. + */ +export interface WsServerSnapshotEvent extends Readonly<{ + state: AgentState; + steps: ReActStep[]; + headId: string; +}> {} + +/** A single ReAct step, streamed live as the agent runs. */ +export interface WsServerStepEvent extends Readonly<{ + step: ReActStep; +}> {} + +/** + * An agent lifecycle transition (e.g. GENERATING when a run starts, the resting + * state when it ends, STOPPING on stop). + */ +export interface WsServerStatusEvent extends Readonly<{ + state: AgentState; +}> {} + +/** An error surfaced to the client (agent not found, bad request, failed run). */ +export interface WsServerErrorEvent extends Readonly<{ + error: string; +}> {} + +/** + * Emitted after a checkout: HEAD moved, carrying the full step list and the + * workflow snapshot at the new head. + * + * @deprecated Redundant and unused. TODO: remove this message and related caller logics. + */ Review Comment: The headChange event is annotated as "unused", but this PR still keeps a reachable `/agents/:id/checkout` route (and tests) that broadcasts headChange. If the intent is “unused by the frontend/UI”, clarify that to avoid misleading future cleanup work. ########## agent-service/src/server.ts: ########## @@ -39,7 +39,8 @@ import type { AgentSettingsApi, ReActStep, } from "./types/agent"; -import { OperatorResultSerializationMode } from "./types/agent"; +import { AgentState, OperatorResultSerializationMode } from "./types/agent"; +import type { WsClientCommand, WsServerEvent, OperatorResultSummaryWs } from "./types/ws"; Review Comment: `OperatorResultSummaryWs` is imported from `types/ws`, but it’s used as the payload type for the HTTP `GET /agents/:id/operator-results` response. Keeping REST DTOs under a `ws/` namespace (and with a `Ws` suffix) makes the API surface harder to navigate and can lead to accidental coupling to WebSocket-only types. ########## agent-service/src/server.ws.spec.ts: ########## @@ -0,0 +1,323 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +// Exercises the /agents/:id/react WebSocket protocol end to end: the snapshot +// sent on connect, the status lifecycle frames, the stop command, the prompt +// request (with a stubbed run), and the error paths. These drive the real +// socket via app.listen + a WebSocket client, since app.handle() does not +// perform WS upgrades. + +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, test } from "bun:test"; +import { buildApp, _resetAgentStoreForTests, _getAgentForTests } from "./server"; +import { env } from "./config/env"; + +const API = env.API_PREFIX; + +let app: ReturnType<typeof buildApp>; +let port: number; +const openSockets: WebSocket[] = []; + +function mintTestToken(): string { + const header = Buffer.from(JSON.stringify({ alg: "HS256", typ: "JWT" })).toString("base64url"); + const payload = Buffer.from( + JSON.stringify({ + sub: "tester", + userId: 1, + email: "[email protected]", + role: "REGULAR", + exp: Math.floor(Date.now() / 1000) + 3600, + }) + ).toString("base64url"); + return `${header}.${payload}.test-signature`; +} + +const TOKEN = mintTestToken(); + +async function createAgent(): Promise<string> { + const res = await app.handle( + new Request(`http://localhost${API}/agents`, { + method: "POST", + headers: { "Content-Type": "application/json", Authorization: `Bearer ${TOKEN}` }, + body: JSON.stringify({ modelType: "test-model" }), + }) + ); + const body = (await res.json()) as { id: string }; + return body.id; +} + +interface Collector { + waitFor(predicate: (m: any) => boolean, timeoutMs?: number): Promise<any>; +} + +// Attaches a message listener immediately (before `open`) so no frame — not even +// the snapshot the server sends on connect — is missed, then resolves waiters +// from a buffer. +function collect(ws: WebSocket): Collector { + const buffer: any[] = []; + const waiters: { predicate: (m: any) => boolean; resolve: (m: any) => void }[] = []; + ws.addEventListener("message", ev => { + let data: any; + try { + data = JSON.parse(ev.data as string); + } catch { + return; + } + buffer.push(data); + const i = waiters.findIndex(w => w.predicate(data)); + if (i >= 0) { + waiters[i].resolve(data); + waiters.splice(i, 1); + } + }); + return { + waitFor(predicate, timeoutMs = 2000) { + const found = buffer.find(predicate); + if (found) return Promise.resolve(found); + return new Promise((resolve, reject) => { + const w = { predicate, resolve }; + waiters.push(w); + setTimeout(() => { + const idx = waiters.indexOf(w); + if (idx >= 0) { + waiters.splice(idx, 1); + reject(new Error("timed out waiting for a matching WS frame")); + } + }, timeoutMs); + }); + }, Review Comment: `waitFor()` installs a timeout but never clears it when the awaited frame arrives. That leaves stray timers running, which can slow the suite and (depending on the runner) keep the event loop alive longer than necessary. ########## agent-service/src/types/ws/server.ts: ########## @@ -0,0 +1,94 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +// Server -> client WebSocket frames for this service's protocol + +import type { AgentState, ReActStep } from "../agent"; +import type { WorkflowContent } from "../workflow"; +import type { CustomUnionType } from "../util"; + +/** + * Wire projection of one operator's execution result, summarized for the + * client: counts and a small record sample instead of full payloads. + */ +export interface OperatorResultSummaryWs { + state: string; + inputTuples: number; + outputTuples: number; + inputPortShapes?: { portIndex: number; rows: number; columns: number }[]; + outputColumns?: number; + error?: string; + warnings?: string[]; + consoleLogCount?: number; + totalRowCount?: number; + sampleRecords?: Record<string, unknown>[]; + resultStatistics?: Record<string, string>; +} + +/** + * Full state pushed once when a client connects: the agent's current lifecycle + * state, the complete step list, and the HEAD pointer. Operator results are not + * included — they are pulled on demand via `GET /operator-results`. + */ Review Comment: This comment says operator results are pulled via `GET /operator-results`, but the actual HTTP route is `GET /agents/:id/operator-results` (under the API prefix). Keeping the correct path in the protocol docs helps avoid client integration mistakes. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
