Merge pull request 'feat(use-cases): add inline use-case definitions' (#44) from feat/inline-use-case into main
All checks were successful
tests / test (push) Successful in 3m20s

This commit is contained in:
goldstein 2026-04-18 19:05:06 +00:00
commit 1481a7baac
9 changed files with 670 additions and 3 deletions

View file

@ -6,6 +6,8 @@ Designed to be used by other on-prem services (e.g. mammon) as a reliable fallba
Status: MVP deployed (2026-04-18) at `http://192.168.68.42:8994` — LAN only. Full reference spec at `docs/spec-core-pipeline.md`; MVP spec at `docs/superpowers/specs/2026-04-18-ix-mvp-design.md`; deploy runbook at `docs/deployment.md`. Status: MVP deployed (2026-04-18) at `http://192.168.68.42:8994` — LAN only. Full reference spec at `docs/spec-core-pipeline.md`; MVP spec at `docs/superpowers/specs/2026-04-18-ix-mvp-design.md`; deploy runbook at `docs/deployment.md`.
Use cases: the built-in registry lives in `src/ix/use_cases/__init__.py` (`bank_statement_header` for MVP). Callers without a registered entry can ship an ad-hoc schema inline via `RequestIX.use_case_inline` (see README "Ad-hoc use cases"); the pipeline builds the Pydantic classes on the fly per request.
## Guiding Principles ## Guiding Principles
- **On-prem always.** All LLM inference, OCR, and user-data processing run on the home server (192.168.68.42). No cloud APIs — OpenAI, Anthropic, Azure, AWS Bedrock/Textract, Google Document AI, Mistral, etc. are not to be used for user data or inference. LLM backend is Ollama (:11434); OCR runs locally (pluggable `OCRClient` interface, first engine: Surya on the RTX 3090); job state lives in local Postgres on the postgis container. The spec's references to Azure / AWS / OpenAI are examples to *replace*, not inherit. - **On-prem always.** All LLM inference, OCR, and user-data processing run on the home server (192.168.68.42). No cloud APIs — OpenAI, Anthropic, Azure, AWS Bedrock/Textract, Google Document AI, Mistral, etc. are not to be used for user data or inference. LLM backend is Ollama (:11434); OCR runs locally (pluggable `OCRClient` interface, first engine: Surya on the RTX 3090); job state lives in local Postgres on the postgis container. The spec's references to Azure / AWS / OpenAI are examples to *replace*, not inherit.

View file

@ -40,6 +40,28 @@ curl -X POST http://192.168.68.42:8994/jobs \
Poll `GET /jobs/{job_id}` until `status` is `done` or `error`. Optionally pass `callback_url` to receive a webhook on completion (one-shot, no retry; polling stays authoritative). Poll `GET /jobs/{job_id}` until `status` is `done` or `error`. Optionally pass `callback_url` to receive a webhook on completion (one-shot, no retry; polling stays authoritative).
### Ad-hoc use cases
For one-offs where a registered use case doesn't exist yet, ship the schema inline:
```jsonc
{
"use_case": "adhoc-invoice", // free-form label (logs/metrics only)
"use_case_inline": {
"use_case_name": "Invoice totals",
"system_prompt": "Extract vendor and total amount.",
"fields": [
{"name": "vendor", "type": "str", "required": true},
{"name": "total", "type": "decimal"},
{"name": "currency", "type": "str", "choices": ["USD", "EUR", "CHF"]}
]
},
// ...ix_client_id, request_id, context...
}
```
When `use_case_inline` is set, the pipeline builds the response schema on the fly and skips the registry. Supported types: `str`, `int`, `float`, `decimal`, `date`, `datetime`, `bool`. `choices` is only allowed on `str` fields. Precedence: inline wins over `use_case` when both are present.
Full REST surface + provenance response shape documented in the MVP design spec. Full REST surface + provenance response shape documented in the MVP design spec.
## Running locally ## Running locally

View file

@ -108,6 +108,25 @@ class ProvenanceOptions(BaseModel):
**Dropped from spec (no-ops under MVP):** `OCROptions.computer_vision_scaling_factor`, `include_page_tags` (always on), `GenAIOptions.use_vision`/`vision_scaling_factor`/`vision_detail`/`reasoning_effort`, `ProvenanceOptions.granularity`/`include_bounding_boxes`/`source_type`/`min_confidence`, `RequestIX.version`. **Dropped from spec (no-ops under MVP):** `OCROptions.computer_vision_scaling_factor`, `include_page_tags` (always on), `GenAIOptions.use_vision`/`vision_scaling_factor`/`vision_detail`/`reasoning_effort`, `ProvenanceOptions.granularity`/`include_bounding_boxes`/`source_type`/`min_confidence`, `RequestIX.version`.
**Ad-hoc use cases (post-MVP add-on).** `RequestIX` carries an optional `use_case_inline: InlineUseCase | None = None`. When set, the pipeline builds the `(Request, Response)` Pydantic class pair on the fly from that inline definition and **skips the registry lookup entirely** — the `use_case` field becomes a free-form label (still required for metrics / logging). Inline definitions look like:
```python
class UseCaseFieldDef(BaseModel):
name: str # valid Python identifier
type: Literal["str", "int", "float", "decimal", "date", "datetime", "bool"]
required: bool = False
description: str | None = None
choices: list[str] | None = None # str-typed fields only; builds Literal[*choices]
class InlineUseCase(BaseModel):
use_case_name: str
system_prompt: str
default_model: str | None = None
fields: list[UseCaseFieldDef]
```
Precedence: `use_case_inline` wins when both are set. Structural errors (dup field name, invalid identifier, `choices` on a non-str type, empty fields list) raise `IX_001_001` (same code as registry miss). The builder lives in `ix.use_cases.inline.build_use_case_classes` and returns fresh classes per call — the pipeline never caches them.
### ResponseIX ### ResponseIX
Identical to spec §2.2 except `FieldProvenance` gains two fields: Identical to spec §2.2 except `FieldProvenance` gains two fields:

View file

@ -83,6 +83,44 @@ class Options(BaseModel):
provenance: ProvenanceOptions = Field(default_factory=ProvenanceOptions) provenance: ProvenanceOptions = Field(default_factory=ProvenanceOptions)
class UseCaseFieldDef(BaseModel):
"""One field in an ad-hoc, caller-defined extraction schema.
The UI (and any other caller that doesn't want to wait on a backend
registry entry) ships one of these per desired output field. The pipeline
builds a fresh Pydantic response class from the list on each request.
``choices`` only applies to ``type == "str"`` it turns the field into a
``Literal[*choices]``. For any other type the builder raises
``IX_001_001``.
"""
model_config = ConfigDict(extra="forbid")
name: str # must be a valid Python identifier
type: Literal["str", "int", "float", "decimal", "date", "datetime", "bool"]
required: bool = False
description: str | None = None
choices: list[str] | None = None
class InlineUseCase(BaseModel):
"""Caller-defined use case bundled into the :class:`RequestIX`.
When present on a request, the pipeline builds the ``(Request, Response)``
Pydantic class pair on the fly from :attr:`fields` and skips the
registered use-case lookup. The registry-based ``use_case`` field is still
required on the request for metrics/logging but becomes a free-form label.
"""
model_config = ConfigDict(extra="forbid")
use_case_name: str
system_prompt: str
default_model: str | None = None
fields: list[UseCaseFieldDef]
class RequestIX(BaseModel): class RequestIX(BaseModel):
"""Top-level job request. """Top-level job request.
@ -90,6 +128,12 @@ class RequestIX(BaseModel):
it; the REST adapter / pg-queue adapter populates it on insert. The field it; the REST adapter / pg-queue adapter populates it on insert. The field
is kept here so the contract is closed-over-construction round-trips is kept here so the contract is closed-over-construction round-trips
(e.g. when the worker re-hydrates a job out of the store). (e.g. when the worker re-hydrates a job out of the store).
When ``use_case_inline`` is present, the pipeline uses it verbatim to
build an ad-hoc ``(Request, Response)`` class pair and skips the registry
lookup; ``use_case`` becomes a free-form label (still required for
metrics/logging). When ``use_case_inline`` is absent, ``use_case`` is
looked up in :data:`ix.use_cases.REGISTRY` as before.
""" """
model_config = ConfigDict(extra="forbid") model_config = ConfigDict(extra="forbid")
@ -101,3 +145,4 @@ class RequestIX(BaseModel):
context: Context context: Context
options: Options = Field(default_factory=Options) options: Options = Field(default_factory=Options)
callback_url: str | None = None callback_url: str | None = None
use_case_inline: InlineUseCase | None = None

View file

@ -34,6 +34,7 @@ from ix.ingestion import (
) )
from ix.pipeline.step import Step from ix.pipeline.step import Step
from ix.use_cases import get_use_case from ix.use_cases import get_use_case
from ix.use_cases.inline import build_use_case_classes
class _Fetcher(Protocol): class _Fetcher(Protocol):
@ -88,9 +89,18 @@ class SetupStep(Step):
async def process( async def process(
self, request_ix: RequestIX, response_ix: ResponseIX self, request_ix: RequestIX, response_ix: ResponseIX
) -> ResponseIX: ) -> ResponseIX:
# 1. Load the use-case pair — early so an unknown name fails before # 1. Load the use-case pair — either from the caller's inline
# we waste time downloading files. # definition (wins over registry) or from the registry by name.
use_case_request_cls, use_case_response_cls = get_use_case(request_ix.use_case) # Done early so an unknown name / bad inline definition fails
# before we waste time downloading files.
if request_ix.use_case_inline is not None:
use_case_request_cls, use_case_response_cls = build_use_case_classes(
request_ix.use_case_inline
)
else:
use_case_request_cls, use_case_response_cls = get_use_case(
request_ix.use_case
)
use_case_request = use_case_request_cls() use_case_request = use_case_request_cls()
# 2. Resolve the per-request scratch directory. ix_id is assigned # 2. Resolve the per-request scratch directory. ix_id is assigned

132
src/ix/use_cases/inline.py Normal file
View file

@ -0,0 +1,132 @@
"""Dynamic Pydantic class builder for caller-supplied use cases.
Input: an :class:`ix.contracts.request.InlineUseCase` carried on the
:class:`~ix.contracts.request.RequestIX`.
Output: a fresh ``(RequestClass, ResponseClass)`` pair with the same shape
as a registered use case. The :class:`~ix.pipeline.setup_step.SetupStep`
calls this when ``request_ix.use_case_inline`` is set, bypassing the
registry lookup entirely.
The builder returns brand-new classes on every call safe to call per
request, so two concurrent jobs can't step on each other's schemas even if
they happen to share a ``use_case_name``. Validation errors map to
``IX_001_001`` (same code the registry-miss path uses); the error is
recoverable from the caller's perspective (fix the JSON and retry), not an
infra problem.
"""
from __future__ import annotations
import keyword
import re
from datetime import date, datetime
from decimal import Decimal
from typing import Any, Literal, cast
from pydantic import BaseModel, ConfigDict, Field, create_model
from ix.contracts.request import InlineUseCase, UseCaseFieldDef
from ix.errors import IXErrorCode, IXException
# Map the ``UseCaseFieldDef.type`` literal to concrete Python types.
_TYPE_MAP: dict[str, type] = {
"str": str,
"int": int,
"float": float,
"decimal": Decimal,
"date": date,
"datetime": datetime,
"bool": bool,
}
def _fail(detail: str) -> IXException:
return IXException(IXErrorCode.IX_001_001, detail=detail)
def _valid_field_name(name: str) -> bool:
"""Require a valid Python identifier that isn't a reserved keyword."""
return name.isidentifier() and not keyword.iskeyword(name)
def _resolve_field_type(field: UseCaseFieldDef) -> Any:
"""Return the annotation for a single field, with ``choices`` honoured."""
base = _TYPE_MAP[field.type]
if field.choices:
if field.type != "str":
raise _fail(
f"field {field.name!r}: 'choices' is only allowed for "
f"type='str' (got {field.type!r})"
)
return Literal[tuple(field.choices)] # type: ignore[valid-type]
return base
def _sanitise_class_name(raw: str) -> str:
"""``re.sub(r"\\W", "_", name)`` + ``Inline_`` prefix.
Keeps the generated class name debuggable (shows up in repr / tracebacks)
while ensuring it's always a valid Python identifier.
"""
return "Inline_" + re.sub(r"\W", "_", raw)
def build_use_case_classes(
inline: InlineUseCase,
) -> tuple[type[BaseModel], type[BaseModel]]:
"""Build a fresh ``(RequestClass, ResponseClass)`` from ``inline``.
* Every call returns new classes. The caller may cache if desired; the
pipeline intentionally does not.
* Raises :class:`~ix.errors.IXException` with code
:attr:`~ix.errors.IXErrorCode.IX_001_001` on any structural problem
(empty fields, bad name, dup name, bad ``choices``).
"""
if not inline.fields:
raise _fail("inline use case must define at least one field")
seen: set[str] = set()
for fd in inline.fields:
if not _valid_field_name(fd.name):
raise _fail(f"field name {fd.name!r} is not a valid Python identifier")
if fd.name in seen:
raise _fail(f"duplicate field name {fd.name!r}")
seen.add(fd.name)
response_fields: dict[str, Any] = {}
for fd in inline.fields:
annotation = _resolve_field_type(fd)
field_info = Field(
...,
description=fd.description,
) if fd.required else Field(
default=None,
description=fd.description,
)
if not fd.required:
annotation = annotation | None
response_fields[fd.name] = (annotation, field_info)
response_cls = create_model( # type: ignore[call-overload]
_sanitise_class_name(inline.use_case_name),
__config__=ConfigDict(extra="forbid"),
**response_fields,
)
request_cls = create_model( # type: ignore[call-overload]
"Inline_Request_" + re.sub(r"\W", "_", inline.use_case_name),
__config__=ConfigDict(extra="forbid"),
use_case_name=(str, inline.use_case_name),
system_prompt=(str, inline.system_prompt),
default_model=(str | None, inline.default_model),
)
return cast(type[BaseModel], request_cls), cast(type[BaseModel], response_cls)
__all__ = ["build_use_case_classes"]

View file

@ -31,6 +31,7 @@ from ix.contracts import (
ResponseIX, ResponseIX,
SegmentCitation, SegmentCitation,
) )
from ix.contracts.request import InlineUseCase, UseCaseFieldDef
class TestFileRef: class TestFileRef:
@ -182,6 +183,32 @@ class TestRequestIX:
with pytest.raises(ValidationError): with pytest.raises(ValidationError):
RequestIX.model_validate({"use_case": "x"}) RequestIX.model_validate({"use_case": "x"})
def test_use_case_inline_defaults_to_none(self) -> None:
r = RequestIX(**self._minimal_payload())
assert r.use_case_inline is None
def test_use_case_inline_roundtrip(self) -> None:
payload = self._minimal_payload()
payload["use_case_inline"] = {
"use_case_name": "adhoc",
"system_prompt": "extract stuff",
"fields": [
{"name": "a", "type": "str", "required": True},
{"name": "b", "type": "int"},
],
}
r = RequestIX.model_validate(payload)
assert r.use_case_inline is not None
assert isinstance(r.use_case_inline, InlineUseCase)
assert r.use_case_inline.use_case_name == "adhoc"
assert len(r.use_case_inline.fields) == 2
assert isinstance(r.use_case_inline.fields[0], UseCaseFieldDef)
# Round-trip through JSON
dumped = r.model_dump_json()
r2 = RequestIX.model_validate_json(dumped)
assert r2.use_case_inline is not None
assert r2.use_case_inline.fields[1].type == "int"
class TestOCRResult: class TestOCRResult:
def test_minimal_defaults(self) -> None: def test_minimal_defaults(self) -> None:

View file

@ -15,6 +15,7 @@ from ix.contracts import (
RequestIX, RequestIX,
ResponseIX, ResponseIX,
) )
from ix.contracts.request import InlineUseCase, UseCaseFieldDef
from ix.contracts.response import _InternalContext from ix.contracts.response import _InternalContext
from ix.errors import IXErrorCode, IXException from ix.errors import IXErrorCode, IXException
from ix.ingestion import FetchConfig from ix.ingestion import FetchConfig
@ -244,6 +245,102 @@ class TestTextOnly:
assert ctx.texts == ["hello", "there"] assert ctx.texts == ["hello", "there"]
class TestInlineUseCase:
def _make_inline_request(
self,
inline: InlineUseCase,
use_case: str = "adhoc-label",
texts: list[str] | None = None,
) -> RequestIX:
return RequestIX(
use_case=use_case,
use_case_inline=inline,
ix_client_id="test",
request_id="r-inline",
context=Context(files=[], texts=texts or ["hello"]),
options=Options(
ocr=OCROptions(use_ocr=True),
provenance=ProvenanceOptions(include_provenance=True),
),
)
async def test_inline_use_case_overrides_registry(self, tmp_path: Path) -> None:
fetcher = FakeFetcher({})
ingestor = FakeIngestor([])
step = SetupStep(
fetcher=fetcher,
ingestor=ingestor,
tmp_dir=tmp_path / "work",
fetch_config=_make_cfg(),
mime_detector=_AlwaysMimePdf(),
)
inline = InlineUseCase(
use_case_name="adhoc",
system_prompt="Extract things.",
fields=[
UseCaseFieldDef(name="vendor", type="str", required=True),
UseCaseFieldDef(name="amount", type="decimal"),
],
)
req = self._make_inline_request(inline)
resp = _make_response()
resp = await step.process(req, resp)
ctx = resp.context
assert ctx is not None
# The response class must have been built from our field list.
resp_cls = ctx.use_case_response # type: ignore[union-attr]
assert set(resp_cls.model_fields.keys()) == {"vendor", "amount"}
# Public display name reflects the inline label.
assert resp.use_case_name == "adhoc"
async def test_inline_precedence_when_both_set(self, tmp_path: Path) -> None:
# ``use_case`` is a valid registered name; ``use_case_inline`` is also
# present. Inline MUST win (documented precedence).
fetcher = FakeFetcher({})
ingestor = FakeIngestor([])
step = SetupStep(
fetcher=fetcher,
ingestor=ingestor,
tmp_dir=tmp_path / "work",
fetch_config=_make_cfg(),
mime_detector=_AlwaysMimePdf(),
)
inline = InlineUseCase(
use_case_name="override",
system_prompt="override prompt",
fields=[UseCaseFieldDef(name="just_me", type="str", required=True)],
)
req = self._make_inline_request(
inline, use_case="bank_statement_header"
)
resp = await step.process(req, _make_response())
resp_cls = resp.context.use_case_response # type: ignore[union-attr]
assert set(resp_cls.model_fields.keys()) == {"just_me"}
async def test_inline_with_bad_field_raises_ix_001_001(
self, tmp_path: Path
) -> None:
fetcher = FakeFetcher({})
ingestor = FakeIngestor([])
step = SetupStep(
fetcher=fetcher,
ingestor=ingestor,
tmp_dir=tmp_path / "work",
fetch_config=_make_cfg(),
mime_detector=_AlwaysMimePdf(),
)
inline = InlineUseCase(
use_case_name="bad",
system_prompt="p",
fields=[UseCaseFieldDef(name="123bad", type="str")],
)
req = self._make_inline_request(inline)
with pytest.raises(IXException) as ei:
await step.process(req, _make_response())
assert ei.value.code is IXErrorCode.IX_001_001
class TestInternalContextShape: class TestInternalContextShape:
async def test_context_is_internal_context_instance(self, tmp_path: Path) -> None: async def test_context_is_internal_context_instance(self, tmp_path: Path) -> None:
fetcher = FakeFetcher({}) fetcher = FakeFetcher({})

View file

@ -0,0 +1,313 @@
"""Tests for :mod:`ix.use_cases.inline` — dynamic Pydantic class builder.
The builder takes an :class:`InlineUseCase` (carried on :class:`RequestIX` as
``use_case_inline``) and produces a fresh ``(RequestClass, ResponseClass)``
pair that the pipeline can consume in place of a registered use case.
"""
from __future__ import annotations
import json
from datetime import date, datetime
from decimal import Decimal
import pytest
from pydantic import BaseModel, ValidationError
from ix.contracts.request import InlineUseCase, UseCaseFieldDef
from ix.errors import IXErrorCode, IXException
from ix.use_cases.inline import build_use_case_classes
class TestUseCaseFieldDef:
def test_minimal(self) -> None:
fd = UseCaseFieldDef(name="foo", type="str")
assert fd.name == "foo"
assert fd.type == "str"
assert fd.required is False
assert fd.description is None
assert fd.choices is None
def test_extra_forbidden(self) -> None:
with pytest.raises(ValidationError):
UseCaseFieldDef.model_validate(
{"name": "foo", "type": "str", "bogus": 1}
)
def test_invalid_type_rejected(self) -> None:
with pytest.raises(ValidationError):
UseCaseFieldDef.model_validate({"name": "foo", "type": "list"})
class TestInlineUseCaseRoundtrip:
def test_json_roundtrip(self) -> None:
iuc = InlineUseCase(
use_case_name="Vendor Total",
system_prompt="Extract invoice total and vendor.",
default_model="qwen3:14b",
fields=[
UseCaseFieldDef(name="vendor", type="str", required=True),
UseCaseFieldDef(
name="total",
type="decimal",
required=True,
description="total amount due",
),
UseCaseFieldDef(
name="currency",
type="str",
choices=["USD", "EUR", "CHF"],
),
],
)
dumped = iuc.model_dump_json()
round = InlineUseCase.model_validate_json(dumped)
assert round == iuc
# JSON is well-formed
json.loads(dumped)
def test_extra_forbidden(self) -> None:
with pytest.raises(ValidationError):
InlineUseCase.model_validate(
{
"use_case_name": "X",
"system_prompt": "p",
"fields": [],
"bogus": 1,
}
)
class TestBuildBasicTypes:
@pytest.mark.parametrize(
"type_name, sample_value, bad_value",
[
("str", "hello", 123),
("int", 42, "nope"),
("float", 3.14, "nope"),
("bool", True, "nope"),
],
)
def test_simple_type(
self, type_name: str, sample_value: object, bad_value: object
) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[UseCaseFieldDef(name="field", type=type_name, required=True)],
)
_req_cls, resp_cls = build_use_case_classes(iuc)
instance = resp_cls(field=sample_value)
assert instance.field == sample_value
with pytest.raises(ValidationError):
resp_cls(field=bad_value)
def test_decimal_type(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[UseCaseFieldDef(name="amount", type="decimal", required=True)],
)
_req_cls, resp_cls = build_use_case_classes(iuc)
instance = resp_cls(amount="12.34")
assert isinstance(instance.amount, Decimal)
assert instance.amount == Decimal("12.34")
def test_date_type(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[UseCaseFieldDef(name="d", type="date", required=True)],
)
_req_cls, resp_cls = build_use_case_classes(iuc)
instance = resp_cls(d="2026-04-18")
assert instance.d == date(2026, 4, 18)
def test_datetime_type(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[UseCaseFieldDef(name="ts", type="datetime", required=True)],
)
_req_cls, resp_cls = build_use_case_classes(iuc)
instance = resp_cls(ts="2026-04-18T10:00:00")
assert isinstance(instance.ts, datetime)
class TestOptionalVsRequired:
def test_required_field_cannot_be_missing(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[UseCaseFieldDef(name="must", type="str", required=True)],
)
_req_cls, resp_cls = build_use_case_classes(iuc)
with pytest.raises(ValidationError):
resp_cls()
def test_optional_field_defaults_to_none(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[UseCaseFieldDef(name="maybe", type="str", required=False)],
)
_req_cls, resp_cls = build_use_case_classes(iuc)
instance = resp_cls()
assert instance.maybe is None
def test_optional_field_schema_allows_null(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[UseCaseFieldDef(name="maybe", type="str", required=False)],
)
_req_cls, resp_cls = build_use_case_classes(iuc)
schema = resp_cls.model_json_schema()
# "maybe" accepts string or null
prop = schema["properties"]["maybe"]
# Pydantic may express Optional as anyOf [str, null] or a type list.
# Either is fine — just assert null is allowed somewhere.
dumped = json.dumps(prop)
assert "null" in dumped
class TestChoices:
def test_choices_for_str_produces_literal(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[
UseCaseFieldDef(
name="kind",
type="str",
required=True,
choices=["a", "b", "c"],
)
],
)
_req_cls, resp_cls = build_use_case_classes(iuc)
inst = resp_cls(kind="a")
assert inst.kind == "a"
with pytest.raises(ValidationError):
resp_cls(kind="nope")
schema = resp_cls.model_json_schema()
# enum or const wind up in a referenced definition; walk the schema
dumped = json.dumps(schema)
assert '"a"' in dumped and '"b"' in dumped and '"c"' in dumped
def test_choices_for_non_str_raises_ix_001_001(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[
UseCaseFieldDef(
name="kind",
type="int",
required=True,
choices=["1", "2"],
)
],
)
with pytest.raises(IXException) as exc:
build_use_case_classes(iuc)
assert exc.value.code is IXErrorCode.IX_001_001
def test_empty_choices_list_ignored(self) -> None:
# An explicitly empty list is as-if choices were unset; builder must
# not break. If the caller sent choices=[] we treat the field as
# plain str.
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[
UseCaseFieldDef(
name="kind", type="str", required=True, choices=[]
)
],
)
_req_cls, resp_cls = build_use_case_classes(iuc)
inst = resp_cls(kind="anything")
assert inst.kind == "anything"
class TestValidation:
def test_duplicate_field_names_raise(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[
UseCaseFieldDef(name="foo", type="str"),
UseCaseFieldDef(name="foo", type="int"),
],
)
with pytest.raises(IXException) as exc:
build_use_case_classes(iuc)
assert exc.value.code is IXErrorCode.IX_001_001
def test_invalid_field_name_raises(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[UseCaseFieldDef(name="123abc", type="str")],
)
with pytest.raises(IXException) as exc:
build_use_case_classes(iuc)
assert exc.value.code is IXErrorCode.IX_001_001
def test_empty_fields_list_raises(self) -> None:
iuc = InlineUseCase(
use_case_name="X", system_prompt="p", fields=[]
)
with pytest.raises(IXException) as exc:
build_use_case_classes(iuc)
assert exc.value.code is IXErrorCode.IX_001_001
class TestResponseClassNaming:
def test_class_name_sanitised(self) -> None:
iuc = InlineUseCase(
use_case_name="Bank / Statement — header!",
system_prompt="p",
fields=[UseCaseFieldDef(name="x", type="str")],
)
_req_cls, resp_cls = build_use_case_classes(iuc)
assert resp_cls.__name__.startswith("Inline_")
# Only alphanumerics and underscores remain.
assert all(c.isalnum() or c == "_" for c in resp_cls.__name__)
def test_fresh_instances_per_call(self) -> None:
iuc = InlineUseCase(
use_case_name="X",
system_prompt="p",
fields=[UseCaseFieldDef(name="x", type="str")],
)
req1, resp1 = build_use_case_classes(iuc)
req2, resp2 = build_use_case_classes(iuc)
assert resp1 is not resp2
assert req1 is not req2
class TestRequestClassShape:
def test_request_class_exposes_prompt_and_default(self) -> None:
iuc = InlineUseCase(
use_case_name="My Case",
system_prompt="Follow directions.",
default_model="qwen3:14b",
fields=[UseCaseFieldDef(name="x", type="str")],
)
req_cls, _resp_cls = build_use_case_classes(iuc)
inst = req_cls()
assert inst.use_case_name == "My Case"
assert inst.system_prompt == "Follow directions."
assert inst.default_model == "qwen3:14b"
assert issubclass(req_cls, BaseModel)
def test_default_model_none_when_unset(self) -> None:
iuc = InlineUseCase(
use_case_name="My Case",
system_prompt="Follow directions.",
fields=[UseCaseFieldDef(name="x", type="str")],
)
req_cls, _resp_cls = build_use_case_classes(iuc)
inst = req_cls()
assert inst.default_model is None