diff --git a/skillopt_sleep/__init__.py b/skillopt_sleep/__init__.py index 9c7581f9..787a7545 100644 --- a/skillopt_sleep/__init__.py +++ b/skillopt_sleep/__init__.py @@ -16,5 +16,14 @@ """ from __future__ import annotations -__all__ = ["__version__"] +from typing import Any + +__all__ = ["__version__", "AgenticReplayBackend"] __version__ = "0.2.0" + + +def __getattr__(name: str) -> Any: + if name == "AgenticReplayBackend": + from skillopt_sleep.agentic_replay import AgenticReplayBackend + return AgenticReplayBackend + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/skillopt_sleep/agentic_replay.py b/skillopt_sleep/agentic_replay.py new file mode 100644 index 00000000..055f5e65 --- /dev/null +++ b/skillopt_sleep/agentic_replay.py @@ -0,0 +1,1332 @@ +"""Tiered Agentic Replay helper and disposable scratch workspace (Issue #155). + +This module implements the first workspace-lifecycle slice for agentic replay: +- A disposable scratch copy (NOT a linked git worktree and NOT an OS/container + sandbox) with an independent bare harness git directory outside the diffed tree. +- Strict allowlist environment scrubbing for child subprocesses, redirecting + ``HOME`` / ``USERPROFILE`` / XDG / CLI config directories to empty locations + outside the diffed working directory. + Note: there is NO OS, user, or network isolation boundary in this slice. A + child running under the same UID can still read files by absolute path + (such as ``~/.ssh`` or ``~/.claude``) or open network sockets directly. +- Deterministic task verification against workspace state/tests, with no + verifier configured treated as ``inconclusive`` (``insufficient_evidence``). +- Explicit opt-in execution (``enabled=False`` by default) for injectable + test/runner callables; live CLI binary execution is not enabled here. +""" +from __future__ import annotations + +import inspect +import json +import os +import re +import shutil +import signal +import stat +import subprocess +import sys +import tarfile +import tempfile +import threading +import time +from dataclasses import dataclass +from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple + +from skillopt_sleep.types import EditRecord, ReplayResult, TaskRecord + +# Outcomes +OUTCOME_PASS = "pass" +OUTCOME_FAIL = "fail" +OUTCOME_INCONCLUSIVE = "inconclusive" + +# Typed inconclusive reasons +REASON_TIMEOUT = "timeout" +REASON_INFRA = "infra" +REASON_FIXTURE_FAILURE = "fixture_failure" +REASON_BUDGET_EXHAUSTED = "budget_exhausted" +REASON_INSUFFICIENT_EVIDENCE = "insufficient_evidence" + +# Strict OS-only allowlist for child subprocesses (matches Backend isolation ethos) +_ENV_ALLOWLIST_COMMON = ( + "PATH", + "LANG", + "LC_ALL", + "LC_CTYPE", + "TZ", + "TERM", + "COLORTERM", +) +_ENV_ALLOWLIST_WINDOWS = ( + "SYSTEMROOT", + "WINDIR", + "COMSPEC", + "PATHEXT", + "OS", + "NUMBER_OF_PROCESSORS", + "PROCESSOR_ARCHITECTURE", + "SYSTEMDRIVE", +) + + +class FixtureSetupError(RuntimeError): + """Raised when fixture validation, git materialization, or baseline setup fails.""" + + +@dataclass +class AgenticReplayConfig: + """Configuration, opt-in gate, and budgets for agentic replay.""" + + enabled: bool = False # Strict boolean opt-in; disabled by default + fixture_root: str = "" # Explicit root required for directory fixtures + max_turns: int = 10 # Turn/action cap per agentic replay + max_tokens: int = 0 # Cumulative token cap across run (0 = unlimited) + timeout_sec: float = 60.0 # Wall-clock timeout in seconds + max_replays_per_run: int = 5 # Per-night/run cap on agentic replays + isolate_env: bool = True # Use allowlisted env + isolated empty config dirs + min_decidable_actions: int = 1 # Minimum actions required to score vs inconclusive + min_coverage: float = 0.5 # Minimum pairwise decidable val fraction for gate + pinned_commit: str = "" # Optional commit materialized via git archive + max_output_bytes: int = 64 * 1024 # Cap buffered stdout/stderr per subprocess + max_diff_bytes: int = 256 * 1024 # Cap buffered unified diff per replay + + def __post_init__(self) -> None: + self.enabled = self.enabled is True + self.isolate_env = bool(self.isolate_env) + + +def agentic_config_from_sleep_config(cfg: Mapping[str, Any]) -> AgenticReplayConfig: + """Build an :class:`AgenticReplayConfig` from a SleepConfig/dict mapping.""" + return AgenticReplayConfig( + enabled=cfg.get("agentic_replay", False) is True, + fixture_root=str(cfg.get("agentic_fixture_root", "") or ""), + max_turns=int(cfg.get("agentic_max_turns", 10) or 0), + max_tokens=int(cfg.get("agentic_max_tokens", 0) or 0), + timeout_sec=float(cfg.get("agentic_timeout_sec", 60.0) or 60.0), + max_replays_per_run=int(cfg.get("agentic_max_replays", 5) or 0), + isolate_env=bool(cfg.get("agentic_isolate_env", True)), + min_decidable_actions=int(cfg.get("agentic_min_actions", 1) or 0), + min_coverage=float(cfg.get("agentic_min_coverage", 0.5) or 0.5), + ) + + +def _ensure_within_root(target: str, root: str) -> str: + """Resolve ``target`` and verify it stays strictly within ``root``.""" + if not root or not str(root).strip(): + raise FixtureSetupError("fixture_root must be configured when loading fixture directories") + real_root = os.path.realpath(root) + if not os.path.isdir(real_root): + raise FixtureSetupError(f"fixture_root is not an existing directory: {root!r}") + candidate = target if os.path.isabs(target) else os.path.join(real_root, target) + real_target = os.path.realpath(candidate) + try: + common = os.path.commonpath([real_root, real_target]) + except ValueError as exc: + raise FixtureSetupError(f"Fixture path {target!r} is outside fixture_root {root!r}") from exc + if common != real_root: + raise FixtureSetupError(f"Fixture path {target!r} escapes fixture_root {root!r}") + if not os.path.isdir(real_target): + raise FixtureSetupError(f"Fixture path is not a directory: {target!r}") + return real_target + + +def _read_stream_capped(stream: Any, cap_bytes: int, dest: List[Tuple[bytes, bool]]) -> None: + """Drain ``stream`` in chunks, retaining at most ``cap_bytes`` bytes in memory.""" + buf = bytearray() + truncated = False + limit = max(0, int(cap_bytes)) + try: + while True: + chunk = stream.read(65536) + if not chunk: + break + if len(buf) < limit: + take = limit - len(buf) + buf.extend(chunk[:take]) + if len(chunk) > take: + truncated = True + else: + truncated = True + except OSError: + pass + finally: + try: + stream.close() + except OSError: + pass + dest.append((bytes(buf), truncated)) + + +def _kill_process_tree(proc: subprocess.Popen[bytes]) -> None: + """Terminate ``proc`` and any descendants started in its process group.""" + if os.name == "nt": + try: + subprocess.run( + ["taskkill", "/F", "/T", "/PID", str(proc.pid)], + capture_output=True, + timeout=5.0, + ) + except (subprocess.SubprocessError, OSError): + pass + try: + proc.kill() + except OSError: + pass + else: + try: + os.killpg(proc.pid, signal.SIGKILL) + except OSError: + try: + proc.kill() + except OSError: + pass + + +def _run_bounded_subprocess( + cmd: Sequence[str], + *, + cwd: str, + env: Optional[Dict[str, str]] = None, + stdin_bytes: Optional[bytes] = None, + timeout_sec: float = 60.0, + max_output_bytes: int = 64 * 1024, +) -> Tuple[bool, int, bytes, bytes, bool, bool]: + """Run ``cmd`` in its own process group with bounded output capture. + + Returns ``(timed_out, returncode, stdout_bytes, stderr_bytes, stdout_trunc, stderr_trunc)``. + """ + if isinstance(cmd, (str, bytes)) or not cmd: + raise ValueError("cmd must be a non-empty sequence of string arguments") + cmd_list = [str(part) for part in cmd] + + popen_kwargs: Dict[str, Any] = { + "cwd": cwd, + "env": env, + "stdin": subprocess.PIPE if stdin_bytes is not None else subprocess.DEVNULL, + "stdout": subprocess.PIPE, + "stderr": subprocess.PIPE, + } + if os.name == "nt": + popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP + else: + popen_kwargs["start_new_session"] = True + + proc = subprocess.Popen(cmd_list, **popen_kwargs) + + out_holder: List[Tuple[bytes, bool]] = [] + err_holder: List[Tuple[bytes, bool]] = [] + out_thread = threading.Thread( + target=_read_stream_capped, + args=(proc.stdout, max_output_bytes, out_holder), + daemon=True, + ) + err_thread = threading.Thread( + target=_read_stream_capped, + args=(proc.stderr, max_output_bytes, err_holder), + daemon=True, + ) + out_thread.start() + err_thread.start() + + in_thread: Optional[threading.Thread] = None + if stdin_bytes is not None and proc.stdin is not None: + def _feed_stdin() -> None: + try: + proc.stdin.write(stdin_bytes) # type: ignore[union-attr] + proc.stdin.flush() # type: ignore[union-attr] + except OSError: + pass + finally: + try: + proc.stdin.close() # type: ignore[union-attr] + except OSError: + pass + + in_thread = threading.Thread(target=_feed_stdin, daemon=True) + in_thread.start() + + timed_out = False + try: + proc.wait(timeout=max(0.05, float(timeout_sec))) + except subprocess.TimeoutExpired: + timed_out = True + _kill_process_tree(proc) + try: + proc.wait(timeout=2.0) + except subprocess.TimeoutExpired: + pass + + if in_thread is not None: + in_thread.join(timeout=1.0) + out_thread.join(timeout=1.5) + err_thread.join(timeout=1.5) + + out_bytes, out_trunc = out_holder[0] if out_holder else (b"", False) + err_bytes, err_trunc = err_holder[0] if err_holder else (b"", False) + rc = proc.returncode if proc.returncode is not None else -1 + return timed_out, rc, out_bytes, err_bytes, out_trunc, err_trunc + + +class AgenticWorkspace: + """Disposable scratch copy for running one agentic replay. + + Directory layout places ``home``, ``tmp``, XDG/CLI config directories, + prompt payload files, and ``harness.git`` alongside—NOT inside—``work``, + so agent tool writes to ``$HOME`` or git metadata cannot pollute ``get_diff()``. + """ + + def __init__( + self, + base_dir: Optional[str] = None, + pinned_commit: Optional[str] = None, + isolate_env: bool = True, + *, + fixture_root: str = "", + max_diff_bytes: int = 256 * 1024, + max_output_bytes: int = 64 * 1024, + ) -> None: + self.base_dir = base_dir + self.pinned_commit = pinned_commit or "" + self.isolate_env = bool(isolate_env) + self.fixture_root = fixture_root + self.max_diff_bytes = max(1024, int(max_diff_bytes)) + self.max_output_bytes = max(1024, int(max_output_bytes)) + + self._root_dir = tempfile.mkdtemp(prefix="skillopt_agentic_") + self.workspace_path = os.path.join(self._root_dir, "work") + self.home_dir = os.path.join(self._root_dir, "home") + self._tmp_dir = os.path.join(self._root_dir, "tmp") + self._xdg_config_home = os.path.join(self._root_dir, "xdg_config") + self._xdg_data_home = os.path.join(self._root_dir, "xdg_data") + self._xdg_cache_home = os.path.join(self._root_dir, "xdg_cache") + self._xdg_state_home = os.path.join(self._root_dir, "xdg_state") + self._claude_config_dir = os.path.join(self._root_dir, "claude_config") + self._codex_home = os.path.join(self._root_dir, "codex_home") + self._prompt_dir = os.path.join(self._root_dir, "prompt") + self._harness_git_dir = os.path.join(self._root_dir, "harness.git") + self._empty_hooks_dir = os.path.join(self._root_dir, "empty_hooks") + self._empty_gitconfig = os.path.join(self._root_dir, "empty_gitconfig") + self._is_setup = False + + def __enter__(self) -> "AgenticWorkspace": + self.setup() + return self + + def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + self.cleanup() + + def _ensure_layout_dirs(self) -> None: + for d in ( + self.workspace_path, + self.home_dir, + self._tmp_dir, + self._xdg_config_home, + self._xdg_data_home, + self._xdg_cache_home, + self._xdg_state_home, + self._claude_config_dir, + self._codex_home, + self._prompt_dir, + self._harness_git_dir, + self._empty_hooks_dir, + ): + os.makedirs(d, exist_ok=True) + with open(self._empty_gitconfig, "wb"): + pass + + def _allowlisted_env(self) -> Dict[str, str]: + """Minimal allowlisted child environment with isolated empty home/config dirs.""" + self._ensure_layout_dirs() + allowlist = list(_ENV_ALLOWLIST_COMMON) + if os.name == "nt": + allowlist.extend(_ENV_ALLOWLIST_WINDOWS) + env: Dict[str, str] = {} + for key in allowlist: + val = os.environ.get(key) + if val is not None: + env[key] = val + + env.update({ + "HOME": self.home_dir, + "USERPROFILE": self.home_dir, + "TMPDIR": self._tmp_dir, + "TMP": self._tmp_dir, + "TEMP": self._tmp_dir, + "XDG_CONFIG_HOME": self._xdg_config_home, + "XDG_DATA_HOME": self._xdg_data_home, + "XDG_CACHE_HOME": self._xdg_cache_home, + "XDG_STATE_HOME": self._xdg_state_home, + "CLAUDE_CONFIG_DIR": self._claude_config_dir, + "CODEX_HOME": self._codex_home, + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_CONFIG_GLOBAL": self._empty_gitconfig, + "GIT_TERMINAL_PROMPT": "0", + "GIT_ASKPASS": "", + "SSH_ASKPASS": "", + "NO_COLOR": "1", + }) + if os.name == "nt": + drive, tail = os.path.splitdrive(self.home_dir) + if drive: + env["HOMEDRIVE"] = drive + env["HOMEPATH"] = tail or "\\" + return env + + def build_isolated_env(self) -> Dict[str, str]: + """Return the child subprocess environment according to ``isolate_env``.""" + self._ensure_layout_dirs() + if self.isolate_env: + return self._allowlisted_env() + return dict(os.environ) + + def _safe_git_prefix(self, git_dir: str, work_tree: Optional[str] = None) -> List[str]: + cmd = [ + "git", + f"--git-dir={git_dir}", + ] + if work_tree is not None: + cmd.append(f"--work-tree={work_tree}") + cmd.extend([ + "-c", f"core.hooksPath={self._empty_hooks_dir}", + "-c", "core.fsmonitor=", + "-c", "diff.external=", + "-c", "commit.gpgsign=false", + "-c", "tag.gpgsign=false", + ]) + return cmd + + def _run_harness_git( + self, + args: Sequence[str], + *, + git_dir: Optional[str] = None, + work_tree: Optional[str] = None, + cwd: Optional[str] = None, + timeout_sec: float = 30.0, + max_output_bytes: Optional[int] = None, + ) -> Tuple[bytes, bool]: + """Run harness-side git strictly under allowlisted env without repo config hooks.""" + env = self._allowlisted_env() + cmd = ( + self._safe_git_prefix(git_dir, work_tree) + if git_dir is not None + else [ + "git", + "-c", f"core.hooksPath={self._empty_hooks_dir}", + "-c", "core.fsmonitor=", + "-c", "diff.external=", + "-c", "commit.gpgsign=false", + "-c", "tag.gpgsign=false", + ] + ) + cmd.extend(args) + timed_out, rc, out_b, err_b, out_trunc, _ = _run_bounded_subprocess( + cmd, + cwd=cwd or self._root_dir, + env=env, + timeout_sec=timeout_sec, + max_output_bytes=max_output_bytes or self.max_output_bytes, + ) + if timed_out: + raise FixtureSetupError(f"Harness git command timed out: {' '.join(args[:3])}") + if rc != 0: + err_msg = err_b.decode("utf-8", errors="replace").strip() + raise FixtureSetupError( + f"Harness git command failed (exit {rc}): {err_msg or ' '.join(args[:3])}" + ) + return out_b, out_trunc + + def _check_no_dotgit_pointers_or_alternates( + self, + directory: str, + *, + allow_top_git_dir: bool, + ) -> None: + """Reject linked ``.git`` files/symlinks and ``objects/info/alternates``.""" + top_dotgit = os.path.join(directory, ".git") + if os.path.islink(top_dotgit): + raise FixtureSetupError(f"Symlinked .git is not allowed in {directory!r}") + if os.path.isfile(top_dotgit): + raise FixtureSetupError( + f"Linked .git file is not allowed without archive materialization: {directory!r}" + ) + if os.path.isdir(top_dotgit): + if not allow_top_git_dir: + raise FixtureSetupError(f"Unexpected .git directory in {directory!r}") + alt = os.path.join(top_dotgit, "objects", "info", "alternates") + if os.path.exists(alt) or os.path.islink(alt): + raise FixtureSetupError(f"Git alternates file is not allowed: {alt!r}") + + for dirpath, dirnames, filenames in os.walk(directory, followlinks=False): + # Top-level .git dir was checked above; skip descending into it for nested checks + if dirpath == directory and ".git" in dirnames: + dirnames.remove(".git") + for name in filenames: + if name == ".git": + raise FixtureSetupError( + f"Nested .git file/submodule pointer is not allowed: {os.path.join(dirpath, name)!r}" + ) + for name in list(dirnames): + full = os.path.join(dirpath, name) + if name == ".git": + if os.path.islink(full) or not allow_top_git_dir: + raise FixtureSetupError(f"Nested .git entry is not allowed: {full!r}") + if os.path.islink(full): + # Check symlink does not point outside directory + real_dir = os.path.realpath(directory) + real_target = os.path.realpath(full) + try: + common = os.path.commonpath([real_dir, real_target]) + except ValueError as exc: + raise FixtureSetupError(f"Symlink escapes directory: {full!r}") from exc + if common != real_dir: + raise FixtureSetupError(f"Symlink escapes directory: {full!r}") + + def _check_source_repo_alternates(self, source_dir: str) -> None: + """Verify git object store for ``source_dir`` has no ``objects/info/alternates``.""" + for flag in ("--git-dir", "--git-common-dir"): + out_b, _ = self._run_harness_git(["-C", source_dir, "rev-parse", flag]) + raw_dir = out_b.decode("utf-8", errors="replace").strip() + if not raw_dir: + continue + resolved = raw_dir if os.path.isabs(raw_dir) else os.path.join(source_dir, raw_dir) + alt = os.path.join(os.path.realpath(resolved), "objects", "info", "alternates") + if os.path.exists(alt) or os.path.islink(alt): + raise FixtureSetupError(f"Git alternates file is not allowed in fixture: {alt!r}") + + def _materialize_via_git_archive(self, source_dir: str, pinned_commit: str) -> None: + """Materialize ``pinned_commit`` from ``source_dir`` without copying any ``.git``.""" + if not re.fullmatch(r"[A-Za-z0-9._/@^{}~-]+", pinned_commit) or pinned_commit.startswith("-"): + raise FixtureSetupError(f"Invalid pinned_commit reference: {pinned_commit!r}") + self._check_source_repo_alternates(source_dir) + + sha_bytes, _ = self._run_harness_git([ + "-C", source_dir, "rev-parse", "--verify", "--end-of-options", f"{pinned_commit}^{{commit}}", + ]) + resolved_sha = sha_bytes.decode("utf-8", errors="replace").strip() + if not re.fullmatch(r"[0-9a-fA-F]{7,64}", resolved_sha): + raise FixtureSetupError(f"Unexpected commit resolution for {pinned_commit!r}") + + archive_path = os.path.join(self._root_dir, "fixture_archive.tar") + env = self._allowlisted_env() + cmd = [ + "git", + "-C", source_dir, + "-c", f"core.hooksPath={self._empty_hooks_dir}", + "-c", "core.fsmonitor=", + "-c", "commit.gpgsign=false", + "archive", + "--format=tar", + resolved_sha, + ] + try: + with open(archive_path, "wb") as out_f: + popen_kw: Dict[str, Any] = { + "cwd": self._root_dir, + "env": env, + "stdin": subprocess.DEVNULL, + "stdout": out_f, + "stderr": subprocess.PIPE, + } + if os.name == "nt": + popen_kw["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP + else: + popen_kw["start_new_session"] = True + proc = subprocess.Popen(cmd, **popen_kw) + err_holder: List[Tuple[bytes, bool]] = [] + err_t = threading.Thread( + target=_read_stream_capped, + args=(proc.stderr, self.max_output_bytes, err_holder), + daemon=True, + ) + err_t.start() + try: + proc.wait(timeout=30.0) + except subprocess.TimeoutExpired as exc: + _kill_process_tree(proc) + err_t.join(timeout=1.0) + raise FixtureSetupError("git archive timed out") from exc + err_t.join(timeout=1.0) + if proc.returncode != 0: + err_b = err_holder[0][0] if err_holder else b"" + raise FixtureSetupError( + f"git archive failed for {pinned_commit!r}: " + f"{err_b.decode('utf-8', errors='replace').strip()}" + ) + + real_work = os.path.realpath(self.workspace_path) + with tarfile.open(archive_path, mode="r:") as tf: + safe_members: List[tarfile.TarInfo] = [] + for member in tf.getmembers(): + if member.name in ("", ".") or member.name.startswith("./"): + rel = member.name[2:] if member.name.startswith("./") else "" + else: + rel = member.name + norm_parts = [p for p in rel.replace("\\", "/").split("/") if p] + if not norm_parts: + if member.isdir(): + continue + raise FixtureSetupError("Unsafe empty archive member") + if os.path.isabs(rel) or os.path.splitdrive(rel)[0] or ".." in norm_parts: + raise FixtureSetupError(f"Archive path escape rejected: {member.name!r}") + if ".git" in norm_parts: + raise FixtureSetupError(f"Archive member inside .git rejected: {member.name!r}") + if member.isdev() or member.isfifo() or member.ischr() or member.isblk(): + raise FixtureSetupError(f"Special device member rejected: {member.name!r}") + dest_path = os.path.realpath(os.path.join(real_work, *norm_parts)) + if os.path.commonpath([real_work, dest_path]) != real_work: + raise FixtureSetupError(f"Archive target escapes workspace: {member.name!r}") + if member.issym() or member.islnk(): + link_base = real_work if member.islnk() else os.path.dirname(dest_path) + link_target = os.path.realpath(os.path.join(link_base, member.linkname)) + if os.path.commonpath([real_work, link_target]) != real_work: + raise FixtureSetupError( + f"Archive link target escapes workspace: {member.linkname!r}" + ) + safe_members.append(member) + tf.extractall(path=self.workspace_path, members=safe_members) + except (tarfile.TarError, OSError) as exc: + raise FixtureSetupError(f"Failed extracting archived fixture: {exc}") from exc + finally: + if os.path.exists(archive_path): + try: + os.remove(archive_path) + except OSError: + pass + + def _copy_plain_fixture(self, source_dir: str) -> None: + """Copy plain fixture files into ``workspace_path`` without copying ``.git``.""" + self._check_no_dotgit_pointers_or_alternates(source_dir, allow_top_git_dir=True) + real_source = os.path.realpath(source_dir) + for dirpath, dirnames, filenames in os.walk(real_source, followlinks=False): + dirnames[:] = [d for d in dirnames if d != ".git"] + rel_dir = os.path.relpath(dirpath, real_source) + dst_dir = ( + self.workspace_path + if rel_dir == "." + else os.path.join(self.workspace_path, rel_dir) + ) + os.makedirs(dst_dir, exist_ok=True) + for fname in filenames: + if fname == ".git": + continue + src_file = os.path.join(dirpath, fname) + if os.path.islink(src_file): + real_target = os.path.realpath(src_file) + try: + common = os.path.commonpath([real_source, real_target]) + except ValueError as exc: + raise FixtureSetupError(f"Fixture file symlink escapes root: {src_file!r}") from exc + if common != real_source: + raise FixtureSetupError(f"Fixture file symlink escapes root: {src_file!r}") + continue + shutil.copy2(src_file, os.path.join(dst_dir, fname)) + + def setup(self) -> None: + """Materialize workspace copy and initialize separate harness + local git baselines.""" + if self._is_setup: + return + self._ensure_layout_dirs() + + try: + self._run_harness_git(["init", "--bare", self._harness_git_dir]) + + if self.base_dir: + source_dir = ( + _ensure_within_root(self.base_dir, self.fixture_root) + if self.fixture_root + else os.path.realpath(self.base_dir) + ) + if not os.path.isdir(source_dir): + raise FixtureSetupError(f"Fixture directory does not exist: {self.base_dir!r}") + + if self.pinned_commit: + self._materialize_via_git_archive(source_dir, self.pinned_commit) + else: + self._copy_plain_fixture(source_dir) + elif self.pinned_commit: + raise FixtureSetupError("pinned_commit requires a base_dir git fixture") + + # Seed harness bare git dir with baseline commit (never touches workspace .git) + self._run_harness_git( + ["add", "-A"], + git_dir=self._harness_git_dir, + work_tree=self.workspace_path, + cwd=self.workspace_path, + ) + self._run_harness_git( + [ + "-c", "user.name=SkillOpt Agentic Scratch", + "-c", "user.email=agentic@skillopt.local", + "commit", "--allow-empty", "-m", "baseline", + ], + git_dir=self._harness_git_dir, + work_tree=self.workspace_path, + cwd=self.workspace_path, + ) + + # Give the scratch copy its own self-contained .git so agent commands like + # `git rev-parse HEAD` work without ever seeing the host repository. + self._run_harness_git(["-C", self.workspace_path, "init"]) + self._run_harness_git(["-C", self.workspace_path, "add", "-A"]) + self._run_harness_git([ + "-C", self.workspace_path, + "-c", "user.name=SkillOpt Agentic Scratch", + "-c", "user.email=agentic@skillopt.local", + "commit", "--allow-empty", "-m", "baseline", + ]) + self._is_setup = True + except FixtureSetupError: + raise + except (subprocess.SubprocessError, OSError) as exc: + raise FixtureSetupError(f"Workspace setup failed: {exc}") from exc + + def build_prompt_text(self, task: TaskRecord, skill: str = "", memory: str = "") -> str: + """Render a structured text prompt for the task, skill, and memory.""" + parts = [f"Task: {task.intent}"] + if task.context_excerpt: + parts.append(f"Context:\n{task.context_excerpt}") + if skill: + parts.append(f"Skill:\n{skill}") + if memory: + parts.append(f"Memory:\n{memory}") + return "\n\n".join(parts) + + def write_prompt_payload(self, task: TaskRecord, skill: str = "", memory: str = "") -> str: + """Write prompt JSON outside ``workspace_path`` so it stays out of ``get_diff()``.""" + self._ensure_layout_dirs() + prompt_file = os.path.join(self._prompt_dir, "task_payload.json") + payload = { + "task_id": task.id, + "intent": task.intent, + "context_excerpt": task.context_excerpt, + "skill": skill, + "memory": memory, + "prompt": self.build_prompt_text(task, skill=skill, memory=memory), + } + with open(prompt_file, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False, indent=2) + return prompt_file + + def run_subprocess( + self, + cmd: Sequence[str], + *, + stdin_text: Optional[str] = None, + timeout_sec: float = 60.0, + env: Optional[Dict[str, str]] = None, + max_output_bytes: Optional[int] = None, + ) -> Tuple[bool, int, str, str]: + """Execute a list-argv command inside ``workspace_path`` with bounded pipes. + + Returns ``(timed_out, returncode, stdout_text, stderr_text)``. + """ + if not self._is_setup: + self.setup() + effective_env = env if env is not None else self.build_isolated_env() + cap = self.max_output_bytes if max_output_bytes is None else max_output_bytes + stdin_b = stdin_text.encode("utf-8", errors="replace") if stdin_text is not None else None + timed_out, rc, out_b, err_b, out_trunc, err_trunc = _run_bounded_subprocess( + cmd, + cwd=self.workspace_path, + env=effective_env, + stdin_bytes=stdin_b, + timeout_sec=timeout_sec, + max_output_bytes=cap, + ) + out_txt = out_b.decode("utf-8", errors="replace") + err_txt = err_b.decode("utf-8", errors="replace") + if out_trunc: + out_txt += f"\n... [stdout truncated at {cap} bytes]" + if err_trunc: + err_txt += f"\n... [stderr truncated at {cap} bytes]" + return timed_out, rc, out_txt, err_txt + + def get_diff(self) -> str: + """Compute unified diff via harness ``harness.git`` without reading ``work/.git/config``.""" + if not self._is_setup or not os.path.isdir(self.workspace_path): + return "" + self._check_no_dotgit_pointers_or_alternates(self.workspace_path, allow_top_git_dir=True) + self._run_harness_git( + ["add", "-N", "."], + git_dir=self._harness_git_dir, + work_tree=self.workspace_path, + cwd=self.workspace_path, + ) + diff_b, truncated = self._run_harness_git( + ["diff", "--no-ext-diff", "--no-textconv", "HEAD"], + git_dir=self._harness_git_dir, + work_tree=self.workspace_path, + cwd=self.workspace_path, + max_output_bytes=self.max_diff_bytes, + ) + text = diff_b.decode("utf-8", errors="replace") + if truncated: + text += f"\n... [diff truncated at {self.max_diff_bytes} bytes]\n" + return text + + def cleanup(self) -> None: + """Remove temporary scratch directories (clearing read-only git object attrs).""" + if not os.path.exists(self._root_dir): + self._is_setup = False + return + + def _handle_readonly(func: Any, path: str, _exc: Any) -> None: + try: + os.chmod(path, stat.S_IWRITE | stat.S_IREAD) + func(path) + except OSError: + pass + + if sys.version_info >= (3, 12): + shutil.rmtree(self._root_dir, onexc=_handle_readonly) + else: + shutil.rmtree(self._root_dir, onerror=_handle_readonly) + # Ensure clean removal even if a transient handle delayed deletion + if os.path.exists(self._root_dir): + shutil.rmtree(self._root_dir, ignore_errors=True) + self._is_setup = False + + +class TaskVerifier: + """Deterministic task verifier acting as the primary acceptance gate.""" + + def __init__( + self, + test_cmd: Optional[Sequence[str]] = None, + expected_files: Optional[Dict[str, Optional[str]]] = None, + verifier_fn: Optional[Callable[[str], Tuple[bool, str]]] = None, + timeout_sec: float = 30.0, + max_output_bytes: int = 64 * 1024, + ) -> None: + if test_cmd is not None and isinstance(test_cmd, (str, bytes)): + raise ValueError("test_cmd must be a list/sequence of string arguments") + self.test_cmd = [str(part) for part in test_cmd] if test_cmd else None + self.expected_files = dict(expected_files) if expected_files else {} + self.verifier_fn = verifier_fn + self.timeout_sec = float(timeout_sec) + self.max_output_bytes = max(1024, int(max_output_bytes)) + + def verify( + self, + workspace_path: str, + *, + env: Optional[Dict[str, str]] = None, + ) -> Tuple[str, float, float, str, str]: + """Run checks against ``workspace_path``. + + Returns ``(outcome, hard, soft, rationale, inconclusive_reason)``. + """ + if not self.verifier_fn and not self.expected_files and not self.test_cmd: + return ( + OUTCOME_INCONCLUSIVE, + 0.0, + 0.0, + "No task verifier configured", + REASON_INSUFFICIENT_EVIDENCE, + ) + + if self.verifier_fn is not None: + try: + passed, msg = self.verifier_fn(workspace_path) + if passed: + return (OUTCOME_PASS, 1.0, 1.0, msg or "Verifier passed", "") + return (OUTCOME_FAIL, 0.0, 0.0, msg or "Verifier failed", "") + except Exception as exc: + return ( + OUTCOME_INCONCLUSIVE, + 0.0, + 0.0, + f"Verifier error: {exc}", + REASON_INFRA, + ) + + real_ws = os.path.realpath(workspace_path) + for rel_path, expected_substr in self.expected_files.items(): + full_path = os.path.realpath(os.path.join(real_ws, rel_path)) + try: + common = os.path.commonpath([real_ws, full_path]) + except ValueError: + common = "" + if common != real_ws: + return ( + OUTCOME_INCONCLUSIVE, + 0.0, + 0.0, + f"Expected file path escapes workspace: {rel_path}", + REASON_FIXTURE_FAILURE, + ) + if not os.path.exists(full_path): + return (OUTCOME_FAIL, 0.0, 0.0, f"Expected file missing: {rel_path}", "") + if expected_substr is not None: + try: + with open(full_path, "r", encoding="utf-8", errors="replace") as f: + content = f.read() + except OSError as exc: + return (OUTCOME_FAIL, 0.0, 0.0, f"Cannot read {rel_path}: {exc}", "") + if expected_substr not in content: + return ( + OUTCOME_FAIL, + 0.0, + 0.0, + f"Expected content missing in {rel_path}", + "", + ) + + if self.test_cmd: + try: + timed_out, rc, out_b, err_b, out_trunc, err_trunc = _run_bounded_subprocess( + self.test_cmd, + cwd=workspace_path, + env=env, + timeout_sec=self.timeout_sec, + max_output_bytes=self.max_output_bytes, + ) + if timed_out: + return ( + OUTCOME_INCONCLUSIVE, + 0.0, + 0.0, + f"Verifier command timed out after {self.timeout_sec}s", + REASON_TIMEOUT, + ) + if rc == 0: + return (OUTCOME_PASS, 1.0, 1.0, "Verifier test command passed", "") + out_s = (out_b + err_b).decode("utf-8", errors="replace").strip() + suffix = " [truncated]" if (out_trunc or err_trunc) else "" + return ( + OUTCOME_FAIL, + 0.0, + 0.0, + f"Verifier exited with {rc}: {out_s[:300]}{suffix}", + "", + ) + except (subprocess.SubprocessError, OSError, ValueError) as exc: + return ( + OUTCOME_INCONCLUSIVE, + 0.0, + 0.0, + f"Verifier infrastructure error: {exc}", + REASON_INFRA, + ) + + return (OUTCOME_PASS, 1.0, 1.0, "All configured verification checks passed", "") + + +def _invoke_runner( + runner: Callable[..., Any], + workspace_path: str, + env: Dict[str, str], + context: Dict[str, Any], +) -> Any: + """Call ``runner`` with ``(workspace_path, env, context)`` or ``(workspace_path, env)``.""" + try: + sig = inspect.signature(runner) + except (ValueError, TypeError): + return runner(workspace_path, env, context) + params = list(sig.parameters.values()) + accepts_varargs = any(p.kind == inspect.Parameter.VAR_POSITIONAL for p in params) + positional = [ + p for p in params + if p.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD) + ] + if accepts_varargs or len(positional) >= 3: + return runner(workspace_path, env, context) + if any(p.kind == inspect.Parameter.KEYWORD_ONLY and p.name == "context" for p in params): + return runner(workspace_path, env, context=context) + return runner(workspace_path, env) + + +class AgenticReplayBackend: + """Tiered opt-in agentic replay backend running in disposable scratch copies.""" + + name = "agentic" + + def __init__( + self, + config: Optional[AgenticReplayConfig] = None, + *, + base_backend: Optional[Any] = None, + agent_runner: Optional[Callable[..., Any]] = None, + default_verifier: Optional[TaskVerifier] = None, + ) -> None: + self.config = config or AgenticReplayConfig() + self.base_backend = base_backend + self.agent_runner = agent_runner + self.default_verifier = default_verifier + self._lock = threading.Lock() + self.replays_run = 0 + self.tokens_spent = 0 + self.preferences = "" + + def reset_run_budget(self) -> None: + """Reset per-night/run replay and token counters.""" + with self._lock: + self.replays_run = 0 + self.tokens_spent = 0 + + def tokens_used(self) -> int: + with self._lock: + local = self.tokens_spent + base = self.base_backend.tokens_used() if self.base_backend is not None else 0 + return local + base + + def attempt( + self, task: TaskRecord, skill: str, memory: str, sample_id: int = 0 + ) -> str: + if self.base_backend is not None: + return self.base_backend.attempt(task, skill, memory, sample_id=sample_id) + return "" + + def reflect( + self, + failures: List[Tuple[TaskRecord, ReplayResult]], + successes: List[Tuple[TaskRecord, ReplayResult]], + skill: str, + memory: str, + *, + edit_budget: int = 4, + evolve_skill: bool = True, + evolve_memory: bool = True, + **kwargs: Any, + ) -> List[EditRecord]: + """Delegate reflection to ``base_backend``, excluding inconclusive runs.""" + filtered_failures = [ + (t, r) for (t, r) in failures + if getattr(r, "outcome", "") != OUTCOME_INCONCLUSIVE + ] + filtered_successes = [ + (t, r) for (t, r) in successes + if getattr(r, "outcome", "") != OUTCOME_INCONCLUSIVE + ] + if self.base_backend is not None: + return self.base_backend.reflect( + filtered_failures, + filtered_successes, + skill, + memory, + edit_budget=edit_budget, + evolve_skill=evolve_skill, + evolve_memory=evolve_memory, + **kwargs, + ) + return [] + + def judge(self, task: TaskRecord, response: str) -> Tuple[float, float, str]: + if self.base_backend is not None: + return self.base_backend.judge(task, response) + return (0.0, 0.0, "No base backend judge available") + + def _reserve_budget(self) -> Optional[str]: + """Thread-safely reserve one replay slot, returning fail reason if exhausted.""" + with self._lock: + if ( + self.config.max_replays_per_run > 0 + and self.replays_run >= self.config.max_replays_per_run + ): + return "Max agentic replays budget reached for this run" + if ( + self.config.max_tokens > 0 + and self.tokens_spent >= self.config.max_tokens + ): + return "Max agentic token budget reached for this run" + self.replays_run += 1 + return None + + def _resolve_fixture_dir( + self, task: TaskRecord, fixture_dir: Optional[str] + ) -> Optional[str]: + """Resolve an optional fixture directory exclusively from ``fixture_root``.""" + requested = fixture_dir + if not requested and isinstance(task.judge, Mapping): + raw = task.judge.get("fixture_dir") + if isinstance(raw, str) and raw.strip(): + requested = raw.strip() + if requested: + return _ensure_within_root(requested, self.config.fixture_root) + if self.config.fixture_root: + return _ensure_within_root(self.config.fixture_root, self.config.fixture_root) + return None + + def replay_agentic_task( + self, + task: TaskRecord, + skill: str = "", + memory: str = "", + *, + verifier: Optional[TaskVerifier] = None, + fixture_dir: Optional[str] = None, + sample_id: int = 0, + ) -> ReplayResult: + """Run one opt-in agentic replay inside a disposable scratch copy.""" + del sample_id + t0 = time.monotonic() + + if self.config.enabled is not True: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_INFRA, + fail_reason="Agentic replay is disabled (set agentic_replay=True to opt in)", + task_type="agentic", + ) + + if self.agent_runner is None: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_INFRA, + fail_reason="No agent_runner configured (live CLI execution is disabled in this slice)", + task_type="agentic", + ) + + budget_err = self._reserve_budget() + if budget_err is not None: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_BUDGET_EXHAUSTED, + fail_reason=budget_err, + task_type="agentic", + ) + + try: + resolved_fixture = self._resolve_fixture_dir(task, fixture_dir) + except FixtureSetupError as exc: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_FIXTURE_FAILURE, + fail_reason=str(exc), + latency_ms=(time.monotonic() - t0) * 1000.0, + task_type="agentic", + ) + + ws = AgenticWorkspace( + base_dir=resolved_fixture, + pinned_commit=self.config.pinned_commit or None, + isolate_env=self.config.isolate_env, + fixture_root=self.config.fixture_root, + max_diff_bytes=self.config.max_diff_bytes, + max_output_bytes=self.config.max_output_bytes, + ) + try: + try: + ws.setup() + except FixtureSetupError as exc: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_FIXTURE_FAILURE, + fail_reason=str(exc), + latency_ms=(time.monotonic() - t0) * 1000.0, + task_type="agentic", + ) + + env = ws.build_isolated_env() + prompt_file = ws.write_prompt_payload(task, skill=skill, memory=memory) + prompt_text = ws.build_prompt_text(task, skill=skill, memory=memory) + prompt_ctx = { + "task": task, + "task_id": task.id, + "intent": task.intent, + "context_excerpt": task.context_excerpt, + "skill": skill, + "memory": memory, + "prompt": prompt_text, + "prompt_file": prompt_file, + "workspace": ws, + "timeout_sec": self.config.timeout_sec, + "max_turns": self.config.max_turns, + "max_output_bytes": self.config.max_output_bytes, + } + + response_text = "" + actions_taken = 0 + turns_taken = 0 + tokens_used_now = 0 + runner_rc = 0 + runner_outcome = "" + runner_reason = "" + runner_fail_msg = "" + + try: + res = _invoke_runner(self.agent_runner, ws.workspace_path, env, prompt_ctx) + if isinstance(res, dict): + response_text = str(res.get("response", "")) + actions_taken = int(res.get("actions", 1)) + turns_taken = int(res.get("turns", actions_taken)) + tokens_used_now = max(0, int(res.get("tokens", 0) or 0)) + raw_rc = res.get("returncode", res.get("exit_code", 0)) + runner_rc = 0 if raw_rc is None else int(raw_rc) + runner_outcome = str(res.get("outcome", "") or "") + runner_reason = str(res.get("inconclusive_reason", "") or "") + runner_fail_msg = str(res.get("fail_reason", "") or "") + elif isinstance(res, tuple) and len(res) >= 2: + response_text = str(res[0]) + actions_taken = int(res[1]) + turns_taken = actions_taken + else: + response_text = str(res) + actions_taken = 1 + turns_taken = 1 + except subprocess.TimeoutExpired: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_TIMEOUT, + fail_reason=f"Agent execution timed out after {self.config.timeout_sec}s", + latency_ms=(time.monotonic() - t0) * 1000.0, + task_type="agentic", + ) + except FixtureSetupError as exc: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_FIXTURE_FAILURE, + fail_reason=str(exc), + latency_ms=(time.monotonic() - t0) * 1000.0, + task_type="agentic", + ) + except Exception as exc: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_INFRA, + fail_reason=f"Agent runner infrastructure error: {exc}", + latency_ms=(time.monotonic() - t0) * 1000.0, + task_type="agentic", + ) + + with self._lock: + self.tokens_spent += tokens_used_now + over_token_budget = ( + self.config.max_tokens > 0 and self.tokens_spent > self.config.max_tokens + ) + + if runner_outcome == OUTCOME_INCONCLUSIVE: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + response=response_text[:4000], + tokens=tokens_used_now, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=runner_reason or REASON_INFRA, + fail_reason=runner_fail_msg or "Runner reported inconclusive outcome", + latency_ms=(time.monotonic() - t0) * 1000.0, + task_type="agentic", + ) + + if runner_rc != 0: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + response=response_text[:4000], + tokens=tokens_used_now, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_INFRA, + fail_reason=runner_fail_msg or f"Agent process exited with code {runner_rc}", + latency_ms=(time.monotonic() - t0) * 1000.0, + task_type="agentic", + ) + + if over_token_budget or ( + self.config.max_turns > 0 and turns_taken > self.config.max_turns + ): + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + response=response_text[:4000], + tokens=tokens_used_now, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_BUDGET_EXHAUSTED, + fail_reason=( + "Max agentic token budget exceeded" + if over_token_budget + else f"Turn budget exceeded ({turns_taken} > {self.config.max_turns})" + ), + latency_ms=(time.monotonic() - t0) * 1000.0, + task_type="agentic", + ) + + if actions_taken < self.config.min_decidable_actions: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + response=response_text[:4000], + tokens=tokens_used_now, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_INSUFFICIENT_EVIDENCE, + fail_reason=( + f"Actions taken ({actions_taken}) below decidable floor " + f"({self.config.min_decidable_actions})" + ), + latency_ms=(time.monotonic() - t0) * 1000.0, + task_type="agentic", + ) + + try: + diff = ws.get_diff() + except FixtureSetupError as exc: + return ReplayResult( + id=task.id, + hard=0.0, + soft=0.0, + response=response_text[:4000], + tokens=tokens_used_now, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_FIXTURE_FAILURE, + fail_reason=str(exc), + latency_ms=(time.monotonic() - t0) * 1000.0, + task_type="agentic", + ) + + active_verifier = ( + verifier + or self.default_verifier + or TaskVerifier( + timeout_sec=self.config.timeout_sec, + max_output_bytes=self.config.max_output_bytes, + ) + ) + outcome, hard, soft, rationale, inc_reason = active_verifier.verify( + ws.workspace_path, env=env + ) + latency_ms = (time.monotonic() - t0) * 1000.0 + fail_reason = "" if outcome == OUTCOME_PASS else rationale + + return ReplayResult( + id=task.id, + hard=hard, + soft=soft, + response=response_text[:4000], + fail_reason=fail_reason, + task_type="agentic", + judge_rationale=rationale, + tokens=tokens_used_now, + latency_ms=latency_ms, + outcome=outcome, + inconclusive_reason=inc_reason, + diff=diff, + ) + finally: + ws.cleanup() diff --git a/skillopt_sleep/backend.py b/skillopt_sleep/backend.py index 605a5716..8b2a4df2 100644 --- a/skillopt_sleep/backend.py +++ b/skillopt_sleep/backend.py @@ -29,7 +29,7 @@ import tempfile import threading from dataclasses import dataclass -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Mapping, Optional, Tuple from skillopt_sleep.types import EditRecord, ReplayResult, TaskRecord @@ -2751,6 +2751,7 @@ def get_backend( opencode_tool_replay: bool = False, azure_endpoint: str = "", project_dir: str = "", + agentic_config: Optional[Mapping[str, Any]] = None, ) -> Backend: n = (name or "mock").strip().lower() if n in {"pi", "pi_cli", "pi_coding_agent", "pi-coding-agent"}: @@ -2781,10 +2782,23 @@ def get_backend( project_dir or os.getcwd(), ".skillopt-sleep-handoff" ) return HandoffBackend(model=model, handoff_dir=hdir) + if n in {"agentic", "agentic_replay", "agentic-replay"}: + from skillopt_sleep.agentic_replay import ( + AgenticReplayBackend, + AgenticReplayConfig, + agentic_config_from_sleep_config, + ) + if isinstance(agentic_config, AgenticReplayConfig): + cfg_obj = agentic_config + elif agentic_config is not None: + cfg_obj = agentic_config_from_sleep_config(agentic_config) + else: + cfg_obj = AgenticReplayConfig() + return AgenticReplayBackend(cfg_obj) if n in {"mock", ""}: return MockBackend() raise ValueError( - f"Unknown backend: {name!r}. Supported backends are: 'azure', 'azure-responses', " + f"Unknown backend: {name!r}. Supported backends are: 'agentic', 'azure', 'azure-responses', " "'claude', 'codex', 'copilot', 'cursor', 'handoff', 'mock', 'opencode', 'pi'." ) @@ -2805,6 +2819,7 @@ def build_backend( azure_endpoint: str = "", preferences: str = "", project_dir: str = "", + agentic_config: Optional[Mapping[str, Any]] = None, ) -> Backend: """Build a single or dual backend. @@ -2813,11 +2828,16 @@ def build_backend( from (backend, model). ``preferences`` (free text) is attached so reflect uses it as a prior (set on the optimizer for dual backends). """ - has_split = any([optimizer_backend, optimizer_model, target_backend, target_model]) - if not has_split: - be = get_backend( - backend, - model=model, + def _call_get(resolved_name: str, resolved_model: str) -> Backend: + extra: Dict[str, Any] = {} + if ( + agentic_config is not None + and (resolved_name or "").strip().lower() in {"agentic", "agentic_replay", "agentic-replay"} + ): + extra["agentic_config"] = agentic_config + return get_backend( + resolved_name, + model=resolved_model, codex_path=codex_path, pi_path=pi_path, cursor_path=cursor_path, @@ -2825,19 +2845,16 @@ def build_backend( opencode_tool_replay=opencode_tool_replay, azure_endpoint=azure_endpoint, project_dir=project_dir, + **extra, ) + + has_split = any([optimizer_backend, optimizer_model, target_backend, target_model]) + if not has_split: + be = _call_get(backend, model) be.preferences = preferences return be - tgt = get_backend(target_backend or backend, model=target_model or model, - codex_path=codex_path, pi_path=pi_path, cursor_path=cursor_path, - opencode_path=opencode_path, opencode_tool_replay=opencode_tool_replay, - azure_endpoint=azure_endpoint, - project_dir=project_dir) - opt = get_backend(optimizer_backend or backend, model=optimizer_model or model, - codex_path=codex_path, pi_path=pi_path, cursor_path=cursor_path, - opencode_path=opencode_path, opencode_tool_replay=opencode_tool_replay, - azure_endpoint=azure_endpoint, - project_dir=project_dir) + tgt = _call_get(target_backend or backend, target_model or model) + opt = _call_get(optimizer_backend or backend, optimizer_model or model) opt.preferences = preferences # reflect runs on the optimizer dual = DualBackend(target=tgt, optimizer=opt) dual.preferences = preferences diff --git a/skillopt_sleep/config.py b/skillopt_sleep/config.py index 64814979..a4a316cd 100644 --- a/skillopt_sleep/config.py +++ b/skillopt_sleep/config.py @@ -68,7 +68,17 @@ "gate_metric": "mixed", # hard | soft | mixed (mixed best for tiny holdouts) "gate_mixed_weight": 0.5, "gate_no_regression": False, # reject any candidate that lowers a val-task score - "replay_mode": "mock", # report label; fresh-worktree replay is not implemented + "replay_mode": "mock", # report label; "mock" | "text" | "agentic" (issue #155) + # ── agentic replay tier (opt-in; issue #155) ─────────────────────────── + "agentic_replay": False, # strict boolean opt-in for agentic replay runner execution + "agentic_max_turns": 10, # turn/action cap per agentic replay + "agentic_max_tokens": 0, # cumulative token cap across replays (0 = unlimited) + "agentic_timeout_sec": 60.0, # wall-clock timeout in seconds + "agentic_max_replays": 5, # nightly budget cap on agentic replays + "agentic_isolate_env": True, # allowlist child environment + empty config dirs + "agentic_min_actions": 1, # decidable floor below which outcome is inconclusive + "agentic_fixture_root": "", # explicit root required to resolve directory fixtures + "agentic_min_coverage": 0.5, # minimum pairwise decidable val fraction for gate # ── dream + recall (opt-in; defaults reproduce the prior single-shot loop) ─ "dream_rollouts": 1, # >1 => multi-rollout contrastive reflection per task "dream_factor": 0, # >0 => add N synthetic variants of each task to the dream diff --git a/skillopt_sleep/consolidate.py b/skillopt_sleep/consolidate.py index cda388fb..eb5457a1 100644 --- a/skillopt_sleep/consolidate.py +++ b/skillopt_sleep/consolidate.py @@ -18,7 +18,7 @@ # research package, so this open-source tool stays decoupled from the paper code). from skillopt_sleep.gate import evaluate_gate, select_gate_score from skillopt_sleep.memory import apply_edits_detailed -from skillopt_sleep.replay import aggregate_scores, replay_batch +from skillopt_sleep.replay import aggregate_scores, pairwise_aggregate_scores, replay_batch from skillopt_sleep.types import ( EditRecord, ReplayResult, @@ -129,6 +129,35 @@ def _holdout_detail(pairs: List[Tuple[TaskRecord, ReplayResult]]) -> List[dict]: return out +def _has_inconclusive(pairs: List[Tuple[TaskRecord, ReplayResult]]) -> bool: + return any(getattr(r, "outcome", "") == "inconclusive" for _t, r in pairs) + + +def _score_val_comparison( + val_tasks: List[TaskRecord], + ref_pairs: List[Tuple[TaskRecord, ReplayResult]], + cand_pairs: List[Tuple[TaskRecord, ReplayResult]], + min_decidable_coverage: float, +) -> Tuple[float, float, float, float, int, int, float, bool]: + """Compute baseline/candidate scores, excluding inconclusive pairs identically on both sides.""" + if _has_inconclusive(ref_pairs) or _has_inconclusive(cand_pairs): + (b_h, b_s), (c_h, c_s), n_decidable, _n_tot, coverage = pairwise_aggregate_scores( + ref_pairs, cand_pairs + ) + n_total = len(val_tasks) if val_tasks else _n_tot + if val_tasks and _n_tot: + coverage = n_decidable / len(val_tasks) + abstained = bool(val_tasks) and ( + n_decidable == 0 or coverage < float(min_decidable_coverage) + ) + return b_h, b_s, c_h, c_s, n_decidable, n_total, coverage, abstained + + b_h, b_s = aggregate_scores(ref_pairs) + c_h, c_s = aggregate_scores(cand_pairs) + n_total = len(val_tasks) + return b_h, b_s, c_h, c_s, n_total, n_total, 1.0, False + + def _task_deltas( tasks: List[TaskRecord], baseline_pairs: List[Tuple[TaskRecord, ReplayResult]], @@ -143,17 +172,17 @@ def _task_deltas( for task in tasks: baseline = baseline_by_id.get(task.id) candidate = candidate_by_id.get(task.id) - baseline_score = ( - select_gate_score(baseline.hard, baseline.soft, metric, mixed_weight) - if baseline is not None - else None - ) - candidate_score = ( - select_gate_score(candidate.hard, candidate.soft, metric, mixed_weight) - if candidate is not None - else None - ) - if baseline_score is None or candidate_score is None: + if baseline is None or candidate is None: + baseline_score = ( + select_gate_score(baseline.hard, baseline.soft, metric, mixed_weight) + if baseline is not None + else None + ) + candidate_score = ( + select_gate_score(candidate.hard, candidate.soft, metric, mixed_weight) + if candidate is not None + else None + ) out.append({ "task_id": task.id, "tags": list(task.tags or []), @@ -163,6 +192,28 @@ def _task_deltas( "scores_are_finite": False, }) continue + + base_inc = getattr(baseline, "outcome", "") == "inconclusive" + cand_inc = getattr(candidate, "outcome", "") == "inconclusive" + if base_inc or cand_inc: + b_val = None if base_inc else _finite_score( + select_gate_score(baseline.hard, baseline.soft, metric, mixed_weight) + ) + c_val = None if cand_inc else _finite_score( + select_gate_score(candidate.hard, candidate.soft, metric, mixed_weight) + ) + out.append({ + "task_id": task.id, + "tags": list(task.tags or []), + "baseline_score": b_val, + "candidate_score": c_val, + "status": "inconclusive", + "scores_are_finite": True, + }) + continue + + baseline_score = select_gate_score(baseline.hard, baseline.soft, metric, mixed_weight) + candidate_score = select_gate_score(candidate.hard, candidate.soft, metric, mixed_weight) scores_are_finite = math.isfinite(baseline_score) and math.isfinite( candidate_score ) @@ -204,6 +255,7 @@ def consolidate( evolve_memory: bool = True, night: int = 1, budget_exhausted: Optional[Callable[[], bool]] = None, + min_decidable_coverage: float = 0.5, ) -> ConsolidationResult: """Run one consolidation epoch: reflect -> bounded edit -> gate. @@ -263,7 +315,7 @@ def _budget_spent(step: str) -> bool: else: evlog.set_phase(backend, "baseline_val") base_pairs = replay_batch(backend, val_tasks, skill, memory) - base_hard, base_soft = aggregate_scores(base_pairs) + base_hard, base_soft = aggregate_scores(base_pairs, exclude_inconclusive=True) holdout_detail = _holdout_detail(base_pairs) base_score = select_gate_score(base_hard, base_soft, gate_metric, gate_mixed_weight) if ev is not None: @@ -275,8 +327,14 @@ def _budget_spent(step: str) -> bool: # ── reflect over TRAIN-split failures/successes ─────────────────────── evlog.set_phase(backend, "train") train_pairs = replay_batch(backend, train_tasks, skill, memory) - failures = [(t, r) for (t, r) in train_pairs if r.hard < 1.0] - successes = [(t, r) for (t, r) in train_pairs if r.hard >= 1.0] + failures = [ + (t, r) for (t, r) in train_pairs + if r.hard < 1.0 and getattr(r, "outcome", "") != "inconclusive" + ] + successes = [ + (t, r) for (t, r) in train_pairs + if r.hard >= 1.0 and getattr(r, "outcome", "") != "inconclusive" + ] cand_skill, cand_memory = skill, memory all_applied: List[EditRecord] = [] @@ -317,7 +375,14 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str: trial_memory = new_doc if which == "memory" else cand_memory evlog.set_phase(backend, f"gate_trial:{which}") pairs = replay_batch(backend, val_tasks, trial_skill, trial_memory) - h, s = aggregate_scores(pairs) + b_h, b_s, h, s, n_decidable, _n_tot, coverage, below_floor = _score_val_comparison( + val_tasks, current_pairs, pairs, min_decidable_coverage + ) + trial_base_score = ( + select_gate_score(b_h, b_s, gate_metric, gate_mixed_weight) + if (_has_inconclusive(current_pairs) or _has_inconclusive(pairs)) + else base_score + ) cand_score = select_gate_score(h, s, gate_metric, gate_mixed_weight) task_deltas = _task_deltas( val_tasks, current_pairs, pairs, gate_metric, gate_mixed_weight @@ -326,21 +391,27 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str: gate_no_regression and any(row["status"] == "regressed" for row in task_deltas) ) - trial_base_score = base_score - improved = cand_score > base_score and not blocked_by_regression - gate_trials.append({ + improved = (not below_floor) and cand_score > trial_base_score and not blocked_by_regression + trial_record: dict = { "target": which, "baseline_score": _finite_score(trial_base_score), "candidate_score": _finite_score(cand_score), "accepted": improved, "blocked_by_regression": blocked_by_regression, "task_deltas": task_deltas, - }) + } + if below_floor or _has_inconclusive(current_pairs) or _has_inconclusive(pairs): + trial_record["abstained_inconclusive"] = below_floor + trial_record["decidable_coverage"] = coverage + trial_record["n_decidable"] = n_decidable + gate_trials.append(trial_record) if ev is not None: ev.log("gate", "trial", target=which, mode="gated", baseline_score=trial_base_score, cand_hard=h, cand_soft=s, cand_score=cand_score, accepted=improved, blocked_by_regression=blocked_by_regression, + abstained_inconclusive=below_floor, + decidable_coverage=coverage, task_deltas=task_deltas, n_edits=len(applied)) if improved: @@ -441,7 +512,11 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str: else: evlog.set_phase(backend, "final_val") final_pairs = replay_batch(backend, val_tasks, cand_skill, cand_memory) - final_hard, final_soft = aggregate_scores(final_pairs) + base_hard_cmp, base_soft_cmp, final_hard, final_soft, n_decidable, _n_tot, coverage, abstain_inconclusive = _score_val_comparison( + val_tasks, base_pairs, final_pairs, min_decidable_coverage + ) + base_hard = base_hard_cmp + base_soft = base_soft_cmp final_score = select_gate_score(final_hard, final_soft, gate_metric, gate_mixed_weight) base_gate_score = select_gate_score(base_hard, base_soft, gate_metric, gate_mixed_weight) final_task_deltas = _task_deltas( @@ -451,65 +526,74 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str: gate_no_regression and any(row["status"] == "regressed" for row in final_task_deltas) ) - if _HAVE_REPO_GATE: - gate = evaluate_gate( - candidate_skill=cand_skill, - cand_hard=final_hard, - current_skill=skill, - current_score=base_gate_score, - best_skill=skill, - best_score=base_gate_score, - best_step=night - 1, - global_step=night, - cand_soft=final_soft, - metric=gate_metric, - mixed_weight=gate_mixed_weight, - ) - action = gate.action - accepted = ( - bool(all_applied) - and final_score > base_gate_score - and not final_blocked_by_regression - ) - else: - action = "accept" if final_score > base_gate_score else "reject" - accepted = ( - bool(all_applied) - and final_score > base_gate_score - and not final_blocked_by_regression - ) - # The gate scores documents, not edit bookkeeping: when every proposed - # edit was dropped during the per-target trials, `all_applied` is empty - # and nothing changed, yet the score comparison can still yield an - # accept-flavoured action. Reporting that as "accept_new_best" while - # `accepted` is False makes the headline contradict the outcome. - if not accepted and action in {"accept", "accept_new_best"}: - action = "reject" - # The gate cannot certify an improvement it measured on the same tasks - # the edits were derived from -- that comparison cannot detect - # overfitting, which is exactly how a reward hack scores 1.0. Abstain - # rather than report a verdict the evidence does not support. Edits are - # still staged, so a human can review them. - if accepted and holdout_leaked: - action = "reject_unverified" + if abstain_inconclusive: + action = "abstain_inconclusive" accepted = False + else: + if _HAVE_REPO_GATE: + gate = evaluate_gate( + candidate_skill=cand_skill, + cand_hard=final_hard, + current_skill=skill, + current_score=base_gate_score, + best_skill=skill, + best_score=base_gate_score, + best_step=night - 1, + global_step=night, + cand_soft=final_soft, + metric=gate_metric, + mixed_weight=gate_mixed_weight, + ) + action = gate.action + accepted = ( + bool(all_applied) + and final_score > base_gate_score + and not final_blocked_by_regression + ) + else: + action = "accept" if final_score > base_gate_score else "reject" + accepted = ( + bool(all_applied) + and final_score > base_gate_score + and not final_blocked_by_regression + ) + # The gate scores documents, not edit bookkeeping: when every proposed + # edit was dropped during the per-target trials, `all_applied` is empty + # and nothing changed, yet the score comparison can still yield an + # accept-flavoured action. Reporting that as "accept_new_best" while + # `accepted` is False makes the headline contradict the outcome. + if not accepted and action in {"accept", "accept_new_best"}: + action = "reject" + # The gate cannot certify an improvement it measured on the same tasks + # the edits were derived from -- that comparison cannot detect + # overfitting, which is exactly how a reward hack scores 1.0. Abstain + # rather than report a verdict the evidence does not support. Edits are + # still staged, so a human can review them. + if accepted and holdout_leaked: + action = "reject_unverified" + accepted = False # A per-target trial can improve and tentatively apply an edit, while a - # later fresh final replay regresses. The returned documents already - # roll back in that case; keep the edit bookkeeping/report consistent - # by moving those tentative edits into the rejected set as well. + # later fresh final replay regresses or becomes inconclusive. The returned + # documents already roll back in that case; keep the edit bookkeeping/report + # consistent by moving those tentative edits into the rejected set as well. if not accepted and all_applied: for edit in all_applied: if edit not in all_rejected: all_rejected.append(edit) all_applied = [] - gate_trials.append({ + final_trial_record: dict = { "target": "final", "baseline_score": _finite_score(base_gate_score), "candidate_score": _finite_score(final_score), "accepted": accepted, "blocked_by_regression": final_blocked_by_regression, "task_deltas": final_task_deltas, - }) + } + if abstain_inconclusive or _has_inconclusive(base_pairs) or _has_inconclusive(final_pairs): + final_trial_record["abstained_inconclusive"] = abstain_inconclusive + final_trial_record["decidable_coverage"] = coverage + final_trial_record["n_decidable"] = n_decidable + gate_trials.append(final_trial_record) if ev is not None: w = max(0.0, min(1.0, float(gate_mixed_weight))) diff --git a/skillopt_sleep/cycle.py b/skillopt_sleep/cycle.py index b3711f7d..7229d61e 100644 --- a/skillopt_sleep/cycle.py +++ b/skillopt_sleep/cycle.py @@ -18,7 +18,7 @@ import time import unicodedata from dataclasses import dataclass -from typing import Dict, List, Optional, Sequence +from typing import Any, Dict, List, Optional, Sequence from skillopt_sleep import evidence from skillopt_sleep.backend import Backend, CursorBackendError, build_backend @@ -118,6 +118,22 @@ def _night_token_budget(cfg: SleepConfig) -> Budget: return Budget(max_tokens=limit or None) +def _agentic_config_payload(cfg: SleepConfig) -> dict[str, Any]: + """Extract agentic replay configuration keys from SleepConfig.""" + keys = ( + "agentic_replay", + "agentic_max_turns", + "agentic_max_tokens", + "agentic_timeout_sec", + "agentic_max_replays", + "agentic_isolate_env", + "agentic_min_actions", + "agentic_fixture_root", + "agentic_min_coverage", + ) + return {k: cfg.get(k) for k in keys if k in cfg.data} + + # ── Model-swap detection (F16) ─────────────────────────────── def _make_model_key(cfg: SleepConfig) -> str: """Stable string identifying the backend object(s) actually used. @@ -142,6 +158,7 @@ def _make_model_key(cfg: SleepConfig) -> str: opencode_tool_replay=cfg.get("opencode_tool_replay", False), azure_endpoint=cfg.get("azure_endpoint", ""), project_dir=cfg.get("invoked_project", "") or os.getcwd(), + agentic_config=_agentic_config_payload(cfg), ) except Exception: backend = str(cfg.get("backend", "mock") or "mock") @@ -179,6 +196,18 @@ def _make_backend_key(backend: Backend) -> str: return f"{name}::{model}" +def _reset_backend_run_budgets(backend: Backend) -> None: + """Reset per-night replay/token budgets on direct or split-role backends.""" + for resolved in ( + backend, + getattr(backend, "target", None), + getattr(backend, "optimizer", None), + ): + reset_fn = getattr(resolved, "reset_run_budget", None) + if callable(reset_fn): + reset_fn() + + def _claude_isolation_modes(backend: Backend, *, reset: bool = False) -> dict[str, list[str]]: """Keep per-night mode evidence for direct and split-role backends.""" roles = { @@ -741,9 +770,11 @@ def run_sleep_cycle( azure_endpoint=cfg.get("azure_endpoint", ""), preferences=cfg.get("preferences", ""), project_dir=project, + agentic_config=_agentic_config_payload(cfg), ) _check_model_change(cfg, state, backend) # F16: warn if model changed between nights _claude_isolation_modes(backend, reset=True) + _reset_backend_run_budgets(backend) night = state.begin_night(clock) started = _now_iso(clock) backend.preferences = cfg.get("preferences", "") @@ -805,6 +836,9 @@ def budget_exhausted() -> bool: cycle_config["opencode_tool_replay"] = ( cfg.get("opencode_tool_replay", False) is True ) + cycle_config["agentic_replay"] = ( + cfg.get("agentic_replay", False) is True + ) ev.log("cycle", "start", night=night, project=project, backend=backend.name, model=cfg.get("model", ""), config=cycle_config) @@ -981,6 +1015,7 @@ def budget_exhausted() -> bool: evolve_memory=cfg.get("evolve_memory", True), night=night, budget_exhausted=budget_exhausted, + min_decidable_coverage=float(cfg.get("agentic_min_coverage", 0.5) or 0.5), ) except CursorBackendError: _discard_unstaged_evidence(staging_dir_pre) @@ -1027,7 +1062,7 @@ def budget_exhausted() -> bool: final_skill = result.new_skill if result.accepted else skill final_memory = result.new_memory if result.accepted else memory test_pairs = replay_batch(backend, test_tasks, final_skill, final_memory) - test_hard, test_soft = aggregate_scores(test_pairs) + test_hard, test_soft = aggregate_scores(test_pairs, exclude_inconclusive=True) ev.log("test", "held_out_score", night=night, n_test=len(test_tasks), hard=round(test_hard, 4), soft=round(test_soft, 4), @@ -1086,6 +1121,9 @@ def budget_exhausted() -> bool: evolve_skill=cfg.get("evolve_skill", True), night=night, budget_exhausted=budget_exhausted, + min_decidable_coverage=float( + cfg.get("agentic_min_coverage", 0.5) or 0.5 + ), ) except CursorBackendError: _discard_unstaged_evidence(staging_dir_pre) diff --git a/skillopt_sleep/dream.py b/skillopt_sleep/dream.py index 4dfdbc85..8afe2783 100644 --- a/skillopt_sleep/dream.py +++ b/skillopt_sleep/dream.py @@ -137,6 +137,7 @@ def dream_consolidate( evolve_memory: bool = True, night: int = 1, budget_exhausted: Optional[Callable[[], bool]] = None, + min_decidable_coverage: float = 0.5, ) -> ConsolidationResult: """Recall similar past experience + dream synthetic variants, then run one gated consolidation epoch over the enlarged training pool. @@ -170,4 +171,5 @@ def dream_consolidate( rollouts_k=dream_rollouts, evolve_skill=evolve_skill, evolve_memory=evolve_memory, night=night, budget_exhausted=budget_exhausted, + min_decidable_coverage=min_decidable_coverage, ) diff --git a/skillopt_sleep/replay.py b/skillopt_sleep/replay.py index 6eba0b86..13087875 100644 --- a/skillopt_sleep/replay.py +++ b/skillopt_sleep/replay.py @@ -32,6 +32,8 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str, """``sample_id`` distinguishes repeated dream rollouts of the same (task, skill, memory) in the attempt cache — without it all K rollouts collapse to one cached response and the contrastive signal is always 0.""" + if hasattr(backend, "replay_agentic_task"): + return backend.replay_agentic_task(task, skill, memory, sample_id=sample_id) import time tools = _required_tools(task) tools_called: List[str] = [] @@ -166,14 +168,61 @@ def replay_batch( return results -def aggregate_scores(pairs: List[Tuple[TaskRecord, ReplayResult]]) -> Tuple[float, float]: +def aggregate_scores( + pairs: List[Tuple[TaskRecord, ReplayResult]], + *, + exclude_inconclusive: bool = False, +) -> Tuple[float, float]: if not pairs: return 0.0, 0.0 + if exclude_inconclusive: + valid = [r for _t, r in pairs if getattr(r, "outcome", "") != "inconclusive"] + if not valid: + return 0.0, 0.0 + hard = sum(r.hard for r in valid) / len(valid) + soft = sum(r.soft for r in valid) / len(valid) + return hard, soft hard = sum(r.hard for _t, r in pairs) / len(pairs) soft = sum(r.soft for _t, r in pairs) / len(pairs) return hard, soft +def pairwise_aggregate_scores( + base_pairs: List[Tuple[TaskRecord, ReplayResult]], + cand_pairs: List[Tuple[TaskRecord, ReplayResult]], +) -> Tuple[Tuple[float, float], Tuple[float, float], int, int, float]: + """Compare baseline and candidate over the exact intersection of decidable task IDs. + + Tasks marked ``outcome == "inconclusive"`` (or missing) on either side are + excluded pairwise so budget exhaustion or infra failures never bias one side. + Returns ``((base_hard, base_soft), (cand_hard, cand_soft), n_decidable, n_total, coverage)``. + """ + all_ids = list(dict.fromkeys([t.id for t, _r in base_pairs] + [t.id for t, _r in cand_pairs])) + n_total = len(all_ids) + if n_total == 0: + return (0.0, 0.0), (0.0, 0.0), 0, 0, 1.0 + + base_by_id = {t.id: r for t, r in base_pairs} + cand_by_id = {t.id: r for t, r in cand_pairs} + shared_ids = [ + tid for tid in all_ids + if base_by_id.get(tid) is not None + and cand_by_id.get(tid) is not None + and getattr(base_by_id[tid], "outcome", "") != "inconclusive" + and getattr(cand_by_id[tid], "outcome", "") != "inconclusive" + ] + n_decidable = len(shared_ids) + coverage = n_decidable / n_total + if n_decidable == 0: + return (0.0, 0.0), (0.0, 0.0), 0, n_total, coverage + + base_hard = sum(base_by_id[tid].hard for tid in shared_ids) / n_decidable + base_soft = sum(base_by_id[tid].soft for tid in shared_ids) / n_decidable + cand_hard = sum(cand_by_id[tid].hard for tid in shared_ids) / n_decidable + cand_soft = sum(cand_by_id[tid].soft for tid in shared_ids) / n_decidable + return (base_hard, base_soft), (cand_hard, cand_soft), n_decidable, n_total, coverage + + def aggregate_cost(pairs: List[Tuple[TaskRecord, ReplayResult]]) -> Tuple[float, float]: """Mean (tokens, latency_ms) per task — the cost objectives.""" if not pairs: diff --git a/skillopt_sleep/types.py b/skillopt_sleep/types.py index d5c053d5..7c4fe61e 100644 --- a/skillopt_sleep/types.py +++ b/skillopt_sleep/types.py @@ -115,6 +115,16 @@ class ReplayResult: # evidence for audit/debugging; they may contain regexes or other grader # implementation details and must not be used as learning context. optimizer_feedback: str = "" + # Three-valued outcome: pass | fail | inconclusive (issue #155). + # Timeouts, budget exhaustion, or environment failures must not be counted + # as evidence that a candidate skill failed. + outcome: str = "" + inconclusive_reason: str = "" # infra | timeout | budget_exhausted | fixture_failure | insufficient_evidence + diff: str = "" # unified diff produced in agentic replay + + def __post_init__(self) -> None: + if not self.outcome: + self.outcome = "pass" if self.hard >= 1.0 else "fail" def to_dict(self) -> Dict[str, Any]: return asdict(self) diff --git a/tests/test_agentic_replay.py b/tests/test_agentic_replay.py new file mode 100644 index 00000000..1ea36578 --- /dev/null +++ b/tests/test_agentic_replay.py @@ -0,0 +1,803 @@ +"""Tests for AgenticReplayBackend and workspace lifecycle slice (Issue #155).""" +from __future__ import annotations + +import os +import subprocess +import sys +import tempfile +import unittest +from concurrent.futures import ThreadPoolExecutor +from unittest import mock + +from skillopt_sleep.agentic_replay import ( + OUTCOME_FAIL, + OUTCOME_INCONCLUSIVE, + OUTCOME_PASS, + REASON_BUDGET_EXHAUSTED, + REASON_FIXTURE_FAILURE, + REASON_INFRA, + REASON_INSUFFICIENT_EVIDENCE, + REASON_TIMEOUT, + AgenticReplayBackend, + AgenticReplayConfig, + AgenticWorkspace, + FixtureSetupError, + TaskVerifier, +) +from skillopt_sleep.backend import MockBackend, build_backend, get_backend +from skillopt_sleep.config import load_config +from skillopt_sleep.consolidate import consolidate +from skillopt_sleep.cycle import run_sleep_cycle +from skillopt_sleep.replay import aggregate_scores, pairwise_aggregate_scores, replay_batch, replay_one +from skillopt_sleep.types import EditRecord, ReplayResult, TaskRecord + + +def _init_git_repo(repo_dir: str, files: dict[str, str] | None = None) -> str: + """Create a small standalone git repository and return its HEAD commit SHA.""" + os.makedirs(repo_dir, exist_ok=True) + for rel, content in (files or {"app.txt": "baseline\n"}).items(): + full = os.path.join(repo_dir, rel) + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, "w", encoding="utf-8") as f: + f.write(content) + git_cfg = ["-c", "user.name=Test", "-c", "user.email=t@example.com", "-c", "commit.gpgsign=false"] + subprocess.run(["git", "init", "-b", "main", repo_dir], check=True, capture_output=True) + subprocess.run(["git", "-C", repo_dir, "add", "."], check=True, capture_output=True) + subprocess.run(["git", "-C", repo_dir, *git_cfg, "commit", "-m", "initial"], check=True, capture_output=True) + res = subprocess.run(["git", "-C", repo_dir, "rev-parse", "HEAD"], check=True, capture_output=True, text=True) + return res.stdout.strip() + + +def _write_fake_agent_script(path: str, body: str) -> str: + with open(path, "w", encoding="utf-8") as f: + f.write(body) + return path + + +def _pid_alive(pid: int) -> bool: + if pid <= 0: + return False + if os.name == "nt": + res = subprocess.run(["tasklist", "/FI", f"PID eq {pid}", "/NH"], capture_output=True, text=True) + return str(pid) in res.stdout + try: + os.kill(pid, 0) + return True + except OSError: + return False + + +class TestReplayResultOutcomes(unittest.TestCase): + def test_default_pass_fail_inference(self): + passed = ReplayResult(id="t1", hard=1.0) + self.assertEqual(passed.outcome, OUTCOME_PASS) + + failed = ReplayResult(id="t2", hard=0.0) + self.assertEqual(failed.outcome, OUTCOME_FAIL) + + mixed = ReplayResult(id="t3", hard=0.5) + self.assertEqual(mixed.outcome, OUTCOME_FAIL) + + def test_explicit_inconclusive_outcome(self): + res = ReplayResult( + id="t1", + hard=0.0, + outcome=OUTCOME_INCONCLUSIVE, + inconclusive_reason=REASON_TIMEOUT, + diff="--- a/file\n+++ b/file\n", + ) + self.assertEqual(res.outcome, OUTCOME_INCONCLUSIVE) + self.assertEqual(res.inconclusive_reason, REASON_TIMEOUT) + self.assertIn("--- a/file", res.diff) + + d = res.to_dict() + self.assertEqual(d["outcome"], OUTCOME_INCONCLUSIVE) + self.assertEqual(d["inconclusive_reason"], REASON_TIMEOUT) + self.assertIn("--- a/file", d["diff"]) + + +class TestAgenticOptInAndConfigWiring(unittest.TestCase): + def test_disabled_by_default_blocks_execution(self): + called = False + + def _runner(ws_path, env, ctx): + nonlocal called + called = True + return {"response": "ran", "actions": 1} + + for flag in (False, "true", 1): + cfg = AgenticReplayConfig(enabled=flag) # type: ignore[arg-type] + self.assertFalse(cfg.enabled) + backend = AgenticReplayBackend( + cfg, + agent_runner=_runner, + default_verifier=TaskVerifier(verifier_fn=lambda _p: (True, "ok")), + ) + res = backend.replay_agentic_task(TaskRecord(id="t1", project="/p", intent="do task")) + self.assertFalse(called) + self.assertEqual(res.outcome, OUTCOME_INCONCLUSIVE) + self.assertEqual(res.inconclusive_reason, REASON_INFRA) + + def test_config_keys_wired_through_load_config_build_backend_and_cycle(self): + with tempfile.TemporaryDirectory() as tmp: + fixture_root = os.path.join(tmp, "fixtures") + os.makedirs(fixture_root) + cfg = load_config( + backend="agentic", + agentic_replay=True, + agentic_max_turns=3, + agentic_max_tokens=1500, + agentic_timeout_sec=12.5, + agentic_max_replays=7, + agentic_isolate_env=False, + agentic_min_actions=2, + agentic_fixture_root=fixture_root, + agentic_min_coverage=0.75, + ) + payload = {k: cfg.get(k) for k in cfg.data if k.startswith("agentic_")} + be_alias = get_backend("agentic_replay", agentic_config=payload) + self.assertEqual(be_alias.name, "agentic") + self.assertTrue(be_alias.config.enabled) + + be = build_backend(backend="agentic", agentic_config=payload) + self.assertIsInstance(be, AgenticReplayBackend) + # Disabled runner invocation via replay_one fails closed cleanly + disabled_be = get_backend("agentic") + res_dis = replay_one(disabled_be, TaskRecord(id="r1", project="/p", intent="hi"), "", "") + self.assertEqual(res_dis.outcome, OUTCOME_INCONCLUSIVE) + self.assertEqual(res_dis.inconclusive_reason, REASON_INFRA) + self.assertTrue(be.config.enabled) + self.assertEqual(be.config.max_turns, 3) + self.assertEqual(be.config.max_tokens, 1500) + self.assertEqual(be.config.timeout_sec, 12.5) + self.assertEqual(be.config.max_replays_per_run, 7) + self.assertFalse(be.config.isolate_env) + self.assertEqual(be.config.min_decidable_actions, 2) + self.assertEqual(be.config.fixture_root, fixture_root) + self.assertEqual(be.config.min_coverage, 0.75) + + # Per-night reset clears replays_run at start of run_sleep_cycle + be.replays_run = 6 + be.tokens_spent = 999 + proj = os.path.join(tmp, "proj") + os.makedirs(proj) + run_cfg = load_config( + projects="invoked", + invoked_project=proj, + state_dir=os.path.join(tmp, "state"), + claude_home=os.path.join(tmp, "claude"), + evidence_log=False, + evolve_skill=False, + evolve_memory=False, + ) + run_sleep_cycle(run_cfg, backend=be, seed_tasks=[], dry_run=True) + self.assertEqual(be.replays_run, 0) + self.assertEqual(be.tokens_spent, 0) + + # Verify run_sleep_cycle itself constructs AgenticReplayBackend from SleepConfig + built_backends = [] + real_builder = build_backend + + def _capture_builder(**kwargs): + res_be = real_builder(**kwargs) + built_backends.append(res_be) + return res_be + + agentic_cfg = load_config( + backend="agentic", + agentic_replay=True, + agentic_max_turns=4, + agentic_max_replays=9, + projects="invoked", + invoked_project=proj, + state_dir=os.path.join(tmp, "state2"), + claude_home=os.path.join(tmp, "claude"), + evidence_log=False, + evolve_skill=False, + evolve_memory=False, + ) + with mock.patch("skillopt_sleep.cycle.build_backend", side_effect=_capture_builder): + run_sleep_cycle(agentic_cfg, seed_tasks=[], dry_run=True) + self.assertGreaterEqual(len(built_backends), 1) + self.assertIsInstance(built_backends[-1], AgenticReplayBackend) + self.assertTrue(built_backends[-1].config.enabled) + self.assertEqual(built_backends[-1].config.max_turns, 4) + self.assertEqual(built_backends[-1].config.max_replays_per_run, 9) + + def test_lazy_import_of_agentic_backend_in_package_init(self): + script = ( + "import sys, skillopt_sleep; " + "assert 'skillopt_sleep.agentic_replay' not in sys.modules; " + "from skillopt_sleep import AgenticReplayBackend; " + "assert AgenticReplayBackend.name == 'agentic'" + ) + res = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True) + self.assertEqual(res.returncode, 0, res.stderr) + + +class TestAgenticWorkspaceAndFakeAgentSubprocess(unittest.TestCase): + def test_workspace_scratch_copy_env_allowlist_and_home_outside_diff(self): + sensitive = { + "TEST_SECRET_KEY": "supersecret", + "AWS_ACCESS_KEY_ID": "AKIAEXAMPLE", + "GITHUB_TOKEN": "ghp_example", + "GOOGLE_APPLICATION_CREDENTIALS": "/tmp/creds.json", + "KUBECONFIG": "/tmp/kubeconfig", + "DATABASE_URL": "postgres://user:pass@host/db", + "GIT_ASKPASS": "/tmp/askpass.sh", + "DOCKER_HOST": "tcp://127.0.0.1:2375", + "DBUS_SESSION_BUS_ADDRESS": "unix:path=/run/user/1000/bus", + "IDENTITY_HEADER": "secret-hdr", + "IDENTITY_ENDPOINT": "http://169.254.169.254", + "CLAUDE_CONFIG_DIR": "/real/claude", + "CODEX_HOME": "/real/codex", + "HTTP_PROXY": "http://corp:8080", + "https_proxy": "https://corp:8080", + "no_proxy": "localhost", + } + with mock.patch.dict(os.environ, sensitive): + with AgenticWorkspace(isolate_env=True) as ws: + self.assertTrue(os.path.isdir(ws.workspace_path)) + self.assertFalse( + os.path.commonpath([ + os.path.realpath(ws.workspace_path), + os.path.realpath(ws.home_dir), + ]) == os.path.realpath(ws.workspace_path), + "HOME must sit outside diffed workspace_path", + ) + git_res = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=ws.workspace_path, + capture_output=True, + text=True, + ) + self.assertEqual(git_res.returncode, 0) + self.assertGreaterEqual(len(git_res.stdout.strip()), 7) + + env = ws.build_isolated_env() + for forbidden in ( + "TEST_SECRET_KEY", "AWS_ACCESS_KEY_ID", "GITHUB_TOKEN", + "GOOGLE_APPLICATION_CREDENTIALS", "KUBECONFIG", "DATABASE_URL", + "DOCKER_HOST", "DBUS_SESSION_BUS_ADDRESS", + "IDENTITY_HEADER", "IDENTITY_ENDPOINT", + "HTTP_PROXY", "https_proxy", "no_proxy", + ): + self.assertNotIn(forbidden, env) + + self.assertEqual(env["GIT_ASKPASS"], "") + self.assertEqual(env["GIT_CONFIG_NOSYSTEM"], "1") + self.assertTrue(os.path.isfile(env["GIT_CONFIG_GLOBAL"])) + self.assertEqual(os.path.getsize(env["GIT_CONFIG_GLOBAL"]), 0) + self.assertEqual(env["HOME"], ws.home_dir) + self.assertEqual(os.listdir(env["CLAUDE_CONFIG_DIR"]), []) + self.assertEqual(os.listdir(env["CODEX_HOME"]), []) + self.assertNotIn(".sandbox_home", env["HOME"]) + + # Verify fake agent binary sees allowlisted env via subprocess and writes to HOME without dirtying diff + script_dir = tempfile.mkdtemp(prefix="skillopt_fake_") + try: + script_path = _write_fake_agent_script( + os.path.join(script_dir, "agent.py"), + ( + "import json, os\n" + "assert 'KUBECONFIG' not in os.environ\n" + "assert 'DATABASE_URL' not in os.environ\n" + "assert 'https_proxy' not in os.environ\n" + "with open(os.path.join(os.environ['HOME'], 'cache.txt'), 'w') as hf:\n" + " hf.write('temp-home-artifact\\n')\n" + "with open('sample.txt', 'w') as wf:\n" + " wf.write('hello world\\n')\n" + "print(json.dumps({'home': os.environ['HOME']}))\n" + ), + ) + timed_out, rc, out, err = ws.run_subprocess([sys.executable, script_path], env=env) + self.assertFalse(timed_out, err) + self.assertEqual(rc, 0, err) + finally: + for name in os.listdir(script_dir): + os.remove(os.path.join(script_dir, name)) + os.rmdir(script_dir) + + diff = ws.get_diff() + self.assertIn("sample.txt", diff) + self.assertIn("+hello world", diff) + self.assertNotIn("cache.txt", diff) + self.assertNotIn("temp-home-artifact", diff) + + self.assertFalse(os.path.exists(ws.workspace_path)) + + def test_fake_agent_worktree_fixture_cannot_touch_real_repo(self): + with tempfile.TemporaryDirectory() as tmp: + fixture_root = os.path.join(tmp, "fixtures") + real_repo = os.path.join(fixture_root, "real_repo") + wt_dir = os.path.join(fixture_root, "wt") + head_sha = _init_git_repo(real_repo, {"tracked.txt": "v1\n"}) + subprocess.run( + ["git", "-C", real_repo, "worktree", "add", "--detach", wt_dir, head_sha], + check=True, + capture_output=True, + ) + hook_sentinel = os.path.join(real_repo, ".git", "hooks", "pwn_marker") + fake_script = _write_fake_agent_script( + os.path.join(tmp, "wt_agent.py"), + ( + "import os, subprocess, sys\n" + "common = subprocess.check_output(['git', 'rev-parse', '--git-common-dir'], text=True).strip()\n" + "main_head = subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip()\n" + "with open('edited.txt', 'w') as f:\n" + " f.write('updated\\n')\n" + "subprocess.run(['git', 'add', 'edited.txt'], check=True)\n" + "subprocess.run(['git', '-c', 'user.name=A', '-c', 'user.email=a@a', 'commit', '-m', 'agent'], check=True)\n" + "new_sha = subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip()\n" + "subprocess.run(['git', 'update-ref', 'refs/heads/main', new_sha], check=True)\n" + "with open(os.path.join(common, 'hooks', 'pwn_marker'), 'w') as hf:\n" + " hf.write('touched')\n" + ), + ) + + # 1. Unpinned worktree fixture must be rejected because .git is a linked file + be_unpinned = AgenticReplayBackend( + AgenticReplayConfig(enabled=True, fixture_root=fixture_root), + agent_runner=lambda _ws, _env: {"response": "ok", "actions": 1}, + default_verifier=TaskVerifier(verifier_fn=lambda _p: (True, "ok")), + ) + unpinned_res = be_unpinned.replay_agentic_task( + TaskRecord(id="wt-unpinned", project=real_repo, intent="edit"), + fixture_dir="wt", + ) + self.assertEqual(unpinned_res.outcome, OUTCOME_INCONCLUSIVE) + self.assertEqual(unpinned_res.inconclusive_reason, REASON_FIXTURE_FAILURE) + + # 2. Pinned worktree fixture materializes without .git pointer and isolates real_repo + def _run_wt(ws_path, env, ctx): + ws = ctx["workspace"] + to, rc, out, err = ws.run_subprocess([sys.executable, fake_script], env=env) + return {"response": out, "actions": 1, "returncode": rc} + + be_pinned = AgenticReplayBackend( + AgenticReplayConfig(enabled=True, fixture_root=fixture_root, pinned_commit=head_sha), + agent_runner=_run_wt, + default_verifier=TaskVerifier(expected_files={"edited.txt": "updated"}), + ) + pinned_res = be_pinned.replay_agentic_task( + TaskRecord(id="wt-pinned", project=real_repo, intent="edit"), + fixture_dir="wt", + ) + self.assertEqual(pinned_res.outcome, OUTCOME_PASS, pinned_res.fail_reason) + self.assertIn("edited.txt", pinned_res.diff) + + # Real repository's refs/heads/main and hooks dir are untouched + real_main = subprocess.run( + ["git", "-C", real_repo, "rev-parse", "refs/heads/main"], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + self.assertEqual(real_main, head_sha) + self.assertFalse(os.path.exists(hook_sentinel)) + subprocess.run(["git", "-C", real_repo, "worktree", "remove", "--force", wt_dir], capture_output=True) + + def test_alternates_rejected(self): + with tempfile.TemporaryDirectory() as tmp: + repo = os.path.join(tmp, "repo") + sha = _init_git_repo(repo) + alt_file = os.path.join(repo, ".git", "objects", "info", "alternates") + os.makedirs(os.path.dirname(alt_file), exist_ok=True) + with open(alt_file, "w", encoding="utf-8") as f: + f.write("/some/other/objects\n") + + with self.assertRaises(FixtureSetupError): + AgenticWorkspace(base_dir=repo, fixture_root=tmp).setup() + with self.assertRaises(FixtureSetupError): + AgenticWorkspace(base_dir=repo, pinned_commit=sha, fixture_root=tmp).setup() + + def test_fake_agent_git_config_diff_external_and_host_gpgsign(self): + with tempfile.TemporaryDirectory() as tmp: + broken_global_cfg = os.path.join(tmp, "host_gitconfig") + with open(broken_global_cfg, "w", encoding="utf-8") as f: + f.write("[commit]\n\tgpgsign = true\n[gpg]\n\tprogram = nonexistent-gpg-bin\n") + + hijack_marker = os.path.join(tmp, "DIFF_EXTERNAL_EXECUTED") + fake_agent = _write_fake_agent_script( + os.path.join(tmp, "hijack_agent.py"), + ( + "import os, sys\n" + f"marker = {hijack_marker!r}\n" + "cfg_path = os.path.join('.git', 'config')\n" + "with open(cfg_path, 'a', encoding='utf-8') as f:\n" + " f.write(f\"\\n[diff]\\n\\texternal = {sys.executable} -c \\\"open({marker!r}, 'w').write('pwnd')\\\"\\n\")\n" + "with open('feature.txt', 'w', encoding='utf-8') as f:\n" + " f.write('real feature\\n')\n" + ), + ) + + with mock.patch.dict(os.environ, {"GIT_CONFIG_GLOBAL": broken_global_cfg}): + be = AgenticReplayBackend( + AgenticReplayConfig(enabled=True), + agent_runner=lambda _ws, env, ctx: ( + lambda res: {"response": res[2], "actions": 1, "returncode": res[1]} + )(ctx["workspace"].run_subprocess([sys.executable, fake_agent], env=env)), + default_verifier=TaskVerifier(expected_files={"feature.txt": "real feature"}), + ) + res = be.replay_agentic_task(TaskRecord(id="git-cfg", project=tmp, intent="add feature")) + + self.assertEqual(res.outcome, OUTCOME_PASS, res.fail_reason) + self.assertIn("+real feature", res.diff) + self.assertFalse(os.path.exists(hijack_marker)) + + def test_fake_agent_mined_intent_flags_and_fixture_root_containment(self): + with tempfile.TemporaryDirectory() as tmp: + fixture_root = os.path.join(tmp, "fixtures") + allowed_dir = os.path.join(fixture_root, "proj") + outside_dir = os.path.join(tmp, "outside_secret") + os.makedirs(allowed_dir) + os.makedirs(outside_dir) + with open(os.path.join(allowed_dir, "allowed.txt"), "w", encoding="utf-8") as f: + f.write("ok\n") + with open(os.path.join(outside_dir, "secret.txt"), "w", encoding="utf-8") as f: + f.write("topsecret\n") + + fake_agent = _write_fake_agent_script( + os.path.join(tmp, "prompt_reader.py"), + ( + "import json, os, sys\n" + "stdin_intent = sys.stdin.read()\n" + "with open('seen_intent.txt', 'w', encoding='utf-8') as f:\n" + " f.write(stdin_intent)\n" + "payload_idx = sys.argv.index('--payload') + 1\n" + "payload = json.load(open(sys.argv[payload_idx], encoding='utf-8'))\n" + "with open('seen_skill.txt', 'w', encoding='utf-8') as f:\n" + " f.write(payload['skill'] + '|' + payload['memory'])\n" + ), + ) + + # 1. Task with flag-like intent ("--help") & context_excerpt pointing at existing outside_dir: + # outside_dir must NOT be treated as a fixture! Intent and skill must arrive via stdin/file. + def _runner(_ws_path, env, ctx): + ws = ctx["workspace"] + to, rc, out, err = ws.run_subprocess( + [sys.executable, fake_agent, "--payload", ctx["prompt_file"]], + stdin_text=ctx["intent"], + env=env, + ) + self.assertFalse(os.path.exists(os.path.join(ws.workspace_path, "secret.txt"))) + return {"response": out or "done", "actions": 1, "returncode": rc} + + be = AgenticReplayBackend( + AgenticReplayConfig(enabled=True, fixture_root=fixture_root), + agent_runner=_runner, + default_verifier=TaskVerifier(expected_files={ + "allowed.txt": "ok", + "seen_intent.txt": "--help --malicious", + "seen_skill.txt": "# Skill Body|# Memory Body", + }), + ) + task = TaskRecord( + id="flag-intent", + project=tmp, + intent="--help --malicious", + context_excerpt=outside_dir, + ) + res = be.replay_agentic_task( + task, + skill="# Skill Body", + memory="# Memory Body", + fixture_dir="proj", + ) + self.assertEqual(res.outcome, OUTCOME_PASS, res.fail_reason) + + # 2. Explicit fixture_dir outside fixture_root fails closed with REASON_FIXTURE_FAILURE + escape_res = be.replay_agentic_task( + task, + fixture_dir=os.path.join("..", "outside_secret"), + ) + self.assertEqual(escape_res.outcome, OUTCOME_INCONCLUSIVE) + self.assertEqual(escape_res.inconclusive_reason, REASON_FIXTURE_FAILURE) + + def test_timeout_kills_process_tree_and_caps_output_and_diff(self): + with tempfile.TemporaryDirectory() as tmp: + pidfile = os.path.join(tmp, "grandchild.pid") + spawner = _write_fake_agent_script( + os.path.join(tmp, "spawner.py"), + ( + "import subprocess, sys, time\n" + "p = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'])\n" + f"open({pidfile!r}, 'w').write(str(p.pid))\n" + "time.sleep(30)\n" + ), + ) + noisy = _write_fake_agent_script( + os.path.join(tmp, "noisy.py"), + ( + "import sys\n" + "sys.stdout.write('X' * 50000)\n" + "open('huge.txt', 'w').write('Y\\n' * 10000)\n" + ), + ) + + with AgenticWorkspace(max_output_bytes=1024, max_diff_bytes=2048) as ws: + timed_out, rc, _out, _err = ws.run_subprocess( + [sys.executable, spawner], timeout_sec=0.4 + ) + self.assertTrue(timed_out) + self.assertNotEqual(rc, 0) + if os.path.exists(pidfile): + gpid = int(open(pidfile, encoding="utf-8").read().strip()) + self.assertFalse(_pid_alive(gpid), "grandchild process should be terminated") + + to2, rc2, out2, _ = ws.run_subprocess( + [sys.executable, noisy], timeout_sec=5.0 + ) + self.assertFalse(to2) + self.assertEqual(rc2, 0) + self.assertLess(len(out2), 2048) + self.assertIn("truncated", out2) + diff = ws.get_diff() + self.assertLess(len(diff), 4096) + self.assertIn("diff truncated", diff) + + +class TestTaskVerifierAndOutcomes(unittest.TestCase): + def test_verifier_checks_and_no_verifier_inconclusive(self): + with tempfile.TemporaryDirectory() as tmp: + # No verifier configured must be inconclusive (insufficient_evidence), NOT pass + empty_v = TaskVerifier() + outcome, hard, soft, _rat, reason = empty_v.verify(tmp) + self.assertEqual(outcome, OUTCOME_INCONCLUSIVE) + self.assertEqual(reason, REASON_INSUFFICIENT_EVIDENCE) + self.assertEqual(hard, 0.0) + self.assertEqual(soft, 0.0) + + verifier = TaskVerifier(verifier_fn=lambda _p: (True, "all checks passed")) + self.assertEqual(verifier.verify(tmp)[:3], (OUTCOME_PASS, 1.0, 1.0)) + + failing = TaskVerifier(verifier_fn=lambda _p: (False, "assertion failed")) + self.assertEqual(failing.verify(tmp)[:3], (OUTCOME_FAIL, 0.0, 0.0)) + + exp = TaskVerifier(expected_files={"out.json": '"ok"'}) + self.assertEqual(exp.verify(tmp)[0], OUTCOME_FAIL) + with open(os.path.join(tmp, "out.json"), "w", encoding="utf-8") as f: + f.write('{"status": "ok"}') + self.assertEqual(exp.verify(tmp)[0], OUTCOME_PASS) + + timeout_v = TaskVerifier(test_cmd=[sys.executable, "-c", "import time; time.sleep(5)"], timeout_sec=0.3) + t_out = timeout_v.verify(tmp) + self.assertEqual(t_out[0], OUTCOME_INCONCLUSIVE) + self.assertEqual(t_out[4], REASON_TIMEOUT) + + def test_nonzero_agent_exit_is_infra_and_bad_pin_is_fixture_failure(self): + with tempfile.TemporaryDirectory() as tmp: + fixture_root = os.path.join(tmp, "fixtures") + repo = os.path.join(fixture_root, "repo") + _init_git_repo(repo) + + be_bad_pin = AgenticReplayBackend( + AgenticReplayConfig( + enabled=True, + fixture_root=fixture_root, + pinned_commit="deadbeefdeadbeefdeadbeef", + ), + agent_runner=lambda _ws, _env: {"response": "ok", "actions": 1}, + default_verifier=TaskVerifier(verifier_fn=lambda _p: (True, "ok")), + ) + pairs = replay_batch(be_bad_pin, [TaskRecord(id="badpin", project=repo, intent="test")], "", "") + self.assertEqual(len(pairs), 1) + self.assertEqual(pairs[0][1].outcome, OUTCOME_INCONCLUSIVE) + self.assertEqual(pairs[0][1].inconclusive_reason, REASON_FIXTURE_FAILURE) + + be_exit = AgenticReplayBackend( + AgenticReplayConfig(enabled=True), + agent_runner=lambda _ws, _env: {"response": "crash", "actions": 1, "returncode": 2}, + default_verifier=TaskVerifier(verifier_fn=lambda _p: (True, "ok")), + ) + res = be_exit.replay_agentic_task(TaskRecord(id="crash", project=tmp, intent="fail")) + self.assertEqual(res.outcome, OUTCOME_INCONCLUSIVE) + self.assertEqual(res.inconclusive_reason, REASON_INFRA) + + def test_budgets_and_thread_safety(self): + be = AgenticReplayBackend( + AgenticReplayConfig( + enabled=True, + max_turns=2, + max_tokens=100, + max_replays_per_run=3, + min_decidable_actions=1, + ), + agent_runner=lambda _ws, _env: {"response": "ok", "actions": 1, "turns": 1, "tokens": 0}, + default_verifier=TaskVerifier(verifier_fn=lambda _p: (True, "ok")), + ) + # Parallel replay batch of 6 tasks against cap 3 must yield strictly 3 pass + 3 budget_exhausted + tasks = [TaskRecord(id=f"p{i}", project="/p", intent="run") for i in range(6)] + with ThreadPoolExecutor(max_workers=4) as pool: + results = list(pool.map(lambda t: be.replay_agentic_task(t), tasks)) + passes = [r for r in results if r.outcome == OUTCOME_PASS] + budget = [r for r in results if r.inconclusive_reason == REASON_BUDGET_EXHAUSTED] + self.assertEqual(len(passes), 3) + self.assertEqual(len(budget), 3) + self.assertEqual(sum(r.tokens for r in results), 0) # No hardcoded fake 500-token constant + + # Check turn cap and token cap enforcement + be2 = AgenticReplayBackend( + AgenticReplayConfig(enabled=True, max_turns=2, max_tokens=50, max_replays_per_run=5), + agent_runner=lambda _ws, _env: {"response": "ok", "actions": 1, "turns": 3, "tokens": 10}, + default_verifier=TaskVerifier(verifier_fn=lambda _p: (True, "ok")), + ) + turn_over = be2.replay_agentic_task(TaskRecord(id="t-turn", project="/p", intent="run")) + self.assertEqual(turn_over.inconclusive_reason, REASON_BUDGET_EXHAUSTED) + + be3 = AgenticReplayBackend( + AgenticReplayConfig(enabled=True, max_turns=5, max_tokens=50, max_replays_per_run=5), + agent_runner=lambda _ws, _env: {"response": "ok", "actions": 1, "turns": 1, "tokens": 60}, + default_verifier=TaskVerifier(verifier_fn=lambda _p: (True, "ok")), + ) + tok_over = be3.replay_agentic_task(TaskRecord(id="t-tok", project="/p", intent="run")) + self.assertEqual(tok_over.inconclusive_reason, REASON_BUDGET_EXHAUSTED) + + +class TestReflectionAndPairwiseGateSemantics(unittest.TestCase): + def test_tiered_reflection_excludes_inconclusive_via_spy(self): + base = MockBackend() + spy_reflect = mock.MagicMock(wraps=base.reflect) + base.reflect = spy_reflect # type: ignore[method-assign] + backend = AgenticReplayBackend(AgenticReplayConfig(enabled=True), base_backend=base) + + t_fail = TaskRecord(id="f1", project="/p", intent="fail deterministically") + t_timeout = TaskRecord(id="to1", project="/p", intent="timed out") + failures = [ + (t_fail, ReplayResult(id="f1", hard=0.0, outcome=OUTCOME_FAIL)), + (t_timeout, ReplayResult(id="to1", hard=0.0, outcome=OUTCOME_INCONCLUSIVE, inconclusive_reason=REASON_TIMEOUT)), + ] + successes = [ + (TaskRecord(id="s1", project="/p", intent="ok"), ReplayResult(id="s1", hard=1.0, outcome=OUTCOME_PASS)), + (TaskRecord(id="s_inc", project="/p", intent="bad"), ReplayResult(id="s_inc", hard=1.0, outcome=OUTCOME_INCONCLUSIVE)), + ] + + backend.reflect( + failures, + successes, + skill="# s", + memory="# m", + edit_budget=2, + evolve_skill=True, + evolve_memory=False, + ) + spy_reflect.assert_called_once() + passed_failures = spy_reflect.call_args.args[0] + passed_successes = spy_reflect.call_args.args[1] + self.assertEqual([t.id for t, _r in passed_failures], ["f1"]) + self.assertEqual([t.id for t, _r in passed_successes], ["s1"]) + + def test_aggregate_and_pairwise_scores(self): + t1 = TaskRecord(id="t1", project="/p", intent="1") + t2 = TaskRecord(id="t2", project="/p", intent="2") + t3 = TaskRecord(id="t3", project="/p", intent="3") + pairs = [ + (t1, ReplayResult(id="t1", hard=1.0, soft=1.0, outcome=OUTCOME_PASS)), + (t2, ReplayResult(id="t2", hard=0.0, soft=0.0, outcome=OUTCOME_FAIL)), + (t3, ReplayResult(id="t3", hard=0.0, soft=0.0, outcome=OUTCOME_INCONCLUSIVE, inconclusive_reason=REASON_INFRA)), + ] + self.assertAlmostEqual(aggregate_scores(pairs)[0], 1 / 3) + self.assertAlmostEqual(aggregate_scores(pairs, exclude_inconclusive=True)[0], 0.5) + + base_pairs = [ + (t1, ReplayResult(id="t1", hard=0.0, soft=0.0, outcome=OUTCOME_FAIL)), + (t2, ReplayResult(id="t2", hard=1.0, soft=1.0, outcome=OUTCOME_PASS)), + (t3, ReplayResult(id="t3", hard=1.0, soft=1.0, outcome=OUTCOME_PASS)), + ] + cand_pairs = [ + (t1, ReplayResult(id="t1", hard=1.0, soft=1.0, outcome=OUTCOME_PASS)), + (t2, ReplayResult(id="t2", hard=1.0, soft=1.0, outcome=OUTCOME_PASS)), + (t3, ReplayResult(id="t3", hard=0.0, soft=0.0, outcome=OUTCOME_INCONCLUSIVE, inconclusive_reason=REASON_BUDGET_EXHAUSTED)), + ] + b_pair, c_pair, n_dec, n_tot, cov = pairwise_aggregate_scores(base_pairs, cand_pairs) + self.assertEqual((n_dec, n_tot), (2, 3)) + self.assertAlmostEqual(cov, 2 / 3) + self.assertAlmostEqual(b_pair[0], 0.5) # baseline scored ONLY on {t1, t2} + self.assertAlmostEqual(c_pair[0], 1.0) # candidate scored ONLY on {t1, t2} + + def test_consolidate_4_task_budget_exhaustion_abstains_pairwise(self): + # Point 8 scenario: max_replays_per_run=5 on 4 val tasks + 1 train task. + # Baseline val gets 4 passes (replays 1..4), train uses replay 5 -> + # candidate trial & final val get inconclusive budget_exhausted. + val_tasks = [TaskRecord(id=f"v{i}", project="/p", intent="solve", split="val") for i in range(4)] + train_task = TaskRecord(id="tr0", project="/p", intent="learn", split="train") + tasks = [train_task, *val_tasks] + + class _ProposedEditBackend(MockBackend): + def reflect( + self, failures, successes, skill, memory, *, + edit_budget=4, evolve_skill=True, evolve_memory=True, + ): + if evolve_skill: + return [EditRecord(target="skill", op="add", content="New rule", rationale="fix")] + return [] + + def _runner(ws_path, _env, ctx): + return {"response": "ok", "actions": 1} + + # Replays 1..4 = baseline val pass; replay 5 = train pass; + # candidate trial runs v0 (replay 6) then v1..v3 budget_exhausted! + be_partial = AgenticReplayBackend( + AgenticReplayConfig(enabled=True, max_replays_per_run=6), + base_backend=_ProposedEditBackend(), + agent_runner=_runner, + default_verifier=TaskVerifier(verifier_fn=lambda _p: (True, "pass")), + ) + res = consolidate( + be_partial, + tasks, + skill="# Base", + memory="", + evolve_memory=False, + gate_metric="hard", + min_decidable_coverage=0.5, + ) + self.assertFalse(res.accepted) + self.assertEqual(res.gate_action, "abstain_inconclusive") + self.assertEqual(res.new_skill, "# Base") + self.assertEqual(res.applied_edits, []) + # Crucially: pairwise baseline/candidate never drop to 0.25 due to inconclusive zeros + self.assertGreaterEqual(res.baseline_score, 0.0) + trial_skill = [g for g in res.gate_trials if g["target"] == "skill"][0] + # Candidate trial had v0 pass (replay 6) and v1..v3 budget_exhausted => baseline_score==1.0, candidate_score==1.0, abstained! + self.assertAlmostEqual(trial_skill["baseline_score"], 1.0) + self.assertAlmostEqual(trial_skill["candidate_score"], 1.0) + self.assertTrue(trial_skill["abstained_inconclusive"]) + self.assertAlmostEqual(trial_skill["decidable_coverage"], 0.25) + + def test_consolidate_pairwise_accepts_above_coverage_floor_with_no_regression(self): + val_tasks = [TaskRecord(id=f"v{i}", project="/p", intent=f"val {i}", split="val") for i in range(4)] + train_task = TaskRecord(id="tr0", project="/p", intent="learn", split="train") + tasks = [train_task, *val_tasks] + + class _EditBackend(MockBackend): + def reflect( + self, failures, successes, skill, memory, *, + edit_budget=4, evolve_skill=True, evolve_memory=True, + ): + if evolve_skill: + return [EditRecord(target="skill", op="add", content="Improved rule", rationale="improve v0")] + return [] + + def _batch_side_effect(_backend, batch_tasks, skill, _memory, **_kw): + out = [] + improved = "Improved rule" in skill + for t in batch_tasks: + if t.id == "v0": + out.append((t, ReplayResult(id=t.id, hard=1.0 if improved else 0.0, soft=1.0 if improved else 0.0))) + elif t.id in {"v1", "v2"}: + out.append((t, ReplayResult(id=t.id, hard=1.0, soft=1.0))) + elif t.id == "v3": + # Candidate hits timeout on v3; baseline passed v3. Pairwise excludes v3 from both sides! + if improved: + out.append((t, ReplayResult(id=t.id, hard=0.0, outcome=OUTCOME_INCONCLUSIVE, inconclusive_reason=REASON_TIMEOUT))) + else: + out.append((t, ReplayResult(id=t.id, hard=1.0, soft=1.0))) + else: + out.append((t, ReplayResult(id=t.id, hard=0.0, outcome=OUTCOME_FAIL))) + return out + + with mock.patch("skillopt_sleep.consolidate.replay_batch", side_effect=_batch_side_effect): + res = consolidate( + _EditBackend(), + tasks, + skill="# Base", + memory="", + evolve_memory=False, + gate_metric="hard", + gate_no_regression=True, + min_decidable_coverage=0.5, + ) + + self.assertTrue(res.accepted) + self.assertEqual(res.gate_action, "accept_new_best") + # Baseline on {v0, v1, v2} = 2/3, Candidate on {v0, v1, v2} = 3/3 (coverage 0.75 >= 0.5) + self.assertAlmostEqual(res.baseline_score, 2 / 3) + self.assertAlmostEqual(res.candidate_score, 1.0) + final_trial = res.gate_trials[-1] + v3_row = [r for r in final_trial["task_deltas"] if r["task_id"] == "v3"][0] + self.assertEqual(v3_row["status"], "inconclusive") + + +if __name__ == "__main__": + unittest.main()