import { writable, derived, get, type Writable, type Readable } from "svelte/store";
import {
type Lane,
type LanePayload,
type LaneEventRecord,
type ClientMsgV2,
type ServerMsgV2,
type SpeechReceipt,
type CellOutputRecord,
type CellExecutionPayload,
type ScratchpadPayload,
type InputPayload,
type LifecyclePayload,
type CapabilityHealthObservation,
type WorkItemDiagnostic,
type CandidateRevisionDiagnostic,
type ProvenanceMetadata,
type OriginKind,
resolveCapabilityHealth,
validateSpeechStatus,
parseServerMsgV2,
extractLaneRecord,
validateEventRecord,
normalizeHealthObservation,
normalizeWorkItem,
normalizeCandidate,
normalizeProvenance,
extractDiagnostics,
type OrientationDiagnostic,
normalizeOrientation,
extractLinkedDiagnostics,
type LinkedDiagnostic,
type DiagnosticItem,
separateThoughtsAndProse,
isFailurePayload,
isDiagnosticPayload,
} from "./protocol-v2";
export interface ScratchpadEntry {
id: string;
seq?: number;
created_ms: number;
step?: number;
thoughts: string;
prose: string;
active: boolean;
toolProposed?: {
id: string;
name: string;
arguments: string;
};
tokens?: number;
duration_ms?: number;
/** A scratchpad payload the daemon reported as a failure rather than reasoning. */
isFailure?: boolean;
}
export type { CellOutputRecord };
export interface CellExecutionItem {
id: string;
seq?: number;
created_ms: number;
command?: string;
/** `interrupted` is a real daemon outcome; unknown values are still rendered as words, never dropped. */
status: "ok" | "error" | "yielded" | "running" | "interrupted" | (string & {});
outputs: CellOutputRecord[];
truncated?: boolean;
exit_code?: number;
duration_ms?: number;
}
export interface SpeechEntry {
id: string;
seq?: number;
created_ms: number;
receipt: SpeechReceipt;
}
export interface InputEntry {
id: string;
seq?: number;
created_ms: number;
source: string;
text: string;
provenance?: ProvenanceMetadata;
}
export interface LifecycleEntry {
id: string;
seq?: number;
created_ms: number;
kind: "idle" | "streaming" | "tool" | "yielded" | "completed" | "failed" | "interrupted" | "reset" | string;
step?: number;
seconds?: number;
reason?: string;
details?: string;
provenance?: ProvenanceMetadata;
health?: CapabilityHealthObservation[];
work_items?: WorkItemDiagnostic[];
candidate?: CandidateRevisionDiagnostic | null;
orientation?: OrientationDiagnostic | null;
data?: unknown;
/** Set when this lifecycle record is a diagnostic interception rather than a real lifecycle event. */
isDiagnostic?: boolean;
/** Set when the record is a reported failure rather than a state transition. */
isFailure?: boolean;
}
export interface DiagnosticSnapshot {
seq: number;
health: CapabilityHealthObservation[];
work_items: WorkItemDiagnostic[];
candidate: CandidateRevisionDiagnostic | null;
orientation?: OrientationDiagnostic | null;
target?: LinkedDiagnostic;
context?: LinkedDiagnostic;
output?: LinkedDiagnostic;
posture?: LinkedDiagnostic;
adoption?: LinkedDiagnostic;
saved_ms: number;
}
export type TimelineItem =
| { lane: "scratchpad"; id: string; seq?: number; created_ms: number; data: ScratchpadEntry }
| { lane: "cell"; id: string; seq?: number; created_ms: number; data: CellExecutionItem }
| { lane: "speech"; id: string; seq?: number; created_ms: number; data: SpeechEntry }
| { lane: "input"; id: string; seq?: number; created_ms: number; data: InputEntry }
| { lane: "lifecycle"; id: string; seq?: number; created_ms: number; data: LifecycleEntry };
export interface ConnectionStatus {
state: "disconnected" | "connecting" | "connected" | "error";
sessionId: string;
lastSeenSeq: number;
isSyncing: boolean;
error?: string;
reconnectAttempts: number;
rejectedCount: number;
lastRejection?: { seq: number; reason: string };
/** Records behind the loaded window that the daemon says it can still serve. */
hasEarlier?: boolean;
/**
* The record the session's active context fold drained through. Everything at or below it is carried by
* the context summary rather than replayed, so a window that starts above it starts there *because* of
* the fold. It is a fact about the session, not about one replay: the view decides whether the window
* on screen is above it.
*/
foldSeq?: number;
/** True while a press of "load earlier history" is in flight. */
loadingEarlier?: boolean;
}
/**
* How many records a fresh console asks for. The daemon answers with the newest window of at most this
* size, anchored at its own last compaction, and says how far back its history goes. The console never
* asks for a whole session: a transcript is read from its live end.
*/
export const WINDOW_LIMIT = 2000;
/** One press of "load earlier history" asks for this many records behind the window. */
export const HISTORY_PAGE = 500;
export class StreamingClientV2 {
private ws: WebSocket | null = null;
private url: string;
public readonly serverId: string;
public readonly sessionId: string;
private lastSeenSeq: number = 0;
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
private reconnectAttempts = 0;
private maxReconnectAttempts = 20;
private baseBackoffMs = 500;
private maxBackoffMs = 10000;
private backoffFactor = 1.6;
private manualDisconnect = false;
public eventBuffer: Map<number, LaneEventRecord> = new Map();
private pendingResync = false;
// Svelte Stores
public readonly connectionStore: Writable<ConnectionStatus>;
public readonly scratchpadStore: Writable<ScratchpadEntry[]>;
public readonly cellStore: Writable<CellExecutionItem[]>;
public readonly speechStore: Writable<SpeechEntry[]>;
public readonly timelineStore: Writable<TimelineItem[]>;
public readonly healthStore: Writable<CapabilityHealthObservation[]>;
public readonly workItemsStore: Writable<WorkItemDiagnostic[]>;
public readonly candidateStore: Writable<CandidateRevisionDiagnostic | null>;
public readonly orientationStore: Writable<OrientationDiagnostic | null>;
public readonly targetStore: Writable<LinkedDiagnostic>;
public readonly contextStore: Writable<LinkedDiagnostic>;
public readonly outputStore: Writable<LinkedDiagnostic>;
public readonly postureStore: Writable<LinkedDiagnostic>;
public readonly adoptionStore: Writable<LinkedDiagnostic>;
public readonly diagnosticsStore: Writable<DiagnosticItem[]>;
/** Last operator command acknowledgement. The control surface must be able to report a refusal. */
public readonly operatorAckStore: Writable<{ action: string; success: boolean; message: string | null; at: number } | null>;
/** Latest `session_status` payload, opaque: the daemon owns its shape. */
public readonly sessionStatusStore: Writable<Record<string, unknown> | null>;
// Active transient entries for live streaming deltas
private activeScratchpadId: string | null = null;
private activeCellId: string | null = null;
private activeSpeechId: string | null = null;
private pendingRequests = new Map<string, { text: string; source: string; timestamp: number }>();
/**
* Every seq the timeline holds, in either direction. A replay, a reconnect and a press of "load
* earlier history" can each deliver a record this console already has, and this is what tells them
* apart without walking the transcript.
*/
private readonly seenSeqs = new Set<number>();
/** The oldest record the daemon can still serve, when it says so. Nothing said = it cannot page. */
private earliestSeq: number | null = null;
/** True between asking for a history page and its `sync_completed`. */
private backfilling = false;
/** The floor the last history page asked behind, so an empty answer can be told from a full one. */
private beforeBackfillFloor = 0;
/** The start of the newest range the daemon announced, until its first record lands. */
private windowStart: number | null = null;
constructor(
url: string = "ws://127.0.0.1:8765",
sessionId: string = "default",
serverId?: string
) {
this.url = url;
this.sessionId = sessionId;
this.serverId = serverId || this.deriveServerId(url);
this.connectionStore = writable<ConnectionStatus>({
state: "disconnected",
sessionId,
lastSeenSeq: 0,
isSyncing: false,
reconnectAttempts: 0,
rejectedCount: 0,
});
this.scratchpadStore = writable<ScratchpadEntry[]>([]);
this.cellStore = writable<CellExecutionItem[]>([]);
this.speechStore = writable<SpeechEntry[]>([]);
this.timelineStore = writable<TimelineItem[]>([]);
this.healthStore = writable<CapabilityHealthObservation[]>([]);
this.workItemsStore = writable<WorkItemDiagnostic[]>([]);
this.candidateStore = writable<CandidateRevisionDiagnostic | null>(null);
this.orientationStore = writable<OrientationDiagnostic | null>(null);
this.targetStore = writable<LinkedDiagnostic>(null);
this.contextStore = writable<LinkedDiagnostic>(null);
this.outputStore = writable<LinkedDiagnostic>(null);
this.postureStore = writable<LinkedDiagnostic>(null);
this.adoptionStore = writable<LinkedDiagnostic>(null);
this.diagnosticsStore = writable<DiagnosticItem[]>([]);
this.operatorAckStore = writable<{ action: string; success: boolean; message: string | null; at: number } | null>(null);
this.sessionStatusStore = writable<Record<string, unknown> | null>(null);
// Nothing is remembered between loads. The window is the daemon's to hand out, and a cursor
// remembered across a reload could only be ahead of the records it claims to cover.
}
private deriveServerId(url: string): string {
try {
const parsed = new URL(url);
return parsed.host || parsed.pathname || "default";
} catch {
return url.replace(/[^a-zA-Z0-9_-]/g, "_") || "default";
}
}
/**
* The daemon is the only holder of a session's history; this console keeps a window of it and nothing
* else. No cache, no remembered cursor, no second copy to keep in step: the window *is* the local
* state, and asking the daemon again is what a fresh page does.
*/
private windowFloor(): number {
// Until the first record of a range lands, the range's announced start is the floor: the daemon
// said what it is sending, and a console that waited for the records would grow the row that
// offers the rest of the history over a reader who is already reading the transcript.
return get(this.timelineStore)[0]?.seq ?? this.windowStart ?? 0;
}
/** What the console knows about the history behind its window, for the row that asks for more. */
private publishHistory(patch: Partial<ConnectionStatus> = {}) {
const floor = this.windowFloor();
const hasEarlier = this.earliestSeq !== null && floor > this.earliestSeq;
this.updateConnection({ hasEarlier, loadingEarlier: this.backfilling, ...patch });
}
/**
* Ask for the records behind the loaded window. This is the reader pressing a button, not a client
* catching up, so it moves neither the live cursor nor the floor: the answer is ingested as history
* and the loaded window grows downward.
*/
public loadEarlier(limit: number = HISTORY_PAGE): boolean {
if (this.backfilling) return false;
// The window is still arriving: a page behind a floor that is still moving would ask for records
// the console is about to receive anyway.
if (get(this.connectionStore).isSyncing) return false;
if (!this.ws || this.ws.readyState !== 1) return false;
const floor = this.windowFloor();
if (this.earliestSeq === null || floor <= this.earliestSeq) return false;
this.backfilling = true;
this.beforeBackfillFloor = floor;
this.updateConnection({ loadingEarlier: true });
this.send({ type: "history", session_id: this.sessionId, before_seq: floor, limit });
return true;
}
public connect(url?: string, sessionId?: string) {
if (url) this.url = url;
if (sessionId && sessionId !== this.sessionId) {
(this as any).sessionId = sessionId;
this.lastSeenSeq = 0;
this.earliestSeq = null;
this.seenSeqs.clear();
this.timelineStore.set([]);
this.eventBuffer.clear();
this.activeScratchpadId = null;
this.activeCellId = null;
}
this.manualDisconnect = false;
this.clearReconnectTimer();
if (this.ws) {
try {
this.ws.close();
} catch {
// ignore
}
this.ws = null;
}
this.updateConnection({ state: "connecting", error: undefined });
try {
this.ws = new WebSocket(this.url);
this.setupSocketHandlers();
} catch (err) {
this.updateConnection({
state: "error",
error: err instanceof Error ? err.message : String(err),
});
this.scheduleReconnect();
}
}
public disconnect() {
this.manualDisconnect = true;
this.clearReconnectTimer();
if (this.ws) {
try {
this.ws.close();
} catch {
// ignore
}
this.ws = null;
}
this.updateConnection({ state: "disconnected" });
}
private setupSocketHandlers() {
if (!this.ws) return;
this.ws.onopen = () => {
this.reconnectAttempts = 0;
this.updateConnection({
state: "connected",
reconnectAttempts: 0,
error: undefined,
});
// What this client wants depends on what it holds. Nothing loaded: the daemon's newest window, not
// its whole history — `after_seq: 0` with a `limit` is that request, and a daemon that does not know
// about windows ignores the limit and replays everything, which is what it already did. A window
// already on screen wants only the gap ahead of its live cursor, and says so without a limit.
const storesEmpty = get(this.timelineStore).length === 0;
this.send({
type: "subscribe",
session_id: this.sessionId,
after_seq: storesEmpty ? 0 : this.lastSeenSeq,
...(storesEmpty ? { limit: WINDOW_LIMIT } : {}),
});
};
this.ws.onmessage = (event: MessageEvent) => {
try {
const raw = typeof event.data === "string" ? event.data : "";
if (!raw) return;
const msg = parseServerMsgV2(raw);
this.handleServerMsg(msg);
} catch (err) {
console.warn("streaming-v2: failed to parse incoming message", err);
}
};
this.ws.onerror = () => {
this.updateConnection({ state: "error", error: "websocket error" });
};
this.ws.onclose = () => {
if (!this.manualDisconnect) {
this.updateConnection({ state: "disconnected" });
this.scheduleReconnect();
} else {
this.updateConnection({ state: "disconnected" });
}
};
}
private scheduleReconnect() {
if (this.manualDisconnect || this.reconnectTimer) return;
this.reconnectAttempts += 1;
const delay = Math.min(
this.baseBackoffMs * Math.pow(this.backoffFactor, this.reconnectAttempts - 1),
this.maxBackoffMs
);
const jitter = Math.random() * 200;
const finalDelay = Math.round(delay + jitter);
this.updateConnection({
reconnectAttempts: this.reconnectAttempts,
});
this.reconnectTimer = setTimeout(() => {
this.reconnectTimer = null;
if (!this.manualDisconnect) {
this.connect();
}
}, finalDelay);
}
private clearReconnectTimer() {
if (this.reconnectTimer) {
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
}
}
public send(msg: ClientMsgV2) {
if (this.ws && this.ws.readyState === 1) {
this.ws.send(JSON.stringify(msg));
}
}
public sendInput(text: string, source: string = "operator", clientRequestId?: string): string {
const trimmed = text.trim();
if (!trimmed) return "";
if (!this.ws || this.ws.readyState !== 1) {
throw new Error("Cannot send input: client is disconnected");
}
const reqId = clientRequestId || `req_${Date.now()}_${Math.random().toString(36).slice(2, 9)}`;
this.pendingRequests.set(reqId, {
text: trimmed,
source,
timestamp: Date.now(),
});
this.send({
type: "send_input",
session_id: this.sessionId,
source,
text: trimmed,
client_request_id: reqId,
});
return reqId;
}
public interrupt(executionId?: string | null) {
this.send({
type: "interrupt",
session_id: this.sessionId,
execution_id: executionId ?? null,
});
}
public resetKernel() {
this.send({
type: "reset_kernel",
session_id: this.sessionId,
});
}
private restoreLinkedSnapshot(snapshot: Partial<DiagnosticSnapshot>) {
if (snapshot.target !== undefined) this.targetStore.set(snapshot.target);
if (snapshot.context !== undefined) this.contextStore.set(snapshot.context);
if (snapshot.output !== undefined) this.outputStore.set(snapshot.output);
if (snapshot.posture !== undefined) this.postureStore.set(snapshot.posture);
if (snapshot.adoption !== undefined) this.adoptionStore.set(snapshot.adoption);
}
public setHealth(health: CapabilityHealthObservation[]) {
this.healthStore.set(health.map(normalizeHealthObservation));
}
public setWorkItems(items: WorkItemDiagnostic[]) {
this.workItemsStore.set(items.map(normalizeWorkItem));
}
public setCandidate(candidate: CandidateRevisionDiagnostic | null) {
this.candidateStore.set(candidate ? normalizeCandidate(candidate) : null);
}
public setOrientation(orientation: OrientationDiagnostic | null) {
this.orientationStore.set(orientation ? normalizeOrientation(orientation) : null);
}
public clearSessionState() {
this.lastSeenSeq = 0;
this.earliestSeq = null;
this.backfilling = false;
this.windowStart = null;
this.updateConnection({ foldSeq: undefined, hasEarlier: undefined, loadingEarlier: undefined });
this.seenSeqs.clear();
this.eventBuffer.clear();
this.activeScratchpadId = null;
this.activeCellId = null;
this.activeSpeechId = null;
this.pendingRequests.clear();
this.scratchpadStore.set([]);
this.cellStore.set([]);
this.speechStore.set([]);
this.timelineStore.set([]);
this.healthStore.set([]);
this.workItemsStore.set([]);
this.candidateStore.set(null);
this.orientationStore.set(null);
this.targetStore.set(null);
this.contextStore.set(null);
this.outputStore.set(null);
this.postureStore.set(null);
this.adoptionStore.set(null);
this.diagnosticsStore.set([]);
this.updateConnection({ lastSeenSeq: 0, isSyncing: false, rejectedCount: 0, lastRejection: undefined });
}
public handleServerMsg(msg: ServerMsgV2) {
switch (msg.type) {
case "sync_started": {
// The range and the floor it sits above arrive together, so the row that offers the records
// behind the window is in the first frame of the replay rather than appearing over a reader who
// is already reading it.
if (msg.earliest_seq !== undefined && msg.earliest_seq !== null) {
this.earliestSeq = msg.earliest_seq;
}
// A session fact, sent with every range: the window on screen may or may not be above it.
if (msg.fold_seq !== undefined && msg.fold_seq !== null) {
this.updateConnection({ foldSeq: msg.fold_seq });
}
this.windowStart = msg.from_seq;
this.updateConnection({ isSyncing: true });
this.publishHistory();
break;
}
case "sync_completed": {
// Drain any buffered events up to latest watermark
this.drainBuffer();
const wasBackfilling = this.backfilling;
this.backfilling = false;
// The daemon is the one that knows how far back its history goes. Silence here means it cannot
// serve anything behind the window, and a console that guessed otherwise would offer a button
// that does nothing.
if (msg.earliest_seq !== undefined && msg.earliest_seq !== null) {
this.earliestSeq = msg.earliest_seq;
}
// A window of its own is not an advance: a page behind it must never move the live cursor.
if (!wasBackfilling && msg.latest_seq > this.lastSeenSeq) {
this.setLastSeenSeq(msg.latest_seq);
}
// Sequence numbers are sparse, so a floor below the window is a bound on where history could
// be, not a promise that any of it exists. A page that brought back nothing is the answer: the
// button offered a press that would return the same nothing every time.
if (wasBackfilling && this.earliestSeq !== null && get(this.timelineStore)[0]?.seq === this.beforeBackfillFloor) {
this.earliestSeq = get(this.timelineStore)[0]?.seq ?? 0;
}
this.updateConnection({ isSyncing: false });
this.publishHistory();
break;
}
case "event": {
const record = extractLaneRecord(msg);
try {
validateEventRecord(record);
} catch (err) {
const reason =
err instanceof Error
? err.message
: typeof (err as any)?.message === "string"
? (err as any).message
: String(err);
this.connectionStore.update((s) => ({
...s,
rejectedCount: s.rejectedCount + 1,
lastRejection: { seq: record.seq, reason },
}));
throw err;
}
this.processSequencedRecord(record);
break;
}
case "live_chunk": {
this.handleLiveChunk(msg);
break;
}
case "ack": {
if (msg.client_request_id) {
this.pendingRequests.delete(msg.client_request_id);
}
break;
}
case "operator_ack": {
this.operatorAckStore.set({
action: msg.action,
success: msg.success,
message: msg.message ?? null,
at: Date.now(),
});
if (!msg.success) {
this.updateConnection({
error: `${msg.action}: ${msg.message ?? "operator command failed"}`,
});
}
break;
}
case "session_status": {
this.sessionStatusStore.set(
msg.state && typeof msg.state === "object" ? (msg.state as Record<string, unknown>) : null
);
break;
}
case "pong":
case "session_list":
break;
case "error": {
console.error("streaming-v2 server error:", msg.code, msg.message);
this.updateConnection({ error: `${msg.code}: ${msg.message}` });
break;
}
}
}
// Idempotent sequence cursor processor supporting sparse monotonic sequences
private processSequencedRecord(record: LaneEventRecord) {
const seq = record.seq;
// Deduplication: drop only if the timeline already represents this sequence. Membership is kept as
// a set because a replay, a reconnect and a history page all re-deliver records, and a scan of the
// transcript per record is the cost this console has spent the most time removing.
if (this.seenSeqs.has(seq)) {
this.acknowledge();
return;
}
// A record behind the loaded window is history: the reader asked for it, and it can never be an
// advance. It goes into the timeline where its sequence puts it (see `updateTimelineItem`) and the
// live cursor stays where it was.
const floor = get(this.timelineStore)[0]?.seq;
const history = floor !== undefined && seq < floor;
this.applyEvent(record, history ? "history" : "live");
this.seenSeqs.add(seq);
if (!history && seq > this.lastSeenSeq) {
this.setLastSeenSeq(seq);
}
this.drainBuffer();
if (history) this.publishHistory();
this.acknowledge();
}
private acknowledge() {
this.send({
type: "ack",
session_id: this.sessionId,
seq: this.lastSeenSeq,
});
}
private drainBuffer() {
if (this.eventBuffer.size === 0) return;
const sortedSeqs = Array.from(this.eventBuffer.keys()).sort((a, b) => a - b);
for (const seq of sortedSeqs) {
if (seq > this.lastSeenSeq) {
const record = this.eventBuffer.get(seq)!;
this.eventBuffer.delete(seq);
this.applyEvent(record);
this.setLastSeenSeq(seq);
}
}
}
private setLastSeenSeq(seq: number) {
this.lastSeenSeq = seq;
this.updateConnection({ lastSeenSeq: seq });
}
private finishActiveScratchpad() {
const activeId = this.activeScratchpadId;
if (!activeId) return;
this.activeScratchpadId = null;
const active = get(this.scratchpadStore).find((entry) => entry.id === activeId);
if (!active) return;
const finalized: ScratchpadEntry = { ...active, active: false };
this.scratchpadStore.update((entries) =>
entries.map((entry) => entry.id === activeId ? finalized : entry)
);
this.updateTimelineItem({
lane: "scratchpad",
id: activeId,
created_ms: finalized.created_ms,
data: finalized,
});
}
private discardActiveScratchpad() {
const activeId = this.activeScratchpadId;
if (!activeId) return;
this.activeScratchpadId = null;
this.scratchpadStore.update((entries) => entries.filter((entry) => entry.id !== activeId));
this.timelineStore.update((items) => items.filter((item) => item.id !== activeId));
}
// Handle transient live chunks before discrete event commit
private handleLiveChunk(chunk: {
lane: Lane;
execution_id?: string | null;
stream?: string | null;
delta: string;
}) {
if (chunk.lane === "scratchpad") {
const isThought = chunk.stream === "thought" || chunk.stream === "think";
this.scratchpadStore.update((entries) => {
let entry = entries.find((e) => e.id === this.activeScratchpadId && e.active);
if (!entry) {
const newId = `scratchpad_${Date.now()}_${Math.random().toString(36).slice(2, 7)}`;
this.activeScratchpadId = newId;
entry = {
id: newId,
created_ms: Date.now(),
thoughts: "",
prose: "",
active: true,
};
entries = [...entries, entry];
}
if (isThought) {
entry.thoughts += chunk.delta;
} else {
entry.prose += chunk.delta;
}
// Sync with timeline
this.updateTimelineItem({
lane: "scratchpad",
id: entry.id,
created_ms: entry.created_ms,
data: { ...entry },
});
return entries;
});
} else if (chunk.lane === "cell") {
const execId = chunk.execution_id || this.activeCellId || `cell_${Date.now()}`;
this.activeCellId = execId;
const streamKind = (chunk.stream as "stdout" | "stderr" | "result") || "stdout";
this.cellStore.update((entries) => {
let entry = entries.find((e) => e.id === execId);
if (!entry) {
entry = {
id: execId,
created_ms: Date.now(),
status: "running",
outputs: [],
};
entries = [...entries, entry];
}
const lastOutput = entry.outputs[entry.outputs.length - 1];
if (lastOutput && lastOutput.stream === streamKind) {
lastOutput.text += chunk.delta;
} else {
entry.outputs.push({
stream: streamKind,
text: chunk.delta,
timestamp: Date.now(),
});
// Bound memory for massive bursts
if (entry.outputs.length > 1000) {
entry.outputs = entry.outputs.slice(-1000);
entry.truncated = true;
}
}
this.updateTimelineItem({
lane: "cell",
id: entry.id,
created_ms: entry.created_ms,
data: { ...entry, outputs: [...entry.outputs] },
});
return entries;
});
} else if (chunk.lane === "speech") {
this.speechStore.update((entries) => {
let entry = entries.find((e) => e.id === this.activeSpeechId);
if (!entry) {
const newId = `speech_live_${Date.now()}_${Math.random().toString(36).slice(2, 7)}`;
this.activeSpeechId = newId;
entry = {
id: newId,
created_ms: Date.now(),
receipt: {
effect_id: newId,
status: "held",
text: "",
},
};
entries = [...entries, entry];
}
entry.receipt.text += chunk.delta;
this.updateTimelineItem({
lane: "speech",
id: entry.id,
created_ms: entry.created_ms,
data: { ...entry, receipt: { ...entry.receipt } },
});
return entries;
});
}
}
// Ingest sequenced events into individual stores & unified timeline
/**
* `mode` is the difference between a record the session just produced and a record the reader asked
* to see from behind the window. Only the live stream has an entry a delta is currently streaming
* into, so only a live record may be merged into one: history arriving while a cell streams must
* never be mistaken for the cell that is streaming.
*/
private applyEvent(record: LaneEventRecord, mode: "live" | "history" = "live") {
const { seq, lane, payload, created_ms } = record;
const merging = mode === "live";
switch (lane) {
case "scratchpad": {
const sp = payload as ScratchpadPayload;
const rawText = (sp.text || (sp as any).content || (sp as any).thoughts || "").trim();
// 1. Intercept failure JSON on scratchpad lane and route to diagnostics/input correlation
let failureCheck = isFailurePayload(sp);
if (!failureCheck.isFailure && rawText) {
failureCheck = isFailurePayload(rawText);
}
if (failureCheck.isFailure) {
const diagItem: DiagnosticItem = {
id: `diag_fail_${seq}`,
seq,
created_ms: created_ms || Date.now(),
kind: "failure",
title: "Turn / Tool Failure",
summary: failureCheck.reason,
data: sp,
};
this.diagnosticsStore.update((items) => [...items, diagItem]);
this.updateTimelineItem({
lane: "lifecycle",
id: diagItem.id,
seq,
created_ms: diagItem.created_ms,
data: {
id: diagItem.id,
created_ms: diagItem.created_ms,
kind: "failed",
reason: failureCheck.reason,
details: failureCheck.reason,
data: sp,
isFailure: true,
},
});
break;
}
// 2. Intercept context receipt, model usage, or diagnostic payloads on scratchpad lane
let diagCheck = isDiagnosticPayload(sp);
if (!diagCheck.isDiagnostic && rawText) {
diagCheck = isDiagnosticPayload(rawText);
}
if (diagCheck.isDiagnostic && diagCheck.kind) {
const diagItem: DiagnosticItem = {
id: `diag_${diagCheck.kind}_${seq}`,
seq,
created_ms: created_ms || Date.now(),
kind: diagCheck.kind,
title: diagCheck.kind,
data: diagCheck.data || sp,
};
this.diagnosticsStore.update((items) => [...items, diagItem]);
this.updateTimelineItem({
lane: "lifecycle",
id: diagItem.id,
seq,
created_ms: diagItem.created_ms,
data: {
id: diagItem.id,
created_ms: diagItem.created_ms,
kind: diagCheck.kind,
data: diagCheck.data || sp,
isDiagnostic: true,
},
});
break;
}
// 3. Orientation / frontier diagnostic interception
if (typeof rawText === "string" && rawText.startsWith("{") && (rawText.includes("orientation.") || rawText.includes("frontier."))) {
try {
const parsed = JSON.parse(rawText);
const orient = normalizeOrientation(parsed);
if (orient) {
this.orientationStore.set(orient);
break;
}
} catch {}
}
// 4. Genuine scratchpad entry: keep provider reasoning and scratchpad prose distinct
let { thoughts, prose } = separateThoughtsAndProse(sp);
if (sp.delta) {
if (sp.kind === "thought") thoughts += sp.delta;
else prose += sp.delta;
}
const entryId = `sp_${seq}`;
this.discardActiveScratchpad();
this.scratchpadStore.update((entries) => {
const entry: ScratchpadEntry = {
id: entryId,
seq,
created_ms: created_ms || Date.now(),
thoughts,
prose,
active: sp.active ?? false,
step: sp.step,
toolProposed: sp.tool_call,
tokens: sp.tokens,
duration_ms: sp.duration_ms,
};
this.updateTimelineItem({
lane: "scratchpad",
id: entry.id,
seq,
created_ms: entry.created_ms,
data: entry,
});
return [...entries, entry];
});
break;
}
case "cell": {
const cp = payload as any;
const execId = cp.execution_id || `cell_${seq}`;
let status: "ok" | "error" | "yielded" | "running" = "ok";
if (cp.type === "started") {
status = "running";
} else if (cp.state) {
status = cp.state === "interrupted" ? "error" : cp.state;
} else if (cp.status) {
status = cp.status;
}
const outputs: CellOutputRecord[] = [];
if (Array.isArray(cp.output)) {
for (const o of cp.output) {
outputs.push({ stream: o.stream, text: o.text, timestamp: created_ms });
}
}
if (Array.isArray(cp.outputs)) {
outputs.push(...cp.outputs);
}
if (cp.stream && cp.text) {
outputs.push({
stream: cp.stream,
text: cp.text,
timestamp: created_ms,
});
}
const truncated = Boolean(cp.truncated || (cp.truncated_bytes && cp.truncated_bytes > 0));
const command = cp.code || cp.command;
const activeCell = merging ? this.activeCellId : null;
this.cellStore.update((entries) => {
const idx = entries.findIndex((e) => e.id === execId || e.id === activeCell);
let entry: CellExecutionItem;
if (idx >= 0) {
const prev = entries[idx];
const prevId = prev.id;
let combinedOutputs = outputs.length > 0 ? [...prev.outputs, ...outputs] : prev.outputs;
let isTruncated = truncated || (prev.truncated ?? false);
if (combinedOutputs.length > 1000) {
combinedOutputs = combinedOutputs.slice(-1000);
isTruncated = true;
}
entry = {
...prev,
id: execId,
seq,
created_ms: created_ms || prev.created_ms,
command: command || prev.command,
status,
outputs: combinedOutputs,
truncated: isTruncated,
exit_code: cp.exit_code ?? prev.exit_code,
duration_ms: cp.duration_ms ?? prev.duration_ms,
};
entries[idx] = entry;
if (this.activeCellId === execId || this.activeCellId === prevId) {
this.activeCellId = null;
}
this.updateTimelineItem({
lane: "cell",
id: entry.id,
seq,
created_ms: entry.created_ms,
data: entry,
}, prevId);
return entries;
} else {
let initialOutputs = outputs;
let isTruncated = truncated;
if (initialOutputs.length > 1000) {
initialOutputs = initialOutputs.slice(-1000);
isTruncated = true;
}
entry = {
id: execId,
seq,
created_ms: created_ms || Date.now(),
command,
status,
outputs: initialOutputs,
truncated: isTruncated,
exit_code: cp.exit_code,
duration_ms: cp.duration_ms,
};
entries.push(entry);
}
if (this.activeCellId === execId) {
this.activeCellId = null;
}
this.updateTimelineItem({
lane: "cell",
id: entry.id,
seq,
created_ms: entry.created_ms,
data: entry,
});
return entries;
});
break;
}
case "speech": {
const raw = payload as any;
const rawStatus = raw.status;
const rawType = raw.type;
// Strictly validate status: ONLY explicit 'confirmed' becomes confirmed speech.
// 'held', 'failed', or unknown statuses must NEVER default to 'confirmed'.
let validatedStatus: "confirmed" | "held" | "failed";
let reason: string | undefined = raw.reason;
if (rawStatus === "confirmed" && (rawType === "sent" || rawType === undefined)) {
validatedStatus = "confirmed";
} else if (rawStatus === "failed" || rawType === "failed") {
validatedStatus = "failed";
reason = reason || raw.error || "delivery failed";
} else if (rawStatus === "held" || rawType === "held") {
validatedStatus = "held";
reason = reason || "held by supervisor policy";
} else {
// Unknown or missing status: NEVER default to confirmed! Treat visibly as held/error.
validatedStatus = "held";
const statusStr = typeof rawStatus === "string" ? rawStatus : "unvalidated";
reason = reason || `unvalidated delivery status: ${statusStr}`;
}
const receipt: SpeechReceipt = {
effect_id: raw.effect_id || `eff_${seq}`,
status: validatedStatus,
event_id: raw.event_id ?? null,
reason,
text: raw.text || "",
author: raw.author,
provenance: raw.provenance,
};
const entryId = `speech_${seq}_${receipt.effect_id || Math.random().toString(36).slice(2, 6)}`;
const speechEntry: SpeechEntry = {
id: entryId,
seq,
created_ms: created_ms || Date.now(),
receipt,
};
const activeSpeech = merging ? this.activeSpeechId : null;
this.speechStore.update((entries) => {
if (activeSpeech) {
const activeIdx = entries.findIndex((e) => e.id === activeSpeech);
if (activeIdx >= 0) {
const prev = entries[activeIdx];
const prevId = prev.id;
const updated: SpeechEntry = {
...prev,
id: entryId,
seq,
created_ms: created_ms || prev.created_ms,
receipt: {
...receipt,
text: receipt.text || prev.receipt.text,
},
};
entries[activeIdx] = updated;
this.activeSpeechId = null;
this.updateTimelineItem({
lane: "speech",
id: updated.id,
seq,
created_ms: updated.created_ms,
data: updated,
}, prevId);
return entries;
}
}
const newEntries = [...entries, speechEntry];
newEntries.sort((a, b) => {
if (a.seq !== undefined && b.seq !== undefined) return a.seq - b.seq;
return a.created_ms - b.created_ms;
});
this.updateTimelineItem({
lane: "speech",
id: speechEntry.id,
seq,
created_ms: speechEntry.created_ms,
data: speechEntry,
});
return newEntries;
});
break;
}
case "input": {
const ip = payload as InputPayload;
const entryId = `input_${seq}`;
const inputEntry: InputEntry = {
id: entryId,
seq,
created_ms: created_ms || Date.now(),
source: ip.source || "operator",
text: ip.text || "",
provenance: ip.provenance,
};
this.updateTimelineItem({
lane: "input",
id: inputEntry.id,
seq,
created_ms: inputEntry.created_ms,
data: inputEntry,
});
break;
}
case "lifecycle": {
const lp = payload as LifecyclePayload;
if (lp.kind === "turn.outcome" || lp.kind === "turn.failed") {
this.finishActiveScratchpad();
}
const entryId = `life_${seq}`;
const extracted = extractDiagnostics(lp);
const linked = extractLinkedDiagnostics(lp.data);
if (extracted.health !== undefined) {
this.healthStore.set(extracted.health);
}
if (extracted.work_items !== undefined) {
this.workItemsStore.set(extracted.work_items);
}
if (extracted.candidate !== undefined) {
this.candidateStore.set(extracted.candidate);
}
if (extracted.orientation !== undefined) {
this.orientationStore.set(extracted.orientation);
}
// Diagnostics live in memory for the life of the page: they are re-derived from the window on
// load, so there is nothing about them worth writing down.
if (linked.target !== undefined) this.targetStore.set(linked.target);
if (linked.context !== undefined) this.contextStore.set(linked.context);
if (linked.output !== undefined) this.outputStore.set(linked.output);
if (linked.posture !== undefined) this.postureStore.set(linked.posture);
if (linked.adoption !== undefined) this.adoptionStore.set(linked.adoption);
const diagItem: DiagnosticItem = {
id: entryId,
seq,
created_ms: created_ms || Date.now(),
kind: lp.kind,
title: lp.kind,
summary: lp.reason || lp.details,
data: lp.data || lp,
};
this.diagnosticsStore.update((items) => [...items, diagItem]);
const lifeEntry: LifecycleEntry = {
id: entryId,
seq,
created_ms: created_ms || Date.now(),
kind: lp.kind,
step: lp.step,
seconds: lp.seconds,
reason: lp.reason,
details: lp.details,
provenance: lp.provenance ? normalizeProvenance(lp.provenance) : undefined,
health: extracted.health,
work_items: extracted.work_items,
candidate: extracted.candidate,
orientation: extracted.orientation,
data: lp.data,
};
this.updateTimelineItem({
lane: "lifecycle",
id: lifeEntry.id,
seq,
created_ms: lifeEntry.created_ms,
data: lifeEntry,
});
break;
}
}
}
private updateTimelineItem(item: TimelineItem, prevId?: string) {
let next: TimelineItem[] = [];
this.timelineStore.update((timeline) => {
const idx = timeline.findIndex((t) => t.id === item.id || (prevId !== undefined && t.id === prevId));
if (idx >= 0) {
next = [...timeline];
next[idx] = item;
} else {
// A record with no place yet goes where its sequence says, so a history page prepends itself
// and a live append lands at the end without either path having to know which it is.
next = [...timeline, item];
next.sort((a, b) => {
if (a.seq !== undefined && b.seq !== undefined) {
return a.seq - b.seq;
}
return a.created_ms - b.created_ms;
});
}
return next;
});
}
private updateConnection(patch: Partial<ConnectionStatus>) {
this.connectionStore.update((prev) => ({ ...prev, ...patch }));
}
}
// Singleton or session-keyed factory
const instances = new Map<string, StreamingClientV2>();
export function getStreamingClient(
url: string = "ws://127.0.0.1:8765",
sessionId: string = "default",
serverId?: string
): StreamingClientV2 {
const key = `${url}::${sessionId}`;
if (!instances.has(key)) {
instances.set(key, new StreamingClientV2(url, sessionId, serverId));
}
return instances.get(key)!;
}