Vim/hyp 184 cli hs follow command for websocket streams

[vimmotions]

hs stream — Live WebSocket Stream CLI + Interactive TUI

Summary

Adds a new hs stream command that connects to a deployed stack's WebSocket and streams live entity data to the terminal. This is the CLI equivalent of what the TypeScript SDK does programmatically — but with filtering, recording, time-travel, and an interactive TUI for exploration.

This was the highest-leverage missing UX feature: users could deploy stacks with hs up but had no way to observe live stream data without writing code.

What's new

Core streaming (hs stream <View> --url <wss://...>)

• Connects to any deployed stack's WebSocket endpoint
• Subscribes to a view using the standard Entity/mode syntax (e.g. PumpfunToken/list)
• Outputs merged entity state as NDJSON (one JSON object per line) to stdout
• --raw mode outputs unmerged WebSocket frames directly
• Pipe-friendly: works with | jq, | head, | grep, etc.
• URL resolution: --url explicit, --stack from hyperstack.toml, or auto-match from config
• Subscription controls: --key, --take, --skip, --no-snapshot, --after

Filtering & triggers

• --where DSL with 10 operators: =, !=, >, >=, <, <=, ~regex, !~regex, ? (exists), !? (not exists)
• Dot-path support for nested fields: --where "info.symbol=TRUMP"
• Multiple --where flags are ANDed together
• --first exits immediately after the first entity matches the filter
• --select projects specific fields: --select "info.name,info.symbol"
• --ops filters by operation type: --ops upsert,patch
• --count shows a running update count on stderr

Agent-friendly output

• --no-dna outputs NO_DNA v1 envelope format with lifecycle events (connected, snapshot_complete, entity_update, disconnected)
• Every TUI interaction has a non-interactive CLI equivalent for agent consumption

Recording & replay

• --save snapshot.json records all frames with timestamps
• --duration 30 auto-stops recording after N seconds
• --load snapshot.json replays through the same merge/filter pipeline (no WebSocket needed)

Entity history & time-travel

• EntityStore tracks per-entity history with a ring buffer (default 1000 entries)
• --history --key <key> outputs full update history as JSON
• --at N --key <key> shows entity state at a specific point in history
• --diff --key <key> shows field-level changes between updates

Interactive TUI (--tui)

Behind --features tui (ratatui + crossterm):

• Split-pane layout: entity list (30%) + JSON detail view (70%)
• JSON syntax coloring: keys in cyan, strings green, numbers magenta, booleans yellow
• Deep search (/): filters entities by searching all values in the JSON tree, not just keys
• Time-travel: h/l step through entity version history, with diff view toggle (d)
• Vim motions: j/k, G/gg, Ctrl+d/Ctrl+u, n (next match), number prefixes (10j)
• Pause/resume (p): freeze the stream while exploring
• Save snapshot (s): dump current recording to JSON file
• Raw frame toggle (r)
• Auto-scrolling list: selection always stays visible in both directions

SDK changes

• deep_merge_with_append made public for reuse
• parse_frame, parse_snapshot_entities, try_parse_subscribed_frame, ClientMessage, SnapshotEntity exported from hyperstack-sdk

Usage examples

# Stream all entities as NDJSON
hs stream PumpfunToken/list --url wss://my-stack.stack.usehyperstack.com

# Find first token with a specific symbol
hs stream PumpfunToken/list --url wss://... --where "info.symbol=TRUMP" --first

# Record 30 seconds, replay later in TUI
hs stream PumpfunToken/list --url wss://... --save capture.json --duration 30
hs stream --load capture.json --tui

# Interactive exploration
hs stream PumpfunToken/list --url wss://... --tui

# Pipe to jq
hs stream PumpfunToken/list --url wss://... --raw | jq '.data[0].data.info.name'

New files

cli/src/commands/stream/
├── mod.rs          # Command args, URL resolution, entry point
├── client.rs       # WebSocket connection, frame processing, replay
├── filter.rs       # --where DSL parser & evaluator (9 unit tests)
├── output.rs       # NDJSON, NO_DNA, raw formatters
├── snapshot.rs     # --save/--load file I/O
├── store.rs        # EntityStore with history ring buffer (5 unit tests)
└── tui/
    ├── mod.rs      # Terminal setup, event loop, key dispatch
    ├── app.rs      # App state, vim motions, entity management
    └── ui.rs       # ratatui layout, JSON coloring, widgets

New dependencies (cli)

• hyperstack-sdk — reuse Frame types, subscription protocol, merge logic
• tokio, futures-util, tokio-tungstenite — async WebSocket client
• ratatui, crossterm — TUI (optional, behind tui feature flag)

Tests

• 14 unit tests: 9 for filter DSL (all operators, nested paths, type coercion), 5 for EntityStore (merge, patch, history, diff)
• Manually tested against my live stack deployment with PumpfunToken/list view

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
hyperstack-docs	Ready	Preview, Comment	Mar 25, 2026 10:35pm

[greptile-apps]

Greptile Summary

This PR adds a comprehensive hs stream command that connects to a deployed stack's WebSocket endpoint and streams live entity data to the terminal. It covers the full feature surface described in the PR: NDJSON/NoDna/raw output modes, a --where filter DSL with 10 operators, --select projection, --save/--load snapshot recording and replay, per-entity history with a ring buffer, and an optional interactive TUI built on ratatui/crossterm. The SDK is extended minimally to export previously internal types needed by the CLI.

• The architecture is cleanly layered: mod.rs handles URL resolution and argument validation, client.rs owns the WebSocket loop and frame dispatch, filter.rs/store.rs/snapshot.rs/output.rs each have a single responsibility, and the TUI is isolated behind a Cargo feature flag.
• 14 unit tests cover all 10 filter operators (including edge cases like absent fields and type coercion) and 5 EntityStore operations (upsert, patch, diff, delete, compute_diff).
• The main non-blocking issues found are: (1) a BufWriter that is flushed on every write in output.rs, negating its batching benefit; (2) two redundant duplicate self.history_anchor = None assignments in tui/app.rs; and (3) crossterm::event::poll used as a blocking call inside an async fn (acceptable with multi-threaded Tokio but not idiomatic).
• No reconnection/retry logic is present — a dropped connection ends the stream permanently, which may surprise users in long-lived streaming scenarios.

Confidence Score: 5/5

• This PR is safe to merge; all identified issues are non-blocking style improvements with no impact on correctness.
• The implementation is large but well-structured, with clear module boundaries, 14 passing unit tests covering filter DSL and EntityStore, atomic snapshot writes, and defensive error handling throughout. No P0/P1 bugs were found — only three P2 style suggestions (BufWriter flush semantics, duplicate assignments in TUI, and synchronous event::poll in async context). The SDK changes are purely additive (visibility widening). The feature is self-contained behind a new subcommand and optional Cargo feature, so there is no regression risk to existing commands.
• No files require special attention; cli/src/commands/stream/output.rs (BufWriter flush) and cli/src/commands/stream/tui/app.rs (duplicate assignments) have the P2 items worth a quick pass before or after merge.

Important Files Changed

Filename	Overview
cli/src/commands/stream/client.rs	Core WebSocket streaming engine: connects, subscribes, processes frames (snapshot/upsert/patch/delete), handles --save recording, --duration timer, Ctrl+C, and a replay path. Logic is well-structured with a shared StreamState. No critical bugs found; Operation::Subscribed is guarded before process_frame so it is never recorded.
cli/src/commands/stream/filter.rs	Filter DSL with 10 operators, two-char precedence correctly handled, dot-path resolution, and select_fields projection. 9 unit tests cover all operators. NotEq/NotRegex intentionally returns true for absent fields (documented in tests).
cli/src/commands/stream/mod.rs	Command entry point: URL resolution (explicit → --stack → entity-name auto-match → single-stack fallback), TUI conflict guards, and subscription builder. Clean layering with good user-facing error messages.
cli/src/commands/stream/output.rs	NDJSON/NoDna/raw formatters and StdoutWriter. BufWriter is flushed on every writeln call, negating its batching benefit for high-throughput streams — either use raw Stdout with a lock, or flush only at natural batch boundaries.
cli/src/commands/stream/snapshot.rs	Frame recorder (100k cap with warning) and player. Uses atomic tmp-then-rename write for safety. Manual JSON streaming avoids holding the full JSON string in memory. SnapshotPlayer::load reads the entire file via read_to_string before parsing; for max-size snapshots (~100k frames × ~1 KB) this could require ~100 MB of RAM.
cli/src/commands/stream/store.rs	Per-entity history ring buffer (VecDeque, 1000-entry cap). at, at_absolute, diff_at, and history methods are consistent: index 0 = latest. 5 unit tests cover upsert, patch, diff, delete, and compute_diff.
cli/src/commands/stream/tui/app.rs	TUI app state, vim motions, entity management, and history browsing. Has two redundant duplicate self.history_anchor = None assignments in HistoryForward (lines 402–404) and HistoryNewest (lines 425–426) — harmless but look like copy-paste artifacts. The compensate_history_anchor logic for ring-buffer eviction is thoughtfully implemented.
cli/src/commands/stream/tui/mod.rs	Terminal setup with panic hook restoration, WS reader task with drop counter, and main event loop. crossterm::event::poll(tick_rate) is a blocking syscall inside an async fn — works with multi-threaded Tokio but consumes a worker thread for up to 50 ms per tick; crossterm's async EventStream would be more correct.
cli/src/commands/stream/tui/ui.rs	Ratatui layout (header/split-pane/timeline/status). JSON syntax coloring uses a heuristic line-by-line approach relying on serde_json pretty-print format. Timeline and status bar are informative without being cluttered.
rust/hyperstack-sdk/src/lib.rs	Re-exports parse_frame, parse_snapshot_entities, try_parse_subscribed_frame, SnapshotEntity, ClientMessage, and deep_merge_with_append from the SDK for use by the CLI — minimal, additive change with no breaking modifications.

Sequence Diagram

sequenceDiagram
    participant User as User / CLI
    participant Mod as stream/mod.rs
    participant Client as client.rs
    participant WS as WebSocket Server
    participant Out as output.rs
    participant Snap as snapshot.rs

    User->>Mod: hs stream View --url wss://...
    Mod->>Mod: resolve_url(), validate_ws_url()
    Mod->>Client: stream(url, view, args)
    Client->>Client: build_state() (parse filters, open recorder)
    Client->>WS: connect_async()
    WS-->>Client: HTTP 101 Upgrade
    Client->>WS: ClientMessage::Subscribe(sub)
    WS-->>Client: Subscribed frame

    loop Live stream
        WS-->>Client: Binary/Text frame
        Client->>Client: parse_frame()
        alt Snapshot frame
            Client->>Client: parse_snapshot_entities()
            Client->>Client: EntityStore::upsert() + entities HashMap
            Client->>Out: emit_entity() → writeln NDJSON/NoDna
        else Upsert/Create
            Client->>Client: entities.insert()
            Client->>Out: emit_entity()
        else Patch
            Client->>Client: deep_merge_with_append()
            Client->>Out: emit_entity()
        else Delete
            Client->>Client: entities.remove()
            Client->>Out: print_delete()
        end
        opt --save active
            Client->>Snap: recorder.record(frame)
        end
        opt --first matched
            Client-->>User: break (exit 0)
        end
        opt --duration elapsed / Ctrl+C
            Client->>WS: ws_tx.close()
            Client-->>User: break
        end
    end

    opt --save path given
        Client->>Snap: recorder.save(path)
        Snap->>Snap: tmp write → atomic rename
    end
    opt --history/--at/--diff
        Client->>Out: output_history_if_requested()
    end

Loading

Comments Outside Diff (4)

1. cli/src/commands/stream/tui/app.rs, line 402-404 (link)

   Duplicate history_anchor assignment

   self.history_anchor = None; is assigned twice back-to-back in both the HistoryForward arm (lines 402–404) and the HistoryNewest arm (lines 425–426). These are dead writes — the second assignment in each block is unreachable in any meaningful sense and is never observed. While this produces no incorrect behavior, it looks like a copy-paste artifact and should be cleaned up to avoid future confusion when modifying this logic.

2. cli/src/commands/stream/tui/app.rs, line 424-427 (link)

   Duplicate history_anchor = None in HistoryNewest

   Same pattern as the duplicate in HistoryForward — self.history_anchor = None; is written twice consecutively. One of these two statements should be removed.

3. cli/src/commands/stream/output.rs, line 24-29 (link)

   BufWriter flushed on every write, negating buffering benefit

   self.inner.flush() is called after every writeln!, which means each line triggers its own write syscall — the same behavior as using io::Stdout directly. BufWriter's purpose is to coalesce multiple small writes into a single syscall; flushing on every write eliminates that benefit and the overhead of the BufWriter wrapper is paid for nothing.

   For the pipe-friendly use case (| jq, | head), per-line flushing is indeed desirable to ensure downstream consumers receive data promptly. If that's the intent, consider either:

   1. Removing BufWriter and using io::Stdout with explicit locking, or
   2. Keeping BufWriter but only flushing periodically (e.g. after each frame batch) and relying on Drop::drop to flush at exit.

   (Flush on drop — already implemented — will then ensure final output is not lost.)

4. cli/src/commands/stream/tui/mod.rs, line 198-199 (link)

   Synchronous event::poll blocks a Tokio worker thread

   crossterm::event::poll(tick_rate) is a blocking system call (it may block up to the full 50 ms timeout). Calling it inside an async fn (run_loop) without spawn_blocking means it occupies a Tokio worker thread for up to 50 ms per iteration. With the default multi-threaded runtime this doesn't stall the WebSocket reader task (it runs on a different thread), but it does reduce the effective thread-pool size during the poll and prevents other Tokio tasks from being scheduled on that thread.

   Consider wrapping the poll + read in tokio::task::spawn_blocking or using crossterm's async event stream (EventStream from the crossterm crate's event-stream feature) to keep the executor fully cooperative.

Prompt To Fix All With AI

This is a comment left during a code review.
Path: cli/src/commands/stream/tui/app.rs
Line: 402-404

Comment:
**Duplicate `history_anchor` assignment**

`self.history_anchor = None;` is assigned twice back-to-back in both the `HistoryForward` arm (lines 402–404) and the `HistoryNewest` arm (lines 425–426). These are dead writes — the second assignment in each block is unreachable in any meaningful sense and is never observed. While this produces no incorrect behavior, it looks like a copy-paste artifact and should be cleaned up to avoid future confusion when modifying this logic.

```suggestion
                self.history_anchor = None;
                self.history_position = 0;
```

How can I resolve this? If you propose a fix, please make it concise.

---

This is a comment left during a code review.
Path: cli/src/commands/stream/tui/app.rs
Line: 424-427

Comment:
**Duplicate `history_anchor = None` in `HistoryNewest`**

Same pattern as the duplicate in `HistoryForward` — `self.history_anchor = None;` is written twice consecutively. One of these two statements should be removed.

```suggestion
                self.history_position = 0;
                self.history_anchor = None;
                self.scroll_offset = 0;
```

How can I resolve this? If you propose a fix, please make it concise.

---

This is a comment left during a code review.
Path: cli/src/commands/stream/output.rs
Line: 24-29

Comment:
**`BufWriter` flushed on every write, negating buffering benefit**

`self.inner.flush()` is called after every `writeln!`, which means each line triggers its own `write` syscall — the same behavior as using `io::Stdout` directly. `BufWriter`'s purpose is to coalesce multiple small writes into a single syscall; flushing on every write eliminates that benefit and the overhead of the `BufWriter` wrapper is paid for nothing.

For the pipe-friendly use case (`| jq`, `| head`), per-line flushing is indeed desirable to ensure downstream consumers receive data promptly. If that's the intent, consider either:
1. Removing `BufWriter` and using `io::Stdout` with explicit locking, or
2. Keeping `BufWriter` but only flushing periodically (e.g. after each frame batch) and relying on `Drop::drop` to flush at exit.

```suggestion
    pub fn writeln(&mut self, line: &str) -> Result<()> {
        writeln!(self.inner, "{}", line)?;
        Ok(())
    }
```
(Flush on drop — already implemented — will then ensure final output is not lost.)

How can I resolve this? If you propose a fix, please make it concise.

---

This is a comment left during a code review.
Path: cli/src/commands/stream/tui/mod.rs
Line: 198-199

Comment:
**Synchronous `event::poll` blocks a Tokio worker thread**

`crossterm::event::poll(tick_rate)` is a blocking system call (it may block up to the full 50 ms timeout). Calling it inside an `async fn` (`run_loop`) without `spawn_blocking` means it occupies a Tokio worker thread for up to 50 ms per iteration. With the default multi-threaded runtime this doesn't stall the WebSocket reader task (it runs on a different thread), but it does reduce the effective thread-pool size during the poll and prevents other Tokio tasks from being scheduled on that thread.

Consider wrapping the poll + read in `tokio::task::spawn_blocking` or using `crossterm`'s async event stream (`EventStream` from the `crossterm` crate's `event-stream` feature) to keep the executor fully cooperative.

How can I resolve this? If you propose a fix, please make it concise.

_{Reviews (32): Last reviewed commit: "fix: swap sort/save key bindings for erg..." | Re-trigger Greptile}

[adiman9]

This is a belter. Works really nice. Few comments:

1. Would be nice to have some way to dynamically sort the entities. By _seq or any other field
2. See below image for how arrays render. Would be nice if we could render inline up to certain width before we break to this format.

3. When the json for an entity overflows the container I want j/k to move me up and down the json, not navigate between entities. This is especially true after i've hit enter to drill into an entity. I'm now in the context of the entity and expect jk to navigate within it 4. When using version history during streaming it jumps me around. I'm assuming its using offset from current version rather than an absolute offset? Or maybe to do with rolling window of 1k, as things roll off we jump? Idk

[adiman9]

I was using hs stream OreRound/latest --url wss://ore.stack.usehyperstack.com --tui btw

[adiman9]

Looks good!