diff --git a/CLAUDE.md b/CLAUDE.md index e9a2511..95ffdcf 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -334,6 +334,8 @@ The event **translator** (`loop_adapters/translate.rs`) is the ONE bus subscribe - A sub-agent is a **tool**, not an interception: `DelegateTool` (registered under the legacy names `execute_task` / `execute_subtask`, D11, each keeping its exact legacy schema) opens a child frame and runs a normal loop in it. The parent simply awaits a slow tool call. Max depth `MAX_AGENT_DEPTH = 5`. - **Parallel batches are the kernel's generic fan-out**: a round whose calls are all `concurrency_safe` (a sync delegate is) runs concurrently, bounded by `max_parallel_calls`. The ordering invariant is unchanged — ids allocated in call order (phase 1) → concurrent execution (phase 2) → recording in call order (phase 3) — so the model reconstructs results by id. Any mixed batch stays sequential. Siblings share the session scratchpad; concurrent writes to the same key are last-writer-wins by design. - `mode: "async"` submits a durable `scheduled_jobs` row through `loop_adapters/async_task.rs::CronExecutor` and returns a receipt immediately; when the job finishes, `DurableSink` writes the result into the parent conversation (synthetic assistant + a completed `task_completed` call) and resumes it. `mode: "cron"` is scheduling, not delegation, and stays on the cron interface tool. +- **An async task ends in the conversation that started it, whatever happened to it** — and `cron::run_job` is shaped so it cannot do otherwise: one `JobOutcome` classification, then *one* `match job.kind` delivery site for every ending. It used to branch on `Ok`/`Err` first and route by kind only inside `Ok`, so a failure or a kill went out as a "Cron job … failed" notification to the **home** source (`/sethome`) while the parent sat waiting for a `task_completed` that never came — the wrong chat *and* a wedged conversation. The sink has a single channel by design: to the model, "it broke" is a result like any other and must not be overlookable, so the failure is delivered as prose (with whatever partial output the run produced). A cron job has no parent conversation and keeps the home notification — the future plan is to let its creator name a destination. Cancellation is a third outcome, not a flavour of failure: `job_runs.status` always had `'cancelled'` in its CHECK and nothing wrote it, and the classifier keys on the **typed** `session::handler::TurnCancelled` error, never on the message text. +- **The chat shows what it started.** `ServerEvent::TaskUpdate` announces an async task's state to the source of its parent conversation only (a cron job belongs to nobody's chat), and `GET /api/{source}/tasks` (`db::scheduled_jobs::list_for_parent_session`) answers the same question at load time — running tasks plus failures from the last 30 minutes, because the event is a broadcast with no replay and a browser reload would otherwise empty a chat that still has work under it. Successes are absent from that query on purpose: a finished task's result is already a message in the conversation. The strip itself is `web/components/shared/agent-tasks.js` (`renderTaskStrip`), rendered above the composer on desktop and mobile from state owned by `ChatSession`; the drill-in is `#session/{id}`, gated on `_canOpenTaskSession` because the mobile shell routes a fixed set of sections and would silently swallow that hash. - A child's model is **never inherited** from the parent: passing a concrete name would bypass AUTO selection, so sub-agents auto-select unless explicitly overridden (`args.client` → `meta.json client` → AUTO by strength). - `list_agents` returns **task** agents only (never `chat`/`system` ones like the entry agent). diff --git a/Cargo.lock b/Cargo.lock index 8c4d2b7..622831c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4178,7 +4178,7 @@ checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" [[package]] name = "skald" -version = "0.1.2" +version = "0.2.0" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 9f01d2c..e8496de 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ resolver = "2" [package] name = "skald" -version = "0.1.2" +version = "0.2.0" edition = "2024" [features] diff --git a/crates/core-api/src/events.rs b/crates/core-api/src/events.rs index 2137924..498a641 100644 --- a/crates/core-api/src/events.rs +++ b/crates/core-api/src/events.rs @@ -285,6 +285,38 @@ pub enum ServerEvent { SecurityGroupSelected { group: String, }, + /// A background task (`execute_task` with `mode: "async"`) started by this + /// conversation changed state. + /// + /// Emitted only for async tasks, and only to the source of the conversation + /// that started one: a cron job belongs to nobody's chat. It drives a live + /// view and nothing else — a client that misses it is merely out of date, + /// never out of sync, because the task's real ending is delivered into the + /// conversation's own history. + TaskUpdate { + job_id: i64, + title: String, + agent_id: String, + /// The task's own session — `#session/{id}` shows what it is doing. + session_id: Option, + state: TaskState, + /// Why it ended badly. Set for `Failed` and `Cancelled`. + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, + }, +} + +/// The lifecycle state of a background task in a [`ServerEvent::TaskUpdate`]. +/// Mirrors `job_runs.status`, plus the `Running` state that table only records +/// by omission. +#[derive(Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum TaskState { + Running, + Completed, + Failed, + /// Stopped by a human before it finished. + Cancelled, } impl ServerEvent { @@ -324,6 +356,7 @@ impl ServerEvent { Self::TurnRunning { .. } => "turn_running", Self::ClientSelected { .. } => "client_selected", Self::SecurityGroupSelected { .. } => "security_group_selected", + Self::TaskUpdate { .. } => "task_update", } } } diff --git a/crates/skald-core/src/cron/mod.rs b/crates/skald-core/src/cron/mod.rs index 732e0f8..01ae9db 100644 --- a/crates/skald-core/src/cron/mod.rs +++ b/crates/skald-core/src/cron/mod.rs @@ -10,11 +10,13 @@ use tokio::sync::mpsc; use tokio::time::Duration; use tracing::{error, info}; +use core_api::events::{ServerEvent, TaskState}; use core_api::system_bus::{SystemEvent, SystemEventBus}; use crate::chat_hub::ChatHub; use crate::db::chat_sessions; use crate::db::scheduled_jobs::{self, ScheduledJob}; +use crate::session::handler::TurnCancelled; use crate::session::manager::ChatSessionManager; pub struct TaskManager { @@ -389,6 +391,11 @@ async fn run_job( } } + // The conversation that asked for this task learns it started, so the chat's + // background-task strip can show it without polling. Cron jobs are excluded + // on purpose: they belong to nobody's conversation. + emit_task_update(pool, hub, job, Some(session_id), TaskState::Running, None).await; + let job_context = format!( "[Job context]\nJob ID: {} — {}\nTime: {} UTC", job.id, job.title, @@ -442,94 +449,217 @@ async fn run_job( .map(|t| t.to_rfc3339()) }; - match handle_result { - Ok(_) => { - record_job_run(pool, job.id, session_id, &started_at.to_rfc3339(), - &completed_at.to_rfc3339(), duration_ms, - "completed", final_response.as_deref(), None).await?; - scheduled_jobs::finish_run(pool, job.id, next_run_at.as_deref()).await?; + // ── Outcome ────────────────────────────────────────────────────────────── + // + // One classification, one delivery site, for **every** ending. The previous + // shape branched on `Ok`/`Err` first and only routed by `kind` inside the + // `Ok` arm, so a failed or killed async task never reached the conversation + // that started it: it went out as a "Cron job … failed" notification to the + // home source, while the parent sat waiting for a `task_completed` that + // would never come. An async task ends in its parent conversation whatever + // happened to it — that is the rule this shape makes structural. + let outcome = JobOutcome::classify(handle_result); + let error_text = outcome.error(); - task_mgr.system_bus.send(SystemEvent::JobCompleted { - job_id: job.id, - origin_ref: job.origin_ref.clone(), - result: final_response.clone(), - error: None, - }); + record_job_run(pool, job.id, session_id, &started_at.to_rfc3339(), + &completed_at.to_rfc3339(), duration_ms, + outcome.run_status(), + outcome.is_ok().then_some(final_response.as_deref()).flatten(), + error_text.as_deref()).await?; + scheduled_jobs::finish_run(pool, job.id, next_run_at.as_deref()).await?; - match job.kind.as_str() { - "cron" => { - if let Some(hub) = hub { - let outcome = final_response.as_deref().unwrap_or("(no output)"); - hub.notify(crate::notification::Notification { - source: "cron".into(), - event_type: "cron_result".into(), - summary: format!( - "Cron job \"{}\" (ID {}) completed: {}", - job.title, job.id, outcome, - ), - event_time: Utc::now().to_rfc3339(), - refs: serde_json::json!({ "job_id": job.id, "title": job.title }), - }).await.ok(); - } - } - "async" => { - if let Some(parent_id) = job.parent_session_id { - if let Some(hub) = hub { - inject_async_result( - &task_mgr.pool, - hub, - parent_id, - job.id, - &job.title, - final_response.as_deref().unwrap_or("(no output)"), - ).await; - } - } - } - _ => {} // sync: result was already returned inline via add_job_sync - } + task_mgr.system_bus.send(SystemEvent::JobCompleted { + job_id: job.id, + origin_ref: job.origin_ref.clone(), + result: outcome.is_ok().then(|| final_response.clone()).flatten(), + error: error_text.clone(), + }); - info!("{} task {} done", job.kind, job.id); - Ok(final_response) - } - Err(e) => { - let err_str = e.to_string(); - record_job_run(pool, job.id, session_id, &started_at.to_rfc3339(), - &completed_at.to_rfc3339(), duration_ms, - "failed", None, Some(&err_str)).await?; - scheduled_jobs::finish_run(pool, job.id, next_run_at.as_deref()).await?; - - task_mgr.system_bus.send(SystemEvent::JobCompleted { - job_id: job.id, - origin_ref: job.origin_ref.clone(), - result: None, - error: Some(err_str.clone()), - }); + emit_task_update( + pool, hub, job, Some(session_id), + outcome.task_state(), error_text.as_deref(), + ).await; + match job.kind.as_str() { + "cron" => { if let Some(hub) = hub { hub.notify(crate::notification::Notification { source: "cron".into(), - event_type: "cron_error".into(), - summary: format!( - "Cron job \"{}\" (ID {}) failed: {} (check the logs)", - job.title, job.id, err_str, - ), + event_type: outcome.notification_event_type().into(), + summary: outcome.cron_summary(job, final_response.as_deref()), event_time: Utc::now().to_rfc3339(), refs: serde_json::json!({ "job_id": job.id, "title": job.title }), }).await.ok(); } - Err(e) + } + "async" => { + if let (Some(parent_id), Some(hub)) = (job.parent_session_id, hub) { + inject_async_result( + &task_mgr.pool, + hub, + parent_id, + job.id, + &job.title, + &outcome.delivery_text(final_response.as_deref()), + ).await; + } + } + _ => {} // sync: the result was already returned inline via add_job_sync + } + + match outcome { + JobOutcome::Completed => { + info!("{} task {} done", job.kind, job.id); + Ok(final_response) + } + JobOutcome::Failed(e) | JobOutcome::Cancelled(e) => Err(e), + } +} + +/// How a job run ended. Cancellation is a third state, not a flavour of +/// failure: `job_runs.status` has always had `'cancelled'` in its CHECK and +/// nothing ever wrote it, so a task the user killed was indistinguishable in +/// the history from one that broke. +enum JobOutcome { + Completed, + Failed(anyhow::Error), + /// Stopped by a human (`/kill`, `/stop`). + Cancelled(anyhow::Error), +} + +impl JobOutcome { + fn classify(result: Result<()>) -> Self { + match result { + Ok(()) => Self::Completed, + Err(e) if e.downcast_ref::().is_some() => Self::Cancelled(e), + Err(e) => Self::Failed(e), + } + } + + fn is_ok(&self) -> bool { + matches!(self, Self::Completed) + } + + fn run_status(&self) -> &'static str { + match self { + Self::Completed => "completed", + Self::Failed(_) => "failed", + Self::Cancelled(_) => "cancelled", + } + } + + fn task_state(&self) -> TaskState { + match self { + Self::Completed => TaskState::Completed, + Self::Failed(_) => TaskState::Failed, + Self::Cancelled(_) => TaskState::Cancelled, + } + } + + /// The error text, for the run log and the WS event. `None` when the run + /// completed — a cancellation *has* one, since "stopped by the user" is + /// what the history should say. + fn error(&self) -> Option { + match self { + Self::Completed => None, + Self::Failed(e) => Some(e.to_string()), + Self::Cancelled(_) => Some("Stopped by the user before it finished.".to_string()), + } + } + + fn notification_event_type(&self) -> &'static str { + match self { + Self::Completed => "cron_result", + _ => "cron_error", + } + } + + fn cron_summary(&self, job: &ScheduledJob, final_response: Option<&str>) -> String { + match self { + Self::Completed => format!( + "Cron job \"{}\" (ID {}) completed: {}", + job.title, job.id, final_response.unwrap_or("(no output)"), + ), + Self::Failed(e) => format!( + "Cron job \"{}\" (ID {}) failed: {e} (check the logs)", + job.title, job.id, + ), + Self::Cancelled(_) => format!( + "Cron job \"{}\" (ID {}) was stopped before it finished.", + job.title, job.id, + ), + } + } + + /// What the parent conversation is told. The model reads this as the result + /// of the `task_completed` call, so a failure has to *say* it failed — + /// prose, not a status code — and carry whatever the task did produce + /// before dying, which is usually the only clue about why. + fn delivery_text(&self, final_response: Option<&str>) -> String { + let partial = |body: String| match final_response { + Some(r) if !r.trim().is_empty() => + format!("{body}\n\nLast thing the task said before stopping:\n{r}"), + _ => body, + }; + match self { + Self::Completed => final_response.unwrap_or("(no output)").to_string(), + Self::Failed(e) => partial(format!( + "This task FAILED — it never produced a final answer.\n\nError: {e}" + )), + Self::Cancelled(_) => partial( + "This task was STOPPED by the user before it finished. \ + Its work is incomplete; do not present it as done." + .to_string(), + ), } } } -/// Delivers an async task's result to the parent session through the loop's +/// Announces an async task's state to the conversation that started it, over +/// that source's WebSocket. Best-effort and silent on failure: it drives a +/// live view, never a state transition — the truth is `scheduled_jobs` plus the +/// result delivered into the parent's history. +/// +/// A cron job has no parent conversation, so it emits nothing. +async fn emit_task_update( + pool: &SqlitePool, + hub: Option<&Arc>, + job: &ScheduledJob, + session_id: Option, + state: TaskState, + error: Option<&str>, +) { + if job.kind != "async" { return; } + let (Some(hub), Some(parent_id)) = (hub, job.parent_session_id) else { return }; + + let Ok(Some(parent)) = chat_sessions::find_by_id(pool, parent_id).await else { return }; + + hub.emit(core_api::events::GlobalEvent { + source: Some(parent.source), + session_id: Some(parent_id), + event: ServerEvent::TaskUpdate { + job_id: job.id, + title: job.title.clone(), + agent_id: job.agent_id.clone(), + session_id, + state, + error: error.map(str::to_string), + }, + }); +} + +/// Delivers an async task's **outcome** to the parent session through the loop's /// [`AsyncResultSink`] seam (blueprint §7.2): the library writes the synthetic /// assistant message + completed `task_completed` call, and Skald's /// [`DurableSink`] resumes the parent so the model reads it right away. /// -/// Failures are logged, never propagated: the job itself succeeded, and losing -/// the delivery must not mark it failed. +/// `result` is whatever the conversation should be told — an answer, or the +/// prose that says the task failed or was stopped. The sink has one channel and +/// that is deliberate: to the model reading it, "it broke" is a result like any +/// other, and one it must not be able to overlook. +/// +/// A delivery failure is logged, never propagated: it cannot change how the run +/// itself is recorded. async fn inject_async_result( pool: &Arc, hub: &Arc, diff --git a/crates/skald-core/src/db/scheduled_jobs.rs b/crates/skald-core/src/db/scheduled_jobs.rs index 74bb403..e2a2484 100644 --- a/crates/skald-core/src/db/scheduled_jobs.rs +++ b/crates/skald-core/src/db/scheduled_jobs.rs @@ -81,6 +81,90 @@ pub async fn list_interrupted(pool: &SqlitePool) -> Result> { Ok(rows) } +/// One background (`async`) task as the conversation that started it sees it. +/// A flattened join of the job with its latest run — the chat cares about a +/// task's *current* state, not its scheduling row. +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct SessionTask { + pub job_id: i64, + pub title: String, + pub agent_id: String, + /// The task's own session (`#session/{id}`). + pub session_id: Option, + /// `running` or `failed` — the only two states this query returns. + pub state: String, + pub error: Option, + /// When it started, normalised to RFC 3339 (see [`normalise_ts`]). + pub started_at: Option, +} + +/// The background tasks one conversation should still be showing: everything +/// running right now, plus failures from the last `failed_within_minutes`. +/// +/// Those are the two states a person can still act on — and the reason this +/// query exists at all is the browser reload: the strip is driven by +/// `ServerEvent::TaskUpdate`, which is a live broadcast with no replay, so +/// without a load-time read a refresh would empty a chat that still has work +/// running under it. Successes are deliberately absent: a completed task's +/// result is already a message in the conversation, which is a better place to +/// read it than a status chip. +pub async fn list_for_parent_session( + pool: &SqlitePool, + parent_session_id: i64, + failed_within_minutes: i64, +) -> Result> { + let rows = sqlx::query_as::<_, SessionTask>( + "SELECT sj.id AS job_id, + sj.title AS title, + sj.agent_id AS agent_id, + COALESCE(sj.running_session_id, jr.session_id) AS session_id, + CASE WHEN sj.running_session_id IS NOT NULL + THEN 'running' ELSE jr.status END AS state, + jr.error AS error, + COALESCE(sj.running_since, jr.started_at) AS started_at + FROM scheduled_jobs sj + LEFT JOIN job_runs jr + ON jr.id = (SELECT id FROM job_runs + WHERE job_id = sj.id ORDER BY id DESC LIMIT 1) + WHERE sj.kind = 'async' + AND sj.parent_session_id = ? + AND (sj.running_session_id IS NOT NULL + -- `datetime()` on both sides, never a raw string compare: + -- `completed_at` is RFC 3339 (`…T…+00:00`) and the cutoff is + -- SQLite-shaped, and `'T' > ' '` makes every same-day row + -- compare as newer than the cutoff — a window that lets + -- through everything it was meant to exclude. + OR (jr.status = 'failed' + AND datetime(jr.completed_at) >= datetime('now', ?))) + ORDER BY sj.id", + ) + .bind(parent_session_id) + .bind(format!("-{failed_within_minutes} minutes")) + .fetch_all(pool) + .await?; + + Ok(rows.into_iter() + .map(|mut t| { t.started_at = t.started_at.as_deref().and_then(normalise_ts); t }) + .collect()) +} + +/// The two timestamp shapes this table mixes, as one RFC 3339 string: +/// `running_since` is written by SQLite's `datetime('now')` (`Y-m-d H:M:S`, +/// UTC, no offset) while `job_runs.started_at` is already RFC 3339. A client +/// that guesses wrong is off by its own timezone, so the guess is made here. +fn normalise_ts(raw: &str) -> Option { + use chrono::{DateTime, NaiveDateTime, Utc}; + DateTime::parse_from_rfc3339(raw) + .map(|d| d.with_timezone(&Utc)) + .ok() + .or_else(|| { + NaiveDateTime::parse_from_str(raw, "%Y-%m-%d %H:%M:%S") + .ok() + .map(|n| n.and_utc()) + }) + .map(|d| d.to_rfc3339()) +} + pub async fn create( pool: &SqlitePool, title: &str, @@ -202,3 +286,88 @@ pub async fn finish_run( .await?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + /// One conversation (session 1) with a background task in each state, plus + /// the rows the query must not pick up: another conversation's task, and a + /// cron job (which belongs to nobody's chat). + async fn seeded() -> SqlitePool { + let pool = SqlitePool::connect("sqlite::memory:").await.unwrap(); + crate::db::create_owner_tables(&pool).await.unwrap(); + + let q = |sql: &'static str| sqlx::query(sql).execute(&pool); + q("INSERT INTO chat_sessions (id, title, source) VALUES (1, 'chat', 'web')").await.unwrap(); + q("INSERT INTO chat_sessions (id, title, source) VALUES (2, 'other', 'mobile')").await.unwrap(); + + let job = |id: i64, title: &'static str, kind: &'static str, + parent: Option, running: Option| { + sqlx::query( + "INSERT INTO scheduled_jobs + (id, title, cron, prompt, agent_id, kind, parent_session_id, + running_session_id, running_since, single_run) + VALUES (?, ?, '', 'p', 'researcher', ?, ?, ?, '2026-08-04 10:00:00', 1)", + ) + .bind(id).bind(title).bind(kind).bind(parent).bind(running) + .execute(&pool) + }; + job(1, "still going", "async", Some(1), Some(11)).await.unwrap(); + job(2, "just broke", "async", Some(1), None).await.unwrap(); + job(3, "finished ok", "async", Some(1), None).await.unwrap(); + job(4, "broke a while ago", "async", Some(1), None).await.unwrap(); + job(5, "someone else's", "async", Some(2), Some(55)).await.unwrap(); + job(6, "nightly digest", "cron", None, Some(66)).await.unwrap(); + + let run = |job_id: i64, session: i64, status: &'static str, completed: String| { + sqlx::query( + "INSERT INTO job_runs (job_id, session_id, started_at, completed_at, + duration_ms, status, error) + VALUES (?, ?, '2026-08-04T10:00:00+00:00', ?, 10, ?, 'boom')", + ) + .bind(job_id).bind(session).bind(completed).bind(status) + .execute(&pool) + }; + // RFC 3339, exactly as `run_job` writes it — the shape the window has to + // cope with. A test that seeded SQLite-shaped strings here would pass + // against a plain string comparison that production data defeats. + let now = chrono::Utc::now(); + let at = |m: i64| (now - chrono::Duration::minutes(m)).to_rfc3339(); + run(2, 22, "failed", at(1)).await.unwrap(); + run(3, 33, "completed", at(1)).await.unwrap(); + run(4, 44, "failed", at(120)).await.unwrap(); + + pool + } + + /// The strip shows what is running plus what has just broken — and nothing + /// that belongs to another conversation, to the schedule, or to yesterday. + #[tokio::test] + async fn a_conversation_sees_its_running_and_recently_failed_tasks() { + let pool = seeded().await; + let tasks = list_for_parent_session(&pool, 1, 30).await.unwrap(); + + let seen: Vec<_> = tasks.iter().map(|t| (t.job_id, t.state.as_str())).collect(); + assert_eq!(seen, vec![(1, "running"), (2, "failed")]); + + // The drill-in target: the running job's live session, the failed one's run. + assert_eq!(tasks[0].session_id, Some(11)); + assert_eq!(tasks[1].session_id, Some(22)); + assert_eq!(tasks[1].error.as_deref(), Some("boom")); + } + + /// `running_since` is SQLite-shaped and `job_runs.started_at` is RFC 3339; + /// both leave here as RFC 3339, or a browser reads one of them in the wrong + /// timezone and shows an elapsed counter hours off. + #[tokio::test] + async fn started_at_is_normalised_to_rfc3339() { + let pool = seeded().await; + let tasks = list_for_parent_session(&pool, 1, 30).await.unwrap(); + for task in &tasks { + let raw = task.started_at.as_deref().expect("a started task has a start time"); + chrono::DateTime::parse_from_rfc3339(raw) + .unwrap_or_else(|e| panic!("job {} start time {raw:?}: {e}", task.job_id)); + } + } +} diff --git a/crates/skald-core/src/session/handler/mod.rs b/crates/skald-core/src/session/handler/mod.rs index 9571f66..720e262 100644 --- a/crates/skald-core/src/session/handler/mod.rs +++ b/crates/skald-core/src/session/handler/mod.rs @@ -81,6 +81,24 @@ pub(super) enum TurnOutcome { Exhausted, } +/// A turn stopped by a human — `/stop` in the chat, or an admin killing a +/// running job — as opposed to one that failed. +/// +/// It is a **typed** error carried by the `anyhow::Error` `handle_message` +/// returns, so a caller that cares about the difference (the cron runner, which +/// records `cancelled` rather than `failed` and words the delivery accordingly) +/// classifies it with `downcast_ref` and never by matching the message text. +#[derive(Debug)] +pub struct TurnCancelled; + +impl std::fmt::Display for TurnCancelled { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("Turn cancelled by user") + } +} + +impl std::error::Error for TurnCancelled {} + /// Truncate `s` to at most `max_chars` characters, appending `…` when it was /// longer. Char-boundary safe: a raw `&s[..n]` byte slice panics when byte `n` /// lands inside a multi-byte UTF-8 character (e.g. an em-dash or emoji straddling @@ -641,7 +659,7 @@ impl ChatSessionHandler { info!(session_id = self.session_id, "handle_message cancelled by user"); // The "Cancelled by user." error event was already emitted by // the translator (root LoopEvent::Cancelled). - Err(anyhow::anyhow!("Turn cancelled by user")) + Err(anyhow::Error::new(TurnCancelled)) } TurnOutcome::Exhausted => { error!(session_id = self.session_id, max_rounds = self.max_tool_rounds, "tool-call loop exhausted without final answer"); diff --git a/docs/index.md b/docs/index.md index 8da7702..27bb824 100644 --- a/docs/index.md +++ b/docs/index.md @@ -4,7 +4,7 @@ This folder is written for **you, the assistant**, not for the human directly. I Keep answers grounded in what's actually enabled and configured for this instance — check with the relevant tool (e.g. list installed/enabled plugins) rather than assuming everything described here is turned on. A feature documented here may not be enabled on this particular instance. -This index will grow over time. Right now it covers memory, projects, system agents, access grants, voice input and plugins; more sections (agents, connectors, security groups, shared folders…) will be added later. +This index will grow over time. Right now it covers memory, projects, background tasks, system agents, access grants, voice input and plugins; more sections (agents, connectors, security groups, shared folders…) will be added later. ## Features @@ -13,6 +13,7 @@ This index will grow over time. Right now it covers memory, projects, system age | [memory.md](memory.md) | Private and shared memory: what goes where, the indexes and history log, why some shared facts can't be changed on request | | [projects.md](projects.md) | Projects: shared folders with their own assistant chat, a live file explorer, and member sharing | | [system-agents.md](system-agents.md) | Background agents that run on a schedule (event triage, the two memory lints, the nightly conversation review of a supervised account): what they watch, why they only ever report, why a run can be skipped, and their settings | +| [tasks.md](tasks.md) | Background tasks: the strip above the message box, following one live, stopping one, and how every outcome comes back to the conversation | | [settings.md](settings.md) | The admin's Config page: interface language, the compaction model picker, debug mode | | [access.md](access.md) | Who can use which plugin or connector: the open default, removing access per person, and the role switch that keeps children out of it | | [voice.md](voice.md) | Voice input: configuring a transcription model, and why the microphone button does nothing unless the page is served over HTTPS or localhost | diff --git a/docs/tasks.md b/docs/tasks.md new file mode 100644 index 0000000..0a1696e --- /dev/null +++ b/docs/tasks.md @@ -0,0 +1,37 @@ +# Background tasks (work that keeps running while you talk) + +Some requests take minutes rather than seconds — reading a long document, searching the web thoroughly, crunching a folder of files. For those, the assistant can hand the work to a **background task**: a second agent that goes off and does it while the conversation carries on. The user does not have to sit and wait, and can keep asking about other things. + +This is different from the two neighbouring things it is easy to confuse it with: + +- A **sub-task** (the ordinary kind) runs *inside* the current answer. The conversation waits for it, and its progress is visible in the transcript as it happens. +- A **scheduled job** (a cron job) runs at a time of day, on repeat, and belongs to nobody's conversation. Those live on the **Tasks** page and report to the home chat. +- A **background task** belongs to the conversation that started it, and comes back to it. + +## Seeing them: the strip above the message box + +While a background task is running, a small strip appears **just above the composer**, in the desktop chat and the mobile one alike. One line per task: its title, which agent is doing it, and how long it has been going. + +- **Clicking a task opens its own page**, where its work is shown live — the same view used for any background agent. This is the answer to "what is it actually doing?", which the chat itself cannot show: the task is a separate conversation. +- **The ■ button stops a task.** It stops there and then; whatever it had done so far is not thrown away, but it is incomplete, and the assistant is told so. +- The strip survives a page reload. It shows what is running *now*, so a task that finished while the browser was closed will not be there — but its result will be in the conversation, which is the better place to read it. + +## How a task comes back + +**Every** background task ends up back in the conversation that started it. There is no case where the user has to go looking for the outcome: + +- **It succeeded** — its answer arrives as a message, and the assistant carries on from there, usually with a summary. +- **It failed** — the conversation is told it failed and why, together with whatever the task managed to say before it broke. The assistant should treat this as a real result and say so plainly, not quietly ignore it. +- **It was stopped** by the user — the conversation is told the work is incomplete. It must not be presented as if it had finished. + +A finished task's line disappears from the strip after a few seconds. A **failed** one stays, so the reason can be read, until it is dismissed with the ✕. + +## When a user asks about them + +Common questions and the honest answers: + +- *"Is it still running?"* — the strip is the answer; if the strip is empty, nothing of theirs is running. +- *"What is it doing?"* — click the task's line. +- *"It has been going for ages."* — a task has no time limit; stopping it with ■ is always available, and stopping is not the same as failing. +- *"Where did the result go?"* — into this conversation, always. If it is not there yet, the task has not finished. +- *"Show me everything that ever ran."* — the **Tasks** page (sidebar → Tasks) has the full history, including scheduled jobs; the strip only covers the current conversation. diff --git a/src/frontend/api/cron.rs b/src/frontend/api/cron.rs index 78aeb8f..95ff664 100644 --- a/src/frontend/api/cron.rs +++ b/src/frontend/api/cron.rs @@ -5,7 +5,7 @@ use axum::{ }; use serde::Deserialize; -use skald_core::db::{scheduled_jobs, job_runs}; +use skald_core::db::{scheduled_jobs, job_runs, sources}; use std::sync::Arc; use skald_core::skald::Skald; use super::{ApiError, guard::AuthUser, require_context}; @@ -130,6 +130,55 @@ pub async fn kill_job( Ok(StatusCode::ACCEPTED) } +// ── GET /api/{source}/tasks ─────────────────────────────────────────────────── + +/// Failures stay in a conversation's task strip this long. Long enough that a +/// reload right after one still shows it, short enough that yesterday's does not. +const FAILED_TASK_WINDOW_MINUTES: i64 = 30; + +#[derive(Deserialize)] +pub struct SourcePath { pub source: String } + +#[derive(serde::Serialize)] +pub struct SessionTaskResponse { + pub job_id: i64, + pub title: String, + pub agent_id: String, + pub session_id: Option, + pub state: String, + pub error: Option, + pub started_at: Option, +} + +/// The background tasks the chat for `source` should be showing right now. +/// +/// The strip's live updates ride `ServerEvent::TaskUpdate`, a broadcast with no +/// replay — so this is what a page reload reads to get its state back. An +/// unknown source (no session yet) is an empty list, not an error: a chat that +/// has never run is a legitimate caller. +pub async fn session_tasks( + State(skald): State>, + Extension(auth): Extension, + Path(p): Path, +) -> Result>, ApiError> { + let ctx = require_context(&skald, &auth.user_id).await?; + let Some(session_id) = sources::active_session_id(&ctx.pool, &p.source).await? else { + return Ok(Json(vec![])); + }; + let tasks = scheduled_jobs::list_for_parent_session( + &ctx.pool, session_id, FAILED_TASK_WINDOW_MINUTES, + ).await?; + Ok(Json(tasks.into_iter().map(|t| SessionTaskResponse { + job_id: t.job_id, + title: t.title, + agent_id: t.agent_id, + session_id: t.session_id, + state: t.state, + error: t.error, + started_at: t.started_at, + }).collect())) +} + pub async fn list_runs( State(skald): State>, Extension(auth): Extension, diff --git a/src/frontend/api/mod.rs b/src/frontend/api/mod.rs index b615cd8..2bd503f 100644 --- a/src/frontend/api/mod.rs +++ b/src/frontend/api/mod.rs @@ -72,6 +72,8 @@ pub fn router() -> Router> { .route("/sessions/{id}", get(sessions::get_session_detail)) .route("/web/messages", get(sessions::web_messages)) .route("/{source}/messages", get(sessions::source_messages)) + // Background (`execute_task mode=async`) tasks of this source's session. + .route("/{source}/tasks", get(cron::session_tasks)) // File attachments: streamed to disk, so the default body-size limit is // disabled on this route only. .route("/{source}/uploads", post(uploads::upload).layer(DefaultBodyLimit::disable())) diff --git a/web/components/copilot.js b/web/components/copilot.js index d08f951..6920393 100644 --- a/web/components/copilot.js +++ b/web/components/copilot.js @@ -2,6 +2,7 @@ import { html, nothing } from 'lit'; import { ChatSession } from '../lib/chat-session.js'; import { t, I18nMixin } from '../lib/i18n.js'; import { renderMsg, renderAttachmentChips } from './copilot-render.js'; +import { renderTaskStrip } from './shared/agent-tasks.js'; // Built-in (server-handled) slash commands shown at the top of the composer // autocomplete. Custom commands (from `commands//`) are fetched from @@ -59,6 +60,10 @@ export class AppCopilot extends I18nMixin(ChatSession) { this._onPageChange = this._onPageChange.bind(this); } + // The desktop shell routes `#session/{id}`, so a background task's row links + // through to what it is doing. + get _canOpenTaskSession() { return true; } + connectedCallback() { super.connectedCallback?.(); this._restoreState(); @@ -398,6 +403,7 @@ export class AppCopilot extends I18nMixin(ChatSession) {
+ ${renderTaskStrip(this)} ${this._renderNoModelsBanner()}
e.preventDefault()} diff --git a/web/components/shared/agent-tasks.js b/web/components/shared/agent-tasks.js new file mode 100644 index 0000000..1b6d173 --- /dev/null +++ b/web/components/shared/agent-tasks.js @@ -0,0 +1,98 @@ +import { html, nothing } from 'lit'; +import { t } from '../../lib/i18n.js'; + +/** + * The background-task strip: what the agent handed off to run on its own. + * + * Rendered just above the composer, and only when there is something to say — + * an empty strip would be a permanent slice of chrome for an occasional event. + * It sits outside the message flow on purpose: a task started twenty messages + * ago scrolls away exactly when it matters most, and the transcript is a record + * of what was said, not a dashboard. + * + * State lives on the chat component (`ChatSession._tasks`), which owns the + * WebSocket the updates arrive on; this file only knows how it looks. Shared by + * the desktop copilot and the mobile chat page. + */ + +const STATE_ICON = { + running: 'bi-arrow-repeat agent-task-spin', + completed: 'bi-check-circle-fill', + failed: 'bi-exclamation-triangle-fill', + cancelled: 'bi-slash-circle', +}; + +/** Elapsed time since an ISO timestamp, coarse on purpose (`12s`, `4m 03s`). */ +function elapsed(startedAt) { + if (!startedAt) return ''; + const ms = Date.now() - new Date(startedAt).getTime(); + if (!Number.isFinite(ms) || ms < 0) return ''; + const s = Math.floor(ms / 1000); + if (s < 60) return `${s}s`; + const m = Math.floor(s / 60); + if (m < 60) return `${m}m ${String(s % 60).padStart(2, '0')}s`; + return `${Math.floor(m / 60)}h ${String(m % 60).padStart(2, '0')}m`; +} + +function renderTask(host, task) { + const running = task.state === 'running'; + // The task's own session page is the drill-in: it already shows, live, what a + // background agent is doing. Without it "a task is running" is a fact you can + // do nothing with. + const openable = task.session_id != null && host._canOpenTaskSession; + + return html` +
+ + + + + ${running + ? html`` + : html``} +
+ `; +} + +export function renderTaskStrip(host) { + const tasks = host._tasks ?? []; + if (tasks.length === 0) return nothing; + + const running = tasks.filter(x => x.state === 'running').length; + + return html` +
+
+ + ${running > 0 + ? t('chat.tasks.running_n', { n: running }) + : t('chat.tasks.title')} + ${t('chat.tasks.see_all')} +
+ ${tasks.map(task => renderTask(host, task))} +
+ `; +} diff --git a/web/components/shared/chat-page.js b/web/components/shared/chat-page.js index e276d54..6f09843 100644 --- a/web/components/shared/chat-page.js +++ b/web/components/shared/chat-page.js @@ -2,6 +2,7 @@ import { html, nothing } from 'lit'; import { ChatSession } from '../../lib/chat-session.js'; import { t } from '../../lib/i18n.js'; import { renderMsg, renderAttachmentChips } from '../copilot-render.js'; +import { renderTaskStrip } from './agent-tasks.js'; export class ChatPage extends ChatSession { static properties = { @@ -193,6 +194,7 @@ export class ChatPage extends ChatSession {
+ ${renderTaskStrip(this)} ${this._renderNoModelsBanner()}
e.preventDefault()} diff --git a/web/css/agent-tasks.css b/web/css/agent-tasks.css new file mode 100644 index 0000000..e53d37a --- /dev/null +++ b/web/css/agent-tasks.css @@ -0,0 +1,151 @@ +/* ── Background-task strip ────────────────────────────────────────────────── + The list of `execute_task mode=async` tasks the current conversation started, + rendered above the composer by `renderTaskStrip` (shared by the desktop + copilot and the mobile chat page). Quiet by default: it sits next to the + composer, so it must read as a status line, never as a second UI. */ + +.agent-tasks { + display: flex; + flex-direction: column; + gap: 0.25rem; + margin-bottom: 0.5rem; + padding: 0.4rem 0.45rem 0.45rem; + border: 1px solid var(--toolbar-border); + border-radius: var(--radius-sm); + background: var(--msg-assistant-bg); +} + +.agent-tasks-head { + display: flex; + align-items: center; + gap: 0.4rem; + padding: 0 0.25rem 0.1rem; + font-size: 0.72rem; + font-weight: 600; + letter-spacing: 0.02em; + text-transform: uppercase; + color: var(--placeholder-color); +} + +.agent-tasks-all { + margin-left: auto; + font-size: 0.7rem; + font-weight: 500; + text-transform: none; + letter-spacing: 0; + color: var(--placeholder-color); + text-decoration: none; +} + +.agent-tasks-all:hover { color: var(--accent); text-decoration: underline; } + +/* ── One task ── */ + +.agent-task { + display: flex; + align-items: center; + gap: 0.5rem; + padding: 0.3rem 0.4rem; + border-radius: var(--radius-sm); + background: var(--card-bg); + border: 1px solid transparent; +} + +.agent-task--failed { + border-color: rgba(var(--bs-danger-rgb, 220, 53, 69), 0.35); +} + +.agent-task--completed, +.agent-task--cancelled { opacity: 0.7; } + +.agent-task-icon { + flex-shrink: 0; + font-size: 0.85rem; + color: var(--accent); +} + +.agent-task--completed .agent-task-icon { color: var(--tool-subagent); } +.agent-task--failed .agent-task-icon { color: var(--bs-danger, #dc3545); } +.agent-task--cancelled .agent-task-icon { color: var(--placeholder-color); } + +/* The body is a button: clicking opens the task's own session page. */ +.agent-task-body { + flex: 1; + min-width: 0; + display: flex; + flex-direction: column; + align-items: flex-start; + gap: 0.05rem; + padding: 0; + border: none; + background: none; + text-align: left; + color: inherit; + cursor: pointer; +} + +.agent-task-body:disabled { cursor: default; } + +.agent-task-title { + max-width: 100%; + font-size: 0.8rem; + font-weight: 500; + line-height: 1.3; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.agent-task-body:not(:disabled):hover .agent-task-title { + color: var(--accent); + text-decoration: underline; +} + +.agent-task-meta, +.agent-task-error { + max-width: 100%; + font-size: 0.7rem; + line-height: 1.3; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + color: var(--placeholder-color); +} + +.agent-task-error { color: var(--bs-danger, #dc3545); } + +.agent-task-action { + flex-shrink: 0; + display: flex; + align-items: center; + justify-content: center; + width: 1.5rem; + height: 1.5rem; + padding: 0; + border: none; + border-radius: var(--radius-sm); + background: none; + color: var(--placeholder-color); + cursor: pointer; + transition: background 0.15s, color 0.15s; +} + +.agent-task-action:hover { + background: var(--accent-soft); + color: var(--accent); +} + +/* ── The running spinner ── */ + +.agent-task-spin { + display: inline-block; + animation: agent-task-spin 1.4s linear infinite; +} + +@keyframes agent-task-spin { + to { transform: rotate(360deg); } +} + +@media (prefers-reduced-motion: reduce) { + .agent-task-spin { animation: none; } +} diff --git a/web/i18n/en.js b/web/i18n/en.js index 3505d16..62650a1 100644 --- a/web/i18n/en.js +++ b/web/i18n/en.js @@ -84,6 +84,15 @@ export default { 'chat.mic.unsupported': 'This browser does not support voice recording.', 'chat.mic.denied': 'Microphone access was denied. Allow it for this site in your browser settings, then try again.', 'chat.mic.failed': 'Could not start recording: {error}', + 'chat.tasks.title': 'Background tasks', + 'chat.tasks.running_n': 'Background tasks · {n} running', + 'chat.tasks.open': 'See what this task is doing', + 'chat.tasks.stop': 'Stop this task', + 'chat.tasks.dismiss': 'Remove from the list', + 'chat.tasks.completed': 'done', + 'chat.tasks.failed': 'failed', + 'chat.tasks.cancelled': 'stopped', + 'chat.tasks.see_all': 'All tasks', // ── Copilot render ───────────────────────────────────────────────────────── 'copilot.open_in_viewer': 'Open in viewer', diff --git a/web/i18n/fr.js b/web/i18n/fr.js index 664ec0b..b5805e0 100644 --- a/web/i18n/fr.js +++ b/web/i18n/fr.js @@ -84,6 +84,15 @@ export default { 'chat.mic.unsupported': 'Ce navigateur ne prend pas en charge l\'enregistrement vocal.', 'chat.mic.denied': 'Accès au microphone refusé. Autorisez-le pour ce site dans les réglages du navigateur, puis réessayez.', 'chat.mic.failed': 'Impossible de démarrer l\'enregistrement : {error}', + 'chat.tasks.title': 'Tâches en arrière-plan', + 'chat.tasks.running_n': 'Tâches en arrière-plan · {n} en cours', + 'chat.tasks.open': 'Voir ce que fait cette tâche', + 'chat.tasks.stop': 'Arrêter cette tâche', + 'chat.tasks.dismiss': 'Retirer de la liste', + 'chat.tasks.completed': 'terminée', + 'chat.tasks.failed': 'échouée', + 'chat.tasks.cancelled': 'arrêtée', + 'chat.tasks.see_all': 'Toutes les tâches', // ── Copilot render ───────────────────────────────────────────────────────── 'copilot.open_in_viewer': 'Ouvrir dans le visualiseur', diff --git a/web/i18n/it.js b/web/i18n/it.js index 9005e5f..2b42794 100644 --- a/web/i18n/it.js +++ b/web/i18n/it.js @@ -84,6 +84,15 @@ export default { 'chat.mic.unsupported': 'Questo browser non supporta la registrazione vocale.', 'chat.mic.denied': 'Accesso al microfono negato. Consentilo per questo sito nelle impostazioni del browser e riprova.', 'chat.mic.failed': 'Impossibile avviare la registrazione: {error}', + 'chat.tasks.title': 'Attività in background', + 'chat.tasks.running_n': 'Attività in background · {n} in corso', + 'chat.tasks.open': 'Guarda cosa sta facendo questa attività', + 'chat.tasks.stop': 'Ferma questa attività', + 'chat.tasks.dismiss': 'Togli dalla lista', + 'chat.tasks.completed': 'completata', + 'chat.tasks.failed': 'fallita', + 'chat.tasks.cancelled': 'fermata', + 'chat.tasks.see_all': 'Tutte le attività', // ── Copilot render ───────────────────────────────────────────────────────── 'copilot.open_in_viewer': 'Apri nel visualizzatore', diff --git a/web/index.html b/web/index.html index 8449d27..81616cb 100644 --- a/web/index.html +++ b/web/index.html @@ -45,6 +45,7 @@ + diff --git a/web/lib/chat-session.js b/web/lib/chat-session.js index 380dd71..c0ab34e 100644 --- a/web/lib/chat-session.js +++ b/web/lib/chat-session.js @@ -59,6 +59,11 @@ export class ChatSession extends LightElement { // Pending attachments for the message being composed (shown as chips above // the textarea; uploaded to disk on selection, sent with the next message). _attachments: { state: true }, + // Background tasks (`execute_task mode=async`) this conversation started. + // Each entry: { job_id, title, agent_id, session_id, state, error, started_at }. + _tasks: { state: true }, + // Bumped once a second while a task is running, so the elapsed counters move. + _taskTick: { state: true }, }; // Live events whose arrival implies a turn is in flight (used to restore the @@ -110,6 +115,11 @@ export class ChatSession extends LightElement { // Each entry: { name, path, mimetype, filesize, uploading? }. While an upload // is in flight the entry has `uploading: true` and no `path` yet. this._attachments = []; + this._tasks = []; + this._taskTick = 0; + this._taskTimer = null; + // Timers that drop a finished task from the strip after a grace period. + this._taskDropTimers = new Map(); this._onAuthRestored = this._onAuthRestored.bind(this); } @@ -119,13 +129,16 @@ export class ChatSession extends LightElement { // Fire-and-forget: availability of a transcription provider determines // whether the mic button is rendered at all. this._checkTranscribe(); - await Promise.all([this._loadProviders(), this._loadHistory()]); + await Promise.all([this._loadProviders(), this._loadHistory(), this._loadTasks()]); this._connectWS(); } disconnectedCallback() { super.disconnectedCallback?.(); window.removeEventListener('auth-restored', this._onAuthRestored); + this._stopTaskClock(); + for (const timer of this._taskDropTimers.values()) clearTimeout(timer); + this._taskDropTimers.clear(); } // ── Source identity — override in subclass ──────────────────────────────────── @@ -146,7 +159,8 @@ export class ChatSession extends LightElement { this._activeSource = source; this._messages = []; this._waiting = false; - await this._loadHistory(); + this._tasks = []; + await Promise.all([this._loadHistory(), this._loadTasks()]); this._connectWS(); } @@ -209,6 +223,119 @@ export class ChatSession extends LightElement { } } + // ── Background tasks ────────────────────────────────────────────────────────── + // + // The agent can hand work to a background task (`execute_task mode="async"`) + // and keep talking. Until now the only trace of one was the receipt in the + // transcript and a row on the Tasks page, so "is it still going?" had no + // answer in the chat itself. `_tasks` is that answer: a small live list of + // the tasks *this* conversation started. + // + // Live updates arrive as `task_update` over the WebSocket — a broadcast with + // no replay, which is why `_loadTasks` runs on every load and reconnect: it + // is what makes the strip survive a browser refresh. + + // A finished task lingers this long before disappearing on its own. Its result + // is already in the conversation by then; the chip is just the hand-off. + static TASK_DROP_MS = 20000; + + /** + * Whether this surface can open a task's own session page (`#session/{id}`). + * The mobile shell routes a fixed set of sections and silently falls back to + * the chat for anything else, so there the row is shown without a link rather + * than with one that quietly navigates somewhere wrong. + */ + get _canOpenTaskSession() { return false; } + + // Failures the user has dismissed with the ✕. Kept across reloads (the server + // keeps reporting a recent failure, and dismissing it should mean dismissed). + static _DISMISSED_KEY = 'skald.dismissedTasks'; + + _dismissedTasks() { + try { + return new Set(JSON.parse(localStorage.getItem(ChatSession._DISMISSED_KEY) ?? '[]')); + } catch { return new Set(); } + } + + _rememberDismissed(jobId) { + const ids = [...this._dismissedTasks(), jobId].slice(-50); + try { localStorage.setItem(ChatSession._DISMISSED_KEY, JSON.stringify(ids)); } catch { /* private mode */ } + } + + async _loadTasks() { + try { + const res = await fetch(`/api/${this._source}/tasks`); + if (!res.ok) throw new Error(`HTTP ${res.status}`); + const dismissed = this._dismissedTasks(); + this._tasks = (await res.json()).filter(t => !dismissed.has(t.job_id)); + this._syncTaskClock(); + } catch (e) { + console.warn('Could not load background tasks:', e.message); + } + } + + /** Insert or advance one task from a `task_update` event. */ + _upsertTask(task) { + if (this._dismissedTasks().has(task.job_id)) return; + const idx = this._tasks.findIndex(t => t.job_id === task.job_id); + const prev = idx >= 0 ? this._tasks[idx] : null; + // Keep the fields the event does not carry (a terminal update has no + // `started_at`, and the elapsed counter should not reset at the finish line). + const next = { ...prev, ...task, started_at: task.started_at ?? prev?.started_at ?? null }; + this._tasks = idx >= 0 + ? this._tasks.map((t, i) => (i === idx ? next : t)) + : [...this._tasks, next]; + + // A failure stays until dismissed — it is the only place the reason is + // readable at a glance. Everything else clears itself. + if (next.state === 'completed' || next.state === 'cancelled') { + this._scheduleTaskDrop(next.job_id); + } + this._syncTaskClock(); + } + + _scheduleTaskDrop(jobId) { + clearTimeout(this._taskDropTimers.get(jobId)); + this._taskDropTimers.set(jobId, setTimeout(() => { + this._taskDropTimers.delete(jobId); + this._tasks = this._tasks.filter(t => t.job_id !== jobId); + this._syncTaskClock(); + }, ChatSession.TASK_DROP_MS)); + } + + _dismissTask(jobId) { + this._rememberDismissed(jobId); + clearTimeout(this._taskDropTimers.get(jobId)); + this._taskDropTimers.delete(jobId); + this._tasks = this._tasks.filter(t => t.job_id !== jobId); + this._syncTaskClock(); + } + + /** Stop a running task. The kill lands as a `task_update` like any other end. */ + async _stopTask(jobId) { + try { + const res = await fetch(`/api/cron/jobs/${jobId}/kill`, { method: 'POST' }); + if (!res.ok) throw new Error(await res.text() || `HTTP ${res.status}`); + } catch (e) { + console.warn('Could not stop task:', e.message); + } + } + + /** The 1 s clock runs only while something is actually running. */ + _syncTaskClock() { + const running = this._tasks.some(t => t.state === 'running'); + if (running && !this._taskTimer) { + this._taskTimer = setInterval(() => { this._taskTick++; }, 1000); + } else if (!running) { + this._stopTaskClock(); + } + } + + _stopTaskClock() { + clearInterval(this._taskTimer); + this._taskTimer = null; + } + // ── WebSocket ───────────────────────────────────────────────────────────────── _connectWS() { @@ -224,6 +351,9 @@ export class ChatSession extends LightElement { if (this._reconnecting) { this._reconnecting = false; this._resyncOnReconnect(); + // A task that ended while the socket was down emitted its `task_update` + // into the void: re-read the authoritative list. + this._loadTasks(); } if (this._hasPendingTools) { ws.send(JSON.stringify({ type: 'resume' })); @@ -315,6 +445,9 @@ export class ChatSession extends LightElement { this._cancelStreamFlush(); this._messages = []; this._waiting = false; + // Tasks belong to the conversation that started them; this is a new one. + this._tasks = []; + this._stopTaskClock(); try { const res = await fetch(`/api/sessions?source=${this._source}`, { method: 'POST' }); if (!res.ok) throw new Error(`HTTP ${res.status}`); @@ -605,6 +738,19 @@ export class ChatSession extends LightElement { this._selectedGroup = msg.group; break; + case 'task_update': + // A background task this conversation started changed state. + this._upsertTask({ + job_id: msg.job_id, + title: msg.title, + agent_id: msg.agent_id, + session_id: msg.session_id ?? null, + state: msg.state, + error: msg.error ?? null, + started_at: msg.state === 'running' ? new Date().toISOString() : null, + }); + break; + case 'llm_failed': this._waiting = false; this._dropStreaming(); diff --git a/web/mobile.html b/web/mobile.html index 0c0c580..6f67442 100644 --- a/web/mobile.html +++ b/web/mobile.html @@ -43,6 +43,7 @@ +