Fix: stop leaking held-out reference answers to the meta/feedback agents

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 `<task_dir>/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 `<task_dir>/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

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)

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 ``<protected>/**`` 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
 

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)

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