Give each answer request its own thread, and send the contract's fields

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)

specs/010-search-page-answers/contracts/answer_endpoint.md

@@ -47,13 +47,36 @@ 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/<st_id>`.
 
+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 `<a href=...>` 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
 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.
 

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)
 

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")

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 `<a href=...>` 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"<a\b[^>]*>|</a\s*>", 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

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)