diff --git a/skills/skill-creator/scripts/run_eval.py b/skills/skill-creator/scripts/run_eval.py index e58c70be..342bce31 100755 --- a/skills/skill-creator/scripts/run_eval.py +++ b/skills/skill-creator/scripts/run_eval.py @@ -9,6 +9,7 @@ import argparse import json import os import select +import shutil import subprocess import sys import time @@ -42,19 +43,41 @@ def run_single_query( ) -> bool: """Run a single query and return whether the skill was triggered. - Creates a command file in .claude/commands/ so it appears in Claude's - available_skills list, then runs `claude -p` with the raw query. - Uses --include-partial-messages to detect triggering early from - stream events (content_block_start) rather than waiting for the - full assistant message, which only arrives after tool execution. + Creates a command file in a per-run .claude/commands/ directory so it + appears in Claude's available_skills list, then runs `claude -p` with + the raw query. Uses --include-partial-messages to detect triggering + early from stream events (content_block_start) rather than waiting + for the full assistant message, which only arrives after tool + execution. + + Each run gets its own project root under .eval-runs// (see + issues #1552 and #1559): with a shared .claude/commands/, N parallel + workers would each see their own copy plus the N-1 copies belonging + to other in-flight runs — N near-identical descriptions competing for + the same query, so a run has only a ~1/N chance of being credited. """ unique_id = uuid.uuid4().hex[:8] clean_name = f"{skill_name}-skill-{unique_id}" - project_commands_dir = Path(project_root) / ".claude" / "commands" + run_root = Path(project_root) / ".eval-runs" / unique_id + project_commands_dir = run_root / ".claude" / "commands" command_file = project_commands_dir / f"{clean_name}.md" try: project_commands_dir.mkdir(parents=True, exist_ok=True) + # Mirror the project's top-level entries so the run's working + # directory still looks like a real project (an empty scratch dir + # makes the model explore with Bash and invoke nothing — issue + # #1559). Skip .claude and .eval-runs themselves. + for entry in Path(project_root).iterdir(): + if entry.name in (".claude", ".eval-runs"): + continue + link = run_root / entry.name + if not link.exists(): + try: + link.symlink_to(entry, target_is_directory=entry.is_dir()) + except OSError: + pass # e.g. Windows without Developer Mode; isolation holds + # Use YAML block scalar to avoid breaking on quotes in description indented_desc = "\n ".join(skill_description.split("\n")) command_content = ( @@ -86,7 +109,7 @@ def run_single_query( cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, - cwd=project_root, + cwd=str(run_root), env=env, ) @@ -97,12 +120,103 @@ def run_single_query( pending_tool_name = None accumulated_json = "" + def parse_line(line: str) -> bool | None: + """Parse one stream-json line. + + Returns True once the skill is confirmed triggered, or None to + keep reading. Never returns False mid-stream: a non-skill tool + call (Bash orientation, etc.) or a completed non-matching block + is not a verdict — the skill may still fire in a later turn + (issue #1559). Only the final `result` event (or the drained + tail) is conclusive. + """ + nonlocal triggered, pending_tool_name, accumulated_json + line = line.strip() + if not line: + return None + + try: + event = json.loads(line) + except json.JSONDecodeError: + return None + + # Early detection via stream events + if event.get("type") == "stream_event": + se = event.get("event", {}) + se_type = se.get("type", "") + + if se_type == "content_block_start": + cb = se.get("content_block", {}) + if cb.get("type") == "tool_use": + tool_name = cb.get("name", "") + if tool_name in ("Skill", "Read"): + pending_tool_name = tool_name + accumulated_json = "" + else: + # Not the skill — keep watching (issue #1559). + pending_tool_name = None + accumulated_json = "" + + elif se_type == "content_block_delta" and pending_tool_name: + delta = se.get("delta", {}) + if delta.get("type") == "input_json_delta": + accumulated_json += delta.get("partial_json", "") + if clean_name in accumulated_json: + return True + + elif se_type in ("content_block_stop", "message_stop"): + if pending_tool_name and clean_name in accumulated_json: + return True + # A completed non-matching block isn't conclusive — + # reset and keep reading (issue #1559). + pending_tool_name = None + accumulated_json = "" + + # Fallback: full assistant message + elif event.get("type") == "assistant": + message = event.get("message", {}) + for content_item in message.get("content", []): + if content_item.get("type") != "tool_use": + continue + tool_name = content_item.get("name", "") + tool_input = content_item.get("input", {}) + if tool_name == "Skill" and clean_name in tool_input.get("skill", ""): + triggered = True + return True + if tool_name == "Read" and clean_name in tool_input.get("file_path", ""): + triggered = True + return True + # No matching tool use in this message — keep reading. + + elif event.get("type") == "result": + return triggered + + return None + + def consume_buffer() -> bool | None: + """Parse all complete lines currently buffered; returns a + verdict (True/False) or None if no verdict found yet.""" + nonlocal buffer + while "\n" in buffer: + line, buffer = buffer.split("\n", 1) + verdict = parse_line(line) + if verdict is not None: + return verdict + return None + try: while time.time() - start_time < timeout: if process.poll() is not None: remaining = process.stdout.read() if remaining: buffer += remaining.decode("utf-8", errors="replace") + # The drained tail must be parsed too (issue #1552): + # when the child's output lands in one go — the common + # case for a single-turn answer — the verdict is only + # in this final read. + verdict = consume_buffer() + if verdict is not None: + return verdict break ready, _, _ = select.select([process.stdout], [], [], 1.0) @@ -114,61 +228,9 @@ def run_single_query( break buffer += chunk.decode("utf-8", errors="replace") - while "\n" in buffer: - line, buffer = buffer.split("\n", 1) - line = line.strip() - if not line: - continue - - try: - event = json.loads(line) - except json.JSONDecodeError: - continue - - # Early detection via stream events - if event.get("type") == "stream_event": - se = event.get("event", {}) - se_type = se.get("type", "") - - if se_type == "content_block_start": - cb = se.get("content_block", {}) - if cb.get("type") == "tool_use": - tool_name = cb.get("name", "") - if tool_name in ("Skill", "Read"): - pending_tool_name = tool_name - accumulated_json = "" - else: - return False - - elif se_type == "content_block_delta" and pending_tool_name: - delta = se.get("delta", {}) - if delta.get("type") == "input_json_delta": - accumulated_json += delta.get("partial_json", "") - if clean_name in accumulated_json: - return True - - elif se_type in ("content_block_stop", "message_stop"): - if pending_tool_name: - return clean_name in accumulated_json - if se_type == "message_stop": - return False - - # Fallback: full assistant message - elif event.get("type") == "assistant": - message = event.get("message", {}) - for content_item in message.get("content", []): - if content_item.get("type") != "tool_use": - continue - tool_name = content_item.get("name", "") - tool_input = content_item.get("input", {}) - if tool_name == "Skill" and clean_name in tool_input.get("skill", ""): - triggered = True - elif tool_name == "Read" and clean_name in tool_input.get("file_path", ""): - triggered = True - return triggered - - elif event.get("type") == "result": - return triggered + verdict = consume_buffer() + if verdict is not None: + return verdict finally: # Clean up process on any exit path (return, exception, timeout) if process.poll() is None: @@ -177,8 +239,12 @@ def run_single_query( return triggered finally: - if command_file.exists(): - command_file.unlink() + # Remove the per-run root (and .eval-runs itself once empty). + shutil.rmtree(run_root, ignore_errors=True) + try: + run_root.parent.rmdir() + except OSError: + pass def run_eval( diff --git a/skills/skill-creator/tests/test_run_eval.py b/skills/skill-creator/tests/test_run_eval.py new file mode 100644 index 00000000..8a271c3a --- /dev/null +++ b/skills/skill-creator/tests/test_run_eval.py @@ -0,0 +1,330 @@ +"""Tests for run_eval.py trigger detection (issues #1552 and #1559). + +These tests drive `run_single_query` with a fake `claude -p` subprocess +whose stdout is a real pipe fed canned stream-json events, so no Claude +Code installation is needed. + +Bugs covered: + #1552-1 the drained tail of the subprocess stream is never parsed, so + a single-turn answer (common case) always scores "not triggered" + #1552-2 parallel workers share one .claude/commands directory, so each + claude -p sees N-1 other workers' skill copies and usually + triggers a different one + #1559-1 a non-skill first tool call (Bash orientation, git status, ls) + ends the run before the skill can fire + #1559-2 a different skill firing first also ends the run; only a + positive match is conclusive + +Run with: python -m pytest skills/skill-creator/tests/ +""" + +import json +import os +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +TESTS_DIR = Path(__file__).resolve().parent +SKILL_CREATOR_DIR = TESTS_DIR.parent +sys.path.insert(0, str(SKILL_CREATOR_DIR)) +sys.path.insert(0, str(SKILL_CREATOR_DIR / "scripts")) + +from scripts import run_eval # noqa: E402 + + +# --------------------------------------------------------------------------- +# Stream event builders (Claude Code stream-json shapes) +# --------------------------------------------------------------------------- + + +def _event(se_type, **se_kwargs): + return {"type": "stream_event", "event": {"type": se_type, **se_kwargs}} + + +def tool_start(name, tool_id="toolu_1"): + return _event( + "content_block_start", + index=0, + content_block={"type": "tool_use", "id": tool_id, "name": name, "input": {}}, + ) + + +def tool_delta(partial_json): + return _event( + "content_block_delta", + index=0, + delta={"type": "input_json_delta", "partial_json": partial_json}, + ) + + +def block_stop(): + return _event("content_block_stop", index=0) + + +def message_stop(): + return _event("message_stop") + + +def result_event(): + return {"type": "result", "subtype": "success", "result": ""} + + +def stream(*events) -> bytes: + """Serialize events as newline-delimited stream-json bytes.""" + return ("\n".join(json.dumps(e) for e in events) + "\n").encode() + + +def skill_stream(clean_name, *, before=None, other_first=False): + """A stream where the target skill is called (optionally after other + tool calls / a different skill completing first).""" + events: list[dict] = [] + if before: + events.extend(before) + if other_first: + events.extend([ + tool_start("Skill", tool_id="toolu_other"), + tool_delta('{"skill":"some-other-skill-00000000"}'), + block_stop(), + ]) + events.extend([ + tool_start("Skill"), + tool_delta(f'{{"skill":"{clean_name}"}}'), + block_stop(), + result_event(), + ]) + return stream(*events) + + +class FakeClaudeProcess: + """A fake subprocess whose stdout is a real pipe with canned output. + + exit_after: number of poll() calls returning None before reporting + exit. 0 = the process already exited (all output lands in one go, + the drained-tail case); 1 = one streaming iteration, then exit. + """ + + def __init__(self, data: bytes, exit_after: int = 0): + read_fd, write_fd = os.pipe() + os.write(write_fd, data) + os.close(write_fd) + self.stdout = os.fdopen(read_fd, "rb", closefd=True) + self._polls = 0 + self._exit_after = exit_after + self.killed = False + self.waited = False + + def poll(self): + self._polls += 1 + if self._polls <= self._exit_after: + return None + return 0 + + def kill(self): + self.killed = True + + def wait(self): + self.waited = True + + +@pytest.fixture +def fake_claude(monkeypatch): + """Install a fake Popen; returns (install, captured_kwargs).""" + + captured = {} + + def install(data: bytes, exit_after: int = 0): + proc = FakeClaudeProcess(data, exit_after=exit_after) + + def fake_popen(cmd, **kwargs): + captured["cmd"] = cmd + captured["kwargs"] = kwargs + # Snapshot the per-run command file at spawn time (the real + # run deletes it in cleanup). + cwd = Path(kwargs["cwd"]) + commands_dir = cwd / ".claude" / "commands" + captured["command_files"] = [ + f.name for f in commands_dir.glob("*.md") + ] if commands_dir.is_dir() else [] + return proc + + monkeypatch.setattr(run_eval.subprocess, "Popen", fake_popen) + # The streaming branch calls select on the pipe; force "readable" + # so the test is portable (Windows select only handles sockets). + monkeypatch.setattr( + run_eval.select, + "select", + lambda r, w, x, timeout=None: (list(r), [], []), + ) + return proc + + return install, captured + + +@pytest.fixture +def fixed_clean_name(monkeypatch): + """Make the run's generated clean_name deterministic.""" + monkeypatch.setattr( + run_eval.uuid, "uuid4", lambda: SimpleNamespace(hex="deadbeef12345678") + ) + return "my-skill-skill-deadbeef" + + +def run(query="do the thing", project_root=None, tmp_path=None, timeout=10): + root = str(project_root or tmp_path) + return run_eval.run_single_query( + query=query, + skill_name="my-skill", + skill_description="Does a thing.", + timeout=timeout, + project_root=root, + ) + + +def skill_call_stream(clean_name): + """Minimal stream: the target skill is invoked and the run ends.""" + return stream( + tool_start("Skill"), + tool_delta(f'{{"skill":"{clean_name}"}}'), + block_stop(), + result_event(), + ) + + +# --------------------------------------------------------------------------- +# Drained tail is parsed (#1552-1) +# --------------------------------------------------------------------------- + + +def test_drained_tail_is_parsed(fake_claude, fixed_clean_name, tmp_path): + """All output lands in one read after the process exits — the common + single-turn case. The old code read the tail and never parsed it.""" + install, _ = fake_claude + install(skill_call_stream(fixed_clean_name), exit_after=0) + assert run(project_root=tmp_path) is True + + +def test_drained_tail_without_trigger_is_false(fake_claude, fixed_clean_name, tmp_path): + install, _ = fake_claude + install(stream(tool_start("Bash"), tool_delta('{"command":"ls"}'), block_stop(), result_event()), exit_after=0) + assert run(project_root=tmp_path) is False + + +# --------------------------------------------------------------------------- +# Streaming branch still detects triggers +# --------------------------------------------------------------------------- + + +def test_streaming_chunks_are_parsed(fake_claude, fixed_clean_name, tmp_path): + install, _ = fake_claude + install(skill_call_stream(fixed_clean_name), exit_after=1) + assert run(project_root=tmp_path) is True + + +# --------------------------------------------------------------------------- +# Non-skill first tool call is not a verdict (#1559-1) +# --------------------------------------------------------------------------- + + +def test_non_skill_first_tool_call_keeps_reading(fake_claude, fixed_clean_name, tmp_path): + """The model opens with a Bash orientation call (git status), then + calls the skill. Old code returned False on the Bash tool_use.""" + install, _ = fake_claude + before = [ + tool_start("Bash", tool_id="toolu_bash"), + tool_delta('{"command":"git status"}'), + block_stop(), + ] + install(skill_stream(fixed_clean_name, before=before), exit_after=0) + assert run(project_root=tmp_path) is True + + +def test_read_orientation_then_skill(fake_claude, fixed_clean_name, tmp_path): + """A Read call before the skill must not end the run either.""" + install, _ = fake_claude + before = [ + tool_start("Read", tool_id="toolu_read"), + tool_delta('{"file_path":"CLAUDE.md"}'), + block_stop(), + ] + install(skill_stream(fixed_clean_name, before=before), exit_after=0) + assert run(project_root=tmp_path) is True + + +# --------------------------------------------------------------------------- +# A different skill completing first is not a verdict (#1559-2) +# --------------------------------------------------------------------------- + + +def test_other_skill_completing_first_keeps_reading(fake_claude, fixed_clean_name, tmp_path): + """Another Skill block starts and completes before the target fires. + Old code returned a verdict as soon as any Skill block completed.""" + install, _ = fake_claude + install(skill_stream(fixed_clean_name, other_first=True), exit_after=0) + assert run(project_root=tmp_path) is True + + +def test_message_stop_before_trigger_keeps_reading(fake_claude, fixed_clean_name, tmp_path): + """A turn boundary (message_stop) before the skill fires must not be + treated as a negative verdict.""" + install, _ = fake_claude + before = [ + tool_start("Bash", tool_id="toolu_bash"), + tool_delta('{"command":"pwd"}'), + block_stop(), + message_stop(), + ] + install(skill_stream(fixed_clean_name, before=before), exit_after=0) + assert run(project_root=tmp_path) is True + + +# --------------------------------------------------------------------------- +# Per-run commands directory isolation (#1552-2 / #1559-3) +# --------------------------------------------------------------------------- + + +def test_command_file_lives_in_per_run_root(fake_claude, fixed_clean_name, tmp_path): + install, captured = fake_claude + install(skill_stream(fixed_clean_name), exit_after=0) + + assert run(project_root=tmp_path) is True + + cwd = Path(captured["kwargs"]["cwd"]) + # The run happened in its own .eval-runs/ root, not the project's + # shared .claude/commands/ directory, and that root held exactly the + # run's own command file. + assert cwd.parent.name == ".eval-runs" + assert cwd.name == "deadbeef" + assert captured["command_files"] == [f"{fixed_clean_name}.md"] + assert not (tmp_path / ".claude").exists() + + +def test_run_root_cleaned_up(fake_claude, fixed_clean_name, tmp_path): + install, _ = fake_claude + install(skill_stream(fixed_clean_name), exit_after=0) + assert run(project_root=tmp_path) is True + eval_runs = tmp_path / ".eval-runs" + if eval_runs.exists(): + assert not any(eval_runs.iterdir()), "run roots must be removed after each query" + + +def test_two_runs_do_not_share_a_commands_dir(fake_claude, fixed_clean_name, monkeypatch, tmp_path): + """Sequential runs each get their own root; the second must not see + the first's command file (parallel workers use the same path).""" + install, captured = fake_claude + install(skill_stream(fixed_clean_name), exit_after=0) + assert run(project_root=tmp_path) is True + first_cwd = Path(captured["kwargs"]["cwd"]) + + # Second run: a fresh uuid makes a fresh clean_name. + install(skill_stream("my-skill-skill-0badf00d"), exit_after=0) + monkeypatch.setattr( + run_eval.uuid, "uuid4", lambda: SimpleNamespace(hex="0badf00d12345678") + ) + assert run(project_root=tmp_path) is True + second_cwd = Path(captured["kwargs"]["cwd"]) + + assert first_cwd != second_cwd + # The first run's command file was removed by its own cleanup. + assert not (tmp_path / ".eval-runs").exists() or not any((tmp_path / ".eval-runs").iterdir())