diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index b81d542c191d..ac8c38f09a5a 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -922,6 +922,83 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } + // Goals: the decider emits these, the projection table has carried a + // goal_json column since migration 041, and the snapshot query reads + // it - but nothing ever wrote it, so every read model reported no + // goal. `thread.goal.continue` then refused with "no Active Goal to + // continue" and autonomous continuation never ran. + case "thread.goal-set": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) { + return; + } + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + goal: { + objective: event.payload.objective, + status: event.payload.status, + createdAt: event.payload.createdAt, + updatedAt: event.payload.updatedAt, + }, + updatedAt: event.payload.updatedAt, + }); + return; + } + + case "thread.goal-paused": + case "thread.goal-resumed": + case "thread.goal-blocked": + case "thread.goal-usage-limited": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) { + return; + } + const goal = existingRow.value.goal; + if (goal == null) { + return; + } + const status = + event.type === "thread.goal-paused" + ? "paused" + : event.type === "thread.goal-resumed" + ? "active" + : event.type === "thread.goal-blocked" + ? "blocked" + : "usageLimited"; + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + goal: { ...goal, status, updatedAt: event.payload.updatedAt }, + updatedAt: event.payload.updatedAt, + }); + return; + } + + case "thread.goal-cleared": + case "thread.goal-completed": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) { + return; + } + // Completing a goal keeps it visible as completed; clearing removes + // it, which is the difference the composer badge renders. + const goal = existingRow.value.goal; + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + goal: + event.type === "thread.goal-completed" && goal != null + ? { ...goal, status: "complete", updatedAt: event.payload.updatedAt } + : null, + updatedAt: event.payload.updatedAt, + }); + return; + } + case "thread.deleted": { attachmentSideEffects.deletedThreadIds.add(event.payload.threadId); const existingRow = yield* projectionThreadRepository.getById({ diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 2af64f323342..11fb2390208d 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -446,6 +446,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { snoozedAt: null, pinnedAt: "2026-02-24T00:00:01.000Z", pinOrderKey: "gm", + activeOrderKey: null, titleRegeneration: null, goal: null, session: { diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 7bf815b0c654..cc1001c7bb6c 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -17,6 +17,7 @@ import { type TurnId, } from "@t3tools/contracts"; import { isTemporaryWorktreeBranch, WORKTREE_BRANCH_PREFIX } from "@t3tools/shared/git"; +import { buildGoalContinuationPrompt } from "@t3tools/shared/goalContinuation"; import * as Cache from "effect/Cache"; import * as Cause from "effect/Cause"; import * as Crypto from "effect/Crypto"; @@ -1205,26 +1206,43 @@ const make = Effect.gen(function* () { return; } - const message = thread.messages.find((entry) => entry.id === event.payload.messageId); - if (!message || message.role !== "user") { + // A Continuation starts a Turn with no user message: the Objective is + // rendered into T3-authored prompt text instead. + const messageId = event.payload.messageId; + const continuationPrompt = + messageId === undefined && thread.goal?.status === "active" + ? buildGoalContinuationPrompt(thread.goal.objective) + : null; + if (messageId === undefined && continuationPrompt === null) { + return; + } + + const message = + messageId === undefined ? null : thread.messages.find((entry) => entry.id === messageId); + if (messageId !== undefined && (!message || message.role !== "user")) { yield* appendProviderFailureActivity({ threadId: event.payload.threadId, kind: "provider.turn.start.failed", summary: "Provider turn start failed", - detail: `User message '${event.payload.messageId}' was not found for turn start request.`, + detail: `User message '${messageId}' was not found for turn start request.`, turnId: null, createdAt: event.payload.createdAt, }); return; } + const messageText = continuationPrompt ?? message?.text ?? ""; + const attachments = message?.attachments; + // First-turn work (worktree branch, title generation) belongs to the // first REAL user message: corrections never count, and with queued // messages present the turn must be for that first message specifically. const isFirstUserMessageTurn = + message !== null && + message !== undefined && !isCorrectionMessage(message) && thread.messages.find((entry) => entry.role === "user" && !isCorrectionMessage(entry))?.id === - event.payload.messageId; + messageId; if (isFirstUserMessageTurn) { const project = yield* resolveProject(thread.projectId); const generationCwd = @@ -1233,8 +1251,8 @@ const make = Effect.gen(function* () { projects: project ? [project] : [], }) ?? process.cwd(); const generationInput = { - messageText: message.text, - ...(message.attachments !== undefined ? { attachments: message.attachments } : {}), + messageText, + ...(attachments !== undefined ? { attachments } : {}), ...(event.payload.titleSeed !== undefined ? { titleSeed: event.payload.titleSeed } : {}), }; @@ -1292,9 +1310,9 @@ const make = Effect.gen(function* () { const sendTurnRequest = yield* buildSendTurnRequestForThread({ threadId: event.payload.threadId, - ...(event.payload.messageId === undefined ? {} : { messageId: event.payload.messageId }), - messageText: message.text, - ...(message.attachments !== undefined ? { attachments: message.attachments } : {}), + ...(messageId === undefined ? {} : { messageId }), + messageText, + ...(attachments !== undefined ? { attachments } : {}), ...(event.payload.modelSelection !== undefined ? { modelSelection: event.payload.modelSelection } : {}), diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 595044abd578..b8bb13c9cbcd 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -14,6 +14,9 @@ import { ThreadSnoozedPayload as ContractsThreadSnoozedPayloadSchema, ThreadUnsnoozedPayload as ContractsThreadUnsnoozedPayloadSchema, ThreadGoalSetPayload as ContractsThreadGoalSetPayloadSchema, + ThreadTurnQueuedPayload as ContractsThreadTurnQueuedPayloadSchema, + ThreadQueuedTurnDispatchedPayload as ContractsThreadQueuedTurnDispatchedPayloadSchema, + ThreadQueuedTurnCancelledPayload as ContractsThreadQueuedTurnCancelledPayloadSchema, ThreadGoalPausedPayload as ContractsThreadGoalPausedPayloadSchema, ThreadGoalResumedPayload as ContractsThreadGoalResumedPayloadSchema, ThreadGoalClearedPayload as ContractsThreadGoalClearedPayloadSchema, @@ -55,6 +58,9 @@ export const ThreadUnsettledPayload = ContractsThreadUnsettledPayloadSchema; export const ThreadSnoozedPayload = ContractsThreadSnoozedPayloadSchema; export const ThreadUnsnoozedPayload = ContractsThreadUnsnoozedPayloadSchema; export const ThreadGoalSetPayload = ContractsThreadGoalSetPayloadSchema; +export const ThreadTurnQueuedPayload = ContractsThreadTurnQueuedPayloadSchema; +export const ThreadQueuedTurnDispatchedPayload = ContractsThreadQueuedTurnDispatchedPayloadSchema; +export const ThreadQueuedTurnCancelledPayload = ContractsThreadQueuedTurnCancelledPayloadSchema; export const ThreadGoalPausedPayload = ContractsThreadGoalPausedPayloadSchema; export const ThreadGoalResumedPayload = ContractsThreadGoalResumedPayloadSchema; export const ThreadGoalClearedPayload = ContractsThreadGoalClearedPayloadSchema; diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 002e355e2b2f..dee2d0dbc56b 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -21,6 +21,12 @@ import { ThreadArchivedPayload, ThreadCreatedPayload, ThreadDeletedPayload, + ThreadGoalClearedPayload, + ThreadQueuedTurnCancelledPayload, + ThreadQueuedTurnDispatchedPayload, + ThreadTurnQueuedPayload, + ThreadGoalPausedPayload, + ThreadGoalSetPayload, ThreadInteractionModeSetPayload, ThreadMetaUpdatedPayload, ThreadMessageCorrectedPayload, @@ -486,7 +492,12 @@ export function projectEvent( ); case "thread.active-reordered": - return decodeForEvent(ThreadActiveReorderedPayload, event.payload, event.type, "payload").pipe( + return decodeForEvent( + ThreadActiveReorderedPayload, + event.payload, + event.type, + "payload", + ).pipe( Effect.map((payload) => ({ ...nextBase, threads: updateThread(nextBase.threads, payload.threadId, { @@ -528,6 +539,131 @@ export function projectEvent( })), ); + // Queued turns had the same gap as goals: the pipeline marked the message + // queued in SQL, but the read model the decider validates against never + // did, so dispatching one was refused with "no longer waiting". + case "thread.turn-queued": + return decodeForEvent(ThreadTurnQueuedPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => { + // A goal continuation queues without a user message. + const messageId = payload.messageId; + if (messageId === undefined) return nextBase; + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) return nextBase; + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + messages: thread.messages.map((entry) => + entry.id === messageId ? { ...entry, deliveryState: "queued" as const } : entry, + ), + }), + }; + }), + ); + + case "thread.queued-turn-dispatched": + return decodeForEvent( + ThreadQueuedTurnDispatchedPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => { + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) return nextBase; + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + messages: thread.messages.map((entry) => { + if (entry.id !== payload.messageId) return entry; + const { deliveryState: _dropped, ...delivered } = entry; + return delivered; + }), + }), + }; + }), + ); + + case "thread.queued-turn-cancelled": + return decodeForEvent( + ThreadQueuedTurnCancelledPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => { + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) return nextBase; + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + messages: thread.messages.filter((entry) => entry.id !== payload.messageId), + }), + }; + }), + ); + + // Goals were emitted by the decider and persisted as events, but no + // projector case ever applied them, so the read model always reported no + // goal: setting one appeared to work and then `thread.goal.continue` + // refused with "no Active Goal to continue". + case "thread.goal-set": + return decodeForEvent(ThreadGoalSetPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + goal: { + objective: payload.objective, + status: payload.status, + createdAt: payload.createdAt, + updatedAt: payload.updatedAt, + }, + updatedAt: payload.updatedAt, + }), + })), + ); + + case "thread.goal-paused": + case "thread.goal-resumed": + case "thread.goal-blocked": + case "thread.goal-usage-limited": + case "thread.goal-completed": + return decodeForEvent(ThreadGoalPausedPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => { + const status = + event.type === "thread.goal-paused" + ? ("paused" as const) + : event.type === "thread.goal-resumed" + ? ("active" as const) + : event.type === "thread.goal-blocked" + ? ("blocked" as const) + : event.type === "thread.goal-usage-limited" + ? ("usageLimited" as const) + : ("complete" as const); + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + const goal = thread?.goal; + if (goal == null) return nextBase; + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + goal: { ...goal, status, updatedAt: payload.updatedAt }, + updatedAt: payload.updatedAt, + }), + }; + }), + ); + + case "thread.goal-cleared": + return decodeForEvent(ThreadGoalClearedPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + goal: null, + updatedAt: payload.updatedAt, + }), + })), + ); + case "thread.interaction-mode-set": return decodeForEvent( ThreadInteractionModeSetPayload,