add pane subsystem to the daemon + IPC event broadcast

Implements the Tier-1 pane subsystem from D-006: spawn / list / focus /
close / write / resize / tail commands, plus pane.spawned / output /
exit / resized / focused / closed events. PaneRegistry owns per-pane
PtySession lifecycles and id generation (p_N); a DaemonEventSink seam
keeps pane code decoupled from the IPC server package.

DaemonServer.broadcast() fans events out to every connected client.
Per-client subsystem/id filtering (`tail --filter pane:p_7`) is
deferred — Tier 1 broadcasts everything and the subscriber discards.

Panes carry a `kind:` field (terminal | claude). Step 7 (builtin.claude)
adds the claude-specific pane flow on top of this generic substrate —
the subsystem itself stays neutral.

14 new core tests: registry unit coverage (spawn → pane.spawned event,
output → base64 events, write/resize/close round-trips, idempotent
close, claude kind on the wire) plus dispatcher coverage (argv
validation, unknown-id → not-found, text vs bytes_b64, etc). All 37
core tests pass in ~3s under test-core.

Co-Authored-By: Claude <noreply@anthropic.com>

lib/src/daemon/pane_commands.dart

@@ -0,0 +1,173 @@
+/// Registers pane.* command handlers on a [DaemonDispatcher].
+///
+/// Verb list matches D-006's subsystem table:
+///   pane.spawn pane.list pane.focus pane.close
+///   pane.write pane.resize pane.tail
+///
+/// `pane.tail` is a no-op on the command surface — events are pushed
+/// over the same socket. The name exists for parity with the CLI
+/// (`clide pane tail --events`) which subscribes to the event stream.
+library;
+
+import 'dart:convert';
+
+import '../ipc/envelope.dart';
+import '../ipc/schema_v1.dart';
+import '../panes/pane.dart';
+import '../panes/registry.dart';
+import 'dispatcher.dart';
+
+void registerPaneCommands(DaemonDispatcher d, PaneRegistry registry) {
+  d.register('pane.spawn', (req) => _spawn(req, registry));
+  d.register('pane.list', (req) => _list(req, registry));
+  d.register('pane.close', (req) => _close(req, registry));
+  d.register('pane.write', (req) => _write(req, registry));
+  d.register('pane.resize', (req) => _resize(req, registry));
+  d.register('pane.focus', (req) => _focus(req, registry));
+  d.register('pane.tail', (req) => _tail(req, registry));
+}
+
+IpcResponse _userErr(String id, String message, {String? hint}) =>
+    IpcResponse.err(
+      id: id,
+      error: IpcError(
+        code: IpcExitCode.userError,
+        kind: IpcErrorKind.userError,
+        message: message,
+        hint: hint,
+      ),
+    );
+
+IpcResponse _notFound(String id, String message) => IpcResponse.err(
+      id: id,
+      error: IpcError(
+        code: IpcExitCode.notFound,
+        kind: IpcErrorKind.notFound,
+        message: message,
+      ),
+    );
+
+Future<IpcResponse> _spawn(IpcRequest req, PaneRegistry registry) async {
+  final args = req.args;
+  final rawArgv = args['argv'];
+  if (rawArgv is! List || rawArgv.isEmpty) {
+    return _userErr(req.id, 'argv is required and non-empty');
+  }
+  final argv = rawArgv.whereType<String>().toList();
+  if (argv.length != rawArgv.length) {
+    return _userErr(req.id, 'argv entries must be strings');
+  }
+
+  final rawKind = args['kind'];
+  PaneKind kind;
+  try {
+    kind = rawKind is String ? PaneKind.parse(rawKind) : PaneKind.terminal;
+  } on ArgumentError catch (e) {
+    return _userErr(req.id, e.message?.toString() ?? 'bad kind');
+  }
+
+  final envArg = args['env'];
+  Map<String, String>? env;
+  if (envArg is Map) {
+    env = {
+      for (final e in envArg.entries)
+        '${e.key}': '${e.value}',
+    };
+  }
+
+  try {
+    final pane = await registry.spawn(
+      kind: kind,
+      argv: argv,
+      cwd: args['cwd'] as String?,
+      env: env,
+      cols: (args['cols'] as num?)?.toInt() ?? 80,
+      rows: (args['rows'] as num?)?.toInt() ?? 24,
+      title: args['title'] as String?,
+      ptycPath: (args['ptyc_path'] as String?) ?? 'ptyc',
+    );
+    return IpcResponse.ok(id: req.id, data: pane.toJson());
+  } catch (e) {
+    return IpcResponse.err(
+      id: req.id,
+      error: IpcError(
+        code: IpcExitCode.toolError,
+        kind: IpcErrorKind.toolError,
+        message: 'pane.spawn failed: $e',
+      ),
+    );
+  }
+}
+
+Future<IpcResponse> _list(IpcRequest req, PaneRegistry registry) async {
+  return IpcResponse.ok(
+    id: req.id,
+    data: {'panes': [for (final p in registry.panes) p.toJson()]},
+  );
+}
+
+Future<IpcResponse> _close(IpcRequest req, PaneRegistry registry) async {
+  final id = req.args['id'] as String?;
+  if (id == null) return _userErr(req.id, 'id is required');
+  if (registry.get(id) == null) return _notFound(req.id, 'no such pane: $id');
+  await registry.close(id);
+  return IpcResponse.ok(id: req.id, data: {'id': id});
+}
+
+Future<IpcResponse> _write(IpcRequest req, PaneRegistry registry) async {
+  final id = req.args['id'] as String?;
+  if (id == null) return _userErr(req.id, 'id is required');
+  if (registry.get(id) == null) return _notFound(req.id, 'no such pane: $id');
+
+  List<int> bytes;
+  final rawBytes = req.args['bytes_b64'];
+  final rawText = req.args['text'];
+  if (rawBytes is String) {
+    try {
+      bytes = base64Decode(rawBytes);
+    } on FormatException {
+      return _userErr(req.id, 'bytes_b64 is not valid base64');
+    }
+  } else if (rawText is String) {
+    bytes = utf8.encode(rawText);
+  } else {
+    return _userErr(req.id, 'write requires bytes_b64 or text');
+  }
+
+  final n = registry.write(id, bytes);
+  return IpcResponse.ok(id: req.id, data: {'id': id, 'written': n});
+}
+
+Future<IpcResponse> _resize(IpcRequest req, PaneRegistry registry) async {
+  final id = req.args['id'] as String?;
+  final cols = (req.args['cols'] as num?)?.toInt();
+  final rows = (req.args['rows'] as num?)?.toInt();
+  if (id == null || cols == null || rows == null) {
+    return _userErr(req.id, 'id, cols, rows are required');
+  }
+  if (registry.get(id) == null) return _notFound(req.id, 'no such pane: $id');
+  registry.resize(id, cols: cols, rows: rows);
+  return IpcResponse.ok(id: req.id, data: {'id': id, 'cols': cols, 'rows': rows});
+}
+
+Future<IpcResponse> _focus(IpcRequest req, PaneRegistry registry) async {
+  final id = req.args['id'] as String?;
+  if (id == null) return _userErr(req.id, 'id is required');
+  if (registry.get(id) == null) return _notFound(req.id, 'no such pane: $id');
+  // Focus is advisory on the daemon side — UIs track their own focus
+  // state. We just emit the event so subscribers know what changed.
+  registry.events.emit(IpcEvent(
+    subsystem: 'pane',
+    kind: 'pane.focused',
+    timestamp: DateTime.now().toUtc(),
+    data: {'id': id},
+  ));
+  return IpcResponse.ok(id: req.id, data: {'id': id});
+}
+
+Future<IpcResponse> _tail(IpcRequest req, PaneRegistry registry) async {
+  // No-op — events are already pushed over the client's socket by the
+  // server's broadcast. The response just acknowledges that the
+  // subscription is in place.
+  return IpcResponse.ok(id: req.id, data: {'subscribed': true});
+}

lib/src/ipc/server.dart

@@ -8,8 +8,7 @@ typedef RequestDispatcher = Future<IpcResponse> Function(IpcRequest request);
 
 /// Unix-socket JSON-lines server. Each connection is an independent
 /// bidirectional line-framed stream: client writes requests, daemon
-/// writes responses (and events, later). Tier 0 handles request→response
-/// only; event broadcasting lands with the first feature that emits.
+/// writes responses + events on the same socket.
 class DaemonServer {
   DaemonServer({
     required this.socketPath,
@@ -22,6 +21,23 @@ class DaemonServer {
   ServerSocket? _server;
   final Set<Socket> _clients = {};
 
+  /// Broadcast [event] to every currently-connected client.
+  ///
+  /// Future tuning: per-client subsystem/id filter (`tail --filter
+  /// pane:p_7`). For Tier 1 every client sees everything. Sockets
+  /// that error on write are silently dropped; the client's read side
+  /// will notice the close.
+  void broadcast(IpcEvent event) {
+    final line = event.encode();
+    for (final c in List<Socket>.from(_clients)) {
+      try {
+        c.writeln(line);
+      } catch (_) {
+        _clients.remove(c);
+      }
+    }
+  }
+
   Future<void> start() async {
     final addr = InternetAddress(socketPath, type: InternetAddressType.unix);
     try {

lib/src/panes/event_sink.dart

@@ -0,0 +1,31 @@
+/// Narrow interface through which subsystems emit events.
+///
+/// The daemon server implements this (broadcasts to every connected
+/// IPC client); tests can provide a recording fake. Keeping the
+/// emitter behind an interface means subsystems don't depend on the
+/// server package, which keeps the dep graph pointing the right way
+/// (server depends on subsystems, not the other way round).
+library;
+
+import '../ipc/envelope.dart';
+
+abstract class DaemonEventSink {
+  void emit(IpcEvent event);
+}
+
+/// In-memory recording sink for tests + for composing multi-sink
+/// scenarios (e.g. tee to both the wire and an audit log).
+class RecordingEventSink implements DaemonEventSink {
+  final List<IpcEvent> events = [];
+
+  @override
+  void emit(IpcEvent event) => events.add(event);
+
+  /// Convenience: filter to a single subsystem (`pane`, `git`, …).
+  Iterable<IpcEvent> ofSubsystem(String subsystem) =>
+      events.where((e) => e.subsystem == subsystem);
+
+  /// Convenience: filter to a specific `type` (`pane.spawned`, …).
+  Iterable<IpcEvent> ofKind(String kind) =>
+      events.where((e) => e.kind == kind);
+}

lib/src/panes/pane.dart

@@ -0,0 +1,60 @@
+/// A single active pane in the daemon.
+///
+/// Owns a [PtySession] plus whatever pane-level metadata the UI +
+/// CLI need. The `kind:` field distinguishes general-purpose terminal
+/// panes from Claude panes (D-041, not yet landed) from whatever
+/// future pane-shaped surface Tier 1+ grows.
+library;
+
+import '../pty/session.dart';
+
+/// Kind of a pane. Keep this enum small and explicit — each kind
+/// typically pairs with a bundled extension that manages its
+/// lifecycle (`builtin.terminal`, `builtin.claude`).
+enum PaneKind {
+  terminal,
+  claude;
+
+  String get wire => name;
+
+  static PaneKind parse(String s) {
+    return PaneKind.values.firstWhere(
+      (v) => v.wire == s,
+      orElse: () => throw ArgumentError.value(s, 'kind', 'unknown pane kind'),
+    );
+  }
+}
+
+/// A live pane. Thin wrapper over [PtySession] — the registry is what
+/// owns the session lifecycle; consumers of this class read state and
+/// call [write] / [resize] via the session.
+class Pane {
+  Pane({
+    required this.id,
+    required this.kind,
+    required this.session,
+    required this.argv,
+    this.cwd,
+    this.title,
+  });
+
+  final String id;
+  final PaneKind kind;
+  final PtySession session;
+  final List<String> argv;
+  final String? cwd;
+  final String? title;
+
+  int get pid => session.pid;
+  bool get isClosed => session.isClosed;
+
+  Map<String, Object?> toJson() => {
+        'id': id,
+        'kind': kind.wire,
+        'pid': pid,
+        'argv': argv,
+        if (cwd != null) 'cwd': cwd,
+        if (title != null) 'title': title,
+        'closed': isClosed,
+      };
+}

lib/src/panes/registry.dart

@@ -0,0 +1,129 @@
+/// [PaneRegistry] — daemon-side state for all live panes.
+///
+/// Owns the [PtySession] per pane, generates `p_N` ids, and forwards
+/// pty output + lifecycle changes as IPC events via a [DaemonEventSink].
+/// Pane commands (pane.spawn / list / write / resize / close) resolve
+/// against this registry; extension UIs subscribe to the emitted events.
+library;
+
+import 'dart:async';
+import 'dart:convert';
+import 'dart:typed_data';
+
+import '../ipc/envelope.dart';
+import '../pty/session.dart';
+import 'event_sink.dart';
+import 'pane.dart';
+
+class PaneRegistry {
+  PaneRegistry({required this.events});
+
+  final DaemonEventSink events;
+  final Map<String, Pane> _panes = {};
+  final Map<String, StreamSubscription<Uint8List>> _subs = {};
+  int _nextId = 1;
+
+  /// All currently-live panes (not yet closed).
+  Iterable<Pane> get panes => _panes.values;
+
+  Pane? get(String id) => _panes[id];
+
+  /// Spawn a child under a PTY and wire its output to events.
+  ///
+  /// [ptycPath] is plumbed through to [PtySession.spawn]; callers that
+  /// have a dev-built `ptyc/bin/ptyc` or a non-PATH install can point
+  /// at it explicitly.
+  Future<Pane> spawn({
+    required PaneKind kind,
+    required List<String> argv,
+    String? cwd,
+    Map<String, String>? env,
+    int cols = 80,
+    int rows = 24,
+    String? title,
+    String ptycPath = 'ptyc',
+  }) async {
+    final id = 'p_${_nextId++}';
+    final session = await PtySession.spawn(
+      argv: argv,
+      cwd: cwd,
+      env: env,
+      cols: cols,
+      rows: rows,
+      ptycPath: ptycPath,
+    );
+    final pane = Pane(
+      id: id,
+      kind: kind,
+      session: session,
+      argv: argv,
+      cwd: cwd,
+      title: title,
+    );
+    _panes[id] = pane;
+
+    _emit('pane.spawned', id, pane.toJson());
+
+    _subs[id] = session.output.listen(
+      (bytes) => _emit('pane.output', id, {
+        'bytes_b64': base64Encode(bytes),
+      }),
+      onDone: () => _onExit(pane),
+    );
+
+    return pane;
+  }
+
+  /// Send bytes to a pane's stdin.
+  int write(String id, List<int> bytes) {
+    final p = _panes[id];
+    if (p == null || p.isClosed) return 0;
+    return p.session.write(bytes);
+  }
+
+  /// Resize a pane + emit `pane.resized`.
+  void resize(String id, {required int cols, required int rows}) {
+    final p = _panes[id];
+    if (p == null || p.isClosed) return;
+    p.session.resize(cols: cols, rows: rows);
+    _emit('pane.resized', id, {'cols': cols, 'rows': rows});
+  }
+
+  /// Close a pane + emit `pane.closed`. Idempotent.
+  Future<void> close(String id) async {
+    final p = _panes[id];
+    if (p == null) return;
+    await p.session.close();
+    await _subs[id]?.cancel();
+    _subs.remove(id);
+    _panes.remove(id);
+    _emit('pane.closed', id, const {});
+  }
+
+  /// Close every pane. Called on daemon shutdown.
+  Future<void> shutdown() async {
+    for (final id in List<String>.from(_panes.keys)) {
+      await close(id);
+    }
+  }
+
+  // -- internals ----------------------------------------------------------
+
+  void _onExit(Pane p) {
+    if (_panes.containsKey(p.id)) {
+      _emit('pane.exit', p.id, const {});
+      // Don't auto-close — keep the pane entry so `list` can show the
+      // exited state until the consumer explicitly closes. A future
+      // tuning knob (keep-vs-reap policy per kind) can change this.
+    }
+  }
+
+  void _emit(String kind, String id, Map<String, Object?> data) {
+    events.emit(IpcEvent(
+      subsystem: 'pane',
+      kind: kind,
+      timestamp: DateTime.now().toUtc(),
+      data: {'id': id, ...data},
+    ));
+  }
+}