🤖 refactor: Effect Phase 11 PR 5 — DesktopLive group layers + DesktopWiringLive; thin ServiceContainer; StreamManager runner param

src/node/services/agentSession.ts

@@ -128,6 +128,7 @@ import { getTotalCost } from "@/common/utils/tokens/usageAggregator";
 import type { CompactionCompletionMetadata } from "@/common/types/compaction";
 import { CompactionHandler } from "./compactionHandler";
 import { RetryManager, type RetryFailureError, type RetryStatusEvent } from "./retryManager";
+import type { EffectRunner } from "./di/effectRunner";
 import type { TelemetryService } from "./telemetryService";
 import type { BackgroundProcessManager } from "./backgroundProcessManager";
 
@@ -543,6 +544,13 @@ export interface AgentSessionStreamManager {
   isStreaming(workspaceId: string): boolean;
   getStreamInfo(workspaceId: string): AgentSessionActiveStreamInfo | undefined;
   replayStream(workspaceId: string, options?: { afterTimestamp?: number }): Promise<void>;
+  /**
+   * The runner the stream manager's clock-driven fibers use; the session's
+   * `RetryManager` schedules its backoff on the same clock. Absent on test
+   * doubles and the AIService fallback, which leaves RetryManager on the
+   * global runtime.
+   */
+  readonly effectRunner?: EffectRunner;
 }
 
 /** Keeps AgentSession coupled only to the AI operations and events it consumes. */
@@ -952,7 +960,8 @@ export class AgentSession {
       async () => {
         await this.retryActiveStream();
       },
-      (event) => this.handleRetryStatusChange(event)
+      (event) => this.handleRetryStatusChange(event),
+      this.streamManager.effectRunner
     );
 
     this.attachAiListeners();

src/node/services/di/layers/app.ts

@@ -4,7 +4,7 @@ import { AppFiberScopeLive } from "@/node/services/di/appFiberScope";
 import { EffectRunnerLive } from "@/node/services/di/effectRunner";
 import type { AppTags } from "@/node/services/di/tags";
 import { CoreLive, MemoryMetaLive } from "./core";
-import { CoreOptionsFromDesktopLive, CrossCuttingLive } from "./desktop";
+import { CoreOptionsFromDesktopLive, CrossCuttingLive, DesktopLive } from "./desktop";
 import { StoresLive } from "./stores";
 
 /**
@@ -20,14 +20,16 @@ import { StoresLive } from "./stores";
  * captures its building context, so placing it there keeps that context to the
  * stores plus references (`Clock`, …). Above them the graph replays the
  * constructor's former order: memory metadata, the cross-cutting services, the
- * core options derived from them, then the staged core graph (which reads
- * `MemoryMeta` and `WorkspaceMcpOverrides` from those layers directly).
+ * core options derived from them, the staged core graph (which reads
+ * `MemoryMeta` and `WorkspaceMcpOverrides` from those layers directly), and
+ * finally the desktop group layers with their wiring.
  */
 export function AppLive(stores: ConfigStores): Layer.Layer<AppTags> {
   const runtimeSeams = AppFiberScopeLive.pipe(
     Layer.provideMerge(EffectRunnerLive.pipe(Layer.provideMerge(StoresLive(stores))))
   );
-  return CoreLive.pipe(
+  return DesktopLive.pipe(
+    Layer.provideMerge(CoreLive),
     Layer.provideMerge(CoreOptionsFromDesktopLive),
     Layer.provideMerge(CrossCuttingLive),
     Layer.provideMerge(MemoryMetaLive),

src/node/services/di/layers/core.ts

@@ -13,7 +13,7 @@ import { AIService } from "@/node/services/aiService";
 import { BackgroundProcessManager } from "@/node/services/backgroundProcessManager";
 import type { CoreOptions, CoreServices, CoreServicesOptions } from "@/node/services/coreServices";
 import { AppFiberScopeLive } from "@/node/services/di/appFiberScope";
-import { EffectRunnerLive } from "@/node/services/di/effectRunner";
+import { EffectRunnerLive, EffectRunnerTag } from "@/node/services/di/effectRunner";
 import {
   AI,
   BackgroundProcessManagerTag,
@@ -90,11 +90,18 @@ export class CoreOptionsTag extends Context.Service<CoreOptionsTag, CoreOptions>
 
 /**
  * What the roots must provide beneath `CoreLive`: the stores, the options,
- * and the two always-present collaborators the desktop builds elsewhere
- * (`MemoryMetaLive`; `WorkspaceMcpOverrides` from `CrossCuttingLive`). CLI
- * roots supply the defaults (`MemoryMetaLive`, `WorkspaceMcpOverridesDefaultLive`).
+ * the runtime's `EffectRunner` (the base seam in both roots; StreamManager's
+ * clock-driven fibers run through it), and the two always-present
+ * collaborators the desktop builds elsewhere (`MemoryMetaLive`;
+ * `WorkspaceMcpOverrides` from `CrossCuttingLive`). CLI roots supply the
+ * defaults (`MemoryMetaLive`, `WorkspaceMcpOverridesDefaultLive`).
  */
-export type CoreInputTags = StoreTags | CoreOptionsTag | MemoryMeta | WorkspaceMcpOverrides;
+export type CoreInputTags =
+  | StoreTags
+  | CoreOptionsTag
+  | EffectRunnerTag
+  | MemoryMeta
+  | WorkspaceMcpOverrides;
 
 /** Memory metadata sidecar; scope root derives from the xum home (`config.rootDir`). */
 export const MemoryMetaLive: Layer.Layer<MemoryMeta, never, ConfigTag> = Layer.effect(
@@ -220,8 +227,13 @@ export const StreamManagerLive = Layer.effect(
   StreamManagerTag,
   Effect.gen(function* () {
     const providerService = yield* Provider;
-    return new StreamManager(yield* History, yield* SessionUsage, () =>
-      providerService.getConfig()
+    return new StreamManager(
+      yield* History,
+      yield* SessionUsage,
+      () => providerService.getConfig(),
+      // Default event sink: AIService installs itself as the sink (S3).
+      undefined,
+      yield* EffectRunnerTag
     );
   })
 );