feat: Add streaming pipe support for async loading in.

crates/pikl-core/src/model/event.rs

@@ -53,6 +53,7 @@ pub enum Action {
     RemoveItems(Vec<usize>),
     ProcessHookResponse(HookResponse),
     CloseMenu,
+    StreamingDone,
 }
 
 /// Broadcast from the menu loop to all subscribers
@@ -81,6 +82,8 @@ pub struct ViewState {
     pub mode: Mode,
     pub selection_count: usize,
     pub multi_enabled: bool,
+    /// True while stdin is still streaming items in.
+    pub streaming: bool,
     /// Column headers for table mode. None when not in
     /// table mode.
     pub columns: Option<Vec<ColumnHeader>>,

crates/pikl-core/src/runtime/input.rs

@@ -8,6 +8,15 @@ use tokio::io::AsyncBufReadExt;
 use crate::error::PiklError;
 use crate::item::Item;
 
+/// Parse a single line into a JSON Value. Tries JSON first,
+/// falls back to wrapping as a string. Used by the streaming
+/// stdin reader in the CLI binary, which sends raw Values
+/// via `Action::AddItems`.
+pub fn parse_line_to_value(line: &str) -> serde_json::Value {
+    serde_json::from_str::<serde_json::Value>(line)
+        .unwrap_or_else(|_| serde_json::Value::String(line.to_string()))
+}
+
 /// Try to parse a line as JSON. Falls back to wrapping
 /// it as a plain-text string. The `label_key` controls
 /// which JSON key is used as the display label for object
@@ -116,4 +125,39 @@ mod tests {
         assert_eq!(items.len(), 1);
         assert_eq!(items[0].label(), "quoted string");
     }
+
+    #[test]
+    fn parse_line_to_value_plain_text() {
+        let v = parse_line_to_value("hello world");
+        assert_eq!(v, serde_json::Value::String("hello world".to_string()));
+    }
+
+    #[test]
+    fn parse_line_to_value_json_object() {
+        let v = parse_line_to_value(r#"{"label": "foo", "count": 42}"#);
+        assert!(v.is_object());
+        assert_eq!(v["label"], "foo");
+        assert_eq!(v["count"], 42);
+    }
+
+    #[test]
+    fn parse_line_to_value_invalid_json() {
+        let v = parse_line_to_value("{not valid json}");
+        assert_eq!(
+            v,
+            serde_json::Value::String("{not valid json}".to_string())
+        );
+    }
+
+    #[test]
+    fn parse_line_to_value_json_string() {
+        let v = parse_line_to_value(r#""quoted""#);
+        assert_eq!(v, serde_json::Value::String("quoted".to_string()));
+    }
+
+    #[test]
+    fn parse_line_to_value_json_number() {
+        let v = parse_line_to_value("42");
+        assert_eq!(v, serde_json::json!(42));
+    }
 }

crates/pikl-core/src/runtime/menu.rs

@@ -154,6 +154,7 @@ pub struct MenuRunner<M: MutableMenu> {
     generation: u64,
     selection: SelectionState,
     visual_anchor: Option<usize>,
+    streaming: bool,
 }
 
 impl<M: MutableMenu> MenuRunner<M> {
@@ -180,6 +181,7 @@ impl<M: MutableMenu> MenuRunner<M> {
             generation: 0,
             selection: SelectionState::new(),
             visual_anchor: None,
+            streaming: false,
         };
         (runner, action_tx)
     }
@@ -312,6 +314,7 @@ impl<M: MutableMenu> MenuRunner<M> {
             mode: self.mode,
             selection_count: self.selection.count(),
             multi_enabled: self.selection.multi_enabled,
+            streaming: self.streaming,
             columns,
             generation: self.generation,
         }
@@ -557,6 +560,11 @@ impl<M: MutableMenu> MenuRunner<M> {
                 self.apply_action(action)
             }
             Action::CloseMenu => ActionOutcome::Closed,
+            Action::StreamingDone => {
+                debug!("streaming done");
+                self.streaming = false;
+                ActionOutcome::Broadcast
+            }
         }
     }
 
@@ -566,6 +574,11 @@ impl<M: MutableMenu> MenuRunner<M> {
         self.mode = mode;
     }
 
+    /// Mark the menu as actively streaming items from stdin.
+    pub fn set_streaming(&mut self, v: bool) {
+        self.streaming = v;
+    }
+
     /// Run the menu event loop. Consumes actions and
     /// broadcasts events.
     ///
@@ -1824,4 +1837,54 @@ mod tests {
         // City "toronto" (7) vs header "City" (4): width = 7
         assert_eq!(cols[2].width, 7);
     }
+
+    // -- Streaming tests --
+
+    #[test]
+    fn streaming_done_sets_streaming_false() {
+        let (mut m, _tx) = test_menu();
+        m.set_streaming(true);
+        m.run_filter();
+        assert!(m.build_view_state().streaming);
+        let outcome = m.apply_action(Action::StreamingDone);
+        assert!(matches!(outcome, ActionOutcome::Broadcast));
+        assert!(!m.build_view_state().streaming);
+    }
+
+    #[test]
+    fn add_items_on_empty_streaming_menu() {
+        let (mut m, _tx) = MenuRunner::new(JsonMenu::new(vec![], "label".to_string()));
+        m.set_streaming(true);
+        m.run_filter();
+        m.apply_action(Action::Resize { height: 20 });
+
+        assert_eq!(m.menu.total(), 0);
+        assert!(m.build_view_state().streaming);
+
+        m.apply_action(Action::AddItems(vec![
+            Value::String("one".into()),
+            Value::String("two".into()),
+        ]));
+        assert_eq!(m.menu.total(), 2);
+        assert_eq!(m.menu.filtered_count(), 2);
+
+        // Confirm on non-empty streaming menu works
+        let outcome = m.apply_action(Action::Confirm);
+        assert!(matches!(outcome, ActionOutcome::Selected { .. }));
+    }
+
+    #[test]
+    fn confirm_on_empty_streaming_menu_is_noop() {
+        let (mut m, _tx) = MenuRunner::new(JsonMenu::new(vec![], "label".to_string()));
+        m.set_streaming(true);
+        m.run_filter();
+        let outcome = m.apply_action(Action::Confirm);
+        assert!(matches!(outcome, ActionOutcome::NoOp));
+    }
+
+    #[test]
+    fn view_state_streaming_defaults_false() {
+        let mut m = ready_menu();
+        assert!(!m.build_view_state().streaming);
+    }
 }

crates/pikl-core/src/script/action_fd/parse.rs

@@ -161,6 +161,7 @@ pub fn parse_action(line_number: usize, line: &str) -> Result<ScriptAction, Scri
         "show-ui" => Ok(ScriptAction::ShowUi),
         "show-tui" => Ok(ScriptAction::ShowTui),
         "show-gui" => Ok(ScriptAction::ShowGui),
+        "streaming-done" => Ok(ScriptAction::Core(Action::StreamingDone)),
         _ => Err(ScriptError {
             line: line_number,
             source_line: line.to_string(),
@@ -678,6 +679,14 @@ mod tests {
         );
     }
 
+    #[test]
+    fn parse_streaming_done() {
+        assert_eq!(
+            parse_action(1, "streaming-done").unwrap_or(ScriptAction::Comment),
+            ScriptAction::Core(Action::StreamingDone)
+        );
+    }
+
     #[test]
     fn parse_set_mode_wrong_case() {
         assert!(parse_action(1, "set-mode Insert").is_err());

crates/pikl-tui/src/lib.rs

@@ -374,7 +374,10 @@ fn render_menu(
         ),
         Span::raw(filter_text),
         Span::styled(
-            format!("  {filtered_count}/{total_count}"),
+            format!(
+                "  {filtered_count}/{total_count}{}",
+                if vs.streaming { "..." } else { "" }
+            ),
             Style::default().fg(Color::DarkGray),
         ),
     ];
@@ -691,6 +694,7 @@ mod tests {
             mode: Mode::Insert,
             selection_count: 0,
             multi_enabled: false,
+            streaming: false,
             columns: None,
             generation: 1,
         }
@@ -1166,6 +1170,7 @@ mod tests {
             mode: Mode::Insert,
             selection_count: 0,
             multi_enabled: false,
+            streaming: false,
             columns: None,
             generation: 1,
         };
@@ -1175,6 +1180,29 @@ mod tests {
         assert!(prompt.contains("0/0"));
     }
 
+    #[test]
+    fn streaming_indicator_shows_ellipsis() {
+        let mut vs = sample_view_state();
+        vs.streaming = true;
+        let backend = render_to_backend(40, 5, &vs, "");
+        let prompt = line_text(&backend, 0);
+        assert!(
+            prompt.contains("..."),
+            "streaming indicator should show '...': {prompt}"
+        );
+    }
+
+    #[test]
+    fn no_streaming_indicator_when_done() {
+        let vs = sample_view_state();
+        let backend = render_to_backend(40, 5, &vs, "");
+        let prompt = line_text(&backend, 0);
+        assert!(
+            !prompt.contains("..."),
+            "should not show '...' when not streaming: {prompt}"
+        );
+    }
+
     #[test]
     fn narrow_viewport_truncates() {
         let vs = sample_view_state();
@@ -1269,6 +1297,7 @@ mod tests {
             mode: Mode::Insert,
             selection_count: 0,
             multi_enabled: false,
+            streaming: false,
             columns: Some(vec![
                 ColumnHeader {
                     display_name: "name".into(),

crates/pikl/src/main.rs

@@ -16,7 +16,7 @@ use pikl_core::error::PiklError;
 use pikl_core::event::{Action, MenuResult, Mode};
 use pikl_core::format::FormatTemplate;
 use pikl_core::hook::{HookEventKind, HookHandler};
-use pikl_core::input::read_items_sync;
+use pikl_core::input::{parse_line_to_value, read_items_sync};
 use pikl_core::item::Item;
 use pikl_core::json_menu::JsonMenu;
 use pikl_core::menu::MenuRunner;
@@ -201,10 +201,6 @@ fn main() {
         std::process::exit(2);
     }
 
-    let timeout = cli
-        .stdin_timeout
-        .unwrap_or(if script.is_some() { 30 } else { 0 });
-
     let input_format = match InputFormat::from_str_opt(&cli.input_format) {
         Some(f) => f,
         None => {
@@ -217,22 +213,55 @@ fn main() {
         }
     };
 
-    let (items, csv_headers) = match read_stdin_with_timeout(timeout, &cli.label_key, input_format)
+    // Detect streaming mode: piped stdin, no action-fd, not CSV/TSV.
-    {
+    let streaming = script.is_none()
-        Ok(r) => r,
+        && !std::io::stdin().is_terminal()
-        Err(e) => {
+        && !matches!(input_format, InputFormat::Csv | InputFormat::Tsv);
-            let _ = writeln!(std::io::stderr().lock(), "pikl: {e}");
+
+    // Save the original stdin fd for the streaming reader before
+    // we dup2 /dev/tty onto fd 0.
+    #[cfg(unix)]
+    let saved_stdin_fd: Option<i32> = if streaming {
+        let fd = unsafe { libc::dup(libc::STDIN_FILENO) };
+        if fd < 0 {
+            let _ = writeln!(
+                std::io::stderr().lock(),
+                "pikl: failed to dup stdin: {}",
+                std::io::Error::last_os_error()
+            );
             std::process::exit(2);
         }
+        Some(fd)
+    } else {
+        None
     };
 
-    if items.is_empty() {
+    let (items, csv_headers) = if streaming {
-        let _ = writeln!(
+        // Streaming mode: start with empty items, read in background later.
-            std::io::stderr().lock(),
+        (Vec::new(), None)
-            "pikl: empty input. nothing to pick from"
+    } else {
-        );
+        let timeout = cli
-        std::process::exit(2);
+            .stdin_timeout
-    }
+            .unwrap_or(if script.is_some() { 30 } else { 0 });
+
+        let (items, csv_headers) =
+            match read_stdin_with_timeout(timeout, &cli.label_key, input_format) {
+                Ok(r) => r,
+                Err(e) => {
+                    let _ = writeln!(std::io::stderr().lock(), "pikl: {e}");
+                    std::process::exit(2);
+                }
+            };
+
+        if items.is_empty() {
+            let _ = writeln!(
+                std::io::stderr().lock(),
+                "pikl: empty input. nothing to pick from"
+            );
+            std::process::exit(2);
+        }
+        (items, csv_headers)
+    };
 
     // Resolve column config: explicit --columns, or auto-generate from CSV headers
     let column_config = if let Some(ref cols) = cli.columns {
@@ -288,12 +317,18 @@ fn main() {
         rt.block_on(run_headless(items, &cli, script, start_mode, column_config))
     } else {
         let history_entries = history.as_ref().map(|h| h.entries().to_vec());
+        #[cfg(unix)]
+        let saved_fd = saved_stdin_fd;
+        #[cfg(not(unix))]
+        let saved_fd: Option<i32> = None;
         rt.block_on(run_interactive(
             items,
             &cli,
             start_mode,
             column_config,
             history_entries,
+            streaming,
+            saved_fd,
         ))
     };
 
@@ -461,11 +496,14 @@ async fn run_interactive(
     start_mode: Mode,
     column_config: Option<ColumnConfig>,
     filter_history: Option<Vec<String>>,
+    streaming: bool,
+    saved_stdin_fd: Option<i32>,
 ) -> Result<MenuResult, PiklError> {
     let (mut menu, action_tx) = MenuRunner::new(build_menu(items, cli, column_config));
     menu.set_initial_mode(start_mode);
     menu.set_multi(cli.multi);
     menu.set_selection_order(cli.selection_order);
+    menu.set_streaming(streaming);
 
     if let Some((_handler, dispatcher)) = build_hook_handler(cli, &action_tx) {
         menu.set_dispatcher(dispatcher);
@@ -509,6 +547,51 @@ async fn run_interactive(
         let _ = signal_tx.send(Action::Cancel).await;
     });
 
+    // Spawn background stdin reader for streaming mode.
+    if let Some(fd) = saved_stdin_fd {
+        let stream_tx = action_tx.clone();
+        tokio::task::spawn_blocking(move || {
+            #[cfg(unix)]
+            {
+                use std::io::BufRead;
+                use std::os::unix::io::FromRawFd;
+
+                let file = unsafe { std::fs::File::from_raw_fd(fd) };
+                let mut reader = std::io::BufReader::new(file);
+                let mut batch = Vec::with_capacity(100);
+                let mut line = String::new();
+
+                loop {
+                    line.clear();
+                    match reader.read_line(&mut line) {
+                        Ok(0) => break,
+                        Ok(_) => {
+                            let trimmed = line.trim_end();
+                            if !trimmed.is_empty() {
+                                batch.push(parse_line_to_value(trimmed));
+                            }
+                            if (batch.len() >= 100 || reader.buffer().is_empty())
+                                && !batch.is_empty()
+                            {
+                                let _ = stream_tx.blocking_send(Action::AddItems(
+                                    std::mem::take(&mut batch),
+                                ));
+                            }
+                        }
+                        Err(e) => {
+                            tracing::warn!(%e, "stdin read error");
+                            break;
+                        }
+                    }
+                }
+                if !batch.is_empty() {
+                    let _ = stream_tx.blocking_send(Action::AddItems(batch));
+                }
+                let _ = stream_tx.blocking_send(Action::StreamingDone);
+            }
+        });
+    }
+
     let tui_handle =
         tokio::spawn(async move { pikl_tui::run(action_tx, event_rx, filter_history).await });
 

docs/DEVPLAN.md

@@ -170,7 +170,7 @@ through hooks and structured I/O. A handler hook can
 receive hover events and emit commands to modify menu
 state.
 
-## Phase 4: Multi-Select
+## Phase 4: Multi-Select ✓
 
 Selection with undo/redo. No marks or registers: filtering
 already handles navigation, and selections surviving filter
@@ -275,21 +275,30 @@ pikl instance over the socket, push items, read state, and
 subscribe to events. A named session remembers filter
 history across invocations.
 
-## Phase 7: Streaming & Watched Sources
+## Phase 7: Streaming Stdin ✓
 
-Live, dynamic menus.
+Auto-detected streaming when stdin is a pipe (not action-fd,
+not CSV/TSV). The menu opens immediately with zero items and
+streams them in as they arrive.
+
+`--watch` was dropped from this phase. External tools like
+`inotifywait -m ~/walls | pikl` compose naturally with
+streaming stdin. No built-in file watcher needed.
 
 **Deliverables:**
-- Async/streaming stdin (items arrive over time, list
+- Auto-detected streaming stdin (items arrive over time,
-  updates progressively)
+  list updates progressively)
-- Streaming output (on-hover events emitted to stdout)
+- `StreamingDone` action signals end of input
-- `--watch path` for file/directory watching
+- `streaming` flag on ViewState for frontend indicators
-- `--watch-extensions` filter
+- TUI shows `...` suffix on the count while streaming
-- notify crate integration (inotify on Linux, FSEvents
+- Background reader with batched `AddItems` (up to 100
-  on macOS)
+  items per batch, flushes immediately for slow sources)
+- `streaming-done` action-fd command for test scripts
+- No new CLI flags: streaming is auto-detected
 
-**Done when:** `find / -name '*.log' | pikl` populates
+**Done when:** `seq 1 10000 | pikl` opens immediately and
-progressively. A watched directory updates the list live.
+items stream in. Slow sources show items appearing one at
+a time with a `...` indicator.
 
 ## Phase 8: GUI Frontend (Wayland + X11)
 

docs/use-cases/wallpaper-picker.md

@@ -126,7 +126,7 @@ bind = $mod, B, exec, wallpicker copy
 | Hook debouncing | 3 | Don't flood hyprctl |
 | Fuzzy filtering | 2 | Filter by filename |
 | Manifest files | 3 | Reusable config |
-| Watched sources | 7 | Live update on new files |
+| Streaming stdin | 7 | Progressive load from find |
 | Sessions | 6 | Remember last position |
 
 ## Open Questions