Experimental/brain-of-reese
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
brain-of-reese/tests/integration/test_chat_api.py
T
ducoterra bef24e05e2
Build and Push Containers / build-and-push-app (push) Successful in 1m54s
Details
Build and Push Containers / build-and-push-db (push) Failing after 13s
Details
phase: 123_chat_image_questions 
All gates green. Verification complete.

**Phase 123 — final verification pass (all 4 tasks already in `complete/`)**

- Verified the full implementation is in the working tree: `app/api/chat_images.py` (upload/serve pair), `ChatRequest.image`/`ChatMessage.image` (path-validated, omitted-when-None), toggle-off + stale-file hinted error frames, `build_user_content` multimodal build at both sites (chat.py deflected branch + `run_agent`), config-gated composer attach/preview/upload-then-send, restore + shared rendering, CSP `img-src 'self' data:` carve-out, mock-LLM capture buffer.
- `uv run pytest` → **2796 passed**, exit 0 (unit + integration).
- `uv run pytest --cov=app --cov-report=term-missing` → **TOTAL 99%** (29/4615 missed; phase-123 modules 99–100%).
- `uv run pytest tests/e2e/test_chat_image_questions.py -v --no-cov` → **5 passed** in isolation.
- `uv run ruff check . && uv run pyright` → clean (0 errors).

**Completion criteria:** (1) attach→send→multimodal text+image to the model, bubble/reload/shared all render it, saved chat stores the PATH with `"base64" not in json.dumps(stored)` — **verified** (E2E tests 1–4 + integration round-trip); (2) `BOR_IMAGES=false` — control hidden, exact hinted error frame, zero model calls / no query_log row — **verified** (E2E test 5 + integration); (3) text-only byte-identical (`content` stays a plain `str`) — **verified** (unit + integration); (4) all gates green — **verified**; (5) commit + phase move — left to the harness per pipeline rules (no `git add`/`commit` run).

No defects found; no live-infrastructure changes (repo + local dev DB only). **Next pending phase: none** — 123 is the last phase in `todo/`.
2026-09-25 05:19:18 -04:00

2510 lines
104 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Integration: POST /api/chat — the RAG turn end-to-end.
Real Postgres (compose) seeded from ``tests/fixtures/docs/`` through the
real importer; the LLM client is a deterministic in-process fake
(token-overlap embeddings, canned streamed answer), so no network is
needed and the cosine ordering is meaningful: the Kubernetes question
retrieves the Kubernetes document.
Requires: podman compose up -d db
"""
from __future__ import annotations
import asyncio
import base64
import hashlib
import io
import json
import logging
import math
import re
import uuid
from collections.abc import Iterator
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
import pytest
from fastapi.testclient import TestClient
from pydantic import ValidationError
from sqlalchemy import delete, func, select, text
from sqlalchemy.orm import Session
from app.api import chat as chat_api
from app.api import chat_images
from app.config import Settings, get_settings
from app.main import app as fastapi_app
from app.models import Chunk, Document, GitSource, QueryLog
from app.rag import agent
from app.rag.agent import AGENT_TOOLS, READ_TRUNCATION_NOTICE
from app.rag.importer import import_sources
from app.rag.llm import EmbeddingError, LLMError, StreamPiece, ToolCallPiece, ToolResultPiece
from app.rag.prompts import build_high_prompt
from app.rag.retriever import TRUNCATION_MARKER
from app.schemas import ChatDoneEvent, SourceRef
from tests.conftest import ADMIN_PASSWORD
#: The fixture documents' fixed creation date (phase 106, D5): the
#: ``read`` result's second line is the row's ``created_at`` UTC date
#: part — a fixed value keeps the read-result pins deterministic
#: (instead of the ``now()`` server default of a bare insert).
_FIXTURE_CREATED_AT = datetime(2024, 6, 15, 12, 0, 0, tzinfo=UTC)
if TYPE_CHECKING:
from app.rag.scaffolding import ScaffoldingFilter
FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "docs"
QUESTION = "How is my Kubernetes cluster set up?"
OFF_TOPIC = "How do I bake sourdough bread?"
DIM = 768
_TOKEN_RE = re.compile(r"[a-z0-9]+")
def _token_vec(text: str) -> list[float]:
"""Bag-of-words unit vector — same algorithm as the E2E mock, so the
cosine behaviour here matches what the story E2E sees."""
vec = [0.0] * DIM
for tok in _TOKEN_RE.findall(text.lower()):
vec[int(hashlib.md5(tok.encode()).hexdigest(), 16) % DIM] += 1.0
norm = math.sqrt(sum(v * v for v in vec)) or 1.0
return [v / norm for v in vec]
class FakeRagLLM:
"""Duck-typed :class:`app.rag.llm.LLMClient` stand-in for the chat path."""
def __init__(
self,
answer: str = "Hey — you've got this! Talos, Cilium, three nodes. 🧠",
thinking: str = "",
embed_error: Exception | None = None,
stream_error: Exception | None = None,
fail_mid_stream: bool = False,
tool_script: list[list[StreamPiece | ToolCallPiece]] | None = None,
embed_fail_count: int = 0,
stream_fail_count: int = 0,
answer_sequence: list[str] | None = None,
) -> None:
self.settings = Settings(_env_file=None) # pyright: ignore[reportCallIssue]
self.embed_batches = 0
self.answer = answer
self.thinking = thinking
self.embed_error = embed_error
self.stream_error = stream_error
self.fail_mid_stream = fail_mid_stream
#: Phase 71: per-request canned answers (the recovery matrix):
#: request *i* (0-based, in ``seen_messages`` order) yields
#: ``answer_sequence[i]``; once exhausted it falls back to
#: ``answer``. ``None`` keeps the single-``answer`` behavior.
self.answer_sequence = answer_sequence
#: Phase 67: the first N ``embed_one`` calls raise an
#: ``EmbeddingError`` (then succeed) — a dead-then-recovered
#: embeddings endpoint for the retry loop.
self.embed_fail_count = embed_fail_count
#: Phase 67: the first N ``chat_stream`` requests die with an
#: ``LLMError`` BEFORE any piece (then succeed) — a dead-then-
#: recovered answer endpoint for the pre-first-piece retry rule.
self.stream_fail_count = stream_fail_count
self.question_embeds: list[str] = []
#: Phase 74: assistant history messages may carry
#: ``reasoning_content`` — the dict values stay strings, but the
#: key set is wider than the pre-phase ``{role, content}`` shape.
self.seen_messages: list[list[dict[str, Any]]] = []
#: Every request's ``tools`` value (phase 37) — ``None`` is the
#: pre-phase request shape (the key is absent from the payload).
self.seen_tools: list[list[dict[str, Any]] | None] = []
#: Canned per-agent-round piece lists (phase 37): ``tool_script[i]``
#: is yielded for the *i*-th request that carries a non-None
#: ``tools`` parameter (a request the agent loop is offering tools
#: on). A request without tools — the deflected direct path, the
#: cap-forced answer request, or the kill-switch
#: (``agent_max_rounds=0``) single-request path — always yields the
#: thinking + answer stream below, so a deflected turn through this
#: fake is byte-identical to the plain fake's output.
self.tool_script: list[list[StreamPiece | ToolCallPiece]] = list(tool_script or [])
async def embed(self, texts: list[str]) -> list[list[float]]:
self.embed_batches += 1
return [_token_vec(t) for t in texts]
async def chat(
self, messages: list[dict[str, str]], model: str | None = None
) -> str:
"""Deterministic ``lite`` stand-in for the import-time summaries
(phase 30) — same convention as ``tests.fakes.FakeEmbedder.chat``."""
user = next((m["content"] for m in messages if m.get("role") == "user"), "")
first = user.split()
return "Summary of " + (first[0] if first else "<empty>")
async def embed_one(self, text: str) -> list[float]:
if self.embed_error is not None:
raise self.embed_error
if self.embed_fail_count > 0:
self.embed_fail_count -= 1
self.question_embeds.append(text)
raise EmbeddingError("simulated embeddings endpoint failure")
self.question_embeds.append(text)
return _token_vec(text)
def _answer_for_request(self) -> str:
"""The canned answer for the request that was just recorded
(phase 71 ``answer_sequence``; ``None`` → the single answer)."""
if self.answer_sequence is None:
return self.answer
index = len(self.seen_messages) - 1
if index < len(self.answer_sequence):
return self.answer_sequence[index]
return self.answer
async def chat_stream(
self,
messages: list[dict[str, Any]],
tools: list[dict[str, Any]] | None = None,
scaffolding: ScaffoldingFilter | None = None,
):
"""Typed stream (phase 17): ``thinking`` slices (same 12-char
cadence as content) **before** the content pieces. With the
default ``thinking=""`` this yields content-only pieces — today's
behavior, new yield type. Phase 37: *tools* is the agent loop's
``tools=…`` passthrough (recorded in ``seen_tools``); a request
with tools consumes the next ``tool_script`` entry, if any.
Phase 71: *scaffolding* mirrors ``LLMClient.chat_stream`` — the
canned content pieces are fed through the caller's filter (an
empty clean result yields nothing) and the held tail is flushed
on normal completion, so a scaffolding-only canned answer streams
zero content pieces and leaves ``stripped_chars`` behind for the
recovery policy to key on. ``None`` (e.g. pre-phase callers) keeps
the byte-identical raw path."""
self.seen_messages.append(messages)
self.seen_tools.append(tools)
if self.stream_error is not None:
raise self.stream_error
if self.stream_fail_count > 0:
self.stream_fail_count -= 1
raise LLMError("simulated pre-piece endpoint failure")
mid_stream_drop = False
raw: list[StreamPiece | ToolCallPiece]
if tools is not None and self.tool_script:
raw = self.tool_script.pop(0)
elif self.fail_mid_stream:
raw = [StreamPiece("content", "partial ")]
mid_stream_drop = True
else:
answer = self._answer_for_request()
raw = cast(
"list[StreamPiece | ToolCallPiece]",
[
StreamPiece("thinking", self.thinking[i : i + 12])
for i in range(0, len(self.thinking), 12)
]
+ [
StreamPiece("content", answer[i : i + 12])
for i in range(0, len(answer), 12)
],
)
if scaffolding is None:
for piece in raw:
yield piece
if mid_stream_drop:
raise LLMError("mid-stream dropout")
return
for piece in raw:
if isinstance(piece, StreamPiece) and piece.kind == "content":
cleaned = scaffolding.feed(piece.text)
if cleaned:
yield StreamPiece("content", cleaned)
else:
yield piece
if mid_stream_drop:
# The tail is NOT flushed on a failed stream — the real
# client only flushes a cleanly completed one.
raise LLMError("mid-stream dropout")
tail = scaffolding.flush()
if tail:
yield StreamPiece("content", tail)
@pytest.fixture()
def seeded_kb(db) -> Iterator[FakeRagLLM]:
"""Fresh Postgres with the fixture docs imported (real pipeline)."""
db.execute(text("TRUNCATE chunks, documents, query_log"))
db.commit()
llm = FakeRagLLM()
summary = asyncio.run(import_sources([FIXTURES], llm, session=db))
assert summary.added == 13 # A9 formats (phase 47 added quadlet+j2); .hidden/ skipped
yield llm
db.execute(text("TRUNCATE chunks, documents, query_log"))
db.commit()
@pytest.fixture(autouse=True)
def _admin_signed_in(client: TestClient) -> None:
"""Phase 79 (task 03): ``POST /api/chat`` is user-gated — every turn
in this module runs as the signed-in ADMIN, so the shared ``client``
logs in once per test (the TestClient cookie jar carries the session
for every request of the test). The anonymous 401 contract itself is
pinned in ``test_auth_api.py``."""
r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"
def _stream_chat(client: TestClient, message: str) -> tuple[int, str, list[dict[str, Any]]]:
with client.stream("POST", "/api/chat", json={"message": message}) as r:
assert r.status_code == 200
assert r.headers["content-type"].startswith("text/event-stream")
buf = ""
frames: list[dict[str, Any]] = []
for part in r.iter_text():
buf += part
while "\n\n" in buf:
frame, buf = buf.split("\n\n", 1)
frame = frame.strip()
if frame.startswith("data:"):
frames.append(json.loads(frame.removeprefix("data:").strip()))
assert buf.strip() == "", "stream must end on a frame boundary"
return r.status_code, r.headers["content-type"], frames
def test_chat_streams_deltas_then_done_with_sources(client, db, seeded_kb: FakeRagLLM) -> None:
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
deltas = [f for f in frames if f.get("type") == "delta"]
assert len(deltas) >= 2 # genuinely streamed
assert "".join(d["text"] for d in deltas) == seeded_kb.answer
assert not any(f.get("type") == "error" for f in frames)
done = [f for f in frames if f.get("type") == "done"]
assert len(done) == 1
assert frames[-1]["type"] == "done" # done is the final event
assert done[0]["deflected"] is False
assert done[0]["suggestions"] == []
# Phase 119 (LOCKED A1 — the phase-118 A4 union retired): the
# citation surface is the agent's READ docs only — this turn's
# canned LLM never emits a tool call, so nothing was read and the
# grounded done frame chips nothing (an accepted, owner-directed
# consequence — the answer prose names the doc it used). The
# retrieval stays durably recorded (118-A3; pinned in the
# query_log test below this one).
assert done[0]["sources"] == []
# The LLM received the locked HIGH prompt — the ``<documents>`` block
# seeds the document's stored SUMMARY (phase 118, LOCKED A6: summary
# seeding re-revises the pre-phase full-text contract; the full text
# reaches the context only through the capped ``read`` tool). The
# summarizer's code-appended pointer line proves the summary block is
# present; the doc's full body is no longer seeded.
(system, user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
assert user["content"] == QUESTION
assert "<relevance>HIGH</relevance>" in system["content"]
assert "DEFLECT_MODE" not in system["content"]
assert "<documents>" in system["content"]
section = system["content"].split("<documents>", 1)[1].split("</documents>", 1)[0]
assert "Summary of" in section # the fake lite model's summary text
assert "Source: docs/homelab/kubernetes.md" in section # code-appended pointer
assert "Talos Linux" not in section # full doc no longer seeded (A6)
assert "HONESTY GATE" in system["content"]
def test_chat_streams_thinking_before_deltas(client, db, seeded_kb: FakeRagLLM) -> None:
"""Phase 17: ``thinking`` frames precede every ``delta`` frame and
reassemble to the model's reasoning; the ``done`` contract is
unchanged."""
thinker = FakeRagLLM(
thinking=(
"Step 1: parse the question. Step 2: check the kubernetes doc. "
"Step 3: name Talos, Cilium, three nodes. Step 4: answer."
)
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: thinker
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
thinking = [f for f in frames if f.get("type") == "thinking"]
deltas = [f for f in frames if f.get("type") == "delta"]
assert len(thinking) >= 1 # genuinely streamed
assert len(deltas) >= 2
# Every thinking frame precedes every delta frame.
ordered = [f["type"] for f in frames if f["type"] in ("thinking", "delta")]
assert ordered == ["thinking"] * len(thinking) + ["delta"] * len(deltas)
assert all(set(f.keys()) == {"type", "text"} for f in thinking)
assert "".join(f["text"] for f in thinking) == thinker.thinking
assert "".join(d["text"] for d in deltas) == thinker.answer
# Done still last; the citation surface is unchanged by the thinking
# extension — phase 119 (A1): read docs only, nothing read ⇒ none.
done = frames[-1]
assert done["type"] == "done"
assert done["deflected"] is False
assert done["suggestions"] == []
assert done["sources"] == []
assert not any(f.get("type") == "error" for f in frames)
def test_chat_thinking_suppressed_when_disabled(
client, db, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Phase 17 kill-switch: ``BOR_STREAM_THINKING=0`` drops every
``thinking`` frame; the delta stream is byte-identical to the
thinking-free case."""
thinker = FakeRagLLM(thinking="hidden reasoning that must never reach the wire")
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: thinker
# Same honesty gate the conftest/module already use (mock-calibrated
# 0.30 from the environment) — only the kill-switch changes.
live = get_settings()
monkeypatch.setattr(
chat_api,
"get_settings",
lambda: Settings(
_env_file=None, # pyright: ignore[reportCallIssue]
relevance_threshold=live.relevance_threshold,
stream_thinking=False,
),
)
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
assert not any(f.get("type") == "thinking" for f in frames)
deltas = [f for f in frames if f.get("type") == "delta"]
assert "".join(d["text"] for d in deltas) == thinker.answer
assert frames[-1]["type"] == "done"
assert not any(f.get("type") == "error" for f in frames)
def test_chat_writes_query_log_row(client, db, seeded_kb: FakeRagLLM) -> None:
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
_stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
rows = db.scalars(select(QueryLog)).all()
assert len(rows) == 1
row = rows[0]
assert row.question == QUESTION
assert row.deflected is False
total_chunks = db.scalar(select(func.count()).select_from(Chunk))
# chunk_hits is the fused candidate set (cosine top-N ∪ FTS top-N).
assert 1 <= row.chunk_hits <= total_chunks
assert row.top_score > 0.0 # genuine token-overlap cosine, best hit
assert row.top_score <= 1.0
# Phase 118 (LOCKED A3): the durable record is the FULL retrieval —
# the suggested tier (ranks 1–5) + the related tier (ranks 6–7) +
# the agent's reads (none on this turn), for this question.
for path in (
"docs/homelab/kubernetes.md", # rank 1
"docs/homelab/templates/deploy.j2", # rank 2
"docs/homelab/ssh/ssh_aliases.txt", # rank 3
"docs/homelab/container_gitlab/gitlab.md", # rank 4
"docs/deployments/new-service.md", # rank 5
"docs/homelab/quadlet/cache.volume", # rank 6 (related)
"docs/homelab/quadlet/compose.container", # rank 7 (related)
):
assert path in row.sources
assert row.latency_ms >= 0
# Why the gate answered (A8 revised): cosine over the threshold OR a
# lexical hit. The mock-calibrated threshold (0.30, see tests/conftest.py)
# makes the cosine branch true here; the FTS branch is covered too —
# "kubernetes" / "cluster" match the doc's tsvector.
thr = get_settings().relevance_threshold
assert row.top_score >= thr or (row.fts_hits or 0) > 0
assert (row.fts_hits or 0) >= 1 # the lexical branch really fired
def test_off_topic_question_deflects_honestly(client, db, seeded_kb: FakeRagLLM) -> None:
"""Phase 04 contract: weak retrieval ⇒ honest deflection, no fake answer."""
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
_, _, frames = _stream_chat(client, OFF_TOPIC)
finally:
fastapi_app.dependency_overrides.clear()
assert not any(f.get("type") == "error" for f in frames)
deltas = [f for f in frames if f.get("type") == "delta"]
assert len(deltas) >= 2 # the LLM is still called (voice stays chippy)
done = frames[-1]
assert done["type"] == "done"
assert done["deflected"] is True
# 2-3 alternative chips, all non-empty, derived from real titles/topics.
assert 2 <= len(done["suggestions"]) <= 3
assert all(s.strip() for s in done["suggestions"])
assert any(
"Deploying a New Service" in s for s in done["suggestions"]
), "the best weak-hit title must be offered as a chip"
# Phase 112 (A8 revised, TODO L2): a deflected turn cites nothing —
# done.sources is the citation surface (the UI chips every entry as
# "the answer used this"), and the weak hits are scored docs, not
# citations. (Pre-phase: they rode the wire as sources.)
assert done["sources"] == []
# The LLM saw the LOW prompt: DEFLECT_MODE + titles, never doc content.
(system, user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
assert user["content"] == OFF_TOPIC
assert "<relevance>LOW</relevance>" in system["content"]
assert "DEFLECT_MODE" in system["content"]
assert "HONESTY GATE" in system["content"]
assert "Talos Linux" not in system["content"] # full doc content never sent
assert "<documents>" not in system["content"]
# Durable record: deflected=true + the weak top_score. Deflection is
# only reached when the cosine is under the threshold AND no chunk
# FTS-matches the question — so fts_hits must be zero here.
row = db.scalars(select(QueryLog)).one()
assert row.question == OFF_TOPIC
assert row.deflected is True
assert 0.0 < row.top_score < get_settings().relevance_threshold
assert row.fts_hits == 0
assert row.chunk_hits >= 1
# The retrieval stays durably recorded for threshold tuning
# (observability unchanged — query_log records retrieval, not
# citations; the done frame's [] above is the citation surface).
assert row.sources
def test_done_frame_carries_related_tier_on_grounded_turn(
client, db, seeded_kb: FakeRagLLM
) -> None:
"""Phase 118 (A3) × phase 119 (A1): a grounded turn's done frame
carries the related tier — the ranked docs from rank 6+ after the
suggested set, capped at ``related_max_docs`` (2) — in ``related``,
disjoint from the read-only ``sources`` (nothing was read on this
turn ⇒ no chips; the never-read suggested tier is not on the wire).
The durable record keeps the FULL retrieval (suggested + related +
read, 118-A3 stands)."""
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
done = frames[-1]
assert done["deflected"] is False
# Phase 119 (A1): the citation surface is the agent's READ docs
# only — no read on this turn ⇒ no chips (the retired A4 union
# would have carried the suggested ranks 1–5 here).
sources = [(s["source"], s["path"]) for s in done["sources"]]
assert sources == []
related = done["related"]
# Rank 6–7 for the Kubernetes question (after the top-5 suggested
# set), capped at related_max_docs.
assert [(s["source"], s["path"]) for s in related] == [
("docs", "homelab/quadlet/cache.volume"),
("docs", "homelab/quadlet/compose.container"),
]
assert len(related) <= get_settings().related_max_docs
# The related tier never overlaps the citation surface (the dedupe is
# by (source, path) — the same pattern as the cited docs).
related_keys = {(s["source"], s["path"]) for s in related}
assert set(sources).isdisjoint(related_keys)
# Every ref carries the chip identity fields (the UI row reuses them).
assert all(s["title"] for s in related)
# Durable record: the full retrieval (suggested + related) is logged.
row = db.scalars(select(QueryLog)).one()
assert "docs/homelab/quadlet/cache.volume" in row.sources
assert "docs/homelab/kubernetes.md" in row.sources
def test_deflected_done_frame_carries_weak_hits_in_related(
client, db, seeded_kb: FakeRagLLM
) -> None:
"""Phase 118: on a deflected turn done.sources stays [] (the
phase-112 contract — a deflected answer cites nothing) and
done.related carries rank 6+ after the suggested set (capped at
``related_max_docs``) — the weak hits' visibility home; the weak
hits themselves are the suggested tier (no floor, A3). The durable
record still carries the retrieval (LOCKED A3)."""
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
_, _, frames = _stream_chat(client, OFF_TOPIC)
done = frames[-1]
assert done["deflected"] is True
assert done["sources"] == [] # a deflected answer cites nothing
# Rank 6–7 for the sourdough question (after the top-5 suggested
# set), capped at related_max_docs.
related = done["related"]
assert len(related) <= get_settings().related_max_docs
assert [s["path"] for s in related] == [
"homelab/backups.md",
"homelab/container_gitlab/gitlab-compose.yaml",
]
assert all(s["title"] for s in related)
assert done["suggestions"] # the "Maybe try" chips are unchanged
# Durable record: the weak retrieval stays logged for tuning (A3).
row = db.scalars(select(QueryLog)).one()
assert row.deflected is True
assert row.sources # the weak-hit paths, for threshold tuning
finally:
fastapi_app.dependency_overrides.clear()
def test_related_doc_read_by_agent_is_cited_not_related(
client, db, seeded_kb: FakeRagLLM
) -> None:
"""Phase 119 (A1) × phase 37: an agent-read doc is a chip by
definition — when the agent ``read``s a rank-6+ doc (the related
tier, "nearby docs"), it is the done frame's ONLY chip (the read
docs ARE the citation surface since phase 119; the never-read
suggested tier is not on the wire) and is EXCLUDED from
done.related (a "nearby doc" that was actually used must not read
as nearby — unchanged intent, the dedupe keyed on read docs)."""
scripted = FakeRagLLM(
tool_script=[
[
ToolCallPiece(
id="call_1",
name="read",
arguments={"path": "docs/homelab/quadlet/cache.volume"},
)
]
]
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
done = frames[-1]
assert done["deflected"] is False
sources = [(s["source"], s["path"]) for s in done["sources"]]
# A1: read docs only — the read related doc is the ONLY chip (the
# retired A4 union would have carried 6 here).
assert sources == [("docs", "homelab/quadlet/cache.volume")] # read ⇒ chip
related = [(s["source"], s["path"]) for s in done["related"]]
assert ("docs", "homelab/quadlet/cache.volume") not in related
assert set(sources).isdisjoint(set(related))
# The OTHER related-tier doc (compose.container, rank 7) stays in the tier.
assert ("docs", "homelab/quadlet/compose.container") in related
def test_keyword_question_grounded_by_lexical_hit_despite_weak_cosine(
client, db, seeded_kb: FakeRagLLM,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Phase 09: a name-your-tool question the vector model barely ranks
("kafkabridge" only appears in static-dns.json) must still be grounded
via the FTS branch — HIGH when cosine >= lexical_support_floor AND
fts_hits > 0 (A8 revised 2026-09-14).
The conftest floor (0.15) is above the mock's cosine (~0.134), so we
lower the floor here so the corroborated-lexical path fires."""
from app.config import get_settings # noqa: E402
monkeypatch.setenv("BOR_LEXICAL_SUPPORT_FLOOR", "0.10")
# get_settings is lru_cached — clear the cache so the new env var takes effect.
get_settings.cache_clear()
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
_, _, frames = _stream_chat(client, "How does kafkabridge work?")
done = frames[-1]
assert done["type"] == "done"
assert done["deflected"] is False # weak cosine, but a lexical hit
assert done["suggestions"] == []
# Phase 119 (A1): read docs only — nothing was read ⇒ no chips;
# the lexical hit stays in the durable record (asserted below).
assert done["sources"] == []
(system, _user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
assert "<relevance>HIGH</relevance>" in system["content"] # grounded prompt
row = db.scalars(select(QueryLog)).one()
assert row.deflected is False
assert row.top_score < get_settings().relevance_threshold # weak vector score
assert (row.fts_hits or 0) >= 1 # …and it is the FTS hit that grounds it
assert "docs/homelab/networking/static-dns.json" in row.sources
finally:
# The cache clear is LAST — an assertion that calls get_settings()
# after the clear would re-populate the lru_cache with the
# monkeypatched value and leak it into the next test.
fastapi_app.dependency_overrides.clear()
get_settings.cache_clear()
def test_chat_empty_kb_streams_empty_sources(client, db) -> None:
db.execute(text("TRUNCATE chunks, documents, query_log"))
db.commit()
llm = FakeRagLLM()
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: llm
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
# Nothing retrieved ⇒ nothing to pretend to know: honest deflection.
done = frames[-1]
assert done["type"] == "done"
assert done["deflected"] is True
assert done["sources"] == []
assert 2 <= len(done["suggestions"]) <= 3 # onboarding fallback chips
(system, _user) = llm.seen_messages[0][0], llm.seen_messages[0][1]
assert "DEFLECT_MODE" in system["content"]
assert "nothing close at all" in system["content"]
row = db.scalars(select(QueryLog)).one()
assert row.deflected is True
assert row.top_score == 0.0
assert row.chunk_hits == 0
assert row.sources == ""
def test_chat_embed_failure_yields_error_event(
client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Phase 67: a dead embeddings endpoint retries on the configured
budget — one ``retry`` frame per restart (the attempt about to be
tried, 1-based) — and settles on the existing terminal error frame;
no query_log row. Zero delay keeps the exhaustion path fast."""
broken = FakeRagLLM(embed_error=EmbeddingError("embeddings endpoint down"))
live = get_settings()
monkeypatch.setattr(
chat_api, "get_settings", lambda: _retry_settings(live)
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
retries = live.llm_retries
assert [f["type"] for f in frames] == ["retry"] * retries + ["error"]
assert [f["attempt"] for f in frames if f["type"] == "retry"] == list(
range(2, retries + 2)
)
assert all(
f["max_attempts"] == retries + 1 for f in frames if f["type"] == "retry"
)
assert "embedding" in frames[-1]["detail"]
assert db.scalars(select(QueryLog)).all() == []
def test_chat_mid_stream_failure_yields_error_after_partial_deltas(client, db) -> None:
broken = FakeRagLLM(fail_mid_stream=True)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
assert [f["type"] for f in frames] == ["delta", "error"]
assert "dropped the connection" in frames[1]["detail"]
# No done event, no log row for a turn that never completed.
assert db.scalars(select(QueryLog)).all() == []
def test_error_event_matches_contract_shape(
client, db, seeded_kb, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The SSE error event (PLAN §4) is exactly ``{type, detail, hint}`` —
the client's loading-feedback state machine (phase 06) keys off the
``type``/``detail`` shape to flip to the error state and re-enable
the send button; ``hint`` (phase 114, TODO L6) is additive — present
as ``null`` on reachability frames, old clients ignore it.
``llm_retries=0`` keeps this a single-attempt turn: the contract under
test is the error frame itself, not the phase-67 retry loop."""
broken = FakeRagLLM(embed_error=EmbeddingError("embeddings endpoint down"))
live = get_settings()
monkeypatch.setattr(
chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=0)
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
assert len(frames) == 1
event = frames[0]
assert set(event.keys()) == {"type", "detail", "hint"}
assert event["type"] == "error"
assert isinstance(event["detail"], str) and event["detail"]
assert event["hint"] is None # reachability frame — no too-long hint
def test_chat_db_down_returns_503_json(client, monkeypatch) -> None:
monkeypatch.setattr(chat_api, "db_available", lambda: False)
r = client.post("/api/chat", json={"message": "hello"})
assert r.status_code == 503
assert "offline" in r.json()["detail"]
def test_chat_retrieval_failure_yields_error_event(client, db, seeded_kb, monkeypatch) -> None:
def boom(*_a: Any, **_k: Any) -> Any:
raise RuntimeError("db exploded")
monkeypatch.setattr(chat_api, "retrieve", boom)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
assert [f["type"] for f in frames] == ["error"]
assert "offline mid-question" in frames[0]["detail"]
assert db.scalars(select(QueryLog)).all() == []
class _BrokenCommitSession:
"""Pass-through session whose ``commit()`` raises (query_log failure)."""
def __init__(self, real: Any) -> None:
self._real = real
def __enter__(self) -> _BrokenCommitSession:
return self
def __exit__(self, *args: Any) -> None:
self._real.close()
def commit(self) -> None:
raise RuntimeError("query_log commit failed")
def __getattr__(self, name: str) -> Any:
return getattr(self._real, name)
def test_chat_query_log_failure_still_sends_done(
client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
"""SEC-14-04: even when the query_log write fails, the answer still
goes out. The chat endpoint uses short-lived sessions (SessionLocal)
for query_log writes — monkeypatch SessionLocal to return a broken
session that fails on commit."""
from app.db import SessionLocal as real_SessionLocal
def broken_session_factory():
real = real_SessionLocal()
return _BrokenCommitSession(real)
monkeypatch.setattr(chat_api, "SessionLocal", broken_session_factory)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
# The answer (and the done event) went out despite the log-row failure.
assert [f["type"] for f in frames if f["type"] == "delta"]
assert frames[-1]["type"] == "done"
assert frames[-1]["deflected"] is False
# ---------- phase 37: agent document tools on grounded turns ----------
async def _collect_run_agent(
llm: FakeRagLLM,
db: Session,
system_prompt: str,
settings: Settings,
seed_docs: list[Document],
) -> tuple[list[Any], agent.AgentHolder]:
"""Consume one ``run_agent`` turn, returning the yielded pieces (in
order) and the holder. Phase 95 (task 01): the direct agent-loop
drive — the agent-loop yield order on the real prompt path, the
complement of the endpoint-level ``tool_result`` SSE tests below
(task 02)."""
holder = agent.AgentHolder()
pieces: list[Any] = []
async for piece in agent.run_agent(
llm, # pyright: ignore[reportArgumentType] # duck-typed LLMClient
lambda: db, # SEC-14-04: session factory (integration tests reuse the fixture session)
system_prompt=system_prompt,
user_message=QUESTION,
seed_docs=seed_docs,
settings=settings,
holder=holder,
):
pieces.append(piece)
return pieces, holder
def test_read_cap_truncates_and_yields_tool_result_on_real_prompt_path(
db,
) -> None:
"""Phase 95 (task 01): on the REAL prompt path (a real Postgres
document + the real ``build_high_prompt``), a ``read`` of a document
LONGER than ``settings.read_max_chars`` truncates the result the model
sees — first ``cap`` chars + the shared :data:`TRUNCATION_MARKER` + the
pinned grep-pointer notice — and ``run_agent`` yields exactly ONE
``ToolResultPiece``: AFTER the read's ``tool`` frame (the matching
``ToolCallPiece``) and BEFORE the next model round. The endpoint-level
``tool_result`` SSE frame is asserted separately below (task 02); this
pins the agent-loop yield order on the real prompt path."""
cap = 100
content = "K" * (cap + 40) # 40 chars over the cap
doc = Document(
id=uuid.uuid4(),
source="docs",
path="big.md",
full_path="/tmp/big.md",
title="Big Doc",
content=content,
content_hash="1" * 64,
created_at=_FIXTURE_CREATED_AT,
)
db.add(doc)
db.commit()
try:
# The real prompt path: the actual HIGH prompt for the one doc.
system_prompt = build_high_prompt([doc])
settings = Settings(_env_file=None, read_max_chars=cap) # pyright: ignore[reportCallIssue]
scripted = FakeRagLLM(
tool_script=[
[
ToolCallPiece(
id="call_1", name="read", arguments={"path": "docs/big.md"}
)
]
# the answer request (tools still offered, script
# exhausted) falls back to the thinking + answer stream
]
)
pieces, holder = asyncio.run(
_collect_run_agent(scripted, db, system_prompt, settings, seed_docs=[])
)
# The model's context carried the truncated read — the first cap
# chars, then the shared marker + the pinned grep-pointer notice
# (so a downstream grep is the model's path to the rest). The
# fake records the (mutated-in-place) messages list, so the same
# tool message is aliased across requests — they all carry the
# same content; take the last.
tool_msgs = [
m for r in scripted.seen_messages for m in r if m.get("role") == "tool"
]
assert tool_msgs, "the executed read must be appended as a tool message"
body = tool_msgs[-1]["content"]
# Phase 106, D5: the date rides every read — the SECOND line
# (first line byte-identical — the mock's header contract).
assert body.startswith(
"Document docs/big.md:\ndate: 2024-06-15\n" + content[:cap]
)
assert TRUNCATION_MARKER in body
assert (
READ_TRUNCATION_NOTICE.format(shown=cap, total=len(content)) in body
)
# The yield order: the read's ToolCallPiece, then the ONE
# ToolResultPiece, then the next round's answer content.
kinds: list[str] = []
for p in pieces:
if isinstance(p, ToolCallPiece):
kinds.append("toolcall")
elif isinstance(p, ToolResultPiece):
kinds.append("toolresult")
elif isinstance(p, StreamPiece):
kinds.append(p.kind)
assert kinds.count("toolresult") == 1
assert kinds.index("toolcall") < kinds.index("toolresult")
assert kinds.index("toolresult") < kinds.index("content")
# The piece carries (argument, shown, total) — the raw
# source/path the model passed (what the tool frame carries), the
# cap kept, the true length.
(result_piece,) = [p for p in pieces if isinstance(p, ToolResultPiece)]
assert result_piece.name == "read"
assert result_piece.argument == "docs/big.md"
assert result_piece.truncated is True
assert result_piece.chars_shown == cap
assert result_piece.chars_total == len(content)
# Holder accounting: a truncated read is still a SUCCESSFUL call
# (counted + added to context); the tuple is the signal only.
assert holder.tool_calls == 1
assert holder.read_docs == [doc]
assert holder.read_truncations == [("docs/big.md", cap, len(content))]
finally:
db.delete(doc)
db.commit()
def test_read_at_or_under_cap_yields_no_tool_result_on_real_prompt_path(
db,
) -> None:
"""Phase 95 (task 01): the complement — a ``read`` of a document at or
under the cap on the real prompt path is byte-identical to the
pre-phase-95 agent loop: NO ``ToolResultPiece``, no holder entry, no
marker in the model's context."""
cap = 100
content = "K" * cap # exactly at the cap → fits, not truncated
doc = Document(
id=uuid.uuid4(),
source="docs",
path="fits.md",
full_path="/tmp/fits.md",
title="Fits Doc",
content=content,
content_hash="2" * 64,
created_at=_FIXTURE_CREATED_AT,
)
db.add(doc)
db.commit()
try:
system_prompt = build_high_prompt([doc])
settings = Settings(_env_file=None, read_max_chars=cap) # pyright: ignore[reportCallIssue]
scripted = FakeRagLLM(
tool_script=[
[
ToolCallPiece(
id="call_1", name="read", arguments={"path": "docs/fits.md"}
)
]
]
)
pieces, holder = asyncio.run(
_collect_run_agent(scripted, db, system_prompt, settings, seed_docs=[])
)
# No ToolResultPiece, no holder entry.
assert not any(isinstance(p, ToolResultPiece) for p in pieces)
assert holder.read_truncations == []
# The model's context is the whole document, the pre-phase-95
# read result plus the phase-106 D5 date line (no marker, no
# notice). (The fake aliases the mutated messages list, so
# take the last tool msg.)
tool_msgs = [
m for r in scripted.seen_messages for m in r if m.get("role") == "tool"
]
assert tool_msgs, "the executed read must be appended as a tool message"
assert (
tool_msgs[-1]["content"]
== "Document docs/fits.md:\ndate: 2024-06-15\n" + content
)
assert TRUNCATION_MARKER not in tool_msgs[-1]["content"]
# Still a successful read.
assert holder.tool_calls == 1
assert holder.read_docs == [doc]
finally:
db.delete(doc)
db.commit()
def _insert_big_doc(db, content: str) -> Document:
"""One bare ``documents`` row (no chunks — the ``read`` lookup is a
(source, path) identity match, not a retrieval) for the SSE-level
read-cap tests: a document the model can only reach through the
``read`` tool."""
doc = Document(
id=uuid.uuid4(),
source="docs",
path="big-read.md",
full_path="/tmp/big-read.md",
title="Big Read Doc",
content=content,
content_hash="3" * 64,
created_at=_FIXTURE_CREATED_AT,
)
db.add(doc)
db.commit()
return doc
def test_truncated_read_streams_tool_result_frame_after_tool_frame(
client,
db,
seeded_kb: FakeRagLLM,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Phase 95 (task 02, the A15 extension): a grounded turn whose
scripted ``read`` hits a document LONGER than ``read_max_chars``
(the cap lowered via the settings override — the task 01
``Settings(_env_file=None, read_max_chars=…)`` pattern) streams the
``tool`` → ``tool_result`` → ``delta…`` → ``done`` sequence: EXACTLY
ONE ``tool_result`` frame, AFTER the matching ``tool`` frame (the
line is already on screen) and BEFORE the next round's first frame,
with the right shape and counts (``chars_shown`` = the cap,
``chars_total`` = the true length). The model's context carried the
truncated read (marker + pinned grep-pointer notice); the read is
still cited (a truncated read is a successful call)."""
cap = 100
content = "K" * (cap + 150)
doc = _insert_big_doc(db, content)
live = get_settings()
monkeypatch.setattr(
chat_api,
"get_settings",
lambda: Settings(
_env_file=None, # pyright: ignore[reportCallIssue]
relevance_threshold=live.relevance_threshold,
read_max_chars=cap,
),
)
scripted = FakeRagLLM(
tool_script=[
[
ToolCallPiece(
id="call_1", name="read", arguments={"path": "docs/big-read.md"}
)
]
# the answer request still carries the tools (1 round < the
# default cap of 10); the script is exhausted, so the fake
# falls back to the thinking + answer stream
]
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
db.delete(doc)
db.commit()
types = [f["type"] for f in frames]
assert "error" not in types
tool_i = types.index("tool")
tool_result_i = types.index("tool_result")
# Exactly one tool_result frame…
assert types.count("tool_result") == 1
# …AFTER the matching tool frame and BEFORE the next model round's
# first frame (the answer's deltas): tool → tool_result → delta…
assert tool_i + 1 == tool_result_i
assert tool_result_i < min(i for i, t in enumerate(types) if t == "delta")
# The frame's exact shape: the additive seventh event type carries
# the name/argument of the matching tool frame + the counts.
frame = frames[tool_result_i]
assert set(frame) == {
"type",
"name",
"argument",
"truncated",
"chars_shown",
"chars_total",
}
assert frame["name"] == frames[tool_i]["name"] == "read"
assert frame["argument"] == frames[tool_i]["argument"] == "docs/big-read.md"
assert frame["truncated"] is True
assert frame["chars_shown"] == cap # the cap kept
assert frame["chars_total"] == len(content) # the true length
# The LLM's context carried the honest truncation: first cap chars +
# the shared marker + the pinned grep-pointer notice (the fake
# aliases the mutated messages list — take the last tool msg).
tool_msgs = [
m for r in scripted.seen_messages for m in r if m.get("role") == "tool"
]
assert tool_msgs
body = tool_msgs[-1]["content"]
# Phase 106, D5: the date rides every read — the SECOND line.
assert body.startswith(f"Document docs/big-read.md:\ndate: 2024-06-15\n{content[:cap]}")
assert TRUNCATION_MARKER in body
assert READ_TRUNCATION_NOTICE.format(shown=cap, total=len(content)) in body
# The truncated read is still a SUCCESSFUL call — it is the frame's
# only chip (phase 119, A1: read docs only; the suggested kubernetes
# doc was never read, so it is not on the wire).
done = frames[-1]
assert done["type"] == "done" and done["deflected"] is False
assert [(s["source"], s["path"]) for s in done["sources"]] == [
("docs", "big-read.md")
]
def test_untruncated_read_streams_no_tool_result_frame(
client,
db,
seeded_kb: FakeRagLLM,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Phase 95 (task 02): the complement at the SSE level — a ``read``
of a document AT OR UNDER the cap (the same long document, cap
raised past its true length) streams NO ``tool_result`` frame (one
frame = one noteworthy event; the six pre-existing event types are
byte-identical), the ``tool`` frame is unchanged, and the model's
context is the WHOLE document (no marker, no notice)."""
content = "K" * 250
doc = _insert_big_doc(db, content)
live = get_settings()
monkeypatch.setattr(
chat_api,
"get_settings",
lambda: Settings(
_env_file=None, # pyright: ignore[reportCallIssue]
relevance_threshold=live.relevance_threshold,
read_max_chars=10_000, # far over the doc's true length
),
)
scripted = FakeRagLLM(
tool_script=[
[
ToolCallPiece(
id="call_1", name="read", arguments={"path": "docs/big-read.md"}
)
]
]
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
db.delete(doc)
db.commit()
types = [f["type"] for f in frames]
assert "error" not in types
assert types.count("tool_result") == 0 # one frame = one noteworthy event
assert types.count("tool") == 1
(tool_frame,) = [f for f in frames if f["type"] == "tool"]
assert set(tool_frame) == {"type", "name", "argument"} # byte-identical shape
assert tool_frame["argument"] == "docs/big-read.md"
assert frames[-1]["type"] == "done" and frames[-1]["deflected"] is False
# The model saw the WHOLE document — no marker, no notice.
tool_msgs = [
m for r in scripted.seen_messages for m in r if m.get("role") == "tool"
]
assert tool_msgs
assert (
tool_msgs[-1]["content"]
== "Document docs/big-read.md:\ndate: 2024-06-15\n" + content
)
assert TRUNCATION_MARKER not in tool_msgs[-1]["content"]
def test_grounded_turn_streams_tool_frames_and_cites_read_doc(
client, db, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture
) -> None:
"""(a) Grounded turn with tool calls: the event sequence is
``thinking?/tool/tool/delta…/done``; ``done.sources`` is the read
document (phase 119, A1 — the read docs are the citation surface)
and the ``query_log`` row includes it (deduped, order preserved);
the per-turn log line carries ``tool_calls=2``. Phase 45: the
agent loop keeps offering the tools for the whole turn — the round
cap (not per-tool budgets) is the bound."""
scripted = FakeRagLLM(
tool_script=[
[
StreamPiece("thinking", "Let me list what is indexed…"),
ToolCallPiece(id="call_1", name="ls", arguments={}),
],
[
ToolCallPiece(
id="call_2",
name="read",
arguments={"path": "docs/homelab/backups.md"},
)
],
# the answer request still carries the tools (2 rounds < the
# default cap of 10); the fake's tool_script is exhausted, so
# it falls back to the thinking + answer stream
]
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
caplog.set_level(logging.INFO, logger="app.chat")
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
types = [f["type"] for f in frames]
assert types[0] == "thinking"
assert types[1] == "tool" and types[2] == "tool" # the two executed calls
assert "error" not in types
assert types[3:-1] == ["delta"] * (len(types) - 4) # deltas, then done last
assert frames[-1]["type"] == "done"
list_frame, read_frame = frames[1], frames[2]
assert set(list_frame) == {"type", "name", "argument"}
assert list_frame["name"] == "ls"
assert list_frame["argument"] is None # no ``path`` argument was passed
assert set(read_frame) == {"type", "name", "argument"}
assert read_frame["name"] == "read"
# Phase 70: the frame's argument is the single string the model
# passed — the combined ``source/path``.
assert read_frame["argument"] == "docs/homelab/backups.md"
deltas = [f for f in frames if f["type"] == "delta"]
assert len(deltas) >= 2 # genuinely streamed
assert "".join(d["text"] for d in deltas) == scripted.answer
done = frames[-1]
assert done["deflected"] is False
# Phase 119 (A1): done.sources = the agent's READ docs only — the
# read doc is the frame's only chip (the suggested retrieval docs
# are not on the wire; they stay in the durable record below).
sources = [(s["source"], s["path"]) for s in done["sources"]]
assert sources == [("docs", "homelab/backups.md")] # the read doc is cited
assert done["sources"][0]["title"] == "Backup Strategy"
# Phase 45: the tools stay offered on every request — the round cap
# (not spent budgets) bounds the loop, and the model answered while
# still being offered the tools (2 rounds < default cap 10).
assert len(scripted.seen_messages) == 3
assert scripted.seen_tools[0] == AGENT_TOOLS
assert scripted.seen_tools[1] == AGENT_TOOLS
assert scripted.seen_tools[2] == AGENT_TOOLS
# The query_log row carries the same combined source list.
(row,) = db.scalars(select(QueryLog)).all()
assert row.deflected is False
assert "docs/homelab/kubernetes.md" in row.sources
assert row.sources.endswith(", docs/homelab/backups.md") # the read doc, last
# The required per-turn log line (PLAN §9 extension) counts both calls
# and lists the combined sources (retrieval + read).
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert lines and "tool_calls=2" in lines[-1]
assert "'docs/homelab/kubernetes.md'" in lines[-1]
assert "'docs/homelab/backups.md'" in lines[-1]
assert "scaffold_stripped=0" in lines[-1] # phase 71: uniform clean-turn field
def test_grounded_turn_streams_grep_tool_frames(
client, db, seeded_kb: FakeRagLLM
) -> None:
"""Phase 68 (renamed ``grep`` in phase 70): a scripted ``grep`` call
streams as ``{type: "tool", name: "grep", argument: <pattern>}`` —
the raw pattern is the frame's ``argument`` (the UI renders the
"searching for" line from it). A non-string pattern — a model error
the backend refuses — yields ``argument: null``. A grep adds no
source (locked A5): with no read on the turn, ``done.sources`` is
empty (phase 119, A1)."""
scripted = FakeRagLLM(
tool_script=[
[
ToolCallPiece(id="call_1", name="grep", arguments={"pattern": "Cilium"}),
],
[
ToolCallPiece(
id="call_2",
name="grep",
arguments={"pattern": 42}, # model error: non-string
),
],
# the answer request still carries the tools (2 rounds < the
# default cap of 10); the fake's tool_script is exhausted, so
# it falls back to the thinking + answer stream
]
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
types = [f["type"] for f in frames]
assert "error" not in types
assert len(scripted.seen_tools) == 3 # both greps executed (rounds)
tool_frames = [f for f in frames if f["type"] == "tool"]
assert len(tool_frames) == 2
first, second = tool_frames
assert set(first) == {"type", "name", "argument"}
assert first["name"] == "grep"
assert first["argument"] == "Cilium" # the raw pattern
assert set(second) == {"type", "name", "argument"}
assert second["name"] == "grep"
assert second["argument"] is None # the non-string pattern → null
# The greps still answered: deltas, then a grounded done.
assert [f for f in frames if f["type"] == "delta"]
done = frames[-1]
assert done["type"] == "done" and done["deflected"] is False
# A grep adds no source (A5) and nothing was read (A1) ⇒ no chips.
assert done["sources"] == []
def test_tool_frames_carry_the_model_arguments_regardless_of_execution(
client, db, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture
) -> None:
"""Phase 70 pins: the frame's ``argument`` is the single string
argument the model passed — an ``ls`` frame carries the scope when
the model gave one (null only when it is omitted, pinned above) —
and frame emission is execution-independent: a rejected call (an
unknown ``read`` path) still streams its frame with the model's
argument as-is. The rejected read adds no source (``done.sources``
carries no chip — the read failed and nothing else was read,
phase 119 A1), and rejected calls count nothing
(``tool_calls=1`` — only the executed scoped ``ls``)."""
# The scoped ``ls`` source-name check reads the registry — insert a
# row resolving to ``docs`` (the fixture's source name) and delete
# it again afterwards.
src = GitSource(url="https://github.com/reese/docs.git", kind="git")
db.add(src)
db.commit()
try:
scripted = FakeRagLLM(
tool_script=[
[ToolCallPiece(id="call_1", name="ls", arguments={"path": "docs"})],
[
ToolCallPiece(
id="call_2", name="read", arguments={"path": "docs/homelab/nope.md"}
)
],
]
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
caplog.set_level(logging.INFO, logger="app.chat")
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
finally:
db.execute(delete(GitSource).where(GitSource.id == src.id))
db.commit()
types = [f["type"] for f in frames]
assert "error" not in types
# Both calls stream a frame — the rejected read included.
tool_frames = [f for f in frames if f["type"] == "tool"]
assert len(tool_frames) == 2
ls_frame, read_frame = tool_frames
assert set(ls_frame) == {"type", "name", "argument"}
assert ls_frame["name"] == "ls"
assert ls_frame["argument"] == "docs" # the model's scope, as passed
assert set(read_frame) == {"type", "name", "argument"}
assert read_frame["name"] == "read"
# The rejected call's frame still carries the model's argument as
# passed — frame emission is execution-independent.
assert read_frame["argument"] == "docs/homelab/nope.md"
# The rejected read adds no source — and the read-only surface is
# empty on this turn (nothing was read; the refused read cites
# nothing, phase 119 A1).
done = frames[-1]
assert done["type"] == "done" and done["deflected"] is False
assert done["sources"] == []
# The rejected call counts nothing — only the executed scoped ls.
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert lines and "tool_calls=1" in lines[-1]
def test_deflected_turn_stays_byte_identical_without_tools(
client, db, seeded_kb: FakeRagLLM
) -> None:
"""(b) Deflected turn: the agent loop never runs — no ``tool``
frames, and the frame sequence is byte-identical to the plain fake's
direct-``chat_stream`` output even for a fake scripted to call tools
(its script is never consumed). The LLM was called once, without a
``tools`` key."""
# The scripted read targets a doc OUTSIDE the OFF_TOPIC retrieval
# top-7 (tables.md ranks 11th — not suggested, not rank 6+ related),
# so "never read" stays distinguishable from "retrieved" in the
# durable record below (phase 118: backups.md — the pre-phase read
# target — now rides the rank-6+ related tier, durably recorded).
scripted = FakeRagLLM(
tool_script=[
[ToolCallPiece(id="call_1", name="ls", arguments={})],
[
ToolCallPiece(
id="call_2",
name="read",
arguments={"path": "docs/homelab/tables.md"},
)
],
[StreamPiece("content", "never used — the agent never runs")],
]
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
_, _, baseline = _stream_chat(client, OFF_TOPIC)
finally:
fastapi_app.dependency_overrides.clear()
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
_, _, frames = _stream_chat(client, OFF_TOPIC)
finally:
fastapi_app.dependency_overrides.clear()
assert frames == baseline # byte-identical to the direct path
assert not any(f["type"] == "tool" for f in frames)
assert frames[-1]["type"] == "done" and frames[-1]["deflected"] is True
assert len(scripted.tool_script) == 3 # the script was never consumed
assert len(scripted.seen_messages) == 1
assert scripted.seen_tools == [None] # one request, no tools key
# The read document never sneaks into the deflected turn's record.
(row,) = [
r
for r in db.scalars(select(QueryLog)).all()
if r.question == OFF_TOPIC
][-1:]
assert row.deflected is True
assert "tables.md" not in row.sources
def test_zero_max_rounds_reproduce_pre_phase_single_request(
client,
db,
seeded_kb: FakeRagLLM,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
"""(c) ``BOR_AGENT_MAX_ROUNDS=0``: no ``tool`` frames, exactly one
request **without** a ``tools`` key (the pre-phase request shape),
``done.sources`` unchanged, and ``tool_calls=0`` in the log line —
the kill switch survives the phase-45 budget removal."""
scripted = FakeRagLLM(
tool_script=[
[ToolCallPiece(id="call_1", name="ls", arguments={})],
[
ToolCallPiece(
id="call_2",
name="read",
arguments={"path": "docs/homelab/backups.md"},
)
],
]
)
live = get_settings()
monkeypatch.setattr(
chat_api,
"get_settings",
lambda: Settings(
_env_file=None, # pyright: ignore[reportCallIssue]
relevance_threshold=live.relevance_threshold,
agent_max_rounds=0,
),
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
caplog.set_level(logging.INFO, logger="app.chat")
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
assert not any(f["type"] == "tool" for f in frames)
assert "error" not in [f["type"] for f in frames]
done = frames[-1]
assert done["type"] == "done"
assert done["deflected"] is False
# Phase 119 (A1): nothing was read ⇒ no chips (the kill-switch
# turn's citation surface is empty; the retrieval stays in the
# durable record below).
assert done["sources"] == []
# Exactly one request, and it carried no ``tools`` key at all — the
# scripted tool calls were never even offered a chance.
assert len(scripted.seen_messages) == 1
assert scripted.seen_tools == [None]
assert len(scripted.tool_script) == 2 # never consumed
(row,) = db.scalars(select(QueryLog)).all()
assert "docs/homelab/kubernetes.md" in row.sources
assert "backups.md" not in row.sources
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert lines and "tool_calls=0" in lines[-1]
assert "scaffold_stripped=0" in lines[-1] # phase 71: uniform clean-turn field
def test_tool_execution_db_failure_yields_error_event(
client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A tool call that hits a dead DB mid-stream gets the same structured
``error`` event as the pre-stream retrieval path — never a severed
stream (the "never stale" contract, PLAN §7.4)."""
scripted = FakeRagLLM(
tool_script=[[ToolCallPiece(id="call_1", name="ls", arguments={})]]
)
def boom(*_a: Any, **_k: Any) -> Any:
raise RuntimeError("db exploded mid tool call")
# Phase 94: the no-arg ``ls`` executes through ``ls_top`` — the
# failure hook moves with the rewrite (same contract: the tool
# frame goes out first, the structured error ends the turn).
monkeypatch.setattr(agent, "ls_top", boom)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
# The ``tool`` frame went out first (the model requested the call);
# the failed execution ends the turn with the structured error event.
assert [f["type"] for f in frames] == ["tool", "error"]
assert frames[0]["name"] == "ls"
assert "offline mid-question" in frames[1]["detail"]
assert db.scalars(select(QueryLog)).all() == [] # no row for a failed turn
# ---------- phase 67: LLM retries before the first token ----------
def _retry_settings(live: Settings, **overrides: Any) -> Settings:
"""Settings for the retry tests: the live (mock-calibrated) threshold
plus the phase-67 knobs, with a ZERO delay so the suite never sleeps.
(The 5 s default is unit-pinned in ``tests/unit/test_config.py``.)"""
kwargs: dict[str, Any] = {
"relevance_threshold": live.relevance_threshold,
"llm_retry_delay": 0.0,
}
kwargs.update(overrides)
return Settings(_env_file=None, **kwargs) # pyright: ignore[reportCallIssue]
def test_embed_failure_retries_then_turn_completes(
client,
db,
seeded_kb: FakeRagLLM,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A dead-then-recovered embeddings endpoint: one SSE ``retry`` frame
(the attempt about to be tried, 1-based) ahead of the normal answer
frames; the turn completes and the per-turn log line counts the
retry (``retries=1``)."""
flaky = FakeRagLLM(embed_fail_count=1)
live = get_settings()
monkeypatch.setattr(
chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=1)
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: flaky
try:
caplog.set_level(logging.INFO, logger="app.chat")
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
assert frames[0] == {"type": "retry", "attempt": 2, "max_attempts": 2}
assert not any(f["type"] == "error" for f in frames)
deltas = [f for f in frames if f["type"] == "delta"]
assert len(deltas) >= 2
assert "".join(d["text"] for d in deltas) == flaky.answer
assert frames[-1]["type"] == "done"
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert lines and "retries=1" in lines[-1]
assert "scaffold_stripped=0" in lines[-1] # phase 71: uniform clean-turn field
def test_embed_failure_exhausts_retries_then_terminal_error(
client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A dead embeddings endpoint (``llm_retries=2`` → 3 attempts): one
``retry`` frame per restart (attempts 2 and 3 of 3), then the
EXISTING terminal error frame — the copy is unchanged, no query_log
row."""
dead = FakeRagLLM(embed_fail_count=99) # every attempt fails
live = get_settings()
monkeypatch.setattr(
chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=2)
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: dead
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
assert [f["type"] for f in frames] == ["retry", "retry", "error"]
assert [f["attempt"] for f in frames if f["type"] == "retry"] == [2, 3]
assert all(f["max_attempts"] == 3 for f in frames if f["type"] == "retry")
assert "embedding" in frames[-1]["detail"]
assert db.scalars(select(QueryLog)).all() == []
def test_deflected_stream_retries_before_the_first_piece(
client,
db,
seeded_kb: FakeRagLLM,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Deflected answer stream: the first attempt dies before any piece,
the restart streams — a ``retry`` frame ahead of the deltas, the
request restarted with the same messages (no tools key), and the
per-turn log line counts the retry."""
flaky = FakeRagLLM(stream_fail_count=1)
live = get_settings()
monkeypatch.setattr(
chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=2)
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: flaky
try:
caplog.set_level(logging.INFO, logger="app.chat")
_, _, frames = _stream_chat(client, OFF_TOPIC)
finally:
fastapi_app.dependency_overrides.clear()
assert frames[0] == {"type": "retry", "attempt": 2, "max_attempts": 3}
rest = frames[1:]
assert all(f["type"] in ("delta", "done") for f in rest)
assert "".join(f["text"] for f in rest if f["type"] == "delta") == flaky.answer
assert rest[-1]["type"] == "done" and rest[-1]["deflected"] is True
assert len(flaky.seen_messages) == 2 # the request was restarted
assert flaky.seen_tools == [None, None] # …byte-identical (no tools key)
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert lines and "retries=1" in lines[-1]
assert "scaffold_stripped=0" in lines[-1] # phase 71: uniform clean-turn field
def test_deflected_stream_failure_after_first_frame_is_terminal(
client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Locked A2: a stream failure AFTER the first output frame is
terminal — no ``retry`` frame, the existing error copy, no row (a
partial answer is never redone)."""
broken = FakeRagLLM(fail_mid_stream=True)
live = get_settings()
monkeypatch.setattr(
chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=3)
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
try:
_, _, frames = _stream_chat(client, OFF_TOPIC)
finally:
fastapi_app.dependency_overrides.clear()
assert [f["type"] for f in frames] == ["delta", "error"]
assert not any(f["type"] == "retry" for f in frames)
assert "dropped the connection" in frames[1]["detail"]
assert db.scalars(select(QueryLog)).all() == []
def test_zero_retries_keep_the_pre_phase_wire_shape(
client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The ``BOR_LLM_RETRIES=0`` kill switch: one attempt, the existing
terminal error frame, no ``retry`` frames — the pre-phase-67
byte-identical wire shape."""
broken = FakeRagLLM(embed_error=EmbeddingError("embeddings endpoint down"))
live = get_settings()
monkeypatch.setattr(
chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=0)
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
assert len(frames) == 1
assert frames[0]["type"] == "error"
assert "embedding" in frames[0]["detail"]
assert not any(f["type"] == "retry" for f in frames)
# ---------- phase 71: the deterministic scaffolding guardrail (deflected path) ----------
def _scaffold_span() -> str:
"""The raw span from the 2026-09-03 incident (the E2E mock's trigger,
task 05) — a complete span the filter strips in full."""
return "<|tool_call_start|>[read(path='/homelab/backup-notes.md')]<|tool_call_end|>"
def test_deflected_scaffolding_only_reply_recovers_once(
client,
db,
seeded_kb: FakeRagLLM,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
"""(a) A deflected reply that is pure scaffolding streams ZERO delta
frames (no raw tokens on the wire); the one bounded recovery —
``tools=None``, the correction folded into the single system prompt,
a fresh filter, the same retry budget — streams the clean answer, the
turn settles with ``done`` + a query_log row, and the log line counts
the stripped chars (the recovery does not bump ``retries=N``)."""
span = _scaffold_span()
clean = "I don't have that on hand — try one of the chips below?"
flaky = FakeRagLLM(answer_sequence=[span, clean])
live = get_settings()
monkeypatch.setattr(chat_api, "get_settings", lambda: _retry_settings(live))
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: flaky
try:
caplog.set_level(logging.INFO, logger="app.chat")
_, _, frames = _stream_chat(client, OFF_TOPIC)
finally:
fastapi_app.dependency_overrides.clear()
# The raw tokens never reach the wire; the deltas reassemble to the
# clean recovery answer.
assert span not in json.dumps(frames)
deltas = [f for f in frames if f["type"] == "delta"]
assert "".join(d["text"] for d in deltas) == clean
assert not any(f["type"] == "error" for f in frames)
done = frames[-1]
assert done["type"] == "done" and done["deflected"] is True
# Exactly two requests: the stripped round + the one recovery, both
# without a tools key…
assert len(flaky.seen_messages) == 2
assert flaky.seen_tools == [None, None]
# …and the recovery's system prompt is the ORIGINAL deflected prompt
# with the correction folded in (a single system message — the user
# message stays last).
recovered = flaky.seen_messages[1]
assert len(recovered) == 2
assert recovered[1] == {"role": "user", "content": OFF_TOPIC}
first_system = flaky.seen_messages[0][0]["content"]
assert "DEFLECT_MODE" in first_system
assert recovered[0] == {
"role": "system",
"content": first_system + "\n" + agent.CORRECTION_INSTRUCTION,
}
# The turn settled normally: one query_log row…
(row,) = db.scalars(select(QueryLog)).all()
assert row.deflected is True
# …and the log line carries the summed stripped count (the clean
# recovery stripped nothing) with retries untouched.
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert lines and f"scaffold_stripped={len(span)}" in lines[-1]
assert "retries=0" in lines[-1] # the recovery is not an endpoint-retry
def test_deflected_scaffolding_twice_settles_malformed(
client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
"""(b) The recovery answer is scaffolding again — a second empty
reply is terminal: the DEDICATED error frame (the exact copy), no
``done``, no query_log row — the standard ``LLMError`` terminal
shape (phase 114: the additive ``hint`` field is ``null`` here) —
and no third request (at most one recovery per turn)."""
span = _scaffold_span()
dead = FakeRagLLM(answer_sequence=[span, span])
live = get_settings()
monkeypatch.setattr(chat_api, "get_settings", lambda: _retry_settings(live))
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: dead
try:
_, _, frames = _stream_chat(client, OFF_TOPIC)
finally:
fastapi_app.dependency_overrides.clear()
assert [f["type"] for f in frames] == ["error"]
assert frames[0]["detail"] == (
"The model returned a malformed reply — please try again."
)
assert set(frames[0].keys()) == {
"type",
"detail",
"hint",
} # the contract shape (phase 114: additive hint — null here)
assert frames[0]["hint"] is None
assert span not in json.dumps(frames)
assert not any(f["type"] == "done" for f in frames)
assert db.scalars(select(QueryLog)).all() == []
assert len(dead.seen_messages) == 2 # round + one recovery — no more
assert dead.seen_tools == [None, None]
def test_deflected_mixed_scaffolding_and_content_needs_no_recovery(
client,
db,
seeded_kb: FakeRagLLM,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
"""(c) Real visible content plus scaffolding: the clean remainder
streams (no raw tokens on the wire), NO recovery runs, and the log
line counts the stripped span (``scaffold_stripped>0``)."""
span = _scaffold_span()
mixed = f"I don't have that. {span} Try the chips below?"
flaky = FakeRagLLM(answer=mixed)
live = get_settings()
monkeypatch.setattr(chat_api, "get_settings", lambda: _retry_settings(live))
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: flaky
try:
caplog.set_level(logging.INFO, logger="app.chat")
_, _, frames = _stream_chat(client, OFF_TOPIC)
finally:
fastapi_app.dependency_overrides.clear()
assert span not in json.dumps(frames)
deltas = [f for f in frames if f["type"] == "delta"]
assert "".join(d["text"] for d in deltas) == "I don't have that. Try the chips below?"
assert not any(f["type"] == "error" for f in frames)
assert frames[-1]["type"] == "done"
assert len(flaky.seen_messages) == 1 # the clean content stands — no recovery
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert lines and f"scaffold_stripped={len(span)}" in lines[-1]
# ---------- phase 74: client-provided history (with prior thinking) ----------
#: The client's prior turns (oldest first — the ``bor.chat.v1`` record
#: minus the current question): two user turns, two brain turns, the
#: FIRST brain turn carrying a prior thinking block (A4) and the second
#: not (the ``reasoning_content`` gate has both shapes on one request).
HISTORY: list[dict[str, Any]] = [
{"who": "user", "text": "What port does Tailscale run on?"},
{
"who": "brain",
"text": "Tailscale runs on 41641/udp.",
"thinking": "The Tailscale wire protocol uses 41641/udp.",
},
{"who": "user", "text": "And the subnet router?"},
{"who": "brain", "text": "The subnet router shares the same port."},
]
#: What :func:`app.rag.prompts.history_to_messages` must produce for
#: :data:`HISTORY` — chronological, ``reasoning_content`` ONLY on the
#: turn that had thinking.
HISTORY_MESSAGES: list[dict[str, Any]] = [
{"role": "user", "content": "What port does Tailscale run on?"},
{
"role": "assistant",
"content": "Tailscale runs on 41641/udp.",
"reasoning_content": "The Tailscale wire protocol uses 41641/udp.",
},
{"role": "user", "content": "And the subnet router?"},
{"role": "assistant", "content": "The subnet router shares the same port."},
]
def _stream_chat_with_history(
client: TestClient, message: str, history: list[dict[str, Any]]
) -> list[dict[str, Any]]:
"""Phase 74 variant of :func:`_stream_chat`: sends ``history`` (the
client's prior turns, oldest first) in the request body."""
with client.stream(
"POST", "/api/chat", json={"message": message, "history": history}
) as r:
assert r.status_code == 200
assert r.headers["content-type"].startswith("text/event-stream")
buf = ""
frames: list[dict[str, Any]] = []
for part in r.iter_text():
buf += part
while "\n\n" in buf:
frame, buf = buf.split("\n\n", 1)
frame = frame.strip()
if frame.startswith("data:"):
frames.append(json.loads(frame.removeprefix("data:").strip()))
assert buf.strip() == "", "stream must end on a frame boundary"
return frames
def test_deflected_turn_forwards_history_with_prior_thinking(
client,
db,
seeded_kb: FakeRagLLM,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A DEFLECTED turn sends the prior turns — chronological, with the
prior brain turn's thinking as ``reasoning_content`` — between the
LOW system prompt and the current question (A2/A3/A4); the per-turn
log line carries ``history_msgs=4``."""
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
caplog.set_level(logging.INFO, logger="app.chat")
frames = _stream_chat_with_history(client, OFF_TOPIC, HISTORY)
finally:
fastapi_app.dependency_overrides.clear()
assert frames[-1]["type"] == "done"
assert frames[-1]["deflected"] is True
assert len(seeded_kb.seen_messages) == 1
(messages,) = seeded_kb.seen_messages
assert messages[0]["role"] == "system"
assert "DEFLECT_MODE" in messages[0]["content"] # the LOW prompt
assert messages[1:-1] == HISTORY_MESSAGES # the prior turns, chronological
assert messages[-1] == {"role": "user", "content": OFF_TOPIC}
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert lines and "history_msgs=4" in lines[-1]
def test_grounded_turn_forwards_history_through_the_agent(
client,
db,
seeded_kb: FakeRagLLM,
caplog: pytest.LogCaptureFixture,
) -> None:
"""The GROUNDED agent branch receives the same block: its first
request is ``[HIGH system, *history, current question]`` (the tool
rounds then append to that same list); ``history_msgs=4``."""
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
caplog.set_level(logging.INFO, logger="app.chat")
frames = _stream_chat_with_history(client, QUESTION, HISTORY)
finally:
fastapi_app.dependency_overrides.clear()
assert frames[-1]["type"] == "done"
assert frames[-1]["deflected"] is False
assert len(seeded_kb.seen_messages) == 1 # the canned answer ends the loop
(messages,) = seeded_kb.seen_messages
assert messages[0]["role"] == "system"
assert "<tools>" in messages[0]["content"] # the HIGH prompt
assert messages[1:-1] == HISTORY_MESSAGES
assert messages[-1] == {"role": "user", "content": QUESTION}
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert lines and "history_msgs=4" in lines[-1]
def test_endpoint_two_turn_history_reaches_the_llm(
client,
db,
seeded_kb: FakeRagLLM,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Phase 108 layer 2 (TODO L4 — the owner's follow-up, server wire):
the owner's exact 2-turn history (Q1 "What is my name?" / R1 "Your
name is Reese.") plus the follow-up "What did I just ask you?" must
reach the LLM as ``[system, user Q1, assistant R1, user Q2]`` — the
two prior turns NOT dropped (the reported missing-first-turn symptom
would be their absence from this captured request). The pin is
branch-agnostic BY DESIGN: phase 74 pinned the history splice on
BOTH branches, so whichever the seeded KB retrieves to (LOW/deflected
or HIGH/grounded), the captured request must carry the full prior
exchange. The seeded-KB fixture is used because it is the file's
standard idiom (and keeps the cosine gate exercised like production);
the canned fake answer ends the turn on exactly one LLM request on
either branch."""
history = [
{"who": "user", "text": "What is my name?"},
{"who": "brain", "text": "Your name is Reese."},
]
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
caplog.set_level(logging.INFO, logger="app.chat")
frames = _stream_chat_with_history(client, "What did I just ask you?", history)
finally:
fastapi_app.dependency_overrides.clear()
assert frames[-1]["type"] == "done" # the turn completes (either branch)
assert len(seeded_kb.seen_messages) == 1 # the canned answer ends the turn
(messages,) = seeded_kb.seen_messages
assert messages[0]["role"] == "system"
assert messages[1:-1] == [
{"role": "user", "content": "What is my name?"},
{"role": "assistant", "content": "Your name is Reese."},
] # the FULL prior exchange — chronological, nothing dropped
assert messages[-1] == {"role": "user", "content": "What did I just ask you?"}
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert lines and "history_msgs=2" in lines[-1]
def test_request_without_history_sends_exactly_system_and_user(
client,
db,
seeded_kb: FakeRagLLM,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Byte-identical pin (A2): a request WITHOUT ``history`` sends
exactly the two-message ``[system, user]`` request on BOTH branches
(deflected + grounded), and the per-turn log line carries
``history_msgs=0``."""
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
caplog.set_level(logging.INFO, logger="app.chat")
_stream_chat(client, OFF_TOPIC) # deflected branch
_stream_chat(client, QUESTION) # grounded branch
finally:
fastapi_app.dependency_overrides.clear()
assert len(seeded_kb.seen_messages) == 2
for messages in seeded_kb.seen_messages:
assert [m["role"] for m in messages] == ["system", "user"]
assert seeded_kb.seen_messages[0][1] == {"role": "user", "content": OFF_TOPIC}
assert seeded_kb.seen_messages[1][1] == {"role": "user", "content": QUESTION}
lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
assert len(lines) == 2
assert all("history_msgs=0" in line for line in lines)
def test_history_rejects_unknown_who(client, db) -> None:
"""Schema pin: ``who`` is a ``Literal["user", "brain"]`` — anything
else is a 422 at the boundary (the same trust model as the saved-
chat ``ChatMessage``)."""
r = client.post(
"/api/chat",
json={"message": "hi", "history": [{"who": "alien", "text": "x"}]},
)
assert r.status_code == 422
def test_history_rejects_more_than_100_entries(client, db) -> None:
"""Schema pin: the DoS sanity ceiling is 100 turns — 101 is a 422
(the config budgets do the real trimming; this only keeps a
pathological body from wasting the mapper's work)."""
r = client.post(
"/api/chat",
json={
"message": "hi",
"history": [{"who": "user", "text": f"q{i}"} for i in range(101)],
},
)
assert r.status_code == 422
def test_done_event_serializes_column_maximum_source_refs() -> None:
"""Phase 83 A3 pin: ``SourceRef`` is SHARED by the SSE ``done``
event and the saved-chat surface — the boundary caps added there
(``source`` ≤ 120, ``path`` ≤ 1000, ``title`` ≤ 500) mirror the
``documents`` column lengths EXACTLY, so a server-built event from
a full-length row (values at the column maxima) still constructs
and serializes byte-identical: the SSE contract is provably
unaffected. The one-over caps raise — only client-saved refs can
ever trip a cap, never a server-built ref."""
event = ChatDoneEvent(
deflected=False,
sources=[SourceRef(source="s" * 120, path="p" * 1000, title="t" * 500)],
suggestions=[],
)
assert event.model_dump() == {
"type": "done",
"deflected": False,
"sources": [{"source": "s" * 120, "path": "p" * 1000, "title": "t" * 500}],
# Phase 113: the additive related tier defaults to [] (the
# key is always present on new frames; old clients ignore it).
"related": [],
"suggestions": [],
}
# The caps sit exactly ON the column maxima: one over any of them
# is rejected (a row could never hold such a value in the first
# place — the string columns enforce the same lengths).
with pytest.raises(ValidationError):
SourceRef(source="s" * 121, path="p" * 1000, title="t" * 500)
with pytest.raises(ValidationError):
SourceRef(source="s" * 120, path="p" * 1001, title="t" * 500)
with pytest.raises(ValidationError):
SourceRef(source="s" * 120, path="p" * 1000, title="t" * 501)
def test_done_event_related_defaults_empty_and_old_payload_parses() -> None:
"""Phase 113 back-compat pin: ``related`` defaults to ``[]`` — a
pre-phase-113 done frame (no ``related`` key) still parses, and a
frame with the field round-trips it (PLAN §4: old clients ignore
unknown fields, so the field is additive in both directions)."""
old_payload = {
"type": "done",
"deflected": True,
"sources": [],
"suggestions": ["Maybe try X?"],
}
event = ChatDoneEvent(**old_payload)
assert event.related == []
assert event.model_dump() == {**old_payload, "related": []}
new_payload = {
"deflected": False,
"sources": [SourceRef(source="docs", path="a.md", title="A")],
"related": [SourceRef(source="docs", path="b.md", title="B")],
"suggestions": [],
}
dumped = ChatDoneEvent(**new_payload).model_dump()
assert [r["path"] for r in dumped["related"]] == ["b.md"]
assert [r["path"] for r in dumped["sources"]] == ["a.md"]
# ---------------------------------------------------------------------------
# Phase 122, task 05 — the SSE source frame's OPTIONAL image_url (the
# shared ``source_ref_with_image`` builder: present on an image doc's
# ref only, omitted — never null — on every text ref). Task 06
# finalizes the phase-122 suite here with the story-level pins.
# ---------------------------------------------------------------------------
def _seed_image_doc(db) -> Document:
"""An ``is_image`` documents row the agent can ``read`` (no chunks
needed — the read tool resolves the row by ``(source, path)`` and
serves its ``content``; the frame needs ``is_image`` + ``id``
only). The fixture's teardown truncates the tables."""
doc = Document(
id=uuid.uuid4(),
source="docs",
path="pic.png",
full_path="/tmp/pic.png",
title="pic",
content="A red square on a white background.",
summary="A red square on a white background.",
content_hash="0" * 64,
created_at=_FIXTURE_CREATED_AT,
is_image=True,
image_path="/tmp/pic.png",
)
db.add(doc)
db.commit()
return doc
def test_grounded_turn_reading_image_doc_frame_carries_image_url_only_for_it(
client, db, seeded_kb: FakeRagLLM
) -> None:
"""Task 05 wire contract: a mocked grounded answer whose agent
READS an image doc → the done frame's ref for THAT doc alone
carries ``image_url`` (the bytes route the frontend's sources
block renders from); the text-doc related refs carry NO
``image_url`` key at all (the omission rule — the key is absent,
never null)."""
doc = _seed_image_doc(db)
scripted = FakeRagLLM(
tool_script=[
[
ToolCallPiece(
id="call_1",
name="read",
arguments={"path": "docs/pic.png"},
)
]
]
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
_, _, frames = _stream_chat(client, QUESTION)
finally:
fastapi_app.dependency_overrides.clear()
done = frames[-1]
assert done["deflected"] is False
# The read image doc is the citation surface (phase 119, A1) — and
# its ref is the frame's ONLY image_url carrier.
sources = done["sources"]
assert [(s["source"], s["path"]) for s in sources] == [("docs", "pic.png")]
assert sources[0]["image_url"] == f"/api/documents/{doc.id}/image"
# The related tier (ranks 6–7 for the Kubernetes question) is text
# docs — the key is ABSENT on every one of them, not null.
related = done["related"]
assert related
for ref in related:
assert "image_url" not in ref
def test_text_only_grounded_turn_frame_has_no_image_url_key_anywhere(
client, db, seeded_kb: FakeRagLLM
) -> None:
"""The omission rule at the BYTE level (the phase's byte-identity
criterion): a grounded turn whose cited + related docs are ALL
text docs serializes a done frame with no ``image_url`` key
anywhere — checked on the raw wire text (not a re-serialized
dict), and every ref keeps exactly the pre-phase-122 key set."""
scripted = FakeRagLLM(
tool_script=[
[
ToolCallPiece(
id="call_1",
name="read",
arguments={"path": "docs/homelab/kubernetes.md"},
)
]
]
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
try:
with client.stream("POST", "/api/chat", json={"message": QUESTION}) as r:
assert r.status_code == 200
buf = ""
raw_frames: list[str] = []
for part in r.iter_text():
buf += part
while "\n\n" in buf:
frame, buf = buf.split("\n\n", 1)
frame = frame.strip()
if frame.startswith("data:"):
raw_frames.append(frame)
finally:
fastapi_app.dependency_overrides.clear()
done_raw = [f for f in raw_frames if '"type": "done"' in f]
assert len(done_raw) == 1
# The BYTE check: the key is absent from the wire text itself.
assert "image_url" not in done_raw[0]
done = json.loads(done_raw[0].removeprefix("data:").strip())
assert done["deflected"] is False
assert [(s["source"], s["path"]) for s in done["sources"]] == [
("docs", "homelab/kubernetes.md")
]
for ref in done["sources"] + done["related"]:
assert set(ref) == {"source", "path", "title"} # the pre-122 key set
# ---------------------------------------------------------------------------
# Phase 123 (task 01): image questions — the upload → chat flow delivers
# the multimodal user message to the model (both construction sites), the
# toggle-off / stale frames settle before any model call (no record),
# text-only turns stay byte-identical, and the saved/shared chat round-
# trips the stored PATH (never base64 — LOCKED A5).
# ---------------------------------------------------------------------------
#: A real 1×1 transparent PNG (the phase-122 fixture bytes) — the server
#: is content-agnostic (the extension + size gates only), but a well-
#: formed fixture keeps the data-URL pin honest.
PNG_1X1 = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAAC0lEQVR4nGP4DwQACfsD/fteaysAAAAASUVORK5CYII="
)
@pytest.fixture()
def chat_image_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Point the question-image store (the upload/serve pair) AND the
turn pipeline at a tmp dir, with the ``images`` toggle ON — the
env-calibrated settings (the conftest mock thresholds) plus the
phase-123 pair (``images`` + ``chat_image_dir``); the monkeypatch
fixture reverts both patches."""
store = tmp_path / "chat-images"
store.mkdir()
settings = Settings(
_env_file=None, # pyright: ignore[reportCallIssue]
images=True,
chat_image_dir=str(store),
)
monkeypatch.setattr(chat_images, "get_settings", lambda: settings)
monkeypatch.setattr(chat_api, "get_settings", lambda: settings)
return store
def _upload_image(client: TestClient, name: str = "screenshot.png") -> str:
"""One question-image upload (the composer's step before send) —
returns the served path (the value the chat request accepts)."""
r = client.post(
"/api/chat-images",
files={"file": (name, io.BytesIO(PNG_1X1), "image/png")},
)
assert r.status_code == 200, r.text
return r.json()["path"]
def _stream_chat_with(
client: TestClient, body: dict[str, Any]
) -> tuple[int, str, list[dict[str, Any]]]:
"""``_stream_chat`` with an arbitrary body (the ``image`` key)."""
with client.stream("POST", "/api/chat", json=body) as r:
assert r.status_code == 200
assert r.headers["content-type"].startswith("text/event-stream")
buf = ""
frames: list[dict[str, Any]] = []
for part in r.iter_text():
buf += part
while "\n\n" in buf:
frame, buf = buf.split("\n\n", 1)
frame = frame.strip()
if frame.startswith("data:"):
frames.append(json.loads(frame.removeprefix("data:").strip()))
assert buf.strip() == "", "stream must end on a frame boundary"
return r.status_code, r.headers["content-type"], frames
def _expected_image_content(question: str) -> list[dict[str, Any]]:
"""The multimodal user content list the model must receive for a
question that carried the fixture image (the text part + the
image_url part — a data URL built server-side from the stored bytes
+ the phase-122 mime map)."""
return [
{"type": "text", "text": question},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{base64.b64encode(PNG_1X1).decode('ascii')}"
},
},
]
def test_image_turn_delivers_the_multimodal_user_message(
client, db, seeded_kb: FakeRagLLM, chat_image_store: Path
) -> None:
"""The full flow (LOCKED A5): the client uploads FIRST, then the
turn request carries the returned path — and the GROUNDED branch
(construction site 2: ``run_agent`` builds its own ``[system,
*history, user]`` from the value chat.py passes — the pinned flow)
delivers the multimodal content list to the model: the text part +
the image_url data URL that decodes to the stored bytes."""
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
path = _upload_image(client)
_, _, frames = _stream_chat_with(
client, {"message": QUESTION, "image": path}
)
finally:
fastapi_app.dependency_overrides.clear()
assert frames[-1]["type"] == "done"
assert frames[-1]["deflected"] is False
assert len(seeded_kb.seen_messages) == 1
assert seeded_kb.seen_messages[0][-1] == {
"role": "user",
"content": _expected_image_content(QUESTION),
}
# the data URL really is the stored bytes (decode + compare)
url = seeded_kb.seen_messages[0][-1]["content"][1]["image_url"]["url"]
assert base64.b64decode(url.split(",", 1)[1]) == PNG_1X1
def test_deflected_image_turn_delivers_the_multimodal_user_message(
client, db, seeded_kb: FakeRagLLM, chat_image_store: Path
) -> None:
"""Construction site 1 (the deflected branch consumes chat.py's
``messages`` list directly): an off-topic question + image still
delivers the SAME multimodal list — the LOW prompt path never
drops the attachment."""
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
path = _upload_image(client)
_, _, frames = _stream_chat_with(
client, {"message": OFF_TOPIC, "image": path}
)
finally:
fastapi_app.dependency_overrides.clear()
assert frames[-1]["type"] == "done"
assert frames[-1]["deflected"] is True
assert len(seeded_kb.seen_messages) == 1
assert seeded_kb.seen_messages[0][-1] == {
"role": "user",
"content": _expected_image_content(OFF_TOPIC),
}
def test_image_turn_with_toggle_off_yields_the_hinted_frame_and_calls_nothing(
client, db, seeded_kb: FakeRagLLM, chat_image_store: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The ``BOR_IMAGES`` gate: off with an image set → the phase-114
error frame with the EXACT detail + hint (ONE terminal frame — no
``done``), and NO model call (zero embeds, zero requests) and NO
record (the rejected turn saves nothing — the existing error-path
convention)."""
monkeypatch.setattr(
chat_api,
"get_settings",
lambda: Settings(
_env_file=None, # pyright: ignore[reportCallIssue]
images=False,
chat_image_dir=str(chat_image_store),
),
)
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
path = _upload_image(client)
_, _, frames = _stream_chat_with(
client, {"message": "What is in this image?", "image": path}
)
finally:
fastapi_app.dependency_overrides.clear()
assert [f["type"] for f in frames] == ["error"]
assert frames[0] == {
"type": "error",
"detail": "Image support is turned off on this server.",
"hint": "Enable BOR_IMAGES in the server's .env (and restart) to ask with an image.",
}
# NO model call: the QUESTION embed never ran (``question_embeds``
# counts ``embed_one`` calls — the import's batch embeds are a
# different counter) and the answer endpoint was never asked.
assert seeded_kb.question_embeds == []
assert seeded_kb.seen_messages == []
assert db.scalars(select(QueryLog)).all() == [] # NO query_log row
def test_image_turn_with_a_missing_stored_file_yields_the_stale_frame(
client, db, seeded_kb: FakeRagLLM, chat_image_store: Path
) -> None:
"""The stale-path edge (the file was deleted out-of-band): the SAME
frame shape with the "no longer available" detail and NO hint (the
banner's default copy is the honest fallback) — again before any
model call, no record."""
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
stale = "/api/chat-images/" + "e" * 32 + ".png" # well-formed, absent
_, _, frames = _stream_chat_with(
client, {"message": "What is in this image?", "image": stale}
)
finally:
fastapi_app.dependency_overrides.clear()
assert [f["type"] for f in frames] == ["error"]
assert frames[0] == {
"type": "error",
"detail": "That image is no longer available.",
"hint": None,
}
assert seeded_kb.question_embeds == [] # NO model call (see the toggle-off test)
assert seeded_kb.seen_messages == []
assert db.scalars(select(QueryLog)).all() == []
def test_text_only_turn_is_byte_identical_with_images_enabled(
client, db, seeded_kb: FakeRagLLM, chat_image_store: Path
) -> None:
"""``image=None`` with the toggle ON: the user message is the PLAIN
STRING (not a content list) — the multimodal branch is inert, and
the turn behaves exactly as pre-phase (done frame, the question
embedded whole)."""
fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
try:
_, _, frames = _stream_chat_with(client, {"message": QUESTION})
finally:
fastapi_app.dependency_overrides.clear()
assert frames[-1]["type"] == "done"
assert seeded_kb.seen_messages[0][-1] == {"role": "user", "content": QUESTION}
assert isinstance(seeded_kb.seen_messages[0][-1]["content"], str)
assert seeded_kb.question_embeds == [QUESTION[: 1200]]
def test_saved_chat_round_trips_the_user_image_path(
client, db, chat_image_store: Path
) -> None:
"""Persistence (LOCKED A5): the user record carries the stored
PATH — the stored JSONB and the saved-chat response round-trip it,
the brain record carries NO image key, and NOTHING base64 crosses
the storage boundary. A text-only record stays without the key
(the phase-50 byte-identical contract for text-only chats)."""
db.execute(text("TRUNCATE saved_chats"))
db.commit()
try:
path = _upload_image(client)
r = client.post(
"/api/chats",
json={
"messages": [
{"who": "user", "text": "What is in this image?", "image": path},
{"who": "brain", "text": "A cat, by the look of it."},
]
},
)
assert r.status_code == 201, r.text
body = r.json()
assert body["messages"][0]["image"] == path
assert "image" not in body["messages"][1] # brain records never carry it
# The RAW stored JSONB: the path, never base64 (LOCKED A5)
raw = db.execute(
text("SELECT messages FROM saved_chats WHERE id = :id"),
{"id": body["id"]},
).scalar_one()
assert raw[0]["image"] == path
assert "base64" not in json.dumps(raw)
# The admin GET round-trips it losslessly
got = client.get(f"/api/chats/{body['id']}")
assert got.status_code == 200
assert got.json()["messages"][0]["image"] == path
assert "image" not in got.json()["messages"][1]
finally:
db.execute(text("TRUNCATE saved_chats"))
db.commit()
def test_shared_chat_serves_the_user_image_path(
client, db, chat_image_store: Path
) -> None:
"""The shared view is faithful: the public snapshot carries the
user's image path (the public ``messages`` shape already includes
the optional key), and the image bytes themselves are publicly
servable (the image is part of the chat's content — phase 55 A1)."""
db.execute(text("TRUNCATE saved_chats"))
db.commit()
try:
path = _upload_image(client)
r = client.post(
"/api/chats",
json={
"share": True,
"messages": [
{"who": "user", "text": "What is in this image?", "image": path},
{"who": "brain", "text": "A cat, by the look of it."},
],
},
)
assert r.status_code == 201, r.text
share_url = r.json()["share_url"]
token = share_url.rsplit("/", 1)[-1]
anon = TestClient(fastapi_app)
got = anon.get(f"/api/shared/{token}")
assert got.status_code == 200
messages = got.json()["messages"]
assert messages[0]["image"] == path
assert "image" not in messages[1]
img = anon.get(path)
assert img.status_code == 200
assert img.content == PNG_1X1
finally:
db.execute(text("TRUNCATE saved_chats"))
db.commit()
Reference in New Issue View Git Blame Copy Permalink