## File: README.md Emdash is a desktop app for running AI coding agents in parallel. Each task runs in its own Git worktree, so you can explore multiple fixes or features at once, review the diffs, and merge what works. It works with local projects and remote machines over SSH. Bring the CLI agents you already use: Claude Code, Codex, OpenCode, Amp, and more. ## What You Can Do - Run multiple coding agents at once without juggling terminals. - Keep every agent isolated in its own Git worktree and branch. - Send issues and tickets from Linear, GitHub, Jira, GitLab, Asana, Featurebase, Monday.com, Forgejo, Plain, or YouTrack into an agent. - Review diffs, create pull requests, inspect CI checks, and merge from one place. - Work locally or on your own remote machines over SSH/SFTP. ## Installation | Platform | Install | | --- | --- | | macOS | `brew install --cask emdash` · [Apple Silicon](https://github.com/generalaction/emdash/releases/latest/download/emdash-arm64.dmg) · [Intel](https://github.com/generalaction/emdash/releases/latest/download/emdash-x64.dmg) | | Windows | [Installer](https://github.com/generalaction/emdash/releases/latest/download/emdash-x64.msi) · [Portable](https://github.com/generalaction/emdash/releases/latest/download/emdash-x64.exe) | | Linux x64 | [AppImage](https://github.com/generalaction/emdash/releases/latest/download/emdash-x86_64.AppImage) · [DEB](https://github.com/generalaction/emdash/releases/latest/download/emdash-amd64.deb) · [RPM](https://github.com/generalaction/emdash/releases/latest/download/emdash-x86_64.rpm) | | Linux ARM64 | [AppImage](https://github.com/generalaction/emdash/releases/latest/download/emdash-arm64.AppImage) · [DEB](https://github.com/generalaction/emdash/releases/latest/download/emdash-arm64.deb) · [RPM](https://github.com/generalaction/emdash/releases/latest/download/emdash-aarch64.rpm) | See the [latest release](https://github.com/generalaction/emdash/releases/latest) for all desktop builds. ## Agents Emdash detects installed provider CLIs automatically. It supports agents like Claude Code, Codex, Cursor, OpenCode, Amp, Devin, Qwen Code, Droid, and GitHub Copilot. For agents with lifecycle-hook support, Emdash installs marker-tagged entries in the agent's user-level config. These hooks let Emdash track status, notifications, and resumable sessions, and silently do nothing when the agent runs outside an Emdash session. See [Providers](https://emdash.sh/docs/providers) for the full list, setup commands, and provider-specific behavior. ## Remote Projects Connect to remote machines with SSH/SFTP and run the same parallel workflow on remote codebases. Emdash supports SSH agent, key, and password authentication, including OpenSSH certificates held by the agent, with credentials stored in your OS keychain. See [Remote Projects](https://emdash.sh/docs/remote-projects) for setup details. ## Privacy Emdash is local-first. App state is stored in a local SQLite database, and Emdash does not send your code or chats to Emdash servers. Agent CLIs may send code, prompts, and context to their own providers. Their data handling depends on the provider you choose. Telemetry is optional and can be disabled in Settings or by launching with: ```bash TELEMETRY_ENABLED=false ``` See [Telemetry](https://emdash.sh/docs/telemetry) for details. ## Contributing Contributions are welcome. Read the [Contributing Guide](CONTRIBUTING.md), open an issue, or join the [Discord](https://discord.gg/f2fv7YxuR2). ## License Licensed under the [Apache-2.0 license](LICENSE.md). --- ## File: agents/README.md # Agent Docs This directory is the system of record for agent-facing repo guidance. Keep topic pages small, specific, and mechanically checkable where possible. ## Recommended Reading Order 1. `quickstart.md` 2. `architecture/overview.md` 3. the task-specific page for the area you are changing ## Directory Layout - `architecture/` - system structure and major code ownership boundaries - [`core-modules.md`](architecture/core-modules.md) — Core runtime, service, and primitive module types, platform surfaces, dependency rules, and subpath exports - [`workspace-server.md`](architecture/workspace-server.md) — protocol versioning, negotiation handshake, and behavior-change rules for the remote workspace daemon - [`state-ownership.md`](architecture/state-ownership.md) — desktop-owned entities, host-owned resources, bindings, and the placement rule for operation behavior - [`acp-runtime.md`](architecture/acp-runtime.md) — ACP runtime, session manager, connection pool, cells, live models, and API contract ownership - [`git-runtime.md`](architecture/git-runtime.md) — Git contract/runtime ownership, session lifecycle, and nested wire composition - [`path-system.md`](architecture/path-system.md) — host-aware path identity, resource URI ownership, and future migration boundaries - `workflows/` - task-oriented procedures like testing, worktrees, remote development, and Nx task orchestration - `integrations/` - provider, MCP, and external service guidance - [`integration-plugins.md`](integrations/integration-plugins.md) — integration auth and credential contracts, stable account identity, and issue-plugin boundaries - `risky-areas/` - places where incorrect changes are expensive - `conventions/` - coding contracts and repo rules - [`ui-kit.md`](conventions/ui-kit.md) — which UI kit to use: `@emdash/ui` for components, feature Tailwind for layout one-offs, where new primitives go, `cn()` location, theming - [`ui-styling.md`](conventions/ui-styling.md) — vanilla-extract styling rules for `packages/ui` ## Maintenance Rules - Prefer one page per concrete topic. - Avoid volatile counts unless you can verify them cheaply. - Link to the source-of-truth file paths. - Update the smallest relevant page instead of expanding `AGENTS.md`. --- ## File: apps/workspace-server/docs/client-connection.md # Workspace Server Client Connection This document describes the target desktop-to-workspace-server connection model. Only macOS and Linux remotes are in scope for this design. ## Topology ```mermaid flowchart LR subgraph desktop [Desktop Main Process] source["ResourceCache per target"] reconnecting[reconnectingTransport] local[Local Unix socket connector] sshManager[SshConnectionManager] proxy[SshClientProxy] end subgraph host [Workspace Host] socket["Unix socket ~/.emdash/workspace-server/run/workspace.sock"] hub[createWireSessionHub] controller[workspaceWireContract controller] runtimes["required core runtime subprocesses"] end source --> reconnecting reconnecting --> local local --> socket reconnecting --> sshManager sshManager --> proxy proxy -->|"openssh_forwardOutStreamLocal"| socket socket --> hub hub --> controller controller --> runtimes ``` The workspace server daemon listens on a Unix domain socket. The socket lives in a user-owned directory with mode `0700`; SSH authentication plus filesystem permissions form the access boundary. Desktop clients do not connect to a TCP port on the remote host. `serve --socket` is the daemon serving mode. `serve --stdio` serves the same wire controller over stdin/stdout and exists as a test and debugging harness. The daemon runs every core runtime in a required supervised subprocess in both socket and stdio modes. Clients reach ACP, agent config, automations, conversations, file search, files, Git, resource usage, terminals, TUI agents, and the workspace registry through the corresponding aggregate client domains. Live state, jobs, event streams, mutations, terminal logs, and file blob channels all use the same reconnectable workspace wire connection. ## Desktop Utility Shape The desktop should expose a resource cache keyed by the full workspace-server target. Local sockets are the first implementation; SSH becomes another stream source without changing handshake or reconnect behavior. Including the socket path in the key avoids conflating custom daemon sockets on the same host. ```ts type WorkspaceServerTarget = | { kind: 'local-socket'; socketPath: string } | { kind: 'ssh'; sshConnectionId: string; socketPath: string }; const workspaces = createResourceCache({ key: workspaceServerTargetKey, idleTtlMs: 30_000, async create(target: WorkspaceServerTarget, scope) { let negotiated: WireInitializeResult | undefined; const transport = reconnectingTransport( () => openInitializedWorkspaceTransport(target, (value) => { negotiated = value; }), { shouldRetry: (error) => !(error instanceof WorkspaceProtocolError), } ); scope.add(() => transport.close()); const connection = connect(transport); scope.add(() => connection.dispose()); const workspace = client(workspaceWireContract, connection); await transport.ready(); return { client: workspace, connection, negotiated: () => negotiated }; }, }); async function openInitializedWorkspaceTransport( target: WorkspaceServerTarget, onInitialized: (value: WireInitializeResult) => void ): Promise { const stream = await openWorkspaceServerStream(target); const candidate = ownedStreamTransport(stream); const handshakeConnection = connect(candidate); const handshakeClient = client(workspaceWireContract, handshakeConnection); try { const initialized = await handshakeClient.initialize({ protocolVersion: PROTOCOL_VERSION, client: { id: clientId, appVersion }, }); if (!initialized.success) throw new WorkspaceProtocolError(initialized.error); onInitialized(initialized.data); return candidate; } catch (error) { candidate.close?.(); throw error; } finally { handshakeConnection.dispose(); } } ``` `openWorkspaceServerStream()` initially opens a manually-started local daemon's Unix socket. The later SSH implementation obtains the stable proxy from `SshConnectionManager`, ensures the daemon exists, and opens a streamlocal channel to the same socket path. `ownedStreamTransport()` wraps `streamTransport(stream, stream)` and destroys the underlying socket/channel from `close()`. `streamTransport()` itself only releases its parsing listeners because it does not own the streams passed to it. `ensureWorkspaceDaemon()` is a future desktop bootstrap step. It should call the server-side lifecycle CLI to probe the socket, start the daemon if absent, and leave version upgrades to the wire update flow. After the daemon exists, each `connectOnce` uses the current ssh2 client exposed by `SshClientProxy` to open an `openssh_forwardOutStreamLocal` channel and adapts it with `streamTransport(channel, channel)`. A convenience method may be added to the proxy when the SSH connector is implemented. ## Handshake and Readiness Invariants Each `connectOnce` attempt owns one physical stream and performs the `initialize` handshake before returning it to `reconnectingTransport`. Therefore: 1. `initialize` is the first workspace-server procedure on every candidate stream. 2. A compatible response records the negotiated feature level and daemon identity. 3. Only then can the stable transport flush queued messages and announce reconnect. 4. `connect()` reattaches live topics after that reconnect notification. 5. An incompatible response closes the candidate, rejects `ready()`, and stops retries. The temporary handshake `Connection` is disposed without closing a successful candidate. The stable outer `Connection` is the only connection that remains subscribed after readiness. ## Failure Semantics - SSH connection drops: the streamlocal channel closes, `streamTransport` emits disconnect, and `reconnectingTransport` retries `connectOnce`. - SSH reconnects: `SshConnectionManager.connect(connectionId)` reuses or waits for the managed SSH connection using the existing credential resolution. - Workspace daemon restarts: `connectOnce` reopens the Unix socket, initializes the new daemon generation, and exposes its new `daemonId` only after negotiation. - Live attachments: `connect()` observes `onReconnect` from `reconnectingTransport` after initialization, re-attaches topics, and replicas resync from fresh snapshots. - Pending calls: in-flight calls reject with `DISCONNECTED`; callers decide whether procedure retries are safe. Live model mutations can use mutation IDs and retry options. - Calls made while disconnected: `reconnectingTransport` queues non-blob messages in a bounded drop-oldest buffer. Features should wait for `ready()` and use cancellation/deadlines so an evicted request cannot wait indefinitely. - Protocol mismatch: `initialize` returns a typed `Result` error with `action: 'upgrade-client' | 'upgrade-server'`. `shouldRetry` classifies it as permanent; the owner surfaces the error and invalidates the cache entry. The future update flow should route `upgrade-server` into workspace-server self-update. ## Server Modes Socket mode: ```bash emdash-workspace-server serve --socket emdash-workspace-server serve --socket ~/.emdash/workspace-server/run/workspace.sock ``` The server ensures the socket directory exists, probes before unlinking stale socket files, and serves each accepted `net.Socket` through `createWireSessionHub`. Daemon lifecycle: ```bash emdash-workspace-server start emdash-workspace-server status emdash-workspace-server stop ``` Lifecycle commands use socket mode by default. They derive sidecar files from the socket path so custom paths remain self-contained: ```text ~/.emdash/workspace-server/run/workspace.sock ~/.emdash/workspace-server/run/workspace.sock.pid ~/.emdash/workspace-server/run/workspace.sock.lock ~/.emdash/workspace-server/run/workspace.sock.log ~/.emdash/workspace-server/state/ ``` `start` probes the socket first, acquires the lock, spawns `serve --socket` as a detached process when needed, and waits for `health` to succeed. `stop` reads the pid file and sends `SIGTERM`, relying on the daemon signal handlers to dispose the socket server and unlink the socket file. `status` connects to the socket and calls `health`. ## ACP Session IDs ACP `start` and `resume` return `{ sessionId }` directly to the desktop client through the workspace wire connection. The desktop side should persist the returned session id when remote ACP session resume is wired. Stdio mode: ```bash emdash-workspace-server serve --stdio ``` Stdio mode must not write logs to stdout because stdout is the wire protocol channel. It is intended for local integration tests and manual debugging, not as the production SSH transport. ## Related Docs - `apps/workspace-server/docs/daemon.md` - `packages/wire/docs/api/serving.md` - `packages/wire/docs/api/transports.md` - `packages/wire/docs/runtime/lifecycle.md` - `packages/wire/docs/runtime/workers.md` --- ## File: apps/workspace-server/docs/daemon.md # Workspace Server Daemon The workspace server is a per-user Node process that exposes the `workspaceWireContract` over either a Unix domain socket or stdio. The production remote path is socket mode: the desktop connects over SSH stream-local forwarding to the daemon's Unix socket, and the daemon serves each accepted socket as an independent wire session. Stdio mode runs the same controller over stdin/stdout. It is primarily a test and debugging harness because stdout is reserved for the wire protocol. ## Process Roles The command line has two roles: - `serve` is the foreground server process. In socket mode, this is the daemon. - `start`, `stop`, and `status` are short-lived lifecycle commands that manage or inspect the daemon process. ```mermaid flowchart LR cliStart["CLI: start"] -->|"probe health"| socket["workspace.sock"] cliStart -->|"spawn detached serve"| daemon["CLI: serve --socket"] daemon -->|"listen"| socket daemon -->|"write pid"| pidFile["workspace.sock.pid"] cliStatus["CLI: status"] -->|"wire health()"| socket cliStop["CLI: stop"] -->|"read pid, SIGTERM"| daemon desktop["Desktop over SSH"] -->|"streamlocal channel"| socket socket --> sessionHub["WireSessionHub"] sessionHub --> controller["workspaceWireContract controller"] daemon --> workerHost["WireWorkerHost"] workerHost --> runtimes["required core runtime subprocesses"] workerHost --> watcher["FS watch subprocess"] watcher --> runtimes controller --> runtimes runtimes -->|"spawn provider CLIs and PTYs"| agentCli["host-local processes"] ``` ## Commands Foreground serving: ```bash emdash-workspace-server serve --socket emdash-workspace-server serve --socket ~/.emdash/workspace-server/run/workspace.sock emdash-workspace-server serve --stdio ``` Daemon lifecycle: ```bash emdash-workspace-server start emdash-workspace-server status emdash-workspace-server stop ``` For backwards compatibility, invoking the binary with only flags still means `serve`: ```bash emdash-workspace-server --socket emdash-workspace-server --stdio ``` Lifecycle commands use socket mode. Passing `--socket ` selects a custom daemon instance; otherwise they use the default socket path. ## Runtime Files All runtime files are derived from the socket path so custom socket paths remain self-contained: ```text ~/.emdash/workspace-server/run/workspace.sock ~/.emdash/workspace-server/run/workspace.sock.pid ~/.emdash/workspace-server/run/workspace.sock.lock ~/.emdash/workspace-server/run/workspace.sock.log ~/.emdash/workspace-server/state/acp-attachments/ ~/.emdash/workspace-server/state/acp-session-intents.json ~/.emdash/workspace-server/state/tui-agent-session-intents.json ~/.emdash/workspace-server/state/automations.db ~/.emdash/workspace-server/state/conversations.db ~/.emdash/workspace-server/state/file-search.db ~/.emdash/workspace-server/state/workspace-registry.db ~/.emdash/workspace-server/state/host-dependencies.json ``` - `.sock` is the Unix domain socket accepted by the foreground daemon. - `.pid` records the daemon process id after socket serving starts. - `.lock` prevents competing `start` calls from spawning duplicate daemons. - `.log` receives stdout/stderr from the detached daemon process. - `state/` stores runtime-owned databases, intents, attachments, and dependency selections. ACP attachments are laid out per conversation. The socket directory is created with mode `0700`, so the filesystem boundary is the local user account. Desktop clients reach the socket through SSH, not through a TCP listener. ## Start Flow `start` is intentionally conservative: 1. Derive socket, pid, lock, and log paths. 2. Acquire the exclusive lock file. 3. Probe the socket with the wire `health` procedure. 4. If the daemon is already healthy, return successfully without spawning. 5. Remove stale pid/socket files when no healthy daemon is present. 6. Spawn `serve --socket ` as a detached child with stdout/stderr redirected to the log file. 7. Wait until the socket answers `health`, then exit. The lifecycle command does not keep the daemon alive after spawning it. Once the detached `serve` process is healthy, it owns the socket and pid file until it exits. ## Serve Flow `serve --socket` owns the long-lived server lifecycle: 1. Create the runtime state directories and the in-process host-dependency store. 2. Start every required runtime and FS-watch subprocess and wait for readiness. 3. Build the aggregate controller from the required runtime clients. 4. Create the socket directory. 5. Probe any existing socket file before unlinking it. A live socket blocks startup; a dead socket file is removed. 6. Listen on the Unix socket and write the pid file. 7. For every accepted `net.Socket`, convert it to a `WireTransport` and open a session in `createWireSessionHub`. 8. On `SIGINT` or `SIGTERM`, dispose all wire sessions, close the server, unlink the socket, remove the pid file, stop the runtime children, and exit. The session hub is what lets one daemon serve multiple client connections while keeping per-client live subscriptions, in-flight calls, and blob channels scoped to the connection that created them. ## Runtime Workers Both serve modes mount every core runtime as a child process. The parent creates workers through `WireWorkerHost` using executable paths declared in the workspace-server worker manifest. Each child entry initializes structured process logging and calls `runWireComponentWorker(...)`. The worker dependency graph mirrors the desktop host: - Conversations, FS watch, terminals, and resource usage have no runtime-worker dependencies. - ACP and TUI agents consume conversations and the parent-owned host-dependency resolver; agent config consumes the resolver only. - Files and file search consume FS watch; Git consumes FS watch and host dependencies. - The workspace registry consumes FS watch, ACP, terminals, and TUI agents. - Automations starts last and consumes the workspace registry, ACP sessions, TUI sessions, and the conversation index. Every runtime is required. Failure to start any worker aborts server startup, while failures after readiness use the worker host's supervision policy. Runtime logs are emitted as structured stderr lines and forwarded by the parent process. ACP `start` and `resume` return `{ sessionId }` through the ACP API. The connected desktop client owns persistence for those returned ids when remote ACP consumption is added. ## Stop Flow `stop` uses the pid file rather than a wire shutdown RPC: 1. Read `.pid`. 2. If the pid file is missing or the process is already gone, remove the stale pid file and report `not-running`. 3. Send `SIGTERM` to the recorded pid. 4. Poll until the process is gone and the socket no longer answers `health`. 5. Remove the pid file. There is no automatic `SIGKILL` escalation. A stuck process should be surfaced to the caller rather than hidden. ## Status Flow `status` connects to the Unix socket and calls `health`. A healthy response prints the socket path, version, and uptime. A failed probe exits non-zero and reports whether the daemon is not running or unhealthy. ## Code Map - `src/index.ts` parses config, dispatches commands, and installs signal handlers for `serve`. - `src/config.ts` defines command and serve-mode parsing. - `src/daemon/paths.ts` centralizes socket, pid, lock, and log paths. - `src/daemon/probe.ts` implements the wire `health` probe. - `src/daemon/start.ts` implements locked detached startup and health waiting. - `src/daemon/stop.ts` implements pid-file based `SIGTERM` shutdown. - `src/daemon/status.ts` implements the status probe. - `src/wire/serve-socket.ts` owns Unix socket serving and maps each accepted socket into a wire session. - `src/wire/serve-stdio.ts` serves the same controller over stdin/stdout. - `src/api/controller.ts` builds the `workspaceWireContract` controller. - `src/gateway/workspace-workers.ts` owns the complete required worker graph and runtime state configuration. - `src/gateway/worker-manifest.ts` selects the shared Core entries and app-local plugin entries for each packaged worker executable. - `src/gateway/worker-paths.ts` resolves packaged worker executable paths at runtime. - `src/gateway/entries/` contains the ACP, agent-config, and TUI-agent entries that inject the application plugin registry; all other workers reuse Core runtime entries directly. - `src/runtime/paths.ts` owns runtime database, intent, and attachment paths. ## Desktop Integration Desktop-side daemon bootstrap is intentionally separate from this server package. The intended flow is: 1. Resolve or establish the SSH connection. 2. Run `emdash-workspace-server start` on the remote host. 3. Open an SSH stream-local channel to the socket path. 4. Wrap that channel in `streamTransport(channel, channel)`. 5. Create a wire client and call `initialize`. See `client-connection.md` for the target desktop-to-daemon connection model. See `packaging.md` for self-contained release artifact targets, layout, build commands, and Linux compatibility constraints. --- ## File: packages/wire/examples/README.md # @emdash/wire Examples These examples show the public surface of `@emdash/wire`. Every example imports the package entrypoints only (`/rpc`, `/live`, `/state`, via their in-repo source paths) and splits the authoritative server side from the client-side binding so the boundary looks like a real transport without adding an adapter. ## Documentation For conceptual docs that explain how the examples fit together, see [../docs/README.md](../docs/README.md). Run them from the repository root: ```bash pnpm --filter @emdash/wire run example:state-kernel pnpm --filter @emdash/wire run example:live-log pnpm --filter @emdash/wire run example:event-stream pnpm --filter @emdash/wire run example:mailbox pnpm --filter @emdash/wire run example:cancellation pnpm --filter @emdash/wire run example:api-definition pnpm --filter @emdash/wire run example:dedupe pnpm --filter @emdash/wire run example:job-contract ``` Examples: - `state-kernel/` demonstrates the state kernel end to end: `cell` + `family` + `expose` on the server, `remote()` + `observe` + `optimistic` on the client, including an optimistic mutation settling against the authoritative cursor. - `live-log/` demonstrates a contract-level `liveLog()` endpoint served by a `LiveLogSource`, consumed through `createLiveLogReplicaCache` and `ReplicaLog`. - `event-stream/` demonstrates a contract-level `eventStream()` endpoint served by `createEventStreamHost` with client-side subscribe and gap handling. - `mailbox/` demonstrates bounded local producer/consumer handoff, graceful close, and explicit overflow. - `cancellation/` demonstrates procedure cancellation with `AbortSignal` and server-side abort on disconnect. - `api-definition/` isolates contract definition with `defineContract`, `procedure`, `liveModel`, `liveLog`, and live model contract mutations. - `dedupe/` demonstrates server-side `deduplicate()` middleware from `@emdash/shared/requests` for in-flight procedure calls. - `job-contract/` demonstrates the contract-level `liveJob()` endpoint with start, progress, cancellation, terminal result, and reattach. - `scope/` and `resource-cache/` demonstrate the `@emdash/shared/concurrency` primitives the live layer builds on. --- ## File: packages/wire/docs/live/event-stream.md # Event Streams `eventStream({ key, event })` exposes a keyed server-to-client event channel for loss-tolerant notifications. It reuses the wire live attachment protocol but does not retain events or expose meaningful snapshots. Use event streams when subscribers can recover from missed events by resyncing from another source of truth, such as a file tree, cache, or database query. Use `liveLog` for retained append-only text and `liveModel` for convergent state. ## Contract ```ts const api = defineContract({ fileEvents: eventStream({ key: z.object({ rootPath: z.string() }), event: z.object({ kind: z.enum(['create', 'update', 'delete']), path: z.string(), }), }), }); ``` The key addresses an event stream instance. The event schema types the client-side `onEvent` callback and documents the payload shape. ## Server Most servers use `createEventStreamHost()`: ```ts const fileEvents = createEventStreamHost(api.fileEvents); const controller = createController(api, { fileEvents, }); fileEvents.emit( { rootPath: '/repo' }, { kind: 'update', path: '/repo/package.json' } ); ``` `emit()` is fire-and-forget. If no client is attached to that key, the event is dropped. When the last subscriber detaches, the host discards the keyed source. If attachment presence owns an external resource, pass lifecycle hooks to the host. `onActive` runs when the first subscriber attaches to a key, and `onIdle` runs when the last subscriber detaches: ```ts const resourceEvents = resourcedStream({ key: z.object({ rootPath: z.string() }), event: z.object({ kind: z.string(), path: z.string() }), }); const fileEvents = createEventStreamHost(resourceEvents, { onActive(key) { void startWatcher(key); }, onIdle(key) { void stopWatcher(key); }, }); ``` This is useful for resources that should be recreated after reconnect. The wire client automatically reattaches live topics after reconnect, so `onActive` runs again in the new server process. When a successful attach must mean that the resource is ready, define the endpoint with `resourcedStream()` and provide `activate`. Activation receives a signal and resolves to its owned disposer: ```ts const fileEvents = createEventStreamHost(api.fileEvents, { async activate(key, signal) { const watcher = await startWatcher(key, { signal }); return () => watcher.dispose(); }, }); ``` Concurrent subscribers share one activation. The host aborts pending activation when the final subscriber leaves, disposes a late result, and keeps activation alive when another subscriber still owns the key. You can also pass a resolver to `createController()` if another component owns the source: ```ts createController(api, { fileEvents: (key) => sources.get(key.rootPath), }); ``` ## Client ```ts const unsubscribe = await client.fileEvents.subscribe( { rootPath: '/repo' }, { onEvent(event) { console.log(event.kind, event.path); }, onGap() { void reloadFileTree(); }, onError(error, { retrying }) { if (!retrying) showFileEventStreamError(error.message); }, signal, } ); ``` `onGap` runs after the underlying transport reattaches. Events may have been lost while the link was disconnected or while a supervised subprocess restarted, so consumers should refresh any derived state in that callback. Initial attachment does not call `onGap`; subscribers should perform their own initial load before or after subscribing. The `subscribe()` promise rejects if the initial attachment fails. After an attachment is established, `DISCONNECTED` reattach failures call `onError` with `retrying: true` and remain registered for a later reconnect. Non-retryable reattach errors call `onError` with `retrying: false` and terminate the attachment. ## Semantics - Events are at-most-once and only delivered to currently attached clients. - No historical events are retained for late subscribers. - Event stream keys are validated by controller validation. - Event payloads are not validated on the push path, matching other live updates. - Forwarding preserves upstream gaps and reattach errors even when the downstream transport remains connected. ## Example See [../../examples/event-stream/client.ts](../../examples/event-stream/client.ts) for a runnable in-process example. --- ## File: packages/wire/docs/live/live-job.md # Live Jobs Live jobs model async work as a live state machine. They are useful when callers need both a final result and a stream of progress updates. ## State Shape `liveJobStateSchema(progress, result, error)` creates a schema for: ```ts type LiveJobState = | { status: 'running'; startedAt: number; progress: P[]; progressCount: number } | { status: 'succeeded'; startedAt: number; finishedAt: number; progress: P[]; result: R } | { status: 'failed'; startedAt: number; finishedAt: number; progress: P[]; error?: E; cause?: SerializedError } | { status: 'cancelled'; startedAt: number; finishedAt: number; progress: P[] }; ``` `progress` is a retained ring of recent entries. `progressCount` is the total number of progress events emitted, including entries that may have rolled out of the retained ring. Terminal states retain lifecycle timestamps and the most recent progress entries, but they do not carry `progressCount`. ## Server `LiveJobSource` accepts a handler that returns `Result`: ```ts const jobs = new LiveJobSource( async (input, ctx) => { ctx.progress({ step: 'checkout' }); await build(input, ctx.signal); ctx.progress({ step: 'package' }); return ok({ artifact: `${input.target}.zip` }); }, { toError: (error) => ({ message: error instanceof Error ? error.message : String(error) }) } ); const { jobId } = jobs.start({ target: 'desktop' }); ``` Options: - `generation`: optional fixed generation for every job state model. - `terminalRetainMs`: how long terminal state remains attachable. Default: `LIVE_JOB_TERMINAL_RETAIN_MS` (5 minutes). - `idFactory`: custom job id factory, useful for deterministic tests. - `clock`: custom timestamp source, useful for deterministic tests. - `toError`: optional fallback for unexpected thrown errors. Domain failures should be returned as `err(error)` from the handler. - `onRunStarted`, `onRunChanged`, and `onRunEvicted`: optional hooks for callers that want to maintain their own job listing. Handlers receive `LiveJobContext`: - `ctx.progress(value)` appends retained progress and emits an update. - `ctx.signal` is an `AbortSignal`. If the signal is aborted, the job finishes as `cancelled`. Return `ok(result)` for successful completion and `err(error)` for expected domain failures. If the handler throws and `toError` is configured, the thrown value is mapped into typed `error`; otherwise the failed state carries a serialized `cause`. `source(jobId)` returns a read-only live source for that job. `snapshot(jobId)` and `getState(jobId)` are convenience accessors. `cancel(id)` aborts a running job. `dispose()` is async: it aborts running jobs, waits for cooperative handlers to settle, clears eviction timers, and removes all job state. Terminal job state is retained for `LIVE_JOB_TERMINAL_RETAIN_MS` (5 minutes) so late clients can reattach by id. This retention is process-local, not durable. ## Consumers Consumers normally reach jobs through the API layer. The contract client starts, cancels, and attaches to jobs without retaining local state: ```ts const { jobId } = await contractClient.build.start({ target: 'desktop' }); const handle = contractClient.build.handle(jobId); const detach = await handle.attach((update) => { console.log(update.delta); }); await contractClient.build.cancel(jobId); detach(); ``` Use `createLiveJobReplicaCache()` when a consumer wants a `result` promise, progress callbacks, ref-counted attachment, and terminal-state retention. If the job fails, `ReplicaJob.result` rejects with `LiveJobFailedError`; if it is cancelled, it rejects with `LiveJobCancelledError`. Progress emitted by a seed is suppressed so reattaching clients do not replay old progress as fresh events. ## Contract Endpoint The API contract layer can expose a job directly: ```ts const api = defineContract({ build: liveJob({ input: z.object({ target: z.string() }), progress: z.object({ step: z.string() }), result: z.object({ artifact: z.string() }), error: z.object({ message: z.string() }), }), }); ``` On the server, implement the endpoint with `{ run, toError? }`: ```ts const controller = createController(api, { build: { run: async (input, ctx) => { ctx.progress({ step: 'compile' }); return ok({ artifact: `${input.target}.zip` }); }, toError: (error) => ({ message: error instanceof Error ? error.message : String(error), }), }, }); ``` The typed client exposes a live job client handle. Use it directly for low-level forwarding, or wrap it in `createLiveJobReplicaCache()` when a consumer wants ref-counted local state: ```ts const jobs = createLiveJobReplicaCache(api.build, contractClient.build, { lingerMs: 30_000 }); const lease = await jobs.start({ target: 'desktop' }); const handle = await lease.ready(); handle.onProgress((progress) => console.log(progress.step)); console.log(await handle.result); await lease.release(); const reattachedLease = jobs.acquire(handle.jobId); const reattached = await reattachedLease.ready(); console.log(await reattached.result); await reattachedLease.release(); await jobs.dispose(); ``` `ReplicaJob` exposes `jobId`, `ready`, `result`, `getState()`, `onProgress(cb)`, and `cancel()`. The `LiveJobReplicaCache` manager owns ref counting, subscription sharing, and terminal-state retention. `LiveJobSource` accepts a `Clock` for deterministic start/finish timestamps and terminal retention eviction. Server disposal is awaitable: it cancels active job runs, waits for cooperative handlers to settle, and disposes retention timers before returning. Cancellation at the procedure/wire level is documented in [serving](../api/serving.md#cancellation). Job cancellation is domain-level: `ReplicaJob.cancel()` calls the generated `.cancel` procedure, which calls `LiveJobSource.cancel(jobId)`. See [../../examples/job-contract/client.ts](../../examples/job-contract/client.ts). --- ## File: packages/wire/docs/live/live-log.md # Live Logs Live logs are append-only text streams with retained tail snapshots. They are a good fit for terminal output, build logs, and agent process output where clients need to attach late and still see recent context. ## Server `LiveLogSource` stores a bounded text buffer. `append(chunk)` emits `{ chunk }` deltas to subscribers and updates the retained snapshot: ```ts const server = new LiveLogSource({ generation: 3000, maxBufferBytes: 12 }); export function appendLine(line: string): void { server.append(`${line}\n`); } ``` Options: - `generation`: optional fixed generation, useful in tests. - `maxBufferBytes`: retained buffer size. The default is 1 MiB. Snapshots contain `LiveLogSnapshotData`: ```ts type LiveLogSnapshotData = { baseOffset: number; text: string; truncated: boolean; }; ``` `baseOffset` is the byte offset of `text` in the logical full log. When the server drops old bytes because `maxBufferBytes` was exceeded, `truncated` is `true`. The server keeps at least the newest chunk even if it exceeds the byte limit. `reseed()` starts a new generation and resets sequence to `0`. With no argument, it clears retained text and resets `baseOffset` to `0`; with retained snapshot data, it seeds the buffer from that data. ## Consumers Consumers normally reach logs through the API layer. The client handle is useful when you want to stream directly into a terminal or log viewer: ```ts const output = contractClient.activity.handle(session); const snapshot = await output.snapshot(); term.write(snapshot.data.text); const detach = await output.attach((update) => { term.write((update.delta as { chunk: string }).chunk); }); ``` Use `createLiveLogReplicaCache()` when a process wants a retained local buffer that can also be served downstream. Without a custom sink, the replica keeps an eager bounded `LiveLogSource` buffer and `text()` remains readable. With a custom `LogSink`, the sink owns storage; readable `LogStore`s add `text()`, while write-only sinks can stream directly into xterm without a duplicate text buffer. The replica's internal log follower tracks the followed generation/sequence plus the UTF-8 byte offset already written to its sink. On reconnect or explicit `refresh()`, it splices from the retained snapshot when the generation still matches, appending only missing bytes instead of resetting rendered output. Generation changes or evicted gaps still reset the sink from the retained snapshot. The underlying protocol follower resyncs on update-before-seed, generation mismatch, sequence gap, or invalid log delta. Resync events use the same `resync` instrumentation hook as live models. ## API Layer Usage Contracts expose logs with `liveLog({ key })`: ```ts const api = defineContract({ activity: liveLog({ key: sessionKeySchema }), }); const controller = createController(api, { activity: () => activityLogServer, }); const activity = contractClient.activity.handle(session); const detach = await activity.attach((update) => { console.log((update.delta as { chunk: string }).chunk); }); detach(); ``` The API layer handles topic encoding, snapshots, attachment, and detachment. See [../../examples/live-log/client.ts](../../examples/live-log/client.ts). --- ## File: packages/wire/docs/live/live-model.md # Live States and Protocol Live states are reactive JSON state containers. A server owns authoritative state, emits ordered Immer patches, and clients apply those patches locally. When a client detects a gap, it refetches a snapshot and resumes from the new generation. ## Protocol Terms The shared protocol lives in `src/live/protocol`. - `LiveSnapshot` is the full state at a cursor: `{ generation, sequence, timestamp, data }`. - `LiveUpdate` is an ordered delta: `{ generation, baseSequence, sequence, timestamp, delta, mutationIds? }`. - `LiveCursor` is `{ generation, sequence }` and identifies a point in a live state stream. - `LiveSource` is the common server-side shape consumed by the API layer: `snapshot()` plus `subscribe(cb)`. `generation` changes when the server calls `reseed()`. `sequence` increments for each effective patch in a generation. `baseSequence` must equal the client's current sequence; otherwise the client has missed an update and resyncs from `snapshot()`. ## Server `LiveStateSource` owns one authoritative state object. Mutate it with `produce()`, not by mutating the original object: ```ts const server = new LiveStateSource( { tasks: [{ id: 'task-1', title: 'Read the plan', done: false }], filter: 'all', }, 1000 ); const cursor = server.produce( (draft) => { draft.tasks.push({ id: 'task-2', title: 'Apply the first patch', done: false }); }, { mutationIds: ['example-add-task'] } ); ``` `produce()` returns the cursor containing the change. If the mutator is a no-op, no update is emitted and the current cursor is returned. `snapshot()` deep clones the current state. `reseed(next?)` replaces the generation, resets sequence to `0`, optionally replaces state, and forces clients to resync on the next observed update. ## Consumers Consumers normally reach live states through a live model contract. Use the contract client when you only need protocol access: ```ts const model = contractClient.conversation.state({ sessionId }, 'state'); const snapshot = await model.snapshot(); const detach = await model.attach((update) => { console.log(update.delta); }); ``` Wrap that live state client handle in `createLiveModelReplicaCache()` when a process wants local state, mutation settling, ref counting, or a downstream `LiveSource`; see [replicas](./replicas.md#model-replicas). The package also has low-level protocol followers internally. They consume snapshots and ordered updates, then resync when: - an update arrives before `seed()`. - `generation` differs from the local generation. - `baseSequence` differs from the local sequence. - Immer patch application throws. - schema validation fails in non-production builds. The reported resync reasons are `generation`, `sequence-gap`, `patch-failed`, and `validation`. Schema validation is skipped when `NODE_ENV === 'production'`; the generation and sequence checks are the primary correctness mechanism. Every follower is constructed with a required resync failure policy that decides what happens when the snapshot refetch itself fails. Wire ships two canned policies: `resyncRetry({ schedule? })` retries on a bounded backoff until success or dispose, and `resyncMarkStale()` gives up while keeping staleness observable (`stale` on the client) until the next update or reattach triggers a fresh attempt. Production replicas use `resyncRetry`. Every failure fires the `resyncFailed` instrumentation event. `refresh()` resolves only when the follower is fresh: concurrent calls coalesce onto the in-flight resync and a trigger arriving mid-reseed re-runs the loop before resolution. Disposing a follower-based client rejects all of its pending waiters — including wait-forever waiters (`timeoutMs <= 0`) — with a disposed error. ## Cursor and Mutation Waiters `waitForCursor(cursor, timeoutMs?)` resolves when the client has reached a cursor. `waitForMutation(mutationId, timeoutMs?)` resolves when an applied update contains the mutation id, or when a seed/resync lands because the snapshot is authoritative. These waiters are used by mutation settling; see [mutations](./mutations.md). See [../../examples/state-kernel/client.ts](../../examples/state-kernel/client.ts) for an end-to-end example.