route Claude transcript content through the MessageBus

The Claude pane owned a TranscriptReader directly via the controller.
Insert a TranscriptPublisher that tails the transcript and republishes
each ConversationItem onto the kernel MessageBus; the view's controller
subscribes through ConversationController.fromBus. The subscription is
attached before the reader's first poll, so the initial tail isn't
missed on the broadcast bus.

This decouples reading from rendering: the team work (T-139/T-140) can
run one observer that publishes per-agent channels while the lead tile
and each teammate tile subscribe independently.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -76,6 +76,10 @@ heading, and (b) bumping `pubspec.yaml` `version:` in the same commit.
 
 ### Changed
 
+- Claude conversation content now flows through the kernel MessageBus —
+  a reader tails the transcript and publishes items; the pane subscribes.
+  Decouples reading from rendering so the upcoming team panels can show
+  one lead plus a tile per teammate (T-137).
 - In-process IPC dispatch swapped for socket loopback (T-127). The
   Flutter UI's `DaemonClient` now talks to its own `IpcServer` over
   the same per-workspace Unix socket the C `clide` client uses — one

lib/builtin/claude/src/claude_pane.dart

@@ -11,6 +11,8 @@ import 'conversation_controller.dart';
 import 'conversation_view.dart';
 import 'session_naming.dart';
 import 'tmux_session.dart' as tmux;
+import 'transcript_publisher.dart';
+import 'transcript_reader.dart';
 
 class ClaudePane extends StatefulWidget {
   const ClaudePane({
@@ -38,6 +40,7 @@ class _ClaudePaneState extends State<ClaudePane> {
 
   StreamSubscription<DaemonEvent>? _eventSub;
   ConversationController? _conversation;
+  TranscriptPublisher? _feed;
   String? _paneId;
   String? _sessionName;
   String? _error;
@@ -60,6 +63,8 @@ class _ClaudePaneState extends State<ClaudePane> {
   void dispose() {
     _conversation?.dispose();
     _conversation = null;
+    unawaited(_feed?.dispose());
+    _feed = null;
     _eventSub?.cancel();
     _eventSub = null;
     final id = _paneId;
@@ -194,9 +199,14 @@ class _ClaudePaneState extends State<ClaudePane> {
     if (!mounted) return;
     _paneId = resp.data['id'] as String?;
     // Render the conversation natively from the transcript (T-137/D-75)
-    // rather than the PTY's TUI output. claude runs in tmux; we tail its
-    // transcript JSONL for the workspace.
-    _conversation = ConversationController.forWorkspace(repoRoot);
+    // rather than the PTY's TUI output. claude runs in tmux; a reader
+    // tails its transcript JSONL and a publisher fans the items onto the
+    // kernel MessageBus, which the view's controller subscribes to. The
+    // subscription is wired before the reader's first poll so the initial
+    // tail is never missed.
+    final messages = _kernel()!.messages;
+    _feed = TranscriptPublisher(messages: messages, reader: TranscriptReader(repoRoot));
+    _conversation = ConversationController.fromBus(messages: messages);
     _subscribe();
     setState(() {});
   }

lib/builtin/claude/src/conversation_controller.dart

@@ -9,7 +9,9 @@ library;
 
 import 'dart:async';
 
+import 'package:clide/builtin/claude/src/transcript_publisher.dart';
 import 'package:clide/builtin/claude/src/transcript_reader.dart';
+import 'package:clide/kernel/src/events/message_bus.dart';
 import 'package:flutter/foundation.dart';
 
 class ConversationController extends ChangeNotifier {
@@ -23,11 +25,18 @@ class ConversationController extends ChangeNotifier {
     _sub = stream.listen(_onItem);
   }
 
-  /// Convenience: build a controller backed by a live [TranscriptReader]
-  /// for [workspacePath].
-  factory ConversationController.forWorkspace(String workspacePath) {
-    final reader = TranscriptReader(workspacePath);
-    return ConversationController(stream: reader.stream, onDispose: reader.dispose);
+  /// Build a controller fed from the kernel [MessageBus] — it consumes
+  /// the [ConversationItem]s a [TranscriptPublisher] writes onto
+  /// [publisher]/[channel]. Decouples the view from the reader so several
+  /// panels can render the same conversation (team work, T-139/T-140).
+  factory ConversationController.fromBus({
+    required MessageBus messages,
+    String channel = ClaudeConversation.leadChannel,
+    Future<void> Function()? onDispose,
+  }) {
+    final stream =
+        messages.subscribe(publisher: ClaudeConversation.publisher, channel: channel).map((m) => m.data[ClaudeConversation.itemKey] as ConversationItem);
+    return ConversationController(stream: stream, onDispose: onDispose);
   }
 
   final Future<void> Function()? _onDispose;

lib/builtin/claude/src/transcript_publisher.dart

@@ -0,0 +1,60 @@
+/// Bridges a [TranscriptReader] onto the kernel [MessageBus] (epic T-132,
+/// D-75).
+///
+/// One reader tails a workspace transcript; this publisher republishes
+/// every [ConversationItem] as a bus [Message]. Any number of Claude
+/// panels can then subscribe to the same conversation via the bus instead
+/// of each owning its own reader — the decoupling the team panels
+/// (T-139/T-140) need, where a single observer feeds the lead tile plus a
+/// tile per teammate.
+library;
+
+import 'dart:async';
+
+import 'package:clide/builtin/claude/src/transcript_reader.dart';
+import 'package:clide/kernel/src/events/message_bus.dart';
+
+/// Bus addressing for Claude conversation content.
+abstract final class ClaudeConversation {
+  /// Publisher id under which conversation items are published.
+  static const publisher = 'builtin.claude';
+
+  /// Channel for the lead (or single) Claude pane's conversation.
+  static const leadChannel = 'conversation';
+
+  /// Channel for a teammate's conversation (team work, T-139/T-140).
+  static String teammateChannel(String agentId) => 'conversation/$agentId';
+
+  /// Key under which the [ConversationItem] travels in a [Message]'s data.
+  static const itemKey = 'item';
+}
+
+class TranscriptPublisher {
+  /// Starts republishing [reader]'s items onto [messages] under
+  /// [ClaudeConversation.publisher] / [channel]. The subscription is
+  /// attached synchronously, so a controller that subscribes before the
+  /// reader's first poll never misses the initial tail.
+  TranscriptPublisher({
+    required MessageBus messages,
+    required TranscriptReader reader,
+    this.channel = ClaudeConversation.leadChannel,
+  })  : _messages = messages,
+        _reader = reader {
+    _sub = _reader.stream.listen((item) {
+      _messages.publish(ClaudeConversation.publisher, channel, {
+        ClaudeConversation.itemKey: item,
+      });
+    });
+  }
+
+  final MessageBus _messages;
+  final TranscriptReader _reader;
+  final String channel;
+  late final StreamSubscription<ConversationItem> _sub;
+
+  /// Stops publishing and tears down the underlying reader.
+  Future<void> dispose() async {
+    await _sub.cancel();
+    await _reader.dispose();
+  }
+}

test/builtin/claude/conversation_view_test.dart

@@ -8,7 +8,9 @@ import 'dart:async';
 
 import 'package:clide/builtin/claude/src/conversation_controller.dart';
 import 'package:clide/builtin/claude/src/conversation_view.dart';
+import 'package:clide/builtin/claude/src/transcript_publisher.dart';
 import 'package:clide/builtin/claude/src/transcript_reader.dart';
+import 'package:clide/kernel/src/events/message_bus.dart';
 import 'package:clide/widgets/widgets.dart';
 import 'package:flutter/services.dart';
 import 'package:flutter_test/flutter_test.dart';
@@ -78,6 +80,35 @@ void main() {
     });
   });
 
+  group('ConversationController.fromBus', () {
+    test('consumes items published on its publisher/channel', () async {
+      final bus = MessageBus();
+      addTearDown(bus.dispose);
+      final c = ConversationController.fromBus(messages: bus);
+      addTearDown(c.dispose);
+
+      bus.publish(ClaudeConversation.publisher, ClaudeConversation.leadChannel, {ClaudeConversation.itemKey: _user('hi')});
+      bus.publish(ClaudeConversation.publisher, ClaudeConversation.leadChannel, {ClaudeConversation.itemKey: _asst('hello')});
+      await Future<void>.delayed(const Duration(milliseconds: 20));
+
+      expect(c.items, hasLength(2));
+      expect(c.items.first, isA<UserMessage>());
+    });
+
+    test('ignores other publishers and channels', () async {
+      final bus = MessageBus();
+      addTearDown(bus.dispose);
+      final c = ConversationController.fromBus(messages: bus);
+      addTearDown(c.dispose);
+
+      bus.publish('someone.else', ClaudeConversation.leadChannel, {ClaudeConversation.itemKey: _user('nope')});
+      bus.publish(ClaudeConversation.publisher, 'conversation/other', {ClaudeConversation.itemKey: _user('nope')});
+      await Future<void>.delayed(const Duration(milliseconds: 20));
+
+      expect(c.items, isEmpty);
+    });
+  });
+
   group('ConversationView', () {
     late KernelFixture f;
     setUp(() async => f = await KernelFixture.create());

test/builtin/claude/transcript_publisher_test.dart

@@ -0,0 +1,83 @@
+/// Tests for TranscriptPublisher — bridges a TranscriptReader onto the
+/// kernel MessageBus (T-137/D-75). Pure Dart: MessageBus + reader have no
+/// Flutter dependency, so this runs under `package:test`.
+library;
+
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:clide/builtin/claude/src/transcript_publisher.dart';
+import 'package:clide/builtin/claude/src/transcript_reader.dart';
+import 'package:clide/kernel/src/events/message_bus.dart';
+import 'package:test/test.dart';
+
+Map<String, dynamic> _userLine(String uuid, String text) => {
+      'type': 'user',
+      'uuid': uuid,
+      'parentUuid': '',
+      'isSidechain': false,
+      'version': '2.1.143',
+      'timestamp': '2026-05-16T08:53:06.708Z',
+      'message': {'role': 'user', 'content': text},
+    };
+
+Map<String, dynamic> _asstLine(String uuid, String text) => {
+      'type': 'assistant',
+      'uuid': uuid,
+      'parentUuid': '',
+      'isSidechain': false,
+      'version': '2.1.143',
+      'timestamp': '2026-05-16T08:53:07.708Z',
+      'message': {
+        'role': 'assistant',
+        'content': [
+          {'type': 'text', 'text': text}
+        ],
+      },
+    };
+
+void main() {
+  group('TranscriptPublisher', () {
+    late Directory base;
+    const workspace = '/pub/ws';
+
+    setUp(() async => base = await Directory.systemTemp.createTemp('transcript_publisher_test_'));
+    tearDown(() async => base.delete(recursive: true));
+
+    test('republishes reader items onto the bus (lead channel + item key)', () async {
+      final dir = Directory('${base.path}/${workspace.replaceAll('/', '-')}');
+      await dir.create(recursive: true);
+      File('${dir.path}/session-abc.jsonl').writeAsStringSync(
+        '${[_userLine('u1', 'hello'), _asstLine('a1', 'hi there')].map(jsonEncode).join('\n')}\n',
+      );
+
+      final bus = MessageBus();
+      addTearDown(bus.dispose);
+      final received = <Message>[];
+      // Subscribe before the publisher starts the reader's first poll.
+      final sub = bus.subscribe(publisher: ClaudeConversation.publisher, channel: ClaudeConversation.leadChannel).listen(received.add);
+
+      final reader = TranscriptReader(
+        workspace,
+        projectsBase: base.path,
+        pollInterval: const Duration(milliseconds: 20),
+      );
+      final pub = TranscriptPublisher(messages: bus, reader: reader);
+
+      await Future<void>.delayed(const Duration(milliseconds: 200));
+      await sub.cancel();
+      await pub.dispose();
+
+      expect(received, hasLength(2));
+      expect(received.every((m) => m.data[ClaudeConversation.itemKey] is ConversationItem), isTrue);
+      final items = received.map((m) => m.data[ClaudeConversation.itemKey]).toList();
+      expect(items.first, isA<UserMessage>());
+      expect((items.first as UserMessage).text, 'hello');
+      expect(items[1], isA<AssistantTextMessage>());
+    });
+
+    test('teammateChannel namespaces by agentId', () {
+      expect(ClaudeConversation.teammateChannel('coder@team-x'), 'conversation/coder@team-x');
+    });
+  });
+}