From 21fa535bde600657f6fc8e1617814a27146af5c0 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Thu, 11 Jun 2026 11:29:02 +0100 Subject: [PATCH] feat(admin): shelf lifecycle + A2A channel admin surface Adds the admin surface per the taOS#774 contract and the long-standing A2A channel cleanup ask. Six new POST endpoints, all fail-closed behind the configured server token (403 when no token is set). Shelf lifecycle: POST /shelves creates or returns an existing shelf (idempotent by shelf_id, validated against ^[a-z][a-z0-9_-]{0,62}$). POST /shelves/{id}/archive soft-hides vector rows via valid_to with a hidden_by:shelf-archive: metadata marker so unarchive restores only those rows. expect_empty=true returns 409 when the shelf has active rows. Archive/unarchive events recorded to the zero-loss archive. A2A channel admin: delete-channel hides a channel from /a2a/channels and /a2a/messages feeds; rename-channel adds an alias (sends to old name are redirected, reads of new name include old history); supersede-message hides one message by id. State persisted in data/a2a-admin-state.json via atomic tmp+os.replace writes. The a2a_feed, a2a_channels, and a2a_send service functions all consult the sidecar at query time. 28 new tests in tests/test_admin_surface.py. Full suite 746 tests green. --- CHANGELOG.md | 20 ++ taosmd/admin.py | 451 +++++++++++++++++++++++++++++ taosmd/http_server.py | 177 ++++++++++++ taosmd/service.py | 173 +++++++++++- tests/test_admin_surface.py | 545 ++++++++++++++++++++++++++++++++++++ 5 files changed, 1357 insertions(+), 9 deletions(-) create mode 100644 taosmd/admin.py create mode 100644 tests/test_admin_surface.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 54967a14..ac3ee87a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,26 @@ ## Unreleased ### Added +- **Admin surface: shelf lifecycle and A2A channel admin (taOS#774).** New + `taosmd/admin.py` module and six HTTP endpoints gated behind the configured + server token (fail-closed: 403 when no token is set, 401 on wrong token). + Shelf lifecycle: `POST /shelves` creates or returns an existing shelf + (idempotent by `shelf_id`; shelf_id must match `^[a-z][a-z0-9_-]{0,62}$`); + `POST /shelves/{id}/archive` soft-hides the shelf's active vector rows by + stamping `valid_to` and embedding a `hidden_by: shelf-archive:` marker + in each row's metadata so `POST /shelves/{id}/unarchive` can restore exactly + those rows and nothing else (rows superseded for other reasons such as + corrections are not resurrected). `?expect_empty=true` returns 409 without + archiving when the shelf has active rows. Archive and unarchive events are + appended to the zero-loss archive. A2A channel admin: `POST + /a2a/admin/delete-channel` soft-deletes a channel so it is hidden from + `/a2a/channels` and `/a2a/messages` responses while messages remain in the + archive; `POST /a2a/admin/rename-channel` adds a channel alias so sends to + the old name are redirected and reads of the new name include history from + the old name (stored rows are not mutated); `POST + /a2a/admin/supersede-message` hides one message by id from feed responses. + The deleted-channels set, alias map, and superseded-message set are persisted + in `data/a2a-admin-state.json` via atomic tmp+os.replace writes. - **Project-scoped grants (taOS#744).** Registry grant rows now carry an optional `project_id` field; `GrantsVerifier.has_grant` accepts a matching `project_id` keyword argument and a grant row with no `project_id` acts as diff --git a/taosmd/admin.py b/taosmd/admin.py new file mode 100644 index 00000000..1bc71025 --- /dev/null +++ b/taosmd/admin.py @@ -0,0 +1,451 @@ +"""Admin surface: shelf lifecycle and A2A channel admin operations. + +This module provides the service-layer logic and sidecar state for: + +- Shelf lifecycle (POST /shelves, POST /shelves/{id}/archive, + POST /shelves/{id}/unarchive) per the taOS#774 contract. +- A2A channel admin (delete-channel, rename-channel, supersede-message) + that hide or redirect content without mutating the zero-loss archive. + +Shelf lifecycle +--------------- +A shelf IS an agent registration (per agents.py). Creating a shelf calls +ensure_agent with optional project_id and display_name stored in the agent +record's metadata field. Archiving a shelf soft-hides its vector rows using +the same valid_to supersede machinery as the correction path, but tags each +hidden row's metadata with ``hidden_by: "shelf-archive:"`` so that +unarchive can restore exactly those rows and nothing else. + +A2A admin sidecar +----------------- +The deleted-channels set, channel-alias map, and superseded-message set are +persisted in a small JSON sidecar under the data dir at +``data/a2a-admin-state.json``. Writes use atomic tmp+os.replace exactly like +agents.py and config.py. + +At query time (in service.py wrappers) callers load this sidecar and filter: +- a2a_channels() skips channels in deleted_channels +- a2a_feed() skips channels in deleted_channels, resolves aliases, skips + message ids in superseded_messages +- a2a_send() redirects sends to renamed channels via the alias map +""" + +from __future__ import annotations + +import json +import logging +import os +import re +import time +from pathlib import Path +from typing import Any + +from .agents import AgentRegistry, InvalidAgentNameError, NAME_RE +from .archive import EVENT_A2A + +logger = logging.getLogger(__name__) + +# Marker prefix embedded in metadata so unarchive only un-hides rows +# that were hidden by this specific shelf-archive operation. +_SHELF_ARCHIVE_MARKER = "shelf-archive" + + +# --------------------------------------------------------------------------- +# A2A admin sidecar +# --------------------------------------------------------------------------- + +class A2AAdminState: + """Manages the persisted sidecar for A2A admin operations. + + State shape:: + + { + "deleted_channels": ["chan1", "chan2"], + "channel_aliases": {"old_name": "new_name"}, + "superseded_messages": [42, 99] + } + + All writes are atomic (tmp + os.replace). The sidecar is read fresh on + each operation so multiple processes / threads see a consistent view. + """ + + def __init__(self, data_dir: Path | str) -> None: + self._path = Path(data_dir) / "a2a-admin-state.json" + + def _read(self) -> dict[str, Any]: + if not self._path.exists(): + return {"deleted_channels": [], "channel_aliases": {}, "superseded_messages": []} + try: + raw = json.loads(self._path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + raw = {} + return { + "deleted_channels": list(raw.get("deleted_channels") or []), + "channel_aliases": dict(raw.get("channel_aliases") or {}), + "superseded_messages": list(raw.get("superseded_messages") or []), + } + + def _write(self, state: dict[str, Any]) -> None: + self._path.parent.mkdir(parents=True, exist_ok=True) + tmp = self._path.with_suffix(".json.tmp") + tmp.write_text(json.dumps(state, indent=2), encoding="utf-8") + os.replace(tmp, self._path) + + # ----- reads ---------------------------------------------------------- + + def deleted_channels(self) -> set[str]: + return set(self._read()["deleted_channels"]) + + def channel_aliases(self) -> dict[str, str]: + return self._read()["channel_aliases"] + + def superseded_messages(self) -> set[int]: + return set(int(x) for x in self._read()["superseded_messages"]) + + def resolve_channel(self, channel: str) -> str: + """Return the canonical channel name after following any alias chain.""" + aliases = self.channel_aliases() + visited: set[str] = set() + while channel in aliases and channel not in visited: + visited.add(channel) + channel = aliases[channel] + return channel + + # ----- writes ---------------------------------------------------------- + + def delete_channel(self, channel: str) -> None: + state = self._read() + if channel not in state["deleted_channels"]: + state["deleted_channels"].append(channel) + self._write(state) + + def add_alias(self, from_channel: str, to_channel: str) -> None: + """Map from_channel -> to_channel. Overwrites an existing mapping.""" + state = self._read() + state["channel_aliases"][from_channel] = to_channel + self._write(state) + + def supersede_message(self, msg_id: int) -> None: + state = self._read() + if msg_id not in state["superseded_messages"]: + state["superseded_messages"].append(msg_id) + self._write(state) + + +# --------------------------------------------------------------------------- +# Shelf operations +# --------------------------------------------------------------------------- + +async def shelf_create( + shelf_id: str, + *, + project_id: str | None = None, + display_name: str | None = None, + data_dir: Path | str, +) -> dict: + """Create or return an existing shelf (agent registration). + + Returns ``{"shelf": {...}, "created": bool}``. shelf_id must match + ``^[a-z][a-z0-9_-]{0,62}$``; raises ``InvalidAgentNameError`` (400) + otherwise. Idempotent: if the shelf already exists (and is not archived), + returns it with ``created=False``. + """ + if not NAME_RE.match(shelf_id): + raise InvalidAgentNameError( + f"shelf_id must match ^[a-z][a-z0-9_-]{{0,62}}$ (got {shelf_id!r})" + ) + + registry = AgentRegistry(data_dir) + data = registry._read() + + existing = next((a for a in data["agents"] if a["name"] == shelf_id), None) + if existing is not None: + # Return existing shelf (even if archived -- caller can unarchive separately) + rec = dict(existing) + return {"shelf": rec, "created": False} + + # Create new registration. Store project_id and display_name in the record. + from .agents import AgentRecord, _default_librarian # noqa: PLC0415 + record = AgentRecord( + name=shelf_id, + display_name=display_name or shelf_id, + created_at=int(time.time()), + librarian=_default_librarian(), + ) + rec_dict = record.to_dict() + # Store extra metadata on the record + if project_id: + rec_dict["project_id"] = project_id + if display_name: + rec_dict["display_name"] = display_name + + data["agents"].append(rec_dict) + registry._write(data) + registry._agent_dir(shelf_id).mkdir(parents=True, exist_ok=True) + + return {"shelf": rec_dict, "created": True} + + +async def shelf_archive( + shelf_id: str, + *, + expect_empty: bool = False, + data_dir: Path | str, + stores: dict, +) -> dict: + """Archive a shelf: mark it archived and soft-hide its vector rows. + + If ``expect_empty=True`` and the shelf has active vector rows, returns + a ``409 Conflict`` signal by raising ``ShelfNotEmptyError``. Archiving + an already-archived shelf is a no-op (returns ``rows_hidden=0``). + + Archive events are recorded to the zero-loss archive (event_type "shelf"). + """ + registry = AgentRegistry(data_dir) + data = registry._read() + agent_rec = next((a for a in data["agents"] if a["name"] == shelf_id), None) + if agent_rec is None: + raise ShelfNotFoundError(f"shelf {shelf_id!r} not found") + + # Already archived: no-op + meta = agent_rec.get("metadata") or {} + if meta.get("archived_at"): + return {"archived": True, "rows_hidden": 0} + + vmem = stores["vector"] + archive = stores["archive"] + + # Count active rows for this shelf + rows = _get_active_shelf_rows(vmem, shelf_id) + + if expect_empty and rows: + raise ShelfNotEmptyError( + f"shelf {shelf_id!r} has {len(rows)} active row(s); " + "use expect_empty=false to archive anyway" + ) + + ts = time.time() + marker = f"{_SHELF_ARCHIVE_MARKER}:{ts}" + + # Soft-hide the rows by patching their metadata to include the marker + # and stamping valid_to. We need to update metadata_json before setting + # valid_to so the marker is visible for unarchive. + hidden = 0 + for row in rows: + row_id = row["id"] + row_meta = row["metadata"] + row_meta["hidden_by"] = marker + vmem._conn.execute( + "UPDATE vector_memory SET valid_to = ?, metadata_json = ? " + "WHERE id = ? AND valid_to IS NULL", + (ts, json.dumps(row_meta), row_id), + ) + hidden += 1 + if hidden: + vmem._conn.commit() + vmem._bm25_dirty = True + + # Mark the agent record archived + if "metadata" not in agent_rec or not isinstance(agent_rec["metadata"], dict): + agent_rec["metadata"] = {} + agent_rec["metadata"]["archived_at"] = ts + registry._write(data) + + # Record archive event + await archive.record( + event_type="shelf", + data={"action": "archived", "shelf_id": shelf_id, "rows_hidden": hidden}, + agent_name=shelf_id, + summary=f"shelf {shelf_id} archived; {hidden} row(s) hidden", + ) + + return {"archived": True, "rows_hidden": hidden} + + +async def shelf_unarchive( + shelf_id: str, + *, + data_dir: Path | str, + stores: dict, +) -> dict: + """Unarchive a shelf: clear archived_at and restore only shelf-archive-hidden rows. + + Rows superseded for OTHER reasons (corrections, manual supersede) are NOT + restored. Only rows whose ``hidden_by`` metadata marker starts with + ``"shelf-archive:"`` are restored. + """ + registry = AgentRegistry(data_dir) + data = registry._read() + agent_rec = next((a for a in data["agents"] if a["name"] == shelf_id), None) + if agent_rec is None: + raise ShelfNotFoundError(f"shelf {shelf_id!r} not found") + + vmem = stores["vector"] + archive = stores["archive"] + + # Find rows hidden by this shelf's archive operation(s) and restore them + rows_restored = 0 + prefix = f"{_SHELF_ARCHIVE_MARKER}:" + hidden_rows = vmem._conn.execute( + "SELECT id, metadata_json FROM vector_memory WHERE valid_to IS NOT NULL" + ).fetchall() + + for row in hidden_rows: + try: + meta = json.loads(row["metadata_json"]) + except (json.JSONDecodeError, TypeError): + meta = {} + hidden_by = meta.get("hidden_by", "") + if not isinstance(hidden_by, str): + continue + # Check the marker belongs to this shelf + # Format: "shelf-archive:" -- we only restore rows hidden by *this* shelf's + # archive operations. We check that they match the agent's known archived_at + # or any prior archived_at for this shelf. + # Simpler: match rows where hidden_by starts with "shelf-archive:" AND the + # row's agent metadata matches shelf_id. + if not hidden_by.startswith(prefix): + continue + # Verify agent association + row_agent = meta.get("agent") + if row_agent is not None and row_agent != shelf_id: + continue + + # Clear the hidden_by marker and restore the row + del meta["hidden_by"] + vmem._conn.execute( + "UPDATE vector_memory SET valid_to = NULL, metadata_json = ? WHERE id = ?", + (json.dumps(meta), row["id"]), + ) + rows_restored += 1 + + if rows_restored: + vmem._conn.commit() + vmem._bm25_dirty = True + + # Clear archived_at on the agent record + if "metadata" in agent_rec and isinstance(agent_rec["metadata"], dict): + agent_rec["metadata"].pop("archived_at", None) + registry._write(data) + + # Record archive event + await archive.record( + event_type="shelf", + data={"action": "unarchived", "shelf_id": shelf_id, "rows_restored": rows_restored}, + agent_name=shelf_id, + summary=f"shelf {shelf_id} unarchived; {rows_restored} row(s) restored", + ) + + return {"archived": False, "rows_restored": rows_restored} + + +# --------------------------------------------------------------------------- +# A2A admin operations +# --------------------------------------------------------------------------- + +async def a2a_admin_delete_channel( + channel: str, + *, + data_dir: Path | str, + stores: dict, +) -> dict: + """Soft-delete a channel: record an archive event and hide it from feeds. + + Messages stay in the archive (zero-loss). The deleted-channels set in the + sidecar is consulted at query time by a2a_channels and a2a_feed. + """ + state = A2AAdminState(data_dir) + archive = stores["archive"] + + state.delete_channel(channel) + + await archive.record( + event_type=EVENT_A2A, + data={"admin_action": "delete_channel", "channel": channel}, + app_id=channel, + summary=f"A2A admin: channel {channel!r} deleted", + ) + + return {"deleted": True, "channel": channel} + + +async def a2a_admin_rename_channel( + from_channel: str, + to_channel: str, + *, + data_dir: Path | str, + stores: dict, +) -> dict: + """Rename a channel by adding an alias: old -> new. + + New sends to the old name are redirected to the new name. Reads of the + new name include the old name's history (the alias map is consulted at + query time). Stored rows are NOT mutated. Renaming again re-points the + alias. + """ + if not from_channel or not to_channel: + raise ValueError("both 'from' and 'to' channel names are required") + if from_channel == to_channel: + raise ValueError("'from' and 'to' channel names must differ") + + state = A2AAdminState(data_dir) + archive = stores["archive"] + + state.add_alias(from_channel, to_channel) + + await archive.record( + event_type=EVENT_A2A, + data={"admin_action": "rename_channel", "from": from_channel, "to": to_channel}, + app_id=to_channel, + summary=f"A2A admin: channel {from_channel!r} renamed to {to_channel!r}", + ) + + return {"renamed": True, "from": from_channel, "to": to_channel} + + +async def a2a_admin_supersede_message( + msg_id: int, + *, + data_dir: Path | str, + stores: dict, +) -> dict: + """Hide one message from feed responses. Archive row is untouched. + + The message id is added to the superseded-messages set in the sidecar; + a2a_feed and a2a_messages filter it out at query time. + """ + state = A2AAdminState(data_dir) + state.supersede_message(msg_id) + return {"superseded": True, "id": msg_id} + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _get_active_shelf_rows(vmem, shelf_id: str) -> list[dict]: + """Return active (valid_to IS NULL) rows whose metadata agent == shelf_id.""" + rows = vmem._conn.execute( + "SELECT id, text, metadata_json FROM vector_memory WHERE valid_to IS NULL" + ).fetchall() + result = [] + for row in rows: + try: + meta = json.loads(row["metadata_json"]) + except (json.JSONDecodeError, TypeError): + meta = {} + if meta.get("agent") == shelf_id: + result.append({"id": row["id"], "text": row["text"], "metadata": meta}) + return result + + +# --------------------------------------------------------------------------- +# Exceptions +# --------------------------------------------------------------------------- + +class ShelfNotFoundError(KeyError): + """Raised when referencing a shelf that does not exist.""" + + +class ShelfNotEmptyError(ValueError): + """Raised when archiving with expect_empty=True and the shelf has rows.""" diff --git a/taosmd/http_server.py b/taosmd/http_server.py index b4e7dfb3..19b65970 100644 --- a/taosmd/http_server.py +++ b/taosmd/http_server.py @@ -71,6 +71,14 @@ ``POST /tasks/{id}/edges`` ``{"to_id", "type", "created_by"}`` -> edge record ``POST /tasks/{id}/edges/remove`` ``{"to_id", "type"}`` -> edge record with removed_ts +Admin endpoints (all require a configured server token; 403 if none is set) +``POST /shelves`` ``{"shelf_id", "project_id"?, "display_name"?}`` -> ``{"shelf": {...}, "created": bool}`` +``POST /shelves/{id}/archive`` ``?expect_empty=true`` -> ``{"archived": true, "rows_hidden": int}`` +``POST /shelves/{id}/unarchive`` -> ``{"archived": false, "rows_restored": int}`` +``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}`` + Inspection UI ------------- ``GET /`` (and ``GET /ui``) serves a single self-contained HTML page: one @@ -615,6 +623,28 @@ def _check_token(self, path: str) -> bool: return auth[len("Bearer "):].strip() == _server_token return False + def _check_admin_token(self) -> bool: + """Return True when the request carries the correct admin token. + + Admin endpoints FAIL CLOSED: if no server token is configured, all + admin requests return 403. This is the inverse of ``_check_token`` + which passes everything through when no token is configured. + If a token is configured the Bearer must match it exactly. + Returns False and writes the error response when auth fails; the + caller must return immediately in that case. + """ + if not _server_token: + self._send_json( + 403, + {"error": "admin surface requires a configured server token"}, + ) + return False + auth = self.headers.get("Authorization", "") + if auth.startswith("Bearer ") and auth[len("Bearer "):].strip() == _server_token: + return True + self._send_json(401, {"error": "Unauthorized"}) + return False + def _apply_token_binding(self, identity: str | None, project: str | None) -> tuple[str | None, bool]: """Apply optional registry-token project binding for data endpoints. @@ -768,6 +798,32 @@ def _dispatch(self, method: str) -> None: self._send_json(404, {"error": "task id required"}) else: self._handle_task_update(task_id) + # ----- admin surface: shelf lifecycle --------------------- + elif method == "POST" and path == "/shelves": + self._handle_admin_shelf_create() + elif method == "POST" and path.startswith("/shelves/"): + rest = path[len("/shelves/"):] + if rest.endswith("/archive"): + shelf_id = rest[: -len("/archive")] + if not shelf_id: + self._send_json(404, {"error": "shelf id required"}) + else: + self._handle_admin_shelf_archive(shelf_id, query) + elif rest.endswith("/unarchive"): + shelf_id = rest[: -len("/unarchive")] + if not shelf_id: + self._send_json(404, {"error": "shelf id required"}) + else: + self._handle_admin_shelf_unarchive(shelf_id) + else: + self._send_json(404, {"error": f"unknown shelf action: {rest}"}) + # ----- admin surface: A2A channel admin ------------------- + elif method == "POST" and path == "/a2a/admin/delete-channel": + self._handle_admin_a2a_delete_channel() + elif method == "POST" and path == "/a2a/admin/rename-channel": + self._handle_admin_a2a_rename_channel() + elif method == "POST" and path == "/a2a/admin/supersede-message": + self._handle_admin_a2a_supersede_message() elif method == "GET" and _serve_dashboard and self._try_serve_static(parts.path): return # static asset served elif method == "GET" and _serve_dashboard: @@ -1204,6 +1260,123 @@ def _handle_task_remove_edge(self, from_id: str) -> None: raise _BadRequest(str(exc)) from exc self._send_json(200, result) + # ----- admin: shelf lifecycle ------------------------------------ + + def _handle_admin_shelf_create(self) -> None: + if not self._check_admin_token(): + return + body = self._read_json_body() + shelf_id = body.get("shelf_id") + project_id = body.get("project_id") + display_name = body.get("display_name") + if not isinstance(shelf_id, str) or not shelf_id: + raise _BadRequest("'shelf_id' (non-empty string) is required") + if project_id is not None and not isinstance(project_id, str): + raise _BadRequest("'project_id' must be a string when provided") + if display_name is not None and not isinstance(display_name, str): + raise _BadRequest("'display_name' must be a string when provided") + from .admin import InvalidAgentNameError # noqa: PLC0415 + try: + result = runner.run( + service.admin_shelf_create( + shelf_id, + project_id=project_id, + display_name=display_name, + data_dir=data_dir, + ) + ) + except InvalidAgentNameError as exc: + raise _BadRequest(str(exc)) from exc + status = 200 + self._send_json(status, result) + + def _handle_admin_shelf_archive(self, shelf_id: str, query: dict) -> None: + if not self._check_admin_token(): + return + expect_empty_raw = (query.get("expect_empty") or [None])[0] + expect_empty = ( + expect_empty_raw is not None and expect_empty_raw.lower() == "true" + ) + from .admin import ShelfNotFoundError, ShelfNotEmptyError # noqa: PLC0415 + try: + result = runner.run( + service.admin_shelf_archive( + shelf_id, + expect_empty=expect_empty, + data_dir=data_dir, + ) + ) + except ShelfNotFoundError as exc: + self._send_json(404, {"error": str(exc)}) + return + except ShelfNotEmptyError as exc: + self._send_json(409, {"error": str(exc)}) + return + self._send_json(200, result) + + def _handle_admin_shelf_unarchive(self, shelf_id: str) -> None: + if not self._check_admin_token(): + return + from .admin import ShelfNotFoundError # noqa: PLC0415 + try: + result = runner.run( + service.admin_shelf_unarchive(shelf_id, data_dir=data_dir) + ) + except ShelfNotFoundError as exc: + self._send_json(404, {"error": str(exc)}) + return + self._send_json(200, result) + + # ----- admin: A2A channel admin ---------------------------------- + + def _handle_admin_a2a_delete_channel(self) -> None: + if not self._check_admin_token(): + return + body = self._read_json_body() + channel = body.get("channel") + if not isinstance(channel, str) or not channel: + raise _BadRequest("'channel' (non-empty string) is required") + result = runner.run( + service.admin_a2a_delete_channel(channel, data_dir=data_dir) + ) + self._send_json(200, result) + + def _handle_admin_a2a_rename_channel(self) -> None: + if not self._check_admin_token(): + return + body = self._read_json_body() + from_channel = body.get("from") + to_channel = body.get("to") + if not isinstance(from_channel, str) or not from_channel: + raise _BadRequest("'from' (non-empty string) is required") + if not isinstance(to_channel, str) or not to_channel: + raise _BadRequest("'to' (non-empty string) is required") + try: + result = runner.run( + service.admin_a2a_rename_channel( + from_channel, to_channel, data_dir=data_dir + ) + ) + except ValueError as exc: + raise _BadRequest(str(exc)) from exc + self._send_json(200, result) + + def _handle_admin_a2a_supersede_message(self) -> None: + if not self._check_admin_token(): + return + body = self._read_json_body() + msg_id = body.get("id") + if msg_id is None: + raise _BadRequest("'id' (integer) is required") + try: + msg_id = int(msg_id) + except (TypeError, ValueError) as exc: + raise _BadRequest("'id' must be an integer") from exc + result = runner.run( + service.admin_a2a_supersede_message(msg_id, data_dir=data_dir) + ) + self._send_json(200, result) + return TaosmdHandler @@ -1254,6 +1427,10 @@ def serve(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, data_dir=None) -> "GET /a2a/channels, GET /a2a/members, " "POST /tasks, GET /tasks, GET /tasks/ready, GET /tasks/prime, " "POST /tasks/{id}, POST /tasks/{id}/edges, POST /tasks/{id}/edges/remove") + print("Admin (token required): POST /shelves, POST /shelves/{id}/archive, " + "POST /shelves/{id}/unarchive, " + "POST /a2a/admin/delete-channel, POST /a2a/admin/rename-channel, " + "POST /a2a/admin/supersede-message") try: httpd.serve_forever() except KeyboardInterrupt: diff --git a/taosmd/service.py b/taosmd/service.py index 028e4fa7..b0c2086a 100644 --- a/taosmd/service.py +++ b/taosmd/service.py @@ -298,6 +298,12 @@ async def a2a_send( return await remote.a2a_send(sender, body, thread=thread, reply_to=reply_to) stores = await _api._ensure_stores(data_dir) archive = stores["archive"] + # Redirect sends to renamed channels: if the target thread has been aliased + # to a new name, route the message to the canonical name instead. + if data_dir is not None: + from .admin import A2AAdminState # noqa: PLC0415 + _admin = A2AAdminState(data_dir) + thread = _admin.resolve_channel(thread) row_id = await archive.record( event_type=EVENT_A2A, data={"from": sender, "body": body, "thread": thread, "reply_to": reply_to}, @@ -334,12 +340,42 @@ async def a2a_feed( return await remote.a2a_feed(thread=thread, since=since, limit=limit) stores = await _api._ensure_stores(data_dir) archive = stores["archive"] - rows = await archive.query( - event_type=EVENT_A2A, - app_id=thread, - since=since, - limit=limit, - ) + + # Apply admin alias resolution: reads of a new channel name include history + # from the old name. Resolve thread through the alias map so callers + # querying the canonical name see both old and new messages. + resolved_thread = thread + alias_sources: list[str] = [] + if data_dir is not None: + from .admin import A2AAdminState # noqa: PLC0415 + _admin_state = A2AAdminState(data_dir) + _aliases = _admin_state.channel_aliases() + _deleted = _admin_state.deleted_channels() + _superseded = _admin_state.superseded_messages() + # Find all channel names that alias to thread (so we can include + # their history when querying the canonical name). + if thread is not None: + alias_sources = [k for k, v in _aliases.items() if v == thread] + else: + _deleted = set() + _superseded = set() + alias_sources = [] + + # Query with no thread filter when we need to merge history from aliases + if alias_sources and thread is not None: + rows_all = await archive.query(event_type=EVENT_A2A, since=since, limit=limit * 10) + rows = [ + r for r in rows_all + if (r.get("app_id") == thread or r.get("app_id") in alias_sources) + ] + rows = rows[:limit] + else: + rows = await archive.query( + event_type=EVENT_A2A, + app_id=thread, + since=since, + limit=limit, + ) # archive.query returns newest-first; A2A feed is displayed oldest-first. rows = list(reversed(rows)) result = [] @@ -348,12 +384,22 @@ async def a2a_feed( data = json.loads(row.get("data_json", "{}")) except (json.JSONDecodeError, TypeError): data = {} + # Skip admin-suppressed items + row_id = row["id"] + if row_id in _superseded: + continue + msg_thread = data.get("thread") or row.get("app_id") or "general" + if msg_thread in _deleted: + continue + # Skip admin-action rows (they have no "from" field) + if data.get("admin_action"): + continue result.append({ - "id": row["id"], + "id": row_id, "ts": row["timestamp"], "from": data.get("from"), "body": data.get("body"), - "thread": data.get("thread"), + "thread": msg_thread, "reply_to": data.get("reply_to"), }) return result @@ -381,13 +427,35 @@ async def a2a_channels(*, data_dir=None) -> list[dict]: archive = stores["archive"] rows = await archive.query(event_type=EVENT_A2A, limit=100_000) + # Load admin state once for filtering + deleted: set[str] = set() + aliases: dict[str, str] = {} + superseded: set[int] = set() + if data_dir is not None: + from .admin import A2AAdminState # noqa: PLC0415 + _admin = A2AAdminState(data_dir) + deleted = _admin.deleted_channels() + aliases = _admin.channel_aliases() + superseded = _admin.superseded_messages() + channels: dict[str, dict] = {} for row in rows: try: data = json.loads(row.get("data_json", "{}")) except (json.JSONDecodeError, TypeError): data = {} + # Skip admin-action rows and superseded messages + if data.get("admin_action"): + continue + if row["id"] in superseded: + continue thread = data.get("thread") or row.get("app_id") or "general" + # Redirect aliased channels to their canonical name + if thread in aliases: + thread = aliases[thread] + # Skip deleted channels + if thread in deleted: + continue sender = data.get("from") or "" ts = row.get("timestamp", 0.0) @@ -610,7 +678,94 @@ async def task_remove_edge( ) +# --------------------------------------------------------------------------- +# Admin surface service wrappers +# --------------------------------------------------------------------------- + +async def admin_shelf_create( + shelf_id: str, + *, + project_id: str | None = None, + display_name: str | None = None, + data_dir=None, +) -> dict: + """Create or return an existing shelf. Returns ``{"shelf": {...}, "created": bool}``.""" + if data_dir is None: + stores = await _api._ensure_stores(data_dir) + data_dir = stores["data_dir"] + from .admin import shelf_create # noqa: PLC0415 + return await shelf_create( + shelf_id, project_id=project_id, display_name=display_name, data_dir=data_dir, + ) + + +async def admin_shelf_archive( + shelf_id: str, + *, + expect_empty: bool = False, + data_dir=None, +) -> dict: + """Archive a shelf, soft-hiding its vector rows.""" + stores = await _api._ensure_stores(data_dir) + if data_dir is None: + data_dir = stores["data_dir"] + from .admin import shelf_archive # noqa: PLC0415 + return await shelf_archive( + shelf_id, expect_empty=expect_empty, data_dir=data_dir, stores=stores, + ) + + +async def admin_shelf_unarchive( + shelf_id: str, + *, + data_dir=None, +) -> dict: + """Unarchive a shelf, restoring only shelf-archive-hidden rows.""" + stores = await _api._ensure_stores(data_dir) + if data_dir is None: + data_dir = stores["data_dir"] + from .admin import shelf_unarchive # noqa: PLC0415 + return await shelf_unarchive(shelf_id, data_dir=data_dir, stores=stores) + + +async def admin_a2a_delete_channel(channel: str, *, data_dir=None) -> dict: + """Soft-delete an A2A channel.""" + stores = await _api._ensure_stores(data_dir) + if data_dir is None: + data_dir = stores["data_dir"] + from .admin import a2a_admin_delete_channel # noqa: PLC0415 + return await a2a_admin_delete_channel(channel, data_dir=data_dir, stores=stores) + + +async def admin_a2a_rename_channel( + from_channel: str, + to_channel: str, + *, + data_dir=None, +) -> dict: + """Rename an A2A channel via alias.""" + stores = await _api._ensure_stores(data_dir) + if data_dir is None: + data_dir = stores["data_dir"] + from .admin import a2a_admin_rename_channel # noqa: PLC0415 + return await a2a_admin_rename_channel( + from_channel, to_channel, data_dir=data_dir, stores=stores + ) + + +async def admin_a2a_supersede_message(msg_id: int, *, data_dir=None) -> dict: + """Supersede (hide) a single A2A message from feeds.""" + stores = await _api._ensure_stores(data_dir) + if data_dir is None: + data_dir = stores["data_dir"] + from .admin import a2a_admin_supersede_message # noqa: PLC0415 + return await a2a_admin_supersede_message(msg_id, data_dir=data_dir, stores=stores) + + __all__ = ["ingest", "search", "pending_list", "pending_resolve", "reconcile", "stats", "supersede", "a2a_send", "a2a_feed", "a2a_channels", "a2a_members", "task_create", "task_list", "task_ready", "task_prime", - "task_update", "task_add_edge", "task_remove_edge"] + "task_update", "task_add_edge", "task_remove_edge", + "admin_shelf_create", "admin_shelf_archive", "admin_shelf_unarchive", + "admin_a2a_delete_channel", "admin_a2a_rename_channel", + "admin_a2a_supersede_message"] diff --git a/tests/test_admin_surface.py b/tests/test_admin_surface.py new file mode 100644 index 00000000..67cb1e38 --- /dev/null +++ b/tests/test_admin_surface.py @@ -0,0 +1,545 @@ +"""Tests for the admin surface: shelf lifecycle and A2A channel admin. + +The admin endpoints require a configured server token and FAIL CLOSED when +none is set (403). These tests exercise both the no-token-403 path and the +correct-token happy path. + +Uses the live_server fixture pattern from tests/test_http_server.py with a +variant that sets a server token via env or config. +""" + +from __future__ import annotations + +import json +import os +import threading +import urllib.error +import urllib.request +from pathlib import Path + +import pytest + +from taosmd import api as taosmd_api +from taosmd import config as taosmd_config +from taosmd import http_server + + +# --------------------------------------------------------------------------- +# HTTP helpers +# --------------------------------------------------------------------------- + +_TOKEN = "test-admin-token-abc123" + + +def _patch_embedder(stores: dict) -> None: + """Deterministic 8-dim hash embedder; same as test_http_server.py.""" + vmem = stores["vector"] + + async def _fake_embed(text: str, task: str = "search_document") -> list[float]: + h = hash(text) & 0xFFFFFFFF + return [((h >> (i * 4)) & 0xFF) / 255.0 for i in range(8)] + + vmem.embed = _fake_embed # type: ignore[assignment] + + +def _post(url: str, payload, token: str | None = None) -> tuple[int, dict]: + data = json.dumps(payload).encode() + headers = {"Content-Type": "application/json"} + if token: + headers["Authorization"] = f"Bearer {token}" + req = urllib.request.Request(url, data=data, headers=headers, method="POST") + return _send(req) + + +def _get(url: str, token: str | None = None) -> tuple[int, dict]: + headers = {} + if token: + headers["Authorization"] = f"Bearer {token}" + return _send(urllib.request.Request(url, headers=headers, method="GET")) + + +def _send(req) -> tuple[int, dict]: + try: + with urllib.request.urlopen(req, timeout=10) as resp: + return resp.status, json.loads(resp.read().decode()) + except urllib.error.HTTPError as exc: + return exc.code, json.loads(exc.read().decode()) + + +# --------------------------------------------------------------------------- +# Fixtures: server with token, server without token +# --------------------------------------------------------------------------- + +def _make_token_server(tmp_path, monkeypatch): + """Helper that creates a token-gated server and returns (url, httpd, data_dir).""" + data_dir = tmp_path / "taosmd-data" + data_dir.mkdir() + monkeypatch.setattr(taosmd_api, "_stores_cache", {}) + monkeypatch.setenv("TAOSMD_TOKEN", _TOKEN) + + httpd = http_server.make_server("127.0.0.1", 0, data_dir=str(data_dir)) + stores = httpd.service_loop.run(taosmd_api._ensure_stores(str(data_dir))) + _patch_embedder(stores) + + host, port = httpd.server_address[:2] + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + return f"http://{host}:{port}", httpd, str(data_dir), thread + + +def _teardown_server(httpd, thread): + httpd.shutdown() + httpd.server_close() + thread.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() + + +@pytest.fixture +def live_server_with_token(tmp_path, monkeypatch): + """Live server with a configured bearer token for admin tests.""" + base, httpd, data_dir, thread = _make_token_server(tmp_path, monkeypatch) + try: + yield base + finally: + _teardown_server(httpd, thread) + + +@pytest.fixture +def live_server_with_token_and_loop(tmp_path, monkeypatch): + """Live server with token that also yields the httpd (for service loop access).""" + base, httpd, data_dir, thread = _make_token_server(tmp_path, monkeypatch) + try: + yield base, httpd + finally: + _teardown_server(httpd, thread) + + +@pytest.fixture +def live_server_no_token(tmp_path, monkeypatch): + """Live server with NO configured token (admin endpoints should return 403).""" + data_dir = tmp_path / "taosmd-data" + data_dir.mkdir() + monkeypatch.setattr(taosmd_api, "_stores_cache", {}) + monkeypatch.delenv("TAOSMD_TOKEN", raising=False) + + httpd = http_server.make_server("127.0.0.1", 0, data_dir=str(data_dir)) + stores = httpd.service_loop.run(taosmd_api._ensure_stores(str(data_dir))) + _patch_embedder(stores) + + host, port = httpd.server_address[:2] + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + try: + yield f"http://{host}:{port}" + finally: + httpd.shutdown() + httpd.server_close() + thread.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() + + +# --------------------------------------------------------------------------- +# Auth rule: no-token-configured returns 403 on every admin route +# --------------------------------------------------------------------------- + +_ADMIN_ROUTES = [ + ("/shelves", {"shelf_id": "testshelf"}), + ("/shelves/testshelf/archive", {}), + ("/shelves/testshelf/unarchive", {}), + ("/a2a/admin/delete-channel", {"channel": "general"}), + ("/a2a/admin/rename-channel", {"from": "old", "to": "new"}), + ("/a2a/admin/supersede-message", {"id": 1}), +] + + +@pytest.mark.parametrize("path,payload", _ADMIN_ROUTES) +def test_no_token_configured_returns_403(live_server_no_token, path, payload): + """When no server token is configured all admin endpoints return 403.""" + status, body = _post(f"{live_server_no_token}{path}", payload, token=None) + assert status == 403, f"expected 403 for {path}, got {status}: {body}" + assert "admin surface requires a configured server token" in body.get("error", "") + + +@pytest.mark.parametrize("path,payload", _ADMIN_ROUTES) +def test_wrong_token_returns_401(live_server_with_token, path, payload): + """When a token IS configured, a wrong token returns 401.""" + status, body = _post( + f"{live_server_with_token}{path}", payload, token="wrong-token" + ) + assert status == 401, f"expected 401 for {path} with wrong token, got {status}: {body}" + + +# --------------------------------------------------------------------------- +# Shelf create: idempotence +# --------------------------------------------------------------------------- + +def test_shelf_create_new(live_server_with_token): + status, body = _post( + f"{live_server_with_token}/shelves", + {"shelf_id": "myshelf", "display_name": "My Shelf"}, + token=_TOKEN, + ) + assert status == 200, body + assert body["created"] is True + assert body["shelf"]["name"] == "myshelf" + assert body["shelf"]["display_name"] == "My Shelf" + + +def test_shelf_create_idempotent(live_server_with_token): + _post( + f"{live_server_with_token}/shelves", + {"shelf_id": "myshelf2"}, + token=_TOKEN, + ) + status, body = _post( + f"{live_server_with_token}/shelves", + {"shelf_id": "myshelf2"}, + token=_TOKEN, + ) + assert status == 200, body + assert body["created"] is False + assert body["shelf"]["name"] == "myshelf2" + + +def test_shelf_create_with_project_id(live_server_with_token): + status, body = _post( + f"{live_server_with_token}/shelves", + {"shelf_id": "proj-shelf", "project_id": "proj-xyz"}, + token=_TOKEN, + ) + assert status == 200, body + assert body["created"] is True + assert body["shelf"].get("project_id") == "proj-xyz" + + +def test_shelf_create_invalid_id_returns_400(live_server_with_token): + status, body = _post( + f"{live_server_with_token}/shelves", + {"shelf_id": "InvalidCaps"}, + token=_TOKEN, + ) + assert status == 400, body + assert "shelf_id" in body.get("error", "").lower() or "must match" in body.get("error", "") + + +def test_shelf_create_missing_id_returns_400(live_server_with_token): + status, body = _post( + f"{live_server_with_token}/shelves", + {}, + token=_TOKEN, + ) + assert status == 400 + assert "shelf_id" in body.get("error", "") + + +# --------------------------------------------------------------------------- +# Shelf archive / unarchive: hides and restores rows +# --------------------------------------------------------------------------- + +def _ingest(base_url: str, text: str, agent: str, token: str) -> None: + status, body = _post( + f"{base_url}/ingest", + {"text": text, "agent": agent}, + token=token, + ) + assert status == 200, f"ingest failed: {body}" + + +def _search(base_url: str, query: str, agent: str, token: str) -> list: + # Use mode=bm25 so we only hit the vector store BM25 index and do not get + # archive FTS results, which come from the zero-loss archive and are NOT + # suppressed by shelf-archive (the spec only hides vector rows). + status, body = _post( + f"{base_url}/search", + {"query": query, "agent": agent, "limit": 10, "mode": "bm25"}, + token=token, + ) + assert status == 200, f"search failed: {body}" + return body.get("hits", []) + + +def test_archive_hides_rows_from_search(live_server_with_token): + base = live_server_with_token + shelf = "archivetest" + + # Create shelf and ingest some content + _post(f"{base}/shelves", {"shelf_id": shelf}, token=_TOKEN) + _ingest(base, "Unique text for archive test alpha", shelf, _TOKEN) + _ingest(base, "Another unique text for archive test beta", shelf, _TOKEN) + + # Rows should be visible before archive + hits_before = _search(base, "unique text for archive test", shelf, _TOKEN) + assert hits_before, "expected hits before archive" + + # Archive the shelf + status, body = _post(f"{base}/shelves/{shelf}/archive", {}, token=_TOKEN) + assert status == 200, body + assert body["archived"] is True + assert body["rows_hidden"] >= 2 + + # Rows should now be hidden from search + hits_after = _search(base, "unique text for archive test", shelf, _TOKEN) + assert not hits_after, f"expected no hits after archive, got {hits_after}" + + +def test_unarchive_restores_rows(live_server_with_token): + base = live_server_with_token + shelf = "restoretest" + + _post(f"{base}/shelves", {"shelf_id": shelf}, token=_TOKEN) + _ingest(base, "Restore test content gamma", shelf, _TOKEN) + + # Archive + status, body = _post(f"{base}/shelves/{shelf}/archive", {}, token=_TOKEN) + assert status == 200, body + assert body["rows_hidden"] >= 1 + + # Confirm hidden + hits = _search(base, "Restore test content gamma", shelf, _TOKEN) + assert not hits, "expected no hits after archive" + + # Unarchive + status, body = _post(f"{base}/shelves/{shelf}/unarchive", {}, token=_TOKEN) + assert status == 200, body + assert body["archived"] is False + assert body["rows_restored"] >= 1 + + # Rows should be visible again + hits = _search(base, "Restore test content gamma", shelf, _TOKEN) + assert hits, "expected hits after unarchive" + assert "Restore test content gamma" in hits[0]["text"] + + +def test_unarchive_does_not_resurrect_non_shelf_superseded_rows( + live_server_with_token_and_loop, +): + """Rows superseded for non-archive reasons are not restored by unarchive. + + Strategy: create two rows, manually supersede one via the server's service + loop (simulating a correction-supersede), then archive+unarchive the shelf + and verify only the shelf-archived row is restored. + """ + base, httpd = live_server_with_token_and_loop + shelf = "supersede-test" + + _post(f"{base}/shelves", {"shelf_id": shelf}, token=_TOKEN) + _ingest(base, "Row superseded for contradiction reason", shelf, _TOKEN) + _ingest(base, "Row to be shelf-archived delta", shelf, _TOKEN) + + # Use the server's own service loop to supersede the first row, simulating + # a correction-supersede (not a shelf-archive). The service loop owns the + # SQLite connection so we must run async operations there. + stores_cache = taosmd_api._stores_cache + data_dir_key = next(iter(stores_cache)) + stores = stores_cache[data_dir_key] + vmem = stores["vector"] + + async def _do_supersede(): + await vmem.supersede_matching("Row superseded for contradiction reason") + + httpd.service_loop.run(_do_supersede()) + + # Now archive the shelf + status, body = _post(f"{base}/shelves/{shelf}/archive", {}, token=_TOKEN) + assert status == 200, body + rows_hidden_by_archive = body["rows_hidden"] + # Only the "delta" row should have been hidden (the other was already superseded) + assert rows_hidden_by_archive == 1 + + # Unarchive + status, body = _post(f"{base}/shelves/{shelf}/unarchive", {}, token=_TOKEN) + assert status == 200, body + # Only the shelf-archive-hidden row should be restored + assert body["rows_restored"] == 1 + + # The "contradiction" row should remain hidden (it was superseded, not shelf-archived). + # Check by text rather than relying on BM25 not returning the other row. + hits = _search(base, "Row to be shelf-archived delta", shelf, _TOKEN) + hit_texts = [h["text"] for h in hits] + assert "Row superseded for contradiction reason" not in hit_texts, ( + "contradiction-superseded row should remain hidden after unarchive" + ) + + # The "delta" row should now be visible + assert any("delta" in t for t in hit_texts), ( + f"shelf-archived row should be restored after unarchive; hits: {hit_texts}" + ) + + +def test_archive_no_op_when_already_archived(live_server_with_token): + base = live_server_with_token + shelf = "noop-archive" + _post(f"{base}/shelves", {"shelf_id": shelf}, token=_TOKEN) + _post(f"{base}/shelves/{shelf}/archive", {}, token=_TOKEN) + + # Second archive is a no-op + status, body = _post(f"{base}/shelves/{shelf}/archive", {}, token=_TOKEN) + assert status == 200, body + assert body["archived"] is True + assert body["rows_hidden"] == 0 + + +def test_archive_expect_empty_409_when_rows_exist(live_server_with_token): + base = live_server_with_token + shelf = "expect-empty-test" + _post(f"{base}/shelves", {"shelf_id": shelf}, token=_TOKEN) + _ingest(base, "Content that prevents empty archive", shelf, _TOKEN) + + status, body = _post( + f"{base}/shelves/{shelf}/archive?expect_empty=true", {}, token=_TOKEN + ) + assert status == 409, body + assert "active row" in body.get("error", "").lower() or "expect_empty" in body.get("error", "").lower() + + +def test_archive_expect_empty_succeeds_when_no_rows(live_server_with_token): + base = live_server_with_token + shelf = "empty-shelf-test" + _post(f"{base}/shelves", {"shelf_id": shelf}, token=_TOKEN) + + status, body = _post( + f"{base}/shelves/{shelf}/archive?expect_empty=true", {}, token=_TOKEN + ) + assert status == 200, body + assert body["archived"] is True + assert body["rows_hidden"] == 0 + + +# --------------------------------------------------------------------------- +# A2A admin: delete channel +# --------------------------------------------------------------------------- + +def test_a2a_delete_channel_hides_from_channels(live_server_with_token): + base = live_server_with_token + + # Send a message to a channel + _post(f"{base}/a2a/send", {"from": "alice", "body": "hello", "thread": "to-delete"}, token=_TOKEN) + + # Confirm it's visible + status, body = _get(f"{base}/a2a/channels", token=_TOKEN) + assert status == 200 + channels = {c["channel"] for c in body["channels"]} + assert "to-delete" in channels + + # Delete the channel + status, body = _post( + f"{base}/a2a/admin/delete-channel", + {"channel": "to-delete"}, + token=_TOKEN, + ) + assert status == 200, body + assert body["deleted"] is True + + # Channel should no longer appear in /a2a/channels + status, body = _get(f"{base}/a2a/channels", token=_TOKEN) + assert status == 200 + channels_after = {c["channel"] for c in body["channels"]} + assert "to-delete" not in channels_after + + +def test_a2a_delete_channel_hides_from_messages(live_server_with_token): + base = live_server_with_token + + _post(f"{base}/a2a/send", {"from": "bob", "body": "secret message", "thread": "secret-chan"}, token=_TOKEN) + _post(f"{base}/a2a/admin/delete-channel", {"channel": "secret-chan"}, token=_TOKEN) + + # Messages on the deleted channel should not appear + status, body = _get(f"{base}/a2a/messages?thread=secret-chan", token=_TOKEN) + assert status == 200 + assert body.get("messages") == [] or not any( + m.get("thread") == "secret-chan" for m in body.get("messages", []) + ) + + +# --------------------------------------------------------------------------- +# A2A admin: rename channel +# --------------------------------------------------------------------------- + +def test_a2a_rename_redirects_sends(live_server_with_token): + base = live_server_with_token + + # Send some old-name history + _post(f"{base}/a2a/send", {"from": "alice", "body": "old msg", "thread": "old-name"}, token=_TOKEN) + + # Rename old-name -> new-name + status, body = _post( + f"{base}/a2a/admin/rename-channel", + {"from": "old-name", "to": "new-name"}, + token=_TOKEN, + ) + assert status == 200, body + assert body["renamed"] is True + + # Send a new message to old-name; it should be redirected to new-name + _post(f"{base}/a2a/send", {"from": "bob", "body": "redirected msg", "thread": "old-name"}, token=_TOKEN) + + # new-name should contain both the old history and the redirected message + status, body = _get(f"{base}/a2a/messages?thread=new-name", token=_TOKEN) + assert status == 200 + texts = [m.get("body") for m in body.get("messages", [])] + assert "old msg" in texts, f"old history should appear in new-name: {texts}" + assert "redirected msg" in texts, f"redirected send should appear in new-name: {texts}" + + +def test_a2a_rename_merges_old_history_into_new(live_server_with_token): + """Reading the new channel name includes messages originally in the old name.""" + base = live_server_with_token + + _post(f"{base}/a2a/send", {"from": "x", "body": "history item", "thread": "alpha"}, token=_TOKEN) + + _post(f"{base}/a2a/admin/rename-channel", {"from": "alpha", "to": "beta"}, token=_TOKEN) + + status, body = _get(f"{base}/a2a/messages?thread=beta", token=_TOKEN) + assert status == 200 + texts = [m.get("body") for m in body.get("messages", [])] + assert "history item" in texts + + +# --------------------------------------------------------------------------- +# A2A admin: supersede message +# --------------------------------------------------------------------------- + +def test_a2a_supersede_message_hides_it(live_server_with_token): + base = live_server_with_token + + status, receipt = _post( + f"{base}/a2a/send", + {"from": "alice", "body": "message to suppress", "thread": "gen"}, + token=_TOKEN, + ) + assert status == 200 + msg_id = receipt["id"] + + # Confirm it's visible + status, body = _get(f"{base}/a2a/messages?thread=gen", token=_TOKEN) + ids_before = [m["id"] for m in body.get("messages", [])] + assert msg_id in ids_before + + # Supersede it + status, body = _post( + f"{base}/a2a/admin/supersede-message", + {"id": msg_id}, + token=_TOKEN, + ) + assert status == 200, body + assert body["superseded"] is True + assert body["id"] == msg_id + + # It should no longer appear in the feed + status, body = _get(f"{base}/a2a/messages?thread=gen", token=_TOKEN) + ids_after = [m["id"] for m in body.get("messages", [])] + assert msg_id not in ids_after