scitix · LikiosSedo · Jul 22, 2026
diff --git 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/<taskId>:subscribe` (only while the
   task is non-terminal; a terminal task returns `400`). Or just fall back to polling
   `GET .../tasks/<taskId>` — 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
@@ -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
@@ -240,14 +240,19 @@ 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",
       method: "POST",
       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<Array<{ c: number }>>(
+      `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<Array<{ state: string; artifact_text: string }>>(
+      "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<Array<{ state: string; artifact_text: string }>>(
+      "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<Array<{ artifact_text: string }>>(
+      "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({