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/specs/010-search-page-answers/contracts/answer_endpoint.md b/specs/010-search-page-answers/contracts/answer_endpoint.md index 5ba4e83..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 @@ -54,6 +72,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. diff --git a/specs/010-search-page-answers/tasks.md b/specs/010-search-page-answers/tasks.md index fdc634e..08b5046 100644 --- a/specs/010-search-page-answers/tasks.md +++ b/specs/010-search-page-answers/tasks.md @@ -27,21 +27,26 @@ 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) +- [x] T011a [US1] Strip inline HTML anchors from the token stream in src/util/anchor_strip.py; the contract promises prose without them and the chat prompt emits them, split across ~20 fragments (PR #236) +- [x] T013a [US1] Run the endpoint end to end against a real graph: release 97, answered in 19.5-44.2s, 12 citations, anchors 0 (PR #236) +- [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) diff --git a/src/api/answer.py b/src/api/answer.py index 9c14033..e5df02c 100644 --- a/src/api/answer.py +++ b/src/api/answer.py @@ -17,7 +17,10 @@ verifying key -- stops the process at startup. """ +import asyncio import json +import time +import uuid from collections.abc import AsyncIterator from typing import Any @@ -26,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 @@ -35,6 +39,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) @@ -54,7 +68,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,28 +91,66 @@ 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" + # 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 for event in graph.astream_answer( - body.question, PROFILE, thread_id=f"search-{id(body)}" - ): - 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": + text = stripper.feed(event.text) + if text: + yield _sse("token", {"text": 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 # terminal event to stop waiting. logger.exception("answering %r failed", body.question[:80]) state = "failed" - yield _sse("done", {"state": state}) + held = stripper.flush() + if held: + yield _sse("token", {"text": held}) + yield _sse( + "done", {"state": state, "seconds": round(time.monotonic() - started, 1)} + ) return StreamingResponse(stream(), media_type="text/event-stream") 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/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) diff --git a/tests/api/test_answer_endpoint.py b/tests/api/test_answer_endpoint.py index 6a5ac0f..0b1690c 100644 --- a/tests/api/test_answer_endpoint.py +++ b/tests/api/test_answer_endpoint.py @@ -10,6 +10,8 @@ a failure ending with a terminal event rather than a hang. """ +import asyncio +import json import time from collections.abc import AsyncIterator from typing import Any @@ -126,7 +128,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 +171,140 @@ 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 + + +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" diff --git a/tests/util/test_anchor_strip.py b/tests/util/test_anchor_strip.py new file mode 100644 index 0000000..ec61e55 --- /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) # noqa: S311 - splitting a test string, not crypto + 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