feat(cookbook): add profiling runner and manifest guards

Co-Authored-By: bot_apk <apk@cognition.ai>

cookbook/capability_routing_pilot/PROFILING_PLAN.md

@@ -2,21 +2,23 @@
 
 ## Current phase
 
-This phase labels the pinned public SWE-bench dev candidates with GPT-5.4-mini and drafts an availability-aware task manifest. The hard label budget is $5. The completed label run cost $0.109114 for 220 candidates. No solver, Jev, smoke, or other model calls are part of this phase
+This phase labels the pinned public SWE-bench dev candidates with GPT-5.4-mini and freezes an availability-aware task manifest. The hard label budget is $5. The completed label run cost $0.109114 for 220 candidates. No solver, Jev, smoke, gold, or noop execution was launched
 
 The candidate corpus is `princeton-nlp/SWE-bench` dev at revision `e48e2bd1e9fecd5bbd641e9414ac59da9f2e69f6`, excluding the five held-out IDs listed in the draft manifest. The original five training tasks remain eligible, but no earlier outcomes may be reused. The labeler sees only sanitized problem statements and uses the existing rubric, a 120-second timeout, no retries, and concurrency bounded at four. Label records are append-only and do not write to `costs_aux.jsonl`
 
-Image discovery uses registry manifest requests only and does not pull images. Each supported category is walked in deterministic seed-hash order until 30 images are confirmed or its candidates are exhausted. The process records every inspected candidate and leaves the rest explicitly not inspected. It prefers HEAD before reading manifest content, paces requests at one per second, uses at most four workers, and honors bounded `Retry-After` backoff without retry loops. Only HTTP 404 is recorded as unavailable. Authentication, rate-limit, and other errors remain errors and are not selected. No local Docker credentials were present, so discovery uses the anonymous registry token flow
+Image discovery used registry manifest requests only and did not pull images. It inspected 80 of 220 candidates, found 62 accessible images, and received 18 HTTP 401 responses. One fresh anonymous token scoped to each repository was retried once. The JWT access claim was empty and the HEAD still returned 401 for each. An empty claim can indicate a missing or private repository, so absence is unverified. These rows are frozen as `access_error` exclusions, not confirmed unavailable images. No `Retry-After` or `RateLimit-Remaining` header was present
 
 ## Draft cohort
 
-The deterministic selection seed is `litellm-capability-profiling-v1`. Candidates are ordered by `sha256(seed + task_id)` within each category, and the first 30 confirmed public-image tasks are selected from `repository_debugging` and `feature_implementation`. `shell_build` and `algorithms_data` are explicitly unsupported. Unknown labels are excluded from solves and reported. The manifest remains a draft until reviewed and frozen
+The deterministic selection seed is `litellm-capability-profiling-v1`. Candidates are ordered by `sha256(seed + task_id)` within each category, and the first 30 accessible tasks are selected from `repository_debugging` and `feature_implementation`. Each has at least 30 accessible candidates. `shell_build` and `algorithms_data` are explicitly unsupported. Unknown labels are excluded from solves and reported. The frozen artifact is `profiling/profiling_manifest.json`; availability and access-error bias are recorded there
 
-The effective API ceiling is $185, consisting of a $5 labeling allowance and a $180 solver allowance. The solver limit is 180 runs, at most 60 tasks across three solver models. A future run uses $1 per run, at most 50 steps, and a 600-second timeout, with three workers. Keep at least 50 GiB free. Process images in batches of three and clean them up after validation and solving. Validate each small image batch with gold and noop runs before solving. Compile an intermediate capability map after each batch with `final=false`. Freeze the reviewed manifest before any solve
+The effective API ceiling is $185, consisting of a $5 labeling allowance and a $180 solver allowance. The solver limit is 180 runs, at most 60 tasks across three solver models. A future run uses $1 per run, at most 50 steps, and a 600-second timeout, with three workers. The runner accepts `--max-batches N` to stop after a bounded number of three-task batches. For task index `i` in manifest order, model order rotates by `i % 3`: `haiku-4.5, sonnet-5, opus-5`, then `sonnet-5, opus-5, haiku-4.5`, then `opus-5, haiku-4.5, sonnet-5`. Each run record stores that task's model order
+
+Keep at least 50 GiB free and abort before any image pull if free disk is below 60 GiB. Process images in batches of three, validate with gold and noop before solving, and never replace a task that fails preflight. Clean only task-owned image tags or IDs after a batch, then verify disk; do not run `docker system prune` or `docker image prune`. Write raw results and a `final=false` map checkpoint after each batch, with a progress file for resume. The current runner entry point only builds dry-run schedules and checkpoints; it launches no pulls, gold/noop checks, or solves
 
 ## Profile records and routing
 
-Task-level records use tenant `public_dev_poc`. Profile keys include category, model alias, the concrete returned model version, deployment ID when reported, harness and budget versions, and a canonical harness-configuration hash. Infrastructure attempts contribute spend and latency but not graded evidence. Repeated graded attempts for one task and key are rejected
+Task-level records use tenant `public_dev_poc`. Profile keys include category, model alias, the concrete returned model version, deployment ID when reported, harness and budget versions, and a canonical harness-configuration hash. The map compiler and runner validate task membership and the manifest SHA-256. Infrastructure attempts contribute spend and latency but not graded evidence. Repeated graded attempts for one task and key are rejected
 
 Profiles use Beta(1,1) posterior summaries and a 5th-percentile lower bound. At least 13 graded tasks are required before a cell can qualify. This is a necessary threshold, not a general adequacy claim. Selection has no global fallback. The map builder reads only `profiling_runs.jsonl`, never the prior `runs.jsonl`
 

cookbook/capability_routing_pilot/poc/profile_manifest.py

@@ -3,7 +3,7 @@ import hashlib
 import json
 from collections.abc import Sequence
 from dataclasses import dataclass
-from itertools import takewhile
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Final
 
@@ -41,13 +41,20 @@ class ImageAvailability:
     compressed_size_bytes: int | None
     error: str | None
     manifest_response_received: bool
+    token_scope_ok: bool | None = None
+    retry_status: int | None = None
+    auth_rate_limit_remaining: str | None = None
+    auth_retry_after: str | None = None
+    rate_limit_remaining: str | None = None
+    retry_after: str | None = None
+    check_error_final: bool = False
 
 
 def _hash_order(task_id: str) -> str:
     return hashlib.sha256(f"{SEED}{task_id}".encode()).hexdigest()
 
 
-def _prefix_selected_ids(
+def _selected_ids(
     category: str,
     labels: Sequence[CandidateLabel],
     image_by_id: dict[str, ImageAvailability],
@@ -58,17 +65,9 @@ def _prefix_selected_ids(
             key=lambda task_id: (_hash_order(task_id), task_id),
         )
     )
-    resolved_prefix: Final = tuple(
-        takewhile(
-            lambda task_id: task_id in image_by_id and image_by_id[task_id].exists is not None,
-            ordered_ids,
-        )
-    )
-    return tuple(
-        task_id
-        for task_id in resolved_prefix
-        if image_by_id[task_id].exists is True
-    )[:MAX_PER_CATEGORY]
+    return tuple(task_id for task_id in ordered_ids if task_id in image_by_id and image_by_id[task_id].exists is True)[
+        :MAX_PER_CATEGORY
+    ]
 
 
 def _candidate_record(
@@ -83,6 +82,8 @@ def _candidate_record(
         if image.exists is True
         else "unavailable"
         if image.exists is False
+        else "access_error"
+        if image.error == "access_error"
         else "check_error"
     )
     return {
@@ -95,9 +96,7 @@ def _candidate_record(
         "call_id": label.call_id,
         "label_cost_usd": label.cost_usd,
         "image": (
-            image.image
-            if image is not None
-            else f"swebench/sweb.eval.x86_64.{task_id.replace('__', '_1776_').lower()}"
+            image.image if image is not None else f"swebench/sweb.eval.x86_64.{task_id.replace('__', '_1776_').lower()}"
         ),
         "image_available": image.exists if image is not None else None,
         "image_lookup_status": status,
@@ -105,12 +104,19 @@ def _candidate_record(
         "compressed_size_bytes": image.compressed_size_bytes if image is not None else None,
         "image_error": image.error if image is not None else None,
         "manifest_response_received": image.manifest_response_received if image is not None else False,
+        "token_scope_ok": image.token_scope_ok if image is not None else None,
+        "retry_status": image.retry_status if image is not None else None,
+        "auth_rate_limit_remaining": image.auth_rate_limit_remaining if image is not None else None,
+        "auth_retry_after": image.auth_retry_after if image is not None else None,
+        "rate_limit_remaining": image.rate_limit_remaining if image is not None else None,
+        "retry_after": image.retry_after if image is not None else None,
     }
 
 
 def build_draft_manifest(
     labels: Sequence[CandidateLabel],
     images: Sequence[ImageAvailability],
+    frozen_at_utc: str | None = None,
 ) -> dict[str, object]:
     label_by_id: Final = {label.task_id: label for label in labels}
     image_by_id: Final = {image.task_id: image for image in images}
@@ -119,16 +125,12 @@ def build_draft_manifest(
     if not set(image_by_id).issubset(label_by_id):
         raise ValueError("image results must belong to labeled candidates")
     records: Final = tuple(
-        _candidate_record(task_id, label_by_id[task_id], image_by_id.get(task_id))
-        for task_id in sorted(label_by_id)
+        _candidate_record(task_id, label_by_id[task_id], image_by_id.get(task_id)) for task_id in sorted(label_by_id)
     )
     category_counts: Final = {
         category: {
             "labeled": sum(label.primary_category == category for label in labels),
-            "inspected": sum(
-                label.primary_category == category and label.task_id in image_by_id
-                for label in labels
-            ),
+            "inspected": sum(label.primary_category == category and label.task_id in image_by_id for label in labels),
             "with_image": sum(
                 label.primary_category == category
                 and label.task_id in image_by_id
@@ -154,8 +156,7 @@ def build_draft_manifest(
                 for label in labels
             ),
             "not_inspected": sum(
-                label.primary_category == category and label.task_id not in image_by_id
-                for label in labels
+                label.primary_category == category and label.task_id not in image_by_id for label in labels
             ),
             "selected": 0,
             "selection_status": "profiled" if category in SUPPORTED_CATEGORIES else "unsupported_category",
@@ -164,9 +165,7 @@ def build_draft_manifest(
         for category in ALL_CATEGORIES
     }
     selected_ids: Final = {
-        category: _prefix_selected_ids(category, labels, image_by_id)
-        if category in SUPPORTED_CATEGORIES
-        else ()
+        category: _selected_ids(category, labels, image_by_id) if category in SUPPORTED_CATEGORIES else ()
         for category in ALL_CATEGORIES
     }
     categories_with_selection: Final = {
@@ -184,11 +183,12 @@ def build_draft_manifest(
         for category, counts in category_counts.items()
     }
     return {
-        "frozen": False,
+        "frozen": frozen_at_utc is not None,
+        "frozen_at_utc": frozen_at_utc,
         "tenant_id": PUBLIC_DEV_TENANT,
         "seed": SEED,
-        "rule": f"Per supported category, sort by sha256(seed + task_id), stop at the first unresolved check, and select up to {MAX_PER_CATEGORY} confirmed images",
-        "availability_bias": "Selection includes only tasks with confirmed public images; image lookup errors block later IDs in deterministic order",
+        "rule": f"Per supported category, sort by sha256(seed + task_id) and select the first {MAX_PER_CATEGORY} confirmed accessible images",
+        "availability_bias": "Selection includes only publicly accessible images at freeze time; access errors and unavailable images are excluded, so the cohort is availability-biased",
         "profiled_categories": SUPPORTED_CATEGORIES,
         "unsupported_categories": ("shell_build", "algorithms_data"),
         "heldout_exclusion_list": (
@@ -203,14 +203,23 @@ def build_draft_manifest(
             {
                 "task_id": record["task_id"],
                 "category": record["primary_category"],
-                "status": "check_error",
+                "status": record["image_lookup_status"],
+                "reason": record["image_error"],
+                "retry_status": record["retry_status"],
+                "token_scope_ok": record["token_scope_ok"],
+                "auth_rate_limit_remaining": record["auth_rate_limit_remaining"],
+                "auth_retry_after": record["auth_retry_after"],
+                "rate_limit_remaining": record["rate_limit_remaining"],
+                "retry_after": record["retry_after"],
             }
             for record in records
-            if record["image_available"] is None and record["image_lookup_status"] == "check_error"
+            if record["image_available"] is None and record["image_lookup_status"] in {"check_error", "access_error"}
         ),
         "unknown_count": category_counts["unknown"]["labeled"],
         "image_unavailable_count": sum(record["image_available"] is False for record in records),
-        "image_lookup_error_count": sum(record["image_lookup_status"] == "check_error" for record in records),
+        "image_lookup_error_count": sum(
+            record["image_lookup_status"] in {"check_error", "access_error"} for record in records
+        ),
         "image_not_inspected_count": sum(record["image_lookup_status"] == "not_inspected" for record in records),
         "selected_count": sum(len(ids) for ids in selected_ids.values()),
         "categories": categories_with_selection,
@@ -268,6 +277,13 @@ def _load_images(path: Path) -> tuple[ImageAvailability, ...]:
             compressed_size_bytes=value.get("compressed_size_bytes"),
             error=value.get("error"),
             manifest_response_received=value.get("manifest_response_received", False),
+            token_scope_ok=value.get("token_scope_ok"),
+            retry_status=value.get("retry_status"),
+            auth_rate_limit_remaining=value.get("auth_rate_limit_remaining"),
+            auth_retry_after=value.get("auth_retry_after"),
+            rate_limit_remaining=value.get("registry_rate_limit_remaining"),
+            retry_after=value.get("registry_retry_after"),
+            check_error_final=value.get("check_error_final", False),
         )
         for value in values
         if isinstance(value.get("task_id"), str)
@@ -281,6 +297,20 @@ def _load_images(path: Path) -> tuple[ImageAvailability, ...]:
         )
         and (value.get("error") is None or isinstance(value.get("error"), str))
         and isinstance(value.get("manifest_response_received", False), bool)
+        and (value.get("token_scope_ok") is None or isinstance(value.get("token_scope_ok"), bool))
+        and (
+            value.get("retry_status") is None
+            or isinstance(value.get("retry_status"), int)
+            and not isinstance(value.get("retry_status"), bool)
+        )
+        and (value.get("auth_rate_limit_remaining") is None or isinstance(value.get("auth_rate_limit_remaining"), str))
+        and (value.get("auth_retry_after") is None or isinstance(value.get("auth_retry_after"), str))
+        and (
+            value.get("registry_rate_limit_remaining") is None
+            or isinstance(value.get("registry_rate_limit_remaining"), str)
+        )
+        and (value.get("registry_retry_after") is None or isinstance(value.get("registry_retry_after"), str))
+        and isinstance(value.get("check_error_final", False), bool)
     )
     if len(images) != len(values):
         raise ValueError(f"{path.name} contains invalid image result fields")
@@ -292,10 +322,12 @@ def main() -> None:
     parser.add_argument("--labels", type=Path, default=ROOT / "profiling_labels.jsonl")
     parser.add_argument("--images", type=Path, default=ROOT / "profiling_candidate_images.json")
     parser.add_argument("--output", type=Path, default=ROOT / "profiling_manifest_draft.json")
+    parser.add_argument("--freeze", action="store_true")
     arguments = parser.parse_args()
     manifest: Final = build_draft_manifest(
         _load_labels(arguments.labels),
         _load_images(arguments.images),
+        datetime.now(UTC).isoformat() if arguments.freeze else None,
     )
     arguments.output.write_text(json.dumps(manifest, indent=2) + "\n")
     print(

cookbook/capability_routing_pilot/poc/profiles.py

@@ -4,7 +4,7 @@ import math
 from collections.abc import Mapping, Sequence
 from dataclasses import asdict, dataclass
 from enum import Enum
-from itertools import groupby
+from itertools import chain, groupby
 from pathlib import Path
 from typing import Final
 
@@ -203,11 +203,42 @@ def attempt_from_mapping(value: object) -> TaskAttempt:
 
 
 def read_attempts_jsonl(path: Path) -> tuple[TaskAttempt, ...]:
-    return tuple(
-        attempt_from_mapping(json.loads(line))
-        for line in path.read_text().splitlines()
-        if line.strip()
-    )
+    return tuple(attempt_from_mapping(json.loads(line)) for line in path.read_text().splitlines() if line.strip())
+
+
+def _manifest_category_ids(value: object) -> tuple[str, ...]:
+    row: Final = _record_mapping(value)
+    selected_ids: Final = row.get("selected_ids")
+    if not isinstance(selected_ids, (list, tuple)) or not all(
+        isinstance(task_id, str) and task_id for task_id in selected_ids
+    ):
+        raise ValueError("manifest category selected_ids must be non-empty strings")
+    return tuple(selected_ids)
+
+
+def manifest_selection(path: Path, expected_sha256: str) -> tuple[tuple[str, ...], str]:
+    raw_manifest: Final = path.read_bytes()
+    actual_sha256: Final = hashlib.sha256(raw_manifest).hexdigest()
+    if actual_sha256 != expected_sha256:
+        raise ValueError("manifest sha256 mismatch")
+    manifest: Final = _record_mapping(json.loads(raw_manifest))
+    categories: Final = _record_mapping(manifest.get("categories"))
+    task_ids: Final = tuple(chain.from_iterable(_manifest_category_ids(category) for category in categories.values()))
+    if len(task_ids) != len(set(task_ids)):
+        raise ValueError("manifest selected task IDs must be unique")
+    return task_ids, actual_sha256
+
+
+def validate_manifest_attempt(
+    task_id: str,
+    attempt_manifest_sha256: str,
+    selected_task_ids: Sequence[str],
+    manifest_sha256: str,
+) -> None:
+    if task_id not in selected_task_ids:
+        raise ValueError(f"task_id is not selected in the manifest: {task_id}")
+    if attempt_manifest_sha256 != manifest_sha256:
+        raise ValueError("attempt manifest sha256 mismatch")
 
 
 def _status(
@@ -316,9 +347,7 @@ def _cheapest(profiles: Sequence[Profile]) -> Profile:
     return min(
         profiles,
         key=lambda profile: (
-            profile.cost_per_graded_task_usd
-            if profile.cost_per_graded_task_usd is not None
-            else math.inf,
+            profile.cost_per_graded_task_usd if profile.cost_per_graded_task_usd is not None else math.inf,
             profile.key.model_alias,
             profile.key.returned_model_version,
         ),
@@ -353,15 +382,11 @@ def select(
         return Selection(DecisionStatus.SELECTED, _cheapest(qualified), False, "cheapest_qualified")
     if quality_first:
         evidenced: Final = tuple(
-            profile
-            for profile in matching
-            if profile.graded_count >= min_evidence and profile.successes > 0
+            profile for profile in matching if profile.graded_count >= min_evidence and profile.successes > 0
         )
         if evidenced:
             best_lower_bound: Final = max(profile.lower_5pct for profile in evidenced)
-            best: Final = _cheapest(
-                tuple(profile for profile in evidenced if profile.lower_5pct == best_lower_bound)
-            )
+            best: Final = _cheapest(tuple(profile for profile in evidenced if profile.lower_5pct == best_lower_bound))
             return Selection(DecisionStatus.SELECTED, best, True, "quality_first_threshold_unmet")
     return Selection(DecisionStatus.NO_QUALIFIED_MODEL, None, False, "no_qualified_model")
 
@@ -383,13 +408,10 @@ def compile_map(
         "unknown",
     )
     matching: Final = tuple(
-        profile
-        for profile in profiles
-        if profile.key.tenant_id == tenant_id and _versions_match(profile, versions)
+        profile for profile in profiles if profile.key.tenant_id == tenant_id and _versions_match(profile, versions)
     )
     category_profiles: Final = {
-        category: tuple(profile for profile in matching if profile.key.category == category)
-        for category in categories
+        category: tuple(profile for profile in matching if profile.key.category == category) for category in categories
     }
     return {
         "final": final,
@@ -437,15 +459,24 @@ def compile_map(
 
 def compile_map_from_profiling_runs(
     path: Path,
+    manifest_path: Path,
+    expected_manifest_sha256: str,
     tenant_id: str,
     versions: ProfileVersions,
     tau: float = 0.80,
     min_evidence: int = 13,
 ) -> dict[str, object]:
     _validate_thresholds(tau, min_evidence)
-    if path.is_symlink():
-        raise ValueError("capability maps may only read profiling_runs.jsonl")
-    profiles: Final = aggregate_profiles(read_attempts_jsonl(path), tau, min_evidence)
+    selected_task_ids, actual_manifest_sha256 = manifest_selection(manifest_path, expected_manifest_sha256)
+    attempts: Final = read_attempts_jsonl(path)
+    for attempt in attempts:
+        validate_manifest_attempt(
+            attempt.task_id,
+            attempt.manifest_sha256,
+            selected_task_ids,
+            actual_manifest_sha256,
+        )
+    profiles: Final = aggregate_profiles(attempts, tau, min_evidence)
     return compile_map(tenant_id, versions, profiles, tau, min_evidence, final=False)
 
 

cookbook/capability_routing_pilot/poc/run_profiling.py

@@ -0,0 +1,323 @@
+import argparse
+import hashlib
+import json
+import shutil
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Final
+
+from poc import run_pilot
+from poc.profiles import (
+    PUBLIC_DEV_TENANT,
+    HarnessConfig,
+    ProfileVersions,
+    compile_map_from_profiling_runs,
+    harness_config_sha256,
+    manifest_selection,
+    validate_manifest_attempt,
+)
+
+ROOT = Path(__file__).resolve().parents[1]
+MODEL_ORDER: Final = ("haiku-4.5", "sonnet-5", "opus-5")
+PHASE_BUDGET_USD: Final = 180.0
+BATCH_SIZE: Final = 3
+WORKERS: Final = 3
+MIN_FREE_BYTES: Final = 60 * 1024**3
+MANIFEST_SHA256: Final = "15720dee12d40b62780837ed12a403f3951e9c5bb1ed18ec0d5d5d1f4776bff4"
+
+
+@dataclass(frozen=True, slots=True)
+class ProfileConfig:
+    root_dir: Path
+    manifest_path: Path
+    manifest_sha256: str
+    tenant_id: str = PUBLIC_DEV_TENANT
+
+    @property
+    def profile_dir(self) -> Path:
+        return self.root_dir / "profiling"
+
+    @property
+    def runs_path(self) -> Path:
+        return self.profile_dir / "profiling_runs.jsonl"
+
+
+@dataclass(frozen=True, slots=True)
+class ProfileTask:
+    index: int
+    task_id: str
+    category: str
+    model_order: tuple[str, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class ProfileBatch:
+    index: int
+    tasks: tuple[ProfileTask, ...]
+
+
+def _mapping(value: object, field: str) -> Mapping[str, object]:
+    if not isinstance(value, Mapping):
+        raise ValueError(f"{field} must be a JSON object")
+    return value
+
+
+def model_order_for_task(index: int) -> tuple[str, ...]:
+    offset: Final = index % len(MODEL_ORDER)
+    return MODEL_ORDER[offset:] + MODEL_ORDER[:offset]
+
+
+def _category_tasks(
+    category: str,
+    manifest_category: Mapping[str, object],
+    index_offset: int,
+) -> tuple[ProfileTask, ...]:
+    selected_ids: Final = manifest_category.get("selected_ids")
+    if not isinstance(selected_ids, list) or not all(isinstance(task_id, str) for task_id in selected_ids):
+        raise ValueError(f"{category} selected_ids must be strings")
+    return tuple(
+        ProfileTask(
+            index=index_offset + index,
+            task_id=task_id,
+            category=category,
+            model_order=model_order_for_task(index_offset + index),
+        )
+        for index, task_id in enumerate(selected_ids)
+    )
+
+
+def profile_batches(config: ProfileConfig, max_batches: int | None = None) -> tuple[ProfileBatch, ...]:
+    if max_batches is not None and max_batches < 1:
+        raise ValueError("max_batches must be positive")
+    selected_ids, _ = manifest_selection(config.manifest_path, config.manifest_sha256)
+    manifest: Final = _mapping(json.loads(config.manifest_path.read_text()), "manifest")
+    categories: Final = _mapping(manifest.get("categories"), "manifest categories")
+    repository_tasks: Final = _category_tasks(
+        "repository_debugging",
+        _mapping(categories.get("repository_debugging"), "repository_debugging"),
+        0,
+    )
+    feature_tasks: Final = _category_tasks(
+        "feature_implementation",
+        _mapping(categories.get("feature_implementation"), "feature_implementation"),
+        len(repository_tasks),
+    )
+    tasks: Final = repository_tasks + feature_tasks
+    if tuple(task.task_id for task in tasks) != selected_ids:
+        raise ValueError("profile categories differ from manifest selection")
+    batches: Final = tuple(
+        ProfileBatch(index=index, tasks=tasks[start : start + BATCH_SIZE])
+        for index, start in enumerate(range(0, len(tasks), BATCH_SIZE), start=1)
+    )
+    return batches if max_batches is None else batches[:max_batches]
+
+
+def require_disk_for_pull(free_bytes: int) -> None:
+    if free_bytes < MIN_FREE_BYTES:
+        raise OSError("abort profiling batch before pull: less than 60 GiB free")
+
+
+def _single_result_value(result: Mapping[str, object], field: str) -> str | None:
+    value: Final = result.get(field, ())
+    values: Final = (value,) if isinstance(value, str) else value
+    if not isinstance(values, (list, tuple)) or not all(isinstance(item, str) for item in values):
+        raise ValueError(f"{field} must contain strings")
+    if len(values) > 1:
+        raise ValueError(f"{field} must contain at most one value")
+    return values[0] if values else None
+
+
+def profile_guard(
+    config: ProfileConfig,
+    free_bytes: Callable[[], int] = lambda: shutil.disk_usage("/").free,
+    committed_spend: Callable[[], float] | None = None,
+) -> run_pilot.DispatchGuard:
+    config.profile_dir.mkdir(parents=True, exist_ok=True)
+    spend_reader: Final = (
+        committed_spend
+        if committed_spend is not None
+        else lambda: run_pilot.committed_spend_usd(config.runs_path, config.profile_dir / "costs_aux.jsonl")
+    )
+    return run_pilot.DispatchGuard(
+        config.profile_dir / "dispatch_stops.jsonl",
+        budget=PHASE_BUDGET_USD,
+        free_bytes=free_bytes,
+        committed_spend=spend_reader,
+    )
+
+
+def record_task_result(
+    config: ProfileConfig,
+    task: ProfileTask,
+    result: Mapping[str, object],
+    versions: ProfileVersions,
+) -> None:
+    task_id: Final = result.get("task_id")
+    manifest_sha256: Final = result.get("manifest_sha256", config.manifest_sha256)
+    if task_id != task.task_id or not isinstance(manifest_sha256, str):
+        raise ValueError("task result must match the scheduled task and include manifest_sha256")
+    selected_ids, actual_sha256 = manifest_selection(config.manifest_path, config.manifest_sha256)
+    validate_manifest_attempt(task.task_id, manifest_sha256, selected_ids, actual_sha256)
+    model_alias: Final = result.get("model")
+    if not isinstance(model_alias, str) or model_alias not in task.model_order:
+        raise ValueError("task result model must match the scheduled model order")
+    returned_model: Final = _single_result_value(result, "returned_models") or "unreturned"
+    deployment_id: Final = _single_result_value(result, "deployment_ids")
+    error_class: Final = result.get("error_class")
+    is_infra: Final = error_class in ("infra", "accounting")
+    success: Final = result.get("success")
+    if not is_infra and not isinstance(success, bool):
+        raise ValueError("graded task results must include a boolean success")
+    solve_latency: Final = result.get("solve_latency_s", 0.0)
+    eval_latency: Final = result.get("eval_latency_s", 0.0)
+    latency: Final = float(solve_latency) + float(eval_latency)
+    harness_sha256: Final = hashlib.sha256(Path(run_pilot.__file__).read_bytes()).hexdigest()
+    run_pilot.append_jsonl(
+        config.runs_path,
+        {
+            **result,
+            "task_id": task.task_id,
+            "tenant_id": config.tenant_id,
+            "category": task.category,
+            "model_alias": model_alias,
+            "returned_model_version": returned_model,
+            "deployment_id": deployment_id,
+            "harness_version": versions.harness_version,
+            "budget_version": versions.budget_version,
+            "harness_config_hash": versions.harness_config_hash,
+            "is_infra": is_infra,
+            "success": None if is_infra else success,
+            "cost_usd": result.get("cost_usd", 0.0),
+            "latency_s": latency,
+            "primary_label": task.category,
+            "secondary_label": None,
+            "manifest_sha256": manifest_sha256,
+            "harness_sha256": harness_sha256,
+            "model_order": task.model_order,
+            "returned_model": returned_model,
+            "x_litellm_model_id": deployment_id,
+            "error_class": error_class,
+            "raw_result": dict(result),
+        },
+    )
+
+
+def _profile_versions(root_dir: Path) -> ProfileVersions:
+    model_manifest: Final = _mapping(
+        json.loads((root_dir / "poc" / "pilot_models.json").read_text()),
+        "pilot model manifest",
+    )
+    solver_data: Final = _mapping(model_manifest.get("solvers"), "pilot solvers")
+    model_versions: Final = tuple(
+        (alias, str(_mapping(solver_data.get(alias), alias)["gateway_model"])) for alias in MODEL_ORDER
+    )
+    model_deployments: Final = tuple(
+        (alias, str(_mapping(solver_data.get(alias), alias)["deployment_id"])) for alias in MODEL_ORDER
+    )
+    run_pilot_sha256: Final = hashlib.sha256((root_dir / "poc" / "run_pilot.py").read_bytes()).hexdigest()
+    config: Final = HarnessConfig(
+        harness_version="run_pilot_attempt_v1",
+        budget_version="budget_1usd_50_600_v1",
+        per_run_budget_usd=1.0,
+        step_limit=50,
+        wall_timeout_s=600,
+        mini_swe_agent_version=run_pilot._package_version("mini-swe-agent"),
+        guard_sha256=run_pilot_sha256,
+    )
+    return ProfileVersions(
+        harness_version=config.harness_version,
+        budget_version=config.budget_version,
+        harness_config_hash=harness_config_sha256(config),
+        model_versions=model_versions,
+        model_deployments=model_deployments,
+    )
+
+
+def _write_checkpoint(config: ProfileConfig, batch: ProfileBatch, versions: ProfileVersions) -> None:
+    config.profile_dir.mkdir(parents=True, exist_ok=True)
+    config.runs_path.touch(exist_ok=True)
+    capability_map: Final = compile_map_from_profiling_runs(
+        config.runs_path,
+        config.manifest_path,
+        config.manifest_sha256,
+        config.tenant_id,
+        versions,
+    )
+    serialized_map: Final = json.dumps(capability_map, indent=2, sort_keys=True) + "\n"
+    (config.profile_dir / f"capability_map_batch_{batch.index:03d}.json").write_text(serialized_map)
+    (config.profile_dir / "capability_map_intermediate.json").write_text(serialized_map)
+    progress: Final = {
+        "dry_run": True,
+        "manifest_sha256": config.manifest_sha256,
+        "tenant_id": config.tenant_id,
+        "completed_batches": batch.index,
+        "next_batch": batch.index + 1,
+        "tasks": [
+            {
+                "task_id": task.task_id,
+                "category": task.category,
+                "model_order": task.model_order,
+                "status": "planned",
+                "preflight_status": "not_run_dry_run",
+            }
+            for task in batch.tasks
+        ],
+    }
+    (config.profile_dir / "progress.json").write_text(json.dumps(progress, indent=2) + "\n")
+
+
+def _completed_batches(config: ProfileConfig) -> int:
+    path: Final = config.profile_dir / "progress.json"
+    if not path.exists():
+        return 0
+    progress: Final = _mapping(json.loads(path.read_text()), "progress")
+    if progress.get("manifest_sha256") != config.manifest_sha256:
+        raise ValueError("progress manifest sha256 mismatch")
+    if progress.get("tenant_id") != config.tenant_id:
+        raise ValueError("progress tenant mismatch")
+    completed: Final = progress.get("completed_batches")
+    if not isinstance(completed, int) or isinstance(completed, bool) or completed < 0:
+        raise ValueError("progress completed_batches must be a non-negative integer")
+    return completed
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Validate and checkpoint profiling batches without launching solves")
+    parser.add_argument("--root", type=Path, default=ROOT)
+    parser.add_argument("--manifest", type=Path)
+    parser.add_argument("--manifest-sha256", default=MANIFEST_SHA256)
+    parser.add_argument("--tenant-id", default=PUBLIC_DEV_TENANT)
+    parser.add_argument("--max-batches", type=int)
+    arguments = parser.parse_args()
+    default_manifest: Final = arguments.root / "profiling" / "profiling_manifest.json"
+    manifest_path: Final = (
+        arguments.manifest
+        if arguments.manifest is not None
+        else default_manifest
+        if default_manifest.exists()
+        else arguments.root / "profiling_manifest.json"
+    )
+    config: Final = ProfileConfig(arguments.root, manifest_path, arguments.manifest_sha256, arguments.tenant_id)
+    completed_batches: Final = _completed_batches(config)
+    pending_batches: Final = profile_batches(config)[completed_batches:]
+    batches: Final = pending_batches if arguments.max_batches is None else pending_batches[: arguments.max_batches]
+    if arguments.max_batches is not None and arguments.max_batches < 1:
+        raise ValueError("max_batches must be positive")
+    versions: Final = _profile_versions(config.root_dir)
+    for batch in batches:
+        _write_checkpoint(config, batch, versions)
+    print(
+        json.dumps(
+            {
+                "dry_run": True,
+                "batches": len(batches),
+                "planned_runs": sum(len(batch.tasks) * len(MODEL_ORDER) for batch in batches),
+                "workers": WORKERS,
+            }
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

cookbook/capability_routing_pilot/profiling/SHA256SUMS

@@ -1,4 +1,4 @@
 fd9fc7ba96711dfc9f2befbd10759c88fe6fa80487741c9c8190c501774fdec4  profiling_manifest_draft.json
 15720dee12d40b62780837ed12a403f3951e9c5bb1ed18ec0d5d5d1f4776bff4  profiling_manifest.json
-04eda96c46cab341e71fc6b691298268d227d86abab14cd3d5cfd4496eef6ef6  SUMMARY.md
-116fc2bebcc8197270bdb5c6449d7cba93f8f2c8e0e0c3b1f3ee904069ebbc81  ../PROFILING_PLAN.md
+634ecebaf14051e151efa326ddca2526435b801868fb521713e9586c1e7c88c7  SUMMARY.md
+ad4d97bafedd8ba2fba4a030dd9a8dde923b5e6c9133714751e6777f0baf8edf  ../PROFILING_PLAN.md

cookbook/capability_routing_pilot/profiling/SUMMARY.md

@@ -1,12 +1,12 @@
 # Profiling discovery summary
 
-The label pass completed for 220 candidates at a total cost of $0.109114. Counts were repository_debugging 158, feature_implementation 61, shell_build 0, algorithms_data 0, and unknown 1. Existing label records were reused; no model calls were made in this phase
+The label pass completed for 220 candidates at a total cost of $0.109114. Counts were repository_debugging 158, feature_implementation 61, shell_build 0, algorithms_data 0, and unknown 1. The runner and manifest-guard work made no additional model calls
 
 The image inventory checked 80 of 220 candidates using manifest lookups only. It found 62 accessible images, 0 confirmed unavailable images, and 18 HTTP 401 access errors. Each error received one fresh anonymous token scoped to its repository and one HEAD retry. All 18 retries still returned 401, and each decoded token had an empty repository access claim. A single `docker manifest inspect` on one candidate also returned unauthorized. An empty claim can mean a missing or private repository, so absence is not confirmed. These candidates are recorded as `access_error` exclusions, not unavailable. No further discovery was needed because both supported categories already had at least 30 accessible images
 
 Per-category counts are:
 
-| Category | Labeled | Checked | Available | Unavailable | Errors | Not inspected | Draft selected | Shortfall |
+| Category | Labeled | Checked | Available | Unavailable | Errors | Not inspected | Frozen selected | Shortfall |
 | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |
 | repository_debugging | 158 | 40 | 32 | 0 | 8 | 118 | 30 | 0 |
 | feature_implementation | 61 | 40 | 30 | 0 | 10 | 21 | 30 | 0 |
@@ -14,7 +14,7 @@ Per-category counts are:
 | algorithms_data | 0 | 0 | 0 | 0 | 0 | 0 | 0 | unsupported |
 | unknown | 1 | 0 | 0 | 0 | 0 | 1 | 0 | excluded |
 
-The frozen manifest selects the first 30 accessible images in each supported category, ordered by `sha256(seed + task_id)`. It was frozen at `2026-10-21T03:24:04.314265+00:00`. The 18 `access_error` exclusions and retry results are included in the manifest. Individual timestamps for the original successful checks were not persisted, so the freeze timestamp records when the available inventory was finalized
+The frozen manifest selects the first 30 accessible images in each supported category, ordered by `sha256(seed + task_id)`. It was frozen at `2026-10-21T03:24:04.314265+00:00` with SHA-256 `15720dee12d40b62780837ed12a403f3951e9c5bb1ed18ec0d5d5d1f4776bff4`. The 18 `access_error` exclusions and retry results are included in the manifest. Individual timestamps for the original successful checks were not persisted, so the freeze timestamp records when the available inventory was finalized
 
 No 429 response was recorded. `RateLimit-Remaining` and `Retry-After` were absent from all 18 retry responses and are persisted as null. No credentials were used or recorded
 

cookbook/capability_routing_pilot/tests/test_profile_manifest.py

@@ -10,15 +10,13 @@ from poc.profile_manifest import (
 
 
 def test_manifest_selects_deterministic_public_images_and_marks_unsupported_categories() -> None:
-    labels = tuple(
-        CandidateLabel(f"repo-{index}", "repository_debugging", None)
-        for index in range(35)
-    ) + tuple(
-        CandidateLabel(f"feature-{index}", "feature_implementation", None)
-        for index in range(32)
-    ) + (
-        CandidateLabel("shell-1", "shell_build", None),
-        CandidateLabel("unknown-1", "unknown", None),
+    labels = (
+        tuple(CandidateLabel(f"repo-{index}", "repository_debugging", None) for index in range(35))
+        + tuple(CandidateLabel(f"feature-{index}", "feature_implementation", None) for index in range(32))
+        + (
+            CandidateLabel("shell-1", "shell_build", None),
+            CandidateLabel("unknown-1", "unknown", None),
+        )
     )
     images = tuple(
         ImageAvailability(
@@ -32,7 +30,11 @@ def test_manifest_selects_deterministic_public_images_and_marks_unsupported_cate
         )
         for label in labels
     )
-    manifest = build_draft_manifest(tuple(reversed(labels)), tuple(reversed(images)))
+    manifest = build_draft_manifest(
+        tuple(reversed(labels)),
+        tuple(reversed(images)),
+        frozen_at_utc="2026-10-21T00:00:00+00:00",
+    )
     categories = manifest["categories"]
     repo_expected = tuple(
         sorted(
@@ -46,7 +48,8 @@ def test_manifest_selects_deterministic_public_images_and_marks_unsupported_cate
             key=lambda task_id: (hashlib.sha256(f"{SEED}{task_id}".encode()).hexdigest(), task_id),
         )[:MAX_PER_CATEGORY]
     )
-    assert manifest["frozen"] is False
+    assert manifest["frozen"] is True
+    assert manifest["frozen_at_utc"] == "2026-10-21T00:00:00+00:00"
     assert categories["repository_debugging"]["selected_ids"] == repo_expected
     assert categories["feature_implementation"]["selected_ids"] == feature_expected
     assert categories["shell_build"]["selection_status"] == "unsupported_category"
@@ -65,7 +68,21 @@ def test_manifest_reports_image_unavailable_and_lookup_errors_without_selecting_
     images = (
         ImageAvailability("available", "image/a", True, "sha256:a", 100, None, True),
         ImageAvailability("unavailable", "image/b", False, None, None, None, True),
-        ImageAvailability("error", "image/c", None, None, None, "http_429", True),
+        ImageAvailability(
+            "error",
+            "image/c",
+            None,
+            None,
+            None,
+            "access_error",
+            True,
+            token_scope_ok=False,
+            retry_status=401,
+            auth_rate_limit_remaining="17",
+            auth_retry_after="2",
+            rate_limit_remaining="41",
+            retry_after="3",
+        ),
     )
     image_by_id = {image.task_id: image for image in images}
     manifest = build_draft_manifest(labels, images)
@@ -85,15 +102,24 @@ def test_manifest_reports_image_unavailable_and_lookup_errors_without_selecting_
             ),
         )
     )
-    blocker = next(
-        task_id
-        for task_id in ordered_ids
-        if task_id not in {image.task_id for image in images} or image_by_id[task_id].exists is None
-    )
     assert counts["selected_ids"] == tuple(
         task_id
-        for task_id in ordered_ids[: ordered_ids.index(blocker)]
-        if image_by_id[task_id].exists is True
+        for task_id in ordered_ids
+        if image_by_id.get(task_id) is not None and image_by_id[task_id].exists is True
     )
     assert manifest["image_unavailable_count"] == 1
     assert manifest["image_lookup_error_count"] == 1
+    assert manifest["error_exclusions"] == (
+        {
+            "task_id": "error",
+            "category": "repository_debugging",
+            "status": "access_error",
+            "reason": "access_error",
+            "retry_status": 401,
+            "token_scope_ok": False,
+            "auth_rate_limit_remaining": "17",
+            "auth_retry_after": "2",
+            "rate_limit_remaining": "41",
+            "retry_after": "3",
+        },
+    )

cookbook/capability_routing_pilot/tests/test_profiles.py

@@ -1,3 +1,6 @@
+import hashlib
+import json
+from dataclasses import asdict
 from pathlib import Path
 
 import pytest
@@ -192,7 +195,15 @@ def test_missing_zero_and_insufficient_evidence_are_not_selected() -> None:
     assert no_evidence.status is SupportStatus.NO_EVIDENCE
     assert zero_success.status is SupportStatus.UNSUPPORTED_ZERO_SUCCESS
     assert insufficient.status is SupportStatus.INSUFFICIENT_EVIDENCE
-    assert select(PUBLIC_DEV_TENANT, "repository_debugging", _versions(), (),).decision is DecisionStatus.NO_QUALIFIED_MODEL
+    assert (
+        select(
+            PUBLIC_DEV_TENANT,
+            "repository_debugging",
+            _versions(),
+            (),
+        ).decision
+        is DecisionStatus.NO_QUALIFIED_MODEL
+    )
     assert (
         select(
             PUBLIC_DEV_TENANT,
@@ -396,17 +407,76 @@ def test_capability_map_is_intermediate_and_contains_bound_and_version_notes() -
     assert capability_map["versions"]["harness_config_hash"] == CONFIG_HASH
 
 
-def test_map_builder_rejects_old_runs_jsonl_and_reads_only_profiling_runs(tmp_path: Path) -> None:
+def test_map_builder_checks_manifest_membership_and_digest(tmp_path: Path) -> None:
     versions = _versions()
-    old_runs = tmp_path / "runs.jsonl"
-    old_runs.write_text("")
-    capability_map = compile_map_from_profiling_runs(old_runs, PUBLIC_DEV_TENANT, versions)
-    assert capability_map["profiles"] == []
-    aliased_old_runs = tmp_path / "profiling_runs.jsonl"
-    aliased_old_runs.symlink_to(old_runs)
-    with pytest.raises(ValueError, match="only read profiling_runs.jsonl"):
-        compile_map_from_profiling_runs(aliased_old_runs, PUBLIC_DEV_TENANT, versions)
+    manifest = tmp_path / "manifest.json"
+    manifest_bytes = json.dumps({"categories": {"repository_debugging": {"selected_ids": ["task-0"]}}}).encode()
+    manifest.write_bytes(manifest_bytes)
+    manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest()
+    attempt = _attempts("haiku", 1, 1, 0.1)[0]
+    attempt_row = {
+        **asdict(attempt),
+        **asdict(attempt.key),
+        "manifest_sha256": manifest_sha256,
+    }
+    profiling_runs = tmp_path / "profiling_runs.jsonl"
+    profiling_runs.write_text(json.dumps(attempt_row) + "\n")
+
+    capability_map = compile_map_from_profiling_runs(
+        profiling_runs,
+        manifest,
+        manifest_sha256,
+        PUBLIC_DEV_TENANT,
+        versions,
+    )
+
+    assert capability_map["profiles"][0]["graded_count"] == 1
     assert capability_map["final"] is False
+    with pytest.raises(ValueError, match="manifest sha256 mismatch"):
+        compile_map_from_profiling_runs(
+            profiling_runs,
+            manifest,
+            "wrong-digest",
+            PUBLIC_DEV_TENANT,
+            versions,
+        )
+
+
+@pytest.mark.parametrize(
+    ("field", "value", "message"),
+    (
+        ("task_id", "outside-manifest", "not selected"),
+        ("manifest_sha256", "wrong-digest", "attempt manifest sha256 mismatch"),
+    ),
+)
+def test_map_builder_rejects_attempt_membership_or_digest_mismatch(
+    tmp_path: Path,
+    field: str,
+    value: str,
+    message: str,
+) -> None:
+    manifest = tmp_path / "manifest.json"
+    manifest_bytes = json.dumps({"categories": {"repository_debugging": {"selected_ids": ["task-0"]}}}).encode()
+    manifest.write_bytes(manifest_bytes)
+    manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest()
+    attempt = _attempts("haiku", 1, 1, 0.1)[0]
+    attempt_row = {
+        **asdict(attempt),
+        **asdict(attempt.key),
+        "manifest_sha256": manifest_sha256,
+        field: value,
+    }
+    profiling_runs = tmp_path / "profiling_runs.jsonl"
+    profiling_runs.write_text(json.dumps(attempt_row) + "\n")
+
+    with pytest.raises(ValueError, match=message):
+        compile_map_from_profiling_runs(
+            profiling_runs,
+            manifest,
+            manifest_sha256,
+            PUBLIC_DEV_TENANT,
+            _versions(),
+        )
 
 
 def test_harness_hash_is_canonical_json_over_the_full_budget_config() -> None:

cookbook/capability_routing_pilot/tests/test_run_profiling.py

@@ -0,0 +1,121 @@
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+
+from poc.run_profiling import (
+    PHASE_BUDGET_USD,
+    ProfileConfig,
+    ProfileTask,
+    model_order_for_task,
+    profile_batches,
+    profile_guard,
+    record_task_result,
+    require_disk_for_pull,
+)
+from poc.profiles import ProfileVersions
+
+
+def _config(tmp_path: Path, task_ids: tuple[str, ...] = ("task-0", "task-1", "task-2", "task-3")) -> ProfileConfig:
+    manifest = tmp_path / "profiling_manifest.json"
+    manifest_bytes = json.dumps(
+        {
+            "categories": {
+                "repository_debugging": {"selected_ids": task_ids},
+                "feature_implementation": {"selected_ids": []},
+                "shell_build": {"selected_ids": []},
+                "algorithms_data": {"selected_ids": []},
+                "unknown": {"selected_ids": []},
+            }
+        }
+    ).encode()
+    manifest.write_bytes(manifest_bytes)
+    return ProfileConfig(
+        root_dir=tmp_path,
+        manifest_path=manifest,
+        manifest_sha256=hashlib.sha256(manifest_bytes).hexdigest(),
+    )
+
+
+def test_max_batches_preserves_manifest_order_and_balances_model_order(tmp_path: Path) -> None:
+    config = _config(tmp_path)
+
+    batches = profile_batches(config, max_batches=1)
+
+    assert len(batches) == 1
+    assert tuple(task.task_id for task in batches[0].tasks) == ("task-0", "task-1", "task-2")
+    assert tuple(model_order_for_task(index) for index in range(3)) == (
+        ("haiku-4.5", "sonnet-5", "opus-5"),
+        ("sonnet-5", "opus-5", "haiku-4.5"),
+        ("opus-5", "haiku-4.5", "sonnet-5"),
+    )
+    assert batches[0].tasks[1].model_order == model_order_for_task(1)
+
+
+def test_runner_record_boundary_checks_membership_and_manifest_sha(tmp_path: Path) -> None:
+    config = _config(tmp_path, ("task-0",))
+    task = ProfileTask(0, "task-0", "repository_debugging", model_order_for_task(0))
+    versions = ProfileVersions("harness-v1", "budget-v1", "config-hash", ())
+
+    with pytest.raises(ValueError, match="scheduled task"):
+        record_task_result(
+            config,
+            task,
+            {"task_id": "outside-manifest", "manifest_sha256": config.manifest_sha256},
+            versions,
+        )
+    outside_task = ProfileTask(1, "outside-manifest", "repository_debugging", model_order_for_task(1))
+    with pytest.raises(ValueError, match="not selected in the manifest"):
+        record_task_result(
+            config,
+            outside_task,
+            {"task_id": "outside-manifest", "manifest_sha256": config.manifest_sha256},
+            versions,
+        )
+    with pytest.raises(ValueError, match="attempt manifest sha256 mismatch"):
+        record_task_result(
+            config,
+            task,
+            {"task_id": "task-0", "manifest_sha256": "wrong-digest"},
+            versions,
+        )
+
+    record_task_result(
+        config,
+        task,
+        {
+            "task_id": "task-0",
+            "manifest_sha256": config.manifest_sha256,
+            "model": "haiku-4.5",
+            "returned_models": ["anthropic/claude-haiku-4-5-20251001"],
+            "deployment_ids": ["deployment-1"],
+            "error_class": "solver",
+            "success": True,
+            "cost_usd": 0.25,
+            "solve_latency_s": 3.0,
+            "eval_latency_s": 2.0,
+        },
+        versions,
+    )
+    result = json.loads(config.runs_path.read_text())
+    assert result["tenant_id"] == "public_dev_poc"
+    assert result["category"] == "repository_debugging"
+    assert result["model_order"] == ["haiku-4.5", "sonnet-5", "opus-5"]
+    assert result["returned_model_version"] == "anthropic/claude-haiku-4-5-20251001"
+    assert result["x_litellm_model_id"] == "deployment-1"
+    assert result["raw_result"]["model"] == "haiku-4.5"
+
+
+def test_profile_guard_enforces_phase_cap_and_pre_pull_disk_reserve(tmp_path: Path) -> None:
+    config = _config(tmp_path, ("task-0",))
+    guard = profile_guard(
+        config,
+        free_bytes=lambda: 100 * 1024**3,
+        committed_spend=lambda: PHASE_BUDGET_USD,
+    )
+
+    assert guard.acquire("task-0") is False
+    require_disk_for_pull(60 * 1024**3)
+    with pytest.raises(OSError, match="less than 60 GiB"):
+        require_disk_for_pull(60 * 1024**3 - 1)