diff --git a/docs/SURFACE.md b/docs/SURFACE.md index 13e6d03d3..7d66b7726 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -128,15 +128,42 @@ No process runs between events: the handler wakes, executes to its next await, p `close-merge-request` in addition to comments and discussions; the earlier comment-only gap was closed by the pinned relay-helpers release. - The initial local memory slice supports `recall` and `why` in authored flows, - with no journal step for either read. Script scope is stable across runs of + The local memory slice supports `recall`, `why`, and `learn` in authored + flows. None of them is a journal step: reads journal nothing, and `learn` + appends a record to the root run's `memory-learn` stream. Script scope is stable across runs of the same flow file and name; reads cannot widen it to another flow. The existing `memory: { script: true }` header enables an eager reachability check; direct `.memory` use also triggers it. Aliased access is checked at call time. The local Node SDK and an existing readable SQLite DB are required (`AI_HIST_DB` overrides `defaultDbPath()`); JSONL fallback is disabled. - `learn` and `memory: { agent: true }` refuse pending the journal-backed write - and identity-scoped agent follow-ups. CLI-only operation is also deferred. + `learn(finding)` journals first. It stages the finding's compacted + trajectory file (the format `ai-hist sync` ingests) under a temporary name, + appends the finding to the durable root run's `memory-learn` stream, and + only then renames the file into place at + `/flows-memory//.trajectories/compacted/.json`, so + nothing is recallable that the journal does not record, and disk failures + happen before anything is journaled. Files live beside the ai-hist DB, not + the flow source, so a flow deployed from a read-only checkout can still + learn. A file is derived only from its commit, which it cites + (`journal: { runId, stream, id }`); if another attempt on the root + committed the same finding first, that commit wins, so every attempt + publishes the same bytes. A resumed body that re-runs `learn` finds the + commit and publishes it without journaling again. Journal errors propagate + unchanged, so a read interruption or lost lease leaves the root resumable. + `learn` without a durable root run refuses (`memory_unjournaled`). `recall` + and `why` read the ai-hist DB plus the scope's learned files, so the next + run sees a finding before any sync; a synced copy is deduplicated in favour + of the DB row (an unsynced `recall` entry has a negative `id`), and a + damaged learned file fails the read (`memory_unreachable`) rather than + being skipped. The id hashes scope and finding, so re-learning a finding + atomically replaces one file. Trajectory files need no lock: each is + staged privately and published by atomic rename, and `learn` never + modifies the ai-hist DB; ordering between writers comes from the root run's + stream. A file takes the DB's mode, and a failed `learn` + (`memory_finding_invalid`, `memory_unwritable`) publishes nothing. A body + must await `learn`: a call whose outcome the body never consumed fails the + run with `unawaited_step`. `memory: { agent: true }` refuses pending the + identity-scoped agent follow-up. CLI-only operation is also deferred. 4. **`{{prev}}` / return-value chaining.** Output flows downward implicitly; naming steps is for reaching back, not bookkeeping. 5. **Headers are optional escalation.** identity, memory, budget, tools appear only when used. The empty header is the common case. [Budget headers and spend](BUDGET.md) specifies parsing, prices, journal attribution, and admission limits. diff --git a/packages/sdk/src/authored-flow-error.ts b/packages/sdk/src/authored-flow-error.ts index 307343791..243ca56c3 100644 --- a/packages/sdk/src/authored-flow-error.ts +++ b/packages/sdk/src/authored-flow-error.ts @@ -26,6 +26,12 @@ export type AuthoredFlowExecutionErrorCode = | 'journal_protocol_violation' | 'missing_completion' | 'memory_unreachable' + /** `f.memory.learn` was given a finding without non-empty string fields. */ + | 'memory_finding_invalid' + /** `f.memory.learn` could not persist the finding to script memory. */ + | 'memory_unwritable' + /** `f.memory.learn` ran without a durable root run to journal the finding in. */ + | 'memory_unjournaled' | 'operation_after_completion' | 'operation_callback_failed' | 'step_failed' diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts index 8205e261d..9ba457558 100644 --- a/packages/sdk/src/authored-flow-executor.ts +++ b/packages/sdk/src/authored-flow-executor.ts @@ -10,7 +10,7 @@ import type { SlackCall } from './slack-writeback.js'; import { checkMcpHeader, McpPreflightError } from './cli/check-typescript.js'; import { buildMcpProxy, runMcpEffect } from './authored-mcp.js'; import { AuthoredBudget } from './authored-budget.js'; -import { assertMemoryReachable, authoredMemory, scriptMemoryScope } from './authored-memory.js'; +import { assertMemoryReachable, authoredMemory, journalLearnLedger, MemoryWriteTracker, scriptMemoryScope } from './authored-memory.js'; import { authoredDeterministicRunner, authoredWorkerRunner } from './authored-worker-step.js'; import { isSurfaceFlowCompletionReason, isSurfaceRunCompletionReason } from './authored-step-output.js'; import { @@ -285,6 +285,7 @@ export async function executeAuthoredFlow( const journalSteps: AuthoredFlowJournalStep[] = []; const authoredSteps: AuthoredFlowOperation[] = []; + const memoryWrites = new MemoryWriteTracker(); const lifecycle = new AuthoredFlowLifecycle(); const activities = new AuthoredActivities(journal, options.rootRunId); const edgeOverrides = new Map(); @@ -563,6 +564,8 @@ export async function executeAuthoredFlow( scriptMemoryScope(flowPath, definition.name), () => assertOperationAllowed('memory', definition.name, requestedCompletion), definition.header.memory?.script !== false, + memoryWrites, + options.rootRunId === undefined ? undefined : journalLearnLedger(journal, options.rootRunId), ), run: runOperation, llm: llmOperation, @@ -818,6 +821,10 @@ export async function executeAuthoredFlow( bodyFailure = error; } if (bodyFailed) { + // An in-flight learn must land or fail before this attempt gives the root + // back: a re-driven attempt reads the memory-learn stream, and must not + // race a commit this one is still making. + await memoryWrites.settle(); try { worker.stop(bodyFailure); await stopAuthoredOperations(authoredSteps, bodyFailure); @@ -850,6 +857,8 @@ export async function executeAuthoredFlow( 'missing_completion', `flow "${definition.name}" returned without done()`, ); + // Same as a failed body: no learn may outlive the attempt that started it. + await memoryWrites.settle(); try { worker.stop(missingCompletion); await stopAuthoredOperations(authoredSteps, missingCompletion); @@ -860,6 +869,14 @@ export async function executeAuthoredFlow( throw missingCompletion; } try { + // A `learn` whose outcome the body never awaited may have failed, or may + // still be racing done(), so the run cannot claim the finding was kept. + // Let each settle, then fail the run rather than report success over it. + const unawaitedLearns = await memoryWrites.unawaited(); + if (unawaitedLearns > 0) { + throw new AuthoredFlowExecutionError('unawaited_step', + `flow "${definition.name}" returned with ${unawaitedLearns} unawaited f.memory.learn call${unawaitedLearns === 1 ? '' : 's'}; await each learn before done()`); + } await verifyAuthoredOperations(definition.name, authoredSteps, lifecycle); } catch (error) { try { diff --git a/packages/sdk/src/authored-memory.ts b/packages/sdk/src/authored-memory.ts index 62ab491da..f78e079aa 100644 --- a/packages/sdk/src/authored-memory.ts +++ b/packages/sdk/src/authored-memory.ts @@ -1,9 +1,10 @@ -import type { MemoryHelper } from '@relayflows/surface'; -import { createHash } from 'node:crypto'; -import { access, stat } from 'node:fs/promises'; +import type { HistoryEntry, MemoryFinding, MemoryHelper, MemoryRecallOptions, TrajectoryEntry } from '@relayflows/surface'; +import { createHash, randomUUID } from 'node:crypto'; +import { access, chmod, mkdir, readFile, readdir, rename, rm, stat, writeFile } from 'node:fs/promises'; import { constants } from 'node:fs'; -import { dirname, join, resolve } from 'node:path'; +import { basename, dirname, join, resolve } from 'node:path'; import { AuthoredFlowExecutionError } from './authored-flow-error.js'; +import type { JournalClient } from './journal-client.js'; import { preflightMemory } from './preflight.js'; /** Stable across runs; disjoint from other flow files and other named flows. */ @@ -27,29 +28,458 @@ export async function assertMemoryReachable(): Promise { if (refusal) throw new AuthoredFlowExecutionError('memory_unreachable', refusal.message); } +/** + * The promise `learn` returns. Its constructor is not `Promise`, so `await`, + * `Promise.all` and friends reach it through `then`, which records that the + * body consumed the result. + */ +class LearnCall extends Promise { + static override get [Symbol.species](): PromiseConstructor { return Promise; } + observed = false; + settled = false; + override then( + onFulfilled?: ((value: void) => A | PromiseLike) | null, + onRejected?: ((reason: unknown) => B | PromiseLike) | null, + ): Promise { + this.observed = true; + return super.then(onFulfilled, onRejected); + } +} + +/** Lets the executor find `learn` calls whose outcome the body never consumed. */ +export class MemoryWriteTracker { + private readonly calls: LearnCall[] = []; + track(write: Promise): Promise { + const call = new LearnCall((resolve, reject) => { void write.then(resolve, reject); }); + // Bypass the override: marking the rejection handled must not count as the body consuming it. + void Promise.prototype.then.call(call, () => { call.settled = true; }, () => { call.settled = true; }); + this.calls.push(call); + return call; + } + /** Wait for every call to settle, so no write outlives the execution that started it. */ + async settle(): Promise { + await Promise.allSettled(this.calls.map(call => Promise.prototype.then.call(call, () => {}, () => {}))); + } + /** Every call the body did not await, after letting each one settle. */ + async unawaited(): Promise { + const unobserved = this.calls.filter(call => !call.observed || !call.settled); + await Promise.allSettled(unobserved.map(call => Promise.prototype.then.call(call, () => {}, () => {}))); + return unobserved.length; + } +} + +/** Root-run stream holding every finding `learn` committed: the source of truth the trajectory files are projected from. */ +export const MEMORY_LEARN_STREAM = 'memory-learn'; + +export interface LearnRecord { + readonly memory: 'learn'; + readonly id: string; + readonly scope: string; + readonly decision: Required; + readonly learnedAt: string; +} + +export interface CommittedLearn { + readonly record: LearnRecord; + readonly runId: string; +} + +/** Where `learn` journals a finding before it may become recallable. */ +export interface LearnLedger { + /** The durable root run this ledger journals on. */ + readonly runId: string; + /** The commit already recorded for `id`, if any. */ + recorded(id: string): Promise; + /** + * Journal `record`, then return the commit that wins for its id: the first + * one in the stream. Usually that is `record`; if another attempt on this + * root committed the same finding first, it is theirs, so every attempt + * projects the same bytes. + */ + append(record: LearnRecord): Promise; +} + +function canonicalLearnRecord(raw: unknown): LearnRecord | undefined { + if (typeof raw !== 'object' || raw === null) return undefined; + const value = raw as Record; + const decision = value.decision as Record | undefined; + if (value.memory !== 'learn' || typeof value.id !== 'string' || typeof value.scope !== 'string' + || typeof value.learnedAt !== 'string' || typeof decision !== 'object' || decision === null) return undefined; + const { question, chosen, reasoning, alternatives } = decision; + if (typeof question !== 'string' || typeof chosen !== 'string' || typeof reasoning !== 'string' + || !Array.isArray(alternatives) || alternatives.some(item => typeof item !== 'string')) return undefined; + // Fixed field order, never the record as it came back: the journal stores + // stream messages with sorted keys, and the projected file must be + // byte-identical whether it is written now or republished on resume. + return { + memory: 'learn', id: value.id, scope: value.scope, + decision: { question, chosen, reasoning, alternatives: [...alternatives] as string[] }, + learnedAt: value.learnedAt, + }; +} + +/** + * A ledger on the durable root run's `memory-learn` stream. The stream is read + * once per execution and re-read after each append, so a resumed or re-driven + * body finds what any earlier attempt committed. A failed read is not cached: + * the next call reads again. Journal errors propagate unchanged, so a read + * interruption or a lost lease keeps its resumable meaning for the root. + */ +export function journalLearnLedger(journal: Pick, rootRunId: string): LearnLedger { + let loaded: Promise> | undefined; + async function read(): Promise> { + const commits = new Map(); + let offset = 0; + for (;;) { + const page = await journal.streamRead(rootRunId, MEMORY_LEARN_STREAM, offset, 1000); + for (const message of page.messages) { + const record = canonicalLearnRecord((message as { message?: unknown }).message ?? message); + // The first commit for an id wins; it is the one its file is built from. + if (record !== undefined && !commits.has(record.id)) commits.set(record.id, { record, runId: rootRunId }); + } + if (page.messages.length === 0 || page.next_offset <= offset) break; + offset = page.next_offset; + } + return commits; + } + function load(): Promise> { + loaded ??= read().catch(error => { loaded = undefined; throw error; }); + return loaded; + } + return { + runId: rootRunId, + async recorded(id) { + return (await load()).get(id); + }, + async append(record) { + await journal.streamAppend(rootRunId, MEMORY_LEARN_STREAM, record); + loaded = undefined; + return (await load()).get(record.id) ?? { record, runId: rootRunId }; + }, + }; +} + export function authoredMemory( scope: string, assertOpen: () => void, enabled: boolean, + tracker?: MemoryWriteTracker, + ledger?: LearnLedger, ): MemoryHelper { - async function read(action: (reader: import('ai-hist').AiHist) => T): Promise { - assertOpen(); + function assertEnabled(): void { if (!enabled) throw new AuthoredFlowExecutionError('unsupported_header', 'f.memory requires script memory; memory.script is false'); + } + async function read(action: (reader: import('ai-hist').AiHist, learned: LearnedTrajectory[]) => T): Promise { + assertOpen(); + assertEnabled(); await assertMemoryReachable(); assertOpen(); - const { openAiHist } = await import('ai-hist'); + const { defaultDbPath, openAiHist } = await import('ai-hist'); + const learned = await readLearned(defaultDbPath(), scope); const reader = await openAiHist({ projectScope: scope, fallback: 'error' }); - try { return action(reader); } finally { reader.close(); } + try { return action(reader, learned); } finally { reader.close(); } + } + /** + * Journal first, then publish. The trajectory file is staged under a name + * no read lists, the finding is committed to the root run's `memory-learn` + * stream, and only then is the file renamed into place. Nothing is + * recallable that the journal does not record, and the failures a disk + * write can have (space, permissions) happen while staging, before anything + * is journaled. If the process dies between the commit and the rename, the + * resumed body re-runs `learn`, finds the commit, and publishes the same + * file from it without journaling again. With no durable root run there is + * nowhere to journal, so `learn` refuses rather than write unrecorded memory. + */ + const learning = new Map>(); + async function learnOnce(decision: Required, journaled: LearnLedger): Promise { + const { dbPath, mode } = await learnTarget(); + const id = learnedId(scope, decision); + const existing = await journaled.recorded(id); + if (existing !== undefined) { + await publishLearned(dbPath, existing, mode); + return; + } + const record: LearnRecord = { memory: 'learn', id, scope, decision, learnedAt: new Date().toISOString() }; + let staged: StagedLearned; + try { + staged = await stageLearned(dbPath, { record, runId: journaled.runId }, mode); + } catch (error) { + throw new AuthoredFlowExecutionError('memory_unwritable', `f.memory.learn could not stage the finding: ${error instanceof Error ? error.message : String(error)}`); + } + try { + // The last point a body that already called done() can be refused: past + // here the finding is journaled and will be published. + assertOpen(); + const winner = await journaled.append(record); + try { + await staged.publish(winner); + } catch (error) { + throw new AuthoredFlowExecutionError('memory_unwritable', `f.memory.learn journaled the finding (root run ${winner.runId}, stream ${MEMORY_LEARN_STREAM}, id ${id}) but could not publish it: ${error instanceof Error ? error.message : String(error)}`); + } + } finally { + await staged.discard(); + } + } + async function learn(finding: MemoryFinding): Promise { + assertOpen(); + assertEnabled(); + const decision = normalizeFinding(finding); + if (ledger === undefined) { + throw new AuthoredFlowExecutionError('memory_unjournaled', 'f.memory.learn requires a durable root run to journal the finding; run the flow through flows run'); + } + const id = learnedId(scope, decision); + // Concurrent learns of one finding in this execution share one commit. + let pending = learning.get(id); + if (pending === undefined) { + pending = learnOnce(decision, ledger); + learning.set(id, pending); + void pending.then(() => learning.delete(id), () => learning.delete(id)); + } + await pending; } return { - recall: (query, options) => read(reader => reader.search(query, { ...options, project: scope })), - why: task => read(reader => { - const entry = reader.whyForTask(task); - return entry?.projectId === scope ? [entry] : []; + recall: (query, options) => read((reader, learned) => recallMerged(reader, learned, scope, query, options)), + why: task => read((reader, learned) => { + const synced = reader.whyForTask(task); + const candidates = [ + ...(synced?.projectId === scope ? [synced] : []), + ...learned.map(entry => entry.trajectory).filter(entry => trajectoryMatches(entry, task)), + ]; + const newest = candidates.sort((a, b) => b.timestampMs - a.timestampMs)[0]; + return newest === undefined ? [] : [newest]; }), - async learn() { - assertOpen(); - throw new AuthoredFlowExecutionError('unsupported_verb', 'f.memory.learn requires the follow-up journal-backed trajectory writer'); + learn(finding) { + const write = learn(finding); + return tracker === undefined ? write : tracker.track(write); + }, + }; +} + +function normalizeFinding(finding: MemoryFinding): Required { + const text = (value: unknown, field: string): string => { + if (typeof value !== 'string' || value.trim() === '') { + throw new AuthoredFlowExecutionError('memory_finding_invalid', `f.memory.learn requires a non-empty string ${field}`); + } + return value; + }; + const alternatives = finding?.alternatives ?? []; + if (!Array.isArray(alternatives) || alternatives.some(item => typeof item !== 'string')) { + throw new AuthoredFlowExecutionError('memory_finding_invalid', 'f.memory.learn alternatives must be an array of strings'); + } + return { + question: text(finding?.question, 'question'), + chosen: text(finding?.chosen, 'chosen'), + reasoning: text(finding?.reasoning, 'reasoning'), + alternatives: [...alternatives], + }; +} + +/** + * Learned findings live beside the ai-hist database, not beside the flow + * source: a flow deployed from a read-only checkout, package or image can + * still learn. The directory is keyed by the scope's hash, and each file + * still names the scope as its `projectId`. Reads here do not need + * `ai-hist sync`. For sync to index them too, add `/flows-memory` to + * `TRAJECTORY_ROOT` together with the roots it already scans: setting it + * replaces ai-hist's default `~/Projects` scan rather than adding to it. + */ +const learnedDir = (dbPath: string, scope: string): string => + join(dirname(dbPath), 'flows-memory', basename(scope), '.trajectories', 'compacted'); +const LEARNED_PREFIX = 'flows-memory-'; + +const learnedId = (scope: string, decision: Required): string => + `${LEARNED_PREFIX}${createHash('sha256').update(JSON.stringify([scope, decision])).digest('hex').slice(0, 32)}`; + +/** + * `learn` never reads the ai-hist database, so it does not load it: it needs + * only the directory it lives in and its mode. A missing database is the same + * refusal reads give. + */ +async function learnTarget(): Promise<{ dbPath: string; mode: number }> { + const { defaultDbPath } = await import('ai-hist'); + const dbPath = defaultDbPath(); + const info = await stat(dbPath).catch(() => undefined); + if (info === undefined || !info.isFile()) { + throw new AuthoredFlowExecutionError('memory_unreachable', `memory database ${dbPath} is missing or not a file`); + } + return { dbPath, mode: info.mode & 0o777 }; +} + +function learnedContent({ record, runId }: CommittedLearn): string { + return `${JSON.stringify({ + id: record.id, + version: 1, + type: 'compacted', + projectId: record.scope, + task: { title: record.decision.question, description: null }, + status: 'completed', + startedAt: record.learnedAt, + completedAt: record.learnedAt, + decisions: [record.decision], + retrospective: { summary: null, approach: null, learnings: [], confidence: null }, + journal: { runId, stream: MEMORY_LEARN_STREAM, id: record.id }, + }, null, 2)}\n`; +} + +interface StagedLearned { + /** Rename into place, rewriting first if the winning commit is not the one staged. */ + publish(winner: CommittedLearn): Promise; + discard(): Promise; +} + +/** + * Write a committed finding's compacted trajectory file under a temporary + * name. Its content is derived only from the commit (including when it was + * learned and which root run journaled it), so every publication of one + * commit writes the same bytes. The final name is a hash of scope and + * finding, so re-learning a finding atomically replaces one file. The file + * takes the database's mode, so a private history does not leak findings + * through a world-readable file. The ai-hist database is never modified. + */ +async function stageLearned(dbPath: string, staging: CommittedLearn, mode: number): Promise { + const file = join(learnedDir(dbPath, staging.record.scope), `${staging.record.id}.json`); + await mkdir(dirname(file), { recursive: true }); + const staged = `${file}.${process.pid}.${randomUUID()}.tmp`; + let content = learnedContent(staging); + const write = async () => { + await writeFile(staged, content, { mode }); + await chmod(staged, mode); + }; + try { + await write(); + } catch (error) { + await rm(staged, { force: true }); + throw error; + } + return { + async publish(winner) { + const winning = learnedContent(winner); + if (winning !== content) { + content = winning; + await write(); + } + await rename(staged, file); }, + discard: () => rm(staged, { force: true }), }; } + +async function publishLearned(dbPath: string, committed: CommittedLearn, mode: number): Promise { + let staged: StagedLearned; + try { + staged = await stageLearned(dbPath, committed, mode); + } catch (error) { + throw new AuthoredFlowExecutionError('memory_unwritable', `f.memory.learn could not publish journaled finding ${committed.record.id}: ${error instanceof Error ? error.message : String(error)}`); + } + try { + await staged.publish(committed); + } catch (error) { + throw new AuthoredFlowExecutionError('memory_unwritable', `f.memory.learn could not publish journaled finding ${committed.record.id}: ${error instanceof Error ? error.message : String(error)}`); + } finally { + await staged.discard(); + } +} + +interface LearnedTrajectory { + trajectory: TrajectoryEntry; + history: HistoryEntry; +} + +/** This scope's learned findings, read straight from disk so they need no `ai-hist sync`. */ +async function readLearned(dbPath: string, scope: string): Promise { + const dir = learnedDir(dbPath, scope); + let names: string[]; + try { names = await readdir(dir); } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return []; + throw error; + } + const learned = await Promise.all(names + .filter(name => name.startsWith(LEARNED_PREFIX) && name.endsWith('.json')) + .map(async name => { + const path = join(dir, name); + try { + return learnedEntry(scope, path, await readFile(path, 'utf8'), Math.trunc((await stat(path)).mtimeMs)); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return undefined; + // Files here are only ever published whole, so an unreadable one is + // real damage; recalling around it would silently drop journaled memory. + throw new AuthoredFlowExecutionError('memory_unreachable', `learned finding ${path} is unreadable: ${error instanceof Error ? error.message : String(error)}`); + } + })); + return learned.filter((entry): entry is LearnedTrajectory => entry !== undefined); +} + +function learnedEntry(scope: string, path: string, text: string, updatedMs: number): LearnedTrajectory { + const raw: unknown = JSON.parse(text); + if (typeof raw !== 'object' || raw === null || Array.isArray(raw)) throw new Error('not a trajectory object'); + const value = raw as Record; + const decisions = value.decisions; + if (typeof value.id !== 'string' || !Array.isArray(decisions)) throw new Error('missing id or decisions'); + const parsed = decisions.map(decision => { + const d = decision as Record | null; + if (typeof d !== 'object' || d === null || typeof d.question !== 'string' || typeof d.chosen !== 'string' + || typeof d.reasoning !== 'string' || !Array.isArray(d.alternatives) || d.alternatives.some(a => typeof a !== 'string')) { + throw new Error('malformed decision'); + } + return { question: d.question, chosen: d.chosen, reasoning: d.reasoning, alternatives: d.alternatives as string[] }; + }); + const task = (typeof value.task === 'object' && value.task !== null ? value.task : {}) as Record; + const title = typeof task.title === 'string' ? task.title : null; + const description = typeof task.description === 'string' ? task.description : null; + const id = value.id; + const completedAt = typeof value.completedAt === 'string' ? value.completedAt : null; + const timestampMs = completedAt !== null && Number.isFinite(Date.parse(completedAt)) ? Date.parse(completedAt) : updatedMs; + // Same field order ai-hist's buildSearchText uses, so matching agrees with a later sync. + const searchText = [id, scope, 'completed', title, description, + ...parsed.flatMap(d => [d.question, d.chosen, d.reasoning, ...d.alternatives])] + .filter((part): part is string => typeof part === 'string' && part.length > 0).join('\n'); + return { + trajectory: { + id, version: 1, personaId: null, projectId: scope, + task: { title, description }, + status: 'completed', startedAt: typeof value.startedAt === 'string' ? value.startedAt : null, completedAt, + decisions: parsed, retrospective: { summary: null, approach: null, learnings: [], confidence: null }, + searchText, path, updatedMs, timestampMs, + }, + // Not yet in the ai-hist database, so it has no row id; a negative id + // derived from the trajectory id marks it as read from disk. + history: { + id: -Number.parseInt(createHash('sha256').update(id).digest('hex').slice(0, 12), 16), + source: 'trajectory', sessionId: id, project: scope, prompt: searchText, timestampMs, gitBranch: null, + }, + }; +} + +const includesFolded = (haystack: string | null | undefined, needle: string): boolean => + (haystack ?? '').toLowerCase().includes(needle.toLowerCase()); + +function trajectoryMatches(entry: TrajectoryEntry, query: string): boolean { + const needle = query.trim(); + if (needle === '') return false; + return [entry.searchText, entry.task.title, entry.task.description, entry.personaId, entry.projectId] + .some(field => includesFolded(field, needle)); +} + +/** ai-hist's `search` semantics over the database plus this scope's not-yet-synced findings. */ +function recallMerged( + reader: import('ai-hist').AiHist, + learned: LearnedTrajectory[], + scope: string, + query: string, + options: MemoryRecallOptions | undefined, +): HistoryEntry[] { + const limit = options?.limit ?? 50; + const needle = query.trim(); + const fromDisk = learned.map(entry => entry.history).filter(entry => + (needle === '' || includesFolded(entry.prompt, needle) || includesFolded(entry.project, needle)) + && (options?.source === undefined || options.source === 'trajectory') + && options?.tag === undefined + && (options?.beforeMs === undefined || entry.timestampMs < options.beforeMs)); + // Fetching `limit + fromDisk.length` rows guarantees a synced copy of any + // disk entry that makes the cut is among them, so it dedupes correctly. + const synced = reader.search(query, { ...options, project: scope, limit: limit + fromDisk.length }); + const syncedIds = new Set(synced.filter(entry => entry.source === 'trajectory').map(entry => entry.sessionId)); + return [...synced, ...fromDisk.filter(entry => !syncedIds.has(entry.sessionId))] + .sort((a, b) => b.timestampMs - a.timestampMs) + .slice(0, limit); +} diff --git a/packages/sdk/src/cli/direct-run.ts b/packages/sdk/src/cli/direct-run.ts index cca83dedd..e419d244d 100644 --- a/packages/sdk/src/cli/direct-run.ts +++ b/packages/sdk/src/cli/direct-run.ts @@ -27,6 +27,7 @@ import { connect, emptyReport, fromCheckReport, + memoryWriteFailure, protocolFailure, suspendedExecution, socketFor, @@ -219,6 +220,10 @@ export async function runDirectFlow( if (error instanceof AuthoredFlowExecutionError && error.code === 'result_unreadable') { return completedResultUnreadableReport('run', base, socketPath, error); } + if (error instanceof AuthoredFlowExecutionError) { + const memoryFailure = memoryWriteFailure('run', base, socketPath, error, base.rootRunId); + if (memoryFailure !== undefined) return memoryFailure; + } if (isReadInterruption(error)) return daemonUnresponsiveReport('run', base, socketPath, error, base.rootRunId, options, dataDir); if (error instanceof AuthoredFlowExecutionError && error.code === 'root_lease_lost') { return rootLeaseLostReport('run', base, socketPath, error); diff --git a/packages/sdk/src/cli/run.ts b/packages/sdk/src/cli/run.ts index 0a76bc242..2a1348b40 100644 --- a/packages/sdk/src/cli/run.ts +++ b/packages/sdk/src/cli/run.ts @@ -418,6 +418,10 @@ export async function resumeFlow( if (error instanceof AuthoredFlowExecutionError && error.code === 'result_unreadable') { return completedResultUnreadableReport('resume', base, socketPath, error); } + if (error instanceof AuthoredFlowExecutionError) { + const memoryFailure = memoryWriteFailure('resume', base, socketPath, error, runId); + if (memoryFailure !== undefined) return memoryFailure; + } if (isReadInterruption(error)) return daemonUnresponsiveReport('resume', base, socketPath, error, runId, options, dataDir); if (error instanceof AuthoredFlowExecutionError && error.code === 'root_lease_lost') { return rootLeaseLostReport('resume', base, socketPath, error, runId); @@ -1098,6 +1102,41 @@ async function inspectOutOfBandStep( const EXPIRED_LEASE_POLL_MS = 250; +/** + * `f.memory` failures keep their own kind instead of collapsing into + * `protocol_error`. By then the root has been terminalized as failed, so the + * report names it the way a step failure does. + */ +export function memoryWriteFailure( + command: RunCommand, + base: CheckReport | RunReport, + socketPath: string, + error: AuthoredFlowExecutionError, + fallbackRunId?: string, +): RunExecution | undefined { + if (error.code !== 'memory_finding_invalid' && error.code !== 'memory_unwritable' + && error.code !== 'memory_unjournaled' && error.code !== 'memory_unreachable') return undefined; + const rootRunId = error.rootRunId ?? fallbackRunId; + const runId = error.runId ?? rootRunId; + return { + exitCode: 1, + report: { + ...fromBase(command, base), + ok: false, + ...(runId === undefined ? {} : { runId }), + ...(rootRunId === undefined ? {} : { rootRunId }), + socketPath, + status: 'failed', + completionReason: 'step_failed', + diagnostics: [...base.diagnostics, { + severity: 'failure', kind: error.code, + // The `: ` prefix `AuthoredFlowExecutionError` adds is redundant once the diagnostic is labelled. + message: error.message.startsWith(`${error.code}: `) ? error.message.slice(error.code.length + 2) : error.message, + }], + }, + }; +} + export function protocolFailure( command: RunCommand, base: CheckReport | RunReport, diff --git a/packages/sdk/src/failure-kinds.ts b/packages/sdk/src/failure-kinds.ts index bb1e23cd7..9c0259590 100644 --- a/packages/sdk/src/failure-kinds.ts +++ b/packages/sdk/src/failure-kinds.ts @@ -138,6 +138,14 @@ export const RUN_FAILURE_KINDS = [ 'gate_failed', /** `flows answer` named a wait the run is not asking: unknown, or already answered. */ 'human_wait_unknown', + /** `f.memory.learn` was given a finding without non-empty string fields. */ + 'memory_finding_invalid', + /** `f.memory.learn` could not persist the finding to script memory. */ + 'memory_unwritable', + /** `f.memory.learn` ran without a durable root run to journal the finding in. */ + 'memory_unjournaled', + /** `f.memory` lost its database, or found a damaged learned finding, after the run started. */ + 'memory_unreachable', /** * `--local-agent` cannot be honoured for this invocation, so it is refused * rather than accepted and ignored. A local agent worker is admitted at run diff --git a/packages/sdk/tests/f-memory.test.ts b/packages/sdk/tests/f-memory.test.ts index 1c57e3df4..4d15431a2 100644 --- a/packages/sdk/tests/f-memory.test.ts +++ b/packages/sdk/tests/f-memory.test.ts @@ -1,13 +1,29 @@ import { execFileSync } from 'node:child_process'; -import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { createHash } from 'node:crypto'; +import { chmodSync, existsSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs'; +import { createRequire } from 'node:module'; import { tmpdir } from 'node:os'; -import { join } from 'node:path'; +import { basename, join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { flow } from '@relayflows/surface'; import { afterEach, beforeEach, expect, it, vi } from 'vitest'; + +let failRename: ((to: string) => boolean) | undefined; +vi.mock('node:fs/promises', async importOriginal => { + const actual = await importOriginal(); + return { + ...actual, + rename: async (from: string, to: string) => { + if (failRename?.(to)) throw Object.assign(new Error('injected rename failure'), { code: 'EIO' }); + return actual.rename(from, to); + }, + }; +}); import { openAiHist } from 'ai-hist'; -import { authoredMemory, scriptMemoryScope } from '../src/authored-memory.js'; +import { authoredMemory, journalLearnLedger, MEMORY_LEARN_STREAM, scriptMemoryScope } from '../src/authored-memory.js'; import { executeAuthoredFlow } from '../src/authored-flow-executor.js'; +import { AuthoredFlowExecutionError } from '../src/authored-flow-error.js'; +import { emptyReport, memoryWriteFailure } from '../src/cli/run.js'; import { JournalClient } from '../src/journal-client.js'; import { preflightMemory } from '../src/preflight.js'; @@ -23,10 +39,43 @@ beforeEach(() => { }); afterEach(() => { vi.unstubAllEnvs(); rmSync(dir, { recursive: true, force: true }); }); +/** The daemon's stream storage, with the kernel's paging contract (`engine/remote.rs` `read_stream`). */ +function streamStore() { + const stored = new Map(); + const key = (runId: string, stream: string): string => `${runId}\u0000${stream}`; + const appended: Array<{ stream: string; message: unknown; fileExisted: boolean }> = []; + return { + stored, appended, + list: (runId: string, stream: string) => stored.get(key(runId, stream)) ?? [], + async streamAppend(runId: string, stream: string, message: unknown) { + const list = stored.get(key(runId, stream)) ?? []; + stored.set(key(runId, stream), list); + appended.push({ stream, message, fileExisted: existsSync(compactedDir()) && readdirSync(compactedDir()).some(name => name.endsWith('.json')) }); + // Stored the way the journal stores it: keys sorted. + list.push(JSON.parse(JSON.stringify(message, Object.keys(flatKeys(message)).sort()))); + return { offset: list.length - 1 }; + }, + async streamRead(runId: string, stream: string, fromOffset: number, limit: number) { + const messages = (stored.get(key(runId, stream)) ?? []).slice(fromOffset, fromOffset + limit); + return { messages, next_offset: fromOffset + messages.length }; + }, + }; +} +function flatKeys(value: unknown, keys: Record = {}): Record { + if (typeof value === 'object' && value !== null) { + for (const [k, v] of Object.entries(value)) { keys[k] = true; flatKeys(v, keys); } + } + return keys; +} +let store: ReturnType; +beforeEach(() => { store = streamStore(); }); +const memoryFor = (forScope: string) => authoredMemory(forScope, () => {}, true, undefined, journalLearnLedger(store, 'root-1')); +const compactedDir = (forScope = scope) => join(dir, 'flows-memory', basename(forScope), '.trajectories', 'compacted'); + it('reads the seeded local SQLite database with cloud unused and fallback disabled', async () => { const db = await openAiHist({ dbPath, projectScope: scope, fallback: 'error' }); try { expect(db.search('retry safely')).toHaveLength(1); } finally { db.close(); } - const memory = authoredMemory(scope, () => {}, true); + const memory = memoryFor(scope); expect(await memory.recall('retry safely')).toMatchObject([{ id: 1, project: scope }]); expect(await memory.why('retry safely')).toMatchObject([{ id: 'run-1', decisions: [{ chosen: 'idempotency key', reasoning: 'avoid duplicate writes' }], @@ -36,9 +85,9 @@ it('reads the seeded local SQLite database with cloud unused and fallback disabl }); it('cannot widen script scope using a raw project option or another flow name', async () => { - const memory = authoredMemory(scope, () => {}, true); + const memory = memoryFor(scope); expect(await memory.recall('', { project: `${scope}-other` } as never)).toMatchObject([{ id: 1 }]); - const other = authoredMemory(scriptMemoryScope(join(dir, 'test.flow.ts'), 'another-flow'), () => {}, true); + const other = memoryFor(scriptMemoryScope(join(dir, 'test.flow.ts'), 'another-flow')); expect(await other.recall('retry safely')).toEqual([]); expect(await other.why('retry safely')).toEqual([]); }); @@ -94,11 +143,324 @@ it('maps an unavailable provider probe to memory_unreachable', async () => { .toMatchObject({ severity: 'refusal', kind: 'memory_unreachable' }); }); -it('fails closed for deferred writes, agent scope, disabled script memory, and reads after done', async () => { - await expect(authoredMemory(scope, () => {}, true).learn({ question: 'q', chosen: 'a', reasoning: 'r' })) - .rejects.toMatchObject({ code: 'unsupported_verb' }); +function completingJournal(): JournalClient { + const journal = new JournalClient('/unused'); + vi.spyOn(journal, 'streamAppend').mockImplementation((runId, stream, message) => store.streamAppend(runId, stream, message)); + vi.spyOn(journal, 'streamRead').mockImplementation((runId, stream, from, limit) => store.streamRead(runId, stream, from, limit ?? 100)); + vi.spyOn(journal, 'runStart').mockResolvedValue({ + run_id: 'completion', status: 'completed', completion_reason: 'success', completed_steps: 1, + }); + vi.spyOn(journal, 'journalRead').mockResolvedValue({ entries: [{ + entry_type: 'step.completed', step_id: 'complete-1', payload: { + completionReason: 'success', disposition: 'step_done', + output: { exit_code: 0, stdout_tail: '', stderr_tail: '' }, + }, + }] } as never); + return journal; +} + +it('persists a learned finding that a later run of the same flow recalls', async () => { + const flowPath = join(dir, 'test.flow.ts'); + const finding = { + question: 'how should the webhook retry', chosen: 'exponential backoff', + reasoning: 'the provider rate-limits bursts', alternatives: ['fixed delay'], + }; + await executeAuthoredFlow(flow('memory-example', { memory: { script: true } }, async f => { + expect(await f.memory.why('webhook retry')).toEqual([]); + await f.memory.learn(finding); + f.done('success'); + }), completingJournal(), undefined, { flowPath, rootRunId: 'root-1' }); + + let recalled: unknown[] = []; + let why: unknown[] = []; + await executeAuthoredFlow(flow('memory-example', { memory: { script: true } }, async f => { + recalled = await f.memory.recall('exponential backoff'); + why = await f.memory.why('webhook retry'); + f.done('success'); + }), completingJournal(), undefined, { flowPath, rootRunId: 'root-1' }); + expect(recalled).toMatchObject([{ source: 'trajectory', project: scope }]); + expect(why).toMatchObject([{ projectId: scope, decisions: [finding] }]); + + // The trajectory file is what a later `ai-hist sync` re-ingests. + const compacted = compactedDir(); + const [file] = readdirSync(compacted); + expect(JSON.parse(readFileSync(join(compacted, file!), 'utf8'))).toMatchObject({ projectId: scope, decisions: [finding] }); +}); + +it('learns idempotently and only into its own script scope', async () => { + const memory = memoryFor(scope); + const finding = { question: 'cache invalidation', chosen: 'ttl', reasoning: 'simple' }; + await memory.learn(finding); + await memory.learn(finding); + expect(await memory.recall('cache invalidation')).toHaveLength(1); + expect(readdirSync(compactedDir())).toHaveLength(1); + expect(await memory.recall('retry safely')).toHaveLength(1); + + const other = memoryFor(scriptMemoryScope(join(dir, 'test.flow.ts'), 'another-flow')); + expect(await other.recall('cache invalidation')).toEqual([]); + expect(await other.why('cache invalidation')).toEqual([]); +}); + +it('rejects invalid findings before writing, and unwritable stores', async () => { + const memory = memoryFor(scope); + await expect(memory.learn({ question: '', chosen: 'a', reasoning: 'r' })) + .rejects.toMatchObject({ code: 'memory_finding_invalid' }); + await expect(memory.learn({ question: 'q', chosen: 'a', reasoning: 'r', alternatives: [1] } as never)) + .rejects.toMatchObject({ code: 'memory_finding_invalid' }); + expect(existsSync(join(dir, 'flows-memory'))).toBe(false); + if (process.getuid?.() !== 0) { + chmodSync(dir, 0o500); + try { + await expect(memory.learn({ question: 'q', chosen: 'a', reasoning: 'r' })) + .rejects.toMatchObject({ code: 'memory_unwritable' }); + } finally { chmodSync(dir, 0o700); } + } +}); + +const dbDigest = () => createHash('sha256').update(readFileSync(dbPath)).digest('hex'); + +it('keeps concurrent findings without touching the shared database', async () => { + const before = dbDigest(); + const memory = memoryFor(scope); + await Promise.all(Array.from({ length: 6 }, (_, i) => + memory.learn({ question: `parallel finding ${i}`, chosen: 'c', reasoning: 'r' }))); + expect(await memory.recall('parallel finding')).toHaveLength(6); + expect(dbDigest()).toBe(before); + expect(readdirSync(compactedDir()).filter(name => !name.endsWith('.json'))).toEqual([]); +}); + +it('dedupes a finding once ai-hist sync has indexed it, preferring the synced row', async () => { + const memory = memoryFor(scope); + await memory.learn({ question: 'synced finding', chosen: 'c', reasoning: 'r' }); + const [diskEntry] = await memory.recall('synced finding'); + expect(diskEntry).toMatchObject({ source: 'trajectory', project: scope }); + expect(diskEntry!.id).toBeLessThan(0); + + // What `ai-hist sync` writes for the trajectory file. + const [name] = readdirSync(compactedDir()); + const raw = JSON.parse(readFileSync(join(compactedDir(), name!), 'utf8')); + const require = createRequire(import.meta.url); + const SQL = await createRequire(require.resolve('ai-hist/package.json'))('sql.js')(); + const db = new SQL.Database(readFileSync(dbPath)); + const ts = Date.parse(raw.completedAt); + db.run('INSERT INTO trajectories VALUES (?, 1, NULL, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?)', [ + raw.id, scope, raw.task.title, 'completed', raw.startedAt, raw.completedAt, + JSON.stringify(raw.decisions), JSON.stringify(raw.retrospective), diskEntry!.prompt, join(compactedDir(), name!), ts, ts]); + db.run('INSERT INTO history VALUES (?, ?, ?, ?, ?, ?, ?)', [99, 'trajectory', raw.id, scope, diskEntry!.prompt, ts, null]); + writeFileSync(dbPath, Buffer.from(db.export())); + db.close(); + + expect(await memory.recall('synced finding')).toMatchObject([{ id: 99, sessionId: raw.id }]); + expect(await memory.why('synced finding')).toMatchObject([{ id: raw.id }]); +}); + +it('applies recall options to findings not yet synced', async () => { + const memory = memoryFor(scope); + await memory.learn({ question: 'option finding', chosen: 'c', reasoning: 'r' }); + expect(await memory.recall('option finding', { source: 'claude' })).toEqual([]); + expect(await memory.recall('option finding', { tag: 'anything' })).toEqual([]); + expect(await memory.recall('option finding', { beforeMs: 1 })).toEqual([]); + expect(await memory.recall('', { limit: 1 })).toHaveLength(1); + expect(await memory.recall('')).toHaveLength(2); +}); + +it.skipIf(process.platform === 'win32')('gives the trajectory file the database file mode', async () => { + chmodSync(dbPath, 0o600); + await memoryFor(scope).learn({ question: 'private', chosen: 'c', reasoning: 'r' }); + expect(statSync(join(compactedDir(), readdirSync(compactedDir())[0]!)).mode & 0o777).toBe(0o600); +}); + +it('leaves nothing behind when the trajectory cannot be published', async () => { + const memory = memoryFor(scope); + failRename = to => to.startsWith(compactedDir()) && to.endsWith('.json'); + try { + await expect(memory.learn({ question: 'doomed', chosen: 'c', reasoning: 'r' })) + .rejects.toMatchObject({ code: 'memory_unwritable' }); + } finally { failRename = undefined; } + expect(readdirSync(compactedDir())).toEqual([]); + expect(await memory.recall('doomed')).toEqual([]); +}); + +it('fails the run when the body does not await learn', async () => { + await expect(executeAuthoredFlow(flow('memory-example', { memory: { script: true } }, async f => { + void f.memory.learn({ question: 'unawaited', chosen: 'c', reasoning: 'r' }); + f.done('success'); + }), completingJournal(), undefined, { flowPath: join(dir, 'test.flow.ts'), rootRunId: 'root-1' })) + .rejects.toMatchObject({ code: 'unawaited_step' }); +}); + +it('fails the run for an unawaited learn that already failed before the body returned', async () => { + await expect(executeAuthoredFlow(flow('memory-example', { memory: { script: true } }, async f => { + void f.memory.learn({ question: '', chosen: 'c', reasoning: 'r' }); + await new Promise(resolve => setTimeout(resolve, 50)); + f.done('success'); + }), completingJournal(), undefined, { flowPath: join(dir, 'test.flow.ts'), rootRunId: 'root-1' })) + .rejects.toMatchObject({ code: 'unawaited_step' }); +}); + +it('accepts a learn failure the body awaited and handled', async () => { + const result = await executeAuthoredFlow(flow('memory-example', { memory: { script: true } }, async f => { + await f.memory.learn({ question: '', chosen: 'c', reasoning: 'r' }).catch(() => undefined); + await Promise.all([f.memory.learn({ question: 'all', chosen: 'c', reasoning: 'r' })]); + f.done('success'); + }), completingJournal(), undefined, { flowPath: join(dir, 'test.flow.ts'), rootRunId: 'root-1' }); + expect(result.completionReason).toBe('success'); +}); + +it('journals the finding on the root run before it becomes recallable, and cites the commit', async () => { + const finding = { question: 'journal first', chosen: 'c', reasoning: 'r', alternatives: ['x'] }; + await memoryFor(scope).learn(finding); + expect(store.appended).toEqual([{ stream: MEMORY_LEARN_STREAM, message: expect.objectContaining({ + memory: 'learn', scope, decision: finding, + }), fileExisted: false }]); + const [name] = readdirSync(compactedDir()); + expect(JSON.parse(readFileSync(join(compactedDir(), name!), 'utf8'))) + .toMatchObject({ journal: { runId: 'root-1', stream: MEMORY_LEARN_STREAM, id: expect.stringMatching(/^flows-memory-/) }, decisions: [finding] }); +}); + +it('refuses to learn without a durable root run, writing nothing', async () => { + await expect(authoredMemory(scope, () => {}, true).learn({ question: 'q', chosen: 'c', reasoning: 'r' })) + .rejects.toMatchObject({ code: 'memory_unjournaled' }); + await expect(executeAuthoredFlow(flow('memory-example', { memory: { script: true } }, async f => { + await f.memory.learn({ question: 'q', chosen: 'c', reasoning: 'r' }); + f.done('success'); + }), completingJournal(), undefined, { flowPath: join(dir, 'test.flow.ts') })) + .rejects.toMatchObject({ code: 'memory_unjournaled' }); + expect(existsSync(join(dir, 'flows-memory'))).toBe(false); +}); + +it('publishes nothing when the journal append fails, and passes the journal error through unchanged', async () => { + const journalDown = new Error('journal down'); + vi.spyOn(store, 'streamAppend').mockRejectedValueOnce(journalDown); + const memory = memoryFor(scope); + await expect(memory.learn({ question: 'not journaled', chosen: 'c', reasoning: 'r' })).rejects.toBe(journalDown); + expect(readdirSync(compactedDir())).toEqual([]); + expect(await memory.recall('not journaled')).toEqual([]); +}); + +it('reads the stream again after a failed read instead of caching the failure', async () => { + const interrupted = new Error('read interrupted'); + vi.spyOn(store, 'streamRead').mockRejectedValueOnce(interrupted); + const memory = memoryFor(scope); + const finding = { question: 'retry after blip', chosen: 'c', reasoning: 'r' }; + await expect(memory.learn(finding)).rejects.toBe(interrupted); + await memory.learn(finding); + expect(store.list('root-1', MEMORY_LEARN_STREAM)).toHaveLength(1); + expect(await memory.recall('retry after blip')).toHaveLength(1); +}); + +it('publishes the first commit when another attempt on the root committed the same finding first', async () => { + const realAppend = store.streamAppend.bind(store); + vi.spyOn(store, 'streamAppend').mockImplementationOnce(async (runId, stream, message) => { + // A re-driven attempt's append lands first, with its own learnedAt. + await realAppend(runId, stream, { ...(message as object), learnedAt: '2020-01-01T00:00:00.000Z' }); + return realAppend(runId, stream, message); + }); + await memoryFor(scope).learn({ question: 'raced finding', chosen: 'c', reasoning: 'r' }); + const [name] = readdirSync(compactedDir()); + expect(JSON.parse(readFileSync(join(compactedDir(), name!), 'utf8'))).toMatchObject({ completedAt: '2020-01-01T00:00:00.000Z' }); +}); + +it('lets an in-flight learn land before a body without done() hands the root back', async () => { + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const realAppend = store.streamAppend.bind(store); + vi.spyOn(store, 'streamAppend').mockImplementation(async (runId, stream, message) => { + if (stream === MEMORY_LEARN_STREAM) await gate; + return realAppend(runId, stream, message); + }); + const incomplete = executeAuthoredFlow(flow('memory-example', { memory: { script: true } }, async f => { + void f.memory.learn({ question: 'no done', chosen: 'c', reasoning: 'r' }); + await new Promise(resolve => setTimeout(resolve, 20)); + }), completingJournal(), undefined, { flowPath: join(dir, 'test.flow.ts'), rootRunId: 'root-1' }); + let settled = false; + void incomplete.catch(() => { settled = true; }); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(settled).toBe(false); + release(); + await expect(incomplete).rejects.toMatchObject({ code: 'missing_completion' }); + expect(store.list('root-1', MEMORY_LEARN_STREAM)).toHaveLength(1); +}); + +it('lets an in-flight learn land before a failed body hands the root back', async () => { + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const realAppend = store.streamAppend.bind(store); + vi.spyOn(store, 'streamAppend').mockImplementation(async (runId, stream, message) => { + if (stream === MEMORY_LEARN_STREAM) await gate; + return realAppend(runId, stream, message); + }); + const failing = executeAuthoredFlow(flow('memory-example', { memory: { script: true } }, async f => { + void f.memory.learn({ question: 'in flight', chosen: 'c', reasoning: 'r' }); + await new Promise(resolve => setTimeout(resolve, 20)); + throw new Error('body failed'); + }), completingJournal(), undefined, { flowPath: join(dir, 'test.flow.ts'), rootRunId: 'root-1' }); + let settled = false; + void failing.catch(() => { settled = true; }); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(settled).toBe(false); + release(); + await expect(failing).rejects.toThrow('body failed'); + expect(store.list('root-1', MEMORY_LEARN_STREAM)).toHaveLength(1); + expect(readdirSync(compactedDir()).filter(name => name.endsWith('.json'))).toHaveLength(1); +}); + +it('on resume, republishes a journaled finding from its commit without journaling it again', async () => { + const finding = { question: 'resumed finding', chosen: 'c', reasoning: 'r', alternatives: [] }; + await memoryFor(scope).learn(finding); + const [name] = readdirSync(compactedDir()); + const published = readFileSync(join(compactedDir(), name!), 'utf8'); + // The crash window: journaled, but the file never landed. + rmSync(join(compactedDir(), name!)); + // A resumed body is a new execution: a fresh ledger reading the same root stream. + const resumed = memoryFor(scope); + expect(await resumed.recall('resumed finding')).toEqual([]); + await resumed.learn(finding); + expect(store.list('root-1', MEMORY_LEARN_STREAM)).toHaveLength(1); + expect(readFileSync(join(compactedDir(), name!), 'utf8')).toBe(published); + expect(await resumed.recall('resumed finding')).toHaveLength(1); +}); + +it('journals concurrent learns of one finding once', async () => { + const memory = memoryFor(scope); + const finding = { question: 'same finding', chosen: 'c', reasoning: 'r' }; + await Promise.all([memory.learn(finding), memory.learn(finding), memory.learn(finding)]); + expect(store.list('root-1', MEMORY_LEARN_STREAM)).toHaveLength(1); +}); + +it('fails recall closed on a damaged learned file instead of skipping it', async () => { + const memory = memoryFor(scope); + await memory.learn({ question: 'damaged', chosen: 'c', reasoning: 'r' }); + const [name] = readdirSync(compactedDir()); + for (const damage of ['{ not json', 'null', '[]', '{"id":"x","decisions":[null]}']) { + writeFileSync(join(compactedDir(), name!), damage); + await expect(memory.recall('damaged')).rejects.toMatchObject({ code: 'memory_unreachable' }); + await expect(memory.why('damaged')).rejects.toMatchObject({ code: 'memory_unreachable' }); + } +}); + +it('keeps learned files out of the flow source directory', async () => { + await memoryFor(scope).learn({ question: 'placement', chosen: 'c', reasoning: 'r' }); + expect(existsSync(scope)).toBe(false); + expect(readdirSync(compactedDir())).toHaveLength(1); +}); + +it('reports learn failures with their own kind, not protocol_error', () => { + for (const code of ['memory_finding_invalid', 'memory_unwritable', 'memory_unjournaled', 'memory_unreachable'] as const) { + const error = Object.assign(new AuthoredFlowExecutionError(code, 'nope'), { rootRunId: 'root-1' }); + expect(memoryWriteFailure('run', emptyReport('run'), '/sock', error)).toMatchObject({ exitCode: 1, report: { + runId: 'root-1', rootRunId: 'root-1', status: 'failed', completionReason: 'step_failed', + diagnostics: [{ severity: 'failure', kind: code, message: 'nope' }], + } }); + } + expect(memoryWriteFailure('run', emptyReport('run'), '/sock', new AuthoredFlowExecutionError('step_failed', 'x'))).toBeUndefined(); +}); + +it('fails closed for agent scope, disabled script memory, and memory use after done', async () => { await expect(authoredMemory(scope, () => {}, false).recall('q')) .rejects.toMatchObject({ code: 'unsupported_header' }); + await expect(authoredMemory(scope, () => {}, false).learn({ question: 'q', chosen: 'a', reasoning: 'r' })) + .rejects.toMatchObject({ code: 'unsupported_header' }); await expect(executeAuthoredFlow(flow('agent', { memory: { agent: true } }, async f => f.done('success')), new JournalClient('/unused'))) .rejects.toMatchObject({ code: 'unsupported_header' }); await expect(executeAuthoredFlow(flow('closed', async f => { diff --git a/packages/sdk/tests/memory-learn-live.test.ts b/packages/sdk/tests/memory-learn-live.test.ts new file mode 100644 index 000000000..482f9e710 --- /dev/null +++ b/packages/sdk/tests/memory-learn-live.test.ts @@ -0,0 +1,62 @@ +import { execFileSync } from 'node:child_process'; +import { existsSync, readdirSync, readFileSync, writeFileSync } from 'node:fs'; +import { basename, join, resolve } from 'node:path'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { MEMORY_LEARN_STREAM, scriptMemoryScope } from '../src/authored-memory.js'; +import { JournalClient } from '../src/journal-client.js'; +import { socketPathFor } from '../src/daemon-connection.js'; +import { chainFixture } from './flow-chain-fixture.js'; + +const closes: Array<() => Promise> = []; +afterEach(async () => { + for (const close of closes.splice(0).reverse()) await close(); + vi.unstubAllEnvs(); +}); + +describe('f.memory.learn through the built CLI and a real kernel', () => { + it('journals the finding on the root run, then a later run of the same flow recalls it', async () => { + const fixture = chainFixture(); + closes.push(() => fixture.close()); + const dbPath = join(fixture.root, 'ai-history.db'); + const scope = scriptMemoryScope(fixture.flowPath, 'learner'); + execFileSync(process.execPath, [resolve('../../testdata/memory/seed.mjs'), dbPath, scope]); + vi.stubEnv('AI_HIST_DB', dbPath); + + writeFileSync(fixture.flowPath, `import { flow } from '@relayflows/surface'; +export default flow('learner', { memory: { script: true } }, async f => { + await f.memory.learn({ question: 'how should the importer page', chosen: 'cursor pagination', + reasoning: 'offsets skip rows under concurrent inserts', alternatives: ['offset pagination'] }); + f.done('success'); +}); +`); + const learned = fixture.invoke('run', '--no-observer-link', '--json', '--data-dir', fixture.data, fixture.flowPath, '--input', '{}'); + expect(learned.status, learned.stderr + learned.stdout).toBe(0); + const report = JSON.parse(learned.stdout) as { runId: string; completionReason: string }; + expect(report.completionReason).toBe('success'); + + const journal = new JournalClient(socketPathFor(fixture.data)); + await journal.connect(); + await journal.hello('memory-learn-live-test'); + const records = (await journal.streamRead(report.runId, MEMORY_LEARN_STREAM, 0, 1000)).messages + .map(message => (message as { message?: unknown }).message ?? message) as Array>; + journal.close(); + expect(records).toMatchObject([{ memory: 'learn', scope, decision: { chosen: 'cursor pagination' } }]); + + const compacted = join(fixture.root, 'flows-memory', basename(scope), '.trajectories', 'compacted'); + const [file] = readdirSync(compacted); + expect(JSON.parse(readFileSync(join(compacted, file!), 'utf8'))) + .toMatchObject({ id: records[0]!.id, journal: { runId: report.runId, stream: MEMORY_LEARN_STREAM, id: records[0]!.id } }); + expect(existsSync(scope)).toBe(false); + + writeFileSync(fixture.flowPath, `import { flow } from '@relayflows/surface'; +export default flow('learner', { memory: { script: true } }, async f => { + const [why] = await f.memory.why('importer page'); + const hits = await f.memory.recall('cursor pagination'); + f.done(why?.decisions[0]?.chosen === 'cursor pagination' && hits.length === 1 ? 'success' : 'step_failed'); +}); +`); + const recalled = fixture.invoke('run', '--no-observer-link', '--json', '--data-dir', fixture.data, fixture.flowPath, '--input', '{}'); + expect(recalled.status, recalled.stderr + recalled.stdout).toBe(0); + expect(JSON.parse(recalled.stdout)).toMatchObject({ completionReason: 'success' }); + }, 60_000); +}); diff --git a/packages/surface/AUTHORING.md b/packages/surface/AUTHORING.md index 2f342d720..0a3f3c820 100644 --- a/packages/surface/AUTHORING.md +++ b/packages/surface/AUTHORING.md @@ -211,6 +211,14 @@ export interface AgentOptions { * `transport: 'relay'`. See docs/SURFACE.md. */ cwd?: string; + /** + * Continue a CLI session instead of starting a new one: an earlier step's + * `AgentResult.sessionId`. Claude and Codex resume it from their own + * session store (`~/.claude`, `~/.codex`), so it works only where that + * store survives — such as a sandbox reused for the same pull request. + * Other CLIs refuse the step. Not supported with `transport: 'relay'`. + */ + resume?: string; /** * Dispatch transport (flows#385). `'direct'` (default) spawns the CLI as a * local subprocess. `'relay'` posts to agent-relay so the agent registers @@ -253,6 +261,12 @@ export interface AgentResult { * final message is a JSON object (that object is the output, unmodified). */ artifacts: string[]; + /** + * The CLI session this step ran as (Claude's session id, Codex's thread + * id), when the CLI reported one. Pass it as a later step's `resume` to + * continue the same conversation. + */ + sessionId?: string; } ``` @@ -1276,7 +1290,7 @@ export interface MemoryHelper { recall(query: string, options?: MemoryRecallOptions): Promise; /** Best matching decision trajectory, or an empty array. No journal step. */ why(task: string): Promise; - /** Reserved for the journal-backed write slice; currently refuses. */ + /** Journals the finding on the durable root run, then persists it to this flow's script memory; later runs see it via recall/why. Idempotent per finding. Must be awaited. */ learn(finding: MemoryFinding): Promise; } ``` diff --git a/packages/surface/src/memory.ts b/packages/surface/src/memory.ts index e6742806c..2924fba4e 100644 --- a/packages/surface/src/memory.ts +++ b/packages/surface/src/memory.ts @@ -15,6 +15,6 @@ export interface MemoryHelper { recall(query: string, options?: MemoryRecallOptions): Promise; /** Best matching decision trajectory, or an empty array. No journal step. */ why(task: string): Promise; - /** Reserved for the journal-backed write slice; currently refuses. */ + /** Journals the finding on the durable root run, then persists it to this flow's script memory; later runs see it via recall/why. Idempotent per finding. Must be awaited. */ learn(finding: MemoryFinding): Promise; } diff --git a/testdata/memory/README.md b/testdata/memory/README.md index edcec7dbe..ee653e24d 100644 --- a/testdata/memory/README.md +++ b/testdata/memory/README.md @@ -1,7 +1,16 @@ # Minimal local script-memory example -This slice implements `f.memory.recall` and `f.memory.why` through ai-hist -0.4.1. Reads have no journal steps. `memory: { script: true }` is already part +This slice implements `f.memory.recall`, `f.memory.why`, and `f.memory.learn` +through ai-hist 0.4.1. Reads journal nothing. `learn` stages a compacted +trajectory file, appends the finding to the durable root run's +`memory-learn` stream, then renames the file into place at +`/flows-memory//.trajectories/compacted/.json`; the +file cites the commit. ai-hist 0.4.1 has no trajectory writer and its database is +never modified; reads merge the database with the scope's learned files. A +resumed body re-running `learn` republishes from the commit instead of +journaling twice, and `learn` refuses (`memory_unjournaled`) without a root +run. `tests/memory-learn-live.test.ts` exercises it through the built CLI and +a real kernel. `memory: { script: true }` is already part of the surface header contract and now works in the internal authored executor. `seed.mjs` creates a new deterministic SQLite database with two script scopes, @@ -42,10 +51,6 @@ than ai-hist's automatic scan of unrelated local JSONL history. ## Deferred acceptance work -- `learn` and journal effect record/confirm idempotency. ai-hist 0.4.1 exposes - reads and tag writes, but no public trajectory writer. The authored executor - also lacks a durable root run ID. `learn` explicitly refuses until both seams - are available; no unjournaled write is substituted. - Identity-scoped agent context injection. `memory.agent: true` explicitly refuses; it must not silently read script scope. - CLI-only provider operation and automatic creation of a fresh database. @@ -54,5 +59,5 @@ than ai-hist's automatic scan of unrelated local JSONL history. - Automatic trajectory import and the behavioural Gate 5 acceptance example. - Cloud push and pair mode, per the original exclusions. -This is a read-only proof slice, not completion of Gate 5 or the full original +This is a local slice, not completion of Gate 5 or the full original recall/why/learn acceptance matrix. The lead owns PR creation and CI review.