twine_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 exit_callback(
205        &self,
206    ) -> impl Fn(TerminalId, Result<TerminalExit, String>, TerminalObservation) + Send + Sync + 'static
207    {
208        let inner = Arc::downgrade(&self.inner);
209        let output = Arc::clone(&self.terminal_output);
210        move |terminal_id, result, observation| {
211            let Some(inner) = inner.upgrade() else { return };
212            let Ok(mut inner) = inner.lock() else {
213                warn!(
214                    terminal_id = terminal_id.value(),
215                    "application state lock poisoned after terminal exit"
216                );
217                return;
218            };
219            inner.record_shell_observations(Vec::new());
220            if inner.pending_shell_marks.is_empty()
221                && let Ok(observations) = output.take_shell_observations()
222            {
223                inner.record_shell_observations(observations);
224            }
225            inner.record_terminal_exit_at(terminal_id, result, observation);
226            inner.finish_shell_process_endings(&output);
227        }
228    }
229}
230
231impl Inner {
232    /// Records how a shell ended, in the terminal's status and as an event, and ends the workflow
233    /// that owns it once all of that workflow's shells have ended.
234    #[cfg(test)]
235    pub(super) fn record_terminal_exit(
236        &mut self,
237        terminal_id: TerminalId,
238        result: Result<TerminalExit, String>,
239    ) {
240        self.record_terminal_exit_at(
241            terminal_id,
242            result,
243            TerminalObservation {
244                integrated_shell: false,
245                observed_at: crate::workflow::timestamp(),
246                byte_offset: 0,
247                boundary_sizes: Some(Vec::new()),
248            },
249        );
250    }
251
252    pub(super) fn record_terminal_exit_at(
253        &mut self,
254        terminal_id: TerminalId,
255        result: Result<TerminalExit, String>,
256        observation: TerminalObservation,
257    ) {
258        if self.terminals.get(&terminal_id) != Some(&TerminalStatus::Running) {
259            return;
260        }
261        let (status, event) = match result {
262            Ok(exit) => (
263                TerminalStatus::Exited(exit.clone()),
264                StateEvent::TerminalExited { terminal_id, exit },
265            ),
266            Err(message) => (
267                TerminalStatus::Failed {
268                    message: message.clone(),
269                },
270                StateEvent::TerminalFailed {
271                    terminal_id,
272                    message,
273                },
274            ),
275        };
276        let (trace_status, trace_kind, trace_message) = match &status {
277            TerminalStatus::Exited(exit) => (
278                crate::TraceSpanStatus::Exited,
279                crate::TraceEventKind::ProcessExited,
280                match &exit.signal {
281                    Some(signal) => format!("Process exited with signal {signal}."),
282                    None => format!("Process exited with code {}.", exit.exit_code),
283                },
284            ),
285            TerminalStatus::Failed { message } => (
286                crate::TraceSpanStatus::Failed,
287                crate::TraceEventKind::ProcessFailed,
288                format!("Process failed: {message}"),
289            ),
290            TerminalStatus::Running => return,
291        };
292        let observed_at = observation.observed_at;
293        if !self.defer_shell_process_ending(
294            terminal_id,
295            &observation,
296            trace_status,
297            trace_kind,
298            &trace_message,
299        ) && let Err(error) = self.end_trace(
300            terminal_id,
301            observation,
302            trace_status,
303            trace_kind,
304            &trace_message,
305        ) {
306            warn!(%error, "failed to persist process trace");
307        }
308        let agent_status = match &status {
309            TerminalStatus::Failed { .. } => crate::RunAgentStatus::Failed,
310            _ => crate::RunAgentStatus::Exited,
311        };
312        self.terminals.insert(terminal_id, status);
313        self.record_run_agent_exit(terminal_id, agent_status);
314        if let Some(workflow) = self.workflows.workflows.iter_mut().find(|workflow| {
315            workflow.status == WorkflowStatus::Running
316                && workflow.terminal_ids().contains(&terminal_id)
317        }) && let Some(status) = ended_status(workflow, &self.terminals)
318        {
319            // Only a running workflow ends here, so a cancelled agent keeps its cancelled status. An
320            // exit only means the processes ended, never that the work succeeded.
321            workflow.status = status;
322            workflow.ended_at = Some(observed_at.max(workflow.started_at));
323            if workflow.kind == WorkflowKind::SingleAgent {
324                if let Err(error) = self
325                    .folders
326                    .store()
327                    .update_agent_status(workflow.workflow_id, workflow.status)
328                {
329                    warn!(%error, "failed to record the agent's exit");
330                }
331                tracing::info!(
332                    workflow_id = workflow.workflow_id.0,
333                    status = ?workflow.status,
334                    "agent exited"
335                );
336            }
337            let workflow = workflow.clone();
338            if let Err(error) = self
339                .events
340                .append(EventKind::State(StateEvent::WorkflowChanged(workflow)))
341            {
342                warn!(%error, "failed to record workflow exit");
343            }
344        }
345        if let Err(error) = self.events.append(EventKind::State(event)) {
346            warn!(terminal_id = terminal_id.value(), %error, "failed to record terminal exit");
347        }
348    }
349
350    fn record_run_agent_exit(
351        &mut self,
352        terminal_id: TerminalId,
353        agent_status: crate::RunAgentStatus,
354    ) {
355        if let Some(index) = self.workflows.workflows.iter().position(|workflow| {
356            workflow.run.is_some() && workflow.agents.iter().any(|a| a.terminal_id == terminal_id)
357        }) {
358            let mut workflow = self.workflows.workflows[index].clone();
359            let id = workflow
360                .agents
361                .iter()
362                .find(|a| a.terminal_id == terminal_id)
363                .expect("terminal belongs to workflow")
364                .agent_id
365                .0;
366            let run = workflow.run.as_mut().expect("workflow has a run");
367            if let Some(agent) = run.agents.iter_mut().find(|a| a.agent_id == id)
368                && agent.status == crate::RunAgentStatus::Running
369            {
370                agent.status = agent_status;
371                // Exiting never completes a role or advances the run.
372                if let Err(error) = self.publish_run_best_effort(index, workflow) {
373                    warn!(%error, "failed to record workflow agent exit");
374                }
375            }
376        }
377    }
378}
379
380/// A workflow runs while any shell it owns runs. Once all have ended it has failed if any failed or
381/// couldn't restart, and otherwise exited. A shell exiting never means the work succeeded.
382fn ended_status(
383    workflow: &Workflow,
384    terminals: &HashMap<TerminalId, TerminalStatus>,
385) -> Option<WorkflowStatus> {
386    if workflow.run.is_some() {
387        return None;
388    }
389    let mut failed = false;
390    for terminal_id in workflow.shells() {
391        match terminals.get(&terminal_id) {
392            Some(TerminalStatus::Running) => return None,
393            Some(TerminalStatus::Exited(_)) => {}
394            Some(TerminalStatus::Failed { .. }) | None => failed = true,
395        }
396    }
397    Some(if failed {
398        WorkflowStatus::Failed
399    } else {
400        WorkflowStatus::Exited
401    })
402}
403
404#[cfg(test)]
405mod tests {
406    use std::sync::mpsc;
407    use std::thread;
408    use std::time::{Duration, Instant};
409
410    use super::*;
411    use crate::Command;
412    use crate::config::Config;
413    use crate::workflow::{Agent, AgentId, SessionId, WorkflowId, WorkflowKind};
414
415    fn workflow(terminal_id: u64, agent_terminal_ids: &[u64]) -> Workflow {
416        Workflow {
417            workflow_id: WorkflowId(1),
418            session_id: SessionId(1),
419            name: "Agents".to_owned(),
420            kind: if agent_terminal_ids.is_empty() {
421                WorkflowKind::Terminal
422            } else {
423                WorkflowKind::Agents
424            },
425            harness: None,
426            terminal_id: TerminalId::from_value(terminal_id),
427            agents: agent_terminal_ids
428                .iter()
429                .zip(1..)
430                .map(|(terminal_id, agent_id)| Agent {
431                    agent_id: AgentId(agent_id),
432                    role: "Worker".to_owned(),
433                    terminal_id: TerminalId::from_value(*terminal_id),
434                })
435                .collect(),
436            status: WorkflowStatus::Running,
437            started_at: 0,
438            ended_at: None,
439            restored: false,
440            terminal_history: Vec::new(),
441            run: None,
442        }
443    }
444
445    #[test]
446    fn a_workflow_ends_after_all_of_its_shells_and_fails_if_any_did() {
447        let exited = TerminalStatus::Exited(TerminalExit {
448            exit_code: 0,
449            signal: None,
450        });
451        let failed = TerminalStatus::Failed {
452            message: "wait failed".to_owned(),
453        };
454        let terminals = HashMap::from([
455            (TerminalId::from_value(1), TerminalStatus::Running),
456            (TerminalId::from_value(2), exited.clone()),
457            (TerminalId::from_value(3), exited),
458            (TerminalId::from_value(4), failed),
459        ]);
460        assert_eq!(ended_status(&workflow(1, &[]), &terminals), None);
461        assert_eq!(
462            ended_status(&workflow(2, &[]), &terminals),
463            Some(WorkflowStatus::Exited)
464        );
465        assert_eq!(
466            ended_status(&workflow(4, &[]), &terminals),
467            Some(WorkflowStatus::Failed)
468        );
469        assert_eq!(ended_status(&workflow(0, &[2, 1]), &terminals), None);
470        assert_eq!(
471            ended_status(&workflow(0, &[2, 3]), &terminals),
472            Some(WorkflowStatus::Exited)
473        );
474        assert_eq!(
475            ended_status(&workflow(0, &[2, 4]), &terminals),
476            Some(WorkflowStatus::Failed)
477        );
478        // An agent whose shell couldn't restart has no terminal.
479        assert_eq!(
480            ended_status(&workflow(0, &[0, 3]), &terminals),
481            Some(WorkflowStatus::Failed)
482        );
483    }
484
485    fn size() -> TerminalSize {
486        TerminalSize {
487            rows: 24,
488            columns: 80,
489            pixel_width: 0,
490            pixel_height: 0,
491        }
492    }
493
494    fn start(app: &Application, folder: &Path, request: u64) -> TerminalId {
495        let receipt = app
496            .handle_command(
497                RequestId(request),
498                Command::StartTerminal {
499                    working_directory: folder.into(),
500                    size: size(),
501                },
502            )
503            .unwrap();
504        assert_eq!(receipt.disposition, CommandDisposition::Accepted);
505        app.snapshot()
506            .unwrap()
507            .terminals
508            .last()
509            .unwrap()
510            .terminal_id
511    }
512
513    #[test]
514    fn application_reads_closed_transcripts_after_relaunch_without_reusing_ids() {
515        let data = tempfile::tempdir().unwrap();
516        let folder = tempfile::tempdir().unwrap();
517        let app = Application::with_config(data.path(), Config::default()).unwrap();
518        let id = start(&app, folder.path(), 1);
519        let command = [
520            br"printf '\377\000\033[H__TWINE_HISTORY__\n'".as_slice(),
521            b"\n",
522        ]
523        .concat();
524        app.write_terminal_input(id, &command).unwrap();
525        let deadline = Instant::now() + Duration::from_secs(3);
526        let bytes = loop {
527            while app.next_terminal_chunk().unwrap().is_some() {}
528            let TranscriptRead::Output(page) =
529                app.read_terminal_transcript(id, 0, 64 * 1024).unwrap()
530            else {
531                panic!("new terminal history should be retained");
532            };
533            if page
534                .bytes
535                .windows(5)
536                .any(|bytes| bytes == [0xff, 0, 0x1b, b'[', b'H'])
537            {
538                break page.bytes;
539            }
540            assert!(
541                Instant::now() < deadline,
542                "shell did not produce binary history"
543            );
544            thread::sleep(Duration::from_millis(5));
545        };
546        app.handle_command(RequestId(2), Command::CloseTerminal { terminal_id: id })
547            .unwrap();
548        let expected = app
549            .read_terminal_transcript(id, 1, bytes.len() - 1)
550            .unwrap();
551        drop(app);
552        let app = Application::with_config(data.path(), Config::default()).unwrap();
553        assert_eq!(
554            app.read_terminal_transcript(id, 1, bytes.len() - 1)
555                .unwrap(),
556            expected
557        );
558        assert!(start(&app, folder.path(), 3).value() > id.value());
559    }
560
561    #[test]
562    fn terminal_and_workflow_allocation_wait_without_holding_application_state() {
563        for workflow in [
564            None,
565            Some(WorkflowKind::Terminal),
566            Some(WorkflowKind::Agents),
567        ] {
568            let data = tempfile::tempdir().unwrap();
569            let folder = tempfile::tempdir().unwrap();
570            let app = Arc::new(Application::with_config(data.path(), Config::default()).unwrap());
571            app.handle_command(
572                RequestId(1),
573                Command::OpenFolder {
574                    path: folder.path().into(),
575                },
576            )
577            .unwrap();
578            let id = start(&app, folder.path(), 2);
579            let stalled = app.terminal_output.stall_recording_worker(id);
580            let command = if let Some(kind) = workflow {
581                Command::CreateWorkflow {
582                    folder: folder.path().into(),
583                    session_id: None,
584                    kind,
585                    roles: if kind == WorkflowKind::Agents {
586                        vec!["Implementer".into(), "Reviewer".into()]
587                    } else {
588                        Vec::new()
589                    },
590                    size: size(),
591                }
592            } else {
593                Command::StartTerminal {
594                    working_directory: folder.path().into(),
595                    size: size(),
596                }
597            };
598            let (sent, result) = mpsc::sync_channel(1);
599            let starting = Arc::clone(&app);
600            let starter = thread::spawn(move || {
601                sent.send(starting.handle_command(RequestId(3), command))
602                    .unwrap();
603            });
604            assert_state_available_during_allocation(&app);
605            assert!(matches!(result.try_recv(), Err(mpsc::TryRecvError::Empty)));
606            drop(stalled);
607            assert_eq!(
608                result
609                    .recv_timeout(Duration::from_secs(3))
610                    .unwrap()
611                    .unwrap()
612                    .disposition,
613                CommandDisposition::Accepted
614            );
615            starter.join().unwrap();
616        }
617    }
618
619    #[test]
620    fn workflow_restoration_waits_without_holding_application_state() {
621        for kind in [WorkflowKind::Terminal, WorkflowKind::Agents] {
622            let data = tempfile::tempdir().unwrap();
623            let folder = tempfile::tempdir().unwrap();
624            let app = Arc::new(Application::with_config(data.path(), Config::default()).unwrap());
625            app.handle_command(
626                RequestId(1),
627                Command::OpenFolder {
628                    path: folder.path().into(),
629                },
630            )
631            .unwrap();
632            app.handle_command(
633                RequestId(2),
634                Command::CreateWorkflow {
635                    folder: folder.path().into(),
636                    session_id: None,
637                    kind,
638                    roles: if kind == WorkflowKind::Agents {
639                        vec!["Implementer".into(), "Reviewer".into()]
640                    } else {
641                        Vec::new()
642                    },
643                    size: size(),
644                },
645            )
646            .unwrap();
647            let id = app.snapshot().unwrap().workflows.workflows[0].terminal_ids()[0];
648            let stalled = app.terminal_output.stall_recording_worker(id);
649            let restoring = Arc::clone(&app);
650            let restore = thread::spawn(move || restoring.restore_workflows().unwrap());
651            assert_state_available_during_allocation(&app);
652            drop(stalled);
653            restore.join().unwrap();
654        }
655    }
656
657    fn assert_state_available_during_allocation(app: &Arc<Application>) {
658        let deadline = Instant::now() + Duration::from_secs(2);
659        while !app.terminal_output.allocation_is_pending() {
660            assert!(
661                Instant::now() < deadline,
662                "allocation did not reach the stalled worker"
663            );
664            thread::yield_now();
665        }
666        let (sent, received) = mpsc::sync_channel(1);
667        let observing = Arc::clone(app);
668        let observer = thread::spawn(move || {
669            let snapshot = observing.snapshot().unwrap();
670            observing.events_after(snapshot.sequence, 8).unwrap();
671            sent.send(()).unwrap();
672        });
673        received
674            .recv_timeout(Duration::from_secs(1))
675            .expect("state must remain available while storage is stalled");
676        observer.join().unwrap();
677    }
678}
twine_core/application/terminals.rs

twine_core/application/
terminals.rs
