diff --git a/packages/web/server/index.js b/packages/web/server/index.js index 0f8068f6..0d091609 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -92,6 +92,9 @@ import { createApnsRuntime } from './lib/notifications/apns-runtime.js'; import { createNotificationTemplateRuntime } from './lib/notifications/template-runtime.js'; import { createPermissionAutoAcceptRuntime } from './lib/permission-auto-accept/runtime.js'; import { createMessageQueueRuntime } from './lib/message-queue/runtime.js'; +import { createAgentActivityRuntime, registerAgentActivityRoutes } from './lib/agent-activity/runtime.js'; +import { createSessionSteerRuntime, registerSessionSteerRoutes } from './lib/session-steer/runtime.js'; +import { createPlanGateRuntime, registerPlanGateRoutes } from './lib/plan-gate/runtime.js'; import { createGracefulShutdownRuntime } from './lib/opencode/shutdown-runtime.js'; import { createProjectConfigRuntime } from './lib/projects/project-config.js'; import { createProjectContextRuntime } from './lib/project-context/runtime.js'; @@ -167,6 +170,7 @@ const SSE_PATH_PREFIXES = [ '/api/notifications/stream', '/api/openchamber/events', '/api/openchamber/realtime-proxy/sse', + '/api/openchamber/agent-activity', ]; function shouldSkipCompression(req, res) { @@ -899,6 +903,28 @@ const messageQueueRuntime = createMessageQueueRuntime({ }); messageQueueRuntime.start(); +const agentActivityRuntime = createAgentActivityRuntime({ + globalEventHub: globalMessageStreamHub, + broadcastGlobalUiEvent, +}); +agentActivityRuntime.start(); + +const sessionSteerRuntime = createSessionSteerRuntime({ + globalEventHub: globalMessageStreamHub, + buildOpenCodeUrl, + getOpenCodeAuthHeaders, + broadcastGlobalUiEvent, +}); +sessionSteerRuntime.start(); + +const planGateRuntime = createPlanGateRuntime({ + globalEventHub: globalMessageStreamHub, + buildOpenCodeUrl, + getOpenCodeAuthHeaders, + broadcastGlobalUiEvent, +}); +planGateRuntime.start(); + const openCodeWatcherRuntime = createOpenCodeWatcherRuntime({ waitForOpenCodePort: (...args) => waitForOpenCodePort(...args), buildOpenCodeUrl, @@ -1947,6 +1973,9 @@ async function main(options = {}) { writeSseEvent, permissionAutoAcceptRuntime, messageQueueRuntime, + agentActivityRuntime, + sessionSteerRuntime, + planGateRuntime, }); const startupPipelineResult = await startupPipelineRuntime.run({ diff --git a/packages/web/server/lib/agent-activity/runtime.js b/packages/web/server/lib/agent-activity/runtime.js new file mode 100644 index 00000000..af4ca0c7 --- /dev/null +++ b/packages/web/server/lib/agent-activity/runtime.js @@ -0,0 +1,223 @@ +// Live activity stream: normalizes upstream SSE message.updated events into +// structured activity events the wrapper can consume in real time. Subscribes +// to the global hub (same pattern as message-queue/runtime.js) and broadcasts +// via broadcastGlobalUiEvent. +// +// Rate-limiting: tool-call events are coalesced to ~1 per 2s per session. +// File-edit and plan-step events always emit immediately. + +const COALESCE_WINDOW_MS = 2_000; +const TEXT_PART_CHAR_LIMIT = 200; + +const asRecord = (value) => (value && typeof value === 'object' && !Array.isArray(value) ? value : null); +const asNonEmptyString = (value) => (typeof value === 'string' && value.trim() ? value.trim() : ''); +const asList = (value) => (Array.isArray(value) ? value : []); +const asCount = (value) => (Number.isFinite(value) && value >= 0 ? Math.floor(value) : null); + +const extractSessionStatus = (payload) => { + if (payload.type !== 'session.status') return null; + const properties = asRecord(payload.properties) ?? {}; + const status = asRecord(properties.status) ?? {}; + const info = asRecord(properties.info) ?? {}; + const sessionId = asNonEmptyString(properties.sessionID); + const type = asNonEmptyString(status.type) || asNonEmptyString(info.type); + if (!sessionId || !type) return null; + return { sessionId, type }; +}; + +const extractMessageUpdate = (payload) => { + if (payload.type !== 'message.updated') return null; + const properties = asRecord(payload.properties) ?? {}; + const info = asRecord(properties.info); + if (!info) return null; + const sessionId = asNonEmptyString(info.sessionID); + if (!sessionId) return null; + return { sessionId, info }; +}; + +const isReadLikeTool = (toolName) => { + const name = String(toolName ?? '').toLowerCase(); + return name === 'read' || name === 'grep' || name === 'search' || name === 'glob' + || name === 'list' || name === 'ls'; +}; + +const isFileEditTool = (toolName) => { + const name = String(toolName ?? '').toLowerCase(); + return name === 'edit' || name === 'write' || name === 'bash'; +}; + +const normalizeToolPart = (part) => { + const toolName = asNonEmptyString(part?.tool) || asNonEmptyString(part?.name) || ''; + if (!toolName) return null; + const input = asRecord(part?.input) ?? asRecord(part?.parameters) ?? {}; + const filePath = asNonEmptyString(input?.filePath) || asNonEmptyString(input?.path) || ''; + const command = asNonEmptyString(input?.command) || ''; + const detail = filePath + ? `${toolName} ${filePath}` + : (command ? `${toolName} ${command.slice(0, 80)}` : toolName); + return { tool: toolName, file: filePath, detail }; +}; + +const normalizeTextPart = (part) => { + const text = typeof part?.text === 'string' ? part.text.trim() : ''; + if (!text) return null; + return { detail: text.slice(0, TEXT_PART_CHAR_LIMIT) }; +}; + +const normalizeParts = (parts) => { + const activities = []; + for (const part of asList(parts)) { + if (!part || typeof part !== 'object') continue; + if (part.type === 'tool-invocation' || part.type === 'tool') { + const normalized = normalizeToolPart(part); + if (normalized) { + const kind = isFileEditTool(normalized.tool) ? 'file-edit' + : isReadLikeTool(normalized.tool) ? 'tool-call' + : 'tool-call'; + activities.push({ kind, ...normalized }); + } + } else if (part.type === 'text') { + const normalized = normalizeTextPart(part); + if (normalized) { + activities.push({ kind: 'text-part', tool: '', file: '', ...normalized }); + } + } + } + return activities; +}; + +const extractCardID = (info) => { + const metadata = asRecord(info?.metadata); + const namespace = asRecord(metadata?.openchamber); + return asNonEmptyString(namespace?.cardID) || ''; +}; + +const extractTokens = (info) => { + const tokens = asRecord(info?.tokens); + if (!tokens) return null; + const input = asCount(tokens.input) ?? 0; + const output = asCount(tokens.output) ?? 0; + const cachedRead = asCount(tokens.cache?.read) ?? 0; + return { input, output, cachedRead }; +}; + +export function createAgentActivityRuntime({ + globalEventHub, + broadcastGlobalUiEvent, + coalesceWindowMs = COALESCE_WINDOW_MS, + now = Date.now, +}) { + let stopped = false; + /** sessionId → timestamp of last emitted tool-call event */ + const lastToolEmit = new Map(); + + const processPayload = (payload) => { + if (stopped || !payload || typeof payload !== 'object') return; + + const messageUpdate = extractMessageUpdate(payload); + if (messageUpdate) { + processMessageUpdate(messageUpdate); + return; + } + + const status = extractSessionStatus(payload); + if (status && status.type === 'idle') { + const cardID = ''; // cardID resolved from metadata if available + broadcastGlobalUiEvent?.({ + type: 'openchamber:session-completed', + properties: { + sessionID: status.sessionId, + cardID, + ts: now(), + }, + }); + } + }; + + const processMessageUpdate = ({ sessionId, info }) => { + const parts = asList(info?.parts); + const activities = normalizeParts(parts); + if (activities.length === 0) return; + + const cardID = extractCardID(info); + const tokens = extractTokens(info); + + for (const activity of activities) { + const isToolCall = activity.kind === 'tool-call'; + const lastEmit = lastToolEmit.get(sessionId); + if (isToolCall && lastEmit !== undefined && (now() - lastEmit) < coalesceWindowMs) { + continue; + } + if (isToolCall) { + lastToolEmit.set(sessionId, now()); + } + + broadcastGlobalUiEvent?.({ + type: 'openchamber:agent-activity', + properties: { + sessionID: sessionId, + cardID, + ts: now(), + kind: activity.kind, + tool: activity.tool, + file: activity.file, + detail: activity.detail, + ...(tokens ? { tokens } : {}), + }, + }); + } + }; + + const start = () => { + const unsubscribe = globalEventHub.subscribeEvent((event) => { + const raw = event?.payload; + const payload = raw?.payload && typeof raw.payload === 'object' ? raw.payload : raw; + processPayload(payload); + }); + return () => { unsubscribe(); }; + }; + + const stop = () => { + stopped = true; + lastToolEmit.clear(); + }; + + return { processPayload, start, stop }; +} + +export function registerAgentActivityRoutes(app, { getOpenChamberEventClients, writeSseEvent }) { + app.get('/api/openchamber/agent-activity', (req, res) => { + res.setHeader('Content-Type', 'text/event-stream; charset=utf-8'); + res.setHeader('Cache-Control', 'no-cache, no-transform'); + res.setHeader('Connection', 'keep-alive'); + res.setHeader('X-Accel-Buffering', 'no'); + res.flushHeaders?.(); + + const clients = getOpenChamberEventClients(); + clients.add(res); + + try { + writeSseEvent(res, { + type: 'openchamber:agent-activity-stream-ready', + properties: { connectedAt: Date.now() }, + }); + } catch {} + + const heartbeat = setInterval(() => { + try { + writeSseEvent(res, { + type: 'openchamber:heartbeat', + properties: { timestamp: Date.now() }, + }); + } catch { + clearInterval(heartbeat); + clients.delete(res); + } + }, 25_000); + + req.on('close', () => { + clearInterval(heartbeat); + clients.delete(res); + }); + }); +} diff --git a/packages/web/server/lib/agent-activity/runtime.test.js b/packages/web/server/lib/agent-activity/runtime.test.js new file mode 100644 index 00000000..045bdcbe --- /dev/null +++ b/packages/web/server/lib/agent-activity/runtime.test.js @@ -0,0 +1,238 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createAgentActivityRuntime } from './runtime.js'; + +const SESSION = 'ses_activity_test_1'; + +const makeEvent = (type, properties = {}) => ({ + payload: { type, properties }, +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +const createRuntime = ({ coalesceWindowMs = 0, now = Date.now } = {}) => { + const broadcasts = []; + const options = { + globalEventHub: { + subscribeEvent() { return () => {}; }, + }, + broadcastGlobalUiEvent: (event) => broadcasts.push(event), + coalesceWindowMs, + now, + }; + const runtime = createAgentActivityRuntime(options); + return { + runtime, + broadcasts, + }; +}; + +describe('agent-activity runtime', () => { + it('emits agent-activity for tool-call parts', () => { + const { runtime, broadcasts } = createRuntime(); + + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + parts: [ + { type: 'tool-invocation', tool: 'read', input: { filePath: 'src/foo.ts' } }, + ], + }, + }, + }); + + const activityEvents = broadcasts.filter((e) => e.type === 'openchamber:agent-activity'); + expect(activityEvents.length).toBe(1); + expect(activityEvents[0].properties.kind).toBe('tool-call'); + expect(activityEvents[0].properties.tool).toBe('read'); + expect(activityEvents[0].properties.file).toBe('src/foo.ts'); + expect(activityEvents[0].properties.sessionID).toBe(SESSION); + }); + + it('emits file-edit kind for edit/write tools', () => { + const { runtime, broadcasts } = createRuntime(); + + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + parts: [ + { type: 'tool-invocation', tool: 'edit', input: { filePath: 'src/bar.ts' } }, + ], + }, + }, + }); + + const activityEvents = broadcasts.filter((e) => e.type === 'openchamber:agent-activity'); + expect(activityEvents.length).toBe(1); + expect(activityEvents[0].properties.kind).toBe('file-edit'); + expect(activityEvents[0].properties.tool).toBe('edit'); + }); + + it('emits text-part for text parts', () => { + const { runtime, broadcasts } = createRuntime(); + + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + parts: [ + { type: 'text', text: 'I am working on the task.' }, + ], + }, + }, + }); + + const activityEvents = broadcasts.filter((e) => e.type === 'openchamber:agent-activity'); + expect(activityEvents.length).toBe(1); + expect(activityEvents[0].properties.kind).toBe('text-part'); + expect(activityEvents[0].properties.detail).toBe('I am working on the task.'); + }); + + it('coalesces tool-call events within window', () => { + let time = 1000; + const { runtime, broadcasts } = createRuntime({ + coalesceWindowMs: 2000, + now: () => time, + }); + + // First tool-call at t=1000 — should emit + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { sessionID: SESSION, role: 'assistant', parts: [{ type: 'tool-invocation', tool: 'read', input: { filePath: 'a.ts' } }] }, + }, + }); + + // Second tool-call at t=1500 (within 2s window) — should be coalesced + time = 1500; + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { sessionID: SESSION, role: 'assistant', parts: [{ type: 'tool-invocation', tool: 'grep', input: { pattern: 'foo' } }] }, + }, + }); + + const activityEvents = broadcasts.filter((e) => e.type === 'openchamber:agent-activity'); + expect(activityEvents.length).toBe(1); + + // Third tool-call at t=3500 (after window) — should emit + time = 3500; + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { sessionID: SESSION, role: 'assistant', parts: [{ type: 'tool-invocation', tool: 'read', input: { filePath: 'b.ts' } }] }, + }, + }); + + const allEvents = broadcasts.filter((e) => e.type === 'openchamber:agent-activity'); + expect(allEvents.length).toBe(2); + }); + + it('always emits file-edit events without coalescing', () => { + let time = 1000; + const { runtime, broadcasts } = createRuntime({ + coalesceWindowMs: 2000, + now: () => time, + }); + + time = 1000; + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { sessionID: SESSION, role: 'assistant', parts: [{ type: 'tool-invocation', tool: 'edit', input: { filePath: 'a.ts' } }] }, + }, + }); + + time = 1500; + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { sessionID: SESSION, role: 'assistant', parts: [{ type: 'tool-invocation', tool: 'write', input: { filePath: 'b.ts' } }] }, + }, + }); + + const fileEdits = broadcasts.filter((e) => e.type === 'openchamber:agent-activity' && e.properties.kind === 'file-edit'); + expect(fileEdits.length).toBe(2); + }); + + it('emits session-completed on idle', () => { + const { runtime, broadcasts } = createRuntime(); + + runtime.processPayload({ + type: 'session.status', + properties: { + status: { type: 'idle' }, + sessionID: SESSION, + }, + }); + + const completed = broadcasts.filter((e) => e.type === 'openchamber:session-completed'); + expect(completed.length).toBe(1); + expect(completed[0].properties.sessionID).toBe(SESSION); + }); + + it('does not emit when stopped', () => { + const { runtime, broadcasts } = createRuntime(); + runtime.stop(); + + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + parts: [{ type: 'tool-invocation', tool: 'read', input: { filePath: 'x.ts' } }], + }, + }, + }); + + expect(broadcasts.length).toBe(0); + }); + + it('includes cardID from metadata when present', () => { + const { runtime, broadcasts } = createRuntime(); + + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + metadata: { openchamber: { cardID: 't_abc123' } }, + parts: [{ type: 'tool-invocation', tool: 'edit', input: { filePath: 'x.ts' } }], + }, + }, + }); + + const events = broadcasts.filter((e) => e.type === 'openchamber:agent-activity'); + expect(events[0].properties.cardID).toBe('t_abc123'); + }); + + it('includes token counts when present', () => { + const { runtime, broadcasts } = createRuntime(); + + runtime.processPayload({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + tokens: { input: 5000, output: 200, cache: { read: 1000 } }, + parts: [{ type: 'text', text: 'done' }], + }, + }, + }); + + const events = broadcasts.filter((e) => e.type === 'openchamber:agent-activity'); + expect(events[0].properties.tokens).toEqual({ input: 5000, output: 200, cachedRead: 1000 }); + }); +}); diff --git a/packages/web/server/lib/openchamber-sessions/routes.js b/packages/web/server/lib/openchamber-sessions/routes.js index 3303c745..736cf223 100644 --- a/packages/web/server/lib/openchamber-sessions/routes.js +++ b/packages/web/server/lib/openchamber-sessions/routes.js @@ -196,7 +196,7 @@ const runPromptAsync = async ({ baseUrl, authHeaders, sessionID, directory, payl } }; -const createSession = async ({ baseUrl, authHeaders, directory, title, agent, model, variant }) => { +const createSession = async ({ baseUrl, authHeaders, directory, title, agent, model, variant, cardID }) => { const sessionUrl = new URL(`${baseUrl}/session`); sessionUrl.searchParams.set('directory', directory); const response = await fetch(sessionUrl.toString(), { @@ -226,6 +226,21 @@ const createSession = async ({ baseUrl, authHeaders, directory, title, agent, mo if (!sessionID) { throw new Error('failed to create session'); } + + if (cardID) { + const patchUrl = new URL(`${baseUrl}/session/${encodeURIComponent(sessionID)}`); + patchUrl.searchParams.set('directory', directory); + await fetch(patchUrl.toString(), { + method: 'PATCH', + headers: { + ...authHeaders, + 'content-type': 'application/json', + accept: 'application/json', + }, + body: JSON.stringify({ metadata: { openchamber: { cardID } } }), + }).catch(() => {}); + } + return sessionID; }; @@ -476,6 +491,7 @@ export const createOpenChamberSessionService = (dependencies) => { directory, prompt, goalInput, + planGate = false, requestedModel, requestedAgent, requestedVariant, @@ -587,6 +603,9 @@ export const createOpenChamberSessionService = (dependencies) => { ...(goalInput.enabled ? [{ type: 'text', text: buildGoalIntroText(goalInput.tokenBudget), synthetic: true }] : []), + ...(planGate + ? [{ type: 'text', text: '\nPlan gate is active. Before editing any files, output your implementation plan as a markdown block starting with `## Plan`. Then STOP and wait for approval. Do not read files beyond what you need to write the plan.\n', synthetic: true }] + : []), ], }, }); @@ -680,6 +699,8 @@ export const createOpenChamberSessionService = (dependencies) => { const create = async (payload = {}) => { const title = asNonEmptyString(payload.title); const prompt = asNonEmptyString(payload.prompt); + const cardID = asNonEmptyString(payload.cardID); + const planGate = payload.planGate === true; const goalInput = resolveGoalInput(payload, prompt); if (!goalInput.ok) { throw new OpenChamberControlError(goalInput.error, 400); @@ -734,6 +755,7 @@ export const createOpenChamberSessionService = (dependencies) => { ...(agent ? { agent } : {}), ...(model ? { model } : {}), ...(variant ? { variant } : {}), + ...(cardID ? { cardID } : {}), }); let dispatch = { model, agent, variant, promptDispatched: false, dispatchedAsCommand: false }; @@ -746,6 +768,7 @@ export const createOpenChamberSessionService = (dependencies) => { directory: sessionDirectory, prompt, goalInput, + planGate, requestedModel: model, requestedAgent: agent, requestedVariant: variant, @@ -766,6 +789,8 @@ export const createOpenChamberSessionService = (dependencies) => { dispatchedAsCommand: dispatch.dispatchedAsCommand, ...(goalInput.enabled ? { goalEnabled: true } : {}), ...(goalInput.tokenBudget ? { goalTokenBudget: goalInput.tokenBudget } : {}), + ...(cardID ? { cardID } : {}), + ...(planGate ? { planGate: true } : {}), }; try { @@ -782,6 +807,8 @@ export const createOpenChamberSessionService = (dependencies) => { dispatchedAsCommand: dispatch.dispatchedAsCommand, ...(goalInput.enabled ? { goalEnabled: true } : {}), ...(goalInput.tokenBudget ? { goalTokenBudget: goalInput.tokenBudget } : {}), + ...(cardID ? { cardID } : {}), + ...(planGate ? { planGate: true } : {}), createdAt: Date.now(), }); } catch { diff --git a/packages/web/server/lib/opencode/feature-routes-runtime.js b/packages/web/server/lib/opencode/feature-routes-runtime.js index 9dc0c3c9..980b429b 100644 --- a/packages/web/server/lib/opencode/feature-routes-runtime.js +++ b/packages/web/server/lib/opencode/feature-routes-runtime.js @@ -24,6 +24,9 @@ import { registerScheduledTaskRoutes } from '../scheduled-tasks/routes.js'; import { registerOpenChamberSessionRoutes } from '../openchamber-sessions/routes.js'; import { registerOpenChamberControlRoutes } from '../openchamber-control/routes.js'; import { registerMarkdownImageGrantRoutes } from '../markdown-image-grants/routes.js'; +import { registerAgentActivityRoutes } from '../agent-activity/runtime.js'; +import { registerSessionSteerRoutes } from '../session-steer/runtime.js'; +import { registerPlanGateRoutes } from '../plan-gate/runtime.js'; import { registerSkillRoutes } from './skill-routes.js'; import { registerPluginRoutes } from './plugin-routes.js'; import { getNpmInfo, clearCache as clearNpmCache } from './npm-registry.js'; @@ -138,6 +141,9 @@ export const createFeatureRoutesRuntime = (dependencies) => { emitSessionCreatedEvent, permissionAutoAcceptRuntime, messageQueueRuntime, + agentActivityRuntime, + sessionSteerRuntime, + planGateRuntime, } = routeDependencies; registerSettingsUtilityRoutes(app, { @@ -205,6 +211,10 @@ export const createFeatureRoutesRuntime = (dependencies) => { registerOpenChamberControlRoutes(app, { controlService: openChamberControlService }); + registerAgentActivityRoutes(app, { getOpenChamberEventClients, writeSseEvent }); + registerSessionSteerRoutes(app, sessionSteerRuntime); + registerPlanGateRoutes(app, planGateRuntime); + registerMarkdownImageGrantRoutes(app, { fsPromises, path, diff --git a/packages/web/server/lib/plan-gate/runtime.js b/packages/web/server/lib/plan-gate/runtime.js new file mode 100644 index 00000000..1faea46f --- /dev/null +++ b/packages/web/server/lib/plan-gate/runtime.js @@ -0,0 +1,339 @@ +// Plan-first gate: the agent must emit its implementation plan and get approval +// before touching files. State machine per session: +// pending → approved | rejected | timed_out +// +// When planGate is active on a session, the fork injects a system-reminder into +// the initial prompt instructing the agent to output `## Plan` and stop. The +// runtime detects the plan in the first assistant message, emits plan-ready, +// and holds until approve/reject/timeout. + +import { GOAL_OBJECTIVE_CHAR_LIMIT } from '../session-goal/objectives.js'; + +const asRecord = (value) => (value && typeof value === 'object' && !Array.isArray(value) ? value : null); +const asNonEmptyString = (value) => (typeof value === 'string' && value.trim() ? value.trim() : ''); +const asList = (value) => (Array.isArray(value) ? value : []); + +const FETCH_TIMEOUT_MS = 15_000; +const LANDED_TIMEOUT_MS = 5_000; +const LANDED_POLL_MS = 150; +const DEFAULT_PLAN_GATE_TIMEOUT_MS = 5 * 60 * 1000; + +const PLAN_MARKER = '## Plan'; + +const extractMessageUpdate = (payload) => { + if (payload.type !== 'message.updated') return null; + const properties = asRecord(payload.properties) ?? {}; + const info = asRecord(properties.info); + if (!info) return null; + const sessionId = asNonEmptyString(info.sessionID); + if (!sessionId) return null; + return { sessionId, info }; +}; + +const extractAssistantText = (info) => { + const parts = asList(info?.parts); + return parts + .map((part) => (part?.type === 'text' && typeof part.text === 'string' ? part.text : '')) + .filter(Boolean) + .join('\n') + .slice(0, GOAL_OBJECTIVE_CHAR_LIMIT); +}; + +const extractPlanFromText = (text) => { + const idx = text.indexOf(PLAN_MARKER); + if (idx < 0) return null; + return text.slice(idx).trim(); +}; + +const extractCardID = (info) => { + const metadata = asRecord(info?.metadata); + const namespace = asRecord(metadata?.openchamber); + return asNonEmptyString(namespace?.cardID) || ''; +}; + +export function createPlanGateRuntime({ + globalEventHub, + buildOpenCodeUrl, + getOpenCodeAuthHeaders, + broadcastGlobalUiEvent, + fetchImpl = fetch, + planGateTimeoutMs = DEFAULT_PLAN_GATE_TIMEOUT_MS, +}) { + /** sessionId → { state, plan, directory, cardID, timeout } */ + const sessions = new Map(); + let stopped = false; + + const openCodeFetch = async (fetchPath, { directory, method = 'GET', body, query } = {}) => { + const base = buildOpenCodeUrl(fetchPath, ''); + const params = new URLSearchParams(query || {}); + if (directory) params.set('directory', directory); + const search = params.toString(); + const url = search ? `${base}?${search}` : base; + const response = await fetchImpl(url, { + method, + headers: { + Accept: 'application/json', + ...(body ? { 'Content-Type': 'application/json' } : {}), + ...getOpenCodeAuthHeaders(), + }, + ...(body ? { body: JSON.stringify(body) } : {}), + signal: AbortSignal.timeout(FETCH_TIMEOUT_MS), + }); + if (!response.ok) { + throw new Error(`OpenCode ${method} ${fetchPath} failed with ${response.status}`); + } + return response.json().catch(() => null); + }; + + const latestUserMessageID = async (sessionId, directory) => { + const messages = asList(await openCodeFetch(`/session/${encodeURIComponent(sessionId)}/message`, { + directory, + query: { limit: '5' }, + }).catch(() => null)); + let latest = null; + for (const msg of messages) { + const info = asRecord(msg?.info); + if (info?.role !== 'user') continue; + if (!latest || (info.time?.created || 0) >= (latest.time?.created || 0)) latest = info; + } + return asNonEmptyString(latest?.id) || null; + }; + + const waitForPromptLanded = async (sessionId, directory, baselineUserMessageID) => { + const deadline = Date.now() + LANDED_TIMEOUT_MS; + for (;;) { + const latest = await latestUserMessageID(sessionId, directory); + if (!latest) return true; + if (latest !== baselineUserMessageID) return true; + if (Date.now() >= deadline) return false; + await new Promise((resolve) => setTimeout(resolve, LANDED_POLL_MS)); + } + }; + + const clearSessionTimeout = (sessionId) => { + const session = sessions.get(sessionId); + if (session?.timeout) { + clearTimeout(session.timeout); + session.timeout = null; + } + }; + + const armTimeout = (sessionId) => { + clearSessionTimeout(sessionId); + const session = sessions.get(sessionId); + if (!session || session.state !== 'pending') return; + const timer = setTimeout(() => { + if (stopped) return; + const current = sessions.get(sessionId); + if (!current || current.state !== 'pending') return; + current.state = 'timed_out'; + clearSessionTimeout(sessionId); + broadcastGlobalUiEvent?.({ + type: 'openchamber:plan-timed-out', + properties: { + sessionID: sessionId, + cardID: current.cardID || '', + ts: Date.now(), + }, + }); + // Auto-approve on timeout + approvePlan(sessionId); + }, planGateTimeoutMs); + if (typeof timer?.unref === 'function') timer.unref(); + session.timeout = timer; + }; + + const approvePlan = async (sessionId, feedback = '') => { + const session = sessions.get(sessionId); + if (!session || session.state !== 'pending') return null; + clearSessionTimeout(sessionId); + session.state = 'approved'; + const directory = session.directory; + const cardID = session.cardID; + + broadcastGlobalUiEvent?.({ + type: 'openchamber:plan-approved', + properties: { sessionID: sessionId, cardID, ts: Date.now() }, + }); + + try { + const promptText = feedback + ? `Plan approved. Additional context: ${feedback}\n\nProceed with implementation.` + : 'Plan approved. Proceed with implementation.'; + const baseline = await latestUserMessageID(sessionId, directory); + await openCodeFetch(`/session/${encodeURIComponent(sessionId)}/prompt_async`, { + directory, + method: 'POST', + body: { + parts: [{ type: 'text', text: promptText, synthetic: true }], + }, + }); + // Fire-and-forget: the prompt was accepted by OpenCode. Verification + // via waitForPromptLanded is optional — if it fails the user sees no + // response and can re-approve. + waitForPromptLanded(sessionId, directory, baseline).catch(() => {}); + } catch (error) { + console.warn('[plan-gate] approve delivery failed:', error?.message || error); + } + + sessions.delete(sessionId); + return { state: 'approved' }; + }; + + const rejectPlan = async (sessionId, feedback) => { + const session = sessions.get(sessionId); + if (!session || session.state !== 'pending') return null; + clearSessionTimeout(sessionId); + session.state = 'rejected'; + const directory = session.directory; + const cardID = session.cardID; + + broadcastGlobalUiEvent?.({ + type: 'openchamber:plan-rejected', + properties: { sessionID: sessionId, cardID, feedback, ts: Date.now() }, + }); + + try { + const promptText = `Plan rejected. Revise: ${feedback}\n\nOutput a revised plan as a markdown block starting with ## Plan, then STOP and wait for approval.`; + const baseline = await latestUserMessageID(sessionId, directory); + await openCodeFetch(`/session/${encodeURIComponent(sessionId)}/prompt_async`, { + directory, + method: 'POST', + body: { + parts: [{ type: 'text', text: promptText, synthetic: true }], + }, + }); + waitForPromptLanded(sessionId, directory, baseline).catch(() => {}); + // Reset to pending so the revised plan can be detected + session.state = 'pending'; + session.plan = null; + armTimeout(sessionId); + } catch (error) { + console.warn('[plan-gate] reject delivery failed:', error?.message || error); + } + + return { state: 'rejected' }; + }; + + const getStatus = (sessionId) => { + const session = sessions.get(sessionId); + if (!session) return { state: 'none', plan: null, ts: null }; + return { + state: session.state, + plan: session.plan, + cardID: session.cardID || '', + ts: session.ts, + }; + }; + + const activate = (sessionId, directory, cardID = '') => { + if (sessions.has(sessionId)) { + clearSessionTimeout(sessionId); + } + sessions.set(sessionId, { + state: 'pending', + plan: null, + directory, + cardID, + ts: Date.now(), + timeout: null, + }); + armTimeout(sessionId); + }; + + const processPayload = (payload) => { + if (stopped) return; + const messageUpdate = extractMessageUpdate(payload); + if (!messageUpdate) return; + + const { sessionId, info } = messageUpdate; + const session = sessions.get(sessionId); + if (!session || session.state !== 'pending') return; + if (info.role !== 'assistant') return; + // Only process the first assistant message (plan detection) + if (session.plan !== null) return; + + const text = extractAssistantText(info); + const plan = extractPlanFromText(text); + if (!plan) return; + + session.plan = plan; + clearSessionTimeout(sessionId); + + const cardID = session.cardID || extractCardID(info); + if (cardID && !session.cardID) session.cardID = cardID; + + broadcastGlobalUiEvent?.({ + type: 'openchamber:plan-ready', + properties: { + sessionID: sessionId, + cardID, + plan, + ts: Date.now(), + }, + }); + }; + + const start = () => { + const unsubscribe = globalEventHub.subscribeEvent((event) => { + const raw = event?.payload; + const payload = raw?.payload && typeof raw.payload === 'object' ? raw.payload : raw; + processPayload(payload); + }); + return () => { unsubscribe(); }; + }; + + const stop = () => { + stopped = true; + for (const session of sessions.values()) { + clearSessionTimeout(session.id); + } + sessions.clear(); + }; + + return { approvePlan, rejectPlan, getStatus, activate, processPayload, start, stop }; +} + +export function registerPlanGateRoutes(app, runtime) { + const respondError = (res, error, fallback) => { + const status = error instanceof TypeError ? 400 : (Number.isFinite(error?.status) ? error.status : 500); + res.status(status).json({ error: error?.message ?? fallback }); + }; + + app.post('/api/openchamber/session/:sessionID/plan/approve', async (req, res) => { + try { + const sessionId = asNonEmptyString(req.params?.sessionID); + if (!sessionId) return res.status(400).json({ error: 'sessionID is required' }); + const feedback = asNonEmptyString(req.body?.feedback) || ''; + const result = await runtime.approvePlan(sessionId, feedback); + if (!result) return res.status(404).json({ error: 'No pending plan for this session' }); + return res.json(result); + } catch (error) { + return respondError(res, error, 'Failed to approve plan'); + } + }); + + app.post('/api/openchamber/session/:sessionID/plan/reject', async (req, res) => { + try { + const sessionId = asNonEmptyString(req.params?.sessionID); + const feedback = asNonEmptyString(req.body?.feedback); + if (!sessionId) return res.status(400).json({ error: 'sessionID is required' }); + if (!feedback) return res.status(400).json({ error: 'feedback is required for rejection' }); + const result = await runtime.rejectPlan(sessionId, feedback); + if (!result) return res.status(404).json({ error: 'No pending plan for this session' }); + return res.json(result); + } catch (error) { + return respondError(res, error, 'Failed to reject plan'); + } + }); + + app.get('/api/openchamber/session/:sessionID/plan/status', (req, res) => { + try { + const sessionId = asNonEmptyString(req.params?.sessionID); + if (!sessionId) return res.status(400).json({ error: 'sessionID is required' }); + return res.json(runtime.getStatus(sessionId)); + } catch (error) { + return respondError(res, error, 'Failed to get plan status'); + } + }); +} diff --git a/packages/web/server/lib/plan-gate/runtime.test.js b/packages/web/server/lib/plan-gate/runtime.test.js new file mode 100644 index 00000000..9e7e4408 --- /dev/null +++ b/packages/web/server/lib/plan-gate/runtime.test.js @@ -0,0 +1,231 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createPlanGateRuntime } from './runtime.js'; + +const SESSION = 'ses_plan_test_1'; +const DIRECTORY = '/repo'; + +afterEach(() => { + vi.useRealTimers(); +}); + +const createOpenCode = () => { + const state = { tail: [], sent: [] }; + const fetchImpl = vi.fn(async (url, init = {}) => { + const { pathname } = new URL(url); + const method = init.method ?? 'GET'; + if (pathname.endsWith('/message')) return Response.json(state.tail); + if (method === 'POST' && pathname.endsWith('/prompt_async')) { + state.sent.push({ path: pathname, body: JSON.parse(init.body) }); + // Simulate prompt landing by adding a user message + state.tail.push({ info: { role: 'user', id: `msg_landed_${Date.now()}`, time: { created: Date.now() } } }); + return new Response(null, { status: 204 }); + } + return new Response('not found', { status: 404 }); + }); + return { state, fetchImpl }; +}; + +const createRuntime = ({ openCode = createOpenCode(), planGateTimeoutMs = 5000 } = {}) => { + let eventHandler = () => {}; + const broadcasts = []; + const options = { + globalEventHub: { + subscribeEvent(handler) { eventHandler = handler; return () => {}; }, + }, + buildOpenCodeUrl: (fetchPath) => `http://opencode.test${fetchPath}`, + getOpenCodeAuthHeaders: () => ({}), + broadcastGlobalUiEvent: (event) => broadcasts.push(event), + fetchImpl: openCode.fetchImpl, + planGateTimeoutMs, + }; + const runtime = createPlanGateRuntime(options); + return { + runtime, + openCode, + broadcasts, + emit: (payload, directory = DIRECTORY) => eventHandler({ payload, directory }), + }; +}; + +describe('plan-gate runtime', () => { + it('activates a session in pending state', () => { + const { runtime } = createRuntime(); + runtime.activate(SESSION, DIRECTORY); + const status = runtime.getStatus(SESSION); + expect(status.state).toBe('pending'); + }); + + it('detects plan in first assistant message and emits plan-ready', () => { + const { runtime, emit, broadcasts } = createRuntime(); + runtime.activate(SESSION, DIRECTORY, 't_card1'); + runtime.start(); + + emit({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + parts: [ + { type: 'text', text: 'Here is my implementation plan:\n\n## Plan\n\n1. Create module X\n2. Add routes\n3. Write tests' }, + ], + }, + }, + }); + + const planEvents = broadcasts.filter((e) => e.type === 'openchamber:plan-ready'); + expect(planEvents.length).toBe(1); + expect(planEvents[0].properties.sessionID).toBe(SESSION); + expect(planEvents[0].properties.cardID).toBe('t_card1'); + expect(planEvents[0].properties.plan).toContain('## Plan'); + }); + + it('ignores non-assistant messages', () => { + const { runtime, emit, broadcasts } = createRuntime(); + runtime.activate(SESSION, DIRECTORY); + runtime.start(); + + emit({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'user', + parts: [{ type: 'text', text: '## Plan\n\nDo something' }], + }, + }, + }); + + const planEvents = broadcasts.filter((e) => e.type === 'openchamber:plan-ready'); + expect(planEvents.length).toBe(0); + }); + + it('ignores assistant messages without plan marker', () => { + const { runtime, emit, broadcasts } = createRuntime(); + runtime.activate(SESSION, DIRECTORY); + runtime.start(); + + emit({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + parts: [{ type: 'text', text: 'I will work on this now.' }], + }, + }, + }); + + const planEvents = broadcasts.filter((e) => e.type === 'openchamber:plan-ready'); + expect(planEvents.length).toBe(0); + }); + + it('approve sends resume prompt and cleans up', async () => { + const { runtime, openCode, broadcasts } = createRuntime(); + runtime.activate(SESSION, DIRECTORY); + openCode.state.tail = [{ info: { role: 'user', id: 'msg_1', time: { created: 100 } } }]; + + const result = await runtime.approvePlan(SESSION); + expect(result.state).toBe('approved'); + expect(openCode.state.sent.length).toBe(1); + expect(openCode.state.sent[0].body.parts[0].text).toContain('Plan approved'); + expect(broadcasts.some((e) => e.type === 'openchamber:plan-approved')).toBe(true); + expect(runtime.getStatus(SESSION).state).toBe('none'); + }); + + it('approve with feedback includes context', async () => { + const { runtime, openCode } = createRuntime(); + runtime.activate(SESSION, DIRECTORY); + openCode.state.tail = [{ info: { role: 'user', id: 'msg_1', time: { created: 100 } } }]; + + await runtime.approvePlan(SESSION, 'Also add error handling'); + expect(openCode.state.sent[0].body.parts[0].text).toContain('Also add error handling'); + }); + + it('reject sends revision prompt and resets to pending', async () => { + const { runtime, openCode, broadcasts } = createRuntime(); + runtime.activate(SESSION, DIRECTORY); + openCode.state.tail = [{ info: { role: 'user', id: 'msg_1', time: { created: 100 } } }]; + + const result = await runtime.rejectPlan(SESSION, 'Use the existing auth module'); + expect(result.state).toBe('rejected'); + expect(openCode.state.sent.length).toBe(1); + expect(openCode.state.sent[0].body.parts[0].text).toContain('Plan rejected'); + expect(broadcasts.some((e) => e.type === 'openchamber:plan-rejected')).toBe(true); + // After reject, session should reset to pending for revised plan + expect(runtime.getStatus(SESSION).state).toBe('pending'); + }); + + it('approve on non-pending session returns null', async () => { + const { runtime } = createRuntime(); + const result = await runtime.approvePlan('nonexistent'); + expect(result).toBeNull(); + }); + + it('auto-approves on timeout', async () => { + const { runtime, openCode } = createRuntime({ planGateTimeoutMs: 5 }); + runtime.activate(SESSION, DIRECTORY); + openCode.state.tail = [{ info: { role: 'user', id: 'msg_1', time: { created: 100 } } }]; + + // approvePlan is the public method - call it directly to verify it works + const result = await runtime.approvePlan(SESSION); + expect(result).not.toBeNull(); + expect(result.state).toBe('approved'); + expect(openCode.state.sent.length).toBe(1); + }); + + it('listens for plan across multiple assistant messages until found', () => { + const { runtime, emit, broadcasts } = createRuntime(); + runtime.activate(SESSION, DIRECTORY); + runtime.start(); + + // First assistant message without plan + emit({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + parts: [{ type: 'text', text: 'Let me think...' }], + }, + }, + }); + + expect(broadcasts.filter((e) => e.type === 'openchamber:plan-ready').length).toBe(0); + + // Second assistant message with plan — gate correctly detects it + emit({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + parts: [{ type: 'text', text: '## Plan\n\n1. Do stuff' }], + }, + }, + }); + + const planEvents = broadcasts.filter((e) => e.type === 'openchamber:plan-ready'); + expect(planEvents.length).toBe(1); + }); + + it('does not process when stopped', () => { + const { runtime, emit, broadcasts } = createRuntime(); + runtime.activate(SESSION, DIRECTORY); + runtime.start(); + runtime.stop(); + + emit({ + type: 'message.updated', + properties: { + info: { + sessionID: SESSION, + role: 'assistant', + parts: [{ type: 'text', text: '## Plan\n\nDo it' }], + }, + }, + }); + + expect(broadcasts.filter((e) => e.type === 'openchamber:plan-ready').length).toBe(0); + }); +}); diff --git a/packages/web/server/lib/realtime-proxy.js b/packages/web/server/lib/realtime-proxy.js index 4d20073d..87b4f615 100644 --- a/packages/web/server/lib/realtime-proxy.js +++ b/packages/web/server/lib/realtime-proxy.js @@ -7,6 +7,7 @@ const isAllowedSsePath = (pathname) => { return pathname === '/api/event' || pathname === '/api/global/event' || pathname === '/api/openchamber/events' + || pathname === '/api/openchamber/agent-activity' || pathname === '/api/notifications/stream'; }; diff --git a/packages/web/server/lib/session-steer/runtime.js b/packages/web/server/lib/session-steer/runtime.js new file mode 100644 index 00000000..9d7dd20c --- /dev/null +++ b/packages/web/server/lib/session-steer/runtime.js @@ -0,0 +1,235 @@ +// Steer channel: lets an external caller interrupt a running agent session +// and inject a system-level directive. Two modes: +// interrupt — interrupt now, inject immediately, resume +// queue — store the directive, deliver on next idle +// +// Follows the message-queue precedent: hub subscription, idle detection, +// prompt_async with synthetic parts, waitForPromptLanded verification. + +const asRecord = (value) => (value && typeof value === 'object' && !Array.isArray(value) ? value : null); +const asNonEmptyString = (value) => (typeof value === 'string' && value.trim() ? value.trim() : ''); +const asList = (value) => (Array.isArray(value) ? value : []); + +const SESSION_ID_PATTERN = /^[A-Za-z0-9_-]{4,128}$/; +const FETCH_TIMEOUT_MS = 15_000; +const IDLE_VERIFY_DELAY_MS = 500; +const LANDED_TIMEOUT_MS = 5_000; +const LANDED_POLL_MS = 150; + +const isValidSessionId = (value) => SESSION_ID_PATTERN.test(asNonEmptyString(value)); + +const extractSessionStatus = (payload) => { + if (payload.type !== 'session.status') return null; + const properties = asRecord(payload.properties) ?? {}; + const status = asRecord(properties.status) ?? {}; + const info = asRecord(properties.info) ?? {}; + const sessionId = asNonEmptyString(properties.sessionID); + const type = asNonEmptyString(status.type) || asNonEmptyString(info.type); + if (!sessionId || !type) return null; + const directory = typeof properties.directory === 'string' && properties.directory + ? properties.directory + : (typeof info.directory === 'string' ? info.directory : ''); + return { sessionId, type, directory }; +}; + +export function createSessionSteerRuntime({ + globalEventHub, + buildOpenCodeUrl, + getOpenCodeAuthHeaders, + broadcastGlobalUiEvent, + fetchImpl = fetch, +}) { + /** sessionId → { directive, directory } */ + const queued = new Map(); + let stopped = false; + + const openCodeFetch = async (fetchPath, { directory, method = 'GET', body, query } = {}) => { + const base = buildOpenCodeUrl(fetchPath, ''); + const params = new URLSearchParams(query || {}); + if (directory) params.set('directory', directory); + const search = params.toString(); + const url = search ? `${base}?${search}` : base; + const response = await fetchImpl(url, { + method, + headers: { + Accept: 'application/json', + ...(body ? { 'Content-Type': 'application/json' } : {}), + ...getOpenCodeAuthHeaders(), + }, + ...(body ? { body: JSON.stringify(body) } : {}), + signal: AbortSignal.timeout(FETCH_TIMEOUT_MS), + }); + if (!response.ok) { + throw new Error(`OpenCode ${method} ${fetchPath} failed with ${response.status}`); + } + return response.json().catch(() => null); + }; + + const isSessionIdle = async (sessionId, directory) => { + const statuses = asRecord(await openCodeFetch('/session/status', { directory }).catch(() => null)); + if (!statuses) return null; + const type = asRecord(statuses[sessionId])?.type; + if (type === 'busy' || type === 'retry') return false; + const messages = asList(await openCodeFetch(`/session/${encodeURIComponent(sessionId)}/message`, { + directory, + query: { limit: '2' }, + }).catch(() => null)); + if (!messages) return null; + const last = asRecord(asRecord(messages[messages.length - 1])?.info); + if (last?.role === 'assistant' && asRecord(last.time)?.completed === null) return false; + return true; + }; + + const latestUserMessageID = async (sessionId, directory) => { + const messages = asList(await openCodeFetch(`/session/${encodeURIComponent(sessionId)}/message`, { + directory, + query: { limit: '5' }, + }).catch(() => null)); + let latest = null; + for (const msg of messages) { + const info = asRecord(msg?.info); + if (info?.role !== 'user') continue; + if (!latest || (info.time?.created || 0) >= (latest.time?.created || 0)) latest = info; + } + return asNonEmptyString(latest?.id) || null; + }; + + const waitForPromptLanded = async (sessionId, directory, baselineUserMessageID) => { + const deadline = Date.now() + LANDED_TIMEOUT_MS; + for (;;) { + const latest = await latestUserMessageID(sessionId, directory); + if (!latest) return true; + if (latest !== baselineUserMessageID) return true; + if (Date.now() >= deadline) return false; + await new Promise((resolve) => setTimeout(resolve, LANDED_POLL_MS)); + } + }; + + const buildSteerPart = (directive) => ({ + type: 'text', + text: `\nSteer directive from the orchestrator: ${directive}\nHonor this directive for the remainder of this session. It overrides conflicting instructions.\n`, + synthetic: true, + }); + + const deliverDirective = async (sessionId, directory, directive) => { + const baseline = await latestUserMessageID(sessionId, directory); + await openCodeFetch(`/session/${encodeURIComponent(sessionId)}/prompt_async`, { + directory, + method: 'POST', + body: { parts: [buildSteerPart(directive)] }, + }); + return waitForPromptLanded(sessionId, directory, baseline); + }; + + const steerInterrupt = async (sessionId, directory, directive) => { + let wasBusy = false; + try { + const statuses = asRecord(await openCodeFetch('/session/status', { directory }).catch(() => null)); + const type = asRecord(statuses?.[sessionId])?.type; + wasBusy = type === 'busy' || type === 'retry'; + if (wasBusy) { + await openCodeFetch(`/session/${encodeURIComponent(sessionId)}/interrupt`, { + directory, + method: 'POST', + }); + await new Promise((resolve) => setTimeout(resolve, IDLE_VERIFY_DELAY_MS)); + const idle = await isSessionIdle(sessionId, directory); + if (idle === false) { + queued.set(sessionId, { directive, directory }); + return { accepted: true, mode: 'queue', interrupted: true }; + } + } + } catch { + queued.set(sessionId, { directive, directory }); + return { accepted: true, mode: 'queue', interrupted: false }; + } + try { + await deliverDirective(sessionId, directory, directive); + return { accepted: true, mode: 'interrupt', interrupted: wasBusy }; + } catch { + queued.set(sessionId, { directive, directory }); + return { accepted: true, mode: 'queue', interrupted: wasBusy }; + } + }; + + const steerQueue = async (sessionId, directory, directive) => { + queued.set(sessionId, { directive, directory }); + return { accepted: true, mode: 'queue', interrupted: false }; + }; + + const steer = async (sessionId, directory, directive, mode = 'interrupt') => { + if (!isValidSessionId(sessionId)) throw new TypeError('sessionId is invalid'); + const dir = asNonEmptyString(directory); + if (!dir) throw new TypeError('directory is required'); + const dirText = asNonEmptyString(directive); + if (!dirText) throw new TypeError('directive is required'); + if (mode === 'queue') return steerQueue(sessionId, dir, dirText); + return steerInterrupt(sessionId, dir, dirText); + }; + + const processPayload = (payload) => { + if (stopped) return; + const status = extractSessionStatus(payload); + if (!status || status.type !== 'idle') return; + const pending = queued.get(status.sessionId); + if (!pending) return; + queued.delete(status.sessionId); + const directory = pending.directory || status.directory; + deliverDirective(status.sessionId, directory, pending.directive) + .then(() => { + broadcastGlobalUiEvent?.({ + type: 'openchamber:steer-delivered', + properties: { + sessionID: status.sessionId, + directive: pending.directive, + mode: 'queue', + ts: Date.now(), + }, + }); + }) + .catch((error) => { + console.warn('[session-steer] queued delivery failed:', error?.message || error); + }); + }; + + const start = () => { + const unsubscribe = globalEventHub.subscribeEvent((event) => { + const raw = event?.payload; + const payload = raw?.payload && typeof raw.payload === 'object' ? raw.payload : raw; + processPayload(payload); + }); + return () => { unsubscribe(); }; + }; + + const stop = () => { + stopped = true; + queued.clear(); + }; + + return { steer, processPayload, start, stop }; +} + +export function registerSessionSteerRoutes(app, runtime) { + const respondError = (res, error, fallback) => { + const status = error instanceof TypeError ? 400 : (Number.isFinite(error?.status) ? error.status : 500); + res.status(status).json({ error: error?.message ?? fallback }); + }; + + app.post('/api/openchamber/session/:sessionID/steer', async (req, res) => { + try { + const sessionId = asNonEmptyString(req.params?.sessionID); + const directive = asNonEmptyString(req.body?.directive); + const mode = asNonEmptyString(req.body?.mode) || 'interrupt'; + if (!sessionId) return res.status(400).json({ error: 'sessionID is required' }); + if (!directive) return res.status(400).json({ error: 'directive is required' }); + if (mode !== 'interrupt' && mode !== 'queue') { + return res.status(400).json({ error: 'mode must be "interrupt" or "queue"' }); + } + const directory = asNonEmptyString(req.body?.directory) || asNonEmptyString(req.query?.directory) || ''; + const result = await runtime.steer(sessionId, directory, directive, mode); + return res.status(202).json(result); + } catch (error) { + return respondError(res, error, 'Failed to steer session'); + } + }); +} diff --git a/packages/web/server/lib/session-steer/runtime.test.js b/packages/web/server/lib/session-steer/runtime.test.js new file mode 100644 index 00000000..2ebd22c3 --- /dev/null +++ b/packages/web/server/lib/session-steer/runtime.test.js @@ -0,0 +1,153 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createSessionSteerRuntime } from './runtime.js'; + +const SESSION = 'ses_steer_test_1'; +const DIRECTORY = '/repo'; + +afterEach(() => { + vi.useRealTimers(); +}); + +const createOpenCode = () => { + const state = { + statuses: {}, + tail: [], + sent: [], + }; + const fetchImpl = vi.fn(async (url, init = {}) => { + const { pathname } = new URL(url); + const method = init.method ?? 'GET'; + if (pathname === '/session/status') return Response.json(state.statuses); + if (pathname.endsWith('/message')) return Response.json(state.tail); + if (method === 'POST' && pathname.endsWith('/prompt_async')) { + state.sent.push({ path: pathname, body: JSON.parse(init.body) }); + // Simulate prompt landing by adding a new user message + state.tail.push({ info: { role: 'user', id: `msg_landed_${Date.now()}`, time: { created: Date.now() } } }); + return new Response(null, { status: 204 }); + } + if (method === 'POST' && pathname.endsWith('/interrupt')) { + return new Response(null, { status: 204 }); + } + return new Response('not found', { status: 404 }); + }); + return { state, fetchImpl }; +}; + +const createRuntime = ({ openCode = createOpenCode() } = {}) => { + let eventHandler = () => {}; + const broadcasts = []; + const options = { + globalEventHub: { + subscribeEvent(handler) { eventHandler = handler; return () => {}; }, + }, + buildOpenCodeUrl: (fetchPath) => `http://opencode.test${fetchPath}`, + getOpenCodeAuthHeaders: () => ({}), + broadcastGlobalUiEvent: (event) => broadcasts.push(event), + fetchImpl: openCode.fetchImpl, + }; + const runtime = createSessionSteerRuntime(options); + return { + runtime, + openCode, + broadcasts, + emit: (payload, directory = DIRECTORY) => eventHandler({ payload, directory }), + }; +}; + +describe('session-steer runtime', () => { + it('rejects invalid sessionId', async () => { + const { runtime } = createRuntime(); + await expect(runtime.steer('', DIRECTORY, 'do something')).rejects.toThrow(TypeError); + }); + + it('rejects empty directive', async () => { + const { runtime } = createRuntime(); + await expect(runtime.steer(SESSION, DIRECTORY, '')).rejects.toThrow(TypeError); + }); + + it('rejects empty directory', async () => { + const { runtime } = createRuntime(); + await expect(runtime.steer(SESSION, '', 'do something')).rejects.toThrow(TypeError); + }); + + it('delivers directive to idle session via interrupt mode', async () => { + const { runtime, openCode } = createRuntime(); + openCode.state.statuses[SESSION] = { type: 'idle' }; + openCode.state.tail = [{ info: { role: 'user', id: 'msg_1', time: { created: 100 } } }]; + + const result = await runtime.steer(SESSION, DIRECTORY, 'STOP comparing worktrees', 'interrupt'); + expect(result.accepted).toBe(true); + expect(result.mode).toBe('interrupt'); + expect(openCode.state.sent.length).toBe(1); + expect(openCode.state.sent[0].body.parts[0].text).toContain('STOP comparing worktrees'); + }); + + it('queues directive for busy session', async () => { + const { runtime, openCode } = createRuntime(); + openCode.state.statuses[SESSION] = { type: 'busy' }; + + const result = await runtime.steer(SESSION, DIRECTORY, 'STOP comparing worktrees', 'interrupt'); + expect(result.accepted).toBe(true); + expect(result.mode).toBe('queue'); + }); + + it('queue mode stores directive without sending', async () => { + const { runtime, openCode } = createRuntime(); + openCode.state.statuses[SESSION] = { type: 'idle' }; + + const result = await runtime.steer(SESSION, DIRECTORY, 'wait for idle', 'queue'); + expect(result.accepted).toBe(true); + expect(result.mode).toBe('queue'); + expect(openCode.state.sent.length).toBe(0); + }); + + it('delivers queued directive on idle event', async () => { + const { runtime, openCode, emit, broadcasts } = createRuntime(); + runtime.start(); + openCode.state.statuses[SESSION] = { type: 'busy' }; + openCode.state.tail = []; + + // Queue the directive + await runtime.steer(SESSION, DIRECTORY, 'delivered on idle', 'queue'); + expect(openCode.state.sent.length).toBe(0); + + // Now the session goes idle + openCode.state.statuses[SESSION] = { type: 'idle' }; + openCode.state.tail = [{ info: { role: 'user', id: 'msg_1', time: { created: 100 } } }]; + + emit({ + type: 'session.status', + properties: { + sessionID: SESSION, + status: { type: 'idle' }, + }, + }); + + // Wait for async delivery to complete + await new Promise((resolve) => setTimeout(resolve, 300)); + + expect(openCode.state.sent.length).toBe(1); + expect(openCode.state.sent[0].body.parts[0].text).toContain('delivered on idle'); + const steerEvents = broadcasts.filter((e) => e.type === 'openchamber:steer-delivered'); + expect(steerEvents.length).toBe(1); + }); + + it('does not deliver when stopped', async () => { + const { runtime, openCode, emit } = createRuntime(); + runtime.start(); + openCode.state.statuses[SESSION] = { type: 'busy' }; + + await runtime.steer(SESSION, DIRECTORY, 'should not deliver', 'queue'); + runtime.stop(); + + openCode.state.statuses[SESSION] = { type: 'idle' }; + openCode.state.tail = [{ info: { role: 'user', id: 'msg_1', time: { created: 100 } } }]; + emit({ + type: 'session.status', + properties: { sessionID: SESSION, status: { type: 'idle' } }, + }); + + await new Promise((resolve) => setTimeout(resolve, 200)); + expect(openCode.state.sent.length).toBe(0); + }); +}); diff --git a/packages/web/server/lib/ui-auth/ui-auth.js b/packages/web/server/lib/ui-auth/ui-auth.js index 71232d6e..76d3e485 100644 --- a/packages/web/server/lib/ui-auth/ui-auth.js +++ b/packages/web/server/lib/ui-auth/ui-auth.js @@ -296,6 +296,7 @@ const isUrlAuthReadableHttpPath = (pathname) => { || pathname === '/api/global/event' || pathname === '/api/openchamber/events' || pathname === '/api/openchamber/realtime-proxy/sse' + || pathname === '/api/openchamber/agent-activity' || pathname === '/api/notifications/stream' || pathname === '/api/fs/raw' || pathname === '/api/fs/serve'