From 737e7fd83bde7ed04d8a42a9c250ad8f16341524 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Thu, 23 Jul 2026 09:38:13 +0000 Subject: [PATCH 1/8] feat: add A2A tools for Codex --- inkbox_codex/tools.py | 48 ++++++++++++++++++++++++++++++ pyproject.toml | 2 +- tests/test_tools.py | 68 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 117 insertions(+), 1 deletion(-) diff --git a/inkbox_codex/tools.py b/inkbox_codex/tools.py index 8134f2f..57814ee 100644 --- a/inkbox_codex/tools.py +++ b/inkbox_codex/tools.py @@ -265,6 +265,36 @@ def _str_list(desc: str = "") -> JsonSchema: "Remove a contact from the address book by contact id. Look it up first to confirm the target.", _schema({"contact_id": _str("Contact id.")}, ["contact_id"]), ), + ToolSpec( + "inkbox_a2a_call", + "Send a task to an A2A 1.0 Agent Card.", + _schema({ + "card_url": _str("Agent Card URL."), + "text": _str("Task text."), + "context_id": _str("Optional context to continue."), + "task_id": _str("Optional task requesting more input."), + "message_id": _str("Stable idempotency id."), + }, ["card_url", "text"]), + ), + ToolSpec( + "inkbox_a2a_check", + "Fetch an A2A task, or wait until it stops.", + _schema({ + "card_url": _str("Agent Card URL."), + "task_id": _str("Remote task id."), + "wait": {"type": "boolean"}, + }, ["card_url", "task_id"]), + ), + ToolSpec( + "inkbox_a2a_reply", + "Reply to a remote A2A task that requested more input.", + _schema({ + "card_url": _str("Agent Card URL."), + "task_id": _str("Remote task id."), + "text": _str("Reply text."), + "message_id": _str("Stable idempotency id."), + }, ["card_url", "task_id", "text"]), + ), ] @@ -695,6 +725,24 @@ def _run() -> Any: client.contacts.delete(str(args["contact_id"])) return {"deleted": str(args["contact_id"])} + if name in {"inkbox_a2a_call", "inkbox_a2a_check", "inkbox_a2a_reply"}: + a2a = _identity().a2a_client() + try: + target = a2a.fetch_card(str(args["card_url"])) + if name == "inkbox_a2a_check": + if args.get("wait"): + return a2a.wait(target, str(args["task_id"])) + return a2a.get_task(target, str(args["task_id"])) + return a2a.send( + target, + text=str(args["text"]), + context_id=args.get("context_id") or None, + task_id=args.get("task_id") or None, + message_id=args.get("message_id") or None, + ) + finally: + a2a.close() + raise ValueError(f"unknown Inkbox tool: {name}") try: diff --git a/pyproject.toml b/pyproject.toml index bb2b7a1..6a25292 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "codex-plugin" -version = "0.1.4" +version = "0.1.5" description = "Inkbox bridge for Codex — talk to your coding agent over email, SMS, iMessage, and voice" requires-python = ">=3.11" dependencies = [ diff --git a/tests/test_tools.py b/tests/test_tools.py index 91492fb..6131736 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -51,6 +51,7 @@ def __init__(self): self.transcript_call_id = None self.sent_texts = [] self.sent_imessages = [] + self.a2a = _FakeA2AClient() def place_call(self, **kwargs): self.place_call_kwargs = kwargs @@ -75,6 +76,34 @@ def send_text(self, **kwargs): self.sent_texts.append(kwargs) return type("Message", (), {"id": "sms-1"})() + def a2a_client(self): + return self.a2a + + +class _FakeA2AClient: + def __init__(self): + self.calls = [] + self.closed = False + + def fetch_card(self, card_url): + self.calls.append(("fetch_card", card_url)) + return {"rpc_url": "https://target.example/a2a"} + + def send(self, target, **kwargs): + self.calls.append(("send", target, kwargs)) + return {"kind": "task", "task": {"id": "task-1"}} + + def get_task(self, target, task_id): + self.calls.append(("get_task", target, task_id)) + return {"id": task_id, "state": "TASK_STATE_WORKING"} + + def wait(self, target, task_id): + self.calls.append(("wait", target, task_id)) + return {"id": task_id, "state": "TASK_STATE_COMPLETED"} + + def close(self): + self.closed = True + class _FakeContacts: def __init__(self): @@ -131,6 +160,9 @@ def test_coding_agent_tool_tier_is_registered(): "inkbox_create_contact", "inkbox_update_contact", "inkbox_delete_contact", + "inkbox_a2a_call", + "inkbox_a2a_check", + "inkbox_a2a_reply", } assert names == expected @@ -147,6 +179,42 @@ def test_get_and_delete_contact_tools(): assert client.contacts.deleted == ["contact-1"] +def test_a2a_tools_send_check_and_reply(): + client = _FakeClient() + card_url = "https://target.example/card" + + sent = _call( + client, + "inkbox_a2a_call", + {"card_url": card_url, "text": "Investigate.", "message_id": "msg-1"}, + ) + checked = _call( + client, + "inkbox_a2a_check", + {"card_url": card_url, "task_id": "task-1", "wait": True}, + ) + replied = _call( + client, + "inkbox_a2a_reply", + { + "card_url": card_url, + "task_id": "task-1", + "text": "More context.", + "message_id": "msg-2", + }, + ) + + assert sent["task"]["id"] == "task-1" + assert checked["state"] == "TASK_STATE_COMPLETED" + assert replied["task"]["id"] == "task-1" + assert ( + "wait", + {"rpc_url": "https://target.example/a2a"}, + "task-1", + ) in client.identity.a2a.calls + assert client.identity.a2a.closed is True + + def test_place_call_writes_context_and_tags_websocket_url(tmp_path, monkeypatch): monkeypatch.setenv("INKBOX_CODEX_HOME", str(tmp_path)) client = _FakeClient() From 8351df44e527c751203b23003ea08aba96a26450 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Thu, 23 Jul 2026 10:31:45 +0000 Subject: [PATCH 2/8] feat: serve and track A2A delegations --- README.md | 6 + inkbox_codex/a2a_delegations.py | 106 ++++++++++++++ inkbox_codex/config.py | 10 ++ inkbox_codex/gateway.py | 240 +++++++++++++++++++++++++++++++- inkbox_codex/sessions.py | 33 ++++- inkbox_codex/tools.py | 132 +++++++++++++++++- tests/test_a2a_gateway.py | 172 +++++++++++++++++++++++ tests/test_tools.py | 72 +++++++++- 8 files changed, 756 insertions(+), 15 deletions(-) create mode 100644 inkbox_codex/a2a_delegations.py create mode 100644 tests/test_a2a_gateway.py diff --git a/README.md b/README.md index 64eb514..300cd64 100644 --- a/README.md +++ b/README.md @@ -261,6 +261,12 @@ The agent reaches you (or third parties) through an in-process MCP server: - `inkbox_list_imessage_conversations` · `inkbox_get_imessage_conversation` — browse iMessage threads and history (find the `conversation_id` to send into). - `inkbox_lookup_contact` · `inkbox_list_contacts` · `inkbox_get_contact` — resolve and read address-book contacts (reverse-lookup by email/phone, free-text search, or full record by id). - `inkbox_create_contact` · `inkbox_update_contact` · `inkbox_delete_contact` — save, edit, and remove organization-wide contacts. Changes affect the shared address book. vCard export/import is not exposed. +- `inkbox_a2a_call` · `inkbox_a2a_check` · `inkbox_a2a_reply` — delegate work to another agent and follow its task. +- `inkbox_a2a_complete` · `inkbox_a2a_ask_caller` · `inkbox_a2a_fail` — commit the outcome of a verified inbound A2A task. These tools are rejected outside that task's isolated session. + +Inbound A2A serving and A2A client tools require Inkbox SDK 0.5.5 or newer. +Other bridge features continue to support the base SDK range in +`pyproject.toml`. On a live call, the OpenAI Realtime voice agent additionally gets `consult_agent`, `register_post_call_action` / `edit_post_call_action` / `delete_post_call_action`, and `hang_up_call` — see [Voice](#voice). diff --git a/inkbox_codex/a2a_delegations.py b/inkbox_codex/a2a_delegations.py new file mode 100644 index 0000000..e604e4a --- /dev/null +++ b/inkbox_codex/a2a_delegations.py @@ -0,0 +1,106 @@ +"""Durable caller-side A2A delegation routing records.""" + +from __future__ import annotations + +import json +import os +import threading +import time +from pathlib import Path +from typing import Any, Dict, Optional +from urllib.parse import urlsplit + +_LOCK = threading.Lock() + + +def _path() -> Path: + root = Path( + os.getenv("INKBOX_CODEX_HOME") + or (Path.home() / ".inkbox-codex") + ) + return root / "a2a_delegations.json" + + +def _origin(url: str) -> str: + parsed = urlsplit(url) + return f"{parsed.scheme.lower()}://{parsed.netloc.lower()}" + + +def _read() -> Dict[str, Dict[str, Any]]: + try: + loaded = json.loads(_path().read_text()) + return loaded if isinstance(loaded, dict) else {} + except FileNotFoundError: + return {} + + +def _write(records: Dict[str, Dict[str, Any]]) -> None: + path = _path() + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(records, indent=2, sort_keys=True) + "\n") + tmp.chmod(0o600) + os.replace(tmp, path) + path.chmod(0o600) + + +def record_before_send( + *, + identity_id: str, + rpc_url: str, + card_url: str, + message_id: str, + session_key: Optional[str], + context_id: Optional[str] = None, + task_id: Optional[str] = None, +) -> str: + """Persist retry and routing data before the SendMessage request.""" + origin = _origin(rpc_url) + context_key = context_id or f"pending:{message_id}" + key = f"{identity_id}|{origin}|{context_key}" + with _LOCK: + records = _read() + records[key] = { + "identity_id": identity_id, + "origin": origin, + "card_url": card_url, + "context_id": context_id, + "task_id": task_id, + "message_id": message_id, + "session_key": session_key, + "updated_at": time.time(), + } + _write(records) + return key + + +def promote_after_send( + pending_key: str, + *, + context_id: str, + task_id: str, +) -> None: + """Promote a provisional record to its canonical context key.""" + with _LOCK: + records = _read() + record = records.pop(pending_key, None) + if record is None: + return + record["context_id"] = context_id + record["task_id"] = task_id + record["updated_at"] = time.time() + key = ( + f"{record['identity_id']}|{record['origin']}|{context_id}" + ) + records[key] = record + _write(records) + + +def find_by_task(task_id: str) -> Optional[Dict[str, Any]]: + """Return the newest local delegation record for a remote task.""" + matches = [ + record + for record in _read().values() + if str(record.get("task_id") or "") == task_id + ] + return max(matches, key=lambda item: item.get("updated_at", 0), default=None) diff --git a/inkbox_codex/config.py b/inkbox_codex/config.py index 9de7c9b..ef122a9 100644 --- a/inkbox_codex/config.py +++ b/inkbox_codex/config.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import os from dataclasses import dataclass, field from pathlib import Path @@ -51,6 +52,15 @@ def channel_hints_path() -> Path: return root / "channel_hints.json" +def a2a_turn_context_path(chat_id: str) -> Path: + """Return the trusted cross-process context file for one Codex session.""" + root = Path(os.getenv("INKBOX_CODEX_HOME") or (Path.home() / ".inkbox-codex")) + path = root / "a2a_turn_contexts" + path.mkdir(parents=True, exist_ok=True) + digest = hashlib.sha256(chat_id.encode()).hexdigest() + return path / f"{digest}.json" + + def env_flag(name: str, default: bool = False) -> bool: raw = os.getenv(name) if raw is None: diff --git a/inkbox_codex/gateway.py b/inkbox_codex/gateway.py index f099421..fc5e4dc 100644 --- a/inkbox_codex/gateway.py +++ b/inkbox_codex/gateway.py @@ -67,6 +67,7 @@ call_contexts_dir, inkbox_client_kwargs, ) + from .a2a_delegations import find_by_task as find_a2a_delegation from .media import download_media, inbound_media_note from .prompts import contact_marker, strip_markdown from .realtime import ( @@ -79,6 +80,7 @@ from .webhook_providers import match_provider except ImportError: # pragma: no cover - direct local import/test fallback from config import DEFAULT_WEBHOOK_PATH, INKBOX_WS_PATH, BridgeConfig, call_contexts_dir, inkbox_client_kwargs + from a2a_delegations import find_by_task as find_a2a_delegation from media import download_media, inbound_media_note from prompts import contact_marker, strip_markdown from realtime import ( @@ -333,6 +335,13 @@ def _voice_consult_prompt( "imessage.delivery_failed", "imessage.reaction_received", ] +A2A_EVENTS = [ + "a2a.task.created", + "a2a.task.message", + "a2a.task.canceled", + "a2a.sent_task.updated", +] +A2A_TERMINAL_STATES = {"completed", "failed", "canceled", "rejected"} # Mail: inbound plus the two delivery-failure transitions that feed the loop # (_on_mail_delivery_failed). The success transitions stay unsubscribed — they # would pay signature cost on every outbound email for no behaviour. @@ -495,6 +504,10 @@ def __init__(self, cfg: BridgeConfig): # delivery-failure loop can stop waking the agent after # OUTBOUND_FAILURE_MAX_ATTEMPTS. Reset on inbound / delivered / TTL. self._outbound_failure_state: Dict[str, Dict[str, float]] = {} + self._a2a_registry_path = ( + Path.home() / ".inkbox-codex" / "a2a_tasks.json" + ) + self._a2a_jobs: Dict[str, set[asyncio.Task[Any]]] = {} # ------------------------------------------------------------------ # Lifecycle @@ -548,6 +561,7 @@ async def run(self) -> None: health_fn=self.health_report, on_send_failure=self._note_sync_send_failure, ) + await self._catch_up_a2a_tasks() logger.info( "[bridge] ready — %s / %s / %s → Codex in %s", @@ -657,9 +671,11 @@ def _reconcile(owner_kw: Dict[str, Any], event_types: List[str]) -> None: self.cfg.identity, webhook_url, ws_url, ) + identity_events = list(A2A_EVENTS) if getattr(identity, "imessage_enabled", False): - _reconcile({"agent_identity_id": identity.id}, IMESSAGE_EVENTS) - logger.info("[bridge] iMessage for %s → %s", self.cfg.identity, webhook_url) + identity_events = [*IMESSAGE_EVENTS, *identity_events] + _reconcile({"agent_identity_id": identity.id}, identity_events) + logger.info("[bridge] identity events for %s → %s", self.cfg.identity, webhook_url) async def _cleanup(self) -> None: if self.sessions is not None: @@ -760,7 +776,9 @@ def _is_known_inkbox_event(self, event_type: "str | None", envelope: Dict[str, A Returns: bool: True for a recognised Inkbox event shape. """ - if event_type and event_type.startswith(("message.", "text.", "imessage.")): + if event_type and event_type.startswith( + ("message.", "text.", "imessage.", "a2a.") + ): return True explicit_call_id = envelope.get("call_id") or envelope.get("callId") generic_id = envelope.get("id") @@ -875,6 +893,8 @@ async def _route_inkbox_event( return await self._on_imessage_received(envelope) if event_type == "imessage.reaction_received": return await self._on_imessage_reaction_received(envelope) + if event_type.startswith("a2a."): + return await self._on_a2a_event(envelope) # Outbound delivery failures: tell the agent its message didn't land so # it can retry or reach the human another way. if event_type == "text.delivery_failed": @@ -1036,6 +1056,220 @@ def _field(*names: str) -> str: await self.sessions.get(chat_id).handle_inbound(text, "external", meta) return web.json_response({"ok": True, "external": source_name}) + def _read_a2a_registry(self) -> Dict[str, Any]: + try: + loaded = json.loads(self._a2a_registry_path.read_text()) + return loaded if isinstance(loaded, dict) else {} + except FileNotFoundError: + return {} + + def _write_a2a_registry( + self, + key: str, + data: Dict[str, Any], + state: str, + ) -> None: + current = self._read_a2a_registry() + current[key] = { + "task_id": str(data.get("task_id") or ""), + "message_id": str(data.get("message_id") or ""), + "context_id": str(data.get("context_id") or ""), + "state": state, + "updated_at": time.time(), + } + self._a2a_registry_path.parent.mkdir(parents=True, exist_ok=True) + tmp = self._a2a_registry_path.with_suffix(".tmp") + tmp.write_text(json.dumps(current, indent=2, sort_keys=True) + "\n") + tmp.chmod(0o600) + os.replace(tmp, self._a2a_registry_path) + self._a2a_registry_path.chmod(0o600) + + def _track_a2a_job( + self, + task_id: str, + registry_key: str, + data: Dict[str, Any], + ) -> None: + job = asyncio.create_task(self._run_a2a_turn(registry_key, data)) + self._a2a_jobs.setdefault(task_id, set()).add(job) + job.add_done_callback( + lambda done, value=task_id: self._a2a_jobs.get(value, set()).discard(done) + ) + + @staticmethod + def _a2a_event_data(task: Any) -> Dict[str, Any]: + message = task.messages[-1] if task.messages else None + return { + "task_id": str(task.id), + "context_id": str(task.context_id), + "state": str(getattr(task.state, "value", task.state)), + "caller": { + "identity_id": str(task.caller.identity_id), + "organization_id": task.caller.organization_id, + "handle": task.caller.handle, + }, + "message_id": ( + str(message.message_id) + if message is not None + else f"task:{task.id}" + ), + "parts": message.parts if message is not None else [], + } + + async def _on_a2a_event( + self, + envelope: Dict[str, Any], + ) -> "web.Response": + event_type = str(envelope.get("event_type") or "") + data = envelope.get("data") if isinstance(envelope.get("data"), dict) else {} + task_id = str(data.get("task_id") or "") + context_id = str(data.get("context_id") or "") + message_id = str(data.get("message_id") or envelope.get("id") or "") + if not task_id or not context_id: + return web.json_response({"ok": True, "ignored": "invalid-a2a-event"}) + if event_type == "a2a.task.canceled": + for job in list(self._a2a_jobs.get(task_id, set())): + job.cancel() + self._a2a_jobs.pop(task_id, None) + return web.json_response({"ok": True}) + if event_type == "a2a.sent_task.updated": + delegation = find_a2a_delegation(task_id) + session_key = str((delegation or {}).get("session_key") or "") + if self.sessions is not None and session_key: + parts = data.get("parts") if isinstance(data.get("parts"), list) else [] + text = "\n".join( + str(part.get("text")) + for part in parts + if isinstance(part, dict) and part.get("text") + ) + prompt = ( + f"[inkbox:a2a_sent_task_updated task_id={task_id} " + f"context_id={context_id} state={data.get('state') or 'unknown'}]\n" + "An A2A task you delegated changed state. Use " + "inkbox_a2a_check or inkbox_a2a_reply with the stored " + f"Agent Card URL {(delegation or {}).get('card_url') or 'unknown'} " + "if follow-up is needed." + ) + if text: + prompt = f"{prompt}\n\nRemote agent message:\n{text}" + await self.sessions.get(session_key).handle_inbound( + prompt, + "external", + { + "a2a_task_id": task_id, + "a2a_context_id": context_id, + }, + ) + else: + logger.info( + "[bridge] outbound A2A task updated without a local session: %s", + task_id, + ) + return web.json_response({"ok": True}) + + key = f"{task_id}:{message_id}" + if key in self._read_a2a_registry(): + return web.json_response({"ok": True, "deduped": True}) + self._write_a2a_registry(key, data, "queued") + self._track_a2a_job(task_id, key, data) + return web.json_response({"ok": True}) + + async def _run_a2a_turn( + self, + registry_key: str, + data: Dict[str, Any], + ) -> None: + task_id = str(data["task_id"]) + context_id = str(data["context_id"]) + caller = data.get("caller") if isinstance(data.get("caller"), dict) else {} + parts = data.get("parts") if isinstance(data.get("parts"), list) else [] + text = "\n".join( + str(part.get("text")) + for part in parts + if isinstance(part, dict) and part.get("text") + ) + marker = ( + f"[inkbox:a2a_task caller=@{str(caller.get('handle') or 'unknown').lstrip('@')} " + f"caller_org={caller.get('organization_id') or 'unknown'}]" + ) + context = { + "task_id": task_id, + "message_id": str(data.get("message_id") or ""), + "context_id": context_id, + "reply_intent_committed": False, + } + self._write_a2a_registry(registry_key, data, "running") + try: + if self.sessions is None: + return + reply = await self.sessions.get( + f"a2a:{self._identity.id}:{context_id}" + ).run_consult( + f"{marker}\n{text}".rstrip(), + a2a_context=context, + ) + if ( + not context["reply_intent_committed"] + and reply.strip() + and reply.strip().upper() != "[SILENT]" + ): + authoritative = await asyncio.to_thread( + self._identity.a2a_task, task_id + ) + state = str( + getattr(authoritative.state, "value", authoritative.state) + ) + if state not in A2A_TERMINAL_STATES: + await asyncio.to_thread( + self._identity.a2a_reply, + task_id, + intent="complete", + text=reply, + ) + self._write_a2a_registry(registry_key, data, "finalized") + except asyncio.CancelledError: + authoritative = await asyncio.to_thread( + self._identity.a2a_task, task_id + ) + state = str(getattr(authoritative.state, "value", authoritative.state)) + if state in A2A_TERMINAL_STATES: + self._write_a2a_registry(registry_key, data, "finalized") + raise + except Exception: + logger.exception("[bridge] A2A turn failed: %s", task_id) + + async def _catch_up_a2a_tasks(self) -> None: + try: + for key, entry in self._read_a2a_registry().items(): + if entry.get("state") == "finalized": + continue + task_id = str(entry.get("task_id") or "") + if not task_id or self._a2a_jobs.get(task_id): + continue + full = await asyncio.to_thread(self._identity.a2a_task, task_id) + state = str(getattr(full.state, "value", full.state)) + data = self._a2a_event_data(full) + if state in A2A_TERMINAL_STATES: + self._write_a2a_registry(key, data, "finalized") + else: + self._track_a2a_job(task_id, key, data) + + tasks = await asyncio.to_thread( + lambda: list(self._identity.iter_a2a_tasks(state="submitted")) + ) + for task in tasks: + full = await asyncio.to_thread(self._identity.a2a_task, task.id) + data = self._a2a_event_data(full) + await self._on_a2a_event( + { + "id": f"catchup:{task.id}:{data['message_id']}", + "event_type": "a2a.task.created", + "data": data, + } + ) + except Exception: + logger.exception("[bridge] A2A catch-up failed") + @staticmethod def _thread_key(prefix: str, value: Any) -> Optional[str]: raw = str(value or "").strip() diff --git a/inkbox_codex/sessions.py b/inkbox_codex/sessions.py index a515c73..2f04925 100644 --- a/inkbox_codex/sessions.py +++ b/inkbox_codex/sessions.py @@ -22,7 +22,7 @@ try: from .codex_client import CodexAppServerClient, CodexAppServerError - from .config import BridgeConfig + from .config import BridgeConfig, a2a_turn_context_path from .escalation import ( PendingInteraction, format_codex_approval_request, @@ -33,7 +33,7 @@ from .prompts import build_channel_prompt, frame_inbound except ImportError: # pragma: no cover - direct local import/test fallback from codex_client import CodexAppServerClient, CodexAppServerError - from config import BridgeConfig + from config import BridgeConfig, a2a_turn_context_path from escalation import ( PendingInteraction, format_codex_approval_request, @@ -77,6 +77,7 @@ class _Turn: # True for a one-shot turn spawned to recover from a rejected reply send. # A recovery turn that itself fails to send is not recovered again (no loop). recovery: bool = False + a2a_context: Optional[Dict[str, Any]] = None # Leading slash-commands the human can text to steer the conversation itself. # The bridge acts on these locally — they never reach Codex as a turn. @@ -629,6 +630,14 @@ async def _run_turn(self, turn: _Turn) -> None: self._interrupting = False # fresh turn starts un-interrupted self._current_turn = turn typing_task: Optional[asyncio.Task] = None + a2a_context_path: Optional[Path] = None + if turn.a2a_context is not None: + a2a_context_path = a2a_turn_context_path(self.chat_id) + tmp = a2a_context_path.with_suffix(".tmp") + tmp.write_text(json.dumps(turn.a2a_context, sort_keys=True) + "\n") + tmp.chmod(0o600) + os.replace(tmp, a2a_context_path) + a2a_context_path.chmod(0o600) try: client = await self._ensure_client() # Keep a typing indicator alive on the human's channel for the whole @@ -664,6 +673,15 @@ async def _run_turn(self, turn: _Turn) -> None: return raise finally: + if a2a_context_path is not None: + try: + persisted = json.loads(a2a_context_path.read_text()) + turn.a2a_context["reply_intent_committed"] = bool( + persisted.get("reply_intent_committed") + ) + except (FileNotFoundError, json.JSONDecodeError): + pass + a2a_context_path.unlink(missing_ok=True) self._turn_active = False self._current_turn = None if typing_task is not None: @@ -726,7 +744,12 @@ async def _deliver_reply(self, turn: _Turn, reply: str) -> None: _Turn(text=_send_rejected_prompt(reply, reason), recovery=True) ) - async def run_consult(self, query: str) -> str: + async def run_consult( + self, + query: str, + *, + a2a_context: Optional[Dict[str, Any]] = None, + ) -> str: """Run one Codex turn and RETURN its text (don't send it). Used by the Realtime voice bridge, post-call actions, and delivery- @@ -746,7 +769,9 @@ async def run_consult(self, query: str) -> str: """ loop = asyncio.get_running_loop() future: asyncio.Future[str] = loop.create_future() - await self._queue.put(_Turn(text=query, future=future)) + await self._queue.put( + _Turn(text=query, future=future, a2a_context=a2a_context) + ) if self._worker is None or self._worker.done(): self._worker = asyncio.create_task(self._drain()) return await future diff --git a/inkbox_codex/tools.py b/inkbox_codex/tools.py index 57814ee..f264a96 100644 --- a/inkbox_codex/tools.py +++ b/inkbox_codex/tools.py @@ -16,6 +16,8 @@ import secrets import sys import time +import uuid +from contextvars import ContextVar from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, List, Tuple @@ -27,15 +29,39 @@ from media import file_to_email_attachment try: - from .config import INKBOX_WS_PATH, call_contexts_dir, channel_hints_path + from .config import ( + INKBOX_WS_PATH, + a2a_turn_context_path, + call_contexts_dir, + channel_hints_path, + ) + from .a2a_delegations import ( + find_by_task, + promote_after_send, + record_before_send, + ) except ImportError: # pragma: no cover - direct local import/test fallback - from config import INKBOX_WS_PATH, call_contexts_dir, channel_hints_path + from config import ( + INKBOX_WS_PATH, + a2a_turn_context_path, + call_contexts_dir, + channel_hints_path, + ) + from a2a_delegations import ( + find_by_task, + promote_after_send, + record_before_send, + ) JsonSchema = Dict[str, Any] SMS_MAX_LENGTH = 1600 IMESSAGE_MAX_LENGTH = 18995 +A2A_TURN_CONTEXT: ContextVar[Dict[str, Any] | None] = ContextVar( + "inkbox_codex_a2a_turn", + default=None, +) @dataclass(frozen=True) @@ -295,6 +321,21 @@ def _str_list(desc: str = "") -> JsonSchema: "message_id": _str("Stable idempotency id."), }, ["card_url", "task_id", "text"]), ), + ToolSpec( + "inkbox_a2a_complete", + "Complete the active inbound A2A task with a final answer.", + _schema({"text": _str("Final answer.")}, ["text"]), + ), + ToolSpec( + "inkbox_a2a_ask_caller", + "Ask the caller for more input on the active inbound A2A task.", + _schema({"text": _str("Question for the caller.")}, ["text"]), + ), + ToolSpec( + "inkbox_a2a_fail", + "Fail the active inbound A2A task with a reason.", + _schema({"reason": _str("Failure reason.")}, ["reason"]), + ), ] @@ -726,23 +767,102 @@ def _run() -> Any: return {"deleted": str(args["contact_id"])} if name in {"inkbox_a2a_call", "inkbox_a2a_check", "inkbox_a2a_reply"}: - a2a = _identity().a2a_client() + identity = _identity() + a2a = identity.a2a_client() try: target = a2a.fetch_card(str(args["card_url"])) if name == "inkbox_a2a_check": if args.get("wait"): return a2a.wait(target, str(args["task_id"])) return a2a.get_task(target, str(args["task_id"])) - return a2a.send( + task_id = str(args.get("task_id") or "") or None + existing = find_by_task(task_id) if task_id else None + message_id = str(args.get("message_id") or uuid.uuid4()) + pending_key = record_before_send( + identity_id=str(identity.id), + rpc_url=str( + getattr(target, "rpc_url", None) + or target["rpc_url"] + ), + card_url=str(args["card_url"]), + message_id=message_id, + context_id=( + args.get("context_id") + or (existing or {}).get("context_id") + or None + ), + task_id=task_id, + session_key=( + os.environ.get("INKBOX_CODEX_CHAT_ID") + or (existing or {}).get("session_key") + or None + ), + ) + result = a2a.send( target, text=str(args["text"]), context_id=args.get("context_id") or None, - task_id=args.get("task_id") or None, - message_id=args.get("message_id") or None, + task_id=task_id, + message_id=message_id, ) + task = getattr(result, "task", None) + if task is None and isinstance(result, dict): + task = result.get("task") + result_task_id = getattr(task, "id", None) + context_id = getattr(task, "context_id", None) + if isinstance(task, dict): + result_task_id = task.get("id") + context_id = task.get("context_id") or task.get("contextId") + if result_task_id and context_id: + promote_after_send( + pending_key, + context_id=str(context_id), + task_id=str(result_task_id), + ) + return result finally: a2a.close() + if name in { + "inkbox_a2a_complete", + "inkbox_a2a_ask_caller", + "inkbox_a2a_fail", + }: + context = A2A_TURN_CONTEXT.get() + context_path = None + if context is None: + chat_id = (os.environ.get("INKBOX_CODEX_CHAT_ID") or "").strip() + if chat_id: + context_path = a2a_turn_context_path(chat_id) + try: + loaded = json.loads(context_path.read_text()) + context = loaded if isinstance(loaded, dict) else None + except (FileNotFoundError, json.JSONDecodeError): + context = None + if context is None: + raise RuntimeError( + "This tool is only available during an inbound A2A task" + ) + intent = { + "inkbox_a2a_complete": "complete", + "inkbox_a2a_ask_caller": "ask_caller", + "inkbox_a2a_fail": "fail", + }[name] + text = str(args["reason"] if name == "inkbox_a2a_fail" else args["text"]) + result = _identity().a2a_reply( + context["task_id"], + intent=intent, + text=text, + ) + context["reply_intent_committed"] = True + if context_path is not None: + tmp = context_path.with_suffix(".tmp") + tmp.write_text(json.dumps(context, sort_keys=True) + "\n") + tmp.chmod(0o600) + os.replace(tmp, context_path) + context_path.chmod(0o600) + return result + raise ValueError(f"unknown Inkbox tool: {name}") try: diff --git a/tests/test_a2a_gateway.py b/tests/test_a2a_gateway.py new file mode 100644 index 0000000..4ca6845 --- /dev/null +++ b/tests/test_a2a_gateway.py @@ -0,0 +1,172 @@ +import asyncio +import json +import types + +import pytest + +from inkbox_codex import gateway as gateway_mod +from inkbox_codex.gateway import InkboxGateway + + +@pytest.fixture(autouse=True) +def fake_web(monkeypatch): + monkeypatch.setattr( + gateway_mod, + "web", + types.SimpleNamespace( + json_response=lambda payload: types.SimpleNamespace( + status=200, + text=json.dumps(payload), + ) + ), + ) + + +class _Session: + def __init__(self): + self.calls = [] + self.inbound = [] + + async def run_consult(self, prompt, *, a2a_context=None): + self.calls.append((prompt, a2a_context)) + return "Completed." + + async def handle_inbound(self, prompt, mode, meta): + self.inbound.append((prompt, mode, meta)) + + +class _Sessions: + def __init__(self): + self.session = _Session() + self.keys = [] + + def get(self, key): + self.keys.append(key) + return self.session + + +def _gateway(tmp_path): + gateway = object.__new__(InkboxGateway) + gateway._a2a_registry_path = tmp_path / "a2a.json" + gateway._a2a_jobs = {} + gateway._identity = types.SimpleNamespace( + id="identity-1", + a2a_task=lambda _task_id: types.SimpleNamespace(state="submitted"), + a2a_reply=lambda task_id, **kwargs: gateway.replies.append( + (task_id, kwargs) + ), + ) + gateway.replies = [] + gateway.sessions = _Sessions() + return gateway + + +def _event(): + return { + "id": "evt-1", + "event_type": "a2a.task.created", + "data": { + "task_id": "task-1", + "context_id": "context-1", + "message_id": "message-1", + "caller": { + "identity_id": "caller-1", + "organization_id": "org-1", + "handle": "caller", + }, + "parts": [{"text": "Investigate."}], + }, + } + + +def test_a2a_gateway_persists_dedupes_and_completes(tmp_path, monkeypatch): + async def inline(function, *args, **kwargs): + return function(*args, **kwargs) + + monkeypatch.setattr(gateway_mod.asyncio, "to_thread", inline) + gateway = _gateway(tmp_path) + + async def scenario(): + first = await gateway._on_a2a_event(_event()) + await asyncio.gather(*gateway._a2a_jobs["task-1"]) + second = await gateway._on_a2a_event(_event()) + return first, second + + first, second = asyncio.run(scenario()) + registry = json.loads(gateway._a2a_registry_path.read_text()) + + assert first.status == 200 + assert json.loads(second.text)["deduped"] is True + assert registry["task-1:message-1"]["state"] == "finalized" + assert gateway.sessions.keys[0] == "a2a:identity-1:context-1" + assert gateway.replies == [ + ("task-1", {"intent": "complete", "text": "Completed."}) + ] + + +def test_a2a_gateway_resumes_nonfinal_registry_entries(tmp_path, monkeypatch): + async def inline(function, *args, **kwargs): + return function(*args, **kwargs) + + monkeypatch.setattr(gateway_mod.asyncio, "to_thread", inline) + gateway = _gateway(tmp_path) + task = types.SimpleNamespace( + id="task-1", + context_id="context-1", + state="working", + caller=types.SimpleNamespace( + identity_id="caller-1", + organization_id="org-1", + handle="caller", + ), + messages=[ + types.SimpleNamespace( + message_id="message-1", + parts=[{"text": "Resume this."}], + ) + ], + ) + gateway._identity.a2a_task = lambda _task_id: task + gateway._identity.iter_a2a_tasks = lambda **_kwargs: iter(()) + gateway._write_a2a_registry( + "task-1:message-1", + _event()["data"], + "running", + ) + + async def scenario(): + await gateway._catch_up_a2a_tasks() + await asyncio.gather(*gateway._a2a_jobs["task-1"]) + + asyncio.run(scenario()) + registry = json.loads(gateway._a2a_registry_path.read_text()) + + assert registry["task-1:message-1"]["state"] == "finalized" + assert gateway.sessions.session.calls[0][0].endswith("Resume this.") + + +def test_a2a_sent_update_returns_to_the_delegating_session( + tmp_path, + monkeypatch, +): + gateway = _gateway(tmp_path) + monkeypatch.setattr( + gateway_mod, + "find_a2a_delegation", + lambda _task_id: { + "session_key": "contact-1", + "card_url": "https://target.example/card", + }, + ) + event = _event() + event["event_type"] = "a2a.sent_task.updated" + event["data"]["state"] = "input_required" + event["data"]["parts"] = [{"text": "Which region?"}] + + asyncio.run(gateway._on_a2a_event(event)) + + assert gateway.sessions.keys[0] == "contact-1" + prompt, mode, meta = gateway.sessions.session.inbound[0] + assert "Which region?" in prompt + assert mode == "external" + assert meta["a2a_task_id"] == "task-1" diff --git a/tests/test_tools.py b/tests/test_tools.py index 6131736..fd29fa2 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -11,7 +11,8 @@ @pytest.fixture(autouse=True) -def _run_to_thread_inline(monkeypatch): +def _run_to_thread_inline(monkeypatch, tmp_path): + monkeypatch.setenv("INKBOX_CODEX_HOME", str(tmp_path)) async def immediate(func, /, *args, **kwargs): return func(*args, **kwargs) @@ -40,6 +41,7 @@ class _FakeTranscript: class _FakeIdentity: def __init__(self): + self.id = "identity-1" self.phone_number = type( "Phone", (), @@ -52,6 +54,7 @@ def __init__(self): self.sent_texts = [] self.sent_imessages = [] self.a2a = _FakeA2AClient() + self.a2a_replies = [] def place_call(self, **kwargs): self.place_call_kwargs = kwargs @@ -79,6 +82,10 @@ def send_text(self, **kwargs): def a2a_client(self): return self.a2a + def a2a_reply(self, task_id, **kwargs): + self.a2a_replies.append((task_id, kwargs)) + return {"id": task_id, "state": kwargs["intent"]} + class _FakeA2AClient: def __init__(self): @@ -91,7 +98,10 @@ def fetch_card(self, card_url): def send(self, target, **kwargs): self.calls.append(("send", target, kwargs)) - return {"kind": "task", "task": {"id": "task-1"}} + return { + "kind": "task", + "task": {"id": "task-1", "context_id": "context-1"}, + } def get_task(self, target, task_id): self.calls.append(("get_task", target, task_id)) @@ -163,6 +173,9 @@ def test_coding_agent_tool_tier_is_registered(): "inkbox_a2a_call", "inkbox_a2a_check", "inkbox_a2a_reply", + "inkbox_a2a_complete", + "inkbox_a2a_ask_caller", + "inkbox_a2a_fail", } assert names == expected @@ -215,6 +228,61 @@ def test_a2a_tools_send_check_and_reply(): assert client.identity.a2a.closed is True +def test_a2a_intent_tools_require_trusted_turn_context(): + client = _FakeClient() + + outside = _call(client, "inkbox_a2a_complete", {"text": "Done."}) + token = tools_mod.A2A_TURN_CONTEXT.set( + { + "task_id": "task-1", + "message_id": "message-1", + "context_id": "context-1", + "reply_intent_committed": False, + } + ) + try: + inside = _call(client, "inkbox_a2a_ask_caller", {"text": "Which region?"}) + context = tools_mod.A2A_TURN_CONTEXT.get() + finally: + tools_mod.A2A_TURN_CONTEXT.reset(token) + + assert "only available during an inbound A2A task" in outside["error"] + assert inside["state"] == "ask_caller" + assert context["reply_intent_committed"] is True + assert client.identity.a2a_replies == [ + ("task-1", {"intent": "ask_caller", "text": "Which region?"}) + ] + + +def test_a2a_intent_tools_share_context_with_the_mcp_process( + tmp_path, + monkeypatch, +): + monkeypatch.setenv("INKBOX_CODEX_HOME", str(tmp_path)) + monkeypatch.setenv("INKBOX_CODEX_CHAT_ID", "a2a:identity-1:context-1") + context_path = tools_mod.a2a_turn_context_path( + "a2a:identity-1:context-1" + ) + context_path.write_text( + json.dumps( + { + "task_id": "task-1", + "message_id": "message-1", + "context_id": "context-1", + "reply_intent_committed": False, + } + ) + ) + client = _FakeClient() + + result = _call(client, "inkbox_a2a_complete", {"text": "Done."}) + + assert result["state"] == "complete" + assert json.loads(context_path.read_text())[ + "reply_intent_committed" + ] is True + + def test_place_call_writes_context_and_tags_websocket_url(tmp_path, monkeypatch): monkeypatch.setenv("INKBOX_CODEX_HOME", str(tmp_path)) client = _FakeClient() From 77209d407fd2e6457bb581a3fddc2584e16e547c Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 05:17:23 +0000 Subject: [PATCH 3/8] fix: tolerate staged A2A rollout --- inkbox_codex/gateway.py | 20 +++++++++- tests/test_gateway_incoming_call_config.py | 46 +++++++++++++++++++--- 2 files changed, 60 insertions(+), 6 deletions(-) diff --git a/inkbox_codex/gateway.py b/inkbox_codex/gateway.py index fc5e4dc..2418a9b 100644 --- a/inkbox_codex/gateway.py +++ b/inkbox_codex/gateway.py @@ -348,6 +348,14 @@ def _voice_consult_prompt( MAIL_EVENTS = ["message.received", "message.bounced", "message.failed"] +def _is_unsupported_a2a_event_types(exc: Exception) -> bool: + detail = str(getattr(exc, "detail", exc)) + return ( + getattr(exc, "status_code", None) == 422 + and any(event_type in detail for event_type in A2A_EVENTS) + ) + + def _outbound_failure_keys( mode: str, conversation_id: Any, @@ -674,7 +682,17 @@ def _reconcile(owner_kw: Dict[str, Any], event_types: List[str]) -> None: identity_events = list(A2A_EVENTS) if getattr(identity, "imessage_enabled", False): identity_events = [*IMESSAGE_EVENTS, *identity_events] - _reconcile({"agent_identity_id": identity.id}, identity_events) + try: + _reconcile({"agent_identity_id": identity.id}, identity_events) + except Exception as exc: + if not _is_unsupported_a2a_event_types(exc): + raise + logger.warning( + "[bridge] Inkbox API does not support A2A webhook events yet; " + "continuing without A2A delivery until the backend is upgraded" + ) + if getattr(identity, "imessage_enabled", False): + _reconcile({"agent_identity_id": identity.id}, IMESSAGE_EVENTS) logger.info("[bridge] identity events for %s → %s", self.cfg.identity, webhook_url) async def _cleanup(self) -> None: diff --git a/tests/test_gateway_incoming_call_config.py b/tests/test_gateway_incoming_call_config.py index 1167176..7cda256 100644 --- a/tests/test_gateway_incoming_call_config.py +++ b/tests/test_gateway_incoming_call_config.py @@ -10,16 +10,30 @@ class _FakeSubscriptions: + def __init__(self): + self.created = [] + def list(self, **_kwargs): return [] - def create(self, **_kwargs): + def create(self, **kwargs): + self.created.append(kwargs) return None def delete(self, _sub_id): return None +class _UnsupportedA2ASubscriptions(_FakeSubscriptions): + def create(self, **kwargs): + if any(event.startswith("a2a.") for event in kwargs["event_types"]): + error = RuntimeError("unsupported A2A events") + error.status_code = 422 + error.detail = "a2a.task.created is not a valid event type" + raise error + return super().create(**kwargs) + + class _FakePhoneNumbers: def __init__(self): self.updates = [] @@ -29,9 +43,11 @@ def update(self, phone_id, **kwargs): class _FakeInkbox: - def __init__(self, identity): + def __init__(self, identity, subscriptions=None): self._identity = identity - self.webhooks = types.SimpleNamespace(subscriptions=_FakeSubscriptions()) + self.webhooks = types.SimpleNamespace( + subscriptions=subscriptions or _FakeSubscriptions() + ) self.phone_numbers = _FakePhoneNumbers() def get_identity(self, _handle): @@ -68,9 +84,9 @@ def _legacy_identity(**kwargs): return legacy -def _patched_gateway(identity): +def _patched_gateway(identity, subscriptions=None): gw = InkboxGateway(BridgeConfig(identity="codex-agent", allow_all_users=True)) - gw._inkbox = _FakeInkbox(identity) + gw._inkbox = _FakeInkbox(identity, subscriptions) gw._public_url = "https://agent.inkboxwire.com" gw._public_host = "agent.inkboxwire.com" gw._patch_identity_objects() @@ -126,3 +142,23 @@ def test_legacy_sdk_without_number_cannot_configure_and_skips(): gw = _patched_gateway(identity) assert gw._inkbox.phone_numbers.updates == [] + + +def test_a2a_subscription_falls_back_to_imessage_on_older_api(): + subscriptions = _UnsupportedA2ASubscriptions() + _patched_gateway( + _Identity(phone=False, imessage=True), + subscriptions=subscriptions, + ) + + assert subscriptions.created[-1]["event_types"] == gateway_mod.IMESSAGE_EVENTS + + +def test_a2a_only_subscription_is_skipped_on_older_api(): + subscriptions = _UnsupportedA2ASubscriptions() + _patched_gateway( + _Identity(phone=False, imessage=False), + subscriptions=subscriptions, + ) + + assert subscriptions.created == [] From 8ee7795dc8d8438a7a2a4a8e71c1deaaf1dd3372 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 05:31:18 +0000 Subject: [PATCH 4/8] fix: handle SDK A2A preflight validation --- inkbox_codex/gateway.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/inkbox_codex/gateway.py b/inkbox_codex/gateway.py index 2418a9b..824623d 100644 --- a/inkbox_codex/gateway.py +++ b/inkbox_codex/gateway.py @@ -351,8 +351,11 @@ def _voice_consult_prompt( def _is_unsupported_a2a_event_types(exc: Exception) -> bool: detail = str(getattr(exc, "detail", exc)) return ( - getattr(exc, "status_code", None) == 422 - and any(event_type in detail for event_type in A2A_EVENTS) + any(event_type in detail for event_type in A2A_EVENTS) + and ( + getattr(exc, "status_code", None) == 422 + or "does not belong to any known channel" in detail + ) ) From 27af569c87363a0f9a99a98a483d31388f21f16a Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 21:23:50 +0000 Subject: [PATCH 5/8] Add filterable A2A history tools --- .github/workflows/tests.yml | 2 +- README.md | 5 +- inkbox_codex/doctor.py | 2 +- inkbox_codex/gateway.py | 2 +- inkbox_codex/setup_wizard.py | 4 +- inkbox_codex/tools.py | 93 ++++++++++++++++++++++++++++++++++++ pyproject.toml | 2 +- tests/test_setup_wizard.py | 8 ++-- tests/test_tools.py | 80 +++++++++++++++++++++++++++++++ 9 files changed, 185 insertions(+), 13 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index b37b0ba..128e68a 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -62,7 +62,7 @@ jobs: - name: Install bridge + test deps run: | uv pip install --system \ - "inkbox @ git+https://github.com/inkbox-ai/inkbox.git@199bbd27c8dab2f70e379de52ca9cc910b0e141d#subdirectory=sdk/python" \ + "inkbox @ git+https://github.com/inkbox-ai/inkbox.git@fdea6e55ac117f246624852aedeef0feabe08b9a#subdirectory=sdk/python" \ -e . pytest # @alpha is the prerelease channel cut from codex main near-daily — the diff --git a/README.md b/README.md index 300cd64..b675cb2 100644 --- a/README.md +++ b/README.md @@ -262,11 +262,10 @@ The agent reaches you (or third parties) through an in-process MCP server: - `inkbox_lookup_contact` · `inkbox_list_contacts` · `inkbox_get_contact` — resolve and read address-book contacts (reverse-lookup by email/phone, free-text search, or full record by id). - `inkbox_create_contact` · `inkbox_update_contact` · `inkbox_delete_contact` — save, edit, and remove organization-wide contacts. Changes affect the shared address book. vCard export/import is not exposed. - `inkbox_a2a_call` · `inkbox_a2a_check` · `inkbox_a2a_reply` — delegate work to another agent and follow its task. +- `inkbox_list_a2a_tasks` · `inkbox_list_a2a_messages` — page and search this identity's inbound and outbound A2A history, with participant, task, context, role, state, and timestamp filters. - `inkbox_a2a_complete` · `inkbox_a2a_ask_caller` · `inkbox_a2a_fail` — commit the outcome of a verified inbound A2A task. These tools are rejected outside that task's isolated session. -Inbound A2A serving and A2A client tools require Inkbox SDK 0.5.5 or newer. -Other bridge features continue to support the base SDK range in -`pyproject.toml`. +The bridge requires Inkbox SDK 0.5.6 or newer. On a live call, the OpenAI Realtime voice agent additionally gets `consult_agent`, `register_post_call_action` / `edit_post_call_action` / `delete_post_call_action`, and `hang_up_call` — see [Voice](#voice). diff --git a/inkbox_codex/doctor.py b/inkbox_codex/doctor.py index 8768934..45a88b8 100644 --- a/inkbox_codex/doctor.py +++ b/inkbox_codex/doctor.py @@ -33,7 +33,7 @@ def run_doctor() -> List[Tuple[str, bool, str]]: import inkbox # noqa: F401 checks.append(("inkbox SDK", True, "installed")) except ImportError: - checks.append(("inkbox SDK", False, "pip install 'inkbox>=0.5.1,<1.0.0'")) + checks.append(("inkbox SDK", False, "pip install 'inkbox>=0.5.6,<1.0.0'")) try: import aiohttp # noqa: F401 diff --git a/inkbox_codex/gateway.py b/inkbox_codex/gateway.py index 824623d..7f9fd0f 100644 --- a/inkbox_codex/gateway.py +++ b/inkbox_codex/gateway.py @@ -533,7 +533,7 @@ async def run(self) -> None: if not AIOHTTP_AVAILABLE: raise RuntimeError("aiohttp is not installed; run: pip install aiohttp") if not INKBOX_AVAILABLE: - raise RuntimeError("inkbox SDK is not installed; run: pip install 'inkbox>=0.5.1,<1.0.0'") + raise RuntimeError("inkbox SDK is not installed; run: pip install 'inkbox>=0.5.6,<1.0.0'") if not self.cfg.api_key or not self.cfg.identity: raise RuntimeError("INKBOX_API_KEY and INKBOX_IDENTITY must be set (see README)") diff --git a/inkbox_codex/setup_wizard.py b/inkbox_codex/setup_wizard.py index 87e7121..80ac897 100644 --- a/inkbox_codex/setup_wizard.py +++ b/inkbox_codex/setup_wizard.py @@ -35,8 +35,8 @@ # Packages the wizard itself needs to talk to Inkbox during setup. The # gateway's Codex CLI dependency is checked by doctor. -INKBOX_REQUIREMENTS = ("inkbox>=0.5.1,<1.0.0", "aiohttp>=3.9") -MIN_INKBOX_VERSION = (0, 5, 0) +INKBOX_REQUIREMENTS = ("inkbox>=0.5.6,<1.0.0", "aiohttp>=3.9") +MIN_INKBOX_VERSION = (0, 5, 6) _BRACKETED_PASTE_PATTERN = re.compile(r"\x1b\[\s*200~|\x1b\[\s*201~") # Bundled avatar attached to the agent's Inkbox contact card during setup. diff --git a/inkbox_codex/tools.py b/inkbox_codex/tools.py index f264a96..ec55728 100644 --- a/inkbox_codex/tools.py +++ b/inkbox_codex/tools.py @@ -96,6 +96,20 @@ def _int(desc: str = "") -> JsonSchema: return schema +def _enum(values: List[str], desc: str = "") -> JsonSchema: + schema = _str(desc) + schema["enum"] = values + return schema + + +def _bounded_int(minimum: int, maximum: int, desc: str = "") -> JsonSchema: + schema = _int(desc) + schema["minimum"] = minimum + schema["maximum"] = maximum + schema["default"] = 50 + return schema + + def _str_list(desc: str = "") -> JsonSchema: schema: JsonSchema = {"type": "array", "items": {"type": "string"}} if desc: @@ -321,6 +335,59 @@ def _str_list(desc: str = "") -> JsonSchema: "message_id": _str("Stable idempotency id."), }, ["card_url", "task_id", "text"]), ), + ToolSpec( + "inkbox_list_a2a_tasks", + "List this identity's A2A task history. Direction defaults to inbound; " + "use the optional participant, state, context, keyword, timestamp, and " + "cursor filters to narrow or continue the result.", + _schema({ + "direction": _enum( + ["inbound", "outbound", "both"], + "Optional history direction.", + ), + "requester_handle": _str("Optional requester identity handle."), + "worker_handle": _str("Optional worker identity handle."), + "state": _enum( + [ + "submitted", + "working", + "input_required", + "auth_required", + "completed", + "failed", + "canceled", + "rejected", + ], + "Optional task lifecycle state.", + ), + "context_id": _str("Optional A2A context id."), + "query": _str("Optional keyword search across task messages."), + "since": _str("Optional ISO 8601 lower timestamp bound."), + "cursor": _str("Opaque next_cursor from the previous page."), + "limit": _bounded_int(1, 100, "Page size from 1 to 100."), + }), + ), + ToolSpec( + "inkbox_list_a2a_messages", + "List messages from this identity's inbound and outbound A2A history. " + "Use the optional participant, task, context, role, keyword, timestamp, " + "and cursor filters to narrow or continue the result.", + _schema({ + "direction": _enum( + ["inbound", "outbound", "both"], + "Optional history direction.", + ), + "requester_handle": _str("Optional requester identity handle."), + "worker_handle": _str("Optional worker identity handle."), + "task_id": _str("Optional A2A task id."), + "context_id": _str("Optional A2A context id."), + "role": _enum(["caller", "agent"], "Optional A2A message role."), + "query": _str("Optional keyword search across message text."), + "since": _str("Optional ISO 8601 lower timestamp bound."), + "cursor": _str("Opaque next_cursor from the previous page."), + "limit": _bounded_int(1, 100, "Page size from 1 to 100."), + }), + ), ToolSpec( "inkbox_a2a_complete", "Complete the active inbound A2A task with a final answer.", @@ -766,6 +833,32 @@ def _run() -> Any: client.contacts.delete(str(args["contact_id"])) return {"deleted": str(args["contact_id"])} + if name in {"inkbox_list_a2a_tasks", "inkbox_list_a2a_messages"}: + limit = int(args.get("limit") or 50) + if not 1 <= limit <= 100: + raise ValueError("limit must be between 1 and 100") + options: Dict[str, Any] = { + "direction": str(args.get("direction") or "").strip() or None, + "requester_handle": ( + str(args.get("requester_handle") or "").strip() or None + ), + "worker_handle": ( + str(args.get("worker_handle") or "").strip() or None + ), + "context_id": str(args.get("context_id") or "").strip() or None, + "q": str(args.get("query") or "").strip() or None, + "since": str(args.get("since") or "").strip() or None, + "cursor": str(args.get("cursor") or "").strip() or None, + "limit": limit, + } + identity = _identity() + if name == "inkbox_list_a2a_tasks": + options["state"] = str(args.get("state") or "").strip() or None + return identity.a2a_tasks(**options) + options["task_id"] = str(args.get("task_id") or "").strip() or None + options["role"] = str(args.get("role") or "").strip() or None + return identity.a2a_messages(**options) + if name in {"inkbox_a2a_call", "inkbox_a2a_check", "inkbox_a2a_reply"}: identity = _identity() a2a = identity.a2a_client() diff --git a/pyproject.toml b/pyproject.toml index 6a25292..44d7172 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ description = "Inkbox bridge for Codex — talk to your coding agent over email, requires-python = ">=3.11" dependencies = [ "aiohttp>=3.9", - "inkbox>=0.5.1,<1.0.0", + "inkbox>=0.5.6,<1.0.0", "segno>=1.5", # terminal QR codes for the iMessage connect step ] diff --git a/tests/test_setup_wizard.py b/tests/test_setup_wizard.py index 1ed37a1..09df6b1 100644 --- a/tests/test_setup_wizard.py +++ b/tests/test_setup_wizard.py @@ -88,7 +88,7 @@ def test_install_command_prefers_uv_when_available(monkeypatch): "install", "--python", "/tmp/venv/bin/python", - "inkbox>=0.5.1,<1.0.0", + "inkbox>=0.5.6,<1.0.0", "aiohttp>=3.9", ]] @@ -98,10 +98,10 @@ def test_install_command_falls_back_to_pip_and_ensurepip(monkeypatch): monkeypatch.setattr(setup_wizard.shutil, "which", lambda _name: None) assert setup_wizard._install_commands() == [ - [["/tmp/venv/bin/python", "-m", "pip", "install", "inkbox>=0.5.1,<1.0.0", "aiohttp>=3.9"]], + [["/tmp/venv/bin/python", "-m", "pip", "install", "inkbox>=0.5.6,<1.0.0", "aiohttp>=3.9"]], [ ["/tmp/venv/bin/python", "-m", "ensurepip", "--upgrade"], - ["/tmp/venv/bin/python", "-m", "pip", "install", "inkbox>=0.5.1,<1.0.0", "aiohttp>=3.9"], + ["/tmp/venv/bin/python", "-m", "pip", "install", "inkbox>=0.5.6,<1.0.0", "aiohttp>=3.9"], ], ] @@ -120,7 +120,7 @@ def fail_import(): out = capsys.readouterr().out assert "/tmp/venv/bin/python" in out assert "uv pip install --python" in out - assert "inkbox>=0.5.1,<1.0.0" in out + assert "inkbox>=0.5.6,<1.0.0" in out # ---------------------------------------------------------------------- diff --git a/tests/test_tools.py b/tests/test_tools.py index fd29fa2..492e68e 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -55,6 +55,7 @@ def __init__(self): self.sent_imessages = [] self.a2a = _FakeA2AClient() self.a2a_replies = [] + self.a2a_history_calls = [] def place_call(self, **kwargs): self.place_call_kwargs = kwargs @@ -86,6 +87,14 @@ def a2a_reply(self, task_id, **kwargs): self.a2a_replies.append((task_id, kwargs)) return {"id": task_id, "state": kwargs["intent"]} + def a2a_tasks(self, **kwargs): + self.a2a_history_calls.append(("tasks", kwargs)) + return {"items": [{"id": "task-1"}], "next_cursor": "task-next"} + + def a2a_messages(self, **kwargs): + self.a2a_history_calls.append(("messages", kwargs)) + return {"items": [{"id": "message-1"}], "next_cursor": "message-next"} + class _FakeA2AClient: def __init__(self): @@ -173,6 +182,8 @@ def test_coding_agent_tool_tier_is_registered(): "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", @@ -228,6 +239,75 @@ def test_a2a_tools_send_check_and_reply(): assert client.identity.a2a.closed is True +def test_a2a_history_tools_forward_filters_and_pagination(): + client = _FakeClient() + tasks = _call( + client, + "inkbox_list_a2a_tasks", + { + "direction": "both", + "requester_handle": "requester", + "worker_handle": "worker", + "state": "completed", + "context_id": "context-1", + "query": "summary", + "since": "2026-07-01T00:00:00Z", + "cursor": "task-cursor", + "limit": 3, + }, + ) + messages = _call( + client, + "inkbox_list_a2a_messages", + { + "direction": "outbound", + "requester_handle": "requester", + "worker_handle": "worker", + "task_id": "task-1", + "context_id": "context-1", + "role": "agent", + "query": "done", + "since": "2026-07-01T00:00:00Z", + "cursor": "message-cursor", + "limit": 4, + }, + ) + + assert tasks["next_cursor"] == "task-next" + assert messages["next_cursor"] == "message-next" + assert client.identity.a2a_history_calls[0] == ( + "tasks", + { + "direction": "both", + "requester_handle": "requester", + "worker_handle": "worker", + "context_id": "context-1", + "q": "summary", + "since": "2026-07-01T00:00:00Z", + "cursor": "task-cursor", + "limit": 3, + "state": "completed", + }, + ) + assert client.identity.a2a_history_calls[1] == ( + "messages", + { + "direction": "outbound", + "requester_handle": "requester", + "worker_handle": "worker", + "context_id": "context-1", + "q": "done", + "since": "2026-07-01T00:00:00Z", + "cursor": "message-cursor", + "limit": 4, + "task_id": "task-1", + "role": "agent", + }, + ) + invalid = _call(client, "inkbox_list_a2a_messages", {"limit": 101}) + assert invalid["error"] == "limit must be between 1 and 100" + + def test_a2a_intent_tools_require_trusted_turn_context(): client = _FakeClient() From 73654194fc5344d0b9bd6a8b2b11e6775cc18a13 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 21:38:39 +0000 Subject: [PATCH 6/8] Fix unpublished SDK installs in live CI --- .github/workflows/canary.yml | 5 ++++- .github/workflows/live-channels.yml | 5 ++++- .github/workflows/live-external-events.yml | 5 ++++- .github/workflows/live-voice.yml | 5 ++++- 4 files changed, 16 insertions(+), 4 deletions(-) diff --git a/.github/workflows/canary.yml b/.github/workflows/canary.yml index 6c7bdca..e161ca4 100644 --- a/.github/workflows/canary.yml +++ b/.github/workflows/canary.yml @@ -32,7 +32,10 @@ jobs: node-version: "22" - name: Install bridge + test deps - run: pip install -e . pytest + run: | + pip install \ + "inkbox @ git+https://github.com/inkbox-ai/inkbox.git@fdea6e55ac117f246624852aedeef0feabe08b9a#subdirectory=sdk/python" \ + -e . pytest # @alpha is the prerelease channel cut from codex main near-daily — the # freshest main build available without compiling the host from source. diff --git a/.github/workflows/live-channels.yml b/.github/workflows/live-channels.yml index cdf9f07..3b66432 100644 --- a/.github/workflows/live-channels.yml +++ b/.github/workflows/live-channels.yml @@ -69,7 +69,10 @@ jobs: mkdir -p "$RUNNER_TEMP/codex-home" "$RUNNER_TEMP/project" - name: Install bridge + test deps - run: pip install -e . pytest + run: | + pip install \ + "inkbox @ git+https://github.com/inkbox-ai/inkbox.git@fdea6e55ac117f246624852aedeef0feabe08b9a#subdirectory=sdk/python" \ + -e . pytest # @alpha is the prerelease channel cut from codex main near-daily — the # freshest main build available without compiling the host from source. diff --git a/.github/workflows/live-external-events.yml b/.github/workflows/live-external-events.yml index 1ef86de..6f4fda9 100644 --- a/.github/workflows/live-external-events.yml +++ b/.github/workflows/live-external-events.yml @@ -59,7 +59,10 @@ jobs: mkdir -p "$RUNNER_TEMP/codex-home" "$RUNNER_TEMP/project" - name: Install bridge + test deps - run: pip install -e . pytest + run: | + pip install \ + "inkbox @ git+https://github.com/inkbox-ai/inkbox.git@fdea6e55ac117f246624852aedeef0feabe08b9a#subdirectory=sdk/python" \ + -e . pytest # @alpha is the prerelease channel cut from codex main near-daily — the # freshest main build available without compiling the host from source. diff --git a/.github/workflows/live-voice.yml b/.github/workflows/live-voice.yml index 2bbfd95..26e8124 100644 --- a/.github/workflows/live-voice.yml +++ b/.github/workflows/live-voice.yml @@ -78,7 +78,10 @@ jobs: # uvicorn[standard] matters: the bare install can't accept WebSocket upgrades, # and the driver's call-media endpoint is a WebSocket. - name: Install bridge + test deps - run: pip install -e . pytest fastapi 'uvicorn[standard]' + run: | + pip install \ + "inkbox @ git+https://github.com/inkbox-ai/inkbox.git@fdea6e55ac117f246624852aedeef0feabe08b9a#subdirectory=sdk/python" \ + -e . pytest fastapi 'uvicorn[standard]' # @alpha is the prerelease channel cut from codex main near-daily — the # freshest main build available without compiling the host from source. From 70018fe96758afd6823cf5a7c59023f000cc3bff Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 21:48:48 +0000 Subject: [PATCH 7/8] Split A2A webhook subscriptions by channel --- inkbox_codex/gateway.py | 41 ++++++++++++++++------ tests/test_gateway_incoming_call_config.py | 17 +++++++++ 2 files changed, 47 insertions(+), 11 deletions(-) diff --git a/inkbox_codex/gateway.py b/inkbox_codex/gateway.py index 7f9fd0f..fed4745 100644 --- a/inkbox_codex/gateway.py +++ b/inkbox_codex/gateway.py @@ -634,16 +634,34 @@ def _patch_identity_objects(self) -> None: ws_url = f"wss://{self._public_host}{INKBOX_WS_PATH}" identity = self._inkbox.get_identity(self.cfg.identity) - def _reconcile(owner_kw: Dict[str, Any], event_types: List[str]) -> None: + def _reconcile( + owner_kw: Dict[str, Any], + event_types: List[str], + *, + subscription_url: str = webhook_url, + ) -> None: existing = self._inkbox.webhooks.subscriptions.list(**owner_kw) + desired_families = { + event_type.split(".", 1)[0] for event_type in event_types + } for sub in existing: - if sub.url == webhook_url and set(sub.event_types) == set(event_types): + if ( + sub.url == subscription_url + and set(sub.event_types) == set(event_types) + ): return # already wired - if sub.url.endswith(DEFAULT_WEBHOOK_PATH): - # A previous bridge install — replace it. + existing_families = { + event_type.split(".", 1)[0] for event_type in sub.event_types + } + if ( + sub.url.split("?", 1)[0].endswith(DEFAULT_WEBHOOK_PATH) + and desired_families & existing_families + ): + # Replace only this event channel. One identity may have + # separate iMessage and A2A subscriptions at the same URL. self._inkbox.webhooks.subscriptions.delete(sub.id) self._inkbox.webhooks.subscriptions.create( - url=webhook_url, event_types=event_types, **owner_kw + url=subscription_url, event_types=event_types, **owner_kw ) if identity.mailbox is not None: @@ -682,11 +700,12 @@ def _reconcile(owner_kw: Dict[str, Any], event_types: List[str]) -> None: self.cfg.identity, webhook_url, ws_url, ) - identity_events = list(A2A_EVENTS) - if getattr(identity, "imessage_enabled", False): - identity_events = [*IMESSAGE_EVENTS, *identity_events] try: - _reconcile({"agent_identity_id": identity.id}, identity_events) + _reconcile( + {"agent_identity_id": identity.id}, + A2A_EVENTS, + subscription_url=f"{webhook_url}?channel=a2a", + ) except Exception as exc: if not _is_unsupported_a2a_event_types(exc): raise @@ -694,8 +713,8 @@ def _reconcile(owner_kw: Dict[str, Any], event_types: List[str]) -> None: "[bridge] Inkbox API does not support A2A webhook events yet; " "continuing without A2A delivery until the backend is upgraded" ) - if getattr(identity, "imessage_enabled", False): - _reconcile({"agent_identity_id": identity.id}, IMESSAGE_EVENTS) + if getattr(identity, "imessage_enabled", False): + _reconcile({"agent_identity_id": identity.id}, IMESSAGE_EVENTS) logger.info("[bridge] identity events for %s → %s", self.cfg.identity, webhook_url) async def _cleanup(self) -> None: diff --git a/tests/test_gateway_incoming_call_config.py b/tests/test_gateway_incoming_call_config.py index 7cda256..9cbd9bd 100644 --- a/tests/test_gateway_incoming_call_config.py +++ b/tests/test_gateway_incoming_call_config.py @@ -154,6 +154,23 @@ def test_a2a_subscription_falls_back_to_imessage_on_older_api(): assert subscriptions.created[-1]["event_types"] == gateway_mod.IMESSAGE_EVENTS +def test_a2a_and_imessage_use_channel_coherent_subscriptions(): + subscriptions = _FakeSubscriptions() + _patched_gateway( + _Identity(phone=False, imessage=True), + subscriptions=subscriptions, + ) + + assert [created["event_types"] for created in subscriptions.created] == [ + gateway_mod.A2A_EVENTS, + gateway_mod.IMESSAGE_EVENTS, + ] + assert [created["url"] for created in subscriptions.created] == [ + "https://agent.inkboxwire.com/webhook?channel=a2a", + "https://agent.inkboxwire.com/webhook", + ] + + def test_a2a_only_subscription_is_skipped_on_older_api(): subscriptions = _UnsupportedA2ASubscriptions() _patched_gateway( From 018aaa722acecc03f31e93f8aaebb71f1d9417ae Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Fri, 24 Jul 2026 21:57:47 +0000 Subject: [PATCH 8/8] Test A2A subscription channel isolation --- tests/test_gateway_incoming_call_config.py | 26 ++++++++++++++++++---- 1 file changed, 22 insertions(+), 4 deletions(-) diff --git a/tests/test_gateway_incoming_call_config.py b/tests/test_gateway_incoming_call_config.py index 9cbd9bd..9ff385f 100644 --- a/tests/test_gateway_incoming_call_config.py +++ b/tests/test_gateway_incoming_call_config.py @@ -10,18 +10,20 @@ class _FakeSubscriptions: - def __init__(self): + def __init__(self, existing=()): self.created = [] + self.existing = list(existing) + self.deleted = [] def list(self, **_kwargs): - return [] + return list(self.existing) def create(self, **kwargs): self.created.append(kwargs) return None - def delete(self, _sub_id): - return None + def delete(self, sub_id): + self.deleted.append(sub_id) class _UnsupportedA2ASubscriptions(_FakeSubscriptions): @@ -171,6 +173,22 @@ def test_a2a_and_imessage_use_channel_coherent_subscriptions(): ] +def test_imessage_reconcile_preserves_existing_a2a_channel_subscription(): + a2a = types.SimpleNamespace( + id="sub-a2a", + url="https://agent.inkboxwire.com/webhook?channel=a2a", + event_types=gateway_mod.A2A_EVENTS, + ) + subscriptions = _FakeSubscriptions([a2a]) + _patched_gateway( + _Identity(phone=False, imessage=True), + subscriptions=subscriptions, + ) + + assert subscriptions.deleted == [] + assert subscriptions.created[-1]["event_types"] == gateway_mod.IMESSAGE_EVENTS + + def test_a2a_only_subscription_is_skipped_on_older_api(): subscriptions = _UnsupportedA2ASubscriptions() _patched_gateway(