diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index da783dba..2a96f65b 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -5,6 +5,7 @@ on: tags: ["v*"] permissions: + contents: write id-token: write jobs: @@ -19,3 +20,10 @@ jobs: - run: pip install build - run: python -m build - uses: pypa/gh-action-pypi-publish@release/v1 + + - name: Create GitHub Release + uses: softprops/action-gh-release@v2 + with: + generate_release_notes: true + draft: false + prerelease: ${{ contains(github.ref, '-') }} diff --git a/tests.json b/tests.json index 0f0bd9b4..27021611 100644 --- a/tests.json +++ b/tests.json @@ -59,7 +59,7 @@ "user_prompt": "Change the default port from 8000 to 9000 in both server.py and config.py", "setup": { "files": { - "server.py": "from config import PORT\n\ndef run():\n print(f'Listening on port {PORT}')\n", + "server.py": "import socket\n\ndef run():\n sock = socket.socket()\n sock.bind(('localhost', 8000))\n print('Server running on port 8000')\n", "config.py": "PORT = 8000\nHOST = 'localhost'\n" } }, @@ -126,7 +126,7 @@ "app.py": "import sqlite3\nfrom flask import Flask, jsonify\n\napp = Flask(__name__)\nDB = 'data.db'\n\ndef get_db():\n return sqlite3.connect(DB)\n\n@app.route('/users')\ndef list_users():\n db = get_db()\n users = db.execute('SELECT * FROM users').fetchall()\n db.close()\n return jsonify(users)\n\n@app.route('/users/')\ndef get_user(uid):\n db = get_db()\n user = db.execute('SELECT * FROM users WHERE id=?', (uid,)).fetchone()\n db.close()\n return jsonify(user)\n\nif __name__ == '__main__':\n app.run(port=8000)\n" } }, - "expected_actions": [{ "tool": "plan" }], + "expected_actions": [{ "tool": "create_plan" }], "match_mode": "subset" }, { @@ -175,6 +175,24 @@ { "tool": "man", "args_pattern": { "page": "tar" } } ], "match_mode": "subset" + }, + { + "id": "math-calculation", + "description": "Use the math tool for precise calculations, not bash or mental math", + "user_prompt": "What is 2^64 - 1? Use the math tool to calculate it precisely.", + "expected_actions": [ + { "tool": "math", "args_pattern": { "code": "2.*64" } } + ], + "match_mode": "subset" + }, + { + "id": "web-search-query", + "description": "Use web_search for general knowledge lookups, not web_fetch", + "user_prompt": "Search the web for the current population of Tokyo", + "expected_actions": [ + { "tool": "web_search", "args_pattern": { "query": "Tokyo" } } + ], + "match_mode": "subset" } ] } diff --git a/tests/test_session.py b/tests/test_session.py index 89eaff65..62079734 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -203,8 +203,8 @@ class TestPlanExec: "id": tc_id, "type": "function", "function": { - "name": "plan", - "arguments": json.dumps({"prompt": prior_prompt}), + "name": "create_plan", + "arguments": json.dumps({"goal": prior_prompt}), }, } ], @@ -239,7 +239,7 @@ class TestPlanExec: m for m in messages if m["role"] == "assistant" and m.get("tool_calls") ] assert len(assistant_with_tc) == 1 - assert assistant_with_tc[0]["tool_calls"][0]["function"]["name"] == "plan" + assert assistant_with_tc[0]["tool_calls"][0]["function"]["name"] == "create_plan" # The real tool result is forwarded with its original content tool_msgs = [m for m in messages if m["role"] == "tool"] diff --git a/tests/test_tools_schema.py b/tests/test_tools_schema.py index da8b3ffe..8787b745 100644 --- a/tests/test_tools_schema.py +++ b/tests/test_tools_schema.py @@ -97,7 +97,7 @@ class TestToolsMetadata: "web_fetch": "url", "web_search": "query", "task": "prompt", - "plan": "prompt", + "create_plan": "goal", "remember": "key", "recall": "query", "forget": "key", diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 92cf82e0..0fb99d19 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -547,13 +547,15 @@ class ChatSession: " write_file(path='hello.py', content='...')\n\n" "Find something across files → search:\n" " search(query='test_')\n\n" - "Complex or multi-step task → plan first:\n" - " plan(prompt='refactor database from API')\n\n" + "Plan, design, or think through an approach → create_plan:\n" + " create_plan(goal='refactor database from API')\n\n" "Run a command, git, or tests → bash:\n" " bash(command='git log -5')\n" " bash(command='pytest')\n\n" "Retrieve a URL → web_fetch:\n" " web_fetch(url='https://example.com')\n\n" + "Search the web for information → web_search:\n" + " web_search(query='current population of Tokyo')\n\n" "Look up documentation → man:\n" " man(page='tar')", ] @@ -1464,7 +1466,7 @@ class ChatSession: # Post-plan gate: prompt user on main thread after plan completes for i, item in enumerate(items): if ( - item.get("func_name") == "plan" + item.get("func_name") == "create_plan" and not item.get("error") and not item.get("denied") and not self.auto_approve @@ -1544,7 +1546,7 @@ class ChatSession: "web_search": self._prepare_web_search, "tool_search": self._prepare_tool_search, "task": self._prepare_task, - "plan": self._prepare_plan, + "create_plan": self._prepare_plan, "remember": self._prepare_remember, "recall": self._prepare_recall, "forget": self._prepare_forget, @@ -2094,26 +2096,26 @@ class ChatSession: def _prepare_plan(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]: """Prepare a planning agent for approval.""" - prompt = (args.get("prompt") or "").strip() - if not prompt: + goal = args.get("goal", "").strip() + if not goal: return { "call_id": call_id, - "func_name": "plan", - "header": "\u2717 plan: empty prompt", + "func_name": "create_plan", + "header": "\u2717 create_plan: empty goal", "preview": "", "needs_approval": False, - "error": "Error: empty prompt", + "error": "Error: empty goal", } - preview_text = prompt[:300] + ("..." if len(prompt) > 300 else "") + preview_text = goal[:300] + ("..." if len(goal) > 300 else "") return { "call_id": call_id, - "func_name": "plan", - "header": "\u2699 plan (planning agent)", + "func_name": "create_plan", + "header": "\u2699 create_plan (planning agent)", "preview": f" {DIM}{preview_text}{RESET}", "needs_approval": True, - "approval_label": "plan", + "approval_label": "create_plan", "execute": self._exec_plan, - "prompt": prompt, + "prompt": goal, } def _prepare_remember(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]: @@ -2577,7 +2579,7 @@ class ChatSession: tool_name = tc_dict["function"]["name"] # Guard 1: block recursive agent calls. - if tool_name in ("task", "plan"): + if tool_name in ("task", "create_plan"): output = "Error: agents cannot spawn further agents" # Guard 2: tool not in this agent's API tool list. elif tool_name not in tool_names: @@ -2698,7 +2700,7 @@ class ChatSession: for i, msg in enumerate(self.messages): if msg.get("role") == "assistant" and msg.get("tool_calls"): for tc in msg["tool_calls"]: - if tc.get("function", {}).get("name") == "plan": + if tc.get("function", {}).get("name") == "create_plan": tc_id = tc["id"] for j in range(i + 1, len(self.messages)): if ( diff --git a/turnstone/eval.py b/turnstone/eval.py index 9a651082..d492c100 100644 --- a/turnstone/eval.py +++ b/turnstone/eval.py @@ -16,6 +16,7 @@ import contextlib import difflib import io import json +import math import os import re import shutil @@ -24,16 +25,36 @@ import tempfile import textwrap import time from collections.abc import Iterator +from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed +from concurrent.futures import TimeoutError as FuturesTimeoutError from datetime import datetime from typing import Any from openai import OpenAI -from turnstone.core.providers import LLMProvider, create_provider +from turnstone.core.providers import LLMProvider, create_client, create_provider from turnstone.core.session import ChatSession from turnstone.core.storage import init_storage, reset_storage from turnstone.core.tools import PRIMARY_KEY_MAP, TOOLS +# ─── Provider auto-detection ────────────────────────────────────────────────── + + +def _detect_provider(base_url: str) -> str: + """Infer provider name from a base URL.""" + if "anthropic.com" in base_url: + return "anthropic" + return "openai" + + +def _make_client_and_provider(base_url: str, api_key: str) -> tuple[Any, LLMProvider]: + """Create a client + provider pair, auto-detecting the provider.""" + name = _detect_provider(base_url) + client = create_client(name, base_url=base_url, api_key=api_key) + provider = create_provider(name) + return client, provider + + # ─── ANSI & logging helpers ─────────────────────────────────────────────────── DIM = "\033[2m" @@ -174,16 +195,22 @@ class HeadlessSession(ChatSession): ) self.tool_call_log: list[dict[str, Any]] = [] self.auto_approve = True + self._total_usage: dict[str, int] = {"prompt": 0, "completion": 0} if system_prompt_override is not None: self._override_system_prompt(system_prompt_override) + @property + def total_usage(self) -> dict[str, int]: + """Accumulated token usage across all turns.""" + return self._total_usage + def _override_system_prompt(self, content: str) -> None: - """Replace the developer message content with a custom prompt.""" + """Replace the system/developer message content with a custom prompt.""" for i, msg in enumerate(self.system_messages): - if msg["role"] == "developer": - self.system_messages[i] = {"role": "developer", "content": content} + if msg["role"] in ("developer", "system"): + self.system_messages[i] = {"role": msg["role"], "content": content} return - self.system_messages.append({"role": "developer", "content": content}) + self.system_messages.append({"role": "system", "content": content}) def send_headless( self, @@ -223,6 +250,10 @@ class HeadlessSession(ChatSession): ) elapsed = time.monotonic() - t0 + if result.usage: + self._total_usage["prompt"] += result.usage.prompt_tokens + self._total_usage["completion"] += result.usage.completion_tokens + assistant_msg: dict[str, Any] = { "role": "assistant", "content": result.content or None, @@ -332,12 +363,15 @@ def _run_single_test( context_window: int, verbose: bool = False, log_prefix: str = "", + test_timeout: int = 300, ) -> dict[str, Any]: """Run a single test case once in an isolated temp directory. - Must be called serially — uses os.chdir which is process-global. + Uses os.chdir (process-global), so concurrent calls must run in + separate processes (see _run_and_score_subprocess / --parallel). - Returns dict with keys: tool_log, final_content, message_count, elapsed. + Returns dict with keys: tool_log, final_content, message_count, + elapsed, usage. """ workdir = tempfile.mkdtemp(prefix="turnstone_eval_") original_cwd = os.getcwd() @@ -370,27 +404,50 @@ def _run_single_test( tool_timeout=30, reasoning_effort=reasoning_effort, context_window=context_window, + tool_truncation=2000, ) - max_turns = case.get("max_turns", 10) + max_turns = case.get("max_turns", 15) # Retry on transient API errors to avoid poisoning eval scores tool_log: list[dict[str, Any]] = [] _last_err: Exception | None = None for _attempt in range(3): + executor: ThreadPoolExecutor | None = None try: - tool_log = session.send_headless( + executor = ThreadPoolExecutor(max_workers=1) + future = executor.submit( + session.send_headless, case["user_prompt"], max_turns=max_turns, verbose=verbose, log_prefix=log_prefix, ) + try: + tool_log = future.result(timeout=test_timeout) + except FuturesTimeoutError: + # shutdown(wait=False) lets the TimeoutError propagate + # immediately instead of blocking until the thread finishes. + # The orphaned thread will eventually exit on its own. + # Note: threads cannot be force-killed in CPython. In the + # parallel path this is moot since each test runs in a + # subprocess that can be terminated. The serial path + # accepts the leak as a trade-off for simpler code. + executor.shutdown(wait=False, cancel_futures=True) + executor = None + raise TimeoutError(f"Test timed out after {test_timeout}s") from None + else: + executor.shutdown(wait=False) + executor = None break + except TimeoutError: + raise except Exception as _e: _last_err = _e if _attempt < 2: - import time as _time - - _time.sleep(2**_attempt) + time.sleep(2**_attempt) + finally: + if executor is not None: + executor.shutdown(wait=False) else: raise _last_err or RuntimeError("send_headless failed after 3 attempts") @@ -406,6 +463,7 @@ def _run_single_test( "final_content": final_content, "message_count": len(session.messages), "elapsed": round(elapsed, 1), + "usage": session.total_usage, } finally: reset_storage() @@ -413,6 +471,79 @@ def _run_single_test( shutil.rmtree(workdir, ignore_errors=True) +def _run_and_score_subprocess(params: dict[str, Any]) -> dict[str, Any]: + """Subprocess worker for parallel test execution. + + Creates its own OpenAI client (the parent's is not picklable), + runs a single test case once, scores it, and returns a serializable + result dict. Must live at module scope for ProcessPoolExecutor. + """ + client = OpenAI( + base_url=params["base_url"], + api_key=params["api_key"], + ) + case: dict[str, Any] = params["case"] + run_tokens = 0 + + try: + run_result = _run_single_test( + client=client, + model=params["model"], + system_prompt=params["system_prompt"], + case=case, + temperature=params["temperature"], + max_tokens=params["max_tokens"], + reasoning_effort=params["reasoning_effort"], + context_window=params["context_window"], + verbose=False, + log_prefix="", + test_timeout=params["test_timeout"], + ) + + score_result = score_run( + tool_log=run_result["tool_log"], + expected_actions=case.get("expected_actions", []), + match_mode=case.get("match_mode", "ordered_subset"), + ) + + score_result["tool_sequence"] = [t["tool"] for t in run_result["tool_log"]] + score_result["tool_args"] = [{t["tool"]: t["args"]} for t in run_result["tool_log"]] + score_result["elapsed"] = run_result.get("elapsed", 0) + run_tokens = sum(run_result.get("usage", {}).values()) + + # Detect JSON dumped into final channel (tool call not made) + fc = (run_result.get("final_content") or "").strip() + if fc and not score_result["pass"]: + has_json = bool( + re.search( + r'\{\s*"(tool|function|name|arguments|command|query|path|url)"', + fc, + ) + ) + if has_json: + score_result["json_dump"] = True + + except Exception as e: + score_result = { + "pass": False, + "score": 0.0, + "matched": [], + "unmatched": list(range(len(case.get("expected_actions", [])))), + "extra_tools": [], + "detail": f"Error: {e}", + "tool_sequence": [], + "tool_args": [], + "elapsed": 0, + } + + return { + "case_id": params["case_id"], + "run_idx": params["run_idx"], + "score_result": score_result, + "run_tokens": run_tokens, + } + + # ─── Scoring ───────────────────────────────────────────────────────────────── @@ -561,6 +692,186 @@ def score_run( # ─── Iteration runner ──────────────────────────────────────────────────────── +def _format_eta(seconds: float) -> str: + """Format seconds as a human-readable ETA string.""" + if seconds <= 0: + return "" + if seconds < 60: + return f"~{seconds:.0f}s left" + m, s = divmod(int(seconds), 60) + if m < 60: + return f"~{m}m{s}s left" + h, m = divmod(m, 60) + return f"~{h}h{m}m left" + + +def _aggregate_case_results( + cases: list[dict[str, Any]], + case_results: dict[str, Any], + total_tokens: int, +) -> dict[str, Any]: + """Build the iteration result dict from per-case results.""" + agg_total_runs = sum(len(cr["runs"]) for cr in case_results.values()) + agg_total_passes = sum(sum(1 for r in cr["runs"] if r["pass"]) for cr in case_results.values()) + total_json_dumps = sum( + sum(1 for r in cr["runs"] if r.get("json_dump")) for cr in case_results.values() + ) + return { + "cases": case_results, + "aggregate": { + "total_cases": len(cases), + "total_runs": agg_total_runs, + "overall_pass_rate": agg_total_passes / agg_total_runs if agg_total_runs else 0, + "json_dumps": total_json_dumps, + "overall_avg_score": ( + sum(cr["avg_score"] for cr in case_results.values()) / len(case_results) + if case_results + else 0 + ), + "per_case_pass_rates": {cid: cr["pass_rate"] for cid, cr in case_results.items()}, + "total_tokens": total_tokens, + }, + } + + +def _run_iteration_parallel( + base_url: str, + api_key: str, + model: str, + system_prompt: str, + cases: list[dict[str, Any]], + n_runs: int, + temperature: float, + max_tokens: int, + reasoning_effort: str, + context_window: int, + test_timeout: int, + parallel: int, +) -> dict[str, Any]: + """Run all test cases in parallel using ProcessPoolExecutor.""" + # Build work items for every (case, run) combination + work_items: list[dict[str, Any]] = [] + for case in cases: + case_id = case["id"] + case_n = case.get("n_runs", n_runs) + for run_idx in range(case_n): + work_items.append( + { + "base_url": base_url, + "api_key": api_key, + "model": model, + "system_prompt": system_prompt, + "case": case, + "case_id": case_id, + "run_idx": run_idx, + "temperature": temperature, + "max_tokens": max_tokens, + "reasoning_effort": reasoning_effort, + "context_window": context_window, + "test_timeout": test_timeout, + } + ) + + total_planned = len(work_items) + max_workers = min(parallel, total_planned) if parallel > 0 else total_planned + max_workers = max(max_workers, 1) + + print(f"\n Running {total_planned} tests across {max_workers} workers...") + + # Collect results grouped by case_id (preserving case order) + case_runs: dict[str, list[tuple[int, dict[str, Any]]]] = {case["id"]: [] for case in cases} + completed = 0 + total_passes = 0 + total_tokens = 0 + t0 = time.monotonic() + + with ProcessPoolExecutor(max_workers=max_workers) as pool: + futures = {pool.submit(_run_and_score_subprocess, item): item for item in work_items} + + for future in as_completed(futures): + try: + # Workers enforce their own test_timeout internally; + # this outer timeout catches process-level hangs. + result = future.result(timeout=test_timeout + 30) + except Exception as exc: + item = futures[future] + result = { + "case_id": item["case_id"], + "run_idx": item["run_idx"], + "score_result": { + "pass": False, + "score": 0.0, + "matched": [], + "unmatched": list(range(len(item["case"].get("expected_actions", [])))), + "extra_tools": [], + "detail": f"Subprocess error: {exc}", + "tool_sequence": [], + "tool_args": [], + "elapsed": 0, + }, + "run_tokens": 0, + } + cid = result["case_id"] + ridx = result["run_idx"] + sr = result["score_result"] + tok = result["run_tokens"] + + case_runs[cid].append((ridx, sr)) + + # Progress + completed += 1 + if sr.get("pass"): + total_passes += 1 + total_tokens += tok + + passed = sr["pass"] + sc = GREEN if passed else RED + sl = "PASS" if passed else "FAIL" + tools = sr.get("tool_sequence", []) + elapsed = sr.get("elapsed", 0) + jf = f" {YELLOW}[JSON_DUMP]{RESET}" if sr.get("json_dump") else "" + + wall = time.monotonic() - t0 + avg = wall / completed + remaining = total_planned - completed + eta = _format_eta(avg * remaining) if remaining > 0 else "" + rate = total_passes / completed + tok_k = total_tokens / 1000 + + prog = f"[{completed}/{total_planned} | {rate:.0%}" + if total_tokens > 0: + prog += f" | {tok_k:.0f}k tok" + if eta: + prog += f" | {eta}" + prog += "]" + + print( + f" {DIM}{cid}{RESET} run {ridx + 1}: " + f"{sc}[{sl}]{RESET} " + f"score={sr['score']:.2f} " + f"tools={tools}" + f"{jf}" + f" {DIM}({elapsed:.1f}s) {prog}{RESET}" + ) + + # Build case_results in original case order + case_results: dict[str, Any] = {} + for case in cases: + cid = case["id"] + # Sort by run_idx so results are in order + sorted_runs = [sr for _, sr in sorted(case_runs[cid], key=lambda x: x[0])] + pass_count = sum(1 for r in sorted_runs if r["pass"]) + case_results[cid] = { + "runs": sorted_runs, + "pass_rate": pass_count / len(sorted_runs) if sorted_runs else 0, + "avg_score": ( + sum(r["score"] for r in sorted_runs) / len(sorted_runs) if sorted_runs else 0 + ), + } + + return _aggregate_case_results(cases, case_results, total_tokens) + + def _run_iteration( client: Any, model: str, @@ -572,10 +883,37 @@ def _run_iteration( reasoning_effort: str, context_window: int, verbose: bool = False, + test_timeout: int = 300, + fast_fail: bool = True, + parallel: int = 1, + base_url: str = "", + api_key: str = "", ) -> dict[str, Any]: """Run all test cases n_runs times and score them.""" + if parallel > 1 and base_url: + return _run_iteration_parallel( + base_url=base_url, + api_key=api_key, + model=model, + system_prompt=system_prompt, + cases=cases, + n_runs=n_runs, + temperature=temperature, + max_tokens=max_tokens, + reasoning_effort=reasoning_effort, + context_window=context_window, + test_timeout=test_timeout, + parallel=parallel, + ) + case_results: dict[str, Any] = {} + total_planned_runs = sum(c.get("n_runs", n_runs) for c in cases) + completed_runs = 0 + iter_total_passes = 0 + total_tokens = 0 + iter_t0 = time.monotonic() + for ci, case in enumerate(cases): case_id = case["id"] case_n = case.get("n_runs", n_runs) @@ -587,6 +925,7 @@ def _run_iteration( for run_idx in range(case_n): log_prefix = f" [{run_idx + 1}/{case_n}]" + run_tokens = 0 try: run_result = _run_single_test( @@ -600,6 +939,7 @@ def _run_iteration( context_window=context_window, verbose=verbose, log_prefix=log_prefix, + test_timeout=test_timeout, ) score_result = score_run( @@ -611,6 +951,7 @@ def _run_iteration( score_result["tool_sequence"] = [t["tool"] for t in run_result["tool_log"]] score_result["tool_args"] = [{t["tool"]: t["args"]} for t in run_result["tool_log"]] score_result["elapsed"] = run_result.get("elapsed", 0) + run_tokens = sum(run_result.get("usage", {}).values()) # Detect JSON dumped into final channel (tool call not made) fc = (run_result.get("final_content") or "").strip() @@ -633,8 +974,10 @@ def _run_iteration( "extra_tools": [], "detail": f"Error: {e}", "tool_sequence": [], + "tool_args": [], "elapsed": 0, } + run_tokens = 0 passed = score_result["pass"] status_color = GREEN if passed else RED @@ -653,6 +996,60 @@ def _run_iteration( runs.append(score_result) + # Update progress counters + completed_runs += 1 + if score_result.get("pass"): + iter_total_passes += 1 + total_tokens += run_tokens + + # Progress stats + iter_elapsed = time.monotonic() - iter_t0 + avg_per_run = iter_elapsed / completed_runs if completed_runs else 0 + remaining = total_planned_runs - completed_runs + eta = _format_eta(avg_per_run * remaining) if remaining > 0 else "" + running_rate = iter_total_passes / completed_runs if completed_runs else 0 + tok_k = total_tokens / 1000 + + progress = f" {DIM}[{completed_runs}/{total_planned_runs} | {running_rate:.0%}" + if total_tokens > 0: + progress += f" | {tok_k:.0f}k tok" + if eta: + progress += f" | {eta}" + progress += f"]{RESET}" + print(progress) + + # Fast-fail: skip remaining runs if first ceil(n/2) all score 0.0 + if ( + fast_fail + and len(runs) >= math.ceil(case_n / 2) + and all(r["score"] == 0.0 for r in runs) + ): + skipped = case_n - len(runs) + if skipped > 0: + _log( + f" {DIM}(skipped {skipped} remaining" + f" run{'s' if skipped != 1 else ''}" + f" — all 0.0){RESET}", + dim=True, + ) + for _ in range(skipped): + runs.append( + { + "pass": False, + "score": 0.0, + "matched": [], + "unmatched": list(range(len(case.get("expected_actions", [])))), + "extra_tools": [], + "detail": "Skipped (fast-fail)", + "tool_sequence": [], + "tool_args": [], + "elapsed": 0, + "skipped": True, + } + ) + completed_runs += skipped + break + pass_count = sum(1 for r in runs if r["pass"]) case_results[case_id] = { "runs": runs, @@ -660,78 +1057,105 @@ def _run_iteration( "avg_score": (sum(r["score"] for r in runs) / len(runs) if runs else 0), } - # Aggregate - total_runs = sum(len(cr["runs"]) for cr in case_results.values()) - total_passes = sum(sum(1 for r in cr["runs"] if r["pass"]) for cr in case_results.values()) - total_json_dumps = sum( - sum(1 for r in cr["runs"] if r.get("json_dump")) for cr in case_results.values() - ) - - return { - "cases": case_results, - "aggregate": { - "total_cases": len(cases), - "total_runs": total_runs, - "overall_pass_rate": total_passes / total_runs if total_runs else 0, - "json_dumps": total_json_dumps, - "overall_avg_score": ( - sum(cr["avg_score"] for cr in case_results.values()) / len(case_results) - if case_results - else 0 - ), - "per_case_pass_rates": {cid: cr["pass_rate"] for cid, cr in case_results.items()}, - }, - } + return _aggregate_case_results(cases, case_results, total_tokens) # ─── Prompt optimizer ──────────────────────────────────────────────────────── OPTIMIZER_SYSTEM = """\ -You are a text rewriter. You receive a developer prompt (instructions \ +You are a prompt optimizer. You receive a developer prompt (instructions \ for a coding assistant on how to use its tools) and test results \ -showing how well the assistant followed them. Rewrite the prompt so \ -the assistant picks the right tools more often. +showing how well the assistant followed them. -Context: Tests score whether the assistant calls specific tools in \ -the right order. The critical failure modes to address: \ -(1) responding with only text when a tool call is needed — the \ -assistant must ALWAYS call a tool, (2) using write_file to rewrite \ -an entire file instead of edit_file for small changes, (3) not calling \ -plan(prompt='...') when asked to think through a complex task, \ -(4) searching for a file before creating it with write_file. When a \ -test shows 100%, preserve whatever phrasing drove that behavior. +GOAL: Make targeted, minimal changes so more tests pass. One fix per \ +failing test case — do NOT rewrite the whole prompt. -Style: direct imperative instructions organized with newlines. Short \ -sentences. Concrete tool call examples like bash(command='git log -5'). +KEEP / DISCARD RULE: +- Any phrasing associated with a 100% pass rate test: KEEP VERBATIM. +- For failing tests: diagnose why the assistant chose wrong, then add \ +or adjust the MINIMUM phrasing needed to fix that specific failure. +- If removing a line yields equal or better results, remove it. \ +Simpler is always better at equal scores. -Length: no longer than 130% of the original prompt's length. +WHAT YOU CAN CHANGE: +- Add imperative instructions for specific tool-choice scenarios. +- Reword unclear directives that cause the wrong tool to be selected. +- Add concrete examples like bash(command='git log -5'). -Output ONLY the rewritten prompt. No commentary, no fences.\ +WHAT YOU CANNOT CHANGE: +- The list of available tools or their schemas. +- The overall structure (system prompt for a coding assistant). +- Phrasing tied to 100% pass rate cases. + +FAILURE MODE DIAGNOSIS: +- If the assistant responded with only text (no tool call) → add a \ +rule: "ALWAYS call a tool. Never respond with only text." +- If write_file was used instead of edit_file for a small change → add: \ +"Use edit_file for modifying existing files. Only use write_file for \ +new files." +- If create_plan() was not called for a complex task → add: "When asked to \ +think through a problem, call create_plan(goal='...')." +- If the assistant searched for a file before creating it → add: \ +"When told to create a new file, use write_file directly." +- If the tool sequence is correct but arguments are wrong → adjust \ +the example arguments, not the tool selection logic. + +STYLE: direct imperative sentences. One instruction per line. \ +Concrete tool call examples where helpful. + +LENGTH: no longer than 130% of the original prompt's length. + +Output ONLY the modified prompt. No commentary, no fences.\ """ OBSERVER_SYSTEM = """\ -You edit the rewriter's instructions shown below. The rewriter takes a \ -paragraph and test results, then rewrites the paragraph to score higher. \ -Your job: tune the rewriter's instructions so it does a better job. +You edit the optimizer's instructions shown below. The optimizer takes \ +a developer prompt and test results, then modifies the prompt to score \ +higher. Your job: tune the optimizer's instructions so it does a \ +better job. -Your output replaces the rewriter's instructions. It must stay at the \ -same level — telling the rewriter HOW to rewrite, not doing the \ -rewriting yourself. +Your output replaces the optimizer's instructions. It must stay at \ +the same level — telling the optimizer HOW to modify prompts, not \ +modifying prompts yourself. Example of the right level (abbreviated): \"\"\" -You are a text rewriter. You receive a paragraph and test results... -Style: flowing prose, no bullet points... -Length: aim for 600-1200 chars... -Be bold — reword, restructure... +You are a prompt optimizer. You receive a developer prompt and test \ +results... +GOAL: Make targeted, minimal changes so more tests pass... +KEEP / DISCARD RULE: Any phrasing tied to 100% pass rate: keep... +FAILURE MODE DIAGNOSIS: If text-only response → add "always call \ +a tool"... +STYLE: direct imperative sentences... \"\"\" -Make 2-3 targeted edits based on the iteration history. Remove guidance \ -that isn't working. Stay under 150% of the input length. +TREND ANALYSIS — look for these patterns: +- Improving: the optimizer's approach is working. Make small \ +refinements only. +- Plateau (same score for 2+ iterations): the optimizer is stuck. \ +Remove ineffective guidance and try a different strategy. +- Regression (score dropped): the last change hurt. Instruct the \ +optimizer to revert that type of change and try something else. +- Oscillation (score goes up/down): the optimizer is making changes \ +that are too broad. Tell it to make smaller, more targeted edits. -Output ONLY the modified rewriter instructions.\ +ESCALATION: If scores have not improved for 3+ iterations, instruct \ +the optimizer to try more radical approaches — restructure sections, \ +change the instruction style, or add/remove entire categories of \ +guidance. + +SCOPE — you can adjust: +- Which failure modes the optimizer should prioritize. +- Whether it should add examples, restructure, or simplify. +- Length and style guidance. +- Diagnosis-to-fix mappings. + +Make 2-3 targeted edits based on the iteration history. Remove \ +guidance that isn't working. Stay under 150% of the input length. + +Output ONLY the modified optimizer instructions.\ """ @@ -804,9 +1228,9 @@ def _observe_and_update_optimizer( behavior_notes.append(f"Iteration {idx} ({score:.0%}): {style}") user_content = ( - f"## Rewriter Instructions (edit these)\n" + f"## Optimizer Instructions (edit these)\n" f"```\n{optimizer_system}\n```\n\n" - f"## What the Rewriter Produced (do NOT mimic this)\n" + f"## What the Optimizer Produced (do NOT mimic this)\n" + "\n".join(behavior_notes) + "\n\n## Iteration History\n" + "\n".join(parts) @@ -889,12 +1313,27 @@ def _propose_prompt_modification( history_text = "\n".join(history_parts) if history_parts else "(first iteration)" - user_content = ( - f"## Current Prompt\n```\n{current_prompt}\n```\n\n" - f"## Test Results (iteration {iteration_result.get('iteration', '?')})\n" - + "\n\n".join(summary_parts) - + f"\n\n## Score History\n{history_text}" - + "\n\nPropose an improved prompt. Output ONLY the new prompt text." + # Separate passing vs failing cases for clearer diagnosis + passing = [p for p in summary_parts if p.startswith("[PASS]")] + failing = [p for p in summary_parts if not p.startswith("[PASS]")] + + user_content = f"## Current Prompt\n```\n{current_prompt}\n```\n\n" + if passing: + user_content += ( + "## Passing Tests (DO NOT change phrasing that drives these)\n" + + "\n\n".join(passing) + + "\n\n" + ) + if failing: + user_content += ( + "## Failing Tests (diagnose each, make ONE targeted fix per case)\n" + + "\n\n".join(failing) + + "\n\n" + ) + user_content += ( + f"## Score History\n{history_text}\n\n" + "Make the MINIMUM change needed to fix failing tests without " + "breaking passing ones. Output ONLY the modified prompt text." ) prov = provider or create_provider("openai") @@ -940,6 +1379,90 @@ def _simple_diff(old: str, new: str) -> str: return "".join(diff) +# ─── Summary & reporting ───────────────────────────────────────────────────── + + +def _print_summary_table(iter_result: dict[str, Any]) -> None: + """Print a formatted summary table for an iteration's results.""" + cases = iter_result.get("cases", {}) + if not cases: + return + + # Column widths + max_id = max(len(cid) for cid in cases) + max_id = max(max_id, 4) # min "CASE" header + + header = f" {'CASE'.ljust(max_id)} {'PASS':>5} {'AVG':>5} {'RUNS':>5} {'TIME':>6} STATUS" + print(f"\n{BOLD}{header}{RESET}") + print(f" {'─' * (max_id + 34)}") + + total_passes = 0 + total_runs_count = 0 + total_elapsed = 0.0 + + for case_id, cr in cases.items(): + runs = cr.get("runs", []) + passes = sum(1 for r in runs if r.get("pass")) + run_count = len(runs) + avg_score = cr.get("avg_score", 0) + elapsed = sum(r.get("elapsed", 0) for r in runs) + pass_rate = cr.get("pass_rate", 0) + + total_passes += passes + total_runs_count += run_count + total_elapsed += elapsed + + if pass_rate == 1.0: + status = f"{GREEN}PASS{RESET}" + elif pass_rate >= 0.5: + status = f"{YELLOW}WEAK{RESET}" + else: + status = f"{RED}FAIL{RESET}" + + print( + f" {case_id.ljust(max_id)} {pass_rate:>4.0%} {avg_score:>5.2f} " + f"{passes:>2}/{run_count:<2} {elapsed:>5.1f}s {status}" + ) + + # Totals + overall_pass_rate = total_passes / total_runs_count if total_runs_count else 0 + overall_avg = iter_result.get("aggregate", {}).get("overall_avg_score", 0) + json_dumps = iter_result.get("aggregate", {}).get("json_dumps", 0) + jd_str = f" {YELLOW}({json_dumps} json_dumps){RESET}" if json_dumps else "" + + print(f" {'─' * (max_id + 34)}") + print( + f" {BOLD}{'TOTAL'.ljust(max_id)}{RESET} {overall_pass_rate:>4.0%} " + f"{overall_avg:>5.2f} " + f"{total_passes:>2}/{total_runs_count:<2} " + f"{total_elapsed:>5.1f}s{jd_str}" + ) + + +def _append_summary_tsv(path: str, iter_result: dict[str, Any], case_ids: list[str]) -> None: + """Append one row per iteration to a TSV summary file.""" + write_header = not os.path.exists(path) or os.path.getsize(path) == 0 + agg = iter_result.get("aggregate", {}) + per_case = agg.get("per_case_pass_rates", {}) + + with open(path, "a") as f: + if write_header: + cols = ["iter", "timestamp", "pass_rate", "avg_score", "runs", "json_dumps"] + [ + f"case:{cid}" for cid in case_ids + ] + f.write("\t".join(cols) + "\n") + + vals = [ + str(iter_result.get("iteration", "")), + iter_result.get("timestamp", ""), + f"{agg.get('overall_pass_rate', 0):.2f}", + f"{agg.get('overall_avg_score', 0):.2f}", + str(agg.get("total_runs", 0)), + str(agg.get("json_dumps", 0)), + ] + [f"{per_case.get(cid, 0):.2f}" for cid in case_ids] + f.write("\t".join(vals) + "\n") + + # ─── Main optimization loop ───────────────────────────────────────────────── @@ -956,20 +1479,60 @@ def run_optimization( output_file: str = "eval_results.json", context_window: int = 131072, verbose: bool = False, + test_timeout: int = 300, + suite_timeout: int = 0, + fast_fail: bool = True, + parallel: int = 1, + optimizer_base_url: str | None = None, + optimizer_model: str | None = None, + observer_base_url: str | None = None, + observer_model: str | None = None, ) -> dict[str, Any]: - """Main optimization loop.""" + """Main optimization loop. + + Three separate model roles: + - test: the model being evaluated (base_url / model) + - optimizer: rewrites the developer prompt (--optimizer-*) + - observer: tunes the optimizer's strategy (--observer-*) + + Inheritance: observer defaults ← optimizer defaults ← test defaults. + Provider is auto-detected from the base URL. + """ + # --- Test model (always OpenAI-compatible for headless eval) --- + api_key = os.environ.get("OPENAI_API_KEY", "dummy") client = OpenAI( base_url=base_url, - api_key=os.environ.get("OPENAI_API_KEY", "dummy"), + api_key=api_key, ) - - eval_provider = create_provider("openai") - if not model: from turnstone.core.model_registry import detect_model model, _ = detect_model(client) + # --- Optimizer model (inherits from test if not specified) --- + opt_base = optimizer_base_url or base_url + opt_model = optimizer_model or model + opt_key_env = ( + "ANTHROPIC_API_KEY" if _detect_provider(opt_base) == "anthropic" else "OPENAI_API_KEY" + ) + opt_key = os.environ.get(opt_key_env, api_key) + opt_client, opt_provider = _make_client_and_provider(opt_base, opt_key) + + # --- Observer model (inherits from optimizer if not specified) --- + obs_base = observer_base_url or opt_base + obs_model = observer_model or opt_model + obs_key_env = ( + "ANTHROPIC_API_KEY" if _detect_provider(obs_base) == "anthropic" else "OPENAI_API_KEY" + ) + obs_key = os.environ.get(obs_key_env, opt_key) + obs_client, obs_provider = _make_client_and_provider(obs_base, obs_key) + + # Log role assignments if any differ from the test model + if opt_model != model or opt_base != base_url: + _log(f" Optimizer: {opt_model} @ {opt_base}", dim=True) + if obs_model != opt_model or obs_base != opt_base: + _log(f" Observer: {obs_model} @ {obs_base}", dim=True) + # Load test cases with open(test_file) as f: suite: dict[str, Any] = json.load(f) @@ -998,7 +1561,12 @@ def run_optimization( reasoning_effort=reasoning_effort, context_window=context_window, ) - initial_prompt = next(m["content"] for m in tmp.system_messages if m["role"] == "developer") + initial_prompt = next( + (m["content"] for m in tmp.system_messages if m["role"] in ("developer", "system")), + None, + ) + if initial_prompt is None: + raise SystemExit("No developer prompt found. Provide one with --prompt ") # Strip memory reminder — it's a runtime artifact, not part of the prompt initial_prompt = re.sub( r"\n*REMINDER: You currently have \d+ memories stored\..*$", @@ -1011,6 +1579,10 @@ def run_optimization( "meta": { "model": model, "base_url": base_url, + "optimizer_model": opt_model, + "optimizer_base_url": opt_base, + "observer_model": obs_model, + "observer_base_url": obs_base, "started": datetime.now().isoformat(), "test_suite": test_file, "n_runs_default": resolved_n_runs, @@ -1019,8 +1591,21 @@ def run_optimization( } current_optimizer_system = OPTIMIZER_SYSTEM + tsv_path = os.path.splitext(output_file)[0] + ".tsv" + case_ids = [c["id"] for c in cases] + + suite_t0 = time.monotonic() for iteration in range(max_iterations): + if suite_timeout > 0: + suite_elapsed = time.monotonic() - suite_t0 + if suite_elapsed >= suite_timeout: + print( + f"\nSuite timeout ({suite_timeout}s) reached" + f" after {suite_elapsed:.0f}s. Stopping." + ) + break + print(f"\n{'=' * 60}") print(f"Iteration {iteration}") print(f"{'=' * 60}") @@ -1036,6 +1621,11 @@ def run_optimization( reasoning_effort=reasoning_effort, context_window=context_window, verbose=verbose, + test_timeout=test_timeout, + fast_fail=fast_fail, + parallel=parallel, + base_url=base_url, + api_key=api_key, ) iter_result["iteration"] = iteration iter_result["prompt"] = current_prompt @@ -1049,17 +1639,11 @@ def run_optimization( with open(output_file, "w") as f: json.dump(results, f, indent=2) - # Print summary - agg = iter_result["aggregate"] - json_dumps = agg.get("json_dumps", 0) - jd_str = f" {YELLOW}json_dumps={json_dumps}{RESET}" if json_dumps else "" - print(f"\nOverall pass rate: {agg['overall_pass_rate']:.0%}{jd_str}") - print(f"Overall avg score: {agg['overall_avg_score']:.2f}") - for case_id, rate in agg["per_case_pass_rates"].items(): - status = "PASS" if rate == 1.0 else "FAIL" - print(f" [{status}] {case_id}: {rate:.0%}") + _print_summary_table(iter_result) + _append_summary_tsv(tsv_path, iter_result, case_ids) # Check if all passing + agg = iter_result["aggregate"] if agg["overall_pass_rate"] == 1.0: print("\nAll test cases passing! Stopping.") break @@ -1071,11 +1655,11 @@ def run_optimization( _log(" Observer updating optimizer prompt...", dim=True) try: new_opt = _observe_and_update_optimizer( - client, - model, + obs_client, + obs_model, current_optimizer_system, results["iterations"], - provider=eval_provider, + provider=obs_provider, ) if new_opt != current_optimizer_system: opt_diff = _simple_diff(current_optimizer_system, new_opt) @@ -1102,14 +1686,14 @@ def run_optimization( print("\nOptimizing prompt...") try: new_prompt = _propose_prompt_modification( - client=client, - model=model, + client=opt_client, + model=opt_model, current_prompt=current_prompt, test_cases=cases, iteration_result=iter_result, history=results["iterations"], optimizer_system=current_optimizer_system, - provider=eval_provider, + provider=opt_provider, ) except Exception as e: _log(f" Prompt modification failed: {e}", dim=True) @@ -1169,6 +1753,26 @@ def main() -> None: default=None, help="Model name (default: auto-detect)", ) + parser.add_argument( + "--optimizer-model", + default=None, + help="Model for prompt optimization (default: same as --model)", + ) + parser.add_argument( + "--optimizer-base-url", + default=None, + help="Base URL for optimizer model (default: same as --base-url)", + ) + parser.add_argument( + "--observer-model", + default=None, + help="Model for meta-optimization (default: same as --optimizer-model)", + ) + parser.add_argument( + "--observer-base-url", + default=None, + help="Base URL for observer model (default: same as --optimizer-base-url)", + ) parser.add_argument( "--prompt", default=None, @@ -1220,6 +1824,29 @@ def main() -> None: default="eval_results.json", help="Output results file (default: eval_results.json)", ) + parser.add_argument( + "--test-timeout", + type=int, + default=300, + help="Per-test timeout in seconds (default: 300)", + ) + parser.add_argument( + "--suite-timeout", + type=int, + default=0, + help="Total suite timeout in seconds (default: unlimited)", + ) + parser.add_argument( + "--no-fast-fail", + action="store_true", + help="Disable early termination when all initial runs score 0.0", + ) + parser.add_argument( + "--parallel", + type=int, + default=1, + help="Parallel workers (default: 1=serial, 0=auto)", + ) parser.add_argument( "-v", "--verbose", @@ -1252,6 +1879,14 @@ def main() -> None: output_file=args.output, context_window=args.context_window, verbose=args.verbose, + test_timeout=args.test_timeout, + suite_timeout=args.suite_timeout, + fast_fail=not args.no_fast_fail, + parallel=args.parallel if args.parallel != 0 else (os.cpu_count() or 4), + optimizer_base_url=args.optimizer_base_url, + optimizer_model=args.optimizer_model, + observer_base_url=args.observer_base_url, + observer_model=args.observer_model, ) diff --git a/turnstone/tools/create_plan.json b/turnstone/tools/create_plan.json new file mode 100644 index 00000000..de6696da --- /dev/null +++ b/turnstone/tools/create_plan.json @@ -0,0 +1,15 @@ +{ + "name": "create_plan", + "description": "Create a structured plan before taking action. An autonomous agent explores the available context, identifies what needs to change, and writes a step-by-step plan. Call this tool when the user asks to plan, design, or think through an approach, or when a task is complex, touches multiple areas, or has unclear scope.", + "parameters": { + "type": "object", + "properties": { + "goal": { + "type": "string", + "description": "The goal and scope of the plan, including any constraints." + } + }, + "required": ["goal"] + }, + "primary_key": "goal" +} diff --git a/turnstone/tools/plan.json b/turnstone/tools/plan.json deleted file mode 100644 index ea330762..00000000 --- a/turnstone/tools/plan.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "name": "plan", - "description": "Plan before implementing. An autonomous agent explores the codebase and writes a structured plan to .plan-.md (unique per workstream to avoid collisions). If a plan for this workstream already exists it is re-read and refined rather than overwritten from scratch. Use plan BEFORE writing code — when the user asks to build, add, refactor, or change something that touches multiple files or has unclear scope. The plan identifies files to modify, existing patterns to reuse, and risks to consider.", - "parameters": { - "type": "object", - "properties": { - "prompt": { - "type": "string", - "description": "What to plan — the goal, constraints, and scope." - } - }, - "required": ["prompt"] - }, - "primary_key": "prompt" -} diff --git a/turnstone/tools/web_fetch.json b/turnstone/tools/web_fetch.json index 2fda4df2..3a87b7b1 100644 --- a/turnstone/tools/web_fetch.json +++ b/turnstone/tools/web_fetch.json @@ -1,6 +1,6 @@ { "name": "web_fetch", - "description": "Fetch a URL and extract specific information from it. You must provide a question or extraction guidance — the page is fetched, analyzed, and only relevant information is returned (not raw page content).", + "description": "Fetch a URL and extract specific information from it. You must provide a question or extraction guidance. The page is fetched, analyzed, and only relevant information is returned (not raw page content).", "parameters": { "type": "object", "properties": {