Serve governed console workflows and verify protocol traces

apps/console/src/control.ts

@@ -0,0 +1,67 @@
+import type { ConsoleAdapters } from "./index.ts";
+import { createHash, createHmac, randomBytes } from "node:crypto";
+
+type ControlStore = { id: string; endpoint: string };
+type ControlOptions = { stores: ControlStore[]; capability?: string; attestationSecret?: string; fetcher?: typeof fetch };
+
+function unavailable(): never {
+  throw new Error("ConsoleAdapterUnavailable");
+}
+
+function selected(stores: ControlStore[], id: string | null): ControlStore {
+  if (id !== null) {
+    const store = stores.find((candidate) => candidate.id === id);
+    if (store) return store;
+    throw new Error("ConsoleStoreUnknown");
+  }
+  if (stores.length === 1) return stores[0];
+  throw new Error("ConsoleStoreSelectionRequired");
+}
+
+function controlDocument(document: unknown, editing: boolean): { store: string | null; body: { expected: number; document?: string; nonce?: string } } {
+  if (document === null || typeof document !== "object" || Array.isArray(document)) throw new Error("ConsoleRequestMalformed");
+  const value = document as Record<string, unknown>;
+  if (value.store !== undefined && typeof value.store !== "string") throw new Error("ConsoleRequestMalformed");
+  if (!Number.isSafeInteger(value.expected) || (value.expected as number) < 1) throw new Error("ConsoleRequestMalformed");
+  if (editing && typeof value.document !== "string") throw new Error("ConsoleRequestMalformed");
+  if (!editing && (typeof value.nonce !== "string" || !/^[0-9a-f]{48}$/.test(value.nonce))) throw new Error("ConsoleRequestMalformed");
+  const keys = Object.keys(value);
+  if (keys.some((key) => key !== "store" && key !== "expected" && (key !== "document" || !editing) && (key !== "nonce" || editing))) throw new Error("ConsoleRequestMalformed");
+  return { store: typeof value.store === "string" ? value.store : null,
+    body: editing ? { expected: value.expected as number, document: value.document as string } : { expected: value.expected as number, nonce: value.nonce as string } };
+}
+
+export function createLiveControl({ stores, capability, attestationSecret, fetcher = fetch }: ControlOptions): ConsoleAdapters["control"] {
+  async function call(store: ControlStore, path: string, token: string, body?: object, subject?: string): Promise<unknown> {
+    const serialized = body ? JSON.stringify(body) : undefined;
+    const headers: Record<string, string> = { Authorization: `Bearer ${token}`, ...(body ? { "Content-Type": "application/json" } : {}) };
+    if (body && subject && attestationSecret) {
+      const timestamp = Math.floor(Date.now() / 1000).toString();
+      const nonce = randomBytes(16).toString("hex");
+      const digest = createHash("sha256").update(serialized!).digest("hex");
+      const signature = createHmac("sha256", attestationSecret).update(`POST\n${path}\n${digest}\n${subject}\n${timestamp}\n${nonce}`).digest("hex");
+      Object.assign(headers, { "X-Contextful-Operator": subject, "X-Contextful-Operator-Time": timestamp, "X-Contextful-Operator-Nonce": nonce, "X-Contextful-Operator-Signature": signature });
+    }
+    const response = await fetcher(new URL(path, store.endpoint), {
+      method: body ? "POST" : "GET",
+      headers,
+      ...(serialized ? { body: serialized } : {}),
+    });
+    if (!response.ok) throw new Error(`ConsoleControlRefused:${response.status}`);
+    return response.json();
+  }
+  return {
+    workflows: async (_operator, id) => capability ? call(selected(stores, id), "/control/workflows", capability) : unavailable(),
+    record: async (_operator, id) => capability ? call(selected(stores, id), "/control/record", capability) : unavailable(),
+    edit: async (document, token, operator) => {
+      if (!capability || token !== capability) return unavailable();
+      const input = controlDocument(document, true);
+      return call(selected(stores, input.store), "/control/edit", token, input.body, operator.subject);
+    },
+    apply: async (document, token, operator) => {
+      if (!capability || token !== capability) return unavailable();
+      const input = controlDocument(document, false);
+      return call(selected(stores, input.store), "/control/apply", token, input.body, operator.subject);
+    },
+  };
+}

apps/console/src/entry.ts

@@ -5,6 +5,7 @@ import { registryFromEnv } from "../../gateway/src/index.ts";
 import { createLiveAnswer } from "./answer_live.ts";
 import { createLiveBrowse } from "./live_browse.ts";
 import type { ConsoleAdapters, Identity } from "./index.ts";
+import { createLiveControl } from "./control.ts";
 import { serveConsole } from "./server.ts";
 
 type HostedAdapters = Pick<ConsoleAdapters, "turn" | "control"> &
@@ -95,6 +96,8 @@ async function main(): Promise<void> {
     brief: live?.brief,
     briefBudgetMs: live?.briefBudgetMs,
     redactView: live?.redactView,
+    control: createLiveControl({ stores: registry.entries, capability: process.env.CONTEXTFUL_ADMIN_CAPABILITY,
+      attestationSecret: process.env.CONTEXTFUL_ADMIN_CAPABILITY ? required("CONTEXTFUL_CONTROL_ATTESTATION_SECRET") : undefined }),
   };
   if (modulePath) {
     const absolute = isAbsolute(modulePath) ? modulePath : resolve(modulePath);

apps/console/src/index.ts

@@ -346,6 +346,7 @@ export function createConsole(adapters: ConsoleAdapters): { fetch: (request: Req
       } else {
         const store = url.searchParams.get("store");
         if (store !== null && !adapters.stores.some((entry) => entry.id === store)) return refusal("ConsoleRequestMalformed", 400);
+        if (request.method === "GET" && path === "/admin/api/stores") return json(adapters.stores);
         if (request.method === "GET" && path === "/admin/api/workflows") return json(await adapters.control.workflows(operator, store));
         if (request.method === "GET" && path === "/admin/api/record") return json(await adapters.control.record(operator, store));
         if (request.method === "GET" && path === "/admin/api/packs") {
@@ -386,6 +387,10 @@ store.addEventListener('change',()=>{void showBrief();void refreshBrowse()});asO
 form.addEventListener('submit',async event=>{event.preventDefault();const question=questionField.value.trim();if(!question)return;brief.hidden=true;const response=await fetch('/query/api/ask',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify({store:store.value,question})});const answer=await response.json();transcript.replaceChildren();const item=document.createElement('article');item.textContent=question+'\\n'+(answer.answer??answer.error?.identifier??'No answer');transcript.append(item);widgets.replaceChildren();for(const widget of answer.widgets??[])widgets.append(draw(widget));if(answer.sources?.length){const sources=document.createElement('p');sources.textContent='Sources: '+answer.sources.map(x=>x.title??x.url).join(', ');transcript.append(sources)}});
 </script></html>`;
 
-const adminPage = `<!doctype html><html lang="en"><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>Admin · Contextful</title>${sharedStyle}<div class="shell"><nav aria-label="Console"><strong>Contextful</strong><a href="/query">Query</a><a href="/admin" aria-current="page">Admin</a></nav><main><h1>Store operations</h1><p class="muted">Pipelines, schedules, steps and run outcomes come from the store.</p><section><h2>Workflow canvas</h2><div id="canvas" class="grid"></div></section><section><h2>Operational record</h2><div id="record"></div></section><section><h2>Control document</h2><label for="document">Document</label><textarea id="document"></textarea><p><button id="edit">Edit</button> <button id="apply">Apply</button></p><p id="result" role="status"></p></section></main></div><script>
-const canvas=document.getElementById('canvas'),record=document.getElementById('record');fetch('/admin/api/workflows').then(r=>r.json()).then(data=>{for(const pipeline of data.pipelines??[]){const node=document.createElement('article');node.className='node';node.textContent=[pipeline.id,pipeline.schedule,...(pipeline.steps??[]),...(pipeline.runs??[]).map(run=>run.status)].filter(Boolean).join(' · ');canvas.append(node)}if(!canvas.children.length)canvas.textContent='No workflows are published.'});fetch('/admin/api/record').then(r=>r.json()).then(data=>{record.textContent=JSON.stringify(data,null,2)});for(const action of ['edit','apply'])document.getElementById(action).addEventListener('click',async()=>{const text=document.getElementById('document').value;let documentValue;try{documentValue=JSON.parse(text)}catch{document.getElementById('result').textContent='Enter a valid JSON document.';return}const response=await fetch('/admin/api/'+action,{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify(documentValue)});document.getElementById('result').textContent=response.ok?action+' complete':(await response.json()).error?.identifier??'Request failed'});
+const adminPage = `<!doctype html><html lang="en"><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>Admin · Contextful</title>${sharedStyle}<div class="shell"><nav aria-label="Console"><strong>Contextful</strong><a href="/query">Query</a><a href="/admin" aria-current="page">Admin</a></nav><main><h1>Store operations</h1><p class="muted">Pipelines, schedules, steps and run outcomes come from the store.</p><label for="admin-store">Store</label> <select id="admin-store" required></select><section><h2>Workflow canvas</h2><div id="canvas" class="grid"></div></section><section><h2>Operational record</h2><div id="record"></div></section><section><h2>Control document</h2><p id="version" class="muted"></p><label for="document">TOML document</label><textarea id="document"></textarea><p><button id="edit">Save draft</button> <button id="apply">Apply draft</button></p><p id="result" role="status"></p></section></main></div><script>
+const canvas=document.getElementById('canvas'),record=document.getElementById('record'),version=document.getElementById('version'),result=document.getElementById('result'),store=document.getElementById('admin-store');let applied=null,draftNonce=null;
+async function workflows(){applied=null;draftNonce=null;if(!store.value)return;const selected=store.value;const response=await fetch('/admin/api/workflows?store='+encodeURIComponent(selected));const data=await response.json();if(selected!==store.value)return;if(!response.ok){result.textContent=data.error?.identifier??'Workflow request failed';return}applied=data.applied;version.textContent=applied===null?'No applied version':'Applied version '+applied;canvas.replaceChildren();for(const pipeline of data.pipelines??[]){const node=document.createElement('article');node.className='node';const run=data.runs?.[pipeline.id];node.textContent=[pipeline.id,pipeline.schedule,...(pipeline.steps??[]),run?.status,run?.run_id].filter(Boolean).join(' · ');canvas.append(node)}if(!canvas.children.length)canvas.textContent='No workflows are published.'}
+async function refresh(){await workflows();if(!store.value)return;const response=await fetch('/admin/api/record?store='+encodeURIComponent(store.value));const data=await response.json();record.textContent=response.ok?JSON.stringify(data,null,2):data.error?.identifier??'Record request failed'}
+store.addEventListener('change',refresh);fetch('/admin/api/stores').then(r=>r.json()).then(rows=>{for(const row of rows){const option=document.createElement('option');option.value=row.id;option.textContent=row.label;store.append(option)}refresh()});
+for(const action of ['edit','apply'])document.getElementById(action).addEventListener('click',async()=>{if(!Number.isSafeInteger(applied)||!store.value){result.textContent='Select a store with an applied version.';return}if(action==='apply'&&!draftNonce){result.textContent='Save a draft before applying.';return}const body=action==='edit'?{store:store.value,expected:applied,document:document.getElementById('document').value}:{store:store.value,expected:applied,nonce:draftNonce};const response=await fetch('/admin/api/'+action,{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify(body)});const answer=await response.json();result.textContent=response.ok?action+' complete':answer.error?.identifier??'Request failed';if(response.ok&&action==='edit')draftNonce=answer.nonce;if(response.ok&&action==='apply')await refresh()});
 </script></html>`;

apps/console/test/console.test.ts

@@ -1,5 +1,6 @@
 import test from "node:test";
 import assert from "node:assert/strict";
+import { Script } from "node:vm";
 import { generateKeyPairSync, sign } from "node:crypto";
 import { createConsole, issueCognitoSession } from "../src/index.ts";
 import { serveConsole } from "../src/server.ts";
@@ -94,6 +95,12 @@ test("Query keeps one composer and transcript and delegates a sourced answer", a
   assert.match(page, /id="composer"/);
   assert.match(page, /id="transcript"/);
   assert.match(page, /id="widgets"/);
+  for (const id of ["brief", "chips", "insights", "file-gallery", "file-preview"]) {
+    assert.match(page, new RegExp(`id="${id}"`));
+  }
+  const script = page.match(/<script>([\s\S]*?)<\/script>/)?.[1];
+  assert.ok(script);
+  assert.doesNotThrow(() => new Script(script));
   const answer = await request("/query/api/ask", { ...access(queryAudience), origin: "https://console.example" }, { method: "POST", body: JSON.stringify({ store: "field-notes", question: "What arrived?" }) });
   assert.equal(answer.status, 200);
   assert.equal((await answer.json() as { sources: unknown[] }).sources.length, 1);
@@ -319,6 +326,24 @@ test("Admin pack listing reaches only the injected listing adapter", async () =>
   assert.deepEqual(calls, ["field-notes:packs/"]);
 });
 
+test("Admin names the selected store on every operational request and shows run outcomes", async () => {
+  const app = createConsole({
+    identity: { kind: "access", issuer, queryAudience, adminAudience, publicKey },
+    stores: [{ id: "one", label: "Store one" }, { id: "two", label: "Store two" }],
+    adminCapability: "server-secret", turn: async () => ({ answer: "", sources: [], widgets: [] }),
+    read: { list: async () => [] },
+    control: { workflows: async () => ({}), record: async () => ({}), edit: async () => ({}), apply: async () => ({}) },
+  });
+  const page = await (await app.fetch(new Request("https://console.example/admin", { headers: access(adminAudience) }))).text();
+  const stores = await app.fetch(new Request("https://console.example/admin/api/stores", { headers: access(adminAudience) }));
+  assert.deepEqual(await stores.json(), [{ id: "one", label: "Store one" }, { id: "two", label: "Store two" }]);
+  assert.match(page, /<select id="admin-store"/);
+  assert.match(page, /\/admin\/api\/workflows\?store=/);
+  assert.match(page, /\/admin\/api\/record\?store=/);
+  assert.match(page, /store:store\.value/);
+  assert.match(page, /data\.runs/);
+});
+
 test("Query builds its widget from rows and sanitizes view props before delivery", async () => {
   const app = createConsole({
     identity: { kind: "access", issuer, queryAudience, adminAudience, publicKey },