fix(orchestration): keep Goals and queued turns working in the live read model

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({

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: {

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 }
         : {}),

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;

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,