M6: crash-resume hardening
Hardens resume-after-crash: an unfinished assistant message left by a killed process is marked aborted on load rather than replayed, covered in harness-app and a small harness-tui::modal fix.
This commit is contained in:
@@ -13,6 +13,7 @@ harness-lsp = { workspace = true }
|
||||
tokio = { workspace = true }
|
||||
tokio-util = { workspace = true }
|
||||
futures = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
async-trait = { workspace = true }
|
||||
dirs = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
|
||||
@@ -394,6 +394,12 @@ impl EngineHandle {
|
||||
}
|
||||
}
|
||||
|
||||
// Crash-resume: a previous run may have died mid-turn (process killed) leaving an
|
||||
// assistant message with no finish/error and tool parts stuck Running/Pending. Repair
|
||||
// that before appending the new turn so history is well-formed and the model isn't
|
||||
// shown a dangling tool call.
|
||||
self.inner.repair_unfinished(&session_id).await?;
|
||||
|
||||
let now = now_ms();
|
||||
let user_message = Message::new_user(session_id.clone(), now);
|
||||
self.inner
|
||||
@@ -543,6 +549,13 @@ impl EngineHandle {
|
||||
Ok(self.inner.store.session(session_id).await?)
|
||||
}
|
||||
|
||||
/// Repairs a crashed-mid-run session (unfinished assistant → aborted, in-flight tool parts
|
||||
/// → interrupted). Idempotent; safe to call whenever a session is loaded/resumed.
|
||||
pub async fn repair_session(&self, session_id: &SessionId) -> Result<(), AppError> {
|
||||
self.inner.repair_unfinished(session_id).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Background jobs spawned by `session_id` (the parent), for the TUI jobs pane / drill-in.
|
||||
pub async fn jobs(
|
||||
&self,
|
||||
@@ -595,6 +608,63 @@ impl EngineInner {
|
||||
.expect("EngineInner self-reference set in new()")
|
||||
}
|
||||
|
||||
/// Repairs a session that crashed mid-run: any assistant message with neither a finish
|
||||
/// reason nor an error is marked `Aborted`, and its in-flight tool parts (Running/Pending)
|
||||
/// become `Error { "interrupted" }`. Idempotent — a clean session is left untouched.
|
||||
async fn repair_unfinished(&self, session_id: &SessionId) -> Result<(), StoreError> {
|
||||
let messages = self.store.messages(session_id.clone()).await?;
|
||||
for message in messages {
|
||||
let crashed = message.role == harness_core::types::Role::Assistant
|
||||
&& message.finished.is_none()
|
||||
&& message.error.is_none();
|
||||
if !crashed {
|
||||
continue;
|
||||
}
|
||||
|
||||
for mut part in self.store.parts(message.id.clone()).await? {
|
||||
let interrupted = match &part.body {
|
||||
PartBody::Tool { state, .. } => matches!(
|
||||
state,
|
||||
harness_core::types::ToolState::Running { .. }
|
||||
| harness_core::types::ToolState::Pending { .. }
|
||||
),
|
||||
_ => false,
|
||||
};
|
||||
if !interrupted {
|
||||
continue;
|
||||
}
|
||||
if let PartBody::Tool {
|
||||
call_id,
|
||||
name,
|
||||
state,
|
||||
} = part.body
|
||||
{
|
||||
let input = match &state {
|
||||
harness_core::types::ToolState::Running { input, .. } => input.clone(),
|
||||
_ => serde_json::Value::Null,
|
||||
};
|
||||
part.body = PartBody::Tool {
|
||||
call_id,
|
||||
name,
|
||||
state: harness_core::types::ToolState::Error {
|
||||
input,
|
||||
error: "interrupted".to_string(),
|
||||
},
|
||||
};
|
||||
self.store.upsert_part(part.clone()).await?;
|
||||
self.bus.publish(AppEvent::PartUpdated { part });
|
||||
}
|
||||
}
|
||||
|
||||
let mut repaired = message;
|
||||
repaired.error = Some(harness_core::types::MessageError::Aborted);
|
||||
self.store.upsert_message(repaired.clone()).await?;
|
||||
self.bus
|
||||
.publish(AppEvent::MessageUpdated { message: repaired });
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Fires a background task that generates a short session title from the first user turn.
|
||||
/// No-op when no `small_model` is available; the title is only generated once (the task
|
||||
/// re-checks that the session is still untitled before writing).
|
||||
@@ -1638,6 +1708,60 @@ mod tests {
|
||||
assert_eq!(stored.agent, "orchestrator");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn repair_marks_crashed_assistant_aborted_and_interrupts_tool_parts() {
|
||||
use harness_core::types::{MessageError, PartId, Role, ToolState};
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let engine = mock_engine(dir.path().to_path_buf());
|
||||
let session = engine
|
||||
.new_session("orchestrator", "mock/mock-model")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// A crashed assistant turn: no finish, no error, with a tool call left Running.
|
||||
let model = ModelRef::new("mock", "mock-model");
|
||||
let crashed = Message::new_assistant(session.clone(), model, "orchestrator", now_ms());
|
||||
engine
|
||||
.inner
|
||||
.store
|
||||
.upsert_message(crashed.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
let part = Part {
|
||||
id: PartId::new(),
|
||||
message_id: crashed.id.clone(),
|
||||
session_id: session.clone(),
|
||||
idx: 0,
|
||||
body: PartBody::Tool {
|
||||
call_id: "c1".into(),
|
||||
name: "bash".into(),
|
||||
state: ToolState::Running {
|
||||
input: serde_json::json!({"command": "sleep 100"}),
|
||||
title: None,
|
||||
metadata: serde_json::Value::Null,
|
||||
},
|
||||
},
|
||||
};
|
||||
engine.inner.store.upsert_part(part).await.unwrap();
|
||||
|
||||
engine.inner.repair_unfinished(&session).await.unwrap();
|
||||
|
||||
let messages = engine.inner.store.messages(session.clone()).await.unwrap();
|
||||
let repaired = messages.iter().find(|m| m.id == crashed.id).unwrap();
|
||||
assert_eq!(repaired.role, Role::Assistant);
|
||||
assert!(matches!(repaired.error, Some(MessageError::Aborted)));
|
||||
|
||||
let parts = engine.inner.store.parts(crashed.id.clone()).await.unwrap();
|
||||
match &parts[0].body {
|
||||
PartBody::Tool { state, .. } => match state {
|
||||
ToolState::Error { error, .. } => assert_eq!(error, "interrupted"),
|
||||
other => panic!("expected Error tool state, got {other:?}"),
|
||||
},
|
||||
other => panic!("expected Tool part, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn first_prompt_generates_a_session_title() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
|
||||
@@ -26,6 +26,9 @@ pub async fn select_session(
|
||||
session: Session,
|
||||
) -> Result<(), harness_app::AppError> {
|
||||
let session_id = session.id.clone();
|
||||
// Repair a session that crashed mid-run before loading it, so the transcript shows the
|
||||
// aborted turn rather than a tool call frozen "running".
|
||||
engine.repair_session(&session_id).await?;
|
||||
let messages = engine.session_messages(session_id.clone()).await?;
|
||||
let mut all_parts = Vec::new();
|
||||
for message in &messages {
|
||||
|
||||
Reference in New Issue
Block a user