From 06e03b6fc731de2e61efe904beee31eff54fd0f3 Mon Sep 17 00:00:00 2001 From: sepehr-safari Date: Sun, 12 Jul 2026 17:04:45 +0300 Subject: [PATCH] feat(ui): live-capture streaming model + update_fx adoption MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- CURRENT_STATE.md | 2 +- src/main.zig | 30 ++++++- src/ui/workspace.zig | 195 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 225 insertions(+), 2 deletions(-) diff --git a/CURRENT_STATE.md b/CURRENT_STATE.md index ef8c2d7..c24b4cf 100644 --- a/CURRENT_STATE.md +++ b/CURRENT_STATE.md @@ -178,7 +178,7 @@ live timeline with OS notifications on critical failures. | `repo` (tooling, CI) | ✅ done for S0 | | `docs` (docs, ADRs) | ✅ done for S0 | | `ocpp` (engine) | ✅ S2 + ingestion (#29) + reports (#41) + anonymize (#42) + diff (#43) + replay core (#44); O(n) detection pending (#36) | -| `ui` (native views) | ✅ S3 inspector (#27–#32) + replay transport (#44) | +| `ui` (native views) | ✅ S3 inspector (#27–#32) + replay transport (#44); 🔨 live-capture streaming model + `update_fx`/`fx.spawn` (#59 pt1) | | `capture` (live proxy) | 🔨 S5 in progress: WS transport (#54) + frame decode (#55) + MITM proxy (#56) | | `cli` (headless) | ✅ inspect/report/diff/anonymize/ci/scenario (#45) + live `capture` (#57) | | `conformance` | ✅ done for S2 (15/15, `contract-v1`) | diff --git a/src/main.zig b/src/main.zig index a2755fc..25ee049 100644 --- a/src/main.zig +++ b/src/main.zig @@ -58,6 +58,33 @@ const InspectorApp = native_sdk.UiApp(Model, Msg); /// own ingestion policy on the bytes (raised for trusted files in #29). const max_trace_file_bytes: usize = 256 * 1024 * 1024; +/// The effects-capable update: runs the pure `workspace.update`, then issues the +/// live-capture effects — on start, spawn the capture worker (`studio capture … +/// --ndjson`), whose stdout NDJSON lines stream back as `capture_line` Msgs; on +/// stop, cancel it. See ADR-0009. +fn updateFx(model: *Model, msg: Msg, fx: *InspectorApp.Effects) void { + workspace.update(model, msg); + switch (msg) { + .start_capture => if (model.live.status == .capturing) { + const argv = [_][]const u8{ + model.selfExe(), "capture", + "--listen", model.live.listen(), + "--upstream", model.live.upstream(), + "--ndjson", + }; + fx.spawn(.{ + .key = model.live.key, + .argv = &argv, + .output = .lines, + .on_line = InspectorApp.Effects.lineMsg(.capture_line), + .on_exit = InspectorApp.Effects.exitMsg(.capture_exit), + }); + }, + .stop_capture => fx.cancel(model.live.key), + else => {}, + } +} + pub fn main(init: std.process.Init) !void { // Second face: if argv names a CLI subcommand, run it to completion and exit // 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 { .name = "studio", .scene = shell_scene, .canvas_label = canvas_label, - .update = update, + .update_fx = updateFx, .view = view, }); defer app_state.destroy(); @@ -101,6 +128,7 @@ fn openTracesFromArgs(model: *Model, init: std.process.Init) void { const alloc = std.heap.page_allocator; const args = init.minimal.args.toSlice(alloc) catch return; defer alloc.free(args); + if (args.len > 0) model.setSelfExe(args[0]); if (args.len <= 1) return; for (args[1..]) |path| { const name = std.fs.path.basename(path); diff --git a/src/ui/workspace.zig b/src/ui/workspace.zig index 8cb8236..00fd3a8 100644 --- a/src/ui/workspace.zig +++ b/src/ui/workspace.zig @@ -16,6 +16,8 @@ const parser = ocpp.parser; const timeline = ocpp.timeline; const detection = ocpp.detection; const types = ocpp.types; +const EffectLine = native_sdk.EffectLine; +const EffectExit = native_sdk.EffectExit; /// Upper bound on simultaneously open traces (workspace tabs). A hard cap keeps /// 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); /// Max length of the timeline free-text search query. pub const max_search_len: usize = 128; +/// Cap on the accumulated live-capture NDJSON recording — bounds the live +/// trace's memory. (An on-disk `--out` recording, when the CLI writes one, is +/// separate and unbounded.) Oldest whole lines drop once past it. +pub const max_live_bytes: usize = 8 * 1024 * 1024; +/// Max bytes for a live-capture endpoint string (`host:port`, `ws://…`). +pub const live_config_len: usize = 128; +/// Display name for the live-capture trace. +pub const live_trace_name = "live capture"; + /// Timeline search + filter state. Facets AND-compose; an inactive filter shows /// the whole timeline. Lives on the `Model` (never memcpy'd) and stores the /// search query as a fixed buffer + length + caret — no stored slice pointer — @@ -161,6 +172,45 @@ pub const Msg = union(enum) { toggle_severity_filter: types.FailureSeverity, /// Reset every facet and the search query. clear_filters, + /// Start a live capture using the model's configured endpoints. + start_capture, + /// Stop the running live capture. + stop_capture, + /// One NDJSON event line from the live-capture worker's stdout. + capture_line: EffectLine, + /// The live-capture worker exited (or was cancelled / rejected). + capture_exit: EffectExit, +}; + +/// The lifecycle of a live-capture session. +pub const CaptureStatus = enum { idle, capturing, stopped, failed }; + +/// Live-capture state: a streaming proxy session recorded into a growing trace. +/// The worker's NDJSON stream accumulates in `ndjson`; the live `trace` is +/// re-derived from it (reusing the whole engine — `loadTrace`), so the inspector +/// renders it exactly like any opened trace, viewport-bounded for free. Kept off +/// the `traces` array so tab compaction never disturbs it. +pub const LiveCapture = struct { + status: CaptureStatus = .idle, + /// `fx.spawn` identity for this session (for `fx.cancel`); bumped each start. + key: u64 = 0, + listen_buf: [live_config_len]u8 = undefined, + listen_len: usize = 0, + upstream_buf: [live_config_len]u8 = undefined, + upstream_len: usize = 0, + /// Accumulated NDJSON recording (bounded to `max_live_bytes`). + ndjson: std.ArrayList(u8) = .empty, + /// The live trace, re-derived from `ndjson`; owns its own arena. + trace: LoadedTrace = .{}, + exit_code: i32 = 0, + exit_reason: ?native_sdk.EffectExitReason = null, + + pub fn listen(self: *const LiveCapture) []const u8 { + return self.listen_buf[0..self.listen_len]; + } + pub fn upstream(self: *const LiveCapture) []const u8 { + return self.upstream_buf[0..self.upstream_len]; + } }; pub const Model = struct { @@ -178,6 +228,12 @@ pub const Model = struct { /// Search + filter applied to the active trace's timeline. Model-owned (not /// per-trace) so the query buffer keeps a stable address. filter: Filter = .{}, + /// Live-capture state (streaming proxy session). + live: LiveCapture = .{}, + /// The running executable's path (argv[0]), for self-spawning the capture + /// worker; set once at startup. + self_exe_buf: [1024]u8 = undefined, + self_exe_len: usize = 0, // --- derived (never stored) ------------------------------------------- @@ -299,6 +355,89 @@ pub const Model = struct { self.filter = .{}; } + // --- live capture ----------------------------------------------------- + + /// Set the live-capture endpoints (before starting). Over-long values clamp. + pub fn configureCapture(self: *Model, listen_addr: []const u8, upstream: []const u8) void { + const ln = @min(listen_addr.len, live_config_len); + @memcpy(self.live.listen_buf[0..ln], listen_addr[0..ln]); + self.live.listen_len = ln; + const un = @min(upstream.len, live_config_len); + @memcpy(self.live.upstream_buf[0..un], upstream[0..un]); + self.live.upstream_len = un; + } + + /// Record the running executable's path (argv[0]) for self-spawning. + pub fn setSelfExe(self: *Model, path: []const u8) void { + const n = @min(path.len, self.self_exe_buf.len); + @memcpy(self.self_exe_buf[0..n], path[0..n]); + self.self_exe_len = n; + } + pub fn selfExe(self: *const Model) []const u8 { + return self.self_exe_buf[0..self.self_exe_len]; + } + + /// Begin a live capture: clear the recording, mark capturing, fresh spawn + /// key. The `fx.spawn` itself is issued by the app's `update_fx`. + pub fn startCapture(self: *Model) void { + if (self.live.status == .capturing) return; + self.live.ndjson.clearRetainingCapacity(); + self.live.trace.deinit(self.backing); + self.live.trace = .{}; + self.live.key +%= 1; + if (self.live.key == 0) self.live.key = 1; + self.live.exit_reason = null; + self.live.status = .capturing; + } + + /// Stop the running capture (marks stopped; `update_fx` issues `fx.cancel`). + pub fn stopCapture(self: *Model) void { + if (self.live.status == .capturing) self.live.status = .stopped; + } + + /// Ingest one NDJSON event line from the capture worker: accumulate it + /// (bounded), then re-derive the live trace so the timeline reflects it. + pub fn appendCaptureLine(self: *Model, line: []const u8) void { + if (self.live.status != .capturing) return; + if (line.len == 0) return; + self.live.ndjson.appendSlice(self.backing, line) catch return; + self.live.ndjson.append(self.backing, '\n') catch return; + self.trimLiveBuffer(); + self.reloadLiveTrace(); + } + + /// The capture worker exited: its reason decides the terminal status. + pub fn captureExited(self: *Model, exit: EffectExit) void { + self.live.exit_code = exit.code; + self.live.exit_reason = exit.reason; + self.live.status = switch (exit.reason) { + .cancelled, .exited => .stopped, + .signaled, .rejected, .spawn_failed => .failed, + }; + } + + /// Drop whole oldest lines from the front once the recording passes the cap. + fn trimLiveBuffer(self: *Model) void { + const items = self.live.ndjson.items; + if (items.len <= max_live_bytes) return; + var cut = items.len - max_live_bytes; + while (cut < items.len and items[cut] != '\n') cut += 1; + if (cut < items.len) cut += 1; // past the newline: keep whole lines + std.mem.copyForwards(u8, items[0 .. items.len - cut], items[cut..]); + self.live.ndjson.items.len -= cut; + } + + /// Re-derive the live trace from the accumulated NDJSON, preserving the + /// selected row when it still exists. + fn reloadLiveTrace(self: *Model) void { + const prev = self.live.trace.selected_event; + self.live.trace.deinit(self.backing); + self.live.trace = loadTrace(self.backing, live_trace_name, self.live.ndjson.items); + if (prev) |s| { + if (s < self.live.trace.events.len) self.live.trace.selected_event = s; + } + } + /// Close trace `index`, freeing its arena and compacting the tab list. pub fn closeTrace(self: *Model, index: usize) void { if (index >= self.trace_count) return; @@ -325,6 +464,8 @@ pub const Model = struct { while (i < self.trace_count) : (i += 1) self.traces[i].deinit(self.backing); self.trace_count = 0; self.active = 0; + self.live.trace.deinit(self.backing); + self.live.ndjson.deinit(self.backing); } }; @@ -350,6 +491,10 @@ pub fn update(model: *Model, msg: Msg) void { .toggle_type_filter => |t| model.filter.message_type = if (model.filter.message_type == t) null else t, .toggle_severity_filter => |s| model.filter.severity = if (model.filter.severity == s) null else s, .clear_filters => model.clearFilters(), + .start_capture => model.startCapture(), + .stop_capture => model.stopCapture(), + .capture_line => |l| model.appendCaptureLine(l.line), + .capture_exit => |e| model.captureExited(e), } } @@ -680,3 +825,53 @@ test "the workspace is bounded at max_open_traces" { while (i < max_open_traces + 3) : (i += 1) update(&model, .open_sample); try testing.expectEqual(max_open_traces, model.trace_count); } + +test "live capture streams NDJSON lines into a growing, correlated trace" { + var model = Model{ .backing = testing.allocator }; + defer model.deinitAll(); + + model.configureCapture("127.0.0.1:9000", "ws://10.0.0.5:9000/ocpp"); + update(&model, .start_capture); + try testing.expectEqual(CaptureStatus.capturing, model.live.status); + try testing.expectEqual(@as(usize, 0), model.live.trace.events.len); + + // Each line is exactly what `studio capture --ndjson` emits (see proxy.zig). + const lines = [_][]const u8{ + "{\"timestamp\":1705312800000,\"direction\":\"CS_TO_CSMS\",\"message\":[2,\"m1\",\"BootNotification\",{}]}", + "{\"timestamp\":1705312800500,\"direction\":\"CSMS_TO_CS\",\"message\":[3,\"m1\",{\"status\":\"Accepted\"}]}", + "{\"timestamp\":1705312801000,\"direction\":\"CS_TO_CSMS\",\"message\":[2,\"m2\",\"Heartbeat\",{}]}", + }; + for (lines) |line| update(&model, .{ .capture_line = .{ .key = model.live.key, .line = line } }); + + try testing.expectEqual(@as(usize, 3), model.live.trace.events.len); + try testing.expectEqualStrings("BootNotification", model.live.trace.events[0].action.?); + try testing.expectEqual(types.Direction.cs_to_csms, model.live.trace.events[0].direction); + // The live trace correlates and detects exactly like an opened trace. + try testing.expect(model.live.trace.sessions.len >= 1); + try testing.expect(!model.live.trace.isError()); +} + +test "live capture status transitions and gates lines when not capturing" { + var model = Model{ .backing = testing.allocator }; + defer model.deinitAll(); + + update(&model, .start_capture); + try testing.expectEqual(CaptureStatus.capturing, model.live.status); + update(&model, .stop_capture); + try testing.expectEqual(CaptureStatus.stopped, model.live.status); + + // A rejected spawn marks the session failed. + update(&model, .start_capture); + update(&model, .{ .capture_exit = .{ .key = model.live.key, .reason = .rejected } }); + try testing.expectEqual(CaptureStatus.failed, model.live.status); + + // Lines that arrive after the session ends are ignored. + update(&model, .{ .capture_line = .{ .key = model.live.key, .line = "{\"message\":[2,\"m9\",\"Heartbeat\",{}]}" } }); + try testing.expectEqual(@as(usize, 0), model.live.trace.events.len); + + // A fresh start bumps the cancel key and clears the prior recording. + const prev_key = model.live.key; + update(&model, .start_capture); + try testing.expect(model.live.key != prev_key); + try testing.expectEqual(@as(usize, 0), model.live.ndjson.items.len); +}