From b9cd204c2680684e84a26dae2dcdc42c37de89b9 Mon Sep 17 00:00:00 2001 From: Yuanqing Zhao <2604121+yuanqingz@users.noreply.github.com> Date: Mon, 10 Aug 2026 01:53:33 -0700 Subject: [PATCH] fix(memory-plugin): harden recall and capture state --- .../scripts/shared/recall-core.mjs | 13 +- examples/codex-memory-plugin/DESIGN.md | 58 +- examples/codex-memory-plugin/README.md | 57 +- .../scripts/auto-capture.mjs | 119 ++-- .../scripts/auto-recall.mjs | 321 +++++++++-- .../scripts/auto-recall.test.mjs | 447 ++++++++++++++- .../scripts/pre-compact-capture.mjs | 174 +++--- .../scripts/pre-compact-capture.test.mjs | 171 ++++++ .../scripts/recall-compressor-profile.mjs | 109 +++- .../recall-compressor-profile.test.mjs | 63 ++- .../scripts/session-start-commit.mjs | 76 ++- .../scripts/session-state.mjs | 535 +++++++++++++++++- .../scripts/session-state.test.mjs | 448 +++++++++++++++ .../scripts/shared/recall-core.mjs | 13 +- .../servers/experience-tools.mjs | 132 ++++- .../servers/experience-tools.test.mjs | 144 ++++- .../skills/ov-experience-memory/SKILL.md | 36 +- .../memory-plugin-shared/lib/recall-core.mjs | 13 +- .../memory-plugin-shared/recall-core.test.mjs | 42 +- .../lib/shared/recall-core.mjs | 13 +- .../shared/recall-core.mjs | 13 +- examples/skills/ov-experience-memory/SKILL.md | 36 +- .../scripts/shared/recall-core.mjs | 13 +- 23 files changed, 2685 insertions(+), 361 deletions(-) create mode 100644 examples/codex-memory-plugin/scripts/pre-compact-capture.test.mjs create mode 100644 examples/codex-memory-plugin/scripts/session-state.test.mjs diff --git a/examples/claude-code-memory-plugin/scripts/shared/recall-core.mjs b/examples/claude-code-memory-plugin/scripts/shared/recall-core.mjs index a6e94eef5a..8c53dcea96 100644 --- a/examples/claude-code-memory-plugin/scripts/shared/recall-core.mjs +++ b/examples/claude-code-memory-plugin/scripts/shared/recall-core.mjs @@ -40,7 +40,8 @@ function scaleQuotas(limit, weights) { const order = Object.keys(weights); const quotas = Object.fromEntries(order.map((key) => [key, 0])); if (slots < order.length) { - for (const key of order) quotas[key] = 1; + const priority = [...order].sort((a, b) => weights[b] - weights[a]); + for (const key of priority.slice(0, slots)) quotas[key] = 1; return quotas; } @@ -61,10 +62,12 @@ function scaleQuotas(limit, weights) { } function legacyMemoryQuotas(limit) { - return { - ...scaleQuotas(limit, { events: 10, entities: 10, preferences: 3 }), - experiences: 0, - }; + return scaleQuotas(limit, { + events: 10, + entities: 10, + experiences: 3, + preferences: 3, + }); } function codingQuotas(limit) { diff --git a/examples/codex-memory-plugin/DESIGN.md b/examples/codex-memory-plugin/DESIGN.md index ee0e7827d4..da719c1848 100644 --- a/examples/codex-memory-plugin/DESIGN.md +++ b/examples/codex-memory-plugin/DESIGN.md @@ -17,7 +17,7 @@ events imply "context for a particular codex `session_id` is gone". memory extractor) at session-end-equivalent moments. `/messages` auto-creates the OV session, so the plugin does not call session create. - **State file** — `~/.openviking/codex-plugin-state/.json`, - shape `{ codexSessionId, ovSessionId, capturedTurnCount, createdAt, lastUpdatedAt }`. + shape `{ codexSessionId, ovSessionId, capturedTurnCount, revision, createdAt, lastUpdatedAt }`. - **Active window** — state files whose `lastUpdatedAt` is within `ACTIVE_WINDOW_MS` (default 2 min) of "now". Used to detect "the codex session that just ended". @@ -228,6 +228,7 @@ OV session id, while commits create additional archives under that session. "codexSessionId": "0193af...", // codex thread id "ovSessionId": "cx-0193af...-or-null", // null means "committed, awaiting next Stop" "capturedTurnCount": 7, // turns from transcript already appended + "revision": 4, // monotonic across save, clear, and recreation "createdAt": 1715000000000, "lastUpdatedAt": 1715000300000 } @@ -238,7 +239,37 @@ Legacy state files from earlier plugin versions may still contain a UUID next resolve. The migration window for preserving old UUID sessions has closed. -State files are atomic-write (tmpfile + rename) to survive crash mid-write. +State files use a unique tmpfile + rename and recover the newest complete, +session-matching tmp after a crash. A monotonic revision counter lives beside +the permanent per-session lock baton; `clear` advances that tombstone before +removing the final file, so state recreation cannot reset the generation or +make a stale SessionStart snapshot look current. + +Stop, PreCompact, and SessionStart run each same-session state/remote-I/O +lifecycle under that cross-process baton. Acquire atomically renames +`available` to a unique `owner-` path; release and dead-owner recovery +move only that unique source, avoiding pathname ABA. Automatic abandoned-owner +recovery is intentionally limited to the same host and Linux PID namespace, +where machine id, PID-namespace identity, boot id, namespace-local PID, and +`/proc` start time can prove that the exact owner process is gone. Legacy claims +without namespace identity and claims from another host or container namespace +are never stolen on elapsed time alone because OpenViking does not provide a +fencing token for the remote session operations; operators sharing one state +directory across hosts or PID namespaces must recover an abandoned baton +explicitly. Automatic dead-owner recovery is Linux-only; other platforms do not +have this full identity tuple and therefore fail safe to manual recovery. +Ownerless contender claim metadata is pruned on a later acquisition only when +the same Linux identity proof confirms that its process is dead; otherwise it is +left for manual cleanup under the same cross-host/container safety rule. + +The ordinary state-lock wait is 60 seconds so an in-flight append or commit is +not mistaken for abandonment. The default Stop hook launches its writer in a +detached process, so Codex's 30-second Stop deadline does not bound that +writer. If an explicitly synchronous Stop writer is killed, the durable +revision/tmp recovery above lets a later hook resume from the last accepted +batch. PreCompact instead waits at most 20 seconds by default and reports a +deferred capture on contention; this preserves most of its 60-second hook +deadline for catch-up and commit work without discarding recoverable progress. ## Configuration @@ -247,6 +278,8 @@ Env var overrides for tuning without rebuilding: | Var | Default | Purpose | |---|---|---| | `OPENVIKING_CODEX_STATE_DIR` | `~/.openviking/codex-plugin-state` | state file dir | +| `OPENVIKING_CODEX_STATE_LOCK_TIMEOUT_MS` | `60000` | maximum wait for the same-session cross-process baton | +| `OPENVIKING_PRECOMPACT_STATE_LOCK_TIMEOUT_MS` | `20000` | PreCompact-specific baton wait; timeout reports deferral and preserves state | | `OPENVIKING_CODEX_ACTIVE_WINDOW_MS` | `120000` (2 min) | rule-3 active window | | `OPENVIKING_CODEX_IDLE_TTL_MS` | `1800000` (30 min) | idle sweep TTL | | `OPENVIKING_RECALL_TIMEOUT_MS` | `120000` (2 min) | whole UserPromptSubmit auto-recall deadline | @@ -254,7 +287,7 @@ Env var overrides for tuning without rebuilding: | `OPENVIKING_RECALL_COMPRESS_MODEL` | unset | custom first-choice compressor model; `off` disables compression | | `OPENVIKING_RECALL_COMPRESS_THINKING` | unset | custom `model_reasoning_effort`; `default` means omit override; alias `OPENVIKING_RECALL_COMPRESS_REASONING_EFFORT` | | `OPENVIKING_RECALL_COMPRESS_DETECT_ON_STARTUP` | `1` | recreate/cache compressor profile during every `SessionStart` | -| `OPENVIKING_RECALL_COMPRESS_DETECT_TIMEOUT_MS` | `15000` | per-candidate compressor probe timeout | +| `OPENVIKING_RECALL_COMPRESS_DETECT_TIMEOUT_MS` | `15000` | compatibility setting for older installers; current detection does not launch a startup probe | | `OPENVIKING_RECALL_COMPRESS_DETECT_TTL_MS` | `604800000` (7 days) | cache TTL used by `UserPromptSubmit` reads | | `OPENVIKING_RESUME_ARCHIVE_INJECT` | `1` | inject latest archive summary on `source=resume` when no live OV session is open | | `OPENVIKING_RESUME_ARCHIVE_TOKEN_BUDGET` | `32000` | token budget for `/sessions/{id}/context` on resume | @@ -288,13 +321,15 @@ codex -m -c 'model_reasoning_effort="low"' exec ... `thinking=default` omits the `model_reasoning_effort` override. This is important for model families whose default effort is tuned by Codex. -Model availability is re-probed at every `SessionStart`, not in every -`UserPromptSubmit`. Recreating the profile on each session start catches -cross-session env/config changes. The detector writes +Model availability is resolved from Codex's model catalogue when the cached +profile is missing, expired, or marked runtime-failed; it is not probed with a +child process on every `SessionStart` or `UserPromptSubmit`. The detector writes `recall-compressor-profile.json` under `OPENVIKING_CODEX_STATE_DIR` and -auto-recall reads that cache. Cache misses in auto-recall use the first -candidate directly and fall back to deterministic digest if `codex exec` -fails. +auto-recall reads that cache. A prompt tries at most two distinct candidates +inside one shared timeout. If every attempt fails, it injects nothing and +records those failed models; it does not turn unverified candidates into a +deterministic digest. Explicitly configured `off` still uses deterministic +formatting without `codex exec`. Fallback order: @@ -302,7 +337,10 @@ Fallback order: `OPENVIKING_RECALL_COMPRESS_THINKING`) 2. `gpt-5.3-codex-spark`, thinking `default` 3. `gpt-5.6-luna`, thinking `low` -4. off (deterministic digest, no child `codex exec`) + +When a configured model is distinct from both defaults, only it and the +primary default fit in the first prompt's two-attempt budget; an untried +fallback remains eligible on a later prompt. Configured `off` (`OPENVIKING_RECALL_COMPRESS=0`, model `off`, or thinking `off`) skips all probing and writes a disabled profile. diff --git a/examples/codex-memory-plugin/README.md b/examples/codex-memory-plugin/README.md index e3dd023ac9..d898911f67 100644 --- a/examples/codex-memory-plugin/README.md +++ b/examples/codex-memory-plugin/README.md @@ -204,23 +204,31 @@ On `resume`, the script skips commit/sweep. It still injects the profile block. Codex injects `additionalContext` into the model turn, so memories arrive without an extra tool call. By default the hook runs a Codex compression pass over recalled candidates before injection, dropping weakly-related memories and preserving only a short digest. If the compressor returns `NO_RELEVANT_MEMORY`, empty text, or non-digest chatter, the hook emits `{}` and injects nothing. The whole hook has its own `OPENVIKING_RECALL_TIMEOUT_MS` deadline (default 120s); the bundled `hooks.json` gives Codex 130s so the script can return `{}` before Codex kills it. Digests may keep `viking://` source URIs and point the model at the OpenViking MCP `read`/`search` tools for details when the inline bullet is intentionally short. The outer `` wrapper is deterministic, not compressor-generated; capture strips it to distinguish recalled context from the user's prompt. Set `OPENVIKING_RECALL_COMPRESS=0` to fall back to deterministic short formatting. -The compressor profile is recreated on every `SessionStart` and cached under `OPENVIKING_CODEX_STATE_DIR` so cross-session config changes are picked up but each `UserPromptSubmit` does not probe models. Default fallback order: +The compressor profile is resolved on `SessionStart` when the cache is missing, +expired, or marked runtime-failed, then cached under +`OPENVIKING_CODEX_STATE_DIR`. A prompt tries at most two distinct model +candidates within one shared timeout budget. If every runtime attempt fails, +recall fails closed and injects nothing; a weak deterministic digest is never +substituted merely because the relevance check failed. Explicitly disabling +compression still uses the deterministic formatter. Candidate order is: 1. configured `OPENVIKING_RECALL_COMPRESS_MODEL` + `OPENVIKING_RECALL_COMPRESS_THINKING` 2. `gpt-5.3-codex-spark` with thinking `default` 3. `gpt-5.6-luna` with thinking `low` -4. off (deterministic digest, no `codex exec` compression) + +If a configured model occupies the first slot, the remaining candidate can be +retried on a later prompt after failed models are recorded. Config knobs: | Env var | Default | Meaning | |---|---|---| -| `OPENVIKING_RECALL_LIMIT` | `10` | Legacy quota-scaling input; explicit values are converted to six coding quotas, not enforced as a final result cap. | +| `OPENVIKING_RECALL_LIMIT` | `10` | Legacy quota-scaling input; an explicit value is distributed across coding categories and the resulting quotas sum to that value. | | `OPENVIKING_RECALL_COMPRESS` | `1` | Set `0` / `off` to disable `codex exec` compression. | | `OPENVIKING_RECALL_COMPRESS_MODEL` | unset | Custom first-choice compressor model. Set `off` to disable compression. | | `OPENVIKING_RECALL_COMPRESS_THINKING` | unset | Custom `model_reasoning_effort`; `default` omits the Codex config override. Alias: `OPENVIKING_RECALL_COMPRESS_REASONING_EFFORT`. | | `OPENVIKING_RECALL_COMPRESS_DETECT_ON_STARTUP` | `1` | Recreate/cache compressor profile in `SessionStart`. | -| `OPENVIKING_RECALL_COMPRESS_DETECT_TIMEOUT_MS` | `15000` | Per-candidate startup probe timeout. | +| `OPENVIKING_RECALL_COMPRESS_DETECT_TIMEOUT_MS` | `15000` | Compatibility setting retained for older installers; current detection reads the local model catalogue and does not launch a startup probe. | | `OPENVIKING_RECALL_COMPRESS_DETECT_TTL_MS` | `604800000` | Cache TTL used by `UserPromptSubmit` when reading the latest profile. | | `OPENVIKING_RECALL_MAX_TOKENS` | `1600` | Token budget the server assembles the context block within, independent of the local compressor input limit. | | `OPENVIKING_RECALL_DEDUP_TURNS` | `5` | Cross-turn cooldown: URIs served in the last N turns are skipped. | @@ -233,9 +241,9 @@ that endpoint fall back to `/api/v1/search/recall`, and that outcome is cached s only the first turn pays for the probe. Server-owned Context defaults are omitted unless explicitly configured, so the plugin follows the server instead of copying values such as `limit=10` or `max_tokens=1600`. An explicit legacy `recallLimit` -is converted to per-category coding quotas, not a final result cap. Values -from 1 through 5 therefore produce an effective total quota of 6, one retrieval -slot for each coding domain. Local `codex exec` compression is +is converted to per-category coding quotas whose total equals the configured +value; when the value is smaller than the number of categories, only the +highest-priority categories receive a slot. Local `codex exec` compression is unchanged and still runs on top of whichever path answered. Client-side knobs can also live in `~/.openviking/ovcli.conf` under @@ -249,6 +257,15 @@ defaults. After a successful append, Stop reads the session meta and commits when `pending_tokens >= OPENVIKING_COMMIT_TOKEN_THRESHOLD` (default `20000`). Threshold commits pass `keep_recent_count=OPENVIKING_COMMIT_KEEP_RECENT_COUNT` (default `10`) so the newest turns remain live for continuity while older context is archived and extracted. `PreCompact` still commits everything before compaction. +Same-session state transitions are serialized across hook processes and every +accepted append batch advances durable state before the next batch starts. The +default Stop path launches its writer asynchronously, so Codex's 30-second +hook deadline does not cut short a writer waiting on the 60-second state-lock +budget. If asynchronous writing is disabled and Codex terminates a synchronous +writer, the next hook recovers a complete temporary state file, resumes from +the last recorded turn count, and safely reclaims a confirmed-dead owner in the +same host and Linux PID namespace. + ### PreCompact (deterministic commit) `pre-compact-capture.mjs`: @@ -257,6 +274,32 @@ After a successful append, Stop reads the session meta and commits when `pending 2. Commit the long-lived OV session so the extractor runs against the full pre-compact transcript 3. Reset `ovSessionId` to `null` so the next `Stop` re-derives the same `cx-` and appends the post-compact half under that deterministic OV session id +PreCompact uses a shorter state-lock wait (20 seconds by default; override +with `OPENVIKING_PRECOMPACT_STATE_LOCK_TIMEOUT_MS`) so lock contention cannot +silently consume its entire 60-second hook deadline. On timeout it emits a +`systemMessage`, leaves durable capture progress untouched, and lets a later +hook catch up. + +If multiple hosts share `OPENVIKING_CODEX_STATE_DIR`, the plugin never steals +a different host's owner baton based only on elapsed time: without a server +fencing token that could overlap remote writes. After a host failure, an +operator must first verify that the remote owner process is gone and then +recover the abandoned baton manually. Prefer a host-local state directory +unless that operational coordination is available. + +The same fail-safe rule applies when multiple containers share that directory. +Linux automatic recovery requires matching machine-id and PID namespace +identity; legacy claims or claims from another PID namespace are left for +manual recovery rather than treating an invisible container process as dead. +Automatic dead-owner recovery is Linux-only; other platforms also require +manual recovery because hostname and PID alone cannot prove host/process +identity safely. + +The same rule applies to a contender that crashes before acquiring the baton: +a later acquisition removes its ownerless claim metadata only when the same +Linux identity tuple proves that process is dead. Unverifiable remote, legacy, +container, or non-Linux claims remain available for manual cleanup. + ### Known gap: SIGTERM / Ctrl+C / `/exit` are silent Codex fires no hook on process exit. `/compact` is the only fully-deterministic "context disappearing" signal. If you `/exit` without `/compact`, the OV session for that codex session_id stays open. Two fallbacks recover the orphan: diff --git a/examples/codex-memory-plugin/scripts/auto-capture.mjs b/examples/codex-memory-plugin/scripts/auto-capture.mjs index 91758854a2..566219f17a 100644 --- a/examples/codex-memory-plugin/scripts/auto-capture.mjs +++ b/examples/codex-memory-plugin/scripts/auto-capture.mjs @@ -31,7 +31,7 @@ import { } from "./capture-utils.mjs"; import { loadConfig } from "./config.mjs"; import { createLogger } from "./debug-log.mjs"; -import { loadState, resolveOvSessionId, saveState } from "./session-state.mjs"; +import { resolveOvSessionId, withStateTransaction } from "./session-state.mjs"; import { maybeDetach, readHookStdin } from "./shared/async-writer.mjs"; import { sendSessionMessages } from "./shared/batch-send.mjs"; import { resolveEffectivePeerId } from "./shared/workspace-peer.mjs"; @@ -117,7 +117,7 @@ async function readTranscriptTurns(transcriptPath) { } } -async function appendTurns(ovSessionId, turns, state) { +async function appendTurns(ovSessionId, turns, state, save) { const payloads = turns.map((turn) => { const body = turn.parts?.length ? { role: turn.role, parts: turn.parts } @@ -128,7 +128,7 @@ async function appendTurns(ovSessionId, turns, state) { const r = await sendSessionMessages(fetchJSONRes, ovSessionId, payloads, { onSent: async (n) => { state.capturedTurnCount += n; - await saveState(state); + await save(state); }, }); return r.sent; @@ -189,73 +189,74 @@ async function main() { const sessionId = input.session_id || "unknown"; const transcriptPath = input.transcript_path || null; - const state = await loadState(sessionId); - activePeerId = cfg.peerId || state.workspacePeerId || resolveEffectivePeerId({ cfg, cwd: process.cwd() }).peerId; - log("start", { sessionId, transcriptPath, hasPeer: Boolean(activePeerId) }); + await withStateTransaction(sessionId, async ({ state, save }) => { + activePeerId = cfg.peerId || state.workspacePeerId || resolveEffectivePeerId({ cfg, cwd: process.cwd() }).peerId; + log("start", { sessionId, transcriptPath, hasPeer: Boolean(activePeerId) }); - const health = await fetchJSON("/health"); - if (!health) { - logError("health_check", "server unreachable or unhealthy"); - noop(); - return; - } + const health = await fetchJSON("/health"); + if (!health) { + logError("health_check", "server unreachable or unhealthy"); + noop(); + return; + } - const allTurns = await readTranscriptTurns(transcriptPath); + const allTurns = await readTranscriptTurns(transcriptPath); - // Post-compact transcript-shrink defense: codex's /compact may rewrite or - // truncate transcript_path. If allTurns has fewer entries than we cached, - // our slice math would underflow and silently drop turns. Reset the - // counter so the next slice captures everything in the new transcript. - // See DESIGN.md "Post-compact transcript shrink". - if (allTurns.length < state.capturedTurnCount) { - log("transcript_shrink_detected", { - cached: state.capturedTurnCount, - observed: allTurns.length, - }); - state.capturedTurnCount = 0; - } + // Post-compact transcript-shrink defense: codex's /compact may rewrite or + // truncate transcript_path. If allTurns has fewer entries than we cached, + // our slice math would underflow and silently drop turns. Reset the + // counter so the next slice captures everything in the new transcript. + // See DESIGN.md "Post-compact transcript shrink". + if (allTurns.length < state.capturedTurnCount) { + log("transcript_shrink_detected", { + cached: state.capturedTurnCount, + observed: allTurns.length, + }); + state.capturedTurnCount = 0; + } - const newTurns = allTurns.slice(state.capturedTurnCount); + const newTurns = allTurns.slice(state.capturedTurnCount); - log("transcript_parse", { - totalTurns: allTurns.length, - previouslyCaptured: state.capturedTurnCount, - newTurns: newTurns.length, - }); + log("transcript_parse", { + totalTurns: allTurns.length, + previouslyCaptured: state.capturedTurnCount, + newTurns: newTurns.length, + }); - if (cfg.captureMode === "keyword" && newTurns.length > 0 && !hasCaptureKeyword(newTurns)) { - log("skip", { stage: "capture_mode", reason: "keyword mode without capture trigger" }); - await saveState(state); - noop(); - return; - } + if (cfg.captureMode === "keyword" && newTurns.length > 0 && !hasCaptureKeyword(newTurns)) { + log("skip", { stage: "capture_mode", reason: "keyword mode without capture trigger" }); + await save(state); + noop(); + return; + } - let added = 0; - let ovSessionId = ""; - let commitInfo = { committed: false, pendingTokens: 0, commitCount: 0, totalMessageCount: 0 }; - if (newTurns.length > 0) { - ovSessionId = resolveOvSessionId(state); - if (!ovSessionId) { - logError("resolve_ov_session", "failed to derive OV session id"); - } else { - added = await appendTurns(ovSessionId, newTurns, state); - log("appended", { ovSessionId, added }); - commitInfo = await maybeCommitByThreshold(ovSessionId, added); + let added = 0; + let ovSessionId = ""; + let commitInfo = { committed: false, pendingTokens: 0, commitCount: 0, totalMessageCount: 0 }; + if (newTurns.length > 0) { + ovSessionId = resolveOvSessionId(state); + if (!ovSessionId) { + logError("resolve_ov_session", "failed to derive OV session id"); + } else { + added = await appendTurns(ovSessionId, newTurns, state, save); + log("appended", { ovSessionId, added }); + commitInfo = await maybeCommitByThreshold(ovSessionId, added); + } } - } - await saveState(state); + await save(state); - // could also sweep here, deliberately not — see header comment + DESIGN.md §5. + // could also sweep here, deliberately not — see header comment + DESIGN.md §5. - if (added > 0) { - noop( - `appended ${added} turn(s) to OpenViking session ${state.ovSessionId}` + - (commitInfo.committed ? " (committed)" : ""), - ); - } else { - noop(); - } + if (added > 0) { + noop( + `appended ${added} turn(s) to OpenViking session ${state.ovSessionId}` + + (commitInfo.committed ? " (committed)" : ""), + ); + } else { + noop(); + } + }); } function hasCaptureKeyword(turns) { diff --git a/examples/codex-memory-plugin/scripts/auto-recall.mjs b/examples/codex-memory-plugin/scripts/auto-recall.mjs index 8489e61247..1a1175c925 100644 --- a/examples/codex-memory-plugin/scripts/auto-recall.mjs +++ b/examples/codex-memory-plugin/scripts/auto-recall.mjs @@ -20,10 +20,13 @@ import { loadConfig } from "./config.mjs"; import { trySpawnCodex } from "./codex-launch.mjs"; import { createLogger } from "./debug-log.mjs"; import { + buildRecallCompressorCandidates, buildCodexExecArgs, + cacheRecallCompressorProfile, fallbackRecallCompressorProfile, loadCachedRecallCompressorProfile, markRecallCompressorRuntimeFailed, + recallCompressionExplicitlyOff, } from "./recall-compressor-profile.mjs"; import { deriveOvSessionId } from "./session-state.mjs"; import { @@ -42,6 +45,8 @@ let emitted = false; let activeCompressor = null; let recallDeadline = null; const DEFAULT_FINAL_RECALL_CHARS = 6500; +const EXCLUDED_EXPERIENCE_STATUSES = new Set(["deprecated", "archived"]); +const EXPERIENCE_SIDECAR_FILENAMES = new Set([".abstract.md", ".overview.md", ".relations.json"]); function output(obj, exitAfter = false) { if (emitted) return; @@ -307,12 +312,146 @@ async function readMemoryContent(uri) { return null; } -function assembledToRecallResult(rendered, entries) { - const items = entries +function isRecord(value) { + return Boolean(value) && typeof value === "object" && !Array.isArray(value); +} + +function experienceUriInfo(uri) { + const value = String(uri || "").trim(); + const looksLikeExperience = /^viking:\/\//i.test(value) + && /\/memories\/experiences(?:\/|$)/i.test(value); + let parsed; + try { + parsed = new URL(value); + } catch { + return { isExperience: looksLikeExperience, canonical: false, value }; + } + if (parsed.protocol !== "viking:") { + return { isExperience: looksLikeExperience, canonical: false, value }; + } + const parts = [parsed.hostname, ...parsed.pathname.split("/").filter(Boolean)]; + let memoryRoot = -1; + if (parts[0] === "user") { + if (parts.length > 5 && parts[2] === "peers" && parts[4] === "memories") memoryRoot = 4; + else if (parts.length > 3 && parts[2] === "memories") memoryRoot = 2; + else if (parts.length > 4 && parts[1] === "peers" && parts[3] === "memories") memoryRoot = 3; + else if (parts.length > 2 && parts[1] === "memories") memoryRoot = 1; + } else if (parts.length > 3 && parts[0] === "agent" && parts[2] === "memories") { + memoryRoot = 2; + } + if (memoryRoot < 0 || parts[memoryRoot + 1] !== "experiences") { + // Fail closed for malformed/unknown Viking namespaces that still visibly + // target the Experience directory; treating them as ordinary memories + // would bypass authoritative lifecycle hydration. + return { isExperience: looksLikeExperience, canonical: false, value }; + } + const relative = parts.slice(memoryRoot + 2); + const basename = relative.at(-1) || ""; + const canonical = Boolean( + !parsed.search + && !parsed.hash + && relative.length > 0 + && relative.every((segment) => segment && segment !== "." && segment !== "..") + && !EXPERIENCE_SIDECAR_FILENAMES.has(basename), + ); + return { isExperience: true, canonical, value }; +} + +function normalizedStatus(...sources) { + let status = ""; + for (const source of sources) { + if (isRecord(source) && typeof source.status === "string" && source.status.trim()) { + status = source.status.trim().toLowerCase(); + } + } + return status; +} + +function parseAuthoritativeExperienceDocument(value) { + const objectValue = isRecord(value) ? value : null; + const raw = objectValue + ? objectValue.raw_content ?? objectValue.raw ?? objectValue.content + : value; + if (typeof raw !== "string" || !raw.trim()) return null; + + const hasMemoryFieldsMarker = /\s*$/i.exec(raw); + let fields = null; + if (hasMemoryFieldsMarker) { + if (!trailer) return null; + try { + fields = JSON.parse(trailer[1].trim()); + } catch { + return null; + } + if (!isRecord(fields)) return null; + } else if (objectValue) { + // Some server versions return raw content and authoritative metadata as + // separate JSON members. Legacy documents can also have no lifecycle + // metadata at all; that is an eligible unknown status, not a parse error. + if (isRecord(objectValue.attrs)) fields = objectValue.attrs; + if (isRecord(objectValue.metadata)) fields = { ...(fields || {}), ...objectValue.metadata }; + if (Object.hasOwn(objectValue, "status")) fields = { ...(fields || {}), status: objectValue.status }; + } else { + // A non-empty legacy raw string without MEMORY_FIELDS has status="". + fields = {}; + } + + const content = (trailer ? raw.slice(0, trailer.index) : raw).trim(); + if (!content) return null; + return { + content, + status: normalizedStatus(objectValue?.attrs, objectValue?.metadata, objectValue, fields), + }; +} + +async function readAuthoritativeExperience(uri) { + const info = experienceUriInfo(uri); + if (!info.isExperience) return { isExperience: false, document: null }; + if (!info.canonical) { + log("experience_drop", { uri: info.value, reason: "noncanonical_uri" }); + return { isExperience: true, document: null }; + } + const result = await fetchJSON( + `/api/v1/content/read?uri=${encodeURIComponent(info.value)}&raw=true`, + ); + if (!result.ok) { + log("experience_drop", { uri: info.value, reason: "raw_read_failed", status: result.status || 0 }); + return { isExperience: true, document: null }; + } + const document = parseAuthoritativeExperienceDocument(result.result); + if (!document) { + log("experience_drop", { uri: info.value, reason: "invalid_or_empty_raw_metadata" }); + return { isExperience: true, document: null }; + } + if (EXCLUDED_EXPERIENCE_STATUSES.has(document.status)) { + log("experience_drop", { uri: info.value, reason: "lifecycle_status", status: document.status }); + return { isExperience: true, document: null }; + } + return { isExperience: true, document }; +} + +async function enforceExperienceLifecycle(items) { + const checked = await Promise.all(items.map(async (item) => { + const result = await readAuthoritativeExperience(item?.uri); + if (!result.isExperience) return { item, experienceContent: null }; + if (!result.document) return null; + return { item, experienceContent: result.document.content }; + })); + const kept = checked.filter(Boolean); + return { kept, filtered: kept.length !== items.length }; +} + +async function assembledToRecallResult(rendered, entries) { + const normalizedItems = entries .map(normalizeContextEntry) .map((entry) => ({ ...entry, score: clampScore(entry.score) })) .filter((entry) => entry.uri && entry.text); - const context = rendered + const lifecycle = await enforceExperienceLifecycle(normalizedItems); + const items = lifecycle.kept.map(({ item, experienceContent }) => ( + experienceContent === null ? item : { ...item, text: experienceContent } + )); + const renderedContext = rendered ? [ "OpenViking memory digest:", rendered, @@ -320,6 +459,11 @@ function assembledToRecallResult(rendered, entries) { "More detail: use the OpenViking MCP recall/read/search tools with cited viking:// URIs if needed.", ].join("\n") : ""; + // Once any entry is removed, the server-rendered block is no longer safe: it + // still contains the excluded entry's body. Rebuild only from retained items. + const context = lifecycle.filtered || normalizedItems.length !== entries.length + ? fallbackDigest(items) + : renderedContext; return { context, items }; } @@ -341,7 +485,7 @@ async function recallViaServerAssembly(query, ovSessionId = "") { log, }); if (assembled) { - return assembledToRecallResult(assembled.rendered, assembled.entries); + return await assembledToRecallResult(assembled.rendered, assembled.entries); } const body = buildRecallEndpointBody(cfg); @@ -352,7 +496,7 @@ async function recallViaServerAssembly(query, ovSessionId = "") { log("recall_endpoint_fallback", { status: result.status || 0 }); return null; } - return assembledToRecallResult( + return await assembledToRecallResult( String(result.result?.rendered || "").trim(), Array.isArray(result.result?.entries) ? result.result.entries : [], ); @@ -410,15 +554,34 @@ function normalizeCompressedContext(text) { return truncateText(appendMcpRetrievalHint(value), 4000); } -async function getRecallCompressorProfile() { +async function getRecallCompressorProfiles() { + if (recallCompressionExplicitlyOff(cfg)) return []; const cached = await loadCachedRecallCompressorProfile(cfg); - if (cached) return cached; - const fallback = fallbackRecallCompressorProfile(cfg); - log("compress_profile_cache_miss", fallback); - return fallback; + const failedModels = new Set( + cached?.source === "runtime_failed" + ? [...(cached.failedModels || []), cached.failedModel || ""].filter(Boolean) + : [], + ); + const candidates = []; + if (cached?.enabled) candidates.push(cached); + if (!cached) { + const fallback = fallbackRecallCompressorProfile(cfg); + log("compress_profile_cache_miss", fallback); + if (fallback.enabled) candidates.push(fallback); + } + candidates.push(...buildRecallCompressorCandidates(cfg)); + + const seenModels = new Set(); + return candidates.filter((profile) => { + if (!profile?.enabled && profile?.enabled !== undefined) return false; + if (!profile?.model || failedModels.has(profile.model)) return false; + if (seenModels.has(profile.model)) return false; + seenModels.add(profile.model); + return true; + }).slice(0, 2); } -async function runCodexCompressor(prompt, profile) { +async function runCodexCompressor(prompt, profile, timeoutMs) { const tmp = await mkdtemp(join(tmpdir(), "ov-recall-compress-")); const outputPath = join(tmp, "last-message.txt"); const args = buildCodexExecArgs(profile, outputPath); @@ -436,39 +599,28 @@ async function runCodexCompressor(prompt, profile) { let done = false; let timedOut = false; let stderr = ""; - const finish = (value, { runtimeFailed = false } = {}) => { + const finish = (value) => { if (done) return; done = true; if (activeCompressor === child) activeCompressor = null; clearTimeout(timer); - if (runtimeFailed) { - // Mark the profile as runtime_failed so subsequent UPS calls in - // this same codex session skip compress (avoids burning - // ~recallCompressTimeoutMs per turn on a guaranteed-to-fail - // spawn). Next SessionStart's cache-first detect treats this - // marker as a cache miss and re-resolves against the current - // catalogue, so a transient failure self-recovers across codex - // restarts. Best-effort write; failure is non-fatal. - markRecallCompressorRuntimeFailed(cfg, { failedModel: profile.model || "" }) - .catch(() => {}); - } resolve(value); }; const launch = trySpawnCodex(args, { env, stdio: ["pipe", "ignore", "pipe"] }); if (launch.error) { logError("compress_spawn", launch.error); - finish(null, { runtimeFailed: true }); + finish(null); return; } child = launch.child; activeCompressor = child; timer = setTimeout(() => { timedOut = true; - logError("compress_timeout", `timed out after ${cfg.recallCompressTimeoutMs}ms`); + logError("compress_timeout", `timed out after ${timeoutMs}ms`); try { child.kill("SIGKILL"); } catch { /* best effort */ } - }, cfg.recallCompressTimeoutMs); + }, timeoutMs); child.stderr.on("data", (chunk) => { stderr += chunk.toString(); @@ -476,11 +628,11 @@ async function runCodexCompressor(prompt, profile) { }); child.on("error", (err) => { logError("compress_spawn", err); - finish(null, { runtimeFailed: true }); + finish(null); }); child.on("close", async (code) => { if (timedOut) { - finish(null, { runtimeFailed: true }); + finish(null); return; } if (code !== 0) { @@ -488,14 +640,14 @@ async function runCodexCompressor(prompt, profile) { profile, error: stderr.trim().slice(-1000) || `codex exited ${code}`, }); - finish(null, { runtimeFailed: true }); + finish(null); return; } try { finish(await readFile(outputPath, "utf-8")); } catch (err) { logError("compress_read", err); - finish(null, { runtimeFailed: true }); + finish(null); } }); child.stdin.end(prompt); @@ -506,11 +658,14 @@ async function runCodexCompressor(prompt, profile) { } async function compressMemoryContext(userPrompt, items) { - if (!cfg.recallCompress) return null; - const profile = await getRecallCompressorProfile(); - if (!profile.enabled) { - log("compress_skip", { reason: "profile disabled", profile }); - return null; + if (recallCompressionExplicitlyOff(cfg)) { + log("compress_skip", { reason: "explicitly disabled" }); + return { status: "disabled", context: "" }; + } + const profiles = await getRecallCompressorProfiles(); + if (profiles.length === 0) { + log("compress_skip", { reason: "no usable profiles" }); + return { status: "failed", context: "" }; } const perItemChars = Math.max(500, Math.floor(cfg.recallCompressMaxInputChars / Math.max(1, items.length))); const payload = { @@ -540,11 +695,44 @@ Task: Input JSON: ${JSON.stringify(payload, null, 2)} `; - const raw = await runCodexCompressor(prompt, profile); - if (raw === null) return null; - const compressed = normalizeCompressedContext(raw); - log("compressed", { inputCount: items.length, chars: compressed.length, profile }); - return compressed; + const failedModels = []; + // `recallCompressTimeoutMs` is one total budget, not a per-model budget. + // Divide the remaining time across the attempts still available so a hung + // primary cannot consume the fallback's entire window or overrun the hook. + const compressorDeadline = Date.now() + cfg.recallCompressTimeoutMs; + for (const [attempt, profile] of profiles.entries()) { + const remainingMs = compressorDeadline - Date.now(); + if (remainingMs <= 0) break; + const attemptsLeft = profiles.length - attempt; + const attemptTimeoutMs = Math.max(1, Math.floor(remainingMs / attemptsLeft)); + log("compress_attempt", { attempt: attempt + 1, attemptTimeoutMs, profile }); + const raw = await runCodexCompressor(prompt, profile, attemptTimeoutMs); + if (raw === null) { + failedModels.push(profile.model || ""); + continue; + } + // A runtime_failed cache can exclude the primary before this loop, making a + // working fallback attempt 0 rather than attempt 1. Promote every successful + // profile when the cache does not already describe that exact model/profile; + // attempt position is not a reliable signal of whether promotion is needed. + const cached = await loadCachedRecallCompressorProfile(cfg); + const cachedThinking = String(cached?.thinking || "default").toLowerCase(); + const profileThinking = String(profile?.thinking || "default").toLowerCase(); + if ( + !cached?.enabled + || cached.model !== profile.model + || cachedThinking !== profileThinking + ) { + await cacheRecallCompressorProfile(cfg, profile); + } + const compressed = normalizeCompressedContext(raw); + log("compressed", { inputCount: items.length, chars: compressed.length, profile }); + return { status: "ok", context: compressed }; + } + + await markRecallCompressorRuntimeFailed(cfg, { failedModels }); + log("compress_fail_closed", { failedModels }); + return { status: "failed", context: "" }; } async function main() { @@ -601,15 +789,14 @@ async function main() { emit(); return; } - const compressedContext = endpointRecall.items.length > 0 + const compression = endpointRecall.items.length > 0 ? await compressMemoryContext(userPrompt, endpointRecall.items) - : null; - const endpointFallback = cfg.recallCompress && endpointRecall.items.length > 0 - ? fallbackDigest(endpointRecall.items) - : endpointRecall.context; - const memoryContext = compressedContext === null - ? endpointFallback - : compressedContext; + : { status: "disabled", context: "" }; + const memoryContext = endpointRecall.items.length === 0 + ? endpointRecall.context + : compression.status === "disabled" + ? (cfg.recallCompress ? fallbackDigest(endpointRecall.items) : endpointRecall.context) + : compression.context; if (!memoryContext) { log("skip", { stage: "recall_endpoint", reason: "compressor found no relevant memory" }); emit(); @@ -617,7 +804,7 @@ async function main() { } log("recall_endpoint", { chars: memoryContext.length, - compressed: compressedContext !== null, + compressed: compression.status === "ok", entryCount: endpointRecall.items.length, }); emit(memoryContext); @@ -635,8 +822,23 @@ async function main() { const processed = postProcess(allMemories, candidateLimit, cfg.scoreThreshold); log("post_process", { beforeCount: allMemories.length, afterCount: processed.length }); + // Validate the full ranked candidate pool before selecting the final width. + // Otherwise an archived top hit would consume a slot, be removed after pick, + // and prevent a lower-ranked eligible memory from backfilling it. + const lifecycle = await enforceExperienceLifecycle(processed); + const eligibleProcessed = lifecycle.kept.map(({ item }) => item); + const experienceContentByUri = new Map( + lifecycle.kept + .filter(({ experienceContent }) => experienceContent !== null) + .map(({ item, experienceContent }) => [item.uri, experienceContent]), + ); + log("experience_lifecycle", { + beforeCount: processed.length, + afterCount: eligibleProcessed.length, + }); + const profile = buildQueryProfile(userPrompt); - const ranked = [...processed] + const ranked = [...eligibleProcessed] .map((item) => ({ item, breakdown: getRankingBreakdown(item, profile) })) .sort((a, b) => b.breakdown.finalScore - a.breakdown.finalScore); @@ -646,24 +848,29 @@ async function main() { } } else { log("ranking_summary", { - candidateCount: processed.length, + candidateCount: eligibleProcessed.length, topCandidates: ranked.slice(0, 5).map((entry) => ({ uri: entry.item.uri, finalScore: entry.breakdown.finalScore })), }); } - const memories = pickMemories(processed, cfg.recallLimit, userPrompt); + const memories = pickMemories(eligibleProcessed, cfg.recallLimit, userPrompt); if (memories.length === 0) { log("skip", { stage: "pick", reason: "no memories survived ranking" }); emit(); return; } - log("picked", { pickedCount: memories.length, uris: memories.map((m) => m.uri) }); + log("picked", { + pickedCount: memories.length, + uris: memories.map((item) => item.uri), + }); const memoryItems = await Promise.all( memories.map(async (item) => { let text = (item.abstract || item.overview || item.uri).trim(); - if (item.level === 2) { + if (experienceContentByUri.has(item.uri)) { + text = experienceContentByUri.get(item.uri); + } else if (item.level === 2) { const content = await readMemoryContent(item.uri); if (content) text = content; } @@ -676,8 +883,10 @@ async function main() { }), ); - const compressedContext = await compressMemoryContext(userPrompt, memoryItems); - const memoryContext = compressedContext === null ? fallbackDigest(memoryItems) : compressedContext; + const compression = await compressMemoryContext(userPrompt, memoryItems); + const memoryContext = compression.status === "disabled" + ? fallbackDigest(memoryItems) + : compression.context; emit(memoryContext); } diff --git a/examples/codex-memory-plugin/scripts/auto-recall.test.mjs b/examples/codex-memory-plugin/scripts/auto-recall.test.mjs index 97a1f52afb..e5353552ef 100644 --- a/examples/codex-memory-plugin/scripts/auto-recall.test.mjs +++ b/examples/codex-memory-plugin/scripts/auto-recall.test.mjs @@ -7,6 +7,7 @@ import { dirname, join } from "node:path"; import test from "node:test"; import { fileURLToPath } from "node:url"; import { resolveCodexLaunch, trySpawnCodex } from "./codex-launch.mjs"; +import { markRecallCompressorRuntimeFailed } from "./recall-compressor-profile.mjs"; const SCRIPT_DIR = dirname(fileURLToPath(import.meta.url)); @@ -111,7 +112,12 @@ async function withFakeCodex(output, fn, { exitCode = 0 } = {}) { const callLog = join(binDir, "calls.log"); await writeFile(executable, `#!/bin/sh output_path="" +model="" while [ "$#" -gt 0 ]; do + if [ "$1" = "-m" ]; then + shift + model="$1" + fi if [ "$1" = "--output-last-message" ]; then shift output_path="$1" @@ -119,7 +125,13 @@ while [ "$#" -gt 0 ]; do shift done cat >/dev/null -printf 'called\\n' >> "$FAKE_CODEX_CALL_LOG" +printf '%s\\n' "$model" >> "$FAKE_CODEX_CALL_LOG" +if [ "$FAKE_CODEX_HANG" = "1" ]; then + while :; do :; done +fi +if [ -n "$FAKE_CODEX_FAIL_MODEL" ] && [ "$model" = "$FAKE_CODEX_FAIL_MODEL" ]; then + exit 1 +fi if [ "$FAKE_CODEX_EXIT_CODE" -ne 0 ]; then exit "$FAKE_CODEX_EXIT_CODE" fi @@ -148,11 +160,29 @@ async function runEndpointCompressionCase({ compressorOutput, exitCode = 0, extraEnv = {}, + seedFailedModels = [], }) { const stateDir = await mkdtemp(join(tmpdir(), "ov-auto-recall-endpoint-compress-")); let requestBody = null; try { + if (seedFailedModels.length > 0) { + const previousStateDir = process.env.OPENVIKING_CODEX_STATE_DIR; + process.env.OPENVIKING_CODEX_STATE_DIR = stateDir; + try { + await markRecallCompressorRuntimeFailed({ + recallCompress: true, + recallCompressModel: "", + recallCompressThinking: "", + recallCompressConfigured: false, + recallCompressDetectTtlMs: 604_800_000, + }, { failedModels: seedFailedModels }); + } finally { + if (previousStateDir === undefined) delete process.env.OPENVIKING_CODEX_STATE_DIR; + else process.env.OPENVIKING_CODEX_STATE_DIR = previousStateDir; + } + } return await withFakeCodex(compressorOutput, async ({ callLog, env }) => { + const startedAt = Date.now(); const result = await withMockOpenViking(async (req, res) => { const url = new URL(req.url, "http://127.0.0.1"); if (req.method === "GET" && url.pathname === "/health") { @@ -189,9 +219,17 @@ async function runEndpointCompressionCase({ }, )); const compressorCallLog = await readFile(callLog, "utf-8").catch(() => ""); + const compressorModels = compressorCallLog.trim().split("\n").filter(Boolean); + const cachedProfile = JSON.parse( + await readFile(join(stateDir, "recall-compressor-profile.json"), "utf-8") + .catch(() => "null"), + )?.profile || null; return { output: JSON.parse(result.stdout.trim()), - compressorCalls: compressorCallLog.trim().split("\n").filter(Boolean).length, + compressorCalls: compressorModels.length, + compressorModels, + cachedProfile, + elapsedMs: Date.now() - startedAt, requestBody, }; }, { exitCode }); @@ -268,10 +306,12 @@ test("auto-recall asks the context face with the derived OpenViking session id", assert.equal(requests[0].body.limit, undefined); assert.equal( Object.values(requests[0].body.quotas).reduce((sum, quota) => sum + quota, 0), - 6, + 1, ); assert.equal(requests[0].body.max_tokens, 800); assert.equal(requests[0].body.dedup_turns, 5); + assert.equal(requests[0].body.quotas.resources, 1); + assert.equal(requests[0].body.quotas.experiences, 0); assert.equal(requests[0].body.target_uri, undefined); } finally { await rm(stateDir, { recursive: true, force: true }); @@ -348,13 +388,303 @@ test("auto-recall prefers the server recall endpoint when available", async () = "/api/v1/search/search", "/api/v1/search/recall", ]); - assert.equal(Object.values(requests[1].body.quotas).reduce((sum, quota) => sum + quota, 0), 3); + assert.equal(Object.values(requests[1].body.quotas).reduce((sum, quota) => sum + quota, 0), 2); + assert.equal(requests[1].body.quotas.experiences, 0); assert.equal(requests[1].body.max_chars, 6500); } finally { await rm(stateDir, { recursive: true, force: true }); } }); +test("auto-recall authoritatively filters deprecated Experience entries and rebuilds rendered context", async () => { + const stateDir = await mkdtemp(join(tmpdir(), "ov-auto-recall-experience-context-")); + const activeUri = "viking://user/zeus/memories/experiences/active-case.md"; + const deprecatedUri = "viking://user/zeus/memories/experiences/old-case.md"; + const eventUri = "viking://user/zeus/memories/events/safe-event.md"; + const rawReads = []; + + try { + await withMockOpenViking(async (req, res) => { + const url = new URL(req.url, "http://127.0.0.1"); + if (req.method === "GET" && url.pathname === "/health") { + writeJson(res, { status: "ok", result: { ok: true } }); + return; + } + if (req.method === "POST" && url.pathname === "/api/v1/search/search") { + writeJson(res, { + status: "ok", + result: { + entries: [ + { uri: activeUri, category: "experiences", score: 0.9, text: "ACTIVE EXPERIENCE SUMMARY" }, + { uri: deprecatedUri, category: "experiences", score: 0.8, text: "DEPRECATED SECRET BODY" }, + { uri: eventUri, category: "events", score: 0.7, text: "SAFE EVENT SUMMARY" }, + ], + rendered: [ + `${activeUri}ACTIVE EXPERIENCE SUMMARY`, + `${deprecatedUri}DEPRECATED SECRET BODY`, + `${eventUri}SAFE EVENT SUMMARY`, + ].join("\n"), + stats: { returned: 3 }, + }, + }); + return; + } + if (req.method === "GET" && url.pathname === "/api/v1/content/read") { + const uri = url.searchParams.get("uri"); + rawReads.push({ uri, raw: url.searchParams.get("raw") }); + if (uri === activeUri) { + // Legacy Experience: non-empty authoritative raw with no lifecycle + // metadata remains eligible with status="". + writeJson(res, { status: "ok", result: "authoritative legacy active body" }); + } else { + writeJson(res, { + status: "ok", + result: "authoritative deprecated body\n\n", + }); + } + return; + } + res.writeHead(404, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "error", error: "not found" })); + }, async (baseUrl) => { + const result = await runAutoRecall( + { prompt: "reuse a relevant past case", session_id: "codex:experience-context" }, + { + OPENVIKING_AUTO_RECALL: "1", + OPENVIKING_CODEX_STATE_DIR: stateDir, + OPENVIKING_STATE_DIR: stateDir, + OPENVIKING_CONFIG_FILE: join(stateDir, "missing-ov.conf"), + OPENVIKING_CLI_CONFIG_FILE: join(stateDir, "missing-ovcli.conf"), + OPENVIKING_CREDENTIAL_SOURCE: "env", + OPENVIKING_RECALL_COMPRESS: "0", + OPENVIKING_RECALL_LIMIT: "3", + OPENVIKING_RECALL_TIMEOUT_MS: "10000", + OPENVIKING_MIN_QUERY_LENGTH: "1", + OPENVIKING_SCORE_THRESHOLD: "0", + OPENVIKING_TIMEOUT_MS: "5000", + OPENVIKING_URL: baseUrl, + }, + ); + + const output = JSON.parse(result.stdout.trim()); + const context = output.hookSpecificOutput.additionalContext; + assert.match(context, /authoritative legacy active body/); + assert.match(context, /SAFE EVENT SUMMARY/); + assert.doesNotMatch(context, /ACTIVE EXPERIENCE SUMMARY/); + assert.doesNotMatch(context, /DEPRECATED SECRET BODY/); + assert.doesNotMatch(context, //); + }); + + assert.deepEqual(rawReads, [ + { uri: activeUri, raw: "true" }, + { uri: deprecatedUri, raw: "true" }, + ]); + } finally { + await rm(stateDir, { recursive: true, force: true }); + } +}); + +test("legacy search fails closed for Experience raw read, empty raw, and metadata parse failures", async () => { + const stateDir = await mkdtemp(join(tmpdir(), "ov-auto-recall-experience-legacy-")); + const uris = { + failed: "viking://user/zeus/memories/experiences/read-failed.md", + empty: "viking://user/zeus/memories/experiences/empty.md", + malformed: "viking://user/zeus/memories/experiences/malformed.md", + }; + const rawReads = []; + + try { + await withMockOpenViking(async (req, res) => { + const url = new URL(req.url, "http://127.0.0.1"); + if (req.method === "GET" && url.pathname === "/health") { + writeJson(res, { status: "ok", result: { ok: true } }); + return; + } + if (req.method === "POST" && url.pathname === "/api/v1/search/recall") { + writeStatusJson(res, 404, { status: "error", error: "not found" }); + return; + } + if (req.method === "POST" && url.pathname === "/api/v1/search/search") { + const body = await readRequestBody(req); + if (body.mode === "context") { + writeStatusJson(res, 400, { + status: "error", + error: "Extra inputs are not permitted: mode", + }); + return; + } + if (body.target_uri === "viking://user/zeus/memories") { + writeJson(res, { + status: "ok", + result: { + memories: Object.values(uris).map((uri, index) => ({ + uri, + level: 2, + score: 0.9 - index / 10, + category: "experiences", + abstract: `candidate ${index}`, + })), + skills: [], + }, + }); + return; + } + writeJson(res, { status: "ok", result: { memories: [], skills: [] } }); + return; + } + if (req.method === "GET" && url.pathname === "/api/v1/content/read") { + const uri = url.searchParams.get("uri"); + rawReads.push({ uri, raw: url.searchParams.get("raw") }); + if (uri === uris.failed) { + writeStatusJson(res, 503, { status: "error", error: "unavailable" }); + } else if (uri === uris.empty) { + writeJson(res, { status: "ok", result: "" }); + } else { + writeJson(res, { + status: "ok", + result: "body\n\n", + }); + } + return; + } + res.writeHead(404, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "error", error: "not found" })); + }, async (baseUrl) => { + const result = await runAutoRecall( + { prompt: "find prior cases", session_id: "codex:experience-legacy" }, + { + OPENVIKING_AUTO_RECALL: "1", + OPENVIKING_CODEX_STATE_DIR: stateDir, + OPENVIKING_STATE_DIR: stateDir, + OPENVIKING_CONFIG_FILE: join(stateDir, "missing-ov.conf"), + OPENVIKING_CLI_CONFIG_FILE: join(stateDir, "missing-ovcli.conf"), + OPENVIKING_CREDENTIAL_SOURCE: "env", + OPENVIKING_USER: "zeus", + OPENVIKING_RECALL_COMPRESS: "0", + OPENVIKING_RECALL_LIMIT: "3", + OPENVIKING_RECALL_TIMEOUT_MS: "10000", + OPENVIKING_MIN_QUERY_LENGTH: "1", + OPENVIKING_SCORE_THRESHOLD: "0", + OPENVIKING_TIMEOUT_MS: "5000", + OPENVIKING_URL: baseUrl, + }, + ); + assert.deepEqual(JSON.parse(result.stdout.trim()), {}); + }); + + assert.deepEqual( + rawReads.map(({ uri }) => uri).sort(), + Object.values(uris).sort(), + ); + assert.ok(rawReads.every(({ raw }) => raw === "true")); + } finally { + await rm(stateDir, { recursive: true, force: true }); + } +}); + +test("legacy search backfills a lower eligible memory after an archived top Experience", async () => { + const stateDir = await mkdtemp(join(tmpdir(), "ov-auto-recall-experience-backfill-")); + const archivedUri = "viking://user/zeus/memories/experiences/archived-top.md"; + const eventUri = "viking://user/zeus/memories/events/lower-eligible.md"; + const contentReads = []; + + try { + await withMockOpenViking(async (req, res) => { + const url = new URL(req.url, "http://127.0.0.1"); + if (req.method === "GET" && url.pathname === "/health") { + writeJson(res, { status: "ok", result: { ok: true } }); + return; + } + if (req.method === "POST" && url.pathname === "/api/v1/search/recall") { + writeStatusJson(res, 404, { status: "error", error: "not found" }); + return; + } + if (req.method === "POST" && url.pathname === "/api/v1/search/search") { + const body = await readRequestBody(req); + if (body.mode === "context") { + writeStatusJson(res, 400, { + status: "error", + error: "Extra inputs are not permitted: mode", + }); + return; + } + if (body.target_uri === "viking://user/zeus/memories") { + writeJson(res, { + status: "ok", + result: { + memories: [ + { + uri: archivedUri, + level: 2, + score: 0.99, + category: "experiences", + abstract: "archived top candidate", + }, + { + uri: eventUri, + level: 2, + score: 0.5, + category: "events", + abstract: "lower eligible candidate", + }, + ], + skills: [], + }, + }); + return; + } + writeJson(res, { status: "ok", result: { memories: [], skills: [] } }); + return; + } + if (req.method === "GET" && url.pathname === "/api/v1/content/read") { + const uri = url.searchParams.get("uri"); + contentReads.push({ uri, raw: url.searchParams.get("raw") }); + if (uri === archivedUri) { + writeJson(res, { + status: "ok", + result: "old guidance\n\n", + }); + } else { + writeJson(res, { status: "ok", result: "lower eligible recalled detail" }); + } + return; + } + res.writeHead(404, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "error", error: "not found" })); + }, async (baseUrl) => { + const result = await runAutoRecall( + { prompt: "find the eligible memory", session_id: "codex:experience-backfill" }, + { + OPENVIKING_AUTO_RECALL: "1", + OPENVIKING_CODEX_STATE_DIR: stateDir, + OPENVIKING_STATE_DIR: stateDir, + OPENVIKING_CONFIG_FILE: join(stateDir, "missing-ov.conf"), + OPENVIKING_CLI_CONFIG_FILE: join(stateDir, "missing-ovcli.conf"), + OPENVIKING_CREDENTIAL_SOURCE: "env", + OPENVIKING_USER: "zeus", + OPENVIKING_RECALL_COMPRESS: "0", + OPENVIKING_RECALL_LIMIT: "1", + OPENVIKING_RECALL_TIMEOUT_MS: "10000", + OPENVIKING_MIN_QUERY_LENGTH: "1", + OPENVIKING_SCORE_THRESHOLD: "0", + OPENVIKING_TIMEOUT_MS: "5000", + OPENVIKING_URL: baseUrl, + }, + ); + const context = JSON.parse(result.stdout.trim()).hookSpecificOutput.additionalContext; + assert.match(context, /lower eligible recalled detail/); + assert.doesNotMatch(context, /old guidance|archived top candidate/); + }); + + assert.deepEqual(contentReads, [ + { uri: archivedUri, raw: "true" }, + { uri: eventUri, raw: null }, + ]); + } finally { + await rm(stateDir, { recursive: true, force: true }); + } +}); + test("auto-recall applies the relevance compressor to server recall entries", async () => { const result = await runEndpointCompressionCase({ prompt: "Explain HTTP 429", @@ -374,7 +704,57 @@ test("auto-recall applies the relevance compressor to server recall entries", as assert.equal(result.requestBody.max_chars, 18000); }); -test("auto-recall falls back to a bounded deterministic digest when endpoint compression fails", async () => { +test("auto-recall tries the next compressor model after a runtime failure", async () => { + const result = await runEndpointCompressionCase({ + prompt: "Which editor do I prefer?", + entry: { + uri: "viking://user/zeus/memories/preferences/editor.md", + score: 0.91, + type: "preferences", + mode: "summary", + summary: "Use Vim", + }, + rendered: "Use Vim", + compressorOutput: [ + "OpenViking memory digest:", + "- Use Vim (viking://user/zeus/memories/preferences/editor.md)", + ].join("\n"), + extraEnv: { FAKE_CODEX_FAIL_MODEL: "gpt-5.3-codex-spark" }, + }); + + assert.match(result.output.hookSpecificOutput.additionalContext, /Use Vim/); + assert.equal(result.compressorCalls, 2); + assert.deepEqual(result.compressorModels, ["gpt-5.3-codex-spark", "gpt-5.6-luna"]); + assert.equal(result.cachedProfile.model, "gpt-5.6-luna"); + assert.equal(result.cachedProfile.enabled, true); + assert.ok(result.elapsedMs < 2000, `immediate fallback took ${result.elapsedMs}ms`); +}); + +test("auto-recall promotes a successful attempt-zero profile after prior failures exclude primary", async () => { + const result = await runEndpointCompressionCase({ + prompt: "Which editor do I prefer?", + entry: { + uri: "viking://user/zeus/memories/preferences/editor.md", + score: 0.91, + type: "preferences", + mode: "summary", + summary: "Use Vim", + }, + rendered: "Use Vim", + compressorOutput: [ + "OpenViking memory digest:", + "- Use Vim (viking://user/zeus/memories/preferences/editor.md)", + ].join("\n"), + seedFailedModels: ["gpt-5.3-codex-spark"], + }); + + assert.deepEqual(result.compressorModels, ["gpt-5.6-luna"]); + assert.equal(result.cachedProfile.enabled, true); + assert.equal(result.cachedProfile.model, "gpt-5.6-luna"); + assert.equal(result.cachedProfile.thinking, "low"); +}); + +test("auto-recall fails closed when every compressor model fails", async () => { const result = await runEndpointCompressionCase({ prompt: "Which editor do I prefer?", entry: { @@ -389,12 +769,38 @@ test("auto-recall falls back to a bounded deterministic digest when endpoint com exitCode: 1, }); - assert.match(result.output.hookSpecificOutput.additionalContext, /Use Vim/); - assert.doesNotMatch(result.output.hookSpecificOutput.additionalContext, //); - assert.equal(result.compressorCalls, 1); + assert.deepEqual(result.output, {}); + assert.equal(result.compressorCalls, 2); + assert.deepEqual(result.cachedProfile.failedModels, [ + "gpt-5.3-codex-spark", + "gpt-5.6-luna", + ]); }); -test("auto-recall preserves recalled memory when compressor spawn throws synchronously", async () => { +test("compressor retries share one total timeout budget", async () => { + const result = await runEndpointCompressionCase({ + prompt: "Which editor do I prefer?", + entry: { + uri: "viking://user/zeus/memories/preferences/editor.md", + score: 0.91, + type: "preferences", + mode: "summary", + summary: "Use Vim", + }, + rendered: "Use Vim", + compressorOutput: "unused", + extraEnv: { + FAKE_CODEX_HANG: "1", + OPENVIKING_RECALL_COMPRESS_TIMEOUT_MS: "1000", + }, + }); + + assert.deepEqual(result.output, {}); + assert.equal(result.compressorCalls, 2); + assert.ok(result.elapsedMs < 1800, `shared timeout budget took ${result.elapsedMs}ms`); +}); + +test("auto-recall fails closed when compressor spawn throws synchronously", async () => { const preloadDir = await mkdtemp(join(tmpdir(), "ov-sync-spawn-failure-")); const preloadPath = join(preloadDir, "throw-codex-spawn.cjs"); await writeFile(preloadPath, ` @@ -425,14 +831,33 @@ syncBuiltinESMExports(); extraEnv: { NODE_OPTIONS: `--require=${preloadPath}` }, }); - assert.match(result.output.hookSpecificOutput.additionalContext, /Use Vim/); - assert.doesNotMatch(result.output.hookSpecificOutput.additionalContext, //); + assert.deepEqual(result.output, {}); assert.equal(result.compressorCalls, 0); } finally { await rm(preloadDir, { recursive: true, force: true }); } }); +test("auto-recall keeps deterministic recall when compression is explicitly disabled", async () => { + const result = await runEndpointCompressionCase({ + prompt: "Which editor do I prefer?", + entry: { + uri: "viking://user/zeus/memories/preferences/editor.md", + score: 0.91, + type: "preferences", + mode: "summary", + summary: "Use Vim", + }, + rendered: "Use Vim", + compressorOutput: "unused", + extraEnv: { OPENVIKING_RECALL_COMPRESS_MODEL: "off" }, + }); + + assert.match(result.output.hookSpecificOutput.additionalContext, /Use Vim/); + assert.doesNotMatch(result.output.hookSpecificOutput.additionalContext, //); + assert.equal(result.compressorCalls, 0); +}); + test("auto-recall expands configured user in memory search target", async () => { const stateDir = await mkdtemp(join(tmpdir(), "ov-auto-recall-user-target-")); const requests = []; diff --git a/examples/codex-memory-plugin/scripts/pre-compact-capture.mjs b/examples/codex-memory-plugin/scripts/pre-compact-capture.mjs index c6406d4567..fb3227df4c 100644 --- a/examples/codex-memory-plugin/scripts/pre-compact-capture.mjs +++ b/examples/codex-memory-plugin/scripts/pre-compact-capture.mjs @@ -26,13 +26,17 @@ import { } from "./capture-utils.mjs"; import { loadConfig } from "./config.mjs"; import { createLogger } from "./debug-log.mjs"; -import { loadState, resolveOvSessionId, saveState } from "./session-state.mjs"; +import { resolveOvSessionId, withStateTransaction } from "./session-state.mjs"; import { sendSessionMessages } from "./shared/batch-send.mjs"; import { resolveEffectivePeerId } from "./shared/workspace-peer.mjs"; const cfg = loadConfig(); const { log, logError } = createLogger("pre-compact"); let activePeerId = cfg.peerId || ""; +const PRECOMPACT_STATE_LOCK_TIMEOUT_MS = (() => { + const configured = Number(process.env.OPENVIKING_PRECOMPACT_STATE_LOCK_TIMEOUT_MS); + return Number.isFinite(configured) && configured > 0 ? Math.floor(configured) : 20_000; +})(); function output(obj) { process.stdout.write(JSON.stringify(obj) + "\n"); @@ -107,7 +111,7 @@ async function readTranscriptTurns(transcriptPath) { } } -async function appendTurns(ovSessionId, turns) { +async function appendTurns(ovSessionId, turns, state, save) { const payloads = turns.map((turn) => { const body = turn.parts?.length ? { role: turn.role, parts: turn.parts } @@ -115,7 +119,15 @@ async function appendTurns(ovSessionId, turns) { if (activePeerId) body.peer_id = activePeerId; return body; }); - const r = await sendSessionMessages(fetchJSONRes, ovSessionId, payloads); + const r = await sendSessionMessages(fetchJSONRes, ovSessionId, payloads, { + // Persist after every successful batch/message. If Codex terminates this + // hook after the server accepted a prefix, the next hook starts after that + // durable prefix instead of replaying it into the OpenViking session. + onSent: async (count) => { + state.capturedTurnCount += count; + await save(state); + }, + }); return r.sent; } @@ -140,94 +152,100 @@ async function main() { const sessionId = input.session_id || "unknown"; const transcriptPath = input.transcript_path || null; const trigger = input.trigger || "auto"; - const state = await loadState(sessionId); - activePeerId = cfg.peerId || state.workspacePeerId || resolveEffectivePeerId({ cfg, cwd: process.cwd() }).peerId; - log("start", { sessionId, transcriptPath, trigger, hasPeer: Boolean(activePeerId) }); - - const health = await fetchJSON("/health"); - if (!health) { - logError("health_check", "server unreachable"); - noop(); - return; - } - - const allTurns = await readTranscriptTurns(transcriptPath); - const newTurns = allTurns.slice(state.capturedTurnCount); + try { + await withStateTransaction(sessionId, async ({ state, save }) => { + activePeerId = cfg.peerId || state.workspacePeerId || resolveEffectivePeerId({ cfg, cwd: process.cwd() }).peerId; + log("start", { sessionId, transcriptPath, trigger, hasPeer: Boolean(activePeerId) }); - log("transcript_parse", { - totalTurns: allTurns.length, - previouslyCaptured: state.capturedTurnCount, - newTurns: newTurns.length, - }); + const health = await fetchJSON("/health"); + if (!health) { + logError("health_check", "server unreachable"); + noop(); + return; + } - if (allTurns.length === 0 && !state.ovSessionId) { - log("skip", { stage: "nothing_to_commit", reason: "no transcript and no open OV session" }); - noop(); - return; - } + const allTurns = await readTranscriptTurns(transcriptPath); + const newTurns = allTurns.slice(state.capturedTurnCount); - if (newTurns.length > 0 && !state.ovSessionId && cfg.captureMode === "keyword" && !hasCaptureKeyword(newTurns)) { - log("skip", { stage: "capture_mode", reason: "keyword mode without capture trigger" }); - await saveState(state); - noop(); - return; - } + log("transcript_parse", { + totalTurns: allTurns.length, + previouslyCaptured: state.capturedTurnCount, + newTurns: newTurns.length, + }); - if (newTurns.length > 0) { - const ovSessionId = resolveOvSessionId(state); - if (!ovSessionId) { - logError("resolve_ov_session", "failed to derive OV session id for catch-up"); + if (allTurns.length === 0 && !state.ovSessionId) { + log("skip", { stage: "nothing_to_commit", reason: "no transcript and no open OV session" }); noop(); return; } - const added = await appendTurns(ovSessionId, newTurns); - state.capturedTurnCount += added; - log("appended_catchup", { ovSessionId, added }); - if (added < newTurns.length) { - logError("append_failed_keep_state", { ovSessionId, attempted: newTurns.length, added }); - await saveState(state); - noop(`pre-compact catch-up append incomplete for ${ovSessionId}; state preserved for retry`); + + if (newTurns.length > 0 && !state.ovSessionId && cfg.captureMode === "keyword" && !hasCaptureKeyword(newTurns)) { + log("skip", { stage: "capture_mode", reason: "keyword mode without capture trigger" }); + await save(state); + noop(); return; } - } - if (!state.ovSessionId) { - log("skip", { stage: "commit", reason: "no OV session for this codex session" }); - await saveState(state); - noop(); - return; - } - - const ovSessionId = state.ovSessionId; - const commit = await fetchJSON( - `/api/v1/sessions/${encodeURIComponent(ovSessionId)}/commit`, - { method: "POST", body: JSON.stringify({}) }, - ); - - // Commit failure handling (see DESIGN.md "Commit failure"): if /commit - // fails (server unreachable, non-2xx, timeout) we MUST NOT reset - // ovSessionId — keep state intact so the next sweep / SessionStart can - // retry. A transient OV outage shouldn't lose a session's memory. - if (!commit) { - logError("commit_failed_keep_state", { ovSessionId }); - await saveState(state); // bumps lastUpdatedAt only, keeps ovSessionId - noop(`pre-compact commit attempted on ${ovSessionId}; result unavailable (state preserved for retry)`); - return; - } + if (newTurns.length > 0) { + const ovSessionId = resolveOvSessionId(state); + if (!ovSessionId) { + logError("resolve_ov_session", "failed to derive OV session id for catch-up"); + noop(); + return; + } + const added = await appendTurns(ovSessionId, newTurns, state, save); + log("appended_catchup", { ovSessionId, added }); + if (added < newTurns.length) { + logError("append_failed_keep_state", { ovSessionId, attempted: newTurns.length, added }); + await save(state); + noop(`pre-compact catch-up append incomplete for ${ovSessionId}; state preserved for retry`); + return; + } + } - log("commit", { - ovSessionId, - archived: commit.archived ?? false, - taskId: commit.task_id, - status: commit.status, - }); + if (!state.ovSessionId) { + log("skip", { stage: "commit", reason: "no OV session for this codex session" }); + await save(state); + noop(); + return; + } - // Reset OV session for the post-compact half. Keep capturedTurnCount so - // we don't re-capture pre-compact turns when Stop fires next. - state.ovSessionId = null; - await saveState(state); + const ovSessionId = state.ovSessionId; + const commit = await fetchJSON( + `/api/v1/sessions/${encodeURIComponent(ovSessionId)}/commit`, + { method: "POST", body: JSON.stringify({}) }, + ); + + // Commit failure handling (see DESIGN.md "Commit failure"): if /commit + // fails (server unreachable, non-2xx, timeout) we MUST NOT reset + // ovSessionId — keep state intact so the next sweep / SessionStart can + // retry. A transient OV outage shouldn't lose a session's memory. + if (!commit) { + logError("commit_failed_keep_state", { ovSessionId }); + await save(state); // bumps lastUpdatedAt only, keeps ovSessionId + noop(`pre-compact commit attempted on ${ovSessionId}; result unavailable (state preserved for retry)`); + return; + } - noop(`OpenViking session ${ovSessionId} is committed`); + log("commit", { + ovSessionId, + archived: commit.archived ?? false, + taskId: commit.task_id, + status: commit.status, + }); + + // Reset OV session for the post-compact half. Keep capturedTurnCount so + // we don't re-capture pre-compact turns when Stop fires next. + state.ovSessionId = null; + await save(state); + + noop(`OpenViking session ${ovSessionId} is committed`); + }, { lockTimeoutMs: PRECOMPACT_STATE_LOCK_TIMEOUT_MS }); + } catch (error) { + if (error?.code !== "OPENVIKING_STATE_LOCK_TIMEOUT") throw error; + log("state_lock_timeout", { sessionId, timeoutMs: PRECOMPACT_STATE_LOCK_TIMEOUT_MS }); + noop("OpenViking pre-compact capture deferred because another same-session writer is still active; durable state is preserved for the next hook"); + } } function hasCaptureKeyword(turns) { diff --git a/examples/codex-memory-plugin/scripts/pre-compact-capture.test.mjs b/examples/codex-memory-plugin/scripts/pre-compact-capture.test.mjs new file mode 100644 index 0000000000..5ff1727c5c --- /dev/null +++ b/examples/codex-memory-plugin/scripts/pre-compact-capture.test.mjs @@ -0,0 +1,171 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import http from "node:http"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import test from "node:test"; +import { fileURLToPath } from "node:url"; + +import { withStateTransaction } from "./session-state.mjs"; + +const SCRIPT_DIR = dirname(fileURLToPath(import.meta.url)); + +function readRequestBody(req) { + return new Promise((resolve, reject) => { + const chunks = []; + req.on("data", (chunk) => chunks.push(chunk)); + req.on("end", () => { + try { + resolve(JSON.parse(Buffer.concat(chunks).toString("utf-8") || "null")); + } catch (error) { + reject(error); + } + }); + req.on("error", reject); + }); +} + +function writeJson(res, value) { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify(value)); +} + +async function withMockOpenViking(handler, fn) { + const server = http.createServer((req, res) => { + handler(req, res).catch((error) => { + res.writeHead(500, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "error", error: String(error?.stack || error) })); + }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + try { + return await fn(`http://127.0.0.1:${server.address().port}`); + } finally { + await new Promise((resolve) => server.close(resolve)); + } +} + +function startPreCompact(input, env) { + const cleanEnv = { ...process.env }; + for (const key of Object.keys(cleanEnv)) { + if (key.startsWith("OPENVIKING_")) delete cleanEnv[key]; + } + const child = spawn(process.execPath, [join(SCRIPT_DIR, "pre-compact-capture.mjs")], { + env: { ...cleanEnv, ...env }, + stdio: ["pipe", "pipe", "pipe"], + }); + let stdout = ""; + let stderr = ""; + child.stdout.on("data", (chunk) => { stdout += chunk.toString(); }); + child.stderr.on("data", (chunk) => { stderr += chunk.toString(); }); + const closed = new Promise((resolve, reject) => { + child.on("error", reject); + child.on("close", (code, signal) => resolve({ code, signal, stdout, stderr })); + }); + child.stdin.end(JSON.stringify(input)); + return { child, closed }; +} + +test("pre-compact persists each accepted batch before attempting the next one", async () => { + const stateDir = await mkdtemp(join(tmpdir(), "ov-pre-compact-crash-")); + const transcriptPath = join(stateDir, "transcript.jsonl"); + const batches = []; + let markSecondBatch; + const secondBatchSeen = new Promise((resolve) => { markSecondBatch = resolve; }); + + try { + const entries = Array.from({ length: 101 }, (_, index) => ({ + payload: { + message: { role: "user", content: `durable pre-compact turn ${index}` }, + }, + })); + await writeFile(transcriptPath, entries.map((entry) => JSON.stringify(entry)).join("\n")); + + await withMockOpenViking(async (req, res) => { + const url = new URL(req.url, "http://127.0.0.1"); + if (req.method === "GET" && url.pathname === "/health") { + writeJson(res, { status: "ok", result: { ok: true } }); + return; + } + if (req.method === "POST" && url.pathname.endsWith("/messages/batch")) { + batches.push(await readRequestBody(req)); + if (batches.length === 1) { + writeJson(res, { status: "ok", result: { ok: true } }); + } else { + // Seeing the next request proves the first batch's awaited onSent + // callback (including its atomic state save) has completed. + markSecondBatch(); + } + return; + } + res.writeHead(404, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "error", error: "not found" })); + }, async (baseUrl) => { + const running = startPreCompact( + { session_id: "precompact-crash", transcript_path: transcriptPath, trigger: "auto" }, + { + OPENVIKING_AUTO_COMMIT_ON_COMPACT: "1", + OPENVIKING_CODEX_STATE_DIR: stateDir, + OPENVIKING_CONFIG_FILE: join(stateDir, "missing-ov.conf"), + OPENVIKING_CLI_CONFIG_FILE: join(stateDir, "missing-ovcli.conf"), + OPENVIKING_CREDENTIAL_SOURCE: "env", + OPENVIKING_CAPTURE_TIMEOUT_MS: "5000", + OPENVIKING_TIMEOUT_MS: "5000", + OPENVIKING_URL: baseUrl, + }, + ); + await secondBatchSeen; + running.child.kill("SIGKILL"); + const exit = await running.closed; + assert.equal(exit.signal, "SIGKILL", exit.stderr); + }); + + assert.equal(batches[0].messages.length, 100); + assert.equal(batches[1].messages.length, 1); + const state = JSON.parse(await readFile(join(stateDir, "precompact-crash.json"), "utf-8")); + assert.equal(state.capturedTurnCount, 100); + assert.equal(state.revision, 1); + } finally { + await rm(stateDir, { recursive: true, force: true }); + } +}); + +test("pre-compact reports lock contention before consuming its whole hook deadline", async () => { + const stateDir = await mkdtemp(join(tmpdir(), "ov-pre-compact-lock-timeout-")); + const previousStateDir = process.env.OPENVIKING_CODEX_STATE_DIR; + process.env.OPENVIKING_CODEX_STATE_DIR = stateDir; + let releaseHolder; + let markHeld; + const held = new Promise((resolve) => { markHeld = resolve; }); + const gate = new Promise((resolve) => { releaseHolder = resolve; }); + const holder = withStateTransaction("contended", async () => { + markHeld(); + await gate; + }); + + try { + await held; + const running = startPreCompact( + { session_id: "contended", transcript_path: join(stateDir, "missing.jsonl") }, + { + OPENVIKING_AUTO_COMMIT_ON_COMPACT: "1", + OPENVIKING_CODEX_STATE_DIR: stateDir, + OPENVIKING_CONFIG_FILE: join(stateDir, "missing-ov.conf"), + OPENVIKING_CLI_CONFIG_FILE: join(stateDir, "missing-ovcli.conf"), + OPENVIKING_CREDENTIAL_SOURCE: "env", + OPENVIKING_PRECOMPACT_STATE_LOCK_TIMEOUT_MS: "30", + }, + ); + const exit = await running.closed; + assert.equal(exit.code, 0, exit.stderr); + const output = JSON.parse(exit.stdout.trim()); + assert.match(output.systemMessage, /another same-session writer is still active/); + } finally { + releaseHolder(); + await holder; + if (previousStateDir === undefined) delete process.env.OPENVIKING_CODEX_STATE_DIR; + else process.env.OPENVIKING_CODEX_STATE_DIR = previousStateDir; + await rm(stateDir, { recursive: true, force: true }); + } +}); diff --git a/examples/codex-memory-plugin/scripts/recall-compressor-profile.mjs b/examples/codex-memory-plugin/scripts/recall-compressor-profile.mjs index 5e7b32e8fd..04ceb7a2d9 100644 --- a/examples/codex-memory-plugin/scripts/recall-compressor-profile.mjs +++ b/examples/codex-memory-plugin/scripts/recall-compressor-profile.mjs @@ -1,13 +1,16 @@ +import { randomUUID } from "node:crypto"; import { readFile, rm } from "node:fs/promises"; import { homedir } from "node:os"; import { join } from "node:path"; import { mkdir, rename, writeFile } from "node:fs/promises"; -import { getStateDir } from "./session-state.mjs"; +import { getStateDir, withStateTransaction } from "./session-state.mjs"; const DEFAULT_PRIMARY = { model: "gpt-5.3-codex-spark", thinking: "default", source: "default_primary" }; const DEFAULT_FALLBACK = { model: "gpt-5.6-luna", thinking: "low", source: "default_fallback" }; -const PROFILE_SCHEMA_VERSION = 3; +const PROFILE_SCHEMA_VERSION = 4; const DEFAULT_CODEX_HOME = join(homedir(), ".codex"); +const PROFILE_CACHE_LOCK_ID = "__recall-compressor-profile-cache__"; +const PROFILE_CACHE_LOCK_TIMEOUT_MS = 2_000; function isOff(value) { return /^(?:0|false|no|off|none|disabled)$/i.test(String(value || "").trim()); @@ -148,17 +151,46 @@ export async function loadCachedRecallCompressorProfile(cfg) { } } -async function saveRecallCompressorProfile(cfg, profile) { +async function withProfileCacheLock(callback) { + return withStateTransaction( + PROFILE_CACHE_LOCK_ID, + callback, + { lockTimeoutMs: PROFILE_CACHE_LOCK_TIMEOUT_MS }, + ); +} + +async function writeRecallCompressorProfile(cfg, profile) { await mkdir(getStateDir(), { recursive: true }); const final = profilePath(); - const tmp = `${final}.tmp`; - await writeFile(tmp, JSON.stringify({ - schemaVersion: PROFILE_SCHEMA_VERSION, - checkedAt: Date.now(), - configKey: configKey(cfg), - profile, - })); - await rename(tmp, final); + const tmp = `${final}.tmp-${process.pid}-${randomUUID()}`; + try { + await writeFile(tmp, JSON.stringify({ + schemaVersion: PROFILE_SCHEMA_VERSION, + checkedAt: Date.now(), + configKey: configKey(cfg), + profile, + })); + await rename(tmp, final); + } finally { + // The rename removes tmp on success. On write/rename failure, leave no + // process-shared fixed temp (or unique crash residue) for a later hook to + // publish accidentally. + await rm(tmp, { force: true }).catch(() => {}); + } +} + +async function saveRecallCompressorProfile(cfg, profile) { + return withProfileCacheLock(() => writeRecallCompressorProfile(cfg, profile)); +} + +/** Persist a compressor that completed successfully so later hooks start there. */ +export async function cacheRecallCompressorProfile(cfg, profile) { + try { + await withProfileCacheLock(() => writeRecallCompressorProfile( + cfg, + { ...profile, enabled: true }, + )); + } catch { /* best effort */ } } /** @@ -168,26 +200,44 @@ async function saveRecallCompressorProfile(cfg, profile) { */ export async function invalidateRecallCompressorProfileCache() { try { - await rm(profilePath(), { force: true }); + await withProfileCacheLock(() => rm(profilePath(), { force: true })); } catch { /* best effort */ } } /** - * Record a runtime compress failure as a disabled profile. UserPromptSubmit - * reads the cached profile directly, so writing `enabled: false` here makes - * subsequent UPS calls within the same codex session skip compress (and - * fall back to deterministic digest) instead of paying ~recallCompressTimeoutMs - * per turn on a guaranteed-to-fail spawn. The next SessionStart's cache- - * first detect treats `source === "runtime_failed"` as a cache miss and - * re-resolves from the current catalogue, so a transient failure does not - * permanently disable compress across codex restarts. + * Record compressor models that failed at runtime. UserPromptSubmit skips those + * models within the same Codex session, trying any untried candidate before it + * fails closed. The next SessionStart treats `source === "runtime_failed"` as + * a cache miss and re-resolves from the current catalogue, so a transient + * failure does not permanently disable compression across Codex restarts. */ -export async function markRecallCompressorRuntimeFailed(cfg, { failedModel = "" } = {}) { +export async function markRecallCompressorRuntimeFailed( + cfg, + { failedModel = "", failedModels = [] } = {}, +) { try { - await saveRecallCompressorProfile(cfg, { - enabled: false, - source: "runtime_failed", - failedModel: String(failedModel || ""), + await withProfileCacheLock(async () => { + const current = await loadCachedRecallCompressorProfile(cfg); + const previousFailures = current?.source === "runtime_failed" + ? current.failedModels || [] + : []; + const models = [...new Set( + [...previousFailures, ...failedModels, failedModel] + .map((model) => String(model || "").trim()) + .filter(Boolean), + )]; + + // A concurrent prompt may already have promoted a fallback that this + // attempt never tried. Keep that positive result; only disable an + // enabled cached profile when this attempt actually failed its model. + if (current?.enabled && current.model && !models.includes(current.model)) return; + + await writeRecallCompressorProfile(cfg, { + enabled: false, + source: "runtime_failed", + failedModel: models.at(-1) || "", + failedModels: models, + }); }); } catch { /* best effort */ } } @@ -250,8 +300,8 @@ export async function resolveRecallCompressorProfile(cfg, logger = {}, env = pro * SessionStart no longer probes models with a subprocess on every fire. * Instead it loads the cached profile and only resolves (a cheap * models_cache.json read) when nothing is cached or the cache is stale. - * The runtime compress path invalidates the cache on failure, which is - * what triggers the next re-resolve. + * The runtime compress path records exhausted candidates, which is what + * triggers the next SessionStart re-resolve. */ export async function detectRecallCompressorProfile(cfg, logger = {}, env = process.env) { const { log } = logger; @@ -261,7 +311,10 @@ export async function detectRecallCompressorProfile(cfg, logger = {}, env = proc return cached; } if (cached && cached.source === "runtime_failed") { - log?.("compress_profile_recover", { failedModel: cached.failedModel || "" }); + log?.("compress_profile_recover", { + failedModel: cached.failedModel || "", + failedModels: cached.failedModels || [], + }); } if (!cfg.recallCompressDetectOnStartup) { log?.("compress_profile_skip", { reason: "detect disabled and no usable cache" }); diff --git a/examples/codex-memory-plugin/scripts/recall-compressor-profile.test.mjs b/examples/codex-memory-plugin/scripts/recall-compressor-profile.test.mjs index 2ea21a1f37..afa7375f4f 100644 --- a/examples/codex-memory-plugin/scripts/recall-compressor-profile.test.mjs +++ b/examples/codex-memory-plugin/scripts/recall-compressor-profile.test.mjs @@ -1,11 +1,12 @@ import assert from "node:assert/strict"; -import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import test from "node:test"; import { detectRecallCompressorProfile, + cacheRecallCompressorProfile, invalidateRecallCompressorProfileCache, loadCachedRecallCompressorProfile, loadCodexModelsCache, @@ -226,12 +227,68 @@ test("markRecallCompressorRuntimeFailed writes a disabled profile cached for UPS await writeModelsCache(codexHome, ["gpt-5.6-luna"]); await resolveRecallCompressorProfile(baseCfg(), {}, { CODEX_HOME: codexHome }); - await markRecallCompressorRuntimeFailed(baseCfg(), { failedModel: "gpt-5.3-codex-spark" }); + await markRecallCompressorRuntimeFailed(baseCfg(), { failedModel: "gpt-5.6-luna" }); const cached = await loadCachedRecallCompressorProfile(baseCfg()); assert.ok(cached, "expected cached profile to exist"); assert.equal(cached.enabled, false); assert.equal(cached.source, "runtime_failed"); - assert.equal(cached.failedModel, "gpt-5.3-codex-spark"); + assert.equal(cached.failedModel, "gpt-5.6-luna"); + assert.deepEqual(cached.failedModels, ["gpt-5.6-luna"]); + }); +}); + +test("cacheRecallCompressorProfile promotes a working runtime fallback", async () => { + await withTempState(async () => { + await cacheRecallCompressorProfile(baseCfg(), { + model: "gpt-5.6-luna", + thinking: "low", + source: "default_fallback", + }); + const cached = await loadCachedRecallCompressorProfile(baseCfg()); + assert.equal(cached.enabled, true); + assert.equal(cached.model, "gpt-5.6-luna"); + }); +}); + +test("concurrent profile resolutions publish valid cache without shared tmp races", async () => { + await withTempState(async ({ stateDir, codexHome }) => { + await writeModelsCache(codexHome, ["gpt-5.3-codex-spark", "gpt-5.6-luna"]); + await Promise.all(Array.from({ length: 8 }, () => resolveRecallCompressorProfile( + baseCfg(), + {}, + { CODEX_HOME: codexHome }, + ))); + + const persisted = JSON.parse( + await readFile(join(stateDir, "recall-compressor-profile.json"), "utf-8"), + ); + assert.equal(persisted.profile.enabled, true); + assert.equal(persisted.profile.model, "gpt-5.3-codex-spark"); + assert.deepEqual( + (await readdir(stateDir)).filter((file) => file.startsWith("recall-compressor-profile.json.tmp-")), + [], + ); + }); +}); + +test("concurrent unrelated failure cannot overwrite a promoted fallback", async () => { + await withTempState(async () => { + for (let i = 0; i < 16; i += 1) { + await invalidateRecallCompressorProfileCache(); + await Promise.all([ + cacheRecallCompressorProfile(baseCfg(), { + model: "gpt-5.6-luna", + thinking: "low", + source: "default_fallback", + }), + markRecallCompressorRuntimeFailed(baseCfg(), { + failedModel: "gpt-5.3-codex-spark", + }), + ]); + const cached = await loadCachedRecallCompressorProfile(baseCfg()); + assert.equal(cached?.enabled, true); + assert.equal(cached?.model, "gpt-5.6-luna"); + } }); }); diff --git a/examples/codex-memory-plugin/scripts/session-start-commit.mjs b/examples/codex-memory-plugin/scripts/session-start-commit.mjs index b81b6ed8e4..31143ab58d 100644 --- a/examples/codex-memory-plugin/scripts/session-start-commit.mjs +++ b/examples/codex-memory-plugin/scripts/session-start-commit.mjs @@ -40,7 +40,7 @@ import { loadConfig } from "./config.mjs"; import { createLogger } from "./debug-log.mjs"; import { detectRecallCompressorProfile } from "./recall-compressor-profile.mjs"; -import { clearState, deriveOvSessionId, listStates, loadState, saveState } from "./session-state.mjs"; +import { deriveOvSessionId, listStates, loadState, withStateTransaction } from "./session-state.mjs"; import { buildProfileBlock } from "./shared/profile-inject.mjs"; import { resolveEffectivePeerId } from "./shared/workspace-peer.mjs"; @@ -233,33 +233,54 @@ async function buildResumeArchiveContext(newSessionId) { * Returns { committed: bool, ovSessionId: string|null }. */ async function commitAndClear(state, reason) { - if (state.ovSessionId) { - const ovSessionId = state.ovSessionId; - const commit = await commitOvSession(state.ovSessionId); - if (!commit) { - logError("commit_failed_keep_state", { + return withStateTransaction(state.codexSessionId, async ({ state: freshState, clear }) => { + // listStates() takes only a short scan lock and releases it before this + // remote commit lifecycle. Revalidate its snapshot after acquiring the + // full session lock so a Stop hook that refreshed this state after the scan + // cannot be committed or cleared by a stale SessionStart decision. + if ( + freshState.revision !== state.revision + || freshState.lastUpdatedAt !== state.lastUpdatedAt + || freshState.capturedTurnCount !== state.capturedTurnCount + || freshState.ovSessionId !== state.ovSessionId + ) { + log("commit_skip_state_changed", { reason, codexSessionId: state.codexSessionId, - ovSessionId: state.ovSessionId, + snapshotUpdatedAt: state.lastUpdatedAt, + currentUpdatedAt: freshState.lastUpdatedAt, }); return { committed: false, ovSessionId: null }; } - log("commit", { - reason, - codexSessionId: state.codexSessionId, - ovSessionId, - archived: commit.archived ?? false, - taskId: commit.task_id, - status: commit.status, - }); - await clearState(state.codexSessionId); - return { committed: true, ovSessionId }; - } - // No OV session attached — nothing to commit on the server, but the local - // state file is still stale and should be removed. - log("clear_no_ov", { reason, codexSessionId: state.codexSessionId }); - await clearState(state.codexSessionId); - return { committed: true, ovSessionId: null }; + + if (freshState.ovSessionId) { + const ovSessionId = freshState.ovSessionId; + const commit = await commitOvSession(ovSessionId); + if (!commit) { + logError("commit_failed_keep_state", { + reason, + codexSessionId: freshState.codexSessionId, + ovSessionId, + }); + return { committed: false, ovSessionId: null }; + } + log("commit", { + reason, + codexSessionId: freshState.codexSessionId, + ovSessionId, + archived: commit.archived ?? false, + taskId: commit.task_id, + status: commit.status, + }); + await clear(); + return { committed: true, ovSessionId }; + } + // No OV session attached — nothing to commit on the server, but the local + // state file is still stale and should be removed. + log("clear_no_ov", { reason, codexSessionId: freshState.codexSessionId }); + await clear(); + return { committed: true, ovSessionId: null }; + }); } function describeCommittedSessions(ovSessionIds) { @@ -288,10 +309,11 @@ async function main() { const effectivePeer = resolveEffectivePeerId({ cfg, cwd }); activePeerId = effectivePeer.peerId; if (newSessionId !== "unknown") { - const state = await loadState(newSessionId); - await saveState({ - ...state, - workspacePeerId: effectivePeer.source === "workspace" ? effectivePeer.peerId : "", + await withStateTransaction(newSessionId, async ({ state, save }) => { + await save({ + ...state, + workspacePeerId: effectivePeer.source === "workspace" ? effectivePeer.peerId : "", + }); }); } log("start", { diff --git a/examples/codex-memory-plugin/scripts/session-state.mjs b/examples/codex-memory-plugin/scripts/session-state.mjs index 9837c4768d..56d6c0c6aa 100644 --- a/examples/codex-memory-plugin/scripts/session-state.mjs +++ b/examples/codex-memory-plugin/scripts/session-state.mjs @@ -10,12 +10,19 @@ * State directory: $OPENVIKING_CODEX_STATE_DIR or ~/.openviking/codex-plugin-state */ -import { mkdir, readFile, readdir, rename, rm, writeFile } from "node:fs/promises"; -import { homedir } from "node:os"; -import { join } from "node:path"; +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, readdir, readlink, rename, rm, stat, writeFile } from "node:fs/promises"; +import { homedir, hostname } from "node:os"; +import { basename, join } from "node:path"; import { deriveCodexSessionId } from "./shared/session-model.mjs"; const DEFAULT_STATE_DIR = join(homedir(), ".openviking", "codex-plugin-state"); +const DEFAULT_LOCK_TIMEOUT_MS = 60_000; +const DEFAULT_LOCK_RETRY_MS = 25; +const LIST_STATES_LOCK_TIMEOUT_MS = 250; +const AVAILABLE_LOCK_FILE = "available"; +const OWNER_LOCK_PREFIX = "owner-"; +const CLAIM_LOCK_PREFIX = "claim-"; export function getStateDir() { return process.env.OPENVIKING_CODEX_STATE_DIR || DEFAULT_STATE_DIR; @@ -42,62 +49,542 @@ function statePath(codexSessionId) { return join(getStateDir(), `${safeId(codexSessionId)}.json`); } -function defaultState(codexSessionId) { +function lockPath(codexSessionId) { + return `${statePath(codexSessionId)}.lock`; +} + +function revisionPath(codexSessionId) { + return join(lockPath(codexSessionId), "revision"); +} + +function tempBasename(codexSessionId) { + return `${basename(statePath(codexSessionId))}.tmp`; +} + +function defaultState(codexSessionId, revision = 0) { const now = Date.now(); return { codexSessionId, ovSessionId: null, workspacePeerId: "", capturedTurnCount: 0, + revision, createdAt: now, lastUpdatedAt: now, }; } -export async function loadState(codexSessionId) { +function positiveDuration(value, fallback, minimum = 1) { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed >= minimum ? Math.floor(parsed) : fallback; +} + +function stateRevision(state) { + const revision = Number(state?.revision); + return Number.isSafeInteger(revision) && revision >= 0 ? revision : 0; +} + +function lockOptions(options = {}) { + return { + lockTimeoutMs: positiveDuration( + options.lockTimeoutMs ?? process.env.OPENVIKING_CODEX_STATE_LOCK_TIMEOUT_MS, + DEFAULT_LOCK_TIMEOUT_MS, + ), + retryDelayMs: positiveDuration(options.retryDelayMs, DEFAULT_LOCK_RETRY_MS), + }; +} + +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function readJsonFile(path) { + try { + return JSON.parse(await readFile(path, "utf-8")); + } catch { + return null; + } +} + +let machineIdentityPromise; +function currentMachineIdentity() { + if (!machineIdentityPromise) { + machineIdentityPromise = (async () => { + if (process.platform !== "linux") return null; + for (const path of ["/etc/machine-id", "/var/lib/dbus/machine-id"]) { + try { + const value = (await readFile(path, "utf-8")).trim(); + if (value) return `linux:${value}`; + } catch { /* try the next stable machine-id location */ } + } + return null; + })(); + } + return machineIdentityPromise; +} + +function parseProcStat(raw) { + const end = raw.lastIndexOf(")"); + const firstSpace = raw.indexOf(" "); + if (end < 0 || firstSpace < 0) return null; + const namespacePid = raw.slice(0, firstSpace); + // After comm, index 0 is field 3; starttime is field 22 (index 19). + const startTime = raw.slice(end + 1).trim().split(/\s+/)[19]; + return namespacePid && startTime ? { namespacePid, startTime } : null; +} + +async function currentPidNamespaceIdentity() { + if (process.platform !== "linux") return null; + try { + return await readlink("/proc/self/ns/pid"); + } catch { + return null; + } +} + +async function currentProcessStartIdentity() { + if (process.platform !== "linux") return null; + try { + // `/proc/self/stat` reveals the PID in this process's mounted PID namespace. + // The claim separately records that namespace identity; include Linux's + // boot id so a reboot cannot make (namespace-pid,start-ticks) look reused. + const [raw, bootId] = await Promise.all([ + readFile("/proc/self/stat", "utf-8"), + readFile("/proc/sys/kernel/random/boot_id", "utf-8"), + ]); + const parsed = parseProcStat(raw); + return parsed + ? `linux:${bootId.trim()}:${parsed.namespacePid}:${parsed.startTime}` + : null; + } catch { + return null; + } +} + +async function recordedProcessIsAlive(identity) { + const match = /^linux:([^:]+):(\d+):(\d+)$/.exec(String(identity || "")); + if (!match) return null; + const [, expectedBootId, namespacePid, expectedStartTime] = match; + let bootId; try { - const raw = await readFile(statePath(codexSessionId), "utf-8"); + bootId = await readFile("/proc/sys/kernel/random/boot_id", "utf-8"); + } catch { + // Losing access to the boot identity is not proof the owner died. + return null; + } + if (bootId.trim() !== expectedBootId) return false; + try { + const raw = await readFile(`/proc/${namespacePid}/stat`, "utf-8"); + return parseProcStat(raw)?.startTime === expectedStartTime; + } catch (error) { + if (error?.code === "ENOENT") return false; + return null; + } +} + +async function ensureLockRoot(codexSessionId) { + const root = lockPath(codexSessionId); + const staging = `${root}.init-${process.pid}-${randomUUID()}`; + let installed = false; + try { + // Build a complete lock directory off to the side, then publish it with one + // directory rename. A crash can leave an unreferenced staging directory but + // never a canonical lock root missing its baton. + await mkdir(staging); + await writeFile(join(staging, AVAILABLE_LOCK_FILE), "", { flag: "wx" }); + await rename(staging, root); + installed = true; + } catch (error) { + if (error?.code !== "EEXIST" && error?.code !== "ENOTEMPTY") throw error; + } finally { + if (!installed) await rm(staging, { recursive: true, force: true }).catch(() => {}); + } + return root; +} + +function ownerToken(file) { + return file.startsWith(OWNER_LOCK_PREFIX) ? file.slice(OWNER_LOCK_PREFIX.length) : ""; +} + +function claimToken(file) { + return file.startsWith(CLAIM_LOCK_PREFIX) && file.endsWith(".json") + ? file.slice(CLAIM_LOCK_PREFIX.length, -".json".length) + : ""; +} + +async function claimOwnerIsDefinitelyDead(owner) { + // Automatic recovery needs identities that have the same meaning to both + // processes. Hostname/PID alone is insufficient across NFS clients, container + // PID namespaces, PID reuse, and non-Linux hosts. + if (process.platform !== "linux" || !owner || owner.hostname !== hostname()) return false; + const [localMachineIdentity, localPidNamespaceIdentity] = await Promise.all([ + currentMachineIdentity(), + currentPidNamespaceIdentity(), + ]); + if ( + !owner.machineIdentity + || !localMachineIdentity + || owner.machineIdentity !== localMachineIdentity + || !owner.pidNamespaceIdentity + || !localPidNamespaceIdentity + || owner.pidNamespaceIdentity !== localPidNamespaceIdentity + ) return false; + return await recordedProcessIsAlive(owner.processStartIdentity) === false; +} + +async function cleanupDeadUnownedClaims(root, keepToken = "") { + let files; + try { + files = await readdir(root); + } catch { + return; + } + const ownerTokens = new Set(files.map(ownerToken).filter(Boolean)); + await Promise.all(files.map(async (file) => { + const token = claimToken(file); + if (!token || token === keepToken || ownerTokens.has(token)) return; + const path = join(root, file); + const owner = await readJsonFile(path); + if (await claimOwnerIsDefinitelyDead(owner)) { + // A live contender can have a claim before it owns the baton. Remove only + // an ownerless claim whose recorded process is positively proven dead; + // remote, legacy, and non-Linux claims remain fail-safe/manual cleanup. + await rm(path, { force: true }).catch(() => {}); + } + })); +} + +async function reclaimDeadSameHostOwner(root) { + let files; + try { + files = await readdir(root); + } catch { + return false; + } + // If the baton is available there is no active owner to reclaim. Any + // leftover claim metadata is harmless and cleaned by its contender. + if (files.includes(AVAILABLE_LOCK_FILE)) return false; + + for (const file of files) { + const token = ownerToken(file); + if (!token) continue; + const ownerPath = join(root, file); + const claimPath = join(root, `${CLAIM_LOCK_PREFIX}${token}.json`); + const owner = await readJsonFile(claimPath); + // Never expire a different host's lease from wall-clock age. Without a + // remote fencing token, timeout-based stealing can overlap a paused but + // still-live owner. Cross-host crash recovery is therefore explicit/manual. + if (!owner || owner.hostname !== hostname()) continue; + const dead = await claimOwnerIsDefinitelyDead(owner); + // On non-Linux platforms hostname + PID is not a trustworthy host/process + // identity (hostnames can collide and PIDs can be reused). Without Linux's + // machine/PID-namespace/start tuple, leave recovery to an operator. + if (!dead) continue; + + try { + // The source path is unique to this owner token. If the owner released + // and another process acquired in between our check and this rename, this + // exact source is gone; we can never rename/delete the new owner's path. + await rename(ownerPath, join(root, AVAILABLE_LOCK_FILE)); + } catch (error) { + if (error?.code === "ENOENT") continue; + throw error; + } + await rm(claimPath, { force: true }).catch(() => {}); + return true; + } + return false; +} + +async function acquireStateLock(codexSessionId, options = {}) { + const { lockTimeoutMs, retryDelayMs } = lockOptions(options); + await mkdir(getStateDir(), { recursive: true }); + const root = await ensureLockRoot(codexSessionId); + const token = randomUUID(); + const ownerPath = join(root, `${OWNER_LOCK_PREFIX}${token}`); + const claimPath = join(root, `${CLAIM_LOCK_PREFIX}${token}.json`); + const deadline = Date.now() + lockTimeoutMs; + let acquired = false; + + await writeFile(claimPath, JSON.stringify({ + token, + pid: process.pid, + hostname: hostname(), + machineIdentity: await currentMachineIdentity(), + pidNamespaceIdentity: await currentPidNamespaceIdentity(), + processStartIdentity: await currentProcessStartIdentity(), + acquiredAt: Date.now(), + }), { flag: "wx" }); + + try { + while (true) { + try { + await rename(join(root, AVAILABLE_LOCK_FILE), ownerPath); + acquired = true; + // A contender killed before acquiring leaves only claim metadata. The + // current fenced owner can safely prune claims whose recorded Linux + // process is conclusively dead, without touching live/remote contenders. + await cleanupDeadUnownedClaims(root, token).catch(() => {}); + break; + } catch (error) { + if (error?.code !== "ENOENT") throw error; + } + + if (await reclaimDeadSameHostOwner(root)) continue; + if (Date.now() >= deadline) { + const error = new Error(`Timed out waiting for OpenViking Codex state lock: ${codexSessionId}`); + error.code = "OPENVIKING_STATE_LOCK_TIMEOUT"; + throw error; + } + await sleep(Math.min(retryDelayMs, Math.max(1, deadline - Date.now()))); + } + } finally { + if (!acquired) await rm(claimPath, { force: true }).catch(() => {}); + } + + return async () => { + try { + // Release moves only our unique source path. If a confirmed-dead-owner + // recovery already moved it, ENOENT is harmless and cannot affect the + // current owner, whose source path contains a different token. + await rename(ownerPath, join(root, AVAILABLE_LOCK_FILE)); + } catch (error) { + if (error?.code !== "ENOENT") throw error; + } finally { + await rm(claimPath, { force: true }).catch(() => {}); + } + }; +} + +function isTempFile(file, codexSessionId) { + const prefix = tempBasename(codexSessionId); + return file === prefix || file.startsWith(`${prefix}-`); +} + +async function listTempPaths(codexSessionId) { + try { + const dir = getStateDir(); + return (await readdir(dir)) + .filter((file) => isTempFile(file, codexSessionId)) + .map((file) => join(dir, file)); + } catch { + return []; + } +} + +async function readCompleteState(path, codexSessionId) { + try { + const raw = await readFile(path, "utf-8"); const parsed = JSON.parse(raw); - return { ...defaultState(codexSessionId), ...parsed }; + if (!parsed || parsed.codexSessionId !== codexSessionId) return null; + const info = await stat(path); + return { + path, + state: { ...defaultState(codexSessionId), ...parsed }, + revision: stateRevision(parsed), + writtenAt: Number(parsed.lastUpdatedAt) || info.mtimeMs, + mtimeMs: info.mtimeMs, + }; } catch { - return defaultState(codexSessionId); + return null; } } -export async function saveState(state) { +async function removeTempFiles(codexSessionId, except = "") { + const paths = await listTempPaths(codexSessionId); + await Promise.all(paths + .filter((path) => path !== except) + .map((path) => rm(path, { force: true }).catch(() => {}))); +} + +async function readPersistedRevision(codexSessionId) { + try { + const revision = Number(await readFile(revisionPath(codexSessionId), "utf-8")); + return Number.isSafeInteger(revision) && revision >= 0 ? revision : 0; + } catch { + return 0; + } +} + +async function writePersistedRevision(codexSessionId, revision) { + const final = revisionPath(codexSessionId); + const tmp = `${final}.tmp-${process.pid}-${randomUUID()}`; + let renamed = false; + try { + await writeFile(tmp, String(revision)); + await rename(tmp, final); + renamed = true; + } finally { + if (!renamed) await rm(tmp, { force: true }).catch(() => {}); + } +} + +async function loadStateUnlocked(codexSessionId) { + const persistedRevision = await readPersistedRevision(codexSessionId); + const final = statePath(codexSessionId); + const persisted = await readCompleteState(final, codexSessionId); + // A process can die after writeFile completes but before rename, leaving a + // valid old final beside a newer complete temp. Compare both; do not discard + // the completed update merely because the old final is still parseable. This + // also understands the legacy fixed `.json.tmp` name. + const tempCandidates = (await Promise.all( + (await listTempPaths(codexSessionId)).map((path) => readCompleteState(path, codexSessionId)), + )).filter(Boolean); + const candidates = [persisted, ...tempCandidates] + .filter(Boolean) + .sort((a, b) => (b.revision - a.revision) + || (b.writtenAt - a.writtenAt) + || (b.mtimeMs - a.mtimeMs)); + const winner = candidates[0]; + if (!winner) { + await removeTempFiles(codexSessionId); + return defaultState(codexSessionId, persistedRevision); + } + + if (winner.revision < persistedRevision) { + // clearState writes a tombstone revision before removing the final file. + // If it crashed between those steps, the higher counter proves this + // otherwise-valid final/temp belongs to the pre-clear generation. + await removeTempFiles(codexSessionId); + await rm(final, { force: true }); + return defaultState(codexSessionId, persistedRevision); + } + + if (winner.path !== final) await rename(winner.path, final); + await removeTempFiles(codexSessionId); + if (winner.revision > persistedRevision) { + // Repair the complementary save crash window: state rename succeeded but + // the monotonic counter update did not. + await writePersistedRevision(codexSessionId, winner.revision); + } + return winner.state; +} + +async function saveStateUnlocked(state, revision = stateRevision(state) + 1) { if (!state || !state.codexSessionId) return; await mkdir(getStateDir(), { recursive: true }); - const next = { ...state, lastUpdatedAt: Date.now() }; - // Atomic write (tmpfile + rename) so a crash mid-write can't leave a - // truncated/corrupt state file. See DESIGN.md "State file schema". + const next = { ...state, revision, lastUpdatedAt: Date.now() }; const final = statePath(state.codexSessionId); - const tmp = `${final}.tmp`; - await writeFile(tmp, JSON.stringify(next)); - await rename(tmp, final); + const tmp = `${final}.tmp-${process.pid}-${randomUUID()}`; + let renamed = false; + try { + await writeFile(tmp, JSON.stringify(next)); + await rename(tmp, final); + renamed = true; + await writePersistedRevision(state.codexSessionId, revision); + Object.assign(state, next); + return next; + } finally { + if (!renamed) await rm(tmp, { force: true }).catch(() => {}); + } } -export async function clearState(codexSessionId) { +async function clearStateUnlocked(codexSessionId, revision) { + // Persist a tombstone generation before deleting the old final. On restart, + // loadStateUnlocked ignores any state whose revision predates this counter. + await writePersistedRevision(codexSessionId, revision); + // Remove recoverable temps first. If this process is killed during clear, + // either the old final still exists or all state is gone; an old temp can + // never be left alone and resurrect a state that was intentionally cleared. + await removeTempFiles(codexSessionId); + await rm(statePath(codexSessionId), { force: true }); +} + +/** + * Run a complete read/remote-work/write lifecycle under one cross-process, + * per-session lock. Codex invokes different hooks in different Node processes, + * so a module-local promise queue is not sufficient. + */ +export async function withStateTransaction(codexSessionId, callback, options = {}) { + if (!codexSessionId) throw new Error("codexSessionId is required for a state transaction"); + if (typeof callback !== "function") throw new TypeError("state transaction callback must be a function"); + const release = await acquireStateLock(codexSessionId, options); try { - await rm(statePath(codexSessionId), { force: true }); - } catch { /* best effort */ } + const state = await loadStateUnlocked(codexSessionId); + let revision = stateRevision(state); + return await callback({ + state, + save: async (next = state) => { + if (!next || next.codexSessionId !== codexSessionId) { + const error = new Error(`State transaction ${codexSessionId} cannot save another session`); + error.code = "OPENVIKING_STATE_SESSION_MISMATCH"; + throw error; + } + const saved = await saveStateUnlocked(next, Math.max(revision, stateRevision(next)) + 1); + revision = saved.revision; + // Keep the transaction's originally-loaded snapshot revision current + // even when a caller saved a replacement object via `{ ...state }`. + state.revision = saved.revision; + state.lastUpdatedAt = saved.lastUpdatedAt; + return saved; + }, + clear: async () => { + revision += 1; + await clearStateUnlocked(codexSessionId, revision); + state.revision = revision; + return revision; + }, + }); + } finally { + await release(); + } +} + +export async function loadState(codexSessionId) { + return withStateTransaction(codexSessionId, ({ state }) => state); +} + +export async function saveState(state) { + if (!state || !state.codexSessionId) return; + return withStateTransaction(state.codexSessionId, ({ state: current, save }) => { + if (stateRevision(state) !== stateRevision(current)) { + const error = new Error(`Refusing stale OpenViking Codex state save: ${state.codexSessionId}`); + error.code = "OPENVIKING_STALE_STATE_SAVE"; + throw error; + } + return save(state); + }); +} + +export async function clearState(codexSessionId) { + return withStateTransaction(codexSessionId, ({ clear }) => clear()); } export async function listStates() { try { const dir = getStateDir(); const files = await readdir(dir); - const out = []; + const sessionIds = new Set(); for (const file of files) { - // .json only — atomic writes briefly create `.json.tmp`, skipped - // by this check (endsWith(".json") is false for ".json.tmp"). - if (!file.endsWith(".json")) continue; + // Scan final files plus recoverable legacy/unique temps. A crash may + // leave only a complete temp, and SessionStart's orphan sweep must still + // discover that session rather than silently stranding it forever. + if (!file.endsWith(".json") && !/\.json\.tmp(?:-|$)/.test(file)) continue; try { const raw = await readFile(join(dir, file), "utf-8"); const parsed = JSON.parse(raw); - if (parsed?.codexSessionId) out.push(parsed); + if (parsed?.codexSessionId) sessionIds.add(parsed.codexSessionId); } catch { /* skip */ } } - return out; + + const states = await Promise.all([...sessionIds].map(async (codexSessionId) => { + try { + return await withStateTransaction(codexSessionId, async ({ state }) => { + // loadStateUnlocked (called by the transaction) has now selected and + // recovered the newest complete candidate. Re-check that a final file + // really exists so a concurrently-cleared candidate does not turn into + // a phantom default state in the sweep. + return await readCompleteState(statePath(codexSessionId), codexSessionId) + ? state + : null; + }, { lockTimeoutMs: LIST_STATES_LOCK_TIMEOUT_MS }); + } catch { + // A busy live hook is safer to omit from this sweep; the next + // SessionStart will retry after it releases the session lock. + return null; + } + })); + return states.filter(Boolean); } catch { return []; } diff --git a/examples/codex-memory-plugin/scripts/session-state.test.mjs b/examples/codex-memory-plugin/scripts/session-state.test.mjs new file mode 100644 index 0000000000..268861eafd --- /dev/null +++ b/examples/codex-memory-plugin/scripts/session-state.test.mjs @@ -0,0 +1,448 @@ +import assert from "node:assert/strict"; +import { execFile } from "node:child_process"; +import { mkdir, mkdtemp, readFile, readdir, readlink, rename, rm, writeFile } from "node:fs/promises"; +import { hostname, tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { promisify } from "node:util"; + +import { + clearState, + listStates, + loadState, + saveState, + withStateTransaction, +} from "./session-state.mjs"; + +const execFileAsync = promisify(execFile); +const STATE_MODULE_URL = new URL("./session-state.mjs", import.meta.url).href; + +async function useTemporaryStateDir(t) { + const previous = process.env.OPENVIKING_CODEX_STATE_DIR; + const dir = await mkdtemp(join(tmpdir(), "openviking-codex-state-")); + process.env.OPENVIKING_CODEX_STATE_DIR = dir; + t.after(async () => { + if (previous === undefined) delete process.env.OPENVIKING_CODEX_STATE_DIR; + else process.env.OPENVIKING_CODEX_STATE_DIR = previous; + await rm(dir, { recursive: true, force: true }); + }); + return dir; +} + +async function waitForFile(path, timeoutMs = 2_000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + try { + await readFile(path); + return; + } catch { /* worker has not acquired the lock yet */ } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error(`timed out waiting for ${path}`); +} + +function transactionWorkerSource() { + return ` + import { writeFile } from "node:fs/promises"; + import { withStateTransaction } from ${JSON.stringify(STATE_MODULE_URL)}; + await withStateTransaction(process.env.SESSION_ID, async ({ state, save }) => { + if (process.env.MARKER) await writeFile(process.env.MARKER, "locked"); + const observed = state.capturedTurnCount; + await new Promise((resolve) => setTimeout(resolve, Number(process.env.DELAY_MS || 0))); + state.capturedTurnCount = observed + 1; + await save(); + }); + `; +} + +async function installLockOwner(dir, sessionId, owner) { + const token = owner.token; + const root = join(dir, `${sessionId}.json.lock`); + await mkdir(root); + await writeFile(join(root, "available"), ""); + await writeFile(join(root, `claim-${token}.json`), JSON.stringify(owner)); + await rename(join(root, "available"), join(root, `owner-${token}`)); + return root; +} + +async function lockEntries(path) { + return (await readdir(path)).sort(); +} + +async function currentLinuxLockIdentity() { + if (process.platform !== "linux") return {}; + let machineId = ""; + for (const path of ["/etc/machine-id", "/var/lib/dbus/machine-id"]) { + try { + machineId = (await readFile(path, "utf-8")).trim(); + if (machineId) break; + } catch { /* try the next machine-id location */ } + } + const rawStat = await readFile("/proc/self/stat", "utf-8"); + const namespacePid = rawStat.slice(0, rawStat.indexOf(" ")); + const startTime = rawStat.slice(rawStat.lastIndexOf(")") + 1).trim().split(/\s+/)[19]; + const bootId = (await readFile("/proc/sys/kernel/random/boot_id", "utf-8")).trim(); + return { + machineIdentity: machineId ? `linux:${machineId}` : null, + pidNamespaceIdentity: await readlink("/proc/self/ns/pid"), + bootId, + processStartIdentity: `linux:${bootId}:${namespacePid}:${startTime}`, + }; +} + +test("serializes same-session transactions across Node processes", async (t) => { + const dir = await useTemporaryStateDir(t); + const marker = join(dir, "first.locked"); + const baseEnv = { + ...process.env, + OPENVIKING_CODEX_STATE_DIR: dir, + SESSION_ID: "shared-session", + }; + + const first = execFileAsync(process.execPath, ["--input-type=module", "-e", transactionWorkerSource()], { + env: { ...baseEnv, MARKER: marker, DELAY_MS: "150" }, + }); + await waitForFile(marker); + const second = execFileAsync(process.execPath, ["--input-type=module", "-e", transactionWorkerSource()], { + env: { ...baseEnv, DELAY_MS: "0" }, + }); + await Promise.all([first, second]); + + const state = await loadState("shared-session"); + assert.equal(state.capturedTurnCount, 2); + assert.equal(state.revision, 2); + assert.deepEqual((await readdir(dir)).filter((name) => name.includes(".tmp")), []); + assert.deepEqual(await lockEntries(join(dir, "shared-session.json.lock")), ["available", "revision"]); +}); + +test("clear waits for an in-flight transaction and cannot be undone by a stale save", async (t) => { + const dir = await useTemporaryStateDir(t); + const marker = join(dir, "writer.locked"); + const writer = execFileAsync(process.execPath, ["--input-type=module", "-e", transactionWorkerSource()], { + env: { + ...process.env, + OPENVIKING_CODEX_STATE_DIR: dir, + SESSION_ID: "clear-race", + MARKER: marker, + DELAY_MS: "150", + }, + }); + await waitForFile(marker); + + const clearing = clearState("clear-race"); + await Promise.all([writer, clearing]); + assert.equal((await loadState("clear-race")).capturedTurnCount, 0); + assert.equal((await readdir(dir)).includes("clear-race.json"), false); + assert.deepEqual(await lockEntries(join(dir, "clear-race.json.lock")), ["available", "revision"]); +}); + +test("recovers only the newest complete session-matching temp file", async (t) => { + const dir = await useTemporaryStateDir(t); + await writeFile(join(dir, "recover.json"), JSON.stringify({ + codexSessionId: "recover", + capturedTurnCount: 1, + revision: 1, + lastUpdatedAt: 500, + })); + await writeFile(join(dir, "recover.json.tmp"), JSON.stringify({ + codexSessionId: "recover", + capturedTurnCount: 3, + revision: 1, + lastUpdatedAt: 1_000, + })); + await writeFile(join(dir, "recover.json.tmp-123-complete"), JSON.stringify({ + codexSessionId: "recover", + capturedTurnCount: 7, + revision: 2, + lastUpdatedAt: 200, + })); + await writeFile(join(dir, "recover.json.tmp-123-wrong-session"), JSON.stringify({ + codexSessionId: "other", + capturedTurnCount: 99, + lastUpdatedAt: 300, + })); + await writeFile(join(dir, "recover.json.tmp-123-truncated"), "{"); + + const recovered = await loadState("recover"); + assert.equal(recovered.capturedTurnCount, 7); + assert.equal(JSON.parse(await readFile(join(dir, "recover.json"), "utf-8")).capturedTurnCount, 7); + assert.deepEqual((await readdir(dir)).filter((name) => name.startsWith("recover.json.tmp")), []); +}); + +test("listStates recovers a complete temp even when no final file exists", async (t) => { + const dir = await useTemporaryStateDir(t); + await writeFile(join(dir, "orphan.json.tmp-456-complete"), JSON.stringify({ + codexSessionId: "orphan", + ovSessionId: "cx-orphan", + capturedTurnCount: 4, + lastUpdatedAt: Date.now() - 60_000, + })); + await writeFile(join(dir, "broken.json.tmp-456-incomplete"), "{"); + + const states = await listStates(); + assert.deepEqual(states.map((state) => state.codexSessionId), ["orphan"]); + assert.equal(states[0].capturedTurnCount, 4); + assert.equal(JSON.parse(await readFile(join(dir, "orphan.json"), "utf-8")).ovSessionId, "cx-orphan"); + assert.equal((await readdir(dir)).some((name) => name.startsWith("orphan.json.tmp")), false); +}); + +test("listStates quickly omits a busy session instead of consuming the hook deadline", async (t) => { + const dir = await useTemporaryStateDir(t); + await writeFile(join(dir, "busy-scan.json"), JSON.stringify({ + codexSessionId: "busy-scan", + capturedTurnCount: 4, + revision: 1, + lastUpdatedAt: Date.now(), + })); + const path = await installLockOwner(dir, "busy-scan", { + token: "live-scan-owner", + pid: process.pid, + hostname: hostname(), + // A legacy/missing identity is deliberately non-reclaimable on Linux. + processStartIdentity: null, + acquiredAt: Date.now(), + }); + + const startedAt = Date.now(); + const states = await listStates(); + const elapsedMs = Date.now() - startedAt; + + assert.deepEqual(states, []); + assert.ok(elapsedMs < 1_500, `busy scan took ${elapsedMs}ms`); + assert.ok((await readdir(path)).includes("owner-live-scan-owner")); +}); + +test("reclaims a lock whose same-Linux-PID-namespace owner process is gone", async (t) => { + if (process.platform !== "linux") { + t.skip("automatic dead-owner recovery is Linux-specific"); + return; + } + const dir = await useTemporaryStateDir(t); + const linuxIdentity = await currentLinuxLockIdentity(); + const path = await installLockOwner(dir, "stale", { + token: "abandoned", + pid: 2_147_483_647, + hostname: hostname(), + machineIdentity: linuxIdentity.machineIdentity, + pidNamespaceIdentity: linuxIdentity.pidNamespaceIdentity, + processStartIdentity: process.platform === "linux" + ? `linux:${linuxIdentity.bootId}:2147483647:1` + : null, + acquiredAt: Date.now(), + }); + + await saveState({ codexSessionId: "stale", capturedTurnCount: 1 }); + assert.equal((await loadState("stale")).capturedTurnCount, 1); + assert.deepEqual(await lockEntries(path), ["available", "revision"]); +}); + +test("does not steal a live lock and times out with a stable error code", async (t) => { + const dir = await useTemporaryStateDir(t); + const path = await installLockOwner(dir, "live", { + token: "live-owner", + pid: process.pid, + hostname: hostname(), + processStartIdentity: null, + acquiredAt: Date.now(), + }); + + await assert.rejects( + withStateTransaction("live", async () => {}, { + lockTimeoutMs: 30, + retryDelayMs: 5, + }), + (error) => error?.code === "OPENVIKING_STATE_LOCK_TIMEOUT", + ); + assert.ok((await readdir(path)).includes("owner-live-owner")); +}); + +test("uses process start identity to reclaim a reused Linux PID safely", async (t) => { + if (process.platform !== "linux") { + t.skip("automatic dead-owner recovery is Linux-specific"); + return; + } + const dir = await useTemporaryStateDir(t); + const linuxIdentity = await currentLinuxLockIdentity(); + const path = await installLockOwner(dir, "pid-reuse", { + token: "old-process", + pid: process.pid, + hostname: hostname(), + machineIdentity: linuxIdentity.machineIdentity, + pidNamespaceIdentity: linuxIdentity.pidNamespaceIdentity, + processStartIdentity: "linux:not-the-current-boot:1:1", + acquiredAt: Date.now(), + }); + + await withStateTransaction("pid-reuse", async ({ state, save }) => { + state.capturedTurnCount = 1; + await save(); + }); + assert.equal((await loadState("pid-reuse")).capturedTurnCount, 1); + assert.deepEqual(await lockEntries(path), ["available", "revision"]); +}); + +test("never auto-reclaims a Linux owner from a missing or different PID namespace", async (t) => { + if (process.platform !== "linux") { + t.skip("PID namespace identity is Linux-specific"); + return; + } + const dir = await useTemporaryStateDir(t); + const linuxIdentity = await currentLinuxLockIdentity(); + for (const [sessionId, pidNamespaceIdentity] of [ + ["legacy-namespace", null], + ["different-namespace", "pid:[different-namespace]"], + ]) { + const path = await installLockOwner(dir, sessionId, { + token: "unfenced-owner", + pid: 2_147_483_647, + hostname: hostname(), + machineIdentity: linuxIdentity.machineIdentity, + pidNamespaceIdentity, + processStartIdentity: `linux:${linuxIdentity.bootId}:2147483647:1`, + acquiredAt: 1, + }); + + await assert.rejects( + withStateTransaction(sessionId, async () => {}, { + lockTimeoutMs: 30, + retryDelayMs: 5, + }), + (error) => error?.code === "OPENVIKING_STATE_LOCK_TIMEOUT", + ); + assert.ok((await readdir(path)).includes("owner-unfenced-owner")); + } +}); + +test("never timeout-reclaims another host's owner without server fencing", async (t) => { + const dir = await useTemporaryStateDir(t); + const path = await installLockOwner(dir, "remote", { + token: "remote-owner", + pid: 2_147_483_647, + hostname: "another-host.example", + processStartIdentity: "1", + acquiredAt: 1, + }); + + await assert.rejects( + withStateTransaction("remote", async () => {}, { lockTimeoutMs: 30, retryDelayMs: 5 }), + (error) => error?.code === "OPENVIKING_STATE_LOCK_TIMEOUT", + ); + assert.ok((await readdir(path)).includes("owner-remote-owner")); +}); + +test("prunes only positively-dead ownerless claims after acquiring the baton", async (t) => { + if (process.platform !== "linux") { + t.skip("automatic dead-claim cleanup is Linux-specific"); + return; + } + const dir = await useTemporaryStateDir(t); + const sessionId = "orphan-claims"; + await loadState(sessionId); + const path = join(dir, `${sessionId}.json.lock`); + const linuxIdentity = await currentLinuxLockIdentity(); + await writeFile(join(path, "claim-dead-contender.json"), JSON.stringify({ + token: "dead-contender", + hostname: hostname(), + machineIdentity: linuxIdentity.machineIdentity, + pidNamespaceIdentity: linuxIdentity.pidNamespaceIdentity, + processStartIdentity: `linux:${linuxIdentity.bootId}:2147483647:1`, + })); + await writeFile(join(path, "claim-live-contender.json"), JSON.stringify({ + token: "live-contender", + hostname: hostname(), + machineIdentity: linuxIdentity.machineIdentity, + pidNamespaceIdentity: linuxIdentity.pidNamespaceIdentity, + processStartIdentity: linuxIdentity.processStartIdentity, + })); + await writeFile(join(path, "claim-remote-contender.json"), JSON.stringify({ + token: "remote-contender", + hostname: "another-host.example", + machineIdentity: linuxIdentity.machineIdentity, + pidNamespaceIdentity: linuxIdentity.pidNamespaceIdentity, + processStartIdentity: `linux:${linuxIdentity.bootId}:2147483647:1`, + })); + + await loadState(sessionId); + const files = await lockEntries(path); + assert.equal(files.includes("claim-dead-contender.json"), false); + assert.equal(files.includes("claim-live-contender.json"), true); + assert.equal(files.includes("claim-remote-contender.json"), true); +}); + +test("standalone save rejects a stale revision instead of overwriting newer state", async (t) => { + await useTemporaryStateDir(t); + const stale = await loadState("cas"); + await saveState({ ...stale, capturedTurnCount: 1 }); + + await assert.rejects( + saveState({ ...stale, capturedTurnCount: 99 }), + (error) => error?.code === "OPENVIKING_STALE_STATE_SAVE", + ); + const current = await loadState("cas"); + assert.equal(current.capturedTurnCount, 1); + assert.equal(current.revision, 1); +}); + +test("a transaction cannot write another session while holding the wrong lock", async (t) => { + const dir = await useTemporaryStateDir(t); + await assert.rejects( + withStateTransaction("session-a", async ({ save }) => save({ + codexSessionId: "session-b", + capturedTurnCount: 99, + })), + (error) => error?.code === "OPENVIKING_STATE_SESSION_MISMATCH", + ); + assert.equal((await readdir(dir)).includes("session-b.json"), false); + assert.equal((await loadState("session-a")).capturedTurnCount, 0); +}); + +test("revision remains monotonic across clear and state recreation", async (t) => { + await useTemporaryStateDir(t); + const initial = await loadState("generation"); + const first = await saveState({ ...initial, capturedTurnCount: 1 }); + assert.equal(first.revision, 1); + + await clearState("generation"); + const cleared = await loadState("generation"); + assert.equal(cleared.revision, 2); + assert.equal(cleared.capturedTurnCount, 0); + + const recreated = await saveState({ ...cleared, capturedTurnCount: 1 }); + assert.equal(recreated.revision, 3); +}); + +test("a persisted clear tombstone suppresses an old final after a crash", async (t) => { + const dir = await useTemporaryStateDir(t); + await loadState("tombstone"); // initializes the permanent lock root + await writeFile(join(dir, "tombstone.json"), JSON.stringify({ + codexSessionId: "tombstone", + capturedTurnCount: 8, + revision: 4, + lastUpdatedAt: Date.now(), + })); + await writeFile(join(dir, "tombstone.json.lock", "revision"), "5"); + + const state = await loadState("tombstone"); + assert.equal(state.revision, 5); + assert.equal(state.capturedTurnCount, 0); + assert.equal((await readdir(dir)).includes("tombstone.json"), false); +}); + +test("different session ids remain independently concurrent", async (t) => { + await useTemporaryStateDir(t); + let releaseFirst; + let signalFirst; + const firstEntered = new Promise((resolve) => { signalFirst = resolve; }); + const firstGate = new Promise((resolve) => { releaseFirst = resolve; }); + const first = withStateTransaction("session-a", async () => { + signalFirst(); + await firstGate; + }); + await firstEntered; + + let secondEntered = false; + await withStateTransaction("session-b", async () => { secondEntered = true; }); + assert.equal(secondEntered, true); + releaseFirst(); + await first; +}); diff --git a/examples/codex-memory-plugin/scripts/shared/recall-core.mjs b/examples/codex-memory-plugin/scripts/shared/recall-core.mjs index a6e94eef5a..8c53dcea96 100644 --- a/examples/codex-memory-plugin/scripts/shared/recall-core.mjs +++ b/examples/codex-memory-plugin/scripts/shared/recall-core.mjs @@ -40,7 +40,8 @@ function scaleQuotas(limit, weights) { const order = Object.keys(weights); const quotas = Object.fromEntries(order.map((key) => [key, 0])); if (slots < order.length) { - for (const key of order) quotas[key] = 1; + const priority = [...order].sort((a, b) => weights[b] - weights[a]); + for (const key of priority.slice(0, slots)) quotas[key] = 1; return quotas; } @@ -61,10 +62,12 @@ function scaleQuotas(limit, weights) { } function legacyMemoryQuotas(limit) { - return { - ...scaleQuotas(limit, { events: 10, entities: 10, preferences: 3 }), - experiences: 0, - }; + return scaleQuotas(limit, { + events: 10, + entities: 10, + experiences: 3, + preferences: 3, + }); } function codingQuotas(limit) { diff --git a/examples/codex-memory-plugin/servers/experience-tools.mjs b/examples/codex-memory-plugin/servers/experience-tools.mjs index 5df915ce25..aa9505ebc9 100644 --- a/examples/codex-memory-plugin/servers/experience-tools.mjs +++ b/examples/codex-memory-plugin/servers/experience-tools.mjs @@ -3,6 +3,7 @@ const DEFAULT_LIMIT = 5; const MAX_LIMIT = 20; const DEFAULT_TIMEOUT_MS = 15000; const EXPERIENCE_SIDECAR_FILENAMES = new Set([".abstract.md", ".overview.md", ".relations.json"]); +const EXCLUDED_EXPERIENCE_STATUSES = new Set(["deprecated", "archived"]); const EXPERIENCE_TOOL_DEFINITIONS = [ { @@ -131,6 +132,96 @@ async function readJsonResponse(response) { return payload; } +function isRecord(value) { + return Boolean(value) && typeof value === "object" && !Array.isArray(value); +} + +function sanitizeMetadataValue(value, depth = 0) { + if (value === null || typeof value === "boolean") return value; + if (typeof value === "number") return Number.isFinite(value) ? value : undefined; + if (typeof value === "string") return value.slice(0, 4096); + if (depth >= 3) return undefined; + if (Array.isArray(value)) { + return value.slice(0, 50) + .map((item) => sanitizeMetadataValue(item, depth + 1)) + .filter((item) => item !== undefined); + } + if (!isRecord(value)) return undefined; + const result = {}; + for (const [key, item] of Object.entries(value).slice(0, 50)) { + const safe = sanitizeMetadataValue(item, depth + 1); + if (safe !== undefined) result[String(key).slice(0, 128)] = safe; + } + return result; +} + +function safeExperienceMetadata(...sources) { + const metadata = {}; + for (const source of sources) { + if (!isRecord(source)) continue; + if (typeof source.status === "string" && source.status.trim()) { + metadata.status = source.status.trim().toLowerCase().slice(0, 64); + } + const version = Number(source.version); + if (Number.isSafeInteger(version) && version >= 0) metadata.version = version; + if (typeof source.curated_at === "string" && source.curated_at.trim()) { + metadata.curated_at = source.curated_at.trim().slice(0, 512); + } + const curatedFrom = source.curated_from ?? source.from; + if (curatedFrom !== undefined) { + const safe = sanitizeMetadataValue(curatedFrom); + if (safe !== undefined) metadata.curated_from = safe; + } + } + return metadata; +} + +function parseExperienceDocument(rawContent, ...metadataSources) { + const raw = String(rawContent || ""); + const match = /\s*$/i.exec(raw); + let fields = {}; + let content = raw; + let metadataValid = true; + if (match) { + content = raw.slice(0, match.index).trimEnd(); + try { + const parsed = JSON.parse(match[1].trim()); + if (isRecord(parsed)) fields = parsed; + else metadataValid = false; + } catch { + metadataValid = false; + } + } + return { + content, + metadata: safeExperienceMetadata(...metadataSources, fields), + metadataValid, + }; +} + +function documentFromReadPayload(payload) { + const value = payload?.result; + if (!isRecord(value)) return parseExperienceDocument(value); + const content = value.raw_content ?? value.raw ?? value.content ?? ""; + return parseExperienceDocument(content, value.attrs, value.metadata, value); +} + +function withMetadata(payload, metadata) { + return Object.keys(metadata).length > 0 ? { ...payload, metadata } : payload; +} + +async function fetchExperienceDocument(uri, config, fetchImpl) { + const url = new URL(`${normalizedBaseUrl(config)}/api/v1/content/read`); + url.searchParams.set("uri", uri); + url.searchParams.set("raw", "true"); + const response = await fetchImpl(url.toString(), { + method: "GET", + headers: requestHeaders(config), + signal: requestSignal(config), + }); + return documentFromReadPayload(await readJsonResponse(response)); +} + async function searchExperience(args, config, fetchImpl) { const query = String(args?.query || "").trim(); if (!query) return errorResult("search_experience requires a non-empty query"); @@ -147,14 +238,29 @@ async function searchExperience(args, config, fetchImpl) { }); const payload = await readJsonResponse(response); const memories = Array.isArray(payload?.result?.memories) ? payload.result.memories : []; - const results = memories - .filter((item) => isExperienceUri(item?.uri, config?.user)) - .map((item) => ({ + const candidates = memories.filter((item) => isExperienceUri(item?.uri, config?.user)); + const hydrated = await Promise.all(candidates.map(async (item) => { + const directMetadata = safeExperienceMetadata(item?.attrs, item?.metadata, item); + try { + const document = await fetchExperienceDocument(item.uri, config, fetchImpl); + if (!document.metadataValid) return null; + return { item, metadata: { ...directMetadata, ...document.metadata } }; + } catch { + // Lifecycle state is authoritative only in the raw memory document. + // A missing/failed hydration must not turn an archived or deprecated + // Experience back into executable guidance. + return null; + } + })); + const results = hydrated + .filter(Boolean) + .filter(({ metadata }) => !EXCLUDED_EXPERIENCE_STATUSES.has(metadata.status)) + .map(({ item, metadata }) => withMetadata({ uri: item.uri, title: titleFromUri(item.uri), score: Number.isFinite(Number(item.score)) ? Number(item.score) : 0, snippet: String(item.abstract || item.overview || ""), - })); + }, metadata)); return result({ results }); } catch (error) { return errorResult(error instanceof Error ? error.message : error); @@ -167,14 +273,16 @@ async function readExperience(args, config, fetchImpl) { return errorResult("read_experience requires an Experience URI owned by the current user"); } try { - const url = `${normalizedBaseUrl(config)}/api/v1/content/read?uri=${encodeURIComponent(uri)}`; - const response = await fetchImpl(url, { - method: "GET", - headers: requestHeaders(config), - signal: requestSignal(config), - }); - const payload = await readJsonResponse(response); - return result({ uri, content: String(payload?.result || "") }); + const document = await fetchExperienceDocument(uri, config, fetchImpl); + if (!document.metadataValid) { + return errorResult("read_experience refuses malformed Experience lifecycle metadata"); + } + if (EXCLUDED_EXPERIENCE_STATUSES.has(document.metadata.status)) { + return errorResult( + `read_experience refuses Experience status=${document.metadata.status}`, + ); + } + return result(withMetadata({ uri, content: document.content }, document.metadata)); } catch (error) { return errorResult(error instanceof Error ? error.message : error); } diff --git a/examples/codex-memory-plugin/servers/experience-tools.test.mjs b/examples/codex-memory-plugin/servers/experience-tools.test.mjs index 24099d096d..073ce52105 100644 --- a/examples/codex-memory-plugin/servers/experience-tools.test.mjs +++ b/examples/codex-memory-plugin/servers/experience-tools.test.mjs @@ -31,6 +31,11 @@ test("search_experience searches only the current user's Experience directory", const provider = createExperienceToolProvider({ fetchImpl: async (url, options) => { calls.push({ url: String(url), options }); + if (options.method === "GET") { + return new Response(JSON.stringify({ + result: '## Approach\n先验证身份。\n\n', + }), { status: 200, headers: { "content-type": "application/json" } }); + } return new Response(JSON.stringify({ ok: true, result: { @@ -78,7 +83,7 @@ test("search_experience searches only the current user's Experience directory", { config }, ); - assert.equal(calls.length, 1); + assert.equal(calls.length, 2); assert.equal(calls[0].url, "http://openviking.test/api/v1/search/find"); assert.equal(calls[0].options.method, "POST"); assert.equal(calls[0].options.headers.Authorization, "Bearer test-key"); @@ -97,6 +102,12 @@ test("search_experience searches only the current user's Experience directory", title: "无订单号换货处理", score: 0.82, snippet: "先验证身份,再逐个定位订单。", + metadata: { + status: "active", + version: 3, + curated_at: "2026-08-10", + curated_from: ["case-17"], + }, }, ], }); @@ -108,7 +119,10 @@ test("read_experience reads a canonical Experience URI", async () => { const provider = createExperienceToolProvider({ fetchImpl: async (url, options) => { calls.push({ url: String(url), options }); - return new Response(JSON.stringify({ ok: true, result: "## Approach\n先验证用户身份。" }), { + return new Response(JSON.stringify({ + ok: true, + result: '## Approach\n先验证用户身份。\n\n', + }), { status: 200, headers: { "content-type": "application/json" }, }); @@ -123,10 +137,132 @@ test("read_experience reads a canonical Experience URI", async () => { assert.equal(calls.length, 1); assert.equal( calls[0].url, - `http://openviking.test/api/v1/content/read?uri=${encodeURIComponent(uri)}`, + `http://openviking.test/api/v1/content/read?uri=${encodeURIComponent(uri)}&raw=true`, ); assert.equal(calls[0].options.method, "GET"); - assert.deepEqual(toolPayload(result), { uri, content: "## Approach\n先验证用户身份。" }); + assert.deepEqual(toolPayload(result), { + uri, + content: "## Approach\n先验证用户身份。", + metadata: { + status: "draft", + version: 2, + curated_from: { project: "returns" }, + }, + }); +}); + +test("search_experience excludes deprecated and archived hits while keeping draft explicit", async () => { + const statuses = new Map([ + ["active.md", "active"], + ["draft.md", "draft"], + ["deprecated.md", "deprecated"], + ["archived.md", "ARCHIVED"], + ]); + const provider = createExperienceToolProvider({ + fetchImpl: async (url, options) => { + if (options.method === "POST") { + return new Response(JSON.stringify({ + result: { + memories: [...statuses.keys()].map((name, index) => ({ + uri: `viking://user/test/memories/experiences/${name}`, + score: 1 - index / 10, + abstract: name, + })), + }, + }), { status: 200 }); + } + const name = new URL(String(url)).searchParams.get("uri").split("/").at(-1); + return new Response(JSON.stringify({ + result: `content\n\n`, + }), { status: 200 }); + }, + }); + + const response = await provider.callTool( + { name: "search_experience", arguments: { query: "status" } }, + { config }, + ); + const payload = toolPayload(response); + assert.deepEqual(payload.results.map((item) => item.title), ["active", "draft"]); + assert.equal(payload.results[1].metadata.status, "draft"); +}); + +test("search_experience drops candidates when authoritative raw reads fail", async () => { + const provider = createExperienceToolProvider({ + fetchImpl: async (_url, options) => { + if (options.method === "GET") return new Response("missing", { status: 404 }); + return new Response(JSON.stringify({ + result: { + memories: [{ + uri: "viking://user/test/memories/experiences/legacy.md", + score: 0.5, + abstract: "legacy server", + }], + }, + }), { status: 200 }); + }, + }); + + const response = await provider.callTool( + { name: "search_experience", arguments: { query: "legacy" } }, + { config }, + ); + assert.deepEqual(toolPayload(response).results, []); +}); + +test("Experience tools fail closed on malformed lifecycle metadata", async () => { + const uri = "viking://user/test/memories/experiences/malformed.md"; + const provider = createExperienceToolProvider({ + fetchImpl: async (_url, options) => { + if (options.method === "POST") { + return new Response(JSON.stringify({ + result: { memories: [{ uri, score: 0.7, abstract: "malformed" }] }, + }), { status: 200 }); + } + return new Response(JSON.stringify({ + result: "content\n\n", + }), { status: 200 }); + }, + }); + + const searchResponse = await provider.callTool( + { name: "search_experience", arguments: { query: "malformed" } }, + { config }, + ); + assert.deepEqual(toolPayload(searchResponse).results, []); + + const readResponse = await provider.callTool( + { name: "read_experience", arguments: { uri } }, + { config }, + ); + assert.equal(readResponse.isError, true); + assert.match(readResponse.content[0].text, /malformed Experience lifecycle metadata/); +}); + +test("read_experience rejects deprecated and archived lifecycle states", async () => { + const provider = createExperienceToolProvider({ + fetchImpl: async (url) => { + const uri = new URL(String(url)).searchParams.get("uri"); + const status = uri.endsWith("archived.md") ? "ARCHIVED" : "deprecated"; + return new Response(JSON.stringify({ + result: `content\n\n`, + }), { status: 200 }); + }, + }); + + for (const status of ["deprecated", "archived"]) { + const response = await provider.callTool( + { + name: "read_experience", + arguments: { + uri: `viking://user/test/memories/experiences/${status}.md`, + }, + }, + { config }, + ); + assert.equal(response.isError, true); + assert.match(response.content[0].text, new RegExp(`status=${status}`)); + } }); test("read_experience rejects non-Experience URIs without an HTTP request", async () => { diff --git a/examples/codex-memory-plugin/skills/ov-experience-memory/SKILL.md b/examples/codex-memory-plugin/skills/ov-experience-memory/SKILL.md index 1faa749e2d..9b5b8930ce 100644 --- a/examples/codex-memory-plugin/skills/ov-experience-memory/SKILL.md +++ b/examples/codex-memory-plugin/skills/ov-experience-memory/SKILL.md @@ -5,7 +5,7 @@ description: > experiences with search_experience, read selected experiences with read_experience, and leave standard tool parts in the committed session so OpenViking can report recall and injection usage. -version: 2026.7.9 +version: 2026.8.10 tags: - openviking - experience-memory @@ -56,7 +56,13 @@ Output schema: "uri": "viking://user//memories/experiences/example.md", "title": "example", "score": 0.82, - "snippet": "Short summary or matched situation" + "snippet": "Short summary or matched situation", + "metadata": { + "status": "production", + "version": 3, + "curated_at": "2026-08-10T00:00:00Z", + "curated_from": ["case-17"] + } } ] } @@ -94,7 +100,12 @@ Output schema: ```json { "uri": "viking://user//memories/experiences/example.md", - "content": "Experience Markdown body" + "content": "Experience Markdown body", + "metadata": { + "status": "draft", + "version": 2, + "curated_from": {"project": "example"} + } } ``` @@ -112,13 +123,28 @@ event for `tool_input.uri` or `tool_output.uri`. In this design, reading an experience through `read_experience` means the experience was injected into the prompt. +## Metadata and Status Contract + +`metadata` is optional on both tools for compatibility with older OpenViking +servers. When present, it contains only the allowlisted fields `status`, +`version`, `curated_at`, and `curated_from`; callers must not expect internal +provenance fields. `status` is normalized to lowercase. + +`search_experience` omits experiences whose status is `deprecated` or +`archived`. Results with `production`, `staging`, `draft`, an unknown status, or no status can +still be returned. Treat `draft` as provisional guidance and validate it against +the current task. `read_experience` revalidates the document metadata and rejects +a direct URI whose status is `deprecated` or `archived`; treat that tool error as +the experience being unavailable and do not inject it. + ## Recommended Flow 1. When a task begins, build a short query from the latest user instruction, current plan, active skill name, and important tool/environment context. 2. Call `search_experience` before final prompt assembly. -3. Review returned titles/snippets and select only experiences likely to affect - execution. +3. Review returned titles, snippets, and optional metadata. Select only + experiences likely to affect execution; prefer `production` entries, treat + `draft` as provisional, and skip `deprecated` or `archived` entries. 4. Call `read_experience` for selected experience URIs. 5. Inject the returned Markdown into the prompt under an explicit experience section. diff --git a/examples/memory-plugin-shared/lib/recall-core.mjs b/examples/memory-plugin-shared/lib/recall-core.mjs index 5a92081d23..2d68e8c160 100644 --- a/examples/memory-plugin-shared/lib/recall-core.mjs +++ b/examples/memory-plugin-shared/lib/recall-core.mjs @@ -39,7 +39,8 @@ function scaleQuotas(limit, weights) { const order = Object.keys(weights); const quotas = Object.fromEntries(order.map((key) => [key, 0])); if (slots < order.length) { - for (const key of order) quotas[key] = 1; + const priority = [...order].sort((a, b) => weights[b] - weights[a]); + for (const key of priority.slice(0, slots)) quotas[key] = 1; return quotas; } @@ -60,10 +61,12 @@ function scaleQuotas(limit, weights) { } function legacyMemoryQuotas(limit) { - return { - ...scaleQuotas(limit, { events: 10, entities: 10, preferences: 3 }), - experiences: 0, - }; + return scaleQuotas(limit, { + events: 10, + entities: 10, + experiences: 3, + preferences: 3, + }); } function codingQuotas(limit) { diff --git a/examples/memory-plugin-shared/recall-core.test.mjs b/examples/memory-plugin-shared/recall-core.test.mjs index a1b97ddd0c..c836a1b0e2 100644 --- a/examples/memory-plugin-shared/recall-core.test.mjs +++ b/examples/memory-plugin-shared/recall-core.test.mjs @@ -25,9 +25,9 @@ test("context requests preserve the configured recall width and server budget", recallCompressMaxInputChars: 18000, }); - assert.equal(Object.values(body.quotas).reduce((sum, quota) => sum + quota, 0), 6); + assert.equal(Object.values(body.quotas).reduce((sum, quota) => sum + quota, 0), 1); assert.equal(body.quotas.resources, 1); - assert.equal(body.quotas.skills, 1); + assert.equal(body.quotas.skills, 0); assert.equal(body.max_tokens, 800); assert.equal(body.purpose, "coding"); }); @@ -59,6 +59,44 @@ test("coding-agent fallback recall explicitly uses the 0.35 threshold", () => { const body = buildRecallEndpointBody({}); assert.equal(body.min_score, 0.35); + assert.ok(body.quotas.experiences >= 1); + assert.equal( + Object.values(body.quotas).reduce((sum, quota) => sum + quota, 0), + 10, + ); +}); + +test("legacy fallback keeps Experience within a three-result recall limit", () => { + const body = buildRecallEndpointBody({ recallLimit: 3 }); + + assert.equal( + Object.values(body.quotas).reduce((sum, quota) => sum + quota, 0), + 3, + ); + assert.equal(body.quotas.events, 1); + assert.equal(body.quotas.entities, 1); + assert.equal(body.quotas.experiences, 1); + assert.equal(body.quotas.preferences, 0); +}); + +test("configured quota totals never exceed the requested recall width", () => { + for (let limit = 1; limit <= 12; limit += 1) { + const legacy = buildRecallEndpointBody({ recallLimit: limit }); + const context = buildContextSearchBody({ + recallLimit: limit, + recallLimitConfigured: true, + }); + assert.equal( + Object.values(legacy.quotas).reduce((sum, quota) => sum + quota, 0), + limit, + `legacy limit=${limit}`, + ); + assert.equal( + Object.values(context.quotas).reduce((sum, quota) => sum + quota, 0), + limit, + `context limit=${limit}`, + ); + } }); test("buildRecallBlock injects context assembled by the server", async () => { diff --git a/examples/opencode-plugin/lib/shared/recall-core.mjs b/examples/opencode-plugin/lib/shared/recall-core.mjs index a6e94eef5a..8c53dcea96 100644 --- a/examples/opencode-plugin/lib/shared/recall-core.mjs +++ b/examples/opencode-plugin/lib/shared/recall-core.mjs @@ -40,7 +40,8 @@ function scaleQuotas(limit, weights) { const order = Object.keys(weights); const quotas = Object.fromEntries(order.map((key) => [key, 0])); if (slots < order.length) { - for (const key of order) quotas[key] = 1; + const priority = [...order].sort((a, b) => weights[b] - weights[a]); + for (const key of priority.slice(0, slots)) quotas[key] = 1; return quotas; } @@ -61,10 +62,12 @@ function scaleQuotas(limit, weights) { } function legacyMemoryQuotas(limit) { - return { - ...scaleQuotas(limit, { events: 10, entities: 10, preferences: 3 }), - experiences: 0, - }; + return scaleQuotas(limit, { + events: 10, + entities: 10, + experiences: 3, + preferences: 3, + }); } function codingQuotas(limit) { diff --git a/examples/pi-coding-agent-extension/shared/recall-core.mjs b/examples/pi-coding-agent-extension/shared/recall-core.mjs index a6e94eef5a..8c53dcea96 100644 --- a/examples/pi-coding-agent-extension/shared/recall-core.mjs +++ b/examples/pi-coding-agent-extension/shared/recall-core.mjs @@ -40,7 +40,8 @@ function scaleQuotas(limit, weights) { const order = Object.keys(weights); const quotas = Object.fromEntries(order.map((key) => [key, 0])); if (slots < order.length) { - for (const key of order) quotas[key] = 1; + const priority = [...order].sort((a, b) => weights[b] - weights[a]); + for (const key of priority.slice(0, slots)) quotas[key] = 1; return quotas; } @@ -61,10 +62,12 @@ function scaleQuotas(limit, weights) { } function legacyMemoryQuotas(limit) { - return { - ...scaleQuotas(limit, { events: 10, entities: 10, preferences: 3 }), - experiences: 0, - }; + return scaleQuotas(limit, { + events: 10, + entities: 10, + experiences: 3, + preferences: 3, + }); } function codingQuotas(limit) { diff --git a/examples/skills/ov-experience-memory/SKILL.md b/examples/skills/ov-experience-memory/SKILL.md index 1faa749e2d..9b5b8930ce 100644 --- a/examples/skills/ov-experience-memory/SKILL.md +++ b/examples/skills/ov-experience-memory/SKILL.md @@ -5,7 +5,7 @@ description: > experiences with search_experience, read selected experiences with read_experience, and leave standard tool parts in the committed session so OpenViking can report recall and injection usage. -version: 2026.7.9 +version: 2026.8.10 tags: - openviking - experience-memory @@ -56,7 +56,13 @@ Output schema: "uri": "viking://user//memories/experiences/example.md", "title": "example", "score": 0.82, - "snippet": "Short summary or matched situation" + "snippet": "Short summary or matched situation", + "metadata": { + "status": "production", + "version": 3, + "curated_at": "2026-08-10T00:00:00Z", + "curated_from": ["case-17"] + } } ] } @@ -94,7 +100,12 @@ Output schema: ```json { "uri": "viking://user//memories/experiences/example.md", - "content": "Experience Markdown body" + "content": "Experience Markdown body", + "metadata": { + "status": "draft", + "version": 2, + "curated_from": {"project": "example"} + } } ``` @@ -112,13 +123,28 @@ event for `tool_input.uri` or `tool_output.uri`. In this design, reading an experience through `read_experience` means the experience was injected into the prompt. +## Metadata and Status Contract + +`metadata` is optional on both tools for compatibility with older OpenViking +servers. When present, it contains only the allowlisted fields `status`, +`version`, `curated_at`, and `curated_from`; callers must not expect internal +provenance fields. `status` is normalized to lowercase. + +`search_experience` omits experiences whose status is `deprecated` or +`archived`. Results with `production`, `staging`, `draft`, an unknown status, or no status can +still be returned. Treat `draft` as provisional guidance and validate it against +the current task. `read_experience` revalidates the document metadata and rejects +a direct URI whose status is `deprecated` or `archived`; treat that tool error as +the experience being unavailable and do not inject it. + ## Recommended Flow 1. When a task begins, build a short query from the latest user instruction, current plan, active skill name, and important tool/environment context. 2. Call `search_experience` before final prompt assembly. -3. Review returned titles/snippets and select only experiences likely to affect - execution. +3. Review returned titles, snippets, and optional metadata. Select only + experiences likely to affect execution; prefer `production` entries, treat + `draft` as provisional, and skip `deprecated` or `archived` entries. 4. Call `read_experience` for selected experience URIs. 5. Inject the returned Markdown into the prompt under an explicit experience section. diff --git a/examples/zcode-memory-plugin/scripts/shared/recall-core.mjs b/examples/zcode-memory-plugin/scripts/shared/recall-core.mjs index a6e94eef5a..8c53dcea96 100644 --- a/examples/zcode-memory-plugin/scripts/shared/recall-core.mjs +++ b/examples/zcode-memory-plugin/scripts/shared/recall-core.mjs @@ -40,7 +40,8 @@ function scaleQuotas(limit, weights) { const order = Object.keys(weights); const quotas = Object.fromEntries(order.map((key) => [key, 0])); if (slots < order.length) { - for (const key of order) quotas[key] = 1; + const priority = [...order].sort((a, b) => weights[b] - weights[a]); + for (const key of priority.slice(0, slots)) quotas[key] = 1; return quotas; } @@ -61,10 +62,12 @@ function scaleQuotas(limit, weights) { } function legacyMemoryQuotas(limit) { - return { - ...scaleQuotas(limit, { events: 10, entities: 10, preferences: 3 }), - experiences: 0, - }; + return scaleQuotas(limit, { + events: 10, + entities: 10, + experiences: 3, + preferences: 3, + }); } function codingQuotas(limit) {