chore: import upstream snapshot with attribution

tests/test_agents_smoke.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from pathlib import Path
+import py_compile
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+AGENTS_DIR = ROOT / "agents"
+AGENT_FILES = sorted(
+    path for path in AGENTS_DIR.glob("*.py") if path.name != "__init__.py"
+)
+AGENT_IDS = [path.name for path in AGENT_FILES]
+
+
+@pytest.mark.parametrize("agent_path", AGENT_FILES, ids=AGENT_IDS)
+def test_agent_scripts_compile(agent_path: Path) -> None:
+    _ = py_compile.compile(str(agent_path), doraise=True)
+
+
+def test_agent_scripts_exist() -> None:
+    assert AGENT_FILES, "expected at least one agent script"

tests/test_compaction_tool_pairs.py

@@ -0,0 +1,255 @@
+import importlib.util
+import os
+import sys
+import tempfile
+import types
+import unittest
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+MODULES = {
+    "s08": REPO_ROOT / "s08_context_compact" / "code.py",
+    "s09": REPO_ROOT / "s09_memory" / "code.py",
+    "s20": REPO_ROOT / "s20_comprehensive" / "code.py",
+}
+
+
+def load_module(name: str, path: Path, temp_cwd: Path):
+    fake_anthropic = types.ModuleType("anthropic")
+
+    class FakeAnthropic:
+        def __init__(self, *args, **kwargs):
+            self.messages = types.SimpleNamespace(create=None)
+
+    fake_dotenv = types.ModuleType("dotenv")
+    setattr(fake_anthropic, "Anthropic", FakeAnthropic)
+    setattr(fake_dotenv, "load_dotenv", lambda override=True: None)
+
+    previous_anthropic = sys.modules.get("anthropic")
+    previous_dotenv = sys.modules.get("dotenv")
+    previous_cwd = Path.cwd()
+    previous_model = os.environ.get("MODEL_ID")
+    previous_key = os.environ.get("ANTHROPIC_API_KEY")
+
+    spec = importlib.util.spec_from_file_location(name, path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Unable to load {path}")
+    module = importlib.util.module_from_spec(spec)
+
+    sys.modules["anthropic"] = fake_anthropic
+    sys.modules["dotenv"] = fake_dotenv
+    os.environ["MODEL_ID"] = "test-model"
+    os.environ["ANTHROPIC_API_KEY"] = "test-key"
+    try:
+        os.chdir(temp_cwd)
+        spec.loader.exec_module(module)
+        return module
+    finally:
+        os.chdir(previous_cwd)
+        if previous_anthropic is None:
+            sys.modules.pop("anthropic", None)
+        else:
+            sys.modules["anthropic"] = previous_anthropic
+        if previous_dotenv is None:
+            sys.modules.pop("dotenv", None)
+        else:
+            sys.modules["dotenv"] = previous_dotenv
+        if previous_model is None:
+            os.environ.pop("MODEL_ID", None)
+        else:
+            os.environ["MODEL_ID"] = previous_model
+        if previous_key is None:
+            os.environ.pop("ANTHROPIC_API_KEY", None)
+        else:
+            os.environ["ANTHROPIC_API_KEY"] = previous_key
+
+
+def assistant_text():
+    return {"role": "assistant", "content": [types.SimpleNamespace(type="text", text="ok")]}
+
+
+def user_text():
+    return {"role": "user", "content": "continue"}
+
+
+def tool_use_message(tool_id="tool-1"):
+    return {
+        "role": "assistant",
+        "content": [types.SimpleNamespace(type="tool_use", id=tool_id, name="bash")],
+    }
+
+
+def tool_result_message(tool_id="tool-1"):
+    return {
+        "role": "user",
+        "content": [{"type": "tool_result", "tool_use_id": tool_id, "content": "ok"}],
+    }
+
+
+def message_has_tool_use(message):
+    content = message.get("content")
+    return (
+        message.get("role") == "assistant"
+        and isinstance(content, list)
+        and any(getattr(block, "type", None) == "tool_use" for block in content)
+    )
+
+
+def assert_no_orphan_tool_results(testcase, messages):
+    for idx, message in enumerate(messages):
+        content = message.get("content")
+        if message.get("role") != "user" or not isinstance(content, list):
+            continue
+        if not any(isinstance(block, dict) and block.get("type") == "tool_result" for block in content):
+            continue
+        testcase.assertGreater(idx, 0)
+        testcase.assertTrue(message_has_tool_use(messages[idx - 1]), messages)
+
+
+class CompactionToolPairTests(unittest.TestCase):
+    def test_snip_compact_keeps_head_tool_pair(self):
+        messages = [
+            user_text(),
+            assistant_text(),
+            tool_use_message("head-tool"),
+            tool_result_message("head-tool"),
+            assistant_text(),
+            user_text(),
+            assistant_text(),
+            user_text(),
+            assistant_text(),
+            user_text(),
+        ]
+
+        for name, path in MODULES.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
+                module = load_module(f"{name}_head_under_test", path, Path(tmp))
+                if name == "s09":
+                    compacted = module.snip_compact(list(messages), mx=6)
+                else:
+                    compacted = module.snip_compact(list(messages), max_messages=6)
+                self.assertEqual(compacted[2], messages[2])
+                self.assertEqual(compacted[3], messages[3])
+                assert_no_orphan_tool_results(self, compacted)
+
+    def test_snip_compact_keeps_tail_tool_pair(self):
+        messages = [
+            user_text(),
+            assistant_text(),
+            user_text(),
+            assistant_text(),
+            user_text(),
+            assistant_text(),
+            tool_use_message("tail-tool"),
+            tool_result_message("tail-tool"),
+            assistant_text(),
+            user_text(),
+        ]
+
+        for name, path in MODULES.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
+                module = load_module(f"{name}_under_test", path, Path(tmp))
+                if name == "s09":
+                    compacted = module.snip_compact(list(messages), mx=6)
+                else:
+                    compacted = module.snip_compact(list(messages), max_messages=6)
+                assert_no_orphan_tool_results(self, compacted)
+
+    def test_reactive_compact_keeps_tail_tool_pair(self):
+        messages = [
+            user_text(),
+            assistant_text(),
+            user_text(),
+            tool_use_message("reactive-tool"),
+            tool_result_message("reactive-tool"),
+            assistant_text(),
+            user_text(),
+            assistant_text(),
+            user_text(),
+        ]
+
+        for name, path in MODULES.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
+                module = load_module(f"{name}_reactive_under_test", path, Path(tmp))
+                module.write_transcript = lambda _messages: Path("transcript.jsonl")
+                module.summarize_history = lambda _messages: "summary"
+                compacted = module.reactive_compact(list(messages))
+                self.assertEqual(compacted[1], messages[3])
+                assert_no_orphan_tool_results(self, compacted)
+
+    def test_reactive_compact_summarizes_only_old_history(self):
+        messages = [
+            user_text(),
+            assistant_text(),
+            user_text(),
+            assistant_text(),
+            user_text(),
+            assistant_text(),
+            user_text(),
+            assistant_text(),
+            user_text(),
+        ]
+
+        for name, path in MODULES.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
+                module = load_module(f"{name}_reactive_oldhist_under_test", path, Path(tmp))
+                module.write_transcript = lambda _messages: Path("transcript.jsonl")
+                captured = {}
+
+                def fake_summarize(passed, _store=captured):
+                    _store["messages"] = list(passed)
+                    return "summary"
+
+                module.summarize_history = fake_summarize
+                compacted = module.reactive_compact(list(messages))
+                # The summary must cover only the old history, not the kept tail.
+                self.assertEqual(captured["messages"], messages[:4])
+                # The recent tail is appended verbatim after the summary message.
+                self.assertEqual(compacted[1:], messages[4:])
+                assert_no_orphan_tool_results(self, compacted)
+
+    def test_reactive_compact_summary_excludes_tail_pair_pulled_in(self):
+        # A tool_use/tool_result pair straddles the tail boundary, so the
+        # adjustment pulls the tool_use into the kept tail. The summary must
+        # cover only what stays trimmed (messages[:adjusted_tail_start]), i.e.
+        # it must not re-summarize the tool_use that is kept verbatim.
+        messages = [
+            user_text(),
+            assistant_text(),
+            user_text(),
+            tool_use_message("reactive-tool"),
+            tool_result_message("reactive-tool"),
+            assistant_text(),
+            user_text(),
+            assistant_text(),
+            user_text(),
+        ]
+
+        for name, path in MODULES.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as tmp:
+                module = load_module(f"{name}_reactive_pairscope_under_test", path, Path(tmp))
+                module.write_transcript = lambda _messages: Path("transcript.jsonl")
+                captured = {}
+
+                def fake_summarize(passed, _store=captured):
+                    _store["messages"] = list(passed)
+                    return "summary"
+
+                module.summarize_history = fake_summarize
+                compacted = module.reactive_compact(list(messages))
+                # tail_start starts at 4, decrements to 3 to keep the pair intact.
+                self.assertEqual(captured["messages"], messages[:3])
+                self.assertEqual(compacted[1], messages[3])
+                self.assertEqual(compacted[1:], messages[3:])
+                assert_no_orphan_tool_results(self, compacted)
+
+    def test_s20_has_tool_use_still_accepts_content_blocks(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            module = load_module("s20_has_tool_use_under_test", MODULES["s20"], Path(tmp))
+            self.assertTrue(module.has_tool_use([types.SimpleNamespace(type="tool_use")]))
+            self.assertFalse(module.has_tool_use([types.SimpleNamespace(type="text")]))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_s_full_background.py

@@ -0,0 +1,67 @@
+import importlib.util
+import os
+import sys
+import tempfile
+import types
+import unittest
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+MODULE_PATH = REPO_ROOT / "agents" / "s_full.py"
+
+
+def load_s_full_module(temp_cwd: Path):
+    fake_anthropic = types.ModuleType("anthropic")
+
+    class FakeAnthropic:
+        def __init__(self, *args, **kwargs):
+            self.messages = types.SimpleNamespace(create=None)
+
+    fake_dotenv = types.ModuleType("dotenv")
+    setattr(fake_anthropic, "Anthropic", FakeAnthropic)
+    setattr(fake_dotenv, "load_dotenv", lambda override=True: None)
+
+    previous_anthropic = sys.modules.get("anthropic")
+    previous_dotenv = sys.modules.get("dotenv")
+    previous_cwd = Path.cwd()
+    spec = importlib.util.spec_from_file_location("s_full_under_test", MODULE_PATH)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Unable to load {MODULE_PATH}")
+    module = importlib.util.module_from_spec(spec)
+
+    sys.modules["anthropic"] = fake_anthropic
+    sys.modules["dotenv"] = fake_dotenv
+    try:
+        os.chdir(temp_cwd)
+        os.environ.setdefault("MODEL_ID", "test-model")
+        spec.loader.exec_module(module)
+        return module
+    finally:
+        os.chdir(previous_cwd)
+        if previous_anthropic is None:
+            sys.modules.pop("anthropic", None)
+        else:
+            sys.modules["anthropic"] = previous_anthropic
+        if previous_dotenv is None:
+            sys.modules.pop("dotenv", None)
+        else:
+            sys.modules["dotenv"] = previous_dotenv
+
+
+class BackgroundManagerTests(unittest.TestCase):
+    def test_check_returns_running_placeholder_when_result_is_none(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            module = load_s_full_module(Path(tmp))
+            manager = module.BackgroundManager()
+            manager.tasks["abc123"] = {
+                "status": "running",
+                "command": "sleep 1",
+                "result": None,
+            }
+
+            self.assertEqual(manager.check("abc123"), "[running] (running)")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_todo_write_string_input.py

@@ -0,0 +1,115 @@
+import importlib.util
+import os
+import sys
+import tempfile
+import types
+import unittest
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+COURSE_MODULES = [
+    ("s05", REPO_ROOT / "s05_todo_write" / "code.py"),
+    ("s06", REPO_ROOT / "s06_subagent" / "code.py"),
+    ("s07", REPO_ROOT / "s07_skill_loading" / "code.py"),
+    ("s08", REPO_ROOT / "s08_context_compact" / "code.py"),
+    ("s20", REPO_ROOT / "s20_comprehensive" / "code.py"),
+]
+
+
+def load_course_module(module_name: str, module_path: Path, temp_cwd: Path):
+    fake_anthropic = types.ModuleType("anthropic")
+
+    class FakeAnthropic:
+        def __init__(self, *args, **kwargs):
+            self.messages = types.SimpleNamespace(create=None)
+
+    fake_dotenv = types.ModuleType("dotenv")
+    fake_yaml = types.ModuleType("yaml")
+    setattr(fake_anthropic, "Anthropic", FakeAnthropic)
+    setattr(fake_dotenv, "load_dotenv", lambda override=True: None)
+    setattr(fake_yaml, "safe_load", lambda text: {})
+    setattr(fake_yaml, "YAMLError", Exception)
+
+    previous_modules = {
+        "anthropic": sys.modules.get("anthropic"),
+        "dotenv": sys.modules.get("dotenv"),
+        "yaml": sys.modules.get("yaml"),
+    }
+    previous_cwd = Path.cwd()
+    previous_model_id = os.environ.get("MODEL_ID")
+
+    spec = importlib.util.spec_from_file_location(f"{module_name}_todo_test", module_path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Unable to load {module_path}")
+    module = importlib.util.module_from_spec(spec)
+
+    sys.modules["anthropic"] = fake_anthropic
+    sys.modules["dotenv"] = fake_dotenv
+    sys.modules["yaml"] = fake_yaml
+    try:
+        os.chdir(temp_cwd)
+        os.environ["MODEL_ID"] = "test-model"
+        spec.loader.exec_module(module)
+        return module
+    finally:
+        os.chdir(previous_cwd)
+        if previous_model_id is None:
+            os.environ.pop("MODEL_ID", None)
+        else:
+            os.environ["MODEL_ID"] = previous_model_id
+        for name, previous in previous_modules.items():
+            if previous is None:
+                sys.modules.pop(name, None)
+            else:
+                sys.modules[name] = previous
+
+
+class TodoWriteStringInputTests(unittest.TestCase):
+    def test_issue_340_accepts_json_array_string(self):
+        for module_name, module_path in COURSE_MODULES:
+            with self.subTest(module=module_name), tempfile.TemporaryDirectory() as tmp:
+                module = load_course_module(module_name, module_path, Path(tmp))
+
+                result = module.run_todo_write(
+                    '[{"content": "inspect repo", "status": "pending"}]'
+                )
+
+                self.assertIn("Updated 1", result)
+                self.assertEqual(
+                    module.CURRENT_TODOS,
+                    [{"content": "inspect repo", "status": "pending"}],
+                )
+
+    def test_issue_340_accepts_python_list_repr_string(self):
+        for module_name, module_path in COURSE_MODULES:
+            with self.subTest(module=module_name), tempfile.TemporaryDirectory() as tmp:
+                module = load_course_module(module_name, module_path, Path(tmp))
+
+                result = module.run_todo_write(
+                    "[{'content': 'write tests', 'status': 'in_progress'}]"
+                )
+
+                self.assertIn("Updated 1", result)
+                self.assertEqual(
+                    module.CURRENT_TODOS,
+                    [{"content": "write tests", "status": "in_progress"}],
+                )
+
+    def test_issue_340_does_not_eval_string_inputs(self):
+        for module_name, module_path in COURSE_MODULES:
+            with self.subTest(module=module_name), tempfile.TemporaryDirectory() as tmp:
+                tmp_path = Path(tmp)
+                marker = tmp_path / "eval_was_executed"
+                module = load_course_module(module_name, module_path, tmp_path)
+
+                result = module.run_todo_write(
+                    f"__import__('pathlib').Path({str(marker)!r}).write_text('bad')"
+                )
+
+                self.assertIn("Error:", result)
+                self.assertFalse(marker.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()