diff --git a/docs/interactive-s07-capacity.md b/docs/interactive-s07-capacity.md new file mode 100644 index 0000000..91c6606 --- /dev/null +++ b/docs/interactive-s07-capacity.md @@ -0,0 +1,34 @@ +# FlowKit Interactive P0-A — S07 bounded Checkpoint / receipt liveness contract + +> Stage 1 P1 fix only. FlowKit is a deterministic JavaScript workflow engine, **not** a business artifact repository or Context service. No new user file storage, object store, context graph, GC or TTL is introduced here. + +## Capacity, identity and explicit outcomes + +- The **hard** canonical JSON Checkpoint limit remains **2 MiB**, preserving its existing external corruption/read guard. +- The Store reserves **256 KiB** of capacity for deterministic failure and recovery. Creation and every **nonterminal** transition must stay at or below **1.75 MiB** (`MAX_ACTIVE_CHECKPOINT_BYTES`); this includes Pending preparation/publication, so an impossible-to-recover Pending is not published merely because its waiting snapshot fit the absolute maximum. +- There is **no arbitrary per-Agent 128 KiB limit**. Agent result JSON has the existing **2 MiB hard canonicalization ceiling**, but whether it can be accepted depends on the entire **1.75 MiB active Checkpoint** including State, Pending, Candidate, and receipt cache. Larger business outputs should be saved by the **Host's user-selected storage** before signing/returning a small reference in the existing Agent result. FlowKit never decides where files are saved or how business content expires. +- When a Host-authenticated, matching Agent receipt exceeds the inline bound, or a Coordinator transition would exceed the active Checkpoint budget, the Coordinator deterministically commits `errored(FK_I_CHECKPOINT_CAPACITY)` with original State intact, clears Pending/Candidate, appends a Journal entry, and then reports the error to the Host. It **does not** pretend to have accepted the receipt, does not manufacture `acceptedRevision`, and never emits a second Pending. Another Host may use `load` to inspect the durable terminal state. A different process racing this transition is resolved by optimistic CAS. +- An initial State that would overflow the active limit is rejected **before** Run creation; users should pass small values or Host-managed references. +- The soft budget applies to nonterminal transitions; a terminal error may use the remaining reserved capacity. Under the Store's new admission rules every newly written active Checkpoint has enough space for a bounded terminal error even after its Pending is removed. +- Malformed or oversized **untrusted** input rejected before Host authentication does not authorize an attacker to terminate another user's Run. The Host verifies provenance before invoking the private Coordinator. + +## Receipt index — still FlowKit execution data + +The default SQLite Store adds a narrowly scoped table `fk_i_receipt_index`: + +`(namespace, run_id, request_id, node_id, attempt, digest, accepted_revision, entry_hash)` + +This table is an **immutable idempotency index**, *not* a business result repository. A successful receipt CAS inserts the index row, updates the Checkpoint, and appends the audit Journal **inside one SQLite IMMEDIATE transaction**. `getReceipt(runId, requestId)` is namespace-scoped and verifies stored digest/row identity against the committed receipt Journal entry. + +The Checkpoint retains at most the 16 most recent receipt summaries for low-cost inspection; all earlier accepted receipts remain accessible from the immutable SQLite index by `requestId` with their original `acceptedRevision` and digest. Existing Checkpoints whose full inline receipt history predates this feature are backfilled from their verified committed Journal records under the next receipt CAS before bounded compaction; no evidence is silently discarded. + +Existing `CheckpointStore` adapters without the optional `getReceipt` capability retain legacy inline receipt handling. The reference SQLite adapter implements the new index and enforces the capacity contract. Other adapters must uphold equivalent atomicity and capacity/terminal behavior to claim S07 conformance. + +## Non-goals and evidence + +- No file I/O API or Context integration. The Host chooses where business outputs and temporary files live and when they expire. +- No automatic pruning of the SQLite receipt index, Journal or Program bytes. Run retention and schema migration are separate future work; an audit index must not disappear while an accepted Run may be queried. +- The M6 near-2MiB adversarial seed must now fail **before** creating an unacceptably large waiting Checkpoint, while the original valid 64-KiB Agent result proceeds from the prior safe Checkpoint. Oversized inline results must end in durable `errored`, preserving State, journal integrity and immutable Program identity. +- Stage 1 cannot claim S13 real Codex/Claude/Grok model/tool integration or mark M6 complete. + +The change keeps the Program's routing authority, the 2 MiB hard integrity limit, namespace and artifact pins, and the existing Standard API untouched. diff --git a/docs/interactive.md b/docs/interactive.md index 434c960..a48139a 100644 --- a/docs/interactive.md +++ b/docs/interactive.md @@ -83,6 +83,24 @@ try { } ~~~ +## S07 capacity and idempotency boundary + +The reference SQLite Store enforces a 2 MiB hard Checkpoint bound and reserves +256 KiB for a durable terminal error. The capacity of any inline Agent result +depends on the **full** Checkpoint size; no additional arbitrary 128-KiB limit +is imposed. Larger business data belongs to the Host-selected storage +provider; Agent results should return only a small user-defined reference. + +A signed matching inline result that exceeds the bound leads to a durable +`errored(FK_I_CHECKPOINT_CAPACITY)` transition **without an acceptedRevision**. +For accepted receipts the Store maintains a separate immutable SQLite +execution-only receipt index, while the Checkpoint retains only a bounded +recent-receipt cache. Older accepted request IDs retain their exact +acknowledgment across restarts, including after proof expiry. + +See [the focused S07 storage contract](./interactive-s07-capacity.md). No +business-artifact persistence, file GC or Context module is added. + ## Invariants and recovery 1. **Program is the routing authority.** The generated `advance(view,event)` diff --git a/src/interactive/canonical.mjs b/src/interactive/canonical.mjs index 0aaf1fc..20acbfe 100644 --- a/src/interactive/canonical.mjs +++ b/src/interactive/canonical.mjs @@ -6,6 +6,15 @@ import {types as nodeTypes} from 'node:util'; */ export const MAX_CHECKPOINT_BYTES = 2 * 1024 * 1024; export const MAX_JOURNAL_BYTES = 2 * 1024 * 1024; +// The reference Store admits nonterminal Checkpoints only below this budget: +// a failed Agent receipt must still leave enough bytes for a durable error exit. +export const CHECKPOINT_ERROR_RESERVE_BYTES = 256 * 1024; +export const MAX_ACTIVE_CHECKPOINT_BYTES = MAX_CHECKPOINT_BYTES - CHECKPOINT_ERROR_RESERVE_BYTES; +// There is no extra arbitrary per-Agent 128-KiB limit: the hard serialization +// bound stays 2 MiB. Admission is decided by the *whole* current Checkpoint +// against MAX_ACTIVE_CHECKPOINT_BYTES, not by payload size in isolation. +// Large business artifacts remain Host-owned. +export const MAX_INLINE_AGENT_RESULT_BYTES = MAX_CHECKPOINT_BYTES; function validUnicode(value) { for (let i = 0; i < value.length; i += 1) { diff --git a/src/interactive/coordinator.mjs b/src/interactive/coordinator.mjs index fc93bae..10d72e6 100644 --- a/src/interactive/coordinator.mjs +++ b/src/interactive/coordinator.mjs @@ -1,11 +1,12 @@ import {randomUUID} from 'node:crypto'; -import {canonicalBytes, safeJSONSnapshot, sha256Bytes, assertSHA256} from './canonical.mjs'; +import {MAX_INLINE_AGENT_RESULT_BYTES, canonicalBytes, safeJSONSnapshot, sha256Bytes, assertSHA256} from './canonical.mjs'; import {loadInteractiveProgram} from './load.mjs'; const FINISHED = new Set(['succeeded','failed','errored']); // P0-A has at most 500 static nodes and 11 attempts per Agent. // Do not cap legitimate 500-node workflows below their worst-case 5,500 receipts. const MAX_RECEIPTS = 8192; +const MAX_INLINE_RECEIPT_CACHE = 16; const MAX_REASONS = 32; const MAX_REASON_LENGTH = 512; @@ -185,11 +186,19 @@ function assertRuntime(program,runtime) { } } function initialCheckpoint(namespace, runId, program, policyHash, state) { + let snapshot; + try { snapshot=copy(state); } + catch(error) { + if (error instanceof RangeError && /Canonical JSON exceeds maximum bytes/.test(error.message)) { + throw failure('FK_I_CHECKPOINT_CAPACITY', 'initial State exceeds Checkpoint hard limit'); + } + throw error; + } return { schemaVersion:1,namespace,runId,revision:0, artifactHash:program.artifactHash,policyHash,status:'running', blockId:program.programTable.entryBlockId, - phase:'enter',state:copy(state), + phase:'enter',state:snapshot, controlStack:[],activationSeq:0,receipts:[], pending:null,prepared:null,candidate:null,terminal:null, contextError:null, @@ -257,6 +266,42 @@ export function createInteractiveCoordinator({store, runtime, policy = {}, maxSt throw error; } } + function failCapacity(cp) { + if (FINISHED.has(cp.status)) return {committed:false,checkpoint:cp}; + const nodeId=cp.blockId.slice(0,cp.blockId.lastIndexOf('::')); + const next={ + ...cp,revision:cp.revision+1,status:'errored', + terminal:{status:'errored',nodeId,causeCode:'FK_I_CHECKPOINT_CAPACITY'}, + pending:null,prepared:null,candidate:null, + pauseReason:null, + }; + // This terminal transition may use the Store's reserved 256 KiB; it + // never drops State or accepted receipt identities. + return store.compareAndSwap({ + runId:cp.runId,expectedRevision:cp.revision,nextCheckpoint:next, + transitionId:stableTransitionId(cp,'capacity-terminal'), + journalEntry:{kind:'capacity-terminal',causeCode:'FK_I_CHECKPOINT_CAPACITY', + from:cp.blockId}, + }); + } + function casOrCapacityExit(cp, args) { + try { return store.compareAndSwap(args); } + catch(error) { + if (error?.code!=='FK_I_CHECKPOINT_CAPACITY') throw error; + const terminal=failCapacity(cp); + return {...terminal,capacityExit:terminal.committed}; + } + } + function oversizedInline(value) { + try { canonicalBytes(value,MAX_INLINE_AGENT_RESULT_BYTES);return false; } + catch(error) { + if (error instanceof RangeError && /Canonical JSON exceeds maximum bytes/.test(error.message)) { + return true; + } + throw error; + } + } + function stepCAS(cp, proposal, eventKind, extras={}) { const next = {...cp,revision:cp.revision+1}; if (proposal.kind === 'raise-error') { @@ -306,7 +351,7 @@ export function createInteractiveCoordinator({store, runtime, policy = {}, maxSt next.activationSeq=cp.activationSeq+1; } const transitionId=stableTransitionId(cp,eventKind + '/' + proposal.kind); - const write=store.compareAndSwap({ + const write=casOrCapacityExit(cp,{ runId:cp.runId,expectedRevision:cp.revision,nextCheckpoint:next,transitionId, journalEntry:{ kind:proposal.kind,event:eventKind,from:cp.blockId, @@ -385,7 +430,7 @@ export function createInteractiveCoordinator({store, runtime, policy = {}, maxSt if (cp.status === 'paused') { if (!['stage-budget','skill-temporary'].includes(cp.pauseReason)) return asPublicFrame(cp); const unpause={...cp,status:'running',revision:cp.revision+1,pauseReason:null}; - const committed=store.compareAndSwap({runId,expectedRevision:cp.revision, + const committed=casOrCapacityExit(cp,{runId,expectedRevision:cp.revision, nextCheckpoint:unpause,transitionId:stableTransitionId(cp,'resume-budget'), journalEntry:{kind:'resume-budget'}}); if (!committed.committed) continue; @@ -404,7 +449,7 @@ export function createInteractiveCoordinator({store, runtime, policy = {}, maxSt if (['FK_R014','FK_I_TEMPORARY_SKILL'].includes(error?.code)) { const paused={...cp,revision:cp.revision+1,status:'paused', pauseReason:'skill-temporary'}; - const result=store.compareAndSwap({runId,expectedRevision:cp.revision, + const result=casOrCapacityExit(cp,{runId,expectedRevision:cp.revision, nextCheckpoint:paused,transitionId:stableTransitionId(cp,'pause-skill'), journalEntry:{kind:'pause',reason:'skill-temporary'}}); if (!result.committed) continue; @@ -451,7 +496,7 @@ export function createInteractiveCoordinator({store, runtime, policy = {}, maxSt } const paused={...cp,revision:cp.revision+1,status:'paused', pauseReason:'stage-budget'}; - const result=store.compareAndSwap({runId,expectedRevision:cp.revision, + const result=casOrCapacityExit(cp,{runId,expectedRevision:cp.revision, nextCheckpoint:paused,transitionId:stableTransitionId(cp,'pause-budget'), journalEntry:{kind:'pause',reason:'stage-budget'}}); if (result.committed) return asPublicFrame(result.checkpoint); @@ -496,15 +541,25 @@ export function createInteractiveCoordinator({store, runtime, policy = {}, maxSt throw failure('FK_I_INVALID_RECEIPT','Agent failure requires bounded {message:string}'); } } - const submission=copy(kind==='agent-result' + // Size-only rejection must not require serializing an enormous receipt + // into the 2 MiB Checkpoint budget before we can mark a matched Run errored. + const rawSubmission=kind==='agent-result' ? {runId,requestId,nodeId,attempt,kind,result} - : {runId,requestId,nodeId,attempt,kind,failure:safeFailure}); - const digest=fingerprint(submission); + : {runId,requestId,nodeId,attempt,kind,failure:safeFailure}; + // Both the value AND its actual receipt envelope must be serializable + // under the protocol's 2 MiB bound. Otherwise snapshotting first would + // throw a raw RangeError while leaving an authenticated Pending waiting. + const inlineTooLarge=kind==='agent-result' + && (oversizedInline(result) || oversizedInline(rawSubmission)); + const submission=inlineTooLarge ? null : copy(rawSubmission); + const digest=inlineTooLarge ? null : fingerprint(submission); for (;;) { const cp=store.load(runId); if (!cp) throw failure('FK_I_RUN_NOT_FOUND','run is not accessible'); const program=await loadPinned(cp); - const already=cp.receipts.find(r=>r.requestId===requestId); + const indexed=typeof store.getReceipt==='function'; + const already=(indexed ? store.getReceipt(runId,requestId) : null) + ?? cp.receipts.find(r=>r.requestId===requestId); if (already) { if (already.digest!==digest || already.nodeId!==nodeId || already.attempt!==attempt) { throw failure('FK_I_RECEIPT_CONFLICT','Different receipt with accepted requestId'); @@ -522,13 +577,22 @@ export function createInteractiveCoordinator({store, runtime, policy = {}, maxSt || pending.requestId!==requestId || pending.nodeId!==nodeId || pending.attempt!==attempt || frame.nodeId!==nodeId || frame.attempt!==attempt || block.nodeId!==nodeId - || cp.receipts.length >= MAX_RECEIPTS) { + || (!indexed && cp.receipts.length >= MAX_RECEIPTS)) { throw failure('FK_I_STALE_RECEIPT','Wrong request, node, attempt, or exhausted ledger'); } const accept=block.acceptTarget; if (!block.legalTargets.includes(accept) || accept!==nodeId+'::validate') { throw failure('FK_I_INVALID_TRANSITION','Awaiting receipt target is not certified'); } + // An authenticated Host may signal a business artifact by a small + // reference. FlowKit never stores oversized inline results or invents + // an external business-data provider. The durable failure is NOT an Ack. + if (inlineTooLarge) { + const terminal=failCapacity(cp); + if (!terminal.committed) continue; + throw failure('FK_I_CHECKPOINT_CAPACITY', + 'inline result exceeds the configured execution budget; Run is errored'); + } const ledgerEntry={requestId,nodeId,attempt,digest,acceptedRevision:cp.revision+1}; const next={ ...cp,revision:cp.revision+1,blockId:accept,phase:'validate', @@ -537,12 +601,18 @@ export function createInteractiveCoordinator({store, runtime, policy = {}, maxSt submissionKind:kind, ...(kind==='agent-result'?{result:submission.result} :{failure:submission.failure}),validated:false}, - receipts:[...cp.receipts,ledgerEntry], + receipts:indexed + ? [...cp.receipts.slice(-(MAX_INLINE_RECEIPT_CACHE-1)),ledgerEntry] + : [...cp.receipts,ledgerEntry], }; - const persisted=store.compareAndSwap({runId,expectedRevision:cp.revision, + const persisted=casOrCapacityExit(cp,{runId,expectedRevision:cp.revision, nextCheckpoint:next,transitionId:stableTransitionId(cp,'receipt:'+requestId), journalEntry:{kind:'receipt-accepted',nodeId,attempt,requestId,digest}}); if (!persisted.committed) continue; + if (persisted.capacityExit) { + throw failure('FK_I_CHECKPOINT_CAPACITY', + 'receipt could not be accepted; Run is durably errored'); + } const current=await drive(runId); return {acceptedRevision:ledgerEntry.acceptedRevision,...current,duplicate:false}; } diff --git a/src/interactive/sqlite-store.mjs b/src/interactive/sqlite-store.mjs index 3420369..0f2db54 100644 --- a/src/interactive/sqlite-store.mjs +++ b/src/interactive/sqlite-store.mjs @@ -1,12 +1,24 @@ import {DatabaseSync} from 'node:sqlite'; import {Buffer} from 'node:buffer'; import { - MAX_CHECKPOINT_BYTES, MAX_JOURNAL_BYTES, assertSHA256, + MAX_CHECKPOINT_BYTES, MAX_ACTIVE_CHECKPOINT_BYTES, MAX_JOURNAL_BYTES, assertSHA256, canonicalBytes, decodeCanonical, safeJSONSnapshot, sha256Bytes, } from './canonical.mjs'; const TERMINAL = new Set(['succeeded', 'failed', 'errored']); const STATUSES = new Set(['running', 'waiting', 'paused', ...TERMINAL]); +function boundedCheckpointBytes(input) { + try { return canonicalBytes(input); } + catch (error) { + if (error instanceof RangeError && /Canonical JSON exceeds maximum bytes/.test(error.message)) { + throw failure('FK_I_CHECKPOINT_CAPACITY', 'Checkpoint exceeds 2 MiB hard limit'); + } + throw error; + } +} +function boundedCheckpoint(input) { + return JSON.parse(boundedCheckpointBytes(input).toString('utf8')); +} function failure(code, detail) { return Object.assign(new Error(code + ': ' + detail), {code}); @@ -89,6 +101,15 @@ export function createSqliteCheckpointStore(file, {namespace} = {}) { ' PRIMARY KEY (namespace, run_id), UNIQUE (namespace, start_key),', ' FOREIGN KEY (namespace, artifact_hash) REFERENCES fk_i_artifacts(namespace, artifact_hash)', ') STRICT;', + 'CREATE TABLE IF NOT EXISTS fk_i_receipt_index (', + ' namespace TEXT NOT NULL, run_id TEXT NOT NULL, request_id TEXT NOT NULL,', + ' node_id TEXT NOT NULL, attempt INTEGER NOT NULL CHECK(attempt >= 1),', + ' digest TEXT NOT NULL, accepted_revision INTEGER NOT NULL CHECK(accepted_revision >= 1),', + ' entry_hash TEXT NOT NULL,', + ' PRIMARY KEY (namespace, run_id, request_id),', + ' UNIQUE (namespace, run_id, accepted_revision),', + ' FOREIGN KEY (namespace, run_id) REFERENCES fk_i_runs(namespace, run_id)', + ') STRICT;', 'CREATE TABLE IF NOT EXISTS fk_i_journal (', ' namespace TEXT NOT NULL, run_id TEXT NOT NULL, revision INTEGER NOT NULL,', ' transition_id TEXT NOT NULL, previous_hash TEXT, checkpoint_hash TEXT NOT NULL,', @@ -111,6 +132,9 @@ export function createSqliteCheckpointStore(file, {namespace} = {}) { const insertRun = db.prepare('INSERT INTO fk_i_runs(namespace, run_id, start_key, start_intent_hash, revision, artifact_hash, checkpoint_json, checkpoint_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?)'); const updateRun = db.prepare('UPDATE fk_i_runs SET revision = ?, checkpoint_json = ?, checkpoint_hash = ? WHERE namespace = ? AND run_id = ? AND revision = ?'); const insertJournal = db.prepare('INSERT INTO fk_i_journal(namespace, run_id, revision, transition_id, previous_hash, checkpoint_hash, entry_json, entry_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?)'); + const readReceipt = db.prepare('SELECT * FROM fk_i_receipt_index WHERE namespace = ? AND run_id = ? AND request_id = ?'); + const insertReceipt = db.prepare('INSERT OR IGNORE INTO fk_i_receipt_index(namespace, run_id, request_id, node_id, attempt, digest, accepted_revision, entry_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?)'); + const journalAtRevision = db.prepare('SELECT * FROM fk_i_journal WHERE namespace = ? AND run_id = ? AND revision = ?'); function transaction(mode, work) { db.exec(mode === 'write' ? 'BEGIN IMMEDIATE' : 'BEGIN'); @@ -136,6 +160,55 @@ export function createSqliteCheckpointStore(file, {namespace} = {}) { } return cp; } + function receiptProof(runId, record) { + const journalRow = journalAtRevision.get(namespace, runId, record.acceptedRevision); + if (!journalRow || journalRow.entry_hash !== record.entryHash) { + throw failure('FK_I_STORE_CORRUPT', 'receipt index has no matching journal revision'); + } + const detail = decodeCanonical(journalRow.entry_json, + sha256Bytes(Buffer.from(journalRow.entry_json, 'utf8')), MAX_JOURNAL_BYTES); + if (journalRow.entry_hash !== journalDigest(namespace, runId, record.acceptedRevision, + journalRow.transition_id, journalRow.previous_hash, journalRow.checkpoint_hash, detail) + || detail.kind !== 'receipt-accepted' + || detail.nodeId !== record.nodeId || detail.attempt !== record.attempt + || detail.requestId !== record.requestId || detail.digest !== record.digest) { + throw failure('FK_I_STORE_CORRUPT', 'receipt index differs from immutable journal'); + } + } + function putReceiptIndex(runId, record, entryHash) { + identifier(record.requestId, 'receipt.requestId'); + identifier(record.nodeId, 'receipt.nodeId'); + if (!Number.isSafeInteger(record.attempt) || record.attempt < 1 + || !Number.isSafeInteger(record.acceptedRevision) || record.acceptedRevision < 1) { + throw failure('FK_I_STORE_CORRUPT', 'invalid receipt index identity'); + } + assertSHA256(record.digest); + assertSHA256(entryHash); + insertReceipt.run(namespace, runId, record.requestId, record.nodeId, record.attempt, + record.digest, record.acceptedRevision, entryHash); + const saved = readReceipt.get(namespace, runId, record.requestId); + if (!saved || saved.node_id !== record.nodeId || saved.attempt !== record.attempt + || saved.digest !== record.digest || saved.accepted_revision !== record.acceptedRevision + || saved.entry_hash !== entryHash) { + throw failure('FK_I_RECEIPT_CONFLICT', 'accepted request identity changed'); + } + } + function legacyIndexBackfill(runId, receipts) { + for (const receipt of receipts ?? []) { + if (!receipt || !Number.isSafeInteger(receipt.acceptedRevision)) { + throw failure('FK_I_STORE_CORRUPT', 'invalid legacy receipt index'); + } + const entry = journalAtRevision.get(namespace, runId, receipt.acceptedRevision); + if (!entry) throw failure('FK_I_STORE_CORRUPT', 'legacy receipt journal missing'); + const record = { + requestId:receipt.requestId,nodeId:receipt.nodeId,attempt:receipt.attempt, + digest:receipt.digest,acceptedRevision:receipt.acceptedRevision, + entryHash:entry.entry_hash, + }; + receiptProof(runId, record); + putReceiptIndex(runId, record, entry.entry_hash); + } + } function appendJournal(runId, revision, transitionId, checkpointHash, previousHash, entry) { identifier(transitionId, 'transitionId'); const detail = safeJSONSnapshot(entry, MAX_JOURNAL_BYTES); @@ -216,19 +289,21 @@ export function createSqliteCheckpointStore(file, {namespace} = {}) { */ createRun({startKey, startIntent = null, checkpoint}) { identifier(startKey, 'startKey'); - const base = safeJSONSnapshot(checkpoint); + const base = boundedCheckpoint(checkpoint); identifier(base?.runId, 'checkpoint.runId'); assertCheckpoint(base, namespace, base.runId, 0); if (base.startKey !== undefined || base.startIntentHash !== undefined) { throw failure('FK_I_INVALID_CHECKPOINT', 'start identity is store-generated'); } const intent = safeJSONSnapshot(startIntent); - const startIntentHash = sha256Bytes(canonicalBytes({ + const startIntentHash = sha256Bytes(boundedCheckpointBytes({ namespace, startKey, artifactHash:base.artifactHash, state:base.state, policyHash:base.policyHash ?? null, intent, })); const initial = {...base, startKey, startIntentHash}; - const checkpointBytes = canonicalBytes(initial); + const checkpointBytes = boundedCheckpointBytes(initial); + // The reserve must already exist before a Run can publish its first Pending. + // Read/duplicate-start of a pre-upgrade Run is not blocked by this check. const checkpointHash = sha256Bytes(checkpointBytes); return transaction('write', () => { const existing = byStartKey.get(namespace, startKey); @@ -242,6 +317,9 @@ export function createSqliteCheckpointStore(file, {namespace} = {}) { if (read.get(namespace, initial.runId)) { throw failure('FK_I_RUN_CONFLICT', 'runId already belongs to another start'); } + if (checkpointBytes.byteLength > MAX_ACTIVE_CHECKPOINT_BYTES) { + throw failure('FK_I_CHECKPOINT_CAPACITY', 'initial Run would leave no error reserve'); + } if (!artifact.get(namespace, initial.artifactHash)) { throw failure('FK_I_PROGRAM_MISMATCH', 'Program bytes must be registered first'); } @@ -265,7 +343,7 @@ export function createSqliteCheckpointStore(file, {namespace} = {}) { if (!Number.isSafeInteger(expectedRevision) || expectedRevision < 0) { throw new TypeError('expectedRevision must be a nonnegative safe integer'); } - const next = safeJSONSnapshot(nextCheckpoint); + const next = boundedCheckpoint(nextCheckpoint); const journal = safeJSONSnapshot(journalEntry, MAX_JOURNAL_BYTES); return transaction('write', () => { const row = read.get(namespace, runId); @@ -283,17 +361,68 @@ export function createSqliteCheckpointStore(file, {namespace} = {}) { assertCheckpoint(next, namespace, runId, expectedRevision + 1, row.artifact_hash); immutableIdentity(current, next); const bytes = canonicalBytes(next); + if (!TERMINAL.has(next.status) && bytes.byteLength > MAX_ACTIVE_CHECKPOINT_BYTES) { + throw failure('FK_I_CHECKPOINT_CAPACITY', 'nonterminal Checkpoint reserve exhausted'); + } const checkpointHash = sha256Bytes(bytes); const last = latest.get(namespace, runId); if (!last) throw failure('FK_I_STORE_CORRUPT', 'missing journal head'); const updated = updateRun.run(next.revision, bytes.toString('utf8'), checkpointHash, namespace, runId, expectedRevision); if (updated.changes !== 1) throw failure('FK_I_CAS_CONFLICT', 'revision changed during transaction'); - appendJournal(runId, next.revision, transitionId, checkpointHash, last.entry_hash, journal); + const entryHash = appendJournal(runId, next.revision, transitionId, checkpointHash, + last.entry_hash, journal); + if (journal?.kind === 'receipt-accepted') { + // Backfill old inline receipts *before* a future Coordinator may + // compact its cache. Every old index row is verified against Journal. + legacyIndexBackfill(runId, current.receipts); + if (current.status !== 'waiting' || next.phase !== 'validate' + || next.candidate?.requestId !== journal.requestId + || next.candidate?.nodeId !== journal.nodeId + || next.candidate?.attempt !== journal.attempt) { + throw failure('FK_I_INVALID_RECEIPT', 'receipt transition does not match the Pending'); + } + putReceiptIndex(runId, { + requestId:journal.requestId,nodeId:journal.nodeId,attempt:journal.attempt, + digest:journal.digest,acceptedRevision:next.revision, + },entryHash); + } return {committed:true, checkpoint:JSON.parse(bytes.toString('utf8'))}; }); }, + /** Immutable, namespaced accepted-receipt lookup without growing CP. */ + getReceipt(runId, requestId) { + identifier(runId, 'runId'); + identifier(requestId, 'requestId'); + return transaction('read', () => { + const run = read.get(namespace, runId); + if (!run) return null; + const cp = decodeRow(run); + const record = readReceipt.get(namespace, runId, requestId); + const indexed = record ? { + requestId:record.request_id,nodeId:record.node_id,attempt:record.attempt, + digest:record.digest,acceptedRevision:record.accepted_revision, + entryHash:record.entry_hash, + } : null; + if (indexed) { + receiptProof(runId, indexed); + return { + requestId:indexed.requestId,nodeId:indexed.nodeId,attempt:indexed.attempt, + digest:indexed.digest,acceptedRevision:indexed.acceptedRevision, + }; + } + // Runs created before the S07 index keep their original inline history + // until it is transactionally backfilled by the next accepted receipt. + const legacy = cp.receipts?.find(x => x.requestId === requestId); + if (!legacy) return null; + const historical = journalAtRevision.get(namespace, runId, legacy.acceptedRevision); + if (!historical) throw failure('FK_I_STORE_CORRUPT', 'legacy receipt journal missing'); + receiptProof(runId, {...legacy,entryHash:historical.entry_hash}); + return safeJSONSnapshot(legacy); + }); + }, + /** Audit chain is verified on each read, independently of a live Coordinator. */ listJournal(runId) { identifier(runId, 'runId'); diff --git a/src/interactive/trusted-host.mjs b/src/interactive/trusted-host.mjs index 1c27ae0..315e226 100644 --- a/src/interactive/trusted-host.mjs +++ b/src/interactive/trusted-host.mjs @@ -237,9 +237,12 @@ export function createTrustedInteractiveHost({ // introduced. NEW receipts still require fresh proofs and signatures // are checked in both cases; raw Coordinator access stays Host-private. const receiptDigest=sha256Bytes(canonicalBytes(normalized)); - const alreadyAccepted=Array.isArray(cp.receipts) && cp.receipts.some(r=> - r.requestId===normalized.requestId && r.nodeId===normalized.nodeId - && r.attempt===normalized.attempt && r.digest===receiptDigest); + const previous=typeof store.getReceipt==='function' + ? store.getReceipt(normalized.runId,normalized.requestId) + : cp.receipts?.find(r=>r.requestId===normalized.requestId); + const alreadyAccepted=previous?.requestId===normalized.requestId + && previous.nodeId===normalized.nodeId + && previous.attempt===normalized.attempt && previous.digest===receiptDigest; if (!Number.isSafeInteger(current) || current<=0 || statement.issuedAt>current+clockSkewMs || (statement.expiresAt{for(const dir of directories)await rm(dir,{recursive:true,force:true})}); +const definition=(count=1)=>{ + const nodes=[]; + for(let i=0;i 'Return a structured result', + validateResult:(_schema,result)=>typeof result?.answer==='string'?[]:['Expected answer'], +}; +async function compiled(count=1){ + const analysis=await analyzeSource(definition(count),{ + filename:'s07.workflow.js',projectRoot:process.cwd(),skillCheck:false, + }); + assert.equal(analysis.diagnostics.some(d=>d.severity==='error'),false, + JSON.stringify(analysis.diagnostics)); + const lowered=lowerInteractive(analysis); + assert.ok(lowered.ir,JSON.stringify(lowered.diagnostics)); + const emitted=emitInteractive(lowered.ir); + assert.ok(emitted.code,JSON.stringify(emitted.diagnostics)); + return {programBytes:Buffer.from(emitted.code),programIdentity:emitted.programIdentity}; +} +async function fixture(t,count=1){ + const dir=await mkdtemp(path.join(tmpdir(),'flowkit-s07-')); + directories.push(dir); + const file=path.join(dir,'test.db'); + const store=createSqliteCheckpointStore(file,{namespace:'s07'}); + t.after(()=>store.close()); + const source=await compiled(count); + const api=createInteractiveCoordinator({store,runtime}); + return {file,store,api,source}; +} +function receipt(frame,result={answer:'ok'}){ + return {runId:frame.runId,requestId:frame.pending.requestId, + nodeId:frame.pending.nodeId,attempt:frame.pending.attempt,result}; +} +async function waiting(f,{runId='run-s07',state={message:'small'},startKey='start-s07'}={}){ + const frame=await f.api.start({...f.source,startKey,runId,state}); + assert.equal(frame.status,'waiting'); + return frame; +} +function checkJournal(store,runId){ + const cp=store.load(runId); + const journal=store.listJournal(runId); + assert.equal(journal.length,cp.revision+1); + assert.deepEqual(journal.map(r=>r.revision),Array.from({length:cp.revision+1},(_,i)=>i)); + return {cp,journal}; +} +function fillWaitingTo(f,original,limit,headroom=75){ + const current=f.store.load(original.runId); + const next={ + ...current,revision:current.revision+1, + state:{...current.state,filler:''}, + }; + const base=canonicalBytes(next).byteLength; + assert.ok(limit-base-headroom>0); + next.state.filler='x'.repeat(limit-base-headroom); + return next; +} + +test('S07 original 2 MiB waiting seed is rejected BEFORE publishing unacceptably large State; 64 KiB valid receipt still succeeds',async t=>{ + const f=await fixture(t); + const w=await waiting(f); + const before=canonicalBytes(f.store.load(w.runId)); + const journal=f.store.listJournal(w.runId); + const huge=fillWaitingTo(f,w,MAX_CHECKPOINT_BYTES); + assert.throws(()=>f.store.compareAndSwap({ + runId:w.runId,expectedRevision:w.revision,nextCheckpoint:huge, + transitionId:'s07-invalid-near-hard',journalEntry:{kind:'test-large'}, + }),{code:'FK_I_CHECKPOINT_CAPACITY'}); + assert.deepEqual(canonicalBytes(f.store.load(w.runId)),before); + assert.deepEqual(f.store.listJournal(w.runId),journal); + const complete=await f.api.complete(receipt(w,{answer:'v'.repeat(64*1024)})); + assert.equal(complete.status,'succeeded'); + assert.equal(complete.state.answer.length,64*1024); + assert.equal(f.store.getReceipt(w.runId,w.pending.requestId).acceptedRevision, + complete.acceptedRevision); + checkJournal(f.store,w.runId); +}); + +test('S07 valid 64 KiB receipt at safe active headroom deterministically errors instead of stranding waiting',async t=>{ + const f=await fixture(t); + const w=await waiting(f); + const expanded=fillWaitingTo(f,w,MAX_ACTIVE_CHECKPOINT_BYTES); + assert.equal(f.store.compareAndSwap({ + runId:w.runId,expectedRevision:w.revision,nextCheckpoint:expanded, + transitionId:'s07-near-soft',journalEntry:{kind:'test-reserved'}, + }).committed,true); + const before=f.store.load(w.runId); + assert.equal(before.status,'waiting'); + assert.ok(MAX_ACTIVE_CHECKPOINT_BYTES-canonicalBytes(before).byteLength<=75); + await assert.rejects(f.api.complete(receipt(w,{answer:'v'.repeat(64*1024)})), + {code:'FK_I_CHECKPOINT_CAPACITY'}); + const ended=f.store.load(w.runId); + assert.equal(ended.status,'errored'); + assert.equal(ended.terminal.causeCode,'FK_I_CHECKPOINT_CAPACITY'); + assert.deepEqual(ended.state,before.state); + assert.equal(ended.pending,null); + assert.equal(ended.candidate,null); + assert.equal(f.store.getReceipt(w.runId,w.pending.requestId),null); + const revision=ended.revision; + assert.equal((await f.api.resume({runId:w.runId})).revision,revision); + assert.equal((await f.api.resume({runId:w.runId})).status,'errored'); + await assert.rejects(f.api.complete(receipt(w)),/FK_I_STALE_RECEIPT/); + checkJournal(f.store,w.runId); +}); + +test('S07 oversized Host-owned inline Agent output is a durable terminal capacity error; no false acceptedRevision',async t=>{ + for(const bytes of [MAX_ACTIVE_CHECKPOINT_BYTES+1,MAX_CHECKPOINT_BYTES+1024]){ + const f=await fixture(t); + const runId='too-large-'+bytes; + const w=await waiting(f,{runId,startKey:runId,state:{valuable:'kept'}}); + const oversized=receipt(w,{answer:'z'.repeat(bytes)}); + await assert.rejects(f.api.complete(oversized),{code:'FK_I_CHECKPOINT_CAPACITY'}); + const snapshot=f.store.load(runId); + assert.equal(snapshot.status,'errored'); + assert.equal(snapshot.terminal.causeCode,'FK_I_CHECKPOINT_CAPACITY'); + assert.deepEqual(snapshot.state,{valuable:'kept'}); + assert.equal(snapshot.receipts.length,0); + assert.equal(f.store.getReceipt(runId,w.pending.requestId),null); + checkJournal(f.store,runId); + } +}); + +test('S07 initial State exceeding soft OR hard budget deterministically fails before creating Run',async t=>{ + const f=await fixture(t); + for(const bytes of [MAX_ACTIVE_CHECKPOINT_BYTES,MAX_CHECKPOINT_BYTES+1024]){ + const runId='oversized-initial-'+bytes; + const state={filler:'a'.repeat(bytes)}; + await assert.rejects(f.api.start({...f.source, + runId,startKey:runId,state}),{code:'FK_I_CHECKPOINT_CAPACITY'}); + assert.equal(f.store.load(runId),null); + assert.deepEqual(f.store.listJournal(runId),[]); + } +}); + +test('S07 21 Agent receipts retain a constant-size Checkpoint cache but all old Ack identities in SQLite index',async t=>{ + const f=await fixture(t,21); + let current=await waiting(f); + let first,originalReceipt; + for(let i=0;i<21;i++){ + assert.equal(current.pending.nodeId,'agent'+i); + const submitted=receipt(current,{answer:'done-'+i}); + const accepted=await f.api.complete(submitted); + if(i===0){first=accepted.acceptedRevision;originalReceipt=submitted} + assert.equal(f.store.getReceipt(current.runId,submitted.requestId).acceptedRevision, + accepted.acceptedRevision); + current=accepted; + } + assert.equal(current.status,'succeeded'); + const final=f.store.load(current.runId); + assert.equal(final.receipts.length,16); + assert.ok(!final.receipts.some(x=>x.requestId===originalReceipt.requestId)); + assert.equal(f.store.getReceipt(current.runId,originalReceipt.requestId).acceptedRevision,first); + const reopened=createSqliteCheckpointStore(f.file,{namespace:'s07'}); + t.after(()=>reopened.close()); + assert.equal(reopened.getReceipt(current.runId,originalReceipt.requestId).acceptedRevision,first); + const replay=await f.api.complete(originalReceipt); + assert.equal(replay.duplicate,true); + assert.equal(replay.acceptedRevision,first); + assert.equal(replay.revision,current.revision); + checkJournal(f.store,current.runId); +}); + +test('S07 M5 signed duplicate recovery checks index even after the old receipt leaves Checkpoint cache',async t=>{ + const f=await fixture(t,21); + const keys=generateKeyPairSync('ed25519'); + let clock=1794400000000; + const host=createTrustedInteractiveHost({ + store:f.store,runtime, + trustedPublicKeys:{'host':keys.publicKey}, + trustedArtifacts:[f.source.programIdentity.artifactHash], + allowedInputKeys:[],disclosurePolicyId:'s07-public-v1', + authorizeDisclosure:async()=>true,now:()=>clock, + }); + let frame=await host.start({...f.source, + startKey:'host-s07',runId:'host-s07',state:{counter:0}}); + assert.equal(frame.status,'waiting'); + let originalSigned=null,originalReceipt=null,firstRevision=null; + for(let i=0;i<21;i++){ + const task=await host.getPublicAgentTask({runId:frame.runId}); + assert.ok(task); + const item={ + runId:task.runId,requestId:task.requestId,nodeId:task.nodeId, + attempt:task.attempt,kind:'agent-result',result:{answer:'signed-'+i}, + }; + const statement=createAgentReceiptStatement({ + run:f.store.load(item.runId),receipt:item,keyId:'host',proofId:'s07-proof-'+i, + issuedAt:clock-1000,expiresAt:clock+5000, + }); + const attestation={statement, + signature:sign(null,encodeAgentReceiptStatement(statement),keys.privateKey) + .toString('base64url'), + }; + const completed=await host.accept({receipt:item,attestation}); + if(i===0) {originalReceipt=item;originalSigned=attestation;firstRevision=completed.acceptedRevision;} + frame=completed; + } + assert.equal(frame.status,'succeeded'); + assert.equal(f.store.load(frame.runId).receipts.length,16); + clock+=3600000; + const old=await host.accept({receipt:originalReceipt,attestation:originalSigned}); + assert.equal(old.duplicate,true); + assert.equal(old.acceptedRevision,firstRevision); + checkJournal(f.store,frame.runId); +}); + +test('S07 pre-upgrade inline receipt history is transactionally indexed before the next receipt is committed',async t=>{ + const f=await fixture(t,2); + const first=await waiting(f); + const receiptA=receipt(first,{answer:'a'}); + const afterA=await f.api.complete(receiptA); + assert.equal(afterA.status,'waiting'); + const original=f.store.load(first.runId).receipts[0]; + const db=new DatabaseSync(f.file); + try { + const count=db.prepare('DELETE FROM fk_i_receipt_index WHERE namespace=? AND run_id=? AND request_id=?') + .run('s07',first.runId,receiptA.requestId).changes; + assert.equal(count,1); + } finally {db.close()} + assert.deepEqual(f.store.getReceipt(first.runId,receiptA.requestId),original); + const afterB=await f.api.complete(receipt(afterA,{answer:'b'})); + assert.equal(afterB.status,'succeeded'); + assert.equal(f.store.getReceipt(first.runId,receiptA.requestId).acceptedRevision,original.acceptedRevision); + const db2=new DatabaseSync(f.file); + try { + assert.equal(db2.prepare('SELECT COUNT(*) AS n FROM fk_i_receipt_index WHERE namespace=? AND run_id=?') + .get('s07',first.runId).n,2); + }finally{db2.close()} +}); + +test('S07 corrupting receipt index identity cannot turn a different body into accepted duplicate',async t=>{ + const f=await fixture(t); + const w=await waiting(f); + await f.api.complete(receipt(w)); + const db=new DatabaseSync(f.file); + try{ + db.prepare('UPDATE fk_i_receipt_index SET digest=? WHERE namespace=? AND run_id=? AND request_id=?') + .run('sha256:'+'e'.repeat(64),'s07',w.runId,w.pending.requestId); + }finally{db.close()} + assert.throws(()=>f.store.getReceipt(w.runId,w.pending.requestId),{code:'FK_I_STORE_CORRUPT'}); + await assert.rejects(f.api.complete(receipt(w)),{code:'FK_I_STORE_CORRUPT'}); +}); + + +test('S07 pre-upgrade already-durable near-2MiB waiting Run has a terminal recovery exit without discarding State',async t=>{ + const f=await fixture(t); + const w=await waiting(f,{runId:'legacy-near-limit',startKey:'legacy-near-limit'}); + const old=f.store.load(w.runId); + const next=fillWaitingTo(f,w,MAX_CHECKPOINT_BYTES); + const bytes=canonicalBytes(next); + assert.ok(bytes.byteLength > MAX_ACTIVE_CHECKPOINT_BYTES); + const checkpointHash=sha256Bytes(bytes); + const transitionId='pre-upgrade-soft-limit-checkpoint'; + const entry={kind:'legacy-large-checkpoint-test'}; + const db=new DatabaseSync(f.file); + try { + db.exec('BEGIN IMMEDIATE'); + const previous=db.prepare( + 'SELECT entry_hash FROM fk_i_journal WHERE namespace=? AND run_id=? ORDER BY revision DESC LIMIT 1' + ).get('s07',w.runId); + assert.ok(previous); + const entryHash=sha256Bytes(canonicalBytes({ + namespace:'s07',runId:w.runId,revision:next.revision, + transitionId,previousHash:previous.entry_hash,checkpointHash,entry, + })); + db.prepare('UPDATE fk_i_runs SET revision=?, checkpoint_json=?, checkpoint_hash=? WHERE namespace=? AND run_id=?') + .run(next.revision,bytes.toString('utf8'),checkpointHash,'s07',w.runId); + db.prepare('INSERT INTO fk_i_journal(namespace,run_id,revision,transition_id,previous_hash,checkpoint_hash,entry_json,entry_hash) VALUES (?,?,?,?,?,?,?,?)') + .run('s07',w.runId,next.revision,transitionId,previous.entry_hash, + checkpointHash,canonicalBytes(entry).toString('utf8'),entryHash); + db.exec('COMMIT'); + }catch(error){ + db.exec('ROLLBACK');throw error; + }finally{db.close()} + const before=f.store.load(w.runId); + assert.equal(before.status,'waiting'); + await assert.rejects(f.api.complete(receipt(w,{answer:'x'.repeat(64*1024)})), + {code:'FK_I_CHECKPOINT_CAPACITY'}); + const final=f.store.load(w.runId); + assert.equal(final.status,'errored'); + assert.equal(final.terminal.causeCode,'FK_I_CHECKPOINT_CAPACITY'); + assert.deepEqual(final.state,before.state); + assert.equal(final.candidate,null); + assert.equal(final.pending,null); + assert.equal(f.store.getReceipt(w.runId,w.pending.requestId),null); + checkJournal(f.store,w.runId); +}); + +test('S07 capacity guard at publish and after an accepted receipt uses deterministic terminal without bogus Ack',async t=>{ + const f=await fixture(t); + let failPublish=true; + const wrap={ + ...f.store, + compareAndSwap(args){ + if(failPublish && args.journalEntry?.kind==='publish-pending'){ + failPublish=false; + throw Object.assign(new Error('Injected Store exhaustion'), + {code:'FK_I_CHECKPOINT_CAPACITY'}); + } + return f.store.compareAndSwap(args); + }, + }; + const host=createInteractiveCoordinator({store:wrap,runtime}); + const ended=await host.start({...f.source,runId:'publish-cap',startKey:'publish-cap', + state:{tiny:'unchanged'}}); + assert.equal(ended.status,'errored'); + assert.equal(ended.terminal.causeCode,'FK_I_CHECKPOINT_CAPACITY'); + assert.deepEqual(f.store.load(ended.runId).state,{tiny:'unchanged'}); + assert.equal(f.store.load(ended.runId).receipts.length,0); + assert.equal(f.store.listJournal(ended.runId).filter(x=>x.entry.kind==='publish-pending').length,0); + + const w=await waiting(f,{runId:'commit-cap',startKey:'commit-cap'}); + let failCommit=true; + const wrapCommit={ + ...f.store, + compareAndSwap(args){ + if(failCommit && args.journalEntry?.kind==='commit-state'){ + failCommit=false; + throw Object.assign(new Error('Injected commit capacity'), + {code:'FK_I_CHECKPOINT_CAPACITY'}); + } + return f.store.compareAndSwap(args); + }, + }; + const coor=createInteractiveCoordinator({store:wrapCommit,runtime}); + const original=receipt(w,{answer:'accepted'}); + const result=await coor.complete(original); + assert.equal(result.status,'errored'); + assert.equal(result.terminal.causeCode,'FK_I_CHECKPOINT_CAPACITY'); + assert.equal(result.duplicate,false); + assert.ok(result.acceptedRevision>w.revision); + const cp=f.store.load(w.runId); + assert.deepEqual(cp.state,{message:'small'}); + assert.equal(f.store.getReceipt(w.runId,original.requestId).acceptedRevision, + result.acceptedRevision); + const duplicate=await f.api.complete(original); + assert.equal(duplicate.duplicate,true); + assert.equal(duplicate.acceptedRevision,result.acceptedRevision); + checkJournal(f.store,w.runId); +}); + + +test('S07 M5 authentic signed oversize receipt closes Run durably without issuing a false Ack',async t=>{ + const f=await fixture(t); + const keys=generateKeyPairSync('ed25519'); + const now=1794400000000; + const host=createTrustedInteractiveHost({ + store:f.store,runtime,trustedPublicKeys:{provider:keys.publicKey}, + trustedArtifacts:[f.source.programIdentity.artifactHash], + allowedInputKeys:[],disclosurePolicyId:'s07-oversize-v1', + authorizeDisclosure:async()=>true,now:()=>now, + }); + const frame=await host.start({...f.source, + startKey:'signed-overlimit',runId:'signed-overlimit',state:{retain:'yes'}}); + assert.equal(frame.status,'waiting'); + const signedReceipt={ + runId:frame.runId,requestId:frame.pending.requestId,nodeId:frame.pending.nodeId, + attempt:frame.pending.attempt,kind:'agent-result', + result:{answer:'x'.repeat(MAX_ACTIVE_CHECKPOINT_BYTES)}, + }; + const statement=createAgentReceiptStatement({ + run:f.store.load(frame.runId),receipt:signedReceipt,keyId:'provider', + proofId:'signed-oversize',issuedAt:now-1000,expiresAt:now+20000, + }); + const signature=sign(null,encodeAgentReceiptStatement(statement),keys.privateKey) + .toString('base64url'); + await assert.rejects(host.accept({receipt:signedReceipt, + attestation:{statement,signature}}),{code:'FK_I_CHECKPOINT_CAPACITY'}); + const stopped=await host.load({runId:frame.runId}); + assert.equal(stopped.status,'errored'); + assert.equal(stopped.terminal.causeCode,'FK_I_CHECKPOINT_CAPACITY'); + assert.deepEqual(stopped.state,{retain:'yes'}); + assert.equal(f.store.getReceipt(frame.runId,frame.pending.requestId),null); + checkJournal(f.store,frame.runId); +}); + + +test('S07 moderately large valid inline output is still accepted when whole Checkpoint has capacity',async t=>{ + const f=await fixture(t); + const frame=await waiting(f,{runId:'moderate-inline',startKey:'moderate-inline'}); + const completed=await f.api.complete(receipt(frame,{answer:'z'.repeat(256*1024)})); + assert.equal(completed.status,'succeeded'); + assert.equal(completed.state.answer.length,256*1024); + assert.equal(f.store.getReceipt(frame.runId,frame.pending.requestId).acceptedRevision, + completed.acceptedRevision); + checkJournal(f.store,frame.runId); +}); + + +test('S07 result alone fits hard JSON limit but full Agent receipt exceeds it: durable terminal, not raw RangeError waiting',async t=>{ + const f=await fixture(t); + const w=await waiting(f,{runId:'envelope-edge',startKey:'envelope-edge'}); + const result={answer:'q'.repeat(MAX_CHECKPOINT_BYTES-140)}; + assert.ok(canonicalBytes(result).byteLengthcanonicalBytes(envelope),/Canonical JSON exceeds maximum bytes/); + await assert.rejects(f.api.complete(response),{code:'FK_I_CHECKPOINT_CAPACITY'}); + const stopped=f.store.load(w.runId); + assert.equal(stopped.status,'errored'); + assert.equal(stopped.terminal.causeCode,'FK_I_CHECKPOINT_CAPACITY'); + assert.deepEqual(stopped.state,{message:'small'}); + assert.equal(f.store.getReceipt(w.runId,w.pending.requestId),null); + checkJournal(f.store,w.runId); +}); diff --git a/types/interactive.d.ts b/types/interactive.d.ts index c50bfc5..2a1caad 100644 --- a/types/interactive.d.ts +++ b/types/interactive.d.ts @@ -115,6 +115,18 @@ export interface CheckpointStore { journalEntry?: unknown; }): {committed:boolean;checkpoint:InteractiveCheckpoint}; listJournal(runId: string): unknown[]; + /** + * Optional execution-only receipt index (implemented by the default SQLite + * Store). Resolves acceptedRevision/digest after a receipt leaves the + * Checkpoint's bounded recent-receipts cache. + */ + getReceipt?(runId: string, requestId: string): { + requestId: string; + nodeId: string; + attempt: number; + digest: string; + acceptedRevision: number; + } | null; } export declare function createSqliteCheckpointStore( file: string, options: {namespace:string}