diff --git a/examples/opencode-plugin/index.mjs b/examples/opencode-plugin/index.mjs index b6c9e3c3a6..961ea6c083 100644 --- a/examples/opencode-plugin/index.mjs +++ b/examples/opencode-plugin/index.mjs @@ -83,7 +83,7 @@ export async function OpenVikingPlugin({ client, directory }) { }, dispose: async () => { - await sessionManager.flushAll({ commit: true }) + await sessionManager.flushAll({ commit: true, shutdown: true }) log("INFO", "plugin", "OpenViking plugin disposed") }, } diff --git a/examples/opencode-plugin/lib/config.mjs b/examples/opencode-plugin/lib/config.mjs index 6f5e8ad853..beb1259bc4 100644 --- a/examples/opencode-plugin/lib/config.mjs +++ b/examples/opencode-plugin/lib/config.mjs @@ -1,14 +1,9 @@ import fs from "fs" import path from "path" import { homedir } from "os" -import { buildUserAgent, readManifestVersion, resolveOpenVikingCredentials } from "./shared/credentials.mjs" +import { resolveOpenVikingCredentials } from "./shared/credentials.mjs" import { resolveEffectivePeerId } from "./shared/workspace-peer.mjs" -const USER_AGENT = buildUserAgent( - "opencode", - readManifestVersion(new URL("../package.json", import.meta.url)), -) - const DEFAULT_CONFIG = { endpoint: "http://127.0.0.1:1933", apiKey: "", @@ -42,6 +37,7 @@ const DEFAULT_CONFIG = { captureToolMaxChars: 2000, commitTokenThreshold: 20000, commitKeepRecentCount: 10, + periodicFlushIntervalMs: 60000, profileTokenBudget: 10000, resumeContextBudget: 32000, noAutoInject: false, @@ -148,6 +144,7 @@ function applyBehaviorConfig(config, fileConfig = {}) { "captureToolMaxChars", "commitTokenThreshold", "commitKeepRecentCount", + "periodicFlushIntervalMs", "profileTokenBudget", "resumeContextBudget", "noAutoInject", @@ -221,7 +218,6 @@ function normalizeConfig(config) { config.baseUrl = config.endpoint config.accountId = config.account config.userId = config.user - config.userAgent = USER_AGENT config.timeoutMs = normalizeNumber(config.timeoutMs, DEFAULT_CONFIG.timeoutMs, 1000, 300000) config.repoContext.cacheTtlMs = normalizeNumber( config.repoContext.cacheTtlMs, @@ -248,6 +244,7 @@ function normalizeConfig(config) { ? Math.max(0, Math.round(commitKeepRecentCount)) : DEFAULT_CONFIG.commitKeepRecentCount config.profileTokenBudget = Math.max(500, Math.round(Number(config.profileTokenBudget) || 10000)) + config.periodicFlushIntervalMs = Math.max(10000, Math.round(Number(config.periodicFlushIntervalMs) || 60000)) config.resumeContextBudget = Math.max(1024, Math.round(Number(config.resumeContextBudget) || 32000)) if (!Array.isArray(config.bypassSessionPatterns)) config.bypassSessionPatterns = [] diff --git a/examples/opencode-plugin/lib/memory-session.mjs b/examples/opencode-plugin/lib/memory-session.mjs index 6646a68965..ca5e3c30b5 100644 --- a/examples/opencode-plugin/lib/memory-session.mjs +++ b/examples/opencode-plugin/lib/memory-session.mjs @@ -27,10 +27,17 @@ export function createMemorySessionManager({ config, pluginRoot }) { const statePath = path.join(pluginRoot, "openviking-session-state.json") const oldSessionMapPath = path.join(pluginRoot, "openviking-session-map.json") let saveTimer = null + let flushTimer = null + let periodicFlushRunning = false + let periodicFlushPromise = null + let shuttingDown = false + let savePromise = Promise.resolve() + let saveCounter = 0 async function init() { if (config.autoCapture) await migrateLegacySessionMap() await loadState() + await sweepStaleTempFiles() const health = await fetchJSON(config, "/health", {}, { timeoutMs: 5000 }) if (health.ok) { await replayPending( @@ -38,6 +45,100 @@ export function createMemorySessionManager({ config, pluginRoot }) { (stage, data) => log("DEBUG", "pending", stage, data), ) } + startPeriodicFlush() + } + + function startPeriodicFlush() { + if (!config.autoCapture) return + if (shuttingDown) return + const intervalMs = Number(config.periodicFlushIntervalMs) > 0 ? config.periodicFlushIntervalMs : 60000 + if (flushTimer) clearInterval(flushTimer) + flushTimer = setInterval(() => { + // A tick may have been queued just before flushAll's clearInterval; skip it + // during teardown so a periodic flush cannot start concurrently with flushAll. + if (shuttingDown) return + runPeriodicFlush().catch((error) => { + log("ERROR", "session", "Periodic flush failed", { error: error?.message }) + }) + }, intervalMs) + if (typeof flushTimer.unref === "function") flushTimer.unref() + log("INFO", "session", "Periodic flush timer started", { intervalMs }) + } + + async function runPeriodicFlush() { + if (periodicFlushRunning) return + periodicFlushRunning = true + const done = (async () => { + try { + for (const [opencodeSessionId, state] of sessions.entries()) { + let hasPending = false + for (const message of state.messages.values()) { + if (!message.captured) { + hasPending = true + break + } + } + if (hasPending) { + await flushSession(opencodeSessionId, { commit: false, reason: "periodic" }) + } + } + } finally { + periodicFlushRunning = false + } + })() + periodicFlushPromise = done + return done + } + + async function sweepStaleTempFiles() { + // saveState writes to a unique temp file (`${statePath}.${pid}.${n}.tmp`) then + // renames it onto statePath. A crash between writeFile and rename leaves an + // orphan temp behind; sweep them on startup so they don't accumulate forever. + // We parse the PID from the filename and skip files whose PID is still alive + // to avoid removing a concurrent process's in-flight temp — but only if the + // file is recent, so a recycled PID can't shield an orphan temp forever. + const RECENT_TEMP_MS = 10 * 60 * 1000 + try { + const dir = path.dirname(statePath) + const base = path.basename(statePath) + const entries = await fs.promises.readdir(dir) + for (const name of entries) { + if (name.startsWith(`${base}.`) && name.endsWith(".tmp")) { + const fullPath = path.join(dir, name) + const match = name.match(/\.(\d+)\.\d+\.tmp$/) + if (match) { + let alive = false + try { + process.kill(Number(match[1]), 0) // probe: process is alive + alive = true + } catch (err) { + // EPERM (or other non-ESRCH) — process may be alive (possibly a + // different OS user); treat as alive to be safe. ESRCH — dead. + alive = err?.code !== "ESRCH" + } + if (alive) { + // A live PID may still be writing this temp. Only skip it while it + // is recent; a stale temp under a recycled PID is reclaimed below. + let recent = true + try { + const stat = await fs.promises.stat(fullPath) + recent = Date.now() - stat.mtimeMs < RECENT_TEMP_MS + } catch { + recent = false // vanished/unreadable — fall through to unlink + } + if (recent) continue + } + } + try { + await fs.promises.unlink(fullPath) + } catch { + // best effort; ignore files removed by a concurrent process + } + } + } + } catch (error) { + log("DEBUG", "persistence", "Temp sweep skipped", { error: error?.message }) + } } async function loadState() { @@ -64,17 +165,34 @@ export function createMemorySessionManager({ config, pluginRoot }) { } async function saveState() { + // Serialize saves within this process. The debounced saveTimer, flushSession + // and runPeriodicFlush can all call saveState concurrently; without chaining, + // two writes to the same temp path could interleave and corrupt the file that + // then gets renamed onto the real state path. + const run = savePromise.then(runSaveState, runSaveState) + savePromise = run.catch(() => {}) + return run + } + + async function runSaveState() { + const tempPath = `${statePath}.${process.pid}.${saveCounter++}.tmp` try { const persisted = {} for (const [opencodeSessionId, state] of sessions.entries()) { persisted[opencodeSessionId] = serializeSessionState(state) } - const tempPath = `${statePath}.tmp` await fs.promises.writeFile(tempPath, JSON.stringify({ version: 2, sessions: persisted, lastSaved: Date.now() }, null, 2), "utf8") await fs.promises.rename(tempPath, statePath) log("DEBUG", "persistence", "Session state saved", { count: sessions.size }) } catch (error) { log("ERROR", "persistence", "Failed to save session state", { error: error?.message }) + // Best-effort cleanup so a failed rename does not leave an orphan temp file. + try { + await fs.promises.unlink(tempPath) + } catch {} + // Rethrow so callers awaiting saveState() can observe persistent failures + // (e.g. ENOSPC/EACCES). Fire-and-forget callers already attach a .catch. + throw error } } @@ -242,36 +360,75 @@ export function createMemorySessionManager({ config, pluginRoot }) { debouncedSaveState() } - async function flushAll({ commit = false } = {}) { + async function flushAll({ commit = false, shutdown = false } = {}) { + if (shutdown) shuttingDown = true if (saveTimer) { clearTimeout(saveTimer) saveTimer = null } + if (flushTimer) { + clearInterval(flushTimer) + flushTimer = null + } + // Drain any in-flight periodic flush so it cannot interleave with the loop + // below and make commit ordering nondeterministic at teardown. + if (periodicFlushPromise) { + try { + await periodicFlushPromise + } catch { + // periodic flush errors are already logged at their source + } + } for (const sessionId of sessions.keys()) { await flushSession(sessionId, { commit, reason: "flushAll" }) } await saveState() + // If this was not a shutdown, keep periodic flushing alive. + if (!shutdown) startPeriodicFlush() } - async function flushSession(opencodeSessionId, { commit = false, reason = "manual" } = {}) { + async function flushSession(opencodeSessionId, { commit = false, reason = "manual", skipThreshold = false } = {}) { if (!opencodeSessionId) return false const state = sessions.get(opencodeSessionId) if (!state) return false - const added = await flushPendingMessages(opencodeSessionId, state) - if (commit && config.autoCapture) { - await commitOvSession(state.ovSessionId, { force: true, reason }) - } else if (added > 0) { - await maybeCommitByThreshold(state) + // Serialize overlapping flushes on the same session. flushPendingMessages + // only marks messages captured=true after the network send resolves, so two + // concurrent flushes (periodic timer + session.idle, or a slow flush the + // timer did not await) would read the same captured=false batch and send it + // twice. Chaining on state.flushing forces them to run one-after-another, + // so each subsequent flush re-reads the post-send captured state. + const run = async () => { + const added = await flushPendingMessages(opencodeSessionId, state) + if (commit && config.autoCapture) { + await commitOvSession(state.ovSessionId, { force: true, reason }) + } else if (added > 0 && !skipThreshold) { + await maybeCommitByThreshold(state) + } + await saveState() + return true } - await saveState() - return true + + const previous = state.flushing || Promise.resolve() + const current = previous.then(run, run) + // Store the same promise the next flush will chain on. Swallow the rejection + // on the gate branch (callers still observe it via `current`) so a failed run + // does not become an unhandled rejection when nothing chains onto it. We do + // NOT reset state.flushing afterwards: chaining on an already-settled promise + // is effectively free, and an auto-reset opens a window where a rapidly queued + // flush sees `undefined` and starts in parallel instead of serializing. + state.flushing = current.catch(() => {}) + return current } async function commitSession(sessionId, opencodeSessionId, abortSignal) { if (opencodeSessionId) { - const state = sessions.get(opencodeSessionId) - if (state) await flushPendingMessages(opencodeSessionId, state) + // Route through flushSession so the send goes through the same + // per-session serialization gate. Calling flushPendingMessages directly + // would bypass state.flushing and could double-send a batch that a + // concurrent periodic/idle flush is already sending. skipThreshold avoids + // a redundant threshold commit here since we force-commit right below. + await flushSession(opencodeSessionId, { commit: false, reason: "tool", skipThreshold: true }) } return commitOvSession(sessionId, { force: true, abortSignal, reason: "tool" }) } diff --git a/examples/opencode-plugin/tests/memory-session.test.mjs b/examples/opencode-plugin/tests/memory-session.test.mjs index 679ef58dac..2578799df5 100644 --- a/examples/opencode-plugin/tests/memory-session.test.mjs +++ b/examples/opencode-plugin/tests/memory-session.test.mjs @@ -1,7 +1,7 @@ import test from "node:test" import assert from "node:assert/strict" import { createServer } from "node:http" -import { mkdtemp, rm } from "node:fs/promises" +import { mkdtemp, rm, readdir, writeFile, utimes } from "node:fs/promises" import { tmpdir } from "node:os" import { join } from "node:path" import { createMemorySessionManager } from "../lib/memory-session.mjs" @@ -15,8 +15,24 @@ async function withTempDir(prefix, fn) { } } -async function withCaptureServer(fn) { +// Find a PID that is definitely not running. process.kill(pid, 0) throws +// ESRCH for a non-existent PID; probe upward from a large base so the test +// does not rely on a hard-coded value that could be alive on high-pid_max +// systems. +function findDeadPid() { + for (let pid = 999983; pid < 4194304; pid += 104729) { + try { + process.kill(pid, 0) + } catch (err) { + if (err?.code === "ESRCH") return pid + } + } + throw new Error("could not find a dead PID") +} + +async function withCaptureServer(fn, options = {}) { const requests = [] + const { onBatch } = options const server = createServer(async (req, res) => { let body = "" req.setEncoding("utf8") @@ -27,6 +43,9 @@ async function withCaptureServer(fn) { if (req.url === "/health") { res.end(JSON.stringify({ status: "ok" })) } else if (req.url?.startsWith("/api/v1/sessions/") && req.url.endsWith("/messages/batch")) { + // Optional hook lets a test hold a batch open so it can observe an + // in-flight send (e.g. a periodic flush) racing another operation. + if (typeof onBatch === "function") await onBatch(req.url) res.end(JSON.stringify({ status: "ok", result: { accepted: true } })) } else if (req.url?.startsWith("/api/v1/sessions/")) { res.end(JSON.stringify({ status: "ok", result: { pending_tokens: 0 } })) @@ -213,3 +232,163 @@ test("assistant messages are captured even when finish is not stop", async () => }) }) }) + +async function seedUserMessage(manager, sessionId, messageId, text) { + await manager.handleEvent({ + type: "message.updated", + properties: { info: { id: messageId, sessionID: sessionId, role: "user" } }, + }) + await manager.handleEvent({ + type: "message.part.updated", + properties: { + part: { id: `${messageId}-part`, messageID: messageId, sessionID: sessionId, type: "text", text }, + }, + }) +} + +test("concurrent flushSession on the same session sends each pending batch once", async () => { + await withCaptureServer(async ({ endpoint, requests }) => { + await withTempDir("ov-oc-session-", async (dir) => { + const manager = createMemorySessionManager({ config: baseConfig(endpoint), pluginRoot: dir }) + await manager.init() + await manager.handleEvent({ type: "session.created", properties: { info: { id: "oc-gate" } } }) + await seedUserMessage(manager, "oc-gate", "msg-gate-1", "Gate serialization must prevent double sends.") + + // Fire two overlapping flushes without awaiting the first. + await Promise.all([ + manager.flushSession("oc-gate", { commit: false }), + manager.flushSession("oc-gate", { commit: false }), + ]) + + const batches = requests.filter((r) => r.url === "/api/v1/sessions/oc-oc-gate/messages/batch") + assert.equal(batches.length, 1, "overlapping flushes must not double-send the same message") + await manager.flushAll({ commit: false }) + }) + }) +}) + +test("flushAll shutdown drains an in-flight periodic flush before its own loop", async () => { + // Gate the first batch send so a periodic flush is genuinely mid-flight when + // flushAll({shutdown:true}) runs. This exercises the drain branch + // (`if (periodicFlushPromise) await periodicFlushPromise`) in flushAll and + // proves the shutdown waits for in-flight periodic work instead of racing it. + let releaseBatch + const batchGate = new Promise((resolve) => { + releaseBatch = resolve + }) + let firstBatchSeen + const firstBatch = new Promise((resolve) => { + firstBatchSeen = resolve + }) + let batchCount = 0 + + await withCaptureServer( + async ({ endpoint, requests }) => { + await withTempDir("ov-oc-session-", async (dir) => { + const manager = createMemorySessionManager({ + config: { ...baseConfig(endpoint), periodicFlushIntervalMs: 10 }, + pluginRoot: dir, + }) + await manager.init() + await manager.handleEvent({ type: "session.created", properties: { info: { id: "oc-drain" } } }) + await seedUserMessage(manager, "oc-drain", "msg-drain-1", "Teardown must drain periodic work.") + + // Wait until the periodic timer has actually started a batch send that + // is now blocked in the gate, then tear down while it is in-flight. + await firstBatch + const shutdown = manager.flushAll({ commit: true, shutdown: true }) + // Let the in-flight periodic batch complete; flushAll must have awaited it. + releaseBatch() + await shutdown + + const batches = requests.filter((r) => r.url === "/api/v1/sessions/oc-oc-drain/messages/batch") + assert.equal(batches.length, 1, "shutdown must not double-send the in-flight periodic batch") + const commits = requests.filter((r) => r.method === "POST" && r.url === "/api/v1/sessions/oc-oc-drain/commit") + assert.equal(commits.length, 1, "shutdown with commit:true must commit exactly once") + // The gated batch must have been fully sent before the commit fired. + const order = requests + .filter((r) => r.method === "POST" && r.url?.startsWith("/api/v1/sessions/oc-oc-drain/")) + .map((r) => r.url) + assert.deepEqual( + order, + ["/api/v1/sessions/oc-oc-drain/messages/batch", "/api/v1/sessions/oc-oc-drain/commit"], + "the drained periodic batch must complete before the shutdown commit", + ) + }) + }, + { + onBatch: async (url) => { + if (url === "/api/v1/sessions/oc-oc-drain/messages/batch") { + batchCount += 1 + if (batchCount === 1) { + firstBatchSeen() + await batchGate + } + } + }, + }, + ) +}) + +test("commitSession routes the send through the per-session gate", async () => { + await withCaptureServer(async ({ endpoint, requests }) => { + await withTempDir("ov-oc-session-", async (dir) => { + const manager = createMemorySessionManager({ config: baseConfig(endpoint), pluginRoot: dir }) + await manager.init() + await manager.handleEvent({ type: "session.created", properties: { info: { id: "oc-tool" } } }) + await seedUserMessage(manager, "oc-tool", "msg-tool-1", "Tool commit must flush pending first.") + + await manager.commitSession("oc-oc-tool", "oc-tool") + + const batches = requests.filter((r) => r.url === "/api/v1/sessions/oc-oc-tool/messages/batch") + assert.equal(batches.length, 1, "commitSession must flush pending messages before committing") + const order = requests + .filter((r) => r.method === "POST" && r.url?.startsWith("/api/v1/sessions/oc-oc-tool/")) + .map((r) => r.url) + assert.deepEqual( + order, + ["/api/v1/sessions/oc-oc-tool/messages/batch", "/api/v1/sessions/oc-oc-tool/commit"], + "the batch send must happen before the commit", + ) + }) + }) +}) + +test("startup sweep reclaims orphan temp files from a dead PID but keeps live/recent ones", async () => { + await withCaptureServer(async ({ endpoint }) => { + await withTempDir("ov-oc-session-", async (dir) => { + const statePath = join(dir, "openviking-session-state.json") + // Dead PID: derived at runtime so the test is not flaky on systems with + // a high pid_max where a hard-coded large PID could be alive. + const deadPid = findDeadPid() + const deadTmp = `${statePath}.${deadPid}.0.tmp` + // Live PID (this test process), recent mtime -> must be kept. + const liveRecentTmp = `${statePath}.${process.pid}.0.tmp` + // Live PID but stale mtime (simulated recycled PID) -> must be reclaimed. + const liveStaleTmp = `${statePath}.${process.pid}.1.tmp` + await writeFile(deadTmp, "orphan") + await writeFile(liveRecentTmp, "in-flight") + await writeFile(liveStaleTmp, "recycled-pid-orphan") + const stale = new Date(Date.now() - 30 * 60 * 1000) + await utimes(liveStaleTmp, stale, stale) + + const manager = createMemorySessionManager({ config: baseConfig(endpoint), pluginRoot: dir }) + await manager.init() + + const remaining = (await readdir(dir)).filter((n) => n.endsWith(".tmp")) + assert.ok( + !remaining.includes(`openviking-session-state.json.${deadPid}.0.tmp`), + "dead-PID orphan must be removed", + ) + assert.ok( + remaining.includes(`openviking-session-state.json.${process.pid}.0.tmp`), + "recent live-PID temp must be kept", + ) + assert.ok( + !remaining.includes(`openviking-session-state.json.${process.pid}.1.tmp`), + "stale live-PID temp (recycled PID) must be reclaimed", + ) + await manager.flushAll({ commit: false, shutdown: true }) + }) + }) +})