Skip to content

Commit 354dca8

Browse files
feat(ui): live-capture streaming model + update_fx adoption (#67)
Part 1 of the live-capture view (#59): the effects-channel adoption and the streaming state machine (ADR-0009). Interactive view controls + a config form are part 2. - The app adopts update_fx (main.zig): the effects-capable update runs the pure workspace.update, then issues fx.spawn on start_capture (self-spawning `studio capture … --ndjson`, whose stdout NDJSON lines stream back as capture_line Msgs) and fx.cancel on stop_capture. main captures argv[0] as the self-exe for the spawn argv. - workspace gains a LiveCapture state machine: start/stop/line/exit Msgs; the captured NDJSON stream accumulates (bounded to max_live_bytes — oldest whole lines drop) and the live trace is re-derived from it via loadTrace, reusing the whole engine, so it correlates + detects exactly like an opened trace and the inspector renders it viewport-bounded. Kept off the traces array so tab compaction never disturbs it. - Headless tests: streamed lines append into a growing, correlated trace; status transitions (start/stop/exit reasons); lines gate when not capturing; a fresh start bumps the cancel key and clears the recording. Verified: native test 154/154, native build (ReleaseFast) compiles the fx.spawn app path, zig fmt clean.
1 parent 71f7b08 commit 354dca8

3 files changed

Lines changed: 225 additions & 2 deletions

File tree

‎CURRENT_STATE.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -178,7 +178,7 @@ live timeline with OS notifications on critical failures.
178178
| `repo` (tooling, CI) | ✅ done for S0 |
179179
| `docs` (docs, ADRs) | ✅ done for S0 |
180180
| `ocpp` (engine) | ✅ S2 + ingestion (#29) + reports (#41) + anonymize (#42) + diff (#43) + replay core (#44); O(n) detection pending (#36) |
181-
| `ui` (native views) | ✅ S3 inspector (#27–#32) + replay transport (#44) |
181+
| `ui` (native views) | ✅ S3 inspector (#27–#32) + replay transport (#44); 🔨 live-capture streaming model + `update_fx`/`fx.spawn` (#59 pt1) |
182182
| `capture` (live proxy) | 🔨 S5 in progress: WS transport (#54) + frame decode (#55) + MITM proxy (#56) |
183183
| `cli` (headless) | ✅ inspect/report/diff/anonymize/ci/scenario (#45) + live `capture` (#57) |
184184
| `conformance` | ✅ done for S2 (15/15, `contract-v1`) |

‎src/main.zig‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,33 @@ const InspectorApp = native_sdk.UiApp(Model, Msg);
5858
/// own ingestion policy on the bytes (raised for trusted files in #29).
5959
const max_trace_file_bytes: usize = 256 * 1024 * 1024;
6060

61+
/// The effects-capable update: runs the pure `workspace.update`, then issues the
62+
/// live-capture effects — on start, spawn the capture worker (`studio capture …
63+
/// --ndjson`), whose stdout NDJSON lines stream back as `capture_line` Msgs; on
64+
/// stop, cancel it. See ADR-0009.
65+
fn updateFx(model: *Model, msg: Msg, fx: *InspectorApp.Effects) void {
66+
workspace.update(model, msg);
67+
switch (msg) {
68+
.start_capture => if (model.live.status == .capturing) {
69+
const argv = [_][]const u8{
70+
model.selfExe(), "capture",
71+
"--listen", model.live.listen(),
72+
"--upstream", model.live.upstream(),
73+
"--ndjson",
74+
};
75+
fx.spawn(.{
76+
.key = model.live.key,
77+
.argv = &argv,
78+
.output = .lines,
79+
.on_line = InspectorApp.Effects.lineMsg(.capture_line),
80+
.on_exit = InspectorApp.Effects.exitMsg(.capture_exit),
81+
});
82+
},
83+
.stop_capture => fx.cancel(model.live.key),
84+
else => {},
85+
}
86+
}
87+
6188
pub fn main(init: std.process.Init) !void {
6289
// Second face: if argv names a CLI subcommand, run it to completion and exit
6390
// BEFORE any window/app setup — no GUI is created. A bare trace path (or no
@@ -70,7 +97,7 @@ pub fn main(init: std.process.Init) !void {
7097
.name = "studio",
7198
.scene = shell_scene,
7299
.canvas_label = canvas_label,
73-
.update = update,
100+
.update_fx = updateFx,
74101
.view = view,
75102
});
76103
defer app_state.destroy();
@@ -101,6 +128,7 @@ fn openTracesFromArgs(model: *Model, init: std.process.Init) void {
101128
const alloc = std.heap.page_allocator;
102129
const args = init.minimal.args.toSlice(alloc) catch return;
103130
defer alloc.free(args);
131+
if (args.len > 0) model.setSelfExe(args[0]);
104132
if (args.len <= 1) return;
105133
for (args[1..]) |path| {
106134
const name = std.fs.path.basename(path);

‎src/ui/workspace.zig‎

Lines changed: 195 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@ const parser = ocpp.parser;
1616
const timeline = ocpp.timeline;
1717
const detection = ocpp.detection;
1818
const types = ocpp.types;
19+
const EffectLine = native_sdk.EffectLine;
20+
const EffectExit = native_sdk.EffectExit;
1921

2022
/// Upper bound on simultaneously open traces (workspace tabs). A hard cap keeps
2123
/// the tab strip and per-trace arenas bounded; opening past it is a no-op.
@@ -38,6 +40,15 @@ pub const PayloadCollapse = std.StaticBitSet(max_payload_tree_nodes);
3840
/// Max length of the timeline free-text search query.
3941
pub const max_search_len: usize = 128;
4042

43+
/// Cap on the accumulated live-capture NDJSON recording — bounds the live
44+
/// trace's memory. (An on-disk `--out` recording, when the CLI writes one, is
45+
/// separate and unbounded.) Oldest whole lines drop once past it.
46+
pub const max_live_bytes: usize = 8 * 1024 * 1024;
47+
/// Max bytes for a live-capture endpoint string (`host:port`, `ws://…`).
48+
pub const live_config_len: usize = 128;
49+
/// Display name for the live-capture trace.
50+
pub const live_trace_name = "live capture";
51+
4152
/// Timeline search + filter state. Facets AND-compose; an inactive filter shows
4253
/// the whole timeline. Lives on the `Model` (never memcpy'd) and stores the
4354
/// search query as a fixed buffer + length + caret — no stored slice pointer —
@@ -161,6 +172,45 @@ pub const Msg = union(enum) {
161172
toggle_severity_filter: types.FailureSeverity,
162173
/// Reset every facet and the search query.
163174
clear_filters,
175+
/// Start a live capture using the model's configured endpoints.
176+
start_capture,
177+
/// Stop the running live capture.
178+
stop_capture,
179+
/// One NDJSON event line from the live-capture worker's stdout.
180+
capture_line: EffectLine,
181+
/// The live-capture worker exited (or was cancelled / rejected).
182+
capture_exit: EffectExit,
183+
};
184+
185+
/// The lifecycle of a live-capture session.
186+
pub const CaptureStatus = enum { idle, capturing, stopped, failed };
187+
188+
/// Live-capture state: a streaming proxy session recorded into a growing trace.
189+
/// The worker's NDJSON stream accumulates in `ndjson`; the live `trace` is
190+
/// re-derived from it (reusing the whole engine — `loadTrace`), so the inspector
191+
/// renders it exactly like any opened trace, viewport-bounded for free. Kept off
192+
/// the `traces` array so tab compaction never disturbs it.
193+
pub const LiveCapture = struct {
194+
status: CaptureStatus = .idle,
195+
/// `fx.spawn` identity for this session (for `fx.cancel`); bumped each start.
196+
key: u64 = 0,
197+
listen_buf: [live_config_len]u8 = undefined,
198+
listen_len: usize = 0,
199+
upstream_buf: [live_config_len]u8 = undefined,
200+
upstream_len: usize = 0,
201+
/// Accumulated NDJSON recording (bounded to `max_live_bytes`).
202+
ndjson: std.ArrayList(u8) = .empty,
203+
/// The live trace, re-derived from `ndjson`; owns its own arena.
204+
trace: LoadedTrace = .{},
205+
exit_code: i32 = 0,
206+
exit_reason: ?native_sdk.EffectExitReason = null,
207+
208+
pub fn listen(self: *const LiveCapture) []const u8 {
209+
return self.listen_buf[0..self.listen_len];
210+
}
211+
pub fn upstream(self: *const LiveCapture) []const u8 {
212+
return self.upstream_buf[0..self.upstream_len];
213+
}
164214
};
165215

166216
pub const Model = struct {
@@ -178,6 +228,12 @@ pub const Model = struct {
178228
/// Search + filter applied to the active trace's timeline. Model-owned (not
179229
/// per-trace) so the query buffer keeps a stable address.
180230
filter: Filter = .{},
231+
/// Live-capture state (streaming proxy session).
232+
live: LiveCapture = .{},
233+
/// The running executable's path (argv[0]), for self-spawning the capture
234+
/// worker; set once at startup.
235+
self_exe_buf: [1024]u8 = undefined,
236+
self_exe_len: usize = 0,
181237

182238
// --- derived (never stored) -------------------------------------------
183239

@@ -299,6 +355,89 @@ pub const Model = struct {
299355
self.filter = .{};
300356
}
301357

358+
// --- live capture -----------------------------------------------------
359+
360+
/// Set the live-capture endpoints (before starting). Over-long values clamp.
361+
pub fn configureCapture(self: *Model, listen_addr: []const u8, upstream: []const u8) void {
362+
const ln = @min(listen_addr.len, live_config_len);
363+
@memcpy(self.live.listen_buf[0..ln], listen_addr[0..ln]);
364+
self.live.listen_len = ln;
365+
const un = @min(upstream.len, live_config_len);
366+
@memcpy(self.live.upstream_buf[0..un], upstream[0..un]);
367+
self.live.upstream_len = un;
368+
}
369+
370+
/// Record the running executable's path (argv[0]) for self-spawning.
371+
pub fn setSelfExe(self: *Model, path: []const u8) void {
372+
const n = @min(path.len, self.self_exe_buf.len);
373+
@memcpy(self.self_exe_buf[0..n], path[0..n]);
374+
self.self_exe_len = n;
375+
}
376+
pub fn selfExe(self: *const Model) []const u8 {
377+
return self.self_exe_buf[0..self.self_exe_len];
378+
}
379+
380+
/// Begin a live capture: clear the recording, mark capturing, fresh spawn
381+
/// key. The `fx.spawn` itself is issued by the app's `update_fx`.
382+
pub fn startCapture(self: *Model) void {
383+
if (self.live.status == .capturing) return;
384+
self.live.ndjson.clearRetainingCapacity();
385+
self.live.trace.deinit(self.backing);
386+
self.live.trace = .{};
387+
self.live.key +%= 1;
388+
if (self.live.key == 0) self.live.key = 1;
389+
self.live.exit_reason = null;
390+
self.live.status = .capturing;
391+
}
392+
393+
/// Stop the running capture (marks stopped; `update_fx` issues `fx.cancel`).
394+
pub fn stopCapture(self: *Model) void {
395+
if (self.live.status == .capturing) self.live.status = .stopped;
396+
}
397+
398+
/// Ingest one NDJSON event line from the capture worker: accumulate it
399+
/// (bounded), then re-derive the live trace so the timeline reflects it.
400+
pub fn appendCaptureLine(self: *Model, line: []const u8) void {
401+
if (self.live.status != .capturing) return;
402+
if (line.len == 0) return;
403+
self.live.ndjson.appendSlice(self.backing, line) catch return;
404+
self.live.ndjson.append(self.backing, '\n') catch return;
405+
self.trimLiveBuffer();
406+
self.reloadLiveTrace();
407+
}
408+
409+
/// The capture worker exited: its reason decides the terminal status.
410+
pub fn captureExited(self: *Model, exit: EffectExit) void {
411+
self.live.exit_code = exit.code;
412+
self.live.exit_reason = exit.reason;
413+
self.live.status = switch (exit.reason) {
414+
.cancelled, .exited => .stopped,
415+
.signaled, .rejected, .spawn_failed => .failed,
416+
};
417+
}
418+
419+
/// Drop whole oldest lines from the front once the recording passes the cap.
420+
fn trimLiveBuffer(self: *Model) void {
421+
const items = self.live.ndjson.items;
422+
if (items.len <= max_live_bytes) return;
423+
var cut = items.len - max_live_bytes;
424+
while (cut < items.len and items[cut] != '\n') cut += 1;
425+
if (cut < items.len) cut += 1; // past the newline: keep whole lines
426+
std.mem.copyForwards(u8, items[0 .. items.len - cut], items[cut..]);
427+
self.live.ndjson.items.len -= cut;
428+
}
429+
430+
/// Re-derive the live trace from the accumulated NDJSON, preserving the
431+
/// selected row when it still exists.
432+
fn reloadLiveTrace(self: *Model) void {
433+
const prev = self.live.trace.selected_event;
434+
self.live.trace.deinit(self.backing);
435+
self.live.trace = loadTrace(self.backing, live_trace_name, self.live.ndjson.items);
436+
if (prev) |s| {
437+
if (s < self.live.trace.events.len) self.live.trace.selected_event = s;
438+
}
439+
}
440+
302441
/// Close trace `index`, freeing its arena and compacting the tab list.
303442
pub fn closeTrace(self: *Model, index: usize) void {
304443
if (index >= self.trace_count) return;
@@ -325,6 +464,8 @@ pub const Model = struct {
325464
while (i < self.trace_count) : (i += 1) self.traces[i].deinit(self.backing);
326465
self.trace_count = 0;
327466
self.active = 0;
467+
self.live.trace.deinit(self.backing);
468+
self.live.ndjson.deinit(self.backing);
328469
}
329470
};
330471

@@ -350,6 +491,10 @@ pub fn update(model: *Model, msg: Msg) void {
350491
.toggle_type_filter => |t| model.filter.message_type = if (model.filter.message_type == t) null else t,
351492
.toggle_severity_filter => |s| model.filter.severity = if (model.filter.severity == s) null else s,
352493
.clear_filters => model.clearFilters(),
494+
.start_capture => model.startCapture(),
495+
.stop_capture => model.stopCapture(),
496+
.capture_line => |l| model.appendCaptureLine(l.line),
497+
.capture_exit => |e| model.captureExited(e),
353498
}
354499
}
355500

@@ -680,3 +825,53 @@ test "the workspace is bounded at max_open_traces" {
680825
while (i < max_open_traces + 3) : (i += 1) update(&model, .open_sample);
681826
try testing.expectEqual(max_open_traces, model.trace_count);
682827
}
828+
829+
test "live capture streams NDJSON lines into a growing, correlated trace" {
830+
var model = Model{ .backing = testing.allocator };
831+
defer model.deinitAll();
832+
833+
model.configureCapture("127.0.0.1:9000", "ws://10.0.0.5:9000/ocpp");
834+
update(&model, .start_capture);
835+
try testing.expectEqual(CaptureStatus.capturing, model.live.status);
836+
try testing.expectEqual(@as(usize, 0), model.live.trace.events.len);
837+
838+
// Each line is exactly what `studio capture --ndjson` emits (see proxy.zig).
839+
const lines = [_][]const u8{
840+
"{\"timestamp\":1705312800000,\"direction\":\"CS_TO_CSMS\",\"message\":[2,\"m1\",\"BootNotification\",{}]}",
841+
"{\"timestamp\":1705312800500,\"direction\":\"CSMS_TO_CS\",\"message\":[3,\"m1\",{\"status\":\"Accepted\"}]}",
842+
"{\"timestamp\":1705312801000,\"direction\":\"CS_TO_CSMS\",\"message\":[2,\"m2\",\"Heartbeat\",{}]}",
843+
};
844+
for (lines) |line| update(&model, .{ .capture_line = .{ .key = model.live.key, .line = line } });
845+
846+
try testing.expectEqual(@as(usize, 3), model.live.trace.events.len);
847+
try testing.expectEqualStrings("BootNotification", model.live.trace.events[0].action.?);
848+
try testing.expectEqual(types.Direction.cs_to_csms, model.live.trace.events[0].direction);
849+
// The live trace correlates and detects exactly like an opened trace.
850+
try testing.expect(model.live.trace.sessions.len >= 1);
851+
try testing.expect(!model.live.trace.isError());
852+
}
853+
854+
test "live capture status transitions and gates lines when not capturing" {
855+
var model = Model{ .backing = testing.allocator };
856+
defer model.deinitAll();
857+
858+
update(&model, .start_capture);
859+
try testing.expectEqual(CaptureStatus.capturing, model.live.status);
860+
update(&model, .stop_capture);
861+
try testing.expectEqual(CaptureStatus.stopped, model.live.status);
862+
863+
// A rejected spawn marks the session failed.
864+
update(&model, .start_capture);
865+
update(&model, .{ .capture_exit = .{ .key = model.live.key, .reason = .rejected } });
866+
try testing.expectEqual(CaptureStatus.failed, model.live.status);
867+
868+
// Lines that arrive after the session ends are ignored.
869+
update(&model, .{ .capture_line = .{ .key = model.live.key, .line = "{\"message\":[2,\"m9\",\"Heartbeat\",{}]}" } });
870+
try testing.expectEqual(@as(usize, 0), model.live.trace.events.len);
871+
872+
// A fresh start bumps the cancel key and clears the prior recording.
873+
const prev_key = model.live.key;
874+
update(&model, .start_capture);
875+
try testing.expect(model.live.key != prev_key);
876+
try testing.expectEqual(@as(usize, 0), model.live.ndjson.items.len);
877+
}

0 commit comments

Comments
 (0)