Files
brain-of-reese/app/rag/agent.py
T
ducoterra 055c0b5d85
Build and Push Containers / build-and-push-app (push) Successful in 1m39s
Build and Push Containers / build-and-push-db (push) Successful in 11s
feat(rag): pass chat history with prior thinking to the LLM
Phase 74 (TODO.md L4): a follow-up question now reaches the model WITH
the conversation so far — every prior user/brain turn and the prior
thinking blocks on brain turns (preserve-thinking) — while
POST /api/chat stays stateless (A10): the client provides the history
in the request body and the server stores nothing new.

Server (task 01):
- ChatRequest.history: optional list[HistoryTurn] (who: user|brain,
  text, optional thinking) — absent/empty keeps the request
  byte-identical to pre-phase-74 (the two-message [system, user]
  request; the kill-switch semantics are pinned in the integration
  suite).
- app.rag.prompts.history_to_messages: pure mapper — walks the turns
  newest-first against the settings budgets (history_max_turns=40 /
  history_max_chars=24000, BOR_HISTORY_MAX_TURNS /
  BOR_HISTORY_MAX_CHARS); a capped turn is dropped WHOLE (never cut
  mid-answer); the kept window is returned oldest-first; brain turns
  carry their thinking as reasoning_content (A4) only when
  non-empty.
- Both branches feed it: the deflected path splices it between the
  system prompt and the current user message (the phase-71 recovery
  still rebuilds from messages[1:]), the grounded agent receives
  run_agent(..., history=hist); llm.py's message params widen to
  list[dict[str, Any]] (string-only messages stay byte-identical on
  the wire — the SDK passes message dicts through verbatim).
- The per-turn log line (PLAN §9) gains history_msgs=N after
  kb_chars=N.
- Pins: tests/unit/test_history.py (mapper: mapping, reasoning
  gating, both budgets, drop-whole, ordering, empty default),
  tests/unit/test_config.py (the two settings + env overrides),
  tests/unit/test_agent.py (the history splice + the default),
  tests/integration/test_chat_api.py (deflected AND grounded forward
  the history incl. reasoning_content, no-history byte-identity, 422
  pins, the log field).

Client (task 02):
- runTurn — the single funnel for fresh send / phase-49 retry /
  phase-53 stale-regen — sends history = the conversation record
  minus the current question, with thinking only on brain records
  that streamed one (undefined drops the key from the JSON, the
  record's convention); the question is never duplicated into the
  history.

Wire proof (task 03):
- The mock's echo my history marker (HISTORY_TRIGGER) answers with
  the deterministic history echo — history: N prior messages; last
  answer tail: <last 24 chars>; thinking: yes|no — checked BEFORE
  the DEFLECT_MODE branch (like TABLE_TRIGGER), so it fires on both
  turn branches whatever the gate says; the module docstring records
  the user/assistant-only history invariant that keeps every
  existing (tool-result-classified) marker flow unaffected.
- tests/e2e/test_llm_history.py (isolated): a grounded follow-up and
  a deflected follow-up both receive history: 2 prior messages +
  thinking: yes + the byte-exact tail of turn 1's answer (derived
  from the persisted bor.chat.v1 record — the same array the client
  maps into the body); a cold start receives history: 0 prior
  messages / last answer tail: none / thinking: no.
- Regressions green in isolation: chat_rag, chat_history (phase 50),
  agent_document_tools, harness_aligned_tools, stop_generation,
  retry_answer, response_to_docs.
2026-09-05 16:04:40 -04:00

1084 lines
52 KiB
Python

"""Agent loop: the grounded-turn document tools (phase 37, task 03; the
harness-aligned ``ls``/``read``/``grep`` surface, phase 70).
Probe verdict (task 01 — ``uv run python -m scripts.llm_probe --tools``
run live against aipi): **``probe: turbo tool_calls=supported 2026-08-26``**
— ``turbo`` answers OpenAI ``tools`` requests with
``finish_reason="tool_calls"`` and streams the calls as indexed
``delta.tool_calls`` partials (id + name on the first partial, arguments
in fragments). This module therefore uses the **native tool-calling
path**: tool calls arrive as :class:`app.rag.llm.ToolCallPiece` values
from ``chat_stream(messages, tools=AGENT_TOOLS)``. The prompt-based
JSON-block fallback (documented in the task file) is *not* implemented —
it exists only for a "not supported"/"intermittent" verdict, and the
probe came back "supported".
Real-model gate (phase 72, task 05 — live vs the configured chat
model; re-run 2026-09-04 on the controlled fixture KB — see
``TOOL_CALLING_TESTING.md``): the bare-path teaching (did-you-mean
refusals) makes every trap self-correct in exactly one round — zero
cap hits, zero repeat loops. Locked derived battery (phase 72,
executed ≥ 90 % bar): ``gate: lite FAIL turns=10 answered=10 caps=0
tool-turns=10 calls 5/15 executed (33%) contract 12/15 (80%) 2026-09-04
(wall 47.7s)`` — the bar is blocked by :data:`ALREADY_IN_CONTEXT` dedupe
refusals on the corrected re-reads, a copy-invariant model behavior
(five copy variants, 2026-09-03 → 04) and an app-semantics decision
(TOOL_CALLING_TESTING.md §7). Controlled fixture battery (the 2026-09-04
methodology — contract accuracy ≥ 90 %): PASS on three consecutive runs,
``contract 11/11 (100%)`` / ``12/13 (92%)`` / ``11/11 (100%)``.
Loop contract (one grounded chat turn; the API layer wires this in,
task 04):
1. The model is offered the three OpenAI functions in :data:`AGENT_TOOLS`
for the whole turn — phase 45 removed the phase-37 per-tool budgets
(owner permission 2026-08-27, ``TODO.md`` L8: "allow the LLM to make
as many tool calls as it wants"): ``ls``, ``read`` and ``grep`` can
each be called as many times as the model needs, re-lists and
re-greps included. With ``settings.agent_max_rounds``
(``BOR_AGENT_MAX_ROUNDS``, default 10) at 0 the loop makes exactly
one request with ``tools=None`` — byte-identical to the
pre-phase-37 chat path (the kill switch). Phase 70 (owner permission
2026-09-03: "match existing harnesses as much as possible") renamed
and reshaped the tools to the harness-trained surface —
``ls(path?)`` / ``read(path)`` / ``grep(pattern, path?)``, the
pi.dev tool shapes the model was trained on: the combined
``source/path`` string is the canonical document identity in every
tool argument, refusal, and result header, and the old two-argument
split (with its self-correction and "teach the split" refusals) is
gone — the model's combined form is now simply correct. The
phase-68 A5 match/output contract rides along under the new name.
2. Each tool call the model emits is executed server-side against
Postgres only (no LLM, no network): ``ls`` returns the indexed
catalog — one ``source: X | path: Y | title: Z`` line per document
(phase 63: labeled fields — unambiguous for LLM parsing),
``GET /api/docs`` order (uncapped in v1; the UI never shows it, only
the model does) — optionally scoped to one source name (a ``path``
argument matching no source name is a refusal; a registered source
with no indexed documents lists as ``0 documents:`` and counts) —
``read`` takes the combined ``source/path`` string, splits it at the
FIRST ``'/'`` (source names are directory basenames — they can never
contain ``'/'``), and returns the document's **full** content
(A7-revised contract: never truncated) — and ``grep`` greps the
indexed documents (or the one document a combined ``source/path``
names) for a case-insensitive fixed substring and returns up to 20
``source/path:line: text`` match lines (owner-locked A5, phase 68),
each line truncated to 200 chars. A grep is a **locator**, not a
context-adder: it never appends to the answer context (only
``read`` does — ``holder.read_docs`` is untouched by a grep).
The 2026-09-05 incident teaching (the harness prior is that grep
takes a REGEX; this grep is a fixed substring and the contract does
not change): when a grep RAN but found nothing and the pattern is
regex-shaped (:func:`looks_like_regex`) with a non-empty
:func:`plain_form`, the no-match result is the TEACHING line —
:data:`NO_MATCHES_REGEX` / :data:`NO_MATCHES_REGEX_SCOPED` — which
states the plain-substring contract and hands over the plain-form
retry hint; a grep that matched, or a no-match for a plain pattern
(or one that reduces to nothing), is byte-identical to the ordinary
line. Still a (counted) result, never a refusal.
3. Rejected calls get a one-line refusal and count in nothing
(``holder.tool_calls`` tracks executed calls only): unknown tool name
→ ``"Unknown tool."``; a ``read`` without a usable ``path`` (missing,
blank or non-string) → ``"read requires a string argument
'path'."``; a ``grep`` without a usable ``pattern`` (missing, blank
or non-string) → ``"grep requires a string argument
'pattern'."``; a scoped ``ls`` whose (stripped) ``path`` contains a
``/`` — a document path where a source name belongs (source names
are directory basenames and can never contain one; the 2026-09-03
incident's ``ls(path='app/rag/importer.py')``) →
:data:`LS_PATH_NOT_A_SOURCE`, the document-path teaching line with
the argument echoed; a scoped ``ls`` whose ``path`` names no
registered source (no ``/`` — the incident's ``ls(path='.')``)
→ :data:`NO_SOURCE_NOT_A_DIRECTORY`, the no-source refusal with the
teaching parenthetical appended; a document
already in context (seed or previously read) →
:data:`ALREADY_IN_CONTEXT` (phase 72, task 05 gate iteration:
the line names the correct action — answer from the text already
in the prompt, do not call read again — so a fired refusal ends
the loop instead of inviting a repeat); an unknown document (a ``read`` or scoped ``grep`` whose
combined ``source/path`` matches nothing — a bare source name, which
can never be a document, included) → ``"No document at '…' — check
the ls output."`` with the argument echoed as passed (the model sees
its own form) — EXCEPT the phase-72 "did you mean …?" teaching
(task 02): when the argument is a path (contains ``/``) that matches
an indexed document's ``path`` (exact or as a ``/arg`` suffix,
case-sensitive, catalog order — :func:`find_path_candidates`, a pure
catalog lookup, one bulk query, called only from this refusal path),
the refusal names the combined identity instead — exactly one match
→ :data:`NO_DOCUMENT_DID_YOU_MEAN` (``did you mean
'source/path'?``), two or more → :data:`NO_DOCUMENT_DID_YOU_MEAN_MANY`
(up to :data:`SUGGESTION_LIMIT` identities), so the harness-prior
misuse (the bare document path missing the source prefix,
``read('app/rag/importer.py')``) self-corrects in one round; it is
still a refusal (counts in nothing, consumes a round — no silent
argument normalization), and a bare argument (no ``/``) or a
zero-candidate path keeps the line above byte-identical (the bare
form never hits the DB). A grep that ran but found nothing is NOT a rejection
— its ``"No matches for …"`` line is a (counted) result. A rejected
call still consumes a *round* in the loop, so a pathological stream
that keeps emitting rejected calls is bounded by the cap (point 4).
4. Every call the model emits is appended back to the message history as
the assistant tool-call message + the tool result (refusals included),
consumes one round, and the model is called again. At the round cap —
``max_rounds = settings.agent_max_rounds`` (``BOR_AGENT_MAX_ROUNDS``,
default 10) — the loop forces one final retried no-tools request
(``chat_stream_retried`` with ``tools=None``) and returns: the cap is
the **only** forced exit (besides "the stream carried no calls"), and
it bounds pathological rejected-call streams.
5. A rare stream that carries both content and a tool call keeps the
content (it was already emitted) **and** still runs the tool.
6. *holder* (an :class:`AgentHolder`) records the read documents and the
number of executed tool calls (re-lists included); the API layer
(task 04) reads it after the stream to extend ``done.sources`` /
``query_log.sources`` and the per-turn log line (``tool_calls=N``).
7. Retries (phase 67, owner-locked A2): every model request — each tool
round and the forced final ``tools=None`` call — goes through
``chat_stream_retried``: a round that dies before its first piece is
restarted with the SAME messages (up to ``settings.llm_retries``
restarts, a flat ``settings.llm_retry_delay`` between attempts, each
preceded by a :class:`app.rag.llm.RetryPiece` the API layer turns into
an SSE ``retry`` frame); a round that already streamed a piece fails
the turn as before (no partial answer is ever redone). Retries are
invisible to the round cap: a round that needed a retry still consumes
exactly one round. With ``settings.llm_retries=0`` every request is a
single plain attempt (the pre-phase-67 path).
Scaffolding guardrail (phase 71, deterministic only — owner permission
2026-09-03: "deterministic guardrails only right now, forget using a
model for that"): every model request (each round, the forced final,
and any recovery) runs its ``delta.content`` through a fresh caller-
owned :class:`app.rag.scaffolding.ScaffoldingFilter`, so raw
``<|tool_call_start|>…<|tool_call_end|>`` tokens can never reach the
user as answer text. A round that ends with NO visible content AND a
non-empty strip (the scaffolding was the whole "answer") gets exactly
ONE bounded recovery: one extra request with ``tools=None``, the same
messages with :data:`CORRECTION_INSTRUCTION` folded into the original
single system message, a fresh filter, and the same phase-67 retry
budget. A recovery that also comes back empty — or a round with no
strip and no content (today's empty/thinking-only answer) — settles as
before; a second empty reply raises :class:`MalformedReplyError` (the
API layer turns it into the dedicated error frame). A round with real
visible content plus scaffolding needs no recovery (the clean content
stands), and a scaffolding-only round that also carried tool calls
needs none either (the tool ran) — the policy keys on the no-calls
exit only. No model participates in detection or repair: the
registry + the fixed retry policy are the whole guardrail.
The DB accessors (:func:`list_catalog`, :func:`list_source_names`,
:func:`find_document`, :func:`all_documents`) and the
:func:`grep_document` line matcher are module-level functions so unit
tests can monkeypatch them without a database.
"""
from __future__ import annotations
import json
import logging
import re
from collections.abc import AsyncIterator, Sequence
from dataclasses import dataclass, field
from typing import Any
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.config import Settings
from app.models import Document
from app.rag.git_sources import effective_sources
from app.rag.llm import (
LLMClient,
LLMError,
RetryPiece,
StreamPiece,
ToolCallPiece,
chat_stream_retried,
)
from app.rag.scaffolding import ScaffoldingFilter
from app.rag.source_removal import resolve_source_name
logger = logging.getLogger("app.agent")
#: The three agent tools (phase 70: the harness-aligned surface —
#: ``ls`` / ``read`` / ``grep``, the pi.dev tool shapes the model was
#: trained on, replacing the phase-37 list/read and phase-68 search
#: names): OpenAI function
#: definitions passed as ``tools=AGENT_TOOLS`` to ``chat_stream`` for
#: the whole grounded turn — phase 45 removed the per-tool budgets; the
#: round cap (``BOR_AGENT_MAX_ROUNDS``) is the only bound. The combined
#: ``source/path`` string is the canonical document identity in every
#: argument (phase 70, owner permission 2026-09-03).
AGENT_TOOLS: list[dict[str, Any]] = [
{
"type": "function",
"function": {
"name": "ls",
"description": (
"List the indexed documents as `source: X | path: Y | "
"title: Z` lines. Call one tool at a time — wait for "
"this result before your next call."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": (
"Source name to list one source's documents "
"(e.g. 'homelab') — a source name, not a "
"file or directory path; omit to list "
"every document. This is the only tool "
"whose `path` is a source name — for "
"`read` and `grep` it must be a document's "
"combined `source/path`."
),
}
},
"required": [],
},
},
},
{
"type": "function",
"function": {
"name": "read",
"description": (
"Do not call this tool for a document already shown in "
"the <documents> section, even when the user asks you to "
"open or read it — its full text is already in your "
"prompt; answer directly from it. Use it only to add a "
"document NOT already in <documents> to your context, "
"by its combined `source/path` string. Call one tool at "
"a time — wait for this result before your next call."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": (
"The document to add to your context, as the "
"combined `source/path` string exactly as "
"shown in the `ls` output (e.g. "
"'homelab/active/container_caddy/caddy.md'). "
"A bare document path (without the source "
"name) will not resolve. Only pass a document "
"NOT already shown in the <documents> "
"section — it is already in your context; do "
"not re-read it."
),
}
},
"required": ["path"],
},
},
},
{
"type": "function",
"function": {
"name": "grep",
"description": (
"Search the indexed documents for an exact string "
"(case-insensitive) and return up to 20 matching lines "
"as `source/path:line: text` — a locator, not a "
"context-adder: read the winner with `read`. The "
"pattern is a plain substring, NEVER a regex — if a "
"pattern with regex syntax (like '.*' or '\\.') comes "
"back with no matches, retry with the plain text you "
"expect to see. For a normal search pass ONLY `pattern` "
"— it searches every document and that is how you "
"search the knowledge base; never pass a source name "
"as `path` (a source name is not a document). Call one "
"tool at a time — wait for this result before your "
"next call."
),
"parameters": {
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": (
"The exact text to search for (a plain "
"substring, not a regex — no '.*', no "
"'\\.', no character classes)"
),
},
"path": {
"type": "string",
"description": (
"Rarely needed — only for re-searching one "
"document you already know: that document's "
"combined `source/path` identity (e.g. "
"'homelab/ansible/inventory.yaml'). Never a "
"source name. A bare document path (without "
"the source name) will not resolve. Omit it "
"for a normal search (pass only `pattern`)."
),
},
},
"required": ["pattern"],
},
},
},
]
#: Tool refusal texts (phase 37): rejected calls count in nothing
#: (``holder.tool_calls`` tracks executed calls); the round cap bounds
#: their pathological repetition (phase 45). The in-context line is a
#: phase-72, task 05 gate-iteration teaching (live telemetry: the
#: ``lite`` model obeyed the user's "open it / read it" and re-read
#: seed-context documents, then repeated the call against the terse
#: phase-37 line — the refusal itself carried no correct action): same
#: behavior (a refusal: counts in nothing, consumes a round, changes
#: no context), the copy now names the action, so even a fired
#: refusal ends the loop instead of inviting a repeat.
ALREADY_IN_CONTEXT = (
"Already in your context — the full text is already in your "
"prompt. Do not call read on it again; answer from that text."
)
UNKNOWN_TOOL = "Unknown tool."
MISSING_READ_ARGS = "read requires a string argument 'path'."
MISSING_SEARCH_ARGS = "grep requires a string argument 'pattern'."
#: Teaching refusal for a scoped ``ls`` whose stripped ``path``
#: contains a ``/`` (phase 72): a source name is a directory basename
#: and can never contain one, so the argument is a document path passed
#: where a source name belongs (the 2026-09-03 incident's
#: ``ls(path='app/rag/importer.py')``). One ``{path}`` field — the
#: argument echoed; a fixed template states the correct contract
#: instead of the terse pre-phase-72 line, so the harness-prior misuse
#: self-corrects in one round.
LS_PATH_NOT_A_SOURCE = (
"'{path}' looks like a document path, not a source name. The "
"'path' argument of ls filters by source name (e.g. 'homelab') — "
"omit it to list every document, or read a document by its "
"combined 'source/path' string."
)
#: The no-source ``ls`` refusal with the teaching parenthetical
#: appended (phase 72): used when a stripped scope has no ``/`` and
#: matches no registered source (the incident's ``ls(path='.')``). The
#: prefix — the pre-phase-72 line — stays byte-identical; one ``{scope}``
#: field, the argument echoed.
NO_SOURCE_NOT_A_DIRECTORY = (
"No source named '{scope}' — check the ls output. (The 'path' "
"argument is a source name, not a directory — omit it to list "
"every document.)"
)
#: Teaching refusal for a ``read`` / scoped ``grep`` argument that
#: resolves to no combined identity but matches ONE indexed document's
#: ``path`` (phase 72, task 02): names the exact combined
#: ``source/path`` identity to use, so the harness-prior misuse — the
#: bare document path missing the source prefix
#: (``read('app/rag/importer.py')``) — self-corrects in one round.
#: One each of the fields ``{arg}`` (the argument echoed as passed),
#: ``{source}`` and ``{path}`` (the one candidate). Still a refusal:
#: it counts in nothing and consumes a round (no silent argument
#: normalization).
NO_DOCUMENT_DID_YOU_MEAN = (
"No document at '{arg}' — did you mean '{source}/{path}'?"
)
#: The ambiguous form of the same teaching (phase 72, task 02): the
#: argument matches SEVERAL indexed documents' ``path`` (the same path
#: under several sources). ``{candidates}`` holds up to
#: :data:`SUGGESTION_LIMIT` combined ``source/path`` identities, each
#: single-quoted, joined with ``", "`` in catalog order; ``{arg}`` is
#: the argument echoed as passed.
NO_DOCUMENT_DID_YOU_MEAN_MANY = (
"No document at '{arg}' — did you mean one of: {candidates}?"
)
#: Cap on the suggested combined identities per "did you mean …?"
#: refusal (phase 72, task 02): the same document ``path`` under
#: several sources suggests up to this many (catalog order, the rest
#: dropped).
SUGGESTION_LIMIT = 3
#: The harness-owned recovery line (phase 71, task 03) — folded into the
#: ORIGINAL single system message of the one bounded recovery request
#: (``system_prompt + "\n" + CORRECTION_INSTRUCTION``; provider-safe,
#: the user message stays last). Verbatim constant: the E2E mock
#: (task 05) keys on a stable substring of it, so it must not drift.
CORRECTION_INSTRUCTION: str = (
"Your previous reply contained raw tool-call markup, which is not "
"interpreted here. Answer the user's question directly in plain "
"text — no tool syntax."
)
class MalformedReplyError(LLMError):
"""The model kept replying in raw tool-scaffolding (phase 71).
Raised ONLY by the recovery policy — :func:`run_agent` (grounded
path) and ``app.api.chat`` (deflected path) — when the one bounded
``tools=None`` recovery still comes back with no visible content.
It is never raised from inside a stream, so
:func:`app.rag.llm.chat_stream_retried`'s retry-before-first-piece
rule never sees it. The API layer catches it BEFORE the generic
:class:`LLMError` handler and settles the turn with the dedicated
"malformed reply" error frame (no ``done``, no ``query_log`` row).
Deterministic only (owner permission 2026-09-03): no model
participates in detection or repair.
"""
#: Search caps (owner-locked A5, phase 68): a global per-call match cap
#: (across documents, in catalog order) and a per-match-line char limit.
SEARCH_MAX_MATCHES = 20
SEARCH_LINE_LIMIT = 200
#: No-match result lines (templates — the pattern is truncated to 100
#: chars before formatting, to keep a long pattern from bloating the
#: tool result). A no-match line is a *result* of an executed grep,
#: not a refusal (see the module docstring, point 3).
NO_MATCHES = "No matches for '{pattern}' in the knowledge base."
NO_MATCHES_SCOPED = "No matches for '{pattern}' in {source}/{path}."
#: Teaching no-match lines (the 2026-09-05 incident — owner chat:
#: "Qwen 3.8" question failed repeatedly). The harness-trained prior is
#: that ``grep(pattern)`` takes a REGEX (pi.dev's grep, ripgrep, grep
#: itself); this app's grep is a case-insensitive FIXED SUBSTRING
#: (owner-locked A5 — the contract does not change). A regex-shaped
#: pattern (``qwen.*3\.8``, ``qwen 3\.8``) can therefore never match, and
#: the bare no-match line above made the turbo model trust the miss and
#: end the turn with a wrong "I searched the entire knowledge base"
#: refusal. These lines are the deterministic teaching: the same result
#: (a no-match is still a *result* — counted, no context added) with the
#: correct contract stated and a plain-text retry hint (the
#: :func:`plain_form` of the pattern, when non-empty). Deterministic only
#: (owner permission 2026-09-03: "deterministic guardrails only right
#: now"): no model participates in detection or repair.
NO_MATCHES_REGEX = (
"No matches for '{pattern}'. grep matches a plain substring "
"(case-insensitive), not a regex — '.*', '\\.' and the like are "
"literal text here, so that pattern can never match. Retry with the "
"plain text you expect to see (e.g. '{plain}')."
)
NO_MATCHES_REGEX_SCOPED = (
"No matches for '{pattern}' in {source}/{path}. grep matches a plain "
"substring (case-insensitive), not a regex — retry with the plain "
"text you expect to see (e.g. '{plain}')."
)
#: One of these metacharacters anywhere in the pattern marks it
#: regex-shaped (the detection for the teaching no-match lines above).
#: Plain substrings that happen to contain one ("C++", "a|b") are
#: affected only on a NO-MATCH — a pattern that matched literally still
#: gets its ordinary result, so the teaching can never suppress a real
#: hit.
_REGEX_META_RE = re.compile(r"[*+?(){}\[\]|\\^$]")
# Backslash + a non-alphanumeric char is an escaped literal (``\\.`` →
# ``.``); backslash + an alphanumeric is a class shorthand (``\\d``,
# ``\\w``, ``\\s``) with no plain-text equivalent (→ dropped).
_ESCAPE_RE = re.compile(r"\\(.)")
def looks_like_regex(pattern: str) -> bool:
"""True when *pattern* carries regex metacharacters (see
:data:`_REGEX_META_RE`). Pure detection — the grep itself stays a
fixed substring (owner-locked A5)."""
return _REGEX_META_RE.search(pattern) is not None
def plain_form(pattern: str) -> str:
"""A deterministic plain-text retry hint for a regex-shaped pattern.
The hint is the pattern reduced to literal text, in this order:
drop ``.*`` runs on the RAW pattern first (the wildcard — an escaped
``\\.`` + ``\\*`` pair carries no raw ``.*`` run, so a literal dot
survives), unescape (``\\.`` → ``.``; class shorthands like ``\\d``
dropped), keep only the FIRST ``|`` alternative, drop character
classes (``[0-9]``), drop quantifier runs (``*``, ``+``, ``?``,
``{2,3}``), drop group parens and anchors (contents kept). Whitespace
is preserved. ``qwen.*3\\.8`` → ``qwen3.8`` (the incident's exact
recovery), ``llama\\.cpp`` → ``llama.cpp``, ``qwen[0-9]+`` →
``qwen``. A pattern made of pure metacharacters reduces to ``""`` —
callers then fall back to the ordinary no-match line (no hint).
"""
p = re.sub(r"\.\*", "", pattern) # .* wildcard runs (raw form)
p = _ESCAPE_RE.sub(lambda m: m.group(1) if not m.group(1).isalnum() else "", p)
p = p.split("|", 1)[0] # first alternative only — a hint, not an answer
p = re.sub(r"\[[^\]]*\]", "", p) # character classes carry no literal text
p = re.sub(r"\{[^{}]*\}", "", p) # {n} / {n,} / {n,m} quantifiers
p = re.sub(r"[*+?]+", "", p) # stray * + ? quantifiers
p = p.replace("(", "").replace(")", "")
p = p.replace("^", "").replace("$", "")
return p.strip()
def list_catalog(db: Session) -> list[tuple[str, str, str]]:
"""Every indexed document as ``(source, path, title)``.
Ordered by ``(source, path)`` — the same order as ``GET /api/docs``.
Module-level (not a method) so unit tests can monkeypatch it.
"""
rows = db.execute(
select(Document.source, Document.path, Document.title).order_by(
Document.source, Document.path
)
).all()
return [(source, path, title) for source, path, title in rows]
def list_source_names(db: Session) -> list[str]:
"""Every registered source name, deduped, in registry order.
The source registry (the ``git_sources`` rows — the
``BOR_GIT_SOURCES`` env fallback while the table is empty) is the
source of truth for *source* names independent of document count:
a registered source with no indexed documents still lists (as
``0 documents:`` — the scoped ``ls`` must not refuse it as unknown).
Names resolve exactly as the import pipeline indexes them
(:func:`app.rag.source_removal.resolve_source_name` — reuse, or a
scoped ``ls`` would judge the wrong names unknown, the phase-69
"RAG consistent with the registry" invariant); two rows resolving
to the same name (the phase-69 sibling case) share documents, so
the name is listed once. Module-level (not a method) so unit tests
can monkeypatch it.
"""
rows, _origin = effective_sources(db)
names: list[str] = []
for row in rows:
name = resolve_source_name(row)
if name not in names:
names.append(name)
return names
def find_document(db: Session, source: str, path: str) -> Document | None:
"""The indexed document at ``(source, path)``, or ``None``.
Module-level (not a method) so unit tests can monkeypatch it.
"""
return db.scalar(
select(Document).where(Document.source == source, Document.path == path)
)
def _resolve_path(db: Session, combined: str) -> tuple[Document | None, str, str]:
"""The combined ``source/path`` identity → document (phase 70).
The canonical document identity in every tool argument, refusal and
result header is the combined string exactly as printed in the
``ls`` output, the ``Document …`` result headers, and the grep
result lines. Source names are directory basenames (``app.rag.importer``:
``source = root.name``) and can never contain a ``'/'``, so the
split at the FIRST slash is exact: the part before is the source
name, the part after is the path. Returns ``(doc, source, path)``
with the split pair (so callers can echo the canonical form, e.g.
the scoped no-match line); no ``'/'`` in the argument →
``(None, combined, "")`` — a bare source name is never a document
(no DB lookup; the refusal echoes the argument as passed).
"""
if "/" not in combined:
return None, combined, ""
source, _, path = combined.partition("/")
return find_document(db, source, path), source, path
def all_documents(db: Session) -> list[Document]:
"""Every indexed document (full rows), ordered by ``(source, path)``
— catalog order.
The whole-KB ``grep`` path loads all contents in this one bulk query
(catalog order is the locked match order, owner-locked A5).
Module-level (not a method) so unit tests can monkeypatch it.
"""
return list(
db.execute(
select(Document).order_by(Document.source, Document.path)
).scalars()
)
def find_path_candidates(db: Session, arg: str) -> list[tuple[str, str, str]]:
"""The indexed documents a bare document *arg* names by ``path``.
Phase 72, task 02: a ``read`` / scoped-``grep`` argument that
resolves to no combined identity but *is* a document path (contains
``/``) is matched against the indexed ``Document.path`` values so
the refusal can name the combined ``source/path`` identity to use
(the "did you mean …?" teaching). The documents whose ``path``
equals *arg* (the exact bare path) or ends with ``f"/{arg}"`` (the
file is nested deeper — the suffix match) — in catalog order (the
:func:`all_documents` order), case-sensitive (these are file
paths) — as ``(source, path, title)`` triples. One bulk query via
:func:`all_documents` (at most one); called ONLY from the refusal
path of :func:`_execute_tool` (never on the happy path) and only
when *arg* contains ``/`` (a bare name keeps today's no-DB-lookup
refusal). Module-level (not a method) so unit tests can
monkeypatch it.
"""
return [
(doc.source, doc.path, doc.title)
for doc in all_documents(db)
if doc.path == arg or doc.path.endswith(f"/{arg}")
]
def _no_document_refusal(db: Session, arg: str) -> str:
"""The no-document refusal for an unresolved ``read`` / scoped-
``grep`` argument (phase 72, task 02).
The pre-phase-72 line — the argument echoed as passed — whenever
there is nothing to suggest: a bare argument (no ``/`` — a bare
source name or any other bare name gets the no-DB-lookup refusal,
byte-identical to today) or a path-like argument that matches no
indexed document's ``path`` (zero candidates). A path-like argument
(contains ``/``) that matches exactly one indexed document's
``path`` gets :data:`NO_DOCUMENT_DID_YOU_MEAN` (the combined
identity named); two or more get :data:`NO_DOCUMENT_DID_YOU_MEAN_MANY`
(up to :data:`SUGGESTION_LIMIT`, catalog order). Deterministic
only: the suggestion is a pure catalog lookup, no model. A refusal
still counts in nothing and consumes a round.
"""
if "/" in arg:
candidates = find_path_candidates(db, arg)
if len(candidates) == 1:
source, path, _title = candidates[0]
return NO_DOCUMENT_DID_YOU_MEAN.format(arg=arg, source=source, path=path)
if len(candidates) > 1:
identities = ", ".join(
f"'{source}/{path}'"
for source, path, _title in candidates[:SUGGESTION_LIMIT]
)
return NO_DOCUMENT_DID_YOU_MEAN_MANY.format(arg=arg, candidates=identities)
return f"No document at '{arg}' — check the ls output."
def grep_document(content: str, pattern: str) -> list[tuple[int, str]]:
"""Every line of *content* that contains *pattern*, in file order.
Case-insensitive **fixed substring** (owner-locked A5: no regex — no
ReDoS surface, a simple contract for the model). Returns
``(1-based line number, line.rstrip())`` pairs; an empty *content*
never matches a non-empty pattern.
"""
needle = pattern.lower()
return [
(number, line.rstrip())
for number, line in enumerate(content.split("\n"), start=1)
if needle in line.lower()
]
@dataclass
class AgentHolder:
"""Per-turn agent state the API layer reads after the stream (task 04).
``read_docs``: the documents ``read`` added to the context, in read
order (deduped — re-reading a document appends nothing).
``tool_calls``: how many tool calls executed (re-lists included);
rejected calls (unknown tool, unknown/missing arguments or document,
already-in-context) do not count. Drives the per-turn log line's
``tool_calls=N`` field (task 04).
``scaffold_stripped``: how many chars of tool-scaffolding the
turn's filters removed across the turn's requests (rounds + the
forced final + any recovery, phase 71) — drives the per-turn log
line's ``scaffold_stripped=N`` field on grounded turns (the
deflected path computes its own total in ``app.api.chat``).
"""
read_docs: list[Document] = field(default_factory=list)
tool_calls: int = 0
scaffold_stripped: int = 0
def _execute_tool(
db: Session,
call: ToolCallPiece,
seed_docs: Sequence[Document],
holder: AgentHolder,
) -> str:
"""Execute one tool call server-side (DB only).
Returns the tool result text. A successful call bumps
``holder.tool_calls`` (a successful ``read`` also appends the
:class:`Document` to ``holder.read_docs``; a ``grep`` never does —
it is a locator, locked A5); rejected calls return their refusal
line and count in nothing. A grep that ran but found nothing is
still a successful (counted) call — its no-match line is a result,
not a refusal. Document targets are combined ``source/path``
strings, resolved by :func:`_resolve_path` (the canonical identity,
phase 70).
"""
if call.name == "ls":
raw_path = call.arguments.get("path")
scope = raw_path.strip() if isinstance(raw_path, str) else ""
rows = list_catalog(db)
if scope:
if "/" in scope:
# A source name (a directory basename) can never
# contain '/' — this is a document path where a source
# name belongs (phase 72): teach the contract; no
# registry lookup needed, counts in nothing, consumes
# a round like every refusal.
return LS_PATH_NOT_A_SOURCE.format(path=scope)
if scope not in list_source_names(db):
# The no-source refusal with the teaching parenthetical
# (phase 72) — the prefix byte-identical to the
# pre-phase-72 line; counts in nothing, consumes a
# round like every refusal.
return NO_SOURCE_NOT_A_DIRECTORY.format(scope=scope)
rows = [row for row in rows if row[0] == scope]
listing = f"{len(rows)} documents:\n" + "\n".join(
f"source: {source} | path: {path} | title: {title}"
for source, path, title in rows
)
holder.tool_calls += 1
return listing
if call.name == "read":
raw_path = call.arguments.get("path")
arg = raw_path.strip() if isinstance(raw_path, str) else ""
if not arg:
return MISSING_READ_ARGS
known = {(doc.source, doc.path) for doc in (*seed_docs, *holder.read_docs)}
# The dedupe check needs no DB: the split pair of a combined
# identity that is in context is in `known` as-is (the resolve
# below would find the same document).
if "/" in arg:
src, _, p = arg.partition("/")
if (src, p) in known:
return ALREADY_IN_CONTEXT
doc, _source, _path = _resolve_path(db, arg)
if doc is None:
# Echo the argument as passed — the model sees its own form
# (a bare argument can never be a document, no DB lookup);
# a path-like argument that matches an indexed document's
# path gets the "did you mean …?" teaching (phase 72).
return _no_document_refusal(db, arg)
holder.read_docs.append(doc)
holder.tool_calls += 1
return f"Document {doc.source}/{doc.path}:\n{doc.content}"
if call.name == "grep":
raw_pattern = call.arguments.get("pattern")
pattern = raw_pattern.strip() if isinstance(raw_pattern, str) else ""
if not pattern:
return MISSING_SEARCH_ARGS
raw_path = call.arguments.get("path")
scope = raw_path.strip() if isinstance(raw_path, str) else ""
scoped_to: tuple[str, str] | None = None
if scope:
target, src, p = _resolve_path(db, scope)
if target is None:
# The same phase-72 "did you mean …?" teaching as the
# read branch (a refusal — not counted, no context).
return _no_document_refusal(db, scope)
docs: list[Document] = [target]
scoped_to = (src, p) # the resolved (canonical) identity
else:
docs = all_documents(db)
matches: list[str] = []
for doc in docs:
for lineno, line in grep_document(doc.content, pattern):
matches.append(
f"{doc.source}/{doc.path}:{lineno}: {line[:SEARCH_LINE_LIMIT]}"
)
if len(matches) >= SEARCH_MAX_MATCHES:
break
if len(matches) >= SEARCH_MAX_MATCHES:
break # the global cap is hit — stop scanning
holder.tool_calls += 1 # the grep executed (no-match counts too)
# Locked A5: a grep never adds context — read_docs untouched.
if not matches:
shown = pattern[:100] # keep a long pattern short in the line
# The 2026-09-05 incident teaching: a regex-shaped pattern
# (the harness prior) can NEVER match a fixed-substring
# grep, so a no-match for one is not "the KB lacks this" —
# it is "the pattern was in the wrong form". Teach the
# contract and hand over the plain-form retry hint (when the
# reduction is non-empty); a plain pattern (or a pattern
# that reduces to nothing) keeps the ordinary line
# byte-identical.
plain = plain_form(pattern) if looks_like_regex(pattern) else ""
if plain:
if scoped_to is not None:
return NO_MATCHES_REGEX_SCOPED.format(
pattern=shown,
source=scoped_to[0],
path=scoped_to[1],
plain=plain[:100],
)
return NO_MATCHES_REGEX.format(pattern=shown, plain=plain[:100])
if scoped_to is not None:
# The scoped no-match line is keyed on the resolved
# source/path (== the argument, stripped).
return NO_MATCHES_SCOPED.format(
pattern=shown, source=scoped_to[0], path=scoped_to[1]
)
return NO_MATCHES.format(pattern=shown)
return "\n".join(matches)
return UNKNOWN_TOOL
async def run_agent(
llm: LLMClient,
db: Session,
*,
system_prompt: str,
user_message: str,
seed_docs: Sequence[Document],
settings: Settings,
holder: AgentHolder,
history: Sequence[dict[str, Any]] = (),
) -> AsyncIterator[StreamPiece | ToolCallPiece | RetryPiece]:
"""Run the grounded-turn tool loop, yielding every stream piece.
Every piece (``thinking`` / ``content`` / tool calls /
:class:`RetryPiece`) is yielded as it arrives; the API layer (task 04)
turns tool-call pieces into SSE ``tool`` events and retry pieces into
SSE ``retry`` events. After the loop finishes, *holder* carries the
read documents and the executed tool-call count (re-lists included).
History (phase 74, TODO L4): *history* is the client's prior turns
already mapped to model messages by
:func:`app.rag.prompts.history_to_messages` (trimmed newest-first
against the settings budgets; assistant turns carry their prior
thinking as ``reasoning_content``). It is spliced between the system
prompt and the current user message —
``[system, *history, user]`` — and everything downstream (the tool
rounds, the phase-71 recovery rebuilding from ``messages[1:]``, the
retry restarts) already operates on that one ``messages`` list,
unchanged. ``()`` (the default) keeps the pre-phase-74 two-message
request byte-identical.
Retries (phase 67, owner-locked A2): every model request goes through
:func:`chat_stream_retried` — a failed round is retried **before** its
first piece (same messages, ``settings.llm_retries`` restarts, a flat
``settings.llm_retry_delay``); a round that already streamed pieces
fails the turn as before.
Scaffolding recovery (phase 71, deterministic only): every request —
each round, the forced final, and any recovery — runs its content
through a fresh :class:`app.rag.scaffolding.ScaffoldingFilter`. A
round that ends with NO visible content but a non-empty strip (the
scaffolding was the whole "answer") gets exactly ONE recovery:
``tools=None``, :data:`CORRECTION_INSTRUCTION` folded into the
original single system message (the rest of the history — user
message and tool results — unchanged), a fresh filter, the same
retry budget. A clean recovery ends the turn; a second empty reply
raises :class:`MalformedReplyError` (terminal — the API layer turns
it into the dedicated error frame). A round with visible content
plus scaffolding needs no recovery (the clean content stands), and
a scaffolding-only round that also carried tool calls needs none
(the tool ran) — the policy keys on the no-calls exit only. The
per-span strip warning log (each span truncated to 200 chars) is
the capture mechanism for new registry entries; *holder* accumulates
the turn's ``scaffold_stripped`` total for the API layer's log line.
``seed_docs`` are the documents the retrieval already put in context
(they shape the *system_prompt* the caller built); re-reading one of
them is rejected with :data:`ALREADY_IN_CONTEXT` (the phase-72
teaching line — answer from the text already in the prompt) — the
rejection counts in nothing, but it still consumes a round.
"""
messages: list[dict[str, Any]] = [
{"role": "system", "content": system_prompt},
*history, # phase 74: the client's prior turns (empty by default)
{"role": "user", "content": user_message},
]
# Phase 45: no per-tool budgets — the tools stay offered for the
# whole turn, bounded by the round cap. ``0`` is the no-tools kill
# switch: exactly one request with ``tools=None`` (the pre-phase-37
# path).
max_rounds = settings.agent_max_rounds
tools: list[dict[str, Any]] | None = AGENT_TOOLS if max_rounds > 0 else None
rounds = 0
while True:
calls: list[ToolCallPiece] = []
# Phase 71: one fresh filter per round (one per model request —
# the retry attempts of this logical request share it: a restart
# only happens while the filter was never fed). Content-only:
# thinking pieces pass through raw.
round_filter = ScaffoldingFilter()
round_content = 0 # visible (clean) content chars this round
# Phase 48: bind the round's stream so a consumer abandon
# (GeneratorExit into the yield below) tears down the in-flight
# model stream deterministically — not GC-dependent. Phase 67:
# the round goes through the retry primitive — a failure before
# the first piece restarts the request (locked A2) after a
# RetryPiece; closing the OUTER generator propagates GeneratorExit
# into ``chat_stream_retried``, whose own ``finally`` closes the
# in-flight inner ``chat_stream``, so teardown stays deterministic
# on consumer abandon. Awaiting ``aclose()`` in the ``finally`` is
# safe because it does not yield; on a fully consumed round it is
# a quiet no-op.
stream = chat_stream_retried(
llm,
messages,
tools=tools,
retries=settings.llm_retries,
delay=settings.llm_retry_delay,
scaffolding=round_filter,
)
try:
async for piece in stream:
if isinstance(piece, ToolCallPiece):
calls.append(piece)
elif isinstance(piece, StreamPiece) and piece.kind == "content":
round_content += len(piece.text)
yield piece
finally:
await stream.aclose()
# Phase 71: the per-strip-event capture log — one warning per
# stripped span, truncated to 200 chars (how a new scaffolding
# format gets captured and added to the registry) — and the turn
# total for the API layer's ``scaffold_stripped=N`` log field.
holder.scaffold_stripped += round_filter.stripped_chars
for span in round_filter.stripped_spans:
logger.warning(
"agent: stripped %d chars of tool-scaffolding in round %d: %r",
len(span),
rounds + 1,
span[:200],
)
if not calls:
if round_content > 0:
return # the answer was streamed (the clean content stands)
if round_filter.stripped_chars == 0:
# Empty/thinking-only answer — today's behavior, unchanged
# (the UI handles it); the guardrail keys on a strip.
return
# Phase 71: the scaffolding was the whole "answer" — the ONE
# bounded recovery (a fixed policy, not a conversation):
# ``tools=None``, the correction folded into the ORIGINAL
# single system message (provider-safe — the user message and
# any tool history stay in place), a fresh filter, the same
# phase-67 retry budget.
logger.warning(
"agent: round %d was pure tool-scaffolding (%d chars stripped) "
"— running the one bounded recovery",
rounds + 1,
round_filter.stripped_chars,
)
messages_recovered = [
{
"role": "system",
"content": system_prompt + "\n" + CORRECTION_INSTRUCTION,
},
*messages[1:],
]
recovery_filter = ScaffoldingFilter()
recovered = chat_stream_retried(
llm,
messages_recovered,
tools=None,
retries=settings.llm_retries,
delay=settings.llm_retry_delay,
scaffolding=recovery_filter,
)
recovery_content = 0
try:
async for piece in recovered:
if isinstance(piece, StreamPiece) and piece.kind == "content":
recovery_content += len(piece.text)
yield piece
finally:
await recovered.aclose()
holder.scaffold_stripped += recovery_filter.stripped_chars
for span in recovery_filter.stripped_spans:
logger.warning(
"agent: stripped %d chars of tool-scaffolding in the "
"recovery after round %d: %r",
len(span),
rounds + 1,
span[:200],
)
if recovery_content > 0:
return # the recovery answered — the turn ends
# The second empty reply is terminal (at most one recovery per
# turn). Raised OUTSIDE the stream, so chat_stream_retried's
# retry rule never sees it; the API layer catches it before
# the generic LLMError handler.
logger.warning(
"agent: the recovery reply was still empty "
"(scaffold_stripped=%d) — settling with a malformed-reply error",
holder.scaffold_stripped,
)
raise MalformedReplyError(
f"the model answered in raw tool-scaffolding twice in a row "
f"(round {rounds + 1} plus one recovery) — no clean answer "
"to stream"
)
call = calls[0] # a stream can carry several calls; run the first
result = _execute_tool(db, call, seed_docs, holder)
rounds += 1 # every call the model emits consumes a round
logger.info(
"agent tool=%s args=%s round=%d/%d",
call.name,
json.dumps(call.arguments, ensure_ascii=False)[:200],
rounds,
max_rounds,
)
messages.append(
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": call.id,
"type": "function",
"function": {
"name": call.name,
"arguments": json.dumps(call.arguments),
},
}
],
}
)
messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
if rounds >= max_rounds:
logger.warning(
"agent round cap reached (rounds=%d) — forcing a final "
"no-tools answer",
rounds,
)
# Phase 48: the forced final answer gets the same explicit
# teardown as the loop rounds (consumer abandon mid-final
# answer must still close the model's stream). Phase 67: the
# forced call retries under the same locked-A2 rule as the
# loop rounds. Phase 71: the forced final runs through a
# fresh filter too — raw scaffolding can never reach the
# user from ANY grounded request.
final_filter = ScaffoldingFilter()
final = chat_stream_retried(
llm,
messages,
tools=None,
retries=settings.llm_retries,
delay=settings.llm_retry_delay,
scaffolding=final_filter,
)
final_content = 0
try:
async for piece in final:
if isinstance(piece, StreamPiece) and piece.kind == "content":
final_content += len(piece.text)
yield piece
finally:
await final.aclose()
# Phase 71: the same capture log + turn total; a
# scaffolding-only forced final (this turn used no recovery,
# so nothing is doubled up) settles with the same terminal
# malformed-reply error rather than a silently empty answer.
holder.scaffold_stripped += final_filter.stripped_chars
for span in final_filter.stripped_spans:
logger.warning(
"agent: stripped %d chars of tool-scaffolding in the "
"forced final answer (round %d): %r",
len(span),
rounds + 1,
span[:200],
)
if final_content == 0 and final_filter.stripped_chars > 0:
logger.warning(
"agent: the forced final answer was pure tool-scaffolding "
"— settling with a malformed-reply error"
)
raise MalformedReplyError(
"the forced final answer was raw tool-scaffolding — no "
"clean answer to stream"
)
return