fix: preserve evaluator input and audit contracts across platforms

This commit is contained in:
haelyra
2026-10-11 15:39:40 -04:00
5 changed files with 78 additions and 26 deletions
+6 -3
View File
@@ -37,9 +37,12 @@ cryptographic signature or proof of who ran the experiment.
grader to score `reference` 1 and the start workspace, the `naive` trap and every
`decoy-*` overlay below 1. A decoy has the right shape and the wrong content, so
a grader that checks only the form fails the check. Both runners count only valid
trials: an attempt that ended in a provider error or timeout stays in
`results.jsonl` with its own transcript, is reported apart, and is retried on
resume. A gated trial whose hook never ran stops the run and is not recorded.
trials: authentication-related provider errors stop the run before the attempt
is appended to `results.jsonl`. The transcript is retained, but there is no
attempt record. Other provider-error or timeout attempts stay in `results.jsonl`
with their own transcripts, are reported apart, and are retried on resume.
A gated trial whose hook never ran also stops the run without a `results.jsonl`
record.
```bash
node docker/gateguard-effectiveness/run-intent.js --check-graders
+5 -3
View File
@@ -104,10 +104,12 @@ transcript and decision metrics from the hook; the questions asked; how many of
the scenario's `evidence` patterns the agent stated in its own text (for
example the three importers); turns, tokens, cost and wall time as reported by
Claude; and provider errors or timeouts. The full transcript of each attempt is
kept in `transcripts/`. An attempt that ends in a provider error or timeout
stays in `results.jsonl` but is not a trial: summaries count valid trials only,
kept in `transcripts/`. An authentication-related provider error stops the run
before the attempt is appended to `results.jsonl`; its transcript is retained,
but there is no attempt record. Other provider-error or timeout attempts
stay in `results.jsonl` but are not trials: summaries count valid trials only,
report the failed attempts apart, and a resumed run retries their keys. A gated
trial whose hook never ran is not recorded at all; the run stops instead.
trial whose hook never ran also has no `results.jsonl` record; the run stops instead.
## Analysis
@@ -370,7 +370,8 @@ def _read_file_or_text(path: Optional[str], *, required: bool = False) -> Option
with open(path) as f:
return f.read()
except OSError as error:
if error.errno not in (errno.ENOENT, errno.ENAMETOOLONG):
windows_inline_task = error.errno == errno.EINVAL and sys.platform == "win32" and not required
if error.errno not in (errno.ENOENT, errno.ENAMETOOLONG) and not windows_inline_task:
raise
if required:
print(f"Error: output file '{path}' not found", file=sys.stderr)
@@ -370,7 +370,8 @@ def _read_file_or_text(path: Optional[str], *, required: bool = False) -> Option
with open(path) as f:
return f.read()
except OSError as error:
if error.errno not in (errno.ENOENT, errno.ENAMETOOLONG):
windows_inline_task = error.errno == errno.EINVAL and sys.platform == "win32" and not required
if error.errno not in (errno.ENOENT, errno.ENAMETOOLONG) and not windows_inline_task:
raise
if required:
print(f"Error: output file '{path}' not found", file=sys.stderr)
+63 -18
View File
@@ -8,22 +8,20 @@ import re
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace
from types import ModuleType, SimpleNamespace
from typing import NoReturn
import pytest
SCRIPT = Path(__file__).resolve().parents[1] / "skills/agent-self-evaluation/scripts/evaluate.py"
SPEC = importlib.util.spec_from_file_location("agent_self_evaluation", SCRIPT)
evaluator = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = evaluator
SPEC.loader.exec_module(evaluator)
PI_SCRIPT = SCRIPT.parents[3] / "pi/core/skills/agent-self-evaluation/scripts/evaluate.py"
def args(**overrides):
def args(**overrides: object) -> argparse.Namespace:
return argparse.Namespace(**{"interactive": False, "task": None, "output": None, **overrides})
def test_output_file_uses_piped_task(tmp_path, monkeypatch):
def test_output_file_uses_piped_task(evaluator: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
output = tmp_path / "response.txt"
output.write_text("All tests passed.", encoding="utf-8")
monkeypatch.setattr(sys, "stdin", io.StringIO("Add retry logic"))
@@ -31,7 +29,7 @@ def test_output_file_uses_piped_task(tmp_path, monkeypatch):
assert evaluator._read_input(args(output=str(output))) == ("Add retry logic", "All tests passed.")
def test_explicit_task_takes_precedence_over_pipe(tmp_path, monkeypatch):
def test_explicit_task_takes_precedence_over_pipe(evaluator: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
output = tmp_path / "response.txt"
output.write_text("Answer", encoding="utf-8")
stream = io.StringIO("Ignored pipe")
@@ -41,7 +39,7 @@ def test_explicit_task_takes_precedence_over_pipe(tmp_path, monkeypatch):
assert stream.tell() == 0
def test_output_file_without_task_does_not_read_terminal(tmp_path, monkeypatch):
def test_output_file_without_task_does_not_read_terminal(evaluator: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
output = tmp_path / "response.txt"
output.write_text("Answer", encoding="utf-8")
monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
@@ -49,36 +47,36 @@ def test_output_file_without_task_does_not_read_terminal(tmp_path, monkeypatch):
assert evaluator._read_input(args(output=str(output))) == (None, "Answer")
def test_no_output_file_keeps_stdin_as_output(monkeypatch):
def test_no_output_file_keeps_stdin_as_output(evaluator: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(sys, "stdin", io.StringIO("Piped answer"))
assert evaluator._read_input(args(task="Inline task")) == ("Inline task", "Piped answer")
def test_interactive_mode_keeps_prompt_and_stdin(monkeypatch):
def test_interactive_mode_keeps_prompt_and_stdin(evaluator: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("builtins.input", lambda _prompt: " Interactive task ")
monkeypatch.setattr(sys, "stdin", io.StringIO("Interactive answer"))
assert evaluator._read_input(args(interactive=True)) == ("Interactive task", "Interactive answer")
def test_task_can_still_be_read_from_file(tmp_path, monkeypatch):
def test_task_can_still_be_read_from_file(evaluator: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
task = tmp_path / "task.txt"
task.write_text("File task", encoding="utf-8")
monkeypatch.setattr(sys, "stdin", io.StringIO("Piped answer"))
assert evaluator._read_input(args(task=str(task))) == ("File task", "Piped answer")
def test_missing_output_file_still_fails(tmp_path, monkeypatch, capsys):
def test_missing_output_file_still_fails(evaluator: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
monkeypatch.setattr(sys, "stdin", io.StringIO("Task"))
with pytest.raises(SystemExit, match="1"):
evaluator._read_input(args(output=str(tmp_path / "missing.txt")))
assert "not found" in capsys.readouterr().err
def test_cli_uses_piped_task_for_conciseness_score(tmp_path):
def test_cli_uses_piped_task_for_conciseness_score(evaluator: ModuleType, tmp_path: Path) -> None:
output = tmp_path / "response.txt"
output.write_text("Verified with tests. " * 25, encoding="utf-8")
result = subprocess.run(
[sys.executable, "-X", "utf8", str(SCRIPT), "--output", str(output)],
[sys.executable, "-X", "utf8", str(evaluator.__file__), "--output", str(output)],
input="Fix bug",
capture_output=True,
encoding="utf-8",
@@ -88,7 +86,7 @@ def test_cli_uses_piped_task_for_conciseness_score(tmp_path):
assert re.search(r"Conciseness[^\n]*3/5", result.stdout)
def test_long_inline_task_remains_text(tmp_path, monkeypatch):
def test_long_inline_task_remains_text(evaluator: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
output = tmp_path / "response.txt"
output.write_text("Answer", encoding="utf-8")
task = "Long explicit task description. " * 60
@@ -97,11 +95,58 @@ def test_long_inline_task_remains_text(tmp_path, monkeypatch):
@pytest.mark.parametrize("error_code", [errno.EACCES, errno.EISDIR])
def test_task_read_does_not_hide_other_file_errors(monkeypatch, error_code):
def fail_open(_path):
def test_task_read_does_not_hide_other_file_errors(evaluator: ModuleType, monkeypatch: pytest.MonkeyPatch, error_code: int) -> None:
def fail_open(_path: str) -> NoReturn:
raise OSError(error_code, "Cannot read task")
monkeypatch.setattr("builtins.open", fail_open)
with pytest.raises(OSError) as captured:
evaluator._read_file_or_text("task-file")
assert captured.value.errno == error_code
@pytest.fixture(params=[SCRIPT, PI_SCRIPT], ids=["canonical", "pi"])
def evaluator(request: pytest.FixtureRequest) -> ModuleType:
spec = importlib.util.spec_from_file_location("task_input_evaluator", request.param)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
@pytest.mark.parametrize("platform", ["win32", "linux", "darwin"])
@pytest.mark.parametrize("required", [False, True], ids=["optional", "required"])
@pytest.mark.parametrize(
"error_code", [errno.EINVAL, errno.ENOENT, errno.ENAMETOOLONG, errno.EACCES, errno.EISDIR, errno.EIO]
)
def test_file_error_fallback_is_platform_and_mode_scoped(
evaluator: ModuleType,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
platform: str,
required: bool,
error_code: int,
) -> None:
task = "Long inline task. " * 60
error = OSError(error_code, "Cannot read file")
def fail_open(_path: str) -> NoReturn:
raise error
monkeypatch.setattr(evaluator.sys, "platform", platform)
monkeypatch.setattr("builtins.open", fail_open)
can_fallback = error_code in (errno.ENOENT, errno.ENAMETOOLONG) or (
error_code == errno.EINVAL and platform == "win32" and not required
)
if can_fallback and not required:
assert evaluator._read_file_or_text(task) == task
elif can_fallback:
with pytest.raises(SystemExit) as captured_exit:
evaluator._read_file_or_text(task, required=True)
assert captured_exit.value.code == 1
assert "not found" in capsys.readouterr().err
else:
with pytest.raises(OSError) as captured_error:
evaluator._read_file_or_text(task, required=required)
assert captured_error.value is error