diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cea29fb0..6cae9f42 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -106,12 +106,15 @@ jobs: sudo apt-get install -y tmux npm install --global @opencode/cli@2.2.6 --no-audit --no-fund npm ci --prefix plugin/gateway-sidebar-v2 --ignore-scripts --no-audit --no-fund + npm ci --prefix plugin/gateway-core --ignore-scripts --no-audit --no-fund + npm --prefix plugin/gateway-core run build - name: Validate V2 sidebar and live terminal behavior run: | npm --prefix plugin/gateway-sidebar-v2 run typecheck npm --prefix plugin/gateway-sidebar-v2 test npm --prefix plugin/gateway-sidebar-v2 run test:live -- --opencode-bin "$(command -v opencode)" + python plugin/gateway-core/test/v2/execution_status_live_smoke.py --opencode-bin "$(command -v opencode)" --with-tui gateway-v2-permission: runs-on: ubuntu-latest @@ -145,6 +148,7 @@ jobs: python scripts/gateway_v2_permission_rejection_live_smoke.py python scripts/gateway_v2_tool_error_live_smoke.py python scripts/gateway_v2_shell_live_smoke.py + python plugin/gateway-core/test/v2/execution_status_live_smoke.py windows-bounded-subprocess: runs-on: windows-latest diff --git a/Makefile b/Makefile index 8cb77039..1f8edafc 100644 --- a/Makefile +++ b/Makefile @@ -13,7 +13,7 @@ OPENCODE_RESUME_E2E_VERSION := 1.18.18 help: ## Show available targets @awk 'BEGIN {FS = ":.*##"} /^[a-zA-Z0-9_-]+:.*##/ {printf "%-14s %s\n", $$1, $$2}' $(MAKEFILE_LIST) -PYTHON_TARGETS := data-lifecycle-help validate build-agents build-agents-check build-agents-prune release-index-update docs-automation-summary-update docs-automation-check pages-readiness-check release-note-validation-check release-note-quality-check plan-hygiene-check wave-linkage-check wave-handoff-summary wave-completion-update quality-fast quality-strict quality-off quality-status tasker-production-live-smoke rolling-decomposition-live gateway-status gateway-enable gateway-disable gateway-doctor gateway-secret-redaction-smoke gateway-v2-dependency-probe gateway-v2-dependency-profile-probe gateway-v2-auxiliary-redaction-probe gateway-v2-slice-smoke gateway-resume-redaction-e2e gateway-resume-redaction-e2e-prebuilt gateway-execution-status-live-smoke gateway-turn-watch gateway-turn-watch-webhook harness-wave2-task4-smoke notify-icons-generate notify-icons-select reservation-status task-lease-status selftest doctor doctor-json sqlite-doctor sqlite-doctor-json devtools-status hooks-install install-test install-test-full release-check release +PYTHON_TARGETS := data-lifecycle-help validate build-agents build-agents-check build-agents-prune release-index-update docs-automation-summary-update docs-automation-check pages-readiness-check release-note-validation-check release-note-quality-check plan-hygiene-check wave-linkage-check wave-handoff-summary wave-completion-update quality-fast quality-strict quality-off quality-status tasker-production-live-smoke rolling-decomposition-live gateway-status gateway-enable gateway-disable gateway-doctor gateway-secret-redaction-smoke gateway-v2-dependency-probe gateway-v2-dependency-profile-probe gateway-v2-auxiliary-redaction-probe gateway-v2-execution-status-live-smoke gateway-v2-slice-smoke gateway-resume-redaction-e2e gateway-resume-redaction-e2e-prebuilt gateway-execution-status-live-smoke gateway-turn-watch gateway-turn-watch-webhook harness-wave2-task4-smoke notify-icons-generate notify-icons-select reservation-status task-lease-status selftest doctor doctor-json sqlite-doctor sqlite-doctor-json devtools-status hooks-install install-test install-test-full release-check release $(PYTHON_TARGETS): python-check @@ -118,6 +118,11 @@ gateway-v2-auxiliary-redaction-probe: python-check ## Measure four V2 request ki npm --prefix plugin/gateway-core run build $(PYTHON) scripts/gateway_v2_auxiliary_redaction_probe.py --opencode-bin "$(OPENCODE_BIN)" +.PHONY: gateway-v2-execution-status-live-smoke +gateway-v2-execution-status-live-smoke: python-check ## Test the experimental V2 status writer and sidebar + npm --prefix plugin/gateway-core run build + $(PYTHON) plugin/gateway-core/test/v2/execution_status_live_smoke.py --opencode-bin "$(OPENCODE_BIN)" --with-tui + gateway-v2-slice-smoke: ## Test experimental V2 security adapters against localhost (not full gateway parity) npm --prefix plugin/gateway-core run build $(PYTHON) scripts/gateway_v2_context_live_smoke.py diff --git a/docs/plan/doing/opencode-v2-migration.md b/docs/plan/doing/opencode-v2-migration.md index 3074bb5d..7d18bdcb 100644 --- a/docs/plan/doing/opencode-v2-migration.md +++ b/docs/plan/doing/opencode-v2-migration.md @@ -39,14 +39,24 @@ The experimental V2 execution sidebar is a separate, unconfigured TUI plugin under `plugin/gateway-sidebar-v2`. It reads the existing private, bounded status format for the displayed session's location. Its unit tests and isolated tmux smoke cover status updates and plugin unload/reload on OpenCode 2.2.6. It does -not provide a V2 status writer or establish gateway security parity; leave it -unconfigured until the V2 gateway and its activation gates are complete. +not establish gateway security parity; leave it unconfigured until the V2 +gateway and its activation gates are complete. + +The separate, unconfigured `plugin/gateway-core/src/v2/execution-status.ts` +adapter translates V2 tool and session events into the existing bounded status +ledger. It checks each session's actual location against its own plugin +location and records generic milestones rather than tool output. Deletion and +move cleanup are best-effort for sessions observed during that plugin lifetime; +old entries remain bounded. A private-server smoke exercises a real V2 shell, +session deletion and, on 2.2.6, the rendered sidebar. This adapter is not a V2 +gateway entrypoint and does not enforce or qualify security policy. ```sh npm ci --prefix plugin/gateway-sidebar-v2 --ignore-scripts npm --prefix plugin/gateway-sidebar-v2 run typecheck npm --prefix plugin/gateway-sidebar-v2 test npm --prefix plugin/gateway-sidebar-v2 run test:live -- --opencode-bin /path/to/opencode-2.2.6 +make gateway-v2-execution-status-live-smoke OPENCODE_BIN=/path/to/opencode-2.2.6 ``` The separate [native dependency-boundary probe](../../specs/opencode-v2-dependency-boundary-contract.md) diff --git a/plugin/gateway-core/dist/v2/execution-status.d.ts b/plugin/gateway-core/dist/v2/execution-status.d.ts new file mode 100644 index 00000000..d72a6f13 --- /dev/null +++ b/plugin/gateway-core/dist/v2/execution-status.d.ts @@ -0,0 +1,42 @@ +import type { Registration } from "./types.js"; +type ToolEvent = { + tool: string; + sessionID: string; + input: unknown; + status?: string; + result?: unknown; +}; +interface StatusToolHooks { + hook(name: "execute.before" | "execute.after", callback: (event: ToolEvent) => Promise): Promise; +} +interface StatusContext { + app: { + version: string; + }; + location: { + directory: string; + }; + session: { + get(input: { + sessionID: string; + }): Promise<{ + location?: { + directory?: unknown; + }; + }>; + }; + tool: StatusToolHooks; + event: { + subscribe(options: { + types: string[]; + signal: AbortSignal; + }): AsyncIterable; + }; +} +/** Unconfigured V2 adapter; status reporting is never an enforcement boundary. */ +export declare function registerV2ExecutionStatus(context: StatusContext, options: { + enabled: boolean; + maxSessions: number; + maxLabelChars: number; +}): Promise<() => Promise>; +export {}; diff --git a/plugin/gateway-core/dist/v2/execution-status.js b/plugin/gateway-core/dist/v2/execution-status.js new file mode 100644 index 00000000..4bf78955 --- /dev/null +++ b/plugin/gateway-core/dist/v2/execution-status.js @@ -0,0 +1,181 @@ +import { isAbsolute, resolve } from "node:path"; +import { createExecutionStatusHook } from "../hooks/execution-status/index.js"; +const CONTROL_CHARACTER = /[\u0000-\u001f\u007f]/; +const TOOLS = new Map([ + ["shell", "bash"], + ["edit", "edit"], + ["write", "write"], + ["patch", "apply_patch"], + ["subagent", "task"], +]); +const MAX_PENDING = 256; +const MAX_SESSIONS = 256; +const MAX_LOOKUPS = 8; +const LOOKUP_TIMEOUT_MS = 1000; +const SUPPORTED_VERSIONS = ["2.2.5", "2.2.6"]; +function isRecord(value) { + return !!value && typeof value === "object" && !Array.isArray(value); +} +function validSessionID(value) { + return typeof value === "string" && value.length > 0 && value.length <= 160 && + value === value.trim() && !CONTROL_CHARACTER.test(value); +} +/** Unconfigured V2 adapter; status reporting is never an enforcement boundary. */ +export async function registerV2ExecutionStatus(context, options) { + if (!options.enabled) + return async () => { }; + if (!SUPPORTED_VERSIONS.includes(context.app.version) || + typeof context.location.directory !== "string" || !isAbsolute(context.location.directory) || + !Number.isInteger(options.maxSessions) || options.maxSessions < 1 || options.maxSessions > 64 || + !Number.isInteger(options.maxLabelChars) || options.maxLabelChars < 24 || options.maxLabelChars > 160) { + throw new Error("Invalid V2 execution-status configuration"); + } + const directory = resolve(context.location.directory); + const status = createExecutionStatusHook({ directory, ...options }); + const known = new Set(); + const tails = new Map(); + const abort = new AbortController(); + const registrations = []; + let pending = 0; + let activeLookups = 0; + let disposed = false; + const payload = (id) => ({ input: { sessionID: id } }); + const remember = (id) => { + known.delete(id); + known.add(id); + if (known.size > MAX_SESSIONS) { + const first = known.values().next().value; + if (first !== undefined) + known.delete(first); + } + }; + const enqueue = (id, work) => { + if (disposed || !validSessionID(id) || pending >= MAX_PENDING) + return Promise.resolve(); + const previous = tails.get(id) ?? Promise.resolve(); + pending += 1; + const current = previous.then(async () => { + if (!disposed) + await work(id); + }).catch(() => { }); + tails.set(id, current); + return current.finally(() => { + pending -= 1; + if (tails.get(id) === current) + tails.delete(id); + }); + }; + const belongsHere = async (id) => { + if (disposed || activeLookups >= MAX_LOOKUPS) + return false; + activeLookups += 1; + const lookup = Promise.resolve().then(() => context.session.get({ sessionID: id })) + .finally(() => { activeLookups -= 1; }); + let timer; + try { + const timeout = new Promise((done) => { + timer = setTimeout(() => done(undefined), LOOKUP_TIMEOUT_MS); + }); + const session = await Promise.race([lookup, timeout]); + if (disposed || !session || !isRecord(session.location)) + return false; + const location = session.location.directory; + if (typeof location !== "string" || !isAbsolute(location)) + return false; + if (resolve(location) !== directory) { + if (known.delete(id)) + await status.event("session.deleted", payload(id)); + return false; + } + if (!known.has(id)) + await status.event("session.created", payload(id)); + if (disposed) + return false; + remember(id); + return true; + } + catch { + return false; + } + finally { + if (timer) + clearTimeout(timer); + } + }; + const toolEvent = (phase, event) => { + try { + const name = typeof event.tool === "string" ? TOOLS.get(event.tool.toLowerCase()) : undefined; + if (!name) + return Promise.resolve(); + const args = isRecord(event.input) && typeof event.input.command === "string" + ? { command: event.input.command } + : {}; + const metadata = isRecord(event.result) && isRecord(event.result.metadata) && + typeof event.result.metadata.exit === "number" && Number.isFinite(event.result.metadata.exit) + ? { exit: event.result.metadata.exit } + : {}; + const input = { tool: name, sessionID: event.sessionID, args }; + const kind = phase === "execute.before" + ? "tool.execute.before" + : event.status === "error" ? "tool.execute.before.error" : "tool.execute.after"; + if (phase === "execute.after" && event.status !== "completed" && event.status !== "error") + return Promise.resolve(); + return enqueue(event.sessionID, async (id) => { + if (!(await belongsHere(id)) || disposed) + return; + await status.event(kind, { input, output: { metadata } }); + }); + } + catch { + return Promise.resolve(); + } + }; + const sessionEvent = async (event) => { + if (!isRecord(event) || !isRecord(event.data)) + return; + const type = event.type; + if (type !== "session.created" && type !== "session.moved" && type !== "session.deleted") + return; + await enqueue(event.data.sessionID, async (id) => { + if (type === "session.deleted") { + if (known.delete(id)) + await status.event("session.deleted", payload(id)); + return; + } + await belongsHere(id); + }); + }; + try { + registrations.push(await context.tool.hook("execute.before", (event) => toolEvent("execute.before", event))); + registrations.push(await context.tool.hook("execute.after", (event) => toolEvent("execute.after", event))); + const stream = context.event.subscribe({ types: ["session.created", "session.moved", "session.deleted"], signal: abort.signal }); + void (async () => { + try { + for await (const event of stream) { + if (disposed) + break; + try { + await sessionEvent(event); + } + catch { + // A malformed event must not end the subscription. + } + } + } + catch { + // Status reporting is best-effort; tool execution must continue. + } + })(); + } + catch { + disposed = true; + abort.abort(); + await Promise.allSettled(registrations.map(async (registration) => registration.dispose())); + throw new Error("Unable to register V2 execution status"); + } + return async () => { + disposed = true; + abort.abort(); + await Promise.allSettled(registrations.map(async (registration) => registration.dispose())); + }; +} diff --git a/plugin/gateway-core/src/v2/execution-status.ts b/plugin/gateway-core/src/v2/execution-status.ts new file mode 100644 index 00000000..46a7b5c8 --- /dev/null +++ b/plugin/gateway-core/src/v2/execution-status.ts @@ -0,0 +1,187 @@ +import { isAbsolute, resolve } from "node:path" + +import { createExecutionStatusHook } from "../hooks/execution-status/index.js" +import type { Registration } from "./types.js" + +type ToolEvent = { + tool: string + sessionID: string + input: unknown + status?: string + result?: unknown +} + +interface StatusToolHooks { + hook(name: "execute.before" | "execute.after", callback: (event: ToolEvent) => Promise): Promise +} + +interface StatusContext { + app: { version: string } + location: { directory: string } + session: { get(input: { sessionID: string }): Promise<{ location?: { directory?: unknown } }> } + tool: StatusToolHooks + event: { subscribe(options: { types: string[]; signal: AbortSignal }): AsyncIterable } +} + +const CONTROL_CHARACTER = /[\u0000-\u001f\u007f]/ +const TOOLS = new Map([ + ["shell", "bash"], + ["edit", "edit"], + ["write", "write"], + ["patch", "apply_patch"], + ["subagent", "task"], +]) +const MAX_PENDING = 256 +const MAX_SESSIONS = 256 +const MAX_LOOKUPS = 8 +const LOOKUP_TIMEOUT_MS = 1000 +const SUPPORTED_VERSIONS = ["2.2.5", "2.2.6"] + +function isRecord(value: unknown): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value) +} + +function validSessionID(value: unknown): value is string { + return typeof value === "string" && value.length > 0 && value.length <= 160 && + value === value.trim() && !CONTROL_CHARACTER.test(value) +} + +/** Unconfigured V2 adapter; status reporting is never an enforcement boundary. */ +export async function registerV2ExecutionStatus( + context: StatusContext, + options: { enabled: boolean; maxSessions: number; maxLabelChars: number }, +): Promise<() => Promise> { + if (!options.enabled) return async () => {} + if (!SUPPORTED_VERSIONS.includes(context.app.version) || + typeof context.location.directory !== "string" || !isAbsolute(context.location.directory) || + !Number.isInteger(options.maxSessions) || options.maxSessions < 1 || options.maxSessions > 64 || + !Number.isInteger(options.maxLabelChars) || options.maxLabelChars < 24 || options.maxLabelChars > 160) { + throw new Error("Invalid V2 execution-status configuration") + } + const directory = resolve(context.location.directory) + const status = createExecutionStatusHook({ directory, ...options }) + const known = new Set() + const tails = new Map>() + const abort = new AbortController() + const registrations: Registration[] = [] + let pending = 0 + let activeLookups = 0 + let disposed = false + + const payload = (id: string): { input: { sessionID: string } } => ({ input: { sessionID: id } }) + const remember = (id: string): void => { + known.delete(id) + known.add(id) + if (known.size > MAX_SESSIONS) { + const first = known.values().next().value + if (first !== undefined) known.delete(first) + } + } + const enqueue = (id: unknown, work: (id: string) => Promise): Promise => { + if (disposed || !validSessionID(id) || pending >= MAX_PENDING) return Promise.resolve() + const previous = tails.get(id) ?? Promise.resolve() + pending += 1 + const current = previous.then(async () => { + if (!disposed) await work(id) + }).catch(() => {}) + tails.set(id, current) + return current.finally(() => { + pending -= 1 + if (tails.get(id) === current) tails.delete(id) + }) + } + const belongsHere = async (id: string): Promise => { + if (disposed || activeLookups >= MAX_LOOKUPS) return false + activeLookups += 1 + const lookup = Promise.resolve().then(() => context.session.get({ sessionID: id })) + .finally(() => { activeLookups -= 1 }) + let timer: ReturnType | undefined + try { + const timeout = new Promise((done) => { + timer = setTimeout(() => done(undefined), LOOKUP_TIMEOUT_MS) + }) + const session = await Promise.race([lookup, timeout]) + if (disposed || !session || !isRecord(session.location)) return false + const location = session.location.directory + if (typeof location !== "string" || !isAbsolute(location)) return false + if (resolve(location) !== directory) { + if (known.delete(id)) await status.event("session.deleted", payload(id)) + return false + } + if (!known.has(id)) await status.event("session.created", payload(id)) + if (disposed) return false + remember(id) + return true + } catch { + return false + } finally { + if (timer) clearTimeout(timer) + } + } + const toolEvent = (phase: "execute.before" | "execute.after", event: ToolEvent): Promise => { + try { + const name = typeof event.tool === "string" ? TOOLS.get(event.tool.toLowerCase()) : undefined + if (!name) return Promise.resolve() + const args = isRecord(event.input) && typeof event.input.command === "string" + ? { command: event.input.command } + : {} + const metadata = isRecord(event.result) && isRecord(event.result.metadata) && + typeof event.result.metadata.exit === "number" && Number.isFinite(event.result.metadata.exit) + ? { exit: event.result.metadata.exit } + : {} + const input = { tool: name, sessionID: event.sessionID, args } + const kind = phase === "execute.before" + ? "tool.execute.before" + : event.status === "error" ? "tool.execute.before.error" : "tool.execute.after" + if (phase === "execute.after" && event.status !== "completed" && event.status !== "error") return Promise.resolve() + return enqueue(event.sessionID, async (id) => { + if (!(await belongsHere(id)) || disposed) return + await status.event(kind, { input, output: { metadata } }) + }) + } catch { + return Promise.resolve() + } + } + const sessionEvent = async (event: unknown): Promise => { + if (!isRecord(event) || !isRecord(event.data)) return + const type = event.type + if (type !== "session.created" && type !== "session.moved" && type !== "session.deleted") return + await enqueue(event.data.sessionID, async (id) => { + if (type === "session.deleted") { + if (known.delete(id)) await status.event("session.deleted", payload(id)) + return + } + await belongsHere(id) + }) + } + + try { + registrations.push(await context.tool.hook("execute.before", (event) => toolEvent("execute.before", event))) + registrations.push(await context.tool.hook("execute.after", (event) => toolEvent("execute.after", event))) + const stream = context.event.subscribe({ types: ["session.created", "session.moved", "session.deleted"], signal: abort.signal }) + void (async (): Promise => { + try { + for await (const event of stream) { + if (disposed) break + try { + await sessionEvent(event) + } catch { + // A malformed event must not end the subscription. + } + } + } catch { + // Status reporting is best-effort; tool execution must continue. + } + })() + } catch { + disposed = true + abort.abort() + await Promise.allSettled(registrations.map(async (registration) => registration.dispose())) + throw new Error("Unable to register V2 execution status") + } + return async () => { + disposed = true + abort.abort() + await Promise.allSettled(registrations.map(async (registration) => registration.dispose())) + } +} diff --git a/plugin/gateway-core/test/v2/execution-status.test.mjs b/plugin/gateway-core/test/v2/execution-status.test.mjs new file mode 100644 index 00000000..fa6e8e78 --- /dev/null +++ b/plugin/gateway-core/test/v2/execution-status.test.mjs @@ -0,0 +1,194 @@ +import assert from "node:assert/strict" +import { existsSync, mkdtempSync, mkdirSync, readFileSync, rmSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import test from "node:test" + +import { loadGatewayState } from "../../dist/state/storage.js" +import { registerV2ExecutionStatus } from "../../dist/v2/execution-status.js" + +const options = { enabled: true, maxSessions: 8, maxLabelChars: 80 } + +function stream() { + const queue = [] + let waiting + let closed = false + return { + subscribe({ signal }) { + signal.addEventListener("abort", () => { + closed = true + if (waiting) { waiting({ done: true }); waiting = undefined } + }, { once: true }) + return { [Symbol.asyncIterator]() { return { + next() { + if (queue.length) return Promise.resolve({ value: queue.shift(), done: false }) + if (closed) return Promise.resolve({ done: true }) + return new Promise((resolve) => { waiting = resolve }) + }, + } } } + }, + emit(value) { + if (closed) return + if (waiting) { waiting({ value, done: false }); waiting = undefined } + else queue.push(value) + }, + get closed() { return closed }, + } +} + +function fixture() { + const root = mkdtempSync(join(tmpdir(), "v2-execution-status-")) + const directory = join(root, "project") + const other = join(root, "other") + mkdirSync(directory) + mkdirSync(other) + const sessions = new Map() + const hooks = new Map() + const disposed = [] + const events = stream() + const context = { + app: { version: "2.2.6" }, + location: { directory }, + session: { async get({ sessionID }) { + if (!sessions.has(sessionID)) throw new Error("missing session") + return { location: { directory: sessions.get(sessionID) } } + } }, + tool: { async hook(name, callback) { + hooks.set(name, callback) + return { async dispose() { disposed.push(name) } } + } }, + event: events, + } + return { root, directory, other, sessions, hooks, disposed, events, context, + state: (id) => loadGatewayState(directory)?.executionStatus?.sessions[id], + close: () => rmSync(root, { recursive: true, force: true }), + } +} + +async function until(predicate) { + for (let attempt = 0; attempt < 100; attempt += 1) { + if (predicate()) return + await new Promise((resolve) => setTimeout(resolve, 10)) + } + throw new Error("status event did not settle") +} + +test("V2 tool events produce bounded milestones without recording command output", async () => { + const f = fixture() + try { + f.sessions.set("ses-one", f.directory) + const dispose = await registerV2ExecutionStatus(f.context, options) + const event = { tool: "shell", sessionID: "ses-one", input: { command: "make validate SYNTHETIC_SECRET", secret: "SYNTHETIC_SECRET" } } + await f.hooks.get("execute.before")(event) + assert.equal(f.state("ses-one").next, "Run validation") + await f.hooks.get("execute.after")({ ...event, status: "completed", result: { metadata: { exit: 0, secret: "SYNTHETIC_SECRET" } } }) + assert.equal(f.state("ses-one").last, "Validation passed") + assert.equal(f.state("ses-one").next, "Review changes") + assert.doesNotMatch(readFileSync(join(f.directory, ".opencode", "gateway-core.state.json"), "utf8"), /SYNTHETIC_SECRET/) + await f.hooks.get("execute.after")({ ...event, status: "completed", result: { metadata: { exit: 1 } } }) + assert.equal(f.state("ses-one").last, "Validation failed") + await f.hooks.get("execute.after")({ ...event, status: "completed", result: {} }) + assert.equal(f.state("ses-one").last, "Validation completed") + await f.hooks.get("execute.after")({ ...event, status: "error" }) + assert.equal(f.state("ses-one").next, "Fix validation") + await dispose() + assert.deepEqual(f.disposed.sort(), ["execute.after", "execute.before"]) + assert.equal(f.events.closed, true) + } finally { f.close() } +}) + +test("V2 status stays in its own location and removes known moved or deleted sessions", async () => { + const f = fixture() + try { + f.sessions.set("ses-one", f.directory) + f.sessions.set("ses-other", f.other) + const dispose = await registerV2ExecutionStatus(f.context, options) + f.events.emit({ type: "session.created", data: { sessionID: "ses-one" } }) + await until(() => f.state("ses-one")?.last === "Session ready") + await f.hooks.get("execute.before")({ tool: "write", sessionID: "ses-other", input: {} }) + assert.equal(existsSync(join(f.other, ".opencode")), false) + assert.equal(f.state("ses-other"), undefined) + f.sessions.set("ses-one", f.other) + f.events.emit({ type: "session.moved", data: { sessionID: "ses-one" } }) + await until(() => f.state("ses-one") === undefined) + f.sessions.set("ses-one", f.directory) + f.events.emit({ type: "session.created", data: { sessionID: "ses-one" } }) + await until(() => !!f.state("ses-one")) + f.events.emit({ type: "session.deleted", data: { sessionID: "ses-one" } }) + await until(() => f.state("ses-one") === undefined) + assert.equal(existsSync(join(f.other, ".opencode")), false) + await dispose() + } finally { f.close() } +}) + +test("V2 status does not write after disposal or for malformed events", async () => { + const f = fixture() + try { + let finishLookup + f.context.session.get = () => new Promise((resolve) => { finishLookup = resolve }) + const dispose = await registerV2ExecutionStatus(f.context, options) + const pending = f.hooks.get("execute.before")({ tool: "write", sessionID: "ses-one", input: {} }) + await until(() => !!finishLookup) + await dispose() + finishLookup({ location: { directory: f.directory } }) + await pending + await f.hooks.get("execute.before")({ tool: "write", sessionID: "ses-two", input: {} }) + f.events.emit({ type: "session.created", data: { sessionID: "ses-two" } }) + assert.equal(existsSync(join(f.directory, ".opencode")), false) + } finally { f.close() } +}) + +test("V2 status orders concurrent events and survives a malformed session event", async () => { + const f = fixture() + try { + let finishLookup + f.context.session.get = () => new Promise((resolve) => { finishLookup = resolve }) + const dispose = await registerV2ExecutionStatus(f.context, options) + f.events.emit(new Proxy({}, { get() { throw new Error("private detail") } })) + const event = { tool: "shell", sessionID: "ses-one", input: { command: "make validate" } } + const before = f.hooks.get("execute.before")(event) + const after = f.hooks.get("execute.after")({ ...event, status: "completed", result: { metadata: { exit: 0 } } }) + await until(() => !!finishLookup) + f.context.session.get = async () => ({ location: { directory: f.directory } }) + finishLookup({ location: { directory: f.directory } }) + await Promise.all([before, after]) + assert.equal(f.state("ses-one").last, "Validation passed") + f.sessions.set("ses-two", f.directory) + f.events.emit({ type: "session.created", data: { sessionID: "ses-two" } }) + await until(() => f.state("ses-two")?.last === "Session ready") + await dispose() + } finally { f.close() } +}) + +test("V2 status cleans up partial registration and validates configuration", async () => { + const f = fixture() + try { + let disposed = false + f.context.tool.hook = async (name) => { + if (name === "execute.after") throw new Error("synthetic private detail") + return { async dispose() { disposed = true } } + } + await assert.rejects(registerV2ExecutionStatus(f.context, options), /^Error: Unable to register V2 execution status$/) + assert.equal(disposed, true) + await assert.rejects(registerV2ExecutionStatus(f.context, { ...options, maxSessions: 0 }), /Invalid V2 execution-status configuration/) + f.context.app.version = "2.3.0" + await assert.rejects(registerV2ExecutionStatus(f.context, options), /Invalid V2 execution-status configuration/) + const disabled = await registerV2ExecutionStatus(f.context, { ...options, enabled: false }) + await disabled() + } finally { f.close() } +}) + +test("V2 status bounds a stalled session lookup and ignores malformed tool events", async () => { + const f = fixture() + try { + f.context.session.get = () => new Promise(() => {}) + const dispose = await registerV2ExecutionStatus(f.context, options) + const start = Date.now() + await f.hooks.get("execute.before")({ tool: "write", sessionID: "ses-stalled", input: {} }) + assert.ok(Date.now() - start < 4000) + await f.hooks.get("execute.before")(new Proxy({}, { get() { throw new Error("private detail") } })) + await f.hooks.get("execute.before")({ tool: "constructor", sessionID: "ses-unknown", input: {} }) + assert.equal(existsSync(join(f.directory, ".opencode")), false) + await dispose() + } finally { f.close() } +}) diff --git a/plugin/gateway-core/test/v2/execution_status_live_smoke.py b/plugin/gateway-core/test/v2/execution_status_live_smoke.py new file mode 100644 index 00000000..27953ce5 --- /dev/null +++ b/plugin/gateway-core/test/v2/execution_status_live_smoke.py @@ -0,0 +1,473 @@ +"""Exercise the unconfigured V2 status writer with a private server and provider.""" + +from __future__ import annotations + +import argparse +import base64 +import json +import os +import secrets +import shutil +import signal +import socket +import stat +import subprocess +import tempfile +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.error import URLError +from urllib.request import ProxyHandler, Request, build_opener + +ROOT = Path(__file__).resolve().parents[4] +CANARY = "V2StatusOutputSynthetic_123456789" +MAX_REQUEST = 1024 * 1024 + + +class Provider(ThreadingHTTPServer): + daemon_threads = True + block_on_close = False + + def __init__(self) -> None: + super().__init__(("127.0.0.1", 0), Handler) + self.lock = threading.Lock() + self.requests = 0 + self.offered = False + self.continued = False + + +class Handler(BaseHTTPRequestHandler): + def setup(self) -> None: + super().setup() + self.connection.settimeout(3) + + def log_message(self, _format: str, *args: object) -> None: + return + + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length") or "0") + if length <= 0 or length > MAX_REQUEST: + self.send_error(400) + return + payload = json.loads(self.rfile.read(length)) + messages = payload.get("messages", []) + offered = any( + item.get("function", {}).get("name") == "shell" + for item in payload.get("tools", []) + ) + finished = any(item.get("role") == "tool" for item in messages) + provider: Provider = self.server # type: ignore[assignment] + with provider.lock: + provider.requests += 1 + provider.offered |= offered + provider.continued |= finished + if provider.requests > 8: + self.send_error(429) + return + if offered and not finished: + delta = { + "role": "assistant", + "tool_calls": [ + { + "index": 0, + "id": "call_status_probe", + "type": "function", + "function": { + "name": "shell", + "arguments": json.dumps({"command": "make validate"}), + }, + } + ], + } + reason = "tool_calls" + else: + delta, reason = ( + {"role": "assistant", "content": "STATUS_PROBE_DONE"}, + "stop", + ) + chunks = [ + { + "id": "status-probe", + "object": "chat.completion.chunk", + "created": 1, + "model": "mock", + "choices": [{"index": 0, "delta": delta, "finish_reason": None}], + }, + { + "id": "status-probe", + "object": "chat.completion.chunk", + "created": 1, + "model": "mock", + "choices": [{"index": 0, "delta": {}, "finish_reason": reason}], + }, + ] + body = ( + "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks) + + "data: [DONE]\n\n" + ).encode() + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--opencode-bin", default="opencode") + parser.add_argument("--with-tui", action="store_true") + args = parser.parse_args() + found = shutil.which(args.opencode_bin) + if not found or (args.with_tui and not shutil.which("tmux")): + raise RuntimeError("OpenCode and, for TUI testing, tmux are required") + binary = str(Path(found).resolve()) + adapter = ROOT / "plugin/gateway-core/dist/v2/execution-status.js" + sidebar = ROOT / "plugin/gateway-sidebar-v2" + if not adapter.is_file(): + raise RuntimeError("build gateway-core before running the smoke") + provider = Provider() + thread = threading.Thread(target=provider.serve_forever, daemon=True) + thread.start() + try: + with tempfile.TemporaryDirectory(prefix="gateway-v2-status-live-") as raw: + root = Path(raw) + home, project, fake_bin = root / "home", root / "project", root / "bin" + config, plugin = home / ".config/opencode", project / "status-probe" + config.mkdir(parents=True) + plugin.mkdir(parents=True) + fake_bin.mkdir() + make = fake_bin / "make" + make.write_text( + f"#!/bin/sh\n[ \"$#\" -eq 1 ] && [ \"$1\" = validate ] || exit 77\nprintf '%s\\n' '{CANARY}'\n" + ) + make.chmod(0o700) + (plugin / "index.ts").write_text( + 'import { Plugin } from "@opencode/plugin"\n' + f"import {{ registerV2ExecutionStatus }} from {json.dumps(adapter.as_uri())}\n" + 'export default Plugin.define({ id: "v2-status-live-probe", async setup(ctx) {\n' + " return await registerV2ExecutionStatus(ctx, { enabled: true, maxSessions: 8, maxLabelChars: 80 })\n" + "} })\n" + ) + (project / "opencode.json").write_text( + json.dumps( + { + "plugins": [str(plugin)], + "model": "capture/mock", + "permissions": [ + {"action": "*", "resource": "*", "effect": "allow"} + ], + "providers": { + "capture": { + "name": "Local test", + "package": "@opencode/ai/providers/openai-compatible", + "settings": { + "baseURL": f"http://127.0.0.1:{provider.server_port}/v1", + "apiKey": "local-test-only", + }, + "models": { + "mock": { + "name": "mock", + "capabilities": {"tools": True}, + } + }, + } + }, + } + ) + ) + if args.with_tui: + (config / "cli.json").write_text( + json.dumps( + {"plugins": [str(sidebar)], "session": {"sidebar": "auto"}} + ) + ) + password = secrets.token_urlsafe(24) + env = { + key: os.environ[key] + for key in ("USER", "LOGNAME", "LANG", "LC_ALL") + if key in os.environ + } + env.update( + { + "HOME": str(home), + "PATH": f"{fake_bin}:/usr/bin:/bin:/opt/homebrew/bin", + "SHELL": "/bin/sh", + "TMPDIR": str(root), + "XDG_CONFIG_HOME": str(home / ".config"), + "XDG_DATA_HOME": str(home / ".local/share"), + "XDG_CACHE_HOME": str(home / ".cache"), + "XDG_STATE_HOME": str(home / ".local/state"), + "CI": "1", + "TERM": "xterm-256color", + "OPENCODE_PASSWORD": password, + "OPENCODE_DISABLE_MODELS_FETCH": "1", + "OPENCODE_DISABLE_DEFAULT_PLUGINS": "1", + "OPENCODE_DISABLE_LSP_DOWNLOAD": "1", + } + ) + version = subprocess.run( + [binary, "--version"], + cwd=project, + env=env, + capture_output=True, + text=True, + timeout=10, + check=True, + ).stdout.strip() + if version not in {"opencode v2.2.5", "opencode v2.2.6"} or ( + args.with_tui and version != "opencode v2.2.6" + ): + raise RuntimeError("unsupported OpenCode version for status smoke") + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + base = f"http://127.0.0.1:{port}" + auth = base64.b64encode(f"opencode:{password}".encode()).decode() + opener = build_opener(ProxyHandler({})) + + def request( + method: str, path: str, value: object = None, timeout: int = 3 + ) -> object: + body = json.dumps(value).encode() if value is not None else None + req = Request( + base + path, + data=body, + method=method, + headers={ + "Authorization": "Basic " + auth, + "Content-Type": "application/json", + }, + ) + with opener.open(req, timeout=timeout) as response: + data = response.read(MAX_REQUEST + 1) + if len(data) > MAX_REQUEST: + raise RuntimeError("private server response exceeded bound") + return json.loads(data) if data else None + + with (root / "server.log").open("wb") as log: + server = subprocess.Popen( + [binary, "serve", "--hostname", "127.0.0.1", "--port", str(port)], + cwd=project, + env=env, + stdin=subprocess.DEVNULL, + stdout=log, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + tmux_target: str | None = None + try: + for _ in range(100): + if server.poll() is not None: + raise RuntimeError("private server exited") + try: + if request("GET", "/api/info").get("pid") == server.pid: # type: ignore[union-attr] + break + except (OSError, URLError): + time.sleep(0.1) + else: + raise RuntimeError("private server did not become ready") + + def mutate( + method: str, path: str, value: object = None, timeout: int = 3 + ) -> object: + if ( + server.poll() is not None + or request("GET", "/api/info").get("pid") != server.pid + ): # type: ignore[union-attr] + raise RuntimeError("private server identity changed") + return request(method, path, value, timeout) + + created = mutate( + "POST", + "/api/session", + { + "location": {"directory": str(project)}, + "model": {"providerID": "capture", "id": "mock"}, + "title": "Private status smoke", + }, + ) + session = created["data"]["id"] # type: ignore[index] + if args.with_tui: + name = "ai-oc-v2-status-" + secrets.token_hex(5) + result = subprocess.run( + [ + "tmux", + "new-session", + "-d", + "-P", + "-F", + "#{session_id}", + "-s", + name, + "-x", + "160", + "-y", + "45", + "-c", + str(project), + "env", + "-i", + *(f"{key}={value}" for key, value in env.items()), + binary, + "--server", + base, + "--session", + session, + str(project), + ], + capture_output=True, + text=True, + timeout=10, + check=True, + ) + tmux_target = result.stdout.strip() + if ( + not tmux_target.startswith("$") + or not tmux_target[1:].isdigit() + ): + raise RuntimeError("unexpected private tmux identity") + mutate( + "POST", + f"/api/session/{session}/prompt", + {"text": "Run the private validation fixture."}, + ) + mutate( + "POST", f"/api/experimental/session/{session}/wait", timeout=30 + ) + state_path = project / ".opencode/gateway-core.state.json" + + def entries() -> object: + try: + return json.loads(state_path.read_text())[ + "executionStatus" + ]["sessions"] + except (OSError, ValueError, KeyError, TypeError): + return None + + for _ in range(100): + current = entries() + item = ( + current.get(session) if isinstance(current, dict) else None + ) + if ( + isinstance(item, dict) + and item.get("last") == "Validation passed" + and item.get("next") == "Review changes" + ): + break + time.sleep(0.1) + else: + raise RuntimeError("real V2 tool event did not update status") + if CANARY in state_path.read_text(): + raise RuntimeError("tool output appeared in status state") + if ( + state_path.is_symlink() + or state_path.parent.is_symlink() + or stat.S_IMODE(state_path.stat().st_mode) != 0o600 + or stat.S_IMODE(state_path.parent.stat().st_mode) != 0o700 + ): + raise RuntimeError("status state is not private") + if args.with_tui: + for _ in range(100): + screen = subprocess.run( + ["tmux", "capture-pane", "-t", tmux_target, "-p"], + capture_output=True, + text=True, + timeout=3, + check=True, + ).stdout + if ( + "Validation passed" in screen + and "Review changes" in screen + ): + break + time.sleep(0.15) + else: + raise RuntimeError( + "V2 sidebar did not render the generated status" + ) + with provider.lock: + if not provider.offered or not provider.continued: + raise RuntimeError( + "synthetic provider did not observe the shell round trip" + ) + other = mutate( + "POST", + "/api/session", + { + "location": {"directory": str(project)}, + "model": {"providerID": "capture", "id": "mock"}, + "title": "Other private session", + }, + ) + other_id = other["data"]["id"] # type: ignore[index] + if other_id == session: + raise RuntimeError("private session identities collided") + for _ in range(100): + current = entries() + other_entry = ( + current.get(other_id) if isinstance(current, dict) else None + ) + if ( + isinstance(other_entry, dict) + and other_entry.get("last") == "Session ready" + ): + break + time.sleep(0.1) + else: + raise RuntimeError("second session status was not created") + mutate("DELETE", f"/api/session/{session}") + for _ in range(100): + current = entries() + if ( + isinstance(current, dict) + and session not in current + and current.get(other_id) == other_entry + ): + break + time.sleep(0.1) + else: + raise RuntimeError( + "deleted status remained or another session changed" + ) + print( + json.dumps( + { + "result": "PASS", + "version": version, + "tool_status": True, + "deleted_status": True, + "tui_rendered": args.with_tui, + } + ) + ) + return 0 + finally: + if ( + tmux_target + and tmux_target.startswith("$") + and tmux_target[1:].isdigit() + ): + subprocess.run( + ["tmux", "kill-session", "-t", tmux_target], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=5, + check=False, + ) + if server.poll() is None: + os.killpg(server.pid, signal.SIGTERM) + try: + server.wait(timeout=5) + except subprocess.TimeoutExpired: + os.killpg(server.pid, signal.SIGKILL) + server.wait(timeout=5) + finally: + provider.shutdown() + provider.server_close() + thread.join(timeout=5) + + +if __name__ == "__main__": + raise SystemExit(main())