From 0ab97129b214af0bfce109cd28e5947d28d33c83 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Thu, 13 Aug 2026 14:15:49 +0000 Subject: [PATCH 1/4] fix(a2a): stop the bus read path returning 200 with silence Three distinct ways GET /api/a2a/bus/messages answered "success, nothing here" when the truth was "your request was wrong", all measured against the live proxy: channel=all -> 200, zero messages channel=doesnotexist -> 200, zero messages (byte-identical) since_id=2430 -> silently dropped; 500 messages from id 1890 `all` is the all-threads idiom the raw bus and `taosmd a2a-watch` document, and it exists precisely so a reader cannot miss a thread created after it started. On this proxy it was forwarded as a thread literally named "all", matched nothing, and returned 200. An agent following our own onboarding guide against the path we are about to recommend to every new agent got a permanently silent bus and a success code confirming it. The cursor case is the same shape: an ignored param is indistinguishable from one that works, so an incremental reader re-read the whole window on every poll while believing it held a cursor. - `all` and `*` read every thread (spelled "omit thread" on the bus) - an unrecognised query param is a 400 naming the accepted set, never a silent no-op - an empty result for a NAMED channel reports channel_known, so a typo is distinguishable from a quiet channel; the probe fails OPEN so an unreachable bus never accuses the caller of a typo - `thread` accepted as an alias for `channel` (the raw bus's own name) - `since` documented and validated as a message ts, not an id Reported by @taOSmd-dev while verifying the authenticated read path. Their report also said `since=` was ignored; measured, it is not -- the raw bus does honour it, and the test asserting that passes both before and after, so it is deliberately not counted among the fixes. All 7 discriminating tests fail against the unfixed route. --- changelog.d/bus-read-silent-empty.md | 1 + tests/test_a2a_bus.py | 131 +++++++++++++++++++++++++++ tinyagentos/routes/a2a_bus.py | 118 +++++++++++++++++++----- 3 files changed, 228 insertions(+), 22 deletions(-) create mode 100644 changelog.d/bus-read-silent-empty.md diff --git a/changelog.d/bus-read-silent-empty.md b/changelog.d/bus-read-silent-empty.md new file mode 100644 index 000000000..e71113aa7 --- /dev/null +++ b/changelog.d/bus-read-silent-empty.md @@ -0,0 +1 @@ +- Fixed three ways `GET /api/a2a/bus/messages` returned HTTP 200 and nothing, leaving a reader silently disconnected: `channel=all` (the idiom the raw bus and `taosmd a2a-watch` document for "every thread") was forwarded as a channel literally named `all` and matched nothing; an unknown channel name was indistinguishable from a quiet one; and unrecognised cursor params such as `since_id` were silently dropped, so an incremental reader re-read the whole window every poll believing it held a cursor. `all` and `*` now read every thread, an unrecognised query param is a 400 naming the accepted set, an empty result for a named channel reports `channel_known`, and `thread` is accepted as an alias for `channel`. diff --git a/tests/test_a2a_bus.py b/tests/test_a2a_bus.py index 7c7533693..1ebd8002c 100644 --- a/tests/test_a2a_bus.py +++ b/tests/test_a2a_bus.py @@ -135,3 +135,134 @@ def httpx_connect_error(): import httpx return httpx.ConnectError("connection refused") + + +# --------------------------------------------------------------------------- +# Read-path ambiguity: silence that reads as success +# +# Each of these was measured against the LIVE proxy before the fix. The shared +# failure shape is a 200 that carries no information: an agent following our own +# onboarding guide got a permanently quiet bus and a success code confirming it. +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +@respx.mock +async def test_channel_all_reads_every_thread(client): + """channel=all must read ALL threads, not a channel literally named "all". + + `all` is the idiom the raw bus and `taosmd a2a-watch` document, and it exists + so a reader cannot miss a thread created after it started. Before the fix the + proxy forwarded it as thread=all, the bus had no such thread, and the caller + got HTTP 200 with zero messages -- forever. + """ + route = respx.get(f"{_BUS}/a2a/messages").mock( + return_value=Response(200, json={"messages": [ + {"id": 7, "ts": 10.0, "from": "@a", "body": "x", "thread": "build", "reply_to": None}, + ]}) + ) + resp = await client.get("/api/a2a/bus/messages", params={"channel": "all"}) + assert resp.status_code == 200 + assert resp.json()["messages"][0]["id"] == 7 + # All-threads is spelled "omit the thread param" on the bus. + assert "thread" not in route.calls.last.request.url.params + + +@pytest.mark.asyncio +@respx.mock +async def test_wildcard_channel_reads_every_thread(client): + """`*` means the same as `all` and must not be rejected.""" + route = respx.get(f"{_BUS}/a2a/messages").mock( + return_value=Response(200, json={"messages": []}) + ) + resp = await client.get("/api/a2a/bus/messages", params={"channel": "*"}) + assert resp.status_code == 200 + assert "thread" not in route.calls.last.request.url.params + # All-threads empty is genuinely empty; there is no channel name to doubt. + assert "channel_known" not in resp.json() + + +@pytest.mark.asyncio +@respx.mock +async def test_unknown_channel_is_distinguishable_from_empty(client): + """A typo'd channel and a quiet channel must not look identical. + + Before the fix `channel=doesnotexist` and `channel=build` with nothing new + both returned exactly {"messages": [], "available": true}. + """ + respx.get(f"{_BUS}/a2a/messages").mock(return_value=Response(200, json={"messages": []})) + respx.get(f"{_BUS}/a2a/channels").mock( + return_value=Response(200, json={"channels": [{"channel": "build"}]}) + ) + + unknown = await client.get("/api/a2a/bus/messages", params={"channel": "doesnotexist"}) + assert unknown.status_code == 200 + assert unknown.json()["channel_known"] is False + + known = await client.get("/api/a2a/bus/messages", params={"channel": "build"}) + assert known.status_code == 200 + assert known.json()["channel_known"] is True + + +@pytest.mark.asyncio +@respx.mock +async def test_channel_probe_fails_open_when_bus_list_unreachable(client): + """If the channel list cannot be fetched, do not accuse the caller of a typo.""" + respx.get(f"{_BUS}/a2a/messages").mock(return_value=Response(200, json={"messages": []})) + respx.get(f"{_BUS}/a2a/channels").mock(side_effect=httpx_connect_error()) + + resp = await client.get("/api/a2a/bus/messages", params={"channel": "build"}) + assert resp.status_code == 200 + assert resp.json()["channel_known"] is True + + +@pytest.mark.asyncio +async def test_unknown_query_param_is_400_not_a_silent_noop(client): + """An ignored cursor param is indistinguishable from one that works. + + Measured live before the fix: `since_id=2430` was silently dropped and the + endpoint returned 500 messages starting at id 1890, so an incremental reader + re-read the whole window every poll believing it held a cursor. + """ + for bad in ("since_id", "after", "from_id"): + resp = await client.get("/api/a2a/bus/messages", params={"channel": "build", bad: "2430"}) + assert resp.status_code == 400, f"{bad} was accepted" + body = resp.json() + assert bad in body["error"] + assert "since" in body["hint"] + + +@pytest.mark.asyncio +@respx.mock +async def test_thread_is_accepted_as_an_alias_for_channel(client): + """`thread` is the raw bus's own name for this; accept it rather than 400.""" + route = respx.get(f"{_BUS}/a2a/messages").mock( + return_value=Response(200, json={"messages": [ + {"id": 1, "ts": 1.0, "from": "@a", "body": "x", "thread": "ops", "reply_to": None}, + ]}) + ) + resp = await client.get("/api/a2a/bus/messages", params={"thread": "ops"}) + assert resp.status_code == 200 + assert route.calls.last.request.url.params["thread"] == "ops" + + +@pytest.mark.asyncio +async def test_since_rejects_an_id_shaped_cursor(client): + """`since` is a message ts, not an id. Say so instead of quietly mis-reading.""" + resp = await client.get( + "/api/a2a/bus/messages", params={"channel": "build", "since": "not-a-ts"} + ) + assert resp.status_code == 400 + assert "ts" in resp.json()["error"] + + +@pytest.mark.asyncio +@respx.mock +async def test_since_is_forwarded_as_the_cursor(client): + """A valid ts cursor reaches the bus (the raw bus does honour it).""" + route = respx.get(f"{_BUS}/a2a/messages").mock( + return_value=Response(200, json={"messages": []}) + ) + await client.get( + "/api/a2a/bus/messages", params={"channel": "build", "since": "1786630185.75"} + ) + assert route.calls.last.request.url.params["since"] == "1786630185.75" diff --git a/tinyagentos/routes/a2a_bus.py b/tinyagentos/routes/a2a_bus.py index 0e17920f9..120b26d81 100644 --- a/tinyagentos/routes/a2a_bus.py +++ b/tinyagentos/routes/a2a_bus.py @@ -100,39 +100,78 @@ async def bus_channels(request: Request): return {"channels": channels, "available": True} +# Query params this endpoint understands. Anything else is a 400 rather than a +# silent no-op: an ignored cursor param is indistinguishable from a cursor that +# works, so a reader that passes `since_id=` believes it is reading incrementally +# while it re-reads the whole window forever. Measured on the live proxy before +# this changed: `since_id=2430` returned 500 messages starting at id 1890. +_MESSAGES_PARAMS = frozenset({"channel", "thread", "limit", "since"}) + +# Selectors meaning "every thread", not one named channel. `all` is the idiom +# the raw bus and `taosmd a2a-watch` document, and it exists precisely so a +# reader cannot miss a thread created after it started. Forwarded here as "omit +# the thread param", which is how the bus itself spells all-threads. +_ALL_CHANNELS = frozenset({"all", "*"}) + + @router.get("/api/a2a/bus/messages") -async def bus_messages( - request: Request, - channel: str = "", - limit: int = 100, - since: float | None = None, -): - """Read messages from one bus channel, oldest-first as the bus returns them. +async def bus_messages(request: Request): + """Read messages from the bus, oldest-first as the bus returns them. Authorized readers: an admin session, the host local token, or an active agent registry JWT holding the ``a2a_receive`` scope. - ``channel`` is required and maps to the bus ``thread`` query param; ``*`` is - rejected here (it is only meaningful as an all-threads selector on the - stream endpoint). ``limit`` is clamped to 1..500. ``since`` is forwarded - verbatim to the bus as the cursor (a message ``ts``); the bus replays - everything after it, so an agent can resume from the highest ``ts`` it has - processed. On a bus error this returns an empty list with ``available: - false`` and HTTP 200. + ``channel`` is required and maps to the bus ``thread`` query param; + ``thread`` is accepted as an alias because that is the raw bus's own name + for it. ``channel=all`` (or ``*``) reads every thread. ``limit`` is clamped + to 1..500. ``since`` is the cursor and is a message ``ts`` (a float), NOT an + id -- it is forwarded verbatim and the bus replays everything after it. Any + other query param is a 400. On a bus error this returns an empty list with + ``available: false`` and HTTP 200. + + A named channel that the bus does not know is reported as ``channel_known: + false`` alongside the empty list. Channels on this bus exist only once + something has been posted to them, so an unknown name and a channel nobody + has written to yet are the same state -- but a reader that quietly gets a + 200 and nothing else cannot tell a typo from a quiet channel, and stays + silent forever believing it is connected. """ await _authorize_bus_read(request) - if not channel: - return JSONResponse({"error": "channel required"}, status_code=400) - if channel == "*": + + unknown = sorted(set(request.query_params.keys()) - _MESSAGES_PARAMS) + if unknown: return JSONResponse( - {"error": "wildcard channel not supported here; use /api/a2a/bus/stream for all-threads"}, + { + "error": f"unknown query parameter(s): {', '.join(unknown)}", + "accepted": sorted(_MESSAGES_PARAMS), + "hint": "the cursor is 'since' and takes a message ts (float), not an id", + }, status_code=400, ) + channel = request.query_params.get("channel") or request.query_params.get("thread") or "" + if not channel: + return JSONResponse({"error": "channel required"}, status_code=400) + + try: + limit = int(request.query_params.get("limit", 100)) + except ValueError: + return JSONResponse({"error": "limit must be an integer"}, status_code=400) limit = max(1, min(500, limit)) - params: dict = {"thread": channel, "limit": limit} - if since is not None: - params["since"] = since + + since_raw = request.query_params.get("since") + params: dict = {"limit": limit} + if channel not in _ALL_CHANNELS: + params["thread"] = channel + if since_raw is not None: + try: + params["since"] = float(since_raw) + except ValueError: + return JSONResponse( + {"error": "since must be a message ts (float), not an id"}, + status_code=400, + ) + bus = _bus_url() try: async with httpx.AsyncClient(timeout=5.0) as client: @@ -147,7 +186,42 @@ async def bus_messages( return JSONResponse({"messages": [], "available": False}, status_code=200) messages = data.get("messages", []) if isinstance(data, dict) else [] - return {"messages": messages, "available": True} + body: dict = {"messages": messages, "available": True} + + # Only pay for the channel lookup when the answer is empty AND a specific + # channel was named -- that is the only case where "no messages" is + # ambiguous, and it keeps the normal read at one bus call. + if not messages and channel not in _ALL_CHANNELS: + body["channel_known"] = await _channel_exists(bus, channel) + return body + + +async def _channel_exists(bus: str, channel: str) -> bool: + """True if *channel* appears in the bus channel list. + + Fails OPEN (returns True) when the channel list cannot be fetched: an + unreachable bus must not be reported to the caller as "your channel name is + wrong", which would send them chasing a typo that does not exist. + """ + try: + async with httpx.AsyncClient(timeout=5.0) as client: + resp = await client.get(f"{bus}/a2a/channels") + resp.raise_for_status() + data = resp.json() + except Exception as exc: # noqa: BLE001 + logger.warning("A2A bus channel probe failed (%s): %s", bus, exc) + return True + channels = data.get("channels", []) if isinstance(data, dict) else [] + # The bus spells it "channel" in this payload (verified against the live bus: + # {"channels":[{"channel":"build","members":[...],...}]}) while the SAME + # concept is "thread" on /a2a/messages. Accept both rather than trust one + # spelling; an empty name set here would silently report every channel as + # unknown, which is the exact false alarm this function exists to avoid. + names = { + (c.get("channel") or c.get("name") or c.get("thread")) if isinstance(c, dict) else c + for c in channels + } + return channel in names @router.get("/api/a2a/bus/stream") From 2d95618a86084fc2fc8f2c03cb4cfb29e2782867 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Thu, 13 Aug 2026 14:41:42 +0000 Subject: [PATCH 2/4] fix(a2a): align channel=* with the stream endpoint and document the read contract The existing test asserted bus_messages 400s on channel=*, with the rationale 'all-threads is stream-only'. That was true only because bus_messages had not implemented all-threads, not because reading every thread here was unwanted: the stream endpoint has always accepted * and forwarded no thread param. The inconsistency pushed callers toward 'all', which silently matched a thread literally named 'all' and returned an empty 200 forever. Also documents the read contract in docs/agent-coordination.md, since this is the path every new agent is told to use. --- docs/agent-coordination.md | 17 +++++++++++++++++ tests/test_routes_a2a_bus_stream.py | 16 +++++++++++++--- 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/docs/agent-coordination.md b/docs/agent-coordination.md index d5504c719..d3170acd4 100644 --- a/docs/agent-coordination.md +++ b/docs/agent-coordination.md @@ -31,6 +31,23 @@ The CI gate merges. Open the PR, let required checks run, and let the gate merge If you cannot proceed, post `[BLOCKED] ` on the coordination bus. Do not guess or silently work around a blocker. +## Reading the bus + +Read through the controller with your own registry token, not the raw bus port: +`GET /api/a2a/bus/messages?channel=` with `Authorization: Bearer `. + +- `channel=all` (or `*`) reads **every** thread. Use it unless you deliberately want one + channel. A named channel cannot show you a thread created after you started watching. +- `since` is the cursor and takes a message **ts** (a float), not an id. Passing an id + reads as a 1970 timestamp and quietly returns everything, every poll. +- Any other query param is a `400`. An unrecognised cursor param is never silently + ignored, because an ignored cursor is indistinguishable from one that works. +- An empty result for a **named** channel carries `channel_known`. If it is `false`, the + channel name is wrong; a quiet channel and a typo are otherwise identical. + +If the bus is silent, check `channel_known` and your cursor before concluding nobody is +talking. A read that returns `200` with nothing is the failure mode that looks like peace. + ## Identity rules Work as jaylfc. Do not add AI attribution to commits, PRs, or issues. Do not use em dashes in any output. diff --git a/tests/test_routes_a2a_bus_stream.py b/tests/test_routes_a2a_bus_stream.py index 12396b449..7bf690421 100644 --- a/tests/test_routes_a2a_bus_stream.py +++ b/tests/test_routes_a2a_bus_stream.py @@ -278,15 +278,25 @@ async def test_messages_forwards_since(self, agent_app, client): assert call_kwargs["params"]["thread"] == "general" assert float(call_kwargs["params"]["since"]) == 42.0 - async def test_messages_rejects_wildcard_channel(self, agent_app, client): - """bus_messages rejects channel=* (all-threads is stream-only).""" + async def test_messages_wildcard_channel_reads_all_threads(self, agent_app, client): + """bus_messages treats channel=* as all-threads, matching the stream endpoint. + + This previously asserted a 400 with the rationale "all-threads is + stream-only" -- true only because bus_messages had not implemented + all-threads, not because reading every thread here was unwanted. The + stream endpoint has always accepted `*` and forwarded no thread param + (see test_stream_wildcard_channel_all_threads above), so rejecting the + same selector on the sibling read endpoint was an inconsistency that + pushed callers toward `all` -- which silently matched a thread literally + named "all" and returned an empty 200 forever. + """ _, token = await _mint_agent(agent_app, scopes=("a2a_receive",)) resp = await client.get( "/api/a2a/bus/messages", params={"channel": "*"}, headers={"Authorization": f"Bearer {token}"}, ) - assert resp.status_code == 400 + assert resp.status_code == 200 @pytest.mark.asyncio From a17183d03e925e05a01245d90ca5fa02140dc7a2 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Thu, 13 Aug 2026 14:49:52 +0000 Subject: [PATCH 3/4] ci(deleted-symbols-gate): retrigger on PR body edits so the waiver trailer works The guard documents a "Removes-Intentionally:" trailer as the way to waive a deliberate deletion, but the workflow only listened for opened/synchronize/ reopened. Adding the trailer by editing the PR body therefore never re-ran the gate, and re-running the failed job replays the stale event payload carrying the old body -- so the waiver was unreachable without an unrelated code push. store-wiring-gate.yml already carries this exact line and comment for the same reason. Proven both ways against this PR's own violation: without PR_BODY the script exits 1 naming the symbol, with the trailer it prints the waiver and exits 0. --- .github/workflows/deleted-symbols-gate.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/deleted-symbols-gate.yml b/.github/workflows/deleted-symbols-gate.yml index 239abb1de..76f534df0 100644 --- a/.github/workflows/deleted-symbols-gate.yml +++ b/.github/workflows/deleted-symbols-gate.yml @@ -14,6 +14,9 @@ name: Deleted symbols gate on: pull_request: + # "edited" so a waiver trailer added by editing the PR body retriggers the + # gate (a re-run replays the stale event payload with the old body). + types: [opened, synchronize, reopened, edited] branches: [master, dev] jobs: From 86dab57edcc2543b946d68035d4a4aa9d2fb58e5 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Thu, 13 Aug 2026 15:12:07 +0000 Subject: [PATCH 4/4] fix(a2a): close three more silent-read paths found in review All three are the same defect this PR exists to fix, and two of them were reintroduced by the fix itself -- worth stating plainly rather than folding quietly. 1. channel + thread disagreeing was a silent drop. `thread` is an ALIAS for `channel`, so passing both with different values has no correct reading, and preferring one silently reads a channel the caller did not ask for. Now a 400 naming both values. Identical values stay accepted -- with a paired test, so the check cannot pass by rejecting every request that carries the alias. 2. `since` accepted non-finite cursors. float() takes "nan", "inf" and "-inf"; a NaN cursor makes every bus-side comparison false, so the reader gets an empty window and a 200 confirming it, forever. That is exactly the silence the cursor validation was added to end. Now a 400. 3. _channel_exists only failed open on TRANSPORT failure. A bus answering 200 with an error body left the channel list empty and reported every channel as unknown -- accusing the caller of a typo because of a fault on the bus side. It now fails open on any payload it cannot read, discriminating on the `channels` KEY rather than on the list being empty, so a bus that genuinely knows no channels still reports unknown (pinned by its own test). Red-first: the three defect tests fail against the previous commit's route ("accused a typo on payload {'error': 'bus is having a bad day'}"), the two control tests pass both before and after by design. 39 green across all three bus test files. Found by kilo (1, 3) and CodeRabbit (2) -- all three accepted. --- tests/test_a2a_bus.py | 86 +++++++++++++++++++++++++++++++++++ tinyagentos/routes/a2a_bus.py | 44 ++++++++++++++++-- 2 files changed, 127 insertions(+), 3 deletions(-) diff --git a/tests/test_a2a_bus.py b/tests/test_a2a_bus.py index 1ebd8002c..66e7e4992 100644 --- a/tests/test_a2a_bus.py +++ b/tests/test_a2a_bus.py @@ -266,3 +266,89 @@ async def test_since_is_forwarded_as_the_cursor(client): "/api/a2a/bus/messages", params={"channel": "build", "since": "1786630185.75"} ) assert route.calls.last.request.url.params["since"] == "1786630185.75" + + +@pytest.mark.asyncio +async def test_channel_and_thread_disagreeing_is_400(client): + """`thread` is an ALIAS for `channel`, so both-with-different-values has no + correct reading -- and silently preferring one drops the other, which is the + same "ignored param reads as a working one" defect this endpoint was fixed + for. Reintroducing it through the alias would be the quietest possible + regression: the caller names the channel they want and reads another one. + """ + resp = await client.get( + "/api/a2a/bus/messages", params={"channel": "build", "thread": "ops"} + ) + assert resp.status_code == 400 + body = resp.json() + assert "build" in body["error"] and "ops" in body["error"] + + +@pytest.mark.asyncio +@respx.mock +async def test_channel_and_thread_agreeing_is_accepted(client): + """The 400 above must fire on DISAGREEMENT, not on the alias being present. + + Without this pair the check could reject every request carrying both params + and still pass its own test. + """ + route = respx.get(f"{_BUS}/a2a/messages").mock( + return_value=Response(200, json={"messages": []}) + ) + resp = await client.get( + "/api/a2a/bus/messages", params={"channel": "build", "thread": "build"} + ) + assert resp.status_code == 200 + assert route.calls.last.request.url.params["thread"] == "build" + + +@pytest.mark.asyncio +async def test_since_rejects_non_finite_cursors(client): + """`float()` accepts "nan", "inf" and "-inf". + + A NaN cursor makes every comparison on the bus side false, so the reader + gets an empty window and a 200 confirming it, forever -- the exact silence + this endpoint was fixed for, smuggled back in through the validator that was + supposed to close it. + """ + for bad in ("nan", "NaN", "inf", "-inf", "Infinity"): + resp = await client.get( + "/api/a2a/bus/messages", params={"channel": "build", "since": bad} + ) + assert resp.status_code == 400, f"{bad} was accepted as a cursor" + assert "finite" in resp.json()["error"] + + +@pytest.mark.asyncio +@respx.mock +async def test_channel_probe_fails_open_on_an_unreadable_payload(client): + """Fail open on a payload we cannot read, not only on transport failure. + + A bus returning HTTP 200 with an error body leaves the channel list empty, + which would report every channel as unknown -- accusing the caller of a typo + because of a fault on the bus side. The docstring promised fail-open; only + the transport path delivered it. + """ + respx.get(f"{_BUS}/a2a/messages").mock(return_value=Response(200, json={"messages": []})) + for payload in ({"error": "bus is having a bad day"}, ["build", "ops"], "nope"): + respx.get(f"{_BUS}/a2a/channels").mock(return_value=Response(200, json=payload)) + resp = await client.get("/api/a2a/bus/messages", params={"channel": "build"}) + assert resp.status_code == 200 + assert resp.json()["channel_known"] is True, f"accused a typo on payload {payload!r}" + + +@pytest.mark.asyncio +@respx.mock +async def test_channel_probe_still_reports_unknown_on_a_real_empty_list(client): + """The fail-open above must not swallow the real signal. + + A bus that genuinely knows no channels answers with the `channels` key and + an empty list -- that is a real "unknown", and it has to stay reportable or + the typo-distinction feature is gone. + """ + respx.get(f"{_BUS}/a2a/messages").mock(return_value=Response(200, json={"messages": []})) + respx.get(f"{_BUS}/a2a/channels").mock(return_value=Response(200, json={"channels": []})) + + resp = await client.get("/api/a2a/bus/messages", params={"channel": "build"}) + assert resp.status_code == 200 + assert resp.json()["channel_known"] is False diff --git a/tinyagentos/routes/a2a_bus.py b/tinyagentos/routes/a2a_bus.py index 120b26d81..48c51f017 100644 --- a/tinyagentos/routes/a2a_bus.py +++ b/tinyagentos/routes/a2a_bus.py @@ -25,6 +25,7 @@ from __future__ import annotations import logging +import math import os import httpx @@ -149,7 +150,24 @@ async def bus_messages(request: Request): status_code=400, ) - channel = request.query_params.get("channel") or request.query_params.get("thread") or "" + # `thread` is an alias for `channel`, so passing BOTH with different values + # has no correct interpretation -- and picking one silently drops the other, + # which is the same "ignored param reads as a working one" failure this + # endpoint is being fixed for. Identical values are harmless and allowed. + chan_param = request.query_params.get("channel") or "" + thread_param = request.query_params.get("thread") or "" + if chan_param and thread_param and chan_param != thread_param: + return JSONResponse( + { + "error": ( + "channel and thread are the same parameter and disagree: " + f"channel={chan_param!r} thread={thread_param!r}" + ), + "hint": "pass one of them, not both", + }, + status_code=400, + ) + channel = chan_param or thread_param if not channel: return JSONResponse({"error": "channel required"}, status_code=400) @@ -165,12 +183,22 @@ async def bus_messages(request: Request): params["thread"] = channel if since_raw is not None: try: - params["since"] = float(since_raw) + since_val = float(since_raw) except ValueError: return JSONResponse( {"error": "since must be a message ts (float), not an id"}, status_code=400, ) + # float() happily accepts "nan", "inf" and "-inf". A NaN cursor makes + # every comparison on the bus side false, so the reader gets a silent + # empty window forever and a 200 confirming it -- the exact failure this + # endpoint is being fixed for, reintroduced through the validator. + if not math.isfinite(since_val): + return JSONResponse( + {"error": "since must be a finite message ts (float), not an id"}, + status_code=400, + ) + params["since"] = since_val bus = _bus_url() try: @@ -211,7 +239,17 @@ async def _channel_exists(bus: str, channel: str) -> bool: except Exception as exc: # noqa: BLE001 logger.warning("A2A bus channel probe failed (%s): %s", bus, exc) return True - channels = data.get("channels", []) if isinstance(data, dict) else [] + # Fail open on a payload we cannot read, not just on a transport failure. + # A 200 carrying an error body (or anything that is not the channel list) + # would otherwise leave `channels` empty and report every channel as + # unknown -- an accusation of a typo, generated by a bus-side fault, which + # is precisely the false alarm this probe exists to avoid. + if not isinstance(data, dict) or "channels" not in data: + logger.warning( + "A2A bus channel probe got an unreadable payload from %s: %r", bus, data + ) + return True + channels = data.get("channels") or [] # The bus spells it "channel" in this payload (verified against the live bus: # {"channels":[{"channel":"build","members":[...],...}]}) while the SAME # concept is "thread" on /a2a/messages. Accept both rather than trust one