1import {
2 OrchestratorMcpFailure,
3 OrchestratorMcpThreadListResult,
4 OrchestratorMcpThreadReadResult,
5 OrchestratorMcpThreadSendResult,
6 OrchestratorMcpThreadWaitResult,
7 ProjectId,
8 RunId,
9 ThreadId,
10 TrimmedNonEmptyString,
11 type OrchestratorMcpThreadListItem,
12 type OrchestratorMcpThreadRun,
13 type OrchestratorMcpThreadStatus,
14 type OrchestratorMcpThreadTimelineItem,
15} from "@t3tools/contracts";
16import * as HostProcess from "@t3tools/shared/HostProcess";
17import * as Console from "effect/Console";
18import * as Effect from "effect/Effect";
19import * as FileSystem from "effect/FileSystem";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as Path from "effect/Path";
23import * as References from "effect/References";
24import * as Runtime from "effect/Runtime";
25import * as Schema from "effect/Schema";
26import * as Stdio from "effect/Stdio";
27import * as Stream from "effect/Stream";
28import { Argument, Command, Flag, GlobalFlag } from "effect/cli";
29
30import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
31import * as ServerConfig from "../config.ts";
32import { mcpToolSession, type McpToolSession } from "../mcp/AcpMcpStdioBridge.ts";
33import { expandHomePath } from "../os-jank.ts";
34import * as ProcessRunner from "../processRunner.ts";
35import { readLiveServerRuntimeState } from "../serverRuntimeState.ts";
36import { projectLocationFlags, resolveCliAuthConfig } from "./config.ts";
37
38/**
39 * Short enough that a run which stops to ask the user something is noticed
40 * soon, since `t3_thread_wait` only returns on a terminal run.
41 */
42const WAIT_SLICE_MS = 15_000;
43const WAKE_GRACE_MS = 10_000;
44const WAKE_POLL_MS = 500;
45const TERMINAL_RUN_STATUSES = new Set<OrchestratorMcpThreadRun["status"]>([
46 "completed",
47 "failed",
48 "cancelled",
49 "interrupted",
50 "rolled_back",
51]);
52const READ_PAGE_LIMIT = 100;
53const READ_ITEM_MAX_CHARS = 20_000;
54const TITLE_MAX_CHARS = 80;
55
56export class ThreadServerNotRunningError extends Schema.TaggedError<ThreadServerNotRunningError>()(
57 "ThreadServerNotRunningError",
58 { baseDir: Schema.String },
59) {
60 override readonly [Runtime.errorReported] = false;
61 override get message(): string {
62 return `No running T3 Code server uses ${this.baseDir}. Start T3 Code, then retry.`;
63 }
64}
65
66export class ThreadServerUnreachableError extends Schema.TaggedError<ThreadServerUnreachableError>()(
67 "ThreadServerUnreachableError",
68 { origin: Schema.String, cause: Schema.Defect() },
69) {
70 override readonly [Runtime.errorReported] = false;
71 override get message(): string {
72 const reason = this.cause instanceof Error ? ` ${this.cause.message}` : "";
73 return `Could not reach the T3 Code server at ${this.origin}.${reason}`;
74 }
75}
76
77export class ThreadToolFailedError extends Schema.TaggedError<ThreadToolFailedError>()(
78 "ThreadToolFailedError",
79 { tool: Schema.String, detail: Schema.String },
80) {
81 override readonly [Runtime.errorReported] = false;
82 override get message(): string {
83 return this.detail;
84 }
85}
86
87export class ThreadUsageError extends Schema.TaggedError<ThreadUsageError>()("ThreadUsageError", {
88 detail: Schema.String,
89}) {
90 override readonly [Runtime.errorReported] = false;
91 override get message(): string {
92 return this.detail;
93 }
94}
95
96export class ThreadProjectNotFoundError extends Schema.TaggedError<ThreadProjectNotFoundError>()(
97 "ThreadProjectNotFoundError",
98 { identifier: Schema.String },
99) {
100 override readonly [Runtime.errorReported] = false;
101 override get message(): string {
102 return `No T3 Code project matches '${this.identifier}'. Pass a project id or a path inside a project.`;
103 }
104}
105
106/** Exit code 2 lets a script tell "needs a human" apart from a failed run. */
107export class ThreadNeedsInputError extends Schema.TaggedError<ThreadNeedsInputError>()(
108 "ThreadNeedsInputError",
109 { threadId: ThreadId },
110) {
111 override readonly [Runtime.errorReported] = false;
112 override readonly [Runtime.errorExitCode] = 2;
113
114 override get message(): string {
115 return `Thread ${this.threadId} is waiting for an answer. Open it in T3 Code to respond.`;
116 }
117}
118
119export class ThreadRunEndedError extends Schema.TaggedError<ThreadRunEndedError>()(
120 "ThreadRunEndedError",
121 { threadId: ThreadId, status: Schema.String, error: Schema.optional(Schema.String) },
122) {
123 override readonly [Runtime.errorReported] = false;
124 override get message(): string {
125 const reason = this.error === undefined ? "." : `: ${this.error}`;
126 return `Thread ${this.threadId} stopped with status ${this.status}${reason}`;
127 }
128}
129
130const ThreadCliError = Schema.Union([
131 ThreadServerNotRunningError,
132 ThreadServerUnreachableError,
133 ThreadToolFailedError,
134 ThreadUsageError,
135 ThreadProjectNotFoundError,
136 ThreadNeedsInputError,
137 ThreadRunEndedError,
138]);
139const isThreadCliError = Schema.is(ThreadCliError);
140
141/**
142 * These errors are expected outcomes for a script, so they print as one line
143 * on stderr instead of the runtime's logged stack trace.
144 */
145const reportThreadErrors = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
146 effect.pipe(
147 Effect.tapError((error) =>
148 isThreadCliError(error) ? Console.error(error.message) : Effect.void,
149 ),
150 );
151
152const decodeFailureText = Schema.decodeUnknownOption(Schema.fromJsonString(OrchestratorMcpFailure));
153const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
154const printJson = (value: unknown) => Console.log(encodeJson(value));
155
156const ToolResult = Schema.Struct({
157 isError: Schema.optional(Schema.Boolean),
158 structuredContent: Schema.optional(Schema.Unknown),
159 content: Schema.optional(
160 Schema.Array(Schema.Struct({ type: Schema.String, text: Schema.optional(Schema.String) })),
161 ),
162});
163
164const decodeToolResult = Schema.decodeUnknownEffect(ToolResult);
165
166const ProjectListResult = Schema.Struct({
167 projects: Schema.Array(
168 Schema.Struct({ id: ProjectId, title: Schema.String, workspaceRoot: Schema.String }),
169 ),
170 nextCursor: Schema.NullOr(Schema.Int),
171});
172
173const ThreadLaunchResult = Schema.Struct({
174 threadId: ThreadId,
175 projectId: ProjectId,
176 runId: Schema.NullOr(RunId),
177});
178
179const toolFailureDetail = (result: typeof ToolResult.Type) => {
180 const text = result.content?.find((part) => part.text !== undefined)?.text ?? "";
181 return Option.match(decodeFailureText(text), {
182 onNone: () => text || "The tool call failed.",
183 onSome: (failure) => failure.message,
184 });
185};
186
187const makeThreadTools = (session: McpToolSession, origin: string) => {
188 const unexpected = (tool: string) =>
189 new ThreadToolFailedError({
190 tool,
191 detail: `The server sent an unexpected ${tool} result. Update the t3 CLI to match the server.`,
192 });
193 return {
194 call: Effect.fn("cli.thread.callTool")(function* <S extends Schema.Decoder<unknown>>(
195 tool: string,
196 args: Readonly<Record<string, unknown>>,
197 schema: S,
198 ): Effect.fn.Return<S["Type"], ThreadServerUnreachableError | ThreadToolFailedError> {
199 const raw = yield* session
200 .call(tool, args)
201 .pipe(Effect.mapError((cause) => new ThreadServerUnreachableError({ origin, cause })));
202 const result = yield* decodeToolResult(raw).pipe(Effect.mapError(() => unexpected(tool)));
203 if (result.isError === true) {
204 return yield* new ThreadToolFailedError({ tool, detail: toolFailureDetail(result) });
205 }
206 return yield* Schema.decodeUnknownEffect(schema)(result.structuredContent).pipe(
207 Effect.mapError(() => unexpected(tool)),
208 );
209 }),
210 };
211};
212export type ThreadTools = ReturnType<typeof makeThreadTools>;
213
214/**
215 * Signs in as a short-lived MCP client of the running server, the same way an
216 * outside agent does, and revokes that session when the command ends.
217 */
218const withThreadTools = Effect.fn("cli.thread.withThreadTools")(function* <A, E, R>(
219 flags: { readonly baseDir: Option.Option<string> },
220 run: (tools: ThreadTools) => Effect.Effect<A, E, R>,
221) {
222 const logLevel = yield* GlobalFlag.LogLevel;
223 const config = yield* resolveCliAuthConfig(flags, logLevel);
224 return yield* Effect.gen(function* () {
225 const runtimeState = yield* readLiveServerRuntimeState(config.serverRuntimeStatePath);
226 if (Option.isNone(runtimeState)) {
227 return yield* new ThreadServerNotRunningError({ baseDir: config.baseDir });
228 }
229 const origin = runtimeState.value.origin;
230 const environmentAuth = yield* EnvironmentAuth.EnvironmentAuth;
231 const issued = yield* Effect.acquireRelease(
232 environmentAuth.issueMcpClientSession({
233 label: "t3 thread cli",
234 // The CLI runs as the machine's owner. New threads still start at
235 // the project's default mode, never at this ceiling by default.
236 access: "full-access",
237 peerLink: false,
238 client: { deviceType: "bot" },
239 }),
240 (session) =>
241 environmentAuth.revokeSession(session.sessionId).pipe(Effect.ignore({ log: true })),
242 );
243 const session = yield* mcpToolSession({
244 endpoint: new URL("/mcp", origin).href,
245 authorization: `Bearer ${issued.token}`,
246 clientName: "t3-thread-cli",
247 }).pipe(Effect.mapError((cause) => new ThreadServerUnreachableError({ origin, cause })));
248 return yield* run(makeThreadTools(session, origin));
249 }).pipe(
250 Effect.scoped,
251 Effect.provide(
252 EnvironmentAuth.layerRuntime.pipe(
253 Layer.provide(ServerConfig.layer(config)),
254 Layer.provide(Layer.succeed(References.MinimumLogLevel, config.logLevel)),
255 ),
256 ),
257 );
258});
259
260const readMessageArgument = Effect.fn("cli.thread.readMessage")(function* (argument: string) {
261 const text =
262 argument === "-"
263 ? yield* Stdio.Stdio.pipe(
264 Effect.flatMap((stdio) => stdio.stdin.pipe(Stream.decodeText(), Stream.mkString)),
265 )
266 : argument;
267 if (text.trim().length === 0) {
268 return yield* new ThreadUsageError({ detail: "The message is empty." });
269 }
270 return text;
271});
272
273const listAllProjects = Effect.fn("cli.thread.listAllProjects")(function* (tools: ThreadTools) {
274 const projects: Array<(typeof ProjectListResult.Type)["projects"][number]> = [];
275 let cursor: number | null = 0;
276 while (cursor !== null) {
277 const page: typeof ProjectListResult.Type = yield* tools.call(
278 "t3_project_list",
279 { cursor, limit: 100 },
280 ProjectListResult,
281 );
282 projects.push(...page.projects);
283 cursor = page.nextCursor;
284 }
285 return projects;
286});
287
288/** The checkouts of the git repository at `directory`, or none outside a repository. */
289const listGitCheckouts = Effect.fn("cli.thread.listGitCheckouts")(function* (directory: string) {
290 const processRunner = yield* ProcessRunner.ProcessRunner;
291 const list = (args: ReadonlyArray<string>) =>
292 processRunner
293 .run({ command: "git", args: ["worktree", "list", "--porcelain", ...args], cwd: directory })
294 .pipe(
295 Effect.option,
296 Effect.map((result) =>
297 Option.isSome(result) && result.value.code === 0 ? result.value.stdout : undefined,
298 ),
299 );
300 // Pre-2.36 Git rejects worktree list -z; normalize its line output to NUL records.
301 const stdout = (yield* list(["-z"])) ?? (yield* list([]))?.replaceAll("\n", "\0") ?? "";
302 return stdout.split("\0\0").flatMap((record) => {
303 const [first = "", ...attributes] = record.split("\0");
304 return first.startsWith("worktree ") && !attributes.includes("bare")
305 ? [first.slice("worktree ".length)]
306 : [];
307 });
308});
309
310/**
311 * A project id, or a path inside a project's folder (deepest project wins).
312 * A path inside a git worktree resolves as the same path in the repository's
313 * other checkouts.
314 */
315export const resolveProject = Effect.fn("cli.thread.resolveProject")(function* (
316 tools: ThreadTools,
317 identifier: string,
318) {
319 const projects = yield* listAllProjects(tools);
320 const byId = projects.find((project) => project.id === identifier);
321 if (byId !== undefined) return byId;
322
323 const path = yield* Path.Path;
324 const fileSystem = yield* FileSystem.FileSystem;
325 // Stored roots keep the path the project was added with, which may pass
326 // through a symlink (macOS /tmp), so both sides compare as real paths.
327 const realPath = (value: string) =>
328 fileSystem.realPath(value).pipe(Effect.orElseSucceed(() => value));
329 const target = yield* realPath(
330 path.resolve(yield* HostProcess.WorkingDirectory, yield* expandHomePath(identifier)),
331 );
332 const roots = yield* Effect.forEach(projects, (project) =>
333 Effect.map(realPath(project.workspaceRoot), (root) => ({ project, root })),
334 );
335 const contains = (root: string, candidate: string) => {
336 const relative = path.relative(root, candidate);
337 return (
338 relative === "" ||
339 (relative !== ".." && !relative.startsWith(`..${path.sep}`) && !path.isAbsolute(relative))
340 );
341 };
342 const deepestContaining = (candidate: string) =>
343 roots
344 .filter(({ root }) => contains(root, candidate))
345 .toSorted((left, right) => right.root.length - left.root.length)[0]?.project;
346
347 const containing = deepestContaining(target);
348 if (containing !== undefined) return containing;
349 const targetInfo = yield* fileSystem.stat(target).pipe(Effect.option);
350 const gitDirectory =
351 Option.isSome(targetInfo) && targetInfo.value.type === "File" ? path.dirname(target) : target;
352 const listed = yield* listGitCheckouts(gitDirectory).pipe(
353 Effect.flatMap((paths) => Effect.forEach(paths, realPath)),
354 Effect.provide(ProcessRunner.layer),
355 );
356 const ownCheckout = listed
357 .filter((checkout) => contains(checkout, target))
358 .toSorted((left, right) => right.length - left.length)[0];
359 if (ownCheckout === undefined) {
360 return yield* new ThreadProjectNotFoundError({ identifier });
361 }
362 const relative = path.relative(ownCheckout, target);
363 const matches = new Map<ProjectId, (typeof projects)[number]>();
364 for (const checkout of listed) {
365 if (checkout === ownCheckout) continue;
366 const project = deepestContaining(path.join(checkout, relative));
367 if (project !== undefined) matches.set(project.id, project);
368 }
369 const [match, ...others] = matches.values();
370 if (match === undefined) {
371 return yield* new ThreadProjectNotFoundError({ identifier });
372 }
373 if (others.length > 0) {
374 const names = [match, ...others].map((project) => `${project.title} (${project.id})`);
375 return yield* new ThreadUsageError({
376 detail: `'${identifier}' is in a git worktree shared by several projects: ${names.join(", ")}. Pass a project id instead.`,
377 });
378 }
379 return match;
380});
381
382/** Every item on a timeline view, oldest first. */
383const readTimeline = Effect.fn("cli.thread.readTimeline")(function* (
384 tools: ThreadTools,
385 threadId: ThreadId,
386 view: "messages" | "activity",
387) {
388 const items: Array<OrchestratorMcpThreadTimelineItem> = [];
389 let afterPosition: number | undefined;
390 let page: OrchestratorMcpThreadReadResult;
391 do {
392 page = yield* tools.call(
393 "t3_thread_read",
394 {
395 threadId,
396 view,
397 limit: READ_PAGE_LIMIT,
398 maxCharsPerItem: READ_ITEM_MAX_CHARS,
399 runLimit: 1,
400 ...(afterPosition === undefined ? {} : { afterPosition }),
401 },
402 OrchestratorMcpThreadReadResult,
403 );
404 items.push(...page.items);
405 afterPosition = page.nextPosition ?? undefined;
406 } while (page.hasMore && afterPosition !== undefined);
407 return { thread: page.thread, items };
408});
409
410/**
411 * The latest `count` user and assistant messages, oldest first. Positions
412 * count every timeline item, so the read starts near the end and widens
413 * backward only while tool and plan items leave too few messages.
414 */
415export const readLatestMessages = Effect.fn("cli.thread.readLatestMessages")(function* (
416 tools: ThreadTools,
417 threadId: ThreadId,
418 count: number,
419) {
420 let { thread } = yield* tools.call(
421 "t3_thread_read",
422 { threadId, limit: 1, runLimit: 1 },
423 OrchestratorMcpThreadReadResult,
424 );
425 let messages: Array<OrchestratorMcpThreadTimelineItem> = [];
426 // The newest window has no upper bound, so items added since the first
427 // read are still shown.
428 let before: number | undefined;
429 let start = thread.itemCount;
430 let span = Math.max(count, READ_PAGE_LIMIT);
431 do {
432 const end = before;
433 start = Math.max(0, start - span);
434 const window: Array<OrchestratorMcpThreadTimelineItem> = [];
435 let afterPosition = start - 1;
436 let page: OrchestratorMcpThreadReadResult;
437 do {
438 page = yield* tools.call(
439 "t3_thread_read",
440 {
441 threadId,
442 view: "messages",
443 limit: READ_PAGE_LIMIT,
444 maxCharsPerItem: READ_ITEM_MAX_CHARS,
445 runLimit: 1,
446 ...(afterPosition < 0 ? {} : { afterPosition }),
447 },
448 OrchestratorMcpThreadReadResult,
449 );
450 window.push(
451 ...page.items.filter(
452 (item) =>
453 (item.type === "user_message" || item.type === "assistant_message") &&
454 (end === undefined || item.position < end),
455 ),
456 );
457 afterPosition = page.nextPosition ?? Number.POSITIVE_INFINITY;
458 } while (page.hasMore && afterPosition < (end ?? Number.POSITIVE_INFINITY) - 1);
459 thread = page.thread;
460 // A window can hold the whole --messages count, too many to spread as arguments.
461 messages = window.concat(messages);
462 before = start;
463 span *= 2;
464 } while (messages.length < count && start > 0);
465 return { thread, messages: messages.slice(Math.max(0, messages.length - count)) };
466});
467
468/** Item text past the read's per-item cap. */
469const readFullText = Effect.fn("cli.thread.readFullText")(function* (
470 tools: ThreadTools,
471 threadId: ThreadId,
472 item: OrchestratorMcpThreadTimelineItem,
473) {
474 let text = item.text ?? "";
475 let nextTextOffset = item.textTruncated ? (item.nextTextOffset ?? null) : null;
476 while (nextTextOffset !== null) {
477 const page = yield* tools.call(
478 "t3_thread_read",
479 {
480 threadId,
481 itemId: item.itemId,
482 textOffset: nextTextOffset,
483 runLimit: 1,
484 maxCharsPerItem: READ_ITEM_MAX_CHARS,
485 },
486 OrchestratorMcpThreadReadResult,
487 );
488 const chunk = page.items.find((candidate) => candidate.itemId === item.itemId);
489 if (chunk === undefined) break;
490 text += chunk.text ?? "";
491 nextTextOffset = chunk.textTruncated ? (chunk.nextTextOffset ?? null) : null;
492 }
493 return text;
494});
495
496/** The run's last assistant message, or its last error when it did not complete. */
497const runResultText = Effect.fn("cli.thread.runResultText")(function* (
498 tools: ThreadTools,
499 threadId: ThreadId,
500 runId: RunId,
501 type: "assistant_message" | "error",
502) {
503 const { items } = yield* readTimeline(
504 tools,
505 threadId,
506 type === "assistant_message" ? "messages" : "activity",
507 );
508 const item = items.findLast((candidate) => candidate.runId === runId && candidate.type === type);
509 return item === undefined ? null : yield* readFullText(tools, threadId, item);
510});
511
512type WaitOutcome =
513 | {
514 readonly type: "ended";
515 readonly runId: RunId | null;
516 readonly status: OrchestratorMcpThreadStatus;
517 }
518 | { readonly type: "needs_input" };
519
520/**
521 * Whether the thread has nothing left to do: no active run and no delegated
522 * or background work. Only then is the work a later wait settles on this
523 * send's own.
524 */
525const threadIsIdle = Effect.fn("cli.thread.threadIsIdle")(function* (
526 tools: ThreadTools,
527 threadId: ThreadId,
528) {
529 const waited = yield* tools.call(
530 "t3_thread_wait",
531 { threadId, timeoutMs: 1, untilDelegatedWorkSettled: true },
532 OrchestratorMcpThreadWaitResult,
533 );
534 return !waited.timedOut;
535});
536
537/**
538 * The runs after `runId`, and whether one holds a message someone sent.
539 * Background work wakes a run with a notification, never a user message, so
540 * such a run is another request and its reply is not this one's.
541 */
542const runsAfter = Effect.fn("cli.thread.runsAfter")(function* (
543 tools: ThreadTools,
544 threadId: ThreadId,
545 runId: RunId,
546) {
547 const { recentRuns } = yield* tools.call(
548 "t3_thread_read",
549 { threadId, limit: 1, runLimit: 50 },
550 OrchestratorMcpThreadReadResult,
551 );
552 const own = recentRuns.find((run) => run.runId === runId);
553 if (own === undefined) return { later: [], sent: true };
554 const later = recentRuns.filter((run) => run.ordinal > own.ordinal);
555 if (later.length === 0) return { later, sent: false };
556 const { items } = yield* readTimeline(tools, threadId, "messages");
557 const sent = items.some(
558 (item) => item.type === "user_message" && later.some((run) => run.runId === item.runId),
559 );
560 return { later, sent };
561});
562
563/**
564 * Waits for this exact run, following a restart to its continuation, so a
565 * message another client sends meanwhile cannot stand in for its result.
566 * With `followWakes`, a completed run is followed further to the runs its
567 * background work wakes, as long as no one else sends to the thread.
568 */
569export const waitForRun = Effect.fn("cli.thread.waitForRun")(function* (
570 tools: ThreadTools,
571 threadId: ThreadId,
572 runId: RunId,
573 followWakes = false,
574) {
575 const waitSlice = (timeoutMs: number, untilDelegatedWorkSettled?: true) =>
576 tools.call(
577 "t3_thread_wait",
578 {
579 threadId,
580 runId,
581 timeoutMs,
582 ...(untilDelegatedWorkSettled ? { untilDelegatedWorkSettled } : {}),
583 },
584 OrchestratorMcpThreadWaitResult,
585 );
586 const ended = (waited: OrchestratorMcpThreadWaitResult) =>
587 ({ type: "ended", runId: waited.runId, status: waited.status }) satisfies WaitOutcome;
588 const ownRun = Effect.gen(function* () {
589 while (true) {
590 const waited = yield* waitSlice(WAIT_SLICE_MS);
591 if (!waited.timedOut) return ended(waited);
592 // A pending request on this run, or on an earlier run this one is queued
593 // behind, blocks it until someone answers.
594 const { thread } = yield* tools.call(
595 "t3_thread_read",
596 { threadId, limit: 1, runLimit: 1 },
597 OrchestratorMcpThreadReadResult,
598 );
599 if (thread.pendingRequestCount > 0) {
600 // The count is thread-wide: if this run ended since the slice, the
601 // request belongs to a later run.
602 const recheck = yield* waitSlice(1);
603 return recheck.timedOut ? ({ type: "needs_input" } satisfies WaitOutcome) : ended(recheck);
604 }
605 }
606 });
607 const outcome = yield* ownRun;
608 if (!followWakes || outcome.type !== "ended" || outcome.status !== "completed") return outcome;
609 const finalRunId = outcome.runId ?? runId;
610 // A provider records the run that background work wakes only after that
611 // work has ended, so the thread briefly looks settled in between.
612 const wakeRunRecorded = Effect.gen(function* () {
613 for (let waited = 0; waited < WAKE_GRACE_MS; waited += WAKE_POLL_MS) {
614 yield* Effect.sleep(WAKE_POLL_MS);
615 const after = yield* runsAfter(tools, threadId, finalRunId);
616 if (after.sent) return false;
617 if (after.later.length > 0) return true;
618 }
619 return false;
620 });
621 let workOutlivedRun = false;
622 while (true) {
623 const waited = yield* waitSlice(WAIT_SLICE_MS, true);
624 // `resultRun` is the thread's latest finished run, whoever sent it.
625 const result = waited.timedOut ? undefined : waited.resultRun;
626 const after = yield* runsAfter(tools, threadId, finalRunId);
627 if (after.sent) return outcome;
628 if (!waited.timedOut) {
629 if (result !== undefined && after.later.some((run) => run.runId === result.runId)) {
630 return { type: "ended", runId: result.runId, status: result.status } satisfies WaitOutcome;
631 }
632 if (!workOutlivedRun) return outcome;
633 if (after.later.some((run) => !TERMINAL_RUN_STATUSES.has(run.status))) continue;
634 if (after.later.length === 0 && (yield* wakeRunRecorded)) continue;
635 return outcome;
636 }
637 workOutlivedRun = true;
638 const { thread } = yield* tools.call(
639 "t3_thread_read",
640 { threadId, limit: 1, runLimit: 1 },
641 OrchestratorMcpThreadReadResult,
642 );
643 // Earlier runs have ended, so the request is on a run this one's work woke.
644 if (thread.pendingRequestCount > 0) return { type: "needs_input" } satisfies WaitOutcome;
645 }
646});
647
648/**
649 * With --json the result always reaches stdout, so a script keeps the thread
650 * id even when the run fails.
651 */
652const finishRun = Effect.fn("cli.thread.finishRun")(function* (
653 tools: ThreadTools,
654 input: {
655 readonly threadId: ThreadId;
656 readonly runId: RunId;
657 readonly json: boolean;
658 readonly followWakes: boolean;
659 },
660) {
661 const outcome = yield* waitForRun(tools, input.threadId, input.runId, input.followWakes);
662 const runId = outcome.type === "ended" ? (outcome.runId ?? input.runId) : input.runId;
663 const status = outcome.type === "ended" ? outcome.status : "waiting";
664 const reply =
665 status === "completed"
666 ? yield* runResultText(tools, input.threadId, runId, "assistant_message")
667 : null;
668 const error =
669 status === "completed" || status === "waiting"
670 ? null
671 : yield* runResultText(tools, input.threadId, runId, "error");
672 if (input.json) {
673 yield* printJson({ threadId: input.threadId, runId, status, reply, error });
674 } else if (reply !== null) {
675 yield* Console.log(reply);
676 }
677 if (outcome.type === "needs_input") {
678 return yield* new ThreadNeedsInputError({ threadId: input.threadId });
679 }
680 if (status !== "completed") {
681 return yield* new ThreadRunEndedError({
682 threadId: input.threadId,
683 status,
684 ...(error === null ? {} : { error }),
685 });
686 }
687});
688
689export const titleFromMessage = (message: string) => {
690 const firstLine = message.trim().split("\n")[0]?.trim() ?? "";
691 if (firstLine.length <= TITLE_MAX_CHARS) return firstLine;
692 let end = TITLE_MAX_CHARS - 1;
693 const previous = firstLine.charCodeAt(end - 1);
694 const next = firstLine.charCodeAt(end);
695 if (previous >= 0xd800 && previous <= 0xdbff && next >= 0xdc00 && next <= 0xdfff) {
696 end -= 1;
697 }
698 return `${firstLine.slice(0, end).trimEnd()}…`;
699};
700
701const launchWorkspaceStrategy = (input: {
702 readonly worktree: Option.Option<string>;
703 readonly branch: Option.Option<string>;
704}) =>
705 Option.match(input.worktree, {
706 onNone: () =>
707 Option.isSome(input.branch)
708 ? Effect.fail(new ThreadUsageError({ detail: "--branch needs --worktree <base-ref>." }))
709 : Effect.void,
710 onSome: (baseRef) =>
711 Effect.succeed({
712 type: "worktree" as const,
713 baseRef,
714 ...Option.match(input.branch, { onNone: () => ({}), onSome: (branch) => ({ branch }) }),
715 }),
716 });
717
718const jsonFlag = Flag.Boolean("json").pipe(
719 Flag.withDescription("Print JSON instead of text."),
720 Flag.withDefault(false),
721);
722const waitFlag = Flag.Boolean("wait").pipe(
723 Flag.withDescription(
724 "Wait for the agent to finish and print its reply. Exits 1 if the run fails and 2 if it waits for an answer.",
725 ),
726 Flag.withDefault(false),
727);
728const messageArgument = Argument.String("message").pipe(
729 Argument.withDescription("Message for the agent, or - to read it from stdin."),
730);
731const threadArgument = Argument.String("thread").pipe(Argument.withDescription("Thread id."));
732
733const decodeThreadIdValue = Schema.decodeUnknownEffect(ThreadId);
734const decodeTitle = Schema.decodeUnknownEffect(TrimmedNonEmptyString);
735const decodeThreadId = (value: string) =>
736 decodeThreadIdValue(value.trim()).pipe(
737 Effect.mapError(() => new ThreadUsageError({ detail: `'${value}' is not a thread id.` })),
738 );
739
740const threadListCommand = Command.make("list", {
741 ...projectLocationFlags,
742 project: Flag.String("project").pipe(
743 Flag.withDescription("Project id or a path inside the project. Defaults to this directory."),
744 Flag.withDefault("."),
745 ),
746 json: jsonFlag,
747}).pipe(
748 Command.withDescription("List a project's threads, newest first."),
749 Command.withHandler((flags) =>
750 withThreadTools(
751 flags,
752 Effect.fn("cli.thread.list")(function* (tools) {
753 const project = yield* resolveProject(tools, flags.project);
754 const threads: Array<OrchestratorMcpThreadListItem> = [];
755 let cursor: number | null = 0;
756 while (cursor !== null) {
757 const page: OrchestratorMcpThreadListResult = yield* tools.call(
758 "t3_thread_list",
759 { projectId: project.id, includeSubagents: false, limit: 100, cursor },
760 OrchestratorMcpThreadListResult,
761 );
762 threads.push(...page.threads);
763 cursor = page.nextCursor;
764 }
765 if (flags.json) {
766 return yield* printJson(threads);
767 }
768 for (const thread of threads) {
769 yield* Console.log(`${thread.threadId} ${thread.status.padEnd(11)} ${thread.title}`);
770 }
771 }),
772 ).pipe(reportThreadErrors),
773 ),
774);
775
776const threadStartCommand = Command.make("start", {
777 ...projectLocationFlags,
778 project: Argument.String("project").pipe(
779 Argument.withDescription("Project id or a path inside the project, such as ."),
780 ),
781 message: messageArgument,
782 title: Flag.String("title").pipe(
783 Flag.withDescription("Thread title. Defaults to the message's first line."),
784 Flag.optional,
785 ),
786 provider: Flag.String("provider").pipe(
787 Flag.withDescription("Provider instance id. Pass with --model."),
788 Flag.optional,
789 ),
790 model: Flag.String("model").pipe(
791 Flag.withDescription("Model slug. Defaults to the project's default model."),
792 Flag.optional,
793 ),
794 worktree: Flag.String("worktree").pipe(
795 Flag.withDescription(
796 "Run in a new git worktree based on this branch, tag, or commit, instead of the project folder.",
797 ),
798 Flag.optional,
799 ),
800 branch: Flag.String("branch").pipe(
801 Flag.withDescription("Branch name for the new worktree."),
802 Flag.optional,
803 ),
804 wait: waitFlag,
805 json: jsonFlag,
806}).pipe(
807 Command.withDescription("Start a thread in a project and print its id."),
808 Command.withHandler((flags) =>
809 Effect.gen(function* () {
810 const message = yield* readMessageArgument(flags.message);
811 const workspaceStrategy = yield* launchWorkspaceStrategy(flags);
812 if (Option.isSome(flags.provider) !== Option.isSome(flags.model)) {
813 return yield* new ThreadUsageError({ detail: "--provider and --model go together." });
814 }
815 const title = yield* decodeTitle(
816 Option.getOrElse(flags.title, () => titleFromMessage(message)),
817 ).pipe(Effect.mapError(() => new ThreadUsageError({ detail: "The title is empty." })));
818 yield* withThreadTools(
819 flags,
820 Effect.fn("cli.thread.start")(function* (tools) {
821 const project = yield* resolveProject(tools, flags.project);
822 const launched = yield* tools.call(
823 "t3_thread_launch",
824 {
825 projectId: project.id,
826 title,
827 message,
828 ...(workspaceStrategy === undefined ? {} : { workspaceStrategy }),
829 ...(Option.isSome(flags.provider) && Option.isSome(flags.model)
830 ? {
831 modelSelection: {
832 instanceId: flags.provider.value,
833 model: flags.model.value,
834 },
835 }
836 : {}),
837 },
838 ThreadLaunchResult,
839 );
840 if (flags.wait && launched.runId !== null) {
841 if (!flags.json) yield* Console.error(`Thread ${launched.threadId}`);
842 return yield* finishRun(tools, {
843 threadId: launched.threadId,
844 runId: launched.runId,
845 json: flags.json,
846 followWakes: true,
847 });
848 }
849 yield* flags.json ? printJson(launched) : Console.log(launched.threadId);
850 }),
851 );
852 }).pipe(reportThreadErrors),
853 ),
854);
855
856const threadSendCommand = Command.make("send", {
857 ...projectLocationFlags,
858 thread: threadArgument,
859 message: messageArgument,
860 wait: waitFlag,
861 json: jsonFlag,
862}).pipe(
863 Command.withDescription(
864 "Send a message to a thread. A busy thread gets it as its next turn; nothing is interrupted.",
865 ),
866 Command.withHandler((flags) =>
867 Effect.gen(function* () {
868 const threadId = yield* decodeThreadId(flags.thread);
869 const message = yield* readMessageArgument(flags.message);
870 yield* withThreadTools(
871 flags,
872 Effect.fn("cli.thread.send")(function* (tools) {
873 // Work already running when this is sent is not this message's to wait for.
874 const followWakes = flags.wait ? yield* threadIsIdle(tools, threadId) : false;
875 const sent = yield* tools.call(
876 "t3_thread_send",
877 { threadId, message, mode: "queue" },
878 OrchestratorMcpThreadSendResult,
879 );
880 if (flags.wait) {
881 return yield* finishRun(tools, {
882 threadId,
883 runId: sent.runId,
884 json: flags.json,
885 followWakes,
886 });
887 }
888 yield* flags.json
889 ? printJson(sent)
890 : Console.log(sent.delivery === "queued" ? "Queued." : "Sent.");
891 }),
892 );
893 }).pipe(reportThreadErrors),
894 ),
895);
896
897const threadShowCommand = Command.make("show", {
898 ...projectLocationFlags,
899 thread: threadArgument,
900 messages: Flag.Int("messages").pipe(
901 Flag.withDescription("How many of the latest messages to print."),
902 Flag.withDefault(10),
903 ),
904 json: jsonFlag,
905}).pipe(
906 Command.withDescription("Show a thread's state and its latest messages."),
907 Command.withHandler((flags) =>
908 Effect.gen(function* () {
909 const threadId = yield* decodeThreadId(flags.thread);
910 yield* withThreadTools(
911 flags,
912 Effect.fn("cli.thread.show")(function* (tools) {
913 const { thread, messages: latest } = yield* readLatestMessages(
914 tools,
915 threadId,
916 flags.messages,
917 );
918 const messages = yield* Effect.forEach(latest, (item) =>
919 readFullText(tools, threadId, item).pipe(
920 Effect.map((text) => ({
921 role: item.type === "user_message" ? "user" : "assistant",
922 runId: item.runId,
923 text,
924 })),
925 ),
926 );
927 if (flags.json) {
928 return yield* printJson({ thread, messages });
929 }
930 yield* Console.log(thread.title);
931 yield* Console.log(
932 [thread.threadId, thread.status, `${thread.providerInstanceId}/${thread.model}`].join(
933 " ",
934 ),
935 );
936 if (thread.worktreePath !== null) {
937 yield* Console.log(`${thread.branch ?? "(no branch)"} ${thread.worktreePath}`);
938 }
939 if (thread.pendingRequestCount > 0) {
940 yield* Console.log(`Waiting for ${thread.pendingRequestCount} answer(s).`);
941 }
942 for (const entry of messages) {
943 yield* Console.log(`\n[${entry.role}]\n${entry.text}`);
944 }
945 }),
946 );
947 }).pipe(reportThreadErrors),
948 ),
949);
950
951export const threadCommand = Command.make("thread").pipe(
952 Command.withDescription("Start, message, and read threads on the running server."),
953 Command.withSubcommands([
954 threadListCommand,
955 threadStartCommand,
956 threadSendCommand,
957 threadShowCommand,
958 ]),
959);