From d28b0ad3e8f4780184e8bd32852e22013fea829c Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Thu, 23 Jul 2026 09:40:59 +0000 Subject: [PATCH 1/4] feat: add A2A task tools --- CHANGELOG.md | 6 ++ README.md | 8 +- package-lock.json | 4 +- package.json | 2 +- src/tools/a2a.ts | 137 +++++++++++++++++++++++++ src/tools/index.ts | 2 + tests/contract/tool-vocabulary.test.ts | 8 +- tests/unit/a2a.test.ts | 112 ++++++++++++++++++++ 8 files changed, 271 insertions(+), 8 deletions(-) create mode 100644 src/tools/a2a.ts create mode 100644 tests/unit/a2a.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index cf5c0c9..5db4a2a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## 0.1.1 (unreleased) + +- Adds identity-bound A2A 1.0 tools for sending, checking, waiting on, and + replying to remote tasks. Outbound calls and replies use the existing + approval and recipient-allowlist controls. + ## 0.1.0 (unreleased) Initial release. diff --git a/README.md b/README.md index 96213ba..4018436 100644 --- a/README.md +++ b/README.md @@ -111,12 +111,13 @@ groups are enabled. ## Tools -27 tools are enabled by default; 21 more are opt-in (see +30 tools are enabled by default; 21 more are opt-in (see [Enabling more tools](#enabling-more-tools)). Names are stable — treat renames as breaking. | Group | Enabled by default | Opt-in | |---|---|---| +| `a2a` | `inkbox_a2a_call`, `inkbox_a2a_check`, `inkbox_a2a_reply` | — | | `email` | `inkbox_send_email`, `inkbox_list_unread_emails`, `inkbox_list_emails`, `inkbox_get_email`, `inkbox_get_email_thread` | `inkbox_forward_email`, `inkbox_mark_emails_read` | | `sms` | `inkbox_send_sms`, `inkbox_list_text_conversations`, `inkbox_get_text_conversation` | `inkbox_list_texts`, `inkbox_get_text`, `inkbox_mark_text_read`, `inkbox_mark_text_conversation_read` | | `imessage` | `inkbox_send_imessage`, `inkbox_list_imessage_conversations`, `inkbox_get_imessage_conversation` | `inkbox_imessage_triage_number`, `inkbox_list_imessage_assignments`, `inkbox_send_imessage_reaction`, `inkbox_mark_imessage_conversation_read` | @@ -154,10 +155,11 @@ export default async (input: any) => InkboxPlugin(input, { ## Outbound safety -Sends and calls are gated before anything leaves: +Sends, calls, and A2A writes are gated before anything leaves: - **Approval prompts** (default): `inkbox_send_email`, `inkbox_send_sms`, - `inkbox_send_imessage`, `inkbox_forward_email`, and `inkbox_place_call` + `inkbox_send_imessage`, `inkbox_forward_email`, `inkbox_place_call`, + `inkbox_a2a_call`, and `inkbox_a2a_reply` request approval through opencode's native permission system. Approve once, or persist an allow rule from the prompt. - **Recipient allowlist**: set `outbound.allowedRecipients` (exact email diff --git a/package-lock.json b/package-lock.json index 929f057..7534dca 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@inkbox/opencode-plugin", - "version": "0.1.0", + "version": "0.1.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@inkbox/opencode-plugin", - "version": "0.1.0", + "version": "0.1.1", "license": "MIT", "dependencies": { "@inkbox/sdk": ">=0.5.1 <1.0.0", diff --git a/package.json b/package.json index 3bb4504..031aa40 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@inkbox/opencode-plugin", - "version": "0.1.0", + "version": "0.1.1", "private": true, "description": "Inkbox for opencode — give your agent an email address, a phone number (SMS/MMS + voice), iMessage, contacts, notes, and an encrypted credential vault.", "license": "MIT", diff --git a/src/tools/a2a.ts b/src/tools/a2a.ts new file mode 100644 index 0000000..39bebbe --- /dev/null +++ b/src/tools/a2a.ts @@ -0,0 +1,137 @@ +import { z } from "zod"; +import { runTool } from "../errors.js"; +import { formatJson } from "../format.js"; +import { approveOutbound } from "../permissions.js"; +import type { RegisteredTool, ToolDeps } from "./types.js"; + +const callArgs = { + cardUrl: z.string().url().describe("A2A Agent Card URL."), + text: z.string().min(1).describe("Task text."), + contextId: z.string().describe("Optional context to continue.").optional(), + taskId: z.string().describe("Optional task requesting more input.").optional(), + messageId: z.string().describe("Stable idempotency id.").optional(), +}; + +const checkArgs = { + cardUrl: z.string().url().describe("A2A Agent Card URL."), + taskId: z.string().min(1).describe("Remote task id."), + wait: z + .boolean() + .describe("Wait until the task reaches a final or input-required state.") + .optional(), +}; + +const replyArgs = { + cardUrl: z.string().url().describe("A2A Agent Card URL."), + taskId: z.string().min(1).describe("Remote task id."), + text: z.string().min(1).describe("Reply text."), + messageId: z.string().describe("Stable idempotency id.").optional(), +}; + +type CallArgs = z.infer>; +type CheckArgs = z.infer>; +type ReplyArgs = z.infer>; + +async function clientFor(deps: ToolDeps): Promise { + const identity = await deps.runtime.getIdentity(); + const factory = (identity as any).a2aClient; + if (typeof factory !== "function") { + throw new Error("This A2A tool requires @inkbox/sdk with identity.a2aClient() support."); + } + return factory.call(identity); +} + +export function a2aTools(deps: ToolDeps): RegisteredTool[] { + const { config } = deps; + return [ + { + name: "inkbox_a2a_call", + group: "a2a", + defaultEnabled: true, + definition: { + description: + "Send a task to an A2A 1.0 Agent Card. Keep the returned task and context ids for later checks or replies.", + args: callArgs, + async execute(args: CallArgs, ctx) { + return runTool(async () => { + await approveOutbound(ctx, config, { + tool: "inkbox_a2a_call", + recipients: [args.cardUrl], + summary: `Send an A2A task to ${args.cardUrl}`, + metadata: { cardUrl: args.cardUrl }, + }); + const a2a = await clientFor(deps); + try { + const target = await a2a.fetchCard(args.cardUrl); + return formatJson( + await a2a.send(target, { + text: args.text, + contextId: args.contextId, + taskId: args.taskId, + messageId: args.messageId, + }), + ); + } finally { + a2a.close?.(); + } + }); + }, + }, + }, + { + name: "inkbox_a2a_check", + group: "a2a", + defaultEnabled: true, + definition: { + description: "Fetch an A2A task, or wait until it reaches a final or input-required state.", + args: checkArgs, + async execute(args: CheckArgs) { + return runTool(async () => { + const a2a = await clientFor(deps); + try { + const target = await a2a.fetchCard(args.cardUrl); + const task = args.wait + ? await a2a.wait(target, args.taskId) + : await a2a.getTask(target, args.taskId); + return formatJson(task); + } finally { + a2a.close?.(); + } + }); + }, + }, + }, + { + name: "inkbox_a2a_reply", + group: "a2a", + defaultEnabled: true, + definition: { + description: "Reply to a remote A2A task that requested more input.", + args: replyArgs, + async execute(args: ReplyArgs, ctx) { + return runTool(async () => { + await approveOutbound(ctx, config, { + tool: "inkbox_a2a_reply", + recipients: [args.cardUrl], + summary: `Reply to A2A task ${args.taskId} at ${args.cardUrl}`, + metadata: { cardUrl: args.cardUrl, taskId: args.taskId }, + }); + const a2a = await clientFor(deps); + try { + const target = await a2a.fetchCard(args.cardUrl); + return formatJson( + await a2a.send(target, { + taskId: args.taskId, + text: args.text, + messageId: args.messageId, + }), + ); + } finally { + a2a.close?.(); + } + }); + }, + }, + }, + ]; +} diff --git a/src/tools/index.ts b/src/tools/index.ts index 70233f2..b0dbead 100644 --- a/src/tools/index.ts +++ b/src/tools/index.ts @@ -1,4 +1,5 @@ import type { ToolDefinition } from "@opencode-ai/plugin"; +import { a2aTools } from "./a2a.js"; import { accessTools } from "./access.js"; import { callReadTools } from "./call-reads.js"; import { contactRuleTools } from "./contact-rules.js"; @@ -23,6 +24,7 @@ const EMPTY_GATING: GatingSummary = { enabled: [], disabledByDefault: [], groups function buildGroups(deps: ToolDeps, getGating: () => GatingSummary): RegisteredTool[] { return [ + ...a2aTools(deps), ...sendEmailTools(deps), ...forwardEmailTools(deps), ...emailReadTools(deps), diff --git a/tests/contract/tool-vocabulary.test.ts b/tests/contract/tool-vocabulary.test.ts index a36f88b..53778b7 100644 --- a/tests/contract/tool-vocabulary.test.ts +++ b/tests/contract/tool-vocabulary.test.ts @@ -32,6 +32,9 @@ function stubDeps(): ToolDeps { } const DEFAULT_ENABLED = [ + "inkbox_a2a_call", + "inkbox_a2a_check", + "inkbox_a2a_reply", "inkbox_send_email", "inkbox_send_sms", "inkbox_send_imessage", @@ -93,6 +96,7 @@ const SENSITIVE = [ ].sort(); const GROUPS = [ + "a2a", "email", "sms", "imessage", @@ -108,10 +112,10 @@ const GROUPS = [ describe("tool vocabulary", () => { const all = buildAllTools(stubDeps()); - it("ships exactly the expected 48 tools", () => { + it("ships exactly the expected 51 tools", () => { const names = all.map((t) => t.name).sort(); expect(names).toEqual([...DEFAULT_ENABLED, ...OPT_IN].sort()); - expect(names).toHaveLength(48); + expect(names).toHaveLength(51); }); it("has no duplicate tool names", () => { diff --git a/tests/unit/a2a.test.ts b/tests/unit/a2a.test.ts new file mode 100644 index 0000000..0f51f57 --- /dev/null +++ b/tests/unit/a2a.test.ts @@ -0,0 +1,112 @@ +import { describe, expect, it, vi } from "vitest"; +import { defaultGatewayConfig } from "../../src/config.js"; +import { a2aTools } from "../../src/tools/a2a.js"; + +function makeCtx() { + return { + ask: vi.fn(async () => {}), + abort: new AbortController().signal, + } as any; +} + +function makeDeps() { + const a2a = { + fetchCard: vi.fn(async (url: string) => ({ rpcUrl: `${url}/rpc` })), + send: vi.fn(async () => ({ kind: "task", task: { id: "task-1" } })), + getTask: vi.fn(async () => ({ id: "task-1", status: { state: "TASK_STATE_WORKING" } })), + wait: vi.fn(async () => ({ id: "task-1", status: { state: "TASK_STATE_COMPLETED" } })), + close: vi.fn(), + }; + const identity = { a2aClient: vi.fn(async () => a2a) }; + return { + a2a, + deps: { + runtime: { + getIdentity: vi.fn(async () => identity), + getClient: vi.fn(async () => ({})), + }, + config: { + apiKey: "key", + identity: "agent", + vaultKeyEnvVar: "INKBOX_VAULT_KEY", + tools: { enable: [], disable: [] }, + outbound: { allowedRecipients: [], approval: "ask", askTimeoutMs: 0 }, + gateway: defaultGatewayConfig(), + }, + vault: { keyEnvVar: "INKBOX_VAULT_KEY", getCredentials: vi.fn() }, + } as any, + }; +} + +function getTool(name: string, deps: any) { + const tool = a2aTools(deps).find((item) => item.name === name); + if (!tool) throw new Error(`missing ${name}`); + return tool; +} + +describe("a2aTools", () => { + it("sends a task behind the outbound approval gate", async () => { + const { a2a, deps } = makeDeps(); + const ctx = makeCtx(); + + const result = await getTool("inkbox_a2a_call", deps).definition.execute( + { + cardUrl: "https://target.example/card", + text: "Investigate.", + messageId: "msg-1", + }, + ctx, + ); + + expect(ctx.ask).toHaveBeenCalledWith( + expect.objectContaining({ + permission: "inkbox_a2a_call", + patterns: ["https://target.example/card"], + }), + ); + expect(a2a.send).toHaveBeenCalledWith( + { rpcUrl: "https://target.example/card/rpc" }, + expect.objectContaining({ text: "Investigate.", messageId: "msg-1" }), + ); + expect(result).toContain('"task-1"'); + expect(a2a.close).toHaveBeenCalledTimes(1); + }); + + it("waits for a task without requesting outbound approval", async () => { + const { a2a, deps } = makeDeps(); + const ctx = makeCtx(); + + const result = await getTool("inkbox_a2a_check", deps).definition.execute( + { + cardUrl: "https://target.example/card", + taskId: "task-1", + wait: true, + }, + ctx, + ); + + expect(ctx.ask).not.toHaveBeenCalled(); + expect(a2a.wait).toHaveBeenCalledWith({ rpcUrl: "https://target.example/card/rpc" }, "task-1"); + expect(result).toContain("TASK_STATE_COMPLETED"); + }); + + it("replies to an input-required task behind approval", async () => { + const { a2a, deps } = makeDeps(); + const ctx = makeCtx(); + + await getTool("inkbox_a2a_reply", deps).definition.execute( + { + cardUrl: "https://target.example/card", + taskId: "task-1", + text: "More context.", + }, + ctx, + ); + + expect(ctx.ask).toHaveBeenCalledTimes(1); + expect(a2a.send).toHaveBeenCalledWith( + { rpcUrl: "https://target.example/card/rpc" }, + expect.objectContaining({ taskId: "task-1", text: "More context." }), + ); + }); +}); From 64d593206ec6473fcf0d6b85d237db044d5e7690 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Thu, 23 Jul 2026 10:31:45 +0000 Subject: [PATCH 2/4] feat: serve and track A2A delegations --- CHANGELOG.md | 9 +- README.md | 4 +- src/a2a-context.ts | 20 +++ src/a2a-delegations.ts | 90 ++++++++++ src/gateway/a2a.ts | 230 +++++++++++++++++++++++++ src/gateway/index.ts | 9 + src/gateway/sessions.ts | 52 ++++++ src/gateway/subscriptions.ts | 15 +- src/gateway/types.ts | 3 + src/tools/a2a.ts | 125 ++++++++++++-- tests/contract/tool-vocabulary.test.ts | 7 +- tests/gateway/a2a.test.ts | 125 ++++++++++++++ tests/gateway/commands.test.ts | 2 + tests/gateway/dispatch.test.ts | 2 + tests/gateway/subscriptions.test.ts | 25 ++- tests/unit/a2a.test.ts | 52 +++++- 16 files changed, 734 insertions(+), 36 deletions(-) create mode 100644 src/a2a-context.ts create mode 100644 src/a2a-delegations.ts create mode 100644 src/gateway/a2a.ts create mode 100644 tests/gateway/a2a.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 5db4a2a..f3d36fe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,9 +2,12 @@ ## 0.1.1 (unreleased) -- Adds identity-bound A2A 1.0 tools for sending, checking, waiting on, and - replying to remote tasks. Outbound calls and replies use the existing - approval and recipient-allowlist controls. +- Adds identity-bound A2A 1.0 client tools plus durable inbound task serving: + context-scoped sessions, restart catch-up, task-addressed cancellation, and + explicit complete/ask/fail intents. Outbound calls and replies use the + existing approval and recipient-allowlist controls. +- A2A features require `@inkbox/sdk` 0.5.5 or newer. The plugin keeps its + existing base dependency range for installations that do not use A2A. ## 0.1.0 (unreleased) diff --git a/README.md b/README.md index 4018436..21a1922 100644 --- a/README.md +++ b/README.md @@ -111,13 +111,13 @@ groups are enabled. ## Tools -30 tools are enabled by default; 21 more are opt-in (see +33 tools are enabled by default; 21 more are opt-in (see [Enabling more tools](#enabling-more-tools)). Names are stable — treat renames as breaking. | Group | Enabled by default | Opt-in | |---|---|---| -| `a2a` | `inkbox_a2a_call`, `inkbox_a2a_check`, `inkbox_a2a_reply` | — | +| `a2a` | `inkbox_a2a_call`, `inkbox_a2a_check`, `inkbox_a2a_reply`, `inkbox_a2a_complete`, `inkbox_a2a_ask_caller`, `inkbox_a2a_fail` | — | | `email` | `inkbox_send_email`, `inkbox_list_unread_emails`, `inkbox_list_emails`, `inkbox_get_email`, `inkbox_get_email_thread` | `inkbox_forward_email`, `inkbox_mark_emails_read` | | `sms` | `inkbox_send_sms`, `inkbox_list_text_conversations`, `inkbox_get_text_conversation` | `inkbox_list_texts`, `inkbox_get_text`, `inkbox_mark_text_read`, `inkbox_mark_text_conversation_read` | | `imessage` | `inkbox_send_imessage`, `inkbox_list_imessage_conversations`, `inkbox_get_imessage_conversation` | `inkbox_imessage_triage_number`, `inkbox_list_imessage_assignments`, `inkbox_send_imessage_reaction`, `inkbox_mark_imessage_conversation_read` | diff --git a/src/a2a-context.ts b/src/a2a-context.ts new file mode 100644 index 0000000..8764742 --- /dev/null +++ b/src/a2a-context.ts @@ -0,0 +1,20 @@ +export interface ActiveA2ATurn { + taskId: string; + messageId: string; + contextId: string; + replyIntentCommitted: boolean; +} + +const turns = new Map(); + +export function setActiveA2ATurn(sessionID: string, turn: ActiveA2ATurn): void { + turns.set(sessionID, turn); +} + +export function clearActiveA2ATurn(sessionID: string, turn: ActiveA2ATurn): void { + if (turns.get(sessionID) === turn) turns.delete(sessionID); +} + +export function activeA2ATurn(sessionID: string): ActiveA2ATurn | undefined { + return turns.get(sessionID); +} diff --git a/src/a2a-delegations.ts b/src/a2a-delegations.ts new file mode 100644 index 0000000..f2f148f --- /dev/null +++ b/src/a2a-delegations.ts @@ -0,0 +1,90 @@ +import * as fs from "node:fs"; +import * as path from "node:path"; +import { gatewayHome } from "./gateway/state.js"; + +export interface A2ADelegationRecord { + identityId: string; + origin: string; + cardUrl: string; + contextId?: string; + taskId?: string; + messageId: string; + sessionId?: string; + updatedAt: number; +} + +type Records = Record; + +function filePath(): string { + return path.join(gatewayHome(), "a2a-delegations.json"); +} + +function origin(url: string): string { + return new URL(url).origin.toLowerCase(); +} + +function read(): Records { + try { + const loaded = JSON.parse(fs.readFileSync(filePath(), "utf8")); + return loaded && typeof loaded === "object" ? loaded : {}; + } catch (error: any) { + if (error?.code === "ENOENT") return {}; + throw error; + } +} + +function write(records: Records): void { + const target = filePath(); + fs.mkdirSync(path.dirname(target), { recursive: true }); + const tmp = `${target}.${process.pid}.tmp`; + fs.writeFileSync(tmp, `${JSON.stringify(records, null, 2)}\n`, { + mode: 0o600, + }); + fs.renameSync(tmp, target); + fs.chmodSync(target, 0o600); +} + +export function recordBeforeSend(input: { + identityId: string; + rpcUrl: string; + cardUrl: string; + contextId?: string; + taskId?: string; + messageId: string; + sessionId?: string; +}): string { + const resolvedOrigin = origin(input.rpcUrl); + const contextKey = input.contextId ?? `pending:${input.messageId}`; + const key = `${input.identityId}|${resolvedOrigin}|${contextKey}`; + const records = read(); + records[key] = { + identityId: input.identityId, + origin: resolvedOrigin, + cardUrl: input.cardUrl, + contextId: input.contextId, + taskId: input.taskId, + messageId: input.messageId, + sessionId: input.sessionId, + updatedAt: Date.now(), + }; + write(records); + return key; +} + +export function promoteAfterSend(pendingKey: string, contextId: string, taskId: string): void { + const records = read(); + const record = records[pendingKey]; + if (!record) return; + delete records[pendingKey]; + record.contextId = contextId; + record.taskId = taskId; + record.updatedAt = Date.now(); + records[`${record.identityId}|${record.origin}|${contextId}`] = record; + write(records); +} + +export function findDelegationByTask(taskId: string): A2ADelegationRecord | undefined { + return Object.values(read()) + .filter((record) => record.taskId === taskId) + .sort((a, b) => b.updatedAt - a.updatedAt)[0]; +} diff --git a/src/gateway/a2a.ts b/src/gateway/a2a.ts new file mode 100644 index 0000000..2f88c68 --- /dev/null +++ b/src/gateway/a2a.ts @@ -0,0 +1,230 @@ +import type { ActiveA2ATurn } from "../a2a-context.js"; +import { findDelegationByTask } from "../a2a-delegations.js"; +import type { InkboxRuntime } from "../client.js"; +import type { StateStore } from "./state.js"; +import type { GatewayLogger, SessionManager, VerifiedEvent } from "./types.js"; + +const TERMINAL = new Set(["completed", "failed", "canceled", "rejected"]); + +interface A2AEventData { + task_id: string; + context_id: string; + state?: string; + message_id?: string; + caller?: { + identity_id?: string; + organization_id?: string; + handle?: string; + }; + parts?: Array>; +} + +interface RegistryEntry { + taskId: string; + contextId: string; + messageId: string; + state: "queued" | "running" | "finalized"; + data: A2AEventData; + updatedAt: number; +} + +export interface A2AHandler { + handles(event: VerifiedEvent): boolean; + handle(event: VerifiedEvent): Promise; + catchUp(): Promise; +} + +function eventData(event: VerifiedEvent): A2AEventData | undefined { + const data = event.body.data; + if (!data || typeof data !== "object" || Array.isArray(data)) return undefined; + const value = data as A2AEventData; + return value.task_id && value.context_id ? value : undefined; +} + +function registry(state: StateStore): Record { + const value = state.read().a2aTasks; + return value && typeof value === "object" ? (value as Record) : {}; +} + +function persist( + state: StateStore, + key: string, + data: A2AEventData, + status: RegistryEntry["state"], +): void { + state.update({ + a2aTasks: { + ...registry(state), + [key]: { + taskId: data.task_id, + contextId: data.context_id, + messageId: data.message_id ?? "", + state: status, + data, + updatedAt: Date.now(), + }, + }, + }); +} + +export function createA2AHandler(deps: { + inkbox: InkboxRuntime; + sessions: SessionManager; + state: StateStore; + logger: GatewayLogger; +}): A2AHandler { + const running = new Map>>(); + + async function identity(): Promise { + return deps.inkbox.getIdentity() as Promise; + } + + async function run(key: string, data: A2AEventData): Promise { + const id = await identity(); + const taskId = data.task_id; + const chatKey = `a2a:${id.id}:${data.context_id}`; + const context: ActiveA2ATurn = { + taskId, + contextId: data.context_id, + messageId: data.message_id ?? "", + replyIntentCommitted: false, + }; + const caller = data.caller ?? {}; + const body = (data.parts ?? []) + .map((part) => (typeof part.text === "string" ? part.text : "")) + .filter(Boolean) + .join("\n"); + const marker = + `[inkbox:a2a_task caller=@${String(caller.handle ?? "unknown").replace(/^@/, "")} ` + + `caller_org=${caller.organization_id ?? "unknown"}]`; + persist(deps.state, key, data, "running"); + try { + const reply = await deps.sessions.runA2A(chatKey, `${marker}\n${body}`.trim(), context); + if ( + !context.replyIntentCommitted && + reply?.trim() && + reply.trim().toUpperCase() !== "[SILENT]" + ) { + const task = await id.a2aTask(taskId); + if (!TERMINAL.has(String(task.state))) { + await id.a2aReply(taskId, { intent: "complete", text: reply }); + } + } + persist(deps.state, key, data, "finalized"); + } catch (error) { + deps.logger.error("a2a.turn_failed", { taskId, error: String(error) }); + } + } + + function start(key: string, data: A2AEventData): void { + const job = run(key, data); + const jobs = running.get(data.task_id) ?? new Set>(); + jobs.add(job); + running.set(data.task_id, jobs); + void job.finally(() => { + jobs.delete(job); + if (jobs.size === 0) running.delete(data.task_id); + }); + } + + return { + handles(event) { + return event.provider === "inkbox" && event.eventType?.startsWith("a2a.") === true; + }, + + async handle(event) { + const type = event.eventType ?? ""; + const data = eventData(event); + if (!data) return true; + if (type === "a2a.sent_task.updated") { + const delegation = findDelegationByTask(data.task_id); + const chatKey = delegation?.sessionId + ? Object.entries(deps.state.read().sessions).find( + ([, sessionId]) => sessionId === delegation.sessionId, + )?.[0] + : undefined; + if (chatKey) { + const text = (data.parts ?? []) + .map((part) => (typeof part.text === "string" ? part.text : "")) + .filter(Boolean) + .join("\n"); + await deps.sessions.runCapture( + chatKey, + `[inkbox:a2a_sent_task_updated task_id=${data.task_id} ` + + `context_id=${data.context_id} state=${data.state ?? "unknown"}]\n` + + "An A2A task you delegated changed state. Use " + + "inkbox_a2a_check or inkbox_a2a_reply with the stored Agent Card " + + `URL ${delegation?.cardUrl ?? "unknown"} if follow-up is needed.` + + (text ? `\n\nRemote agent message:\n${text}` : ""), + ); + } else { + deps.logger.info("a2a.sent_task_updated_without_session", { + taskId: data.task_id, + }); + } + return true; + } + if (type === "a2a.task.canceled") { + const id = await identity(); + await deps.sessions.abortA2A(`a2a:${id.id}:${data.context_id}`, data.task_id); + return true; + } + const messageId = data.message_id ?? event.body.id?.toString() ?? ""; + const key = `${data.task_id}:${messageId}`; + if (registry(deps.state)[key]) return true; + const normalized = { ...data, message_id: messageId }; + persist(deps.state, key, normalized, "queued"); + start(key, normalized); + return true; + }, + + async catchUp() { + const id = await identity(); + if ( + typeof id.a2aTask !== "function" || + typeof id.iterA2ATasks !== "function" || + typeof id.a2aReply !== "function" + ) { + deps.logger.warn("a2a.sdk_upgrade_required", { + requiredVersion: "0.5.5", + }); + return; + } + for (const [key, entry] of Object.entries(registry(deps.state))) { + if (entry.state === "finalized") continue; + try { + const task = await id.a2aTask(entry.taskId); + if (TERMINAL.has(String(task.state))) { + persist(deps.state, key, entry.data, "finalized"); + } else { + start(key, entry.data); + } + } catch (error) { + deps.logger.warn("a2a.registry_reconcile_failed", { + taskId: entry.taskId, + error: String(error), + }); + } + } + for await (const task of id.iterA2ATasks({ state: "submitted" })) { + const message = task.messages.at(-1); + const data: A2AEventData = { + task_id: String(task.id), + context_id: String(task.contextId), + state: String(task.state), + caller: { + identity_id: String(task.caller.identityId), + organization_id: task.caller.organizationId, + handle: task.caller.handle, + }, + message_id: message?.messageId ?? `task:${task.id}`, + parts: message?.parts ?? [], + }; + const key = `${data.task_id}:${data.message_id}`; + if (registry(deps.state)[key]) continue; + persist(deps.state, key, data, "queued"); + start(key, data); + } + }, + }; +} diff --git a/src/gateway/index.ts b/src/gateway/index.ts index 492ced6..d5fe928 100644 --- a/src/gateway/index.ts +++ b/src/gateway/index.ts @@ -1,6 +1,7 @@ import type { OpencodeClient } from "@opencode-ai/sdk"; import type { InkboxRuntime } from "../client.js"; import type { ResolvedConfig } from "../config.js"; +import { createA2AHandler } from "./a2a.js"; import { createBurstBuffer } from "./burst.js"; import { handleCommand } from "./commands.js"; import { createContactResolver } from "./contacts.js"; @@ -70,6 +71,12 @@ export async function startGateway(opts: StartGatewayOptions): Promise {}); @@ -162,6 +170,7 @@ export async function startGateway(opts: StartGatewayOptions): Promise { + if (a2a.handles(event)) return a2a.handle(event); return dispatchEvent( { config: opts.config, diff --git a/src/gateway/sessions.ts b/src/gateway/sessions.ts index 603b6f5..1ff0374 100644 --- a/src/gateway/sessions.ts +++ b/src/gateway/sessions.ts @@ -1,4 +1,5 @@ import type { OpencodeClient } from "@opencode-ai/sdk"; +import { type ActiveA2ATurn, clearActiveA2ATurn, setActiveA2ATurn } from "../a2a-context.js"; import type { InkboxRuntime } from "../client.js"; import type { ResolvedConfig } from "../config.js"; import { buildIdentitySystem, frameCapture, frameInbound } from "./prompts.js"; @@ -22,6 +23,7 @@ interface QueuedTurn { // True for a follow-up turn enqueued after a delivery failure, so a second // failure doesn't spawn another recovery (bounded to one attempt). recovered?: boolean; + a2aContext?: ActiveA2ATurn; resolve: (out: string | undefined) => void; reject: (err: unknown) => void; } @@ -34,6 +36,7 @@ interface PerKey { runningKind?: TurnKind; // Set to interrupt the in-flight normal turn so its partial output is dropped. interruptNormal: boolean; + runningA2ATaskId?: string; } export interface SessionManagerDeps { @@ -166,6 +169,10 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager { if (turn.kind === "normal") entry.interruptNormal = false; try { const sessionID = await ensureSession(chatKey); + if (turn.a2aContext) { + entry.runningA2ATaskId = turn.a2aContext.taskId; + setActiveA2ATurn(sessionID, turn.a2aContext); + } let out: string | undefined; try { out = await runPrompt(sessionID, turn.text, turn.agent); @@ -207,6 +214,12 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager { } catch (err) { deps.logger.error("turn.failed", { chatKey, error: String(err) }); turn.reject(err); + } finally { + const sessionID = deps.state.getSession(chatKey); + if (sessionID && turn.a2aContext) { + clearActiveA2ATurn(sessionID, turn.a2aContext); + } + entry.runningA2ATaskId = undefined; } } } finally { @@ -287,6 +300,45 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager { }); }, + async runA2A(chatKey, framedText, context) { + if (closing) return undefined; + const entry = per(chatKey); + return new Promise((resolve, reject) => { + entry.queue.push({ + kind: "capture", + text: framedText, + deliver: false, + a2aContext: context, + resolve, + reject, + }); + void drain(chatKey); + }); + }, + + async abortA2A(chatKey, taskId) { + const entry = keys.get(chatKey); + if (!entry) return false; + const kept: QueuedTurn[] = []; + let removed = false; + for (const turn of entry.queue) { + if (turn.a2aContext?.taskId === taskId) { + turn.resolve(undefined); + removed = true; + } else { + kept.push(turn); + } + } + entry.queue = kept; + if (entry.runningA2ATaskId !== taskId) return removed; + const sessionID = deps.state.getSession(chatKey); + if (!sessionID) return removed; + await deps.opencode.session + .abort({ path: { id: sessionID }, query: { directory: deps.directory } }) + .catch(() => {}); + return true; + }, + async resetSession(chatKey) { // Abort any in-flight turn first so a pending escalation on the old // session doesn't get orphaned by the mapping being cleared. diff --git a/src/gateway/subscriptions.ts b/src/gateway/subscriptions.ts index f52f666..621c39b 100644 --- a/src/gateway/subscriptions.ts +++ b/src/gateway/subscriptions.ts @@ -19,6 +19,13 @@ export const IMESSAGE_EVENT_TYPES = [ "imessage.reaction_received", "imessage.delivery_failed", ]; +export const A2A_EVENT_TYPES = [ + "a2a.task.created", + "a2a.task.message", + "a2a.task.canceled", + "a2a.sent_task.updated", +]; +export const IDENTITY_EVENT_TYPES = [...IMESSAGE_EVENT_TYPES, ...A2A_EVENT_TYPES]; export interface ReconcileResult { created: number; @@ -121,9 +128,11 @@ export async function reconcileSubscriptions( if (identity.phoneNumber) { await reconcileOwner("phone", { phoneNumberId: identity.phoneNumber.id }, PHONE_EVENT_TYPES); } - if (identity.imessageEnabled) { - await reconcileOwner("imessage", { agentIdentityId: identity.id }, IMESSAGE_EVENT_TYPES); - } + await reconcileOwner( + "identity", + { agentIdentityId: identity.id }, + identity.imessageEnabled ? IDENTITY_EVENT_TYPES : A2A_EVENT_TYPES, + ); if (deps.config.gateway.voice.enabled) { await wireIncomingCalls(deps, identity, base, webhookUrl); diff --git a/src/gateway/types.ts b/src/gateway/types.ts index 38940fe..3b71393 100644 --- a/src/gateway/types.ts +++ b/src/gateway/types.ts @@ -1,4 +1,5 @@ import type { OpencodeClient } from "@opencode-ai/sdk"; +import type { ActiveA2ATurn } from "../a2a-context.js"; import type { InkboxRuntime } from "../client.js"; import type { ResolvedConfig } from "../config.js"; import type { StateStore } from "./state.js"; @@ -103,6 +104,8 @@ export interface SessionManager { // Run an already-framed turn and return the assistant text without // delivering it anywhere (used by the voice bridge to speak the reply). runText(chatKey: string, framedText: string): Promise; + runA2A(chatKey: string, framedText: string, context: ActiveA2ATurn): Promise; + abortA2A(chatKey: string, taskId: string): Promise; // Control-command support. resetSession(chatKey: string): Promise; abortTurn(chatKey: string): Promise; diff --git a/src/tools/a2a.ts b/src/tools/a2a.ts index 39bebbe..a8eb6ee 100644 --- a/src/tools/a2a.ts +++ b/src/tools/a2a.ts @@ -1,4 +1,6 @@ import { z } from "zod"; +import { activeA2ATurn } from "../a2a-context.js"; +import { findDelegationByTask, promoteAfterSend, recordBeforeSend } from "../a2a-delegations.js"; import { runTool } from "../errors.js"; import { formatJson } from "../format.js"; import { approveOutbound } from "../permissions.js"; @@ -27,10 +29,18 @@ const replyArgs = { text: z.string().min(1).describe("Reply text."), messageId: z.string().describe("Stable idempotency id.").optional(), }; +const completeArgs = { + text: z.string().min(1).describe("Final answer."), +}; +const failArgs = { + reason: z.string().min(1).describe("Failure reason."), +}; type CallArgs = z.infer>; type CheckArgs = z.infer>; type ReplyArgs = z.infer>; +type CompleteArgs = z.infer>; +type FailArgs = z.infer>; async function clientFor(deps: ToolDeps): Promise { const identity = await deps.runtime.getIdentity(); @@ -62,15 +72,28 @@ export function a2aTools(deps: ToolDeps): RegisteredTool[] { }); const a2a = await clientFor(deps); try { + const identity = await deps.runtime.getIdentity(); const target = await a2a.fetchCard(args.cardUrl); - return formatJson( - await a2a.send(target, { - text: args.text, - contextId: args.contextId, - taskId: args.taskId, - messageId: args.messageId, - }), - ); + const messageId = args.messageId ?? crypto.randomUUID(); + const pendingKey = recordBeforeSend({ + identityId: String(identity.id), + rpcUrl: String(target.rpcUrl), + cardUrl: args.cardUrl, + contextId: args.contextId, + taskId: args.taskId, + messageId, + sessionId: ctx.sessionID, + }); + const result = await a2a.send(target, { + text: args.text, + contextId: args.contextId, + taskId: args.taskId, + messageId, + }); + if (result.task?.id && result.task?.contextId) { + promoteAfterSend(pendingKey, String(result.task.contextId), String(result.task.id)); + } + return formatJson(result); } finally { a2a.close?.(); } @@ -118,14 +141,28 @@ export function a2aTools(deps: ToolDeps): RegisteredTool[] { }); const a2a = await clientFor(deps); try { + const identity = await deps.runtime.getIdentity(); const target = await a2a.fetchCard(args.cardUrl); - return formatJson( - await a2a.send(target, { - taskId: args.taskId, - text: args.text, - messageId: args.messageId, - }), - ); + const existing = findDelegationByTask(args.taskId); + const messageId = args.messageId ?? crypto.randomUUID(); + const pendingKey = recordBeforeSend({ + identityId: String(identity.id), + rpcUrl: String(target.rpcUrl), + cardUrl: args.cardUrl, + contextId: existing?.contextId, + taskId: args.taskId, + messageId, + sessionId: ctx.sessionID ?? existing?.sessionId, + }); + const result = await a2a.send(target, { + taskId: args.taskId, + text: args.text, + messageId, + }); + if (result.task?.contextId) { + promoteAfterSend(pendingKey, String(result.task.contextId), args.taskId); + } + return formatJson(result); } finally { a2a.close?.(); } @@ -133,5 +170,63 @@ export function a2aTools(deps: ToolDeps): RegisteredTool[] { }, }, }, + { + name: "inkbox_a2a_complete", + group: "a2a", + defaultEnabled: true, + definition: { + description: "Complete the active inbound A2A task with a final answer.", + args: completeArgs, + async execute(args: CompleteArgs, ctx) { + return inboundIntent(deps, ctx.sessionID, "complete", args.text); + }, + }, + }, + { + name: "inkbox_a2a_ask_caller", + group: "a2a", + defaultEnabled: true, + definition: { + description: "Ask the caller for more input on the active inbound A2A task.", + args: completeArgs, + async execute(args: CompleteArgs, ctx) { + return inboundIntent(deps, ctx.sessionID, "ask_caller", args.text); + }, + }, + }, + { + name: "inkbox_a2a_fail", + group: "a2a", + defaultEnabled: true, + definition: { + description: "Fail the active inbound A2A task with a reason.", + args: failArgs, + async execute(args: FailArgs, ctx) { + return inboundIntent(deps, ctx.sessionID, "fail", args.reason); + }, + }, + }, ]; } + +async function inboundIntent( + deps: ToolDeps, + sessionID: string, + intent: "complete" | "ask_caller" | "fail", + text: string, +): Promise { + return runTool(async () => { + const context = activeA2ATurn(sessionID); + if (!context) { + throw new Error("This tool is only available during an inbound A2A task"); + } + const identity = await deps.runtime.getIdentity(); + const reply = (identity as any).a2aReply; + if (typeof reply !== "function") { + throw new Error("This A2A tool requires @inkbox/sdk with identity.a2aReply() support."); + } + const result = await reply.call(identity, context.taskId, { intent, text }); + context.replyIntentCommitted = true; + return formatJson(result); + }); +} diff --git a/tests/contract/tool-vocabulary.test.ts b/tests/contract/tool-vocabulary.test.ts index 53778b7..cf41b08 100644 --- a/tests/contract/tool-vocabulary.test.ts +++ b/tests/contract/tool-vocabulary.test.ts @@ -35,6 +35,9 @@ const DEFAULT_ENABLED = [ "inkbox_a2a_call", "inkbox_a2a_check", "inkbox_a2a_reply", + "inkbox_a2a_complete", + "inkbox_a2a_ask_caller", + "inkbox_a2a_fail", "inkbox_send_email", "inkbox_send_sms", "inkbox_send_imessage", @@ -112,10 +115,10 @@ const GROUPS = [ describe("tool vocabulary", () => { const all = buildAllTools(stubDeps()); - it("ships exactly the expected 51 tools", () => { + it("ships exactly the expected 54 tools", () => { const names = all.map((t) => t.name).sort(); expect(names).toEqual([...DEFAULT_ENABLED, ...OPT_IN].sort()); - expect(names).toHaveLength(51); + expect(names).toHaveLength(54); }); it("has no duplicate tool names", () => { diff --git a/tests/gateway/a2a.test.ts b/tests/gateway/a2a.test.ts new file mode 100644 index 0000000..91c28b4 --- /dev/null +++ b/tests/gateway/a2a.test.ts @@ -0,0 +1,125 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { promoteAfterSend, recordBeforeSend } from "../../src/a2a-delegations.js"; +import { createA2AHandler } from "../../src/gateway/a2a.js"; +import { createStateStore } from "../../src/gateway/state.js"; + +function event() { + return { + provider: "inkbox", + verified: true, + eventType: "a2a.task.created", + body: { + id: "evt-1", + data: { + task_id: "task-1", + context_id: "context-1", + state: "submitted", + message_id: "message-1", + caller: { + identity_id: "caller-1", + organization_id: "org-1", + handle: "caller", + }, + parts: [{ text: "Investigate." }], + }, + }, + headers: {}, + }; +} + +describe("createA2AHandler", () => { + beforeEach(() => { + process.env.INKBOX_OPENCODE_HOME = `${process.env.TMPDIR ?? "/tmp"}/opencode-a2a-gateway-${crypto.randomUUID()}`; + }); + + it("persists before ack, dedupes, and guarded-completes", async () => { + const state = createStateStore( + `${process.env.TMPDIR ?? "/tmp"}/opencode-a2a-${crypto.randomUUID()}`, + ); + const a2aReply = vi.fn(async () => ({ id: "task-1", state: "completed" })); + const identity = { + id: "identity-1", + a2aTask: vi.fn(async () => ({ id: "task-1", state: "submitted" })), + a2aReply, + }; + const sessions = { + runA2A: vi.fn(async () => "Completed."), + abortA2A: vi.fn(async () => true), + }; + const handler = createA2AHandler({ + inkbox: { + getIdentity: vi.fn(async () => identity), + getClient: vi.fn(), + } as any, + sessions: sessions as any, + state, + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }); + + expect(await handler.handle(event())).toBe(true); + expect((state.read().a2aTasks as any)["task-1:message-1"]).toBeDefined(); + expect(await handler.handle(event())).toBe(true); + await vi.waitFor(() => { + expect(a2aReply).toHaveBeenCalledWith("task-1", { + intent: "complete", + text: "Completed.", + }); + }); + expect(sessions.runA2A).toHaveBeenCalledTimes(1); + expect((state.read().a2aTasks as any)["task-1:message-1"].state).toBe("finalized"); + }); + + it("cancels only the addressed task on its context session", async () => { + const abortA2A = vi.fn(async () => true); + const handler = createA2AHandler({ + inkbox: { + getIdentity: vi.fn(async () => ({ id: "identity-1" })), + getClient: vi.fn(), + } as any, + sessions: { abortA2A } as any, + state: createStateStore( + `${process.env.TMPDIR ?? "/tmp"}/opencode-a2a-${crypto.randomUUID()}`, + ), + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }); + const canceled = event(); + canceled.eventType = "a2a.task.canceled"; + + await handler.handle(canceled); + + expect(abortA2A).toHaveBeenCalledWith("a2a:identity-1:context-1", "task-1"); + }); + + it("injects sent-task updates into the delegating session", async () => { + const state = createStateStore( + `${process.env.TMPDIR ?? "/tmp"}/opencode-a2a-${crypto.randomUUID()}`, + ); + state.setSession("contact-1", "session-1"); + const key = recordBeforeSend({ + identityId: "identity-1", + rpcUrl: "https://target.example/a2a", + cardUrl: "https://target.example/card", + messageId: "message-1", + sessionId: "session-1", + }); + promoteAfterSend(key, "context-1", "task-1"); + const runCapture = vi.fn(async () => "Handled."); + const handler = createA2AHandler({ + inkbox: { + getIdentity: vi.fn(async () => ({ id: "identity-1" })), + getClient: vi.fn(), + } as any, + sessions: { runCapture } as any, + state, + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }); + const updated = event(); + updated.eventType = "a2a.sent_task.updated"; + updated.body.data.state = "input_required"; + updated.body.data.parts = [{ text: "Which region?" }]; + + await handler.handle(updated); + + expect(runCapture).toHaveBeenCalledWith("contact-1", expect.stringContaining("Which region?")); + }); +}); diff --git a/tests/gateway/commands.test.ts b/tests/gateway/commands.test.ts index ebdaeea..0d6772a 100644 --- a/tests/gateway/commands.test.ts +++ b/tests/gateway/commands.test.ts @@ -9,6 +9,8 @@ function makeSessions(over: Record = {}) { handleInbound: vi.fn(async () => {}), runCapture: vi.fn(async () => undefined), runText: vi.fn(async () => undefined), + runA2A: vi.fn(async () => undefined), + abortA2A: vi.fn(async () => false), resetSession: vi.fn(async () => {}), abortTurn: vi.fn(async () => true), status: vi.fn(() => ({ busy: false, sessionID: undefined as string | undefined })), diff --git a/tests/gateway/dispatch.test.ts b/tests/gateway/dispatch.test.ts index d46f254..6273792 100644 --- a/tests/gateway/dispatch.test.ts +++ b/tests/gateway/dispatch.test.ts @@ -37,6 +37,8 @@ function makeDeps(over: Partial = {}): DispatchDeps { handleInbound: vi.fn(async () => {}), runCapture: vi.fn(async () => undefined), runText: vi.fn(async () => undefined), + runA2A: vi.fn(async () => undefined), + abortA2A: vi.fn(async () => false), resetSession: vi.fn(async () => {}), abortTurn: vi.fn(async () => false), status: vi.fn(() => ({ busy: false })), diff --git a/tests/gateway/subscriptions.test.ts b/tests/gateway/subscriptions.test.ts index 2bc56dd..9c299f6 100644 --- a/tests/gateway/subscriptions.test.ts +++ b/tests/gateway/subscriptions.test.ts @@ -1,7 +1,8 @@ import { describe, expect, it, vi } from "vitest"; import type { ResolvedConfig } from "../../src/config.js"; import { - IMESSAGE_EVENT_TYPES, + A2A_EVENT_TYPES, + IDENTITY_EVENT_TYPES, MAILBOX_EVENT_TYPES, PHONE_EVENT_TYPES, reconcileSubscriptions, @@ -148,7 +149,7 @@ describe("reconcileSubscriptions", () => { expect(subs.create).toHaveBeenCalledWith({ agentIdentityId: "ident-1", url: WEBHOOK_URL, - eventTypes: IMESSAGE_EVENT_TYPES, + eventTypes: IDENTITY_EVENT_TYPES, }); expect(subs.update).not.toHaveBeenCalled(); }); @@ -170,7 +171,7 @@ describe("reconcileSubscriptions", () => { id: "sub-im", agentIdentityId: "ident-1", url: WEBHOOK_URL, - eventTypes: IMESSAGE_EVENT_TYPES, + eventTypes: IDENTITY_EVENT_TYPES, }, ]); const result = await reconcileSubscriptions(makeDeps(makeIdentity(), subs), PUBLIC_URL); @@ -193,8 +194,12 @@ describe("reconcileSubscriptions", () => { const identity = makeIdentity({ phoneNumber: null, imessageEnabled: false }); const result = await reconcileSubscriptions(makeDeps(identity, subs), PUBLIC_URL); - expect(result).toEqual({ created: 0, updated: 0, unchanged: 1 }); - expect(subs.create).not.toHaveBeenCalled(); + expect(result).toEqual({ created: 1, updated: 0, unchanged: 1 }); + expect(subs.create).toHaveBeenCalledWith({ + agentIdentityId: "ident-1", + url: WEBHOOK_URL, + eventTypes: A2A_EVENT_TYPES, + }); expect(subs.update).not.toHaveBeenCalled(); }); @@ -234,14 +239,18 @@ describe("reconcileSubscriptions", () => { ); }); - it("skips the imessage subscription when iMessage is disabled", async () => { + it("keeps the identity A2A subscription when iMessage is disabled", async () => { const subs = makeSubscriptions(); const identity = makeIdentity({ imessageEnabled: false }); const result = await reconcileSubscriptions(makeDeps(identity, subs), PUBLIC_URL); - expect(result).toEqual({ created: 2, updated: 0, unchanged: 0 }); + expect(result).toEqual({ created: 3, updated: 0, unchanged: 0 }); const owners = subs.create.mock.calls.map(([options]) => options); - expect(owners.some((o: Record) => "agentIdentityId" in o)).toBe(false); + expect(owners).toContainEqual({ + agentIdentityId: "ident-1", + url: WEBHOOK_URL, + eventTypes: A2A_EVENT_TYPES, + }); }); it("does not touch the incoming-call action when voice is disabled", async () => { diff --git a/tests/unit/a2a.test.ts b/tests/unit/a2a.test.ts index 0f51f57..db9c05d 100644 --- a/tests/unit/a2a.test.ts +++ b/tests/unit/a2a.test.ts @@ -1,9 +1,11 @@ -import { describe, expect, it, vi } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { clearActiveA2ATurn, setActiveA2ATurn } from "../../src/a2a-context.js"; import { defaultGatewayConfig } from "../../src/config.js"; import { a2aTools } from "../../src/tools/a2a.js"; function makeCtx() { return { + sessionID: "session-1", ask: vi.fn(async () => {}), abort: new AbortController().signal, } as any; @@ -12,14 +14,25 @@ function makeCtx() { function makeDeps() { const a2a = { fetchCard: vi.fn(async (url: string) => ({ rpcUrl: `${url}/rpc` })), - send: vi.fn(async () => ({ kind: "task", task: { id: "task-1" } })), + send: vi.fn(async () => ({ + kind: "task", + task: { id: "task-1", contextId: "context-1" }, + })), getTask: vi.fn(async () => ({ id: "task-1", status: { state: "TASK_STATE_WORKING" } })), wait: vi.fn(async () => ({ id: "task-1", status: { state: "TASK_STATE_COMPLETED" } })), close: vi.fn(), }; - const identity = { a2aClient: vi.fn(async () => a2a) }; + const identity = { + id: "identity-1", + a2aClient: vi.fn(async () => a2a), + a2aReply: vi.fn(async (taskId: string, options: unknown) => ({ + id: taskId, + ...(options as object), + })), + }; return { a2a, + identity, deps: { runtime: { getIdentity: vi.fn(async () => identity), @@ -45,6 +58,10 @@ function getTool(name: string, deps: any) { } describe("a2aTools", () => { + beforeEach(() => { + process.env.INKBOX_OPENCODE_HOME = `${process.env.TMPDIR ?? "/tmp"}/opencode-a2a-tools-${crypto.randomUUID()}`; + }); + it("sends a task behind the outbound approval gate", async () => { const { a2a, deps } = makeDeps(); const ctx = makeCtx(); @@ -109,4 +126,33 @@ describe("a2aTools", () => { expect.objectContaining({ taskId: "task-1", text: "More context." }), ); }); + + it("gates inbound intents to the active A2A session", async () => { + const { deps, identity } = makeDeps(); + const ctx = makeCtx(); + const context = { + taskId: "task-1", + messageId: "message-1", + contextId: "context-1", + replyIntentCommitted: false, + }; + const tool = getTool("inkbox_a2a_ask_caller", deps); + + await expect(tool.definition.execute({ text: "Which region?" }, ctx)).rejects.toThrow( + /only available/, + ); + setActiveA2ATurn("session-1", context); + try { + const result = await tool.definition.execute({ text: "Which region?" }, ctx); + expect(result).toContain("ask_caller"); + } finally { + clearActiveA2ATurn("session-1", context); + } + + expect(identity.a2aReply).toHaveBeenCalledWith("task-1", { + intent: "ask_caller", + text: "Which region?", + }); + expect(context.replyIntentCommitted).toBe(true); + }); }); From 01cc3ccb7ec5af1db9b01d181f4188f30ae62155 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 05:17:23 +0000 Subject: [PATCH 3/4] fix: tolerate staged A2A rollout --- src/gateway/subscriptions.ts | 23 +++++++++++++ tests/gateway/subscriptions.test.ts | 53 +++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+) diff --git a/src/gateway/subscriptions.ts b/src/gateway/subscriptions.ts index 621c39b..86a8438 100644 --- a/src/gateway/subscriptions.ts +++ b/src/gateway/subscriptions.ts @@ -49,6 +49,14 @@ function sameEventTypes(a: string[], b: string[]): boolean { return setA.size === setB.size && [...setA].every((e) => setB.has(e)); } +function isUnsupportedA2AEventTypes(err: unknown): boolean { + const message = inkboxErrorMessage(err); + return ( + message.includes("Validation error (422)") && + A2A_EVENT_TYPES.some((eventType) => message.includes(eventType)) + ); +} + function normalizePublicUrl(publicUrl: string): string { const base = publicUrl.trim().replace(/\/+$/, ""); if (!/^https?:\/\//.test(base)) { @@ -115,6 +123,21 @@ export async function reconcileSubscriptions( }); } } catch (err) { + if (kind === "identity" && isUnsupportedA2AEventTypes(err)) { + const fallbackEventTypes = eventTypes.filter( + (eventType) => !A2A_EVENT_TYPES.includes(eventType), + ); + deps.logger.warn( + "Inkbox API does not support A2A webhook events yet; " + + (fallbackEventTypes.length + ? "reconciling the identity subscription without A2A events" + : "skipping the A2A-only identity subscription"), + ); + if (fallbackEventTypes.length) { + await reconcileOwner(kind, owner, fallbackEventTypes); + } + return; + } throw new Error( `Failed to reconcile ${kind} webhook subscription for ${webhookUrl}: ` + inkboxErrorMessage(err), diff --git a/tests/gateway/subscriptions.test.ts b/tests/gateway/subscriptions.test.ts index 9c299f6..1307c54 100644 --- a/tests/gateway/subscriptions.test.ts +++ b/tests/gateway/subscriptions.test.ts @@ -1,8 +1,10 @@ +import { InkboxAPIError } from "@inkbox/sdk"; import { describe, expect, it, vi } from "vitest"; import type { ResolvedConfig } from "../../src/config.js"; import { A2A_EVENT_TYPES, IDENTITY_EVENT_TYPES, + IMESSAGE_EVENT_TYPES, MAILBOX_EVENT_TYPES, PHONE_EVENT_TYPES, reconcileSubscriptions, @@ -253,6 +255,57 @@ describe("reconcileSubscriptions", () => { }); }); + it("falls back to legacy identity events when the API rejects A2A event types", async () => { + const subs = makeSubscriptions(); + subs.create.mockImplementation(async (options: { url: string; eventTypes: string[] }) => { + if (options.eventTypes.some((eventType) => A2A_EVENT_TYPES.includes(eventType))) { + throw new InkboxAPIError(422, { detail: "a2a.task.created is not a valid event type" }); + } + return { + id: "sub-created", + url: options.url, + eventTypes: options.eventTypes, + signingKey: null, + }; + }); + const deps = makeDeps(makeIdentity(), subs); + + const result = await reconcileSubscriptions(deps, PUBLIC_URL); + + expect(result).toEqual({ created: 3, updated: 0, unchanged: 0 }); + expect(subs.create).toHaveBeenLastCalledWith({ + agentIdentityId: "ident-1", + url: WEBHOOK_URL, + eventTypes: IMESSAGE_EVENT_TYPES, + }); + expect(deps.logger.warn).toHaveBeenCalledWith( + expect.stringContaining("does not support A2A webhook events yet"), + ); + }); + + it("continues without an identity subscription when only A2A events are unsupported", async () => { + const subs = makeSubscriptions(); + subs.create.mockImplementation(async (options: { url: string; eventTypes: string[] }) => { + if (options.eventTypes.some((eventType) => A2A_EVENT_TYPES.includes(eventType))) { + throw new InkboxAPIError(422, { detail: "a2a.task.created is not a valid event type" }); + } + return { + id: "sub-created", + url: options.url, + eventTypes: options.eventTypes, + signingKey: null, + }; + }); + const deps = makeDeps(makeIdentity({ imessageEnabled: false }), subs); + + const result = await reconcileSubscriptions(deps, PUBLIC_URL); + + expect(result).toEqual({ created: 2, updated: 0, unchanged: 0 }); + expect(deps.logger.warn).toHaveBeenCalledWith( + expect.stringContaining("skipping the A2A-only identity subscription"), + ); + }); + it("does not touch the incoming-call action when voice is disabled", async () => { const identity = makeIdentity(); await reconcileSubscriptions( From 20b750ae9b3226e758106e3024062c3666168e0d Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 05:31:18 +0000 Subject: [PATCH 4/4] fix: handle SDK A2A preflight validation --- src/gateway/subscriptions.ts | 5 +++-- tests/gateway/subscriptions.test.ts | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/gateway/subscriptions.ts b/src/gateway/subscriptions.ts index 86a8438..79b13ab 100644 --- a/src/gateway/subscriptions.ts +++ b/src/gateway/subscriptions.ts @@ -52,8 +52,9 @@ function sameEventTypes(a: string[], b: string[]): boolean { function isUnsupportedA2AEventTypes(err: unknown): boolean { const message = inkboxErrorMessage(err); return ( - message.includes("Validation error (422)") && - A2A_EVENT_TYPES.some((eventType) => message.includes(eventType)) + A2A_EVENT_TYPES.some((eventType) => message.includes(eventType)) && + (message.includes("Validation error (422)") || + message.includes("does not belong to any known channel")) ); } diff --git a/tests/gateway/subscriptions.test.ts b/tests/gateway/subscriptions.test.ts index 1307c54..5538782 100644 --- a/tests/gateway/subscriptions.test.ts +++ b/tests/gateway/subscriptions.test.ts @@ -259,7 +259,7 @@ describe("reconcileSubscriptions", () => { const subs = makeSubscriptions(); subs.create.mockImplementation(async (options: { url: string; eventTypes: string[] }) => { if (options.eventTypes.some((eventType) => A2A_EVENT_TYPES.includes(eventType))) { - throw new InkboxAPIError(422, { detail: "a2a.task.created is not a valid event type" }); + throw new Error("event_type 'a2a.task.created' does not belong to any known channel"); } return { id: "sub-created",