dguiducci/Skald-Circle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: show a conversation its own background tasks, and give it back every outcome
Nightly Build / build (push) Successful in 7m34s
Details

Browse Source 
An `execute_task mode="async"` was invisible from the chat that started it.
The only trace was the receipt in the transcript and a row on the Tasks page
— which does not say *which* of those rows the assistant just spawned — so
"is it still going?" had no answer where the question is asked.

Worse, a task that did not simply succeed never came back at all. `run_job`
branched on `Ok`/`Err` first and routed by `job.kind` only inside the `Ok`
arm, so a failure or a kill left through the `Err` arm's unconditional
`hub.notify` — the home source (`/sethome`), worded "Cron job … failed" —
while the parent conversation sat waiting for a `task_completed` that would
never arrive. The wrong chat, and a wedged one.

The fix is a shape, not a branch: one `JobOutcome` classification, then one
`match job.kind` delivery site for every ending. An async task now ends in
its parent conversation whatever happened to it. The sink has a single
channel deliberately — to the model reading it, "it broke" is a result like
any other and must not be overlookable — so a failure is delivered as prose,
carrying whatever partial output the run produced, which is usually the only
clue about why. A cron job keeps the home notification: it belongs to nobody's
conversation. Cancellation becomes a third outcome rather than a flavour of
failure (`job_runs.status` has always had `'cancelled'` in its CHECK and
nothing ever wrote it), classified off the new typed `TurnCancelled` error so
nothing keys on a message string.

The strip above the composer is the visible half. `ServerEvent::TaskUpdate`
announces state to the source of the parent conversation only; the list is
`renderTaskStrip` (shared by the desktop copilot and the mobile chat), fed by
state on `ChatSession`. Each row links to `#session/{id}` — the page that
already shows, live, what a background agent is doing, and without which
"a task is running" is a fact you can do nothing with. Stopping is the
existing kill endpoint. A finished row clears itself after 20 s (its result
is in the conversation by then); a failed one stays until dismissed, and the
dismissal is remembered across reloads.

`GET /api/{source}/tasks` is what makes the strip survive a browser refresh:
the event is a broadcast with no replay, so without a load-time read a reload
would empty a chat that still has work running under it. It answers with the
running tasks plus failures from the last 30 minutes — the two states a person
can still act on. Successes are absent on purpose. Its window compares through
`datetime()` on both sides: `completed_at` is RFC 3339 and the cutoff is
SQLite-shaped, and `'T' > ' '` would let every same-day row through a window
meant to exclude it.

Not addressed, and worth doing next: a cron job's result should go where its
creator says, not always to the home chat.
This commit is contained in:
dguiducci
2026-08-04 19:13:30 +01:00
parent e356741435
commit daaceff6ba
21 changed files with 949 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/skald-core/src/cron/mod.rs
+199 -69
View File
@@ -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::<TurnCancelled>().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<String> {
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<ChatHub>>,
job: &ScheduledJob,
session_id: Option<i64>,
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<SqlitePool>,
hub: &Arc<ChatHub>,
crates/skald-core/src/db/scheduled_jobs.rs
+169
View File
@@ -81,6 +81,90 @@ pub async fn list_interrupted(pool: &SqlitePool) -> Result<Vec<ScheduledJob>> {
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<i64>,
/// `running` or `failed` — the only two states this query returns.
pub state: String,
pub error: Option<String>,
/// When it started, normalised to RFC 3339 (see [`normalise_ts`]).
pub started_at: Option<String>,
}
/// 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<Vec<SessionTask>> {
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<String> {
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<i64>, running: Option<i64>| {
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));
}
}
}
crates/skald-core/src/session/handler/mod.rs
+19 -1
View File
@@ -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");