twine_core/
application.rs

1use std::collections::HashMap;
2use std::path::{Path, PathBuf};
3use std::sync::{Arc, Mutex, MutexGuard};
4
5use thiserror::Error;
6use tracing::{debug, info};
7
8use crate::config::Config;
9use crate::event::{CommandResult, Event, EventError, EventJournal, EventKind, StateEvent};
10use crate::folder::{FolderError, FolderState, Folders};
11use crate::store::{Store, StoreError};
12use crate::terminal::{
13    TerminalError, TerminalId, TerminalManager, TerminalSize, TerminalState, TerminalStatus,
14    TerminalStream, TranscriptError, TranscriptRecorder,
15};
16
17use crate::harness::{HarnessId, LoginPath};
18use crate::workflow::{SessionId, WorkflowId, WorkflowKind, WorkflowState};
19
20mod agents;
21mod configuration;
22mod files;
23mod git;
24mod harness_steps;
25#[cfg(test)]
26mod recovery;
27mod resume_agents;
28mod run_inputs;
29mod runs;
30mod sessions;
31mod terminals;
32mod traces;
33mod windows;
34mod workflow_types;
35mod workflows;
36
37const DATABASE_FILE_NAME: &str = "twine.db";
38const DEFAULT_EVENT_CAPACITY: usize = 4_096;
39const DEFAULT_TERMINAL_CAPACITY_BYTES: usize = 1024 * 1024;
40const DEFAULT_TERMINAL_CAPACITY_CHUNKS: usize = 4_096;
41
42#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
43pub struct RequestId(pub u64);
44
45#[derive(Clone, Debug, Eq, PartialEq)]
46pub enum Command {
47    Ping,
48    /// Applies the current user config to this runtime without restarting its processes.
49    ReloadConfig,
50    ValidateWorkflowType {
51        definition: crate::WorkflowTypeDefinition,
52    },
53    SaveWorkflowType {
54        expected_definition: Option<Box<crate::WorkflowTypeDefinition>>,
55        source: Option<crate::WorkflowTypeRef>,
56        definition: crate::WorkflowTypeDefinition,
57    },
58    DeleteWorkflowType {
59        reference: crate::WorkflowTypeRef,
60        expected_definition: Box<crate::WorkflowTypeDefinition>,
61    },
62    /// Opens the folder at an absolute path and records it as the most recent folder.
63    OpenFolder {
64        path: PathBuf,
65    },
66    /// Attaches a window to a persisted restore entry, including an unavailable folder.
67    RestoreFolder {
68        path: PathBuf,
69    },
70    /// Closes the open folder, so the window shows the start page.
71    CloseFolder,
72    /// Used by a departing window; an old folder must not close its replacement.
73    CloseFolderIfOpen {
74        path: PathBuf,
75    },
76    RemoveRecentFolder {
77        path: PathBuf,
78    },
79    RefreshGitBranch {
80        folder: PathBuf,
81    },
82    CreateSession {
83        folder: PathBuf,
84        name: String,
85    },
86    RenameSession {
87        session_id: SessionId,
88        name: String,
89    },
90    SelectSession {
91        session_id: SessionId,
92    },
93    DeleteSession {
94        session_id: SessionId,
95    },
96    CreateWorkflow {
97        folder: PathBuf,
98        session_id: Option<SessionId>,
99        kind: WorkflowKind,
100        /// One role per agent, in order. Agents workflows need at least one; other kinds take none.
101        roles: Vec<String>,
102        size: TerminalSize,
103    },
104    ActivateWorkflow {
105        workflow_id: WorkflowId,
106    },
107    NameDraftWorkflow {
108        workflow_id: WorkflowId,
109        name: String,
110    },
111    CloseWorkflow {
112        workflow_id: WorkflowId,
113    },
114    /// Runs a draft workflow as a single agent: `harness` started with `prompt`, or waiting for
115    /// the user in its terminal when `prompt` is empty.
116    StartAgent {
117        workflow_id: WorkflowId,
118        harness: HarnessId,
119        /// The harness's model, or its own default when absent.
120        model: Option<String>,
121        /// The harness's effort level, or its own default when absent.
122        effort: Option<String>,
123        /// Skips the harness's permission prompts, where it has them.
124        yolo: bool,
125        prompt: String,
126        size: TerminalSize,
127    },
128    ResumeAgent {
129        workflow_id: WorkflowId,
130        session: String,
131    },
132    /// Stops a running agent, leaving its workflow open.
133    CancelAgent {
134        workflow_id: WorkflowId,
135    },
136    /// Starts a workflow run. With an empty `prompt`, the first stage asks the user for the task
137    /// and reports it when it completes.
138    StartWorkflowRun {
139        workflow_id: WorkflowId,
140        workflow_type: crate::WorkflowTypeRef,
141        prompt: String,
142        roles: Vec<crate::RoleLaunch>,
143        size: TerminalSize,
144    },
145    CompleteWorkflowRole {
146        workflow_id: WorkflowId,
147        agent_id: crate::AgentId,
148        generation: u64,
149        signal: crate::CompletionSignal,
150    },
151    /// A follow-up to a first-stage agent starts another cycle with the same conversations.
152    ContinueWorkflowRun {
153        workflow_id: WorkflowId,
154        agent_id: crate::AgentId,
155        generation: u64,
156        mode_revision: u64,
157    },
158    SetWorkflowIndividualMode {
159        workflow_id: WorkflowId,
160        generation: u64,
161        mode_revision: u64,
162        individual_mode: bool,
163    },
164    SubmitWorkflowTask {
165        workflow_id: WorkflowId,
166        generation: u64,
167        prompt: String,
168    },
169    ChooseWorkflowResult {
170        workflow_id: WorkflowId,
171        generation: u64,
172        agent_id: crate::AgentId,
173    },
174    CancelWorkflowRun {
175        workflow_id: WorkflowId,
176    },
177    StartTerminal {
178        working_directory: PathBuf,
179        size: TerminalSize,
180    },
181    CloseTerminal {
182        terminal_id: TerminalId,
183    },
184}
185
186#[derive(Clone, Debug, Eq, PartialEq)]
187pub enum CommandDisposition {
188    Accepted,
189    Rejected { code: String, message: String },
190}
191
192#[derive(Clone, Debug, Eq, PartialEq)]
193pub struct CommandReceipt {
194    pub request_id: RequestId,
195    pub disposition: CommandDisposition,
196}
197
198#[derive(Clone, Copy, Debug, Eq, PartialEq)]
199pub enum ApplicationState {
200    Ready,
201}
202
203#[derive(Clone, Debug, Eq, PartialEq)]
204pub struct Snapshot {
205    pub sequence: u64,
206    pub state: ApplicationState,
207    pub config: Config,
208    pub folders: FolderState,
209    pub terminals: Vec<TerminalState>,
210    pub workflows: WorkflowState,
211    pub traces: Vec<crate::TraceSummary>,
212    pub workflow_types: Vec<crate::WorkflowType>,
213}
214
215#[derive(Debug)]
216struct Inner {
217    state: ApplicationState,
218    config: Config,
219    folders: Folders,
220    events: EventJournal,
221    terminals: HashMap<TerminalId, TerminalStatus>,
222    workflows: WorkflowState,
223    trace_spans: HashMap<TerminalId, crate::TraceSpanId>,
224    step_terminals: HashMap<TerminalId, WorkflowId>,
225    pending_trace_endings: HashMap<TerminalId, traces::PendingTraceEnding>,
226    command_shells: std::collections::HashSet<TerminalId>,
227    shell_process_endings: HashMap<TerminalId, traces::PendingTraceEnding>,
228    pending_shell_marks: std::collections::VecDeque<crate::terminal::ShellObservation>,
229    pending_shell_workflows: HashMap<TerminalId, crate::Workflow>,
230    workflow_types_cache: Vec<crate::WorkflowType>,
231    workflow_types_polled: Option<std::time::Instant>,
232}
233
234pub struct Application {
235    window_storage: Option<windows::WindowStorageLease>,
236    /// Window runtimes share their recorder; only its final owner shuts it down.
237    shared_recording: bool,
238    files: crate::files::FileWatcher,
239    // Serialize lifetime commands without holding state while joining terminal supervisors.
240    commands: Mutex<()>,
241    inner: Arc<Mutex<Inner>>,
242    terminal_output: Arc<TerminalStream>,
243    terminals: TerminalManager,
244    /// Shared with threads that list harness models.
245    login_path: Arc<LoginPath>,
246    model_lists: crate::harness::models::ModelListWorkers,
247    /// Makes harness lookup search only this `PATH`, so tests don't depend on installed tools.
248    #[cfg(test)]
249    harness_path: Option<std::ffi::OsString>,
250    run_processes: Mutex<HashMap<WorkflowId, runs::RunProcesses>>,
251    harness_steps: Mutex<HashMap<TerminalId, harness_steps::HarnessRecording>>,
252    history: Mutex<crate::harness::history::HistoryWorkers>,
253}
254
255impl Application {
256    /// Creates the application with production queue limits and the user's config file, keeping
257    /// its database in `data_directory`, and reopens the folder that was open when Twine last quit.
258    ///
259    /// # Errors
260    ///
261    /// Returns an error if the database cannot be opened or migrated, transcript storage cannot
262    /// be opened or is owned by another core, or the initial ready event cannot be recorded.
263    pub fn new(data_directory: &Path) -> Result<Self, ApplicationError> {
264        Self::with_config(data_directory, Config::load_user())
265    }
266
267    /// Creates the application like [`Application::new`], but with already validated
268    /// configuration instead of the user's config file.
269    ///
270    /// # Errors
271    ///
272    /// Returns an error if the database cannot be opened or migrated, transcript storage cannot
273    /// be opened or is owned by another core, or the initial ready event cannot be recorded.
274    pub fn with_config(data_directory: &Path, config: Config) -> Result<Self, ApplicationError> {
275        // Acquire directory ownership before migrating or recovering any durable state.
276        let transcripts = Arc::new(TranscriptRecorder::open(
277            &data_directory.join("transcripts"),
278        )?);
279        let store = Store::open(&data_directory.join(DATABASE_FILE_NAME))?;
280        Self::with_store(store, config, DEFAULT_EVENT_CAPACITY, transcripts)
281    }
282
283    /// Creates the application with an in-memory database, default configuration, and an
284    /// injectable event limit for deterministic tests.
285    #[cfg(test)]
286    pub(crate) fn with_event_capacity(event_capacity: usize) -> Result<Self, ApplicationError> {
287        let mut application = Self::with_store(
288            Store::open_in_memory()?,
289            Config::default(),
290            event_capacity,
291            Arc::new(TranscriptRecorder::temporary()?),
292        )?;
293        application
294            .terminals
295            .set_test_shell(PathBuf::from("/bin/sh"));
296        Ok(application)
297    }
298
299    fn with_store(
300        mut store: Store,
301        config: Config,
302        event_capacity: usize,
303        transcripts: Arc<TranscriptRecorder>,
304    ) -> Result<Self, ApplicationError> {
305        // Recover every folder, including closed or currently unavailable folders, before ready.
306        store.recover_interrupted_work()?;
307        let folders = Folders::restore(store)?;
308        Self::with_folders(folders, config, event_capacity, transcripts, false)
309    }
310
311    fn with_folders(
312        mut folders: Folders,
313        config: Config,
314        event_capacity: usize,
315        transcripts: Arc<TranscriptRecorder>,
316        shared_recording: bool,
317    ) -> Result<Self, ApplicationError> {
318        folders.store().configure_trace_storage(&config.traces)?;
319        let mut events = EventJournal::new(event_capacity)?;
320        events.append(EventKind::State(StateEvent::ApplicationReady))?;
321
322        let terminal_output = Arc::new(TerminalStream::new(
323            DEFAULT_TERMINAL_CAPACITY_BYTES,
324            DEFAULT_TERMINAL_CAPACITY_CHUNKS,
325            transcripts,
326        )?);
327        let terminals = TerminalManager::new(Arc::clone(&terminal_output));
328        #[cfg(test)]
329        let terminals = {
330            let mut terminals = terminals;
331            // Unit tests must not depend on the developer's login shell or startup config.
332            terminals.set_test_shell(PathBuf::from("/bin/sh"));
333            terminals
334        };
335        let application = Self {
336            window_storage: None,
337            shared_recording,
338            files: crate::files::FileWatcher::new()?,
339            run_processes: Mutex::new(HashMap::new()),
340            harness_steps: Mutex::new(HashMap::new()),
341            history: Mutex::new(
342                crate::harness::history::HistoryWorkers::new()
343                    .map_err(crate::StoreError::TracePayload)?,
344            ),
345            commands: Mutex::new(()),
346            inner: Arc::new(Mutex::new(Inner {
347                state: ApplicationState::Ready,
348                config,
349                folders,
350                events,
351                terminals: HashMap::new(),
352                workflows: WorkflowState::default(),
353                trace_spans: HashMap::new(),
354                step_terminals: HashMap::new(),
355                pending_trace_endings: HashMap::new(),
356                command_shells: std::collections::HashSet::new(),
357                shell_process_endings: HashMap::new(),
358                pending_shell_marks: std::collections::VecDeque::new(),
359                pending_shell_workflows: HashMap::new(),
360                workflow_types_cache: Vec::new(),
361                workflow_types_polled: None,
362            })),
363            terminal_output: Arc::clone(&terminal_output),
364            terminals,
365            model_lists: crate::harness::models::ModelListWorkers::default(),
366            // Unit tests must not start the developer's login shell.
367            #[cfg(test)]
368            login_path: Arc::new(LoginPath::ready(None)),
369            #[cfg(not(test))]
370            login_path: Arc::new(LoginPath::spawn(crate::terminal::login_shell())),
371            #[cfg(test)]
372            harness_path: None,
373        };
374
375        application.restore_workflows()?;
376        let _ = application.trace_storage_status(None)?;
377        info!("application core initialized");
378        Ok(application)
379    }
380
381    /// Handles a command and returns its immediate acceptance result.
382    ///
383    /// # Errors
384    ///
385    /// Returns an error if application state cannot be accessed or the resulting event cannot be
386    /// recorded.
387    #[expect(clippy::too_many_lines, reason = "one short arm per command")]
388    pub fn handle_command(
389        &self,
390        request_id: RequestId,
391        command: Command,
392    ) -> Result<CommandReceipt, ApplicationError> {
393        let _command_guard = self
394            .commands
395            .lock()
396            .map_err(|_| ApplicationError::Poisoned)?;
397        self.lock_inner()?.retry_trace_endings();
398        self.poll_harness_steps()?;
399        self.poll_shell_observations()?;
400        let disposition = match command {
401            Command::ReloadConfig => self.reload_config()?,
402            Command::ValidateWorkflowType { definition } => {
403                self.validate_workflow_type(request_id, &definition)?
404            }
405            Command::SaveWorkflowType {
406                source,
407                definition,
408                expected_definition,
409            } => self.save_workflow_type(
410                request_id,
411                source,
412                &definition,
413                expected_definition.as_deref(),
414            )?,
415            Command::DeleteWorkflowType {
416                reference,
417                expected_definition,
418            } => self.delete_workflow_type(request_id, reference, &expected_definition)?,
419            Command::Ping => {
420                let mut inner = self.lock_inner()?;
421                inner.events.append(EventKind::CommandCompleted {
422                    request_id,
423                    result: CommandResult::Pong,
424                })?;
425                CommandDisposition::Accepted
426            }
427            Command::OpenFolder { path } => self.change_folder(Some(&path))?,
428            Command::RestoreFolder { path } => self.restore_window_folder(&path)?,
429            Command::CloseFolder => self.change_folder(None)?,
430            Command::CloseFolderIfOpen { path } => {
431                if self.lock_inner()?.folders.state().open_folder.as_deref() == Some(path.as_path())
432                {
433                    self.change_folder(None)?
434                } else {
435                    CommandDisposition::Accepted
436                }
437            }
438            Command::RemoveRecentFolder { path } => self
439                .lock_inner()?
440                .update_folders(|folders| folders.remove_recent(&path))?,
441            Command::RefreshGitBranch { folder } => self.refresh_git_branch(&folder)?,
442            Command::CreateSession { folder, name } => {
443                self.create_session(request_id, &folder, &name)?
444            }
445            Command::RenameSession { session_id, name } => {
446                self.rename_session(request_id, session_id, &name)?
447            }
448            Command::SelectSession { session_id } => self.select_session(request_id, session_id)?,
449            Command::DeleteSession { session_id } => self.delete_session(request_id, session_id)?,
450            Command::CreateWorkflow {
451                folder,
452                session_id,
453                kind,
454                roles,
455                size,
456            } => self.create_workflow(request_id, &folder, session_id, kind, &roles, size)?,
457            Command::ActivateWorkflow { workflow_id } => {
458                self.activate_workflow(request_id, workflow_id)?
459            }
460            Command::NameDraftWorkflow { workflow_id, name } => {
461                self.name_draft_workflow(workflow_id, &name)?
462            }
463            Command::CloseWorkflow { workflow_id } => {
464                self.close_workflow(request_id, workflow_id)?
465            }
466            Command::StartAgent {
467                workflow_id,
468                harness,
469                model,
470                effort,
471                yolo,
472                prompt,
473                size,
474            } => self.start_agent(
475                request_id,
476                workflow_id,
477                harness,
478                (model.as_deref(), effort.as_deref(), yolo),
479                &prompt,
480                size,
481            )?,
482            Command::ResumeAgent {
483                workflow_id,
484                session,
485            } => self.resume_agent(request_id, workflow_id, &session)?,
486            Command::CancelAgent { workflow_id } => self.cancel_agent(request_id, workflow_id)?,
487            Command::StartWorkflowRun {
488                workflow_id,
489                workflow_type,
490                prompt,
491                roles,
492                size,
493            } => self.start_workflow_run(workflow_id, workflow_type, prompt, &roles, size)?,
494            Command::CompleteWorkflowRole {
495                workflow_id,
496                agent_id,
497                generation,
498                signal,
499            } => self.complete_workflow_role(workflow_id, agent_id, generation, signal)?,
500            Command::ContinueWorkflowRun {
501                workflow_id,
502                agent_id,
503                generation,
504                mode_revision,
505            } => self.continue_workflow_run(workflow_id, agent_id, generation, mode_revision)?,
506            Command::SetWorkflowIndividualMode {
507                workflow_id,
508                generation,
509                mode_revision,
510                individual_mode,
511            } => self.set_workflow_individual_mode(
512                workflow_id,
513                generation,
514                mode_revision,
515                individual_mode,
516            )?,
517            Command::SubmitWorkflowTask {
518                workflow_id,
519                generation,
520                prompt,
521            } => self.submit_workflow_task(workflow_id, generation, prompt)?,
522            Command::ChooseWorkflowResult {
523                workflow_id,
524                generation,
525                agent_id,
526            } => self.choose_workflow_result(workflow_id, generation, agent_id)?,
527            Command::CancelWorkflowRun { workflow_id } => self.cancel_workflow_run(workflow_id)?,
528            Command::StartTerminal {
529                working_directory,
530                size,
531            } => self.start_terminal_command(request_id, &working_directory, size)?,
532            Command::CloseTerminal { terminal_id } => {
533                self.close_terminal_command(request_id, terminal_id)?
534            }
535        };
536        self.prune_run_processes()?;
537        match &disposition {
538            CommandDisposition::Accepted => debug!(request_id = request_id.0, "command accepted"),
539            CommandDisposition::Rejected { code, .. } => {
540                debug!(request_id = request_id.0, code, "command rejected");
541            }
542        }
543        Ok(CommandReceipt {
544            request_id,
545            disposition,
546        })
547    }
548
549    /// Returns an atomic snapshot of application state and the latest event sequence.
550    ///
551    /// # Errors
552    ///
553    /// Returns an error if application state cannot be accessed.
554    pub fn snapshot(&self) -> Result<Snapshot, ApplicationError> {
555        self.poll_harness_steps()?;
556        self.poll_shell_observations()?;
557        self.poll_workflow_signals()?;
558        let mut inner = self.lock_inner()?;
559        let workflow_types =
560            crate::workflow_type::WorkflowCatalog::new(inner.folders.store()).list()?;
561        inner.workflow_types_cache.clone_from(&workflow_types);
562        let mut terminals = inner
563            .terminals
564            .iter()
565            .map(|(terminal_id, status)| TerminalState {
566                terminal_id: *terminal_id,
567                status: status.clone(),
568            })
569            .collect::<Vec<_>>();
570        terminals.sort_by_key(|terminal| terminal.terminal_id.value());
571        Ok(Snapshot {
572            sequence: inner.events.latest_sequence(),
573            state: inner.state,
574            config: inner.config.clone(),
575            folders: inner.folders.state().clone(),
576            terminals,
577            workflows: inner.workflows.clone(),
578            traces: match &inner.folders.state().open_folder {
579                Some(folder) => inner.folders.read_store().trace_summaries(folder)?,
580                None => Vec::new(),
581            },
582            workflow_types,
583        })
584    }
585
586    /// Returns at most `limit` events whose sequence is strictly greater than `sequence`.
587    ///
588    /// # Errors
589    ///
590    /// Returns an error if the cursor has expired, the limit is invalid, or application state
591    /// cannot be accessed.
592    pub fn events_after(
593        &self,
594        sequence: u64,
595        limit: usize,
596    ) -> Result<Vec<Event>, ApplicationError> {
597        self.poll_shell_observations()?;
598        self.poll_workflow_signals()?;
599        let mut inner = self.lock_inner()?;
600        let previous = inner.folders.state().clone();
601        if inner.folders.refresh_shared()? && *inner.folders.state() != previous {
602            let folders = inner.folders.state().clone();
603            inner
604                .events
605                .append(EventKind::State(StateEvent::FoldersChanged(folders)))?;
606        }
607        if inner
608            .workflow_types_polled
609            .is_none_or(|time| time.elapsed() >= std::time::Duration::from_secs(1))
610        {
611            inner.workflow_types_polled = Some(std::time::Instant::now());
612            let types = crate::workflow_type::WorkflowCatalog::new(inner.folders.store()).list()?;
613            if types != inner.workflow_types_cache {
614                let initialized = !inner.workflow_types_cache.is_empty();
615                inner.workflow_types_cache.clone_from(&types);
616                if initialized {
617                    inner
618                        .events
619                        .append(EventKind::State(StateEvent::WorkflowTypesChanged(types)))?;
620                }
621            }
622        }
623        Ok(inner.events.after(sequence, limit)?)
624    }
625
626    fn lock_inner(&self) -> Result<MutexGuard<'_, Inner>, ApplicationError> {
627        self.inner.lock().map_err(|_| ApplicationError::Poisoned)
628    }
629}
630
631impl Inner {
632    /// Applies a folder change and records the new folder state as an event if it changed. Invalid
633    /// paths and folders that can't be opened reject the command; store failures are errors.
634    fn update_folders(
635        &mut self,
636        change: impl FnOnce(&mut Folders) -> Result<(), FolderError>,
637    ) -> Result<CommandDisposition, ApplicationError> {
638        let previous = self.folders.state().clone();
639        let result = change(&mut self.folders);
640        if *self.folders.state() != previous {
641            self.events
642                .append(EventKind::State(StateEvent::FoldersChanged(
643                    self.folders.state().clone(),
644                )))?;
645        }
646
647        let (code, error) = match result {
648            Ok(()) => return Ok(CommandDisposition::Accepted),
649            Err(FolderError::Store(error)) => return Err(error.into()),
650            Err(error @ FolderError::InvalidPath) => ("invalidPath", error),
651            Err(error @ FolderError::Missing) => ("folderNotFound", error),
652            Err(error @ FolderError::Inaccessible) => ("folderInaccessible", error),
653        };
654        Ok(rejection(code, &error))
655    }
656}
657
658impl Drop for Application {
659    fn drop(&mut self) {
660        // Model discovery may be waiting on the login shell. Cancel it before joining listings,
661        // including requests still held by the UI after this runtime closes.
662        self.login_path.cancel();
663        self.model_lists.shutdown();
664        self.login_path.shutdown();
665        let _ = self.poll_harness_steps();
666        let _ = self.poll_shell_observations();
667        if let Ok(mut inner) = self.inner.lock() {
668            inner.retry_trace_endings();
669            let ids = inner
670                .trace_spans
671                .keys()
672                .chain(inner.step_terminals.keys())
673                .copied()
674                .collect::<std::collections::HashSet<_>>();
675            for id in ids {
676                if let Ok(observation) = self.terminals.observe(id)
677                    && let Err(error) =
678                        inner.stop_trace(id, observation, "Process stopped when Twine quit.")
679                {
680                    tracing::warn!(%error, "failed to persist trace during shutdown");
681                }
682            }
683            // Exit callbacks are ignored after explicit shutdown takes ownership of stopping.
684            inner.terminals.clear();
685        }
686        self.terminals.shutdown();
687        if !self.shared_recording {
688            self.terminal_output.shutdown_recording();
689        }
690        self.window_storage.take();
691    }
692}
693
694fn rejection(code: &str, error: &dyn std::error::Error) -> CommandDisposition {
695    CommandDisposition::Rejected {
696        code: code.to_owned(),
697        message: error.to_string(),
698    }
699}
700
701#[derive(Debug, Error)]
702pub enum ApplicationError {
703    #[error("Could not create a workflow signal inbox: {0}")]
704    CompletionInbox(std::io::Error),
705    #[error(transparent)]
706    Harness(#[from] crate::harness::HarnessError),
707    #[error(transparent)]
708    Catalog(#[from] crate::CatalogError),
709    #[error(transparent)]
710    Files(#[from] crate::files::FileError),
711    #[error(transparent)]
712    Event(#[from] EventError),
713    #[error("session or workflow identifiers exhausted")]
714    IdExhausted,
715    #[error("application state lock is poisoned")]
716    Poisoned,
717    #[error(transparent)]
718    Store(#[from] StoreError),
719    #[error(transparent)]
720    Terminal(#[from] TerminalError),
721    #[error(transparent)]
722    Transcript(#[from] TranscriptError),
723    #[error(transparent)]
724    Trace(#[from] crate::TraceError),
725}
726
727#[cfg(test)]
728mod tests {
729    use std::sync::Arc;
730
731    use super::*;
732    use crate::config::{ColorScheme, Config};
733    use crate::folder::{UnavailableFolder, UnavailableReason};
734
735    #[test]
736    fn snapshot_then_events_has_no_gap_or_repeat() {
737        let data = tempfile::tempdir().expect("a data directory should be available");
738        let application = Application::with_config(data.path(), Config::default())
739            .expect("application should initialize");
740        let snapshot = application.snapshot().expect("snapshot should succeed");
741
742        let receipt = application
743            .handle_command(RequestId(41), Command::Ping)
744            .expect("command should succeed");
745        let events = application
746            .events_after(snapshot.sequence, 16)
747            .expect("events should be available");
748
749        assert_eq!(receipt.request_id, RequestId(41));
750        assert_eq!(receipt.disposition, CommandDisposition::Accepted);
751        assert_eq!(events.len(), 1);
752        assert_eq!(events[0].sequence, snapshot.sequence + 1);
753        assert_eq!(
754            events[0].kind,
755            EventKind::CommandCompleted {
756                request_id: RequestId(41),
757                result: CommandResult::Pong,
758            }
759        );
760        assert_eq!(
761            application
762                .events_after(events[0].sequence, 16)
763                .expect("up-to-date cursor should succeed"),
764            []
765        );
766    }
767
768    #[test]
769    fn expired_cursor_is_explicit() {
770        let application =
771            Application::with_event_capacity(2).expect("application should initialize");
772        application
773            .handle_command(RequestId(1), Command::Ping)
774            .expect("first command should succeed");
775        application
776            .handle_command(RequestId(2), Command::Ping)
777            .expect("second command should succeed");
778
779        let error = application
780            .events_after(0, 16)
781            .expect_err("the evicted cursor should fail");
782
783        assert!(matches!(
784            error,
785            ApplicationError::Event(EventError::CursorExpired {
786                requested: 0,
787                oldest_available: 2,
788            })
789        ));
790    }
791
792    #[test]
793    fn folder_commands_publish_folder_state_after_the_snapshot() {
794        let folder = tempfile::tempdir().expect("a folder should be available");
795        let application =
796            Application::with_event_capacity(16).expect("application should initialize");
797        let snapshot = application.snapshot().expect("snapshot should succeed");
798        assert_eq!(snapshot.folders, FolderState::default());
799
800        let receipt = application
801            .handle_command(
802                RequestId(1),
803                Command::OpenFolder {
804                    path: folder.path().to_owned(),
805                },
806            )
807            .expect("command should succeed");
808        assert_eq!(receipt.disposition, CommandDisposition::Accepted);
809        let events = application
810            .events_after(snapshot.sequence, 16)
811            .expect("events should be available");
812        let [event] = events.as_slice() else {
813            panic!("expected one event, got {events:?}");
814        };
815        assert_eq!(event.sequence, snapshot.sequence + 1);
816        let EventKind::State(StateEvent::FoldersChanged(folders)) = &event.kind else {
817            panic!("expected a folder event, got {event:?}");
818        };
819        assert_eq!(folders.open_folder.as_deref(), Some(folder.path()));
820        assert_eq!(
821            application.snapshot().expect("snapshot should succeed"),
822            Snapshot {
823                sequence: event.sequence,
824                state: ApplicationState::Ready,
825                config: Config::default(),
826                folders: folders.clone(),
827                terminals: Vec::new(),
828                workflows: WorkflowState::default(),
829                traces: Vec::new(),
830                workflow_types: application.workflow_types().unwrap(),
831            }
832        );
833    }
834
835    #[test]
836    fn rejected_terminal_commands_explain_themselves() {
837        let application =
838            Application::with_event_capacity(16).expect("application should initialize");
839        let missing = {
840            let folder = tempfile::tempdir().expect("a folder should be available");
841            folder.path().to_owned()
842        }; // Dropping `folder` deletes it.
843        let sequence = application
844            .snapshot()
845            .expect("snapshot should succeed")
846            .sequence;
847
848        let start = application
849            .handle_command(
850                RequestId(1),
851                Command::StartTerminal {
852                    working_directory: missing,
853                    size: TerminalSize {
854                        rows: 24,
855                        columns: 80,
856                        pixel_width: 800,
857                        pixel_height: 480,
858                    },
859                },
860            )
861            .expect("command should succeed");
862        assert!(matches!(
863            start.disposition,
864            CommandDisposition::Rejected { ref code, .. } if code == "terminalStartFailed"
865        ));
866
867        let close = application
868            .handle_command(
869                RequestId(2),
870                Command::CloseTerminal {
871                    terminal_id: TerminalId::from_value(99),
872                },
873            )
874            .expect("command should succeed");
875        assert!(matches!(
876            close.disposition,
877            CommandDisposition::Rejected { ref code, .. } if code == "terminalCloseFailed"
878        ));
879
880        let snapshot = application.snapshot().expect("snapshot should succeed");
881        assert_eq!(snapshot.sequence, sequence);
882        assert_eq!(snapshot.terminals, []);
883    }
884
885    #[test]
886    fn rejected_folder_commands_explain_themselves() {
887        let application =
888            Application::with_event_capacity(16).expect("application should initialize");
889        let missing = {
890            let folder = tempfile::tempdir().expect("a folder should be available");
891            folder.path().to_owned()
892        }; // Dropping `folder` deletes it.
893        let sequence = application
894            .snapshot()
895            .expect("snapshot should succeed")
896            .sequence;
897
898        let receipt = application
899            .handle_command(
900                RequestId(1),
901                Command::OpenFolder {
902                    path: missing.clone(),
903                },
904            )
905            .expect("command should succeed");
906        assert!(matches!(
907            receipt.disposition,
908            CommandDisposition::Rejected { ref code, .. } if code == "folderNotFound"
909        ));
910        let snapshot = application.snapshot().expect("snapshot should succeed");
911        assert_eq!(snapshot.sequence, sequence + 1);
912        assert_eq!(
913            snapshot.folders.unavailable_folder,
914            Some(UnavailableFolder {
915                path: missing,
916                reason: UnavailableReason::Missing,
917            })
918        );
919
920        let receipt = application
921            .handle_command(
922                RequestId(2),
923                Command::RemoveRecentFolder {
924                    path: PathBuf::from("relative"),
925                },
926            )
927            .expect("command should succeed");
928        assert!(matches!(
929            receipt.disposition,
930            CommandDisposition::Rejected { ref code, .. } if code == "invalidPath"
931        ));
932        assert_eq!(
933            application
934                .snapshot()
935                .expect("snapshot should succeed")
936                .sequence,
937            snapshot.sequence,
938            "a command that changes nothing publishes no event"
939        );
940    }
941
942    #[test]
943    fn a_new_application_reopens_the_last_folder_from_its_data_directory() {
944        let data = tempfile::tempdir().expect("a data directory should be available");
945        let folder = tempfile::tempdir().expect("a folder should be available");
946        let application = Application::with_config(data.path(), Config::default())
947            .expect("application should initialize");
948        application
949            .handle_command(
950                RequestId(1),
951                Command::OpenFolder {
952                    path: folder.path().to_owned(),
953                },
954            )
955            .expect("command should succeed");
956        drop(application);
957
958        let relaunched = Application::with_config(data.path(), Config::default())
959            .expect("application should initialize");
960        assert_eq!(
961            relaunched
962                .snapshot()
963                .expect("snapshot should succeed")
964                .folders
965                .open_folder
966                .as_deref(),
967            Some(folder.path())
968        );
969    }
970
971    #[test]
972    fn concurrent_snapshots_and_commands_keep_sequences_contiguous() {
973        const COMMANDS_PER_THREAD: usize = 500;
974        const THREADS: usize = 4;
975
976        let application = Arc::new(
977            Application::with_event_capacity(3_000).expect("application should initialize"),
978        );
979        let snapshot = application.snapshot().expect("snapshot should succeed");
980        let finished = Arc::new(std::sync::atomic::AtomicUsize::new(0));
981        let handles: Vec<_> = (0..THREADS)
982            .map(|thread| {
983                let application = Arc::clone(&application);
984                let finished = Arc::clone(&finished);
985                std::thread::spawn(move || {
986                    for request in 0..COMMANDS_PER_THREAD {
987                        application
988                            .handle_command(
989                                RequestId(
990                                    u64::try_from(thread * COMMANDS_PER_THREAD + request)
991                                        .expect("test request ID should fit"),
992                                ),
993                                Command::Ping,
994                            )
995                            .expect("command should succeed");
996                        if request % 25 == 0 {
997                            std::thread::yield_now();
998                        }
999                    }
1000                    finished.fetch_add(1, std::sync::atomic::Ordering::Release);
1001                })
1002            })
1003            .collect();
1004
1005        while finished.load(std::sync::atomic::Ordering::Acquire) != THREADS {
1006            let concurrent_snapshot = application.snapshot().expect("snapshot should succeed");
1007            let events = application
1008                .events_after(concurrent_snapshot.sequence, THREADS * COMMANDS_PER_THREAD)
1009                .expect("snapshot cursor should remain available");
1010            for (index, event) in events.iter().enumerate() {
1011                assert_eq!(
1012                    event.sequence,
1013                    concurrent_snapshot.sequence
1014                        + u64::try_from(index).expect("test index should fit")
1015                        + 1
1016                );
1017            }
1018        }
1019
1020        for handle in handles {
1021            handle.join().expect("command thread should not panic");
1022        }
1023
1024        let events = application
1025            .events_after(snapshot.sequence, THREADS * COMMANDS_PER_THREAD)
1026            .expect("all events should remain available");
1027        assert_eq!(events.len(), THREADS * COMMANDS_PER_THREAD);
1028        for (index, event) in events.iter().enumerate() {
1029            assert_eq!(
1030                event.sequence,
1031                snapshot.sequence + u64::try_from(index).expect("test index should fit") + 1
1032            );
1033        }
1034    }
1035
1036    #[test]
1037    fn snapshot_contains_loaded_config() {
1038        let directory = tempfile::tempdir().unwrap();
1039        let path = directory.path().join("config.toml");
1040        std::fs::write(
1041            &path,
1042            "[appearance]\ncolor_scheme = 'dark'\n[terminal]\nfont_family = 'Menlo'\nfont_size = 15.5\n",
1043        )
1044        .unwrap();
1045        let application =
1046            Application::with_config(directory.path(), Config::load(&path).config).unwrap();
1047        assert_eq!(
1048            application
1049                .snapshot()
1050                .unwrap()
1051                .config
1052                .appearance
1053                .color_scheme,
1054            ColorScheme::Dark
1055        );
1056        let config = application.snapshot().unwrap().config;
1057        assert_eq!(config.terminal.font_family, "Menlo");
1058        assert!((config.terminal.font_size.points() - 15.5).abs() < f64::EPSILON);
1059    }
1060}
twine_core/application.rs

twine_core/
application.rs
