aaronrene / knowtation public
flow-authoring.mjs
542 lines 19.3 KB
Raw
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6 docs: record AIP-b SD-21 land (KN #308) Human 11 days ago
1 /**
2 * Flow authoring write-back facade (Phase 7A-L1b).
3 *
4 * A typed facade over the existing `/proposals` lifecycle (SD-4): drafting,
5 * editing, or importing a Flow becomes a standard proposal targeting the Flow's
6 * mirror note. There is **no second write path** — review/evaluation/approve/
7 * apply and the optimistic-concurrency check are the same machinery notes use.
8 * The Flow index changes **only** at approve→apply, by reconciling the approved
9 * mirror back into the store as a new `(flow_id, version)` row.
10 *
11 * `FLOW_AUTHORING_WRITES` defaults **off**; when off every propose/import returns
12 * `403 FLOW_AUTHORING_DISABLED` and no write path is reachable.
13 *
14 * @see docs/FLOW-AUTHORING-WRITEBACK-CONTRACT-7A-L1.md
15 * @see docs/FLOW-STORE-CONTRACT-7A-10.md
16 */
17
18 import fs from 'fs';
19 import path from 'path';
20
21 import { fnv1a64Hex, stableStringify } from '../note-state-id.mjs';
22 import {
23 validateFlowBundle,
24 flowDefinitionForClient,
25 latestStoredFlow,
26 upsertFlowVersion,
27 loadFlowStore,
28 parseSemver,
29 compareSemver,
30 FLOW_ID_RE,
31 SEMVER_RE,
32 } from './flow-store.mjs';
33 import {
34 resolveFlowVisibleScopes,
35 resolveFlowWriteAuthority,
36 } from './flow-scope.mjs';
37 import {
38 readVaultExternalAgentPolicy,
39 validateImportExternalTools,
40 } from './external-agent.mjs';
41 import { validateImportAutomatableSteps } from './flow-execution.mjs';
42 import {
43 SCOOLING_FLOW_EXTERNAL_REF_RE,
44 resolveOptionalScoolingExternalRef,
45 readProposeExternalRefRaw,
46 } from '../scooling-external-ref.mjs';
47
48 export const FLOW_STATE_ID_PREFIX = 'flowst1_';
49 export const FLOW_AUTHORING_POLICY_FILE = 'hub_flow_authoring_policy.json';
50 export const FLOW_PROPOSAL_SCHEMA = 'knowtation.flow_proposal/v0';
51 export const FLOW_PROPOSAL_SOURCE = 'flow';
52 export const FLOW_REVIEW_QUEUE = 'flow-authoring';
53
54 /** @typedef {import('./flow-scope.mjs').FlowScope} FlowScope */
55 /** @typedef {'new'|'edit'|'import'} FlowProposeKind */
56
57 /**
58 * Canonicalize a flow record to the stable subset used by `flowStateId`.
59 * Mirrors `flowDefinitionForClient` so a token computed from a `flow get`
60 * payload reproduces server-side byte-for-byte.
61 *
62 * @param {Record<string, unknown>} flow
63 * @returns {Record<string, unknown>}
64 */
65 function canonicalFlowForState(flow) {
66 return {
67 schema: 'knowtation.flow/v0',
68 flow_id: flow.flow_id,
69 title: flow.title,
70 version: flow.version,
71 scope: flow.scope,
72 summary: flow.summary,
73 tags: Array.isArray(flow.tags) ? flow.tags : [],
74 steps: Array.isArray(flow.steps) ? flow.steps : [],
75 inputs: Array.isArray(flow.inputs) ? flow.inputs : [],
76 vault_mirror_path: typeof flow.vault_mirror_path === 'string' ? flow.vault_mirror_path : null,
77 updated: flow.updated,
78 truncated: flow.truncated === true,
79 };
80 }
81
82 /**
83 * @param {Record<string, unknown>} step
84 * @returns {Record<string, unknown>}
85 */
86 function canonicalStepForState(step) {
87 return {
88 schema: 'knowtation.flow_step/v0',
89 step_id: step.step_id,
90 flow_id: step.flow_id,
91 ordinal: step.ordinal,
92 owned_job: step.owned_job,
93 instruction: step.instruction,
94 trigger: step.trigger,
95 when_not_to_run: step.when_not_to_run,
96 requires: Array.isArray(step.requires) ? step.requires : [],
97 boundaries: Array.isArray(step.boundaries) ? step.boundaries : [],
98 skill_refs: Array.isArray(step.skill_refs) ? step.skill_refs : [],
99 inputs: Array.isArray(step.inputs) ? step.inputs : [],
100 outputs: Array.isArray(step.outputs) ? step.outputs : [],
101 output_shape: step.output_shape,
102 verification: step.verification,
103 automatable: step.automatable,
104 };
105 }
106
107 /**
108 * Deterministic optimistic-concurrency token over a flow definition + ordered
109 * steps. `flowst1_<16 hex>` = FNV-1a 64-bit over the key-sorted canonical
110 * content. Reuses `fnv1a64Hex` + `stableStringify` from `lib/note-state-id.mjs`.
111 *
112 * @param {Record<string, unknown>} flow
113 * @param {Record<string, unknown>[]} steps
114 * @returns {string}
115 */
116 export function flowStateId(flow, steps) {
117 const orderedSteps = [...(Array.isArray(steps) ? steps : [])]
118 .map((s) => canonicalStepForState(s))
119 .sort((a, b) => Number(a.ordinal) - Number(b.ordinal));
120 const payload = stableStringify({
121 flow: canonicalFlowForState(flow || {}),
122 steps: orderedSteps,
123 });
124 return FLOW_STATE_ID_PREFIX + fnv1a64Hex(Buffer.from(payload, 'utf8'));
125 }
126
127 /**
128 * State token for a flow that must still be **absent** (propose-new). Mirrors
129 * the note `absentNoteStateId` sentinel.
130 *
131 * @returns {string}
132 */
133 export function absentFlowStateId() {
134 return FLOW_STATE_ID_PREFIX + fnv1a64Hex(Buffer.from([0x00]));
135 }
136
137 /** @param {unknown} v */
138 function envTriState(v) {
139 if (v === '1' || v === 'true') return true;
140 if (v === '0' || v === 'false') return false;
141 return null;
142 }
143
144 /**
145 * @param {string} dataDir
146 * @returns {{ flow_authoring_writes_enabled?: boolean, flow_authoring_forbidden?: boolean }}
147 */
148 export function readFlowAuthoringPolicyFile(dataDir) {
149 if (!dataDir) return {};
150 const fp = path.join(dataDir, FLOW_AUTHORING_POLICY_FILE);
151 try {
152 if (!fs.existsSync(fp)) return {};
153 const j = JSON.parse(fs.readFileSync(fp, 'utf8'));
154 if (!j || typeof j !== 'object') return {};
155 const out = {};
156 if (typeof j.flow_authoring_writes_enabled === 'boolean') {
157 out.flow_authoring_writes_enabled = j.flow_authoring_writes_enabled;
158 }
159 if (typeof j.flow_authoring_forbidden === 'boolean') {
160 out.flow_authoring_forbidden = j.flow_authoring_forbidden;
161 }
162 return out;
163 } catch {
164 return {};
165 }
166 }
167
168 /**
169 * Whether durable Flow authoring writes are enabled (tri-state, default OFF).
170 * Precedence: explicit `FLOW_AUTHORING_WRITES` env (1/true|0/false) overrides the
171 * policy file; else file; else default `false`.
172 *
173 * @param {string} dataDir
174 * @returns {boolean}
175 */
176 export function getFlowAuthoringWritesEnabled(dataDir) {
177 const fromEnv = envTriState(process.env.FLOW_AUTHORING_WRITES);
178 if (fromEnv !== null) return fromEnv;
179 return readFlowAuthoringPolicyFile(dataDir).flow_authoring_writes_enabled === true;
180 }
181
182 /**
183 * Whether an org/classroom policy forbids authoring entirely (default false).
184 *
185 * @param {string} dataDir
186 * @returns {boolean}
187 */
188 export function getFlowAuthoringForbidden(dataDir) {
189 const fromEnv = envTriState(process.env.FLOW_AUTHORING_FORBIDDEN);
190 if (fromEnv !== null) return fromEnv;
191 return readFlowAuthoringPolicyFile(dataDir).flow_authoring_forbidden === true;
192 }
193
194 /**
195 * Server-derive `auto_approvable` from the bundle's verification kinds. A draft
196 * has no `auto_approvable` field; any `human_review` step ⇒ `false` so a draft
197 * can never self-authorize.
198 *
199 * @param {{ verification?: { kind?: string } }[]} steps
200 * @returns {boolean}
201 */
202 export function deriveAutoApprovable(steps) {
203 if (!Array.isArray(steps) || steps.length === 0) return false;
204 return steps.every((s) => s?.verification?.kind && s.verification.kind !== 'human_review');
205 }
206
207 /**
208 * @param {string} flowId
209 * @returns {string}
210 */
211 function defaultMirrorPath(flowId) {
212 const slug = flowId.replace(/^flow_/, '').replace(/_/g, '-');
213 return `meta/flows/${slug}.md`;
214 }
215
216 /**
217 * @param {Set<FlowScope>} [a]
218 * @param {Set<FlowScope>} [b]
219 * @returns {Set<FlowScope>}
220 */
221 function unionScopes(a, b) {
222 const out = new Set();
223 if (a) for (const s of a) out.add(s);
224 if (b) for (const s of b) out.add(s);
225 if (out.size === 0) out.add('personal');
226 return out;
227 }
228
229 /**
230 * @param {object} input
231 * @returns {{ visibleScopes: Set<FlowScope>, ambiguous: boolean }}
232 */
233 function resolveWriteScopes(input) {
234 if (input.ambiguous === true) {
235 return { visibleScopes: new Set(['personal']), ambiguous: true };
236 }
237 if (input.visibleScopes instanceof Set) {
238 return { visibleScopes: input.visibleScopes, ambiguous: false };
239 }
240 return resolveFlowVisibleScopes({
241 dataDir: input.dataDir,
242 userId: input.userId,
243 vaultId: input.vaultId,
244 role: input.role,
245 cliScopes: input.cliScopes,
246 });
247 }
248
249 /**
250 * @param {number} status
251 * @param {string} code
252 * @param {string} [error]
253 */
254 function refuse(status, code, error) {
255 return { ok: false, status, error: error ?? code, code };
256 }
257
258 /**
259 * THE one handler — MCP `flow_propose`/`flow_import`, Hub `POST /api/v1/flows`
260 * (+`/{id}/proposals`, `/import`), and CLI `flow propose|import` all converge
261 * here. Validates the bundle, resolves write authority server-side, runs the
262 * propose-time concurrency precheck, and delegates to the proposal create
263 * lifecycle. Never writes the Flow index (that happens only at approve→apply).
264 *
265 * @param {{
266 * dataDir: string,
267 * vaultId: string,
268 * userId?: string,
269 * role?: string,
270 * cliScopes?: FlowScope[],
271 * visibleScopes?: Set<FlowScope>,
272 * ambiguous?: boolean,
273 * kind: FlowProposeKind,
274 * flow?: unknown,
275 * steps?: unknown,
276 * bundle?: { flow?: unknown, steps?: unknown },
277 * intent?: unknown,
278 * flowId?: string,
279 * baseVersion?: string,
280 * baseStateId?: string,
281 * externalRef?: string,
282 * external_ref?: string,
283 * sourceVaultHint?: string,
284 * sessionBound?: boolean,
285 * createProposal: (dataDir: string, input: object) => { proposal_id: string } | Promise<{ proposal_id: string }>,
286 * starterDir?: string,
287 * }} input
288 * @returns {Promise<{ ok: true, payload: object } | { ok: false, status: number, error: string, code: string }>}
289 */
290 export async function handleFlowProposeRequest(input) {
291 const isImport = input.kind === 'import';
292 const malformedCode = isImport ? 'FLOW_IMPORT_BUNDLE_MALFORMED' : 'FLOW_DRAFT_INVALID';
293
294 // Gating — fail closed before any work.
295 if (getFlowAuthoringForbidden(input.dataDir)) {
296 return refuse(403, 'FLOW_AUTHORING_POLICY_FORBIDDEN', 'Flow authoring forbidden by policy');
297 }
298 if (!getFlowAuthoringWritesEnabled(input.dataDir)) {
299 return refuse(403, 'FLOW_AUTHORING_DISABLED', 'Flow authoring writes are disabled');
300 }
301
302 if (typeof input.createProposal !== 'function') {
303 return refuse(500, 'RUNTIME_ERROR', 'createProposal is required');
304 }
305
306 // Intent — required, untrusted, recorded verbatim.
307 const intent = typeof input.intent === 'string' ? input.intent.trim() : '';
308 if (!intent) {
309 return refuse(400, malformedCode, 'intent is required');
310 }
311
312 // Bundle shape + anatomy completeness.
313 const rawBundle = isImport
314 ? input.bundle && typeof input.bundle === 'object'
315 ? input.bundle
316 : {}
317 : { flow: input.flow, steps: input.steps };
318 const validated = validateFlowBundle(rawBundle);
319 if (!validated.ok) {
320 return refuse(400, malformedCode, validated.reason);
321 }
322 const { flow, steps } = validated;
323
324 // For an edit, the request flow_id must match the bundle.
325 if (input.kind === 'edit') {
326 const requestedId = typeof input.flowId === 'string' ? input.flowId.trim() : '';
327 if (requestedId && requestedId !== flow.flow_id) {
328 return refuse(400, 'FLOW_DRAFT_INVALID', 'flow_id mismatch between path and bundle');
329 }
330 }
331
332 // Scope resolution (deny-by-default; ambiguous fails closed).
333 const resolved = resolveWriteScopes(input);
334 if (resolved.ambiguous) {
335 return refuse(400, 'FLOW_SCOPE_AMBIGUOUS', 'Ambiguous flow scope');
336 }
337
338 // Write authority — scope × role, server-side; no scope widening from inside.
339 const authority = resolveFlowWriteAuthority(resolved.visibleScopes, flow.scope);
340 if (!authority.ok) {
341 const code = isImport && authority.code === 'FLOW_SCOPE_DENIED'
342 ? 'FLOW_IMPORT_SCOPE_DENIED'
343 : authority.code;
344 return refuse(authority.status, code, authority.error);
345 }
346
347 // Import sandbox: external_tool refs must be in vault allowlist (FLOW-V0-SPEC §6 item 3).
348 if (isImport) {
349 const vaultPolicy = readVaultExternalAgentPolicy(input.dataDir);
350 const externalCheck = validateImportExternalTools(steps, vaultPolicy.allowedTools);
351 if (!externalCheck.ok && vaultPolicy.importPolicy === 'reject') {
352 return refuse(403, 'FLOW_IMPORT_EXTERNAL_TOOL_DENIED', 'Import declares tools outside allowlist');
353 }
354 const automatableCheck = validateImportAutomatableSteps(steps, input.dataDir);
355 if (!automatableCheck.ok) {
356 return refuse(403, 'FLOW_IMPORT_AUTOMATABLE_DENIED', 'Import declares automatable steps where policy forbids');
357 }
358 }
359
360 // Optimistic concurrency precheck (fast fail; approve re-checks authoritatively).
361 const store = loadFlowStore(input.dataDir);
362 const vault = store.vaults[input.vaultId];
363 const current = vault ? latestStoredFlow(vault, flow.flow_id) : null;
364 // Only flows in a scope the actor may read are "visible"; others are absent to them.
365 const currentVisible =
366 current && resolved.visibleScopes.has(current.flow.scope) ? current : null;
367
368 let proposalBaseStateId;
369 let proposalBaseVersion = null;
370
371 if (input.kind === 'edit') {
372 const baseVersion = typeof input.baseVersion === 'string' ? input.baseVersion.trim() : '';
373 const baseStateId = typeof input.baseStateId === 'string' ? input.baseStateId.trim() : '';
374 if (!baseVersion || !SEMVER_RE.test(baseVersion) || !baseStateId.startsWith(FLOW_STATE_ID_PREFIX)) {
375 return refuse(400, 'FLOW_DRAFT_INVALID', 'edit requires base_version + base_state_id');
376 }
377 // No existence leak: an unreadable/missing flow is uniformly unknown_flow.
378 if (!currentVisible) {
379 return refuse(404, 'unknown_flow', 'unknown_flow');
380 }
381 const canonical = flowDefinitionForClient(currentVisible.flow, currentVisible.steps);
382 const serverStateId = flowStateId(canonical.flow, canonical.steps);
383 if (currentVisible.flow.version !== baseVersion || serverStateId !== baseStateId) {
384 return refuse(409, 'FLOW_LINEAGE_CONFLICT', 'flow changed since edit was based');
385 }
386 const next = parseSemver(flow.version);
387 const base = parseSemver(baseVersion);
388 if (!next || !base || compareSemver(next, base) <= 0) {
389 return refuse(400, 'FLOW_DRAFT_INVALID', 'flow.version must be greater than base_version');
390 }
391 proposalBaseStateId = baseStateId;
392 proposalBaseVersion = baseVersion;
393 } else {
394 // New (and import-as-new): the flow_id must still be absent in the actor's scope.
395 if (currentVisible) {
396 return refuse(409, 'FLOW_LINEAGE_CONFLICT', 'flow_id already exists in scope');
397 }
398 proposalBaseStateId = absentFlowStateId();
399 }
400
401 const autoApprovable = deriveAutoApprovable(steps);
402
403 // FLOW-WRITE-LIVE §FWL.4.1 — optional scooling.flow: ref (malformed → 400; absent → ok).
404 // Import lineage hints (sourceVaultHint / free-form muse refs) MUST NOT substitute.
405 const refGate = resolveFlowProposeExternalRef(input);
406 if (!refGate.ok) return refGate;
407
408 // Build the mirror-note proposal (review-before-write; no index write here).
409 const mirrorPath = flow.vault_mirror_path || defaultMirrorPath(flow.flow_id);
410 const body = JSON.stringify({ flow, steps }, null, 2);
411 const frontmatter = {
412 type: 'flow',
413 flow_id: flow.flow_id,
414 flow_version: flow.version,
415 scope: flow.scope,
416 };
417
418 const proposal = await Promise.resolve(
419 input.createProposal(input.dataDir, {
420 path: mirrorPath,
421 body,
422 frontmatter,
423 intent,
424 base_state_id: proposalBaseStateId,
425 ...(refGate.externalRef ? { external_ref: refGate.externalRef } : {}),
426 source: FLOW_PROPOSAL_SOURCE,
427 vault_id: input.vaultId,
428 proposed_by: typeof input.userId === 'string' && input.userId.trim() ? input.userId.trim() : undefined,
429 review_queue: FLOW_REVIEW_QUEUE,
430 flow_meta: {
431 kind: isImport ? 'import' : input.kind,
432 base_version: proposalBaseVersion,
433 base_state_id: proposalBaseStateId,
434 },
435 ...(typeof input.sessionBound === 'boolean' ? { session_bound: input.sessionBound } : {}),
436 }),
437 );
438
439 return {
440 ok: true,
441 payload: {
442 schema: FLOW_PROPOSAL_SCHEMA,
443 proposal_id: proposal.proposal_id,
444 flow_id: flow.flow_id,
445 base_version: proposalBaseVersion,
446 base_state_id: input.kind === 'edit' ? proposalBaseStateId : null,
447 scope: flow.scope,
448 auto_approvable: autoApprovable,
449 status: 'proposed',
450 review_queue: FLOW_REVIEW_QUEUE,
451 },
452 };
453 }
454
455 /**
456 * Optional Scooling Flow `external_ref` on propose (§FWL.4.1).
457 * Malformed → 400; absent → ok/undefined (propose may succeed; not admitted).
458 *
459 * @param {object} input
460 * @returns {{ ok: true, externalRef: string|undefined } | { ok: false, status: number, error: string, code: string }}
461 */
462 function resolveFlowProposeExternalRef(input) {
463 const resolved = resolveOptionalScoolingExternalRef(
464 readProposeExternalRefRaw(input),
465 SCOOLING_FLOW_EXTERNAL_REF_RE,
466 );
467 if (!resolved.ok) {
468 return refuse(resolved.status, resolved.code, resolved.error);
469 }
470 return { ok: true, externalRef: resolved.externalRef };
471 }
472
473 /**
474 * Approve-time **authoritative** concurrency re-check + bundle parse for a Flow
475 * proposal (the binding check). Run BEFORE the mirror note is written so a
476 * conflict short-circuits with no partial state.
477 *
478 * @param {string} dataDir
479 * @param {object} proposal - the stored proposal (source === 'flow').
480 * @returns {{ ok: true, vaultId: string, flow: object, steps: object[] } | { ok: false, status: number, error: string, code: string }}
481 */
482 export function precheckApprovedFlowProposal(dataDir, proposal) {
483 let parsed;
484 try {
485 parsed = JSON.parse(typeof proposal.body === 'string' ? proposal.body : '');
486 } catch {
487 return refuse(400, 'FLOW_DRAFT_INVALID', 'flow proposal body is not valid JSON');
488 }
489 const validated = validateFlowBundle(parsed);
490 if (!validated.ok) {
491 return refuse(400, 'FLOW_DRAFT_INVALID', validated.reason);
492 }
493 const { flow, steps } = validated;
494 const vaultId = typeof proposal.vault_id === 'string' && proposal.vault_id.trim()
495 ? proposal.vault_id.trim()
496 : 'default';
497 const meta = proposal.flow_meta && typeof proposal.flow_meta === 'object' ? proposal.flow_meta : {};
498 const kind = meta.kind === 'edit' ? 'edit' : 'new';
499
500 const store = loadFlowStore(dataDir);
501 const vault = store.vaults[vaultId];
502 const current = vault ? latestStoredFlow(vault, flow.flow_id) : null;
503
504 if (kind === 'edit') {
505 if (!current) {
506 return refuse(409, 'FLOW_LINEAGE_CONFLICT', 'flow disappeared before approve');
507 }
508 const baseVersion = typeof meta.base_version === 'string' ? meta.base_version : '';
509 const baseStateId = typeof meta.base_state_id === 'string' ? meta.base_state_id : '';
510 const canonical = flowDefinitionForClient(current.flow, current.steps);
511 const serverStateId = flowStateId(canonical.flow, canonical.steps);
512 if (current.flow.version !== baseVersion || serverStateId !== baseStateId) {
513 return refuse(409, 'FLOW_LINEAGE_CONFLICT', 'flow changed since edit was based');
514 }
515 const next = parseSemver(flow.version);
516 const base = parseSemver(baseVersion);
517 if (!next || !base || compareSemver(next, base) <= 0) {
518 return refuse(400, 'FLOW_DRAFT_INVALID', 'flow.version must be greater than base_version');
519 }
520 } else if (current) {
521 return refuse(409, 'FLOW_LINEAGE_CONFLICT', 'flow_id already exists');
522 }
523
524 return { ok: true, vaultId, flow, steps };
525 }
526
527 /**
528 * Apply a pre-checked Flow bundle into the index as a new `(flow_id, version)`
529 * row. Call AFTER the mirror note write succeeds; the bundle has already been
530 * validated by {@link precheckApprovedFlowProposal} so this cannot fail on shape.
531 *
532 * @param {string} dataDir
533 * @param {string} vaultId
534 * @param {object} flow
535 * @param {object[]} steps
536 * @returns {{ created: boolean, version: string }}
537 */
538 export function applyFlowProposalToIndex(dataDir, vaultId, flow, steps) {
539 return upsertFlowVersion(dataDir, vaultId, flow, steps);
540 }
541
542 export { FLOW_ID_RE };
File History 1 commit
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6 docs: record AIP-b SD-21 land (KN #308) Human 11 days ago