theater-cli 0.3.7

Command-line interface for Theater actor system
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
use anyhow::Result;
use clap::{Parser, ValueEnum};
use std::collections::HashMap;
use std::fs::{self, OpenOptions};
use std::io::Write;
use std::path::PathBuf;
use tokio::sync::mpsc;
use tracing::{debug, error};

use crate::{error::CliError, CommandContext};
use theater::chain::ChainEvent;
use theater::config::actor_manifest::{
    RuntimeHostConfig, StoreHandlerConfig, SupervisorHostConfig, TcpHandlerConfig,
    TerminalHandlerConfig, TimerHandlerConfig,
};
use theater::handler::HandlerRegistry;
use theater::messages::TheaterCommand;
use theater::pack_bridge::{Value, ValueType};
use theater::theater_runtime::TheaterRuntime;
use theater::utils::resolve_reference;
use theater::ManifestConfig;
use theater::TheaterId;
use theater_handler_loop::LoopHandler;
use theater_handler_message_server::{MessageRouter, MessageServerHandler};
use theater_handler_rpc::RpcHandler;
use theater_handler_runtime::RuntimeHandler;
use theater_handler_store::StoreHandler;
use theater_handler_supervisor::SupervisorHandler;
use theater_handler_tcp::TcpHandler;
use theater_handler_terminal::TerminalHandler;
use theater_handler_timer::TimerHandler;

/// Output format for chain events
#[derive(Debug, Clone, Copy, ValueEnum, Default)]
pub enum EventFormat {
    /// JSON format (one JSON object per line)
    Json,
    /// Short format (compact, one line per event)
    #[default]
    Short,
    /// Full format (complete event data, multi-line)
    Full,
}

#[derive(Debug, Parser)]
pub struct StartArgs {
    /// Path or URL to the actor manifest file
    #[arg(default_value = "manifest.toml")]
    pub manifest: String,

    /// Output chain events from all actors
    #[arg(long)]
    pub events: bool,

    /// Format for event output (used with --events)
    #[arg(long, value_enum, default_value = "short")]
    pub events_format: EventFormat,

    /// Directory to persist chain events (one file per actor)
    #[arg(long)]
    pub chain_dir: Option<PathBuf>,

    /// Skip calling the actor's init function after spawning
    #[arg(long)]
    pub no_init: bool,

    /// Disable actor log output to stdout
    #[arg(long)]
    pub no_actor_logs: bool,
}

/// Format a chain event with actor ID prefix using ChainEvent's Display impl (short)
fn format_event_short(event: &ChainEvent, actor_id: &TheaterId) -> String {
    let id_str = actor_id.to_string();
    let short_id = &id_str[..8.min(id_str.len())];
    format!("[{}] {}\n", short_id, event)
}

/// Format a chain event with full data (multi-line, complete)
fn format_event_full(event: &ChainEvent, actor_id: &TheaterId) -> String {
    let id_str = actor_id.to_string();
    let short_id = &id_str[..8.min(id_str.len())];
    let hash_hex = hex::encode(&event.hash);
    let parent_hex = event
        .parent_hash
        .as_ref()
        .map(hex::encode)
        .unwrap_or_else(|| "none".to_string());
    let data_str = String::from_utf8_lossy(&event.data);

    format!(
        "EVENT [{}] {}\nparent: {}\ntype: {}\nsize: {}\n{}\n\n",
        short_id,
        hash_hex,
        parent_hex,
        event.event_type,
        event.data.len(),
        data_str
    )
}

/// Format a chain event as JSON for stdout
fn format_event_json(event: &ChainEvent, actor_id: &TheaterId) -> String {
    let json = serde_json::json!({
        "actor_id": actor_id.to_string(),
        "hash": hex::encode(&event.hash),
        "parent_hash": event.parent_hash.as_ref().map(hex::encode),
        "event_type": event.event_type,
        "data": format!("{} bytes (pack-encoded)", event.data.len())
    });
    serde_json::to_string(&json).unwrap_or_else(|_| "{}".to_string())
}

/// Manages chain file writers for multiple actors
struct ChainFileManager {
    dir: PathBuf,
    files: HashMap<TheaterId, std::fs::File>,
}

impl ChainFileManager {
    fn new(dir: PathBuf) -> Result<Self, CliError> {
        fs::create_dir_all(&dir).map_err(|e| {
            CliError::file_operation_failed("create directory", dir.display().to_string(), e)
        })?;
        Ok(Self {
            dir,
            files: HashMap::new(),
        })
    }

    fn write_event(&mut self, actor_id: &TheaterId, event: &ChainEvent) -> Result<(), CliError> {
        let file = self.files.entry(*actor_id).or_insert_with(|| {
            let path = self.dir.join(format!("{}.chain", actor_id));
            OpenOptions::new()
                .create(true)
                .append(true)
                .open(&path)
                .expect("Failed to open chain file")
        });

        let block = format_event_full(event, actor_id);
        file.write_all(block.as_bytes()).map_err(|e| {
            CliError::file_operation_failed("write event", format!("{}.chain", actor_id), e)
        })?;
        file.flush().map_err(|e| {
            CliError::file_operation_failed("flush", format!("{}.chain", actor_id), e)
        })?;
        Ok(())
    }
}

/// Create a handler registry with all Theater handlers
fn create_handler_registry(
    theater_tx: mpsc::Sender<TheaterCommand>,
    show_actor_logs: bool,
) -> HandlerRegistry {
    let mut registry = HandlerRegistry::new();

    // Runtime handler - provides log, get-chain, shutdown
    let runtime_config = RuntimeHostConfig {};
    registry.register(
        RuntimeHandler::new(runtime_config, theater_tx.clone(), None)
            .with_show_logs(show_actor_logs),
    );

    // Store handler - provides content storage
    let store_config = StoreHandlerConfig::default();
    registry.register(StoreHandler::new(store_config, None));

    // Supervisor handler - allows spawning/managing child actors
    let supervisor_config = SupervisorHostConfig {};
    registry.register(SupervisorHandler::new(supervisor_config, None));

    // Message server handler - inter-actor messaging
    let message_router = MessageRouter::new();
    registry.register(MessageServerHandler::new(None, message_router.clone()));

    // RPC handler - direct actor-to-actor function calls
    registry.register(RpcHandler::new(theater_tx.clone()));

    // TCP handler - TCP server/client functionality
    let tcp_config = TcpHandlerConfig {
        listen: None,
        max_connections: None,
        ..Default::default()
    };
    registry.register(TcpHandler::new(tcp_config));

    // Terminal handler - stdin/stdout/stderr for interactive CLI apps
    let terminal_config = TerminalHandlerConfig::default();
    registry.register(TerminalHandler::new(terminal_config));

    // Timer handler - periodic tick callbacks for game loops, polling, etc.
    let timer_config = TimerHandlerConfig::default();
    registry.register(TimerHandler::new(timer_config));

    // Loop handler - cooperative looping with yield points
    registry.register(LoopHandler::new());

    registry
}

/// Execute the start command - spin up a local runtime and run the actor
pub async fn execute_async(args: &StartArgs, ctx: &CommandContext) -> Result<(), CliError> {
    debug!("Starting actor from manifest: {}", args.manifest);

    // Resolve the manifest reference (file path, URL, or store path)
    let manifest_bytes = resolve_reference(&args.manifest).await.map_err(|e| {
        CliError::invalid_manifest(format!(
            "Failed to resolve manifest reference '{}': {}",
            args.manifest, e
        ))
    })?;

    let manifest_content = String::from_utf8(manifest_bytes).map_err(|e| {
        CliError::invalid_manifest(format!("Manifest content is not valid UTF-8: {}", e))
    })?;

    // Set up chain file manager if --chain-dir is specified
    let mut chain_file_manager = if let Some(ref dir) = args.chain_dir {
        Some(ChainFileManager::new(dir.clone())?)
    } else {
        None
    };

    // Create the TheaterRuntime in-process
    let (theater_tx, theater_rx) = mpsc::channel::<TheaterCommand>(32);
    let handler_registry = create_handler_registry(theater_tx.clone(), !args.no_actor_logs);

    let mut runtime = TheaterRuntime::new(
        theater_tx.clone(),
        theater_rx,
        None, // no channel events forwarding needed
        handler_registry,
    )
    .await
    .map_err(|e| CliError::server_error(format!("Failed to create runtime: {}", e)))?;

    // Set up global event subscription (receives events from ALL actors)
    let (global_events_tx, mut global_events_rx) = mpsc::channel(256);
    runtime.add_global_subscription(global_events_tx);

    // Spawn the runtime event loop in a background task
    let runtime_handle = tokio::spawn(async move {
        if let Err(e) = runtime.run().await {
            error!("Theater runtime error: {}", e);
        }
    });

    // Parse the manifest
    let manifest = ManifestConfig::from_toml_str(&manifest_content)
        .map_err(|e| CliError::invalid_manifest(format!("Failed to parse manifest: {}", e)))?;

    // Resolve WASM path relative to manifest directory
    let wasm_path = if manifest.package.starts_with('/') || manifest.package.contains("://") {
        // Absolute path or URL - use as is
        manifest.package.clone()
    } else {
        // Relative path - resolve relative to manifest's directory
        let manifest_path = std::path::Path::new(&args.manifest);
        if let Some(manifest_dir) = manifest_path.parent() {
            manifest_dir
                .join(&manifest.package)
                .to_string_lossy()
                .to_string()
        } else {
            manifest.package.clone()
        }
    };

    // Load WASM bytes
    let wasm_bytes = resolve_reference(&wasm_path).await.map_err(|e| {
        CliError::server_error(format!("Failed to load WASM from '{}': {}", wasm_path, e))
    })?;

    // Spawn the actor
    let (response_tx, response_rx) = tokio::sync::oneshot::channel();

    // Set up a supervisor channel so we get notified when the actor exits
    let (supervisor_tx, mut supervisor_rx) = mpsc::channel(32);

    theater_tx
        .send(TheaterCommand::SpawnActor {
            wasm_bytes,
            name: Some(manifest.name.clone()),
            manifest: Some(manifest),
            init_bytes: None,
            response_tx,
            supervisor_tx: Some(supervisor_tx),
            subscription_tx: None, // Using global subscription instead
        })
        .await
        .map_err(|e| CliError::server_error(format!("Failed to send spawn command: {}", e)))?;

    // Wait for the actor to start
    let actor_id = match response_rx.await {
        Ok(Ok(id)) => {
            debug!("Actor started: {}", id);
            id
        }
        Ok(Err(e)) => {
            return Err(CliError::server_error(format!(
                "Failed to start actor: {}",
                e
            )));
        }
        Err(e) => {
            return Err(CliError::server_error(format!(
                "Failed to receive spawn response: {}",
                e
            )));
        }
    };

    // Call init unless --no-init flag is set
    if !args.no_init {
        // Get the actor handle
        let (handle_tx, handle_rx) = tokio::sync::oneshot::channel();
        theater_tx
            .send(TheaterCommand::GetActorHandle {
                actor_id,
                response_tx: handle_tx,
            })
            .await
            .map_err(|e| CliError::server_error(format!("Failed to get actor handle: {}", e)))?;

        let actor_handle = match handle_rx.await {
            Ok(Some(handle)) => handle,
            Ok(None) => {
                return Err(CliError::server_error("Actor handle not found".to_string()));
            }
            Err(e) => {
                return Err(CliError::server_error(format!(
                    "Failed to receive actor handle: {}",
                    e
                )));
            }
        };

        // Build init state (None for now)
        let init_state = Value::Option {
            inner_type: ValueType::List(Box::new(ValueType::U8)),
            value: None,
        };

        // Call init
        let init_params = Value::Tuple(vec![init_state]);
        debug!("Calling init on actor {}", actor_id);
        let _init_result = actor_handle
            .call_function("theater:simple/actor.init".to_string(), init_params)
            .await
            .map_err(|e| CliError::server_error(format!("Failed to call init: {:?}", e)))?;
        debug!("Init completed");
    }

    // Now wait for either:
    // - The actor to exit (supervisor notification)
    // - Ctrl+C
    // - Shutdown token cancellation
    //
    // Output modes:
    // - Default: print only log messages as [actor-id] message
    // - --events: print all chain events as JSON
    // - --chain-dir: also persist events to files
    loop {
        tokio::select! {
            // Actor result (exit/error)
            result = supervisor_rx.recv() => {
                match result {
                    Some(actor_result) => {
                        debug!("Actor exited: {:?}", actor_result);
                        match actor_result {
                            theater::messages::ActorResult::Success(success) => {
                                if let Some(output) = success.result {
                                    // Write actor result to stdout
                                    let _ = std::io::stdout().write_all(&output);
                                    let _ = std::io::stdout().flush();
                                }
                            }
                            theater::messages::ActorResult::Error(err) => {
                                eprintln!("Actor error: {}", err.error);
                                std::process::exit(1);
                            }
                            theater::messages::ActorResult::ExternalStop(_) => {
                                debug!("Actor stopped externally");
                            }
                        }
                        break;
                    }
                    None => {
                        // Supervisor channel closed, actor is done
                        debug!("Supervisor channel closed");
                        break;
                    }
                }
            }

            // Global event subscription (all actors)
            event = global_events_rx.recv() => {
                if let Some((event_actor_id, event_result)) = event {
                    match event_result {
                        Ok(chain_event) => {
                            // Persist to chain file if enabled
                            if let Some(ref mut manager) = chain_file_manager {
                                if let Err(e) = manager.write_event(&event_actor_id, &chain_event) {
                                    eprintln!("Warning: failed to write chain event: {}", e);
                                }
                            }

                            // Output events if --events mode is enabled
                            // (Actor logs are printed directly by RuntimeHandler, not extracted here)
                            if args.events {
                                match args.events_format {
                                    EventFormat::Json => {
                                        println!("{}", format_event_json(&chain_event, &event_actor_id));
                                    }
                                    EventFormat::Short => {
                                        print!("{}", format_event_short(&chain_event, &event_actor_id));
                                    }
                                    EventFormat::Full => {
                                        print!("{}", format_event_full(&chain_event, &event_actor_id));
                                    }
                                }
                            }

                            // Check for root actor shutdown
                            if event_actor_id == actor_id && chain_event.event_type == "shutdown" {
                                break;
                            }
                        }
                        Err(e) => {
                            debug!("Actor error event: {:?}", e);
                        }
                    }
                }
            }

            // Ctrl+C
            _ = tokio::signal::ctrl_c() => {
                debug!("Received Ctrl+C, stopping actor {}", actor_id);
                eprintln!("\nStopping actor...");

                let (stop_tx, stop_rx) = tokio::sync::oneshot::channel();
                let _ = theater_tx.send(TheaterCommand::StopActor {
                    actor_id,
                    response_tx: stop_tx,
                }).await;

                // Wait briefly for graceful shutdown
                match tokio::time::timeout(
                    tokio::time::Duration::from_secs(5),
                    stop_rx,
                ).await {
                    Ok(Ok(Ok(()))) => debug!("Actor stopped gracefully"),
                    _ => debug!("Actor stop timed out or failed"),
                }
                break;
            }

            // Shutdown token
            _ = ctx.shutdown_token.cancelled() => {
                debug!("Shutdown token cancelled");
                break;
            }
        }
    }

    // Drop the theater_tx to signal the runtime to stop
    drop(theater_tx);

    // Wait for runtime to finish (with timeout)
    let _ = tokio::time::timeout(tokio::time::Duration::from_secs(5), runtime_handle).await;

    Ok(())
}