From 579221de3f6c1864ab29e6c576188704794bed32 Mon Sep 17 00:00:00 2001 From: sheehanmunim Date: Wed, 19 Aug 2026 15:44:29 -0400 Subject: [PATCH] fix(orchestration): keep Goals and queued turns working in the live read model Goals could be set but never drove work, and a queued message could never be dispatched. Two gaps, both in the read model the decider validates against: - `projector.ts` (the in-memory projection behind `OrchestrationEngine`) ignored every `thread.goal-*` and queued-turn event, so `thread.goal` stayed null and a queued message never got `deliveryState: "queued"`. The SQL pipeline handled them; the live model did not, so the decider refused both `thread.goal.continue` and `thread.queued-turn.dispatch`. - `ProviderCommandReactor` still required a user message for every turn start. A Continuation starts a Turn with no message, so the whole autonomous loop died with "User message 'undefined' was not found". Restored the Objective prompt path that an earlier integration merge dropped. Also adds the goal cases to the SQL pipeline (`goal_json` had a column since migration 041 and nothing wrote it) and the `activeOrderKey` the shell snapshot now returns. Orchestration suite: 439 passing, was 433 with 6 failures. Model: Claude Opus 5, harness: Claude Code --- .../Layers/ProjectionPipeline.ts | 77 ++++++++++ .../Layers/ProjectionSnapshotQuery.test.ts | 1 + .../Layers/ProviderCommandReactor.ts | 36 +++-- apps/server/src/orchestration/Schemas.ts | 6 + apps/server/src/orchestration/projector.ts | 138 +++++++++++++++++- 5 files changed, 248 insertions(+), 10 deletions(-) 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,