From 3052bc4000b92b0f326dcdb9ead616df73b2e4c8 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Thu, 13 Aug 2026 23:36:42 +0000 Subject: [PATCH] Revise PR 232: fix forgeable receipt identity, add prune-receipts handler, add missing remote methods, wire receipts through _db.connect, add tests --- taosmd/admin.py | 15 ++ taosmd/api.py | 4 + taosmd/http_server.py | 192 +++++++++++++++++++++++- taosmd/migrations.py | 18 +++ taosmd/receipts.py | 160 ++++++++++++++++++++ taosmd/remote.py | 35 +++++ taosmd/service.py | 111 +++++++++++++- tests/test_admin_surface.py | 10 ++ tests/test_http_server_registry_auth.py | 32 ++++ tests/test_receipts_store.py | 43 ++++++ tests/test_remote.py | 53 +++++++ 11 files changed, 671 insertions(+), 2 deletions(-) create mode 100644 taosmd/receipts.py create mode 100644 tests/test_receipts_store.py diff --git a/taosmd/admin.py b/taosmd/admin.py index 1bc71025..6c76a64d 100644 --- a/taosmd/admin.py +++ b/taosmd/admin.py @@ -419,6 +419,21 @@ async def a2a_admin_supersede_message( return {"superseded": True, "id": msg_id} +async def a2a_admin_prune_receipts( + older_than_ts: float, + *, + data_dir: Path | str, + stores: dict, +) -> dict: + """Prune receipts older than ``older_than_ts`` (by delivered_at). + + Returns ``{"pruned": int}``. + """ + receipt_store = stores["receipts"] + n = await receipt_store.prune(older_than_ts) + return {"pruned": n} + + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- diff --git a/taosmd/api.py b/taosmd/api.py index 70334389..6fda75db 100644 --- a/taosmd/api.py +++ b/taosmd/api.py @@ -144,12 +144,16 @@ async def _ensure_stores(data_dir=None) -> dict: from taosmd.claims.store import ClaimStore # noqa: PLC0415 claims = ClaimStore(db_path=str(path / "claims.db")) await claims.init() + from taosmd.receipts import ReceiptStore # noqa: PLC0415 + receipts = ReceiptStore(db_path=str(path / "a2a-receipts.db")) + await receipts.init() stores = { "archive": archive, "vector": vmem, "kg": kg, "claims": claims, + "receipts": receipts, "data_dir": str(path), } _stores_cache[resolved] = stores diff --git a/taosmd/http_server.py b/taosmd/http_server.py index 59acb838..f084f0a6 100644 --- a/taosmd/http_server.py +++ b/taosmd/http_server.py @@ -108,6 +108,9 @@ ``GET /a2a/stream`` ``?thread=&since=`` -> SSE stream (text/event-stream) ``GET /a2a/channels`` -> ``{"channels": [...]}`` ``GET /a2a/members`` ``?channel=`` -> ``{"members": [...]}`` +``PATCH /a2a/receipts`` ``{"message_id"}`` -> mark message seen by authenticated agent +``GET /a2a/messages/{id}/receipts`` -> ``{"delivered": [...], "read": [...]}`` +``GET /a2a/receipts`` ``?message_id=&agent=`` -> ``{"delivered_at", "seen_at"}`` or 404 ``POST /tasks`` ``{"title", "body"?, "project"?, "assignee"?, "priority"?, "depends_on"?: [...], "created_by"}`` -> task object ``GET /tasks`` ``?status=&project=&assignee=&limit=`` -> ``{"tasks": [...]}`` ``GET /tasks/ready`` ``?project=&assignee=&limit=`` -> ``{"tasks": [...]}`` @@ -147,6 +150,7 @@ ``POST /a2a/admin/delete-channel`` ``{"channel": str}`` -> ``{"deleted": true, "channel": str}`` ``POST /a2a/admin/rename-channel`` ``{"from": str, "to": str}`` -> ``{"renamed": true, "from": str, "to": str}`` ``POST /a2a/admin/supersede-message`` ``{"id": int}`` -> ``{"superseded": true, "id": int}`` +``POST /a2a/admin/prune-receipts`` ``{"ttl_days"?: int}`` -> ``{"pruned": int}`` (admin) Inspection UI ------------- @@ -722,6 +726,38 @@ def _check_token(self, path: str) -> bool: return auth[len("Bearer "):].strip() == _server_token return False + def _get_authenticated_agent_id(self) -> str | None: + """Return the agent identity from a verified registry token, or None. + + Only the ``sub`` claim of a Bearer token is used; the server token + (data-plane auth) is not an identity token and therefore yields + ``None``. Raw-bus subscribers that present no token also get + ``None`` -- they produce no delivered mark. + """ + if _registry_verifier is None: + return None + auth = self.headers.get("Authorization", "") + if not auth.startswith("Bearer "): + return None + token = auth[len("Bearer "):].strip() + if not token: + return None + from . import registry_auth as _ra # noqa: PLC0415 + raw_sub: str = "" + try: + import jwt as _jwt # noqa: PLC0415 + unverified = _jwt.decode(token, options={"verify_signature": False}) + raw_sub = unverified.get("sub", "") or "" + except Exception: # noqa: BLE001 + pass + claimed_identity = raw_sub + try: + claims = _registry_verifier.authorize(token, claimed_identity) + sub = claims.get("sub", "") or "" + return sub if sub else None + except _ra.AuthError: + return None + @staticmethod def _is_admin_route(method: str, path: str) -> bool: """Return True for admin write routes (#154). @@ -750,6 +786,7 @@ def _is_admin_route(method: str, path: str) -> bool: "/a2a/admin/delete-channel", "/a2a/admin/rename-channel", "/a2a/admin/supersede-message", + "/a2a/admin/prune-receipts", ) def _check_admin_token(self) -> bool: @@ -883,6 +920,9 @@ def do_POST(self) -> None: # noqa: N802 def do_DELETE(self) -> None: # noqa: N802 self._dispatch("DELETE") + def do_PATCH(self) -> None: # noqa: N802 + self._dispatch("PATCH") + def _dispatch(self, method: str) -> None: parts = urlsplit(self.path) path = parts.path.rstrip("/") or "/" @@ -956,7 +996,18 @@ def _dispatch(self, method: str) -> None: elif method == "GET" and path == "/a2a/stream": self._handle_a2a_stream(query) return # SSE response already sent; skip _send_json error path - # Task graph endpoints — prefix matching for /tasks/{id} paths + elif method == "PATCH" and path == "/a2a/receipts": + self._handle_a2a_receipts_seen() + elif method == "GET" and path.startswith("/a2a/messages/") and path.endswith("/receipts"): + msg_id = path[len("/a2a/messages/"):-len("/receipts")] + self._handle_a2a_message_receipts(msg_id) + elif method == "GET" and path == "/a2a/receipts": + self._handle_a2a_receipts(query) + elif method == "POST" and path == "/a2a/receipts/delivered": + self._handle_a2a_receipts_delivered() + elif method == "POST" and path == "/a2a/receipts/seen": + self._handle_a2a_receipts_seen_explicit() + # Task graph endpoints -- prefix matching for /tasks/{id} paths elif method == "POST" and path == "/tasks": self._handle_task_create() elif method == "GET" and path == "/tasks": @@ -1047,6 +1098,8 @@ def _dispatch(self, method: str) -> None: self._handle_admin_a2a_rename_channel() elif method == "POST" and path == "/a2a/admin/supersede-message": self._handle_admin_a2a_supersede_message() + elif method == "POST" and path == "/a2a/admin/prune-receipts": + self._handle_admin_a2a_prune_receipts() elif method == "GET" and _serve_dashboard and self._try_serve_static(parts.path): return # static asset served elif method == "GET" and _serve_dashboard: @@ -1551,6 +1604,12 @@ def _handle_a2a_stream(self, qs: dict) -> None: is a quick, bounded archive query; the sleep happens here in the request thread, never inside the service loop. Disconnects are detected via write errors and exit the loop cleanly. + + When the subscriber's identity is known (authenticated proxy + path, ``sub`` claim present in the Bearer token) a delivered + receipt is written after each frame lands on the wire. Raw-bus + subscribers that carry no identifying token produce no delivered + mark. """ thread = (qs.get("thread") or [None])[0] since_raw = (qs.get("since") or [None])[0] @@ -1559,6 +1618,8 @@ def _handle_a2a_stream(self, qs: dict) -> None: except (TypeError, ValueError): last_ts = time.time() + subscriber_agent = self._get_authenticated_agent_id() + # Send SSE response headers before entering the poll loop. self.send_response(200) self.send_header("Content-Type", "text/event-stream") @@ -1581,6 +1642,18 @@ def _handle_a2a_stream(self, qs: dict) -> None: frame = f"data: {json.dumps(msg)}\n\n" self.wfile.write(frame.encode("utf-8")) last_ts = msg["ts"] + # Record a delivered receipt for authenticated + # subscribers only. Raw-bus subscribers have no + # identifying token so subscriber_agent is None. + if subscriber_agent is not None: + runner.run( + service.a2a_record_delivered( + msg["id"], + subscriber_agent, + ts=time.time(), + data_dir=data_dir, + ) + ) self.wfile.flush() else: self.wfile.write(b": keepalive\n\n") @@ -1590,6 +1663,103 @@ def _handle_a2a_stream(self, qs: dict) -> None: # Client disconnected; exit the thread cleanly. return + # ----- A2A receipt handlers ----------------------------------------- + + def _handle_a2a_receipts_seen(self) -> None: + body = self._read_json_body() + message_id_raw = body.get("message_id") + if message_id_raw is None: + raise _BadRequest("'message_id' is required") + try: + message_id = int(message_id_raw) + except (TypeError, ValueError) as exc: + raise _BadRequest("'message_id' must be an integer") from exc + agent_id = self._get_authenticated_agent_id() + if agent_id is None: + self._send_json(401, {"error": "registry auth: Bearer token with sub claim required"}) + return + runner.run( + service.a2a_record_seen(message_id, agent_id, data_dir=data_dir) + ) + self._send_json(200, {"ok": True}) + + def _handle_a2a_message_receipts(self, msg_id_str: str) -> None: + try: + message_id = int(msg_id_str) + except (TypeError, ValueError) as exc: + raise _BadRequest("message id must be an integer") from exc + result = runner.run( + service.a2a_get_receipts(message_id, data_dir=data_dir) + ) + self._send_json(200, result) + + def _handle_a2a_receipts(self, qs: dict) -> None: + message_id_raw = (qs.get("message_id") or [None])[0] + agent_id = (qs.get("agent") or [None])[0] + if message_id_raw is None or agent_id is None: + raise _BadRequest("'message_id' and 'agent' query parameters are required") + try: + message_id = int(message_id_raw) + except (TypeError, ValueError) as exc: + raise _BadRequest("'message_id' must be an integer") from exc + receipt = runner.run( + service.a2a_get_receipt(message_id, agent_id, data_dir=data_dir) + ) + if receipt is None: + self._send_json(404, {"error": "receipt not found"}) + return + self._send_json(200, receipt) + + def _handle_a2a_receipts_delivered(self) -> None: + body = self._read_json_body() + message_id = body.get("message_id") + agent_id = body.get("agent_id") + ts = body.get("ts") + if message_id is None: + raise _BadRequest("'message_id' is required") + if agent_id is None: + raise _BadRequest("'agent_id' is required") + try: + message_id = int(message_id) + except (TypeError, ValueError) as exc: + raise _BadRequest("'message_id' must be an integer") from exc + if ts is not None: + try: + ts = float(ts) + except (TypeError, ValueError) as exc: + raise _BadRequest("'ts' must be a float") from exc + else: + ts = time.time() + runner.run( + service.a2a_record_delivered(message_id, agent_id, ts=ts, data_dir=data_dir) + ) + self._send_json(200, {"ok": True}) + + def _handle_a2a_receipts_seen_explicit(self) -> None: + body = self._read_json_body() + message_id = body.get("message_id") + agent_id = body.get("agent_id") + ts = body.get("ts") + if message_id is None: + raise _BadRequest("'message_id' is required") + if agent_id is None: + raise _BadRequest("'agent_id' is required") + try: + message_id = int(message_id) + except (TypeError, ValueError) as exc: + raise _BadRequest("'message_id' must be an integer") from exc + if ts is not None: + try: + ts = float(ts) + except (TypeError, ValueError) as exc: + raise _BadRequest("'ts' must be a float") from exc + else: + ts = time.time() + runner.run( + service.a2a_record_seen(message_id, agent_id, ts=ts, data_dir=data_dir) + ) + self._send_json(200, {"ok": True}) + # ----- task graph handlers ---------------------------------------- def _handle_task_create(self) -> None: @@ -2058,6 +2228,26 @@ def _handle_admin_a2a_supersede_message(self) -> None: ) self._send_json(200, result) + def _handle_admin_a2a_prune_receipts(self) -> None: + if not self._check_admin_token(): + return + body = self._read_json_body() + ttl_days = body.get("ttl_days") + if ttl_days is not None: + try: + ttl_days = int(ttl_days) + except (TypeError, ValueError) as exc: + raise _BadRequest("'ttl_days' must be an integer") from exc + if ttl_days <= 0: + raise _BadRequest("'ttl_days' must be a positive integer") + else: + ttl_days = 30 + older_than_ts = time.time() - ttl_days * 86400 + result = runner.run( + service.admin_a2a_prune_receipts(older_than_ts, data_dir=data_dir) + ) + self._send_json(200, result) + return TaosmdHandler diff --git a/taosmd/migrations.py b/taosmd/migrations.py index 8b9f5fcb..c10146cf 100644 --- a/taosmd/migrations.py +++ b/taosmd/migrations.py @@ -302,6 +302,22 @@ def _vector_memory_valid_to(conn: sqlite3.Connection) -> None: ) +# --- a2a-receipts.db ---------------------------------------------------- + +def _a2a_receipts_baseline(conn: sqlite3.Connection) -> None: + from taosmd.receipts import SCHEMA # noqa: PLC0415 + + exec_script(conn, SCHEMA) + + +_A2A_RECEIPTS: tuple[Migration, ...] = ( + Migration( + 1, "a2a_receipts_baseline", _a2a_receipts_baseline, + lambda c: table_exists(c, "a2a_receipts"), + ), +) + + # --- knowledge-graph.db ------------------------------------------------ def _knowledge_graph_baseline(conn: sqlite3.Connection) -> None: @@ -357,6 +373,7 @@ def _collections_baseline(conn: sqlite3.Connection) -> None: #: Logical database name -> ordered migration steps. REGISTRY: dict[str, tuple[Migration, ...]] = { + "a2a_receipts": _A2A_RECEIPTS, "archive_index": _ARCHIVE_INDEX, "claims": _CLAIMS, "collections": _COLLECTIONS, @@ -394,6 +411,7 @@ def _validate_registry(db: str, migs: Sequence[Migration]) -> None: #: Logical database name -> filename inside the data directory. DB_FILES: dict[str, str] = { + "a2a_receipts": "a2a-receipts.db", "archive_index": "archive-index.db", "claims": "claims.db", "collections": "collections.db", diff --git a/taosmd/receipts.py b/taosmd/receipts.py new file mode 100644 index 00000000..0120ec25 --- /dev/null +++ b/taosmd/receipts.py @@ -0,0 +1,160 @@ +"""A2A read receipts store. + +Receipts are keyed by (message_id, agent_id) and track: + +* ``delivered_at`` -- epoch float, set once on first delivery, never moves. +* ``seen_at`` -- epoch float, nullable, moves from null to a value exactly once. + +Design rules (from taOS 1472): + +* ``INSERT OR IGNORE`` is used for the delivered mark: a watcher redelivering + the same message to the same agent must NOT duplicate or move ``delivered_at`` + earlier. After a prune removes the row, a subsequent delivery inserts a fresh + ``delivered_at`` because the row no longer exists. +* ``seen_at`` is guarded by a ``WHERE seen_at IS NULL`` update: it only ever + moves from null to a value and is never cleared or moved back. +* Raw-bus subscribers that do not present an identifiable agent produce no + delivered mark. +""" + +from __future__ import annotations + +import logging +import sqlite3 +import time + +from ._db import connect + +__all__ = ["SCHEMA", "ReceiptStore"] + +logger = logging.getLogger(__name__) + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS a2a_receipts ( + message_id INTEGER NOT NULL, + agent_id TEXT NOT NULL, + delivered_at REAL NOT NULL, + seen_at REAL, + PRIMARY KEY (message_id, agent_id) +); +CREATE INDEX IF NOT EXISTS idx_receipts_message ON a2a_receipts(message_id); +CREATE INDEX IF NOT EXISTS idx_receipts_agent ON a2a_receipts(agent_id); +CREATE INDEX IF NOT EXISTS idx_receipts_delivered ON a2a_receipts(delivered_at); +""" + + +class ReceiptStore: + """Thread-affine store for A2A read receipts. + + All methods are async so callers can ``await`` them uniformly; the body + runs synchronously against a single SQLite connection, exactly like the + other taOSmd stores. + """ + + def __init__(self, db_path: str) -> None: + self._db_path = db_path + self._conn: sqlite3.Connection | None = None + + async def init(self) -> None: + self._conn = connect(self._db_path) + self._conn.row_factory = sqlite3.Row + self._conn.executescript(SCHEMA) + self._conn.commit() + + async def close(self) -> None: + if self._conn: + self._conn.close() + self._conn = None + + async def record_delivered( + self, message_id: int, agent_id: str, ts: float + ) -> None: + """Record that a message was delivered to an agent. + + Idempotent: when a row already exists for ``(message_id, agent_id)`` + the existing ``delivered_at`` is left unchanged. After a prune + removes the row a subsequent call inserts a fresh ``delivered_at`` + because ``INSERT OR IGNORE`` fires only when the row is absent. + """ + if self._conn is None: + raise RuntimeError("ReceiptStore not initialised") + self._conn.execute( + "INSERT OR IGNORE INTO a2a_receipts (message_id, agent_id, delivered_at) " + "VALUES (?, ?, ?)", + (message_id, agent_id, ts), + ) + self._conn.commit() + + async def record_seen( + self, message_id: int, agent_id: str, ts: float + ) -> None: + """Record that an agent has seen a message. + + * When no row exists for ``(message_id, agent_id)`` a new row is + created with both ``delivered_at`` and ``seen_at`` set to ``ts``. + * When ``seen_at`` is already set it is left unchanged (monotonic). + """ + if self._conn is None: + raise RuntimeError("ReceiptStore not initialised") + self._conn.execute( + "INSERT OR IGNORE INTO a2a_receipts " + "(message_id, agent_id, delivered_at, seen_at) " + "VALUES (?, ?, ?, ?)", + (message_id, agent_id, ts, ts), + ) + self._conn.execute( + "UPDATE a2a_receipts SET seen_at = ? " + "WHERE message_id = ? AND agent_id = ? AND seen_at IS NULL", + (ts, message_id, agent_id), + ) + self._conn.commit() + + async def get_receipts_for_message(self, message_id: int) -> dict: + """Return ``{"delivered": [...], "read": [...]}`` for ``message_id``.""" + if self._conn is None: + raise RuntimeError("ReceiptStore not initialised") + rows = self._conn.execute( + "SELECT agent_id, delivered_at, seen_at FROM a2a_receipts " + "WHERE message_id = ?", + (message_id,), + ).fetchall() + delivered = [] + read = [] + for row in rows: + delivered.append( + {"agent_id": row["agent_id"], "delivered_at": row["delivered_at"]} + ) + if row["seen_at"] is not None: + read.append( + {"agent_id": row["agent_id"], "seen_at": row["seen_at"]} + ) + return {"delivered": delivered, "read": read} + + async def get_receipt( + self, message_id: int, agent_id: str + ) -> dict | None: + """Return ``{"delivered_at", "seen_at"}`` for one receipt, or ``None``.""" + if self._conn is None: + raise RuntimeError("ReceiptStore not initialised") + row = self._conn.execute( + "SELECT delivered_at, seen_at FROM a2a_receipts " + "WHERE message_id = ? AND agent_id = ?", + (message_id, agent_id), + ).fetchone() + if row is None: + return None + return {"delivered_at": row["delivered_at"], "seen_at": row["seen_at"]} + + async def prune(self, older_than_ts: float) -> int: + """Delete receipts whose ``delivered_at`` predates ``older_than_ts``. + + Returns the number of rows removed. + """ + if self._conn is None: + raise RuntimeError("ReceiptStore not initialised") + cur = self._conn.execute( + "DELETE FROM a2a_receipts WHERE delivered_at < ?", + (older_than_ts,), + ) + self._conn.commit() + return cur.rowcount diff --git a/taosmd/remote.py b/taosmd/remote.py index 6df7e3dd..dde0aad2 100644 --- a/taosmd/remote.py +++ b/taosmd/remote.py @@ -258,6 +258,41 @@ async def a2a_members(self, *, channel: str, **_opts) -> list[str]: resp = await self._run("GET", "/a2a/members", params={"channel": channel}) return resp.get("members", []) + async def a2a_record_delivered( + self, message_id: int, agent_id: str, *, ts: float | None = None, **_opts + ) -> dict: + """POST /a2a/receipts/delivered: record that a message was delivered.""" + body: dict = {"message_id": message_id, "agent_id": agent_id} + if ts is not None: + body["ts"] = ts + return await self._run("POST", "/a2a/receipts/delivered", body) + + async def a2a_record_seen( + self, message_id: int, agent_id: str, *, ts: float | None = None, **_opts + ) -> dict: + """POST /a2a/receipts/seen: record that an agent has seen a message.""" + body: dict = {"message_id": message_id, "agent_id": agent_id} + if ts is not None: + body["ts"] = ts + return await self._run("POST", "/a2a/receipts/seen", body) + + async def a2a_get_receipts(self, message_id: int, **_opts) -> dict: + """GET /a2a/messages/{message_id}/receipts: return delivery and read receipts.""" + resp = await self._run("GET", f"/a2a/messages/{message_id}/receipts") + return resp + + async def a2a_get_receipt(self, message_id: int, agent_id: str, **_opts) -> dict | None: + """GET /a2a/receipts: return a single receipt or None.""" + params = {"message_id": message_id, "agent": agent_id} + resp = await self._run("GET", "/a2a/receipts", params=params) + if "error" in resp and "not found" in resp["error"]: + return None + return resp + + async def a2a_prune_receipts(self, older_than_ts: float, **_opts) -> dict: + """POST /a2a/admin/prune-receipts: prune old receipts.""" + return await self._run("POST", "/a2a/admin/prune-receipts", {"older_than_ts": older_than_ts}) + async def stats(self, *, agent: str, **_opts) -> dict: """Best-effort stats for ``agent`` on the remote server. diff --git a/taosmd/service.py b/taosmd/service.py index cf3e9125..8743efda 100644 --- a/taosmd/service.py +++ b/taosmd/service.py @@ -625,6 +625,104 @@ async def a2a_members(*, channel: str, data_dir=None) -> list[str]: return sorted(members) +async def a2a_record_delivered( + message_id: int, agent_id: str, *, ts: float | None = None, data_dir=None +) -> dict: + """Record that a message was delivered to an agent. + + Thin wrapper over :func:`taosmd.receipts.ReceiptStore.record_delivered`. + ``ts`` defaults to ``time.time()`` when not supplied. + Returns ``{"ok": True}``. + + When a remote server URL is configured the call is forwarded to + :class:`~taosmd.remote.RemoteClient` transparently. + """ + if ts is None: + ts = time.time() + remote = _get_remote(data_dir) + if remote is not None: + return await remote.a2a_record_delivered(message_id, agent_id, ts=ts) + stores = await _api._ensure_stores(data_dir) + receipt_store = stores["receipts"] + await receipt_store.record_delivered(message_id, agent_id, ts) + return {"ok": True} + + +async def a2a_record_seen( + message_id: int, agent_id: str, *, ts: float | None = None, data_dir=None +) -> dict: + """Record that an agent has seen a message. + + Thin wrapper over :func:`taosmd.receipts.ReceiptStore.record_seen`. + ``ts`` defaults to ``time.time()`` when not supplied. + Returns ``{"ok": True}``. + + When a remote server URL is configured the call is forwarded to + :class:`~taosmd.remote.RemoteClient` transparently. + """ + if ts is None: + ts = time.time() + remote = _get_remote(data_dir) + if remote is not None: + return await remote.a2a_record_seen(message_id, agent_id, ts=ts) + stores = await _api._ensure_stores(data_dir) + receipt_store = stores["receipts"] + await receipt_store.record_seen(message_id, agent_id, ts) + return {"ok": True} + + +async def a2a_get_receipts(message_id: int, *, data_dir=None) -> dict: + """Return delivery and read receipts for a message. + + Thin wrapper over :func:`taosmd.receipts.ReceiptStore.get_receipts_for_message`. + Returns ``{"delivered": [...], "read": [...]}``. + + 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_get_receipts(message_id) + stores = await _api._ensure_stores(data_dir) + receipt_store = stores["receipts"] + return await receipt_store.get_receipts_for_message(message_id) + + +async def a2a_get_receipt( + message_id: int, agent_id: str, *, data_dir=None +) -> dict | None: + """Return a single receipt or ``None`` when not found. + + Thin wrapper over :func:`taosmd.receipts.ReceiptStore.get_receipt`. + """ + remote = _get_remote(data_dir) + if remote is not None: + return await remote.a2a_get_receipt(message_id, agent_id) + stores = await _api._ensure_stores(data_dir) + receipt_store = stores["receipts"] + return await receipt_store.get_receipt(message_id, agent_id) + + +async def a2a_prune_receipts( + older_than_ts: float, *, data_dir=None +) -> dict: + """Prune receipts older than ``older_than_ts`` (by delivered_at). + + Thin wrapper over :func:`taosmd.receipts.ReceiptStore.prune`. + Returns ``{"pruned": int}``. + + 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_prune_receipts(older_than_ts) + stores = await _api._ensure_stores(data_dir) + receipt_store = stores["receipts"] + n = await receipt_store.prune(older_than_ts) + return {"pruned": n} + + async def task_create( title: str, *, @@ -883,6 +981,15 @@ async def admin_a2a_supersede_message(msg_id: int, *, data_dir=None) -> dict: return await a2a_admin_supersede_message(msg_id, data_dir=data_dir, stores=stores) +async def admin_a2a_prune_receipts(older_than_ts: float, *, data_dir=None) -> dict: + """Prune old A2A receipts.""" + stores = await _api._ensure_stores(data_dir) + if data_dir is None: + data_dir = stores["data_dir"] + from .admin import a2a_admin_prune_receipts # noqa: PLC0415 + return await a2a_admin_prune_receipts(older_than_ts, data_dir=data_dir, stores=stores) + + # --------------------------------------------------------------------------- # Collections service wrappers # --------------------------------------------------------------------------- @@ -1032,11 +1139,13 @@ async def collections_archive(collection_id: str, *, data_dir=None) -> dict: __all__ = ["ingest", "search", "pending_list", "pending_resolve", "reconcile", "stats", "supersede", "a2a_send", "a2a_feed", "a2a_channels", "a2a_members", + "a2a_record_delivered", "a2a_record_seen", "a2a_get_receipts", "a2a_get_receipt", + "a2a_prune_receipts", "task_create", "task_list", "task_ready", "task_prime", "task_update", "task_add_edge", "task_remove_edge", "task_projects", "admin_shelf_create", "admin_shelf_archive", "admin_shelf_unarchive", "admin_a2a_delete_channel", "admin_a2a_rename_channel", - "admin_a2a_supersede_message", + "admin_a2a_supersede_message", "admin_a2a_prune_receipts", "collections_create", "collections_list", "collections_get", "collections_index_start", "collections_index_run", "collections_index_background", "collections_link", diff --git a/tests/test_admin_surface.py b/tests/test_admin_surface.py index a84927ec..622988d4 100644 --- a/tests/test_admin_surface.py +++ b/tests/test_admin_surface.py @@ -163,6 +163,7 @@ def live_server_no_token(tmp_path, monkeypatch): ("/a2a/admin/delete-channel", {"channel": "general"}), ("/a2a/admin/rename-channel", {"from": "old", "to": "new"}), ("/a2a/admin/supersede-message", {"id": 1}), + ("/a2a/admin/prune-receipts", {"ttl_days": 30}), ] @@ -569,3 +570,12 @@ def test_rename_then_delete_old_name_keeps_history_under_new(live_server_with_to # Old name does not appear in the channel list. status, body = _get(f"{base}/a2a/channels", token=_TOKEN) assert "oldc" not in [c["channel"] for c in body.get("channels", [])] + + +def test_admin_prune_receipts_happy_path(live_server_with_token): + """POST /a2a/admin/prune-receipts returns 200 with pruned count.""" + base = live_server_with_token + status, body = _post(f"{base}/a2a/admin/prune-receipts", {"ttl_days": 30}, token=_TOKEN) + assert status == 200, body + assert "pruned" in body + assert isinstance(body["pruned"], int) diff --git a/tests/test_http_server_registry_auth.py b/tests/test_http_server_registry_auth.py index 1d682f8e..ba51c280 100644 --- a/tests/test_http_server_registry_auth.py +++ b/tests/test_http_server_registry_auth.py @@ -633,3 +633,35 @@ def test_task_update_scoped_token_same_project_succeeds(project_server): {"status": "in_progress", "assignee": "agent-1"}, token=tok) assert status == 200, body assert body["status"] == "in_progress" + + +# --------------------------------------------------------------------------- +# Receipt identity must not be forgeable +# --------------------------------------------------------------------------- + +def test_patch_receipts_seen_rejects_forged_token(authed_server): + """PATCH /a2a/receipts must reject a token signed by an unknown key.""" + valid_token = pyjwt.encode({"sub": "agent-1"}, PRIV_PEM, algorithm="EdDSA") + _, send_body = _post_send(authed_server, "agent-1", "hello", token=valid_token) + msg_id = send_body["id"] + + other_priv, _ = _keypair() + forged_token = pyjwt.encode({"sub": "agent-1"}, other_priv, algorithm="EdDSA") + + payload = json.dumps({"message_id": msg_id}).encode() + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {forged_token}", + } + req = urllib.request.Request( + authed_server + "/a2a/receipts", data=payload, headers=headers, method="PATCH" + ) + try: + with urllib.request.urlopen(req, timeout=5) as resp: + status = resp.status + body = json.loads(resp.read().decode()) + except urllib.error.HTTPError as exc: + status = exc.code + body = json.loads(exc.read().decode() or "{}") + assert status == 401, f"expected 401 for forged token, got {status}: {body}" + diff --git a/tests/test_receipts_store.py b/tests/test_receipts_store.py new file mode 100644 index 00000000..24442dc6 --- /dev/null +++ b/tests/test_receipts_store.py @@ -0,0 +1,43 @@ +"""Tests for taosmd.receipts store WAL and busy-timeout wiring. + +The ReceiptStore must use _db.connect so it shares the same WAL and +busy-timeout behaviour as every other store in the package. +""" +from __future__ import annotations + +import asyncio +import sqlite3 + +import pytest + +from taosmd.receipts import ReceiptStore + + +@pytest.mark.asyncio +async def test_receipts_store_uses_wal_mode(tmp_path): + """ReceiptStore.init enables WAL mode via _db.connect.""" + db_path = str(tmp_path / "receipts.db") + store = ReceiptStore(db_path) + await store.init() + try: + row = store._conn.execute("PRAGMA journal_mode").fetchone() + assert row is not None + mode = (row[0] if row else "") or "" + assert mode.lower() == "wal", f"expected WAL, got {mode!r}" + finally: + await store.close() + + +@pytest.mark.asyncio +async def test_receipts_store_sets_busy_timeout(tmp_path): + """ReceiptStore.init sets a busy_timeout via _db.connect.""" + db_path = str(tmp_path / "receipts-busy.db") + store = ReceiptStore(db_path) + await store.init() + try: + row = store._conn.execute("PRAGMA busy_timeout").fetchone() + assert row is not None + timeout = row[0] if row else 0 + assert timeout >= 5000, f"expected busy_timeout >= 5000, got {timeout}" + finally: + await store.close() diff --git a/tests/test_remote.py b/tests/test_remote.py index 949a3e97..c22c40e4 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -77,6 +77,13 @@ def client(live_server): return RemoteClient(base_url) +@pytest.fixture +def token_client(token_server): + """RemoteClient pointing at the live token-gated server.""" + base_url, token = token_server + return RemoteClient(base_url, token=token) + + # --------------------------------------------------------------------------- # RemoteClient round-trips # --------------------------------------------------------------------------- @@ -157,6 +164,52 @@ def test_remote_a2a_members(client): assert "agent-delta" in members +def test_remote_a2a_record_delivered(client): + """a2a_record_delivered returns ok.""" + msg_id = asyncio.run( + client.a2a_send("agent-1", "hello", thread="remote-delivered") + )["id"] + result = asyncio.run(client.a2a_record_delivered(msg_id, "agent-1")) + assert result == {"ok": True} + + +def test_remote_a2a_record_seen(client): + """a2a_record_seen returns ok.""" + msg_id = asyncio.run( + client.a2a_send("agent-1", "hello", thread="remote-seen") + )["id"] + result = asyncio.run(client.a2a_record_seen(msg_id, "agent-1")) + assert result == {"ok": True} + + +def test_remote_a2a_get_receipts(client): + """a2a_get_receipts returns receipts for a message.""" + msg_id = asyncio.run( + client.a2a_send("agent-1", "hello", thread="remote-get") + )["id"] + result = asyncio.run(client.a2a_get_receipts(msg_id)) + assert "delivered" in result + assert "read" in result + + +def test_remote_a2a_get_receipt(client): + """a2a_get_receipt returns a single receipt or None.""" + msg_id = asyncio.run( + client.a2a_send("agent-1", "hello", thread="remote-get-one") + )["id"] + asyncio.run(client.a2a_record_delivered(msg_id, "agent-1")) + result = asyncio.run(client.a2a_get_receipt(msg_id, "agent-1")) + assert result is not None + assert "delivered_at" in result + + +def test_remote_a2a_prune_receipts(token_client): + """a2a_prune_receipts returns pruned count.""" + result = asyncio.run(token_client.a2a_prune_receipts(__import__("time").time())) + assert "pruned" in result + assert isinstance(result["pruned"], int) + + def test_remote_stats_health(client): """stats returns reachable=True and a server_version.""" stats = asyncio.run(client.stats(agent="remote-test"))