feat: refine hosted runner, kernel, and timeout/reproducibility services

Author: ALikesToCode

services/hosted-runner/src/server.test.ts

@@ -33,11 +33,13 @@ async function start(
   authorizeToken: HostedRunnerServerOptions["authorizeToken"] = (token) =>
     Promise.resolve(token === "scoped-token"),
   onJobSettled?: HostedRunnerServerOptions["onJobSettled"],
+  releaseIdentity?: HostedRunnerServerOptions["releaseIdentity"],
 ) {
   const server = createHostedRunnerServer({
     authorizeToken,
     processJob,
     ...(onJobSettled === undefined ? {} : { onJobSettled }),
+    ...(releaseIdentity === undefined ? {} : { releaseIdentity }),
   });
   servers.push(server);
   server.listen(0, "127.0.0.1");
@@ -103,13 +105,20 @@ describe("hosted runner HTTP service", () => {
 
   it("accepts one authenticated job and reports readiness without exposing inputs", async () => {
     const processJob = vi.fn(async () => undefined);
-    const baseUrl = await start(processJob);
+    const runnerSourceCommit = "b".repeat(40);
+    const runnerImageDigest = `sha256:${"c".repeat(64)}`;
+    const baseUrl = await start(processJob, undefined, undefined, {
+      runnerSourceCommit,
+      runnerImageDigest,
+    });
 
     const ready = await fetch(`${baseUrl}/ready`);
     expect(ready.status).toBe(200);
     await expect(ready.json()).resolves.toEqual({
       status: "ready",
       service: "counterlab-hosted-runner",
+      runnerSourceCommit,
+      runnerImageDigest,
     });
 
     const accepted = await fetch(`${baseUrl}/jobs`, {

services/hosted-runner/src/server.ts

@@ -30,6 +30,10 @@ const DispatchSchema = z
   .strict();
 
 export type HostedRunnerServerOptions = {
+  releaseIdentity?: {
+    runnerSourceCommit: string;
+    runnerImageDigest: string;
+  };
   authorizeToken(
     token: string,
     jobId: string,
@@ -100,6 +104,7 @@ export function createHostedRunnerServer(options: HostedRunnerServerOptions) {
       respond(response, 200, {
         status: "ready",
         service: "counterlab-hosted-runner",
+        ...(options.releaseIdentity ?? {}),
       });
       return;
     }
@@ -217,6 +222,10 @@ async function startProductionServer(): Promise<void> {
   const authJson = process.env.CODEX_AUTH_JSON;
   const runnerVerifyingPublicKey =
     process.env.COUNTERLAB_RUNNER_VERIFYING_PUBLIC_KEY;
+  const runnerSourceCommit =
+    process.env.COUNTERLAB_RUNNER_SOURCE_COMMIT?.trim() ?? "";
+  const runnerImageDigest =
+    process.env.COUNTERLAB_RUNNER_IMAGE_DIGEST?.trim() ?? "";
   delete process.env.CODEX_AUTH_JSON;
   delete process.env.COUNTERLAB_RUNNER_VERIFYING_PUBLIC_KEY;
   if (authJson === undefined || authJson.trim().length === 0) {
@@ -230,6 +239,17 @@ async function startProductionServer(): Promise<void> {
       "COUNTERLAB_RUNNER_VERIFYING_PUBLIC_KEY is required by the hosted runner",
     );
   }
+  const releaseIdentityMissing =
+    runnerSourceCommit.length === 0 && runnerImageDigest.length === 0;
+  if (
+    !releaseIdentityMissing &&
+    (!/^[a-f0-9]{40}$/u.test(runnerSourceCommit) ||
+      !/^sha256:[a-f0-9]{64}$/u.test(runnerImageDigest))
+  ) {
+    throw new Error(
+      "Runner release identity must contain an exact source commit and image digest",
+    );
+  }
   const workspaceRoot = process.env.COUNTERLAB_RUNNER_WORK_ROOT ?? "/work/jobs";
   const codexHomeRoot =
     process.env.COUNTERLAB_CODEX_HOME_ROOT ?? "/run/counterlab-codex";
@@ -260,6 +280,9 @@ async function startProductionServer(): Promise<void> {
     current?: ReturnType<typeof createHostedRunnerServer>;
   } = {};
   const server = createHostedRunnerServer({
+    ...(releaseIdentityMissing
+      ? {}
+      : { releaseIdentity: { runnerSourceCommit, runnerImageDigest } }),
     async authorizeToken(scopedToken, jobId, purpose, controlPlaneOrigin) {
       try {
         await verifyRunnerJobToken(scopedToken, runnerVerifyingPublicKey, {

services/kernel/src/counterlab_kernel/hosted_patch.py

@@ -166,6 +166,88 @@
     "resultHash",
 })
 _V5_TRANSFER_CHECK_KEYS = frozenset({"invariant", "passed", "evidence"})
+_V5_TRANSFER_POLICIES: dict[str, dict[str, Any]] = {
+    "entity_leakage": {
+        "taskId": "forecasting-future-leakage-01",
+        "evaluatorVersion": "counterlab-transfer-v1",
+        "selectedStrategy": "time_ordered_holdout",
+        "identifiedRisks": ["centered_window_reads_future"],
+        "requiredEvidence": frozenset(
+            {"center_true_uses_later_targets", "random_split_mixes_dates"}
+        ),
+        "allowedEvidence": frozenset(
+            {
+                "center_true_uses_later_targets",
+                "random_split_mixes_dates",
+                "metric_is_mae",
+            }
+        ),
+        "checks": [
+            {
+                "invariant": "TIME_AWARE_EVALUATION",
+                "passed": True,
+                "evidence": (
+                    "A deployment-aligned forecast trains on earlier dates and "
+                    "evaluates later dates."
+                ),
+            },
+            {
+                "invariant": "FUTURE_INFORMATION_RISK",
+                "passed": True,
+                "evidence": (
+                    "A centered target window reads observations that do not "
+                    "exist at prediction time."
+                ),
+            },
+            {
+                "invariant": "EVIDENCE_GROUNDED",
+                "passed": True,
+                "evidence": (
+                    "Both center=True and the shuffled date split must be "
+                    "selected as code evidence."
+                ),
+            },
+        ],
+    },
+    "class_imbalance": {
+        "taskId": "manufacturing-defect-transfer-01",
+        "evaluatorVersion": "counterlab-imbalance-transfer-v1",
+        "selectedStrategy": "reject_accuracy_only",
+        "identifiedRisks": ["recall_and_pr_auc"],
+        "requiredEvidence": frozenset(
+            {"zero_true_positives", "rare_base_rate"}
+        ),
+        "allowedEvidence": frozenset(
+            {"zero_true_positives", "rare_base_rate", "many_true_negatives"}
+        ),
+        "checks": [
+            {
+                "invariant": "ACCURACY_CLAIM_REJECTED",
+                "passed": True,
+                "evidence": (
+                    "Overall accuracy cannot support deployment when the "
+                    "classifier misses every rare defective part."
+                ),
+            },
+            {
+                "invariant": "MINORITY_METRICS_SELECTED",
+                "passed": True,
+                "evidence": (
+                    "Defect recall and PR-AUC expose performance on the rare "
+                    "class and relative to its base rate."
+                ),
+            },
+            {
+                "invariant": "EVIDENCE_GROUNDED",
+                "passed": True,
+                "evidence": (
+                    "Use both the zero true-positive count and the 1% base rate "
+                    "as evidence."
+                ),
+            },
+        ],
+    },
+}
 
 
 class HostedPatchError(ValueError):
@@ -200,6 +282,56 @@ def _strict_fields(
         )
 
 
+def _validate_v5_transfer_semantics(
+    concept: str, transfer: Mapping[str, Any]
+) -> None:
+    policy = _V5_TRANSFER_POLICIES.get(concept)
+    if policy is None:
+        raise HostedPatchError("v5 patch transfer Subject Pack is not registered")
+    evidence = _sequence(
+        transfer.get("evidenceChoices"), "transferResult.evidenceChoices"
+    )
+    evidence_set = set(evidence)
+    required_evidence = policy["requiredEvidence"]
+    allowed_evidence = policy["allowedEvidence"]
+    checks = [
+        dict(_mapping(check, "transfer check"))
+        for check in _sequence(transfer.get("checks"), "transferResult.checks")
+    ]
+    if (
+        transfer.get("taskId") != policy["taskId"]
+        or transfer.get("evaluatorVersion") != policy["evaluatorVersion"]
+        or transfer.get("selectedStrategy") != policy["selectedStrategy"]
+        or transfer.get("identifiedRisks") != policy["identifiedRisks"]
+        or len(evidence) != len(evidence_set)
+        or evidence != sorted(evidence)
+        or not required_evidence.issubset(evidence_set)
+        or not evidence_set.issubset(allowed_evidence)
+        or any(
+            not isinstance(check.get("evidence"), str)
+            or not check["evidence"].strip()
+            for check in checks
+        )
+        or [
+            {
+                "invariant": check.get("invariant"),
+                "passed": check.get("passed"),
+            }
+            for check in checks
+        ]
+        != [
+            {
+                "invariant": check["invariant"],
+                "passed": check["passed"],
+            }
+            for check in policy["checks"]
+        ]
+    ):
+        raise HostedPatchError(
+            "v5 patch transfer does not match the fixed evaluator semantics"
+        )
+
+
 def _validate_v5_patch_bundle(bundle: Mapping[str, Any]) -> None:
     _strict_fields(bundle, _V5_PATCH_BUNDLE_KEYS, "v5 patch bundle")
     manifest = _mapping(bundle.get("artifactManifest"), "artifactManifest")
@@ -267,6 +399,7 @@ def _validate_v5_patch_bundle(bundle: Mapping[str, Any]) -> None:
     concept = belief.get("concept")
     if concept not in _PATCH_OPERATIONS:
         raise HostedPatchError("v5 patch Subject Pack is not registered")
+    _validate_v5_transfer_semantics(str(concept), transfer)
     if (
         manifest.get("support", {}).get("status") != "SUPPORTED"
         or belief.get("supportState") != "SUPPORTED"

services/kernel/src/counterlab_kernel/hosted_run.py

@@ -647,6 +647,7 @@ def _execute_v5_interactive(bundle: Mapping[str, Any]) -> dict[str, Any]:
             "scientific-candidate-verifier-v1",
             "scientific-candidate-verifier-v2",
             "scientific-candidate-verifier-v3",
+            "scientific-candidate-verifier-v4",
         }
         or compile_authority.get("projectionAdapterVersion")
         != "experiment-ir-v5-to-plan-v2-v1"
@@ -1094,6 +1095,7 @@ def _execute_v5(bundle: Mapping[str, Any]) -> dict[str, Any]:
             "scientific-candidate-verifier-v1",
             "scientific-candidate-verifier-v2",
             "scientific-candidate-verifier-v3",
+            "scientific-candidate-verifier-v4",
         }
         or provenance.get("scorerVersion") != selection.get("scorerVersion")
         or provenance.get("projectionAdapterVersion")

services/kernel/src/counterlab_kernel/imbalance_transfer.py

@@ -10,7 +10,7 @@
 
 _TASK: dict[str, Any] = {
     "schemaVersion": "1",
-    "id": "manufacturing-defect-imbalance-01",
+    "id": "manufacturing-defect-transfer-01",
     "concept": "class_imbalance_transfer",
     "title": "Would this inspection model catch rare defects?",
     "scenario": (
@@ -108,6 +108,8 @@ def evaluate_manufacturing_transfer(
         raise ValueError(f"unknown decision choice: {decision_choice}")
     if metric_choice not in _choice_ids("metricChoices"):
         raise ValueError(f"unknown metric choice: {metric_choice}")
+    if len(set(evidence_choices)) != len(evidence_choices):
+        raise ValueError("duplicate evidence choice")
     unknown_evidence = sorted(
         set(evidence_choices).difference(_choice_ids("evidenceChips"))
     )
@@ -129,8 +131,8 @@ def evaluate_manufacturing_transfer(
             "id": "MINORITY_METRICS_SELECTED",
             "passed": metric_choice == _CORRECT_METRIC,
             "feedback": (
-                "Defect recall and PR-AUC expose performance on the rare class and "
-                "relative to its base rate."
+                "Defect recall and PR-AUC expose performance on the rare class "
+                "and relative to its base rate."
             ),
         },
         {

services/kernel/src/counterlab_kernel/transfer.py

@@ -124,6 +124,8 @@ def evaluate_forecasting_transfer(
         raise ValueError(f"unknown strategy choice: {strategy_choice}")
     if risk_choice not in risk_ids:
         raise ValueError(f"unknown risk choice: {risk_choice}")
+    if len(set(evidence_choices)) != len(evidence_choices):
+        raise ValueError("duplicate evidence choice")
     unknown_evidence = sorted(set(evidence_choices).difference(evidence_ids))
     if unknown_evidence:
         raise ValueError(
@@ -178,4 +180,3 @@ def evaluate_forecasting_transfer(
     }
     result["resultHash"] = sha256_json(result)
     return result
-

services/kernel/tests/test_current_sample_patch_fixture.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from hashlib import sha256
+import json
+from pathlib import Path
+
+from counterlab_kernel.canonical import sha256_json
+from counterlab_kernel.patching import (
+    compile_sample_notebook_patch,
+    verify_sample_notebook_patch,
+)
+from counterlab_kernel.transfer import evaluate_forecasting_transfer
+
+
+ROOT = Path(__file__).resolve().parents[3]
+EVIDENCE = ROOT / "fixtures/public/leakage_sample_patch_v1"
+ORIGINAL = ROOT / "fixtures/notebooks/customer_churn_leakage.ipynb"
+FIXTURE = ROOT / "fixtures/public/customer_churn.csv"
+
+
+def read_json(path: Path) -> dict[str, object]:
+    value = json.loads(path.read_text(encoding="utf-8"))
+    assert isinstance(value, dict)
+    return value
+
+
+def file_sha256(path: Path) -> str:
+    return sha256(path.read_bytes()).hexdigest()
+
+
+def passing_transfer() -> dict[str, object]:
+    return evaluate_forecasting_transfer(
+        strategy_choice="time_ordered_holdout",
+        risk_choice="centered_window_reads_future",
+        evidence_choices=[
+            "center_true_uses_later_targets",
+            "random_split_mixes_dates",
+        ],
+    )
+
+
+def test_checked_in_current_patch_is_source_and_byte_bound() -> None:
+    patch = read_json(EVIDENCE / "patch-kernel-result.json")
+    metadata = read_json(EVIDENCE / "customer_churn_leakage.patch.json")
+    verification = read_json(EVIDENCE / "patch-verification.json")
+    patched = EVIDENCE / "customer_churn_leakage.patched.ipynb"
+
+    assert patch["status"] == "VERIFIED"
+    assert patch["originalSha256"] == file_sha256(ORIGINAL)
+    assert patch["patchedSha256"] == file_sha256(patched)
+    assert metadata["sourceNotebookSha256"] == patch["originalSha256"]
+    assert metadata["patchedNotebookSha256"] == patch["patchedSha256"]
+    assert verification["candidateSha256"] == patch["patchedSha256"]
+    metadata_without_hash = dict(metadata)
+    recorded_metadata_hash = metadata_without_hash.pop("metadataHash")
+    assert sha256_json(metadata_without_hash) == recorded_metadata_hash
+
+
+def test_current_patch_regenerates_deterministically(tmp_path: Path) -> None:
+    checked = read_json(EVIDENCE / "patch-kernel-result.json")
+    regenerated = compile_sample_notebook_patch(
+        original_notebook=ORIGINAL,
+        fixture_csv=FIXTURE,
+        output_dir=tmp_path / "patch",
+        transfer_result=passing_transfer(),
+    )
+    assert regenerated["status"] == "VERIFIED"
+
+    checked_portable = {
+        key: value
+        for key, value in checked.items()
+        if key not in {"patchFile", "metadataFile"}
+    }
+    regenerated_portable = {
+        key: value
+        for key, value in regenerated.items()
+        if key not in {"patchPath", "metadataPath"}
+    }
+    assert regenerated_portable == checked_portable
+    assert (
+        verify_sample_notebook_patch(
+            original_notebook=ORIGINAL,
+            candidate_notebook=Path(str(regenerated["patchPath"])),
+            fixture_csv=FIXTURE,
+        )["status"]
+        == "VERIFIED"
+    )