diff --git a/Cargo.lock b/Cargo.lock index da10786..d492f13 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1192,6 +1192,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs4" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c29c30684418547d476f0b48e84f4821639119c483b1eccd566c8cd0cd05f521" +dependencies = [ + "rustix", + "windows-sys 0.52.0", +] + [[package]] name = "fsevent-sys" version = "4.1.0" @@ -1970,6 +1980,7 @@ dependencies = [ "async-trait", "domain", "fastembed", + "fs4", "futures-util", "git2", "hex", @@ -2290,6 +2301,12 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "linux-raw-sys" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab" + [[package]] name = "litemap" version = "0.8.2" @@ -3526,6 +3543,19 @@ dependencies = [ "semver", ] +[[package]] +name = "rustix" +version = "0.38.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154" +dependencies = [ + "bitflags 2.12.1", + "errno", + "libc", + "linux-raw-sys", + "windows-sys 0.59.0", +] + [[package]] name = "rustls" version = "0.23.40" diff --git a/Cargo.toml b/Cargo.toml index b87fcda..3b16b95 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,6 +21,7 @@ serde_json = "1" thiserror = "2" async-trait = "0.1" futures-util = "0.3" +fs4 = "0.12" tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "fs", "io-util", "time"] } hex = "0.4" sha2 = "0.10" diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 469e797..d26340b 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -30,6 +30,7 @@ use application::{ UpdateProjectMcpToolPermissionsInput, UpdateProjectPermissionsInput, UpdateProjectSystemPermissionsInput, UpdateSkillInput, }; +use backend::stream::OutputSink; use domain::ports::ModelServerRuntime; use domain::ports::PtyHandle; @@ -38,9 +39,9 @@ use crate::dto::{ parse_layout_id, parse_memory_slug, parse_model_server_id, parse_node_id, parse_profile_id, parse_project_id, parse_session_id, parse_skill_id, parse_task_id, parse_template_id, parse_ticket_id, save_model_server_input, AgentDriftListDto, AgentDto, AgentListDto, - AppExitWorkGuardStateDto, AssignSkillRequestDto, AttachLiveAgentRequestDto, - AttachLiveAgentResponseDto, BackgroundTaskDto, ChangeAgentProfileDto, - ChangeAgentProfileRequestDto, CloneOpenCodeProfileFromSeedRequestDto, + AppExitWorkGuardStateDto, AssignSkillRequestDto, AttachBackgroundTaskResultDto, + AttachLiveAgentRequestDto, AttachLiveAgentResponseDto, BackgroundTaskDto, + ChangeAgentProfileDto, ChangeAgentProfileRequestDto, CloneOpenCodeProfileFromSeedRequestDto, CloneProfileFromSeedRequestDto, ConfigureProfilesRequestDto, ConversationDetailsDto, CreateAgentFromTemplateRequestDto, CreateAgentRequestDto, CreateLayoutRequestDto, CreateLayoutResultDto, CreateMemoryRequestDto, CreateProjectRequestDto, CreateSkillRequestDto, @@ -77,6 +78,7 @@ use crate::embedded_server::{ }; use crate::pty::{PtyBridge, PtyChunk}; use crate::state::{AppState, FocusedProjectDto}; +use crate::stream::TauriChannelSink; use domain::{DeviceId, SkillRef, SkillScope}; use uuid::Uuid; @@ -3745,6 +3747,54 @@ pub async fn retry_background_task( .map_err(ErrorDto::from) } +/// `attach_background_task` — attach a UI output subscriber to a command-backed +/// background task. +/// +/// If the task is still live, the response contains PTY scrollback and a fresh +/// live subscription is pumped to `on_output`. If it is already terminal, the +/// response contains only the persisted output tail and no PTY subscription is +/// attempted. +/// +/// # Errors +/// Returns an [`ErrorDto`] (`INVALID` for malformed id, `NOT_FOUND` if the task +/// has no live handle and no terminal result, `PROCESS`/`STORE` on backend +/// failure). +#[tauri::command] +pub async fn attach_background_task( + task_id: String, + on_output: Channel, + state: State<'_, AppState>, +) -> Result { + let id = parse_task_id(&task_id)?; + let plan = state + .background_task_attach_plan(id) + .await + .map_err(ErrorDto::from)?; + let live = plan.live_handle.is_some(); + + if let Some(handle) = plan.live_handle { + match state.pty_port.subscribe_output(&handle) { + Ok(stream) => { + std::thread::spawn(move || { + let sink = TauriChannelSink::new(on_output); + for chunk in stream { + if sink.send(chunk).is_err() { + break; + } + } + }); + } + Err(e) => return Err(ErrorDto::from(AppError::from(e))), + } + } + + Ok(AttachBackgroundTaskResultDto { + task_id: plan.task_id.to_string(), + scrollback: plan.scrollback, + live, + }) +} + /// `list_background_tasks` — read the background-task read-model for a project, /// optionally narrowed to one owning agent. /// diff --git a/crates/app-tauri/src/lib.rs b/crates/app-tauri/src/lib.rs index a11990f..2b4a230 100644 --- a/crates/app-tauri/src/lib.rs +++ b/crates/app-tauri/src/lib.rs @@ -353,6 +353,7 @@ pub fn run() { commands::spawn_background_command, commands::cancel_background_task, commands::retry_background_task, + commands::attach_background_task, commands::list_background_tasks, plugins::plugin_list_plugins, plugins::plugin_review_package, diff --git a/crates/backend/src/dto.rs b/crates/backend/src/dto.rs index ec07b76..d2a97a5 100644 --- a/crates/backend/src/dto.rs +++ b/crates/backend/src/dto.rs @@ -562,6 +562,20 @@ pub struct ReattachResultDto { pub scrollback: Vec, } +/// Response DTO for `attach_background_task`: the retained bytes to repaint for +/// a background command task before optional live bytes arrive on the channel. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AttachBackgroundTaskResultDto { + /// The task that was attached (echoed back for the frontend). + pub task_id: String, + /// Recent output bytes: PTY scrollback for a live task, persisted tail for a + /// terminal task. + pub scrollback: Vec, + /// Whether a live PTY subscription was installed for subsequent output. + pub live: bool, +} + /// Request DTO for `write_terminal`. #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase")] diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index b6bba99..f12d4cc 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -60,19 +60,19 @@ use domain::ports::{ EmbedderProfileStore, EmbedderPromptStore, EventBus, FileSystem, GitPort, IdGenerator, IssueNumberAllocator, IssueStore, McpToolPermissionStore, MemoryRecall, MemoryStore, ModelArtifactDownloader, PermissionStore, PluginManifestValidator, PluginMcpSupervisor, - PluginPackageStore, PluginRegistryStore, ProcessSpawner, ProfileStore, ProjectStore, PtyPort, - RuntimePermissionProbe, ScheduledTask, Scheduler, SecretStore, SkillStore, SprintStore, - StructuredSessionEnvironmentPreparer, SystemPermissionStore, TemplateStore, ToolInvoker, - WakeError, WakeReason, WindowStateStore, + PluginPackageStore, PluginRegistryStore, ProcessSpawner, ProfileStore, ProjectStore, PtyHandle, + PtyPort, RuntimePermissionProbe, ScheduledTask, Scheduler, SecretStore, SkillStore, + SprintStore, StructuredSessionEnvironmentPreparer, SystemPermissionStore, TemplateStore, + ToolInvoker, WakeError, WakeReason, WindowStateStore, }; use domain::profile::{ AgentProfile, ContextInjection, McpConfigStrategy, McpTransport, StructuredAdapter, }; use domain::remote::RemoteKind; use domain::{ - AgentId, AgentInbox, BackgroundTask, BackgroundTaskWakePolicy, DomainEvent, EmbedderProfile, - InboxError, InboxItem, InboxItemKind, InboxReceiptStatus, InboxSource, Project, ProjectId, - TaskId, TicketId, + AgentId, AgentInbox, BackgroundTask, BackgroundTaskResult, BackgroundTaskWakePolicy, + DomainEvent, EmbedderProfile, InboxError, InboxItem, InboxItemKind, InboxReceiptStatus, + InboxSource, Project, ProjectId, TaskId, TicketId, }; use serde_json::{json, Map, Value}; use uuid::Uuid; @@ -1099,6 +1099,8 @@ pub struct BackendCore { pub cancel_background_task: Arc, /// Retry a terminal command task under a fresh task id. pub retry_background_task: Arc, + /// Concrete command runner retained for infrastructure-only live PTY attach. + pub background_command_runner: Arc, /// Store handle used by `list_background_tasks` to read the task read-model. pub background_task_store: Arc, // --- Plugins (#43) --- @@ -1255,7 +1257,93 @@ pub struct BackendCore { pub template_tool_binder: Arc, } +/// Backend decision for attaching a UI subscriber to a background task. +pub struct BackgroundTaskAttachPlan { + /// Attached task id. + pub task_id: TaskId, + /// Bytes to repaint immediately. + pub scrollback: Vec, + /// Live PTY handle to subscribe to. `None` means the task is terminal and + /// the persisted tail is the whole attach payload. + pub live_handle: Option, +} + +fn map_background_task_port_error(err: BackgroundTaskPortError) -> AppError { + match err { + BackgroundTaskPortError::NotFound => AppError::NotFound("background task".to_owned()), + BackgroundTaskPortError::AlreadyExists => { + AppError::Invalid("background task already exists".to_owned()) + } + BackgroundTaskPortError::Invalid(msg) => AppError::Invalid(msg), + BackgroundTaskPortError::Runner(msg) => AppError::Process(msg), + BackgroundTaskPortError::Store(msg) => AppError::Store(msg), + } +} + +fn background_task_tail_bytes(task: &BackgroundTask) -> Vec { + match &task.result { + Some(BackgroundTaskResult::Success { + stdout_tail, + stderr_tail, + .. + }) + | Some(BackgroundTaskResult::Failure { + stdout_tail, + stderr_tail, + .. + }) => [stdout_tail.as_deref(), stderr_tail.as_deref()] + .into_iter() + .flatten() + .collect::>() + .join("") + .into_bytes(), + Some(BackgroundTaskResult::Cancelled { reason, .. }) + | Some(BackgroundTaskResult::Expired { reason, .. }) => reason.clone().into_bytes(), + None => Vec::new(), + } +} + impl BackendCore { + /// Builds an attach plan for a background task output view. + /// + /// A running task uses the runner's live PTY handle plus PTY scrollback. + /// A completed task intentionally returns only the persisted output tail and + /// no handle, so callers do not subscribe to a dead PTY. + /// + /// # Errors + /// Returns [`AppError::NotFound`] when the task is neither live in the + /// runner nor terminal in the store, and [`AppError::Process`] / + /// [`AppError::Store`] for PTY/store failures. + pub async fn background_task_attach_plan( + &self, + task_id: TaskId, + ) -> Result { + if let Some(handle) = self.background_command_runner.pty_handle_for(task_id) { + let scrollback = self.pty_port.scrollback(&handle).map_err(AppError::from)?; + return Ok(BackgroundTaskAttachPlan { + task_id, + scrollback, + live_handle: Some(handle), + }); + } + + let task = self + .background_task_store + .get(task_id) + .await + .map_err(map_background_task_port_error)? + .ok_or_else(|| AppError::NotFound("background task".to_owned()))?; + if task.is_terminal() { + return Ok(BackgroundTaskAttachPlan { + task_id, + scrollback: background_task_tail_bytes(&task), + live_handle: None, + }); + } + + Err(AppError::NotFound("background task live handle".to_owned())) + } + /// **Composition root.** Builds all adapters and use cases. /// /// `app_data_dir` is the machine-local IDE data directory (ARCHITECTURE @@ -2827,6 +2915,7 @@ impl BackendCore { spawn_background_command, cancel_background_task, retry_background_task, + background_command_runner: Arc::clone(&background_runner), background_task_store: Arc::clone(&background_tasks_port), review_plugin_package, install_plugin_from_archive, diff --git a/crates/infrastructure/Cargo.toml b/crates/infrastructure/Cargo.toml index 14008de..cb8ced4 100644 --- a/crates/infrastructure/Cargo.toml +++ b/crates/infrastructure/Cargo.toml @@ -17,6 +17,7 @@ tokio = { workspace = true, features = ["process", "time"] } uuid = { workspace = true } async-trait = { workspace = true } futures-util = { workspace = true } +fs4 = { workspace = true } # Ergonomic error enums for the MCP adapter (tool-mapping / transport errors). thiserror = { workspace = true } serde = { workspace = true } diff --git a/crates/infrastructure/src/background_task/runner.rs b/crates/infrastructure/src/background_task/runner.rs index d577c4a..6d6b368 100644 --- a/crates/infrastructure/src/background_task/runner.rs +++ b/crates/infrastructure/src/background_task/runner.rs @@ -101,6 +101,17 @@ impl CommandBackgroundRunner { u64::try_from(self.clock.now_millis().max(0)).unwrap_or(0) } + /// Returns the live PTY handle for a running task, when this runner still + /// owns one. + #[must_use] + pub fn pty_handle_for(&self, task_id: TaskId) -> Option { + self.running + .lock() + .expect("runner registry poisoned") + .get(&task_id) + .map(|control| control.pty_handle.clone()) + } + /// Detached worker driving one command to its single completion. #[allow(clippy::too_many_arguments)] async fn run_to_completion( diff --git a/crates/infrastructure/src/lib.rs b/crates/infrastructure/src/lib.rs index eb89443..98019ef 100644 --- a/crates/infrastructure/src/lib.rs +++ b/crates/infrastructure/src/lib.rs @@ -25,6 +25,7 @@ pub mod id; pub mod input; pub mod inspector; pub mod issues; +pub mod lock; pub mod mailbox; pub mod model_catalogue; pub mod model_server; @@ -68,6 +69,7 @@ pub use inspector::{ transcript_activity_token, ClaudeTranscriptInspector, ClaudeTranscriptTurnWatcher, }; pub use issues::{FsIssueNumberAllocator, FsIssueStore}; +pub use lock::{acquire_app_data_dir_lock, AppDataDirLock, AppDataDirLockError}; pub use mailbox::InMemoryMailbox; pub use model_catalogue::{ EmbeddedCompatibilityMatrix, HttpProviderModelCatalogue, ProcessCliVersionReader, diff --git a/crates/infrastructure/tests/background_task_runner.rs b/crates/infrastructure/tests/background_task_runner.rs index e60e1aa..32bd387 100644 --- a/crates/infrastructure/tests/background_task_runner.rs +++ b/crates/infrastructure/tests/background_task_runner.rs @@ -53,15 +53,6 @@ impl FakePty { self.state.lock().unwrap().scrollback = bytes.to_vec(); } - fn last_handle(&self) -> PtyHandle { - self.state - .lock() - .unwrap() - .handle - .clone() - .expect("spawned handle") - } - fn kill_count(&self) -> usize { self.state.lock().unwrap().kills } @@ -205,7 +196,9 @@ async fn ui_subscriber_does_not_interfere_with_runner_completion() { .spawn(spawn_spec(task_id, None)) .await .expect("spawn succeeds"); - let handle = pty.last_handle(); + let handle = runner + .pty_handle_for(task_id) + .expect("runner exposes live PTY handle"); let _ui_stream = pty.subscribe_output(&handle).expect("ui subscribes"); pty.complete(Some(0)); @@ -229,6 +222,68 @@ async fn ui_subscriber_does_not_interfere_with_runner_completion() { ); } +#[tokio::test] +async fn double_ui_subscriber_uses_independent_pty_streams() { + let pty = Arc::new(FakePty::default()); + let clock = Arc::new(FakeClock::default()); + let (runner, _completions) = runner_with(Arc::clone(&pty), clock); + let task_id = TaskId::from_uuid(id(14)); + + runner + .spawn(spawn_spec(task_id, None)) + .await + .expect("spawn succeeds"); + let handle = runner + .pty_handle_for(task_id) + .expect("runner exposes live PTY handle"); + + let _first = pty.subscribe_output(&handle).expect("first ui subscribes"); + let _second = pty.subscribe_output(&handle).expect("second ui subscribes"); + + assert_eq!( + pty.subscribe_count(), + 2, + "each UI attach must get its own PTY subscription" + ); + assert_eq!( + runner.pty_handle_for(task_id), + Some(handle), + "UI subscribers must not remove the runner's live handle" + ); +} + +#[tokio::test] +async fn completed_task_no_longer_exposes_live_pty_handle_for_attach() { + let pty = Arc::new(FakePty::default()); + pty.set_scrollback(b"final tail"); + let clock = Arc::new(FakeClock::default()); + let (runner, mut completions) = runner_with(Arc::clone(&pty), clock); + let task_id = TaskId::from_uuid(id(15)); + + runner + .spawn(spawn_spec(task_id, None)) + .await + .expect("spawn succeeds"); + assert!(runner.pty_handle_for(task_id).is_some()); + pty.complete(Some(0)); + + let completion = tokio::time::timeout( + Duration::from_secs(1), + tokio::task::spawn_blocking(move || completions.next().expect("completion")), + ) + .await + .expect("completion arrives") + .expect("completion thread joins"); + + assert_eq!(completion.task_id, task_id); + assert_eq!(runner.pty_handle_for(task_id), None); + assert_eq!( + pty.subscribe_count(), + 0, + "attach after completion must not subscribe to a dead PTY handle" + ); +} + #[tokio::test] async fn deadline_expires_and_kills_when_wait_never_resolves() { let pty = Arc::new(FakePty::default()); diff --git a/frontend/src/adapters/http/requestResponseGateways.ts b/frontend/src/adapters/http/requestResponseGateways.ts index 9382dbb..01c3ef4 100644 --- a/frontend/src/adapters/http/requestResponseGateways.ts +++ b/frontend/src/adapters/http/requestResponseGateways.ts @@ -71,8 +71,10 @@ import type { SkillGateway, TemplateGateway, WorkStateGateway, + BackgroundTaskAttachment, } from "@/ports"; import { normalizeProjectWorkState } from "../workStateNormalization"; +import { unsupportedOnWeb } from "./unsupported"; import { normalizeTurnPage } from "../conversationNormalization"; import { normalizeProfileModelCatalog } from "../profileCatalog"; import type { HttpInvoker } from "./httpInvoker"; @@ -443,6 +445,12 @@ export class HttpWorkStateGateway implements WorkStateGateway { const state = await this.http.invoke("get_project_work_state", { projectId }); return normalizeProjectWorkState(state); } + async attachBackgroundTask( + _taskId: string, + _onData: (bytes: Uint8Array) => void, + ): Promise { + return unsupportedOnWeb("Attaching to a background task live stream"); + } async cancelBackgroundTask(taskId: string): Promise { await this.http.invoke("cancel_background_task", { taskId }); } diff --git a/frontend/src/adapters/mock/index.ts b/frontend/src/adapters/mock/index.ts index 888b6d2..18b16da 100644 --- a/frontend/src/adapters/mock/index.ts +++ b/frontend/src/adapters/mock/index.ts @@ -124,6 +124,7 @@ import type { FocusedProject, FocusedProjectGateway, WorkStateGateway, + BackgroundTaskAttachment, } from "@/ports"; import { normalizeProjectWorkState } from "../workStateNormalization"; import { applyOperation, singleLeafTree } from "@/features/layout/layout"; @@ -2519,18 +2520,76 @@ export class MockPermissionGateway implements PermissionGateway { export class MockWorkStateGateway implements WorkStateGateway { private states = new Map(); + private backgroundTaskStreams = new Map< + string, + { + scrollback: Uint8Array; + live: boolean; + sinks: Set<(bytes: Uint8Array) => void>; + } + >(); /** Seeds the read-model returned for a project (deterministic tests/dev). */ _setProjectWorkState(projectId: string, state: unknown): void { this.states.set(projectId, normalizeProjectWorkState(state)); } + /** Seeds attach output for a background task (deterministic tests/dev). */ + _setBackgroundTaskAttachment( + taskId: string, + attachment: { scrollback?: string | Uint8Array; live?: boolean }, + ): void { + const encoder = new TextEncoder(); + const scrollback = + typeof attachment.scrollback === "string" + ? encoder.encode(attachment.scrollback) + : attachment.scrollback ?? new Uint8Array(); + this.backgroundTaskStreams.set(taskId, { + scrollback, + live: attachment.live ?? false, + sinks: new Set(), + }); + } + + /** Emits a live output chunk to current subscribers (deterministic tests/dev). */ + _emitBackgroundTaskOutput(taskId: string, bytes: string | Uint8Array): void { + const stream = this.backgroundTaskStreams.get(taskId); + if (!stream) return; + const encoder = new TextEncoder(); + const chunk = typeof bytes === "string" ? encoder.encode(bytes) : bytes; + for (const sink of stream.sinks) sink(chunk); + } + + _backgroundTaskSubscriberCount(taskId: string): number { + return this.backgroundTaskStreams.get(taskId)?.sinks.size ?? 0; + } + async getProjectWorkState(projectId: string): Promise { return structuredClone( this.states.get(projectId) ?? { agents: [], conversations: [] }, ); } + async attachBackgroundTask( + taskId: string, + onData: (bytes: Uint8Array) => void, + ): Promise { + const stream = this.backgroundTaskStreams.get(taskId) ?? { + scrollback: new Uint8Array(), + live: false, + sinks: new Set<(bytes: Uint8Array) => void>(), + }; + if (stream.live) stream.sinks.add(onData); + return { + taskId, + scrollback: Uint8Array.from(stream.scrollback), + live: stream.live, + detach: () => { + stream.sinks.delete(onData); + }, + }; + } + async cancelBackgroundTask(_taskId: string): Promise { // No-op in the mock; real refresh is driven by domain events. } diff --git a/frontend/src/adapters/workState.ts b/frontend/src/adapters/workState.ts index 9f523ca..75e2cc7 100644 --- a/frontend/src/adapters/workState.ts +++ b/frontend/src/adapters/workState.ts @@ -5,18 +5,44 @@ * command name; features consume the gateway port through DI. */ -import { invoke } from "@tauri-apps/api/core"; +import { Channel, invoke } from "@tauri-apps/api/core"; import type { ProjectWorkState } from "@/domain"; -import type { WorkStateGateway } from "@/ports"; +import type { BackgroundTaskAttachment, WorkStateGateway } from "@/ports"; import { normalizeProjectWorkState } from "./workStateNormalization"; +interface AttachBackgroundTaskResponse { + taskId: string; + scrollback: number[]; + live: boolean; +} + export class TauriWorkStateGateway implements WorkStateGateway { async getProjectWorkState(projectId: string): Promise { const state = await invoke("get_project_work_state", { projectId }); return normalizeProjectWorkState(state); } + async attachBackgroundTask( + taskId: string, + onData: (bytes: Uint8Array) => void, + ): Promise { + const channel = new Channel(); + channel.onmessage = (chunk) => onData(Uint8Array.from(chunk)); + const res = await invoke("attach_background_task", { + taskId, + onOutput: channel, + }); + return { + taskId: res.taskId, + scrollback: Uint8Array.from(res.scrollback), + live: res.live, + detach: () => { + channel.onmessage = () => {}; + }, + }; + } + async cancelBackgroundTask(taskId: string): Promise { await invoke("cancel_background_task", { taskId }); } diff --git a/frontend/src/features/projects/ProjectsView.ls7.test.tsx b/frontend/src/features/projects/ProjectsView.ls7.test.tsx index 331acd5..9965b22 100644 --- a/frontend/src/features/projects/ProjectsView.ls7.test.tsx +++ b/frontend/src/features/projects/ProjectsView.ls7.test.tsx @@ -66,6 +66,12 @@ function fixedWorkState(): WorkStateGateway { }; return { getProjectWorkState: async () => structuredClone(state), + attachBackgroundTask: async (taskId) => ({ + taskId, + scrollback: new Uint8Array(), + live: false, + detach: () => {}, + }), cancelBackgroundTask: async () => {}, retryBackgroundTask: async () => {}, }; diff --git a/frontend/src/features/workstate/ProjectWorkStatePanel.tsx b/frontend/src/features/workstate/ProjectWorkStatePanel.tsx index f9d8ecf..2d8e4ea 100644 --- a/frontend/src/features/workstate/ProjectWorkStatePanel.tsx +++ b/frontend/src/features/workstate/ProjectWorkStatePanel.tsx @@ -3,7 +3,7 @@ * and idle/busy state from the backend read-model, plus the current input queue. */ -import { Component, type ReactNode, useState } from "react"; +import { Component, type ReactNode, useEffect, useRef, useState } from "react"; import type { AgentTicketState, @@ -14,6 +14,7 @@ import type { InboxItem, LeafCell, } from "@/domain"; +import type { BackgroundTaskAttachment } from "@/ports"; import { useGateways } from "@/app/di"; import { leaves } from "@/features/layout/layout"; import { useLayout } from "@/features/layout/useLayout"; @@ -337,11 +338,36 @@ function BackgroundTaskRow({ }) { const { workState } = useGateways(); const [open, setOpen] = useState(false); + const [liveOpen, setLiveOpen] = useState(false); + const [liveOutput, setLiveOutput] = useState(""); const [actionBusy, setActionBusy] = useState(false); + const [liveBusy, setLiveBusy] = useState(false); const [message, setMessage] = useState(null); + const attachmentRef = useRef(null); + const attachSeqRef = useRef(0); const hasOutput = Boolean(task.stdoutTail || task.stderrTail); const canCancel = task.status === "running" || task.status === "pending"; const canRetry = task.status === "failed" || task.status === "cancelled"; + const canAttachLive = canCancel; + + function detachLive(): void { + attachSeqRef.current += 1; + attachmentRef.current?.detach(); + attachmentRef.current = null; + setLiveOpen(false); + } + + useEffect(() => { + return () => { + attachSeqRef.current += 1; + attachmentRef.current?.detach(); + attachmentRef.current = null; + }; + }, [task.taskId]); + + useEffect(() => { + if (!canAttachLive && attachmentRef.current) detachLive(); + }, [canAttachLive]); async function runAction( action: (taskId: string) => Promise, @@ -358,6 +384,36 @@ function BackgroundTaskRow({ } } + async function toggleLive(): Promise { + if (liveOpen || attachmentRef.current) { + detachLive(); + return; + } + + const seq = attachSeqRef.current + 1; + attachSeqRef.current = seq; + setLiveBusy(true); + setMessage(null); + setLiveOutput(""); + try { + const decoder = new TextDecoder(); + const attachment = await workState.attachBackgroundTask(task.taskId, (bytes) => { + setLiveOutput((prev) => prev + decoder.decode(bytes)); + }); + if (seq !== attachSeqRef.current) { + attachment.detach(); + return; + } + attachmentRef.current = attachment; + setLiveOutput(decoder.decode(attachment.scrollback)); + setLiveOpen(true); + } catch (e) { + setMessage(e instanceof Error ? e.message : String(e)); + } finally { + if (seq === attachSeqRef.current) setLiveBusy(false); + } + } + return (
  • @@ -386,6 +442,15 @@ function BackgroundTaskRow({ > Cancel +
    • ); } diff --git a/frontend/src/features/workstate/workstate.test.tsx b/frontend/src/features/workstate/workstate.test.tsx index 0773008..8663ebf 100644 --- a/frontend/src/features/workstate/workstate.test.tsx +++ b/frontend/src/features/workstate/workstate.test.tsx @@ -336,6 +336,64 @@ describe("ProjectWorkStatePanel", () => { ).toBeNull(); }); + it("attaches, repaints, streams, and detaches live background task output", async () => { + const workState = new MockWorkStateGateway(); + workState._setProjectWorkState(PROJECT_ID, { + agents: [ + { + agentId: "agent-live-task", + name: "Runner", + profileId: "codex", + busy: { state: "idle" }, + backgroundTasks: [ + { + taskId: "task-live-output", + kind: "command", + state: "running", + createdAtMs: 10, + updatedAtMs: 10, + }, + ], + }, + ], + }); + workState._setBackgroundTaskAttachment("task-live-output", { + scrollback: "already printed\n", + live: true, + }); + const attachSpy = vi.spyOn(workState, "attachBackgroundTask"); + + const view = renderPanel(workState); + const list = await screen.findByLabelText("Runner background tasks"); + + fireEvent.click(within(list).getByRole("button", { name: "Live" })); + + expect(attachSpy).toHaveBeenCalledWith("task-live-output", expect.any(Function)); + expect( + (await within(list).findByLabelText("task task-liv live output")).textContent, + ).toContain("already printed"); + expect(workState._backgroundTaskSubscriberCount("task-live-output")).toBe(1); + + workState._emitBackgroundTaskOutput("task-live-output", "new line\n"); + await waitFor(() => + expect( + within(list).getByLabelText("task task-liv live output").textContent, + ).toContain("new line"), + ); + + fireEvent.click(within(list).getByRole("button", { name: "Detach" })); + await waitFor(() => + expect(workState._backgroundTaskSubscriberCount("task-live-output")).toBe(0), + ); + expect(within(list).queryByLabelText("task task-liv live output")).toBeNull(); + + fireEvent.click(within(list).getByRole("button", { name: "Live" })); + await within(list).findByLabelText("task task-liv live output"); + expect(workState._backgroundTaskSubscriberCount("task-live-output")).toBe(1); + view.unmount(); + expect(workState._backgroundTaskSubscriberCount("task-live-output")).toBe(0); + }); + it("renders a legacy agent without tickets", async () => { const workState = new MockWorkStateGateway(); workState._setProjectWorkState(PROJECT_ID, { diff --git a/frontend/src/ports/index.ts b/frontend/src/ports/index.ts index f9f3f14..d3e673f 100644 --- a/frontend/src/ports/index.ts +++ b/frontend/src/ports/index.ts @@ -388,6 +388,18 @@ export interface ReattachResult { scrollback: Uint8Array; } +/** A UI subscription attached to one background task output stream. */ +export interface BackgroundTaskAttachment { + /** Attached task id, echoed by the backend. */ + taskId: string; + /** Retained bytes to repaint immediately before live chunks arrive. */ + scrollback: Uint8Array; + /** Whether subsequent live bytes are expected on the supplied callback. */ + live: boolean; + /** Detaches the local UI subscriber without cancelling the task. */ + detach(): void; +} + /** Projects: create/open/close/list (L2). */ export interface ProjectGateway { /** Lists the projects known to the registry. */ @@ -932,6 +944,15 @@ export interface PermissionGateway { export interface WorkStateGateway { /** Reads the current per-agent live/offline and idle/busy state for a project. */ getProjectWorkState(projectId: string): Promise; + /** + * Attaches a UI output subscriber to a background task. Running tasks replay + * PTY scrollback and then stream live chunks; terminal tasks return only their + * persisted output tail. + */ + attachBackgroundTask( + taskId: string, + onData: (bytes: Uint8Array) => void, + ): Promise; /** * Cancels a running/pending background task by id. The read-model refreshes * through the `backgroundTaskChanged` domain event.