"""Unit: schema boundary caps for the saved-chat surface (phase 83, task 01). The first schema-boundary suite: every A1 cap in ``app/schemas.py`` is pinned at BOTH boundaries — a value exactly at the cap validates and one past it raises a pydantic ``ValidationError`` naming the offending field (the 422-shaped boundary response, house style, phase 56 precedent). Plus the regression pin: a realistic ``bor.chat.v1`` payload validates cleanly and round-trips ``model_dump()`` (the stored-shape contract — the caps added value bounds only, no key/shape change). """ from __future__ import annotations import pytest from pydantic import ValidationError from app.schemas import ( ChatMessage, ChatRequest, SavedChatCreate, SavedChatUpdate, SourceRef, ToolCall, ) # --- boundary constants (phase 83, A1 — mirror sources in app/schemas.py) --- TEXT_CAP = 32_000 # mirrors HistoryTurn.text / .thinking SOURCES_CAP = 20 SUGGESTIONS_CAP = 50 CHIP_CAP = 200 TOOLS_CAP = 50 SOURCE_CAP = 120 # documents.source String(120) PATH_CAP = 1000 # documents.path String(1000) TITLE_CAP = 500 # documents.title String(500) NAME_CAP = 100 # ToolCall.name ARGUMENT_CAP = 2000 # ToolCall.argument MESSAGES_CAP = 200 # SavedChatCreate/Update.messages QUESTION_CAP = 4_000 # ChatRequest.message — the cap the composer mirrors (phase 104) def _source_ref() -> dict: return {"source": "Homelab", "path": "kubernetes.md", "title": "K8s"} def _tool_call() -> dict: # Phase 95 (task 02): the truncation record rides the same entry — # the CURRENT full shape (additive fields, defaults for a plain read). return { "name": "read", "argument": "Homelab/kubernetes.md", "truncated": False, "chars_shown": None, "chars_total": None, } def _user_message(text: str = "How did I install k3s?") -> ChatMessage: return ChatMessage.model_validate({"who": "user", "text": text}) def _failed_loc(exc: ValidationError, *loc: object) -> None: """Assert the first ``ValidationError`` names exactly the given field path (``loc`` tuple, e.g. ``("text",)`` or ``("messages", 0, "text")``) — the 422 must point at the offender.""" assert exc.errors()[0]["loc"] == tuple(loc), exc.errors() # --------------------------------------------------------------------------- # ChatMessage.text / .thinking # --------------------------------------------------------------------------- def test_chat_message_text_at_cap_validates() -> None: msg = _user_message("a" * TEXT_CAP) assert len(msg.text) == TEXT_CAP def test_chat_message_text_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: _user_message("a" * (TEXT_CAP + 1)) _failed_loc(exc.value, "text") def test_chat_message_thinking_at_cap_validates() -> None: msg = ChatMessage.model_validate({"who": "brain", "text": "ok", "thinking": "t" * TEXT_CAP}) assert len(msg.thinking or "") == TEXT_CAP def test_chat_message_thinking_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ChatMessage.model_validate( {"who": "brain", "text": "ok", "thinking": "t" * (TEXT_CAP + 1)} ) _failed_loc(exc.value, "thinking") def test_chat_message_thinking_none_still_valid() -> None: assert _user_message().thinking is None assert ChatMessage.model_validate( {"who": "brain", "text": "ok", "thinking": None} ).thinking is None # --------------------------------------------------------------------------- # ChatMessage.sources / .suggestions / .tools # --------------------------------------------------------------------------- def test_chat_message_sources_at_cap_validates() -> None: msg = ChatMessage.model_validate( {"who": "brain", "text": "ok", "sources": [_source_ref() for _ in range(SOURCES_CAP)]} ) assert len(msg.sources or []) == SOURCES_CAP def test_chat_message_sources_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ChatMessage.model_validate( { "who": "brain", "text": "ok", "sources": [_source_ref() for _ in range(SOURCES_CAP + 1)], } ) _failed_loc(exc.value, "sources") def test_chat_message_suggestions_at_cap_validates() -> None: msg = ChatMessage.model_validate( {"who": "brain", "text": "ok", "suggestions": [f"chip {i}" for i in range(SUGGESTIONS_CAP)]} ) assert len(msg.suggestions or []) == SUGGESTIONS_CAP def test_chat_message_suggestions_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ChatMessage.model_validate( { "who": "brain", "text": "ok", "suggestions": [f"chip {i}" for i in range(SUGGESTIONS_CAP + 1)], } ) _failed_loc(exc.value, "suggestions") def test_chat_message_suggestion_chip_at_cap_validates() -> None: msg = ChatMessage.model_validate( {"who": "brain", "text": "ok", "suggestions": ["c" * CHIP_CAP]} ) assert msg.suggestions is not None assert len(msg.suggestions[0]) == CHIP_CAP def test_chat_message_suggestion_chip_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ChatMessage.model_validate( {"who": "brain", "text": "ok", "suggestions": ["c" * (CHIP_CAP + 1)]} ) # the offender is the ITEM (the list itself is well under max_items). _failed_loc(exc.value, "suggestions", 0) def test_chat_message_tools_at_cap_validates() -> None: msg = ChatMessage.model_validate( {"who": "brain", "text": "ok", "tools": [_tool_call() for _ in range(TOOLS_CAP)]} ) assert len(msg.tools or []) == TOOLS_CAP def test_chat_message_tools_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ChatMessage.model_validate( {"who": "brain", "text": "ok", "tools": [_tool_call() for _ in range(TOOLS_CAP + 1)]} ) _failed_loc(exc.value, "tools") # --------------------------------------------------------------------------- # ChatRequest.message (the 4,000-char question cap — phase 104, A3) # --------------------------------------------------------------------------- def test_chat_request_message_at_cap_validates() -> None: """EXACTLY 4,000 chars passes — the cap admits "a code example of a few dozen lines" (owner A3) and the composer's maxlength + counter mirror this exact boundary (the 4,000/4,000 E2E submit must not 422).""" req = ChatRequest.model_validate({"message": "a" * QUESTION_CAP}) assert len(req.message) == QUESTION_CAP def test_chat_request_message_one_over_cap_rejects() -> None: """4,001 chars is the 422 the UI now prevents blind: the server cap stays the backstop (pre-existing, untouched — this phase only pins it at the boundary, phase-83 pattern), naming ``message`` in the error loc.""" with pytest.raises(ValidationError) as exc: ChatRequest.model_validate({"message": "a" * (QUESTION_CAP + 1)}) _failed_loc(exc.value, "message") # --------------------------------------------------------------------------- # SourceRef (documents column-length mirrors) # --------------------------------------------------------------------------- def test_source_ref_source_at_cap_validates() -> None: ref = SourceRef.model_validate( {"source": "s" * SOURCE_CAP, "path": "a.md", "title": "A"} ) assert len(ref.source) == SOURCE_CAP def test_source_ref_source_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: SourceRef.model_validate({"source": "s" * (SOURCE_CAP + 1), "path": "a.md", "title": "A"}) _failed_loc(exc.value, "source") def test_source_ref_path_at_cap_validates() -> None: ref = SourceRef.model_validate( {"source": "Homelab", "path": "p" * PATH_CAP, "title": "A"} ) assert len(ref.path) == PATH_CAP def test_source_ref_path_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: SourceRef.model_validate( {"source": "Homelab", "path": "p" * (PATH_CAP + 1), "title": "A"} ) _failed_loc(exc.value, "path") def test_source_ref_title_at_cap_validates() -> None: ref = SourceRef.model_validate( {"source": "Homelab", "path": "a.md", "title": "t" * TITLE_CAP} ) assert len(ref.title) == TITLE_CAP def test_source_ref_title_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: SourceRef.model_validate( {"source": "Homelab", "path": "a.md", "title": "t" * (TITLE_CAP + 1)} ) _failed_loc(exc.value, "title") # --------------------------------------------------------------------------- # ToolCall # --------------------------------------------------------------------------- def test_tool_call_name_at_cap_validates() -> None: call = ToolCall.model_validate({"name": "n" * NAME_CAP, "argument": None}) assert len(call.name) == NAME_CAP def test_tool_call_name_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ToolCall.model_validate({"name": "n" * (NAME_CAP + 1), "argument": None}) _failed_loc(exc.value, "name") def test_tool_call_argument_at_cap_validates() -> None: call = ToolCall.model_validate({"name": "read", "argument": "a" * ARGUMENT_CAP}) assert len(call.argument or "") == ARGUMENT_CAP def test_tool_call_argument_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ToolCall.model_validate({"name": "read", "argument": "a" * (ARGUMENT_CAP + 1)}) _failed_loc(exc.value, "argument") def test_tool_call_argument_none_still_valid() -> None: assert ToolCall.model_validate({"name": "ls", "argument": None}).argument is None assert ToolCall.model_validate({"name": "ls"}).argument is None # --------------------------------------------------------------------------- # SavedChatCreate / SavedChatUpdate .messages # --------------------------------------------------------------------------- def _minimal_messages(n: int) -> list[dict]: return [{"who": "user" if i % 2 == 0 else "brain", "text": f"m{i}"} for i in range(n)] def test_saved_chat_create_messages_at_cap_validates() -> None: payload = SavedChatCreate.model_validate( {"title": "big", "messages": _minimal_messages(MESSAGES_CAP)} ) assert len(payload.messages) == MESSAGES_CAP def test_saved_chat_create_messages_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: SavedChatCreate.model_validate( {"title": "bigger", "messages": _minimal_messages(MESSAGES_CAP + 1)} ) _failed_loc(exc.value, "messages") def test_saved_chat_update_messages_at_cap_validates() -> None: payload = SavedChatUpdate.model_validate({"messages": _minimal_messages(MESSAGES_CAP)}) assert len(payload.messages) == MESSAGES_CAP def test_saved_chat_update_messages_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: SavedChatUpdate.model_validate({"messages": _minimal_messages(MESSAGES_CAP + 1)}) _failed_loc(exc.value, "messages") # --------------------------------------------------------------------------- # JSON-shape regression: same KEYS as before, value bounds only # --------------------------------------------------------------------------- def test_extra_keys_still_forbidden() -> None: """``extra="forbid"`` is untouched — a stray key is still a 422-shaped rejection (the accepted/rejected KEYS did not change with phase 83).""" with pytest.raises(ValidationError) as exc: ChatMessage.model_validate({"who": "user", "text": "hi", "html": "x"}) _failed_loc(exc.value, "html") def test_minimal_message_still_validates() -> None: """Optional keys may be ABSENT exactly as pre-phase-83.""" msg = ChatMessage.model_validate({"who": "user", "text": "hi"}) assert ( msg.sources, msg.related, msg.deflected, msg.suggestions, msg.thinking, msg.tools, msg.stopped, ) == ( None, None, None, None, None, None, None, ) # --------------------------------------------------------------------------- # Realistic-payload round-trip (the stored-shape contract) # --------------------------------------------------------------------------- def test_realistic_bor_chat_v1_payload_round_trips() -> None: """A full ``bor.chat.v1``-shaped record (4–8 messages mixing user/brain, one brain message with ``thinking`` + ``tools`` + ``sources``, one with ``suggestions`` + ``stopped``) validates cleanly and ``model_dump()`` of the messages equals the input dict — ``None``-keys preserved (the phase-50/51 byte-identical stored shape; the caps added value bounds only, no key/shape change).""" messages: list[dict] = [ { "who": "user", "text": "How did I install k3s on the new node?", "sources": None, "related": None, "deflected": None, "suggestions": None, "thinking": None, "tools": None, "stopped": None, }, { "who": "brain", "text": "Your k3s cluster runs on three nodes — here's how it happened.", "sources": [ {"source": "Homelab", "path": "kubernetes.md", "title": "Kubernetes Cluster"}, {"source": "Deployments", "path": "k3s-install.md", "title": "k3s Install Notes"}, ], # Phase 113 related-doc tier — an ACCEPTED key (the phase-113 # omission of this field made extra="forbid" 422 every # done-time auto-save carrying it, so grounded turns' brain # messages never persisted — the A2 quiet failure swallowed # the 422). Round-trips like sources. "related": [ {"source": "Homelab", "path": "traefik.md", "title": "Traefik Notes"} ], "deflected": False, "suggestions": None, "thinking": "The kubernetes doc covers the cluster layout…", "tools": [ # Phase 95 (task 02): the current full shape — one # truncated read (the marker record) + a plain ls. { "name": "read", "argument": "Homelab/kubernetes.md", "truncated": True, "chars_shown": 128_000, "chars_total": 204_000, }, { "name": "ls", "argument": None, "truncated": False, "chars_shown": None, "chars_total": None, }, ], "stopped": None, }, { "who": "user", "text": "And what ports does Traefik expose?", "sources": None, "related": None, "deflected": None, "suggestions": None, "thinking": None, "tools": None, "stopped": None, }, { "who": "brain", "text": "Traefik exposes 80/443 on every node.", "sources": None, "related": None, "deflected": None, "suggestions": ["What is the Traefik dashboard password?"], "thinking": None, "tools": None, "stopped": True, # the owner stopped the generation mid-answer }, ] payload = SavedChatCreate.model_validate({"title": "k3s install", "messages": messages}) dumped = payload.model_dump() # the stored shape is byte-identical: every message dict — None-keys # and all — round-trips unchanged (and the top-level record keeps its # ``share`` default). assert dumped["messages"] == messages assert dumped["share"] is False def test_realistic_payload_round_trips_through_update_model() -> None: """Same record re-Saved through ``SavedChatUpdate`` (the re-Save upsert surface) — identical stored shape.""" msg = { "who": "brain", "text": "answer", "sources": [_source_ref()], "related": None, "deflected": None, "suggestions": ["follow-up?"], "thinking": "scratchpad", "tools": [_tool_call()], "stopped": None, } payload = SavedChatUpdate.model_validate({"messages": [msg]}) assert payload.model_dump()["messages"] == [msg] # --------------------------------------------------------------------------- # Phase 95 (task 02): ToolCall truncation fields + ChatToolResultEvent # --------------------------------------------------------------------------- def test_tool_call_round_trips_with_truncation_fields() -> None: """The CURRENT full shape (phase 95 task 02): a truncated read's record (``truncated: True`` + the two non-negative counts) validates and round-trips ``model_dump()`` unchanged — the save payload carries it with zero other change (the UI re-renders the marker from it).""" raw = { "name": "read", "argument": "Homelab/big.md", "truncated": True, "chars_shown": 128_000, "chars_total": 204_000, } call = ToolCall.model_validate(raw) assert call.truncated is True assert call.chars_shown == 128_000 assert call.chars_total == 204_000 assert call.model_dump() == raw def test_tool_call_old_shape_validates_with_defaults() -> None: """Backward-compat (the phase-50 rule): a saved chat written BEFORE phase 95 — tool records without the truncation fields — validates UNCHANGED: ``truncated`` defaults to False (the marker is absent), the counts to None. No migration (``ChatMessage.tools`` is JSON); only the dump gains the additive keys with their defaults.""" old_shape = {"name": "read", "argument": "Homelab/kubernetes.md"} call = ToolCall.model_validate(old_shape) assert call.truncated is False assert call.chars_shown is None assert call.chars_total is None dumped = call.model_dump() assert dumped["name"] == "read" and dumped["argument"] == "Homelab/kubernetes.md" assert dumped["truncated"] is False assert dumped["chars_shown"] is None and dumped["chars_total"] is None def test_tool_call_old_shape_message_still_round_trips_as_record() -> None: """The record-level backward-compat: a pre-phase-95 brain message (old-shape ``tools``) validates inside ``ChatMessage`` and dumps back as a VALID record of the same shape (the frontend renders it without the marker — ``truncated`` falsy).""" old_message = { "who": "brain", "text": "You've got this!", "sources": None, "deflected": False, "suggestions": None, "thinking": None, "tools": [ {"name": "read", "argument": "Homelab/kubernetes.md"}, {"name": "read_document", "argument": "Homelab/legacy.md"}, ], "stopped": None, } msg = ChatMessage.model_validate(old_message) assert all(t.truncated is False for t in (msg.tools or [])) # Re-validating the dump is a no-op (lossless record round-trip). ChatMessage.model_validate(msg.model_dump()) def test_tool_call_counts_reject_negative() -> None: """Phase 83 bounds philosophy: the counts are non-negative (``ge=0``) — a negative count is not a real record.""" with pytest.raises(ValidationError): ToolCall.model_validate( { "name": "read", "argument": "x", "truncated": True, "chars_shown": -1, "chars_total": 5, } ) with pytest.raises(ValidationError): ToolCall.model_validate( { "name": "read", "argument": "x", "truncated": True, "chars_shown": 5, "chars_total": -1, } ) def test_chat_tool_result_event_shape() -> None: """The A15 extension's wire shape (phase 95 task 02): the seventh, OPTIONAL SSE event type — ``{type, name, argument, truncated, chars_shown, chars_total}`` — with the pinned field order, the ``tool_result`` default, ``truncated`` defaulting True (the emission trigger), and the non-negative counts.""" from app.schemas import ChatToolResultEvent ev = ChatToolResultEvent( name="read", argument="docs/big.md", truncated=True, chars_shown=128_000, chars_total=204_000, ) dumped = ev.model_dump() assert list(dumped) == [ "type", "name", "argument", "truncated", "chars_shown", "chars_total", ] assert dumped == { "type": "tool_result", "name": "read", "argument": "docs/big.md", "truncated": True, "chars_shown": 128_000, "chars_total": 204_000, } # The emission trigger defaults: the pump always builds the frame from # a piece, so a frame that ever exists carries the truncation truth. minimal = ChatToolResultEvent( name="read", argument=None, chars_shown=0, chars_total=0 ).model_dump() assert minimal["truncated"] is True def test_chat_tool_result_event_counts_reject_negative() -> None: """The frame's counts are non-negative (``ge=0``), like the record's.""" from app.schemas import ChatToolResultEvent with pytest.raises(ValidationError): ChatToolResultEvent(name="read", argument="x", chars_shown=-1, chars_total=5) with pytest.raises(ValidationError): ChatToolResultEvent(name="read", argument="x", chars_shown=5, chars_total=-1)