diff --git a/CHANGELOG.md b/CHANGELOG.md index 645de5e1..b1d94d5d 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/lib/builtin/claude/src/claude_pane.dart b/lib/builtin/claude/src/claude_pane.dart index 12bf11c2..d750e8ff 100644 --- a/lib/builtin/claude/src/claude_pane.dart +++ b/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 { StreamSubscription? _eventSub; ConversationController? _conversation; + TranscriptPublisher? _feed; String? _paneId; String? _sessionName; String? _error; @@ -60,6 +63,8 @@ class _ClaudePaneState extends State { 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 { 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(() {}); } diff --git a/lib/builtin/claude/src/conversation_controller.dart b/lib/builtin/claude/src/conversation_controller.dart index 7ab8777a..4687eb0f 100644 --- a/lib/builtin/claude/src/conversation_controller.dart +++ b/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 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 Function()? _onDispose; diff --git a/lib/builtin/claude/src/transcript_publisher.dart b/lib/builtin/claude/src/transcript_publisher.dart new file mode 100644 index 00000000..a14bddf0 --- /dev/null +++ b/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 _sub; + + /// Stops publishing and tears down the underlying reader. + Future dispose() async { + await _sub.cancel(); + await _reader.dispose(); + } +} diff --git a/test/builtin/claude/conversation_view_test.dart b/test/builtin/claude/conversation_view_test.dart index 2650d8a2..14af3de5 100644 --- a/test/builtin/claude/conversation_view_test.dart +++ b/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.delayed(const Duration(milliseconds: 20)); + + expect(c.items, hasLength(2)); + expect(c.items.first, isA()); + }); + + 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.delayed(const Duration(milliseconds: 20)); + + expect(c.items, isEmpty); + }); + }); + group('ConversationView', () { late KernelFixture f; setUp(() async => f = await KernelFixture.create()); diff --git a/test/builtin/claude/transcript_publisher_test.dart b/test/builtin/claude/transcript_publisher_test.dart new file mode 100644 index 00000000..26808533 --- /dev/null +++ b/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 _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 _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 = []; + // 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.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()); + expect((items.first as UserMessage).text, 'hello'); + expect(items[1], isA()); + }); + + test('teammateChannel namespaces by agentId', () { + expect(ClaudeConversation.teammateChannel('coder@team-x'), 'conversation/coder@team-x'); + }); + }); +}