feat(background-task): rendu live des tâches de fond — subscriber UI + canal IPC (#58)

Câble un canal attachable au flux de sortie d'une tâche de fond (runner
infrastructure + commande app-tauri + port/adaptateurs frontend) et le
panneau ProjectWorkStatePanel s'y abonne pour un rendu live au lieu d'un
état figé au dernier snapshot.

Validations obtenues avant commit :
- cargo test -p infrastructure --test background_task_runner : vert
- cargo check -p backend -p app-tauri : vert
- npx vitest run src/features/workstate/workstate.test.tsx : vert
- npm run typecheck : vert
- verdict QA #58 : vert

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-27 16:48:29 +02:00
parent 6f532d7434
commit b734c237d3
17 changed files with 528 additions and 23 deletions

View File

@ -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<PtyChunk>,
state: State<'_, AppState>,
) -> Result<AttachBackgroundTaskResultDto, ErrorDto> {
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.
///

View File

@ -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,

View File

@ -562,6 +562,20 @@ pub struct ReattachResultDto {
pub scrollback: Vec<u8>,
}
/// 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<u8>,
/// 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")]

View File

@ -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<CancelBackgroundTask>,
/// Retry a terminal command task under a fresh task id.
pub retry_background_task: Arc<RetryBackgroundTask>,
/// Concrete command runner retained for infrastructure-only live PTY attach.
pub background_command_runner: Arc<CommandBackgroundRunner>,
/// Store handle used by `list_background_tasks` to read the task read-model.
pub background_task_store: Arc<dyn BackgroundTaskStore>,
// --- Plugins (#43) ---
@ -1255,7 +1257,93 @@ pub struct BackendCore {
pub template_tool_binder: Arc<LateBoundTemplateToolProvider>,
}
/// 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<u8>,
/// 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<PtyHandle>,
}
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<u8> {
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::<Vec<_>>()
.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<BackgroundTaskAttachPlan, AppError> {
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,

View File

@ -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 }

View File

@ -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<PtyHandle> {
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(

View File

@ -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,

View File

@ -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());