feat(http_server): add /a2a/threads and /a2a/threads/{thread}/messages endpoints - #223
Closed
jaylfc wants to merge 2 commits into
Closed
feat(http_server): add /a2a/threads and /a2a/threads/{thread}/messages endpoints#223jaylfc wants to merge 2 commits into
jaylfc wants to merge 2 commits into
Conversation
Implement A2A bus read APIs for thread listing and cursor pagination:
- a2a_threads: returns threads participated in by an agent
- a2a_thread_messages: returns messages from a thread with cursor pagination
- Before/after cursors accept message IDs (not timestamps) to avoid epoch trap
- Thread list ordered by latest activity descending
Adds new endpoints:
- GET /a2a/threads
- GET /a2a/threads/{thread}/messages
📝 WalkthroughWalkthroughThe service layer adds A2A thread summaries and thread-message retrieval with local archive filtering, alias handling, pagination, and remote forwarding. HTTP routing exposes the new endpoints, while the ChangesA2A thread access
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant HTTPDispatcher
participant A2AService
participant RemoteClient
participant LocalArchive
Client->>HTTPDispatcher: GET /a2a/threads or /a2a/threads/{thread}/messages
HTTPDispatcher->>A2AService: invoke thread service function
alt Remote server configured
A2AService->>RemoteClient: forward thread request
RemoteClient-->>A2AService: return thread data
else Local archive
A2AService->>LocalArchive: query EVENT_A2A rows
LocalArchive-->>A2AService: return filtered or paginated rows
end
A2AService-->>HTTPDispatcher: return thread response
HTTPDispatcher-->>Client: send response
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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 |
PR Summary by QodoAdd A2A thread listing and thread-message pagination APIs
AI Description
Diagram
High-Level Assessment
Files changed (2)
|
![kilo-code-bot[bot]](https://avatars.githubusercontent.com/in/2193792?s=60&v=4){kind=small}
| return sorted(members) | ||
|
|
||
|
|
||
| async def a2a_threads(*, agent: str | None = None, data_dir=None) -> list[dict]: |
There was a problem hiding this comment.
WARNING: Duplicate a2a_threads definition — this first definition (lines 628–724) is dead code because a second a2a_threads is defined at line 1260 and overwrites it.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| data = {} | ||
|
|
||
| # Skip superseded messages | ||
| from .admin import A2AAdminState # noqa: PLC0415 |
There was a problem hiding this comment.
SUGGESTION: Inefficient loop — A2AAdminState is instantiated inside the row loop (lines 664–668), creating a new object for every archive row. Move this outside the loop.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| return result | ||
|
|
||
|
|
||
| async def a2a_thread_messages( |
There was a problem hiding this comment.
WARNING: Duplicate a2a_thread_messages definition — this first definition (lines 727–852) is dead code because a second definition at line 1362 overwrites it. It also contains incomplete cursor pagination (lines 800–803, 845–849).
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
|
|
||
| # Limit and pagination | ||
| params.append(limit) | ||
| if before is not None or after is not None: |
There was a problem hiding this comment.
WARNING: Cursor pagination is not implemented — before and after values are accepted but never used in the SQL WHERE clause, so the query returns all matching rows regardless of cursor.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| elif thread == row.get("app_id"): | ||
| result.append(msg) | ||
|
|
||
| # Apply cursor-based pagination if needed |
There was a problem hiding this comment.
WARNING: Cursor pagination is not implemented — this pass means cursor values are never applied to filter the result set.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| elif method == "GET" and path == "/a2a/stream": | ||
| self._handle_a2a_stream(query) | ||
| return # SSE response already sent; skip _send_json error path | ||
| elif method == "GET" and path == "/a2a/threads": |
There was a problem hiding this comment.
WARNING: Duplicate route registration — /a2a/threads and /a2a/threads/{thread}/messages routes are registered twice (lines 956–959 and 963–966). The second registration is unreachable dead code.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
|
|
||
| # Sort result based on cursor direction | ||
| if before is not None: | ||
| # Already oldest-first from our loop |
There was a problem hiding this comment.
SUGGESTION: Misleading comment — all_messages is sorted newest-first, so the loop appends in newest-first order, not oldest-first. This comment is incorrect and could cause future bugs.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
Code Review SummaryStatus: 7 Issues Found | Recommendation: Address before merge Overview
Issue Details (click to expand)WARNING
SUGGESTION
Files Reviewed (3 files)
Fix these issues in Kilo Cloud Reviewed by step-3.7-flash · Input: 64.8K · Output: 18K · Cached: 270.5K |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🧹 Nitpick comments (2)
taosmd/service.py (2)
1481-1493: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
__all__is not sorted (RUF022).Ruff flags this; apply the isort-style ordering it suggests.
🤖 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 1481 - 1493, Sort the exported names in __all__ using the isort-style ordering required by RUF022, preserving every existing symbol and its grouping semantics where applicable.Source: Linters/SAST tools
1362-1478: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winNo tests accompany these endpoints.
The PR description acknowledges the card asked for tests and none were added; the defects above (unbound SQL parameters, non-existent table,
sqlite3.Row.get, always-Nonelast_message) would all have been caught by a single happy-path test against a tmpdata_dir. Want me to drafttests/test_a2a_threads.pycovering listing order, alias resolution, deleted-channel suppression, and before/after cursor paging?🤖 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 1362 - 1478, Add tests for a2a_thread_messages using a temporary data_dir and seeded archive data. Cover default listing order, alias resolution, deleted-channel suppression, and before/after cursor pagination; ensure the happy path exercises SQL parameter binding, archive row decoding, and returned message fields so runtime errors such as sqlite3.Row.get or missing last_message are detected.
🤖 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/http_server.py`:
- Around line 956-959: Implement the missing TaosmdHandler methods
_handle_a2a_threads and _handle_a2a_thread_messages, or remove their routing
branches. For the per-thread messages handler, URL-decode and extract the thread
segment, then call service.a2a_thread_messages with thread, before, after, and
limit parameters; ensure both routes return their service results through the
existing response flow.
- Around line 960-966: Remove the duplicate GET route branches for
"/a2a/threads" and "/a2a/threads/" messages from the request-dispatch chain,
preserving the earlier handlers and the "/a2a/stream" SSE behavior.
In `@taosmd/service.py`:
- Around line 628-852: Remove the duplicate definitions of a2a_threads and
a2a_thread_messages in this block, retaining the later implementations that are
active at import time. Delete the entire unreachable block, including its broken
per-row A2AAdminState construction and incomplete cursor query logic.
- Around line 1260-1272: Update the local implementation of a2a_threads to honor
the agent parameter by filtering returned threads to those where the principal
participates, including matching data.get("from") and any modeled recipients;
keep the remote forwarding behavior unchanged and ensure unfiltered results
remain available only when agent is unset.
- Around line 1401-1405: Update the thread-history query around resolved_thread
and alias_sources to include rows whose app_id matches the resolved thread or
any pre-rename alias, preserving renamed-channel history like a2a_feed. Keep
alias_sources applied to the query rather than leaving it unused. Make the
response’s "thread" field consistently use the chosen identifier
(resolved_thread or caller-supplied thread) and document that behavior.
- Around line 1449-1478: Update the cursor filtering in the pagination flow to
use exclusive bounds, excluding the message whose id equals before or after.
Correct the before branch ordering to match its documented pagination direction
by explicitly sorting the filtered result appropriately, rather than relying on
the existing all_messages order; preserve newest-first ordering for after and
apply the existing limit afterward.
- Around line 1315-1343: Update the thread aggregation logic around the thread
initialization and last-message update so the first newest-first row seeds
last_message immediately. Rename created_ts to a name representing the latest
timestamp, update all comparisons and downstream sorting references accordingly,
and preserve descending latest-activity ordering for each thread.
- Around line 1407-1447: Replace the hand-written SQL and direct archive._conn
access in the local message fetch with archive.query(event_type=EVENT_A2A,
app_id=resolved_thread, limit=...) as used by a2a_feed/a2a_channels. Filter
returned rows in Python using A2AAdminState.superseded_messages(), skip rows
containing data.admin_action, and preserve the deleted-channel/alias-history
behavior; normalize rows to dictionaries before reading data_json.
---
Nitpick comments:
In `@taosmd/service.py`:
- Around line 1481-1493: Sort the exported names in __all__ using the
isort-style ordering required by RUF022, preserving every existing symbol and
its grouping semantics where applicable.
- Around line 1362-1478: Add tests for a2a_thread_messages using a temporary
data_dir and seeded archive data. Cover default listing order, alias resolution,
deleted-channel suppression, and before/after cursor pagination; ensure the
happy path exercises SQL parameter binding, archive row decoding, and returned
message fields so runtime errors such as sqlite3.Row.get or missing last_message
are detected.
🪄 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: 113f1fde-0891-44ab-94f1-0dc6aed7c1a2
📒 Files selected for processing (3)
taosmd/__main__.pytaosmd/http_server.pytaosmd/service.py
💤 Files with no reviewable changes (1)
- taosmd/main.py
Comment on lines
+956
to
+959
| elif method == "GET" and path == "/a2a/threads": | ||
| self._handle_a2a_threads(query) | ||
| elif method == "GET" and path.startswith("/a2a/threads/") and "/messages" in path: | ||
| self._handle_a2a_thread_messages(query) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm whether the routed handlers exist anywhere in the repo.
rg -nP --type=py '\bdef\s+_handle_a2a_thread(s|_messages)\s*\(' -C2
# Show every reference to the handler names.
rg -nP --type=py '_handle_a2a_thread(s|_messages)\b' -C1
# Confirm the RemoteClient side of the contract exists too.
rg -nP --type=py '\b(async\s+)?def\s+a2a_thread(s|_messages)\s*\(' -C3Repository: jaylfc/taosmd
Length of output: 151
🏁 Script executed:
#!/bin/bash
set -u
echo "== candidate files =="
git ls-files '*.py' | sed -n '1,200p'
echo "== handler references =="
rg -n --type=py '_handle_a2a_thread|a2a_thread(s|_messages)\b|a2a_threads\b' -C2 .
echo "== http_server relevant sections =="
for f in $(git ls-files '*.py' | rg '(^|/)http_server\.py$|(^|/)service\.py$|taosmd/'); do
if [ -f "$f" ]; then
echo "--- $f outline/matches ---"
rg -n 'class TaosmdHandler|def do_GET|def _handle_a2a|a2a_thread_messages|a2a_threads|before="|after="|limit=' "$f" -C3 || true
fi
doneRepository: jaylfc/taosmd
Length of output: 40229
🏁 Script executed:
#!/bin/bash
set -u
echo "== http_server TaosmdHandler class range =="
wc -l taosmd/http_server.py
sed -n '920,1085p' taosmd/http_server.py | cat -n
echo "== a2a message handler params/validation =="
sed -n '1512,1534p' taosmd/http_server.py | cat -n
echo "== task/collection path parsing examples =="
sed -n '967,995p' taosmd/http_server.py | cat -n
rg -n --type=py '_handle_task(_[a-z_]+)?\b|_handle_collection(_[a-z_]+)?\b|collections/' taosmd/http_server.py -C3
echo "== service a2a_thread_messages signature/body =="
sed -n '727,790p' taosmd/service.py | cat -n
sed -n '1362,1405p' taosmd/service.py | cat -n
echo "== remote a2a thread methods =="
rg -n --type=py 'a2a_threads|a2a_thread_messages|thread' taosmd/remote.py -C4Repository: jaylfc/taosmd
Length of output: 32743
🏁 Script executed:
#!/bin/bash
set -u
echo "== handler definitions =="
rg -n --type=py '\bdef\s+_handle_([^(\s]+)' taosmd/http_server.py | sed -n '900,1000p'
echo "== exact method existence check =="
python3 - <<'PY'
from pathlib import Path
src = Path('taosmd/http_server.py').read_text()
for needle in ['def _handle_a2a_threads', 'def _handle_a2a_thread_messages', 'def _handle_task_create', 'def _handle_task_update']:
print(needle, 'FOUND' if needle in src else 'NOT FOUND')
PYRepository: jaylfc/taosmd
Length of output: 348
Add the A2A thread routes before exposing them.
_handle_a2a_threads and _handle_a2a_thread_messages are referenced from the TaosmdHandler routing but are not defined in taosmd/http_server.py, so any request to these routes can escape to the generic exception handler and return 500. Implement these handlers or remove the routes. For the per-thread messages route, also parse the URL-decoded thread segment and pass it to service.a2a_thread_messages, whose signature requires thread plus before/after/limit.
🤖 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 956 - 959, Implement the missing
TaosmdHandler methods _handle_a2a_threads and _handle_a2a_thread_messages, or
remove their routing branches. For the per-thread messages handler, URL-decode
and extract the thread segment, then call service.a2a_thread_messages with
thread, before, after, and limit parameters; ensure both routes return their
service results through the existing response flow.
Comment on lines
960
to
+966
| elif method == "GET" and path == "/a2a/stream": | ||
| self._handle_a2a_stream(query) | ||
| return # SSE response already sent; skip _send_json error path | ||
| elif method == "GET" and path == "/a2a/threads": | ||
| self._handle_a2a_threads(query) | ||
| elif method == "GET" and path.startswith("/a2a/threads/") and "/messages" in path: | ||
| self._handle_a2a_thread_messages(query) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Lines 963-966 are unreachable duplicates of Lines 956-959.
The same two elif conditions appear twice in one chain; the second pair can never be evaluated. Delete it.
🧹 Proposed fix
elif method == "GET" and path == "/a2a/stream":
self._handle_a2a_stream(query)
return # SSE response already sent; skip _send_json error path
- elif method == "GET" and path == "/a2a/threads":
- self._handle_a2a_threads(query)
- elif method == "GET" and path.startswith("/a2a/threads/") and "/messages" in path:
- self._handle_a2a_thread_messages(query)📝 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.
Suggested change
| elif method == "GET" and path == "/a2a/stream": | |
| self._handle_a2a_stream(query) | |
| return # SSE response already sent; skip _send_json error path | |
| elif method == "GET" and path == "/a2a/threads": | |
| self._handle_a2a_threads(query) | |
| elif method == "GET" and path.startswith("/a2a/threads/") and "/messages" in path: | |
| self._handle_a2a_thread_messages(query) | |
| elif method == "GET" and path == "/a2a/stream": | |
| self._handle_a2a_stream(query) | |
| return # SSE response already sent; skip _send_json error path |
🤖 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 960 - 966, Remove the duplicate GET route
branches for "/a2a/threads" and "/a2a/threads/" messages from the
request-dispatch chain, preserving the earlier handlers and the "/a2a/stream"
SSE behavior.
Comment on lines
+628
to
+852
| async def a2a_threads(*, agent: str | None = None, data_dir=None) -> list[dict]: | ||
| """Return a list of threads the principal participates in. | ||
|
|
||
| Returns each thread with thread identifier, title (if present), | ||
| kind, participants, and last_message (with id, ts, from, body_preview). | ||
| Threads are ordered by latest activity descending. | ||
|
|
||
| When a remote server URL is configured the call is forwarded to | ||
| :class:`~taosmd.remote.RemoteClient` transparently. | ||
| """ | ||
| remote = _get_remote(data_dir) | ||
| if remote is not None: | ||
| return await remote.a2a_threads(agent=agent) | ||
| stores = await _api._ensure_stores(data_dir) | ||
| archive = stores["archive"] | ||
|
|
||
| # Load admin state for filtering deleted channels | ||
| deleted_channels = set() | ||
| aliases: dict[str, str] = {} | ||
| if data_dir is not None: | ||
| from .admin import A2AAdminState # noqa: PLC0415 | ||
| _admin = A2AAdminState(data_dir) | ||
| deleted_channels = _admin.deleted_channels() | ||
| aliases = _admin.channel_aliases() | ||
|
|
||
| # Aggregate thread activity | ||
| threads: dict[str, dict] = {} | ||
| rows = await archive.query(event_type=EVENT_A2A, limit=100_000) | ||
|
|
||
| for row in rows: | ||
| try: | ||
| data = json.loads(row.get("data_json", "{}")) | ||
| except (json.JSONDecodeError, TypeError): | ||
| data = {} | ||
|
|
||
| # Skip superseded messages | ||
| from .admin import A2AAdminState # noqa: PLC0415 | ||
| _admin_state = A2AAdminState(data_dir) | ||
| _superseded = _admin_state.superseded_messages() | ||
| if row["id"] in _superseded: | ||
| continue | ||
|
|
||
| # Resolve thread through aliases | ||
| thread = data.get("thread") or row.get("app_id") or "general" | ||
| if thread in aliases: | ||
| thread = aliases[thread] | ||
|
|
||
| # Skip deleted channels (except when they're being merged as alias history) | ||
| alias_sources = [k for k, v in aliases.items() if v == thread] | ||
| if thread in deleted_channels and thread not in alias_sources: | ||
| continue | ||
|
|
||
| if thread not in threads: | ||
| threads[thread] = { | ||
| "thread": thread, | ||
| "title": data.get("title"), | ||
| "kind": data.get("kind"), | ||
| "participants": [], | ||
| "message_count": 0, | ||
| "last_message": None, | ||
| "created_ts": row["timestamp"], | ||
| } | ||
|
|
||
| # Add participant | ||
| sender = data.get("from") or "" | ||
| if sender and sender not in threads[thread]["participants"]: | ||
| threads[thread]["participants"].append(sender) | ||
|
|
||
| # Update message count | ||
| threads[thread]["message_count"] += 1 | ||
|
|
||
| # Update last message | ||
| msg = { | ||
| "id": row["id"], | ||
| "ts": row["timestamp"], | ||
| "from": sender, | ||
| "body_preview": (data.get("body") or "")[:100], | ||
| } | ||
| if row["timestamp"] > threads[thread]["created_ts"]: | ||
| threads[thread]["last_message"] = msg | ||
| threads[thread]["created_ts"] = row["timestamp"] | ||
|
|
||
| # Convert to list and add unread_count field (future enhancement) | ||
| result = [] | ||
| for thread_data in threads.values(): | ||
| result.append({ | ||
| "thread": thread_data["thread"], | ||
| "title": thread_data["title"], | ||
| "kind": thread_data["kind"], | ||
| "participants": thread_data["participants"], | ||
| "last_message": thread_data["last_message"], | ||
| "unread_count": 0, # Reserved for receipts work (tsk-fhltad) | ||
| }) | ||
|
|
||
| # Order by latest activity descending | ||
| result.sort(key=lambda t: t["last_message"]["ts"] if t["last_message"] else 0, reverse=True) | ||
| return result | ||
|
|
||
|
|
||
| async def a2a_thread_messages( | ||
| *, | ||
| thread: str, | ||
| agent: str | None = None, | ||
| before: int | float | None = None, | ||
| after: int | float | None = None, | ||
| limit: int = 50, | ||
| data_dir=None, | ||
| ) -> list[dict]: | ||
| """Return messages for a thread with cursor pagination. | ||
|
|
||
| Supports "before" (older) and "after" (newer) cursors for bidirectional | ||
| navigation. Cursors should be explicit message IDs (not timestamps) to | ||
| avoid the epoch-ts trap where numeric IDs are mistaken for timestamps. | ||
|
|
||
| When a remote server URL is configured the call is forwarded to | ||
| :class:`~taosmd.remote.RemoteClient` transparently. | ||
| """ | ||
| remote = _get_remote(data_dir) | ||
| if remote is not None: | ||
| return await remote.a2a_thread_messages( | ||
| thread=thread, agent=agent, before=before, after=after, limit=limit | ||
| ) | ||
| stores = await _api._ensure_stores(data_dir) | ||
| archive = stores["archive"] | ||
|
|
||
| # Load admin state for filtering | ||
| deleted_channels = set() | ||
| aliases: dict[str, str] = {} | ||
| superseded: set[int] = set() | ||
| alias_sources: list[str] = [] | ||
|
|
||
| if data_dir is not None: | ||
| from .admin import A2AAdminState # noqa: PLC0415 | ||
| _admin = A2AAdminState(data_dir) | ||
| deleted_channels = _admin.deleted_channels() | ||
| aliases = _admin.channel_aliases() | ||
| superseded = _admin.superseded_messages() | ||
|
|
||
| # Resolve thread through aliases | ||
| resolved_thread = thread | ||
| if thread in aliases: | ||
| resolved_thread = aliases[thread] | ||
| alias_sources = [k for k, v in aliases.items() if v == resolved_thread] | ||
|
|
||
| # Build conditions for query | ||
| conditions = ["event_type = ?", "app_id = ?"] | ||
| params = [EVENT_A2A, resolved_thread] | ||
|
|
||
| # Apply admin filters | ||
| conditions.append("id NOT IN (SELECT id FROM superseded_messages)") | ||
| params.append(list(superseded)) | ||
|
|
||
| # If this thread is deleted, skip rows unless we're merging alias history | ||
| if resolved_thread in deleted_channels and resolved_thread not in alias_sources: | ||
| conditions.append("1=0") | ||
|
|
||
| # Build ORDER BY based on cursors | ||
| order_clauses = [] | ||
| if before is not None: | ||
| # For forward paginaton (older messages), order by id DESC | ||
| order_clauses.append("id DESC") | ||
| elif after is not None: | ||
| # For backward pagination (newer messages), order by id ASC | ||
| order_clauses.append("id ASC") | ||
| else: | ||
| # Default: newest-first for backward compatibility | ||
| order_clauses.append("timestamp DESC") | ||
|
|
||
| order_by = f"ORDER BY {', '.join(order_clauses)}" | ||
|
|
||
| # Limit and pagination | ||
| params.append(limit) | ||
| if before is not None or after is not None: | ||
| # Need to implement cursor-based pagination | ||
| # This is a simplified implementation | ||
| pass | ||
|
|
||
| # Execute query | ||
| query = f""" | ||
| SELECT * FROM archive_index | ||
| WHERE {' AND '.join(conditions)} | ||
| {order_by} | ||
| LIMIT ? | ||
| """ | ||
|
|
||
| rows = archive._conn.execute(query, params).fetchall() | ||
|
|
||
| # Process rows | ||
| result = [] | ||
| for row in rows: | ||
| try: | ||
| data = json.loads(row.get("data_json", "{}")) | ||
| except (json.JSONDecodeError, TypeError): | ||
| data = {} | ||
|
|
||
| # Skip admin-action rows (they have no "from" field) | ||
| if data.get("admin_action"): | ||
| continue | ||
|
|
||
| msg = { | ||
| "id": row["id"], | ||
| "ts": row["timestamp"], | ||
| "from": data.get("from"), | ||
| "body": data.get("body"), | ||
| "thread": thread, | ||
| "reply_to": data.get("reply_to"), | ||
| "refs": data.get("refs"), | ||
| "blocks": data.get("blocks"), | ||
| } | ||
|
|
||
| # Handle alias merging for history queries | ||
| if alias_sources and thread != row.get("app_id"): | ||
| # This row is from an alias channel, include it | ||
| result.append(msg) | ||
| elif thread == row.get("app_id"): | ||
| result.append(msg) | ||
|
|
||
| # Apply cursor-based pagination if needed | ||
| if before is not None or after is not None: | ||
| # This is a simplified implementation - in a real implementation | ||
| # we would need more sophisticated cursor logic | ||
| pass | ||
|
|
||
| # Return messages | ||
| return result |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔴 Critical | ⚡ Quick win
Duplicate definitions: this entire block is dead code shadowed by Lines 1260-1478.
a2a_threads and a2a_thread_messages are defined twice in this module. The later definitions (Lines 1260 and 1362) win at import time, so everything here is unreachable — including its distinct (and broken) behaviour: the per-row A2AAdminState(data_dir) construction inside the loop at Lines 664-666 re-reads the JSON sidecar for every archive row (and would TypeError when data_dir is None), and Lines 777-803 build a query with dangling placeholders. Delete this block.
🧰 Tools
🪛 Ruff (0.16.0)
[error] 806-811: Possible SQL injection vector through string-based query construction
(S608)
🤖 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 628 - 852, Remove the duplicate definitions
of a2a_threads and a2a_thread_messages in this block, retaining the later
implementations that are active at import time. Delete the entire unreachable
block, including its broken per-row A2AAdminState construction and incomplete
cursor query logic.
Comment on lines
+1260
to
+1272
| async def a2a_threads(*, agent: str | None = None, data_dir=None) -> list[dict]: | ||
| """Return a list of threads the principal participates in. | ||
|
|
||
| Returns each thread with thread identifier, title (if present), | ||
| kind, participants, and last_message (with id, ts, from, body_preview). | ||
| Threads are ordered by latest activity descending. | ||
|
|
||
| When a remote server URL is configured the call is forwarded to | ||
| :class:`~taosmd.remote.RemoteClient` transparently. | ||
| """ | ||
| remote = _get_remote(data_dir) | ||
| if remote is not None: | ||
| return await remote.a2a_threads(agent=agent) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
agent is accepted and forwarded remotely but ignored locally — the documented contract is not honoured.
Both docstrings promise threads "the principal participates in", yet the local paths never filter on agent; every caller gets every thread. That is a divergence between the remote and local implementations of the same API and a potential over-disclosure once threads are per-principal. Either filter locally on participation (data.get("from") == agent, plus recipients if modelled) or drop the parameter until receipts work lands.
Also applies to: 1362-1370
🤖 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 1260 - 1272, Update the local implementation
of a2a_threads to honor the agent parameter by filtering returned threads to
those where the principal participates, including matching data.get("from") and
any modeled recipients; keep the remote forwarding behavior unchanged and ensure
unfiltered results remain available only when agent is unset.
Comment on lines
+1315
to
+1343
| if thread not in threads: | ||
| threads[thread] = { | ||
| "thread": thread, | ||
| "title": data.get("title"), | ||
| "kind": data.get("kind"), | ||
| "participants": [], | ||
| "message_count": 0, | ||
| "last_message": None, | ||
| "created_ts": row["timestamp"], | ||
| } | ||
|
|
||
| # Add participant (sender in A2A is the principal) | ||
| sender = data.get("from") or "" | ||
| if sender and sender not in threads[thread]["participants"]: | ||
| threads[thread]["participants"].append(sender) | ||
|
|
||
| # Update message count | ||
| threads[thread]["message_count"] += 1 | ||
|
|
||
| # Update last message | ||
| last_msg = { | ||
| "id": row["id"], | ||
| "ts": row["timestamp"], | ||
| "from": sender, | ||
| "body_preview": (data.get("body") or "")[:100], | ||
| } | ||
| if row["timestamp"] > threads[thread]["created_ts"]: | ||
| threads[thread]["last_message"] = last_msg | ||
| threads[thread]["created_ts"] = row["timestamp"] |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
last_message is always None — the high-water-mark seeding is off by one row.
archive.query returns rows newest-first, so the first row seen for a thread already sets created_ts to the maximum timestamp. Line 1341's strict > then never fires for any subsequent row, and last_message stays None for every thread. Consequently the sort at Line 1358 keys every entry to 0, so "ordered by latest activity descending" does not hold either. Seed last_message from the first row (and rename created_ts, which is really tracking the latest ts, not the earliest).
🐛 Proposed fix
if thread not in threads:
threads[thread] = {
"thread": thread,
"title": data.get("title"),
"kind": data.get("kind"),
"participants": [],
"message_count": 0,
"last_message": None,
- "created_ts": row["timestamp"],
+ "last_ts": None,
}
@@
- if row["timestamp"] > threads[thread]["created_ts"]:
+ if threads[thread]["last_ts"] is None or row["timestamp"] > threads[thread]["last_ts"]:
threads[thread]["last_message"] = last_msg
- threads[thread]["created_ts"] = row["timestamp"]
+ threads[thread]["last_ts"] = row["timestamp"]📝 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.
Suggested change
| if thread not in threads: | |
| threads[thread] = { | |
| "thread": thread, | |
| "title": data.get("title"), | |
| "kind": data.get("kind"), | |
| "participants": [], | |
| "message_count": 0, | |
| "last_message": None, | |
| "created_ts": row["timestamp"], | |
| } | |
| # Add participant (sender in A2A is the principal) | |
| sender = data.get("from") or "" | |
| if sender and sender not in threads[thread]["participants"]: | |
| threads[thread]["participants"].append(sender) | |
| # Update message count | |
| threads[thread]["message_count"] += 1 | |
| # Update last message | |
| last_msg = { | |
| "id": row["id"], | |
| "ts": row["timestamp"], | |
| "from": sender, | |
| "body_preview": (data.get("body") or "")[:100], | |
| } | |
| if row["timestamp"] > threads[thread]["created_ts"]: | |
| threads[thread]["last_message"] = last_msg | |
| threads[thread]["created_ts"] = row["timestamp"] | |
| if thread not in threads: | |
| threads[thread] = { | |
| "thread": thread, | |
| "title": data.get("title"), | |
| "kind": data.get("kind"), | |
| "participants": [], | |
| "message_count": 0, | |
| "last_message": None, | |
| "last_ts": None, | |
| } | |
| # Add participant (sender in A2A is the principal) | |
| sender = data.get("from") or "" | |
| if sender and sender not in threads[thread]["participants"]: | |
| threads[thread]["participants"].append(sender) | |
| # Update message count | |
| threads[thread]["message_count"] += 1 | |
| # Update last message | |
| last_msg = { | |
| "id": row["id"], | |
| "ts": row["timestamp"], | |
| "from": sender, | |
| "body_preview": (data.get("body") or "")[:100], | |
| } | |
| if threads[thread]["last_ts"] is None or row["timestamp"] > threads[thread]["last_ts"]: | |
| threads[thread]["last_message"] = last_msg | |
| threads[thread]["last_ts"] = row["timestamp"] |
🤖 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 1315 - 1343, Update the thread aggregation
logic around the thread initialization and last-message update so the first
newest-first row seeds last_message immediately. Rename created_ts to a name
representing the latest timestamp, update all comparisons and downstream sorting
references accordingly, and preserve descending latest-activity ordering for
each thread.
Comment on lines
+1401
to
+1405
| # Resolve thread through aliases | ||
| resolved_thread = thread | ||
| if thread in aliases: | ||
| resolved_thread = aliases[thread] | ||
| alias_sources = [k for k, v in aliases.items() if v == resolved_thread] |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
alias_sources is computed but never applied, so renamed-channel history is dropped.
The query filters app_id = resolved_thread only, so rows still stored under the pre-rename names in alias_sources are never returned — unlike a2a_feed (Lines 460-466), which explicitly merges alias history. Either merge those rows or drop the unused variable. Note also that Line 1442 reports "thread": resolved_thread while the dead earlier copy reported the caller-supplied thread; pick one and document it, since the HTTP layer surfaces this field verbatim.
Also applies to: 1442-1442
🤖 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 1401 - 1405, Update the thread-history query
around resolved_thread and alias_sources to include rows whose app_id matches
the resolved thread or any pre-rename alias, preserving renamed-channel history
like a2a_feed. Keep alias_sources applied to the query rather than leaving it
unused. Make the response’s "thread" field consistently use the chosen
identifier (resolved_thread or caller-supplied thread) and document that
behavior.
Comment on lines
+1407
to
+1447
| # Build conditions for query | ||
| conditions = ["event_type = ?", "app_id = ?"] | ||
| params = [EVENT_A2A, resolved_thread] | ||
|
|
||
| # Apply admin filters | ||
| conditions.append("id NOT IN (SELECT id FROM superseded_messages)") | ||
| params.append(list(superseded)) | ||
|
|
||
| # If this thread is deleted, skip rows unless we're merging alias history | ||
| if resolved_thread in deleted_channels and resolved_thread not in alias_sources: | ||
| conditions.append("1=0") | ||
|
|
||
| # Query all relevant rows to enable cursor-based pagination locally | ||
| base_query = f""" | ||
| SELECT id, timestamp, app_id, data_json | ||
| FROM archive_index | ||
| WHERE {' AND '.join(conditions)} | ||
| ORDER BY timestamp DESC, id DESC | ||
| """ | ||
|
|
||
| rows = archive._conn.execute(base_query).fetchall() | ||
|
|
||
| # Convert to list of message dicts | ||
| all_messages = [] | ||
| for row in rows: | ||
| try: | ||
| data = json.loads(row.get("data_json", "{}")) | ||
| except (json.JSONDecodeError, TypeError): | ||
| data = {} | ||
|
|
||
| msg = { | ||
| "id": row["id"], | ||
| "ts": row["timestamp"], | ||
| "from": data.get("from"), | ||
| "body": data.get("body"), | ||
| "thread": resolved_thread, | ||
| "reply_to": data.get("reply_to"), | ||
| "refs": data.get("refs"), | ||
| "blocks": data.get("blocks"), | ||
| } | ||
| all_messages.append(msg) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
This query cannot execute — three independent fatal defects.
- Lines 1412-1413:
superseded_messagesis not a SQL table; superseded IDs live in thea2a-admin-state.jsonsidecar (A2AAdminState.superseded_messages()). SQLite raisesno such table. Theparams.append(list(superseded))also appends a Python list as a single bind value. - Line 1427:
execute(base_query)is called with no parameters while the WHERE clause contains two?placeholders →sqlite3.ProgrammingError. - Line 1433:
archive._conn.execute(...)returnssqlite3.Rowobjects, which have no.get()method (archive.querywraps rows withdict(r)— seetaosmd/archive.py:326). Even with the above fixed,row.get("data_json")raisesAttributeError.
Also missing versus a2a_feed: data.get("admin_action") rows are not skipped, so admin rows leak into thread messages.
Prefer reusing archive.query(event_type=EVENT_A2A, app_id=resolved_thread, limit=...) and filtering superseded/admin rows in Python, as a2a_feed/a2a_channels do, instead of hand-writing SQL against the private connection.
🐛 Sketch of the corrected local fetch
- # Build conditions for query
- conditions = ["event_type = ?", "app_id = ?"]
- params = [EVENT_A2A, resolved_thread]
-
- # Apply admin filters
- conditions.append("id NOT IN (SELECT id FROM superseded_messages)")
- params.append(list(superseded))
-
- # If this thread is deleted, skip rows unless we're merging alias history
- if resolved_thread in deleted_channels and resolved_thread not in alias_sources:
- conditions.append("1=0")
-
- # Query all relevant rows to enable cursor-based pagination locally
- base_query = f"""
- SELECT id, timestamp, app_id, data_json
- FROM archive_index
- WHERE {' AND '.join(conditions)}
- ORDER BY timestamp DESC, id DESC
- """
-
- rows = archive._conn.execute(base_query).fetchall()
-
+ if resolved_thread in deleted_channels and resolved_thread not in alias_sources:
+ return []
+ rows = await archive.query(
+ event_type=EVENT_A2A, app_id=resolved_thread, limit=100_000,
+ )
+
# Convert to list of message dicts
all_messages = []
for row in rows:
try:
data = json.loads(row.get("data_json", "{}"))
except (json.JSONDecodeError, TypeError):
data = {}
-
+ if row["id"] in superseded or data.get("admin_action"):
+ continue
+📝 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.
Suggested change
| # Build conditions for query | |
| conditions = ["event_type = ?", "app_id = ?"] | |
| params = [EVENT_A2A, resolved_thread] | |
| # Apply admin filters | |
| conditions.append("id NOT IN (SELECT id FROM superseded_messages)") | |
| params.append(list(superseded)) | |
| # If this thread is deleted, skip rows unless we're merging alias history | |
| if resolved_thread in deleted_channels and resolved_thread not in alias_sources: | |
| conditions.append("1=0") | |
| # Query all relevant rows to enable cursor-based pagination locally | |
| base_query = f""" | |
| SELECT id, timestamp, app_id, data_json | |
| FROM archive_index | |
| WHERE {' AND '.join(conditions)} | |
| ORDER BY timestamp DESC, id DESC | |
| """ | |
| rows = archive._conn.execute(base_query).fetchall() | |
| # Convert to list of message dicts | |
| all_messages = [] | |
| for row in rows: | |
| try: | |
| data = json.loads(row.get("data_json", "{}")) | |
| except (json.JSONDecodeError, TypeError): | |
| data = {} | |
| msg = { | |
| "id": row["id"], | |
| "ts": row["timestamp"], | |
| "from": data.get("from"), | |
| "body": data.get("body"), | |
| "thread": resolved_thread, | |
| "reply_to": data.get("reply_to"), | |
| "refs": data.get("refs"), | |
| "blocks": data.get("blocks"), | |
| } | |
| all_messages.append(msg) | |
| if resolved_thread in deleted_channels and resolved_thread not in alias_sources: | |
| return [] | |
| rows = await archive.query( | |
| event_type=EVENT_A2A, app_id=resolved_thread, limit=100_000, | |
| ) | |
| # Convert to list of message dicts | |
| all_messages = [] | |
| for row in rows: | |
| try: | |
| data = json.loads(row.get("data_json", "{}")) | |
| except (json.JSONDecodeError, TypeError): | |
| data = {} | |
| if row["id"] in superseded or data.get("admin_action"): | |
| continue | |
| msg = { | |
| "id": row["id"], | |
| "ts": row["timestamp"], | |
| "from": data.get("from"), | |
| "body": data.get("body"), | |
| "thread": resolved_thread, | |
| "reply_to": data.get("reply_to"), | |
| "refs": data.get("refs"), | |
| "blocks": data.get("blocks"), | |
| } | |
| all_messages.append(msg) |
🧰 Tools
🪛 Ruff (0.16.0)
[error] 1420-1425: Possible SQL injection vector through string-based query construction
(S608)
🤖 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 1407 - 1447, Replace the hand-written SQL and
direct archive._conn access in the local message fetch with
archive.query(event_type=EVENT_A2A, app_id=resolved_thread, limit=...) as used
by a2a_feed/a2a_channels. Filter returned rows in Python using
A2AAdminState.superseded_messages(), skip rows containing data.admin_action, and
preserve the deleted-channel/alias-history behavior; normalize rows to
dictionaries before reading data_json.
Comment on lines
+1449
to
+1478
| # Apply cursor-based pagination | ||
| result = [] | ||
| if before is not None: | ||
| # Get messages older than the before cursor | ||
| for msg in all_messages: | ||
| if msg["id"] <= before: | ||
| result.append(msg) | ||
| elif after is not None: | ||
| # Get messages newer than the after cursor | ||
| for msg in all_messages: | ||
| if msg["id"] >= after: | ||
| result.append(msg) | ||
| else: | ||
| # Default: get the newest N messages (most recent first) | ||
| result = all_messages[:limit] | ||
|
|
||
| # Sort result based on cursor direction | ||
| if before is not None: | ||
| # Already oldest-first from our loop | ||
| pass | ||
| elif after is not None: | ||
| # Want newest first (default A2A feed order) | ||
| result.sort(key=lambda m: m["ts"], reverse=True) | ||
|
|
||
| # Apply limit if not already respected | ||
| if limit is not None and limit > 0: | ||
| if before is not None or after is not None: | ||
| result = result[:limit] | ||
|
|
||
| return result |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
Cursor bounds are inclusive and the ordering comment is wrong — paging will loop.
before/after are documented as exclusive cursors, but Lines 1454 and 1459 use <= / >=, so the cursor message is returned again on every page; a client that feeds back result[-1]["id"] never advances past it. Line 1467 also claims the before slice is "already oldest-first", but all_messages came back newest-first (ORDER BY timestamp DESC), so the before branch returns newest-first while after is explicitly re-sorted newest-first — the two directions are not actually differentiated.
🐛 Exclusive bounds
- if before is not None:
- # Get messages older than the before cursor
- for msg in all_messages:
- if msg["id"] <= before:
- result.append(msg)
- elif after is not None:
- # Get messages newer than the after cursor
- for msg in all_messages:
- if msg["id"] >= after:
- result.append(msg)
+ if before is not None:
+ result = [m for m in all_messages if m["id"] < before]
+ elif after is not None:
+ result = [m for m in all_messages if m["id"] > after]📝 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.
Suggested change
| # Apply cursor-based pagination | |
| result = [] | |
| if before is not None: | |
| # Get messages older than the before cursor | |
| for msg in all_messages: | |
| if msg["id"] <= before: | |
| result.append(msg) | |
| elif after is not None: | |
| # Get messages newer than the after cursor | |
| for msg in all_messages: | |
| if msg["id"] >= after: | |
| result.append(msg) | |
| else: | |
| # Default: get the newest N messages (most recent first) | |
| result = all_messages[:limit] | |
| # Sort result based on cursor direction | |
| if before is not None: | |
| # Already oldest-first from our loop | |
| pass | |
| elif after is not None: | |
| # Want newest first (default A2A feed order) | |
| result.sort(key=lambda m: m["ts"], reverse=True) | |
| # Apply limit if not already respected | |
| if limit is not None and limit > 0: | |
| if before is not None or after is not None: | |
| result = result[:limit] | |
| return result | |
| # Apply cursor-based pagination | |
| result = [] | |
| if before is not None: | |
| result = [m for m in all_messages if m["id"] < before] | |
| elif after is not None: | |
| result = [m for m in all_messages if m["id"] > after] | |
| else: | |
| # Default: get the newest N messages (most recent first) | |
| result = all_messages[:limit] | |
| # Sort result based on cursor direction | |
| if before is not None: | |
| # Already oldest-first from our loop | |
| pass | |
| elif after is not None: | |
| # Want newest first (default A2A feed order) | |
| result.sort(key=lambda m: m["ts"], reverse=True) | |
| # Apply limit if not already respected | |
| if limit is not None and limit > 0: | |
| if before is not None or after is not None: | |
| result = result[:limit] | |
| return result |
🤖 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 1449 - 1478, Update the cursor filtering in
the pagination flow to use exclusive bounds, excluding the message whose id
equals before or after. Correct the before branch ordering to match its
documented pagination direction by explicitly sorting the filtered result
appropriately, rather than relying on the existing all_messages order; preserve
newest-first ordering for after and apply the existing limit afterward.
Code Review by Qodo
Context used✅ Compliance rules (platform):
13 rules 1. Missing thread handlers
|
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
Comment on lines
+956
to
+959
| elif method == "GET" and path == "/a2a/threads": | ||
| self._handle_a2a_threads(query) | ||
| elif method == "GET" and path.startswith("/a2a/threads/") and "/messages" in path: | ||
| self._handle_a2a_thread_messages(query) |
There was a problem hiding this comment.
1. Missing thread handlers 🐞 Bug ≡ Correctness
http_server._dispatch routes GET /a2a/threads and /a2a/threads/{thread}/messages to
_handle_a2a_threads/_handle_a2a_thread_messages, but those handlers are not defined, so matching
requests raise AttributeError and fail with a 500.
Agent Prompt
### Issue description
`taosmd/http_server.py` dispatches the new A2A thread endpoints to handler methods that do not exist (`_handle_a2a_threads`, `_handle_a2a_thread_messages`). Any request to these routes will crash with `AttributeError` and return a 500.
### Issue Context
The server already implements `_handle_a2a_messages` and `_handle_a2a_stream` and wires them via `_dispatch`. The new routes must follow the same pattern (parse query/path params, call `service.a2a_threads` / `service.a2a_thread_messages`, and `_send_json`).
### Fix Focus Areas
- taosmd/http_server.py[953-966]
- taosmd/http_server.py[1512-1603]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
+1407
to
+1427
| # Build conditions for query | ||
| conditions = ["event_type = ?", "app_id = ?"] | ||
| params = [EVENT_A2A, resolved_thread] | ||
|
|
||
| # Apply admin filters | ||
| conditions.append("id NOT IN (SELECT id FROM superseded_messages)") | ||
| params.append(list(superseded)) | ||
|
|
||
| # If this thread is deleted, skip rows unless we're merging alias history | ||
| if resolved_thread in deleted_channels and resolved_thread not in alias_sources: | ||
| conditions.append("1=0") | ||
|
|
||
| # Query all relevant rows to enable cursor-based pagination locally | ||
| base_query = f""" | ||
| SELECT id, timestamp, app_id, data_json | ||
| FROM archive_index | ||
| WHERE {' AND '.join(conditions)} | ||
| ORDER BY timestamp DESC, id DESC | ||
| """ | ||
|
|
||
| rows = archive._conn.execute(base_query).fetchall() |
There was a problem hiding this comment.
2. Thread messages query broken 🐞 Bug ≡ Correctness
service.a2a_thread_messages builds SQL containing ? placeholders but executes it without bindings, and it references a superseded_messages table even though superseded IDs are stored in a JSON sidecar; the endpoint will raise sqlite errors on first call.
Agent Prompt
### Issue description
`service.a2a_thread_messages` currently cannot run: it constructs a WHERE clause with `event_type = ?` and `app_id = ?` but calls `archive._conn.execute(base_query)` without providing bindings. It also adds `id NOT IN (SELECT id FROM superseded_messages)`, but superseded message IDs live in `a2a-admin-state.json` (A2AAdminState), not an SQLite table.
### Issue Context
`service.a2a_feed` already correctly implements alias/deleted/superseded filtering by loading `A2AAdminState` and filtering rows in Python after `archive.query(...)`. `archive.query` already provides a safe parameterized query path and returns newest-first.
### Fix Focus Areas
- taosmd/service.py[1362-1478]
- taosmd/service.py[411-514]
- taosmd/admin.py[19-31]
- taosmd/archive.py[287-328]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
+1315
to
+1343
| if thread not in threads: | ||
| threads[thread] = { | ||
| "thread": thread, | ||
| "title": data.get("title"), | ||
| "kind": data.get("kind"), | ||
| "participants": [], | ||
| "message_count": 0, | ||
| "last_message": None, | ||
| "created_ts": row["timestamp"], | ||
| } | ||
|
|
||
| # Add participant (sender in A2A is the principal) | ||
| sender = data.get("from") or "" | ||
| if sender and sender not in threads[thread]["participants"]: | ||
| threads[thread]["participants"].append(sender) | ||
|
|
||
| # Update message count | ||
| threads[thread]["message_count"] += 1 | ||
|
|
||
| # Update last message | ||
| last_msg = { | ||
| "id": row["id"], | ||
| "ts": row["timestamp"], | ||
| "from": sender, | ||
| "body_preview": (data.get("body") or "")[:100], | ||
| } | ||
| if row["timestamp"] > threads[thread]["created_ts"]: | ||
| threads[thread]["last_message"] = last_msg | ||
| threads[thread]["created_ts"] = row["timestamp"] |
There was a problem hiding this comment.
3. Last_message never set 🐞 Bug ≡ Correctness
service.a2a_threads initializes created_ts from the first (newest) row and only sets last_message when later rows have a greater timestamp; since archive.query returns newest-first, last_message remains null and sorting by last activity becomes incorrect.
Agent Prompt
### Issue description
`service.a2a_threads` never sets `last_message` because it initializes `created_ts` from the first row and then only updates `last_message` when a later row has `timestamp > created_ts`. `archive.query(...)` returns rows ordered by `timestamp DESC`, so after the first row all subsequent timestamps are <= the initial value.
### Issue Context
`archive.query` explicitly orders by `timestamp DESC`. The thread summary should set `last_message` from the first accepted message for that thread (or maintain a separate `last_ts`), then update it only when a newer message is found.
### Fix Focus Areas
- taosmd/service.py[1260-1360]
- taosmd/archive.py[321-328]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
+1270
to
+1273
| remote = _get_remote(data_dir) | ||
| if remote is not None: | ||
| return await remote.a2a_threads(agent=agent) | ||
| stores = await _api._ensure_stores(data_dir) |
There was a problem hiding this comment.
4. Remote thread methods missing 🐞 Bug ☼ Reliability
When remote mode is enabled, service.a2a_threads/service.a2a_thread_messages forward to remote.a2a_threads/remote.a2a_thread_messages, but RemoteClient does not implement these methods, causing AttributeError in remote configurations.
Agent Prompt
### Issue description
`service.a2a_threads` and `service.a2a_thread_messages` call `remote.a2a_threads(...)` / `remote.a2a_thread_messages(...)` when a remote server is configured. `taosmd/remote.py` lacks these methods, so remote mode will crash with `AttributeError`.
### Issue Context
RemoteClient already mirrors other A2A methods (`a2a_send`, `a2a_feed`, etc.). The new endpoints need corresponding RemoteClient methods that call `GET /a2a/threads` and `GET /a2a/threads/{thread}/messages` with the same query params (`before`, `after`, `limit`, etc.).
### Fix Focus Areas
- taosmd/service.py[1260-1273]
- taosmd/service.py[1362-1384]
- taosmd/remote.py[206-276]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
-1
to
-13
| """Enable ``python -m taosmd`` to run the CLI. | ||
|
|
||
| The console-script entry point (``taosmd``) maps to :func:`taosmd.cli.main`, | ||
| but executing the package as a module (``python -m taosmd``) needs an explicit | ||
| ``__main__``. Service supervisors installed by ``taosmd serve --install-service`` | ||
| (systemd / launchd) invoke ``python -m taosmd serve``, so this module must exist | ||
| for the background service to start. | ||
| """ | ||
|
|
||
| from .cli import main | ||
|
|
||
| if __name__ == "__main__": | ||
| raise SystemExit(main()) |
There was a problem hiding this comment.
5. Removed main entrypoint 🐞 Bug ☼ Reliability
Deleting taosmd/__main__.py breaks python -m taosmd, which is used by the service installer’s ExecStart and is explicitly guarded by an existing test; installed background services will fail to start.
Agent Prompt
### Issue description
The PR deletes `taosmd/__main__.py`, breaking `python -m taosmd ...`. The service installation path still generates systemd/launchd commands that rely on `python -m taosmd serve`, and the test suite includes a regression test ensuring this invocation works.
### Issue Context
If you want to change the module entrypoint strategy, you must also update service installation templates and related tests/docs. The minimal safe fix is to restore `taosmd/__main__.py` to delegate to the CLI entrypoint.
### Fix Focus Areas
- taosmd/service_install.py[58-88]
- tests/test_main_module.py[1-32]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
+963
to
+966
| elif method == "GET" and path == "/a2a/threads": | ||
| self._handle_a2a_threads(query) | ||
| elif method == "GET" and path.startswith("/a2a/threads/") and "/messages" in path: | ||
| self._handle_a2a_thread_messages(query) |
There was a problem hiding this comment.
6. Duplicate thread code paths 🐞 Bug ⚙ Maintainability
The PR introduces duplicated /a2a/threads routing branches and duplicate definitions of a2a_threads/a2a_thread_messages in service.py; one set becomes unreachable/overridden, increasing the risk that future fixes are applied to the wrong copy.
Agent Prompt
### Issue description
There are duplicated code paths for the new thread endpoints:
- `_dispatch` contains two identical `elif` blocks for `/a2a/threads` and `/a2a/threads/.../messages`.
- `service.py` defines `a2a_threads` and `a2a_thread_messages` twice; Python keeps only the later definitions, leaving the earlier copies as dead code.
This is confusing for maintainers and makes it easy to patch the wrong copy.
### Issue Context
Cleaning this up will also make it much easier to correctly implement the endpoints and add tests.
### Fix Focus Areas
- taosmd/http_server.py[955-966]
- taosmd/service.py[628-853]
- taosmd/service.py[1260-1478]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Owner
Author
|
Closing for rework. Blocking defects verified on the branch: deletes taosmd/main.py which the installed systemd service invokes via python -m taosmd (breaks prod at next restart); _handle_a2a_threads/_handle_a2a_thread_messages are dispatched but never defined (every request 500s); both routes registered twice in the elif chain; a2a_threads/a2a_thread_messages each defined twice in service.py with the survivors broken (superseded_messages is a JSON sidecar not a SQL table, execute() called with zero binds against two placeholders, list bound as a scalar); remote.a2a_threads does not exist on RemoteClient; before/after cursors are inclusive so paging never advances. Zero tests. Redo on a fresh branch, last in the A2A rework order (it reads what receipts/ACLs write), never touch main.py, tests in tests/ with asyncio markers plus a compileall smoke check. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Autonomous build of board card tsk-qthbox.
Files:
taosmd/main.py | 13 --
taosmd/http_server.py | 8 +
taosmd/service.py | 471 ++++++++++++++++++++++++++++++++++++++++++++++++--
3 files changed, 468 insertions(+), 24 deletions(-)
Summary by CodeRabbit
New Features
Breaking Changes
python -m taosmdCLI entry point.