Skip to content

qnetbench.spec

JSON Schema generation and the published reference corpus. Both are generated from the pydantic models, and a test regenerates and diffs them, so the committed contract cannot drift from the code.

Narrative guide: Traces; the published contract is the spec.

qnetbench.spec

The versioned, machine-readable trace and metric specification (Deliverable 3).

trace_json_schema() / metric_json_schema() emit JSON Schemas generated from the pydantic models, so the committed spec can never drift from the code (a test regenerates and diffs them). generate_reference_corpus() produces the published reference traces a third-party scheduler can consume without importing qnetbench.

SPEC_VERSION module-attribute 

SPEC_VERSION = SCHEMA_VERSION

trace_json_schema 

trace_json_schema() -> dict[str, Any]

JSON Schema for a single trace event (the tagged union of event types).

Source code in qnetbench/spec.py 
28
29
30
def trace_json_schema() -> dict[str, Any]:
    """JSON Schema for a single trace event (the tagged union of event types)."""
    return TypeAdapter(Event).json_schema()

metric_json_schema 

metric_json_schema() -> dict[str, Any]

JSON Schema for the standard metric report.

Source code in qnetbench/spec.py 
33
34
35
def metric_json_schema() -> dict[str, Any]:
    """JSON Schema for the standard metric report."""
    return Report.model_json_schema()

write_specs 

write_specs(out_dir: str | Path) -> list[Path]

Write the trace and metric JSON Schemas to out_dir.

Source code in qnetbench/spec.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def write_specs(out_dir: str | Path) -> list[Path]:
    """Write the trace and metric JSON Schemas to `out_dir`."""
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    schemas = {
        "trace-schema.json": trace_json_schema(),
        "metric-schema.json": metric_json_schema(),
    }
    paths = []
    for name, schema in schemas.items():
        path = out / name
        path.write_text(_dump(schema))
        paths.append(path)
    return paths

generate_reference_corpus 

generate_reference_corpus(
    out_dir: str | Path, seed: int = 0
) -> dict[str, Any]

Run every application once on the reference backend and write the JSONL traces plus a manifest (spec version, seed, per-trace sha256 + event count).

Source code in qnetbench/spec.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def generate_reference_corpus(out_dir: str | Path, seed: int = 0) -> dict[str, Any]:
    """Run every application once on the reference backend and write the JSONL
    traces plus a manifest (spec version, seed, per-trace sha256 + event count)."""
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    entries = []
    for app in available_apps():
        events = run_once(app, seed=seed, backend="reference")
        path = out / f"{app}.jsonl"
        write_trace(path, events)
        digest = hashlib.sha256(path.read_bytes()).hexdigest()
        entries.append({"app": app, "file": path.name, "n_events": len(events), "sha256": digest})
    manifest = {
        "spec_version": SPEC_VERSION,
        "backend": "reference",
        "seed": seed,
        "traces": entries,
    }
    (out / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n")
    return manifest