From bbd8995b3cabefc83464a90d4c63b3bb55777620 Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Thu, 3 Sep 2026 20:43:04 -0700 Subject: [PATCH 1/4] feat(threads): show pending agent questions --- .../features/threads/thread-list-v2-items.tsx | 1 + .../src/features/threads/threadListV2.test.ts | 22 +++ .../src/features/threads/threadListV2.ts | 10 +- .../features/threads/threadPresentation.ts | 13 ++ apps/server/src/mcp/McpHttpServer.ts | 31 +++- .../server/src/mcp/McpSessionRegistry.test.ts | 15 ++ apps/server/src/mcp/McpSessionRegistry.ts | 11 +- .../toolkits/threadAttention/handlers.test.ts | 62 +++++++ .../mcp/toolkits/threadAttention/handlers.ts | 57 ++++++ .../src/mcp/toolkits/threadAttention/tools.ts | 45 +++++ .../Layers/ProjectionPipeline.test.ts | 52 ++++++ .../Layers/ProjectionPipeline.ts | 27 +++ .../Layers/ProjectionSnapshotQuery.test.ts | 4 + .../Layers/ProjectionSnapshotQuery.ts | 12 ++ apps/server/src/orchestration/Schemas.ts | 4 + .../ThreadSettlementPolicy.test.ts | 3 + .../orchestration/ThreadSettlementPolicy.ts | 1 + .../orchestration/decider.attention.test.ts | 166 ++++++++++++++++++ apps/server/src/orchestration/decider.ts | 129 +++++++++++++- .../orchestration/projector.attention.test.ts | 70 ++++++++ apps/server/src/orchestration/projector.ts | 30 ++++ .../Layers/ProjectionRepositories.test.ts | 3 + .../persistence/Layers/ProjectionThreads.ts | 8 +- apps/server/src/persistence/Migrations.ts | 2 + .../048_ProjectionThreadAttention.test.ts | 27 +++ .../048_ProjectionThreadAttention.ts | 16 ++ .../persistence/Services/ProjectionThreads.ts | 2 + .../provider/Layers/ProviderService.test.ts | 45 ++--- .../src/provider/Layers/ProviderService.ts | 31 +--- apps/web/src/components/Sidebar.logic.test.ts | 20 +++ apps/web/src/components/Sidebar.logic.ts | 25 ++- apps/web/src/components/Sidebar.tsx | 39 ++-- .../src/components/ThreadStatusIndicators.tsx | 32 ++-- docs/user/thread-sidebar.md | 13 ++ .../client-runtime/src/state/threadDetail.ts | 1 + .../client-runtime/src/state/threadSettled.ts | 16 +- .../src/state/threadSnoozed.test.ts | 12 ++ packages/contracts/src/orchestration.ts | 58 ++++++ 38 files changed, 1018 insertions(+), 97 deletions(-) create mode 100644 apps/server/src/mcp/toolkits/threadAttention/handlers.test.ts create mode 100644 apps/server/src/mcp/toolkits/threadAttention/handlers.ts create mode 100644 apps/server/src/mcp/toolkits/threadAttention/tools.ts create mode 100644 apps/server/src/orchestration/decider.attention.test.ts create mode 100644 apps/server/src/orchestration/projector.attention.test.ts create mode 100644 apps/server/src/persistence/Migrations/048_ProjectionThreadAttention.test.ts create mode 100644 apps/server/src/persistence/Migrations/048_ProjectionThreadAttention.ts diff --git a/apps/mobile/src/features/threads/thread-list-v2-items.tsx b/apps/mobile/src/features/threads/thread-list-v2-items.tsx index 97c13de56aab..2903fc5f3c44 100644 --- a/apps/mobile/src/features/threads/thread-list-v2-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-v2-items.tsx @@ -55,6 +55,7 @@ const STATUS_LABEL_BY_STATUS: Partial< > = { approval: { label: "Approval", className: "text-adaptive-amber-700-300" }, input: { label: "Input", className: "text-adaptive-indigo-600-300" }, + question: { label: "? Question", className: "text-adaptive-violet-700-300" }, working: { label: "Working", className: "text-adaptive-sky-600-400" }, failed: { label: "Failed", className: "text-adaptive-red-700-300" }, }; diff --git a/apps/mobile/src/features/threads/threadListV2.test.ts b/apps/mobile/src/features/threads/threadListV2.test.ts index 33ae27cc0638..e6cdf1d49def 100644 --- a/apps/mobile/src/features/threads/threadListV2.test.ts +++ b/apps/mobile/src/features/threads/threadListV2.test.ts @@ -151,6 +151,28 @@ describe("resolveThreadListV2Status", () => { expect(resolveThreadListV2Status(thread)).toBe("approval"); }); + it("shows question attention before active work", () => { + expect( + resolveThreadListV2Status( + makeThread({ + id: ThreadId.make("t"), + title: "t", + attention: { kind: "question", raisedAt: NOW }, + session: { + threadId: ThreadId.make("t"), + status: "running", + providerName: "Codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: NOW, + }, + }), + ), + ).toBe("question"); + }); + it("resolves ready for quiescent threads", () => { expect(resolveThreadListV2Status(makeThread({ id: ThreadId.make("t"), title: "t" }))).toBe( "ready", diff --git a/apps/mobile/src/features/threads/threadListV2.ts b/apps/mobile/src/features/threads/threadListV2.ts index 2b44851f9309..e25ce554cfa6 100644 --- a/apps/mobile/src/features/threads/threadListV2.ts +++ b/apps/mobile/src/features/threads/threadListV2.ts @@ -27,7 +27,7 @@ export { snoozeWakeLabel }; * (approval), "in motion" (working), and "broken" (failed). Ready is the * unlabeled resting state. */ -export type ThreadListV2Status = "approval" | "input" | "working" | "failed" | "ready"; +export type ThreadListV2Status = "approval" | "input" | "question" | "working" | "failed" | "ready"; export type ThreadListV2SwipeAction = "archive" | "settle" | "unsettle" | "snooze" | "unsnooze"; export function resolveThreadListV2SnoozeMenuSelection(input: { @@ -126,7 +126,10 @@ export function resolveThreadListV2Enabled(input: { } export function resolveThreadListV2Status( - thread: Pick, + thread: Pick< + EnvironmentThreadShell, + "attention" | "hasPendingApprovals" | "hasPendingUserInput" | "session" + >, ): ThreadListV2Status { if (thread.hasPendingApprovals) { return "approval"; @@ -134,6 +137,9 @@ export function resolveThreadListV2Status( if (thread.hasPendingUserInput) { return "input"; } + if (thread.attention?.kind === "question") { + return "question"; + } if (thread.session?.status === "running" || thread.session?.status === "starting") { return "working"; } diff --git a/apps/mobile/src/features/threads/threadPresentation.ts b/apps/mobile/src/features/threads/threadPresentation.ts index 59cf108a01dd..ec0a60c15408 100644 --- a/apps/mobile/src/features/threads/threadPresentation.ts +++ b/apps/mobile/src/features/threads/threadPresentation.ts @@ -5,6 +5,7 @@ import { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell"; export type ThreadStatusKind = | "pending-approval" | "awaiting-input" + | "question" | "working" | "connecting" | "error" @@ -62,6 +63,18 @@ export function resolveThreadStatus( }; } + if (thread.attention?.kind === "question") { + return { + kind: "question", + label: "Question", + pillClassName: "bg-adaptive-violet-500-a12-a16", + textClassName: "text-adaptive-violet-700-300", + iconColor: "#bf5af2", + iconBackground: "rgba(191,90,242,0.22)", + pulse: false, + }; + } + if (thread.session?.status === "running") { return { kind: "working", diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 44ca928e63bb..1e58bedfabe3 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -22,6 +22,8 @@ import { PreviewSnapshotToolkit, PreviewStandardToolkit, } from "./toolkits/preview/tools.ts"; +import { ThreadAttentionToolkitHandlersLive } from "./toolkits/threadAttention/handlers.ts"; +import { ThreadAttentionToolkit } from "./toolkits/threadAttention/tools.ts"; const unauthorized = HttpServerResponse.jsonUnsafe( { @@ -215,11 +217,26 @@ export const PreviewToolkitRegistrationLive = Layer.mergeAll( PreviewSnapshotRegistrationLive, ); -const McpTransportLive = McpServer.layerHttp({ - name: "T3 Code", - version: packageJson.version, - path: "/mcp", - protocols: [McpProtocol.v2025_06_18], -}).pipe(Layer.provide(McpAuthMiddlewareLive)); +const threadAttentionToolkitRegistration = () => + McpServer.toolkit(ThreadAttentionToolkit).pipe(Layer.provide(ThreadAttentionToolkitHandlersLive)); -export const layer = PreviewToolkitRegistrationLive.pipe(Layer.provideMerge(McpTransportLive)); +const makeMcpTransport = (path: "/mcp" | "/mcp/thread") => + McpServer.layerHttp({ + name: "T3 Code", + version: packageJson.version, + path, + protocols: [McpProtocol.v2025_06_18], + }).pipe(Layer.provide(McpAuthMiddlewareLive)); + +const FullToolkitLive = Layer.mergeAll( + PreviewToolkitRegistrationLive, + threadAttentionToolkitRegistration(), +).pipe(Layer.provideMerge(makeMcpTransport("/mcp"))); + +// Sessions created while agent browser access is disabled still receive the +// attention tools, but preview tools stay absent from discovery entirely. +const ThreadAttentionOnlyToolkitLive = threadAttentionToolkitRegistration().pipe( + Layer.provideMerge(makeMcpTransport("/mcp/thread")), +); + +export const layer = Layer.mergeAll(FullToolkitLive, ThreadAttentionOnlyToolkitLive); diff --git a/apps/server/src/mcp/McpSessionRegistry.test.ts b/apps/server/src/mcp/McpSessionRegistry.test.ts index 1d8aead99d0d..b60530daed43 100644 --- a/apps/server/src/mcp/McpSessionRegistry.test.ts +++ b/apps/server/src/mcp/McpSessionRegistry.test.ts @@ -74,6 +74,21 @@ it.effect("builds MCP endpoints from the bound server host", () => }), ); +it.effect("uses the attention-only endpoint when preview access is disabled", () => + Effect.gen(function* () { + const registry = yield* makeRegistry(() => 1_000); + const issued = yield* registry.issue({ + threadId: ThreadId.make("thread-attention-only"), + providerInstanceId: ProviderInstanceId.make("codex"), + enablePreview: false, + }); + expect(issued.config.endpoint).toBe("http://127.0.0.1:43123/mcp/thread"); + + const token = issued.config.authorizationHeader.replace(/^Bearer\s+/, ""); + expect((yield* registry.resolve(token))?.capabilities.has("preview")).toBe(false); + }), +); + it.effect("expires credentials once their session stops showing signs of life", () => Effect.gen(function* () { let timestamp = 1_000; diff --git a/apps/server/src/mcp/McpSessionRegistry.ts b/apps/server/src/mcp/McpSessionRegistry.ts index f19a4f4e8c49..07797b21b5af 100644 --- a/apps/server/src/mcp/McpSessionRegistry.ts +++ b/apps/server/src/mcp/McpSessionRegistry.ts @@ -14,6 +14,7 @@ import * as McpProviderSession from "./McpProviderSession.ts"; export interface McpCredentialRequest { readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId; + readonly enablePreview?: boolean; } export interface McpIssuedCredential { @@ -98,10 +99,10 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( const state = yield* SynchronizedRef.make({ records: new Map() }); const currentTimeMillis = options.now ? Effect.sync(options.now) : Clock.currentTimeMillis; const livenessWindowMs = options.livenessWindowMs ?? DEFAULT_LIVENESS_WINDOW_MS; - const endpoint = + const endpointBase = httpServer.address._tag === "TcpAddress" - ? `http://${getHttpMcpEndpointHost(httpServer.address.hostname)}:${httpServer.address.port}/mcp` - : "http://127.0.0.1/mcp"; + ? `http://${getHttpMcpEndpointHost(httpServer.address.hostname)}:${httpServer.address.port}` + : "http://127.0.0.1"; const hashToken = (token: string) => crypto @@ -128,7 +129,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( threadId: ThreadId.make(request.threadId), providerSessionId, providerInstanceId: ProviderInstanceId.make(request.providerInstanceId), - capabilities: new Set(["preview"]), + capabilities: request.enablePreview === false ? new Set() : new Set(["preview"]), issuedAt, }; yield* SynchronizedRef.update(state, ({ records }) => { @@ -142,7 +143,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( threadId: scope.threadId, providerSessionId, providerInstanceId: scope.providerInstanceId, - endpoint, + endpoint: `${endpointBase}${request.enablePreview === false ? "/mcp/thread" : "/mcp"}`, authorizationHeader: `Bearer ${rawToken}`, }, }; diff --git a/apps/server/src/mcp/toolkits/threadAttention/handlers.test.ts b/apps/server/src/mcp/toolkits/threadAttention/handlers.test.ts new file mode 100644 index 000000000000..c945ed525a76 --- /dev/null +++ b/apps/server/src/mcp/toolkits/threadAttention/handlers.test.ts @@ -0,0 +1,62 @@ +import { + EnvironmentId, + ProviderInstanceId, + ThreadId, + type OrchestrationCommand, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { OrchestrationEngineService } from "../../../orchestration/Services/OrchestrationEngine.ts"; +import { threadAttentionHandlers } from "./handlers.ts"; + +it.effect("dispatches attention commands only to the authenticated thread", () => { + const commands: Array = []; + const boundThreadId = ThreadId.make("bound-thread"); + const engine = OrchestrationEngineService.of({ + dispatch: (command) => + Effect.sync(() => { + commands.push(command); + return { sequence: commands.length }; + }), + readEvents: () => Stream.empty, + streamDomainEvents: Stream.empty, + subscribeDomainEvents: Effect.succeed(Stream.empty), + latestSequence: Effect.succeed(0), + }); + const invocation: McpInvocationContext.McpInvocationScope = { + environmentId: EnvironmentId.make("environment-1"), + threadId: boundThreadId, + providerSessionId: "provider-session-1", + providerInstanceId: ProviderInstanceId.make("codex"), + capabilities: new Set(), + issuedAt: 0, + }; + + const testLayer = Layer.mergeAll( + NodeServices.layer, + Layer.succeed(OrchestrationEngineService, engine), + Layer.succeed(McpInvocationContext.McpInvocationContext, invocation), + ); + + return Effect.gen(function* () { + const marked = yield* threadAttentionHandlers.set_thread_attention({ + kind: "question", + }); + const cleared = yield* threadAttentionHandlers.clear_thread_attention(); + + expect(marked.attention.kind).toBe("question"); + expect(cleared.attention).toBeNull(); + expect(commands.map((command) => command.type)).toEqual([ + "thread.attention.set", + "thread.attention.clear", + ]); + expect( + commands.every((command) => "threadId" in command && command.threadId === boundThreadId), + ).toBe(true); + }).pipe(Effect.provide(testLayer)); +}); diff --git a/apps/server/src/mcp/toolkits/threadAttention/handlers.ts b/apps/server/src/mcp/toolkits/threadAttention/handlers.ts new file mode 100644 index 000000000000..35174c98b764 --- /dev/null +++ b/apps/server/src/mcp/toolkits/threadAttention/handlers.ts @@ -0,0 +1,57 @@ +import { CommandId, ThreadAttentionToolError } from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import { OrchestrationEngineService } from "../../../orchestration/Services/OrchestrationEngine.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { ThreadAttentionToolkit } from "./tools.ts"; + +const dispatchFailure = () => + new ThreadAttentionToolError({ + message: "Could not update this thread's attention status.", + }); + +export const threadAttentionHandlers = { + set_thread_attention: ({ kind }) => + Effect.gen(function* () { + const invocation = yield* McpInvocationContext.McpInvocationContext; + const engine = yield* OrchestrationEngineService; + const crypto = yield* Crypto.Crypto; + const createdAt = DateTime.formatIso(yield* DateTime.now); + const attention = { kind, raisedAt: createdAt } as const; + yield* engine + .dispatch({ + type: "thread.attention.set", + commandId: CommandId.make( + `mcp:thread-attention:${yield* crypto.randomUUIDv4.pipe(Effect.orDie)}`, + ), + threadId: invocation.threadId, + attention, + createdAt, + }) + .pipe(Effect.mapError(dispatchFailure)); + return { attention }; + }), + clear_thread_attention: () => + Effect.gen(function* () { + const invocation = yield* McpInvocationContext.McpInvocationContext; + const engine = yield* OrchestrationEngineService; + const crypto = yield* Crypto.Crypto; + const createdAt = DateTime.formatIso(yield* DateTime.now); + yield* engine + .dispatch({ + type: "thread.attention.clear", + commandId: CommandId.make( + `mcp:thread-attention:${yield* crypto.randomUUIDv4.pipe(Effect.orDie)}`, + ), + threadId: invocation.threadId, + createdAt, + }) + .pipe(Effect.mapError(dispatchFailure)); + return { attention: null }; + }), +} satisfies Parameters[0]; + +export const ThreadAttentionToolkitHandlersLive = + ThreadAttentionToolkit.toLayer(threadAttentionHandlers); diff --git a/apps/server/src/mcp/toolkits/threadAttention/tools.ts b/apps/server/src/mcp/toolkits/threadAttention/tools.ts new file mode 100644 index 000000000000..cc5a9521073a --- /dev/null +++ b/apps/server/src/mcp/toolkits/threadAttention/tools.ts @@ -0,0 +1,45 @@ +import { ThreadAttention, ThreadAttentionToolError } from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as Schema from "effect/Schema"; +import { Tool, Toolkit } from "effect/unstable/ai"; + +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { OrchestrationEngineService } from "../../../orchestration/Services/OrchestrationEngine.ts"; + +const dependencies = [ + McpInvocationContext.McpInvocationContext, + OrchestrationEngineService, + Crypto.Crypto, +]; +const result = Schema.Struct({ attention: Schema.NullOr(ThreadAttention) }); + +export const SetThreadAttentionTool = Tool.make("set_thread_attention", { + description: + "Mark this thread as waiting for the user to answer a question. Call this immediately before ending a turn only when your final response contains a question that blocks useful progress. The thread is derived from your authenticated session; never identify a thread yourself.", + parameters: Schema.Struct({ kind: Schema.Literal("question") }), + success: result, + failure: ThreadAttentionToolError, + dependencies, +}) + .annotate(Tool.Title, "Mark thread as awaiting an answer") + .annotate(Tool.Readonly, false) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true); + +export const ClearThreadAttentionTool = Tool.make("clear_thread_attention", { + description: + "Clear a question marker you previously set when the question was withdrawn or no answer is required. A user reply clears the marker automatically.", + parameters: Schema.Record(Schema.String, Schema.Never), + success: result, + failure: ThreadAttentionToolError, + dependencies, +}) + .annotate(Tool.Title, "Clear thread attention") + .annotate(Tool.Readonly, false) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true); + +export const ThreadAttentionToolkit = Toolkit.make( + SetThreadAttentionTool, + ClearThreadAttentionTool, +); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 1b8a451175f3..7a3fc614ee40 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -324,6 +324,58 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { unsettledAt: "2026-01-01T00:00:02.000Z", }, ]); + + const raisedAt = "2026-01-01T00:00:03.000Z"; + yield* eventStore.append({ + type: "thread.attention-set", + eventId: EventId.make("evt-attention-set"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + occurredAt: raisedAt, + commandId: CommandId.make("cmd-attention-set"), + causationEventId: null, + correlationId: CommandId.make("cmd-attention-set"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-1"), + attention: { kind: "question", raisedAt }, + updatedAt: raisedAt, + }, + }); + yield* projectionPipeline.bootstrap; + + let attentionRows = yield* sql<{ readonly attention: string | null }>` + SELECT attention_json AS "attention" + FROM projection_threads + WHERE thread_id = 'thread-1' + `; + assert.deepEqual(attentionRows, [ + { attention: '{"kind":"question","raisedAt":"2026-01-01T00:00:03.000Z"}' }, + ]); + + yield* eventStore.append({ + type: "thread.attention-cleared", + eventId: EventId.make("evt-attention-clear"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + occurredAt: "2026-01-01T00:00:04.000Z", + commandId: CommandId.make("cmd-attention-clear"), + causationEventId: null, + correlationId: CommandId.make("cmd-attention-clear"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-1"), + updatedAt: "2026-01-01T00:00:04.000Z", + }, + }); + yield* projectionPipeline.bootstrap; + + attentionRows = yield* sql<{ readonly attention: string | null }>` + SELECT attention_json AS "attention" + FROM projection_threads + WHERE thread_id = 'thread-1' + `; + assert.deepEqual(attentionRows, [{ attention: null }]); }), ); }); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 1ba1b6afa4f3..2a85a0615307 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -650,6 +650,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti pinOrderKey: null, titleRegenerationRequestId: null, titleRegenerationStartedAt: null, + attention: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, @@ -811,6 +812,32 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } + case "thread.attention-set": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) return; + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + attention: event.payload.attention, + updatedAt: event.payload.updatedAt, + }); + return; + } + + case "thread.attention-cleared": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) return; + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + attention: null, + updatedAt: event.payload.updatedAt, + }); + return; + } + case "thread.meta-updated": { const existingRow = yield* projectionThreadRepository.getById({ threadId: event.payload.threadId, diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 682b280bf4d0..12b135abd917 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -91,6 +91,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { has_actionable_proposed_plan, pinned_at, pin_order_key, + attention_json, created_at, updated_at, deleted_at @@ -112,6 +113,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { 0, '2026-02-24T00:00:01.000Z', 'gm', + '{"kind":"question","raisedAt":"2026-02-24T00:00:02.500Z"}', '2026-02-24T00:00:02.000Z', '2026-02-24T00:00:03.000Z', NULL @@ -338,6 +340,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { pinnedAt: "2026-02-24T00:00:01.000Z", pinOrderKey: "gm", titleRegeneration: null, + attention: { kind: "question", raisedAt: "2026-02-24T00:00:02.500Z" }, deletedAt: null, messages: [ { @@ -466,6 +469,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { pinnedAt: "2026-02-24T00:00:01.000Z", pinOrderKey: "gm", titleRegeneration: null, + attention: { kind: "question", raisedAt: "2026-02-24T00:00:02.500Z" }, session: { threadId: ThreadId.make("thread-1"), status: "running", diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 0d064a1d6e9c..6f2467057b6b 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -27,6 +27,7 @@ import { ModelSelection, ProjectId, ThreadLinkedPullRequest, + ThreadAttention, ThreadId, } from "@t3tools/contracts"; import * as Arr from "effect/Array"; @@ -101,6 +102,7 @@ const ProjectionThreadDbRowSchema = ProjectionThread.mapFields( Struct.assign({ modelSelection: Schema.fromJsonString(ModelSelection), linkedPullRequest: Schema.NullOr(Schema.fromJsonString(ThreadLinkedPullRequest)), + attention: Schema.NullOr(Schema.fromJsonString(ThreadAttention)), }), ); const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( @@ -485,6 +487,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -523,6 +526,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -563,6 +567,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -1025,6 +1030,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -1957,6 +1963,7 @@ pending_approval_requests AS ( pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), + attention: row.attention, deletedAt: row.deletedAt, messages: messagesByThread.get(row.threadId) ?? [], proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], @@ -2170,6 +2177,7 @@ pending_approval_requests AS ( pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), + attention: row.attention, deletedAt: row.deletedAt, messages: [], proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], @@ -2310,6 +2318,7 @@ pending_approval_requests AS ( pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), + attention: row.attention, session: sessionByThread.get(row.threadId) ?? null, latestUserMessageAt: row.latestUserMessageAt, hasPendingApprovals: row.pendingApprovalCount > 0, @@ -2458,6 +2467,7 @@ pending_approval_requests AS ( pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), + attention: row.attention, session: sessionByThread.get(row.threadId) ?? null, latestUserMessageAt: row.latestUserMessageAt, hasPendingApprovals: row.pendingApprovalCount > 0, @@ -2752,6 +2762,7 @@ pending_approval_requests AS ( pinnedAt: threadRow.value.pinnedAt, pinOrderKey: threadRow.value.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(threadRow.value), + attention: threadRow.value.attention, session: Option.isSome(sessionRow) ? mapSessionRow(sessionRow.value) : null, latestUserMessageAt: threadRow.value.latestUserMessageAt, hasPendingApprovals: threadRow.value.pendingApprovalCount > 0, @@ -2990,6 +3001,7 @@ pending_approval_requests AS ( pinnedAt: threadRow.value.pinnedAt, pinOrderKey: threadRow.value.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(threadRow.value), + attention: threadRow.value.attention, deletedAt: null, messages: messageRows.map((row) => { const message = { diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 7e866cf89592..046fc67ee25d 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -22,6 +22,8 @@ import { ThreadTurnDiffCompletedPayload as ContractsThreadTurnDiffCompletedPayloadSchema, ThreadRevertedPayload as ContractsThreadRevertedPayloadSchema, ThreadActivityAppendedPayload as ContractsThreadActivityAppendedPayloadSchema, + ThreadAttentionSetPayload as ContractsThreadAttentionSetPayloadSchema, + ThreadAttentionClearedPayload as ContractsThreadAttentionClearedPayloadSchema, ThreadTurnStartRequestedPayload as ContractsThreadTurnStartRequestedPayloadSchema, ThreadTurnInterruptRequestedPayload as ContractsThreadTurnInterruptRequestedPayloadSchema, ThreadApprovalResponseRequestedPayload as ContractsThreadApprovalResponseRequestedPayloadSchema, @@ -55,6 +57,8 @@ export const ThreadSessionSetPayload = ContractsThreadSessionSetPayloadSchema; export const ThreadTurnDiffCompletedPayload = ContractsThreadTurnDiffCompletedPayloadSchema; export const ThreadRevertedPayload = ContractsThreadRevertedPayloadSchema; export const ThreadActivityAppendedPayload = ContractsThreadActivityAppendedPayloadSchema; +export const ThreadAttentionSetPayload = ContractsThreadAttentionSetPayloadSchema; +export const ThreadAttentionClearedPayload = ContractsThreadAttentionClearedPayloadSchema; export const ThreadTurnStartRequestedPayload = ContractsThreadTurnStartRequestedPayloadSchema; export const ThreadTurnInterruptRequestedPayload = diff --git a/apps/server/src/orchestration/ThreadSettlementPolicy.test.ts b/apps/server/src/orchestration/ThreadSettlementPolicy.test.ts index a70fb9f68176..e821b118cbbc 100644 --- a/apps/server/src/orchestration/ThreadSettlementPolicy.test.ts +++ b/apps/server/src/orchestration/ThreadSettlementPolicy.test.ts @@ -153,6 +153,9 @@ describe("resolveAutoSettlementAt", () => { expect(decide(makeThread({ snoozedUntil: "2026-08-29T00:00:00.000Z" }))).toBe(false); expect(decide(makeThread({ hasPendingApprovals: true }))).toBe(false); expect(decide(makeThread({ hasPendingUserInput: true }))).toBe(false); + expect( + decide(makeThread({ attention: { kind: "question", raisedAt: "2026-08-20T00:00:00.000Z" } })), + ).toBe(false); expect(decide(makeThread({ backgroundLiveness: "working" }))).toBe(false); expect(decide(makeThread({ backgroundLiveness: "monitoring" }))).toBe(false); expect( diff --git a/apps/server/src/orchestration/ThreadSettlementPolicy.ts b/apps/server/src/orchestration/ThreadSettlementPolicy.ts index 5270c30eebe2..c0b7764a941b 100644 --- a/apps/server/src/orchestration/ThreadSettlementPolicy.ts +++ b/apps/server/src/orchestration/ThreadSettlementPolicy.ts @@ -93,6 +93,7 @@ export function resolveAutoSettlementAt(input: { export function isAutoSettlementCandidate(thread: OrchestrationThreadShell, now: string): boolean { if (thread.archivedAt !== null || thread.settledOverride !== null) return false; if (thread.hasPendingApprovals || thread.hasPendingUserInput) return false; + if (thread.attention != null) return false; if (thread.session?.status === "starting" || thread.session?.status === "running") return false; if (thread.backgroundLiveness != null) return false; if (threadHasQueuedTurnStart(thread, now)) return false; diff --git a/apps/server/src/orchestration/decider.attention.test.ts b/apps/server/src/orchestration/decider.attention.test.ts new file mode 100644 index 000000000000..de8ebdfd746e --- /dev/null +++ b/apps/server/src/orchestration/decider.attention.test.ts @@ -0,0 +1,166 @@ +import { + CommandId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationCommand, + type OrchestrationReadModel, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { decideOrchestrationCommand } from "./decider.ts"; + +const NOW = "2026-01-01T00:00:00.000Z"; +const threadId = ThreadId.make("thread-1"); + +function makeReadModel( + overrides: Partial = {}, +): OrchestrationReadModel { + return { + snapshotSequence: 0, + projects: [], + threads: [ + { + id: threadId, + projectId: ProjectId.make("project-1"), + title: "Thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.4", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: null, + createdAt: NOW, + updatedAt: NOW, + archivedAt: null, + settledOverride: null, + settledAt: null, + snoozedUntil: null, + snoozedAt: null, + attention: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + ...overrides, + }, + ], + updatedAt: NOW, + }; +} + +const decide = (command: OrchestrationCommand, readModel = makeReadModel()) => + decideOrchestrationCommand({ command, readModel }).pipe( + Effect.map((result) => (Array.isArray(result) ? result : [result])), + ); + +it.layer(NodeServices.layer)("thread attention decider", (it) => { + it.effect("sets question attention and wakes settled and snoozed threads", () => + Effect.gen(function* () { + const events = yield* decide( + { + type: "thread.attention.set", + commandId: CommandId.make("attention-set"), + threadId, + attention: { kind: "question", raisedAt: NOW }, + createdAt: NOW, + }, + makeReadModel({ + settledOverride: "settled", + settledAt: NOW, + snoozedUntil: "2026-01-02T00:00:00.000Z", + snoozedAt: NOW, + }), + ); + + expect(events.map((event) => event.type)).toEqual([ + "thread.unsettled", + "thread.unsnoozed", + "thread.attention-set", + ]); + }), + ); + + it.effect("clears question attention after the user's reply and before turn start", () => + Effect.gen(function* () { + const events = yield* decide( + { + type: "thread.turn.start", + commandId: CommandId.make("turn-start"), + threadId, + message: { + messageId: MessageId.make("message-1"), + role: "user", + text: "My answer", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + createdAt: NOW, + }, + makeReadModel({ attention: { kind: "question", raisedAt: NOW } }), + ); + + expect(events.map((event) => event.type)).toEqual([ + "thread.message-sent", + "thread.attention-cleared", + "thread.turn-start-requested", + ]); + }), + ); + + it.effect("manual settle dismisses attention but automatic settle is blocked", () => + Effect.gen(function* () { + const readModel = makeReadModel({ + attention: { kind: "question", raisedAt: NOW }, + }); + const manualEvents = yield* decide( + { + type: "thread.settle", + commandId: CommandId.make("settle"), + threadId, + }, + readModel, + ); + expect(manualEvents.map((event) => event.type)).toEqual([ + "thread.settled", + "thread.attention-cleared", + ]); + + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-settle", + commandId: CommandId.make("auto-settle"), + threadId, + snapshotSequence: 0, + settledAt: NOW, + }, + readModel, + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationThreadSettleBlockedError"); + }), + ); + + it.effect("does not snooze a thread that is waiting for an answer", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.snooze", + commandId: CommandId.make("snooze"), + threadId, + snoozedUntil: "1970-01-02T00:00:00.000Z", + }, + readModel: makeReadModel({ attention: { kind: "question", raisedAt: NOW } }), + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index b336053ac9e1..dcfa40aff623 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -446,6 +446,9 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }), ); } + if (command.type === "thread.auto-settle" && thread.attention != null) { + return yield* new OrchestrationThreadSettleBlockedError({ threadId: command.threadId }); + } // The server owns settle eligibility. A stale command must not settle // a thread whose session is coming alive or working. if (thread.session?.status === "starting" || thread.session?.status === "running") { @@ -460,7 +463,9 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" const occurredAt = yield* nowIso; // Settling inside the adoption window would hide just-requested work. if (hasQueuedTurnStartForThread(thread, occurredAt)) { - return yield* new OrchestrationThreadSettleBlockedError({ threadId: command.threadId }); + return yield* new OrchestrationThreadSettleBlockedError({ + threadId: command.threadId, + }); } // Settling an already-settled thread re-emits with the original // settledAt: the engine rejects zero-event commands, and bulk-settle / @@ -521,6 +526,21 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }, }); } + if (thread.attention != null) { + companionEvents.push({ + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.attention-cleared", + payload: { + threadId: command.threadId, + updatedAt: occurredAt, + }, + }); + } return companionEvents.length > 0 ? [settledEvent, ...companionEvents] : settledEvent; } @@ -576,11 +596,11 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" // user-input request is the agent waiting on the user, and hiding it // defeats the request. (A running session IS snoozable — snooze only // affects visibility, never the agent.) - if (hasOpenBlockingRequest(thread)) { + if (hasOpenBlockingRequest(thread) || thread.attention != null) { return yield* Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, - detail: `thread ${command.threadId} has a pending approval or user-input request and cannot be snoozed`, + detail: `thread ${command.threadId} needs user attention and cannot be snoozed`, }), ); } @@ -1014,7 +1034,28 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }, }); } - return [...lifecycleResetEvents, userMessageEvent, turnStartRequestedEvent]; + const attentionClearedEvent: Omit | null = + targetThread.attention == null + ? null + : { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.attention-cleared", + payload: { + threadId: command.threadId, + updatedAt: command.createdAt, + }, + }; + return [ + ...lifecycleResetEvents, + userMessageEvent, + ...(attentionClearedEvent === null ? [] : [attentionClearedEvent]), + turnStartRequestedEvent, + ]; } case "thread.turn.interrupt": { @@ -1448,6 +1489,86 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" return [unsettledEvent, activityAppendedEvent]; } + case "thread.attention.set": { + const thread = yield* requireThreadNotArchived({ + readModel, + command, + threadId: command.threadId, + }); + const unchanged = thread.attention?.kind === command.attention.kind; + const unchangedAndVisible = + unchanged && thread.settledOverride === null && thread.snoozedUntil == null; + const attentionSetEvent: Omit = { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.attention-set", + payload: { + threadId: command.threadId, + attention: unchanged ? thread.attention! : command.attention, + updatedAt: unchangedAndVisible ? thread.updatedAt : command.createdAt, + }, + }; + const wakeEvents: Array> = []; + if (thread.settledOverride !== null) { + wakeEvents.push({ + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.unsettled", + payload: { + threadId: command.threadId, + reason: "activity", + updatedAt: command.createdAt, + }, + }); + } + if (thread.snoozedUntil != null) { + wakeEvents.push({ + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.unsnoozed", + payload: { + threadId: command.threadId, + reason: "activity", + updatedAt: command.createdAt, + }, + }); + } + return [...wakeEvents, attentionSetEvent]; + } + + case "thread.attention.clear": { + const thread = yield* requireThreadNotArchived({ + readModel, + command, + threadId: command.threadId, + }); + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.attention-cleared", + payload: { + threadId: command.threadId, + updatedAt: thread.attention == null ? thread.updatedAt : command.createdAt, + }, + }; + } + default: { command satisfies never; const fallback = command as never as { type: string }; diff --git a/apps/server/src/orchestration/projector.attention.test.ts b/apps/server/src/orchestration/projector.attention.test.ts new file mode 100644 index 000000000000..296a9a5624e5 --- /dev/null +++ b/apps/server/src/orchestration/projector.attention.test.ts @@ -0,0 +1,70 @@ +import { + CommandId, + EventId, + ProjectId, + ThreadId, + type OrchestrationEvent, +} from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { createEmptyReadModel, projectEvent } from "./projector.ts"; + +const now = "2026-01-01T00:00:00.000Z"; +const threadId = ThreadId.make("thread-1"); +const makeEvent = ( + sequence: number, + type: OrchestrationEvent["type"], + payload: unknown, +): OrchestrationEvent => + ({ + sequence, + eventId: EventId.make(`event-${sequence}`), + type, + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make(`command-${sequence}`), + causationEventId: null, + correlationId: null, + metadata: {}, + payload: payload as never, + }) as OrchestrationEvent; + +it.effect("projects the thread attention lifecycle", () => + Effect.gen(function* () { + const created = yield* projectEvent( + createEmptyReadModel(now), + makeEvent(1, "thread.created", { + threadId, + projectId: ProjectId.make("project-1"), + title: "Thread", + modelSelection: { provider: "codex", model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }), + ); + expect(created.threads[0]?.attention).toBeNull(); + + const attention = { kind: "question", raisedAt: now } as const; + const marked = yield* projectEvent( + created, + makeEvent(2, "thread.attention-set", { + threadId, + attention, + updatedAt: now, + }), + ); + expect(marked.threads[0]?.attention).toEqual(attention); + + const cleared = yield* projectEvent( + marked, + makeEvent(3, "thread.attention-cleared", { threadId, updatedAt: now }), + ); + expect(cleared.threads[0]?.attention).toBeNull(); + }), +); diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 3cea194bbb44..8d95a6954265 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -16,6 +16,8 @@ import { ProjectDeletedPayload, ProjectMetaUpdatedPayload, ThreadActivityAppendedPayload, + ThreadAttentionClearedPayload, + ThreadAttentionSetPayload, ThreadArchivedPayload, ThreadCreatedPayload, ThreadDeletedPayload, @@ -335,6 +337,7 @@ export function projectEvent( unsettledAt: null, snoozedUntil: null, snoozedAt: null, + attention: null, deletedAt: null, messages: [], activities: [], @@ -483,6 +486,33 @@ export function projectEvent( })), ); + case "thread.attention-set": + return decodeForEvent(ThreadAttentionSetPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + attention: payload.attention, + updatedAt: payload.updatedAt, + }), + })), + ); + + case "thread.attention-cleared": + return decodeForEvent( + ThreadAttentionClearedPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + attention: null, + updatedAt: payload.updatedAt, + }), + })), + ); + case "thread.meta-updated": return decodeForEvent(ThreadMetaUpdatedPayload, event.payload, event.type, "payload").pipe( Effect.map((payload) => ({ diff --git a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts index adc3ca40cbb5..70059fad2ee9 100644 --- a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts @@ -99,6 +99,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { snoozedUntil: null, snoozedAt: null, pinnedAt: null, + attention: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, @@ -163,6 +164,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { snoozedUntil: "2026-03-26T09:00:00.000Z", snoozedAt: "2026-03-25T00:00:00.000Z", pinnedAt: "2026-03-25T00:00:00.000Z", + attention: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, @@ -240,6 +242,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { snoozedUntil: null, snoozedAt: null, pinnedAt: null, + attention: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index d5653a2c8b42..5796178f4609 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -14,12 +14,13 @@ import { ProjectionThreadRepository, type ProjectionThreadRepositoryShape, } from "../Services/ProjectionThreads.ts"; -import { ModelSelection, ThreadLinkedPullRequest } from "@t3tools/contracts"; +import { ModelSelection, ThreadAttention, ThreadLinkedPullRequest } from "@t3tools/contracts"; const ProjectionThreadDbRow = ProjectionThread.mapFields( Struct.assign({ modelSelection: Schema.fromJsonString(ModelSelection), linkedPullRequest: Schema.NullOr(Schema.fromJsonString(ThreadLinkedPullRequest)), + attention: Schema.NullOr(Schema.fromJsonString(ThreadAttention)), }), ); type ProjectionThreadDbRow = typeof ProjectionThreadDbRow.Type; @@ -54,6 +55,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { pin_order_key, title_regeneration_request_id, title_regeneration_started_at, + attention_json, latest_user_message_at, pending_approval_count, pending_user_input_count, @@ -83,6 +85,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.pinOrderKey ?? null}, ${row.titleRegenerationRequestId ?? null}, ${row.titleRegenerationStartedAt ?? null}, + ${row.attention === null ? null : JSON.stringify(row.attention)}, ${row.latestUserMessageAt}, ${row.pendingApprovalCount}, ${row.pendingUserInputCount}, @@ -112,6 +115,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { pin_order_key = excluded.pin_order_key, title_regeneration_request_id = excluded.title_regeneration_request_id, title_regeneration_started_at = excluded.title_regeneration_started_at, + attention_json = excluded.attention_json, latest_user_message_at = excluded.latest_user_message_at, pending_approval_count = excluded.pending_approval_count, pending_user_input_count = excluded.pending_user_input_count, @@ -148,6 +152,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -186,6 +191,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 92dc18291057..5c172d2c7a66 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -59,6 +59,7 @@ import Migration0044 from "./Migrations/044_ClearAutomaticProjectModelDefaults.t import Migration0045 from "./Migrations/045_ProjectionProjectsAutoPull.ts"; import Migration0046 from "./Migrations/046_RepairAutomaticSettlementTimestamps.ts"; import Migration0047 from "./Migrations/047_ProjectionProjectIcon.ts"; +import Migration0048 from "./Migrations/048_ProjectionThreadAttention.ts"; /** * Migration loader with all migrations defined inline. @@ -118,6 +119,7 @@ export const migrationEntries = [ [45, "ProjectionProjectsAutoPull", Migration0045], [46, "RepairAutomaticSettlementTimestamps", Migration0046], [47, "ProjectionProjectIcon", Migration0047], + [48, "ProjectionThreadAttention", Migration0048], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/048_ProjectionThreadAttention.test.ts b/apps/server/src/persistence/Migrations/048_ProjectionThreadAttention.test.ts new file mode 100644 index 000000000000..0a02ae249ca5 --- /dev/null +++ b/apps/server/src/persistence/Migrations/048_ProjectionThreadAttention.test.ts @@ -0,0 +1,27 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; + +it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory()))("048_ProjectionThreadAttention", (it) => { + it.effect("adds nullable attention JSON to thread projections", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 47 }); + yield* runMigrations({ toMigrationInclusive: 48 }); + + const columns = yield* sql<{ + readonly name: string; + readonly notnull: number; + }>` + PRAGMA table_info(projection_threads) + `; + const attention = columns.find((column) => column.name === "attention_json"); + assert.equal(attention?.name, "attention_json"); + assert.equal(attention?.notnull, 0); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/048_ProjectionThreadAttention.ts b/apps/server/src/persistence/Migrations/048_ProjectionThreadAttention.ts new file mode 100644 index 000000000000..1ba9b5891b62 --- /dev/null +++ b/apps/server/src/persistence/Migrations/048_ProjectionThreadAttention.ts @@ -0,0 +1,16 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + + if (!columns.some((column) => column.name === "attention_json")) { + yield* sql` + ALTER TABLE projection_threads + ADD COLUMN attention_json TEXT + `; + } +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index a70548bc110c..f883238e61aa 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -15,6 +15,7 @@ import { ProviderInteractionMode, RuntimeMode, ThreadLinkedPullRequest, + ThreadAttention, ThreadId, TurnId, } from "@t3tools/contracts"; @@ -48,6 +49,7 @@ export const ProjectionThread = Schema.Struct({ pinOrderKey: Schema.optional(Schema.NullOr(Schema.String)), titleRegenerationRequestId: Schema.optional(Schema.NullOr(CommandId)), titleRegenerationStartedAt: Schema.optional(Schema.NullOr(IsoDateTime)), + attention: Schema.NullOr(ThreadAttention), latestUserMessageAt: Schema.NullOr(IsoDateTime), pendingApprovalCount: NonNegativeInt, pendingUserInputCount: NonNegativeInt, diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index fb997b1ab82a..da4ea9d28c78 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -2868,11 +2868,13 @@ boundedListing.layer("ProviderServiceLive session listing", (it) => { }); describe("agent browser access", () => { - const revokedThreads: Array = []; - const startSessionWith = (enableAgentBrowserAccess: boolean, threadId: ThreadId) => Effect.gen(function* () { - const issued: Array = []; + const issued: Array<{ + threadId: ThreadId; + providerInstanceId: ProviderInstanceId; + enablePreview?: boolean; + }> = []; const codex = makeFakeCodexAdapter(); const providerAdapterLayer = Layer.succeed( ProviderAdapterRegistry.ProviderAdapterRegistry, @@ -2887,10 +2889,9 @@ describe("agent browser access", () => { const providerLayer = makeProviderServiceLive({ issueMcpCredential: (request) => Effect.sync(() => { - issued.push(request.threadId); + issued.push(request); return undefined; }), - revokeMcpCredential: (revoked) => Effect.sync(() => void revokedThreads.push(revoked)), }).pipe( Layer.provide(providerAdapterLayer), Layer.provide(directoryLayer), @@ -2918,28 +2919,18 @@ describe("agent browser access", () => { return issued; }); - // Credential issuance is the observable that matters: it is the only place a - // credential is minted, and `/mcp` accepts nothing else, so withholding it is - // what actually denies every provider and external MCP client. - it.effect("requests no MCP credential when agent browser access is off", () => - Effect.gen(function* () { - const issued = yield* startSessionWith(false, asThreadId("thread-browser-off")); - - assert.deepEqual(issued, []); - }).pipe(Effect.provide(NodeServices.layer)), - ); - - it.effect("revokes an already-issued credential when access is off", () => + it.effect("requests an MCP credential without preview when agent browser access is off", () => Effect.gen(function* () { - const threadId = asThreadId("thread-browser-revoke"); - revokedThreads.length = 0; + const threadId = asThreadId("thread-browser-off"); + const issued = yield* startSessionWith(false, threadId); - yield* startSessionWith(false, threadId); - - // Clearing the in-memory map is not enough: a token issued before the - // toggle flipped stays valid against `/mcp` for its whole liveness - // window, and later turns refresh it. - assert.deepEqual(revokedThreads, [threadId]); + assert.deepEqual(issued, [ + { + threadId, + providerInstanceId: codexInstanceId, + enablePreview: false, + }, + ]); }).pipe(Effect.provide(NodeServices.layer)), ); @@ -2949,7 +2940,9 @@ describe("agent browser access", () => { const issued = yield* startSessionWith(true, threadId); - assert.deepEqual(issued, [threadId]); + assert.deepEqual(issued, [ + { threadId, providerInstanceId: codexInstanceId, enablePreview: true }, + ]); }).pipe(Effect.provide(NodeServices.layer)), ); }); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 7171430d636b..d46c15e37f10 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -92,8 +92,6 @@ export interface ProviderServiceLiveOptions { * test see whether a credential was requested at all. */ readonly issueMcpCredential?: typeof McpSessionRegistry.issueActiveMcpCredential; - /** Same seam as `issueMcpCredential`, for observing the deny path's revoke. */ - readonly revokeMcpCredential?: typeof McpSessionRegistry.revokeActiveMcpThread; } type ProviderServiceMethod = @@ -246,8 +244,6 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const serverSettings = yield* ServerSettings.ServerSettingsService; const issueMcpCredential = options?.issueMcpCredential ?? McpSessionRegistry.issueActiveMcpCredential; - const revokeMcpCredential = - options?.revokeMcpCredential ?? McpSessionRegistry.revokeActiveMcpThread; const runtimeEventPubSub = yield* PubSub.unbounded(); const pendingCompactions = new Map(); const timedOutNativeCompactions = new Set(); @@ -259,14 +255,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( return true; }); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); - /** - * Attach the `t3-code` MCP server to the session that is about to start. - * - * This is the only place a credential is minted, so withholding one here is - * what disables agent browser access everywhere: every adapter already - * treats a missing session as "no MCP server", and the `/mcp` endpoint - * accepts nothing but tokens issued from this path. - */ + /** Attach the thread-scoped `t3-code` MCP server to the next session. */ /** * Deny on an unreadable settings file rather than letting the read failure * escape: adding `ServerSettingsError` to `ProviderServiceError` would widen @@ -287,18 +276,12 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) => Effect.gen(function* () { - if (!(yield* agentBrowserAccessEnabled)) { - // Revoke as well as clear. Every other prepare path reaches - // `issueActiveMcpCredential`, which revokes the thread first, so - // skipping it here would leave a previously issued bearer token valid - // against `/mcp` for the rest of its liveness window — and later turns - // would keep refreshing it. A session restart (runtime mode, cwd, - // model) re-prepares without stopping, so it relies on this. - yield* revokeMcpCredential(threadId); - yield* Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId)); - return undefined; - } - const credential = yield* issueMcpCredential({ threadId, providerInstanceId }); + const enablePreview = yield* agentBrowserAccessEnabled; + const credential = yield* issueMcpCredential({ + threadId, + providerInstanceId, + enablePreview, + }); if (credential) { yield* Effect.sync(() => McpProviderSession.setMcpProviderSession(credential.config)); } diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index dbf8fcf78532..042b2bae29fb 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -1170,6 +1170,26 @@ describe("resolveThreadStatusPill", () => { ).toMatchObject({ label: "Awaiting Input", pulse: false }); }); + it("shows a question after native input and before active work", () => { + expect( + resolveThreadStatusPill({ + thread: { + ...baseThread, + attention: { kind: "question", raisedAt: "2026-03-09T10:00:00.000Z" }, + }, + }), + ).toMatchObject({ label: "Question", marker: "?", pulse: false }); + expect( + resolveThreadStatusPill({ + thread: { + ...baseThread, + hasPendingApprovals: true, + attention: { kind: "question", raisedAt: "2026-03-09T10:00:00.000Z" }, + }, + }), + ).toMatchObject({ label: "Pending Approval" }); + }); + it("falls back to working when the thread is actively running without blockers", () => { expect( resolveThreadStatusPill({ diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index b5bb7bc77cc9..28ecd4b83cbf 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -185,18 +185,21 @@ export interface ThreadStatusPill { | "Completed" | "Pending Approval" | "Awaiting Input" + | "Question" | "Plan Ready"; colorClass: string; dotClass: string; pulse: boolean; + marker?: string; } // Rollup order mirrors the per-thread resolver exactly: attention states, // then active work, then the actionable plan prompt, then passive // monitoring. A Monitoring sibling must never hide a Plan Ready thread. const THREAD_STATUS_PRIORITY: Record = { - "Pending Approval": 6, - "Awaiting Input": 5, + "Pending Approval": 7, + "Awaiting Input": 6, + Question: 5, Working: 4, Connecting: 4, "Plan Ready": 3, @@ -206,6 +209,7 @@ const THREAD_STATUS_PRIORITY: Record = { type ThreadStatusInput = Pick< SidebarThreadSummary, + | "attention" | "hasActionableProposedPlan" | "hasPendingApprovals" | "hasPendingUserInput" @@ -515,6 +519,7 @@ export function resolveThreadRowClassName(input: { export type SidebarThreadStatus = | "approval" | "input" + | "question" | "working" | "monitoring" | "failed" @@ -522,7 +527,7 @@ export type SidebarThreadStatus = type SidebarThreadStatusInput = Pick< SidebarThreadSummary, - "hasPendingApprovals" | "hasPendingUserInput" | "session" | "backgroundLiveness" + "attention" | "hasPendingApprovals" | "hasPendingUserInput" | "session" | "backgroundLiveness" >; export function resolveSidebarThreadStatus(thread: SidebarThreadStatusInput): SidebarThreadStatus { @@ -532,6 +537,10 @@ export function resolveSidebarThreadStatus(thread: SidebarThreadStatusInput): Si if (thread.hasPendingUserInput) { return "input"; } + if (thread.attention?.kind === "question") { + return "question"; + } + if (thread.session?.status === "running" || thread.session?.status === "starting") { return "working"; } @@ -725,6 +734,16 @@ export function resolveThreadStatusPill(input: { }; } + if (thread.attention?.kind === "question") { + return { + label: "Question", + colorClass: "text-violet-600 dark:text-violet-300/90", + dotClass: "bg-violet-500 dark:bg-violet-300/90", + pulse: false, + marker: "?", + }; + } + if (thread.session?.status === "running") { return { label: "Working", diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index cf14c9614734..648b03dff3cd 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -918,25 +918,31 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { icon: null, className: "text-indigo-600 dark:text-indigo-300", } - : status === "failed" + : status === "question" ? { - label: "Failed", - icon: null, - className: "text-red-700 dark:text-red-300", + label: "Question", + icon: "question" as const, + className: "text-violet-600 dark:text-violet-300", } - : isWoke + : status === "failed" ? { - label: "Woke", - icon: "woke" as const, - className: "text-amber-700 dark:text-amber-300", + label: "Failed", + icon: null, + className: "text-red-700 dark:text-red-300", } - : isUnread + : isWoke ? { - label: "Done", - icon: "done" as const, - className: "text-emerald-700 dark:text-emerald-300", + label: "Woke", + icon: "woke" as const, + className: "text-amber-700 dark:text-amber-300", } - : null; + : isUnread + ? { + label: "Done", + icon: "done" as const, + className: "text-emerald-700 dark:text-emerald-300", + } + : null; const isWokeStatus = topStatus?.icon === "woke"; const branchMismatch = resolveLocalCheckoutBranchMismatch({ @@ -1534,6 +1540,13 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { ) : topStatus.icon === "done" ? ( + ) : topStatus.icon === "question" ? ( + + ? + ) : null} {/* The label alone is the live region: a role="status" wrapper around the ticking duration would make diff --git a/apps/web/src/components/ThreadStatusIndicators.tsx b/apps/web/src/components/ThreadStatusIndicators.tsx index 1ad0c3139fd8..410966d2b3a4 100644 --- a/apps/web/src/components/ThreadStatusIndicators.tsx +++ b/apps/web/src/components/ThreadStatusIndicators.tsx @@ -488,11 +488,17 @@ export function ThreadStatusLabel({ /> } > - + {status.marker ? ( + + {status.marker} + + ) : ( + + )} {status.label} @@ -509,11 +515,17 @@ export function ThreadStatusLabel({ /> } > - + {status.marker ? ( + + {status.marker} + + ) : ( + + )} {status.label} {status.label} diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 204678eb1d2f..f1c81b764064 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -36,6 +36,19 @@ by older clients on one device no longer control this behavior. When you un-settle a thread, it returns to the top of the active list so you can find it right away. Its timestamps do not change. Other threads keep their positions. +## Questions from agents + +Agents can mark a thread when their latest response contains a question that blocks further work. +The sidebar shows a violet `?`, or **Question** when long status labels are enabled, so the thread +does not get lost among other conversations. Sending a reply clears the marker automatically. +Settling the thread yourself also dismisses it; automatic settlement waits until the question has +been answered or cleared. + +This marker is separate from a provider's structured approval and input prompts. Those keep their +existing, higher-priority status. The agent can currently raise only the `question` attention kind; +the stored attention record is typed so future user-actionable kinds can be added without treating +terminal output as an API. + Right-click a pull request link in a thread and choose **Link to thread** to show that pull request in the sidebar. The thread settles when the linked pull request merges if **Auto-settle merged threads** is enabled. Right-click the same link and choose **Unlink from thread** to remove it. diff --git a/packages/client-runtime/src/state/threadDetail.ts b/packages/client-runtime/src/state/threadDetail.ts index 5a2ffa442e04..b448bce68a48 100644 --- a/packages/client-runtime/src/state/threadDetail.ts +++ b/packages/client-runtime/src/state/threadDetail.ts @@ -64,6 +64,7 @@ export function mergeEnvironmentThread( pinnedAt: shell.pinnedAt, pinOrderKey: shell.pinOrderKey, session: shell.session, + attention: shell.attention, }; } diff --git a/packages/client-runtime/src/state/threadSettled.ts b/packages/client-runtime/src/state/threadSettled.ts index f5209a09e499..73834f47737f 100644 --- a/packages/client-runtime/src/state/threadSettled.ts +++ b/packages/client-runtime/src/state/threadSettled.ts @@ -52,6 +52,7 @@ export function hasQueuedTurnStart( */ export type ThreadSnoozeShell = Pick< OrchestrationThreadShell, + | "attention" | "snoozedUntil" | "snoozedAt" | "hasPendingApprovals" @@ -69,6 +70,7 @@ export type ThreadSnoozeShell = Pick< * the thread from classifying as snoozed. */ export function threadRaisedHandWhileSnoozed(shell: ThreadSnoozeShell): boolean { + if (shell.attention?.kind === "question") return true; if (shell.hasPendingApprovals || shell.hasPendingUserInput) return true; // Only a FRESH failure raises the hand: a thread snoozed while already // failed stays snoozed — that snooze was the user saying "I saw it, not @@ -102,11 +104,21 @@ export function threadRaisedHandWhileSnoozed(shell: ThreadSnoozeShell): boolean export function canSnooze( shell: Pick< OrchestrationThreadShell, - "hasPendingApprovals" | "hasPendingUserInput" | "latestUserMessageAt" | "latestTurn" | "session" + | "attention" + | "hasPendingApprovals" + | "hasPendingUserInput" + | "latestUserMessageAt" + | "latestTurn" + | "session" >, options: { readonly now: string }, ): boolean { - if (shell.hasPendingApprovals || shell.hasPendingUserInput) return false; + if ( + shell.hasPendingApprovals || + shell.hasPendingUserInput || + shell.attention?.kind === "question" + ) + return false; if (hasQueuedTurnStart(shell, options)) return false; return true; } diff --git a/packages/client-runtime/src/state/threadSnoozed.test.ts b/packages/client-runtime/src/state/threadSnoozed.test.ts index 8a62103950bf..4824a5812eac 100644 --- a/packages/client-runtime/src/state/threadSnoozed.test.ts +++ b/packages/client-runtime/src/state/threadSnoozed.test.ts @@ -29,6 +29,7 @@ function makeShell(input: { readonly snoozedAt?: string | null; readonly sessionStatus?: "starting" | "running" | "ready" | "error"; readonly pending?: "approval" | "user-input"; + readonly question?: boolean; readonly turnCompletedAt?: string | null; }): ThreadSnoozeShell { const threadId = ThreadId.make("thread-1"); @@ -37,6 +38,7 @@ function makeShell(input: { snoozedAt: input.snoozedAt ?? (input.snoozedUntil != null ? SNOOZED_AT : null), hasPendingApprovals: input.pending === "approval", hasPendingUserInput: input.pending === "user-input", + attention: input.question ? { kind: "question", raisedAt: SNOOZED_AT } : null, session: input.sessionStatus === undefined ? null @@ -73,6 +75,11 @@ function makeQueuedTurnShell(overrides: Partial = {}): QueuedTu } describe("effectiveSnoozed", () => { + it("raises a question attention even before the scheduled wake", () => { + expect( + effectiveSnoozed(makeShell({ snoozedUntil: FUTURE_WAKE, question: true }), { now: NOW }), + ).toBe(false); + }); it("hides a thread whose wake time is in the future", () => { expect(effectiveSnoozed(makeShell({ snoozedUntil: FUTURE_WAKE }), { now: NOW })).toBe(true); }); @@ -173,6 +180,11 @@ describe("threadRaisedHandWhileSnoozed", () => { }); describe("canSnooze", () => { + it("refuses question attention", () => { + expect( + canSnooze({ ...makeShell({ question: true }), latestUserMessageAt: null }, { now: NOW }), + ).toBe(false); + }); it("allows snoozing quiet and working threads alike", () => { expect(canSnooze({ ...makeShell({}), latestUserMessageAt: null }, { now: NOW })).toBe(true); expect( diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 17cadc6d1d7f..629d19af66bf 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -474,6 +474,14 @@ export const ThreadTitleRegeneration = Schema.Struct({ }); export type ThreadTitleRegeneration = typeof ThreadTitleRegeneration.Type; +/** Agent-managed attention raised on a thread. Add new kinds here as the UI + * gains distinct, user-actionable statuses. */ +export const ThreadAttention = Schema.Struct({ + kind: Schema.Literal("question"), + raisedAt: IsoDateTime, +}); +export type ThreadAttention = typeof ThreadAttention.Type; + export const ThreadLinkedPullRequest = Schema.Struct({ projectId: ProjectId, repository: TrimmedNonEmptyString, @@ -524,6 +532,8 @@ export const OrchestrationThread = Schema.Struct({ pinOrderKey: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), // Pending-only state. Optional so older servers remain compatible. titleRegeneration: Schema.optional(Schema.NullOr(ThreadTitleRegeneration)), + // Optional on the wire so cached snapshots from older servers still decode. + attention: Schema.optional(Schema.NullOr(ThreadAttention)), deletedAt: Schema.NullOr(IsoDateTime), messages: Schema.Array(OrchestrationMessage), proposedPlans: Schema.Array(OrchestrationProposedPlan).pipe( @@ -587,6 +597,7 @@ export const OrchestrationThreadShell = Schema.Struct({ pinnedAt: Schema.optional(Schema.NullOr(IsoDateTime)), pinOrderKey: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), titleRegeneration: Schema.optional(Schema.NullOr(ThreadTitleRegeneration)), + attention: Schema.optional(Schema.NullOr(ThreadAttention)), session: Schema.NullOr(OrchestrationSession), latestUserMessageAt: Schema.NullOr(IsoDateTime), hasPendingApprovals: Schema.Boolean, @@ -1152,6 +1163,21 @@ const ThreadActivityAppendCommand = Schema.Struct({ createdAt: IsoDateTime, }); +const ThreadAttentionSetCommand = Schema.Struct({ + type: Schema.Literal("thread.attention.set"), + commandId: CommandId, + threadId: ThreadId, + attention: ThreadAttention, + createdAt: IsoDateTime, +}); + +const ThreadAttentionClearCommand = Schema.Struct({ + type: Schema.Literal("thread.attention.clear"), + commandId: CommandId, + threadId: ThreadId, + createdAt: IsoDateTime, +}); + const ThreadRevertCompleteCommand = Schema.Struct({ type: Schema.Literal("thread.revert.complete"), commandId: CommandId, @@ -1176,6 +1202,8 @@ const InternalOrchestrationCommand = Schema.Union([ ThreadProposedPlanUpsertCommand, ThreadTurnDiffCompleteCommand, ThreadActivityAppendCommand, + ThreadAttentionSetCommand, + ThreadAttentionClearCommand, ThreadRevertCompleteCommand, ThreadTitleRegenerationCompleteCommand, ]); @@ -1217,6 +1245,8 @@ export const OrchestrationEventType = Schema.Literals([ "thread.proposed-plan-upserted", "thread.turn-diff-completed", "thread.activity-appended", + "thread.attention-set", + "thread.attention-cleared", ]); export type OrchestrationEventType = typeof OrchestrationEventType.Type; @@ -1455,6 +1485,17 @@ export const ThreadActivityAppendedPayload = Schema.Struct({ activity: OrchestrationThreadActivity, }); +export const ThreadAttentionSetPayload = Schema.Struct({ + threadId: ThreadId, + attention: ThreadAttention, + updatedAt: IsoDateTime, +}); + +export const ThreadAttentionClearedPayload = Schema.Struct({ + threadId: ThreadId, + updatedAt: IsoDateTime, +}); + /** * Which client connection dispatched the command that produced an event. * Stamped by the orchestration engine on client-dispatched commands; absent on @@ -1635,6 +1676,16 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.activity-appended"), payload: ThreadActivityAppendedPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.attention-set"), + payload: ThreadAttentionSetPayload, + }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.attention-cleared"), + payload: ThreadAttentionClearedPayload, + }), ]); export type OrchestrationEvent = typeof OrchestrationEvent.Type; @@ -1864,6 +1915,13 @@ export class OrchestrationDispatchCommandError extends Schema.TaggedErrorClass()( + "ThreadAttentionToolError", + { + message: TrimmedNonEmptyString, + }, +) {} + export class OrchestrationGetTurnDiffError extends Schema.TaggedErrorClass()( "OrchestrationGetTurnDiffError", { From 9993263aeef570312e325aa26ed06ab2d7ce2492 Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Thu, 3 Sep 2026 20:57:18 -0700 Subject: [PATCH 2/4] fix(mcp): enforce thread-only credentials --- apps/server/src/mcp/McpHttpServer.test.ts | 9 +++ apps/server/src/mcp/McpHttpServer.ts | 78 +++++++++++-------- .../client-runtime/src/state/threadSettled.ts | 3 + .../src/state/threadSnoozed.test.ts | 18 ++++- 4 files changed, 75 insertions(+), 33 deletions(-) diff --git a/apps/server/src/mcp/McpHttpServer.test.ts b/apps/server/src/mcp/McpHttpServer.test.ts index fa2880f9c364..f270600a56c2 100644 --- a/apps/server/src/mcp/McpHttpServer.test.ts +++ b/apps/server/src/mcp/McpHttpServer.test.ts @@ -24,6 +24,15 @@ const invocation = { capabilities: new Set(["preview"] as const), issuedAt: 1, }; + +it("keeps attention-only credentials off the full MCP endpoint", () => { + const attentionOnlyInvocation = { ...invocation, capabilities: new Set<"preview">() }; + + expect(McpHttpServer.canInvokeMcpEndpoint("/mcp", attentionOnlyInvocation)).toBe(false); + expect(McpHttpServer.canInvokeMcpEndpoint("/mcp/thread", attentionOnlyInvocation)).toBe(true); + expect(McpHttpServer.canInvokeMcpEndpoint("/mcp", invocation)).toBe(true); +}); + const client = McpSchema.McpServerClient.of({ clientId: 1, protocolVersion: "2025-06-18", diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 1e58bedfabe3..97b637d20255 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -65,37 +65,51 @@ export const normalizeMcpHttpResponse = ( : response; }; -const makeMcpAuthMiddleware = McpSessionRegistry.McpSessionRegistry.pipe( - Effect.map((registry): McpAuthMiddleware => - Effect.fn("McpHttpServer.authenticateRequest")(function* (httpEffect) { - const request = yield* HttpServerRequest.HttpServerRequest; - const authorization = request.headers.authorization; - const token = - authorization?.startsWith("Bearer ") === true - ? authorization.slice("Bearer ".length).trim() - : ""; - const invocation = yield* registry.resolve(token); - if (!invocation) { - // Without this the only symptom of a dead credential is the agent - // quietly losing the whole `t3-code` toolkit for the rest of its - // session, with nothing on the server to explain why. - yield* Effect.logWarning("rejected MCP request with an unusable credential", { - reason: token.length === 0 ? "missing_bearer_token" : "unknown_or_expired_token", - }); - return unauthorized; - } - return yield* httpEffect.pipe( - Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), - Effect.map(normalizeMcpHttpResponse), - ); - }), - ), - Effect.withSpan("McpHttpServer.makeAuthMiddleware"), -); +type McpEndpointPath = "/mcp" | "/mcp/thread"; + +export const canInvokeMcpEndpoint = ( + path: McpEndpointPath, + invocation: McpInvocationContext.McpInvocationScope, +): boolean => path === "/mcp/thread" || invocation.capabilities.has("preview"); + +const makeMcpAuthMiddleware = (path: McpEndpointPath) => + McpSessionRegistry.McpSessionRegistry.pipe( + Effect.map((registry): McpAuthMiddleware => + Effect.fn("McpHttpServer.authenticateRequest")(function* (httpEffect) { + const request = yield* HttpServerRequest.HttpServerRequest; + const authorization = request.headers.authorization; + const token = + authorization?.startsWith("Bearer ") === true + ? authorization.slice("Bearer ".length).trim() + : ""; + const invocation = yield* registry.resolve(token); + if (!invocation || !canInvokeMcpEndpoint(path, invocation)) { + // Without this the only symptom of a dead credential is the agent + // quietly losing the whole `t3-code` toolkit for the rest of its + // session, with nothing on the server to explain why. + yield* Effect.logWarning("rejected MCP request with an unusable credential", { + reason: + token.length === 0 + ? "missing_bearer_token" + : invocation + ? "insufficient_capability" + : "unknown_or_expired_token", + }); + return unauthorized; + } + return yield* httpEffect.pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), + Effect.map(normalizeMcpHttpResponse), + ); + }), + ), + Effect.withSpan("McpHttpServer.makeAuthMiddleware"), + ); -const McpAuthMiddlewareLive = HttpRouter.middleware<{ - provides: McpInvocationContext.McpInvocationContext; -}>()(makeMcpAuthMiddleware).layer; +const makeMcpAuthMiddlewareLive = (path: McpEndpointPath) => + HttpRouter.middleware<{ + provides: McpInvocationContext.McpInvocationContext; + }>()(makeMcpAuthMiddleware(path)).layer; const previewSnapshotFailure = (cause: Cause.Cause) => { if (Cause.hasInterrupts(cause) || cause.reasons.some(Cause.isDieReason)) { @@ -220,13 +234,13 @@ export const PreviewToolkitRegistrationLive = Layer.mergeAll( const threadAttentionToolkitRegistration = () => McpServer.toolkit(ThreadAttentionToolkit).pipe(Layer.provide(ThreadAttentionToolkitHandlersLive)); -const makeMcpTransport = (path: "/mcp" | "/mcp/thread") => +const makeMcpTransport = (path: McpEndpointPath) => McpServer.layerHttp({ name: "T3 Code", version: packageJson.version, path, protocols: [McpProtocol.v2025_06_18], - }).pipe(Layer.provide(McpAuthMiddlewareLive)); + }).pipe(Layer.provide(makeMcpAuthMiddlewareLive(path))); const FullToolkitLive = Layer.mergeAll( PreviewToolkitRegistrationLive, diff --git a/packages/client-runtime/src/state/threadSettled.ts b/packages/client-runtime/src/state/threadSettled.ts index 73834f47737f..8cce6e2b1ab3 100644 --- a/packages/client-runtime/src/state/threadSettled.ts +++ b/packages/client-runtime/src/state/threadSettled.ts @@ -165,6 +165,9 @@ export function threadWokeAt( // indicator the user already cleared by visiting (snoozedUntil is newer // than that visit's lastVisitedAt). if (threadRaisedHandWhileSnoozed(shell)) { + if (shell.attention?.kind === "question") { + return shell.attention.raisedAt; + } if ( shell.snoozedAt != null && shell.latestTurn?.state === "completed" && diff --git a/packages/client-runtime/src/state/threadSnoozed.test.ts b/packages/client-runtime/src/state/threadSnoozed.test.ts index 4824a5812eac..c8a8228d9358 100644 --- a/packages/client-runtime/src/state/threadSnoozed.test.ts +++ b/packages/client-runtime/src/state/threadSnoozed.test.ts @@ -30,6 +30,7 @@ function makeShell(input: { readonly sessionStatus?: "starting" | "running" | "ready" | "error"; readonly pending?: "approval" | "user-input"; readonly question?: boolean; + readonly questionRaisedAt?: string; readonly turnCompletedAt?: string | null; }): ThreadSnoozeShell { const threadId = ThreadId.make("thread-1"); @@ -38,7 +39,9 @@ function makeShell(input: { snoozedAt: input.snoozedAt ?? (input.snoozedUntil != null ? SNOOZED_AT : null), hasPendingApprovals: input.pending === "approval", hasPendingUserInput: input.pending === "user-input", - attention: input.question ? { kind: "question", raisedAt: SNOOZED_AT } : null, + attention: input.question + ? { kind: "question", raisedAt: input.questionRaisedAt ?? SNOOZED_AT } + : null, session: input.sessionStatus === undefined ? null @@ -293,6 +296,19 @@ describe("threadWokeAt", () => { ).toBe("2026-04-10T10:30:00.000Z"); }); + it("reports when a question raised the thread from snooze", () => { + expect( + threadWokeAt( + makeShell({ + snoozedUntil: FUTURE_WAKE, + question: true, + questionRaisedAt: "2026-04-10T11:30:00.000Z", + }), + { now: NOW }, + ), + ).toBe("2026-04-10T11:30:00.000Z"); + }); + it("falls back to session activity for blocked/failed early wakes", () => { expect( threadWokeAt(makeShell({ snoozedUntil: FUTURE_WAKE, sessionStatus: "error" }), { From 20ba22584c5eed8865ae88e5df09bac7f45e2ef6 Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Thu, 3 Sep 2026 22:06:58 -0700 Subject: [PATCH 3/4] fix(orchestration): clear stale question attention --- .../orchestration/decider.attention.test.ts | 20 +++++++++++++++ apps/server/src/orchestration/decider.ts | 25 ++++++++++++++++--- .../src/orchestration/projector.test.ts | 1 + .../src/components/ThreadStatusIndicators.tsx | 2 +- 4 files changed, 44 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration/decider.attention.test.ts b/apps/server/src/orchestration/decider.attention.test.ts index de8ebdfd746e..1e5b3fddf466 100644 --- a/apps/server/src/orchestration/decider.attention.test.ts +++ b/apps/server/src/orchestration/decider.attention.test.ts @@ -163,4 +163,24 @@ it.layer(NodeServices.layer)("thread attention decider", (it) => { expect(error._tag).toBe("OrchestrationCommandInvariantError"); }), ); + + it.effect("clears question attention after reverting past the question", () => + Effect.gen(function* () { + const events = yield* decide( + { + type: "thread.revert.complete", + commandId: CommandId.make("revert-complete"), + threadId, + turnCount: 1, + createdAt: NOW, + }, + makeReadModel({ attention: { kind: "question", raisedAt: NOW } }), + ); + + expect(events.map((event) => event.type)).toEqual([ + "thread.reverted", + "thread.attention-cleared", + ]); + }), + ); }); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index dcfa40aff623..0bd583515e5d 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1415,12 +1415,12 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" } case "thread.revert.complete": { - yield* requireThread({ + const thread = yield* requireThread({ readModel, command, threadId: command.threadId, }); - return { + const revertedEvent = { ...(yield* withEventBase({ aggregateKind: "thread", aggregateId: command.threadId, @@ -1432,7 +1432,26 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" threadId: command.threadId, turnCount: command.turnCount, }, - }; + } as const; + if (thread.attention == null) { + return revertedEvent; + } + return [ + revertedEvent, + { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.attention-cleared", + payload: { + threadId: command.threadId, + updatedAt: command.createdAt, + }, + }, + ]; } case "thread.activity.append": { diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index dad3d07370f9..15841252f561 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -75,6 +75,7 @@ describe("orchestration projector", () => { expect(next.threads).toEqual([ { id: "thread-1", + attention: null, projectId: "project-1", title: "demo", modelSelection: { diff --git a/apps/web/src/components/ThreadStatusIndicators.tsx b/apps/web/src/components/ThreadStatusIndicators.tsx index 410966d2b3a4..812aa4ab7cb9 100644 --- a/apps/web/src/components/ThreadStatusIndicators.tsx +++ b/apps/web/src/components/ThreadStatusIndicators.tsx @@ -521,7 +521,7 @@ export function ThreadStatusLabel({ ) : ( From bb2bc1a50d4eface4ce8f26ff32b3637b1228423 Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Thu, 3 Sep 2026 22:54:43 -0700 Subject: [PATCH 4/4] fix(orchestration): clear questions when turns abort --- .../orchestration/decider.attention.test.ts | 62 +++++++++++++++++++ apps/server/src/orchestration/decider.ts | 46 ++++++++++++-- 2 files changed, 104 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration/decider.attention.test.ts b/apps/server/src/orchestration/decider.attention.test.ts index 1e5b3fddf466..4eab7f9d395e 100644 --- a/apps/server/src/orchestration/decider.attention.test.ts +++ b/apps/server/src/orchestration/decider.attention.test.ts @@ -183,4 +183,66 @@ it.layer(NodeServices.layer)("thread attention decider", (it) => { ]); }), ); + + it.effect("clears question attention when the user interrupts the turn", () => + Effect.gen(function* () { + const events = yield* decide( + { + type: "thread.turn.interrupt", + commandId: CommandId.make("turn-interrupt"), + threadId, + createdAt: NOW, + }, + makeReadModel({ attention: { kind: "question", raisedAt: NOW } }), + ); + + expect(events.map((event) => event.type)).toEqual([ + "thread.turn-interrupt-requested", + "thread.attention-cleared", + ]); + }), + ); + + it.effect("clears question attention on failed turns but preserves delivered questions", () => + Effect.gen(function* () { + const session = { + threadId, + providerName: "Codex", + runtimeMode: "full-access" as const, + activeTurnId: null, + lastError: null, + updatedAt: NOW, + }; + const readModel = makeReadModel({ attention: { kind: "question", raisedAt: NOW } }); + + for (const status of ["error", "interrupted"] as const) { + const events = yield* decide( + { + type: "thread.session.set", + commandId: CommandId.make(`session-${status}`), + threadId, + session: { ...session, status }, + createdAt: NOW, + }, + readModel, + ); + expect(events.map((event) => event.type)).toEqual([ + "thread.session-set", + "thread.attention-cleared", + ]); + } + + const readyEvents = yield* decide( + { + type: "thread.session.set", + commandId: CommandId.make("session-ready"), + threadId, + session: { ...session, status: "ready" }, + createdAt: NOW, + }, + readModel, + ); + expect(readyEvents.map((event) => event.type)).toEqual(["thread.session-set"]); + }), + ); }); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 0bd583515e5d..786f2df92add 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1059,12 +1059,12 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" } case "thread.turn.interrupt": { - yield* requireThread({ + const thread = yield* requireThread({ readModel, command, threadId: command.threadId, }); - return { + const interruptRequestedEvent = { ...(yield* withEventBase({ aggregateKind: "thread", aggregateId: command.threadId, @@ -1077,7 +1077,26 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" ...(command.turnId !== undefined ? { turnId: command.turnId } : {}), createdAt: command.createdAt, }, - }; + } as const; + if (thread.attention == null) { + return interruptRequestedEvent; + } + return [ + interruptRequestedEvent, + { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.attention-cleared", + payload: { + threadId: command.threadId, + updatedAt: command.createdAt, + }, + }, + ]; } case "thread.approval.respond": { @@ -1281,6 +1300,23 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" session: command.session, }, }; + const terminalAttentionClearEvent: Omit | null = + thread.attention != null && + (command.session.status === "error" || command.session.status === "interrupted") + ? { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.attention-cleared", + payload: { + threadId: command.threadId, + updatedAt: command.createdAt, + }, + } + : null; // Only a session coming alive is activity worth waking a settled thread // for — status writes like ready/stopped/error arrive after the fact and // must not fight a user's explicit settle. Snooze is deliberately NOT @@ -1293,7 +1329,9 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" command.session.status === "starting" || command.session.status === "running"; // Real activity resets ANY override (settled wakes, active unpins). if (thread.settledOverride === null || !isSessionActivity) { - return sessionSetEvent; + return terminalAttentionClearEvent === null + ? sessionSetEvent + : [sessionSetEvent, terminalAttentionClearEvent]; } const unsettledEvent: Omit = { ...(yield* withEventBase({