fix: integrate six focused workflow repairs with canonical registry identity

This commit is contained in:
haelyra committed 2026-10-11 11:19:21 -04:00
commit 1abc625de9
16 files changed
+541 -35

No files matched your search

@@ -19,6 +19,7 @@ For production use, pair with an LLM judge for semantic understanding.
"""
import argparse
import errno
import re
import sys
from dataclasses import dataclass, field
@@ -368,7 +369,9 @@ def _read_file_or_text(path: Optional[str], *, required: bool = False) -> Option
try:
with open(path) as f:
return f.read()
except FileNotFoundError:
except OSError as error:
if error.errno not in (errno.ENOENT, errno.ENAMETOOLONG):
raise
if required:
print(f"Error: output file '{path}' not found", file=sys.stderr)
sys.exit(1)
@@ -382,7 +385,10 @@ def _read_input(args: argparse.Namespace) -> tuple[Optional[str], str]:
print("Paste agent output (Ctrl+D to finish):")
return task, sys.stdin.read()
if args.output:
return _read_file_or_text(args.task), _read_file_or_text(args.output, required=True) or ""
task = _read_file_or_text(args.task)
if task is None and not sys.stdin.isatty():
task = sys.stdin.read()
return task, _read_file_or_text(args.output, required=True) or ""
return _read_file_or_text(args.task), sys.stdin.read()
+5 -1
View File
@@ -94,8 +94,12 @@ def _safe_sandbox_dir(scenario_id: str) -> Path:
"""Sanitize scenario ID and ensure path stays within sandbox base."""
safe_id = re.sub(r"[^a-zA-Z0-9\-_]", "_", scenario_id)
path = SANDBOX_BASE / safe_id
resolved_path = path.resolve()
resolved_base = SANDBOX_BASE.resolve()
if not safe_id or resolved_path == resolved_base:
raise ValueError("Scenario sandbox must be a subdirectory of the sandbox base")
# Validate path stays within sandbox base (raises ValueError on traversal)
path.resolve().relative_to(SANDBOX_BASE.resolve())
resolved_path.relative_to(resolved_base)
return path
+38 -1
View File
@@ -9,7 +9,13 @@ from pathlib import Path
from unittest.mock import patch
import pytest
from scripts.runner import _parse_stream_json, _setup_sandbox, run_scenario
from scripts.runner import (
SANDBOX_BASE,
_parse_stream_json,
_safe_sandbox_dir,
_setup_sandbox,
run_scenario,
)
@dataclass(frozen=True)
@@ -21,6 +27,37 @@ class _FakeScenario:
setup_commands: tuple[str, ...] = ()
class TestSafeSandboxDirectory:
def test_empty_id_cannot_delete_the_shared_base(self) -> None:
with (
patch("scripts.runner.Path.exists", return_value=True),
patch("scripts.runner.shutil.rmtree") as remove,
patch("scripts.runner.Path.mkdir") as mkdir,
patch("scripts.runner.subprocess.run") as execute,
):
execute.return_value = subprocess.CompletedProcess([], 0, stdout="", stderr="")
with pytest.raises(ValueError, match="sandbox"):
run_scenario(_FakeScenario(id=""))
remove.assert_not_called()
mkdir.assert_not_called()
execute.assert_not_called()
def test_resolved_base_is_not_a_scenario_directory(self) -> None:
with patch("scripts.runner.Path.resolve", return_value=SANDBOX_BASE):
with pytest.raises(ValueError, match="sandbox"):
_safe_sandbox_dir("alias")
def test_valid_id_keeps_existing_sanitization(self) -> None:
assert _safe_sandbox_dir("scenario/one") == SANDBOX_BASE / "scenario_one"
def test_resolved_escape_is_still_rejected(self, tmp_path, monkeypatch) -> None:
base = tmp_path / "sandbox"
monkeypatch.setattr("scripts.runner.SANDBOX_BASE", base)
with patch.object(Path, "resolve", side_effect=[tmp_path / "outside", base]):
with pytest.raises(ValueError):
_safe_sandbox_dir("escape")
class TestSetupSandboxSkipsShellBuiltins:
"""Setup commands containing shell builtins (cd/pushd/popd) must be skipped.