diff --git a/README.md b/README.md index aa5450e9..e4316082 100644 --- a/README.md +++ b/README.md @@ -162,7 +162,9 @@ runs as a Python worker or as a native worker on CUDA. Open-Jev-27B-v1.1 and Open-Jev-9B run on the same native Rust/CUDA worker. Cua-S1 also has a Python screenshot worker, Valen-Preview-0923 has a Python reference worker for text and one-image state, CLM has a stub-encoder contract recipe, and OmniJev-4B -v1.1 has its checkpoint export and CPU contract, with no worker yet: +v1.1 has its checkpoint export and CPU contract, with no worker yet. D3 +(Decision 3.0) has a Python reference worker that runs the checkpoint's own +pinned runtime for text state: | Model | Status | | --- | --- | @@ -176,6 +178,7 @@ v1.1 has its checkpoint export and CPU contract, with no worker yet: | CLM-v0.1-8B | [External worker with a CPU stub encoder](recipe/clm/README.md); contract checks only, real Qwen3-8B decisions unverified by this recipe | | LFM2.5-350M | [Transformers worker](recipe/lfm2/README.md); text state and independent choice questions | | OmniJev-4B v1.1 | [Checkpoint export and CPU contract](recipe/omnijev/README.md) only; the native worker is in progress ([#114](https://github.com/ThinkFlowLab/system1-omni/issues/114)) | +| D3 (Decision 3.0) | [Python reference worker](recipe/d3/reference.md) on the checkpoint's pinned runtime; `choice`/`noul`/`score` text state, media refused until stage 2; author-run on A100 (evidence in the integration PR) | [Supported models and hardware](docs/supported-models.md) lists the devices and where each worker has been run. diff --git a/docs/supported-models.md b/docs/supported-models.md index c92e1aff..3ca05477 100644 --- a/docs/supported-models.md +++ b/docs/supported-models.md @@ -23,6 +23,7 @@ Models that are being added are also tracked in issues labeled [new model](https | LFM2.5-350M | [Transformers choice worker](../recipe/lfm2/README.md) | Tiny-model regression tests; full checkpoint unverified | L40S FP16 [recorded execution checks](../recipe/lfm2/RESULTS.md) | Unverified | Python 3.12, [pinned dependencies](../src/models/lfm2/requirements.lock.txt), explicit candidate batch size | | CLM-v0.1-8B | [Native Rust engine](../src/models/clm/README.md), `clm-run`, in front of a Qwen3-8B `/v1/embeddings` server | Engine-only agreement **4.5e-06** on identical vectors ([#29](https://github.com/ThinkFlowLab/system1-omni/pull/29), [validation](../recipe/clm/native/VALIDATION.md)); contract checks documented ([#23](https://github.com/ThinkFlowLab/system1-omni/pull/23)) | **Validated on compute capability 8.9** ([#29](https://github.com/ThinkFlowLab/system1-omni/pull/29), [validation](../recipe/clm/native/VALIDATION.md)) | Unverified | Rust to build and a separate Qwen3-8B `/v1/embeddings` server for the encoder; the engine itself has no device-specific code | | OmniJev-4B v1.1 | None yet: [checkpoint export and CPU contract](../recipe/omnijev/README.md); the native worker is in progress ([#114](https://github.com/ThinkFlowLab/system1-omni/issues/114)) | Request contract, preparation, heads and finishing checked against the reference ([#120](https://github.com/ThinkFlowLab/system1-omni/pull/120)); no inference | Not yet | Not supported | Python 3.10 reference environment for the export and fixtures | +| D3 (Decision 3.0) | [Python reference worker](../recipe/d3/reference.md) running the checkpoint's pinned runtime | Unverified | Author-run on NVIDIA A100 (sm_80): three-kind serving smoke with byte-identical frontend forwarding and reference parity (exact decisions, max delta 0.0, 348 == 348 tokens); evidence in the integration PR. Owner: @MrDongsls. | Unverified | Python 3.12, the [pinned runtime requirements](../recipe/d3/requirements.txt), the [pinned package](https://huggingface.co/vllm-sr/d3) (~52 GB) | - **Validated:** covered by the recipe on `main` or by the checks in the linked merged pull request. - **Author-run:** recorded by the recipe author on the stated hardware and not independently reproduced; the row names its owner. @@ -31,6 +32,8 @@ Models that are being added are also tracked in issues labeled [new model](https The Cua-S1 workers answer `choice` questions only. LAYA's English worker, Open-Jev and CLM support `choice`, `score`, and `noul` text questions. +D3's reference worker serves the same three kinds for text state; its media +inputs stay refused for now and will be integrated soon. CLM has a row per path: the merged `clm-serve` recipe exercises those answer shapes with stub embeddings and says nothing about decision quality, while the native engine's CUDA row is the path through a real Qwen3-8B encoder. Its engine-only figure comes from the same comparison run diff --git a/mkdocs.yml b/mkdocs.yml index 7033af43..055b7c24 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -75,4 +75,5 @@ nav: - OmniJev preparation: recipe/omnijev/README.md - CLM stub-encoder contract: recipe/clm/README.md - LFM2.5-350M choice worker: recipe/lfm2/README.md + - D3 reference worker: recipe/d3/reference.md - Contributing: CONTRIBUTING.md diff --git a/recipe/README.md b/recipe/README.md index ce860871..32f83cd0 100644 --- a/recipe/README.md +++ b/recipe/README.md @@ -25,6 +25,9 @@ For a first real decision, follow the [complete CPU walkthrough](../docs/getting regenerate the CPU fixtures; the native worker is not available yet. - [CLM behind the frontend](clm/README.md): run CLM's own server behind the frontend on CPU with a stub encoder, and what the response comparison has to allow for. +- [D3 reference worker](d3/reference.md): download the pinned vllm-sr/d3 package, + verify its runtime manifest, and serve `choice`/`noul`/`score` text decisions + through the Python worker and the Rust frontend (serving and parity validated on A100). - [LFM2.5-350M choice worker](lfm2/README.md): score text decisions with independent questions, hybrid cache branching and configurable candidate batches. diff --git a/recipe/d3/compare_reference.py b/recipe/d3/compare_reference.py new file mode 100644 index 00000000..b7ef07de --- /dev/null +++ b/recipe/d3/compare_reference.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 +"""Compare the d3 worker pipeline against the pinned upstream server logic. + +Runs each request through two paths that share one loaded pinned model, so +the comparison isolates this repository's admission and orchestration from +the checkpoint's own code: + +- reference: the decide() body of the upstream ``d3_server.py`` (prepare, + reject over-limit and invalid questions with 422, run, respond), driven + directly on the loaded runtime; +- worker: the production ``frontend.d3.decide`` pipeline (protocol envelope, + prepared-error admission, run, respond, response finishing). + +Both paths execute the same verified ``d3_runtime``, so the criteria are +exact decision agreement (including embedded ``invalid_model_output`` +shapes), equal token accounting, and a probability tolerance of 1e-6 +(``--tolerance``). ``--export`` additionally freezes the oracle fixture — +token ids, probabilities and answers per question — for the future native +worker, and writes it only after a passing, non-rejected comparison. +Requires the pinned d3 package and a working device; see +recipe/d3/reference.md. GPU use is opt-in: run this tool by hand only. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +HERE = Path(__file__).resolve().parent +PROBABILITY_TOLERANCE = 1e-6 + + +def parse_args(argv=None): + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--model", required=True, help="pinned d3 package directory") + parser.add_argument("--device", default="cuda:0") + parser.add_argument("--batch-size", type=int, default=8) + parser.add_argument("--verify", default="fast", choices=("fast", "full", "none")) + parser.add_argument( + "--fixture", + default=str(HERE / "example-request.json"), + help="request body (default: the recipe fixture)", + ) + parser.add_argument("--tolerance", type=float, default=PROBABILITY_TOLERANCE) + parser.add_argument( + "--export", + default=None, + help="write the frozen oracle fixture (token ids, probabilities, answers); " + "only a passing, non-rejected comparison is written", + ) + return parser.parse_args(argv) + + +def reference_decide(model, body): + """The upstream d3_server.py decide() body, without FastAPI. + + Returns the 200 response dict, or a ("422", detail) tuple when the + upstream server would have rejected the request before execution. + """ + + prepared = model.prepare(body.get("state"), body.get("questions")) + over = [e for e in prepared.errors.values() if e["error"] == "max_length_exceeded"] + if over: + return ("422", over[0]["message"]) + invalid = {k: e["message"] for k, e in prepared.errors.items()} + if invalid: + return ("422", {"invalid_questions": invalid}) + probabilities, tokens = model.run(prepared) + return model.respond(prepared, probabilities, tokens) + + +def _numeric_delta(answer, reference, problems, deltas): + """Compare one answer pair; returns exact-match problems and max deltas.""" + + if set(answer) != set(reference): + problems.append(f"fields {sorted(answer)} != {sorted(reference)}") + return + if answer.get("type") != reference.get("type"): + problems.append(f"type {answer.get('type')!r} != {reference.get('type')!r}") + if "error" in reference: + if answer.get("error") != reference.get("error") or answer.get( + "message" + ) != reference.get("message"): + problems.append("embedded error shape differs") + return + if "choice" in reference and answer.get("choice") != reference["choice"]: + problems.append(f"choice {answer.get('choice')!r} != {reference['choice']!r}") + if "noul" in reference: + deltas.append(abs(float(answer["noul"]) - float(reference["noul"]))) + if "score" in reference: + deltas.append(abs(float(answer["score"]) - float(reference["score"]))) + if "legend" in reference and answer.get("legend") != reference.get("legend"): + problems.append("legend differs") + if "probabilities" in reference: + if list(answer.get("probabilities") or ()) != list(reference["probabilities"]): + problems.append("probability key order differs") + else: + deltas.extend( + abs(float(a) - float(r)) + for a, r in zip( + answer["probabilities"].values(), reference["probabilities"].values() + ) + ) + if "confidence" in reference: + deltas.append(abs(float(answer["confidence"]) - float(reference["confidence"]))) + + +def compare_case(name, worker_response, reference, tolerance=PROBABILITY_TOLERANCE): + """Compare one worker response against one reference result. + + Returns (report lines, passed). + """ + + lines = [f"case: {name}", "reference path: d3_server decide()", "worker model: d3"] + problems = [] + deltas: list[float] = [] + if isinstance(reference, tuple): + rejection = worker_response.get("worker rejected the request") + if not isinstance(rejection, dict): + problems.append(f"reference rejected the request: {reference[1]}") + else: + detail = reference[1] + body = rejection.get("body") + agreed = rejection.get("status") == 422 and body == ( + detail if isinstance(detail, dict) else {"detail": detail} + ) + lines.append(f"worker rejection: {rejection.get('status')} {body}") + lines.append("rejection agreement: " + ("PASS" if agreed else "FAIL")) + return lines, bool(agreed) + elif "worker rejected the request" in worker_response: + problems.append(str(worker_response["worker rejected the request"])) + else: + if list(worker_response["answers"]) != list(reference["answers"]): + problems.append("answer order differs") + for qid, ref_answer in reference["answers"].items(): + _numeric_delta(worker_response["answers"][qid], ref_answer, problems, deltas) + usage, ref_usage = worker_response["usage"], reference["usage"] + lines.append( + f"input_tokens: {usage.get('input_tokens')}" + f" {'==' if usage.get('input_tokens') == ref_usage.get('input_tokens') else '!='}" + f" {ref_usage.get('input_tokens')}" + ) + lines.append(f"output_tokens: {usage.get('output_tokens')} == 0") + max_delta = max(deltas, default=0.0) + lines.append(f"max numeric delta: {max_delta:.10e}") + lines.append("decision agreement: " + ("PASS" if not problems else "FAIL: " + "; ".join(problems[:4]))) + tolerance_ok = max_delta <= tolerance + lines.append(f"numeric tolerance <= {tolerance:g}: " + ("PASS" if tolerance_ok else "FAIL")) + tokens_exact = ( + not isinstance(reference, tuple) + and worker_response["usage"] == reference["usage"] + and worker_response["usage"].get("output_tokens") == 0 + ) + lines.append("token accounting: " + ("PASS" if tokens_exact else "FAIL")) + return lines, not problems and tolerance_ok and tokens_exact + + +def export_allowed(ok, worker_response, reference): + """The oracle freezes a served 200 response, never a rejection. + + A comparison can pass on an agreed 422 rejection, but that path has no + answers or token ids to freeze — the worker side is a synthetic empty + response with ``-1`` usage — so only a passing non-rejected case may + overwrite the fixture. + """ + + return ( + ok + and not isinstance(reference, tuple) + and "worker rejected the request" not in worker_response + ) + + +def export_oracle(body, prepared, probabilities, worker_response, path): + """Freeze the oracle fixture for the future native worker.""" + + oracle = { + "request": body, + "token_ids": {q: list(prepared.sequences[q]) for q in sorted(prepared.sequences)}, + "probabilities": {q: list(values) for q, values in sorted(probabilities.items())}, + "answers": worker_response["answers"], + "usage": worker_response["usage"], + } + Path(path).write_text(json.dumps(oracle, indent=1) + "\n", encoding="utf-8") + return oracle + + +def main(argv=None): + args = parse_args(argv) + sys.path.insert(0, str(HERE.parents[1] / "src")) + from frontend.d3 import decide + from models.d3.engine import D3Executor + + executor = D3Executor( + args.model, + device=args.device, + batch_size=args.batch_size, + verify=args.verify, + ) + body = json.loads(Path(args.fixture).read_text(encoding="utf-8")) + failed = 0 + try: + raw = json.dumps(body).encode("utf-8") + try: + worker_response = decide(raw, executor) + except Exception as exc: # noqa: BLE001 - reported through the comparison + status = getattr(exc, "status", 500) + message = str(exc) + rejection_body = getattr(exc, "body", None) + if rejection_body is None: + rejection_body = {"detail": message} + worker_response = { + "answers": {}, + "usage": {"input_tokens": -1, "output_tokens": -1}, + "worker rejected the request": {"status": status, "body": rejection_body}, + } + reference = reference_decide(executor.model, body) + lines, ok = compare_case(Path(args.fixture).name, worker_response, reference, args.tolerance) + print("\n".join(lines)) + if args.export and export_allowed(ok, worker_response, reference): + prepared = executor.prepare(body.get("state"), dict(body["questions"])) + probabilities, _ = executor.run(prepared) + export_oracle(body, prepared, probabilities, worker_response, args.export) + print(f"oracle exported: {args.export}") + elif args.export: + print("oracle export skipped: no passing 200-response comparison to freeze") + print() + if not ok: + failed += 1 + finally: + executor.close() + print("compare_reference: ok" if not failed else "compare_reference: FAILED") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/recipe/d3/example-request.json b/recipe/d3/example-request.json new file mode 100644 index 00000000..f3eac261 --- /dev/null +++ b/recipe/d3/example-request.json @@ -0,0 +1,27 @@ +{ + "model": "d3", + "state": "The customer's card was charged twice for one order and they opened a support ticket reporting the duplicate charge.", + "questions": { + "refund": { + "type": "choice", + "instructions": "Decide the refund action.", + "criteria": { + "refund": "Refund the duplicate charge now", + "wait": "Wait for manual review" + } + }, + "urgent": { + "type": "noul", + "instructions": "Is this urgent for the customer?", + "criteria": { + "false": "Routine duplicate charge", + "true": "Customer hardship reported" + } + }, + "severity": { + "type": "score", + "instructions": "Rate the severity of the billing error.", + "criteria": ["minor", "moderate", "major", "critical"] + } + } +} diff --git a/recipe/d3/reference.md b/recipe/d3/reference.md new file mode 100644 index 00000000..7c81ac73 --- /dev/null +++ b/recipe/d3/reference.md @@ -0,0 +1,268 @@ +# D3 reference worker + +This recipe serves D3 (Decision 3.0) through the Python reference worker and +the existing Rust frontend. The model-side implementation is under +[`src/models/d3/`](../../src/models/d3/), and the worker is +[`src/frontend/d3.py`](../../src/frontend/d3.py). + +Run the commands from the repository root on Linux with an NVIDIA GPU and +Python 3.12. The recipe records the procedure; measured results and verbatim +run output live in the pull request description, not here. + +## Scope + +The slice is: + +- package: [`vllm-sr/d3`](https://huggingface.co/vllm-sr/d3) at the pinned + revision below; +- the Python reference worker behind the Rust frontend, executing the + checkpoint's own pinned runtime (`d3_runtime.py` and siblings) after a + full-hash manifest gate; +- text/JSON state and the three typed kinds — `choice`, `noul`, `score` — + in one request; +- BF16 backbone with FP32 readout, noncausal full attention, last-position + pooling, one forward pass per question. + +Out of scope for this slice: `images`/`videos` (rejected with 422; upstream +supports them per-item at 8,000,000 bytes per image and 32,000,000 per +video), upstream's FastAPI server features (API key, request ids, server +timing), and any native Rust/CUDA executor. + +## Architecture and ownership + +~~~text +Rust frontend + -> Python worker + -> admit: envelope (protocol), pinned-runtime question errors (preprocess) + -> verify: manifest identity pin (repo id + runtime digests), full + SHA-256 of the manifest runtime table, then import it (engine) + -> execute: the checkpoint's own d3_runtime prepare/run/respond + -> finish: response validation + embedded-failure warning (postprocess) +~~~ + +The ownership boundary follows +[`docs/architecture.md`](../../docs/architecture.md) and the +[model contract](../../src/models/d3/README.md): + +- `protocol.py` owns the five-field envelope and the slice-scoped media + refusal; +- `preprocess.py` owns the pre-execution 422 mapping + (`invalid_questions` map, plain `max_length_exceeded` message); +- `engine.py` owns the pinned runtime and the manifest gate stronger than + the upstream default; +- `postprocess.py` owns the pure answer formulas and the warning on + embedded `invalid_model_output` answers (partial success stays HTTP 200 + but never passes silently); +- `frontend.d3` owns HTTP lifecycle and orchestration; +- The Rust frontend forwards `/v1/systemone` and `/health` unchanged. + +## Pinned artifacts + +| Artifact | Value | +| --- | --- | +| Package | [vllm-sr/d3](https://huggingface.co/vllm-sr/d3) | +| Pinned revision | `cbff3a3cb0b074b6585fe9f747b9f9f5557a0e75` | +| Manifest | `d3-package-manifest/1`, built 2026-10-11T03:30:34Z | +| Runtime gate | `repo_id` + `runtime.files_sha256` pinned per revision in `engine.py`, 9 files, full SHA-256 before import, no unlisted importables in the package root | +| License | Apache-2.0 (package metadata) | + +The pinned revision predates later upstream commits (the repository's +`lastModified` is later than the manifest build time), which is exactly why +the revision pin plus the two manifest hash tables exist. Re-check before +changing the pin: + +~~~sh +curl -sS "https://huggingface.co/api/models/vllm-sr/d3/revision/cbff3a3cb0b074b6585fe9f747b9f9f5557a0e75" | jq -r '.sha' +.venv/bin/python - <<'PY' +import json, hashlib, pathlib +root = pathlib.Path("weights/d3") +manifest = json.loads((root / "MODEL_MANIFEST.json").read_text()) +for name, digest in manifest["runtime"]["files_sha256"].items(): + actual = hashlib.sha256((root / name).read_bytes()).hexdigest() + assert actual == digest, f"{name}: {actual} != {digest}" +print("runtime table: 9 files verified") +PY +~~~ + +## Install and prepare the artifacts + +The environment installs from [`recipe/d3/requirements.txt`](requirements.txt), +a verbatim copy of the pinned package's own file (SHA-256 `57a19384…be8a32`, +matching the manifest's `runtime.files_sha256` entry; the worker's startup +gate re-verifies the package copy against the same digest). PyTorch wheels +must match the host CUDA runtime; the file's comment records that upstream +validated with transformers 5.17.0 on ROCm (torch 2.13). + +~~~sh +python3.12 -m venv .venv +.venv/bin/python -m pip install --upgrade pip +.venv/bin/python -m pip install -r recipe/d3/requirements.txt pytest +~~~ + +~~~sh +export D3_PKG="$PWD/weights/d3" +export D3_REV=cbff3a3cb0b074b6585fe9f747b9f9f5557a0e75 +mkdir -p weights +.venv/bin/hf download vllm-sr/d3 --revision "$D3_REV" --local-dir "$D3_PKG" +~~~ + +The package is about 52 GB (26.09B BF16 parameters plus the FP32 +(255, 5120) readout). The worker refuses to start when +`MODEL_MANIFEST.json` is missing, when the manifest is not the pinned +revision's own (a self-consistent manifest from a later snapshot carries a +different `repo_id` or runtime table), when any of the 9 runtime files +fails its full-hash check, or when the package root holds an importable +entry the runtime table does not name (an extra `*.py` file or +`__init__.py` package, which the `sys.path` setup would let shadow a +dependency) — that gate runs before any package code is imported. + +## Start the worker + +~~~sh +PYTHONPATH=src .venv/bin/python -m frontend.d3 \ + --model "$D3_PKG" --device cuda:0 --port 8000 +~~~ + +The worker verifies the runtime table, loads the package through the +pinned runtime's own `verify="fast"` path (pass `--verify full` to hash +every file, a minutes-level operation over 52 GB), compiles the kernels via +warmup, and only then binds the port: + +~~~sh +curl -sS -i http://127.0.0.1:8000/health +~~~ + +Expected shape (the pinned package reports no input limit): + +~~~json +{"status":"ready","model":"d3","max_input_tokens":null,"modalities":["text","image","video"]} +~~~ + +## Example request + +The complete runnable request is in +[`example-request.json`](example-request.json): one text state with one +`choice`, one `noul` and one `score` question. + +~~~sh +curl -sS http://127.0.0.1:8000/v1/systemone \ + -H 'Content-Type: application/json' \ + --data-binary @recipe/d3/example-request.json +~~~ + +Expected response shape: + +~~~json +{ + "model": "d3", + "answers": { + "refund": {"type": "choice", "choice": "...", "probabilities": {"...": 0.0}, "confidence": 0.0}, + "urgent": {"type": "noul", "noul": 0.0}, + "severity": {"type": "score", "score": 0.0, "probabilities": {"0": 0.0}, "confidence": 0.0, + "legend": {"0": "minor", "1": "moderate", "2": "major", "3": "critical"}} + }, + "usage": {"input_tokens": 0, "output_tokens": 0} +} +~~~ + +This is a serving smoke test, not an accuracy result; the exact numbers come +from the checkpoint and are recorded by the parity tool below. + +## Start and verify the Rust frontend + +~~~sh +cargo build --release --locked +OMNI_JEV_BIND=127.0.0.1:8080 \ +OMNI_JEV_BACKEND_URL=http://127.0.0.1:8000 \ + target/release/omni-jev +~~~ + +Send the same request through port 8080 and compare byte-for-byte: + +~~~sh +curl -fsS http://127.0.0.1:8000/v1/systemone \ + -H 'Content-Type: application/json' \ + --data-binary @recipe/d3/example-request.json > /tmp/d3-worker.json +curl -fsS http://127.0.0.1:8080/v1/systemone \ + -H 'Content-Type: application/json' \ + --data-binary @recipe/d3/example-request.json > /tmp/d3-frontend.json +cmp /tmp/d3-worker.json /tmp/d3-frontend.json +~~~ + +Worker responses are forwarded unchanged. Frontend-generated 502/504 +responses are frontend errors, not model responses. + +## Error behavior + +The request, response and error contract — the three error layers, the +partial-success warning, answer shapes and usage counting — is the +[model contract](../../src/models/d3/README.md). In short: transport errors +follow the shared worker discipline; invalid or over-limit questions reject +the whole request with 422 (`{"invalid_questions": {...}}` or a plain +`detail` naming the token count); a question with undecodable model output +keeps its 200 with the failure embedded and a worker warning logged. + +## Repository-side validation + +~~~sh +PYTHONPATH=src .venv/bin/python -m pytest tests/d3 -q +PYTHONPATH=src .venv/bin/python -m compileall -q src/models/d3 src/frontend/d3.py tests/d3 +.venv/bin/python recipe/d3/test_compare_reference.py +cargo fmt --all --check +cargo clippy --workspace --locked --all-targets -- -D warnings +cargo test --workspace --locked +cargo build --workspace --release --locked +~~~ + +Run these from the repository root. The recorded outputs of a run belong in +the pull request description, not in this recipe. + +## Reference parity (author-side, opt-in) + +[`compare_reference.py`](compare_reference.py) drives one fixture through +two paths that share the loaded pinned model: the upstream +`d3_server.py` decide body (prepare, 422 admission, run, respond) and the +production `frontend.d3.decide` pipeline. The criteria are exact decision +agreement including embedded `invalid_model_output` shapes, equal token +accounting, and a numeric tolerance of 1e-6. `--export` freezes the oracle +fixture (token ids, probabilities, answers) for the future native worker, +and writes it only after a passing, non-rejected comparison — a failed run +or an agreed 422 never touches the existing fixture. + +~~~sh +PYTHONPATH=src .venv/bin/python recipe/d3/compare_reference.py \ + --model "$D3_PKG" --device cuda:0 \ + --export tests/d3/fixtures/oracle.json +~~~ + +Run it by hand on a GPU host and record the verbatim output, host and +package revision in the pull request description. The flash-linear-attention +kernels and the PyTorch reference GDN implementation are not assumed to +agree numerically; measure the delta on this fixture before freezing any +tolerance for a native worker. The exported oracle fixture +(`tests/d3/fixtures/oracle.json`) is the frozen comparison baseline for the +stage-3 native worker and is committed with the integration. + +## Troubleshooting and limits + +- The package directory must be the pinned revision's download; the worker + refuses missing manifests, foreign manifests (wrong `repo_id` or runtime + table) and any runtime-file hash mismatch at startup. +- Upstream moved after the manifest was built — do not track `main` of the + repository; re-pin deliberately and re-record the revision and the pinned + runtime table (`PINNED_RUNTIME_SHA256`) in `src/models/d3/engine.py`. +- `images`/`videos` non-empty lists are rejected with 422 in this slice; + the 8 MiB body cap is slice-scoped and will re-align to the upstream + per-item budgets when media opens (stage 2). +- `/health` reports the loaded model's modalities even while this slice + refuses media; the 422 marks the serving boundary, not the checkpoint's + capability. +- `/health` has no `authentication` field: this worker has no API key + (a deliberate difference from upstream, which reads `DECISION_API_KEY`). +- Without optional `causal_conv1d`, the pinned runtime falls back to its + PyTorch reference implementation of that op (the runtime prints a notice); + correctness is unaffected, only speed. +- The noncausal full-attention hook inside the pinned runtime is adapted + from perplexity-ai/pplx-decider-v1.1-27b (Apache-2.0). +- GPU evidence for the supported-models table is recorded in the + integration pull request, not in this recipe. diff --git a/recipe/d3/requirements.txt b/recipe/d3/requirements.txt new file mode 100644 index 00000000..f5d3847b --- /dev/null +++ b/recipe/d3/requirements.txt @@ -0,0 +1,18 @@ +# d3 runtime. Validated with transformers 5.17.0 on ROCm (torch 2.13). +torch>=2.8 +transformers==5.17.0 +accelerate>=1.10 +safetensors>=0.6 +huggingface_hub>=1.0 +# GPU kernels for the Gated DeltaNet layers; without them transformers runs its slower PyTorch reference. +flash-linear-attention==0.5.2 +einops +# Image inputs: PIL decodes the images; the processor's torchvision backend resizes them as in evaluation +# (install the torchvision build that matches torch; without it transformers falls back to a PIL resize). +pillow>=10 +torchvision +# Video files: OpenCV decodes them with FFmpeg (frame arrays need no decoder). +opencv-python-headless +# d3_server.py only: +# fastapi +# uvicorn diff --git a/recipe/d3/test_compare_reference.py b/recipe/d3/test_compare_reference.py new file mode 100644 index 00000000..c77927b6 --- /dev/null +++ b/recipe/d3/test_compare_reference.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +"""Regression tests for compare_reference.py. + +Checks the pure comparison and export helpers without the pinned d3 package, +weights or a GPU: the PASS/FAIL verdicts of compare_case, embedded-error +shape agreement, the oracle export gate, and the oracle export structure. + + python recipe/d3/test_compare_reference.py +""" + +from __future__ import annotations + +import json +import sys +import tempfile +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE)) + +import compare_reference as tool + + +def worker_response(answers, input_tokens=17): + return { + "model": "d3", + "answers": answers, + "usage": {"input_tokens": input_tokens, "output_tokens": 0}, + } + + +CHOICE = { + "type": "choice", + "choice": "wait", + "probabilities": {"refund": 0.25, "wait": 0.75}, + "confidence": 0.18872187554086717, +} +NOUL = {"type": "noul", "noul": 0.75} +SCORE = { + "type": "score", + "score": 1.0, + "probabilities": {"0": 0.25, "1": 0.5, "2": 0.25}, + "confidence": 0.18872187554086717, + "legend": {"0": "minor", "1": "moderate", "2": "major"}, +} + + +def reference(): + return worker_response({"refund": dict(CHOICE), "urgent": dict(NOUL), "severity": dict(SCORE)}) + + +def main() -> None: + results: list[tuple[str, bool, str]] = [] + + def check(name: str, condition: bool, detail: str = "") -> None: + results.append((name, condition, detail)) + + # 1. An exactly matching worker response passes every verdict. + lines, ok = tool.compare_case("fixture", worker_response( + {"refund": dict(CHOICE), "urgent": dict(NOUL), "severity": dict(SCORE)} + ), reference()) + report = "\n".join(lines) + check( + "exact match passes", + ok + and "decision agreement: PASS" in report + and "numeric tolerance <= 1e-06: PASS" in report + and "token accounting: PASS" in report, + report, + ) + + # 2. A tiny delta stays inside the declared tolerance. + drifted = dict(CHOICE) + drifted["probabilities"] = {"refund": 0.25 + 1e-9, "wait": 0.75 - 1e-9} + drifted["confidence"] = CHOICE["confidence"] + 1e-9 + lines, ok = tool.compare_case( + "fixture", worker_response({"refund": drifted, "urgent": dict(NOUL), "severity": dict(SCORE)}), + reference(), + ) + check("delta within tolerance passes", ok, "\n".join(lines)) + + # 3. Each failure mode fails on its own. + flipped = dict(CHOICE) + flipped["choice"] = "refund" + lines, ok = tool.compare_case( + "fixture", worker_response({"refund": flipped, "urgent": dict(NOUL), "severity": dict(SCORE)}), + reference(), + ) + check( + "decision mismatch fails", + not ok and "decision agreement: FAIL" in "\n".join(lines), + "\n".join(lines), + ) + + lines, ok = tool.compare_case( + "fixture", worker_response( + {"refund": dict(CHOICE), "urgent": dict(NOUL), "severity": dict(SCORE)}, + input_tokens=18, + ), reference(), + ) + check( + "token mismatch fails", + not ok and "token accounting: FAIL" in "\n".join(lines) + and "input_tokens: 18 != 17" in "\n".join(lines), + "\n".join(lines), + ) + + reordered = {"urgent": dict(NOUL), "refund": dict(CHOICE), "severity": dict(SCORE)} + lines, ok = tool.compare_case("fixture", worker_response(reordered), reference()) + check( + "answer order mismatch fails", + not ok and "answer order differs" in "\n".join(lines), + "\n".join(lines), + ) + + # 4. Embedded invalid_model_output shapes must agree between the paths. + bad = { + "type": "noul", + "error": "invalid_model_output", + "message": "need one finite non-negative probability per option", + } + lines, ok = tool.compare_case( + "fixture", worker_response({"refund": dict(bad), "urgent": dict(NOUL), "severity": dict(SCORE)}), + worker_response({"refund": dict(bad), "urgent": dict(NOUL), "severity": dict(SCORE)}), + ) + check("matching embedded errors pass", ok, "\n".join(lines)) + + other_message = dict(bad) + other_message["message"] = "different failure" + lines, ok = tool.compare_case( + "fixture", worker_response({"refund": dict(other_message), "urgent": dict(NOUL), "severity": dict(SCORE)}), + worker_response({"refund": dict(bad), "urgent": dict(NOUL), "severity": dict(SCORE)}), + ) + check( + "embedded error mismatch fails", + not ok and "embedded error shape differs" in "\n".join(lines), + "\n".join(lines), + ) + + # 5. Rejection agreement: matching 422s pass, mismatches fail. + lines, ok = tool.compare_case("fixture", worker_response({}), ("422", "too long")) + check( + "worker answering a rejected request fails", + not ok and "reference rejected the request: too long" in "\n".join(lines), + "\n".join(lines), + ) + + rejected = worker_response({}) + rejected["worker rejected the request"] = { + "status": 422, + "body": {"invalid_questions": {"bad": "unsupported question type"}}, + } + lines, ok = tool.compare_case( + "fixture", rejected, ("422", {"invalid_questions": {"bad": "unsupported question type"}}) + ) + check( + "matching 422 rejections pass", + ok and "rejection agreement: PASS" in "\n".join(lines), + "\n".join(lines), + ) + + wrong_body = dict(rejected) + wrong_body["worker rejected the request"] = { + "status": 422, + "body": {"detail": "different"}, + } + lines, ok = tool.compare_case( + "fixture", wrong_body, ("422", {"invalid_questions": {"bad": "unsupported question type"}}) + ) + check( + "mismatched rejection bodies fail", + not ok and "rejection agreement: FAIL" in "\n".join(lines), + "\n".join(lines), + ) + + non_422 = dict(rejected) + non_422["worker rejected the request"] = {"status": 500, "body": {"detail": "x"}} + lines, ok = tool.compare_case("fixture", non_422, ("422", "too long")) + check( + "non-422 worker rejection fails", + not ok and "rejection agreement: FAIL" in "\n".join(lines), + "\n".join(lines), + ) + + # 6. The oracle export freezes the full stage-3 comparison surface. + body = {"model": "d3", "state": "s", "questions": {"q": {"type": "noul"}}} + + class Prepared: + sequences = {"q": [1, 2, 3]} + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "oracle.json" + oracle = tool.export_oracle( + body, + Prepared(), + {"q": [0.25, 0.75]}, + worker_response({"q": dict(NOUL)}), + path, + ) + reread = json.loads(path.read_text(encoding="utf-8")) + check( + "oracle export round-trips", + oracle == reread + and oracle["token_ids"] == {"q": [1, 2, 3]} + and oracle["probabilities"] == {"q": [0.25, 0.75]} + and oracle["answers"]["q"]["noul"] == 0.75 + and oracle["usage"] == {"input_tokens": 17, "output_tokens": 0}, + json.dumps(oracle)[:200], + ) + + # 7. The oracle is frozen only for a passing, non-rejected comparison. + served = worker_response({"q": dict(NOUL)}) + check( + "export allowed for a passing 200 case", + tool.export_allowed(True, served, worker_response({"q": dict(NOUL)})), + ) + check( + "export refused for a failed comparison", + not tool.export_allowed(False, served, worker_response({"q": dict(NOUL)})), + ) + mutual_422 = worker_response({}) + mutual_422["worker rejected the request"] = {"status": 422, "body": {"detail": "too long"}} + check( + "export refused for an agreed 422 (nothing to freeze)", + not tool.export_allowed(True, mutual_422, ("422", "too long")), + ) + check( + "export refused when only the worker rejected", + not tool.export_allowed(True, mutual_422, worker_response({"q": dict(NOUL)})), + ) + + failed = 0 + for name, passed, detail in results: + print(f" {'ok ' if passed else 'FAIL'} {name}") + if not passed: + failed += 1 + print(" " + detail.strip().replace("\n", "\n ")[:400]) + print( + "compare_reference tests: ok" + if not failed + else f"compare_reference tests: {failed} FAILED" + ) + sys.exit(1 if failed else 0) + + +if __name__ == "__main__": + main() diff --git a/src/frontend/d3.py b/src/frontend/d3.py new file mode 100644 index 00000000..b9fcae38 --- /dev/null +++ b/src/frontend/d3.py @@ -0,0 +1,190 @@ +"""Python reference worker for d3 (Decision 3.0). + +The Rust frontend remains the public transport layer. This worker owns only +HTTP lifecycle and request orchestration; wire admission lives in +models.d3.protocol, prepared-error admission in models.d3.preprocess, the +pinned runtime and its manifest verification in models.d3.engine, and +response finishing in models.d3.postprocess. + +The error surface has three layers: transport errors follow the shared +Python-worker discipline; a request with an invalid or over-limit question is +rejected whole with HTTP 422 before any execution (nothing is truncated); and +a question whose model output cannot be decoded keeps its HTTP 200 response +with the failure embedded in that question's answer — plus a worker warning, +because a silent partial failure is otherwise undetectable. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import socket +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any + +from models.d3.engine import VERIFY_MODES, D3Executor +from models.d3.postprocess import finish_response +from models.d3.preprocess import reject_pre_execution_errors +from models.d3.protocol import MAX_BODY, RequestError, parse_body, parse_request + +LOG = logging.getLogger(__name__) + + +def decide(raw: bytes, executor: D3Executor) -> dict[str, Any]: + """Admit, execute, and finish one complete request.""" + + request = parse_request(parse_body(raw)) + questions = dict(request.questions) + try: + prepared = executor.prepare(request.state, questions) + except ValueError as exc: + # The pinned runtime is the validation authority for what the wire + # layer cannot foresee (envelope-level question problems); its + # rejections are client errors, not worker failures. + raise RequestError(f"request rejected by the pinned runtime: {exc}", 422) from exc + reject_pre_execution_errors(prepared) + probabilities, tokens = executor.run(prepared) + response = executor.respond(prepared, probabilities, tokens) + return finish_response(tuple(questions), response) + + +class WorkerServer(ThreadingHTTPServer): + daemon_threads = False + + def __init__(self, address, executor: D3Executor): + self.executor = executor + self.inference_lock = threading.Lock() + self._executor_closed = False + super().__init__(address, Handler) + + def server_close(self): + super().server_close() + if not self._executor_closed: + self.executor.close() + self._executor_closed = True + + +class Handler(BaseHTTPRequestHandler): + def setup(self): + super().setup() + self.connection.settimeout(15) + + def log_message(self, format, *args): + # Do not log paths, prompts, or arbitrary request headers. + pass + + def send_json(self, status: int, value: dict): + raw = json.dumps(value, ensure_ascii=False, allow_nan=False).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + try: + self.wfile.write(raw) + except (BrokenPipeError, ConnectionResetError): + pass + + def _send_request_error(self, exc: RequestError): + if exc.body is not None: + self.send_json(exc.status, exc.body) + else: + self.send_json(exc.status, {"detail": str(exc)}) + + def do_GET(self): + if self.path == "/health": + executor = self.server.executor + self.send_json( + 200, + { + "status": "ready", + "model": executor.model_name, + "max_input_tokens": executor.max_length, + "modalities": list(executor.modalities), + }, + ) + else: + self.send_json(404, {"detail": "unknown route"}) + + def do_POST(self): + if self.path != "/v1/systemone": + self.send_json(404, {"detail": "unknown route"}) + return + if self.headers.get("Transfer-Encoding"): + self.send_json(411, {"detail": "Content-Length is required; chunked requests are unsupported"}) + return + lengths = self.headers.get_all("Content-Length", []) + if len(lengths) != 1: + self.send_json(411, {"detail": "one Content-Length is required"}) + return + try: + length = int(lengths[0]) + except ValueError: + self.send_json(400, {"detail": "invalid Content-Length"}) + return + if length < 0 or length > MAX_BODY: + self.send_json(413, {"detail": "request exceeds body limit"}) + return + if self.headers.get_content_type() != "application/json": + self.send_json(415, {"detail": "Content-Type must be application/json"}) + return + if not self.server.inference_lock.acquire(blocking=False): + self.send_json(503, {"detail": "worker busy"}) + return + try: + raw = self.rfile.read(length) + if len(raw) != length: + self.send_json(400, {"detail": "incomplete body"}) + return + self.send_json(200, decide(raw, self.server.executor)) + except RequestError as exc: + self._send_request_error(exc) + except (TimeoutError, socket.timeout): + self.send_json(408, {"detail": "request body timed out"}) + except Exception as exc: + LOG.error("d3 inference failed: %s", type(exc).__name__) + self.send_json(500, {"detail": "inference failed"}) + finally: + self.server.inference_lock.release() + + +def parse_args(argv=None): + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--model", required=True, help="pinned d3 package directory") + parser.add_argument("--device", default=None) + parser.add_argument("--batch-size", type=int, default=8) + parser.add_argument("--verify", default="fast", choices=VERIFY_MODES) + parser.add_argument("--name", help="served model name (default: the package's model name)") + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8000) + return parser.parse_args(argv) + + +def main(argv=None): + args = parse_args(argv) + logging.basicConfig(level=logging.INFO) + executor = D3Executor( + args.model, + device=args.device, + batch_size=args.batch_size, + verify=args.verify, + model_name=args.name, + ) + server = None + try: + executor.warmup() + server = WorkerServer((args.host, args.port), executor) + LOG.info("%s worker ready on %s:%s", executor.model_name, args.host, args.port) + server.serve_forever() + except KeyboardInterrupt: + pass + finally: + if server is not None: + server.server_close() + else: + executor.close() + + +if __name__ == "__main__": + main() diff --git a/src/models/d3/README.md b/src/models/d3/README.md new file mode 100644 index 00000000..40faf986 --- /dev/null +++ b/src/models/d3/README.md @@ -0,0 +1,140 @@ +# D3 reference contract + +This directory contains the first System1-Omni integration slice for +D3 (Decision 3.0, [`vllm-sr/d3`](https://huggingface.co/vllm-sr/d3)). It is a +reference-serving boundary only: a Python worker that runs the checkpoint's +own pinned runtime for text state and the three typed question kinds +(`choice`, `noul`, `score`) on `/v1/systemone`. Native CUDA execution and +multimodal inputs are not part of this slice. + +## Pinned artifacts + +| Artifact | Value | +| --- | --- | +| Package | [vllm-sr/d3](https://huggingface.co/vllm-sr/d3) | +| Pinned revision | `cbff3a3cb0b074b6585fe9f747b9f9f5557a0e75` | +| Manifest schema | `d3-package-manifest/1` (built 2026-10-11T03:30:34Z) | +| Model identity | `model_name: "d3"`, 26.09B parameters (manifest-reported) | +| License | Apache-2.0 (Qwen3.8-27B base and d3 weights/readout/runtime) | +| Runtime dependencies | the checkpoint's own `requirements.txt` (torch>=2.8, transformers==5.17.0, flash-linear-attention==0.5.2) | + +The checkpoint ships its runtime as flat Python files next to the weights +(`d3_runtime.py` and siblings). Upstream `verify_package` walks only the +39-entry `files_sha256` table and, in fast mode, skips hashing files over +64 MiB — so by default it never verifies the runtime table it executes. +This worker additionally hashes every file of +`runtime.files_sha256` (8 runtime modules plus `requirements.txt`) in full +**before importing anything** from the package directory, and refuses to +serve packages without a manifest. It also rejects importable entries the +runtime table does not name — an extra `*.py` file or `__init__.py` package +in the package root — because that root is placed first on `sys.path`, where +such an entry would shadow a dependency of the pinned runtime. Because hash +agreement with a manifest +proves only internal consistency — a later snapshot ships a self-consistent +table of its own — the gate first requires the manifest to carry exactly +the `repo_id` and `runtime.files_sha256` table checked in for the pinned +revision in `engine.py`; a foreign manifest is refused before any file is +hashed. Shard and config verification uses the +upstream `verify` mechanism (`fast` default, `full` via CLI). + +## Current serving scope + +- The request envelope is upstream's five-field form: + `{"model", "state", "questions", "images", "videos"}`. Unknown fields are + rejected with HTTP 422. `model` is accepted and ignored. +- `state` is any JSON value; the pinned runtime renders it (empty and `""` + render as `(empty)`). +- `questions` is a nonempty mapping of nonempty string IDs to questions of + three kinds, validated by the pinned runtime: + - `choice`: `criteria` maps 1–255 nonempty string keys to descriptions + (any JSON value); + - `noul`: `criteria` is optional and may only describe `"false"` and + `"true"`; + - `score`: `criteria` is an ordered list of 2–10 distinct nonempty levels. +- `images`/`videos` are accepted only as absent, `null` or empty lists; + anything else is rejected with HTTP 422. Serving media is stage 2; when it + opens, the limits re-align to the upstream per-item budgets (8,000,000 + bytes per image, 32,000,000 per video) instead of this slice's 8 MiB + transport cap. +- No token generation: one forward pass per question scores 255 single-token + answer codes at the last prompt position. + +## Answers + +Each answer is built by the pinned runtime from per-option probabilities +(mirrored as pure functions in `postprocess.py` for the CPU contract tests): + +- `choice`: `{"type", "choice", "probabilities", "confidence"}` — + ties go to the earliest option in request order; `probabilities` are + renormalized per option. +- `noul`: `{"type": "noul", "noul": p_true}` — the renormalized probability + of `true`; no probabilities or confidence. +- `score`: `{"type", "score", "probabilities", "confidence", + "legend"}` — `score` is the 0-based expectation over the renormalized + level distribution, the answer keys are the stringified level indices, and + `legend` maps each key to its canonical level text. +- `confidence` is one minus the normalized entropy clipped to [0, 1] + (1.0 for fewer than two options) — the upstream "Decision 2.0 measure", + not max-p. + +`usage.input_tokens` sums each runnable question's prompt tokens; +`usage.output_tokens` is 0. The pinned package reports no +`max_input_tokens` limit (the manifest records `null`), so +`/health` reports `max_input_tokens: null` and `max_length_exceeded` cannot +trigger for it; the admission path stays for packages that set a limit. + +## Errors + +Three layers, matching the upstream server: + +1. Transport follows the shared Python-worker discipline: `400` (unusable + JSON body, mismatched or invalid `Content-Length`), `408` body timeout, + `411` without exactly one `Content-Length`, `413` over the slice-scoped + 8 MiB body limit, `415` non-JSON content type, `503` while another + inference holds the lock, `404` unknown routes, `500` anything else + (generic detail, no request content in logs). +2. Before execution, the pinned runtime's per-question validation rejects + the whole request with HTTP 422: invalid questions produce + `{"invalid_questions": {name: message}}` in request order, and an + over-limit question produces a plain `detail` naming the actual token + count and limit — nothing is ever truncated. `max_length_exceeded` wins + over invalid questions. +3. After execution, a question whose probabilities cannot be decoded + (non-finite, negative, or all zero) keeps its HTTP 200 response with + `{"type", "error": "invalid_model_output", "message"}` embedded in that + question's answer; the other questions are still answered. This partial + success is new to the repository, so the worker logs one warning per + embedded failure — naming the question and the message, never request + content — to keep it detectable. + +`/health` returns `{"status", "model", "max_input_tokens", "modalities"}` +after loading and warmup; the upstream `authentication` field is dropped +(this worker has no API key). + +## Ownership boundary + +- `protocol.py` owns the envelope: body parsing, unknown-field and + slice-scoped media rejection, question-ID admission. +- `preprocess.py` owns the mapping of the pinned runtime's per-question + preparation errors to the two pre-execution 422 shapes. +- `engine.py` owns the pinned runtime, the identity pin and the pre-import + manifest gate (full runtime hashes, no unlisted importables), device + state, and warmup. +- `postprocess.py` owns the pure answer formulas, response finishing, and + the embedded-failure warning. +- `src/frontend/d3.py` owns HTTP lifecycle and orchestration. +- The Rust frontend owns transport and forwarding, not model state. + +The server-side error surface and envelope naming are recorded as this +repository's stance and remain coordinated with +[system1-omni#61](https://github.com/ThinkFlowLab/system1-omni/issues/61); +the noncausal full-attention hook inside the pinned runtime is adapted from +[perplexity-ai/pplx-decider-v1.1-27b](https://github.com/perplexity-ai/pplx-decider-v1.1-27b) +(Apache-2.0), as its module docstring records. + +The current tests are CPU-only and do not require model weights, torch or a +GPU (`PYTHONPATH=src python -m pytest tests/d3 -q`). The real worker path +additionally requires the pinned package and a CUDA-capable environment; +the procedure is [recipe/d3/reference.md](../../recipe/d3/reference.md) and +measured-run evidence is recorded in the integration pull request, not in +the recipe. diff --git a/src/models/d3/__init__.py b/src/models/d3/__init__.py new file mode 100644 index 00000000..4d727dbb --- /dev/null +++ b/src/models/d3/__init__.py @@ -0,0 +1,5 @@ +"""D3 (Decision 3.0) model contract and reference-serving adapters.""" + +from .protocol import MODEL_NAME + +__all__ = ["MODEL_NAME"] diff --git a/src/models/d3/engine.py b/src/models/d3/engine.py new file mode 100644 index 00000000..9327306c --- /dev/null +++ b/src/models/d3/engine.py @@ -0,0 +1,234 @@ +"""D3 executor for the pinned code-readout checkpoint. + +The executor owns the pinned runtime, the manifest verification that gates its +import, device state, and warmup. It consumes raw state/questions through the +upstream ``prepare``/``run``/``respond`` surface and stays free of HTTP +concerns; wire admission lives in protocol.py and preprocess.py. + +The checkpoint ships its own runtime as flat Python files next to the +weights. Those files are the code this process executes, so — stronger than +the upstream default, whose ``verify_package`` walks only the 39-entry +``files_sha256`` table and skips hashing files over 64 MiB in fast mode — the +executor hashes every file of the manifest's ``runtime.files_sha256`` table +in full before importing anything from the package directory, and refuses +unlisted importable entries (an extra ``*.py`` file or ``__init__.py`` +package) in the package root: that root goes first on ``sys.path``, so such +an entry would shadow a dependency the pinned runtime imports. Hash +agreement proves only internal consistency — a later snapshot ships a +self-consistent table of its own — so the gate first requires the manifest +to carry exactly the repo id and runtime digests checked in for +``REFERENCE_REVISION``. +""" + +from __future__ import annotations + +import hashlib +import json +import sys +import threading +from pathlib import Path +from typing import Any + +REFERENCE_REPO = "vllm-sr/d3" +REFERENCE_REVISION = "cbff3a3cb0b074b6585fe9f747b9f9f5557a0e75" +MANIFEST_NAME = "MODEL_MANIFEST.json" +MANIFEST_SCHEMA = "d3-package-manifest/1" +VERIFY_MODES = ("fast", "full", "none") +# The runtime.files_sha256 table of REFERENCE_REVISION, transcribed from the +# pinned package's manifest; repinning means updating this table together +# with REFERENCE_REVISION. +PINNED_RUNTIME_SHA256 = { + "d3_runtime.py": "6c049aa7d3da8da3738602a2812572879d0a47b03c28c3e33f36d7300257de89", + "d3_fast.py": "68a1fb9a2c3bde9ac2763bf504decae592ca12722698fbde1b39a31262955535", + "d3_kernels.py": "8f7534ca1e1846ebf7c4c73058002f3be4dbf0d5d426f0815150ae0cbe7677bd", + "modeling_d3.py": "f8f8db23ff58f1a990ffa24977d744a50e1b4446d02a64c22070967705b6dbef", + "pipeline_d3.py": "4f9297c7c74e596e9b67df91a1e6823cfada48bf781336f141974ff6a2f1038e", + "d3_server.py": "f9b8f2481461a074f469b4f58da3b0d7004824096528366d5b1f23187dd6e692", + "d3_engine.py": "086238942762df45047a32cd9a8e2bedd9b63167eb40be260c776de567979f53", + "requirements.txt": "57a1938448bc240bdf4fee5d1b9cbc8450109fa1f6cee11e94dc04f274be8a32", + "d3_format.py": "e5036154d2e54793f59b320c8726632957bc343c382bac26d825b263a6e9ce62", +} + + +def _sha256_safe(path: Path) -> str: + """Might encounter a large file, read in chunks""" + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def load_manifest(root: Path) -> dict[str, Any]: + """Read and schema-check the package manifest; refuse packages without one.""" + + path = Path(root) / MANIFEST_NAME + if not path.is_file(): + raise ValueError( + f"the package is missing {MANIFEST_NAME}; this worker only serves verified packages" + ) + manifest = json.loads(path.read_text(encoding="utf-8")) + if manifest.get("schema") != MANIFEST_SCHEMA: + raise ValueError(f"{MANIFEST_NAME} is not a {MANIFEST_SCHEMA} manifest") + return manifest + + +def verify_pinned_package(manifest: dict[str, Any]) -> None: + """Refuse any package that is not the pinned reference snapshot. + + A package whose files hash to its own manifest is only internally + consistent: a later snapshot rewrites its ``runtime.files_sha256`` and + would pass the full-hash gate on a changed baseline. The manifest must + therefore carry exactly the repo id and runtime digests checked in for + ``REFERENCE_REVISION`` before any file is hashed or imported. + """ + + repo_id = manifest.get("repo_id") + if repo_id != REFERENCE_REPO: + raise ValueError( + f"{MANIFEST_NAME} declares repo_id {repo_id!r}; this worker only " + f"serves {REFERENCE_REPO} at {REFERENCE_REVISION}" + ) + table = (manifest.get("runtime") or {}).get("files_sha256") + if table != PINNED_RUNTIME_SHA256: + raise ValueError( + f"{MANIFEST_NAME} runtime.files_sha256 is not the table pinned for " + f"{REFERENCE_REPO}@{REFERENCE_REVISION}; re-pin deliberately or " + "download the pinned revision" + ) + + +def verify_runtime_files(root: Path, manifest: dict[str, Any]) -> list[str]: + """Hash every runtime file of the manifest in full and return their names. + + Must run before any package code is imported: the runtime table names the + files this process is about to execute, so unlike the upstream shard + verification there is no size shortcut. It also refuses importable + entries the table does not name — the package root is about to go first + on ``sys.path``, where an extra ``*.py`` file or ``__init__.py`` package + would shadow one of the pinned runtime's dependencies. + """ + + table = (manifest.get("runtime") or {}).get("files_sha256") + if not isinstance(table, dict) or not table: + raise ValueError(f"{MANIFEST_NAME} has no runtime.files_sha256 table") + problems: list[str] = [] + for name, digest in sorted(table.items()): + if Path(name).name != name or name in (".", ".."): + raise ValueError(f"unsafe runtime file name {name!r}") + target = Path(root) / name + if not target.is_file(): + problems.append(f"missing runtime file {name}") + continue + if _sha256_safe(target) != digest: + problems.append(f"{name}: runtime SHA-256 differs from {MANIFEST_NAME}") + for entry in sorted(Path(root).iterdir()): + if entry.name in table: + continue + if entry.is_file() and entry.suffix == ".py": + problems.append(f"unexpected importable file {entry.name} in the package root") + elif entry.is_dir() and (entry / "__init__.py").is_file(): + problems.append(f"unexpected importable package {entry.name}/ in the package root") + if problems: + raise ValueError("runtime verification failed: " + "; ".join(problems[:8])) + return sorted(table) + + +class D3Executor: + """Load and drive one immutable pinned d3 runtime instance.""" + + model_name = "d3" + + def __init__( + self, + package: str | Path, + *, + device: str | None = None, + batch_size: int = 8, + verify: str = "fast", + model_name: str | None = None, + max_length: int | None = None, + readout_dtype: str | None = None, + permutation_average: bool = False, + ) -> None: + if verify not in VERIFY_MODES: + raise ValueError(f"verify must be one of {VERIFY_MODES}") + root = Path(package) + if not root.is_dir(): + raise ValueError(f"the d3 package directory does not exist: {root}") + manifest = load_manifest(root) + verify_pinned_package(manifest) + verify_runtime_files(root, manifest) + + # Import the just-verified runtime the way upstream d3_server.py does: + # the package directory goes first on sys.path so the flat runtime + # modules (d3_runtime, d3_format, ...) resolve to the checkpoint's own + # copies. One worker process serves one package. + root_str = str(root) + if root_str not in sys.path: + sys.path.insert(0, root_str) + import d3_runtime + + self.model = d3_runtime.D3.from_pretrained( + root, + device=device, + batch_size=batch_size, + verify=verify, + model_name=model_name, + max_length=max_length, + readout_dtype=readout_dtype, + permutation_average=permutation_average, + ) + self.model_name = self.model.model_name + self.max_length = self.model.max_length + self._image_unavailable = self.model.image_unavailable + self._video_unavailable = self.model.video_unavailable + self._lock = threading.RLock() + self._closed = False + + @property + def modalities(self) -> list[str]: + """The loaded model's input modalities, in upstream /health order.""" + + names = ["text", "image"] if self._image_unavailable is None else ["text"] + return names + ["video"] if self._video_unavailable is None else names + + def prepare(self, state: Any, questions: dict[str, Any]) -> Any: + """Validate and tokenize one request; per-question failures land in + ``prepared.errors`` instead of raising (envelope problems raise).""" + + with self._lock: + if self._closed: + raise RuntimeError("D3 executor is closed") + return self.model.prepare(state, questions) + + def run(self, prepared: Any) -> tuple[dict[str, list[float]], int]: + """Probabilities per runnable question and the input token count.""" + + with self._lock: + if self._closed: + raise RuntimeError("D3 executor is closed") + return self.model.run(prepared) + + def respond(self, prepared: Any, probabilities: dict[str, list[float]], tokens: int) -> dict: + """Build the System One response, embedding per-question model failures.""" + + with self._lock: + if self._closed: + raise RuntimeError("D3 executor is closed") + return self.model.respond(prepared, probabilities, tokens) + + def warmup(self) -> None: + """Compile and autotune the kernels before readiness.""" + + with self._lock: + if self._closed: + raise RuntimeError("D3 executor is closed") + self.model.warmup() + + def close(self) -> None: + with self._lock: + if self._closed: + return + self._closed = True + self.model = None diff --git a/src/models/d3/postprocess.py b/src/models/d3/postprocess.py new file mode 100644 index 00000000..8dadd45c --- /dev/null +++ b/src/models/d3/postprocess.py @@ -0,0 +1,170 @@ +"""Pure d3 answer reconstruction and response finishing. + +The runtime's ``respond`` builds each answer from per-option probabilities. +Its formulas are mirrored here as pure functions — first-max tie-breaking, +renormalization, the entropy confidence, the score expectation and the legend +canonicalization — so the CPU contract tests can lock them without weights +and the recipe oracle can re-derive them on a GPU host. ``finish_response`` +is the runtime-facing half: it passes the response through unchanged, warns +on every embedded ``invalid_model_output`` answer (a silent partial failure +is undetectable otherwise; the log never contains request content) and +rejects anything that breaks the response contract. +""" + +from __future__ import annotations + +import json +import logging +import math +from collections.abc import Sequence +from typing import Any + +LOG = logging.getLogger(__name__) + +INVALID_MODEL_OUTPUT = "invalid_model_output" + + +def describe(value: Any) -> str: + """Upstream's option-text rendering: strings pass, other JSON is dumped.""" + + return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) + + +def canonical(value: Any) -> str: + """Upstream's legend canonicalization (sorted keys, compact separators).""" + + return ( + value + if isinstance(value, str) + else json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + ) + + +def confidence(values: Sequence[float]) -> float: + """One minus the normalized entropy, clipped to [0, 1]. + + Upstream docstring: "the Decision 2.0 measure". Not max-p. + """ + + if len(values) < 2: + return 1.0 + entropy = -sum(p * math.log(p) for p in values if p > 0) + return max(0.0, min(1.0, 1.0 - entropy / math.log(len(values)))) + + +def renormalized(probabilities: Sequence[float]) -> list[float]: + """Validate and renormalize one probability per option (upstream to_answer).""" + + values = [float(value) for value in probabilities] + if ( + len(values) < 1 + or any(not math.isfinite(value) or value < 0 for value in values) + or sum(values) <= 0 + ): + raise ValueError("need one finite non-negative probability per option") + total = sum(values) + return [value / total for value in values] + + +def choice_answer(keys: Sequence[str], probabilities: Sequence[float]) -> dict[str, Any]: + """``{"type", "choice", "probabilities", "confidence"}``. + + Ties go to the earliest option in order (upstream's max over indices). + """ + + values = renormalized(probabilities) + if len(values) != len(keys): + raise ValueError("need one finite non-negative probability per option") + best = max(range(len(values)), key=values.__getitem__) + return { + "type": "choice", + "choice": keys[best], + "probabilities": dict(zip(keys, values)), + "confidence": confidence(values), + } + + +def noul_answer(probabilities: Sequence[float]) -> dict[str, Any]: + """``{"type": "noul", "noul": p_true}`` — no probabilities or confidence.""" + + values = renormalized(probabilities) + if len(values) != 2: + raise ValueError("need one finite non-negative probability per option") + return {"type": "noul", "noul": values[1]} + + +def score_answer(levels: Sequence[Any], probabilities: Sequence[float]) -> dict[str, Any]: + """``{"type", "score", "probabilities", "confidence", "legend"}``. + + The answer keys are the stringified level indices; ``score`` is the + 0-based expectation and ``legend`` maps each key to its canonical level. + """ + + values = renormalized(probabilities) + if len(values) != len(levels): + raise ValueError("need one finite non-negative probability per level") + keys = [str(index) for index in range(len(levels))] + return { + "type": "score", + "score": sum(index * value for index, value in enumerate(values)), + "probabilities": dict(zip(keys, values)), + "confidence": confidence(values), + "legend": {key: canonical(level) for key, level in zip(keys, levels)}, + } + + +def to_answer( + kind: str, + keys: Sequence[str], + levels: Sequence[Any] | None, + probabilities: Sequence[float], +) -> dict[str, Any]: + """Dispatch on the question kind (upstream product_answer shapes).""" + + if kind == "noul": + return noul_answer(probabilities) + if kind == "choice": + return choice_answer(keys, probabilities) + if kind == "score": + return score_answer(levels or (), probabilities) + raise ValueError(f"unsupported question type {kind!r}") + + +def finish_response(expected_names: Sequence[str], response: dict[str, Any]) -> dict[str, Any]: + """Pass the runtime response through, warning on embedded model failures. + + ``invalid_model_output`` answers stay in the HTTP 200 response (partial + success; the other questions are still answered) but each one logs a + warning naming the question and the failure — never request content. + Any other embedded error means admission failed to intercept it and is a + worker bug, not a client error. + """ + + if not isinstance(response, dict) or set(response) != {"model", "answers", "usage"}: + raise RuntimeError("the pinned runtime returned an unexpected response envelope") + answers = response["answers"] + if not isinstance(answers, dict) or tuple(answers) != tuple(expected_names): + raise RuntimeError("answers do not match request question order") + usage = response["usage"] + if ( + not isinstance(usage, dict) + or set(usage) != {"input_tokens", "output_tokens"} + or not isinstance(usage["input_tokens"], int) + or usage["input_tokens"] < 0 + or usage["output_tokens"] != 0 + ): + raise RuntimeError("the pinned runtime returned invalid usage accounting") + for name, answer in answers.items(): + if not isinstance(answer, dict) or "error" not in answer: + continue + if answer.get("error") == INVALID_MODEL_OUTPUT: + LOG.warning( + "question %s was answered with invalid_model_output: %s", + name, + answer.get("message"), + ) + else: + raise RuntimeError( + f"the pinned runtime embedded a pre-execution error for question {name!r}" + ) + return response diff --git a/src/models/d3/preprocess.py b/src/models/d3/preprocess.py new file mode 100644 index 00000000..7f8ea08b --- /dev/null +++ b/src/models/d3/preprocess.py @@ -0,0 +1,52 @@ +"""Admission of the pinned runtime's prepared-request errors. + +``D3.prepare`` validates every question and reports failures per question in +``prepared.errors`` instead of raising. The upstream server turns that map +into HTTP 422 before any model execution; this module owns the same mapping +for the reference worker: a ``max_length_exceeded`` question rejects the whole +request with its plain message (nothing is ever truncated), and invalid +questions reject it with an ``invalid_questions`` body preserving request +order. +""" + +from __future__ import annotations + +from typing import Any + +from .protocol import RequestError + +MAX_LENGTH_EXCEEDED = "max_length_exceeded" + + +def reject_pre_execution_errors(prepared: Any) -> None: + """Turn ``prepared.errors`` into request errors before any execution. + + ``max_length_exceeded`` wins over invalid questions, mirroring the + upstream server: its message names the actual token count and limit. + """ + + errors = getattr(prepared, "errors", None) + if not isinstance(errors, dict): + raise RuntimeError("the pinned runtime returned no per-question error map") + + over = [ + entry + for entry in errors.values() + if isinstance(entry, dict) and entry.get("error") == MAX_LENGTH_EXCEEDED + ] + if over: + message = str(over[0].get("message") or "a question prompt exceeds the maximum context length") + raise RequestError(message, 422) + invalid: dict[str, str] = {} + for name, entry in errors.items(): + # The pinned runtime always reports a string message; anything else + # degrades to a stable placeholder instead of a repr leaking into + # the JSON body. + message = entry.get("message") if isinstance(entry, dict) else None + invalid[name] = message if isinstance(message, str) and message else "invalid question" + if invalid: + # Insertion order of the runtime's map is the request order. + raise RequestError( + f"request contains {len(invalid)} invalid question(s)", 422, + body={"invalid_questions": invalid}, + ) diff --git a/src/models/d3/protocol.py b/src/models/d3/protocol.py new file mode 100644 index 00000000..a314c4f6 --- /dev/null +++ b/src/models/d3/protocol.py @@ -0,0 +1,127 @@ +"""Wire contract for the d3 reference worker. + +This module deliberately has no torch dependency and validates only the +request envelope: the five-field System One request, unknown-field rejection, +and the slice-scoped media refusal. Per-question validation belongs to the +checkpoint's pinned runtime, whose aggregated rejections ``preprocess`` maps +to HTTP 422. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any + +MODEL_NAME = "d3" +# Slice-scoped transport cap. Upstream d3_server.py sets no total-body limit +# (its budgets are per image and per video); stage 2 re-aligns this when the +# media paths open. +MAX_BODY = 8 * 1024 * 1024 +REQUEST_FIELDS = {"model", "state", "questions", "images", "videos"} + + +class RequestError(ValueError): + """A request cannot be evaluated under the reference contract.""" + + def __init__(self, message: str, status: int = 422, body: dict | None = None) -> None: + super().__init__(message) + self.status = status + # A full JSON error body ({"invalid_questions": {...}}); None sends + # the default {"detail": message} shape. + self.body = body + + +class MalformedJSON(RequestError): + """The body is not a usable UTF-8 JSON object.""" + + def __init__(self, message: str) -> None: + super().__init__(message, 400) + + +@dataclass(frozen=True) +class Request: + """One admitted request: the raw upstream fields, in request order. + + ``model`` is accepted and ignored (upstream semantics). ``state`` passes + through unchanged; the pinned runtime renders it. ``questions`` keeps the + request order for response reconstruction. + """ + + model: Any + state: Any + questions: tuple[tuple[str, Any], ...] + + +def _object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise MalformedJSON(f"duplicate JSON key {key!r}") + result[key] = value + return result + + +def _nonfinite(value: str) -> None: + raise MalformedJSON(f"non-finite JSON number {value} is not supported") + + +def parse_body(raw: bytes) -> dict[str, Any]: + """Decode a bounded JSON body and reject ambiguous JSON values.""" + + if len(raw) > MAX_BODY: + raise RequestError("request body exceeds 8 MiB", 413) + try: + value = json.loads( + raw.decode("utf-8"), + object_pairs_hook=_object, + parse_constant=_nonfinite, + ) + # Reject lone surrogates and values that cannot be emitted as UTF-8. + json.dumps(value, ensure_ascii=False, allow_nan=False).encode("utf-8") + except MalformedJSON: + raise + except (UnicodeError, ValueError, RecursionError) as exc: + raise MalformedJSON("request body must contain valid JSON and UTF-8 text") from exc + if not isinstance(value, dict): + raise MalformedJSON("request must be a JSON object") + return value + + +def _media(value: dict[str, Any], field: str) -> None: + """Admit an absent, null or empty media list; refuse anything else.""" + + media = value.get(field) + if media is None: + return + if not isinstance(media, list): + raise RequestError(f"{field} must be a list of base64 data URLs") + if media: + raise RequestError( + f"this reference worker serves text state only; non-empty {field}" + " are not supported yet (stage 2)" + ) + + +def parse_request(value: dict[str, Any]) -> Request: + """Validate the system1-omni wire request envelope. + + Unknown fields are rejected with upstream's message; ``model`` is not + checked (upstream ignores it); per-question content validation is left + to the pinned runtime. + """ + + unknown = set(value) - REQUEST_FIELDS + if unknown: + raise RequestError(f"Unknown request fields: {sorted(unknown)}") + + for field in ("images", "videos"): + _media(value, field) + + questions = value.get("questions") + if not isinstance(questions, dict) or not questions: + raise RequestError("questions must be a nonempty mapping of question IDs to questions") + if any(not isinstance(name, str) or not name for name in questions): + raise RequestError("question IDs must be nonempty strings") + + return Request(value.get("model"), value.get("state"), tuple(questions.items())) diff --git a/tests/d3/__init__.py b/tests/d3/__init__.py new file mode 100644 index 00000000..d1b3ac34 --- /dev/null +++ b/tests/d3/__init__.py @@ -0,0 +1 @@ +"""D3 contract tests.""" diff --git a/tests/d3/fixtures/MODEL_MANIFEST.json b/tests/d3/fixtures/MODEL_MANIFEST.json new file mode 100644 index 00000000..29205f13 --- /dev/null +++ b/tests/d3/fixtures/MODEL_MANIFEST.json @@ -0,0 +1,674 @@ +{ + "schema": "d3-package-manifest/1", + "model_name": "d3", + "repo_id": "vllm-sr/d3", + "format_id": "d3-code-readout-v1", + "prompt": "d3", + "attention_mode": "noncausal_full_attention", + "max_input_tokens": null, + "identity": { + "model_sha256": "0e439bebbbbf80d9e1b601c56c55a6fc3a6502b4e396cbbb5f3c3f3acc1d607b", + "files_sha256": { + "chat_template.jinja": "c3cf9e34abf4f9e36c2d72165aa9c132d3e2a725b6c2586aaa3a8af9d7a81041", + "model-00001-of-00011.safetensors": "4ecb29a7d1e2d1d0ad97008e5405a11d510a8a712a0b6491928359176fd56029", + "model-00002-of-00011.safetensors": "41dc277e43c9c5734656807da9f856f932338125a68ca6ec16b71dbdd6f97d62", + "model-00003-of-00011.safetensors": "46959de0f2202f7d6c53d122abff46c5f49f53871844eee01d60f4c035080146", + "model-00004-of-00011.safetensors": "946e87b8fa934091dc942622e5b72d16b8eb4d4571cafb2dbe36cb1022ea9275", + "model-00005-of-00011.safetensors": "87c7ce2751c7bf147394aa0413ae5e509907d52e16f59adb4a2db18ee023cef7", + "model-00006-of-00011.safetensors": "e4da6d081a7ef2012413537ef0e9cf6872df07b395cb65c08c05f9f79abb974a", + "model-00007-of-00011.safetensors": "262d7506f49ff1596aff610dccd46f4a33d6d9dab386bc5795c5abcab9a4d683", + "model-00008-of-00011.safetensors": "9e29bd831f8582119505018aeb3047030b698fe71fdd63bb38de2622b85534ed", + "model-00009-of-00011.safetensors": "0f1e3dfda5583dd38d61da22145451c8507ff7858a4293aa1e0e48ff20f9c319", + "model-00010-of-00011.safetensors": "f7a650a697b92b6561355624ae89b1aba2fc7df01572ff92a6531fa1c79cca14", + "model-00011-of-00011.safetensors": "9842116c181985fec72fcb9560f5c9b849800f6322f8304dcc466990758c7355", + "readout.safetensors": "5319cbf3ccdc5d3ba348b318d7b54acfc3a861d3ff94b578e787852c8726055a", + "tokenizer.json": "0997f410c57a1f4e53b09e4be8f4a172d90edd9564368fb0847030937229b9f3", + "tokenizer_config.json": "b11349aafa7cdc6a320767cf7ceb29ed82f7eda5d65e8e0819e76f0ce947bf27" + }, + "decision": { + "format_version": 1, + "format_id": "d3-code-readout-v1", + "prompt": "d3", + "base_model": "Qwen/Qwen3.8-27B", + "revision": "1d4bf0f2ff6012fd82039f2fa52739d0dd7c60c0", + "codes": [ + "A", + "B", + "C", + "D", + "E", + "F", + "G", + "H", + "I", + "J", + "K", + "L", + "M", + "N", + "O", + "P", + "Q", + "R", + "S", + "T", + "U", + "V", + "W", + "X", + "Y", + "Z", + "AA", + "AB", + "AC", + "AD", + "AE", + "AF", + "AG", + "AH", + "AI", + "AJ", + "AK", + "AL", + "AM", + "AN", + "AO", + "AP", + "AQ", + "AR", + "AS", + "AT", + "AU", + "AV", + "AW", + "AX", + "AY", + "AZ", + "BA", + "BB", + "BC", + "BD", + "BE", + "BF", + "BG", + "BH", + "BI", + "BJ", + "BK", + "BL", + "BM", + "BN", + "BO", + "BP", + "BR", + "BS", + "BT", + "BU", + "BV", + "BW", + "BX", + "BY", + "CA", + "CB", + "CC", + "CD", + "CE", + "CF", + "CG", + "CH", + "CI", + "CK", + "CL", + "CM", + "CN", + "CO", + "CP", + "CR", + "CS", + "CT", + "CU", + "CV", + "CW", + "CX", + "CY", + "DA", + "DB", + "DC", + "DD", + "DE", + "DF", + "DG", + "DH", + "DI", + "DJ", + "DK", + "DL", + "DM", + "DN", + "DO", + "DP", + "DR", + "DS", + "DT", + "DU", + "DV", + "DW", + "DX", + "DY", + "EA", + "EB", + "EC", + "ED", + "EE", + "EF", + "EG", + "EH", + "EI", + "EK", + "EL", + "EM", + "EN", + "EO", + "EP", + "EQ", + "ER", + "ES", + "ET", + "EU", + "EV", + "EW", + "EX", + "EZ", + "FA", + "FB", + "FC", + "FD", + "FE", + "FF", + "FG", + "FH", + "FI", + "FK", + "FL", + "FM", + "FN", + "FO", + "FP", + "FR", + "FS", + "FT", + "FU", + "FW", + "FX", + "FY", + "GA", + "GB", + "GC", + "GD", + "GE", + "GF", + "GG", + "GH", + "GI", + "GL", + "GM", + "GN", + "GO", + "GP", + "GR", + "GS", + "GT", + "GU", + "GV", + "GW", + "GX", + "GY", + "HA", + "HB", + "HC", + "HD", + "HE", + "HF", + "HG", + "HH", + "HI", + "HK", + "HL", + "HM", + "HN", + "HO", + "HP", + "HQ", + "HR", + "HS", + "HT", + "HU", + "HV", + "HW", + "HX", + "HY", + "HZ", + "IA", + "IB", + "IC", + "ID", + "IE", + "IF", + "IG", + "IH", + "II", + "IJ", + "IK", + "IL", + "IM", + "IN", + "IO", + "IP", + "IQ", + "IR", + "IS", + "IT", + "IU", + "IV", + "IW", + "IX", + "IZ", + "JA", + "JB", + "JC", + "JD", + "JE", + "JI", + "JJ", + "JK", + "JM", + "JO", + "JP", + "JR", + "JS", + "JT" + ], + "token_ids": [ + 32, + 33, + 34, + 35, + 36, + 37, + 38, + 39, + 40, + 41, + 42, + 43, + 44, + 45, + 46, + 47, + 48, + 49, + 50, + 51, + 52, + 53, + 54, + 55, + 56, + 57, + 5840, + 1803, + 1646, + 1745, + 13276, + 8018, + 1825, + 28946, + 15015, + 29595, + 11568, + 939, + 1354, + 1058, + 18183, + 2456, + 88898, + 905, + 1846, + 802, + 33869, + 7839, + 14006, + 2860, + 2926, + 22828, + 6844, + 9798, + 4738, + 9265, + 11261, + 19278, + 36513, + 93801, + 8335, + 14544, + 85266, + 9110, + 28000, + 15137, + 4525, + 25261, + 12717, + 7116, + 17078, + 14497, + 57339, + 74909, + 52072, + 19305, + 4887, + 12607, + 3580, + 6281, + 2036, + 9362, + 8533, + 2080, + 10911, + 2925, + 3040, + 9690, + 27731, + 8023, + 6901, + 8702, + 6211, + 1123, + 16307, + 18990, + 64045, + 63037, + 33380, + 6151, + 3392, + 5449, + 3967, + 1113, + 5095, + 51923, + 49600, + 17099, + 51483, + 17756, + 16037, + 8135, + 30237, + 5683, + 9992, + 7444, + 5751, + 10284, + 20887, + 59884, + 52396, + 16103, + 67547, + 18535, + 8006, + 7263, + 1425, + 6878, + 14453, + 9097, + 44072, + 76089, + 68720, + 2662, + 2629, + 923, + 6548, + 8924, + 52194, + 622, + 1515, + 1300, + 37523, + 44473, + 36530, + 3152, + 93924, + 3505, + 15731, + 6542, + 14176, + 11091, + 1686, + 11660, + 80440, + 18836, + 26132, + 5934, + 24794, + 40229, + 3660, + 11361, + 10191, + 8225, + 3860, + 78413, + 17680, + 15900, + 78138, + 15653, + 5213, + 22150, + 39500, + 10460, + 35131, + 21563, + 43194, + 27127, + 3697, + 20011, + 24368, + 15058, + 23658, + 8362, + 16035, + 24583, + 52857, + 38403, + 60456, + 81519, + 40097, + 16522, + 29722, + 21756, + 18567, + 1736, + 48043, + 87013, + 22456, + 23165, + 55523, + 13097, + 50397, + 41741, + 23073, + 6401, + 86538, + 16585, + 11622, + 2464, + 84982, + 75516, + 36984, + 58795, + 47217, + 59675, + 5681, + 3151, + 1271, + 887, + 5203, + 2685, + 1849, + 72470, + 5370, + 74063, + 27629, + 1655, + 1728, + 669, + 3682, + 3191, + 59865, + 2712, + 1580, + 922, + 77243, + 2990, + 78493, + 5228, + 2750, + 42711, + 44568, + 56402, + 48236, + 38993, + 43559, + 61250, + 32942, + 86302, + 25310, + 26313, + 81770, + 12185, + 78382 + ], + "temperature": 1.0, + "attention_mode": "noncausal_full_attention", + "pooling": "last", + "max_length": null, + "readout_dtype": "float32" + } + }, + "parameters": { + "text": 25624600064, + "vision": 460730096, + "readout": 1305600, + "loaded": 26086635760 + }, + "base_model": { + "repo_id": "Qwen/Qwen3.8-27B", + "revision": "1d4bf0f2ff6012fd82039f2fa52739d0dd7c60c0" + }, + "runtime": { + "files_sha256": { + "d3_runtime.py": "6c049aa7d3da8da3738602a2812572879d0a47b03c28c3e33f36d7300257de89", + "d3_fast.py": "68a1fb9a2c3bde9ac2763bf504decae592ca12722698fbde1b39a31262955535", + "d3_kernels.py": "8f7534ca1e1846ebf7c4c73058002f3be4dbf0d5d426f0815150ae0cbe7677bd", + "modeling_d3.py": "f8f8db23ff58f1a990ffa24977d744a50e1b4446d02a64c22070967705b6dbef", + "pipeline_d3.py": "4f9297c7c74e596e9b67df91a1e6823cfada48bf781336f141974ff6a2f1038e", + "d3_server.py": "f9b8f2481461a074f469b4f58da3b0d7004824096528366d5b1f23187dd6e692", + "d3_engine.py": "086238942762df45047a32cd9a8e2bedd9b63167eb40be260c776de567979f53", + "requirements.txt": "57a1938448bc240bdf4fee5d1b9cbc8450109fa1f6cee11e94dc04f274be8a32", + "d3_format.py": "e5036154d2e54793f59b320c8726632957bc343c382bac26d825b263a6e9ce62" + } + }, + "licence": { + "spdx": "apache-2.0", + "components": [ + { + "component": "Qwen/Qwen3.8-27B", + "licence": "apache-2.0" + }, + { + "component": "d3 (Decision 3.0) weights, readout and runtime", + "licence": "apache-2.0" + } + ] + }, + "built_utc": "2026-10-11T03:30:34+00:00", + "files_sha256": { + "LICENSE": "c71d239df91726fc519c6eb72d318ec65820627232b2f796219e87dcf35d0ab4", + "NOTICE": "61f4ea488e1fed0c3fce97171506a23152c2591116442f78f15827fb9e6af469", + "README.md": "62c45b5b40c4236aa52b65fb7c83da623c35edab4df7469f104c1597c4235d58", + "assets/banner.png": "eb7ecc949232e1de544247899dd55ee629a36cfe276042d5795f65178b9c64c5", + "assets/example-receipt.png": "b9dbb8b103bacd45f1de824d844a3c07b81d57cb1a7466712ef060ad7dcb3926", + "assets/example-video.mp4": "f4968ef89d07e97f20df4fd8047510a7a8e9df5c2d07c35e363a97927f164ab9", + "assets/index-areas.png": "602d4e76437b22dec2c5389b870349fc2ac9a68ee757ea67b0db10182962458d", + "assets/index-pareto.png": "c907a692a854298baebdb11ff5010cf24fd8890ce123b29c9a8c6df6faf8b25d", + "chat_template.jinja": "c3cf9e34abf4f9e36c2d72165aa9c132d3e2a725b6c2586aaa3a8af9d7a81041", + "config.json": "7e16284fafd2d54b73c10073c0391bfd6804886ceaf2cbe37fdb62d4c6813dea", + "d3_engine.py": "086238942762df45047a32cd9a8e2bedd9b63167eb40be260c776de567979f53", + "d3_fast.py": "68a1fb9a2c3bde9ac2763bf504decae592ca12722698fbde1b39a31262955535", + "d3_format.py": "e5036154d2e54793f59b320c8726632957bc343c382bac26d825b263a6e9ce62", + "d3_kernels.py": "8f7534ca1e1846ebf7c4c73058002f3be4dbf0d5d426f0815150ae0cbe7677bd", + "d3_runtime.py": "6c049aa7d3da8da3738602a2812572879d0a47b03c28c3e33f36d7300257de89", + "d3_server.py": "f9b8f2481461a074f469b4f58da3b0d7004824096528366d5b1f23187dd6e692", + "decision_config.json": "6b80ca11bd6ba481df4b3c187db8a5d1983786563ca05e2edccb0bdac0d2fc4f", + "merges.txt": "a9d356d7bdf1ef4949e3e748e95b8e10ad9d4e2e838eddc38a0a7b6b94d1db8d", + "model-00001-of-00011.safetensors": "4ecb29a7d1e2d1d0ad97008e5405a11d510a8a712a0b6491928359176fd56029", + "model-00002-of-00011.safetensors": "41dc277e43c9c5734656807da9f856f932338125a68ca6ec16b71dbdd6f97d62", + "model-00003-of-00011.safetensors": "46959de0f2202f7d6c53d122abff46c5f49f53871844eee01d60f4c035080146", + "model-00004-of-00011.safetensors": "946e87b8fa934091dc942622e5b72d16b8eb4d4571cafb2dbe36cb1022ea9275", + "model-00005-of-00011.safetensors": "87c7ce2751c7bf147394aa0413ae5e509907d52e16f59adb4a2db18ee023cef7", + "model-00006-of-00011.safetensors": "e4da6d081a7ef2012413537ef0e9cf6872df07b395cb65c08c05f9f79abb974a", + "model-00007-of-00011.safetensors": "262d7506f49ff1596aff610dccd46f4a33d6d9dab386bc5795c5abcab9a4d683", + "model-00008-of-00011.safetensors": "9e29bd831f8582119505018aeb3047030b698fe71fdd63bb38de2622b85534ed", + "model-00009-of-00011.safetensors": "0f1e3dfda5583dd38d61da22145451c8507ff7858a4293aa1e0e48ff20f9c319", + "model-00010-of-00011.safetensors": "f7a650a697b92b6561355624ae89b1aba2fc7df01572ff92a6531fa1c79cca14", + "model-00011-of-00011.safetensors": "9842116c181985fec72fcb9560f5c9b849800f6322f8304dcc466990758c7355", + "model.safetensors.index.json": "dc39d771cb240f775399e1845b11d081d8e007f14eecae3b780bc6272b128a46", + "modeling_d3.py": "f8f8db23ff58f1a990ffa24977d744a50e1b4446d02a64c22070967705b6dbef", + "pipeline_d3.py": "4f9297c7c74e596e9b67df91a1e6823cfada48bf781336f141974ff6a2f1038e", + "preprocessor_config.json": "27225450ac9c6529872ee1924fcb0962ff5634834f817040f444118116f4e516", + "readout.safetensors": "5319cbf3ccdc5d3ba348b318d7b54acfc3a861d3ff94b578e787852c8726055a", + "requirements.txt": "57a1938448bc240bdf4fee5d1b9cbc8450109fa1f6cee11e94dc04f274be8a32", + "tokenizer.json": "0997f410c57a1f4e53b09e4be8f4a172d90edd9564368fb0847030937229b9f3", + "tokenizer_config.json": "b11349aafa7cdc6a320767cf7ceb29ed82f7eda5d65e8e0819e76f0ce947bf27", + "video_preprocessor_config.json": "7768af27c1fafa9cc9011c1dc20067e03f8915e03b63504550e11d5066986d13", + "vocab.json": "ce99b4cb2983d118806ce0a8b777a35b093e2000a503ebde25853284c9dfa003" + }, + "files_bytes": { + "LICENSE": 11357, + "NOTICE": 100, + "README.md": 7892, + "assets/banner.png": 2233137, + "assets/example-receipt.png": 65918, + "assets/example-video.mp4": 26300, + "assets/index-areas.png": 126851, + "assets/index-pareto.png": 180468, + "chat_template.jinja": 8952, + "config.json": 3949, + "d3_engine.py": 4413, + "d3_fast.py": 22331, + "d3_format.py": 4882, + "d3_kernels.py": 16129, + "d3_runtime.py": 88800, + "d3_server.py": 10100, + "decision_config.json": 5503, + "merges.txt": 3353259, + "model-00001-of-00011.safetensors": 4997471976, + "model-00002-of-00011.safetensors": 4965144568, + "model-00003-of-00011.safetensors": 4933789248, + "model-00004-of-00011.safetensors": 4965227496, + "model-00005-of-00011.safetensors": 4974750552, + "model-00006-of-00011.safetensors": 4924266240, + "model-00007-of-00011.safetensors": 4974750560, + "model-00008-of-00011.safetensors": 4902229944, + "model-00009-of-00011.safetensors": 4996786840, + "model-00010-of-00011.safetensors": 4902229960, + "model-00011-of-00011.safetensors": 2634155256, + "model.safetensors.index.json": 103954, + "modeling_d3.py": 12079, + "pipeline_d3.py": 2936, + "preprocessor_config.json": 390, + "readout.safetensors": 5222480, + "requirements.txt": 679, + "tokenizer.json": 12809320, + "tokenizer_config.json": 17928, + "video_preprocessor_config.json": 385, + "vocab.json": 6722759 + } +} diff --git a/tests/d3/fixtures/oracle.json b/tests/d3/fixtures/oracle.json new file mode 100644 index 00000000..2b4425f1 --- /dev/null +++ b/tests/d3/fixtures/oracle.json @@ -0,0 +1,442 @@ +{ + "request": { + "model": "d3", + "state": "The customer's card was charged twice for one order and they opened a support ticket reporting the duplicate charge.", + "questions": { + "refund": { + "type": "choice", + "instructions": "Decide the refund action.", + "criteria": { + "refund": "Refund the duplicate charge now", + "wait": "Wait for manual review" + } + }, + "urgent": { + "type": "noul", + "instructions": "Is this urgent for the customer?", + "criteria": { + "false": "Routine duplicate charge", + "true": "Customer hardship reported" + } + }, + "severity": { + "type": "score", + "instructions": "Rate the severity of the billing error.", + "criteria": [ + "minor", + "moderate", + "major", + "critical" + ] + } + } + }, + "token_ids": { + "refund": [ + 248045, + 8678, + 198, + 2523, + 513, + 264, + 5307, + 4560, + 13, + 18307, + 279, + 1528, + 430, + 795, + 11, + 524, + 430, + 10895, + 13, + 4310, + 279, + 3296, + 321, + 1396, + 2904, + 11, + 1179, + 9559, + 440, + 1132, + 279, + 1970, + 314, + 279, + 1786, + 2904, + 13, + 248046, + 198, + 248045, + 846, + 198, + 1349, + 25, + 198, + 760, + 5813, + 579, + 3575, + 557, + 11102, + 10598, + 364, + 799, + 1906, + 321, + 781, + 8660, + 264, + 1761, + 11390, + 12459, + 279, + 21814, + 6545, + 13, + 271, + 14162, + 25, + 198, + 4741, + 562, + 279, + 20325, + 1852, + 13, + 271, + 3670, + 25, + 198, + 32, + 25, + 20325, + 25, + 8288, + 1199, + 279, + 21814, + 6545, + 1381, + 198, + 33, + 25, + 3655, + 25, + 13428, + 364, + 11048, + 3286, + 271, + 20206, + 440, + 1132, + 279, + 1970, + 314, + 279, + 1786, + 2904, + 13, + 248046, + 198, + 248045, + 74455, + 198, + 248068, + 271, + 248069, + 271 + ], + "severity": [ + 248045, + 8678, + 198, + 2523, + 513, + 264, + 5307, + 4560, + 13, + 18307, + 279, + 1528, + 430, + 795, + 11, + 524, + 430, + 10895, + 13, + 4310, + 279, + 3296, + 321, + 1396, + 2904, + 11, + 1179, + 9559, + 440, + 1132, + 279, + 1970, + 314, + 279, + 1786, + 2904, + 13, + 248046, + 198, + 248045, + 846, + 198, + 1349, + 25, + 198, + 760, + 5813, + 579, + 3575, + 557, + 11102, + 10598, + 364, + 799, + 1906, + 321, + 781, + 8660, + 264, + 1761, + 11390, + 12459, + 279, + 21814, + 6545, + 13, + 271, + 14162, + 25, + 198, + 11231, + 279, + 28943, + 314, + 279, + 32419, + 1412, + 13, + 271, + 3670, + 25, + 198, + 32, + 25, + 8652, + 198, + 33, + 25, + 22469, + 198, + 34, + 25, + 3478, + 198, + 35, + 25, + 8751, + 271, + 20206, + 440, + 1132, + 279, + 1970, + 314, + 279, + 1786, + 2904, + 13, + 248046, + 198, + 248045, + 74455, + 198, + 248068, + 271, + 248069, + 271 + ], + "urgent": [ + 248045, + 8678, + 198, + 2523, + 513, + 264, + 5307, + 4560, + 13, + 18307, + 279, + 1528, + 430, + 795, + 11, + 524, + 430, + 10895, + 13, + 4310, + 279, + 3296, + 321, + 1396, + 2904, + 11, + 1179, + 9559, + 440, + 1132, + 279, + 1970, + 314, + 279, + 1786, + 2904, + 13, + 248046, + 198, + 248045, + 846, + 198, + 1349, + 25, + 198, + 760, + 5813, + 579, + 3575, + 557, + 11102, + 10598, + 364, + 799, + 1906, + 321, + 781, + 8660, + 264, + 1761, + 11390, + 12459, + 279, + 21814, + 6545, + 13, + 271, + 14162, + 25, + 198, + 3742, + 411, + 32556, + 364, + 279, + 5813, + 30, + 271, + 3670, + 25, + 198, + 32, + 25, + 68650, + 21814, + 6545, + 198, + 33, + 25, + 11926, + 64301, + 4800, + 271, + 20206, + 440, + 1132, + 279, + 1970, + 314, + 279, + 1786, + 2904, + 13, + 248046, + 198, + 248045, + 74455, + 198, + 248068, + 271, + 248069, + 271 + ] + }, + "probabilities": { + "refund": [ + 0.952499508857727, + 0.04750046133995056 + ], + "severity": [ + 0.044581253081560135, + 0.6166402101516724, + 0.2996678650379181, + 0.0391106903553009 + ], + "urgent": [ + 0.9871630072593689, + 0.0128370001912117 + ] + }, + "answers": { + "refund": { + "type": "choice", + "choice": "refund", + "probabilities": { + "refund": 0.9524995372444254, + "wait": 0.047500462755574666 + }, + "confidence": 0.7243173114537556 + }, + "urgent": { + "type": "noul", + "noul": 0.012837000095568596 + }, + "severity": { + "type": "score", + "score": 1.3333079864586148, + "probabilities": { + "0": 0.0445812522511696, + "1": 0.6166401986658536, + "2": 0.29966785945616925, + "3": 0.03911068962680754 + }, + "confidence": 0.3329772483566522, + "legend": { + "0": "minor", + "1": "moderate", + "2": "major", + "3": "critical" + } + } + }, + "usage": { + "input_tokens": 348, + "output_tokens": 0 + } +} diff --git a/tests/d3/test_engine_manifest.py b/tests/d3/test_engine_manifest.py new file mode 100644 index 00000000..dfecca72 --- /dev/null +++ b/tests/d3/test_engine_manifest.py @@ -0,0 +1,195 @@ +"""Manifest verification tests for the d3 executor's runtime gate. + +Uses the pinned upstream manifest as a fixture and temp directories with +recomputed hashes: no weights, no torch, no network. What is under test is +the worker's own admission — the identity pin that binds the manifest to +REFERENCE_REVISION, and the full-hash gate over the runtime table that runs +before any package code is imported. +""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest + +from models.d3.engine import ( + MANIFEST_SCHEMA, + PINNED_RUNTIME_SHA256, + REFERENCE_REPO, + load_manifest, + verify_pinned_package, + verify_runtime_files, +) + +FIXTURE = Path(__file__).resolve().parent / "fixtures" / "MODEL_MANIFEST.json" +PINNED_RUNTIME_SHA = ( + "6c049aa7d3da8da3738602a2812572879d0a47b03c28c3e33f36d7300257de89" +) + + +def pinned_manifest(): + return json.loads(FIXTURE.read_text(encoding="utf-8")) + + +def build_package(root: Path, files: dict[str, bytes]) -> dict: + """Materialize a same-repo package whose runtime files hash to its own + manifest — the shape a later self-consistent snapshot would have.""" + + manifest = { + "schema": MANIFEST_SCHEMA, + "repo_id": REFERENCE_REPO, + "runtime": { + "files_sha256": { + name: hashlib.sha256(content).hexdigest() for name, content in files.items() + } + }, + } + for name, content in files.items(): + (root / name).write_bytes(content) + (root / "MODEL_MANIFEST.json").write_text(json.dumps(manifest), encoding="utf-8") + return manifest + + +def small_files(): + return { + "d3_runtime.py": b"verified runtime content\n", + "requirements.txt": b"torch>=2.8\n", + } + + +def test_the_pinned_manifest_fixture_shape_is_locked(): + manifest = pinned_manifest() + assert manifest["schema"] == MANIFEST_SCHEMA + assert manifest["repo_id"] == REFERENCE_REPO + assert manifest["model_name"] == "d3" + # The upstream runtime this worker was integrated against; the repo may + # have moved since (drift is why the pin and the hashes exist). + assert manifest["runtime"]["files_sha256"]["d3_runtime.py"] == PINNED_RUNTIME_SHA + assert manifest["runtime"]["files_sha256"] == PINNED_RUNTIME_SHA256 + assert len(manifest["files_sha256"]) == 39 + assert len(manifest["runtime"]["files_sha256"]) == 9 + + +def test_the_pinned_manifest_passes_the_identity_gate(): + assert verify_pinned_package(pinned_manifest()) is None + + +def test_self_consistent_but_unpinned_tables_are_refused(tmp_path): + # What a later snapshot looks like to the full-hash gate: every file + # hashes to the package's own manifest, yet the baseline changed. + manifest = build_package(tmp_path, small_files()) + + with pytest.raises(ValueError, match="not the table pinned for"): + verify_pinned_package(manifest) + + +def test_foreign_repo_ids_are_refused(): + manifest = pinned_manifest() + manifest["repo_id"] = "other/d3" + + with pytest.raises(ValueError, match="this worker only serves"): + verify_pinned_package(manifest) + + +def test_verification_accepts_a_package_matching_the_manifest(tmp_path): + manifest = build_package(tmp_path, small_files()) + assert load_manifest(tmp_path) == manifest + assert verify_runtime_files(tmp_path, manifest) == ["d3_runtime.py", "requirements.txt"] + + +def test_the_full_nine_file_table_verifies(tmp_path): + names = list(pinned_manifest()["runtime"]["files_sha256"]) + assert len(names) == 9 + files = {name: f"content of {name}\n".encode() for name in names} + manifest = build_package(tmp_path, files) + + assert verify_runtime_files(tmp_path, manifest) == sorted(names) + + +def test_tampered_runtime_content_is_rejected(tmp_path): + manifest = build_package(tmp_path, small_files()) + victim = tmp_path / "d3_runtime.py" + victim.write_bytes(victim.read_bytes() + b"tampered\n") + + with pytest.raises(ValueError, match="runtime SHA-256 differs"): + verify_runtime_files(tmp_path, manifest) + + +def test_truncated_runtime_file_is_rejected(tmp_path): + manifest = build_package(tmp_path, small_files()) + victim = tmp_path / "requirements.txt" + victim.write_bytes(victim.read_bytes()[:5]) + + with pytest.raises(ValueError, match="runtime verification failed"): + verify_runtime_files(tmp_path, manifest) + + +def test_missing_runtime_file_is_rejected(tmp_path): + manifest = build_package(tmp_path, small_files()) + (tmp_path / "d3_runtime.py").unlink() + + with pytest.raises(ValueError, match="missing runtime file d3_runtime.py"): + verify_runtime_files(tmp_path, manifest) + + +def test_packages_without_a_manifest_or_runtime_table_are_rejected(tmp_path): + with pytest.raises(ValueError, match="missing MODEL_MANIFEST.json"): + load_manifest(tmp_path) + + (tmp_path / "MODEL_MANIFEST.json").write_text(json.dumps({"schema": "other"})) + with pytest.raises(ValueError, match="not a d3-package-manifest/1 manifest"): + load_manifest(tmp_path) + + manifest = build_package(tmp_path, {}) + del manifest["runtime"] + (tmp_path / "MODEL_MANIFEST.json").write_text(json.dumps(manifest)) + with pytest.raises(ValueError, match="no runtime.files_sha256 table"): + verify_runtime_files(tmp_path, manifest) + + +def test_unsafe_runtime_names_are_rejected_before_any_file_access(tmp_path): + manifest = { + "schema": MANIFEST_SCHEMA, + "runtime": {"files_sha256": {"../escape.py": "0" * 64}}, + } + with pytest.raises(ValueError, match="unsafe runtime file name"): + verify_runtime_files(tmp_path, manifest) + + +def test_extra_importable_files_in_the_package_root_are_refused(tmp_path): + # The package root goes first on sys.path, so an unlisted module there + # executes in place of the installed dependency it shadows. + build_package(tmp_path, small_files()) + (tmp_path / "transformers.py").write_text("shadowing module\n") + + with pytest.raises(ValueError, match="unexpected importable file transformers.py"): + verify_runtime_files(tmp_path, load_manifest(tmp_path)) + + +def test_extra_package_directories_in_the_package_root_are_refused(tmp_path): + build_package(tmp_path, small_files()) + fla = tmp_path / "fla" + fla.mkdir() + (fla / "__init__.py").write_text("shadowing package\n") + + with pytest.raises(ValueError, match="unexpected importable package fla"): + verify_runtime_files(tmp_path, load_manifest(tmp_path)) + + +def test_unimportable_package_root_entries_still_pass(tmp_path): + # The pinned package's own layout shape: an assets/ directory without + # __init__.py, plain data files, and the __pycache__ Python writes next + # to the imported runtime — none of it can shadow an import. + manifest = build_package(tmp_path, small_files()) + assets = tmp_path / "assets" + assets.mkdir() + (assets / "banner.png").write_bytes(b"not a python package\n") + (tmp_path / "LICENSE").write_text("Apache-2.0\n") + pycache = tmp_path / "__pycache__" + pycache.mkdir() + (pycache / "d3_runtime.cpython-312.pyc").write_bytes(b"bytecode\n") + + assert verify_runtime_files(tmp_path, manifest) == ["d3_runtime.py", "requirements.txt"] diff --git a/tests/d3/test_postprocess.py b/tests/d3/test_postprocess.py new file mode 100644 index 00000000..c9c915cf --- /dev/null +++ b/tests/d3/test_postprocess.py @@ -0,0 +1,122 @@ +import logging +import math + +import pytest + +from models.d3.postprocess import ( + canonical, + choice_answer, + confidence, + describe, + finish_response, + noul_answer, + score_answer, + to_answer, +) + + +def test_confidence_is_one_minus_normalized_entropy(): + # H(0.25, 0.75) = 0.562335..., ln 2 = 0.693147... + assert confidence([0.25, 0.75]) == pytest.approx(1 - 0.5623351446188083 / math.log(2)) + assert confidence([1 / 3, 1 / 3, 1 / 3]) == pytest.approx(0.0) + assert confidence([1.0, 0.0]) == pytest.approx(1.0) + assert confidence([0.5]) == 1.0 + # The pinned checkpoint's Decision 2.0 measure, not max-p. + assert confidence([0.25, 0.75]) == pytest.approx(0.18872187554086717) + + +def test_describe_and_canonical_mirror_the_upstream_renderers(): + assert describe("plain") == "plain" + assert describe({"x": 1}) == '{"x": 1}' + assert canonical("plain") == "plain" + assert canonical({"b": 2, "a": 1}) == '{"a":1,"b":2}' + + +def test_choice_answer_renormalizes_and_breaks_ties_by_order(): + result = choice_answer(("up", "down", "left"), [1.0, 2.0, 1.0]) + assert result["type"] == "choice" + assert result["choice"] == "down" + assert result["probabilities"] == {"up": 0.25, "down": 0.5, "left": 0.25} + assert result["confidence"] == pytest.approx(confidence([0.25, 0.5, 0.25])) + + tied = choice_answer(("first", "second"), [0.5, 0.5]) + assert tied["choice"] == "first" + assert tied["confidence"] == pytest.approx(0.0) + + +def test_noul_answer_reports_only_the_true_probability(): + assert noul_answer([1.0, 3.0]) == {"type": "noul", "noul": pytest.approx(0.75)} + assert set(noul_answer([0.2, 0.8])) == {"type", "noul"} + + +def test_score_answer_uses_string_index_keys_and_a_canonical_legend(): + result = score_answer(["minor", {"kind": "major"}, "critical"], [1.0, 2.0, 1.0]) + values = [0.25, 0.5, 0.25] + assert result["type"] == "score" + assert result["score"] == pytest.approx(sum(i * p for i, p in enumerate(values))) + assert result["probabilities"] == {"0": 0.25, "1": 0.5, "2": 0.25} + assert result["confidence"] == pytest.approx(confidence(values)) + assert result["legend"] == { + "0": "minor", + "1": '{"kind":"major"}', + "2": "critical", + } + + +def test_to_answer_rejects_invalid_probabilities(): + for bad in ([0.5], [0.5, -0.1], [0.0, 0.0], [float("nan"), 1.0], [0.5, float("inf")]): + with pytest.raises(ValueError, match="one finite non-negative probability"): + to_answer("choice", ("a", "b"), None, bad) + with pytest.raises(ValueError, match="one finite non-negative probability"): + to_answer("score", ("a", "b"), ["a", "b"], [1.0, 2.0, 3.0]) + with pytest.raises(ValueError, match="unsupported question type"): + to_answer("triangle", ("a",), None, [1.0]) + + +def _response(answers, input_tokens=17): + return { + "model": "d3", + "answers": answers, + "usage": {"input_tokens": input_tokens, "output_tokens": 0}, + } + + +def test_finish_response_passes_a_valid_response_through(caplog): + answers = {"q": choice_answer(("a", "b"), [0.25, 0.75])} + response = _response(answers) + assert finish_response(("q",), response) is response + assert not caplog.records + + +def test_finish_response_warns_on_embedded_invalid_model_output(caplog): + bad = {"type": "noul", "error": "invalid_model_output", "message": "need one finite"} + good = choice_answer(("a", "b"), [0.25, 0.75]) + response = _response({"bad": bad, "good": good}) + + with caplog.at_level(logging.WARNING, logger="models.d3.postprocess"): + finished = finish_response(("bad", "good"), response) + + assert finished is response + assert finished["answers"]["bad"] is bad + assert finished["answers"]["good"] is good + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 1 + assert "bad" in warnings[0].getMessage() + assert "invalid_model_output" in warnings[0].getMessage() + assert "need one finite" in warnings[0].getMessage() + + +def test_finish_response_rejects_broken_contracts(): + with pytest.raises(RuntimeError, match="question order"): + finish_response(("a",), _response({"b": {"type": "noul", "noul": 0.5}})) + with pytest.raises(RuntimeError, match="envelope"): + finish_response(("a",), {"model": "d3", "answers": {}}) + with pytest.raises(RuntimeError, match="usage"): + finish_response(("a",), _response({"a": {}}, input_tokens=-1)) + with pytest.raises(RuntimeError, match="usage"): + bad_usage = _response({"a": {}}) + bad_usage["usage"]["output_tokens"] = 3 + finish_response(("a",), bad_usage) + with pytest.raises(RuntimeError, match="pre-execution error"): + embedded = {"type": None, "error": "invalid_question", "message": "bad"} + finish_response(("a",), _response({"a": embedded})) diff --git a/tests/d3/test_preprocess.py b/tests/d3/test_preprocess.py new file mode 100644 index 00000000..66ac2bea --- /dev/null +++ b/tests/d3/test_preprocess.py @@ -0,0 +1,61 @@ +import pytest + +from models.d3.preprocess import reject_pre_execution_errors +from models.d3.protocol import RequestError + + +class FakePrepared: + def __init__(self, errors): + self.keys = list(errors) + self.errors = dict(errors) + + +def invalid(message="choice criteria must map 1 to 255 option keys to descriptions"): + return {"type": "choice", "error": "invalid_question", "message": message} + + +def over(limit=8192, tokens=9000): + return { + "type": "choice", + "error": "max_length_exceeded", + "message": ( + f"the question prompt has {tokens} tokens, over the maximum context" + f" length of {limit} tokens; nothing was truncated" + ), + } + + +def test_clean_prepared_admits_the_request(): + assert reject_pre_execution_errors(FakePrepared({})) is None + + +def test_invalid_questions_reject_with_an_ordered_map_body(): + prepared = FakePrepared({"b": invalid("b is bad"), "a": invalid("a is bad")}) + with pytest.raises(RequestError) as excinfo: + reject_pre_execution_errors(prepared) + + assert excinfo.value.status == 422 + assert excinfo.value.body == { + "invalid_questions": {"b": "b is bad", "a": "a is bad"} + } + assert list(excinfo.value.body["invalid_questions"]) == ["b", "a"] + + +def test_max_length_rejects_with_the_plain_message_and_wins(): + prepared = FakePrepared({"long": over(), "bad": invalid()}) + with pytest.raises(RequestError) as excinfo: + reject_pre_execution_errors(prepared) + + assert excinfo.value.status == 422 + assert excinfo.value.body is None + assert str(excinfo.value) == prepared.errors["long"]["message"] + assert "9000 tokens" in str(excinfo.value) + assert "nothing was truncated" in str(excinfo.value) + + +def test_runtime_without_an_error_map_is_a_worker_bug(): + class Bare: + errors = None + + with pytest.raises(RuntimeError, match="error map"): + reject_pre_execution_errors(Bare()) diff --git a/tests/d3/test_protocol.py b/tests/d3/test_protocol.py new file mode 100644 index 00000000..9295f6d4 --- /dev/null +++ b/tests/d3/test_protocol.py @@ -0,0 +1,111 @@ +import pytest + +from models.d3.protocol import ( + MAX_BODY, + MODEL_NAME, + MalformedJSON, + RequestError, + parse_body, + parse_request, +) + + +def body(**overrides): + value = { + "model": MODEL_NAME, + "state": "The card was charged twice for one order.", + "questions": { + "refund": { + "type": "choice", + "instructions": "Decide the refund action.", + "criteria": {"refund": "Refund the duplicate charge", "wait": "Wait for review"}, + } + }, + } + value.update(overrides) + return value + + +def test_parse_body_rejects_duplicate_and_nonfinite_json(): + with pytest.raises(MalformedJSON, match="duplicate"): + parse_body(b'{"model":"x","model":"y"}') + with pytest.raises(MalformedJSON, match="non-finite"): + parse_body(b'{"value":NaN}') + with pytest.raises(MalformedJSON): + parse_body(b"{not json") + with pytest.raises(MalformedJSON): + parse_body(b"[1, 2]") + + +def test_parse_body_enforces_the_slice_scoped_body_limit(): + padding = b'{"state":"' + b"a" * (MAX_BODY) + b'"}' + with pytest.raises(RequestError) as excinfo: + parse_body(padding) + assert excinfo.value.status == 413 + + +def test_parse_rejects_unknown_fields_with_the_upstream_message(): + value = body(extra=1) + with pytest.raises(RequestError, match="Unknown request fields: \\['extra'\\]"): + parse_request(value) + + +def test_parse_accepts_and_ignores_any_model_value(): + for model in (MODEL_NAME, "other-model", "", None, 17): + parsed = parse_request(body(model=model)) + assert parsed.model == model + parsed = parse_request( + {k: v for k, v in body().items() if k != "model"} + ) + assert parsed.model is None + + +def test_parse_passes_state_through_in_every_wire_shape(): + for state in ("text", {"goal": "route"}, ["part one"], 3.5, None, ""): + parsed = parse_request(body(state=state)) + assert parsed.state == state + + +def test_parse_admits_empty_media_lists_and_refuses_the_rest(): + for field in ("images", "videos"): + assert parse_request(body(**{field: []})) is not None + assert parse_request(body(**{field: None})) is not None + + value = body(**{field: ["data:image/png;base64,AAAA"]}) + with pytest.raises(RequestError, match="text state only"): + parse_request(value) + + value = body(**{field: "data:image/png;base64,AAAA"}) + with pytest.raises(RequestError, match="must be a list"): + parse_request(value) + + +def test_parse_rejects_empty_or_malformed_questions(): + with pytest.raises(RequestError, match="nonempty mapping"): + parse_request(body(questions={})) + with pytest.raises(RequestError, match="nonempty mapping"): + parse_request(body(questions=[])) + with pytest.raises(RequestError, match="nonempty strings"): + parse_request(body(questions={"": {"type": "noul"}})) + with pytest.raises(RequestError, match="nonempty strings"): + parse_request(body(questions={3: {"type": "noul"}})) + + +def test_parse_preserves_question_request_order(): + questions = { + "zeta": {"type": "noul"}, + "alpha": {"type": "score", "criteria": ["low", "high"]}, + "mid": {"type": "choice", "criteria": {"a": None, "b": None}}, + } + parsed = parse_request(body(questions=questions)) + assert [name for name, _ in parsed.questions] == ["zeta", "alpha", "mid"] + assert dict(parsed.questions)["alpha"] is questions["alpha"] + + +def test_parse_leaves_question_content_to_the_pinned_runtime(): + # The envelope does not judge question bodies: kinds, criteria shapes and + # sizes are validated by the runtime and reported per question. + parsed = parse_request( + body(questions={"weird": {"type": "triangle", "criteria": 17}}) + ) + assert dict(parsed.questions)["weird"] == {"type": "triangle", "criteria": 17} diff --git a/tests/d3/test_worker.py b/tests/d3/test_worker.py new file mode 100644 index 00000000..d24af71b --- /dev/null +++ b/tests/d3/test_worker.py @@ -0,0 +1,253 @@ +"""End-to-end decide() tests with a CPU-only FakeD3: no weights, no torch. + +FakeD3 mirrors the upstream runtime surface the worker drives — +prepare/run/respond plus model_name, max_length and modalities — with the +pinned question-validation rules reimplemented just far enough to produce +upstream-shaped per-question errors, and answer construction delegated to +models.d3.postprocess (which locks the pinned formulas). +""" + +import json +import logging + +import pytest + +from frontend.d3 import decide +from models.d3.postprocess import describe, to_answer +from models.d3.protocol import MODEL_NAME, RequestError + +from .test_protocol import body + + +class FakePrepared: + def __init__(self, keys, questions, errors): + self.keys = list(keys) + self.questions = questions + self.errors = errors + + +class FakeD3: + """The pinned runtime's contract, driven without its code.""" + + model_name = MODEL_NAME + max_length = None + modalities = ["text"] + + def __init__(self, probabilities=None, max_length=None): + self.probabilities = dict(probabilities or {}) + self.max_length = max_length + self.warmed = False + self.closed = False + + # -- upstream normalize_question, reduced to what the tests exercise ---- + @staticmethod + def _normalize(question): + if not isinstance(question, dict): + raise ValueError("a question is an object with a type") + kind = question.get("type") + criteria = question.get("criteria") + if kind == "choice": + if not isinstance(criteria, dict) or not 1 <= len(criteria) <= 255: + raise ValueError("choice criteria must map 1 to 255 option keys to descriptions") + if any(not isinstance(key, str) or not key for key in criteria): + raise ValueError("choice option keys must be nonempty strings") + return {"kind": "choice", "keys": list(criteria), "levels": None} + if kind == "noul": + if criteria is not None and ( + not isinstance(criteria, dict) or set(criteria) - {"false", "true"} + ): + raise ValueError('noul criteria may only describe "false" and "true"') + return {"kind": "noul", "keys": ["false", "true"], "levels": None} + if kind == "score": + if not isinstance(criteria, (list, tuple)) or not 2 <= len(criteria) <= 10: + raise ValueError("score criteria must be an ordered list of 2 to 10 levels") + levels = [describe(level) for level in criteria] + if any(not text for text in levels) or len(set(levels)) != len(levels): + raise ValueError("score levels must be distinct and nonempty") + return { + "kind": "score", + "keys": [str(index) for index in range(len(criteria))], + "levels": list(criteria), + } + raise ValueError(f"unsupported question type {kind!r} (choice, noul or score)") + + def _tokens(self, state, question): + return 8 + len(json.dumps(state)) // 8 + len(json.dumps(question)) // 8 + + def prepare(self, state, questions): + errors = {} + normalized = {} + for key, question in questions.items(): + try: + normalized[key] = self._normalize(question) + except ValueError as exc: + kind = question.get("type") if isinstance(question, dict) else None + errors[key] = {"type": kind, "error": "invalid_question", "message": str(exc)} + prepared = FakePrepared(questions, normalized, errors) + if self.max_length is not None: + for key, question in questions.items(): + if key in errors: + continue + tokens = self._tokens(state, question) + if tokens > self.max_length: + errors[key] = { + "type": normalized[key]["kind"], + "error": "max_length_exceeded", + "message": ( + f"the question prompt has {tokens} tokens, over the maximum" + f" context length of {self.max_length} tokens; nothing was truncated" + ), + } + return prepared + + def run(self, prepared): + probabilities = {} + tokens = 0 + for key in prepared.keys: + if key in prepared.errors: + continue + probabilities[key] = self.probabilities.get(key) or [ + 1.0 for _ in prepared.questions[key]["keys"] + ] + tokens += 8 + 4 * len(probabilities[key]) + return probabilities, tokens + + def respond(self, prepared, probabilities, tokens): + answers = {} + for key in prepared.keys: + if key in prepared.errors: + answers[key] = dict(prepared.errors[key]) + continue + question = prepared.questions[key] + try: + answers[key] = to_answer( + question["kind"], question["keys"], question["levels"], probabilities[key] + ) + except ValueError as exc: + answers[key] = { + "type": question["kind"], + "error": "invalid_model_output", + "message": str(exc), + } + return { + "model": self.model_name, + "answers": answers, + "usage": {"input_tokens": tokens, "output_tokens": 0}, + } + + def warmup(self): + self.warmed = True + + def close(self): + self.closed = True + + +def three_kind_body(): + return { + "model": MODEL_NAME, + "state": {"ticket": "double charge", "customer": "vip"}, + "questions": { + "refund": { + "type": "choice", + "instructions": "Decide the refund action.", + "criteria": {"refund": "Refund the duplicate charge", "wait": "Wait for review"}, + }, + "urgent": { + "type": "noul", + "instructions": "Is this urgent for the customer?", + "criteria": {"true": "Customer hardship"}, + }, + "severity": { + "type": "score", + "instructions": "Rate the severity.", + "criteria": ["minor", "moderate", "major"], + }, + }, + } + + +def test_decide_answers_all_three_kinds_in_one_request(): + executor = FakeD3( + probabilities={ + "refund": [1.0, 3.0], + "urgent": [1.0, 3.0], + "severity": [1.0, 2.0, 1.0], + } + ) + result = decide(json.dumps(three_kind_body()).encode("utf-8"), executor) + + assert result["model"] == "d3" + assert list(result["answers"]) == ["refund", "urgent", "severity"] + assert result["answers"]["refund"]["choice"] == "wait" + assert result["answers"]["refund"]["probabilities"] == {"refund": 0.25, "wait": 0.75} + assert result["answers"]["urgent"] == {"type": "noul", "noul": 0.75} + assert result["answers"]["severity"]["score"] == pytest.approx(1.0) + assert result["answers"]["severity"]["legend"] == {"0": "minor", "1": "moderate", "2": "major"} + assert result["usage"]["output_tokens"] == 0 + assert result["usage"]["input_tokens"] == 8 + 4 * 2 + 8 + 4 * 2 + 8 + 4 * 3 + + +def test_decide_aggregates_invalid_questions_into_one_422(): + value = three_kind_body() + value["questions"]["broken"] = {"type": "triangle"} + value["questions"]["also_broken"] = {"type": "choice", "criteria": {}} + with pytest.raises(RequestError) as excinfo: + decide(json.dumps(value).encode("utf-8"), FakeD3()) + + assert excinfo.value.status == 422 + assert list(excinfo.value.body["invalid_questions"]) == ["broken", "also_broken"] + assert "unsupported question type" in excinfo.value.body["invalid_questions"]["broken"] + assert "choice criteria" in excinfo.value.body["invalid_questions"]["also_broken"] + + +def test_decide_rejects_over_limit_questions_without_truncating(): + value = three_kind_body() + with pytest.raises(RequestError) as excinfo: + decide(json.dumps(value).encode("utf-8"), FakeD3(max_length=16)) + + assert excinfo.value.status == 422 + assert excinfo.value.body is None + assert "tokens, over the maximum context length of 16" in str(excinfo.value) + assert "nothing was truncated" in str(excinfo.value) + + +def test_decide_keeps_partial_success_and_warns(caplog): + executor = FakeD3( + probabilities={"refund": [1.0, 3.0], "urgent": [0.0, 0.0], "severity": [1.0, 2.0, 1.0]} + ) + with caplog.at_level(logging.WARNING, logger="models.d3.postprocess"): + result = decide(json.dumps(three_kind_body()).encode("utf-8"), executor) + + broken = result["answers"]["urgent"] + assert broken["type"] == "noul" + assert broken["error"] == "invalid_model_output" + assert "one finite non-negative probability" in broken["message"] + assert result["answers"]["refund"]["choice"] == "wait" + assert result["answers"]["severity"]["type"] == "score" + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 1 + assert "urgent" in warnings[0].getMessage() + assert "invalid_model_output" in warnings[0].getMessage() + + +def test_decide_maps_runtime_value_errors_to_422(): + class RejectingD3(FakeD3): + def prepare(self, state, questions): + raise ValueError("questions must be a nonempty mapping") + + with pytest.raises(RequestError) as excinfo: + decide(json.dumps(body()).encode("utf-8"), RejectingD3()) + assert excinfo.value.status == 422 + assert "pinned runtime" in str(excinfo.value) + assert "nonempty mapping" in str(excinfo.value) + + +def test_decide_rejects_responses_that_break_question_order(): + class ReorderedD3(FakeD3): + def respond(self, prepared, probabilities, tokens): + response = super().respond(prepared, probabilities, tokens) + response["answers"] = dict(reversed(list(response["answers"].items()))) + return response + + with pytest.raises(RuntimeError, match="question order"): + decide(json.dumps(three_kind_body()).encode("utf-8"), ReorderedD3()) diff --git a/tests/d3/test_worker_server.py b/tests/d3/test_worker_server.py new file mode 100644 index 00000000..a4c0f4ec --- /dev/null +++ b/tests/d3/test_worker_server.py @@ -0,0 +1,246 @@ +"""HTTP tests for the d3 worker handler: status mapping and body limits. + +Starts the real WorkerServer on an ephemeral port with the FakeD3 from +test_worker: no weights, no torch, no GPU. The 408 slow-client timeout is not +covered because the handler's 15 s connection timeout is deliberately not +configurable for tests. + + PYTHONPATH=src python -m pytest tests/d3 -q +""" + +from __future__ import annotations + +import json +import logging +import threading +from http.client import HTTPConnection + +import pytest + +from frontend.d3 import WorkerServer +from models.d3.protocol import MAX_BODY + +from .test_protocol import body +from .test_worker import FakeD3, three_kind_body + + +class FailingExecutor: + model_name = "d3" + max_length = None + modalities = ["text"] + + def prepare(self, state, questions): + raise RuntimeError("boom") + + def close(self): + pass + + +def start(executor): + server = WorkerServer(("127.0.0.1", 0), executor) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server + + +def stop(server): + server.shutdown() + server.server_close() + + +def request(server, method, path, headers=(), payload=b""): + """One raw HTTP request with full header control.""" + + conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=10) + conn.putrequest(method, path, skip_accept_encoding=True) + for name, value in headers: + conn.putheader(name, value) + conn.endheaders(payload) + response = conn.getresponse() + raw = response.read() + status, ctype = response.status, response.getheader("Content-Type") + conn.close() + return status, ctype, raw + + +def post(server, payload, headers=()): + """POST /v1/systemone with JSON defaults; explicit headers replace them.""" + + defaults = {"Content-Type": "application/json", "Content-Length": str(len(payload))} + extra = [] + for name, value in headers: + defaults.pop(name, None) + extra.append((name, value)) + return request( + server, + "POST", + "/v1/systemone", + headers=(*defaults.items(), *extra), + payload=payload, + ) + + +@pytest.fixture() +def server(): + instance = start(FakeD3()) + yield instance + stop(instance) + + +def test_health_reports_model_limits_and_modalities(server): + status, ctype, raw = request(server, "GET", "/health") + assert status == 200 and ctype == "application/json" + health = json.loads(raw) + assert health["status"] == "ready" + assert health["model"] == "d3" + assert health["max_input_tokens"] is None + assert health["modalities"] == ["text"] + assert "authentication" not in health + + status, _, raw = request(server, "GET", "/nope") + assert status == 404 and json.loads(raw) == {"detail": "unknown route"} + + +def test_valid_three_kind_decision_returns_200(server): + status, ctype, raw = post(server, json.dumps(three_kind_body()).encode("utf-8")) + assert status == 200 and ctype == "application/json" + result = json.loads(raw) + assert result["model"] == "d3" + assert result["answers"]["urgent"] == {"type": "noul", "noul": 0.5} + assert result["usage"]["output_tokens"] == 0 + + +def test_unknown_post_route_is_404(server): + status, _, _ = request( + server, + "POST", + "/other", + headers=(("Content-Type", "application/json"), ("Content-Length", "2")), + payload=b"{}", + ) + assert status == 404 + + +def test_length_headers_are_enforced(server): + payload = b"{}" + base = {"Content-Type": "application/json"} + + status, _, raw = post(server, payload, headers=(("Transfer-Encoding", "chunked"),)) + assert status == 411 and "Content-Length" in json.loads(raw)["detail"] + + status, _, _ = request( + server, + "POST", + "/v1/systemone", + headers=[*base.items(), ("Content-Length", "2"), ("Content-Length", "2")], + payload=payload, + ) + assert status == 411 + + conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=10) + conn.putrequest("POST", "/v1/systemone", skip_accept_encoding=True) + conn.putheader("Content-Type", "application/json") + conn.endheaders(payload) + response = conn.getresponse() + response.read() + assert response.status == 411 + conn.close() + + status, _, _ = post(server, payload, headers=(("Content-Length", "abc"),)) + assert status == 400 + + +def test_truncated_body_is_rejected(server): + # The client declares 100 bytes, sends 7 and half-closes its write side: + # the worker must answer 400 (incomplete body) on the readable socket. + import socket + + sock = socket.create_connection(("127.0.0.1", server.server_address[1]), timeout=10) + sock.sendall( + b"POST /v1/systemone HTTP/1.1\r\nHost: localhost\r\n" + b"Content-Type: application/json\r\nContent-Length: 100\r\n\r\n{short" + ) + sock.shutdown(socket.SHUT_WR) + response = b"" + while True: + chunk = sock.recv(4096) + if not chunk: + break + response += chunk + sock.close() + status_line = response.split(b"\r\n", 1)[0] + assert b" 400 " in status_line + assert b"incomplete body" in response + + +def test_body_limit_and_content_type_are_enforced(server): + payload = json.dumps(body()).encode("utf-8") + + status, _, raw = post(server, payload, headers=(("Content-Length", str(MAX_BODY + 1)),)) + assert status == 413 and "body limit" in json.loads(raw)["detail"] + + status, _, raw = post(server, payload, headers=(("Content-Type", "text/plain"),)) + assert status == 415 and "application/json" in json.loads(raw)["detail"] + + +def test_transport_errors_map_to_their_status_and_detail(server): + status, _, raw = post(server, b"{not json") + assert status == 400 and "detail" in json.loads(raw) + + value = body(extra=True) + status, _, raw = post(server, json.dumps(value).encode("utf-8")) + assert status == 422 and "Unknown request fields" in json.loads(raw)["detail"] + + value = body() + value["images"] = ["data:image/png;base64,AAAA"] + status, _, raw = post(server, json.dumps(value).encode("utf-8")) + assert status == 422 and "text state only" in json.loads(raw)["detail"] + + +def test_invalid_questions_return_the_upstream_422_body(server): + value = three_kind_body() + value["questions"]["broken"] = {"type": "triangle"} + status, _, raw = post(server, json.dumps(value).encode("utf-8")) + assert status == 422 + payload = json.loads(raw) + assert set(payload) == {"invalid_questions"} + assert "unsupported question type" in payload["invalid_questions"]["broken"] + + +def test_partial_success_returns_200_and_logs_a_warning(server, caplog): + executor = FakeD3(probabilities={"refund": [1.0, 3.0], "urgent": [0.0, 0.0]}) + partial = start(executor) + try: + with caplog.at_level(logging.WARNING, logger="models.d3.postprocess"): + status, ctype, raw = post( + partial, json.dumps(three_kind_body()).encode("utf-8") + ) + finally: + stop(partial) + + assert status == 200 and ctype == "application/json" + result = json.loads(raw) + broken = result["answers"]["urgent"] + assert broken["error"] == "invalid_model_output" + assert result["answers"]["refund"]["choice"] == "wait" + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 1 + assert "urgent" in warnings[0].getMessage() + + +def test_busy_worker_returns_503(server): + server.inference_lock.acquire() + try: + status, _, raw = post(server, json.dumps(body()).encode("utf-8")) + finally: + server.inference_lock.release() + assert status == 503 and json.loads(raw) == {"detail": "worker busy"} + + +def test_executor_failure_returns_500_without_leaking_details(): + failing = start(FailingExecutor()) + try: + status, _, raw = post(failing, json.dumps(body()).encode("utf-8")) + finally: + stop(failing) + assert status == 500 and json.loads(raw) == {"detail": "inference failed"}