diff --git a/taosmd/archive.py b/taosmd/archive.py index 69bf452..b7c777d 100644 --- a/taosmd/archive.py +++ b/taosmd/archive.py @@ -58,6 +58,8 @@ agent_name TEXT, app_id TEXT, project TEXT, + source TEXT, + source_id TEXT, summary TEXT NOT NULL DEFAULT '', file_path TEXT NOT NULL, line_number INTEGER NOT NULL, @@ -67,6 +69,9 @@ CREATE INDEX IF NOT EXISTS idx_archive_type ON archive_index(event_type); CREATE INDEX IF NOT EXISTS idx_archive_agent ON archive_index(agent_name); CREATE INDEX IF NOT EXISTS idx_archive_app ON archive_index(app_id); +CREATE UNIQUE INDEX IF NOT EXISTS idx_archive_source_uid +ON archive_index (source, source_id) +WHERE source IS NOT NULL AND source_id IS NOT NULL; CREATE TABLE IF NOT EXISTS archive_settings ( key TEXT PRIMARY KEY, @@ -179,8 +184,18 @@ async def record( app_id: str | None = None, summary: str = "", project: str | None = None, + source: str | None = None, + source_id: str | None = None, + timestamp: float | None = None, ) -> int: - """Record an event to the archive. Returns the index row ID.""" + """Record an event to the archive. Returns the index row ID. + + ``timestamp`` overrides the wall-clock time when provided (used by the + A2A batch importer to preserve historical timestamps, #211). ``source`` + and ``source_id`` tag imported A2A messages for idempotent re-import; + when both are non-null the unique index ``idx_archive_source_uid`` + enforces (source, source_id) uniqueness. + """ # Skip user activity events if tracking is disabled if event_type in USER_ACTIVITY_EVENTS and not self._user_tracking_enabled: return -1 @@ -192,13 +207,15 @@ async def record( if key in data and isinstance(data[key], str): data[key], _ = redact_secrets(data[key]) - ts = time.time() + ts = timestamp if timestamp is not None else time.time() event = { "timestamp": ts, "event_type": event_type, "agent_name": agent_name, "app_id": app_id, "project": project, + "source": source, + "source_id": source_id, "summary": summary, "data": data, } @@ -232,9 +249,11 @@ async def record( # Index for fast lookup cursor = self._conn.execute( """INSERT INTO archive_index - (timestamp, event_type, agent_name, app_id, project, summary, file_path, line_number, data_json) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", - (ts, event_type, agent_name, app_id, project, summary, file_path, line_count, json.dumps(data, default=str)), + (timestamp, event_type, agent_name, app_id, project, source, source_id, + summary, file_path, line_number, data_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (ts, event_type, agent_name, app_id, project, source, source_id, + summary, file_path, line_count, json.dumps(data, default=str)), ) # Index in FTS for full-text search @@ -342,6 +361,35 @@ async def get_event(self, event_id: int) -> dict | None: result["data"] = {} return result + async def find_source_ids( + self, source: str, event_type: str = EVENT_A2A + ) -> dict[str, int]: + """Return ``{source_id: archive_id}`` for all events tagged with ``source``. + + Backs idempotent A2A batch import (#211): lets the importer skip rows whose + ``(source, source_id)`` already exists and resolve ``reply_to_source_id`` + to the archive row id of the originally imported message. Only rows with a + non-null ``source_id`` are considered. + """ + rows = self._conn.execute( + "SELECT source_id, id FROM archive_index " + "WHERE source = ? AND source_id IS NOT NULL AND event_type = ?", + (source, event_type), + ).fetchall() + return {row["source_id"]: row["id"] for row in rows} + + async def find_reply_target( + self, source: str, source_id: str, event_type: str = EVENT_A2A + ) -> int | None: + """Return the archive row id for a ``(source, source_id)`` pair, or None.""" + row = self._conn.execute( + "SELECT id FROM archive_index " + "WHERE source = ? AND source_id = ? AND event_type = ? " + "ORDER BY id ASC LIMIT 1", + (source, source_id, event_type), + ).fetchone() + return row["id"] if row else None + async def count( self, event_type: str | None = None, diff --git a/taosmd/capabilities.py b/taosmd/capabilities.py index 50b4d29..5c85414 100644 --- a/taosmd/capabilities.py +++ b/taosmd/capabilities.py @@ -84,9 +84,10 @@ class CapabilityProbe: CapabilityProbe( name="a2a.v1", module="taosmd.service", - symbols=("a2a_send", "a2a_feed", "a2a_channels", "a2a_members"), + symbols=("a2a_send", "a2a_import", "a2a_feed", "a2a_channels", "a2a_members"), route_markers=( '"/a2a/send"', + '"/a2a/import"', '"/a2a/messages"', '"/a2a/stream"', '"/a2a/channels"', diff --git a/taosmd/http_server.py b/taosmd/http_server.py index 59acb83..a6b3c09 100644 --- a/taosmd/http_server.py +++ b/taosmd/http_server.py @@ -102,8 +102,10 @@ ``GET /pending?agent=`` -> ``{"pending": [...]}`` ``POST /pending/resolve`` ``{"id", "decision", "note"?}`` -> resolve result ``POST /a2a/send`` ``{"from", "body", "thread"?, "reply_to"?, "refs"?, "blocks"?}`` -> send receipt - ``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 + ``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 +``POST /a2a/import`` (admin) ``{"source": str, "defer_index"?: bool, "messages": [{"from": str, "thread": str, "body": str, "ts": float, "source_id": str, "reply_to_source_id"?: str|null, "blocks"?: [...], "refs"?: [...]}]}`` -> ``{"imported": n, "skipped": n, "first_id": int|null, "last_id": int|null}`` + Batch-import historical chat history onto the bus (taOSmd #211 Q3a). Idempotent on ``(source, source_id)``: re-runs skip existing rows. ``ts`` is preserved as the archive timestamp, not import time. ``reply_to_source_id`` resolves to the imported archive id of the referenced message (same source); unresolvable -> 400, no write. ``defer_index=true`` archives only and defers vector embedding; run ``taosmd reindex --agent `` afterwards to back-fill vectors from the archive. ``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/channels`` -> ``{"channels": [...]}`` @@ -750,6 +752,7 @@ def _is_admin_route(method: str, path: str) -> bool: "/a2a/admin/delete-channel", "/a2a/admin/rename-channel", "/a2a/admin/supersede-message", + "/a2a/import", ) def _check_admin_token(self) -> bool: @@ -947,6 +950,8 @@ def _dispatch(self, method: str) -> None: self._handle_pending_resolve() elif method == "POST" and path == "/a2a/send": self._handle_a2a_send() + elif method == "POST" and path == "/a2a/import": + self._handle_a2a_import() elif method == "GET" and path == "/a2a/channels": self._handle_a2a_channels() elif method == "GET" and path == "/a2a/members": @@ -1501,6 +1506,26 @@ def _handle_a2a_send(self) -> None: ) self._send_json(200, result) + def _handle_a2a_import(self) -> None: + if not self._check_admin_token(): + return + body = self._read_json_body() + source = body.get("source") + defer_index = body.get("defer_index", False) + messages = body.get("messages") + if not isinstance(source, str) or not source: + raise _BadRequest("'source' (non-empty string) is required") + if not isinstance(messages, list) or not messages: + raise _BadRequest("'messages' (non-empty list) is required") + if not isinstance(defer_index, bool): + raise _BadRequest("'defer_index' must be a boolean when provided") + result = runner.run( + service.a2a_import( + source, messages, defer_index=defer_index, data_dir=data_dir, + ) + ) + self._send_json(200, result) + def _handle_a2a_messages(self, qs: dict) -> None: thread = (qs.get("thread") or [None])[0] since_raw = (qs.get("since") or [None])[0] @@ -2104,7 +2129,7 @@ def serve(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, data_dir=None) -> print("Endpoints: GET /health, GET /version, POST /ingest, POST /ingest/batch, GET|POST /search, " "GET /projects, GET /shelves, " "GET /pending, POST /pending/resolve, " - "POST /a2a/send, GET /a2a/messages, GET /a2a/stream, " + "POST /a2a/send, POST /a2a/import, GET /a2a/messages, GET /a2a/stream, " "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, " @@ -2116,7 +2141,7 @@ def serve(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, data_dir=None) -> "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") + "POST /a2a/admin/supersede-message, POST /a2a/import") try: httpd.serve_forever() except KeyboardInterrupt: diff --git a/taosmd/migrations.py b/taosmd/migrations.py index 8b9f5fc..41830d8 100644 --- a/taosmd/migrations.py +++ b/taosmd/migrations.py @@ -199,6 +199,24 @@ def _archive_index_project(conn: sqlite3.Connection) -> None: add_column(conn, "archive_index", "project", "TEXT") +def _archive_index_source_uid(conn: sqlite3.Connection) -> None: + """Add source/source_id columns for idempotent A2A batch import (#211). + + ``source`` and ``source_id`` tag each imported A2A message with the + external origin and its stable per-source id, so a re-import can skip + rows already present without relying on the JSON payload alone. The + partial unique index enforces (source, source_id) uniqueness at the + database level as a safety net against duplicate writes. + """ + add_column(conn, "archive_index", "source", "TEXT") + add_column(conn, "archive_index", "source_id", "TEXT") + conn.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_archive_source_uid " + "ON archive_index (source, source_id) " + "WHERE source IS NOT NULL AND source_id IS NOT NULL" + ) + + _ARCHIVE_INDEX: tuple[Migration, ...] = ( Migration( 1, "archive_index_baseline", _archive_index_baseline, @@ -208,6 +226,13 @@ def _archive_index_project(conn: sqlite3.Connection) -> None: 2, "archive_index_project", _archive_index_project, lambda c: has_column(c, "archive_index", "project"), ), + Migration( + 3, "archive_index_source_uid", _archive_index_source_uid, + lambda c: ( + has_column(c, "archive_index", "source") + and has_column(c, "archive_index", "source_id") + ), + ), ) diff --git a/taosmd/remote.py b/taosmd/remote.py index 6df7e3d..84794dd 100644 --- a/taosmd/remote.py +++ b/taosmd/remote.py @@ -228,6 +228,24 @@ async def a2a_send( payload["blocks"] = blocks return await self._run("POST", "/a2a/send", payload) + async def a2a_import( + self, + source: str, + messages: list[dict], + *, + defer_index: bool = False, + **_opts, + ) -> dict: + """POST /a2a/import: admin batch-import historical messages onto the remote bus. + + Returns ``{"imported", "skipped", "first_id", "last_id"}`` (taOSmd #211 Q3a). + Uses the client's bearer token, which must be an admin token. + """ + payload: dict = {"source": source, "messages": messages} + if defer_index: + payload["defer_index"] = True + return await self._run("POST", "/a2a/import", payload) + async def a2a_feed( self, *, diff --git a/taosmd/service.py b/taosmd/service.py index cf3e912..dd5abde 100644 --- a/taosmd/service.py +++ b/taosmd/service.py @@ -27,6 +27,7 @@ import json import logging +import math from . import api as _api from . import config as _config @@ -625,6 +626,183 @@ async def a2a_members(*, channel: str, data_dir=None) -> list[str]: return sorted(members) +async def a2a_import( + source: str, + messages: list[dict], + *, + defer_index: bool = False, + data_dir=None, +) -> dict: + """Admin batch-import historical chat messages onto the A2A bus. + + Writes each message as an append-only :data:`~taosmd.archive.EVENT_A2A` + archive event, preserving the caller-supplied historical ``ts`` rather than + the wall-clock import time (taOSmd #211 Q3a). Designed for the taOS chat + history migration: existing conversations become ordinary bus messages with + their original timestamps intact, so thread ordering survives in the archive. + + The whole batch is validated and pre-checked *before any write*; a missing + required field or an unresolvable ``reply_to_source_id`` refuses the entire + batch with a ``ValueError`` (which the HTTP layer maps to 400) and leaves the + archive untouched (fail-loud, zero partial writes). + + Rules: + + * **Idempotent** -- uniqueness is on ``(source, source_id)``. Messages whose + ``source_id`` already exists for this ``source`` (in the archive or earlier + in the same batch) are skipped and counted in ``skipped``, never duplicated. + * **TS preserved** -- the archive row carries the historical ``ts``. Within a + batch, rows are inserted in ascending ``(ts, position-in-batch)`` order so + the auto-increment row-id ordering matches ts ordering (thread ordering + survives equal timestamps via the stable sort). + * **reply_to resolution** -- ``reply_to_source_id`` is resolved to the archive + row id of the matching message (same ``source``); an unresolvable id + refuses the batch. + * **defer_index** -- when ``True``, messages are archived but NOT embedded + into the vector store. Run ``taosmd reindex --agent `` afterwards to + back-fill vector entries from the archive. When ``False`` (default) each + message body is embedded immediately so imported history is searchable. + + Returns ``{"imported": n, "skipped": n, "first_id": int|None, "last_id": int|None}``. + ``first_id`` / ``last_id`` are the archive row ids of the first and last newly + imported messages in ts order (both ``None`` when every message was skipped). + """ + if not isinstance(source, str) or not source: + raise ValueError("source (non-empty string) is required") + if not isinstance(messages, list): + raise ValueError("messages must be a list") + if not messages: + raise ValueError("messages must be a non-empty list") + + # --- Fail-loud validation: every field checked before any write -------- + seen_source_ids: set[str] = set() + for i, msg in enumerate(messages): + if not isinstance(msg, dict): + raise ValueError(f"messages[{i}] must be an object") + for field in ("from", "thread", "body", "source_id"): + val = msg.get(field) + if not isinstance(val, str) or not val: + raise ValueError( + f"messages[{i}].{field} (non-empty string) is required" + ) + ts = msg.get("ts") + if not isinstance(ts, (int, float)) or isinstance(ts, bool) or not math.isfinite(ts): + raise ValueError(f"messages[{i}].ts (finite float) is required") + rti = msg.get("reply_to_source_id") + if rti is not None and not isinstance(rti, str): + raise ValueError(f"messages[{i}].reply_to_source_id must be a string or null") + sid = msg["source_id"] + if sid in seen_source_ids: + raise ValueError( + f"messages[{i}].source_id {sid!r} is duplicated within the batch" + ) + seen_source_ids.add(sid) + + # --- Idempotency + reply_to pre-check (archive is source of truth) --- + stores = await _api._ensure_stores(data_dir) + archive = stores["archive"] + existing = await archive.find_source_ids(source) + # Every reply_to_source_id must resolve to an already-imported message + # (in the archive, or earlier in this same batch by source_id presence). + batch_ids = {msg["source_id"] for msg in messages} + for i, msg in enumerate(messages): + rti = msg.get("reply_to_source_id") + if rti is not None and rti not in existing and rti not in batch_ids: + raise ValueError( + f"messages[{i}].reply_to_source_id {rti!r} does not match any " + f"imported message in source {source!r}" + ) + + # --- Insert in ts order so auto-increment id order == ts order ---------- + # Stable sort on (ts, original position): equal timestamps keep their + # in-batch order, so thread ordering survives. + order = sorted( + range(len(messages)), + key=lambda p: (messages[p]["ts"], p), + ) + + id_map: dict[str, int] = dict(existing) + imported = 0 + skipped = 0 + first_id: int | None = None + last_id: int | None = None + + for pos in order: + msg = messages[pos] + sid = msg["source_id"] + if sid in id_map: + skipped += 1 + continue + + # Resolve reply_to_source_id to the archive row id of the referenced + # message (already imported in the archive or earlier in this batch in + # ts order). A forward reference (target comes later in the batch) + # leaves reply_to=None but keeps reply_to_source_id for traceability. + rti = msg.get("reply_to_source_id") + reply_to: str | None = None + if rti is not None: + target = id_map.get(rti) + if target is not None: + reply_to = str(target) + + data: dict = { + "from": msg["from"], + "body": msg["body"], + "thread": msg["thread"], + "reply_to": reply_to, + "source": source, + "source_id": sid, + } + if rti is not None: + data["reply_to_source_id"] = rti + if msg.get("refs") is not None: + data["refs"] = msg["refs"] + if msg.get("blocks") is not None: + data["blocks"] = msg["blocks"] + + row_id = await archive.record( + event_type=EVENT_A2A, + data=data, + agent_name=msg["from"], + app_id=msg["thread"], + summary=msg["body"][:200], + source=source, + source_id=sid, + timestamp=msg["ts"], + ) + id_map[sid] = row_id + if first_id is None: + first_id = row_id + last_id = row_id + imported += 1 + + # Optional vector embedding (skipped when deferring). + if not defer_index: + meta: dict = { + "agent": msg["from"], + "source": source, + "source_id": sid, + "ts": msg["ts"], + } + if isinstance(row_id, int) and row_id >= 0: + meta["archive_span_id"] = row_id + try: + await stores["vector"].add(msg["body"], metadata=meta) + except Exception: # noqa: BLE001 + logger.warning( + "a2a_import: vector embed failed for source_id %r; " + "message is archived and recoverable", + sid, + ) + + return { + "imported": imported, + "skipped": skipped, + "first_id": first_id, + "last_id": last_id, + } + + async def task_create( title: str, *, @@ -1031,7 +1209,7 @@ 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", + "supersede", "a2a_send", "a2a_import", "a2a_feed", "a2a_channels", "a2a_members", "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",