From af8a76af94c5f8561fa1560ccc5548fadd480786 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Tue, 21 Jul 2026 01:23:30 +0000 Subject: [PATCH 01/16] Remove paused contact-rule guidance --- skills/inkbox-contact-rules/SKILL.md | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/skills/inkbox-contact-rules/SKILL.md b/skills/inkbox-contact-rules/SKILL.md index b17de6e..e847fc1 100644 --- a/skills/inkbox-contact-rules/SKILL.md +++ b/skills/inkbox-contact-rules/SKILL.md @@ -22,9 +22,8 @@ by a human in the Inkbox console (https://inkbox.ai/console/contact-rules). 1. When the user asks "who is blocked?", "can X reach you?", or "why didn't my email arrive?" — list the rules for the matching channel and read the - result: `action: "block"` rejects matching traffic, `action: "allow"` - permits it when a whitelist posture is active, `status: "paused"` means - the rule exists but is not enforced. + result: `action: "block"` rejects matching traffic and `action: "allow"` + permits it when a whitelist posture is active. 2. Match types: mail rules use `exact_email` or `domain`; phone and iMessage rules use `exact_number` (E.164, e.g. `+15551234567`). Phone rules apply to both SMS and voice on the dedicated number. From ae104600715ff8cb879ffe7ad8d503b0a6d14975 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Thu, 23 Jul 2026 09:40:59 +0000 Subject: [PATCH 02/16] feat: add A2A task tools --- CHANGELOG.md | 6 ++ README.md | 8 +- src/tools/a2a.ts | 137 +++++++++++++++++++++++++ src/tools/index.ts | 2 + tests/contract/tool-vocabulary.test.ts | 8 +- tests/unit/a2a.test.ts | 112 ++++++++++++++++++++ 6 files changed, 268 insertions(+), 5 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/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 0547c5de4b601f42d05ddeba9c8a3f1206b742b5 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Thu, 23 Jul 2026 10:31:45 +0000 Subject: [PATCH 03/16] 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 966458d1ff6b436ca250bec2f5908f6924310279 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 05:17:23 +0000 Subject: [PATCH 04/16] 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 3ac63d92b8c4ff552908a9ac38d895dc462bb8e9 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 05:31:18 +0000 Subject: [PATCH 05/16] 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", From d939c1ca501e261f5a40e2cd901cdd7fd02a5ae7 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 21:23:50 +0000 Subject: [PATCH 06/16] Add filterable A2A history tools --- .github/workflows/tests.yml | 6 +- CHANGELOG.md | 11 +-- README.md | 4 +- package-lock.json | 2 +- package.json | 2 +- src/gateway/a2a.ts | 2 +- src/tools/a2a.ts | 101 +++++++++++++++++++++++++ tests/contract/tool-vocabulary.test.ts | 6 +- tests/unit/a2a.test.ts | 68 +++++++++++++++++ 9 files changed, 187 insertions(+), 15 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 46a067e..2832483 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -13,7 +13,7 @@ jobs: - uses: actions/checkout@v7 with: repository: inkbox-ai/inkbox - ref: 199bbd27c8dab2f70e379de52ca9cc910b0e141d + ref: fdea6e55ac117f246624852aedeef0feabe08b9a path: .ci/inkbox - uses: actions/setup-node@v7 with: @@ -39,7 +39,7 @@ jobs: - uses: actions/checkout@v7 with: repository: inkbox-ai/inkbox - ref: 199bbd27c8dab2f70e379de52ca9cc910b0e141d + ref: fdea6e55ac117f246624852aedeef0feabe08b9a path: .ci/inkbox - uses: actions/setup-node@v7 with: @@ -65,7 +65,7 @@ jobs: - uses: actions/checkout@v7 with: repository: inkbox-ai/inkbox - ref: 199bbd27c8dab2f70e379de52ca9cc910b0e141d + ref: fdea6e55ac117f246624852aedeef0feabe08b9a path: .ci/inkbox - uses: actions/setup-node@v7 with: diff --git a/CHANGELOG.md b/CHANGELOG.md index f3d36fe..5391bf4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,16 +6,17 @@ 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. +- Adds paginated task and message history with direction, participant, + lifecycle, context, role, keyword, and timestamp filters. +- The plugin requires `@inkbox/sdk` 0.5.6 or newer. ## 0.1.0 (unreleased) Initial release. -- Requires `@inkbox/sdk` 0.5.1 or newer. -- 48 `inkbox_*` tools across email, SMS/MMS, iMessage, calls, contacts, notes, - contact rules, note access grants, encrypted vault, and diagnostics. 27 are +- Requires `@inkbox/sdk` 0.5.6 or newer. +- 56 `inkbox_*` tools across A2A, email, SMS/MMS, iMessage, calls, contacts, + notes, contact rules, note access grants, encrypted vault, and diagnostics. 35 are enabled by default; the rest are opt-in via the `tools.enable` plugin option (`inkbox_doctor` reports what is off and how to enable it). - Outbound sends and calls gate through opencode's native permission prompts, diff --git a/README.md b/README.md index 21a1922..be2efe6 100644 --- a/README.md +++ b/README.md @@ -111,13 +111,13 @@ groups are enabled. ## Tools -33 tools are enabled by default; 21 more are opt-in (see +35 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`, `inkbox_a2a_complete`, `inkbox_a2a_ask_caller`, `inkbox_a2a_fail` | — | +| `a2a` | `inkbox_a2a_call`, `inkbox_a2a_check`, `inkbox_a2a_reply`, `inkbox_list_a2a_tasks`, `inkbox_list_a2a_messages`, `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/package-lock.json b/package-lock.json index cc43bec..de0c302 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,7 +9,7 @@ "version": "0.2.5", "license": "MIT", "dependencies": { - "@inkbox/sdk": ">=0.5.1 <1.0.0", + "@inkbox/sdk": ">=0.5.6 <1.0.0", "@opencode-ai/sdk": ">=1.17.18 <1.19.0", "ws": "^8.21.0", "zod": "4.1.8" diff --git a/package.json b/package.json index 322eed3..073fe47 100644 --- a/package.json +++ b/package.json @@ -55,7 +55,7 @@ "opencode": ">=1.15.0 <1.19.0" }, "dependencies": { - "@inkbox/sdk": ">=0.5.1 <1.0.0", + "@inkbox/sdk": ">=0.5.6 <1.0.0", "@opencode-ai/sdk": ">=1.17.18 <1.19.0", "ws": "^8.21.0", "zod": "4.1.8" diff --git a/src/gateway/a2a.ts b/src/gateway/a2a.ts index 2f88c68..bd487e7 100644 --- a/src/gateway/a2a.ts +++ b/src/gateway/a2a.ts @@ -186,7 +186,7 @@ export function createA2AHandler(deps: { typeof id.a2aReply !== "function" ) { deps.logger.warn("a2a.sdk_upgrade_required", { - requiredVersion: "0.5.5", + requiredVersion: "0.5.6", }); return; } diff --git a/src/tools/a2a.ts b/src/tools/a2a.ts index a8eb6ee..a3bca5f 100644 --- a/src/tools/a2a.ts +++ b/src/tools/a2a.ts @@ -35,12 +35,44 @@ const completeArgs = { const failArgs = { reason: z.string().min(1).describe("Failure reason."), }; +const historyCommonArgs = { + direction: z.enum(["inbound", "outbound", "both"]).optional(), + requesterHandle: z.string().min(1).optional(), + workerHandle: z.string().min(1).optional(), + contextId: z.string().min(1).optional(), + query: z.string().min(1).max(500).optional(), + since: z.string().min(1).describe("ISO 8601 lower timestamp bound.").optional(), + cursor: z.string().min(1).describe("Opaque next_cursor from the previous page.").optional(), + limit: z.number().int().min(1).max(100).default(50), +}; +const taskHistoryArgs = { + ...historyCommonArgs, + state: z + .enum([ + "submitted", + "working", + "input_required", + "auth_required", + "completed", + "failed", + "canceled", + "rejected", + ]) + .optional(), +}; +const messageHistoryArgs = { + ...historyCommonArgs, + taskId: z.string().min(1).optional(), + role: z.enum(["caller", "agent"]).optional(), +}; type CallArgs = z.infer>; type CheckArgs = z.infer>; type ReplyArgs = z.infer>; type CompleteArgs = z.infer>; type FailArgs = z.infer>; +type TaskHistoryArgs = z.infer>; +type MessageHistoryArgs = z.infer>; async function clientFor(deps: ToolDeps): Promise { const identity = await deps.runtime.getIdentity(); @@ -170,6 +202,75 @@ export function a2aTools(deps: ToolDeps): RegisteredTool[] { }, }, }, + { + name: "inkbox_list_a2a_tasks", + group: "a2a", + defaultEnabled: true, + definition: { + description: + "List this identity's A2A task history. Direction defaults to inbound; filter by participants, state, context, keywords, or timestamp and follow next_cursor for more.", + args: taskHistoryArgs, + async execute(args: TaskHistoryArgs) { + return runTool(async () => { + const identity = await deps.runtime.getIdentity(); + const list = (identity as any).a2aTasks; + if (typeof list !== "function") { + throw new Error( + "This A2A tool requires @inkbox/sdk with identity.a2aTasks() support.", + ); + } + return formatJson( + await list.call(identity, { + direction: args.direction, + requesterHandle: args.requesterHandle, + workerHandle: args.workerHandle, + state: args.state, + contextId: args.contextId, + q: args.query, + since: args.since, + cursor: args.cursor, + limit: args.limit, + }), + ); + }); + }, + }, + }, + { + name: "inkbox_list_a2a_messages", + group: "a2a", + defaultEnabled: true, + definition: { + description: + "List messages from this identity's inbound and outbound A2A history. Filter by direction, participants, task, context, role, keywords, or timestamp and follow next_cursor for more.", + args: messageHistoryArgs, + async execute(args: MessageHistoryArgs) { + return runTool(async () => { + const identity = await deps.runtime.getIdentity(); + const list = (identity as any).a2aMessages; + if (typeof list !== "function") { + throw new Error( + "This A2A tool requires @inkbox/sdk with identity.a2aMessages() support.", + ); + } + return formatJson( + await list.call(identity, { + direction: args.direction, + requesterHandle: args.requesterHandle, + workerHandle: args.workerHandle, + taskId: args.taskId, + contextId: args.contextId, + role: args.role, + q: args.query, + since: args.since, + cursor: args.cursor, + limit: args.limit, + }), + ); + }); + }, + }, + }, { name: "inkbox_a2a_complete", group: "a2a", diff --git a/tests/contract/tool-vocabulary.test.ts b/tests/contract/tool-vocabulary.test.ts index cf41b08..cf19322 100644 --- a/tests/contract/tool-vocabulary.test.ts +++ b/tests/contract/tool-vocabulary.test.ts @@ -35,6 +35,8 @@ const DEFAULT_ENABLED = [ "inkbox_a2a_call", "inkbox_a2a_check", "inkbox_a2a_reply", + "inkbox_list_a2a_tasks", + "inkbox_list_a2a_messages", "inkbox_a2a_complete", "inkbox_a2a_ask_caller", "inkbox_a2a_fail", @@ -115,10 +117,10 @@ const GROUPS = [ describe("tool vocabulary", () => { const all = buildAllTools(stubDeps()); - it("ships exactly the expected 54 tools", () => { + it("ships exactly the expected 56 tools", () => { const names = all.map((t) => t.name).sort(); expect(names).toEqual([...DEFAULT_ENABLED, ...OPT_IN].sort()); - expect(names).toHaveLength(54); + expect(names).toHaveLength(56); }); it("has no duplicate tool names", () => { diff --git a/tests/unit/a2a.test.ts b/tests/unit/a2a.test.ts index db9c05d..eda5579 100644 --- a/tests/unit/a2a.test.ts +++ b/tests/unit/a2a.test.ts @@ -25,6 +25,14 @@ function makeDeps() { const identity = { id: "identity-1", a2aClient: vi.fn(async () => a2a), + a2aTasks: vi.fn(async (options: unknown) => ({ + items: [{ id: "task-1", options }], + nextCursor: "task-next", + })), + a2aMessages: vi.fn(async (options: unknown) => ({ + items: [{ id: "message-1", options }], + nextCursor: "message-next", + })), a2aReply: vi.fn(async (taskId: string, options: unknown) => ({ id: taskId, ...(options as object), @@ -127,6 +135,66 @@ describe("a2aTools", () => { ); }); + it("lists filtered task and message history with cursors", async () => { + const { deps, identity } = makeDeps(); + const ctx = makeCtx(); + const tasks = await getTool("inkbox_list_a2a_tasks", deps).definition.execute( + { + direction: "both", + requesterHandle: "requester", + workerHandle: "worker", + state: "completed", + contextId: "context-1", + query: "summary", + since: "2026-07-01T00:00:00Z", + cursor: "task-cursor", + limit: 3, + }, + ctx, + ); + const messages = await getTool("inkbox_list_a2a_messages", deps).definition.execute( + { + direction: "outbound", + requesterHandle: "requester", + workerHandle: "worker", + taskId: "task-1", + contextId: "context-1", + role: "agent", + query: "done", + since: "2026-07-01T00:00:00Z", + cursor: "message-cursor", + limit: 4, + }, + ctx, + ); + + expect(tasks).toContain("task-next"); + expect(messages).toContain("message-next"); + expect(identity.a2aTasks).toHaveBeenCalledWith({ + direction: "both", + requesterHandle: "requester", + workerHandle: "worker", + state: "completed", + contextId: "context-1", + q: "summary", + since: "2026-07-01T00:00:00Z", + cursor: "task-cursor", + limit: 3, + }); + expect(identity.a2aMessages).toHaveBeenCalledWith({ + direction: "outbound", + requesterHandle: "requester", + workerHandle: "worker", + taskId: "task-1", + contextId: "context-1", + role: "agent", + q: "done", + since: "2026-07-01T00:00:00Z", + cursor: "message-cursor", + limit: 4, + }); + }); + it("gates inbound intents to the active A2A session", async () => { const { deps, identity } = makeDeps(); const ctx = makeCtx(); From 0fd759b52cfff8f262da15956a309f92affa2bd1 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 21:38:43 +0000 Subject: [PATCH 07/16] Fix unpublished SDK installs in live CI --- .github/workflows/canary.yml | 27 +++++++++++++++++++--- .github/workflows/live-channels.yml | 13 ++++++++++- .github/workflows/live-external-events.yml | 13 ++++++++++- .github/workflows/live-voice.yml | 13 ++++++++++- 4 files changed, 60 insertions(+), 6 deletions(-) diff --git a/.github/workflows/canary.yml b/.github/workflows/canary.yml index ff015f2..04d3536 100644 --- a/.github/workflows/canary.yml +++ b/.github/workflows/canary.yml @@ -12,12 +12,22 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v7 + - uses: actions/checkout@v7 + with: + repository: inkbox-ai/inkbox + ref: fdea6e55ac117f246624852aedeef0feabe08b9a + path: .ci/inkbox - uses: actions/setup-node@v7 with: node-version: 22 cache: npm - - run: npm ci - - run: npm install -D @opencode-ai/plugin@latest + - name: Install dependencies from the unpublished SDK source + run: | + npm ci --prefix .ci/inkbox/sdk/typescript + npm run build --prefix .ci/inkbox/sdk/typescript + npm install --no-save --package-lock=false \ + ./.ci/inkbox/sdk/typescript \ + @opencode-ai/sdk@latest @opencode-ai/plugin@latest - run: npm run lint - run: npm run typecheck - run: npm test @@ -27,11 +37,22 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v7 + - uses: actions/checkout@v7 + with: + repository: inkbox-ai/inkbox + ref: fdea6e55ac117f246624852aedeef0feabe08b9a + path: .ci/inkbox - uses: actions/setup-node@v7 with: node-version: 22 cache: npm - - run: npm ci + - name: Install dependencies from the unpublished SDK source + run: | + npm ci --prefix .ci/inkbox/sdk/typescript + npm run build --prefix .ci/inkbox/sdk/typescript + npm install --no-save --package-lock=false \ + ./.ci/inkbox/sdk/typescript \ + @opencode-ai/sdk@1.17.18 @opencode-ai/plugin@1.17.18 - run: npm install -g opencode-ai@latest - run: bash scripts/smoke-loader.sh diff --git a/.github/workflows/live-channels.yml b/.github/workflows/live-channels.yml index 8c31cb4..be93eec 100644 --- a/.github/workflows/live-channels.yml +++ b/.github/workflows/live-channels.yml @@ -48,11 +48,22 @@ jobs: steps: - uses: actions/checkout@v7 + - uses: actions/checkout@v7 + with: + repository: inkbox-ai/inkbox + ref: fdea6e55ac117f246624852aedeef0feabe08b9a + path: .ci/inkbox - uses: actions/setup-node@v7 with: node-version: 22 cache: npm - - run: npm ci + - name: Install dependencies from the unpublished SDK source + run: | + npm ci --prefix .ci/inkbox/sdk/typescript + npm run build --prefix .ci/inkbox/sdk/typescript + npm install --no-save --package-lock=false \ + ./.ci/inkbox/sdk/typescript \ + @opencode-ai/sdk@1.17.18 @opencode-ai/plugin@1.17.18 - run: npm install -g opencode-ai@latest - name: Start mock model diff --git a/.github/workflows/live-external-events.yml b/.github/workflows/live-external-events.yml index 43ce6e1..ddf7cc4 100644 --- a/.github/workflows/live-external-events.yml +++ b/.github/workflows/live-external-events.yml @@ -38,11 +38,22 @@ jobs: steps: - uses: actions/checkout@v7 + - uses: actions/checkout@v7 + with: + repository: inkbox-ai/inkbox + ref: fdea6e55ac117f246624852aedeef0feabe08b9a + path: .ci/inkbox - uses: actions/setup-node@v7 with: node-version: 22 cache: npm - - run: npm ci + - name: Install dependencies from the unpublished SDK source + run: | + npm ci --prefix .ci/inkbox/sdk/typescript + npm run build --prefix .ci/inkbox/sdk/typescript + npm install --no-save --package-lock=false \ + ./.ci/inkbox/sdk/typescript \ + @opencode-ai/sdk@1.17.18 @opencode-ai/plugin@1.17.18 - run: npm install -g opencode-ai@latest - name: Boot the AUT gateway (real model, external events on) diff --git a/.github/workflows/live-voice.yml b/.github/workflows/live-voice.yml index 7e3f87f..2379f14 100644 --- a/.github/workflows/live-voice.yml +++ b/.github/workflows/live-voice.yml @@ -55,11 +55,22 @@ jobs: steps: - uses: actions/checkout@v7 + - uses: actions/checkout@v7 + with: + repository: inkbox-ai/inkbox + ref: fdea6e55ac117f246624852aedeef0feabe08b9a + path: .ci/inkbox - uses: actions/setup-node@v7 with: node-version: 22 cache: npm - - run: npm ci + - name: Install dependencies from the unpublished SDK source + run: | + npm ci --prefix .ci/inkbox/sdk/typescript + npm run build --prefix .ci/inkbox/sdk/typescript + npm install --no-save --package-lock=false \ + ./.ci/inkbox/sdk/typescript \ + @opencode-ai/sdk@1.17.18 @opencode-ai/plugin@1.17.18 - run: npm install -g opencode-ai@latest - name: Boot the AUT gateway (voice; ${{ matrix.scenario }}) From 1919ea309caf39c77453489bf52796ff194c01e1 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 21:42:43 +0000 Subject: [PATCH 08/16] Use source SDK when staging live plugin --- .github/workflows/live-channels.yml | 1 + .github/workflows/live-external-events.yml | 1 + .github/workflows/live-voice.yml | 1 + scripts/live-aut.sh | 7 ++++++- 4 files changed, 9 insertions(+), 1 deletion(-) diff --git a/.github/workflows/live-channels.yml b/.github/workflows/live-channels.yml index be93eec..bb62042 100644 --- a/.github/workflows/live-channels.yml +++ b/.github/workflows/live-channels.yml @@ -80,6 +80,7 @@ jobs: - name: Boot the AUT gateway (${{ matrix.mode }}) env: MODE: ${{ matrix.mode }} + INKBOX_SDK_PATH: ${{ github.workspace }}/.ci/inkbox/sdk/typescript AUT_INKBOX_API_KEY: ${{ secrets.AUT_INKBOX_API_KEY }} AUT_INKBOX_SIGNING_KEY: ${{ secrets.AUT_INKBOX_SIGNING_KEY }} INKBOX_BASE_URL: ${{ vars.INKBOX_BASE_URL }} diff --git a/.github/workflows/live-external-events.yml b/.github/workflows/live-external-events.yml index ddf7cc4..09ec4c1 100644 --- a/.github/workflows/live-external-events.yml +++ b/.github/workflows/live-external-events.yml @@ -59,6 +59,7 @@ jobs: - name: Boot the AUT gateway (real model, external events on) env: MODE: real + INKBOX_SDK_PATH: ${{ github.workspace }}/.ci/inkbox/sdk/typescript AUT_INKBOX_API_KEY: ${{ secrets.AUT_INKBOX_API_KEY }} AUT_INKBOX_SIGNING_KEY: ${{ secrets.AUT_INKBOX_SIGNING_KEY }} INKBOX_BASE_URL: ${{ vars.INKBOX_BASE_URL }} diff --git a/.github/workflows/live-voice.yml b/.github/workflows/live-voice.yml index 2379f14..40c96e1 100644 --- a/.github/workflows/live-voice.yml +++ b/.github/workflows/live-voice.yml @@ -76,6 +76,7 @@ jobs: - name: Boot the AUT gateway (voice; ${{ matrix.scenario }}) env: MODE: real + INKBOX_SDK_PATH: ${{ github.workspace }}/.ci/inkbox/sdk/typescript AUT_INKBOX_API_KEY: ${{ secrets.AUT_INKBOX_API_KEY }} AUT_INKBOX_SIGNING_KEY: ${{ secrets.AUT_INKBOX_SIGNING_KEY }} INKBOX_BASE_URL: ${{ vars.INKBOX_BASE_URL }} diff --git a/scripts/live-aut.sh b/scripts/live-aut.sh index 5f43655..b119c43 100755 --- a/scripts/live-aut.sh +++ b/scripts/live-aut.sh @@ -10,6 +10,7 @@ # INKBOX_BASE_URL Inkbox API origin (default https://inkbox.ai) # OPENAI_API_KEY required when MODE=real # LIVE_WORKDIR where to stage (default: mktemp) +# INKBOX_SDK_PATH optional local SDK source installed with the plugin # INKBOX_WEBHOOK_SECRET_GITHUB optional; enables external-event turns # # Out (appended to $GITHUB_ENV when present, else printed): @@ -46,7 +47,11 @@ mkdir -p "$WORKDIR" cd "$WORKDIR" git init -q . 2>/dev/null || true npm init -y >/dev/null 2>&1 || true -npm install --silent "$TARBALL" +if [ -n "${INKBOX_SDK_PATH:-}" ]; then + npm install --silent --package-lock=false "$INKBOX_SDK_PATH" "$TARBALL" +else + npm install --silent "$TARBALL" +fi mkdir -p .opencode/plugins .opencode/agent cp "$ROOT/agents/inkbox-channel.md" .opencode/agent/ From 6ebc04948fe185b5a10b4689f4c1d91c8524af1b Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 21:57:47 +0000 Subject: [PATCH 09/16] Split A2A webhook subscriptions by channel --- src/gateway/subscriptions.ts | 58 ++++++++++++++++++++--------- tests/gateway/subscriptions.test.ts | 34 +++++++++++------ 2 files changed, 62 insertions(+), 30 deletions(-) diff --git a/src/gateway/subscriptions.ts b/src/gateway/subscriptions.ts index 79b13ab..c8ea247 100644 --- a/src/gateway/subscriptions.ts +++ b/src/gateway/subscriptions.ts @@ -25,8 +25,6 @@ export const A2A_EVENT_TYPES = [ "a2a.task.canceled", "a2a.sent_task.updated", ]; -export const IDENTITY_EVENT_TYPES = [...IMESSAGE_EVENT_TYPES, ...A2A_EVENT_TYPES]; - export interface ReconcileResult { created: number; updated: number; @@ -49,6 +47,20 @@ function sameEventTypes(a: string[], b: string[]): boolean { return setA.size === setB.size && [...setA].every((e) => setB.has(e)); } +function eventFamilies(eventTypes: string[]): Set { + return new Set(eventTypes.map((eventType) => eventType.split(".", 1)[0])); +} + +function sameWebhookPath(left: string, right: string): boolean { + try { + const leftUrl = new URL(left); + const rightUrl = new URL(right); + return leftUrl.origin === rightUrl.origin && leftUrl.pathname === rightUrl.pathname; + } catch { + return false; + } +} + function isUnsupportedA2AEventTypes(err: unknown): boolean { const message = inkboxErrorMessage(err); return ( @@ -93,14 +105,22 @@ export async function reconcileSubscriptions( kind: string, owner: SubscriptionOwner, eventTypes: string[], + subscriptionUrl = webhookUrl, ): Promise { try { const existing = await client.webhooks.subscriptions.list(owner); - const ours = existing.find((sub) => sub.url === webhookUrl); + const ours = existing.find((sub) => sub.url === subscriptionUrl); + const desiredFamilies = eventFamilies(eventTypes); + const staleOurs = existing.filter( + (sub) => + sub.url !== subscriptionUrl && + sameWebhookPath(sub.url, subscriptionUrl) && + [...eventFamilies(sub.eventTypes)].some((family) => desiredFamilies.has(family)), + ); if (!ours) { const created = await client.webhooks.subscriptions.create({ ...owner, - url: webhookUrl, + url: subscriptionUrl, eventTypes, }); result.created += 1; @@ -123,24 +143,22 @@ export async function reconcileSubscriptions( subscriptionId: ours.id, }); } + for (const stale of staleOurs) { + await client.webhooks.subscriptions.delete(stale.id); + deps.logger.info("removed stale webhook subscription", { + kind, + subscriptionId: stale.id, + }); + } } catch (err) { - if (kind === "identity" && isUnsupportedA2AEventTypes(err)) { - const fallbackEventTypes = eventTypes.filter( - (eventType) => !A2A_EVENT_TYPES.includes(eventType), - ); + if (kind === "a2a" && isUnsupportedA2AEventTypes(err)) { 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"), + "Inkbox API does not support A2A webhook events yet; " + "skipping the A2A subscription", ); - if (fallbackEventTypes.length) { - await reconcileOwner(kind, owner, fallbackEventTypes); - } return; } throw new Error( - `Failed to reconcile ${kind} webhook subscription for ${webhookUrl}: ` + + `Failed to reconcile ${kind} webhook subscription for ${subscriptionUrl}: ` + inkboxErrorMessage(err), ); } @@ -153,10 +171,14 @@ export async function reconcileSubscriptions( await reconcileOwner("phone", { phoneNumberId: identity.phoneNumber.id }, PHONE_EVENT_TYPES); } await reconcileOwner( - "identity", + "a2a", { agentIdentityId: identity.id }, - identity.imessageEnabled ? IDENTITY_EVENT_TYPES : A2A_EVENT_TYPES, + A2A_EVENT_TYPES, + `${webhookUrl}?channel=a2a`, ); + if (identity.imessageEnabled) { + await reconcileOwner("imessage", { agentIdentityId: identity.id }, IMESSAGE_EVENT_TYPES); + } if (deps.config.gateway.voice.enabled) { await wireIncomingCalls(deps, identity, base, webhookUrl); diff --git a/tests/gateway/subscriptions.test.ts b/tests/gateway/subscriptions.test.ts index 5538782..7efcf8a 100644 --- a/tests/gateway/subscriptions.test.ts +++ b/tests/gateway/subscriptions.test.ts @@ -3,7 +3,6 @@ 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, @@ -136,8 +135,8 @@ describe("reconcileSubscriptions", () => { const subs = makeSubscriptions(); const result = await reconcileSubscriptions(makeDeps(makeIdentity(), subs), PUBLIC_URL); - expect(result).toEqual({ created: 3, updated: 0, unchanged: 0 }); - expect(subs.create).toHaveBeenCalledTimes(3); + expect(result).toEqual({ created: 4, updated: 0, unchanged: 0 }); + expect(subs.create).toHaveBeenCalledTimes(4); expect(subs.create).toHaveBeenCalledWith({ mailboxId: "mb-1", url: WEBHOOK_URL, @@ -148,10 +147,15 @@ describe("reconcileSubscriptions", () => { url: WEBHOOK_URL, eventTypes: PHONE_EVENT_TYPES, }); + expect(subs.create).toHaveBeenCalledWith({ + agentIdentityId: "ident-1", + url: `${WEBHOOK_URL}?channel=a2a`, + eventTypes: A2A_EVENT_TYPES, + }); expect(subs.create).toHaveBeenCalledWith({ agentIdentityId: "ident-1", url: WEBHOOK_URL, - eventTypes: IDENTITY_EVENT_TYPES, + eventTypes: IMESSAGE_EVENT_TYPES, }); expect(subs.update).not.toHaveBeenCalled(); }); @@ -161,7 +165,7 @@ describe("reconcileSubscriptions", () => { await reconcileSubscriptions(makeDeps(makeIdentity(), subs), `${PUBLIC_URL}/`); for (const [options] of subs.create.mock.calls) { - expect(options.url).toBe(WEBHOOK_URL); + expect([WEBHOOK_URL, `${WEBHOOK_URL}?channel=a2a`]).toContain(options.url); } }); @@ -169,16 +173,22 @@ describe("reconcileSubscriptions", () => { const subs = makeSubscriptions([ { id: "sub-mb", mailboxId: "mb-1", url: WEBHOOK_URL, eventTypes: ["message.received"] }, { id: "sub-pn", phoneNumberId: "pn-1", url: WEBHOOK_URL, eventTypes: PHONE_EVENT_TYPES }, + { + id: "sub-a2a", + agentIdentityId: "ident-1", + url: `${WEBHOOK_URL}?channel=a2a`, + eventTypes: A2A_EVENT_TYPES, + }, { id: "sub-im", agentIdentityId: "ident-1", url: WEBHOOK_URL, - eventTypes: IDENTITY_EVENT_TYPES, + eventTypes: IMESSAGE_EVENT_TYPES, }, ]); const result = await reconcileSubscriptions(makeDeps(makeIdentity(), subs), PUBLIC_URL); - expect(result).toEqual({ created: 0, updated: 1, unchanged: 2 }); + expect(result).toEqual({ created: 0, updated: 1, unchanged: 3 }); expect(subs.update).toHaveBeenCalledTimes(1); expect(subs.update).toHaveBeenCalledWith("sub-mb", { eventTypes: MAILBOX_EVENT_TYPES }); expect(subs.create).not.toHaveBeenCalled(); @@ -199,7 +209,7 @@ describe("reconcileSubscriptions", () => { expect(result).toEqual({ created: 1, updated: 0, unchanged: 1 }); expect(subs.create).toHaveBeenCalledWith({ agentIdentityId: "ident-1", - url: WEBHOOK_URL, + url: `${WEBHOOK_URL}?channel=a2a`, eventTypes: A2A_EVENT_TYPES, }); expect(subs.update).not.toHaveBeenCalled(); @@ -223,7 +233,7 @@ describe("reconcileSubscriptions", () => { const result = await reconcileSubscriptions(makeDeps(makeIdentity(), subs), PUBLIC_URL); // Foreign subscriptions are ignored entirely; ours are created alongside. - expect(result).toEqual({ created: 3, updated: 0, unchanged: 0 }); + expect(result).toEqual({ created: 4, updated: 0, unchanged: 0 }); expect(subs.update).not.toHaveBeenCalled(); expect(subs.delete).not.toHaveBeenCalled(); }); @@ -233,7 +243,7 @@ describe("reconcileSubscriptions", () => { const identity = makeIdentity({ phoneNumber: null }); 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) => "phoneNumberId" in o)).toBe(false); expect(subs.list).not.toHaveBeenCalledWith( @@ -250,7 +260,7 @@ describe("reconcileSubscriptions", () => { const owners = subs.create.mock.calls.map(([options]) => options); expect(owners).toContainEqual({ agentIdentityId: "ident-1", - url: WEBHOOK_URL, + url: `${WEBHOOK_URL}?channel=a2a`, eventTypes: A2A_EVENT_TYPES, }); }); @@ -302,7 +312,7 @@ describe("reconcileSubscriptions", () => { expect(result).toEqual({ created: 2, updated: 0, unchanged: 0 }); expect(deps.logger.warn).toHaveBeenCalledWith( - expect.stringContaining("skipping the A2A-only identity subscription"), + expect.stringContaining("skipping the A2A subscription"), ); }); From 2ffdec632b71738e30f1c12e04fd070e35356414 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 22:00:27 +0000 Subject: [PATCH 10/16] Tolerate pre-rollout A2A API absence --- src/gateway/a2a.ts | 53 ++++++++++++++++++++++++++------------- tests/gateway/a2a.test.ts | 34 +++++++++++++++++++++++++ 2 files changed, 69 insertions(+), 18 deletions(-) diff --git a/src/gateway/a2a.ts b/src/gateway/a2a.ts index bd487e7..aa4ede5 100644 --- a/src/gateway/a2a.ts +++ b/src/gateway/a2a.ts @@ -46,6 +46,16 @@ function registry(state: StateStore): Record { return value && typeof value === "object" ? (value as Record) : {}; } +function isA2AApiUnavailable(error: unknown): boolean { + return ( + (typeof error === "object" && + error !== null && + "statusCode" in error && + error.statusCode === 404) || + /\bHTTP 404\b/.test(String(error)) + ); +} + function persist( state: StateStore, key: string, @@ -206,24 +216,31 @@ export function createA2AHandler(deps: { }); } } - 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); + try { + 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); + } + } catch (error) { + if (!isA2AApiUnavailable(error)) throw error; + deps.logger.warn("a2a.api_unavailable", { + error: String(error), + }); } }, }; diff --git a/tests/gateway/a2a.test.ts b/tests/gateway/a2a.test.ts index 91c28b4..0628b56 100644 --- a/tests/gateway/a2a.test.ts +++ b/tests/gateway/a2a.test.ts @@ -122,4 +122,38 @@ describe("createA2AHandler", () => { expect(runCapture).toHaveBeenCalledWith("contact-1", expect.stringContaining("Which region?")); }); + + it("continues startup when the A2A API is not deployed yet", async () => { + const unavailable = Object.assign(new Error("HTTP 404: Not Found"), { + statusCode: 404, + }); + const logger = { info: vi.fn(), warn: vi.fn(), error: vi.fn() }; + const identity = { + a2aTask: vi.fn(), + a2aReply: vi.fn(), + iterA2ATasks: vi.fn(() => ({ + [Symbol.asyncIterator]: () => ({ + next: vi.fn(async () => { + throw unavailable; + }), + }), + })), + }; + const handler = createA2AHandler({ + inkbox: { + getIdentity: vi.fn(async () => identity), + getClient: vi.fn(), + } as any, + sessions: {} as any, + state: createStateStore( + `${process.env.TMPDIR ?? "/tmp"}/opencode-a2a-${crypto.randomUUID()}`, + ), + logger, + }); + + await expect(handler.catchUp()).resolves.toBeUndefined(); + expect(logger.warn).toHaveBeenCalledWith("a2a.api_unavailable", { + error: "Error: HTTP 404: Not Found", + }); + }); }); From cce32ceb443306a39e3cf6b43824f608066c2df6 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Mon, 27 Jul 2026 06:34:50 +0000 Subject: [PATCH 11/16] Add live A2A protocol matrix --- .github/workflows/live-a2a.yml | 99 +++++++++ .github/workflows/live-stack.yml | 26 ++- tests/live/a2a_driver.py | 343 +++++++++++++++++++++++++++++++ 3 files changed, 463 insertions(+), 5 deletions(-) create mode 100644 .github/workflows/live-a2a.yml create mode 100644 tests/live/a2a_driver.py diff --git a/.github/workflows/live-a2a.yml b/.github/workflows/live-a2a.yml new file mode 100644 index 0000000..c33ae9b --- /dev/null +++ b/.github/workflows/live-a2a.yml @@ -0,0 +1,99 @@ +name: Live — Agent2Agent + +# Four real protocol legs cover both roles and conversation lengths: +# inbound/outbound × single-turn/multi-turn. The plugin and remote identities +# are preconfigured to allow one another in both directions. +on: + workflow_call: + inputs: + timeout_s: + description: "Seconds to wait for each A2A transition" + required: false + type: string + default: "300" + orchestrated: + description: "True when the full-stack workflow owns the shared live lock" + required: false + type: boolean + default: false + workflow_dispatch: + inputs: + timeout_s: + description: "Seconds to wait for each A2A transition" + required: false + default: "300" + +permissions: + contents: read + +concurrency: + group: ${{ inputs.orchestrated && format('inkbox-live-child-{0}', github.run_id) || 'inkbox-live-aut-tunnel' }} + cancel-in-progress: false + queue: max + +jobs: + a2a: + runs-on: ubuntu-latest + timeout-minutes: 40 + strategy: + fail-fast: false + max-parallel: 1 + matrix: + scenario: + - inbound-single + - inbound-multi + - outbound-single + - outbound-multi + + steps: + - uses: actions/checkout@v7 + + - uses: actions/setup-node@v7 + with: + node-version: 22 + cache: npm + + - uses: actions/setup-python@v6 + with: + python-version: "3.12" + + - name: Install protocol driver + run: pip install 'inkbox==0.5.6' + + - name: Install plugin and host + run: | + npm ci + npm install --no-save --package-lock=false \ + @inkbox/sdk@0.5.6 \ + @opencode-ai/sdk@1.17.18 @opencode-ai/plugin@1.17.18 + npm install -g opencode-ai@latest + + - name: Boot the AUT gateway + env: + MODE: real + AUT_INKBOX_API_KEY: ${{ secrets.AUT_INKBOX_API_KEY }} + AUT_INKBOX_SIGNING_KEY: ${{ secrets.AUT_INKBOX_SIGNING_KEY }} + INKBOX_BASE_URL: ${{ vars.INKBOX_BASE_URL }} + OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} + run: bash scripts/live-aut.sh + + - name: Run ${{ matrix.scenario }} + env: + A2A_SCENARIO: ${{ matrix.scenario }} + A2A_TIMEOUT_S: ${{ inputs.timeout_s || '300' }} + AUT_INKBOX_API_KEY: ${{ secrets.AUT_INKBOX_API_KEY }} + REMOTE_INKBOX_API_KEY: ${{ secrets.REMOTE_INKBOX_API_KEY }} + INKBOX_BASE_URL: ${{ vars.INKBOX_BASE_URL }} + run: python3 tests/live/a2a_driver.py + + - name: Dump gateway logs on failure + if: failure() + run: | + tail -n 300 "$AUT_GATEWAY_LOG" 2>/dev/null || true + tail -n 100 "$AUT_SERVE_LOG" 2>/dev/null || true + + - name: Stop gateway + if: always() + run: | + kill "$AUT_GATEWAY_PID" 2>/dev/null || true + kill "$AUT_SERVE_PID" 2>/dev/null || true diff --git a/.github/workflows/live-stack.yml b/.github/workflows/live-stack.yml index 1ab477a..f9a6e81 100644 --- a/.github/workflows/live-stack.yml +++ b/.github/workflows/live-stack.yml @@ -29,8 +29,22 @@ jobs: orchestrated: true secrets: inherit - voice: + a2a: needs: channels + if: > + !cancelled() && + ((github.event_name == 'pull_request' && github.event.pull_request.draft == false && + github.event.pull_request.head.repo.full_name == github.repository) || + github.event_name == 'workflow_dispatch' || + (github.event_name == 'workflow_run' && github.event.workflow_run.conclusion == 'success' && + github.event.workflow_run.head_branch == 'main')) + uses: ./.github/workflows/live-a2a.yml + with: + orchestrated: true + secrets: inherit + + voice: + needs: a2a if: > !cancelled() && ((github.event_name == 'pull_request' && github.event.pull_request.draft == false && @@ -62,7 +76,7 @@ jobs: full-stack: name: full-stack - needs: [channels, voice, external-events] + needs: [channels, a2a, voice, external-events] if: > !cancelled() && ((github.event_name == 'pull_request' && github.event.pull_request.draft == false && @@ -75,11 +89,12 @@ jobs: - name: Require every live suite to pass env: CHANNELS_RESULT: ${{ needs.channels.result }} + A2A_RESULT: ${{ needs.a2a.result }} VOICE_RESULT: ${{ needs.voice.result }} EXTERNAL_EVENTS_RESULT: ${{ needs.external-events.result }} run: | failed=0 - for suite in CHANNELS VOICE EXTERNAL_EVENTS; do + for suite in CHANNELS A2A VOICE EXTERNAL_EVENTS; do result_var="${suite}_RESULT" result="${!result_var}" echo "$suite: $result" @@ -88,13 +103,14 @@ jobs: exit "$failed" notify: - needs: [channels, voice, external-events, full-stack] + needs: [channels, a2a, voice, external-events, full-stack] if: > always() && github.event_name == 'workflow_run' && github.event.workflow_run.event == 'schedule' && github.event.workflow_run.conclusion == 'success' && github.event.workflow_run.head_branch == 'main' && - (needs.channels.result != 'success' || needs.voice.result != 'success' || + (needs.channels.result != 'success' || needs.a2a.result != 'success' || + needs.voice.result != 'success' || needs.external-events.result != 'success' || needs.full-stack.result != 'success') runs-on: ubuntu-latest timeout-minutes: 2 diff --git a/tests/live/a2a_driver.py b/tests/live/a2a_driver.py new file mode 100644 index 0000000..ba48bf7 --- /dev/null +++ b/tests/live/a2a_driver.py @@ -0,0 +1,343 @@ +#!/usr/bin/env python3 +"""Drive one real A2A scenario between the plugin and a remote identity.""" + +from __future__ import annotations + +import os +import time +import uuid +from typing import Any + +from inkbox import Inkbox + +STOPPED_WIRE_STATES = { + "TASK_STATE_COMPLETED", + "TASK_STATE_FAILED", + "TASK_STATE_CANCELED", + "TASK_STATE_REJECTED", + "TASK_STATE_INPUT_REQUIRED", + "TASK_STATE_AUTH_REQUIRED", +} + + +def _required_env(name: str) -> str: + value = os.environ.get(name, "").strip() + if not value: + raise RuntimeError(f"{name} is required") + return value + + +def _enum_value(value: Any) -> str: + return str(getattr(value, "value", value)) + + +def _identity(client: Inkbox): + mailboxes = client.mailboxes.list() + if len(mailboxes) != 1: + raise RuntimeError("Live A2A credentials must resolve to exactly one mailbox") + handle = mailboxes[0].email_address.split("@", 1)[0] + return client.get_identity(handle), handle + + +def _parts_text(parts: list[dict[str, Any]]) -> str: + return "\n".join( + str(part["text"]) + for part in parts + if isinstance(part, dict) and part.get("text") is not None + ) + + +def _wire_history_text(task: Any) -> str: + return "\n".join( + _parts_text(message.get("parts", [])) + for message in task.raw.get("history", []) + if isinstance(message, dict) + ) + + +def _rest_history_text(task: Any) -> str: + return "\n".join(_parts_text(message.parts) for message in task.messages) + + +def _wait_protocol_task( + a2a: Any, + target: Any, + task_id: str, + *, + expected: set[str], + timeout: float, +) -> Any: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + task = a2a.get_task(target, task_id, history_length=50) + state = _enum_value(task.state) + if state in expected: + return task + if state in STOPPED_WIRE_STATES: + raise AssertionError(f"A2A task stopped in unexpected state {state}") + time.sleep(1) + raise TimeoutError(f"A2A task did not reach {sorted(expected)} before timeout") + + +def _send_task(a2a: Any, target: Any, text: str) -> Any: + result = a2a.send(target, text=text, message_id=str(uuid.uuid4())) + if result.kind != "task" or result.task is None: + raise AssertionError("A2A SendMessage did not return a task") + return result.task + + +def _find_inbound_task(identity: Any, token: str, timeout: float) -> Any: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + page = identity.a2a_tasks(direction="inbound", limit=100) + for item in page.items: + if token in _rest_history_text(item): + return identity.a2a_task(item.id) + time.sleep(1) + raise TimeoutError("Plugin did not create the expected outbound A2A task") + + +def _wait_for_caller_message( + identity: Any, + task_id: str, + token: str, + timeout: float, +) -> Any: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + task = identity.a2a_task(task_id) + for message in task.messages: + if _enum_value(message.role) == "caller" and token in _parts_text(message.parts): + return task + state = _enum_value(task.state) + if state in {"completed", "failed", "canceled", "rejected"}: + raise AssertionError( + f"Outbound A2A task stopped before caller follow-up: {state}" + ) + time.sleep(1) + raise TimeoutError("Plugin did not answer the worker's input request") + + +def _assert_outer_is_open(a2a: Any, target: Any, task_id: str) -> None: + state = _enum_value(a2a.get_task(target, task_id).state) + if state in STOPPED_WIRE_STATES: + raise AssertionError("Plugin completed the outer task before its delegation") + + +def _cancel_if_open(a2a: Any, target: Any, task_id: str) -> None: + try: + state = _enum_value(a2a.get_task(target, task_id).state) + if state not in STOPPED_WIRE_STATES: + a2a.cancel(target, task_id) + except Exception: + pass + + +def _fail_if_open(identity: Any, task_id: str) -> None: + try: + task = identity.a2a_task(task_id) + if _enum_value(task.state) not in { + "completed", + "failed", + "canceled", + "rejected", + }: + identity.a2a_reply( + task_id, + intent="fail", + text="Automated A2A test cleanup.", + ) + except Exception: + pass + + +def _inbound_single(a2a: Any, target: Any, timeout: float, run: str) -> None: + completion = f"a2a-ci-inbound-single-{run}" + task = _send_task( + a2a, + target, + "Complete this A2A task without requesting input. " + f"The final answer must contain `{completion}`.", + ) + try: + final = _wait_protocol_task( + a2a, + target, + task.id, + expected={"TASK_STATE_COMPLETED"}, + timeout=timeout, + ) + if completion not in _wire_history_text(final): + raise AssertionError("Inbound single-turn completion token is missing") + finally: + _cancel_if_open(a2a, target, task.id) + + +def _inbound_multi(a2a: Any, target: Any, timeout: float, run: str) -> None: + answer = f"a2a-ci-answer-{run}" + completion = f"a2a-ci-inbound-multi-{run}" + task = _send_task( + a2a, + target, + "First call inkbox_a2a_ask_caller to request the access code. " + "Do not complete or fail the task before the caller responds. " + "After the caller replies, call inkbox_a2a_complete and include both " + f"the supplied code and `{completion}` in the final answer.", + ) + try: + waiting = _wait_protocol_task( + a2a, + target, + task.id, + expected={"TASK_STATE_INPUT_REQUIRED"}, + timeout=timeout, + ) + a2a.send( + target, + text=answer, + message_id=str(uuid.uuid4()), + context_id=waiting.context_id, + task_id=waiting.id, + ) + final = _wait_protocol_task( + a2a, + target, + task.id, + expected={"TASK_STATE_COMPLETED"}, + timeout=timeout, + ) + history = _wire_history_text(final) + if answer not in history or completion not in history: + raise AssertionError("Inbound multi-turn history is incomplete") + finally: + _cancel_if_open(a2a, target, task.id) + + +def _outbound_single( + a2a: Any, + target: Any, + remote_identity: Any, + remote_card_url: str, + timeout: float, + run: str, +) -> None: + inner_token = f"a2a-ci-outbound-single-task-{run}" + worker_result = f"a2a-ci-outbound-single-result-{run}" + outer = _send_task( + a2a, + target, + "Delegate a new task with inkbox_a2a_call to the Agent Card at " + f"{remote_card_url}. The delegated task text must contain " + f"`{inner_token}`. Wait for it with inkbox_a2a_check and do not " + "complete this outer task first. After the worker completes, call " + "inkbox_a2a_complete and include the worker's response in the answer.", + ) + inner = None + try: + inner = _find_inbound_task(remote_identity, inner_token, timeout) + _assert_outer_is_open(a2a, target, outer.id) + remote_identity.a2a_reply( + inner.id, + intent="complete", + text=worker_result, + ) + final = _wait_protocol_task( + a2a, + target, + outer.id, + expected={"TASK_STATE_COMPLETED"}, + timeout=timeout, + ) + if worker_result not in _wire_history_text(final): + raise AssertionError("Outbound single-turn worker result is missing") + finally: + if inner is not None: + _fail_if_open(remote_identity, inner.id) + _cancel_if_open(a2a, target, outer.id) + + +def _outbound_multi( + a2a: Any, + target: Any, + remote_identity: Any, + remote_card_url: str, + timeout: float, + run: str, +) -> None: + inner_token = f"a2a-ci-outbound-multi-task-{run}" + answer = f"a2a-ci-outbound-multi-answer-{run}" + worker_result = f"a2a-ci-outbound-multi-result-{run}" + outer = _send_task( + a2a, + target, + "Delegate a new task with inkbox_a2a_call to the Agent Card at " + f"{remote_card_url}. The delegated task text must contain " + f"`{inner_token}`. The worker will request input; reply with " + f"`{answer}` using inkbox_a2a_reply, then wait again with " + "inkbox_a2a_check. Do not complete this outer task before the worker. " + "Finally call inkbox_a2a_complete with the worker's response.", + ) + inner = None + try: + inner = _find_inbound_task(remote_identity, inner_token, timeout) + _assert_outer_is_open(a2a, target, outer.id) + remote_identity.a2a_reply( + inner.id, + intent="ask_caller", + text="Provide the requested verification value.", + ) + _wait_for_caller_message(remote_identity, inner.id, answer, timeout) + _assert_outer_is_open(a2a, target, outer.id) + remote_identity.a2a_reply( + inner.id, + intent="complete", + text=worker_result, + ) + final = _wait_protocol_task( + a2a, + target, + outer.id, + expected={"TASK_STATE_COMPLETED"}, + timeout=timeout, + ) + if worker_result not in _wire_history_text(final): + raise AssertionError("Outbound multi-turn worker result is missing") + finally: + if inner is not None: + _fail_if_open(remote_identity, inner.id) + _cancel_if_open(a2a, target, outer.id) + + +def main() -> None: + scenario = _required_env("A2A_SCENARIO") + timeout = float(os.environ.get("A2A_TIMEOUT_S", "240")) + base_url = os.environ.get("INKBOX_BASE_URL", "https://inkbox.ai").rstrip("/") + aut = Inkbox(api_key=_required_env("AUT_INKBOX_API_KEY"), base_url=base_url) + remote = Inkbox(api_key=_required_env("REMOTE_INKBOX_API_KEY"), base_url=base_url) + _, aut_handle = _identity(aut) + remote_identity, remote_handle = _identity(remote) + a2a = remote_identity.a2a_client() + target = a2a.fetch_card(f"{base_url}/a2a/{aut_handle}/card") + remote_card_url = f"{base_url}/a2a/{remote_handle}/card" + run = uuid.uuid4().hex[:12] + try: + if scenario == "inbound-single": + _inbound_single(a2a, target, timeout, run) + elif scenario == "inbound-multi": + _inbound_multi(a2a, target, timeout, run) + elif scenario == "outbound-single": + _outbound_single( + a2a, target, remote_identity, remote_card_url, timeout, run + ) + elif scenario == "outbound-multi": + _outbound_multi( + a2a, target, remote_identity, remote_card_url, timeout, run + ) + else: + raise RuntimeError(f"Unknown A2A_SCENARIO: {scenario}") + finally: + a2a.close() + + +if __name__ == "__main__": + main() From 8e0226e3fb671eae485e8c8a89529c2f39b660a7 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Mon, 27 Jul 2026 07:02:38 +0000 Subject: [PATCH 12/16] Use filtered A2A task discovery in live tests --- tests/live/a2a_driver.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/live/a2a_driver.py b/tests/live/a2a_driver.py index ba48bf7..70ea64a 100644 --- a/tests/live/a2a_driver.py +++ b/tests/live/a2a_driver.py @@ -89,10 +89,11 @@ def _send_task(a2a: Any, target: Any, text: str) -> Any: def _find_inbound_task(identity: Any, token: str, timeout: float) -> Any: deadline = time.monotonic() + timeout while time.monotonic() < deadline: - page = identity.a2a_tasks(direction="inbound", limit=100) + page = identity.a2a_tasks(direction="inbound", q=token, limit=10) for item in page.items: - if token in _rest_history_text(item): - return identity.a2a_task(item.id) + task = identity.a2a_task(item.id) + if token in _rest_history_text(task): + return task time.sleep(1) raise TimeoutError("Plugin did not create the expected outbound A2A task") From 9d7e48e99f2149363b8563ebf6335ab92350a1a4 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Mon, 27 Jul 2026 08:44:05 +0000 Subject: [PATCH 13/16] chore: lock Inkbox SDK 0.5.6 --- package-lock.json | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/package-lock.json b/package-lock.json index de0c302..f2bbd63 100644 --- a/package-lock.json +++ b/package-lock.json @@ -605,9 +605,9 @@ } }, "node_modules/@inkbox/sdk": { - "version": "0.5.1", - "resolved": "https://registry.npmjs.org/@inkbox/sdk/-/sdk-0.5.1.tgz", - "integrity": "sha512-YQPCDsPesZLoscAk+XdreHgjnMco8yeN1ZjMOWXhAvNuX7JVUKttVS5GTer5QtIQ/dFRY+pp9cWjavKkL2AwSw==", + "version": "0.5.6", + "resolved": "https://registry.npmjs.org/@inkbox/sdk/-/sdk-0.5.6.tgz", + "integrity": "sha512-tR9akg37T8+zP9zQ6QO59TDYaDZbbClYnnuRJhxbuTqpg5b0VkzbtJ6v2/TkQCGrr8mwIkizqW3zC5ov3RUv8w==", "license": "MIT", "dependencies": { "@peculiar/x509": "^2.0.0", From 1338d763f6290b041ecff6e0be58999492f19ed8 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Mon, 27 Jul 2026 08:48:54 +0000 Subject: [PATCH 14/16] ci: default the live Inkbox endpoint --- .github/workflows/live-a2a.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/live-a2a.yml b/.github/workflows/live-a2a.yml index c33ae9b..42d7fa7 100644 --- a/.github/workflows/live-a2a.yml +++ b/.github/workflows/live-a2a.yml @@ -73,7 +73,7 @@ jobs: MODE: real AUT_INKBOX_API_KEY: ${{ secrets.AUT_INKBOX_API_KEY }} AUT_INKBOX_SIGNING_KEY: ${{ secrets.AUT_INKBOX_SIGNING_KEY }} - INKBOX_BASE_URL: ${{ vars.INKBOX_BASE_URL }} + INKBOX_BASE_URL: ${{ vars.INKBOX_BASE_URL || 'https://inkbox.ai' }} OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} run: bash scripts/live-aut.sh @@ -83,7 +83,7 @@ jobs: A2A_TIMEOUT_S: ${{ inputs.timeout_s || '300' }} AUT_INKBOX_API_KEY: ${{ secrets.AUT_INKBOX_API_KEY }} REMOTE_INKBOX_API_KEY: ${{ secrets.REMOTE_INKBOX_API_KEY }} - INKBOX_BASE_URL: ${{ vars.INKBOX_BASE_URL }} + INKBOX_BASE_URL: ${{ vars.INKBOX_BASE_URL || 'https://inkbox.ai' }} run: python3 tests/live/a2a_driver.py - name: Dump gateway logs on failure From 78cc01ad03b4b6f072515b0c4579135ec9cda967 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Mon, 27 Jul 2026 08:58:44 +0000 Subject: [PATCH 15/16] Guide inbound A2A outcome handling --- src/gateway/a2a.ts | 10 +++++++++- tests/gateway/a2a.test.ts | 7 +++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/src/gateway/a2a.ts b/src/gateway/a2a.ts index aa4ede5..56b8f94 100644 --- a/src/gateway/a2a.ts +++ b/src/gateway/a2a.ts @@ -5,6 +5,10 @@ import type { StateStore } from "./state.js"; import type { GatewayLogger, SessionManager, VerifiedEvent } from "./types.js"; const TERMINAL = new Set(["completed", "failed", "canceled", "rejected"]); +const INBOUND_TASK_DIRECTIVE = + "You are handling an inbound A2A task. Resolve it with inkbox_a2a_complete, " + + "inkbox_a2a_ask_caller, or inkbox_a2a_fail. When caller input is needed, use " + + "inkbox_a2a_ask_caller and wait for the caller instead of completing the task."; interface A2AEventData { task_id: string; @@ -109,7 +113,11 @@ export function createA2AHandler(deps: { `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); + const reply = await deps.sessions.runA2A( + chatKey, + `${marker}\n${INBOUND_TASK_DIRECTIVE}\n${body}`.trim(), + context, + ); if ( !context.replyIntentCommitted && reply?.trim() && diff --git a/tests/gateway/a2a.test.ts b/tests/gateway/a2a.test.ts index 0628b56..4f5ef9d 100644 --- a/tests/gateway/a2a.test.ts +++ b/tests/gateway/a2a.test.ts @@ -66,6 +66,13 @@ describe("createA2AHandler", () => { }); }); expect(sessions.runA2A).toHaveBeenCalledTimes(1); + expect(sessions.runA2A).toHaveBeenCalledWith( + "a2a:identity-1:context-1", + expect.stringContaining( + "When caller input is needed, use inkbox_a2a_ask_caller and wait for the caller", + ), + expect.objectContaining({ taskId: "task-1", contextId: "context-1" }), + ); expect((state.read().a2aTasks as any)["task-1:message-1"].state).toBe("finalized"); }); From 1b646c7fa7e90d2019c4428f086b40a4cd1b3b42 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Mon, 27 Jul 2026 09:10:53 +0000 Subject: [PATCH 16/16] Share inbound A2A turn context with the host --- src/a2a-context.ts | 62 ++++++++++++++++++++++++++++++++++++++- src/gateway/a2a.ts | 13 ++++++-- src/tools/a2a.ts | 7 +++-- tests/gateway/a2a.test.ts | 29 ++++++++++++++++++ tests/unit/a2a.test.ts | 35 +++++++++++++++++++++- 5 files changed, 139 insertions(+), 7 deletions(-) diff --git a/src/a2a-context.ts b/src/a2a-context.ts index 8764742..9d8b778 100644 --- a/src/a2a-context.ts +++ b/src/a2a-context.ts @@ -1,3 +1,8 @@ +import * as crypto from "node:crypto"; +import * as fs from "node:fs"; +import * as path from "node:path"; +import { gatewayHome } from "./gateway/state.js"; + export interface ActiveA2ATurn { taskId: string; messageId: string; @@ -7,14 +12,69 @@ export interface ActiveA2ATurn { const turns = new Map(); +export function a2aTurnContextPath(sessionID: string): string { + const digest = crypto.createHash("sha256").update(sessionID).digest("hex"); + return path.join(gatewayHome(), "a2a-turn-contexts", `${digest}.json`); +} + +function sameTurn(left: ActiveA2ATurn, right: ActiveA2ATurn): boolean { + return ( + left.taskId === right.taskId && + left.messageId === right.messageId && + left.contextId === right.contextId + ); +} + +function readTurn(sessionID: string): ActiveA2ATurn | undefined { + try { + const value = JSON.parse(fs.readFileSync(a2aTurnContextPath(sessionID), "utf8")); + if ( + !value || + typeof value !== "object" || + typeof value.taskId !== "string" || + typeof value.messageId !== "string" || + typeof value.contextId !== "string" || + typeof value.replyIntentCommitted !== "boolean" + ) { + return undefined; + } + return value; + } catch { + return undefined; + } +} + +function writeTurn(sessionID: string, turn: ActiveA2ATurn): void { + const target = a2aTurnContextPath(sessionID); + fs.mkdirSync(path.dirname(target), { recursive: true, mode: 0o700 }); + const tmp = `${target}.${process.pid}.tmp`; + fs.writeFileSync(tmp, `${JSON.stringify(turn)}\n`, { mode: 0o600 }); + fs.renameSync(tmp, target); + fs.chmodSync(target, 0o600); +} + export function setActiveA2ATurn(sessionID: string, turn: ActiveA2ATurn): void { turns.set(sessionID, turn); + writeTurn(sessionID, turn); } export function clearActiveA2ATurn(sessionID: string, turn: ActiveA2ATurn): void { if (turns.get(sessionID) === turn) turns.delete(sessionID); + const persisted = readTurn(sessionID); + if (!persisted || !sameTurn(persisted, turn)) return; + turn.replyIntentCommitted = persisted.replyIntentCommitted; + try { + fs.unlinkSync(a2aTurnContextPath(sessionID)); + } catch (error: any) { + if (error?.code !== "ENOENT") throw error; + } } export function activeA2ATurn(sessionID: string): ActiveA2ATurn | undefined { - return turns.get(sessionID); + return turns.get(sessionID) ?? readTurn(sessionID); +} + +export function commitActiveA2ATurn(sessionID: string, turn: ActiveA2ATurn): void { + turn.replyIntentCommitted = true; + writeTurn(sessionID, turn); } diff --git a/src/gateway/a2a.ts b/src/gateway/a2a.ts index 56b8f94..2171e44 100644 --- a/src/gateway/a2a.ts +++ b/src/gateway/a2a.ts @@ -4,7 +4,14 @@ 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"]); +const TURN_STOPPED = new Set([ + "input_required", + "auth_required", + "completed", + "failed", + "canceled", + "rejected", +]); const INBOUND_TASK_DIRECTIVE = "You are handling an inbound A2A task. Resolve it with inkbox_a2a_complete, " + "inkbox_a2a_ask_caller, or inkbox_a2a_fail. When caller input is needed, use " + @@ -124,7 +131,7 @@ export function createA2AHandler(deps: { reply.trim().toUpperCase() !== "[SILENT]" ) { const task = await id.a2aTask(taskId); - if (!TERMINAL.has(String(task.state))) { + if (!TURN_STOPPED.has(String(task.state))) { await id.a2aReply(taskId, { intent: "complete", text: reply }); } } @@ -212,7 +219,7 @@ export function createA2AHandler(deps: { if (entry.state === "finalized") continue; try { const task = await id.a2aTask(entry.taskId); - if (TERMINAL.has(String(task.state))) { + if (TURN_STOPPED.has(String(task.state))) { persist(deps.state, key, entry.data, "finalized"); } else { start(key, entry.data); diff --git a/src/tools/a2a.ts b/src/tools/a2a.ts index a3bca5f..b876c4b 100644 --- a/src/tools/a2a.ts +++ b/src/tools/a2a.ts @@ -1,5 +1,5 @@ import { z } from "zod"; -import { activeA2ATurn } from "../a2a-context.js"; +import { activeA2ATurn, commitActiveA2ATurn } from "../a2a-context.js"; import { findDelegationByTask, promoteAfterSend, recordBeforeSend } from "../a2a-delegations.js"; import { runTool } from "../errors.js"; import { formatJson } from "../format.js"; @@ -321,13 +321,16 @@ async function inboundIntent( if (!context) { throw new Error("This tool is only available during an inbound A2A task"); } + if (context.replyIntentCommitted) { + throw new Error("This inbound A2A task already has an outcome"); + } 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; + commitActiveA2ATurn(sessionID, context); return formatJson(result); }); } diff --git a/tests/gateway/a2a.test.ts b/tests/gateway/a2a.test.ts index 4f5ef9d..47437aa 100644 --- a/tests/gateway/a2a.test.ts +++ b/tests/gateway/a2a.test.ts @@ -76,6 +76,35 @@ describe("createA2AHandler", () => { expect((state.read().a2aTasks as any)["task-1:message-1"].state).toBe("finalized"); }); + it("does not overwrite an input-required outcome with default completion", async () => { + const a2aReply = vi.fn(); + const identity = { + id: "identity-1", + a2aTask: vi.fn(async () => ({ id: "task-1", state: "input_required" })), + a2aReply, + }; + const handler = createA2AHandler({ + inkbox: { + getIdentity: vi.fn(async () => identity), + getClient: vi.fn(), + } as any, + sessions: { + runA2A: vi.fn(async () => "Which region?"), + } as any, + state: createStateStore( + `${process.env.TMPDIR ?? "/tmp"}/opencode-a2a-${crypto.randomUUID()}`, + ), + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }); + + await handler.handle(event()); + await vi.waitFor(() => { + expect(identity.a2aTask).toHaveBeenCalledWith("task-1"); + }); + + expect(a2aReply).not.toHaveBeenCalled(); + }); + it("cancels only the addressed task on its context session", async () => { const abortA2A = vi.fn(async () => true); const handler = createA2AHandler({ diff --git a/tests/unit/a2a.test.ts b/tests/unit/a2a.test.ts index eda5579..ac20030 100644 --- a/tests/unit/a2a.test.ts +++ b/tests/unit/a2a.test.ts @@ -1,5 +1,7 @@ +import * as fs from "node:fs"; +import * as path from "node:path"; import { beforeEach, describe, expect, it, vi } from "vitest"; -import { clearActiveA2ATurn, setActiveA2ATurn } from "../../src/a2a-context.js"; +import { a2aTurnContextPath, clearActiveA2ATurn, setActiveA2ATurn } from "../../src/a2a-context.js"; import { defaultGatewayConfig } from "../../src/config.js"; import { a2aTools } from "../../src/tools/a2a.js"; @@ -223,4 +225,35 @@ describe("a2aTools", () => { }); expect(context.replyIntentCommitted).toBe(true); }); + + it("shares inbound turn authorization with the separate host process", async () => { + const { deps, identity } = makeDeps(); + const ctx = makeCtx(); + const context = { + taskId: "task-cross-process", + messageId: "message-cross-process", + contextId: "context-cross-process", + replyIntentCommitted: false, + }; + const target = a2aTurnContextPath(ctx.sessionID); + fs.mkdirSync(path.dirname(target), { recursive: true }); + fs.writeFileSync(target, `${JSON.stringify(context)}\n`, { mode: 0o600 }); + + const result = await getTool("inkbox_a2a_ask_caller", deps).definition.execute( + { text: "Which region?" }, + ctx, + ); + + expect(result).toContain("ask_caller"); + expect(identity.a2aReply).toHaveBeenCalledWith("task-cross-process", { + intent: "ask_caller", + text: "Which region?", + }); + expect(JSON.parse(fs.readFileSync(target, "utf8")).replyIntentCommitted).toBe(true); + expect(fs.statSync(target).mode & 0o777).toBe(0o600); + + clearActiveA2ATurn(ctx.sessionID, context); + expect(context.replyIntentCommitted).toBe(true); + expect(fs.existsSync(target)).toBe(false); + }); });