diff --git a/evaluation/benchmarks/swe_bench/scripts/run_infer.sh b/evaluation/benchmarks/swe_bench/scripts/run_infer.sh new file mode 100755 index 000000000000..7226977d853a --- /dev/null +++ b/evaluation/benchmarks/swe_bench/scripts/run_infer.sh @@ -0,0 +1,141 @@ +#!/usr/bin/env bash +# Bench entry script — invoked by gym's OpenCodeHarnessProcessor.get_run_command(). +# +# Args (positional, must match the order in app.py's get_run_command): +# $1 COMMIT_HASH opencode commit (informational; checkout is done at setup) +# $2 AGENT agent class name (informational) +# $3 MAX_ITER max agent turns +# $4 DATASET dataset name (informational; gym already dispatched) +# $5 SPLIT dataset split (informational) +# $6 EVAL_OUTPUT_DIR where to write trajectories (relative to opencode dir) +# $7 SELECTED_ID instance_id to run +# $8 INSTANCE_DICT_PATH /root/dataset/data.jsonl (single-line JSONL) +# $9 CONFIG_FILE opencode model config JSON (written by gym) +# $10 WORKSPACE_ROOT resolved repo path inside the SIF (gym side decided) +# $11 USER_MESSAGE_PATH pre-rendered user prompt file (workspace baked in) +# $12 SYSTEM_PROMPT_PATH optional system-prompt override +# $13 REPLAY_MESSAGES_PATH optional JSON file of prior chat-completion +# messages to replay before continuing live +# $14 REPLAY_SUBAGENTS_PATH optional causal subagent replay manifest +# +# Environment (set by gym): +# NEMO_GYM_MODEL_SERVER_NAME proxy name on the gym head server +# NEMO_GYM_MODEL_SERVER_BASE_URL base http://host:port for the model server +# NEMO_GYM_METRICS_FPATH path to the metrics JSON to update +# NEMO_GYM_CONFIG_DICT (informational) the gym YAML config blob +# COMMAND_EXEC_TIMEOUT per-bash-command timeout in seconds +# DIVERSIFY_TOOL_NAMES optional: rename tools for RL diversity +# CAMEL_CASE_TOOL_NAMES optional: camelCase tool names +# PATCH_MODE optional: how the model patch is extracted. +# `worktree` (default) = `git diff` of the +# working tree; `committed` = diff of what the +# agent committed, for task families whose +# prompt asks the agent to commit its work. + +set -eo pipefail + +COMMIT_HASH="${1:-}" +AGENT="${2:-OpenCodeAgent}" +MAX_ITER="${3:-100}" +DATASET="${4:-}" +SPLIT="${5:-test}" +EVAL_OUTPUT_DIR="${6:-evaluation/oh}" +SELECTED_ID="${7:-}" +INSTANCE_DICT_PATH="${8:-/root/dataset/data.jsonl}" +CONFIG_FILE="${9:-/tmp/oc_config.json}" +WORKSPACE_ROOT="${10:-}" +USER_MESSAGE_PATH="${11:-}" +SYSTEM_PROMPT_PATH="${12:-}" +REPLAY_MESSAGES_PATH="${13:-}" +REPLAY_SUBAGENTS_PATH="${14:-}" + +if [ -z "$SELECTED_ID" ]; then + echo "ERROR: SELECTED_ID (\$7) is required." + exit 64 +fi +if [ -z "$WORKSPACE_ROOT" ]; then + echo "ERROR: WORKSPACE_ROOT (\$10) is required — gym side resolves the dataset-aware repo path." + exit 65 +fi +if [ -z "$USER_MESSAGE_PATH" ]; then + echo "ERROR: USER_MESSAGE_PATH (\$11) is required — gym side renders the user prompt." + exit 66 +fi +if [ -z "${NEMO_GYM_MODEL_SERVER_NAME:-}" ]; then + echo "ERROR: NEMO_GYM_MODEL_SERVER_NAME not set in env." + exit 67 +fi +if [ -z "${NEMO_GYM_MODEL_SERVER_BASE_URL:-}" ]; then + echo "ERROR: NEMO_GYM_MODEL_SERVER_BASE_URL not set in env." + exit 68 +fi + +# Resolve the opencode root directory. The script lives at +# evaluation/benchmarks/swe_bench/scripts/run_infer.sh — go up four levels. +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +OPENCODE_DIR="$(cd "$SCRIPT_DIR/../../../.." && pwd)" +BENCH_CLI="$OPENCODE_DIR/packages/opencode/src/bench/cli.ts" + +if [ ! -f "$BENCH_CLI" ]; then + echo "ERROR: bench cli.ts not found at $BENCH_CLI" + exit 69 +fi +if ! command -v bun >/dev/null 2>&1; then + echo "ERROR: bun not on PATH (expected /opencode_setup/bun/bin/bun)" + exit 70 +fi + +# Make EVAL_OUTPUT_DIR absolute (relative to opencode dir). +case "$EVAL_OUTPUT_DIR" in + /*) ABS_OUTPUT_DIR="$EVAL_OUTPUT_DIR" ;; + *) ABS_OUTPUT_DIR="$OPENCODE_DIR/$EVAL_OUTPUT_DIR" ;; +esac +mkdir -p "$ABS_OUTPUT_DIR" + +echo "OPENCODE_DIR: $OPENCODE_DIR" +echo "BENCH_CLI: $BENCH_CLI" +echo "AGENT: $AGENT COMMIT: $COMMIT_HASH MAX_ITER: $MAX_ITER" +echo "DATASET: $DATASET SPLIT: $SPLIT SELECTED_ID: $SELECTED_ID" +echo "EVAL_OUTPUT_DIR: $ABS_OUTPUT_DIR" +echo "INSTANCE_DICT_PATH: $INSTANCE_DICT_PATH" +echo "CONFIG_FILE: $CONFIG_FILE" +echo "WORKSPACE_ROOT: $WORKSPACE_ROOT" +echo "USER_MESSAGE_PATH: $USER_MESSAGE_PATH" +echo "PATCH_MODE: ${PATCH_MODE:-worktree (default)}" +echo "SYSTEM_PROMPT_PATH: $SYSTEM_PROMPT_PATH" +echo "REPLAY_MESSAGES_PATH: $REPLAY_MESSAGES_PATH" +echo "REPLAY_SUBAGENTS_PATH: $REPLAY_SUBAGENTS_PATH" +echo "MODEL_SERVER: $NEMO_GYM_MODEL_SERVER_NAME @ $NEMO_GYM_MODEL_SERVER_BASE_URL" + +cmd=( + bun "$BENCH_CLI" + --instance-dict-path "$INSTANCE_DICT_PATH" + --output-dir "$ABS_OUTPUT_DIR" + --config "$CONFIG_FILE" + --max-turns "$MAX_ITER" + --agent-cls "$AGENT" + --dataset "$DATASET" + --split "$SPLIT" + --selected-id "$SELECTED_ID" + --workspace-root "$WORKSPACE_ROOT" + --user-message-file "$USER_MESSAGE_PATH" +) +if [ -n "$SYSTEM_PROMPT_PATH" ]; then + cmd+=(--system-prompt "$SYSTEM_PROMPT_PATH") +fi +if [ -n "$REPLAY_MESSAGES_PATH" ]; then + cmd+=(--replay-messages-file "$REPLAY_MESSAGES_PATH") +fi +if [ -n "$REPLAY_SUBAGENTS_PATH" ]; then + cmd+=(--replay-subagents-file "$REPLAY_SUBAGENTS_PATH") +fi +if [ "${ENABLE_SUBAGENTS:-0}" = "1" ] || [ "${ENABLE_SUBAGENTS:-}" = "true" ]; then + cmd+=(--enable-subagents) +fi +# Omitted entirely when unset so cli.ts keeps its own default (`worktree`). +if [ -n "${PATCH_MODE:-}" ]; then + cmd+=(--patch-mode "$PATCH_MODE") +fi + +echo "Executing: ${cmd[*]}" +exec "${cmd[@]}" diff --git a/packages/opencode/src/bench/bootstrap_repo.ts b/packages/opencode/src/bench/bootstrap_repo.ts new file mode 100644 index 000000000000..82c9c61a2353 --- /dev/null +++ b/packages/opencode/src/bench/bootstrap_repo.ts @@ -0,0 +1,78 @@ +/** + * Bootstrap a git repository inside the workspace when the SIF ships a flat + * source tree without a `.git` directory. + * + * Some dataset SIFs (notably `swe-bench-ext`, and certain SWE-rebench variants) + * copy the repo contents into `/workspace/repo` (or the dataset-specific path) + * without preserving git history. Without `.git`, `runDeepReset` is a silent + * no-op (its `git rev-parse` fails under the outer `|| true`) and + * `captureGitDiff` returns "" — every rollout is recorded as `patch=0 bytes` + * regardless of what the agent did. Port of nv-OpenHands' + * `evaluation/benchmarks/swe_bench/run_infer.py:1142-1156`. + * + * If `.git` already exists, this is a no-op. Otherwise a pristine baseline + * commit is created and tagged `opencode_bench_baseline`. Callers should skip + * `runDeepReset` when this returns `{ freshInit: true }` — the dataset's + * upstream `base_commit` SHA does not exist in the fresh repo, so deep_reset + * would just fail rev-parse and noisily fall through to its nuclear pass. + */ + +import { spawn } from "node:child_process" +import { existsSync } from "node:fs" +import path from "node:path" + +function detectShell(): string | null { + for (const p of ["/bin/bash", "/usr/bin/bash", "/bin/sh", "/usr/bin/sh"]) { + if (existsSync(p)) return p + } + return null +} + +function shellQuote(s: string): string { + return `'${s.replace(/'/g, `'\\''`)}'` +} + +function buildBootstrapCmd(workspaceRoot: string): string { + const q = shellQuote(workspaceRoot) + return ( + `cd ${q} && ` + + `echo "[bootstrap_repo] initializing git repo at ${workspaceRoot}" && ` + + `git config --global --add safe.directory ${q} && ` + + `git init -q && ` + + `git config user.email 'bench@opencode.local' && ` + + `git config user.name 'opencode bench' && ` + + `git add -A && ` + + `git commit -q --allow-empty -m 'opencode bench baseline' && ` + + `git tag -f opencode_bench_baseline HEAD && ` + + `echo "[bootstrap_repo] done; HEAD=$(git rev-parse --short HEAD)"` + ) +} + +export interface BootstrapResult { + freshInit: boolean +} + +export async function bootstrapRepoIfMissing(workspaceRoot: string): Promise { + if (existsSync(path.join(workspaceRoot, ".git"))) { + return { freshInit: false } + } + const shell = detectShell() + if (!shell) { + console.warn(`[bench] bootstrap_repo skipped: no shell found at /bin/{bash,sh} or /usr/bin/{bash,sh}`) + return { freshInit: false } + } + const cmd = buildBootstrapCmd(workspaceRoot) + console.log(`[bench] bootstrap_repo workspace=${workspaceRoot} shell=${shell}`) + const exitCode = await new Promise((resolve) => { + const child = spawn(shell, ["-c", cmd], { + stdio: ["ignore", "inherit", "inherit"], + }) + child.on("close", (code) => resolve(code ?? 0)) + child.on("error", (err) => { + console.warn(`[bench] bootstrap_repo spawn error: ${err}`) + resolve(1) + }) + }) + console.log(`[bench] bootstrap_repo exit=${exitCode}`) + return { freshInit: exitCode === 0 } +} diff --git a/packages/opencode/src/bench/cli.ts b/packages/opencode/src/bench/cli.ts new file mode 100644 index 000000000000..6a566612a6be --- /dev/null +++ b/packages/opencode/src/bench/cli.ts @@ -0,0 +1,632 @@ +/** + * SWE-bench bench CLI driver. + * + * Drives a single SWE-bench instance to completion using opencode's REAL + * agentic loop. We spawn `bun .../src/index.ts run` as a subprocess (with a + * per-instance opencode config that registers our `nemo-gym` provider, a + * SWE-bench agent, and disables compaction) and let it run to idle. + * + * Why subprocess instead of in-process Server.Default? Subprocess is the + * model the user-facing `opencode run` already uses (cli/cmd/run.ts:670–675 + * also uses an in-process fetch but the public entry is `bun .../index.ts`). + * A subprocess gives us: + * - clean process isolation per instance (matters for many parallel SIFs) + * - identical bootstrapping path to `opencode run`, so we don't drift + * - a compact event-type stream on stdout + * + * Trajectory capture: the nemo-gym provider (registered via this config) + * writes `/.json` per LLM call BEFORE returning. On + * exit we capture the model patch (see ./patch.ts for the two modes) and write + * `output.jsonl`. + */ + +import { existsSync, promises as fs, readFileSync } from "node:fs" +import path from "node:path" +import os from "node:os" +import { spawn } from "node:child_process" +import { runDeepReset } from "./deep_reset" +import { bootstrapRepoIfMissing } from "./bootstrap_repo" +import { capturePatch, ensureCommitIdentity, parsePatchMode, recordBaselineCommit, type PatchMode } from "./patch" +import * as BenchTerminalError from "./terminal_error" +// opencode's built-in anthropic system prompt — Bun bundles .txt as a string. +// Used as the default when no --system-prompt override is passed. +import PROMPT_ANTHROPIC from "../session/prompt/anthropic.txt" +import type { NemoGymReplayManifest, NemoGymReplayTurn } from "../provider/sdk/nemo-gym/language-model" +import { parseReplayManifest, parseReplayMessages } from "./replay" + +interface CliArgs { + instanceDictPath: string + outputDir: string + config: string + maxTurns: number + agentCls: string + dataset: string + split: string + selectedId: string + /** Resolved repo path inside the SIF — gym side decided based on dataset_name. */ + workspaceRoot: string + /** Pre-rendered user message file (workspace_path baked in by gym). */ + userMessageFile: string + systemPromptPath?: string + /** Enable opencode's `task` tool (spawns subagent sessions). */ + enableSubagents: boolean + /** + * Path to a JSON file of prior chat-completion-format messages to replay + * before continuing live (trajectory resume). See language-model.ts. + */ + replayMessagesFile?: string + /** Causal parent-task-call -> recorded child-session replay graph. */ + replaySubagentsFile?: string + /** How the model patch is extracted at the end of the run. See ./patch.ts. */ + patchMode: PatchMode +} + +function parseArgs(argv: string[]): CliArgs { + const out: Partial = { + maxTurns: 100, + agentCls: "OpenCodeAgent", + dataset: "", + split: "test", + enableSubagents: false, + patchMode: parsePatchMode(undefined), + } + for (let i = 0; i < argv.length; i++) { + const a = argv[i] + const next = () => argv[++i] + switch (a) { + case "--instance-dict-path": + out.instanceDictPath = next() + break + case "--output-dir": + out.outputDir = next() + break + case "--config": + out.config = next() + break + case "--max-turns": + out.maxTurns = parseInt(next(), 10) + break + case "--agent-cls": + out.agentCls = next() + break + case "--dataset": + out.dataset = next() + break + case "--split": + out.split = next() + break + case "--selected-id": + out.selectedId = next() + break + case "--workspace-root": + out.workspaceRoot = next() + break + case "--user-message-file": + out.userMessageFile = next() + break + case "--system-prompt": + out.systemPromptPath = next() + break + case "--enable-subagents": + out.enableSubagents = true + break + case "--replay-messages-file": + out.replayMessagesFile = next() + break + case "--replay-subagents-file": + out.replaySubagentsFile = next() + break + case "--patch-mode": + out.patchMode = parsePatchMode(next()) + break + default: + if (a.startsWith("--")) throw new Error(`Unknown flag: ${a}`) + } + } + for (const required of [ + "instanceDictPath", + "outputDir", + "config", + "selectedId", + "workspaceRoot", + "userMessageFile", + ] as const) { + if (!out[required]) + throw new Error(`Missing required arg --${required.replace(/[A-Z]/g, (c) => "-" + c.toLowerCase())}`) + } + return out as CliArgs +} + +interface InstanceDict { + instance_id: string + problem_statement: string + repo?: string + repo_name?: string + workspace?: string + base_commit?: string + [key: string]: unknown +} + +async function readInstance(instanceDictPath: string, selectedId: string): Promise { + const text = await fs.readFile(instanceDictPath, "utf8") + const lines = text + .split("\n") + .map((l) => l.trim()) + .filter(Boolean) + const records = lines.map((l) => JSON.parse(l) as InstanceDict) + const match = records.find((r) => r.instance_id === selectedId) ?? records[0] + if (!match) throw new Error(`No instance found in ${instanceDictPath}`) + return match +} + +function loadGymConfig(configPath: string): Record { + return JSON.parse(readFileSync(configPath, "utf8")) +} + +const DEFAULT_SYSTEM_PROMPT = PROMPT_ANTHROPIC + +async function buildConfigDir(args: { + instanceId: string + modelName: string + baseURL: string + completionsDir: string + maxTurns: number + systemPromptPath?: string + enableSubagents: boolean + /** Forced sampling params (RL on-policy requirement); from gym llm.model config. */ + temperature?: number + topP?: number + maxTokens?: number + /** Scripted assistant turns to replay before the agent continues live. */ + replayTurns?: NemoGymReplayTurn[] + /** Subsequent user messages trailing the last replayed turn. */ + replayTrailingUserTexts?: string[] + /** Per-recorded-subagent replay queues and their parent task-call links. */ + replayManifest?: NemoGymReplayManifest +}): Promise<{ tmpRoot: string; configFile: string }> { + const tmpRoot = await fs.mkdtemp(path.join(os.tmpdir(), `bench-${args.instanceId}-`)) + await fs.mkdir(tmpRoot, { recursive: true }) + + const systemPrompt = args.systemPromptPath ? await fs.readFile(args.systemPromptPath, "utf8") : DEFAULT_SYSTEM_PROMPT + + const cfg: Record = { + $schema: "https://opencode.ai/config.json", + provider: { + "nemo-gym": { + npm: "@opencode-ai/nemo-gym", + options: { + baseURL: args.baseURL, + completionsDir: args.completionsDir, + instanceId: args.instanceId, + // Unlimited: model-server backpressure can stall a request for + // tens of minutes at high shard concurrency; we'd rather wait + // than tear down the session and produce an empty model_patch. + // requestTimeoutMs<=0 disables the abort timer in the provider. + retries: Number.MAX_SAFE_INTEGER, + requestTimeoutMs: 0, + // Forced sampling params: NeMo-RL's vLLM worker asserts every + // request's temperature/top_p match the training generation config + // exactly (on-policy). Passed by gym via the llm.model config block. + ...(args.temperature !== undefined ? { temperature: args.temperature } : {}), + ...(args.topP !== undefined ? { topP: args.topP } : {}), + ...(args.maxTokens !== undefined ? { maxTokens: args.maxTokens } : {}), + ...(args.replayTurns?.length ? { replayTurns: args.replayTurns } : {}), + ...(args.replayTrailingUserTexts?.length ? { replayTrailingUserTexts: args.replayTrailingUserTexts } : {}), + ...(args.replayManifest ? { replayManifest: args.replayManifest } : {}), + }, + models: { + [args.modelName]: { + id: args.modelName, + name: args.modelName, + limit: { context: 131072, output: 32768 }, + tool_call: true, + temperature: true, + }, + }, + }, + }, + agent: { + title: { + disable: true, + }, + "swe-bench": { + mode: "primary", + model: `nemo-gym/${args.modelName}`, + prompt: systemPrompt, + // Allow the read+write tool set; disable web/skill/task to keep the + // agent focused on local code editing. + permission: { + edit: { "**": "allow" }, + bash: { + "*": "allow", + + // process termination + "*killall*": "deny", + "*pkill*": "deny", + "*kill -1*": "deny", + "*kill 0*": "deny", + + // filesystem destruction + "*rm -rf /": "deny", + "*rm -rf /*": "deny", + "*rm -rf /bin*": "deny", + "*rm -rf /usr*": "deny", + "*rm -rf /etc*": "deny", + "*rm -rf /var*": "deny", + "*rm -rf /home*": "deny", + "*rm -rf /root*": "deny", + "*rm -rf /opt*": "deny", + "*rm -rf /lib*": "deny", + "*rm -rf /lib64*": "deny", + "*rm -rf /sbin*": "deny", + "*rm -rf /boot*": "deny", + "*rm -rf /dev*": "deny", + "*rm -rf /proc*": "deny", + "*rm -rf /sys*": "deny", + + // system control + // "*shutdown*": "deny", + // "*reboot*": "deny", + // "*poweroff*": "deny", + // "*halt*": "deny", + // "init 0*": "deny", + // "init 6*": "deny", + + // disk devices + "dd *of=/dev/sd*": "deny", + "dd *of=/dev/nvme*": "deny", + "dd *of=/dev/hd*": "deny", + "dd *of=/dev/null*": "deny", + + // git network + "*git fetch*": "deny", + "*git pull*": "deny", + "*git clone*": "deny", + "*git ls-remote*": "deny", + "*git remote add*": "deny", + "*git remote set-url*": "deny", + "*git remote set-head*": "deny", + "*git remote update*": "deny", + "*git remote rename*": "deny", + "*git remote set-branches*": "deny", + "*git submodule add*": "deny", + "*git submodule update*": "deny", + "*git submodule sync*": "deny", + "*git submodule init*": "deny", + "*git archive*--remote*": "deny", + "*git *://*": "deny", + "*git *@*:*": "deny", + + // git history mining + "*git log*--all*": "deny", + "*git log*--branches*": "deny", + "*git log*--remotes*": "deny", + "*git log*--walk-reflogs*": "deny", + "*git log*--grep*": "deny", + "*git rev-list*--all*": "deny", + "*git rev-list*--branches*": "deny", + "*git rev-list*--remotes*": "deny", + "*git rev-list*--grep*": "deny", + "*git shortlog*--all*": "deny", + "*git reflog*": "deny", + "*git cat-file*": "deny", + "*git fsck*": "deny", + "*git verify-pack*": "deny", + "*git unpack-objects*": "deny", + "*git cherry*": "deny", + "*git show*": "deny", + "*git merge-base*--is-ancestor*": "deny", + "*git branch*--contains*": "deny", + "*git tag*--contains*": "deny", + "*git for-each-ref*--contains*": "deny", + + // git internals (substring match on path) + "*.git/logs*": "deny", + "*.git/packed-refs*": "deny", + "*.git/ORIG_HEAD*": "deny", + "*.git/FETCH_HEAD*": "deny", + "*.git/refs*": "deny", + + // online lookups + "*curl *github.com*": "deny", + "*wget *github.com*": "deny", + "*curl *githubusercontent.com*": "deny", + "*wget *githubusercontent.com*": "deny", + "*curl *github.io*": "deny", + "*wget *github.io*": "deny", + }, + }, + tools: { + bash: true, + edit: true, + read: true, + glob: true, + grep: true, + write: true, + apply_patch: true, + webfetch: false, + websearch: false, + task: args.enableSubagents, + skill: false, + todowrite: true, + }, + steps: args.maxTurns, + options: {}, + }, + }, + compaction: { auto: false }, + share: "manual", + } + + const configFile = path.join(tmpRoot, "opencode.jsonc") + await fs.writeFile(configFile, JSON.stringify(cfg, null, 2)) + + return { tmpRoot, configFile } +} + +function runOpencode(args: { + workspaceRoot: string + modelName: string + message: string + env: NodeJS.ProcessEnv + opencodeBin: string + agent: string +}): Promise<{ exitCode: number; stdout: string; stderr: string; terminalError?: BenchTerminalError.Kind }> { + // Use the same bun binary that's currently running — guaranteed to exist + // and avoids PATH lookup quirks under Bun's posix_spawn. + const bunPath = process.execPath + return new Promise((resolve) => { + // Don't set spawn's `cwd` — Bun's posix_spawn on some minimal apptainer + // images ENOENTs whenever cwd is set (libc lacks addchdir_np). Opencode's + // own `--dir ` flag changes the working directory + // internally, so we don't need spawn-level cwd. + const child = spawn( + bunPath, + [ + args.opencodeBin, + "run", + args.message, + "--agent", + args.agent, + "--model", + `nemo-gym/${args.modelName}`, + "--format", + "json", + "--dir", + args.workspaceRoot, + ], + { + env: args.env, + stdio: ["ignore", "pipe", "pipe"], + }, + ) + let stdout = "" + let stderr = "" + let terminalError: BenchTerminalError.Kind | undefined + let terminalSignalBuffer = "" + const observeTerminalSignal = (chunk: string) => { + // Retain enough overlap to recognize a marker split across pipe chunks. + terminalSignalBuffer = (terminalSignalBuffer + chunk).slice(-256) + terminalError = BenchTerminalError.prefer(terminalError, BenchTerminalError.detect(terminalSignalBuffer)) + } + const MAX_KEEP = 256 * 1024 // keep only a bounded tail for error reporting + let lineBuf = "" + child.stdout?.on("data", (b) => { + const chunk = b.toString("utf8") + observeTerminalSignal(chunk) + lineBuf += chunk + let idx: number + while ((idx = lineBuf.indexOf("\n")) >= 0) { + const line = lineBuf.slice(0, idx) + lineBuf = lineBuf.slice(idx + 1) + // Forward the event type so the gym log captures progress cheaply. + process.stdout.write(line + "\n") + stdout = (stdout + line + "\n").slice(-MAX_KEEP) + } + }) + child.stderr?.on("data", (b) => { + const chunk = b.toString("utf8") + observeTerminalSignal(chunk) + stderr = (stderr + chunk).slice(-MAX_KEEP) + process.stderr.write(chunk) + }) + child.on("close", (code) => resolve({ exitCode: code ?? 0, stdout, stderr, terminalError })) + child.on("error", (err) => { + stderr += String(err) + resolve({ exitCode: 999, stdout, stderr, terminalError }) + }) + }) +} + +interface OutputJsonl { + instance_id: string + test_result: { git_patch: string } + metadata: { llm_config: { model: string } } + metrics: Record + error: string | null +} + +async function writeOutputJsonl(evalOutputDir: string, instanceId: string, payload: OutputJsonl): Promise { + const runDir = path.join(evalOutputDir, instanceId, "bench_run") + await fs.mkdir(runDir, { recursive: true }) + const outPath = path.join(runDir, "output.jsonl") + const tmp = `${outPath}.tmp` + await fs.writeFile(tmp, JSON.stringify(payload) + "\n") + await fs.rename(tmp, outPath) + return outPath +} + +function completionsDirFor(evalOutputDir: string, instanceId: string): string { + // Match openhands' on-host glob: /*/*/*/llm_completions//*.json + return path.join(evalOutputDir, instanceId, "bench_run", "llm_completions", instanceId) +} + +function detectOpencodeBin(): string { + // Prefer the pre-bundled artifact at /.bench-build/opencode.js. + // Running un-bundled `src/index.ts` triggers cascading runtime resolution + // failures (TUI JSX runtime not honored, @anthropic-ai/sdk relative .mjs + // paths failing across the isolated install layout). The bundle inlines + // every transitive dep and is opencode's intended deployment shape. + // Falls back to src/index.ts only for dev / when setup_scripts/opencode.sh + // hasn't run. + const here = path.dirname(new URL(import.meta.url).pathname) + // bench/cli.ts → packages/opencode/src/bench → packages/opencode/src → packages/opencode → packages → + const opencodeRoot = path.resolve(here, "..", "..", "..", "..") + const bundled = path.resolve(opencodeRoot, ".bench-build", "opencode.js") + if (existsSync(bundled)) return bundled + return path.resolve(here, "..", "index.ts") +} + +async function main() { + const args = parseArgs(process.argv.slice(2)) + const instance = await readInstance(args.instanceDictPath, args.selectedId) + // workspaceRoot is decided gym-side based on dataset_name; we use it verbatim. + const workspaceRoot = args.workspaceRoot + const gymConfig = loadGymConfig(args.config) + const llmModelCfg = ((gymConfig as Record>).llm?.model ?? {}) as Record< + string, + unknown + > + const modelName = String(llmModelCfg.model ?? "unknown-model") + // Forced sampling params from gym (RL training on-policy requirement). + const forcedTemperature = typeof llmModelCfg.temperature === "number" ? llmModelCfg.temperature : undefined + const forcedTopP = typeof llmModelCfg.top_p === "number" ? llmModelCfg.top_p : undefined + // Optional: force a max_tokens cap; when absent, requests carry none (unlimited). + const forcedMaxTokens = typeof llmModelCfg.max_tokens === "number" ? llmModelCfg.max_tokens : undefined + const baseURL = process.env.NEMO_GYM_MODEL_SERVER_BASE_URL + if (!baseURL) throw new Error("NEMO_GYM_MODEL_SERVER_BASE_URL not set in env (gym harness sets this).") + + const completionsDir = completionsDirFor(args.outputDir, instance.instance_id) + await fs.mkdir(completionsDir, { recursive: true }) + + // Trajectory resume: when a replay file is given, the recorded conversation's + // own first user message becomes the task instruction (byte-for-byte, not + // gym's rendered template — mirrors OpenHands' replay semantics) and the + // recorded assistant turns are threaded into the nemo-gym provider so it + // replays them (re-executing tool calls for real) before continuing live. + let userPrompt: string + let replayTurns: NemoGymReplayTurn[] | undefined + let replayTrailingUserTexts: string[] | undefined + let replayManifest: NemoGymReplayManifest | undefined + if (args.replayMessagesFile) { + const raw = await fs.readFile(args.replayMessagesFile, "utf8") + const parsed = parseReplayMessages(raw) + userPrompt = parsed.initialUserText + replayTurns = parsed.replayTurns + replayTrailingUserTexts = parsed.trailingUserTexts + } else { + // The user message is fully rendered by gym (workspace_path baked in based + // on dataset_name); we just read it as-is and pass it to opencode. + userPrompt = await fs.readFile(args.userMessageFile, "utf8") + } + if (args.replaySubagentsFile) { + replayManifest = parseReplayManifest(await fs.readFile(args.replaySubagentsFile, "utf8")) + } + + const { tmpRoot, configFile } = await buildConfigDir({ + instanceId: instance.instance_id, + modelName, + baseURL, + completionsDir, + maxTurns: args.maxTurns, + systemPromptPath: args.systemPromptPath, + enableSubagents: args.enableSubagents || Boolean(replayManifest?.sessions.length), + temperature: forcedTemperature, + topP: forcedTopP, + maxTokens: forcedMaxTokens, + replayTurns, + replayTrailingUserTexts, + replayManifest, + }) + + const startedAt = Date.now() + const childEnv: NodeJS.ProcessEnv = { + ...process.env, + // Run-isolated opencode state. + OPENCODE_DB: ":memory:", + OPENCODE_DATA: path.join(tmpRoot, "data"), + OPENCODE_CONFIG: configFile, + // The benchmark already runs inside a SIF sandbox, so make that the + // security boundary. This final config override applies to subagents too. + OPENCODE_PERMISSION: JSON.stringify({ "*": "allow" }), + + // Disable opencode's built-in plugin loaders; the bench harness doesn't need them. + OPENCODE_PURE: "1", + // Skip the dynamic env block (working dir + Today's date) in the system + // prompt — keeps the RL prompt-token prefix invariant stable across turns + // (a midnight rollover would otherwise shift `Today's date: ...`). + OPENCODE_DISABLE_ENV_PROMPT: "1", + // Have all agent sessions report terminal states to this bench wrapper. + // This is bench-only and does not alter normal opencode runs. + [BenchTerminalError.ENV]: "1", + // Avoid serializing and piping full event payloads into the gym log. + OPENCODE_BENCH_EVENT_TYPES_ONLY: "1", + } + + // Bootstrap a git repo if the SIF shipped a flat source tree (swe-bench-ext + // and some SWE-rebench variants). Without this, the patch capture returns "" + // and every patch is recorded as 0 bytes. + const { freshInit } = await bootstrapRepoIfMissing(workspaceRoot) + + // Prune git history past base_commit so the agent can't reach future commits. + // Skip when we just freshly initialized: the dataset's upstream base_commit + // SHA doesn't exist in our local repo, so deep_reset would just fail + // rev-parse and fall through to its nuclear pass. The fresh `HEAD` is + // already the correct baseline (also tagged `opencode_bench_baseline`). + if (!freshInit) { + await runDeepReset(workspaceRoot, String(instance.base_commit ?? "")) + } + + // Snapshot the pristine HEAD *after* bootstrap/deep-reset — it is the diff + // base for `--patch-mode committed`. Recorded unconditionally so the log + // always shows what the agent started from. + const baselineCommit = await recordBaselineCommit(workspaceRoot) + console.log(`[bench] patch_mode=${args.patchMode} baseline=${baselineCommit || ""}`) + if (args.patchMode === "committed") { + // The agent is expected to commit; make sure git will let it. + await ensureCommitIdentity(workspaceRoot) + } + + const opencodeBin = detectOpencodeBin() + const result = await runOpencode({ + workspaceRoot, + modelName, + message: userPrompt, + env: childEnv, + opencodeBin, + agent: "swe-bench", + }) + + const patch = await capturePatch(workspaceRoot, args.patchMode, baselineCommit) + const benchRunTime = (Date.now() - startedAt) / 1000 + + const error = BenchTerminalError.toGymError(result.exitCode, result.terminalError) + const outPath = await writeOutputJsonl(args.outputDir, instance.instance_id, { + instance_id: instance.instance_id, + test_result: { git_patch: patch }, + metadata: { llm_config: { model: modelName } }, + metrics: { + bench_run_time: benchRunTime, + opencode_exit_code: result.exitCode, + patch_mode: args.patchMode, + }, + error, + }) + + console.log( + `[bench] wrote ${outPath} (patch=${patch.length} bytes, mode=${args.patchMode}, error=${error ?? "none"})`, + ) + + // Mirror opencode's exit code explicitly. Falling off the end of main() and + // letting Bun drain the event loop produced a flaky exit=1 even when the + // bench wrote output.jsonl cleanly (sqlite migration handles, residual + // child-stdio pipes from the opencode subprocess). Gym's runner treats any + // non-zero apptainer exit as `Agent command failed` and discards the + // already-written patch, so we MUST exit 0 deterministically on success. + process.exit(BenchTerminalError.shouldExitSuccessfully(result.exitCode, result.terminalError) ? 0 : 1) +} + +main().catch((err) => { + console.error(`[bench] fatal: ${err?.stack ?? err}`) + process.exit(2) +}) diff --git a/packages/opencode/src/bench/deep_reset.ts b/packages/opencode/src/bench/deep_reset.ts new file mode 100644 index 000000000000..f127a472111d --- /dev/null +++ b/packages/opencode/src/bench/deep_reset.ts @@ -0,0 +1,169 @@ +/** + * Strip git history past base_commit so the agent can't reach future commits. + * + * Port of nv-OpenHands' `_deep_reset_to_base_commit` + * (evaluation/benchmarks/swe_bench/run_infer.py:774). Two-pass design: + * + * - Careful pass: per-ref iteration with `git for-each-ref`. Preserves + * local branches that don't descend from base, resets branches that do, + * deletes tags/remote-tracking/stash/notes refs past base. + * - Nuclear fallback: batch-delete every tag/remote/stash/notes ref + every + * local branch in two `git update-ref --stdin` calls. Microseconds + * regardless of ref count — handles monorepos with thousands of refs + * where the careful pass times out. + * + * `|| true` at the very end so a busted git state can't kill the agent run. + */ + +import { spawn } from "node:child_process" +import { existsSync } from "node:fs" + +// Some SIFs are minimal and ship without `bash` on PATH, or Bun's posix_spawn +// doesn't fall back to PATH lookup the way `execvp` does — either way, +// spawn("bash", ...) ENOENTs. Probe absolute paths up front; the deep-reset +// script uses only POSIX features, so /bin/sh is a safe fallback if bash is +// absent. +function detectShell(): string | null { + for (const p of ["/bin/bash", "/usr/bin/bash", "/bin/sh", "/usr/bin/sh"]) { + if (existsSync(p)) return p + } + return null +} + +function carefulPass(baseCommit: string): string { + return ( + `echo "[deep_reset:careful] start" && ` + + `BASE=$(git rev-parse --verify ${baseCommit}^{commit}) && ` + + `ORIG_BRANCH=$(git symbolic-ref --short -q HEAD || echo main) && ` + + `echo "[deep_reset:careful] base=$BASE orig_branch=$ORIG_BRANCH" && ` + + `git checkout --detach "$BASE" && ` + + `echo "[deep_reset:careful] resetting local branches descending from base..." && ` + + `git for-each-ref --format="%(refname)" refs/heads | while read -r ref; do ` + + ` tip=$(git rev-parse -q --verify "$ref^{commit}" 2>/dev/null || true); ` + + ` [ -z "$tip" ] && continue; ` + + ` if [ "$tip" != "$BASE" ] && git merge-base --is-ancestor "$BASE" "$tip"; then ` + + ` echo "[deep_reset:careful] reset $ref -> $BASE"; ` + + ` git update-ref "$ref" "$BASE"; ` + + ` fi; ` + + `done && ` + + `echo "[deep_reset:careful] deleting tags/remotes/stash/notes past base..." && ` + + `git for-each-ref --format="%(refname)" refs | while read -r ref; do ` + + ` case "$ref" in refs/heads/*) continue ;; esac; ` + + ` if git symbolic-ref -q "$ref" >/dev/null 2>&1; then continue; fi; ` + + ` tip=$(git rev-parse -q --verify "$ref^{commit}" 2>/dev/null || true); ` + + ` [ -z "$tip" ] && continue; ` + + ` if [ "$tip" != "$BASE" ] && git merge-base --is-ancestor "$BASE" "$tip"; then ` + + ` echo "[deep_reset:careful] delete $ref"; ` + + ` git update-ref -d "$ref"; ` + + ` fi; ` + + `done && ` + + `echo "[deep_reset:careful] removing remotes + transient refs..." && ` + + `for r in $(git remote); do echo "[deep_reset:careful] rm remote $r"; git remote remove "$r"; done; ` + + `gd=$(git rev-parse --git-dir) && ` + + `rm -f "$gd"/FETCH_HEAD "$gd"/ORIG_HEAD "$gd"/MERGE_HEAD "$gd"/CHERRY_PICK_HEAD ` + + `"$gd"/REVERT_HEAD "$gd"/BISECT_HEAD "$gd"/AUTO_MERGE && ` + + `echo "[deep_reset:careful] expiring reflog + gc..." && ` + + `git reflog expire --expire=now --expire-unreachable=now --all && ` + + `git repack -ad && git prune --expire=now && git gc --prune=now && ` + + `git checkout -B "$ORIG_BRANCH" "$BASE" && ` + + `echo "[deep_reset:careful] done; HEAD=$ORIG_BRANCH at $BASE"` + ) +} + +function nuclearPass(baseCommit: string): string { + return ( + `echo "[deep_reset:nuclear] careful pass failed; running batch-delete fallback" && ` + + `BASE=$(git rev-parse --verify ${baseCommit}^{commit}) && ` + + `ORIG_BRANCH=$(git symbolic-ref --short -q HEAD || echo main) && ` + + `echo "[deep_reset:nuclear] base=$BASE orig_branch=$ORIG_BRANCH" && ` + + `git checkout --detach "$BASE" && ` + + `for r in $(git remote); do echo "[deep_reset:nuclear] rm remote $r"; git remote remove "$r"; done; ` + + `echo "[deep_reset:nuclear] batch-delete tags/remotes/stash/notes..." && ` + + `git for-each-ref --format="delete %(refname)" refs/tags refs/remotes refs/stash refs/notes 2>/dev/null ` + + `| git update-ref --stdin; ` + + `echo "[deep_reset:nuclear] batch-delete local branches..." && ` + + `git for-each-ref --format="delete %(refname)" refs/heads | git update-ref --stdin; ` + + `gd=$(git rev-parse --git-dir) && ` + + `rm -f "$gd"/FETCH_HEAD "$gd"/ORIG_HEAD "$gd"/MERGE_HEAD "$gd"/CHERRY_PICK_HEAD ` + + `"$gd"/REVERT_HEAD "$gd"/BISECT_HEAD "$gd"/AUTO_MERGE && ` + + `echo "[deep_reset:nuclear] expiring reflog + gc..." && ` + + `git reflog expire --expire=now --expire-unreachable=now --all && ` + + `git repack -ad && git prune --expire=now && git gc --prune=now && ` + + `git checkout -B "$ORIG_BRANCH" "$BASE" && ` + + `echo "[deep_reset:nuclear] done; HEAD=$ORIG_BRANCH at $BASE"` + ) +} + +export function buildDeepResetCmd(baseCommit: string): string { + return `( ${carefulPass(baseCommit)} ) || ( ${nuclearPass(baseCommit)} ) || true` +} + +function shellQuote(s: string): string { + return `'${s.replace(/'/g, `'\\''`)}'` +} + +// Default budget for the whole deep-reset pipeline (checkout, ref rewrite, +// reflog expire, repack, gc, prune). This is plain git plumbing that should +// finish in seconds even on large repos; the timeout exists so a stuck git +// process (lock contention, a pathological repo) can't silently burn the +// entire per-instance agent timeout before a single LLM call happens. +const DEFAULT_TIMEOUT_MS = 10 * 60_000 +const KILL_GRACE_MS = 10_000 + +export async function runDeepReset( + workspaceRoot: string, + baseCommit: string, + timeoutMs = Number(process.env.OPENCODE_DEEP_RESET_TIMEOUT_MS) || DEFAULT_TIMEOUT_MS, +): Promise { + if (!baseCommit) return + const shell = detectShell() + if (!shell) { + console.warn(`[bench] deep_reset skipped: no shell found at /bin/{bash,sh} or /usr/bin/{bash,sh}`) + return + } + // Bake `cd ` into the shell script instead of passing the `cwd` + // option to spawn(). On some minimal apptainer images Bun's posix_spawn + // ENOENTs whenever a `cwd` is set (libc lacks addchdir_np extension); routing + // the chdir through the shell sidesteps that entirely. + const cmd = `cd ${shellQuote(workspaceRoot)} && ` + buildDeepResetCmd(baseCommit) + console.log(`[bench] deep_reset workspace=${workspaceRoot} base=${baseCommit} shell=${shell} timeout_ms=${timeoutMs}`) + await new Promise((resolve) => { + // detached: true puts the shell in its own process group so a timeout + // can kill the whole tree (git repack/gc children included) via the + // negative-PID group signal — killing just the top-level bash leaves + // orphaned git children running, silently eating the rest of the budget. + const child = spawn(shell, ["-c", cmd], { + stdio: ["ignore", "inherit", "inherit"], + detached: true, + }) + const killTree = (signal: NodeJS.Signals) => { + try { + if (child.pid) process.kill(-child.pid, signal) + else child.kill(signal) + } catch { + // Group already gone (process exited between the timer firing and here). + } + } + let killTimer: ReturnType | undefined + const timeoutTimer = setTimeout(() => { + console.warn(`[bench] deep_reset exceeded ${timeoutMs}ms; sending SIGTERM`) + killTree("SIGTERM") + killTimer = setTimeout(() => { + console.warn(`[bench] deep_reset still alive ${KILL_GRACE_MS}ms after SIGTERM; sending SIGKILL`) + killTree("SIGKILL") + }, KILL_GRACE_MS) + }, timeoutMs) + child.on("close", (code) => { + clearTimeout(timeoutTimer) + clearTimeout(killTimer) + console.log(`[bench] deep_reset exit=${code ?? 0}`) + resolve() + }) + child.on("error", (err) => { + clearTimeout(timeoutTimer) + clearTimeout(killTimer) + console.warn(`[bench] deep_reset spawn error: ${err}`) + resolve() + }) + }) +} diff --git a/packages/opencode/src/bench/patch.ts b/packages/opencode/src/bench/patch.ts new file mode 100644 index 000000000000..135ba335028b --- /dev/null +++ b/packages/opencode/src/bench/patch.ts @@ -0,0 +1,218 @@ +/** + * Model-patch capture for the bench driver. + * + * Two modes, selected by `--patch-mode` (gym passes it through the + * `PATCH_MODE` env var in run_infer.sh): + * + * - `worktree` (DEFAULT, the historical behaviour): mark untracked files + * intent-to-add and take `git diff` of the working tree. Correct for the + * SWE-bench-style prompts that explicitly tell the agent *not* to commit. + * + * - `committed`: ignore the working tree and extract what the agent + * COMMITTED. Required by task families whose problem statement ends with + * "work on this in a new branch from main and commit everything when you + * are done" (e.g. the DeepSWE set). There, the agent commits its solution, + * leaving a clean tree — `git diff` returns "" and every rollout would be + * recorded as a 0-byte patch. + * + * In `committed` mode we diff `baseline..tip`, where `baseline` is the HEAD sha + * captured *before* the agent starts (after bootstrap/deep-reset, so it is the + * dataset's base commit) and `tip` is the most-advanced commit the agent left + * behind. `tip` is searched across HEAD *and* every local branch, because the + * agent may commit on a side branch and then switch back to main — HEAD alone + * would silently yield an empty patch. + * + * The resulting patch is still a plain `baseline -> final tree` unified diff, + * so the eval side (`git reset --hard ` + `git apply`) is + * unchanged regardless of mode. + */ + +import { spawn } from "node:child_process" +import { existsSync } from "node:fs" + +export type PatchMode = "worktree" | "committed" + +export const PATCH_MODES: PatchMode[] = ["worktree", "committed"] + +export const DEFAULT_PATCH_MODE: PatchMode = "worktree" + +export function parsePatchMode(raw: string | undefined): PatchMode { + if (!raw) return DEFAULT_PATCH_MODE + const v = raw.trim().toLowerCase() + if ((PATCH_MODES as string[]).includes(v)) return v as PatchMode + throw new Error(`Invalid --patch-mode "${raw}" (expected one of: ${PATCH_MODES.join(", ")})`) +} + +// git's canonical empty-tree object id. Used as the diff base when the repo has +// no commits at all (unborn HEAD), so a first commit still produces a patch. +const EMPTY_TREE = "4b825dc642cb6eb9a060e54bf8d69288fbee4904" + +// Same rationale as cli.ts/deep_reset.ts: some SIFs ENOENT on bare program +// names through Bun's posix_spawn, so resolve an absolute path up front. +function detectGit(): string { + for (const p of ["/usr/bin/git", "/bin/git", "/usr/local/bin/git"]) { + if (existsSync(p)) return p + } + return "git" +} + +interface GitResult { + stdout: string + exitCode: number +} + +function git(workspaceRoot: string, args: string[]): Promise { + const gitPath = detectGit() + return new Promise((resolve) => { + const child = spawn(gitPath, ["-C", workspaceRoot, ...args], { + env: { ...process.env, GIT_PAGER: "cat" }, + stdio: ["ignore", "pipe", "pipe"], + }) + let stdout = "" + child.stdout?.on("data", (b) => (stdout += b.toString("utf8"))) + // Swallow stderr: every call here is best-effort and a noisy `git` on a + // half-broken repo must not pollute the gym log or fail the rollout. + child.stderr?.on("data", () => {}) + child.on("close", (code) => resolve({ stdout, exitCode: code ?? 0 })) + child.on("error", () => resolve({ stdout: "", exitCode: 1 })) + }) +} + +/** + * Snapshot HEAD before the agent runs. Returns "" when the repo has no commit + * yet (unborn HEAD) or isn't a repo at all; `capturePatch` then falls back to + * the empty tree. + */ +export async function recordBaselineCommit(workspaceRoot: string): Promise { + const res = await git(workspaceRoot, ["rev-parse", "--verify", "--quiet", "HEAD^{commit}"]) + return res.exitCode === 0 ? res.stdout.trim() : "" +} + +/** + * Give git a committer identity if the image doesn't ship one. Without this the + * agent's very first `git commit` dies with "Author identity unknown" and the + * whole rollout scores zero for a reason that has nothing to do with the model. + * Repo-local (`--local`) so we don't mutate anything outside the workspace, and + * only when unset, so a task-provided identity always wins. + */ +export async function ensureCommitIdentity(workspaceRoot: string): Promise { + for (const [key, value] of [ + ["user.email", "agent@opencode.local"], + ["user.name", "opencode agent"], + ]) { + const existing = await git(workspaceRoot, ["config", "--get", key]) + if (existing.exitCode === 0 && existing.stdout.trim()) continue + await git(workspaceRoot, ["config", "--local", key, value]) + } +} + +async function worktreePatch(workspaceRoot: string): Promise { + // Mark untracked files as intent-to-add so newly-created files appear in + // `git diff` without being committed. Plain `git diff` only shows changes + // to tracked files, which silently drops new-file patches the agent wrote. + await git(workspaceRoot, ["add", "-AN"]) + const res = await git(workspaceRoot, ["diff", "--binary"]) + return res.stdout +} + +interface Candidate { + /** commit sha of the ref tip */ + sha: string + /** human label for logging: "HEAD" or the branch name */ + label: string + /** commits reachable from `sha` but not from the baseline */ + ahead: number +} + +/** + * Every commit the agent could have left its work on: HEAD (covers detached + * HEAD and "still on the branch it committed to") plus every local branch + * (covers "committed on a side branch, then checked main back out"). + */ +async function candidateTips(workspaceRoot: string, baseline: string): Promise { + const tips: { sha: string; label: string }[] = [] + + const head = await git(workspaceRoot, ["rev-parse", "--verify", "--quiet", "HEAD^{commit}"]) + if (head.exitCode === 0 && head.stdout.trim()) tips.push({ sha: head.stdout.trim(), label: "HEAD" }) + + const branches = await git(workspaceRoot, ["for-each-ref", "--format=%(objectname) %(refname:short)", "refs/heads"]) + for (const line of branches.stdout.split("\n")) { + const [sha, ...rest] = line.trim().split(" ") + if (!sha) continue + if (tips.some((t) => t.sha === sha)) continue // HEAD already covers this tip + tips.push({ sha, label: rest.join(" ") || sha.slice(0, 8) }) + } + + const out: Candidate[] = [] + for (const tip of tips) { + // `baseline..tip` counts commits reachable from tip but not from baseline. + // Deliberately NOT gated on `merge-base --is-ancestor`: an agent that + // amended or rebased its work leaves a tip that no longer descends from + // baseline, and we still want that work. + const res = await git(workspaceRoot, ["rev-list", "--count", `${baseline}..${tip.sha}`]) + const ahead = res.exitCode === 0 ? parseInt(res.stdout.trim(), 10) : 0 + out.push({ ...tip, ahead: Number.isFinite(ahead) ? ahead : 0 }) + } + return out +} + +/** Count of dirty/untracked paths, reported so a dropped worktree is visible in the log. */ +async function dirtyPathCount(workspaceRoot: string): Promise { + const res = await git(workspaceRoot, ["status", "--porcelain", "--untracked-files=all"]) + return res.stdout.split("\n").filter((l) => l.trim()).length +} + +async function committedPatch(workspaceRoot: string, baselineCommit: string): Promise { + const baseline = baselineCommit || EMPTY_TREE + if (!baselineCommit) { + console.log(`[bench] patch_mode=committed: no pre-run HEAD; diffing against the empty tree`) + } + + const candidates = await candidateTips(workspaceRoot, baseline) + // Most commits past the baseline wins. `candidateTips` puts HEAD first, and + // Array.prototype.sort is stable in every JS engine we ship on, so HEAD wins + // ties against a side branch holding the identical work. + const ranked = candidates.filter((c) => c.ahead > 0).sort((a, b) => b.ahead - a.ahead) + const dirty = await dirtyPathCount(workspaceRoot) + + if (ranked.length === 0) { + console.log( + `[bench] patch_mode=committed: agent left no commit past baseline ${baseline.slice(0, 8)} ` + + `(refs checked: ${candidates.length}, uncommitted paths: ${dirty}) -> empty patch`, + ) + return "" + } + + const chosen = ranked[0]! + const others = ranked + .slice(1) + .map((c) => `${c.label}+${c.ahead}`) + .join(",") + console.log( + `[bench] patch_mode=committed: baseline=${baseline.slice(0, 8)} tip=${chosen.label}@${chosen.sha.slice(0, 8)} ` + + `commits=${chosen.ahead} uncommitted_paths=${dirty}${others ? ` other_refs=[${others}]` : ""}`, + ) + if (dirty > 0) { + console.log( + `[bench] patch_mode=committed: ${dirty} uncommitted path(s) are NOT in the patch ` + + `(the task asked the agent to commit its work)`, + ) + } + + const res = await git(workspaceRoot, ["diff", "--binary", baseline, chosen.sha]) + return res.stdout +} + +/** + * Produce the model patch for `mode`. + * + * @param baselineCommit HEAD sha captured before the agent ran; only used by + * `committed` mode. + */ +export async function capturePatch( + workspaceRoot: string, + mode: PatchMode, + baselineCommit: string = "", +): Promise { + return mode === "committed" ? committedPatch(workspaceRoot, baselineCommit) : worktreePatch(workspaceRoot) +} diff --git a/packages/opencode/src/bench/replay.ts b/packages/opencode/src/bench/replay.ts new file mode 100644 index 000000000000..45aab7fa8e82 --- /dev/null +++ b/packages/opencode/src/bench/replay.ts @@ -0,0 +1,193 @@ +/** + * Parses a replay-messages file (prior chat-completion-format trajectory) for + * the SWE-bench bench harness. Split out of `bench/cli.ts` so it's importable + * without triggering `cli.ts`'s top-level `main()` (which calls + * `process.exit` on missing/invalid CLI args — not test-friendly). + * + * The first user message becomes the task instruction, assistant messages + * become scripted turns replayed in order by the nemo-gym provider (see + * language-model.ts). System / tool messages are skipped — system content is + * handled separately (pinned as the agent's system prompt by gym), and tool + * output is regenerated fresh by actually re-executing each replayed tool + * call against the sandbox, not replayed from recorded text. + * + * A *subsequent* user message (anything after the first) is real request + * content — everything the caller sent must be part of what the model sees. + * Unlike assistant/tool content it isn't reconstructed by replaying it + * through the agent loop (it's inert text, not an action), so each one is + * attached to the replay turn it immediately precedes as `precedingUserTexts` + * (or, if it trails the very last replayed turn with nothing recorded after + * it, returned separately as `trailingUserTexts`). `NemoGymLanguageModel` + * splices these into the outgoing message list on every live model call from + * the first one onward — they're not persisted in opencode's own session + * storage (there's no cheap way to do that without restructuring the bench + * harness to drive multiple `session.prompt()` calls against one long-lived, + * disk-backed session), so the provider re-applies them on every request + * instead of relying on session history to carry them forward. + */ + +import type { NemoGymReplayManifest, NemoGymReplayTurn } from "../provider/sdk/nemo-gym/language-model" + +export interface ReplayChatMessage { + role: "system" | "user" | "assistant" | "tool" + content?: string | Array<{ type?: string; text?: string }> | null + tool_calls?: Array<{ id: string; type?: string; function: { name: string; arguments: string } }> +} + +export function replayMessageText(content: ReplayChatMessage["content"]): string { + if (typeof content === "string") return content + if (Array.isArray(content)) { + return content + .filter((part) => part?.type === "text" || typeof part?.text === "string") + .map((part) => part.text ?? "") + .join("\n") + } + return "" +} + +export interface ParsedReplay { + initialUserText: string + replayTurns: NemoGymReplayTurn[] + /** Subsequent user messages after the last replayed assistant turn (trajectory ends on a user message). */ + trailingUserTexts?: string[] +} + +export function parseReplayMessages(raw: string): ParsedReplay { + const messages = JSON.parse(raw) as ReplayChatMessage[] + + let initialUserText: string | undefined + const replayTurns: NemoGymReplayTurn[] = [] + let pendingUserTexts: string[] = [] + + for (const msg of messages) { + if (msg.role === "system" || msg.role === "tool") continue + if (msg.role === "user") { + const text = replayMessageText(msg.content) + if (initialUserText === undefined) { + initialUserText = text + } else if (text) { + pendingUserTexts.push(text) + } + continue + } + if (msg.role === "assistant") { + replayTurns.push({ + content: typeof msg.content === "string" ? msg.content : replayMessageText(msg.content) || null, + toolCalls: msg.tool_calls?.map((tc) => ({ id: tc.id, name: tc.function.name, arguments: tc.function.arguments })), + ...(pendingUserTexts.length ? { precedingUserTexts: pendingUserTexts } : {}), + }) + pendingUserTexts = [] + } + } + + if (initialUserText === undefined) { + throw new Error("replay-messages-file: no user message found (expected at least one task-instruction message)") + } + + return { + initialUserText, + replayTurns, + ...(pendingUserTexts.length ? { trailingUserTexts: pendingUserTexts } : {}), + } +} + +function replayManifestError(message: string): never { + throw new Error(`replay-subagents-file: ${message}`) +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value) +} + +/** Parse Gym's snake_case causal subagent manifest into provider replay queues. */ +export function parseReplayManifest(raw: string): NemoGymReplayManifest { + const input: unknown = JSON.parse(raw) + if (!isRecord(input)) return replayManifestError("expected an object") + if (input.version !== 1) return replayManifestError("version must be 1") + if (typeof input.root_session_id !== "string" || !input.root_session_id) { + return replayManifestError("root_session_id must be a non-empty string") + } + if (!Array.isArray(input.sessions)) return replayManifestError("sessions must be an array") + + const seen = new Set() + const seenSpawns = new Set() + const sessions = input.sessions.map((value, index) => { + if (!isRecord(value)) return replayManifestError(`sessions[${index}] must be an object`) + const sessionId = value.session_id + const parentSessionId = value.parent_session_id + const spawnCallId = value.spawn_call_id + const spawnIndex = value.spawn_index + if (typeof sessionId !== "string" || !sessionId) { + return replayManifestError(`sessions[${index}].session_id must be a non-empty string`) + } + if (sessionId === input.root_session_id) { + return replayManifestError(`sessions[${index}].session_id duplicates root_session_id`) + } + if (seen.has(sessionId)) return replayManifestError(`duplicate session_id ${sessionId}`) + seen.add(sessionId) + if (typeof parentSessionId !== "string" || !parentSessionId) { + return replayManifestError(`sessions[${index}].parent_session_id must be a non-empty string`) + } + if (typeof spawnCallId !== "string" || !spawnCallId) { + return replayManifestError(`sessions[${index}].spawn_call_id must be a non-empty string`) + } + if (!Number.isInteger(spawnIndex) || (spawnIndex as number) < 0) { + return replayManifestError(`sessions[${index}].spawn_index must be a non-negative integer`) + } + const spawnKey = `${parentSessionId}\u0000${spawnCallId}` + if (seenSpawns.has(spawnKey)) { + return replayManifestError(`duplicate spawn_call_id ${spawnCallId} in parent ${parentSessionId}`) + } + seenSpawns.add(spawnKey) + if (!Array.isArray(value.messages)) { + return replayManifestError(`sessions[${index}].messages must be an array`) + } + + const parsed = parseReplayMessages(JSON.stringify(value.messages)) + return { + sessionId, + parentSessionId, + spawnCallId, + spawnIndex: spawnIndex as number, + ...(typeof value.subagent_type === "string" ? { subagentType: value.subagent_type } : {}), + messageCount: value.messages.length, + // Unlike the root session, every later child user message is recreated + // by replaying its parent's task(task_id=...) call. Injecting the text + // here as well would duplicate resumed-task prompts. + replayTurns: parsed.replayTurns.map(({ content, toolCalls }) => ({ + content, + ...(toolCalls ? { toolCalls } : {}), + })), + } + }) + + const knownParents = new Set([input.root_session_id, ...sessions.map((session) => session.sessionId)]) + for (const session of sessions) { + if (!knownParents.has(session.parentSessionId)) { + return replayManifestError( + `session ${session.sessionId} references unknown parent_session_id ${session.parentSessionId}`, + ) + } + } + + const reachable = new Set([input.root_session_id]) + let changed = true + while (changed) { + changed = false + for (const session of sessions) { + if (reachable.has(session.sessionId) || !reachable.has(session.parentSessionId)) continue + reachable.add(session.sessionId) + changed = true + } + } + const unreachable = sessions.find((session) => !reachable.has(session.sessionId)) + if (unreachable) { + return replayManifestError(`session ${unreachable.sessionId} is not reachable from root_session_id`) + } + + return { + version: 1, + rootSessionId: input.root_session_id, + sessions, + } +} diff --git a/packages/opencode/src/bench/terminal_error.ts b/packages/opencode/src/bench/terminal_error.ts new file mode 100644 index 000000000000..d8a0b64064ef --- /dev/null +++ b/packages/opencode/src/bench/terminal_error.ts @@ -0,0 +1,38 @@ +export type Kind = "max_iteration" | "context_window" + +export const ENV = "OPENCODE_BENCH_TERMINAL_SIGNALS" +export const PREFIX = "[opencode-bench-terminal] " + +export function encode(kind: Kind): string { + return PREFIX + kind +} + +/** Report terminal agent states from any session, including subagents. */ +export function report(kind: Kind): void { + if (process.env[ENV] !== "1") return + process.stderr.write(encode(kind) + "\n") +} + +export function detect(text: string): Kind | undefined { + if (text.includes(encode("context_window"))) return "context_window" + if (text.includes(encode("max_iteration"))) return "max_iteration" + return undefined +} + +/** Context overflow wins when the forced final max-step call also overflows. */ +export function prefer(current: Kind | undefined, incoming: Kind | undefined): Kind | undefined { + if (!current) return incoming + if (!incoming) return current + if (current === "context_window" || incoming === "context_window") return "context_window" + return "max_iteration" +} + +export function toGymError(exitCode: number, kind?: Kind): string | null { + if (kind === "max_iteration") return "maximum iteration reached" + if (kind === "context_window") return "context window exceeded" + return exitCode === 0 ? null : `opencode_exit_${exitCode}` +} + +export function shouldExitSuccessfully(exitCode: number, kind?: Kind): boolean { + return exitCode === 0 || kind !== undefined +} diff --git a/packages/opencode/src/cli/cmd/run.ts b/packages/opencode/src/cli/cmd/run.ts index a05b273e4489..4ab9f3be6868 100644 --- a/packages/opencode/src/cli/cmd/run.ts +++ b/packages/opencode/src/cli/cmd/run.ts @@ -411,6 +411,8 @@ export const RunCommand = effectCmd({ } async function execute(sdk: OpencodeClient) { + const benchEventTypesOnly = process.env.OPENCODE_BENCH_EVENT_TYPES_ONLY === "1" + function tool(part: ToolPart) { try { if (part.tool === ShellID.ToolID) return shell(props(part)) @@ -432,6 +434,10 @@ export const RunCommand = effectCmd({ function emit(type: string, data: Record) { if (args.format === "json") { + if (benchEventTypesOnly) { + process.stdout.write(type + EOL) + return true + } process.stdout.write(JSON.stringify({ type, timestamp: Date.now(), sessionID, ...data }) + EOL) return true } diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index 4c8e447041c0..e975dc863bd6 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -22,8 +22,13 @@ import { McpCommand } from "./cli/cmd/mcp" import { GithubCommand } from "./cli/cmd/github" import { ExportCommand } from "./cli/cmd/export" import { ImportCommand } from "./cli/cmd/import" -import { AttachCommand } from "./cli/cmd/tui/attach" -import { TuiThreadCommand } from "./cli/cmd/tui/thread" +// TUI subcommands (Attach, TuiThread) are dropped from this build of opencode. +// The bench harness (`packages/opencode/src/bench/cli.ts`) only invokes the +// `run` command via subprocess, never the TUI; loading them eagerly here drags +// in `cli/cmd/tui/app.tsx` at startup, which JSX-compiles against +// `@opentui/solid` and trips Bun's runtime JSX resolver into looking for +// `react/jsx-dev-runtime` (a bug we hit when running the un-bundled .ts). +// Removed entirely rather than lazy-loaded — bench has no use for them. import { AcpCommand } from "./cli/cmd/acp" import { EOL } from "os" import { WebCommand } from "./cli/cmd/web" @@ -115,49 +120,56 @@ const cli = yargs(args) run_id: processMetadata.runID, }) - const marker = path.join(Global.Path.data, "opencode.db") - if (!(await Filesystem.exists(marker))) { - const tty = process.stderr.isTTY - process.stderr.write("Performing one time database migration, may take a few minutes..." + EOL) - const width = 36 - const orange = "\x1b[38;5;214m" - const muted = "\x1b[0;2m" - const reset = "\x1b[0m" - let last = -1 - if (tty) process.stderr.write("\x1b[?25l") - try { - await JsonMigration.run(drizzle({ client: Database.Client().$client }), { - progress: (event) => { - const percent = Math.floor((event.current / event.total) * 100) - if (percent === last && event.current !== event.total) return - last = percent - if (tty) { - const fill = Math.round((percent / 100) * width) - const bar = `${"■".repeat(fill)}${"･".repeat(width - fill)}` - process.stderr.write( - `\r${orange}${bar} ${percent.toString().padStart(3)}%${reset} ${muted}${event.label.padEnd(12)} ${event.current}/${event.total}${reset}`, - ) - if (event.current === event.total) process.stderr.write("\n") - } else { - process.stderr.write(`sqlite-migration:${percent}${EOL}`) - } - }, - }) - } finally { - if (tty) process.stderr.write("\x1b[?25h") - else { - process.stderr.write(`sqlite-migration:done${EOL}`) + // In-memory DBs (bench/CI harnesses set OPENCODE_DB=:memory:, see + // bench/cli.ts) start empty every process and never persist a marker + // file, so the on-disk marker check below would always be absent and + // this "one time" migration would actually re-run — as a no-op, since + // there's no legacy JSON storage dir to migrate either, but still + // costing a Database.Client() open + filesystem glob — on every single + // invocation instead of truly once. + if (Database.Path !== ":memory:") { + const marker = path.join(Global.Path.data, "opencode.db") + if (!(await Filesystem.exists(marker))) { + const tty = process.stderr.isTTY + process.stderr.write("Performing one time database migration, may take a few minutes..." + EOL) + const width = 36 + const orange = "\x1b[38;5;214m" + const muted = "\x1b[0;2m" + const reset = "\x1b[0m" + let last = -1 + if (tty) process.stderr.write("\x1b[?25l") + try { + await JsonMigration.run(drizzle({ client: Database.Client().$client }), { + progress: (event) => { + const percent = Math.floor((event.current / event.total) * 100) + if (percent === last && event.current !== event.total) return + last = percent + if (tty) { + const fill = Math.round((percent / 100) * width) + const bar = `${"■".repeat(fill)}${"･".repeat(width - fill)}` + process.stderr.write( + `\r${orange}${bar} ${percent.toString().padStart(3)}%${reset} ${muted}${event.label.padEnd(12)} ${event.current}/${event.total}${reset}`, + ) + if (event.current === event.total) process.stderr.write("\n") + } else { + process.stderr.write(`sqlite-migration:${percent}${EOL}`) + } + }, + }) + } finally { + if (tty) process.stderr.write("\x1b[?25h") + else { + process.stderr.write(`sqlite-migration:done${EOL}`) + } } + process.stderr.write("Database migration complete." + EOL) } - process.stderr.write("Database migration complete." + EOL) } }) .usage("") .completion("completion", "generate shell completion script") .command(AcpCommand) .command(McpCommand) - .command(TuiThreadCommand) - .command(AttachCommand) .command(RunCommand) .command(GenerateCommand) .command(DebugCommand) diff --git a/packages/opencode/src/provider/models-snapshot.d.ts b/packages/opencode/src/provider/models-snapshot.d.ts new file mode 100644 index 000000000000..508ab6ee22fe --- /dev/null +++ b/packages/opencode/src/provider/models-snapshot.d.ts @@ -0,0 +1,3 @@ +// Empty stub committed for the bench harness build path. See models-snapshot.js +// for the rationale. +export declare const snapshot: Record diff --git a/packages/opencode/src/provider/models-snapshot.js b/packages/opencode/src/provider/models-snapshot.js new file mode 100644 index 000000000000..c48d54a8f72a --- /dev/null +++ b/packages/opencode/src/provider/models-snapshot.js @@ -0,0 +1,18 @@ +// @ts-nocheck +// Empty stub committed for the bench harness build path. +// +// Upstream opencode generates this file at build time via `script/generate.ts` +// (which fetches https://models.dev/api.json). For the nemo-gym bench harness +// we only register a single custom provider in the per-instance opencode +// config, so the snapshot is unused — but `bun build` still has to resolve +// `import("./models-snapshot.js")` from `provider/models.ts:137` at static +// analysis time. An empty snapshot satisfies that requirement; the runtime +// `try:` lambda in models.ts handles an empty snapshot gracefully. +// +// `.gitignore` excludes this file because upstream regenerates it. We +// force-add it on the bench branch (sdd/dev) so `bun build --target=bun +// packages/opencode/src/index.ts ...` succeeds without running generate.ts +// (which requires network access to models.dev). If you ever DO want real +// model metadata, run `bun run script/generate.ts` and don't commit the +// regenerated file. +export const snapshot = {} diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index 4013dcee36e7..dbcb319ff284 100644 --- a/packages/opencode/src/provider/provider.ts +++ b/packages/opencode/src/provider/provider.ts @@ -114,6 +114,11 @@ const BUNDLED_PROVIDERS: Record Promise<(opts: any) => BundledSDK> "gitlab-ai-provider": () => import("gitlab-ai-provider").then((m) => m.createGitLab), "@ai-sdk/github-copilot": () => import("./sdk/copilot/copilot-provider").then((m) => m.createOpenaiCompatible), "venice-ai-sdk-provider": () => import("venice-ai-sdk-provider").then((m) => m.createVenice), + // NeMo-Gym custom provider used by the SWE-bench RL rollout harness. + // Routes chat completions through the gym's vllm model server while + // threading prompt/generation token IDs into providerMetadata. The + // bench cli (`bench/cli.ts`) configures this provider per-instance. + "@opencode-ai/nemo-gym": () => import("./sdk/nemo-gym/index").then((m) => m.createNemoGym), } type CustomModelLoader = (sdk: any, modelID: string, options?: Record) => Promise diff --git a/packages/opencode/src/provider/sdk/nemo-gym/index.ts b/packages/opencode/src/provider/sdk/nemo-gym/index.ts new file mode 100644 index 000000000000..bd439b1182d5 --- /dev/null +++ b/packages/opencode/src/provider/sdk/nemo-gym/index.ts @@ -0,0 +1,94 @@ +/** + * NeMo-Gym opencode provider entry. + * + * Provider id: `nemo-gym`. Used by the bench harness for SWE-bench RL rollouts. + * Registered in `provider/provider.ts:BUNDLED_PROVIDERS`. + * + * The factory mirrors `@ai-sdk/openai-compatible`'s shape: `createNemoGym(opts)` + * returns a provider with `.languageModel(modelId)` so opencode's existing + * provider plumbing (Provider.Service.getModel) works without special-casing. + */ + +import { + NemoGymLanguageModel, + type NemoGymLanguageModelConfig, + type NemoGymReplayManifest, + type NemoGymReplayTurn, +} from "./language-model" + +export type { NemoGymReplayManifest, NemoGymReplaySession, NemoGymReplayTurn } from "./language-model" + +export interface CreateNemoGymOptions { + /** Base URL of the gym model server (`http://host:port`). */ + baseURL: string + /** Optional name of the model server (informational; useful for logs). */ + modelServerName?: string + /** Custom request headers. */ + headers?: () => Record + /** + * Where to dump per-call llm_completions/.json files. + * Set per-instance by the bench harness; if absent, no trajectory dump. + */ + completionsDir?: string + /** instance_id to embed in trajectory dump paths/file names. */ + instanceId?: string + /** Per-call HTTP timeout in ms. */ + requestTimeoutMs?: number + /** HTTP retry count on transient errors. */ + retries?: number + /** + * Forced sampling params for RL training (on-policy requirement): when set, + * every request carries exactly these values regardless of what the + * session/agent layer picked. Wired from the gym config's llm.model block. + */ + temperature?: number + topP?: number + /** Optional forced max_tokens; unset = no cap (vLLM generates to remaining context). */ + maxTokens?: number + /** Optional request-order counter shared across all live model calls. */ + turnCounter?: { next(): number } + /** Optional callback invoked after each successful chat-completion. */ + onCompletion?: NemoGymLanguageModelConfig["onCompletion"] + /** + * Scripted assistant turns to replay in the root session before falling + * through to live HTTP calls. Set by the bench harness when the request + * carries a prior trajectory to resume. See language-model.ts's docblock. + */ + replayTurns?: NemoGymReplayTurn[] + /** Subsequent user messages trailing the last replayed turn. See language-model.ts's docblock. */ + replayTrailingUserTexts?: string[] + /** Causal replay graph for child and nested-child sessions. */ + replayManifest?: NemoGymReplayManifest +} + +export interface NemoGymProvider { + languageModel: (modelId: string) => NemoGymLanguageModel +} + +export function createNemoGym(opts: CreateNemoGymOptions): NemoGymProvider { + if (!opts.baseURL) { + throw new Error("createNemoGym: baseURL is required (e.g. http://host:port)") + } + return { + languageModel(modelId: string) { + return new NemoGymLanguageModel(modelId, { + provider: "nemo-gym", + baseURL: opts.baseURL, + modelServerName: opts.modelServerName, + headers: opts.headers, + completionsDir: opts.completionsDir, + instanceId: opts.instanceId, + requestTimeoutMs: opts.requestTimeoutMs, + retries: opts.retries, + temperature: opts.temperature, + topP: opts.topP, + maxTokens: opts.maxTokens, + turnCounter: opts.turnCounter, + onCompletion: opts.onCompletion, + replayTurns: opts.replayTurns, + replayTrailingUserTexts: opts.replayTrailingUserTexts, + replayManifest: opts.replayManifest, + }) + }, + } +} diff --git a/packages/opencode/src/provider/sdk/nemo-gym/language-model.ts b/packages/opencode/src/provider/sdk/nemo-gym/language-model.ts new file mode 100644 index 000000000000..07203dcb21fc --- /dev/null +++ b/packages/opencode/src/provider/sdk/nemo-gym/language-model.ts @@ -0,0 +1,1076 @@ +/** + * NeMo-Gym LanguageModelV3 implementation. + * + * The opencode `processor.ts` agentic loop is unmodified — this is the only + * piece that swaps. Internally we POST to NeMo Gym's `/v1/chat/completions` + * non-streaming, capture token IDs (`prompt_token_ids` / `generation_token_ids` + * / `generation_log_probs`) from the response, and emit a single-shot synthetic + * stream so opencode's streaming handler is happy. + * + * Why non-streaming? RL training requires contiguous, exact token IDs across + * turns. Streaming has them drip in across SSE chunks; non-streaming returns + * them in the final response cleanly. The opencode loop doesn't notice — it + * receives all stream parts at once. + * + * Trajectory dump: every doStream call writes + * `//.json` BEFORE the stream finishes, + * so a tool crash later cannot lose this turn's token IDs. The shape matches + * openhands' `llm_completions//*.json` exactly so gym's + * `get_openhands_trajectory_from_completions` reads it without changes. + * + * Replay: `cfg.replayTurns` drives the root session and `cfg.replayManifest` + * supplies one queue per recorded child. A live child is bound by its + * recorded parent plus the exact task call ID that created it, so parallel + * siblings and nested agents cannot consume one another's turns. Tool-bearing + * calls are answered from that session's scripted queue instead of a real + * HTTP call — same synthesized-stream shape as a real response, so + * opencode's *real* tool-execution path (streamText -> the AI-SDK `tool.execute` closures + * built in session/prompt.ts's resolveTools()) runs each replayed tool call + * for real against the live sandbox. This is required for correctness: + * SWE-bench agents mutate a git workspace and the final patch comes from + * `git diff`, so a replayed "edit"/"bash" call must actually happen on the + * fresh container, not just be asserted via stale recorded output text. + * Once the queue is exhausted, doStream/doGenerate fall through unchanged to + * the real HTTP path and the agent continues live. Scripted turns are never + * dumped to completionsDir: gym derives the replay/live boundary from the + * FIRST dumped completion's cumulative `messages` length, which must be the + * first live call (by then the replay prefix is already in session + * history), not a scripted one. + * + * Subsequent user messages (anything after the trajectory's first user + * message) are real request content, not an action to replay — there's + * nothing to "re-execute" for a plain user turn. Everything the caller sent + * has to be part of what the model sees, starting with the first live call + * and on every call after that. Since opencode's own session storage isn't + * writable from here (see bench/replay.ts's module docblock), these aren't + * persisted anywhere — `_buildRequestParams` re-splices them into the + * outgoing message list, at a fixed position relative to the replayed + * assistant turns, on every live doStream/doGenerate call. + */ + +import { + type LanguageModelV3, + type LanguageModelV3CallOptions, + type LanguageModelV3StreamPart, + type LanguageModelV3Content, + type SharedV3ProviderMetadata, + type SharedV3Warning, +} from "@ai-sdk/provider" +import { promises as fs } from "node:fs" +import path from "node:path" +import { convertToOpenAICompatibleChatMessages } from "../copilot/chat/convert-to-openai-compatible-chat-messages" +import { prepareTools } from "../copilot/chat/openai-compatible-prepare-tools" + +// --------------------------------------------------------------------------- +// Wire types +// --------------------------------------------------------------------------- + +interface ChatRequestMessage { + role: "system" | "user" | "assistant" | "tool" + content?: string | Array | null + tool_calls?: Array<{ + id: string + type: "function" + function: { name: string; arguments: string } + }> + tool_call_id?: string + name?: string + prompt_token_ids?: number[] + generation_token_ids?: number[] + generation_log_probs?: number[] + [key: string]: unknown +} + +interface ChatResponseChoice { + index?: number + finish_reason?: string | null + message: { + role: string + content?: string | null + reasoning_text?: string | null + tool_calls?: Array<{ + id?: string + type?: string + function: { name: string; arguments: string } + }> + prompt_token_ids?: number[] + generation_token_ids?: number[] + generation_log_probs?: number[] + [key: string]: unknown + } +} + +interface ChatResponseUsage { + prompt_tokens?: number | null + completion_tokens?: number | null + total_tokens?: number | null +} + +interface ChatResponse { + id?: string + model?: string + created?: number + choices: ChatResponseChoice[] + usage?: ChatResponseUsage +} + +function contextOverflowStreamError(message: string): string { + return JSON.stringify({ + type: "error", + error: { + code: "context_length_exceeded", + message: message.slice(0, 2_000) || "Input exceeds context window of this model", + }, + }) +} + +function isGymContextOverflowCompletion(choice: ChatResponseChoice): boolean { + // Gym's vLLM wrapper translates an upstream context-overflow HTTP 400 into + // a successful empty completion. The stable signal it returns is exactly + // this pair. `content == null` alone is not enough because valid tool-call + // completions also normally carry null assistant content. + return choice.finish_reason === "length" && choice.message?.content == null +} + +// --------------------------------------------------------------------------- +// Config +// --------------------------------------------------------------------------- + +const TOKEN_ID_FIELDS = ["prompt_token_ids", "generation_token_ids", "generation_log_probs"] as const + +/** A single scripted assistant turn to replay before falling through to live model calls. */ +export interface NemoGymReplayTurn { + content: string | null + toolCalls?: Array<{ id: string; name: string; arguments: string }> + /** + * Subsequent user messages that occurred, in the original trajectory, + * immediately before this turn was originally generated. Not replayed as + * part of the scripted turn itself (a user message isn't an action) — + * spliced into every live request's message list once replay reaches this + * point. See the module docblock. + */ + precedingUserTexts?: string[] +} + +export interface NemoGymReplaySession { + sessionId: string + parentSessionId: string + spawnCallId: string + spawnIndex: number + subagentType?: string + messageCount: number + replayTurns: NemoGymReplayTurn[] + replayTrailingUserTexts?: string[] +} + +export interface NemoGymReplayManifest { + version: 1 + rootSessionId: string + sessions: NemoGymReplaySession[] +} + +export interface NemoGymLanguageModelConfig { + /** Provider id used to namespace providerMetadata. Defaults to "nemo-gym". */ + provider: string + /** Full base URL of the model server (e.g. `http://gym-host:18086`). */ + baseURL: string + /** Optional gym head-server-style model server name; informational only. */ + modelServerName?: string + /** Custom request headers (auth, etc). */ + headers?: () => Record + /** Per-call HTTP timeout in ms. */ + requestTimeoutMs?: number + /** Number of HTTP retry attempts on transient errors. */ + retries?: number + /** + * Forced sampling params for RL training. NeMo-RL's vLLM worker asserts + * that every request's temperature/top_p exactly match the training + * generation config (on-policy requirement) — when set, these override + * whatever the session/agent layer picked, for ALL sessions including + * subagents. + */ + temperature?: number + topP?: number + /** + * Optional forced max_tokens. When unset (the default), requests carry NO + * max_tokens and vLLM generates up to the remaining context — opencode's + * session-level output cap is deliberately ignored. + */ + maxTokens?: number + /** + * Where per-call llm_completions JSONs land. The bench harness builds this + * path; it must match what gym's host-side glob expects. If unset, no + * trajectory dump happens (useful for dev/test). + */ + completionsDir?: string + /** instance_id for the dump file naming + path. Required when completionsDir set. */ + instanceId?: string + /** Optional sink that the bench harness uses to count turns globally. */ + turnCounter?: { next(): number } + /** Optional callback fired after each successful chat completion. */ + onCompletion?: (info: { + turn: number + messages: ChatRequestMessage[] + response: ChatResponse + providerSpecificFields: Record + requestParams: Record + }) => void | Promise + /** + * Scripted assistant turns to replay in the root session before falling + * through to live HTTP calls. See the module docblock for why replayed + * tool calls must run for real rather than just replaying recorded text. + */ + replayTurns?: NemoGymReplayTurn[] + /** + * Subsequent user messages after the last replayed assistant turn, with + * nothing recorded after them in the trajectory (trajectory ends on a + * user message). Spliced into every live request's message list, same as + * NemoGymReplayTurn.precedingUserTexts. See the module docblock. + */ + replayTrailingUserTexts?: string[] + /** Causal parent-task-call -> recorded child-session replay graph. */ + replayManifest?: NemoGymReplayManifest +} + +// --------------------------------------------------------------------------- +// Implementation +// --------------------------------------------------------------------------- + +interface ReplayState { + recordedSessionID: string + recordedParentSessionID?: string + spawnCallID?: string + spawnIndex?: number + subagentType?: string + messageCount: number + turns: NemoGymReplayTurn[] + trailingUserTexts?: string[] + index: number + pendingUserInjections: Array<{ beforeAssistantOrdinal: number; text: string }> + livePrefixMessageCount?: number +} + +interface SessionHeaders { + sessionID: string + parentSessionID: string | undefined + parentToolCallID: string | undefined + agentName: string | undefined +} + +export class NemoGymLanguageModel implements LanguageModelV3 { + readonly specificationVersion = "v3" + readonly modelId: string + readonly provider: string + + private readonly cfg: NemoGymLanguageModelConfig + private cookies: Record = {} + // Per-session turn counter. opencode's session header is `x-session-affinity`; + // subagents spawned via the task tool get their own sessionID, so keeping + // a Map keeps their dump filenames from clobbering the main session's. + private readonly turnCounters: Map = new Map() + private readonly replayByRecordedSession = new Map() + private readonly replayChildBySpawn = new Map() + private readonly liveToRecordedSession = new Map() + private readonly recordedToLiveSession = new Map() + private readonly rootRecordedSessionID: string | undefined + private globalTurn = 0 + private readonly sessionStartGlobalTurn = new Map() + + constructor(modelId: string, cfg: NemoGymLanguageModelConfig) { + this.modelId = modelId + this.provider = cfg.provider + this.cfg = { + ...cfg, + requestTimeoutMs: cfg.requestTimeoutMs ?? 600_000, + retries: cfg.retries ?? 3, + } + this.rootRecordedSessionID = cfg.replayManifest?.rootSessionId ?? (cfg.replayTurns ? "__main__" : undefined) + if (this.rootRecordedSessionID) { + this.replayByRecordedSession.set( + this.rootRecordedSessionID, + this._makeReplayState({ + recordedSessionID: this.rootRecordedSessionID, + messageCount: 0, + turns: cfg.replayTurns ?? [], + trailingUserTexts: cfg.replayTrailingUserTexts, + }), + ) + } + for (const session of cfg.replayManifest?.sessions ?? []) { + this.replayByRecordedSession.set( + session.sessionId, + this._makeReplayState({ + recordedSessionID: session.sessionId, + recordedParentSessionID: session.parentSessionId, + spawnCallID: session.spawnCallId, + spawnIndex: session.spawnIndex, + subagentType: session.subagentType, + messageCount: session.messageCount, + turns: session.replayTurns, + trailingUserTexts: session.replayTrailingUserTexts, + }), + ) + this.replayChildBySpawn.set(this._spawnKey(session.parentSessionId, session.spawnCallId), session.sessionId) + } + } + + private _nextTurn(sessionID: string): number { + const n = (this.turnCounters.get(sessionID) ?? -1) + 1 + this.turnCounters.set(sessionID, n) + return n + } + + private _nextGlobalTurn(sessionID: string): { globalTurn: number; sessionStartGlobalTurn: number } { + const globalTurn = this.cfg.turnCounter?.next() ?? this.globalTurn++ + const sessionStartGlobalTurn = this.sessionStartGlobalTurn.get(sessionID) ?? globalTurn + this.sessionStartGlobalTurn.set(sessionID, sessionStartGlobalTurn) + return { globalTurn, sessionStartGlobalTurn } + } + + private _makeReplayState( + input: Omit, + ): ReplayState { + return { + ...input, + index: 0, + pendingUserInjections: [ + ...input.turns.flatMap((turn, i) => + (turn.precedingUserTexts ?? []).map((text) => ({ beforeAssistantOrdinal: i, text })), + ), + ...(input.trailingUserTexts ?? []).map((text) => ({ + beforeAssistantOrdinal: input.turns.length, + text, + })), + ], + } + } + + private _spawnKey(parentSessionID: string, callID: string): string { + return `${parentSessionID}\u0000${callID}` + } + + private _sessionFromHeaders(headers: unknown): SessionHeaders { + let sid = "" + let pid: string | undefined + let callID: string | undefined + let agentName: string | undefined + if (headers && typeof headers === "object") { + const h = headers as Record + const v = h["x-session-affinity"] ?? h["X-Session-Affinity"] + if (typeof v === "string") sid = v + const p = h["x-parent-session-id"] ?? h["X-Parent-Session-Id"] + if (typeof p === "string") pid = p + const c = h["x-parent-tool-call-id"] ?? h["X-Parent-Tool-Call-Id"] + if (typeof c === "string") callID = c + const a = h["x-opencode-agent"] ?? h["X-Opencode-Agent"] + if (typeof a === "string") agentName = a + } + return { sessionID: sid || "main", parentSessionID: pid, parentToolCallID: callID, agentName } + } + + private _bindReplaySession(liveSessionID: string, recordedSessionID: string): ReplayState | undefined { + const existing = this.liveToRecordedSession.get(liveSessionID) + if (existing && existing !== recordedSessionID) { + throw new Error( + `nemo-gym replay: live session ${liveSessionID} is already bound to ${existing}, cannot bind ${recordedSessionID}`, + ) + } + const otherLive = this.recordedToLiveSession.get(recordedSessionID) + if (otherLive && otherLive !== liveSessionID) { + throw new Error( + `nemo-gym replay: recorded session ${recordedSessionID} is already bound to ${otherLive}, cannot bind ${liveSessionID}`, + ) + } + this.liveToRecordedSession.set(liveSessionID, recordedSessionID) + this.recordedToLiveSession.set(recordedSessionID, liveSessionID) + return this.replayByRecordedSession.get(recordedSessionID) + } + + private _replayState(session: SessionHeaders): ReplayState | undefined { + const recorded = this.liveToRecordedSession.get(session.sessionID) + if (recorded) return this.replayByRecordedSession.get(recorded) + if (!session.parentSessionID) { + if (!this.rootRecordedSessionID) return undefined + return this._bindReplaySession(session.sessionID, this.rootRecordedSessionID) + } + if (!session.parentToolCallID) return undefined + const recordedParent = this.liveToRecordedSession.get(session.parentSessionID) + if (!recordedParent) return undefined + const recordedChild = this.replayChildBySpawn.get(this._spawnKey(recordedParent, session.parentToolCallID)) + if (!recordedChild) return undefined + return this._bindReplaySession(session.sessionID, recordedChild) + } + + // Auxiliary title/summary calls never pass tools, so they must not consume + // a session's scripted agentic-loop turns. + private _popReplayTurn(session: SessionHeaders, hasTools: boolean): NemoGymReplayTurn | undefined { + if (!hasTools) return undefined + const replay = this._replayState(session) + if (!replay || replay.index >= replay.turns.length) return undefined + return replay.turns[replay.index++] + } + + private _rewriteTaskResumeArguments(tool: NonNullable[number]): string { + if (tool.name !== "task" || !tool.arguments.includes('"task_id"')) return tool.arguments + let parsed: unknown + try { + parsed = JSON.parse(tool.arguments) + } catch { + return tool.arguments + } + if (!parsed || typeof parsed !== "object") return tool.arguments + const input = parsed as Record + if (typeof input.task_id !== "string") return tool.arguments + const liveSessionID = this.recordedToLiveSession.get(input.task_id) + if (!liveSessionID) return tool.arguments + return JSON.stringify({ ...input, task_id: liveSessionID }) + } + + private _messageFromReplayTurn(turn: NemoGymReplayTurn): ChatResponseChoice["message"] { + return { + role: "assistant", + content: turn.content, + tool_calls: turn.toolCalls?.map((tc) => ({ + id: tc.id, + type: "function" as const, + function: { name: tc.name, arguments: this._rewriteTaskResumeArguments(tc) }, + })), + } + } + + private _replayFinishReason(turn: NemoGymReplayTurn): string { + return turn.toolCalls && turn.toolCalls.length > 0 ? "tool_calls" : "stop" + } + + // Shared by the real HTTP path and the replay path so both stay in sync: + // reasoning -> text -> tool-call stream parts for one assistant turn. + private _enqueueMessageParts( + controller: ReadableStreamDefaultController, + msg: Pick, + providerMetadata: SharedV3ProviderMetadata, + ) { + // Reasoning content. providerMetadata on the *-end events is persisted by + // opencode's processor as part.metadata and replayed on the next request + // as part.providerOptions["nemo-gym"] — this is how per-turn token IDs + // round-trip so EVERY assistant turn (not just the last) carries them for + // RL training reconstruction. Replay turns carry no token IDs (empty + // providerMetadata), which is fine — gym never reads a dump for them. + if (msg.reasoning_text) { + controller.enqueue({ type: "reasoning-start", id: "reasoning-0" }) + controller.enqueue({ type: "reasoning-delta", id: "reasoning-0", delta: msg.reasoning_text }) + controller.enqueue({ type: "reasoning-end", id: "reasoning-0", providerMetadata }) + } + + // Text content. + if (msg.content) { + controller.enqueue({ type: "text-start", id: "txt-0" }) + controller.enqueue({ type: "text-delta", id: "txt-0", delta: msg.content }) + controller.enqueue({ type: "text-end", id: "txt-0", providerMetadata }) + } + + // Tool calls. Replayed tool-call IDs are reused verbatim (see + // _messageFromReplayTurn) so gym's replay/live boundary matching by + // call_id still lines up. + if (msg.tool_calls) { + for (const tc of msg.tool_calls) { + const tcId = tc.id ?? `call_${Math.random().toString(36).slice(2, 10)}` + controller.enqueue({ + type: "tool-input-start", + id: tcId, + toolName: tc.function.name, + }) + controller.enqueue({ + type: "tool-input-delta", + id: tcId, + delta: tc.function.arguments, + }) + controller.enqueue({ type: "tool-input-end", id: tcId }) + controller.enqueue({ + type: "tool-call", + toolCallId: tcId, + toolName: tc.function.name, + input: tc.function.arguments, + }) + } + } + } + + // Re-splices subsequent-user-message text from the replay trajectory into + // the outgoing message list, in place. Processed in descending ordinal + // order so inserting a later text doesn't shift the index about to be + // looked up for an earlier one. + private _injectPendingUserMessages(messages: ChatRequestMessage[], replay: ReplayState | undefined): void { + if (!replay?.pendingUserInjections.length) return + const byOrdinal = new Map() + for (const { beforeAssistantOrdinal, text } of replay.pendingUserInjections) { + const list = byOrdinal.get(beforeAssistantOrdinal) ?? [] + list.push(text) + byOrdinal.set(beforeAssistantOrdinal, list) + } + const ordinals = [...byOrdinal.keys()].sort((a, b) => b - a) + for (const ordinal of ordinals) { + const idx = this._findAssistantOrdinalIndex(messages, ordinal) + const texts = byOrdinal.get(ordinal)! + messages.splice(idx, 0, ...texts.map((text) => ({ role: "user" as const, content: text }))) + } + } + + // Index right before the `ordinal`-th (0-based) assistant-role message, or + // messages.length if fewer than `ordinal` assistant messages exist yet. + private _findAssistantOrdinalIndex(messages: ChatRequestMessage[], ordinal: number): number { + let seen = 0 + for (let i = 0; i < messages.length; i++) { + if (messages[i].role === "assistant") { + if (seen === ordinal) return i + seen++ + } + } + return messages.length + } + + get supportedUrls() { + return {} as Record + } + + // The streamText path in `session/llm.ts` only calls doStream. We still + // implement doGenerate for completeness / future direct-use. + async doGenerate(options: LanguageModelV3CallOptions) { + const session = this._sessionFromHeaders(options.headers) + const replayTurn = this._popReplayTurn(session, Boolean(options.tools && options.tools.length > 0)) + + if (replayTurn) { + const msg = this._messageFromReplayTurn(replayTurn) + const providerMetadata = this._buildProviderMetadata({}) + const content: LanguageModelV3Content[] = [] + if (msg.content) content.push({ type: "text", text: msg.content, providerMetadata }) + if (msg.tool_calls) { + for (const tc of msg.tool_calls) { + content.push({ + type: "tool-call", + toolCallId: tc.id ?? `call_${Math.random().toString(36).slice(2, 10)}`, + toolName: tc.function.name, + input: tc.function.arguments, + }) + } + } + return { + content, + finishReason: this._mapFinishReason(this._replayFinishReason(replayTurn)), + usage: this._mapUsage(undefined), + providerMetadata, + request: { body: "{}" }, + response: { body: {} }, + warnings: [], + } + } + + const { warnings, loggedMessages, requestParams, globalTurn, sessionStartGlobalTurn } = + await this._buildRequestParams(options, session) + const { responseJson } = await this._postChat(requestParams) + + const choice = responseJson.choices[0] + if (!choice) throw new Error("nemo-gym: empty choices in response") + if (isGymContextOverflowCompletion(choice)) { + throw new Error( + contextOverflowStreamError("NeMo Gym returned an empty length completion for an overlong context"), + ) + } + const msg: ChatResponseChoice["message"] = + choice.message ?? ({ role: "assistant" } as ChatResponseChoice["message"]) + + const providerSpecificFields = this._extractProviderFields(msg) + const providerMetadata = this._buildProviderMetadata(providerSpecificFields) + + const content: LanguageModelV3Content[] = [] + // Part-level providerMetadata round-trips per-turn token IDs (see doStream). + if (msg.content) content.push({ type: "text", text: msg.content, providerMetadata }) + if (msg.reasoning_text) content.push({ type: "reasoning", text: msg.reasoning_text, providerMetadata }) + if (msg.tool_calls) { + for (const tc of msg.tool_calls) { + content.push({ + type: "tool-call", + toolCallId: tc.id ?? `call_${Math.random().toString(36).slice(2, 10)}`, + toolName: tc.function.name, + input: tc.function.arguments, + }) + } + } + + await this._dumpAndNotify({ + messages: loggedMessages, + response: responseJson, + providerSpecificFields, + requestParams, + session, + globalTurn, + sessionStartGlobalTurn, + }) + + return { + content, + finishReason: this._mapFinishReason(choice.finish_reason ?? null), + usage: this._mapUsage(responseJson.usage), + providerMetadata, + request: { body: JSON.stringify(requestParams) }, + response: { body: responseJson }, + warnings, + } + } + + async doStream(options: LanguageModelV3CallOptions) { + const session = this._sessionFromHeaders(options.headers) + const replayTurn = this._popReplayTurn(session, Boolean(options.tools && options.tools.length > 0)) + + if (replayTurn) { + const msg = this._messageFromReplayTurn(replayTurn) + const finishReasonRaw = this._replayFinishReason(replayTurn) + const self = this + const stream = new ReadableStream({ + start(controller) { + controller.enqueue({ type: "stream-start", warnings: [] }) + const providerMetadata = self._buildProviderMetadata({}) + self._enqueueMessageParts(controller, msg, providerMetadata) + // No _dumpAndNotify here — see the module docblock: gym derives the + // replay/live boundary from the FIRST dumped completion, which must + // be the first live call. + controller.enqueue({ + type: "finish", + finishReason: self._mapFinishReason(finishReasonRaw), + usage: self._mapUsage(undefined), + providerMetadata, + }) + controller.close() + }, + }) + return { stream, request: { body: "{}" }, response: {} } + } + + const { warnings, loggedMessages, requestParams, globalTurn, sessionStartGlobalTurn } = + await this._buildRequestParams(options, session) + + // Fire the HTTP call eagerly so any error surfaces synchronously when the + // stream is consumed. We then synthesize parts in `start`. + const self = this + + const stream = new ReadableStream({ + async start(controller) { + controller.enqueue({ type: "stream-start", warnings }) + + try { + const { responseJson } = await self._postChat(requestParams) + + const choice = responseJson.choices[0] + if (!choice) throw new Error("nemo-gym: empty choices in response") + if (isGymContextOverflowCompletion(choice)) { + throw new Error( + contextOverflowStreamError("NeMo Gym returned an empty length completion for an overlong context"), + ) + } + const msg: ChatResponseChoice["message"] = + choice.message ?? ({ role: "assistant" } as ChatResponseChoice["message"]) + + const providerSpecificFields = self._extractProviderFields(msg) + const providerMetadata = self._buildProviderMetadata(providerSpecificFields) + + // Emit response-metadata first. + controller.enqueue({ + type: "response-metadata", + id: responseJson.id, + modelId: responseJson.model, + timestamp: responseJson.created ? new Date(responseJson.created * 1000) : undefined, + }) + + self._enqueueMessageParts(controller, msg, providerMetadata) + + // Persist trajectory BEFORE finishing so a downstream tool crash + // cannot lose this turn's token IDs. + await self._dumpAndNotify({ + messages: loggedMessages, + response: responseJson, + providerSpecificFields, + requestParams, + session, + globalTurn, + sessionStartGlobalTurn, + }) + + controller.enqueue({ + type: "finish", + finishReason: self._mapFinishReason(choice.finish_reason ?? null), + usage: self._mapUsage(responseJson.usage), + providerMetadata, + }) + + controller.close() + } catch (err) { + controller.enqueue({ type: "error", error: err instanceof Error ? err.message : String(err) }) + controller.enqueue({ + type: "finish", + finishReason: { unified: "error", raw: undefined }, + usage: self._mapUsage(undefined), + providerMetadata: {}, + }) + controller.close() + } + }, + }) + + return { + stream, + request: { body: JSON.stringify(requestParams) }, + response: {}, + } + } + + // ----------------------------------------------------------------------- + // Helpers + // ----------------------------------------------------------------------- + + private async _buildRequestParams( + options: LanguageModelV3CallOptions, + session: SessionHeaders, + ): Promise<{ + warnings: SharedV3Warning[] + messages: ChatRequestMessage[] + loggedMessages: ChatRequestMessage[] + tools: unknown + toolChoice: unknown + requestParams: Record + globalTurn: number + sessionStartGlobalTurn: number + }> { + const warnings: SharedV3Warning[] = [] + // Reuse opencode's existing OpenAI-compatible message converter so all + // tool-call / multi-content shapes map identically to the rest of opencode. + const messages = convertToOpenAICompatibleChatMessages(options.prompt) as unknown as ChatRequestMessage[] + + // The gym's vllm proxy (NeMoGymEasyInputMessage) only accepts plain-string + // content on non-assistant messages; the converter emits an ARRAY of parts + // for multi-part user turns (e.g. the synthetic "Attached image(s) from + // tool result:" message) which fails validation server-side with a 500. + // The policy model is text-only anyway, so flatten arrays to a single + // string and stub out non-text parts. + for (const m of messages as Array>) { + const content = m["content"] + if (Array.isArray(content)) { + m["content"] = (content as Array>) + .map((p) => + p?.["type"] === "text" ? String(p["text"] ?? "") : `[${String(p?.["type"] ?? "unknown")} part omitted]`, + ) + .join("\n") + } + } + + // Trajectory resume: splice in any subsequent user messages from the + // replay trajectory (see bench/replay.ts + the module docblock for why + // these can't just be persisted into opencode's own session storage). + // Done here, before loggedMessages is cloned from messages below, so the + // wire request and the trajectory dump gym reads both include them + // identically. Re-applied on every call — nothing else carries them + // forward — at a position fixed relative to the replayed assistant + // turns, so they land in the same chronological spot every time. + const replay = this._replayState(session) + this._injectPendingUserMessages(messages as ChatRequestMessage[], replay) + + // Token-ID handling, mirroring OpenHands' nemo_gym_client.py exactly: + // - WIRE request: token IDs on the MOST RECENT assistant message only + // (the last turn's prompt_token_ids embed the exact token stream of + // the whole conversation; the vllm proxy uses it to verify continuity + // and avoid retokenization drift). + // - LOGGED trajectory (llm_completions/*.json): token IDs on EVERY + // assistant turn — swe_agents app.py materializes the training + // episode from the logged messages, and NeMo-RL needs per-turn + // generation_token_ids/log_probs to build the loss mask over all + // model-generated spans, not just the final turn. + // The IDs round-trip through opencode's part metadata: doStream emits + // providerMetadata on text-end / reasoning-end -> processor stores + // part.metadata -> replay attaches part.providerOptions["nemo-gym"]. + // (The generic converter drops the fields, so we restore them here.) + // Shallow-clone each message so wire-side attach/strip never leaks into + // the logged copy and vice versa. + const loggedMessages = (messages as Array>).map((m) => ({ + ...m, + })) as unknown as ChatRequestMessage[] + if (replay && options.tools?.length && replay.livePrefixMessageCount === undefined) { + replay.livePrefixMessageCount = loggedMessages.length + } + { + const promptAssistants = options.prompt.filter((m) => m.role === "assistant") + const wireAssistants = (messages as Array>).filter((m) => m["role"] === "assistant") + const loggedAssistants = (loggedMessages as unknown as Array>).filter( + (m) => m["role"] === "assistant", + ) + const n = Math.min(promptAssistants.length, wireAssistants.length) + let mostRecentAttached = false + for (let i = n - 1; i >= 0; i--) { + const parts = promptAssistants[i].content + if (!Array.isArray(parts)) continue + for (const part of parts as Array<{ providerOptions?: Record> }>) { + const md = part.providerOptions?.[this.provider] + if (md && TOKEN_ID_FIELDS.every((f) => f in md)) { + for (const f of TOKEN_ID_FIELDS) loggedAssistants[i][f] = md[f] + if (!mostRecentAttached) { + for (const f of TOKEN_ID_FIELDS) wireAssistants[i][f] = md[f] + mostRecentAttached = true + } + break + } + } + } + } + + const { tools, toolChoice, toolWarnings } = prepareTools({ + tools: options.tools, + toolChoice: options.toolChoice, + }) + warnings.push(...toolWarnings) + + // Safeguard: strip token-ID fields from all assistant messages EXCEPT the + // most recent (mirrors nemo_gym_client.py:85-97). With the single-message + // attach above this is normally a no-op, but it keeps the wire contract + // if an upstream change ever attaches more. + { + let lastSeen = false + for (let i = messages.length - 1; i >= 0; i--) { + const m = messages[i] as Record + const hasAll = TOKEN_ID_FIELDS.every((f) => f in m) + if (lastSeen) { + for (const f of TOKEN_ID_FIELDS) delete m[f] + } else if (hasAll) { + lastSeen = true + } + } + } + + const requestParams: Record = { + messages, + // max_tokens is intentionally OMITTED unless the gym config forces one: + // without it vLLM generates up to the remaining context + // (max_model_len - prompt), i.e. "unlimited" output. Sending opencode's + // session-level cap (OUTPUT_TOKEN_MAX=32k) both truncated long turns and + // shrank the usable input window (vLLM rejects input+max_tokens>context). + ...(this.cfg.maxTokens ? { max_tokens: this.cfg.maxTokens } : {}), + // Forced training params (cfg) win over session/agent-level choices: + // NeMo-RL asserts exact temperature/top_p equality on every request. + temperature: this.cfg.temperature ?? options.temperature, + top_p: this.cfg.topP ?? options.topP, + stop: options.stopSequences, + seed: options.seed, + } + // Only include `model` when the caller-supplied modelId is a real value. + // opencode's session resolver falls back to its sentinel `"default"` + // (and to empty string with some misconfigured agents) when no model is + // pinned. We DO NOT want either of those leaking through to OpenAI as a + // literal `model: "default"` — the gym openai_model server's + // `body_dict.setdefault("model", self.config.openai_model)` will fill in + // the policy-configured model name when we omit it instead. + if (this.modelId && this.modelId !== "default") { + requestParams.model = this.modelId + } + if (tools && (tools as unknown[]).length) requestParams.tools = tools + if (toolChoice) requestParams.tool_choice = toolChoice + + // Strip undefineds — vllm errors on null/undefined keys. + for (const k of Object.keys(requestParams)) { + if (requestParams[k] === undefined) delete requestParams[k] + } + + return { + warnings, + messages, + loggedMessages, + tools, + toolChoice, + requestParams, + ...this._nextGlobalTurn(session.sessionID), + } + } + + private async _postChat(params: Record): Promise<{ responseJson: ChatResponse }> { + const url = this._urlFor("/v1/chat/completions") + const headers: Record = { + "Content-Type": "application/json", + Accept: "application/json", + } + // opencode's bundled-provider loader can pass `headers` as either a + // function (matching upstream openai-compatible's schema) OR a plain + // object (when opencode injects defaults from its provider merge layer). + // Handle both — `?.()` would throw on a non-callable object. + let cfgHeaders: Record | undefined + const rawHeaders = this.cfg.headers as unknown + if (typeof rawHeaders === "function") { + cfgHeaders = (rawHeaders as () => Record)() + } else if (rawHeaders && typeof rawHeaders === "object") { + cfgHeaders = rawHeaders as Record + } + if (cfgHeaders) { + for (const [k, v] of Object.entries(cfgHeaders)) if (v != null) headers[k] = v + } + if (Object.keys(this.cookies).length) { + headers.Cookie = Object.entries(this.cookies) + .map(([k, v]) => `${k}=${v}`) + .join("; ") + } + + const retries = this.cfg.retries ?? 3 + const timeoutMs = this.cfg.requestTimeoutMs ?? 0 + let lastErr: unknown = null + for (let attempt = 0; attempt < retries; attempt++) { + const ac = new AbortController() + // timeoutMs<=0 means "no timeout" — don't install the abort timer. + const timer = timeoutMs > 0 ? setTimeout(() => ac.abort(), timeoutMs) : null + try { + const res = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify(params), + signal: ac.signal, + }) + if (timer) clearTimeout(timer) + if (!res.ok) { + const text = await res.text().catch(() => "") + throw new Error(`NeMoGym ${url} ${res.status}: ${text.slice(0, 500)}`) + } + const setCookie = res.headers.get("set-cookie") + if (setCookie) { + for (const part of setCookie.split(/,(?=[^;]+=)/)) { + const [kv] = part.split(";") + const [k, v] = kv.split("=") + if (k && v) this.cookies[k.trim()] = v.trim() + } + } + const responseJson = (await res.json()) as ChatResponse + return { responseJson } + } catch (err) { + if (timer) clearTimeout(timer) + lastErr = err + if (attempt === retries - 1) break + // Cap exponential backoff at 60s so unlimited-retries configs don't blow up the delay. + const backoffMs = Math.min(1000 * 2 ** attempt, 60_000) + await new Promise((r) => setTimeout(r, backoffMs)) + } + } + throw new Error(`NeMoGym chat completions failed after ${retries} attempts: ${String(lastErr)}`) + } + + private _urlFor(p: string): string { + const base = this.cfg.baseURL.endsWith("/") ? this.cfg.baseURL : `${this.cfg.baseURL}/` + return new URL(p.replace(/^\//, ""), base).toString() + } + + private _extractProviderFields(msg: ChatResponseChoice["message"]): Record { + const out: Record = {} + if (Array.isArray(msg.prompt_token_ids)) { + for (const f of TOKEN_ID_FIELDS) { + const v = (msg as Record)[f] + if (v !== undefined) out[f] = v + } + } + return out + } + + private _buildProviderMetadata(providerSpecific: Record): SharedV3ProviderMetadata { + const md: SharedV3ProviderMetadata = { [this.provider]: {} } + for (const [k, v] of Object.entries(providerSpecific)) { + ;(md[this.provider] as Record)[k] = v as never + } + return md + } + + private _mapFinishReason(raw: string | null): { + unified: "stop" | "length" | "tool-calls" | "error" | "other" + raw: string | undefined + } { + if (!raw) return { unified: "other", raw: undefined } + switch (raw) { + case "stop": + return { unified: "stop", raw } + case "length": + return { unified: "length", raw } + case "tool_calls": + case "function_call": + return { unified: "tool-calls", raw } + default: + return { unified: "other", raw } + } + } + + private _mapUsage(raw?: ChatResponseUsage) { + return { + inputTokens: { + total: raw?.prompt_tokens ?? undefined, + noCache: raw?.prompt_tokens ?? undefined, + cacheRead: undefined, + cacheWrite: undefined, + }, + outputTokens: { + total: raw?.completion_tokens ?? undefined, + text: raw?.completion_tokens ?? undefined, + reasoning: undefined, + }, + } + } + + private async _dumpAndNotify(args: { + messages: ChatRequestMessage[] + response: ChatResponse + providerSpecificFields: Record + requestParams: Record + session: SessionHeaders + globalTurn: number + sessionStartGlobalTurn: number + }) { + const turn = this._nextTurn(args.session.sessionID) + const replay = this._replayState(args.session) + const recordedParentSessionID = + replay?.recordedParentSessionID ?? + (args.session.parentSessionID ? this.liveToRecordedSession.get(args.session.parentSessionID) : undefined) + if (this.cfg.onCompletion) { + try { + await this.cfg.onCompletion({ turn, ...args }) + } catch (err) { + console.warn(`[nemo-gym] onCompletion hook threw: ${String(err)}`) + } + } + + if (!this.cfg.completionsDir || !this.cfg.instanceId) return + + try { + await fs.mkdir(this.cfg.completionsDir, { recursive: true }) + const turnStr = String(turn).padStart(4, "0") + const safeModel = this.modelId.replace(/\//g, "__") + // sessionID is part of the filename so subagent dumps don't clobber the + // main session's. Sanitized for filesystem safety. + const safeSession = args.session.sessionID.replace(/[^A-Za-z0-9_-]/g, "_") + const fname = `${safeModel}-${safeSession}-${turnStr}-${Date.now()}.json` + const fpath = path.join(this.cfg.completionsDir, fname) + const kwargs: Record = {} + for (const [k, v] of Object.entries(args.requestParams)) { + if (k !== "messages") kwargs[k] = v + } + const payload = { + messages: args.messages, + response: args.response, + provider_specific_fields: args.providerSpecificFields, + kwargs, + session_id: args.session.sessionID, + parent_session_id: args.session.parentSessionID ?? null, + recorded_session_id: replay?.recordedSessionID ?? null, + recorded_parent_session_id: recordedParentSessionID ?? null, + spawn_call_id: replay?.spawnCallID ?? args.session.parentToolCallID ?? null, + spawn_index: replay?.spawnIndex ?? null, + subagent_type: replay?.subagentType ?? args.session.agentName ?? null, + replay_prefix_message_count: replay?.livePrefixMessageCount ?? null, + turn, + global_turn: args.globalTurn, + session_start_global_turn: args.sessionStartGlobalTurn, + timestamp: Date.now() / 1000, + } + const tmp = `${fpath}.tmp` + await fs.writeFile(tmp, JSON.stringify(payload)) + await fs.rename(tmp, fpath) + } catch (err) { + console.warn(`[nemo-gym] failed to dump completion: ${String(err)}`) + } + } +} diff --git a/packages/opencode/src/session/llm.ts b/packages/opencode/src/session/llm.ts index e76583f2d347..dce7d2e0aa26 100644 --- a/packages/opencode/src/session/llm.ts +++ b/packages/opencode/src/session/llm.ts @@ -37,6 +37,7 @@ export type StreamInput = { user: MessageV2.User sessionID: string parentSessionID?: string + parentToolCallID?: string model: Provider.Model agent: Agent.Info permission?: Permission.Ruleset @@ -381,6 +382,8 @@ const live: Layer.Layer< : { "x-session-affinity": input.sessionID, ...(input.parentSessionID ? { "x-parent-session-id": input.parentSessionID } : {}), + ...(input.parentToolCallID ? { "x-parent-tool-call-id": input.parentToolCallID } : {}), + "x-opencode-agent": input.agent.name, "User-Agent": `opencode/${InstallationVersion}`, }), ...input.model.headers, diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index 237fb527c078..347e9b202b86 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -378,6 +378,8 @@ const messageBase = { export const User = Schema.Struct({ ...messageBase, role: Schema.Literal("user"), + /** Task tool call in the parent session that created this child session. */ + parentToolCallID: Schema.optional(Schema.String), time: Schema.Struct({ created: NonNegativeInt, }), diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index f22da92927d2..727eaed99526 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -24,6 +24,7 @@ import { EventV2 } from "@/v2/event" import { SessionEvent } from "@/v2/session-event" import { Modelv2 } from "@/v2/model" import * as DateTime from "effect/DateTime" +import * as BenchTerminalError from "@/bench/terminal_error" const DOOM_LOOP_THRESHOLD = 3 const log = Log.create({ service: "session.processor" }) @@ -647,6 +648,7 @@ export const layer: Layer.Layer< slog.error("process", { error: errorMessage(e), stack: e instanceof Error ? e.stack : undefined }) const error = parse(e) if (MessageV2.ContextOverflowError.isInstance(error)) { + BenchTerminalError.report("context_window") ctx.needsCompaction = true yield* bus.publish(Session.Event.Error, { sessionID: ctx.sessionID, error }) return diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index fef8c438366c..4e250c788790 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -61,6 +61,7 @@ import * as DateTime from "effect/DateTime" import { eq } from "@/storage/db" import * as Database from "@/storage/db" import { SessionTable } from "./session.sql" +import * as BenchTerminalError from "@/bench/terminal_error" // @ts-ignore globalThis.AI_SDK_LOG_WARNINGS = false @@ -943,6 +944,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the id: input.messageID ?? MessageID.ascending(), role: "user", sessionID: input.sessionID, + parentToolCallID: input.parentToolCallID, time: { created: Date.now() }, tools: input.tools, agent: ag.name, @@ -1495,6 +1497,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the } const maxSteps = agent.steps ?? Infinity const isLastStep = step >= maxSteps + if (isLastStep) BenchTerminalError.report("max_iteration") msgs = yield* insertReminders({ messages: msgs, agent, session }) const msg: MessageV2.Assistant = { @@ -1569,7 +1572,11 @@ NOTE: At any point in time through this workflow you should feel free to ask the sys.skills(agent), sys.environment(model), instruction.system().pipe(Effect.orDie), - MessageV2.toModelMessagesEffect(msgs, model), + // stripMedia: the policy model is text-only; dropping media + // attachments here prevents the synthetic "Attached image(s) + // from tool result:" user message (multi-part content the gym + // vllm proxy rejects) from ever being injected. + MessageV2.toModelMessagesEffect(msgs, model, { stripMedia: true }), ]) const system = [...env, ...instructions, ...(skills ? [skills] : [])] const format = lastUser.format ?? { type: "text" as const } @@ -1580,8 +1587,9 @@ NOTE: At any point in time through this workflow you should feel free to ask the permission: session.permission, sessionID, parentSessionID: session.parentID, + parentToolCallID: lastUser.parentToolCallID, system, - messages: [...modelMsgs, ...(isLastStep ? [{ role: "assistant" as const, content: MAX_STEPS }] : [])], + messages: [...modelMsgs, ...(isLastStep ? [{ role: "user" as const, content: MAX_STEPS }] : [])], tools, model, toolChoice: format.type === "json_schema" ? "required" : undefined, @@ -1807,6 +1815,7 @@ const ModelRef = Schema.Struct({ export const PromptInput = Schema.Struct({ sessionID: SessionID, messageID: Schema.optional(MessageID), + parentToolCallID: Schema.optional(Schema.String), model: Schema.optional(ModelRef), agent: Schema.optional(Schema.String), noReply: Schema.optional(Schema.Boolean), diff --git a/packages/opencode/src/session/system.ts b/packages/opencode/src/session/system.ts index 06c71fa7dbdd..acde90b448d7 100644 --- a/packages/opencode/src/session/system.ts +++ b/packages/opencode/src/session/system.ts @@ -46,6 +46,11 @@ export const layer = Layer.effect( return Service.of({ environment: Effect.fn("SystemPrompt.environment")(function* (model: Provider.Model) { + // Bench / RL mode: skip the dynamic env block entirely. It includes + // `new Date().toDateString()` which would shift prompt tokens across a + // midnight rollover and break the RL contiguity invariant + // (prompt_token_ids[N+1] must extend prompt_token_ids[N]). + if (process.env.OPENCODE_DISABLE_ENV_PROMPT === "1") return [] const ctx = yield* InstanceState.context return [ [ diff --git a/packages/opencode/src/tool/read.ts b/packages/opencode/src/tool/read.ts index bf01fc7d2d5c..3d0d83b04fca 100644 --- a/packages/opencode/src/tool/read.ts +++ b/packages/opencode/src/tool/read.ts @@ -219,25 +219,17 @@ export const ReadTool = Tool.define( const mime = sniffAttachmentMime(sample, AppFileSystem.mimeType(filepath)) const isImage = SUPPORTED_IMAGE_MIMES.has(mime) + // Media input is disabled in this harness: the policy model is + // text-only, so image/PDF attachments would only produce a synthetic + // multi-part user message the gym's vllm proxy can't accept (and a + // base64 blob in session storage). Fail with a clear error so the + // agent moves on instead of retrying. if (isImage || isPdfAttachment(mime)) { - const bytes = yield* fs.readFile(filepath) - const msg = isPdfAttachment(mime) ? "PDF read successfully" : "Image read successfully" - return { - title, - output: msg, - metadata: { - preview: msg, - truncated: false, - loaded: loaded.map((item) => item.filepath), - }, - attachments: [ - { - type: "file" as const, - mime, - url: `data:${mime};base64,${Buffer.from(bytes).toString("base64")}`, - }, - ], - } + return yield* Effect.fail( + new Error( + `Cannot read ${isPdfAttachment(mime) ? "PDF" : "image"} file: ${filepath} (media input is not supported by this model)`, + ), + ) } if (isBinaryFile(filepath, sample)) { diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts index 22e4e5671c89..8e2ade8177a9 100644 --- a/packages/opencode/src/tool/task.ts +++ b/packages/opencode/src/tool/task.ts @@ -100,6 +100,13 @@ export const TaskTool = Tool.define( })) ?? []), ], })) + const spawnMessage = session + ? (yield* sessions.messages({ sessionID: nextSession.id })).find( + (message) => message.info.role === "user" && message.info.parentToolCallID, + ) + : undefined + const spawnToolCallID = + spawnMessage?.info.role === "user" ? (spawnMessage.info.parentToolCallID ?? ctx.callID) : ctx.callID const msg = yield* Effect.sync(() => MessageV2.get({ sessionID: ctx.sessionID, messageID: ctx.messageID })) if (msg.info.role !== "assistant") return yield* Effect.fail(new Error("Not an assistant message")) @@ -138,6 +145,7 @@ export const TaskTool = Tool.define( const result = yield* ops.prompt({ messageID, sessionID: nextSession.id, + parentToolCallID: spawnToolCallID, model: { modelID: model.modelID, providerID: model.providerID, diff --git a/packages/opencode/test/bench/patch.test.ts b/packages/opencode/test/bench/patch.test.ts new file mode 100644 index 000000000000..5aae8f5b1f2e --- /dev/null +++ b/packages/opencode/test/bench/patch.test.ts @@ -0,0 +1,154 @@ +import { $ } from "bun" +import { afterEach, describe, expect, test } from "bun:test" +import fs from "fs/promises" +import os from "os" +import path from "path" +import { capturePatch, ensureCommitIdentity, parsePatchMode, recordBaselineCommit } from "../../src/bench/patch" + +// Deliberately NOT using ../fixture/fixture: src/bench/* is a standalone leaf +// (node:child_process + node:fs only) that runs inside minimal SIF images, and +// this suite should stay runnable without booting the instance/effect stack. +const created: string[] = [] + +afterEach(async () => { + await Promise.all(created.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true }).catch(() => {}))) +}) + +async function repo(withBaseline = true) { + const dir = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "bench-patch-"))) + created.push(dir) + await $`git init -q -b main`.cwd(dir).quiet() + await $`git config user.email bench@opencode.local`.cwd(dir).quiet() + await $`git config user.name bench`.cwd(dir).quiet() + await $`git config commit.gpgsign false`.cwd(dir).quiet() + if (withBaseline) { + await Bun.write(path.join(dir, "app.py"), "def solve():\n return 0\n") + await commit(dir, "baseline") + } + return dir +} + +async function commit(dir: string, message: string) { + await $`git add -A`.cwd(dir).quiet() + await $`git commit -q -m ${message}`.cwd(dir).quiet() +} + +describe("bench patch mode", () => { + test("parsePatchMode defaults to worktree and rejects garbage", () => { + expect(parsePatchMode(undefined)).toBe("worktree") + expect(parsePatchMode("")).toBe("worktree") + expect(parsePatchMode("Committed")).toBe("committed") + expect(() => parsePatchMode("staged")).toThrow() + }) + + test("worktree mode captures uncommitted edits and new files", async () => { + const dir = await repo() + const baseline = await recordBaselineCommit(dir) + + await Bun.write(path.join(dir, "app.py"), "def solve():\n return 1\n") + await Bun.write(path.join(dir, "new_file.py"), "X = 1\n") + + const patch = await capturePatch(dir, "worktree", baseline) + expect(patch).toContain("+ return 1") + expect(patch).toContain("new_file.py") + }) + + test("worktree mode misses work the agent committed (the DeepSWE failure)", async () => { + const dir = await repo() + const baseline = await recordBaselineCommit(dir) + + await Bun.write(path.join(dir, "app.py"), "def solve():\n return 1\n") + await commit(dir, "fix") + + expect(await capturePatch(dir, "worktree", baseline)).toBe("") + }) + + test("committed mode captures a side-branch commit from either HEAD position", async () => { + const dir = await repo() + const baseline = await recordBaselineCommit(dir) + + await $`git checkout -q -b fix/solve`.cwd(dir).quiet() + await Bun.write(path.join(dir, "app.py"), "def solve():\n return 1\n") + await Bun.write(path.join(dir, "new_file.py"), "X = 1\n") + await commit(dir, "fix") + + const onBranch = await capturePatch(dir, "committed", baseline) + expect(onBranch).toContain("+ return 1") + expect(onBranch).toContain("new_file.py") + + // Agent switched back to main after committing: HEAD sits at the baseline, + // so only the local-branch scan finds the work. + await $`git checkout -q main`.cwd(dir).quiet() + expect(await capturePatch(dir, "committed", baseline)).toBe(onBranch) + }) + + test("committed mode picks the ref with the most commits past baseline", async () => { + const dir = await repo() + const baseline = await recordBaselineCommit(dir) + + await $`git checkout -q -b scratch`.cwd(dir).quiet() + await Bun.write(path.join(dir, "scratch.txt"), "debug\n") + await commit(dir, "scratch") + + await $`git checkout -q -b fix/solve main`.cwd(dir).quiet() + for (const n of [1, 2]) { + await Bun.write(path.join(dir, "app.py"), `def solve():\n return ${n}\n`) + await commit(dir, `step${n}`) + } + await $`git checkout -q main`.cwd(dir).quiet() + + const patch = await capturePatch(dir, "committed", baseline) + expect(patch).toContain("+ return 2") + expect(patch).not.toContain("scratch.txt") + }) + + test("committed mode excludes uncommitted leftovers", async () => { + const dir = await repo() + const baseline = await recordBaselineCommit(dir) + + await Bun.write(path.join(dir, "app.py"), "def solve():\n return 1\n") + await commit(dir, "fix") + await Bun.write(path.join(dir, "repro_scratch.py"), "print('debug')\n") + + const patch = await capturePatch(dir, "committed", baseline) + expect(patch).toContain("+ return 1") + expect(patch).not.toContain("repro_scratch.py") + }) + + test("committed mode yields an empty patch when the agent never committed", async () => { + const dir = await repo() + const baseline = await recordBaselineCommit(dir) + + await Bun.write(path.join(dir, "app.py"), "def solve():\n return 1\n") + + expect(await capturePatch(dir, "committed", baseline)).toBe("") + }) + + test("committed mode falls back to the empty tree for an unborn HEAD", async () => { + const dir = await repo(false) + const baseline = await recordBaselineCommit(dir) + expect(baseline).toBe("") + + await Bun.write(path.join(dir, "app.py"), "X = 1\n") + await commit(dir, "first") + + const patch = await capturePatch(dir, "committed", baseline) + expect(patch).toContain("app.py") + expect(patch).toContain("+X = 1") + }) + + test("ensureCommitIdentity fills a missing identity and keeps an existing one", async () => { + const dir = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "bench-patch-"))) + created.push(dir) + await $`git init -q -b main`.cwd(dir).quiet() + // A user-level identity would mask the "missing" case on a dev machine. + await $`git config --local user.useConfigOnly true`.cwd(dir).quiet() + + await ensureCommitIdentity(dir) + expect((await $`git config --get user.email`.cwd(dir).quiet().text()).trim()).not.toBe("") + + await $`git config --local user.email task@example.com`.cwd(dir).quiet() + await ensureCommitIdentity(dir) + expect((await $`git config --get user.email`.cwd(dir).quiet().text()).trim()).toBe("task@example.com") + }) +}) diff --git a/packages/opencode/test/bench/replay.test.ts b/packages/opencode/test/bench/replay.test.ts new file mode 100644 index 000000000000..25ab4c23b67b --- /dev/null +++ b/packages/opencode/test/bench/replay.test.ts @@ -0,0 +1,220 @@ +import { describe, test, expect } from "bun:test" +import { parseReplayManifest, parseReplayMessages, replayMessageText } from "@/bench/replay" + +describe("replayMessageText", () => { + test("returns string content verbatim", () => { + expect(replayMessageText("hello")).toBe("hello") + }) + + test("joins text parts from array content", () => { + expect( + replayMessageText([ + { type: "text", text: "line one" }, + { type: "text", text: "line two" }, + ]), + ).toBe("line one\nline two") + }) + + test("returns empty string for null/undefined content", () => { + expect(replayMessageText(null)).toBe("") + expect(replayMessageText(undefined)).toBe("") + }) +}) + +describe("parseReplayMessages", () => { + test("extracts the first user message as the initial task instruction", () => { + const raw = JSON.stringify([ + { role: "system", content: "sys prompt" }, + { role: "user", content: "fix the bug" }, + ]) + const { initialUserText, replayTurns } = parseReplayMessages(raw) + expect(initialUserText).toBe("fix the bug") + expect(replayTurns).toEqual([]) + }) + + test("skips system and tool messages", () => { + const raw = JSON.stringify([ + { role: "system", content: "sys prompt" }, + { role: "user", content: "fix the bug" }, + { + role: "assistant", + content: null, + tool_calls: [{ id: "call_1", type: "function", function: { name: "bash", arguments: '{"cmd":"ls"}' } }], + }, + { role: "tool", content: "file1.py\n", tool_call_id: "call_1" }, + { role: "assistant", content: "Done.", tool_calls: undefined }, + ]) + const { initialUserText, replayTurns } = parseReplayMessages(raw) + expect(initialUserText).toBe("fix the bug") + expect(replayTurns).toEqual([ + { content: null, toolCalls: [{ id: "call_1", name: "bash", arguments: '{"cmd":"ls"}' }] }, + { content: "Done.", toolCalls: undefined }, + ]) + }) + + test("attaches a subsequent user message to the turn it precedes, not dropped", () => { + const raw = JSON.stringify([ + { role: "user", content: "fix the bug" }, + { + role: "assistant", + content: null, + tool_calls: [{ id: "call_1", type: "function", function: { name: "bash", arguments: '{"cmd":"ls"}' } }], + }, + { role: "tool", content: "file1.py\n", tool_call_id: "call_1" }, + { role: "user", content: "please also fix the other bug" }, + { role: "assistant", content: "Done.", tool_calls: undefined }, + ]) + const { initialUserText, replayTurns } = parseReplayMessages(raw) + expect(initialUserText).toBe("fix the bug") + expect(replayTurns).toEqual([ + { content: null, toolCalls: [{ id: "call_1", name: "bash", arguments: '{"cmd":"ls"}' }] }, + { content: "Done.", toolCalls: undefined, precedingUserTexts: ["please also fix the other bug"] }, + ]) + }) + + test("collects multiple consecutive subsequent user messages onto the same turn, in order", () => { + const raw = JSON.stringify([ + { role: "user", content: "fix the bug" }, + { role: "assistant", content: "ok" }, + { role: "user", content: "also do X" }, + { role: "user", content: "and Y" }, + { role: "assistant", content: "done" }, + ]) + const { replayTurns } = parseReplayMessages(raw) + expect(replayTurns[1].precedingUserTexts).toEqual(["also do X", "and Y"]) + }) + + test("returns trailing user messages separately when the trajectory ends on a user turn", () => { + const raw = JSON.stringify([ + { role: "user", content: "fix the bug" }, + { + role: "assistant", + content: null, + tool_calls: [{ id: "call_1", type: "function", function: { name: "bash", arguments: '{"cmd":"ls"}' } }], + }, + { role: "tool", content: "file1.py\n", tool_call_id: "call_1" }, + { role: "user", content: "now also check the tests" }, + ]) + const { replayTurns, trailingUserTexts } = parseReplayMessages(raw) + expect(replayTurns).toEqual([ + { content: null, toolCalls: [{ id: "call_1", name: "bash", arguments: '{"cmd":"ls"}' }] }, + ]) + expect(trailingUserTexts).toEqual(["now also check the tests"]) + }) + + test("omits empty subsequent user message text", () => { + const raw = JSON.stringify([ + { role: "user", content: "fix the bug" }, + { role: "assistant", content: "ok" }, + { role: "user", content: "" }, + { role: "assistant", content: "done" }, + ]) + const { replayTurns } = parseReplayMessages(raw) + expect(replayTurns[1].precedingUserTexts).toBeUndefined() + }) + + test("preserves tool_call ids verbatim, including multiple calls in one turn", () => { + const raw = JSON.stringify([ + { role: "user", content: "fix the bug" }, + { + role: "assistant", + content: null, + tool_calls: [ + { id: "call_abc", type: "function", function: { name: "read", arguments: '{"path":"a.py"}' } }, + { id: "call_def", type: "function", function: { name: "read", arguments: '{"path":"b.py"}' } }, + ], + }, + ]) + const { replayTurns } = parseReplayMessages(raw) + expect(replayTurns[0].toolCalls?.map((tc) => tc.id)).toEqual(["call_abc", "call_def"]) + }) + + test("joins array-of-parts user content for the initial instruction", () => { + const raw = JSON.stringify([ + { role: "user", content: [{ type: "text", text: "part one" }, { type: "text", text: "part two" }] }, + ]) + const { initialUserText } = parseReplayMessages(raw) + expect(initialUserText).toBe("part one\npart two") + }) + + test("throws when no user message is present", () => { + const raw = JSON.stringify([{ role: "system", content: "sys prompt" }]) + expect(() => parseReplayMessages(raw)).toThrow(/no user message/) + }) +}) + +describe("parseReplayManifest", () => { + test("parses child and nested-child replay queues without relying on array order", () => { + const manifest = parseReplayManifest( + JSON.stringify({ + version: 1, + root_session_id: "recorded-root", + sessions: [ + { + session_id: "recorded-grandchild", + parent_session_id: "recorded-child", + spawn_call_id: "call_nested", + spawn_index: 0, + subagent_type: "explore", + messages: [ + { role: "user", content: "nested work" }, + { role: "assistant", content: "nested result" }, + ], + }, + { + session_id: "recorded-child", + parent_session_id: "recorded-root", + spawn_call_id: "call_child", + spawn_index: 1, + subagent_type: "general", + messages: [ + { role: "user", content: "child work" }, + { + role: "assistant", + content: null, + tool_calls: [ + { + id: "call_nested", + function: { name: "task", arguments: '{"prompt":"nested work"}' }, + }, + ], + }, + ], + }, + ], + }), + ) + + expect(manifest.rootSessionId).toBe("recorded-root") + expect(manifest.sessions.map((session) => session.sessionId)).toEqual([ + "recorded-grandchild", + "recorded-child", + ]) + expect(manifest.sessions[0]).toMatchObject({ + parentSessionId: "recorded-child", + spawnCallId: "call_nested", + messageCount: 2, + replayTurns: [{ content: "nested result" }], + }) + }) + + test("rejects an unlinked parent", () => { + expect(() => + parseReplayManifest( + JSON.stringify({ + version: 1, + root_session_id: "root", + sessions: [ + { + session_id: "child", + parent_session_id: "missing", + spawn_call_id: "call_1", + spawn_index: 0, + messages: [{ role: "user", content: "work" }], + }, + ], + }), + ), + ).toThrow(/unknown parent_session_id/) + }) +}) diff --git a/packages/opencode/test/bench/terminal_error.test.ts b/packages/opencode/test/bench/terminal_error.test.ts new file mode 100644 index 000000000000..5985446408f7 --- /dev/null +++ b/packages/opencode/test/bench/terminal_error.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, test } from "bun:test" +import * as BenchTerminalError from "@/bench/terminal_error" + +describe("bench terminal error signals", () => { + test("detects max-iteration and context-window markers", () => { + expect(BenchTerminalError.detect(`before ${BenchTerminalError.encode("max_iteration")} after`)).toBe( + "max_iteration", + ) + expect(BenchTerminalError.detect(BenchTerminalError.encode("context_window"))).toBe("context_window") + expect(BenchTerminalError.detect("ordinary opencode stderr")).toBeUndefined() + }) + + test("prefers context overflow when both terminal states occur", () => { + expect(BenchTerminalError.prefer("max_iteration", "context_window")).toBe("context_window") + expect(BenchTerminalError.prefer("context_window", "max_iteration")).toBe("context_window") + }) + + test("writes errors that Gym classifies and preserves ordinary exit errors", () => { + expect(BenchTerminalError.toGymError(0, "max_iteration")).toBe("maximum iteration reached") + expect(BenchTerminalError.toGymError(0, "context_window")).toBe("context window exceeded") + expect(BenchTerminalError.toGymError(17)).toBe("opencode_exit_17") + expect(BenchTerminalError.toGymError(0)).toBeNull() + }) + + test("keeps terminal trajectories even when opencode exits nonzero", () => { + expect(BenchTerminalError.shouldExitSuccessfully(1, "context_window")).toBeTrue() + expect(BenchTerminalError.shouldExitSuccessfully(1, "max_iteration")).toBeTrue() + expect(BenchTerminalError.shouldExitSuccessfully(1)).toBeFalse() + }) +}) diff --git a/packages/opencode/test/provider/nemo-gym/context-overflow.test.ts b/packages/opencode/test/provider/nemo-gym/context-overflow.test.ts new file mode 100644 index 000000000000..6e6117163f5b --- /dev/null +++ b/packages/opencode/test/provider/nemo-gym/context-overflow.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, mock, test } from "bun:test" +import type { LanguageModelV3CallOptions, LanguageModelV3StreamPart } from "@ai-sdk/provider" +import { NemoGymLanguageModel } from "@/provider/sdk/nemo-gym/language-model" + +async function drain(stream: ReadableStream): Promise { + const reader = stream.getReader() + const parts: LanguageModelV3StreamPart[] = [] + while (true) { + const { done, value } = await reader.read() + if (done) break + parts.push(value) + } + return parts +} + +const CALL_OPTIONS: LanguageModelV3CallOptions = { + prompt: [{ role: "user", content: [{ type: "text", text: "hi" }] }], + tools: [{ type: "function", name: "bash", inputSchema: { type: "object", properties: {} } }], +} + +describe("NemoGymLanguageModel context overflow", () => { + test("recognizes Gym's null-content length completion as context overflow", async () => { + const originalFetch = globalThis.fetch + const fetchSpy = mock( + async () => + new Response( + JSON.stringify({ + id: "chatcmpl-123", + model: "test-model", + choices: [ + { + index: 0, + finish_reason: "length", + message: { role: "assistant", content: null, tool_calls: null }, + }, + ], + }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ), + ) + // @ts-expect-error test override + globalThis.fetch = fetchSpy + + try { + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + retries: Number.MAX_SAFE_INTEGER, + }) + + const parts = await drain((await model.doStream(CALL_OPTIONS)).stream) + expect(fetchSpy).toHaveBeenCalledTimes(1) + + const error = parts.find((part) => part.type === "error") + expect(error?.type).toBe("error") + if (error?.type !== "error" || typeof error.error !== "string") throw new Error("missing stream error") + expect(JSON.parse(error.error)).toMatchObject({ + type: "error", + error: { code: "context_length_exceeded" }, + }) + } finally { + globalThis.fetch = originalFetch + } + }) +}) diff --git a/packages/opencode/test/provider/nemo-gym/language-model.test.ts b/packages/opencode/test/provider/nemo-gym/language-model.test.ts new file mode 100644 index 000000000000..bfbda261b67b --- /dev/null +++ b/packages/opencode/test/provider/nemo-gym/language-model.test.ts @@ -0,0 +1,535 @@ +import { describe, test, expect, mock } from "bun:test" +import { NemoGymLanguageModel } from "@/provider/sdk/nemo-gym/language-model" +import type { LanguageModelV3CallOptions, LanguageModelV3StreamPart } from "@ai-sdk/provider" + +async function drain(stream: ReadableStream): Promise { + const reader = stream.getReader() + const parts: LanguageModelV3StreamPart[] = [] + while (true) { + const { done, value } = await reader.read() + if (done) break + parts.push(value) + } + return parts +} + +// Real agentic-loop calls always carry the resolved tool registry +// (session/prompt.ts's resolveTools()) — this is what actually distinguishes +// them from auxiliary same-session calls like title/summary generation, +// which never pass tools. See _popReplayTurn's docblock in language-model.ts. +const AGENT_TOOLS: LanguageModelV3CallOptions["tools"] = [ + { type: "function", name: "bash", inputSchema: { type: "object", properties: {} } }, +] + +const CALL_OPTIONS: LanguageModelV3CallOptions = { + prompt: [{ role: "user", content: [{ type: "text", text: "hi" }] }], + tools: AGENT_TOOLS, +} + +describe("NemoGymLanguageModel replay", () => { + test("doStream replays a scripted tool-call turn without hitting the network", async () => { + const fetchSpy = mock(async () => { + throw new Error("network should not be called during replay") + }) + // @ts-expect-error test override + globalThis.fetch = fetchSpy + + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + replayTurns: [{ content: null, toolCalls: [{ id: "call_1", name: "bash", arguments: '{"cmd":"ls"}' }] }], + }) + + const { stream } = await model.doStream(CALL_OPTIONS) + const parts = await drain(stream) + + expect(fetchSpy).not.toHaveBeenCalled() + + const toolCall = parts.find((p) => p.type === "tool-call") + expect(toolCall).toMatchObject({ toolCallId: "call_1", toolName: "bash", input: '{"cmd":"ls"}' }) + + const finish = parts.find((p) => p.type === "finish") + expect(finish).toMatchObject({ finishReason: { unified: "tool-calls" } }) + }) + + test("doStream replays a scripted text-only turn (no tool calls) as finishReason stop", async () => { + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + replayTurns: [{ content: "All done." }], + }) + + const { stream } = await model.doStream(CALL_OPTIONS) + const parts = await drain(stream) + + expect(parts.some((p) => p.type === "tool-call")).toBe(false) + const textDelta = parts.find((p) => p.type === "text-delta") + expect(textDelta).toMatchObject({ delta: "All done." }) + const finish = parts.find((p) => p.type === "finish") + expect(finish).toMatchObject({ finishReason: { unified: "stop" } }) + }) + + test("doStream falls through to the real HTTP path once the replay queue is exhausted", async () => { + const fetchSpy = mock( + async () => + new Response( + JSON.stringify({ + id: "resp_1", + model: "test-model", + choices: [{ finish_reason: "stop", message: { role: "assistant", content: "live turn" } }], + }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ), + ) + // @ts-expect-error test override + globalThis.fetch = fetchSpy + + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + replayTurns: [{ content: "scripted turn" }], + }) + + // First call: replay (no fetch). + await drain((await model.doStream(CALL_OPTIONS)).stream) + expect(fetchSpy).not.toHaveBeenCalled() + + // Second call: replay queue exhausted -> real HTTP path. + const { stream } = await model.doStream(CALL_OPTIONS) + const parts = await drain(stream) + expect(fetchSpy).toHaveBeenCalledTimes(1) + const textDelta = parts.find((p) => p.type === "text-delta") + expect(textDelta).toMatchObject({ delta: "live turn" }) + }) + + test("doStream recognizes Gym's null-content length completion as context overflow", async () => { + const fetchSpy = mock( + async () => + new Response( + JSON.stringify({ + id: "chtcmpl-123", + model: "test-model", + choices: [ + { + index: 0, + finish_reason: "length", + message: { role: "assistant", content: null, tool_calls: null }, + }, + ], + }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ), + ) + // @ts-expect-error test override + globalThis.fetch = fetchSpy + + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + retries: Number.MAX_SAFE_INTEGER, + }) + + const parts = await drain((await model.doStream(CALL_OPTIONS)).stream) + expect(fetchSpy).toHaveBeenCalledTimes(1) + + const error = parts.find((part) => part.type === "error") + expect(error?.type).toBe("error") + if (error?.type !== "error" || typeof error.error !== "string") throw new Error("missing stream error") + expect(JSON.parse(error.error)).toMatchObject({ + type: "error", + error: { code: "context_length_exceeded" }, + }) + }) + + test("replay is scoped to the top-level session — a subagent session (x-parent-session-id set) calls through to HTTP", async () => { + const fetchSpy = mock( + async () => + new Response( + JSON.stringify({ + id: "resp_1", + model: "test-model", + choices: [{ finish_reason: "stop", message: { role: "assistant", content: "subagent turn" } }], + }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ), + ) + // @ts-expect-error test override + globalThis.fetch = fetchSpy + + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + replayTurns: [{ content: "scripted turn" }], + }) + + // A subagent session carries its OWN session id plus x-parent-session-id + // pointing at the main session — session/llm.ts sets both unconditionally + // for every call, subagent or not, so it's parentSessionID's presence + // that identifies a subagent, not the session id string itself. + const subagentOptions: LanguageModelV3CallOptions = { + ...CALL_OPTIONS, + headers: { "x-session-affinity": "ses_subagent_1", "x-parent-session-id": "ses_main" }, + } + const { stream } = await model.doStream(subagentOptions) + const parts = await drain(stream) + expect(fetchSpy).toHaveBeenCalledTimes(1) + expect(parts.find((p) => p.type === "text-delta")).toMatchObject({ delta: "subagent turn" }) + }) + + test("an auxiliary no-tool call on the main session (e.g. opencode's own title/summary generation) does not consume the replay queue", async () => { + const fetchSpy = mock( + async () => + new Response( + JSON.stringify({ + id: "resp_1", + model: "test-model", + choices: [{ finish_reason: "stop", message: { role: "assistant", content: "Fix the parser bug" } }], + }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ), + ) + // @ts-expect-error test override + globalThis.fetch = fetchSpy + + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + replayTurns: [{ content: null, toolCalls: [{ id: "call_1", name: "bash", arguments: "{}" }] }], + }) + + // session/prompt.ts's runLoop forks off a title-generation call on step 1, + // using the SAME session id as the real agentic loop but never passing + // tools. Real x-session-affinity value from opencode: same session, + // no x-parent-session-id (it's not a subagent), no tools. + const titleGenOptions: LanguageModelV3CallOptions = { + prompt: [{ role: "user", content: [{ type: "text", text: "Generate a title" }] }], + headers: { "x-session-affinity": "ses_main" }, + // no tools + } + const { stream } = await model.doStream(titleGenOptions) + await drain(stream) + expect(fetchSpy).toHaveBeenCalledTimes(1) // went straight to HTTP, not replayed + + // The scripted turn is still there for the real agent's own next call + // (same session id, this time with tools). + const realOptions: LanguageModelV3CallOptions = { + ...CALL_OPTIONS, + headers: { "x-session-affinity": "ses_main" }, + } + const realFetchSpy = mock(async () => { + throw new Error("network should not be called — replay turn should still be available") + }) + // @ts-expect-error test override + globalThis.fetch = realFetchSpy + const { stream: realStream } = await model.doStream(realOptions) + const parts = await drain(realStream) + expect(realFetchSpy).not.toHaveBeenCalled() + expect(parts.find((p) => p.type === "tool-call")).toMatchObject({ toolCallId: "call_1" }) + }) + + function fetchSpyCapturingBody() { + const bodies: Array<{ messages: unknown }> = [] + const fetchSpy = mock(async (_url: unknown, init: { body?: string }) => { + bodies.push(JSON.parse(init.body ?? "{}")) + return new Response( + JSON.stringify({ + id: "resp_1", + model: "test-model", + choices: [{ finish_reason: "stop", message: { role: "assistant", content: "live turn" } }], + }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ) + }) + // @ts-expect-error test override + globalThis.fetch = fetchSpy + return { fetchSpy, bodies } + } + + test("subsequent user message (precedingUserTexts) is spliced at the right position relative to already-replayed turns", async () => { + const { bodies } = fetchSpyCapturingBody() + + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + // Attached to turn 0 (ordinal 0): must land BEFORE the first assistant + // message in the eventual wire request, i.e. right after the initial + // user message. + replayTurns: [ + { + content: null, + toolCalls: [{ id: "call_1", name: "bash", arguments: "{}" }], + precedingUserTexts: ["please also fix the other bug"], + }, + ], + }) + + // Turn 0 replays without hitting the network. + await drain((await model.doStream(CALL_OPTIONS)).stream) + expect(bodies).toHaveLength(0) + + // Live call, with a realistic prompt reflecting what the session would + // actually contain by now: the initial user message, then the assistant + // turn + tool result that were just replayed for real. + const grownPrompt: LanguageModelV3CallOptions["prompt"] = [ + { role: "user", content: [{ type: "text", text: "fix the bug" }] }, + { role: "assistant", content: [{ type: "tool-call", toolCallId: "call_1", toolName: "bash", input: {} }] }, + { + role: "tool", + content: [ + { type: "tool-result", toolCallId: "call_1", toolName: "bash", output: { type: "text", value: "ok" } }, + ], + }, + ] + await drain((await model.doStream({ prompt: grownPrompt, tools: AGENT_TOOLS })).stream) + expect(bodies).toHaveLength(1) + + const messages = bodies[0].messages as unknown as Array<{ role: string; content?: unknown }> + expect(messages[0]).toMatchObject({ role: "user", content: "fix the bug" }) + const injectedIdx = messages.findIndex((m) => m.role === "user" && m.content === "please also fix the other bug") + const firstAssistantIdx = messages.findIndex((m) => m.role === "assistant") + expect(injectedIdx).toBeGreaterThan(-1) + expect(firstAssistantIdx).toBeGreaterThan(-1) + // Injected immediately before the assistant turn it originally + // preceded, not appended somewhere arbitrary — and strictly after the + // initial task-instruction message. + expect(injectedIdx).toBe(firstAssistantIdx - 1) + expect(injectedIdx).toBeGreaterThan(0) + }) + + test("injected user message persists on every subsequent live call, not just the first", async () => { + const { bodies } = fetchSpyCapturingBody() + + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + replayTurns: [{ content: "ok", precedingUserTexts: ["please also fix the other bug"] }], + }) + + await drain((await model.doStream(CALL_OPTIONS)).stream) // scripted + await drain((await model.doStream(CALL_OPTIONS)).stream) // live #1 + await drain((await model.doStream(CALL_OPTIONS)).stream) // live #2 + + expect(bodies).toHaveLength(2) + for (const body of bodies) { + const messages = body.messages as Array<{ role: string; content?: unknown }> + expect(messages.some((m) => m.role === "user" && m.content === "please also fix the other bug")).toBe(true) + } + }) + + test("replayTrailingUserTexts is appended once the replay queue is fully drained", async () => { + const { bodies } = fetchSpyCapturingBody() + + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + replayTurns: [{ content: null, toolCalls: [{ id: "call_1", name: "bash", arguments: "{}" }] }], + replayTrailingUserTexts: ["now also check the tests"], + }) + + await drain((await model.doStream(CALL_OPTIONS)).stream) // scripted + await drain((await model.doStream(CALL_OPTIONS)).stream) // live + + expect(bodies).toHaveLength(1) + const messages = bodies[0].messages as unknown as Array<{ role: string; content?: unknown }> + expect(messages[messages.length - 1]).toMatchObject({ role: "user", content: "now also check the tests" }) + }) + + test("binds parallel child queues by parent task call id, not manifest or execution order", async () => { + const fetchSpy = mock(async () => { + throw new Error("network should not be called during replay") + }) + // @ts-expect-error test override + globalThis.fetch = fetchSpy + + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + replayTurns: [ + { + content: null, + toolCalls: [ + { id: "call_a", name: "task", arguments: '{"prompt":"A","subagent_type":"general"}' }, + { id: "call_b", name: "task", arguments: '{"prompt":"B","subagent_type":"explore"}' }, + ], + }, + ], + replayManifest: { + version: 1, + rootSessionId: "recorded-root", + // Deliberately opposite the parent task-call order. + sessions: [ + { + sessionId: "recorded-b", + parentSessionId: "recorded-root", + spawnCallId: "call_b", + spawnIndex: 1, + messageCount: 2, + replayTurns: [{ content: "result B" }], + }, + { + sessionId: "recorded-a", + parentSessionId: "recorded-root", + spawnCallId: "call_a", + spawnIndex: 0, + messageCount: 2, + replayTurns: [{ content: "result A" }], + }, + ], + }, + }) + + await drain((await model.doStream({ ...CALL_OPTIONS, headers: { "x-session-affinity": "live-root" } })).stream) + + const childB = await drain( + ( + await model.doStream({ + ...CALL_OPTIONS, + headers: { + "x-session-affinity": "live-b", + "x-parent-session-id": "live-root", + "x-parent-tool-call-id": "call_b", + }, + }) + ).stream, + ) + const childA = await drain( + ( + await model.doStream({ + ...CALL_OPTIONS, + headers: { + "x-session-affinity": "live-a", + "x-parent-session-id": "live-root", + "x-parent-tool-call-id": "call_a", + }, + }) + ).stream, + ) + + expect(childB.find((part) => part.type === "text-delta")).toMatchObject({ delta: "result B" }) + expect(childA.find((part) => part.type === "text-delta")).toMatchObject({ delta: "result A" }) + expect(fetchSpy).not.toHaveBeenCalled() + }) + + test("recursively binds a nested subagent to the task call in its recorded parent", async () => { + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + replayTurns: [ + { + content: null, + toolCalls: [{ id: "call_child", name: "task", arguments: "{}" }], + }, + ], + replayManifest: { + version: 1, + rootSessionId: "recorded-root", + sessions: [ + { + sessionId: "recorded-child", + parentSessionId: "recorded-root", + spawnCallId: "call_child", + spawnIndex: 0, + messageCount: 3, + replayTurns: [ + { + content: null, + toolCalls: [{ id: "call_grandchild", name: "task", arguments: "{}" }], + }, + ], + }, + { + sessionId: "recorded-grandchild", + parentSessionId: "recorded-child", + spawnCallId: "call_grandchild", + spawnIndex: 0, + messageCount: 2, + replayTurns: [{ content: "nested result" }], + }, + ], + }, + }) + + await drain((await model.doStream({ ...CALL_OPTIONS, headers: { "x-session-affinity": "live-root" } })).stream) + await drain( + ( + await model.doStream({ + ...CALL_OPTIONS, + headers: { + "x-session-affinity": "live-child", + "x-parent-session-id": "live-root", + "x-parent-tool-call-id": "call_child", + }, + }) + ).stream, + ) + const nested = await drain( + ( + await model.doStream({ + ...CALL_OPTIONS, + headers: { + "x-session-affinity": "live-grandchild", + "x-parent-session-id": "live-child", + "x-parent-tool-call-id": "call_grandchild", + }, + }) + ).stream, + ) + + expect(nested.find((part) => part.type === "text-delta")).toMatchObject({ delta: "nested result" }) + }) + + test("rewrites a recorded task_id to the bound live child session id when resuming it", async () => { + const model = new NemoGymLanguageModel("test-model", { + provider: "nemo-gym", + baseURL: "http://unused.invalid", + replayTurns: [ + { content: null, toolCalls: [{ id: "call_child", name: "task", arguments: "{}" }] }, + { + content: null, + toolCalls: [ + { + id: "call_resume", + name: "task", + arguments: '{"task_id":"recorded-child","prompt":"continue"}', + }, + ], + }, + ], + replayManifest: { + version: 1, + rootSessionId: "recorded-root", + sessions: [ + { + sessionId: "recorded-child", + parentSessionId: "recorded-root", + spawnCallId: "call_child", + spawnIndex: 0, + messageCount: 2, + replayTurns: [{ content: "first child result" }], + }, + ], + }, + }) + + const rootHeaders = { "x-session-affinity": "live-root" } + await drain((await model.doStream({ ...CALL_OPTIONS, headers: rootHeaders })).stream) + await drain( + ( + await model.doStream({ + ...CALL_OPTIONS, + headers: { + "x-session-affinity": "live-child", + "x-parent-session-id": "live-root", + "x-parent-tool-call-id": "call_child", + }, + }) + ).stream, + ) + const resumed = await drain((await model.doStream({ ...CALL_OPTIONS, headers: rootHeaders })).stream) + const call = resumed.find((part) => part.type === "tool-call") + expect(call).toMatchObject({ + toolCallId: "call_resume", + input: '{"task_id":"live-child","prompt":"continue"}', + }) + }) +}) diff --git a/packages/opencode/test/tool/task.test.ts b/packages/opencode/test/tool/task.test.ts index f75fcf84b8a9..6cf0a302966c 100644 --- a/packages/opencode/test/tool/task.test.ts +++ b/packages/opencode/test/tool/task.test.ts @@ -196,6 +196,15 @@ describe("tool.task", () => { const sessions = yield* Session.Service const { chat, assistant } = yield* seed() const child = yield* sessions.create({ parentID: chat.id, title: "Existing child" }) + yield* sessions.updateMessage({ + id: MessageID.ascending(), + role: "user", + sessionID: child.id, + parentToolCallID: "call_spawn", + agent: "general", + model: ref, + time: { created: Date.now() }, + }) const tool = yield* TaskTool const def = yield* tool.init() let seen: SessionPrompt.PromptInput | undefined @@ -211,6 +220,7 @@ describe("tool.task", () => { { sessionID: chat.id, messageID: assistant.id, + callID: "call_resume", agent: "build", abort: new AbortController().signal, extra: { promptOps }, @@ -226,6 +236,7 @@ describe("tool.task", () => { expect(result.metadata.sessionId).toBe(child.id) expect(result.output).toContain(`task_id: ${child.id}`) expect(seen?.sessionID).toBe(child.id) + expect(seen?.parentToolCallID).toBe("call_spawn") }), )