Foundations

turn 1   system(1,800) + user(40)                                      =  1,840 tokens in
turn 2   system(1,800) + user(40) + assistant(220) + user(35)           =  2,095 tokens in
turn 9   system(1,800) + 8 earlier turns(2,900) + user(50)              =  4,750 tokens in

# api/schemas.py
class ChatMessage(BaseModel):
    role: Literal["user", "assistant"]
    content: str = Field(min_length=1, max_length=32_000)   # hard upper bound


    @field_validator("content")
    @classmethod
    def _clean(cls, v: str) -> str:
        cleaned = clean_user_text(v)       # strips control, zero-width and bidi characters
        if not cleaned:
            raise ValueError("message is empty after removing control characters")
        return cleaned




class ChatRequest(BaseModel):
    messages: list[ChatMessage] = Field(min_length=1, max_length=50)


    @model_validator(mode="after")
    def _alternating_user_first_and_last(self) -> ChatRequest:
        roles = [m.role for m in self.messages]
        if roles[0] != "user" or roles[-1] != "user":
            raise ValueError("conversation must start and end with a user message")
        if any(a == b for a, b in itertools.pairwise(roles)):
            raise ValueError("roles must alternate")
        return self

# api/chat.py
def _validate_limits(body: ChatRequest, settings: Settings) -> None:
    if len(body.messages) > settings.max_history_messages:
        raise HTTPException(422, f"too many messages (max {settings.max_history_messages})")
    for m in body.messages:
        # Users are capped tightly; prior assistant turns are our own (bounded by
        # max_tokens) and echoed back by the client, so they get a looser cap.
        limit = settings.max_message_chars if m.role == "user" else settings.max_assistant_chars
        if len(m.content) > limit:
            raise HTTPException(422, f"{m.role} message too long (max {limit} chars)")

# llm/types.py
@dataclass(frozen=True, slots=True)
class LLMRequest:
    model: str
    messages: list[dict[str, Any]]
    system: str | list[dict[str, Any]] | None = None
    tools: list[dict[str, Any]] | None = None
    max_tokens: int = 16000
    effort: Effort | None = None               # "low" ... "max": the cost/quality dial
    json_schema: dict[str, Any] | None = None  # structured output
    cache: bool = False                        # cache the stable prefix
    purpose: str = "answer"                    # answer | triage | judge | route: a metrics label




@dataclass(frozen=True, slots=True)
class LLMResult:
    text: str
    content: list[Any]          # raw blocks: replay these unchanged in a tool loop
    stop_reason: str | None
    usage: Usage
    model: str                  # the model that actually served the call
    tool_calls: list[ToolCall] = field(default_factory=list)
    request_id: str | None = None
    served_by_fallback: bool = False


    @property
    def refused(self) -> bool:
        return self.stop_reason == "refusal"




LLMEvent = TextDelta | StreamReset | LLMResult




class LLM(Protocol):
    def stream(self, request: LLMRequest) -> AsyncGenerator[LLMEvent, None]:
        """Yield TextDelta (and rarely StreamReset) events, then exactly one LLMResult."""
        ...

async def complete(llm: LLM, request: LLMRequest) -> LLMResult:
    result: LLMResult | None = None
    async for event in llm.stream(request):
        if isinstance(event, LLMResult):
            result = event
    if result is None:
        raise RuntimeError("LLM stream ended without a result")
    return result

# llm/client.py
class AnthropicLLM:
    async def stream(self, request: LLMRequest) -> AsyncGenerator[LLMEvent, None]:
        kwargs = self._build_kwargs(request)
        try:
            async with self._client.messages.stream(**kwargs) as stream:
                async for event in stream:
                    if event.type == "content_block_delta" and event.delta.type == "text_delta":
                        yield TextDelta(event.delta.text)
                final = await stream.get_final_message()
                request_id = stream.request_id
        except anthropic.APIError as exc:
            raise LLMError.from_anthropic(exc) from exc   # never leak SDK exceptions
        yield _to_result(final, request_id)


    @staticmethod
    def _build_kwargs(request: LLMRequest) -> dict[str, Any]:
        kwargs: dict[str, Any] = {
            "model": request.model,
            "max_tokens": request.max_tokens,
            "messages": request.messages,
        }
        if request.system is not None:
            kwargs["system"] = request.system
        if request.tools:
            kwargs["tools"] = request.tools
        output_config: dict[str, Any] = {}
        if request.effort is not None:
            output_config["effort"] = request.effort
        if request.json_schema is not None:
            output_config["format"] = {"type": "json_schema", "schema": request.json_schema}
        if output_config:
            kwargs["output_config"] = output_config
        if request.cache:
            kwargs["cache_control"] = {"type": "ephemeral"}
        # Deliberately absent: temperature/top_p (rejected by current models) and
        # `thinking` (adaptive by default; effort is the control knob).
        return kwargs

def _to_result(message: Any, request_id: str | None = None) -> LLMResult:
    blocks = list(message.content)
    text = "".join(b.text for b in blocks if b.type == "text")
    tool_calls = [
        ToolCall(id=b.id, name=b.name, input=dict(b.input))
        for b in blocks if b.type == "tool_use"
    ]
    u = message.usage
    usage = Usage(
        input_tokens=u.input_tokens or 0,
        output_tokens=u.output_tokens or 0,
        cache_read_input_tokens=getattr(u, "cache_read_input_tokens", 0) or 0,
        cache_creation_input_tokens=getattr(u, "cache_creation_input_tokens", 0) or 0,
    )
    return LLMResult(text=text, content=blocks, stop_reason=message.stop_reason,
                     usage=usage, model=message.model, tool_calls=tool_calls,
                     request_id=request_id)

event: tool_start
data: {"id": "toolu_01", "name": "search_docs"}

event: delta
data: {"text": "Refunds are available within 30 days "}

event: delta
data: {"text": "of the charge date [policies/refunds.md]."}

event: citations
data: {"sources": [{"path": "policies/refunds.md", "title": "Refund Policy", "section": "Eligibility"}]}

event: done
data: {"stop_reason": "end_turn", "usage": {"input_tokens": 4120, "output_tokens": 96}, "request_id": "b3f1..."}

# api/chat.py (trimmed)
def _sse(event: str, data: dict[str, Any]) -> dict[str, str]:
    return {"event": event, "data": json.dumps(data, ensure_ascii=False)}


async def _to_sse(events, request_id, tel, ...) -> AsyncIterator[dict[str, str]]:
    """Translate agent events to SSE. Always ends the stream with done|error."""
    try:
        async for event in events:
            if isinstance(event, TextDelta):
                yield _sse("delta", {"text": event.text})
            elif isinstance(event, ToolStart):
                yield _sse("tool_start", {"id": event.id, "name": event.name})
            elif isinstance(event, AgentFinished):
                yield _sse("done", {"stop_reason": event.stop_reason, "request_id": request_id})
    except LLMError as exc:
        yield _sse("error", {"code": exc.kind, "retryable": exc.retryable,
                             "message": exc.public_message})
    except (GeneratorExit, asyncio.CancelledError):
        raise                    # the client went away: not an error
    except Exception:
        # HTTP 200 is already on the wire: never let a stream die silently.
        log.exception("chat_unexpected_error")
        yield _sse("error", {"code": "unknown", "retryable": False,
                             "message": "Something went wrong. Please retry."})


@router.post("/chat")
async def chat(body: ChatRequest, request: Request, user_id: ...) -> EventSourceResponse:
    _validate_limits(body, request.app.state.settings)
    events = request.app.state.agent.run([m.model_dump() for m in body.messages], ...)
    return EventSourceResponse(_to_sse(events, request.state.request_id, ...))

// web/app.js
async function* sse(response) {
  const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
  let buf = "";
  for (;;) {
    const { value, done } = await reader.read();
    if (done) return;
    // Normalise on the accumulated buffer: "\r\n" may be split across chunks.
    buf = (buf + value).replace(/\r\n/g, "\n");
    let idx;
    while ((idx = buf.indexOf("\n\n")) >= 0) {
      const raw = buf.slice(0, idx); buf = buf.slice(idx + 2);
      let event = "message", data = "";
      for (const line of raw.split("\n")) {
        if (line.startsWith("event:")) event = line.slice(6).trim();
        else if (line.startsWith("data:")) data += line.slice(5).trim();
      }
      if (data) yield { event, data: JSON.parse(data) };
    }
  }
}

# llm/errors.py
ErrorKind = Literal["rate_limited", "overloaded", "timeout", "connection",
                    "auth", "bad_request", "not_found", "server", "unknown"]


_RETRYABLE = frozenset({"rate_limited", "overloaded", "timeout", "connection", "server"})


class LLMError(Exception):
    @property
    def retryable(self) -> bool:
        return self.kind in _RETRYABLE


    @property
    def public_message(self) -> str:
        """Safe to show end users: never leaks provider internals."""
        return _PUBLIC_MESSAGES[self.kind]


    @classmethod
    def from_anthropic(cls, exc: anthropic.APIError) -> LLMError:
        """Most-specific class first; APITimeoutError subclasses APIConnectionError."""
        if isinstance(exc, anthropic.AuthenticationError | anthropic.PermissionDeniedError):
            kind = "auth"
        elif isinstance(exc, anthropic.RateLimitError):
            kind = "rate_limited"
        elif isinstance(exc, anthropic.APITimeoutError):
            kind = "timeout"
        elif isinstance(exc, anthropic.APIConnectionError):
            kind = "connection"
        elif isinstance(exc, anthropic.APIStatusError) and exc.status_code < 400:
            # Errors sent *inside* an SSE stream carry the 200 of the stream itself;
            # the real classification is in the body.
            kind = _kind_from_body(exc.body)
        ...
        return cls(str(exc), kind=kind, status_code=..., request_id=...)

def build_anthropic_client(settings: Settings) -> anthropic.AsyncAnthropic:
    return anthropic.AsyncAnthropic(
        api_key=...,
        timeout=anthropic.Timeout(settings.llm_timeout_s, connect=settings.llm_connect_timeout_s),
        max_retries=settings.llm_max_retries,
    )

# prompts.py
_FILE_RE = re.compile(r"^(?P<name>[a-z_]+)\.v(?P<version>\d+)\.md$")


@dataclass(frozen=True, slots=True)
class Prompt:
    name: str
    version: int
    text: str
    sha: str


    @property
    def ref(self) -> str:
        return f"{self.name}@v{self.version}"           # "answer@v4" in every log line


    def render(self, **variables: str) -> str:
        return Template(self.text).substitute(**variables)   # a missing variable raises




class PromptRegistry:
    def __init__(self, directory: Path) -> None:
        self._prompts: dict[tuple[str, int], Prompt] = {}
        for path in sorted(directory.glob("*.md")):
            if (match := _FILE_RE.match(path.name)) is None:
                continue
            text = path.read_text(encoding="utf-8").strip()
            sha = hashlib.sha256(text.encode()).hexdigest()[:12]
            self._prompts[(match["name"], int(match["version"]))] = Prompt(
                match["name"], int(match["version"]), text, sha)

# config.py
@model_validator(mode="after")
def _require_key_in_prod(self) -> Settings:
    if self.strict and self.anthropic_api_key is None:
        raise ValueError("ANTHROPIC_API_KEY is required outside dev/test")
    if self.strict and self.llm_backend != "anthropic":
        raise ValueError("DOCSPILOT_LLM_BACKEND must be 'anthropic' outside dev/test")
    if self.strict and self.resolved_embedder == "hash":
        raise ValueError("a real embedder (VOYAGE_API_KEY) is required outside dev/test")
    if self.strict and self.injection_policy == "off":
        raise ValueError("DOCSPILOT_INJECTION_POLICY must not be 'off' outside dev/test")
    if self.strict and (self.user_daily_budget_usd <= 0 or self.global_daily_budget_usd <= 0):
        raise ValueError("daily budgets are required outside dev/test")
    ...

# llm/fake.py
@dataclass
class Script:
    """One scripted model turn."""
    text: str = ""
    tool_calls: Sequence[ToolCall] = ()
    stop_reason: str | None = None
    error: LLMError | None = None
    usage: Usage = field(default_factory=lambda: Usage(input_tokens=10, output_tokens=5))




class FakeLLM:
    """Plays scripted turns in order and records every request it received."""
    def __init__(self, scripts: Sequence[Script] | Callable[[LLMRequest], Script] | None = None):
        self._scripts = scripts if scripts is not None else [Script(text="ok")]
        self.requests: list[LLMRequest] = []

# tests/test_agent.py
async def test_read_tool_error_is_fed_back_and_assistant_blocks_replayed_unchanged() -> None:
    agent, llm, tickets, _ = build([
        Script(tool_calls=[call("t1", "get_ticket", ticket_id="TCK-deadbeef")]),
        Script(text="I could not find that ticket."),
    ])
    events = await collect(agent.run(HISTORY, ctx(tickets)))


    second = llm.requests[1].messages
    block = second[-1]["content"][0]
    assert block["tool_use_id"] == "t1" and block["is_error"] is True
    assert second[-2]["content"][0]["type"] == "tool_use"
    assert isinstance(events[-1], AgentFinished) and events[-1].steps == 2