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

handshake_core/
application.rs
