diff --git a/index.ts b/index.ts index 73f5340..862fe75 100644 --- a/index.ts +++ b/index.ts @@ -1020,6 +1020,7 @@ const plugin = { healthManager?.stop(); auditLogger.close(); client.destroy(); + executor.close(); // Stop task cleanup timer if (cleanupTimer) { diff --git a/package-lock.json b/package-lock.json index 4f424e1..9bfd0a5 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,7 +9,7 @@ "version": "1.5.1", "license": "MIT", "dependencies": { - "@a2a-js/sdk": "^0.3.13", + "@a2a-js/sdk": "^1.0.1", "@bufbuild/protobuf": "^2.11.0", "@grpc/grpc-js": "^1.14.3", "express": "^4.21.2", @@ -29,15 +29,16 @@ } }, "node_modules/@a2a-js/sdk": { - "version": "0.3.13", - "resolved": "https://registry.npmjs.org/@a2a-js/sdk/-/sdk-0.3.13.tgz", - "integrity": "sha512-BZr0f9JVNQs3GKOM9xINWCh6OKIJWZFPyqqVqTym5mxO2Eemc6I/0zL7zWnljHzGdaf5aZQyQN5xa6PSH62q+A==", + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/@a2a-js/sdk/-/sdk-1.0.1.tgz", + "integrity": "sha512-CJQdh3Wzwo8qIx5UUkSJ7+7BEI16PB+MXMHHNSmx8JQsQed2HlQgvx1ENOiKUfYA3PlcEvxIwv14dBblhDuPmw==", "license": "Apache-2.0", "dependencies": { + "jose": "^6.2.3", "uuid": "^11.1.0" }, "engines": { - "node": ">=18" + "node": ">=20" }, "peerDependencies": { "@bufbuild/protobuf": "^2.10.2", @@ -1224,15 +1225,6 @@ "url": "https://github.com/discordjs/discord.js?sponsor" } }, - "node_modules/@discordjs/voice/node_modules/opusscript": { - "version": "0.0.8", - "resolved": "https://registry.npmjs.org/opusscript/-/opusscript-0.0.8.tgz", - "integrity": "sha512-VSTi1aWFuCkRCVq+tx/BQ5q9fMnQ9pVZ3JU4UHKqTkf0ED3fKEPdr+gKAAl3IA2hj9rrP6iyq3hlcJq3HELtNQ==", - "dev": true, - "license": "MIT", - "optional": true, - "peer": true - }, "node_modules/@discordjs/voice/node_modules/prism-media": { "version": "1.3.5", "resolved": "https://registry.npmjs.org/prism-media/-/prism-media-1.3.5.tgz", @@ -8992,6 +8984,15 @@ "jiti": "lib/jiti-cli.mjs" } }, + "node_modules/jose": { + "version": "6.2.10", + "resolved": "https://registry.npmjs.org/jose/-/jose-6.2.10.tgz", + "integrity": "sha512-iiW7J9qRFlGxvCOIBDBDxFePQSn7ZMAnrYGhrrOo6siO/MIqwfyilLR27pkfDgUk+raLuzADS8A3S/KLBisc0g==", + "license": "MIT", + "funding": { + "url": "https://github.com/sponsors/panva" + } + }, "node_modules/json-bigint": { "version": "1.0.0", "resolved": "https://registry.npmjs.org/json-bigint/-/json-bigint-1.0.0.tgz", diff --git a/package.json b/package.json index 29f8639..958c382 100644 --- a/package.json +++ b/package.json @@ -24,7 +24,7 @@ "test": "node --import tsx --test tests/*.test.ts" }, "dependencies": { - "@a2a-js/sdk": "^0.3.13", + "@a2a-js/sdk": "^1.0.1", "@bufbuild/protobuf": "^2.11.0", "@grpc/grpc-js": "^1.14.3", "express": "^4.21.2", diff --git a/scripts/verify-e2e.mts b/scripts/verify-e2e.mts new file mode 100644 index 0000000..174bb18 --- /dev/null +++ b/scripts/verify-e2e.mts @@ -0,0 +1,97 @@ +import express from "express"; +import { TaskState, Role } from "@a2a-js/sdk"; +import type { AgentExecutionEvent, AgentExecutor, ExecutionEventBus, RequestContext } from "@a2a-js/sdk/server"; +import { DefaultRequestHandler, InMemoryTaskStore } from "@a2a-js/sdk/server"; +import { jsonRpcHandler, UserBuilder } from "@a2a-js/sdk/server/express"; +import { readFileSync } from "node:fs"; +import { A2AClient } from "../src/client.js"; + +const PORT = 18999; +const base = `http://127.0.0.1:${PORT}`; + +function makeCardV1() { + return { + name: "Verify Echo", description: "local verification", version: "1.0.0", + supportedInterfaces: [ + { url: `${base}/`, protocolBinding: "JSONRPC", tenant: "", protocolVersion: "1.0" }, + { url: `${base}/`, protocolBinding: "JSONRPC", tenant: "", protocolVersion: "0.3" }, + ], + capabilities: {}, securitySchemes: {}, securityRequirements: [], + defaultInputModes: [], defaultOutputModes: [], skills: [], signatures: [], + }; +} +const cardV03 = { protocolVersion: "0.3.0", name: "Verify Echo", url: `${base}/`, skills: [] }; + +function extractText(ctx: any): string { + for (const p of ctx.userMessage?.parts ?? []) { + if (p?.content?.$case === "text") return String(p.content.value); + if (typeof p?.text === "string") return p.text; + } + return ""; +} + +const executor: AgentExecutor = { + async execute(ctx: RequestContext, bus: ExecutionEventBus) { + const task = { + id: ctx.taskId, contextId: ctx.contextId, + status: { + state: TaskState.TASK_STATE_COMPLETED, + timestamp: new Date().toISOString(), + message: { + messageId: `m-${ctx.taskId}`, contextId: ctx.contextId, taskId: ctx.taskId, + role: Role.ROLE_AGENT, + parts: [{ content: { $case: "text", value: `pong:${extractText(ctx)}` }, metadata: undefined, filename: "", mediaType: "text/plain" }], + metadata: undefined, extensions: [], referenceTaskIds: [], + }, + }, + artifacts: [], history: [], metadata: undefined, + } as never; + bus.publish({ kind: "task", data: task } as unknown as AgentExecutionEvent); + bus.finished(); + }, + async cancelTask(_id: string, _bus: ExecutionEventBus) {}, +}; + +const handler = new DefaultRequestHandler(makeCardV1() as never, new InMemoryTaskStore(), executor); +const app = express(); +app.use(express.json()); +app.get("/card-v1", (_q, r) => r.json(makeCardV1())); +app.get("/card-v03", (_q, r) => r.json(cardV03)); +app.use(jsonRpcHandler({ requestHandler: handler, userBuilder: UserBuilder.noAuthentication, legacyCompat: { enabled: true } })); +app.listen(PORT, "127.0.0.1"); +await new Promise((r) => setTimeout(r, 400)); + +const client = new A2AClient(); +let failures = 0; +async function check(name: string, peer: any) { + try { + const res: any = await client.sendMessage(peer, { text: "ping" }, {}); + const ok = !!res.ok; + console.log(`${name}: ${ok ? "PASS" : "FAIL"}${ok ? "" : " " + JSON.stringify(res.response ?? {}).slice(0, 220)}`); + if (!ok) failures++; + } catch (e: any) { + console.log(`${name}: THREW ${e.message}`); + failures++; + } +} + +await check("A.v1-outbound-vs-real-sdk-server", { name: "v1-peer", agentCardUrl: `${base}/card-v1` }); +await check("B.legacy-v03-outbound-vs-real-sdk-server", { name: "v03-peer", agentCardUrl: `${base}/card-v03` }); + +const peers = JSON.parse(readFileSync("/home/praxis/.openclaw/a2a-peers.json", "utf8")); +const praxTok = peers.prax?.token ?? ""; +if (!praxTok) { + console.log("C.prax-live: SKIP (no token)"); +} else { + const res: any = await Promise.race([ + client.sendMessage( + { name: "praxai", agentCardUrl: "http://192.168.4.209:18800/.well-known/agent-card.json", auth: { type: "bearer", token: praxTok } } as never, + { text: "connectivity probe from Axiom migration verification (fix/95-sdk1-dual-stack)" }, + {}, + ), + new Promise((r) => setTimeout(() => r({ ok: null }), 90_000)), + ]); + console.log(`C.prax-live-real-v03-server: ${res.ok === null ? "PENDING>90s" : res.ok ? "PASS" : "FAIL"}${res.ok === false ? " " + JSON.stringify(res.response ?? {}).slice(0, 220) : ""}`); + if (res.ok === false) failures++; +} +process.exit(failures ? 1 : 0); diff --git a/src/agent-card.ts b/src/agent-card.ts index ffe195c..b3de386 100644 --- a/src/agent-card.ts +++ b/src/agent-card.ts @@ -1,25 +1,36 @@ -import type { AgentCard, AgentSkill } from "@a2a-js/sdk"; +import type { AgentCard, AgentSkill, SecurityScheme } from "@a2a-js/sdk"; import type { GatewayConfig } from "./types.js"; function toSkill(entry: string | { id?: string; name: string; description?: string }, index: number): AgentSkill { - if (typeof entry === "string") { - return { - id: `skill-${index + 1}`, - name: entry, - description: entry, - tags: [], - }; - } + const base = + typeof entry === "string" + ? { id: `skill-${index + 1}`, name: entry, description: entry } + : { + id: entry.id || `skill-${index + 1}`, + name: entry.name, + description: entry.description || entry.name, + }; return { - id: entry.id || `skill-${index + 1}`, - name: entry.name, - description: entry.description || entry.name, + id: base.id, + name: base.name, + description: base.description, tags: [], + examples: [], + inputModes: [], + outputModes: [], + securityRequirements: [], }; } +/** + * Build a v1 AgentCard. + * + * Dual-stack note (#95): the first supportedInterface is preferred by + * conforming clients. We advertise JSONRPC twice — v1.0 and v0.3 — so + * legacy peers can negotiate down without a second card fetch. + */ export function buildAgentCard(config: GatewayConfig): AgentCard { const agentCard = config.agentCard || ({} as GatewayConfig["agentCard"]); const server = config.server || { host: "0.0.0.0", port: 18800 }; @@ -28,17 +39,22 @@ export function buildAgentCard(config: GatewayConfig): AgentCard { const useDerivedGrpcUrl = agentCard.grpcProxy === true; const fallbackHost = server.host === "0.0.0.0" ? "localhost" : server.host; const fallbackUrl = `http://${fallbackHost}:${server.port}/a2a/jsonrpc`; + const jsonRpcUrl = configuredUrl || fallbackUrl; + const restUrl = `${new URL(jsonRpcUrl).origin}/a2a/rest`; const securitySchemes: AgentCard["securitySchemes"] = {}; - const security: AgentCard["security"] = []; + const securityRequirements: AgentCard["securityRequirements"] = []; const security_ = config.security || { inboundAuth: "none", token: "" }; if (security_.inboundAuth === "bearer") { - securitySchemes["bearer"] = { - type: "http", - scheme: "bearer", + const scheme: SecurityScheme = { + scheme: { + $case: "httpAuthSecurityScheme", + value: { description: "", scheme: "Bearer", bearerFormat: "" }, + }, }; - security.push({ bearer: [] }); + securitySchemes["bearer"] = scheme; + securityRequirements.push({ schemes: { bearer: { list: [] } } }); } const grpcPort = server.port + 1; @@ -50,26 +66,27 @@ export function buildAgentCard(config: GatewayConfig): AgentCard { : `${grpcHost}:${grpcPort}`; return { - protocolVersion: "0.3.0", - version: "1.0.0", name: agentCard.name || "OpenClaw A2A Gateway", description: agentCard.description || "A2A bridge for OpenClaw agents", - url: configuredUrl || fallbackUrl, - skills: (agentCard.skills || []).map((entry, index) => toSkill(entry, index)), + version: "1.0.0", + supportedInterfaces: [ + { url: jsonRpcUrl, protocolBinding: "JSONRPC", tenant: "", protocolVersion: "1.0" }, + { url: jsonRpcUrl, protocolBinding: "JSONRPC", tenant: "", protocolVersion: "0.3" }, + { url: restUrl, protocolBinding: "HTTP+JSON", tenant: "", protocolVersion: "1.0" }, + { url: grpcProxy, protocolBinding: "GRPC", tenant: "", protocolVersion: "1.0" }, + ], + provider: undefined, + documentationUrl: undefined, capabilities: { streaming: true, pushNotifications: false, - stateTransitionHistory: false, + extensions: [], }, securitySchemes, - security, - supportsAuthenticatedExtendedCard: false, + securityRequirements, defaultInputModes: ["text"], defaultOutputModes: ["text"], - additionalInterfaces: [ - { url: configuredUrl || fallbackUrl, transport: "JSONRPC" }, - { url: `${new URL(configuredUrl || fallbackUrl).origin}/a2a/rest`, transport: "HTTP+JSON" }, - { url: grpcProxy, transport: "GRPC" }, - ], + skills: (agentCard.skills || []).map((entry, index) => toSkill(entry, index)), + signatures: [], }; } diff --git a/src/client.ts b/src/client.ts index 5a7a210..51c01dc 100644 --- a/src/client.ts +++ b/src/client.ts @@ -10,7 +10,10 @@ import { type HttpHeaders, } from "@a2a-js/sdk/client"; import { GrpcTransportFactory } from "@a2a-js/sdk/client/grpc"; -import type { MessageSendParams, Message } from "@a2a-js/sdk"; +import { Role } from "@a2a-js/sdk"; +import type { Message, SendMessageRequest } from "@a2a-js/sdk"; +import { LEGACY_METHOD_MESSAGE_SEND } from "@a2a-js/sdk/compat/v0_3"; +import { toV1Parts } from "./types-v1.js"; import { ConnectionPool, type ConnectionPoolConfig } from "./connection-pool.js"; import type { OutboundSendResult, PeerConfig, RetryConfig } from "./types.js"; @@ -228,21 +231,37 @@ export class A2AClient { // ------------------------------------------------------------------ const targetAgentId = typeof (message as any)?.agentId === "string" ? String((message as any).agentId) : ""; - const outboundMessage: any = { - kind: "message", + const role = + String((message as any)?.role ?? "user").toLowerCase() === "agent" + ? Role.ROLE_AGENT + : Role.ROLE_USER; + + const outboundMessage: Message = { messageId: (message.messageId as string) || uuidv4(), - role: (message.role as Message["role"]) || "user", - parts: (message.parts as Message["parts"]) || [ - { kind: "text", text: String(message.text || message.message || "") }, - ], + contextId: typeof (message as any)?.contextId === "string" ? (message as any).contextId : "", + taskId: typeof (message as any)?.taskId === "string" ? (message as any).taskId : "", + role, + parts: toV1Parts( + (message.parts as unknown[] | undefined) ?? [ + String(message.text || message.message || ""), + ], + ), + metadata: undefined, + extensions: [], + referenceTaskIds: [], }; if (targetAgentId) { - outboundMessage.agentId = targetAgentId; + // v1 MessageFns.toJSON drops unknown fields, so route agentId through + // metadata for peer-side routing (OpenClaw extension). + outboundMessage.metadata = { ...(outboundMessage.metadata ?? {}), agentId: targetAgentId }; } - const sendParams: MessageSendParams = { + const sendParams: SendMessageRequest = { + tenant: "", message: outboundMessage, + configuration: undefined, + metadata: undefined, }; const serviceParameters: Record = {}; @@ -270,19 +289,48 @@ export class A2AClient { }); const agentCard = await resolver.resolve(baseUrl, path); - // Build transport list: main URL + additionalInterfaces - const mainTransport = agentCard.preferredTransport ?? "JSONRPC"; - const allInterfaces: TransportEndpoint[] = [ - { url: agentCard.url, transport: mainTransport }, - ]; - - if (agentCard.additionalInterfaces) { - for (const iface of agentCard.additionalInterfaces) { - // Avoid duplicate of the main endpoint - if (iface.url !== agentCard.url || iface.transport !== mainTransport) { - allInterfaces.push({ url: iface.url, transport: iface.transport }); - } - } + // Legacy peer detection (#95 dual-stack): SDK 1.x transports negotiate + // v1 only, so speak legacy JSON-RPC directly for v0.3 peers. + const cardAny = agentCard as unknown as Record; + const supportedIfaces = Array.isArray(cardAny.supportedInterfaces) + ? (cardAny.supportedInterfaces as Array>) + : []; + const cardIsLegacy = + supportedIfaces.length === 0 || + String(cardAny.protocolVersion ?? "").startsWith("0."); + if (cardIsLegacy) { + return this.sendViaLegacyJsonRpc(peer, cardAny, outboundMessage, requestOptions); + } + + // Build transport list, normalizing BOTH card generations: + // v0.3: url + preferredTransport + additionalInterfaces[] + // v1: supportedInterfaces[] ({ url, protocolBinding }) + const legacyCard = agentCard as unknown as { + url?: string; + preferredTransport?: string; + additionalInterfaces?: { url?: string; transport?: string }[]; + }; + + const allInterfaces: TransportEndpoint[] = []; + const seenEndpoints = new Set(); + const pushEndpoint = (url?: string, transport?: string) => { + if (!url || !transport) return; + const key = `${transport}:${url}`; + if (seenEndpoints.has(key)) return; + seenEndpoints.add(key); + allInterfaces.push({ url, transport }); + }; + + // v0.3 generation first (preferred entry keeps priority), then v1. + pushEndpoint(legacyCard.url, legacyCard.preferredTransport ?? "JSONRPC"); + for (const iface of legacyCard.additionalInterfaces ?? []) { + pushEndpoint(iface.url, iface.transport); + } + for (const iface of agentCard.supportedInterfaces ?? []) { + pushEndpoint(iface.url, iface.protocolBinding); + } + if (allInterfaces.length === 0) { + pushEndpoint(legacyCard.url, "JSONRPC"); } // Bio-inspired: use adaptive transport ordering when we have stats @@ -389,6 +437,101 @@ export class A2AClient { }; } + /** + * Legacy v0.3 peers: raw JSON-RPC "message/send" with a v0.3-shaped + * message. Keeps real-world 0.3 deployments working under SDK 1.x. + */ + private async sendViaLegacyJsonRpc( + peer: PeerConfig, + card: Record, + message: Message, + requestOptions: { serviceParameters?: Record }, + ): Promise { + // Use the live fetch chain (auth-wrapped, mockable) — NOT the + // module-captured nativeFetch, which predates any globalThis.fetch + // stubbing in tests/embedders. + const doFetch = this.createFetch(peer); + const endpoint = String(card.url ?? ""); + if (!endpoint) { + return { ok: false, statusCode: 500, response: { error: "Legacy Agent Card has no service url" } }; + } + + const extraMeta = (message.metadata ?? {}) as Record; + + const legacyParts = (message.parts ?? []).map((part) => { + const fileMeta: Record = { + ...(part.filename ? { name: part.filename } : {}), + ...(part.mediaType ? { mimeType: part.mediaType } : {}), + }; + const content = part.content; + if (!content) return { kind: "text", text: "" }; + switch (content.$case) { + case "text": + return { kind: "text", text: content.value }; + case "url": + return { kind: "file", file: { uri: content.value, ...fileMeta } }; + case "raw": { + const bytes = Buffer.from(content.value).toString("base64"); + return { kind: "file", file: { bytes, ...fileMeta } }; + } + case "data": + return { kind: "data", data: content.value }; + default: + return { kind: "text", text: "" }; + } + }); + + const legacyMessage: Record = { + kind: "message", + messageId: message.messageId, + role: message.role === Role.ROLE_AGENT ? "agent" : "user", + parts: legacyParts, + ...(message.contextId ? { contextId: message.contextId } : {}), + ...(message.taskId ? { taskId: message.taskId } : {}), + // Preserve the OpenClaw agentId routing extension for legacy peers, + // which expect it as a top-level message field. + ...(typeof extraMeta.agentId === "string" ? { agentId: extraMeta.agentId } : {}), + }; + + const headers: Record = { "content-type": "application/json" }; + const serviceParams = requestOptions.serviceParameters ?? {}; + if (serviceParams.authorization) headers.authorization = serviceParams.authorization; + if (serviceParams["x-api-key"]) headers["x-api-key"] = serviceParams["x-api-key"]; + + try { + const response = await doFetch(endpoint, { + method: "POST", + headers, + body: JSON.stringify({ + jsonrpc: "2.0", + id: `legacy-${Date.now()}-${Math.random().toString(16).slice(2)}`, + method: LEGACY_METHOD_MESSAGE_SEND, + params: { message: legacyMessage }, + }), + }); + + const payload = (await response.json().catch(() => undefined)) as + | { result?: unknown; error?: { code?: number; message?: string } } + | undefined; + + if (!response.ok || !payload || payload.error) { + return { + ok: false, + statusCode: response.status || 500, + response: payload?.error ?? payload ?? { error: `legacy send failed (${response.status})` }, + }; + } + + return { ok: true, statusCode: response.status, response: payload.result }; + } catch (error: unknown) { + return { + ok: false, + statusCode: 500, + response: { error: error instanceof Error ? error.message : String(error) }, + }; + } + } + /** * Send via the SDK's ClientFactory (original single-transport path). */ @@ -396,7 +539,7 @@ export class A2AClient { factory: ClientFactory, baseUrl: string, path: string, - sendParams: MessageSendParams, + sendParams: SendMessageRequest, requestOptions: { serviceParameters?: Record }, ): Promise { try { @@ -423,7 +566,7 @@ export class A2AClient { private async doSendViaTransport( peer: PeerConfig, endpoint: TransportEndpoint, - sendParams: MessageSendParams, + sendParams: SendMessageRequest, requestOptions: { serviceParameters?: Record }, ): Promise { const authFetch = this.createFetch(peer); diff --git a/src/executor.ts b/src/executor.ts index 6387ac2..3f5e96b 100644 --- a/src/executor.ts +++ b/src/executor.ts @@ -4,8 +4,16 @@ import fs from "node:fs"; import path from "node:path"; import os from "node:os"; -import type { Message, Part, Task } from "@a2a-js/sdk"; -import type { AgentExecutor, ExecutionEventBus, RequestContext } from "@a2a-js/sdk/server"; +import { Role, TaskState } from "@a2a-js/sdk"; +import type { Artifact, Message, Part, Task } from "@a2a-js/sdk"; +import type { + AgentExecutionEvent, + AgentExecutor, + ExecutionEventBus, + RequestContext, +} from "@a2a-js/sdk/server"; + +import { textPart, urlFilePart, v1Message } from "./types-v1.js"; import type { GatewayConfig, OpenClawPluginApi } from "./types.js"; import { enqueueDispatch } from "./dispatch-queue.js"; @@ -154,6 +162,29 @@ function extractTextFragments(value: unknown): string[] { return []; } + // v1 protobuf-style Part: content discriminated by $case + const v1Content = asObject(obj.content); + if (v1Content && typeof v1Content.$case === "string") { + if (v1Content.$case === "text") { + const trimmed = typeof v1Content.value === "string" ? v1Content.value.trim() : ""; + return trimmed ? [trimmed] : []; + } + if (v1Content.$case === "url") { + const name = typeof obj.filename === "string" && obj.filename ? ` (${obj.filename})` : ""; + return [`[file] ${String(v1Content.value)}${name}`]; + } + if (v1Content.$case === "raw") { + return [`[file] ${typeof obj.filename === "string" && obj.filename ? obj.filename : "(binary attachment)"}`]; + } + if (v1Content.$case === "data") { + try { + return [JSON.stringify(v1Content.value)]; + } catch { + return []; + } + } + } + if (obj.kind === "text" && typeof obj.text === "string") { const trimmed = obj.text.trim(); return trimmed ? [trimmed] : []; @@ -343,39 +374,70 @@ export function extractUrlsFromText(text: string, existingUrls: string[] = []): return result; } +/** Build a wrapped v1 task lifecycle event for the execution event bus. */ +function taskEvent( + taskId: string, + contextId: string, + history: Message[], + status: { + state: TaskState; + message?: Message | undefined; + timestamp?: string | undefined; + }, + artifacts?: Artifact[], +): AgentExecutionEvent { + const task: Task = { + id: taskId, + contextId, + status: { + state: status.state, + message: status.message ?? undefined, + timestamp: status.timestamp ?? new Date().toISOString(), + }, + history, + artifacts: artifacts ?? [], + metadata: undefined, + }; + return { kind: "task", data: task }; +} + +function agentStatusMessage(contextId: string, taskId: string, text: string): Message { + return v1Message({ + messageId: uuidv4(), + role: Role.ROLE_AGENT, + parts: [textPart(text)], + contextId, + taskId, + }); +} + /** - * Build A2A Part array from an AgentResponse. Produces TextPart for text - * content, FilePart entries for each media URL, and additional FileParts - * for file URLs discovered in the text content. + * Build A2A v1 Part array from an AgentResponse. Produces text parts for + * text content, URL file parts for each media URL, and additional URL + * file parts for file URLs discovered in the text content. */ function buildResponseParts(response: AgentResponse): Part[] { const parts: Part[] = []; if (response.text) { - parts.push({ kind: "text", text: response.text }); + parts.push(textPart(response.text)); } for (const url of response.mediaUrls) { - parts.push({ - kind: "file", - file: { uri: url }, - }); + parts.push(urlFilePart(url)); } // Extract file URLs from text that are not already in mediaUrls if (response.text) { const textUrls = extractUrlsFromText(response.text, response.mediaUrls); for (const url of textUrls) { - parts.push({ - kind: "file", - file: { uri: url }, - }); + parts.push(urlFilePart(url)); } } // Ensure at least one part exists (A2A requires non-empty parts array) if (parts.length === 0) { - parts.push({ kind: "text", text: "" }); + parts.push(textPart("")); } return parts; @@ -1216,6 +1278,7 @@ export class GatewayRpcConnectionPool { this.stopHeartbeat(entry); } }, this.heartbeatIntervalMs); + entry.heartbeatTimer.unref?.(); } private stopHeartbeat(entry: PooledConnection): void { @@ -1232,6 +1295,7 @@ export class GatewayRpcConnectionPool { this.evictConnection(entry); } }, this.idleTimeoutMs); + entry.evictionTimer.unref?.(); } private stopEviction(entry: PooledConnection): void { @@ -1299,41 +1363,29 @@ export class OpenClawAgentExecutor implements AgentExecutor { : rawHistory; // Publish initial "working" state so the task is trackable during async dispatch - const workingTask: Task = { - kind: "task", - id: taskId, - contextId, - status: { - state: "working", + eventBus.publish( + taskEvent(taskId, contextId, existingHistory, { + state: TaskState.TASK_STATE_WORKING, timestamp: new Date().toISOString(), - }, - history: existingHistory, - }; - eventBus.publish(workingTask); + }), + ); // Validate inbound FileParts before dispatching to the agent const fileValidationError = this.validateInboundFileParts(requestContext.userMessage); if (fileValidationError) { this.api.logger.warn(`a2a-gateway: inbound file validation failed: ${fileValidationError}`); - const rejectedMessage: Message = { - kind: "message", - messageId: uuidv4(), - role: "agent", - parts: [{ kind: "text", text: `File validation failed: ${fileValidationError}` }], - contextId, - }; - const rejectedTask: Task = { - kind: "task", - id: taskId, + const rejectedMessage = agentStatusMessage( contextId, - status: { - state: "failed", + taskId, + `File validation failed: ${fileValidationError}`, + ); + eventBus.publish( + taskEvent(taskId, contextId, existingHistory, { + state: TaskState.TASK_STATE_FAILED, message: rejectedMessage, timestamp: new Date().toISOString(), - }, - history: existingHistory, - }; - eventBus.publish(rejectedTask); + }), + ); eventBus.finished(); return; } @@ -1343,17 +1395,12 @@ export class OpenClawAgentExecutor implements AgentExecutor { // Emit periodic heartbeat events while the agent is working. // This keeps SSE connections alive and signals that the task is still in progress. const heartbeat = setInterval(() => { - const heartbeatTask: Task = { - kind: "task", - id: taskId, - contextId, - status: { - state: "working", + eventBus.publish( + taskEvent(taskId, contextId, existingHistory, { + state: TaskState.TASK_STATE_WORKING, timestamp: new Date().toISOString(), - }, - history: existingHistory, - }; - eventBus.publish(heartbeatTask); + }), + ); }, STREAMING_HEARTBEAT_INTERVAL_MS); try { @@ -1371,27 +1418,19 @@ export class OpenClawAgentExecutor implements AgentExecutor { await this.tryHooksWakeFallback(agentId, taskId, contextId, requestContext.userMessage); // Return failed task status so the caller knows dispatch did not succeed. - const failedMessage: Message = { - kind: "message", - messageId: uuidv4(), - role: "agent", - parts: [{ kind: "text", text: `Agent dispatch failed: ${errorMessage}` }], + const failedMessage = agentStatusMessage( contextId, - }; + taskId, + `Agent dispatch failed: ${errorMessage}`, + ); - const failedTask: Task = { - kind: "task", - id: taskId, - contextId, - status: { - state: "failed", + eventBus.publish( + taskEvent(taskId, contextId, existingHistory, { + state: TaskState.TASK_STATE_FAILED, message: failedMessage, timestamp: new Date().toISOString(), - }, - history: existingHistory, - }; - - eventBus.publish(failedTask); + }), + ); eventBus.finished(); return; } @@ -1401,33 +1440,36 @@ export class OpenClawAgentExecutor implements AgentExecutor { // Publish completed Task with artifact (text + optional file parts) const responseParts = buildResponseParts(agentResponse); - const responseMessage: Message = { - kind: "message", + const responseMessage = v1Message({ messageId: uuidv4(), - role: "agent", + role: Role.ROLE_AGENT, parts: responseParts, contextId, + taskId, + }); + + const responseArtifact: Artifact = { + artifactId: uuidv4(), + name: "", + description: "", + parts: responseParts, + metadata: undefined, + extensions: [], }; - const completedTask: Task = { - kind: "task", - id: taskId, - contextId, - status: { - state: "completed", - message: responseMessage, - timestamp: new Date().toISOString(), - }, - history: existingHistory, - artifacts: [ + eventBus.publish( + taskEvent( + taskId, + contextId, + existingHistory, { - artifactId: uuidv4(), - parts: responseParts, + state: TaskState.TASK_STATE_COMPLETED, + message: responseMessage, + timestamp: new Date().toISOString(), }, - ], - }; - - eventBus.publish(completedTask); + [responseArtifact], + ), + ); eventBus.finished(); } @@ -1445,16 +1487,12 @@ export class OpenClawAgentExecutor implements AgentExecutor { return; } - const canceledTask: Task = { - kind: "task", - id: taskId, - contextId, - status: { - state: "canceled", + eventBus.publish( + taskEvent(taskId, contextId, [], { + state: TaskState.TASK_STATE_CANCELED, timestamp: new Date().toISOString(), - }, - }; - eventBus.publish(canceledTask); + }), + ); this.taskContextByTaskId.delete(taskId); eventBus.finished(); } @@ -1614,14 +1652,32 @@ export class OpenClawAgentExecutor implements AgentExecutor { return results; } + // v1 protobuf-style Part: normalize into the legacy {kind:"file",file} + // shape so downstream validation logic stays uniform. + const v1Content = asObject(obj.content); + if (v1Content && typeof v1Content.$case === "string") { + const mimeType = typeof obj.mediaType === "string" ? obj.mediaType : undefined; + const filename = typeof obj.filename === "string" ? obj.filename : undefined; + if (v1Content.$case === "url" && typeof v1Content.value === "string") { + results.push({ + kind: "file", + file: { uri: v1Content.value, mimeType, name: filename }, + }); + return results; + } + if (v1Content.$case === "raw") { + const bytes = + Buffer.isBuffer(v1Content.value) + ? (v1Content.value as Buffer).toString("base64") + : String(v1Content.value ?? ""); + results.push({ kind: "file", file: { bytes, mimeType, name: filename } }); + return results; + } + } + const parts = Array.isArray(obj.parts) ? obj.parts : []; for (const p of parts) { - if (p && typeof p === "object") { - const part = p as Record; - if (part.kind === "file" && part.file) { - results.push(part); - } - } + results.push(...this.extractFileParts(p)); } return results; diff --git a/src/queueing-executor.ts b/src/queueing-executor.ts index 54a9cd4..e1c43de 100644 --- a/src/queueing-executor.ts +++ b/src/queueing-executor.ts @@ -1,6 +1,7 @@ import { v4 as uuidv4 } from "uuid"; -import type { Message, Task } from "@a2a-js/sdk"; +import { Role, TaskState } from "@a2a-js/sdk"; +import type { Message, Task, TaskStatus } from "@a2a-js/sdk"; import type { AgentExecutionEvent, AgentExecutor, @@ -8,6 +9,8 @@ import type { RequestContext, } from "@a2a-js/sdk/server"; +import { taskStateName, textPart, v1Message } from "./types-v1.js"; + import { GatewayTelemetry } from "./telemetry.js"; import { computeSaturationDelay, type SaturationConfig } from "./saturation-model.js"; @@ -25,34 +28,56 @@ interface QueuedTaskEntry { reject: (error: Error) => void; } -type TerminalTaskState = "completed" | "failed" | "canceled" | "rejected"; +/** Terminal task states that end a queued run (v1 enum values). */ +const TERMINAL_STATES: ReadonlySet = new Set([ + TaskState.TASK_STATE_COMPLETED, + TaskState.TASK_STATE_FAILED, + TaskState.TASK_STATE_CANCELED, + TaskState.TASK_STATE_REJECTED, +]); -function statusMessage(contextId: string, text: string): Message { - return { - kind: "message", +function statusMessage(contextId: string, taskId: string, text: string): Message { + return v1Message({ messageId: uuidv4(), - role: "agent", + role: Role.ROLE_AGENT, + parts: [textPart(text)], contextId, - parts: [{ kind: "text", text }], - }; + taskId, + }); } +/** Build a wrapped v1 task lifecycle event for the execution event bus. */ function taskEvent( taskId: string, contextId: string, - state: Task["status"]["state"], + state: TaskState, text?: string, -): Task { - return { - kind: "task", +): AgentExecutionEvent { + const task: Task = { id: taskId, contextId, status: { state, - message: text ? statusMessage(contextId, text) : undefined, + message: text ? statusMessage(contextId, taskId, text) : undefined, timestamp: new Date().toISOString(), }, + artifacts: [], + history: [], + metadata: undefined, }; + return { kind: "task", data: task }; +} + +type BusListener = ((event: AgentExecutionEvent) => void) | (() => void); + +/** Map a v1 TaskState onto the telemetry wire name for finished tasks. */ +function terminalWire( + state: TaskState, +): "completed" | "failed" | "canceled" | "rejected" { + if (state === TaskState.TASK_STATE_COMPLETED) return "completed"; + if (state === TaskState.TASK_STATE_CANCELED) return "canceled"; + if (state === TaskState.TASK_STATE_REJECTED) return "rejected"; + return "failed"; } function createObservedEventBus( @@ -64,19 +89,29 @@ function createObservedEventBus( observer(event); eventBus.publish(event); }, - on(eventName, listener) { - eventBus.on(eventName, listener); + on(eventName: "event" | "finished", listener: BusListener): ExecutionEventBus { + // Forward verbatim; the target bus owns the overload dispatch. + (eventBus.on as (name: "event", listener: unknown) => unknown)( + eventName as "event", + listener, + ); return wrapped; }, - off(eventName, listener) { - eventBus.off(eventName, listener); + off(eventName: "event" | "finished", listener: BusListener): ExecutionEventBus { + (eventBus.off as (name: "event", listener: unknown) => unknown)( + eventName as "event", + listener, + ); return wrapped; }, - once(eventName, listener) { - eventBus.once(eventName, listener); + once(eventName: "event" | "finished", listener: BusListener): ExecutionEventBus { + (eventBus.once as (name: "event", listener: unknown) => unknown)( + eventName as "event", + listener, + ); return wrapped; }, - removeAllListeners(eventName) { + removeAllListeners(eventName?: "event" | "finished"): ExecutionEventBus { eventBus.removeAllListeners(eventName); return wrapped; }, @@ -107,6 +142,12 @@ export class QueueingAgentExecutor implements AgentExecutor { }; } + /** Release resources owned by delegates that expose an explicit close hook. */ + close(): void { + const delegate = this.delegate as AgentExecutor & { close?: () => void }; + delegate.close?.(); + } + execute(requestContext: RequestContext, eventBus: ExecutionEventBus): Promise { return new Promise((resolve, reject) => { const entry: QueuedTaskEntry = { @@ -134,7 +175,7 @@ export class QueueingAgentExecutor implements AgentExecutor { taskEvent( requestContext.taskId, requestContext.contextId, - "rejected", + TaskState.TASK_STATE_REJECTED, "Gateway is overloaded; queue limit reached", ), ); @@ -154,7 +195,7 @@ export class QueueingAgentExecutor implements AgentExecutor { taskEvent( requestContext.taskId, requestContext.contextId, - "submitted", + TaskState.TASK_STATE_SUBMITTED, `Queued for execution (position ${this.queue.length})`, ), ); @@ -168,14 +209,19 @@ export class QueueingAgentExecutor implements AgentExecutor { if (entry) { this.pendingByTaskId.delete(taskId); entry.eventBus.publish( - taskEvent(taskId, entry.requestContext.contextId, "canceled", "Task canceled while queued"), + taskEvent( + taskId, + entry.requestContext.contextId, + TaskState.TASK_STATE_CANCELED, + "Task canceled while queued", + ), ); entry.eventBus.finished(); entry.resolve(); this.telemetry.recordTaskFinish( taskId, entry.requestContext.contextId, - "canceled", + terminalWire(TaskState.TASK_STATE_CANCELED), 0, this.activeTasks, this.queue.length, @@ -190,7 +236,7 @@ export class QueueingAgentExecutor implements AgentExecutor { private async runEntry(entry: QueuedTaskEntry): Promise { const { requestContext } = entry; const startedAt = Date.now(); - let finalState: TerminalTaskState | undefined; + let finalState: TaskState | undefined; let finalErrorMessage: string | undefined; this.queueDelete(requestContext.taskId); @@ -218,22 +264,19 @@ export class QueueingAgentExecutor implements AgentExecutor { ); const observedBus = createObservedEventBus(entry.eventBus, (event) => { - const status = event.kind === "task" || event.kind === "status-update" ? event.status : undefined; + const data = event.data as Partial | undefined; + const status: TaskStatus | undefined = + event.kind === "task" || event.kind === "statusUpdate" ? (data?.status ?? undefined) : undefined; if (!status) { return; } - if ( - status.state === "completed" || - status.state === "failed" || - status.state === "canceled" || - status.state === "rejected" - ) { + if (TERMINAL_STATES.has(status.state)) { finalState = status.state; - if (status.state !== "completed") { - finalErrorMessage = - typeof status.message?.parts?.[0] === "object" && status.message.parts[0]?.kind === "text" - ? status.message.parts[0].text - : finalErrorMessage; + if (status.state !== TaskState.TASK_STATE_COMPLETED) { + const firstPart = status.message?.parts?.[0]; + if (firstPart?.content?.$case === "text") { + finalErrorMessage = firstPart.content.value; + } } } }); @@ -243,7 +286,7 @@ export class QueueingAgentExecutor implements AgentExecutor { entry.resolve(); } catch (error: unknown) { const message = error instanceof Error ? error.message : String(error); - finalState = finalState || "failed"; + finalState = finalState ?? TaskState.TASK_STATE_FAILED; finalErrorMessage = finalErrorMessage || message; entry.reject(error instanceof Error ? error : new Error(message)); return; @@ -254,7 +297,7 @@ export class QueueingAgentExecutor implements AgentExecutor { this.telemetry.recordTaskFinish( requestContext.taskId, requestContext.contextId, - finalState || "failed", + terminalWire(finalState ?? TaskState.TASK_STATE_FAILED), Date.now() - startedAt, this.activeTasks, this.queue.length, diff --git a/src/task-cleanup.ts b/src/task-cleanup.ts index 4a9e494..d5b19b5 100644 --- a/src/task-cleanup.ts +++ b/src/task-cleanup.ts @@ -1,10 +1,18 @@ +import { TaskState } from "@a2a-js/sdk"; + import type { FileTaskStore } from "./task-store.js"; import type { GatewayTelemetry } from "./telemetry.js"; +import { taskStateName } from "./types-v1.js"; type LoggerLike = { info: (msg: string) => void; warn: (msg: string) => void }; -/** Terminal states that are safe to expire. */ -const TERMINAL_STATES = new Set(["completed", "failed", "canceled", "rejected"]); +/** Terminal states that are safe to expire (v1 enum values). */ +const TERMINAL_STATES = new Set([ + TaskState.TASK_STATE_COMPLETED, + TaskState.TASK_STATE_FAILED, + TaskState.TASK_STATE_CANCELED, + TaskState.TASK_STATE_REJECTED, +]); const ACTIVE_CLEANUPS = new WeakSet(); export interface CleanupResult { @@ -56,13 +64,13 @@ export async function runTaskCleanup( continue; } - const { state } = task.status; + const state = task.status?.state ?? TaskState.TASK_STATE_UNSPECIFIED; if (!TERMINAL_STATES.has(state)) { result.skipped += 1; continue; } - const timestamp = task.status.timestamp; + const timestamp = task.status?.timestamp; if (!timestamp) { // No timestamp -> can't determine age -> skip result.skipped += 1; @@ -81,7 +89,7 @@ export async function runTaskCleanup( continue; } - telemetry.recordTaskExpired(taskId, state); + telemetry.recordTaskExpired(taskId, taskStateName(state)); result.expired += 1; } catch (err: unknown) { const msg = err instanceof Error ? err.message : String(err); diff --git a/src/task-recovery.ts b/src/task-recovery.ts index 961da0d..494475d 100644 --- a/src/task-recovery.ts +++ b/src/task-recovery.ts @@ -1,11 +1,19 @@ import { randomUUID } from "node:crypto"; +import { Role, TaskState } from "@a2a-js/sdk"; + import type { FileTaskStore } from "./task-store.js"; +import { taskStateName, textPart, v1Message } from "./types-v1.js"; type LoggerLike = { info: (msg: string) => void; warn: (msg: string) => void }; /** Terminal states that should NOT be recovered — mirrors task-cleanup.ts */ -const TERMINAL_STATES = new Set(["completed", "failed", "canceled", "rejected"]); +const TERMINAL_STATES = new Set([ + TaskState.TASK_STATE_COMPLETED, + TaskState.TASK_STATE_FAILED, + TaskState.TASK_STATE_CANCELED, + TaskState.TASK_STATE_REJECTED, +]); const ACTIVE_RECOVERIES = new WeakSet(); @@ -58,25 +66,26 @@ export async function recoverStaleTasks( continue; } - const { state } = task.status; + const state = task.status?.state ?? TaskState.TASK_STATE_UNSPECIFIED; if (TERMINAL_STATES.has(state)) { result.skipped += 1; continue; } - task.status.state = "failed"; + if (!task.status) { + task.status = { state: TaskState.TASK_STATE_FAILED, message: undefined, timestamp: undefined }; + } + task.status.state = TaskState.TASK_STATE_FAILED; task.status.timestamp = new Date().toISOString(); - task.status.message = { - kind: "message", + task.status.message = v1Message({ messageId: randomUUID(), - role: "agent", + role: Role.ROLE_AGENT, parts: [ - { - kind: "text", - text: `gateway restarted before task completed (was: ${state})`, - }, + textPart(`gateway restarted before task completed (was: ${taskStateName(state)})`), ], - }; + contextId: task.contextId, + taskId: task.id, + }); await store.save(task); result.recovered += 1; diff --git a/src/task-store.ts b/src/task-store.ts index 567d815..c735aa7 100644 --- a/src/task-store.ts +++ b/src/task-store.ts @@ -1,11 +1,59 @@ import { mkdir, readdir, readFile, rename, unlink, writeFile } from "node:fs/promises"; import path from "node:path"; -import type { Task } from "@a2a-js/sdk"; +import { TaskState } from "@a2a-js/sdk"; +import type { ListTasksRequest, ListTasksResponse, Task } from "@a2a-js/sdk"; import type { ServerCallContext, TaskStore } from "@a2a-js/sdk/server"; +function taskReplacer(_key: string, value: unknown): unknown { + if (value instanceof Uint8Array) { + return { __a2aRawBytes: Buffer.from(value).toString("base64") }; + } + const candidate = value as { type?: unknown; data?: unknown } | null; + if (candidate?.type === "Buffer" && Array.isArray(candidate.data)) { + return { __a2aRawBytes: Buffer.from(candidate.data).toString("base64") }; + } + return value; +} + +function taskReviver(_key: string, value: unknown): unknown { + const candidate = value as { type?: unknown; data?: unknown } | null; + const encoded = (value as { __a2aRawBytes?: unknown } | null)?.__a2aRawBytes; + if (typeof encoded === "string") { + return Buffer.from(encoded, "base64"); + } + if ( + candidate?.type === "Buffer" && + Array.isArray(candidate.data) && + candidate.data.every((entry) => Number.isInteger(entry) && entry >= 0 && entry <= 255) + ) { + return Buffer.from(candidate.data); + } + return value; +} + function cloneTask(task: Task): Task { - return JSON.parse(JSON.stringify(task)) as Task; + return JSON.parse(JSON.stringify(task, taskReplacer), taskReviver) as Task; +} + +/** Apply ListTasks projections (history truncation, artifact stripping). */ +function projectTask(task: Task, params: ListTasksRequest): Task { + const projected = cloneTask(task); + + if (!Array.isArray(projected.history)) { + projected.history = []; + } else if (params.historyLength !== undefined) { + projected.history = + params.historyLength === 0 ? [] : projected.history.slice(-params.historyLength); + } + + if (!Array.isArray(projected.artifacts)) { + projected.artifacts = []; + } else if (!params.includeArtifacts) { + projected.artifacts = []; + } + + return projected; } function taskFileName(taskId: string): string { @@ -23,7 +71,7 @@ export class FileTaskStore implements TaskStore { async load(taskId: string, _context?: ServerCallContext): Promise { try { const payload = await readFile(this.taskPath(taskId), "utf8"); - return JSON.parse(payload) as Task; + return JSON.parse(payload, taskReviver) as Task; } catch (error: unknown) { const code = (error as { code?: string } | undefined)?.code; if (code === "ENOENT") { @@ -39,7 +87,7 @@ export class FileTaskStore implements TaskStore { const nextTask = cloneTask(task); const targetPath = this.taskPath(task.id); const tmpPath = `${targetPath}.${process.pid}.${Date.now()}.tmp`; - const payload = `${JSON.stringify(nextTask, null, 2)}\n`; + const payload = `${JSON.stringify(nextTask, taskReplacer, 2)}\n`; await writeFile(tmpPath, payload, "utf8"); @@ -79,6 +127,55 @@ export class FileTaskStore implements TaskStore { } } + /** + * v1 TaskStore.list \u2014 filesystem-backed listing with filtering and + * offset-based pagination (page token encodes the next offset). + */ + async list(params: ListTasksRequest, _context?: ServerCallContext): Promise { + const ids = await this.listAll(); + const afterTs = params.statusTimestampAfter ? Date.parse(params.statusTimestampAfter) : NaN; + const matched: Task[] = []; + + for (const id of ids) { + let task: Task | undefined; + try { + task = await this.load(id); + } catch { + continue; // unreadable/corrupt entry \u2014 skip rather than fail the whole page + } + if (!task) continue; + if (params.contextId && task.contextId !== params.contextId) continue; + if ( + params.status !== TaskState.TASK_STATE_UNSPECIFIED && + task.status?.state !== params.status + ) { + continue; + } + if (!Number.isNaN(afterTs)) { + const ts = task.status?.timestamp ? Date.parse(task.status.timestamp) : NaN; + if (Number.isNaN(ts) || ts < afterTs) continue; + } + matched.push(projectTask(task, params)); + } + + const pageSize = + params.pageSize && params.pageSize > 0 ? Math.min(Math.floor(params.pageSize), 100) : 50; + let offset = 0; + if (params.pageToken) { + const parsed = Number.parseInt(params.pageToken, 10); + if (Number.isInteger(parsed) && parsed > 0) offset = parsed; + } + + const tasks = matched.slice(offset, offset + pageSize); + const nextOffset = offset + pageSize; + return { + tasks, + nextPageToken: nextOffset < matched.length ? String(nextOffset) : "", + pageSize, + totalSize: matched.length, + }; + } + /** List all stored task IDs. */ async listAll(): Promise { try { diff --git a/src/types-v1.ts b/src/types-v1.ts new file mode 100644 index 0000000..b9999af --- /dev/null +++ b/src/types-v1.ts @@ -0,0 +1,128 @@ +/** + * Shared builders for A2A v1 (protobuf-flavored) message structures. + * + * The v1 wire types are generated from protobuf and every field is + * explicitly present (`| undefined` instead of optional `?`). These + * helpers centralize construction so call sites stay readable. + */ +import { Role, TaskState } from "@a2a-js/sdk"; +import type { Message, Part } from "@a2a-js/sdk"; + +/** Human-readable lowercase wire name for a v1 TaskState enum value. */ +export function taskStateName(state: TaskState): string { + const name = TaskState[state]; + return typeof name === "string" ? name.replace(/^TASK_STATE_/, "").toLowerCase() : "unknown"; +} + +/** Build a v1 text Part. */ +export function textPart(value: string, mediaType = "text/plain"): Part { + return { + content: { $case: "text", value }, + metadata: undefined, + filename: "", + mediaType, + }; +} + +/** Build a v1 URL-referenced file Part. */ +export function urlFilePart(url: string, filename = "", mediaType = "application/octet-stream"): Part { + return { + content: { $case: "url", value: url }, + metadata: undefined, + filename, + mediaType, + }; +} + +/** Build a v1 raw-content file Part (value is raw bytes). */ +export function rawFilePart(bytes: Buffer, filename = "", mediaType = "application/octet-stream"): Part { + return { + content: { $case: "raw", value: bytes }, + metadata: undefined, + filename, + mediaType, + }; +} + +/** Build a v1 structured-data Part. */ +export function dataPart(data: unknown): Part { + return { + content: { $case: "data", value: data }, + metadata: undefined, + filename: "", + mediaType: "application/json", + }; +} + +export interface V1MessageInit { + messageId: string; + role: Role; + parts: Part[]; + contextId?: string; + taskId?: string; + metadata?: Record; + extensions?: string[]; + referenceTaskIds?: string[]; +} + +/** Build a complete v1 Message with protobuf-required fields filled. */ +export function v1Message(init: V1MessageInit): Message { + return { + messageId: init.messageId, + contextId: init.contextId ?? "", + taskId: init.taskId ?? "", + role: init.role, + parts: init.parts, + metadata: (init.metadata as Record | undefined) ?? undefined, + extensions: init.extensions ?? [], + referenceTaskIds: init.referenceTaskIds ?? [], + }; +} + +/** + * Coerce arbitrary caller-supplied parts (v0.3 TextPart/FilePart/DataPart, + * plain strings, or already-v1 parts) into v1 Part structures. + */ +export function toV1Part(input: unknown): Part { + if (input === null || input === undefined) return textPart(""); + if (typeof input !== "object") return textPart(String(input)); + + const p = input as Record; + + // Already v1 ($case-discriminated content) + if (p.content && typeof p.content === "object" && "$case" in p.content) { + return p as Part; + } + // v0.3 / shorthand text + if (typeof p.text === "string" && p.kind !== "file") { + return textPart(p.text); + } + // v0.3 DataPart or bare data payload + if (p.kind === "data" || ("data" in p && p.data !== undefined)) { + return dataPart(p.data ?? p.value); + } + // v0.3 FilePart { file: { uri | bytes, mimeType?, name? } } + if (p.file && typeof p.file === "object") { + const f = p.file; + if (typeof f.uri === "string") return urlFilePart(f.uri, f.name ?? "", f.mimeType ?? "application/octet-stream"); + if (typeof f.bytes === "string") { + return rawFilePart(Buffer.from(f.bytes, "base64"), f.name ?? "", f.mimeType ?? "application/octet-stream"); + } + } + // v1-style flat fields without wrapper + if (typeof p.url === "string" && (p.kind === "file" || !("kind" in p))) { + return urlFilePart(p.url, p.filename ?? "", p.mediaType ?? "application/octet-stream"); + } + if (Buffer.isBuffer(p.raw)) return rawFilePart(p.raw, p.filename ?? "", p.mediaType ?? "application/octet-stream"); + if (typeof p.raw === "string" && (p.kind === "file" || p.mediaType)) { + return rawFilePart(Buffer.from(p.raw, "base64"), p.filename ?? "", p.mediaType ?? "application/octet-stream"); + } + + return textPart(JSON.stringify(p)); +} + +/** Convert an unknown parts array into v1 Parts. */ +export function toV1Parts(inputs: unknown): Part[] { + if (Array.isArray(inputs)) return inputs.map(toV1Part); + return [toV1Part(inputs)]; +} diff --git a/tests/a2a-gateway.test.ts b/tests/a2a-gateway.test.ts index a05d34c..386fcfd 100644 --- a/tests/a2a-gateway.test.ts +++ b/tests/a2a-gateway.test.ts @@ -1,6 +1,8 @@ import assert from "node:assert/strict"; import { describe, it } from "node:test"; +import { TaskState } from "@a2a-js/sdk"; + import plugin, { parseConfig } from "../index.js"; import { buildAgentCard } from "../src/agent-card.js"; import { OpenClawAgentExecutor } from "../src/executor.js"; @@ -33,7 +35,8 @@ describe("zero-config install (issue #7)", () => { }); const card = buildAgentCard(minimalConfig as unknown as GatewayConfig) as Record; assert.equal(card.name, "OpenClaw A2A Gateway", "should use default name"); - assert.equal(card.protocolVersion, "0.3.0"); + const defaultInterfaces = card.supportedInterfaces as Array>; + assert.equal(defaultInterfaces?.[0]?.protocolVersion, "1.0"); assert.equal(card.description, "A2A bridge for OpenClaw agents"); }); }); @@ -105,19 +108,21 @@ describe("session key format (PR #9, issue #8)", () => { }); describe("a2a-gateway plugin", () => { - it("builds an Agent Card with protocolVersion 0.3.0 and required fields", async () => { + it("builds a v1 Agent Card and required fields", async () => { const payload = buildAgentCard(makeConfig() as unknown as GatewayConfig) as Record; - assert.equal(payload.protocolVersion, "0.3.0"); + const interfaces = payload.supportedInterfaces as Array>; + assert.ok(Array.isArray(interfaces), "supportedInterfaces should be present"); + assert.equal(interfaces[0]?.protocolVersion, "1.0"); assert.equal(payload.name, "Test Agent"); // Verify spec-required fields assert.ok(payload.securitySchemes !== undefined, "securitySchemes should be present"); - assert.ok(payload.security !== undefined, "security should be present"); + assert.ok(payload.securityRequirements !== undefined, "securityRequirements should be present"); const capabilities = payload.capabilities as Record; assert.equal(capabilities.streaming, true); assert.equal(capabilities.pushNotifications, false); - assert.equal(capabilities.stateTransitionHistory, false); + assert.ok(Array.isArray(capabilities.extensions), "capabilities.extensions should be present"); }); it("defaults published gRPC URL to server.port + 1 when agentCard.grpcProxy is omitted", () => { @@ -134,8 +139,9 @@ describe("a2a-gateway plugin", () => { }); const payload = buildAgentCard(config) as Record; - const interfaces = payload.additionalInterfaces as Array>; - assert.equal(interfaces[2]?.url, "127.0.0.1:18801"); + const interfaces = payload.supportedInterfaces as Array>; + const grpcInterface = interfaces.find((i) => i.protocolBinding === "GRPC"); + assert.equal(grpcInterface?.url, "127.0.0.1:18801"); }); it("derives published gRPC URL from the same base URL when agentCard.grpcProxy is true", () => { @@ -153,8 +159,9 @@ describe("a2a-gateway plugin", () => { }); const payload = buildAgentCard(config) as Record; - const interfaces = payload.additionalInterfaces as Array>; - assert.equal(interfaces[2]?.url, "http://127.0.0.1:18800"); + const interfaces = payload.supportedInterfaces as Array>; + const grpcInterface = interfaces.find((i) => i.protocolBinding === "GRPC"); + assert.equal(grpcInterface?.url, "http://127.0.0.1:18800"); }); it("dispatches inbound messages via gateway RPC", async () => { @@ -195,11 +202,13 @@ describe("a2a-gateway plugin", () => { assert.equal(true, true); assert.equal(finishedCalled, true); - const finalTask = published[published.length - 1] as Record; + const lastEvent = published[published.length - 1] as { kind?: string; data?: Record }; + const finalTask = (lastEvent.kind === "task" ? lastEvent.data : lastEvent) as Record; const status = finalTask.status as Record; const message = status.message as Record; - const parts = message.parts as Array>; - assert.equal(parts[0].text, "Gateway response"); + const parts = message.parts as Array>; + assert.equal(parts[0]?.content?.$case, "text"); + assert.equal(parts[0]?.content?.value, "Gateway response"); } finally { (globalThis as any).WebSocket = originalWebSocket; } @@ -282,11 +291,12 @@ describe("a2a-gateway plugin", () => { assert.equal(finishedCalled, true); - const finalTask = published[published.length - 1] as Record; - assert.equal(finalTask.kind, "task"); + const lastEvent = published[published.length - 1] as { kind?: string; data?: Record }; + assert.equal(lastEvent.kind, "task"); + const finalTask = (lastEvent.data ?? lastEvent) as Record; const status = finalTask.status as Record; // When WebSocket is unavailable, executor publishes a failed state - assert.equal(status.state, "failed"); + assert.equal(status.state, TaskState.TASK_STATE_FAILED); } finally { (globalThis as any).WebSocket = originalWebSocket; } @@ -312,8 +322,10 @@ describe("a2a-gateway plugin", () => { assert.equal(finishedCalled, true); assert.equal(published.length, 1); - assert.equal(published[0].id, "task-1"); - assert.equal(published[0].contextId, "ctx-1"); + const canceledEvent = published[0] as { data?: Record }; + const canceledTask = (canceledEvent.data ?? canceledEvent) as Record; + assert.equal(canceledTask.id, "task-1"); + assert.equal(canceledTask.contextId, "ctx-1"); }); it("inbound FilePart (URI) is formatted as text for the agent", async () => { @@ -634,28 +646,28 @@ describe("a2a-gateway plugin", () => { } as any, ); - const finalTask = published[published.length - 1] as Record; + const lastEvent = published[published.length - 1] as { kind?: string; data?: Record }; + const finalTask = (lastEvent.data ?? lastEvent) as Record; const status = finalTask.status as Record; - assert.equal(status.state, "completed"); + assert.equal(status.state, TaskState.TASK_STATE_COMPLETED); const message = status.message as Record; - const parts = message.parts as Array>; + const parts = message.parts as Array>; - // Should have both TextPart and FilePart - const textParts = parts.filter((p) => p.kind === "text"); - const fileParts = parts.filter((p) => p.kind === "file"); + // Should have both text and URL-file parts + const textParts = parts.filter((p) => p.content?.$case === "text"); + const fileParts = parts.filter((p) => p.content?.$case === "url"); assert.ok(textParts.length >= 1, "should have at least one text part"); assert.equal(fileParts.length, 1, "should have exactly one file part"); - const filePart = fileParts[0] as { kind: string; file: { uri: string } }; - assert.equal(filePart.file.uri, "https://example.com/chart.png"); + assert.equal(fileParts[0]?.content?.value, "https://example.com/chart.png"); // Artifacts should also contain the file part - const artifacts = finalTask.artifacts as Array<{ parts: Array> }>; + const artifacts = finalTask.artifacts as Array<{ parts: Array> }>; assert.ok(artifacts.length >= 1, "should have at least one artifact"); - const artifactFileParts = artifacts[0].parts.filter((p) => p.kind === "file"); + const artifactFileParts = artifacts[0].parts.filter((p) => p.content?.$case === "url"); assert.equal(artifactFileParts.length, 1, "artifact should have one file part"); } finally { (globalThis as any).WebSocket = originalWebSocket; @@ -700,11 +712,12 @@ describe("a2a-gateway plugin", () => { } as any, ); - const finalTask = published[published.length - 1] as Record; + const lastEvent = published[published.length - 1] as { kind?: string; data?: Record }; + const finalTask = (lastEvent.data ?? lastEvent) as Record; const message = (finalTask.status as Record).message as Record; - const parts = message.parts as Array>; + const parts = message.parts as Array>; - const fileParts = parts.filter((p) => p.kind === "file"); + const fileParts = parts.filter((p) => p.content?.$case === "url"); assert.equal(fileParts.length, 2, "should have two file parts for two media URLs"); } finally { (globalThis as any).WebSocket = originalWebSocket; @@ -741,15 +754,16 @@ describe("a2a-gateway plugin", () => { } as any, ); - const finalTask = published[published.length - 1] as Record; + const lastEvent = published[published.length - 1] as { kind?: string; data?: Record }; + const finalTask = (lastEvent.data ?? lastEvent) as Record; const message = (finalTask.status as Record).message as Record; - const parts = message.parts as Array>; + const parts = message.parts as Array>; assert.equal(parts.length, 1, "should have exactly one part"); - assert.equal(parts[0].kind, "text"); - assert.equal(parts[0].text, "Just text, no media"); + assert.equal(parts[0].content?.$case, "text"); + assert.equal(parts[0].content?.value, "Just text, no media"); - const fileParts = parts.filter((p) => p.kind === "file"); + const fileParts = parts.filter((p) => p.content?.$case === "url" || p.content?.$case === "raw"); assert.equal(fileParts.length, 0, "should have no file parts"); } finally { (globalThis as any).WebSocket = originalWebSocket; @@ -766,11 +780,23 @@ describe("a2a-gateway plugin", () => { if (url === "http://mock-peer/.well-known/agent-card.json" || url === "http://mock-peer/.well-known/agent.json") { return new Response( JSON.stringify({ - protocolVersion: "0.3.0", name: "Peer Agent", - // Per A2A spec, the Agent Card `url` field is the service endpoint. - url: "http://mock-peer/a2a/jsonrpc", + version: "1.0.0", + supportedInterfaces: [ + { + url: "http://mock-peer/a2a/jsonrpc", + protocolBinding: "JSONRPC", + tenant: "", + protocolVersion: "1.0", + }, + ], + capabilities: {}, + securitySchemes: {}, + securityRequirements: [], + defaultInputModes: [], + defaultOutputModes: [], skills: [], + signatures: [], }), { status: 200, @@ -789,7 +815,16 @@ describe("a2a-gateway plugin", () => { jsonrpc: "2.0", id: payload.id, result: { - accepted: true, + task: { + id: `task-${payload.id ?? "0"}`, + contextId: "ctx-mock-peer", + status: { + state: "TASK_STATE_COMPLETED", + timestamp: new Date().toISOString(), + }, + artifacts: [], + history: [], + }, }, }), { @@ -824,7 +859,8 @@ describe("a2a-gateway plugin", () => { assert.equal(result.ok, true); assert.equal(received.length, 1); - assert.equal(received[0].method, "message/send"); + // SDK 1.x JSON-RPC transport uses the v1 method name + assert.equal(received[0].method, "SendMessage"); const params = received[0].params as Record; assert.equal(typeof params, "object"); @@ -832,7 +868,7 @@ describe("a2a-gateway plugin", () => { const msg = (params as any)?.message as Record; assert.equal(typeof msg, "object"); // OpenClaw extension: agentId should be forwarded for peer-side routing. - assert.equal(msg.agentId, "peer-agent"); + assert.equal((msg.metadata as Record | undefined)?.agentId, "peer-agent"); } finally { globalThis.fetch = originalFetch; } @@ -848,10 +884,23 @@ describe("a2a-gateway plugin", () => { if (url === "http://mock-peer/.well-known/agent-card.json" || url === "http://mock-peer/.well-known/agent.json") { return new Response( JSON.stringify({ - protocolVersion: "0.3.0", name: "Peer Agent", - url: "http://mock-peer/a2a/jsonrpc", + version: "1.0.0", + supportedInterfaces: [ + { + url: "http://mock-peer/a2a/jsonrpc", + protocolBinding: "JSONRPC", + tenant: "", + protocolVersion: "1.0", + }, + ], + capabilities: {}, + securitySchemes: {}, + securityRequirements: [], + defaultInputModes: [], + defaultOutputModes: [], skills: [], + signatures: [], }), { status: 200, headers: { "content-type": "application/json" } } ); @@ -866,7 +915,18 @@ describe("a2a-gateway plugin", () => { JSON.stringify({ jsonrpc: "2.0", id: payload.id, - result: { accepted: true }, + result: { + task: { + id: `task-${payload.id ?? "0"}`, + contextId: "ctx-mock-peer", + status: { + state: "TASK_STATE_COMPLETED", + timestamp: new Date().toISOString(), + }, + artifacts: [], + history: [], + }, + }, }), { status: 200, headers: { "content-type": "application/json" } } ); @@ -906,16 +966,28 @@ describe("a2a-gateway plugin", () => { const msg = (params as any)?.message as Record; // Verify agentId is forwarded - assert.equal(msg.agentId, "coder", "agentId should be forwarded to peer"); + assert.equal( + (msg.metadata as Record | undefined)?.agentId, + "coder", + "agentId should be forwarded to peer" + ); - // Verify FilePart structure - const parts = msg.parts as Array>; - const fileParts = parts.filter((p) => p.kind === "file"); + // Verify file part structure. On the wire, v1 Part.toJSON flattens the + // $case union ({url}), while in-process objects keep {content:{$case}}. + const parts = msg.parts as Array>; + const fileParts = parts.filter( + (p) => p.content?.$case === "url" || typeof p.url === "string", + ); assert.equal(fileParts.length, 1, "should have one file part"); - const fp = fileParts[0] as { kind: string; file: { uri: string; name: string; mimeType: string } }; - assert.equal(fp.file.uri, "https://example.com/report.pdf"); - assert.equal(fp.file.name, "report.pdf"); - assert.equal(fp.file.mimeType, "application/pdf"); + const fp = fileParts[0] as { + content?: { value?: string }; + url?: string; + filename?: string; + mediaType?: string; + }; + assert.equal(fp.content?.value ?? fp.url, "https://example.com/report.pdf"); + assert.equal(fp.filename, "report.pdf"); + assert.equal(fp.mediaType, "application/pdf"); } finally { globalThis.fetch = originalFetch; } @@ -1007,10 +1079,19 @@ describe("scope verification on connect (issue #54)", () => { // The executor catches dispatch errors and publishes a "failed" task. // Find the failed task event and verify the error message is actionable. - const failedTask = published.find((e: any) => e.status?.state === "failed") as any; + const failedTask = published.find((e: any) => { + const task = e?.kind === "task" ? e.data : e; + return task?.status?.state === TaskState.TASK_STATE_FAILED || + task?.status?.state === "failed"; + }) as any; assert.ok(failedTask, "should publish a failed task when scopes are downgraded"); - const failedText = failedTask.status?.message?.parts?.[0]?.text ?? ""; + const failedPayload = failedTask.kind === "task" ? failedTask.data : failedTask; + const firstPart = failedPayload.status?.message?.parts?.[0]; + const failedText = + firstPart?.content?.$case === "text" + ? String(firstPart.content.value) + : (firstPart?.text ?? ""); assert.ok( failedText.includes("operator.write") || failedText.includes("scope"), `failed task message should mention scope issue but got: ${failedText}`, diff --git a/tests/compat-matrix.test.ts b/tests/compat-matrix.test.ts index 15c08ee..1fe1aa9 100644 --- a/tests/compat-matrix.test.ts +++ b/tests/compat-matrix.test.ts @@ -10,6 +10,7 @@ */ import assert from "node:assert/strict"; import { describe, it } from "node:test"; +import { TaskState } from "@a2a-js/sdk"; import { buildAgentCard } from "../src/agent-card.js"; import { OpenClawAgentExecutor } from "../src/executor.js"; @@ -302,9 +303,9 @@ describe("compat: Agent Card parsing variations", () => { it("buildAgentCard produces valid card regardless of config completeness", () => { // Empty config — all defaults const defaultCard = buildAgentCard({} as unknown as GatewayConfig) as Record; - assert.equal(defaultCard.protocolVersion, "0.3.0"); + assert.equal((defaultCard.supportedInterfaces as Array>)[0]?.protocolVersion, "1.0"); assert.ok(defaultCard.name, "should have a name"); - assert.ok(defaultCard.url, "should have a url"); + assert.ok(Array.isArray(defaultCard.supportedInterfaces), "should have supported interfaces"); assert.ok(Array.isArray(defaultCard.skills), "should have skills array"); // Partial config — only name provided @@ -312,7 +313,7 @@ describe("compat: Agent Card parsing variations", () => { makeConfig({ agentCard: { name: "PartialBot" } }) as unknown as GatewayConfig, ) as Record; assert.equal(partialCard.name, "PartialBot"); - assert.equal(partialCard.protocolVersion, "0.3.0"); + assert.equal((partialCard.supportedInterfaces as Array>)[0]?.protocolVersion, "1.0"); // Full config const fullCard = buildAgentCard( @@ -748,14 +749,14 @@ describe("compat: response format variations", () => { } as any, ); - const finalTask = published[published.length - 1] as Record; + const finalTask = (published[published.length - 1] as any).data as Record; const status = finalTask.status as Record; - assert.equal(status.state, "completed"); + assert.equal(status.state, TaskState.TASK_STATE_COMPLETED); const msg = status.message as Record; const parts = msg.parts as Array>; assert.ok(parts.length >= 1, "should have at least one part"); - assert.equal(parts[0].kind, "text"); + assert.equal((parts[0] as any).content?.$case, "text"); } finally { (globalThis as any).WebSocket = originalWebSocket; } @@ -797,15 +798,15 @@ describe("compat: response format variations", () => { } as any, ); - const finalTask = published[published.length - 1] as Record; + const finalTask = (published[published.length - 1] as any).data as Record; const status = finalTask.status as Record; - assert.equal(status.state, "completed"); + assert.equal(status.state, TaskState.TASK_STATE_COMPLETED); const msg = status.message as Record; const parts = msg.parts as Array>; - const textParts = parts.filter((p) => p.kind === "text"); - const fileParts = parts.filter((p) => p.kind === "file"); + const textParts = parts.filter((p: any) => p.content?.$case === "text"); + const fileParts = parts.filter((p: any) => p.content?.$case === "url" || p.content?.$case === "raw"); assert.ok(textParts.length >= 1, "should have text part"); assert.equal(fileParts.length, 2, "should have two file parts (deduped from mediaUrl + mediaUrls)"); } finally { @@ -983,7 +984,7 @@ describe("compat: transport header variations", () => { body: JSON.stringify({ jsonrpc: "2.0", id: "test-1", - method: "message/send", + method: "SendMessage", params: { message: { messageId: "m1", role: "user", parts: [{ kind: "text", text: "no auth" }] }, }, @@ -1004,9 +1005,9 @@ describe("compat: transport header variations", () => { body: JSON.stringify({ jsonrpc: "2.0", id: "test-2", - method: "message/send", + method: "SendMessage", params: { - message: { messageId: "m2", role: "user", parts: [{ kind: "text", text: "with auth" }] }, + message: { messageId: "m2", role: "ROLE_USER", parts: [{ text: "with auth" }] }, }, }), signal: AbortSignal.timeout(30_000), @@ -1014,7 +1015,7 @@ describe("compat: transport header variations", () => { assert.equal(authResp.status, 200, "should accept request with correct auth"); const authBody = await authResp.json() as Record; - assert.ok(!authBody.error, "should not have error with correct auth"); + assert.ok(!authBody.error, `should not have error with correct auth: ${JSON.stringify(authBody.error)}`); await service!.stop({} as any); } finally { diff --git a/tests/integration-filepart.test.ts b/tests/integration-filepart.test.ts index d0166d9..6df7a60 100644 --- a/tests/integration-filepart.test.ts +++ b/tests/integration-filepart.test.ts @@ -59,20 +59,17 @@ describe("integration: FilePart end-to-end", () => { const jsonRpcRequest = { jsonrpc: "2.0", id: uuidv4(), - method: "message/send", + method: "SendMessage", params: { message: { messageId: uuidv4(), - role: "user", + role: "ROLE_USER", parts: [ - { kind: "text", text: "Analyze this image" }, + { text: "Analyze this image" }, { - kind: "file", - file: { - uri: "https://user-uploads.example.com/photo.jpg", - mimeType: "image/jpeg", - name: "photo.jpg", - }, + url: "https://user-uploads.example.com/photo.jpg", + mediaType: "image/jpeg", + filename: "photo.jpg", }, ], }, @@ -93,32 +90,30 @@ describe("integration: FilePart end-to-end", () => { // The response should be a JSON-RPC result (not error) assert.ok(!body.error, `unexpected error: ${JSON.stringify(body.error)}`); - const result = body.result as Record; - assert.ok(result, "result must exist"); + const result = ((body.result as any)?.task ?? body.result) as Record; + assert.ok(result, `result must exist: ${JSON.stringify(body)}`); // Result should be a Task (kind: "task") with completed status - assert.equal(result.kind, "task"); const status = result.status as Record; - assert.equal(status.state, "completed"); + assert.equal(status.state, "TASK_STATE_COMPLETED"); // Check the message parts in the response const message = status.message as Record; const parts = message.parts as Array>; - const textParts = parts.filter((p) => p.kind === "text"); - const fileParts = parts.filter((p) => p.kind === "file"); + const textParts = parts.filter((p) => typeof p.text === "string"); + const fileParts = parts.filter((p) => typeof p.url === "string" || typeof p.raw === "string"); assert.ok(textParts.length >= 1, "response should have text part"); assert.equal((textParts[0] as any).text, "Here is the generated image"); assert.equal(fileParts.length, 1, "response should have exactly one file part"); - const filePartFile = (fileParts[0] as any).file as { uri: string }; - assert.equal(filePartFile.uri, "https://cdn.example.com/generated-chart.png"); + assert.equal((fileParts[0] as any).url, "https://cdn.example.com/generated-chart.png"); // Also verify artifacts contain FilePart const artifacts = result.artifacts as Array<{ parts: Array> }>; assert.ok(artifacts && artifacts.length >= 1, "should have artifacts"); - const artifactFileParts = artifacts[0].parts.filter((p) => p.kind === "file"); + const artifactFileParts = artifacts[0].parts.filter((p) => typeof p.url === "string" || typeof p.raw === "string"); assert.equal(artifactFileParts.length, 1, "artifact should contain file part"); await service!.stop({} as any); @@ -151,16 +146,17 @@ describe("integration: FilePart end-to-end", () => { const jsonRpcRequest = { jsonrpc: "2.0", id: uuidv4(), - method: "message/send", + method: "SendMessage", params: { message: { messageId: uuidv4(), - role: "user", + role: "ROLE_USER", parts: [ - { kind: "text", text: "Read this" }, + { text: "Read this" }, { - kind: "file", - file: { uri: "file:///etc/passwd", mimeType: "text/plain", name: "passwd" }, + url: "file:///etc/passwd", + mediaType: "text/plain", + filename: "passwd", }, ], }, @@ -176,10 +172,10 @@ describe("integration: FilePart end-to-end", () => { assert.equal(response.status, 200); const body = await response.json() as Record; - const result = body.result as Record; + const result = ((body.result as any)?.task ?? body.result) as Record; assert.ok(result, "result must exist"); const status = result.status as Record; - assert.equal(status.state, "failed", "task should fail for file:// URI"); + assert.equal(status.state, "TASK_STATE_FAILED", "task should fail for file:// URI"); await service!.stop({} as any); } finally { @@ -211,19 +207,16 @@ describe("integration: FilePart end-to-end", () => { const jsonRpcRequest = { jsonrpc: "2.0", id: uuidv4(), - method: "message/send", + method: "SendMessage", params: { message: { messageId: uuidv4(), - role: "user", + role: "ROLE_USER", parts: [ { - kind: "file", - file: { - uri: "https://cdn.example.com/malware.exe", - mimeType: "application/x-executable", - name: "malware.exe", - }, + url: "https://cdn.example.com/malware.exe", + mediaType: "application/x-executable", + filename: "malware.exe", }, ], }, @@ -239,10 +232,10 @@ describe("integration: FilePart end-to-end", () => { assert.equal(response.status, 200); const body = await response.json() as Record; - const result = body.result as Record; + const result = ((body.result as any)?.task ?? body.result) as Record; assert.ok(result, "result must exist"); const status = result.status as Record; - assert.equal(status.state, "failed", "task should fail for disallowed MIME type"); + assert.equal(status.state, "TASK_STATE_FAILED", "task should fail for disallowed MIME type"); await service!.stop({} as any); } finally { @@ -281,19 +274,16 @@ describe("integration: FilePart end-to-end", () => { const jsonRpcRequest = { jsonrpc: "2.0", id: uuidv4(), - method: "message/send", + method: "SendMessage", params: { message: { messageId: uuidv4(), - role: "user", + role: "ROLE_USER", parts: [ { - kind: "file", - file: { - bytes: largeBase64, - mimeType: "image/png", - name: "large.png", - }, + raw: largeBase64, + mediaType: "image/png", + filename: "large.png", }, ], }, @@ -309,10 +299,10 @@ describe("integration: FilePart end-to-end", () => { assert.equal(response.status, 200); const body = await response.json() as Record; - const result = body.result as Record; - assert.ok(result, "result must exist"); + const result = ((body.result as any)?.task ?? body.result) as Record; + assert.ok(result, `result must exist: ${JSON.stringify(body)}`); const status = result.status as Record; - assert.equal(status.state, "failed", "task should fail for oversized inline file"); + assert.equal(status.state, "TASK_STATE_FAILED", "task should fail for oversized inline file"); await service!.stop({} as any); } finally { diff --git a/tests/multi-round.test.ts b/tests/multi-round.test.ts index a524818..c972633 100644 --- a/tests/multi-round.test.ts +++ b/tests/multi-round.test.ts @@ -4,7 +4,8 @@ import os from "node:os"; import path from "node:path"; import { describe, it } from "node:test"; -import type { Task } from "@a2a-js/sdk"; +import { TaskState } from "@a2a-js/sdk"; +import type { AgentExecutionEvent, Task } from "@a2a-js/sdk"; import { OpenClawAgentExecutor } from "../src/executor.js"; import { FileTaskStore } from "../src/task-store.js"; @@ -36,7 +37,14 @@ async function executeRound(executor: OpenClawAgentExecutor, taskId: string, con ); assert.equal(eventBus.isFinished(), true); - assert.equal((eventBus.events.at(-1) as Task).status.state, "completed"); + const finalTask = unwrapTask(eventBus.events.at(-1)); + assert.equal(finalTask?.status?.state, TaskState.TASK_STATE_COMPLETED); +} + +/** Unwrap a v1 execution event into its task payload. */ +function unwrapTask(event: unknown): Task | undefined { + const ev = event as AgentExecutionEvent; + return ev?.kind === "task" ? ev.data : (ev as unknown as Task | undefined); } describe("multi-round conversation routing", () => { @@ -171,15 +179,15 @@ describe("history preservation across rounds", () => { assert.equal(eventBus.isFinished(), true); // The completed task must carry the previous history - const completedTask = eventBus.events.at(-1) as Task; - assert.equal(completedTask.status.state, "completed"); + const completedTask = unwrapTask(eventBus.events.at(-1))!; + assert.equal(completedTask.status?.state, TaskState.TASK_STATE_COMPLETED); assert.ok(completedTask.history, "completed task should have history"); assert.equal(completedTask.history!.length, 2, "should carry 2 previous messages"); assert.equal((completedTask.history![0] as any).messageId, "msg-round-1-user"); assert.equal((completedTask.history![1] as any).messageId, "msg-round-1-agent"); // The working event should also carry history - const workingTask = eventBus.events[0] as Task; + const workingTask = unwrapTask(eventBus.events[0])!; assert.ok(workingTask.history, "working task should have history"); assert.equal(workingTask.history!.length, 2); } finally { @@ -226,8 +234,8 @@ describe("history preservation across rounds", () => { eventBus.bus, ); - const completedTask = eventBus.events.at(-1) as Task; - assert.equal(completedTask.status.state, "completed"); + const completedTask = unwrapTask(eventBus.events.at(-1))!; + assert.equal(completedTask.status?.state, TaskState.TASK_STATE_COMPLETED); assert.ok(completedTask.history, "should have history"); assert.equal(completedTask.history!.length, 200, "should cap at 200 messages"); // Should keep the LATEST 200 (indices 50-249) @@ -261,8 +269,8 @@ describe("history preservation across rounds", () => { eventBus.bus, ); - const completedTask = eventBus.events.at(-1) as Task; - assert.equal(completedTask.status.state, "completed"); + const completedTask = unwrapTask(eventBus.events.at(-1))!; + assert.equal(completedTask.status?.state, TaskState.TASK_STATE_COMPLETED); assert.ok(Array.isArray(completedTask.history), "history should be an array"); assert.equal(completedTask.history!.length, 0, "first round should have empty history"); } finally { diff --git a/tests/p0-runtime.test.ts b/tests/p0-runtime.test.ts index 9bf50c0..4fc05bb 100644 --- a/tests/p0-runtime.test.ts +++ b/tests/p0-runtime.test.ts @@ -4,9 +4,11 @@ import os from "node:os"; import path from "node:path"; import { describe, it } from "node:test"; -import type { Task } from "@a2a-js/sdk"; +import { TaskState } from "@a2a-js/sdk"; +import type { AgentExecutionEvent, Task } from "@a2a-js/sdk"; import type { AgentExecutor, ExecutionEventBus } from "@a2a-js/sdk/server"; +import { textPart } from "../src/types-v1.js"; import { QueueingAgentExecutor } from "../src/queueing-executor.js"; import { FileTaskStore } from "../src/task-store.js"; import { GatewayTelemetry } from "../src/telemetry.js"; @@ -55,22 +57,34 @@ function createEventBus() { function makeTask(taskId: string): Task { return { - kind: "task", id: taskId, contextId: `ctx-${taskId}`, status: { - state: "completed", + state: TaskState.TASK_STATE_COMPLETED, timestamp: new Date().toISOString(), + message: undefined, }, artifacts: [ { artifactId: `artifact-${taskId}`, - parts: [{ kind: "text", text: `done-${taskId}` }], + name: "", + description: "", + parts: [textPart(`done-${taskId}`)], + metadata: undefined, + extensions: [], }, ], + history: [], + metadata: undefined, }; } +/** Unwrap a v1 execution event into its task payload (or undefined). */ +function taskPayload(event: unknown): Task | undefined { + const ev = event as AgentExecutionEvent; + return ev?.kind === "task" ? ev.data : undefined; +} + function makeRequestContext(taskId: string) { return { taskId, @@ -97,7 +111,7 @@ describe("P0 runtime components", () => { assert.ok(restored, "task should be restored from disk"); assert.equal(restored.id, "task-1"); - assert.equal(restored.artifacts?.[0]?.parts?.[0]?.kind, "text"); + assert.equal(restored.artifacts?.[0]?.parts?.[0]?.content?.$case, "text"); } finally { await rm(tasksDir, { recursive: true, force: true }); } @@ -113,7 +127,7 @@ describe("P0 runtime components", () => { const delegate: AgentExecutor = { async execute(requestContext, eventBus) { await gates.get(requestContext.taskId)?.promise; - eventBus.publish(makeTask(requestContext.taskId)); + eventBus.publish({ kind: "task", data: makeTask(requestContext.taskId) }); eventBus.finished(); }, async cancelTask(_taskId, eventBus) { @@ -136,8 +150,8 @@ describe("P0 runtime components", () => { await Promise.resolve(); - assert.equal((bus2.events[0] as Task).status.state, "submitted"); - assert.equal((bus3.events[0] as Task).status.state, "rejected"); + assert.equal(taskPayload(bus2.events[0])?.status?.state, TaskState.TASK_STATE_SUBMITTED); + assert.equal(taskPayload(bus3.events[0])?.status?.state, TaskState.TASK_STATE_REJECTED); assert.equal(bus3.isFinished(), true); gates.get("task-1")?.resolve(); @@ -148,8 +162,8 @@ describe("P0 runtime components", () => { await p2; await p3; - assert.equal((bus1.events.at(-1) as Task).status.state, "completed"); - assert.equal((bus2.events.at(-1) as Task).status.state, "completed"); + assert.equal(taskPayload(bus1.events.at(-1))?.status?.state, TaskState.TASK_STATE_COMPLETED); + assert.equal(taskPayload(bus2.events.at(-1))?.status?.state, TaskState.TASK_STATE_COMPLETED); const snapshot = telemetry.snapshot(); assert.equal(snapshot.tasks.started, 2); @@ -158,4 +172,26 @@ describe("P0 runtime components", () => { assert.equal(snapshot.tasks.rejected, 1); assert.equal(snapshot.tasks.queued, 1); }); + + it("QueueingAgentExecutor closes a resource-owning delegate", () => { + const telemetry = new GatewayTelemetry(silentLogger(), { structuredLogs: false }); + let closed = false; + const delegate: AgentExecutor & { close(): void } = { + async execute() {}, + async cancelTask(_taskId, eventBus) { + eventBus.finished(); + }, + close() { + closed = true; + }, + }; + + const executor = new QueueingAgentExecutor(delegate, telemetry, { + maxConcurrentTasks: 1, + maxQueuedTasks: 0, + }); + + executor.close(); + assert.equal(closed, true); + }); }); diff --git a/tests/task-recovery.test.ts b/tests/task-recovery.test.ts index 13f66f3..c1e8bbc 100644 --- a/tests/task-recovery.test.ts +++ b/tests/task-recovery.test.ts @@ -4,6 +4,7 @@ import os from "node:os"; import path from "node:path"; import { describe, it } from "node:test"; +import { Role, TaskState } from "@a2a-js/sdk"; import type { Task } from "@a2a-js/sdk"; import { recoverStaleTasks } from "../src/task-recovery.js"; @@ -13,17 +14,20 @@ import { silentLogger } from "./helpers.js"; function makeTask( taskId: string, - state: string, + state: TaskState, timestamp?: string, ): Task { return { - kind: "task", id: taskId, contextId: `ctx-${taskId}`, status: { - state: state as any, + state, + message: undefined, ...(timestamp !== undefined ? { timestamp } : {}), }, + artifacts: [], + history: [], + metadata: undefined, }; } @@ -36,7 +40,7 @@ describe("recoverStaleTasks", () => { const dir = await mkdtemp(path.join(os.tmpdir(), "a2a-recovery-")); try { const store = new FileTaskStore(dir); - await store.save(makeTask("task-sub", "submitted", hoursAgo(2))); + await store.save(makeTask("task-sub", TaskState.TASK_STATE_SUBMITTED, hoursAgo(2))); const result = await recoverStaleTasks(store, silentLogger()); assert.equal(result.recovered, 1); @@ -45,7 +49,7 @@ describe("recoverStaleTasks", () => { const task = await store.load("task-sub"); assert.ok(task); - assert.equal(task.status.state, "failed"); + assert.equal(task.status.state, TaskState.TASK_STATE_FAILED); assert.ok(task.status.timestamp); } finally { await rm(dir, { recursive: true, force: true }); @@ -56,14 +60,14 @@ describe("recoverStaleTasks", () => { const dir = await mkdtemp(path.join(os.tmpdir(), "a2a-recovery-")); try { const store = new FileTaskStore(dir); - await store.save(makeTask("task-work", "working", hoursAgo(1))); + await store.save(makeTask("task-work", TaskState.TASK_STATE_WORKING, hoursAgo(1))); const result = await recoverStaleTasks(store, silentLogger()); assert.equal(result.recovered, 1); const task = await store.load("task-work"); assert.ok(task); - assert.equal(task.status.state, "failed"); + assert.equal(task.status.state, TaskState.TASK_STATE_FAILED); } finally { await rm(dir, { recursive: true, force: true }); } @@ -73,17 +77,17 @@ describe("recoverStaleTasks", () => { const dir = await mkdtemp(path.join(os.tmpdir(), "a2a-recovery-")); try { const store = new FileTaskStore(dir); - await store.save(makeTask("t-input", "input-required", hoursAgo(1))); - await store.save(makeTask("t-auth", "auth-required", hoursAgo(1))); - await store.save(makeTask("t-unknown", "unknown", hoursAgo(1))); + await store.save(makeTask("t-input", TaskState.TASK_STATE_INPUT_REQUIRED, hoursAgo(1))); + await store.save(makeTask("t-auth", TaskState.TASK_STATE_AUTH_REQUIRED, hoursAgo(1))); + await store.save(makeTask("t-unknown", TaskState.TASK_STATE_UNSPECIFIED, hoursAgo(1))); const result = await recoverStaleTasks(store, silentLogger()); assert.equal(result.recovered, 3); assert.equal(result.skipped, 0); - assert.equal((await store.load("t-input"))!.status.state, "failed"); - assert.equal((await store.load("t-auth"))!.status.state, "failed"); - assert.equal((await store.load("t-unknown"))!.status.state, "failed"); + assert.equal((await store.load("t-input"))!.status.state, TaskState.TASK_STATE_FAILED); + assert.equal((await store.load("t-auth"))!.status.state, TaskState.TASK_STATE_FAILED); + assert.equal((await store.load("t-unknown"))!.status.state, TaskState.TASK_STATE_FAILED); } finally { await rm(dir, { recursive: true, force: true }); } @@ -93,19 +97,19 @@ describe("recoverStaleTasks", () => { const dir = await mkdtemp(path.join(os.tmpdir(), "a2a-recovery-")); try { const store = new FileTaskStore(dir); - await store.save(makeTask("t-completed", "completed", hoursAgo(1))); - await store.save(makeTask("t-failed", "failed", hoursAgo(1))); - await store.save(makeTask("t-canceled", "canceled", hoursAgo(1))); - await store.save(makeTask("t-rejected", "rejected", hoursAgo(1))); + await store.save(makeTask("t-completed", TaskState.TASK_STATE_COMPLETED, hoursAgo(1))); + await store.save(makeTask("t-failed", TaskState.TASK_STATE_FAILED, hoursAgo(1))); + await store.save(makeTask("t-canceled", TaskState.TASK_STATE_CANCELED, hoursAgo(1))); + await store.save(makeTask("t-rejected", TaskState.TASK_STATE_REJECTED, hoursAgo(1))); const result = await recoverStaleTasks(store, silentLogger()); assert.equal(result.recovered, 0); assert.equal(result.skipped, 4); - assert.equal((await store.load("t-completed"))!.status.state, "completed"); - assert.equal((await store.load("t-failed"))!.status.state, "failed"); - assert.equal((await store.load("t-canceled"))!.status.state, "canceled"); - assert.equal((await store.load("t-rejected"))!.status.state, "rejected"); + assert.equal((await store.load("t-completed"))!.status.state, TaskState.TASK_STATE_COMPLETED); + assert.equal((await store.load("t-failed"))!.status.state, TaskState.TASK_STATE_FAILED); + assert.equal((await store.load("t-canceled"))!.status.state, TaskState.TASK_STATE_CANCELED); + assert.equal((await store.load("t-rejected"))!.status.state, TaskState.TASK_STATE_REJECTED); } finally { await rm(dir, { recursive: true, force: true }); } @@ -115,19 +119,19 @@ describe("recoverStaleTasks", () => { const dir = await mkdtemp(path.join(os.tmpdir(), "a2a-recovery-")); try { const store = new FileTaskStore(dir); - await store.save(makeTask("t1", "submitted", hoursAgo(5))); - await store.save(makeTask("t2", "completed", hoursAgo(1))); - await store.save(makeTask("t3", "working", hoursAgo(3))); - await store.save(makeTask("t4", "failed", hoursAgo(2))); + await store.save(makeTask("t1", TaskState.TASK_STATE_SUBMITTED, hoursAgo(5))); + await store.save(makeTask("t2", TaskState.TASK_STATE_COMPLETED, hoursAgo(1))); + await store.save(makeTask("t3", TaskState.TASK_STATE_WORKING, hoursAgo(3))); + await store.save(makeTask("t4", TaskState.TASK_STATE_FAILED, hoursAgo(2))); const result = await recoverStaleTasks(store, silentLogger()); assert.equal(result.recovered, 2); assert.equal(result.skipped, 2); - assert.equal((await store.load("t1"))!.status.state, "failed"); - assert.equal((await store.load("t2"))!.status.state, "completed"); - assert.equal((await store.load("t3"))!.status.state, "failed"); - assert.equal((await store.load("t4"))!.status.state, "failed"); + assert.equal((await store.load("t1"))!.status.state, TaskState.TASK_STATE_FAILED); + assert.equal((await store.load("t2"))!.status.state, TaskState.TASK_STATE_COMPLETED); + assert.equal((await store.load("t3"))!.status.state, TaskState.TASK_STATE_FAILED); + assert.equal((await store.load("t4"))!.status.state, TaskState.TASK_STATE_FAILED); } finally { await rm(dir, { recursive: true, force: true }); } @@ -137,7 +141,7 @@ describe("recoverStaleTasks", () => { const dir = await mkdtemp(path.join(os.tmpdir(), "a2a-recovery-")); try { const store = new FileTaskStore(dir); - await store.save(makeTask("task-msg", "working", hoursAgo(1))); + await store.save(makeTask("task-msg", TaskState.TASK_STATE_WORKING, hoursAgo(1))); await recoverStaleTasks(store, silentLogger()); @@ -145,12 +149,14 @@ describe("recoverStaleTasks", () => { assert.ok(task); const message = task.status.message; assert.ok(message); - assert.equal(message.kind, "message"); assert.ok(message.messageId, "messageId must be present"); - assert.equal(message.role, "agent"); + assert.equal(message.role, Role.ROLE_AGENT); assert.ok(message.parts.length > 0); - const text = (message.parts[0] as any).text; - assert.ok(text.includes("was: working")); + const first = message.parts[0] as any; + assert.equal(first.content?.$case, "text"); + const text = String(first.content?.value ?? ""); + // taskStateName() renders lowercase wire names ("working") + assert.equal(text.includes("was: working"), true); } finally { await rm(dir, { recursive: true, force: true }); } diff --git a/tests/task-ttl.test.ts b/tests/task-ttl.test.ts index b27dbe3..433e1ff 100644 --- a/tests/task-ttl.test.ts +++ b/tests/task-ttl.test.ts @@ -4,6 +4,7 @@ import os from "node:os"; import path from "node:path"; import { describe, it } from "node:test"; +import { TaskState } from "@a2a-js/sdk"; import type { Task } from "@a2a-js/sdk"; import { runTaskCleanup } from "../src/task-cleanup.js"; @@ -18,17 +19,20 @@ function makeTelemetry() { function makeTask( taskId: string, - state: string, + state: TaskState, timestamp?: string, ): Task { return { - kind: "task", id: taskId, contextId: `ctx-${taskId}`, status: { - state: state as any, + state, + message: undefined, ...(timestamp !== undefined ? { timestamp } : {}), }, + artifacts: [], + history: [], + metadata: undefined, }; } @@ -41,9 +45,9 @@ describe("FileTaskStore extensions", () => { const dir = await mkdtemp(path.join(os.tmpdir(), "a2a-ttl-list-")); try { const store = new FileTaskStore(dir); - await store.save(makeTask("alpha-uno", "completed", hoursAgo(1))); - await store.save(makeTask("beta-dos", "failed", hoursAgo(2))); - await store.save(makeTask("gamma-tres", "working", hoursAgo(3))); + await store.save(makeTask("alpha-uno", TaskState.TASK_STATE_COMPLETED, hoursAgo(1))); + await store.save(makeTask("beta-dos", TaskState.TASK_STATE_FAILED, hoursAgo(2))); + await store.save(makeTask("gamma-tres", TaskState.TASK_STATE_WORKING, hoursAgo(3))); const ids = await store.listAll(); assert.equal(ids.length, 3); @@ -65,7 +69,7 @@ describe("FileTaskStore extensions", () => { const dir = await mkdtemp(path.join(os.tmpdir(), "a2a-ttl-del-")); try { const store = new FileTaskStore(dir); - await store.save(makeTask("doomed-task", "completed", hoursAgo(1))); + await store.save(makeTask("doomed-task", TaskState.TASK_STATE_COMPLETED, hoursAgo(1))); const before = await store.load("doomed-task"); assert.ok(before, "task should exist before delete"); @@ -107,12 +111,12 @@ describe("FileTaskStore extensions", () => { assert.equal(exists, false, "dir should not exist before first save"); // First save creates the directory - await store.save(makeTask("lazy-1", "completed", hoursAgo(1))); + await store.save(makeTask("lazy-1", TaskState.TASK_STATE_COMPLETED, hoursAgo(1))); const entries1 = await readdir(dir); assert.equal(entries1.length, 1); // Second save reuses the directory (no error) - await store.save(makeTask("lazy-2", "completed", hoursAgo(1))); + await store.save(makeTask("lazy-2", TaskState.TASK_STATE_COMPLETED, hoursAgo(1))); const entries2 = await readdir(dir); assert.equal(entries2.length, 2); } finally { @@ -128,14 +132,14 @@ describe("FileTaskStore extensions", () => { const store = new FileTaskStore(path.join(blocker, "tasks")); await assert.rejects( - store.save(makeTask("first-attempt", "completed", hoursAgo(1))), + store.save(makeTask("first-attempt", TaskState.TASK_STATE_COMPLETED, hoursAgo(1))), (error: unknown) => (error as { code?: string } | undefined)?.code === "ENOTDIR", ); await rm(blocker, { force: true }); await mkdir(blocker, { recursive: true }); - await store.save(makeTask("second-attempt", "completed", hoursAgo(1))); + await store.save(makeTask("second-attempt", TaskState.TASK_STATE_COMPLETED, hoursAgo(1))); const recovered = await store.load("second-attempt"); assert.ok(recovered, "save should recover once the directory path becomes valid"); } finally { @@ -152,13 +156,13 @@ describe("Task TTL cleanup", () => { const telemetry = makeTelemetry(); // Old terminal tasks — should be expired - await store.save(makeTask("old-completed-999", "completed", hoursAgo(100))); - await store.save(makeTask("old-failed-888", "failed", hoursAgo(200))); - await store.save(makeTask("old-canceled-777", "canceled", hoursAgo(150))); - await store.save(makeTask("old-rejected-666", "rejected", hoursAgo(80))); + await store.save(makeTask("old-completed-999", TaskState.TASK_STATE_COMPLETED, hoursAgo(100))); + await store.save(makeTask("old-failed-888", TaskState.TASK_STATE_FAILED, hoursAgo(200))); + await store.save(makeTask("old-canceled-777", TaskState.TASK_STATE_CANCELED, hoursAgo(150))); + await store.save(makeTask("old-rejected-666", TaskState.TASK_STATE_REJECTED, hoursAgo(80))); // Fresh terminal task — should survive - await store.save(makeTask("fresh-completed-111", "completed", hoursAgo(1))); + await store.save(makeTask("fresh-completed-111", TaskState.TASK_STATE_COMPLETED, hoursAgo(1))); const ttl72h = 72 * 3_600_000; const result = await runTaskCleanup(store, ttl72h, telemetry, silentLogger()); @@ -192,9 +196,9 @@ describe("Task TTL cleanup", () => { const telemetry = makeTelemetry(); // Old but active — must NOT be deleted - await store.save(makeTask("working-ancient-555", "working", hoursAgo(500))); - await store.save(makeTask("submitted-old-444", "submitted", hoursAgo(300))); - await store.save(makeTask("input-required-333", "input-required", hoursAgo(200))); + await store.save(makeTask("working-ancient-555", TaskState.TASK_STATE_WORKING, hoursAgo(500))); + await store.save(makeTask("submitted-old-444", TaskState.TASK_STATE_SUBMITTED, hoursAgo(300))); + await store.save(makeTask("input-required-333", TaskState.TASK_STATE_INPUT_REQUIRED, hoursAgo(200))); const ttl1h = 1 * 3_600_000; const result = await runTaskCleanup(store, ttl1h, telemetry, silentLogger()); @@ -218,7 +222,7 @@ describe("Task TTL cleanup", () => { const telemetry = makeTelemetry(); // Terminal but no timestamp — should skip - await store.save(makeTask("no-timestamp-222", "completed")); + await store.save(makeTask("no-timestamp-222", TaskState.TASK_STATE_COMPLETED)); const ttl1h = 1 * 3_600_000; const result = await runTaskCleanup(store, ttl1h, telemetry, silentLogger()); @@ -264,7 +268,7 @@ describe("Task TTL cleanup", () => { const store = new FileTaskStore(dir); const telemetry = makeTelemetry(); - await store.save(makeTask("race-1", "completed", hoursAgo(100))); + await store.save(makeTask("race-1", TaskState.TASK_STATE_COMPLETED, hoursAgo(100))); const [first, second] = await Promise.all([ runTaskCleanup(store, 72 * 3_600_000, telemetry, silentLogger()),