"""Operator-invoked, standard-library diagnostic harness. No hosted models or keys.""" from __future__ import annotations import argparse from copy import deepcopy from datetime import datetime, timezone import json import math import os from pathlib import Path import subprocess import time CONDITIONS = ("continuation", "simple_pause", "pit_stop", "processing_control") RATES = ("task_accuracy", "context_retention", "contradiction_rate", "error_recovery") USAGE = ("input_tokens", "output_tokens", "total_tokens", "cost_amount", "cost_currency") RUNNER_SCHEMA = "eclypse.recovery-runner.v1" def condition_order(seed, block_id, conditions): state = 2166136261 for byte in f"{seed}:{block_id}".encode(): state = ((state ^ byte) * 16777619) & 0xffffffff state = state or 0x6d2b79f5 result = list(conditions) for i in range(len(result) - 1, 0, -1): state ^= (state << 13) & 0xffffffff state ^= state >> 17 state ^= (state << 5) & 0xffffffff state &= 0xffffffff j = state % (i + 1) result[i], result[j] = result[j], result[i] return result def now(): return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") def unknown_usage(): return dict.fromkeys(USAGE) def missing_observation(form, error=None): return {"status": "missing", "started_at": None, "ended_at": None, "form_id": form, "counts": {key: {"successes": None, "opportunities": None} for key in RATES}, "latency_ms": None, "usage": unknown_usage(), "error": error} def validate_usage(value): if not isinstance(value, dict) or set(value) != set(USAGE): raise ValueError("invalid_usage") for key in USAGE[:-1]: number = value[key] if number is not None and (type(number) not in (int, float) or not math.isfinite(number) or number < 0 or (key != "cost_amount" and (type(number) is not int or number > 9007199254740991))): raise ValueError("invalid_usage") if value["cost_currency"] is not None and (not isinstance(value["cost_currency"], str) or not 1 <= len(value["cost_currency"]) <= 12): raise ValueError("invalid_currency") if value["cost_amount"] is not None and value["cost_currency"] is None: raise ValueError("cost_currency_required") if all(type(value[key]) is int for key in USAGE[:3]) and value["total_tokens"] != value["input_tokens"] + value["output_tokens"]: raise ValueError("incoherent_token_total") return deepcopy(value) def supplied_evidence(value): if not isinstance(value, dict): raise ValueError("pause_evidence_required") expected = {"request_id", "receipt_hash", "operator_authorization_recorded", "signed_receipt", "classification"} if set(value) != expected or value["classification"] != "supplied_unverified" or value["operator_authorization_recorded"] is not True: raise ValueError("supplied_unverified_pause_evidence_required") receipt = value["signed_receipt"] if not isinstance(receipt, dict) or receipt.get("kind") != "eclypse.pause_receipt" or receipt.get("request_id") != value["request_id"] or receipt.get("receipt_hash") != value["receipt_hash"]: raise ValueError("incoherent_pause_evidence") # Deliberately no signature/ownership assertion. The hosted owned lookup is separate. return deepcopy(value) def invoke(command, request, timeout, environment): started_at, start = now(), time.monotonic_ns() completed = subprocess.run(command, input=json.dumps(request, allow_nan=False), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, env=environment, shell=False, check=False) elapsed, ended_at = (time.monotonic_ns() - start) / 1_000_000, now() if completed.returncode: raise ValueError("runner_nonzero_exit") if len(completed.stdout.encode()) > 1_000_000: raise ValueError("runner_response_too_large") try: response = json.loads(completed.stdout, parse_constant=lambda _: (_ for _ in ()).throw(ValueError())) except (ValueError, TypeError): raise ValueError("runner_invalid_json") from None if not isinstance(response, dict) or response.get("schema") != RUNNER_SCHEMA: raise ValueError("runner_schema_mismatch") if set(response) - {"schema", "state", "status", "counts", "usage", "evidence"}: raise ValueError("runner_unexpected_output_field") state = response.get("state") if not isinstance(state, str) or not 1 <= len(state) <= 4096: raise ValueError("explicit_session_state_required") return response, state, started_at, ended_at, elapsed def measure(command, request, timeout, environment): response, state, started, ended, elapsed = invoke(command, request, timeout, environment) if response.get("status") not in ("ok", "failed"): raise ValueError("runner_status_required") counts = response.get("counts") if not isinstance(counts, dict) or set(counts) != set(RATES): raise ValueError("raw_counts_required") for counter in counts.values(): if not isinstance(counter, dict) or set(counter) != {"successes", "opportunities"}: raise ValueError("raw_denominators_required") for value in counter.values(): if value is not None and (type(value) is not int or value < 0 or value > 9007199254740991): raise ValueError("invalid_raw_count") if all(type(v) is int for v in counter.values()) and counter["successes"] > counter["opportunities"]: raise ValueError("raw_count_exceeds_opportunities") observation = {"status": response["status"], "started_at": started, "ended_at": ended, "form_id": request["form_id"], "counts": deepcopy(counts), "latency_ms": elapsed, "usage": validate_usage(response.get("usage")), "error": None if response["status"] == "ok" else "runner_reported_failed"} return observation, state def execute(plan, command, *, operator_authorized=False, synthetic=False, timeout=120, sleep=time.sleep, environment=None): if operator_authorized is not True: raise ValueError("explicit_operator_authorization_required") if not command or not all(isinstance(part, str) for part in command): raise ValueError("explicit_runner_command_required") report = deepcopy(plan) if report.get("schema") != "eclypse.recovery-diagnostic.v1" or not isinstance(report.get("blocks"), list): raise ValueError("report_plan_required") protocol = report["protocol"] conditions = protocol["conditions"] if (not isinstance(protocol.get("seed"), int) or not 0 <= protocol["seed"] <= 0xffffffff or not set(CONDITIONS[:3]).issubset(conditions) or len(set(conditions)) != len(conditions) or any(condition not in CONDITIONS for condition in conditions) or protocol.get("randomization") != "seeded_fisher_yates_v1"): raise ValueError("predeclared_randomization_required") report["report_kind"], report["operator_authorized"], report["created_at"] = "synthetic" if synthetic else "real", True, now() # Minimal child environment: no credential discovery or automatic API-key inheritance. child_env = environment if environment is not None else {key: os.environ[key] for key in ("PATH", "LANG", "LC_ALL", "TZ", "SYSTEMROOT") if key in os.environ} used_sessions = set() for block in report["blocks"]: order = condition_order(protocol["seed"], block["block_id"], conditions) declared = {run["condition"]: run for run in block["runs"]} if set(declared) != set(conditions) or block["order"] != order: raise ValueError("plan_order_must_match_seed") block["runs"] = [] for condition in order: original = declared[condition] session = original["session_id"] if not isinstance(session, str) or not session or session in used_sessions: raise ValueError("isolated_unique_sessions_required") used_sessions.add(session) before_form, after_form = original["before"]["form_id"], original["after"]["form_id"] if before_form == after_form or any(form not in protocol["heldout_forms"] for form in (before_form, after_form)): raise ValueError("distinct_predeclared_forms_required") intervention = {"kind": condition, "started_at": None, "ended_at": None, "actual_elapsed_ms": None, "requested_pause_seconds": protocol["budgets"][condition]["pause_seconds"], "usage": unknown_usage(), "evidence": None} run = {"condition": condition, "session_id": session, "before": missing_observation(before_form), "intervention": intervention, "after": missing_observation(after_form), "error": None} block["runs"].append(run) request = {"schema": RUNNER_SCHEMA, "block_id": block["block_id"], "condition": condition, "session_id": session, "protocol": protocol, "operator_authorized": True, "state": None} try: request.update(phase="before", form_id=before_form) run["before"], state = measure(command, request, timeout, child_env) if run["before"]["status"] != "ok": raise ValueError("before_failed") if condition in ("continuation", "simple_pause"): intervention["started_at"], start = now(), time.monotonic_ns() if condition == "simple_pause": seconds = protocol["budgets"][condition]["pause_seconds"] if type(seconds) is not int or not 0 <= seconds <= 300: raise ValueError("predeclared_pause_duration_required") sleep(seconds) intervention["ended_at"] = now() intervention["actual_elapsed_ms"] = (time.monotonic_ns() - start) / 1_000_000 intervention["usage"].update(input_tokens=0, output_tokens=0, total_tokens=0) else: request.update(phase="intervention", form_id=None, state=state) response, state, start, end, elapsed = invoke(command, request, timeout, child_env) intervention.update(started_at=start, ended_at=end, actual_elapsed_ms=elapsed, usage=validate_usage(response.get("usage"))) if response.get("status") != "ok": raise ValueError("intervention_failed") if condition == "pit_stop": intervention["evidence"] = supplied_evidence(response.get("evidence")) request.update(phase="after", form_id=after_form, state=state) run["after"], _ = measure(command, request, timeout, child_env) if run["after"]["status"] != "ok": run["error"] = "after_failed" except subprocess.TimeoutExpired: run["error"] = "runner_timeout" except (ValueError, OSError, KeyError, TypeError): # Never copy stdout/stderr, task content, local state paths, or credentials into reports. run["error"] = "runner_or_evidence_failed" return report def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--plan", required=True, type=Path) parser.add_argument("--runner", required=True, help="Operator-selected executable; shell parsing is never used.") parser.add_argument("--runner-arg", action="append", default=[]) parser.add_argument("--output", required=True, type=Path) parser.add_argument("--operator-authorized", action="store_true", required=True) parser.add_argument("--synthetic", action="store_true", help="Only fake runners/fixtures; always labeled synthetic.") parser.add_argument("--timeout-seconds", type=int, default=120) args = parser.parse_args() if not 1 <= args.timeout_seconds <= 3600: parser.error("timeout must be 1–3600 seconds") if args.output.exists(): parser.error("output already exists; select a new report filename") plan = json.loads(args.plan.read_text()) report = execute(plan, [args.runner, *args.runner_arg], operator_authorized=args.operator_authorized, synthetic=args.synthetic, timeout=args.timeout_seconds) args.output.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n") print(f"Wrote {report['report_kind']} descriptive report; supplied evidence remains unverified. Validate with scorer.mjs.") if __name__ == "__main__": main()