fix(server): show idle paragraphs before the assistant turn ends

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

@@ -12,7 +12,6 @@ import {
   type ProjectId,
   ThreadId,
 } from "@t3tools/contracts";
-import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
 import * as ProviderCommitBatch from "./ProviderCommitBatch.ts";
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
@@ -79,9 +78,10 @@ const CurrentProviderBatch = Context.Reference<ProviderBatchPublication | undefi
 );
 
 export interface EventSinkV2Shape {
-  readonly consumeProviderEvents: <E, R, E2, R2>(
-    source: Stream.Stream<ProviderAdapter.ProviderAdapterV2Event, E, R>,
-    process: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect<void, E2, R2>,
+  /** Items that are not provider snapshots, such as a consumer's own control items, never batch. */
+  readonly consumeProviderEvents: <A extends { readonly type: string }, E, R, E2, R2>(
+    source: Stream.Stream<A, E, R>,
+    process: (event: A) => Effect.Effect<void, E2, R2>,
     after: Effect.Effect<boolean, E2, R2>,
     canBatch: Effect.Effect<boolean, never, R2>,
   ) => Effect.Effect<

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

@@ -1,4 +1,3 @@
-import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
 import * as Cause from "effect/Cause";
 import * as Clock from "effect/Clock";
 import * as Config from "effect/Config";
@@ -19,8 +18,7 @@ export const enabled = Config.Boolean("T3CODE_PROVIDER_COMMIT_BATCHING").pipe(
   Effect.orDie,
 );
 
-type Event = ProviderAdapter.ProviderAdapterV2Event;
-const isSnapshot = (event: Event) =>
+const isSnapshot = (event: { readonly type: string }) =>
   event.type === "node.updated" ||
   event.type === "message.updated" ||
   event.type === "turn_item.updated" ||
@@ -232,9 +230,9 @@ export const make = Effect.fn("ProviderCommitBatch.make")(function* (
       ),
     );
   });
-  const consume = Effect.fnUntraced(function* <E, R, E2, R2>(
-    source: Stream.Stream<Event, E, R>,
-    process: (event: Event) => Effect.Effect<void, E2, R2>,
+  const consume = Effect.fnUntraced(function* <A extends { readonly type: string }, E, R, E2, R2>(
+    source: Stream.Stream<A, E, R>,
+    process: (event: A) => Effect.Effect<void, E2, R2>,
     after: Effect.Effect<boolean, E2, R2>,
     canBatch: Effect.Effect<boolean, never, R2> = Effect.succeed(true),
   ) {