diff --git a/docs/concepts.md b/docs/concepts.md index 7d9269a95..dc344e0c1 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -8,7 +8,7 @@ The mental model for benchflow. Read once, then refer back from the how-tos. | Primitive | What it is | |-----------|------------| | **Task** | A directory on disk: a `task.md` document (YAML frontmatter + prompt body) plus `environment/Dockerfile` for the sandbox, `verifier/` checks, and optional `oracle/` — or the legacy split layout (`task.toml` + `instruction.md` + `tests/` + `solution/`). Authored once, evaluated many times. | -| **Agent** | A registered ACP-speaking program (Claude Code, Gemini CLI, OpenCode, etc.). Identified by name (`"gemini"`, `"opencode"`) plus an optional model ID. Use the `acpx/` prefix (e.g. `acpx/gemini`) to route through [ACPX](https://acpx.sh/), a headless ACP client with persistent sessions and crash recovery. | +| **Agent** | A registered ACP-speaking program (Claude Code, Gemini CLI, OpenCode, Ori via BenchFlow's ACP shim, etc.). Identified by name (`"gemini"`, `"opencode"`, `"ori"`) plus an optional model ID. Use the `acpx/` prefix (e.g. `acpx/gemini`) to route through [ACPX](https://acpx.sh/), a headless ACP client with persistent sessions and crash recovery. | | **Environment** | The sandbox where the agent runs and the verifier checks the result. Docker locally, Daytona for cloud, Modal for serverless/GPU. Abstracted behind the `Sandbox` protocol — bring your own sandbox backend. | | **Verifier** | The test runner that scores the rollout. Its entry point is a `test.sh` script (native `verifier/test.sh`, legacy `tests/test.sh`) — which typically runs `pytest` against the workspace the agent left behind. For subjective tasks, use an [LLM-as-judge](./llm-judge.md) verifier scored against a rubric. Outputs `rewards: {reward: float}`. See the [verifier file map](#verifier-file-map) for which file lives where in native vs legacy packages. | | **Rollout** | One agent run on one task. Holds the lifecycle (setup → start → install → execute → verify → cleanup). All higher-level primitives below are built on Rollouts. | diff --git a/docs/getting-started.md b/docs/getting-started.md index 624500025..78da9dbc3 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -49,19 +49,22 @@ uv sync --extra dev --locked ## Auth: OAuth, long-lived token, or API key -You don't need an API key if you're a Claude / Codex / Gemini subscriber. Three options, pick one per agent: +You don't need an API key if you're using a Claude / Codex / Gemini +subscription or an OpenRouter login. Three options, pick one per agent: ### Option 1 — Subscription OAuth from host CLI login If you've logged into the agent's CLI on your host (`claude auth login`, -`codex login`, or the `gemini` interactive flow), benchflow picks up the -credential file and copies it into the sandbox. No API key billing. +`codex login`, `ori login`, or the `gemini` interactive flow), benchflow picks up the +credential file and copies it into the sandbox. No API key needs to be +exported; billing and entitlements still follow the account used by that CLI. | Agent | How to log in on the host | What benchflow detects | Replaces env var | |-------|---------------------------|------------------------|------------------| | `claude-agent-acp` | `claude auth login` (Claude Code CLI) | `~/.claude/.credentials.json` | `ANTHROPIC_API_KEY` | | `codex-acp` | `codex login` (Codex CLI) | `~/.codex/auth.json` | `OPENAI_API_KEY` | | `gemini` | `gemini` (interactive login) | `~/.gemini/oauth_creds.json` | `GEMINI_API_KEY` | +| `ori` | [`ori login`](https://openrouter.ai/docs/guides/ori/harness) | `~/.ori/credentials.json` or `~/.openrouter/credentials.json` | `OPENROUTER_API_KEY` | When benchflow finds the detect file, you'll see: @@ -91,6 +94,7 @@ export ANTHROPIC_API_KEY=sk-ant-... export OPENAI_API_KEY=sk-... export CODEX_API_KEY=sk-... # Codex alias for OPENAI_API_KEY export GEMINI_API_KEY=... +export OPENROUTER_API_KEY=sk-or-... # OpenRouter / Ori export LLM_API_KEY=... # OpenHands / LiteLLM-compatible providers export AZURE_API_KEY=... export AZURE_API_ENDPOINT='https://.openai.azure.com/' @@ -163,6 +167,14 @@ GEMINI_API_KEY=... bench eval run \ # List the registered agents bench agent list + +# Run OpenRouter's native Ori coding harness +OPENROUTER_API_KEY=... bench eval run \ + --source-repo benchflow-ai/skillsbench --source-path tasks \ + --include citation-check \ + --agent ori \ + --model openrouter/anthropic/claude-sonnet-4.6 \ + --sandbox docker ``` `bench eval run` is the primary command for running evaluations — it works for diff --git a/docs/reference/python-api.md b/docs/reference/python-api.md index 2f2cdda9a..a7d2945b1 100644 --- a/docs/reference/python-api.md +++ b/docs/reference/python-api.md @@ -271,6 +271,7 @@ result = await bf.run(config) | `claude-agent-acp` | ACP | ANTHROPIC_API_KEY | `claude` | | `codex-acp` | ACP | OPENAI_API_KEY, CODEX_API_KEY, CODEX_ACCESS_TOKEN, or host login | `codex` | | `opencode` | ACP | inferred from model/provider | — | +| `ori` | ACP (BenchFlow shim over Ori JSONL) | OPENROUTER_API_KEY or host login | — | | `openhands` | ACP | LLM_API_KEY | `oh` | | `pi-acp` | ACP | ANTHROPIC_API_KEY | `pi` | | `openclaw` | ACP | inferred from model | — | diff --git a/docs/running-benchmarks.md b/docs/running-benchmarks.md index 8da6013b9..2eaa67007 100644 --- a/docs/running-benchmarks.md +++ b/docs/running-benchmarks.md @@ -315,6 +315,7 @@ Common choices: | Gemini | `gemini` | `GEMINI_API_KEY` or host login | | Claude Code | `claude-agent-acp` (alias: `claude`) | `ANTHROPIC_API_KEY` or host login | | Codex | `codex-acp` (alias: `codex`) | `OPENAI_API_KEY`, `CODEX_API_KEY`, `CODEX_ACCESS_TOKEN`, or host login | +| OpenRouter Ori | `ori` | `OPENROUTER_API_KEY` or `ori login` | | OpenHands | `openhands` (alias: `oh`) | `LLM_API_KEY` | | Harvey LAB harness | `harvey-lab-harness` (alias: `harvey-lab`) | Provider key matching model | @@ -333,6 +334,35 @@ bench eval run --tasks-dir tasks/edit-pdf --agent acpx/gemini --model gemini-3.1 ACPX is a headless ACP client that adds persistent sessions and crash recovery. The underlying agent's install, env vars, credentials, and skill paths are all preserved. +### OpenRouter Ori harness + +`--agent ori` runs Ori's built-in coding harness through BenchFlow's ACP shim. +The shim drives Ori's headless JSONL runtime, resumes Ori's native session id +for follow-up turns, streams messages and tool calls as ACP events, and returns +cumulative token usage in the ACP prompt response. BenchFlow pins and verifies +the Ori binary before installing it. + +Use an OpenRouter API key: + +```bash +export OPENROUTER_API_KEY=... +bench eval run \ + --tasks-dir tasks/citation-check \ + --agent ori \ + --model openrouter/anthropic/claude-sonnet-4.6 \ + --sandbox docker +``` + +The first `openrouter/` selects BenchFlow's provider; the remainder is the +OpenRouter model slug sent to Ori. For OpenRouter Auto, use +`--model openrouter/openrouter/auto` (also Ori's registry default). + +Alternatively, install Ori on the host, run `ori login`, and unset +`OPENROUTER_API_KEY`. BenchFlow detects either `~/.ori/credentials.json` or +`~/.openrouter/credentials.json`, copies it into the sandbox, and uses Ori's +native login. API-key runs go through BenchFlow's LiteLLM usage gateway; native +login runs use the trusted token totals returned through the ACP shim. + The **Harvey LAB harness** agent is special — it runs Harvey LAB's own agent loop (6 tools, system prompt) inside BenchFlow's sandbox. Use it for parity testing (same agent on both original and converted tasks). diff --git a/src/benchflow/agents/env.py b/src/benchflow/agents/env.py index ebf83e15c..df6d195ce 100644 --- a/src/benchflow/agents/env.py +++ b/src/benchflow/agents/env.py @@ -388,8 +388,8 @@ def uses_native_subscription_auth( """Return True when an agent should use CLI/subscription auth directly. This is the Harbor-style split point: API-key runs can be routed through - LiteLLM, while subscription-auth runs stay on the native Codex/Claude ACP - path and report usage from the agent protocol response. + LiteLLM, while subscription-auth runs stay on the agent's native auth path + and report usage from the ACP response. """ if agent_env.get("BENCHFLOW_PROVIDER_NAME") == "litellm" or any( agent_env.get(key) for key in _LITELLM_RUNTIME_MARKER_KEYS @@ -414,28 +414,26 @@ def uses_native_subscription_auth( or check_subscription_auth(agent, required_key) ) - # Registry-driven Claude-CLI gate: any agent whose subscription_auth - # substitutes ANTHROPIC_API_KEY runs the Claude Code CLI and can take - # OAuth/subscription auth natively (claude-agent-acp, omnigent claude-*). - claude_cfg = AGENTS.get(agent) - if ( - claude_cfg is not None - and claude_cfg.subscription_auth is not None - and claude_cfg.subscription_auth.replaces_env == "ANTHROPIC_API_KEY" - ): - if agent_env.get("ANTHROPIC_API_KEY"): + # Registry-owned policy for native-login ACP agents. The containing + # SubscriptionAuth declares both the provider auth context (replaces_env) + # and any direct token aliases; no provider or agent name is hard-coded in + # this routing layer. + config = AGENTS.get(agent) + subscription = config.subscription_auth if config is not None else None + policy = subscription.native_policy if subscription is not None else None + if subscription is not None and policy is not None: + required_key = subscription.replaces_env + if agent_env.get(required_key): return False if model is not None: from benchflow.agents.registry import infer_env_key_for_model - if infer_env_key_for_model(model) != "ANTHROPIC_API_KEY": + if infer_env_key_for_model(model) != required_key: return False return ( - bool(agent_env.get(_CLAUDE_CODE_OAUTH_TOKEN_ENV)) - or bool(agent_env.get(_CLAUDE_OAUTH_TOKEN_ENV)) - or bool(agent_env.get("ANTHROPIC_AUTH_TOKEN")) + any(bool(agent_env.get(key)) for key in policy.direct_envs) or agent_env.get(_SUBSCRIPTION_AUTH_MARKER) == "1" - or check_subscription_auth(agent, "ANTHROPIC_API_KEY") + or check_subscription_auth(agent, required_key) ) return False @@ -558,7 +556,8 @@ def check_subscription_auth(agent: str, required_key: str) -> bool: sa = agent_cfg.subscription_auth if sa.replaces_env != required_key: return False - return Path(sa.detect_file).expanduser().is_file() + detect_files = sa.detect_files or [sa.detect_file] + return any(Path(path).expanduser().is_file() for path in detect_files) def validate_aws_bedrock_env(agent_env: dict[str, str], model: str) -> None: diff --git a/src/benchflow/agents/ori_acp_shim.py b/src/benchflow/agents/ori_acp_shim.py new file mode 100644 index 000000000..2a17a915e --- /dev/null +++ b/src/benchflow/agents/ori_acp_shim.py @@ -0,0 +1,332 @@ +#!/usr/bin/env python3 +"""ACP-over-stdio shim for OpenRouter Ori's headless coding harness. + +Ori exposes a persistent JSONL CLI rather than an ACP server. This process is +the narrow transport adapter: it speaks ACP on stdin/stdout, invokes +``ori code --output jsonl`` for each prompt, resumes Ori's native session id, +and translates the streamed runtime events into ACP ``session/update`` +notifications. The sibling :mod:`ori_jsonl` and :mod:`ori_events` modules own +tolerant decoding, typed token arithmetic, and event translation. + +All three files are installed into the sandbox and run without BenchFlow installed, +so this module depends only on the Python standard library. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +try: # Installed scripts are sibling top-level modules. + from ori_events import TurnTranslator + from ori_jsonl import OriUsage, decode_line, terminal_result_error +except ModuleNotFoundError: # Package import used by BenchFlow's unit tests. + from .ori_events import TurnTranslator # type: ignore[no-redef] + from .ori_jsonl import OriUsage, decode_line, terminal_result_error + +ORI_BINARY = "/opt/benchflow/bin/ori" +ORI_VERSION = "0.12.0+68f9a36" +_DEFAULT_MODEL = "openrouter/auto" +_EFFORTS = ("none", "minimal", "low", "medium", "high", "xhigh", "max") + + +def send(message: dict[str, Any]) -> None: + """Write one JSON-RPC frame without contaminating stdout.""" + sys.stdout.write(json.dumps(message, ensure_ascii=False) + "\n") + sys.stdout.flush() + + +def recv() -> dict[str, Any]: + """Read the next non-empty JSON-RPC frame.""" + while True: + line = sys.stdin.readline() + if not line: + raise EOFError("stdin closed") + if line.strip(): + value = json.loads(line) + if not isinstance(value, dict): + raise ValueError("JSON-RPC frame must be an object") + return value + + +def _ensure_global_workspace() -> None: + """Create only Ori's missing offline persona metadata, never credentials.""" + home = Path(os.environ.get("BENCHFLOW_AGENT_HOME") or Path.home()) + global_root = home / ".ori" / "global" + (global_root / "features").mkdir(parents=True, exist_ok=True) + ori_md = global_root / "ori.md" + package_json = global_root / "package.json" + if not ori_md.exists(): + ori_md.write_text( + f"---\nmodel: {_DEFAULT_MODEL}\nversion: {ORI_VERSION}\n---\n", + encoding="utf-8", + ) + if not package_json.exists(): + package_json.write_text( + '{\n "name": "benchflow-ori-runtime",\n' + ' "private": true,\n "type": "module"\n}\n', + encoding="utf-8", + ) + + +def build_ori_command( + *, + model: str, + prompt_path: str, + reasoning_effort: str | None = None, + native_session_id: str | None = None, + binary: str | None = None, +) -> list[str]: + """Build Ori's argv as data; prompts never pass through a shell.""" + binary = binary or ORI_BINARY + command = [ + binary, + "code", + "--harness", + "ori", + "--model", + model, + "--approvals", + "self-drive", + "--output", + "jsonl", + ] + if reasoning_effort: + command.extend(["--reasoning-effort", reasoning_effort]) + if native_session_id: + command.extend(["--session", native_session_id]) + command.extend(["--prompt-file", prompt_path]) + return command + + +@dataclass +class SessionState: + cwd: str + model: str + reasoning_effort: str | None = None + native_session_id: str | None = None + usage: OriUsage = field(default_factory=OriUsage) + has_usage: bool = False + + +class OriACPServer: + """Minimal ACP server with one or more independent Ori sessions.""" + + def __init__(self) -> None: + self.sessions: dict[str, SessionState] = {} + self.current_process: subprocess.Popen[str] | None = None + + def handle(self, message: dict[str, Any]) -> None: + method = str(message.get("method") or "") + request_id = message.get("id") + params = message.get("params") + params = params if isinstance(params, dict) else {} + + if method == "initialize": + self._reply( + request_id, + { + "protocolVersion": 1, + "agentCapabilities": { + "loadSession": False, + "promptCapabilities": {"image": False, "audio": False}, + }, + "agentInfo": {"name": "openrouter-ori", "version": ORI_VERSION}, + }, + ) + return + if method == "session/new": + self._new_session(request_id, params) + return + if method == "session/set_model": + state = self._state(params) + model = params.get("modelId") + if not isinstance(model, str) or not model: + raise ValueError("modelId must be a non-empty string") + state.model = model + self._reply(request_id, {}) + return + if method == "session/set_config_option": + self._set_config_option(request_id, params) + return + if method == "session/prompt": + self._prompt(request_id, params) + return + if method == "session/cancel": + if self.current_process is not None and self.current_process.poll() is None: + self.current_process.terminate() + return + if request_id is not None: + self._error(request_id, -32601, f"Method not found: {method}") + + @staticmethod + def _reply(request_id: object, result: dict[str, Any]) -> None: + send({"jsonrpc": "2.0", "id": request_id, "result": result}) + + @staticmethod + def _error(request_id: object, code: int, message: str) -> None: + send( + { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": code, "message": message[-4000:]}, + } + ) + + def _state(self, params: dict[str, Any]) -> SessionState: + session_id = params.get("sessionId") + if not isinstance(session_id, str) or session_id not in self.sessions: + raise ValueError(f"unknown sessionId: {session_id!r}") + return self.sessions[session_id] + + def _new_session(self, request_id: object, params: dict[str, Any]) -> None: + _ensure_global_workspace() + cwd = params.get("cwd") + if not isinstance(cwd, str) or not cwd: + cwd = os.getcwd() + session_id = f"ori-{uuid.uuid4().hex[:12]}" + model = ( + os.environ.get("ORI_MODEL") + or os.environ.get("BENCHFLOW_PROVIDER_MODEL") + or _DEFAULT_MODEL + ) + self.sessions[session_id] = SessionState(cwd=cwd, model=model) + # The registry declares the private config id BenchFlow should use for + # requested effort. Do not advertise a fictitious current value when + # no effort was requested and Ori owns its default. + self._reply(request_id, {"sessionId": session_id}) + + def _set_config_option(self, request_id: object, params: dict[str, Any]) -> None: + state = self._state(params) + if params.get("configId") != "reasoning_effort": + raise ValueError(f"unknown configId: {params.get('configId')!r}") + value = params.get("value") + if value not in _EFFORTS: + raise ValueError(f"unsupported Ori reasoning effort: {value!r}") + state.reasoning_effort = str(value) + self._reply(request_id, {}) + + def _prompt(self, request_id: object, params: dict[str, Any]) -> None: + state = self._state(params) + prompt = "".join( + str(part.get("text") or "") + for part in params.get("prompt", []) + if isinstance(part, dict) and part.get("type") == "text" + ) + session_id = str(params["sessionId"]) + translator = TurnTranslator(session_id, send) + prompt_path = "" + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + prefix="benchflow-ori-prompt-", + suffix=".txt", + delete=False, + ) as prompt_file: + prompt_file.write(prompt) + prompt_path = prompt_file.name + os.chmod(prompt_path, 0o600) + command = build_ori_command( + model=state.model, + prompt_path=prompt_path, + reasoning_effort=state.reasoning_effort, + native_session_id=state.native_session_id, + ) + child_env = { + **os.environ, + "CI": "true", + "ORI_TELEMETRY": "0", + } + self.current_process = subprocess.Popen( + command, + cwd=state.cwd, + env=child_env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="utf-8", + errors="replace", + bufsize=1, + ) + assert self.current_process.stdout is not None + for line_number, line in enumerate(self.current_process.stdout, start=1): + decoded = decode_line(line, line_number) + if decoded is None: + continue + if decoded.document is None: + translator.consume_diagnostic(decoded.line_number, decoded.raw) + else: + translator.consume_document(decoded.document) + return_code = self.current_process.wait() + finally: + self.current_process = None + if prompt_path: + Path(prompt_path).unlink(missing_ok=True) + + if translator.native_session_id: + state.native_session_id = translator.native_session_id + if translator.turn_usage is not None: + state.usage = state.usage + translator.turn_usage + state.has_usage = True + + result = translator.result + if return_code != 0: + detail = ( + terminal_result_error(result) + or translator.last_diagnostic + or f"Ori exited with status {return_code}" + ) + raise RuntimeError( + f"Ori code failed with exit code {return_code}: {detail}" + ) + if result is None: + raise RuntimeError("Ori JSONL stream ended without a terminal result") + if result.get("ok") is not True: + detail = terminal_result_error(result) or "unknown Ori failure" + raise RuntimeError(f"Ori code failed: {detail}") + + response: dict[str, Any] = {"stopReason": "end_turn"} + if state.has_usage: + response["usage"] = state.usage.as_acp() + self._reply(request_id, response) + + +def main() -> None: + server = OriACPServer() + while True: + try: + message = recv() + except EOFError: + break + except Exception as exc: + print(f"ori-acp-shim input error: {exc}", file=sys.stderr) + continue + request_id = message.get("id") + try: + server.handle(message) + except Exception as exc: + if request_id is not None: + server._error(request_id, -32603, f"{type(exc).__name__}: {exc}") + + +if __name__ == "__main__": + main() + + +__all__ = [ + "ORI_BINARY", + "ORI_VERSION", + "OriACPServer", + "SessionState", + "TurnTranslator", + "build_ori_command", + "main", +] diff --git a/src/benchflow/agents/ori_events.py b/src/benchflow/agents/ori_events.py new file mode 100644 index 000000000..896544f9e --- /dev/null +++ b/src/benchflow/agents/ori_events.py @@ -0,0 +1,225 @@ +#!/usr/bin/env python3 +"""Translate decoded Ori runtime records into ACP session updates. + +Installed beside :mod:`ori_acp_shim` and :mod:`ori_jsonl`; stdlib-only. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +try: # Installed scripts are sibling top-level modules. + from ori_jsonl import OriUsage, json_text, runtime_event +except ModuleNotFoundError: # Package import used by BenchFlow's unit tests. + from .ori_jsonl import ( # type: ignore[no-redef] + OriUsage, + json_text, + runtime_event, + ) + +Send = Callable[[dict[str, Any]], None] + +_TEXT_EVENTS = frozenset({"assistant.text.delta", "content.delta"}) +_REASONING_EVENTS = frozenset({"reasoning.delta"}) +_TERMINAL_EVENTS = frozenset( + {"turn.succeeded", "turn.failed", "session.succeeded", "session.failed"} +) +_EVIDENCE_LIMIT = 8000 +_TOOL_CONTENT_LIMIT = 12000 + + +def _content(value: object) -> list[dict[str, object]]: + return [ + { + "type": "content", + "content": { + "type": "text", + "text": json_text(value)[:_TOOL_CONTENT_LIMIT], + }, + } + ] + + +def _tool_kind(name: str) -> str: + lowered = name.lower() + if lowered in {"bash", "shell", "terminal"}: + return "execute" + if lowered in {"read", "read_file"}: + return "read" + if lowered in {"write", "write_file", "edit", "apply_patch"}: + return "edit" + if lowered in {"glob", "grep", "search"}: + return "search" + if lowered in {"browser", "web", "web_search", "web_fetch"}: + return "fetch" + if "plan" in lowered or "think" in lowered: + return "think" + return "other" + + +def _tool_title(name: str, tool_input: object) -> str: + if isinstance(tool_input, dict): + values: dict[str, Any] = {str(key): value for key, value in tool_input.items()} + for key in ( + "command", + "path", + "file_path", + "pattern", + "query", + "prompt", + "name", + ): + value = values.get(key) + if isinstance(value, str) and value.strip(): + return value.strip()[:500] + return name or "tool" + + +class TurnTranslator: + """Stateful mapping from one Ori stream onto one ACP prompt.""" + + def __init__(self, session_id: str, send: Send) -> None: + self.session_id = session_id + self.native_session_id: str | None = None + self.result: dict[str, Any] | None = None + self.turn_usage: OriUsage | None = None + self.last_diagnostic = "" + self._known_tools: set[str] = set() + self._send = send + + def consume_diagnostic(self, line_number: int, raw: str) -> None: + self.last_diagnostic = raw + self._emit_text( + f"[ori diagnostic line {line_number}] {raw[:_EVIDENCE_LIMIT]}\n", + thought=True, + ) + + def consume_document(self, document: dict[str, Any]) -> None: + if document.get("kind") == "result": + self.result = document + self._remember_session_id(document.get("sessionId")) + self._emit_evidence("result", document) + return + + event = runtime_event(document) + if event is None: + self._emit_evidence("record", document) + return + + payload = event.get("payload") + payload = payload if isinstance(payload, dict) else {} + self._remember_session_id(event.get("sessionId") or payload.get("sessionId")) + event_type = str(event.get("type") or "") + + if event_type in _TEXT_EVENTS: + delta = payload.get("delta") + if isinstance(delta, str): + self._emit_text(delta) + return + if event_type in _REASONING_EVENTS: + delta = payload.get("delta") + if isinstance(delta, str): + self._emit_text(delta, thought=True) + return + if event_type == "tool.started": + self._start_tool(payload) + return + if event_type in {"tool.progress", "tool.succeeded", "tool.failed"}: + self._update_tool(event_type, payload) + return + if event_type in _TERMINAL_EVENTS: + usage = OriUsage.from_ori(payload.get("usage")) + if usage is not None: + # Some streams carry the same snapshot in both turn.* and + # session.*. Replacement avoids double-counting one turn. + self.turn_usage = usage + self._emit_evidence(event_type or "runtime.event", event) + + def _notification(self, update: dict[str, Any]) -> None: + self._send( + { + "jsonrpc": "2.0", + "method": "session/update", + "params": {"sessionId": self.session_id, "update": update}, + } + ) + + def _emit_text(self, text: str, *, thought: bool = False) -> None: + if not text: + return + self._notification( + { + "sessionUpdate": ( + "agent_thought_chunk" if thought else "agent_message_chunk" + ), + "content": {"type": "text", "text": text}, + } + ) + + def _remember_session_id(self, value: object) -> None: + if isinstance(value, str) and value: + self.native_session_id = value + + def _emit_evidence(self, label: str, value: object) -> None: + rendered = json_text(value)[:_EVIDENCE_LIMIT] + self._emit_text(f"[ori {label}] {rendered}\n", thought=True) + + def _tool_id(self, payload: dict[str, Any]) -> str: + value = payload.get("toolCallId") + if isinstance(value, str) and value: + return value + return f"ori-tool-{len(self._known_tools) + 1}" + + def _start_tool(self, payload: dict[str, Any]) -> str: + tool_id = self._tool_id(payload) + if tool_id in self._known_tools: + return tool_id + self._known_tools.add(tool_id) + name = str(payload.get("name") or "tool") + tool_input = payload.get("input") + self._notification( + { + "sessionUpdate": "tool_call", + "toolCallId": tool_id, + "title": _tool_title(name, tool_input), + "kind": _tool_kind(name), + "status": "in_progress", + } + ) + if tool_input is not None: + self._notification( + { + "sessionUpdate": "tool_call_update", + "toolCallId": tool_id, + "status": "in_progress", + "content": _content(tool_input), + } + ) + return tool_id + + def _update_tool(self, event_type: str, payload: dict[str, Any]) -> None: + tool_id = self._tool_id(payload) + if tool_id not in self._known_tools: + tool_id = self._start_tool(payload) + status = { + "tool.progress": "in_progress", + "tool.succeeded": "completed", + "tool.failed": "failed", + }[event_type] + output = payload.get("result") + if output is None: + output = payload.get("partialResult") + if output is None and event_type == "tool.failed": + output = payload.get("failure") or payload.get("error") + update: dict[str, Any] = { + "sessionUpdate": "tool_call_update", + "toolCallId": tool_id, + "status": status, + } + if output is not None: + update["content"] = _content(output) + self._notification(update) + + +__all__ = ["TurnTranslator"] diff --git a/src/benchflow/agents/ori_jsonl.py b/src/benchflow/agents/ori_jsonl.py new file mode 100644 index 000000000..73545c020 --- /dev/null +++ b/src/benchflow/agents/ori_jsonl.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +"""Typed, tolerant decoding helpers for Ori's headless JSONL stream. + +This module is installed beside the Ori ACP shim inside the sandbox. It is +therefore deliberately stdlib-only and must not import :mod:`benchflow`. + +Ori normally writes one JSON object per line, but startup/provider diagnostics +can be written as plain text before the terminal JSON result. Those lines are +evidence, not framing errors: :func:`decode_line` preserves them as diagnostic +records so callers can keep decoding later structured events. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any, Literal + + +@dataclass(frozen=True) +class DecodedLine: + """One physical Ori output line without lossy error coercion.""" + + line_number: int + kind: Literal["document", "diagnostic"] + raw: str + document: dict[str, Any] | None = None + + +@dataclass(frozen=True) +class OriUsage: + """Token usage for one Ori turn.""" + + input_tokens: int = 0 + output_tokens: int = 0 + cached_read_tokens: int = 0 + cached_write_tokens: int = 0 + thought_tokens: int = 0 + + @property + def total_tokens(self) -> int: + """ACP total: the sum of every reported token component.""" + return ( + self.input_tokens + + self.output_tokens + + self.cached_read_tokens + + self.cached_write_tokens + + self.thought_tokens + ) + + def __add__(self, other: OriUsage) -> OriUsage: + return OriUsage( + input_tokens=self.input_tokens + other.input_tokens, + output_tokens=self.output_tokens + other.output_tokens, + cached_read_tokens=(self.cached_read_tokens + other.cached_read_tokens), + cached_write_tokens=(self.cached_write_tokens + other.cached_write_tokens), + thought_tokens=self.thought_tokens + other.thought_tokens, + ) + + def as_acp(self) -> dict[str, int]: + """Return the cumulative ACP ``PromptResponse.usage`` wire shape.""" + return { + "inputTokens": self.input_tokens, + "outputTokens": self.output_tokens, + "cachedReadTokens": self.cached_read_tokens, + "cachedWriteTokens": self.cached_write_tokens, + "thoughtTokens": self.thought_tokens, + "totalTokens": self.total_tokens, + } + + @classmethod + def from_ori(cls, value: object) -> OriUsage | None: + """Decode Ori's camelCase usage object, rejecting non-mappings.""" + if not isinstance(value, dict): + return None + usage: dict[str, Any] = {str(key): item for key, item in value.items()} + return cls( + input_tokens=_nonnegative_int(usage.get("inputTokens")), + output_tokens=_nonnegative_int(usage.get("outputTokens")), + cached_read_tokens=_nonnegative_int(usage.get("cacheReadTokens")), + cached_write_tokens=_nonnegative_int(usage.get("cacheCreationTokens")), + # Ori currently exposes reasoning inside outputTokens and has no + # separate thought-token counter. Keep the ACP field explicit. + thought_tokens=0, + ) + + +def _nonnegative_int(value: object) -> int: + try: + return max(int(str(value)), 0) + except (TypeError, ValueError): + return 0 + + +def decode_line(line: str, line_number: int) -> DecodedLine | None: + """Decode one line, preserving plain/invalid JSON as a diagnostic. + + Returning ``None`` for whitespace lets the streaming caller ignore blank + framing without losing the original physical line number. + """ + raw = line.rstrip("\r\n") + if not raw.strip(): + return None + try: + value = json.loads(raw) + except json.JSONDecodeError: + return DecodedLine(line_number, "diagnostic", raw) + if not isinstance(value, dict): + return DecodedLine(line_number, "diagnostic", raw) + return DecodedLine(line_number, "document", raw, value) + + +def runtime_event(document: dict[str, Any]) -> dict[str, Any] | None: + """Unwrap an Ori ``runtime.event`` document, if this is one.""" + wrapper = document.get("event") + if not isinstance(wrapper, dict) or wrapper.get("type") != "runtime.event": + return None + event = wrapper.get("event") + return event if isinstance(event, dict) else None + + +def terminal_result_error(document: dict[str, Any] | None) -> str: + """Extract the stable human-readable error from an Ori result object.""" + if not document: + return "" + error = document.get("error") + if isinstance(error, dict) and isinstance(error.get("message"), str): + return error["message"] + if isinstance(error, str): + return error + return "" + + +def json_text(value: object) -> str: + """Stable compact rendering for raw evidence and tool content.""" + if isinstance(value, str): + return value + try: + return json.dumps(value, ensure_ascii=False, default=str, sort_keys=True) + except (TypeError, ValueError): + return str(value) + + +__all__ = [ + "DecodedLine", + "OriUsage", + "decode_line", + "json_text", + "runtime_event", + "terminal_result_error", +] diff --git a/src/benchflow/agents/registry.py b/src/benchflow/agents/registry.py index 253f602b3..66f72ba59 100644 --- a/src/benchflow/agents/registry.py +++ b/src/benchflow/agents/registry.py @@ -111,6 +111,9 @@ def _apt_install(*packages: str) -> str: _BENCHFLOW_NODE_PREFIX = "/opt/benchflow/node" _BENCHFLOW_JS_AGENT_PREFIX = "/opt/benchflow/js-agents" _BENCHFLOW_BIN_PREFIX = "/opt/benchflow/bin" +_ORI_VERSION = "0.12.0+68f9a36" +_ORI_RELEASE_TAG = "cli-0.12.0-68f9a36" +_ORI_BINARY = f"{_BENCHFLOW_BIN_PREFIX}/ori" # OpenCode-family proxy provider id. OpenCode hard-codes the OpenAI *Responses* # API for the built-in ``openai`` provider id (its ``getModel`` calls @@ -126,6 +129,62 @@ def _apt_install(*packages: str) -> str: f"{_BENCHFLOW_BIN_PREFIX}:{_BENCHFLOW_JS_AGENT_PREFIX}/bin:" f"{_BENCHFLOW_NODE_PREFIX}/bin:$PATH" ) + +# Ori is distributed as a native Bun executable, not an npm package. Pin both +# the release tag and per-platform SHA-256 so sandbox builds are reproducible +# and the install path never executes a floating remote installer. +_ORI_INSTALL = ( + f"BF_ORI_BIN={_ORI_BINARY}; " + f"BF_ORI_VERSION={_ORI_VERSION}; " + f"BF_ORI_RELEASE={_ORI_RELEASE_TAG}; " + 'BF_ORI_CURRENT=""; ' + 'if [ -x "$BF_ORI_BIN" ]; then ' + 'BF_ORI_CURRENT=$(ORI_TELEMETRY=0 "$BF_ORI_BIN" version --human ' + "2>/dev/null | awk 'NF {print $NF; exit}'); " + "fi; " + 'if [ "$BF_ORI_CURRENT" != "$BF_ORI_VERSION" ]; then ' + "if ! command -v curl >/dev/null 2>&1 || " + "! command -v sha256sum >/dev/null 2>&1; then " + "if command -v apt-get >/dev/null 2>&1; then " + "apt-get update -qq && apt-get install -y -qq curl ca-certificates coreutils; " + "elif command -v dnf >/dev/null 2>&1; then " + "dnf -y install curl ca-certificates coreutils; " + "elif command -v apk >/dev/null 2>&1; then " + "apk add --no-cache curl ca-certificates coreutils; " + "else echo 'Ori bootstrap requires curl and sha256sum' >&2; exit 127; fi; " + "fi; " + "BF_ORI_ARCH=$(uname -m); " + 'case "$BF_ORI_ARCH" in ' + "x86_64|amd64) BF_ORI_ARCH=x64 ;; " + "aarch64|arm64) BF_ORI_ARCH=arm64 ;; " + '*) echo "Unsupported architecture for Ori: $BF_ORI_ARCH" >&2; exit 1 ;; ' + "esac; " + 'BF_ORI_LIBC=""; ' + "if command -v ldd >/dev/null 2>&1 && " + "ldd --version 2>&1 | grep -qi musl; then BF_ORI_LIBC=-musl; fi; " + 'BF_ORI_ASSET="ori-linux-${BF_ORI_ARCH}${BF_ORI_LIBC}"; ' + 'case "$BF_ORI_ASSET" in ' + "ori-linux-x64) BF_ORI_SHA=2dffa9f311f8b65fbcbf6a5645c806ba623f14a003010410cd800095bc270b67 ;; " + "ori-linux-arm64) BF_ORI_SHA=d3ee260046c313a785db466db99781fb5acd91bf39ef28384a83ac293f793753 ;; " + "ori-linux-x64-musl) BF_ORI_SHA=b91bb8f01e41f5e8de496b16135db6756c4d71baa30bc230f7990db7e370f837 ;; " + "ori-linux-arm64-musl) BF_ORI_SHA=85a3be536da3337b630f2fa56f3b259961a334f5e7af75c3dd4a5a7873957d34 ;; " + '*) echo "No checksum for Ori asset: $BF_ORI_ASSET" >&2; exit 1 ;; ' + "esac; " + "BF_ORI_TMP=$(mktemp -d /tmp/benchflow-ori-install.XXXXXX); " + 'curl -fsSLo "$BF_ORI_TMP/$BF_ORI_ASSET" ' + '"https://github.com/OpenRouterLabs/ori-releases/releases/download/' + '${BF_ORI_RELEASE}/${BF_ORI_ASSET}"; ' + 'BF_ORI_ACTUAL=$(sha256sum "$BF_ORI_TMP/$BF_ORI_ASSET"); ' + "BF_ORI_ACTUAL=${BF_ORI_ACTUAL%% *}; " + 'if [ "$BF_ORI_ACTUAL" != "$BF_ORI_SHA" ]; then ' + 'echo "Ori checksum mismatch for $BF_ORI_ASSET" >&2; exit 1; fi; ' + f"mkdir -p {_BENCHFLOW_BIN_PREFIX}; " + 'mv -f "$BF_ORI_TMP/$BF_ORI_ASSET" "$BF_ORI_BIN"; ' + 'rmdir "$BF_ORI_TMP"; ' + 'chmod 755 "$BF_ORI_BIN"; ' + "fi; " + 'ORI_TELEMETRY=0 "$BF_ORI_BIN" version --human' +) # Node 22.20.0 supports OpenClaw 2026.6.9. Keep their pin pair in sync. _NODE_INSTALL = ( "export DEBIAN_FRONTEND=noninteractive; " @@ -292,6 +351,11 @@ def _js_agent_launch(binary: str, args: str = "") -> str: # Path to the deepagents ACP shim (runs LangChain's create_deep_agent as an ACP agent) _DEEPAGENTS_SHIM = (Path(__file__).parent / "deepagents_acp_shim.py").read_text() +# Ori's stdlib-only decoder, event translator, and ACP shim deploy together. +_ORI_JSONL = (Path(__file__).parent / "ori_jsonl.py").read_text() +_ORI_EVENTS = (Path(__file__).parent / "ori_events.py").read_text() +_ORI_ACP_SHIM = (Path(__file__).parent / "ori_acp_shim.py").read_text() + def _json_settings_merge(path: str, mutator: str) -> str: """Idempotent JSON-settings merge as a one-line bash snippet.""" @@ -422,6 +486,18 @@ class HostAuthFile: container_path: str # Destination in container (may use {home} placeholder) +@dataclass +class NativeSubscriptionPolicy: + """When subscription credentials may bypass the provider proxy. + + ``direct_envs`` lists agent-native login tokens that activate the same path + without a copied credential file. Model/provider eligibility is derived + from the containing :class:`SubscriptionAuth`'s ``replaces_env`` field. + """ + + direct_envs: tuple[str, ...] = () + + @dataclass class SubscriptionAuth: """Host CLI login credentials that can substitute for an API key. @@ -430,13 +506,17 @@ class SubscriptionAuth: BenchFlow detects the auth files on the host, copies them into the container, and skips the API key requirement. - ``detect_file`` is checked to determine if the user is logged in. + ``detect_file`` is checked to determine if the user is logged in. Agents + with more than one supported credential location may set ``detect_files``; + any existing path then activates subscription auth. All ``files`` are copied into the container when subscription auth is used. """ replaces_env: str # The env var this substitutes, e.g. "ANTHROPIC_API_KEY" detect_file: str # Host path to check for login, e.g. "~/.claude/.credentials.json" files: list[HostAuthFile] = field(default_factory=list) # All files to copy + detect_files: list[str] = field(default_factory=list) # Optional alternatives + native_policy: NativeSubscriptionPolicy | None = None @dataclass @@ -537,6 +617,13 @@ class AgentConfig: subscription_auth=SubscriptionAuth( replaces_env="ANTHROPIC_API_KEY", detect_file="~/.claude/.credentials.json", + native_policy=NativeSubscriptionPolicy( + direct_envs=( + "CLAUDE_CODE_OAUTH_TOKEN", + "CLAUDE_OAUTH_TOKEN", + "ANTHROPIC_AUTH_TOKEN", + ) + ), files=[ HostAuthFile( "~/.claude/.credentials.json", "{home}/.claude/.credentials.json" @@ -725,6 +812,55 @@ class AgentConfig: ), disallow_web_tools_owned_paths=["$HOME/.config/opencode"], ), + "ori": AgentConfig( + name="ori", + description=( + "OpenRouter Ori coding harness via a BenchFlow ACP-over-JSONL shim" + ), + install_cmd=( + f"{_ORI_INSTALL} && " + + _install_python_script( + f"{_BENCHFLOW_BIN_PREFIX}/ori_jsonl.py", _ORI_JSONL + ) + + " && " + + _install_python_script( + f"{_BENCHFLOW_BIN_PREFIX}/ori_events.py", _ORI_EVENTS + ) + + " && " + + _install_python_script( + f"{_BENCHFLOW_BIN_PREFIX}/ori-acp-shim", _ORI_ACP_SHIM + ) + ), + launch_cmd=f"{_BENCHFLOW_BIN_PREFIX}/ori-acp-shim", + protocol="acp", + skill_paths=["$WORKSPACE/.agents/skills"], + requires_env=["OPENROUTER_API_KEY"], + default_model="openrouter/openrouter/auto", + api_protocol="openai-completions", + env_mapping={ + "BENCHFLOW_PROVIDER_BASE_URL": "ORI_OPENROUTER_BASE_URL", + "BENCHFLOW_PROVIDER_API_KEY": "OPENROUTER_API_KEY", + "BENCHFLOW_PROVIDER_MODEL": "ORI_MODEL", + }, + subscription_auth=SubscriptionAuth( + replaces_env="OPENROUTER_API_KEY", + detect_file="~/.ori/credentials.json", + native_policy=NativeSubscriptionPolicy(), + detect_files=[ + "~/.ori/credentials.json", + "~/.openrouter/credentials.json", + ], + files=[ + HostAuthFile("~/.ori/credentials.json", "{home}/.ori/credentials.json"), + HostAuthFile( + "~/.openrouter/credentials.json", + "{home}/.openrouter/credentials.json", + ), + ], + ), + home_dirs=[".ori", ".openrouter"], + acp_effort_config_id="reasoning_effort", + ), "mimo": AgentConfig( name="mimo", description=( @@ -1065,6 +1201,7 @@ def infer_env_key_for_model(model: str) -> str | None: "gemini": "gemini", "pi": "pi-acp", "openclaw": "openclaw", + "ori": "ori", "openhands": "openhands", "oh": "openhands", "harvey-lab": "harvey-lab-harness", diff --git a/src/benchflow/providers/litellm_runtime.py b/src/benchflow/providers/litellm_runtime.py index fc53f1186..cfa6ba1e5 100644 --- a/src/benchflow/providers/litellm_runtime.py +++ b/src/benchflow/providers/litellm_runtime.py @@ -1353,6 +1353,7 @@ def _provider_models_for_proxy_alias( "LLM_BASE_URL", "OPENAI_BASE_URL", "OPENAI_API_BASE", + "ORI_OPENROUTER_BASE_URL", "AZURE_API_ENDPOINT", "AZURE_API_BASE", "AZURE_OPENAI_ENDPOINT", diff --git a/src/benchflow/rollout/__init__.py b/src/benchflow/rollout/__init__.py index 8e73b0f96..b1ac25fd6 100644 --- a/src/benchflow/rollout/__init__.py +++ b/src/benchflow/rollout/__init__.py @@ -1274,6 +1274,21 @@ def _session_factory_entrypoint(self, agent_name: str) -> str | None: return cfg.session_factory return None + def _bind_agent_connection(self, connection: tuple[Any, Any, Any, str]) -> None: + """Bind one newly-created agent session and reset session-local usage. + + Both primary and role-based connects cross this boundary. Keeping the + native-usage checkpoint here prevents cumulative counters from a prior + session being subtracted from a fresh session's smaller totals. + """ + ( + self._acp_client, + self._session, + self._session_adapter, + self._agent_name, + ) = connection + self._native_usage_checkpoint = None + async def connect(self) -> None: """Open an ACP connection to the agent. Can be called multiple times.""" cfg = self._config @@ -1300,12 +1315,7 @@ async def connect(self) -> None: sf_entrypoint = self._session_factory_entrypoint(cfg.primary_agent) self._is_session_factory = sf_entrypoint is not None if sf_entrypoint is not None: - ( - self._acp_client, - self._session, - self._session_adapter, - self._agent_name, - ) = await self._planes.connect_session_factory( + connection = await self._planes.connect_session_factory( env=self._env, agent=cfg.primary_agent, session_factory=sf_entrypoint, @@ -1317,12 +1327,7 @@ async def connect(self) -> None: agent_cwd=self._agent_cwd, ) else: - ( - self._acp_client, - self._session, - self._session_adapter, - self._agent_name, - ) = await self._planes.connect_acp( + connection = await self._planes.connect_acp( env=self._env, agent=cfg.primary_agent, agent_launch=self._agent_launch, @@ -1339,7 +1344,7 @@ async def connect(self) -> None: getattr(self, "_agent_cfg", None), ), ) - self._native_usage_checkpoint = None + self._bind_agent_connection(connection) self._reapply_ask_user_handler() self._attach_trajectory_writer(rollout_dir) @@ -2346,12 +2351,7 @@ async def connect_as(self, role: Role) -> None: sf_entrypoint = self._session_factory_entrypoint(role.agent) self._is_session_factory = sf_entrypoint is not None if sf_entrypoint is not None: - ( - self._acp_client, - self._session, - self._session_adapter, - self._agent_name, - ) = await self._planes.connect_session_factory( + connection = await self._planes.connect_session_factory( env=self._env, agent=role.agent, session_factory=sf_entrypoint, @@ -2365,12 +2365,7 @@ async def connect_as(self, role: Role) -> None: agent_cwd=self._agent_cwd, ) else: - ( - self._acp_client, - self._session, - self._session_adapter, - self._agent_name, - ) = await self._planes.connect_acp( + connection = await self._planes.connect_acp( env=self._env, agent=role.agent, agent_launch=agent_launch, @@ -2385,6 +2380,7 @@ async def connect_as(self, role: Role) -> None: role.agent, getattr(self, "_task", None), agent_cfg ), ) + self._bind_agent_connection(connection) self._reapply_ask_user_handler() self._attach_trajectory_writer(rollout_dir) self._active_role = role @@ -2564,13 +2560,12 @@ def _maybe_classify_api_error(self) -> None: # silent API failure. if not getattr(self, "_executed_prompts", None): return - # Native-subscription runs have NO usage channel: the LiteLLM proxy is + # Native-subscription runs have no proxy evidence: LiteLLM is # deliberately skipped (Harbor-style split) and the CLI authenticates - # itself, so zero tokens + zero tool calls is the expected shape of a - # HEALTHY run for agents whose trajectory carries no tool telemetry - # (e.g. omnigent's flat session events). The zero-signal heuristic is - # meaningless there and would null verifier-granted rewards; real - # failures still surface via the agent error channels. + # itself. Some agents expose trusted ACP usage, while others expose + # neither usage nor tool telemetry. Conservatively skip the + # proxy-oriented zero-signal heuristic for both; real failures still + # surface via agent errors. from benchflow.agents.env import uses_native_subscription_auth config = getattr(self, "_config", None) diff --git a/tests/test_native_acp_usage.py b/tests/test_native_acp_usage.py index ebfea42f2..b81d8f323 100644 --- a/tests/test_native_acp_usage.py +++ b/tests/test_native_acp_usage.py @@ -93,6 +93,39 @@ def test_rollout_native_acp_usage_uses_cumulative_deltas(): } +def test_binding_fresh_session_resets_native_usage_checkpoint(): + """Guards review feedback on PR #1067: connect_as cannot undercount a new session.""" + from benchflow.acp.session import ACPSession + from benchflow.rollout import Rollout + + rollout = Rollout.__new__(Rollout) + first = ACPSession("session-1") + first.record_prompt_usage( + { + "inputTokens": 80, + "outputTokens": 20, + "totalTokens": 100, + } + ) + rollout._bind_agent_connection((None, first, None, "agent")) + rollout._collect_native_acp_usage() + + second = ACPSession("session-2") + second.record_prompt_usage( + { + "inputTokens": 7, + "outputTokens": 3, + "totalTokens": 10, + } + ) + rollout._bind_agent_connection((None, second, None, "agent")) + rollout._collect_native_acp_usage() + + assert rollout._native_usage_metrics["n_input_tokens"] == 87 + assert rollout._native_usage_metrics["n_output_tokens"] == 23 + assert rollout._native_usage_metrics["total_tokens"] == 110 + + def test_rollout_provider_usage_wins_over_native_acp_usage(): """Guards PR #613 follow-up: LiteLLM provider telemetry remains authoritative.""" from benchflow.rollout import Rollout diff --git a/tests/test_ori_agent.py b/tests/test_ori_agent.py new file mode 100644 index 000000000..e607bc7c2 --- /dev/null +++ b/tests/test_ori_agent.py @@ -0,0 +1,292 @@ +"""OpenRouter Ori ACP-shim integration tests.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from benchflow.agents import ori_acp_shim as shim +from benchflow.agents.ori_events import TurnTranslator +from benchflow.agents.ori_jsonl import OriUsage, decode_line +from benchflow.agents.registry import AGENTS + + +def _runtime(event_type: str, payload: dict, *, session_id: str = "native-ori"): + return { + "kind": "event", + "event": { + "type": "runtime.event", + "event": { + "type": event_type, + "payload": payload, + "sessionId": session_id, + }, + }, + } + + +def test_ori_decoder_preserves_diagnostic_before_structured_result() -> None: + """Guards review feedback on PR #1067: plain prelude must not lose JSON evidence.""" + diagnostic = decode_line("provider bootstrap warning\n", 1) + result = decode_line('{"kind":"result","ok":false}\n', 2) + + assert diagnostic is not None + assert diagnostic.kind == "diagnostic" + assert diagnostic.raw == "provider bootstrap warning" + assert result is not None + assert result.kind == "document" + assert result.document == {"kind": "result", "ok": False} + + +def test_ori_usage_total_includes_cache_components() -> None: + """Guards review feedback on PR #1067: ACP total includes every component.""" + usage = OriUsage.from_ori( + { + "inputTokens": 10, + "outputTokens": 6, + "cacheReadTokens": 3, + "cacheCreationTokens": 2, + "contextTokens": 999, + } + ) + + assert usage is not None + assert usage.as_acp() == { + "inputTokens": 10, + "outputTokens": 6, + "cachedReadTokens": 3, + "cachedWriteTokens": 2, + "thoughtTokens": 0, + "totalTokens": 21, + } + + +def test_ori_failed_result_remains_available_after_plain_diagnostic() -> None: + """Guards review feedback on PR #1067: failure diagnostics and result both survive.""" + messages: list[dict] = [] + translator = TurnTranslator("acp-session", messages.append) + translator.consume_diagnostic(1, "provider rejected request") + translator.consume_document( + { + "kind": "result", + "ok": False, + "error": {"message": "invalid model"}, + "sessionId": "native-failed", + } + ) + + assert translator.result is not None + assert translator.result["error"]["message"] == "invalid model" + assert translator.native_session_id == "native-failed" + evidence = "".join( + message["params"]["update"]["content"]["text"] for message in messages + ) + assert "provider rejected request" in evidence + assert "invalid model" in evidence + + +def test_ori_command_is_argv_and_resumes_native_session() -> None: + """Guards PR #1067: prompt text stays in a file and follow-up turns resume.""" + command = shim.build_ori_command( + model="anthropic/claude-sonnet-4.6", + prompt_path="/tmp/prompt with spaces.txt", + reasoning_effort="max", + native_session_id="native-session", + ) + + assert command[:5] == [ + shim.ORI_BINARY, + "code", + "--harness", + "ori", + "--model", + ] + assert command[command.index("--prompt-file") + 1] == ( + "/tmp/prompt with spaces.txt" + ) + assert command[command.index("--session") + 1] == "native-session" + assert command[command.index("--reasoning-effort") + 1] == "max" + + +def test_ori_acp_server_streams_tools_diagnostics_usage_and_resume( + monkeypatch, tmp_path: Path +) -> None: + """Guards PR #1067: Ori is an honest multi-turn ACP adapter with evidence.""" + fake_ori = tmp_path / "fake-ori" + argv_log = tmp_path / "argv.jsonl" + fake_ori.write_text( + "#!/usr/bin/env python3\n" + "import json, os, pathlib, sys\n" + "args = sys.argv[1:]\n" + "prompt = pathlib.Path(args[args.index('--prompt-file') + 1]).read_text()\n" + "with open(os.environ['ORI_FAKE_ARGV_LOG'], 'a', encoding='utf-8') as f:\n" + " f.write(json.dumps({'args': args, 'prompt': prompt}) + '\\n')\n" + "print('provider bootstrap warning')\n" + "def runtime(kind, payload):\n" + " return {'kind':'event','event':{'type':'runtime.event','event':" + "{'type':kind,'payload':payload,'sessionId':'native-ori'}}}\n" + "print(json.dumps(runtime('tool.started', {'toolCallId':'call-1'," + "'name':'bash','input':{'command':'pwd'}})))\n" + "print(json.dumps(runtime('tool.succeeded', {'toolCallId':'call-1'," + "'name':'bash','result':'/workspace'})))\n" + "print(json.dumps(runtime('assistant.text.delta', {'delta':'done'})))\n" + "print(json.dumps(runtime('turn.succeeded', {'usage':{'inputTokens':10," + "'outputTokens':6,'cacheReadTokens':3,'cacheCreationTokens':2}})))\n" + "print(json.dumps({'kind':'result','ok':True,'sessionId':'native-ori'}))\n", + encoding="utf-8", + ) + fake_ori.chmod(0o755) + monkeypatch.setattr(shim, "ORI_BINARY", str(fake_ori)) + monkeypatch.setenv("ORI_FAKE_ARGV_LOG", str(argv_log)) + messages: list[dict] = [] + monkeypatch.setattr(shim, "send", messages.append) + + server = shim.OriACPServer() + server.sessions["acp-session"] = shim.SessionState( + cwd=str(tmp_path), model="anthropic/claude-sonnet-4.6", reasoning_effort="max" + ) + prompt_request = { + "jsonrpc": "2.0", + "id": 1, + "method": "session/prompt", + "params": { + "sessionId": "acp-session", + "prompt": [{"type": "text", "text": "fix $(not-shell)"}], + }, + } + + server.handle(prompt_request) + prompt_request["id"] = 2 + server.handle(prompt_request) + + responses = [message for message in messages if "result" in message] + assert responses[-1]["result"] == { + "stopReason": "end_turn", + "usage": { + "inputTokens": 20, + "outputTokens": 12, + "cachedReadTokens": 6, + "cachedWriteTokens": 4, + "thoughtTokens": 0, + "totalTokens": 42, + }, + } + from benchflow.acp.types import PromptResult + + assert PromptResult.model_validate(responses[-1]["result"]).stop_reason == ( + "end_turn" + ) + updates = [ + message["params"]["update"] + for message in messages + if message.get("method") == "session/update" + ] + assert any(update["sessionUpdate"] == "tool_call" for update in updates) + assert any( + update["sessionUpdate"] == "tool_call_update" + and update["status"] == "completed" + for update in updates + ) + assert any( + update["sessionUpdate"] == "agent_message_chunk" + and update["content"]["text"] == "done" + for update in updates + ) + thought_text = "".join( + update["content"]["text"] + for update in updates + if update["sessionUpdate"] == "agent_thought_chunk" + ) + assert "provider bootstrap warning" in thought_text + assert '"kind": "result"' in thought_text + + invocations = [json.loads(line) for line in argv_log.read_text().splitlines()] + assert invocations[0]["prompt"] == "fix $(not-shell)" + assert "--session" not in invocations[0]["args"] + assert invocations[1]["args"][invocations[1]["args"].index("--session") + 1] == ( + "native-ori" + ) + + +def test_ori_registry_contract_is_pinned_acp_and_manifest_expressible() -> None: + """Guards review feedback on PR #1067: Ori uses the canonical ACP contract.""" + cfg = AGENTS["ori"] + + assert cfg.protocol == "acp" + assert cfg.session_factory == "" + assert cfg.launch_cmd == "/opt/benchflow/bin/ori-acp-shim" + assert cfg.default_model == "openrouter/openrouter/auto" + assert cfg.api_protocol == "openai-completions" + assert cfg.acp_effort_config_id == "reasoning_effort" + assert cfg.env_mapping == { + "BENCHFLOW_PROVIDER_BASE_URL": "ORI_OPENROUTER_BASE_URL", + "BENCHFLOW_PROVIDER_API_KEY": "OPENROUTER_API_KEY", + "BENCHFLOW_PROVIDER_MODEL": "ORI_MODEL", + } + assert "cli-0.12.0-68f9a36" in cfg.install_cmd + assert "2dffa9f311f8b65f" in cfg.install_cmd + assert "d3ee260046c313a7" in cfg.install_cmd + assert "ori-releases/releases/download" in cfg.install_cmd + assert "ori-acp-shim" in cfg.install_cmd + assert "ori_events.py" in cfg.install_cmd + assert "ori_jsonl.py" in cfg.install_cmd + assert "install.sh" not in cfg.install_cmd + assert cfg.subscription_auth is not None + assert cfg.subscription_auth.native_policy is not None + assert cfg.subscription_auth.detect_files == [ + "~/.ori/credentials.json", + "~/.openrouter/credentials.json", + ] + + +def test_ori_native_login_policy_only_applies_to_openrouter_models() -> None: + """Guards PR #1067: typed subscription policy cannot bypass another provider.""" + from benchflow.agents.env import uses_native_subscription_auth + + marker = {"_BENCHFLOW_SUBSCRIPTION_AUTH": "1"} + assert uses_native_subscription_auth( + "ori", "openrouter/anthropic/claude-sonnet-4.6", marker + ) + assert not uses_native_subscription_auth( + "ori", + "openrouter/anthropic/claude-sonnet-4.6", + {**marker, "OPENROUTER_API_KEY": "api-key-wins"}, + ) + assert not uses_native_subscription_auth( + "ori", "deepseek/deepseek-v4-flash", marker + ) + + +def test_ori_login_detects_openrouter_fallback_file( + monkeypatch, tmp_path: Path +) -> None: + """Guards PR #1067: either official Ori credential location is detected.""" + from benchflow.agents.env import check_subscription_auth + + fallback = tmp_path / ".openrouter" / "credentials.json" + fallback.parent.mkdir() + fallback.write_text('{"key":"saved"}') + auth = AGENTS["ori"].subscription_auth + assert auth is not None + monkeypatch.setattr( + auth, + "detect_files", + [str(tmp_path / ".ori" / "credentials.json"), str(fallback)], + ) + + assert check_subscription_auth("ori", "OPENROUTER_API_KEY") + + +def test_ori_openrouter_provider_maps_native_cli_environment() -> None: + """Guards PR #1067: OpenRouter provider routing reaches Ori's native env.""" + from benchflow.agents.env import resolve_agent_env + + resolved = resolve_agent_env( + "ori", + "openrouter/anthropic/claude-sonnet-4.6", + {"OPENROUTER_API_KEY": "test-openrouter-key"}, + ) + + assert resolved["ORI_OPENROUTER_BASE_URL"] == "https://openrouter.ai/api/v1" + assert resolved["OPENROUTER_API_KEY"] == "test-openrouter-key" + assert resolved["ORI_MODEL"] == "anthropic/claude-sonnet-4.6" diff --git a/tests/test_subscription_auth.py b/tests/test_subscription_auth.py index 45da6fb54..61c22d4d7 100644 --- a/tests/test_subscription_auth.py +++ b/tests/test_subscription_auth.py @@ -316,7 +316,12 @@ def test_anthropic_subscription_gate_is_registry_driven(self, monkeypatch): in their registration. Agents without it must never skip the proxy. """ from benchflow.agents.env import uses_native_subscription_auth - from benchflow.agents.registry import AGENTS, AgentConfig, SubscriptionAuth + from benchflow.agents.registry import ( + AGENTS, + AgentConfig, + NativeSubscriptionPolicy, + SubscriptionAuth, + ) cfg = AgentConfig( name="fake-claude-cli", @@ -325,6 +330,9 @@ def test_anthropic_subscription_gate_is_registry_driven(self, monkeypatch): subscription_auth=SubscriptionAuth( replaces_env="ANTHROPIC_API_KEY", detect_file="~/.claude/.credentials.json", + native_policy=NativeSubscriptionPolicy( + direct_envs=("CLAUDE_CODE_OAUTH_TOKEN",) + ), ), ) monkeypatch.setitem(AGENTS, "fake-claude-cli", cfg)