feat(background): runner PTY B8 + boucle sink fermée + refactor point-2
Livre le lot backend B8 des tâches de fond : - infrastructure : runner de commandes concret (CommandBackgroundRunner) sur le port BackgroundTaskRunner, tail borné (bounded_tail/BoundedTail), éclatement du module background_task en sous-modules (mod/runner/tail, sink extrait de l'ancien background_task.rs). - application : nouveau module background exposant les cas d'usage SpawnBackgroundCommand, CancelBackgroundTask, RetryBackgroundTask et le port BackgroundCommandArchive. - domain : refactor point-2 de l'arbitrage Architect — sortie du trait BackgroundCommandArchive de la couche domaine vers application. - app-tauri : câblage runtime (commands, dto, state, lib) des commandes spawn/cancel/retry et de la boucle de complétion sink fermée en composition root. Build workspace + tests application/infrastructure verts (QA). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -30,7 +30,8 @@ use domain::ports::PtyHandle;
|
||||
use crate::dto::{
|
||||
parse_agent_id, parse_close_terminal, parse_delete_profile, parse_layout_id, parse_memory_slug,
|
||||
parse_node_id, parse_profile_id, parse_project_id, parse_session_id, parse_skill_id,
|
||||
parse_template_id, parse_ticket_id, AgentDriftListDto, AgentDto, AgentListDto,
|
||||
parse_task_id, parse_template_id, parse_ticket_id, AgentDriftListDto, AgentDto, AgentListDto,
|
||||
BackgroundTaskDto,
|
||||
AssignSkillRequestDto, AttachLiveAgentRequestDto, AttachLiveAgentResponseDto,
|
||||
ChangeAgentProfileDto, ChangeAgentProfileRequestDto, ConfigureProfilesRequestDto,
|
||||
ConversationDetailsDto, CreateAgentFromTemplateRequestDto, CreateAgentRequestDto,
|
||||
@ -2555,3 +2556,160 @@ pub async fn resolve_memory_links(
|
||||
.map(MemoryLinksDto::from)
|
||||
.map_err(ErrorDto::from)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Background tasks (B8 — first-class background command)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Maps a background-task port error to the frontend [`ErrorDto`] shape.
|
||||
fn background_error(err: domain::ports::BackgroundTaskPortError) -> ErrorDto {
|
||||
use domain::ports::BackgroundTaskPortError as E;
|
||||
let code = match &err {
|
||||
E::NotFound => "NOT_FOUND",
|
||||
E::AlreadyExists | E::Invalid(_) => "INVALID",
|
||||
E::Runner(_) => "PROCESS",
|
||||
E::Store(_) => "STORE",
|
||||
};
|
||||
ErrorDto {
|
||||
code: code.to_owned(),
|
||||
message: err.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// `spawn_background_command` — start a command-backed first-class background
|
||||
/// task whose completion is delivered to the owning agent's inbox.
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns an [`ErrorDto`] (`INVALID` for a malformed id/path, `PROCESS` if the
|
||||
/// command cannot be spawned, `STORE` on persistence failure).
|
||||
#[tauri::command]
|
||||
pub async fn spawn_background_command(
|
||||
request: crate::dto::SpawnBackgroundCommandRequestDto,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<BackgroundTaskDto, ErrorDto> {
|
||||
let project_id = parse_project_id(&request.project_id)?;
|
||||
let owner_agent_id = parse_agent_id(&request.owner_agent_id)?;
|
||||
let cwd = domain::project::ProjectPath::new(request.cwd.clone()).map_err(|_| ErrorDto {
|
||||
code: "INVALID".to_owned(),
|
||||
message: format!("invalid working directory: {}", request.cwd),
|
||||
})?;
|
||||
let command = domain::ports::SpawnSpec {
|
||||
command: request.command,
|
||||
args: request.args,
|
||||
cwd,
|
||||
env: request.env,
|
||||
context_plan: None,
|
||||
sandbox: None,
|
||||
};
|
||||
let wake_policy = if request.record_only {
|
||||
domain::BackgroundTaskWakePolicy::RecordOnly
|
||||
} else {
|
||||
domain::BackgroundTaskWakePolicy::WakeOwner
|
||||
};
|
||||
state
|
||||
.spawn_background_command
|
||||
.execute(application::SpawnBackgroundCommandInput {
|
||||
project_id,
|
||||
owner_agent_id,
|
||||
label: request.label,
|
||||
command,
|
||||
wake_policy,
|
||||
deadline_ms: request.deadline_ms,
|
||||
})
|
||||
.await
|
||||
.map(|out| BackgroundTaskDto::from(out.task))
|
||||
.map_err(ErrorDto::from)
|
||||
}
|
||||
|
||||
/// `cancel_background_task` — request cancellation of a running background task.
|
||||
///
|
||||
/// The terminal `Cancelled` state is written by the completion sink; this returns
|
||||
/// the task as currently persisted (absent if it no longer exists).
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns an [`ErrorDto`] (`INVALID` for a malformed id, `PROCESS`/`STORE` on
|
||||
/// failure).
|
||||
#[tauri::command]
|
||||
pub async fn cancel_background_task(
|
||||
task_id: String,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<Option<BackgroundTaskDto>, ErrorDto> {
|
||||
let id = parse_task_id(&task_id)?;
|
||||
state
|
||||
.cancel_background_task
|
||||
.execute(id)
|
||||
.await
|
||||
.map(|out| out.task.map(BackgroundTaskDto::from))
|
||||
.map_err(ErrorDto::from)
|
||||
}
|
||||
|
||||
/// `retry_background_task` — re-run a terminal command task under a **new** task
|
||||
/// id (the original id is never reused).
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns an [`ErrorDto`] (`INVALID` for a malformed id or a non-retryable task,
|
||||
/// `NOT_FOUND` if the task is unknown, `PROCESS`/`STORE` on failure).
|
||||
#[tauri::command]
|
||||
pub async fn retry_background_task(
|
||||
task_id: String,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<BackgroundTaskDto, ErrorDto> {
|
||||
let id = parse_task_id(&task_id)?;
|
||||
state
|
||||
.retry_background_task
|
||||
.execute(id)
|
||||
.await
|
||||
.map(|out| BackgroundTaskDto::from(out.task))
|
||||
.map_err(ErrorDto::from)
|
||||
}
|
||||
|
||||
/// `list_background_tasks` — read the background-task read-model for a project,
|
||||
/// optionally narrowed to one owning agent.
|
||||
///
|
||||
/// Returns the union of the agent's open tasks and undelivered completions,
|
||||
/// filtered to `projectId`. Without `agentId`, only the project's undelivered
|
||||
/// completions are returned (the store cannot enumerate open tasks project-wide).
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns an [`ErrorDto`] (`INVALID` for a malformed id, `STORE` on failure).
|
||||
#[tauri::command]
|
||||
pub async fn list_background_tasks(
|
||||
project_id: String,
|
||||
agent_id: Option<String>,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<Vec<BackgroundTaskDto>, ErrorDto> {
|
||||
let project_id = parse_project_id(&project_id)?;
|
||||
let agent_id = match &agent_id {
|
||||
Some(raw) => Some(parse_agent_id(raw)?),
|
||||
None => None,
|
||||
};
|
||||
|
||||
let store = &state.background_task_store;
|
||||
let mut by_id: std::collections::HashMap<domain::TaskId, domain::BackgroundTask> =
|
||||
std::collections::HashMap::new();
|
||||
|
||||
if let Some(agent_id) = agent_id {
|
||||
for task in store
|
||||
.list_open_for_agent(agent_id)
|
||||
.await
|
||||
.map_err(background_error)?
|
||||
{
|
||||
by_id.insert(task.id, task);
|
||||
}
|
||||
}
|
||||
for task in store
|
||||
.list_undelivered_completions()
|
||||
.await
|
||||
.map_err(background_error)?
|
||||
{
|
||||
by_id.entry(task.id).or_insert(task);
|
||||
}
|
||||
|
||||
let mut tasks: Vec<domain::BackgroundTask> = by_id
|
||||
.into_values()
|
||||
.filter(|task| task.project_id == project_id)
|
||||
.filter(|task| agent_id.map_or(true, |a| task.owner_agent_id == a))
|
||||
.collect();
|
||||
tasks.sort_by_key(|task| (task.created_at_ms, task.id));
|
||||
Ok(tasks.into_iter().map(BackgroundTaskDto::from).collect())
|
||||
}
|
||||
|
||||
@ -2866,3 +2866,167 @@ mod ls6_pagination_tests {
|
||||
assert!(dto.turns.is_empty() && !dto.has_more && dto.next_anchor.is_none());
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Background tasks (B8 — first-class background command)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
use domain::{
|
||||
BackgroundTask, BackgroundTaskKind, BackgroundTaskResult, BackgroundTaskState, TaskId,
|
||||
};
|
||||
|
||||
/// One first-class background task as seen by the frontend (camelCase wire shape).
|
||||
///
|
||||
/// Mirrors the persisted [`BackgroundTask`], flattening the terminal
|
||||
/// [`BackgroundTaskResult`] into optional `exitCode` / `summary` / tails so the
|
||||
/// UI has a single shape for every lifecycle state. Optional fields are omitted
|
||||
/// from the wire when absent (`skip_serializing_if`).
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct BackgroundTaskDto {
|
||||
/// Stable task id (UUID string).
|
||||
pub task_id: String,
|
||||
/// Owning agent id (UUID string).
|
||||
pub owner_agent_id: String,
|
||||
/// Owning project id (UUID string).
|
||||
pub project_id: String,
|
||||
/// Kind discriminant (`"command"`, `"headlessRendezvous"`, `"sessionResume"`,
|
||||
/// `"maintenance"`).
|
||||
pub kind: String,
|
||||
/// Lifecycle state (`"queued"`, `"running"`, `"waiting"`, `"completed"`,
|
||||
/// `"failed"`, `"cancelled"`, `"expired"`).
|
||||
pub state: String,
|
||||
/// Process exit code, when the terminal result carries one.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub exit_code: Option<i32>,
|
||||
/// Human-readable summary / error / reason of the terminal result.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub summary: Option<String>,
|
||||
/// Bounded stdout tail (merged output for PTY-backed commands).
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub stdout_tail: Option<String>,
|
||||
/// Bounded stderr tail (unset for PTY-backed commands, which merge streams).
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub stderr_tail: Option<String>,
|
||||
/// Creation timestamp, epoch milliseconds.
|
||||
pub created_at_ms: u64,
|
||||
/// Last update timestamp, epoch milliseconds.
|
||||
pub updated_at_ms: u64,
|
||||
}
|
||||
|
||||
/// Kind discriminant string for a [`BackgroundTaskKind`].
|
||||
fn background_kind_label(kind: &BackgroundTaskKind) -> &'static str {
|
||||
match kind {
|
||||
BackgroundTaskKind::Command { .. } => "command",
|
||||
BackgroundTaskKind::HeadlessRendezvous { .. } => "headlessRendezvous",
|
||||
BackgroundTaskKind::SessionResume { .. } => "sessionResume",
|
||||
BackgroundTaskKind::Maintenance { .. } => "maintenance",
|
||||
}
|
||||
}
|
||||
|
||||
/// Lifecycle state string for a [`BackgroundTaskState`].
|
||||
fn background_state_label(state: BackgroundTaskState) -> &'static str {
|
||||
match state {
|
||||
BackgroundTaskState::Queued => "queued",
|
||||
BackgroundTaskState::Running => "running",
|
||||
BackgroundTaskState::Waiting => "waiting",
|
||||
BackgroundTaskState::Completed => "completed",
|
||||
BackgroundTaskState::Failed => "failed",
|
||||
BackgroundTaskState::Cancelled => "cancelled",
|
||||
BackgroundTaskState::Expired => "expired",
|
||||
}
|
||||
}
|
||||
|
||||
impl From<BackgroundTask> for BackgroundTaskDto {
|
||||
fn from(task: BackgroundTask) -> Self {
|
||||
let (exit_code, summary, stdout_tail, stderr_tail) = match &task.result {
|
||||
Some(BackgroundTaskResult::Success {
|
||||
exit_code,
|
||||
summary,
|
||||
stdout_tail,
|
||||
stderr_tail,
|
||||
..
|
||||
}) => (
|
||||
*exit_code,
|
||||
Some(summary.clone()),
|
||||
stdout_tail.clone(),
|
||||
stderr_tail.clone(),
|
||||
),
|
||||
Some(BackgroundTaskResult::Failure {
|
||||
exit_code,
|
||||
error,
|
||||
stdout_tail,
|
||||
stderr_tail,
|
||||
..
|
||||
}) => (
|
||||
*exit_code,
|
||||
Some(error.clone()),
|
||||
stdout_tail.clone(),
|
||||
stderr_tail.clone(),
|
||||
),
|
||||
Some(BackgroundTaskResult::Cancelled { reason, .. })
|
||||
| Some(BackgroundTaskResult::Expired { reason, .. }) => {
|
||||
(None, Some(reason.clone()), None, None)
|
||||
}
|
||||
None => (None, None, None, None),
|
||||
};
|
||||
Self {
|
||||
task_id: task.id.to_string(),
|
||||
owner_agent_id: task.owner_agent_id.to_string(),
|
||||
project_id: task.project_id.to_string(),
|
||||
kind: background_kind_label(&task.kind).to_owned(),
|
||||
state: background_state_label(task.state).to_owned(),
|
||||
exit_code,
|
||||
summary,
|
||||
stdout_tail,
|
||||
stderr_tail,
|
||||
created_at_ms: task.created_at_ms,
|
||||
updated_at_ms: task.updated_at_ms,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Parses a task-id string (UUID) coming from the frontend.
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns an [`ErrorDto`] with code `INVALID` if the string is not a UUID.
|
||||
pub fn parse_task_id(raw: &str) -> Result<TaskId, ErrorDto> {
|
||||
uuid::Uuid::parse_str(raw)
|
||||
.map(TaskId::from_uuid)
|
||||
.map_err(|_| ErrorDto {
|
||||
code: "INVALID".to_owned(),
|
||||
message: format!("invalid task id: {raw}"),
|
||||
})
|
||||
}
|
||||
|
||||
/// Request DTO for `spawn_background_command`.
|
||||
///
|
||||
/// Builds a command-backed [`BackgroundTask`]: the command line runs under the
|
||||
/// project host's PTY and its completion is delivered to `ownerAgentId`.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct SpawnBackgroundCommandRequestDto {
|
||||
/// Owning project id (UUID string).
|
||||
pub project_id: String,
|
||||
/// Agent that owns completion delivery (UUID string).
|
||||
pub owner_agent_id: String,
|
||||
/// Human-facing label.
|
||||
pub label: String,
|
||||
/// Executable to run.
|
||||
pub command: String,
|
||||
/// Arguments.
|
||||
#[serde(default)]
|
||||
pub args: Vec<String>,
|
||||
/// Working directory (absolute path).
|
||||
pub cwd: String,
|
||||
/// Extra environment variables.
|
||||
#[serde(default)]
|
||||
pub env: Vec<(String, String)>,
|
||||
/// When `true` (the default), wake the owner on completion; otherwise only
|
||||
/// record it.
|
||||
#[serde(default)]
|
||||
pub record_only: bool,
|
||||
/// Optional absolute deadline, epoch milliseconds.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub deadline_ms: Option<u64>,
|
||||
}
|
||||
|
||||
@ -214,6 +214,10 @@ pub fn run() {
|
||||
commands::recall_memory,
|
||||
commands::resolve_memory_links,
|
||||
commands::move_tab_to_new_window,
|
||||
commands::spawn_background_command,
|
||||
commands::cancel_background_task,
|
||||
commands::retry_background_task,
|
||||
commands::list_background_tasks,
|
||||
])
|
||||
.run(tauri::generate_context!())
|
||||
.expect("error while running IdeA Tauri application");
|
||||
|
||||
@ -13,7 +13,8 @@ use std::sync::{Arc, Mutex};
|
||||
|
||||
use application::{
|
||||
AgentResumer, AgentWakeService, AppError, AssignSkillToAgent, AttachLiveAgent,
|
||||
ChangeAgentProfile, CheckEmbedderSuggestion, CloseProject, CloseTab, CloseTerminal,
|
||||
BackgroundCommandArchive, CancelBackgroundTask, ChangeAgentProfile, CheckEmbedderSuggestion,
|
||||
CloseProject, CloseTab, CloseTerminal, RetryBackgroundTask, SpawnBackgroundCommand,
|
||||
ConfigureProfiles, ContextGuardUseCases, CreateAgentFromScratch, CreateAgentFromTemplate,
|
||||
CreateLayout, CreateMemory, CreateProject, CreateSkill, CreateTemplate, DeleteAgent,
|
||||
DeleteEmbedderProfile, DeleteLayout, DeleteMemory, DeleteProfile, DeleteSkill, DeleteTemplate,
|
||||
@ -39,7 +40,8 @@ use application::{
|
||||
use async_trait::async_trait;
|
||||
use domain::ports::{
|
||||
AgentContextStore, AgentRuntime, AgentSession, AgentSessionFactory, AgentWakePort,
|
||||
BackgroundTaskPortError, BackgroundTaskStore, Clock, Embedder, EmbedderEnvInspector,
|
||||
BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskStore,
|
||||
Clock, Embedder, EmbedderEnvInspector,
|
||||
EmbedderProfileStore, EmbedderPromptStore, EventBus, FileSystem, GitPort, IdGenerator,
|
||||
MemoryRecall, MemoryStore, PermissionStore, ProcessSpawner, ProfileStore, ProjectStore,
|
||||
PtyPort, ScheduledTask, Scheduler, SkillStore, TemplateStore, WakeError, WakeReason,
|
||||
@ -57,8 +59,9 @@ use serde_json::{json, Map, Value};
|
||||
use uuid::Uuid;
|
||||
|
||||
use infrastructure::{
|
||||
embedder_from_profile, AdaptiveMemoryRecall, BackgroundTaskReadyToDeliver,
|
||||
ClaudePermissionProjector, ClaudeTranscriptInspector, CliAgentRuntime,
|
||||
embedder_from_profile, AdaptiveMemoryRecall, BackgroundCompletionSink,
|
||||
BackgroundTaskReadyToDeliver, ClaudePermissionProjector, ClaudeTranscriptInspector,
|
||||
CliAgentRuntime, CommandBackgroundRunner,
|
||||
CodexPermissionProjector, EmbedderEnvProbe, FsBackgroundTaskStore, FsConversationLog,
|
||||
FsEmbedderProfileStore, FsEmbedderPromptStore, FsHandoffStore, FsLiveStateStore, FsMemoryStore,
|
||||
FsOrchestratorWatcher, FsPermissionStore, FsProfileStore, FsProjectStore,
|
||||
@ -795,6 +798,15 @@ pub struct AppState {
|
||||
// --- Windows (L10) ---
|
||||
/// Detach a tab into a new OS window (persists the workspace topology).
|
||||
pub move_tab: Arc<MoveTabToNewWindow>,
|
||||
// --- Background tasks (B8) ---
|
||||
/// Spawn a command-backed first-class background task.
|
||||
pub spawn_background_command: Arc<SpawnBackgroundCommand>,
|
||||
/// Cancel a running background task.
|
||||
pub cancel_background_task: Arc<CancelBackgroundTask>,
|
||||
/// Retry a terminal command task under a fresh task id.
|
||||
pub retry_background_task: Arc<RetryBackgroundTask>,
|
||||
/// Store handle used by `list_background_tasks` to read the task read-model.
|
||||
pub background_task_store: Arc<dyn BackgroundTaskStore>,
|
||||
// --- Templates & sync (L7) ---
|
||||
/// Create a template in the global store.
|
||||
pub create_template: Arc<CreateTemplate>,
|
||||
@ -1541,6 +1553,49 @@ impl AppState {
|
||||
let background_tasks_port = Arc::clone(&background_tasks) as Arc<dyn BackgroundTaskStore>;
|
||||
let (background_ready_tx, mut background_ready_rx) =
|
||||
tokio::sync::mpsc::unbounded_channel::<BackgroundTaskReadyToDeliver>();
|
||||
|
||||
// --- B8 : runner de commandes + fermeture de la boucle du sink ---
|
||||
// Le runner concret exécute les tâches command-backed sur le `PtyPort` composé
|
||||
// (local ici ; SSH/WSL via `RemoteHost` quand ils atterriront — Liskov). Il
|
||||
// émet UNE complétion par tâche ; le sink (single-writer) persiste l'état
|
||||
// terminal AVANT de signaler la livraison (persist-avant-signal), puis le pont
|
||||
// ready→inbox ci-dessous réveille l'agent propriétaire.
|
||||
let background_runner = Arc::new(CommandBackgroundRunner::new(
|
||||
Arc::clone(&pty_port),
|
||||
Arc::clone(&clock) as Arc<dyn Clock>,
|
||||
));
|
||||
let background_runner_port =
|
||||
Arc::clone(&background_runner) as Arc<dyn BackgroundTaskRunner>;
|
||||
{
|
||||
let sink = Arc::new(BackgroundCompletionSink::new(
|
||||
Arc::clone(&background_tasks_port),
|
||||
background_ready_tx.clone(),
|
||||
));
|
||||
let runner = Arc::clone(&background_runner_port);
|
||||
// `start_from_runner` exige un runtime Tokio ambiant (`Handle::current`) :
|
||||
// on l'appelle donc DANS la tâche async. Le `JoinHandle` du drain est gardé
|
||||
// vivant en l'attendant (il ne se termine qu'à la fermeture du flux de
|
||||
// complétions), ce qui maintient sink + runner en vie pour la session.
|
||||
tauri::async_runtime::spawn(async move {
|
||||
let drain = sink.start_from_runner(runner);
|
||||
let _ = drain.await;
|
||||
});
|
||||
}
|
||||
let spawn_background_command = Arc::new(SpawnBackgroundCommand::new(
|
||||
Arc::clone(&background_tasks_port),
|
||||
Arc::clone(&background_runner_port),
|
||||
Arc::clone(&clock) as Arc<dyn Clock>,
|
||||
Arc::clone(&ids) as Arc<dyn IdGenerator>,
|
||||
));
|
||||
let cancel_background_task = Arc::new(CancelBackgroundTask::new(
|
||||
Arc::clone(&background_tasks_port),
|
||||
Arc::clone(&background_runner_port),
|
||||
));
|
||||
let retry_background_task = Arc::new(RetryBackgroundTask::new(
|
||||
Arc::clone(&background_tasks_port),
|
||||
Arc::clone(&background_runner) as Arc<dyn BackgroundCommandArchive>,
|
||||
Arc::clone(&spawn_background_command),
|
||||
));
|
||||
let background_wake = Arc::new(AgentWakeService::new(
|
||||
Arc::clone(&mediated_inbox) as Arc<dyn AgentInbox>,
|
||||
Arc::clone(&input_mediator),
|
||||
@ -1977,6 +2032,10 @@ impl AppState {
|
||||
fs_port: Arc::clone(&fs_port),
|
||||
home_dir,
|
||||
move_tab,
|
||||
spawn_background_command,
|
||||
cancel_background_task,
|
||||
retry_background_task,
|
||||
background_task_store: Arc::clone(&background_tasks_port),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
300
crates/application/src/background/mod.rs
Normal file
300
crates/application/src/background/mod.rs
Normal file
@ -0,0 +1,300 @@
|
||||
//! First-class background command use cases (B8 sub-task 3).
|
||||
//!
|
||||
//! These use cases own the *application* side of a command-backed background
|
||||
//! task: they allocate the [`TaskId`], create the persisted task
|
||||
//! ([`BackgroundTaskState::Queued`] → [`Running`](BackgroundTaskState::Running))
|
||||
//! and hand a [`BackgroundTaskSpec`] to the runner. They talk **only** to ports:
|
||||
//! [`BackgroundTaskStore`], [`BackgroundTaskRunner`], [`Clock`] and
|
||||
//! [`IdGenerator`], plus the runtime-only [`BackgroundCommandArchive`] abstraction
|
||||
//! defined below (see its doc for why retry recall is not a domain port).
|
||||
//!
|
||||
//! The runner never writes the store; the completion sink does. So beyond the
|
||||
//! initial create/transition, these use cases do not persist terminal state —
|
||||
//! except on a *spawn failure*, where they mark the just-created task `Failed` so
|
||||
//! it never dangles in `Running` with no worker behind it.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use domain::ports::{
|
||||
BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec, BackgroundTaskStore, Clock,
|
||||
IdGenerator, SpawnSpec,
|
||||
};
|
||||
use domain::{
|
||||
AgentId, BackgroundTask, BackgroundTaskKind, BackgroundTaskResult, BackgroundTaskState,
|
||||
BackgroundTaskWakePolicy, ProjectId, TaskId,
|
||||
};
|
||||
|
||||
use crate::error::AppError;
|
||||
|
||||
/// Maps a background-task port error into the application error shape.
|
||||
fn map_port_err(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),
|
||||
}
|
||||
}
|
||||
|
||||
/// Input for [`SpawnBackgroundCommand`].
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SpawnBackgroundCommandInput {
|
||||
/// Owning project.
|
||||
pub project_id: ProjectId,
|
||||
/// Agent that owns completion delivery.
|
||||
pub owner_agent_id: AgentId,
|
||||
/// Human-facing label.
|
||||
pub label: String,
|
||||
/// Command invocation to run.
|
||||
pub command: SpawnSpec,
|
||||
/// Completion wake policy.
|
||||
pub wake_policy: BackgroundTaskWakePolicy,
|
||||
/// Optional absolute deadline, epoch milliseconds.
|
||||
pub deadline_ms: Option<u64>,
|
||||
}
|
||||
|
||||
/// Output of [`SpawnBackgroundCommand`] / [`RetryBackgroundTask`].
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SpawnBackgroundCommandOutput {
|
||||
/// The created task, in its `Running` state.
|
||||
pub task: BackgroundTask,
|
||||
}
|
||||
|
||||
/// Creates a command-backed background task and hands it to the runner.
|
||||
pub struct SpawnBackgroundCommand {
|
||||
store: Arc<dyn BackgroundTaskStore>,
|
||||
runner: Arc<dyn BackgroundTaskRunner>,
|
||||
clock: Arc<dyn Clock>,
|
||||
ids: Arc<dyn IdGenerator>,
|
||||
}
|
||||
|
||||
impl SpawnBackgroundCommand {
|
||||
/// Wires the use case to its ports.
|
||||
#[must_use]
|
||||
pub fn new(
|
||||
store: Arc<dyn BackgroundTaskStore>,
|
||||
runner: Arc<dyn BackgroundTaskRunner>,
|
||||
clock: Arc<dyn Clock>,
|
||||
ids: Arc<dyn IdGenerator>,
|
||||
) -> Self {
|
||||
Self {
|
||||
store,
|
||||
runner,
|
||||
clock,
|
||||
ids,
|
||||
}
|
||||
}
|
||||
|
||||
/// Allocates a task id, persists it (`Queued`→`Running`) and spawns it.
|
||||
///
|
||||
/// # Errors
|
||||
/// [`AppError`] if a store/runner operation or a domain invariant fails.
|
||||
pub async fn execute(
|
||||
&self,
|
||||
input: SpawnBackgroundCommandInput,
|
||||
) -> Result<SpawnBackgroundCommandOutput, AppError> {
|
||||
let task = self
|
||||
.create_and_run(
|
||||
input.project_id,
|
||||
input.owner_agent_id,
|
||||
input.label,
|
||||
input.command,
|
||||
input.wake_policy,
|
||||
input.deadline_ms,
|
||||
)
|
||||
.await?;
|
||||
Ok(SpawnBackgroundCommandOutput { task })
|
||||
}
|
||||
|
||||
/// Shared create/transition/spawn path, reused by retry.
|
||||
async fn create_and_run(
|
||||
&self,
|
||||
project_id: ProjectId,
|
||||
owner_agent_id: AgentId,
|
||||
label: String,
|
||||
command: SpawnSpec,
|
||||
wake_policy: BackgroundTaskWakePolicy,
|
||||
deadline_ms: Option<u64>,
|
||||
) -> Result<BackgroundTask, AppError> {
|
||||
let task_id = TaskId::from_uuid(self.ids.new_uuid());
|
||||
let now = u64::try_from(self.clock.now_millis().max(0)).unwrap_or(0);
|
||||
|
||||
let task = BackgroundTask::new(
|
||||
task_id,
|
||||
project_id,
|
||||
owner_agent_id,
|
||||
BackgroundTaskKind::Command { label },
|
||||
wake_policy,
|
||||
now,
|
||||
deadline_ms,
|
||||
)
|
||||
.map_err(|e| AppError::Invalid(e.to_string()))?;
|
||||
self.store.create(&task).await.map_err(map_port_err)?;
|
||||
|
||||
let running = task
|
||||
.transition(BackgroundTaskState::Running, now)
|
||||
.map_err(|e| AppError::Invalid(e.to_string()))?;
|
||||
self.store.save(&running).await.map_err(map_port_err)?;
|
||||
|
||||
let spec = BackgroundTaskSpec {
|
||||
task_id,
|
||||
project_id,
|
||||
owner_agent_id,
|
||||
kind: running.kind.clone(),
|
||||
wake_policy,
|
||||
command: Some(command),
|
||||
deadline_ms,
|
||||
};
|
||||
|
||||
if let Err(err) = self.runner.spawn(spec).await {
|
||||
// Never leave the task dangling in `Running` with no worker: mark it
|
||||
// Failed so reconcile/UI see a terminal state. Best-effort persist.
|
||||
if let Ok(failed) = running.complete(BackgroundTaskResult::Failure {
|
||||
finished_at_ms: now,
|
||||
exit_code: None,
|
||||
error: format!("failed to spawn background command: {err}"),
|
||||
stdout_tail: None,
|
||||
stderr_tail: None,
|
||||
}) {
|
||||
let _ = self.store.save(&failed).await;
|
||||
}
|
||||
return Err(map_port_err(err));
|
||||
}
|
||||
|
||||
Ok(running)
|
||||
}
|
||||
}
|
||||
|
||||
/// Output of [`CancelBackgroundTask`].
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct CancelBackgroundTaskOutput {
|
||||
/// The task as currently persisted, if it still exists. The terminal
|
||||
/// `Cancelled` state is written by the completion sink, not here.
|
||||
pub task: Option<BackgroundTask>,
|
||||
}
|
||||
|
||||
/// Requests cancellation of a running background task.
|
||||
pub struct CancelBackgroundTask {
|
||||
store: Arc<dyn BackgroundTaskStore>,
|
||||
runner: Arc<dyn BackgroundTaskRunner>,
|
||||
}
|
||||
|
||||
impl CancelBackgroundTask {
|
||||
/// Wires the use case to its ports.
|
||||
#[must_use]
|
||||
pub fn new(store: Arc<dyn BackgroundTaskStore>, runner: Arc<dyn BackgroundTaskRunner>) -> Self {
|
||||
Self { store, runner }
|
||||
}
|
||||
|
||||
/// Signals the runner to cancel `task_id` (idempotent).
|
||||
///
|
||||
/// # Errors
|
||||
/// [`AppError`] if the runner or store fails.
|
||||
pub async fn execute(
|
||||
&self,
|
||||
task_id: TaskId,
|
||||
) -> Result<CancelBackgroundTaskOutput, AppError> {
|
||||
self.runner.cancel(task_id).await.map_err(map_port_err)?;
|
||||
let task = self.store.get(task_id).await.map_err(map_port_err)?;
|
||||
Ok(CancelBackgroundTaskOutput { task })
|
||||
}
|
||||
}
|
||||
|
||||
/// Read-only recall of the command invocation a background task was spawned with.
|
||||
///
|
||||
/// This is a **runtime, in-memory** abstraction, *not* a domain port: the recall
|
||||
/// lives wherever the runner keeps its live invocations for the current session
|
||||
/// and is deliberately **never persisted** (a [`SpawnSpec`] can carry secrets and
|
||||
/// must never land in `.ideai/background-tasks/*.json`, which travels with the
|
||||
/// project). Retry is therefore **session-scoped** in V1 — a task whose runner no
|
||||
/// longer remembers its spec (e.g. after a restart) simply cannot be retried.
|
||||
///
|
||||
/// The persisted [`BackgroundTask`] carries no [`SpawnSpec`] (the command line is
|
||||
/// an execution detail, not domain state). Retry, however, must re-run the *same*
|
||||
/// command under a fresh task id, so it recovers the original invocation from
|
||||
/// whatever component still holds it — the runner. Keeping that recall behind this
|
||||
/// application-level trait lets [`RetryBackgroundTask`] stay decoupled from the
|
||||
/// concrete runner without dragging the concept into the domain.
|
||||
pub trait BackgroundCommandArchive: Send + Sync {
|
||||
/// Returns the [`SpawnSpec`] a command-backed task was spawned with, if the
|
||||
/// runner still remembers it. `None` when the task is unknown, was not
|
||||
/// command-backed, or is no longer in the current session's registry.
|
||||
fn spec_for(&self, task_id: TaskId) -> Option<SpawnSpec>;
|
||||
}
|
||||
|
||||
/// Re-runs a terminal command-backed task under a brand-new task id.
|
||||
///
|
||||
/// The original command is recovered from the [`BackgroundCommandArchive`]; the
|
||||
/// deadline is intentionally dropped (an absolute past deadline would expire the
|
||||
/// retry instantly).
|
||||
pub struct RetryBackgroundTask {
|
||||
store: Arc<dyn BackgroundTaskStore>,
|
||||
archive: Arc<dyn BackgroundCommandArchive>,
|
||||
spawn: Arc<SpawnBackgroundCommand>,
|
||||
}
|
||||
|
||||
impl RetryBackgroundTask {
|
||||
/// Wires the use case to its ports and the spawn use case.
|
||||
#[must_use]
|
||||
pub fn new(
|
||||
store: Arc<dyn BackgroundTaskStore>,
|
||||
archive: Arc<dyn BackgroundCommandArchive>,
|
||||
spawn: Arc<SpawnBackgroundCommand>,
|
||||
) -> Self {
|
||||
Self {
|
||||
store,
|
||||
archive,
|
||||
spawn,
|
||||
}
|
||||
}
|
||||
|
||||
/// Retries `task_id`, returning the freshly-created task.
|
||||
///
|
||||
/// # Errors
|
||||
/// [`AppError::NotFound`] if the task is unknown, [`AppError::Invalid`] if it
|
||||
/// is not a terminal command task or its command is no longer available.
|
||||
pub async fn execute(
|
||||
&self,
|
||||
task_id: TaskId,
|
||||
) -> Result<SpawnBackgroundCommandOutput, AppError> {
|
||||
let old = self
|
||||
.store
|
||||
.get(task_id)
|
||||
.await
|
||||
.map_err(map_port_err)?
|
||||
.ok_or_else(|| AppError::NotFound(format!("background task {task_id}")))?;
|
||||
|
||||
if !old.is_terminal() {
|
||||
return Err(AppError::Invalid(
|
||||
"cannot retry a task that is still running".to_owned(),
|
||||
));
|
||||
}
|
||||
let label = match &old.kind {
|
||||
BackgroundTaskKind::Command { label } => label.clone(),
|
||||
_ => {
|
||||
return Err(AppError::Invalid(
|
||||
"only command-backed tasks can be retried".to_owned(),
|
||||
))
|
||||
}
|
||||
};
|
||||
let command = self.archive.spec_for(task_id).ok_or_else(|| {
|
||||
AppError::Invalid("original command is no longer available for retry".to_owned())
|
||||
})?;
|
||||
|
||||
let task = self
|
||||
.spawn
|
||||
.create_and_run(
|
||||
old.project_id,
|
||||
old.owner_agent_id,
|
||||
label,
|
||||
command,
|
||||
old.wake_policy,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
Ok(SpawnBackgroundCommandOutput { task })
|
||||
}
|
||||
}
|
||||
@ -12,6 +12,7 @@
|
||||
#![warn(missing_docs)]
|
||||
|
||||
pub mod agent;
|
||||
pub mod background;
|
||||
pub mod conversation;
|
||||
pub mod diag;
|
||||
pub mod embedder;
|
||||
@ -47,6 +48,10 @@ pub use agent::{
|
||||
StructuredSessionDescriptor, TurnOutcome, UpdateAgentContext, UpdateAgentContextInput,
|
||||
AGENT_MEMORY_RECALL_BUDGET, CODEX_SUBMIT_DELAY_MS, LIVE_STATE_INJECT_MAX, RESUME_PROMPT,
|
||||
};
|
||||
pub use background::{
|
||||
BackgroundCommandArchive, CancelBackgroundTask, CancelBackgroundTaskOutput, RetryBackgroundTask,
|
||||
SpawnBackgroundCommand, SpawnBackgroundCommandInput, SpawnBackgroundCommandOutput,
|
||||
};
|
||||
pub use conversation::{
|
||||
ConversationArchiveProvider, ReadConversationPage, ReadConversationPageInput, RecordTurn,
|
||||
RotateConversationLog, RotateConversationLogInput, TurnPage, TurnSource, TurnView,
|
||||
|
||||
@ -167,9 +167,9 @@ pub use orchestrator::{
|
||||
};
|
||||
|
||||
pub use ports::{
|
||||
AgentContextStore, AgentRuntime, BackgroundCompletionStream, BackgroundTaskCompletion,
|
||||
BackgroundTaskHandle, BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec,
|
||||
BackgroundTaskStore, Clock, ContextInjectionPlan, DirEntry, Embedder, EmbedderEnvInspector,
|
||||
AgentContextStore, AgentRuntime, BackgroundCompletionStream,
|
||||
BackgroundTaskCompletion, BackgroundTaskHandle, BackgroundTaskPortError, BackgroundTaskRunner,
|
||||
BackgroundTaskSpec, BackgroundTaskStore, Clock, ContextInjectionPlan, DirEntry, Embedder, EmbedderEnvInspector,
|
||||
EmbedderEnvReport, EmbedderError, EmbedderProfileStore, EmbedderPromptDismissal,
|
||||
EmbedderPromptStore, EventBus, EventStream, ExitStatus, FileSystem, FsError, GitCommitInfo,
|
||||
GitError, GitFileStatus, GitPort, GraphCommit, IdGenerator, LiveStateStore, MemoryError,
|
||||
|
||||
22
crates/infrastructure/src/background_task/mod.rs
Normal file
22
crates/infrastructure/src/background_task/mod.rs
Normal file
@ -0,0 +1,22 @@
|
||||
//! First-class background task infrastructure.
|
||||
//!
|
||||
//! Three cohesive pieces close the B2→B8 loop:
|
||||
//!
|
||||
//! - [`sink`] — the durable [`BackgroundCompletionSink`], single writer between a
|
||||
//! runner completion and the ready-to-deliver signal (persist-before-signal).
|
||||
//! - [`tail`] — a UTF-8-safe bounded ring buffer for stdout/stderr tails.
|
||||
//! - [`runner`] — [`CommandBackgroundRunner`], the concrete
|
||||
//! [`BackgroundTaskRunner`](domain::ports::BackgroundTaskRunner) that spawns a
|
||||
//! command-backed task over a composed [`PtyPort`](domain::ports::PtyPort) and
|
||||
//! emits exactly one completion per task.
|
||||
|
||||
mod runner;
|
||||
mod sink;
|
||||
mod tail;
|
||||
|
||||
pub use runner::CommandBackgroundRunner;
|
||||
pub use sink::{
|
||||
start_background_ready_inbox_bridge, BackgroundCompletionSink, BackgroundCompletionSinkError,
|
||||
BackgroundCompletionSinkOutcome, BackgroundReadyInboxBridgeHandle, BackgroundTaskReadyToDeliver,
|
||||
};
|
||||
pub use tail::{bounded_tail, tail_cap_bytes, BoundedTail};
|
||||
308
crates/infrastructure/src/background_task/runner.rs
Normal file
308
crates/infrastructure/src/background_task/runner.rs
Normal file
@ -0,0 +1,308 @@
|
||||
//! [`CommandBackgroundRunner`] — the concrete command-backed background task
|
||||
//! runner (B8 sub-task 2).
|
||||
//!
|
||||
//! # Role
|
||||
//!
|
||||
//! This is the `impl` of the frozen [`BackgroundTaskRunner`] port that the
|
||||
//! composition root wires under the [`BackgroundCompletionSink`]. It spawns a
|
||||
//! command-backed task over a **composed** [`PtyPort`] (resolved by the project's
|
||||
//! `RemoteHost`, so local/SSH/WSL are transparent — Liskov) and, for each task,
|
||||
//! emits **exactly one** [`BackgroundTaskCompletion`] on the stream returned by
|
||||
//! [`subscribe_completions`](BackgroundTaskRunner::subscribe_completions).
|
||||
//!
|
||||
//! It writes **neither** the store **nor** the inbox: the sink is the single
|
||||
//! writer that persists the terminal state before signalling delivery
|
||||
//! (persist-before-signal). The runner's only outward effect is the completion
|
||||
//! event.
|
||||
//!
|
||||
//! # Lifecycle of one task
|
||||
//!
|
||||
//! `spawn` opens the PTY and starts a detached worker. The worker watches the
|
||||
//! PTY's output stream for EOF (natural process exit), a cancel notification, or
|
||||
//! the deadline, whichever comes first. It then snapshots the bounded output tail
|
||||
//! from the PTY scrollback, reaps the real exit status via
|
||||
//! [`PtyPort::kill`](domain::ports::PtyPort::kill), maps that to a terminal
|
||||
//! [`BackgroundTaskResult`], and sends the completion.
|
||||
//!
|
||||
//! # Output tail & the live tee
|
||||
//!
|
||||
//! A PTY **merges** stdout and stderr onto one stream, so the completion carries
|
||||
//! the merged output in `stdout_tail` and leaves `stderr_tail` empty. Exit
|
||||
//! detection reads the PTY's single-consumer output subscription; a *live tee*
|
||||
//! that re-subscribes the same PTY (to render it in an xterm cell) would supersede
|
||||
//! the runner's subscription and break EOF detection — teeing a running command
|
||||
//! therefore requires a multi-consumer PTY broadcast, which is out of scope here.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::mpsc::{self, Receiver, Sender};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use application::BackgroundCommandArchive;
|
||||
use domain::ports::{
|
||||
BackgroundCompletionStream, BackgroundTaskCompletion, BackgroundTaskHandle,
|
||||
BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec, Clock, PtyHandle, PtyPort,
|
||||
SpawnSpec,
|
||||
};
|
||||
use domain::terminal::PtySize;
|
||||
use domain::{BackgroundTaskResult, TaskId};
|
||||
|
||||
use super::tail::{bounded_tail, tail_cap_bytes};
|
||||
|
||||
/// Default PTY geometry for a headless background command (no visible cell).
|
||||
const DEFAULT_SIZE: PtySize = PtySize { rows: 24, cols: 80 };
|
||||
|
||||
/// Why the worker's terminal wait resolved.
|
||||
enum WaitOutcome {
|
||||
/// The process exited on its own (stream hit EOF).
|
||||
Exited,
|
||||
/// A [`cancel`](BackgroundTaskRunner::cancel) was requested.
|
||||
Cancelled,
|
||||
/// The deadline elapsed before exit.
|
||||
Expired,
|
||||
}
|
||||
|
||||
/// Live control handles for one running task (kept in the registry).
|
||||
#[derive(Clone)]
|
||||
struct TaskControl {
|
||||
pty_handle: PtyHandle,
|
||||
cancel: Arc<tokio::sync::Notify>,
|
||||
cancelled: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
/// Command-backed [`BackgroundTaskRunner`] over a composed [`PtyPort`].
|
||||
pub struct CommandBackgroundRunner {
|
||||
pty: Arc<dyn PtyPort>,
|
||||
clock: Arc<dyn Clock>,
|
||||
completion_tx: Sender<BackgroundTaskCompletion>,
|
||||
completion_rx: Mutex<Option<Receiver<BackgroundTaskCompletion>>>,
|
||||
running: Arc<Mutex<HashMap<TaskId, TaskControl>>>,
|
||||
/// Retained command invocations, so [`BackgroundCommandArchive::spec_for`]
|
||||
/// can recover a task's command for retry.
|
||||
specs: Arc<Mutex<HashMap<TaskId, SpawnSpec>>>,
|
||||
}
|
||||
|
||||
impl CommandBackgroundRunner {
|
||||
/// Builds a runner over a composed PTY port and clock.
|
||||
#[must_use]
|
||||
pub fn new(pty: Arc<dyn PtyPort>, clock: Arc<dyn Clock>) -> Self {
|
||||
let (completion_tx, completion_rx) = mpsc::channel();
|
||||
Self {
|
||||
pty,
|
||||
clock,
|
||||
completion_tx,
|
||||
completion_rx: Mutex::new(Some(completion_rx)),
|
||||
running: Arc::new(Mutex::new(HashMap::new())),
|
||||
specs: Arc::new(Mutex::new(HashMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
fn now_ms(&self) -> u64 {
|
||||
u64::try_from(self.clock.now_millis().max(0)).unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Detached worker driving one command to its single completion.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
async fn run_to_completion(
|
||||
pty: Arc<dyn PtyPort>,
|
||||
clock: Arc<dyn Clock>,
|
||||
completion_tx: Sender<BackgroundTaskCompletion>,
|
||||
running: Arc<Mutex<HashMap<TaskId, TaskControl>>>,
|
||||
task_id: TaskId,
|
||||
control: TaskControl,
|
||||
deadline_ms: Option<u64>,
|
||||
) {
|
||||
let outcome = Self::await_terminal(&pty, &control, deadline_ms, &clock).await;
|
||||
// A cancel that raced a natural EOF still wins: the user asked to cancel.
|
||||
let outcome = if control.cancelled.load(Ordering::SeqCst) {
|
||||
WaitOutcome::Cancelled
|
||||
} else {
|
||||
outcome
|
||||
};
|
||||
|
||||
// Snapshot the merged output tail *before* kill tears the session down.
|
||||
let tail = pty
|
||||
.scrollback(&control.pty_handle)
|
||||
.ok()
|
||||
.map(|bytes| bounded_tail(&bytes, tail_cap_bytes()))
|
||||
.filter(|s| !s.is_empty());
|
||||
|
||||
// Reap the real exit status (works for a natural exit and a forced kill).
|
||||
let exit_code = pty
|
||||
.kill(&control.pty_handle)
|
||||
.await
|
||||
.ok()
|
||||
.and_then(|status| status.code);
|
||||
|
||||
let finished_at_ms = u64::try_from(clock.now_millis().max(0)).unwrap_or(0);
|
||||
let result = match outcome {
|
||||
WaitOutcome::Cancelled => BackgroundTaskResult::Cancelled {
|
||||
finished_at_ms,
|
||||
reason: "cancelled by request".to_owned(),
|
||||
},
|
||||
WaitOutcome::Expired => BackgroundTaskResult::Expired {
|
||||
finished_at_ms,
|
||||
reason: "deadline elapsed before completion".to_owned(),
|
||||
},
|
||||
WaitOutcome::Exited => {
|
||||
if exit_code == Some(0) {
|
||||
BackgroundTaskResult::Success {
|
||||
finished_at_ms,
|
||||
exit_code,
|
||||
summary: "command completed successfully".to_owned(),
|
||||
stdout_tail: tail,
|
||||
stderr_tail: None,
|
||||
}
|
||||
} else {
|
||||
BackgroundTaskResult::Failure {
|
||||
finished_at_ms,
|
||||
exit_code,
|
||||
error: match exit_code {
|
||||
Some(code) => format!("command exited with code {code}"),
|
||||
None => "command terminated without an exit code".to_owned(),
|
||||
},
|
||||
stdout_tail: tail,
|
||||
stderr_tail: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
running.lock().expect("runner registry poisoned").remove(&task_id);
|
||||
let _ = completion_tx.send(BackgroundTaskCompletion { task_id, result });
|
||||
}
|
||||
|
||||
/// Resolves when the process exits, is cancelled, or hits its deadline.
|
||||
async fn await_terminal(
|
||||
pty: &Arc<dyn PtyPort>,
|
||||
control: &TaskControl,
|
||||
deadline_ms: Option<u64>,
|
||||
clock: &Arc<dyn Clock>,
|
||||
) -> WaitOutcome {
|
||||
let eof = Self::wait_for_eof(Arc::clone(pty), control.pty_handle.clone());
|
||||
tokio::pin!(eof);
|
||||
|
||||
let sleep_ms = deadline_ms.map(|deadline| {
|
||||
let now = u64::try_from(clock.now_millis().max(0)).unwrap_or(0);
|
||||
deadline.saturating_sub(now)
|
||||
});
|
||||
|
||||
match sleep_ms {
|
||||
Some(ms) => tokio::select! {
|
||||
() = &mut eof => WaitOutcome::Exited,
|
||||
() = control.cancel.notified() => WaitOutcome::Cancelled,
|
||||
() = tokio::time::sleep(std::time::Duration::from_millis(ms)) => WaitOutcome::Expired,
|
||||
},
|
||||
None => tokio::select! {
|
||||
() = &mut eof => WaitOutcome::Exited,
|
||||
() = control.cancel.notified() => WaitOutcome::Cancelled,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolves when the PTY output stream ends (the process has exited).
|
||||
///
|
||||
/// The domain `OutputStream` is a blocking iterator, so it is drained on a
|
||||
/// blocking task and its bytes discarded — the persisted tail is read from the
|
||||
/// PTY scrollback instead, which the reader thread fills regardless of
|
||||
/// subscribers. If the handle is already unknown, resolves immediately so the
|
||||
/// worker proceeds to reap.
|
||||
async fn wait_for_eof(pty: Arc<dyn PtyPort>, handle: PtyHandle) {
|
||||
let Ok(stream) = pty.subscribe_output(&handle) else {
|
||||
return;
|
||||
};
|
||||
let _ = tokio::task::spawn_blocking(move || {
|
||||
for _chunk in stream {}
|
||||
})
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl BackgroundTaskRunner for CommandBackgroundRunner {
|
||||
async fn spawn(
|
||||
&self,
|
||||
spec: BackgroundTaskSpec,
|
||||
) -> Result<BackgroundTaskHandle, BackgroundTaskPortError> {
|
||||
let command = spec.command.clone().ok_or_else(|| {
|
||||
BackgroundTaskPortError::Runner(
|
||||
"command-backed runner requires a command spec".to_owned(),
|
||||
)
|
||||
})?;
|
||||
|
||||
let pty_handle = self
|
||||
.pty
|
||||
.spawn(command.clone(), DEFAULT_SIZE)
|
||||
.await
|
||||
.map_err(|e| BackgroundTaskPortError::Runner(format!("pty spawn failed: {e}")))?;
|
||||
|
||||
self.specs
|
||||
.lock()
|
||||
.expect("runner spec registry poisoned")
|
||||
.insert(spec.task_id, command);
|
||||
|
||||
let control = TaskControl {
|
||||
pty_handle,
|
||||
cancel: Arc::new(tokio::sync::Notify::new()),
|
||||
cancelled: Arc::new(AtomicBool::new(false)),
|
||||
};
|
||||
self.running
|
||||
.lock()
|
||||
.expect("runner registry poisoned")
|
||||
.insert(spec.task_id, control.clone());
|
||||
|
||||
tokio::spawn(Self::run_to_completion(
|
||||
Arc::clone(&self.pty),
|
||||
Arc::clone(&self.clock),
|
||||
self.completion_tx.clone(),
|
||||
Arc::clone(&self.running),
|
||||
spec.task_id,
|
||||
control,
|
||||
spec.deadline_ms,
|
||||
));
|
||||
|
||||
// Touch `now_ms` only to keep the clock wired for future scheduling; the
|
||||
// authoritative timestamps come from the worker at completion.
|
||||
let _ = self.now_ms();
|
||||
Ok(BackgroundTaskHandle {
|
||||
task_id: spec.task_id,
|
||||
})
|
||||
}
|
||||
|
||||
async fn cancel(&self, task_id: TaskId) -> Result<(), BackgroundTaskPortError> {
|
||||
let control = self
|
||||
.running
|
||||
.lock()
|
||||
.expect("runner registry poisoned")
|
||||
.get(&task_id)
|
||||
.cloned();
|
||||
// Unknown task ⇒ already terminal or never ours: cancellation is idempotent.
|
||||
if let Some(control) = control {
|
||||
control.cancelled.store(true, Ordering::SeqCst);
|
||||
control.cancel.notify_one();
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn subscribe_completions(&self) -> BackgroundCompletionStream {
|
||||
let rx = self
|
||||
.completion_rx
|
||||
.lock()
|
||||
.expect("runner completion receiver poisoned")
|
||||
.take()
|
||||
.expect("subscribe_completions must be called exactly once");
|
||||
Box::new(rx.into_iter())
|
||||
}
|
||||
}
|
||||
|
||||
impl BackgroundCommandArchive for CommandBackgroundRunner {
|
||||
fn spec_for(&self, task_id: TaskId) -> Option<SpawnSpec> {
|
||||
self.specs
|
||||
.lock()
|
||||
.expect("runner spec registry poisoned")
|
||||
.get(&task_id)
|
||||
.cloned()
|
||||
}
|
||||
}
|
||||
149
crates/infrastructure/src/background_task/tail.rs
Normal file
149
crates/infrastructure/src/background_task/tail.rs
Normal file
@ -0,0 +1,149 @@
|
||||
//! UTF-8-safe bounded output tail (B8 sub-task 1).
|
||||
//!
|
||||
//! A background command can print megabytes; the completion contract only keeps a
|
||||
//! bounded *tail* (the most recent bytes). [`BoundedTail`] is a byte ring buffer
|
||||
//! capped at construction; [`bounded_tail`] renders the retained bytes as a
|
||||
//! `String`, trimming any partial UTF-8 sequence at the front so a multi-byte
|
||||
//! character never gets sliced by the cap.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
|
||||
use domain::background_task::BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES;
|
||||
|
||||
/// Default tail cap when `IDEA_BG_TAIL_BYTES` is unset (8 KiB).
|
||||
const DEFAULT_TAIL_BYTES: usize = 8 * 1024;
|
||||
/// Floor so a hostile/typo'd env value can't disable the tail entirely.
|
||||
const MIN_TAIL_BYTES: usize = 1024;
|
||||
|
||||
/// Resolves the tail cap in bytes from `IDEA_BG_TAIL_BYTES`.
|
||||
///
|
||||
/// Defaults to 8 KiB, floored at 1 KiB, and clamped to the domain maximum
|
||||
/// ([`BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES`], 16 KiB) so the persisted result
|
||||
/// always validates.
|
||||
#[must_use]
|
||||
pub fn tail_cap_bytes() -> usize {
|
||||
std::env::var("IDEA_BG_TAIL_BYTES")
|
||||
.ok()
|
||||
.and_then(|v| v.trim().parse::<usize>().ok())
|
||||
.unwrap_or(DEFAULT_TAIL_BYTES)
|
||||
.clamp(MIN_TAIL_BYTES, BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES)
|
||||
}
|
||||
|
||||
/// A byte ring buffer that retains only the last `cap` bytes pushed into it.
|
||||
#[derive(Debug)]
|
||||
pub struct BoundedTail {
|
||||
buf: VecDeque<u8>,
|
||||
cap: usize,
|
||||
}
|
||||
|
||||
impl BoundedTail {
|
||||
/// Builds a tail retaining at most `cap` bytes (a `cap` of `0` retains
|
||||
/// nothing but never panics).
|
||||
#[must_use]
|
||||
pub fn with_cap(cap: usize) -> Self {
|
||||
Self {
|
||||
buf: VecDeque::new(),
|
||||
cap,
|
||||
}
|
||||
}
|
||||
|
||||
/// Appends a chunk, dropping the oldest bytes past the cap.
|
||||
pub fn push(&mut self, chunk: &[u8]) {
|
||||
if self.cap == 0 {
|
||||
return;
|
||||
}
|
||||
// If the incoming chunk alone exceeds the cap, only its tail can survive.
|
||||
let start = chunk.len().saturating_sub(self.cap);
|
||||
self.buf.extend(&chunk[start..]);
|
||||
let overflow = self.buf.len().saturating_sub(self.cap);
|
||||
if overflow > 0 {
|
||||
self.buf.drain(0..overflow);
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns whether any byte is retained.
|
||||
#[must_use]
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.buf.is_empty()
|
||||
}
|
||||
|
||||
/// Renders the retained bytes as a UTF-8 string, dropping a leading partial
|
||||
/// multi-byte sequence left dangling by the cap. Returns `None` when empty.
|
||||
#[must_use]
|
||||
pub fn to_tail_string(&self) -> Option<String> {
|
||||
if self.buf.is_empty() {
|
||||
return None;
|
||||
}
|
||||
let bytes: Vec<u8> = self.buf.iter().copied().collect();
|
||||
Some(bounded_tail(&bytes, self.cap))
|
||||
}
|
||||
}
|
||||
|
||||
/// Renders the last `cap` bytes of `bytes` as a lossless-at-the-boundary
|
||||
/// `String`.
|
||||
///
|
||||
/// The slice is taken from the end, then any partial UTF-8 sequence at the *front*
|
||||
/// (a continuation byte with no lead byte, because the cap cut mid-character) is
|
||||
/// skipped so the result is valid UTF-8 without replacement characters at the
|
||||
/// seam. Invalid bytes in the interior are still replaced lossily.
|
||||
#[must_use]
|
||||
pub fn bounded_tail(bytes: &[u8], cap: usize) -> String {
|
||||
let start = bytes.len().saturating_sub(cap);
|
||||
let mut slice = &bytes[start..];
|
||||
// Only realign when we actually truncated the front (start > 0): otherwise a
|
||||
// legitimately leading continuation byte is just invalid input, handled below.
|
||||
if start > 0 {
|
||||
let mut skip = 0;
|
||||
while skip < slice.len() && is_utf8_continuation(slice[skip]) {
|
||||
skip += 1;
|
||||
}
|
||||
slice = &slice[skip..];
|
||||
}
|
||||
String::from_utf8_lossy(slice).into_owned()
|
||||
}
|
||||
|
||||
/// Whether `b` is a UTF-8 continuation byte (`10xxxxxx`).
|
||||
const fn is_utf8_continuation(b: u8) -> bool {
|
||||
b & 0b1100_0000 == 0b1000_0000
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn keeps_only_the_last_cap_bytes() {
|
||||
let mut tail = BoundedTail::with_cap(4);
|
||||
tail.push(b"abcdef");
|
||||
assert_eq!(tail.to_tail_string().as_deref(), Some("cdef"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn push_bigger_than_cap_keeps_tail() {
|
||||
let mut tail = BoundedTail::with_cap(3);
|
||||
tail.push(b"0123456789");
|
||||
assert_eq!(tail.to_tail_string().as_deref(), Some("789"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_tail_is_none() {
|
||||
let tail = BoundedTail::with_cap(8);
|
||||
assert!(tail.is_empty());
|
||||
assert_eq!(tail.to_tail_string(), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cut_multibyte_char_is_trimmed_at_front() {
|
||||
// "é" is 0xC3 0xA9. Cap of 1 would keep only the trailing 0xA9 (a lone
|
||||
// continuation byte); it must be dropped, yielding an empty string.
|
||||
let bytes = "é".as_bytes();
|
||||
assert_eq!(bounded_tail(bytes, 1), "");
|
||||
// Cap of 2 keeps the whole character.
|
||||
assert_eq!(bounded_tail(bytes, 2), "é");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn interior_is_preserved() {
|
||||
assert_eq!(bounded_tail("héllo".as_bytes(), 100), "héllo");
|
||||
}
|
||||
}
|
||||
@ -38,9 +38,9 @@ pub mod store;
|
||||
pub mod timeparse;
|
||||
|
||||
pub use background_task::{
|
||||
start_background_ready_inbox_bridge, BackgroundCompletionSink, BackgroundCompletionSinkError,
|
||||
BackgroundCompletionSinkOutcome, BackgroundReadyInboxBridgeHandle,
|
||||
BackgroundTaskReadyToDeliver,
|
||||
bounded_tail, start_background_ready_inbox_bridge, tail_cap_bytes, BackgroundCompletionSink,
|
||||
BackgroundCompletionSinkError, BackgroundCompletionSinkOutcome, BackgroundReadyInboxBridgeHandle,
|
||||
BackgroundTaskReadyToDeliver, BoundedTail, CommandBackgroundRunner,
|
||||
};
|
||||
pub use clock::SystemClock;
|
||||
pub use conversation::InMemoryConversationRegistry;
|
||||
|
||||
Reference in New Issue
Block a user