From 532135bc9bd842db7e8c1b65f7c8b4ef326481f7 Mon Sep 17 00:00:00 2001
From: Cameron Pfiffer <cameron@pfiffer.org>
Date: Mon, 10 Aug 2026 17:29:44 -0700
Subject: [PATCH] Stream inspector updates from Jazz.
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Push content-dark change signals over SSE so root activity, run state, counts, and selected-source observations refresh without exposing the raw Jazz account to the browser.

👾 Generated with [Letta Code](https://letta.com)

Co-Authored-By: Letta Code <noreply@letta.com>
---
 src/jazz/store.ts      | 16 ++++++++++
 src/web/inspector.ts   | 61 +++++++++++++++++++++++++++++++++++--
 test/inspector.test.ts | 68 ++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 142 insertions(+), 3 deletions(-)

diff --git a/src/jazz/store.ts b/src/jazz/store.ts
index 17a7ba0..854c3e4 100644
--- a/src/jazz/store.ts
+++ b/src/jazz/store.ts
@@ -834,6 +834,22 @@ export class JazzThoughtStore {
     );
   }
 
+  subscribeInspectorActivity(onChange: () => void): () => void {
+    const unsubscribers = [
+      this.db.subscribeAll(
+        thoughtstreamApp.events.where({}).orderBy("observedAt", "desc").limit(1),
+        () => onChange(),
+      ),
+      this.db.subscribeAll(
+        thoughtstreamApp.sources.where({}),
+        () => onChange(),
+      ),
+    ];
+    return () => {
+      for (const unsubscribe of unsubscribers) unsubscribe();
+    };
+  }
+
   async queryConsumerEvents(query: ConsumerEventQuery): Promise<ThoughtEvent[]> {
     if (query.eventTypes.length === 0 || query.acceptedPrivacy.length === 0) return [];
     const rows = await this.db.all(thoughtstreamApp.events.where({
diff --git a/src/web/inspector.ts b/src/web/inspector.ts
index 4ef4655..159a614 100644
--- a/src/web/inspector.ts
+++ b/src/web/inspector.ts
@@ -194,6 +194,10 @@ async function handleRequest(
     });
     return;
   }
+  if (url.pathname === "/api/live") {
+    streamInspectorActivity(store, request, response);
+    return;
+  }
   if (url.pathname === "/api/proposals") {
     sendJson(response, 200, await projectProposalQueue(store));
     return;
@@ -572,7 +576,7 @@ export function renderInspectorHtml(): string {
 <header><h1>thought stream</h1><nav><button id="events-tab" class="active">events</button><button id="runs-tab">runs</button><button id="sources-tab">sources</button><button id="suggestions-tab">suggestions</button><button id="reviews-tab">review</button><button id="artifacts-tab">artifacts</button></nav><span id="status" class="muted">loading</span></header>
 <main><section id="list-pane"><div class="section-head"><span id="list-title">root activity</span><span id="count">…</span></div><div id="filter-bar" class="filter-bar" hidden></div><div id="list"><div class="empty"><span class="spinner" aria-hidden="true"></span>Loading root activity…</div></div></section><section id="detail-pane"><div id="detail"><div class="empty">Select an observation to see what processed it and what, if anything, happened.</div></div></section></main>
 <script>
-const state={snapshot:null,proposals:null,reviews:null,artifacts:null,session:null,loading:{snapshot:true,proposals:true,reviews:true,artifacts:true,session:true},errors:{},tab:'events',selected:null,proposalReceipt:'',eventFilters:{source:'all',activity:'all',processing:'all',query:''}};
+const state={snapshot:null,proposals:null,reviews:null,artifacts:null,session:null,loading:{snapshot:true,proposals:true,reviews:true,artifacts:true,session:true},errors:{},tab:'events',selected:null,proposalReceipt:'',eventFilters:{source:'all',activity:'all',processing:'all',query:''},live:'connecting',snapshotRefreshInFlight:false,snapshotRefreshPending:false};
 const resources={snapshot:{path:'api/snapshot',label:'root activity'},proposals:{path:'api/proposals',label:'suggestions'},reviews:{path:'api/reviews',label:'review queue'},artifacts:{path:'api/artifacts',label:'artifacts'},session:{path:'api/session',label:'session'}};
 const esc=s=>String(s??'').replace(/[&<>"']/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'}[c]));
 const json=value=>esc(JSON.stringify(value,null,2));
@@ -586,9 +590,10 @@ async function load(){
   await loadResource('artifacts');
   await loadResource('proposals');
   await loadResource('reviews');
+  startLiveUpdates();
 }
 function tabResource(){return state.tab==='suggestions'?'proposals':state.tab==='reviews'?'reviews':state.tab==='artifacts'?'artifacts':'snapshot'}
-function updateStatus(){const pending=Object.values(state.loading).filter(Boolean).length;const failed=Object.keys(state.errors).length;const parts=[];if(failed)parts.push(failed+' failed');if(pending)parts.push('loading '+pending);if(!pending&&!failed)parts.push('ready');if(state.session)parts.push(state.session.reviewWriteEnabled?'review enabled':'read-only');document.querySelector('#status').textContent=parts.join(' · ')}
+function updateStatus(){const pending=Object.values(state.loading).filter(Boolean).length;const failed=Object.keys(state.errors).length;const parts=[];if(failed)parts.push(failed+' failed');if(pending)parts.push('loading '+pending);if(!pending&&!failed)parts.push('ready');parts.push(state.live==='live'?'live':state.live==='reconnecting'?'reconnecting':'connecting');if(state.session)parts.push(state.session.reviewWriteEnabled?'review enabled':'read-only');document.querySelector('#status').textContent=parts.join(' · ')}
 function emptyMessage(){return state.tab==='events'?'No root activity has been recorded.':state.tab==='runs'?'No agent runs have been recorded.':state.tab==='sources'?'No configured or observed sources are available.':state.tab==='suggestions'?'No suggestions are waiting.':state.tab==='reviews'?'No reviews are waiting.':'No artifacts are available.'}
 function renderList(){
   const list=document.querySelector('#list'); const isEvents=state.tab==='events'; const isRuns=state.tab==='runs'; const isSuggestions=state.tab==='suggestions'; const isReviews=state.tab==='reviews'; const isArtifacts=state.tab==='artifacts';
@@ -604,7 +609,12 @@ function renderList(){
   list.innerHTML=error?'<div class="empty failed">'+esc(error)+'<div><button class="retry" data-resource="'+esc(resourceKey)+'">Retry</button></div></div>':!ready?'<div class="empty"><span class="spinner" aria-hidden="true"></span>Loading '+esc(resource.label)+'…</div>':values.map(value=>isEvents?eventItem(value):isRuns?runItem(value):isSuggestions?suggestionItem(value):isReviews?reviewItem(value):isArtifacts?artifactItem(value):sourceItem(value)).join('')||'<div class="empty">'+esc(empty)+'</div>';
   list.querySelector('.retry')?.addEventListener('click',event=>loadResource(event.currentTarget.dataset.resource));
   list.querySelectorAll('button.item').forEach(button=>button.addEventListener('click',()=>{list.querySelectorAll('button.item').forEach(item=>item.classList.remove('active'));button.classList.add('active');select(button.dataset.kind,button.dataset.id)}));
+  if(state.selected)list.querySelectorAll('button.item').forEach(button=>button.classList.toggle('active',button.dataset.kind===state.selected.kind&&button.dataset.id===state.selected.id));
 }
+function startLiveUpdates(){const stream=new EventSource('api/live');stream.addEventListener('open',()=>{state.live='live';updateStatus()});stream.addEventListener('change',()=>scheduleSnapshotRefresh());stream.addEventListener('error',()=>{state.live='reconnecting';updateStatus()})}
+let snapshotRefreshTimer;
+function scheduleSnapshotRefresh(){clearTimeout(snapshotRefreshTimer);snapshotRefreshTimer=setTimeout(()=>void refreshSnapshot(),120)}
+async function refreshSnapshot(){if(state.snapshotRefreshInFlight){state.snapshotRefreshPending=true;return}state.snapshotRefreshInFlight=true;try{state.snapshot=await fetchJson(resources.snapshot.path,resources.snapshot.label);delete state.errors.snapshot;renderList();if(state.selected?.kind==='source')showSource(state.selected.id);else if(state.selected&&(state.selected.kind==='event'||state.selected.kind==='run'))await select(state.selected.kind,state.selected.id)}catch(error){state.errors.snapshot=error instanceof Error?error.message:String(error)}finally{state.snapshotRefreshInFlight=false;updateStatus();if(state.snapshotRefreshPending){state.snapshotRefreshPending=false;scheduleSnapshotRefresh()}}}
 function filterEvents(events){const filters=state.eventFilters;const query=filters.query.trim().toLowerCase();return events.filter(event=>{const observation=observationPresentation(event);if(filters.source!=='all'&&event.source!==filters.source)return false;if(filters.activity!=='all'&&observation.title!==filters.activity)return false;const processed=(event.consumerRuns??[]).length>0;if(filters.processing==='processed'&&!processed)return false;if(filters.processing==='unprocessed'&&processed)return false;if(query&&!([observation.title,observation.body,observation.objectLabel,observation.sourceLabel,event.summary,event.source,event.type].join(' ').toLowerCase().includes(query)))return false;return true})}
 function renderEventFilters(events){const bar=document.querySelector('#filter-bar');if(!events){bar.hidden=true;bar.innerHTML='';return}bar.hidden=false;const sourceMap=new Map((state.snapshot?.sources??[]).map(source=>[source.source,sourceLabelFromId(source.source)]));events.forEach(event=>sourceMap.set(event.source,observationSourceLabel(event)));const sourceOptions=[...sourceMap].sort((a,b)=>a[1].localeCompare(b[1])).map(([value,label])=>'<option value="'+esc(value)+'" '+(state.eventFilters.source===value?'selected':'')+'>'+esc(label)+'</option>').join('');const activities=[...new Set(events.map(event=>observationPresentation(event).title))].sort().map(value=>'<option value="'+esc(value)+'" '+(state.eventFilters.activity===value?'selected':'')+'>'+esc(value)+'</option>').join('');bar.innerHTML='<select id="event-source-filter" aria-label="Filter observations by source"><option value="all">All sources</option>'+sourceOptions+'</select><select id="event-activity-filter" aria-label="Filter observations by activity"><option value="all">All activity</option>'+activities+'</select><select id="event-processing-filter" aria-label="Filter observations by processing"><option value="all">Any processing</option><option value="processed" '+(state.eventFilters.processing==='processed'?'selected':'')+'>Processed</option><option value="unprocessed" '+(state.eventFilters.processing==='unprocessed'?'selected':'')+'>Not processed</option></select><input id="event-query" type="search" placeholder="Search observations" value="'+esc(state.eventFilters.query)+'"><button id="event-filter-clear">Clear</button>';bar.querySelector('#event-source-filter').addEventListener('change',event=>{state.eventFilters.source=event.target.value;renderList()});bar.querySelector('#event-activity-filter').addEventListener('change',event=>{state.eventFilters.activity=event.target.value;renderList()});bar.querySelector('#event-processing-filter').addEventListener('change',event=>{state.eventFilters.processing=event.target.value;renderList()});bar.querySelector('#event-query').addEventListener('input',event=>{const cursor=event.target.selectionStart;state.eventFilters.query=event.target.value;renderList();const next=document.querySelector('#event-query');next.focus();next.setSelectionRange(cursor,cursor)});bar.querySelector('#event-filter-clear').addEventListener('click',()=>{state.eventFilters={source:'all',activity:'all',processing:'all',query:''};renderList()})}
 function eventFilterEmptyMessage(){if(state.eventFilters.source!=='all'){const source=state.snapshot?.sources?.find(value=>value.source===state.eventFilters.source);if(source&&!source.hasActivityEvidence)return sourceLabelFromId(source.source)+' is configured but has no durable activity yet.'}return 'No observations match these filters.'}
@@ -628,7 +638,7 @@ function sourceItem(source){
 function suggestionItem(item){return '<button class="item" data-kind="suggestion" data-id="'+esc(item.eventId)+'"><div class="type">'+esc(item.kind)+' <span class="badge '+(item.status==='Failed'?'failed':'')+'">'+esc(item.status)+'</span></div><div class="summary">'+esc(item.proposedText.slice(0,240))+'</div><div class="meta">'+esc(item.reason)+' · '+esc(new Date(item.observedAt).toLocaleString())+'</div></button>'}
 function reviewItem(item){const status=item.decision?item.decision.disposition:'unreviewed';return '<button class="item" data-kind="review" data-id="'+esc(item.id)+'"><div class="type">'+esc(item.campaign.label)+' <span class="badge">'+esc(status)+'</span></div><div class="summary">'+esc(item.prompt.slice(0,240))+'</div><div class="meta">'+esc(item.criterion.label)+' · '+esc(item.privacy)+(item.decision?.externalExportEligible?' · export eligible':'')+'</div></button>'}
 async function select(kind,id){
-  if(kind==='source'){const source=state.snapshot.sources.find(value=>value.source===id);const detail=document.querySelector('#detail');detail.innerHTML=renderSource(source);detail.querySelectorAll('[data-event-id]').forEach(button=>button.addEventListener('click',()=>select('event',button.dataset.eventId)));return}
+  if(kind==='source'){showSource(id);return}
   if(kind==='suggestion'){const item=state.proposals.items.find(value=>value.eventId===id);document.querySelector('#detail').innerHTML=renderSuggestion(item);bindSuggestionForm(item);return}
   if(kind==='review'){const item=state.reviews.items.find(value=>value.id===id);document.querySelector('#detail').innerHTML=renderReview(item);bindReviewForm(item);return}
   if(kind==='artifact'){state.selected={kind,id};document.querySelector('main').classList.add('artifact-selected');document.querySelector('#detail').innerHTML='<button class="artifact-back">← Artifacts</button><div class="empty">Loading artifact…</div>';bindArtifactBack();const response=await fetch('api/artifacts/'+encodeURIComponent(id));const data=await response.json();document.querySelector('#detail').innerHTML=renderArtifact(data);bindArtifactBack();return}
@@ -636,6 +646,7 @@ async function select(kind,id){
   detail.innerHTML='<div class="empty"><span class="spinner" aria-hidden="true"></span>Loading '+label+'…</div>';
   try{const data=await fetchJson('api/'+(kind==='event'?'events/':'runs/')+encodeURIComponent(id),label);if(state.selected?.kind+':'+state.selected?.id!==selection)return;detail.innerHTML=kind==='event'?renderEvent(data):renderRun(data);document.querySelectorAll('#detail [data-event-id]').forEach(button=>button.addEventListener('click',()=>select('event',button.dataset.eventId)))}catch(error){if(state.selected?.kind+':'+state.selected?.id!==selection)return;detail.innerHTML='<div class="empty failed">'+esc(error instanceof Error?error.message:String(error))+'<div><button class="retry" id="detail-retry">Retry</button></div></div>';document.querySelector('#detail-retry')?.addEventListener('click',()=>select(kind,id))}
 }
+function showSource(id){const source=state.snapshot.sources.find(value=>value.source===id);const detail=document.querySelector('#detail');detail.innerHTML=renderSource(source);detail.querySelectorAll('[data-event-id]').forEach(button=>button.addEventListener('click',()=>select('event',button.dataset.eventId)))}
 function eventLink(label,event){return event?'<button data-event-id="'+esc(event.id)+'">'+esc(label)+': '+esc(event.type)+' · '+esc(event.id)+'</button>':''}
 function bskyPostUrl(uri){const parts=String(uri??'').split('/');return parts.length===5&&parts[0]==='at:'&&parts[1]===''&&parts[3]==='app.bsky.feed.post'&&parts[2]&&parts[4]?'https://bsky.app/profile/'+encodeURIComponent(parts[2])+'/post/'+encodeURIComponent(parts[4]):''}
 function sourceLabelFromId(source){if(source==='jetstream:cameron-atproto')return 'Cameron · Bluesky';if(source==='x:public-watch')return 'X public watch';if(source==='x:cameron-public')return 'Cameron · X';if(String(source??'').startsWith('telegram:'))return 'Telegram';if(String(source??'').startsWith('filesystem:'))return 'Filesystem';return humanLabel(source??'Source')}
@@ -769,6 +780,50 @@ function sendJson(response: ServerResponse, status: number, value: unknown): voi
   send(response, status, "application/json; charset=utf-8", JSON.stringify(value));
 }
 
+function streamInspectorActivity(
+  store: JazzThoughtStore,
+  request: IncomingMessage,
+  response: ServerResponse,
+): void {
+  response.writeHead(200, {
+    "content-type": "text/event-stream; charset=utf-8",
+    "cache-control": "no-store, no-transform",
+    connection: "keep-alive",
+    "x-accel-buffering": "no",
+    "x-content-type-options": "nosniff",
+  });
+  if (request.method === "HEAD") {
+    response.end();
+    return;
+  }
+  response.write("retry: 2000\n\n");
+  let timer: NodeJS.Timeout | undefined;
+  const notify = () => {
+    if (timer || response.writableEnded || response.destroyed) return;
+    timer = setTimeout(() => {
+      timer = undefined;
+      if (!response.writableEnded && !response.destroyed) {
+        response.write(`event: change\ndata: ${JSON.stringify({ at: new Date().toISOString() })}\n\n`);
+      }
+    }, 100);
+    timer.unref();
+  };
+  const unsubscribe = store.subscribeInspectorActivity(notify);
+  const heartbeat = setInterval(() => {
+    if (!response.writableEnded && !response.destroyed) response.write(": keepalive\n\n");
+  }, 15_000);
+  heartbeat.unref();
+  const close = () => {
+    request.off("aborted", close);
+    response.off("close", close);
+    if (timer) clearTimeout(timer);
+    clearInterval(heartbeat);
+    unsubscribe();
+  };
+  request.once("aborted", close);
+  response.once("close", close);
+}
+
 function send(response: ServerResponse, status: number, contentType: string, body: string): void {
   response.writeHead(status, {
     "content-type": contentType,
diff --git a/test/inspector.test.ts b/test/inspector.test.ts
index 6e1188b..81e5285 100644
--- a/test/inspector.test.ts
+++ b/test/inspector.test.ts
@@ -114,6 +114,8 @@ describe("thought stream inspector", () => {
     expect(pageHtml).not.toContain("No consumer processed this event.");
     expect(pageHtml).toContain("No root activity has been recorded.");
     expect(pageHtml).toContain("No artifacts are available.");
+    expect(pageHtml).toContain("new EventSource('api/live')");
+    expect(pageHtml).toContain("scheduleSnapshotRefresh");
     expect(pageHtml).toContain("cursor:pointer; overflow-wrap:anywhere");
     expect(pageHtml).toContain("font-size:14px; overflow-wrap:anywhere");
     expect(page.headers.get("content-security-policy")).toContain("frame-ancestors 'none'");
@@ -154,6 +156,46 @@ describe("thought stream inspector", () => {
     expect(await mutation.json()).toEqual({ error: "Inspector is read-only" });
   });
 
+  test("streams Jazz activity signals without exposing event content in the stream", async () => {
+    const project = await temporaryProject();
+    roots.push(project);
+    const store = testStore(project);
+    stores.push(store);
+    const server = await startInspectorServer(store, { port: 0 });
+    servers.push(server);
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("Missing inspector address");
+    const controller = new AbortController();
+    const response = await fetch(`http://127.0.0.1:${address.port}/api/live`, { signal: controller.signal });
+    expect(response.status).toBe(200);
+    expect(response.headers.get("content-type")).toBe("text/event-stream; charset=utf-8");
+    expect(response.headers.get("cache-control")).toBe("no-store, no-transform");
+    if (!response.body) throw new Error("Inspector live stream has no body");
+    const frames = sseFrames(response.body.getReader());
+    expect(await frames.next()).toContain("retry: 2000");
+    expect(await frames.next()).toContain("event: change");
+
+    const sourceText = "must-not-enter-the-live-signal";
+    await store.appendEvent({
+      type: "stream.thought.source.rss.item",
+      schemaVersion: 1,
+      source: "rss:live-fixture",
+      sourceKind: "rss",
+      externalId: "live-item-1",
+      idempotencyKey: "live-item-1",
+      occurredAt: "2026-08-10T17:00:00.000Z",
+      actor: "rss:live-fixture",
+      correlationId: "live-item-1",
+      privacy: "public-source",
+      payload: { title: sourceText },
+    });
+    const change = await frames.next();
+    expect(change).toContain("event: change");
+    expect(change).toContain('"at"');
+    expect(change).not.toContain(sourceText);
+    controller.abort();
+  });
+
   test("puts the semantic agent result ahead of lifecycle machinery", async () => {
     const project = await temporaryProject();
     roots.push(project);
@@ -389,3 +431,29 @@ describe("thought stream inspector", () => {
     await expect(startInspectorServer(store, { host: "0.0.0.0", port: 0 })).rejects.toThrow("loopback");
   });
 });
+
+function sseFrames(reader: ReadableStreamDefaultReader<Uint8Array>): { next(): Promise<string> } {
+  const decoder = new TextDecoder();
+  let buffer = "";
+  return {
+    next: async () => {
+      const deadline = Date.now() + 3_000;
+      while (Date.now() < deadline) {
+        const boundary = buffer.indexOf("\n\n");
+        if (boundary >= 0) {
+          const frame = buffer.slice(0, boundary);
+          buffer = buffer.slice(boundary + 2);
+          return frame;
+        }
+        const remaining = deadline - Date.now();
+        const next = await Promise.race([
+          reader.read(),
+          new Promise<never>((_, reject) => setTimeout(() => reject(new Error("Timed out waiting for inspector SSE frame")), remaining)),
+        ]);
+        if (next.done) throw new Error("Inspector SSE stream ended early");
+        buffer += decoder.decode(next.value, { stream: true });
+      }
+      throw new Error("Timed out waiting for inspector SSE frame");
+    },
+  };
+}
-- 
2.51.2