aaronrene / knowtation public
external-agent-blob-store.mjs
256 lines 8.7 KB
Raw
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6 docs: record AIP-b SD-21 land (KN #308) Human 12 days ago
1 /**
2 * Hosted bridge: persist external protocol state in Netlify Blobs.
3 *
4 * Self-hosted bridge uses DATA_DIR files only. On Netlify, DATA_DIR is ephemeral;
5 * this module hydrates files from Blobs before protocol handlers and persists after writes.
6 */
7
8 import fs from 'fs';
9 import path from 'path';
10 import { FLOW_STORE_FILENAME } from '../../lib/flow/flow-store.mjs';
11 import { hydrateDelegationStoresFromBlob } from './delegation-blob-store.mjs';
12
13 /** @typedef {{ get: (key: string, opts?: { type?: string }) => Promise<string|ArrayBuffer|null>, set: (key: string, value: string) => Promise<void> }} BlobStore */
14
15 export const EXTERNAL_PROTOCOL_BLOB_FILES = [
16 FLOW_STORE_FILENAME,
17 'hub_external_protocol_idempotency.json',
18 'hub_delegation_audit.json',
19 ];
20
21 /**
22 * @param {string} filename
23 * @returns {string}
24 */
25 export function externalProtocolBlobKey(filename) {
26 return `external-protocol/${filename}`;
27 }
28
29 /**
30 * Merge local and blob flow stores without losing fresher task writes.
31 * Task apply may land on a warm lambda before the external-protocol blob reflects it.
32 *
33 * @param {string} localRaw
34 * @param {string} blobRaw
35 * @returns {string}
36 */
37 export function mergeFlowStoreJson(localRaw, blobRaw) {
38 const parse = (raw) => {
39 if (typeof raw !== 'string' || !raw.trim()) return null;
40 try {
41 return JSON.parse(raw);
42 } catch {
43 return null;
44 }
45 };
46
47 const local = parse(localRaw);
48 const blob = parse(blobRaw);
49 if (!local && !blob) return blobRaw || localRaw || '';
50 if (!local) return blobRaw;
51 if (!blob) return localRaw;
52
53 /**
54 * Union local and blob records by natural key; on collision the newer
55 * `updated`/`created` wins (local wins ties). Prevents a warm lambda's stale
56 * local array from masking records another instance persisted to Blobs.
57 *
58 * @param {unknown[]} localArr
59 * @param {unknown[]} blobArr
60 * @param {(rec: Record<string, unknown>) => string|null} keyOf
61 */
62 const mergeByKey = (localArr, blobArr, keyOf) => {
63 /** @type {Map<string, Record<string, unknown>>} */
64 const byKey = new Map();
65 for (const rec of blobArr || []) {
66 if (!rec || typeof rec !== 'object') continue;
67 const key = keyOf(/** @type {Record<string, unknown>} */ (rec));
68 if (key != null) byKey.set(key, /** @type {Record<string, unknown>} */ (rec));
69 }
70 for (const rec of localArr || []) {
71 if (!rec || typeof rec !== 'object') continue;
72 const row = /** @type {Record<string, unknown>} */ (rec);
73 const key = keyOf(row);
74 if (key == null) continue;
75 const existing = byKey.get(key);
76 if (!existing) {
77 byKey.set(key, row);
78 continue;
79 }
80 const tLocal = Date.parse(String(row.updated || row.created || '')) || 0;
81 const tBlob = Date.parse(String(existing.updated || existing.created || '')) || 0;
82 byKey.set(key, tLocal >= tBlob ? row : existing);
83 }
84 return [...byKey.values()];
85 };
86
87 /** @param {string} field @returns {(rec: Record<string, unknown>) => string|null} */
88 const stringKey = (field) => (rec) => (typeof rec[field] === 'string' ? rec[field] : null);
89 const mergeById = (localArr, blobArr, idField) => mergeByKey(localArr, blobArr, stringKey(idField));
90
91 /** Flows are versioned: one record per (flow_id, version). */
92 const flowKey = (rec) =>
93 typeof rec.flow_id === 'string' ? `${rec.flow_id}\0${typeof rec.version === 'string' ? rec.version : ''}` : null;
94 /** Steps key on (flow_id, flow_version, step_id) — 7A-10c store shape. */
95 const stepKey = (rec) =>
96 typeof rec.step_id === 'string' && typeof rec.flow_id === 'string'
97 ? `${rec.flow_id}\0${typeof rec.flow_version === 'string' ? rec.flow_version : ''}\0${rec.step_id}`
98 : null;
99
100 if (!local.vaults) local.vaults = {};
101 for (const [vaultId, blobVault] of Object.entries(blob.vaults || {})) {
102 const localVault =
103 local.vaults[vaultId] && typeof local.vaults[vaultId] === 'object'
104 ? /** @type {Record<string, unknown>} */ (local.vaults[vaultId])
105 : {};
106 const blobVaultObj =
107 blobVault && typeof blobVault === 'object'
108 ? /** @type {Record<string, unknown>} */ (blobVault)
109 : {};
110 // candidates/flows must merge by id too: a warm lambda's stale local store
111 // otherwise masks blob records written by another instance (this made the
112 // approve-time capture apply refuse FLOW_CANDIDATE_NOT_PROMOTABLE while a
113 // cold-lambda retry of the same apply succeeded — 2026-07-31 live).
114 local.vaults[vaultId] = {
115 ...blobVaultObj,
116 ...localVault,
117 tasks: mergeById(
118 /** @type {unknown[]} */ (localVault.tasks),
119 /** @type {unknown[]} */ (blobVaultObj.tasks),
120 'task_id',
121 ),
122 task_loops: mergeById(
123 /** @type {unknown[]} */ (localVault.task_loops),
124 /** @type {unknown[]} */ (blobVaultObj.task_loops),
125 'loop_id',
126 ),
127 candidates: mergeById(
128 /** @type {unknown[]} */ (localVault.candidates),
129 /** @type {unknown[]} */ (blobVaultObj.candidates),
130 'candidate_id',
131 ),
132 flows: mergeByKey(
133 /** @type {unknown[]} */ (localVault.flows),
134 /** @type {unknown[]} */ (blobVaultObj.flows),
135 flowKey,
136 ),
137 steps: mergeByKey(
138 /** @type {unknown[]} */ (localVault.steps),
139 /** @type {unknown[]} */ (blobVaultObj.steps),
140 stepKey,
141 ),
142 runs: mergeById(
143 /** @type {unknown[]} */ (localVault.runs),
144 /** @type {unknown[]} */ (blobVaultObj.runs),
145 'run_id',
146 ),
147 learning_paths: mergeById(
148 /** @type {unknown[]} */ (localVault.learning_paths),
149 /** @type {unknown[]} */ (blobVaultObj.learning_paths),
150 'path_id',
151 ),
152 };
153 }
154
155 return JSON.stringify(local);
156 }
157
158 /**
159 * Load external protocol store files from Blobs into DATA_DIR (hosted cold-start hydration).
160 *
161 * @param {BlobStore|null|undefined} blobStore
162 * @param {string} dataDir
163 */
164 export async function hydrateExternalProtocolStoresFromBlob(blobStore, dataDir) {
165 if (!blobStore || typeof blobStore.get !== 'function') return;
166 fs.mkdirSync(dataDir, { recursive: true });
167 for (const filename of EXTERNAL_PROTOCOL_BLOB_FILES) {
168 const fp = path.join(dataDir, filename);
169 try {
170 const raw = await blobStore.get(externalProtocolBlobKey(filename), { type: 'text' });
171 if (typeof raw === 'string' && raw.trim()) {
172 if (filename === FLOW_STORE_FILENAME && fs.existsSync(fp)) {
173 const localRaw = fs.readFileSync(fp, 'utf8');
174 const merged = mergeFlowStoreJson(localRaw, raw);
175 if (merged.trim()) {
176 fs.writeFileSync(fp, merged, 'utf8');
177 }
178 } else {
179 fs.writeFileSync(fp, raw, 'utf8');
180 }
181 }
182 } catch {
183 /* keep existing file or empty */
184 }
185 }
186 }
187
188 /**
189 * Write external protocol store files from DATA_DIR to Blobs after a mutation.
190 *
191 * @param {BlobStore|null|undefined} blobStore
192 * @param {string} dataDir
193 */
194 export async function persistExternalProtocolStoresToBlob(blobStore, dataDir) {
195 if (!blobStore || typeof blobStore.set !== 'function') return;
196 for (const filename of EXTERNAL_PROTOCOL_BLOB_FILES) {
197 const fp = path.join(dataDir, filename);
198 if (!fs.existsSync(fp)) continue;
199 try {
200 const raw = fs.readFileSync(fp, 'utf8');
201 if (raw.trim()) {
202 await blobStore.set(externalProtocolBlobKey(filename), raw);
203 }
204 } catch {
205 /* non-fatal */
206 }
207 }
208 }
209
210 /**
211 * Run an external protocol mutation with hosted Blob hydrate/persist when available.
212 * Pushes them back if changed.
213 *
214 * @template T
215 * @param {{
216 * blobStore: BlobStore|null|undefined,
217 * dataDir: string,
218 * run: () => T | Promise<T>,
219 * }} opts
220 * @returns {Promise<T>}
221 */
222 export async function withExternalProtocolBlobSync(opts) {
223 if (!opts.blobStore || typeof opts.blobStore.get !== 'function') {
224 return opts.run();
225 }
226
227 await hydrateExternalProtocolStoresFromBlob(opts.blobStore, opts.dataDir);
228 await hydrateDelegationStoresFromBlob(opts.blobStore, opts.dataDir);
229
230 // Snapshot before run to avoid unnecessary blob writes if unmodified
231 const before = new Map();
232 for (const filename of EXTERNAL_PROTOCOL_BLOB_FILES) {
233 const fp = path.join(opts.dataDir, filename);
234 if (fs.existsSync(fp)) {
235 try { before.set(filename, fs.readFileSync(fp, 'utf8')); } catch {}
236 }
237 }
238
239 const result = await opts.run();
240
241 // Persist if changed
242 for (const filename of EXTERNAL_PROTOCOL_BLOB_FILES) {
243 const fp = path.join(opts.dataDir, filename);
244 if (!fs.existsSync(fp)) continue;
245 try {
246 const raw = fs.readFileSync(fp, 'utf8');
247 if (raw.trim() && raw !== before.get(filename)) {
248 await opts.blobStore.set(externalProtocolBlobKey(filename), raw);
249 }
250 } catch {
251 /* non-fatal */
252 }
253 }
254
255 return result;
256 }
File History 1 commit
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6 docs: record AIP-b SD-21 land (KN #308) Human 12 days ago