From aad4c16d6ead4657ee04996eeee08526136e63d7 Mon Sep 17 00:00:00 2001 From: "J. Champagne" Date: Sat, 14 Mar 2026 17:02:16 -0400 Subject: [PATCH] feat: Add support IPC socket control over menu. --- Cargo.lock | 12 + crates/pikl-core/src/model/event.rs | 10 +- crates/pikl-core/src/runtime/menu.rs | 18 +- crates/pikl-test-macros/src/codegen.rs | 338 ++++++++++++++++--------- crates/pikl-test-macros/src/parse.rs | 16 +- crates/pikl-tui/src/lib.rs | 65 ++++- crates/pikl/Cargo.toml | 3 +- crates/pikl/src/ipc/integration.rs | 120 +++++++++ crates/pikl/src/ipc/mod.rs | 55 ++++ crates/pikl/src/ipc/protocol.rs | 212 ++++++++++++++++ crates/pikl/src/ipc/server.rs | 331 ++++++++++++++++++++++++ crates/pikl/src/main.rs | 83 +++++- crates/pikl/src/session.rs | 210 +++++++++++++++ docs/DESIGN.md | 143 ++++++++--- docs/DEVPLAN.md | 70 ++++- 15 files changed, 1501 insertions(+), 185 deletions(-) create mode 100644 crates/pikl/src/ipc/integration.rs create mode 100644 crates/pikl/src/ipc/mod.rs create mode 100644 crates/pikl/src/ipc/protocol.rs create mode 100644 crates/pikl/src/ipc/server.rs create mode 100644 crates/pikl/src/session.rs diff --git a/Cargo.lock b/Cargo.lock index 5306d90..33ebb06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1100,6 +1100,7 @@ dependencies = [ "pikl-core", "pikl-test-macros", "pikl-tui", + "serde", "serde_json", "tokio", "tracing", @@ -1497,6 +1498,16 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +[[package]] +name = "socket2" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" +dependencies = [ + "libc", + "windows-sys", +] + [[package]] name = "static_assertions" version = "1.1.0" @@ -1696,6 +1707,7 @@ dependencies = [ "mio", "pin-project-lite", "signal-hook-registry", + "socket2", "tokio-macros", "windows-sys", ] diff --git a/crates/pikl-core/src/model/event.rs b/crates/pikl-core/src/model/event.rs index 1164168..eed4283 100644 --- a/crates/pikl-core/src/model/event.rs +++ b/crates/pikl-core/src/model/event.rs @@ -119,7 +119,13 @@ pub struct VisibleItem { #[must_use] #[derive(Debug)] pub enum MenuResult { - Selected { items: Vec<(Value, usize)> }, - Quicklist { items: Vec<(Value, usize)> }, + Selected { + items: Vec<(Value, usize)>, + filter_text: String, + }, + Quicklist { + items: Vec<(Value, usize)>, + filter_text: String, + }, Cancelled, } diff --git a/crates/pikl-core/src/runtime/menu.rs b/crates/pikl-core/src/runtime/menu.rs index 0af7c21..3ea34f3 100644 --- a/crates/pikl-core/src/runtime/menu.rs +++ b/crates/pikl-core/src/runtime/menu.rs @@ -190,6 +190,12 @@ impl MenuRunner { self.event_tx.subscribe() } + /// Get a clone of the event broadcast sender. Useful for + /// IPC servers that need to create their own subscribers. + pub fn event_sender(&self) -> broadcast::Sender { + self.event_tx.clone() + } + /// Set a hook handler. Wraps it in a DebouncedDispatcher /// with no debounce (all events fire immediately). Use /// [`set_dispatcher`] for custom debounce settings. @@ -615,7 +621,10 @@ impl MenuRunner { self.emit_hook(HookEvent::Close); let _ = self.event_tx.send(MenuEvent::Selected(items.clone())); - return Ok(MenuResult::Selected { items }); + return Ok(MenuResult::Selected { + items, + filter_text: self.filter_text.to_string(), + }); } ActionOutcome::Quicklist { items } => { let values: Vec = items.iter().map(|(v, _)| v.clone()).collect(); @@ -628,7 +637,10 @@ impl MenuRunner { self.emit_hook(HookEvent::Close); let _ = self.event_tx.send(MenuEvent::Quicklist(values)); - return Ok(MenuResult::Quicklist { items }); + return Ok(MenuResult::Quicklist { + items, + filter_text: self.filter_text.to_string(), + }); } ActionOutcome::Cancelled => { info!("menu cancelled"); @@ -1360,7 +1372,7 @@ mod tests { let result = handle.await.unwrap_or(Ok(MenuResult::Cancelled)); match result { - Ok(MenuResult::Quicklist { items }) => { + Ok(MenuResult::Quicklist { items, .. }) => { // "alpha" matches "al" assert!(!items.is_empty()); for (v, _) in &items { diff --git a/crates/pikl-test-macros/src/codegen.rs b/crates/pikl-test-macros/src/codegen.rs index b94f0e6..c37416e 100644 --- a/crates/pikl-test-macros/src/codegen.rs +++ b/crates/pikl-test-macros/src/codegen.rs @@ -24,6 +24,7 @@ fn gen_module(module: &TestModule) -> syn::Result { TestKind::Filter => gen_filter(case, &module.fixtures)?, TestKind::Nav => gen_nav(case, &module.fixtures)?, TestKind::Menu => gen_menu(case, &module.fixtures)?, + TestKind::Ipc => gen_ipc(case, &module.fixtures)?, }; test_fns.push(tokens); } @@ -63,6 +64,15 @@ fn gen_imports(kind: TestKind) -> TokenStream { use pikl_core::json_menu::JsonMenu; } } + TestKind::Ipc => { + quote! { + use pikl_core::item::Item; + use pikl_core::event::{Action, MenuEvent, MenuResult}; + use pikl_core::menu::MenuRunner; + use pikl_core::json_menu::JsonMenu; + use crate::ipc::server::IpcServer; + } + } } } @@ -179,8 +189,8 @@ fn build_headless_script(actions: &[ActionExpr]) -> String { script.push_str(line); script.push('\n'); } - ActionExpr::AddItems(_) => { - // Not applicable for headless. Items come from stdin. + ActionExpr::AddItems(_) | ActionExpr::IpcSend(_) | ActionExpr::IpcExpect(_) => { + // Not applicable for headless. } } } @@ -191,18 +201,12 @@ fn build_headless_script(actions: &[ActionExpr]) -> String { // Filter // --------------------------------------------------------------------------- -/// Generate a filter unit test: create items, push them -/// into a `FuzzyFilter`, set the query, and assert on -/// matched labels. fn gen_filter(case: &TestCase, fixtures: &Fixtures) -> syn::Result { let test_name = &case.name; - let item_exprs = gen_item_constructors(fixtures); - let query = case.query.as_deref().unwrap_or(""); let mut asserts = Vec::new(); - if let Some(ref expected) = case.match_labels { asserts.push(quote! { let labels: Vec<&str> = (0..f.matched_count()) @@ -235,11 +239,8 @@ fn gen_filter(case: &TestCase, fixtures: &Fixtures) -> syn::Result // Nav // --------------------------------------------------------------------------- -/// Generate a navigation unit test: create a viewport, run -/// movement actions, and assert on cursor/offset. fn gen_nav(case: &TestCase, fixtures: &Fixtures) -> syn::Result { let test_name = &case.name; - let (height, count) = fixtures.viewport.unwrap_or((10, 20)); let height_lit = height; let count_lit = count; @@ -247,22 +248,14 @@ fn gen_nav(case: &TestCase, fixtures: &Fixtures) -> syn::Result { let action_calls = gen_nav_actions(&case.actions)?; let mut asserts = Vec::new(); - if let Some(cursor) = case.cursor { asserts.push(quote! { - assert_eq!( - v.cursor(), #cursor, - "expected cursor {}, got {}", #cursor, v.cursor() - ); + assert_eq!(v.cursor(), #cursor, "expected cursor {}, got {}", #cursor, v.cursor()); }); } - if let Some(offset) = case.offset { asserts.push(quote! { - assert_eq!( - v.scroll_offset(), #offset, - "expected offset {}, got {}", #offset, v.scroll_offset() - ); + assert_eq!(v.scroll_offset(), #offset, "expected offset {}, got {}", #offset, v.scroll_offset()); }); } @@ -278,8 +271,6 @@ fn gen_nav(case: &TestCase, fixtures: &Fixtures) -> syn::Result { }) } -/// Convert DSL action names to `Viewport` method calls -/// (e.g. `move-down` becomes `v.move_down()`). fn gen_nav_actions(actions: &[ActionExpr]) -> syn::Result> { let mut calls = Vec::new(); for action in actions { @@ -288,12 +279,8 @@ fn gen_nav_actions(actions: &[ActionExpr]) -> syn::Result> { let method = Ident::new(&name.replace('-', "_"), Span::call_site()); let needs_count = matches!( name.as_str(), - "move-up" - | "move-down" - | "page-up" - | "page-down" - | "half-page-up" - | "half-page-down" + "move-up" | "move-down" | "page-up" | "page-down" + | "half-page-up" | "half-page-down" ); if needs_count { calls.push(quote! { v.#method(1); }); @@ -304,10 +291,7 @@ fn gen_nav_actions(actions: &[ActionExpr]) -> syn::Result> { _ => { return Err(syn::Error::new( Span::call_site(), - format!( - "nav tests only support simple actions, got: {:?}", - action_debug(action) - ), + format!("nav tests only support simple actions, got: {:?}", action_debug(action)), )); } } @@ -319,73 +303,12 @@ fn gen_nav_actions(actions: &[ActionExpr]) -> syn::Result> { // Menu // --------------------------------------------------------------------------- -/// Generate an async menu state machine test: create a menu, -/// send actions, and assert on the final result (selected -/// item or cancellation). fn gen_menu(case: &TestCase, fixtures: &Fixtures) -> syn::Result { let test_name = &case.name; let item_exprs = gen_item_constructors(fixtures); - let label_key = fixtures.label_key.as_deref().unwrap_or("label"); - let action_sends = gen_menu_actions(&case.actions)?; - - let result_assert = if case.cancelled { - quote! { - assert!( - matches!(result, Ok(MenuResult::Cancelled)), - "expected Cancelled, got: {:?}", result.as_ref().map(|r| format!("{:?}", r)) - ); - } - } else if let Some(ref expected) = case.selected { - quote! { - match &result { - Ok(MenuResult::Selected { items, .. }) => { - assert!(!items.is_empty(), "expected Selected with items, got empty"); - let value = &items[0].0; - let got = value.as_str() - .or_else(|| value.get(#label_key).and_then(|v| v.as_str())) - .unwrap_or(""); - assert_eq!( - got, #expected, - "expected selected {:?}, got value: {:?}", #expected, value - ); - } - other => panic!("expected Selected, got: {:?}", other), - } - } - } else if let Some(ref expected) = case.selected_items { - quote! { - match &result { - Ok(MenuResult::Selected { items, .. }) => { - let mut labels: Vec<&str> = items.iter() - .filter_map(|(v, _)| v.as_str() - .or_else(|| v.get(#label_key).and_then(|v| v.as_str()))) - .collect(); - labels.sort(); - let mut expected = vec![#(#expected),*]; - expected.sort(); - assert_eq!( - labels, expected, - "expected selected_items {:?}, got {:?}", expected, labels - ); - } - other => panic!("expected Selected, got: {:?}", other), - } - } - } else { - return Err(syn::Error::new( - case.name.span(), - format!( - "menu test '{}' has no result assertion (expected selected, selected_items, or cancelled)", - case.name - ), - )); - }; - - // If test expects cancellation via sender drop (no cancel action, no confirm), - // we need to drop tx after sending actions. - let drop_sender = quote! { drop(tx); }; + let result_assert = gen_result_assert(case, label_key)?; let multi_setup = if fixtures.multi == Some(true) { quote! { menu.set_multi(true); } @@ -402,18 +325,13 @@ fn gen_menu(case: &TestCase, fixtures: &Fixtures) -> syn::Result { let mut rx = menu.subscribe(); let handle = tokio::spawn(async move { menu.run().await }); - // Wait for initial state broadcast. let _ = rx.recv().await; - - // Give the viewport some height so confirms work. let _ = tx.send(Action::Resize { height: 50 }).await; let _ = rx.recv().await; - // Send all actions. #(#action_sends)* - // Drop sender so menu loop can exit. - #drop_sender + drop(tx); let result = handle.await.unwrap_or(Ok(MenuResult::Cancelled)); #result_assert @@ -421,8 +339,6 @@ fn gen_menu(case: &TestCase, fixtures: &Fixtures) -> syn::Result { }) } -/// Convert DSL actions to `tx.send(Action::...)` calls for -/// menu tests. fn gen_menu_actions(actions: &[ActionExpr]) -> syn::Result> { let mut sends = Vec::new(); for action in actions { @@ -432,24 +348,20 @@ fn gen_menu_actions(actions: &[ActionExpr]) -> syn::Result> { quote! { let _ = tx.send(#variant).await; } } ActionExpr::Filter(query) => { - quote! { - let _ = tx.send(Action::UpdateFilter(#query.to_string())).await; - } + quote! { let _ = tx.send(Action::UpdateFilter(#query.to_string())).await; } } ActionExpr::AddItems(items) => { let item_exprs: Vec = items .iter() .map(|s| quote! { serde_json::Value::String(#s.to_string()) }) .collect(); - quote! { - let _ = tx.send(Action::AddItems(vec![#(#item_exprs),*])).await; - } + quote! { let _ = tx.send(Action::AddItems(vec![#(#item_exprs),*])).await; } } ActionExpr::Raw(_) => { - return Err(syn::Error::new( - Span::call_site(), - "raw actions are only supported in headless tests", - )); + return Err(syn::Error::new(Span::call_site(), "raw actions are only supported in headless tests")); + } + ActionExpr::IpcSend(_) | ActionExpr::IpcExpect(_) => { + return Err(syn::Error::new(Span::call_site(), "ipc actions are only supported in ipc tests")); } }; sends.push(expr); @@ -457,8 +369,6 @@ fn gen_menu_actions(actions: &[ActionExpr]) -> syn::Result> { Ok(sends) } -/// Map a DSL action name like `"move-down"` to the -/// corresponding `Action::MoveDown` token stream. fn menu_action_variant(name: &str) -> syn::Result { let tokens = match name { "confirm" => quote! { Action::Confirm }, @@ -480,37 +390,213 @@ fn menu_action_variant(name: &str) -> syn::Result { "undo-selection" => quote! { Action::UndoSelection }, "redo-selection" => quote! { Action::RedoSelection }, _ => { - return Err(syn::Error::new( - Span::call_site(), - format!("unknown menu action: '{name}'"), - )); + return Err(syn::Error::new(Span::call_site(), format!("unknown menu action: '{name}'"))); } }; Ok(tokens) } +// --------------------------------------------------------------------------- +// IPC +// --------------------------------------------------------------------------- + +/// Generate an async IPC integration test: create a menu, +/// start an IPC server, connect via Unix socket, and send/ +/// receive JSON commands. +fn gen_ipc(case: &TestCase, fixtures: &Fixtures) -> syn::Result { + let test_name = &case.name; + let test_name_str = test_name.to_string(); + let item_exprs = gen_item_constructors(fixtures); + let label_key = fixtures.label_key.as_deref().unwrap_or("label"); + + let multi_setup = if fixtures.multi == Some(true) { + quote! { menu.set_multi(true); } + } else { + quote! {} + }; + + let has_expect = case.actions.iter().any(|a| matches!(a, ActionExpr::IpcExpect(_))); + let has_result = case.selected.is_some() || case.selected_items.is_some() || case.cancelled; + if !has_expect && !has_result { + return Err(syn::Error::new( + case.name.span(), + format!( + "ipc test '{}' has no assertions (expected ipc-expect, selected, selected_items, or cancelled)", + case.name + ), + )); + } + + let ipc_actions = gen_ipc_actions(&case.actions)?; + + let result_assert = if has_result { + gen_result_assert(case, label_key)? + } else { + quote! { let _ = result; } + }; + + Ok(quote! { + #[tokio::test] + async fn #test_name() { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + use tokio::net::UnixStream; + use std::time::Duration; + + let items = vec![#(#item_exprs),*]; + let (mut menu, tx) = MenuRunner::new(JsonMenu::new(items, #label_key.to_string())); + #multi_setup + let event_sender = menu.event_sender(); + let ipc_event_rx = menu.subscribe(); + + let handle = tokio::spawn(async move { menu.run().await }); + + let _ = tx.send(Action::Resize { height: 50 }).await; + tokio::time::sleep(Duration::from_millis(30)).await; + + // Start IPC server on a temp socket + let socket_path = std::env::temp_dir().join( + format!("pikl-ipc-{}-{}.sock", std::process::id(), #test_name_str) + ); + let _ = std::fs::remove_file(&socket_path); + + let ipc_server = IpcServer::new( + socket_path.clone(), tx.clone(), event_sender, + ); + assert!(ipc_server.start(ipc_event_rx).is_ok(), "failed to start IPC server"); + + tokio::time::sleep(Duration::from_millis(10)).await; + let stream = match UnixStream::connect(&socket_path).await { + Ok(s) => s, + Err(e) => { + drop(tx); + drop(ipc_server); + let _ = handle.await; + assert!(false, "failed to connect to IPC socket: {e}"); + return; + } + }; + let (read_half, write_half) = stream.into_split(); + let mut reader = BufReader::new(read_half); + let mut writer = write_half; + + #(#ipc_actions)* + + drop(writer); + drop(tx); + drop(ipc_server); + let result = handle.await.unwrap_or(Ok(MenuResult::Cancelled)); + #result_assert + } + }) +} + +fn gen_ipc_actions(actions: &[ActionExpr]) -> syn::Result> { + let mut stmts = Vec::new(); + for action in actions { + match action { + ActionExpr::IpcSend(line) => { + stmts.push(quote! { + assert!(writer.write_all(#line.as_bytes()).await.is_ok(), "ipc write failed"); + assert!(writer.write_all(b"\n").await.is_ok(), "ipc write newline failed"); + assert!(writer.flush().await.is_ok(), "ipc flush failed"); + tokio::time::sleep(Duration::from_millis(30)).await; + }); + } + ActionExpr::IpcExpect(expected) => { + stmts.push(quote! { + { + let mut __resp = String::new(); + let __read = tokio::time::timeout( + Duration::from_secs(5), + reader.read_line(&mut __resp) + ).await; + assert!(__read.is_ok(), "IPC response timed out after 5s"); + let __n = __read.unwrap_or(Ok(0)).unwrap_or(0); + assert!(__n > 0, "IPC connection closed before response"); + assert!( + __resp.contains(#expected), + "expected IPC response to contain {:?}, got: {}", + #expected, __resp.trim() + ); + } + }); + } + _ => { + return Err(syn::Error::new( + Span::call_site(), + format!("ipc tests only support ipc-send and ipc-expect actions, got: {:?}", action_debug(action)), + )); + } + } + } + Ok(stmts) +} + // --------------------------------------------------------------------------- // Shared helpers // --------------------------------------------------------------------------- -/// Generate `Item::from_plain_text("...")` expressions for -/// each item in the fixtures. fn gen_item_constructors(fixtures: &Fixtures) -> Vec { match &fixtures.items { - Some(items) => items - .iter() - .map(|s| quote! { Item::from_plain_text(#s) }) - .collect(), + Some(items) => items.iter().map(|s| quote! { Item::from_plain_text(#s) }).collect(), None => Vec::new(), } } -/// Format an action expression for use in error messages. fn action_debug(action: &ActionExpr) -> String { match action { ActionExpr::Simple(name) => name.clone(), ActionExpr::Filter(q) => format!("filter \"{q}\""), ActionExpr::Raw(r) => format!("raw \"{r}\""), ActionExpr::AddItems(items) => format!("add-items {:?}", items), + ActionExpr::IpcSend(s) => format!("ipc-send \"{s}\""), + ActionExpr::IpcExpect(s) => format!("ipc-expect \"{s}\""), + } +} + +/// Generate the result assertion for menu/ipc tests. +fn gen_result_assert(case: &TestCase, label_key: &str) -> syn::Result { + if case.cancelled { + Ok(quote! { + assert!( + matches!(result, Ok(MenuResult::Cancelled)), + "expected Cancelled, got: {:?}", result.as_ref().map(|r| format!("{:?}", r)) + ); + }) + } else if let Some(ref expected) = case.selected { + Ok(quote! { + match &result { + Ok(MenuResult::Selected { items, .. }) => { + assert!(!items.is_empty(), "expected Selected with items, got empty"); + let value = &items[0].0; + let got = value.as_str() + .or_else(|| value.get(#label_key).and_then(|v| v.as_str())) + .unwrap_or(""); + assert_eq!(got, #expected, "expected selected {:?}, got value: {:?}", #expected, value); + } + other => panic!("expected Selected, got: {:?}", other), + } + }) + } else if let Some(ref expected) = case.selected_items { + Ok(quote! { + match &result { + Ok(MenuResult::Selected { items, .. }) => { + let mut labels: Vec<&str> = items.iter() + .filter_map(|(v, _)| v.as_str() + .or_else(|| v.get(#label_key).and_then(|v| v.as_str()))) + .collect(); + labels.sort(); + let mut expected = vec![#(#expected),*]; + expected.sort(); + assert_eq!(labels, expected, "expected selected_items {:?}, got {:?}", expected, labels); + } + other => panic!("expected Selected, got: {:?}", other), + } + }) + } else { + Err(syn::Error::new( + case.name.span(), + format!("test '{}' has no result assertion (expected selected, selected_items, or cancelled)", case.name), + )) } } diff --git a/crates/pikl-test-macros/src/parse.rs b/crates/pikl-test-macros/src/parse.rs index 2fc8cb8..47ed3ec 100644 --- a/crates/pikl-test-macros/src/parse.rs +++ b/crates/pikl-test-macros/src/parse.rs @@ -25,6 +25,7 @@ pub enum TestKind { Filter, Nav, Menu, + Ipc, } pub struct Fixtures { @@ -58,6 +59,10 @@ pub enum ActionExpr { Raw(String), /// `add-items ["a", "b", "c"]` AddItems(Vec), + /// `ipc-send "json line"` — send a JSON command over the IPC socket + IpcSend(String), + /// `ipc-expect "text"` — read a response line and assert it contains text + IpcExpect(String), } // --------------------------------------------------------------------------- @@ -300,6 +305,14 @@ fn parse_action_expr(input: ParseStream) -> syn::Result { let items = parse_string_list(input)?; Ok(ActionExpr::AddItems(items)) } + "ipc-send" => { + let val: LitStr = input.parse()?; + Ok(ActionExpr::IpcSend(val.value())) + } + "ipc-expect" => { + let val: LitStr = input.parse()?; + Ok(ActionExpr::IpcExpect(val.value())) + } _ => Ok(ActionExpr::Simple(name)), } } @@ -317,9 +330,10 @@ fn parse_kind(input: ParseStream) -> syn::Result { "filter" => Ok(TestKind::Filter), "nav" => Ok(TestKind::Nav), "menu" => Ok(TestKind::Menu), + "ipc" => Ok(TestKind::Ipc), other => Err(syn::Error::new( ident.span(), - format!("unknown test kind '{other}', expected headless, filter, nav, or menu"), + format!("unknown test kind '{other}', expected headless, filter, nav, menu, or ipc"), )), } } diff --git a/crates/pikl-tui/src/lib.rs b/crates/pikl-tui/src/lib.rs index 42eb535..5846027 100644 --- a/crates/pikl-tui/src/lib.rs +++ b/crates/pikl-tui/src/lib.rs @@ -39,6 +39,7 @@ pub fn restore_terminal() { pub async fn run( action_tx: mpsc::Sender, mut event_rx: broadcast::Receiver, + filter_history: Option>, ) -> std::io::Result<()> { crossterm::terminal::enable_raw_mode()?; crossterm::execute!(std::io::stderr(), crossterm::terminal::EnterAlternateScreen,)?; @@ -53,7 +54,7 @@ pub async fn run( let _ = crossterm::event::read()?; } - let result = run_inner(&action_tx, &mut event_rx, &mut terminal).await; + let result = run_inner(&action_tx, &mut event_rx, &mut terminal, filter_history).await; // Always clean up terminal, even on error restore_terminal(); @@ -67,6 +68,7 @@ async fn run_inner( action_tx: &mpsc::Sender, event_rx: &mut broadcast::Receiver, terminal: &mut Terminal>, + filter_history: Option>, ) -> std::io::Result<()> { // Send initial resize let size = terminal.size()?; @@ -95,6 +97,11 @@ async fn run_inner( let mut multi_enabled = false; let mut visual_anchor: Option = None; + // Session filter history cycling state + let history = filter_history.unwrap_or_default(); + let mut history_cursor: Option = None; + let mut pre_history_text: Option = None; + loop { if let Some(ref vs) = view_state { let ft = filter_text.clone(); @@ -112,6 +119,62 @@ async fn run_inner( let event = event_result?; match event { Event::Key(key) => { + // History cycling: Ctrl+P/Ctrl+N in insert mode + // with a non-empty history list. + if mode == Mode::Insert && !history.is_empty() { + let history_handled = match (key.code, key.modifiers) { + (KeyCode::Char('p'), KeyModifiers::CONTROL) | (KeyCode::Up, _) => { + if pre_history_text.is_none() { + pre_history_text = Some(filter_text.clone()); + } + let new_cursor = match history_cursor { + Some(c) if c > 0 => c - 1, + Some(_) => 0, + None => history.len() - 1, + }; + history_cursor = Some(new_cursor); + filter_text.clone_from(&history[new_cursor]); + let _ = action_tx + .send(Action::UpdateFilter(filter_text.clone())) + .await; + true + } + (KeyCode::Char('n'), KeyModifiers::CONTROL) => { + if let Some(c) = history_cursor { + if c + 1 < history.len() { + history_cursor = Some(c + 1); + filter_text.clone_from(&history[c + 1]); + } else { + // Past the end: restore original text + history_cursor = None; + if let Some(ref saved) = pre_history_text { + filter_text.clone_from(saved); + } + pre_history_text = None; + } + let _ = action_tx + .send(Action::UpdateFilter(filter_text.clone())) + .await; + true + } else { + // Not cycling, fall through to normal Ctrl+N + false + } + } + _ => { + // Any other key resets history cycling + if history_cursor.is_some() { + history_cursor = None; + pre_history_text = None; + } + false + } + }; + if history_handled { + continue; + } + } + // Handle multi-action keys (Tab, Space in normal/visual) // before the single-action map_key_event path. let handled = handle_multi_action_key( diff --git a/crates/pikl/Cargo.toml b/crates/pikl/Cargo.toml index 2ff8a4e..5f51d40 100644 --- a/crates/pikl/Cargo.toml +++ b/crates/pikl/Cargo.toml @@ -16,7 +16,8 @@ path = "src/main.rs" pikl-core = { path = "../pikl-core" } pikl-tui = { path = "../pikl-tui" } clap = { version = "4", features = ["derive"] } -tokio = { version = "1", features = ["rt-multi-thread", "process", "signal", "io-util"] } +tokio = { version = "1", features = ["rt-multi-thread", "process", "signal", "io-util", "net"] } +serde = { version = "1", features = ["derive"] } serde_json = "1" tracing = "0.1" tracing-subscriber = "0.3" diff --git a/crates/pikl/src/ipc/integration.rs b/crates/pikl/src/ipc/integration.rs new file mode 100644 index 0000000..7f035a9 --- /dev/null +++ b/crates/pikl/src/ipc/integration.rs @@ -0,0 +1,120 @@ +//! IPC integration tests. Uses the proc macro DSL to set up +//! a menu, start an IPC server, and interact over a Unix socket. + +use pikl_test_macros::pikl_tests; + +pikl_tests! { + ipc mod ipc_get_state { + items: ["alpha", "beta", "gamma"]; + + test get_state_returns_item_count { + actions: [ + ipc-send r#"{"action":"get_state","id":"1"}"#, + ipc-expect r#""total_items":3"#, + ] + } + + test get_state_returns_filter_text { + actions: [ + ipc-send r#"{"action":"get_state","id":"1"}"#, + ipc-expect r#""filter_text":"""#, + ] + } + } + + ipc mod ipc_set_filter { + items: ["alpha", "beta", "gamma"]; + + test set_filter_then_get_state { + actions: [ + ipc-send r#"{"action":"set_filter","text":"al"}"#, + ipc-send r#"{"action":"get_state","id":"1"}"#, + ipc-expect r#""total_filtered":1"#, + ] + } + } + + ipc mod ipc_add_items { + items: ["alpha", "beta"]; + + test add_items_increases_count { + actions: [ + ipc-send r#"{"action":"add_items","items":["delta","epsilon"]}"#, + ipc-send r#"{"action":"get_state","id":"1"}"#, + ipc-expect r#""total_items":4"#, + ] + } + } + + ipc mod ipc_navigation { + items: ["alpha", "beta", "gamma"]; + + test move_down_changes_cursor { + actions: [ + ipc-send r#"{"action":"move_down"}"#, + ipc-send r#"{"action":"get_state","id":"1"}"#, + ipc-expect r#""cursor":1"#, + ] + } + + test move_to_bottom_then_state { + actions: [ + ipc-send r#"{"action":"move_to_bottom"}"#, + ipc-send r#"{"action":"get_state","id":"1"}"#, + ipc-expect r#""cursor":2"#, + ] + } + } + + ipc mod ipc_confirm { + items: ["alpha", "beta", "gamma"]; + + test confirm_selects_cursor_item { + actions: [ + ipc-send r#"{"action":"confirm"}"#, + ] + selected: "alpha" + } + + test move_then_confirm { + actions: [ + ipc-send r#"{"action":"move_down"}"#, + ipc-send r#"{"action":"confirm"}"#, + ] + selected: "beta" + } + } + + ipc mod ipc_cancel { + items: ["alpha", "beta"]; + + test cancel_via_ipc { + actions: [ + ipc-send r#"{"action":"cancel"}"#, + ] + cancelled + } + } + + ipc mod ipc_selection { + items: ["alpha", "beta", "gamma"]; + + test get_selection_initially_empty { + actions: [ + ipc-send r#"{"action":"get_selection","id":"1"}"#, + ipc-expect r#""indices":[]"#, + ] + } + } + + ipc mod ipc_errors { + items: ["alpha"]; + + test bad_json_returns_error { + actions: [ + ipc-send r#"not valid json"#, + ipc-expect r#""error":"#, + ] + } + } +} diff --git a/crates/pikl/src/ipc/mod.rs b/crates/pikl/src/ipc/mod.rs new file mode 100644 index 0000000..c9672c6 --- /dev/null +++ b/crates/pikl/src/ipc/mod.rs @@ -0,0 +1,55 @@ +//! IPC module. Provides external control of pikl via Unix socket. +//! Newline-delimited JSON protocol. Off by default, enabled with `--ipc`. + +pub mod protocol; +pub mod server; + +#[cfg(test)] +mod integration; + +use std::path::PathBuf; + +use pikl_core::error::PiklError; + +/// Compute the socket path for a pikl IPC session. +/// +/// Named session: `/run/user/$UID/pikl-{session}.sock` +/// No session: `/run/user/$UID/pikl-{pid}.sock` +pub fn socket_path(session: Option<&str>) -> Result { + let uid = unsafe { libc::getuid() }; + let run_dir = PathBuf::from(format!("/run/user/{uid}")); + if !run_dir.exists() { + return Err(PiklError::Io(std::io::Error::new( + std::io::ErrorKind::NotFound, + format!("/run/user/{uid} does not exist"), + ))); + } + let name = match session { + Some(s) => format!("pikl-{s}.sock"), + None => format!("pikl-{}.sock", std::process::id()), + }; + Ok(run_dir.join(name)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn socket_path_with_session() { + let path = socket_path(Some("test")); + // Should succeed on Linux systems with /run/user/$UID + if let Ok(p) = path { + assert!(p.to_string_lossy().contains("pikl-test.sock")); + } + } + + #[test] + fn socket_path_without_session() { + let path = socket_path(None); + if let Ok(p) = path { + let pid = std::process::id(); + assert!(p.to_string_lossy().contains(&format!("pikl-{pid}.sock"))); + } + } +} diff --git a/crates/pikl/src/ipc/protocol.rs b/crates/pikl/src/ipc/protocol.rs new file mode 100644 index 0000000..2f3653b --- /dev/null +++ b/crates/pikl/src/ipc/protocol.rs @@ -0,0 +1,212 @@ +//! IPC protocol types. Newline-delimited JSON over Unix socket. +//! Write commands are fire-and-forget, read commands use an `id` +//! field for request/response correlation. + +use pikl_core::event::{Action, Mode, ViewState}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// Command sent from an IPC client to pikl. +#[derive(Debug, Deserialize)] +#[serde(tag = "action", rename_all = "snake_case")] +pub enum IpcCommand { + // -- Write commands (no response) -- + AddItems { items: Vec }, + ReplaceItems { items: Vec }, + RemoveItems { indices: Vec }, + SetFilter { text: String }, + MoveUp, + MoveDown, + MoveToTop, + MoveToBottom, + PageUp, + PageDown, + ToggleSelect, + SelectAll, + ClearSelections, + Confirm, + Cancel, + Close, + + // -- Read commands (require id) -- + GetState { id: String }, + GetSelection { id: String }, + + // -- Subscription -- + Subscribe { events: Vec }, + Unsubscribe, +} + +/// Response sent from pikl to an IPC client. +#[derive(Debug, Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum IpcResponse { + State { + id: String, + filter_text: String, + cursor: usize, + total_items: usize, + total_filtered: usize, + mode: String, + selection_count: usize, + }, + Selection { + id: String, + indices: Vec, + }, + Event { + event: String, + #[serde(skip_serializing_if = "Option::is_none")] + data: Option, + }, + Error { + #[serde(skip_serializing_if = "Option::is_none")] + id: Option, + error: String, + }, +} + +/// Convert a write-type IpcCommand into a core Action. +/// Returns None for read/subscribe commands (handled by the server). +pub fn ipc_command_to_action(cmd: &IpcCommand) -> Option { + match cmd { + IpcCommand::AddItems { items } => Some(Action::AddItems(items.clone())), + IpcCommand::ReplaceItems { items } => Some(Action::ReplaceItems(items.clone())), + IpcCommand::RemoveItems { indices } => Some(Action::RemoveItems(indices.clone())), + IpcCommand::SetFilter { text } => Some(Action::UpdateFilter(text.clone())), + IpcCommand::MoveUp => Some(Action::MoveUp(1)), + IpcCommand::MoveDown => Some(Action::MoveDown(1)), + IpcCommand::MoveToTop => Some(Action::MoveToTop), + IpcCommand::MoveToBottom => Some(Action::MoveToBottom), + IpcCommand::PageUp => Some(Action::PageUp(1)), + IpcCommand::PageDown => Some(Action::PageDown(1)), + IpcCommand::ToggleSelect => Some(Action::ToggleSelect), + IpcCommand::SelectAll => Some(Action::SelectAll), + IpcCommand::ClearSelections => Some(Action::ClearSelections), + IpcCommand::Confirm => Some(Action::Confirm), + IpcCommand::Cancel => Some(Action::Cancel), + IpcCommand::Close => Some(Action::CloseMenu), + IpcCommand::GetState { .. } + | IpcCommand::GetSelection { .. } + | IpcCommand::Subscribe { .. } + | IpcCommand::Unsubscribe => None, + } +} + +/// Build an IpcResponse::State from a cached ViewState. +pub fn view_state_to_response(id: String, vs: &ViewState) -> IpcResponse { + let mode = match vs.mode { + Mode::Insert => "insert", + Mode::Normal => "normal", + Mode::Visual => "visual", + }; + IpcResponse::State { + id, + filter_text: vs.filter_text.to_string(), + cursor: vs.cursor, + total_items: vs.total_items, + total_filtered: vs.total_filtered, + mode: mode.to_string(), + selection_count: vs.selection_count, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn deserialize_write_commands() -> Result<(), Box> { + let cases = vec![ + (r#"{"action":"move_up"}"#, "MoveUp"), + (r#"{"action":"move_down"}"#, "MoveDown"), + (r#"{"action":"move_to_top"}"#, "MoveToTop"), + (r#"{"action":"move_to_bottom"}"#, "MoveToBottom"), + (r#"{"action":"page_up"}"#, "PageUp"), + (r#"{"action":"page_down"}"#, "PageDown"), + (r#"{"action":"toggle_select"}"#, "ToggleSelect"), + (r#"{"action":"select_all"}"#, "SelectAll"), + (r#"{"action":"clear_selections"}"#, "ClearSelections"), + (r#"{"action":"confirm"}"#, "Confirm"), + (r#"{"action":"cancel"}"#, "Cancel"), + (r#"{"action":"close"}"#, "Close"), + ]; + + for (json, name) in cases { + let cmd: IpcCommand = serde_json::from_str(json)?; + assert!( + ipc_command_to_action(&cmd).is_some(), + "{name} should produce an action" + ); + } + Ok(()) + } + + #[test] + fn deserialize_set_filter() -> Result<(), Box> { + let json = r#"{"action":"set_filter","text":"hello"}"#; + let cmd: IpcCommand = serde_json::from_str(json)?; + let action = ipc_command_to_action(&cmd); + assert!(matches!(action, Some(Action::UpdateFilter(ref t)) if t == "hello")); + Ok(()) + } + + #[test] + fn deserialize_add_items() -> Result<(), Box> { + let json = r#"{"action":"add_items","items":["foo","bar"]}"#; + let cmd: IpcCommand = serde_json::from_str(json)?; + let action = ipc_command_to_action(&cmd); + assert!(matches!(action, Some(Action::AddItems(ref items)) if items.len() == 2)); + Ok(()) + } + + #[test] + fn deserialize_remove_items() -> Result<(), Box> { + let json = r#"{"action":"remove_items","indices":[0,2]}"#; + let cmd: IpcCommand = serde_json::from_str(json)?; + let action = ipc_command_to_action(&cmd); + assert!(matches!(action, Some(Action::RemoveItems(ref idx)) if idx == &[0, 2])); + Ok(()) + } + + #[test] + fn deserialize_replace_items() -> Result<(), Box> { + let json = r#"{"action":"replace_items","items":["a","b","c"]}"#; + let cmd: IpcCommand = serde_json::from_str(json)?; + let action = ipc_command_to_action(&cmd); + assert!(matches!(action, Some(Action::ReplaceItems(ref items)) if items.len() == 3)); + Ok(()) + } + + #[test] + fn deserialize_get_state() -> Result<(), Box> { + let json = r#"{"action":"get_state","id":"req1"}"#; + let cmd: IpcCommand = serde_json::from_str(json)?; + assert!(ipc_command_to_action(&cmd).is_none()); + Ok(()) + } + + #[test] + fn deserialize_get_selection() -> Result<(), Box> { + let json = r#"{"action":"get_selection","id":"req2"}"#; + let cmd: IpcCommand = serde_json::from_str(json)?; + assert!(ipc_command_to_action(&cmd).is_none()); + Ok(()) + } + + #[test] + fn deserialize_subscribe() -> Result<(), Box> { + let json = r#"{"action":"subscribe","events":["hover","filter"]}"#; + let cmd: IpcCommand = serde_json::from_str(json)?; + assert!(ipc_command_to_action(&cmd).is_none()); + Ok(()) + } + + #[test] + fn deserialize_unsubscribe() -> Result<(), Box> { + let json = r#"{"action":"unsubscribe"}"#; + let cmd: IpcCommand = serde_json::from_str(json)?; + assert!(ipc_command_to_action(&cmd).is_none()); + Ok(()) + } +} diff --git a/crates/pikl/src/ipc/server.rs b/crates/pikl/src/ipc/server.rs new file mode 100644 index 0000000..2dfe266 --- /dev/null +++ b/crates/pikl/src/ipc/server.rs @@ -0,0 +1,331 @@ +//! IPC server. Accepts Unix socket connections and translates +//! JSON commands into Actions. Each client gets its own task. + +use std::path::PathBuf; +use std::sync::Arc; + +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::UnixListener; +use tokio::sync::{broadcast, mpsc, watch, RwLock}; +use tracing::{info, warn}; + +use pikl_core::event::{Action, MenuEvent, ViewState}; + +use super::protocol::{ipc_command_to_action, view_state_to_response, IpcCommand, IpcResponse}; + +/// IPC server that listens on a Unix socket and bridges +/// external commands into the menu's action channel. +/// Dropping the server cancels all background tasks and +/// removes the socket file. +pub struct IpcServer { + socket_path: PathBuf, + action_tx: mpsc::Sender, + state: Arc>>, + event_tx: broadcast::Sender, + /// Dropping the sender signals all tasks to shut down. + _shutdown_tx: watch::Sender, +} + +impl IpcServer { + pub fn new( + socket_path: PathBuf, + action_tx: mpsc::Sender, + event_tx: broadcast::Sender, + ) -> Self { + Self { + socket_path, + action_tx, + state: Arc::new(RwLock::new(None)), + event_tx, + _shutdown_tx: watch::channel(false).0, + } + } + + /// Start listening. Spawns background tasks for state caching + /// and connection handling. Returns immediately. + pub fn start(&self, event_rx: broadcast::Receiver) -> Result<(), std::io::Error> { + // Remove stale socket from a previous crash + if self.socket_path.exists() { + std::fs::remove_file(&self.socket_path)?; + } + + let listener = { + let std_listener = std::os::unix::net::UnixListener::bind(&self.socket_path)?; + std_listener.set_nonblocking(true)?; + UnixListener::from_std(std_listener)? + }; + + // Set socket permissions to 0o700 (owner only) + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let perms = std::fs::Permissions::from_mode(0o700); + std::fs::set_permissions(&self.socket_path, perms)?; + } + + info!(path = %self.socket_path.display(), "IPC server listening"); + + // State cache task: keep a copy of the latest ViewState + let state = Arc::clone(&self.state); + let mut cache_rx = event_rx; + let mut shutdown_rx = self._shutdown_tx.subscribe(); + tokio::spawn(async move { + loop { + tokio::select! { + event = cache_rx.recv() => { + match event { + Ok(MenuEvent::StateChanged(vs)) => { + *state.write().await = Some(vs); + } + Ok( + MenuEvent::Selected(_) + | MenuEvent::Quicklist(_) + | MenuEvent::Cancelled, + ) => break, + Err(broadcast::error::RecvError::Lagged(n)) => { + warn!(skipped = n, "IPC state cache fell behind"); + } + Err(broadcast::error::RecvError::Closed) => break, + } + } + _ = shutdown_rx.changed() => break, + } + } + }); + + // Accept loop + let action_tx = self.action_tx.clone(); + let state = Arc::clone(&self.state); + let event_tx = self.event_tx.clone(); + let mut shutdown_rx = self._shutdown_tx.subscribe(); + tokio::spawn(async move { + loop { + tokio::select! { + result = listener.accept() => { + match result { + Ok((stream, _addr)) => { + let action_tx = action_tx.clone(); + let state = Arc::clone(&state); + let event_tx = event_tx.clone(); + tokio::spawn(async move { + if let Err(e) = + handle_connection(stream, action_tx, state, event_tx).await + { + warn!(%e, "IPC connection error"); + } + }); + } + Err(e) => { + if e.kind() == std::io::ErrorKind::InvalidInput { + break; + } + warn!(%e, "IPC accept error"); + } + } + } + _ = shutdown_rx.changed() => break, + } + } + }); + + Ok(()) + } + +} + +impl Drop for IpcServer { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.socket_path); + } +} + +/// Handle a single client connection. Reads JSON lines, +/// dispatches actions, sends responses. +async fn handle_connection( + stream: tokio::net::UnixStream, + action_tx: mpsc::Sender, + state: Arc>>, + event_tx: broadcast::Sender, +) -> Result<(), std::io::Error> { + let (read_half, mut write_half) = stream.into_split(); + let mut reader = BufReader::new(read_half); + let mut line = String::new(); + + loop { + line.clear(); + let n = reader.read_line(&mut line).await?; + if n == 0 { + break; // client disconnected + } + + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + + let cmd: IpcCommand = match serde_json::from_str(trimmed) { + Ok(c) => c, + Err(e) => { + warn!(input = trimmed, %e, "bad IPC command"); + let resp = IpcResponse::Error { + id: None, + error: format!("invalid JSON: {e}"), + }; + write_response(&mut write_half, &resp).await?; + continue; + } + }; + + match cmd { + // Write commands: convert to action, send, no response + ref c if ipc_command_to_action(c).is_some() => { + if let Some(action) = ipc_command_to_action(&cmd) { + let _ = action_tx.send(action).await; + } + } + + // Read: get_state + IpcCommand::GetState { ref id } => { + let guard = state.read().await; + let resp = match &*guard { + Some(vs) => view_state_to_response(id.clone(), vs), + None => IpcResponse::Error { + id: Some(id.clone()), + error: "no state available yet".to_string(), + }, + }; + write_response(&mut write_half, &resp).await?; + } + + // Read: get_selection + IpcCommand::GetSelection { ref id } => { + let guard = state.read().await; + let resp = match &*guard { + Some(vs) => { + let indices: Vec = vs + .visible_items + .iter() + .filter(|vi| vi.selected) + .map(|vi| vi.index) + .collect(); + IpcResponse::Selection { + id: id.clone(), + indices, + } + } + None => IpcResponse::Error { + id: Some(id.clone()), + error: "no state available yet".to_string(), + }, + }; + write_response(&mut write_half, &resp).await?; + } + + // Subscribe: forward matching events as JSON lines + IpcCommand::Subscribe { ref events } => { + let event_filter: Vec = + events.iter().map(|s| s.to_lowercase()).collect(); + let mut sub_rx = event_tx.subscribe(); + // Spawn a forwarding task. It runs until the client + // disconnects (write fails) or the menu closes. + let mut write_half_clone = write_half; + tokio::spawn(async move { + loop { + match sub_rx.recv().await { + Ok(MenuEvent::StateChanged(vs)) => { + if event_filter.contains(&"state".to_string()) { + let resp = IpcResponse::Event { + event: "state".to_string(), + data: Some(serde_json::json!({ + "filter_text": vs.filter_text.to_string(), + "cursor": vs.cursor, + "total_items": vs.total_items, + "total_filtered": vs.total_filtered, + "selection_count": vs.selection_count, + })), + }; + if write_response(&mut write_half_clone, &resp) + .await + .is_err() + { + break; + } + } + } + Ok(MenuEvent::Selected(items)) => { + if event_filter.contains(&"select".to_string()) { + let resp = IpcResponse::Event { + event: "select".to_string(), + data: Some(serde_json::json!({ + "items": items.iter() + .map(|(v, i)| serde_json::json!({"value": v, "index": i})) + .collect::>() + })), + }; + if write_response(&mut write_half_clone, &resp) + .await + .is_err() + { + break; + } + } + break; + } + Ok(MenuEvent::Quicklist(items)) => { + if event_filter.contains(&"quicklist".to_string()) { + let resp = IpcResponse::Event { + event: "quicklist".to_string(), + data: Some(serde_json::json!({"items": items})), + }; + if write_response(&mut write_half_clone, &resp) + .await + .is_err() + { + break; + } + } + break; + } + Ok(MenuEvent::Cancelled) => { + if event_filter.contains(&"cancel".to_string()) { + let resp = IpcResponse::Event { + event: "cancel".to_string(), + data: None, + }; + let _ = + write_response(&mut write_half_clone, &resp).await; + } + break; + } + Err(broadcast::error::RecvError::Lagged(_)) => continue, + Err(broadcast::error::RecvError::Closed) => break, + } + } + }); + // Subscribe takes over the write half, so we + // can't continue reading. The connection is now + // subscription-only until it closes. + return Ok(()); + } + + IpcCommand::Unsubscribe => { + // Nothing to do if not subscribed + } + + _ => {} + } + } + + Ok(()) +} + +/// Write a JSON response as a single line. +async fn write_response( + writer: &mut tokio::net::unix::OwnedWriteHalf, + resp: &IpcResponse, +) -> Result<(), std::io::Error> { + let json = serde_json::to_string(resp).map_err(|e| std::io::Error::other(e.to_string()))?; + writer.write_all(json.as_bytes()).await?; + writer.write_all(b"\n").await?; + writer.flush().await +} diff --git a/crates/pikl/src/main.rs b/crates/pikl/src/main.rs index 9f77731..3013988 100644 --- a/crates/pikl/src/main.rs +++ b/crates/pikl/src/main.rs @@ -1,5 +1,7 @@ mod handler; mod hook; +mod ipc; +mod session; use std::io::{BufReader, IsTerminal, Write}; use std::sync::Arc; @@ -137,6 +139,14 @@ struct Cli { /// Columns to display in table mode (e.g. "label,meta.age:Age") #[arg(long, value_name = "COLS")] columns: Option, + + /// Enable IPC control via Unix socket + #[arg(long)] + ipc: bool, + + /// Session name (enables filter history, names the IPC socket) + #[arg(long)] + session: Option, } fn main() { @@ -262,13 +272,41 @@ fn main() { } }; + // Load session filter history if --session is set + let mut history = cli.session.as_ref().and_then(|name| { + match session::FilterHistory::load(name) { + Ok(h) => Some(h), + Err(e) => { + tracing::warn!(%e, "failed to load session history"); + None + } + } + }); + // STEP 4: Branch on headless vs interactive let result = if let Some(script) = script { rt.block_on(run_headless(items, &cli, script, start_mode, column_config)) } else { - rt.block_on(run_interactive(items, &cli, start_mode, column_config)) + let history_entries = history.as_ref().map(|h| h.entries().to_vec()); + rt.block_on(run_interactive( + items, + &cli, + start_mode, + column_config, + history_entries, + )) }; + // Save filter text to session history on confirm + if let Some(ref mut h) = history + && let Ok(MenuResult::Selected { ref filter_text, .. }) + | Ok(MenuResult::Quicklist { ref filter_text, .. }) = result + { + let _ = h.append(filter_text).inspect_err(|e| { + tracing::warn!(%e, "failed to save session history"); + }); + } + // STEP 5: Handle result handle_result(result, &cli); } @@ -398,7 +436,7 @@ async fn run_headless( match show_action { ShowAction::Ui | ShowAction::Tui | ShowAction::Gui => { - let tui_handle = tokio::spawn(pikl_tui::run(action_tx, event_rx)); + let tui_handle = tokio::spawn(pikl_tui::run(action_tx, event_rx, None)); let result = menu_handle .await .map_err(|e| PiklError::Io(std::io::Error::other(e.to_string())))??; @@ -422,6 +460,7 @@ async fn run_interactive( cli: &Cli, start_mode: Mode, column_config: Option, + filter_history: Option>, ) -> Result { let (mut menu, action_tx) = MenuRunner::new(build_menu(items, cli, column_config)); menu.set_initial_mode(start_mode); @@ -434,16 +473,44 @@ async fn run_interactive( let event_rx = menu.subscribe(); + // Start IPC server if requested + let _ipc_server = if cli.ipc { + let sock_path = ipc::socket_path(cli.session.as_deref())?; + let server = ipc::server::IpcServer::new( + sock_path, + action_tx.clone(), + menu.event_sender(), + ); + let ipc_event_rx = menu.subscribe(); + server + .start(ipc_event_rx) + .map_err(PiklError::Io)?; + Some(server) + } else { + None + }; + // Handle SIGINT/SIGTERM: restore terminal and exit cleanly. let signal_tx = action_tx.clone(); tokio::spawn(async move { - if let Ok(()) = tokio::signal::ctrl_c().await { - pikl_tui::restore_terminal(); - let _ = signal_tx.send(Action::Cancel).await; + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) { + Ok(mut sigterm) => { + tokio::select! { + _ = tokio::signal::ctrl_c() => {} + _ = sigterm.recv() => {} + } + } + Err(e) => { + tracing::warn!(%e, "failed to register SIGTERM handler, falling back to SIGINT only"); + let _ = tokio::signal::ctrl_c().await; + } } + pikl_tui::restore_terminal(); + let _ = signal_tx.send(Action::Cancel).await; }); - let tui_handle = tokio::spawn(async move { pikl_tui::run(action_tx, event_rx).await }); + let tui_handle = + tokio::spawn(async move { pikl_tui::run(action_tx, event_rx, filter_history).await }); let result = menu.run().await; @@ -456,7 +523,7 @@ async fn run_interactive( fn handle_result(result: Result, cli: &Cli) { let mut out = std::io::stdout().lock(); match result { - Ok(MenuResult::Selected { items }) => { + Ok(MenuResult::Selected { items, .. }) => { if cli.structured { for (value, index) in items { let output = OutputItem { @@ -472,7 +539,7 @@ fn handle_result(result: Result, cli: &Cli) { } } } - Ok(MenuResult::Quicklist { items }) => { + Ok(MenuResult::Quicklist { items, .. }) => { if cli.structured { for (value, index) in items { let output = OutputItem { diff --git a/crates/pikl/src/session.rs b/crates/pikl/src/session.rs new file mode 100644 index 0000000..df92ed0 --- /dev/null +++ b/crates/pikl/src/session.rs @@ -0,0 +1,210 @@ +//! Session filter history. Stores and recalls previous filter +//! strings per named session. History is append-only to a file +//! in `~/.local/state/pikl/sessions/`. + +use std::fs; +use std::io::Write; +use std::path::PathBuf; + +use pikl_core::error::PiklError; + +/// Filter history for a named session. Each entry is a filter +/// string the user confirmed with. Ctrl+P/Ctrl+N cycle through +/// these in the TUI. +pub struct FilterHistory { + entries: Vec, + file_path: PathBuf, +} + +impl FilterHistory { + /// Load history for the given session name. Creates an empty + /// history if the file doesn't exist yet. + pub fn load(session_name: &str) -> Result { + let file_path = history_path(session_name)?; + + let entries = if file_path.exists() { + let content = fs::read_to_string(&file_path)?; + content + .lines() + .filter(|l| !l.is_empty()) + .map(String::from) + .collect() + } else { + Vec::new() + }; + + Ok(Self { entries, file_path }) + } + + /// Append a filter string to the history. Skips empty strings + /// and duplicates of the last entry. + pub fn append(&mut self, filter: &str) -> Result<(), PiklError> { + if filter.is_empty() { + return Ok(()); + } + if self.entries.last().is_some_and(|last| last == filter) { + return Ok(()); + } + + // Create parent dirs if needed + if let Some(parent) = self.file_path.parent() { + fs::create_dir_all(parent)?; + } + + let mut file = fs::OpenOptions::new() + .create(true) + .append(true) + .open(&self.file_path)?; + writeln!(file, "{filter}")?; + + self.entries.push(filter.to_string()); + Ok(()) + } + + /// Read-only access to history entries. + pub fn entries(&self) -> &[String] { + &self.entries + } +} + +#[cfg(test)] +impl FilterHistory { + /// Test-only constructor that uses an explicit file path + /// instead of deriving one from HOME. + fn with_path(file_path: PathBuf) -> Self { + Self { + entries: Vec::new(), + file_path, + } + } +} + +/// Build the history file path for a session name. +/// Uses `$HOME/.local/state/pikl/sessions/{name}.history`. +fn history_path(session_name: &str) -> Result { + let home = std::env::var("HOME").map_err(|_| { + PiklError::Io(std::io::Error::new( + std::io::ErrorKind::NotFound, + "HOME environment variable not set", + )) + })?; + Ok(PathBuf::from(home) + .join(".local/state/pikl/sessions") + .join(format!("{session_name}.history"))) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + + /// Create a temp directory and return a history file path inside it. + fn temp_history_path(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("pikl-test-session-{name}")); + let _ = fs::create_dir_all(&dir); + dir.join("test.history") + } + + /// Clean up the temp directory after a test. + fn cleanup(path: &PathBuf) { + if let Some(parent) = path.parent() { + let _ = fs::remove_dir_all(parent); + } + } + + #[test] + fn load_missing_file_returns_empty() { + let path = temp_history_path("missing"); + // Make sure it doesn't exist + let _ = fs::remove_file(&path); + + let hist = FilterHistory::with_path(path.clone()); + assert!(hist.entries().is_empty()); + cleanup(&path); + } + + #[test] + fn append_and_load_roundtrip() -> Result<(), PiklError> { + let path = temp_history_path("roundtrip"); + let _ = fs::remove_file(&path); + + let mut hist = FilterHistory::with_path(path.clone()); + hist.append("hello")?; + hist.append("world")?; + + // Re-read from disk by constructing a new history that + // loads the file content. + let content = fs::read_to_string(&path)?; + let loaded: Vec = content + .lines() + .filter(|l| !l.is_empty()) + .map(String::from) + .collect(); + + assert_eq!(loaded, vec!["hello", "world"]); + cleanup(&path); + Ok(()) + } + + #[test] + fn append_skips_empty_string() -> Result<(), PiklError> { + let path = temp_history_path("skip-empty"); + let _ = fs::remove_file(&path); + + let mut hist = FilterHistory::with_path(path.clone()); + hist.append("")?; + + assert!(hist.entries().is_empty()); + assert!(!path.exists()); + cleanup(&path); + Ok(()) + } + + #[test] + fn append_deduplicates_consecutive() -> Result<(), PiklError> { + let path = temp_history_path("dedup-consecutive"); + let _ = fs::remove_file(&path); + + let mut hist = FilterHistory::with_path(path.clone()); + hist.append("foo")?; + hist.append("foo")?; + + assert_eq!(hist.entries(), &["foo"]); + cleanup(&path); + Ok(()) + } + + #[test] + fn append_allows_nonconsecutive_duplicates() -> Result<(), PiklError> { + let path = temp_history_path("nonconsecutive"); + let _ = fs::remove_file(&path); + + let mut hist = FilterHistory::with_path(path.clone()); + hist.append("foo")?; + hist.append("bar")?; + hist.append("foo")?; + + assert_eq!(hist.entries(), &["foo", "bar", "foo"]); + cleanup(&path); + Ok(()) + } + + #[test] + fn entries_returns_all() -> Result<(), PiklError> { + let path = temp_history_path("entries-all"); + let _ = fs::remove_file(&path); + + let mut hist = FilterHistory::with_path(path.clone()); + hist.append("alpha")?; + hist.append("beta")?; + hist.append("gamma")?; + + let entries = hist.entries(); + assert_eq!(entries.len(), 3); + assert_eq!(entries[0], "alpha"); + assert_eq!(entries[1], "beta"); + assert_eq!(entries[2], "gamma"); + cleanup(&path); + Ok(()) + } +} diff --git a/docs/DESIGN.md b/docs/DESIGN.md index e18adee..303b805 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -519,13 +519,19 @@ A few built-in themes ship with pikl. `--theme monokai` or ## Sessions -`--session name` persists the menu's state (filter text, -scroll position, selections) across -invocations. State lives in -`~/.local/state/pikl/sessions/`. - +`--session name` enables filter history across invocations. No `--session` = ephemeral, nothing saved. +When a session is active, every filter that leads to a +selection (single, multi, or quicklist) is appended to +`~/.local/state/pikl/sessions/{name}.history`. Empty +filters and consecutive duplicates are skipped. + +On startup with a session name, the history file is loaded. +Ctrl+P / Ctrl+N in insert mode cycles through previous +filters. Selecting a history entry replaces the current +filter text. + Session names are just strings, use shell expansion for dynamic names: @@ -534,8 +540,8 @@ pikl --session "walls-$(hostname)" pikl --session "logs-$USER" ``` -Session history is a log file alongside the state. Other -tools can tail it for observability. +Sessions are lightweight by design. No item persistence, +no scroll position, no selections. Just filter recall. ## Watched Sources @@ -552,18 +558,96 @@ named pipe. ## IPC -While running, pikl-menu listens on a Unix socket -(`/run/user/$UID/pikl-$PID.sock`). External tools can: +Opt-in external control via Unix socket. Enabled with +`--ipc`. Off by default: pikl is ephemeral and shouldn't +have side effects unless asked. -- Push new items -- Remove items -- Update item fields -- Set the filter text -- Read current selection -- Close the menu +Socket path: +- Named session: `/run/user/$UID/pikl-{session}.sock` +- No session: `/run/user/$UID/pikl-{pid}.sock` -Protocol is newline-delimited JSON. Simple enough to use -with `socat` or any language's Unix socket support. +The socket path is logged via tracing on startup and +cleaned up on exit. + +### Protocol + +Newline-delimited JSON, one message per line. Simple +enough for `socat` or any language with Unix sockets. + +### Client-to-pikl Commands + +**Write commands** (fire-and-forget, no response): + +| Action | Payload | Effect | +|---|---|---| +| `add_items` | `{"items": [...]}` | Append items | +| `replace_items` | `{"items": [...]}` | Replace all items | +| `remove_items` | `{"indices": [...]}` | Remove by index | +| `set_filter` | `{"text": "..."}` | Set filter text | +| `move_up` | (none) | Navigate up | +| `move_down` | (none) | Navigate down | +| `move_to_top` | (none) | Jump to top | +| `move_to_bottom` | (none) | Jump to bottom | +| `page_up` | (none) | Page up | +| `page_down` | (none) | Page down | +| `toggle_select` | (none) | Toggle current item | +| `select_all` | (none) | Select all items | +| `clear_selections` | (none) | Clear selections | +| `confirm` | (none) | Confirm selection | +| `cancel` | (none) | Cancel menu | +| `close` | (none) | Close menu | + +**Read commands** (require `id` field, response echoes it): + +| Action | Response | +|---|---| +| `get_state` | `{"id": "...", "state": {"filter": "...", "cursor": 3, "total_items": 150, "total_filtered": 12, "selection_count": 0, "mode": "insert"}}` | +| `get_selection` | `{"id": "...", "selection": [{"label": "...", "index": 3}, ...]}` | + +**Event subscription:** + +| Action | Payload | Effect | +|---|---|---| +| `subscribe` | `{"events": ["hover", "filter", ...]}` | Start receiving events | +| `unsubscribe` | (none) | Stop receiving events | + +Subscribed events are pushed as +`{"event": "hover", "item": {...}, "index": 3}` lines. + +### Multiple Connections + +Multiple clients can connect simultaneously. Each gets +independent subscription state. All writes are serialized +through the same action channel. + +### Auth + +Unix socket permissions (user-only access). No additional +authentication. + +### Architecture + +IPC is just another frontend. It deserializes JSON +messages into `Action` variants and sends them through the +same `mpsc` channel as the TUI or action-fd. Event +subscriptions tap into the `broadcast` channel. No special +core changes needed. + +### Examples + +```sh +# Push items into a running pikl +echo '{"action": "add_items", "items": [{"label": "new"}]}' \ + | socat - UNIX-CONNECT:/run/user/1000/pikl-mypicker.sock + +# Read current state +echo '{"action": "get_state", "id": "1"}' \ + | socat - UNIX-CONNECT:/run/user/1000/pikl-mypicker.sock + +# Subscribe to hover events +echo '{"action": "subscribe", "events": ["hover"]}' \ + | socat - UNIX-CONNECT:/run/user/1000/pikl-mypicker.sock +``` ## Exit Codes @@ -644,8 +728,8 @@ sequentially regardless of where they came from. | TUI | Terminal keypresses (crossterm) | Yes | 1 | | GUI | GUI events (iced) | Yes | 8 | | Action-fd | Pre-validated script from a file descriptor | No (unless `show-ui`) | 1.5 | -| IPC | Unix socket, JSON protocol | Yes (bidirectional) | 6 | -| Lua | LuaJIT script via mlua | Yes (stateful, conditional) | Post-6 | +| IPC | Unix socket, newline-delimited JSON | Yes (bidirectional) | 6 | +| Lua | Embedded LuaJIT via mlua (in-process) | Yes (stateful, conditional) | Post-6 | This framing means new interaction modes don't require core changes: they're just new frontends that push actions. @@ -821,12 +905,15 @@ complexity: extend pikl without touching Rust. 4. **IPC** (phase 6): bidirectional JSON over Unix socket. External tools can read state and send actions while pikl - runs interactively. Good for tool integration. -5. **Lua** (post phase 6): embedded LuaJIT via mlua. Full - stateful scripting: subscribe to events, branch on state, - loops, the works. The Lua runtime is just another - frontend pushing Actions and subscribing to MenuEvents. - For anything complex enough to need a real language. + runs interactively. Language-agnostic: anything that can + write JSON to a socket works. +5. **Lua** (post phase 6): embedded LuaJIT via mlua. + In-process scripting for complex hooks and behaviour. + Direct access to the action/event system, no socket + overhead. Think Neovim's Lua layer: you're extending the + tool, not talking to it from outside. IPC and Lua are + separate integration points. IPC is for external + processes. Lua is for embedded logic. No custom DSL. Action-fd stays simple forever. The jump from "I need conditionals" to "use Lua" is intentional: @@ -848,16 +935,10 @@ with the full writeup. ## Open Questions -- Should marks/registers persist across sessions or only - within a session? - Accessibility: screen reader support for TUI mode? - Should `--watch` support inotify on Linux and FSEvents on macOS, or use `notify` crate to abstract? - Maximum practical item count before we need virtual scrolling? (Probably around 100k) -- Should hooks run in a pool or strictly sequential? - Resolved: exec hooks are one subprocess per event. - Handler hooks are persistent processes. Debounce and - cancel-stale manage concurrency. - Plugin system via WASM for custom filter strategies? (Probably way later if ever) diff --git a/docs/DEVPLAN.md b/docs/DEVPLAN.md index 15ae1a6..913a763 100644 --- a/docs/DEVPLAN.md +++ b/docs/DEVPLAN.md @@ -212,22 +212,68 @@ CSV/TSV input parsing and columnar table rendering. **Done when:** `echo "name,age\nalice,30\nbob,25" | pikl --input-format csv` renders a navigable table with headers. -## Phase 6: Sessions & IPC +## Phase 6: IPC & Session History -Persistence and external control. +Two independent features bundled together. IPC adds live +external control over a Unix socket. Session history gives +lightweight filter recall for repeated workflows. + +### IPC (External Control) + +Opt-in Unix socket server for live control of a running +pikl instance by external processes. **Deliverables:** -- `--session name` for state persistence -- Session state: filter, scroll position, selections -- Session history log file -- Unix socket IPC while running -- IPC commands: push/remove/update items, set filter, - read selection, close -- Protocol: newline-delimited JSON +- `--ipc` flag to enable the socket listener (off by + default, pikl is ephemeral and shouldn't have side + effects unless asked) +- Socket path: `/run/user/$UID/pikl-{session}.sock` + (named session) or `/run/user/$UID/pikl-{pid}.sock` + (no session name) +- Socket path logged via tracing on startup +- Cleanup on exit (normal, cancel, SIGTERM) +- Protocol: newline-delimited JSON, one message per line +- Write commands: `add_items`, `replace_items`, + `remove_items`, `set_filter`, `confirm`, `cancel`, + `close`, plus navigation actions (`move_up`, + `move_down`, `move_to_top`, `move_to_bottom`, + `page_up`, `page_down`, `toggle_select`, `select_all`, + `clear_selections`) +- Read commands: `get_state` (filter, cursor, counts, + mode), `get_selection` (selected items). Both require + an `id` field, response echoes it back. +- Event subscription: `subscribe` with event type list, + `unsubscribe` to stop. Subscribed events pushed as + `{"event": "...", ...}` lines. +- Multiple simultaneous client connections +- Auth: Unix socket permissions (user-only) +- IPC is just another frontend: deserializes JSON into + Actions, optionally subscribes to MenuEvent broadcast -**Done when:** You can close and reopen a session and find -your state intact. External scripts can push items into a -running pikl instance. +**Not in scope:** +- `get_items` (full item list read, potentially large) +- Auto-enable with `--session` +- Auth beyond Unix socket permissions + +### Session Filter History + +**Deliverables:** +- `--session name` flag +- On any confirm (single, multi, quicklist), append + current filter text to + `~/.local/state/pikl/sessions/{name}.history` +- Skip empty filters and consecutive duplicates +- Load history on startup if session file exists +- Ctrl+P / Ctrl+N in insert mode to cycle through + filter history +- Selecting a history entry replaces the current filter + text +- Per-session only, no global history + +**Done when:** An external script can connect to a running +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