Skip to content

tsk-qh3bfs [OPEN] POST /a2a/import: admin batch import with historic #230

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jaylfc wants to merge 2 commits into
base: master
Choose a base branch
from
Open

tsk-qh3bfs [OPEN] POST /a2a/import: admin batch import with historic #230

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
08bacbe
tsk-qh3bfs [OPEN] POST /a2a/import: admin batch import with historic
jaylfc Aug 2, 2026
04e6b07
chore: drop generated artifacts not tracked on master
jaylfc Aug 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 53 additions & 5 deletions taosmd/archive.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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;
Comment on lines +72 to +74

@coderabbitai coderabbitai Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🔴 Critical | 🏗️ Heavy lift

CREATE UNIQUE INDEX on new columns runs before the migration that adds them.

ArchiveStore.init executes INDEX_SCHEMA (Line 114) and only then calls migrations.migrate (Line 118). On an existing database, CREATE TABLE IF NOT EXISTS archive_index is a no-op, so the table still lacks source and source_id when this statement runs. SQLite then raises OperationalError: no such column: source and init() fails for every upgraded installation. Fresh databases are unaffected, which hides the fault in tests that start from an empty data dir.

Move the index creation out of INDEX_SCHEMA and rely on _archive_index_source_uid in taosmd/migrations.py, which already creates the same index after adding the columns.

🐛 Proposed fix: create the index only in the migration 
 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;

migrations._archive_index_source_uid creates idx_archive_source_uid for both fresh and existing databases, because migration 3 runs whenever the columns are missing.

Run the following script to confirm the ordering and that no other code path creates the index first:

#!/bin/bash
# Confirm INDEX_SCHEMA execution precedes migrate() and locate all creators of the index.
rg -n -C4 'executescript|migrations.migrate' taosmd/archive.py
rg -n -C2 'idx_archive_source_uid' -g '*.py'
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@taosmd/archive.py` around lines 72 - 74, Remove the idx_archive_source_uid
CREATE UNIQUE INDEX statement from INDEX_SCHEMA so ArchiveStore.init does not
reference source or source_id before migrations.migrate runs. Keep
_archive_index_source_uid in migrations.py as the sole creator of this index
after adding the columns, preserving index creation for both fresh and upgraded
databases.

Comment on lines +72 to +74

@qodo-code-review qodo-code-review Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remediation recommended

4. Unique index can orphan jsonl 🐞 Bug ☼ Reliability

ArchiveStore.record writes and flushes the JSONL event line before inserting into archive_index;
with the new unique index on (source, source_id), an IntegrityError can occur after the file append,
leaving an unindexed/orphaned archive line and breaking file_path/line_number provenance. This is
most likely under concurrent imports or any caller that bypasses the importer’s pre-check.
Agent Prompt 
### Issue description
The new `(source, source_id)` unique index is enforced only at the SQLite index layer, but `ArchiveStore.record()` appends to the JSONL file first. If the subsequent `INSERT` fails (e.g., due to the unique index), the JSONL line remains without a corresponding index row.

### Issue Context
This PR introduces both the unique index and the ability for callers (a2a_import) to set `source`/`source_id`.

### Fix Focus Areas
- taosmd/archive.py[179-275]
- taosmd/archive.py[52-75]

### Suggested change
When `source` and `source_id` are both non-null:
1) **Before writing JSONL**, query `archive_index` for an existing row id for `(source, source_id)` (optionally also `event_type`) and return it if found.
2) Optionally, still keep the unique index as a hard safety net.
3) Consider wrapping the sqlite insert in try/except for `sqlite3.IntegrityError` and, on conflict, re-query and return the existing id (but only safe if you also avoided writing JSONL first).

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools


CREATE TABLE IF NOT EXISTS archive_settings (
key TEXT PRIMARY KEY,
Expand Down Expand Up @@ -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
Expand All @@ -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,
}
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down
3 changes: 2 additions & 1 deletion taosmd/capabilities.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"',
Expand Down
33 changes: 29 additions & 4 deletions taosmd/http_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 <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)
``GET /a2a/channels`` -> ``{"channels": [...]}``
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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":
Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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, "
Expand All @@ -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:
Expand Down
25 changes: 25 additions & 0 deletions taosmd/migrations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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")
),
Comment on lines +231 to +234

@qodo-code-review qodo-code-review Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remediation recommended

5. Migration detect skips index 🐞 Bug ☼ Reliability

The archive_index_source_uid migration’s detect function only checks for source/source_id columns;
if those columns exist but idx_archive_source_uid is missing, the migration will be treated as
applied and the uniqueness constraint will not be created. This can silently disable the idempotency
safety net on some upgraded databases.
Agent Prompt 
### Issue description
Migration 3 for `archive_index` creates `idx_archive_source_uid` but its detect lambda does not verify the index exists. Databases that have columns but lack the index will never get the constraint.

### Issue Context
This file already provides `index_exists()` and uses it in other migration detect probes.

### Fix Focus Areas
- taosmd/migrations.py[202-236]
- taosmd/migrations.py[126-131]

### Suggested change
Update the detect lambda for `archive_index_source_uid` to require:
- `has_column(..., 'source')`
- `has_column(..., 'source_id')`
- `index_exists(c, 'idx_archive_source_uid')`
This matches the established pattern used elsewhere (e.g., session_catalog taxonomy migration).

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

),
)


Expand Down
18 changes: 18 additions & 0 deletions taosmd/remote.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Comment on lines +231 to +248

@coderabbitai coderabbitai Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Find callers of a2a_import and compare the remote-forwarding pattern.
rg -nP --type=py '\ba2a_import\s*\(' -C4
rg -nP --type=py '_get_remote\s*\(' -C2 taosmd/service.py

Repository: jaylfc/taosmd

Length of output: 151

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== files =="
git ls-files | sed -n '1,120p'

echo "== locate relevant files =="
fd -a 'remote\.py|service\.py' . | sed -n '1,120p'

echo "== search a2a_import occurrences =="
rg -n 'a2a_import|\ba2a_send\b|_get_remote|RemoteClient|/a2a/import' . || true

echo "== list python files under repo =="
git ls-files '*.py' | sed -n '1,200p'

Repository: jaylfc/taosmd

Length of output: 29199

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== git status/stat =="
git status --short
git diff --stat || true
git log --oneline -5 || true

echo "== tracked python files containing import/export/API names =="
git ls-files '*.py' | xargs rg -n 'a2a_import|a2a_send|RemoteClient|_get_remote|/a2a/import|import .*a2a' -- || true

Repository: jaylfc/taosmd

Length of output: 13587

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== service remote forwarding for a2a methods =="
sed -n '340,680p' taosmd/service.py | nl -ba -v340

echo "== http_server a2a_import handler =="
sed -n '1500,1540p' taosmd/http_server.py | nl -ba -v1500

echo "== cli around RemoteClient creation =="
sed -n '500,545p' taosmd/cli.py | nl -ba -v500

echo "== module exported/importable a2a_import references =="
python3 - <<'PY'
from pathlib import Path
import re
root = Path('.')
for p in root.glob('**/*.py'):
    text = p.read_text(errors='ignore')
    if r'\ba2a_import\b' in text:
        # Print non-test, non-import-site occurrences plus import statements that mention it
        for i, line in enumerate(text.splitlines(), 1):
            if line.strip().startswith('from ') and '.service import' in line or re.search(r'\ba2a_import\b', line):
                continue
matches = []
for p in sorted(root.glob('**/*.py')):
    text = p.read_text(errors='ignore')
    occurrences = [(i+1, line.strip()) for i,line in enumerate(text.splitlines(),1) if re.search(r'\ba2a_import\b', line)]
    if not occurrences:
        continue
    print(f'FILE {p}')
    for i,line in occurrences:
        print(f'{i}: {line}')
PY

echo "== __all__ in service/remote =="
sed -n '1200,1220p' taosmd/service.py | nl -ba -v1200
sed -n '410,425p' taosmd/remote.py | nl -ba -v410
sed -n '1,120p' taosmd/__init__.py | nl -ba -v1

Repository: jaylfc/taosmd

Length of output: 239

🏁 Script executed:

#!/bin/bash
set -euo pipefail
export LC_ALL=C

echo "== service remote forwarding for a2 a methods =="
sed -n '340,680p' taosmd/service.py

echo "== http_server a2a_import handler =="
sed -n '1500,1540p' taosmd/http_server.py

echo "== cli around RemoteClient creation =="
sed -n '500,545p' taosmd/cli.py

echo "== python call occurrences =="
python3 - <<'PY'
from pathlib import Path
import re
root = Path('.')
for p in sorted(root.glob('**/*.py')):
    text = p.read_text(errors='ignore')
    occurrences = [(i+1, line.strip()) for i,line in enumerate(text.splitlines(),1) if re.search(r'\ba2a_import\s*\(', line)]
    if not occurrences:
        continue
    print(f'FILE {p}')
    for i,line in occurrences:
        print(f'{i}: {line}')
PY

echo "== __all__ in service/remote/api =="
sed -n '1200,1220p' taosmd/service.py
sed -n '410,425p' taosmd/remote.py
sed -n '1,140p' taosmd/__init__.py
sed -n '1,160p' taosmd/api.py

Repository: jaylfc/taosmd

Length of output: 29868

Forward service.a2a_import through RemoteClient.

service.a2a_import calls _api._ensure_stores(data_dir), and the only Python caller is the HTTP handler, which also passes the local data_dir. Add the same data_dir-aware RemoteClient forwarding as service.a2a_send; otherwise imports from the API with TAOSMD_SERVER_URL always write to the local archive.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@taosmd/remote.py` around lines 231 - 248, Update RemoteClient.a2a_import to
accept and forward the caller’s data_dir, matching the existing service.a2a_send
forwarding pattern. Ensure the HTTP handler passes its local data_dir through
RemoteClient so remote imports use the configured remote service instead of
writing to the local archive.

async def a2a_feed(
self,
*,
Expand Down
Loading
Loading