import { createHash } from "node:crypto"; import { RUN_VIEW_SCHEMA, SESSION_MESSAGE_SCHEMA, SESSION_RUN_VIEW_SCHEMA, SESSION_VIEW_SCHEMA, type ClientInteractiveRequest, type WorkflowDisplay, type WorkflowDisplayStatus, type WorkflowRunListPage, type WorkflowRunQueueView, type WorkflowRunSummary, type WorkflowRunView, type WorkflowSessionMessage, type WorkflowSessionNodeRow, type WorkflowSessionProgressUpdate, type WorkflowSessionRunView, type WorkflowSessionView, } from "../client/view.js"; import type { StateDatabase } from "../state/database.js"; import { canonicalJson, parseJson, type JsonValue } from "../state/json.js"; import { WorkflowMessageStore, type SessionSummaryFilter, type WorkflowMessage, type WorkflowMessageSummary, } from "../state/workflow-messages.js"; import type { WorkflowRunQueueStore, WorkflowRunQueueRecord, WorkflowRunQueueViewRecord, } from "../workflows/queue.js"; import { NODE_ID_MAX_BYTES } from "../workflows/schema.js"; import type { WorkflowRunDisplayState, WorkflowRunStore } from "../workflows/store.js"; import type { WorkflowRunState, WorkflowSessionEntryRecord, WorkflowSessionEventRecord, WorkflowStepRecord, WorkflowTraceEvent, WorkflowUpdateRecord, } from "../workflows/types.js"; import { recoveryStopped } from "./recovery.js"; import type { ServerStateStore } from "./state.js"; export const WORKFLOW_PAGE_KINDS = [ "steps", "trace", "trace_at_step", "session_entries", "session_events", "settings", "follow_ups", "updates", "workflow_messages", ] as const; export type WorkflowPageKind = (typeof WORKFLOW_PAGE_KINDS)[number]; type RunPageRequest = { kind: WorkflowPageKind; cursor: number; }; type ContentRecord = { runId: string; path: string; mediaType: "application/json" | "text/plain"; bytes: Buffer; sha256: string; }; const INLINE_CONTENT_BYTES = 16 * 1024; const VIEW_PAGE_BYTES = 64 * 1024; const VIEW_PAGE_ITEMS = 256; const CONTENT_CHUNK_BYTES = 192 * 1024; const CONTENT_CACHE_BYTES = 64 * 1024 * 1024; const VIEW_CACHE_ITEMS = 64; const TERMINAL_VIEW_RETENTION_MS = 60_000; /** * Rows of context the widget window keeps above the node it follows. The widget * is about ten lines, so a few earlier rows are enough for orientation. */ const SESSION_NODE_LEAD = 4; /** * Text one session field may carry: one node row, the workflow name, the run * title, the run error, or one decision choice label. The widget renders each on * one line it truncates to the terminal width, and the 1 MiB client frame budget * is the external limit that requires this bound. Complete text stays available * through the detailed run view and its node history. */ const SESSION_TEXT_BYTES = 4 * 1024; const SESSION_MESSAGE_BATCH = 32; /** * JSON one session detail may carry, such as a monitor estimate, one progress * payload, or the choice labels of one decision. A larger value is reported as * null, left out, or cut at a choice boundary so the single session frame stays * bounded; the detailed run view carries the complete value. */ const SESSION_DETAIL_JSON_BYTES = 8 * 1024; export class ServerViewStore { private readonly contentRecords = new Map(); private readonly listCache = new Map(); private readonly runCache = new Map(); private readonly sessionCache = new Map(); // The selection walk reads message metadata in the durable order of each step, and // a session whose messages did not change keeps the same answer. The memo below // holds that answer under a key of cheap indexed facts, so a periodic view call // over a long history costs the key instead of the history. It carries the same // item bound as the view caches, so its size follows the viewed sessions and not // the number of sessions the server has ever seen. private readonly selectionCache = new Map< string, { key: string; message: WorkflowMessageSummary | undefined; retentionKnown: boolean; retainedRunId: string | undefined; retainedExpiresAt: number | null; } >(); private contentBytes = 0; private activityRevision = 0; private readonly workflowMessages: WorkflowMessageStore; constructor( private readonly state: StateDatabase, private readonly queue: WorkflowRunQueueStore, private readonly serverState: ServerStateStore, private readonly runs: WorkflowRunStore, private readonly hasLiveRunner: (runId: string) => boolean, private readonly hasActiveSessionTurn: (targetSessionId: string) => boolean, ) { this.workflowMessages = serverState.workflowMessages; } noteWorkflowActivityChange(): void { this.activityRevision += 1; } list(cursor = 0, limit?: number): WorkflowRunListPage { return this.state.readTransaction(() => { const pageSize = Math.min(limit ?? VIEW_PAGE_ITEMS, VIEW_PAGE_ITEMS); const current = this.queue.workflowRunListRevision(); const revision = `${current.revision}:${this.workflowActivityRevision()}`; const cacheKey = `${cursor}:${pageSize}`; const cached = this.listCache.get(cacheKey); if (cached?.revision === revision) { refreshCacheEntry(this.listCache, cacheKey, cached); return cached.page; } const loaded = this.queue.listWorkflowRunViews({ offset: cursor, limit: pageSize }); const summaries = loaded.runs.map((run) => { const display = this.projectDisplay( run.runId, this.display(run, { status: run.runStateStatus as WorkflowRunState["status"], paused: run.paused, error: run.errorMessage, }), ); return { runId: run.runId, // One free-form value must not push a list frame past the client limit. // The complete name stays in the run definition. workflowName: boundSessionText(run.workflowName), originSessionId: run.originSessionId, createdAt: run.createdAt, updatedAt: run.updatedAt, display, manifest: manifest(run, display.status), live: display.status === "running" || display.status === "waiting", possiblyInterrupted: run.status === "parked" && display.status !== "paused", } satisfies WorkflowRunSummary; }); const items = byteBoundedForwardPage(summaries, (summary) => toJson(summary)).map( (item) => item as WorkflowRunSummary, ); const page: WorkflowRunListPage = { schema: "pi-workflows.run-list-page.v1", revision, start: cursor, total: loaded.total, items, }; rememberCacheEntry(this.listCache, cacheKey, { revision, page }); return page; }); } run(runId: string): WorkflowRunView | null { return this.state.readTransaction(() => { const version = this.runVersion(runId); const cached = this.runCache.get(runId); if (cached?.version === version) { refreshCacheEntry(this.runCache, runId, cached); return cached.view; } const view = this.readRun(runId); rememberCacheEntry(this.runCache, runId, { version, view }); return view; }); } page(runId: string, request: RunPageRequest): WorkflowRunView | null { return this.state.readTransaction(() => this.readRun(runId, request)); } private readRun(runId: string, page?: RunPageRequest): WorkflowRunView | null { const queue = this.queue.getWorkflowRunView(runId); const counts = this.runs.readRunViewCounts(runId); if (queue === undefined || counts === null) return null; const graphCursor = page?.kind === "steps" ? clampCursor(page.cursor, counts.steps) : Math.max(0, counts.steps - 1); const traceCursor = page?.kind === "trace_at_step" ? this.runs.traceCursorForStep(runId, page.cursor, counts.trace) : page?.kind === "trace" ? page.cursor : undefined; const stepRange = viewRange(counts.steps, page?.kind === "steps" ? page.cursor : undefined); const traceRange = viewRange(counts.trace, traceCursor); const entryRange = viewRange( counts.sessionEntries, page?.kind === "session_entries" ? page.cursor : undefined, ); const eventRange = viewRange( counts.sessionEvents, page?.kind === "session_events" ? page.cursor : undefined, ); const settingsRange = viewRange( counts.settings, page?.kind === "settings" ? page.cursor : undefined, ); const followUpRange = viewRange( counts.followUps, page?.kind === "follow_ups" ? page.cursor : undefined, ); const updateRange = viewRange( counts.updates, page?.kind === "updates" ? page.cursor : undefined, ); const messageRange = viewRange( counts.workflowMessages, page?.kind === "workflow_messages" ? page.cursor : undefined, ); const loaded = this.runs.readRunView(runId, { steps: stepRange, trace: traceRange, sessionEntries: entryRange, sessionEvents: eventRange, settings: settingsRange, followUps: followUpRange, updates: updateRange, graphCursor, }); if (loaded === null) return null; const stepPage = byteBoundedCandidatePage( loaded.state.steps, stepRange.start, counts.steps, page?.kind === "steps" ? page.cursor : undefined, (step) => this.projectStep(runId, step), ); const tracePage = byteBoundedCandidatePage( loaded.traceEvents, traceRange.start, counts.trace, traceCursor, (event) => this.projectTraceEvent(runId, event), ); const entryPage = byteBoundedCandidatePage( loaded.sessionEntries, entryRange.start, counts.sessionEntries, page?.kind === "session_entries" ? page.cursor : undefined, (entry) => this.projectSessionEntry(runId, entry), ); const eventPage = byteBoundedCandidatePage( loaded.sessionEvents, eventRange.start, counts.sessionEvents, page?.kind === "session_events" ? page.cursor : undefined, (event) => this.projectSessionEvent(runId, event), ); const settingsPage = byteBoundedCandidatePage( loaded.settingsScopes, settingsRange.start, counts.settings, page?.kind === "settings" ? page.cursor : undefined, (scope) => this.projectRecordField(runId, scope, "settings"), ); const followUps = loaded.followUpQueue?.followUps ?? []; const followUpPage = byteBoundedCandidatePage( followUps, followUpRange.start, counts.followUps, page?.kind === "follow_ups" ? page.cursor : undefined, (followUp) => this.projectRecordField(runId, followUp, "prompt"), ); const updates = loaded.state.updates ?? []; const updatePage = byteBoundedCandidatePage( updates, updateRange.start, counts.updates, page?.kind === "updates" ? page.cursor : undefined, (update) => this.projectUpdate(runId, update), ); // Message content loads only for the rows the byte-bounded page shows. const messageSummaries = this.workflowMessages.listRunSummaryPage(runId, messageRange); const messagePage = byteBoundedCandidatePage( messageSummaries, messageRange.start, counts.workflowMessages, page?.kind === "workflow_messages" ? page.cursor : undefined, (summary) => this.projectRecordField(runId, this.workflowMessages.materialize(summary), "content"), ); const followUpQueue = loaded.followUpQueue === null ? null : projectFollowUpQueue(loaded.followUpQueue, followUpPage.items); const completeGraphSteps = loaded.graphSteps.map((step) => toCompactStepJson(step)); const completeTakenTransitions = loaded.takenTransitions.map((transition) => toJson(transition), ); const graphSteps = byteBoundedForwardPage(completeGraphSteps, (step) => step); const takenTransitions = byteBoundedForwardPage( completeTakenTransitions, (transition) => transition, ).filter((transition): transition is string => typeof transition === "string"); const graphHistory = this.projectValue(runId, { steps: completeGraphSteps, transitions: completeTakenTransitions, }); const revision = this.presentationRevision(runId); const display = this.projectDisplay(runId, this.display(queue, loaded.state)); return { schema: RUN_VIEW_SCHEMA, runId, revision, runRevision: this.runs.runRevision(runId), display, manifest: manifest(queue, display.status), state: this.projectState(runId, loaded.state, stepPage.items, updatePage.items), workflow: this.projectWorkflow(runId, loaded.snapshot), queue: projectQueue(queue), updates: updatePage.items, graphSteps, graphStepStart: 0, graphStepTotal: loaded.graphSteps.length, takenTransitions, graphHistory, takenTransitionStart: 0, takenTransitionTotal: loaded.takenTransitions.length, graphCursor, stepStart: stepPage.start, stepTotal: stepPage.total, tracePage, session: { binding: toJson(loaded.sessionBinding), entryPage, eventPage, capture: toJson(loaded.sessionCapture), integrity: toJson(loaded.sessionIntegrity), replayCheckpoint: this.projectReplayCheckpoint( runId, this.runs.readSessionReplayCheckpoint(runId, eventPage.start), ), }, settingsScopes: settingsPage.items, settingsStart: settingsPage.start, settingsTotal: settingsPage.total, followUpQueue: toJson(followUpQueue), followUpStart: followUpPage.start, followUpTotal: followUpPage.total, updateStart: updatePage.start, updateTotal: updatePage.total, workflowMessages: messagePage.items, workflowMessageStart: messagePage.start, workflowMessageTotal: messagePage.total, live: display.status === "running" || display.status === "waiting", possiblyInterrupted: queue.status === "parked" && display.status !== "paused", }; } session( sessionId: string, coordinator: { epoch: string; active: boolean; branchReportRequired: boolean } | null = null, nodeCursor?: number, ): WorkflowSessionView { return this.state.readTransaction(() => { const activeQueue = this.queue.findSessionReservationView(sessionId); const retainedRunId = activeQueue === undefined ? this.retainedTerminalRunIdForView(sessionId) : undefined; const runId = activeQueue?.runId ?? retainedRunId; // Selection is one metadata read. The key must name the exact message, // because a status or entry change can land in the same millisecond as the // write that made it current, and then the aggregate facts stay equal. The // message alone identifies the selection, including a session that holds // no live or retained run. const selected = this.currentWorkflowMessageSummary(sessionId); const version = [ runId ?? "-", runId === undefined ? "-" : this.runVersion(runId), this.pendingSessionRevision(sessionId), this.sessionMessageRevision(sessionId), selected === undefined ? "-" : [selected.workflowMessageId, selected.status, selected.piSessionEntryId ?? "-"].join( ":", ), this.openTurnRevision(sessionId), coordinator?.epoch ?? "-", coordinator?.active === true ? "active" : "idle", coordinator?.branchReportRequired === true ? "report" : "reported", nodeCursor === undefined ? "-" : `${nodeCursor}`, ].join("|"); const cached = this.sessionCache.get(sessionId); if (cached?.version === version) { refreshCacheEntry(this.sessionCache, sessionId, cached); return cached.view; } const message = selected === undefined ? undefined : this.workflowMessages.materialize(selected); const openTurn = this.workflowMessages.openTurnsForSession(sessionId)[0]; const view: WorkflowSessionView = { schema: SESSION_VIEW_SCHEMA, sessionId, run: runId === undefined ? null : this.sessionRun(runId, nodeCursor), interaction: this.currentInteraction(sessionId), workflowMessage: message === undefined ? null : this.sessionMessage(message, this.hasCancelledSource(message)), openWorkflowTurn: openTurn === undefined ? null : openTurn, coordinatorEpoch: coordinator?.epoch ?? null, coordinatorActive: coordinator?.active ?? false, branchReportRequired: coordinator?.branchReportRequired ?? false, }; rememberCacheEntry(this.sessionCache, sessionId, { version, view }); return view; }); } /** The one workflow message Pi must inspect, add, finish, or confirm next. */ currentWorkflowMessage(sessionId: string): WorkflowMessage | undefined { const selected = this.currentWorkflowMessageSummary(sessionId); return selected === undefined ? undefined : this.workflowMessages.materialize(selected); } /** * Selection reads message metadata only. Content loads once, for the message * the session must act on. */ /** The one workflow message Pi must inspect, add, finish, or confirm next. */ private currentWorkflowMessageSummary(sessionId: string): WorkflowMessageSummary | undefined { return this.sessionSelection(sessionId, false).message; } /** The retained terminal run the session view shows, or undefined. */ private retainedTerminalRunIdForView(sessionId: string): string | undefined { return this.sessionSelection(sessionId, true).retainedRunId; } /** * The selection and the retained terminal run of one session, held under the * indexed facts both depend on. The held result follows the item bound of the * view caches, so viewing many sessions cannot grow this map without limit, and a * repeat call over a session whose messages did not change reads no message row. */ private sessionSelection( sessionId: string, wantRetention: boolean, ): { message: WorkflowMessageSummary | undefined; retainedRunId: string | undefined } { const now = Date.now(); const key = this.selectionKey(sessionId); const cached = this.selectionCache.get(sessionId); if ( cached !== undefined && cached.key === key && this.selectionHolds(cached, now, wantRetention) ) { refreshCacheEntry(this.selectionCache, sessionId, cached); return cached; } // The retention walk runs at most once per computation, and only when the view // needs it or the selection falls through to its retained terminal step. let retained: { runId: string; expiresAt: number | null } | undefined; let retentionKnown = false; const retention = () => { if (!retentionKnown) { retentionKnown = true; retained = this.retainedTerminalForView(sessionId, now); } return retained; }; const message = this.selectWorkflowMessageSummary(sessionId, retention); if (wantRetention) retention(); const entry = { key, message, retentionKnown, retainedRunId: retained?.runId, retainedExpiresAt: retained?.expiresAt ?? null, }; rememberCacheEntry(this.selectionCache, sessionId, entry); return entry; } /** * Whether a held selection still answers this session. A message inside the * terminal retention window and a retained run both expire with the clock, not * with a stored write, so both are checked before the result is reused. */ private selectionHolds( entry: { message: WorkflowMessageSummary | undefined; retentionKnown: boolean; retainedRunId: string | undefined; retainedExpiresAt: number | null; }, now: number, wantRetention: boolean, ): boolean { if (wantRetention && !entry.retentionKnown) return false; if (entry.retainedRunId !== undefined && entry.retainedExpiresAt !== null) { if (entry.retainedExpiresAt <= now) return false; } return entry.message === undefined || this.retentionStillHolds(entry.message, now); } /** * Whether a delivered terminal message is still inside its retention window. The * window closes with time and not with a stored write, so the held selection is * checked against the clock before it is reused. */ private retentionStillHolds(message: WorkflowMessageSummary, now: number = Date.now()): boolean { if (message.kind !== "terminal" || message.status !== "sent") return true; const turn = this.workflowMessages.latestTurnForMessage(message.workflowMessageId); return Date.parse(turn?.endedAt ?? message.updatedAt) + TERMINAL_VIEW_RETENTION_MS > now; } /** * The indexed facts the selection depends on: the message revision the database * maintains, the run and request revisions, the state of the session runs, and the * open turns of the session. */ private selectionKey(sessionId: string): string { return [ this.sessionMessageRevision(sessionId), this.workflowActivityRevision(), this.pendingSessionRevision(sessionId), this.openTurnRevision(sessionId), this.sessionRunsRevision(sessionId), ].join("|"); } /** * The count and state of the session runs. Eligibility also depends on whether a * run is paused and whether its status is terminal, so the selection key covers * those facts without reading any message. */ private sessionRunsRevision(sessionId: string): string { const row = this.state.connection .prepare( `SELECT count(*) AS runs, coalesce(sum(CASE WHEN r.paused = 1 THEN 1 ELSE 0 END), 0) AS paused, coalesce(sum(CASE WHEN r.status IN ('queued', 'running', 'waiting') THEN 1 ELSE 0 END), 0) AS live FROM run_bindings b JOIN runs r ON r.run_id = b.run_id WHERE b.origin_session_id = ?`, ) .get(sessionId); return isObjectRecord(row) && typeof row.runs === "number" && typeof row.paused === "number" && typeof row.live === "number" ? `${row.runs}:${row.paused}:${row.live}` : "-"; } /** Walk the session message metadata for the one message Pi must act on. */ private selectWorkflowMessageSummary( sessionId: string, retention: () => { runId: string } | undefined, ): WorkflowMessageSummary | undefined { const openTurn = this.workflowMessages.openTurnsForSession(sessionId)[0]; if (openTurn !== undefined) { const open = this.workflowMessages.readSessionSummary(sessionId, openTurn.workflowMessageId); if (open !== undefined) return open; } // Selection walks one bounded batch of metadata at a time, in the order each // step needs. Each step names the candidate filter it can use, so it never // reads a row it must skip and it stops at the first message it needs. for (const message of this.walkSessionSummaries(sessionId, { status: "pending", newestFirst: false, filter: "eligiblePending", })) { if (this.isMessageEligible(message)) return message; } for (const message of this.walkSessionSummaries(sessionId, { status: "sent", newestFirst: true, filter: "piWork", })) { if (!this.needsPiWork(message)) continue; return message; } for (const message of this.walkSessionSummaries(sessionId, { status: "sent", newestFirst: true, filter: "cancelledStep", })) { if (message.kind !== "step" || !this.hasCancelledSource(message)) continue; if (this.workflowMessages.latestTurnForMessage(message.workflowMessageId) !== undefined) { continue; } return message; } // A retained terminal message stays current until its delivery or first turn finishes. const retainedRunId = retention()?.runId; if (retainedRunId !== undefined) { return this.workflowMessages.readSessionSummaryByRun(sessionId, retainedRunId, "terminal"); } return undefined; } /** * Walk one session's message metadata in bounded batches, in durable order or * newest first, with the named candidate filter for that step. The caller stops * at the first message it needs. */ private *walkSessionSummaries( sessionId: string, options: { status: string | null; newestFirst: boolean; kind?: string; filter?: SessionSummaryFilter; cutoff?: number; }, ): Generator { let lastOrder: number | undefined; for (;;) { const batch = this.workflowMessages.listSessionSummaryBatch(sessionId, { status: options.status, newestFirst: options.newestFirst, ...(options.kind === undefined ? {} : { kind: options.kind }), ...(options.filter === undefined ? {} : { filter: options.filter }), ...(options.cutoff === undefined ? {} : { cutoff: options.cutoff }), ...(lastOrder === undefined ? {} : { lastOrder }), limit: SESSION_MESSAGE_BATCH, }); for (const message of batch) yield message; if (batch.length < SESSION_MESSAGE_BATCH) return; lastOrder = batch[batch.length - 1]?.order; if (lastOrder === undefined) return; } } /** Whether Pi still owes delivery confirmation or a model turn for this message. */ private needsPiWork(message: WorkflowMessageSummary): boolean { return this.openWorkflowMessage([message]) !== undefined; } private currentInteraction(sessionId: string): ClientInteractiveRequest | null { const pending = this.serverState.listPendingInteractions(sessionId); const first = pending[0]; if (first === undefined) return null; return this.projectRecordField( first.runId, first, "contract", ) as unknown as ClientInteractiveRequest; } private sessionMessage( message: WorkflowMessage, deliveryCancelled: boolean, ): WorkflowSessionMessage { return { schema: SESSION_MESSAGE_SCHEMA, workflowMessageId: message.workflowMessageId, runId: message.runId, targetSessionId: message.targetSessionId, kind: message.kind, sourceId: message.sourceId, order: message.order, status: message.status, piSessionEntryId: message.piSessionEntryId, createdAt: message.createdAt, updatedAt: message.updatedAt, triggerTurn: message.content.triggerTurn, customType: message.content.customType, display: message.content.display, content: this.projectValue(message.runId, toJson(message.content)), contentDigest: message.contentDigest, deliveryCancelled, }; } /** * The number of message writes this session has seen. The database maintains it, * so the session view cache key notices every insert, update, and delete without * reading the session's stored messages. */ private sessionMessageRevision(sessionId: string): string { const row = this.state.connection .prepare("SELECT revision FROM session_message_revisions WHERE target_session_id = ?") .get(sessionId); if (row === undefined) return "0"; if (!isObjectRecord(row) || typeof row.revision !== "number") { throw new Error("Session message revision is invalid"); } return `${row.revision}`; } private openTurnRevision(sessionId: string): string { const row = this.state.connection .prepare( `SELECT count(*) AS count, COALESCE(max(started_at), 0) AS startedAt FROM workflow_turns WHERE target_session_id = ? AND state = 'started'`, ) .get(sessionId); if (!isObjectRecord(row)) throw new Error("Session turn revision is invalid"); return `${row.count}:${row.startedAt}`; } /** * Bounded current run projection for Pi. It carries the semantic facts for the * status line and widget and leaves complete history to the detailed view. */ sessionRun(runId: string, nodeCursor?: number): WorkflowSessionRunView | null { const queue = this.queue.getWorkflowRunView(runId); const counts = this.runs.readRunViewCounts(runId); if (queue === undefined || counts === null) return null; const empty = { start: 0, limit: 0 }; const loaded = this.runs.readRunView(runId, { steps: viewRange(counts.steps), trace: empty, sessionEntries: empty, sessionEvents: empty, settings: empty, followUps: empty, // Progress and monitor facts come from their own bounded tail reads, so the // compact run never loads the head of a long update set. updates: empty, graphCursor: 0, }); if (loaded === null) return null; const state = loaded.state; const failureNodeId = this.sessionFailureNodeId(state); const display = this.projectDisplay(runId, this.display(queue, state)); const rows = this.sessionNodeRows(runId, loaded.snapshot, state, failureNodeId); const window = boundedNodeWindow( rows, nodeCursor ?? defaultNodeWindowStart(rows, state, failureNodeId), ); return { schema: SESSION_RUN_VIEW_SCHEMA, runId, revision: this.presentationRevision(runId), runRevision: this.runs.runRevision(runId), queue: projectQueue(queue), display, workflowName: boundSessionText(state.workflowName), runTitle: state.runTitle === undefined ? null : boundSessionText(state.runTitle), paused: state.paused === true, currentNode: boundSessionNodeId(state.currentNode), waitingOn: boundSessionNodeId(state.waitingOn), error: state.error === undefined ? null : boundSessionText(state.error), nodes: window.items, nodeStart: window.start, nodeTotal: window.total, progressUpdates: sessionProgressUpdates( this.runs.readCurrentUpdateTail(runId, { type: "progress", limit: MAX_SESSION_PROGRESS_UPDATES, }), ), monitorEstimate: boundedSessionJson(toJson(state.outputs.estimate ?? null)), monitorSchedule: sessionMonitorSchedule( this.runs.readCurrentUpdateTail(runId, { type: "monitor.schedule", key: "next-check", limit: 1, }), ), live: display.status === "running" || display.status === "waiting", possiblyInterrupted: queue.status === "parked" && display.status !== "paused", }; } /** * One row per definition node, in definition order. Attempt facts come from * durable attempts so the rows stay correct for long histories. Detailed text * travels only for the current, waiting, and most recent failed node. */ private sessionNodeRows( runId: string, snapshot: unknown, state: WorkflowRunState, failureNodeId: string | undefined, ): WorkflowSessionNodeRow[] { const attempts = this.state.connection .prepare( `SELECT node_id AS nodeId, status, attempt_number AS attemptNumber, started_at AS startedAt, finished_at AS finishedAt, settings_change_number AS settingsChangeNumber, error_hash AS errorHash FROM node_attempts WHERE run_id = ? ORDER BY attempt_number`, ) .all(runId) .filter(isNodeAttemptRow); const byNode = new Map(); for (const attempt of attempts) { const facts = byNode.get(attempt.nodeId) ?? { attempts: 0, activeStatus: null, lastStatus: null, lastSettingsChangeNumber: null, startedAt: null, durationMs: null, errorHash: null, }; facts.attempts += 1; if (attempt.startedAt !== null && attempt.finishedAt === null) { facts.startedAt = new Date(attempt.startedAt).toISOString(); } if (attempt.startedAt !== null && attempt.finishedAt !== null) { facts.durationMs = Math.max(0, attempt.finishedAt - attempt.startedAt); } if (attempt.settingsChangeNumber !== null) { facts.lastSettingsChangeNumber = attempt.settingsChangeNumber; } if (isActiveAttemptStatus(attempt.status)) facts.activeStatus = attempt.status; else { // Attempts are ordered by number, so a finished attempt means no earlier // attempt is still active. A leftover unfinished row from a superseded // attempt must not report the node as working after a later attempt // succeeded. facts.activeStatus = null; facts.startedAt = null; facts.lastStatus = attempt.status; facts.errorHash = attempt.errorHash; } byNode.set(attempt.nodeId, facts); } const records = nodeRecords(snapshot); const failure = failureNodeId; const rows: WorkflowSessionNodeRow[] = []; for (const [nodeId, node] of Object.entries(records)) { const facts = byNode.get(nodeId); const detail = nodeId === state.currentNode || nodeId === state.waitingOn || nodeId === failure; rows.push({ nodeId, nodeType: typeof node.nodeType === "string" ? node.nodeType : "unknown", actionExecution: sessionActionExecution(node), state: nodeRowState(state, nodeId, facts), attempts: facts?.attempts ?? 0, settingsChangeNumber: nodeId === state.currentNode ? (state.currentSettingsChangeNumber ?? facts?.lastSettingsChangeNumber ?? null) : (facts?.lastSettingsChangeNumber ?? null), statusDetail: nodeId === state.currentNode && typeof state.statusDetail === "string" ? boundNodeText(state.statusDetail) : null, // The widget shows the current node, or the waiting node while the run // is running, as the node it is working on. Both need their start time // so the elapsed segment stays visible. startedAt: nodeId === state.currentNode || nodeId === state.waitingOn ? (facts?.startedAt ?? null) : null, durationMs: facts?.durationMs ?? null, error: detail ? boundNodeText(this.readAttemptError(facts?.errorHash ?? null)) : null, humanDecision: sessionHumanDecision(node, state, nodeId), summary: nodeId === state.waitingOn && typeof node.summary === "string" ? boundNodeText(node.summary) : null, assistantResponse: isAssistantResponseNode(node), outcome: state.results[nodeId]?.outcome ?? null, }); } return rows; } private readAttemptError(hash: Buffer | null): string | null { if (hash === null) return null; const blob = this.state.readBlob(hash); if (blob === undefined) return null; return blob.content.toString("utf8").slice(0, 512); } /** The last node that did not finish cleanly, in completion order. */ private sessionFailureNodeId(state: WorkflowRunState): string | undefined { let failed: string | undefined; for (const [nodeId, result] of Object.entries(state.results)) { if (result.outcome !== "ok") failed = nodeId; } return failed; } clearTerminal(sessionId: string, runId?: string, now: number = Date.now()): string | null { return this.state.transaction(() => { const retained = this.retainedTerminalRunId(sessionId, now); if (retained === undefined) return null; if (runId !== undefined && retained !== runId) { throw new Error(`Retained terminal workflow does not match run ${runId}`); } this.state.connection .prepare( `INSERT INTO session_terminal_views(target_session_id, cleared_run_id, cleared_at) VALUES (?, ?, ?) ON CONFLICT(target_session_id) DO UPDATE SET cleared_run_id = excluded.cleared_run_id, cleared_at = excluded.cleared_at`, ) .run(sessionId, retained, now); this.sessionCache.delete(sessionId); this.selectionCache.delete(sessionId); return retained; }); } content(runId: string, contentPath: string, offset: number): JsonValue | null { const key = contentKey(runId, contentPath); const record = this.contentRecords.get(key) ?? this.recoverContent(runId, contentPath); if (record === undefined) return null; this.contentRecords.delete(key); this.contentRecords.set(key, record); if (!Number.isSafeInteger(offset) || offset < 0 || offset > record.bytes.byteLength) { throw new Error("Workflow content offset is outside the content range"); } const nextOffset = Math.min(record.bytes.byteLength, offset + CONTENT_CHUNK_BYTES); return { schema: "pi-workflows.content-chunk.v1", runId, path: record.path, mediaType: record.mediaType, bytes: record.bytes.byteLength, sha256: record.sha256, offset, nextOffset, complete: nextOffset === record.bytes.byteLength, data: record.bytes.subarray(offset, nextOffset).toString("base64"), }; } private projectStep(runId: string, step: WorkflowStepRecord): JsonValue { const projected = toJson(step); if (!isJsonObject(projected)) return projected; for (const field of ["prompt", "output", "assistantMessage"] as const) { const value = projected[field]; if (value !== undefined) projected[field] = this.projectValue(runId, value); } return projected; } private projectTraceEvent(runId: string, event: WorkflowTraceEvent): JsonValue { return this.projectRecordField(runId, event, "payload"); } private projectSessionEntry(runId: string, entry: WorkflowSessionEntryRecord): JsonValue { return this.projectRecordField(runId, entry, "entry"); } private projectSessionEvent(runId: string, event: WorkflowSessionEventRecord): JsonValue { return this.projectRecordField(runId, event, "payload"); } private projectUpdate(runId: string, update: WorkflowUpdateRecord): JsonValue { return this.projectRecordField(runId, update, "data"); } private projectReplayCheckpoint(runId: string, value: JsonValue | null): JsonValue { return value === null ? null : this.projectValue(runId, value); } private projectRecordField(runId: string, value: unknown, field: string): JsonValue { const projected = toJson(value); if (isJsonObject(projected)) { const fieldValue = projected[field]; if (fieldValue !== undefined) projected[field] = this.projectValue(runId, fieldValue); } return projected; } private projectState( runId: string, stateValue: WorkflowRunState, steps: JsonValue[], updates: JsonValue[], ): JsonValue { const state = toJson(stateValue); if (!isJsonObject(state)) return state; for (const field of ["input", "outputs", "results", "humanDecision", "finalOutput"] as const) { const value = state[field]; if (value !== undefined) state[field] = this.projectValue(runId, value); } state.steps = steps; state.updates = updates; return state; } private projectWorkflow(runId: string, value: unknown): JsonValue { const original = toJson(value); const workflow = escapeArtifactSentinels(original); if (Buffer.byteLength(canonicalJson(workflow)) <= VIEW_PAGE_BYTES * 2) return workflow; if ( !isJsonObject(workflow) || !isJsonObject(workflow.nodes) || typeof workflow.schema !== "string" || typeof workflow.name !== "string" || typeof workflow.startAt !== "string" || !Array.isArray(workflow.edges) ) { return this.registerContent(runId, original, "application/json"); } const allNodeEntries = Object.entries(workflow.nodes); // A node identity larger than one bounded value is left out, never cut, so one // node entry cannot exceed the page budget on its own. The complete definition // stays reachable through the content reference below. const nodeEntries = allNodeEntries.filter( ([nodeId]) => Buffer.byteLength(nodeId, "utf8") <= NODE_ID_MAX_BYTES, ); const boundedNodeEntries = byteBoundedForwardPage(nodeEntries, ([nodeId, node]) => [ nodeId, this.projectWorkflowNode(runId, node), ]); const nodes = Object.fromEntries( boundedNodeEntries.flatMap((entry) => Array.isArray(entry) && typeof entry[0] === "string" && entry[1] !== undefined ? [[entry[0], entry[1]]] : [], ), ); const edges = byteBoundedForwardPage(workflow.edges, (edge) => this.projectValue(runId, edge)); return { schema: workflow.schema, // One free-form value must not push a run frame past the client limit. The // complete definition stays in the run content, reachable by its digest. name: boundSessionText(workflow.name), startAt: workflow.startAt, nodes, nodeStart: 0, nodeTotal: allNodeEntries.length, edges, edgeStart: 0, edgeTotal: workflow.edges.length, content: this.registerContent(runId, original, "application/json"), }; } private projectWorkflowNode(runId: string, value: JsonValue): JsonValue { if (!isJsonObject(value)) return this.projectValue(runId, value); const projected: Record = {}; for (const field of [ "nodeType", "timeoutMs", "statusDetail", "actionExecution", "settingsRoute", "effect", "mountPath", "localNodeId", "includeTransition", ]) { const fieldValue = value[field]; if (fieldValue !== undefined) projected[field] = this.projectValue(runId, fieldValue); } return projected; } private projectDisplay(runId: string, display: WorkflowDisplay): WorkflowDisplay { if ( display.reason === null || Buffer.byteLength(display.reason, "utf8") <= INLINE_CONTENT_BYTES ) { return display; } return { ...display, reason: "Complete workflow failure details are available.", reasonContent: this.registerContent(runId, display.reason, "text/plain"), }; } private projectValue(runId: string, value: JsonValue): JsonValue { const safeValue = escapeArtifactSentinels(value); const mediaType = typeof value === "string" ? "text/plain" : "application/json"; const bytes = mediaType === "text/plain" ? Buffer.from(value as string, "utf8") : Buffer.from(canonicalJson(value), "utf8"); return bytes.byteLength <= INLINE_CONTENT_BYTES ? safeValue : this.registerContent(runId, value, mediaType); } private registerContent( runId: string, value: JsonValue, mediaType: ContentRecord["mediaType"], ): JsonValue { const bytes = Buffer.from( mediaType === "text/plain" ? (value as string) : canonicalJson(value), "utf8", ); const sha256 = createHash("sha256").update(bytes).digest("hex"); const persistedDigest = this.runs.persistViewContent(runId, bytes, mediaType); if (persistedDigest !== sha256) throw new Error("Workflow view content digest changed"); const extension = mediaType === "text/plain" ? "txt" : "json"; const contentPath = `artifacts/sha256/${sha256}.${extension}`; this.rememberContent({ runId, path: contentPath, mediaType, bytes, sha256, }); return { $artifact: { path: contentPath, mediaType, bytes: bytes.byteLength, sha256, opaque: true, }, }; } private rememberContent(record: ContentRecord): ContentRecord { const key = contentKey(record.runId, record.path); const previous = this.contentRecords.get(key); if (previous !== undefined) this.contentBytes -= previous.bytes.byteLength; this.contentRecords.delete(key); this.contentRecords.set(key, record); this.contentBytes += record.bytes.byteLength; while (this.contentBytes > CONTENT_CACHE_BYTES && this.contentRecords.size > 1) { const oldest = this.contentRecords.entries().next().value as | [string, ContentRecord] | undefined; if (oldest === undefined) break; this.contentRecords.delete(oldest[0]); this.contentBytes -= oldest[1].bytes.byteLength; } return record; } private recoverContent(runId: string, contentPath: string): ContentRecord | undefined { const match = /^artifacts\/sha256\/([0-9a-f]{64})\.(json|txt)$/u.exec(contentPath); if (match === null) return undefined; const mediaType = match[2] === "txt" ? "text/plain" : "application/json"; const blob = this.runs.readContentBlob(runId, match[1] as string, mediaType); if (blob === undefined) return undefined; return this.rememberContent({ runId, path: contentPath, mediaType, bytes: blob.content, sha256: match[1] as string, }); } clearConnection(_connectionId: string): void { // Coordinator fencing is process-local in the server. Durable turn state is // reconciled from the Pi branch after the next coordinator connects. } /** The retained terminal run of one session, if the session view shows one. */ private retainedTerminalRunId(sessionId: string, now: number = Date.now()): string | undefined { return this.retainedTerminalForView(sessionId, now)?.runId; } /** * The retained terminal run of one session, with the moment its retention window * closes. A run kept by a waiting message or a triggering turn has no such moment, * because its state and not the clock holds it. */ private retainedTerminalForView( sessionId: string, now: number, ): { runId: string; expiresAt: number | null } | undefined { // The walk reads bounded batches of retained terminal metadata, newest first, // and stops at the first retained run. Its candidate filter excludes every // terminal message whose retention window has passed. const cutoff = now - TERMINAL_VIEW_RETENTION_MS; for (const message of this.walkSessionSummaries(sessionId, { status: null, newestFirst: true, kind: "terminal", filter: "retainedTerminal", cutoff, })) { const run = this.state.connection .prepare("SELECT status FROM runs WHERE run_id = ?") .get(message.runId); if (!isObjectRecord(run) || !isTerminalStatus(run.status)) continue; const clear = this.state.connection .prepare( `SELECT cleared_run_id AS clearedRunId, cleared_at AS clearedAt FROM session_terminal_views WHERE target_session_id = ?`, ) .get(sessionId); if ( isObjectRecord(clear) && clear.clearedRunId === message.runId && typeof clear.clearedAt === "number" && clear.clearedAt >= Date.parse(message.createdAt) ) { continue; } if (message.status === "pending") return { runId: message.runId, expiresAt: null }; if (message.status !== "sent") continue; const turn = this.workflowMessages.latestTurnForMessage(message.workflowMessageId); if ( message.triggerTurn && !recoveryStopped(this.state, message.workflowMessageId) && (turn === undefined || turn.state === "started") ) { return { runId: message.runId, expiresAt: null }; } const expiresAt = Date.parse(turn?.endedAt ?? message.updatedAt) + TERMINAL_VIEW_RETENTION_MS; if (expiresAt > now) return { runId: message.runId, expiresAt }; } return undefined; } hasCancelledSource(message: WorkflowMessageSummary): boolean { if (message.kind === "terminal") return message.triggerTurn && recoveryStopped(this.state, message.workflowMessageId); if (message.kind === "step") { const request = this.state.connection .prepare("SELECT status FROM interactive_requests WHERE request_id = ? AND run_id = ?") .get(message.sourceId, message.runId); return isObjectRecord(request) && request.status === "cancelled"; } if (message.kind === "followUp") { const source = this.state.connection .prepare("SELECT status FROM workflow_follow_ups WHERE follow_up_id = ? AND run_id = ?") .get(message.sourceId, message.runId); return ( isObjectRecord(source) && (source.status === "cancelled" || source.status === "removed") ); } return false; } private isMessageEligible(message: WorkflowMessageSummary): boolean { if (message.status !== "pending") return false; if (message.kind === "step" || message.kind === "decision") { const request = this.state.connection .prepare( `SELECT i.status, r.paused FROM interactive_requests i JOIN runs r ON r.run_id = i.run_id WHERE i.request_id = ? AND i.run_id = ?`, ) .get(message.sourceId, message.runId); if (!isObjectRecord(request) || request.status !== "pending") return false; return message.kind === "decision" || request.paused === 0; } if (message.kind === "notification") return true; if (message.kind === "terminal") { const reservation = this.state.connection .prepare( `SELECT 1 FROM run_bindings b JOIN runs r ON r.run_id = b.run_id WHERE b.origin_session_id = ? AND r.run_id <> ? AND r.status IN ('queued', 'running', 'waiting') LIMIT 1`, ) .get(message.targetSessionId, message.runId); if (reservation !== undefined) return false; const run = this.state.connection .prepare("SELECT status FROM runs WHERE run_id = ?") .get(message.runId); return ( isObjectRecord(run) && isTerminalStatus(run.status) && (!message.triggerTurn || !recoveryStopped(this.state, message.workflowMessageId)) ); } const source = this.state.connection .prepare( `SELECT f.run_id AS runId, f.order_number AS orderNumber, f.status FROM workflow_follow_ups f WHERE f.follow_up_id = ?`, ) .get(message.sourceId); if ( !isObjectRecord(source) || source.status !== "queued" || typeof source.orderNumber !== "number" || typeof source.runId !== "string" ) { return false; } const run = this.state.connection .prepare("SELECT status FROM runs WHERE run_id = ?") .get(source.runId); if (!isObjectRecord(run) || run.status !== "completed") return false; const terminal = this.workflowMessages .listRunSummaries(source.runId) .filter((candidate) => candidate.kind === "terminal" && candidate.status === "sent") .at(-1); if (terminal === undefined) return false; if (terminal.triggerTurn) { const turn = this.workflowMessages.latestTurnForMessage(terminal.workflowMessageId); if ( turn?.state !== "ended" || turn.stopReason !== "completed" || recoveryStopped(this.state, terminal.workflowMessageId) ) return false; } const prior = this.state.connection .prepare( `SELECT follow_up_id AS followUpId, status FROM workflow_follow_ups WHERE run_id = ? AND order_number < ? ORDER BY order_number`, ) .all(source.runId, source.orderNumber); for (const item of prior) { if ( !isObjectRecord(item) || typeof item.followUpId !== "string" || typeof item.status !== "string" ) { return false; } if (item.status === "removed" || item.status === "cancelled") continue; const priorMessage = this.workflowMessages.latestForSource("followUp", item.followUpId); if ( priorMessage === undefined || this.workflowMessages.latestTurnForMessage(priorMessage.workflowMessageId)?.state !== "ended" ) { return false; } } const reservation = this.state.connection .prepare( `SELECT 1 AS present FROM run_bindings b JOIN runs r ON r.run_id = b.run_id WHERE b.origin_session_id = ? AND r.run_id <> ? AND r.status IN ('queued', 'running', 'waiting') LIMIT 1`, ) .get(message.targetSessionId, message.runId); return reservation === undefined; } private openWorkflowMessage( messages: readonly WorkflowMessageSummary[], ): WorkflowMessageSummary | undefined { for (const message of [...messages].reverse()) { if (message.status !== "sent") continue; if (message.kind === "step") { const request = this.state.connection .prepare( `SELECT i.status, r.paused FROM interactive_requests i JOIN runs r ON r.run_id = i.run_id WHERE i.request_id = ?`, ) .get(message.sourceId); if (isObjectRecord(request) && request.status === "pending" && request.paused === 0) { return message; } } else if ( message.kind === "followUp" || (message.kind === "terminal" && message.triggerTurn && !recoveryStopped(this.state, message.workflowMessageId)) ) { const turn = this.workflowMessages.latestTurnForMessage(message.workflowMessageId); if (turn === undefined || turn.state === "started") return message; } } return undefined; } private workflowActivityRevision(runId?: string): string { const row = runId === undefined ? this.state.connection .prepare( `SELECT COALESCE((SELECT max(updated_at) FROM workflow_messages), 0) AS messageUpdatedAt, COALESCE((SELECT max(COALESCE(ended_at, started_at)) FROM workflow_turns), 0) AS turnUpdatedAt`, ) .get() : this.state.connection .prepare( `SELECT COALESCE((SELECT max(updated_at) FROM workflow_messages WHERE run_id = ?), 0) AS messageUpdatedAt, COALESCE((SELECT max(COALESCE(ended_at, started_at)) FROM workflow_turns WHERE run_id = ?), 0) AS turnUpdatedAt`, ) .get(runId, runId); return isObjectRecord(row) && typeof row.messageUpdatedAt === "number" && typeof row.turnUpdatedAt === "number" ? `${row.messageUpdatedAt}:${row.turnUpdatedAt}${this.activityRevision === 0 ? "" : `-${this.activityRevision}`}` : `0:0${this.activityRevision === 0 ? "" : `-${this.activityRevision}`}`; } private display( queue: WorkflowRunQueueRecord | WorkflowRunQueueViewRecord, state?: WorkflowRunDisplayState | WorkflowRunState | null, ): WorkflowDisplay { return reduceWorkflowDisplay({ queueStatus: queue.status, durableStatus: state?.status, paused: state?.paused === true, ambiguous: this.hasAmbiguousEffect(queue.runId), runnerActive: this.hasLiveRunner(queue.runId), originTurnActive: this.hasActivity(queue.runId), pendingRequestKind: this.pendingRequestKind(queue.runId), requestDeliveryConfirmed: this.requestDeliveryConfirmed(queue.runId), errorMessage: state?.error ?? queue.errorMessage, }); } private hasActivity(runId: string): boolean { const row = this.state.connection .prepare( `SELECT t.target_session_id AS targetSessionId FROM workflow_turns t JOIN workflow_messages m ON m.workflow_message_id = t.workflow_message_id WHERE t.run_id = ? AND t.state = 'started' AND m.kind IN ('step', 'terminal', 'followUp') LIMIT 1`, ) .get(runId); return ( isObjectRecord(row) && typeof row.targetSessionId === "string" && this.hasActiveSessionTurn(row.targetSessionId) ); } private pendingRequestKind(runId: string): WorkflowDisplayFacts["pendingRequestKind"] { const row = this.state.connection .prepare("SELECT kind FROM interactive_requests WHERE run_id = ? AND status = 'pending'") .get(runId) as | { kind: Exclude } | undefined; return row?.kind ?? null; } private requestDeliveryConfirmed(runId: string): boolean { const row = this.state.connection .prepare( `SELECT m.status FROM interactive_requests i JOIN workflow_messages m ON m.source_id = i.request_id AND m.kind = 'step' WHERE i.run_id = ? AND i.status = 'pending' ORDER BY m.order_number DESC LIMIT 1`, ) .get(runId) as { status: string } | undefined; return row?.status === "sent"; } private hasAmbiguousEffect(runId: string): boolean { const row = this.state.connection .prepare( `SELECT 1 AS present FROM effects e JOIN runs r ON r.resource_id = e.source_resource_id WHERE r.run_id = ? AND e.status = 'ambiguous' LIMIT 1`, ) .get(runId); return row !== undefined; } private runVersion(runId: string): string { const row = this.state.connection .prepare( `SELECT res.revision, r.status AS runStatus, r.paused, q.updated_at AS updatedAt, COALESCE(v.presentation_revision, 0) AS presentationRevision FROM runs r JOIN resources res ON res.resource_id = r.resource_id JOIN run_queue q ON q.run_id = r.run_id LEFT JOIN viewer_runs v ON v.run_id = r.run_id WHERE r.run_id = ?`, ) .get(runId); if (!isRunVersionRow(row)) return "missing"; return [ row.revision, row.updatedAt, row.presentationRevision, row.runStatus, row.paused, this.workflowActivityRevision(runId), this.hasLiveRunner(runId), this.hasActivity(runId), this.pendingRequestKind(runId), this.requestDeliveryConfirmed(runId), this.hasAmbiguousEffect(runId), ].join(":"); } private pendingSessionRevision(sessionId: string): string { const row = this.state.connection .prepare( `SELECT count(*) AS count, COALESCE(sum(revision), 0) AS revisionSum, COALESCE(max(updated_at), 0) AS updatedAt FROM interactive_requests WHERE target_session_id = ? AND status = 'pending'`, ) .get(sessionId); if (!isSessionRevisionRow(row)) throw new Error("Session view revision is invalid"); return `${row.count}:${row.revisionSum}:${row.updatedAt}`; } private presentationRevision(runId: string): number { const row = this.state.connection .prepare(`SELECT presentation_revision AS revision FROM viewer_runs WHERE run_id = ?`) .get(runId); return isRevisionRow(row) ? row.revision : 0; } } export type WorkflowDisplayFacts = { queueStatus: WorkflowRunQueueRecord["status"]; durableStatus: WorkflowRunState["status"] | undefined; paused: boolean; ambiguous: boolean; runnerActive: boolean; originTurnActive: boolean; pendingRequestKind: "agent" | "assistant" | "checkpoint" | "decision" | null; requestDeliveryConfirmed: boolean; errorMessage: string | null; }; export function reduceWorkflowDisplay(facts: WorkflowDisplayFacts): WorkflowDisplay { let status: WorkflowDisplayStatus; const activity: WorkflowDisplay["activity"] = facts.runnerActive ? "supervised_runner" : facts.originTurnActive ? "origin_turn" : null; let reason: string | null = null; if ( facts.durableStatus === "completed" || facts.durableStatus === "failed" || facts.durableStatus === "timed_out" || facts.durableStatus === "cancelled" ) { status = facts.durableStatus; reason = facts.ambiguous ? "The run is terminal, but an external effect still needs explicit recovery." : facts.originTurnActive ? "Execution has ended. The model is reviewing the result and safe next actions." : facts.errorMessage; } else if (facts.ambiguous) { status = "ambiguous"; reason = "An external effect needs explicit recovery."; } else if (facts.paused) { status = "paused"; reason = "The workflow is durably paused."; } else if (activity !== null) { status = "running"; reason = facts.originTurnActive ? "The agent is working on the workflow step." : "The workflow runner is working."; } else if (facts.pendingRequestKind !== null || facts.durableStatus === "waiting") { status = "waiting"; reason = facts.pendingRequestKind === "decision" ? "The workflow needs a protected human decision." : facts.pendingRequestKind === "checkpoint" ? "The workflow needs a checkpoint answer." : facts.pendingRequestKind === "agent" ? "The workflow needs its assigned agent result." : facts.pendingRequestKind === "assistant" ? "The workflow needs its assigned visible response." : "The workflow is waiting."; if (facts.pendingRequestKind === "agent" || facts.pendingRequestKind === "assistant") { if (!facts.requestDeliveryConfirmed) reason = "A step is pending delivery. It starts a new model turn after this turn ends."; } } else if (facts.queueStatus === "parked" || facts.queueStatus === "queued") { status = "queued"; reason = facts.queueStatus === "parked" ? (facts.errorMessage ?? "The workflow is ready to resume.") : null; } else if (facts.queueStatus === "done") { status = "completed"; } else if (facts.queueStatus === "failed" || facts.queueStatus === "cancelled") { status = facts.queueStatus; reason = facts.errorMessage; } else { status = "running"; } const controls: WorkflowDisplay["controls"] = []; if (status === "running" || status === "waiting") controls.push("pause", "cancel"); else if (status === "paused") controls.push("resume", "cancel"); else if (status === "queued") { if (facts.queueStatus === "parked") controls.push("resume"); controls.push("cancel"); } if (status === "waiting" || status === "running") { if (facts.pendingRequestKind === "checkpoint") controls.push("answer"); if (facts.pendingRequestKind === "decision") controls.push("human-answer"); if (facts.pendingRequestKind === "agent") controls.push("update", "submit"); } if (["completed", "failed", "timed_out"].includes(status) && facts.originTurnActive) controls.push("cancel"); if (facts.ambiguous) controls.push("review"); return { status, activity, controls, reason }; } function manifest( run: WorkflowRunQueueRecord | WorkflowRunQueueViewRecord, status: WorkflowDisplayStatus, ): JsonValue { return { schema: "pi-workflows.run-manifest.v1", runId: run.runId, // The complete name stays in the run definition, so the list frame cannot // grow with one free-form value. workflowName: boundSessionText(run.workflowName), workflowSource: workflowRootSource(run.workflowSource), startedAt: run.startedAt ?? run.createdAt, ...(run.finishedAt === null ? {} : { finishedAt: run.finishedAt }), status, traceSchema: "pi-workflows.trace-event.v1", paths: { workflow: "server", state: "server", trace: "server", }, }; } // The widget shows a bounded set of progress rows. A run keeps up to 1,024 // current updates, so the newest keys are the ones that matter. const MAX_SESSION_PROGRESS_UPDATES = 16; type NodeRecord = { nodeType?: unknown; actionExecution?: unknown; summary?: unknown; humanDecision?: unknown; expectedOutput?: unknown; }; type NodeAttemptFacts = { attempts: number; activeStatus: string | null; lastStatus: string | null; lastSettingsChangeNumber: number | null; startedAt: string | null; durationMs: number | null; errorHash: Buffer | null; }; type NodeAttemptRow = { nodeId: string; status: string; attemptNumber: number; startedAt: number | null; finishedAt: number | null; settingsChangeNumber: number | null; errorHash: Buffer | null; }; function isNodeAttemptRow(value: unknown): value is NodeAttemptRow { if (!isJsonObject(value)) return false; return ( typeof value.nodeId === "string" && typeof value.status === "string" && typeof value.attemptNumber === "number" && (value.startedAt === null || typeof value.startedAt === "number") && (value.finishedAt === null || typeof value.finishedAt === "number") && (value.settingsChangeNumber === null || typeof value.settingsChangeNumber === "number") && (value.errorHash === null || Buffer.isBuffer(value.errorHash)) ); } function isActiveAttemptStatus(status: string): boolean { return status === "pending" || status === "running" || status === "waiting"; } function nodeRecords(snapshot: unknown): Record { if (!isJsonObject(snapshot) || !isJsonObject(snapshot.nodes)) return {}; const records: Record = {}; for (const [nodeId, node] of Object.entries(snapshot.nodes)) { if (isJsonObject(node)) records[nodeId] = node as NodeRecord; } return records; } function nodeRowState( state: WorkflowRunState, nodeId: string, facts: NodeAttemptFacts | undefined, ): WorkflowSessionNodeRow["state"] { if (facts?.activeStatus === "running") return "running"; if (facts?.activeStatus === "waiting") return "waiting"; if (facts?.activeStatus === "pending") return "pending"; const result = state.results[nodeId]; if (result !== undefined) return result.outcome === "ok" ? "ok" : "failed"; if (facts?.lastStatus === "completed") return "ok"; if (facts !== undefined && facts.lastStatus !== null) return "failed"; return "pending"; } function sessionActionExecution(node: NodeRecord): "function" | "shell" | null { return node.actionExecution === "shell" || node.actionExecution === "function" ? node.actionExecution : null; } function isAssistantResponseNode(node: NodeRecord): boolean { return ( node.nodeType === "agent" && isJsonObject(node.expectedOutput) && node.expectedOutput.kind === "assistant-message" ); } function sessionHumanDecision( node: NodeRecord, state: WorkflowRunState, nodeId: string, ): WorkflowSessionNodeRow["humanDecision"] { const human = isJsonObject(node.humanDecision) ? node.humanDecision : undefined; if (human === undefined) return null; const request = humanDecisionRequest(state.finalOutput); const current = request !== undefined && request.nodeId === nodeId ? request : undefined; const audience = current?.audience ?? (typeof human.audience === "string" ? human.audience : "human"); const choices = isJsonObject(human.choices) ? sessionDecisionChoices( Object.entries(human.choices).flatMap(([value, choice]) => isJsonObject(choice) && typeof choice.label === "string" ? [{ value, label: choice.label }] : [], ), ) : []; const choiceValue = state.humanDecision !== undefined && state.humanDecision.nodeId === nodeId ? state.humanDecision.response.choice : null; return { audience, summary: current?.summary === undefined ? null : boundNodeText(current.summary), choices, choiceValue, presentationDigest: current?.presentationDigest ?? null, }; } /** * Decision choices one session node row may carry. The widget joins the labels * into one line it truncates to the terminal width, and the 1 MiB frame budget * requires the bound. The complete decision stays available through the detailed * run view and its decision record. */ function sessionDecisionChoices( entries: readonly { value: string; label: string }[], ): { value: string; label: string }[] { const choices: { value: string; label: string }[] = []; let bytes = 0; for (const entry of entries) { const label = boundSessionText(entry.label); const size = Buffer.byteLength(label, "utf8") + Buffer.byteLength(entry.value, "utf8"); // Keep at least one choice so the row still names a choice the human sees. if (choices.length > 0 && bytes + size > SESSION_DETAIL_JSON_BYTES) break; bytes += size; choices.push({ value: entry.value, label }); } return choices; } function humanDecisionRequest( value: unknown, ): | { nodeId: string; audience: string; summary: string | null; presentationDigest: string | null } | undefined { if (!isJsonObject(value)) return undefined; if (value.schema !== "pi-workflows.human-decision-request.v1") return undefined; if (typeof value.nodeId !== "string" || typeof value.audience !== "string") return undefined; // A node identity and an audience name identify state, so one that cannot travel // in a bounded frame is left out and never cut. The summary is free-form display // text, which is cut at the shared session bound. if ( Buffer.byteLength(value.nodeId, "utf8") > NODE_ID_MAX_BYTES || Buffer.byteLength(value.audience, "utf8") > SESSION_TEXT_BYTES ) { return undefined; } const presentation = isJsonObject(value.presentation) ? value.presentation : undefined; return { nodeId: value.nodeId, audience: value.audience, summary: typeof presentation?.summary === "string" ? boundSessionText(presentation.summary) : null, presentationDigest: typeof value.presentationDigest === "string" && Buffer.byteLength(value.presentationDigest, "utf8") <= SESSION_TEXT_BYTES ? value.presentationDigest : null, }; } function sessionProgressUpdates( updates: readonly WorkflowUpdateRecord[], ): WorkflowSessionProgressUpdate[] { const progress: WorkflowSessionProgressUpdate[] = []; for (const update of updates) { if (update.type !== "progress") continue; const data = boundedSessionJson(toJson(update.data)); // A payload larger than one session detail is left out rather than cut, so // the compact view never carries a value that looks complete but is not. if (data === null) continue; progress.push({ key: update.key, at: update.at, data }); } // Updates arrive in run revision order, so a bounded set keeps the newest keys // instead of the oldest ones. return progress.slice(-MAX_SESSION_PROGRESS_UPDATES); } function sessionMonitorSchedule( updates: readonly WorkflowUpdateRecord[], ): { nextCheckAt: string; recordedAt: string } | null { for (const update of updates) { if (update.type !== "monitor.schedule" || update.key !== "next-check") continue; if (!isJsonObject(update.data) || typeof update.data.nextCheckAt !== "string") continue; // A schedule that cannot travel in a bounded frame is left out rather than cut, // because a cut instant is another instant. The recorded update stays available // through the run update page. if (Buffer.byteLength(update.data.nextCheckAt, "utf8") > SESSION_TEXT_BYTES) return null; return { nextCheckAt: update.data.nextCheckAt, recordedAt: update.at }; } return null; } function projectQueue( run: WorkflowRunQueueRecord | WorkflowRunQueueViewRecord, ): WorkflowRunQueueView { return { runId: run.runId, workflowName: boundSessionText(run.workflowName), workflowSourceRef: run.workflowSourceRef, initialized: run.initialized, definitionDigest: run.definitionDigest, status: run.status, originSessionId: run.originSessionId, executionMode: run.executionMode, parentRunId: run.parentRunId, rootRunId: run.rootRunId, lineageKind: run.lineageKind, restartNumber: run.restartNumber, parentRunRevision: run.parentRunRevision, errorCode: run.errorCode, createdAt: run.createdAt, updatedAt: run.updatedAt, startedAt: run.startedAt, finishedAt: run.finishedAt, }; } function workflowRootSource(value: unknown): JsonValue { const sourceSet = toJson(value); if (!isJsonObject(sourceSet)) throw new Error("Workflow queue source set is invalid"); const root = sourceSet.root; if (root === undefined || !isJsonObject(root)) { throw new Error("Workflow queue source set is invalid"); } return root; } function isObjectRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } function isTerminalStatus( value: unknown, ): value is "completed" | "failed" | "timed_out" | "cancelled" { return ( value === "completed" || value === "failed" || value === "timed_out" || value === "cancelled" ); } function contentKey(runId: string, contentPath: string): string { return `${runId}\u0000${contentPath}`; } function escapeArtifactSentinels(value: JsonValue): JsonValue { if (Array.isArray(value)) return value.map(escapeArtifactSentinels); if (!isJsonObject(value)) return value; const escaped = Object.fromEntries( Object.entries(value).map(([key, item]) => [key, escapeArtifactSentinels(item)]), ) as JsonValue; return Object.keys(value).length === 1 && (Object.hasOwn(value, "$artifact") || Object.hasOwn(value, "$escaped")) ? { $escaped: escaped } : escaped; } function projectFollowUpQueue(value: unknown, items: JsonValue[]): JsonValue { const queue = toJson(value); if (!isJsonObject(queue)) return queue; delete queue.followUps; queue.items = items; return queue; } function refreshCacheEntry(cache: Map, key: K, value: V): void { cache.delete(key); cache.set(key, value); } function rememberCacheEntry(cache: Map, key: K, value: V): void { refreshCacheEntry(cache, key, value); while (cache.size > VIEW_CACHE_ITEMS) { const oldest = cache.keys().next().value as K | undefined; if (oldest === undefined) break; cache.delete(oldest); } } function viewRange(total: number, cursor?: number): { start: number; limit: number } { const start = workflowPageStart(total, cursor); return { start, limit: Math.min(VIEW_PAGE_ITEMS, Math.max(0, total - start)) }; } function byteBoundedForwardPage( values: readonly T[], project: (value: T) => JsonValue, ): JsonValue[] { const items: JsonValue[] = []; let bytes = 0; for (const value of values) { if (items.length >= VIEW_PAGE_ITEMS) break; const item = project(value); const itemBytes = Buffer.byteLength(canonicalJson(item)) + (items.length === 0 ? 0 : 1); if (items.length > 0 && bytes + itemBytes > VIEW_PAGE_BYTES) break; items.push(item); bytes += itemBytes; } return items; } function byteBoundedCandidatePage( values: readonly T[], candidateStart: number, total: number, requestedCursor: number | undefined, project: (value: T) => JsonValue, ): { start: number; total: number; items: JsonValue[] } { if (values.length === 0) return { start: candidateStart, total, items: [] }; const globalCursor = clampCursor(requestedCursor ?? Math.max(0, total - 1), total); const localCursor = Math.min(Math.max(0, globalCursor - candidateStart), values.length - 1); const page = byteBoundedPage(values, localCursor, project); return { start: candidateStart + page.start, total, items: page.items }; } function byteBoundedPage( values: readonly T[], requestedCursor: number | undefined, project: (value: T) => JsonValue, ): { start: number; total: number; items: JsonValue[] } { const total = values.length; if (total === 0) return { start: 0, total: 0, items: [] }; const cursor = clampCursor(requestedCursor ?? total - 1, total); const selected = project(values[cursor] as T); const selectedBytes = Buffer.byteLength(canonicalJson(selected)); const indexed = new Map([[cursor, selected]]); let pageBytes = selectedBytes; let left = cursor - 1; let right = cursor + 1; let leftBlocked = false; let rightBlocked = false; let preferLeft = true; while (indexed.size < VIEW_PAGE_ITEMS && (!leftBlocked || !rightBlocked)) { const index = preferLeft ? left : right; const inRange = index >= 0 && index < total; if (!inRange) { if (preferLeft) leftBlocked = true; else rightBlocked = true; } else { const item = project(values[index] as T); const itemBytes = Buffer.byteLength(canonicalJson(item)) + 1; if (pageBytes + itemBytes > VIEW_PAGE_BYTES) { if (preferLeft) leftBlocked = true; else rightBlocked = true; } else { indexed.set(index, item); pageBytes += itemBytes; if (preferLeft) left -= 1; else right += 1; } } preferLeft = !preferLeft; } const ordered = [...indexed.entries()].sort( ([leftIndex], [rightIndex]) => leftIndex - rightIndex, ); return { start: ordered[0]?.[0] ?? cursor, total, items: ordered.map(([, item]) => item), }; } export function toCompactStepJson(step: WorkflowStepRecord): JsonValue { return toJson({ attemptId: step.attemptId, nodeId: step.nodeId, nodeType: step.nodeType, outcome: step.outcome, startedAt: step.startedAt, finishedAt: step.finishedAt, prompt: null, output: null, ...(step.settingsScopeId === undefined ? {} : { settingsScopeId: step.settingsScopeId }), ...(step.settingsChangeNumber === undefined ? {} : { settingsChangeNumber: step.settingsChangeNumber }), ...(step.settingsHash === undefined ? {} : { settingsHash: step.settingsHash }), }); } export function workflowPageStart(total: number, cursor?: number): number { if (total <= 256) return 0; if (cursor === undefined) return total - 256; const center = clampCursor(cursor, total); return Math.min(Math.max(0, center - 128), total - 256); } function clampCursor(cursor: number, total: number): number { return total === 0 ? 0 : Math.min(cursor, total - 1); } /** Text one session node row may carry, bounded by the client frame budget. */ function boundNodeText(value: string | null): string | null { return value === null ? null : boundSessionText(value); } /** Free-form session text, bounded for the single frame budget. */ function boundSessionText(value: string): string { const bytes = Buffer.from(value, "utf8"); if (bytes.byteLength <= SESSION_TEXT_BYTES) return value; // The cut must fall on a character boundary. A cut inside a multi-byte sequence // would end the value with a replacement character the widget shows as text. let end = SESSION_TEXT_BYTES; while (end > 0 && ((bytes[end] ?? 0) & 0xc0) === 0x80) end -= 1; return bytes.subarray(0, end).toString("utf8"); } /** * A node identity the session run reports as the node it works on. The window * leaves out a row whose own bytes exceed the frame budget, so an identity of * that size is left out here too. A cut identity would match no row and would * name a node that does not exist. The complete identity stays in the detailed * run view. */ function boundSessionNodeId(value: string | null | undefined): string | null { if (value === null || value === undefined) return null; return Buffer.byteLength(canonicalJson(value), "utf8") > NODE_ID_MAX_BYTES ? null : value; } /** One JSON detail, or null when the detail is too large for the session frame. */ function boundedSessionJson(value: JsonValue): JsonValue { return Buffer.byteLength(canonicalJson(value), "utf8") <= SESSION_DETAIL_JSON_BYTES ? value : null; } /** * The window follows the node the widget shows as working, with a little lead. * A run with no working node shows the node that carries the failure the widget * highlights, or its end, so the useful rows are in the first window. */ function defaultNodeWindowStart( rows: readonly WorkflowSessionNodeRow[], state: WorkflowRunState, failureNodeId?: string, ): number { // The widget projects the same rows, so a row that reports work is the row it // shows as working even when the run state itself carries no working node. let rowFocus: string | undefined; for (const row of rows) { if (row.state === "running" || row.state === "waiting") rowFocus = row.nodeId; } const focus = state.currentNode ?? state.waitingOn ?? rowFocus ?? failureNodeId ?? rows.at(-1)?.nodeId; const index = focus === undefined ? -1 : rows.findIndex((row) => row.nodeId === focus); return index < 0 ? 0 : Math.max(0, index - SESSION_NODE_LEAD); } /** * Byte- and item-bounded node rows that start at the requested row. Widget * scrolling asks for the row after the last one it holds, so a window always * begins where the caller asked and grows forward. */ function boundedNodeWindow( rows: readonly WorkflowSessionNodeRow[], start: number, ): { start: number; total: number; items: WorkflowSessionNodeRow[] } { const total = rows.length; if (total === 0) return { start: 0, total: 0, items: [] }; const first = Math.min(Math.max(0, Math.floor(start)), total - 1); const items: WorkflowSessionNodeRow[] = []; let windowStart = first; let bytes = 0; for (let index = first; index < total && items.length < VIEW_PAGE_ITEMS; index += 1) { const row = rows[index] as WorkflowSessionNodeRow; const rowBytes = Buffer.byteLength(canonicalJson(toJson(row)), "utf8") + 1; // A row must fit by itself, or the window would never move past it. A row // whose own identity exceeds the budget is left out, and the window starts at // the next row instead, so one long node id cannot break the frame the client // needs. A window that already holds rows stops here, so its rows stay // contiguous and the next cursor is exact. Complete node history stays in the // detailed run view. if (rowBytes > VIEW_PAGE_BYTES) { if (items.length === 0) { windowStart = index + 1; continue; } break; } if (items.length > 0 && bytes + rowBytes > VIEW_PAGE_BYTES) break; bytes += rowBytes; items.push(row); } return { start: windowStart, total, items }; } function toJson(value: unknown): JsonValue { return parseJson(canonicalJson(value)); } function isJsonObject(value: unknown): value is { [key: string]: JsonValue } { return typeof value === "object" && value !== null && !Array.isArray(value); } function isRunVersionRow(value: unknown): value is { revision: number; updatedAt: number; presentationRevision: number; runStatus: string; paused: number; } { return ( typeof value === "object" && value !== null && typeof (value as { revision?: unknown }).revision === "number" && typeof (value as { updatedAt?: unknown }).updatedAt === "number" && typeof (value as { presentationRevision?: unknown }).presentationRevision === "number" && typeof (value as { runStatus?: unknown }).runStatus === "string" && typeof (value as { paused?: unknown }).paused === "number" ); } function isSessionRevisionRow(value: unknown): value is { count: number; revisionSum: number; updatedAt: number; } { return ( typeof value === "object" && value !== null && typeof (value as { count?: unknown }).count === "number" && typeof (value as { revisionSum?: unknown }).revisionSum === "number" && typeof (value as { updatedAt?: unknown }).updatedAt === "number" ); } function isRevisionRow(value: unknown): value is { revision: number } { return ( typeof value === "object" && value !== null && typeof (value as { revision?: unknown }).revision === "number" ); }