Skip to content

Commit 55ed569

Browse files
authored
Merge pull request #189 from jayleaton/fix/thread-wait-settled-20261104
fix(server): t3_thread_wait says why it is still waiting and survives a silent environment
2 parents 895fb16 + b91fe81 commit 55ed569

15 files changed

Lines changed: 251 additions & 44 deletions

‎apps/server/src/mcp/OrchestratorMcpService.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3349,6 +3349,7 @@ const make = Effect.gen(function* () {
33493349
: {
33503350
resultRun: { runId: result.resultRun.id, status: runOutcome(result.resultRun) },
33513351
}),
3352+
...(result.workState === undefined ? {} : { workState: result.workState }),
33523353
} satisfies OrchestratorMcpThreadWaitResult;
33533354
}),
33543355
interruptThread: Effect.fn("OrchestratorMcpService.interruptThread")(function* (scope, input) {

‎apps/server/src/mcp/agents/remoteThreads.test.ts‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,7 +127,11 @@ const makeMacBook = (
127127
createdAt: "2026-10-22T03:40:00.000Z",
128128
updatedAt: "2026-10-22T03:44:00.000Z",
129129
});
130+
let readsLeft = Infinity;
130131
const answer = (method: string, args: ReadonlyArray<any>): unknown => {
132+
if (readsLeft <= 0 || (method === "getThread" && readsLeft-- <= 0)) {
133+
throw new Error("The MacBook did not answer.");
134+
}
131135
switch (method) {
132136
case "listEnvironments":
133137
return [
@@ -174,6 +178,10 @@ const makeMacBook = (
174178
calls,
175179
answer,
176180
called: (method: string) => calls.filter((call) => call.method === method),
181+
/** Answers `reads` more thread reads, then nothing at all. */
182+
stopAnsweringAfter: (reads: number) => {
183+
readsLeft = reads;
184+
},
177185
};
178186
};
179187

@@ -508,6 +516,37 @@ it.effect("waiting on a remote child returns when its run completes", () => {
508516
}).pipe(Effect.provide(makeLayer(macbook, newRecord())));
509517
});
510518

519+
it.effect("a remote wait reports an environment that stops answering instead of failing", () => {
520+
const macbook = makeMacBook(["running"]);
521+
return Effect.gen(function* () {
522+
const clock = yield* makeSteppedClock;
523+
const wait = orchestrate("t3_thread_wait", {
524+
environmentId: macBook,
525+
threadId: child,
526+
timeoutMs: 60 * 60 * 1_000,
527+
}).pipe(Effect.provideService(Clock.Clock, clock.stepped));
528+
macbook.stopAnsweringAfter(1);
529+
const fiber = yield* Effect.forkChild(wait);
530+
// Each failed poll is tolerated until the last one.
531+
expect([yield* clock.step, yield* clock.step, yield* clock.step]).toEqual([
532+
1_000, 2_000, 4_000,
533+
]);
534+
expect((yield* Fiber.join(fiber)).result).toMatchObject({
535+
runId: "run-macbook-1",
536+
status: "running",
537+
timedOut: true,
538+
unreachable: expect.stringContaining("did not answer"),
539+
});
540+
// Nothing seen yet: the wait still answers, with no run.
541+
expect((yield* wait).result).toMatchObject({
542+
runId: null,
543+
status: "idle",
544+
timedOut: true,
545+
unreachable: expect.any(String),
546+
});
547+
}).pipe(Effect.provide(makeLayer(macbook, newRecord())));
548+
});
549+
511550
it.effect("reads and sends reach a remote child through its environmentId", () => {
512551
const macbook = makeMacBook();
513552
return Effect.gen(function* () {

‎apps/server/src/mcp/agents/remoteThreads.ts‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -301,22 +301,32 @@ const stillActive = (remote: RemoteEnvironment, threadId: ThreadId, runId: RunId
301301
* The relaying app answers requests but streams nothing back, so a wait on another environment
302302
* polls, backing off to every five seconds. Each poll reads the small thread list and fetches
303303
* the thread itself only once its run is no longer the active one. Without runId, a restart
304-
* moves the wait to the run that continues the work, as a local wait does.
304+
* moves the wait to the run that continues the work. A silent environment yields `unreachable`.
305305
*/
306306
export const waitForThread = Effect.fn("remoteThreads.waitForThread")(function* (
307307
remote: RemoteEnvironment,
308308
input: OrchestratorMcpThreadWaitInput,
309309
): Effect.fn.Return<OrchestratorMcpThreadWaitResult, OrchestratorMcpFailure> {
310310
const deadline = (yield* Clock.currentTimeMillis) + clampWaitTimeoutMs(input.timeoutMs);
311-
let thread = yield* getRemoteThread(remote, input.threadId);
312-
let runId = input.runId ?? thread.runs.at(-1)?.id;
311+
let runId = input.runId;
313312
const result = (status: OrchestratorMcpThreadWaitResult["status"], timedOut: boolean) => ({
314313
environmentId: remote.environmentId,
315314
threadId: input.threadId,
316315
runId: runId ?? null,
317316
status,
318317
timedOut,
319318
});
319+
const unreachable = (
320+
status: OrchestratorMcpThreadWaitResult["status"],
321+
failure: OrchestratorMcpFailure,
322+
) =>
323+
failure.code === "orchestration_error"
324+
? Effect.succeed({ ...result(status, true), unreachable: failure.message })
325+
: Effect.fail(failure);
326+
const first = yield* getRemoteThread(remote, input.threadId).pipe(Effect.result);
327+
if (first._tag === "Failure") return yield* unreachable("idle", first.failure);
328+
let thread = first.success;
329+
runId ??= thread.runs.at(-1)?.id;
320330
if (runId === undefined) return result("idle", false);
321331
let delay = FIRST_POLL;
322332
let failures = 0;
@@ -350,7 +360,7 @@ export const waitForThread = Effect.fn("remoteThreads.waitForThread")(function*
350360
thread = next.success;
351361
failures = 0;
352362
} else if (++failures >= RELAY_FAILURES_TOLERATED) {
353-
return yield* next.failure;
363+
return yield* unreachable(run.status, next.failure);
354364
}
355365
}
356366
});

‎apps/server/src/mcp/toolkits/orchestrator/tools.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -293,7 +293,7 @@ const ThreadSendTool = Tool.make("t3_thread_send", {
293293

294294
const ThreadWaitTool = Tool.make("t3_thread_wait", {
295295
description:
296-
"Wait for a T3 thread run to reach a terminal durable state. Pass environmentId for a thread on another environment. Without runId, the latest run at call time is selected; an idle thread returns immediately. A thread blocked on a question returns early with hasPendingQuestion and pendingRequestId for t3_pending_request, or on an approval with pendingApproval. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. Waiting reports status only and does not acknowledge a delegated result.",
296+
"Wait for a T3 thread run to reach a terminal durable state. Pass environmentId for a thread on another environment. Without runId, the latest run at call time is selected; an idle thread returns immediately. A thread blocked on a question returns early with hasPendingQuestion and pendingRequestId for t3_pending_request, or on an approval with pendingApproval. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. With untilDelegatedWorkSettled, a terminal status with timedOut=true means the run ended but the thread is not done; workState says why. Waiting reports status only and does not acknowledge a delegated result.",
297297
parameters: OrchestratorMcpThreadWaitInput,
298298
success: OrchestratorMcpThreadWaitResult,
299299
failure: OrchestratorMcpFailure,

‎apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3368,9 +3368,12 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => {
33683368
occurredAt: now,
33693369
payload: {
33703370
...providerThread,
3371-
pendingBackgroundTasks: open
3372-
? [{ taskId: "sleep-900", kind: "command" as const }]
3373-
: [],
3371+
pendingBackgroundTasks: [
3372+
...(open
3373+
? [{ taskId: "sleep-900", kind: "command" as const, awaited: true as const }]
3374+
: []),
3375+
{ taskId: "log-watcher", kind: "monitor" as const },
3376+
],
33743377
},
33753378
}
33763379
: {

‎apps/server/src/orchestration-v2/Orchestrator.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@ import {
8181
import { modelSelectionsEqual } from "@t3tools/shared/model";
8282
import { recordThreadOpen } from "@t3tools/shared/paletteSearch";
8383
import {
84+
backgroundWorkHoldsCompletion,
8485
derivePendingBackgroundWork,
8586
pendingBackgroundTurnItems,
8687
turnItemUpdateCanEndBackgroundWork,
@@ -14152,14 +14153,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1415214153

1415314154
// A child whose run ends while its background tasks or its own subagents still run has no
1415414155
// result yet. When that work settles there is no run update to finalize it, so these events
14155-
// do it. A background roster counts once it empties after it was seen non-empty.
14156+
// do it. A background roster counts once it stops holding after it was seen holding.
1415614157
const providerThreadsWithBackgroundWork = new Set<string>();
1415714158
const settlesChildWork = (stored: OrchestrationV2StoredEvent) => {
1415814159
const { event } = stored;
1415914160
if (event.type === "subagent.updated")
1416014161
return !isOrchestrationV2WorkActive(event.payload.status);
1416114162
if (event.type !== "provider-thread.updated") return false;
14162-
if ((event.payload.pendingBackgroundTasks?.length ?? 0) > 0) {
14163+
if (backgroundWorkHoldsCompletion(event.payload.pendingBackgroundTasks ?? [])) {
1416314164
providerThreadsWithBackgroundWork.add(event.payload.id);
1416414165
return false;
1416514166
}

‎apps/server/src/orchestration-v2/SubagentProjection.test.ts‎

Lines changed: 19 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -272,15 +272,25 @@ it("waits for nested work and retains the report across monitor acknowledgements
272272
"result_available",
273273
);
274274
}
275-
assert.equal(
276-
delegatedTaskProgress({
277-
...projection,
278-
providerThreads: [
279-
{ pendingBackgroundTasks: [{ taskId: "background-audit", kind: "command" }] },
280-
],
281-
}).state,
282-
"waiting_for_children",
283-
);
275+
for (const [awaited, state] of [
276+
[true, "waiting_for_children"],
277+
[false, "result_available"],
278+
] as const) {
279+
assert.equal(
280+
delegatedTaskProgress({
281+
...projection,
282+
providerThreads: [
283+
{
284+
pendingBackgroundTasks: [
285+
{ taskId: "background-audit", kind: "command", ...(awaited ? { awaited } : {}) },
286+
{ taskId: "log-watcher", kind: "monitor" },
287+
],
288+
},
289+
],
290+
}).state,
291+
state,
292+
);
293+
}
284294
const pending = {
285295
...run,
286296
id: RunId.make("followup"),

‎apps/server/src/orchestration-v2/ThreadManagementService.test.ts‎

Lines changed: 28 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -406,7 +406,7 @@ it.effect("waitForThread can wait past an ended run for the thread's delegated w
406406
const threadId = ThreadId.make("thread:thread-management:wait-delegated");
407407
const runId = RunId.make("run:thread-management:wait-delegated");
408408
const wakeRunId = RunId.make("run:thread-management:wait-delegated:wake");
409-
const subscribed = yield* Deferred.make<void>();
409+
let subscribed = yield* Deferred.make<void>();
410410
const events = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
411411
const at = (seconds: number) => DateTime.makeUnsafe(seconds * 1_000);
412412
const run = (id: RunId, ordinal: number, status: OrchestrationV2Run["status"], end: number) =>
@@ -464,6 +464,31 @@ it.effect("waitForThread can wait past an ended run for the thread's delegated w
464464
yield* service.waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 }),
465465
).not.toHaveProperty("resultRun");
466466

467+
// A wait that runs out says why the ended run is not the answer yet.
468+
const timedOut = Effect.fnUntraced(function* () {
469+
const waiting = yield* service
470+
.waitForThread({
471+
projectId,
472+
threadId,
473+
runId,
474+
timeoutMs: 1_000,
475+
untilDelegatedWorkSettled: true,
476+
})
477+
.pipe(Effect.forkChild);
478+
yield* TestClock.adjust(Duration.millis(1_000));
479+
return yield* Fiber.join(waiting);
480+
});
481+
expect(yield* timedOut()).toMatchObject({
482+
timedOut: true,
483+
run: { id: runId, status: "completed" },
484+
workState: "waiting_for_children",
485+
});
486+
// The task finishes; its delivery wakes the thread, whose next run answers.
487+
subagents = [{ status: "completed" }];
488+
runs = [...runs, run(wakeRunId, 2, "running", 3)];
489+
expect(yield* timedOut()).toMatchObject({ timedOut: true, workState: "working" });
490+
491+
subscribed = yield* Deferred.make<void>();
467492
const fiber = yield* service
468493
.waitForThread({
469494
projectId,
@@ -475,16 +500,15 @@ it.effect("waitForThread can wait past an ended run for the thread's delegated w
475500
.pipe(Effect.forkChild);
476501
// A wait that returns early fails below instead of hanging here.
477502
yield* Effect.raceFirst(Deferred.await(subscribed), Fiber.await(fiber));
478-
// The task finishes; its delivery wakes the thread, whose next run answers.
479-
subagents = [{ status: "completed" }];
480-
runs = [...runs, run(wakeRunId, 2, "completed", 3)];
503+
runs = [runs[0]!, run(wakeRunId, 2, "completed", 3)];
481504
yield* Queue.offer(events, stored(1, { type: "subagent.updated", payload: {} }));
482505
const result = yield* Fiber.join(fiber);
483506

484507
expect(result).toMatchObject({
485508
timedOut: false,
486509
run: { id: runId, status: "completed" },
487510
resultRun: { id: wakeRunId, status: "completed" },
511+
workState: "result_available",
488512
});
489513
}),
490514
);

‎apps/server/src/orchestration-v2/ThreadManagementService.ts‎

Lines changed: 19 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -197,6 +197,12 @@ export interface ThreadManagementWaitResult {
197197
readonly timedOut: boolean;
198198
/** With `untilDelegatedWorkSettled`, once settled: the run with the thread's final answer. */
199199
readonly resultRun?: OrchestrationV2Run;
200+
/**
201+
* With `untilDelegatedWorkSettled`, once the selected run ended: `working` while
202+
* another run continues the thread, `waiting_for_children` while its tasks or
203+
* awaited background work remain, `result_available` once settled.
204+
*/
205+
readonly workState?: ReturnType<typeof delegatedTaskProgress>["state"];
200206
/** With `untilQuestion`: the live question the thread is blocked on. */
201207
readonly pendingQuestion?: RuntimeRequestId;
202208
/** With `untilQuestion`: the oldest live approval the thread is blocked on. */
@@ -1117,13 +1123,18 @@ const make = Effect.gen(function* () {
11171123
progress.resultRun === undefined &&
11181124
records.runs.every((run) => isTerminalRunStatus(run.status))
11191125
) {
1120-
return latestRun(records);
1126+
return { workState: "result_available" as const, resultRun: latestRun(records) };
1127+
}
1128+
if (progress.state !== "result_available") {
1129+
return { workState: progress.state, resultRun: undefined };
11211130
}
1122-
if (progress.state !== "result_available") return undefined;
1131+
// A pending result is a restart continuation or delivery about to run.
11231132
const pending = yield* orchestrator
11241133
.delegatedTaskResultPending(input.threadId)
11251134
.pipe(Effect.mapError(loadError));
1126-
return pending ? undefined : progress.resultRun;
1135+
return pending
1136+
? { workState: "working" as const, resultRun: undefined }
1137+
: { workState: progress.state, resultRun: progress.resultRun };
11271138
});
11281139
const untilQuestion = input.untilQuestion === true;
11291140
const blockingRequests = getProjectThreadRecords(input, ["runtimeRequests"]).pipe(
@@ -1153,13 +1164,13 @@ const make = Effect.gen(function* () {
11531164
const current =
11541165
outcomeNow.settled && untilDelegatedWorkSettled
11551166
? yield* delegatedWorkResult.pipe(
1156-
Effect.map((resultRun) => ({
1167+
Effect.map((delegated) => ({
11571168
...outcomeNow,
1158-
settled: resultRun !== undefined,
1159-
resultRun,
1169+
...delegated,
1170+
settled: delegated.resultRun !== undefined,
11601171
})),
11611172
)
1162-
: { ...outcomeNow, resultRun: undefined };
1173+
: { ...outcomeNow, resultRun: undefined, workState: undefined };
11631174
const blocked =
11641175
current.settled || !untilQuestion
11651176
? { pendingQuestion: undefined, pendingApproval: undefined }
@@ -1178,6 +1189,7 @@ const make = Effect.gen(function* () {
11781189
run: current.run,
11791190
timedOut: !current.done,
11801191
...(current.resultRun === undefined ? {} : { resultRun: current.resultRun }),
1192+
...(current.workState === undefined ? {} : { workState: current.workState }),
11811193
...(current.pendingQuestion === undefined
11821194
? {}
11831195
: { pendingQuestion: current.pendingQuestion }),

‎apps/server/src/peer/PeerForwarding.test.ts‎

Lines changed: 48 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,8 @@ interface Seen {
6262
readonly clones: ReadonlyArray<SourceControlCloneRepositoryInput>;
6363
/** Commands dispatched and delegated-task stops, as `<type> <threadId>`. */
6464
readonly stops: ReadonlyArray<string>;
65+
/** Waits, counted from 1, that fail as a peer that stops answering would. */
66+
readonly droppedWaits?: ReadonlyArray<number>;
6567
}
6668
const nothingSeen: Seen = {
6769
sends: [],
@@ -137,11 +139,21 @@ const boxThreads = (seen: Ref.Ref<Seen>, linkSession: Ref.Ref<string>) => {
137139
})),
138140
waitForThread: (input) =>
139141
Ref.updateAndGet(seen, (current) => ({ ...current, waits: [...current.waits, input] })).pipe(
140-
Effect.map(({ waits }) => ({
141-
threadId: input.threadId,
142-
run: run(waits.length < 3 ? "running" : "completed"),
143-
timedOut: waits.length < 3,
144-
})),
142+
Effect.flatMap(({ waits, droppedWaits }) =>
143+
droppedWaits?.includes(waits.length)
144+
? Effect.fail(
145+
new ThreadManagement.ThreadManagementProjectionLoadError({
146+
projectId: boxProject,
147+
threadId: input.threadId,
148+
cause: "dropped",
149+
}),
150+
)
151+
: Effect.succeed({
152+
threadId: input.threadId,
153+
run: run(waits.length < 3 ? "running" : "completed"),
154+
timedOut: waits.length < 3,
155+
}),
156+
),
145157
),
146158
});
147159
};
@@ -549,6 +561,37 @@ it.effect("keeps a forwarded wait short and its retries scoped to the caller", (
549561
).pipe(Effect.provide(NodeHttpServer.layerTest)),
550562
);
551563

564+
it.effect("reports a linked environment that stops answering mid-wait instead of failing", () =>
565+
Effect.scoped(
566+
Effect.gen(function* () {
567+
const seen = yield* Ref.make<Seen>({ ...nothingSeen, droppedWaits: [2] });
568+
const linkSession = yield* Ref.make("");
569+
const b = yield* serveBox(seen, linkSession);
570+
const a = yield* makeLaptop;
571+
yield* linkTo(a.links, b, "auto");
572+
yield* Ref.set(linkSession, (yield* b.linkedSessions)[0]!.sessionId);
573+
const wait = a.call("thread:laptop-full", "t3_thread_wait", {
574+
environmentId: box.environmentId,
575+
threadId: boxThread,
576+
timeoutMs: 600_000,
577+
});
578+
579+
const dropped = yield* wait;
580+
expect(dropped.isError).toBe(false);
581+
expect(dropped.structuredContent).toMatchObject({
582+
runId: boxRun,
583+
status: "running",
584+
timedOut: true,
585+
unreachable: expect.any(String),
586+
});
587+
expect((yield* wait).structuredContent).toMatchObject({
588+
status: "completed",
589+
timedOut: false,
590+
});
591+
}),
592+
).pipe(Effect.provide(NodeHttpServer.layerTest)),
593+
);
594+
552595
it.effect("launches in a linked environment with the caller's modes, retry-safe", () =>
553596
Effect.scoped(
554597
Effect.gen(function* () {

0 commit comments

Comments
 (0)