Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -187,8 +187,18 @@ run_dir = run_experiment(
The standard executor retains timeout, cancellation, logs, result paths and child
cleanup. A custom `execute(command, cwd, trial_dir, deadline, metric_key)` must
preserve that trial record contract. Prefer composing `execute_trial` over
reimplementing process lifetime. Metric readers must preserve the finite,
positive metric contract before returning a float.
reimplementing process lifetime. Metric readers convert a result document into a
float; `execute_trial` then applies the one finite, positive metric contract
(`model.protocols.metrics.validate_metric`) to whatever the reader returns, so a
reader that yields `0.0`, a `bool` or a non-finite value records a metric error
rather than entering the evaluation.

Run state is written by the termination path, never assumed. `status.json`
reports `completed` only after the final artifacts were written;
a run stopped by a command state keeps `timeout`, `failed` or `cancelled`; an
exception outside the adapter error contract writes `crashed` and re-raises; and a
failure while writing the terminal artifacts writes `termination_failed` with the
original measurement error beside it.

A new source adapter implements `snapshot(repo, excluded_run_dir)` and returns
`available` plus a stable `snapshot_hash` and explanatory metadata. It must
Expand Down
12 changes: 10 additions & 2 deletions src/inferrsi/harness/evaluation.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,11 @@


def _summary(trials: list[dict]) -> dict | None:
values = [trial["metric"] for trial in trials if trial.get("metric") is not None]
# Only a completed trial recorded a measurement. A timed-out or failed trial
# may still have written a result file, but its metric is not a datapoint;
# `evaluate` reports the exclusion in `reasons` instead of dropping it silently.
measured = [trial for trial in trials if trial.get("status") == "completed"]
values = [trial["metric"] for trial in measured if trial.get("metric") is not None]
if not values:
return None
ordered = sorted(values)
Expand All @@ -25,6 +29,7 @@ def _summary(trials: list[dict]) -> dict | None:
"min": min(values),
"max": max(values),
"range_pct_of_median": spread if math.isfinite(spread) else None,
"excluded_trials": len(trials) - len(values),
}


Expand All @@ -43,7 +48,10 @@ def evaluate(
)
for index, trial in enumerate(trials, 1):
if trial.get("status") != "completed":
incomplete.append(f"{label} trial {index}: {trial.get('status')}")
incomplete.append(
f"{label} trial {index}: {trial.get('status')} trial excluded from "
"the recorded measurements"
)
if trial.get("metric") is None:
incomplete.append(
f"{label} trial {index}: {trial.get('metric_error') or 'metric missing'}"
Expand Down
170 changes: 111 additions & 59 deletions src/inferrsi/harness/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,8 @@ def run_experiment(
_write_json(run_dir / "environment.json", environment)
baseline, candidate, validation = [], [], None
before, after, source_checks = {}, {}, []
state, error = "completed", None
# No state is claimed until the work that would establish it has finished.
state, error = None, None

def event(phase: str, **details: object) -> None:
record = {"time": datetime.now(timezone.utc).isoformat(), "phase": phase, **details}
Expand All @@ -45,6 +46,17 @@ def event(phase: str, **details: object) -> None:
run_dir / "status.json", {"state": "running", "phase": phase, "run_dir": str(run_dir)}
)

def record_status(phase: str, **fields: object) -> None:
_write_json(
run_dir / "status.json",
{
"run_dir": str(run_dir),
"elapsed_seconds": time.monotonic() - started,
"phase": phase,
**fields,
},
)

repos = {
"baseline": Path(spec.repo).resolve(),
"candidate": Path(spec.candidate_repo or spec.repo).resolve(),
Expand All @@ -66,6 +78,82 @@ def execute_checked(
_write_json(trial_dir / "source.json", check)
return record

def terminate() -> None:
"""Write the terminal artifacts; a failure here is reported, not hidden."""
nonlocal after
try:
after = {label: snapshot(repo, run_dir) for label, repo in repos.items()}
_write_json(run_dir / "repositories.final.json", after)
_write_json(run_dir / "source_checks.json", source_checks)
identities = list(before.values()) + list(after.values())
identities.extend(
identity
for check in source_checks
for identity in (check["before"], check["after"])
)
known = bool(before) and all(item.get("available") for item in identities)
unchanged = (
known
and all(
before[label]["snapshot_hash"] == after[label]["snapshot_hash"]
for label in repos
)
and all(
check["before"]["snapshot_hash"]
== check["after"]["snapshot_hash"]
== before[check["side"]]["snapshot_hash"]
for check in source_checks
)
)
integrity = {
"verified": bool(unchanged),
"reason": "Git source snapshots unchanged"
if unchanged
else "Git source snapshot unavailable"
if not known
else "source changed during experiment",
"command_checks": len(source_checks),
}
result = evaluate(spec, baseline, candidate, validation, integrity)
if error:
result["reasons"].append(error)
result["verdict"] = "INCONCLUSIVE"
_write_json(run_dir / "evaluation.json", result)
event("complete", state=state, verdict=result["verdict"])
record_status(
"complete", state=state, verdict=result["verdict"], error=error
)
write_report(run_dir)
except KeyboardInterrupt as exc:
# Interruption during termination is outside the caller's own
# KeyboardInterrupt handler, so the state must be persisted here:
# otherwise it stays `running` (interrupted during the final snapshot)
# or a premature `completed` (interrupted while writing the report).
record_status(
"termination",
state="termination_failed",
verdict=None,
error=error,
requested_state=state,
termination_error=f"KeyboardInterrupt: {exc or 'interrupted during termination'}",
)
raise
except Exception as exc:

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Handle interruption inside termination as well. except Exception does not catch KeyboardInterrupt, and the final terminate() call (line 196) is outside the outer try/KeyboardInterrupt handler (lines 184–185). Ctrl-C during final snapshot can leave the persisted run running; Ctrl-C in write_report can leave it marked completed without the report. This is #179's existing terminal-artifact contract and the interruption handling requested in the prior finding. Explicitly persist cancelled/termination-failed state and preserve any original measurement error for termination KeyboardInterrupt, then re-raise; add regressions for interruption in final snapshot and report writing.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in d5fd02ca.

terminate() now has an explicit except KeyboardInterrupt before except Exception. It records state: termination_failed, phase: termination, the requested_state, the original measurement error, and termination_error: KeyboardInterrupt: ..., then re-raises.

Two regressions, both failing on 01bd59c4:

  • test_interruption_while_writing_the_report_is_recorded_as_termination — write_report raises KeyboardInterrupt after completed was written. Before: status stayed completed. After: termination_failed with requested_state: completed.
  • test_interruption_during_final_snapshot_is_recorded_as_termination — the snapshot hook raises on its fourth call, inside terminate(). Before: status stayed running. After: termination_failed, and no evaluation.json is left behind.

Observed before/after on the parent commit: AssertionError: 'running' != 'termination_failed' and AssertionError: 'completed' != 'termination_failed', both now passing.

The pre-existing KeyboardInterrupt handling for the measurement loop (which records cancelled and still writes the report) is unchanged, since that path reaches terminate() normally.

# The run did not reach a finished state. Every termination failure is
# recorded, not only I/O-shaped ones: the hook and report paths are
# caller-supplied, and an escaped RuntimeError would otherwise leave
# `running` (or a premature `completed`) as the last persisted state.
record_status(
"termination",
state="termination_failed",
verdict=None,
error=error,
requested_state=state,
termination_error=f"{type(exc).__name__}: {exc}",
)
raise

stopped = None
try:
event("source_snapshot")
before = {label: snapshot(repo, run_dir) for label, repo in repos.items()}
Expand All @@ -82,78 +170,42 @@ def execute_checked(
)
(baseline if label == "baseline" else candidate).append(record)
if record["status"] == "cancelled":
state = "cancelled"
raise KeyboardInterrupt
if record["status"] != "completed":
state = "timeout" if record["status"] == "timeout" else "failed"
stopped = record["status"]
break
if state != "completed":
if stopped is not None:
break
if spec.validation_command is not None and state == "completed":
if spec.validation_command is not None and stopped is None:
event("quality_validation")
validation = execute_checked(
spec.validation_command, "candidate", run_dir / "validation", None
)
if validation["status"] != "completed":
state = (
"cancelled"
if validation["status"] == "cancelled"
else "timeout"
if validation["status"] == "timeout"
else "failed"
)
stopped = validation["status"]
elif spec.validation_command is not None:
validation = {"status": "skipped", "reason": "measurement did not complete"}
if stopped is None:
state = "completed"
else:
state = (
"cancelled"
if stopped == "cancelled"
else "timeout"
if stopped == "timeout"
else "failed"
)
except KeyboardInterrupt:
state, error = "cancelled", "interrupted by user"
except (OSError, ValueError, subprocess.SubprocessError) as exc:
state, error = "failed", str(exc)
finally:
after = {label: snapshot(repo, run_dir) for label, repo in repos.items()}
_write_json(run_dir / "repositories.final.json", after)
_write_json(run_dir / "source_checks.json", source_checks)
identities = list(before.values()) + list(after.values())
identities.extend(
identity for check in source_checks for identity in (check["before"], check["after"])
)
known = bool(before) and all(item.get("available") for item in identities)
unchanged = (
known
and all(
before[label]["snapshot_hash"] == after[label]["snapshot_hash"] for label in repos
)
and all(
check["before"]["snapshot_hash"]
== check["after"]["snapshot_hash"]
== before[check["side"]]["snapshot_hash"]
for check in source_checks
)
)
integrity = {
"verified": bool(unchanged),
"reason": "Git source snapshots unchanged"
if unchanged
else "Git source snapshot unavailable"
if not known
else "source changed during experiment",
"command_checks": len(source_checks),
}
result = evaluate(spec, baseline, candidate, validation, integrity)
if error:
result["reasons"].append(error)
result["verdict"] = "INCONCLUSIVE"
_write_json(run_dir / "evaluation.json", result)
event("complete", state=state, verdict=result["verdict"])
_write_json(
run_dir / "status.json",
{
"state": state,
"phase": "complete",
"verdict": result["verdict"],
"run_dir": str(run_dir),
"elapsed_seconds": time.monotonic() - started,
"error": error,
},
except BaseException as exc:
# A hook failure outside the documented error contract still left a run
# directory. Record the crash truthfully, then let the caller see it.
event("crashed")
record_status(
"crashed", state="crashed", verdict=None, error=f"{type(exc).__name__}: {exc}"
)
write_report(run_dir)
raise
terminate()
return run_dir
9 changes: 5 additions & 4 deletions src/inferrsi/harness/scenarios/external.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

from ...infra.artifacts import atomic_write_json as _write_json
from ...infra.executors.local import execute_command
from ..evaluation import read_metric
from ...model.protocols.metrics import read_metric, validate_metric


def execute_trial(
Expand All @@ -22,9 +22,10 @@ def execute_trial(
result_path = Path(record["result_path"])
if metric_key is not None:
try:
record["metric"] = metric_reader(
json.loads(result_path.read_text(encoding="utf-8")), metric_key
)
document = json.loads(result_path.read_text(encoding="utf-8"))
# A reader is a conversion, not an exemption: the common metric
# contract applies here, before the value enters the trial record.
record["metric"] = validate_metric(metric_key, metric_reader(document, metric_key))
except (OSError, ValueError, TypeError, OverflowError) as exc:
record["metric_error"] = str(exc)
_write_json(trial_dir / "execution.json", record)
Expand Down
25 changes: 23 additions & 2 deletions src/inferrsi/infra/artifacts.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,16 @@

import hashlib
import json
import os
import time
import uuid
from pathlib import Path

# Windows refuses os.replace while another handle holds the destination open.
# A bounded retry covers the observed contention without hiding a real failure.
REPLACE_ATTEMPTS = 8
REPLACE_DELAY_SECONDS = 0.05


def digest(value):
return hashlib.sha256(
Expand All @@ -29,8 +37,21 @@ def write_json(path, value):


def atomic_write_json(path: Path, value: object) -> None:
temporary = path.with_suffix(path.suffix + ".tmp")
path = Path(path)
# A per-writer temporary name keeps two writers from sharing one file.
temporary = path.with_name(f"{path.name}.{os.getpid()}.{uuid.uuid4().hex[:8]}.tmp")
temporary.write_text(
json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8"
)
temporary.replace(path)
try:
for attempt in range(REPLACE_ATTEMPTS):
try:
temporary.replace(path)
return
except PermissionError:
if attempt + 1 == REPLACE_ATTEMPTS:
raise
time.sleep(REPLACE_DELAY_SECONDS)
finally:
if temporary.exists():
temporary.unlink()
15 changes: 10 additions & 5 deletions src/inferrsi/model/protocols/metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,19 @@
import math


def validate_metric(key: str, value) -> float:
"""Apply the one numeric contract, whatever produced the value."""
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"metric {key} must be a number")
if not math.isfinite(value) or value <= 0:
raise ValueError(f"metric {key} must be finite and positive")
return float(value)


def read_metric(result: dict, key: str) -> float:
value = result
for part in key.split("."):
if not isinstance(value, dict) or part not in value:
raise ValueError(f"missing metric: {key}")
value = value[part]
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"metric {key} must be a number")
if not math.isfinite(value) or value <= 0:
raise ValueError(f"metric {key} must be finite and positive")
return float(value)
return validate_metric(key, value)
29 changes: 29 additions & 0 deletions tests/test_goal_checkpoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,33 @@ def test_distinct_damaged_inputs_keep_distinct_checkpoint_identities(
name: sha256(compare.ROOT / name) for name in compare.CONTROLLERS}


def test_accepted_criteria_cannot_recommend_acceptance_before_any_trial_ran(
tmp_path, comparison_templates,
):
# Every criterion can be explicitly accepted while the comparison is still
# only prepared. Acceptance of criteria is not execution of the comparison.
path, goal, run = setup_goal(tmp_path, comparison_templates[0])
# A never-executed comparison has no results to cite; the frozen plan is all
# the evidence that exists, and the owner can still accept the criteria.
criterion = goal["criteria"][0]
criterion["evidence"] = [pin(run / "plan/freeze.json", path.parent)]
record = path.parent / "acceptance.json"
write_json(record, {"goal_id": goal["id"], "criterion_id": criterion["id"],
"criterion_sha256": checkpoint.criterion_sha256(goal, criterion),
"accepted_by": goal["acceptance_owner"], "decision": "accepted"})
criterion["acceptance"] = pin(record, path.parent)
write_json(path, goal)
result = checkpoint.build_checkpoint(path, run)
assert result["comparison"]["status"]["execution_state"] == "not_completed"
assert not result["remaining"] and not result["problems"]
assert result["recommendation"]["action"] != "report_accepted"
assert result["recommendation"]["action"] == "request_authorization"
authorize(path, goal)
result = checkpoint.build_checkpoint(path, run)
assert result["recommendation"]["action"] not in ("report_accepted", "wait_acceptance")
assert result["recommendation"]["action"] == "propose_step"


def test_interruption_never_becomes_a_fresh_or_accepted_attempt(tmp_path, comparison_templates):
path, goal, run = setup_goal(tmp_path, comparison_templates[0])
authorize(path, goal)
Expand Down Expand Up @@ -254,6 +281,8 @@ def test_real_client_failures_preserve_evidence_and_choose_the_actual_next_step(
before = files(run)
result = checkpoint.build_checkpoint(path, run)
assert not result["problems"]
# One case is two trials, so an unknown usage on the last trial starts nothing
# further; the schedule completed and the next step is owner acceptance.
expected = "prepare_new_plan" if mode in ("nonzero", "timeout") else "wait_acceptance"
assert result["recommendation"]["action"] == expected
assert result["comparison"]["counts"] == report["counts"]
Expand Down
Loading
Loading