chore: drop generated artifacts not tracked on master - #230
Conversation
📝 WalkthroughWalkthroughThis PR adds historical A2A message import. It extends archive provenance metadata, preserves historical timestamps, enforces source-based idempotency, resolves replies, exposes an admin HTTP endpoint, updates capability detection, and adds a remote client method. ChangesHistorical A2A import
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant A2AClient
participant HTTPServer
participant ImportService
participant ArchiveStore
A2AClient->>HTTPServer: POST /a2a/import
HTTPServer->>ImportService: Validate and import messages
ImportService->>ArchiveStore: Check source IDs and reply targets
ArchiveStore-->>ImportService: Existing IDs and targets
ImportService->>ArchiveStore: Record historical messages
ArchiveStore-->>ImportService: Archive IDs
ImportService-->>HTTPServer: Import result
HTTPServer-->>A2AClient: Response
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
Much better than the closed #218: this actually implements the endpoint. Verified quickly before deep review: the package compiles, the handler exists rather than just being dispatched to, /a2a/import is inside _is_admin_route so the admin token gates it, the archive_import table now goes through the migrations registry instead of riding on a schema constant, remote.py has parity, and the docstring commits to whole-batch refusal plus (source, source_id) idempotency. That is the shape I asked for. It cannot merge yet for one reason: ZERO test files in the diff. This is an admin batch-import endpoint on a bus where 'from' is self-claimed, and its entire value proposition is writing historical messages that other agents will later trust as provenance. Untested is the one thing it cannot be. The rework order on #218 said tests in tests/ with asyncio markers, and the taosmd CI gate now runs them. Required before I review the logic in depth, so we only do that once:
Red-first for at least the admin gate and the whole-batch refusal: show them failing against the unfixed path and paste the runs. Also retitle. The PR is called 'chore: drop generated artifacts not tracked on master', which is the subject of your uv.lock cleanup commit, not the change. That title has now landed on four separate taosmd PRs and it makes the history unreadable. |
PR Summary by QodoAdd admin /a2a/import for idempotent historical A2A message imports
AI Description
Diagram
High-Level Assessment
Files changed (6)
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
taosmd/http_server.py (1)
2129-2144: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
POST /a2a/importis listed as both a general and an admin endpoint.Line 2132 prints the route in the general endpoint list. Line 2144 prints it again under "Admin (admin token required)". The route is admin-only, so the general listing tells operators the wrong thing. Remove it from the general list.
📝 Proposed fix
- "POST /a2a/send, POST /a2a/import, GET /a2a/messages, GET /a2a/stream, " + "POST /a2a/send, GET /a2a/messages, GET /a2a/stream, "🤖 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/http_server.py` around lines 2129 - 2144, Remove POST /a2a/import from the general endpoint list printed by the startup/help output, while retaining it only in the admin endpoint list alongside the other admin routes.taosmd/archive.py (1)
223-257: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftA failed row leaves a partially written import.
recordappends the JSONL line and flushes it before the index insert, anda2a_importwrites one message per iteration with no transactional boundary. If the partial unique index rejects a(source, source_id)pair,sqlite3.IntegrityErrorpropagates: the JSONL file holds a record with no index row, earlier messages stay imported, and the caller receives a 500 with no counts. This contradicts the "fail-loud, zero partial writes" claim in thea2a_importdocstring at Lines 644-647. The service pre-check narrows the window but does not close it, because two concurrent imports of the same source can both pass it.
taosmd/archive.py#L223-L257: catchsqlite3.IntegrityErroraround thearchive_indexinsert, roll back, and return the existing row id for the(source, source_id)pair so re-import stays idempotent at the storage layer.taosmd/service.py#L730-L796: contain per-message write failures so the response still reportsimported,skipped, and the failure, instead of aborting the batch mid-write.🤖 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 223 - 257, Update taosmd/archive.py lines 223-257 around the archive_index insert in record to catch sqlite3.IntegrityError, roll back the failed transaction, and return the existing row ID for the matching (source, source_id) pair so duplicate imports remain idempotent. Update taosmd/service.py lines 730-796 in the a2a_import per-message loop to contain write failures, continue producing a response with imported and skipped counts plus the failure details, and prevent the batch from aborting mid-write.
🧹 Nitpick comments (2)
taosmd/service.py (1)
670-699: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAdd a bound on batch size and message size.
a2a_importaccepts an unboundedmessageslist and unboundedbodystrings.POST /a2a/sendenforces a 64KB per-message limit intaosmd/http_server.pyat Lines 1428-1435; the import path enforces nothing. The whole import runs on the single service loop, so one large batch blocks every other request for its duration, including embedding whendefer_indexis false.The route is admin-gated, so this is a resilience concern rather than an exploit. Add a maximum batch length and a per-message size check in the validation pass, before any write.
🤖 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/service.py` around lines 670 - 699, Add maximum batch-length and per-message body-size validation to the a2a_import validation flow before any writes, reusing the existing 64KB message limit enforced by the send route. Update the validation around source/messages and the loop over messages to reject oversized batches and body strings with clear ValueError messages, while preserving the existing field and duplicate-source checks.taosmd/archive.py (1)
381-392: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the unused reply-target lookup.
find_reply_targetis only defined, anda2a_importresolves targets from thefind_source_idsmap. Removefind_reply_targetunless a caller is added.🤖 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 381 - 392, Remove the unused Archive method find_reply_target, including its query implementation and docstring; retain a2a_import’s existing target resolution through the find_source_ids map and do not add a replacement caller.
🤖 Prompt for all review comments with 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.
Inline comments:
In `@taosmd/archive.py`:
- Around line 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.
In `@taosmd/remote.py`:
- Around line 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.
In `@taosmd/service.py`:
- Around line 658-660: Update the docstring describing reply_to resolution in
the batch-processing method to state that forward references—where the target
message appears later in timestamp order—produce reply_to=None, while
unresolvable ids still reject the batch. Keep the implementation around the
existing forward-reference handling unchanged.
- Around line 763-796: The a2a_import flow archives messages and adds vectors
without populating the knowledge graph. Update a2a_import to invoke
process_conversation_turn(...) for each imported user message using the
available conversation context, or add a documented migration backfill that
performs the same extraction; ensure imported turns are processed even when
vector indexing is deferred.
---
Outside diff comments:
In `@taosmd/archive.py`:
- Around line 223-257: Update taosmd/archive.py lines 223-257 around the
archive_index insert in record to catch sqlite3.IntegrityError, roll back the
failed transaction, and return the existing row ID for the matching (source,
source_id) pair so duplicate imports remain idempotent. Update taosmd/service.py
lines 730-796 in the a2a_import per-message loop to contain write failures,
continue producing a response with imported and skipped counts plus the failure
details, and prevent the batch from aborting mid-write.
In `@taosmd/http_server.py`:
- Around line 2129-2144: Remove POST /a2a/import from the general endpoint list
printed by the startup/help output, while retaining it only in the admin
endpoint list alongside the other admin routes.
---
Nitpick comments:
In `@taosmd/archive.py`:
- Around line 381-392: Remove the unused Archive method find_reply_target,
including its query implementation and docstring; retain a2a_import’s existing
target resolution through the find_source_ids map and do not add a replacement
caller.
In `@taosmd/service.py`:
- Around line 670-699: Add maximum batch-length and per-message body-size
validation to the a2a_import validation flow before any writes, reusing the
existing 64KB message limit enforced by the send route. Update the validation
around source/messages and the loop over messages to reject oversized batches
and body strings with clear ValueError messages, while preserving the existing
field and duplicate-source checks.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: c9ee30f5-220a-4b38-b57a-2d1b90f43121
📒 Files selected for processing (6)
taosmd/archive.pytaosmd/capabilities.pytaosmd/http_server.pytaosmd/migrations.pytaosmd/remote.pytaosmd/service.py
| 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; |
There was a problem hiding this comment.
🩺 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.
| 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) | ||
|
|
There was a problem hiding this comment.
🎯 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.pyRepository: 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' -- || trueRepository: 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 -v1Repository: 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.pyRepository: 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.
| * **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. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Docstring conflicts with the forward-reference behavior.
Lines 658-660 state that reply_to_source_id resolves to the archive row id of the matching message. The code at Lines 741-746 leaves reply_to as None when the target appears later in ts order, as the comment at Lines 737-740 describes. Align the docstring with the implemented behavior so callers know a forward reference produces reply_to=None.
📝 Proposed docstring fix
* **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.
+ row id of the matching message (same ``source``) when that message is
+ already imported or comes earlier in ts order; a forward reference leaves
+ ``reply_to`` as ``None`` and keeps ``reply_to_source_id`` for traceability.
+ An id that matches no message in the archive or the batch refuses the batch.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| * **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. | |
| * **reply_to resolution** -- ``reply_to_source_id`` is resolved to the archive | |
| row id of the matching message (same ``source``) when that message is | |
| already imported or comes earlier in ts order; a forward reference leaves | |
| ``reply_to`` as ``None`` and keeps ``reply_to_source_id`` for traceability. | |
| An id that matches no message in the archive or the batch refuses the batch. |
🤖 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/service.py` around lines 658 - 660, Update the docstring describing
reply_to resolution in the batch-processing method to state that forward
references—where the target message appears later in timestamp order—produce
reply_to=None, while unresolvable ids still reject the batch. Keep the
implementation around the existing forward-reference handling unchanged.
| 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, | ||
| ) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find the extraction entry points and any existing back-fill command.
rg -nP --type=py '\bprocess_conversation_turn\s*\(' -C3
rg -nP --type=py '\breindex\b' -C3 -g '!tests/**'Repository: jaylfc/taosmd
Length of output: 151
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== repository files near taosmd/service.py =="
git ls-files | rg '(^|/)taosmd/service\.py$|(\.py$)' | sed -n '1,120p'
echo
echo "== taosmd/service.py relevant outlines/searches =="
if [ -f taosmd/service.py ]; then
wc -l taosmd/service.py
rg -n "class|def |async def |archive\.record|process_conversation_turn|kg\.add_triple|vector\.add|def import|import_" taosmd/service.py -C2 | sed -n '1,240p'
fi
echo
echo "== broader extraction references =="
rg -nP --type=p 'Archive|archive\.record|process_conversation_turn|kg\.add_t|knowledge|graph|reindex|index_' -C2 -S .Repository: jaylfc/taosmd
Length of output: 11043
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== service a2a_import =="
sed -n '629,815p' taosmd/service.py
echo
echo "== remote.py a2a_import / extract refs =="
if [ -f taosmd/remote.py ]; then
rg -n "a2a_import|record|extract|graph|kg\." taosmd/remote.py -C3
fi
echo
echo "== archive/process_conversation_turn references =="
rg -n -S --glob '*.py' 'Archive\.record|process_conversation_turn|kg\.add_triple|knowledge_graph' .
echo
echo "== reindex references =="
rg -n -S --glob '*.py' 'reindex|def reindex|\.reindex' . -C2Repository: jaylfc/taosmd
Length of output: 43057
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== ingest ingestion path =="
sed -n '240,360p' taosmd/service.py
echo
echo "== process_conversation_turn implementation =="
sed -n '217,335p' taosmd/memory_extractor.py
echo
echo "== normal archive record paths without KG calls? =="
python3 - <<'PY'
import ast, pathlib
for p in [pathlib.PurePath('taosmd/service.py'), pathlib.PurePath('taosmd/api.py'), pathlib.PurePath('taosmd/remote.py'), pathlib.PurePath('taosmd/http_server.py')]:
print(f'\n--- {p} ---')
try:
tree = ast.parse(p)
except FileNotFoundError:
print('missing')
continue
def node_text(n):
s=p.read_text().splitlines()
return '\n'.join(s[n.lineno-1:n.end_lineno]) if hasattr(n,'end_lineno') else ''
for n in ast.walk(tree):
if isinstance(n, ast.Assign) and any(isinstance(target, ast.Name) and target.id in ('message', 'data', 'payload') for target in n.targets):
call = n.value
if isinstance(call, ast.Await):
call = call.value
elif isinstance(call, ast.Call):
pass
else:
continue
name = ast.unparse(call)
if 'archive.record' in name:
print(ast.unparse(n))
print('contains process_conversation_turn:', 'process_conversation_turn' in node_text(n))
print('contains kg.add_triple:', 'kg.add_triple' in node_text(n))
PY
echo
echo "== existing KG back-fill command candidates =="
rg -n -S --glob '*.py' -e 'KnowledgeGraph|TemporalKnowledgeGraph|kg|process_conversation_turn' taosmd/cli.py taosmd/api.py taosmd/service.py -C2Repository: jaylfc/taosmd
Length of output: 11007
Import A2A messages through the KG-extraction path or add a migration step.
a2a_import() only writes archive events and vector rows. Imported user messages must run the required extraction flow; add process_conversation_turn(...) for these turns, or provide and documentation a migration back-fill. taosmd reindex only rebuilds vectors from the archive and does not populate the knowledge graph.
🤖 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/service.py` around lines 763 - 796, The a2a_import flow archives
messages and adds vectors without populating the knowledge graph. Update
a2a_import to invoke process_conversation_turn(...) for each imported user
message using the available conversation context, or add a documented migration
backfill that performs the same extraction; ensure imported turns are processed
even when vector indexing is deferred.
Source: Coding guidelines
Code Review by Qodo
Context used✅ Compliance rules (platform):
13 rules 1. Forward reply IDs unresolved
|
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
| 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") |
There was a problem hiding this comment.
1. a2a_import errors not archived 📘 Rule violation ☼ Reliability
a2a_import can raise request-validation errors before any archive.record() call, so failed import interactions are not archived. This violates the requirement that public entrypoints archive interactions even on failure paths.
Agent Prompt
## Issue description
The new `/a2a/import` entrypoint can fail (e.g., invalid `source`, invalid `messages`, unresolvable `reply_to_source_id`) before any archive write occurs, meaning failure paths are not archived.
## Issue Context
PR Compliance ID 1019881 requires that each public entrypoint processing conversation turns archives the interaction regardless of success/failure (including error cases). Right now, `a2a_import()` validates and raises `ValueError` before any `archive.record()` call, and the HTTP handler raises `_BadRequest` similarly.
## Fix Focus Areas
- taosmd/http_server.py[1509-1527]
- taosmd/service.py[670-715]
- taosmd/service.py[763-772]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
| target = id_map.get(rti) | ||
| if target is not None: | ||
| reply_to = str(target) |
There was a problem hiding this comment.
2. Forward reply ids unresolved 🐞 Bug ≡ Correctness
service.a2a_import accepts reply_to_source_id values that exist later in the same batch, but during insertion it only resolves reply_to when the target is already in id_map, leaving valid forward references permanently stored with reply_to=None. This contradicts the documented /a2a/import contract that reply_to_source_id resolves to an imported archive id (or the batch fails).
Agent Prompt
### Issue description
`a2a_import()` validates that `reply_to_source_id` exists in the archive or somewhere in the batch, but it only resolves `reply_to` from `id_map` at insert time. When the referenced message is inserted later (forward reference in ts order), the importer writes `reply_to=None` and never fixes it, violating the endpoint’s stated behavior.
### Issue Context
Insertion is sorted by `(ts, original_position)` to make archive row ids follow ts order. That ordering must not silently drop reply linkage.
### Fix Focus Areas
- taosmd/service.py[701-777]
- taosmd/http_server.py[104-109]
### Suggested change
Choose one:
1) **Disallow forward references**: during validation, require the referenced message to be in `existing` or in the batch *and* have `ts` <= the current message’s `ts` (with tie-break on position), otherwise 400.
2) **Two-pass resolution** (preferred if you want to allow forward refs): first insert all new messages with `reply_to=None` but keep `reply_to_source_id`; after all inserts, run a second pass to update the stored JSON payload (or record an additional linking event) so every reply can be resolved to the final archive id.
3) **Drop ts-sorting requirement** and insert in input order so reply targets always precede replies (only if acceptable for your historical ordering guarantees).
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
| async def a2a_import( | ||
| source: str, | ||
| messages: list[dict], | ||
| *, |
There was a problem hiding this comment.
3. Import ignores remote routing 🐞 Bug ≡ Correctness
service.a2a_import always imports into local stores and never checks _get_remote(data_dir), unlike other A2A service methods that transparently forward to RemoteClient when a remote server is configured. This can cause admins to import into the wrong (local) bus in remote-configured deployments despite RemoteClient.a2a_import existing.
Agent Prompt
### Issue description
`service.a2a_import()` does not follow the established remote-forwarding contract used by `a2a_send`, `a2a_feed`, etc. When a remote server URL is configured, callers likely expect the import to occur on the remote bus.
### Issue Context
This PR adds `RemoteClient.a2a_import()`, suggesting imports are intended to be callable in remote mode.
### Fix Focus Areas
- taosmd/service.py[345-437]
- taosmd/service.py[629-706]
- taosmd/remote.py[206-248]
### Suggested change
Add at the top of `service.a2a_import()` (after basic argument validation):
- `remote = _get_remote(data_dir)`
- if non-null: `return await remote.a2a_import(source, messages, defer_index=defer_index)`
Then proceed with local-store import only when `remote is None`.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
| 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; |
There was a problem hiding this comment.
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
| lambda c: ( | ||
| has_column(c, "archive_index", "source") | ||
| and has_column(c, "archive_index", "source_id") | ||
| ), |
There was a problem hiding this comment.
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
1 participant
Autonomous build of board card tsk-qh3bfs.
Files:
taosmd/archive.py | 58 ++++++++++++++--
taosmd/capabilities.py | 3 +-
taosmd/http_server.py | 33 +++++++--
taosmd/migrations.py | 25 +++++++
taosmd/remote.py | 18 +++++
taosmd/service.py | 180 ++++++++++++++++++++++++++++++++++++++++++++++++-
6 files changed, 306 insertions(+), 11 deletions(-)
Summary by CodeRabbit