diff --git a/docs/configuration.md b/docs/configuration.md index de30169..ae44b78 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -281,6 +281,33 @@ sia run --task gpqa --max_gen 5 --run_id 1 --focus weights --training_sandbox sa `SIA_META_AGENT_PROFILE` / `SIA_TARGET_AGENT_PROFILE` set the default profile names (overridden by the CLI flags). `SIA_MAX_GENERATIONS`, `SIA_MAX_TURNS`, and `SIA_SANDBOX_MODE` are also honored. +### Held-out feedback summary + +The feedback agent is shown a curated, **reference-answer-free** summary of the held-out +evaluation rather than the raw `results.json` (which can carry the grader's ground-truth +answers — a reward-hacking surface). The summary includes every scalar metric the grader emits, +at any nesting depth (e.g. top-level `accuracy`/`correct`/`total`, or a nested `summary` block); +every list is dropped, since per-item record arrays such as `results`/`details` are where +reference answers live. When the grader opts in by writing a generic `items[]` array, the summary +also adds a capped, reference-answer-free sample of the **failed** items. Two variables tune that +sample: + +| Variable | Default | Description | +| --- | --- | --- | +| `SIA_VERIFIER_PASS_STATUSES` | `CORRECT,PASS,correct` | Comma-separated set of `items[]` `status` values that count as **passing**. Any other status is treated as a failure when selecting which items to surface. Set this when your grader uses a different status vocabulary (e.g. `OK,GREEN`). | +| `SIA_FEEDBACK_FAILURE_SAMPLES` | `20` | Maximum number of failed held-out items included in the summary (diversified across `status` and `group`). Caps the feedback prompt size for large eval sets. | +| `SIA_PRIVATE_DIR_GUARD` | `1` (on) | OS-level backstop: while a meta/feedback agent runs, strip all filesystem permissions on the task's `/data/private` dir so a read fails at the filesystem layer (impl-agnostic, covers the agent impls that ignore the claude-only tool-call guard). The original mode is restored when the agent run ends, including on error. Set `0` to disable. | + +The `items[]` contract is opt-in and gold-free by design: a grader emits one dict per item with +any of the generic keys `id` / `status` / `group` / `category` / `input` / `output` / `detail` +(no reference answer). Graders that don't emit `items[]` simply get the scalar summary. The +held-out directory `/data/private` is additionally protected during meta/feedback agent +runs by two complementary mechanisms: a `claude`-impl tool-call guard that denies file/Bash access +resolving inside it, and an impl-agnostic OS-level guard (`SIA_PRIVATE_DIR_GUARD`, on by default) +that strips its filesystem permissions for the duration of the run. Both are defense-in-depth +behind the primary seal (the curated, reference-answer-free summary above); container-level +sandboxing (`--sandbox docker`) remains the strongest boundary against a determined agent. + ## Notes - The `claude` agent impl only accepts the Claude shortcut names (`haiku`, `sonnet`, `opus`) and an diff --git a/sia/agent_impls/base.py b/sia/agent_impls/base.py index e3ffb54..ab5e172 100644 --- a/sia/agent_impls/base.py +++ b/sia/agent_impls/base.py @@ -53,6 +53,7 @@ async def run_agent( agent_working_directory: str, agent_impl: str = "claude", provider: Provider | None = None, + protected_paths: list[str] | None = None, ) -> None: """Dispatch to the named agent impl. @@ -63,6 +64,19 @@ async def run_agent( agent_working_directory: Working directory for the agent. agent_impl: Which registered impl to use (e.g. "claude", "openhands", "pydantic-ai"). provider: Optional endpoint/credentials for the model (api_key_env, base_url). + protected_paths: Held-out dirs the agent must not access. Honored only by the + "claude" impl (which can enforce a PreToolUse deny hook); other impls ignore it. """ logger.info(f"Using {agent_impl} agent impl") - await get_agent_impl(agent_impl)(model_name, max_turns, prompt, agent_working_directory, provider=provider) + runner = get_agent_impl(agent_impl) + if agent_impl == "claude": + await runner( + model_name, + max_turns, + prompt, + agent_working_directory, + provider=provider, + protected_paths=protected_paths, + ) + else: + await runner(model_name, max_turns, prompt, agent_working_directory, provider=provider) diff --git a/sia/agent_impls/claude.py b/sia/agent_impls/claude.py index 233e718..eb527d1 100644 --- a/sia/agent_impls/claude.py +++ b/sia/agent_impls/claude.py @@ -2,23 +2,148 @@ from __future__ import annotations +import os +from collections.abc import Awaitable, Callable from datetime import datetime +from typing import Any, cast from sia.agent_impls.base import register from sia.logging_setup import get_logger logger = get_logger(__name__) +# Tools that take a path or path-glob input; their inputs are resolved against the +# protected dirs. Bash is handled separately (its command is substring-checked). +_FILE_TOOLS = frozenset({"Read", "Edit", "Write", "NotebookEdit", "Glob", "LS", "Grep"}) -async def run_agent_claude(model_name, max_turns, prompt, agent_working_directory, provider=None): +# Input keys that hold a single path or path-glob across the file tools above. +_PATH_FIELDS = ("file_path", "path", "notebook_path", "pattern") + + +def _looks_like_path(value: str) -> bool: + """Heuristic: a string that contains a separator or a parent-dir hop is path-like.""" + return ("/" in value) or value.startswith("..") or value.startswith("~") + + +def _resolves_inside(candidate: str, protected_real: list[str]) -> bool: + """True when ``candidate`` (a path, possibly with ``..`` or a glob) resolves inside any + protected dir. The leading glob magic is stripped so ``/**`` resolves to the + protected dir itself rather than a literal ``**`` child. + """ + cleaned = candidate.split("*", 1)[0].split("?", 1)[0] + if not cleaned: + cleaned = candidate + real = os.path.realpath(os.path.abspath(cleaned)) + return any(real == prot or real.startswith(prot + os.sep) for prot in protected_real) + + +def _accesses_protected_path(tool_name: str, tool_input: dict, protected_dirs: list[str]) -> bool: + """Return True when a tool call would touch a protected directory. + + File tools are checked by resolving their path-like input fields (handling ``..`` and + glob magic) against the canonical protected dirs. Bash is checked by substring match + of the command against both the canonical absolute protected path and a normalized + relative form (e.g. ``data/private``). Unknown tools and path-free inputs are allowed. + """ + if not protected_dirs: + return False + + protected_real = [os.path.realpath(os.path.abspath(d)) for d in protected_dirs] + + if tool_name == "Bash": + command = tool_input.get("command", "") + if not command: + return False + relative_forms = [os.path.join("data", os.path.basename(p)) for p in protected_dirs] + needles = protected_real + relative_forms + return any(needle in command for needle in needles) + + if tool_name in _FILE_TOOLS: + candidates: list[str] = [] + for field in _PATH_FIELDS: + value = tool_input.get(field) + if isinstance(value, str) and value: + candidates.append(value) + # Catch any other path-like string argument (defensive -- tool schemas vary). + for value in tool_input.values(): + if isinstance(value, str) and value and _looks_like_path(value) and value not in candidates: + candidates.append(value) + return any(_resolves_inside(c, protected_real) for c in candidates) + + return False + + +def _make_protected_path_hook( + protected_dirs: list[str], +) -> Callable[[dict, str | None, Any], Awaitable[dict]]: + """Build a PreToolUse hook callback that denies any tool call touching a protected dir. + + Returns the canonical SDK deny dict (``permissionDecision: "deny"``) when the call + would access a held-out dir, and ``{}`` (allow) otherwise. The reason names the + held-out dir generically so the agent self-corrects without learning its contents. + """ + held_out = ", ".join(protected_dirs) + + async def deny_cb(input: dict, tool_use_id: str | None, context: Any) -> dict: + tool_name = input.get("tool_name", "") + tool_input = input.get("tool_input", {}) or {} + if _accesses_protected_path(tool_name, tool_input, protected_dirs): + return { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": ( + f"Access to the held-out evaluation directory ({held_out}) is blocked. " + "It holds grader-only ground truth; improve the agent from the provided " + "feedback alone." + ), + } + } + return {} + + return deny_cb + + +def _build_claude_options(model_name, max_turns, agent_working_directory, protected_paths=None): + """Construct ClaudeAgentOptions, registering a protected-path deny hook when requested. + + Extracted so the option wiring (notably the PreToolUse hook registration) is unit + testable without invoking the Claude CLI. When ``protected_paths`` is None/empty no hook + is registered, so behavior is byte-identical to the pre-hook path. + """ + from claude_agent_sdk import ClaudeAgentOptions, HookMatcher + + hooks = None + if protected_paths: + deny_cb = _make_protected_path_hook(protected_paths) + # cast: the SDK's HookCallback is a broad TypedDict-union signature the closure + # conforms to at runtime but ty cannot match structurally. + hooks = {"PreToolUse": [HookMatcher(hooks=[cast(Any, deny_cb)])]} + + return ClaudeAgentOptions( + cwd=agent_working_directory, + allowed_tools=["Bash", "Read", "Write", "Edit", "Glob"], + permission_mode="bypassPermissions", + max_turns=max_turns, + model=model_name, + hooks=cast(Any, hooks), + ) + + +async def run_agent_claude(model_name, max_turns, prompt, agent_working_directory, provider=None, protected_paths=None): """Run agent using Claude Code SDK The ``provider`` argument is accepted for a uniform agent-impl signature but ignored: the Claude Code SDK authenticates against Anthropic natively (ANTHROPIC_API_KEY). + When ``protected_paths`` is non-empty, a PreToolUse hook denies every tool call that + would access one of those directories (the task's held-out ground truth). The hook + fires regardless of ``permission_mode="bypassPermissions"``, so normal authoring is + unimpeded and only protected-path access is blocked. + Note: Claude Code automatically saves trajectories to ~/.claude/projects/ """ - from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query + from claude_agent_sdk import ResultMessage, query logger.info(f"Starting agent execution with {model_name} model (max turns: {max_turns})") logger.debug("=" * 80) @@ -31,13 +156,7 @@ async def run_agent_claude(model_name, max_turns, prompt, agent_working_director try: async for message in query( prompt=prompt, - options=ClaudeAgentOptions( - cwd=agent_working_directory, - allowed_tools=["Bash", "Read", "Write", "Edit", "Glob"], - permission_mode="bypassPermissions", - max_turns=max_turns, - model=model_name, - ), + options=_build_claude_options(model_name, max_turns, agent_working_directory, protected_paths), ): logged_content = False diff --git a/sia/config.py b/sia/config.py index b2f2242..582af2f 100644 --- a/sia/config.py +++ b/sia/config.py @@ -8,6 +8,16 @@ from typing import ClassVar +def _to_str_tuple(value: str) -> tuple[str, ...]: + """Parse a comma-separated env-var string into a tuple of trimmed strings.""" + return tuple(part.strip() for part in value.split(",") if part.strip()) + + +def _to_bool(value: str) -> bool: + """Parse a truthy/falsy env-var string ("1"/"true"/"yes"/"on" -> True).""" + return value.strip().lower() in {"1", "true", "yes", "on"} + + @dataclass class Config: """Single source of truth for all SIA configuration defaults.""" @@ -51,6 +61,20 @@ class Config: MAX_CONTEXT_FILE_SIZE: int = 10_000_000 # 10 MB MAX_EXECUTION_LOG_SIZE: int = 50_000_000 # 50 MB + # Verifier->feedback contract. Pass-set: a grader item whose `status` is not in + # this set counts as a failure when building the gold-free eval summary. + VERIFIER_PASS_STATUSES: tuple[str, ...] = ("CORRECT", "PASS", "correct") + # Cap on the number of FAILED held-out items surfaced (reference-answer-free) in + # the feedback prompt's curated eval summary. Diversified across status and group. + FEEDBACK_FAILURE_SAMPLES: int = 20 + + # Held-out-integrity seal — impl-agnostic OS-level backstop. While a meta/feedback + # agent runs, strip all permissions on the task's data/private dir so a read fails at + # the filesystem layer, not just via the prompt notice + claude-impl PreToolUse hook. + # Applied only around the sequential meta/feedback agent calls — never during grading, + # which legitimately reads the dir. Set SIA_PRIVATE_DIR_GUARD=0 to disable. + PRIVATE_DIR_GUARD: bool = True + # Virtual environment packages. VENV_PACKAGES: ClassVar[list[str]] = [ "anthropic", @@ -85,6 +109,9 @@ def from_env(cls) -> Config: "SIA_AGENT_IMPL": ("DEFAULT_AGENT_IMPL", str), "SIA_MAX_TURNS": ("DEFAULT_MAX_TURNS", int), "SIA_SANDBOX_MODE": ("SANDBOX_MODE", str), + "SIA_VERIFIER_PASS_STATUSES": ("VERIFIER_PASS_STATUSES", _to_str_tuple), + "SIA_FEEDBACK_FAILURE_SAMPLES": ("FEEDBACK_FAILURE_SAMPLES", int), + "SIA_PRIVATE_DIR_GUARD": ("PRIVATE_DIR_GUARD", _to_bool), } for env_var, (attr, converter) in env_map.items(): val = os.environ.get(env_var) diff --git a/sia/context_manager.py b/sia/context_manager.py index 2c72ff8..324ca32 100644 --- a/sia/context_manager.py +++ b/sia/context_manager.py @@ -20,6 +20,95 @@ logger = get_logger(__name__) +# Default pass-set for the verifier->feedback contract. A grader item whose `status` +# is not in this set counts as a failure. Mirrors Config.VERIFIER_PASS_STATUSES; the +# config field is read by the orchestrator wiring, this local default keeps +# _extract_metrics self-contained when no config is threaded through. +DEFAULT_VERIFIER_PASS_STATUSES: tuple[str, ...] = ("CORRECT", "PASS", "correct") + +# Caps on the bounded items summary so a large items array cannot bloat context. +ITEMS_SUMMARY_MAX_GROUPS = 20 +ITEMS_SUMMARY_MAX_CATEGORIES = 20 +ITEMS_SUMMARY_MAX_WORST_IDS = 10 + + +def _is_failure(status: Any, pass_statuses: tuple[str, ...]) -> bool: + """A status counts as a failure when it is not in the pass-set.""" + return status not in pass_statuses + + +def summarize_items( + items: list[Any], + pass_statuses: tuple[str, ...] = DEFAULT_VERIFIER_PASS_STATUSES, +) -> dict[str, Any]: + """Compute a bounded summary of a grader's optional `items` array. + + Per the verifier->feedback contract, each item is an open dict in which three + keys are framework-recognized (all optional): `status`, `group`, `category`. + Returns counts that stay bounded regardless of array size: + + - `total`: number of items + - `failures`: count of items whose status is not in the pass-set + - `status_counts`: per-status item counts (all statuses) + - `group_failure_counts`: per-group failure counts (capped, top groups by failures) + - `category_counts`: per-failure-category counts (capped, top categories) + - `worst_ids`: ids of the first failing items (capped) + + Non-dict items are ignored. When an item has no `category`, a coarse category is + derived from its status so the digest still has a category dimension. + """ + status_counts: dict[str, int] = {} + group_failures: dict[str, int] = {} + category_counts: dict[str, int] = {} + worst_ids: list[str] = [] + failures = 0 + total = 0 + + for item in items: + if not isinstance(item, dict): + continue + total += 1 + status = item.get("status") + status_key = str(status) if status is not None else "UNKNOWN" + status_counts[status_key] = status_counts.get(status_key, 0) + 1 + + if not _is_failure(status, pass_statuses): + continue + + failures += 1 + + group = item.get("group") + if group is not None: + group_key = str(group) + group_failures[group_key] = group_failures.get(group_key, 0) + 1 + + category = item.get("category") + category_key = str(category) if category is not None else f"status:{status_key}" + category_counts[category_key] = category_counts.get(category_key, 0) + 1 + + if len(worst_ids) < ITEMS_SUMMARY_MAX_WORST_IDS: + item_id = item.get("id", item.get("question_id")) + if item_id is not None: + worst_ids.append(str(item_id)) + + return { + "total": total, + "failures": failures, + "status_counts": status_counts, + "group_failure_counts": _top_n(group_failures, ITEMS_SUMMARY_MAX_GROUPS), + "category_counts": _top_n(category_counts, ITEMS_SUMMARY_MAX_CATEGORIES), + "worst_ids": worst_ids, + } + + +def _top_n(counts: dict[str, int], cap: int) -> dict[str, int]: + """Return the `cap` highest-count entries, ordered by count desc then key.""" + if len(counts) <= cap: + return dict(sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))) + ranked = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[:cap] + return dict(ranked) + + class ContextManager: """Manages context.md for tracking generation evolution in a run""" @@ -355,8 +444,12 @@ def _extract_metrics(self, gen_dir: str) -> dict[str, Any]: elif isinstance(value, dict): # Skip other nested dicts continue + elif key == "items" and isinstance(value, list) and len(value) > 0: + # Verifier->feedback contract: retain a bounded summary of the + # optional grader `items` array instead of dropping it. + metrics["items_summary"] = summarize_items(value) elif isinstance(value, list) and len(value) > 0: - # Skip lists + # Skip other lists continue # Priority 2: detailed_results.json diff --git a/sia/orchestrator.py b/sia/orchestrator.py index 0411565..8934e62 100644 --- a/sia/orchestrator.py +++ b/sia/orchestrator.py @@ -43,6 +43,7 @@ """ import asyncio +import contextlib import glob import json import os @@ -51,6 +52,7 @@ import traceback from datetime import datetime from pathlib import Path +from typing import Any from sia import __version__, cli from sia.agent_reference import ResolvedAgentReference, copy_reference_into, resolve_agent_reference @@ -59,7 +61,7 @@ from sia.layout import BUNDLED_TASKS, Names, RunLayout, TaskLayout, resolve_task_dir, venv_python_path from sia.logging_setup import configure_logging, get_logger from sia.profiles import MetaAgentProfile, load_meta_agent_profile, load_target_agent_profile -from sia.prompts import build_feedback_prompt, build_meta_prompt +from sia.prompts import HELD_OUT_GROUND_TRUTH_NOTICE, build_feedback_prompt, build_meta_prompt from sia.providers import Provider from sia.results import FeedbackContext, TargetAgentResult from sia.run_setup import RunSetup, TaskFiles, install_requirements, load_task_files, setup_run_directory @@ -67,14 +69,17 @@ __all__ = [ "BUNDLED_TASKS", + "HELD_OUT_GROUND_TRUTH_NOTICE", "RunSetup", "TaskFiles", "build_feedback_prompt", "build_meta_prompt", + "compute_protected_paths", "load_agent_execution", "load_task_files", "main", "resolve_task_dir", + "restricted_access", "run_evaluation", "run_generation", "setup_run_directory", @@ -88,6 +93,54 @@ # ======================== +def compute_protected_paths(task_dir: str) -> list[str]: + """Return the task's held-out ground-truth dirs that meta/feedback agents must not read. + + Generic across all SIA tasks: the public/private convention places grader-only ground + truth under ``/data/private``. Returns the absolute path when that dir exists, + or an empty list (no-op) when the task ships no private dir. + """ + private_dir = os.path.join(task_dir, "data/private") + if os.path.isdir(private_dir): + return [os.path.abspath(private_dir)] + return [] + + +@contextlib.contextmanager +def restricted_access(paths: list[str], enabled: bool): + """Temporarily strip all filesystem permissions on ``paths`` for the duration of the block. + + Impl-agnostic OS-level backstop to the prompt notice + claude-impl PreToolUse tripwire: + while a meta/feedback agent runs, a read of the task's held-out ``data/private`` fails at + the filesystem layer even via a subprocess, an obfuscated shell, or a path the substring + matcher would miss — and for the agent impls that ignore the PreToolUse kwarg, this is the + only filesystem defense. The original mode is restored on exit (including on error). A + no-op when ``enabled`` is False or ``paths`` is empty. + + Process-global by nature (it chmods a shared dir), so it is applied only around the + sequential meta/feedback agent calls — never around grading, which legitimately reads the + dir. Does not defend against a root process or a copy-out-then-read, and a hard kill + (SIGKILL) mid-block leaves the dir stripped until restored; OS-level container sandboxing + (``--sandbox docker``) remains the strongest boundary. + """ + if not enabled or not paths: + yield + return + saved: list[tuple[str, int]] = [] + try: + for path in paths: + try: + saved.append((path, os.stat(path).st_mode)) + os.chmod(path, 0o000) + except OSError as exc: + logger.warning(f"private-dir guard: could not restrict {path}: {exc}") + yield + finally: + for path, mode in saved: + with contextlib.suppress(OSError): + os.chmod(path, mode) + + def load_agent_execution(gen_directory, config: Config | None = None): """ Load execution logs with automatic format detection. @@ -422,6 +475,122 @@ def _run_target_agent( return TargetAgentResult(False, stdout, "", error_msg).as_tuple() +# Generic render whitelist for a single failing eval item. Only these keys reach the +# feedback prompt — no task-shaped key (e.g. a reference answer carried elsewhere) can +# ride along. +_ITEM_RENDER_FIELDS = ("id", "group", "status", "category", "input", "output", "detail") + +_ANTI_REWARD_HACK_FRAMING = ( + "The held-out reference answers are intentionally withheld. Improve the agent by " + "reasoning about WHY these inputs failed (the failing inputs/outputs above) — " + "never by hardcoding or matching specific answers." +) + + +def _select_failures(items: list[Any], pass_statuses: tuple[str, ...], max_failures: int) -> list[dict]: + """Pick up to `max_failures` FAILED items, diversified across status and group. + + Round-robins over (status, group) buckets so a single dominant status or group + cannot crowd out the others — the feedback agent sees a spread of failure modes. + Reads only the generic `status` and `group` item keys. + """ + pass_set = set(pass_statuses) + buckets: dict[tuple[str, str], list[dict]] = {} + for item in items: + if not isinstance(item, dict): + continue + status = item.get("status") + if status in pass_set: + continue + key = (str(status), str(item.get("group"))) + buckets.setdefault(key, []).append(item) + + selected: list[dict] = [] + bucket_lists = list(buckets.values()) + cursor = 0 + while len(selected) < max_failures and any(bucket_lists): + bucket = bucket_lists[cursor % len(bucket_lists)] + if bucket: + selected.append(bucket.pop(0)) + cursor += 1 + if cursor % len(bucket_lists) == 0: + bucket_lists = [b for b in bucket_lists if b] + cursor = 0 + return selected + + +def _render_item(item: dict) -> dict: + """Project a failing item onto the generic render whitelist (present keys only). + + Constructs a NEW dict containing only `_ITEM_RENDER_FIELDS` keys that are present + in the item — never copies the source dict, so no task-specific key (and no + reference answer carried elsewhere) can ride along. + """ + return {field: item[field] for field in _ITEM_RENDER_FIELDS if field in item} + + +def _collect_scalars(data: dict) -> dict: + """Recursively project a results.json mapping onto its scalar metrics only. + + Keeps scalar leaves (numbers, strings, booleans) at any nesting depth and preserves + the surrounding dict shape, so aggregate blocks such as a nested ``summary`` object + survive. Every list is DROPPED: per-item record arrays (e.g. ``results`` / ``details``) + are where a grader's held-out reference answers live, so they never enter the summary. + Empty nested dicts are omitted. + """ + out: dict = {} + for key, value in data.items(): + if isinstance(value, (int, float, str, bool)): + out[key] = value + elif isinstance(value, dict): + nested = _collect_scalars(value) + if nested: + out[key] = nested + # lists are intentionally dropped (potential per-item reference answers) + return out + + +def _build_eval_summary( + eval_data: dict, + env_config: Config, +) -> str: + """Render a curated, reference-answer-free eval summary for the feedback prompt. + + Task-agnostic, and a near drop-in for the previous full results.json dump: + + - Every **scalar** metric is emitted as a JSON object, at any nesting depth, preserving + the original results.json shape and field names (e.g. ``accuracy``, ``correct``, + ``total``, or a nested ``summary`` block) — whatever the grader wrote, in its order. + - The reference-answer-free guarantee comes from DROPPING every list: ``results`` / + ``details`` (the task-shaped per-item records that may carry reference answers) and + any other array are excluded by ``_collect_scalars``. The only per-item channel is + the opt-in, generic ``items[]`` array, surfaced solely through the + ``_ITEM_RENDER_FIELDS`` whitelist. + - When ``items[]`` carries failures, a capped sample (diversified across ``status`` + and ``group``) plus an anti-reward-hack framing line are appended. A grader that + emits only scalars therefore gets output equivalent to the original JSON dump, + minus any answer-bearing arrays. + """ + scalars = _collect_scalars(eval_data) + summary = f"```json\n{json.dumps(scalars, indent=2)}\n```" + + items = eval_data.get("items") + if not isinstance(items, list): + items = [] + + failures = _select_failures(items, env_config.VERIFIER_PASS_STATUSES, env_config.FEEDBACK_FAILURE_SAMPLES) + if failures: + shown = [_render_item(item) for item in failures] + summary += ( + f"\n\n**Sample of FAILED held-out items** " + f"(up to {env_config.FEEDBACK_FAILURE_SAMPLES}, diversified across status and group):\n" + f"```json\n{json.dumps(shown, indent=2)}\n```" + f"\n\n{_ANTI_REWARD_HACK_FRAMING}" + ) + + return summary + + def _build_feedback_context( current_gen: int, gen_dir: str, @@ -504,12 +673,14 @@ def _build_feedback_context( else: with open(results_json_path, encoding="utf-8") as f: eval_data = json.load(f) + # Curated, gold-free summary — the full results dump leaked every + # held-out reference answer to the feedback agent (reward-hacking + # surface + turn pressure). Reads only the generic `items[]` contract. + eval_summary = _build_eval_summary(eval_data, cfg) eval_results_section = f""" **EVALUATION RESULTS**: -```json -{json.dumps(eval_data, indent=2)} -``` +{eval_summary} """ except (json.JSONDecodeError, OSError) as e: eval_results_section = f"\n**EVALUATION RESULTS**: Error loading results.json: {e}\n" @@ -617,16 +788,24 @@ def _run_feedback_agent( write_text(feedback_prompt_path, feedback_agent_prompt) logger.info(f" ✓ Saved feedback agent prompt to: {feedback_prompt_path}") - asyncio.run( - run_agent( - model_name=meta_profile.model, - max_turns=str(env_config.DEFAULT_MAX_TURNS), - prompt=feedback_agent_prompt, - agent_working_directory=next_gen_dir, - agent_impl=meta_profile.agent_impl, - provider=meta_profile.provider, + # dataset_dir is /data/public; the held-out dir is its sibling data/private. + task_dir = os.path.dirname(os.path.dirname(dataset_dir)) + protected_paths = compute_protected_paths(task_dir) + + # OS-level backstop: strip data/private permissions for the duration of the agent run + # (grading has already happened upstream and is outside this block). + with restricted_access(protected_paths, env_config.PRIVATE_DIR_GUARD): + asyncio.run( + run_agent( + model_name=meta_profile.model, + max_turns=str(env_config.DEFAULT_MAX_TURNS), + prompt=feedback_agent_prompt, + agent_working_directory=next_gen_dir, + agent_impl=meta_profile.agent_impl, + provider=meta_profile.provider, + protected_paths=protected_paths, + ) ) - ) next_gen = current_gen + 1 logger.info(f"Feedback agent completed. Created improved agent for generation {next_gen}") @@ -901,16 +1080,22 @@ def main(): write_text(meta_agent_prompt_path, meta_agent_prompt) logger.info(f" ✓ Saved meta-agent prompt to: {meta_agent_prompt_path}") - asyncio.run( - run_agent( - model_name=meta_model, - max_turns=str(env_config.DEFAULT_MAX_TURNS), - prompt=meta_agent_prompt, - agent_working_directory=run_setup.meta_agent_working_directory, - agent_impl=agent_impl, - provider=meta_profile.provider, + meta_protected_paths = compute_protected_paths(task_dir) + + # OS-level backstop: strip data/private permissions while the meta agent creates the + # initial target agent. The grader runs later (Section 5), outside this block. + with restricted_access(meta_protected_paths, env_config.PRIVATE_DIR_GUARD): + asyncio.run( + run_agent( + model_name=meta_model, + max_turns=str(env_config.DEFAULT_MAX_TURNS), + prompt=meta_agent_prompt, + agent_working_directory=run_setup.meta_agent_working_directory, + agent_impl=agent_impl, + provider=meta_profile.provider, + protected_paths=meta_protected_paths, + ) ) - ) # ======================== # SECTION 5: Main Loop - Run Target Agent and Feedback Agent diff --git a/sia/prompts.py b/sia/prompts.py index 4ca47a7..653e67f 100644 --- a/sia/prompts.py +++ b/sia/prompts.py @@ -14,6 +14,17 @@ from sia.providers import Provider from sia.run_setup import TaskFiles +# Generic, task-agnostic instruction injected into the meta- and feedback-agent +# harness-mode prompts. It tells the agent that held-out grader ground truth exists +# in a private directory and must never be accessed — without naming any concrete +# path, so it works for any SIA task. +HELD_OUT_GROUND_TRUTH_NOTICE = ( + "HELD-OUT EVALUATION INTEGRITY: The task's held-out evaluation ground truth lives in a " + "private directory used only by the grader. You MUST NOT read, list, glob, or otherwise " + "access it — improve solely from the feedback provided. Such access is blocked and would " + "invalidate the experiment." +) + def _reference_section(task_files: TaskFiles, reference_dir: str | None) -> str: """The reference paragraph of the meta prompt. @@ -653,6 +664,8 @@ def build_meta_prompt( Here is a sample agent execution trajectory: {json.dumps(task_files.sample_agent_execution, indent=2)} +{HELD_OUT_GROUND_TRUTH_NOTICE} + CRITICAL RULES - FOLLOW EXACTLY: 1. The current working directory is {working_dir}. Create the target_agent.py in the current working directory itself. @@ -942,6 +955,7 @@ def build_feedback_prompt( - Consider error handling, logging mechanisms, and robustness - Build upon successful patterns from previous generations (check context.md) - If execution log shows errors or is incomplete, suggest improvements to ensure proper logging +- {HELD_OUT_GROUND_TRUTH_NOTICE} NOTE: The agent execution log may be incomplete or contain errors if the target agent crashed. If you see an "error" field, focus on making the agent more robust to prevent such failures. """ diff --git a/tests/golden/feedback_prompt.txt b/tests/golden/feedback_prompt.txt index 751d293..4468294 100644 --- a/tests/golden/feedback_prompt.txt +++ b/tests/golden/feedback_prompt.txt @@ -84,5 +84,6 @@ Follow these steps: - Consider error handling, logging mechanisms, and robustness - Build upon successful patterns from previous generations (check context.md) - If execution log shows errors or is incomplete, suggest improvements to ensure proper logging +- HELD-OUT EVALUATION INTEGRITY: The task's held-out evaluation ground truth lives in a private directory used only by the grader. You MUST NOT read, list, glob, or otherwise access it — improve solely from the feedback provided. Such access is blocked and would invalidate the experiment. NOTE: The agent execution log may be incomplete or contain errors if the target agent crashed. If you see an "error" field, focus on making the agent more robust to prevent such failures. diff --git a/tests/golden/meta_prompt.txt b/tests/golden/meta_prompt.txt index fe9bfc1..737de34 100644 --- a/tests/golden/meta_prompt.txt +++ b/tests/golden/meta_prompt.txt @@ -20,6 +20,8 @@ Here is a sample agent execution trajectory: ] } +HELD-OUT EVALUATION INTEGRITY: The task's held-out evaluation ground truth lives in a private directory used only by the grader. You MUST NOT read, list, glob, or otherwise access it — improve solely from the feedback provided. Such access is blocked and would invalidate the experiment. + CRITICAL RULES - FOLLOW EXACTLY: 1. The current working directory is /WORK/run_1/gen_1. Create the target_agent.py in the current working directory itself. diff --git a/tests/golden/meta_prompt_openai.txt b/tests/golden/meta_prompt_openai.txt index 9b1a77c..16b0ab2 100644 --- a/tests/golden/meta_prompt_openai.txt +++ b/tests/golden/meta_prompt_openai.txt @@ -40,6 +40,8 @@ Here is a sample agent execution trajectory: ] } +HELD-OUT EVALUATION INTEGRITY: The task's held-out evaluation ground truth lives in a private directory used only by the grader. You MUST NOT read, list, glob, or otherwise access it — improve solely from the feedback provided. Such access is blocked and would invalidate the experiment. + CRITICAL RULES - FOLLOW EXACTLY: 1. The current working directory is /WORK/run_1/gen_1. Create the target_agent.py in the current working directory itself. diff --git a/tests/test_config.py b/tests/test_config.py index a38ade3..7b664e4 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -2,7 +2,7 @@ from dataclasses import fields -from sia.config import Config +from sia.config import Config, _to_bool, _to_str_tuple def test_default_values(): @@ -53,3 +53,45 @@ def test_config_is_dataclass_with_expected_fields(): "MAX_CONTEXT_FILE_SIZE", } assert expected.issubset(field_names) + + +# --- verifier->feedback contract knobs (gold-free eval summary) --------------- + + +def test_verifier_feedback_knob_defaults(): + cfg = Config() + assert cfg.VERIFIER_PASS_STATUSES == ("CORRECT", "PASS", "correct") + assert cfg.FEEDBACK_FAILURE_SAMPLES == 20 + assert cfg.PRIVATE_DIR_GUARD is True + + +def test_from_env_disables_private_dir_guard(monkeypatch): + monkeypatch.setenv("SIA_PRIVATE_DIR_GUARD", "0") + cfg = Config.from_env() + assert cfg.PRIVATE_DIR_GUARD is False + + +def test_from_env_overrides_verifier_pass_statuses_tuple(monkeypatch): + monkeypatch.setenv("SIA_VERIFIER_PASS_STATUSES", "OK, GREEN ,done") + cfg = Config.from_env() + assert cfg.VERIFIER_PASS_STATUSES == ("OK", "GREEN", "done") + + +def test_from_env_overrides_feedback_failure_samples(monkeypatch): + monkeypatch.setenv("SIA_FEEDBACK_FAILURE_SAMPLES", "5") + cfg = Config.from_env() + assert cfg.FEEDBACK_FAILURE_SAMPLES == 5 + + +def test_to_str_tuple_trims_and_drops_empty_parts(): + assert _to_str_tuple("a, b ,, c") == ("a", "b", "c") + + +def test_to_bool_parses_truthy_and_falsy_strings(): + assert _to_bool("1") is True + assert _to_bool("TRUE") is True + assert _to_bool(" yes ") is True + assert _to_bool("0") is False + assert _to_bool("false") is False + assert _to_bool("") is False + assert _to_str_tuple("") == () diff --git a/tests/test_context_manager.py b/tests/test_context_manager.py index a7aaea3..8ad159b 100644 --- a/tests/test_context_manager.py +++ b/tests/test_context_manager.py @@ -150,3 +150,88 @@ def test_multiple_generations_track_deltas(mock_llm, context_mgr, run_dir): content = (run_dir / "context.md").read_text() assert "Generation 2" in content assert "Modified by feedback agent" in content + + +# --- Bounded items summary (verifier->feedback contract) ------------------- + + +def _write_results(gen_dir, payload): + (gen_dir / "results.json").write_text(json.dumps(payload)) + + +def test_items_summary_counts_status_group_category(context_mgr, run_dir): + gen_dir = run_dir / "gen_1" + _write_results( + gen_dir, + { + "accuracy": 0.5, + "items": [ + {"id": "q0", "status": "CORRECT", "group": "world_1", "category": "ok"}, + {"id": "q1", "status": "WRONG", "group": "world_1", "category": "missing_distinct"}, + {"id": "q2", "status": "WRONG", "group": "world_1", "category": "missing_distinct"}, + {"id": "q3", "status": "EXEC_ERROR", "group": "car_1", "category": "bad_join"}, + ], + }, + ) + + metrics = context_mgr._extract_metrics(str(gen_dir)) + + assert metrics["accuracy"] == 0.5 + summary = metrics["items_summary"] + assert summary["total"] == 4 + assert summary["failures"] == 3 + assert summary["status_counts"] == {"CORRECT": 1, "WRONG": 2, "EXEC_ERROR": 1} + assert summary["group_failure_counts"] == {"world_1": 2, "car_1": 1} + assert summary["category_counts"] == {"missing_distinct": 2, "bad_join": 1} + assert summary["worst_ids"] == ["q1", "q2", "q3"] + + +def test_items_summary_derives_category_when_absent(context_mgr, run_dir): + gen_dir = run_dir / "gen_1" + _write_results( + gen_dir, + { + "items": [ + {"id": "q0", "status": "CORRECT", "group": "g1"}, + {"id": "q1", "status": "WRONG", "group": "g1"}, + ], + }, + ) + + summary = context_mgr._extract_metrics(str(gen_dir))["items_summary"] + + assert summary["failures"] == 1 + assert summary["group_failure_counts"] == {"g1": 1} + # No category on items -> coarse category derived from status. + assert summary["category_counts"] == {"status:WRONG": 1} + + +def test_no_items_key_produces_no_summary(context_mgr, run_dir): + gen_dir = run_dir / "gen_1" + _write_results(gen_dir, {"accuracy": 0.9, "correct": 9, "total": 10}) + + metrics = context_mgr._extract_metrics(str(gen_dir)) + + assert "items_summary" not in metrics + assert metrics["accuracy"] == 0.9 + assert metrics["correct"] == 9 + + +def test_items_summary_stays_bounded_on_large_array(context_mgr, run_dir): + from sia.context_manager import ( + ITEMS_SUMMARY_MAX_CATEGORIES, + ITEMS_SUMMARY_MAX_GROUPS, + ITEMS_SUMMARY_MAX_WORST_IDS, + ) + + gen_dir = run_dir / "gen_1" + items = [{"id": f"q{i}", "status": "WRONG", "group": f"g{i}", "category": f"c{i}"} for i in range(10_000)] + _write_results(gen_dir, {"items": items}) + + summary = context_mgr._extract_metrics(str(gen_dir))["items_summary"] + + assert summary["total"] == 10_000 + assert summary["failures"] == 10_000 + assert len(summary["group_failure_counts"]) <= ITEMS_SUMMARY_MAX_GROUPS + assert len(summary["category_counts"]) <= ITEMS_SUMMARY_MAX_CATEGORIES + assert len(summary["worst_ids"]) <= ITEMS_SUMMARY_MAX_WORST_IDS diff --git a/tests/test_orchestrator_helpers.py b/tests/test_orchestrator_helpers.py index 74b7738..3bd495e 100644 --- a/tests/test_orchestrator_helpers.py +++ b/tests/test_orchestrator_helpers.py @@ -1,8 +1,46 @@ """Unit tests for orchestrator helper functions.""" +import inspect import json +import re -from sia.orchestrator import load_agent_execution +from sia.config import Config +from sia.orchestrator import ( + HELD_OUT_GROUND_TRUTH_NOTICE, + TaskFiles, + _build_eval_summary, + _collect_scalars, + _render_item, + _select_failures, + build_feedback_prompt, + build_meta_prompt, + load_agent_execution, +) + + +def _make_task_files(): + return TaskFiles( + sample_task_descriptions="sample descriptions", + reference_target_agent_py="def main():\n pass\n", + sample_agent_execution={"role": "user"}, + task_md="solve the task", + ) + + +def _build_feedback_prompt(tmp_path): + return build_feedback_prompt( + current_gen=1, + max_gen=3, + task_files=_make_task_files(), + agent_py="print('agent')", + task="task body", + execution_status="SUCCESS", + execution_section="execution details", + run_dir=str(tmp_path), + next_gen_dir=str(tmp_path / "gen_2"), + previous_gens="None", + task_model="claude-haiku-4-5", + ) def test_load_single_trajectory(tmp_path): @@ -48,3 +86,180 @@ def test_load_empty_multi_trajectory_folder(tmp_path): data, is_multi = load_agent_execution(str(tmp_path)) assert is_multi assert "error" in data + + +# --- generic eval summary (task-agnostic items[] contract) ------------------- + + +def _eval_data_with_items(items, results=None): + data = { + "accuracy_percent": 50.0, + "correct": 1, + "wrong_answer": 1, + "exec_error": 0, + "missing": 0, + "items": items, + } + if results is not None: + data["results"] = results + return data + + +def test_eval_summary_ignores_results_and_never_leaks_gold(): + """The framework reads only items[]; a gold sentinel in results[] must not appear.""" + sentinel = "SELECT __GOLD_SENTINEL__" + items = [ + {"id": "q0", "status": "CORRECT", "group": "g1"}, + {"id": "q1", "status": "WRONG", "group": "g1", "input": "ask", "output": "SELECT bad"}, + ] + results = [ + {"id": "q0", "status": "CORRECT", "reference_answer": sentinel}, + {"id": "q1", "status": "WRONG", "reference_answer": sentinel}, + ] + eval_data = _eval_data_with_items(items, results=results) + + summary = _build_eval_summary(eval_data, Config()) + + assert sentinel not in summary + assert "__GOLD_SENTINEL__" not in summary + + +def test_eval_summary_renders_only_failed_items_with_generic_fields(): + items = [ + {"id": "q0", "status": "CORRECT", "group": "g1"}, + {"id": "q1", "status": "WRONG", "group": "g1", "input": "ask", "output": "bad", "detail": "mismatch"}, + {"id": "q2", "status": "EXEC_ERROR", "group": "g2", "input": "ask2", "output": "x", "detail": "no col"}, + ] + summary = _build_eval_summary(_eval_data_with_items(items), Config()) + + # Failed items rendered; the passing item's id is absent from the failure block. + assert "q1" in summary + assert "q2" in summary + assert "no col" in summary + # Top-level scalars are emitted in the original results.json JSON shape. + assert '"accuracy_percent": 50.0' in summary + + +def test_eval_summary_caps_and_stratifies_failures_across_status_and_group(): + cap = 4 + cfg = Config() + cfg.FEEDBACK_FAILURE_SAMPLES = cap + items = [] + # 6 WRONG in g1, 6 EXEC_ERROR in g2 -> selection must spread across both. + for i in range(6): + items.append({"id": f"w{i}", "status": "WRONG", "group": "g1"}) + for i in range(6): + items.append({"id": f"e{i}", "status": "EXEC_ERROR", "group": "g2"}) + + failures = _select_failures(items, cfg.VERIFIER_PASS_STATUSES, cap) + + assert len(failures) == cap + statuses = {f["status"] for f in failures} + assert statuses == {"WRONG", "EXEC_ERROR"} # stratified, not all from one bucket + + +def test_eval_summary_scalars_only_when_no_items(): + """A grader that emits only scalars gets the plain JSON block (original shape).""" + summary = _build_eval_summary({"accuracy_percent": 100.0}, Config()) + assert '"accuracy_percent": 100.0' in summary + assert summary.startswith("```json") + assert "Sample of FAILED" not in summary + + +def test_eval_summary_preserves_scalar_fields_and_drops_gold_arrays(): + """Scalars (incl. `total`) are kept in the original JSON shape; array fields that + may carry reference answers (`results`) are excluded.""" + eval_data = { + "accuracy": 0.9, + "correct": 9, + "total": 10, + "results": [{"id": "q1", "reference_answer": "SECRET"}], # must not appear + } + summary = _build_eval_summary(eval_data, Config()) + assert '"accuracy": 0.9' in summary + assert '"correct": 9' in summary + assert '"total": 10' in summary # the field the original feedback context kept + assert "results" not in summary + assert "SECRET" not in summary + + +def test_eval_summary_surfaces_nested_summary_and_drops_gold_results(): + """Real-grader shape: scalars live in a nested `summary` block and the answer key + lives in a per-item `results[]` list. The summary must surface the nested scalars + and drop the gold-bearing list entirely.""" + eval_data = { + "summary": {"total_questions": 3, "correct": 1, "wrong_answer": 2, "accuracy": 33.3}, + "results": [ + {"question_id": "c1", "expected": "SECRET_MOVE_Nf3", "predicted": "a3", "status": "WRONG"}, + {"question_id": "c2", "expected": "SECRET_MOVE_Qd8", "predicted": "Qd8", "status": "CORRECT"}, + ], + } + summary = _build_eval_summary(eval_data, Config()) + # Nested aggregate scalars are surfaced. + assert '"accuracy": 33.3' in summary + assert '"correct": 1' in summary + # The per-item answer key never appears. + assert "SECRET_MOVE_Nf3" not in summary + assert "SECRET_MOVE_Qd8" not in summary + assert "expected" not in summary + assert "results" not in summary + + +def test_collect_scalars_recurses_dicts_and_drops_lists(): + data = { + "accuracy": 0.5, + "summary": {"correct": 1, "total": 2, "nested": {"deep": "x"}}, + "results": [{"expected": "GOLD"}], # list dropped wholesale + "empty": {}, # empty nested dict omitted + } + out = _collect_scalars(data) + assert out == {"accuracy": 0.5, "summary": {"correct": 1, "total": 2, "nested": {"deep": "x"}}} + assert "results" not in out + assert "empty" not in out + + +def test_render_item_projects_only_generic_whitelist(): + item = { + "id": "q1", + "status": "WRONG", + "group": "g1", + "category": "wrong_answer", + "input": "ask", + "output": "bad", + "detail": "mismatch", + "reference_answer": "secret gold", # must be dropped + "extra_task_key": "g1", # task-specific key, must be dropped + } + rendered = _render_item(item) + assert set(rendered.keys()) == {"id", "status", "group", "category", "input", "output", "detail"} + assert "reference_answer" not in rendered + assert "extra_task_key" not in rendered + + +def test_eval_summary_source_reads_only_generic_keys(): + """Genericity guard: the eval-summary code path must not hardcode task-specific keys.""" + forbidden = ("gold", "reference_answer", "answer_key", "db_id") + for fn in (_build_eval_summary, _collect_scalars, _select_failures, _render_item): + src = inspect.getsource(fn).lower() + for token in forbidden: + assert token not in src, f"{fn.__name__} references task-specific identifier {token!r}" + # `sql` as a standalone word must not appear (substring guard tolerates 'results'). + assert not re.search(r"\bsql\b", src), f"{fn.__name__} references 'sql'" + + +# --- Held-out ground-truth seal: prompt notice ------------------------------- + + +def test_meta_prompt_carries_held_out_notice(): + prompt = build_meta_prompt(_make_task_files(), "claude-haiku-4-5", "/tmp/wd") + assert HELD_OUT_GROUND_TRUTH_NOTICE in prompt + + +def test_feedback_prompt_carries_held_out_notice(tmp_path): + prompt = _build_feedback_prompt(tmp_path) + assert HELD_OUT_GROUND_TRUTH_NOTICE in prompt + + +def test_held_out_notice_names_no_concrete_private_path(): + """The generic notice must not hardcode any task's private directory path.""" + assert "data/private" not in HELD_OUT_GROUND_TRUTH_NOTICE diff --git a/tests/test_protected_paths.py b/tests/test_protected_paths.py new file mode 100644 index 0000000..c5f9654 --- /dev/null +++ b/tests/test_protected_paths.py @@ -0,0 +1,218 @@ +"""Tests for the generic held-out-data seal: the protected-path matcher, +the PreToolUse deny hook, and the ClaudeAgentOptions builder wiring. + +All tests are offline -- they exercise the pure matcher and the option-builder directly, +with no live Claude SDK / CLI invocation. +""" + +import asyncio +import inspect + +import pytest + +from sia.agent_impls.claude import ( + _accesses_protected_path, + _build_claude_options, + _make_protected_path_hook, +) +from sia.orchestrator import restricted_access + +PROTECTED = "/tmp/task/data/private" + + +# --------------------------------------------------------------------------- +# _accesses_protected_path matrix +# --------------------------------------------------------------------------- + + +def test_read_inside_protected_is_blocked(): + assert _accesses_protected_path("Read", {"file_path": f"{PROTECTED}/gold.json"}, [PROTECTED]) + + +def test_read_outside_protected_is_allowed(): + assert not _accesses_protected_path("Read", {"file_path": "/tmp/task/cand_0/target_agent.py"}, [PROTECTED]) + + +def test_bash_cat_protected_is_blocked(): + assert _accesses_protected_path("Bash", {"command": f"cat {PROTECTED}/gold.json"}, [PROTECTED]) + + +def test_bash_public_db_is_allowed(): + assert not _accesses_protected_path("Bash", {"command": "sqlite3 data/public/x.db 'select 1'"}, [PROTECTED]) + + +def test_bash_find_protected_is_blocked(): + assert _accesses_protected_path("Bash", {"command": f"find {PROTECTED} -name '*'"}, [PROTECTED]) + + +def test_bash_relative_protected_form_is_blocked(): + # The normalized relative form `data/private` is matched even without the absolute path. + assert _accesses_protected_path("Bash", {"command": "cd data/private && ls"}, [PROTECTED]) + + +def test_glob_protected_is_blocked(): + assert _accesses_protected_path("Glob", {"pattern": f"{PROTECTED}/**"}, [PROTECTED]) + + +def test_glob_public_is_allowed(): + assert not _accesses_protected_path("Glob", {"pattern": "data/public/*"}, [PROTECTED]) + + +def test_edit_inside_protected_is_blocked(): + assert _accesses_protected_path("Edit", {"file_path": f"{PROTECTED}/gold.json"}, [PROTECTED]) + + +def test_write_inside_protected_is_blocked(): + assert _accesses_protected_path("Write", {"file_path": f"{PROTECTED}/leak.txt"}, [PROTECTED]) + + +def test_relative_dotdot_resolving_to_protected_is_blocked(): + # A `..` path that resolves into the protected dir must be caught after realpath/abspath. + candidate = f"{PROTECTED}/../private/gold.json" + assert _accesses_protected_path("Read", {"file_path": candidate}, [PROTECTED]) + + +def test_unknown_tool_is_allowed(): + assert not _accesses_protected_path("WebSearch", {"query": PROTECTED}, [PROTECTED]) + + +def test_empty_protected_dirs_allows_everything(): + assert not _accesses_protected_path("Read", {"file_path": f"{PROTECTED}/gold.json"}, []) + + +# --------------------------------------------------------------------------- +# Hook output shapes +# --------------------------------------------------------------------------- + + +def _run_hook(tool_name, tool_input, protected_dirs=None): + cb = _make_protected_path_hook(protected_dirs or [PROTECTED]) + return asyncio.run(cb({"tool_name": tool_name, "tool_input": tool_input}, None, None)) + + +def test_hook_denies_protected_access_with_canonical_shape(): + result = _run_hook("Read", {"file_path": f"{PROTECTED}/gold.json"}) + output = result["hookSpecificOutput"] + assert output["hookEventName"] == "PreToolUse" + assert output["permissionDecision"] == "deny" + assert PROTECTED in output["permissionDecisionReason"] + + +def test_hook_allows_non_protected_access_with_empty_dict(): + result = _run_hook("Read", {"file_path": "/tmp/task/cand_0/target_agent.py"}) + assert result == {} + + +# --------------------------------------------------------------------------- +# ClaudeAgentOptions builder wiring +# --------------------------------------------------------------------------- + + +def test_builder_registers_hook_when_protected_paths_present(): + options = _build_claude_options("haiku", "20", "/tmp/wd", protected_paths=[PROTECTED]) + assert options.hooks is not None + matchers = options.hooks["PreToolUse"] + assert len(matchers) == 1 + assert len(matchers[0].hooks) == 1 + + +def test_builder_registers_no_hook_when_protected_paths_empty(): + options = _build_claude_options("haiku", "20", "/tmp/wd", protected_paths=[]) + assert options.hooks is None + + +def test_builder_registers_no_hook_when_protected_paths_none(): + options = _build_claude_options("haiku", "20", "/tmp/wd", protected_paths=None) + assert options.hooks is None + + +# --------------------------------------------------------------------------- +# Orchestrator wiring: compute_protected_paths +# --------------------------------------------------------------------------- + + +def test_compute_protected_paths_returns_private_when_present(tmp_path): + from sia.orchestrator import compute_protected_paths + + private = tmp_path / "data" / "private" + private.mkdir(parents=True) + result = compute_protected_paths(str(tmp_path)) + assert result == [str(private.resolve())] + + +def test_compute_protected_paths_empty_when_absent(tmp_path): + from sia.orchestrator import compute_protected_paths + + (tmp_path / "data" / "public").mkdir(parents=True) + assert compute_protected_paths(str(tmp_path)) == [] + + +# --------------------------------------------------------------------------- +# Genericity: the seal code must stay task-agnostic +# --------------------------------------------------------------------------- + + +def test_seal_code_carries_no_task_specific_identifiers(): + """The seal must reference the held-out dir abstractly, never a concrete task's + SQL/gold/benchmark vocabulary -- otherwise it leaks task knowledge into a generic layer. + """ + import sia.agent_impls.claude as claude_mod + + sources = "\n".join( + inspect.getsource(obj) + for obj in ( + claude_mod._looks_like_path, + claude_mod._resolves_inside, + claude_mod._accesses_protected_path, + claude_mod._make_protected_path_hook, + claude_mod._build_claude_options, + ) + ).lower() + + forbidden = ("sql", "gold", "lawbench", "gpqa", "spaceship", "chess") + leaked = [token for token in forbidden if token in sources] + assert not leaked, f"seal code leaked task-specific identifiers: {leaked}" + + +# --------------------------------------------------------------------------- +# OS-level guard: restricted_access (impl-agnostic backstop) +# --------------------------------------------------------------------------- + + +def _make_private(tmp_path): + private = tmp_path / "data" / "private" + private.mkdir(parents=True) + (private / "answers.json").write_text("gold") + return private + + +def test_restricted_access_strips_then_restores_permissions(tmp_path): + private = _make_private(tmp_path) + original = private.stat().st_mode + + with restricted_access([str(private)], enabled=True): + # All permission bits stripped during the block (root-safe: check the mode, not access). + assert private.stat().st_mode & 0o777 == 0 + assert private.stat().st_mode == original # restored after the block + + +def test_restricted_access_is_noop_when_disabled(tmp_path): + private = _make_private(tmp_path) + original = private.stat().st_mode + with restricted_access([str(private)], enabled=False): + assert private.stat().st_mode == original # untouched + assert private.stat().st_mode == original + + +def test_restricted_access_restores_on_error(tmp_path): + private = _make_private(tmp_path) + original = private.stat().st_mode + with pytest.raises(ValueError), restricted_access([str(private)], enabled=True): + raise ValueError("boom") + assert private.stat().st_mode == original # restored even on exception + + +def test_restricted_access_tolerates_missing_path(tmp_path): + missing = str(tmp_path / "data" / "private") # never created + with restricted_access([missing], enabled=True): + pass # must not raise