diff --git a/.github/workflows/bundle-gate.yml b/.github/workflows/bundle-gate.yml index b344b1a..e741961 100644 --- a/.github/workflows/bundle-gate.yml +++ b/.github/workflows/bundle-gate.yml @@ -44,10 +44,9 @@ jobs: - name: Build a single-file stable plugin from source run: npm run build:plugin - - name: Verify generated entry syntax and tests + - name: Verify generated entry syntax and import run: | node --check src/index.js - npm test bun -e "await import('./src/index.js')" - name: Install current OpenCode CLI @@ -56,8 +55,14 @@ jobs: - name: Show OpenCode version run: npx --no-install opencode --version - - name: Run real-host canary against generated bundle - run: node scripts/host-loop-canary.mjs + - name: Verify real-host Loop and QUEUED Goal steering against clean generated bundle + run: npm run canary:host env: OPENCODE_PRINT_LOGS: "1" OPENCODE_LOG_LEVEL: DEBUG + + - name: Restore locked test dependencies + run: npm ci + + - name: Run generated-bundle regression suite + run: npm test diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f6d73b6..6a6271b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -160,8 +160,8 @@ jobs: - name: Show OpenCode version run: npx --no-install opencode --version - - name: Run real-host Loop canary - run: node scripts/host-loop-canary.mjs + - name: Run real-host Loop and queued Goal steering canaries + run: npm run canary:host env: OPENCODE_PRINT_LOGS: "1" OPENCODE_LOG_LEVEL: DEBUG diff --git a/package.json b/package.json index 07e0d01..f911179 100644 --- a/package.json +++ b/package.json @@ -21,9 +21,9 @@ "build:plugin": "bun build src/source/v1.js --outfile=src/index.js --target=bun --format=esm --external=@opencode-ai/plugin/tool", "build:plugin:npm": "npm run build:plugin", "prepack": "node --check src/index.js", - "check": "node --check src/source/v1.js && node --check src/source/core/args.js && node --check src/source/core/state.js && node --check src/source/core/jobs.js && node --check src/source/core/process.js && node --check src/source/opencode/sdk.js && node --check src/source/opencode/session-context.js && node --check src/source/opencode/command-router.js && node --check src/source/opencode/goal-commands.js && node --check src/source/opencode/loop-commands.js && node --check src/source/opencode/loop-registration.js && node --check src/source/runtime/session-activity.js && node --check src/source/runtime/session-status.js && node --check src/source/runtime/compaction.js && node --check src/source/runtime/executor.js && node --check src/source/runtime/scheduler.js && node --check src/source/runtime/goal-runtime.js && node --check src/source/runtime/goal-policy.js && node --check src/source/runtime/goal-steering.js && node --check src/source/runtime/job-workspace.js && node --check src/source/opencode2/prompt-runtime.js && node --check src/source/legacy-v1.js && node --check src/index.js && node --check scripts/install-node.mjs && node --check scripts/loopd.mjs && node --check scripts/install-test.mjs && node --check scripts/loopd-test.mjs && node --check scripts/smoke-test.mjs && node --check scripts/host-adapter-contract-test.mjs && node --check scripts/command-router-test.mjs && node --check scripts/goal-command-handlers-test.mjs && node --check scripts/loop-command-handlers-test.mjs && node --check scripts/loop-registration-test.mjs && node --check scripts/session-activity-test.mjs && node --check scripts/session-status-test.mjs && node --check scripts/compaction-runtime-test.mjs && node --check scripts/executor-runtime-test.mjs && node --check scripts/scheduler-runtime-test.mjs && node --check scripts/goal-runtime-test.mjs && node --check scripts/goal-policy-test.mjs && node --check scripts/goal-steering-test.mjs && node --check scripts/job-workspace-test.mjs && node --check scripts/v2-prompt-runtime-test.mjs && node --check scripts/v2-prompt-interval-test.mjs && node --check scripts/v2-command-runtime-test.mjs && node --check scripts/v2-command-adapter-test.mjs && node --check scripts/comprehensive-watchdog.mjs && node --check scripts/comprehensive-test.mjs && node --check scripts/host-loop-canary.mjs", + "check": "node --check src/source/v1.js && node --check src/source/core/args.js && node --check src/source/core/state.js && node --check src/source/core/jobs.js && node --check src/source/core/process.js && node --check src/source/opencode/sdk.js && node --check src/source/opencode/session-context.js && node --check src/source/opencode/command-router.js && node --check src/source/opencode/goal-commands.js && node --check src/source/opencode/loop-commands.js && node --check src/source/opencode/loop-registration.js && node --check src/source/runtime/session-activity.js && node --check src/source/runtime/session-status.js && node --check src/source/runtime/compaction.js && node --check src/source/runtime/executor.js && node --check src/source/runtime/scheduler.js && node --check src/source/runtime/goal-runtime.js && node --check src/source/runtime/goal-policy.js && node --check src/source/runtime/goal-steering.js && node --check src/source/runtime/job-workspace.js && node --check src/source/opencode2/prompt-runtime.js && node --check src/source/legacy-v1.js && node --check src/index.js && node --check scripts/install-node.mjs && node --check scripts/loopd.mjs && node --check scripts/install-test.mjs && node --check scripts/loopd-test.mjs && node --check scripts/smoke-test.mjs && node --check scripts/host-adapter-contract-test.mjs && node --check scripts/command-router-test.mjs && node --check scripts/goal-command-handlers-test.mjs && node --check scripts/loop-command-handlers-test.mjs && node --check scripts/loop-registration-test.mjs && node --check scripts/session-activity-test.mjs && node --check scripts/session-status-test.mjs && node --check scripts/compaction-runtime-test.mjs && node --check scripts/executor-runtime-test.mjs && node --check scripts/scheduler-runtime-test.mjs && node --check scripts/goal-runtime-test.mjs && node --check scripts/goal-policy-test.mjs && node --check scripts/goal-steering-test.mjs && node --check scripts/job-workspace-test.mjs && node --check scripts/v2-prompt-runtime-test.mjs && node --check scripts/v2-prompt-interval-test.mjs && node --check scripts/v2-command-runtime-test.mjs && node --check scripts/v2-command-adapter-test.mjs && node --check scripts/comprehensive-watchdog.mjs && node --check scripts/comprehensive-test.mjs && node --check scripts/host-loop-canary.mjs && node --check scripts/host-goal-steering-canary.mjs", "test": "node scripts/command-router-test.mjs && node scripts/goal-command-handlers-test.mjs && node scripts/loop-command-handlers-test.mjs && node scripts/loop-registration-test.mjs && node scripts/session-activity-test.mjs && node scripts/session-status-test.mjs && node scripts/compaction-runtime-test.mjs && node scripts/executor-runtime-test.mjs && node scripts/scheduler-runtime-test.mjs && node scripts/goal-runtime-test.mjs && node scripts/goal-policy-test.mjs && node scripts/goal-steering-test.mjs && node scripts/job-workspace-test.mjs && node scripts/v2-prompt-runtime-test.mjs && node scripts/v2-prompt-interval-test.mjs && node scripts/v2-command-runtime-test.mjs && node scripts/v2-command-adapter-test.mjs && node scripts/install-test.mjs && node scripts/loopd-test.mjs && node scripts/smoke-test.mjs && node scripts/host-adapter-contract-test.mjs && node scripts/comprehensive-watchdog.mjs", - "canary:host": "node scripts/host-loop-canary.mjs", + "canary:host": "node scripts/host-loop-canary.mjs && node scripts/host-goal-steering-canary.mjs", "install:global": "node scripts/install-node.mjs", "pack:zip": "node scripts/make-zip.mjs" }, diff --git a/scripts/goal-steering-test.mjs b/scripts/goal-steering-test.mjs index 5a338d0..523153a 100644 --- a/scripts/goal-steering-test.mjs +++ b/scripts/goal-steering-test.mjs @@ -1,5 +1,6 @@ import assert from "node:assert/strict" import { createGoalSteeringRuntime } from "../src/source/runtime/goal-steering.js" +import { guardLoopOwnedUserMessage, loopOwnedUserMessageGuardActive, clearLoopOwnedUserMessageGuard } from "../src/source/opencode/messages.js" function goal(id = "goal-1") { return { @@ -49,17 +50,22 @@ function assistantEvent(sessionID = "ses-steering", parentID = "user-steering") now: () => 1000, }) - const result = await runtime.handleEvent("/workspace", client, userEvent()) + const result = await runtime.handleUserMessage("/workspace", client, { sessionID: "ses-steering", messageID: "user-steering" }) assert.equal(result.handled, true) assert.equal(result.preempted, true) - assert.equal(aborts, 1, "queued user steering should abort only the active Goal turn") + assert.equal(aborts, 1, "chat.message steering should abort only the active Goal turn before the new prompt dispatches") assert.equal(clears, 1, "the aborted Goal run must no longer be finalized as a completed loop run") assert.equal(runtime.shouldSuppressIdle("ses-steering"), true, "abort-generated idle must be suppressed while steering is queued") assert.deepEqual(state.jobs[0], originalGoal, "normal steering must not pause or rewrite the Goal contract") assert.equal(logs.at(-1)?.type, "goal-user-steering") assert.equal(logs.at(-1)?.detail?.preempted, true) - runtime.handleEvent("/workspace", client, assistantEvent()) + const duplicate = await runtime.handleEvent("/workspace", client, userEvent()) + assert.equal(duplicate.duplicate, true, "message.updated must not abort a steering message already handled by chat.message") + assert.equal(aborts, 1) + assert.equal(clears, 1) + + await runtime.handleEvent("/workspace", client, assistantEvent()) assert.equal(runtime.shouldSuppressIdle("ses-steering"), false, "matching steering assistant turn should release idle suppression") } @@ -75,7 +81,7 @@ function assistantEvent(sessionID = "ses-steering", parentID = "user-steering") isLoopOwnedUserMessage: () => false, now: () => 2000, }) - const result = await runtime.handleEvent("/workspace", { session: { abort: async () => {} } }, userEvent("ses-idle", "user-idle")) + const result = await runtime.handleUserMessage("/workspace", { session: { abort: async () => {} } }, { sessionID: "ses-idle", messageID: "user-idle" }) assert.equal(result.handled, true) assert.equal(result.preempted, false) assert.equal(aborts, 0, "an idle Goal should let the foreground user turn start normally") @@ -95,7 +101,7 @@ function assistantEvent(sessionID = "ses-steering", parentID = "user-steering") isLoopOwnedUserMessage: () => false, now: () => 3000, }) - const result = await runtime.handleEvent("/workspace", { session: { abort: async () => {} } }, userEvent("ses-ordinary", "user-ordinary")) + const result = await runtime.handleUserMessage("/workspace", { session: { abort: async () => {} } }, { sessionID: "ses-ordinary", messageID: "user-ordinary" }) assert.equal(result.handled, true) assert.equal(result.preempted, false) assert.equal(aborts, 0) @@ -110,7 +116,7 @@ function assistantEvent(sessionID = "ses-steering", parentID = "user-steering") appendLoopLog: async () => {}, isLoopOwnedUserMessage: () => true, }) - const result = await runtime.handleEvent("/workspace", { session: {} }, userEvent("ses-owned", "owned-message")) + const result = await runtime.handleUserMessage("/workspace", { session: {} }, { sessionID: "ses-owned", messageID: "owned-message" }) assert.equal(result.loopOwned, true) assert.equal(reads, 0, "Loop-owned synthetic user messages must not be mistaken for human steering") } @@ -128,10 +134,20 @@ function assistantEvent(sessionID = "ses-steering", parentID = "user-steering") now: () => clock, suppressionMs: 100, }) - await runtime.handleEvent("/workspace", { session: { abort: async () => {} } }, userEvent("ses-expire", "user-expire")) + await runtime.handleUserMessage("/workspace", { session: { abort: async () => {} } }, { sessionID: "ses-expire", messageID: "user-expire" }) assert.equal(runtime.shouldSuppressIdle("ses-expire"), true) clock += 101 assert.equal(runtime.shouldSuppressIdle("ses-expire"), false, "lost steering must not suppress autonomous continuation forever") } +{ + const sessionID = "ses-loop-owned-two-stage" + guardLoopOwnedUserMessage(sessionID) + assert.equal(loopOwnedUserMessageGuardActive(sessionID), true, "chat.message should recognize a pending synthetic prompt before OpenCode assigns its ID") + assert.equal(loopOwnedUserMessageGuardActive(sessionID, "synthetic-user-1"), true, "message.updated should consume the pending guard and bind the real synthetic message ID") + assert.equal(loopOwnedUserMessageGuardActive(sessionID, "synthetic-user-1"), true, "the bound synthetic message ID should remain Loop-owned for later duplicate events") + clearLoopOwnedUserMessageGuard(sessionID) + assert.equal(loopOwnedUserMessageGuardActive(sessionID, "synthetic-user-1"), false) +} + console.log("Goal steering runtime tests passed") diff --git a/scripts/host-goal-steering-canary.mjs b/scripts/host-goal-steering-canary.mjs new file mode 100644 index 0000000..86b3f88 --- /dev/null +++ b/scripts/host-goal-steering-canary.mjs @@ -0,0 +1,420 @@ +import assert from "node:assert/strict" +import { existsSync } from "node:fs" +import { createServer } from "node:http" +import net from "node:net" +import { spawn } from "node:child_process" +import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises" +import os from "node:os" +import path from "node:path" +import { fileURLToPath, pathToFileURL } from "node:url" + +const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..") +const isWindows = process.platform === "win32" +const GOAL_OBJECTIVE = "queued steering real host goal objective" +const USER_STEERING = "queued user steering must run before goal continuation" + +function resolveOpenCodeBinary() { + if (!isWindows) return path.join(repoRoot, "node_modules", ".bin", "opencode") + const candidates = [ + path.join(repoRoot, "node_modules", "opencode-windows-x64", "bin", "opencode.exe"), + path.join(repoRoot, "node_modules", "opencode-windows-x64-baseline", "bin", "opencode.exe"), + path.join(repoRoot, "node_modules", "opencode-windows-arm64", "bin", "opencode.exe"), + ] + const found = candidates.find((candidate) => existsSync(candidate)) + if (!found) throw new Error(`OpenCode native Windows binary was not installed. Checked: ${candidates.join(", ")}`) + return found +} + +const opencodeBin = resolveOpenCodeBinary() + +function appendLog(current, chunk, limit = 80_000) { + return (current + String(chunk)).slice(-limit) +} + +async function seedConfigDependencies(dir) { + await mkdir(path.join(dir, "node_modules"), { recursive: true }) + const dependencies = { "@opencode-ai/plugin": "*" } + await writeFile(path.join(dir, "package.json"), `${JSON.stringify({ private: true, dependencies }, null, 2)}\n`) + await writeFile( + path.join(dir, "package-lock.json"), + `${JSON.stringify({ + name: "opencode-loop-goal-steering-canary-config", + lockfileVersion: 3, + requires: true, + packages: { "": { dependencies } }, + }, null, 2)}\n`, + ) + await writeFile(path.join(dir, ".gitignore"), "node_modules\npackage.json\npackage-lock.json\nbun.lock\n.gitignore\n") +} + +async function reservePort() { + return await new Promise((resolve, reject) => { + const server = net.createServer() + server.once("error", reject) + server.listen(0, "127.0.0.1", () => { + const address = server.address() + if (!address || typeof address === "string") return reject(new Error("failed to reserve TCP port")) + server.close((error) => error ? reject(error) : resolve(address.port)) + }) + }) +} + +async function waitForTcp(port, child, logs, timeoutMs = 30_000) { + const deadline = Date.now() + timeoutMs + while (Date.now() < deadline) { + if (child.exitCode !== null) throw new Error(`OpenCode server exited before ready.\n${logs()}`) + const connected = await new Promise((resolve) => { + const socket = net.createConnection({ host: "127.0.0.1", port }) + socket.once("connect", () => { socket.destroy(); resolve(true) }) + socket.once("error", () => resolve(false)) + socket.setTimeout(500, () => { socket.destroy(); resolve(false) }) + }) + if (connected) return + await new Promise((resolve) => setTimeout(resolve, 100)) + } + throw new Error(`timed out waiting for OpenCode server on ${port}\n${logs()}`) +} + +async function stopProcess(child, timeoutMs = 2_000) { + if (!child || child.exitCode !== null) return + child.kill() + await new Promise((resolve) => { + if (child.exitCode !== null) return resolve() + const timer = setTimeout(resolve, timeoutMs) + child.once("close", () => { clearTimeout(timer); resolve() }) + }) +} + +function spawnOpenCode(args, options = {}) { + return spawn(opencodeBin, args, { ...options, windowsHide: true }) +} + +async function runOpenCode(args, { cwd, env, timeoutMs = 60_000 }) { + return await new Promise((resolve, reject) => { + const child = spawnOpenCode(args, { cwd, env }) + let stdout = "" + let stderr = "" + let settled = false + child.stdout?.on("data", (chunk) => { stdout = appendLog(stdout, chunk) }) + child.stderr?.on("data", (chunk) => { stderr = appendLog(stderr, chunk) }) + const finish = (fn, value) => { + if (settled) return + settled = true + clearTimeout(timer) + fn(value) + } + const timer = setTimeout(() => { + void stopProcess(child) + finish(reject, new Error(`OpenCode command timed out: ${args.join(" ")}\nstdout:\n${stdout}\nstderr:\n${stderr}`)) + }, timeoutMs) + child.once("error", (error) => finish(reject, error)) + child.once("close", (code) => { + if (code !== 0) return finish(reject, new Error(`OpenCode command exited ${code}: ${args.join(" ")}\nstdout:\n${stdout}\nstderr:\n${stderr}`)) + finish(resolve, { stdout, stderr }) + }) + }) +} + +function contentText(content) { + if (typeof content === "string") return content + if (!Array.isArray(content)) return "" + return content.map((part) => typeof part?.text === "string" ? part.text : typeof part?.content === "string" ? part.content : "").join("\n") +} + +function lastUserText(body) { + const messages = Array.isArray(body.messages) ? body.messages : [] + for (let index = messages.length - 1; index >= 0; index -= 1) { + if (messages[index]?.role === "user") return contentText(messages[index]?.content) + } + return "" +} + +function streamHeaders(res) { + res.writeHead(200, { + "content-type": "text/event-stream; charset=utf-8", + "cache-control": "no-cache", + connection: "keep-alive", + }) +} + +function writeSse(res, value) { + res.write(`data: ${JSON.stringify(value)}\n\n`) +} + +function streamText(res, content, sequence) { + const id = `chatcmpl-loop-goal-steering-${sequence}` + const created = Math.floor(Date.now() / 1000) + streamHeaders(res) + writeSse(res, { + id, + object: "chat.completion.chunk", + created, + model: "canary", + choices: [{ index: 0, delta: { role: "assistant", content }, finish_reason: null }], + }) + writeSse(res, { + id, + object: "chat.completion.chunk", + created, + model: "canary", + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + usage: { prompt_tokens: 30, completion_tokens: 4, total_tokens: 34 }, + }) + res.end("data: [DONE]\n\n") +} + +function startProvider() { + const stats = { + chatRequests: 0, + goalStarted: 0, + goalClosed: 0, + steeringStarted: 0, + paths: [], + } + const heldGoalResponses = new Set() + + const server = createServer(async (req, res) => { + const url = new URL(req.url ?? "/", "http://127.0.0.1") + stats.paths.push(`${req.method} ${url.pathname}`) + if (req.method === "GET" && url.pathname.endsWith("/models")) { + res.writeHead(200, { "content-type": "application/json" }) + res.end(JSON.stringify({ object: "list", data: [{ id: "canary", object: "model", owned_by: "canary" }] })) + return + } + if (req.method !== "POST" || !url.pathname.endsWith("/chat/completions")) { + res.writeHead(404, { "content-type": "application/json" }) + res.end(JSON.stringify({ error: { message: `unexpected endpoint: ${req.method} ${url.pathname}` } })) + return + } + + let raw = "" + for await (const chunk of req) raw += String(chunk) + const body = raw ? JSON.parse(raw) : {} + stats.chatRequests += 1 + const text = lastUserText(body) + + if (text.includes(USER_STEERING)) { + stats.steeringStarted += 1 + streamText(res, "USER_STEERING_ACK", stats.chatRequests) + return + } + + if (text.includes("EXPERIMENTAL GOAL MODE CONTINUATION") && text.includes(GOAL_OBJECTIVE)) { + stats.goalStarted += 1 + const id = `chatcmpl-loop-goal-held-${stats.chatRequests}` + const created = Math.floor(Date.now() / 1000) + heldGoalResponses.add(res) + res.once("close", () => { + if (heldGoalResponses.delete(res)) stats.goalClosed += 1 + }) + streamHeaders(res) + writeSse(res, { + id, + object: "chat.completion.chunk", + created, + model: "canary", + choices: [{ index: 0, delta: { role: "assistant", content: `GOAL_WORK_${stats.goalStarted}` }, finish_reason: null }], + }) + return + } + + streamText(res, "OK", stats.chatRequests) + }) + + return { + stats, + async listen() { + await new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(0, "127.0.0.1", resolve) + }) + const address = server.address() + if (!address || typeof address === "string") throw new Error("failed to start deterministic Goal steering provider") + return address.port + }, + async close() { + for (const response of heldGoalResponses) response.destroy() + await new Promise((resolve) => server.close(() => resolve())) + }, + } +} + +async function waitFor(predicate, description, diagnostics, timeoutMs = 45_000) { + const deadline = Date.now() + timeoutMs + while (Date.now() < deadline) { + if (await predicate()) return + await new Promise((resolve) => setTimeout(resolve, 50)) + } + throw new Error(`timed out waiting for ${description}\n${await diagnostics()}`) +} + +async function readGoalJob(stateFile) { + try { + const state = JSON.parse(await readFile(stateFile, "utf8")) + return (state.jobs || []).find((job) => String(job?.kind || "").toLowerCase() === "goal") || null + } catch (error) { + if (error?.code === "ENOENT") return null + throw error + } +} + +async function main() { + const workspace = await mkdtemp(path.join(os.tmpdir(), "opencode-loop-goal-steering-host-")) + const home = path.join(workspace, ".home") + const projectConfig = path.join(workspace, ".opencode") + const globalConfig = path.join(home, ".config", "opencode") + const pluginDir = path.join(projectConfig, "plugins") + const commandDir = path.join(projectConfig, "commands") + const agentDir = path.join(projectConfig, "agents") + const provider = startProvider() + const providerPort = await provider.listen() + let server + let serverLog = "" + let commandError = null + + await mkdir(pluginDir, { recursive: true }) + await mkdir(commandDir, { recursive: true }) + await mkdir(agentDir, { recursive: true }) + await seedConfigDependencies(projectConfig) + await seedConfigDependencies(globalConfig) + const pluginEntry = pathToFileURL(path.join(repoRoot, "src", "index.js")).href + await writeFile(path.join(pluginDir, "opencode-loop.js"), `export { default as OpenCodeLoopPlugin } from ${JSON.stringify(pluginEntry)}\n`) + await writeFile(path.join(commandDir, "loop-goal.md"), `---\ndescription: Start a Goal steering host canary\nagent: opencode-loop-local\n---\n\nOpenCode Loop Goal command handled. Reply exactly: OK.\n`) + await writeFile(path.join(agentDir, "opencode-loop-local.md"), `---\ndescription: Local Goal steering command acknowledgement\nmode: primary\npermission:\n "*": deny\n---\n\nReply exactly: OK\n`) + await writeFile(path.join(workspace, "opencode.json"), `${JSON.stringify({ + $schema: "https://opencode.ai/config.json", + model: "canary/canary", + small_model: "canary/canary", + provider: { + canary: { + npm: "@ai-sdk/openai-compatible", + name: "Deterministic Loop Goal Steering Canary", + options: { baseURL: `http://127.0.0.1:${providerPort}/v1`, apiKey: "canary-key" }, + models: { canary: { name: "Deterministic Loop Goal Steering Canary", limit: { context: 100000, output: 4096 } } }, + }, + }, + }, null, 2)}\n`) + + const env = { + ...process.env, + HOME: home, + USERPROFILE: home, + XDG_CONFIG_HOME: path.join(home, ".config"), + XDG_DATA_HOME: path.join(home, ".local", "share"), + XDG_CACHE_HOME: path.join(home, ".cache"), + OPENCODE_DISABLE_AUTOUPDATE: "true", + OPENCODE_DB: ":memory:", + OPENCODE_DISABLE_LSP_DOWNLOAD: "true", + CI: "true", + } + + try { + await runOpenCode(["debug", "config"], { cwd: workspace, env, timeoutMs: 60_000 }) + const port = await reservePort() + server = spawnOpenCode(["serve", "--hostname", "127.0.0.1", "--port", String(port)], { cwd: workspace, env }) + server.stdout?.on("data", (chunk) => { serverLog = appendLog(serverLog, chunk) }) + server.stderr?.on("data", (chunk) => { serverLog = appendLog(serverLog, chunk) }) + await waitForTcp(port, server, () => serverLog) + + const baseURL = `http://127.0.0.1:${port}` + const directoryQuery = `directory=${encodeURIComponent(workspace)}` + const api = async (pathname, init = {}) => { + const separator = pathname.includes("?") ? "&" : "?" + const response = await fetch(`${baseURL}${pathname}${separator}${directoryQuery}`, { + ...init, + headers: { "content-type": "application/json", ...(init.headers ?? {}) }, + signal: init.signal ?? AbortSignal.timeout(30_000), + }) + const text = await response.text() + if (!response.ok) throw new Error(`HTTP ${response.status}: ${text}`) + if (!text) return null + try { return JSON.parse(text) } catch { return text } + } + + const sessionsPayload = await api("/session", { method: "GET", signal: AbortSignal.timeout(45_000) }) + assert.ok(Array.isArray(sessionsPayload?.data ?? sessionsPayload), "GET /session bootstrap did not return an array") + const createdPayload = await api("/session", { method: "POST", body: JSON.stringify({ title: "opencode-loop Goal steering canary" }) }) + const session = createdPayload?.data ?? createdPayload + const sessionID = String(session?.id ?? "") + assert.ok(sessionID, `OpenCode did not create a session: ${JSON.stringify(createdPayload)}`) + + const command = api(`/session/${encodeURIComponent(sessionID)}/command`, { + method: "POST", + body: JSON.stringify({ agent: "build", model: "canary/canary", command: "loop-goal", arguments: GOAL_OBJECTIVE }), + signal: AbortSignal.timeout(90_000), + }).catch((error) => { + commandError = error + return null + }) + + const stateFile = path.join(workspace, ".opencode", "opencode-loop", `${sessionID}.json`) + const diagnostics = async () => { + let state = "missing" + try { state = await readFile(stateFile, "utf8") } catch {} + return `provider=${JSON.stringify(provider.stats)}\ncommandError=${String(commandError ?? "none")}\nstate=${state}\nserver log:\n${serverLog}` + } + + await waitFor( + async () => { + const goal = await readGoalJob(stateFile) + return provider.stats.goalStarted === 1 && goal?.action === GOAL_OBJECTIVE && goal?.goalStatus === "active" && goal?.paused === false + }, + "first autonomous Goal stream to become active", + diagnostics, + ) + + await api(`/session/${encodeURIComponent(sessionID)}/prompt_async`, { + method: "POST", + body: JSON.stringify({ + agent: "build", + model: { providerID: "canary", modelID: "canary" }, + parts: [{ type: "text", text: USER_STEERING }], + }), + signal: AbortSignal.timeout(20_000), + }) + + await waitFor(() => provider.stats.goalClosed >= 1, "active Goal provider stream to be cancelled for queued steering", diagnostics) + await waitFor(() => provider.stats.steeringStarted === 1, "queued user steering to reach the provider", diagnostics) + await waitFor( + async () => { + const goal = await readGoalJob(stateFile) + return goal?.action === GOAL_OBJECTIVE && goal?.goalStatus === "active" && goal?.paused === false + }, + "Goal state to remain active and unchanged during steering", + diagnostics, + ) + await waitFor(() => provider.stats.goalStarted >= 2, "Goal autonomous continuation to resume after steering", diagnostics) + + const goal = await readGoalJob(stateFile) + assert.equal(provider.stats.goalStarted, 2, `Goal should start once before and once after queued steering\n${await diagnostics()}`) + assert.equal(provider.stats.goalClosed, 1, `only the pre-steering Goal stream should be cancelled before cleanup\n${await diagnostics()}`) + assert.equal(provider.stats.steeringStarted, 1, `queued steering must execute exactly one foreground model turn\n${await diagnostics()}`) + assert.equal(goal?.action, GOAL_OBJECTIVE) + assert.equal(goal?.goalStatus, "active") + assert.equal(goal?.paused, false) + assert.equal(server.exitCode, null, `OpenCode server exited during Goal steering canary\n${await diagnostics()}`) + + console.log(JSON.stringify({ + ok: true, + platform: process.platform, + sessionID, + goalStarted: provider.stats.goalStarted, + goalClosed: provider.stats.goalClosed, + steeringStarted: provider.stats.steeringStarted, + goalStatus: goal.goalStatus, + paused: goal.paused, + objective: goal.action, + commandError: commandError ? String(commandError) : null, + }, null, 2)) + void command + } finally { + await stopProcess(server) + await provider.close().catch(() => undefined) + await rm(workspace, { recursive: true, force: true }).catch(() => undefined) + } +} + +main().catch((error) => { + console.error(error?.stack || error) + process.exitCode = 1 +}) diff --git a/src/index.js b/src/index.js index 0647c75..e40aeab 100644 --- a/src/index.js +++ b/src/index.js @@ -909,9 +909,10 @@ function loopOwnedUserMessageGuardActive(sessionID, messageID) { if (id && entry.messageIDs.has(id)) return true; if ((entry.pending || 0) > 0 && (entry.until || 0) >= now()) { + if (!id) + return true; entry.pending -= 1; - if (id) - entry.messageIDs.set(id, now() + LOOP_OWNED_USER_MESSAGE_RETENTION_MS); + entry.messageIDs.set(id, now() + LOOP_OWNED_USER_MESSAGE_RETENTION_MS); loopOwnedUserMessageGuards.set(sessionID, entry); return true; } @@ -3228,6 +3229,7 @@ exit=` + preflight.code + ` // src/source/runtime/goal-steering.js var DEFAULT_STEERING_SUPPRESSION_MS = 5 * 60000; +var DEFAULT_SEEN_USER_MESSAGE_MS = 10 * 60000; function requireFunction8(value, label) { if (typeof value !== "function") throw new TypeError(`createGoalSteeringRuntime requires ${label}`); @@ -3280,7 +3282,9 @@ function createGoalSteeringRuntime(options = {}) { const fireSdk2 = typeof options.fireSdk === "function" ? options.fireSdk : fireSdk; const now2 = typeof options.now === "function" ? options.now : now; const suppressionMs = Number.isFinite(Number(options.suppressionMs)) && Number(options.suppressionMs) > 0 ? Number(options.suppressionMs) : DEFAULT_STEERING_SUPPRESSION_MS; + const seenUserMessageMs = Number.isFinite(Number(options.seenUserMessageMs)) && Number(options.seenUserMessageMs) > 0 ? Number(options.seenUserMessageMs) : DEFAULT_SEEN_USER_MESSAGE_MS; const pendingSteering = new Map; + const seenUserMessages = new Map; function pendingForSession(sessionID) { const entry = pendingSteering.get(sessionID); if (!entry) @@ -3294,6 +3298,33 @@ function createGoalSteeringRuntime(options = {}) { function shouldSuppressIdle(sessionID) { return Boolean(pendingForSession(sessionID)); } + function seenKey(sessionID, messageID) { + return messageID ? `${sessionID}\x00${messageID}` : ""; + } + function alreadyHandled(sessionID, messageID) { + const key = seenKey(sessionID, messageID); + if (!key) + return false; + const expiresAt = seenUserMessages.get(key); + if (!expiresAt) + return false; + if (expiresAt <= now2()) { + seenUserMessages.delete(key); + return false; + } + return true; + } + function rememberHandled(sessionID, messageID) { + const key = seenKey(sessionID, messageID); + if (!key) + return; + const current = now2(); + seenUserMessages.set(key, current + seenUserMessageMs); + for (const [candidate, expiresAt] of seenUserMessages.entries()) { + if (expiresAt <= current) + seenUserMessages.delete(candidate); + } + } function observeAssistantMessage(event) { const assistant = assistantMessageFromEvent(event); if (!assistant) @@ -3308,52 +3339,67 @@ function createGoalSteeringRuntime(options = {}) { pendingSteering.delete(assistant.sessionID); return true; } - async function handleEvent(directory, client, event) { - observeAssistantMessage(event); - const user = userMessageFromEvent(event); - if (!user) + async function handleUserMessage(directory, client, user) { + const sessionID = typeof user?.sessionID === "string" ? user.sessionID : ""; + const messageID = typeof user?.messageID === "string" ? user.messageID : ""; + if (!sessionID) return; - if (isLoopOwnedUserMessage(user.sessionID, user.messageID)) { - return { handled: false, loopOwned: true, ...user }; + if (alreadyHandled(sessionID, messageID)) + return { handled: false, duplicate: true, sessionID, messageID }; + rememberHandled(sessionID, messageID); + if (isLoopOwnedUserMessage(sessionID, messageID)) { + return { handled: false, loopOwned: true, sessionID, messageID }; } - const state = await readState2(directory, user.sessionID); + const state = await readState2(directory, sessionID); const goals = activeGoalJobs(state); if (!goals.length) - return { handled: false, ...user }; - const active = getActiveRun(user.sessionID); + return { handled: false, sessionID, messageID }; + const active = getActiveRun(sessionID); const activeGoalIDs = new Set(goals.map((goal) => goal.id)); const canPreempt = active && activeGoalIDs.has(active.jobId) && isGoalJob(active.job) && typeof client?.session?.abort === "function"; let preempted = false; let abortError = ""; if (canPreempt) { - pendingSteering.set(user.sessionID, { - messageID: user.messageID, + pendingSteering.set(sessionID, { + messageID, goalID: active.jobId, armedAt: now2(), expiresAt: now2() + suppressionMs }); try { - await fireSdk2(client, "session.abort", client.session.abort.bind(client.session), { path: { id: user.sessionID }, body: {} }, { path: { sessionID: user.sessionID }, body: {} }, { sessionID: user.sessionID }); - clearActiveRun(user.sessionID); + await fireSdk2(client, "session.abort", client.session.abort.bind(client.session), { path: { id: sessionID }, body: {} }, { path: { sessionID }, body: {} }, { sessionID }); + clearActiveRun(sessionID); preempted = true; } catch (error) { - pendingSteering.delete(user.sessionID); + pendingSteering.delete(sessionID); abortError = error instanceof Error ? error.message : String(error); } } await appendLoopLog2(directory, "goal-user-steering", { - sessionID: user.sessionID, - messageID: user.messageID, + sessionID, + messageID, goals: goals.length, preempted, ...abortError ? { abortError } : {} }); - return { handled: true, preempted, ...user }; + return { handled: true, preempted, sessionID, messageID }; + } + async function handleEvent(directory, client, event) { + observeAssistantMessage(event); + const user = userMessageFromEvent(event); + if (!user) + return; + return await handleUserMessage(directory, client, user); } function clearSession(sessionID) { pendingSteering.delete(sessionID); + const prefix = `${sessionID}\x00`; + for (const key of seenUserMessages.keys()) + if (key.startsWith(prefix)) + seenUserMessages.delete(key); } return { + handleUserMessage, handleEvent, observeAssistantMessage, shouldSuppressIdle, @@ -3369,6 +3415,7 @@ var workspaceRuntime = createJobWorkspaceRuntime({ toast }); var { snapshotPaths } = workspaceRuntime; var goalPolicy = createGoalExecutionPolicy({ runShellCommand, dangerousShell, toast, appendLoopLog, now }); var schedulerRuntime; +var goalSteeringRuntime; var schedulerBridge = { rememberSession: (...args) => schedulerRuntime.rememberSession(...args), scheduleDueWork: (...args) => schedulerRuntime.scheduleDueWork(...args) @@ -3386,12 +3433,17 @@ var executorRuntime = createLoopExecutor({ var { clearActiveRun, finalizeActiveRun, - maybeRunDueJobs, + maybeRunDueJobs: runDueJobs, sessionIsIdle, updateSessionStatusFromEvent, noteLoopCompactionStarted, noteLoopCompactionCompleted } = executorRuntime; +async function maybeRunDueJobs(directory, client, sessionID, runOptions) { + if (goalSteeringRuntime?.shouldSuppressIdle(sessionID)) + return; + return await runDueJobs(directory, client, sessionID, runOptions); +} schedulerRuntime = createSchedulerRuntime({ sessionIsIdle, finalizeActiveRun, @@ -3401,7 +3453,7 @@ schedulerRuntime = createSchedulerRuntime({ errorMessage: sdkErrorMessage }); var { rememberSession, scheduleIdleWork, scheduleDueWork, stopWatchdog, cancelDueWork } = schedulerRuntime; -var goalSteeringRuntime = createGoalSteeringRuntime({ +goalSteeringRuntime = createGoalSteeringRuntime({ getActiveRun: executorRuntime.getActiveRun, clearActiveRun, isLoopOwnedUserMessage: loopOwnedUserMessageGuardActive, @@ -3552,6 +3604,14 @@ var OpenCodeLoopPlugin = async ({ client, directory }) => { "command.execute.before": async (input, output) => { await handleCommand(directory, client, input, undefined, undefined, output); }, + "chat.message": async (input) => { + const steering = await goalSteeringRuntime.handleUserMessage(directory, client, { + sessionID: input?.sessionID, + messageID: input?.messageID + }); + if (steering?.handled && steering.sessionID) + rememberSession(directory, client, steering.sessionID); + }, "tool.execute.before": async (input) => { markToolCallActive(input); }, diff --git a/src/source/legacy-v1.js b/src/source/legacy-v1.js index 37064ce..5501734 100644 --- a/src/source/legacy-v1.js +++ b/src/source/legacy-v1.js @@ -25,6 +25,7 @@ const { snapshotPaths } = workspaceRuntime const goalPolicy = createGoalExecutionPolicy({ runShellCommand, dangerousShell, toast, appendLoopLog, now }) let schedulerRuntime +let goalSteeringRuntime const schedulerBridge = { rememberSession: (...args) => schedulerRuntime.rememberSession(...args), scheduleDueWork: (...args) => schedulerRuntime.scheduleDueWork(...args), @@ -42,13 +43,18 @@ const executorRuntime = createLoopExecutor({ const { clearActiveRun, finalizeActiveRun, - maybeRunDueJobs, + maybeRunDueJobs: runDueJobs, sessionIsIdle, updateSessionStatusFromEvent, noteLoopCompactionStarted, noteLoopCompactionCompleted, } = executorRuntime +async function maybeRunDueJobs(directory, client, sessionID, runOptions) { + if (goalSteeringRuntime?.shouldSuppressIdle(sessionID)) return + return await runDueJobs(directory, client, sessionID, runOptions) +} + schedulerRuntime = createSchedulerRuntime({ sessionIsIdle, finalizeActiveRun, @@ -59,7 +65,7 @@ schedulerRuntime = createSchedulerRuntime({ }) const { rememberSession, scheduleIdleWork, scheduleDueWork, stopWatchdog, cancelDueWork } = schedulerRuntime -const goalSteeringRuntime = createGoalSteeringRuntime({ +goalSteeringRuntime = createGoalSteeringRuntime({ getActiveRun: executorRuntime.getActiveRun, clearActiveRun, isLoopOwnedUserMessage: loopOwnedUserMessageGuardActive, @@ -213,6 +219,13 @@ export const OpenCodeLoopPlugin = async ({ client, directory }) => { dispose: async () => { disposeRuntime(directory, client) }, tool: goalTools(directory), "command.execute.before": async (input, output) => { await handleCommand(directory, client, input, undefined, undefined, output) }, + "chat.message": async (input) => { + const steering = await goalSteeringRuntime.handleUserMessage(directory, client, { + sessionID: input?.sessionID, + messageID: input?.messageID, + }) + if (steering?.handled && steering.sessionID) rememberSession(directory, client, steering.sessionID) + }, "tool.execute.before": async (input) => { markToolCallActive(input) }, "tool.execute.after": async (input) => { markToolCallFinished(input) }, "experimental.session.compacting": async (input) => { await noteLoopCompactionStarted(directory, input?.sessionID) }, diff --git a/src/source/opencode/messages.js b/src/source/opencode/messages.js index 0b7845f..c27028e 100644 --- a/src/source/opencode/messages.js +++ b/src/source/opencode/messages.js @@ -25,8 +25,12 @@ export function loopOwnedUserMessageGuardActive(sessionID, messageID) { const id = typeof messageID === "string" ? messageID : "" if (id && entry.messageIDs.has(id)) return true if ((entry.pending || 0) > 0 && (entry.until || 0) >= now()) { + // chat.message can fire before OpenCode assigns the synthetic user message + // an ID. Treat that pre-dispatch hook as Loop-owned without consuming the + // pending guard; message.updated will later bind and retain the real ID. + if (!id) return true entry.pending -= 1 - if (id) entry.messageIDs.set(id, now() + LOOP_OWNED_USER_MESSAGE_RETENTION_MS) + entry.messageIDs.set(id, now() + LOOP_OWNED_USER_MESSAGE_RETENTION_MS) loopOwnedUserMessageGuards.set(sessionID, entry) return true } diff --git a/src/source/runtime/goal-steering.js b/src/source/runtime/goal-steering.js index dad7c35..b01e0f2 100644 --- a/src/source/runtime/goal-steering.js +++ b/src/source/runtime/goal-steering.js @@ -5,6 +5,7 @@ import { appendLoopLog as defaultAppendLoopLog } from "../core/process.js" import { fireSdk as defaultFireSdk } from "../opencode/host.js" const DEFAULT_STEERING_SUPPRESSION_MS = 5 * 60_000 +const DEFAULT_SEEN_USER_MESSAGE_MS = 10 * 60_000 function requireFunction(value, label) { if (typeof value !== "function") throw new TypeError(`createGoalSteeringRuntime requires ${label}`) @@ -57,8 +58,12 @@ export function createGoalSteeringRuntime(options = {}) { const suppressionMs = Number.isFinite(Number(options.suppressionMs)) && Number(options.suppressionMs) > 0 ? Number(options.suppressionMs) : DEFAULT_STEERING_SUPPRESSION_MS + const seenUserMessageMs = Number.isFinite(Number(options.seenUserMessageMs)) && Number(options.seenUserMessageMs) > 0 + ? Number(options.seenUserMessageMs) + : DEFAULT_SEEN_USER_MESSAGE_MS const pendingSteering = new Map() + const seenUserMessages = new Map() function pendingForSession(sessionID) { const entry = pendingSteering.get(sessionID) @@ -74,6 +79,32 @@ export function createGoalSteeringRuntime(options = {}) { return Boolean(pendingForSession(sessionID)) } + function seenKey(sessionID, messageID) { + return messageID ? `${sessionID}\u0000${messageID}` : "" + } + + function alreadyHandled(sessionID, messageID) { + const key = seenKey(sessionID, messageID) + if (!key) return false + const expiresAt = seenUserMessages.get(key) + if (!expiresAt) return false + if (expiresAt <= now()) { + seenUserMessages.delete(key) + return false + } + return true + } + + function rememberHandled(sessionID, messageID) { + const key = seenKey(sessionID, messageID) + if (!key) return + const current = now() + seenUserMessages.set(key, current + seenUserMessageMs) + for (const [candidate, expiresAt] of seenUserMessages.entries()) { + if (expiresAt <= current) seenUserMessages.delete(candidate) + } + } + function observeAssistantMessage(event) { const assistant = assistantMessageFromEvent(event) if (!assistant) return false @@ -86,27 +117,30 @@ export function createGoalSteeringRuntime(options = {}) { return true } - async function handleEvent(directory, client, event) { - observeAssistantMessage(event) - const user = userMessageFromEvent(event) - if (!user) return undefined - if (isLoopOwnedUserMessage(user.sessionID, user.messageID)) { - return { handled: false, loopOwned: true, ...user } + async function handleUserMessage(directory, client, user) { + const sessionID = typeof user?.sessionID === "string" ? user.sessionID : "" + const messageID = typeof user?.messageID === "string" ? user.messageID : "" + if (!sessionID) return undefined + if (alreadyHandled(sessionID, messageID)) return { handled: false, duplicate: true, sessionID, messageID } + rememberHandled(sessionID, messageID) + + if (isLoopOwnedUserMessage(sessionID, messageID)) { + return { handled: false, loopOwned: true, sessionID, messageID } } - const state = await readState(directory, user.sessionID) + const state = await readState(directory, sessionID) const goals = activeGoalJobs(state) - if (!goals.length) return { handled: false, ...user } + if (!goals.length) return { handled: false, sessionID, messageID } - const active = getActiveRun(user.sessionID) + const active = getActiveRun(sessionID) const activeGoalIDs = new Set(goals.map((goal) => goal.id)) const canPreempt = active && activeGoalIDs.has(active.jobId) && isGoalJob(active.job) && typeof client?.session?.abort === "function" let preempted = false let abortError = "" if (canPreempt) { - pendingSteering.set(user.sessionID, { - messageID: user.messageID, + pendingSteering.set(sessionID, { + messageID, goalID: active.jobId, armedAt: now(), expiresAt: now() + suppressionMs, @@ -116,34 +150,44 @@ export function createGoalSteeringRuntime(options = {}) { client, "session.abort", client.session.abort.bind(client.session), - { path: { id: user.sessionID }, body: {} }, - { path: { sessionID: user.sessionID }, body: {} }, - { sessionID: user.sessionID }, + { path: { id: sessionID }, body: {} }, + { path: { sessionID }, body: {} }, + { sessionID }, ) - clearActiveRun(user.sessionID) + clearActiveRun(sessionID) preempted = true } catch (error) { - pendingSteering.delete(user.sessionID) + pendingSteering.delete(sessionID) abortError = error instanceof Error ? error.message : String(error) } } await appendLoopLog(directory, "goal-user-steering", { - sessionID: user.sessionID, - messageID: user.messageID, + sessionID, + messageID, goals: goals.length, preempted, ...(abortError ? { abortError } : {}), }) - return { handled: true, preempted, ...user } + return { handled: true, preempted, sessionID, messageID } + } + + async function handleEvent(directory, client, event) { + observeAssistantMessage(event) + const user = userMessageFromEvent(event) + if (!user) return undefined + return await handleUserMessage(directory, client, user) } function clearSession(sessionID) { pendingSteering.delete(sessionID) + const prefix = `${sessionID}\u0000` + for (const key of seenUserMessages.keys()) if (key.startsWith(prefix)) seenUserMessages.delete(key) } return { + handleUserMessage, handleEvent, observeAssistantMessage, shouldSuppressIdle,