diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index 5ab4e24d0a8e1..5252d81b2a22f 100644 --- a/apps/server/src/cli/thread.test.ts +++ b/apps/server/src/cli/thread.test.ts @@ -7,10 +7,12 @@ import * as NetService from "@t3tools/shared/Net"; import * as ConfigProvider from "effect/ConfigProvider"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import * as TestClock from "effect/testing/TestClock"; import * as TestConsole from "effect/testing/TestConsole"; import { Command } from "effect/cli"; import { vi } from "vite-plus/test"; @@ -210,6 +212,133 @@ it.effect("--wait follows the sent run to its continuation's terminal status", ( }), ); +/** The reads that ask whether a run after `runId` holds a message someone sent. */ +const laterRunReads = ( + laterRun: RunId, + items: ReadonlyArray<{ readonly runId: RunId; readonly type: string }>, +) => [ + { + recentRuns: [ + { runId, ordinal: 1 }, + { runId: laterRun, ordinal: 2 }, + ], + }, + { thread: {}, items, nextPosition: null, hasMore: false }, +]; + +it.effect("--wait lasts until background work the agent waits on wakes it and that run ends", () => + Effect.gen(function* () { + const wakeRun = RunId.make("run-2"); + const { tools, calls } = scriptedTools({ + t3_thread_wait: [ + { threadId, runId, status: "completed", timedOut: false }, + { + threadId, + runId, + status: "completed", + timedOut: false, + resultRun: { runId: wakeRun, status: "completed" }, + }, + ], + // The wake run starts from a notification, which the messages view leaves out. + t3_thread_read: laterRunReads(wakeRun, []), + }); + + const outcome = yield* waitForRun(tools, threadId, runId, true); + + assert.deepStrictEqual( + calls.filter((entry) => entry.tool === "t3_thread_wait").map((entry) => entry.args), + [ + { threadId, runId, timeoutMs: 15_000 }, + { threadId, runId, timeoutMs: 15_000, untilDelegatedWorkSettled: true }, + ], + ); + assert.deepStrictEqual(outcome, { type: "ended", runId: wakeRun, status: "completed" }); + }), +); + +it.effect( + "--wait waits out the moment between background work ending and its wake run starting", + () => + Effect.gen(function* () { + const wakeRun = RunId.make("run-2"); + const ownOnly = { recentRuns: [{ runId, ordinal: 1, status: "completed" }] }; + const withWake = (status: string) => [ + { + recentRuns: [ + { runId, ordinal: 1, status: "completed" }, + { runId: wakeRun, ordinal: 2, status }, + ], + }, + { thread: {}, items: [], nextPosition: null, hasMore: false }, + ]; + const { tools } = scriptedTools({ + t3_thread_wait: [ + { threadId, runId, status: "completed", timedOut: false }, + { threadId, runId, status: "completed", timedOut: true }, + // The work ended, and the run it wakes is not recorded yet. + { + threadId, + runId, + status: "completed", + timedOut: false, + resultRun: { runId, status: "completed" }, + }, + { + threadId, + runId, + status: "completed", + timedOut: false, + resultRun: { runId: wakeRun, status: "completed" }, + }, + ], + t3_thread_read: [ + ownOnly, + { thread: { pendingRequestCount: 0 } }, + ownOnly, + ...withWake("running"), + ...withWake("completed"), + ], + }); + + const fiber = yield* Effect.forkChild(waitForRun(tools, threadId, runId, true)); + yield* TestClock.adjust("500 millis"); + + assert.deepStrictEqual(yield* Fiber.join(fiber), { + type: "ended", + runId: wakeRun, + status: "completed", + }); + }), +); + +it.effect( + "--wait keeps its own run's result when another client's later run settles the thread", + () => + Effect.gen(function* () { + const sentRun = RunId.make("run-2"); + const { tools } = scriptedTools({ + t3_thread_wait: [ + { threadId, runId, status: "completed", timedOut: false }, + { + threadId, + runId, + status: "completed", + timedOut: false, + resultRun: { runId: sentRun, status: "failed" }, + }, + ], + t3_thread_read: laterRunReads(sentRun, [{ runId: sentRun, type: "user_message" }]), + }); + + assert.deepStrictEqual(yield* waitForRun(tools, threadId, runId, true), { + type: "ended", + runId, + status: "completed", + }); + }), +); + it.effect("--wait stops when the thread waits for an answer instead of polling forever", () => Effect.gen(function* () { const { tools } = scriptedTools({ diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index 53b7df74be2d4..7e2959c767d04 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -9,6 +9,7 @@ import { ThreadId, TrimmedNonEmptyString, type OrchestratorMcpThreadListItem, + type OrchestratorMcpThreadRun, type OrchestratorMcpThreadStatus, type OrchestratorMcpThreadTimelineItem, } from "@t3tools/contracts"; @@ -39,6 +40,15 @@ import { projectLocationFlags, resolveCliAuthConfig } from "./config.ts"; * soon, since `t3_thread_wait` only returns on a terminal run. */ const WAIT_SLICE_MS = 15_000; +const WAKE_GRACE_MS = 10_000; +const WAKE_POLL_MS = 500; +const TERMINAL_RUN_STATUSES = new Set([ + "completed", + "failed", + "cancelled", + "interrupted", + "rolled_back", +]); const READ_PAGE_LIMIT = 100; const TITLE_MAX_CHARS = 80; @@ -498,35 +508,131 @@ type WaitOutcome = } | { readonly type: "needs_input" }; +/** + * Whether the thread has nothing left to do: no active run and no delegated + * or background work. Only then is the work a later wait settles on this + * send's own. + */ +const threadIsIdle = Effect.fn("cli.thread.threadIsIdle")(function* ( + tools: ThreadTools, + threadId: ThreadId, +) { + const waited = yield* tools.call( + "t3_thread_wait", + { threadId, timeoutMs: 1, untilDelegatedWorkSettled: true }, + OrchestratorMcpThreadWaitResult, + ); + return !waited.timedOut; +}); + +/** + * The runs after `runId`, and whether one holds a message someone sent. + * Background work wakes a run with a notification, never a user message, so + * such a run is another request and its reply is not this one's. + */ +const runsAfter = Effect.fn("cli.thread.runsAfter")(function* ( + tools: ThreadTools, + threadId: ThreadId, + runId: RunId, +) { + const { recentRuns } = yield* tools.call( + "t3_thread_read", + { threadId, limit: 1, runLimit: 50 }, + OrchestratorMcpThreadReadResult, + ); + const own = recentRuns.find((run) => run.runId === runId); + if (own === undefined) return { later: [], sent: true }; + const later = recentRuns.filter((run) => run.ordinal > own.ordinal); + if (later.length === 0) return { later, sent: false }; + const { items } = yield* readTimeline(tools, threadId, "messages"); + const sent = items.some( + (item) => item.type === "user_message" && later.some((run) => run.runId === item.runId), + ); + return { later, sent }; +}); + /** * Waits for this exact run, following a restart to its continuation, so a * message another client sends meanwhile cannot stand in for its result. + * With `followWakes`, a completed run is followed further to the runs its + * background work wakes, as long as no one else sends to the thread. */ export const waitForRun = Effect.fn("cli.thread.waitForRun")(function* ( tools: ThreadTools, threadId: ThreadId, runId: RunId, + followWakes = false, ) { - const waitSlice = (timeoutMs: number) => - tools.call("t3_thread_wait", { threadId, runId, timeoutMs }, OrchestratorMcpThreadWaitResult); + const waitSlice = (timeoutMs: number, untilDelegatedWorkSettled?: true) => + tools.call( + "t3_thread_wait", + { + threadId, + runId, + timeoutMs, + ...(untilDelegatedWorkSettled ? { untilDelegatedWorkSettled } : {}), + }, + OrchestratorMcpThreadWaitResult, + ); const ended = (waited: OrchestratorMcpThreadWaitResult) => ({ type: "ended", runId: waited.runId, status: waited.status }) satisfies WaitOutcome; + const ownRun = Effect.gen(function* () { + while (true) { + const waited = yield* waitSlice(WAIT_SLICE_MS); + if (!waited.timedOut) return ended(waited); + // A pending request on this run, or on an earlier run this one is queued + // behind, blocks it until someone answers. + const { thread } = yield* tools.call( + "t3_thread_read", + { threadId, limit: 1, runLimit: 1 }, + OrchestratorMcpThreadReadResult, + ); + if (thread.pendingRequestCount > 0) { + // The count is thread-wide: if this run ended since the slice, the + // request belongs to a later run. + const recheck = yield* waitSlice(1); + return recheck.timedOut ? ({ type: "needs_input" } satisfies WaitOutcome) : ended(recheck); + } + } + }); + const outcome = yield* ownRun; + if (!followWakes || outcome.type !== "ended" || outcome.status !== "completed") return outcome; + const finalRunId = outcome.runId ?? runId; + // A provider records the run that background work wakes only after that + // work has ended, so the thread briefly looks settled in between. + const wakeRunRecorded = Effect.gen(function* () { + for (let waited = 0; waited < WAKE_GRACE_MS; waited += WAKE_POLL_MS) { + yield* Effect.sleep(WAKE_POLL_MS); + const after = yield* runsAfter(tools, threadId, finalRunId); + if (after.sent) return false; + if (after.later.length > 0) return true; + } + return false; + }); + let workOutlivedRun = false; while (true) { - const waited = yield* waitSlice(WAIT_SLICE_MS); - if (!waited.timedOut) return ended(waited); - // A pending request on this run, or on an earlier run this one is queued - // behind, blocks it until someone answers. + const waited = yield* waitSlice(WAIT_SLICE_MS, true); + // `resultRun` is the thread's latest finished run, whoever sent it. + const result = waited.timedOut ? undefined : waited.resultRun; + const after = yield* runsAfter(tools, threadId, finalRunId); + if (after.sent) return outcome; + if (!waited.timedOut) { + if (result !== undefined && after.later.some((run) => run.runId === result.runId)) { + return { type: "ended", runId: result.runId, status: result.status } satisfies WaitOutcome; + } + if (!workOutlivedRun) return outcome; + if (after.later.some((run) => !TERMINAL_RUN_STATUSES.has(run.status))) continue; + if (after.later.length === 0 && (yield* wakeRunRecorded)) continue; + return outcome; + } + workOutlivedRun = true; const { thread } = yield* tools.call( "t3_thread_read", { threadId, limit: 1, runLimit: 1 }, OrchestratorMcpThreadReadResult, ); - if (thread.pendingRequestCount > 0) { - // The count is thread-wide: if this run ended since the slice, the - // request belongs to a later run. - const recheck = yield* waitSlice(1); - return recheck.timedOut ? ({ type: "needs_input" } satisfies WaitOutcome) : ended(recheck); - } + // Earlier runs have ended, so the request is on a run this one's work woke. + if (thread.pendingRequestCount > 0) return { type: "needs_input" } satisfies WaitOutcome; } }); @@ -540,9 +646,10 @@ const finishRun = Effect.fn("cli.thread.finishRun")(function* ( readonly threadId: ThreadId; readonly runId: RunId; readonly json: boolean; + readonly followWakes: boolean; }, ) { - const outcome = yield* waitForRun(tools, input.threadId, input.runId); + const outcome = yield* waitForRun(tools, input.threadId, input.runId, input.followWakes); const runId = outcome.type === "ended" ? (outcome.runId ?? input.runId) : input.runId; const status = outcome.type === "ended" ? outcome.status : "waiting"; const reply = @@ -727,6 +834,7 @@ const threadStartCommand = Command.make("start", { threadId: launched.threadId, runId: launched.runId, json: flags.json, + followWakes: true, }); } yield* flags.json ? printJson(launched) : Console.log(launched.threadId); @@ -753,13 +861,20 @@ const threadSendCommand = Command.make("send", { yield* withThreadTools( flags, Effect.fn("cli.thread.send")(function* (tools) { + // Work already running when this is sent is not this message's to wait for. + const followWakes = flags.wait ? yield* threadIsIdle(tools, threadId) : false; const sent = yield* tools.call( "t3_thread_send", { threadId, message, mode: "queue" }, OrchestratorMcpThreadSendResult, ); if (flags.wait) { - return yield* finishRun(tools, { threadId, runId: sent.runId, json: flags.json }); + return yield* finishRun(tools, { + threadId, + runId: sent.runId, + json: flags.json, + followWakes, + }); } yield* flags.json ? printJson(sent)