diff --git a/src/projections/activity.ts b/src/projections/activity.ts index af2d9a0..ecc27fb 100644 --- a/src/projections/activity.ts +++ b/src/projections/activity.ts @@ -169,6 +169,57 @@ export async function buildRecentRootActivity( }; } +export async function buildRecentSourceActivity( + store: JazzThoughtStore, + runs: AgentRun[], + source: string, + limit = 20, +): Promise { + const sourceEvents = await store.listEvents({ source }); + const roots = sourceEvents.filter((event) => event.id === event.rootEventId && event.type.startsWith("stream.thought.source.")); + const inputIds = [...new Set(runs.flatMap((run) => run.inputEventIds))]; + const inputEvents = await store.getEvents(inputIds); + const inputsById = new Map(inputEvents.map((event) => [event.id, event])); + const runsByRoot = new Map(); + for (const run of runs) { + const rootIds = new Set(run.inputEventIds + .map((id) => inputsById.get(id)?.rootEventId) + .filter((id): id is string => Boolean(id))); + for (const rootId of rootIds) runsByRoot.set(rootId, [...(runsByRoot.get(rootId) ?? []), run]); + } + const byType: Record = {}; + for (const root of roots) byType[root.type] = (byType[root.type] ?? 0) + 1; + const items = roots.slice(-limit).reverse().map((event): ActivityItem => { + const consumerRuns = (runsByRoot.get(event.id) ?? []).map((run) => ({ + id: run.id, + agentId: run.agentId, + agentVersion: run.agentVersion, + status: run.status, + kind: runKind(run), + description: describeRunResult( + run, + run.inputEventIds.map((id) => inputsById.get(id)).filter((input): input is ThoughtEvent => Boolean(input)), + ), + outputCount: run.outputEventIds.length, + })); + return { + id: event.id, + type: event.type, + occurredAt: event.occurredAt, + source: event.source, + actor: event.actor, + rootEventId: event.rootEventId, + ...(event.parentEventId ? { parentEventId: event.parentEventId } : {}), + privacy: event.privacy, + summary: summarizeRoot(event), + presentation: presentRoot(event), + descendantEventCount: consumerRuns.reduce((total, run) => total + run.outputCount + 1, 0), + consumerRuns, + }; + }); + return { totalEvents: roots.length, byType, items }; +} + export async function rebuildRootActivity(store: JazzThoughtStore, limit = 100): Promise { const projection = await buildRootActivity(store, limit); const events = await store.listEvents(); diff --git a/src/projections/source-health.ts b/src/projections/source-health.ts index d466635..b88864c 100644 --- a/src/projections/source-health.ts +++ b/src/projections/source-health.ts @@ -30,6 +30,7 @@ export interface SourceHealth { }; status: SourceHealthStatus; hasActivityEvidence: boolean; + recordCount: number; cursorId?: string; cursor: JsonObject; lastSuccessAt?: string; @@ -130,6 +131,7 @@ export async function buildSourceHealth(store: JazzThoughtStore, nowMs = Date.no } : {}), status: statusFor(cursor?.lastSuccessAt, cursor?.lastFailureAt, inFlight), hasActivityEvidence, + recordCount: registration?.lastSequence ?? 0, ...(cursor ? { cursorId: cursor.id, cursor: cursor.cursor, diff --git a/src/web/inspector.ts b/src/web/inspector.ts index 159a614..af52f37 100644 --- a/src/web/inspector.ts +++ b/src/web/inspector.ts @@ -1,6 +1,6 @@ import http, { type IncomingMessage, type ServerResponse } from "node:http"; import type { JazzThoughtStore } from "../jazz/store.js"; -import { buildRecentRootActivity, describeRunResult } from "../projections/activity.js"; +import { buildRecentRootActivity, buildRecentSourceActivity, describeRunResult } from "../projections/activity.js"; import { buildSourceHealth } from "../projections/source-health.js"; import { auditRunEvidence } from "../agents/evidence.js"; import type { AgentRun } from "../store/types.js"; @@ -214,6 +214,15 @@ async function handleRequest( sendJson(response, 200, await getArtifactCatalog(store)); return; } + const sourceMatch = url.pathname.match(/^\/api\/sources\/([^/]+)$/); + if (sourceMatch) { + const sourceId = decodeURIComponent(sourceMatch[1]!); + const [runs, sources] = await Promise.all([store.listRuns(), buildSourceHealth(store)]); + const source = sources.find((candidate) => candidate.source === sourceId); + if (!source) return sendJson(response, 404, { error: `Source not found: ${sourceId}` }); + sendJson(response, 200, { source, activity: await buildRecentSourceActivity(store, runs, sourceId) }); + return; + } const artifactContentMatch = url.pathname.match(/^\/api\/artifacts\/([^/]+)\/content$/); if (artifactContentMatch) { const id = decodeURIComponent(artifactContentMatch[1]!); @@ -576,7 +585,7 @@ export function renderInspectorHtml(): string {

thought stream

loading
root activity…
Loading root activity…
Select an observation to see what processed it and what, if anything, happened.