From 7703cf8962ad72e6a57e7a69745d8c00ae0f477a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:31:24 +0000 Subject: [PATCH 1/8] test(server): cover thread wait shell lifecycle and concurrent work --- .../orchestration-v2/ProjectionStore.test.ts | 114 ++++++++++++++++++ 1 file changed, 114 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 74377160a820..aa90eb2b5ab7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -9,6 +9,7 @@ import { MessageId, type ModelSelection, type OrchestrationV2ProviderThread, + type OrchestrationV2TurnItem, type OrchestrationV2ThreadShellSnapshot, NodeId, ProjectId, @@ -183,6 +184,119 @@ const addRolledBackRecoveryCandidate = Effect.fn("addRolledBackRecoveryCandidate return threadId; }); +it.effect.each([ + ["memory", ProjectionStore.layerMemory], + ["SQL", layerTest], +] as const)("%s shells flag only an active turn blocked solely on T3 thread waits", ([, layer]) => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const threadId = yield* addRolledBackRecoveryCandidate("thread-wait"); + const initial = yield* store.getThreadProjection(threadId); + const at = initial.updatedAt; + yield* store.apply({ + id: EventId.make("event:thread-wait:previous-completed"), + type: "run.updated", + threadId, + occurredAt: at, + payload: { ...initial.runs[0]!, status: "completed", completedAt: at }, + }); + const run = { + ...initial.runs[0]!, + id: RunId.make("run:thread-wait:active"), + ordinal: 2, + status: "running" as const, + completedAt: null, + }; + yield* store.apply({ + id: EventId.make("event:thread-wait:active"), + type: "run.created", + threadId, + occurredAt: at, + payload: run, + }); + let sequence = 0; + const updateItem = (payload: OrchestrationV2TurnItem) => + store.apply({ + id: EventId.make(`event:thread-wait:item:${sequence++}`), + type: "turn-item.updated", + threadId, + runId: payload.runId, + driver, + occurredAt: at, + payload, + }); + const assertBlocked = Effect.fnUntraced(function* (expected: boolean) { + const listed = (yield* store.getShellSnapshot()).threads.find((row) => row.id === threadId); + for (const shell of [yield* store.getThreadShell(threadId), listed]) { + assert.equal(shell?.activeRunId, run.id); + assert.equal(shell?.status, "running"); + assert.equal(shell?.blockedOnThreadWait ?? false, expected); + } + }); + const wait = { + ...initial.turnItems[0]!, + id: TurnItemId.make("item:thread-wait:active"), + runId: run.id, + type: "dynamic_tool" as const, + toolName: "mcp__t3_code__t3_thread_wait", + input: { threadId: "child-thread" }, + status: "running" as const, + }; + + yield* updateItem({ + ...wait, + id: TurnItemId.make("item:thread-wait:stale"), + runId: initial.runs[0]!.id, + }); + yield* assertBlocked(false); + const foreign = { + ...wait, + id: TurnItemId.make("item:thread-wait:foreign"), + toolName: "weather.t3_thread_wait", + }; + yield* updateItem(foreign); + yield* assertBlocked(false); + yield* updateItem({ ...foreign, status: "completed", completedAt: at }); + yield* updateItem({ ...wait, status: "pending", startedAt: null }); + yield* assertBlocked(false); + yield* updateItem(wait); + yield* assertBlocked(true); + + const command = { + ...initial.turnItems[0]!, + id: TurnItemId.make("item:thread-wait:concurrent"), + runId: run.id, + }; + yield* updateItem(command); + yield* assertBlocked(false); + yield* updateItem({ ...wait, status: "waiting" }); + yield* assertBlocked(false); + yield* updateItem({ ...command, status: "completed", completedAt: at }); + yield* assertBlocked(true); + yield* updateItem({ ...wait, status: "completed", completedAt: at }); + yield* assertBlocked(false); + + const delegation = { + ...wait, + id: TurnItemId.make("item:thread-wait:delegation"), + toolName: "t3-code-thread_opencode2-adapter_delegate_task", + input: { task: "Review", mode: "async" }, + }; + yield* updateItem(delegation); + yield* assertBlocked(false); + yield* updateItem({ ...delegation, status: "completed", completedAt: at }); + const blockingDelegation = { + ...delegation, + id: TurnItemId.make("item:thread-wait:blocking-delegation"), + input: { task: "Review", mode: "wait" }, + }; + yield* updateItem(blockingDelegation); + yield* assertBlocked(true); + yield* updateItem({ ...blockingDelegation, status: "completed", completedAt: at }); + yield* assertBlocked(false); + }).pipe(Effect.provide(layer)), +); + it.effect.each([ ["memory", ProjectionStore.layerMemory], ["SQL", layerTest], From b5c26087003503bf3abe493d9eaf16a702ad5b22 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:41:28 +0000 Subject: [PATCH 2/8] fix(server): derive when active turns block on T3 thread waits --- .../orchestration-v2/ProjectionStore.test.ts | 36 +++++++++++++++-- .../src/orchestration-v2/ProjectionStore.ts | 40 ++++++++++++++----- packages/contracts/src/orchestrationV2.ts | 1 + .../orchestrationV2PendingBackgroundWork.ts | 34 ++++++++++++++++ 4 files changed, 97 insertions(+), 14 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index aa90eb2b5ab7..30fcb458cc93 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -193,6 +193,14 @@ it.effect.each([ const threadId = yield* addRolledBackRecoveryCandidate("thread-wait"); const initial = yield* store.getThreadProjection(threadId); const at = initial.updatedAt; + const previousCommand = initial.turnItems.find((item) => item.type === "command_execution")!; + const itemBase = { + ...previousCommand, + providerThreadId: previousCommand.providerThreadId ?? null, + providerTurnId: previousCommand.providerTurnId ?? null, + nativeItemRef: previousCommand.nativeItemRef ?? null, + parentItemId: previousCommand.parentItemId ?? null, + }; yield* store.apply({ id: EventId.make("event:thread-wait:previous-completed"), type: "run.updated", @@ -220,7 +228,7 @@ it.effect.each([ id: EventId.make(`event:thread-wait:item:${sequence++}`), type: "turn-item.updated", threadId, - runId: payload.runId, + runId: payload.runId ?? undefined, driver, occurredAt: at, payload, @@ -234,7 +242,7 @@ it.effect.each([ } }); const wait = { - ...initial.turnItems[0]!, + ...itemBase, id: TurnItemId.make("item:thread-wait:active"), runId: run.id, type: "dynamic_tool" as const, @@ -263,7 +271,7 @@ it.effect.each([ yield* assertBlocked(true); const command = { - ...initial.turnItems[0]!, + ...itemBase, id: TurnItemId.make("item:thread-wait:concurrent"), runId: run.id, }; @@ -273,6 +281,16 @@ it.effect.each([ yield* assertBlocked(false); yield* updateItem({ ...command, status: "completed", completedAt: at }); yield* assertBlocked(true); + const fileChange = { + ...command, + id: TurnItemId.make("item:thread-wait:concurrent-edit"), + type: "file_change" as const, + fileName: "example.ts", + }; + yield* updateItem(fileChange); + yield* assertBlocked(false); + yield* updateItem({ ...fileChange, status: "completed", completedAt: at }); + yield* assertBlocked(true); yield* updateItem({ ...wait, status: "completed", completedAt: at }); yield* assertBlocked(false); @@ -290,6 +308,18 @@ it.effect.each([ id: TurnItemId.make("item:thread-wait:blocking-delegation"), input: { task: "Review", mode: "wait" }, }; + yield* updateItem({ + ...command, + id: TurnItemId.make("item:thread-wait:child"), + type: "subagent", + subagentId: NodeId.make("node:thread-wait:child"), + origin: "app_owned", + driver, + providerInstanceId, + childThreadId: ThreadId.make("child-thread"), + prompt: "Review", + result: null, + }); yield* updateItem(blockingDelegation); yield* assertBlocked(true); yield* updateItem({ ...blockingDelegation, status: "completed", completedAt: at }); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 002ef7bafa95..dd2b4edeb5a6 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -69,7 +69,10 @@ import { createOrchestrationV2TurnItemVisibility, isOrchestrationV2SupersededInterrupt, } from "@t3tools/shared/orchestrationV2Timeline"; -import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; +import { + deriveBlockedOnThreadWait, + derivePendingBackgroundWork, +} from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -1014,6 +1017,7 @@ type ShellThreadRow = { readonly latest_run_started_at: string | null; readonly latest_run_completed_at: string | null; readonly active_run_id: string | null; + readonly active_run_status: string | null; readonly activity_run_status: string | null; readonly activity_run_started_at: string | null; readonly last_error: string | null; @@ -1529,6 +1533,7 @@ export function threadShellFromProjection( latestRunStartedAt: latestRun?.startedAt ?? null, latestRunCompletedAt: latestRun?.completedAt ?? null, activeRunId: activeRun?.id ?? null, + blockedOnThreadWait: deriveBlockedOnThreadWait({ activeRun, turnItems: projection.turnItems }), activityRunStatus: activityRun?.status ?? null, activityRunStartedAt: activityRun === null ? null : orchestrationV2RunWorkStartedAt(activityRun), @@ -1634,6 +1639,7 @@ type ShellThreadState = { readonly latestRunStartedAt: DateTime.Utc | null; readonly latestRunCompletedAt: DateTime.Utc | null; readonly activeRunId: RunId | null; + readonly blockedOnThreadWait: boolean; readonly activityRunStatus: ShellActivityRunStatus | null; readonly activityRunStartedAt: DateTime.Utc | null; readonly lastError: string | null; @@ -1786,6 +1792,7 @@ function shellFromState(input: { latestRunStartedAt: input.state.latestRunStartedAt, latestRunCompletedAt: input.state.latestRunCompletedAt, activeRunId: input.state.activeRunId, + blockedOnThreadWait: input.state.blockedOnThreadWait, activityRunStatus: input.state.activityRunStatus, activityRunStartedAt: input.state.activityRunStartedAt, status: input.state.latestRunStatus, @@ -5292,14 +5299,8 @@ export const layer: Layer.Layer = presented.requested_at AS latest_run_requested_at, json_extract(presented.payload_json, '$.startedAt') AS latest_run_started_at, presented.completed_at AS latest_run_completed_at, - ( - SELECT r.run_id - FROM orchestration_v2_projection_runs r - WHERE r.thread_id = t.thread_id - AND r.status IN ('preparing', 'starting', 'running') - ORDER BY r.ordinal DESC, r.run_id DESC - LIMIT 1 - ) AS active_run_id, + active.run_id AS active_run_id, + active.status AS active_run_status, ( SELECT r.status FROM orchestration_v2_projection_runs r @@ -5424,6 +5425,14 @@ export const layer: Layer.Layer = AND i.run_id IS NULL ) AS runless_item_count FROM orchestration_v2_projection_threads t + LEFT JOIN orchestration_v2_projection_runs active ON active.run_id = ( + SELECT r.run_id + FROM orchestration_v2_projection_runs r + WHERE r.thread_id = t.thread_id + AND r.status IN ('preparing', 'starting', 'running') + ORDER BY r.ordinal DESC, r.run_id DESC + LIMIT 1 + ) -- The newest run not waiting in a held queue, matching latestUnheldRun. LEFT JOIN orchestration_v2_projection_runs presented ON presented.run_id = ( SELECT candidate.run_id @@ -5516,7 +5525,8 @@ export const layer: Layer.Layer = ON p.provider_thread_id = i.provider_thread_id LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = i.run_id - WHERE i.type IN ('command_execution', 'dynamic_tool', 'subagent') + WHERE (i.type IN ('command_execution', 'dynamic_tool', 'subagent') + OR (r.status = 'running' AND i.status IN ('running', 'waiting'))) AND i.status NOT IN ('completed', 'interrupted', 'failed', 'cancelled') -- A rolled-back run's items are abandoned, not pending. Without -- this the shell reports Waiting for work no one will finish, @@ -5530,7 +5540,8 @@ export const layer: Layer.Layer = ON p.provider_thread_id = i.provider_thread_id LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = i.run_id - WHERE i.type IN ('command_execution', 'dynamic_tool', 'subagent') + WHERE (i.type IN ('command_execution', 'dynamic_tool', 'subagent') + OR (r.status = 'running' AND i.status IN ('running', 'waiting'))) AND i.status NOT IN ('completed', 'interrupted', 'failed', 'cancelled') AND (i.run_id IS NULL OR r.status <> 'rolled_back') AND i.thread_id IN ${sql.in(threadIds)} @@ -5856,6 +5867,13 @@ export const layer: Layer.Layer = latestRunStartedAt, latestRunCompletedAt, activeRunId: row.active_run_id === null ? null : RunId.make(row.active_run_id), + blockedOnThreadWait: deriveBlockedOnThreadWait({ + activeRun: + row.active_run_id === null || row.active_run_status !== "running" + ? null + : { id: RunId.make(row.active_run_id), status: "running" }, + turnItems: pendingTurnItemsByThreadId.get(thread.id) ?? [], + }), activityRunStartedAt: row.activity_run_started_at === null ? null diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 94a9eca331d1..f91b077dc871 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1822,6 +1822,7 @@ export const OrchestrationV2ThreadShell = Schema.Struct({ activityRunStatus: Schema.optional( Schema.NullOr(Schema.Literals(["preparing", "starting", "running", "waiting"])), ), + blockedOnThreadWait: Schema.optional(Schema.Boolean), status: OrchestrationV2ShellThreadStatus, lastError: Schema.optional(Schema.NullOr(Schema.String)), lastErrorClass: Schema.optional(Schema.NullOr(OrchestrationV2ProviderFailureClass)), diff --git a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts index 15453ae01850..3d07b5f349ea 100644 --- a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts +++ b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts @@ -7,7 +7,9 @@ import type { ThreadPullRequestLink, } from "@t3tools/contracts"; import { isOrchestrationV2WorkActive } from "@t3tools/contracts"; +import * as Predicate from "effect/Predicate"; +import { resolveT3McpToolId } from "./t3McpToolPresentation.ts"; import { threadPullRequestKeyOf } from "./threadPullRequests.ts"; const BACKGROUND_TURN_ITEM_TYPES = new Set([ @@ -48,6 +50,38 @@ export function turnItemUpdateCanEndBackgroundWork( return BACKGROUND_TURN_ITEM_TYPES.has(item.type) && !isOrchestrationV2WorkActive(item.status); } +export function deriveBlockedOnThreadWait(input: { + readonly activeRun: Pick | null; + readonly turnItems: ReadonlyArray; +}): boolean { + if (input.activeRun?.status !== "running") return false; + let hasWait = false; + for (const item of input.turnItems) { + if ( + item.runId !== input.activeRun.id || + (item.status !== "running" && item.status !== "waiting") || + // App-owned rows track the child, including the child a blocking delegate is waiting for. + (item.type === "subagent" && item.origin === "app_owned") + ) { + continue; + } + if (item.type !== "dynamic_tool") return false; + const tool = resolveT3McpToolId(item.toolName); + if ( + tool !== "t3_thread_wait" && + !( + tool === "delegate_task" && + Predicate.isObject(item.input) && + Reflect.get(item.input, "mode") === "wait" + ) + ) { + return false; + } + hasWait = true; + } + return hasWait; +} + export function backgroundWorkHoldsCompletion( tasks: ReadonlyArray>, ): boolean { From 642b5313b1150f5a4601d4c141514e83fba179ea Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:46:14 +0000 Subject: [PATCH 3/8] fix(providers): preserve T3 wait inputs and thread tool identities --- .../Adapters/OpenCode2AdapterV2.test.ts | 44 +++++++++++++++++++ .../src/server/adapter.test.ts | 6 +-- .../provider-cursor/src/server/adapter.ts | 2 +- .../provider-opencode/src/server/toolItems.ts | 2 + 4 files changed, 48 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 4b3c72b11e17..06cd9bd9b61d 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -3202,6 +3202,50 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("keeps thread-scoped T3 waits as dynamic tools through their lifecycle", () => + Effect.gen(function* () { + const tool = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-wait" }; + const name = "t3-code-web_delegate_task"; + const input = { task: "Review", mode: "wait" }; + const { runtime, thread } = yield* resumed([ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.tool.input.started", { ...tool, name }), + event("session.tool.called", { ...tool, name, input, executed: true }), + event("session.tool.success", { + ...tool, + content: [{ type: "text", text: "Reviewed" }], + metadata: {}, + executed: true, + }), + event("session.execution.succeeded", { sessionID: SESSION }), + ]); + const events = yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + Effect.forkScoped, + ); + yield* runtime.startTurn(turnInput(thread)); + const items = (yield* Fiber.join(events)).flatMap((event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId?.endsWith(tool.id) + ? [event.turnItem] + : [], + ); + assert.deepEqual( + items.map((item) => item.type), + ["dynamic_tool", "dynamic_tool", "dynamic_tool"], + ); + assert.deepEqual( + items.map((item) => item.status), + ["running", "running", "completed"], + ); + const last = items.at(-1); + assert.equal(last?.type === "dynamic_tool" ? last.toolName : null, name); + assert.deepEqual(last?.type === "dynamic_tool" ? last.input : null, input); + }).pipe(Effect.scoped), + ); + it.effect( "registers T3's MCP server for the thread alone and removes it when the thread unloads", () => diff --git a/packages/provider-cursor/src/server/adapter.test.ts b/packages/provider-cursor/src/server/adapter.test.ts index be87e4a48196..8aecb9f92dfe 100644 --- a/packages/provider-cursor/src/server/adapter.test.ts +++ b/packages/provider-cursor/src/server/adapter.test.ts @@ -830,11 +830,7 @@ describe("CursorAdapterV2", () => { name: "weather", kind: "integration", }); - assert.deepEqual(item.input, { - providerIdentifier: "weather", - toolName: "get_weather", - args: { city: "Berlin" }, - }); + assert.deepEqual(item.input, { city: "Berlin" }); } const fileSearchItems = events.flatMap((event) => event.type === "turn_item.updated" && diff --git a/packages/provider-cursor/src/server/adapter.ts b/packages/provider-cursor/src/server/adapter.ts index ae495016e9ff..56c21d6c15c4 100644 --- a/packages/provider-cursor/src/server/adapter.ts +++ b/packages/provider-cursor/src/server/adapter.ts @@ -1319,7 +1319,7 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( }) : {}), toolName: cursorToolName(toolCall), - input: toolCall.args, + input: toolCall.type === "mcp" ? toolCall.args.args : toolCall.args, ...(cursorToolOutput(toolCall) === undefined ? {} : { output: cursorToolOutput(toolCall) }), diff --git a/packages/provider-opencode/src/server/toolItems.ts b/packages/provider-opencode/src/server/toolItems.ts index f4551cdd279c..55948e76d02c 100644 --- a/packages/provider-opencode/src/server/toolItems.ts +++ b/packages/provider-opencode/src/server/toolItems.ts @@ -1,4 +1,5 @@ import type { OrchestrationV2TurnItem } from "@t3tools/contracts"; +import { resolveT3McpToolId } from "@t3tools/shared/t3McpToolPresentation"; import { formatReadToolLabel, formatSearchToolLabel } from "@t3tools/shared/toolActivity"; // Search results stay on the timeline wire, so keep their text a preview. @@ -12,6 +13,7 @@ type ToolItemBase = Omit< export function openCodeToolProjectionKind( toolName: string, ): "command_execution" | "file_change" | "file_search" | "web_search" | "dynamic_tool" { + if (resolveT3McpToolId(toolName) !== null) return "dynamic_tool"; const normalized = toolName.toLowerCase(); if (normalized === "todowrite") { return "dynamic_tool"; From 4f2c902886d5802704ecbb56ee65602ea5b750b2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:48:22 +0000 Subject: [PATCH 4/8] perf(server): keep thread wait shell reads on active item indexes --- .../src/orchestration-v2/ProjectionStore.ts | 78 +++++++++++-------- 1 file changed, 46 insertions(+), 32 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index dd2b4edeb5a6..3dd7dcdae94e 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -5516,36 +5516,34 @@ export const layer: Layer.Layer = WHERE thread_id IN ${sql.in(threadIds)} `; - const selectShellPendingTurnItemRows = (threadIds?: ReadonlyArray) => - threadIds === undefined - ? sql` - SELECT i.thread_id, i.payload_json, p.provider_instance_id AS owning_provider_instance_id - FROM orchestration_v2_projection_turn_items i - LEFT JOIN orchestration_v2_projection_provider_threads p - ON p.provider_thread_id = i.provider_thread_id - LEFT JOIN orchestration_v2_projection_runs r - ON r.run_id = i.run_id - WHERE (i.type IN ('command_execution', 'dynamic_tool', 'subagent') - OR (r.status = 'running' AND i.status IN ('running', 'waiting'))) - AND i.status NOT IN ('completed', 'interrupted', 'failed', 'cancelled') - -- A rolled-back run's items are abandoned, not pending. Without - -- this the shell reports Waiting for work no one will finish, - -- matching the item_count query's exclusion above. - AND (i.run_id IS NULL OR r.status <> 'rolled_back') - ` - : sql` - SELECT i.thread_id, i.payload_json, p.provider_instance_id AS owning_provider_instance_id - FROM orchestration_v2_projection_turn_items i - LEFT JOIN orchestration_v2_projection_provider_threads p - ON p.provider_thread_id = i.provider_thread_id - LEFT JOIN orchestration_v2_projection_runs r - ON r.run_id = i.run_id - WHERE (i.type IN ('command_execution', 'dynamic_tool', 'subagent') - OR (r.status = 'running' AND i.status IN ('running', 'waiting'))) - AND i.status NOT IN ('completed', 'interrupted', 'failed', 'cancelled') - AND (i.run_id IS NULL OR r.status <> 'rolled_back') - AND i.thread_id IN ${sql.in(threadIds)} - `; + const selectShellPendingTurnItemRows = ( + threadIds?: ReadonlyArray, + activeRunIds: ReadonlyArray = [], + ) => sql` + SELECT i.thread_id, i.payload_json, p.provider_instance_id AS owning_provider_instance_id + FROM orchestration_v2_projection_turn_items i + LEFT JOIN orchestration_v2_projection_provider_threads p + ON p.provider_thread_id = i.provider_thread_id + LEFT JOIN orchestration_v2_projection_runs r + ON r.run_id = i.run_id + WHERE i.type IN ('command_execution', 'dynamic_tool', 'subagent') + AND i.status NOT IN ('completed', 'interrupted', 'failed', 'cancelled') + AND (i.run_id IS NULL OR r.status <> 'rolled_back') + ${threadIds === undefined ? sql`` : sql`AND i.thread_id IN ${sql.in(threadIds)}`} + ${ + activeRunIds.length === 0 + ? sql`` + : sql` + -- Preserve the pending-item partial index; read other work only from running runs. + UNION ALL + SELECT i.thread_id, i.payload_json, NULL AS owning_provider_instance_id + FROM orchestration_v2_projection_turn_items i + WHERE i.run_id IN ${sql.in(activeRunIds)} + AND i.type NOT IN ('command_execution', 'dynamic_tool', 'subagent') + AND i.status IN ('running', 'waiting') + ` + } + `; const runMapsByThreadId = (input: { readonly runRows: ReadonlyArray; @@ -5961,7 +5959,16 @@ export const layer: Layer.Layer = AND aggregate_kind = 'thread' `, readForThreadIds(selectShellProviderThreadRows), - readForThreadIds(selectShellPendingTurnItemRows), + readForThreadIds((ids) => + selectShellPendingTurnItemRows( + ids, + threadRows.flatMap((row) => + row.active_run_id !== null && row.active_run_status === "running" + ? [row.active_run_id] + : [], + ), + ), + ), ]); return { targetThreadIds, @@ -6068,7 +6075,14 @@ export const layer: Layer.Layer = ? Effect.succeed([] as ReadonlyArray) : selectShellRunItemCounts(forkSourceIds), selectShellProviderThreadRows(threadIds), - selectShellPendingTurnItemRows(threadIds), + selectShellPendingTurnItemRows( + threadIds, + [...rowsByThreadId.values()].flatMap((row) => + row.active_run_id !== null && row.active_run_status === "running" + ? [row.active_run_id] + : [], + ), + ), ]); const { runOrdinalsByThreadId, itemCountsByThreadId } = runMapsByThreadId({ runRows, From 9b6ab5ccb2d878b9ad2233a791c8774a6f0af649 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 04:07:46 +0000 Subject: [PATCH 5/8] fix(server): ignore open plans and text when deriving thread waits --- .../Adapters/AcpAdapterV2.test.ts | 135 ++++++++++++++++++ .../orchestration-v2/ProjectionStore.test.ts | 46 ++++++ .../src/orchestration-v2/ProjectionStore.ts | 2 +- .../orchestrationV2PendingBackgroundWork.ts | 10 ++ 4 files changed, 192 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 35c271ac0f0a..fbad163c263f 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -27,10 +27,12 @@ import { RunId, ThreadId, type OrchestrationV2ProviderThread, + type OrchestrationV2TurnItem, type ProviderOptionDescriptor, } from "@t3tools/contracts"; import { GrokSettings } from "@t3tools/provider-grok/settings"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import { deriveBlockedOnThreadWait } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { resolveSelfInvocation } from "@t3tools/shared/nodeRuntime"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; @@ -980,6 +982,139 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect.each(["agent_message_chunk", "agent_thought_chunk"] as const)( + "keeps named ACP %s open while deriving thread wait execution state", + (sessionUpdate) => + Effect.gen(function* () { + type RuntimeService = AcpSessionRuntime.AcpSessionRuntime["Service"]; + let handler: Parameters[0] | undefined; + const instanceId = ProviderInstanceId.make("acp-named-stream-wait"); + const path = yield* Path.Path; + const adapter = yield* makeAcpAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + makeRuntime: makeMockRuntime({ + childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, + mockAgentPath: yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ), + wrapRuntime: (runtime) => ({ + ...runtime, + handleSessionUpdate: (next) => + Effect.sync(() => { + handler = next; + }).pipe(Effect.andThen(runtime.handleSessionUpdate(next))), + prompt: () => + Effect.gen(function* () { + assert.isDefined(handler); + const updates = [ + { + sessionUpdate, + messageId: "named-stream", + content: { type: "text", text: "Waiting for the child result" }, + }, + { + sessionUpdate: "tool_call", + toolCallId: "wait", + title: "Wait for the child", + status: "in_progress", + rawInput: { + server: "t3-code", + tool: "t3_thread_wait", + arguments: { threadId: "child" }, + }, + _meta: { is_mcp_tool_call: true }, + }, + { + sessionUpdate: "tool_call", + toolCallId: "command", + kind: "execute", + title: "Run command", + status: "in_progress", + rawInput: { command: "pwd" }, + }, + { + sessionUpdate: "tool_call_update", + toolCallId: "command", + status: "completed", + }, + { + sessionUpdate: "tool_call_update", + toolCallId: "wait", + status: "completed", + }, + ] satisfies Array; + for (const update of updates) + yield* handler!({ sessionId: "mock-session-1", update }); + return { stopReason: "end_turn" as const }; + }), + }), + }), + }, + }); + const threadId = ThreadId.make("thread-acp-named-stream-wait"); + const modelSelection = { instanceId, model: "default" }; + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("session-acp-named-stream-wait"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const input = makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + }); + yield* runtime.startTurn(input); + const events = yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ); + const retained = new Map(); + const states: Array = []; + for (const event of events) { + if (event.type !== "turn_item.updated") continue; + const item = event.turnItem; + retained.set(item.id, item); + if (item.type !== "dynamic_tool" && item.type !== "command_execution") continue; + const streams = [...retained.values()].filter( + (entry) => entry.type === "assistant_message" || entry.type === "reasoning", + ); + assert.deepEqual( + streams.map((entry) => [entry.type, entry.status]), + [ + [ + sessionUpdate === "agent_message_chunk" ? "assistant_message" : "reasoning", + "running", + ], + ], + ); + states.push( + deriveBlockedOnThreadWait({ + activeRun: { id: input.runId, status: "running" }, + turnItems: [...retained.values()], + }), + ); + } + assert.deepEqual(states, [true, false, true, false]); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect("starts a new replay message after ACP v2 plan boundaries", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 30fcb458cc93..8b7edd2fbe17 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -13,6 +13,7 @@ import { type OrchestrationV2ThreadShellSnapshot, NodeId, ProjectId, + PlanId, ProviderDriverKind, ProviderInstanceId, ProviderSessionId, @@ -270,6 +271,41 @@ it.effect.each([ yield* updateItem(wait); yield* assertBlocked(true); + yield* updateItem({ + ...wait, + id: TurnItemId.make("item:thread-wait:plan"), + type: "todo_list", + planId: PlanId.make("plan:thread-wait"), + steps: [{ id: "review", text: "Review the child result", status: "running" }], + }); + yield* assertBlocked(true); + yield* updateItem({ + ...wait, + id: TurnItemId.make("item:thread-wait:proposed-plan"), + type: "proposed_plan", + planId: PlanId.make("plan:thread-wait:proposed"), + markdown: "Review the child result", + streaming: true, + }); + yield* updateItem({ + ...wait, + id: TurnItemId.make("item:thread-wait:named-message"), + nativeItemRef: { driver, nativeId: "named-message", strength: "weak" }, + type: "assistant_message", + messageId: MessageId.make("message:thread-wait:named"), + text: "Waiting for the child result", + streaming: true, + }); + yield* updateItem({ + ...wait, + id: TurnItemId.make("item:thread-wait:named-thought"), + nativeItemRef: { driver, nativeId: "named-thought", strength: "weak" }, + type: "reasoning", + text: "I need the child result", + streaming: true, + }); + yield* assertBlocked(true); + const command = { ...itemBase, id: TurnItemId.make("item:thread-wait:concurrent"), @@ -291,6 +327,16 @@ it.effect.each([ yield* assertBlocked(false); yield* updateItem({ ...fileChange, status: "completed", completedAt: at }); yield* assertBlocked(true); + const compaction = { + ...command, + id: TurnItemId.make("item:thread-wait:compaction"), + type: "compaction" as const, + driver, + }; + yield* updateItem(compaction); + yield* assertBlocked(false); + yield* updateItem({ ...compaction, status: "completed", completedAt: at }); + yield* assertBlocked(true); yield* updateItem({ ...wait, status: "completed", completedAt: at }); yield* assertBlocked(false); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 3dd7dcdae94e..66258e980cde 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -5539,7 +5539,7 @@ export const layer: Layer.Layer = SELECT i.thread_id, i.payload_json, NULL AS owning_provider_instance_id FROM orchestration_v2_projection_turn_items i WHERE i.run_id IN ${sql.in(activeRunIds)} - AND i.type NOT IN ('command_execution', 'dynamic_tool', 'subagent') + AND i.type IN ('file_change', 'file_search', 'web_search', 'compaction') AND i.status IN ('running', 'waiting') ` } diff --git a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts index 3d07b5f349ea..8073bcccd0bb 100644 --- a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts +++ b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts @@ -18,6 +18,15 @@ const BACKGROUND_TURN_ITEM_TYPES = new Set([ "subagent", ]); +// Plans and text streams can stay open across tool calls without executing work. +const EXECUTION_TURN_ITEM_TYPES = new Set([ + ...BACKGROUND_TURN_ITEM_TYPES, + "file_change", + "file_search", + "web_search", + "compaction", +]); + const TERMINAL_RUN_STATUSES = new Set([ "cancelled", "completed", @@ -59,6 +68,7 @@ export function deriveBlockedOnThreadWait(input: { for (const item of input.turnItems) { if ( item.runId !== input.activeRun.id || + !EXECUTION_TURN_ITEM_TYPES.has(item.type) || (item.status !== "running" && item.status !== "waiting") || // App-owned rows track the child, including the child a blocking delegate is waiting for. (item.type === "subagent" && item.origin === "app_owned") From a53ea23e35006e0572e6eafb172d399c8ecdc7c1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 04:23:47 +0000 Subject: [PATCH 6/8] fix(cursor): default missing MCP tool arguments to an object --- packages/provider-cursor/src/server/adapter.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/provider-cursor/src/server/adapter.ts b/packages/provider-cursor/src/server/adapter.ts index 56c21d6c15c4..1fb988b68ca2 100644 --- a/packages/provider-cursor/src/server/adapter.ts +++ b/packages/provider-cursor/src/server/adapter.ts @@ -1319,7 +1319,7 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( }) : {}), toolName: cursorToolName(toolCall), - input: toolCall.type === "mcp" ? toolCall.args.args : toolCall.args, + input: toolCall.type === "mcp" ? (toolCall.args.args ?? {}) : toolCall.args, ...(cursorToolOutput(toolCall) === undefined ? {} : { output: cursorToolOutput(toolCall) }), From cf2280556a1fc35e058a0517ecd9ce48ac3c6962 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 20 Oct 2026 02:58:46 +0000 Subject: [PATCH 7/8] fix(server): deduplicate merged projection test import --- apps/server/src/orchestration-v2/ProjectionStore.test.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index a66596ea112c..4897405416c1 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -14,7 +14,6 @@ import { NodeId, PlanId, ProjectId, - PlanId, ProviderDriverKind, ProviderInstanceId, ProviderSessionId, From 0aea756a5dc88db7b299fc0150a14a751554c06d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 20 Oct 2026 03:58:12 +0000 Subject: [PATCH 8/8] fix(mobile): retain callback import across main integration --- apps/mobile/src/features/threads/ThreadDetailScreen.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 500abb7575ad..153d7c681f4c 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -58,7 +58,7 @@ import type { QueuedRunEdit } from "../../state/queued-run-edit"; import type { FollowUpBehavior } from "../../lib/followUpBehavior"; import type { ActiveTurnComposerAction } from "@t3tools/client-runtime/state/composer-dispatch"; import * as Haptics from "expo-haptics"; -import { useContext, useEffect, useLayoutEffect, useRef, useState } from "react"; +import { useCallback, useContext, useEffect, useLayoutEffect, useRef, useState } from "react"; import { AppState, Keyboard,