diff --git a/taosmd/config.py b/taosmd/config.py index f7abac73..771983ee 100644 --- a/taosmd/config.py +++ b/taosmd/config.py @@ -50,6 +50,10 @@ _GENERATOR_PROFILE_KEY = "generator_profile" # Whether A2A registry auth runs in enforce mode (True) or verify-and-warn mode (False). _A2A_AUTH_ENFORCE_KEY = "a2a_auth_enforce" +# Key under which the optional A2A role-resolver base URL is stored. When set, +# the bus can validate and deliver to taOS role handles (@taOS-*) by consulting +# the taOS resolution endpoint; when unset, role recipients are rejected (400). +_ROLE_RESOLVER_URL_KEY = "a2a_role_resolver_url" # Section under which collections settings live. ``allowed_roots`` is the # safety line of the collections contract: source paths must resolve inside # one of these directories. Empty (the default) means collections are off. @@ -602,6 +606,53 @@ def set_a2a_auth_enforce(value: bool, data_dir=None) -> None: _write(data, data_dir) +# --------------------------------------------------------------------------- +# A2A role resolver URL (opt-in delivery-time role resolution, taOS#2155) +# --------------------------------------------------------------------------- + +def get_a2a_role_resolver_url(data_dir=None) -> str | None: + """Return the configured A2A role-resolver base URL, or ``None`` if unset. + + Resolution order (first non-empty wins): + + 1. ``TAOSMD_A2A_ROLE_RESOLVER_URL`` environment variable + 2. ``a2a_role_resolver_url`` key in ``~/.taosmd/config.json`` + + When set, role recipients (``@taOS-*``) are validated and resolved at send + time and annotated at delivery time by consulting the taOS resolution + endpoint. When unset, role recipients are rejected with 400 (no resolver + configured -> fail loud, never guess). + """ + env = os.environ.get("TAOSMD_A2A_ROLE_RESOLVER_URL") + if env and env.strip(): + return env.strip() + url = _read(data_dir).get(_ROLE_RESOLVER_URL_KEY) + if isinstance(url, str) and url.strip(): + return url.strip() + return None + + +def set_a2a_role_resolver_url(url: str, clear: bool = False, data_dir=None) -> None: + """Persist the A2A role-resolver base URL (or clear it). + + Args: + url: Base URL of the taOS resolution endpoint. Ignored when ``clear`` + is True. + clear: when True, remove the setting (role sends revert to 400). + + Raises: + ValueError: when ``clear`` is False and ``url`` is not a non-empty string. + """ + data = _read(data_dir) + if clear: + data.pop(_ROLE_RESOLVER_URL_KEY, None) + else: + if not isinstance(url, str) or not url.strip(): + raise ValueError("url must be a non-empty string (or pass clear=True)") + data[_ROLE_RESOLVER_URL_KEY] = url.strip() + _write(data, data_dir) + + # --------------------------------------------------------------------------- # Collections: allowed roots # --------------------------------------------------------------------------- @@ -689,6 +740,8 @@ def set_collections_allowed_roots(roots, clear: bool = False, data_dir=None) -> "set_serve_dashboard", "get_a2a_auth_enforce", "set_a2a_auth_enforce", + "get_a2a_role_resolver_url", + "set_a2a_role_resolver_url", "MANAGED_BY_STANDALONE", "MANAGED_BY_TAOS", "get_generator_profile", diff --git a/taosmd/http_server.py b/taosmd/http_server.py index 59acb838..0a0d8086 100644 --- a/taosmd/http_server.py +++ b/taosmd/http_server.py @@ -101,11 +101,17 @@ ``GET /shelves?project=`` -> ``{"shelves": [...]}`` ``GET /pending?agent=`` -> ``{"pending": [...]}`` ``POST /pending/resolve`` ``{"id", "decision", "note"?}`` -> resolve result -``POST /a2a/send`` ``{"from", "body", "thread"?, "reply_to"?, "refs"?, "blocks"?}`` -> send receipt +``POST /a2a/send`` ``{"from", "body", "thread"?, "reply_to"?, "refs"?, "blocks"?, "recipient"?}`` -> send receipt + ``recipient``: optional addressee handle (agent ``@handle`` or role + ``@taOS-``, e.g. ``@taOS-PA``); stored verbatim. A role recipient + is validated at send time against the configured resolver + (``a2a_role_resolver_url``): 400 if not resolvable to a single holder, + 503 if unreachable, 400 if no resolver configured. The resolved + identity is never stored -- binding is at delivery (taOS#2155). ``refs``: optional list (<=8) of ``{"kind": doc|report|spec|log, "title", "uri", "sha256"?, "doc_id"?, "version"?, "for"?, "summary"?}`` ``blocks``: optional list of arbitrary objects (no schema validation); when present, ``body`` must be non-empty -``GET /a2a/messages`` ``?thread=&since=&limit=&fields=&format=`` -> ``{"messages": [...]}`` (``fields=id,sender,body`` projects keys; ``format=ndjson`` emits one message per line) -``GET /a2a/stream`` ``?thread=&since=`` -> SSE stream (text/event-stream) +``GET /a2a/messages`` ``?thread=&since=&limit=&recipient=&fields=&format=`` -> ``{"messages": [...]}`` (``fields=id,sender,body`` projects keys; ``format=ndjson`` emits one message per line; ``recipient=`` filters on the verbatim stored handle and, for a role ``@taOS-*``, annotates ``resolved_to`` with the current holder computed at read time) +``GET /a2a/stream`` ``?thread=&since=`` -> SSE stream (text/event-stream; messages carry the verbatim ``recipient`` field when present, no ``resolved_to`` -- resolution is per-reader) ``GET /a2a/channels`` -> ``{"channels": [...]}`` ``GET /a2a/members`` ``?channel=`` -> ``{"members": [...]}`` ``POST /tasks`` ``{"title", "body"?, "project"?, "assignee"?, "priority"?, "depends_on"?: [...], "created_by"}`` -> task object @@ -174,6 +180,7 @@ from urllib.parse import parse_qs, urlsplit from . import __version__, capabilities, config as _config, service +from .role_resolver import RoleResolver, RoleResolveError, is_role_handle # --------------------------------------------------------------------------- # Static webui helpers @@ -578,7 +585,7 @@ def close(self) -> None: def _make_handler(data_dir, runner: _ServiceLoop, verifier=None, - grants_verifier=None): + grants_verifier=None, role_resolver=None): """Build a handler class bound to a fixed ``data_dir``. ThreadingHTTPServer instantiates the handler per request, so the data dir @@ -590,6 +597,13 @@ def _make_handler(data_dir, runner: _ServiceLoop, verifier=None, otherwise. ``GET /health``, ``GET /version``, ``GET /``, ``GET /ui``, and static assets are always open so monitoring probes, capability/drift probes, and the inspection UI keep working. + + ``role_resolver`` is an injected seam (taOSmd #2155): an object exposing + ``resolve(role) -> canonical_id | None`` used to validate role recipients + at send time and annotate the current holder at delivery time. When None + it is built from the configured ``a2a_role_resolver_url``; when neither is + set, role recipients are rejected (400/503) as specified, and plain agent + handles pass through untouched. """ # Read the server-side expected token once at handler-class creation time. # This is the token the *server* checks (not the client's outbound token). @@ -623,9 +637,19 @@ def _make_handler(data_dir, runner: _ServiceLoop, verifier=None, expected_iss=registry_auth.REGISTRY_ISS, ) _grants_verifier = registry_auth.grants_verifier_from_url( - _registry_url, - grants_token=_registry_admin_token, - ) + _registry_url, + grants_token=_registry_admin_token, + ) + + # Role resolver (taOSmd #2155). Injected for tests; otherwise built from the + # configured resolver URL. When None, role recipients are rejected (the bus + # owns the message-side concepts, not the binding; a role cannot be + # validated or delivered without a resolver). + _role_resolver = role_resolver + if _role_resolver is None: + _role_resolver_url = _config.get_a2a_role_resolver_url(data_dir) + if _role_resolver_url: + _role_resolver = RoleResolver(_role_resolver_url) # Paths that are always public regardless of the token setting. _PUBLIC_PATHS = frozenset({"/", "/ui", "/health", "/version"}) @@ -1396,6 +1420,7 @@ def _handle_a2a_send(self) -> None: reply_to = body.get("reply_to") refs = body.get("refs") blocks = body.get("blocks") + recipient = body.get("recipient") if not isinstance(from_, str) or not from_: raise _BadRequest("'from' (non-empty string) is required") # --- Envelope field validation (taOSmd #211) --- @@ -1440,6 +1465,33 @@ def _handle_a2a_send(self) -> None: raise _BadRequest( "'body' (non-empty string) is required when 'blocks' is present" ) + # --- Recipient validation (taOSmd #2155) --- + # recipient: optional addressee handle (agent @handle or role + # @taOS-). Stored VERBATIM. A role recipient is validated at + # send time against the configured resolver -- the binding happens at + # delivery, so the resolved identity is never persisted and holder + # rotation needs no sender reconfiguration. + if recipient is not None: + if not isinstance(recipient, str) or not recipient.strip(): + raise _BadRequest("'recipient' must be a non-empty string when provided") + recipient = recipient.strip() + if is_role_handle(recipient): + if _role_resolver is None: + raise _BadRequest( + f"role recipient {recipient!r} requires a configured " + "role resolver (a2a_role_resolver_url)" + ) + try: + holder = _role_resolver.resolve(recipient) + except RoleResolveError as exc: + # Configured but unreachable: 503, never a silent drop. + self._send_json(503, {"error": f"role resolver unavailable: {exc}"}) + return + # A None result means no single holder -> reject, nothing stored. + if holder is None: + raise _BadRequest( + f"role recipient {recipient!r} does not resolve to a single holder" + ) # Registry auth (opt-in): when a verifier is configured, run the # identity + grant checks and collect any failure reason. # In enforce mode (a2a_auth_enforce=true) failures are rejected with @@ -1495,7 +1547,7 @@ def _handle_a2a_send(self) -> None: service.a2a_send( sender=from_, body=body_text, thread=thread, reply_to=reply_to, - refs=refs, blocks=blocks, + refs=refs, blocks=blocks, recipient=recipient, data_dir=data_dir, ) ) @@ -1507,6 +1559,7 @@ def _handle_a2a_messages(self, qs: dict) -> None: limit_raw = (qs.get("limit") or [50])[0] fields_raw = (qs.get("fields") or [None])[0] fmt = (qs.get("format") or ["json"])[0] + recipient = (qs.get("recipient") or [None])[0] try: since = float(since_raw) if since_raw is not None else None except (TypeError, ValueError) as exc: @@ -1517,9 +1570,42 @@ def _handle_a2a_messages(self, qs: dict) -> None: raise _BadRequest("'limit' must be an integer") from exc if fmt not in ("json", "ndjson"): raise _BadRequest("'format' must be 'json' or 'ndjson'") + # --- Delivery-time role resolution (taOSmd #2155) --- + # A role recipient (e.g. ?recipient=@taOS-PA) is matched against the + # verbatim stored handle by service.a2a_feed, then annotated with the + # CURRENT holder (resolved_to) computed here at read time. The stored + # envelope never carries resolved_to -- resolution is per-reader and + # per-read, so rotating the holder changes the answer without a + # re-send. Fail-loud: no resolver -> 400; unreachable -> 503; a + # reachable resolver that finds no holder is not an error (the role + # simply has no current holder), so resolved_to is omitted in that + # case. SSE does NOT annotate: role-addressed messages ride verbatim + # (recipient present, no resolved_to on the broadcast). + holder: str | None = None + annotate = False + if recipient is not None: + if not isinstance(recipient, str) or not recipient.strip(): + raise _BadRequest("'recipient' must be a non-empty string when provided") + recipient = recipient.strip() + if is_role_handle(recipient): + annotate = True + if _role_resolver is None: + self._send_json(400, {"error": f"role recipient {recipient!r} requires a configured role resolver (a2a_role_resolver_url)"}) + return + try: + holder = _role_resolver.resolve(recipient) + except RoleResolveError as exc: + self._send_json(503, {"error": f"role resolver unavailable: {exc}"}) + return messages = runner.run( - service.a2a_feed(thread=thread, since=since, limit=limit_i, data_dir=data_dir) + service.a2a_feed( + thread=thread, since=since, limit=limit_i, + recipient=recipient, data_dir=data_dir, + ) ) + if annotate and holder is not None: + for msg in messages: + msg["resolved_to"] = holder # Compact mode: ?fields=id,sender,body projects each message down # to the named keys so token-frugal consumers (LLM agents) skip # framing they never read. Unknown names are ignored, never a 400, @@ -2062,7 +2148,7 @@ def _handle_admin_a2a_supersede_message(self) -> None: def make_server(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, data_dir=None, - verifier=None, grants_verifier=None): + verifier=None, grants_verifier=None, role_resolver=None): """Create (but do not start) a :class:`ThreadingHTTPServer`. Useful for tests that need to bind an ephemeral port (``port=0``) and read @@ -2072,6 +2158,9 @@ def make_server(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, data_dir=Non A :class:`_ServiceLoop` is started and attached as ``server.service_loop``; closing it is handled by :func:`serve`. Tests that drive ``make_server`` directly should call ``server.service_loop.close()`` during teardown. + + ``role_resolver`` is an optional injectable seam (taOSmd #2155) for tests; + when omitted it is built from config by :func:`_make_handler`. """ if verifier is not None and grants_verifier is None: # Identity without permission is half the locked auth contract @@ -2084,7 +2173,7 @@ def make_server(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, data_dir=Non runner = _ServiceLoop() httpd = ThreadingHTTPServer( (host, port), - _make_handler(data_dir, runner, verifier, grants_verifier), + _make_handler(data_dir, runner, verifier, grants_verifier, role_resolver), ) httpd.service_loop = runner return httpd diff --git a/taosmd/remote.py b/taosmd/remote.py index 6df7e3dd..b476414c 100644 --- a/taosmd/remote.py +++ b/taosmd/remote.py @@ -212,12 +212,13 @@ async def a2a_send( reply_to: str | None = None, refs: list | None = None, blocks: list | None = None, + recipient: str | None = None, **_opts, ) -> dict: """POST /a2a/send: post a message to the remote A2A bus. Returns the send receipt ``{"id", "from", "thread", "reply_to"}`` - plus ``refs``/``blocks`` when supplied (taOSmd #211). + plus ``refs``/``blocks``/``recipient`` when supplied (taOSmd #211/#2155). """ payload: dict = {"from": sender, "body": body, "thread": thread} if reply_to is not None: @@ -226,6 +227,8 @@ async def a2a_send( payload["refs"] = refs if blocks is not None: payload["blocks"] = blocks + if recipient is not None: + payload["recipient"] = recipient return await self._run("POST", "/a2a/send", payload) async def a2a_feed( @@ -234,17 +237,23 @@ async def a2a_feed( thread: str | None = None, since: float | None = None, limit: int = 50, + recipient: str | None = None, **_opts, ) -> list[dict]: """GET /a2a/messages: return messages from the remote A2A bus, oldest-first. - Returns the ``messages`` list from the server response. + ``recipient`` is forwarded as ``?recipient=`` so the remote server can + apply its verbatim recipient filter (taOSmd #2155). The remote does + not annotate ``resolved_to``; callers reading a role recipient do their + own delivery-time resolution. """ params: dict = {"limit": limit} if thread is not None: params["thread"] = thread if since is not None: params["since"] = since + if recipient is not None: + params["recipient"] = recipient resp = await self._run("GET", "/a2a/messages", params=params) return resp.get("messages", []) diff --git a/taosmd/role_resolver.py b/taosmd/role_resolver.py new file mode 100644 index 00000000..0dbba69b --- /dev/null +++ b/taosmd/role_resolver.py @@ -0,0 +1,144 @@ +"""Role-to-holder resolution for A2A recipient addressing (taOS#2155). + +taosmd owns the message-side concepts; taOS owns the role *binding*. A +role handle (``@taOS-``, e.g. ``@taOS-PA``) is stored verbatim on the +message envelope and resolved at delivery time, so rotating the holder never +requires reconfiguring senders. The resolver is an injected seam: this module +provides the :class:`RoleResolver` that talks to the (taOS-provided) +resolution endpoint, but the HTTP server accepts any object exposing a +compatible ``resolve(role) -> canonical_id | None`` method. + +Fail-loud contract +------------------ +* reachable, exactly one holder -> returns the canonical id +* reachable, no single holder -> returns ``None`` (callers map this to a + 400 at send time, or omit ``resolved_to`` at read time) +* configured but unreachable -> raises :class:`RoleResolveError` + (callers map this to 503: never a silent drop, never a guess) + +Resolution is NEVER cached at send time -- the stored envelope keeps only the +role handle, so rotation is reflected on the next read. A short TTL cache +bounds repeated lookups within a single resolver instance. +""" +from __future__ import annotations + +import json +import logging +import threading +import time +import urllib.error +import urllib.request + +logger = logging.getLogger(__name__) + +# A recipient handle is a *role* when it is prefixed with ``@taOS-`` (the taOS +# PA/role namespace), e.g. ``@taOS-PA``. Bare ``@handle`` strings are agent +# handles and need no resolution. This convention is the message-side shim that +# lets the bus recognise a role recipient without consulting the resolver. +ROLE_PREFIX = "@taOS-" + +# Path (relative to the resolver base URL) that resolves a role to its holder. +# ``{role}`` is the bare role name with the leading ``@`` stripped. +_RESOLVE_PATH = "/api/roles/{role}/resolve" + +# Default cache TTL (seconds) for resolved role -> holder mappings. +_DEFAULT_TTL = 30.0 + + +def is_role_handle(recipient: str | None) -> bool: + """Return True when ``recipient`` is a role handle (``@taOS-...``).""" + return isinstance(recipient, str) and recipient.startswith(ROLE_PREFIX) + + +class RoleResolveError(Exception): + """The role resolver is configured but could not be reached. + + Raised only for *transport* failures (connection refused, DNS failure, + non-2xx other than 404). A reachable resolver that finds no holder returns + ``None`` instead -- that is a valid negative answer, not a failure. + """ + + +class RoleResolver: + """Resolve taOS role handles to the canonical id of their current holder. + + Args: + url: Base URL of the taOS resolution endpoint (e.g. the taOS API). + token: Optional bearer token sent as ``Authorization``. + timeout: Per-request timeout in seconds. + ttl: Cache lifetime in seconds; a resolved role is re-fetched once this + elapses so holder rotation propagates without a restart. + """ + + def __init__(self, url: str, *, token: str | None = None, + timeout: int = 10, ttl: float = _DEFAULT_TTL) -> None: + self._base = url.rstrip("/") + self._token = token + self._timeout = timeout + self._ttl = ttl + # role -> (expiry_monotonic, canonical_id | None) + self._cache: dict[str, tuple[float, str | None]] = {} + self._lock = threading.Lock() + + def resolve(self, role: str) -> str | None: + """Return the canonical id of the single holder of ``role``, else None. + + Raises :class:`RoleResolveError` when the endpoint is unreachable. + A negative answer (no holder / not exactly one) is returned as None. + """ + if not is_role_handle(role): + return None + with self._lock: + cached = self._cache.get(role) + if cached is not None and cached[0] > time.monotonic(): + return cached[1] + result = self._fetch(role) + with self._lock: + self._cache[role] = (time.monotonic() + self._ttl, result) + return result + + def _fetch(self, role: str) -> str | None: + bare = role.lstrip("@") + url = self._base + _RESOLVE_PATH.format(role=bare) + headers: dict[str, str] = {"Accept": "application/json"} + if self._token: + headers["Authorization"] = f"Bearer {self._token}" + req = urllib.request.Request(url, headers=headers, method="GET") + try: + with urllib.request.urlopen(req, timeout=self._timeout) as resp: + body = json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + if exc.code == 404: + return None + raise RoleResolveError( + f"role resolver {url} returned HTTP {exc.code}" + ) from exc + except (urllib.error.URLError, TimeoutError, OSError) as exc: + raise RoleResolveError( + f"role resolver {url} unreachable: {exc}" + ) from exc + # The endpoint reports the single current holder; absent/empty means + # the role has no (single) holder right now. + holder = body.get("holder") if isinstance(body, dict) else None + if not isinstance(holder, str) or not holder: + return None + return holder + + def bust(self, role: str | None = None) -> None: + """Drop cached entries, forcing a re-fetch on the next ``resolve()``. + + ``None`` clears the whole cache; otherwise just ``role`` is evicted. + """ + with self._lock: + if role is None: + self._cache.clear() + else: + self._cache.pop(role, None) + + +__all__ = [ + "RoleResolver", + "RoleResolveError", + "is_role_handle", + "ROLE_PREFIX", +] diff --git a/taosmd/service.py b/taosmd/service.py index cf3e9125..0d3e54e5 100644 --- a/taosmd/service.py +++ b/taosmd/service.py @@ -349,6 +349,7 @@ async def a2a_send( reply_to: str | None = None, refs: list | None = None, blocks: list | None = None, + recipient: str | None = None, data_dir=None, ) -> dict: """Post a message onto the agent-to-agent bus. @@ -364,8 +365,17 @@ async def a2a_send( payload and echoed back in the receipt and on feed/SSE reads. When absent they are omitted from output entirely (no null noise). + ``recipient`` is an optional addressee handle (taOSmd #2155): an agent + ``@handle`` or a role ``@taOS-``. It is stored VERBATIM -- the role + binding is resolved at delivery, never at send time, so holder rotation + does not require re-sending. When absent it is omitted from output. + Send-time role validation (400 on an unresolvable role, 503 on an + unreachable resolver) is performed by the HTTP server before this + function is called; the service layer itself only stores/forwards the + handle. + Returns ``{"id", "from", "thread", "reply_to"}`` plus ``refs`` and/or - ``blocks`` when those were supplied. + ``blocks`` and/or ``recipient`` when those were supplied. When a remote server URL is configured the call is forwarded to :class:`~taosmd.remote.RemoteClient` transparently. @@ -374,11 +384,15 @@ async def a2a_send( raise ValueError("sender must be a non-empty string") if not isinstance(body, str) or not body: raise ValueError("body must be a non-empty string") + if recipient is not None and ( + not isinstance(recipient, str) or not recipient + ): + raise ValueError("recipient must be a non-empty string when provided") remote = _get_remote(data_dir) if remote is not None: return await remote.a2a_send( sender, body, thread=thread, reply_to=reply_to, - refs=refs, blocks=blocks, + refs=refs, blocks=blocks, recipient=recipient, ) stores = await _api._ensure_stores(data_dir) archive = stores["archive"] @@ -393,6 +407,8 @@ async def a2a_send( data["refs"] = refs if blocks is not None: data["blocks"] = blocks + if recipient is not None: + data["recipient"] = recipient row_id = await archive.record( event_type=EVENT_A2A, data=data, @@ -405,6 +421,8 @@ async def a2a_send( receipt["refs"] = refs if blocks is not None: receipt["blocks"] = blocks + if recipient is not None: + receipt["recipient"] = recipient return receipt @@ -413,6 +431,7 @@ async def a2a_feed( thread: str | None = None, since: float | None = None, limit: int = 50, + recipient: str | None = None, data_dir=None, ) -> list[dict]: """Return messages from the agent-to-agent bus, oldest-first. @@ -423,16 +442,25 @@ async def a2a_feed( messages when ``since`` is None). Returns chronological order (oldest first) suitable for chat-style display. + ``recipient`` is an optional verbatim filter (taOSmd #2155): when given, + only messages whose stored ``recipient`` field equals it are returned. + The stored value is always the raw handle (agent ``@handle`` or role + ``@taOS-``); the HTTP layer resolves a role recipient and annotates + the computed holder as ``resolved_to`` at read time. The service layer + itself never resolves -- it only filters on the verbatim stored handle. + Each item has shape ``{"id", "ts", "from", "body", "thread", - "reply_to"}`` plus ``refs`` and/or ``blocks`` when those were - supplied on send (taOSmd #211). + "reply_to"}`` plus ``refs`` and/or ``blocks`` and/or ``recipient`` when + those were supplied on send (taOSmd #211 / #2155). When a remote server URL is configured the call is forwarded to :class:`~taosmd.remote.RemoteClient` transparently. """ remote = _get_remote(data_dir) if remote is not None: - return await remote.a2a_feed(thread=thread, since=since, limit=limit) + return await remote.a2a_feed( + thread=thread, since=since, limit=limit, recipient=recipient, + ) stores = await _api._ensure_stores(data_dir) archive = stores["archive"] @@ -495,6 +523,10 @@ async def a2a_feed( # Skip admin-action rows (they have no "from" field) if data.get("admin_action"): continue + # Verbatim recipient filter: matches the stored handle (agent or role) + # exactly. Role resolution/annotation is the HTTP layer's job. + if recipient is not None and data.get("recipient") != recipient: + continue msg = { "id": row_id, "ts": row["timestamp"], @@ -509,6 +541,9 @@ async def a2a_feed( msg["refs"] = data["refs"] if "blocks" in data: msg["blocks"] = data["blocks"] + # Addressee (taOSmd #2155): stored verbatim, omitted when absent. + if "recipient" in data: + msg["recipient"] = data["recipient"] result.append(msg) return result diff --git a/tests/test_a2a.py b/tests/test_a2a.py index a3406647..e6beff8e 100644 --- a/tests/test_a2a.py +++ b/tests/test_a2a.py @@ -23,6 +23,8 @@ from taosmd import api as taosmd_api from taosmd import http_server, service +from taosmd.remote import RemoteClient +from taosmd.role_resolver import RoleResolveError # --------------------------------------------------------------------------- @@ -690,3 +692,386 @@ def test_http_a2a_blocks_without_body_returns_400(live_server): assert status == 400 assert "blocks" in body["error"] _assert_thread_empty(live_server, "v-blocks-no-body") + + +# --------------------------------------------------------------------------- +# Recipient addressing (taOSmd #2155) +# --------------------------------------------------------------------------- +# +# A message may carry an optional ``recipient`` addressee handle (agent +# ``@handle`` or role ``@taOS-``). It is stored VERBATIM and echoed on +# GET/SSE reads; when absent the record is identical to today. Role +# recipients are validated at send time against an injected RoleResolver seam +# and resolved again at delivery time, so holder rotation never requires a +# re-send. + +class _FakeRoleResolver: + """Injectable RoleResolver stand-in: configurable holders, no cache. + + Mirrors the ``resolve(role) -> canonical_id | None`` contract of + :class:`taosmd.role_resolver.RoleResolver` but is fully mutable so tests + can rotate the holder answer and simulate unreachability. + """ + + def __init__(self, holders=None, unreachable=False): + self._holders = dict(holders or {}) + self.unreachable = unreachable + + def resolve(self, role): + if self.unreachable: + raise RoleResolveError("role resolver unreachable (fake)") + return self._holders.get(role) + + def set_holders(self, holders): + self._holders = dict(holders or {}) + + def set_unreachable(self, flag=True): + self.unreachable = flag + + +@pytest.fixture +def resolver_server(tmp_path, monkeypatch): + """HTTP server with an injected fake role resolver; yields (url, resolver).""" + data_dir = tmp_path / "taosmd-a2a-resolver" + data_dir.mkdir() + monkeypatch.setattr(taosmd_api, "_stores_cache", {}) + resolver = _FakeRoleResolver() + httpd = http_server.make_server( + "127.0.0.1", 0, data_dir=str(data_dir), role_resolver=resolver, + ) + stores = httpd.service_loop.run(taosmd_api._ensure_stores(str(data_dir))) + _patch_embedder(stores) + host, port = httpd.server_address[:2] + t = threading.Thread(target=httpd.serve_forever, daemon=True) + t.start() + try: + yield f"http://{host}:{port}", resolver + finally: + httpd.shutdown() + httpd.server_close() + t.join(timeout=5) + for s in list(taosmd_api._stores_cache.values()): + for store in (s.get("archive"), s.get("vector"), s.get("kg")): + if store and hasattr(store, "close"): + try: + httpd.service_loop.run(store.close()) + except Exception: + pass + httpd.service_loop.close() + + +# --- Recipient round-trip (agent handle) -------------------------------------- + +def test_a2a_send_with_recipient_roundtrip(isolated_data_dir): + """send with recipient (agent handle) -> feed returns it verbatim.""" + _setup_stores(isolated_data_dir) + dd = str(isolated_data_dir) + + receipt = asyncio.run(service.a2a_send( + "taOSmd", "hello to @agent1", thread="rcpt", + recipient="@agent1", data_dir=dd, + )) + assert receipt["recipient"] == "@agent1" + + # Feed returns it verbatim. + msgs = asyncio.run(service.a2a_feed(thread="rcpt", data_dir=dd)) + assert len(msgs) == 1 + assert msgs[0]["recipient"] == "@agent1" + + # Verbatim recipient filter isolates the addressed messages. + filtered = asyncio.run(service.a2a_feed(thread="rcpt", recipient="@agent1", data_dir=dd)) + assert len(filtered) == 1 + other = asyncio.run(service.a2a_feed(thread="rcpt", recipient="@other", data_dir=dd)) + assert other == [] + + +def test_a2a_send_without_recipient_omits_key(isolated_data_dir): + """send without recipient -> receipt and feed output omit the key entirely.""" + _setup_stores(isolated_data_dir) + dd = str(isolated_data_dir) + + receipt = asyncio.run(service.a2a_send("taOSmd", "no recipient", thread="nr", data_dir=dd)) + assert "recipient" not in receipt + + msgs = asyncio.run(service.a2a_feed(thread="nr", data_dir=dd)) + assert len(msgs) == 1 + assert "recipient" not in msgs[0] + + +def test_http_a2a_send_recipient_roundtrip(live_server): + """POST /a2a/send with recipient -> GET /a2a/messages returns it verbatim.""" + status, body = _post( + f"{live_server}/a2a/send", + {"from": "taOSmd", "body": "to an agent", "thread": "http-r", + "recipient": "@agent1"}, + ) + assert status == 200, body + assert body["recipient"] == "@agent1" + + status, body = _get(f"{live_server}/a2a/messages?thread=http-r") + assert status == 200, body + msgs = body["messages"] + assert len(msgs) == 1 + assert msgs[0]["recipient"] == "@agent1" + + # Verbatim filter. + status, body = _get(f"{live_server}/a2a/messages?thread=http-r&recipient=@agent1") + assert status == 200 + assert len(body["messages"]) == 1 + + status, body = _get(f"{live_server}/a2a/messages?thread=http-r&recipient=@other") + assert status == 200 + assert body["messages"] == [] + + +def test_http_a2a_sse_carries_recipient(live_server): + """SSE frame includes the verbatim recipient, never resolved_to.""" + parsed = urllib.parse.urlsplit(live_server) + host = parsed.hostname + port = parsed.port + thread = "sse-rcpt" + + frames_received = [] + error_holder = [] + + def _stream_reader(): + try: + frames_received.extend( + _read_sse_frames(host, port, f"/a2a/stream?thread={thread}", timeout=8.0) + ) + except Exception as exc: # noqa: BLE001 + error_holder.append(exc) + + reader = threading.Thread(target=_stream_reader, daemon=True) + reader.start() + time.sleep(1.5) + + status, body = _post( + live_server + "/a2a/send", + {"from": "sse-sender", "body": "sse recipient payload", + "thread": thread, "recipient": "@agent1"}, + ) + assert status == 200, f"send failed: {body}" + + reader.join(timeout=10) + assert not error_holder, f"SSE reader raised: {error_holder[0]}" + assert frames_received + payloads = [json.loads(f) for f in frames_received] + matching = [p for p in payloads if p.get("body") == "sse recipient payload"] + assert matching, "expected the posted message in SSE frames" + msg = matching[0] + assert msg["recipient"] == "@agent1" + assert "resolved_to" not in msg + + +# --- Role send-time validation (taOS#2155) ---------------------------------- + +def test_role_send_resolvable_accepted(resolver_server): + """A resolvable role recipient is accepted and stored verbatim.""" + url, resolver = resolver_server + resolver.set_holders({"@taOS-PA": "holderA"}) + + status, body = _post( + f"{url}/a2a/send", + {"from": "taOSmd", "body": "role addressed", "thread": "r1", + "recipient": "@taOS-PA"}, + ) + assert status == 200, body + # Stored verbatim -- the resolved identity is NEVER in the receipt. + assert body["recipient"] == "@taOS-PA" + assert "resolved_to" not in body + + status, body = _get(f"{url}/a2a/messages?thread=r1") + assert status == 200 + msgs = body["messages"] + assert len(msgs) == 1 + assert msgs[0]["recipient"] == "@taOS-PA" + + +def test_role_send_unresolvable_returns_400(resolver_server): + """An unresolvable role (no holder) -> 400 naming the role, nothing stored.""" + url, resolver = resolver_server + # holders dict does not contain @taOS-NoSuch -> resolve returns None + status, body = _post( + f"{url}/a2a/send", + {"from": "taOSmd", "body": "bad role", "thread": "r2", + "recipient": "@taOS-NoSuch"}, + ) + assert status == 400 + assert "@taOS-NoSuch" in body["error"] + _assert_thread_empty(url, "r2") + + +def test_role_send_unreachable_returns_503(resolver_server): + """A configured-but-unreachable resolver -> 503, nothing stored (fail-loud).""" + url, resolver = resolver_server + resolver.set_unreachable(True) + + status, body = _post( + f"{url}/a2a/send", + {"from": "taOSmd", "body": "unreachable", "thread": "r3", + "recipient": "@taOS-PA"}, + ) + assert status == 503 + assert "unavailable" in body["error"].lower() or "resolver" in body["error"].lower() + _assert_thread_empty(url, "r3") + + +def test_role_send_no_resolver_returns_400(live_server): + """A role recipient with no resolver configured -> 400, nothing stored.""" + status, body = _post( + f"{live_server}/a2a/send", + {"from": "taOSmd", "body": "no resolver", "thread": "r4", + "recipient": "@taOS-PA"}, + ) + assert status == 400 + assert "resolver" in body["error"].lower() + _assert_thread_empty(live_server, "r4") + + +def test_role_send_verbosely_stored_even_when_role_value_is_arbitrary(live_server): + """A non-role recipient is stored verbatim with no resolver involvement.""" + status, body = _post( + f"{live_server}/a2a/send", + {"from": "taOSmd", "body": "plain agent recipient", "thread": "r5", + "recipient": "@plainhandle"}, + ) + assert status == 200, body + assert body["recipient"] == "@plainhandle" + + +# --- Delivery-time role resolution (taOS#2155) ------------------------------ + +def test_role_read_filter_resolves_to_holder(resolver_server): + """?recipient=@taOS-PA returns role-addressed messages with resolved_to.""" + url, resolver = resolver_server + resolver.set_holders({"@taOS-PA": "holderA"}) + + status, body = _post( + f"{url}/a2a/send", + {"from": "taOSmd", "body": "deliver to role", "thread": "del", + "recipient": "@taOS-PA"}, + ) + assert status == 200, body + + status, body = _get(f"{url}/a2a/messages?thread=del&recipient=@taOS-PA") + assert status == 200 + msgs = body["messages"] + assert len(msgs) == 1 + assert msgs[0]["recipient"] == "@taOS-PA" + assert msgs[0]["resolved_to"] == "holderA" + + +def test_role_read_filter_reflects_rotation(resolver_server): + """Rotating the resolver answer changes resolved_to WITHOUT a re-send. + + This is the core guarantee of delivery-time resolution: the stored + envelope keeps the role handle; the holder is computed fresh on every + read. The test FAILS if anyone caches resolution at send time (the whole + point of the design). + """ + url, resolver = resolver_server + resolver.set_holders({"@taOS-PA": "holderA"}) + + status, body = _post( + f"{url}/a2a/send", + {"from": "taOSmd", "body": "rotate me", "thread": "rot", + "recipient": "@taOS-PA"}, + ) + assert status == 200, body + + # Before rotation: resolves to holderA. + status, body = _get(f"{url}/a2a/messages?thread=rot&recipient=@taOS-PA") + assert status == 200 + assert body["messages"][0]["resolved_to"] == "holderA" + + # Rotate the holder -- the SAME stored message must now resolve differently. + resolver.set_holders({"@taOS-PA": "holderB"}) + status, body = _get(f"{url}/a2a/messages?thread=rot&recipient=@taOS-PA") + assert status == 200 + assert body["messages"][0]["resolved_to"] == "holderB" + assert body["messages"][0]["recipient"] == "@taOS-PA" # untouched + + +def test_role_read_no_resolver_returns_400(live_server): + """A role-filtered read with no resolver -> 400 (fail-loud, never guess).""" + _post(f"{live_server}/a2a/send", + {"from": "taOSmd", "body": "background msg", "thread": "r6"}) + status, body = _get(f"{live_server}/a2a/messages?thread=r6&recipient=@taOS-PA") + assert status == 400 + assert "resolver" in body["error"].lower() + + +def test_role_read_unreachable_returns_503(resolver_server): + """A role-filtered read with an unreachable resolver -> 503 (fail-loud).""" + url, resolver = resolver_server + resolver.set_holders({"@taOS-PA": "holderA"}) + _post(f"{url}/a2a/send", + {"from": "taOSmd", "body": "x", "thread": "r7", "recipient": "@taOS-PA"}) + resolver.set_unreachable(True) + + status, body = _get(f"{url}/a2a/messages?thread=r7&recipient=@taOS-PA") + assert status == 503 + assert "unavailable" in body["error"].lower() or "resolver" in body["error"].lower() + + +def test_role_read_no_holder_omits_resolved_to(resolver_server): + """A reachable role with no current holder returns messages (no resolved_to).""" + url, resolver = resolver_server + resolver.set_holders({"@taOS-PA": "holderA"}) + _post(f"{url}/a2a/send", + {"from": "taOSmd", "body": "orphan role msg", "thread": "r8", + "recipient": "@taOS-PA"}) + # Rotate to no holder (key absent). + resolver.set_holders({}) + + status, body = _get(f"{url}/a2a/messages?thread=r8&recipient=@taOS-PA") + assert status == 200 + msgs = body["messages"] + assert len(msgs) == 1 + assert msgs[0]["recipient"] == "@taOS-PA" + assert "resolved_to" not in msgs[0] + + +# --- Remote path: recipient survives the forwarded call (taOSmd #2155 inert-param check) --- + +def test_remote_a2a_send_recipient_survives(live_server): + """Recipient survives the RemoteClient -> HTTP -> service -> archive chain. + + This targets the call site most likely to silently drop a new parameter + (RemoteClient forwards the body as JSON and the query string as params). + Uses an agent handle so no resolver is required on the server. + """ + base_url = live_server + rc = RemoteClient(base_url) + + receipt = asyncio.run(rc.a2a_send( + "agent-alpha", "remote recipient test", thread="rmt-rcpt", + recipient="@agent1", + )) + assert receipt["from"] == "agent-alpha" + assert receipt["recipient"] == "@agent1" + + msgs = asyncio.run(rc.a2a_feed(thread="rmt-rcpt", recipient="@agent1")) + assert len(msgs) == 1 + assert msgs[0]["recipient"] == "@agent1" + + +def test_remote_a2a_send_role_recipient_survives(resolver_server): + """A role recipient survives the remote path too; resolved_to is server-side.""" + url, resolver = resolver_server + resolver.set_holders({"@taOS-PA": "holderA"}) + rc = RemoteClient(url) + + receipt = asyncio.run(rc.a2a_send( + "agent-alpha", "remote role recipient", thread="rmt-role", + recipient="@taOS-PA", + )) + assert receipt["recipient"] == "@taOS-PA" + assert "resolved_to" not in receipt + + # The server annotates resolved_to on the read; the client receives it. + msgs = asyncio.run(rc.a2a_feed(thread="rmt-role", recipient="@taOS-PA")) + assert len(msgs) == 1 + assert msgs[0]["recipient"] == "@taOS-PA" + assert msgs[0]["resolved_to"] == "holderA" diff --git a/tests/test_config_role_resolver.py b/tests/test_config_role_resolver.py new file mode 100644 index 00000000..6856f73e --- /dev/null +++ b/tests/test_config_role_resolver.py @@ -0,0 +1,44 @@ +"""Tests for the opt-in A2A role-resolver URL config (taOS#2155). + +When unset, role recipients (``@taOS-*``) are rejected at send time. When set +(env or config file), the HTTP server builds a :class:`~taosmd.role_resolver +.RoleResolver` and uses it for send-time validation and delivery-time +resolution. +""" +from __future__ import annotations + +import pytest + +from taosmd import config + + +@pytest.fixture +def data_dir(tmp_path, monkeypatch): + monkeypatch.delenv("TAOSMD_A2A_ROLE_RESOLVER_URL", raising=False) + return str(tmp_path) + + +def test_unset_is_none(data_dir): + assert config.get_a2a_role_resolver_url(data_dir) is None + + +def test_set_then_get_round_trip(data_dir): + config.set_a2a_role_resolver_url("http://taos.local:8000", data_dir=data_dir) + assert config.get_a2a_role_resolver_url(data_dir) == "http://taos.local:8000" + + +def test_env_overrides_config_file(data_dir, monkeypatch): + config.set_a2a_role_resolver_url("http://from-file", data_dir=data_dir) + monkeypatch.setenv("TAOSMD_A2A_ROLE_RESOLVER_URL", "http://from-env") + assert config.get_a2a_role_resolver_url(data_dir) == "http://from-env" + + +def test_clear_returns_none(data_dir): + config.set_a2a_role_resolver_url("http://taos.local", data_dir=data_dir) + config.set_a2a_role_resolver_url("", clear=True, data_dir=data_dir) + assert config.get_a2a_role_resolver_url(data_dir) is None + + +def test_set_empty_string_raises(data_dir): + with pytest.raises(ValueError): + config.set_a2a_role_resolver_url("", data_dir=data_dir)