handshake_core/application/
terminals.rs

1use std::collections::HashMap;
2use std::path::Path;
3use std::sync::Arc;
4
5use tracing::{debug, warn};
6
7use super::{Application, ApplicationError, CommandDisposition, Inner, RequestId, rejection};
8use crate::event::{CommandResult, EventKind, StateEvent};
9use crate::terminal::{
10    TerminalChunk, TerminalExit, TerminalId, TerminalObservation, TerminalSize, TerminalStatus,
11    TranscriptRead,
12};
13use crate::workflow::{Workflow, WorkflowKind, WorkflowStatus};
14
15impl Application {
16    /// Starts a shell and records it as running, or rejects the command if the shell can't start.
17    pub(super) fn start_terminal_command(
18        &self,
19        request_id: RequestId,
20        working_directory: &Path,
21        size: TerminalSize,
22    ) -> Result<CommandDisposition, ApplicationError> {
23        let reserved = match self.terminals.reserve_terminal() {
24            Ok(id) => id,
25            Err(error) => return Ok(rejection("terminalStartFailed", &error)),
26        };
27        // Hold state while the process starts so an immediately exiting shell cannot publish its
28        // exit before the command-completion event.
29        let mut inner = self.lock_inner()?;
30        let terminal_id = match self.start_terminal(reserved, working_directory, size, false) {
31            Ok(terminal_id) => terminal_id,
32            Err(error) => return Ok(rejection("terminalStartFailed", &error)),
33        };
34        inner.terminals.insert(terminal_id, TerminalStatus::Running);
35        if let Err(error) = inner.events.append(EventKind::CommandCompleted {
36            request_id,
37            result: CommandResult::TerminalStarted { terminal_id },
38        }) {
39            inner.terminals.remove(&terminal_id);
40            drop(inner);
41            let _ = self.terminals.close(terminal_id);
42            return Err(error.into());
43        }
44        debug!(
45            request_id = request_id.0,
46            terminal_id = terminal_id.value(),
47            "terminal started"
48        );
49        Ok(CommandDisposition::Accepted)
50    }
51
52    /// Stops a shell and forgets it, or rejects the command if it isn't running.
53    pub(super) fn close_terminal_command(
54        &self,
55        request_id: RequestId,
56        terminal_id: TerminalId,
57    ) -> Result<CommandDisposition, ApplicationError> {
58        {
59            let mut inner = self.lock_inner()?;
60            if inner
61                .workflows
62                .workflows
63                .iter()
64                .any(|workflow| workflow.terminal_ids().contains(&terminal_id))
65            {
66                return Ok(CommandDisposition::Rejected {
67                    code: "terminalOwnedByWorkflow".to_owned(),
68                    message: "Close the workflow to close its terminal.".to_owned(),
69                });
70            }
71            inner.terminals.remove(&terminal_id);
72        }
73        if let Err(error) = self.terminals.close(terminal_id) {
74            return Ok(rejection("terminalCloseFailed", &error));
75        }
76        let mut inner = self.lock_inner()?;
77        inner.events.append(EventKind::CommandCompleted {
78            request_id,
79            result: CommandResult::TerminalClosed { terminal_id },
80        })?;
81        debug!(
82            request_id = request_id.0,
83            terminal_id = terminal_id.value(),
84            "terminal closed"
85        );
86        Ok(CommandDisposition::Accepted)
87    }
88
89    /// Removes the next terminal chunk from the binary output queue.
90    ///
91    /// # Errors
92    ///
93    /// Returns an error if the queue cannot be accessed.
94    pub fn next_terminal_chunk(&self) -> Result<Option<TerminalChunk>, ApplicationError> {
95        Ok(self.terminal_output.next_chunk()?)
96    }
97
98    /// Reads at most `limit` raw bytes beginning exactly at `offset`, after committing output
99    /// already accepted for recording. `limit` must be between 1 and 64 KiB. An offset at the
100    /// recorded end returns an empty page; pruned history returns an explicit expired result.
101    ///
102    /// This performs storage work on the recording worker and waits for its reply. Call it off
103    /// the UI thread. It acquires no application-state or terminal-input locks.
104    ///
105    /// # Errors
106    ///
107    /// Returns an error for an invalid limit, an unknown terminal, an offset beyond the recorded
108    /// end, or unavailable/corrupt storage. Terminal IDs remain readable after terminal closure.
109    pub fn read_terminal_transcript(
110        &self,
111        terminal_id: TerminalId,
112        offset: u64,
113        limit: usize,
114    ) -> Result<TranscriptRead, ApplicationError> {
115        Ok(self
116            .terminal_output
117            .read_transcript(terminal_id, offset, limit)?)
118    }
119
120    /// Begins a transcript read without waiting for recording or storage. `None` means the bounded
121    /// request queue is full; callers may retry while continuing to service live terminals.
122    ///
123    /// # Errors
124    /// Returns an invalid limit, cancellation, or a recording failure.
125    pub fn request_terminal_transcript(
126        &self,
127        terminal_id: TerminalId,
128        offset: u64,
129        limit: usize,
130    ) -> Result<Option<crate::TranscriptRequest>, ApplicationError> {
131        Ok(self
132            .terminal_output
133            .request_transcript(terminal_id, offset, limit)?)
134    }
135
136    /// Writes binary user input to a live terminal.
137    ///
138    /// # Errors
139    ///
140    /// Returns an error if the terminal is not running or the PTY writer fails.
141    pub fn write_terminal_input(
142        &self,
143        terminal_id: TerminalId,
144        bytes: &[u8],
145    ) -> Result<(), ApplicationError> {
146        if let Some(input) = self.workflow_agent_input(terminal_id)? {
147            let mut input = input.lock().map_err(|_| ApplicationError::Poisoned)?;
148            let submission = if input.will_submit(bytes) {
149                self.note_workflow_submission(terminal_id)?
150            } else {
151                None
152            };
153            if let Err(error) = input.write(&self.terminals, terminal_id, bytes) {
154                self.cancel_workflow_submission(submission)?;
155                return Err(error.into());
156            }
157            return Ok(());
158        }
159        Ok(self.terminals.write_input(terminal_id, bytes)?)
160    }
161
162    /// Writes a terminal-generated protocol reply without treating it as a user draft.
163    ///
164    /// # Errors
165    /// Returns an error if the terminal is not running or the PTY writer fails.
166    pub fn write_terminal_response(
167        &self,
168        terminal_id: TerminalId,
169        bytes: &[u8],
170    ) -> Result<(), ApplicationError> {
171        Ok(self.terminals.write_input(terminal_id, bytes)?)
172    }
173
174    /// Changes the kernel PTY size for a live terminal.
175    ///
176    /// # Errors
177    ///
178    /// Returns an error if the size is invalid, the terminal is not running, or resizing fails.
179    pub fn resize_terminal(
180        &self,
181        terminal_id: TerminalId,
182        size: TerminalSize,
183    ) -> Result<(), ApplicationError> {
184        Ok(self.terminals.resize(terminal_id, size)?)
185    }
186
187    pub(super) fn start_terminal(
188        &self,
189        terminal_id: TerminalId,
190        working_directory: &Path,
191        size: TerminalSize,
192        integrate: bool,
193    ) -> Result<TerminalId, ApplicationError> {
194        Ok(self.terminals.start_default_shell(
195            terminal_id,
196            working_directory,
197            size,
198            Arc::new(self.exit_callback()),
199            integrate,
200        )?)
201    }
202
203    /// Records a terminal's process exit in application state.
204    pub(super) fn memory_exit_callback(
205        &self,
206        lease: Option<std::fs::File>,
207    ) -> impl Fn(TerminalId, Result<TerminalExit, String>, TerminalObservation) + Send + Sync + 'static
208    {
209        let callback = self.exit_callback();
210        let lease = std::sync::Mutex::new(lease);
211        move |id, result, observation| {
212            lease
213                .lock()
214                .unwrap_or_else(std::sync::PoisonError::into_inner)
215                .take();
216            callback(id, result, observation);
217        }
218    }
219    pub(super) fn exit_callback(
220        &self,
221    ) -> impl Fn(TerminalId, Result<TerminalExit, String>, TerminalObservation) + Send + Sync + 'static
222    {
223        let inner = Arc::downgrade(&self.inner);
224        let output = Arc::clone(&self.terminal_output);
225        move |terminal_id, result, observation| {
226            let Some(inner) = inner.upgrade() else { return };
227            let Ok(mut inner) = inner.lock() else {
228                warn!(
229                    terminal_id = terminal_id.value(),
230                    "application state lock poisoned after terminal exit"
231                );
232                return;
233            };
234            inner.record_shell_observations(Vec::new());
235            if inner.pending_shell_marks.is_empty()
236                && let Ok(observations) = output.take_shell_observations()
237            {
238                inner.record_shell_observations(observations);
239            }
240            inner.record_terminal_exit_at(terminal_id, result, observation);
241            inner.finish_shell_process_endings(&output);
242        }
243    }
244}
245
246impl Inner {
247    /// Records how a shell ended, in the terminal's status and as an event, and ends the workflow
248    /// that owns it once all of that workflow's shells have ended.
249    #[cfg(test)]
250    pub(super) fn record_terminal_exit(
251        &mut self,
252        terminal_id: TerminalId,
253        result: Result<TerminalExit, String>,
254    ) {
255        self.record_terminal_exit_at(
256            terminal_id,
257            result,
258            TerminalObservation {
259                integrated_shell: false,
260                observed_at: crate::workflow::timestamp(),
261                byte_offset: 0,
262                boundary_sizes: Some(Vec::new()),
263            },
264        );
265    }
266
267    pub(super) fn record_terminal_exit_at(
268        &mut self,
269        terminal_id: TerminalId,
270        result: Result<TerminalExit, String>,
271        observation: TerminalObservation,
272    ) {
273        if self.terminals.get(&terminal_id) != Some(&TerminalStatus::Running) {
274            return;
275        }
276        let (status, event) = match result {
277            Ok(exit) => (
278                TerminalStatus::Exited(exit.clone()),
279                StateEvent::TerminalExited { terminal_id, exit },
280            ),
281            Err(message) => (
282                TerminalStatus::Failed {
283                    message: message.clone(),
284                },
285                StateEvent::TerminalFailed {
286                    terminal_id,
287                    message,
288                },
289            ),
290        };
291        let (trace_status, trace_kind, trace_message) = match &status {
292            TerminalStatus::Exited(exit) => (
293                crate::TraceSpanStatus::Exited,
294                crate::TraceEventKind::ProcessExited,
295                match &exit.signal {
296                    Some(signal) => format!("Process exited with signal {signal}."),
297                    None => format!("Process exited with code {}.", exit.exit_code),
298                },
299            ),
300            TerminalStatus::Failed { message } => (
301                crate::TraceSpanStatus::Failed,
302                crate::TraceEventKind::ProcessFailed,
303                format!("Process failed: {message}"),
304            ),
305            TerminalStatus::Running => return,
306        };
307        let observed_at = observation.observed_at;
308        if !self.defer_shell_process_ending(
309            terminal_id,
310            &observation,
311            trace_status,
312            trace_kind,
313            &trace_message,
314        ) && let Err(error) = self.end_trace(
315            terminal_id,
316            observation,
317            trace_status,
318            trace_kind,
319            &trace_message,
320        ) {
321            warn!(%error, "failed to persist process trace");
322        }
323        let agent_status = match &status {
324            TerminalStatus::Failed { .. } => crate::RunAgentStatus::Failed,
325            _ => crate::RunAgentStatus::Exited,
326        };
327        self.terminals.insert(terminal_id, status);
328        self.record_run_agent_exit(terminal_id, agent_status);
329        if let Some(workflow) = self.workflows.workflows.iter_mut().find(|workflow| {
330            workflow.status == WorkflowStatus::Running
331                && workflow.terminal_ids().contains(&terminal_id)
332        }) && let Some(status) = ended_status(workflow, &self.terminals)
333        {
334            // Only a running workflow ends here, so a cancelled agent keeps its cancelled status. An
335            // exit only means the processes ended, never that the work succeeded.
336            workflow.status = status;
337            workflow.ended_at = Some(observed_at.max(workflow.started_at));
338            if workflow.kind == WorkflowKind::SingleAgent {
339                if let Err(error) = self
340                    .folders
341                    .store()
342                    .update_agent_status(workflow.workflow_id, workflow.status)
343                {
344                    warn!(%error, "failed to record the agent's exit");
345                }
346                tracing::info!(
347                    workflow_id = workflow.workflow_id.0,
348                    status = ?workflow.status,
349                    "agent exited"
350                );
351            }
352            let workflow = workflow.clone();
353            if let Err(error) = self
354                .events
355                .append(EventKind::State(StateEvent::WorkflowChanged(workflow)))
356            {
357                warn!(%error, "failed to record workflow exit");
358            }
359        }
360        if let Err(error) = self.events.append(EventKind::State(event)) {
361            warn!(terminal_id = terminal_id.value(), %error, "failed to record terminal exit");
362        }
363    }
364
365    fn record_run_agent_exit(
366        &mut self,
367        terminal_id: TerminalId,
368        agent_status: crate::RunAgentStatus,
369    ) {
370        if let Some(index) = self.workflows.workflows.iter().position(|workflow| {
371            workflow.run.is_some() && workflow.agents.iter().any(|a| a.terminal_id == terminal_id)
372        }) {
373            let mut workflow = self.workflows.workflows[index].clone();
374            let id = workflow
375                .agents
376                .iter()
377                .find(|a| a.terminal_id == terminal_id)
378                .expect("terminal belongs to workflow")
379                .agent_id
380                .0;
381            let run = workflow.run.as_mut().expect("workflow has a run");
382            if let Some(agent) = run.agents.iter_mut().find(|a| a.agent_id == id)
383                && agent.status == crate::RunAgentStatus::Running
384            {
385                agent.status = agent_status;
386                // Exiting never completes a role or advances the run.
387                if let Err(error) = self.publish_run_best_effort(index, workflow) {
388                    warn!(%error, "failed to record workflow agent exit");
389                }
390            }
391        }
392    }
393}
394
395/// A workflow runs while any shell it owns runs. Once all have ended it has failed if any failed or
396/// couldn't restart, and otherwise exited. A shell exiting never means the work succeeded.
397fn ended_status(
398    workflow: &Workflow,
399    terminals: &HashMap<TerminalId, TerminalStatus>,
400) -> Option<WorkflowStatus> {
401    if workflow.run.is_some() {
402        return None;
403    }
404    let mut failed = false;
405    for terminal_id in workflow.shells() {
406        match terminals.get(&terminal_id) {
407            Some(TerminalStatus::Running) => return None,
408            Some(TerminalStatus::Exited(_)) => {}
409            Some(TerminalStatus::Failed { .. }) | None => failed = true,
410        }
411    }
412    Some(if failed {
413        WorkflowStatus::Failed
414    } else {
415        WorkflowStatus::Exited
416    })
417}
418
419#[cfg(test)]
420mod tests {
421    use std::sync::mpsc;
422    use std::thread;
423    use std::time::{Duration, Instant};
424
425    use super::*;
426    use crate::Command;
427    use crate::config::Config;
428    use crate::workflow::{Agent, AgentId, SessionId, WorkflowId, WorkflowKind};
429
430    fn workflow(terminal_id: u64, agent_terminal_ids: &[u64]) -> Workflow {
431        Workflow {
432            workflow_id: WorkflowId(1),
433            session_id: SessionId(1),
434            name: "Agents".to_owned(),
435            kind: if agent_terminal_ids.is_empty() {
436                WorkflowKind::Terminal
437            } else {
438                WorkflowKind::Agents
439            },
440            harness: None,
441            terminal_id: TerminalId::from_value(terminal_id),
442            agents: agent_terminal_ids
443                .iter()
444                .zip(1..)
445                .map(|(terminal_id, agent_id)| Agent {
446                    agent_id: AgentId(agent_id),
447                    role: "Worker".to_owned(),
448                    terminal_id: TerminalId::from_value(*terminal_id),
449                })
450                .collect(),
451            status: WorkflowStatus::Running,
452            started_at: 0,
453            ended_at: None,
454            custom_name: None,
455            archived: false,
456            restored: false,
457            terminal_history: Vec::new(),
458            run: None,
459        }
460    }
461
462    #[test]
463    fn a_workflow_ends_after_all_of_its_shells_and_fails_if_any_did() {
464        let exited = TerminalStatus::Exited(TerminalExit {
465            exit_code: 0,
466            signal: None,
467        });
468        let failed = TerminalStatus::Failed {
469            message: "wait failed".to_owned(),
470        };
471        let terminals = HashMap::from([
472            (TerminalId::from_value(1), TerminalStatus::Running),
473            (TerminalId::from_value(2), exited.clone()),
474            (TerminalId::from_value(3), exited),
475            (TerminalId::from_value(4), failed),
476        ]);
477        assert_eq!(ended_status(&workflow(1, &[]), &terminals), None);
478        assert_eq!(
479            ended_status(&workflow(2, &[]), &terminals),
480            Some(WorkflowStatus::Exited)
481        );
482        assert_eq!(
483            ended_status(&workflow(4, &[]), &terminals),
484            Some(WorkflowStatus::Failed)
485        );
486        assert_eq!(ended_status(&workflow(0, &[2, 1]), &terminals), None);
487        assert_eq!(
488            ended_status(&workflow(0, &[2, 3]), &terminals),
489            Some(WorkflowStatus::Exited)
490        );
491        assert_eq!(
492            ended_status(&workflow(0, &[2, 4]), &terminals),
493            Some(WorkflowStatus::Failed)
494        );
495        // An agent whose shell couldn't restart has no terminal.
496        assert_eq!(
497            ended_status(&workflow(0, &[0, 3]), &terminals),
498            Some(WorkflowStatus::Failed)
499        );
500    }
501
502    fn size() -> TerminalSize {
503        TerminalSize {
504            rows: 24,
505            columns: 80,
506            pixel_width: 0,
507            pixel_height: 0,
508        }
509    }
510
511    fn start(app: &Application, folder: &Path, request: u64) -> TerminalId {
512        let receipt = app
513            .handle_command(
514                RequestId(request),
515                Command::StartTerminal {
516                    working_directory: folder.into(),
517                    size: size(),
518                },
519            )
520            .unwrap();
521        assert_eq!(receipt.disposition, CommandDisposition::Accepted);
522        app.snapshot()
523            .unwrap()
524            .terminals
525            .last()
526            .unwrap()
527            .terminal_id
528    }
529
530    #[test]
531    fn application_reads_closed_transcripts_after_relaunch_without_reusing_ids() {
532        let data = tempfile::tempdir().unwrap();
533        let folder = tempfile::tempdir().unwrap();
534        let app = Application::with_config(data.path(), Config::default()).unwrap();
535        let id = start(&app, folder.path(), 1);
536        let command = [
537            br"printf '\377\000\033[H__HANDSHAKE_HISTORY__\n'".as_slice(),
538            b"\n",
539        ]
540        .concat();
541        app.write_terminal_input(id, &command).unwrap();
542        let deadline = Instant::now() + Duration::from_secs(3);
543        let bytes = loop {
544            while app.next_terminal_chunk().unwrap().is_some() {}
545            let TranscriptRead::Output(page) =
546                app.read_terminal_transcript(id, 0, 64 * 1024).unwrap()
547            else {
548                panic!("new terminal history should be retained");
549            };
550            if page
551                .bytes
552                .windows(5)
553                .any(|bytes| bytes == [0xff, 0, 0x1b, b'[', b'H'])
554            {
555                break page.bytes;
556            }
557            assert!(
558                Instant::now() < deadline,
559                "shell did not produce binary history"
560            );
561            thread::sleep(Duration::from_millis(5));
562        };
563        app.handle_command(RequestId(2), Command::CloseTerminal { terminal_id: id })
564            .unwrap();
565        let expected = app
566            .read_terminal_transcript(id, 1, bytes.len() - 1)
567            .unwrap();
568        drop(app);
569        let app = Application::with_config(data.path(), Config::default()).unwrap();
570        assert_eq!(
571            app.read_terminal_transcript(id, 1, bytes.len() - 1)
572                .unwrap(),
573            expected
574        );
575        assert!(start(&app, folder.path(), 3).value() > id.value());
576    }
577
578    #[test]
579    fn terminal_and_workflow_allocation_wait_without_holding_application_state() {
580        for workflow in [
581            None,
582            Some(WorkflowKind::Terminal),
583            Some(WorkflowKind::Agents),
584        ] {
585            let data = tempfile::tempdir().unwrap();
586            let folder = tempfile::tempdir().unwrap();
587            let app = Arc::new(Application::with_config(data.path(), Config::default()).unwrap());
588            app.handle_command(
589                RequestId(1),
590                Command::OpenFolder {
591                    path: folder.path().into(),
592                },
593            )
594            .unwrap();
595            let id = start(&app, folder.path(), 2);
596            let stalled = app.terminal_output.stall_recording_worker(id);
597            let command = if let Some(kind) = workflow {
598                Command::CreateWorkflow {
599                    folder: folder.path().into(),
600                    session_id: None,
601                    kind,
602                    roles: if kind == WorkflowKind::Agents {
603                        vec!["Implementer".into(), "Reviewer".into()]
604                    } else {
605                        Vec::new()
606                    },
607                    size: size(),
608                }
609            } else {
610                Command::StartTerminal {
611                    working_directory: folder.path().into(),
612                    size: size(),
613                }
614            };
615            let (sent, result) = mpsc::sync_channel(1);
616            let starting = Arc::clone(&app);
617            let starter = thread::spawn(move || {
618                sent.send(starting.handle_command(RequestId(3), command))
619                    .unwrap();
620            });
621            assert_state_available_during_allocation(&app);
622            assert!(matches!(result.try_recv(), Err(mpsc::TryRecvError::Empty)));
623            drop(stalled);
624            assert_eq!(
625                result
626                    .recv_timeout(Duration::from_secs(3))
627                    .unwrap()
628                    .unwrap()
629                    .disposition,
630                CommandDisposition::Accepted
631            );
632            starter.join().unwrap();
633        }
634    }
635
636    #[test]
637    fn workflow_restoration_waits_without_holding_application_state() {
638        for kind in [WorkflowKind::Terminal, WorkflowKind::Agents] {
639            let data = tempfile::tempdir().unwrap();
640            let folder = tempfile::tempdir().unwrap();
641            let app = Arc::new(Application::with_config(data.path(), Config::default()).unwrap());
642            app.handle_command(
643                RequestId(1),
644                Command::OpenFolder {
645                    path: folder.path().into(),
646                },
647            )
648            .unwrap();
649            app.handle_command(
650                RequestId(2),
651                Command::CreateWorkflow {
652                    folder: folder.path().into(),
653                    session_id: None,
654                    kind,
655                    roles: if kind == WorkflowKind::Agents {
656                        vec!["Implementer".into(), "Reviewer".into()]
657                    } else {
658                        Vec::new()
659                    },
660                    size: size(),
661                },
662            )
663            .unwrap();
664            let id = app.snapshot().unwrap().workflows.workflows[0].terminal_ids()[0];
665            let stalled = app.terminal_output.stall_recording_worker(id);
666            let restoring = Arc::clone(&app);
667            let restore = thread::spawn(move || restoring.restore_workflows().unwrap());
668            assert_state_available_during_allocation(&app);
669            drop(stalled);
670            restore.join().unwrap();
671        }
672    }
673
674    fn assert_state_available_during_allocation(app: &Arc<Application>) {
675        let deadline = Instant::now() + Duration::from_secs(2);
676        while !app.terminal_output.allocation_is_pending() {
677            assert!(
678                Instant::now() < deadline,
679                "allocation did not reach the stalled worker"
680            );
681            thread::yield_now();
682        }
683        let (sent, received) = mpsc::sync_channel(1);
684        let observing = Arc::clone(app);
685        let observer = thread::spawn(move || {
686            let snapshot = observing.snapshot().unwrap();
687            observing.events_after(snapshot.sequence, 8).unwrap();
688            sent.send(()).unwrap();
689        });
690        received
691            .recv_timeout(Duration::from_secs(1))
692            .expect("state must remain available while storage is stalled");
693        observer.join().unwrap();
694    }
695}
handshake_core/application/terminals.rs

handshake_core/application/
terminals.rs
