aaronrene / knowtation public
flow-capture-routes.mjs
377 lines 13.9 KB
Raw
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6 docs: record AIP-b SD-21 land (KN #308) Human 11 days ago
1 /**
2 * Hosted bridge REST routes for Flow capture flywheel
3 * (FLOW-CAPTURE-LIVE-KN-b — FCL-C10 parity with FLOW-WRITE-LIVE-GATEWAY-PROXY).
4 *
5 * Gated by FLOW_CAPTURE_DETECTION_ENABLED / FLOW_CAPTURE_WRITES_ENABLED
6 * (both default OFF → empty observe / 403 FLOW_CAPTURE_*_DISABLED).
7 * Does NOT flip those envs. Does NOT admit T5 self-apply for flow_capture.
8 *
9 * Every mutating route runs inside withExternalProtocolBlobSync and every read
10 * hydrates from Blobs first: hosted Netlify lambdas have ephemeral DATA_DIR, so
11 * a candidate written by observe/propose on one instance must survive to the
12 * approve-time apply on another (CAPTURE-STORE-BLOB-PERSIST fix — without this,
13 * apply refuses with FLOW_CANDIDATE_NOT_PROMOTABLE once the warm lambda recycles).
14 *
15 * @see docs/FLOW-CAPTURE-FLYWHEEL-CONTRACT-7A-L4.md
16 * @see hub/bridge/flow-routes.mjs
17 */
18
19 import {
20 handleFlowCaptureObserveRequest,
21 handleFlowCaptureListRequest,
22 handleFlowCaptureProposeRequest,
23 handleFlowCaptureDismissRequest,
24 } from '../../lib/flow/flow-capture.mjs';
25 import { createCaptureProposalOnCanister } from '../../lib/flow/flow-capture-hosted-proposal.mjs';
26 import { applyApprovedCaptureProposalFromCanister } from '../../lib/flow/flow-capture-hosted-apply.mjs';
27 import { handleFlowListRequest, handleFlowGetRequest } from '../../lib/flow/flow-handlers.mjs';
28 import {
29 withExternalProtocolBlobSync,
30 hydrateExternalProtocolStoresFromBlob,
31 } from './external-agent-blob-store.mjs';
32 import { isSessionBoundActor } from '../gateway/access-token-authz.mjs';
33 import { verifyJwtWithSecretRotation } from '../lib/session-secret-rotation.mjs';
34
35 /**
36 * Map bridge role to capture handler role (member → editor, matching self-hosted hub).
37 *
38 * @param {string} role
39 * @returns {string}
40 */
41 export function bridgeFlowCaptureHandlerRole(role) {
42 const r = typeof role === 'string' ? role.trim().toLowerCase() : '';
43 return r === 'member' || !r ? 'editor' : r;
44 }
45
46 /**
47 * @param {import('express').Express} app
48 * @param {{
49 * dataDir: string,
50 * canisterUrl: string,
51 * canisterHeaders: (extra?: Record<string, string>) => Record<string, string>,
52 * requireBridgeAuth: import('express').RequestHandler,
53 * resolveHostedBridgeContext: (req: import('express').Request, actorUid: string) => Promise<{
54 * ok: boolean,
55 * status?: number,
56 * error?: string,
57 * code?: string,
58 * vaultId?: string,
59 * effectiveCanisterUid?: string,
60 * actorUid?: string,
61 * }>,
62 * effectiveRole: (uid: string, storedRoles: Record<string, string>) => string,
63 * loadRoles: (blobStore: unknown) => Promise<Record<string, string>>,
64 * }} deps
65 */
66 export function registerBridgeFlowCaptureRoutes(app, deps) {
67 const {
68 dataDir,
69 canisterUrl,
70 canisterHeaders,
71 requireBridgeAuth,
72 resolveHostedBridgeContext,
73 effectiveRole,
74 loadRoles,
75 } = deps;
76
77 /**
78 * @param {import('express').Request} req
79 */
80 async function captureHandlerContext(req) {
81 const hctx = await resolveHostedBridgeContext(req, req.uid);
82 if (!hctx.ok) return hctx;
83 const roles = await loadRoles(req.blobStore);
84 const role = bridgeFlowCaptureHandlerRole(effectiveRole(req.uid, roles));
85 return { ok: true, hctx, role };
86 }
87
88 /**
89 * @param {import('express').Request} req
90 * @returns {boolean}
91 */
92 function sessionBoundFromReq(req) {
93 const auth = req.headers.authorization;
94 const token = auth && auth.startsWith('Bearer ') ? auth.slice(7) : null;
95 const secret = process.env.SESSION_SECRET;
96 if (!token || !secret) return false;
97 const payload = verifyJwtWithSecretRotation(token, secret, process.env.SESSION_SECRET_PREVIOUS);
98 return payload ? isSessionBoundActor(payload) : false;
99 }
100
101 /**
102 * @param {{
103 * effectiveCanisterUid: string,
104 * actorUid: string,
105 * vaultId: string,
106 * sessionBound?: boolean,
107 * }} ctx
108 */
109 function hostedCreateProposal(ctx) {
110 return async function createProposal(_dataDir, input) {
111 return createCaptureProposalOnCanister({
112 canisterUrl,
113 sessionBound: ctx.sessionBound === true,
114 headers: canisterHeaders({
115 'X-User-Id': ctx.effectiveCanisterUid,
116 'X-Actor-Id': ctx.actorUid,
117 'X-Vault-Id': ctx.vaultId,
118 }),
119 input: {
120 ...input,
121 vault_id: ctx.vaultId,
122 proposed_by: ctx.actorUid,
123 },
124 });
125 };
126 }
127
128 /**
129 * @param {import('express').Response} res
130 * @param {unknown} err
131 */
132 function sendRouteError(res, err) {
133 const e =
134 err && typeof err === 'object'
135 ? /** @type {{ status?: number, code?: string, message?: string }} */ (err)
136 : {};
137 const status = typeof e.status === 'number' ? e.status : 500;
138 const code = typeof e.code === 'string' ? e.code : 'RUNTIME_ERROR';
139 const message = typeof e.message === 'string' ? e.message : String(err);
140 return res.status(status).json({ error: message, code });
141 }
142
143 app.post('/api/v1/flows/capture/observe', requireBridgeAuth, async (req, res) => {
144 const ctx = await captureHandlerContext(req);
145 if (!ctx.ok) return res.status(ctx.status).json({ error: ctx.error, code: ctx.code });
146
147 const body = req.body && typeof req.body === 'object' ? req.body : {};
148 try {
149 const result = await withExternalProtocolBlobSync({
150 blobStore: req.blobStore ?? null,
151 dataDir,
152 run: () =>
153 handleFlowCaptureObserveRequest({
154 dataDir,
155 vaultId: ctx.hctx.vaultId,
156 userId: req.uid,
157 role: ctx.role,
158 sessionMeta: body,
159 includeLowConfidence: body.include_low_confidence === true,
160 harness: body.harness,
161 }),
162 });
163 if (!result.ok) {
164 return res.status(result.status).json({ error: result.error, code: result.code });
165 }
166 return res.json(result.payload);
167 } catch (err) {
168 return sendRouteError(res, err);
169 }
170 });
171
172 app.get('/api/v1/flows/candidates', requireBridgeAuth, async (req, res) => {
173 const ctx = await captureHandlerContext(req);
174 if (!ctx.ok) return res.status(ctx.status).json({ error: ctx.error, code: ctx.code });
175
176 const limitRaw = req.query.limit != null ? parseInt(String(req.query.limit), 10) : undefined;
177 try {
178 await hydrateExternalProtocolStoresFromBlob(req.blobStore ?? null, dataDir);
179 const result = handleFlowCaptureListRequest({
180 dataDir,
181 vaultId: ctx.hctx.vaultId,
182 userId: req.uid,
183 role: ctx.role,
184 scope: typeof req.query.scope === 'string' ? req.query.scope : undefined,
185 includeLowConfidence: req.query.include_low_confidence === 'true',
186 limit: Number.isFinite(limitRaw) ? limitRaw : undefined,
187 });
188 if (!result.ok) {
189 return res.status(result.status).json({ error: result.error, code: result.code });
190 }
191 return res.json(result.payload);
192 } catch (err) {
193 return sendRouteError(res, err);
194 }
195 });
196
197 app.post('/api/v1/flows/candidates/:id/propose', requireBridgeAuth, async (req, res) => {
198 const ctx = await captureHandlerContext(req);
199 if (!ctx.ok) return res.status(ctx.status).json({ error: ctx.error, code: ctx.code });
200
201 const candidateId =
202 typeof req.params.id === 'string' ? decodeURIComponent(req.params.id).trim() : '';
203 const body = req.body && typeof req.body === 'object' ? req.body : {};
204 try {
205 const result = await withExternalProtocolBlobSync({
206 blobStore: req.blobStore ?? null,
207 dataDir,
208 run: () =>
209 handleFlowCaptureProposeRequest({
210 dataDir,
211 vaultId: ctx.hctx.vaultId,
212 userId: req.uid,
213 role: ctx.role,
214 candidateId,
215 confirmedScope: body.confirmed_scope,
216 scopeWidenAcknowledged: body.scope_widen_acknowledged === true,
217 allowLowConfidence: body.allow_low_confidence === true,
218 forceNewFlow: body.force_new_flow === true,
219 mergeIntoFlowId: body.merge_into_flow_id,
220 intent: body.intent,
221 createProposal: hostedCreateProposal({
222 effectiveCanisterUid: ctx.hctx.effectiveCanisterUid,
223 actorUid: req.uid,
224 vaultId: ctx.hctx.vaultId,
225 sessionBound: sessionBoundFromReq(req),
226 }),
227 }),
228 });
229 if (!result.ok) {
230 const payload = { error: result.error, code: result.code };
231 if (result.merge_into_flow_id) payload.merge_into_flow_id = result.merge_into_flow_id;
232 if (result.overlap != null) payload.overlap = result.overlap;
233 return res.status(result.status).json(payload);
234 }
235 return res.status(201).json(result.payload);
236 } catch (err) {
237 return sendRouteError(res, err);
238 }
239 });
240
241 app.post('/api/v1/flows/candidates/:id/dismiss', requireBridgeAuth, async (req, res) => {
242 const ctx = await captureHandlerContext(req);
243 if (!ctx.ok) return res.status(ctx.status).json({ error: ctx.error, code: ctx.code });
244
245 const candidateId =
246 typeof req.params.id === 'string' ? decodeURIComponent(req.params.id).trim() : '';
247 const body = req.body && typeof req.body === 'object' ? req.body : {};
248 try {
249 const result = await withExternalProtocolBlobSync({
250 blobStore: req.blobStore ?? null,
251 dataDir,
252 run: () =>
253 handleFlowCaptureDismissRequest({
254 dataDir,
255 vaultId: ctx.hctx.vaultId,
256 userId: req.uid,
257 role: ctx.role,
258 candidateId,
259 intent: body.intent,
260 createProposal: hostedCreateProposal({
261 effectiveCanisterUid: ctx.hctx.effectiveCanisterUid,
262 actorUid: req.uid,
263 vaultId: ctx.hctx.vaultId,
264 sessionBound: sessionBoundFromReq(req),
265 }),
266 }),
267 });
268 if (!result.ok) {
269 return res.status(result.status).json({ error: result.error, code: result.code });
270 }
271 return res.status(201).json(result.payload);
272 } catch (err) {
273 return sendRouteError(res, err);
274 }
275 });
276
277 // Hub-complete capture apply (CAPTURE-HOSTED-APPLY-KN-b / CHA-C2).
278 // Called by the gateway post-approve hook; also re-callable by ops after fixing
279 // store state while the proposal is still `approved` (CHA-C11 recovery).
280 // withExternalProtocolBlobSync hydrates hub_flow_store.json before precheck so a
281 // cold lambda sees pending_review candidates (CHA-C3) and persists after apply.
282 app.post(
283 '/api/v1/flows/capture/proposals/:proposal_id/apply-approved',
284 requireBridgeAuth,
285 async (req, res) => {
286 const hctx = await resolveHostedBridgeContext(req, req.uid);
287 if (!hctx.ok) return res.status(hctx.status).json({ error: hctx.error, code: hctx.code });
288
289 const proposalId =
290 typeof req.params.proposal_id === 'string'
291 ? decodeURIComponent(req.params.proposal_id).trim()
292 : '';
293 if (!proposalId) {
294 return res.status(400).json({ error: 'proposal_id required', code: 'BAD_REQUEST' });
295 }
296
297 try {
298 const result = await withExternalProtocolBlobSync({
299 blobStore: req.blobStore ?? null,
300 dataDir,
301 run: () =>
302 applyApprovedCaptureProposalFromCanister({
303 dataDir,
304 canisterUrl,
305 headers: canisterHeaders({
306 'X-User-Id': hctx.effectiveCanisterUid,
307 'X-Actor-Id': req.uid,
308 'X-Vault-Id': hctx.vaultId,
309 }),
310 proposalId,
311 requireApproved: true,
312 }),
313 });
314 if (!result.ok) {
315 return res.status(result.status).json({ error: result.error, code: result.code });
316 }
317 return res.json(result.payload);
318 } catch (err) {
319 return sendRouteError(res, err);
320 }
321 },
322 );
323
324 // Hosted Flow list/get exposure (CHA-C5) — same handlers as self-hosted hub/server.mjs
325 // so a promote apply is observable via Scooling listFlows. Blob hydrate before read.
326 app.get('/api/v1/flows', requireBridgeAuth, async (req, res) => {
327 const ctx = await captureHandlerContext(req);
328 if (!ctx.ok) return res.status(ctx.status).json({ error: ctx.error, code: ctx.code });
329
330 try {
331 await hydrateExternalProtocolStoresFromBlob(req.blobStore ?? null, dataDir);
332 const limitRaw = req.query.limit != null ? parseInt(String(req.query.limit), 10) : undefined;
333 const result = handleFlowListRequest({
334 dataDir,
335 vaultId: ctx.hctx.vaultId,
336 userId: req.uid,
337 role: ctx.role,
338 scope: typeof req.query.scope === 'string' ? req.query.scope : undefined,
339 tag: typeof req.query.tag === 'string' ? req.query.tag : undefined,
340 limit: Number.isFinite(limitRaw) ? limitRaw : undefined,
341 });
342 if (!result.ok) {
343 return res.status(result.status).json({ error: result.error, code: result.code });
344 }
345 return res.json(result.payload);
346 } catch (err) {
347 return sendRouteError(res, err);
348 }
349 });
350
351 // MUST stay registered after GET /api/v1/flows/candidates (above) so 'candidates'
352 // is never treated as a flow id — CHA-C5 static-path ordering.
353 app.get('/api/v1/flows/:id', requireBridgeAuth, async (req, res) => {
354 const ctx = await captureHandlerContext(req);
355 if (!ctx.ok) return res.status(ctx.status).json({ error: ctx.error, code: ctx.code });
356
357 const flowId =
358 typeof req.params.id === 'string' ? decodeURIComponent(req.params.id).trim() : '';
359 try {
360 await hydrateExternalProtocolStoresFromBlob(req.blobStore ?? null, dataDir);
361 const result = handleFlowGetRequest({
362 dataDir,
363 vaultId: ctx.hctx.vaultId,
364 flowId,
365 userId: req.uid,
366 role: ctx.role,
367 version: typeof req.query.version === 'string' ? req.query.version : undefined,
368 });
369 if (!result.ok) {
370 return res.status(result.status).json({ error: result.error, code: result.code });
371 }
372 return res.json(result.payload);
373 } catch (err) {
374 return sendRouteError(res, err);
375 }
376 });
377 }
File History 1 commit
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6 docs: record AIP-b SD-21 land (KN #308) Human 11 days ago