authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-08 02:48:20-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-08 02:48:20-07:00
log37eaf5d293ef83ce86d2ce9a1583792696c82bbc
tree30bc8d0e9da20084bc3325d6636867bf34de84f0
parent85975585579188be20f46d96b4972ce46c33f4bc
signature Signed by PGP key B5690EEEBB952194

fix(server): an agent can settle its own thread when its turn ends (#17145)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

7 files changed, 187 insertions(+), 14 deletions(-)

apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts+7-11
...@@ -767,17 +767,13 @@ describe("orchestrator MCP toolkit", () => {...@@ -767,17 +767,13 @@ describe("orchestrator MCP toolkit", () => {
767 const invoke = (name: string, args: Record<string, unknown>) =>767 const invoke = (name: string, args: Record<string, unknown>) =>
768 invokeAs(invocation, name, args);768 invokeAs(invocation, name, args);
769769
770 const refusedSettle = yield* invoke("t3_thread_organize", { action: "settle" });770 // Settling would stop the session, so the agent's own turn keeps running.
771 expect(refusedSettle.isError).toBe(true);771 const deferredSettle = yield* invoke("t3_thread_organize", { action: "settle" });
772 expect(refusedSettle.structuredContent).toBeUndefined();772 expect(deferredSettle.isError).toBe(false);
773 expect(declaredFailure(refusedSettle)).toEqual({773 expect(deferredSettle.structuredContent).toEqual({ settlesWhenTurnEnds: true });
774 _tag: "OrchestratorMcpFailure",774 const afterDeferredSettle = yield* orchestrator.getThreadProjection(parentThreadId);
775 code: "orchestration_error",775 expect(afterDeferredSettle.thread.settledOverride).not.toBe("settled");
776 message: `Thread ${parentThreadId} has active or blocked work and cannot be settled.`,776 expect(afterDeferredSettle.runs.find((run) => run.id === parentRun?.id)?.status).toBe(
777 });
778 const afterRefusedSettle = yield* orchestrator.getThreadProjection(parentThreadId);
779 expect(afterRefusedSettle.thread.settledOverride).not.toBe("settled");
780 expect(afterRefusedSettle.runs.find((run) => run.id === parentRun?.id)?.status).toBe(
781 "running",777 "running",
782 );778 );
783779
apps/server/src/mcp/toolkits/thread/handlers.ts+6-1
...@@ -292,8 +292,13 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, {...@@ -292,8 +292,13 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, {
292 ),292 ),
293 t3_thread_organize: writesThread((input) =>293 t3_thread_organize: writesThread((input) =>
294 Effect.gen(function* () {294 Effect.gen(function* () {
295 const { threads, projection } = yield* readThread(input.threadId);295 const { threads, projection, caller } = yield* readThread(input.threadId);
296 const common = { commandId: yield* newCommandId(), threadId: projection.thread.id };296 const common = { commandId: yield* newCommandId(), threadId: projection.thread.id };
297 if (input.action === "settle") {
298 return yield* threads
299 .settleThread({ ...common, byOwnAgent: caller?.id === projection.thread.id })
300 .pipe(Effect.mapError(dispatchFailure));
301 }
297 let command: OrchestrationV2Command;302 let command: OrchestrationV2Command;
298 switch (input.action) {303 switch (input.action) {
299 case "snooze":304 case "snooze":
apps/server/src/mcp/toolkits/thread/tools.ts+5-2
...@@ -30,7 +30,7 @@ import * as McpInvocationContext from "../../McpInvocationContext.ts";...@@ -30,7 +30,7 @@ import * as McpInvocationContext from "../../McpInvocationContext.ts";
3030
31const ThreadOrganizeTool = Tool.make("t3_thread_organize", {31const ThreadOrganizeTool = Tool.make("t3_thread_organize", {
32 description:32 description:
33 "Pin, snooze, settle, archive, or mark a thread unread. Omit threadId for this thread. snooze requires snoozedUntil. Existing thread lifecycle rules apply; this does not schedule a future action.",33 "Pin, snooze, settle, archive, or mark a thread unread. Omit threadId for this thread. snooze requires snoozedUntil. Existing thread lifecycle rules apply. Settling this thread takes effect when your turn completes, returning settlesWhenTurnEnds=true; a turn that fails or is interrupted, or a queued message, leaves it active.",
34 parameters: Schema.Struct({34 parameters: Schema.Struct({
35 threadId: Schema.optional(ThreadId),35 threadId: Schema.optional(ThreadId),
36 action: Schema.Literals([36 action: Schema.Literals([
...@@ -46,7 +46,10 @@ const ThreadOrganizeTool = Tool.make("t3_thread_organize", {...@@ -46,7 +46,10 @@ const ThreadOrganizeTool = Tool.make("t3_thread_organize", {
46 ]),46 ]),
47 snoozedUntil: Schema.optional(IsoDateTime),47 snoozedUntil: Schema.optional(IsoDateTime),
48 }),48 }),
49 success: OrchestrationV2DispatchCommandResult,49 success: Schema.Union([
50 OrchestrationV2DispatchCommandResult,
51 Schema.Struct({ settlesWhenTurnEnds: Schema.Literal(true) }),
52 ]),
50 failure: OrchestratorMcpFailure,53 failure: OrchestratorMcpFailure,
51 failureMode: "return" as const,54 failureMode: "return" as const,
52 dependencies: [55 dependencies: [
apps/server/src/orchestration-v2/ThreadManagementService.test.ts+37
...@@ -473,3 +473,40 @@ it.effect("waitForThread reads the run again only when the run updates", () =>...@@ -473,3 +473,40 @@ it.effect("waitForThread reads the run again only when the run updates", () =>
473 expect(reads).toBe(2);473 expect(reads).toBe(2);
474 }),474 }),
475);475);
476
477it.effect.each([
478 { status: "completed" as const, settles: true },
479 { status: "failed" as const, settles: false },
480 { status: "interrupted" as const, settles: false },
481])("settleAfterRun settles only when the run $status", ({ status, settles }) =>
482 Effect.gen(function* () {
483 const projectId = ProjectId.make("project:thread-management:settle-after-run");
484 const threadId = ThreadId.make("thread:thread-management:settle-after-run");
485 const runId = RunId.make("run:thread-management:settle-after-run");
486 const dispatched: Array<string> = [];
487 const layerTest = ThreadManagementService.layer.pipe(
488 Layer.provide(
489 Layer.mock(Orchestrator.OrchestratorV2)({
490 getThreadEventSequence: () => Effect.succeed(0),
491 getThreadRecords: () =>
492 Effect.succeed({
493 thread: { id: threadId, projectId, deletedAt: null },
494 runs: [{ id: runId, status }],
495 } as unknown as OrchestrationV2ThreadProjection),
496 dispatch: (command) =>
497 Effect.sync(() => {
498 dispatched.push(command.type);
499 return { sequence: 1, storedEvents: [] } as never;
500 }),
501 }),
502 ),
503 );
504 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
505 Effect.provide(layerTest),
506 );
507
508 yield* service.settleAfterRun({ projectId, threadId, runId });
509
510 expect(dispatched).toEqual(settles ? ["thread.settle"] : []);
511 }),
512);
apps/server/src/orchestration-v2/ThreadManagementService.ts+62
...@@ -326,6 +326,30 @@ export interface ThreadManagementServiceShape {...@@ -326,6 +326,30 @@ export interface ThreadManagementServiceShape {
326 readonly waitForThread: (326 readonly waitForThread: (
327 input: ThreadManagementWaitInput,327 input: ThreadManagementWaitInput,
328 ) => Effect.Effect<ThreadManagementWaitResult, ThreadManagementError>;328 ) => Effect.Effect<ThreadManagementWaitResult, ThreadManagementError>;
329 /**
330 * Waits for `runId` to end, then settles the thread if the run completed.
331 * Settling stops the provider session, so an agent that settles its own
332 * thread uses this to wait for its turn to end. A run that ends any other
333 * way leaves the thread active, as does anything `thread.settle` refuses
334 * then, such as a queued message.
335 */
336 readonly settleAfterRun: (input: {
337 readonly projectId: ProjectId;
338 readonly threadId: ThreadId;
339 readonly runId: RunId;
340 }) => Effect.Effect<void, ThreadManagementFailure>;
341 /**
342 * Settles a thread. When the thread's own agent asks, it is mid-turn, so
343 * the thread settles through `settleAfterRun` once that turn ends.
344 */
345 readonly settleThread: (input: {
346 readonly threadId: ThreadId;
347 readonly commandId: CommandId;
348 readonly byOwnAgent: boolean;
349 }) => Effect.Effect<
350 { readonly sequence: number } | { readonly settlesWhenTurnEnds: true },
351 Orchestrator.OrchestratorV2Error
352 >;
329 readonly interruptThread: (353 readonly interruptThread: (
330 input: ThreadManagementInterruptInput,354 input: ThreadManagementInterruptInput,
331 ) => Effect.Effect<ThreadManagementInterruptResult, ThreadManagementFailure>;355 ) => Effect.Effect<ThreadManagementInterruptResult, ThreadManagementFailure>;
...@@ -402,8 +426,11 @@ function latestSteerableRun(...@@ -402,8 +426,11 @@ function latestSteerableRun(
402 .toSorted((left, right) => right.ordinal - left.ordinal)[0];426 .toSorted((left, right) => right.ordinal - left.ordinal)[0];
403}427}
404428
429const SETTLE_AFTER_RUN_WAIT_MS = 24 * 60 * 60 * 1_000;
430
405const make = Effect.gen(function* () {431const make = Effect.gen(function* () {
406 const orchestrator = yield* Orchestrator.OrchestratorV2;432 const orchestrator = yield* Orchestrator.OrchestratorV2;
433 const layerScope = yield* Effect.scope;
407 const legacyImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;434 const legacyImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
408435
409 const ensureLegacyTranscript = Effect.fn(436 const ensureLegacyTranscript = Effect.fn(
...@@ -724,6 +751,39 @@ const make = Effect.gen(function* () {...@@ -724,6 +751,39 @@ const make = Effect.gen(function* () {
724 return { threadId: input.threadId, run, timedOut: !isTerminalRunStatus(run.status) };751 return { threadId: input.threadId, run, timedOut: !isTerminalRunStatus(run.status) };
725 });752 });
726753
754 const settleAfterRun: ThreadManagementServiceShape["settleAfterRun"] = Effect.fn(
755 "orchestrationV2.threadManagement.settleAfterRun",
756 )(function* (input) {
757 // A turn that outlives this wait leaves its thread active.
758 const { run } = yield* waitForThread({ ...input, timeoutMs: SETTLE_AFTER_RUN_WAIT_MS });
759 if (run?.status !== "completed") return;
760 yield* dispatch({
761 type: "thread.settle",
762 commandId: CommandId.make(`server:settle-after-run:${input.runId}`),
763 threadId: input.threadId,
764 });
765 });
766
767 const settleThread: ThreadManagementServiceShape["settleThread"] = Effect.fn(
768 "orchestrationV2.threadManagement.settleThread",
769 )(function* (input) {
770 const shell = input.byOwnAgent ? yield* orchestrator.getThreadShell(input.threadId) : null;
771 if (shell != null && shell.activeRunId !== null) {
772 yield* settleAfterRun({
773 projectId: shell.projectId,
774 threadId: shell.id,
775 runId: shell.activeRunId,
776 }).pipe(Effect.ignoreCause({ log: true }), Effect.forkIn(layerScope));
777 return { settlesWhenTurnEnds: true } as const;
778 }
779 const result = yield* dispatch({
780 type: "thread.settle",
781 commandId: input.commandId,
782 threadId: input.threadId,
783 });
784 return { sequence: result.sequence };
785 });
786
727 const interruptThread: ThreadManagementServiceShape["interruptThread"] = (input) =>787 const interruptThread: ThreadManagementServiceShape["interruptThread"] = (input) =>
728 Effect.gen(function* () {788 Effect.gen(function* () {
729 const target = yield* getProjectThreadRecords(input, ["runs", "providerTurns"]);789 const target = yield* getProjectThreadRecords(input, ["runs", "providerTurns"]);
...@@ -829,6 +889,8 @@ const make = Effect.gen(function* () {...@@ -829,6 +889,8 @@ const make = Effect.gen(function* () {
829 listProjectThreads,889 listProjectThreads,
830 sendToThread,890 sendToThread,
831 waitForThread,891 waitForThread,
892 settleAfterRun,
893 settleThread,
832 interruptThread,894 interruptThread,
833 stopDelegatedTasks,895 stopDelegatedTasks,
834 getThreadEventSequence: orchestrator.getThreadEventSequence,896 getThreadEventSequence: orchestrator.getThreadEventSequence,
apps/server/src/orchestration-v2/runtimeLayer.test.ts+68
...@@ -3373,6 +3373,74 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => {...@@ -3373,6 +3373,74 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => {
3373 }),3373 }),
3374 );3374 );
33753375
3376 it.effect("settles a thread its own agent settled once the turn completes", () =>
3377 Effect.gen(function* () {
3378 const orchestrator = yield* Orchestrator.OrchestratorV2;
3379 const eventSink = yield* EventSink.EventSinkV2;
3380 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
3381 const projectId = ProjectId.make("runtime-layer-settle-after-run-project");
3382 const threadId = ThreadId.make("runtime-layer-settle-after-run-thread");
3383
3384 yield* orchestrator.dispatch({
3385 type: "thread.create",
3386 createdBy: "user",
3387 creationSource: "web",
3388 commandId: CommandId.make("runtime-layer-settle-after-run-create"),
3389 threadId,
3390 projectId,
3391 title: "Settle after run",
3392 modelSelection,
3393 runtimeMode: "full-access",
3394 interactionMode: "default",
3395 branch: null,
3396 worktreePath: "/tmp/runtime-layer-settle-after-run",
3397 });
3398 yield* orchestrator.dispatch({
3399 type: "message.dispatch",
3400 createdBy: "user",
3401 creationSource: "web",
3402 commandId: CommandId.make("runtime-layer-settle-after-run-message"),
3403 threadId,
3404 messageId: MessageId.make("runtime-layer-settle-after-run-message"),
3405 text: "Fix it and then settle this thread.",
3406 attachments: [],
3407 modelSelection,
3408 dispatchMode: { type: "start_immediately" },
3409 });
3410 const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0];
3411 if (run === undefined) return yield* Effect.die(new Error("Run missing."));
3412
3413 const settled = yield* orchestrator
3414 .streamStoredEventsFrom({ threadId, afterSequence: 0, eventType: "thread.settled" })
3415 .pipe(Stream.runHead, Effect.forkChild);
3416 const result = yield* threadManagement.settleThread({
3417 threadId,
3418 commandId: CommandId.make("runtime-layer-settle-after-run-settle"),
3419 byOwnAgent: true,
3420 });
3421 assert.deepEqual(result, { settlesWhenTurnEnds: true });
3422 assert.isNull((yield* orchestrator.getThreadProjection(threadId)).thread.settledOverride);
3423 const now = yield* DateTime.now;
3424 yield* eventSink.write({
3425 commandId: CommandId.make("runtime-layer-settle-after-run-completed"),
3426 events: [
3427 {
3428 id: EventId.make("runtime-layer-settle-after-run-completed"),
3429 type: "run.updated",
3430 threadId,
3431 runId: run.id,
3432 occurredAt: now,
3433 payload: { ...run, status: "completed", startedAt: now, completedAt: now },
3434 },
3435 ],
3436 });
3437 yield* Fiber.join(settled);
3438
3439 const projection = yield* orchestrator.getThreadProjection(threadId);
3440 assert.equal(projection.thread.settledOverride, "settled");
3441 }),
3442 );
3443
3376 it.effect("settles past held automatic runs but not held user messages", () =>3444 it.effect("settles past held automatic runs but not held user messages", () =>
3377 Effect.gen(function* () {3445 Effect.gen(function* () {
3378 const orchestrator = yield* Orchestrator.OrchestratorV2;3446 const orchestrator = yield* Orchestrator.OrchestratorV2;
apps/server/src/relay/AgentAwarenessRelay.test.ts+2
...@@ -206,6 +206,8 @@ const makeTestRelay = Effect.fnUntraced(function* (...@@ -206,6 +206,8 @@ const makeTestRelay = Effect.fnUntraced(function* (
206 listProjectThreads: unused,206 listProjectThreads: unused,
207 sendToThread: unused,207 sendToThread: unused,
208 waitForThread: unused,208 waitForThread: unused,
209 settleAfterRun: unused,
210 settleThread: unused,
209 interruptThread: unused,211 interruptThread: unused,
210 stopDelegatedTasks: unused,212 stopDelegatedTasks: unused,
211 getThreadEventSequence: unused,213 getThreadEventSequence: unused,