diff --git a/.github/workflows/canary.yml b/.github/workflows/canary.yml index 9f1a338..020eb35 100644 --- a/.github/workflows/canary.yml +++ b/.github/workflows/canary.yml @@ -27,7 +27,9 @@ jobs: - name: Install bridge + latest SDK run: | - pip install -e . pytest + pip install \ + "inkbox @ git+https://github.com/inkbox-ai/inkbox.git@fdea6e55ac117f246624852aedeef0feabe08b9a#subdirectory=sdk/python" \ + -e . pytest pip install -U claude-agent-sdk - name: Install latest Claude Code CLI diff --git a/.github/workflows/live-channels.yml b/.github/workflows/live-channels.yml index f9bffe3..e95cd3b 100644 --- a/.github/workflows/live-channels.yml +++ b/.github/workflows/live-channels.yml @@ -65,7 +65,10 @@ jobs: node-version: 22 - name: Install bridge - run: pip install -e . pytest + run: | + pip install \ + "inkbox @ git+https://github.com/inkbox-ai/inkbox.git@fdea6e55ac117f246624852aedeef0feabe08b9a#subdirectory=sdk/python" \ + -e . pytest - name: Install Claude Code CLI run: npm install -g @anthropic-ai/claude-code diff --git a/.github/workflows/live-external-events.yml b/.github/workflows/live-external-events.yml index 7358e3f..361e5aa 100644 --- a/.github/workflows/live-external-events.yml +++ b/.github/workflows/live-external-events.yml @@ -63,7 +63,10 @@ jobs: node-version: 22 - name: Install bridge - run: pip install -e . pytest + run: | + pip install \ + "inkbox @ git+https://github.com/inkbox-ai/inkbox.git@fdea6e55ac117f246624852aedeef0feabe08b9a#subdirectory=sdk/python" \ + -e . pytest - name: Install Claude Code CLI run: npm install -g @anthropic-ai/claude-code diff --git a/.github/workflows/live-voice.yml b/.github/workflows/live-voice.yml index 31e2b24..aa4063f 100644 --- a/.github/workflows/live-voice.yml +++ b/.github/workflows/live-voice.yml @@ -56,7 +56,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 + driver 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]' - name: Install Claude Code CLI run: npm install -g @anthropic-ai/claude-code diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index c71dff0..578a254 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -55,7 +55,7 @@ jobs: - name: Install bridge + latest SDK 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 uv pip install --system -U claude-agent-sdk diff --git a/README.md b/README.md index 5003344..2ba317c 100644 --- a/README.md +++ b/README.md @@ -238,6 +238,11 @@ 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_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. + +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_claude/a2a_delegations.py b/inkbox_claude/a2a_delegations.py new file mode 100644 index 0000000..09bd972 --- /dev/null +++ b/inkbox_claude/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_CLAUDE_HOME") + or (Path.home() / ".inkbox-claude") + ) + 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_claude/doctor.py b/inkbox_claude/doctor.py index ad1a31e..477747a 100644 --- a/inkbox_claude/doctor.py +++ b/inkbox_claude/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 claude_agent_sdk # noqa: F401 diff --git a/inkbox_claude/gateway.py b/inkbox_claude/gateway.py index e957ff4..c5656db 100644 --- a/inkbox_claude/gateway.py +++ b/inkbox_claude/gateway.py @@ -64,6 +64,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 ( @@ -76,6 +77,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 ( @@ -388,6 +390,24 @@ def _call_ended_prompt(transcript: Any) -> str: "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"} + + +def _is_unsupported_a2a_event_types(exc: Exception) -> bool: + detail = str(getattr(exc, "detail", exc)) + return ( + 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 + ) + ) def _message_too_long_reason(channel: str, content: str, max_chars: int) -> str: @@ -474,6 +494,10 @@ def __init__(self, cfg: BridgeConfig): # delivery-failure feedback loop stops 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-claude" / "a2a_tasks.json" + ) + self._a2a_jobs: Dict[str, set[asyncio.Task[Any]]] = {} # ------------------------------------------------------------------ # Lifecycle @@ -488,7 +512,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)") @@ -528,6 +552,7 @@ async def run(self) -> None: on_send_rejected=self._note_send_rejection, health_fn=self.health_report, ) + await self._catch_up_a2a_tasks() logger.info( "[bridge] ready — %s / %s / %s → Claude Code in %s", @@ -589,16 +614,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: @@ -636,9 +679,22 @@ def _reconcile(owner_kw: Dict[str, Any], event_types: List[str]) -> None: "[bridge] incoming-call action for %s → %s + %s", self.cfg.identity, webhook_url, ws_url, ) + try: + _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 + 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] iMessage for %s → %s", self.cfg.identity, webhook_url) + logger.info("[bridge] identity events for %s → %s", self.cfg.identity, webhook_url) async def _cleanup(self) -> None: if self.sessions is not None: @@ -787,6 +843,8 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response": response = await self._on_imessage_received(envelope) elif event_type == "imessage.reaction_received": response = await self._on_imessage_reaction_received(envelope) + elif event_type.startswith("a2a."): + response = 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. elif event_type == "text.delivery_failed": @@ -856,7 +914,9 @@ def _is_known_inkbox_event(cls, event_type: "str | None", envelope: Dict[str, An 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 # ``id`` by itself is not a call discriminator: generic external # webhook schemas commonly use a top-level event id. Treat an @@ -2152,6 +2212,227 @@ async def _on_external_event( asyncio.create_task(self._run_external_turn(chat_id, prompt, directive)) return web.json_response({"ok": True}) + 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 + session = self.sessions.get( + f"a2a:{self._identity.id}:{context_id}", + system_prompt_extra=( + "You are handling an inbound A2A task. Use " + "inkbox_a2a_complete, inkbox_a2a_ask_caller, or " + "inkbox_a2a_fail when an explicit outcome is appropriate." + ), + ) + reply = await session.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") + async def _run_external_turn(self, chat_id: str, prompt: str, directive: str) -> None: try: # The directive rides on the session's system prompt (per-event diff --git a/inkbox_claude/sessions.py b/inkbox_claude/sessions.py index 0208e20..93e37ab 100644 --- a/inkbox_claude/sessions.py +++ b/inkbox_claude/sessions.py @@ -88,6 +88,7 @@ class _Turn: text: str future: Optional["asyncio.Future[str]"] = None + 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 Claude as a turn. @@ -647,6 +648,13 @@ async def _run_turn(self, turn: _Turn) -> None: self._current_turn = turn typing_task: Optional[asyncio.Task] = None retried_missing_resume = False + a2a_token = None + if turn.a2a_context is not None: + try: + from .tools import A2A_TURN_CONTEXT + except ImportError: # pragma: no cover + from tools import A2A_TURN_CONTEXT + a2a_token = A2A_TURN_CONTEXT.set(turn.a2a_context) try: while True: try: @@ -697,6 +705,8 @@ async def _run_turn(self, turn: _Turn) -> None: return raise finally: + if a2a_token is not None: + A2A_TURN_CONTEXT.reset(a2a_token) self._turn_active = False self._current_turn = None if typing_task is not None: @@ -766,7 +776,12 @@ async def _deliver_reply(self, turn: _Turn, reply: str) -> None: self.chat_id, self.mode, self.reply_meta, reply, exc ) - 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 Claude Code turn and RETURN its text (don't send it). Used by the Realtime voice bridge, post-call actions, and delivery- @@ -786,7 +801,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_claude/setup_wizard.py b/inkbox_claude/setup_wizard.py index 8b81395..0ef1918 100644 --- a/inkbox_claude/setup_wizard.py +++ b/inkbox_claude/setup_wizard.py @@ -35,8 +35,8 @@ # Packages the wizard itself needs to talk to Inkbox during setup. The # gateway's other dependency (claude-agent-sdk) is checked by doctor. -INKBOX_MIN_VERSION = (0, 5, 0) -INKBOX_REQUIREMENTS = ("inkbox>=0.5.1,<1.0.0", "aiohttp>=3.9") +INKBOX_MIN_VERSION = (0, 5, 6) +INKBOX_REQUIREMENTS = ("inkbox>=0.5.6,<1.0.0", "aiohttp>=3.9") _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_claude/tools.py b/inkbox_claude/tools.py index a9ff2cc..7b633f3 100644 --- a/inkbox_claude/tools.py +++ b/inkbox_claude/tools.py @@ -14,6 +14,7 @@ import mimetypes import secrets import time +import uuid from contextvars import ContextVar from pathlib import Path from typing import Any, Dict, List, Optional, Tuple @@ -26,8 +27,18 @@ try: from .config import INKBOX_WS_PATH, call_contexts_dir + 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 + from a2a_delegations import ( + find_by_task, + promote_after_send, + record_before_send, + ) try: from claude_agent_sdk import create_sdk_mcp_server, tool @@ -48,6 +59,10 @@ # is long-lived and its ``mode`` mutates per inbound message, giving tools a # live view of the conversation's channel. CURRENT_SESSION: ContextVar[Any] = ContextVar("inkbox_claude_current_session", default=None) +A2A_TURN_CONTEXT: ContextVar[Optional[Dict[str, Any]]] = ContextVar( + "inkbox_claude_a2a_turn", + default=None, +) def _mark_tool_delivery(mode: str, target: str) -> None: @@ -729,6 +744,263 @@ def _run(): except Exception as exc: return _error(str(exc)) + @tool( + "inkbox_a2a_call", + "Send a task to an A2A 1.0 Agent Card. Keep the returned task and " + "context ids for later checks or replies.", + {"card_url": str, "text": str, "context_id": str, "task_id": str, + "message_id": str}, + ) + async def inkbox_a2a_call(args: Dict[str, Any]) -> Dict[str, Any]: + def _run(): + identity = _identity() + a2a = identity.a2a_client() + try: + target = a2a.fetch_card(str(args["card_url"])) + message_id = str(args.get("message_id") or uuid.uuid4()) + session = CURRENT_SESSION.get() + 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 None, + task_id=args.get("task_id") or None, + session_key=getattr(session, "chat_id", 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=message_id, + ) + task = getattr(result, "task", None) + if task is None and isinstance(result, dict): + task = result.get("task") + task_id = getattr(task, "id", None) + context_id = getattr(task, "context_id", None) + if isinstance(task, dict): + task_id = task.get("id") + context_id = task.get("context_id") or task.get("contextId") + if task_id and context_id: + promote_after_send( + pending_key, + context_id=str(context_id), + task_id=str(task_id), + ) + return result + finally: + a2a.close() + + try: + return _result(await asyncio.to_thread(_run)) + except Exception as exc: + return _error(str(exc)) + + @tool( + "inkbox_a2a_check", + "Fetch an A2A task, or wait until it reaches a final or input-required state.", + {"card_url": str, "task_id": str, "wait": bool}, + ) + async def inkbox_a2a_check(args: Dict[str, Any]) -> Dict[str, Any]: + def _run(): + a2a = _identity().a2a_client() + try: + target = a2a.fetch_card(str(args["card_url"])) + if args.get("wait"): + return a2a.wait(target, str(args["task_id"])) + return a2a.get_task(target, str(args["task_id"])) + finally: + a2a.close() + + try: + return _result(await asyncio.to_thread(_run)) + except Exception as exc: + return _error(str(exc)) + + @tool( + "inkbox_a2a_reply", + "Reply to a remote A2A task that requested more input.", + {"card_url": str, "task_id": str, "text": str, "message_id": str}, + ) + async def inkbox_a2a_reply(args: Dict[str, Any]) -> Dict[str, Any]: + def _run(): + identity = _identity() + a2a = identity.a2a_client() + try: + target = a2a.fetch_card(str(args["card_url"])) + task_id = str(args["task_id"]) + existing = find_by_task(task_id) or {} + message_id = str(args.get("message_id") or uuid.uuid4()) + session = CURRENT_SESSION.get() + 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=existing.get("context_id"), + task_id=task_id, + session_key=( + getattr(session, "chat_id", None) + or existing.get("session_key") + ), + ) + result = a2a.send( + target, + task_id=task_id, + text=str(args["text"]), + message_id=message_id, + ) + task = getattr(result, "task", None) + if task is None and isinstance(result, dict): + task = result.get("task") + context_id = getattr(task, "context_id", None) + if isinstance(task, dict): + context_id = task.get("context_id") or task.get("contextId") + if context_id: + promote_after_send( + pending_key, + context_id=str(context_id), + task_id=task_id, + ) + return result + finally: + a2a.close() + + try: + return _result(await asyncio.to_thread(_run)) + except Exception as exc: + return _error(str(exc)) + + def _a2a_history_options(args: Dict[str, Any], *, messages: bool) -> Dict[str, Any]: + 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, + } + if messages: + options["task_id"] = str(args.get("task_id") or "").strip() or None + options["role"] = str(args.get("role") or "").strip() or None + else: + options["state"] = str(args.get("state") or "").strip() or None + return options + + @tool( + "inkbox_list_a2a_tasks", + "List this identity's A2A task history with optional direction, " + "participant, state, context, keyword, timestamp, and cursor filters. " + "Direction defaults to inbound.", + { + "direction": str, "requester_handle": str, "worker_handle": str, + "state": str, "context_id": str, "query": str, "since": str, + "cursor": str, "limit": int, + }, + ) + async def inkbox_list_a2a_tasks(args: Dict[str, Any]) -> Dict[str, Any]: + try: + return _result( + await asyncio.to_thread( + _identity().a2a_tasks, + **_a2a_history_options(args, messages=False), + ) + ) + except Exception as exc: + return _error(str(exc)) + + @tool( + "inkbox_list_a2a_messages", + "List messages from this identity's inbound and outbound A2A history " + "with optional participant, task, context, role, keyword, timestamp, " + "and cursor filters.", + { + "direction": str, "requester_handle": str, "worker_handle": str, + "task_id": str, "context_id": str, "role": str, "query": str, + "since": str, "cursor": str, "limit": int, + }, + ) + async def inkbox_list_a2a_messages(args: Dict[str, Any]) -> Dict[str, Any]: + try: + return _result( + await asyncio.to_thread( + _identity().a2a_messages, + **_a2a_history_options(args, messages=True), + ) + ) + except Exception as exc: + return _error(str(exc)) + + def _a2a_intent(intent: str, text: str) -> Any: + context = A2A_TURN_CONTEXT.get() + if context is None: + raise RuntimeError("This tool is only available during an inbound A2A task") + result = _identity().a2a_reply( + context["task_id"], + intent=intent, + text=text, + ) + context["reply_intent_committed"] = True + return result + + @tool( + "inkbox_a2a_complete", + "Complete the active inbound A2A task with a final answer.", + {"text": str}, + ) + async def inkbox_a2a_complete(args: Dict[str, Any]) -> Dict[str, Any]: + try: + return _result( + await asyncio.to_thread( + _a2a_intent, "complete", str(args["text"]) + ) + ) + except Exception as exc: + return _error(str(exc)) + + @tool( + "inkbox_a2a_ask_caller", + "Ask the caller for more input on the active inbound A2A task.", + {"text": str}, + ) + async def inkbox_a2a_ask_caller(args: Dict[str, Any]) -> Dict[str, Any]: + try: + return _result( + await asyncio.to_thread( + _a2a_intent, "ask_caller", str(args["text"]) + ) + ) + except Exception as exc: + return _error(str(exc)) + + @tool( + "inkbox_a2a_fail", + "Fail the active inbound A2A task with a reason.", + {"reason": str}, + ) + async def inkbox_a2a_fail(args: Dict[str, Any]) -> Dict[str, Any]: + try: + return _result( + await asyncio.to_thread( + _a2a_intent, "fail", str(args["reason"]) + ) + ) + except Exception as exc: + return _error(str(exc)) + tools = [ inkbox_whoami, inkbox_send_email, @@ -747,6 +1019,14 @@ def _run(): inkbox_create_contact, inkbox_update_contact, inkbox_delete_contact, + 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, ] server = create_sdk_mcp_server(name="inkbox", version="0.1.0", tools=tools) tool_names = [ @@ -767,5 +1047,13 @@ def _run(): "mcp__inkbox__inkbox_create_contact", "mcp__inkbox__inkbox_update_contact", "mcp__inkbox__inkbox_delete_contact", + "mcp__inkbox__inkbox_a2a_call", + "mcp__inkbox__inkbox_a2a_check", + "mcp__inkbox__inkbox_a2a_reply", + "mcp__inkbox__inkbox_list_a2a_tasks", + "mcp__inkbox__inkbox_list_a2a_messages", + "mcp__inkbox__inkbox_a2a_complete", + "mcp__inkbox__inkbox_a2a_ask_caller", + "mcp__inkbox__inkbox_a2a_fail", ] return server, tool_names diff --git a/pyproject.toml b/pyproject.toml index d894fcc..e0c8c98 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,11 @@ [project] name = "claude-code-plugin" -version = "0.1.3" +version = "0.1.4" description = "Inkbox bridge for Claude Code — talk to your coding agent over email, SMS, iMessage, and voice" requires-python = ">=3.11" dependencies = [ "aiohttp>=3.9", - "inkbox>=0.5.1,<1.0.0", + "inkbox>=0.5.6,<1.0.0", "claude-agent-sdk>=0.1.0", "segno>=1.5", # terminal QR codes for the iMessage connect step ] diff --git a/tests/test_a2a_gateway.py b/tests/test_a2a_gateway.py new file mode 100644 index 0000000..9e0544b --- /dev/null +++ b/tests/test_a2a_gateway.py @@ -0,0 +1,172 @@ +import asyncio +import json +import types + +import pytest + +from inkbox_claude import gateway as gateway_mod +from inkbox_claude.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, system_prompt_extra=""): + self.keys.append((key, system_prompt_extra)) + 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][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][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_gateway_call_config.py b/tests/test_gateway_call_config.py index 0be226f..1cdc909 100644 --- a/tests/test_gateway_call_config.py +++ b/tests/test_gateway_call_config.py @@ -12,15 +12,34 @@ class _FakeSubscriptions: - def __init__(self): + def __init__(self, existing=()): self.created = [] + self.existing = list(existing) + self.deleted = [] def list(self, **_owner): - return [] + return list(self.existing) def create(self, **kwargs): self.created.append(kwargs) + def delete(self, sub_id): + self.deleted.append(sub_id) + + +class _UnsupportedA2ASubscriptions(_FakeSubscriptions): + def __init__(self): + super().__init__() + self.attempted = [] + + def create(self, **kwargs): + self.attempted.append(kwargs) + if any(event.startswith("a2a.") for event in kwargs["event_types"]): + raise ValueError( + "event_type 'a2a.task.created' does not belong to any known channel" + ) + return super().create(**kwargs) + class _FakePhoneNumbers: def __init__(self): @@ -31,9 +50,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): @@ -70,9 +91,9 @@ def _phone(): return types.SimpleNamespace(id="phone-1", number="+15550001111") -def _patched_gateway(identity): +def _patched_gateway(identity, subscriptions=None): gw = gateway.InkboxGateway(BridgeConfig(identity="claude", require_signature=False)) - gw._inkbox = _FakeInkbox(identity) + gw._inkbox = _FakeInkbox(identity, subscriptions) gw._public_url = "https://agent.example" gw._public_host = "agent.example" gw._patch_identity_objects() @@ -132,3 +153,56 @@ def test_legacy_sdk_cannot_configure_imessage_only_identity(): gw = _patched_gateway(identity) assert gw._inkbox.phone_numbers.updated == [] + + +def test_a2a_subscription_falls_back_to_imessage_on_older_api(): + subscriptions = _UnsupportedA2ASubscriptions() + _patched_gateway( + _Identity(phone=None, imessage_enabled=True), + subscriptions=subscriptions, + ) + + assert subscriptions.created[-1]["event_types"] == gateway.IMESSAGE_EVENTS + + +def test_a2a_and_imessage_use_channel_coherent_subscriptions(): + subscriptions = _FakeSubscriptions() + _patched_gateway( + _Identity(phone=None, imessage_enabled=True), + subscriptions=subscriptions, + ) + + assert [created["event_types"] for created in subscriptions.created] == [ + gateway.A2A_EVENTS, + gateway.IMESSAGE_EVENTS, + ] + assert [created["url"] for created in subscriptions.created] == [ + "https://agent.example/webhook?channel=a2a", + "https://agent.example/webhook", + ] + + +def test_imessage_reconcile_preserves_existing_a2a_channel_subscription(): + a2a = types.SimpleNamespace( + id="sub-a2a", + url="https://agent.example/webhook?channel=a2a", + event_types=gateway.A2A_EVENTS, + ) + subscriptions = _FakeSubscriptions([a2a]) + _patched_gateway( + _Identity(phone=None, imessage_enabled=True), + subscriptions=subscriptions, + ) + + assert subscriptions.deleted == [] + assert subscriptions.created[-1]["event_types"] == gateway.IMESSAGE_EVENTS + + +def test_a2a_only_subscription_is_skipped_on_older_api(): + subscriptions = _UnsupportedA2ASubscriptions() + _patched_gateway( + _Identity(phone=None, imessage_enabled=False), + subscriptions=subscriptions, + ) + + assert subscriptions.created == [] diff --git a/tests/test_setup_wizard.py b/tests/test_setup_wizard.py index fb21333..39e3aa4 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 36f0524..5851bb3 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -11,7 +11,8 @@ @pytest.fixture(autouse=True) -def _fake_claude_sdk(monkeypatch): +def _fake_claude_sdk(monkeypatch, tmp_path): + monkeypatch.setenv("INKBOX_CLAUDE_HOME", str(tmp_path)) async def immediate(func, /, *args, **kwargs): return func(*args, **kwargs) @@ -53,6 +54,7 @@ class _FakeTranscript: class _FakeIdentity: def __init__(self): + self.id = "identity-1" self.agent_handle = "claude-agent" self.mailbox = type("Mailbox", (), {"email_address": "claude@inkbox.ai"})() self.phone_number = type( @@ -72,6 +74,9 @@ def __init__(self): self.sent_emails = [] self.sent_texts = [] self.sent_imessages = [] + self.a2a = _FakeA2AClient() + self.a2a_replies = [] + self.a2a_history_calls = [] def place_call(self, **kwargs): self.place_call_kwargs = kwargs @@ -102,6 +107,49 @@ def send_text(self, **kwargs): self.sent_texts.append(kwargs) return type("Message", (), {"id": "sms-1"})() + 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"]} + + 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): + 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", "context_id": "context-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): @@ -165,6 +213,14 @@ 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", + "inkbox_list_a2a_tasks", + "inkbox_list_a2a_messages", + "inkbox_a2a_complete", + "inkbox_a2a_ask_caller", + "inkbox_a2a_fail", } assert set(tools) == expected @@ -182,6 +238,140 @@ def test_get_contact_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_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 == [ + ( + "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", + }, + ), + ( + "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_tasks", {"limit": 101}) + assert invalid["error"] == "limit must be between 1 and 100" + + +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_place_call_writes_context_and_tags_websocket_url(tmp_path, monkeypatch): monkeypatch.setenv("INKBOX_CLAUDE_HOME", str(tmp_path)) client = _FakeClient()