CortexLM · echobt · Sep 10, 2026 · Sep 10, 2026
@@ -34,7 +34,8 @@ Harbor **does not** accept a filesystem path for `-a`. This adaptor therefore
 A one-line `import_path` file inside the agent dir (contents
 `module.path:ClassName`) wins when several classes exist. A built-in name in
 that file is refused: that would ignore miner code the same way `terminus-2`
-did.
+did. The named module is resolved in the evaluate import env (artefact
+parent only) and **rejected** if its origin is outside the staged artefact.
 
 ## Agent selection (miner attach surface)
 
@@ -81,7 +82,9 @@ Off-limits in the artefact (inspect fails the named rule):
 - `no_eval_short_circuit` (and `skip_eval` / `skip_verifier` / `always_pass_eval` / `short_circuit_eval`)
 - `no_tb4_hardcoding` (and `tb4_answers` / `hardcoded_tb4`)
 
-Do not quote those markers in miner code or README inside the tar.
+A file/byte-limit truncation marks the scan incomplete and fails those
+off-limits rules. Unknown rule ids fail closed. Do not quote those markers
+in miner code or README inside the tar.
 
 ## BYOK
 
@@ -137,8 +140,10 @@ the task list.
 }
 ```
 
-`primary_value` is the mean of Harbor trial `verifier_result.rewards.reward`
-only. No measured trial → fail closed, no invented number.
+`primary_value` is the mean of **every** Harbor trial
+`verifier_result.rewards.reward`. Evidence may truncate the serialized
+trial list; the mean does not. No measured trial, or a nonzero Harbor
+exit, → fail closed, no invented number and no leftover `report.json`.
 
 `inspect` writes `$PROOF_OUTPUT_DIR/checklist.json` (no Harbor, no keys).
 

@@ -62,6 +62,13 @@ set +e
 harbor_exit=$?
 set -e
 
+# Nonzero Harbor is an incomplete or failed trial set. Do not invoke the
+# summarizer: a leftover report.json would be a successful score.
+if [ "$harbor_exit" -ne 0 ]; then
+    rm -f "$PROOF_OUTPUT_DIR/report.json"
+    proof_die "harbor exited $harbor_exit; refusing to score a partial or failed run"
+fi
+
 python3 "$HERE/summarize.py" \
     --jobs-dir "$jobs_dir" \
     --log "$log" \

@@ -1,13 +1,16 @@
 #!/usr/bin/env python3
 """Turn a Harbor jobs directory into Proof ``report.json``.
 
-``primary_value`` is the mean of trial ``verifier_result.rewards.reward``
-values that are finite numbers. A trial with no such field is **no
-measurement** — never another field's value (score, accuracy, job-level
-aggregates) as a substitute. Zero measured trials → exit 2, no report.
-
-Evidence includes ``trials`` and a redacted ``harbor_run_tail``. Secret
-values from the owner secrets dir and miner BYOK dir are blanked.
+``primary_value`` is the mean of **every** trial
+``verifier_result.rewards.reward`` that is a finite number. A trial with
+no such field is **no measurement** — never another field's value (score,
+accuracy, job-level aggregates) as a substitute. Zero measured trials →
+exit 2, no report. A nonzero Harbor exit is an incomplete run → exit 2,
+no report (do not publish a partial score).
+
+Evidence serializes at most ``MAX_EVIDENCE_TRIALS`` trial rows; the mean
+always uses the full measured set. Secret values from the owner secrets
+dir and miner BYOK dir are blanked.
 """
 
 from __future__ import annotations
@@ -60,12 +63,11 @@ def _load_json(path: Path) -> Any | None:
 
 
 def collect_trials(jobs_dir: Path) -> list[dict[str, Any]]:
+    """Load every measured trial. Do not cap here — the cap is evidence only."""
     trials: list[dict[str, Any]] = []
     if not jobs_dir.is_dir():
         return trials
     for result_path in sorted(jobs_dir.rglob("result.json")):
-        if len(trials) >= MAX_EVIDENCE_TRIALS:
-            break
         obj = _load_json(result_path)
         reward = trial_reward(obj)
         if reward is None:
@@ -171,12 +173,15 @@ def build_report(
     agent_source: str,
 ) -> dict[str, Any]:
     primary = mean_reward(trials)
+    evidence_trials = trials[:MAX_EVIDENCE_TRIALS]
     return {
         "primary_value": primary,
         "claim_holds": True,
         "evidence": {
-            "trials": trials,
+            "trials": evidence_trials,
             "n_measured": len(trials),
+            "n_evidence_trials": len(evidence_trials),
+            "evidence_truncated": len(trials) > MAX_EVIDENCE_TRIALS,
             "mean_reward": primary,
             "harbor_exit": harbor_exit,
             "harbor_run_tail": log_tail,
@@ -196,6 +201,12 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--agent-source", default="")
     args = parser.parse_args(argv)
 
+    if args.harbor_exit != 0:
+        _fail(
+            f"harbor exited {args.harbor_exit}; refusing to publish a score "
+            "from a failed or incomplete run"
+        )
+
     jobs_dir = Path(args.jobs_dir)
     trials = collect_trials(jobs_dir)
     if not trials:

@@ -4,8 +4,11 @@
 Reads ``PROOF_RULES_FILE`` (a ``RuleSet`` object or a ``[{id, text}]`` array)
 and writes ``checklist.json``. Off-limits markers ``no_eval_short_circuit``
 and ``no_tb4_hardcoding`` fail those rules when they appear in the artefact
-tree. Host-side rules are answered with evidence, not left blank (a missing
-item is recorded red). Secret file contents are never printed.
+tree. A file/byte-limit truncation marks the scan incomplete and **fails**
+those off-limits rules — truncated absence is not a clean pass. Unknown
+rule IDs fail closed. Host-side rules are answered with evidence, not left
+blank (a missing item is recorded red). Secret file contents are never
+printed.
 """
 
 from __future__ import annotations
@@ -72,23 +75,31 @@ def load_rules(path: Path) -> list[dict[str, str]]:
     return out
 
 
-def collect_artefact_text(root: Path | None) -> tuple[str, int, list[str]]:
+def collect_artefact_text(root: Path | None) -> tuple[str, int, list[str], bool]:
+    """Return ``(text, n_scanned, names, incomplete)``.
+
+    ``incomplete`` is true when a file or byte cap stopped the walk before
+    every regular file was considered. Callers must not treat a truncated
+    scan as proof that an off-limits marker is absent.
+    """
     if root is None or not root.is_dir():
-        return "", 0, []
+        return "", 0, [], False
     blobs: list[str] = []
     names: list[str] = []
     total = 0
     n_files = 0
+    incomplete = False
     for path in sorted(root.rglob("*")):
         if not path.is_file():
             continue
         rel = path.relative_to(root).as_posix()
         if ".." in Path(rel).parts:
             continue
+        if n_files >= MAX_FILES or total >= MAX_TOTAL_BYTES:
+            incomplete = True
+            break
         names.append(rel)
         n_files += 1
-        if n_files > MAX_FILES or total > MAX_TOTAL_BYTES:
-            break
         try:
             size = path.stat().st_size
         except OSError:
@@ -103,19 +114,31 @@ def collect_artefact_text(root: Path | None) -> tuple[str, int, list[str]]:
             continue
         total += len(data)
         blobs.append(data.decode("utf-8", errors="replace"))
-    return "\n".join(blobs).lower(), n_files, names
+    return "\n".join(blobs).lower(), n_files, names, incomplete
 
 
 def _contains_any(haystack: str, needles: tuple[str, ...]) -> list[str]:
     return [n for n in needles if n.lower() in haystack]
 
 
+def _off_limits_incomplete(rid: str, n_files: int) -> dict[str, Any]:
+    return {
+        "id": rid,
+        "pass": False,
+        "evidence": _clip(
+            f"artefact scan incomplete (file/byte limit); cannot treat absence "
+            f"of off-limits markers as clean ({n_files} files scanned)"
+        ),
+    }
+
+
 def tick_rule(
     rule: dict[str, str],
     artefact_text: str,
     n_files: int,
     names: list[str],
     has_artefact: bool,
+    incomplete: bool = False,
 ) -> dict[str, Any]:
     rid = rule["id"]
     joined_names = " ".join(names).lower()
@@ -132,6 +155,8 @@ def tick_rule(
                     f"off-limits eval short-circuit marker in artefact: {', '.join(hits)}"
                 ),
             }
+        if incomplete:
+            return _off_limits_incomplete(rid, n_files)
         return {
             "id": rid,
             "pass": True,
@@ -150,6 +175,8 @@ def tick_rule(
                     f"off-limits tb4 hardcoding marker in artefact: {', '.join(hits)}"
                 ),
             }
+        if incomplete:
+            return _off_limits_incomplete(rid, n_files)
         return {
             "id": rid,
             "pass": True,
@@ -184,20 +211,12 @@ def tick_rule(
             ),
         }
 
-    if not has_artefact:
-        return {
-            "id": rid,
-            "pass": False,
-            "evidence": _clip(
-                f"rule {rid}: no artefact tree staged for inspect (PROOF_ARTIFACT_DIR unset or empty)"
-            ),
-        }
     return {
         "id": rid,
-        "pass": True,
+        "pass": False,
         "evidence": _clip(
-            f"rule {rid}: artefact present ({n_files} files); no paid inference; "
-            "no off-limits short-circuit/tb4 markers beyond the dedicated probes"
+            f"rule {rid}: unsupported/unknown rule id; inspect fails closed "
+            f"(artefact_present={has_artefact}, files={n_files})"
         ),
     }
 
@@ -213,10 +232,11 @@ def main(argv: list[str] | None = None) -> int:
     artefact_raw = os.environ.get("PROOF_ARTIFACT_DIR", "")
     artefact_dir = Path(artefact_raw) if artefact_raw else None
     has_artefact = artefact_dir is not None and artefact_dir.is_dir()
-    text, n_files, names = collect_artefact_text(artefact_dir)
+    text, n_files, names, incomplete = collect_artefact_text(artefact_dir)
 
     items = [
-        tick_rule(rule, text, n_files, names, has_artefact) for rule in load_rules(rules_path)
+        tick_rule(rule, text, n_files, names, has_artefact, incomplete)
+        for rule in load_rules(rules_path)
     ]
     out = output_dir / "checklist.json"
     out.write_text(json.dumps(items, indent=2) + "\n", encoding="utf-8")

@@ -93,7 +93,10 @@ proof_select_harbor_agent() {
         pythonpath="$(printf '%s\n' "$resolved" | sed -n 's/^pythonpath=//p' | head -n1)"
         [ -n "$import_path" ] || return 1
         if [ -n "$pythonpath" ]; then
-            export PYTHONPATH="${pythonpath}${PYTHONPATH:+:$PYTHONPATH}"
+            # Evaluate import env is the staged artefact parent only.
+            # Inherited PYTHONPATH would let a miner import_path name a
+            # module that lives outside the artefact.
+            export PYTHONPATH="$pythonpath"
         fi
         chosen="$import_path"
         source="$label"

@@ -9,19 +9,24 @@
 
 This helper turns an artefact directory into ``module:Class`` so evaluate can
 pass ``-a`` without silently falling back to the topic's built-in agent.
-Miner code is **not** executed: class discovery is AST-only.
+Miner code is **not** executed: class discovery is AST-only. A named
+``import_path`` is resolved in the evaluate import env (artefact parent
+only) and rejected when the origin is outside the staged artefact.
 """
 
 from __future__ import annotations
 
 import argparse
 import ast
+import importlib.machinery
+import os
 import sys
 from pathlib import Path
 
 AGENT_BASES = frozenset({"BaseAgent", "BaseInstalledAgent"})
 MAX_PY_BYTES = 256 * 1024
 MAX_PY_FILES = 64
+_NON_FILE_ORIGINS = frozenset({"built-in", "frozen"})
 
 
 def _fail(msg: str, code: int = 2) -> None:
@@ -48,11 +53,82 @@ def _read_import_path_file(agent_dir: Path) -> str | None:
             )
         if ".." in line or "/" in line or "\\" in line:
             _fail(f"{path} is not a Python import path: {line!r}")
+        _assert_import_origin(line, agent_dir)
         return line
     _fail(f"{path} is empty")
     return None
 
 
+def _containment_root(agent_dir: Path) -> Path:
+    raw = os.environ.get("PROOF_ARTIFACT_DIR", "").strip()
+    if raw:
+        art = Path(raw)
+        if art.is_dir():
+            return art.resolve()
+    return agent_dir.resolve()
+
+
+def _is_inside(path: Path, root: Path) -> bool:
+    try:
+        path.resolve().relative_to(root.resolve())
+        return True
+    except (OSError, ValueError):
+        return False
+
+
+def _module_file_candidates(search_root: Path, module_name: str) -> list[Path]:
+    parts = [p for p in module_name.split(".") if p]
+    if not parts or any(p in {".", ".."} or not p.isidentifier() for p in parts):
+        return []
+    base = search_root.joinpath(*parts)
+    return [Path(str(base) + ".py"), base / "__init__.py"]
+
+
+def _resolve_module_origin(module_name: str, search_root: Path) -> Path | None:
+    """Resolve ``module_name`` using only ``search_root`` (no inherited PYTHONPATH).
+
+    Miner code is not executed. PathFinder + on-disk candidates only.
+    """
+    try:
+        spec = importlib.machinery.PathFinder.find_spec(module_name, [str(search_root)])
+    except KeyError:
+        # Parent package is not on sys.modules (implicit namespace / first look).
+        spec = None
+    if spec is not None and isinstance(spec.origin, str) and spec.origin not in _NON_FILE_ORIGINS:
+        origin = Path(spec.origin)
+        if origin.is_file():
+            return origin
+    for candidate in _module_file_candidates(search_root, module_name):
+        if candidate.is_file():
+            return candidate
+    return None
+
+
+def _assert_import_origin(import_path: str, agent_dir: Path) -> None:
+    """Reject ``module:Class`` whose origin is outside the staged artefact."""
+    module_name, sep, class_name = import_path.partition(":")
+    if not sep or not module_name or not class_name:
+        _fail(f"{import_path!r} is not a Python import path module.path:ClassName")
+    if not class_name.isidentifier():
+        _fail(f"{import_path!r} class name is not a Python identifier")
+    search_root = agent_dir.resolve().parent
+    origin = _resolve_module_origin(module_name, search_root)
+    if origin is None:
+        _fail(
+            f"{import_path} does not resolve to a module under {search_root} "
+            "(inherited PYTHONPATH / stdlib / Harbor installs are not miner code)"
+        )
+    bound = _containment_root(agent_dir)
+    if not _is_inside(origin, bound):
+        _fail(
+            f"{import_path} origin {origin.resolve()} is outside staged artefact {bound}"
+        )
+    if not _is_inside(origin, agent_dir) and not _is_inside(origin, search_root):
+        _fail(
+            f"{import_path} origin {origin.resolve()} is outside the evaluate import root"
+        )
+
+
 def _base_attr(node: ast.expr) -> str | None:
     if isinstance(node, ast.Name):
         return node.id
@@ -137,6 +213,7 @@ def discover(agent_dir: Path) -> tuple[str, str]:
             f"{agent_dir} has multiple Harbor agent classes ({', '.join(unique_paths)}); "
             "write a one-line import_path file (module.path:ClassName) to choose"
         )
+    _assert_import_origin(discovered[0][0], agent_dir)
     return discovered[0]