D3: serve Decision 3.0 through a Python reference worker

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.

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

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

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.

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())

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"]
+    }
+  }
+}