From a3ee0141267498d0a204b00a0fe23d94588e0dab Mon Sep 17 00:00:00 2001 From: Tony Ketcham Date: Fri, 17 Jul 2026 16:46:24 -0700 Subject: [PATCH] feat(effort-graph,flatbread): committed-generation bridge completes the ADR-0003/0004 contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both generation tokens existed with no wire between them: the writer's journal generation advanced while LiveSchemaReloader.generation never moved, the injected indexer defaulted to a no-op, and the strict read-your-writes consumer did not exist. EffortGraphIndexer also name-collided with EffortGraphIndex (opposite roles, near-identical names). - Rename: EffortGraphIndexer → CommittedGenerationPublisher, ReindexRequest → CommittedGenerationPublication, option indexer → publisher. journal.ts is rename-only (9+/5-): zero changes to markers, fsync ordering, rollback, replay, or error handling — the crash-safety protocol and its suites are untouched. - New live.ts: createEffortGraphLiveBridge composes writer + publisher + strict reader. The publisher maps journal paths to absolute paths, awaits reloader.notifyChanged({ source: 'writer' }), and throws on a rejected generation — that throw is the entire publish gate: the journal's existing committed-but-unpublished recovery handles it. waitForCommittedGeneration(token) requires both durable generation.json publication and the mapped live schema commit (timeoutMs escape hatch; replaceConfig fallback covers cross-process tokens). - New journalBarrier.ts: journal-aware ReindexBarrier adapter for the watcher path — blocks paths named by uncommitted journal intents, releases on commit/rollback, fails closed on malformed intents, bounded wait (default 10s) so an orphaned txn cannot stall the reloader queue. - Composition root in flatbread's GraphQL server: structural detection of the full effort-graph preset shape (all six effortGraphContent entries) activates barrier + bridge + non-fatal boot recovery and exposes server.effortGraph; without the preset the server is byte-identical to before. Adds the honest workspace edge flatbread → @flatbread/effort-graph (lockfile regenerated). - ADR-0008 records the design, incl. the v1 posture that out-of-process writers get eventual (not strict) consistency; CONTEXT.md gains the "Committed generation" vocabulary entry. ADRs 0001–0007 unedited. Test plan: 17 new AVA tests — live-bridge contract (no publish before reindex commits; failed schema commit leaves the generation unpublished and recover() republishes; strict readers see the mutation), journal-barrier (defer/release/fail-closed/bounded), composition activation/inertness, and 3 end-to-end liveServer tests incl. a real GraphQL query of a mutated effort and boot recovery. All pre-existing suites pass unmodified. Full suite: pnpm verify (293 AVA + 52 vitest). Co-authored-by: Cursor Change-Id: Ia531b2eb6e08223957fc0f56b1b72e561d15661d --- docs/effort-graph/CONTEXT.md | 2 + .../adr/0008-committed-generation-bridge.md | 50 ++++ .../src/__tests__/journal-barrier.test.ts | 104 +++++++ .../src/__tests__/live-bridge.test.ts | 273 ++++++++++++++++++ packages/effort-graph/src/errors.ts | 33 +++ packages/effort-graph/src/index.ts | 2 + packages/effort-graph/src/journal.ts | 14 +- packages/effort-graph/src/journalBarrier.ts | 100 +++++++ packages/effort-graph/src/live.ts | 152 ++++++++++ packages/effort-graph/src/preset.ts | 55 ++++ packages/effort-graph/src/types.ts | 8 +- packages/effort-graph/src/writer.ts | 8 +- packages/flatbread/package.json | 1 + .../graphql/effortGraphComposition.test.ts | 67 +++++ .../src/graphql/effortGraphComposition.ts | 32 ++ packages/flatbread/src/graphql/liveServer.ts | 21 +- .../src/graphql/liveServerEffortGraph.test.ts | 140 +++++++++ pnpm-lock.yaml | 3 + 18 files changed, 1051 insertions(+), 14 deletions(-) create mode 100644 docs/effort-graph/adr/0008-committed-generation-bridge.md create mode 100644 packages/effort-graph/src/__tests__/journal-barrier.test.ts create mode 100644 packages/effort-graph/src/__tests__/live-bridge.test.ts create mode 100644 packages/effort-graph/src/journalBarrier.ts create mode 100644 packages/effort-graph/src/live.ts create mode 100644 packages/flatbread/src/graphql/effortGraphComposition.test.ts create mode 100644 packages/flatbread/src/graphql/effortGraphComposition.ts create mode 100644 packages/flatbread/src/graphql/liveServerEffortGraph.test.ts diff --git a/docs/effort-graph/CONTEXT.md b/docs/effort-graph/CONTEXT.md index 4b65c6eb..aee3aae5 100644 --- a/docs/effort-graph/CONTEXT.md +++ b/docs/effort-graph/CONTEXT.md @@ -8,6 +8,8 @@ The Effort Graph is **built on top of** Flatbread's content-layer vocabulary (se **Operational provenance** (which session produced this, which agent, which model, which DAG run) is captured as **frontmatter fields** on these primitives, not as peer collections. The durable transcript record lives next to the graph under `.flatbread/artifacts/` (see [`packages/proof/README.md`](../../packages/proof/README.md) §Artifact Output). +**Committed generation.** The opaque journal generation token returned by an Effort Graph mutation. It is published only after the full save group has produced a committed live schema (the `CommittedGenerationPublisher` seam — not to be confused with `EffortGraphIndex`, the plan-time read interface). It is a different counter from any process-local live-schema generation; the committed-generation bridge maps the former to the latter for strict readers. + --- ### Effort diff --git a/docs/effort-graph/adr/0008-committed-generation-bridge.md b/docs/effort-graph/adr/0008-committed-generation-bridge.md new file mode 100644 index 00000000..9007e4d3 --- /dev/null +++ b/docs/effort-graph/adr/0008-committed-generation-bridge.md @@ -0,0 +1,50 @@ +# ADR-0008: Committed-generation bridge + +Status: Accepted + +## Context + +ADR-0003 requires the writer to expose a monotonic generation token and an +opt-in strict read. ADR-0004 requires generation publication only after +reindex and live schema swap. Both halves existed (the journal protocol and +the live reloader), but the wire between them did not: `.journal/generation.json` +could advance while `LiveSchemaReloader.generation` never moved. Journal tokens +are durable per root; live generations are process-local across all content. +Equating their values is false after restart and whenever unrelated content +changes. + +## Decision + +The writer now uses `CommittedGenerationPublisher`, renamed from +`EffortGraphIndexer` to separate it from the plan-time `EffortGraphIndex`. +The live adapter maps relative paths to absolute paths, awaits +`notifyChanged({ source: 'writer' })`, and throws on rejected candidates. This +existing callback gate is the publish gate; the journal protocol is unchanged. +The bridge privately maps journal token J to live generation L, and strict +readers require both a live commit and durable publication, with an explicit +timeout escape hatch. + +A disk-backed, journal-aware `ReindexBarrier` defers watcher paths named by +uncommitted intents, releases on the committed marker or rollback removal, +never waits on the reloader or publisher, fails closed on malformed intents, +and bounds its wait so an orphaned transaction cannot stall the serialized +reindex queue forever. + +The Flatbread composition root activates on structural detection of the +complete six-entry `effortGraphContent(root)` shape (paths and refs), attaches +the bridge, attempts non-fatal boot recovery before listening, and exposes +`RunningGraphqlServer.effortGraph`. Flatbread takes a runtime workspace +dependency on effort-graph; effort-graph keeps core type-only, and core learns +no journal semantics. + +## Consequences + +A returned mutation token names a generation whose live schema commit already +completed, providing strict same-process read-your-writes. An out-of-process +writer publishes normally with the no-op publisher; the server watcher observes +its files after commit or rollback and those reads are EVENTUAL, not strict. +A dead external writer can defer intersecting watcher work until lease-safe +recovery; bounded barrier waits convert that to a logged rejected batch rather +than a stalled queue. The watcher may rebuild files the writer just published; +whole-file reads and serialization make that safe. This completes the +ADR-0003/0004 committed-generation contract. diff --git a/packages/effort-graph/src/__tests__/journal-barrier.test.ts b/packages/effort-graph/src/__tests__/journal-barrier.test.ts new file mode 100644 index 00000000..bc5547d0 --- /dev/null +++ b/packages/effort-graph/src/__tests__/journal-barrier.test.ts @@ -0,0 +1,104 @@ +import test from 'ava'; +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { createJournalReindexBarrier } from '../journalBarrier.js'; +import { + EffortGraphBarrierTimeoutError, + EffortGraphCorruptJournalError, +} from '../errors.js'; + +async function fixture() { + const root = await mkdtemp(join(tmpdir(), 'eg-barrier-')); + const txn = join(root, '.journal', 'txns', 'txn'); + await mkdir(txn, { recursive: true }); + await writeFile( + join(txn, 'intent.json'), + JSON.stringify({ + writes: [{ relativePath: 'efforts/a.md' }], + }) + ); + return { root, txn, path: join(root, 'efforts', 'a.md') }; +} + +test.serial( + 'defers watcher paths named by an uncommitted intent and releases on the committed marker', + async (t) => { + const { root, txn, path } = await fixture(); + const pending = createJournalReindexBarrier({ + rootDir: root, + pollIntervalMs: 5, + }).waitUntilReadable([path]); + await new Promise((resolve) => setTimeout(resolve, 20)); + let done = false; + void pending.then(() => (done = true)); + t.false(done); + await writeFile(join(txn, 'committed'), ''); + await pending; + t.pass(); + } +); + +test.serial( + 'releases after rollback removes the uncommitted transaction', + async (t) => { + const { root, txn, path } = await fixture(); + const pending = createJournalReindexBarrier({ + rootDir: root, + pollIntervalMs: 5, + }).waitUntilReadable([path]); + await rm(txn, { recursive: true, force: true }); + await pending; + t.pass(); + } +); + +test.serial( + 'does not defer unrelated paths or committed transactions', + async (t) => { + const { root, txn } = await fixture(); + await writeFile(join(txn, 'committed'), ''); + await createJournalReindexBarrier({ rootDir: root }).waitUntilReadable([ + join(root, 'other.md'), + ]); + t.pass(); + } +); + +test.serial( + 'maps relative intent paths against absolute watcher paths', + async (t) => { + const { root, path } = await fixture(); + const pending = createJournalReindexBarrier({ + rootDir: root, + pollIntervalMs: 5, + }).waitUntilReadable([path]); + await writeFile(join(root, '.journal', 'txns', 'txn', 'committed'), ''); + await pending; + t.pass(); + } +); + +test.serial('fails closed on a malformed active intent', async (t) => { + const { root, txn, path } = await fixture(); + await writeFile(join(txn, 'intent.json'), '{broken'); + await t.throwsAsync( + createJournalReindexBarrier({ rootDir: root }).waitUntilReadable([path]), + { instanceOf: EffortGraphCorruptJournalError } + ); +}); + +test.serial( + 'bounded wait rejects after maxWaitMs for an orphaned transaction', + async (t) => { + const { root, path } = await fixture(); + await t.throwsAsync( + createJournalReindexBarrier({ + rootDir: root, + pollIntervalMs: 2, + maxWaitMs: 10, + }).waitUntilReadable([path]), + { instanceOf: EffortGraphBarrierTimeoutError } + ); + } +); diff --git a/packages/effort-graph/src/__tests__/live-bridge.test.ts b/packages/effort-graph/src/__tests__/live-bridge.test.ts new file mode 100644 index 00000000..3285ce04 --- /dev/null +++ b/packages/effort-graph/src/__tests__/live-bridge.test.ts @@ -0,0 +1,273 @@ +import test from 'ava'; +import { createHash } from 'node:crypto'; +import { mkdir, mkdtemp, readdir, readFile, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { createEffortGraphLiveBridge } from '../live.js'; +import { + EffortGraphGenerationWaitTimeoutError, + EffortGraphReindexFailedError, +} from '../errors.js'; +import type { LiveSchemaReloader, SchemaSnapshot } from '@flatbread/core'; + +function fakeReloader(): { + reloader: LiveSchemaReloader; + release: () => void; + calls: Array<{ paths: readonly string[]; source?: string }>; + reject: { value: boolean }; +} { + let generation = 0; + let snapshot = {} as SchemaSnapshot; + let releaseGate: (() => void) | undefined; + const calls: Array<{ paths: readonly string[]; source?: string }> = []; + const reject = { value: false }; + const reloader = { + get generation() { + return generation; + }, + getSnapshot: () => snapshot, + notifyChanged: async (change: { + paths: readonly string[]; + source?: 'writer'; + }) => { + calls.push(change); + if (releaseGate) + await new Promise( + (resolve) => + (releaseGate = () => { + releaseGate = undefined; + resolve(); + }) + ); + if (reject.value) + return { + status: 'rejected' as const, + generation, + error: new Error('candidate rejected'), + }; + generation += 1; + snapshot = { ...snapshot, generation }; + return { status: 'committed' as const, generation }; + }, + replaceConfig: async () => { + generation += 1; + snapshot = { ...snapshot, generation }; + return { status: 'committed' as const, generation }; + }, + waitForGeneration: async () => snapshot, + } as LiveSchemaReloader; + return { + reloader, + release: () => releaseGate?.(), + calls, + reject, + }; +} + +async function makeBridge() { + const root = await mkdtemp(join(tmpdir(), 'eg-live-')); + const fake = fakeReloader(); + const bridge = await createEffortGraphLiveBridge({ + rootDir: root, + reloader: fake.reloader, + }); + return { root, bridge, fake }; +} + +test.serial('no journal publish before the live reindex commits', async (t) => { + const { root, bridge, fake } = await makeBridge(); + let release!: () => void; + const pendingGate = new Promise((resolve) => (release = resolve)); + fake.reloader.notifyChanged = async (change) => { + fake.calls.push(change); + await pendingGate; + return { status: 'committed', generation: 1 }; + }; + const pending = bridge.writer.mutate({ + type: 'CreateEffort', + title: 'E', + body: '', + }); + let txns: string[] = []; + for (let attempt = 0; attempt < 50; attempt++) { + await new Promise((resolve) => setTimeout(resolve, 10)); + try { + txns = await readdir(join(root, '.journal', 'txns')); + } catch (error) { + if ( + typeof error !== 'object' || + error === null || + !('code' in error) || + (error as { readonly code?: unknown }).code !== 'ENOENT' + ) + throw error; + continue; + } + if ( + txns[0] && + (await readFile(join(root, '.journal', 'txns', txns[0], 'committed')) + .then(() => true) + .catch(() => false)) + ) + break; + } + t.true(txns.length === 1); + t.true( + await readFile(join(root, '.journal', 'txns', txns[0], 'committed')).then( + () => true + ) + ); + t.false( + await readFile(join(root, '.journal', 'generation.json')) + .then(() => true) + .catch(() => false) + ); + release(); + await pending; + t.deepEqual(fake.calls[0].source, 'writer'); + t.true(fake.calls[0].paths[0].startsWith(root)); + t.is( + JSON.parse( + await readFile(join(root, '.journal', 'generation.json'), 'utf8') + ).generation, + 1 + ); +}); + +test.serial( + 'rejected live schema leaves the generation unpublished and recover() publishes after a later commit', + async (t) => { + const { root, bridge, fake } = await makeBridge(); + fake.reject.value = true; + await t.throwsAsync( + bridge.writer.mutate({ type: 'CreateEffort', title: 'E', body: '' }), + { instanceOf: EffortGraphReindexFailedError } + ); + const txns = await readdir(join(root, '.journal', 'txns')); + t.is(txns.length, 1); + t.false( + await readFile(join(root, '.journal', 'generation.json')) + .then(() => true) + .catch(() => false) + ); + fake.reject.value = false; + t.deepEqual(await bridge.writer.recover(), { + action: 'completed', + transactionId: txns[0], + }); + t.is( + JSON.parse( + await readFile(join(root, '.journal', 'generation.json'), 'utf8') + ).generation, + 1 + ); + } +); + +test.serial( + 'a strict reader holding the returned token sees the mutation', + async (t) => { + const { bridge, fake } = await makeBridge(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + fake.reloader.notifyChanged = async () => { + await gate; + return { status: 'committed', generation: 41 }; + }; + fake.reloader.waitForGeneration = async () => ({ + ...fake.reloader.getSnapshot(), + generation: 41, + }); + const waiting = bridge.waitForCommittedGeneration('1', { timeoutMs: 1000 }); + const mutation = bridge.writer.mutate({ + type: 'CreateEffort', + title: 'E', + body: '', + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + release(); + const result = await mutation; + t.is(result.generation, '1'); + t.is( + (await bridge.waitForCommittedGeneration(result.generation)).generation, + 41 + ); + t.is((await waiting).generation, 41); + } +); + +test.serial( + 'strict wait times out for a token that never publishes', + async (t) => { + const { bridge } = await makeBridge(); + await t.throwsAsync( + bridge.waitForCommittedGeneration('1', { timeoutMs: 10 }), + { + instanceOf: EffortGraphGenerationWaitTimeoutError, + } + ); + } +); + +test.serial( + 'published tokens remain strictly readable after bridge recreation', + async (t) => { + const { root, bridge, fake } = await makeBridge(); + const result = await bridge.writer.mutate({ + type: 'CreateEffort', + title: 'E', + body: '', + }); + const recreated = await createEffortGraphLiveBridge({ + rootDir: root, + reloader: fake.reloader, + }); + t.is( + await ( + await recreated.waitForCommittedGeneration(result.generation) + ).generation, + 1 + ); + } +); + +test.serial( + 'recovery replays the publisher idempotently after an unpublished schema success', + async (t) => { + const { root, bridge, fake } = await makeBridge(); + const body = Buffer.from('---\nid: x\n---\n'); + const td = join(root, '.journal', 'txns', 'seed'); + await mkdir(td, { recursive: true }); + await writeFile(join(root, 'efforts.md'), body); + await writeFile( + join(td, 'intent.json'), + JSON.stringify({ + transactionId: 'seed', + targetGeneration: 9, + writes: [ + { + relativePath: 'efforts.md', + before: { exists: false }, + after: { + sha256: createHash('sha256').update(body).digest('hex'), + base64: body.toString('base64'), + }, + }, + ], + touchedIds: [], + }) + ); + await writeFile(join(td, 'committed'), ''); + t.deepEqual(await bridge.writer.recover(), { + action: 'completed', + transactionId: 'seed', + }); + t.is(fake.calls.length, 1); + t.is( + JSON.parse( + await readFile(join(root, '.journal', 'generation.json'), 'utf8') + ).generation, + 9 + ); + } +); diff --git a/packages/effort-graph/src/errors.ts b/packages/effort-graph/src/errors.ts index 49ba1733..a810d07f 100644 --- a/packages/effort-graph/src/errors.ts +++ b/packages/effort-graph/src/errors.ts @@ -1,3 +1,8 @@ +import type { + CommittedGenerationPublication, + GenerationToken, +} from './types.js'; + export class EffortGraphError extends Error { constructor(message: string, readonly code: string) { super(message); @@ -26,3 +31,31 @@ export class EffortGraphCorruptJournalError extends EffortGraphError { super(message, 'EFFORT_GRAPH_CORRUPT_JOURNAL'); } } + +export class EffortGraphLiveSchemaRejectedError extends EffortGraphError { + constructor( + readonly publication: CommittedGenerationPublication, + readonly originalError: Error + ) { + super( + `Live schema rejected generation ${publication.targetGeneration}: ${originalError.message}`, + 'EFFORT_GRAPH_LIVE_SCHEMA_REJECTED' + ); + } +} +export class EffortGraphGenerationWaitTimeoutError extends EffortGraphError { + constructor(readonly token: GenerationToken, readonly timeoutMs: number) { + super( + `Timed out waiting for committed generation ${token} after ${timeoutMs}ms`, + 'EFFORT_GRAPH_GENERATION_WAIT_TIMEOUT' + ); + } +} +export class EffortGraphBarrierTimeoutError extends EffortGraphError { + constructor(readonly paths: readonly string[], readonly maxWaitMs: number) { + super( + `Timed out waiting for journal readability after ${maxWaitMs}ms`, + 'EFFORT_GRAPH_BARRIER_TIMEOUT' + ); + } +} diff --git a/packages/effort-graph/src/index.ts b/packages/effort-graph/src/index.ts index a70c3721..b83804bb 100644 --- a/packages/effort-graph/src/index.ts +++ b/packages/effort-graph/src/index.ts @@ -10,3 +10,5 @@ export * from './snapshot.js'; export * from './decision-lifecycle.js'; export { acquireWriterLock } from './lock.js'; export { recoverJournal } from './journal.js'; +export * from './live.js'; +export * from './journalBarrier.js'; diff --git a/packages/effort-graph/src/journal.ts b/packages/effort-graph/src/journal.ts index 08514bdf..b284aa90 100644 --- a/packages/effort-graph/src/journal.ts +++ b/packages/effort-graph/src/journal.ts @@ -15,7 +15,11 @@ import { EffortGraphLockedError, EffortGraphReindexFailedError, } from './errors.js'; -import type { PlannedWrite, RecoveryResult, ReindexRequest } from './types.js'; +import type { + PlannedWrite, + RecoveryResult, + CommittedGenerationPublication, +} from './types.js'; async function fsyncFile(path: string): Promise { const handle = await open(path, 'r+'); @@ -81,7 +85,7 @@ async function removeTempRemnants(root: string, relativePaths: string[]) { export async function recoverJournal( root: string, - reindex: (r: ReindexRequest) => Promise + publish: (r: CommittedGenerationPublication) => Promise ): Promise { const dir = join(root, '.journal', 'txns'); await mkdir(dir, { recursive: true }); @@ -134,7 +138,7 @@ export async function recoverJournal( } } try { - await reindex({ + await publish({ rootDir: root, transactionId: intent.transactionId, targetGeneration: String(intent.targetGeneration), @@ -164,7 +168,7 @@ export async function commitJournal( token: string, writes: PlannedWrite[], generation: number, - reindex: (r: ReindexRequest) => Promise, + publish: (r: CommittedGenerationPublication) => Promise, verifyLock: () => Promise = async () => true ): Promise { const transactionId = `${Date.now()}-${token.slice(0, 8)}`, @@ -205,7 +209,7 @@ export async function commitJournal( await writeFileDurable(join(td, 'committed'), ''); await fsyncDir(td); try { - await reindex({ + await publish({ rootDir: root, transactionId, targetGeneration: String(generation), diff --git a/packages/effort-graph/src/journalBarrier.ts b/packages/effort-graph/src/journalBarrier.ts new file mode 100644 index 00000000..791b71af --- /dev/null +++ b/packages/effort-graph/src/journalBarrier.ts @@ -0,0 +1,100 @@ +import { access, readFile, readdir } from 'node:fs/promises'; +import { relative, resolve, join, sep } from 'node:path'; +import type { ReindexBarrier } from '@flatbread/core'; +import { + EffortGraphBarrierTimeoutError, + EffortGraphCorruptJournalError, + EffortGraphValidationError, +} from './errors.js'; + +export interface JournalReindexBarrierOptions { + rootDir: string; + pollIntervalMs?: number; + maxWaitMs?: number; +} + +async function blockingWrites(rootDir: string): Promise { + const txnsDir = join(rootDir, '.journal', 'txns'); + let transactions: string[]; + try { + transactions = await readdir(txnsDir); + } catch { + return []; + } + const paths: string[] = []; + for (const transaction of transactions) { + const dir = join(txnsDir, transaction); + const intentPath = join(dir, 'intent.json'); + let intentText: string; + try { + intentText = await readFile(intentPath, 'utf8'); + } catch { + continue; + } + let intent: { writes?: Array<{ relativePath?: string }> }; + try { + intent = JSON.parse(intentText); + } catch { + throw new EffortGraphCorruptJournalError(`Cannot parse ${intentPath}`); + } + let committed = false; + try { + await access(join(dir, 'committed')); + committed = true; + } catch {} + if (committed) continue; + if (!Array.isArray(intent.writes)) { + throw new EffortGraphCorruptJournalError( + `Invalid writes in ${intentPath}` + ); + } + for (const write of intent.writes) { + if (typeof write.relativePath !== 'string') { + throw new EffortGraphCorruptJournalError( + `Invalid relativePath in ${intentPath}` + ); + } + const absolute = resolve(rootDir, write.relativePath); + const outside = relative(rootDir, absolute); + if (outside === '..' || outside.startsWith(`..${sep}`)) { + throw new EffortGraphValidationError( + `Journal path escapes root directory: ${write.relativePath}` + ); + } + paths.push(absolute); + } + } + return paths; +} + +export function createJournalReindexBarrier( + options: JournalReindexBarrierOptions +): ReindexBarrier { + const pollIntervalMs = options.pollIntervalMs ?? 25; + const maxWaitMs = options.maxWaitMs ?? 10_000; + return { + async waitUntilReadable(requestedPaths) { + const requested = requestedPaths.map((path) => resolve(path)); + const started = Date.now(); + for (;;) { + const blocked = await blockingWrites(options.rootDir); + if ( + !blocked.some((write) => + requested.some( + (path) => + path === write || + path.startsWith(`${write}${sep}`) || + write.startsWith(`${path}${sep}`) + ) + ) + ) + return; + if (Date.now() - started >= maxWaitMs) + throw new EffortGraphBarrierTimeoutError(requestedPaths, maxWaitMs); + await new Promise((resolveSleep) => + setTimeout(resolveSleep, pollIntervalMs) + ); + } + }, + }; +} diff --git a/packages/effort-graph/src/live.ts b/packages/effort-graph/src/live.ts new file mode 100644 index 00000000..0ff3613b --- /dev/null +++ b/packages/effort-graph/src/live.ts @@ -0,0 +1,152 @@ +import { readFile } from 'node:fs/promises'; +import { relative, resolve, sep, join } from 'node:path'; +import type { LiveSchemaReloader, SchemaSnapshot } from '@flatbread/core'; +import { + EffortGraphGenerationWaitTimeoutError, + EffortGraphLiveSchemaRejectedError, + EffortGraphValidationError, +} from './errors.js'; +import { createEffortGraphWriter } from './writer.js'; +import type { + CommittedGenerationPublication, + CommittedGenerationPublisher, + EffortGraphWriter, + EffortGraphWriterOptions, + GenerationToken, +} from './types.js'; + +export interface WaitForCommittedGenerationOptions { + timeoutMs?: number; +} +export interface EffortGraphLiveBridge { + readonly rootDir: string; + readonly writer: EffortGraphWriter; + readonly publisher: CommittedGenerationPublisher; + waitForCommittedGeneration( + token: GenerationToken, + options?: WaitForCommittedGenerationOptions + ): Promise; +} +export interface CreateEffortGraphLiveBridgeOptions + extends Omit { + reloader: LiveSchemaReloader; + pollIntervalMs?: number; +} + +async function generationOnDisk(rootDir: string): Promise { + try { + const value = JSON.parse( + await readFile(join(rootDir, '.journal', 'generation.json'), 'utf8') + ).generation; + return Number.isInteger(value) && value > 0 ? value : 0; + } catch { + return 0; + } +} + +function tokenNumber(token: string): number { + if (!/^[1-9]\d*$/.test(token)) { + throw new EffortGraphValidationError( + `Generation token must be a positive integer: ${token}` + ); + } + return Number(token); +} + +export async function createEffortGraphLiveBridge( + options: CreateEffortGraphLiveBridgeOptions +): Promise { + const pollIntervalMs = options.pollIntervalMs ?? 25; + const liveGenerations = new Map(); + // Tokens at or below this mark were already published when the bridge was + // created, so the reloader's initial full build has read their files. + const initialHighWater = await generationOnDisk(options.rootDir); + + const publisher: CommittedGenerationPublisher = { + async publish(publication: CommittedGenerationPublication) { + const absolutePaths = publication.changedPaths.map((path) => { + const absolute = resolve(options.rootDir, path.replaceAll('\\', sep)); + const outside = relative(options.rootDir, absolute); + if (outside === '..' || outside.startsWith(`..${sep}`)) { + throw new EffortGraphValidationError( + `Publication path escapes root directory: ${path}` + ); + } + return absolute; + }); + const result = await options.reloader.notifyChanged({ + paths: absolutePaths, + source: 'writer', + }); + if (result.status === 'rejected') { + throw new EffortGraphLiveSchemaRejectedError(publication, result.error); + } + liveGenerations.set( + Number(publication.targetGeneration), + result.generation + ); + }, + }; + + const writer = createEffortGraphWriter({ + rootDir: options.rootDir, + index: options.index, + clock: options.clock, + randomBytes: options.randomBytes, + lockOptions: options.lockOptions, + publisher, + }); + + async function waitForCommittedGeneration( + token: GenerationToken, + waitOptions: WaitForCommittedGenerationOptions = {} + ): Promise { + const target = tokenNumber(token); + const deadline = + waitOptions.timeoutMs === undefined + ? undefined + : Date.now() + waitOptions.timeoutMs; + const sleep = async () => { + const remaining = + deadline === undefined ? pollIntervalMs : deadline - Date.now(); + if (remaining <= 0) + throw new EffortGraphGenerationWaitTimeoutError( + token, + waitOptions.timeoutMs! + ); + await new Promise((resolveSleep) => + setTimeout(resolveSleep, Math.min(pollIntervalMs, remaining)) + ); + }; + // Pure poll loop: publication durability is a disk fact, and the + // co-located fast path (a token returned by mutate()) resolves on the + // first iteration, so no publish-side wakeup machinery is needed. + for (;;) { + const diskGeneration = await generationOnDisk(options.rootDir); + if (diskGeneration >= target) { + const mapped = liveGenerations.get(target); + if (mapped !== undefined) { + return await options.reloader.waitForGeneration(mapped); + } + if (target <= initialHighWater) return options.reloader.getSnapshot(); + const result = await options.reloader.replaceConfig( + options.reloader.getSnapshot().graph.config + ); + if (result.status === 'rejected') throw result.error; + liveGenerations.set(target, result.generation); + return options.reloader.getSnapshot(); + } + if (liveGenerations.has(target)) { + await options.reloader.waitForGeneration(liveGenerations.get(target)!); + } + await sleep(); + } + } + + return { + rootDir: options.rootDir, + writer, + publisher, + waitForCommittedGeneration, + }; +} diff --git a/packages/effort-graph/src/preset.ts b/packages/effort-graph/src/preset.ts index 6d6cd9ee..4d1bd79d 100644 --- a/packages/effort-graph/src/preset.ts +++ b/packages/effort-graph/src/preset.ts @@ -50,3 +50,58 @@ export function effortGraphContent( }, ]; } + +function equalRefs( + left: Record | undefined, + right: Record | undefined +): boolean { + const leftKeys = Object.keys(left ?? {}).sort(); + const rightKeys = Object.keys(right ?? {}).sort(); + return ( + leftKeys.length === rightKeys.length && + leftKeys.every( + (key, index) => key === rightKeys[index] && left![key] === right![key] + ) + ); +} + +export function findEffortGraphContentRoot( + content: readonly Pick[] +): string | undefined { + const names = [ + 'Effort', + 'Issue', + 'Finding', + 'Decision', + 'Constraint', + 'Risk', + ]; + const effort = content.find( + (entry) => entry.collection === 'Effort' && typeof entry.path === 'string' + ); + if (!effort?.path) return undefined; + const root = effort.path.endsWith('/efforts') + ? effort.path.slice(0, -'/efforts'.length) + : undefined; + if (!root) return undefined; + const expected = effortGraphContent(root); + if ( + names.some( + (name) => + content.filter((entry) => entry.collection === name).length !== 1 + ) + ) + return undefined; + for (const entry of expected) { + const actual = content.find( + (candidate) => candidate.collection === entry.collection + ); + if ( + !actual || + actual.path !== entry.path || + !equalRefs(actual.refs, entry.refs) + ) + return undefined; + } + return root; +} diff --git a/packages/effort-graph/src/types.ts b/packages/effort-graph/src/types.ts index d44e6410..12846b20 100644 --- a/packages/effort-graph/src/types.ts +++ b/packages/effort-graph/src/types.ts @@ -23,15 +23,15 @@ export interface MutationResult { artifacts: WrittenArtifact[]; touched: TouchedArtifact[]; } -export interface ReindexRequest { +export interface CommittedGenerationPublication { rootDir: string; transactionId: string; targetGeneration: GenerationToken; changedPaths: readonly string[]; touchedIds: readonly string[]; } -export interface EffortGraphIndexer { - reindex(request: ReindexRequest): Promise; +export interface CommittedGenerationPublisher { + publish(publication: CommittedGenerationPublication): Promise; } export interface IndexedArtifact { id: string; @@ -56,7 +56,7 @@ export interface WriterLockOptions { export interface EffortGraphWriterOptions { rootDir: string; index?: import('./snapshot.js').EffortGraphSnapshotSource; - indexer?: EffortGraphIndexer; + publisher?: CommittedGenerationPublisher; clock?: () => Date; randomBytes?: (length: number) => Uint8Array; lockOptions?: Partial; diff --git a/packages/effort-graph/src/writer.ts b/packages/effort-graph/src/writer.ts index 74088a12..baf2be11 100644 --- a/packages/effort-graph/src/writer.ts +++ b/packages/effort-graph/src/writer.ts @@ -15,12 +15,12 @@ export function createEffortGraphWriter( options: EffortGraphWriterOptions ): EffortGraphWriter { const snapshotSource = options.index ?? filesystemEffortGraphSnapshotSource, - indexer = options.indexer ?? { reindex: async () => {} }, + publisher = options.publisher ?? { publish: async () => {} }, clock = options.clock ?? (() => new Date()); async function recover() { const lock = await acquireWriterLock(options.rootDir, options.lockOptions); try { - return await recoverJournal(options.rootDir, (r) => indexer.reindex(r)); + return await recoverJournal(options.rootDir, (p) => publisher.publish(p)); } finally { await lock.release(); } @@ -28,7 +28,7 @@ export function createEffortGraphWriter( async function mutate(input: any): Promise { const lock = await acquireWriterLock(options.rootDir, options.lockOptions); try { - await recoverJournal(options.rootDir, (r) => indexer.reindex(r)); + await recoverJournal(options.rootDir, (p) => publisher.publish(p)); let current = 0; try { current = @@ -53,7 +53,7 @@ export function createEffortGraphWriter( lock.token, writes, generation, - (r) => indexer.reindex(r), + (p) => publisher.publish(p), () => lock.verify() ); const artifacts: WrittenArtifact[] = writes.map((w) => { diff --git a/packages/flatbread/package.json b/packages/flatbread/package.json index b19b0151..88bd511a 100644 --- a/packages/flatbread/package.json +++ b/packages/flatbread/package.json @@ -37,6 +37,7 @@ "@flatbread/codegen": "workspace:*", "@flatbread/config": "workspace:*", "@flatbread/core": "workspace:*", + "@flatbread/effort-graph": "workspace:*", "@flatbread/source-filesystem": "workspace:*", "@flatbread/transformer-markdown": "workspace:*", "@flatbread/transformer-yaml": "workspace:*", diff --git a/packages/flatbread/src/graphql/effortGraphComposition.test.ts b/packages/flatbread/src/graphql/effortGraphComposition.test.ts new file mode 100644 index 00000000..b51fce07 --- /dev/null +++ b/packages/flatbread/src/graphql/effortGraphComposition.test.ts @@ -0,0 +1,67 @@ +import test from 'ava'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { effortGraphContent } from '@flatbread/effort-graph'; +import type { + ContentEntry, + LiveSchemaReloader, + SchemaSnapshot, +} from '@flatbread/core'; +import { createEffortGraphComposition } from './effortGraphComposition.js'; + +function fakeReloader(): LiveSchemaReloader { + let generation = 0; + return { + get generation() { + return generation; + }, + getSnapshot: () => ({} as SchemaSnapshot), + notifyChanged: async () => ({ + status: 'committed', + generation: ++generation, + }), + replaceConfig: async () => ({ + status: 'committed', + generation: ++generation, + }), + waitForGeneration: async () => ({} as SchemaSnapshot), + }; +} + +test.serial( + 'activates for the complete effortGraphContent preset', + async (t) => { + const cwd = await mkdtemp(join(tmpdir(), 'eg-composition-')); + t.teardown(() => rm(cwd, { recursive: true, force: true })); + const content = [ + ...effortGraphContent(), + { collection: 'Other', path: 'other' }, + ]; + const composition = createEffortGraphComposition(content, { cwd }); + t.truthy(composition); + t.is(composition!.rootDir, join(cwd, '.flatbread-efforts')); + t.truthy(composition!.barrier); + const bridge = await composition!.attach(fakeReloader()); + t.is(bridge.rootDir, composition!.rootDir); + t.truthy(bridge.writer); + await bridge.writer.mutate({ type: 'CreateEffort', title: 'E', body: '' }); + t.pass(); + } +); + +test.serial('stays inert for ordinary and lookalike content', (t) => { + t.is( + createEffortGraphComposition([{ collection: 'Post', path: 'posts' }], { + cwd: '/tmp', + }), + undefined + ); + const altered = effortGraphContent().map((entry, index) => + index === 2 ? { ...entry, path: `${entry.path}-altered` } : entry + ); + t.is( + createEffortGraphComposition(altered as ContentEntry[], { cwd: '/tmp' }), + undefined + ); +}); diff --git a/packages/flatbread/src/graphql/effortGraphComposition.ts b/packages/flatbread/src/graphql/effortGraphComposition.ts new file mode 100644 index 00000000..e7d13b28 --- /dev/null +++ b/packages/flatbread/src/graphql/effortGraphComposition.ts @@ -0,0 +1,32 @@ +import { resolve } from 'node:path'; +import type { + ContentEntry, + LiveSchemaReloader, + ReindexBarrier, +} from '@flatbread/core'; +import { + createEffortGraphLiveBridge, + createJournalReindexBarrier, + findEffortGraphContentRoot, + type EffortGraphLiveBridge, +} from '@flatbread/effort-graph'; + +export interface EffortGraphComposition { + readonly rootDir: string; + readonly barrier: ReindexBarrier; + attach(reloader: LiveSchemaReloader): Promise; +} + +export function createEffortGraphComposition( + content: readonly ContentEntry[], + options: { cwd: string } +): EffortGraphComposition | undefined { + const presetRoot = findEffortGraphContentRoot(content); + if (!presetRoot) return undefined; + const rootDir = resolve(options.cwd, presetRoot); + return { + rootDir, + barrier: createJournalReindexBarrier({ rootDir }), + attach: (reloader) => createEffortGraphLiveBridge({ rootDir, reloader }), + }; +} diff --git a/packages/flatbread/src/graphql/liveServer.ts b/packages/flatbread/src/graphql/liveServer.ts index f689bca5..00daf4bd 100644 --- a/packages/flatbread/src/graphql/liveServer.ts +++ b/packages/flatbread/src/graphql/liveServer.ts @@ -1,4 +1,3 @@ -import { subscribe } from '@parcel/watcher'; import { generateTypes } from '@flatbread/codegen'; import type { ConfigResult, @@ -11,6 +10,7 @@ import { createWatchCoordinator, type WatchCoordinator, } from '@flatbread/core'; +import type { EffortGraphLiveBridge } from '@flatbread/effort-graph'; import { ApolloServer } from '@apollo/server'; import { expressMiddleware } from '@as-integrations/express5'; import { InMemoryLRUCache } from '@apollo/utils.keyvaluecache'; @@ -18,6 +18,7 @@ import cors from 'cors'; import express, { type RequestHandler } from 'express'; import http from 'http'; import { loadFlatbreadConfig } from '../utils/getSchema'; +import { createEffortGraphComposition } from './effortGraphComposition'; export interface GraphqlServerOptions { port?: number; @@ -28,6 +29,7 @@ export interface GraphqlServerOptions { export interface RunningGraphqlServer { readonly port: number; readonly reloader: LiveSchemaReloader; + readonly effortGraph?: EffortGraphLiveBridge; close(): Promise; } @@ -37,6 +39,12 @@ export async function startGraphqlServer( const cwd = options.cwd ?? process.cwd(); if (!options.config.config) throw new Error('Config is not defined'); const config = options.config.config; + const composition = createEffortGraphComposition(config.content, { cwd }); + if (composition && !config.source.fetchPaths) { + throw new Error( + 'Flatbread effort-graph mode requires the configured source to implement fetchPaths(paths).' + ); + } if (options.watch && !config.source.fetchPaths) { throw new Error( 'Flatbread watch mode requires the configured source to implement fetchPaths(paths).' @@ -91,6 +99,7 @@ export async function startGraphqlServer( ); const reloader = await createLiveSchemaReloader({ config, + barrier: composition?.barrier, commitSchema: async (candidate) => { const next = await startGeneration(candidate.schema); const old = current; @@ -100,6 +109,14 @@ export async function startGraphqlServer( }, }); app.use((req, res, next) => currentMiddleware(req, res, next)); + const effortGraph = await composition?.attach(reloader); + if (effortGraph) { + try { + await effortGraph.writer.recover(); + } catch (error) { + console.error('Flatbread effort-graph recovery failed:', error); + } + } await new Promise((listenResolve) => httpServer.listen({ port: options.port ?? 5050 }, listenResolve) ); @@ -111,6 +128,7 @@ export async function startGraphqlServer( let subscription: { unsubscribe(): Promise } | undefined; let coordinator: WatchCoordinator | undefined; if (options.watch) { + const { subscribe } = await import('@parcel/watcher'); coordinator = createWatchCoordinator({ config, cwd, @@ -166,6 +184,7 @@ export async function startGraphqlServer( return { port, reloader, + effortGraph, async close() { if (closed) return; closed = true; diff --git a/packages/flatbread/src/graphql/liveServerEffortGraph.test.ts b/packages/flatbread/src/graphql/liveServerEffortGraph.test.ts new file mode 100644 index 00000000..b2c20879 --- /dev/null +++ b/packages/flatbread/src/graphql/liveServerEffortGraph.test.ts @@ -0,0 +1,140 @@ +import test from 'ava'; +import { createHash } from 'node:crypto'; +import { mkdir, mkdtemp, readFile, writeFile, rm } from 'node:fs/promises'; +import { join, relative } from 'node:path'; +import filesystem from '@flatbread/source-filesystem'; +import markdownTransformer from '@flatbread/transformer-markdown'; +import { initializeConfig } from '@flatbread/core'; +import { effortGraphContent } from '@flatbread/effort-graph'; +import type { ConfigResult, LoadedFlatbreadConfig } from '@flatbread/core'; +import { startGraphqlServer } from './liveServer.js'; + +async function makeDir() { + const dir = await mkdtemp(join(process.cwd(), '.tmp-effort-live-')); + const root = join(dir, 'graph'); + for (const path of [ + 'efforts', + 'issues', + 'findings', + 'decisions', + 'constraints', + 'risks', + ]) + await mkdir(join(root, path), { recursive: true }); + await mkdir(join(root, 'plain'), { recursive: true }); + return { dir, root, relativeRoot: relative(process.cwd(), root) }; +} + +function config( + root: string, + active: boolean +): ConfigResult { + return { + config: initializeConfig({ + source: filesystem(), + transformer: markdownTransformer(), + content: active + ? effortGraphContent(root) + : [{ collection: 'Plain', path: `${root}/plain` }], + }), + }; +} + +async function query(port: number, source: string) { + const response = await fetch(`http://localhost:${port}`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ query: source }), + }); + return (await response.json()) as { + data?: Record; + errors?: Array<{ message: string }>; + }; +} + +test.serial( + 'active preset exposes the bridge and a mutation is strictly readable end-to-end', + async (t) => { + const fixture = await makeDir(); + t.teardown(() => rm(fixture.dir, { recursive: true, force: true })); + const server = await startGraphqlServer({ + config: config(fixture.relativeRoot, true), + port: 0, + }); + t.teardown(() => server.close()); + t.truthy(server.effortGraph); + const result = await server.effortGraph!.writer.mutate({ + type: 'CreateEffort', + title: 'Committed effort', + body: '', + }); + await server.effortGraph!.waitForCommittedGeneration(result.generation); + const response = await query(server.port, '{ allEfforts { title } }'); + t.deepEqual(response.errors, undefined); + t.deepEqual(response.data?.allEfforts, [{ title: 'Committed effort' }]); + } +); + +test.serial( + 'inactive config exposes no effortGraph and behaves as before', + async (t) => { + const fixture = await makeDir(); + t.teardown(() => rm(fixture.dir, { recursive: true, force: true })); + const server = await startGraphqlServer({ + config: config(fixture.relativeRoot, false), + port: 0, + }); + t.teardown(() => server.close()); + t.is(server.effortGraph, undefined); + } +); + +test.serial( + 'boot recovery completes a committed-unpublished transaction before listen', + async (t) => { + const fixture = await makeDir(); + t.teardown(() => rm(fixture.dir, { recursive: true, force: true })); + const body = Buffer.from( + '---\nid: boot-effort\nstatus: active\n---\n\nBoot\n' + ); + const path = join(fixture.root, 'efforts', 'boot-effort.md'); + await writeFile(path, body); + const txn = join(fixture.root, '.journal', 'txns', 'boot'); + await mkdir(txn, { recursive: true }); + await writeFile( + join(txn, 'intent.json'), + JSON.stringify({ + transactionId: 'boot', + targetGeneration: 7, + writes: [ + { + relativePath: 'efforts/boot-effort.md', + before: { exists: false }, + after: { + sha256: createHash('sha256').update(body).digest('hex'), + base64: body.toString('base64'), + }, + }, + ], + touchedIds: ['boot-effort'], + }) + ); + await writeFile(join(txn, 'committed'), ''); + const server = await startGraphqlServer({ + config: config(fixture.relativeRoot, true), + port: 0, + }); + t.teardown(() => server.close()); + t.is( + JSON.parse( + await readFile( + join(fixture.root, '.journal', 'generation.json'), + 'utf8' + ) + ).generation, + 7 + ); + const snapshot = await server.effortGraph!.waitForCommittedGeneration('7'); + t.is(snapshot.generation, 1); + } +); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e716a8d6..a3bfe2c8 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -372,6 +372,9 @@ importers: '@flatbread/core': specifier: workspace:* version: link:../core + '@flatbread/effort-graph': + specifier: workspace:* + version: link:../effort-graph '@flatbread/source-filesystem': specifier: workspace:* version: link:../source-filesystem