diff --git a/docs/a2a-integration-guide.md b/docs/a2a-integration-guide.md index 7df60c91..bd7fea9b 100644 --- a/docs/a2a-integration-guide.md +++ b/docs/a2a-integration-guide.md @@ -171,6 +171,8 @@ Consumer rules: - **Frame type** = the single top-level key: `task` | `statusUpdate` | `artifactUpdate`. - `artifactUpdate` with `append: true` → **concatenate** `artifact.parts[].text` to build the answer. `lastChunk: true` marks the final chunk. +- `artifactUpdate` with `append: false` → **replace** the current answer. Siclaw + uses this when a complete assistant message corrects missing or divergent deltas. - `statusUpdate` with a terminal `status.state` → the stream is done. - Lines starting with `:` are heartbeat comments (every ~25s) — ignore them. @@ -187,8 +189,12 @@ with requests.post(f"{base}/message:stream", headers=H, stream=True, continue obj = json.loads(line[6:]) if "artifactUpdate" in obj: - for p in obj["artifactUpdate"]["artifact"].get("parts", []): - buf.append(p.get("text", "")) + update = obj["artifactUpdate"] + text = "".join(p.get("text", "") for p in update["artifact"].get("parts", [])) + if update.get("append", False): + buf.append(text) + else: + buf = [text] elif "statusUpdate" in obj: state = obj["statusUpdate"]["status"]["state"] if state.startswith("TASK_STATE_") and state not in ("TASK_STATE_SUBMITTED", "TASK_STATE_WORKING"): @@ -203,7 +209,9 @@ print("".join(buf)) - **Continue the same investigation thread**: pass the same `contextId` on the next `message:send`/`message:stream`. Omit it to start fresh (the server generates one and returns it on the task). Each call still creates a new task; the `contextId` - reuses the underlying session. + reuses the underlying session. Calls in one context are sequential: wait for the + current task to become terminal (or cancel it) before submitting the next one. + A concurrent submission returns `409 CONTEXT_BUSY`. - **Reconnect a dropped stream**: `POST .../tasks/:subscribe` (only while the task is non-terminal; a terminal task returns `400`). Or just fall back to polling `GET .../tasks/` — polling always works. @@ -238,6 +246,7 @@ JSON error envelope (google.rpc shape): | 401 | `UNAUTHENTICATED` | missing/invalid key | | 403 | `PERMISSION_DENIED` | key not bound to this `agentId` | | 404 | `NOT_FOUND` | task does not exist (or not yours) | +| 409 | `ABORTED` | another non-terminal task already owns this context | | 413 | `RESOURCE_EXHAUSTED` | body over 1 MB | | 502 | `UNAVAILABLE`/runtime error | runtime command failed | | 503 | `UNAVAILABLE` | agent runtime not connected | @@ -249,6 +258,8 @@ JSON error envelope (google.rpc shape): - **Text parts only.** `message.parts[].text`; `raw`/`url`/`data`/file parts are rejected. - `message.role`, if present, must be `ROLE_USER`. - Body ≤ **1 MB**; any id (`messageId`/`contextId`) ≤ **255 bytes**. +- Partial artifacts are checkpointed while a task is `WORKING` (first update, + then at most once per `SICLAW_A2A_PROGRESS_FLUSH_MS`, default 3000 ms). - `message.taskId` continuation / `input-required` is **not** supported yet — start a new task in the same `contextId` instead. - Push notifications, Agent Card signing, OAuth/mTLS, and the full A2A JSON-RPC diff --git a/docs/design/2026-06-18-a2a-gateway.md b/docs/design/2026-06-18-a2a-gateway.md index ba166983..a34fb5fa 100644 --- a/docs/design/2026-06-18-a2a-gateway.md +++ b/docs/design/2026-06-18-a2a-gateway.md @@ -91,6 +91,10 @@ Supported payload: - Siclaw stores the external `contextId` separately and maps it to an internal UUID `sessionId`; repeat calls with the same `contextId` reuse the latest internal session for that API key. +- Context reuse is sequential: at most one non-terminal task may own a given + `(agent, API key, contextId)` at a time. A concurrent submission returns + `409 CONTEXT_BUSY`. This prevents two task trackers from consuming the same + session-scoped Runtime event stream. - If no `contextId` is provided, Siclaw creates one UUID and uses it as both the A2A `contextId` and internal `sessionId`. - `message.taskId` continuation/input-required semantics are deferred until @@ -108,6 +112,7 @@ a2a_tasks - api_key_id - context_id - session_id +- active_context_key - state - status_message - artifact_text @@ -124,12 +129,25 @@ history, while `a2a_tasks` answers protocol-level state queries. The table keeps both `context_id` and `session_id` because A2A clients may use non-UUID context identifiers, while Siclaw's internal chat session storage remains UUID-shaped. +`active_context_key` is a nullable SHA-256 lease over `(agent, API key, +contextId)`. A unique index makes admission atomic across concurrent requests +and Portal replicas. Terminal transitions clear it in the same SQL update; +multiple `NULL` values preserve completed task history while allowing the next +task in that context. + +While a task is working, the adapter checkpoints `artifact_text` on the first +text update and then at a throttled interval (default 3 seconds). The final +artifact write is awaited before a terminal status is emitted. `chat_messages` +remains the Runtime transcript; `a2a_tasks` owns the durable A2A protocol output +for each task, so the two persistence domains are related but not interchangeable. + ## Event Mapping | Siclaw chat.event | A2A output | | --- | --- | | `message_update` text delta | `artifactUpdate` append chunk | | `agent_message` text | `artifactUpdate` append chunk | +| assistant `message_end` | reconcile complete message; replace artifact if deltas diverged | | `tool_execution_start` | `statusUpdate` working, current tool metadata | | `tool_execution_end` | `statusUpdate` working, tool finished metadata | | `stream_error` | `statusUpdate` failed | @@ -140,6 +158,10 @@ identifiers, while Siclaw's internal chat session storage remains UUID-shaped. Runtime `chat.sessionStatus`; if Runtime still reports running, the task stays `TASK_STATE_WORKING` even if no recent DB row changed. +`SICLAW_A2A_PROGRESS_FLUSH_MS` controls partial artifact checkpointing. It +defaults to `3000`; set it to `0` to disable working-state checkpoints. Terminal +artifact persistence is never disabled. + ## Re-hosting Behind a Gateway (Future) This contract is intentionally a self-contained boundary adapter, so a downstream diff --git a/src/portal/a2a-gateway.test.ts b/src/portal/a2a-gateway.test.ts index d16117dc..160abb60 100644 --- a/src/portal/a2a-gateway.test.ts +++ b/src/portal/a2a-gateway.test.ts @@ -240,7 +240,7 @@ describe("registerA2aRoutes", () => { expect(sent[2].systemPrompt).toBe("你是 SRE 专家。"); }); - it("maps a non-UUID A2A context to a stable internal Siclaw session", async () => { + it("reuses a non-UUID context session only after the previous task completes", async () => { const { router, conn } = await makeRouter(); const first = await runRoute(router, fakeReq({ url: "/api/v1/a2a/agents/a1/message:send", @@ -248,6 +248,11 @@ describe("registerA2aRoutes", () => { headers: { authorization: `Bearer ${API_KEY}` }, body: { message: { role: "ROLE_USER", contextId: "external-context", parts: [{ text: "first" }] } }, })); + const firstSessionId = (conn.map.sendCommand as any).mock.calls + .find((call: any[]) => call[1] === "chat.send")[2].sessionId; + conn.emit({ sessionId: firstSessionId, event: { type: "prompt_done" } }); + for (let i = 0; i < 3; i++) await new Promise((r) => setImmediate(r)); + const second = await runRoute(router, fakeReq({ url: "/api/v1/a2a/agents/a1/message:send", method: "POST", @@ -259,11 +264,57 @@ describe("registerA2aRoutes", () => { expect(second._status).toBe(200); const sendCalls = (conn.map.sendCommand as any).mock.calls.filter((call: any[]) => call[1] === "chat.send"); expect(sendCalls).toHaveLength(2); - expect(sendCalls[0][2].sessionId).toMatch(/^[0-9a-f-]{36}$/); + expect(sendCalls[0][2].sessionId).toBe(firstSessionId); expect(sendCalls[1][2].sessionId).toBe(sendCalls[0][2].sessionId); expect(parseJsonBody(second).task.contextId).toBe("external-context"); }); + it("rejects a concurrent task in the same context before dispatching it", async () => { + const { router, conn } = await makeRouter(); + const first = await runRoute(router, fakeReq({ + url: "/api/v1/a2a/agents/a1/message:send", + method: "POST", + headers: { authorization: `Bearer ${API_KEY}` }, + body: { message: { role: "ROLE_USER", contextId: "busy-context", parts: [{ text: "first" }] } }, + })); + const second = await runRoute(router, fakeReq({ + url: "/api/v1/a2a/agents/a1/message:send", + method: "POST", + headers: { authorization: `Bearer ${API_KEY}` }, + body: { message: { role: "ROLE_USER", contextId: "busy-context", parts: [{ text: "second" }] } }, + })); + + expect(first._status).toBe(200); + expect(second._status).toBe(409); + const error = parseJsonBody(second).error; + expect(error.status).toBe("ABORTED"); + expect(error.details[0].reason).toBe("CONTEXT_BUSY"); + const sendCalls = (conn.map.sendCommand as any).mock.calls.filter((call: any[]) => call[1] === "chat.send"); + expect(sendCalls).toHaveLength(1); + }); + + it("serializes racing submissions with a database-backed context lease", async () => { + const { router, conn } = await makeRouter(); + const request = (text: string) => runRoute(router, fakeReq({ + url: "/api/v1/a2a/agents/a1/message:send", + method: "POST", + headers: { authorization: `Bearer ${API_KEY}` }, + body: { message: { role: "ROLE_USER", contextId: "racing-context", parts: [{ text }] } }, + })); + + const responses = await Promise.all([request("first"), request("second")]); + expect(responses.map((response) => response._status).sort()).toEqual([200, 409]); + const sendCalls = (conn.map.sendCommand as any).mock.calls.filter((call: any[]) => call[1] === "chat.send"); + expect(sendCalls).toHaveLength(1); + + const [rows] = await getDb().query>( + `SELECT COUNT(*) AS c FROM a2a_tasks + WHERE context_id = ? AND state IN ('TASK_STATE_SUBMITTED', 'TASK_STATE_WORKING')`, + ["racing-context"], + ); + expect(Number(rows[0].c)).toBe(1); + }); + it("rejects oversized A2A context identifiers before dispatching", async () => { const { router, conn } = await makeRouter(); const res = await runRoute(router, fakeReq({ @@ -322,6 +373,99 @@ describe("registerA2aRoutes", () => { expect(res.writableEnded).toBe(true); }); + it("checkpoints a partial artifact while the task is still working", async () => { + const { router, conn } = await makeRouter(); + const createRes = await runRoute(router, fakeReq({ + url: "/api/v1/a2a/agents/a1/message:send", + method: "POST", + headers: { authorization: `Bearer ${API_KEY}` }, + body: { message: { role: "ROLE_USER", parts: [{ text: "stream it" }] } }, + })); + const taskId = parseJsonBody(createRes).task.id; + const sent = (conn.map.sendCommand as any).mock.calls.find((call: any[]) => call[1] === "chat.send"); + + conn.emit({ + sessionId: sent[2].sessionId, + event: { type: "message_update", assistantMessageEvent: { type: "text_delta", delta: "partial" } }, + }); + for (let i = 0; i < 3; i++) await new Promise((r) => setImmediate(r)); + + const [rows] = await getDb().query>( + "SELECT state, artifact_text FROM a2a_tasks WHERE id = ?", + [taskId], + ); + expect(rows[0]).toEqual({ state: "TASK_STATE_WORKING", artifact_text: "partial" }); + }); + + it("uses the full assistant message when no text deltas were emitted", async () => { + const { router, conn } = await makeRouter(); + const createRes = await runRoute(router, fakeReq({ + url: "/api/v1/a2a/agents/a1/message:send", + method: "POST", + headers: { authorization: `Bearer ${API_KEY}` }, + body: { message: { role: "ROLE_USER", parts: [{ text: "answer without deltas" }] } }, + })); + const taskId = parseJsonBody(createRes).task.id; + const sent = (conn.map.sendCommand as any).mock.calls.find((call: any[]) => call[1] === "chat.send"); + + conn.emit({ + sessionId: sent[2].sessionId, + event: { + type: "message_end", + message: { role: "assistant", content: [{ type: "text", text: "complete answer" }] }, + }, + }); + conn.emit({ sessionId: sent[2].sessionId, event: { type: "prompt_done" } }); + for (let i = 0; i < 4; i++) await new Promise((r) => setImmediate(r)); + + const [rows] = await getDb().query>( + "SELECT state, artifact_text FROM a2a_tasks WHERE id = ?", + [taskId], + ); + expect(rows[0]).toEqual({ state: "TASK_STATE_COMPLETED", artifact_text: "complete answer" }); + }); + + it("reconciles divergent deltas to the authoritative message_end content", async () => { + const { router, conn } = await makeRouter(); + const res = fakeRes(); + router.handle(fakeReq({ + url: "/api/v1/a2a/agents/a1/message:stream", + method: "POST", + headers: { authorization: `Bearer ${API_KEY}` }, + body: { message: { role: "ROLE_USER", parts: [{ text: "recover dropped deltas" }] } }, + }), res); + for (let i = 0; i < 3; i++) await new Promise((r) => setImmediate(r)); + const sent = (conn.map.sendCommand as any).mock.calls.find((call: any[]) => call[1] === "chat.send"); + + conn.emit({ + sessionId: sent[2].sessionId, + event: { type: "message_update", assistantMessageEvent: { type: "text_delta", delta: "helx" } }, + }); + conn.emit({ + sessionId: sent[2].sessionId, + event: { + type: "message_end", + message: { role: "assistant", content: [{ type: "text", text: "hello" }] }, + }, + }); + conn.emit({ sessionId: sent[2].sessionId, event: { type: "prompt_done" } }); + for (let i = 0; i < 5; i++) await new Promise((r) => setImmediate(r)); + + const frames = parseSseDataChunks(res); + expect(frames).toContainEqual(expect.objectContaining({ + artifactUpdate: expect.objectContaining({ + append: false, + artifact: expect.objectContaining({ parts: [expect.objectContaining({ text: "hello" })] }), + }), + })); + const taskId = frames.find((frame) => frame.task)?.task.id; + const [rows] = await getDb().query>( + "SELECT artifact_text FROM a2a_tasks WHERE id = ?", + [taskId], + ); + expect(rows[0].artifact_text).toBe("hello"); + }); + it("lists A2A tasks with context and status filters", async () => { const { router } = await makeRouter(); const createRes = await runRoute(router, fakeReq({ diff --git a/src/portal/a2a-gateway.ts b/src/portal/a2a-gateway.ts index 2883cc47..aa7d75cc 100644 --- a/src/portal/a2a-gateway.ts +++ b/src/portal/a2a-gateway.ts @@ -7,6 +7,7 @@ import { type RestRouter, } from "../gateway/rest-router.js"; import { getDb } from "../gateway/db.js"; +import { isUniqueViolation } from "../gateway/dialect-helpers.js"; import type { RuntimeConnectionMap } from "./runtime-connection.js"; import { authenticateApiKey, type ApiKeyAuthResult } from "./api-key-auth.js"; import { resolveAgentModelBinding } from "./chat-gateway.js"; @@ -80,6 +81,10 @@ type A2aStreamResponse = interface ActiveTracker { unsubscribe: () => void; artifactText: string; + currentMessageText: string; + messageStartOffset: number; + lastFlushAt: number; + persistChain: Promise; } const activeTrackers = new Map(); @@ -343,23 +348,32 @@ async function createTaskRecord(params: { apiKeyId: string; contextId: string; sessionId: string; + activeContextKey: string; }): Promise { const db = getDb(); - await db.query( - `INSERT INTO a2a_tasks - (id, agent_id, user_id, api_key_id, context_id, session_id, state, status_message) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, - [ - params.id, - params.agentId, - params.userId, - params.apiKeyId, - params.contextId, - params.sessionId, - "TASK_STATE_SUBMITTED", - "Task submitted to Siclaw", - ], - ); + try { + await db.query( + `INSERT INTO a2a_tasks + (id, agent_id, user_id, api_key_id, context_id, session_id, active_context_key, state, status_message) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + [ + params.id, + params.agentId, + params.userId, + params.apiKeyId, + params.contextId, + params.sessionId, + params.activeContextKey, + "TASK_STATE_SUBMITTED", + "Task submitted to Siclaw", + ], + ); + } catch (err) { + if (isUniqueViolation(err)) { + throw contextBusyError(); + } + throw err; + } return { id: params.id, agentId: params.agentId, @@ -378,6 +392,37 @@ async function createTaskRecord(params: { }; } +function activeContextKey(agentId: string, apiKeyId: string, contextId: string): string { + return crypto.createHash("sha256") + .update(agentId) + .update("\0") + .update(apiKeyId) + .update("\0") + .update(contextId) + .digest("hex"); +} + +function contextBusyError(): A2aHttpError { + return new A2aHttpError( + 409, + "CONTEXT_BUSY", + "Another A2A task is already active in this context; wait for it to finish or cancel it", + ); +} + +async function hasActiveTaskForContext(agentId: string, apiKeyId: string, contextId: string): Promise { + const db = getDb(); + const [rows] = await db.query( + `SELECT id + FROM a2a_tasks + WHERE agent_id = ? AND api_key_id = ? AND context_id = ? + AND state NOT IN (${TERMINAL_STATES_SQL}) + LIMIT 1`, + [agentId, apiKeyId, contextId], + ) as any; + return rows.length > 0; +} + async function loadTaskRecord(agentId: string, taskId: string, apiKeyId: string): Promise { const db = getDb(); const [rows] = await db.query( @@ -475,7 +520,8 @@ async function setTaskState( // mislabel the task (user clicked cancel, DB ends up FAILED). const sql = terminal ? `UPDATE a2a_tasks - SET state = ?, status_message = ?, error = ?, updated_at = CURRENT_TIMESTAMP, + SET state = ?, status_message = ?, error = ?, active_context_key = NULL, + updated_at = CURRENT_TIMESTAMP, last_event_at = CURRENT_TIMESTAMP, completed_at = CURRENT_TIMESTAMP WHERE id = ? AND state NOT IN (${TERMINAL_STATES_SQL})` : `UPDATE a2a_tasks @@ -557,7 +603,12 @@ function buildStatusUpdate( }; } -function buildArtifactUpdate(task: A2aTaskRecord, delta: string, lastChunk = false): A2aStreamResponse { +function buildArtifactUpdate( + task: A2aTaskRecord, + text: string, + lastChunk = false, + append = true, +): A2aStreamResponse { return { artifactUpdate: { taskId: task.id, @@ -565,9 +616,9 @@ function buildArtifactUpdate(task: A2aTaskRecord, delta: string, lastChunk = fal artifact: { artifactId: ASSISTANT_ARTIFACT_ID, name: "Siclaw diagnosis", - parts: [{ text: delta, mediaType: "text/markdown" }], + parts: [{ text, mediaType: "text/markdown" }], }, - append: true, + append, lastChunk, }, }; @@ -603,6 +654,44 @@ function extractDelta(evt: Record): string { return ""; } +function extractAssistantMessageText(evt: Record): string { + if (evt.type !== "message_end") return ""; + const message = (evt as any).message; + if (message?.role !== "assistant") return ""; + if (typeof message.content === "string") return message.content; + if (!Array.isArray(message.content)) return ""; + return message.content + .filter((block: unknown) => block && typeof block === "object" && (block as any).type === "text") + .map((block: unknown) => typeof (block as any).text === "string" ? (block as any).text : "") + .join(""); +} + +function progressFlushMs(): number { + const value = Number(process.env.SICLAW_A2A_PROGRESS_FLUSH_MS ?? 3_000); + return Number.isFinite(value) && value >= 0 ? value : 3_000; +} + +function queueArtifactCheckpoint(taskId: string, tracker: ActiveTracker): void { + const snapshot = tracker.artifactText; + tracker.persistChain = tracker.persistChain + .then(() => setTaskArtifact(taskId, snapshot)) + .catch((err) => { + // A partial checkpoint is recoverability metadata, not a terminal transition. + // Keep streaming and retry with a newer snapshot; the terminal path still performs + // an awaited final write before reporting completion. + console.error(`[a2a-gateway] partial artifact checkpoint failed for task ${taskId}:`, err); + }); +} + +function maybeCheckpointArtifact(taskId: string, tracker: ActiveTracker): void { + const interval = progressFlushMs(); + if (interval <= 0) return; + const now = Date.now(); + if (tracker.lastFlushAt !== 0 && now - tracker.lastFlushAt < interval) return; + tracker.lastFlushAt = now; + queueArtifactCheckpoint(taskId, tracker); +} + function toolName(evt: Record): string | undefined { const name = (evt as any).toolName ?? (evt as any).name; return typeof name === "string" && name ? name : undefined; @@ -620,6 +709,10 @@ function ensureTaskTracker(task: A2aTaskRecord, connectionMap: RuntimeConnection const tracker: ActiveTracker = { artifactText: task.artifactText, + currentMessageText: "", + messageStartOffset: task.artifactText.length, + lastFlushAt: 0, + persistChain: Promise.resolve(), unsubscribe: () => {}, }; @@ -643,14 +736,54 @@ async function handleTrackedEvent( tracker: ActiveTracker, evt: Record, ): Promise { - // Text deltas accumulate in the in-memory tracker and are flushed to a2a_tasks.artifact_text - // only at terminal events (prompt_done / stream_error). The a2a_tasks row is a protocol - // projection, not the durable transcript — chat_messages is the audit source of truth — so a - // crash mid-turn loses the partial artifact from the projection but never from chat history. + if (evt.type === "message_start") { + tracker.currentMessageText = ""; + tracker.messageStartOffset = tracker.artifactText.length; + return; + } + const delta = extractDelta(evt); if (delta) { + if (!tracker.currentMessageText) tracker.messageStartOffset = tracker.artifactText.length; tracker.artifactText += delta; + tracker.currentMessageText += delta; emitTask(task.id, buildArtifactUpdate(task, delta)); + maybeCheckpointArtifact(task.id, tracker); + return; + } + + if (evt.type === "message_end") { + const completeText = extractAssistantMessageText(evt); + if (!completeText) { + tracker.currentMessageText = ""; + return; + } + + if (!tracker.currentMessageText) { + tracker.artifactText += completeText; + emitTask(task.id, buildArtifactUpdate(task, completeText)); + } else if (completeText.startsWith(tracker.currentMessageText)) { + const missingSuffix = completeText.slice(tracker.currentMessageText.length); + if (missingSuffix) { + tracker.artifactText += missingSuffix; + emitTask(task.id, buildArtifactUpdate(task, missingSuffix)); + } + } else { + // A full message_end is the authoritative content for this assistant message. + // If deltas were dropped or rewritten, replace the current message tail and emit + // an append=false snapshot so streaming clients converge with polling clients. + tracker.artifactText = tracker.artifactText.slice(0, tracker.messageStartOffset) + completeText; + emitTask(task.id, buildArtifactUpdate(task, tracker.artifactText, false, false)); + } + tracker.currentMessageText = ""; + tracker.messageStartOffset = tracker.artifactText.length; + // A completed assistant message can be followed by a long tool call with no text + // deltas. Persist it immediately (still serialized behind any earlier checkpoint) + // so polling clients do not remain stuck on the first throttled prefix. + if (progressFlushMs() > 0) { + tracker.lastFlushAt = Date.now(); + queueArtifactCheckpoint(task.id, tracker); + } return; } @@ -682,6 +815,7 @@ async function handleTrackedEvent( return; } const message = streamErrorMessage(evt); + await tracker.persistChain; await setTaskArtifact(task.id, tracker.artifactText); await setTaskState(task.id, "TASK_STATE_FAILED", message, message); emitTask(task.id, buildStatusUpdate(task, "TASK_STATE_FAILED", message)); @@ -695,6 +829,7 @@ async function handleTrackedEvent( stopTaskTracker(task.id); return; } + await tracker.persistChain; await setTaskArtifact(task.id, tracker.artifactText); await setTaskState(task.id, "TASK_STATE_COMPLETED", "Siclaw task completed"); emitTask(task.id, buildArtifactUpdate(task, "", true)); @@ -710,6 +845,13 @@ function stopTaskTracker(taskId: string): void { tracker.unsubscribe(); } +async function persistTrackedArtifact(taskId: string): Promise { + const tracker = activeTrackers.get(taskId); + if (!tracker) return; + await tracker.persistChain; + await setTaskArtifact(taskId, tracker.artifactText); +} + async function submitA2aTask(params: { agentId: string; auth: ApiKeyAuthResult; @@ -729,6 +871,9 @@ async function submitA2aTask(params: { const taskId = crypto.randomUUID(); const contextId = message.contextId || crypto.randomUUID(); + if (await hasActiveTaskForContext(agentId, auth.keyId, contextId)) { + throw contextBusyError(); + } const sessionId = message.contextId ? (await loadSessionIdForContext(agentId, auth.keyId, contextId)) ?? crypto.randomUUID() : contextId; @@ -739,6 +884,7 @@ async function submitA2aTask(params: { apiKeyId: auth.keyId, contextId, sessionId, + activeContextKey: activeContextKey(agentId, auth.keyId, contextId), }); ensureTaskTracker(task, connectionMap); @@ -799,6 +945,7 @@ async function reconcileOrphanedTask(task: A2aTaskRecord): Promise { try { const { auth, task } = await loadAuthorizedTask(req, params.agentId, params.taskId); + let cancelAttempted = false; if (!isTerminalState(task.state)) { + cancelAttempted = true; const result = await connectionMap.sendCommand(params.agentId, "chat.abort", { agentId: params.agentId, userId: auth.createdBy, sessionId: task.sessionId, }); if (!result.ok) throw new A2aHttpError(502, "RUNTIME_ERROR", result.error ?? "Runtime cancel failed"); + await persistTrackedArtifact(task.id); await setTaskState(task.id, "TASK_STATE_CANCELED", "Task canceled by A2A client"); - emitTask(task.id, buildStatusUpdate(task, "TASK_STATE_CANCELED", "Task canceled by A2A client")); stopTaskTracker(task.id); } const latest = await loadTaskRecord(params.agentId, params.taskId, auth.keyId) ?? task; + // A runtime terminal event may win the race while chat.abort is in flight. The + // terminal-immutable SQL guard preserves that state; only emit CANCELED if this + // request actually won, so subscribers never see a contradictory terminal frame. + if (cancelAttempted && latest.state === "TASK_STATE_CANCELED") { + emitTask(task.id, buildStatusUpdate(task, "TASK_STATE_CANCELED", "Task canceled by A2A client")); + } sendA2aJson(res, 200, { task: buildTask(latest) }); } catch (err) { respondA2aError(req, res, err); diff --git a/src/portal/migrate-sqlite.test.ts b/src/portal/migrate-sqlite.test.ts index a676f14b..7a228b7b 100644 --- a/src/portal/migrate-sqlite.test.ts +++ b/src/portal/migrate-sqlite.test.ts @@ -113,6 +113,18 @@ describe("runPortalMigrations on SQLite :memory:", () => { "SELECT name FROM sqlite_master WHERE type = 'index' AND name LIKE 'idx_%'", ); expect(allIdx.map((r) => r.name).sort()).toEqual(expectedIndexes.slice().sort()); + + const [activeContextIndexes] = await db.query>( + "SELECT name, sql FROM sqlite_master WHERE type = 'index' AND name = 'uq_a2a_tasks_active_context'", + ); + expect(activeContextIndexes).toHaveLength(1); + expect(activeContextIndexes[0].sql).toContain("CREATE UNIQUE INDEX"); + }); + + it("adds the A2A active-context lease column", async () => { + await runPortalMigrations(); + const [columns] = await getDb().query>("PRAGMA table_info(`a2a_tasks`)"); + expect(columns.map((column) => column.name)).toContain("active_context_key"); }); it("is resilient to legacy-style pre-populated schema (simulated dump replay)", async () => { diff --git a/src/portal/migrate.ts b/src/portal/migrate.ts index 24fb3560..160a0d22 100644 --- a/src/portal/migrate.ts +++ b/src/portal/migrate.ts @@ -378,6 +378,7 @@ const PORTAL_SCHEMA_SQLS: string[] = [ api_key_id CHAR(36) DEFAULT NULL, context_id VARCHAR(255) NOT NULL, session_id CHAR(36) NOT NULL, + active_context_key VARCHAR(64) DEFAULT NULL, state VARCHAR(40) NOT NULL, status_message TEXT, artifact_text TEXT, @@ -579,6 +580,11 @@ async function createIndexes(): Promise { await ensureIndex(db, "a2a_tasks", "idx_a2a_tasks_agent_key", "agent_id, api_key_id, created_at"); await ensureIndex(db, "a2a_tasks", "idx_a2a_tasks_session", "session_id"); await ensureIndex(db, "a2a_tasks", "idx_a2a_tasks_context_key", "agent_id, api_key_id, context_id, created_at"); + // A non-NULL lease exists only while a task is non-terminal. Its SHA-256 value covers + // (agent, API key, context), so this unique index serializes same-context submissions + // across concurrent requests and Portal replicas. Both SQLite and MySQL allow multiple + // NULL values, which releases the lease without deleting task history. + await ensureUniqueIndex(db, "a2a_tasks", "uq_a2a_tasks_active_context", "active_context_key"); await dropIndexIfExists(db, "a2a_tasks", "idx_a2a_tasks_agent"); await dropIndexIfExists(db, "a2a_tasks", "idx_a2a_tasks_context"); // notifications @@ -659,6 +665,7 @@ export async function runPortalMigrations(): Promise { await safeAlterTable(db, "chat_messages", "delegation_id", "VARCHAR(64) DEFAULT NULL"); await safeAlterTable(db, "chat_messages", "target_agent_id", "CHAR(36) DEFAULT NULL"); await safeAlterTable(db, "chat_messages", "trace_id", "CHAR(32) DEFAULT NULL"); + await safeAlterTable(db, "a2a_tasks", "active_context_key", "VARCHAR(64) DEFAULT NULL"); // Widen delegation_id CHAR(36)→VARCHAR(64) on EXISTING deployments. A group reduce child's id // `${toolCallId}#reduce` reaches 36 chars for a 29-char provider id and would overflow CHAR(36)