Merge pull request #2800 from generalaction/wire-abstraction

feat: wip workspace server worskapce setup

Author: Davidknp

apps/emdash-desktop/electron.vite.config.ts

@@ -3,6 +3,24 @@ import tailwindcss from '@tailwindcss/vite';
 import react from '@vitejs/plugin-react';
 import { defineConfig } from 'electron-vite';
 
+const workspaceAliases = {
+  '@emdash/core/acp/client': resolve('../../packages/core/src/acp/client.ts'),
+  '@emdash/core/acp': resolve('../../packages/core/src/acp/index.ts'),
+  '@emdash/core/agents/plugins/helpers': resolve(
+    '../../packages/core/src/agents/plugins/helpers/index.ts'
+  ),
+  '@emdash/core/agents/plugins': resolve('../../packages/core/src/agents/plugins/index.ts'),
+  '@emdash/plugins/agents/types': resolve('../../packages/plugins/src/agents/types.ts'),
+  '@emdash/plugins/agents': resolve('../../packages/plugins/src/agents/registry.ts'),
+  '@emdash/wire/api': resolve('../../packages/wire/src/api/index.ts'),
+  '@emdash/wire/process': resolve('../../packages/wire/src/process/index.ts'),
+  '@emdash/wire/util/mobx': resolve('../../packages/wire/src/util/mobx/index.ts'),
+  '@emdash/wire/util/process-runtime': resolve(
+    '../../packages/wire/src/util/process-runtime/index.ts'
+  ),
+  '@emdash/wire': resolve('../../packages/wire/src/index.ts'),
+};
+
 export default defineConfig({
   main: {
     root: 'src/main',
@@ -21,6 +39,7 @@ export default defineConfig({
         '@main': resolve('src/main'),
         '@shared': resolve('src/shared'),
         '@root': resolve('.'),
+        ...workspaceAliases,
       },
     },
   },
@@ -30,6 +49,7 @@ export default defineConfig({
       alias: {
         '@shared': resolve('src/shared'),
         '@root': resolve('.'),
+        ...workspaceAliases,
       },
     },
   },
@@ -42,6 +62,7 @@ export default defineConfig({
         '@renderer': resolve('src/renderer'),
         '@shared': resolve('src/shared'),
         '@root': resolve('.'),
+        ...workspaceAliases,
         // cli-agent-plugins metadata/icons chunks transitively reference node:buffer
         // (through hook-config helpers bundled in the same tsdown chunk), even though
         // those helpers never run in the renderer. Alias to the browser-safe polyfill.

apps/emdash-desktop/package.json

@@ -55,6 +55,7 @@
     "@emdash/plugins": "workspace:*",
     "@emdash/shared": "workspace:*",
     "@emdash/ui": "workspace:*",
+    "@emdash/wire": "workspace:*",
     "@fontsource-variable/inter": "^5.2.8",
     "@fontsource-variable/jetbrains-mono": "5.2.8",
     "@gitbeaker/rest": "^43.8.0",

apps/emdash-desktop/src/main/core/acp/agent-status-bridge.ts

@@ -1,35 +1,32 @@
-import { acpLiveTopics, sessionSummarySchema, type SessionSummary } from '@emdash/core/acp';
-import { LiveModelClient, type LiveSnapshot, type LiveUpdate } from '@emdash/core/live';
+import { sessionSummarySchema, type SessionSummary } from '@emdash/core/acp';
 import type { Unsubscribe } from '@emdash/shared';
+import { ReplicaState } from '@emdash/wire';
 import { z } from 'zod';
 import { agentHookService } from '@main/core/agent-hooks/agent-hook-service';
 import { isAppFocused } from '@main/core/agent-hooks/notification';
 import { log } from '@main/lib/logger';
 import { deriveAcpAgentStatusActions, type AcpAgentStatusAction } from './agent-status-transition';
-import { acpWire } from './controller';
-import { acpRuntimeProcessHost } from './runtime-process/host';
+import { getAcpRuntimeHandle } from './controller';
 
 type SessionSummaryList = Record<string, SessionSummary>;
 
 const sessionSummaryListSchema = z.record(z.string(), sessionSummarySchema);
 
 class AcpAgentStatusBridge {
   private readonly summaries = new Map<string, SessionSummary>();
-  private startedUnsubscribe: Unsubscribe | null = null;
-  private wireUnsubscribe: Unsubscribe | null = null;
+  private processExitUnsubscribe: Unsubscribe | null = null;
+  private replica: ReplicaState<SessionSummaryList> | null = null;
   private attaching = false;
 
   initialize(): void {
-    this.startedUnsubscribe = acpRuntimeProcessHost.onStarted(() => {
-      void this.attach().catch((error) => {
-        log.warn('ACP agent status bridge failed to attach', { error: String(error) });
-      });
+    void this.attach().catch((error) => {
+      log.warn('ACP agent status bridge failed to attach', { error: String(error) });
     });
   }
 
   dispose(): void {
-    this.startedUnsubscribe?.();
-    this.startedUnsubscribe = null;
+    this.processExitUnsubscribe?.();
+    this.processExitUnsubscribe = null;
     this.detach();
   }
 
@@ -38,58 +35,37 @@ class AcpAgentStatusBridge {
     this.attaching = true;
     try {
       this.detach();
-
-      const topic = acpLiveTopics.sessionStateList.topic(undefined);
-      const client = new LiveModelClient<SessionSummaryList>(
-        sessionSummaryListSchema,
-        () => acpWire.live.snapshot(topic) as Promise<LiveSnapshot<SessionSummaryList>>,
-        (summaries) => void this.applySummaries(summaries)
-      );
-      const buffer: LiveUpdate[] = [];
-      let seeded = false;
-      let detachLive: Unsubscribe | null = null;
-      let detachWire: Unsubscribe | null = null;
-
-      try {
-        detachLive = await acpWire.live.attach(topic, (update) => {
-          if (seeded) {
-            client.applyUpdate(update);
-          } else {
-            buffer.push(update);
-          }
-        });
-        detachWire = acpWire.onDisconnect(() => {
-          void this.resetAll().catch((error) => {
-            log.warn('ACP agent status bridge failed to reset statuses on disconnect', {
-              error: String(error),
-            });
+      const handle = await getAcpRuntimeHandle();
+      this.processExitUnsubscribe = handle.process.onExit((exit) => {
+        if (exit.willRestart) return;
+        void this.resetAll().catch((error) => {
+          log.warn('ACP agent status bridge failed to reset statuses on disconnect', {
+            error: String(error),
           });
-          this.detach();
         });
-
-        client.seed((await acpWire.live.snapshot(topic)) as LiveSnapshot<SessionSummaryList>);
-        seeded = true;
-        for (const update of buffer) {
-          client.applyUpdate(update);
+        this.detach();
+      });
+      const replica = new ReplicaState<SessionSummaryList>(
+        handle.client.sessions.state(undefined, 'list'),
+        {
+          schema: sessionSummaryListSchema,
+          onChange: (summaries) => void this.applySummaries(summaries),
         }
-
-        this.wireUnsubscribe = () => {
-          detachLive?.();
-          detachWire?.();
-        };
-      } catch (error) {
-        detachLive?.();
-        detachWire?.();
-        throw error;
-      }
+      );
+      await replica.ready;
+      this.replica = replica;
+      this.applySummaries(replica.current());
     } finally {
       this.attaching = false;
     }
   }
 
   private detach(): void {
-    this.wireUnsubscribe?.();
-    this.wireUnsubscribe = null;
+    this.processExitUnsubscribe?.();
+    this.processExitUnsubscribe = null;
+    const replica = this.replica;
+    this.replica = null;
+    if (replica) void replica.dispose();
   }
 
   private applySummaries(nextSummaries: SessionSummaryList): void {

apps/emdash-desktop/src/main/core/acp/controller.ts

@@ -1,10 +1,7 @@
-import type { AcpProcedures } from '@emdash/core/acp';
-import { portWire } from '@emdash/core/wire';
-import { typedProcedures } from '@emdash/core/wire';
-import { exposeWire } from '@main/lib/wire/expose-wire';
-import { acpRuntimeProcessHost } from './runtime-process/host';
-
-export const acpWire = portWire(acpRuntimeProcessHost.transport());
-
-export const acpRuntimeProcedures = typedProcedures<AcpProcedures>(acpWire.procedures);
-export const acpController = exposeWire('acp', acpWire);
+export {
+  disposeAcpRuntimeProcess,
+  getAcpRuntimeClient,
+  getAcpRuntimeHandle,
+  initializeAcpRuntimeProcess,
+  type AcpRuntimeClient,
+} from './runtime-process/host';

apps/emdash-desktop/src/main/core/acp/runtime-process/child-process-host.ts

@@ -2,18 +2,14 @@ import { spawn } from 'node:child_process';
 import { EventEmitter } from 'node:events';
 import { mkdir, readFile, writeFile } from 'node:fs/promises';
 import type {
+  acpHostContract,
   AcpFs,
   AcpProcessHandle,
   AcpProcessHost,
   AcpTerminalExit,
   AcpTerminalProcess,
 } from '@emdash/core/acp';
-import type { UtilityParentPort } from './protocol';
-
-type PendingResolve = {
-  resolve(value: { cli: string; agentEnv: Record<string, string> }): void;
-  reject(error: Error): void;
-};
+import type { ContractClient } from '@emdash/wire/api';
 
 class ChildProcessHandle implements AcpProcessHandle {
   constructor(private readonly child: ReturnType<typeof spawn>) {}
@@ -95,30 +91,13 @@ const fsPort: AcpFs = {
 
 export class ChildAcpProcessHost implements AcpProcessHost {
   readonly fs = fsPort;
-  private readonly pending = new Map<string, PendingResolve>();
-
-  constructor(private readonly parentPort: UtilityParentPort) {}
-
-  handleMessage(message: unknown): void {
-    if (!isResolveResult(message)) return;
-    const pending = this.pending.get(message.requestId);
-    if (!pending) return;
-    this.pending.delete(message.requestId);
-    if (message.ok) {
-      pending.resolve(message.value);
-    } else {
-      pending.reject(new Error(message.error));
-    }
-  }
+
+  constructor(private readonly host: ContractClient<typeof acpHostContract>) {}
 
   resolveSpawnContext(
     providerId: string
   ): Promise<{ cli: string; agentEnv: Record<string, string> }> {
-    const requestId = crypto.randomUUID();
-    return new Promise((resolve, reject) => {
-      this.pending.set(requestId, { resolve, reject });
-      this.parentPort.postMessage({ type: 'resolve-spawn-context', requestId, providerId });
-    });
+    return this.host.resolveSpawnContext({ providerId });
   }
 
   async spawn(spec: {
@@ -155,18 +134,3 @@ export class ChildAcpProcessHost implements AcpProcessHost {
     return new ChildTerminalProcess(child);
   }
 }
-
-function isResolveResult(value: unknown): value is {
-  type: 'resolve-spawn-context-result';
-  requestId: string;
-  ok: boolean;
-  value: { cli: string; agentEnv: Record<string, string> };
-  error: string;
-} {
-  return (
-    typeof value === 'object' &&
-    value !== null &&
-    (value as { type?: unknown }).type === 'resolve-spawn-context-result' &&
-    typeof (value as { requestId?: unknown }).requestId === 'string'
-  );
-}