feat(ovsm): Add osvm-stream function (foundation)

Add osvm-stream built-in function for standalone OVSM scripts.

Goal: Eliminate need for manual server setup
Before: osvm stream --programs pumpfun (separate terminal)
After: (define stream (osvm-stream :alias "pumpfun"))

Implementation (foundation):
- Added eval_osvm_stream to lisp_evaluator
- Added osvm_stream function to streaming module
- Parses :alias, :programs, :tokens, :accounts keywords
- Finds available port (18080-18180 range)
- Spawns background thread for embedded server
- Auto-connects via WebSocket
- Returns stream ID for stream-poll

Current status: PLACEHOLDER
- spawn_internal_server prints debug info
- TODO: Actually embed StreamService in background thread
- TODO: Proper server lifecycle management

Next steps:
1. Refactor StreamService to be embeddable
2. Pass RPC URL configuration
3. Graceful shutdown when script ends
4. Error handling for server failures

Usage (when complete):
```lisp
;; One-line setup - no external server needed!
(define stream (osvm-stream :alias "pumpfun"))

(while (< (- (now) start-time) 60)
  (define events (stream-poll stream :limit 50))
  ...)
```

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: 0xrinegade

crates/ovsm/src/runtime/lisp_evaluator.rs

@@ -369,6 +369,7 @@ impl LispEvaluator {
                     "stream-poll" => self.eval_stream_poll(args),
                     "stream-wait" => self.eval_stream_wait(args),
                     "stream-close" => self.eval_stream_close(args),
+                    "osvm-stream" => self.eval_osvm_stream(args),
                     // LINQ-style functional operations
                     "compact" => self.eval_compact(args),
                     "count-by" => self.eval_count_by(args),
@@ -9107,6 +9108,20 @@ impl LispEvaluator {
         // Call the streaming function with evaluated arguments
         crate::runtime::streaming::stream_close(&evaluated_args)
     }
+
+    /// (osvm-stream &key alias programs tokens) - Spawn internal stream server and connect
+    /// This is a convenience function that combines server spawning + stream-connect
+    /// The server automatically terminates when the script ends
+    fn eval_osvm_stream(&mut self, args: &[crate::parser::Argument]) -> Result<Value> {
+        // Evaluate all arguments to Values
+        let mut evaluated_args = Vec::new();
+        for arg in args {
+            evaluated_args.push(self.evaluate_expression(&arg.value)?);
+        }
+
+        // Call the streaming helper
+        crate::runtime::streaming::osvm_stream(&evaluated_args)
+    }
 }
 
 impl Default for LispEvaluator {

crates/ovsm/src/runtime/streaming.rs

@@ -476,3 +476,156 @@ fn json_to_value(json: &JsonValue) -> Value {
         }
     }
 }
+
+/// Spawn internal OSVM stream server and connect to it
+///
+/// Syntax: `(osvm-stream &key alias programs tokens accounts)`
+///
+/// This is a convenience function that:
+/// 1. Spawns an embedded stream server in a background thread
+/// 2. Automatically connects via WebSocket
+/// 3. Returns stream ID for use with stream-poll
+/// 4. Server auto-terminates when script ends
+///
+/// Parameters:
+/// - `:alias` (optional): Program alias like "pumpfun", "raydium"
+/// - `:programs` (optional): Array of program IDs
+/// - `:tokens` (optional): Array of token symbols/mints
+/// - `:accounts` (optional): Array of account addresses
+///
+/// Returns: Stream ID string
+///
+/// Example:
+/// ```lisp
+/// (define stream (osvm-stream :alias "pumpfun"))
+/// (while true
+///   (define events (stream-poll stream))
+///   ...)
+/// ```
+pub fn osvm_stream(args: &[Value]) -> Result<Value> {
+    // Parse keyword arguments
+    let mut alias: Option<String> = None;
+    let mut programs: Vec<String> = Vec::new();
+    let mut tokens: Vec<String> = Vec::new();
+    let mut accounts: Vec<String> = Vec::new();
+
+    let mut i = 0;
+    while i < args.len() {
+        if let Value::String(key) = &args[i] {
+            if key.starts_with(':') {
+                if i + 1 >= args.len() {
+                    return Err(Error::runtime(format!(
+                        "osvm-stream: missing value for keyword argument {}",
+                        key
+                    )));
+                }
+
+                let value = &args[i + 1];
+                match key.as_str() {
+                    ":alias" => {
+                        if let Value::String(s) = value {
+                            alias = Some(s.clone());
+                        }
+                    }
+                    ":programs" => {
+                        programs = extract_string_array(value)?;
+                    }
+                    ":tokens" => {
+                        tokens = extract_string_array(value)?;
+                    }
+                    ":accounts" => {
+                        accounts = extract_string_array(value)?;
+                    }
+                    _ => {
+                        return Err(Error::runtime(format!(
+                            "osvm-stream: unknown keyword argument {}",
+                            key
+                        )))
+                    }
+                }
+                i += 2;
+            } else {
+                i += 1;
+            }
+        } else {
+            i += 1;
+        }
+    }
+
+    // If alias provided, add to programs list
+    if let Some(alias_name) = alias {
+        programs.push(alias_name);
+    }
+
+    // Find an available port
+    let port = find_available_port()?;
+
+    // Spawn internal stream server in background
+    spawn_internal_server(port, programs.clone(), tokens.clone(), accounts.clone())?;
+
+    // Wait for server to start
+    thread::sleep(Duration::from_millis(1000));
+
+    // Connect via WebSocket
+    let ws_url = format!("ws://127.0.0.1:{}/ws", port);
+    let mut connect_args = vec![Value::String(ws_url)];
+
+    // Add filters if provided
+    if !programs.is_empty() {
+        connect_args.push(Value::String(":programs".to_string()));
+        connect_args.push(Value::Array(Arc::new(
+            programs.into_iter().map(Value::String).collect(),
+        )));
+    }
+    if !tokens.is_empty() {
+        connect_args.push(Value::String(":tokens".to_string()));
+        connect_args.push(Value::Array(Arc::new(
+            tokens.into_iter().map(Value::String).collect(),
+        )));
+    }
+    if !accounts.is_empty() {
+        connect_args.push(Value::String(":accounts".to_string()));
+        connect_args.push(Value::Array(Arc::new(
+            accounts.into_iter().map(Value::String).collect(),
+        )));
+    }
+
+    // Call stream_connect
+    stream_connect(&connect_args)
+}
+
+/// Find an available port for the internal server
+fn find_available_port() -> Result<u16> {
+    use std::net::TcpListener;
+
+    // Try ports 18080-18180
+    for port in 18080..18180 {
+        if TcpListener::bind(("127.0.0.1", port)).is_ok() {
+            return Ok(port);
+        }
+    }
+
+    Err(Error::runtime("Could not find available port for internal stream server".to_string()))
+}
+
+/// Spawn internal stream server in background thread
+fn spawn_internal_server(
+    port: u16,
+    programs: Vec<String>,
+    tokens: Vec<String>,
+    accounts: Vec<String>,
+) -> Result<()> {
+    thread::spawn(move || {
+        // This would need to call the actual stream server code
+        // For now, this is a placeholder that shows the architecture
+        eprintln!("Internal stream server would start on port {} with filters:", port);
+        eprintln!("  Programs: {:?}", programs);
+        eprintln!("  Tokens: {:?}", tokens);
+        eprintln!("  Accounts: {:?}", accounts);
+
+        // TODO: Actually spawn the stream server here
+        // Need to refactor stream service to be embeddable
+    });
+
+    Ok(())
+}