fix(v2): compact full sessions before provider handoffs

apps/mobile/src/state/use-thread-composer-state.ts

@@ -5,6 +5,7 @@ import {
   deriveProviderSubagentStatus,
   deriveRunlessWorkStartedAt,
   deriveThreadActivityRun,
+  deriveThreadIsCompacting,
   deriveThreadRuntime,
   threadRuntimeHasInterruptibleRun,
 } from "@t3tools/client-runtime/state/thread-execution";
@@ -365,6 +366,8 @@ export function useThreadComposerState() {
   );
 
   const isCompacting = useMemo(() => {
+    if (selectedThreadProjection && deriveThreadIsCompacting(selectedThreadProjection.projection))
+      return true;
     const queuedCompact = selectedThreadQueuedMessages.some(
       (message) =>
         message.messageId === dispatchingQueuedMessageId &&
@@ -392,6 +395,7 @@ export function useThreadComposerState() {
     dispatchingQueuedMessageId,
     selectedThreadQueuedMessages,
     selectedThreadRuntime,
+    selectedThreadProjection,
     selectedThreadVisibleTurnItems,
   ]);
 

apps/server/src/orchestration-v2/ContextHandoffDelivery.ts

@@ -7,28 +7,24 @@ import * as Effect from "effect/Effect";
 import * as Schema from "effect/Schema";
 import { historyCost, renderHistory, selectHistory } from "./ContextHandoffBudget.ts";
 
-/** Persist before/after injection: an ambiguous pending delivery requires a fresh native thread. */
-export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextHandoffs")(
-  function* <InjectError = never, PersistError = never, BudgetError = never>(input: {
-    readonly handoffs: ReadonlyArray<OrchestrationV2ContextHandoff>;
-    readonly providerThread: OrchestrationV2ProviderThread;
-    readonly budget: number | Effect.Effect<number, BudgetError>;
-    readonly deferInline?: boolean;
-    readonly alreadyDeliveredItemIds: ReadonlySet<string>;
-    readonly inject?: (
-      history: ProviderAdapterV2HistoricalContext,
-    ) => Effect.Effect<boolean, InjectError>;
-    readonly persist: (handoff: OrchestrationV2ContextHandoff) => Effect.Effect<void, PersistError>;
-  }) {
+type ContextHandoffInput<BudgetError = never> = {
+  readonly handoffs: ReadonlyArray<OrchestrationV2ContextHandoff>;
+  readonly providerThread: OrchestrationV2ProviderThread;
+  readonly budget: number | Effect.Effect<number, BudgetError>;
+  readonly deferInline?: boolean;
+  readonly alreadyDeliveredItemIds: ReadonlySet<string>;
+};
+
+export const prepareContextHandoffs = Effect.fn("orchestrationV2.prepareContextHandoffs")(
+  function* <BudgetError = never>(input: ContextHandoffInput<BudgetError>) {
     const nativeThreadId = input.providerThread.nativeThreadRef?.nativeId ?? undefined;
     const pending = input.handoffs.filter(
       (handoff) =>
         nativeThreadId === undefined ||
         handoff.delivery?.nativeThreadId !== nativeThreadId ||
         handoff.delivery.status === "pending",
     );
-    if (pending.length === 0 || (input.deferInline && input.inject === undefined))
-      return { context: "", delivered: Effect.void };
+    if (pending.length === 0) return undefined;
     const budget = typeof input.budget === "number" ? input.budget : yield* input.budget;
     let coverage = pending
       .map(
@@ -71,9 +67,30 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH
       budget,
     });
     if (historyCost(selected.messages, selected.context) > budget) {
-      if (input.deferInline) return { context: "", delivered: Effect.void };
+      if (input.deferInline) return undefined;
       return yield* new ContextHandoffBudgetError();
     }
+    return { nativeThreadId, pending, selected };
+  },
+);
+
+/** Persist before/after injection: an ambiguous pending delivery requires a fresh native thread. */
+export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextHandoffs")(
+  function* <InjectError = never, PersistError = never, BudgetError = never>(
+    input: ContextHandoffInput<BudgetError> & {
+      readonly inject?: (
+        history: ProviderAdapterV2HistoricalContext,
+      ) => Effect.Effect<boolean, InjectError>;
+      readonly persist: (
+        handoff: OrchestrationV2ContextHandoff,
+      ) => Effect.Effect<void, PersistError>;
+    },
+  ) {
+    if (input.deferInline && input.inject === undefined)
+      return { context: "", delivered: Effect.void };
+    const prepared = yield* prepareContextHandoffs(input);
+    if (prepared === undefined) return { context: "", delivered: Effect.void };
+    const { nativeThreadId, pending, selected } = prepared;
     const omittedItemIds = new Set(selected.omittedItemIds);
     const persist = (status: "pending" | "injected" | "inline") =>
       Effect.forEach(
@@ -146,7 +163,7 @@ export class ContextHandoffBudgetError extends Schema.TaggedError<ContextHandoff
   {},
 ) {
   override get message() {
-    return "Insufficient context allowance for the provider handoff. Compact the target conversation or use a larger-context model; the current request has not been truncated.";
+    return "Insufficient context allowance for the provider handoff. Use a larger-context model or reduce the request; the current request has not been truncated.";
   }
 }
 export class ContextHandoffDeliveryUncertainError extends Schema.TaggedError<ContextHandoffDeliveryUncertainError>()(

apps/server/src/orchestration-v2/EventSink.ts

@@ -10,6 +10,7 @@ import {
   NodeId,
   type ProjectId,
   ThreadId,
+  TurnItemId,
 } from "@t3tools/contracts";
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
@@ -90,6 +91,11 @@ export interface EventSinkV2Shape {
     readonly activeAttemptId: RunAttemptId;
     readonly expectedStatus: OrchestrationV2Run["status"];
     readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
+    readonly effects?: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
+    readonly rejectUnpairedInterrupt?: {
+      readonly requestId: TurnItemId;
+      readonly resultId: TurnItemId;
+    };
   }) => Effect.Effect<
     {
       readonly committed: boolean;
@@ -416,7 +422,22 @@ const baseLayer: Layer.Layer<
             LIMIT 1
           `;
             const current = rows[0];
+            const interrupted =
+              input.rejectUnpairedInterrupt === undefined
+                ? false
+                : (yield* sql`
+                  SELECT 1 FROM orchestration_v2_projection_turn_items AS request
+                  WHERE request.thread_id = ${input.threadId}
+                    AND request.turn_item_id = ${input.rejectUnpairedInterrupt.requestId}
+                    AND NOT EXISTS (
+                      SELECT 1 FROM orchestration_v2_projection_turn_items AS result
+                      WHERE result.thread_id = ${input.threadId}
+                        AND result.turn_item_id = ${input.rejectUnpairedInterrupt.resultId}
+                    )
+                  LIMIT 1
+                `).length > 0;
             if (
+              interrupted ||
               current === undefined ||
               current.status !== input.expectedStatus ||
               current.active_attempt_id !== input.activeAttemptId
@@ -437,9 +458,15 @@ const baseLayer: Layer.Layer<
               events: normalized,
             });
             yield* applyStoredEvents(storedEvents);
+            yield* effectOutbox.enqueue(input.effects ?? []);
             return { committed: true as const, storedEvents };
           }),
-          (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void),
+          (result) =>
+            Effect.gen(function* () {
+              if (!result.committed) return;
+              if (input.effects?.length) yield* effectOutbox.notifyAvailable(input.effects.length);
+              yield* publishStoredEvents(result.storedEvents);
+            }),
         );
       },
     );