From 8d00fc70fdcf5f9fdf3447be4b9a7ee47882228c Mon Sep 17 00:00:00 2001 From: d-robotics Date: Sun, 25 Oct 2026 02:47:05 +0800 Subject: [PATCH 1/4] feat(session): durably hand off deferred steering exactly once --- README.md | 2 + docs/sdk-durable-steering.md | 11 + package-lock.json | 4 +- package.json | 2 +- src/core/agent/moss-agent-types.ts | 3 + src/core/agent/moss-agent.ts | 216 ++++++++---- src/core/index.ts | 2 +- src/core/session/index.ts | 2 +- src/core/session/jsonl-session-store.ts | 86 ++++- src/core/session/session-inbox-store.ts | 187 ++++++++++- src/core/session/session-inbox.ts | 22 +- src/core/session/session.ts | 75 ++++- test/deferred-steer-durable.spec.mjs | 420 ++++++++++++++++++++++++ test/sdk-contract.spec.mjs | 2 + 14 files changed, 933 insertions(+), 101 deletions(-) create mode 100644 docs/sdk-durable-steering.md create mode 100644 test/deferred-steer-durable.spec.mjs diff --git a/README.md b/README.md index 90518e7c5..6257aa0ae 100644 --- a/README.md +++ b/README.md @@ -401,6 +401,8 @@ npm run build && MOSS_REAL_TERMINALS=1 npm run test:filter -- --filter tui-real- ## 文档 +- [嵌入宿主的持久补充指令与会话消费回执](docs/sdk-durable-steering.md) + - [`AGENTS.md`](AGENTS.md) —— 架构、分层规则、子系统导航、工程约定（工作合同） - [`docs/release-policy.md`](docs/release-policy.md) —— 一个版本 / tag 声称了什么，又没声称什么 - [`docs/capability-layer.md`](docs/capability-layer.md) —— MCP / device / skill 能力层 diff --git a/docs/sdk-durable-steering.md b/docs/sdk-durable-steering.md new file mode 100644 index 000000000..33a2f9a1b --- /dev/null +++ b/docs/sdk-durable-steering.md @@ -0,0 +1,11 @@ +# Durable steering in an embedded host + +An accepted steer belongs to the session inbox until its canonical message has been adopted. An unconsumed entry remains pending when a run ends. A new Agent using the same session directory can deliver it on the next run. Delivery uses a stable inbox entry ID, not the message body; two intentionally identical instructions remain distinct deliveries. + +The built-in JSONL store implements `appendMessageOnce(sessionKey, deliveryId, message)`. Its message and delivery receipt share one locked, fsynced append. A retry after append succeeds but inbox acknowledgement fails returns `false` without appending another message. Replacing or compacting history preserves receipts. A duplicate receipt does not refresh the caller's observed history version or authorize an outdated replacement. + +The in-memory store provides receipt idempotence for its own lifetime. Hosts that initially use memory can implement optional `ensureDurableSession(sessionKey)`. Moss requests this transition only before adopting deferred entries. The host must transfer a coherent `loadDeliverySnapshot` containing both messages and delivery IDs into its durable store, under the same session lock as subsequent operations. `replaceMessages` accepts those IDs as an optional third argument. The durable store becomes the sole canonical history; an intentionally empty durable history is not a missing session and must not be filled from an older archive. + +After a producer crashes before it can defer its pending entries, the host can explicitly recover entries for proven terminated run IDs. Admission includes server-generated producer provenance. A different Agent instance, an empty local active-run map, another hostname or an inaccessible PID is not proof of termination. Studio's host uses same-host `ESRCH`; live or unknown producers remain pending. + +Custom stores remain source compatible: these methods are optional. An append-only custom store retains append-then-ack behavior and does **not** gain a crash-safe exactly-once guarantee. A host must fail closed when it cannot transfer delivery receipts coherently. Existing explicit inbox drain APIs still transfer ownership to their caller; they are not automatic durable recovery. diff --git a/package-lock.json b/package-lock.json index f9ba601d4..f609de73f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@rdk-moss/agent", - "version": "0.27.0", + "version": "0.28.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@rdk-moss/agent", - "version": "0.27.0", + "version": "0.28.0", "hasInstallScript": true, "license": "MIT", "dependencies": { diff --git a/package.json b/package.json index 41ff1fd54..17e73eaa1 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@rdk-moss/agent", - "version": "0.27.0", + "version": "0.28.0", "private": false, "type": "module", "license": "MIT", diff --git a/src/core/agent/moss-agent-types.ts b/src/core/agent/moss-agent-types.ts index e336f0fa9..cc4b2e9a1 100644 --- a/src/core/agent/moss-agent-types.ts +++ b/src/core/agent/moss-agent-types.ts @@ -163,6 +163,9 @@ export interface MossAgentConfig workspaceDir?: string; + /** Owner-scoped inbox directory; defaults to workspaceDir's runtime inbox. */ + sessionInboxDir?: string; + /** Shell-write confinement roots for exec tools (v0.9 W1). Undefined = unconstrained. */ execWriteRoots?: string[]; diff --git a/src/core/agent/moss-agent.ts b/src/core/agent/moss-agent.ts index 7aebdec21..a680fa0f6 100644 --- a/src/core/agent/moss-agent.ts +++ b/src/core/agent/moss-agent.ts @@ -1,5 +1,7 @@ import path from 'node:path'; +import os from 'node:os'; import type { LLMMessage } from '../llm/llm-provider.js'; +import type { SessionStore } from '../session/session.js'; import type { Tool, ToolContext, ToolResult } from '../tools/tool-types.js'; import { getRootLogger } from '../../logger.js'; import { buildExperienceBlock, experienceEnabled } from '../experience/experience-library.js'; @@ -85,7 +87,7 @@ import { type SessionInboxDelivery, } from '../session/session-inbox.js'; import { runSessionDrain, type SessionDrainResult } from '../session/session-drain.js'; -import { loadSessionInbox, saveSessionInbox } from '../session/session-inbox-store.js'; +import { loadSessionInbox, updateSessionInbox } from '../session/session-inbox-store.js'; import { getMossWorkspacePaths } from '../../utils/workspace-paths.js'; import { SessionEventLog, type SessionEvent } from '../session/session-event.js'; import { appendSessionEvent, loadSessionEventLog } from '../session/session-event-store.js'; @@ -145,13 +147,15 @@ export class MossAgent { private readonly remoteCompactProvider = createRemoteCompactProviderFromEnv(); private readonly toolHooks: ToolHookRegistry; private readonly inboxes = new Map(); + private readonly inboxProducer = { + host: os.hostname(), + pid: process.pid, + instanceId: crypto.randomUUID(), + }; private readonly activeRunIds = new Map>(); - /** - * Steers admitted after the loop's last poll. The host queue (`takeDeferredSteers`) - * owns them. A later run that the host did not drain delivers each one as a - * normal user message ahead of the new prompt — not as a mid-run steer. - */ - private readonly deferredSteers = new Map(); + /** Polling claims an entry only until its canonical append acknowledges it. */ + private readonly claimedSteers = new Map>(); + private readonly steeringDeliveries = new WeakMap(); private userQuestionAsker?: NonNullable; /** Per-agent run epochs stop parallel instances overwriting shared-session streams. */ private readonly runEpochStore = new Map(); @@ -429,6 +433,9 @@ export class MossAgent { } private inboxFilePath(sessionKey: string): string | undefined { + if (this.config.sessionInboxDir) { + return path.join(this.config.sessionInboxDir, `${encodeURIComponent(sessionKey)}.json`); + } const workspaceDir = this.config?.workspaceDir; if (!workspaceDir) return undefined; return path.join( @@ -439,24 +446,36 @@ export class MossAgent { } private inboxFor(sessionKey: string): SessionInbox { + const file = this.inboxFilePath(sessionKey); + if (file) { + return this.rememberInbox(sessionKey, loadSessionInbox(file)); + } let inbox = this.inboxes.get(sessionKey); if (!inbox) { - const file = this.inboxFilePath(sessionKey); - inbox = file ? loadSessionInbox(file) : new SessionInbox(); + inbox = new SessionInbox(); this.inboxes.set(sessionKey, inbox); } return inbox; } - private persistInbox(sessionKey: string): void { + private rememberInbox(sessionKey: string, inbox: SessionInbox): SessionInbox { + const cached = this.inboxes.get(sessionKey); + if (cached) { + cached.refreshFrom(inbox); + return cached; + } + this.inboxes.set(sessionKey, inbox); + return inbox; + } + + private mutateInbox(sessionKey: string, mutate: (inbox: SessionInbox) => T): T { const file = this.inboxFilePath(sessionKey); - const inbox = this.inboxes.get(sessionKey); - if (!file || !inbox) return; - try { - saveSessionInbox(file, inbox); - } catch (err) { - log.warn('persist_inbox_failed', { error: errorMessage(err), sessionKey }); + if (file) { + const { inbox, result } = updateSessionInbox(file, mutate); + this.rememberInbox(sessionKey, inbox); + return result; } + return mutate(this.inboxFor(sessionKey)); } admit( @@ -464,13 +483,13 @@ export class MossAgent { prompt: string, options?: { delivery?: SessionInboxDelivery; id?: string } ): SessionInboxEntry { - const entry = this.inboxFor(sessionKey).admit({ - prompt, - delivery: options?.delivery, - id: options?.id, - }); - this.persistInbox(sessionKey); - return entry; + return this.mutateInbox(sessionKey, (inbox) => + inbox.admit({ + prompt, + delivery: options?.delivery, + id: options?.id, + }) + ); } inboxPending(sessionKey: string): readonly SessionInboxEntry[] { @@ -481,37 +500,66 @@ export class MossAgent { const activeRuns = this.activeRunIds.get(sessionKey); if (!activeRuns || activeRuns.size !== 1) return null; const runId = activeRuns.values().next().value as string; - const entry = this.inboxFor(sessionKey).admit({ - prompt, - delivery: 'steer', - metadata: { runId }, - }); - this.persistInbox(sessionKey); - return entry; + return this.mutateInbox(sessionKey, (inbox) => + inbox.admit({ + prompt, + delivery: 'steer', + metadata: { runId, producer: this.inboxProducer }, + }) + ); } /** - * Prompts that missed the run they were aimed at. Clears the stash. - * The TUI queues these. A host that never calls this still gets them once, - * as user messages ahead of the next prompt. + * Transfer deferred prompts to the caller, which now owns their delivery. + * Hosts wanting crash-safe automatic delivery should leave them in the inbox. */ takeDeferredSteers(sessionKey: string): string[] { - const prompts = this.deferredSteers.get(sessionKey) ?? []; - if (prompts.length > 0) this.deferredSteers.delete(sessionKey); - return prompts.slice(); + if ( + !this.inboxFor(sessionKey) + .promotableSteers() + .some((entry) => entry.deferred) + ) + return []; + return this.mutateInbox(sessionKey, (inbox) => + inbox + .promotableSteers() + .filter((entry) => entry.deferred) + .map((entry) => { + inbox.promote(entry.id); + return entry.prompt; + }) + ); + } + + /** Host-proven terminated runs only; absence from this Agent is not proof. */ + recoverDeferredSteers(sessionKey: string, terminatedRunIds: readonly string[]): void { + const terminated = new Set(terminatedRunIds); + if (terminated.size === 0) return; + this.mutateInbox(sessionKey, (inbox) => { + for (const entry of inbox.promotableSteers()) { + if (typeof entry.metadata?.runId === 'string' && terminated.has(entry.metadata.runId)) { + inbox.defer(entry.id); + } + } + }); } private takeSteeringMessages(sessionKey: string, runId: string): InternalMessage[] { const inbox = this.inboxFor(sessionKey); + const claimed = this.claimedSteers.get(runId) ?? new Set(); const entries = inbox.promotableSteers().filter((entry) => { const targetRunId = entry.metadata?.runId; - return targetRunId === undefined || targetRunId === runId; + return ( + !entry.deferred && + !claimed.has(entry.id) && + (targetRunId === undefined || targetRunId === runId) + ); }); if (entries.length === 0) return []; const timestamp = Date.now(); const messages = entries.map((entry) => { - inbox.promote(entry.id); - return { + claimed.add(entry.id); + const message: InternalMessage = { role: 'user' as const, content: [ { @@ -521,27 +569,27 @@ export class MossAgent { ], timestamp, }; + this.steeringDeliveries.set(message, entry.id); + return message; }); - this.persistInbox(sessionKey); + this.claimedSteers.set(runId, claimed); return messages; } /** Unconsumed steers for a finished run are deferred, not discarded. */ private deferUnconsumedSteers(sessionKey: string, runId: string): void { - const inbox = this.inboxes.get(sessionKey); - if (!inbox) return; - const prompts: string[] = []; - let changed = false; - for (const entry of inbox.promotableSteers()) { - if (entry.metadata?.runId !== runId) continue; - prompts.push(entry.prompt); - inbox.promote(entry.id); - changed = true; - } - if (changed) this.persistInbox(sessionKey); - if (prompts.length === 0) return; - const existing = this.deferredSteers.get(sessionKey) ?? []; - this.deferredSteers.set(sessionKey, existing.concat(prompts)); + this.claimedSteers.delete(runId); + if ( + !this.inboxFor(sessionKey) + .promotableSteers() + .some((entry) => entry.metadata?.runId === runId) + ) + return; + this.mutateInbox(sessionKey, (inbox) => { + for (const entry of inbox.promotableSteers()) { + if (entry.metadata?.runId === runId) inbox.defer(entry.id); + } + }); } private noteRunStarted(sessionKey: string, runId: string): void { @@ -550,6 +598,21 @@ export class MossAgent { this.activeRunIds.set(sessionKey, activeRuns); } + private async appendInboxDelivery( + store: SessionStore, + sessionKey: string, + deliveryId: string, + message: LLMMessage + ): Promise { + // Built-in stores commit message and receipt together. Legacy custom + // stores retain append-then-ack compatibility, without crash-safe dedupe. + const appended = store.appendMessageOnce + ? await store.appendMessageOnce(sessionKey, deliveryId, message) + : (await store.appendMessage(sessionKey, message), true); + this.mutateInbox(sessionKey, (inbox) => inbox.promote(deliveryId)); + return appended; + } + private noteRunFinished(sessionKey: string, runId: string): void { const activeRuns = this.activeRunIds.get(sessionKey); if (!activeRuns) return; @@ -565,9 +628,13 @@ export class MossAgent { const drain = await runSessionDrain({ inbox: this.inboxFor(sessionKey), runTurn: async (promoted) => { + // Like takeDeferredSteers, explicit drain transfers delivery ownership + // to the caller of chat rather than the automatic steering path. + this.mutateInbox(sessionKey, (inbox) => { + for (const entry of promoted) inbox.promote(entry.id); + }); for (const entry of promoted) chats.push(await this.chat(sessionKey, entry.prompt, options)); - this.persistInbox(sessionKey); return { continue: false }; }, }); @@ -905,25 +972,41 @@ ${result.stderr ?? ''}`.trim(); } } + const deferredEntries = this.inboxFor(sessionKey) + .promotableSteers() + .filter((entry) => entry.deferred); + if (deferredEntries.length > 0 && store.ensureDurableSession) { + await abortable(store.ensureDurableSession(sessionKey), abortSignal); + } const loadedMessages = (await abortable( store.loadMessages(sessionKey), abortSignal )) as unknown as InternalMessage[]; const messages = fromSessionMessages(loadedMessages); const isFirstUserTurn = !messages.some((message) => message.role === 'user'); - // Deferred steers belong to the previous run. Deliver them as ordinary - // user messages ahead of this prompt. takeDeferredSteers clears the stash, - // so a host that already queued them does not get a second copy. - for (const prompt of this.takeDeferredSteers(sessionKey)) { - const text = prompt.trim(); - if (!text) continue; + // Keep the durable pending record until canonical append + receipt commits. + // A retry after append but before ack sees the receipt and reloads history. + for (const entry of deferredEntries) { const prior: InternalMessage = { role: 'user', - content: text, + content: entry.prompt, timestamp: Date.now(), }; - messages.push(prior); - await store.appendMessage(sessionKey, prior as unknown as LLMMessage); + const appended = await this.appendInboxDelivery( + store, + sessionKey, + entry.id, + prior as unknown as LLMMessage + ); + if (appended) messages.push(prior); + else + messages.splice( + 0, + messages.length, + ...fromSessionMessages( + (await store.loadMessages(sessionKey)) as unknown as InternalMessage[] + ) + ); } const userMsg: InternalMessage = { role: 'user', @@ -1452,7 +1535,10 @@ ${result.stderr ?? ''}`.trim(); contextTokens, steeringEngine: this.steeringEngine ?? undefined, appendMessage: async (key, msg) => { - await store.appendMessage(key, msg as unknown as LLMMessage); + const deliveryId = this.steeringDeliveries.get(msg as InternalMessage); + if (deliveryId) { + await this.appendInboxDelivery(store, key, deliveryId, msg as unknown as LLMMessage); + } else await store.appendMessage(key, msg as unknown as LLMMessage); }, replaceMessages: async (key, nextMessages) => { await store.replaceMessages(key, nextMessages as unknown as LLMMessage[]); diff --git a/src/core/index.ts b/src/core/index.ts index ebacb211c..8fed553cd 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -105,7 +105,7 @@ export { export type { SteeringRule, SteeringContext, SteeringResult } from './loop/index.js'; export { InMemorySessionStore } from './session/index.js'; -export type { SessionStore, SessionMeta } from './session/index.js'; +export type { SessionStore, SessionMeta, SessionDeliverySnapshot } from './session/index.js'; export { JsonlSessionStore } from './session/index.js'; export type { JsonlSessionStoreConfig } from './session/index.js'; export { diff --git a/src/core/session/index.ts b/src/core/session/index.ts index 19ade947f..7ee363b5f 100644 --- a/src/core/session/index.ts +++ b/src/core/session/index.ts @@ -1,5 +1,5 @@ export { InMemorySessionStore } from './session.js'; -export type { SessionStore, SessionMeta } from './session.js'; +export type { SessionStore, SessionMeta, SessionDeliverySnapshot } from './session.js'; export { JsonlSessionStore } from './jsonl-session-store.js'; export type { JsonlSessionStoreConfig } from './jsonl-session-store.js'; export { diff --git a/src/core/session/jsonl-session-store.ts b/src/core/session/jsonl-session-store.ts index eadd1676f..27bc3a8f3 100644 --- a/src/core/session/jsonl-session-store.ts +++ b/src/core/session/jsonl-session-store.ts @@ -2,7 +2,7 @@ import crypto from 'node:crypto'; import * as fsp from 'node:fs/promises'; import * as path from 'node:path'; import type { LLMMessage } from '../llm/llm-provider.js'; -import type { SessionStore, SessionMeta } from './session.js'; +import type { SessionStore, SessionMeta, SessionDeliverySnapshot } from './session.js'; import { WriteChain } from '../../utils/write-chain.js'; import { ensureMossRuntimeGitignore } from '../../utils/workspace-paths.js'; import { ErrorCode, MossError } from '../../errors.js'; @@ -17,8 +17,8 @@ export interface JsonlSessionStoreConfig { } type JsonlSessionEntry = - | { type: 'message'; message: LLMMessage; ts?: number } - | { type: 'state_replace'; messages: LLMMessage[]; ts?: number }; + | { type: 'message'; message: LLMMessage; deliveryId?: string; ts?: number } + | { type: 'state_replace'; messages: LLMMessage[]; deliveryIds?: string[]; ts?: number }; type DurableFileHandle = { writeFile: (data: string, encoding: BufferEncoding) => Promise; @@ -343,36 +343,46 @@ export class JsonlSessionStore implements SessionStore { private replayMessagesFromContent(raw: string): { messages: LLMMessage[]; + deliveryIds: Set; malformedCount: number; } { const lines = raw.split('\n').filter((l) => l.trim()); const messages: LLMMessage[] = []; + const deliveryIds = new Set(); let malformedCount = 0; for (const line of lines) { try { const entry = JSON.parse(line) as JsonlSessionEntry; if (entry.type === 'message' && entry.message) { messages.push(entry.message); + if (typeof entry.deliveryId === 'string') deliveryIds.add(entry.deliveryId); } else if (entry.type === 'state_replace' && Array.isArray(entry.messages)) { messages.splice(0, messages.length, ...entry.messages); + for (const id of entry.deliveryIds ?? []) { + if (typeof id === 'string') deliveryIds.add(id); + } } } catch { malformedCount++; } } - return { messages, malformedCount }; + return { messages, deliveryIds, malformedCount }; } async loadMessages(sessionKey: string): Promise { + return (await this.loadDeliverySnapshot(sessionKey)).messages; + } + + async loadDeliverySnapshot(sessionKey: string): Promise { const filePath = this.sessionPath(sessionKey); - let result: LLMMessage[] = []; + let result: SessionDeliverySnapshot = { messages: [], deliveryIds: [] }; await this.enqueueWrite(filePath, async () => { result = await this.withFileLock(filePath, async () => { const raw = await this.readRaw(filePath); this.observedContentVersions.set(filePath, this.contentVersion(raw)); this.replacementConflicts.delete(filePath); - if (!raw) return []; - const { messages, malformedCount } = this.replayMessagesFromContent(raw); + if (!raw) return { messages: [], deliveryIds: [] }; + const { messages, deliveryIds, malformedCount } = this.replayMessagesFromContent(raw); if (malformedCount > 0) { const pct = messages.length > 0 @@ -385,15 +395,33 @@ export class JsonlSessionStore implements SessionStore { `If the damage is severe, start a fresh session with \`moss\`.` ); } - return messages; + return { messages, deliveryIds: [...deliveryIds] }; }); }); return result; } async appendMessage(sessionKey: string, message: LLMMessage): Promise { + await this.appendSessionMessage(sessionKey, message); + } + + async appendMessageOnce( + sessionKey: string, + deliveryId: string, + message: LLMMessage + ): Promise { + if (!deliveryId) throw new Error('A non-empty deliveryId is required'); + return this.appendSessionMessage(sessionKey, message, deliveryId); + } + + private async appendSessionMessage( + sessionKey: string, + message: LLMMessage, + deliveryId?: string + ): Promise { const filePath = this.sessionPath(sessionKey); - const entry = redactJsonLine({ type: 'message', message, ts: Date.now() }); + const entry = redactJsonLine({ type: 'message', message, deliveryId, ts: Date.now() }); + let appended = false; const mutate = async () => { let isNewSession = false; await this.enqueueWrite(filePath, async () => { @@ -402,6 +430,16 @@ export class JsonlSessionStore implements SessionStore { this.noteRuntimeGitignore(); isNewSession = this.maxSessions > 0 && !(await this.fileExists(filePath)); const currentRaw = await this.readRaw(filePath); + if ( + deliveryId && + this.replayMessagesFromContent(currentRaw).deliveryIds.has(deliveryId) + ) { + // Receipt lookup precedes the optimistic version guard: a retry of + // a committed delivery is safe even from a stale store instance. + // It does not refresh that instance's history view: only an actual + // load may authorize a later replacement of concurrent messages. + return; + } const observedVersion = this.observedContentVersions.get(filePath); const currentVersion = this.contentVersion(currentRaw); if (observedVersion !== undefined && observedVersion !== currentVersion) { @@ -414,6 +452,7 @@ export class JsonlSessionStore implements SessionStore { }); } await this.appendLineDurably(filePath, entry + '\n'); + appended = true; const raw = await this.readRaw(filePath); this.observedContentVersions.set(filePath, this.contentVersion(raw)); }); @@ -422,15 +461,18 @@ export class JsonlSessionStore implements SessionStore { }; if (this.maxSessions > 0) await this.withDirectoryMutationLock(mutate); else await mutate(); + return appended; } - async replaceMessages(sessionKey: string, messages: LLMMessage[]): Promise { + async replaceMessages( + sessionKey: string, + messages: LLMMessage[], + options?: { deliveryIds?: readonly string[] } + ): Promise { + if (options?.deliveryIds?.some((id) => typeof id !== 'string' || !id)) { + throw new Error('Delivery snapshot contains an invalid deliveryId'); + } const filePath = this.sessionPath(sessionKey); - const entry = redactJsonLine({ - type: 'state_replace', - messages, - ts: Date.now(), - }); const mutate = async () => { let isNewSession = false; await this.enqueueWrite(filePath, async () => { @@ -457,7 +499,19 @@ export class JsonlSessionStore implements SessionStore { // Prunes dead lines (prior snapshots + superseded messages) that would // otherwise accumulate: replay ignores everything before the latest // state_replace, so appended-but-dead bytes grew ~ history × replaces. - const nextRaw = entry + '\n'; + const deliveryIds = [ + ...new Set([ + ...this.replayMessagesFromContent(currentRaw).deliveryIds, + ...(options?.deliveryIds ?? []), + ]), + ]; + const nextRaw = + redactJsonLine({ + type: 'state_replace', + messages, + deliveryIds, + ts: Date.now(), + }) + '\n'; await this.rewriteFileDurably(filePath, nextRaw); this.observedContentVersions.set(filePath, this.contentVersion(nextRaw)); }); diff --git a/src/core/session/session-inbox-store.ts b/src/core/session/session-inbox-store.ts index 9d92e9125..a0ffb7f9c 100644 --- a/src/core/session/session-inbox-store.ts +++ b/src/core/session/session-inbox-store.ts @@ -1,25 +1,192 @@ +import crypto from 'node:crypto'; import fs from 'node:fs'; import path from 'node:path'; import { SessionInbox, type SessionInboxEntry } from './session-inbox.js'; +type InboxLockOwner = { pid: number; token: string }; + +function syncDirectory(directory: string): void { + if (process.platform === 'win32') return; + const fd = fs.openSync(directory, 'r'); + try { + fs.fsyncSync(fd); + } finally { + fs.closeSync(fd); + } +} + +function ensureDirectory(directory: string): void { + if (fs.existsSync(directory)) return; + const parent = path.dirname(directory); + ensureDirectory(parent); + try { + fs.mkdirSync(directory); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error; + } + syncDirectory(directory); + syncDirectory(parent); +} + +function readOwner(file: string): InboxLockOwner | undefined { + try { + const owner = JSON.parse(fs.readFileSync(file, 'utf8')) as Partial; + return Number.isSafeInteger(owner.pid) && + (owner.pid ?? 0) > 0 && + typeof owner.token === 'string' && + owner.token.length >= 16 + ? (owner as InboxLockOwner) + : undefined; + } catch { + return undefined; + } +} + +function isDead(pid: number): boolean { + try { + process.kill(pid, 0); + return false; + } catch (error) { + return (error as NodeJS.ErrnoException).code === 'ESRCH'; + } +} + +/** The synchronous admission API cannot wait on an async lock in this process. */ +function withInboxLock(file: string, operation: () => T): T { + ensureDirectory(path.dirname(file)); + const lock = `${file}.lock`; + const owner: InboxLockOwner = { pid: process.pid, token: crypto.randomUUID() }; + const candidate = `${lock}.candidate-${owner.token}`; + const publish = (target: string): boolean => { + let published = false; + try { + fs.linkSync(candidate, target); + published = true; + syncDirectory(path.dirname(file)); + return true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'EEXIST') return false; + if (published && readOwner(target)?.token === owner.token) { + // A failed directory fsync must not strand a live owner that prevents + // the same process from retrying an admission it was told had failed. + try { + fs.unlinkSync(target); + syncDirectory(path.dirname(file)); + } catch { + /* preserve the original durability error */ + } + } + throw error; + } + }; + const release = (target: string): void => { + if (readOwner(target)?.token !== owner.token) + throw new Error(`Session inbox lock owner changed: ${target}`); + fs.unlinkSync(target); + syncDirectory(path.dirname(file)); + }; + let acquired = false; + try { + const fd = fs.openSync(candidate, 'wx', 0o600); + try { + fs.writeFileSync(fd, JSON.stringify(owner)); + fs.fsyncSync(fd); + } finally { + fs.closeSync(fd); + } + acquired = publish(lock); + if (!acquired) { + // One recovery gate serializes dead-owner replacement. A stranded or + // malformed recovery owner fails closed; elapsed time never proves death. + const observed = readOwner(lock); + const recovery = `${lock}.recovery`; + if (!observed || !isDead(observed.pid) || !publish(recovery)) { + throw new Error(`Session inbox is locked or has an unknown owner: ${file}`); + } + try { + const current = readOwner(lock); + if (!current || current.token !== observed.token || !isDead(current.pid)) { + throw new Error(`Session inbox lock owner changed during recovery: ${file}`); + } + fs.unlinkSync(lock); + syncDirectory(path.dirname(file)); + acquired = publish(lock); + if (!acquired) throw new Error(`Session inbox lock recovery conflicted: ${file}`); + } finally { + release(recovery); + } + } + return operation(); + } finally { + try { + if (acquired) release(lock); + } finally { + fs.rmSync(candidate, { force: true }); + } + } +} + export function loadSessionInbox(filePath: string): SessionInbox { let raw: string; try { raw = fs.readFileSync(filePath, 'utf8'); - } catch { - return new SessionInbox(); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return new SessionInbox(); + throw error; + } + const entries = JSON.parse(raw) as unknown; + if ( + !Array.isArray(entries) || + !entries.every( + (entry: Partial) => + entry && + typeof entry.id === 'string' && + typeof entry.prompt === 'string' && + (entry.delivery === 'steer' || entry.delivery === 'queue') && + Number.isSafeInteger(entry.admittedSeq) && + (entry.admittedSeq ?? 0) > 0 && + (entry.promotedSeq === undefined || + (Number.isSafeInteger(entry.promotedSeq) && entry.promotedSeq > 0)) && + (entry.deferred === undefined || typeof entry.deferred === 'boolean') + ) + ) + throw new Error(`Invalid session inbox: ${filePath}`); + if (new Set(entries.map((entry: SessionInboxEntry) => entry.id)).size !== entries.length) { + throw new Error(`Duplicate entries in session inbox: ${filePath}`); } + return SessionInbox.fromEntries(entries as SessionInboxEntry[]); +} + +function writeSessionInbox(filePath: string, inbox: SessionInbox): void { + const tmp = `${filePath}.tmp-${crypto.randomUUID()}`; + let fd: number | undefined; try { - const entries = JSON.parse(raw) as SessionInboxEntry[]; - return Array.isArray(entries) ? SessionInbox.fromEntries(entries) : new SessionInbox(); - } catch { - return new SessionInbox(); + fd = fs.openSync(tmp, 'wx', 0o600); + fs.writeFileSync(fd, JSON.stringify(inbox.toEntries(), null, 2)); + fs.fsyncSync(fd); + fs.closeSync(fd); + fd = undefined; + fs.renameSync(tmp, filePath); + syncDirectory(path.dirname(filePath)); + } finally { + if (fd !== undefined) fs.closeSync(fd); + fs.rmSync(tmp, { force: true }); } } export function saveSessionInbox(filePath: string, inbox: SessionInbox): void { - fs.mkdirSync(path.dirname(filePath), { recursive: true }); - const tmp = `${filePath}.tmp`; - fs.writeFileSync(tmp, JSON.stringify(inbox.toEntries(), null, 2)); - fs.renameSync(tmp, filePath); + withInboxLock(filePath, () => writeSessionInbox(filePath, inbox)); +} + +/** Reload and mutate under one lock so separate Agent instances cannot lose entries. */ +export function updateSessionInbox( + filePath: string, + mutate: (inbox: SessionInbox) => T +): { inbox: SessionInbox; result: T } { + return withInboxLock(filePath, () => { + const inbox = loadSessionInbox(filePath); + const result = mutate(inbox); + writeSessionInbox(filePath, inbox); + return { inbox, result }; + }); } diff --git a/src/core/session/session-inbox.ts b/src/core/session/session-inbox.ts index d42fa5883..2eccb1d07 100644 --- a/src/core/session/session-inbox.ts +++ b/src/core/session/session-inbox.ts @@ -11,6 +11,9 @@ export interface SessionInboxEntry { promotedSeq?: number; + /** The target run ended before canonical adoption. Still pending until ack. */ + deferred?: boolean; + readonly metadata?: Readonly>; } @@ -31,8 +34,6 @@ export class SessionInboxConflictError extends Error { } } -let autoId = 0; - export class SessionInbox { private readonly entries: SessionInboxEntry[] = []; private admittedCounter = 0; @@ -41,7 +42,7 @@ export class SessionInbox { static readonly MAX_ENTRIES = 500; admit(input: SessionInboxAdmitInput): SessionInboxEntry { - const id = input.id ?? `inbox_${Date.now().toString(36)}_${(autoId++).toString(36)}`; + const id = input.id ?? `inbox_${randomUUID()}`; if (this.entries.some((e) => e.id === id)) throw new SessionInboxConflictError(id); const entry: SessionInboxEntry = { id, @@ -78,6 +79,13 @@ export class SessionInbox { return this.entries.some((e) => e.promotedSeq === undefined); } + defer(id: string): SessionInboxEntry { + const entry = this.entries.find((e) => e.id === id); + if (!entry) throw new Error(`Session inbox has no entry "${id}"`); + if (entry.promotedSeq === undefined) entry.deferred = true; + return entry; + } + private pruneIfNeeded(): void { while (this.entries.length > SessionInbox.MAX_ENTRIES) { const idx = this.entries.findIndex((e) => e.promotedSeq !== undefined); @@ -90,6 +98,13 @@ export class SessionInbox { return this.entries.map((e) => ({ ...e })); } + /** Refresh persisted state without invalidating an active drain's inbox view. */ + refreshFrom(inbox: SessionInbox): void { + this.entries.splice(0, this.entries.length, ...inbox.toEntries()); + this.admittedCounter = inbox.admittedCounter; + this.promotedCounter = inbox.promotedCounter; + } + static fromEntries(entries: readonly SessionInboxEntry[]): SessionInbox { const inbox = new SessionInbox(); for (const e of entries) inbox.entries.push({ ...e }); @@ -98,3 +113,4 @@ export class SessionInbox { return inbox; } } +import { randomUUID } from 'node:crypto'; diff --git a/src/core/session/session.ts b/src/core/session/session.ts index 66a455509..d04183d99 100644 --- a/src/core/session/session.ts +++ b/src/core/session/session.ts @@ -8,12 +8,42 @@ export interface SessionMeta { messageCount: number; } +export interface SessionDeliverySnapshot { + messages: LLMMessage[]; + deliveryIds: string[]; +} + export interface SessionStore { loadMessages(sessionKey: string): Promise; + /** One coherent history and receipt image, for a canonical store handoff. */ + loadDeliverySnapshot?(sessionKey: string): Promise; + appendMessage(sessionKey: string, message: LLMMessage): Promise; - replaceMessages(sessionKey: string, messages: LLMMessage[]): Promise; + /** + * Atomically adopt a delivery and its message. Returns false for an already + * adopted ID, including after replacement/compaction. The caller must reuse + * an ID only for the same logical message. Durability follows the store: + * an in-memory implementation only deduplicates for its own lifetime. + * Custom stores may omit this method; appendMessage alone cannot provide a + * crash-safe acknowledgement boundary. + */ + appendMessageOnce?(sessionKey: string, deliveryId: string, message: LLMMessage): Promise; + + /** + * Optional one-way adoption into durable canonical storage, invoked before + * automatic deferred delivery. Capture the current history atomically, then + * route all later reads and mutations to that same durable history. Do not + * restore messages from a separate delivery journal after compaction. + */ + ensureDurableSession?(sessionKey: string): Promise; + + replaceMessages( + sessionKey: string, + messages: LLMMessage[], + options?: { deliveryIds?: readonly string[] } + ): Promise; listSessions(): Promise; @@ -24,12 +54,25 @@ export interface SessionStore { export class InMemorySessionStore implements SessionStore { private sessions = new Map(); + private readonly deliveries = new Map>(); async loadMessages(sessionKey: string): Promise { return [...(this.sessions.get(sessionKey)?.messages ?? [])]; } + async loadDeliverySnapshot(sessionKey: string): Promise { + // Both reads happen synchronously, before yielding to another mutation. + return { + messages: [...(this.sessions.get(sessionKey)?.messages ?? [])], + deliveryIds: [...(this.deliveries.get(sessionKey) ?? [])], + }; + } + async appendMessage(sessionKey: string, message: LLMMessage): Promise { + this.appendToSession(sessionKey, message); + } + + private appendToSession(sessionKey: string, message: LLMMessage): void { let session = this.sessions.get(sessionKey); if (!session) { session = { @@ -43,7 +86,28 @@ export class InMemorySessionStore implements SessionStore { session.meta.messageCount = session.messages.length; } - async replaceMessages(sessionKey: string, messages: LLMMessage[]): Promise { + async appendMessageOnce( + sessionKey: string, + deliveryId: string, + message: LLMMessage + ): Promise { + if (!deliveryId) throw new Error('A non-empty deliveryId is required'); + const deliveries = this.deliveries.get(sessionKey) ?? new Set(); + if (deliveries.has(deliveryId)) return false; + this.appendToSession(sessionKey, message); + deliveries.add(deliveryId); + this.deliveries.set(sessionKey, deliveries); + return true; + } + + async replaceMessages( + sessionKey: string, + messages: LLMMessage[], + options?: { deliveryIds?: readonly string[] } + ): Promise { + if (options?.deliveryIds?.some((id) => typeof id !== 'string' || !id)) { + throw new Error('Delivery snapshot contains an invalid deliveryId'); + } let session = this.sessions.get(sessionKey); if (!session) { session = { @@ -55,6 +119,12 @@ export class InMemorySessionStore implements SessionStore { session.messages = [...messages]; session.meta.updatedAt = Date.now(); session.meta.messageCount = messages.length; + if (options?.deliveryIds) { + this.deliveries.set( + sessionKey, + new Set([...(this.deliveries.get(sessionKey) ?? []), ...options.deliveryIds]) + ); + } } async listSessions(): Promise { @@ -63,6 +133,7 @@ export class InMemorySessionStore implements SessionStore { async deleteSession(sessionKey: string): Promise { this.sessions.delete(sessionKey); + this.deliveries.delete(sessionKey); } async exists(sessionKey: string): Promise { diff --git a/test/deferred-steer-durable.spec.mjs b/test/deferred-steer-durable.spec.mjs new file mode 100644 index 000000000..21eb7810d --- /dev/null +++ b/test/deferred-steer-durable.spec.mjs @@ -0,0 +1,420 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import * as fs from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; + +const source = process.env.MOSS_TEST_SOURCE === '1'; +const modulePath = (name) => `../${source ? 'src' : 'dist'}/${name}.${source ? 'ts' : 'js'}`; +const { MossAgent } = await import(modulePath('core/agent/moss-agent')); +const { JsonlSessionStore } = await import(modulePath('core/session/jsonl-session-store')); +const { InMemorySessionStore } = await import(modulePath('core/session/session')); +const { runProcess } = await import(modulePath('utils/run-process')); +const { loadSessionInbox, updateSessionInbox } = await import( + modulePath('core/session/session-inbox-store') +); +const fsSync = (await import('node:fs')).default; + +const user = (content) => ({ role: 'user', content }); +const users = (messages) => + messages + .filter((message) => message.role === 'user') + .map((message) => + typeof message.content === 'string' + ? message.content + : message.content + .filter((block) => block.type === 'text') + .map((block) => block.text) + .join('\n') + ); + +async function temporary(t) { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'moss-durable-steer-')); + t.after(() => fs.rm(dir, { recursive: true, force: true })); + return dir; +} + +function provider(blocked = false) { + let entered; + const ready = new Promise((resolve) => { + entered = resolve; + }); + const requests = []; + return { + ready, + requests, + id: 'durable-steer-test', + displayName: 'durable-steer-test', + async stream(options, onEvent) { + requests.push(options.messages); + entered(); + if (blocked) + await new Promise((_resolve, reject) => { + const fail = () => reject(new Error('aborted')); + if (options.abortSignal?.aborted) fail(); + else options.abortSignal?.addEventListener('abort', fail, { once: true }); + }); + onEvent({ type: 'message_start' }); + onEvent({ type: 'content_block_start' }); + onEvent({ type: 'content_block_delta', text: 'ok' }); + onEvent({ type: 'content_block_stop' }); + onEvent({ type: 'message_delta', stopReason: 'end_turn' }); + onEvent({ type: 'message_stop' }); + return { stopReason: 'end_turn', content: [{ type: 'text', text: 'ok' }] }; + }, + }; +} + +function agent( + t, + workspaceDir, + llmProvider, + sessionStore = new JsonlSessionStore({ dir: path.join(workspaceDir, 'sessions') }) +) { + const value = new MossAgent({ + workspaceDir, + sessionStore, + llmProvider, + baseSystemPrompt: 'test', + domainPrompt: false, + includeAgentBehaviorPrompt: false, + includeLanguagePolicyPrompt: false, + enableSteering: false, + enableFollowUpGuard: false, + maxAgentTurns: 2, + }); + t.after(() => value.close()); + return value; +} + +async function leaveDeferred(t, dir, key, text) { + const llm = provider(true); + const value = agent(t, dir, llm); + const controller = new AbortController(); + const run = value.chat(key, 'first', { abortSignal: controller.signal }); + await llm.ready; + const admitted = value.steer(key, text); + assert.ok(admitted); + controller.abort(); + await run.catch(() => {}); + await value.close(); + return admitted; +} + +for (const kind of ['memory', 'jsonl']) + test(`${kind} append-once is scoped by delivery ID and survives replacement`, async (t) => { + const dir = await temporary(t); + const store = kind === 'memory' ? new InMemorySessionStore() : new JsonlSessionStore({ dir }); + assert.equal(typeof store.appendMessageOnce, 'function'); + assert.equal( + await store.appendMessageOnce('owner-a:session', 'delivery-1', user('same body')), + true + ); + assert.equal( + await store.appendMessageOnce('owner-a:session', 'delivery-1', user('same body')), + false + ); + assert.equal( + await store.appendMessageOnce('owner-a:session', 'delivery-2', user('same body')), + true + ); + assert.deepEqual(users(await store.loadMessages('owner-a:session')), [ + 'same body', + 'same body', + ]); + await store.replaceMessages('owner-a:session', [user('compacted')]); + const restarted = kind === 'memory' ? store : new JsonlSessionStore({ dir }); + assert.equal( + await restarted.appendMessageOnce('owner-a:session', 'delivery-1', user('same body')), + false + ); + assert.deepEqual(users(await restarted.loadMessages('owner-a:session')), ['compacted']); + assert.equal( + await restarted.appendMessageOnce('owner-b:session', 'delivery-1', user('same body')), + true + ); + await restarted.deleteSession('owner-a:session'); + assert.equal( + await restarted.appendMessageOnce('owner-a:session', 'delivery-1', user('new session')), + true + ); + }); + +test('independent Agents retain deferred steer without owner/session leakage', async (t) => { + const dir = await temporary(t); + const key = 'owner-a:session'; + await leaveDeferred(t, dir, key, 'durable constraint'); + for (const unrelated of ['owner-b:session', 'owner-a:another']) { + const llm = provider(); + await agent(t, dir, llm).chat(unrelated, 'next'); + assert.equal( + users(llm.requests.at(-1)).filter((text) => text === 'durable constraint').length, + 0 + ); + } + const llm = provider(); + const next = agent(t, dir, llm); + await next.chat(key, 'next'); + const texts = users(llm.requests.at(-1)); + assert.equal(texts.filter((text) => text === 'durable constraint').length, 1); + assert.ok(texts.indexOf('durable constraint') < texts.findIndex((text) => text.includes('next'))); + assert.equal(next.inboxPending(key).length, 0); + await next.chat(key, 'again'); + assert.equal( + users(llm.requests.at(-1)).filter((text) => text === 'durable constraint').length, + 1 + ); +}); + +test('append committed before lost acknowledgement is retried without duplication', async (t) => { + const dir = await temporary(t); + const key = 'owner-a:session'; + await leaveDeferred(t, dir, key, 'ack crash constraint'); + const durable = new JsonlSessionStore({ dir: path.join(dir, 'sessions') }); + assert.equal(typeof durable.appendMessageOnce, 'function'); + const append = durable.appendMessageOnce.bind(durable); + durable.appendMessageOnce = async (...args) => { + await append(...args); + throw new Error('simulated stop after durable append before inbox acknowledgement'); + }; + await assert.rejects( + agent(t, dir, provider(), durable).chat(key, 'interrupted'), + /after durable append/ + ); + const llm = provider(); + await agent(t, dir, llm).chat(key, 'retry'); + assert.equal( + users(llm.requests.at(-1)).filter((text) => text === 'ack crash constraint').length, + 1 + ); +}); + +test('two process restart rounds carry one copy and isolate owner/session', async (t) => { + const dir = await temporary(t); + const child = async (mode, key, text, hardStop = false) => { + const agentModule = new URL(modulePath('core/agent/moss-agent'), import.meta.url).href; + const storeModule = new URL(modulePath('core/session/jsonl-session-store'), import.meta.url) + .href; + const program = ` + import { MossAgent } from ${JSON.stringify(agentModule)}; + import { JsonlSessionStore } from ${JSON.stringify(storeModule)}; + const provider = ${provider.toString()}; + const users = ${users.toString()}; + const llm = provider(${mode === 'admit'}); + const key = ${JSON.stringify(key)}; + const text = ${JSON.stringify(text)}; + const value = new MossAgent({ workspaceDir: ${JSON.stringify(dir)}, sessionStore: new JsonlSessionStore({ dir: ${JSON.stringify(path.join(dir, 'sessions'))} }), llmProvider: llm, baseSystemPrompt: 'test', domainPrompt: false, includeAgentBehaviorPrompt: false, includeLanguagePolicyPrompt: false, enableSteering: false, enableFollowUpGuard: false, maxAgentTurns: 2 }); + if (${mode === 'admit'}) { + const controller = new AbortController(); + const run = value.chat(key, 'first', { runId: 'terminated-run', abortSignal: controller.signal }); + await llm.ready; + if (!value.steer(key, text)) throw new Error('steer was not admitted'); + if (${hardStop}) process.exit(0); + controller.abort(); + await run.catch(() => {}); + } else { + if (${hardStop}) value.recoverDeferredSteers(key, ['terminated-run']); + await value.chat(key, 'next'); + const first = users(llm.requests.at(-1)).filter((item) => item === text).length; + await value.chat(key, 'again'); + const second = users(llm.requests.at(-1)).filter((item) => item === text).length; + console.log(JSON.stringify({ first, second })); + } + await value.close(); + `; + const result = await runProcess(process.execPath, { + args: [...process.execArgv, '--input-type=module', '--eval', program], + timeout: 20_000, + }); + assert.equal(result.exitCode, 0, result.stderr); + if (mode !== 'admit') return JSON.parse(result.stdout.trim().split('\n').at(-1)); + }; + for (const hardStop of [false, true]) { + const key = `owner-a:restart-${Number(hardStop)}`; + const text = `restart constraint ${Number(hardStop)}`; + await child('admit', key, text, hardStop); + assert.deepEqual(await child('resume', `owner-b:restart-${Number(hardStop)}`, text), { + first: 0, + second: 0, + }); + assert.deepEqual(await child('resume', 'owner-a:unrelated', text), { first: 0, second: 0 }); + assert.deepEqual(await child('resume', key, text, hardStop), { first: 1, second: 1 }); + } +}); + +test('failed append does not acknowledge a deferred delivery', async (t) => { + const dir = await temporary(t); + const key = 'owner-a:session'; + const entry = await leaveDeferred(t, dir, key, 'retry after failed append'); + const store = new JsonlSessionStore({ dir: path.join(dir, 'sessions') }); + store.appendMessageOnce = async () => { + throw new Error('disk unavailable'); + }; + const interrupted = agent(t, dir, provider(), store); + await assert.rejects(interrupted.chat(key, 'next'), /disk unavailable/); + assert.equal( + interrupted.inboxPending(key).some((pending) => pending.id === entry.id), + true + ); + const llm = provider(); + await agent(t, dir, llm).chat(key, 'retry'); + assert.equal( + users(llm.requests.at(-1)).filter((text) => text === 'retry after failed append').length, + 1 + ); +}); + +test('durable continuation is requested only for a deferred handoff and failure keeps it pending', async (t) => { + const dir = await temporary(t); + const key = 'owner-a:session'; + const store = new JsonlSessionStore({ dir: path.join(dir, 'sessions') }); + let adoptions = 0; + store.ensureDurableSession = async () => { + adoptions++; + throw new Error('capture failed'); + }; + await agent(t, dir, provider(), store).chat(key, 'ordinary'); + assert.equal(adoptions, 0); + const entry = await leaveDeferred(t, dir, key, 'needs durable adoption'); + assert.equal(entry.metadata.producer.host, os.hostname()); + assert.equal(entry.metadata.producer.pid, process.pid); + assert.equal(typeof entry.metadata.producer.instanceId, 'string'); + const pending = agent(t, dir, provider(), store); + await assert.rejects(pending.chat(key, 'next'), /capture failed/); + assert.equal(adoptions, 1); + assert.equal( + pending.inboxPending(key).some((item) => item.id === entry.id), + true + ); +}); + +test('in-run steering is acknowledged only after append-once and survives a lost acknowledgement', async (t) => { + const dir = await temporary(t); + const key = 'owner-a:session'; + const store = new JsonlSessionStore({ dir: path.join(dir, 'sessions') }); + const append = store.appendMessageOnce.bind(store); + let committed; + store.appendMessageOnce = async (...args) => { + committed = args[1]; + await append(...args); + throw new Error('in-run acknowledgement interrupted'); + }; + let calls = 0; + const llm = provider(); + llm.stream = async (_options, onEvent) => { + calls++; + const response = + calls === 1 + ? { + stopReason: 'tool_use', + content: [{ type: 'tool_use', id: 'inspect-1', name: 'inspect', input: {} }], + } + : { stopReason: 'end_turn', content: [{ type: 'text', text: 'ok' }] }; + onEvent({ type: 'message_start' }); + for (const block of response.content) { + if (block.type === 'tool_use') { + onEvent({ type: 'content_block_start', toolUse: { id: block.id, name: block.name } }); + onEvent({ type: 'content_block_delta', partialJson: '{}' }); + } else { + onEvent({ type: 'content_block_start' }); + onEvent({ type: 'content_block_delta', text: block.text }); + } + onEvent({ type: 'content_block_stop' }); + } + onEvent({ type: 'message_delta', stopReason: response.stopReason }); + onEvent({ type: 'message_stop' }); + return response; + }; + const first = agent(t, dir, llm, store); + let entry; + first.tools.register({ + name: 'inspect', + description: 'inspect', + metadata: { sideEffectClass: 'readonly' }, + inputSchema: { type: 'object', properties: {} }, + async execute() { + entry = first.steer(key, 'live steering constraint'); + return 'done'; + }, + }); + await assert.rejects(first.chat(key, 'inspect'), /in-run acknowledgement interrupted/); + assert.equal(committed, entry.id, 'the internal message keeps its original delivery identity'); + assert.equal( + first.inboxPending(key).some((item) => item.id === entry.id), + true + ); + const resumed = provider(); + await agent(t, dir, resumed).chat(key, 'resume'); + assert.equal( + users(resumed.requests.at(-1)).filter((text) => text.includes('live steering constraint')) + .length, + 1 + ); +}); + +test('inbox corruption and pre-commit persistence failures cannot look like successful admission', async (t) => { + const dir = await temporary(t); + const file = path.join(dir, 'inbox.json'); + await fs.writeFile(file, '{bad json'); + assert.throws(() => loadSessionInbox(file)); + await fs.rm(file); + updateSessionInbox(file, (inbox) => inbox.admit({ id: 'existing', prompt: 'keep' })); + const rename = fsSync.renameSync; + try { + fsSync.renameSync = (from, to) => { + if (to === file) throw new Error('rename before commit failed'); + return rename(from, to); + }; + assert.throws( + () => updateSessionInbox(file, (inbox) => inbox.admit({ id: 'not-admitted', prompt: 'new' })), + /before commit failed/ + ); + } finally { + fsSync.renameSync = rename; + } + assert.deepEqual( + loadSessionInbox(file) + .pending() + .map((entry) => entry.id), + ['existing'] + ); + updateSessionInbox(file, (inbox) => inbox.admit({ id: 'retry', prompt: 'works' })); + assert.deepEqual( + loadSessionInbox(file) + .pending() + .map((entry) => entry.id), + ['existing', 'retry'] + ); +}); + +test('a receipt retry does not authorize replacement from a stale history view', async (t) => { + const dir = await temporary(t); + const first = new JsonlSessionStore({ dir }); + await first.appendMessageOnce('session', 'delivery-1', user('base')); + const stale = new JsonlSessionStore({ dir }); + await stale.loadMessages('session'); + await first.appendMessage('session', user('concurrent addition')); + assert.equal(await stale.appendMessageOnce('session', 'delivery-1', user('base')), false); + await assert.rejects( + stale.replaceMessages('session', [user('stale compacted')]), + /Session changed before replacement/ + ); + assert.deepEqual(users(await first.loadMessages('session')), ['base', 'concurrent addition']); +}); + +test('durable inbox refresh preserves queued work admitted during an explicit drain', async (t) => { + const dir = await temporary(t); + const llm = provider(); + const original = llm.stream.bind(llm); + let value; + let calls = 0; + llm.stream = async (...args) => { + if (calls++ === 0) value.admit('queue', 'second queued prompt', { delivery: 'queue' }); + return original(...args); + }; + value = agent(t, dir, llm); + value.admit('queue', 'first queued prompt', { delivery: 'queue' }); + const result = await value.drainInbox('queue'); + assert.equal(result.chats.length, 2); + assert.equal(value.inboxPending('queue').length, 0); +}); diff --git a/test/sdk-contract.spec.mjs b/test/sdk-contract.spec.mjs index 82029f919..e235f9e2e 100644 --- a/test/sdk-contract.spec.mjs +++ b/test/sdk-contract.spec.mjs @@ -347,6 +347,8 @@ test('typed API surface resolves through dist/index.d.ts', async () => { for (const symbol of [ 'MossAgent', 'MossAgentConfig', + 'SessionStore', + 'SessionDeliverySnapshot', 'runAgentLoop', 'MossError', 'builtinTools', From c86a6f569baca92957c22b741743e8352db31277 Mon Sep 17 00:00:00 2001 From: d-robotics Date: Sun, 25 Oct 2026 03:54:40 +0800 Subject: [PATCH 2/4] fix(cli): release doctor probe response bodies --- src/cli/doctor-board-proxy.ts | 7 +- src/cli/doctor-rdk-deps.ts | 7 + src/cli/doctor.ts | 7 +- test/cli-doctor-response-cleanup.spec.mjs | 159 ++++++++++++++++++++++ 4 files changed, 176 insertions(+), 4 deletions(-) create mode 100644 test/cli-doctor-response-cleanup.spec.mjs diff --git a/src/cli/doctor-board-proxy.ts b/src/cli/doctor-board-proxy.ts index ba5f9be1a..68d0800c2 100644 --- a/src/cli/doctor-board-proxy.ts +++ b/src/cli/doctor-board-proxy.ts @@ -103,18 +103,21 @@ export async function probeGatewayReachability(input: { const timeoutMs = input.timeoutMs ?? GATEWAY_PROBE_MS; const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), timeoutMs); + let response: Response | undefined; try { - const res = await fetchFn(target, { + response = await fetchFn(target, { method: 'GET', headers: { 'user-agent': 'moss-doctor/1.0' }, signal: controller.signal, }); - return { kind: 'ok', host, status: res.status }; + return { kind: 'ok', host, status: response.status }; } catch (err) { const msg = err instanceof Error ? err.message : String(err); return { kind: 'unreachable', host, error: msg }; } finally { clearTimeout(timer); + controller.abort(); + await response?.body?.cancel().catch(() => {}); } } diff --git a/src/cli/doctor-rdk-deps.ts b/src/cli/doctor-rdk-deps.ts index 01879e3fd..9bf0162e3 100644 --- a/src/cli/doctor-rdk-deps.ts +++ b/src/cli/doctor-rdk-deps.ts @@ -149,12 +149,14 @@ export async function probeVerifiedWhls(input: { for (const item of list) { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), timeoutMs); + const responses: Response[] = []; try { const res = await fetchFn(item.url, { method: 'HEAD', signal: controller.signal, headers: { 'user-agent': 'moss-doctor/1.0' }, }); + responses.push(res); // Some CDNs reject HEAD; fall back to ranged GET. if (res.status === 405 || res.status === 501) { const getRes = await fetchFn(item.url, { @@ -162,6 +164,7 @@ export async function probeVerifiedWhls(input: { signal: controller.signal, headers: { 'user-agent': 'moss-doctor/1.0', range: 'bytes=0-0' }, }); + responses.push(getRes); results.push({ board: item.board, url: item.url, @@ -175,6 +178,10 @@ export async function probeVerifiedWhls(input: { results.push({ board: item.board, url: item.url, ok: false }); } finally { clearTimeout(timer); + controller.abort(); + for (const response of responses) { + await response.body?.cancel().catch(() => {}); + } } } return results; diff --git a/src/cli/doctor.ts b/src/cli/doctor.ts index 257814f07..b9fd0f2fb 100644 --- a/src/cli/doctor.ts +++ b/src/cli/doctor.ts @@ -393,16 +393,19 @@ async function probeKeylessSearchReachability( ): Promise<'ok' | 'http-error' | 'unreachable'> { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), timeoutMs); + let response: Response | undefined; try { - const res = await fetch('https://www.bing.com/search?q=moss+agent', { + response = await fetch('https://www.bing.com/search?q=moss+agent', { headers: { 'user-agent': 'Mozilla/5.0 (compatible; moss-doctor/1.0)' }, signal: controller.signal, }); - return res.ok ? 'ok' : 'http-error'; + return response.ok ? 'ok' : 'http-error'; } catch { return 'unreachable'; } finally { clearTimeout(timer); + controller.abort(); + await response?.body?.cancel().catch(() => {}); } } diff --git a/test/cli-doctor-response-cleanup.spec.mjs b/test/cli-doctor-response-cleanup.spec.mjs new file mode 100644 index 000000000..10a39b75a --- /dev/null +++ b/test/cli-doctor-response-cleanup.spec.mjs @@ -0,0 +1,159 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs/promises'; +import { createServer } from 'node:http'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { fileURLToPath, pathToFileURL } from 'node:url'; +import { probeGatewayReachability } from '../dist/cli/doctor-board-proxy.js'; +import { probeVerifiedWhls } from '../dist/cli/doctor-rdk-deps.js'; +import { runProcess } from '../dist/utils/run-process.js'; + +const root = path.dirname(path.dirname(fileURLToPath(import.meta.url))); + +test('gateway releases its response even when body cancellation rejects', async () => { + let signal; + let cancelled = false; + const result = await probeGatewayReachability({ + baseUrl: 'https://gateway.example/v1', + fetchImpl: async (_url, init) => { + signal = init.signal; + return new Response( + new ReadableStream({ + cancel() { + cancelled = true; + throw new Error('body cancellation failed'); + }, + }), + { status: 401 } + ); + }, + }); + assert.deepEqual(result, { kind: 'ok', host: 'gateway.example', status: 401 }); + assert.equal(cancelled, true); + assert.equal(signal.aborted, true); +}); + +for (const [headStatus, getStatus] of [[200], [405, 206], [501, 500], [405, 'reject']]) { + test(`whl releases every acquired response for HEAD ${headStatus} / GET ${getStatus ?? 'unused'}`, async () => { + const cancelled = []; + const signals = []; + const result = await probeVerifiedWhls({ + urls: [{ board: 'x5', url: 'https://whl.example/x5.whl' }], + fetchImpl: async (_url, init) => { + signals.push(init.signal); + if (init.method === 'GET' && getStatus === 'reject') throw new Error('GET failed'); + return new Response( + new ReadableStream({ + cancel() { + cancelled.push(init.method); + throw new Error('body cancellation failed'); + }, + }), + { status: init.method === 'HEAD' ? headStatus : getStatus } + ); + }, + }); + const status = getStatus === 'reject' ? undefined : (getStatus ?? headStatus); + assert.deepEqual(result, [ + { + board: 'x5', + url: 'https://whl.example/x5.whl', + ok: status === 200 || status === 206, + ...(status === undefined ? {} : { status }), + }, + ]); + assert.deepEqual( + cancelled, + getStatus === undefined || getStatus === 'reject' ? ['HEAD'] : ['HEAD', 'GET'] + ); + assert.equal( + signals.every((signal) => signal.aborted), + true + ); + }); +} + +// Only the selected probe uses real HTTP. Other probe responses are explicit +// closed fixtures; no request is sent to the public URLs printed by doctor. +for (const selected of ['gateway', 'search', 'whl']) { + test(`doctor exits after reporting a ${selected} response whose body never ends`, async (t) => { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), 'moss-doctor-body-')); + t.after(() => fs.rm(directory, { recursive: true, force: true })); + const requests = []; + const server = createServer((req, res) => { + requests.push({ method: req.method, url: req.url }); + if (req.method === 'HEAD') { + res.writeHead(405); + res.end(); + return; + } + res.writeHead(selected === 'whl' ? 206 : 200, { 'content-type': 'text/plain' }); + res.write('headers and one chunk, deliberately without an end'); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + t.after(async () => { + server.closeAllConnections(); + await new Promise((resolve) => server.close(resolve)); + }); + const endpoint = `http://127.0.0.1:${server.address().port}`; + const preload = path.join(directory, 'probe-fixture.mjs'); + await fs.writeFile( + preload, + ` + const realFetch = globalThis.fetch; + globalThis.fetch = async (input, init) => { + const url = new URL(String(input)); + const kind = url.hostname === 'www.bing.com' ? 'search' : url.pathname.endsWith('.whl') ? 'whl' : 'gateway'; + if (kind !== ${JSON.stringify(selected)}) return new Response(null, { status: kind === 'gateway' ? 421 : 200 }); + return realFetch(${JSON.stringify(endpoint)} + '/' + kind, init); + }; + ` + ); + const env = { + PATH: process.env.PATH ?? '', + HOME: directory, + MOSS_CONFIG_DIR: path.join(directory, 'config'), + MOSS_NO_BUNDLED_DEFAULT: '1', + MOSS_RDK_DOCS_AUTO: '0', + OPENAI_API_KEY: 'sk-doctor-local-fixture', + LANG: 'C', + LC_ALL: 'C', + TERM: 'dumb', + NO_COLOR: '1', + FORCE_COLOR: '0', + ...(process.env.SystemRoot ? { SystemRoot: process.env.SystemRoot } : {}), + }; + let result; + try { + result = await runProcess(process.execPath, { + args: [ + '--import', + pathToFileURL(preload).href, + path.join(root, 'dist', 'cli.js'), + 'doctor', + '-C', + directory, + ], + cwd: directory, + env, + timeout: 4_000, + stdin: '', + }); + } catch (error) { + assert.match( + error.stderr ?? '', + /sessions: no sessions directory yet/, + 'the complete report was printed before the process got stuck' + ); + throw error; + } + assert.equal(result.exitCode, 0, result.stderr); + assert.match(result.stderr, /sessions: no sessions directory yet/); + assert.equal( + requests.some((request) => request.method === 'GET' && request.url === `/${selected}`), + true, + 'the selected probe used the real loopback HTTP stream' + ); + }); +} From 1d52edb55dc5cad4a96d60670d066afac23346dd Mon Sep 17 00:00:00 2001 From: d-robotics Date: Sun, 25 Oct 2026 04:23:48 +0800 Subject: [PATCH 3/4] fix(session): separate records after interrupted JSONL tails --- src/core/session/jsonl-session-store.ts | 5 +- test/jsonl-session-tail-recovery.spec.mjs | 204 ++++++++++++++++++++++ 2 files changed, 208 insertions(+), 1 deletion(-) create mode 100644 test/jsonl-session-tail-recovery.spec.mjs diff --git a/src/core/session/jsonl-session-store.ts b/src/core/session/jsonl-session-store.ts index 27bc3a8f3..bc5299d34 100644 --- a/src/core/session/jsonl-session-store.ts +++ b/src/core/session/jsonl-session-store.ts @@ -451,7 +451,10 @@ export class JsonlSessionStore implements SessionStore { recoverable: true, }); } - await this.appendLineDurably(filePath, entry + '\n'); + // A stopped write may leave a complete or partial unterminated line. + // Preserve its bytes while isolating the new record in this same lock. + const separator = currentRaw.length > 0 && !currentRaw.endsWith('\n') ? '\n' : ''; + await this.appendLineDurably(filePath, separator + entry + '\n'); appended = true; const raw = await this.readRaw(filePath); this.observedContentVersions.set(filePath, this.contentVersion(raw)); diff --git a/test/jsonl-session-tail-recovery.spec.mjs b/test/jsonl-session-tail-recovery.spec.mjs new file mode 100644 index 000000000..e70c277d9 --- /dev/null +++ b/test/jsonl-session-tail-recovery.spec.mjs @@ -0,0 +1,204 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { JsonlSessionStore } from '../dist/index.js'; +import { ErrorCode } from '../dist/errors.js'; +import { runProcess } from '../dist/utils/run-process.js'; + +const key = 'owner:tail-session'; +const user = (content) => ({ role: 'user', content }); +const initial = user('complete history'); +const tailMessage = user('complete final record'); +const completeTail = JSON.stringify({ type: 'message', message: tailMessage, deliveryId: 'tail' }); +const tails = [ + { + name: 'partial JSON without newline', + raw: '{"type":"message","message":', + messages: [], + receipts: [], + }, + { + name: 'complete JSON without newline', + raw: completeTail, + messages: [tailMessage], + receipts: ['tail'], + }, + { + name: 'complete JSON with newline', + raw: completeTail + '\n', + messages: [tailMessage], + receipts: ['tail'], + }, +]; + +async function fixture(t, tail) { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'moss-jsonl-tail-')); + t.after(() => fs.rm(dir, { recursive: true, force: true })); + const file = path.join(dir, `${encodeURIComponent(key)}.jsonl`); + const raw = JSON.stringify({ type: 'message', message: initial }) + '\n' + tail; + // Deterministic images of a stopped writer; subsequent writes and reads use + // real stores in separate OS processes, with no mocked persistence. + await fs.writeFile(file, raw); + return { dir, file, raw }; +} + +async function child(dir, body) { + const moduleUrl = new URL('../dist/index.js', import.meta.url).href; + const result = await runProcess(process.execPath, { + args: [ + '--input-type=module', + '--eval', + ` + import { JsonlSessionStore } from ${JSON.stringify(moduleUrl)}; + const store = new JsonlSessionStore({ dir: ${JSON.stringify(dir)} }); + const key = ${JSON.stringify(key)}; + const user = (content) => ({ role: 'user', content }); + ${body} + `, + ], + cwd: dir, + timeout: 10_000, + }); + assert.equal(result.exitCode, 0, result.stderr); + return JSON.parse(result.stdout.trim()); +} + +for (const tail of tails) { + test(`append-once survives process restarts after ${tail.name}`, async (t) => { + const { dir, file, raw } = await fixture(t, tail.raw); + assert.equal( + await child( + dir, + ` + console.log(JSON.stringify(await store.appendMessageOnce(key, 'first', user('first delivery')))); + ` + ), + true + ); + const afterFirst = await fs.readFile(file, 'utf8'); + const restarted = await child( + dir, + ` + const before = await store.loadDeliverySnapshot(key); + const retry = await store.appendMessageOnce(key, 'first', user('first delivery')); + const second = await store.appendMessageOnce(key, 'second', user('second delivery')); + console.log(JSON.stringify({ before, retry, second })); + ` + ); + assert.deepEqual(restarted.before, { + messages: [initial, ...tail.messages, user('first delivery')], + deliveryIds: [...tail.receipts, 'first'], + }); + assert.equal(restarted.retry, false); + assert.equal(restarted.second, true); + const final = await child( + dir, + ` + const retry = await store.appendMessageOnce(key, 'second', user('second delivery')); + console.log(JSON.stringify({ retry, snapshot: await store.loadDeliverySnapshot(key) })); + ` + ); + assert.equal(final.retry, false); + assert.deepEqual(final.snapshot, { + messages: [initial, ...tail.messages, user('first delivery'), user('second delivery')], + deliveryIds: [...tail.receipts, 'first', 'second'], + }); + assert.equal(afterFirst.slice(0, raw.length), raw, 'existing tail bytes are never rewritten'); + const appendedLines = afterFirst.slice(raw.length).split('\n'); + if (!raw.endsWith('\n')) + assert.equal(appendedLines.shift(), '', 'a new line isolates the old tail'); + assert.equal(JSON.parse(appendedLines[0]).deliveryId, 'first'); + assert.deepEqual(appendedLines.slice(1), [''], 'only one complete new record is appended'); + }); + + test(`ordinary append survives process restarts after ${tail.name}`, async (t) => { + const { dir, file, raw } = await fixture(t, tail.raw); + assert.deepEqual( + await child( + dir, + ` + await store.appendMessage(key, user('ordinary first')); + console.log(JSON.stringify(await store.loadMessages(key))); + ` + ), + [initial, ...tail.messages, user('ordinary first')] + ); + assert.deepEqual( + await child( + dir, + ` + await store.appendMessage(key, user('ordinary second')); + console.log(JSON.stringify(await store.loadMessages(key))); + ` + ), + [initial, ...tail.messages, user('ordinary first'), user('ordinary second')] + ); + assert.equal((await fs.readFile(file, 'utf8')).slice(0, raw.length), raw); + }); +} + +test('receipt hits do not mutate an unterminated tail', async (t) => { + const { dir, file, raw } = await fixture(t, completeTail); + assert.equal( + await child( + dir, + ` + console.log(JSON.stringify(await store.appendMessageOnce(key, 'tail', user('must not append')))); + ` + ), + false + ); + assert.equal(await fs.readFile(file, 'utf8'), raw); +}); + +test('external tail changes still reject stale append and replacement without mutation', async (t) => { + const { dir, file } = await fixture(t, completeTail); + const stale = new JsonlSessionStore({ dir }); + const messages = await stale.loadMessages(key); + await fs.appendFile(file, '\n{"type":"message","message":'); + const changed = await fs.readFile(file, 'utf8'); + for (const append of [ + () => stale.appendMessageOnce(key, 'new', user('stale delivery')), + () => stale.appendMessage(key, user('stale ordinary append')), + () => stale.replaceMessages(key, messages), + ]) { + await assert.rejects(append, (error) => error.code === ErrorCode.SESSION_PERSIST_FAILED); + assert.equal(await fs.readFile(file, 'utf8'), changed); + } +}); + +test('replacement preserves a receipt from a complete unterminated tail', async (t) => { + const { dir, file } = await fixture(t, completeTail); + await child( + dir, + ` + await store.loadMessages(key); + await store.replaceMessages(key, [user('compacted')]); + console.log('true'); + ` + ); + const result = await child( + dir, + ` + const retry = await store.appendMessageOnce(key, 'tail', user('must not replay')); + const next = await store.appendMessageOnce(key, 'next', user('next delivery')); + console.log(JSON.stringify({ retry, next, snapshot: await store.loadDeliverySnapshot(key) })); + ` + ); + assert.deepEqual(result, { + retry: false, + next: true, + snapshot: { + messages: [user('compacted'), user('next delivery')], + deliveryIds: ['tail', 'next'], + }, + }); + const lines = (await fs.readFile(file, 'utf8')) + .trimEnd() + .split('\n') + .map((line) => JSON.parse(line)); + assert.equal(lines.length, 2); + assert.equal(lines[0].type, 'state_replace'); +}); From 27eca3776bc84c54753bea2ee0498b12b8971161 Mon Sep 17 00:00:00 2001 From: d-robotics Date: Sun, 25 Oct 2026 04:41:44 +0800 Subject: [PATCH 4/4] fix(session): preserve durable receipt identity on retry --- src/core/session/jsonl-session-store.ts | 19 +- .../jsonl-session-receipt-durability.spec.mjs | 237 ++++++++++++++++++ 2 files changed, 253 insertions(+), 3 deletions(-) create mode 100644 test/jsonl-session-receipt-durability.spec.mjs diff --git a/src/core/session/jsonl-session-store.ts b/src/core/session/jsonl-session-store.ts index bc5299d34..d2ba54563 100644 --- a/src/core/session/jsonl-session-store.ts +++ b/src/core/session/jsonl-session-store.ts @@ -420,7 +420,12 @@ export class JsonlSessionStore implements SessionStore { deliveryId?: string ): Promise { const filePath = this.sessionPath(sessionKey); - const entry = redactJsonLine({ type: 'message', message, deliveryId, ts: Date.now() }); + const entry = JSON.stringify({ + type: 'message', + message: JSON.parse(redactJsonLine(message)) as LLMMessage, + deliveryId, + ts: Date.now(), + }); let appended = false; const mutate = async () => { let isNewSession = false; @@ -438,6 +443,14 @@ export class JsonlSessionStore implements SessionStore { // a committed delivery is safe even from a stale store instance. // It does not refresh that instance's history view: only an actual // load may authorize a later replacement of concurrent messages. + // A prior append may have written the receipt but failed its fsync. + const handle = await fsp.open(filePath, 'r+'); + try { + await handle.sync(); + } finally { + await handle.close(); + } + await syncDirectoryWith(path.dirname(filePath), durableDirectoryOperations); return; } const observedVersion = this.observedContentVersions.get(filePath); @@ -509,9 +522,9 @@ export class JsonlSessionStore implements SessionStore { ]), ]; const nextRaw = - redactJsonLine({ + JSON.stringify({ type: 'state_replace', - messages, + messages: JSON.parse(redactJsonLine(messages)) as LLMMessage[], deliveryIds, ts: Date.now(), }) + '\n'; diff --git a/test/jsonl-session-receipt-durability.spec.mjs b/test/jsonl-session-receipt-durability.spec.mjs new file mode 100644 index 000000000..6e2557cf1 --- /dev/null +++ b/test/jsonl-session-receipt-durability.spec.mjs @@ -0,0 +1,237 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs/promises'; +import { syncBuiltinESMExports } from 'node:module'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { JsonlSessionStore } from '../dist/index.js'; +import { runProcess } from '../dist/utils/run-process.js'; + +const key = 'owner:receipt-session'; +const user = (content) => ({ role: 'user', content }); +const eio = (message) => Object.assign(new Error(message), { code: 'EIO' }); + +async function fixture(t) { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'moss-jsonl-receipt-')); + t.after(() => fs.rm(dir, { recursive: true, force: true })); + return { dir, file: path.join(dir, `${encodeURIComponent(key)}.jsonl`) }; +} + +function interceptOpen(t, intercept) { + const original = fs.open; + fs.open = (target, flags, ...args) => intercept(original, target, flags, ...args); + syncBuiltinESMExports(); + t.after(() => { + fs.open = original; + syncBuiltinESMExports(); + }); +} + +async function restartedSnapshot(dir) { + const moduleUrl = new URL('../dist/index.js', import.meta.url).href; + const result = await runProcess(process.execPath, { + args: [ + '--input-type=module', + '--eval', + ` + import { JsonlSessionStore } from ${JSON.stringify(moduleUrl)}; + const store = new JsonlSessionStore({ dir: ${JSON.stringify(dir)} }); + console.log(JSON.stringify(await store.loadDeliverySnapshot(${JSON.stringify(key)}))); + `, + ], + cwd: dir, + timeout: 10_000, + }); + assert.equal(result.exitCode, 0, result.stderr); + return JSON.parse(result.stdout.trim()); +} + +test('a duplicate receipt retries failed file fsync before acknowledging and closes every handle', async (t) => { + const { dir, file } = await fixture(t); + let failing = true; + let syncAttempts = 0; + let closes = 0; + const flagsSeen = []; + interceptOpen(t, async (open, target, flags, ...args) => { + const handle = await open(target, flags, ...args); + if (target === file) { + flagsSeen.push(flags); + const sync = handle.sync.bind(handle); + const close = handle.close.bind(handle); + handle.sync = async () => { + syncAttempts++; + if (failing) throw eio('persistent data fsync failure'); + await sync(); + }; + handle.close = async () => { + closes++; + await close(); + }; + } + return handle; + }); + await assert.rejects( + new JsonlSessionStore({ dir }).appendMessageOnce(key, 'delivery', user('one delivery')), + { code: 'EIO' } + ); + assert.equal(syncAttempts, 1); + assert.equal(closes, 1); + const written = await fs.readFile(file, 'utf8'); + await assert.rejects( + new JsonlSessionStore({ dir }).appendMessageOnce(key, 'delivery', user('one delivery')), + { code: 'EIO' } + ); + assert.equal(syncAttempts, 2, 'a visible receipt cannot bypass the failed persistence barrier'); + assert.equal(closes, 2); + failing = false; + assert.equal( + await new JsonlSessionStore({ dir }).appendMessageOnce(key, 'delivery', user('one delivery')), + false + ); + assert.equal(syncAttempts, 3); + assert.equal(closes, 3); + assert.deepEqual(flagsSeen, ['a', 'r+', 'r+']); + assert.equal(await fs.readFile(file, 'utf8'), written, 'retries never append or rewrite history'); + assert.deepEqual(await restartedSnapshot(dir), { + messages: [user('one delivery')], + deliveryIds: ['delivery'], + }); +}); + +if (process.platform !== 'win32') { + test('a duplicate receipt requires its parent directory barrier and closes on failure', async (t) => { + const { dir, file } = await fixture(t); + await new JsonlSessionStore({ dir }).appendMessageOnce(key, 'delivery', user('one delivery')); + const written = await fs.readFile(file, 'utf8'); + let failing = true; + let armed = false; + let directoryAttempts = 0; + let dataCloses = 0; + let barrierCloses = 0; + interceptOpen(t, async (open, target, flags, ...args) => { + const handle = await open(target, flags, ...args); + if (target === file) { + assert.equal(flags, 'r+'); + const sync = handle.sync.bind(handle); + const close = handle.close.bind(handle); + handle.sync = async () => { + await sync(); + armed = true; + }; + handle.close = async () => { + dataCloses++; + await close(); + }; + } else if (target === dir && armed) { + armed = false; + const sync = handle.sync.bind(handle); + const close = handle.close.bind(handle); + handle.sync = async () => { + directoryAttempts++; + if (failing) throw eio('receipt directory fsync failure'); + await sync(); + }; + handle.close = async () => { + barrierCloses++; + await close(); + }; + } + return handle; + }); + for (let attempt = 1; attempt <= 2; attempt++) { + await assert.rejects( + new JsonlSessionStore({ dir }).appendMessageOnce(key, 'delivery', user('one delivery')), + { code: 'EIO' } + ); + assert.equal(directoryAttempts, attempt); + assert.equal(dataCloses, attempt); + assert.equal(barrierCloses, attempt); + } + failing = false; + assert.equal( + await new JsonlSessionStore({ dir }).appendMessageOnce(key, 'delivery', user('one delivery')), + false + ); + assert.equal(directoryAttempts, 3); + assert.equal(dataCloses, 3); + assert.equal(barrierCloses, 3); + assert.equal(await fs.readFile(file, 'utf8'), written); + }); +} + +test('a receipt barrier cannot recreate a data file removed before opening it', async (t) => { + const { dir, file } = await fixture(t); + await new JsonlSessionStore({ dir }).appendMessageOnce(key, 'delivery', user('one delivery')); + let opens = 0; + interceptOpen(t, async (open, target, flags, ...args) => { + if (target === file) { + opens++; + await fs.unlink(file); + assert.equal(flags, 'r+', 'the receipt barrier must only open an existing file'); + } + return open(target, flags, ...args); + }); + await assert.rejects( + new JsonlSessionStore({ dir }).appendMessageOnce(key, 'delivery', user('one delivery')), + { code: 'ENOENT' } + ); + assert.equal(opens, 1); + await assert.rejects(fs.access(file), { code: 'ENOENT' }); +}); + +const firstId = 'sk-fixture000000000000000000000000000000'; +const secondId = 'sk-fixture111111111111111111111111111111'; +const importedId = 'sk-fixture222222222222222222222222222222'; +const bodySecret = 'sk-bodysecret333333333333333333333333333'; +const password = 'SyntheticPassword123'; +const sensitive = user(`api_key=${bodySecret}\nDB_PASSWORD=${password}`); + +test('append preserves exact receipt IDs while redacting message secrets', async (t) => { + const { dir, file } = await fixture(t); + assert.equal( + await new JsonlSessionStore({ dir }).appendMessageOnce(key, firstId, sensitive), + true + ); + assert.equal( + await new JsonlSessionStore({ dir }).appendMessageOnce(key, firstId, sensitive), + false + ); + assert.equal( + await new JsonlSessionStore({ dir }).appendMessageOnce(key, secondId, sensitive), + true + ); + const snapshot = await restartedSnapshot(dir); + assert.deepEqual(snapshot.deliveryIds, [firstId, secondId]); + assert.equal(snapshot.messages.length, 2); + const raw = await fs.readFile(file, 'utf8'); + assert.equal(raw.includes(bodySecret), false); + assert.equal(raw.includes(password), false); + assert.match(JSON.stringify(snapshot.messages), /\[REDACTED\]/); +}); + +test('replacement preserves carried and imported receipt IDs while redacting message secrets', async (t) => { + const { dir, file } = await fixture(t); + // A valid existing receipt image isolates replacement from append redaction. + await fs.writeFile( + file, + JSON.stringify({ type: 'message', message: user('before'), deliveryId: firstId }) + '\n' + ); + const store = new JsonlSessionStore({ dir }); + await store.loadMessages(key); + await store.replaceMessages(key, [sensitive], { deliveryIds: [importedId] }); + const raw = await fs.readFile(file, 'utf8'); + assert.equal(raw.includes(bodySecret), false); + assert.equal(raw.includes(password), false); + assert.deepEqual(await restartedSnapshot(dir), { + messages: [user('api_key=[REDACTED]\nDB_PASSWORD=[REDACTED]')], + deliveryIds: [firstId, importedId], + }); + for (const id of [firstId, importedId]) { + assert.equal(await new JsonlSessionStore({ dir }).appendMessageOnce(key, id, sensitive), false); + } + assert.equal( + await new JsonlSessionStore({ dir }).appendMessageOnce(key, secondId, sensitive), + true + ); + assert.deepEqual((await restartedSnapshot(dir)).deliveryIds, [firstId, importedId, secondId]); +});