capture-store-blob-persist.test.mjs
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6
docs: record AIP-b SD-21 land (KN #308)
Human
10 days ago
| 1 | /** |
| 2 | * CAPTURE-STORE-BLOB-PERSIST — seven-tier regression suite. |
| 3 | * |
| 4 | * Root cause (found live 2026-07-31, prop-1785500300353491755): the hosted |
| 5 | * bridge capture routes (observe / candidates / propose / dismiss) wrote the |
| 6 | * flow store only to the lambda's ephemeral DATA_DIR. A candidate created by |
| 7 | * observe+propose evaporated when the warm lambda recycled, so the Hub-complete |
| 8 | * apply at approve time refused with FLOW_CANDIDATE_NOT_PROMOTABLE and no Flow |
| 9 | * was indexed even though the canister proposal was approved. |
| 10 | * |
| 11 | * These tiers boot the REAL registerBridgeFlowCaptureRoutes express app against |
| 12 | * an in-memory Netlify-Blobs stand-in and simulate separate lambda instances as |
| 13 | * separate DATA_DIRs sharing one blob store. Every tier fails against the |
| 14 | * pre-fix routes (no blob hydrate/persist on observe/candidates/propose/dismiss). |
| 15 | * |
| 16 | * Tiers: unit, integration, e2e, stress, data-integrity, performance, security. |
| 17 | */ |
| 18 | |
| 19 | import { describe, it, before, beforeEach, after } from 'node:test'; |
| 20 | import assert from 'node:assert/strict'; |
| 21 | import fs from 'node:fs'; |
| 22 | import os from 'node:os'; |
| 23 | import path from 'node:path'; |
| 24 | import http from 'node:http'; |
| 25 | import { performance } from 'node:perf_hooks'; |
| 26 | |
| 27 | import express from 'express'; |
| 28 | |
| 29 | import { registerBridgeFlowCaptureRoutes } from '../hub/bridge/flow-capture-routes.mjs'; |
| 30 | import { |
| 31 | externalProtocolBlobKey, |
| 32 | mergeFlowStoreJson, |
| 33 | } from '../hub/bridge/external-agent-blob-store.mjs'; |
| 34 | import { FLOW_STORE_FILENAME } from '../lib/flow/flow-store.mjs'; |
| 35 | import { validSessionMeta } from './fixtures/flow/capture-helpers.mjs'; |
| 36 | |
| 37 | const tmpRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'capture-blob-persist-')); |
| 38 | const ACTOR = 'google:learner-persist'; |
| 39 | const VAULT = 'default'; |
| 40 | const FLOW_STORE_BLOB_KEY = externalProtocolBlobKey(FLOW_STORE_FILENAME); |
| 41 | |
| 42 | /** In-memory Netlify-Blobs stand-in shared across simulated lambda instances. */ |
| 43 | function fakeBlobStore(initial = {}) { |
| 44 | const store = new Map(Object.entries(initial)); |
| 45 | const sets = []; |
| 46 | return { |
| 47 | store, |
| 48 | sets, |
| 49 | get: async (key) => (store.has(key) ? store.get(key) : null), |
| 50 | set: async (key, value) => { |
| 51 | sets.push(key); |
| 52 | store.set(key, value); |
| 53 | }, |
| 54 | }; |
| 55 | } |
| 56 | |
| 57 | /** |
| 58 | * Mock canister: stores POSTed proposals, serves GETs, flips status on approve. |
| 59 | * Mirrors the row shape parseCanisterProposalGetBody expects. |
| 60 | */ |
| 61 | function startMockCanister() { |
| 62 | const rows = new Map(); |
| 63 | let seq = 0; |
| 64 | const app = express(); |
| 65 | app.use(express.json()); |
| 66 | app.post('/api/v1/proposals', (req, res) => { |
| 67 | seq += 1; |
| 68 | const id = `prop-persist-${seq}`; |
| 69 | rows.set(id, { |
| 70 | proposal_id: id, |
| 71 | status: 'proposed', |
| 72 | path: req.body.path, |
| 73 | body: req.body.body ?? '', |
| 74 | intent: req.body.intent ?? '', |
| 75 | frontmatter: req.body.frontmatter ?? {}, |
| 76 | base_state_id: req.body.base_state_id ?? '', |
| 77 | vault_id: VAULT, |
| 78 | }); |
| 79 | res.status(201).json({ proposal_id: id, path: req.body.path, status: 'proposed' }); |
| 80 | }); |
| 81 | app.get('/api/v1/proposals/:id', (req, res) => { |
| 82 | const row = rows.get(req.params.id); |
| 83 | if (!row) return res.status(404).json({ error: 'Not found', code: 'NOT_FOUND' }); |
| 84 | res.json(row); |
| 85 | }); |
| 86 | const srv = http.createServer(app); |
| 87 | return new Promise((resolve) => { |
| 88 | srv.listen(0, '127.0.0.1', () => { |
| 89 | resolve({ |
| 90 | rows, |
| 91 | url: `http://127.0.0.1:${srv.address().port}`, |
| 92 | close: () => new Promise((r) => srv.close(() => r())), |
| 93 | }); |
| 94 | }); |
| 95 | }); |
| 96 | } |
| 97 | |
| 98 | /** |
| 99 | * Boot ONE simulated lambda instance: real capture routes, own DATA_DIR, |
| 100 | * shared blob store. Auth/context deps are stubbed (auth is not under test). |
| 101 | */ |
| 102 | function startInstance({ dataDir, blobStore, canisterUrl, role = 'admin' }) { |
| 103 | fs.mkdirSync(dataDir, { recursive: true }); |
| 104 | const app = express(); |
| 105 | app.use(express.json()); |
| 106 | registerBridgeFlowCaptureRoutes(app, { |
| 107 | dataDir, |
| 108 | canisterUrl, |
| 109 | canisterHeaders: (extra = {}) => ({ ...extra }), |
| 110 | requireBridgeAuth: (req, _res, next) => { |
| 111 | req.uid = ACTOR; |
| 112 | req.blobStore = blobStore; |
| 113 | next(); |
| 114 | }, |
| 115 | resolveHostedBridgeContext: async (_req, actorUid) => ({ |
| 116 | ok: true, |
| 117 | vaultId: VAULT, |
| 118 | effectiveCanisterUid: actorUid, |
| 119 | actorUid, |
| 120 | }), |
| 121 | effectiveRole: () => role, |
| 122 | loadRoles: async () => ({}), |
| 123 | }); |
| 124 | const srv = http.createServer(app); |
| 125 | return new Promise((resolve) => { |
| 126 | srv.listen(0, '127.0.0.1', () => { |
| 127 | resolve({ |
| 128 | url: `http://127.0.0.1:${srv.address().port}`, |
| 129 | close: () => new Promise((r) => srv.close(() => r())), |
| 130 | }); |
| 131 | }); |
| 132 | }); |
| 133 | } |
| 134 | |
| 135 | /** JSON helper against an instance. */ |
| 136 | async function call(instance, method, route, body) { |
| 137 | const res = await fetch(`${instance.url}${route}`, { |
| 138 | method, |
| 139 | headers: { 'Content-Type': 'application/json', Accept: 'application/json' }, |
| 140 | ...(body !== undefined ? { body: JSON.stringify(body) } : {}), |
| 141 | }); |
| 142 | const text = await res.text(); |
| 143 | let json = {}; |
| 144 | try { |
| 145 | json = text ? JSON.parse(text) : {}; |
| 146 | } catch { |
| 147 | json = { raw: text }; |
| 148 | } |
| 149 | return { status: res.status, json }; |
| 150 | } |
| 151 | |
| 152 | /** Observe on an instance and return the created candidate_id. */ |
| 153 | async function observeCandidate(instance, sessionId) { |
| 154 | const r = await call(instance, 'POST', '/api/v1/flows/capture/observe', { |
| 155 | ...validSessionMeta(sessionId ? { session_id: sessionId } : {}), |
| 156 | harness: 'test', |
| 157 | }); |
| 158 | assert.equal(r.status, 200, `observe failed: ${JSON.stringify(r.json)}`); |
| 159 | assert.equal(r.json.detection_authorized, true); |
| 160 | assert.ok(r.json.candidates.length >= 1, 'observe returned no candidates'); |
| 161 | return r.json.candidates[0].candidate_id; |
| 162 | } |
| 163 | |
| 164 | let instanceSeq = 0; |
| 165 | /** Fresh DATA_DIR per simulated cold lambda. */ |
| 166 | function freshDataDir() { |
| 167 | instanceSeq += 1; |
| 168 | return path.join(tmpRoot, `instance-${instanceSeq}`); |
| 169 | } |
| 170 | |
| 171 | let canister; |
| 172 | before(async () => { |
| 173 | process.env.FLOW_CAPTURE_DETECTION_ENABLED = '1'; |
| 174 | process.env.FLOW_CAPTURE_WRITES_ENABLED = '1'; |
| 175 | canister = await startMockCanister(); |
| 176 | }); |
| 177 | after(async () => { |
| 178 | delete process.env.FLOW_CAPTURE_DETECTION_ENABLED; |
| 179 | delete process.env.FLOW_CAPTURE_WRITES_ENABLED; |
| 180 | await canister.close(); |
| 181 | fs.rmSync(tmpRoot, { recursive: true, force: true }); |
| 182 | }); |
| 183 | |
| 184 | // --------------------------------------------------------------------------- |
| 185 | |
| 186 | describe('CAPTURE-STORE-BLOB-PERSIST — unit', () => { |
| 187 | it('observe persists the flow store (with the candidate) to the blob store', async () => { |
| 188 | const blob = fakeBlobStore(); |
| 189 | const a = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 190 | try { |
| 191 | const candidateId = await observeCandidate(a); |
| 192 | const raw = blob.store.get(FLOW_STORE_BLOB_KEY); |
| 193 | assert.ok(typeof raw === 'string' && raw.includes(candidateId), |
| 194 | 'flow store blob missing the observed candidate (pre-fix regression)'); |
| 195 | } finally { |
| 196 | await a.close(); |
| 197 | } |
| 198 | }); |
| 199 | |
| 200 | it('candidates list hydrates from blob on a cold instance', async () => { |
| 201 | const blob = fakeBlobStore(); |
| 202 | const a = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 203 | let candidateId; |
| 204 | try { |
| 205 | candidateId = await observeCandidate(a); |
| 206 | } finally { |
| 207 | await a.close(); |
| 208 | } |
| 209 | const b = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 210 | try { |
| 211 | const r = await call(b, 'GET', '/api/v1/flows/candidates'); |
| 212 | assert.equal(r.status, 200); |
| 213 | const ids = (r.json.candidates || []).map((c) => c.candidate_id); |
| 214 | assert.ok(ids.includes(candidateId), |
| 215 | 'cold instance did not hydrate candidates from blob (pre-fix regression)'); |
| 216 | } finally { |
| 217 | await b.close(); |
| 218 | } |
| 219 | }); |
| 220 | }); |
| 221 | |
| 222 | describe('CAPTURE-STORE-BLOB-PERSIST — unit (warm-lambda stale merge)', () => { |
| 223 | // Live failure 2026-07-31 (prop-1785526040570098296): a WARM lambda whose |
| 224 | // local store predated the candidate hydrated the blob, but the merge let the |
| 225 | // stale local candidates array mask the blob's — apply refused |
| 226 | // FLOW_CANDIDATE_NOT_PROMOTABLE while a cold-lambda retry succeeded. |
| 227 | it('blob-only candidates/flows survive a merge against a stale local store', () => { |
| 228 | const staleLocal = JSON.stringify({ |
| 229 | vaults: { |
| 230 | [VAULT]: { flows: [], steps: [], runs: [], candidates: [], tasks: [], task_loops: [] }, |
| 231 | }, |
| 232 | }); |
| 233 | const blob = JSON.stringify({ |
| 234 | vaults: { |
| 235 | [VAULT]: { |
| 236 | candidates: [ |
| 237 | { candidate_id: 'cand_blob_only', status: 'pending_review', updated: '2026-07-31T19:00:00Z' }, |
| 238 | ], |
| 239 | flows: [ |
| 240 | { flow_id: 'flow_blob_only', version: '0.1.0', updated: '2026-07-31T19:00:00Z' }, |
| 241 | ], |
| 242 | steps: [ |
| 243 | { step_id: 'step_1', flow_id: 'flow_blob_only', flow_version: '0.1.0' }, |
| 244 | ], |
| 245 | runs: [{ run_id: 'run_blob_only' }], |
| 246 | tasks: [], |
| 247 | task_loops: [], |
| 248 | }, |
| 249 | }, |
| 250 | }); |
| 251 | const merged = JSON.parse(mergeFlowStoreJson(staleLocal, blob)); |
| 252 | const vault = merged.vaults[VAULT]; |
| 253 | assert.equal(vault.candidates.length, 1, 'stale local candidates masked blob candidate'); |
| 254 | assert.equal(vault.candidates[0].candidate_id, 'cand_blob_only'); |
| 255 | assert.equal(vault.flows.length, 1, 'stale local flows masked blob flow'); |
| 256 | assert.equal(vault.steps.length, 1, 'stale local steps masked blob step'); |
| 257 | assert.equal(vault.runs.length, 1, 'stale local runs masked blob run'); |
| 258 | }); |
| 259 | |
| 260 | it('newer record wins on key collision; distinct flow versions both survive', () => { |
| 261 | const local = JSON.stringify({ |
| 262 | vaults: { |
| 263 | [VAULT]: { |
| 264 | candidates: [ |
| 265 | { candidate_id: 'cand_x', status: 'promoted', updated: '2026-07-31T20:00:00Z' }, |
| 266 | ], |
| 267 | flows: [{ flow_id: 'flow_x', version: '0.2.0', updated: '2026-07-31T20:00:00Z' }], |
| 268 | }, |
| 269 | }, |
| 270 | }); |
| 271 | const blob = JSON.stringify({ |
| 272 | vaults: { |
| 273 | [VAULT]: { |
| 274 | candidates: [ |
| 275 | { candidate_id: 'cand_x', status: 'pending_review', updated: '2026-07-31T19:00:00Z' }, |
| 276 | ], |
| 277 | flows: [{ flow_id: 'flow_x', version: '0.1.0', updated: '2026-07-31T19:00:00Z' }], |
| 278 | }, |
| 279 | }, |
| 280 | }); |
| 281 | const merged = JSON.parse(mergeFlowStoreJson(local, blob)); |
| 282 | const vault = merged.vaults[VAULT]; |
| 283 | assert.equal(vault.candidates.length, 1); |
| 284 | assert.equal(vault.candidates[0].status, 'promoted', 'older blob record overwrote newer local'); |
| 285 | assert.equal(vault.flows.length, 2, 'distinct flow versions collapsed'); |
| 286 | }); |
| 287 | }); |
| 288 | |
| 289 | describe('CAPTURE-STORE-BLOB-PERSIST — integration', () => { |
| 290 | it('propose on a cold instance finds the blob-persisted candidate and persists its state change', async () => { |
| 291 | const blob = fakeBlobStore(); |
| 292 | const a = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 293 | let candidateId; |
| 294 | try { |
| 295 | candidateId = await observeCandidate(a); |
| 296 | } finally { |
| 297 | await a.close(); |
| 298 | } |
| 299 | |
| 300 | const b = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 301 | try { |
| 302 | const r = await call(b, 'POST', `/api/v1/flows/candidates/${candidateId}/propose`, { |
| 303 | confirmed_scope: 'personal', |
| 304 | intent: 'promote across lambda instances', |
| 305 | }); |
| 306 | assert.equal(r.status, 201, `propose failed: ${JSON.stringify(r.json)}`); |
| 307 | assert.equal(r.json.candidate_id, candidateId); |
| 308 | // Propose keeps the candidate pending_review in the flow store (the |
| 309 | // pending-proposal guard reads the proposals store — canister-side when |
| 310 | // hosted). The regression property is that the candidate must still be |
| 311 | // present and promotable in the shared blob after the cold-instance call. |
| 312 | const parsed = JSON.parse(blob.store.get(FLOW_STORE_BLOB_KEY)); |
| 313 | const rec = parsed.vaults[VAULT].candidates.find((c) => c.candidate_id === candidateId); |
| 314 | assert.ok(rec, 'candidate lost from blob after cold-instance propose'); |
| 315 | assert.equal(rec.status, 'pending_review'); |
| 316 | } finally { |
| 317 | await b.close(); |
| 318 | } |
| 319 | }); |
| 320 | |
| 321 | it('dismiss on a cold instance finds the blob-persisted candidate', async () => { |
| 322 | const blob = fakeBlobStore(); |
| 323 | const a = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 324 | let candidateId; |
| 325 | try { |
| 326 | candidateId = await observeCandidate(a, 'd'.repeat(64)); |
| 327 | } finally { |
| 328 | await a.close(); |
| 329 | } |
| 330 | const b = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 331 | try { |
| 332 | const r = await call(b, 'POST', `/api/v1/flows/candidates/${candidateId}/dismiss`, { |
| 333 | intent: 'dismiss across lambda instances', |
| 334 | }); |
| 335 | assert.equal(r.status, 201, `dismiss failed: ${JSON.stringify(r.json)}`); |
| 336 | } finally { |
| 337 | await b.close(); |
| 338 | } |
| 339 | }); |
| 340 | }); |
| 341 | |
| 342 | describe('CAPTURE-STORE-BLOB-PERSIST — e2e', () => { |
| 343 | it('observe→propose→approve→apply-approved across four cold instances yields a listed Flow', async () => { |
| 344 | const blob = fakeBlobStore(); |
| 345 | |
| 346 | // Instance A: observe. |
| 347 | const a = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 348 | let candidateId; |
| 349 | try { |
| 350 | candidateId = await observeCandidate(a, 'e'.repeat(64)); |
| 351 | } finally { |
| 352 | await a.close(); |
| 353 | } |
| 354 | |
| 355 | // Instance B: propose (creates canister proposal). |
| 356 | const b = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 357 | let proposalId; |
| 358 | try { |
| 359 | const r = await call(b, 'POST', `/api/v1/flows/candidates/${candidateId}/propose`, { |
| 360 | confirmed_scope: 'personal', |
| 361 | intent: 'e2e promote', |
| 362 | }); |
| 363 | assert.equal(r.status, 201, `propose failed: ${JSON.stringify(r.json)}`); |
| 364 | proposalId = r.json.proposal_id; |
| 365 | } finally { |
| 366 | await b.close(); |
| 367 | } |
| 368 | |
| 369 | // Approve on the (durable) canister — the operator's Hub click. |
| 370 | canister.rows.get(proposalId).status = 'approved'; |
| 371 | |
| 372 | // Instance C: Hub-complete apply (this exact call failed live pre-fix). |
| 373 | const c = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 374 | let flowId; |
| 375 | try { |
| 376 | const r = await call(c, 'POST', `/api/v1/flows/capture/proposals/${proposalId}/apply-approved`, {}); |
| 377 | assert.equal(r.status, 200, |
| 378 | `apply-approved refused (${r.json.code}) — live bug regression`); |
| 379 | assert.equal(r.json.proposal_kind, 'flow_candidate_promote'); |
| 380 | flowId = r.json.flow_id; |
| 381 | assert.ok(flowId, 'apply payload missing flow_id'); |
| 382 | } finally { |
| 383 | await c.close(); |
| 384 | } |
| 385 | |
| 386 | // Instance D: the promoted Flow is observable (CHA-C5). |
| 387 | const d = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 388 | try { |
| 389 | const list = await call(d, 'GET', '/api/v1/flows'); |
| 390 | assert.equal(list.status, 200); |
| 391 | const ids = (list.json.flows || []).map((f) => f.flow_id || f.id); |
| 392 | assert.ok(ids.includes(flowId), 'promoted flow not listed on cold instance'); |
| 393 | const one = await call(d, 'GET', `/api/v1/flows/${flowId}`); |
| 394 | assert.equal(one.status, 200); |
| 395 | } finally { |
| 396 | await d.close(); |
| 397 | } |
| 398 | }); |
| 399 | }); |
| 400 | |
| 401 | describe('CAPTURE-STORE-BLOB-PERSIST — e2e (warm stale lambda)', () => { |
| 402 | it('apply-approved on a WARM instance with a stale local store still finds the blob candidate', async () => { |
| 403 | const blob = fakeBlobStore(); |
| 404 | |
| 405 | // Instance A: observe + propose (candidate + proposal in blob/canister). |
| 406 | const a = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 407 | let candidateId; |
| 408 | let proposalId; |
| 409 | try { |
| 410 | candidateId = await observeCandidate(a, '7'.repeat(64)); |
| 411 | const r = await call(a, 'POST', `/api/v1/flows/candidates/${candidateId}/propose`, { |
| 412 | confirmed_scope: 'personal', |
| 413 | intent: 'warm stale lambda regression', |
| 414 | }); |
| 415 | assert.equal(r.status, 201, `propose failed: ${JSON.stringify(r.json)}`); |
| 416 | proposalId = r.json.proposal_id; |
| 417 | } finally { |
| 418 | await a.close(); |
| 419 | } |
| 420 | canister.rows.get(proposalId).status = 'approved'; |
| 421 | |
| 422 | // Instance B: WARM — its local DATA_DIR already holds a stale flow store |
| 423 | // (vault exists, candidate absent), exactly the live 2026-07-31 failure. |
| 424 | const staleDir = freshDataDir(); |
| 425 | fs.mkdirSync(staleDir, { recursive: true }); |
| 426 | fs.writeFileSync( |
| 427 | path.join(staleDir, FLOW_STORE_FILENAME), |
| 428 | JSON.stringify({ |
| 429 | vaults: { |
| 430 | [VAULT]: { flows: [], steps: [], runs: [], candidates: [], projections: [], tasks: [], task_loops: [] }, |
| 431 | }, |
| 432 | }), |
| 433 | 'utf8', |
| 434 | ); |
| 435 | const b = await startInstance({ dataDir: staleDir, blobStore: blob, canisterUrl: canister.url }); |
| 436 | try { |
| 437 | const r = await call(b, 'POST', `/api/v1/flows/capture/proposals/${proposalId}/apply-approved`, {}); |
| 438 | assert.equal(r.status, 200, |
| 439 | `warm stale lambda refused apply (${r.json.code}) — live 2026-07-31 regression`); |
| 440 | assert.ok(r.json.flow_id, 'apply payload missing flow_id'); |
| 441 | } finally { |
| 442 | await b.close(); |
| 443 | } |
| 444 | }); |
| 445 | }); |
| 446 | |
| 447 | describe('CAPTURE-STORE-BLOB-PERSIST — stress', () => { |
| 448 | it('12 candidates across alternating instances all survive', async () => { |
| 449 | const blob = fakeBlobStore(); |
| 450 | const created = []; |
| 451 | for (let i = 0; i < 12; i += 1) { |
| 452 | const inst = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 453 | try { |
| 454 | created.push(await observeCandidate(inst, String(i % 10).repeat(64))); |
| 455 | } finally { |
| 456 | await inst.close(); |
| 457 | } |
| 458 | } |
| 459 | const reader = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 460 | try { |
| 461 | const r = await call(reader, 'GET', '/api/v1/flows/candidates?limit=50'); |
| 462 | assert.equal(r.status, 200); |
| 463 | const ids = new Set((r.json.candidates || []).map((c) => c.candidate_id)); |
| 464 | for (const id of created) { |
| 465 | assert.ok(ids.has(id), `candidate ${id} lost across instances`); |
| 466 | } |
| 467 | } finally { |
| 468 | await reader.close(); |
| 469 | } |
| 470 | }); |
| 471 | }); |
| 472 | |
| 473 | describe('CAPTURE-STORE-BLOB-PERSIST — data-integrity', () => { |
| 474 | it('blob flow store stays valid JSON and candidate fields survive the round-trip intact', async () => { |
| 475 | const blob = fakeBlobStore(); |
| 476 | const a = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 477 | let candidateId; |
| 478 | try { |
| 479 | candidateId = await observeCandidate(a, 'f'.repeat(64)); |
| 480 | } finally { |
| 481 | await a.close(); |
| 482 | } |
| 483 | |
| 484 | const raw = blob.store.get(FLOW_STORE_BLOB_KEY); |
| 485 | const parsed = JSON.parse(raw); |
| 486 | const rec = parsed.vaults[VAULT].candidates.find((c) => c.candidate_id === candidateId); |
| 487 | assert.ok(rec, 'candidate record absent from blob JSON'); |
| 488 | assert.equal(rec.status, 'pending_review'); |
| 489 | assert.equal(rec.schema, 'knowtation.flow_candidate/v0'); |
| 490 | assert.ok(Array.isArray(rec.draft_steps) && rec.draft_steps.length > 0); |
| 491 | |
| 492 | const b = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 493 | try { |
| 494 | const r = await call(b, 'GET', '/api/v1/flows/candidates'); |
| 495 | const back = (r.json.candidates || []).find((c) => c.candidate_id === candidateId); |
| 496 | assert.ok(back, 'candidate missing after hydration'); |
| 497 | assert.equal(back.status, 'pending_review'); |
| 498 | } finally { |
| 499 | await b.close(); |
| 500 | } |
| 501 | }); |
| 502 | |
| 503 | it('does not clobber unrelated vault state already in the blob (merge, not overwrite)', async () => { |
| 504 | const seeded = { |
| 505 | version: 1, |
| 506 | vaults: { |
| 507 | other_vault: { flows: [], candidates: [], tasks: [{ task_id: 't-keep', updated: '2026-07-30T00:00:00Z' }] }, |
| 508 | }, |
| 509 | }; |
| 510 | const blob = fakeBlobStore({ [FLOW_STORE_BLOB_KEY]: JSON.stringify(seeded) }); |
| 511 | const a = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 512 | try { |
| 513 | await observeCandidate(a, 'a'.repeat(64)); |
| 514 | } finally { |
| 515 | await a.close(); |
| 516 | } |
| 517 | const parsed = JSON.parse(blob.store.get(FLOW_STORE_BLOB_KEY)); |
| 518 | assert.ok(parsed.vaults.other_vault, 'unrelated vault dropped from blob'); |
| 519 | assert.equal(parsed.vaults.other_vault.tasks[0].task_id, 't-keep'); |
| 520 | }); |
| 521 | }); |
| 522 | |
| 523 | describe('CAPTURE-STORE-BLOB-PERSIST — performance', () => { |
| 524 | it('observe with blob sync stays under 250ms p95 (in-memory blob)', async () => { |
| 525 | const blob = fakeBlobStore(); |
| 526 | const inst = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 527 | try { |
| 528 | const samples = []; |
| 529 | for (let i = 0; i < 20; i += 1) { |
| 530 | const t0 = performance.now(); |
| 531 | await call(inst, 'POST', '/api/v1/flows/capture/observe', { |
| 532 | ...validSessionMeta({ session_id: String(i % 10).repeat(64) }), |
| 533 | harness: 'test', |
| 534 | }); |
| 535 | samples.push(performance.now() - t0); |
| 536 | } |
| 537 | samples.sort((x, y) => x - y); |
| 538 | const p95 = samples[Math.floor(samples.length * 0.95) - 1] ?? samples[samples.length - 1]; |
| 539 | assert.ok(p95 < 250, `observe p95=${p95.toFixed(1)}ms exceeds 250ms budget`); |
| 540 | } finally { |
| 541 | await inst.close(); |
| 542 | } |
| 543 | }); |
| 544 | }); |
| 545 | |
| 546 | describe('CAPTURE-STORE-BLOB-PERSIST — security', () => { |
| 547 | it('refused propose (wrong scope) does not mutate the blob store', async () => { |
| 548 | const blob = fakeBlobStore(); |
| 549 | const a = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 550 | let candidateId; |
| 551 | try { |
| 552 | candidateId = await observeCandidate(a, '9'.repeat(64)); |
| 553 | } finally { |
| 554 | await a.close(); |
| 555 | } |
| 556 | const before = blob.store.get(FLOW_STORE_BLOB_KEY); |
| 557 | |
| 558 | const b = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 559 | try { |
| 560 | const r = await call(b, 'POST', `/api/v1/flows/candidates/${candidateId}/propose`, { |
| 561 | confirmed_scope: 'org', |
| 562 | intent: 'scope widen without ack must refuse', |
| 563 | }); |
| 564 | assert.ok(r.status >= 400, `expected refusal, got ${r.status}`); |
| 565 | assert.equal(blob.store.get(FLOW_STORE_BLOB_KEY), before, |
| 566 | 'refused propose mutated the persisted store'); |
| 567 | } finally { |
| 568 | await b.close(); |
| 569 | } |
| 570 | }); |
| 571 | |
| 572 | it('persisted blob carries no raw session content (content-minimized candidates only)', async () => { |
| 573 | const blob = fakeBlobStore(); |
| 574 | const a = await startInstance({ dataDir: freshDataDir(), blobStore: blob, canisterUrl: canister.url }); |
| 575 | try { |
| 576 | const r = await call(a, 'POST', '/api/v1/flows/capture/observe', { |
| 577 | ...validSessionMeta({ session_id: '8'.repeat(64) }), |
| 578 | prompt: 'RAW PROMPT MUST NOT PERSIST', |
| 579 | completion: 'RAW COMPLETION MUST NOT PERSIST', |
| 580 | harness: 'test', |
| 581 | }); |
| 582 | // Payload-bearing meta is refused by the handler; if a variant were accepted, |
| 583 | // the persisted blob still must not carry raw content. |
| 584 | const raw = blob.store.get(FLOW_STORE_BLOB_KEY) || ''; |
| 585 | assert.ok(!raw.includes('RAW PROMPT MUST NOT PERSIST'), 'raw prompt leaked into blob'); |
| 586 | assert.ok(!raw.includes('RAW COMPLETION MUST NOT PERSIST'), 'raw completion leaked into blob'); |
| 587 | assert.ok(r.status === 200 || r.status === 400); |
| 588 | } finally { |
| 589 | await a.close(); |
| 590 | } |
| 591 | }); |
| 592 | }); |
File History
1 commit
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6
docs: record AIP-b SD-21 land (KN #308)
Human
10 days ago