From 703da9035ea4898811bd5290c6233ec1de0f7534 Mon Sep 17 00:00:00 2001 From: Dirk Riemann Date: Sat, 18 Apr 2026 21:01:27 +0200 Subject: [PATCH] feat(use-cases): add inline use-case definitions Adds RequestIX.use_case_inline so callers can define ad-hoc extraction schemas in the request itself, bypassing the backend registry. The pipeline builds a fresh (Request, Response) Pydantic class pair per call via ix.use_cases.inline.build_use_case_classes; structural errors (dup field, bad identifier, choices-on-non-str, empty fields) raise IX_001_001 to match the registry-miss path. Inline wins when both use_case and use_case_inline are set. Existing REST callers see no behavioural change. Co-Authored-By: Claude Opus 4.7 (1M context) --- AGENTS.md | 2 + README.md | 22 ++ .../specs/2026-04-18-ix-mvp-design.md | 19 ++ src/ix/contracts/request.py | 45 +++ src/ix/pipeline/setup_step.py | 16 +- src/ix/use_cases/inline.py | 132 ++++++++ tests/unit/test_contracts.py | 27 ++ tests/unit/test_setup_step.py | 97 ++++++ tests/unit/test_use_case_inline.py | 313 ++++++++++++++++++ 9 files changed, 670 insertions(+), 3 deletions(-) create mode 100644 src/ix/use_cases/inline.py create mode 100644 tests/unit/test_use_case_inline.py diff --git a/AGENTS.md b/AGENTS.md index ef55c1b..885745a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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`. +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 - **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. diff --git a/README.md b/README.md index 1906459..7c065ae 100644 --- a/README.md +++ b/README.md @@ -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). +### 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. ## Running locally diff --git a/docs/superpowers/specs/2026-04-18-ix-mvp-design.md b/docs/superpowers/specs/2026-04-18-ix-mvp-design.md index 7e9356a..06926ed 100644 --- a/docs/superpowers/specs/2026-04-18-ix-mvp-design.md +++ b/docs/superpowers/specs/2026-04-18-ix-mvp-design.md @@ -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`. +**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 Identical to spec §2.2 except `FieldProvenance` gains two fields: diff --git a/src/ix/contracts/request.py b/src/ix/contracts/request.py index 58e3d3c..c11e113 100644 --- a/src/ix/contracts/request.py +++ b/src/ix/contracts/request.py @@ -83,6 +83,44 @@ class Options(BaseModel): 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): """Top-level job request. @@ -90,6 +128,12 @@ class RequestIX(BaseModel): 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 (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") @@ -101,3 +145,4 @@ class RequestIX(BaseModel): context: Context options: Options = Field(default_factory=Options) callback_url: str | None = None + use_case_inline: InlineUseCase | None = None diff --git a/src/ix/pipeline/setup_step.py b/src/ix/pipeline/setup_step.py index 707f34d..d73da79 100644 --- a/src/ix/pipeline/setup_step.py +++ b/src/ix/pipeline/setup_step.py @@ -34,6 +34,7 @@ from ix.ingestion import ( ) from ix.pipeline.step import Step from ix.use_cases import get_use_case +from ix.use_cases.inline import build_use_case_classes class _Fetcher(Protocol): @@ -88,9 +89,18 @@ class SetupStep(Step): async def process( self, request_ix: RequestIX, response_ix: ResponseIX ) -> ResponseIX: - # 1. Load the use-case pair — early so an unknown name fails before - # we waste time downloading files. - use_case_request_cls, use_case_response_cls = get_use_case(request_ix.use_case) + # 1. Load the use-case pair — either from the caller's inline + # definition (wins over registry) or from the registry by name. + # 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() # 2. Resolve the per-request scratch directory. ix_id is assigned diff --git a/src/ix/use_cases/inline.py b/src/ix/use_cases/inline.py new file mode 100644 index 0000000..2ba79f1 --- /dev/null +++ b/src/ix/use_cases/inline.py @@ -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"] diff --git a/tests/unit/test_contracts.py b/tests/unit/test_contracts.py index 8f5d1a1..4e81c53 100644 --- a/tests/unit/test_contracts.py +++ b/tests/unit/test_contracts.py @@ -31,6 +31,7 @@ from ix.contracts import ( ResponseIX, SegmentCitation, ) +from ix.contracts.request import InlineUseCase, UseCaseFieldDef class TestFileRef: @@ -182,6 +183,32 @@ class TestRequestIX: with pytest.raises(ValidationError): 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: def test_minimal_defaults(self) -> None: diff --git a/tests/unit/test_setup_step.py b/tests/unit/test_setup_step.py index 3376c22..e3e8853 100644 --- a/tests/unit/test_setup_step.py +++ b/tests/unit/test_setup_step.py @@ -15,6 +15,7 @@ from ix.contracts import ( RequestIX, ResponseIX, ) +from ix.contracts.request import InlineUseCase, UseCaseFieldDef from ix.contracts.response import _InternalContext from ix.errors import IXErrorCode, IXException from ix.ingestion import FetchConfig @@ -244,6 +245,102 @@ class TestTextOnly: 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: async def test_context_is_internal_context_instance(self, tmp_path: Path) -> None: fetcher = FakeFetcher({}) diff --git a/tests/unit/test_use_case_inline.py b/tests/unit/test_use_case_inline.py new file mode 100644 index 0000000..e8a1b99 --- /dev/null +++ b/tests/unit/test_use_case_inline.py @@ -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