1import { z } from "zod";2import { sessionUsageSnapshotSchema } from "../../../../src/session/sessionUsageSchema";3import { parseStructuredToolInput } from "../../../../src/shared/structuredInput";4import {5 SERVER_ERROR_CODES,6 SERVER_ERROR_SOURCES,7 type
ServerErrorCode,
8 type ServerErrorSource,
9} from "../../../../src/types";
10import type { SessionEvent } from "../lib/wsProtocol";
11import { safeParseSessionEvent } from "../lib/wsProtocol";
12import {
13 clearModelStreamReplayRuntime,
14 createModelStreamReplayRuntime,
15 type ModelStreamChunkEvent,
16 type ModelStreamRawEvent,
17 type ModelStreamReplayRuntime,
18 type ModelStreamUpdate,
19 mapModelStreamChunk,
20 replayModelStreamRawEvent,
21 shouldIgnoreNormalizedChunkForRawBackedTurn,
22} from "./modelStream";
23import type { FeedItem, ThreadAgentSummary, ThreadRuntime, TranscriptEvent } from "./types";
24
25export type ThreadModelStreamRuntime = {
26 activeTurnId: string | null;
27 assistantItemIdByStream: Map<string, string>;
28 assistantTextByStream: Map<string, string>;
29 assistantTextHistoryInTurn: string[];
30 lastAssistantStreamKeyByTurn: Map<string, string>;
31 completedAssistantStreamKeys: Set<string>;
32 reasoningItemIdByStream: Map<string, string>;
33 reasoningTextByStream: Map<string, string>;
34 reasoningTextsSeenInTurn: Set<string>;
35 reasoningTextHistoryInTurn: string[];
36 reasoningTurns: Set<string>;
37 toolItemIdByKey: Map<string, string>;
38 latestToolKeyByTurnAndName: Map<string, string>;
39 toolInputByKey: Map<string, string>;
40 lastAssistantTurnId: string | null;
41 lastReasoningTurnId: string | null;
42 replay: ModelStreamReplayRuntime;
43};
44
45export type ThreadModelStreamFeedOps = {
46 makeId: () => string;
47 nowIso: () => string;
48 pushFeedItem: (item: FeedItem) => void;
49 insertFeedItemBefore?: (beforeItemId: string, item: FeedItem) => void;
50 updateFeedItem: (itemId: string, update: (item: FeedItem) => FeedItem) => void;
51 flushPendingContent?: () => void;
52 onToolTerminal?: () => void;
53};
54
55export type TranscriptUsageState = Pick<ThreadRuntime, "sessionUsage" | "lastTurnUsage">;
56
57type DeveloperDiagnosticSessionEvent = Extract<
58 SessionEvent,
59 {
60 type: "observability_status" | "session_backup_state" | "harness_context";
61 }
62>;
63
64const serverErrorCodeSet = new Set<string>(SERVER_ERROR_CODES);
65const serverErrorSourceSet = new Set<string>(SERVER_ERROR_SOURCES);
66
67function normalizeServerErrorCode(value: unknown): ServerErrorCode {
68 return typeof value === "string" && serverErrorCodeSet.has(value)
69 ? (value as ServerErrorCode)
70 : "internal_error";
71}
72
73function normalizeServerErrorSource(value: unknown): ServerErrorSource {
74 return typeof value === "string" && serverErrorSourceSet.has(value)
75 ? (value as ServerErrorSource)
76 : "session";
77}
78
79function yesNo(value: boolean): string {
80 return value ? "yes" : "no";
81}
82
83function humanizeUnderscoreLabel(value: string): string {
84 return value.replace(/_/g, " ");
85}
86
87function formatObservabilityDiagnosticLine(evt: {
88 enabled: boolean;
89 health: { status?: unknown; reason?: unknown; message?: unknown };
90 config?: unknown;
91}): string {
92 const configured =
93 isRecord(evt.config) && typeof evt.config.configured === "boolean"
94 ? evt.config.configured
95 : false;
96 const healthStatus = typeof evt.health.status === "string" ? evt.health.status : "unknown";
97 const healthReason = typeof evt.health.reason === "string" ? evt.health.reason : "unknown";
98 const healthMessage = previewValue(evt.health.message);
99 const healthDetail = healthMessage ? `${healthReason}: ${healthMessage}` : healthReason;
100 return `Observability: enabled=${yesNo(evt.enabled)}, configured=${yesNo(configured)}, health=${healthStatus} (${healthDetail})`;
101}
102
103function formatSessionBackupDiagnosticLine(evt: { reason?: unknown; backup?: unknown }): string {
104 const reason =
105 typeof evt.reason === "string" && evt.reason.trim().length > 0
106 ? humanizeUnderscoreLabel(evt.reason)
107 : "update";
108 const status =
109 isRecord(evt.backup) && typeof evt.backup.status === "string" ? evt.backup.status : "unknown";
110 const checkpointCount =
111 isRecord(evt.backup) && Array.isArray(evt.backup.checkpoints)
112 ? evt.backup.checkpoints.length
113 : null;
114 return checkpointCount === null
115 ? `Session backup (${reason}): status=${status}`
116 : `Session backup (${reason}): status=${status}, checkpoints=${checkpointCount}`;
117}
118
119function formatHarnessContextDiagnosticLine(evt: { context?: unknown }): string {
120 if (evt.context === null || evt.context === undefined) {
121 return "Harness context cleared";
122 }
123 if (!isRecord(evt.context)) {
124 return "Harness context updated";
125 }
126
127 const details: string[] = [];
128 if (typeof evt.context.taskId === "string" && evt.context.taskId.trim().length > 0) {
129 details.push(`taskId=${evt.context.taskId}`);
130 }
131 if (typeof evt.context.runId === "string" && evt.context.runId.trim().length > 0) {
132 details.push(`runId=${evt.context.runId}`);
133 }
134 if (typeof evt.context.objective === "string" && evt.context.objective.trim().length > 0) {
135 details.push(`objective=${previewValue(evt.context.objective, 80)}`);
136 }
137 if (Array.isArray(evt.context.acceptanceCriteria)) {
138 details.push(`acceptanceCriteria=${evt.context.acceptanceCriteria.length}`);
139 }
140 if (Array.isArray(evt.context.constraints)) {
141 details.push(`constraints=${evt.context.constraints.length}`);
142 }
143 return details.length > 0
144 ? `Harness context updated: ${details.join(", ")}`
145 : "Harness context updated";
146}
147
148export function developerDiagnosticSystemLineFromSessionEvent(
149 evt: DeveloperDiagnosticSessionEvent,
150): string {
151 switch (evt.type) {
152 case "observability_status":
153 return formatObservabilityDiagnosticLine(evt);
154 case "session_backup_state":
155 return formatSessionBackupDiagnosticLine(evt);
156 case "harness_context":
157 return formatHarnessContextDiagnosticLine(evt);
158 }
159}
160
161function developerDiagnosticSystemLineFromPayload(payload: unknown): string | null {
162 const parsed = transcriptDeveloperDiagnosticPayloadSchema.safeParse(payload);
163 if (!parsed.success) return null;
164
165 switch (parsed.data.type) {
166 case "observability_status":
167 return formatObservabilityDiagnosticLine(parsed.data);
168 case "session_backup_state":
169 return formatSessionBackupDiagnosticLine(parsed.data);
170 case "harness_context":
171 return formatHarnessContextDiagnosticLine(parsed.data);
172 }
173}
174
175export function unhandledEventSystemLine(type: string): string {
176 return `Unhandled event: ${type}`;
177}
178
179export function createThreadModelStreamRuntime(): ThreadModelStreamRuntime {
180 return {
181 activeTurnId: null,
182 assistantItemIdByStream: new Map(),
183 assistantTextByStream: new Map(),
184 assistantTextHistoryInTurn: [],
185 lastAssistantStreamKeyByTurn: new Map(),
186 completedAssistantStreamKeys: new Set(),
187 reasoningItemIdByStream: new Map(),
188 reasoningTextByStream: new Map(),
189 reasoningTextsSeenInTurn: new Set(),
190 reasoningTextHistoryInTurn: [],
191 reasoningTurns: new Set(),
192 toolItemIdByKey: new Map(),
193 latestToolKeyByTurnAndName: new Map(),
194 toolInputByKey: new Map(),
195 lastAssistantTurnId: null,
196 lastReasoningTurnId: null,
197 replay: createModelStreamReplayRuntime(),
198 };
199}
200
201export function clearThreadModelStreamRuntime(runtime: ThreadModelStreamRuntime) {
202 runtime.activeTurnId = null;
203 clearStepLocalModelStreamRuntime(runtime, { snapshotReasoning: false, snapshotAssistant: false });
204 runtime.completedAssistantStreamKeys.clear();
205 runtime.assistantTextHistoryInTurn = [];
206 runtime.reasoningTextsSeenInTurn.clear();
207 runtime.reasoningTextHistoryInTurn = [];
208 runtime.reasoningTurns.clear();
209 runtime.toolItemIdByKey.clear();
210 runtime.latestToolKeyByTurnAndName.clear();
211 runtime.toolInputByKey.clear();
212 runtime.lastReasoningTurnId = null;
213 clearModelStreamReplayRuntime(runtime.replay);
214}
215
216function normalizeReasoningText(text: string): string | null {
217 const normalized = text.trim();
218 return normalized.length > 0 ? normalized : null;
219}
220
221function rememberStreamedReasoningText(runtime: ThreadModelStreamRuntime, text: string) {
222 const normalized = normalizeReasoningText(text);
223 if (!normalized) return;
224 runtime.reasoningTextsSeenInTurn.add(normalized);
225 runtime.reasoningTextHistoryInTurn.push(normalized);
226}
227
228function rememberStreamedAssistantText(runtime: ThreadModelStreamRuntime, text: string) {
229 if (!hasVisibleAssistantText(text)) return;
230 runtime.assistantTextHistoryInTurn.push(text);
231}
232
233function rememberStreamedReasoningTexts(runtime: ThreadModelStreamRuntime) {
234 for (const text of runtime.reasoningTextByStream.values()) {
235 rememberStreamedReasoningText(runtime, text);
236 }
237}
238
239function rememberStreamedAssistantTexts(runtime: ThreadModelStreamRuntime) {
240 for (const text of runtime.assistantTextByStream.values()) {
241 rememberStreamedAssistantText(runtime, text);
242 }
243}
244
245function clearStepLocalModelStreamRuntime(
246 runtime: ThreadModelStreamRuntime,
247 opts: { snapshotReasoning?: boolean; snapshotAssistant?: boolean } = {},
248) {
249 if (opts.snapshotAssistant !== false) {
250 rememberStreamedAssistantTexts(runtime);
251 }
252 if (opts.snapshotReasoning !== false) {
253 rememberStreamedReasoningTexts(runtime);
254 }
255 runtime.assistantItemIdByStream.clear();
256 runtime.assistantTextByStream.clear();
257 runtime.lastAssistantStreamKeyByTurn.clear();
258 runtime.reasoningItemIdByStream.clear();
259 runtime.reasoningTextByStream.clear();
260 runtime.lastAssistantTurnId = null;
261 runtime.lastReasoningTurnId = null;
262}
263
264export function hasMatchingStreamedReasoningText(
265 runtime: ThreadModelStreamRuntime,
266 text: string,
267): boolean {
268 const normalized = normalizeReasoningText(text);
269 if (!normalized) return false;
270 if (runtime.reasoningTextsSeenInTurn.has(normalized)) return true;
271
272 for (const current of runtime.reasoningTextByStream.values()) {
273 if (normalizeReasoningText(current) === normalized) {
274 return true;
275 }
276 }
277
278 const aggregate = normalizeTranscriptReplayText(
279 [
280 ...runtime.reasoningTextHistoryInTurn,
281 ...[...runtime.reasoningTextByStream.values()]
282 .map((current) => normalizeReasoningText(current))
283 .filter((current): current is string => current !== null),
284 ].join("\n\n"),
285 );
286 if (aggregate && aggregate === normalizeTranscriptReplayText(normalized)) {
287 return true;
288 }
289 return false;
290}
291
292function clearStepLocalToolRuntime(runtime: ThreadModelStreamRuntime) {
293 runtime.toolItemIdByKey.clear();
294 runtime.latestToolKeyByTurnAndName.clear();
295 runtime.toolInputByKey.clear();
296}
297
298import { isTerminalToolState } from "./toolFeedState";
299
300function isRecord(value: unknown): value is Record<string, unknown> {
301 return typeof value === "object" && value !== null && !Array.isArray(value);
302}
303
304function sortAgentSummaries(agents: ThreadAgentSummary[]): ThreadAgentSummary[] {
305 return [...agents].sort((left, right) => {
306 const updatedDiff = Date.parse(right.updatedAt) - Date.parse(left.updatedAt);
307 if (Number.isFinite(updatedDiff) && updatedDiff !== 0) return updatedDiff;
308 return left.title.localeCompare(right.title);
309 });
310}
311
312function shouldReplaceAgentSummary(
313 existing: ThreadAgentSummary,
314 nextAgent: ThreadAgentSummary,
315): boolean {
316 const existingTs = Date.parse(existing.updatedAt);
317 const nextTs = Date.parse(nextAgent.updatedAt);
318 if (Number.isFinite(existingTs) && Number.isFinite(nextTs) && existingTs !== nextTs) {
319 return nextTs > existingTs;
320 }
321 return true;
322}
323
324export function upsertAgentSummary(
325 agents: ThreadAgentSummary[],
326 nextAgent: ThreadAgentSummary,
327): ThreadAgentSummary[] {
328 const existing = agents.find((agent) => agent.agentId === nextAgent.agentId);
329 if (existing && !shouldReplaceAgentSummary(existing, nextAgent)) {
330 return agents;
331 }
332 const nextAgents = agents.filter((agent) => agent.agentId !== nextAgent.agentId);
333 nextAgents.push(nextAgent);
334 return sortAgentSummaries(nextAgents);
335}
336
337function isTurnUsagePayload(payload: unknown): payload is {
338 type: "turn_usage";
339 turnId: string;
340 usage: {
341 promptTokens: number;
342 completionTokens: number;
343 totalTokens: number;
344 cachedPromptTokens?: number;
345 cacheWritePromptTokens?: number;
346 reasoningOutputTokens?: number;
347 estimatedCostUsd?: number;
348 };
349} {
350 if (!isRecord(payload)) return false;
351 if (payload.type !== "turn_usage") return false;
352 if (typeof payload.turnId !== "string") return false;
353 if (!isRecord(payload.usage)) return false;
354 const hasCanonicalFields =
355 typeof payload.usage.promptTokens === "number" &&
356 typeof payload.usage.completionTokens === "number" &&
357 typeof payload.usage.totalTokens === "number";
358 if (!hasCanonicalFields) return false;
359 if (
360 payload.usage.cachedPromptTokens !== undefined &&
361 typeof payload.usage.cachedPromptTokens !== "number"
362 ) {
363 return false;
364 }
365 if (
366 payload.usage.cacheWritePromptTokens !== undefined &&
367 typeof payload.usage.cacheWritePromptTokens !== "number"
368 ) {
369 return false;
370 }
371 if (
372 payload.usage.reasoningOutputTokens !== undefined &&
373 typeof payload.usage.reasoningOutputTokens !== "number"
374 ) {
375 return false;
376 }
377 if (
378 payload.usage.estimatedCostUsd !== undefined &&
379 typeof payload.usage.estimatedCostUsd !== "number"
380 ) {
381 return false;
382 }
383 return true;
384}
385
386function isSessionUsagePayload(payload: unknown): payload is {
387 type: "session_usage";
388 usage: ThreadRuntime["sessionUsage"];
389} {
390 if (!isRecord(payload)) return false;
391 if (payload.type !== "session_usage") return false;
392 if (payload.usage === null) return true;
393 return sessionUsageSnapshotSchema.safeParse(payload.usage).success;
394}
395
396export function extractUsageStateFromTranscript(
397 transcript: TranscriptEvent[],
398): TranscriptUsageState {
399 let sessionUsage: ThreadRuntime["sessionUsage"] = null;
400 let lastTurnUsage: ThreadRuntime["lastTurnUsage"] = null;
401
402 for (const evt of transcript) {
403 if (evt.direction !== "server") continue;
404 const payload = evt.payload;
405 if (isSessionUsagePayload(payload)) {
406 sessionUsage = payload.usage;
407 continue;
408 }
409 if (isTurnUsagePayload(payload)) {
410 lastTurnUsage = {
411 turnId: payload.turnId,
412 usage: payload.usage,
413 };
414 }
415 }
416
417 return { sessionUsage, lastTurnUsage };
418}
419
420function previewValue(value: unknown, maxChars = 160): string {
421 if (value === undefined) return "";
422 if (typeof value === "string") {
423 return value.length > maxChars ? `${value.slice(0, maxChars - 1)}...` : value;
424 }
425 try {
426 const raw = JSON.stringify(value);
427 if (!raw) return "";
428 return raw.length > maxChars ? `${raw.slice(0, maxChars - 1)}...` : raw;
429 } catch {
430 const fallback = String(value);
431 return fallback.length > maxChars ? `${fallback.slice(0, maxChars - 1)}...` : fallback;
432 }
433}
434
435function incompleteToolStreamResult(error?: unknown): Record<string, unknown> {
436 return {
437 error: error ?? "Turn failed before the tool call completed.",
438 };
439}
440
441function normalizeToolArgsFromInput(inputText: string, existingArgs?: unknown): unknown {
442 const parsed = parseStructuredToolInput(inputText);
443 const base = isRecord(existingArgs) ? existingArgs : {};
444 const { input: _discardInput, ...rest } = base;
445
446 if (isRecord(parsed)) {
447 return { ...rest, ...parsed };
448 }
449
450 if (Object.keys(rest).length > 0) {
451 return { ...rest, input: inputText };
452 }
453
454 return { input: inputText };
455}
456
457function toolTurnNameKey(turnId: string, name: string): string {
458 return `${turnId}:${name}`;
459}
460
461function toolSyntheticApprovalKey(turnId: string, approvalId: string): string {
462 return `${turnId}:approval:${approvalId}`;
463}
464
465function toolNameFromApproval(toolCall: unknown): string {
466 if (isRecord(toolCall)) {
467 const toolName = toolCall.toolName;
468 if (typeof toolName === "string" && toolName.trim().length > 0) return toolName;
469 const name = toolCall.name;
470 if (typeof name === "string" && name.trim().length > 0) return name;
471 }
472 return "tool";
473}
474
475function toolArgsFromApproval(toolCall: unknown): unknown {
476 if (!isRecord(toolCall)) return undefined;
477 if ("args" in toolCall) return toolCall.args;
478 if ("input" in toolCall) return toolCall.input;
479 return undefined;
480}
481
482function rememberLatestToolKey(
483 stream: ThreadModelStreamRuntime,
484 turnId: string,
485 name: string,
486 fullKey: string,
487) {
488 stream.latestToolKeyByTurnAndName.set(toolTurnNameKey(turnId, name), fullKey);
489}
490
491function shouldReuseLatestToolItemByName(name: string): boolean {
492 return name !== "nativeWebSearch" && name !== "nativeUrlContext";
493}
494
495function resolveToolItem(
496 stream: ThreadModelStreamRuntime,
497 turnId: string,
498 key: string,
499 name: string,
500): { fullKey: string; itemId?: string } {
501 const fullKey = `${turnId}:${key}`;
502 const directItemId = stream.toolItemIdByKey.get(fullKey);
503 if (directItemId) {
504 rememberLatestToolKey(stream, turnId, name, fullKey);
505 return { fullKey, itemId: directItemId };
506 }
507
508 if (!shouldReuseLatestToolItemByName(name)) {
509 return { fullKey };
510 }
511
512 const latestKey = stream.latestToolKeyByTurnAndName.get(toolTurnNameKey(turnId, name));
513 if (!latestKey) {
514 return { fullKey };
515 }
516
517 const latestItemId = stream.toolItemIdByKey.get(latestKey);
518 if (!latestItemId) {
519 return { fullKey };
520 }
521
522 if (latestKey !== fullKey) {
523 stream.toolItemIdByKey.delete(latestKey);
524 const latestInput = stream.toolInputByKey.get(latestKey);
525 if (latestInput !== undefined) {
526 stream.toolInputByKey.delete(latestKey);
527 stream.toolInputByKey.set(fullKey, latestInput);
528 }
529 }
530
531 stream.toolItemIdByKey.set(fullKey, latestItemId);
532 rememberLatestToolKey(stream, turnId, name, fullKey);
533 return { fullKey, itemId: latestItemId };
534}
535
536export function shouldSuppressRawDebugLogLine(line: string): boolean {
537 const trimmed = line.trim();
538 if (!trimmed) return false;
539
540 if (/^raw stream part:/i.test(trimmed)) return true;
541 if (/response\.function_call_arguments\./i.test(trimmed)) return true;
542 if (/response\.reasoning(?:_|\.|[a-z])/i.test(trimmed)) return true;
543 if (/"type"\s*:\s*"response\./i.test(trimmed)) return true;
544 if (/\bobfuscation\b/i.test(trimmed)) return true;
545
546 return false;
547}
548
549function recentAssistantTextSinceLastUser(feed: FeedItem[]): string {
550 const assistantTexts: string[] = [];
551 for (let i = feed.length - 1; i >= 0; i -= 1) {
552 const item = feed[i];
553 if (item.kind !== "message") continue;
554 if (item.role === "user") break;
555 const text = item.text.trim();
556 if (item.role === "assistant" && text) {
557 assistantTexts.push(text);
558 }
559 }
560
561 if (assistantTexts.length === 0) return "";
562 return assistantTexts.reverse().join("\n\n").trim();
563}
564
565function normalizeTranscriptReplayText(text: string): string {
566 return text
567 .replace(/\r\n?/g, "\n")
568 .split("\n")
569 .map((line) => line.trim())
570 .join("\n")
571 .replace(/\n{3,}/g, "\n\n")
572 .trim();
573}
574
575function hasVisibleAssistantText(text: string): boolean {
576 return text.trim().length > 0;
577}
578
579export function shouldSkipAssistantMessageAfterStreamReplay(
580 stream: ThreadModelStreamRuntime,
581 assistantText: string,
582 feed: FeedItem[] = [],
583): boolean {
584 const normalizedAssistantText = normalizeTranscriptReplayText(assistantText);
585 if (!normalizedAssistantText) return true;
586
587 if (stream.lastAssistantTurnId) {
588 const assistantKey = stream.lastAssistantStreamKeyByTurn.get(stream.lastAssistantTurnId);
589 const streamed = normalizeTranscriptReplayText(
590 assistantKey ? (stream.assistantTextByStream.get(assistantKey) ?? "") : "",
591 );
592 if (streamed) {
593 if (normalizedAssistantText === streamed) return true;
594 if (streamed.startsWith(normalizedAssistantText)) return true;
595
596 if (stream.replay.rawBackedTurns.has(stream.lastAssistantTurnId)) {
597 return normalizedAssistantText.endsWith(streamed);
598 }
599 }
600 }
601
602 const exactStreamedAssistantText = normalizeTranscriptReplayText(
603 [...stream.assistantTextHistoryInTurn, ...stream.assistantTextByStream.values()].join(""),
604 );
605 if (exactStreamedAssistantText) {
606 if (normalizedAssistantText === exactStreamedAssistantText) return true;
607 if (exactStreamedAssistantText.startsWith(normalizedAssistantText)) return true;
608
609 if (
610 stream.lastAssistantTurnId &&
611 stream.replay.rawBackedTurns.has(stream.lastAssistantTurnId)
612 ) {
613 return normalizedAssistantText.endsWith(exactStreamedAssistantText);
614 }
615 }
616
617 const aggregatedAssistantText = normalizeTranscriptReplayText(
618 recentAssistantTextSinceLastUser(feed),
619 );
620 if (!aggregatedAssistantText) return false;
621 if (normalizedAssistantText === aggregatedAssistantText) return true;
622
623 if (stream.lastAssistantTurnId && stream.replay.rawBackedTurns.has(stream.lastAssistantTurnId)) {
624 return normalizedAssistantText.endsWith(aggregatedAssistantText);
625 }
626
627 return false;
628}
629
630export function reasoningInsertBeforeAssistantAfterStreamReplay(
631 stream: ThreadModelStreamRuntime,
632): string | null {
633 const turnId = stream.lastAssistantTurnId;
634 if (!turnId) return null;
635
636 const assistantKey = stream.lastAssistantStreamKeyByTurn.get(turnId);
637 if (!assistantKey) return null;
638
639 return stream.assistantItemIdByStream.get(assistantKey) ?? null;
640}
641
642function modelStreamSystemLine(update: ModelStreamUpdate): string | null {
643 if (update.kind === "turn_abort") {
644 const reason = previewValue(update.reason);
645 return reason ? `Turn aborted: ${reason}` : "Turn aborted";
646 }
647
648 if (update.kind === "turn_error") {
649 const detail = previewValue(update.error);
650 return detail ? `Stream error: ${detail}` : "Stream error";
651 }
652
653 if (update.kind === "reasoning_start") {
654 return `Reasoning started (${update.mode})`;
655 }
656
657 if (update.kind === "reasoning_end") {
658 return `Reasoning ended (${update.mode})`;
659 }
660
661 if (update.kind === "source") {
662 const sourcePreview = previewValue(update.source);
663 return sourcePreview ? `Source: ${sourcePreview}` : "Source";
664 }
665
666 if (update.kind === "file") {
667 const filePreview = previewValue(update.file);
668 return filePreview ? `File: ${filePreview}` : "File";
669 }
670
671 if (update.kind === "unknown") {
672 const payloadPreview = previewValue(update.payload);
673 return payloadPreview
674 ? `Unhandled stream part (${update.partType}): ${payloadPreview}`
675 : `Unhandled stream part (${update.partType})`;
676 }
677
678 return null;
679}
680
681function applyModelStreamUpdate(
682 stream: ThreadModelStreamRuntime,
683 update: ModelStreamUpdate,
684 ops: ThreadModelStreamFeedOps,
685) {
686 const push = (item: FeedItem) => {
687 ops.pushFeedItem(item);
688 };
689
690 const appendReasoningDelta = (
691 turnId: string,
692 streamId: string,
693 mode: Extract<FeedItem, { kind: "reasoning" }>["mode"],
694 text: string,
695 ) => {
696 stream.lastReasoningTurnId = turnId;
697 stream.reasoningTurns.add(turnId);
698 const key = `${turnId}:${streamId}`;
699 const itemId = stream.reasoningItemIdByStream.get(key);
700 const nextText = `${stream.reasoningTextByStream.get(key) ?? ""}${text}`;
701 stream.reasoningTextByStream.set(key, nextText);
702 if (itemId) {
703 ops.updateFeedItem(itemId, (item) =>
704 item.kind === "reasoning" ? { ...item, mode, text: nextText } : item,
705 );
706 } else if (nextText) {
707 const id = ops.makeId();
708 stream.reasoningItemIdByStream.set(key, id);
709 const item: FeedItem = {
710 id,
711 kind: "reasoning",
712 mode,
713 ts: ops.nowIso(),
714 text: nextText,
715 };
716 const beforeAssistantId =
717 turnId === stream.lastAssistantTurnId
718 ? reasoningInsertBeforeAssistantAfterStreamReplay(stream)
719 : null;
720 if (beforeAssistantId && ops.insertFeedItemBefore) {
721 ops.insertFeedItemBefore(beforeAssistantId, item);
722 } else {
723 push(item);
724 }
725 }
726 };
727
728 const appendAssistantDelta = (turnId: string, streamId: string, text: string) => {
729 if (!text) return;
730 stream.lastAssistantTurnId = turnId;
731 const assistantKey = `${turnId}:${streamId}`;
732 stream.completedAssistantStreamKeys.delete(assistantKey);
733 stream.lastAssistantStreamKeyByTurn.set(turnId, assistantKey);
734 const itemId = stream.assistantItemIdByStream.get(assistantKey);
735 const nextText = `${stream.assistantTextByStream.get(assistantKey) ?? ""}${text}`;
736 stream.assistantTextByStream.set(assistantKey, nextText);
737 if (itemId) {
738 ops.updateFeedItem(itemId, (item) =>
739 item.kind === "message" && item.role === "assistant" ? { ...item, text: nextText } : item,
740 );
741 } else if (hasVisibleAssistantText(nextText)) {
742 const id = ops.makeId();
743 stream.assistantItemIdByStream.set(assistantKey, id);
744 push({ id, kind: "message", role: "assistant", ts: ops.nowIso(), text: nextText });
745 }
746 };
747
748 const failActiveToolStreams = (turnId: string | null, error?: unknown) => {
749 for (const [fullKey, itemId] of [...stream.toolItemIdByKey.entries()]) {
750 if (turnId && !fullKey.startsWith(`${turnId}:`)) continue;
751 ops.updateFeedItem(itemId, (item) => {
752 if (item.kind !== "tool" || isTerminalToolState(item.state)) return item;
753 return { ...item, state: "output-error", result: incompleteToolStreamResult(error) };
754 });
755 }
756 ops.onToolTerminal?.();
757 };
758
759 if (update.kind === "turn_start") {
760 if (stream.activeTurnId !== update.turnId) {
761 clearThreadModelStreamRuntime(stream);
762 stream.activeTurnId = update.turnId;
763 } else {
764 clearStepLocalModelStreamRuntime(stream);
765 clearStepLocalToolRuntime(stream);
766 }
767 return;
768 }
769
770 if (
771 update.kind === "turn_finish" ||
772 update.kind === "step_start" ||
773 update.kind === "step_finish" ||
774 update.kind === "assistant_text_start"
775 ) {
776 // Keep these as state-only boundaries to avoid noisy transcript/feed reconstruction.
777 return;
778 }
779
780 if (update.kind === "turn_error") {
781 failActiveToolStreams(update.turnId, update.error);
782 }
783
784 if (update.kind === "turn_abort") {
785 failActiveToolStreams(update.turnId, update.reason);
786 }
787
788 if (update.kind === "assistant_text_end") {
789 const assistantKey = `${update.turnId}:${update.streamId}`;
790 const itemId = stream.assistantItemIdByStream.get(assistantKey);
791 if (itemId && update.annotations) {
792 ops.updateFeedItem(itemId, (item) =>
793 item.kind === "message" && item.role === "assistant"
794 ? { ...item, annotations: update.annotations }
795 : item,
796 );
797 }
798 return;
799 }
800
801 if (update.kind === "assistant_delta") {
802 appendAssistantDelta(update.turnId, update.streamId, update.text);
803 return;
804 }
805
806 if (update.kind === "reasoning_start" || update.kind === "reasoning_end") {
807 return;
808 }
809
810 if (update.kind === "reasoning_delta") {
811 appendReasoningDelta(update.turnId, update.streamId, update.mode, update.text);
812 return;
813 }
814
815 if (update.kind === "tool_input_start") {
816 const { fullKey, itemId } = resolveToolItem(stream, update.turnId, update.key, update.name);
817 stream.toolInputByKey.delete(fullKey);
818 if (itemId) {
819 ops.updateFeedItem(itemId, (item) =>
820 item.kind === "tool" && !isTerminalToolState(item.state)
821 ? {
822 ...item,
823 name: update.name,
824 state: "input-streaming",
825 args: update.args ?? item.args,
826 }
827 : item,
828 );
829 return;
830 }
831
832 const id = ops.makeId();
833 stream.toolItemIdByKey.set(fullKey, id);
834 rememberLatestToolKey(stream, update.turnId, update.name, fullKey);
835 push({
836 id,
837 kind: "tool",
838 ts: ops.nowIso(),
839 name: update.name,
840 state: "input-streaming",
841 args: update.args,
842 });
843 return;
844 }
845
846 if (update.kind === "tool_input_delta") {
847 const fullKey = `${update.turnId}:${update.key}`;
848 const nextInput = `${stream.toolInputByKey.get(fullKey) ?? ""}${update.delta}`;
849 stream.toolInputByKey.set(fullKey, nextInput);
850 const itemId = stream.toolItemIdByKey.get(fullKey);
851 if (!itemId) {
852 const id = ops.makeId();
853 stream.toolItemIdByKey.set(fullKey, id);
854 push({
855 id,
856 kind: "tool",
857 ts: ops.nowIso(),
858 name: "tool",
859 state: "input-streaming",
860 args: normalizeToolArgsFromInput(nextInput),
861 });
862 return;
863 }
864 ops.updateFeedItem(itemId, (item) => {
865 if (item.kind !== "tool") return item;
866 if (isTerminalToolState(item.state)) return item;
867 return {
868 ...item,
869 state: item.state === "approval-requested" ? item.state : "input-streaming",
870 args: normalizeToolArgsFromInput(nextInput, item.args),
871 };
872 });
873 return;
874 }
875
876 if (update.kind === "tool_input_end") {
877 const { fullKey, itemId } = resolveToolItem(stream, update.turnId, update.key, update.name);
878 const nextInput = stream.toolInputByKey.get(fullKey) ?? "";
879
880 if (itemId) {
881 ops.updateFeedItem(itemId, (item) =>
882 item.kind === "tool" && !isTerminalToolState(item.state)
883 ? {
884 ...item,
885 name: update.name,
886 state: item.state === "approval-requested" ? item.state : "input-available",
887 args: nextInput ? normalizeToolArgsFromInput(nextInput, item.args) : item.args,
888 }
889 : item,
890 );
891 return;
892 }
893
894 if (!nextInput) return;
895 const id = ops.makeId();
896 stream.toolItemIdByKey.set(fullKey, id);
897 rememberLatestToolKey(stream, update.turnId, update.name, fullKey);
898 push({
899 id,
900 kind: "tool",
901 ts: ops.nowIso(),
902 name: update.name,
903 state: "input-available",
904 args: normalizeToolArgsFromInput(nextInput),
905 });
906 return;
907 }
908
909 if (update.kind === "tool_call") {
910 const { fullKey, itemId } = resolveToolItem(stream, update.turnId, update.key, update.name);
911 if (itemId) {
912 ops.updateFeedItem(itemId, (item) =>
913 item.kind === "tool" && !isTerminalToolState(item.state)
914 ? {
915 ...item,
916 name: update.name,
917 state: item.state === "approval-requested" ? item.state : "input-available",
918 args: update.args ?? item.args,
919 }
920 : item,
921 );
922 return;
923 }
924
925 const id = ops.makeId();
926 stream.toolItemIdByKey.set(fullKey, id);
927 rememberLatestToolKey(stream, update.turnId, update.name, fullKey);
928 push({
929 id,
930 kind: "tool",
931 ts: ops.nowIso(),
932 name: update.name,
933 state: "input-available",
934 args: update.args,
935 });
936 return;
937 }
938
939 if (
940 update.kind === "tool_result" ||
941 update.kind === "tool_error" ||
942 update.kind === "tool_output_denied"
943 ) {
944 const { fullKey, itemId } = resolveToolItem(stream, update.turnId, update.key, update.name);
945 const result =
946 update.kind === "tool_result"
947 ? update.result
948 : update.kind === "tool_error"
949 ? { error: update.error }
950 : { denied: true, reason: update.reason };
951 const state =
952 update.kind === "tool_result"
953 ? "output-available"
954 : update.kind === "tool_error"
955 ? "output-error"
956 : "output-denied";
957 const completedAt = ops.nowIso();
958
959 if (itemId) {
960 ops.updateFeedItem(itemId, (item) =>
961 item.kind === "tool"
962 ? {
963 ...item,
964 name: update.name,
965 state,
966 result,
967 completedAt: item.completedAt ?? completedAt,
968 }
969 : item,
970 );
971 } else {
972 const id = ops.makeId();
973 stream.toolItemIdByKey.set(fullKey, id);
974 rememberLatestToolKey(stream, update.turnId, update.name, fullKey);
975 push({ id, kind: "tool", ts: completedAt, name: update.name, state, result, completedAt });
976 }
977
978 ops.onToolTerminal?.();
979 return;
980 }
981
982 if (update.kind === "tool_approval_request") {
983 const name = toolNameFromApproval(update.toolCall);
984 const latestKey = stream.latestToolKeyByTurnAndName.get(toolTurnNameKey(update.turnId, name));
985 const itemId = latestKey ? stream.toolItemIdByKey.get(latestKey) : undefined;
986
987 if (itemId) {
988 ops.updateFeedItem(itemId, (item) =>
989 item.kind === "tool"
990 ? {
991 ...item,
992 name,
993 state: "approval-requested",
994 args: item.args ?? toolArgsFromApproval(update.toolCall),
995 approval: { approvalId: update.approvalId, toolCall: update.toolCall },
996 }
997 : item,
998 );
999 ops.flushPendingContent?.();
1000 return;
1001 }
1002
1003 const id = ops.makeId();
1004 const syntheticKey = toolSyntheticApprovalKey(update.turnId, update.approvalId);
1005 stream.toolItemIdByKey.set(syntheticKey, id);
1006 rememberLatestToolKey(stream, update.turnId, name, syntheticKey);
1007 push({
1008 id,
1009 kind: "tool",
1010 ts: ops.nowIso(),
1011 name,
1012 state: "approval-requested",
1013 args: toolArgsFromApproval(update.toolCall),
1014 approval: { approvalId: update.approvalId, toolCall: update.toolCall },
1015 });
1016 ops.flushPendingContent?.();
1017 return;
1018 }
1019
1020 const systemLine = modelStreamSystemLine(update);
1021 if (systemLine) {
1022 push({ id: ops.makeId(), kind: "system", ts: ops.nowIso(), line: systemLine });
1023 }
1024}
1025
1026function appendModelStreamUpdateToFeed(
1027 out: FeedItem[],
1028 ts: string,
1029 stream: ThreadModelStreamRuntime,
1030 update: ModelStreamUpdate,
1031) {
1032 applyModelStreamUpdate(stream, update, {
1033 makeId: () => crypto.randomUUID(),
1034 nowIso: () => ts,
1035 pushFeedItem: (item) => out.push(item),
1036 insertFeedItemBefore: (beforeItemId, item) => {
1037 const idx = out.findIndex((entry) => entry.id === beforeItemId);
1038 if (idx < 0) {
1039 out.push(item);
1040 return;
1041 }
1042 out.splice(idx, 0, item);
1043 },
1044 updateFeedItem: (itemId, updateItem) => {
1045 const idx = out.findIndex((item) => item.id === itemId);
1046 if (idx < 0) return;
1047 const current = out[idx];
1048 if (!current) return;
1049 out[idx] = updateItem(current);
1050 },
1051 });
1052}
1053
1054function failActiveToolStreamsInFeed(
1055 out: FeedItem[],
1056 stream: ThreadModelStreamRuntime,
1057 turnId: string | null,
1058 error?: unknown,
1059) {
1060 for (const [fullKey, itemId] of [...stream.toolItemIdByKey.entries()]) {
1061 if (turnId && !fullKey.startsWith(`${turnId}:`)) continue;
1062 const idx = out.findIndex((item) => item.id === itemId);
1063 if (idx < 0) continue;
1064 const current = out[idx];
1065 if (current?.kind !== "tool" || isTerminalToolState(current.state)) continue;
1066 out[idx] = { ...current, state: "output-error", result: incompleteToolStreamResult(error) };
1067 }
1068}
1069
1070const transcriptPayloadTypeSchema = z
1071 .object({
1072 type: z.string(),
1073 })
1074 .passthrough();
1075
1076const transcriptUserMessagePayloadSchema = z
1077 .object({
1078 type: z.literal("user_message"),
1079 text: z.unknown().optional(),
1080 clientMessageId: z.string().optional(),
1081 })
1082 .passthrough();
1083
1084const transcriptModelStreamPayloadSchema = z
1085 .object({
1086 type: z.literal("model_stream_chunk"),
1087 })
1088 .passthrough();
1089
1090const transcriptModelStreamRawPayloadSchema = z
1091 .object({
1092 type: z.literal("model_stream_raw"),
1093 })
1094 .passthrough();
1095
1096const transcriptAssistantMessagePayloadSchema = z
1097 .object({
1098 type: z.literal("assistant_message"),
1099 text: z.unknown().optional(),
1100 })
1101 .passthrough();
1102
1103const transcriptReasoningPayloadSchema = z
1104 .object({
1105 type: z.literal("reasoning"),
1106 kind: z.enum(["reasoning", "summary"]).optional(),
1107 text: z.unknown().optional(),
1108 })
1109 .passthrough();
1110
1111const transcriptAssistantReasoningPayloadSchema = z
1112 .object({
1113 type: z.literal("assistant_reasoning"),
1114 text: z.unknown().optional(),
1115 })
1116 .passthrough();
1117
1118const transcriptReasoningSummaryPayloadSchema = z
1119 .object({
1120 type: z.literal("reasoning_summary"),
1121 text: z.unknown().optional(),
1122 })
1123 .passthrough();
1124
1125const transcriptTodosPayloadSchema = z
1126 .object({
1127 type: z.literal("todos"),
1128 todos: z.unknown().optional(),
1129 })
1130 .passthrough();
1131
1132const transcriptLogPayloadSchema = z
1133 .object({
1134 type: z.literal("log"),
1135 line: z.unknown().optional(),
1136 })
1137 .passthrough();
1138
1139const transcriptErrorPayloadSchema = z
1140 .object({
1141 type: z.literal("error"),
1142 message: z.unknown().optional(),
1143 code: z.unknown().optional(),
1144 source: z.unknown().optional(),
1145 })
1146 .passthrough();
1147
1148const transcriptSessionBusyPayloadSchema = z
1149 .object({
1150 type: z.literal("session_busy"),
1151 busy: z.boolean().optional(),
1152 })
1153 .passthrough();
1154
1155const transcriptObservabilityStatusPayloadSchema = z
1156 .object({
1157 type: z.literal("observability_status"),
1158 enabled: z.boolean(),
1159 health: z.object({
1160 status: z.unknown(),
1161 reason: z.unknown(),
1162 message: z.unknown().optional(),
1163 }),
1164 config: z.unknown().optional(),
1165 })
1166 .passthrough();
1167
1168const transcriptSessionBackupStatePayloadSchema = z
1169 .object({
1170 type: z.literal("session_backup_state"),
1171 reason: z.unknown().optional(),
1172 backup: z.unknown().optional(),
1173 })
1174 .passthrough();
1175
1176const transcriptHarnessContextPayloadSchema = z
1177 .object({
1178 type: z.literal("harness_context"),
1179 context: z.unknown().nullable().optional(),
1180 })
1181 .passthrough();
1182
1183const transcriptDeveloperDiagnosticPayloadSchema = z.discriminatedUnion("type", [
1184 transcriptObservabilityStatusPayloadSchema,
1185 transcriptSessionBackupStatePayloadSchema,
1186 transcriptHarnessContextPayloadSchema,
1187]);
1188
1189const transcriptFeedPayloadSchema = z.discriminatedUnion("type", [
1190 transcriptUserMessagePayloadSchema,
1191 transcriptModelStreamPayloadSchema,
1192 transcriptModelStreamRawPayloadSchema,
1193 transcriptAssistantMessagePayloadSchema,
1194 transcriptReasoningPayloadSchema,
1195 transcriptAssistantReasoningPayloadSchema,
1196 transcriptReasoningSummaryPayloadSchema,
1197 transcriptTodosPayloadSchema,
1198 transcriptLogPayloadSchema,
1199 transcriptErrorPayloadSchema,
1200 transcriptSessionBusyPayloadSchema,
1201]);
1202
1203const transcriptFeedSuppressedTypes = new Set([
1204 "steer_message",
1205 "steer_accepted",
1206 "agent_spawned",
1207 "agent_list",
1208 "agent_status",
1209 "agent_wait_result",
1210 "set_session_usage_budget",
1211 "turn_usage",
1212 "session_usage",
1213 "budget_warning",
1214 "budget_exceeded",
1215]);
1216
1217export function extractAgentStateFromTranscript(events: TranscriptEvent[]): ThreadAgentSummary[] {
1218 let agents: ThreadAgentSummary[] = [];
1219
1220 for (const evt of events) {
1221 const parsed = safeParseSessionEvent(evt.payload);
1222 if (!parsed) continue;
1223
1224 if (parsed.type === "agent_spawned" || parsed.type === "agent_status") {
1225 agents = upsertAgentSummary(agents, parsed.agent);
1226 continue;
1227 }
1228
1229 if (parsed.type === "agent_list") {
1230 agents = sortAgentSummaries(parsed.agents);
1231 continue;
1232 }
1233
1234 if (parsed.type === "agent_wait_result") {
1235 for (const agent of parsed.agents) {
1236 agents = upsertAgentSummary(agents, agent);
1237 }
1238 }
1239 }
1240
1241 return agents;
1242}
1243
1244export function mapTranscriptToFeed(events: TranscriptEvent[]): FeedItem[] {
1245 const out: FeedItem[] = [];
1246 const seenUser = new Set<string>();
1247 const stream = createThreadModelStreamRuntime();
1248 const makeId = () => crypto.randomUUID();
1249
1250 for (const evt of events) {
1251 const parsedPayload = transcriptFeedPayloadSchema.safeParse(evt.payload);
1252 if (!parsedPayload.success) {
1253 const developerDiagnosticLine = developerDiagnosticSystemLineFromPayload(evt.payload);
1254 if (developerDiagnosticLine) {
1255 out.push({
1256 id: makeId(),
1257 kind: "system",
1258 ts: evt.ts,
1259 line: developerDiagnosticLine,
1260 });
1261 continue;
1262 }
1263
1264 const parsedTypeOnly = transcriptPayloadTypeSchema.safeParse(evt.payload);
1265 if (!parsedTypeOnly.success) continue;
1266 if (transcriptFeedSuppressedTypes.has(parsedTypeOnly.data.type)) continue;
1267 out.push({
1268 id: makeId(),
1269 kind: "system",
1270 ts: evt.ts,
1271 line: unhandledEventSystemLine(parsedTypeOnly.data.type),
1272 });
1273 continue;
1274 }
1275
1276 const payload = parsedPayload.data;
1277
1278 if (payload.type === "user_message") {
1279 clearThreadModelStreamRuntime(stream);
1280 const cmid = payload.clientMessageId ?? "";
1281 if (cmid && seenUser.has(cmid)) continue;
1282 if (cmid) seenUser.add(cmid);
1283 out.push({
1284 id: cmid || makeId(),
1285 kind: "message",
1286 role: "user",
1287 ts: evt.ts,
1288 text: String(payload.text ?? ""),
1289 });
1290 continue;
1291 }
1292
1293 if (payload.type === "model_stream_chunk") {
1294 if (
1295 shouldIgnoreNormalizedChunkForRawBackedTurn(stream.replay, payload as ModelStreamChunkEvent)
1296 ) {
1297 continue;
1298 }
1299 const mapped = mapModelStreamChunk(payload as ModelStreamChunkEvent);
1300 if (mapped) appendModelStreamUpdateToFeed(out, evt.ts, stream, mapped);
1301 continue;
1302 }
1303
1304 if (payload.type === "model_stream_raw") {
1305 const updates = replayModelStreamRawEvent(stream.replay, payload as ModelStreamRawEvent);
1306 for (const update of updates) {
1307 appendModelStreamUpdateToFeed(out, evt.ts, stream, update);
1308 }
1309 continue;
1310 }
1311
1312 if (payload.type === "assistant_message") {
1313 const text = String(payload.text ?? "");
1314 if (shouldSkipAssistantMessageAfterStreamReplay(stream, text, out)) continue;
1315 out.push({
1316 id: makeId(),
1317 kind: "message",
1318 role: "assistant",
1319 ts: evt.ts,
1320 text,
1321 });
1322 continue;
1323 }
1324
1325 if (
1326 payload.type === "reasoning" ||
1327 payload.type === "assistant_reasoning" ||
1328 payload.type === "reasoning_summary"
1329 ) {
1330 if (hasMatchingStreamedReasoningText(stream, String(payload.text ?? ""))) continue;
1331 const mode =
1332 payload.type === "reasoning_summary"
1333 ? "summary"
1334 : payload.type === "reasoning"
1335 ? payload.kind === "summary"
1336 ? "summary"
1337 : "reasoning"
1338 : "reasoning";
1339 const item: FeedItem = {
1340 id: makeId(),
1341 kind: "reasoning",
1342 mode,
1343 ts: evt.ts,
1344 text: String(payload.text ?? ""),
1345 };
1346 const beforeAssistantId = reasoningInsertBeforeAssistantAfterStreamReplay(stream);
1347 if (beforeAssistantId) {
1348 const assistantIndex = out.findIndex((entry) => entry.id === beforeAssistantId);
1349 if (assistantIndex >= 0) {
1350 out.splice(assistantIndex, 0, item);
1351 continue;
1352 }
1353 }
1354 out.push(item);
1355 continue;
1356 }
1357
1358 if (payload.type === "todos") {
1359 out.push({
1360 id: makeId(),
1361 kind: "todos",
1362 ts: evt.ts,
1363 todos: Array.isArray(payload.todos) ? payload.todos : [],
1364 });
1365 continue;
1366 }
1367
1368 if (payload.type === "log") {
1369 const line = String(payload.line ?? "");
1370 if (shouldSuppressRawDebugLogLine(line)) continue;
1371 out.push({ id: makeId(), kind: "log", ts: evt.ts, line });
1372 continue;
1373 }
1374
1375 if (payload.type === "error") {
1376 failActiveToolStreamsInFeed(out, stream, stream.activeTurnId, payload.message);
1377 out.push({
1378 id: makeId(),
1379 kind: "error",
1380 ts: evt.ts,
1381 message: String(payload.message ?? ""),
1382 code: normalizeServerErrorCode(payload.code),
1383 source: normalizeServerErrorSource(payload.source),
1384 });
1385 continue;
1386 }
1387
1388 if (payload.type === "session_busy") {
1389 if (payload.busy === false) {
1390 const failed =
1391 isRecord(payload) && (payload.outcome === "error" || payload.outcome === "cancelled");
1392 const turnId =
1393 isRecord(payload) && typeof payload.turnId === "string" ? payload.turnId : null;
1394 if (failed) {
1395 failActiveToolStreamsInFeed(out, stream, turnId ?? stream.activeTurnId);
1396 }
1397 clearThreadModelStreamRuntime(stream);
1398 }
1399 out.push({
1400 id: makeId(),
1401 kind: "system",
1402 ts: evt.ts,
1403 line: `[${payload.type}]`,
1404 });
1405 }
1406 }
1407
1408 return out;
1409}
1410
1411export function applyModelStreamUpdateToThreadFeed(
1412 stream: ThreadModelStreamRuntime,
1413 update: ModelStreamUpdate,
1414 ops: ThreadModelStreamFeedOps,
1415) {
1416 applyModelStreamUpdate(stream, update, ops);
1417}
1418
1419export function buildContextPreamble(feed: FeedItem[], maxPairs = 10): string {
1420 const pairs: Array<{ role: "user" | "assistant"; text: string }> = [];
1421 for (let i = feed.length - 1; i >= 0; i--) {
1422 const item = feed[i];
1423 if (item.kind !== "message") continue;
1424 pairs.push({ role: item.role, text: item.text });
1425 if (pairs.length >= maxPairs * 2) break;
1426 }
1427 pairs.reverse();
1428
1429 if (pairs.length === 0) return "";
1430
1431 const lines: string[] = ["Context (previous thread transcript):", ""];
1432 for (const p of pairs) {
1433 lines.push(`${p.role === "user" ? "User" : "Assistant"}: ${p.text}`);
1434 lines.push("");
1435 }
1436 lines.push("---", "");
1437 return lines.join("\n");
1438}
1439
