From f872efb9eee75fec731e119eef1a81519c4f8844 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 01:59:44 +0000 Subject: [PATCH 1/5] test(cli): thread --wait waits for the run background work wakes Fails on main: waitForRun never passes untilDelegatedWorkSettled to t3_thread_wait, so --wait ends with the run that called t3_background_wait. Repro for #34688. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/cli/thread.test.ts | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index 5ab4e24d0a8e1..52dc1ed5dcbd2 100644 --- a/apps/server/src/cli/thread.test.ts +++ b/apps/server/src/cli/thread.test.ts @@ -210,6 +210,31 @@ it.effect("--wait follows the sent run to its continuation's terminal status", ( }), ); +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, + resultRun: { runId: wakeRun, status: "completed" }, + }, + ], + }); + + const outcome = yield* waitForRun(tools, threadId, runId); + + assert.deepStrictEqual( + calls.map((entry) => entry.args), + [{ threadId, runId, timeoutMs: 15_000, untilDelegatedWorkSettled: true }], + ); + assert.deepStrictEqual(outcome, { type: "ended", runId: wakeRun, status: "completed" }); + }), +); + it.effect("--wait stops when the thread waits for an answer instead of polling forever", () => Effect.gen(function* () { const { tools } = scriptedTools({ From 16c4739c10911ce00ed171351e92130f76fbae04 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 02:05:25 +0000 Subject: [PATCH 2/5] fix(cli): thread --wait returns once background work the agent waits on has ended waitForRun never asked t3_thread_wait to wait for delegated and background work, so `t3 thread start --wait` and `t3 thread send --wait` returned after the first run ended and printed its interim reply. It now passes untilDelegatedWorkSettled and reports resultRun, the run with the thread's final answer, for the reply and exit code. Fixes pingdotgg/t3code#34688 AI-assisted; verified with the tests and tools listed above. --- apps/server/src/cli/thread.test.ts | 4 ++-- apps/server/src/cli/thread.ts | 17 +++++++++++++---- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index 52dc1ed5dcbd2..d2cf502f63b9d 100644 --- a/apps/server/src/cli/thread.test.ts +++ b/apps/server/src/cli/thread.test.ts @@ -203,8 +203,8 @@ it.effect("--wait follows the sent run to its continuation's terminal status", ( assert.deepStrictEqual( calls.filter((entry) => entry.tool === "t3_thread_wait").map((entry) => entry.args), [ - { threadId, runId, timeoutMs: 15_000 }, - { threadId, runId, timeoutMs: 15_000 }, + { threadId, runId, timeoutMs: 15_000, untilDelegatedWorkSettled: true }, + { threadId, runId, timeoutMs: 15_000, untilDelegatedWorkSettled: true }, ], ); }), diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index 53b7df74be2d4..ae05db9c4d4f9 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -499,8 +499,9 @@ type WaitOutcome = | { readonly type: "needs_input" }; /** - * Waits for this exact run, following a restart to its continuation, so a - * message another client sends meanwhile cannot stand in for its result. + * Waits for this exact run, following a restart to its continuation and the + * runs its background work wakes, so a message another client sends meanwhile + * cannot stand in for its result. */ export const waitForRun = Effect.fn("cli.thread.waitForRun")(function* ( tools: ThreadTools, @@ -508,9 +509,17 @@ export const waitForRun = Effect.fn("cli.thread.waitForRun")(function* ( runId: RunId, ) { const waitSlice = (timeoutMs: number) => - tools.call("t3_thread_wait", { threadId, runId, timeoutMs }, OrchestratorMcpThreadWaitResult); + tools.call( + "t3_thread_wait", + { threadId, runId, timeoutMs, untilDelegatedWorkSettled: true }, + OrchestratorMcpThreadWaitResult, + ); const ended = (waited: OrchestratorMcpThreadWaitResult) => - ({ type: "ended", runId: waited.runId, status: waited.status }) satisfies WaitOutcome; + ({ + type: "ended", + runId: waited.resultRun?.runId ?? waited.runId, + status: waited.resultRun?.status ?? waited.status, + }) satisfies WaitOutcome; while (true) { const waited = yield* waitSlice(WAIT_SLICE_MS); if (!waited.timedOut) return ended(waited); From b752e2fd20da2b0fdb297a134d7df1e5e23a08e6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 02:17:00 +0000 Subject: [PATCH 3/5] fix(cli): thread --wait reports another client's later run as before untilDelegatedWorkSettled settles thread-wide and its resultRun is the thread's latest finished run, whoever sent it. --wait now takes resultRun only when every run after its own was woken by a notification; once a run after its own holds a sent message, it stops waiting and reports its own run, as before, including when that later run waits for an answer. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/cli/thread.test.ts | 53 +++++++++++++++++++++-- apps/server/src/cli/thread.ts | 69 ++++++++++++++++++++++++++---- 2 files changed, 111 insertions(+), 11 deletions(-) diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index d2cf502f63b9d..e66f8ca9cffb4 100644 --- a/apps/server/src/cli/thread.test.ts +++ b/apps/server/src/cli/thread.test.ts @@ -186,6 +186,20 @@ const scriptedTools = (script: Record>) => { return { tools: { call } satisfies ThreadTools, calls }; }; +/** 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 follows the sent run to its continuation's terminal status", () => Effect.gen(function* () { const continuation = RunId.make("run-2"); @@ -223,18 +237,46 @@ it.effect("--wait lasts until background work the agent waits on wakes it and th 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); assert.deepStrictEqual( - calls.map((entry) => entry.args), + calls.filter((entry) => entry.tool === "t3_thread_wait").map((entry) => entry.args), [{ threadId, runId, timeoutMs: 15_000, untilDelegatedWorkSettled: true }], ); assert.deepStrictEqual(outcome, { 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, + resultRun: { runId: sentRun, status: "failed" }, + }, + ], + t3_thread_read: laterRunReads(sentRun, [{ runId: sentRun, type: "user_message" }]), + }); + + assert.deepStrictEqual(yield* waitForRun(tools, threadId, runId), { + 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({ @@ -251,12 +293,17 @@ it.effect("--wait stops when the thread waits for an answer instead of polling f it.effect("--wait reports its run's result when a later run is the one waiting", () => Effect.gen(function* () { + const sentRun = RunId.make("run-2"); const { tools } = scriptedTools({ + // The later run's request keeps the thread-wide wait from settling. t3_thread_wait: [ { threadId, runId, status: "running", timedOut: true }, - { threadId, runId, status: "completed", timedOut: false }, + { threadId, runId, status: "completed", timedOut: true }, + ], + t3_thread_read: [ + { thread: { pendingRequestCount: 1 } }, + ...laterRunReads(sentRun, [{ runId: sentRun, type: "user_message" }]), ], - t3_thread_read: [{ thread: { pendingRequestCount: 1 } }], }); assert.deepStrictEqual(yield* waitForRun(tools, threadId, runId), { diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index ae05db9c4d4f9..1cee102404bc2 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -498,6 +498,41 @@ type WaitOutcome = } | { readonly type: "needs_input" }; +/** + * Whether a run after `runId` 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 sentRunAfter = Effect.fn("cli.thread.sentRunAfter")(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 true; + const later = new Set( + recentRuns.filter((run) => run.ordinal > own.ordinal).map((run) => run.runId), + ); + if (later.size === 0) return false; + const { items } = yield* readTimeline(tools, threadId, "messages"); + return items.some( + (item) => item.type === "user_message" && item.runId !== null && later.has(item.runId), + ); +}); + +const TERMINAL_STATUSES: ReadonlyArray = [ + "completed", + "failed", + "cancelled", + "interrupted", + "rolled_back", +]; + /** * Waits for this exact run, following a restart to its continuation and the * runs its background work wakes, so a message another client sends meanwhile @@ -514,15 +549,30 @@ export const waitForRun = Effect.fn("cli.thread.waitForRun")(function* ( { threadId, runId, timeoutMs, untilDelegatedWorkSettled: true }, OrchestratorMcpThreadWaitResult, ); - const ended = (waited: OrchestratorMcpThreadWaitResult) => - ({ - type: "ended", - runId: waited.resultRun?.runId ?? waited.runId, - status: waited.resultRun?.status ?? waited.status, - }) satisfies WaitOutcome; + const own = (waited: OrchestratorMcpThreadWaitResult) => + ({ type: "ended", runId: waited.runId, status: waited.status }) satisfies WaitOutcome; + // `resultRun` is the thread's latest finished run, whoever sent it. + const ended = Effect.fnUntraced(function* (waited: OrchestratorMcpThreadWaitResult) { + const result = waited.resultRun; + if ( + result === undefined || + waited.runId === null || + result.runId === waited.runId || + (yield* sentRunAfter(tools, threadId, waited.runId)) + ) { + return own(waited); + } + return { type: "ended", runId: result.runId, status: result.status } satisfies WaitOutcome; + }); + // The thread-wide wait also outlasts runs other clients sent after this one. + const othersHoldThread = (waited: OrchestratorMcpThreadWaitResult) => + waited.runId !== null && TERMINAL_STATUSES.includes(waited.status) + ? sentRunAfter(tools, threadId, waited.runId) + : Effect.succeed(false); while (true) { const waited = yield* waitSlice(WAIT_SLICE_MS); - if (!waited.timedOut) return ended(waited); + if (!waited.timedOut) return yield* ended(waited); + if (yield* othersHoldThread(waited)) return own(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( @@ -534,7 +584,10 @@ export const waitForRun = Effect.fn("cli.thread.waitForRun")(function* ( // 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); + if (!recheck.timedOut) return yield* ended(recheck); + return (yield* othersHoldThread(recheck)) + ? own(recheck) + : ({ type: "needs_input" } satisfies WaitOutcome); } } }); From 67c5386a8bbfd6c23e96bf4b07217406ee426c67 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 02:27:17 +0000 Subject: [PATCH 4/5] fix(cli): thread --wait follows background wakes only after its own run completes The thread-wide untilDelegatedWorkSettled wait counted runs and delegated work that were not this request's: an earlier run's question, work running before the send, or an earlier run picked as resultRun. --wait now waits for its own run exactly as before, and only once that run completes, and only if the thread was idle when it was sent, keeps waiting thread-wide for the runs its background work wakes. It takes resultRun only when that run comes after its own and no one else sent to the thread in between. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/cli/thread.test.ts | 52 +++++------ apps/server/src/cli/thread.ts | 134 +++++++++++++++++------------ 2 files changed, 107 insertions(+), 79 deletions(-) diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index e66f8ca9cffb4..2d94228c97229 100644 --- a/apps/server/src/cli/thread.test.ts +++ b/apps/server/src/cli/thread.test.ts @@ -186,20 +186,6 @@ const scriptedTools = (script: Record>) => { return { tools: { call } satisfies ThreadTools, calls }; }; -/** 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 follows the sent run to its continuation's terminal status", () => Effect.gen(function* () { const continuation = RunId.make("run-2"); @@ -217,18 +203,33 @@ it.effect("--wait follows the sent run to its continuation's terminal status", ( assert.deepStrictEqual( calls.filter((entry) => entry.tool === "t3_thread_wait").map((entry) => entry.args), [ - { threadId, runId, timeoutMs: 15_000, untilDelegatedWorkSettled: true }, - { threadId, runId, timeoutMs: 15_000, untilDelegatedWorkSettled: true }, + { threadId, runId, timeoutMs: 15_000 }, + { threadId, runId, timeoutMs: 15_000 }, ], ); }), ); +/** 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, @@ -241,11 +242,14 @@ it.effect("--wait lasts until background work the agent waits on wakes it and th t3_thread_read: laterRunReads(wakeRun, []), }); - const outcome = yield* waitForRun(tools, threadId, runId); + 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, untilDelegatedWorkSettled: true }], + [ + { threadId, runId, timeoutMs: 15_000 }, + { threadId, runId, timeoutMs: 15_000, untilDelegatedWorkSettled: true }, + ], ); assert.deepStrictEqual(outcome, { type: "ended", runId: wakeRun, status: "completed" }); }), @@ -258,6 +262,7 @@ it.effect( const sentRun = RunId.make("run-2"); const { tools } = scriptedTools({ t3_thread_wait: [ + { threadId, runId, status: "completed", timedOut: false }, { threadId, runId, @@ -269,7 +274,7 @@ it.effect( t3_thread_read: laterRunReads(sentRun, [{ runId: sentRun, type: "user_message" }]), }); - assert.deepStrictEqual(yield* waitForRun(tools, threadId, runId), { + assert.deepStrictEqual(yield* waitForRun(tools, threadId, runId, true), { type: "ended", runId, status: "completed", @@ -293,17 +298,12 @@ it.effect("--wait stops when the thread waits for an answer instead of polling f it.effect("--wait reports its run's result when a later run is the one waiting", () => Effect.gen(function* () { - const sentRun = RunId.make("run-2"); const { tools } = scriptedTools({ - // The later run's request keeps the thread-wide wait from settling. t3_thread_wait: [ { threadId, runId, status: "running", timedOut: true }, - { threadId, runId, status: "completed", timedOut: true }, - ], - t3_thread_read: [ - { thread: { pendingRequestCount: 1 } }, - ...laterRunReads(sentRun, [{ runId: sentRun, type: "user_message" }]), + { threadId, runId, status: "completed", timedOut: false }, ], + t3_thread_read: [{ thread: { pendingRequestCount: 1 } }], }); assert.deepStrictEqual(yield* waitForRun(tools, threadId, runId), { diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index 1cee102404bc2..225c9b18e4689 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -499,11 +499,28 @@ type WaitOutcome = | { readonly type: "needs_input" }; /** - * Whether a run after `runId` 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. + * 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 sentRunAfter = Effect.fn("cli.thread.sentRunAfter")(function* ( +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, @@ -514,81 +531,83 @@ const sentRunAfter = Effect.fn("cli.thread.sentRunAfter")(function* ( OrchestratorMcpThreadReadResult, ); const own = recentRuns.find((run) => run.runId === runId); - if (own === undefined) return true; + if (own === undefined) return { later: new Set(), sent: true }; const later = new Set( recentRuns.filter((run) => run.ordinal > own.ordinal).map((run) => run.runId), ); - if (later.size === 0) return false; + if (later.size === 0) return { later, sent: false }; const { items } = yield* readTimeline(tools, threadId, "messages"); - return items.some( + const sent = items.some( (item) => item.type === "user_message" && item.runId !== null && later.has(item.runId), ); + return { later, sent }; }); -const TERMINAL_STATUSES: ReadonlyArray = [ - "completed", - "failed", - "cancelled", - "interrupted", - "rolled_back", -]; - /** - * Waits for this exact run, following a restart to its continuation and the - * runs its background work wakes, so a message another client sends meanwhile - * cannot stand in for its result. + * 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) => + const waitSlice = (timeoutMs: number, untilDelegatedWorkSettled?: true) => tools.call( "t3_thread_wait", - { threadId, runId, timeoutMs, untilDelegatedWorkSettled: true }, + { + threadId, + runId, + timeoutMs, + ...(untilDelegatedWorkSettled ? { untilDelegatedWorkSettled } : {}), + }, OrchestratorMcpThreadWaitResult, ); - const own = (waited: OrchestratorMcpThreadWaitResult) => + const ended = (waited: OrchestratorMcpThreadWaitResult) => ({ type: "ended", runId: waited.runId, status: waited.status }) satisfies WaitOutcome; - // `resultRun` is the thread's latest finished run, whoever sent it. - const ended = Effect.fnUntraced(function* (waited: OrchestratorMcpThreadWaitResult) { - const result = waited.resultRun; - if ( - result === undefined || - waited.runId === null || - result.runId === waited.runId || - (yield* sentRunAfter(tools, threadId, waited.runId)) - ) { - return own(waited); + 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); + } } - return { type: "ended", runId: result.runId, status: result.status } satisfies WaitOutcome; }); - // The thread-wide wait also outlasts runs other clients sent after this one. - const othersHoldThread = (waited: OrchestratorMcpThreadWaitResult) => - waited.runId !== null && TERMINAL_STATUSES.includes(waited.status) - ? sentRunAfter(tools, threadId, waited.runId) - : Effect.succeed(false); + const outcome = yield* ownRun; + if (!followWakes || outcome.type !== "ended" || outcome.status !== "completed") return outcome; + const finalRunId = outcome.runId ?? runId; while (true) { - const waited = yield* waitSlice(WAIT_SLICE_MS); - if (!waited.timedOut) return yield* ended(waited); - if (yield* othersHoldThread(waited)) return own(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) { + return result !== undefined && after.later.has(result.runId) + ? ({ type: "ended", runId: result.runId, status: result.status } satisfies WaitOutcome) + : outcome; + } 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); - if (!recheck.timedOut) return yield* ended(recheck); - return (yield* othersHoldThread(recheck)) - ? own(recheck) - : ({ type: "needs_input" } satisfies WaitOutcome); - } + // 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; } }); @@ -602,9 +621,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 = @@ -789,6 +809,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); @@ -815,13 +836,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) From 5819dda70110d8f168d591007c0dc6bc5ebbe563 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 02:52:56 +0000 Subject: [PATCH 5/5] fix(cli): thread --wait waits for the wake run that background work starts A provider records the run that background work wakes a moment after that work ends, and the thread-wide wait settles in between, so --wait still reported the run that called t3_background_wait. When work outlived the CLI's run and the wait settles with no run after it, --wait now looks for the wake run for up to 10 seconds and keeps waiting once it appears; a later run that is still active is waited for too. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/cli/thread.test.ts | 57 ++++++++++++++++++++++++++++++ apps/server/src/cli/thread.ts | 43 +++++++++++++++++----- 2 files changed, 91 insertions(+), 9 deletions(-) diff --git a/apps/server/src/cli/thread.test.ts b/apps/server/src/cli/thread.test.ts index 2d94228c97229..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"; @@ -255,6 +257,61 @@ it.effect("--wait lasts until background work the agent waits on wakes it and th }), ); +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", () => diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index 225c9b18e4689..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; @@ -531,14 +541,12 @@ const runsAfter = Effect.fn("cli.thread.runsAfter")(function* ( OrchestratorMcpThreadReadResult, ); const own = recentRuns.find((run) => run.runId === runId); - if (own === undefined) return { later: new Set(), sent: true }; - const later = new Set( - recentRuns.filter((run) => run.ordinal > own.ordinal).map((run) => run.runId), - ); - if (later.size === 0) return { later, sent: false }; + 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" && item.runId !== null && later.has(item.runId), + (item) => item.type === "user_message" && later.some((run) => run.runId === item.runId), ); return { later, sent }; }); @@ -590,6 +598,18 @@ export const waitForRun = Effect.fn("cli.thread.waitForRun")(function* ( 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, true); // `resultRun` is the thread's latest finished run, whoever sent it. @@ -597,10 +617,15 @@ export const waitForRun = Effect.fn("cli.thread.waitForRun")(function* ( const after = yield* runsAfter(tools, threadId, finalRunId); if (after.sent) return outcome; if (!waited.timedOut) { - return result !== undefined && after.later.has(result.runId) - ? ({ type: "ended", runId: result.runId, status: result.status } satisfies WaitOutcome) - : outcome; + 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 },