feat(runner): plan.v1 / result.v1 and engine-computed completeness

dokimi/changelog.d/20261021_120000_napetrov_run-plan-result.md

@@ -0,0 +1,7 @@
+### Added
+
+- **Run plan and result artifacts** — two new artifact kinds, `plan.v1` and `result.v1`. A plan lists every cell a run will measure (arm × model × task × rep, each with a stable `cell_id`) and records the subjects, suites, models, harnesses, grader digest, budget estimate and the one `varied_variable` (a list for a 2×2 design). A result gives each cell one status: `pass`, `fail`, `inconclusive`, `invalid`, `infra_error` or `not_run`, and refers to its plan by `plan_digest`. `dokimi.runner.plan.completeness` computes planned minus completed cells. A run is `complete` only when every planned cell is `pass`, `fail` or `inconclusive`. A result for a different plan, or one with unplanned or duplicated cells, is `invalid`.
+
+### Changed
+
+- **External-catalog smoke gate** — `scripts/check_smoke_runs.py` now declares its planned cells and checks them with `completeness`. A planned cell missing from `runs.json` fails the gate even when every count in the summary is clean.

dokimi/dokimi/artifacts.py

@@ -9,8 +9,9 @@
 localization results), ``novelty`` (what an artifact's content adds to one
 model), ``guidance_reach`` (L1 reach probes), ``repo_readiness`` (L0 claim
 checks), ``tool_preference`` (which of two equal tools a model picks, and
-what moves it) and ``repo_environment`` (static inspection of what a
-repository offers an agent). This module stamps each with a
+what moves it), ``repo_environment`` (static inspection of what a
+repository offers an agent), ``plan`` (the cells a run promises to measure) and
+``result`` (one status per planned cell). This module stamps each with a
 ``schema_version`` and validates it against the JSON Schema in
 ``dokimi/schemas/`` on save and load, so fixtures and baselines stay
 forward-compatible and self-describing.
@@ -50,6 +51,8 @@
     "repo_readiness": "repo_readiness.v1",
     "tool_preference": "tool_preference.v1",
     "repo_environment": "repo_environment.v1",
+    "plan": "plan.v1",
+    "result": "result.v1",
 }
 
 # kind -> superseded versions still accepted on *load*, newest first. An artifact written under an

dokimi/dokimi/runner/plan.py

@@ -0,0 +1,144 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""Planned cells versus completed cells: the ``plan.v1`` / ``result.v1`` pair.
+
+A plan lists every cell (arm x model x task x rep) a run promises to measure; a result
+gives each cell one status. :func:`completeness` is the one place that decides whether
+a run is complete, computed as planned minus completed. A run that finishes 2 of its 6
+planned cells is ``incomplete`` however well those two did, and a result that points at
+another plan (digest mismatch) or names cells the plan does not have is ``invalid``.
+
+Everything here is pure: no file, model or process is touched. Reading and writing the
+artifacts goes through :mod:`dokimi.artifacts` (kinds ``plan`` and ``result``).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import itertools
+import json
+from collections import Counter
+from dataclasses import dataclass, field
+from typing import Any, Iterable, Mapping
+
+#: Statuses that are a measurement of the cell. ``invalid``, ``infra_error`` and
+#: ``not_run`` are recorded, but the cell still counts as missing.
+MEASURED = frozenset({"pass", "fail", "inconclusive"})
+STATUSES = ("pass", "fail", "inconclusive", "invalid", "infra_error", "not_run")
+
+
+def cell_id(arm: str, model: str | None, task: str, rep: int) -> str:
+    """Stable id of one cell: the same four coordinates give the same id in every plan."""
+    key = json.dumps([arm, model, task, int(rep)], separators=(",", ":"), ensure_ascii=True)
+    return "c-" + hashlib.sha256(key.encode("utf-8")).hexdigest()[:16]
+
+
+def plan_cells(arms: Iterable[str], models: Iterable[str | None], tasks: Iterable[str],
+               reps: int = 1) -> list[dict[str, Any]]:
+    """The full arm x model x task x rep grid, in that nesting order, each with its id."""
+    if reps < 1:
+        raise ValueError(f"reps must be at least 1, got {reps}")
+    return [
+        {"cell_id": cell_id(arm, model, task, rep), "arm": arm, "model": model, "task": task, "rep": rep}
+        for arm, model, task, rep in itertools.product(list(arms), list(models), list(tasks),
+                                                       range(1, reps + 1))
+    ]
+
+
+def plan_digest(plan: Mapping[str, Any]) -> str:
+    """``sha256:<hex>`` of the plan's canonical JSON, ``generated_at`` left out."""
+    body = {k: v for k, v in plan.items() if k != "generated_at"}
+    text = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
+    return "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+@dataclass(frozen=True)
+class Completeness:
+    """What :func:`completeness` found. ``state`` is ``complete``, ``incomplete`` or ``invalid``."""
+
+    state: str
+    planned: int
+    measured: int
+    missing: tuple[str, ...]
+    statuses: dict[str, int]
+    problems: tuple[str, ...] = field(default=())
+
+    @property
+    def complete(self) -> bool:
+        return self.state == "complete"
+
+    def as_dict(self) -> dict[str, Any]:
+        return {"state": self.state, "planned": self.planned, "measured": self.measured,
+                "missing": list(self.missing), "statuses": dict(self.statuses),
+                "problems": list(self.problems)}
+
+
+def _statuses(result_cells: Any, problems: list[str]) -> dict[str, str]:
+    """``cell_id -> status`` of the result; an unknown status is appended to ``problems``."""
+    status: dict[str, str] = {}
+    for cell in result_cells if isinstance(result_cells, list) else []:
+        cid = cell.get("cell_id") if isinstance(cell, Mapping) else None
+        st = cell.get("status") if isinstance(cell, Mapping) else None
+        if not isinstance(cid, str):
+            continue  # reported by completeness
+        if isinstance(st, str) and st in STATUSES:
+            status[cid] = st
+        else:
+            problems.append(f"result cell {cid} has an unknown status: {st!r}")
+    return status
+
+
+def completeness(plan: Mapping[str, Any], result: Mapping[str, Any]) -> Completeness:
+    """Planned minus completed cells of ``result`` against ``plan``.
+
+    - ``invalid``: the result's ``plan_digest`` is not this plan's, the plan repeats a cell
+      id, a cell has no string ``cell_id`` or a result cell has a status outside
+      :data:`STATUSES`, or the result names a cell the plan lacks or gives one cell two
+      statuses. An ``invalid`` result reports ``measured=0``.
+    - ``incomplete``: some planned cell has no measured status (absent, ``not_run``,
+      ``infra_error`` or ``invalid``); those ids are in ``missing``, in plan order.
+    - ``complete``: every planned cell is ``pass``, ``fail`` or ``inconclusive``. Complete
+      says nothing about whether the run passed; that is a gate's call.
+    """
+    problems: list[str] = []
+
+    def _ids(side: str, cells: Any) -> list[str]:
+        """The string ``cell_id`` of each cell; a cell without one is a problem, not a crash."""
+        ids: list[str] = []
+        for cell in cells if isinstance(cells, list) else []:
+            cid = cell.get("cell_id") if isinstance(cell, Mapping) else None
+            if isinstance(cid, str):
+                ids.append(cid)
+            else:
+                problems.append(f"{side} has a cell without a string cell_id: {cell!r}")
+        return ids
+
+    planned = _ids("plan", plan.get("cells", []))
+    expected, actual = plan_digest(plan), result.get("plan_digest")
+    if actual != expected:
+        problems.append(f"result is for plan {actual}, not {expected}")
+    dup_planned = sorted(k for k, n in Counter(planned).items() if n > 1)
+    if dup_planned:
+        problems.append(f"plan repeats cell ids: {dup_planned}")
+    result_cells = result.get("cells", [])
+    seen = Counter(_ids("result", result_cells))
+    dup_result = sorted(k for k, n in seen.items() if n > 1)
+    if dup_result:
+        problems.append(f"result gives more than one status to cells: {dup_result}")
+    unplanned = sorted(set(seen) - set(planned))
+    if unplanned:
+        problems.append(f"result has cells the plan does not: {unplanned}")
+    status = _statuses(result_cells, problems)
+    counts = Counter(status.get(cid, "not_run") for cid in dict.fromkeys(planned))
+    missing = tuple(cid for cid in dict.fromkeys(planned) if status.get(cid, "not_run") not in MEASURED)
+    if problems:
+        state = "invalid"
+    elif missing or not planned:
+        state = "incomplete"
+    else:
+        state = "complete"
+    measured = 0 if problems else len(set(planned)) - len(missing)
+    return Completeness(state=state, planned=len(set(planned)), measured=measured,
+                        missing=missing, statuses={s: counts[s] for s in STATUSES if counts[s]},
+                        problems=tuple(problems))

dokimi/dokimi/schemas/plan.v1.json

@@ -0,0 +1,91 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://github.com/intel/dokimi/schemas/plan.v1.json",
+  "title": "Run plan v1",
+  "description": "What a run is going to measure, declared before any paid call: the subjects and suites it reads, the models, harnesses and grader it uses, the one variable its arms vary, and every cell (arm x model x task x rep) it promises to run. A result.v1 points back at a plan by plan_digest (dokimi.runner.plan.plan_digest); completeness is computed from the plan's cells, never from what happened to finish.",
+  "type": "object",
+  "required": ["schema_version", "subjects", "suites", "models", "harnesses", "grader", "varied_variable", "cells",
+               "budget"],
+  "additionalProperties": false,
+  "properties": {
+    "schema_version": {"const": "plan.v1"},
+    "generated_at": {"type": "string", "description": "Left out of plan_digest: re-stamping a plan does not change it."},
+    "subjects": {
+      "type": "array",
+      "description": "What is under test: a skill, repository, doc source or profile, pinned by commit and content.",
+      "items": {
+        "type": "object",
+        "required": ["id", "sha", "fingerprint"],
+        "additionalProperties": false,
+        "properties": {
+          "id": {"type": "string", "minLength": 1},
+          "kind": {"type": "string"},
+          "sha": {"type": ["string", "null"], "description": "commit the subject was read at; null when it is not in git"},
+          "fingerprint": {"type": ["string", "null"], "description": "digest of the subject's content as the run sees it"}
+        }
+      }
+    },
+    "suites": {
+      "type": "array",
+      "minItems": 1,
+      "items": {
+        "type": "object",
+        "required": ["id", "hash", "reasons"],
+        "additionalProperties": false,
+        "properties": {
+          "id": {"type": "string", "minLength": 1},
+          "hash": {"type": "string", "minLength": 1, "description": "digest of the suite's task/question content"},
+          "reasons": {"type": "array", "items": {"type": "string"},
+                      "description": "why this suite was selected (changed file, label, policy)"}
+        }
+      }
+    },
+    "models": {"type": "array", "minItems": 1, "items": {"type": ["string", "null"]},
+               "description": "requested model ids; null for a cell that calls no model (an oracle or probe harness)"},
+    "harnesses": {"type": "array", "minItems": 1, "items": {"type": "string", "minLength": 1}},
+    "grader": {
+      "type": "object",
+      "required": ["id", "digest"],
+      "additionalProperties": false,
+      "properties": {
+        "id": {"type": "string", "minLength": 1},
+        "digest": {"type": ["string", "null"], "description": "digest of the verifier or judge prompt the run grades with"}
+      }
+    },
+    "varied_variable": {
+      "description": "The one thing the arms differ in. A list names a factorial design (2x2: code and guidance), not several free variables.",
+      "oneOf": [
+        {"type": "string", "minLength": 1},
+        {"type": "array", "minItems": 1, "uniqueItems": true, "items": {"type": "string", "minLength": 1}}
+      ]
+    },
+    "cells": {
+      "type": "array",
+      "minItems": 1,
+      "items": {
+        "type": "object",
+        "required": ["cell_id", "arm", "model", "task", "rep"],
+        "additionalProperties": false,
+        "properties": {
+          "cell_id": {"type": "string", "minLength": 1,
+                      "description": "dokimi.runner.plan.cell_id(arm, model, task, rep): stable across re-plans"},
+          "arm": {"type": "string", "minLength": 1},
+          "model": {"type": ["string", "null"]},
+          "task": {"type": "string", "minLength": 1},
+          "rep": {"type": "integer", "minimum": 1}
+        }
+      }
+    },
+    "budget": {
+      "type": "object",
+      "required": ["cells"],
+      "additionalProperties": false,
+      "properties": {
+        "cells": {"type": "integer", "minimum": 0},
+        "agent_calls": {"type": ["integer", "null"], "minimum": 0},
+        "cost_usd": {"type": ["number", "null"], "minimum": 0},
+        "wall_sec": {"type": ["number", "null"], "minimum": 0}
+      }
+    }
+  }
+}

dokimi/dokimi/schemas/result.v1.json

@@ -0,0 +1,29 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://github.com/intel/dokimi/schemas/result.v1.json",
+  "title": "Run result v1",
+  "description": "One status per cell of the plan.v1 the run was given (plan_digest). A planned cell absent here is not_run; dokimi.runner.plan.completeness computes planned minus completed and refuses to call a run complete while any planned cell lacks a measured status (pass, fail or inconclusive).",
+  "type": "object",
+  "required": ["schema_version", "plan_digest", "cells"],
+  "additionalProperties": false,
+  "properties": {
+    "schema_version": {"const": "result.v1"},
+    "generated_at": {"type": "string"},
+    "plan_digest": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"},
+    "execution_id": {"type": ["string", "null"], "description": "the run's execution id, when the producer has one"},
+    "cells": {
+      "type": "array",
+      "items": {
+        "type": "object",
+        "required": ["cell_id", "status"],
+        "additionalProperties": false,
+        "properties": {
+          "cell_id": {"type": "string", "minLength": 1},
+          "status": {"enum": ["pass", "fail", "inconclusive", "invalid", "infra_error", "not_run"]},
+          "model": {"type": ["string", "null"], "description": "the model that answered (resolved), which may differ from the planned id"},
+          "detail": {"type": ["string", "null"]}
+        }
+      }
+    }
+  }
+}

dokimi/scripts/check_smoke_runs.py

@@ -10,6 +10,9 @@
 - every cell ran and passed: ``n > 0``, ``passed == n`` and no errored cell.
   ``pass_rate`` alone is not enough, because an errored cell is left out of its
   denominator (one pass + one errored cell reads ``pass_rate == 1.0``);
+- every planned cell (each harness on :data:`TASKS`, once) has a measured
+  status, computed by :func:`dokimi.runner.plan.completeness`: a cell missing
+  from the artifact fails the gate however well the others did;
 - each ``docker-probe`` row shows the egress proxy denying ``example.com``
   both as a CONNECT (``:443``) and as a plain-HTTP GET (``:80``, a separate
   code path): "closed" alone also fits a sidecar nobody reached.
@@ -22,9 +25,32 @@
 
 import json
 import sys
+from pathlib import Path
 from typing import Any
 
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:  # run as a script from CI, not installed
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from dokimi.harnesses.runner import is_errored
+from dokimi.runner.plan import cell_id, completeness, plan_cells, plan_digest
+
 HARNESSES = ("docker-oracle", "docker-probe")
+#: The task the workflow runs (``tasks run --tasks`` in agent-quality.yml).
+TASKS = ("onetbb-histogram-local-aggregation",)
+
+
+def planned() -> dict[str, Any]:
+    """The cells this gate expects; the oracle and probe call no model."""
+    return {"cells": plan_cells(HARNESSES, [None], TASKS)}
+
+
+def result_for(plan: dict[str, Any], data: dict[str, Any]) -> dict[str, Any]:
+    """The ``task_runs.v1`` rows as result cells of ``plan``."""
+    cells = [{"cell_id": cell_id(row["harness"], None, row.get("task", ""), row.get("repeat", 1)),
+              "status": "infra_error" if is_errored(row) else "pass" if row.get("passed") else "fail"}
+             for row in data["results"]]
+    return {"plan_digest": plan_digest(plan), "cells": cells}
 
 
 def harness_ok(stats: dict[str, Any]) -> bool:
@@ -39,6 +65,15 @@ def smoke_ok(data: dict[str, Any], out=sys.stdout) -> bool:
         print(f"{harness}: {stats['passed']}/{stats['n']} pass, "
               f"errored={stats.get('errored', 0)} (rate={stats.get('pass_rate')})", file=out)
         ok = harness_ok(stats) and ok
+    plan = planned()
+    found = completeness(plan, result_for(plan, data))
+    print(f"cells: {found.measured}/{found.planned} measured ({found.state})", file=out)
+    names = {c["cell_id"]: f"{c['arm']} on {c['task']}" for c in plan["cells"]}
+    for missing in found.missing:
+        print(f"  not measured: {names[missing]}", file=out)
+    for problem in found.problems:
+        print(f"  {problem}", file=out)
+    ok = found.complete and ok
     for row in data["results"]:
         metrics = row.get("metrics") or {}
         egress = metrics.get("egress") or {}