Revise PR #230: the new index breaks every existing install on upgrade

changelog.d/tsk-sgoilz-upgrade-and-import.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Revised PR #230: fixed upgrade path broken on every existing install by removing source/source_id from INDEX_SCHEMA and changing migration guard from has_column to index_exists; also added _get_remote branch to a2a_import so remote-configured installs don't silently write history to local archive

taosmd/archive.py

@@ -179,8 +179,16 @@ async def record(
         app_id: str | None = None,
         summary: str = "",
         project: 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,7 +200,7 @@ 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,
@@ -342,6 +350,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,

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"',

taosmd/http_server.py

@@ -105,10 +105,17 @@
 ``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
+<<<<<<< HEAD
 ``GET  /a2a/messages``     ``?thread=&since=&limit=&fields=&format=``  -> ``{"messages": [...]}`` (``fields=id,sender,body`` projects keys; ``format=ndjson`` emits one message per line; ``since`` is an epoch timestamp in seconds, not a message id; values below 1e9 return 400)
 ``GET  /a2a/stream``       ``?thread=&since=``             -> SSE stream (text/event-stream); ``since`` is an epoch timestamp in seconds, not a message id; values below 1e9 return 400
 ``GET  /a2a/threads``      ``?principal=``                  -> ``{"threads": [...]}`` thread list for the principal, ordered by latest activity desc; each entry has ``thread``, ``kind``, ``participants``, ``last_message: {id, ts, from, body_preview}`` (see notes below)
 ``GET  /a2a/threads/{thread}/messages`` ``?before=&after=&limit=`` -> ``{"thread", "messages": [...]}`` cursor-paginated message envelope, oldest-first
+=======
+``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 <from>`` 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)
+>>>>>>> 04e6b07af77bb30d797e0867a2647196c18b7c28
 ``GET  /a2a/channels``                                     -> ``{"channels": [...]}``
 ``GET  /a2a/members``      ``?channel=<name>``             -> ``{"members": [...]}``
 ``POST /tasks``            ``{"title", "body"?, "project"?, "assignee"?, "priority"?, "depends_on"?: [...], "created_by"}`` -> task object
@@ -830,6 +837,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:
@@ -1029,6 +1037,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":
@@ -1608,6 +1618,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]
@@ -2232,7 +2262,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, "
@@ -2244,7 +2274,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:

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,

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,
         *,