1import {
2 isOrchestrationV2WorkActive,
3 OrchestrationV2CommandOutputError,
4 type OrchestrationV2CommandOutputFrame,
5 type OrchestrationV2SubscribeCommandOutputInput,
6 type OrchestrationV2TurnItem,
7} from "@t3tools/contracts";
8import {
9 appendTerminalOutput,
10 EMPTY_TERMINAL_OUTPUT,
11 normalizeTerminalOutput,
12 type TerminalOutputState,
13 TERMINAL_OUTPUT_TAIL_CHARS,
14 terminalOutputResumeText,
15} from "@t3tools/shared/terminalOutput";
16import * as Cause from "effect/Cause";
17import * as Clock from "effect/Clock";
18import * as Context from "effect/Context";
19import * as Effect from "effect/Effect";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as Queue from "effect/Queue";
23import * as Schema from "effect/Schema";
24import * as Stream from "effect/Stream";
25
26import * as EventSink from "./EventSink.ts";
27import * as ProjectionStore from "./ProjectionStore.ts";
28
29/** Output bypasses orchestration events; expanded rows receive tails, coalesced appends, then persisted output. */
30
31/** Fastest a subscriber receives frames while output flows. */
32const COMMAND_OUTPUT_FRAME_INTERVAL_MS = 100;
33/** After a subscriber falls behind and is sent the whole tail, it waits this long. */
34const COMMAND_OUTPUT_REPLACE_INTERVAL_MS = 500;
35/** Raw text a subscriber may accumulate before it is sent the tail instead. */
36const COMMAND_OUTPUT_MAX_APPEND_CHARS = 16 * 1024;
37/** Evict the least recently written command beyond this limit. */
38const MAX_LIVE_COMMANDS = 64;
39/** Retain finished tails for settlements that omit output. */
40const MAX_FINISHED_COMMANDS = 16;
41const MAX_FINAL_SOURCE_CHARS = TERMINAL_OUTPUT_TAIL_CHARS * 4;
42
43interface Subscriber {
44 pendingAppend: string;
45 needsReplace: boolean;
46 /** `needsReplace` was caused by the subscriber falling behind. */
47 overflowed: boolean;
48 readonly wake: Queue.Queue<void>;
49}
50
51export type CommandOutputTarget = OrchestrationV2SubscribeCommandOutputInput;
52
53export interface CommandOutputHubShape {
54 readonly append: (input: CommandOutputTarget & { readonly chunk: string }) => Effect.Effect<void>;
55 /** Retains the tail briefly for settlements without output; otherwise rows use persisted output. */
56 readonly finish: (input: CommandOutputTarget) => Effect.Effect<void>;
57 readonly subscribe: (
58 input: CommandOutputTarget,
59 ) => Stream.Stream<OrchestrationV2CommandOutputFrame, OrchestrationV2CommandOutputError>;
60}
61
62export class CommandOutputHub extends Context.Service<CommandOutputHub, CommandOutputHubShape>()(
63 "t3/orchestration-v2/CommandOutputHub",
64) {}
65
66const isCommandOutputError = Schema.is(OrchestrationV2CommandOutputError);
67
68function commandKey(target: CommandOutputTarget): string {
69 return `${target.threadId}\u0000${target.itemId}`;
70}
71
72const decodeLegacyClaudeBashResult = Schema.decodeUnknownOption(
73 Schema.fromJsonString(
74 Schema.Struct({ stdout: Schema.String, stderr: Schema.String, interrupted: Schema.Boolean }),
75 ),
76);
77
78function legacyClaudeBashOutput(output: string): string {
79 if (!output.trimStart().startsWith('{"stdout"')) return output;
80 const parsed = decodeLegacyClaudeBashResult(output);
81 if (Option.isNone(parsed)) return output;
82 const out = parsed.value.stdout.trim().length > 0 ? parsed.value.stdout : "";
83 const err = parsed.value.stderr.trim().length > 0 ? parsed.value.stderr : "";
84 return out.length > 0 && err.length > 0 && !out.endsWith("\n") ? `${out}\n${err}` : out + err;
85}
86
87function persistedCommandOutput(output: string | undefined): {
88 readonly text: string;
89 readonly truncated: boolean;
90} {
91 if (output === undefined) return { text: "", truncated: false };
92 let source = output;
93 if (output.length > MAX_FINAL_SOURCE_CHARS) {
94 // Cut at a newline to avoid splitting escape sequences.
95 source = output.slice(output.length - MAX_FINAL_SOURCE_CHARS);
96 const newline = source.indexOf("\n");
97 if (newline !== -1 && newline + 1 < source.length) source = source.slice(newline + 1);
98 }
99 const normalized = normalizeTerminalOutput(source);
100 return {
101 text: normalized.text,
102 truncated: normalized.truncated || source.length !== output.length,
103 };
104}
105
106function setBounded<V>(
107 map: Map<string, V>,
108 key: string,
109 value: V,
110 max: number,
111 onEvict?: (key: string) => void,
112): void {
113 map.delete(key);
114 map.set(key, value);
115 if (map.size <= max) return;
116 const oldest = map.keys().next().value!;
117 map.delete(oldest);
118 onEvict?.(oldest);
119}
120
121export const make = Effect.gen(function* () {
122 const eventSink = yield* EventSink.EventSinkV2;
123 const projections = yield* ProjectionStore.ProjectionStoreV2;
124 const tails = new Map<string, TerminalOutputState>();
125 const finished = new Map<string, TerminalOutputState>();
126 const subscribers = new Map<string, Set<Subscriber>>();
127
128 const wakeAll = (key: string, chunk: string) =>
129 Effect.forEach(
130 subscribers.get(key) ?? [],
131 (subscriber) => {
132 if (!subscriber.needsReplace) {
133 subscriber.pendingAppend += chunk;
134 if (subscriber.pendingAppend.length > COMMAND_OUTPUT_MAX_APPEND_CHARS) {
135 subscriber.needsReplace = true;
136 subscriber.overflowed = true;
137 subscriber.pendingAppend = "";
138 }
139 }
140 return Queue.offer(subscriber.wake, undefined);
141 },
142 { discard: true },
143 );
144
145 const append: CommandOutputHubShape["append"] = ({ chunk, ...target }) =>
146 Effect.suspend(() => {
147 if (chunk.length === 0) return Effect.void;
148 const key = commandKey(target);
149 const tail = appendTerminalOutput(tails.get(key) ?? EMPTY_TERMINAL_OUTPUT, chunk);
150 setBounded(tails, key, tail, MAX_LIVE_COMMANDS, (oldest) => {
151 // Eviction invalidates viewers' state; resync on their next frame.
152 for (const subscriber of subscribers.get(oldest) ?? []) {
153 subscriber.needsReplace = true;
154 subscriber.pendingAppend = "";
155 }
156 });
157 return wakeAll(key, chunk);
158 });
159
160 const finish: CommandOutputHubShape["finish"] = (target) =>
161 Effect.sync(() => {
162 const key = commandKey(target);
163 const tail = tails.get(key);
164 if (tail === undefined) return;
165 tails.delete(key);
166 setBounded(finished, key, tail, MAX_FINISHED_COMMANDS);
167 });
168
169 const subscribe: CommandOutputHubShape["subscribe"] = (target) =>
170 Stream.unwrap(
171 Effect.gen(function* () {
172 const key = commandKey(target);
173 const wake = yield* Queue.sliding<void>(1);
174 const subscriber: Subscriber = {
175 pendingAppend: "",
176 needsReplace: true,
177 overflowed: false,
178 wake,
179 };
180 // Subscribe before reading to avoid losing intervening chunks.
181 const keySubscribers = subscribers.get(key) ?? new Set<Subscriber>();
182 keySubscribers.add(subscriber);
183 subscribers.set(key, keySubscribers);
184 yield* Effect.addFinalizer(() =>
185 Effect.sync(() => {
186 keySubscribers.delete(subscriber);
187 if (keySubscribers.size === 0 && subscribers.get(key) === keySubscribers) {
188 subscribers.delete(key);
189 }
190 }),
191 );
192
193 // Persisted items provide settlement and final output; snapshot-only providers (ACP, Pi) also stream here.
194 let snapshotOutput: string | undefined;
195 let lost: Cause.Cause<unknown> | undefined;
196 let settled: Extract<OrchestrationV2TurnItem, { type: "command_execution" }> | undefined;
197 const observe = (item: OrchestrationV2TurnItem | undefined) =>
198 Effect.suspend(() => {
199 if (item === undefined || item.type !== "command_execution") return Effect.void;
200 if (!isOrchestrationV2WorkActive(item.status)) {
201 settled = item;
202 } else if (item.output !== undefined && item.output !== snapshotOutput) {
203 snapshotOutput = item.output;
204 } else {
205 return Effect.void;
206 }
207 subscriber.needsReplace = true;
208 return Queue.offer(wake, undefined);
209 });
210 const fromSequence = yield* eventSink.latestSequence({ threadId: target.threadId });
211 yield* eventSink
212 .stream({
213 threadId: target.threadId,
214 afterSequence: fromSequence,
215 eventType: "turn-item.updated",
216 })
217 .pipe(
218 Stream.runForEach((stored) =>
219 stored.event.type === "turn-item.updated" && stored.event.payload.id === target.itemId
220 ? observe(stored.event.payload)
221 : Effect.void,
222 ),
223 // A lost event stream cannot settle the row; end with an error.
224 Effect.catchCause((cause) =>
225 Effect.sync(() => {
226 lost = cause;
227 }).pipe(Effect.andThen(Queue.offer(wake, undefined))),
228 ),
229 Effect.forkScoped,
230 );
231 const item = yield* projections.getTurnItem(target);
232 // Non-command rows would hold output subscriptions open forever.
233 if (item === null || item.type !== "command_execution") {
234 return yield* new OrchestrationV2CommandOutputError({
235 threadId: target.threadId,
236 message: "No such command.",
237 });
238 }
239 yield* observe(item);
240 yield* Queue.offer(wake, undefined);
241
242 let lastFrameAt: number | undefined;
243 let throttle = COMMAND_OUTPUT_FRAME_INTERVAL_MS;
244 const nextFrame = Effect.gen(function* (): Effect.fn.Return<
245 OrchestrationV2CommandOutputFrame,
246 OrchestrationV2CommandOutputError
247 > {
248 yield* Queue.take(wake);
249 if (lost !== undefined) {
250 return yield* new OrchestrationV2CommandOutputError({
251 threadId: target.threadId,
252 message: "Lost the thread's event stream.",
253 cause: Cause.squash(lost),
254 });
255 }
256 const now = yield* Clock.currentTimeMillis;
257 if (lastFrameAt !== undefined && now - lastFrameAt < throttle) {
258 yield* Effect.sleep(throttle - (now - lastFrameAt));
259 }
260 lastFrameAt = yield* Clock.currentTimeMillis;
261 throttle = COMMAND_OUTPUT_FRAME_INTERVAL_MS;
262 const tail = tails.get(key) ?? finished.get(key);
263 if (settled !== undefined) {
264 if (settled.output === undefined && tail !== undefined) {
265 return {
266 kind: "replace",
267 text: tail.text,
268 truncated: tail.truncated,
269 running: false,
270 };
271 }
272 const output = settled.output && legacyClaudeBashOutput(settled.output);
273 return { kind: "replace", ...persistedCommandOutput(output), running: false };
274 }
275 if (subscriber.needsReplace || tail === undefined) {
276 if (subscriber.overflowed) throttle = COMMAND_OUTPUT_REPLACE_INTERVAL_MS;
277 subscriber.needsReplace = false;
278 subscriber.overflowed = false;
279 subscriber.pendingAppend = "";
280 if (tail === undefined) {
281 return { kind: "replace", ...persistedCommandOutput(snapshotOutput), running: true };
282 }
283 // Include terminal state so clients interpret later appends identically.
284 return {
285 kind: "replace",
286 text: terminalOutputResumeText(tail),
287 truncated: tail.truncated,
288 running: true,
289 };
290 }
291 const text = subscriber.pendingAppend;
292 subscriber.pendingAppend = "";
293 return { kind: "append", text, truncated: tail.truncated, running: true };
294 });
295 return Stream.fromEffectRepeat(nextFrame).pipe(
296 Stream.filter((frame) => frame.kind !== "append" || frame.text.length > 0),
297 Stream.takeUntil((frame) => !frame.running),
298 );
299 }).pipe(
300 Effect.mapError((cause) =>
301 isCommandOutputError(cause)
302 ? cause
303 : new OrchestrationV2CommandOutputError({
304 threadId: target.threadId,
305 message: "Failed to read command output.",
306 cause,
307 }),
308 ),
309 ),
310 );
311
312 return CommandOutputHub.of({ append, finish, subscribe });
313});
314
315export const layer: Layer.Layer<
316 CommandOutputHub,
317 never,
318 EventSink.EventSinkV2 | ProjectionStore.ProjectionStoreV2
319> = Layer.effect(CommandOutputHub, make);