From 592230dc6a8611c74102f9c644018f6aea043a6a Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Thu, 17 Sep 2026 21:09:47 +0000 Subject: [PATCH 01/10] Give each answer request its own thread The thread_id was `search-{id(body)}`. CPython reuses the address of a freed object immediately, so request bodies -- allocated and discarded per request -- collide almost always. Measured over 200 requests through the real endpoint: 38 distinct threads, 192 of the 200 on a shared one, one thread serving 14 different questions. That matters because `chat_history` is checkpointed state annotated with `add_messages`, and the rephraser injects it through a MessagesPlaceholder. Two requests on one thread means one stranger's question and answer become the context that rephrases the next stranger's question. The test uses 50 requests deliberately. The two-request version I wrote first passed against the broken code -- sequential bodies usually do get distinct addresses -- which would have made it a test that could not find the bug it was written for. Co-Authored-By: Claude Opus 5 --- src/api/answer.py | 27 +++++++-- tests/api/test_answer_endpoint.py | 97 ++++++++++++++++++++++++++++++- 2 files changed, 119 insertions(+), 5 deletions(-) diff --git a/src/api/answer.py b/src/api/answer.py index 9c14033..df29245 100644 --- a/src/api/answer.py +++ b/src/api/answer.py @@ -18,6 +18,8 @@ """ import json +import time +import uuid from collections.abc import AsyncIterator from typing import Any @@ -54,7 +56,9 @@ def _refusal(reason: str) -> StreamingResponse: """ async def body() -> AsyncIterator[str]: - yield _sse("done", {"state": "refused"}) + # `seconds` too: a refusal is a `done` like any other, and the caller + # parses one shape. It is ~0 because refusal precedes the graph. + yield _sse("done", {"state": "refused", "seconds": 0.0}) logger.info("answer request refused: %s", reason) return StreamingResponse(body(), media_type="text/event-stream") @@ -75,12 +79,25 @@ async def answer(request: Request, body: AnswerRequest) -> StreamingResponse: graph = get_graph() + # A fresh thread per request, never `id(body)`. `chat_history` is checkpointed + # state annotated with `add_messages`, and the rephraser injects it through a + # MessagesPlaceholder, so two requests sharing a thread means one stranger's + # question and answer rephrase the next stranger's question. CPython reuses the + # address of a freed object immediately: measured over 200 requests, `id(body)` + # produced 38 distinct threads and put 192 of them on a shared one. + + # Resolved at startup, not per request: the graph is built from the same + # bundles, so startup is what is actually being served. None if unknown -- + # a missing release costs the caller cache invalidation, not an answer. + release = getattr(request.app.state, "release", None) + async def stream() -> AsyncIterator[str]: - yield _sse("start", {"answered": True}) + yield _sse("start", {"release": release, "answered": True}) + started = time.monotonic() state = "failed" try: async for event in graph.astream_answer( - body.question, PROFILE, thread_id=f"search-{id(body)}" + body.question, PROFILE, thread_id=f"search-{uuid.uuid4()}" ): if event.kind == "token": yield _sse("token", {"text": event.text}) @@ -97,6 +114,8 @@ async def stream() -> AsyncIterator[str]: # terminal event to stop waiting. logger.exception("answering %r failed", body.question[:80]) state = "failed" - yield _sse("done", {"state": state}) + yield _sse( + "done", {"state": state, "seconds": round(time.monotonic() - started, 1)} + ) return StreamingResponse(stream(), media_type="text/event-stream") diff --git a/tests/api/test_answer_endpoint.py b/tests/api/test_answer_endpoint.py index 6a5ac0f..8197594 100644 --- a/tests/api/test_answer_endpoint.py +++ b/tests/api/test_answer_endpoint.py @@ -10,6 +10,7 @@ a failure ending with a terminal event rather than a hang. """ +import json import time from collections.abc import AsyncIterator from typing import Any @@ -126,7 +127,9 @@ def test_no_token_means_no_model_call(keys: tuple[str, str], stub: _StubGraph) - f"{PREFIX}/answer", json={"question": "anything", "human_token": ""} ) assert response.status_code == 200 - assert _events(response.text) == [("done", '{"state": "refused"}')] + events = _events(response.text) + assert [name for name, _ in events] == ["done"] + assert json.loads(events[0][1])["state"] == "refused" assert stub.calls == 0, "the graph must not be touched for a refused request" @@ -167,3 +170,95 @@ def test_an_empty_question_is_rejected_by_validation( ) assert response.status_code == 422 assert stub.calls == 0 + + +class _ThreadRecordingGraph(_StubGraph): + """Records the thread_id each request runs under.""" + + def __init__(self) -> None: + super().__init__() + self.thread_ids: list[str] = [] + + async def astream_answer(self, *_a: Any, **kwargs: Any) -> AsyncIterator[AnswerEvent]: + self.thread_ids.append(kwargs["thread_id"]) + for event in self._events: + yield event + + +def test_separate_requests_do_not_share_a_thread( + keys: tuple[str, str], monkeypatch: pytest.MonkeyPatch +) -> None: + """Two askers must not land on one checkpointer thread. + + `chat_history` is checkpointed state annotated with `add_messages`, and the + rephraser injects it via a MessagesPlaceholder. Sharing a thread_id means one + stranger's question and answer become the context that rephrases the next + stranger's question. + + Why 50 requests and not 2: the original bug keyed the thread on `id(body)`, + and two sequential requests usually get distinct addresses, so a two-request + version of this test passed against the broken code. At 50 it fails every + time -- measured over 200 requests, the broken version produced 38 distinct + threads and put 192 of them on a shared one. + """ + private, public = keys + graph = _ThreadRecordingGraph() + monkeypatch.setattr("api.answer.get_graph", lambda: graph) + client = _client(public) + + requests = 50 + for index in range(requests): + response = client.post( + f"{PREFIX}/answer", + json={"question": f"question {index}", "human_token": _token(private)}, + ) + assert response.status_code == 200 + + assert len(graph.thread_ids) == requests + assert len(set(graph.thread_ids)) == requests, ( + f"{requests} requests shared {requests - len(set(graph.thread_ids))} threads" + ) + + +def test_start_carries_the_release_and_done_carries_seconds( + keys: tuple[str, str], stub: _StubGraph +) -> None: + """The contract's fields, pinned. + + Both were specified in contracts/answer_endpoint.md and both were missing + from the first implementation. The website codes against that document, so a + field it promises and we never send is a defect on their side, not ours. + `release` is what makes FR-007 cache invalidation possible at all. + """ + private, public = keys + app = FastAPI() + app.include_router(router, prefix=PREFIX) + app.state.human_token_key = public + app.state.release = 97 + client = TestClient(app) + + response = client.post( + f"{PREFIX}/answer", + json={"question": "what is CDK5", "human_token": _token(private)}, + ) + events = dict(_events(response.text)) + + assert json.loads(events["start"])["release"] == 97 + done = json.loads(events["done"]) + assert done["state"] == "answered" + assert isinstance(done["seconds"], float) + + +def test_a_refusal_has_the_same_done_shape_as_an_answer( + keys: tuple[str, str], stub: _StubGraph +) -> None: + """One shape, so the caller parses `done` one way.""" + _, public = keys + client = _client(public) + + response = client.post(f"{PREFIX}/answer", json={"question": "what is CDK5"}) + done = json.loads(dict(_events(response.text))["done"]) + + assert done["state"] == "refused" + assert set(done) == {"state", "seconds"} + assert stub.calls == 0 From 307a81ee7c5e3f45f8a8c6cb5a35b6e780738eec Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Thu, 17 Sep 2026 21:09:47 +0000 Subject: [PATCH 02/10] Send the contract's release and seconds fields contracts/answer_endpoint.md promises `start` carrying `release` and `done` carrying `seconds`. The implementation sent neither. The website codes against that document, so a promised field we never send is a defect on their side. `release` is not cosmetic: FR-007 exists so a cached answer can be invalidated after a release, and without the field there is nothing to invalidate on. It comes from the bundle directory name, which is the only place it is recorded. Bundles can differ -- reactome is at 97 while userguide is at 95 -- so get_release is per-bundle and the endpoint reports reactome's, that being the pathway content an answer is about. Resolved at startup, because the graph is built from the same bundles and so startup describes what is actually served. Refusals carry `seconds` too, so `done` has one shape to parse. Co-Authored-By: Claude Opus 5 --- bin/chat-fastapi.py | 6 ++++++ src/util/embedding_environment.py | 21 +++++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/bin/chat-fastapi.py b/bin/chat-fastapi.py index 2683f4d..9b53a9e 100644 --- a/bin/chat-fastapi.py +++ b/bin/chat-fastapi.py @@ -16,6 +16,7 @@ from agent.registry import build_graph, set_graph from api.answer import router as answer_router from util.captcha_scope import is_captcha_exempt +from util.embedding_environment import EmbeddingEnvironment from util.human_token import load_verifying_key from util.logging import logging from util.secrets import SECRET_NAMES, get_secret, load_secrets_to_environ @@ -43,6 +44,11 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]: # Before the graph: a missing verifying key must stop the process, and # spending 52 seconds building a graph first only delays the failure. _app.state.human_token_key = load_verifying_key() + # The release the served answers are built from, for FR-007 cache + # invalidation. Read here rather than per request because the graph below is + # built from these same bundles, so this value describes what is served even + # if the pointer file changes underneath a running process. + _app.state.release = EmbeddingEnvironment.get_release("reactome") graph = build_graph() set_graph(graph) logging.info("Agent graph ready in %.1fs", time.monotonic() - started) diff --git a/src/util/embedding_environment.py b/src/util/embedding_environment.py index 9560672..07434ec 100644 --- a/src/util/embedding_environment.py +++ b/src/util/embedding_environment.py @@ -63,6 +63,27 @@ def require_dir(cls, key: str) -> Path: ) return directory + @classmethod + def get_release(cls, key: str) -> int | None: + """The Reactome release a bundle was built from, e.g. 97. + + Read from the bundle directory name (`.../reactome/Release97`), which is + the only place it is recorded -- nothing writes a manifest. Returns None + rather than raising: a missing release must not stop the service, it only + costs a caller the ability to invalidate a cached answer (FR-007). + + Bundles can sit at different releases -- reactome at 97 while userguide is + at 95 -- so this is per-bundle and the caller says which one it means. + """ + path = cls._get().embeddings.get(key) + if path is None: + return None + name = path.name + if not name.startswith("Release"): + return None + suffix = name[len("Release") :] + return int(suffix) if suffix.isdigit() else None + @classmethod def get_model(cls, key: str) -> str: return str(cls._get().embeddings[key].parent.parent) From fde108ccf6a398d4edb7a2526e3cbca8445f3bc5 Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Thu, 17 Sep 2026 21:11:28 +0000 Subject: [PATCH 03/10] Format the test file Co-Authored-By: Claude Opus 5 --- tests/api/test_answer_endpoint.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/tests/api/test_answer_endpoint.py b/tests/api/test_answer_endpoint.py index 8197594..e221b25 100644 --- a/tests/api/test_answer_endpoint.py +++ b/tests/api/test_answer_endpoint.py @@ -179,7 +179,9 @@ def __init__(self) -> None: super().__init__() self.thread_ids: list[str] = [] - async def astream_answer(self, *_a: Any, **kwargs: Any) -> AsyncIterator[AnswerEvent]: + async def astream_answer( + self, *_a: Any, **kwargs: Any + ) -> AsyncIterator[AnswerEvent]: self.thread_ids.append(kwargs["thread_id"]) for event in self._events: yield event @@ -215,9 +217,9 @@ def test_separate_requests_do_not_share_a_thread( assert response.status_code == 200 assert len(graph.thread_ids) == requests - assert len(set(graph.thread_ids)) == requests, ( - f"{requests} requests shared {requests - len(set(graph.thread_ids))} threads" - ) + assert ( + len(set(graph.thread_ids)) == requests + ), f"{requests} requests shared {requests - len(set(graph.thread_ids))} threads" def test_start_carries_the_release_and_done_carries_seconds( From 94ee252bb8e84e422e2001eed121f708f45df587 Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Thu, 17 Sep 2026 21:11:55 +0000 Subject: [PATCH 04/10] Mark what shipped, and record what the review found T007-T013, T015, T016 and T018 shipped in #235 but nothing was checked off. T014a and T014b are the two defects the adversarial review of that PR found. T014 (SC-003) and T017 (FR-008 rate limiting) remain genuinely open. Co-Authored-By: Claude Opus 5 --- specs/010-search-page-answers/tasks.md | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/specs/010-search-page-answers/tasks.md b/specs/010-search-page-answers/tasks.md index fdc634e..26e5934 100644 --- a/specs/010-search-page-answers/tasks.md +++ b/specs/010-search-page-answers/tasks.md @@ -27,21 +27,23 @@ website repo entirely. Speed is Phase 5 and does not gate the handover. arrives, tokens stream, citations resolve to real stable IDs, and `done` carries a state. -- [ ] T007 [P] [US1] Add src/util/human_token.py: verify signature and expiry only, stateless, no consumption tracking -- [ ] T008 [P] [US1] Test human_token in tests/util/test_human_token.py: valid, expired, wrong key, tampered payload, absent — every failure refuses -- [ ] T009 [US1] Refuse at startup in src/util/human_token.py when the verifying key is missing or unreadable, rather than accepting everything (Principle IV) -- [ ] T010 [US1] Add the SSE endpoint in src/api/answer.py implementing contracts/answer_endpoint.md: start, token, citation, done -- [ ] T011 [US1] Emit citations from retrieved documents' `st_id` metadata, deduplicated — never by parsing anchors out of the model's prose -- [ ] T012 [US1] Mount the router in bin/chat-fastapi.py and let the captcha middleware pass /chat/api/ through, since the endpoint verifies its own caller -- [ ] T013 [US1] Test over HTTP with a real client in tests/api/test_answer_endpoint.py, not by calling the handler — mounting order and middleware only interact in the served path (Principle I) +- [x] T007 [P] [US1] Add src/util/human_token.py: verify signature and expiry only, stateless, no consumption tracking +- [x] T008 [P] [US1] Test human_token in tests/util/test_human_token.py: valid, expired, wrong key, tampered payload, absent — every failure refuses +- [x] T009 [US1] Refuse at startup in src/util/human_token.py when the verifying key is missing or unreadable, rather than accepting everything (Principle IV) +- [x] T010 [US1] Add the SSE endpoint in src/api/answer.py implementing contracts/answer_endpoint.md: start, token, citation, done +- [x] T011 [US1] Emit citations from retrieved documents' `st_id` metadata, deduplicated — never by parsing anchors out of the model's prose +- [x] T012 [US1] Mount the router in bin/chat-fastapi.py and let the captcha middleware pass /chat/api/ through, since the endpoint verifies its own caller +- [x] T013 [US1] Test over HTTP with a real client in tests/api/test_answer_endpoint.py, not by calling the handler — mounting order and middleware only interact in the served path (Principle I) - [ ] T014 [US1] Assert the answer matches the chat UI's for the same question (SC-003); two surfaces that can disagree is a defect +- [x] T014a [US1] Give each request its own checkpointer thread in src/api/answer.py; `id(body)` put 192 of 200 requests on a shared thread, and `chat_history` is checkpointed state the rephraser reads (PR #236) +- [x] T014b [US1] Send `release` on start and `seconds` on done per contracts/answer_endpoint.md; both were promised to the website and neither was implemented (PR #236) ## Phase 4: User Story 2 — no answer without a verified person (P1) -- [ ] T015 [US2] Refuse missing, expired, malformed and wrongly-signed tokens before any model call, in src/api/answer.py -- [ ] T016 [US2] Test that no model call happens for a refused request in tests/api/test_answer_endpoint.py, by asserting on a patched graph rather than on timing (SC-002) +- [x] T015 [US2] Refuse missing, expired, malformed and wrongly-signed tokens before any model call, in src/api/answer.py +- [x] T016 [US2] Test that no model call happens for a refused request in tests/api/test_answer_endpoint.py, by asserting on a patched graph rather than on timing (SC-002) - [ ] T017 [P] [US2] Rate limit per token as a backstop; the budget is the website's, enforced before the call reaches here (FR-008) -- [ ] T018 [US2] Return `state: failed` with no partial answer on any internal error, so the page renders no panel (FR-006) +- [x] T018 [US2] Return `state: failed` with no partial answer on any internal error, so the page renders no panel (FR-006) ## Phase 5: Latency (does NOT gate the handover) From 4824ac4e25a8af908dfde4cdc5b9a6999b9ee2ba Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Thu, 17 Sep 2026 21:16:56 +0000 Subject: [PATCH 05/10] Bound the answer, so a stuck upstream cannot hold the connection FR-006 names timeout alongside error, and nothing implemented it. The only bound was the LLM client's `request_timeout=360.0` -- six minutes, per model call, and six calls run around one answer. A stuck upstream could hold a connection for over half an hour and never send `done`, on a page whose whole contract is that it must be safe to ignore. 120s is about twice the worst complete answer measured, so it does not cut off answers that were going to arrive; it converts an unbounded hang into a bounded one. TimeoutError is logged distinctly from a crash so an operator can tell a slow upstream from a broken one. The stand-in hangs for a finite 5s rather than an hour, so a regression fails the suite in five seconds instead of hanging it. Verified that the assertion catches an ineffective bound, not merely a missing constant: with the timeout raised above the hang it reports "stream ran 5.0s; the timeout did not fire". Co-Authored-By: Claude Opus 5 --- src/api/answer.py | 48 +++++++++++++++++++++++-------- tests/api/test_answer_endpoint.py | 44 ++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 12 deletions(-) diff --git a/src/api/answer.py b/src/api/answer.py index df29245..3e2a05c 100644 --- a/src/api/answer.py +++ b/src/api/answer.py @@ -17,6 +17,7 @@ verifying key -- stops the process at startup. """ +import asyncio import json import time import uuid @@ -37,6 +38,16 @@ PROFILE = "react-to-me" +# FR-006 names timeout alongside error, and nothing here implemented it. The only +# bound was the LLM client's `request_timeout=360.0` -- six minutes per model call, +# and six calls run around one answer, so a pathological request could hold a +# connection for over half an hour and never send `done`. +# +# 120s is about twice the worst complete answer measured (max 31.5s, with the first +# answer token near 36s on a heavy question), so it does not cut off answers that +# were going to arrive; it converts an unbounded hang into a bounded one. +ANSWER_TIMEOUT_SECONDS = 120.0 + class AnswerRequest(BaseModel): question: str = Field(min_length=1, max_length=2000) @@ -96,18 +107,31 @@ async def stream() -> AsyncIterator[str]: started = time.monotonic() state = "failed" try: - async for event in graph.astream_answer( - body.question, PROFILE, thread_id=f"search-{uuid.uuid4()}" - ): - if event.kind == "token": - yield _sse("token", {"text": event.text}) - elif event.kind == "citation": - yield _sse( - "citation", - {"st_id": event.st_id, "display_name": event.display_name}, - ) - elif event.kind == "done": - state = event.state or "failed" + async with asyncio.timeout(ANSWER_TIMEOUT_SECONDS): + async for event in graph.astream_answer( + body.question, PROFILE, thread_id=f"search-{uuid.uuid4()}" + ): + if event.kind == "token": + yield _sse("token", {"text": event.text}) + elif event.kind == "citation": + yield _sse( + "citation", + { + "st_id": event.st_id, + "display_name": event.display_name, + }, + ) + elif event.kind == "done": + state = event.state or "failed" + except TimeoutError: + # Distinct from the crash below so an operator can tell a slow + # upstream from a broken one. + logger.warning( + "answering %r exceeded %.0fs", + body.question[:80], + ANSWER_TIMEOUT_SECONDS, + ) + state = "failed" except Exception: # Deliberately broad, and deliberately not re-raised: a half-written # SSE stream cannot become an HTTP error code, and the caller needs a diff --git a/tests/api/test_answer_endpoint.py b/tests/api/test_answer_endpoint.py index e221b25..0b1690c 100644 --- a/tests/api/test_answer_endpoint.py +++ b/tests/api/test_answer_endpoint.py @@ -10,6 +10,7 @@ a failure ending with a terminal event rather than a hang. """ +import asyncio import json import time from collections.abc import AsyncIterator @@ -264,3 +265,46 @@ def test_a_refusal_has_the_same_done_shape_as_an_answer( assert done["state"] == "refused" assert set(done) == {"state", "seconds"} assert stub.calls == 0 + + +class _HangingGraph(_StubGraph): + """Never yields its second event, like an upstream that stopped responding.""" + + async def astream_answer(self, *_a: Any, **_k: Any) -> AsyncIterator[AnswerEvent]: + self.calls += 1 + yield AnswerEvent(kind="token", text="partial ") + await asyncio.sleep( + 5 + ) # finite, so a regression fails fast instead of hanging CI + + +def test_a_hanging_upstream_still_ends_the_stream( + keys: tuple[str, str], monkeypatch: pytest.MonkeyPatch +) -> None: + """FR-006 names timeout alongside error, and nothing implemented it. + + The only bound was the LLM client's 360s request_timeout, once per model call + and six calls per answer, so a stuck upstream could hold the connection for + over half an hour and never send `done`. The search page must not depend on + this service being up. + + The timeout is patched to 0.25s so the test is quick, and the stand-in hangs + for a finite 5s so that a regression fails in five seconds rather than hanging + the suite for an hour. + """ + private, public = keys + graph = _HangingGraph() + monkeypatch.setattr("api.answer.get_graph", lambda: graph) + monkeypatch.setattr("api.answer.ANSWER_TIMEOUT_SECONDS", 0.25) + client = _client(public) + + started = time.monotonic() + response = client.post( + f"{PREFIX}/answer", + json={"question": "what is CDK5", "human_token": _token(private)}, + ) + elapsed = time.monotonic() - started + + events = dict(_events(response.text)) + assert json.loads(events["done"])["state"] == "failed" + assert elapsed < 2, f"stream ran {elapsed:.1f}s; the timeout did not fire" From db3a1a716fae651f66bc44d42defba075f860f48 Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Thu, 17 Sep 2026 21:17:14 +0000 Subject: [PATCH 06/10] Tell the website the stream is bounded The 120s ceiling is something a caller needs to know, so it belongs in the contract they code against rather than only in our source. Co-Authored-By: Claude Opus 5 --- specs/010-search-page-answers/contracts/answer_endpoint.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/specs/010-search-page-answers/contracts/answer_endpoint.md b/specs/010-search-page-answers/contracts/answer_endpoint.md index 5ba4e83..70f352f 100644 --- a/specs/010-search-page-answers/contracts/answer_endpoint.md +++ b/specs/010-search-page-answers/contracts/answer_endpoint.md @@ -54,6 +54,11 @@ produces a `done` with a non-`answered` state. The website renders no panel. The search page must never be slower or broken because this service is down (FR-006, SC-004). +**The stream is bounded.** The server gives up after 120 seconds and sends `done` +with `state: failed`. A caller still needs its own timeout -- a dropped connection +sends nothing -- but the server will not hold one open indefinitely. 120s is about +twice the slowest complete answer measured; it is a ceiling, not a target. + **No answer without a token.** Refused before any model call, not after (FR-003). Search pages get crawled, and every crawled search reaching the model is a bill. From 86931db6af78c436621c21235685a4af5281d0bc Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Thu, 17 Sep 2026 21:17:25 +0000 Subject: [PATCH 07/10] Record the timeout task Co-Authored-By: Claude Opus 5 --- specs/010-search-page-answers/tasks.md | 1 + 1 file changed, 1 insertion(+) diff --git a/specs/010-search-page-answers/tasks.md b/specs/010-search-page-answers/tasks.md index 26e5934..ef4c21d 100644 --- a/specs/010-search-page-answers/tasks.md +++ b/specs/010-search-page-answers/tasks.md @@ -44,6 +44,7 @@ state. - [x] T016 [US2] Test that no model call happens for a refused request in tests/api/test_answer_endpoint.py, by asserting on a patched graph rather than on timing (SC-002) - [ ] T017 [P] [US2] Rate limit per token as a backstop; the budget is the website's, enforced before the call reaches here (FR-008) - [x] T018 [US2] Return `state: failed` with no partial answer on any internal error, so the page renders no panel (FR-006) +- [x] T018a [US2] Bound the answer at 120s in src/api/answer.py; FR-006 names timeout and only the LLM client's 360s-per-call limit existed, so a stuck upstream could hold a connection for over half an hour (PR #236) ## Phase 5: Latency (does NOT gate the handover) From e3ca53543f890022268f959e6fcf4836273a3031 Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Thu, 17 Sep 2026 21:35:56 +0000 Subject: [PATCH 08/10] Strip the anchors the contract promised would not be there The contract says citations are separate events so the website can style links itself, and calls out that returning prose with embedded HTML anchors "would force the search page to parse them back out". The endpoint returned exactly that: the `react-to-me` profile is the chat UI's, and its prompt asks for inline links. Found by running the endpoint end to end against a real graph for the first time. It cannot be fixed a fragment at a time -- measured live, one anchor arrives as twenty-odd fragments: ' <', 'a', ' href', '="', 'https', '://', 'react', 'ome', '.org', '/content', '/detail', '/R', '-H', 'SA', ... A caller rendering incrementally would show that verbatim before it snapped into a link, and the page would be injecting model-generated HTML into its own DOM. AnchorStripper holds back only what could still become an anchor, so prose like "x < y" passes straight through. Every case is tested at every split point, plus character-at-a-time and random splits, because a fragment-at-a-time implementation passes a whole-anchor test and fails the real stream. Verified live: anchors in prose 10 -> 0, with citations still arriving as structured events and the answer intact. The anchor's text is kept -- removing it would break a sentence with a linked phrase mid-clause -- which leaves a trailing pathway title where the model used a link as a citation. Noted in the contract as cosmetic. Co-Authored-By: Claude Opus 5 --- .../contracts/answer_endpoint.md | 18 +++ src/api/answer.py | 13 ++- src/util/anchor_strip.py | 78 +++++++++++++ tests/util/test_anchor_strip.py | 109 ++++++++++++++++++ 4 files changed, 217 insertions(+), 1 deletion(-) create mode 100644 src/util/anchor_strip.py create mode 100644 tests/util/test_anchor_strip.py diff --git a/specs/010-search-page-answers/contracts/answer_endpoint.md b/specs/010-search-page-answers/contracts/answer_endpoint.md index 70f352f..ce78bb5 100644 --- a/specs/010-search-page-answers/contracts/answer_endpoint.md +++ b/specs/010-search-page-answers/contracts/answer_endpoint.md @@ -47,6 +47,24 @@ anchors -- what the chat UI does today -- would force the search page to parse t back out and re-style them. Stable IDs resolve at `reactome.org/content/detail/`. +At most **12** citations are sent. Measured against the live endpoint this cap is +binding on ordinary questions, so treat it as the most relevant few rather than +the complete set. + +### What `token` text contains + +**Markdown, never HTML.** Headings and lists appear; anchors do not. The answer +prompt is the chat UI's and does emit inline `` links, so the endpoint +strips them -- across fragment boundaries, because the model streams one anchor as +twenty-odd pieces (`' <'`, `'a'`, `' href'`, `'="'`, `'https'`, ...). A caller that +rendered fragments as they arrived would otherwise show the raw tag before it +became a link. + +The anchor's *text* is kept, since removing it would break any sentence with a +linked phrase in the middle. Where the model used a link as a trailing citation, +that leaves the pathway title as a bare clause -- cosmetic, and the structured +`citation` events are the reliable source for links. + ## Properties worth holding to **It must be safe to ignore.** Any failure, timeout, refusal or unverified caller diff --git a/src/api/answer.py b/src/api/answer.py index 3e2a05c..e5df02c 100644 --- a/src/api/answer.py +++ b/src/api/answer.py @@ -29,6 +29,7 @@ from pydantic import BaseModel, Field from agent.registry import get_graph +from util.anchor_strip import AnchorStripper from util.human_token import TokenRejectedError, verify from util.logging import logging @@ -106,13 +107,20 @@ async def stream() -> AsyncIterator[str]: yield _sse("start", {"release": release, "answered": True}) started = time.monotonic() state = "failed" + # The `react-to-me` prompt is the chat UI's and asks for inline anchors. + # The contract promises this caller prose without them, citations being + # separate events, so they come out here -- across fragment boundaries, + # because one anchor arrives as twenty-odd fragments. + stripper = AnchorStripper() try: async with asyncio.timeout(ANSWER_TIMEOUT_SECONDS): async for event in graph.astream_answer( body.question, PROFILE, thread_id=f"search-{uuid.uuid4()}" ): if event.kind == "token": - yield _sse("token", {"text": event.text}) + text = stripper.feed(event.text) + if text: + yield _sse("token", {"text": text}) elif event.kind == "citation": yield _sse( "citation", @@ -138,6 +146,9 @@ async def stream() -> AsyncIterator[str]: # terminal event to stop waiting. logger.exception("answering %r failed", body.question[:80]) state = "failed" + held = stripper.flush() + if held: + yield _sse("token", {"text": held}) yield _sse( "done", {"state": state, "seconds": round(time.monotonic() - started, 1)} ) diff --git a/src/util/anchor_strip.py b/src/util/anchor_strip.py new file mode 100644 index 0000000..0829b97 --- /dev/null +++ b/src/util/anchor_strip.py @@ -0,0 +1,78 @@ +"""Remove HTML anchors from a token stream that is split at arbitrary points. + +contracts/answer_endpoint.md promises the search page prose without embedded +anchors -- citations arrive as their own events so the website can style links +itself. The `react-to-me` profile is the chat UI's, and its prompt asks for +inline `` links, so the endpoint inherited them. + +They cannot be stripped a fragment at a time. Measured on the live endpoint, one +anchor arrived as twenty-odd fragments: ' <', 'a', ' href', '="', 'https', '://', +'react', 'ome', '.org', '/content', '/detail', '/R', '-H', 'SA', ... A caller +rendering incrementally would show that verbatim before it became a link. + +So this holds back only the part of the buffer that could still turn into an +anchor, and passes everything else straight through. +""" + +import re + +# A complete opening or closing anchor tag. +_TAG = re.compile(r"]*>|", re.IGNORECASE) + +# An unclosed '<...' is held only while it could still become an anchor. Beyond +# this it is treated as prose: real tags here are about seventy characters, and +# holding forever would strand text that never arrives. +_MAX_HELD = 200 + + +class AnchorStripper: + """Feed fragments in, get anchor-free text out. Call `flush` at the end.""" + + def __init__(self) -> None: + self._buffer = "" + + def feed(self, text: str) -> str: + self._buffer += text + out: list[str] = [] + while self._buffer: + start = self._buffer.find("<") + if start == -1: + out.append(self._buffer) + self._buffer = "" + break + out.append(self._buffer[:start]) + self._buffer = self._buffer[start:] + + match = _TAG.match(self._buffer) + if match: + self._buffer = self._buffer[match.end() :] + continue + if not self._could_become_anchor(): + out.append("<") + self._buffer = self._buffer[1:] + continue + break # Incomplete: wait for more. + return "".join(out) + + def _could_become_anchor(self) -> bool: + """True while the held '<...' might still complete into an anchor tag. + + Checked before length, so prose like "x < y" is released immediately + rather than waiting: the character after '<' settles it. + """ + held = self._buffer + if len(held) == 1: + return True # Just '<'; the next character decides. + if ">" in held: + return False # A complete tag that _TAG already declined. + if len(held) > _MAX_HELD: + return False + after = held[1] + if after == "/": + return len(held) == 2 or held[2] in "aA" + return after in "aA" + + def flush(self) -> str: + """Whatever is still held, emitted as prose. A truncated tag is not one.""" + remaining, self._buffer = self._buffer, "" + return remaining diff --git a/tests/util/test_anchor_strip.py b/tests/util/test_anchor_strip.py new file mode 100644 index 0000000..050cd64 --- /dev/null +++ b/tests/util/test_anchor_strip.py @@ -0,0 +1,109 @@ +"""The anchor stripper, driven the way the model actually streams. + +The bug this exists for is not "an anchor is present" but "an anchor is split". +A fragment-at-a-time regex passes a test that feeds whole anchors and fails on +the real stream, so every case here is also run at every possible split point. +""" + +import random + +import pytest + +from util.anchor_strip import AnchorStripper + +# Verbatim from the live endpoint, 2026-09-17: how one anchor actually arrived. +LIVE_FRAGMENTS = [ + " <", + "a", + " href", + '="', + "https", + "://", + "react", + "ome", + ".org", + "/content", + "/detail", + "/R", + "-H", + "SA", + "-", + "960", + "1234", + '">', + "ABCA", + "1", + " transports", + " cholesterol", + "", + " out", + " of", + " the", + " cell", + ".", +] + + +def _through(stripper: AnchorStripper, pieces: list[str]) -> str: + return "".join(stripper.feed(piece) for piece in pieces) + stripper.flush() + + +def _every_split(text: str) -> list[list[str]]: + return [[text[:i], text[i:]] for i in range(len(text) + 1)] + + +def test_the_live_fragment_sequence_loses_its_anchor_and_keeps_its_words() -> None: + assert _through(AnchorStripper(), LIVE_FRAGMENTS) == ( + " ABCA1 transports cholesterol out of the cell." + ) + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ('text', "text"), + ("before link after", "before link after"), + ("upper", "upper"), + ("no anchors at all", "no anchors at all"), + ("", ""), + # Prose that merely contains '<' must not be held back or eaten. + ("if x < y then", "if x < y then"), + ("a < b and c > d", "a < b and c > d"), + # Non-anchor markup is prose here; only anchors are the contract's problem. + ("bold", "bold"), + ("orphan close", "orphan close"), + ("two one and two", "two one and two"), + ], +) +def test_cases_hold_at_every_split_point(raw: str, expected: str) -> None: + assert _through(AnchorStripper(), [raw]) == expected + for pieces in _every_split(raw): + assert _through(AnchorStripper(), pieces) == expected, f"split: {pieces!r}" + + +def test_character_at_a_time_is_the_same_as_all_at_once() -> None: + """The worst split there is.""" + raw = 'CDK5 p25 binds.' + assert _through(AnchorStripper(), list(raw)) == "CDK5 p25 binds." + + +def test_random_splits_agree_with_whole_input() -> None: + raw = "x a y b z < w" + whole = _through(AnchorStripper(), [raw]) + rng = random.Random(0) + for _ in range(200): + pieces, rest = [], raw + while rest: + cut = rng.randint(1, min(4, len(rest))) + pieces.append(rest[:cut]) + rest = rest[cut:] + assert _through(AnchorStripper(), pieces) == whole + + +def test_an_unclosed_tag_is_not_swallowed_at_the_end() -> None: + """A stream that dies mid-tag must not silently lose the text before it.""" + stripper = AnchorStripper() + out = stripper.feed('kept text a y b z < w" whole = _through(AnchorStripper(), [raw]) - rng = random.Random(0) + rng = random.Random(0) # noqa: S311 - splitting a test string, not crypto for _ in range(200): pieces, rest = [], raw while rest: