diff --git a/agentlab/cli/task.py b/agentlab/cli/task.py index 6763616..4dc48bf 100644 --- a/agentlab/cli/task.py +++ b/agentlab/cli/task.py @@ -337,6 +337,7 @@ def _print_smoke_test_result(evaluation: object) -> None: def _print_failed_reference_checks(verification: ReferenceVerification) -> None: checks = verification.setup_checks + verification.baseline_checks checks += [verification.artifact_check] + verification.target_checks + checks += verification.hidden_verifier.checks for check in checks: if check.passed: continue diff --git a/agentlab/evidence/results.py b/agentlab/evidence/results.py index f8fbc73..1bb6dee 100644 --- a/agentlab/evidence/results.py +++ b/agentlab/evidence/results.py @@ -107,6 +107,11 @@ def to_result_dict(run: Any) -> Dict[str, Any]: "diff_path": str(run.agent_run.diff_path), "run_dir": str(run.run_dir), } + hidden_verifier = _hidden_verifier_to_dict( + getattr(run, "hidden_verifier", None) + ) + if hidden_verifier is not None: + result["hidden_verifier"] = hidden_verifier if setup_created_untracked_changed_paths: result["setup_created_untracked_changed_paths"] = ( setup_created_untracked_changed_paths @@ -202,6 +207,11 @@ def reference_verification_to_result_dict(verification: Any) -> Dict[str, Any]: "diff_path": _display_path(verification.diff_path, output_dir), "run_dir": _display_path(output_dir, output_dir), } + hidden_verifier = _hidden_verifier_to_dict( + getattr(verification, "hidden_verifier", None) + ) + if hidden_verifier is not None: + result["hidden_verifier"] = hidden_verifier if setup_created_untracked_changed_paths: result["setup_created_untracked_changed_paths"] = ( setup_created_untracked_changed_paths @@ -243,6 +253,20 @@ def _check_to_dict(check: CheckResult) -> Dict[str, Any]: } +def _hidden_verifier_to_dict(hidden_verifier: Any) -> Dict[str, Any] | None: + if hidden_verifier is None or not getattr(hidden_verifier, "configured", False): + return None + return { + "configured": True, + "patch": getattr(hidden_verifier, "patch", None), + "checks": [ + _check_to_dict(check) + for check in getattr(hidden_verifier, "checks", []) + ], + "restore_notes": list(getattr(hidden_verifier, "restore_notes", [])), + } + + def _reference_artifact_to_dict(artifact: Any) -> Dict[str, Any] | None: if artifact is None: return None diff --git a/agentlab/evidence/review_proposals.py b/agentlab/evidence/review_proposals.py index f4a4d0f..64f4faf 100644 --- a/agentlab/evidence/review_proposals.py +++ b/agentlab/evidence/review_proposals.py @@ -405,6 +405,13 @@ def _default_evidence(context: ReviewProposalContext) -> tuple[str, ...]: passed = check.get("passed") if command: evidence.append(f"check: passed={passed}; command={command}") + for check in _hidden_verifier_checks(result)[:3]: + command = str(check.get("command") or "").strip() + passed = check.get("passed") + if command: + evidence.append( + f"hidden verifier: passed={passed}; command={command}" + ) if context.report_excerpt: evidence.append(f"report.md excerpt: {context.report_excerpt}") @@ -429,6 +436,14 @@ def _looks_like_setup_issue(result: OutcomeEvidence) -> bool: str(check.get("command") or ""), ] ) + for check in _hidden_verifier_checks(result): + text_parts.extend( + [ + str(check.get("stdout") or ""), + str(check.get("stderr") or ""), + str(check.get("command") or ""), + ] + ) text = "\n".join(text_parts).lower() return any( needle in text @@ -444,6 +459,16 @@ def _looks_like_setup_issue(result: OutcomeEvidence) -> bool: ) +def _hidden_verifier_checks(result: OutcomeEvidence) -> list[Mapping[str, Any]]: + hidden_verifier = result.raw.get("hidden_verifier") + if not isinstance(hidden_verifier, Mapping): + return [] + checks = hidden_verifier.get("checks") + if not isinstance(checks, list): + return [] + return [check for check in checks if isinstance(check, Mapping)] + + def _read_excerpt(path: Path | None, max_chars: int = 600) -> str: if path is None: return "" diff --git a/agentlab/execution/hidden_verifier.py b/agentlab/execution/hidden_verifier.py new file mode 100644 index 0000000..ba61ff5 --- /dev/null +++ b/agentlab/execution/hidden_verifier.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +import shutil +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import Mapping + +from agentlab.execution.commands import isolated_git_env +from agentlab.execution.commands import run_commands +from agentlab.execution.commands import run_git +from agentlab.execution.scoring import CheckResult +from agentlab.tasks import EvalTask + + +@dataclass(frozen=True) +class HiddenVerifierResult: + configured: bool = False + patch: str | None = None + checks: list[CheckResult] = field(default_factory=list) + restore_notes: list[str] = field(default_factory=list) + + +def run_hidden_verifier( + task: EvalTask, + workspace: Path, + env: Mapping[str, str], +) -> HiddenVerifierResult: + verifier = task.hidden_verifier + if verifier is None: + return HiddenVerifierResult() + if task.source_path is None: + return HiddenVerifierResult( + configured=True, + patch=verifier.patch, + checks=[ + CheckResult( + command=f"hidden verifier patch: {verifier.patch}", + returncode=1, + stderr="hidden verifier requires task.source_path", + ) + ], + ) + + patch_path = task.source_path.parent / verifier.patch + with _workspace_snapshot(workspace) as snapshot: + checks: list[CheckResult] = [] + restore_notes: list[str] = [] + try: + apply_check = _apply_hidden_patch(workspace, patch_path, verifier.patch) + checks.append(apply_check) + if apply_check.passed: + checks.extend(run_commands(verifier.commands, workspace, env=env)) + finally: + restore_notes.extend(snapshot.restore()) + if restore_notes: + checks.append( + CheckResult( + command="restore hidden verifier workspace", + returncode=1, + stderr="\n".join(restore_notes), + ) + ) + + return HiddenVerifierResult( + configured=True, + patch=verifier.patch, + checks=checks, + restore_notes=restore_notes, + ) + + +def _apply_hidden_patch( + workspace: Path, + patch_path: Path, + display_path: str, +) -> CheckResult: + completed = run_git( + ["apply", str(patch_path.resolve())], + cwd=workspace, + env=isolated_git_env(), + ) + return CheckResult( + command=f"git apply hidden verifier patch: {display_path}", + returncode=completed.returncode, + stdout=completed.stdout, + stderr=completed.stderr, + ) + + +class _WorkspaceSnapshot: + def __init__(self, workspace: Path, snapshot: Path): + self._workspace = workspace + self._snapshot = snapshot + + def restore(self) -> list[str]: + notes: list[str] = [] + try: + _replace_worktree_contents(self._snapshot, self._workspace) + except OSError as exc: + notes.append(f"failed to restore hidden verifier worktree: {exc}") + return notes + + +class _workspace_snapshot: + def __init__(self, workspace: Path): + self._workspace = workspace + self._temp: tempfile.TemporaryDirectory[str] | None = None + self._snapshot: _WorkspaceSnapshot | None = None + + def __enter__(self) -> _WorkspaceSnapshot: + self._temp = tempfile.TemporaryDirectory(prefix="agentlab-hidden-verifier-") + root = Path(self._temp.name) + snapshot_path = root / "worktree" + _copy_worktree_contents(self._workspace, snapshot_path) + self._snapshot = _WorkspaceSnapshot(self._workspace, snapshot_path) + return self._snapshot + + def __exit__(self, exc_type, exc, tb) -> None: + if self._temp is not None: + self._temp.cleanup() + + +def _copy_worktree_contents(source: Path, destination: Path) -> None: + destination.mkdir(parents=True, exist_ok=True) + for child in source.iterdir(): + target = destination / child.name + if child.is_symlink(): + shutil.copy2(child, target, follow_symlinks=False) + elif child.is_dir(): + shutil.copytree(child, target, symlinks=True) + else: + shutil.copy2(child, target) + + +def _replace_worktree_contents(snapshot: Path, workspace: Path) -> None: + for child in workspace.iterdir(): + if child.is_dir() and not child.is_symlink(): + shutil.rmtree(child) + else: + child.unlink() + _copy_worktree_contents(snapshot, workspace) diff --git a/agentlab/execution/phases.py b/agentlab/execution/phases.py index 4da0bd6..b64f440 100644 --- a/agentlab/execution/phases.py +++ b/agentlab/execution/phases.py @@ -7,6 +7,8 @@ from agentlab.execution.commands import run_commands from agentlab.execution.changed_paths import capture_change_baseline from agentlab.execution.changed_paths import capture_diff_details_preserving_index +from agentlab.execution.hidden_verifier import HiddenVerifierResult +from agentlab.execution.hidden_verifier import run_hidden_verifier from agentlab.tasks.environment import build_task_environment from agentlab.runtime.patches import count_patch_lines from agentlab.execution.scoring import CheckResult @@ -35,6 +37,9 @@ class TaskExecution: baseline_checks: list[CheckResult] = field(default_factory=list) action_checks: list[CheckResult] = field(default_factory=list) target_checks: list[CheckResult] = field(default_factory=list) + hidden_verifier: HiddenVerifierResult = field( + default_factory=HiddenVerifierResult + ) files_changed: list[str] = field(default_factory=list) lines_added: int = 0 lines_deleted: int = 0 @@ -84,7 +89,12 @@ def execute_task_phases( patch_stats = count_patch_lines( resolved_diff_path.read_text(encoding="utf-8") ) - target_checks = run_commands(task.test, prepared.path, env=task_env) + hidden_verifier = run_hidden_verifier(task, prepared.path, task_env) + target_checks = ( + [] + if hidden_verifier.restore_notes + else run_commands(task.test, prepared.path, env=task_env) + ) all_checks = ( setup_checks + baseline_checks @@ -96,6 +106,7 @@ def execute_task_phases( all_checks, captured_diff.files_changed, agent_error=action_result.agent_error, + hidden_checks=hidden_verifier.checks, ) return TaskExecution( task=task, @@ -105,6 +116,7 @@ def execute_task_phases( baseline_checks=baseline_checks, action_checks=action_result.checks, target_checks=target_checks, + hidden_verifier=hidden_verifier, files_changed=captured_diff.files_changed, lines_added=patch_stats.lines_added, lines_deleted=patch_stats.lines_deleted, diff --git a/agentlab/execution/runner.py b/agentlab/execution/runner.py index 0d97c19..016c923 100644 --- a/agentlab/execution/runner.py +++ b/agentlab/execution/runner.py @@ -2,13 +2,14 @@ import time import uuid -from dataclasses import dataclass, replace +from dataclasses import dataclass, field, replace from pathlib import Path from agentlab.agents.base import AgentAdapter from agentlab.agents.base import AgentRun from agentlab.reports.trial_markdown import render_markdown_report from agentlab.execution.scoring import Score +from agentlab.execution.hidden_verifier import HiddenVerifierResult from agentlab.execution.phases import TaskActionResult from agentlab.execution.phases import execute_task_phases from agentlab.tasks import EvalTask @@ -22,6 +23,9 @@ class EvaluationRun: run_dir: Path report_path: Path result_path: Path + hidden_verifier: HiddenVerifierResult = field( + default_factory=HiddenVerifierResult + ) workspace_history_policy: str = "unknown" workspace_base_ref: str = "unknown" @@ -73,6 +77,7 @@ def run_agent(workspace: Path, _task_env: object) -> TaskActionResult: run_dir=run_dir, report_path=report_path, result_path=result_path, + hidden_verifier=execution.hidden_verifier, workspace_history_policy=execution.workspace_history_policy, workspace_base_ref=execution.workspace_base_ref, ) diff --git a/agentlab/execution/scoring.py b/agentlab/execution/scoring.py index 19194b0..39ab274 100644 --- a/agentlab/execution/scoring.py +++ b/agentlab/execution/scoring.py @@ -35,17 +35,25 @@ def calculate_grader_outcome( checks: Iterable[CheckResult], files_changed: Sequence[str], agent_error: Optional[str] = None, + hidden_checks: Iterable[CheckResult] = (), ) -> Score: check_results = list(checks) + hidden_check_results = list(hidden_checks) notes = _outcome_notes(task, files_changed) - checks_passed = ( + public_checks_passed = ( all(check.passed for check in check_results) if task.success.tests_must_pass else True ) + hidden_checks_passed = all(check.passed for check in hidden_check_results) return Score( - tests_passed=agent_error is None and checks_passed and not notes, + tests_passed=( + agent_error is None + and public_checks_passed + and hidden_checks_passed + and not notes + ), checks=check_results, notes=notes, ) diff --git a/agentlab/reports/operability_evidence.py b/agentlab/reports/operability_evidence.py index 8f42062..1dee15c 100644 --- a/agentlab/reports/operability_evidence.py +++ b/agentlab/reports/operability_evidence.py @@ -118,10 +118,31 @@ def _verifier_state_facts(results: list[OutcomeEvidence]) -> list[tuple[str, obj ("final_grader_status", _result_values(results, lambda result: result.status)), ("checks_array", _coverage(results, lambda result: bool(result.checks))), ("graders_array", _coverage(results, lambda result: bool(result.graders))), + ( + "hidden_verifier_configured", + _coverage(results, _hidden_verifier_configured), + ), + ( + "hidden_verifier_checks", + _coverage(results, _hidden_verifier_checks), + ), ("intermediate_verifier_movement", UNKNOWN), ] +def _hidden_verifier_configured(result: OutcomeEvidence) -> bool: + hidden_verifier = result.raw.get("hidden_verifier") + return isinstance(hidden_verifier, Mapping) + + +def _hidden_verifier_checks(result: OutcomeEvidence) -> bool: + hidden_verifier = result.raw.get("hidden_verifier") + if not isinstance(hidden_verifier, Mapping): + return False + checks = hidden_verifier.get("checks") + return isinstance(checks, list) and bool(checks) + + def _halt_reasons_facts(results: list[OutcomeEvidence]) -> list[tuple[str, object]]: return [ ( diff --git a/agentlab/reports/trial_markdown.py b/agentlab/reports/trial_markdown.py index 238bc59..26ce10e 100644 --- a/agentlab/reports/trial_markdown.py +++ b/agentlab/reports/trial_markdown.py @@ -72,16 +72,18 @@ def render_markdown_report(run: "EvaluationRun") -> str: lines.extend(["", "## Scope Oracle Metadata", ""]) lines.extend(scope_oracle_lines) - lines.extend(["", "## Code-Based Graders", ""]) + lines.extend(["", "## Public Graders", ""]) if not run.score.checks: - lines.append("No code-based graders were configured.") + lines.append("No public graders were configured.") else: lines.extend( _render_check(command_index, check) for command_index, check in enumerate(run.score.checks, 1) ) + lines.extend(_render_hidden_verifier_section(run)) + if run.score.notes: lines.extend(["", "## Grader Notes", ""]) lines.extend(f"- {note}" for note in run.score.notes) @@ -182,17 +184,19 @@ def render_reference_report(verification: "ReferenceVerification") -> str: lines.extend(["", "## Scope Oracle Metadata", ""]) lines.extend(scope_oracle_lines) - lines.extend(["", "## Code-Based Graders", ""]) + lines.extend(["", "## Public Graders", ""]) checks = verification.all_checks if not checks: - lines.append("No code-based graders were configured.") + lines.append("No public graders were configured.") else: lines.extend( _render_check(command_index, check) for command_index, check in enumerate(checks, 1) ) + lines.extend(_render_hidden_verifier_section(verification)) + if verification.notes: lines.extend(["", "## Grader Notes", ""]) lines.extend(f"- {note}" for note in verification.notes) @@ -218,6 +222,32 @@ def _render_check(index: int, check: CheckResult) -> str: return "\n".join(lines) +def _render_hidden_verifier_section(run: object) -> list[str]: + hidden_verifier = getattr(run, "hidden_verifier", None) + if hidden_verifier is None or not getattr(hidden_verifier, "configured", False): + return [] + lines = [ + "", + "## Hidden Verifier", + "", + f"- Patch: `{getattr(hidden_verifier, 'patch', None)}`", + ] + restore_notes = list(getattr(hidden_verifier, "restore_notes", [])) + if restore_notes: + lines.append("- Restore warnings:") + lines.extend(f" - {note}" for note in restore_notes) + checks = list(getattr(hidden_verifier, "checks", [])) + if not checks: + lines.extend(["", "No hidden verifier checks ran."]) + return lines + lines.append("") + lines.extend( + _render_check(command_index, check) + for command_index, check in enumerate(checks, 1) + ) + return lines + + def _render_scope_oracle_metadata(task: object) -> list[str]: success = getattr(task, "success", None) consent_style = getattr(task, "consent_style", None) diff --git a/agentlab/tasks/__init__.py b/agentlab/tasks/__init__.py index 0520396..1a246a0 100644 --- a/agentlab/tasks/__init__.py +++ b/agentlab/tasks/__init__.py @@ -2,6 +2,7 @@ EVAL_TYPES, TASK_BUNDLE_FILENAMES, EvalTask, + HiddenVerifier, ReferenceArtifact, SuccessCriteria, TaskBundle, @@ -27,6 +28,7 @@ "TASK_BUNDLE_FILENAMES", "BoundaryViolation", "EvalTask", + "HiddenVerifier", "ReferenceArtifact", "SuccessCriteria", "TaskBundle", diff --git a/agentlab/tasks/cards.py b/agentlab/tasks/cards.py index d4ee3b1..c996e3c 100644 --- a/agentlab/tasks/cards.py +++ b/agentlab/tasks/cards.py @@ -49,6 +49,7 @@ def render_task_card(bundle: TaskBundle) -> str: _environment_text(task), "", *_visible_validation_section(task), + *_hidden_verifier_section(task), "## Graders", "", "### Setup", @@ -99,6 +100,21 @@ def _visible_validation_section(task: EvalTask) -> list[str]: ] +def _hidden_verifier_section(task: EvalTask) -> list[str]: + verifier = task.hidden_verifier + if verifier is None: + return [] + command_count = len(verifier.commands) + command_word = "command" if command_count == 1 else "commands" + return [ + "## Hidden Verifier", + "", + f"- Patch: `{verifier.patch}`", + f"- Commands: `{command_count} {command_word} configured`", + "", + ] + + def _environment_text(task: EvalTask) -> str: lines = describe_task_environment(task) if not lines: diff --git a/agentlab/tasks/model.py b/agentlab/tasks/model.py index 98352c8..b54bf90 100644 --- a/agentlab/tasks/model.py +++ b/agentlab/tasks/model.py @@ -3,7 +3,7 @@ import glob import json from dataclasses import dataclass, field -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import Any, Dict, Iterable, List, Mapping, Optional import yaml # type: ignore @@ -34,6 +34,12 @@ class ReferenceArtifact: commit: Optional[str] = None +@dataclass(frozen=True) +class HiddenVerifier: + patch: str + commands: List[str] + + @dataclass(frozen=True) class EvalTask: id: str @@ -50,6 +56,7 @@ class EvalTask: baseline: List[str] = field(default_factory=list) test: List[str] = field(default_factory=list) visible_validation: List[str] = field(default_factory=list) + hidden_verifier: Optional[HiddenVerifier] = None environment_path: List[str] = field(default_factory=list) environment: Dict[str, str] = field(default_factory=dict) success: SuccessCriteria = field(default_factory=SuccessCriteria) @@ -132,6 +139,10 @@ def from_mapping( mapping.get("visible_validation", []), "visible_validation", ), + hidden_verifier=_hidden_verifier( + mapping.get("hidden_verifier"), + source_path, + ), environment_path=_environment_path( mapping.get("environment_path", []), "environment_path", @@ -273,6 +284,27 @@ def _string_list(value: Any, field_name: str) -> List[str]: return [str(item) for item in value] +def _nonempty_string_list(value: Any, field_name: str) -> List[str]: + entries = _string_list(value, field_name) + if not entries: + raise TaskLoadError(f"{field_name} must contain at least one value") + return entries + + +def _nonempty_strict_string_list(value: Any, field_name: str) -> List[str]: + if value is None: + raise TaskLoadError(f"{field_name} must contain at least one value") + if not isinstance(value, list): + raise TaskLoadError(f"{field_name} must be a list") + if not value: + raise TaskLoadError(f"{field_name} must contain at least one value") + if not all(isinstance(item, str) for item in value): + raise TaskLoadError(f"{field_name} must contain only strings") + if any(not item.strip() for item in value): + raise TaskLoadError(f"{field_name} must not contain blank strings") + return [item for item in value] + + def _environment_path(value: Any, field_name: str) -> List[str]: entries = _string_list(value, field_name) for entry in entries: @@ -366,6 +398,57 @@ def _reference_artifact( raise TaskLoadError("reference_artifact.type must be one of: patch, commit") +def _hidden_verifier( + value: Any, + source_path: Optional[Path], +) -> Optional[HiddenVerifier]: + if value is None: + return None + mapping = _mapping(value, "hidden_verifier") + allowed_keys = {"patch", "commands"} + unknown_keys = sorted(str(key) for key in mapping if key not in allowed_keys) + if unknown_keys: + raise TaskLoadError( + "hidden_verifier contains unknown field(s): " + + ", ".join(unknown_keys) + ) + patch = _hidden_patch_path(mapping.get("patch"), source_path) + commands = _nonempty_strict_string_list( + mapping.get("commands"), + "hidden_verifier.commands", + ) + return HiddenVerifier(patch=patch, commands=commands) + + +def _hidden_patch_path(value: Any, source_path: Optional[Path]) -> str: + patch = _required_string(value, "hidden_verifier.patch") + path = PurePosixPath(patch) + if path.is_absolute() or ".." in path.parts: + raise TaskLoadError( + "hidden_verifier.patch must be a relative path inside the bundle" + ) + if path.name in {"", "."}: + raise TaskLoadError("hidden_verifier.patch is invalid") + if path.suffix not in {".patch", ".diff"}: + raise TaskLoadError( + "hidden_verifier.patch must end with .patch or .diff" + ) + if source_path is not None: + bundle_dir = source_path.parent + candidate = bundle_dir / patch + if not candidate.is_file(): + raise TaskLoadError(f"hidden_verifier.patch does not exist: {patch}") + try: + candidate.resolve(strict=True).relative_to( + bundle_dir.resolve(strict=True) + ) + except (OSError, ValueError) as exc: + raise TaskLoadError( + "hidden_verifier.patch must stay inside the bundle" + ) from exc + return patch + + def _required_string(value: Any, field_name: str) -> str: parsed = _optional_string(value, field_name) if parsed is None or parsed == "": diff --git a/agentlab/tasks/reference.py b/agentlab/tasks/reference.py index 59e1725..7bf00d2 100644 --- a/agentlab/tasks/reference.py +++ b/agentlab/tasks/reference.py @@ -11,6 +11,7 @@ from agentlab.evidence.results import reference_verification_to_result_dict from agentlab.execution.scoring import CheckResult from agentlab.execution.scoring import Score +from agentlab.execution.hidden_verifier import HiddenVerifierResult from agentlab.execution.phases import TaskActionResult from agentlab.execution.phases import execute_task_phases from agentlab.tasks import EvalTask @@ -29,6 +30,9 @@ class ReferenceVerification: setup_checks: list[CheckResult] = field(default_factory=list) baseline_checks: list[CheckResult] = field(default_factory=list) target_checks: list[CheckResult] = field(default_factory=list) + hidden_verifier: HiddenVerifierResult = field( + default_factory=HiddenVerifierResult + ) files_changed: list[str] = field(default_factory=list) lines_added: int = 0 lines_deleted: int = 0 @@ -107,6 +111,7 @@ def reference_diff_path(_workspace: Path) -> Path: setup_checks=execution.setup_checks, baseline_checks=execution.baseline_checks, target_checks=execution.target_checks, + hidden_verifier=execution.hidden_verifier, files_changed=execution.files_changed, lines_added=execution.lines_added, lines_deleted=execution.lines_deleted, diff --git a/docs/design.md b/docs/design.md index 6a90c56..a219b15 100644 --- a/docs/design.md +++ b/docs/design.md @@ -144,6 +144,13 @@ commands in `test` remain grader-facing by default: they run after the agent phase and contribute to the deterministic grader outcome, but they are not automatically injected into agent prompts. +Task bundles can also configure one `hidden_verifier` overlay with a patch and +commands. Hidden verifier details stay out of agent prompts and generated task +cards show only summary metadata. After the agent phase, the runner captures the +model diff first, applies the hidden verifier patch temporarily, runs hidden +commands in the task environment, records those checks separately, and restores +the workspace to the model-produced state. + ## Task Environments Task setup should provision the dependencies needed by the deterministic graders @@ -197,9 +204,9 @@ configuration metadata. ## Graders And Outcomes The current evaluation harness uses deterministic code-based graders: setup -commands, baseline assertions, target assertions, and post-change assertions. -These are fast, cheap, reproducible, and appropriate for early coding-agent -evals. +commands, baseline assertions, public target assertions, and optional hidden +verifier assertions. These are fast, cheap, reproducible, and appropriate for +early coding-agent evals. Reports emphasize the outcome: the final patch, changed files, command results, and grader pass/fail status. Tool-call and transcript graders should be added diff --git a/tests/task_execution_helpers.py b/tests/task_execution_helpers.py index 976eeee..f9352b9 100644 --- a/tests/task_execution_helpers.py +++ b/tests/task_execution_helpers.py @@ -20,7 +20,7 @@ def _repo_with_file(self, root, contents, *, gitignore=None): commit = self._git(["rev-parse", "HEAD"], repo).stdout.strip() return repo, commit - def _git(self, args, cwd, *, input_text=None): + def _git(self, args, cwd, *, input_text=None, check=True): completed = subprocess.run( ["git"] + args, cwd=str(cwd), @@ -28,6 +28,6 @@ def _git(self, args, cwd, *, input_text=None): capture_output=True, input=input_text, ) - if completed.returncode != 0: + if check and completed.returncode != 0: self.fail(completed.stderr) return completed diff --git a/tests/test_agent_prompts.py b/tests/test_agent_prompts.py index b6cf6b0..74bec47 100644 --- a/tests/test_agent_prompts.py +++ b/tests/test_agent_prompts.py @@ -5,7 +5,7 @@ from agentlab.agents.manual import _print_manual_instructions from agentlab.agents.prompts import build_agent_prompt -from agentlab.tasks import EvalTask, SuccessCriteria +from agentlab.tasks import EvalTask, HiddenVerifier, SuccessCriteria class AgentPromptTest(unittest.TestCase): @@ -88,6 +88,34 @@ def test_scope_oracle_metadata_is_not_injected(self): self.assertNotIn("forbidden_paths", prompt) self.assertNotIn("allowed_paths", prompt) + def test_hidden_verifier_is_not_injected(self): + task = EvalTask( + id="hidden-task", + title="Hidden task", + repo="https://github.com/example/repo", + commit="abc123", + language="python", + prompt="Fix the behavior.", + visible_validation=["pytest tests/focused_check.py"], + hidden_verifier=HiddenVerifier( + patch="verifier.patch", + commands=["pytest tests/hidden_behavior.py"], + ), + ) + + prompt = build_agent_prompt(task) + output = StringIO() + with redirect_stdout(output): + _print_manual_instructions(task, Path("/tmp/workspace")) + printed = output.getvalue() + + self.assertIn("pytest tests/focused_check.py", prompt) + self.assertNotIn("verifier.patch", prompt) + self.assertNotIn("pytest tests/hidden_behavior.py", prompt) + self.assertIn("pytest tests/focused_check.py", printed) + self.assertNotIn("verifier.patch", printed) + self.assertNotIn("pytest tests/hidden_behavior.py", printed) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_operability_evidence.py b/tests/test_operability_evidence.py index e7a56a5..ecd0f14 100644 --- a/tests/test_operability_evidence.py +++ b/tests/test_operability_evidence.py @@ -65,6 +65,8 @@ def test_render_agent_harness_operability_table_uses_raw_result_fields(self): self.assertIn("normalized_or_derived_stop_reason", table) self.assertIn("first_class_halt_reason_taxonomy: `unknown`", table) self.assertIn("intermediate_verifier_movement: `unknown`", table) + self.assertIn("hidden_verifier_configured: `unknown`", table) + self.assertIn("hidden_verifier_checks: `unknown`", table) self.assertIn( "budget_operator_interruption_taxonomy: `unknown`", table, @@ -136,6 +138,39 @@ def test_render_agent_harness_operability_table_keeps_reasoning_effort_out_of_bu self.assertIn("turn_or_step_budget: `unknown`", table) self.assertNotIn('turn_or_step_budget: `{"reasoning_effort": "xhigh"}`', table) + def test_render_agent_harness_operability_table_reports_hidden_verifier_state(self): + result = _result( + trial_id="codex-hidden-verifier", + agent_name="codex", + model_name="gpt-test", + status="passed", + success=True, + config={"agent_harness": "codex"}, + input_tokens=10, + output_tokens=5, + cost_usd=None, + commands_run=[], + checks=[], + graders=[], + hidden_verifier={ + "patch": "verifier.patch", + "checks": [ + { + "command": "pytest tests/hidden_behavior.py", + "returncode": 0, + "passed": True, + } + ], + }, + ) + + table = "\n".join(render_agent_harness_operability_table([result])) + + self.assertIn("checks_array: `unknown`", table) + self.assertIn("graders_array: `unknown`", table) + self.assertIn("hidden_verifier_configured: `1/1`", table) + self.assertIn("hidden_verifier_checks: `1/1`", table) + def _result( *, @@ -148,10 +183,13 @@ def _result( input_tokens: int | None, output_tokens: int | None, cost_usd: float | None, + commands_run: list[object] | None = None, + checks: list[object] | None = None, + graders: list[object] | None = None, + hidden_verifier: dict[str, object] | None = None, ): run_dir = Path("/tmp") / trial_id - return normalize_outcome_evidence( - { + payload = { "trial_kind": "agent_trial", "trial_id": trial_id, "run_id": trial_id, @@ -170,14 +208,22 @@ def _result( "files_changed": ["app.py"], "lines_added": 3, "lines_deleted": 1, - "commands_run": ["pytest"], - "checks": [{"name": "pytest", "status": status}], - "graders": [{"name": "pytest"}], + "commands_run": ["pytest"] if commands_run is None else commands_run, + "checks": ( + [{"name": "pytest", "status": status}] + if checks is None + else checks + ), + "graders": [{"name": "pytest"}] if graders is None else graders, "report_path": str(run_dir / "report.md"), "transcript_path": str(run_dir / "transcript.md"), "diff_path": str(run_dir / "diff.patch"), "run_dir": str(run_dir), - }, + } + if hidden_verifier is not None: + payload["hidden_verifier"] = hidden_verifier + return normalize_outcome_evidence( + payload, run_dir=run_dir, ) diff --git a/tests/test_reference.py b/tests/test_reference.py index 9526925..73bcc40 100644 --- a/tests/test_reference.py +++ b/tests/test_reference.py @@ -69,6 +69,101 @@ def test_verifies_patch_reference_artifact(self): self.assertEqual(verification.workspace_history_policy, "base_only") assert_base_only_repository(self, verification.workspace) + def test_reference_verification_runs_hidden_verifier(self): + if shutil.which("git") is None: + self.skipTest("git is required for reference verification") + + with tempfile.TemporaryDirectory() as temp: + temp_path = Path(temp) + repo = temp_path / "repo" + init_repo(repo) + commit = commit_file(repo, "app.txt", "before\n", message="initial") + + (repo / "app.txt").write_text("after\n", encoding="utf-8") + patch = git(["diff"], repo).stdout + git(["checkout", "--", "app.txt"], repo) + + bundle = temp_path / "task" + bundle.mkdir() + (bundle / "reference.patch").write_text(patch, encoding="utf-8") + (bundle / "verifier.patch").write_text( + "\n".join( + [ + "diff --git a/hidden_check.py b/hidden_check.py", + "new file mode 100644", + "--- /dev/null", + "+++ b/hidden_check.py", + "@@ -0,0 +1,2 @@", + "+from pathlib import Path", + "+assert Path('app.txt').read_text() == 'after\\n'", + "", + ] + ), + encoding="utf-8", + ) + hidden_command = f"{sys.executable} hidden_check.py" + (bundle / "task.yaml").write_text( + textwrap.dedent( + f""" + id: reference-hidden-task + title: Reference hidden task + repo: {repo} + commit: {commit} + language: text + prompt: Change before to after. + reference_artifact: + type: patch + path: reference.patch + hidden_verifier: + patch: verifier.patch + commands: + - {hidden_command} + """ + ), + encoding="utf-8", + ) + + task = load_task(bundle) + verification = verify_reference( + task, + temp_path / "work", + write_artifacts=True, + ) + result = json.loads( + (bundle / "reference-result.json").read_text(encoding="utf-8") + ) + report = (bundle / "reference-report.md").read_text(encoding="utf-8") + + self.assertTrue(verification.success) + self.assertEqual(verification.files_changed, ["app.txt"]) + self.assertEqual( + [check.command for check in verification.hidden_verifier.checks], + [ + "git apply hidden verifier patch: verifier.patch", + hidden_command, + ], + ) + self.assertFalse((verification.workspace / "hidden_check.py").exists()) + self.assertEqual( + [check["command"] for check in result["hidden_verifier"]["checks"]], + [ + "git apply hidden verifier patch: verifier.patch", + hidden_command, + ], + ) + self.assertNotIn(hidden_command, result["commands_run"]) + self.assertNotIn( + hidden_command, + [check["command"] for check in result["checks"]], + ) + self.assertNotIn( + hidden_command, + [grader["assertion"] for grader in result["graders"]], + ) + self.assertIn("## Public Graders", report) + self.assertIn("## Hidden Verifier", report) + self.assertIn(f"Assertion `{hidden_command}`: passed", report) + def test_commit_reference_artifact_is_converted_to_patch(self): if shutil.which("git") is None: self.skipTest("git is required for reference verification") diff --git a/tests/test_runner.py b/tests/test_runner.py index 495ffd9..8aabc40 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -6,8 +6,9 @@ from pathlib import Path from agentlab.agents.manual import ManualAgentAdapter +from agentlab.agents.base import AgentRun from agentlab.execution.runner import _run_id, run_task -from agentlab.tasks import EvalTask, SuccessCriteria +from agentlab.tasks import EvalTask, HiddenVerifier, SuccessCriteria from tests.git_fixtures import commit_file from tests.git_fixtures import git from tests.git_fixtures import head @@ -154,6 +155,101 @@ def test_visible_validation_is_not_executed_or_serialized_as_grader(self): self.assertNotIn(visible_command, report) self.assertNotIn(visible_command, json.dumps(result)) + def test_hidden_verifier_is_serialized_separately_from_public_graders(self): + if shutil.which("git") is None: + self.skipTest("git is required for workspace preparation") + + with tempfile.TemporaryDirectory() as temp: + temp_path = Path(temp) + repo = temp_path / "repo" + init_repo(repo) + commit = commit_file(repo, "app.txt", "before\n", message="initial") + bundle = temp_path / "bundle" + bundle.mkdir() + task_file = bundle / "task.yaml" + task_file.write_text("id: hidden-task\n", encoding="utf-8") + (bundle / "verifier.patch").write_text( + "\n".join( + [ + "diff --git a/hidden_check.py b/hidden_check.py", + "new file mode 100644", + "--- /dev/null", + "+++ b/hidden_check.py", + "@@ -0,0 +1,2 @@", + "+from pathlib import Path", + "+assert Path('app.txt').read_text() == 'after\\n'", + "", + ] + ), + encoding="utf-8", + ) + public_command = f"{sys.executable} -c \"print('public ok')\"" + hidden_command = f"{sys.executable} hidden_check.py" + + task = EvalTask( + id="hidden-serialization-task", + title="Hidden serialization task", + repo=str(repo), + commit=commit, + language="text", + prompt="Change app.txt.", + test=[public_command], + hidden_verifier=HiddenVerifier( + patch="verifier.patch", + commands=[hidden_command], + ), + source_path=task_file, + ) + + class EditingAdapter: + name = "editing" + + def run(self, task, workspace, run_dir): + (workspace / "app.txt").write_text("after\n", encoding="utf-8") + transcript_path = run_dir / "transcript.txt" + transcript_path.write_text("edited\n", encoding="utf-8") + return AgentRun( + agent_name=self.name, + task_id=task.id, + transcript_path=transcript_path, + diff_path=run_dir / "diff.patch", + ) + + evaluation = run_task(task, EditingAdapter(), temp_path / "runs") + result = json.loads(evaluation.result_path.read_text(encoding="utf-8")) + report = evaluation.report_path.read_text(encoding="utf-8") + + self.assertTrue(evaluation.score.tests_passed) + self.assertEqual(evaluation.agent_run.commands_run, [public_command]) + self.assertEqual( + [check["command"] for check in result["checks"]], + [public_command], + ) + self.assertEqual( + [grader["assertion"] for grader in result["graders"]], + [public_command], + ) + self.assertEqual(result["hidden_verifier"]["patch"], "verifier.patch") + self.assertEqual( + [check["command"] for check in result["hidden_verifier"]["checks"]], + [ + "git apply hidden verifier patch: verifier.patch", + hidden_command, + ], + ) + self.assertIn("## Public Graders", report) + self.assertIn(public_command, report) + self.assertIn("## Hidden Verifier", report) + self.assertIn(hidden_command, report) + self.assertFalse( + ( + evaluation.run_dir + / "workspace" + / "hidden-serialization-task" + / "hidden_check.py" + ).exists() + ) + def test_max_files_changed_is_enforced(self): if shutil.which("git") is None: self.skipTest("git is required for workspace preparation") diff --git a/tests/test_task_cards.py b/tests/test_task_cards.py index 0895085..fc99eae 100644 --- a/tests/test_task_cards.py +++ b/tests/test_task_cards.py @@ -165,6 +165,22 @@ def test_renders_visible_validation_when_configured(self): self.assertIn("- `pytest tests/focused_check.py`", card) self.assertIn("### Target\n\n- `pytest tests/test_demo.py`", card) + def test_renders_hidden_verifier_summary_without_commands(self): + with tempfile.TemporaryDirectory() as temp: + bundle_dir = _write_task_bundle( + Path(temp) / "example-suite", + "demo-001", + hidden_verifier=True, + ) + bundle = load_task_bundle(bundle_dir) + + card = render_task_card(bundle) + + self.assertIn("## Hidden Verifier", card) + self.assertIn("- Patch: `verifier.patch`", card) + self.assertIn("- Commands: `1 command configured`", card) + self.assertNotIn("pytest tests/hidden_behavior.py", card) + def _write_task_bundle( suite_dir: Path, @@ -175,10 +191,16 @@ def _write_task_bundle( tags: list[str] | None = None, scope_oracle: bool = False, visible_validation: bool = False, + hidden_verifier: bool = False, ) -> Path: bundle_dir = suite_dir / task_id bundle_dir.mkdir(parents=True) (bundle_dir / "reference.patch").write_text("diff --git a/demo.py b/demo.py\n") + if hidden_verifier: + (bundle_dir / "verifier.patch").write_text( + "diff --git a/tests/hidden_behavior.py b/tests/hidden_behavior.py\n", + encoding="utf-8", + ) tag_lines = "\n".join(f" - {tag}" for tag in (tags or ["bugfix", "python"])) (bundle_dir / "task.yaml").write_text( f"""\ @@ -200,6 +222,7 @@ def _write_task_bundle( baseline: - pytest {_visible_validation(visible_validation)} +{_hidden_verifier(hidden_verifier)} test: - pytest tests/test_demo.py environment_path: @@ -247,5 +270,16 @@ def _visible_validation(enabled: bool) -> str: return "visible_validation:\n - pytest tests/focused_check.py" +def _hidden_verifier(enabled: bool) -> str: + if not enabled: + return "" + return ( + "hidden_verifier:\n" + " patch: verifier.patch\n" + " commands:\n" + " - pytest tests/hidden_behavior.py" + ) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_task_execution.py b/tests/test_task_execution.py index caece9f..dc9a335 100644 --- a/tests/test_task_execution.py +++ b/tests/test_task_execution.py @@ -9,7 +9,7 @@ from agentlab.execution.scoring import CheckResult from agentlab.execution.phases import TaskActionResult from agentlab.execution.phases import execute_task_phases -from agentlab.tasks import EvalTask +from agentlab.tasks import EvalTask, HiddenVerifier, SuccessCriteria from tests.task_execution_helpers import TaskExecutionGitMixin @@ -76,6 +76,287 @@ def action(workspace, _task_env): execution.workspace_base_ref, ) + def test_hidden_verifier_runs_after_model_diff_and_restores_workspace(self): + if shutil.which("git") is None: + self.skipTest("git is required for task execution") + + with tempfile.TemporaryDirectory() as temp: + temp_path = Path(temp) + repo, commit = self._repo_with_file(temp_path, "before\n") + task_file = _hidden_task_file( + temp_path, + _new_file_patch( + "hidden_check.py", + [ + "import subprocess", + "from pathlib import Path", + "subprocess.run(['git', 'config', 'hidden.verifier', 'true'], check=True)", + "assert Path('app.txt').read_text() == 'after\\n'", + ], + ), + ) + hidden_command = f"{sys.executable} hidden_check.py" + task = EvalTask( + id="hidden-success-task", + title="Hidden success task", + repo=str(repo), + commit=commit, + language="text", + prompt="Change app.txt.", + hidden_verifier=HiddenVerifier( + patch="verifier.patch", + commands=[hidden_command], + ), + success=SuccessCriteria(tests_must_pass=False), + source_path=task_file, + ) + + def action(workspace, _task_env): + (workspace / "app.txt").write_text("after\n", encoding="utf-8") + return TaskActionResult() + + execution = execute_task_phases( + task, + temp_path / "workspace", + action, + temp_path / "diff.patch", + ) + + self.assertTrue(execution.score.tests_passed) + self.assertEqual(execution.files_changed, ["app.txt"]) + self.assertEqual(execution.lines_added, 1) + self.assertEqual(execution.lines_deleted, 1) + self.assertFalse((execution.workspace / "hidden_check.py").exists()) + config_check = self._git( + ["config", "--get", "hidden.verifier"], + execution.workspace, + check=False, + ) + self.assertNotEqual(config_check.returncode, 0) + self.assertTrue(execution.hidden_verifier.configured) + self.assertEqual( + [check.command for check in execution.hidden_verifier.checks], + [ + "git apply hidden verifier patch: verifier.patch", + hidden_command, + ], + ) + self.assertEqual(execution.all_checks, []) + + def test_hidden_verifier_restore_preserves_directory_symlinks(self): + if shutil.which("git") is None: + self.skipTest("git is required for task execution") + + with tempfile.TemporaryDirectory() as temp: + temp_path = Path(temp) + repo, commit = self._repo_with_file(temp_path, "before\n") + outside_dir = temp_path / "outside-target" + outside_dir.mkdir() + (outside_dir / "secret.txt").write_text("outside\n", encoding="utf-8") + task_file = _hidden_task_file( + temp_path, + _new_file_patch("hidden_check.py", ["print('hidden ok')"]), + ) + task = EvalTask( + id="hidden-symlink-task", + title="Hidden symlink task", + repo=str(repo), + commit=commit, + language="text", + prompt="Create a symlink.", + hidden_verifier=HiddenVerifier( + patch="verifier.patch", + commands=[f"{sys.executable} hidden_check.py"], + ), + source_path=task_file, + ) + + def action(workspace, _task_env): + os.symlink(outside_dir, workspace / "linked-dir") + return TaskActionResult() + + execution = execute_task_phases( + task, + temp_path / "workspace", + action, + temp_path / "diff.patch", + ) + + restored_link = execution.workspace / "linked-dir" + self.assertTrue(execution.score.tests_passed) + self.assertTrue(restored_link.is_symlink()) + self.assertEqual(os.readlink(restored_link), str(outside_dir)) + + def test_visible_test_tampering_cannot_substitute_for_hidden_verifier(self): + if shutil.which("git") is None: + self.skipTest("git is required for task execution") + + with tempfile.TemporaryDirectory() as temp: + temp_path = Path(temp) + repo = temp_path / "repo" + repo.mkdir() + self._git(["init"], repo) + self._git(["config", "user.email", "agentlab@example.com"], repo) + self._git(["config", "user.name", "Agent Lab"], repo) + (repo / "app.txt").write_text("broken\n", encoding="utf-8") + (repo / "visible_test.py").write_text( + "raise SystemExit('visible test still expects a fix')\n", + encoding="utf-8", + ) + self._git(["add", "."], repo) + self._git(["commit", "-m", "initial"], repo) + commit = self._git(["rev-parse", "HEAD"], repo).stdout.strip() + task_file = _hidden_task_file( + temp_path, + _new_file_patch( + "hidden_check.py", + [ + "from pathlib import Path", + "assert Path('app.txt').read_text() == 'fixed\\n'", + ], + ), + ) + visible_command = f"{sys.executable} visible_test.py" + hidden_command = f"{sys.executable} hidden_check.py" + task = EvalTask( + id="hidden-fail-task", + title="Hidden fail task", + repo=str(repo), + commit=commit, + language="python", + prompt="Fix app.txt.", + test=[visible_command], + hidden_verifier=HiddenVerifier( + patch="verifier.patch", + commands=[hidden_command], + ), + source_path=task_file, + ) + + def action(workspace, _task_env): + (workspace / "visible_test.py").write_text( + "print('visible test bypassed')\n", + encoding="utf-8", + ) + return TaskActionResult() + + execution = execute_task_phases( + task, + temp_path / "workspace", + action, + temp_path / "diff.patch", + ) + + self.assertFalse(execution.score.tests_passed) + self.assertTrue(execution.target_checks[0].passed) + self.assertFalse(execution.hidden_verifier.checks[1].passed) + self.assertEqual(execution.files_changed, ["visible_test.py"]) + self.assertFalse((execution.workspace / "hidden_check.py").exists()) + + def test_restore_failure_fails_hidden_verifier_and_skips_target_checks(self): + if shutil.which("git") is None: + self.skipTest("git is required for task execution") + + with tempfile.TemporaryDirectory() as temp: + temp_path = Path(temp) + repo, commit = self._repo_with_file(temp_path, "before\n") + task_file = _hidden_task_file( + temp_path, + _new_file_patch( + "hidden_check.py", + ["print('hidden ok')"], + ), + ) + target_command = f"{sys.executable} -c \"print('target should skip')\"" + task = EvalTask( + id="hidden-restore-fail-task", + title="Hidden restore fail task", + repo=str(repo), + commit=commit, + language="text", + prompt="Do nothing.", + test=[target_command], + hidden_verifier=HiddenVerifier( + patch="verifier.patch", + commands=[f"{sys.executable} hidden_check.py"], + ), + source_path=task_file, + ) + + with mock.patch( + "agentlab.execution.hidden_verifier._replace_worktree_contents", + side_effect=OSError("boom"), + ): + execution = execute_task_phases( + task, + temp_path / "workspace", + lambda _workspace, _task_env: TaskActionResult(), + temp_path / "diff.patch", + ) + + self.assertFalse(execution.score.tests_passed) + self.assertEqual(execution.target_checks, []) + self.assertEqual( + execution.hidden_verifier.checks[-1].command, + "restore hidden verifier workspace", + ) + self.assertFalse(execution.hidden_verifier.checks[-1].passed) + self.assertEqual( + execution.hidden_verifier.restore_notes, + ["failed to restore hidden verifier worktree: boom"], + ) + + def test_hidden_verifier_patch_apply_failure_fails_without_commands(self): + if shutil.which("git") is None: + self.skipTest("git is required for task execution") + + with tempfile.TemporaryDirectory() as temp: + temp_path = Path(temp) + repo, commit = self._repo_with_file(temp_path, "before\n") + task_file = _hidden_task_file( + temp_path, + "\n".join( + [ + "diff --git a/app.txt b/app.txt", + "--- a/app.txt", + "+++ b/app.txt", + "@@ -1 +1 @@", + "-missing", + "+hidden", + "", + ] + ), + ) + hidden_command = f"{sys.executable} -c \"raise SystemExit('no')\"" + task = EvalTask( + id="hidden-apply-fail-task", + title="Hidden apply fail task", + repo=str(repo), + commit=commit, + language="text", + prompt="Do nothing.", + hidden_verifier=HiddenVerifier( + patch="verifier.patch", + commands=[hidden_command], + ), + source_path=task_file, + ) + + execution = execute_task_phases( + task, + temp_path / "workspace", + lambda _workspace, _task_env: TaskActionResult(), + temp_path / "diff.patch", + ) + + self.assertFalse(execution.score.tests_passed) + self.assertEqual(len(execution.hidden_verifier.checks), 1) + self.assertFalse(execution.hidden_verifier.checks[0].passed) + self.assertEqual( + (execution.workspace / "app.txt").read_text(encoding="utf-8"), + "before\n", + ) + def test_agent_error_makes_grader_outcome_fail(self): if shutil.which("git") is None: self.skipTest("git is required for task execution") @@ -208,5 +489,29 @@ def action(_workspace, task_env): self.assertEqual(execution.files_changed, []) +def _hidden_task_file(root: Path, verifier_patch: str) -> Path: + bundle = root / "bundle" + bundle.mkdir(exist_ok=True) + task_file = bundle / "task.yaml" + task_file.write_text("id: hidden-task\n", encoding="utf-8") + (bundle / "verifier.patch").write_text(verifier_patch, encoding="utf-8") + return task_file + + +def _new_file_patch(path: str, lines: list[str]) -> str: + added = "\n".join(f"+{line}" for line in lines) + return "\n".join( + [ + f"diff --git a/{path} b/{path}", + "new file mode 100644", + "--- /dev/null", + f"+++ b/{path}", + f"@@ -0,0 +1,{len(lines)} @@", + added, + "", + ] + ) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_tasks.py b/tests/test_tasks.py index ce68059..f4b6f28 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -5,6 +5,7 @@ from agentlab.tasks import ( EvalTask, + HiddenVerifier, TaskLoadError, discover_task_bundles, discover_task_files, @@ -114,6 +115,131 @@ def test_loads_folded_command_list_items(self): ) self.assertEqual(task.test, ['python -c "print(\'still one command\')"']) + def test_loads_hidden_verifier_from_bundle(self): + with tempfile.TemporaryDirectory() as temp: + bundle = Path(temp) + (bundle / "verifier.patch").write_text( + "diff --git a/tests/hidden.py b/tests/hidden.py\n", + encoding="utf-8", + ) + task_file = bundle / "task.yaml" + task_file.write_text( + textwrap.dedent( + """ + id: hidden-task + title: Hidden verifier task + repo: https://github.com/example/demo + commit: abc123 + language: python + prompt: Fix it. + hidden_verifier: + patch: verifier.patch + commands: + - pytest tests/hidden.py + """ + ), + encoding="utf-8", + ) + + task = load_task(task_file) + + self.assertEqual( + task.hidden_verifier, + HiddenVerifier( + patch="verifier.patch", + commands=["pytest tests/hidden.py"], + ), + ) + + def test_rejects_invalid_hidden_verifier(self): + base = { + "id": "demo-001", + "title": "Demo task", + "repo": "https://github.com/example/demo", + "commit": "abc123", + "language": "python", + "prompt": "Fix it.", + } + invalid_values = [ + {"patch": "/tmp/verifier.patch", "commands": ["pytest"]}, + {"patch": "../verifier.patch", "commands": ["pytest"]}, + {"patch": "verifier.txt", "commands": ["pytest"]}, + {"patch": "verifier.patch", "commands": [123]}, + {"patch": "verifier.patch", "commands": [""]}, + {"patch": "verifier.patch", "commands": [" "]}, + {"patch": "verifier.patch", "commands": []}, + {"commands": ["pytest"]}, + {"patch": "verifier.patch"}, + {"patch": "verifier.patch", "commands": ["pytest"], "timeout": 30}, + {"patch": "verifier.patch", "commands": ["pytest"], "phases": []}, + ] + for hidden_verifier in invalid_values: + with self.subTest(hidden_verifier=hidden_verifier): + with self.assertRaises(TaskLoadError): + EvalTask.from_mapping( + { + **base, + "hidden_verifier": hidden_verifier, + } + ) + + def test_rejects_missing_hidden_verifier_patch_file_for_bundle(self): + with tempfile.TemporaryDirectory() as temp: + task_file = Path(temp) / "task.yaml" + task_file.write_text( + textwrap.dedent( + """ + id: hidden-task + title: Hidden verifier task + repo: https://github.com/example/demo + commit: abc123 + language: python + prompt: Fix it. + hidden_verifier: + patch: verifier.patch + commands: + - pytest tests/hidden.py + """ + ), + encoding="utf-8", + ) + + with self.assertRaisesRegex(TaskLoadError, "does not exist"): + load_task(task_file) + + def test_rejects_hidden_verifier_patch_symlink_escape(self): + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + outside_patch = root / "outside.patch" + outside_patch.write_text( + "diff --git a/app.py b/app.py\n", + encoding="utf-8", + ) + bundle = root / "bundle" + bundle.mkdir() + (bundle / "verifier.patch").symlink_to(outside_patch) + task_file = bundle / "task.yaml" + task_file.write_text( + textwrap.dedent( + """ + id: hidden-task + title: Hidden verifier task + repo: https://github.com/example/demo + commit: abc123 + language: python + prompt: Fix it. + hidden_verifier: + patch: verifier.patch + commands: + - pytest tests/hidden.py + """ + ), + encoding="utf-8", + ) + + with self.assertRaisesRegex(TaskLoadError, "inside the bundle"): + load_task(task_file) + def test_loads_boundary_metadata_and_consent_style(self): task = EvalTask.from_mapping( load_task_mapping(