Files
oh-my-pi/scripts/rate-edit-tool.py
T
can1357 c62ab2d53e chore(benchmarks-misc-fixes): cleaned benchmark task/pid validation
- Added `vim` as an edit variant in benchmark CLI/config and rating script coverage.
- Expanded benchmark execution so `vim` is treated as a mutation tool for retries, stats, and edit intent checks.
- Adjusted `TaskTool` output schema precedence so explicit params override agent frontmatter.
- Fixed `TaskTool` success counting by excluding aborted tasks from success totals.
- Improved validation guidance in `SubmitResultTool`/`TodoWriteTool` for clearer recovery when payloads are missing or invalid.
- Added background command PID regression coverage in `executeBash` to confirm a real, terminateable PID is returned.
2026-04-13 12:27:10 +02:00

1661 lines
57 KiB
Python
Executable File

#!/usr/bin/env python3
from __future__ import annotations
import argparse
import asyncio
import json
import os
import random
import shutil
import sys
import tempfile
import textwrap
import threading
import time
from dataclasses import asdict, dataclass, field, is_dataclass
from pathlib import Path
from typing import Any, TextIO
from rich import box
from rich.console import Console, Group
from rich.live import Live
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "python/omp-rpc/src"))
from omp_rpc import ( # noqa: E402
AgentEndEvent,
AutoRetryEndEvent,
AutoRetryStartEvent,
ExtensionUiRequest,
MessageEndEvent,
MessageUpdateEvent,
RpcClient,
RpcError,
RpcNotification,
TodoAutoClearEvent,
TodoItem,
TodoReminderEvent,
TodoPhase,
ToolExecutionEndEvent,
ToolExecutionStartEvent,
ToolExecutionUpdateEvent,
TurnEndEvent,
TurnStartEvent,
assistant_text,
)
MODELS = [
"openrouter/moonshotai/kimi-k2.5",
"openrouter/anthropic/claude-haiku-4.5",
"openrouter/anthropic/claude-sonnet-4.6",
"openrouter/google/gemini-3-flash-preview",
"openrouter/z-ai/glm-5-turbo",
"openrouter/minimax/minimax-m2.7",
]
ORACLE_MODEL = "openrouter/anthropic/claude-opus-4.6"
PROMPT = textwrap.dedent(
"""\
You are evaluating the code-reading and code-editing tools on files in this directory.
{FIXTURE_SURFACE}
Work in this order:
1. Map the surface area. Identify operations, selectors, and addressing modes that actually work. Note differences across file types.
2. Exercise the supported paths. Read: whole-file, structural chunks, nested members, line ranges, raw source. Edit/vim: replace, insert into containers, insert before/after, delete. On `main.md`, check how addressing differs from code files.
3. Push into awkward cases. Test first/last-child edits, indentation preservation, decorators, docstrings, enum variants, markdown tables, and fenced code blocks. Note whether error messages were clear and actionable.
4. Verify after edits. Re-read each file after meaningful edits and confirm no unintended changes.
Report concrete findings:
- what required workarounds
- what was impossible
- which errors were clear vs unclear
- what was ambiguous or under-documented
- suggested improvements
Be specific. Generic success summaries are not useful.
"""
).strip()
FINAL_REVIEW_PROMPT = textwrap.dedent(
"""\
Your prior turn completed without a final written review in the assistant text.
Write the final review now as markdown only.
Do not perform more tool calls.
Summarize the concrete findings from the work already completed:
- awkward workflows or required workarounds
- impossible operations
- clear vs unclear errors
- ambiguous or under-documented behavior
- changes that would improve trustworthiness and usability
"""
).strip()
ORACLE_REVIEW_PROMPT = textwrap.dedent(
"""\
<context>
You are the oracle reviewer for a tool-evaluation benchmark.
You are reading multiple independent reviews of the same edit-tool session.
Synthesize only what is supported by the supplied reviews.
This matters. Be specific and conservative.
</context>
<instructions>
1. Deduplicate overlapping observations into one finding.
2. Separate well-supported findings from weaker signals. When evidence is mixed or thin, lower confidence instead of overstating.
3. Cite which review models reported each finding and summarize the concrete evidence they observed.
4. End with practical improvement areas prioritized by expected trust and usability gains.
Output markdown only with this exact structure:
# Oracle synthesis
## Findings
- Finding: ...
- Confidence: High | Medium | Low
- Evidence: cite the models and the behavior they observed
- Improvement area: the concrete product or UX change this finding points to
## Improvement areas
1. ...
2. ...
If the reviews do not support any concrete finding, say so explicitly under `## Findings`.
</instructions>
<data>
{{REVIEWS}}
</data>
<instructions>
Use only the supplied reviews. Output markdown only.
</instructions>
"""
).strip()
TODOS = [
"Map the current read, edit, and vim surface area on main.ts, main.rs, main.py, and main.md.",
"Exercise supported read, edit, and vim paths with concrete before/after verification across code and prose fixtures.",
"Probe awkward selector, indentation, and boundary cases including decorators, docstrings, tables, and fenced blocks.",
"Summarize what was awkward, impossible, ambiguous, or under-documented with concrete examples.",
]
TS_FIXTURE = textwrap.dedent(
"""\
function sealed(_target: Function): void {}
function trace(_label: string) {
return function (
_target: object,
_propertyKey: string,
descriptor: PropertyDescriptor,
): PropertyDescriptor {
return descriptor;
};
}
/** Log severity levels emitted by the demo server. */
export enum LogLevel {
Debug = "DEBUG",
Info = "INFO",
Warn = "WARN",
Error = "ERROR",
}
/** Shared runtime configuration used by parsing and request handling. */
export interface Config {
host: string;
port: number;
logLevel: LogLevel;
tags: Record<string, string>;
}
export type Result<T> = { ok: true; value: T } | { ok: false; error: string };
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null;
}
/** Parse a small JSON config blob into a typed config object. */
export function parseConfig(raw: string): Result<Config> {
try {
const parsed: unknown = JSON.parse(raw);
if (!isRecord(parsed)) {
return { ok: false, error: "config must be an object" };
}
if (typeof parsed.host !== "string" || typeof parsed.port !== "number") {
return { ok: false, error: "missing required fields" };
}
return {
ok: true,
value: {
host: parsed.host,
port: parsed.port,
logLevel: (parsed.logLevel as LogLevel | undefined) ?? LogLevel.Info,
tags: (parsed.tags as Record<string, string> | undefined) ?? {},
},
};
} catch (error) {
return { ok: false, error: `parse failed: ${String(error)}` };
}
}
/** Tiny request handler surface with decorators and attached comments. */
@sealed
export class Server {
#config: Config;
#running = false;
#history: string[] = [];
constructor(config: Config) {
this.#config = config;
}
/** Begin serving requests. */
@trace("start")
start(): void {
if (this.#running) {
throw new Error("already running");
}
this.#running = true;
// Record the first lifecycle transition for later inspection.
this.#history.push(`started:${this.getAddress()}`);
}
stop(): void {
if (!this.#running) return;
this.#running = false;
this.#history.push("stopped");
}
isRunning(): boolean {
return this.#running;
}
getAddress(): string {
return `${this.#config.host}:${this.#config.port}`;
}
handleRequest(method: string, path: string): Result<string> {
if (!this.#running) {
return { ok: false, error: "server not running" };
}
switch (method) {
case "GET":
return { ok: true, value: `fetched ${path}` };
case "POST":
return { ok: true, value: `created ${path}` };
case "DELETE":
return { ok: true, value: `deleted ${path}` };
default:
return { ok: false, error: `unknown method: ${method}` };
}
}
history(): string[] {
return [...this.#history];
}
}
/** Format a timestamped log line. */
export function formatLog(level: LogLevel, message: string): string {
const timestamp = new Date().toISOString();
return `[${timestamp}] [${level}] ${message}`;
}
/** Generic fixed-size queue used to test nested members. */
export class RingBuffer<T> {
#items: T[] = [];
#capacity: number;
constructor(capacity: number) {
this.#capacity = capacity;
}
push(item: T): void {
if (this.#items.length >= this.#capacity) {
this.#items.shift();
}
this.#items.push(item);
}
peek(): T | undefined {
return this.#items[this.#items.length - 1];
}
toArray(): T[] {
return [...this.#items];
}
}
"""
).strip() + "\n"
RUST_FIXTURE = textwrap.dedent(
"""\
use std::collections::HashMap;
use std::fmt;
/// Log severity levels emitted by the demo server.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LogLevel {
Debug,
Info,
Warn,
Error,
}
impl fmt::Display for LogLevel {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
LogLevel::Debug => write!(f, "DEBUG"),
LogLevel::Info => write!(f, "INFO"),
LogLevel::Warn => write!(f, "WARN"),
LogLevel::Error => write!(f, "ERROR"),
}
}
}
/// Shared runtime configuration used by parsing and request handling.
#[derive(Debug, Clone)]
pub struct Config {
pub host: String,
pub port: u16,
pub log_level: LogLevel,
pub tags: HashMap<String, String>,
}
/// Result of handling a request.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum OpResult {
Ok(String),
Err(String),
}
/// Parse a `host:port` config string into a structured config.
#[must_use]
pub fn parse_config(raw: &str) -> Result<Config, String> {
let parts: Vec<&str> = raw.splitn(2, ':').collect();
if parts.len() != 2 {
return Err("expected host:port format".into());
}
let port = parts[1]
.parse::<u16>()
.map_err(|error| format!("bad port: {}", error))?;
Ok(Config {
host: parts[0].to_string(),
port,
log_level: LogLevel::Info,
tags: HashMap::new(),
})
}
/// Request handling surface for trait and impl edits.
pub trait Handler {
fn handle(&self, method: &str, path: &str) -> OpResult;
}
/// Small in-memory server used as an edit surface.
#[derive(Debug)]
pub struct Server {
config: Config,
running: bool,
history: Vec<String>,
}
impl Server {
#[inline]
pub fn new(config: Config) -> Self {
Self {
config,
running: false,
history: Vec::new(),
}
}
/// Begin serving requests.
pub fn start(&mut self) -> Result<(), String> {
if self.running {
return Err("already running".into());
}
self.running = true;
// Record the first lifecycle transition for later inspection.
self.history
.push(format!("started:{}", self.address()));
Ok(())
}
pub fn stop(&mut self) {
if !self.running {
return;
}
self.running = false;
self.history.push("stopped".into());
}
#[must_use]
pub fn is_running(&self) -> bool {
self.running
}
#[must_use]
pub fn address(&self) -> String {
format!("{}:{}", self.config.host, self.config.port)
}
#[must_use]
pub fn history(&self) -> &[String] {
&self.history
}
}
impl Handler for Server {
fn handle(&self, method: &str, path: &str) -> OpResult {
if !self.running {
return OpResult::Err("not running".into());
}
match method {
"GET" => OpResult::Ok(format!("fetched {}", path)),
"POST" => OpResult::Ok(format!("created {}", path)),
"DELETE" => OpResult::Ok(format!("deleted {}", path)),
_ => OpResult::Err(format!("unknown method: {}", method)),
}
}
}
/// Format a log line with its level prefix.
#[must_use]
pub fn format_log(level: LogLevel, message: &str) -> String {
format!("[{}] {}", level, message)
}
/// Fixed-size buffer used to test nested impl members.
#[allow(dead_code)]
pub struct RingBuffer {
items: Vec<String>,
capacity: usize,
}
impl RingBuffer {
pub fn new(capacity: usize) -> Self {
Self {
items: Vec::with_capacity(capacity),
capacity,
}
}
pub fn push(&mut self, item: String) {
if self.items.len() >= self.capacity {
self.items.remove(0);
}
self.items.push(item);
}
#[must_use]
pub fn peek(&self) -> Option<&str> {
self.items.last().map(|item| item.as_str())
}
#[must_use]
pub fn as_slice(&self) -> &[String] {
&self.items
}
}
"""
).strip() + "\n"
PYTHON_FIXTURE = textwrap.dedent(
"""\
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable
def traced(label: str):
def decorator(fn):
def wrapper(*args, **kwargs):
return fn(*args, **kwargs)
wrapper.__name__ = fn.__name__
wrapper.__doc__ = fn.__doc__
return wrapper
return decorator
@dataclass(slots=True)
class Config:
host: str
port: int
tags: dict[str, str] = field(default_factory=dict)
class Server:
# Small indentation-sensitive server surface for edit tests.
def __init__(self, config: Config) -> None:
self._config = config
self._history: list[str] = []
self._running = False
@traced("start")
def start(self) -> None:
if self._running:
raise RuntimeError("already running")
self._running = True
self._history.append(f"started:{self.address}")
def stop(self) -> None:
if not self._running:
return
self._running = False
self._history.append("stopped")
@property
def address(self) -> str:
return f"{self._config.host}:{self._config.port}"
def handle(self, method: str, path: str) -> str:
if not self._running:
raise RuntimeError("server not running")
match method:
case "GET":
return f"fetched {path}"
case "POST":
return f"created {path}"
case _:
raise ValueError(f"unknown method: {method}")
def history(self) -> list[str]:
return list(self._history)
def parse_config(raw: str) -> Config:
host, port = raw.split(":", 1)
return Config(host=host.strip(), port=int(port))
def write_report(lines: Iterable[str], target: Path) -> None:
target.write_text("\n".join(lines) + "\n", encoding="utf-8")
"""
).strip() + "\n"
MARKDOWN_FIXTURE = textwrap.dedent(
"""\
---
title: Tooling Evaluation Notes
owner: Fixtures Team
---
# Fixture Overview
This markdown fixture is intentionally prose-heavy.
It should help reveal how read and edit behave when the file is not routed through a language AST.
## Acceptance checklist
- [ ] Verify heading edits keep spacing intact.
- [ ] Verify list edits preserve indentation.
- [ ] Verify table edits do not destroy alignment beyond what the tool promises.
- [ ] Verify fenced code blocks remain fenced after edits.
## Comparison table
| Surface | Expected stress |
| --- | --- |
| `main.ts` | Structural chunk addressing |
| `main.rs` | Enum and impl member edits |
| `main.py` | Indentation-sensitive blocks |
| `main.md` | Prose and block-level text edits |
## Embedded examples
```python
def greet(name: str) -> str:
return f"hello {name}"
```
```json
{
"mode": "demo",
"strict": true
}
```
## Notes
1. Paragraph edits should preserve blank lines.
2. List insertions should not collapse into one paragraph.
3. Deleting this section should not damage the fenced blocks above.
"""
).strip() + "\n"
REFERENCE_FILES = {
"PROMPT.md": PROMPT + "\n",
"main.ts": TS_FIXTURE,
"main.rs": RUST_FIXTURE,
"main.py": PYTHON_FIXTURE,
"main.md": MARKDOWN_FIXTURE,
}
FIXTURES: tuple[tuple[str, str], ...] = (
("typescript", "main.ts"),
("rust", "main.rs"),
("python", "main.py"),
("markdown", "main.md"),
)
FIXTURE_DESCRIPTIONS: dict[str, str] = {
"typescript": "TypeScript/AST",
"rust": "Rust/AST",
"python": "indentation-sensitive",
"markdown": "prose/non-AST",
}
WORKSPACE_FILES = {
"main.ts": TS_FIXTURE,
"main.rs": RUST_FIXTURE,
"main.py": PYTHON_FIXTURE,
"main.md": MARKDOWN_FIXTURE,
}
def build_fixture_prompt(fixture_language: str, fixture_file: str) -> str:
description = FIXTURE_DESCRIPTIONS.get(fixture_language, fixture_language)
surface = f"Test surface: `{fixture_file}` ({description})."
return PROMPT.format(FIXTURE_SURFACE=surface) + "\n\nKeep all operations and edits scoped to this fixture."
@dataclass
class ModelResult:
model: str
fixture: str
status: str
started_at: float
finished_at: float
workspace: str
jsonl_path: str
review_path: str
turns: int
tool_calls: int
thinking_chars: int
text_chars: int
token_input: int | None
token_output: int | None
token_total: int | None
todo_completed: int
todo_total: int
todo_current: str | None
error: str | None
session_state: dict[str, Any] | None
@dataclass
class ModelProgress:
model: str
label: str
status: str = "pending"
turns: int = 0
tool_calls: int = 0
thinking_chars: int = 0
text_chars: int = 0
token_input: int | None = None
token_output: int | None = None
token_total: int | None = None
todo_completed: int = 0
todo_total: int = 0
todo_current: str | None = None
last_activity: str = "waiting"
last_thinking: str | None = None
last_text: str | None = None
duration_seconds: float | None = None
error: str | None = None
todo_order: list[str] = field(default_factory=list)
todo_items: dict[str, tuple[str, str]] = field(default_factory=dict)
TOOL_WHITELIST = ("read", "edit", "vim", "todo_write", "report_tool_issue")
MODEL_LABEL_WIDTH = 30
STATUS_WIDTH = 7
TOKENS_WIDTH = 9
TODOS_WIDTH = 18
ACTIVITY_WIDTH_FLOOR = 24
THINKING_SNIPPET_LIMIT = 80
TEXT_SNIPPET_LIMIT = 64
def shorten_model_name(model: str) -> str:
if model.startswith("openrouter/"):
return model.removeprefix("openrouter/")
return model
def collapse_whitespace(value: str) -> str:
return " ".join(value.split())
def truncate_text(value: str | None, width: int) -> str:
if width <= 0:
return ""
text = collapse_whitespace(value or "")
if len(text) <= width:
return text
if width <= 1:
return text[:width]
return text[: width - 1] + "…"
def format_count(value: int | None) -> str:
if value is None:
return "-"
if value >= 1_000_000:
return f"{value / 1_000_000:.1f}M"
if value >= 1_000:
return f"{value / 1_000:.1f}k"
return str(value)
def extract_usage_tokens(message: dict[str, Any]) -> tuple[int | None, int | None, int | None]:
usage = message.get("usage")
if not isinstance(usage, dict):
return None, None, None
token_input = usage.get("input")
token_output = usage.get("output")
token_total = usage.get("totalTokens")
if not isinstance(token_total, int) and isinstance(token_input, int) and isinstance(token_output, int):
token_total = token_input + token_output
return (
token_input if isinstance(token_input, int) else None,
token_output if isinstance(token_output, int) else None,
token_total if isinstance(token_total, int) else None,
)
def build_todo_state_from_phases(phases: tuple[TodoPhase, ...]) -> tuple[list[str], dict[str, tuple[str, str]]]:
order: list[str] = []
items: dict[str, tuple[str, str]] = {}
for phase in phases:
for task in phase.tasks:
order.append(task.id)
items[task.id] = (task.content, task.status)
return order, items
def seed_todo_state(todos: list[str]) -> tuple[list[str], dict[str, tuple[str, str]]]:
order: list[str] = []
items: dict[str, tuple[str, str]] = {}
for index, content in enumerate(todos, start=1):
task_id = f"task-{index}"
order.append(task_id)
items[task_id] = (content, "pending")
return order, items
def summarize_todo_state(order: list[str], items: dict[str, tuple[str, str]]) -> tuple[int, int, str | None]:
if not order:
return 0, 0, None
completed = 0
current: str | None = None
for task_id in order:
content, status = items.get(task_id, ("", "pending"))
if status == "completed":
completed += 1
elif current is None and status == "in_progress":
current = content
if current is None:
for task_id in order:
content, status = items.get(task_id, ("", "pending"))
if status == "pending":
current = content
break
return completed, len(order), current
def apply_todo_ops(progress: ModelProgress, args: Any) -> None:
if not isinstance(args, dict):
return
raw_ops = args.get("ops")
if not isinstance(raw_ops, list):
return
for raw_op in raw_ops:
if not isinstance(raw_op, dict):
continue
op = raw_op.get("op")
if op == "replace":
raw_phases = raw_op.get("phases")
if isinstance(raw_phases, list):
phases: list[TodoPhase] = []
for phase_index, raw_phase in enumerate(raw_phases, start=1):
if not isinstance(raw_phase, dict):
continue
raw_tasks = raw_phase.get("tasks")
if not isinstance(raw_tasks, list):
continue
tasks: list[TodoItem] = []
for task_index, raw_task in enumerate(raw_tasks, start=1):
if not isinstance(raw_task, dict):
continue
content = raw_task.get("content")
status = raw_task.get("status")
if not isinstance(content, str) or not isinstance(status, str):
continue
task_id = raw_task.get("id")
if not isinstance(task_id, str) or not task_id:
task_id = f"task-{task_index}"
tasks.append(TodoItem(id=task_id, content=content, status=status, notes=None, details=None))
phase_id = raw_phase.get("id")
name = raw_phase.get("name")
if not isinstance(name, str) or not name:
name = f"Phase {phase_index}"
if not isinstance(phase_id, str) or not phase_id:
phase_id = f"phase-{phase_index}"
phases.append(TodoPhase(id=phase_id, name=name, tasks=tuple(tasks)))
progress.todo_order, progress.todo_items = build_todo_state_from_phases(tuple(phases))
elif op == "update":
task_id = raw_op.get("id")
if not isinstance(task_id, str) or task_id not in progress.todo_items:
continue
content, status = progress.todo_items[task_id]
next_content = raw_op.get("content") if isinstance(raw_op.get("content"), str) else content
next_status = raw_op.get("status") if isinstance(raw_op.get("status"), str) else status
progress.todo_items[task_id] = (next_content, next_status)
elif op == "add_task":
phase = raw_op.get("phase")
task_payload = raw_op.get("task")
if not isinstance(task_payload, dict):
continue
task_id = task_payload.get("id")
if not isinstance(task_id, str) or not task_id:
task_id = raw_op.get("id") if isinstance(raw_op.get("id"), str) else f"task-{len(progress.todo_order) + 1}"
content = task_payload.get("content")
status = task_payload.get("status")
if not isinstance(content, str) or not isinstance(status, str):
continue
if task_id not in progress.todo_items:
insert_after = raw_op.get("after") if isinstance(raw_op.get("after"), str) else None
if insert_after in progress.todo_order:
index = progress.todo_order.index(insert_after) + 1
progress.todo_order.insert(index, task_id)
else:
progress.todo_order.append(task_id)
progress.todo_items[task_id] = (content, status)
elif op == "remove_task":
task_id = raw_op.get("id")
if not isinstance(task_id, str):
continue
progress.todo_items.pop(task_id, None)
progress.todo_order = [candidate for candidate in progress.todo_order if candidate != task_id]
elif op == "add_phase":
raw_tasks = raw_op.get("tasks")
if not isinstance(raw_tasks, list):
continue
for raw_task in raw_tasks:
if not isinstance(raw_task, dict):
continue
content = raw_task.get("content")
status = raw_task.get("status")
if not isinstance(content, str) or not isinstance(status, str):
continue
task_id = raw_task.get("id")
if not isinstance(task_id, str) or not task_id:
task_id = f"task-{len(progress.todo_order) + 1}"
progress.todo_order.append(task_id)
progress.todo_items[task_id] = (content, status)
progress.todo_completed, progress.todo_total, progress.todo_current = summarize_todo_state(
progress.todo_order, progress.todo_items
)
class ProgressPrinter:
def __init__(self, runs: list[tuple[str, str]], *, stream: TextIO | None = None, interactive: bool | None = None) -> None:
self._lock = threading.Lock()
self._stream = sys.stdout if stream is None else stream
self._interactive = self._stream.isatty() if interactive is None else interactive
self._console = Console(file=self._stream, force_terminal=self._interactive, soft_wrap=False)
self._model_order = [run_id for run_id, _ in runs]
self._states = {
run_id: ModelProgress(model=run_id, label=label)
for run_id, label in runs
}
self._fixtures_dir: str | None = None
self._results_dir: str | None = None
self._closed = False
self._final_message: str | None = None
self._live: Live | None = None
if self._interactive:
self._live = Live(
self._build_renderable_locked(),
console=self._console,
auto_refresh=False,
transient=False,
)
self._live.start()
def configure(self, *, fixtures_dir: Path, results_dir: Path) -> None:
with self._lock:
self._fixtures_dir = str(fixtures_dir)
self._results_dir = str(results_dir)
self._refresh_locked()
def mark_starting(self, model: str) -> None:
self._mutate_model(model, status="boot", last_activity="starting rpc")
def mark_ready(self, model: str) -> None:
self._mutate_model(model, status="ready", last_activity="rpc ready")
def seed_todos(self, model: str, todos: list[str]) -> None:
with self._lock:
progress = self._states[model]
progress.todo_order, progress.todo_items = seed_todo_state(todos)
progress.todo_completed, progress.todo_total, progress.todo_current = summarize_todo_state(
progress.todo_order, progress.todo_items
)
progress.last_activity = f"seeded {progress.todo_total} todos"
self._refresh_locked()
def mark_prompt_submitted(self, model: str) -> None:
self._mutate_model(model, status="run", last_activity="prompt submitted")
def mark_turn_start(self, model: str, turns: int) -> None:
self._mutate_model(model, status="run", turns=turns)
def mark_turn_end(self, model: str, turns: int) -> None:
self._mutate_model(model, turns=turns)
def note_tool_start(self, model: str, tool_name: str, intent: str | None, tool_calls: int, args: Any) -> None:
with self._lock:
progress = self._states[model]
progress.status = "run"
progress.tool_calls = tool_calls
if tool_name == "todo_write":
apply_todo_ops(progress, args)
detail = truncate_text(intent, 36)
progress.last_activity = f"{tool_name} · {detail}" if detail else tool_name
self._refresh_locked()
def note_tool_end(self, model: str, tool_name: str, is_error: bool | None) -> None:
if is_error:
self._mutate_model(model, last_activity=f"{tool_name} failed")
def note_todo_reminder(self, model: str, todos: tuple[TodoItem, ...]) -> None:
with self._lock:
progress = self._states[model]
progress.todo_order = [task.id for task in todos]
progress.todo_items = {task.id: (task.content, task.status) for task in todos}
progress.todo_completed, progress.todo_total, progress.todo_current = summarize_todo_state(
progress.todo_order, progress.todo_items
)
self._refresh_locked()
def note_todo_auto_clear(self, model: str) -> None:
with self._lock:
progress = self._states[model]
progress.todo_completed = progress.todo_total
progress.todo_current = None
self._refresh_locked()
def note_thinking(self, model: str, delta: str, total_chars: int) -> None:
with self._lock:
progress = self._states[model]
progress.status = "think"
progress.thinking_chars = total_chars
progress.last_thinking = truncate_text(delta, THINKING_SNIPPET_LIMIT)
progress.last_activity = progress.last_thinking or "thinking"
self._refresh_locked()
def note_text(self, model: str, delta: str, total_chars: int) -> None:
with self._lock:
progress = self._states[model]
progress.status = "run"
progress.text_chars = total_chars
progress.last_text = truncate_text(delta, TEXT_SNIPPET_LIMIT)
progress.last_activity = progress.last_text or "drafting"
self._refresh_locked()
def note_usage(self, model: str, token_input: int | None, token_output: int | None, token_total: int | None) -> None:
self._mutate_model(
model,
token_input=token_input,
token_output=token_output,
token_total=token_total,
)
def mark_completed(self, model: str, duration_seconds: float) -> None:
self._mutate_model(model, status="done", duration_seconds=duration_seconds, last_activity="completed")
def mark_failed(self, model: str, error: str) -> None:
self._mutate_model(model, status="failed", error=error, last_activity=truncate_text(error, 72))
def finish(self, message: str) -> None:
with self._lock:
if self._closed:
return
self._final_message = message
if self._live is not None:
self._live.update(self._build_renderable_locked(), refresh=True)
self._live.stop()
else:
self._console.print(self._build_renderable_locked())
self._closed = True
def _mutate_model(self, model: str, **changes: Any) -> None:
with self._lock:
progress = self._states[model]
for key, value in changes.items():
setattr(progress, key, value)
self._refresh_locked()
def _refresh_locked(self) -> None:
if self._live is None:
return
self._live.update(self._build_renderable_locked(), refresh=True)
def _build_renderable_locked(self) -> Group:
done = sum(1 for state in self._states.values() if state.status == "done")
failed = sum(1 for state in self._states.values() if state.status == "failed")
active = sum(1 for state in self._states.values() if state.status not in {"pending", "done", "failed"})
summary = Text()
summary.append(f"done {done}/{len(self._states)}", style="bold green")
summary.append(" • ", style="dim")
summary.append(f"active {active}", style="bold cyan")
summary.append(" • ", style="dim")
summary.append(f"failed {failed}", style="bold red" if failed else "green")
results_line = Text()
results_line.append("results ", style="bold")
results_line.append(self._results_dir or "-", style="cyan")
config_line = Text()
config_line.append("fixtures ", style="bold")
config_line.append(self._fixtures_dir or "-", style="cyan")
config_line.append(" • ", style="dim")
config_line.append("tools ", style="bold")
config_line.append("|".join(TOOL_WHITELIST), style="magenta")
header = Panel(
Group(summary, results_line, config_line),
title="rate-edit-tool",
border_style="cyan",
box=box.ROUNDED,
)
table = Table(box=box.SIMPLE_HEAVY, expand=True, show_lines=False)
table.add_column("Model", style="bold", width=34, min_width=34, max_width=34)
table.add_column("State", no_wrap=True, width=7)
table.add_column("Turns", justify="right", no_wrap=True, width=5)
table.add_column("Tools", justify="right", no_wrap=True, width=5)
table.add_column("Tokens", justify="right", no_wrap=True, width=8)
table.add_column("Todos", ratio=1, min_width=24)
for model in self._model_order:
state = self._states[model]
table.add_row(
self._model_text(state),
self._status_text(state.status),
str(state.turns),
str(state.tool_calls),
format_count(state.token_total),
self._todo_text(state),
)
if self._final_message:
footer = Panel(self._final_message, border_style="green" if failed == 0 else "red", box=box.ROUNDED)
return Group(header, table, footer)
return Group(header, table)
@staticmethod
def _status_text(status: str) -> Text:
styles = {
"pending": "dim",
"boot": "yellow",
"ready": "blue",
"run": "cyan",
"think": "magenta",
"done": "green",
"failed": "bold red",
}
return Text(status, style=styles.get(status, "white"))
@staticmethod
def _todo_text(state: ModelProgress) -> str:
if state.todo_total == 0:
return "-"
summary = f"{state.todo_completed}/{state.todo_total}"
if state.todo_current:
summary = f"{summary} · {truncate_text(state.todo_current, 24)}"
return summary
@staticmethod
def _model_text(state: ModelProgress) -> Text:
text = Text(truncate_text(state.label, 34), style="bold")
activity = state.error if state.status == "failed" and state.error else state.last_activity
if activity and activity not in {"waiting", "completed"}:
text.append("\n")
text.append(truncate_text(activity, 34), style="dim")
return text
def slugify(value: str) -> str:
return "".join(char if char.isalnum() or char in "._-" else "_" for char in value)
def materialize_workspace(target_dir: Path, fixture_file: str | None = None) -> None:
if fixture_file is not None:
content = WORKSPACE_FILES[fixture_file]
(target_dir / fixture_file).write_text(content)
return
for name, content in WORKSPACE_FILES.items():
(target_dir / name).write_text(content)
def sync_reference_fixtures(fixtures_dir: Path) -> None:
fixtures_dir.mkdir(parents=True, exist_ok=True)
keep = set(REFERENCE_FILES)
for child in fixtures_dir.iterdir():
if child.name not in keep:
if child.is_dir():
shutil.rmtree(child)
else:
child.unlink()
for name, content in REFERENCE_FILES.items():
(fixtures_dir / name).write_text(content)
def require_openrouter_key() -> str:
key = os.environ.get("OPENROUTER_API_KEY")
if not key:
raise SystemExit("OPENROUTER_API_KEY is not set")
return key
def resolve_omp_bin(raw: str | None) -> str:
if raw:
return raw
found = shutil.which("omp")
if not found:
raise SystemExit("Could not find `omp` on PATH. Set --omp-bin or OMP_BIN.")
return found
def serialize_notification(notification: Any) -> dict[str, Any]:
if is_dataclass(notification):
return asdict(notification)
if isinstance(notification, dict):
return dict(notification)
data = getattr(notification, "__dict__", None)
if isinstance(data, dict):
return dict(data)
return {"value": repr(notification)}
class ModelRunRecorder:
def __init__(self, run_id: str, model: str, fixture: str, printer: ProgressPrinter, jsonl_path: Path) -> None:
self.run_id = run_id
self.model = model
self.fixture = fixture
self.printer = printer
self.jsonl_path = jsonl_path
self.turns = 0
self.tool_calls = 0
self.thinking_chars = 0
self.text_chars = 0
self.token_input: int | None = None
self.token_output: int | None = None
self.token_total: int | None = None
self.todo_completed = 0
self.todo_total = 0
self.todo_current: str | None = None
self.review_sections: list[str] = []
self.agent_ended = False
self.auto_retry_active = False
self.auto_retry_delay_ms = 0
self.last_event_at = time.monotonic()
self._consumed_assistant_messages = 0
self._event_lock = threading.Lock()
def _touch(self) -> None:
self.last_event_at = time.monotonic()
def record_notification(self, notification: RpcNotification) -> None:
self._touch()
self._append_jsonl(serialize_notification(notification))
def record_ui(self, request: ExtensionUiRequest) -> None:
self._touch()
if request.method in {"notify", "setStatus", "setTitle", "set_editor_text"}:
return
if request.method == "setWidget" and request.widget_key == "autoresearch":
return
def record_turn_start(self, _event: TurnStartEvent) -> None:
self._touch()
self.turns += 1
self.printer.mark_turn_start(self.run_id, self.turns)
def record_turn_end(self, _event: TurnEndEvent) -> None:
self._touch()
self.printer.mark_turn_end(self.run_id, self.turns)
def record_tool_execution_start(self, event: ToolExecutionStartEvent) -> None:
self._touch()
self.tool_calls += 1
self.printer.note_tool_start(self.run_id, event.tool_name, event.intent, self.tool_calls, event.args)
def record_tool_execution_update(self, _event: ToolExecutionUpdateEvent) -> None:
self._touch()
return
def record_tool_execution_end(self, event: ToolExecutionEndEvent) -> None:
self._touch()
self.printer.note_tool_end(self.run_id, event.tool_name, event.is_error)
def record_auto_retry_start(self, event: AutoRetryStartEvent) -> None:
self._touch()
self.auto_retry_active = True
self.auto_retry_delay_ms = event.delay_ms
def record_auto_retry_end(self, _event: AutoRetryEndEvent) -> None:
self._touch()
self.auto_retry_active = False
self.auto_retry_delay_ms = 0
def record_message_end(self, event: MessageEndEvent) -> None:
self._touch()
message = event.message
if not isinstance(message, dict) or message.get("role") != "assistant":
return
text = assistant_text(message)
if not isinstance(text, str) or not text.strip():
return
self.review_sections.append(text.strip())
self._consumed_assistant_messages += 1
token_input, token_output, token_total = extract_usage_tokens(message)
if token_total is not None and (self.token_total is None or token_total >= self.token_total):
self.token_input = token_input
self.token_output = token_output
self.token_total = token_total
self.printer.note_usage(self.run_id, token_input, token_output, token_total)
def record_agent_end(self, event: AgentEndEvent) -> None:
self._touch()
self.agent_ended = True
assistant_count = 0
for message in event.messages:
if not isinstance(message, dict) or message.get("role") != "assistant":
continue
text = assistant_text(message)
if assistant_count >= self._consumed_assistant_messages and isinstance(text, str) and text.strip():
self.review_sections.append(text.strip())
assistant_count += 1
self._consumed_assistant_messages = assistant_count
for message in reversed(event.messages):
if isinstance(message, dict) and message.get("role") == "assistant":
token_input, token_output, token_total = extract_usage_tokens(message)
self.token_input = token_input
self.token_output = token_output
self.token_total = token_total
self.printer.note_usage(self.run_id, token_input, token_output, token_total)
break
def record_message_update(self, event: MessageUpdateEvent) -> None:
self._touch()
assistant_event = event.assistant_message_event
partial = assistant_event.get("partial")
if isinstance(partial, dict):
token_input, token_output, token_total = extract_usage_tokens(partial)
if token_total is not None and (self.token_total is None or token_total >= self.token_total):
self.token_input = token_input
self.token_output = token_output
self.token_total = token_total
self.printer.note_usage(self.run_id, token_input, token_output, token_total)
delta_type = assistant_event.get("type")
delta = assistant_event.get("delta")
if not isinstance(delta, str):
return
if delta_type == "thinking_delta":
self.thinking_chars += len(delta)
self.printer.note_thinking(self.run_id, delta, self.thinking_chars)
elif delta_type == "text_delta":
self.text_chars += len(delta)
self.printer.note_text(self.run_id, delta, self.text_chars)
def record_todo_reminder(self, event: TodoReminderEvent) -> None:
self._touch()
self.todo_completed = sum(1 for task in event.todos if task.status == "completed")
self.todo_total = len(event.todos)
in_progress = next((task.content for task in event.todos if task.status == "in_progress"), None)
pending = next((task.content for task in event.todos if task.status == "pending"), None)
self.todo_current = in_progress or pending
self.printer.note_todo_reminder(self.run_id, event.todos)
def record_todo_auto_clear(self, _event: TodoAutoClearEvent) -> None:
self._touch()
self.todo_completed = self.todo_total
self.todo_current = None
self.printer.note_todo_auto_clear(self.run_id)
def sync_final_todos(self, phases: tuple[TodoPhase, ...]) -> None:
order, items = build_todo_state_from_phases(phases)
self.todo_completed, self.todo_total, self.todo_current = summarize_todo_state(order, items)
flattened = tuple(task for phase in phases for task in phase.tasks)
self.printer.note_todo_reminder(self.run_id, flattened)
def build_review_markdown(self) -> str:
if not self.review_sections:
return ""
return "\n\n-----------\n\n".join(self.review_sections)
def is_effectively_complete(self, *, quiet_seconds: float) -> bool:
return (
len(self.review_sections) > 0
and self.todo_total > 0
and self.todo_completed >= self.todo_total
and not self.auto_retry_active
and (time.monotonic() - self.last_event_at) >= quiet_seconds
)
def _append_jsonl(self, payload: dict[str, Any]) -> None:
with self._event_lock:
with self.jsonl_path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(payload) + "\n")
def run_model_sync(
*,
model: str,
fixture_language: str,
fixture_file: str,
omp_bin: str,
results_dir: Path,
workspace_root: Path,
timeout: float,
printer: ProgressPrinter,
openrouter_key: str,
) -> ModelResult:
started_at = time.time()
model_slug = slugify(model)
fixture_slug = slugify(fixture_language)
run_id = f"{model}|{fixture_slug}"
workspace = workspace_root / model_slug / fixture_slug
workspace.mkdir(parents=True, exist_ok=True)
materialize_workspace(workspace, fixture_file=fixture_file)
review_slug = slugify(shorten_model_name(model))
review_path = results_dir / f"review_{review_slug}_{fixture_slug}.md"
jsonl_path = Path(tempfile.gettempdir()) / f"rate-edit-tool-{results_dir.name}-{model_slug}-{fixture_slug}.jsonl"
jsonl_path.unlink(missing_ok=True)
jsonl_path.touch()
recorder = ModelRunRecorder(
run_id=run_id,
model=model,
fixture=fixture_language,
printer=printer,
jsonl_path=jsonl_path,
)
printer.mark_starting(run_id)
error_message: str | None = None
session_state: dict[str, Any] | None = None
try:
with RpcClient(
executable=omp_bin,
model=model,
cwd=workspace,
env={"OPENROUTER_API_KEY": openrouter_key, "PI_STRICT_EDIT_MODE": "1"},
thinking="high",
tools=TOOL_WHITELIST,
no_skills=True,
no_rules=True,
no_session=True,
startup_timeout=30.0,
request_timeout=30.0,
) as client:
client.on_notification(recorder.record_notification)
client.on_turn_start(recorder.record_turn_start)
client.on_turn_end(recorder.record_turn_end)
client.on_tool_execution_start(recorder.record_tool_execution_start)
client.on_tool_execution_update(recorder.record_tool_execution_update)
client.on_tool_execution_end(recorder.record_tool_execution_end)
client.on_auto_retry_start(recorder.record_auto_retry_start)
client.on_auto_retry_end(recorder.record_auto_retry_end)
client.on_agent_end(recorder.record_agent_end)
client.on_message_update(recorder.record_message_update)
client.on_message_end(recorder.record_message_end)
client.on_todo_reminder(recorder.record_todo_reminder)
client.on_todo_auto_clear(recorder.record_todo_auto_clear)
client.on_ui_request(recorder.record_ui)
client.install_headless_ui()
printer.mark_ready(run_id)
client.set_todos(TODOS)
printer.seed_todos(run_id, TODOS)
deadline = time.monotonic() + timeout
def wait_for_settle() -> None:
last_timeout: RpcError | None = None
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
if last_timeout is not None:
raise last_timeout
raise RpcError("Timed out waiting for agent to settle")
try:
client.wait_for_idle(timeout=min(remaining, 60.0))
if recorder.auto_retry_active:
time.sleep(min(max(recorder.auto_retry_delay_ms / 1000.0, 0.2), 2.0))
continue
if recorder.agent_ended:
grace = min(0.5, max(deadline - time.monotonic(), 0.0))
if grace > 0:
time.sleep(grace)
if recorder.auto_retry_active:
time.sleep(min(max(recorder.auto_retry_delay_ms / 1000.0, 0.2), 2.0))
continue
return
if recorder.is_effectively_complete(quiet_seconds=2.0):
return
time.sleep(0.2)
except RpcError as error:
if "Timed out waiting for agent_end" not in str(error):
raise
last_timeout = error
if recorder.is_effectively_complete(quiet_seconds=2.0):
return
printer.mark_prompt_submitted(run_id)
client.prompt(build_fixture_prompt(fixture_language, fixture_file))
wait_for_settle()
review_markdown = recorder.build_review_markdown()
if not review_markdown.strip():
printer.mark_prompt_submitted(run_id)
client.prompt(FINAL_REVIEW_PROMPT)
wait_for_settle()
review_markdown = recorder.build_review_markdown()
if not review_markdown.strip():
raise RpcError("Agent completed without final review text after retry")
stats = client.get_session_stats()
todo_phases = client.get_todos()
recorder.sync_final_todos(todo_phases)
if (recorder.token_total is None or recorder.token_total <= 0) and stats.tokens.total > 0:
recorder.token_input = stats.tokens.input
recorder.token_output = stats.tokens.output
recorder.token_total = stats.tokens.total
printer.note_usage(run_id, recorder.token_input, recorder.token_output, recorder.token_total)
review_path.write_text(review_markdown)
provider, model_id = model.split("/", 1)
session_state = {
"model": {"provider": provider, "id": model_id},
"thinkingLevel": "high",
}
status = "ok"
except Exception as error: # noqa: BLE001
error_message = f"{type(error).__name__}: {error}" if str(error) else type(error).__name__
printer.mark_failed(run_id, error_message)
status = "failed"
finished_at = time.time()
duration_seconds = round(finished_at - started_at, 3)
if status == "ok":
printer.mark_completed(run_id, duration_seconds)
return ModelResult(
model=model,
fixture=fixture_language,
status=status,
started_at=started_at,
finished_at=finished_at,
workspace=str(workspace),
jsonl_path=str(jsonl_path),
review_path=str(review_path),
turns=recorder.turns,
tool_calls=recorder.tool_calls,
thinking_chars=recorder.thinking_chars,
text_chars=recorder.text_chars,
token_input=recorder.token_input,
token_output=recorder.token_output,
token_total=recorder.token_total,
todo_completed=recorder.todo_completed,
todo_total=recorder.todo_total,
todo_current=recorder.todo_current,
error=error_message,
session_state=session_state,
)
def build_oracle_review_prompt(results: list[ModelResult]) -> str:
review_sections: list[str] = []
for result in sorted(results, key=lambda candidate: (candidate.model, candidate.fixture)):
review_text = Path(result.review_path).read_text(encoding="utf-8").strip()
if not review_text:
continue
review_sections.append(
textwrap.dedent(
f"""\
<review>
<model>{result.model}</model>
<fixture>{result.fixture}</fixture>
<path>{result.review_path}</path>
{review_text}
</review>
"""
).strip()
)
if not review_sections:
raise ValueError("No review content available for oracle synthesis")
review_payload = "\n\n".join(review_sections)
return ORACLE_REVIEW_PROMPT.replace("{{REVIEWS}}", review_payload)
def run_oracle_review_sync(
*,
model: str,
omp_bin: str,
results: list[ModelResult],
results_dir: Path,
timeout: float,
openrouter_key: str,
) -> str:
prompt = build_oracle_review_prompt(results)
with RpcClient(
executable=omp_bin,
model=model,
cwd=results_dir,
env={"OPENROUTER_API_KEY": openrouter_key, "PI_STRICT_EDIT_MODE": "1"},
thinking="high",
tools=(),
no_skills=True,
no_rules=True,
no_session=True,
startup_timeout=30.0,
request_timeout=30.0,
) as client:
client.install_headless_ui()
client.prompt_and_wait(prompt, timeout=timeout)
review_markdown = client.get_last_assistant_text()
if not isinstance(review_markdown, str) or not review_markdown.strip():
raise RpcError("Oracle model completed without synthesis text")
return review_markdown.strip()
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run OpenRouter fixture evaluations through omp RPC mode.")
parser.add_argument("--omp-bin", default=os.environ.get("OMP_BIN"))
parser.add_argument("--fixtures-dir", default=os.path.expanduser("~/tmp/fixtures"))
parser.add_argument("--results-dir")
parser.add_argument("--timeout", type=float, default=900.0, help="Per run timeout in seconds.")
parser.add_argument("--model", dest="models", action="append", help="Repeat to limit execution to specific models.")
parser.add_argument("--oracle-model", default=ORACLE_MODEL, help="Model used to synthesize findings across all reviews.")
return parser.parse_args()
async def run_all(args: argparse.Namespace) -> int:
openrouter_key = require_openrouter_key()
omp_bin = resolve_omp_bin(args.omp_bin)
fixtures_dir = Path(args.fixtures_dir).expanduser()
sync_reference_fixtures(fixtures_dir)
timestamp = time.strftime("%Y%m%d-%H%M%S")
tmp_root = Path(tempfile.gettempdir())
results_dir = Path(args.results_dir) if args.results_dir else tmp_root / f"omp-fixture-runs-{timestamp}"
results_dir.mkdir(parents=True, exist_ok=True)
workspace_root = tmp_root / f"rate-edit-tool-workspaces-{timestamp}"
workspace_root.mkdir(parents=True, exist_ok=True)
selected_models = args.models or MODELS
model_fixtures = {model: random.sample(list(FIXTURES), min(2, len(FIXTURES))) for model in selected_models}
run_specs = [
(f"{model}|{fixture_language}", f"{shorten_model_name(model)}:{fixture_language}")
for model in selected_models
for fixture_language, _ in model_fixtures[model]
]
printer = ProgressPrinter(run_specs)
printer.configure(fixtures_dir=fixtures_dir, results_dir=results_dir)
tasks = [
asyncio.to_thread(
run_model_sync,
model=model,
fixture_language=fixture_language,
fixture_file=fixture_file,
omp_bin=omp_bin,
results_dir=results_dir,
workspace_root=workspace_root,
timeout=args.timeout,
printer=printer,
openrouter_key=openrouter_key,
)
for model in selected_models
for fixture_language, fixture_file in model_fixtures[model]
]
results = await asyncio.gather(*tasks)
failures = sum(1 for result in results if result.status != "ok")
if failures:
printer.finish(f"{failures}/{len(results)} run(s) failed")
return 1
oracle_synthesis = await asyncio.to_thread(
run_oracle_review_sync,
model=args.oracle_model,
omp_bin=omp_bin,
results=results,
results_dir=results_dir,
timeout=args.timeout,
openrouter_key=openrouter_key,
)
printer.finish(
f"{len(results)} review file(s) completed. Oracle synthesis:\n\n{oracle_synthesis}"
)
return 0
def main() -> int:
args = parse_args()
return asyncio.run(run_all(args))
if __name__ == "__main__":
raise SystemExit(main())