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 84d8ddc6920e..c7545b45a4d3 100644 --- a/apps/mobile/src/features/threads/thread-list-v2-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-v2-items.tsx @@ -70,6 +70,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" }, waiting: { label: "Waiting", className: "text-adaptive-amber-700-300" }, 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 5badb27a6744..a700bbb3fb73 100644 --- a/apps/mobile/src/features/threads/threadListV2.test.ts +++ b/apps/mobile/src/features/threads/threadListV2.test.ts @@ -268,6 +268,28 @@ describe("resolveThreadListV2Status", () => { expect(shouldShowActionWaitingIndicator(working, "working")).toBe(false); }); + 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 0c30bcf720c4..9225c6d8d4f8 100644 --- a/apps/mobile/src/features/threads/threadListV2.ts +++ b/apps/mobile/src/features/threads/threadListV2.ts @@ -28,7 +28,14 @@ export { snoozeWakeLabel }; * (approval), "in motion" (working), and "broken" (failed). Ready is the * unlabeled resting state. */ -export type ThreadListV2Status = "approval" | "input" | "working" | "waiting" | "failed" | "ready"; +export type ThreadListV2Status = + | "approval" + | "input" + | "question" + | "working" + | "waiting" + | "failed" + | "ready"; export type ThreadListV2SwipeAction = "archive" | "settle" | "unsettle" | "snooze" | "unsnooze"; export type ThreadListV2CleanupAction = "retry-worktree-cleanup" | "keep-worktree"; @@ -138,7 +145,7 @@ export function resolveThreadListV2Enabled(input: { export function resolveThreadListV2Status( thread: Pick< EnvironmentThreadShell, - "hasPendingApprovals" | "hasPendingUserInput" | "session" | "actionResume" + "actionResume" | "attention" | "hasPendingApprovals" | "hasPendingUserInput" | "session" >, ): ThreadListV2Status { if (thread.hasPendingApprovals) { @@ -147,6 +154,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 78bf5497cd3c..ddc02fbaa8f6 100644 --- a/apps/mobile/src/features/threads/threadPresentation.ts +++ b/apps/mobile/src/features/threads/threadPresentation.ts @@ -6,6 +6,7 @@ import { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell"; export type ThreadStatusKind = | "pending-approval" | "awaiting-input" + | "question" | "working" | "waiting" | "connecting" @@ -114,6 +115,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.test.ts b/apps/server/src/mcp/McpHttpServer.test.ts index 10086caadf0b..127e5af659b6 100644 --- a/apps/server/src/mcp/McpHttpServer.test.ts +++ b/apps/server/src/mcp/McpHttpServer.test.ts @@ -34,6 +34,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 5c04b5cfcfbb..59947ff65577 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -24,6 +24,8 @@ import { } from "./toolkits/preview/tools.ts"; import { ActionResumeToolkitHandlersLive } from "./toolkits/actionResume/handlers.ts"; import { ActionResumeToolkit } from "./toolkits/actionResume/tools.ts"; +import { ThreadAttentionToolkitHandlersLive } from "./toolkits/threadAttention/handlers.ts"; +import { ThreadAttentionToolkit } from "./toolkits/threadAttention/tools.ts"; const unauthorized = HttpServerResponse.jsonUnsafe( { @@ -65,37 +67,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 McpAuthMiddlewareLive = HttpRouter.middleware<{ - provides: McpInvocationContext.McpInvocationContext; -}>()(makeMcpAuthMiddleware).layer; +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 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)) { @@ -221,14 +237,27 @@ export const ActionResumeToolkitRegistrationLive = McpServer.toolkit(ActionResum Layer.provide(ActionResumeToolkitHandlersLive), ); -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)); + +const makeMcpTransport = (path: McpEndpointPath) => + McpServer.layerHttp({ + name: "T3 Code", + version: packageJson.version, + path, + protocols: [McpProtocol.v2025_06_18], + }).pipe(Layer.provide(makeMcpAuthMiddlewareLive(path))); -export const layer = Layer.mergeAll( +const FullToolkitLive = Layer.mergeAll( PreviewToolkitRegistrationLive, ActionResumeToolkitRegistrationLive, -).pipe(Layer.provideMerge(McpTransportLive)); + 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 c847b88dcd7c..c43bc52dd1db 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,8 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( threadId: ThreadId.make(request.threadId), providerSessionId, providerInstanceId: ProviderInstanceId.make(request.providerInstanceId), - capabilities: new Set(["preview", "action-resume"]), + capabilities: + request.enablePreview === false ? new Set() : new Set(["preview", "action-resume"]), issuedAt, }; yield* SynchronizedRef.update(state, ({ records }) => { @@ -142,7 +144,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..299932daf22e --- /dev/null +++ b/apps/server/src/mcp/toolkits/threadAttention/handlers.test.ts @@ -0,0 +1,63 @@ +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({ + getTurnRequestWaitState: () => Effect.succeed({ kind: "correlation-not-found" }), + 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 544f9d3d004a..819c6b09a1c5 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -367,6 +367,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 ac2876577dd5..2bcbd528f7de 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -665,6 +665,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti annotation: null, worktreeCleanup: null, latestUserMessageId: null, + attention: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, @@ -858,6 +859,31 @@ 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 21d0ebe1bc08..8751c9578553 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -95,6 +95,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { pinned_at, pin_order_key, annotation_json, + attention_json, created_at, updated_at, deleted_at @@ -118,6 +119,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { '2026-02-24T00:00:01.000Z', 'gm', '{"body":"# Follow up","anchorMessageId":"message-1","createdAt":"2026-02-24T00:00:02.500Z","updatedAt":"2026-02-24T00:00:02.500Z","resolvedAt":null}', + '{"kind":"question","raisedAt":"2026-02-24T00:00:02.500Z"}', '2026-02-24T00:00:02.000Z', '2026-02-24T00:00:03.000Z', NULL @@ -354,6 +356,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { updatedAt: "2026-02-24T00:00:02.500Z", resolvedAt: null, }, + attention: { kind: "question", raisedAt: "2026-02-24T00:00:02.500Z" }, deletedAt: null, messages: [ { @@ -491,6 +494,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { updatedAt: "2026-02-24T00:00:02.500Z", resolvedAt: 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 f3b827f14e97..3c9d469b6256 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -26,6 +26,7 @@ import { ModelSelection, ProjectId, ThreadLinkedPullRequest, + ThreadAttention, ThreadId, ThreadAnnotation, ThreadWorktreeCleanup, @@ -106,6 +107,7 @@ const ProjectionThreadDbRowSchema = ProjectionThread.mapFields( linkedPullRequest: Schema.NullOr(Schema.fromJsonString(ThreadLinkedPullRequest)), annotation: Schema.NullOr(Schema.fromJsonString(ThreadAnnotation)), worktreeCleanup: Schema.optional(Schema.NullOr(Schema.fromJsonString(ThreadWorktreeCleanup))), + attention: Schema.NullOr(Schema.fromJsonString(ThreadAttention)), }), ); const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( @@ -500,6 +502,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { annotation_json AS "annotation", worktree_cleanup_json AS "worktreeCleanup", latest_user_message_id AS "latestUserMessageId", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -542,6 +545,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { annotation_json AS "annotation", worktree_cleanup_json AS "worktreeCleanup", latest_user_message_id AS "latestUserMessageId", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -586,6 +590,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { annotation_json AS "annotation", worktree_cleanup_json AS "worktreeCleanup", latest_user_message_id AS "latestUserMessageId", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -1053,6 +1058,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { annotation_json AS "annotation", worktree_cleanup_json AS "worktreeCleanup", latest_user_message_id AS "latestUserMessageId", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -1961,6 +1967,7 @@ pending_approval_requests AS ( titleRegeneration: mapTitleRegeneration(row), ...(row.annotation !== null ? { annotation: row.annotation } : {}), ...(row.worktreeCleanup != null ? { worktreeCleanup: row.worktreeCleanup } : {}), + attention: row.attention, deletedAt: row.deletedAt, messages: messagesByThread.get(row.threadId) ?? [], proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], @@ -2178,6 +2185,7 @@ pending_approval_requests AS ( ...(row.annotation !== null ? { annotation: row.annotation } : {}), latestUserMessageId: row.latestUserMessageId, ...(row.worktreeCleanup != null ? { worktreeCleanup: row.worktreeCleanup } : {}), + attention: row.attention, deletedAt: row.deletedAt, messages: [], proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], @@ -2323,6 +2331,7 @@ pending_approval_requests AS ( ...(row.worktreeCleanup != null ? { worktreeCleanup: row.worktreeCleanup } : {}), + attention: row.attention, session: sessionByThread.get(row.threadId) ?? null, latestUserMessageAt: row.latestUserMessageAt, hasPendingApprovals: row.pendingApprovalCount > 0, @@ -2473,6 +2482,7 @@ pending_approval_requests AS ( persistent: (row.persistent ?? 0) > 0, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), + attention: row.attention, session: sessionByThread.get(row.threadId) ?? null, latestUserMessageAt: row.latestUserMessageAt, hasPendingApprovals: row.pendingApprovalCount > 0, @@ -2775,6 +2785,7 @@ pending_approval_requests AS ( ...(threadRow.value.worktreeCleanup != null ? { worktreeCleanup: threadRow.value.worktreeCleanup } : {}), + attention: threadRow.value.attention, session: Option.isSome(sessionRow) ? mapSessionRow(sessionRow.value) : null, latestUserMessageAt: threadRow.value.latestUserMessageAt, hasPendingApprovals: threadRow.value.pendingApprovalCount > 0, @@ -3016,6 +3027,7 @@ pending_approval_requests AS ( pinOrderKey: threadRow.value.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(threadRow.value), ...(threadRow.value.annotation !== null ? { annotation: threadRow.value.annotation } : {}), + attention: threadRow.value.attention, deletedAt: null, messages: messageRows.map((row) => { const message = { diff --git a/apps/server/src/orchestration/Layers/ThreadDeletionReactor.test.ts b/apps/server/src/orchestration/Layers/ThreadDeletionReactor.test.ts index 3e61349b1544..44f3d61b88c5 100644 --- a/apps/server/src/orchestration/Layers/ThreadDeletionReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ThreadDeletionReactor.test.ts @@ -201,6 +201,7 @@ function cleanupRow( annotation: null, worktreeCleanup: cleanup, latestUserMessageId: null, + attention: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 29eb2040182b..ec4b247da33d 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -25,6 +25,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, @@ -62,6 +64,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..4eab7f9d395e --- /dev/null +++ b/apps/server/src/orchestration/decider.attention.test.ts @@ -0,0 +1,248 @@ +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"); + }), + ); + + 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", + ]); + }), + ); + + 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 16c047cb98ff..32661cb668b3 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -832,6 +832,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") { @@ -846,7 +849,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 / @@ -907,6 +912,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; } @@ -962,11 +982,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`, }), ); } @@ -1545,16 +1565,37 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }, }); } - return [...lifecycleResetEvents, turnMessageEvent, 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, + turnMessageEvent, + ...(attentionClearedEvent === null ? [] : [attentionClearedEvent]), + turnStartRequestedEvent, + ]; } 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, @@ -1567,7 +1608,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": { @@ -1760,6 +1820,23 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" 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 @@ -1772,7 +1849,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({ @@ -1894,12 +1973,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, @@ -1911,7 +1990,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": { @@ -1968,6 +2066,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.test.ts b/apps/server/src/orchestration/projector.test.ts index 4049c070d294..7dbc28f292f2 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -76,6 +76,7 @@ describe("orchestration projector", () => { { id: "thread-1", annotation: null, + attention: null, projectId: "project-1", title: "demo", modelSelection: { diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 4018a9439d28..6e81fd494ec8 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -20,6 +20,8 @@ import { ProjectDeletedPayload, ProjectMetaUpdatedPayload, ThreadActivityAppendedPayload, + ThreadAttentionClearedPayload, + ThreadAttentionSetPayload, ThreadArchivedPayload, ThreadPersistenceChangedPayload, ThreadCreatedPayload, @@ -334,6 +336,7 @@ export function projectEvent( persistent: false, annotation: null, worktreeCleanup: null, + attention: null, deletedAt: null, messages: [], activities: [], @@ -537,6 +540,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 00cbb7357728..8d4cb9a3c1b6 100644 --- a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts @@ -118,6 +118,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { resolvedAt: null, }, latestUserMessageId: MessageId.make("message-1"), + attention: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, @@ -176,7 +177,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { }), ); - it.effect("stores SQL NULL for thread fields omitted by pre-annotation events", () => + it.effect("stores SQL NULL for thread fields omitted by older events", () => Effect.gen(function* () { const threads = yield* ProjectionThreadRepository; const sql = yield* SqlClient.SqlClient; @@ -212,16 +213,19 @@ projectionRepositoriesLayer("Projection repositories", (it) => { const rows = yield* sql<{ readonly annotation: string | null; + readonly attention: string | null; readonly latestUserMessageId: string | null; }>` SELECT annotation_json AS annotation, + attention_json AS attention, latest_user_message_id AS "latestUserMessageId" FROM projection_threads WHERE thread_id = 'thread-before-annotations' `; assert.deepStrictEqual(rows[0], { annotation: null, + attention: null, latestUserMessageId: null, }); @@ -229,6 +233,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { threadId: ThreadId.make("thread-before-annotations"), }); assert.strictEqual(Option.getOrNull(persisted)?.annotation, null); + assert.strictEqual(Option.getOrNull(persisted)?.attention, null); assert.strictEqual(Option.getOrNull(persisted)?.latestUserMessageId, null); }), ); @@ -261,6 +266,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { pinnedAt: "2026-03-25T00:00:00.000Z", annotation: null, latestUserMessageId: null, + attention: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, @@ -338,6 +344,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 2dcf4e935490..47a09c82f2b2 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -21,6 +21,7 @@ import { import { ModelSelection, ThreadAnnotation, + ThreadAttention, ThreadLinkedPullRequest, ThreadWorktreeCleanup, } from "@t3tools/contracts"; @@ -31,6 +32,7 @@ const ProjectionThreadDbRow = ProjectionThread.mapFields( linkedPullRequest: Schema.NullOr(Schema.fromJsonString(ThreadLinkedPullRequest)), annotation: Schema.NullOr(Schema.fromJsonString(ThreadAnnotation)), worktreeCleanup: Schema.NullOr(Schema.fromJsonString(ThreadWorktreeCleanup)), + attention: Schema.NullOr(Schema.fromJsonString(ThreadAttention)), }), ); type ProjectionThreadDbRow = typeof ProjectionThreadDbRow.Type; @@ -69,6 +71,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { annotation_json, worktree_cleanup_json, latest_user_message_id, + attention_json, latest_user_message_at, pending_approval_count, pending_user_input_count, @@ -102,6 +105,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.annotation == null ? null : JSON.stringify(row.annotation)}, ${row.worktreeCleanup == null ? null : JSON.stringify(row.worktreeCleanup)}, ${row.latestUserMessageId ?? null}, + ${row.attention == null ? null : JSON.stringify(row.attention)}, ${row.latestUserMessageAt}, ${row.pendingApprovalCount}, ${row.pendingUserInputCount}, @@ -135,6 +139,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { annotation_json = excluded.annotation_json, worktree_cleanup_json = excluded.worktree_cleanup_json, latest_user_message_id = excluded.latest_user_message_id, + 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, @@ -175,6 +180,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { annotation_json AS "annotation", worktree_cleanup_json AS "worktreeCleanup", latest_user_message_id AS "latestUserMessageId", + attention_json AS "attention", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -217,6 +223,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { annotation_json AS "annotation", worktree_cleanup_json AS "worktreeCleanup", latest_user_message_id AS "latestUserMessageId", + 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 f94678fe923e..ac3dea46fd30 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -68,6 +68,7 @@ import Migration0053 from "./Migrations/053_ProjectionThreadLinkedPullRequest.ts import Migration0054 from "./Migrations/054_ProjectionThreadsUnsettledAt.ts"; import Migration0055 from "./Migrations/055_ProjectionThreadMessageSource.ts"; import Migration0056 from "./Migrations/056_ProjectionThreadsPersistent.ts"; +import Migration0057 from "./Migrations/057_ProjectionThreadAttention.ts"; /** * Migration loader with all migrations defined inline. @@ -136,6 +137,7 @@ export const migrationEntries = [ [54, "ProjectionThreadsUnsettledAt", Migration0054], [55, "ProjectionThreadMessageSource", Migration0055], [56, "ProjectionThreadsPersistent", Migration0056], + [57, "ProjectionThreadAttention", Migration0057], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/057_ProjectionThreadAttention.test.ts b/apps/server/src/persistence/Migrations/057_ProjectionThreadAttention.test.ts new file mode 100644 index 000000000000..a8b203bbd212 --- /dev/null +++ b/apps/server/src/persistence/Migrations/057_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()))("057_ProjectionThreadAttention", (it) => { + it.effect("adds nullable attention JSON to thread projections", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 56 }); + yield* runMigrations({ toMigrationInclusive: 57 }); + + 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/057_ProjectionThreadAttention.ts b/apps/server/src/persistence/Migrations/057_ProjectionThreadAttention.ts new file mode 100644 index 000000000000..1ba9b5891b62 --- /dev/null +++ b/apps/server/src/persistence/Migrations/057_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 7b4d4ccd916a..f0c70a17476a 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -17,6 +17,7 @@ import { RuntimeMode, ThreadLinkedPullRequest, ThreadAnnotation, + ThreadAttention, ThreadWorktreeCleanup, ThreadId, TurnId, @@ -55,6 +56,7 @@ export const ProjectionThread = Schema.Struct({ annotation: Schema.NullOr(ThreadAnnotation), worktreeCleanup: Schema.optional(Schema.NullOr(ThreadWorktreeCleanup)), latestUserMessageId: Schema.NullOr(MessageId), + attention: Schema.NullOr(ThreadAttention), latestUserMessageAt: Schema.NullOr(IsoDateTime), pendingApprovalCount: NonNegativeInt, pendingUserInputCount: NonNegativeInt, @@ -66,6 +68,7 @@ export type ProjectionThread = typeof ProjectionThread.Type; export const UpsertProjectionThreadInput = Schema.Struct({ ...ProjectionThread.fields, annotation: Schema.optional(Schema.NullOr(ThreadAnnotation)), + attention: Schema.optional(Schema.NullOr(ThreadAttention)), latestUserMessageId: Schema.optional(Schema.NullOr(MessageId)), }); export type UpsertProjectionThreadInput = typeof UpsertProjectionThreadInput.Type; diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index f61e9054578b..08ce8cd662aa 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -2617,11 +2617,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, @@ -2636,10 +2638,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), @@ -2667,28 +2668,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)), ); @@ -2698,7 +2689,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 8ddba586f191..0caba97f2b79 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -76,8 +76,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 = @@ -230,18 +228,9 @@ 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 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 @@ -262,18 +251,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 84935ed90ff0..2b65c1e692f7 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -1300,6 +1300,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 934094547b5b..c88da1885c88 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -217,18 +217,21 @@ export interface ThreadStatusPill { | "Plan Ready" | "Deleting" | "Deleting (Queued)" + | "Question" | "Cleanup failed"; 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, Waiting: 3, @@ -242,6 +245,7 @@ const THREAD_STATUS_PRIORITY: Record = { type ThreadStatusInput = Pick< SidebarThreadSummary, + | "attention" | "hasActionableProposedPlan" | "hasPendingApprovals" | "hasPendingUserInput" @@ -553,6 +557,7 @@ export function resolveThreadRowClassName(input: { export type SidebarThreadStatus = | "approval" | "input" + | "question" | "working" | "waiting" | "monitoring" @@ -564,11 +569,12 @@ export type SidebarThreadStatus = type SidebarThreadStatusInput = Pick< SidebarThreadSummary, + | "actionResume" + | "attention" + | "backgroundLiveness" | "hasPendingApprovals" | "hasPendingUserInput" | "session" - | "backgroundLiveness" - | "actionResume" | "worktreeCleanup" >; @@ -582,6 +588,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"; } @@ -805,6 +815,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 d8f1807afc58..e926017168ee 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -873,37 +873,43 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { icon: "waiting" as const, className: "text-yellow-700 dark:text-yellow-300", } - : status === "approval" + : status === "question" ? { - label: "Approval", - icon: null, - className: "text-amber-700 dark:text-amber-300", + label: "Question", + icon: "question" as const, + className: "text-violet-600 dark:text-violet-300", } - : status === "input" + : status === "approval" ? { - label: "Input", + label: "Approval", icon: null, - className: "text-indigo-600 dark:text-indigo-300", + className: "text-amber-700 dark:text-amber-300", } - : status === "failed" + : status === "input" ? { - label: "Failed", + label: "Input", icon: null, - className: "text-red-700 dark:text-red-300", + className: "text-indigo-600 dark:text-indigo-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({ @@ -1611,6 +1617,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 9af7e86f867a..22028310e7e9 100644 --- a/apps/web/src/components/ThreadStatusIndicators.tsx +++ b/apps/web/src/components/ThreadStatusIndicators.tsx @@ -474,7 +474,9 @@ export function ThreadStatusLabel({ status, compact = false, }: { - status: Pick & { label: string }; + status: Pick & { + label: string; + }; compact?: boolean; }) { if (compact) { @@ -489,12 +491,21 @@ export function ThreadStatusLabel({ /> } > - + {status.marker ? ( + + {status.marker} + + ) : ( + + )} {status.label} @@ -511,12 +522,21 @@ 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 ba382a50a878..524375588130 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -34,6 +34,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 34942020ccd7..d059e2819347 100644 --- a/packages/client-runtime/src/state/threadDetail.ts +++ b/packages/client-runtime/src/state/threadDetail.ts @@ -66,6 +66,7 @@ export function mergeEnvironmentThread( pinOrderKey: shell.pinOrderKey, annotation: shell.annotation, 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..8cce6e2b1ab3 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; } @@ -153,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 8a62103950bf..c8a8228d9358 100644 --- a/packages/client-runtime/src/state/threadSnoozed.test.ts +++ b/packages/client-runtime/src/state/threadSnoozed.test.ts @@ -29,6 +29,8 @@ function makeShell(input: { readonly snoozedAt?: string | null; 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"); @@ -37,6 +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: input.questionRaisedAt ?? SNOOZED_AT } + : null, session: input.sessionStatus === undefined ? null @@ -73,6 +78,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 +183,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( @@ -281,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" }), { diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 303e8bf0f9bf..fec28621f897 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -685,6 +685,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, @@ -797,6 +805,8 @@ export const OrchestrationThread = Schema.Struct({ // hydrating message bodies and attachments for every thread. latestUserMessageId: Schema.optional(Schema.NullOr(MessageId)), worktreeCleanup: Schema.optional(Schema.NullOr(ThreadWorktreeCleanup)), + // 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( @@ -863,6 +873,7 @@ export const OrchestrationThreadShell = Schema.Struct({ titleRegeneration: Schema.optional(Schema.NullOr(ThreadTitleRegeneration)), annotation: Schema.optional(Schema.NullOr(ThreadAnnotation)), worktreeCleanup: Schema.optional(Schema.NullOr(ThreadWorktreeCleanup)), + attention: Schema.optional(Schema.NullOr(ThreadAttention)), session: Schema.NullOr(OrchestrationSession), latestUserMessageAt: Schema.NullOr(IsoDateTime), hasPendingApprovals: Schema.Boolean, @@ -1507,6 +1518,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, @@ -1565,6 +1591,8 @@ const InternalOrchestrationCommand = Schema.Union([ ThreadProposedPlanUpsertCommand, ThreadTurnDiffCompleteCommand, ThreadActivityAppendCommand, + ThreadAttentionSetCommand, + ThreadAttentionClearCommand, ThreadRevertCompleteCommand, ThreadTitleRegenerationCompleteCommand, ThreadTurnRequestResolveCommand, @@ -1616,6 +1644,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; @@ -1887,6 +1917,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 @@ -2102,6 +2143,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; @@ -2331,6 +2382,13 @@ export class OrchestrationDispatchCommandError extends Schema.TaggedErrorClass()( + "ThreadAttentionToolError", + { + message: TrimmedNonEmptyString, + }, +) {} + export class OrchestrationGetTurnDiffError extends Schema.TaggedErrorClass()( "OrchestrationGetTurnDiffError", { diff --git a/packages/shared/src/agentAwareness.test.ts b/packages/shared/src/agentAwareness.test.ts index 28e07c04e6f3..8b99ea473421 100644 --- a/packages/shared/src/agentAwareness.test.ts +++ b/packages/shared/src/agentAwareness.test.ts @@ -29,6 +29,7 @@ function thread( | "updatedAt" | "hasPendingApprovals" | "hasPendingUserInput" + | "attention" > { return { id: "thread-1" as ThreadId, @@ -39,6 +40,7 @@ function thread( updatedAt: NOW, hasPendingApprovals: false, hasPendingUserInput: false, + attention: null, ...overrides, }; } @@ -102,6 +104,19 @@ describe("projectThreadAwareness", () => { }); }); + it("projects question attention as waiting for input", () => { + const state = projectThreadAwareness({ + environmentId: "env-1" as EnvironmentId, + project, + thread: thread({ + attention: { kind: "question", raisedAt: NOW }, + }), + }); + + expect(state?.phase).toBe("waiting_for_input"); + expect(state?.headline).toBe("Waiting for input"); + }); + it("projects completed turns as completed even when teardown settled them as interrupted", () => { const finishedTurn = { turnId: "turn-1" as TurnId, diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index c0f5842eb7c5..2ec5bdb1c076 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -40,6 +40,7 @@ export interface ProjectThreadAwarenessInput { | "updatedAt" | "hasPendingApprovals" | "hasPendingUserInput" + | "attention" >; } @@ -83,6 +84,9 @@ function resolveThreadAwarenessPhase( if (thread.hasPendingUserInput) { return "waiting_for_input"; } + if (thread.attention?.kind === "question") { + return "waiting_for_input"; + } if (thread.session?.status === "error" || thread.latestTurn?.state === "error") { return "failed"; }