replay-latest ValueStream for session state streams (T-386, T-274)

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<T> (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 <noreply@anthropic.com>
This commit is contained in:
2026-06-12 01:05:08 +02:00
co-authored by Claude Fable 5
parent 0e7353bf9c
commit 5f9c054420
8 changed files with 244 additions and 4 deletions
@@ -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; 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', '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 ('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;
+23
View File
@@ -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. 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); 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<T> 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<T> 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);
+6
View File
@@ -84,6 +84,12 @@ heading, and (b) bumping `pubspec.yaml` `version:` in the same commit.
pane stops spinning, clears any unanswerable permission prompt, reports pane stops spinning, clears any unanswerable permission prompt, reports
the exit in the status line, and logs the stderr tail. (T-361) 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.** - **Accepting ExitPlanMode now leaves plan mode in the conversation panel.**
Approving Claude's plan (the ExitPlanMode tool) transitioned the underlying Approving Claude's plan (the ExitPlanMode tool) transitioned the underlying
session out of plan mode, but clide's tracked permission mode didn't follow, session out of plan mode, but clide's tracked permission mode didn't follow,
@@ -19,6 +19,7 @@ import 'dart:convert';
import 'dart:io'; import 'dart:io';
import 'package:clide/builtin/claude/src/transcript_reader.dart'; 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. /// The claude subprocess, abstracted so tests drive it without spawning.
/// Fakes `extend` this and override what they drive; the defaults below /// Fakes `extend` this and override what they drive; the defaults below
@@ -251,7 +252,9 @@ class StreamJsonSession {
/// round-trips are answered by [_handleMcpMessage]. /// round-trips are answered by [_handleMcpMessage].
final List<McpServer> _mcpServers; final List<McpServer> _mcpServers;
final _items = StreamController<ConversationItem>.broadcast(); final _items = StreamController<ConversationItem>.broadcast();
final _statusCtl = StreamController<SessionStatus>.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<SessionStatus>();
final _sessionIdCtl = StreamController<String>.broadcast(); final _sessionIdCtl = StreamController<String>.broadcast();
StreamSubscription<String>? _sub; StreamSubscription<String>? _sub;
SessionStatus _status = const SessionStatus(); SessionStatus _status = const SessionStatus();
@@ -284,7 +287,7 @@ class StreamJsonSession {
/// Prompts awaiting a [resolvePrompt] decision, in arrival order. The head /// Prompts awaiting a [resolvePrompt] decision, in arrival order. The head
/// is the one currently shown in the composer zone. /// is the one currently shown in the composer zone.
final _queue = <ToolPrompt>[]; final _queue = <ToolPrompt>[];
final _pendingCtl = StreamController<ToolPrompt?>.broadcast(); final _pendingCtl = ValueStream<ToolPrompt?>.seeded(null);
/// tool_use_ids that surfaced as a prompt — the view hides their raw /// 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. /// 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 /// Whether a turn is in flight (between a send and claude's `result`). Drives
/// the composer's Stop affordance. /// the composer's Stop affordance.
bool _busy = false; bool _busy = false;
final _busyCtl = StreamController<bool>.broadcast(); final _busyCtl = ValueStream<bool>.seeded(false);
bool get busy => _busy; bool get busy => _busy;
Stream<bool> get busyStream => _busyCtl.stream; Stream<bool> get busyStream => _busyCtl.stream;
+1
View File
@@ -26,6 +26,7 @@ export 'src/ipc/paths.dart';
export 'src/ipc/schema_v1.dart'; export 'src/ipc/schema_v1.dart';
export 'src/panes/event_sink.dart'; export 'src/panes/event_sink.dart';
export 'src/panes/pane.dart' show Pane, PaneKind; export 'src/panes/pane.dart' show Pane, PaneKind;
export 'src/util/value_stream.dart' show ValueStream;
// clideName, clideTagline, clideVersion, clideRepository, clideCommit, // clideName, clideTagline, clideVersion, clideRepository, clideCommit,
// clideDate live in lib/src/build_info.g.dart, regenerated by every // clideDate live in lib/src/build_info.g.dart, regenerated by every
+70
View File
@@ -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<T> {
ValueStream();
ValueStream.seeded(T value) : _value = value, _hasValue = true;
final StreamController<T> _ctl = StreamController<T>.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<T> get stream {
late StreamController<T> out;
StreamSubscription<T>? sub;
out = StreamController<T>(
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<void> close() => _ctl.close();
}
@@ -201,6 +201,21 @@ void main() {
expect(statuses, hasLength(1)); 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<void>.delayed(Duration.zero);
final late = <SessionStatus>[];
session.statusStream.listen(late.add);
await Future<void>.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 { test('captures the claude session id from the first event carrying it (T-185)', () async {
final ids = <String>[]; final ids = <String>[];
session.sessionIdResolved.listen(ids.add); session.sessionIdResolved.listen(ids.add);
@@ -647,7 +662,9 @@ void main() {
proc.emit(jsonEncode({'type': 'result', 'subtype': 'success'})); proc.emit(jsonEncode({'type': 'result', 'subtype': 'success'}));
await Future<void>.delayed(Duration.zero); await Future<void>.delayed(Duration.zero);
expect(session.busy, isFalse); 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 { test('dispose kills the process', () async {
+117
View File
@@ -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<int>();
v.add(1);
v.add(2);
final got = <int>[];
v.stream.listen(got.add);
await Future<void>.delayed(Duration.zero);
expect(got, [2]);
});
test('an unseeded holder replays nothing until the first add', () async {
final v = ValueStream<int>();
final got = <int>[];
v.stream.listen(got.add);
await Future<void>.delayed(Duration.zero);
expect(got, isEmpty);
v.add(7);
await Future<void>.delayed(Duration.zero);
expect(got, [7]);
});
test('seeded constructor provides the initial value', () async {
final v = ValueStream<bool>.seeded(false);
expect(v.hasValue, isTrue);
expect(v.value, isFalse);
final got = <bool>[];
v.stream.listen(got.add);
await Future<void>.delayed(Duration.zero);
expect(got, [false]);
});
test('live updates flow to existing subscribers', () async {
final v = ValueStream<String>();
final got = <String>[];
v.stream.listen(got.add);
v.add('a');
v.add('b');
await Future<void>.delayed(Duration.zero);
expect(got, ['a', 'b']);
});
test('each stream access gives every subscriber its own replay', () async {
final v = ValueStream<int>.seeded(5);
final a = <int>[];
final b = <int>[];
v.stream.listen(a.add);
v.stream.listen(b.add);
v.add(6);
await Future<void>.delayed(Duration.zero);
expect(a, [5, 6]);
expect(b, [5, 6]);
});
test('value throws before the first add; valueOrNull is null', () {
final v = ValueStream<int>();
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<String?>.seeded(null);
expect(v.hasValue, isTrue);
expect(v.valueOrNull, isNull);
final got = <String?>[];
v.stream.listen(got.add);
await Future<void>.delayed(Duration.zero);
expect(got, [null]);
});
test('close ends derived streams; a post-close subscriber still gets the replay', () async {
final v = ValueStream<int>();
v.add(3);
final done = <String>[];
v.stream.listen((_) {}, onDone: () => done.add('a'));
await v.close();
await Future<void>.delayed(Duration.zero);
expect(done, ['a']);
expect(v.isClosed, isTrue);
final got = <int>[];
var closed = false;
v.stream.listen(got.add, onDone: () => closed = true);
await Future<void>.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<int>.seeded(1);
v.close();
v.add(2);
expect(v.value, 2);
});
test('pause/resume on a derived stream buffers updates', () async {
final v = ValueStream<int>();
final got = <int>[];
final sub = v.stream.listen(got.add);
v.add(1);
await Future<void>.delayed(Duration.zero);
sub.pause();
v.add(2);
await Future<void>.delayed(Duration.zero);
expect(got, [1]);
sub.resume();
await Future<void>.delayed(Duration.zero);
expect(got, [1, 2]);
});
}