Open History to see the last commit
1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { assert, it } from "@effect/vitest";
4import { AuthSessionId, ProjectId, RunId, ThreadId } from "@t3tools/contracts";
5import * as HostProcess from "@t3tools/shared/HostProcess";
6import * as NetService from "@t3tools/shared/Net";
7import * as ConfigProvider from "effect/ConfigProvider";
8import * as DateTime from "effect/DateTime";
9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as FileSystem from "effect/FileSystem";
12import * as Layer from "effect/Layer";
13import * as Path from "effect/Path";
14import { ChildProcess, ChildProcessSpawner } from "effect/process";
15import * as TestClock from "effect/testing/TestClock";
16import * as TestConsole from "effect/testing/TestConsole";
17import { Command } from "effect/cli";
18import { vi } from "vite-plus/test";
19
20import * as McpBridge from "../mcp/AcpMcpStdioBridge.ts";
21import {
22 makePersistedServerRuntimeState,
23 persistServerRuntimeState,
24} from "../serverRuntimeState.ts";
25import {
26 readLatestMessages,
27 resolveProject,
28 threadCommand,
29 type ThreadTools,
30 titleFromMessage,
31 waitForRun,
32} from "./thread.ts";
33
34vi.mock("../auth/EnvironmentAuth.ts", async (importOriginal) => {
35 const actual = await importOriginal<typeof import("../auth/EnvironmentAuth.ts")>();
36 return {
37 ...actual,
38 layerRuntime: Layer.mock(actual.EnvironmentAuth)({
39 issueMcpClientSession: () =>
40 Effect.map(DateTime.now, (expiresAt) => ({
41 sessionId: AuthSessionId.make("test-session"),
42 token: "synthetic-token",
43 expiresAt,
44 })),
45 revokeSession: () => Effect.succeed(true),
46 }),
47 };
48});
49
50const threadId = ThreadId.make("thread-1");
51const runId = RunId.make("run-1");
52
53it.effect(
54 "start resolves home-relative project paths while preserving ids and relative paths",
55 () =>
56 Effect.gen(function* () {
57 const fs = yield* FileSystem.FileSystem;
58 const path = yield* Path.Path;
59 const root = yield* fs
60 .makeTempDirectoryScoped({ prefix: "t3-thread-project-path-" })
61 .pipe(Effect.flatMap((directory) => fs.realPath(directory)));
62 const home = path.join(root, "home");
63 const cwd = path.join(root, "current");
64 const workspaceRoot = path.join(home, "My Project");
65 const nestedRoot = path.join(workspaceRoot, "nested");
66 const baseDir = path.join(root, "state");
67 yield* fs.makeDirectory(nestedRoot, { recursive: true });
68 yield* fs.makeDirectory(cwd);
69 const projects = [
70 { id: ProjectId.make("current-project"), title: "Current", workspaceRoot: cwd },
71 { id: ProjectId.make("home-project"), title: "Home", workspaceRoot: home },
72 { id: ProjectId.make("intended-project"), title: "Intended", workspaceRoot },
73 { id: ProjectId.make("nested-project"), title: "Nested", workspaceRoot: nestedRoot },
74 { id: ProjectId.make("~/literal-id"), title: "Exact id", workspaceRoot: cwd },
75 ];
76 yield* persistServerRuntimeState({
77 path: path.join(baseDir, "userdata", "server-runtime.json"),
78 state: yield* makePersistedServerRuntimeState({
79 config: { host: "127.0.0.1", devUrl: undefined },
80 port: 3_773,
81 }),
82 });
83 using _session = vi.spyOn(McpBridge, "mcpToolSession").mockReturnValue(
84 Effect.succeed({
85 call: (tool, args) => {
86 switch (tool) {
87 case "t3_project_list":
88 return Effect.succeed({ structuredContent: { projects, nextCursor: null } });
89 case "t3_thread_launch":
90 return Effect.succeed({
91 structuredContent: { threadId, projectId: args.projectId, runId: null },
92 });
93 default:
94 return Effect.die(`unexpected ${tool} call`);
95 }
96 },
97 }),
98 );
99
100 for (const [identifier, expectedProjectId] of [
101 ["intended-project", "intended-project"],
102 ["~/literal-id", "~/literal-id"],
103 [workspaceRoot, "intended-project"],
104 [path.relative(cwd, workspaceRoot), "intended-project"],
105 [".", "current-project"],
106 ["~/My Project", "intended-project"],
107 ["~", "home-project"],
108 ["~/My Project/nested", "nested-project"],
109 ["~/My Project/missing", "intended-project"],
110 ] as const) {
111 yield* Command.runWith(threadCommand, { version: "0.0.0" })([
112 "start",
113 identifier,
114 "Inspect this project",
115 "--json",
116 "--base-dir",
117 baseDir,
118 ]).pipe(
119 Effect.provideService(HostProcess.HomeDirectory, home),
120 Effect.provideService(HostProcess.WorkingDirectory, cwd),
121 );
122 assert.strictEqual(
123 (yield* TestConsole.logLines).at(-1),
124 JSON.stringify({ threadId, projectId: expectedProjectId, runId: null }),
125 identifier,
126 );
127 }
128 }).pipe(
129 Effect.scoped,
130 Effect.provide(
131 Layer.mergeAll(
132 NodeServices.layer,
133 NodeCrypto.layer,
134 NetService.layer,
135 ConfigProvider.layer(ConfigProvider.fromEnv({ env: { T3CODE_PORT: "3773" } })),
136 ),
137 ),
138 ),
139);
140
141it.effect("does not sign in to whatever took over a dead server's port", () =>
142 Effect.gen(function* () {
143 const fs = yield* FileSystem.FileSystem;
144 const path = yield* Path.Path;
145 const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-thread-dead-server-" });
146 const state = yield* makePersistedServerRuntimeState({
147 config: { host: "127.0.0.1", devUrl: undefined },
148 port: 3_773,
149 });
150 yield* persistServerRuntimeState({
151 path: path.join(baseDir, "userdata", "server-runtime.json"),
152 // Above Linux's PID ceiling, as in pair.test.ts: no process can hold it.
153 state: { ...state, pid: 4_194_305 },
154 });
155 using session = vi
156 .spyOn(McpBridge, "mcpToolSession")
157 .mockReturnValue(Effect.die("no session should open"));
158
159 const error = yield* Command.runWith(threadCommand, { version: "0.0.0" })([
160 "list",
161 "--base-dir",
162 baseDir,
163 ]).pipe(Effect.flip);
164
165 assert.strictEqual(error._tag, "ThreadServerNotRunningError");
166 assert.strictEqual(session.mock.calls.length, 0);
167 }).pipe(
168 Effect.scoped,
169 Effect.provide(
170 Layer.mergeAll(
171 NodeServices.layer,
172 NodeCrypto.layer,
173 NetService.layer,
174 ConfigProvider.layer(ConfigProvider.fromEnv({ env: { T3CODE_PORT: "3773" } })),
175 ),
176 ),
177 ),
178);
179
180/** Answers each tool call with the next scripted response for that tool. */
181const scriptedTools = (script: Record<string, Array<unknown>>) => {
182 const calls: Array<{ readonly tool: string; readonly args: unknown }> = [];
183 const call = ((tool: string, args: unknown) => {
184 calls.push({ tool, args });
185 const next = script[tool]?.shift();
186 return next === undefined ? Effect.die(`unscripted ${tool} call`) : Effect.succeed(next);
187 }) as ThreadTools["call"];
188 return { tools: { call } satisfies ThreadTools, calls };
189};
190
191it.effect("--wait follows the sent run to its continuation's terminal status", () =>
192 Effect.gen(function* () {
193 const continuation = RunId.make("run-2");
194 const { tools, calls } = scriptedTools({
195 t3_thread_wait: [
196 { threadId, runId, status: "running", timedOut: true },
197 { threadId, runId: continuation, status: "failed", timedOut: false },
198 ],
199 t3_thread_read: [{ thread: { pendingRequestCount: 0 } }],
200 });
201
202 const outcome = yield* waitForRun(tools, threadId, runId);
203
204 assert.deepStrictEqual(outcome, { type: "ended", runId: continuation, status: "failed" });
205 assert.deepStrictEqual(
206 calls.filter((entry) => entry.tool === "t3_thread_wait").map((entry) => entry.args),
207 [
208 { threadId, runId, timeoutMs: 15_000 },
209 { threadId, runId, timeoutMs: 15_000 },
210 ],
211 );
212 }),
213);
214
215/** The reads that ask whether a run after `runId` holds a message someone sent. */
216const laterRunReads = (
217 laterRun: RunId,
218 items: ReadonlyArray<{ readonly runId: RunId; readonly type: string }>,
219) => [
220 {
221 recentRuns: [
222 { runId, ordinal: 1 },
223 { runId: laterRun, ordinal: 2 },
224 ],
225 },
226 { thread: {}, items, nextPosition: null, hasMore: false },
227];
228
229it.effect("--wait lasts until background work the agent waits on wakes it and that run ends", () =>
230 Effect.gen(function* () {
231 const wakeRun = RunId.make("run-2");
232 const { tools, calls } = scriptedTools({
233 t3_thread_wait: [
234 { threadId, runId, status: "completed", timedOut: false },
235 {
236 threadId,
237 runId,
238 status: "completed",
239 timedOut: false,
240 resultRun: { runId: wakeRun, status: "completed" },
241 },
242 ],
243 // The wake run starts from a notification, which the messages view leaves out.
244 t3_thread_read: laterRunReads(wakeRun, []),
245 });
246
247 const outcome = yield* waitForRun(tools, threadId, runId, true);
248
249 assert.deepStrictEqual(
250 calls.filter((entry) => entry.tool === "t3_thread_wait").map((entry) => entry.args),
251 [
252 { threadId, runId, timeoutMs: 15_000 },
253 { threadId, runId, timeoutMs: 15_000, untilDelegatedWorkSettled: true },
254 ],
255 );
256 assert.deepStrictEqual(outcome, { type: "ended", runId: wakeRun, status: "completed" });
257 }),
258);
259
260it.effect(
261 "--wait waits out the moment between background work ending and its wake run starting",
262 () =>
263 Effect.gen(function* () {
264 const wakeRun = RunId.make("run-2");
265 const ownOnly = { recentRuns: [{ runId, ordinal: 1, status: "completed" }] };
266 const withWake = (status: string) => [
267 {
268 recentRuns: [
269 { runId, ordinal: 1, status: "completed" },
270 { runId: wakeRun, ordinal: 2, status },
271 ],
272 },
273 { thread: {}, items: [], nextPosition: null, hasMore: false },
274 ];
275 const { tools } = scriptedTools({
276 t3_thread_wait: [
277 { threadId, runId, status: "completed", timedOut: false },
278 { threadId, runId, status: "completed", timedOut: true },
279 // The work ended, and the run it wakes is not recorded yet.
280 {
281 threadId,
282 runId,
283 status: "completed",
284 timedOut: false,
285 resultRun: { runId, status: "completed" },
286 },
287 {
288 threadId,
289 runId,
290 status: "completed",
291 timedOut: false,
292 resultRun: { runId: wakeRun, status: "completed" },
293 },
294 ],
295 t3_thread_read: [
296 ownOnly,
297 { thread: { pendingRequestCount: 0 } },
298 ownOnly,
299 ...withWake("running"),
300 ...withWake("completed"),
301 ],
302 });
303
304 const fiber = yield* Effect.forkChild(waitForRun(tools, threadId, runId, true));
305 yield* TestClock.adjust("500 millis");
306
307 assert.deepStrictEqual(yield* Fiber.join(fiber), {
308 type: "ended",
309 runId: wakeRun,
310 status: "completed",
311 });
312 }),
313);
314
315it.effect(
316 "--wait keeps its own run's result when another client's later run settles the thread",
317 () =>
318 Effect.gen(function* () {
319 const sentRun = RunId.make("run-2");
320 const { tools } = scriptedTools({
321 t3_thread_wait: [
322 { threadId, runId, status: "completed", timedOut: false },
323 {
324 threadId,
325 runId,
326 status: "completed",
327 timedOut: false,
328 resultRun: { runId: sentRun, status: "failed" },
329 },
330 ],
331 t3_thread_read: laterRunReads(sentRun, [{ runId: sentRun, type: "user_message" }]),
332 });
333
334 assert.deepStrictEqual(yield* waitForRun(tools, threadId, runId, true), {
335 type: "ended",
336 runId,
337 status: "completed",
338 });
339 }),
340);
341
342it.effect("--wait stops when the thread waits for an answer instead of polling forever", () =>
343 Effect.gen(function* () {
344 const { tools } = scriptedTools({
345 t3_thread_wait: [
346 { threadId, runId, status: "queued", timedOut: true },
347 { threadId, runId, status: "queued", timedOut: true },
348 ],
349 t3_thread_read: [{ thread: { pendingRequestCount: 1 } }],
350 });
351
352 assert.deepStrictEqual(yield* waitForRun(tools, threadId, runId), { type: "needs_input" });
353 }),
354);
355
356it.effect("--wait reports its run's result when a later run is the one waiting", () =>
357 Effect.gen(function* () {
358 const { tools } = scriptedTools({
359 t3_thread_wait: [
360 { threadId, runId, status: "running", timedOut: true },
361 { threadId, runId, status: "completed", timedOut: false },
362 ],
363 t3_thread_read: [{ thread: { pendingRequestCount: 1 } }],
364 });
365
366 assert.deepStrictEqual(yield* waitForRun(tools, threadId, runId), {
367 type: "ended",
368 runId,
369 status: "completed",
370 });
371 }),
372);
373
374/** Serves t3_thread_read pages the way the server does: positions count every item. */
375const timelineTools = (types: ReadonlyArray<string>) => {
376 const items = types.map((type, position) => ({ position, itemId: `item-${position}`, type }));
377 const served: Array<number> = [];
378 const call = ((_tool: string, input: unknown) => {
379 const args = input as { readonly afterPosition?: number; readonly limit: number };
380 // Scanning one match past the page keeps reads of huge fixtures linear.
381 const matching: typeof items = [];
382 for (let position = (args.afterPosition ?? -1) + 1; position < items.length; position++) {
383 const item = items[position]!;
384 if (["user_message", "assistant_message", "proposed_plan"].includes(item.type)) {
385 matching.push(item);
386 if (matching.length > args.limit) break;
387 }
388 }
389 const page = matching.slice(0, args.limit);
390 served.push(page.length);
391 return Effect.succeed({
392 thread: { itemCount: items.length },
393 items: page,
394 nextPosition: page.at(-1)?.position ?? null,
395 hasMore: matching.length > page.length,
396 });
397 }) as ThreadTools["call"];
398 const latest = (count: number) =>
399 items
400 .filter((item) => item.type === "user_message" || item.type === "assistant_message")
401 .slice(-count)
402 .map((item) => item.itemId);
403 return { tools: { call } satisfies ThreadTools, served, latest };
404};
405
406const tail = (tools: number) => [...Array<string>(tools).fill("tool_call"), "assistant_message"];
407
408it.effect("show reads the end of a long thread, not its whole history", () =>
409 Effect.gen(function* () {
410 const { tools, served, latest } = timelineTools([
411 ...Array.from({ length: 1000 }, (_, index) =>
412 index % 2 === 0 ? "user_message" : "assistant_message",
413 ),
414 ...Array.from({ length: 10 }, () => tail(30)).flat(),
415 "proposed_plan",
416 ]);
417
418 const { messages } = yield* readLatestMessages(tools, threadId, 10);
419
420 assert.deepStrictEqual(
421 messages.map((item) => item.itemId),
422 latest(10),
423 );
424 assert.isBelow(
425 served.reduce((sum, count) => sum + count, 0),
426 100,
427 );
428 }),
429);
430
431it.effect("show widens back to the first item when the thread has fewer messages", () =>
432 Effect.gen(function* () {
433 const { tools, latest } = timelineTools([
434 "user_message",
435 ...Array.from({ length: 4 }, () => tail(150)).flat(),
436 ]);
437
438 const { messages } = yield* readLatestMessages(tools, threadId, 10);
439
440 assert.deepStrictEqual(
441 messages.map((item) => item.itemId),
442 latest(10),
443 );
444 assert.lengthOf(messages, 5);
445 }),
446);
447
448it.effect("show returns a --messages count larger than one call can spread", () =>
449 Effect.gen(function* () {
450 const { tools, latest } = timelineTools(Array<string>(200_000).fill("assistant_message"));
451
452 const { messages } = yield* readLatestMessages(tools, threadId, 200_000);
453
454 assert.deepStrictEqual(
455 messages.map((item) => item.itemId),
456 latest(200_000),
457 );
458 }),
459);
460
461it("titles from a long first line without splitting an emoji", () => {
462 const title = titleFromMessage(`${"a".repeat(78)}😀 and the rest of a long first line`);
463 assert.strictEqual(title, `${"a".repeat(78)}…`);
464});
465
466const git = Effect.fn("test.git")(function* (cwd: string, args: ReadonlyArray<string>) {
467 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
468 const exitCode = yield* spawner.exitCode(ChildProcess.make("git", args, { cwd }));
469 assert.strictEqual(Number(exitCode), 0, `git ${args.join(" ")}`);
470});
471
472const commitAll = (cwd: string) =>
473 Effect.andThen(
474 git(cwd, ["add", "."]),
475 git(cwd, [
476 "-c",
477 "user.name=Test",
478 "-c",
479 "user.email=test@example.test",
480 "commit",
481 "--quiet",
482 "-m",
483 "initial",
484 ]),
485 );
486
487it.live("resolves a path inside a git worktree of the project's repository to that project", () =>
488 Effect.gen(function* () {
489 const fileSystem = yield* FileSystem.FileSystem;
490 const path = yield* Path.Path;
491 const root = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-thread-worktree-" });
492 const repo = path.join(root, "sample");
493 const worktree = path.join(root, "worktrees", "sample", "t3-8ff64c24");
494 const feature = path.join(root, "wt", "feature");
495 const featureWorktree = path.join(root, "worktrees", "feature", "t3-1a2b3c4d");
496 yield* fileSystem.makeDirectory(path.join(repo, "pkg"), { recursive: true });
497 yield* fileSystem.writeFileString(path.join(repo, "pkg", "index.ts"), "export {};\n");
498 yield* git(repo, ["init", "--quiet", "-b", "main"]);
499 yield* commitAll(repo);
500 yield* git(repo, ["worktree", "add", "--quiet", "-b", "t3/8ff64c24", worktree]);
501 yield* git(repo, ["worktree", "add", "--quiet", "-b", "feature", feature]);
502 yield* git(feature, ["worktree", "add", "--quiet", "-b", "t3/1a2b3c4d", featureWorktree]);
503
504 const project = (id: string, workspaceRoot: string) => ({
505 id: ProjectId.make(id),
506 title: path.basename(workspaceRoot),
507 workspaceRoot,
508 });
509 const sample = project("project-sample", repo);
510 const featureProject = project("project-feature", feature);
511 const resolveFrom = (
512 projects: ReadonlyArray<ReturnType<typeof project>>,
513 workingDirectory: string,
514 identifier = ".",
515 ) =>
516 resolveProject(
517 scriptedTools({ t3_project_list: [{ projects, nextCursor: null }] }).tools,
518 identifier,
519 ).pipe(Effect.provideService(HostProcess.WorkingDirectory, workingDirectory));
520
521 assert.deepStrictEqual(yield* resolveFrom([sample], worktree), sample);
522 assert.deepStrictEqual(yield* resolveFrom([sample], path.join(worktree, "pkg")), sample);
523 assert.deepStrictEqual(yield* resolveFrom([sample], worktree, "pkg/index.ts"), sample);
524 assert.deepStrictEqual(yield* resolveFrom([featureProject], featureWorktree), featureProject);
525 const ambiguous = yield* Effect.flip(resolveFrom([sample, featureProject], featureWorktree));
526 assert.strictEqual(ambiguous._tag, "ThreadUsageError");
527 assert.include(ambiguous.message, "sample (project-sample)");
528 assert.include(ambiguous.message, "feature (project-feature)");
529 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
530);