From 5f9c054420d39c7e957598cd98f589df69bbbacc Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Fri, 12 Jun 2026 01:05:08 +0200 Subject: [PATCH] replay-latest ValueStream for session state streams (T-386, T-274) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Broadcast streams drop the current value for late subscribers — the shape behind T-274: the init event fires while spawn() is still awaiting the transcript-tail read, before the pane subscribes, so the status bar stayed blank. New pure-Dart ValueStream (no rxdart — prefer-zero-deps) replays the latest value to each new subscriber; statusStream, busyStream, and pendingPromptStream in the claude builtin now use it. busyStream subscribers see the current state first (seeded false), which the busy test now asserts. Co-Authored-By: Claude Fable 5 --- .pql/changelog/ticket_history/2026-06.sql | 3 + .pql/changelog/tickets/2026-06.sql | 23 ++++ CHANGELOG.md | 6 + .../claude/src/stream_json_session.dart | 9 +- lib/clide.dart | 1 + lib/src/util/value_stream.dart | 70 +++++++++++ .../claude/stream_json_session_test.dart | 19 ++- test/util/value_stream_test.dart | 117 ++++++++++++++++++ 8 files changed, 244 insertions(+), 4 deletions(-) create mode 100644 lib/src/util/value_stream.dart create mode 100644 test/util/value_stream_test.dart diff --git a/.pql/changelog/ticket_history/2026-06.sql b/.pql/changelog/ticket_history/2026-06.sql index 9c486ca1..f43d0ab7 100644 --- a/.pql/changelog/ticket_history/2026-06.sql +++ b/.pql/changelog/ticket_history/2026-06.sql @@ -3994,3 +3994,6 @@ UPDATE 2026-06-08: the active hang did NOT reproduce — clide is running fine i T-361 (done, 2026-06-12) added the session-level building blocks this ticket can reuse: StreamJsonSession now watches the process exit code (SessionEnd with stderr tail, replay-latest via session.end) and the pane surfaces ''claude exited (code N) — /clear to restart''. A resume that dies at spawn now surfaces instead of hanging silently; what remains here is the timeout/fallback for a resume that starts but never produces the init event, and resume-decided-by-content.', NULL, '2026-06-11 22:53:52', '2026-06-11 22:53:52', '2026-06-11 22:53:52', NULL, '5a99913696126412ccc7b18f75d3ec15', 2) ON CONFLICT(hash) DO NOTHING; INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHBN5F0F8SDF15P21DNKT1W', 'status', 'backlog', 'in_progress', NULL, '2026-06-11 22:54:18', '2026-06-11 22:54:18', '2026-06-11 22:54:18', NULL, '2cadeabff0a0a89bbcc06db35dfa2f15', 2) ON CONFLICT(hash) DO NOTHING; INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHBN5F0F8SDF15P21DNKT1W', 'status', 'in_progress', 'done', NULL, '2026-06-11 22:57:42', '2026-06-11 22:57:42', '2026-06-11 22:57:42', NULL, 'cb060646e7e967d607dc013a75ec0a28', 2) ON CONFLICT(hash) DO NOTHING; +INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHD098CV2N73823KX4Z99P4', 'status', 'backlog', 'in_progress', NULL, '2026-06-11 23:01:24', '2026-06-11 23:01:24', '2026-06-11 23:01:24', NULL, '33ef6141e9af3e6957def829e635b138', 2) ON CONFLICT(hash) DO NOTHING; +INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FB0TNQM5W6VN98RQM6S22X28', 'status', 'backlog', 'done', NULL, '2026-06-11 23:04:46', '2026-06-11 23:04:46', '2026-06-11 23:04:46', NULL, '5b0e12802a8f9c0b90f8b08e97f85e29', 2) ON CONFLICT(hash) DO NOTHING; +INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHD098CV2N73823KX4Z99P4', 'status', 'in_progress', 'done', NULL, '2026-06-11 23:04:46', '2026-06-11 23:04:46', '2026-06-11 23:04:46', NULL, '5d8d8c7a17da4894510db957e7af3c3f', 2) ON CONFLICT(hash) DO NOTHING; diff --git a/.pql/changelog/tickets/2026-06.sql b/.pql/changelog/tickets/2026-06.sql index 6ff02135..ae2d3037 100644 --- a/.pql/changelog/tickets/2026-06.sql +++ b/.pql/changelog/tickets/2026-06.sql @@ -4066,3 +4066,26 @@ Fix: generate a token in the lock file (Claude Code /ide lock format has a slot INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHBN5F0F8SDF15P21DNKT1W', 'bug', '06FBHBGHNEQTAEPGNJKN42C1E8', 'MCP HTTP server exposes the full dispatcher with zero auth', 'lib/src/ipc/mcp_server.dart:138-195, started unconditionally at boot (lib/main.dart:174-180). D-71 threat model (another user on the same host must not drive my IDE) is enforced with 0600 on the unix socket — then bypassed wholesale by an unauthenticated localhost HTTP port that, since D-86, serves every clide verb as a tool. Fix: generate a token in the lock file (Claude Code /ide lock format has a slot for it) and require the auth header on every request. Tests: request without token is rejected; token round-trips via the lock file.', 'done', 'critical', NULL, NULL, NULL, '2026-06-11 21:55:13', '2026-06-11 22:57:42', NULL, '14de71ed19e921a370001097887a7e6f', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); +INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHD098CV2N73823KX4Z99P4', 'task', '06FBHBGHNEQTAEPGNJKN42C1E8', 'ValueStream (replay-latest) wrapper; retrofit statusStream, busyStream, pendingPromptStream', 'Broadcast streams that carry STATE (not events) drop the current value for late subscribers. This one shape caused T-274 (status bar blank — root cause appended there), the meta sidebar''s manual compensation, and the prompt-stream''s initialData workaround. + +Fix: write one small ValueStream wrapper (a broadcast stream that replays the latest value to each new subscriber, plus a .value getter) in the kernel; retrofit statusStream, busyStream, and pendingPromptStream in the claude builtin; delete the per-site workarounds it obsoletes. No third-party dep (rxdart) — prefer-zero-deps; the wrapper is ~30 LOC. + +Acceptance: unit tests for the wrapper (late subscriber gets latest value; no value yet = no synthetic emit unless seeded); T-274 repro covered: subscribing after the init event still yields the status. Closing this should make T-274 fixable in one line at the call site.', 'in_progress', 'high', NULL, NULL, NULL, '2026-06-11 22:01:07', '2026-06-11 23:01:24', NULL, '490fa0773537afc3ec2be0756960dc05', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); +INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FB0TNQM5W6VN98RQM6S22X28', 'bug', '06FB0TNQM5TWC00GW0P3X02HZW', 'Claude bottom status bar often stays empty / doesn''t update', 'The bottom status-bar context slot for the Claude pane (model · permission-mode · context, T-145/T-150) frequently renders empty and doesn''t update. The slot is fed by the active pane''s status widget via the focus service (PaneContextStatusItem -> ClidePane.statusWidget in claude_pane.dart), sourced from StreamJsonSession.statusStream (model/permissionMode from the ''system/init'' event; cost/contextWindow from ''result'' events in stream_json_session.dart). + +Repro: open clide; the status line is often blank and stays blank until/unless a turn completes (or never populates). + +Likely suspects to investigate: +- status only published while the pane is the focused contribution (active==true) — if focus isn''t on the Claude pane, the slot clears. +- statusStream may not emit until the first ''result''/''init'' event; a resumed session (--resume) may not re-emit init, so model/mode never arrive. +- _statusWidget returns null when _status.isEmpty AND no skills, so an unstarted/!init session shows nothing. +- focus-slot wiring (FocusTracker) may not re-publish on pane (re)build / session rebind. + +Acceptance: the Claude status line shows model · mode · context promptly after a session starts/resumes and stays current across turns and focus changes; add a test covering the resumed-session (no fresh init) case. + +Root cause found and verified by the 2026-06-11 Fable review (fable-ous.md, epic T-359): statusStream is a plain broadcast controller — the system/init event fires while spawn() is still awaiting a 256KB transcript-tail read, before the pane ever subscribes (lib/builtin/claude/src/claude_pane.dart:322, lib/builtin/claude/src/session_orchestrator.dart:222-225). Fix: seed from session.status on bind, or make the stream replay-latest (see the ValueStream systemic ticket under T-359).', 'done', 'medium', NULL, NULL, NULL, '2026-06-08 07:26:13', '2026-06-11 23:04:46', NULL, '97024e51051e9ac66e65376e953a5faf', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); +INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHD098CV2N73823KX4Z99P4', 'task', '06FBHBGHNEQTAEPGNJKN42C1E8', 'ValueStream (replay-latest) wrapper; retrofit statusStream, busyStream, pendingPromptStream', 'Broadcast streams that carry STATE (not events) drop the current value for late subscribers. This one shape caused T-274 (status bar blank — root cause appended there), the meta sidebar''s manual compensation, and the prompt-stream''s initialData workaround. + +Fix: write one small ValueStream wrapper (a broadcast stream that replays the latest value to each new subscriber, plus a .value getter) in the kernel; retrofit statusStream, busyStream, and pendingPromptStream in the claude builtin; delete the per-site workarounds it obsoletes. No third-party dep (rxdart) — prefer-zero-deps; the wrapper is ~30 LOC. + +Acceptance: unit tests for the wrapper (late subscriber gets latest value; no value yet = no synthetic emit unless seeded); T-274 repro covered: subscribing after the init event still yields the status. Closing this should make T-274 fixable in one line at the call site.', 'done', 'high', NULL, NULL, NULL, '2026-06-11 22:01:07', '2026-06-11 23:04:46', NULL, '9cd3d1777cfbe56c117a91aa3f303855', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); diff --git a/CHANGELOG.md b/CHANGELOG.md index 2be93a3d..c5a2c435 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -84,6 +84,12 @@ heading, and (b) bumping `pubspec.yaml` `version:` in the same commit. pane stops spinning, clears any unanswerable permission prompt, reports the exit in the status line, and logs the stderr tail. (T-361) +- **The Claude status bar populates reliably after a session starts.** The + session's init event often fired before the pane subscribed and the plain + broadcast stream dropped it, leaving the model/mode/context line blank; + session state streams now replay their latest value to late subscribers. + (T-274, T-386) + - **Accepting ExitPlanMode now leaves plan mode in the conversation panel.** Approving Claude's plan (the ExitPlanMode tool) transitioned the underlying session out of plan mode, but clide's tracked permission mode didn't follow, diff --git a/lib/builtin/claude/src/stream_json_session.dart b/lib/builtin/claude/src/stream_json_session.dart index 2c82f4b2..46a02f00 100644 --- a/lib/builtin/claude/src/stream_json_session.dart +++ b/lib/builtin/claude/src/stream_json_session.dart @@ -19,6 +19,7 @@ import 'dart:convert'; import 'dart:io'; import 'package:clide/builtin/claude/src/transcript_reader.dart'; +import 'package:clide/src/util/value_stream.dart'; /// The claude subprocess, abstracted so tests drive it without spawning. /// Fakes `extend` this and override what they drive; the defaults below @@ -251,7 +252,9 @@ class StreamJsonSession { /// round-trips are answered by [_handleMcpMessage]. final List _mcpServers; final _items = StreamController.broadcast(); - final _statusCtl = StreamController.broadcast(); + // State, not events — replay-latest so a subscriber that binds after the + // init event still sees the current status (T-386; root cause of T-274). + final _statusCtl = ValueStream(); final _sessionIdCtl = StreamController.broadcast(); StreamSubscription? _sub; SessionStatus _status = const SessionStatus(); @@ -284,7 +287,7 @@ class StreamJsonSession { /// Prompts awaiting a [resolvePrompt] decision, in arrival order. The head /// is the one currently shown in the composer zone. final _queue = []; - final _pendingCtl = StreamController.broadcast(); + final _pendingCtl = ValueStream.seeded(null); /// tool_use_ids that surfaced as a prompt — the view hides their raw /// tool-use card while pending (it shows as a prompt) but keeps the result. @@ -309,7 +312,7 @@ class StreamJsonSession { /// Whether a turn is in flight (between a send and claude's `result`). Drives /// the composer's Stop affordance. bool _busy = false; - final _busyCtl = StreamController.broadcast(); + final _busyCtl = ValueStream.seeded(false); bool get busy => _busy; Stream get busyStream => _busyCtl.stream; diff --git a/lib/clide.dart b/lib/clide.dart index 074964da..188cec1d 100644 --- a/lib/clide.dart +++ b/lib/clide.dart @@ -26,6 +26,7 @@ export 'src/ipc/paths.dart'; export 'src/ipc/schema_v1.dart'; export 'src/panes/event_sink.dart'; export 'src/panes/pane.dart' show Pane, PaneKind; +export 'src/util/value_stream.dart' show ValueStream; // clideName, clideTagline, clideVersion, clideRepository, clideCommit, // clideDate live in lib/src/build_info.g.dart, regenerated by every diff --git a/lib/src/util/value_stream.dart b/lib/src/util/value_stream.dart new file mode 100644 index 00000000..10cc9b73 --- /dev/null +++ b/lib/src/util/value_stream.dart @@ -0,0 +1,70 @@ +/// Replay-latest broadcast value holder (T-386). +/// +/// Broadcast streams drop the current value for late subscribers — the +/// recurring bug factory behind T-274 (status bar blank because the +/// `system/init` event fired before the pane subscribed) and the +/// per-site `initialData` workarounds. A [ValueStream] carries STATE, +/// not events: every new subscriber immediately receives the latest +/// value (when one exists), then live updates. +/// +/// Pure Dart — usable from the IPC/daemon layer and under `dart test`. +library; + +import 'dart:async'; + +class ValueStream { + ValueStream(); + + ValueStream.seeded(T value) : _value = value, _hasValue = true; + + final StreamController _ctl = StreamController.broadcast(); + T? _value; + bool _hasValue = false; + + /// Whether a value has been added (or seeded) yet. A fresh, unseeded + /// holder replays nothing — subscribers wait for the first [add]. + bool get hasValue => _hasValue; + + /// The latest value, or null before the first [add]. For a nullable + /// [T], disambiguate with [hasValue]. + T? get valueOrNull => _value; + + /// The latest value. Throws [StateError] before the first [add] — + /// callers that can race the first value should use [valueOrNull]. + T get value { + if (!_hasValue) throw StateError('ValueStream has no value yet'); + return _value as T; + } + + void add(T value) { + _value = value; + _hasValue = true; + if (!_ctl.isClosed) _ctl.add(value); + } + + /// A stream that replays the latest value (if any) to its subscriber, + /// then follows live updates. Each access returns a fresh + /// single-subscription stream, so every listener gets its own replay. + Stream get stream { + late StreamController out; + StreamSubscription? sub; + out = StreamController( + onListen: () { + if (_hasValue) out.add(_value as T); + if (_ctl.isClosed) { + out.close(); + return; + } + sub = _ctl.stream.listen(out.add, onError: out.addError, onDone: out.close); + }, + onPause: () => sub?.pause(), + onResume: () => sub?.resume(), + onCancel: () => sub?.cancel(), + ); + return out.stream; + } + + bool get isClosed => _ctl.isClosed; + + Future close() => _ctl.close(); +} diff --git a/test/builtin/claude/stream_json_session_test.dart b/test/builtin/claude/stream_json_session_test.dart index 0195aad2..e3fe618b 100644 --- a/test/builtin/claude/stream_json_session_test.dart +++ b/test/builtin/claude/stream_json_session_test.dart @@ -201,6 +201,21 @@ void main() { expect(statuses, hasLength(1)); }); + // T-274 root cause: the init event fired before the pane subscribed and + // the plain broadcast stream dropped it — the status bar stayed blank. + test('subscribing AFTER the init event still yields the status (T-274/T-386)', () async { + proc.emit(initEvent()); + await Future.delayed(Duration.zero); + + final late = []; + session.statusStream.listen(late.add); + await Future.delayed(Duration.zero); + + expect(late, hasLength(1), reason: 'replay-latest delivers the current status to late binders'); + expect(late.single.model, 'claude-opus-4-7'); + expect(late.single.permissionMode, 'default'); + }); + test('captures the claude session id from the first event carrying it (T-185)', () async { final ids = []; session.sessionIdResolved.listen(ids.add); @@ -647,7 +662,9 @@ void main() { proc.emit(jsonEncode({'type': 'result', 'subtype': 'success'})); await Future.delayed(Duration.zero); expect(session.busy, isFalse); - expect(busy, [true, false]); + // Leading false is the replayed seed — busyStream tells a new + // subscriber the CURRENT state before the live updates (T-386). + expect(busy, [false, true, false]); }); test('dispose kills the process', () async { diff --git a/test/util/value_stream_test.dart b/test/util/value_stream_test.dart new file mode 100644 index 00000000..1a8f9dec --- /dev/null +++ b/test/util/value_stream_test.dart @@ -0,0 +1,117 @@ +/// Tests for [ValueStream] — the replay-latest state holder (T-386). +library; + +import 'package:clide/src/util/value_stream.dart'; +import 'package:test/test.dart'; + +void main() { + test('a late subscriber receives the latest value immediately', () async { + final v = ValueStream(); + v.add(1); + v.add(2); + final got = []; + v.stream.listen(got.add); + await Future.delayed(Duration.zero); + expect(got, [2]); + }); + + test('an unseeded holder replays nothing until the first add', () async { + final v = ValueStream(); + final got = []; + v.stream.listen(got.add); + await Future.delayed(Duration.zero); + expect(got, isEmpty); + v.add(7); + await Future.delayed(Duration.zero); + expect(got, [7]); + }); + + test('seeded constructor provides the initial value', () async { + final v = ValueStream.seeded(false); + expect(v.hasValue, isTrue); + expect(v.value, isFalse); + final got = []; + v.stream.listen(got.add); + await Future.delayed(Duration.zero); + expect(got, [false]); + }); + + test('live updates flow to existing subscribers', () async { + final v = ValueStream(); + final got = []; + v.stream.listen(got.add); + v.add('a'); + v.add('b'); + await Future.delayed(Duration.zero); + expect(got, ['a', 'b']); + }); + + test('each stream access gives every subscriber its own replay', () async { + final v = ValueStream.seeded(5); + final a = []; + final b = []; + v.stream.listen(a.add); + v.stream.listen(b.add); + v.add(6); + await Future.delayed(Duration.zero); + expect(a, [5, 6]); + expect(b, [5, 6]); + }); + + test('value throws before the first add; valueOrNull is null', () { + final v = ValueStream(); + expect(() => v.value, throwsStateError); + expect(v.valueOrNull, isNull); + expect(v.hasValue, isFalse); + }); + + test('a nullable type can hold null as a real value', () async { + final v = ValueStream.seeded(null); + expect(v.hasValue, isTrue); + expect(v.valueOrNull, isNull); + final got = []; + v.stream.listen(got.add); + await Future.delayed(Duration.zero); + expect(got, [null]); + }); + + test('close ends derived streams; a post-close subscriber still gets the replay', () async { + final v = ValueStream(); + v.add(3); + final done = []; + v.stream.listen((_) {}, onDone: () => done.add('a')); + await v.close(); + await Future.delayed(Duration.zero); + expect(done, ['a']); + expect(v.isClosed, isTrue); + + final got = []; + var closed = false; + v.stream.listen(got.add, onDone: () => closed = true); + await Future.delayed(Duration.zero); + expect(got, [3], reason: 'the last value survives close for late readers'); + expect(closed, isTrue); + }); + + test('add after close updates the value without throwing', () { + final v = ValueStream.seeded(1); + v.close(); + v.add(2); + expect(v.value, 2); + }); + + test('pause/resume on a derived stream buffers updates', () async { + final v = ValueStream(); + final got = []; + final sub = v.stream.listen(got.add); + v.add(1); + await Future.delayed(Duration.zero); + sub.pause(); + v.add(2); + await Future.delayed(Duration.zero); + expect(got, [1]); + sub.resume(); + await Future.delayed(Duration.zero); + expect(got, [1, 2]); + }); +}