RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(mock-provider): deterministic OpenAI-compatible endpoint for local and CI use (#1094) 

* feat(mock-provider): deterministic OpenAI-compatible endpoint for local and CI use

Adds src/mock_provider/, a standalone ASGI app that lets Honcho run with no
model provider, no API key, and no spend. It answers /v1/chat/completions and
/v1/embeddings with obviously-synthetic content derived from the request, so
the same request always produces the same response.

It runs as its own service from the standard Honcho image with a different
entrypoint, the way api and deriver already differ, so there is no second image
to build or keep in digest-sync. The app imports nothing from src.config or
src.db, so it boots even when the rest of the stack is misconfigured.

The chat endpoint generates from the JSON Schema it is sent rather than
answering with prose. That matters because a prose answer does not fail loudly:
repair_response_model_json swallows the parse error and returns an empty
PromptRepresentation, which reads as "the deriver found nothing" rather than
"the mock is wrong". Generation resolves $ref/$defs indirection, caps recursion
for reasoning-tree schemas, and covers json_object mode by recovering the
schema Honcho injects into the prompt. Embeddings are hash-derived, so
identical input yields an identical vector.

Tests drive the production OpenAIBackend and _EmbeddingClient against the app
over ASGI, including the strict json_schema transform that
chat.completions.parse() applies. Verified end to end against a real stack:
messages in, conclusions and 1536-dim embeddings written to pgvector, with no
calls to any real provider.

Mock embeddings carry no semantic similarity, so recall against this provider
must use lexical search. CONTRIBUTING notes that, and the load_dotenv(override=
True) behaviour that lets a stale repo .env win over exported environment
variables.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* refactor(mock-provider): validate requests with Pydantic models

Review feedback: hand-coercing the request bodies was defended on the grounds
that FastAPI answers a malformed body with a 422, and a 422 mid-deriver-run
reads as a Honcho bug. That argues against the default handler, not against the
models. Registering an exception handler fixes it — and the resulting behaviour
is more faithful, not less, because the real API answers a bad request with a
400 and an `error` envelope, which is now exactly what the mock returns.

Adds src/mock_provider/schemas.py with ChatCompletionRequest and
EmbeddingsRequest. Every model allows extra fields and every field is optional,
so validation fires on a wrong type rather than on a parameter the mock has not
heard of — a new upstream parameter must not turn a working setup into a hard
failure. dimensions is a StrictInt because bool is an int subclass and a JSON
`true` would otherwise mean a one-dimensional vector.

coerce.py stays, narrowed to serving schema_gen, which walks arbitrary
caller-supplied JSON Schema and is untyped by nature. response_format likewise
stays dict[str, Any]: only its envelope is worth typing.

Also records why schema_gen does not reuse src/utils/schema_conversion.py
despite the overlapping $ref/$defs handling — it builds a model class rather
than an instance, raises by contract where a mock must degrade, and rejects
both allOf and the recursive $ref that reasoning-tree schemas rely on.

Documents that LLM_OPENAI_API_KEY is only tested for truthiness; the previous
wording read as though the value had to be the literal string "sandbox".

Re-verified end to end after the refactor: 6 messages in, 4 conclusions and 6
1536-dim embeddings out, every real request answered 200, no calls to any real
provider.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(mock-provider): honour include_usage, generate prefixItems tuples

Three fidelity gaps where the mock answered a request differently from the
API it stands in for:

- The usage chunk was emitted on every stream. The real API sends it only
  when stream_options.include_usage is set, so a caller that did not opt in
  had to skip a trailing chunk with an empty choices array. stream_options
  is now a typed model, which also rejects a non-boolean include_usage
  instead of reading it as truthy.
- A fixed-length tuple is prefixItems with no items, which is what Pydantic
  emits for tuple[str, int]. Reading only items returned [], failing the
  minItems the same schema carries — the silent-empty failure schema_gen
  exists to avoid.
- A zero or negative dimensions was silently replaced with 1536, answering
  a bad request with a plausible-looking vector rather than a 400.

Three further deviations from JSON Schema are left in place and documented
where they occur: allOf merges properties first-wins, oneOf is treated as
anyOf, and string pattern is ignored. None is reachable from a Honcho
response model — no model emits prefixItems or oneOf, and the only pattern
constraints are on API request models — and each fix costs more than the
unreachable path is worth.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(mock-provider): strict request booleans, bounded recursion, multipleOf

Second CodeRabbit pass. All four findings reproduced first; none is reachable
from a Honcho response model, but two trace back to the previous commit.

- `include_usage` and `stream` were plain `bool`, which Pydantic coerces from
  "yes"/"on"/"true"/"1". The comment added last commit claimed a string had to
  fail here, and it did not — the test only passed because "definitely" is not
  a recognised bool literal. Both are StrictBool now, matching why `dimensions`
  is StrictInt, and the tests cover the truthy strings that actually coerced.
- `_generate_array` returned the prefix alone when `items` was absent, so
  prefixItems plus a larger minItems undershot its own schema. Absent `items`
  leaves those positions unconstrained rather than disallowed, so the shortfall
  is filled to minItems — a bare `{"type": "array"}` still generates nothing.
- A required, non-nullable recursive $ref hit RecursionError: MAX_DEPTH only
  terminates a cycle that offers a `default` or a nullable branch, and
  `_generate_object` keeps descending into required properties. HARD_MAX_DEPTH
  degrades to an empty container instead, since a mock must not turn its own
  defect into a 500. Bounded, not plumbed into an error response — the
  unreachable path does not justify touching the request path.
- `_bounded_int` ignored `multipleOf` while honouring minimum, maximum and both
  exclusive bounds; 9 of 12 sampled paths produced a non-multiple. Values now
  snap onto a multiple inside the bounds, and an unsatisfiable window keeps the
  bounds. A fractional `multipleOf` is still ignored, as documented.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* docs(mock-provider): correct the reason fractional multipleOf is dropped

The docstring claimed honouring it would mean returning a non-integer from an
integer schema. That is wrong: 3 is an integer and a multiple of 1.5. The real
reason is that it needs exact-decimal arithmetic to keep float drift from
deciding validity, and no Honcho response model emits multipleOf at all.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Eugene Eisenstein 2026-09-03 14:20:47 -04:00 committed by GitHub
parent 7cf865c969
commit afbc517cbc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 1888 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
CONTRIBUTING.md
View File

@ -219,6 +219,45 @@ uv run python -m src.deriver # background worker
Everything Python goes through `uv run`. Redis is optional for local development; without it
caching is simply disabled.
### Running without a model provider
`src/mock_provider/` is a deterministic, OpenAI-compatible endpoint, so you can run the full
stack with no provider account, no API key, and no spend. It answers `/v1/chat/completions`
and `/v1/embeddings` with obviously-synthetic content derived from the request, and the same
request always produces the same response. Run it from the standard image or the repo:
```bash
uv run fastapi run --host 0.0.0.0 --port 8106 src/mock_provider/main.py
```
Then point Honcho at it. All three variables are required:
```bash
export LLM_OPENAI_API_KEY=any-non-empty-string # only truthiness is checked
export LLM_OPENAI_BASE_URL=http://localhost:8106/v1
export EMBEDDING_MODEL_CONFIG__OVERRIDES__BASE_URL=http://localhost:8106/v1
```
The key's *value* is never checked — the mock reads no Authorization header, and Honcho only
tests it for truthiness before building the client (`src/llm/registry.py`). Set the base URL
without it and the client is never constructed, so the base URL is silently ignored. Keep the
value obviously fake, so a module that ever escapes the override 401s rather than spends.
Embeddings resolve through a separate client that reads the base URL only from the per-module
override, so without the third variable your embedding calls go to `api.openai.com` for real.
Do not set any per-module credential override (`..._OVERRIDES__API_KEY` / `API_KEY_ENV`) —
that makes the module ignore the global base URL.
Two things to know:
- **A repo `.env` beats your exported environment.** `src/config.py` calls
`load_dotenv(override=True)` at import, so a stale `.env` silently wins over the variables
above. Set `PYTHON_DOTENV_DISABLED=1` (and `HONCHO_CONFIG_TOML_DISABLED=1` for a local
`config.toml`) when you need the environment to be the only input.
- **Mock embeddings are hash-derived and carry no semantic similarity.** Two paraphrases are as
far apart as two unrelated strings. Recall against this provider must use lexical/full-text
search; anything asserting on vector ranking needs a real embedding provider.
## Making the change
### Branches and commits

1
src/mock_provider/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Deterministic OpenAI-compatible provider for local and CI use."""

214
src/mock_provider/chat.py Normal file
View File

@ -0,0 +1,214 @@
"""OpenAI-compatible ``/chat/completions``, answered without inference."""
from __future__ import annotations
import hashlib
import json
import re
from collections.abc import AsyncIterator
from typing import Any
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from src.mock_provider.coerce import as_dict, as_str
from src.mock_provider.schema_gen import generate
from src.mock_provider.schemas import ChatCompletionRequest, ChatMessage
router = APIRouter(tags=["mock-provider"])
# Honcho's json_object mode injects the schema into the prompt text rather than
# into response_format (see _apply_json_object_mode in the OpenAI backend), so
# the only machine-readable copy of the schema is inside a message.
_SCHEMA_HINT = re.compile(r"schema:\s*(\{)", re.IGNORECASE)
def _completion_id(body: ChatCompletionRequest) -> str:
"""Stable id, so a replayed request is byte-identical."""
digest = hashlib.sha256(
body.model_dump_json(exclude_none=True).encode()
).hexdigest()
return f"chatcmpl-mock-{digest[:24]}"
def _extract_balanced_json(text: str, start: int) -> dict[str, Any] | None:
"""Read one balanced ``{...}`` beginning at ``start`` and parse it.
A plain regex cannot do this — a JSON Schema contains nested objects, and
braces inside string literals must not count toward the depth.
"""
depth = 0
in_string = False
escaped = False
for index in range(start, len(text)):
char = text[index]
if in_string:
if escaped:
escaped = False
elif char == "\\":
escaped = True
elif char == '"':
in_string = False
continue
if char == '"':
in_string = True
elif char == "{":
depth += 1
elif char == "}":
depth -= 1
if depth == 0:
try:
parsed = json.loads(text[start : index + 1])
except json.JSONDecodeError:
return None
return as_dict(parsed)
return None
def _schema_from_messages(messages: list[ChatMessage]) -> dict[str, Any] | None:
"""Recover an injected schema from the prompt, for json_object mode."""
for message in reversed(messages):
content = as_str(message.content)
if content is None:
continue
for match in _SCHEMA_HINT.finditer(content):
candidate = _extract_balanced_json(content, match.start(1))
if candidate and ("properties" in candidate or "$defs" in candidate):
return candidate
return None
def _response_content(body: ChatCompletionRequest) -> str:
"""The assistant message body: schema-conforming JSON, or prose."""
response_format = body.response_format
if response_format is not None:
kind = as_str(response_format.get("type"))
if kind == "json_schema":
wrapper = as_dict(response_format.get("json_schema"))
if wrapper is not None:
schema = as_dict(wrapper.get("schema"))
if schema is not None:
return json.dumps(generate(schema))
# A json_schema request whose schema we cannot read must not fall
# through to prose — that is the silent-empty failure this mock
# exists to avoid. An empty object at least parses.
return "{}"
if kind == "json_object":
schema = _schema_from_messages(body.messages)
return json.dumps(generate(schema)) if schema else "{}"
return (
"[mock] This is a synthetic response from Honcho's mock provider. "
"No model was called."
)
def _usage(body: ChatCompletionRequest, content: str) -> dict[str, int]:
"""Rough token accounting, so cost telemetry has plausible numbers."""
prompt_chars = 0
for message in body.messages:
text = as_str(message.content)
if text is not None:
prompt_chars += len(text)
prompt_tokens = max(1, prompt_chars // 4)
completion_tokens = max(1, len(content) // 4)
return {
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens,
}
def _created() -> int:
# Fixed rather than time-based: a mock that changes its output between
# identical calls defeats the point.
return 1577836800 # 2020-01-01T00:00:00Z
async def _stream(
completion_id: str, model: str, content: str, usage: dict[str, int] | None
) -> AsyncIterator[bytes]:
"""Stream ``content``, ending on a usage chunk when ``usage`` is given."""
def chunk(payload: dict[str, Any]) -> bytes:
return f"data: {json.dumps(payload)}\n\n".encode()
base = {
"id": completion_id,
"object": "chat.completion.chunk",
"created": _created(),
"model": model,
}
yield chunk(
{
**base,
"choices": [
{
"index": 0,
"delta": {"role": "assistant", "content": ""},
"finish_reason": None,
}
],
}
)
yield chunk(
{
**base,
"choices": [
{"index": 0, "delta": {"content": content}, "finish_reason": None}
],
}
)
yield chunk(
{
**base,
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
}
)
# The usage chunk is conditional: the real API emits it only when
# stream_options.include_usage is set, and ends the stream on it — so it
# must come last and must carry choices: []. Honcho's own backend always
# asks for it (_build_params in the OpenAI backend), but a caller that does
# not must not receive a chunk it never requested.
if usage is not None:
yield chunk({**base, "choices": [], "usage": usage})
yield b"data: [DONE]\n\n"
@router.post("/chat/completions")
async def chat_completions(body: ChatCompletionRequest) -> Any:
model = body.model or "mock-model"
content = _response_content(body)
usage = _usage(body, content)
completion_id = _completion_id(body)
if body.stream:
include_usage = (
body.stream_options is not None and body.stream_options.include_usage
)
return StreamingResponse(
_stream(completion_id, model, content, usage if include_usage else None),
media_type="text/event-stream",
)
return {
"id": completion_id,
"object": "chat.completion",
"created": _created(),
"model": model,
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": content,
"refusal": None,
"tool_calls": None,
},
"logprobs": None,
"finish_reason": "stop",
}
],
"usage": usage,
}

34
src/mock_provider/coerce.py Normal file
View File

@ -0,0 +1,34 @@
"""Typed narrowing for values decoded from JSON.
``isinstance(value, dict)`` on an ``Any`` narrows to ``dict[Unknown, Unknown]``,
which spreads unknown types through everything downstream. These helpers narrow
and pin the element types in one step.
"""
from __future__ import annotations
from typing import Any, cast
def as_dict(value: object) -> dict[str, Any] | None:
"""The value as a JSON object, or None if it is not one."""
return cast("dict[str, Any]", value) if isinstance(value, dict) else None
def as_list(value: object) -> list[Any] | None:
"""The value as a JSON array, or None if it is not one."""
return cast("list[Any]", value) if isinstance(value, list) else None
def as_str(value: object) -> str | None:
"""The value as a JSON string, or None if it is not one."""
return value if isinstance(value, str) else None
def as_int(value: object) -> int | None:
"""The value as a JSON integer, or None if it is not one.
``bool`` is excluded: it is an ``int`` subclass, and a JSON ``true`` reaching
a size or dimension field is a malformed request, not the number one.
"""
return value if isinstance(value, int) and not isinstance(value, bool) else None

97
src/mock_provider/embeddings.py Normal file
View File

@ -0,0 +1,97 @@
"""OpenAI-compatible ``/embeddings``, answered from a content hash."""
from __future__ import annotations
import base64
import hashlib
import struct
from typing import Any
from fastapi import APIRouter
from src.mock_provider.schemas import EmbeddingsRequest
router = APIRouter(tags=["mock-provider"])
# Honcho's default. EmbeddingClient._validate_embedding_dimensions raises when a
# vector comes back at the wrong width, and validate_embedding_schema refuses to
# boot when the width disagrees with the pgvector column, so the request's own
# `dimensions` is honoured whenever it is present.
DEFAULT_DIMENSIONS = 1536
def content_to_embedding(content: str, dimensions: int) -> list[float]:
"""A deterministic vector for ``content``.
Identical input yields an identical vector, and different inputs differ —
which is what deduplication logic needs. It carries no semantic similarity:
two paraphrases are as far apart as two unrelated strings. Anything
asserting on ranking quality must not use this provider.
Mirrors ``_content_to_embedding`` in tests/conftest.py.
"""
digest = hashlib.sha256(content.encode()).digest()
return [(digest[i % len(digest)] / 255.0) * 2 - 1 for i in range(dimensions)]
def _encode_base64(vector: list[float]) -> str:
"""Little-endian float32, which is what the OpenAI SDK decodes."""
return base64.b64encode(struct.pack(f"<{len(vector)}f", *vector)).decode()
def _normalize_input(
raw: str | list[str] | list[int] | list[list[int]] | None,
) -> list[str]:
"""Flatten the request input into one string per embedding to return.
Token-array inputs are rendered back to a stable string rather than
rejected — the vector only has to be deterministic, not meaningful.
"""
if raw is None:
return []
if isinstance(raw, str):
return [raw]
# A flat list of ints is one tokenized input, not many single-token ones.
if raw and all(isinstance(item, int) for item in raw):
return [",".join(str(item) for item in raw)]
texts: list[str] = []
for item in raw:
if isinstance(item, str):
texts.append(item)
elif isinstance(item, list):
texts.append(",".join(str(part) for part in item))
else:
texts.append(str(item))
return texts
@router.post("/embeddings")
async def embeddings(body: EmbeddingsRequest) -> Any:
texts = _normalize_input(body.input)
# A non-positive width is rejected by the request model, so absent is the
# only case left to fill in.
dimensions = body.dimensions if body.dimensions is not None else DEFAULT_DIMENSIONS
data: list[dict[str, Any]] = []
for index, text in enumerate(texts):
vector = content_to_embedding(text, dimensions)
data.append(
{
"object": "embedding",
"index": index,
"embedding": (
vector
if body.encoding_format == "float"
else _encode_base64(vector)
),
}
)
prompt_tokens = max(1, sum(len(text) for text in texts) // 4)
return {
"object": "list",
"data": data,
"model": body.model or "mock-embedding",
"usage": {"prompt_tokens": prompt_tokens, "total_tokens": prompt_tokens},
}

93
src/mock_provider/main.py Normal file
View File

@ -0,0 +1,93 @@
"""A deterministic, OpenAI-compatible provider for local and CI use.
Lets Honcho run with no model provider, no API key, and no spend. It answers
``/v1/chat/completions`` and ``/v1/embeddings`` with obviously-synthetic content
derived from the request, so the same request always produces the same response.
Runs as its own service from the standard Honcho image:
fastapi run --host 0.0.0.0 src/mock_provider/main.py
Point Honcho at it with three variables — all three are required:
LLM_OPENAI_API_KEY=any-non-empty-string # only truthiness is checked; value ignored
LLM_OPENAI_BASE_URL=http://mock-provider:8000/v1
EMBEDDING_MODEL_CONFIG__OVERRIDES__BASE_URL=http://mock-provider:8000/v1
The key's *value* is never checked — not by this mock, which reads no
Authorization header, and not by Honcho, which only tests it for truthiness
before constructing the client (``src/llm/registry.py``). Set the base URL
without it and the client is never built, so the base URL is silently ignored.
Keep the value obviously fake: if a module ever escapes the base-URL override it
then 401s against the real provider instead of spending.
Embeddings resolve through a separate client that reads the base URL only from
the per-module override, so without the third variable embedding calls go to
api.openai.com for real. Do not set any per-module credential override
(``..._OVERRIDES__API_KEY`` / ``API_KEY_ENV``) — that makes the module ignore the
global base URL.
Embeddings are hash-derived and carry no semantic similarity. Recall assertions
against this provider must use lexical/full-text search, not vector ranking.
"""
from __future__ import annotations
from typing import Any
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from src.mock_provider import chat, embeddings
app = FastAPI(
title="Honcho Mock Provider",
description="Deterministic OpenAI-compatible endpoint for local and CI use.",
version="1.0.0",
)
@app.exception_handler(RequestValidationError)
async def openai_error_response(
_request: Request, exc: RequestValidationError
) -> JSONResponse:
"""Answer a malformed request the way the real API does.
FastAPI's default is a 422 carrying its own error shape. Mid-run that reads
as a Honcho bug rather than a bad request, and it is not what an OpenAI
client expects — the real API returns 400 with an ``error`` envelope, so
that is what a faithful mock returns.
"""
return JSONResponse(
status_code=400,
content={
"error": {
"message": f"Invalid request: {exc.errors()}",
"type": "invalid_request_error",
"param": None,
"code": None,
}
},
)
# Mounted at both prefixes so the base URL works with or without /v1.
for _router in (chat.router, embeddings.router):
app.include_router(_router, prefix="/v1")
app.include_router(_router)
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok", "provider": "mock"}
@app.get("/{path:path}")
async def catch_all(path: str) -> dict[str, Any]:
"""Answer any other GET, so a bare ``/`` works as a container healthcheck.
Deliberately GET-only: an unimplemented POST returns 405 rather than a
plausible-looking 200, so a missing endpoint fails loudly.
"""
return {"object": "mock", "path": path, "detail": "mock provider placeholder"}

373
src/mock_provider/schema_gen.py Normal file
View File

@ -0,0 +1,373 @@
"""Generate a conforming instance from a JSON Schema.
The deriver is a structured-output caller: it sends a schema and parses the
reply back into a Pydantic model. A mock that answers with prose does not fail
loudly — ``repair_response_model_json`` swallows the error and hands back an
empty ``PromptRepresentation``, which reads as "the deriver found nothing"
rather than "the mock is wrong". So generation is driven by the schema that was
actually sent, ``$ref`` indirection and all.
Values are derived from a hash of the property path, so the same schema always
produces the same instance and two different fields never collide.
Not reused from ``src/utils/schema_conversion.py``, despite the overlapping
``$ref``/``$defs`` handling, because that module answers a different question and
does so under an incompatible contract. It builds a Pydantic *model class* where
this needs an *instance*; it raises by design (conversion doubles as validation,
surfaced to callers as a 422) where a mock must degrade rather than turn its own
defect into a 500; and it rejects both ``allOf`` and recursive ``$ref`` — the
latter being ordinary input here, since reasoning-tree schemas nest premises
inside conclusions.
"""
from __future__ import annotations
import hashlib
from typing import Any
from src.mock_provider.coerce import as_dict, as_int, as_list, as_str
# Depth cap for self-referential schemas. Reasoning-tree models nest premises
# inside conclusions, so a $ref cycle is normal input, not a malformed schema.
MAX_DEPTH = 6
# Absolute cap. Past MAX_DEPTH a cycle is expected to terminate on a `default`
# or a nullable/optional branch; a required, non-nullable self-reference has
# neither and would recurse until Python raises RecursionError. Degrading to an
# empty container may violate the schema, but a mock must not turn its own
# defect into a 500. Set well clear of MAX_DEPTH so no schema that terminates
# on its own ever reaches it.
HARD_MAX_DEPTH = MAX_DEPTH * 4
_WORDS = (
"synthetic",
"placeholder",
"mock",
"sample",
"fixture",
"stub",
"generated",
"example",
"inert",
"dummy",
)
def _seed(path: str) -> int:
return int.from_bytes(hashlib.sha256(path.encode()).digest()[:8], "big")
def _phrase(path: str, words: int = 6) -> str:
"""An obviously-synthetic sentence, stable for a given path."""
seed = _seed(path)
picked = [_WORDS[(seed >> (i * 5)) % len(_WORDS)] for i in range(words)]
return f"[mock] {' '.join(picked)}"
def _resolve(schema: dict[str, Any], root: dict[str, Any]) -> dict[str, Any]:
"""Follow a local ``$ref`` chain to the schema it points at.
Only local refs are supported: the mock never fetches over the network, and
Pydantic's ``model_json_schema()`` only ever emits ``#/$defs/...``.
"""
seen: set[str] = set()
current = schema
while "$ref" in current:
ref = as_str(current["$ref"])
if ref is None or not ref.startswith("#/") or ref in seen:
return {}
seen.add(ref)
target: dict[str, Any] | None = root
for part in ref[2:].split("/"):
if target is None or part not in target:
return {}
target = as_dict(target[part])
if target is None:
return {}
current = target
return current
def _merge_all_of(schema: dict[str, Any], root: dict[str, Any]) -> dict[str, Any]:
"""Flatten ``allOf`` into the parent so one pass can read properties off it."""
branches = as_list(schema.get("allOf"))
if branches is None:
return schema
merged: dict[str, Any] = {k: v for k, v in schema.items() if k != "allOf"}
for branch in branches:
resolved_branch = as_dict(branch)
if resolved_branch is None:
continue
resolved = _resolve(resolved_branch, root)
for key, value in resolved.items():
if key == "properties":
properties = as_dict(value)
if properties is not None:
# First branch to define a property wins. Strictly, `allOf`
# requires every branch's constraints to apply, so a schema
# splitting `minimum` and `maximum` for one property across
# two branches generates a value satisfying only one of
# them. Not merged recursively because Pydantic's `allOf` is
# always a $ref plus sibling annotations — it never repeats
# a property key, let alone with conflicting constraints.
existing = as_dict(merged.get("properties")) or {}
merged["properties"] = {**properties, **existing}
continue
if key == "required":
required = as_list(value)
if required is not None:
previous = as_list(merged.get("required")) or []
merged["required"] = list({*previous, *required})
continue
merged.setdefault(key, value)
return merged
def _infer_type(schema: dict[str, Any]) -> str:
"""Best-effort type when the schema omits an explicit ``type``."""
declared = schema.get("type")
if (name := as_str(declared)) is not None:
return name
if (names := as_list(declared)) is not None:
# Nullable unions arrive as ["string", "null"]; prefer the real type.
for candidate in names:
if (candidate_name := as_str(candidate)) and candidate_name != "null":
return candidate_name
return "null"
if "properties" in schema:
return "object"
if "items" in schema:
return "array"
return "string"
def generate(schema: dict[str, Any], root: dict[str, Any] | None = None) -> Any:
"""Build a value satisfying ``schema``.
``root`` carries the document that ``$ref`` resolves against; it defaults to
``schema`` itself, which is the shape Pydantic emits.
"""
return _generate(schema, root if root is not None else schema, "$", 0)
def _generate(
schema: dict[str, Any], root: dict[str, Any], path: str, depth: int
) -> Any:
resolved = _merge_all_of(_resolve(schema, root), root)
if "const" in resolved:
return resolved["const"]
enum = as_list(resolved.get("enum"))
if enum:
return enum[_seed(path) % len(enum)]
if depth >= MAX_DEPTH and "default" in resolved:
return resolved["default"]
# `oneOf` is treated as `anyOf`: a branch is picked without checking that
# the result matches only that one. A `oneOf` whose branches overlap can
# therefore yield a value matching several, which `oneOf` forbids. Enforcing
# the cardinality needs a full JSON Schema validator to test the candidate
# against every branch, and Pydantic emits `anyOf` for unions — never
# `oneOf` — so nothing Honcho sends reaches the distinction.
for key in ("anyOf", "oneOf"):
branches = as_list(resolved.get(key))
if branches:
return _generate(_pick_branch(branches, root, depth), root, path, depth)
kind = _infer_type(resolved)
# Only the two recursive kinds need the absolute cap; scalars terminate.
if kind == "object":
if depth >= HARD_MAX_DEPTH:
return {}
return _generate_object(resolved, root, path, depth)
if kind == "array":
if depth >= HARD_MAX_DEPTH:
return []
return _generate_array(resolved, root, path, depth)
if kind == "integer":
return _bounded_int(resolved, path)
if kind == "number":
return float(_bounded_int(resolved, path))
if kind == "boolean":
return _seed(path) % 2 == 0
if kind == "null":
return None
return _generate_string(resolved, path)
def _pick_branch(
branches: list[Any], root: dict[str, Any], depth: int
) -> dict[str, Any]:
"""Choose a union member, preferring a non-null one.
Past the depth cap the order flips: a nullable recursive field terminates on
``null`` instead of nesting another level.
"""
resolved: list[dict[str, Any]] = []
for branch in branches:
branch_dict = as_dict(branch)
if branch_dict is not None:
resolved.append(_resolve(branch_dict, root))
if not resolved:
return {}
if depth >= MAX_DEPTH:
nulls = [b for b in resolved if _infer_type(b) == "null"]
if nulls:
return nulls[0]
non_null = [b for b in resolved if _infer_type(b) != "null"]
return non_null[0] if non_null else resolved[0]
def _generate_object(
schema: dict[str, Any], root: dict[str, Any], path: str, depth: int
) -> dict[str, Any]:
properties = as_dict(schema.get("properties"))
if properties is None:
return {}
# OpenAI structured outputs run in strict mode, where every property is
# required. Emitting the full property set satisfies both strict and loose
# schemas, so `required` is only consulted to decide what to drop once the
# depth cap has been hit.
declared_required = as_list(schema.get("required"))
required: set[str] = (
{name for name in (as_str(item) for item in declared_required) if name}
if declared_required is not None
else set(properties)
)
result: dict[str, Any] = {}
for name, subschema in properties.items():
if depth >= MAX_DEPTH and name not in required:
continue
child = as_dict(subschema)
if child is None:
continue
result[name] = _generate(child, root, f"{path}.{name}", depth + 1)
return result
def _generate_array(
schema: dict[str, Any], root: dict[str, Any], path: str, depth: int
) -> list[Any]:
# A fixed-length tuple is `prefixItems` with no `items`, which is what
# Pydantic emits for `tuple[str, int]`. Reading only `items` would return []
# for it and fail the minItems/maxItems the same schema carries.
prefix: list[Any] = []
prefix_items = as_list(schema.get("prefixItems"))
if prefix_items is not None:
for index, entry in enumerate(prefix_items):
child = as_dict(entry)
if child is not None:
prefix.append(_generate(child, root, f"{path}[{index}]", depth + 1))
items = as_dict(schema.get("items"))
min_items = as_int(schema.get("minItems"))
max_items = as_int(schema.get("maxItems"))
count = 2
if min_items is not None:
count = max(count, min_items)
if max_items is not None:
count = min(count, max_items)
if depth >= MAX_DEPTH:
count = min_items or 0
# `items` describes the positions after the prefix, so only the shortfall is
# filled. With `items` absent those positions are unconstrained rather than
# disallowed: an empty schema stands in, and the target drops to whatever
# minItems demands, so a bare `{"type": "array"}` still generates nothing.
trailing = items if items is not None else {}
target = count if items is not None else min(count, min_items or 0)
return prefix + [
_generate(trailing, root, f"{path}[{len(prefix) + i}]", depth + 1)
for i in range(max(0, target - len(prefix)))
]
def _generate_string(schema: dict[str, Any], path: str) -> str:
fmt = as_str(schema.get("format"))
if fmt == "date-time":
return "2020-01-01T00:00:00Z"
if fmt == "date":
return "2020-01-01"
if fmt == "uuid":
stem = hashlib.sha256(path.encode()).hexdigest()[:8]
return f"{stem}-0000-4000-8000-000000000000"
if fmt in ("uri", "url"):
return "https://mock.invalid/placeholder"
if fmt == "email":
return "placeholder@mock.invalid"
# `pattern` is not honoured: this phrase fails any regex narrower than it,
# so a pattern-constrained string generates a value its own schema rejects.
# Satisfying an arbitrary regex needs a generator library, and no Honcho
# response model carries a `pattern` — the only ones in the codebase are on
# API request models, which are never sent as a response_format.
value = _phrase(path)
min_length = as_int(schema.get("minLength"))
max_length = as_int(schema.get("maxLength"))
if min_length is not None and len(value) < min_length:
value = value.ljust(min_length, "x")
if max_length is not None and len(value) > max_length:
value = value[:max_length]
return value
def _bounded_int(schema: dict[str, Any], path: str) -> int:
low = as_int(schema.get("minimum"))
if (
low is None
and (exclusive := as_int(schema.get("exclusiveMinimum"))) is not None
):
low = exclusive + 1
high = as_int(schema.get("maximum"))
if (
high is None
and (exclusive := as_int(schema.get("exclusiveMaximum"))) is not None
):
high = exclusive - 1
if low is not None and high is not None:
span = high - low
value = low + (_seed(path) % (span + 1) if span > 0 else 0)
elif low is not None:
value = low + (_seed(path) % 8)
elif high is not None:
value = high - (_seed(path) % 8)
else:
value = _seed(path) % 100
return _snap_to_multiple(value, as_int(schema.get("multipleOf")), low, high)
def _snap_to_multiple(
value: int, multiple: int | None, low: int | None, high: int | None
) -> int:
"""Move ``value`` onto a multiple of ``multiple``, staying within bounds.
Integer ``multipleOf`` only. The spec allows a fractional one, and an
integer can satisfy it (3 is a multiple of 1.5), but honouring it needs
exact-decimal arithmetic to avoid float drift deciding validity. ``as_int``
rejects it, so the constraint is dropped rather than approximated — no
Honcho response model emits ``multipleOf`` at all.
"""
if multiple is None or multiple <= 0:
return value
# Floor division, so a negative value snaps down to the next multiple below.
snapped = (value // multiple) * multiple
if low is not None and snapped < low:
snapped = -(-low // multiple) * multiple # smallest multiple >= low
if high is not None and snapped > high:
snapped = (high // multiple) * multiple # largest multiple <= high
# No multiple exists in the window, so the schema is unsatisfiable. An
# in-range value breaks the constraint the caller is less likely to check.
if (low is not None and snapped < low) or (high is not None and snapped > high):
return value
return snapped

71
src/mock_provider/schemas.py Normal file
View File

@ -0,0 +1,71 @@
"""Request models for the mock provider's OpenAI-compatible endpoints.
Validating the request envelope rather than hand-coercing it makes the mock
behave like the thing it mocks: real OpenAI answers a malformed request with a
400 and an error envelope, and ``openai_error_response`` in ``main`` turns
Pydantic's failure into exactly that.
Two deliberate choices:
- ``extra="allow"`` on every model, and every field optional. Validation should
fire on a wrong *type* (a string where a list belongs), never on a field this
mock has not heard of — otherwise a new upstream parameter turns a working
setup into a hard failure.
- Open-ended payloads stay ``dict[str, Any]``. ``response_format`` carries an
arbitrary caller-supplied JSON Schema, so only its envelope is worth typing;
``schema_gen`` walks the rest.
"""
from __future__ import annotations
from typing import Annotated, Any, ClassVar, Literal
from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictInt
class MockRequest(BaseModel):
"""Permissive base: unknown fields pass through untouched."""
model_config: ClassVar[ConfigDict] = ConfigDict(extra="allow")
class ChatMessage(MockRequest):
role: str | None = None
# Multimodal requests send a list of content parts rather than a string, so
# this cannot narrow further.
content: Any = None
class StreamOptions(MockRequest):
# Typed rather than left as a dict because the usage chunk is conditional on
# it. StrictBool for the same reason `dimensions` is StrictInt: plain `bool`
# coerces "yes"/"on"/"true"/"1", so a string would quietly decide the shape
# of the stream instead of failing the way the real API does.
include_usage: StrictBool = False
class ChatCompletionRequest(MockRequest):
model: str | None = None
messages: list[ChatMessage] = []
response_format: dict[str, Any] | None = None
tools: list[dict[str, Any]] | None = None
# StrictBool because this one field decides between two response *shapes* —
# a JSON body or an SSE stream — so coercing a string here is the difference
# between a working client and one that hangs waiting for events.
stream: StrictBool = False
stream_options: StreamOptions | None = None
class EmbeddingsRequest(MockRequest):
# Every input shape the OpenAI embeddings API accepts. Pydantic's smart
# union keeps list[str] and list[int] apart instead of coercing one to the
# other.
input: str | list[str] | list[int] | list[list[int]] | None = None
model: str | None = None
# StrictInt because bool is an int subclass: a JSON `true` here would
# otherwise silently become a one-dimensional vector. gt=0 because the real
# API rejects a non-positive width, and substituting the default instead
# would answer a bad request with a plausible-looking vector.
dimensions: Annotated[StrictInt, Field(gt=0)] | None = None
# The SDK omits this only when it wants base64, so absent means base64.
encoding_format: Literal["float", "base64"] = "base64"

3
tests/conftest.py
View File

@ -91,6 +91,9 @@ _RUNTIME_MOCK_TEST_BLOCKLIST_PREFIXES = (
# Pure JWT scope tests — operate on src.security directly, no DB needed.
"tests/test_security.py",
"tests/test_generate_jwt_script.py",
# The mock provider is a standalone ASGI app with no database or LLM of its
# own; the runtime mocks would patch the very seams it exists to replace.
"tests/mock_provider/",
)
_LIVE_LLM_MARKER = "live_llm"

221
tests/mock_provider/test_honcho_contract.py Normal file
View File

@ -0,0 +1,221 @@
"""Drive Honcho's real provider clients against the mock over ASGI.
The unit tests assert the mock's own output. These assert the hop that actually
matters: ``OpenAIBackend`` and ``EmbeddingClient`` — the production classes,
unpatched — talking to the mock through the genuine OpenAI SDK, including the
``strict: true`` json_schema transform that ``chat.completions.parse()`` applies
on the way out and the Pydantic validation it applies on the way back.
"""
from __future__ import annotations
from typing import Any
import httpx
import pytest
from openai import AsyncOpenAI
from src.config import EmbeddingModelConfig
from src.embedding_client import _EmbeddingClient # pyright: ignore[reportPrivateUsage]
from src.llm.backends.openai import OpenAIBackend
from src.mock_provider.embeddings import content_to_embedding
from src.mock_provider.main import app
from src.utils.representation import PromptRepresentation
MESSAGES: list[dict[str, Any]] = [
{"role": "user", "content": "I switched the service from pip to uv last week."}
]
@pytest.fixture
def openai_client() -> AsyncOpenAI:
return AsyncOpenAI(
api_key="sandbox",
base_url="http://mock-provider.invalid/v1",
http_client=httpx.AsyncClient(transport=httpx.ASGITransport(app=app)),
)
@pytest.mark.asyncio
async def test_backend_parses_the_deriver_response_model(
openai_client: AsyncOpenAI,
) -> None:
"""The production path: parse() with a Pydantic response_format."""
backend = OpenAIBackend(openai_client)
result = await backend.complete(
model="mock-model",
messages=MESSAGES,
max_tokens=512,
response_format=PromptRepresentation,
)
assert isinstance(result.content, PromptRepresentation)
# An empty explicit list is what a prose-answering mock silently produces,
# so it is the specific thing worth asserting against.
assert result.content.explicit
assert result.output_tokens > 0
@pytest.mark.asyncio
async def test_backend_json_object_mode_recovers_the_schema(
openai_client: AsyncOpenAI,
) -> None:
"""json_object mode carries the schema in the prompt, not response_format."""
backend = OpenAIBackend(openai_client)
result = await backend.complete(
model="mock-model",
messages=MESSAGES,
max_tokens=512,
response_format=PromptRepresentation,
extra_params={"structured_output_mode": "json_object"},
)
assert isinstance(result.content, PromptRepresentation)
assert result.content.explicit
@pytest.mark.asyncio
async def test_backend_with_tools_uses_json_schema_and_still_parses(
openai_client: AsyncOpenAI,
) -> None:
"""Non-strict tools force create() + explicit json_schema instead of parse()."""
backend = OpenAIBackend(openai_client)
result = await backend.complete(
model="mock-model",
messages=MESSAGES,
max_tokens=512,
response_format=PromptRepresentation,
tools=[
{
"type": "function",
"function": {
"name": "search_memory",
"description": "Search memory",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
},
},
}
],
)
assert isinstance(result.content, PromptRepresentation)
assert result.content.explicit
@pytest.mark.asyncio
async def test_backend_plain_completion(openai_client: AsyncOpenAI) -> None:
backend = OpenAIBackend(openai_client)
result = await backend.complete(
model="mock-model", messages=MESSAGES, max_tokens=128
)
assert isinstance(result.content, str)
assert "[mock]" in result.content
assert result.finish_reason == "stop"
@pytest.mark.asyncio
async def test_backend_stream_yields_content_then_a_usage_terminator(
openai_client: AsyncOpenAI,
) -> None:
backend = OpenAIBackend(openai_client)
chunks = [
chunk
async for chunk in backend.stream(
model="mock-model", messages=MESSAGES, max_tokens=128
)
]
assert "[mock]" in "".join(chunk.content or "" for chunk in chunks)
terminator = chunks[-1]
assert terminator.is_done
assert terminator.finish_reason == "stop"
# None here means the stream ended without a usage chunk, which is the
# failure mode when stream_options.include_usage goes unanswered.
assert terminator.output_tokens is not None
assert terminator.output_tokens > 0
def _embedding_client(dimensions: int, encoding_format: str) -> _EmbeddingClient:
# The public EmbeddingClient is a settings-driven singleton wrapper; the
# transport behaviour under test lives on the implementation it wraps.
return _EmbeddingClient(
EmbeddingModelConfig(
model="text-embedding-3-small",
transport="openai",
api_key="sandbox",
base_url="http://mock-provider.invalid/v1",
),
vector_dimensions=dimensions,
max_input_tokens=8192,
max_tokens_per_request=300000,
send_dimensions=True,
encoding_format=encoding_format, # pyright: ignore[reportArgumentType]
)
@pytest.fixture(autouse=True)
def _route_embedding_client_over_asgi( # pyright: ignore[reportUnusedFunction]
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Give the embedding client's AsyncOpenAI an ASGI transport.
EmbeddingClient builds its own client internally, so the transport has to be
injected at construction rather than passed in.
"""
original = AsyncOpenAI.__init__
def patched(self: AsyncOpenAI, *args: Any, **kwargs: Any) -> None:
kwargs.setdefault(
"http_client",
httpx.AsyncClient(transport=httpx.ASGITransport(app=app)),
)
original(self, *args, **kwargs)
monkeypatch.setattr(AsyncOpenAI, "__init__", patched)
@pytest.mark.asyncio
@pytest.mark.parametrize("encoding_format", ["float", "base64"])
async def test_embedding_client_round_trip(encoding_format: str) -> None:
"""Covers both wire encodings; base64 is what the SDK uses by default."""
client = _embedding_client(1536, encoding_format)
vector = await client.embed("I switched the service from pip to uv.")
# _validate_embedding_dimensions raises on a width mismatch, so reaching
# here already proves the width is right; assert the values too.
assert len(vector) == 1536
assert vector == pytest.approx( # pyright: ignore[reportUnknownMemberType]
content_to_embedding("I switched the service from pip to uv.", 1536),
abs=1e-6,
)
@pytest.mark.asyncio
async def test_embedding_client_honours_a_non_default_dimension() -> None:
"""send_dimensions=True forwards `dimensions`; the mock must obey it."""
client = _embedding_client(256, "float")
assert len(await client.embed("hello")) == 256
@pytest.mark.asyncio
async def test_embedding_client_batches() -> None:
"""_validate_embedding_count rejects a mismatched count."""
client = _embedding_client(1536, "float")
texts = [f"observation number {index}" for index in range(12)]
vectors = await client.simple_batch_embed(texts)
assert len(vectors) == len(texts)
assert all(len(vector) == 1536 for vector in vectors)
assert len({tuple(vector) for vector in vectors}) == len(texts)

742
tests/mock_provider/test_mock_provider.py Normal file
View File

@ -0,0 +1,742 @@
"""Contract tests for the mock provider.
The failure this guards against is silent: when the mock answers a structured
request with something the deriver cannot parse, ``repair_response_model_json``
falls back to an empty ``PromptRepresentation`` and the run looks like "the
deriver found nothing" rather than "the mock is broken". So the assertions here
are about parseability against real Honcho models, not about response shape.
"""
from __future__ import annotations
import base64
import hashlib
import json
import struct
from collections.abc import Callable
from typing import Any
import pytest
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field
from src.mock_provider.coerce import as_dict
from src.mock_provider.embeddings import content_to_embedding
from src.mock_provider.main import app
from src.mock_provider.schema_gen import HARD_MAX_DEPTH, MAX_DEPTH, generate
from src.utils.representation import PromptRepresentation
# A $ref/$defs schema, which is what Pydantic emits for any nested model and the
# indirection a naive generator silently drops.
PROBE_SCHEMA: dict[str, Any] = {
"$defs": {
"Item": {
"type": "object",
"properties": {
"name": {"type": "string"},
"count": {"type": "integer", "minimum": 1, "maximum": 5},
},
"required": ["name", "count"],
}
},
"type": "object",
"properties": {
"label": {"type": "string"},
"items": {"type": "array", "items": {"$ref": "#/$defs/Item"}},
},
"required": ["label", "items"],
}
class ProbeItem(BaseModel):
name: str
count: int = Field(ge=1, le=5)
class Probe(BaseModel):
label: str
items: list[ProbeItem]
@pytest.fixture
def client() -> TestClient:
return TestClient(app)
def _post_chat(client: TestClient, **payload: Any) -> dict[str, Any]:
payload.setdefault("model", "mock-model")
payload.setdefault("messages", [{"role": "user", "content": "hello"}])
response = client.post("/v1/chat/completions", json=payload)
assert response.status_code == 200, response.text
return response.json()
def _json_schema_format(schema: dict[str, Any], name: str) -> dict[str, Any]:
return {
"type": "json_schema",
"json_schema": {"name": name, "schema": schema, "strict": True},
}
# --- structured output ------------------------------------------------------
def test_json_schema_request_round_trips_into_its_pydantic_model(
client: TestClient,
) -> None:
body = _post_chat(
client, response_format=_json_schema_format(PROBE_SCHEMA, "Probe")
)
content = body["choices"][0]["message"]["content"]
probe = Probe.model_validate_json(content)
assert probe.label
assert probe.items, "$ref array must not come back empty"
assert all(1 <= item.count <= 5 for item in probe.items)
def test_deriver_response_model_round_trips() -> None:
"""The real model the deriver parses, not a stand-in."""
schema = PromptRepresentation.model_json_schema()
content = json.dumps(generate(schema))
representation = PromptRepresentation.model_validate_json(content)
assert representation.explicit, (
"an empty explicit list is exactly the silent failure this mock avoids"
)
def test_json_schema_response_is_never_prose(client: TestClient) -> None:
body = _post_chat(
client, response_format=_json_schema_format(PROBE_SCHEMA, "Probe")
)
json.loads(body["choices"][0]["message"]["content"])
def test_unreadable_json_schema_still_returns_parseable_json(
client: TestClient,
) -> None:
body = _post_chat(
client,
response_format={"type": "json_schema", "json_schema": {"name": "Broken"}},
)
assert json.loads(body["choices"][0]["message"]["content"]) == {}
def test_json_object_mode_recovers_the_schema_from_the_prompt(
client: TestClient,
) -> None:
"""json_object mode puts the schema in the prompt, not in response_format."""
body = _post_chat(
client,
messages=[
{"role": "user", "content": "Extract facts."},
{
"role": "user",
"content": "Respond with valid JSON matching this schema:\n"
+ json.dumps(PROBE_SCHEMA),
},
],
response_format={"type": "json_object"},
)
Probe.model_validate_json(body["choices"][0]["message"]["content"])
def test_json_object_mode_without_a_schema_returns_an_empty_object(
client: TestClient,
) -> None:
body = _post_chat(client, response_format={"type": "json_object"})
assert json.loads(body["choices"][0]["message"]["content"]) == {}
def test_plain_request_returns_prose(client: TestClient) -> None:
body = _post_chat(client)
content = body["choices"][0]["message"]["content"]
assert "[mock]" in content
with pytest.raises(json.JSONDecodeError):
json.loads(content)
def test_tools_request_does_not_emit_tool_calls(client: TestClient) -> None:
"""The tool loop must terminate; a mock that calls tools would spin."""
body = _post_chat(
client,
tools=[
{
"type": "function",
"function": {"name": "search_memory", "parameters": {}},
}
],
)
assert body["choices"][0]["message"]["tool_calls"] is None
assert body["choices"][0]["finish_reason"] == "stop"
def test_identical_requests_are_byte_identical(client: TestClient) -> None:
payload: dict[str, Any] = {
"model": "mock-model",
"messages": [{"role": "user", "content": "determinism"}],
"response_format": _json_schema_format(PROBE_SCHEMA, "Probe"),
}
first = client.post("/v1/chat/completions", json=payload).json()
second = client.post("/v1/chat/completions", json=payload).json()
assert first == second
def test_usage_is_reported(client: TestClient) -> None:
body = _post_chat(client)
usage = body["usage"]
assert usage["total_tokens"] == usage["prompt_tokens"] + usage["completion_tokens"]
assert usage["completion_tokens"] > 0
# --- schema generation edge cases -------------------------------------------
def test_recursive_schema_terminates() -> None:
"""Reasoning trees nest premises inside conclusions, so cycles are normal."""
schema: dict[str, Any] = {
"$defs": {
"Node": {
"type": "object",
"properties": {
"value": {"type": "string"},
"child": {"anyOf": [{"$ref": "#/$defs/Node"}, {"type": "null"}]},
},
"required": ["value", "child"],
}
},
"$ref": "#/$defs/Node",
}
node: dict[str, Any] | None = generate(schema)
depth = 0
while node is not None and node.get("child") is not None:
node = node["child"]
depth += 1
assert depth < 50, "recursive schema did not terminate"
_SCALAR_CASES: list[tuple[str, dict[str, Any], Callable[[Any], bool]]] = [
("enum", {"type": "string", "enum": ["a", "b"]}, lambda v: v in ("a", "b")),
("const", {"const": 7}, lambda v: v == 7),
("boolean", {"type": "boolean"}, lambda v: isinstance(v, bool)),
("null", {"type": "null"}, lambda v: v is None),
("number", {"type": "number"}, lambda v: isinstance(v, float)),
("nullable-union", {"type": ["string", "null"]}, lambda v: isinstance(v, str)),
("pinned-int", {"type": "integer", "minimum": 3, "maximum": 3}, lambda v: v == 3),
(
"exclusive-bounds",
{"type": "integer", "exclusiveMinimum": 1, "exclusiveMaximum": 3},
lambda v: v == 2,
),
(
"date-time",
{"type": "string", "format": "date-time"},
lambda v: str(v).endswith("Z"),
),
("min-length", {"type": "string", "minLength": 400}, lambda v: len(v) >= 400),
("max-length", {"type": "string", "maxLength": 4}, lambda v: len(v) == 4),
(
"min-items",
{"type": "array", "items": {"type": "string"}, "minItems": 3},
lambda v: len(v) >= 3,
),
(
"max-items",
{"type": "array", "items": {"type": "string"}, "maxItems": 1},
lambda v: len(v) == 1,
),
]
@pytest.mark.parametrize(
("schema", "check"),
[(schema, check) for _, schema, check in _SCALAR_CASES],
ids=[name for name, _, _ in _SCALAR_CASES],
)
def test_scalar_schema_forms(
schema: dict[str, Any], check: Callable[[Any], bool]
) -> None:
assert check(generate(schema))
def test_all_of_is_flattened() -> None:
schema: dict[str, Any] = {
"allOf": [
{
"type": "object",
"properties": {"a": {"type": "string"}},
"required": ["a"],
},
{
"type": "object",
"properties": {"b": {"type": "integer"}},
"required": ["b"],
},
]
}
result = generate(schema)
assert isinstance(result["a"], str)
assert isinstance(result["b"], int)
def test_fixed_tuple_schema_round_trips() -> None:
"""Pydantic emits a fixed tuple as `prefixItems` with no `items`.
Reading only `items` yields [], which fails the minItems the same schema
carries — the silent-empty failure this module exists to avoid.
"""
class Tupled(BaseModel):
pair: tuple[str, int]
schema = Tupled.model_json_schema()
assert "prefixItems" in schema["properties"]["pair"]
result = generate(schema)
assert isinstance(result["pair"], list)
Tupled.model_validate(result)
def test_prefix_items_are_followed_by_homogeneous_items() -> None:
"""A variadic tuple constrains leading positions and the rest by `items`."""
schema: dict[str, Any] = {
"type": "array",
"prefixItems": [{"type": "string"}, {"type": "integer"}],
"items": {"type": "boolean"},
"minItems": 4,
}
result = generate(schema)
assert len(result) == 4
assert isinstance(result[0], str)
assert isinstance(result[1], int)
assert all(isinstance(value, bool) for value in result[2:])
def test_min_items_is_met_when_items_is_omitted() -> None:
"""Absent `items` leaves trailing positions unconstrained, not disallowed."""
schema: dict[str, Any] = {
"type": "array",
"prefixItems": [{"type": "string"}],
"minItems": 3,
}
result = generate(schema)
assert len(result) == 3
assert isinstance(result[0], str)
@pytest.mark.parametrize(
("constraints", "multiple"),
[
({"minimum": 0, "maximum": 100, "multipleOf": 10}, 10),
({"minimum": 7, "maximum": 9, "multipleOf": 4}, 4),
({"minimum": -100, "maximum": 0, "multipleOf": 25}, 25),
({"minimum": 5, "multipleOf": 3}, 3),
({"maximum": -5, "multipleOf": 3}, 3),
({"multipleOf": 6}, 6),
],
)
def test_multiple_of_is_honoured_within_bounds(
constraints: dict[str, Any], multiple: int
) -> None:
"""Path-seeded values land off the multiple unless snapped back onto it."""
low = constraints.get("minimum")
high = constraints.get("maximum")
# Several paths, because a single one can satisfy the constraint by luck.
for index in range(12):
schema: dict[str, Any] = {
"type": "object",
"properties": {f"f{index}": {"type": "integer", **constraints}},
"required": [f"f{index}"],
}
value = generate(schema)[f"f{index}"]
assert value % multiple == 0, f"{value} is not a multiple of {multiple}"
if low is not None:
assert value >= low
if high is not None:
assert value <= high
def test_unsatisfiable_multiple_of_stays_within_bounds() -> None:
"""No multiple of 10 lies in [3, 7], so the bounds win over the multiple."""
schema: dict[str, Any] = {
"type": "integer",
"minimum": 3,
"maximum": 7,
"multipleOf": 10,
}
result = generate(schema)
assert 3 <= result <= 7
def test_required_recursive_ref_terminates_instead_of_overflowing() -> None:
"""A required, non-nullable cycle has no `default` or null branch to stop on.
MAX_DEPTH alone does not save it — `_generate_object` keeps descending into
required properties — so the absolute cap has to.
"""
schema: dict[str, Any] = {
"$defs": {
"Node": {
"type": "object",
"properties": {"child": {"$ref": "#/$defs/Node"}},
"required": ["child"],
}
},
"$ref": "#/$defs/Node",
}
node = as_dict(generate(schema))
depth = 0
# The cap returns {}, so an empty dict is the terminator.
while node:
node = as_dict(node["child"])
depth += 1
assert depth <= HARD_MAX_DEPTH, "absolute depth cap did not hold"
assert depth > MAX_DEPTH, "should descend past the soft cap before stopping"
def test_required_recursive_array_terminates_instead_of_overflowing() -> None:
"""minItems >= 1 keeps `_generate_array` from emptying out at the soft cap."""
schema: dict[str, Any] = {
"$defs": {
"Node": {
"type": "object",
"properties": {
"kids": {
"type": "array",
"items": {"$ref": "#/$defs/Node"},
"minItems": 1,
}
},
"required": ["kids"],
}
},
"$ref": "#/$defs/Node",
}
generate(schema) # must not raise RecursionError
def test_generation_is_stable_across_calls() -> None:
assert generate(PROBE_SCHEMA) == generate(PROBE_SCHEMA)
def test_sibling_fields_of_the_same_type_differ() -> None:
"""Path-seeded, so a schema of identical fields is not all one value."""
schema: dict[str, Any] = {
"type": "object",
"properties": {
"first": {"type": "string"},
"second": {"type": "string"},
},
"required": ["first", "second"],
}
result = generate(schema)
assert result["first"] != result["second"]
# --- embeddings -------------------------------------------------------------
def test_embeddings_default_to_1536_and_are_stable(client: TestClient) -> None:
payload = {
"model": "text-embedding-3-small",
"input": "hello",
"encoding_format": "float",
}
first = client.post("/v1/embeddings", json=payload)
assert first.status_code == 200, first.text
vector = first.json()["data"][0]["embedding"]
assert len(vector) == 1536
assert all(-1.0 <= value <= 1.0 for value in vector)
assert client.post("/v1/embeddings", json=payload).json() == first.json()
def test_embeddings_honour_the_requested_dimension(client: TestClient) -> None:
"""A width mismatch raises in EmbeddingClient and blocks startup."""
response = client.post(
"/v1/embeddings",
json={"input": "hello", "dimensions": 256, "encoding_format": "float"},
)
assert len(response.json()["data"][0]["embedding"]) == 256
def test_different_inputs_give_different_vectors(client: TestClient) -> None:
response = client.post(
"/v1/embeddings",
json={"input": ["alpha", "beta"], "encoding_format": "float"},
)
data = response.json()["data"]
assert len(data) == 2
assert [item["index"] for item in data] == [0, 1]
assert data[0]["embedding"] != data[1]["embedding"]
def test_batch_returns_one_embedding_per_input(client: TestClient) -> None:
"""EmbeddingClient._validate_embedding_count rejects any other count."""
texts = [f"text-{index}" for index in range(17)]
response = client.post(
"/v1/embeddings", json={"input": texts, "encoding_format": "float"}
)
assert len(response.json()["data"]) == len(texts)
def test_base64_is_the_default_encoding_and_decodes_to_the_float_vector(
client: TestClient,
) -> None:
"""The SDK omits encoding_format precisely when it wants base64."""
response = client.post("/v1/embeddings", json={"input": "hello"})
encoded = response.json()["data"][0]["embedding"]
assert isinstance(encoded, str)
raw = base64.b64decode(encoded)
decoded = list(struct.unpack(f"<{len(raw) // 4}f", raw))
assert len(decoded) == 1536
expected = content_to_embedding("hello", 1536)
assert decoded == pytest.approx(expected, abs=1e-6) # pyright: ignore[reportUnknownMemberType]
def test_embedding_matches_the_test_suite_helper() -> None:
"""Kept in step with _content_to_embedding in tests/conftest.py.
Both must derive the same vector from the same text, so a suite that mocks
the embedding client in-process and one that talks to this provider over
HTTP agree on what a given string embeds to.
"""
digest = hashlib.sha256(b"hello").digest()
expected = [(digest[i % len(digest)] / 255.0) * 2 - 1 for i in range(8)]
assert content_to_embedding("hello", 8) == pytest.approx(expected) # pyright: ignore[reportUnknownMemberType]
# --- routing ----------------------------------------------------------------
def test_routes_are_mounted_with_and_without_the_v1_prefix(
client: TestClient,
) -> None:
for path in ("/v1/chat/completions", "/chat/completions"):
response = client.post(
path, json={"model": "m", "messages": [{"role": "user", "content": "x"}]}
)
assert response.status_code == 200, path
def test_unimplemented_post_returns_405_not_a_plausible_200(
client: TestClient,
) -> None:
"""A catch-all POST would make a missing endpoint look like it worked."""
assert client.post("/v1/completions", json={}).status_code == 405
def test_health_and_catch_all_get(client: TestClient) -> None:
assert client.get("/health").json()["status"] == "ok"
assert client.get("/").status_code == 200
# --- request validation -----------------------------------------------------
def test_malformed_body_returns_an_openai_error_envelope(client: TestClient) -> None:
"""A bad request must look like the real API's, not like FastAPI's 422.
Mid-run, a 422 in FastAPI's own error shape reads as a Honcho bug rather
than a bad request, and no OpenAI client knows how to interpret it.
"""
response = client.post(
"/v1/embeddings", json={"input": "hello", "dimensions": "not-a-number"}
)
assert response.status_code == 400
error = response.json()["error"]
assert error["type"] == "invalid_request_error"
assert error["message"]
assert set(error) == {"message", "type", "param", "code"}
def test_boolean_dimensions_is_rejected_not_silently_coerced(
client: TestClient,
) -> None:
"""bool is an int subclass, so `true` would otherwise mean 1 dimension."""
response = client.post(
"/v1/embeddings", json={"input": "hello", "dimensions": True}
)
assert response.status_code == 400
@pytest.mark.parametrize("dimensions", [0, -1])
def test_non_positive_dimensions_is_rejected_not_defaulted(
client: TestClient, dimensions: int
) -> None:
"""Substituting 1536 would answer a bad request with a plausible vector."""
response = client.post(
"/v1/embeddings",
json={"input": "hello", "dimensions": dimensions, "encoding_format": "float"},
)
assert response.status_code == 400
assert response.json()["error"]["type"] == "invalid_request_error"
def test_unknown_fields_are_accepted(client: TestClient) -> None:
"""Validation must fire on wrong types, never on unrecognised parameters.
A new upstream parameter should not turn a working setup into a hard
failure, so every model allows extras.
"""
body = _post_chat(
client,
temperature=0.7,
max_completion_tokens=256,
reasoning_effort="minimal",
some_parameter_invented_next_year=True,
)
assert body["choices"][0]["finish_reason"] == "stop"
def test_wrongly_typed_messages_are_rejected(client: TestClient) -> None:
response = client.post(
"/v1/chat/completions", json={"model": "m", "messages": "not-a-list"}
)
assert response.status_code == 400
assert response.json()["error"]["type"] == "invalid_request_error"
@pytest.mark.parametrize(
("payload", "expected"),
[
({"input": "solo"}, 1),
({"input": ["a", "b", "c"]}, 3),
({"input": [1, 2, 3]}, 1),
({"input": [[1, 2], [3, 4]]}, 2),
({"input": None}, 0),
],
ids=["string", "list-of-strings", "token-array", "token-arrays", "null"],
)
def test_every_documented_input_shape_is_accepted(
client: TestClient, payload: dict[str, Any], expected: int
) -> None:
"""A flat int list is one tokenized input, not many single-token ones."""
response = client.post(
"/v1/embeddings", json={**payload, "encoding_format": "float"}
)
assert response.status_code == 200, response.text
assert len(response.json()["data"]) == expected
# --- streaming --------------------------------------------------------------
def _stream_chunks(
client: TestClient, stream_options: dict[str, Any] | None = None
) -> list[dict[str, Any]]:
"""The SSE payloads of a streaming completion, `[DONE]` asserted and dropped."""
body: dict[str, Any] = {
"model": "mock-model",
"messages": [{"role": "user", "content": "stream please"}],
"stream": True,
}
if stream_options is not None:
body["stream_options"] = stream_options
with client.stream("POST", "/v1/chat/completions", json=body) as response:
assert response.status_code == 200
lines = [
line[len("data: ") :]
for line in response.iter_lines()
if line.startswith("data: ")
]
assert lines[-1] == "[DONE]"
return [json.loads(line) for line in lines[:-1]]
def test_stream_emits_content_then_a_final_usage_chunk(client: TestClient) -> None:
"""The backend ends the stream on the usage chunk, so it must come last."""
chunks = _stream_chunks(client, {"include_usage": True})
content = "".join(
chunk["choices"][0]["delta"].get("content", "")
for chunk in chunks
if chunk["choices"]
)
assert "[mock]" in content
assert any(
chunk["choices"] and chunk["choices"][0]["finish_reason"] == "stop"
for chunk in chunks
)
usage_chunk = chunks[-1]
assert usage_chunk["usage"]["completion_tokens"] > 0
assert usage_chunk["choices"] == []
@pytest.mark.parametrize(
"stream_options",
[None, {}, {"include_usage": False}],
ids=["absent", "empty", "false"],
)
def test_stream_without_include_usage_emits_no_usage_chunk(
client: TestClient, stream_options: dict[str, Any] | None
) -> None:
"""The real API sends the usage chunk only when asked, so neither does this.
A caller that did not opt in must not have to skip a trailing chunk with an
empty `choices` array.
"""
chunks = _stream_chunks(client, stream_options)
assert all("usage" not in chunk for chunk in chunks)
assert chunks[-1]["choices"][0]["finish_reason"] == "stop"
content = "".join(
chunk["choices"][0]["delta"].get("content", "") for chunk in chunks
)
assert "[mock]" in content
@pytest.mark.parametrize("value", ["definitely", "yes", "on", "true", "1", 1])
def test_non_boolean_include_usage_is_rejected(client: TestClient, value: Any) -> None:
"""The usage chunk is conditional on this, so a wrong type must 400.
The truthy strings matter more than the nonsense one: plain `bool` coerces
"yes"/"on"/"true"/"1", so without StrictBool a string would silently decide
whether the stream carries usage.
"""
response = client.post(
"/v1/chat/completions",
json={
"model": "mock-model",
"messages": [{"role": "user", "content": "x"}],
"stream": True,
"stream_options": {"include_usage": value},
},
)
assert response.status_code == 400
assert response.json()["error"]["type"] == "invalid_request_error"
@pytest.mark.parametrize("value", ["yes", "true", "1", 1])
def test_non_boolean_stream_is_rejected(client: TestClient, value: Any) -> None:
"""`stream` picks between a JSON body and an SSE stream, so it must be exact."""
response = client.post(
"/v1/chat/completions",
json={
"model": "mock-model",
"messages": [{"role": "user", "content": "x"}],
"stream": value,
},
)
assert response.status_code == 400
assert response.json()["error"]["type"] == "invalid_request_error"