Treat intentional cancellation as a non-failure terminal

This commit is contained in:
2026-10-01 17:04:05 +09:00
parent 0ec2b3ad86
commit 64ed50f6c0
17 changed files with 256 additions and 15 deletions
+6
View File
@@ -991,6 +991,9 @@ pub enum SessionSnapshotEntryData {
source: RunResumeSource,
active_run_turn_count: usize,
},
/// Durable terminal for an intentional cancellation. Consumers must not
/// present this lifecycle result as an execution failure.
RunCancelled,
RunError {
message: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -1737,6 +1740,9 @@ pub enum RunResult {
Finished,
Paused,
LimitReached,
/// The logical Run ended because the user intentionally cancelled it after
/// observable work. Unlike `RolledBack`, committed history is retained.
Cancelled,
/// The accepted Method::Submit produced no assistant/tool output before
/// user interruption, so the Worker rolled the submit-time turn state back
/// to its pre-submit snapshot. Clients should treat the Worker as Idle and
@@ -1075,6 +1075,17 @@ fn project_session_snapshot_for_segment(
},
));
}
LogEntry::RunCancelled { ts } => {
active_run_turn_count = None;
entries.push(legacy_entry(
&session_key,
segment_id.as_ref(),
log_index,
0,
*ts,
SessionSnapshotEntryData::RunCancelled,
));
}
LogEntry::RunErrored {
ts,
message,
@@ -2849,6 +2860,27 @@ mod tests {
));
}
#[test]
fn cancelled_run_projects_as_a_non_failure_terminal() {
let snapshot = project_session_snapshot(
crate::new_session_id(),
&[
LogEntry::Invoke {
ts: 10,
trigger: protocol::InvokeKind::UserSend,
},
LogEntry::RunCancelled { ts: 20 },
],
);
assert_eq!(snapshot.entries.len(), 1);
assert_eq!(snapshot.entries[0].timestamp, 20);
assert!(matches!(
snapshot.entries[0].data,
SessionSnapshotEntryData::RunCancelled
));
}
#[test]
fn retained_snapshot_read_projects_active_segment_without_rewriting_files() {
let root = tempfile::tempdir().unwrap();
+32 -4
View File
@@ -28,7 +28,7 @@ use crate::logged_item::LoggedItem;
/// `run()`/`resume()` (current callers persist a single TurnEnd at
/// run completion); the fork-point seq for `at_turn_index` is the
/// preceding `Invoke` entry, not the TurnEnd.
/// - `RunCompleted` / `RunErrored` — marks end of a `run()` or `resume()` call
/// - `RunCompleted` / `RunCancelled` / `RunErrored` — terminal run outcomes
/// - `PausedTurnAbandoned` — explicit abandon/cancel of a paused interrupted turn
/// - `ConfigChanged` — `RequestConfig` mutation
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -86,9 +86,10 @@ pub enum LogEntry {
/// serde tag already occupies `"kind"`.
///
/// Replay marks the run interrupted until a terminal `RunCompleted`,
/// `RunErrored`, or `PausedTurnAbandoned` entry proves how it ended. This
/// makes a process/disk failure between Invoke and its terminal record
/// restore conservatively instead of re-running a dangling tool call.
/// `RunCancelled`, `RunErrored`, or `PausedTurnAbandoned` entry proves how
/// it ended. This makes a process/disk failure between Invoke and its
/// terminal record restore conservatively instead of re-running a dangling
/// tool call.
Invoke { ts: u64, trigger: InvokeKind },
/// Canonical user submission with its exact model-visible entries. Typed
@@ -152,6 +153,10 @@ pub enum LogEntry {
active_run_turn_count: usize,
},
/// The logical Run ended because the user intentionally cancelled it.
/// This is terminal, but is not an execution failure.
RunCancelled { ts: u64 },
/// `run()` / `resume()` が `EngineError` で終了した。
/// `EngineError` は `Serialize` 不可なので `message` のみ lossy 保持する。
/// Audit-only metadata: replay は `interrupted` のみ反映する。
@@ -389,6 +394,10 @@ pub fn collect_state(entries: &[LogEntry]) -> RestoredState {
state.last_run_interrupted = true;
state.active_run_turn_count = Some(*active_run_turn_count);
}
LogEntry::RunCancelled { .. } => {
state.last_run_interrupted = false;
state.active_run_turn_count = None;
}
LogEntry::RunErrored { interrupted, .. } => {
state.last_run_interrupted = *interrupted;
state.active_run_turn_count = None;
@@ -889,6 +898,25 @@ mod tests {
assert_eq!(state.active_run_turn_count, Some(2));
}
#[test]
fn replay_cancelled_run_is_terminal_without_becoming_a_failure() {
let state = collect_state(&[
LogEntry::Invoke {
ts: 100,
trigger: InvokeKind::UserSend,
},
LogEntry::TurnEnd {
ts: 200,
turn_count: 1,
},
LogEntry::RunCancelled { ts: 300 },
]);
assert!(!state.last_run_interrupted);
assert_eq!(state.active_run_turn_count, None);
assert_eq!(state.turn_count, 1);
}
#[test]
fn legacy_yielded_run_completed_remains_resumable_but_is_not_terminal_success() {
let entry: LogEntry = serde_json::from_value(serde_json::json!({
+3 -2
View File
@@ -2545,7 +2545,8 @@ impl App {
}
}
SessionSnapshotEntryData::RunYielded { .. }
| SessionSnapshotEntryData::RunResumed { .. } => {
| SessionSnapshotEntryData::RunResumed { .. }
| SessionSnapshotEntryData::RunCancelled => {
// Durable logical-Run transitions are intentionally not
// presentation rows and never restore runtime progress.
}
@@ -3536,7 +3537,7 @@ mod completion_flow_tests {
#[test]
fn non_rolled_back_run_end_keeps_submitted_blocks_and_does_not_restore_input() {
for result in [RunResult::Paused, RunResult::Finished] {
for result in [RunResult::Paused, RunResult::Finished, RunResult::Cancelled] {
let mut app = App::new("test".into());
let submitted = submit_text(&mut app, "normal run");
app.handle_worker_event(Event::UserMessage {
+1
View File
@@ -66,6 +66,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
WorkerRunResult::Finished => println!("(finished)"),
WorkerRunResult::Paused => println!("(paused)"),
WorkerRunResult::LimitReached => println!("(turn limit reached)"),
WorkerRunResult::Cancelled => println!("(cancelled)"),
WorkerRunResult::Interrupted { message, .. } => println!("(interrupted: {message})"),
WorkerRunResult::RolledBack => println!("(empty turn rolled back)"),
}
+6
View File
@@ -2794,6 +2794,12 @@ where
WorkerRunResult::Paused => (WorkerStatus::Paused, RunResult::Paused),
WorkerRunResult::LimitReached => (WorkerStatus::Idle, RunResult::LimitReached),
WorkerRunResult::RolledBack => (WorkerStatus::Idle, RunResult::RolledBack),
WorkerRunResult::Cancelled if pause_requested => {
(WorkerStatus::Paused, RunResult::Paused)
}
WorkerRunResult::Cancelled => {
(WorkerStatus::Idle, RunResult::Cancelled)
}
WorkerRunResult::Interrupted { .. } if pause_requested => {
let _ = working_event_tx.send(Event::RunEnd { result: RunResult::Paused });
return (WorkerStatus::Paused, shutdown_requested, false);
@@ -632,7 +632,7 @@ fn extraction_disposition(
ExtractionDisposition::Failed(error.source.to_string())
}
Ok(run) => match &run.lifecycle {
WorkerRunResult::RolledBack => {
WorkerRunResult::Cancelled | WorkerRunResult::RolledBack => {
ExtractionDisposition::Cancelled("memory-extract cancelled".to_string())
}
WorkerRunResult::Interrupted { message, .. } => {
+5
View File
@@ -213,6 +213,7 @@ where
let outcome = match worker.run_text(&input).await {
Ok(lifecycle @ WorkerRunResult::Finished)
| Ok(lifecycle @ WorkerRunResult::Paused)
| Ok(lifecycle @ WorkerRunResult::Cancelled)
| Ok(lifecycle @ WorkerRunResult::RolledBack) => Ok(InternalWorkerResult {
usage: last_usage.lock().ok().and_then(|slot| slot.clone()),
identity,
@@ -333,6 +334,10 @@ fn classify_internal_turn_result(
Ok(WorkerRunResult::Interrupted { message, .. }) => {
(InternalWorkerSessionStatus::Stopped, Some(message))
}
Ok(WorkerRunResult::Cancelled) => (
InternalWorkerSessionStatus::Stopped,
Some("internal Worker run was cancelled".to_string()),
),
Ok(WorkerRunResult::RolledBack) => (
InternalWorkerSessionStatus::Stopped,
Some("internal Worker run was cancelled before AI output".to_string()),
@@ -64,6 +64,7 @@ pub fn live_log_entry_event(entry: LogEntry) -> Option<Event> {
| LogEntry::AnnotatedToolResult { .. }
| LogEntry::RunYielded { .. }
| LogEntry::RunResumed { .. }
| LogEntry::RunCancelled { .. }
| LogEntry::RunErrored { .. }) => {
let mut projected =
session_store::public_snapshot::project_current_session_snapshot(&[entry]);
+1
View File
@@ -125,6 +125,7 @@ impl SegmentLogSink {
| LogEntry::Invoke { .. }
| LogEntry::RunYielded { .. }
| LogEntry::RunResumed { .. }
| LogEntry::RunCancelled { .. }
| LogEntry::RunErrored { .. }
)
}
+1
View File
@@ -287,6 +287,7 @@ impl SessionCapture {
SessionSnapshotEntryData::SystemItem { .. }
| SessionSnapshotEntryData::RunYielded { .. }
| SessionSnapshotEntryData::RunResumed { .. }
| SessionSnapshotEntryData::RunCancelled
| SessionSnapshotEntryData::RunError { .. } => return None,
};
Some(CapturedHistoryEntry {
+57 -2
View File
@@ -4974,6 +4974,10 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
self.last_run_interrupted = false;
Ok(WorkerRunResult::LimitReached)
}
EngineRunExit::Interrupted(RunInterruptionReason::Cancelled) => {
self.last_run_interrupted = false;
Ok(WorkerRunResult::Cancelled)
}
EngineRunExit::Interrupted(reason) => {
self.last_run_interrupted = true;
Ok(WorkerRunResult::Interrupted {
@@ -5446,6 +5450,11 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
active_run_turn_count,
})?;
}
EngineRunExit::Interrupted(RunInterruptionReason::Cancelled) => {
self.commit_entry(LogEntry::RunCancelled {
ts: segment_log::now_millis(),
})?;
}
EngineRunExit::Interrupted(reason) => {
self.commit_entry(LogEntry::RunErrored {
ts: segment_log::now_millis(),
@@ -7376,6 +7385,8 @@ pub enum WorkerRunResult {
Paused,
/// The worker reached its configured max_turns limit.
LimitReached,
/// The logical Run was intentionally cancelled after observable work.
Cancelled,
/// The run was interrupted by a known or unexpected terminal cause.
Interrupted { code: ErrorCode, message: String },
/// The submit-time user turn was rolled back because no AI output was materialized.
@@ -9335,6 +9346,48 @@ mod build_summary_prompt_tests {
assert_eq!(worker.engine().active_run_turn_count(), None);
}
#[tokio::test]
async fn intentional_cancel_commits_a_non_failure_terminal() {
let dir = tempfile::tempdir().unwrap();
let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap();
let mut worker = Worker::new(
minimal_manifest(),
Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient),
store.clone(),
WorkerWorkspaceContext::no_workspace(),
WorkerFilesystemAuthority::None,
Scope::empty(),
)
.await
.unwrap();
worker.ensure_segment_head().await.unwrap();
worker.engine_mut().set_active_run_turn_count(Some(1));
let result = worker
.handle_worker_result(
EngineRunExit::Interrupted(RunInterruptionReason::Cancelled),
0,
)
.await
.unwrap();
assert_eq!(result, WorkerRunResult::Cancelled);
let entries = store
.read_all(worker.session_id(), worker.segment_id())
.unwrap();
assert!(
entries
.iter()
.any(|entry| matches!(entry, LogEntry::RunCancelled { .. }))
);
assert!(
!entries
.iter()
.any(|entry| matches!(entry, LogEntry::RunErrored { .. }))
);
assert!(!worker.last_run_interrupted);
}
#[tokio::test]
async fn fresh_run_clears_interrupted_budget_before_pre_run_compaction() {
let dir = tempfile::tempdir().unwrap();
@@ -10522,7 +10575,7 @@ mod build_summary_prompt_tests {
}
#[tokio::test]
async fn interrupted_result_terminalizes_orphan_before_run_completed() {
async fn cancelled_result_terminalizes_orphan_before_run_terminal() {
let dir = tempfile::tempdir().unwrap();
let manifest = minimal_manifest();
let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap();
@@ -10631,7 +10684,9 @@ mod build_summary_prompt_tests {
.position(|entry| {
matches!(
entry,
LogEntry::RunCompleted { .. } | LogEntry::RunErrored { .. }
LogEntry::RunCompleted { .. }
| LogEntry::RunCancelled { .. }
| LogEntry::RunErrored { .. }
)
})
.expect("durable final run status");
+65
View File
@@ -3693,6 +3693,71 @@ async fn empty_turn_rollback_removes_only_the_most_recent_turn() {
);
}
#[tokio::test]
async fn cancel_after_assistant_output_is_a_non_failure_terminal() {
let client = MockClient::sequential(vec![MockResponse::Hang(vec![
LlmEvent::text_block_start(0),
LlmEvent::text_delta(0, "committed before cancel"),
LlmEvent::text_block_stop(0, None),
])]);
let worker = make_worker(client).await;
let handle = spawn_controller(worker).await;
let mut rx = handle.subscribe();
handle
.send(Method::submit_text(
protocol::new_submission_request_id(),
"cancel this active turn",
))
.await
.unwrap();
assert!(
drain_until(
&mut rx,
std::time::Duration::from_secs(2),
|event| matches!(event, Event::TextDone { .. })
)
.await,
"assistant output should be durable before cancellation"
);
handle
.send(Method::Cancel {
command: worker_command(&handle),
})
.await
.unwrap();
assert!(
drain_until(
&mut rx,
std::time::Duration::from_secs(2),
|event| matches!(
event,
Event::RunEnd {
result: protocol::RunResult::Cancelled
}
)
)
.await,
"intentional cancellation must emit RunEnd::Cancelled, not Event::Error"
);
wait_for_status(&handle, WorkerStatus::Idle).await;
let (entries, _) = handle.sink.subscribe_with_snapshot();
assert!(
entries
.iter()
.any(|entry| matches!(entry, LogEntry::RunCancelled { .. })),
"cancelled run must keep a durable non-failure terminal: {entries:?}"
);
assert!(
!entries
.iter()
.any(|entry| matches!(entry, LogEntry::RunErrored { .. })),
"intentional cancellation must not be restored as a run error: {entries:?}"
);
}
#[tokio::test]
async fn pause_after_assistant_token_does_not_rollback() {
let client = MockClient::sequential(vec![MockResponse::Hang(vec![
File diff suppressed because one or more lines are too long
+3 -3
View File
@@ -40,7 +40,7 @@ export type TurnResult = "finished" | "paused";
export type InvokeKind = "user_send" | "notify" | "worker_event" | "system_reminder" | "wakeup";
export type RunResult = "finished" | "paused" | "limit_reached" | "rolled_back";
export type RunResult = "finished" | "paused" | "limit_reached" | "cancelled" | "rolled_back";
export type ToolResultDisposition = "success" | "error" | "interrupted" | "cancelled" | "outcome_unknown";
@@ -142,7 +142,7 @@ export type RunResumeSource = "compaction" | "pause";
export type RunFailureKind = "engine" | "compaction";
export type SessionSnapshotEntryData = { "kind": "user_input", segments: Array<Segment>, } | { "kind": "message", role: SessionMessageRole, content: Array<SessionContentPart>, } | { "kind": "tool_call", call_id: string, name: string, arguments: string, } | { "kind": "tool_result", call_id: string, summary: string, content?: string | null, is_error: boolean, attachments?: Array<SessionToolAttachment>, } | { "kind": "system_item", item_kind: string, content: string, data?: unknown, } | { "kind": "run_yielded", reason: RunYieldReason, active_run_turn_count: number, } | { "kind": "run_resumed", source: RunResumeSource, active_run_turn_count: number, } | { "kind": "run_error", message: string, failure?: RunFailureKind | null, };
export type SessionSnapshotEntryData = { "kind": "user_input", segments: Array<Segment>, } | { "kind": "message", role: SessionMessageRole, content: Array<SessionContentPart>, } | { "kind": "tool_call", call_id: string, name: string, arguments: string, } | { "kind": "tool_result", call_id: string, summary: string, content?: string | null, is_error: boolean, attachments?: Array<SessionToolAttachment>, } | { "kind": "system_item", item_kind: string, content: string, data?: unknown, } | { "kind": "run_yielded", reason: RunYieldReason, active_run_turn_count: number, } | { "kind": "run_resumed", source: RunResumeSource, active_run_turn_count: number, } | { "kind": "run_cancelled" } | { "kind": "run_error", message: string, failure?: RunFailureKind | null, };
export type SessionSnapshotEntry = {
/**
@@ -153,7 +153,7 @@ entry_id: string,
/**
* Timestamp copied from the durable log record that commits this entry.
*/
timestamp: number, provenance: SessionEntryProvenance, derived_from?: Array<string>, } & ({ "kind": "user_input", segments: Array<Segment>, } | { "kind": "message", role: SessionMessageRole, content: Array<SessionContentPart>, } | { "kind": "tool_call", call_id: string, name: string, arguments: string, } | { "kind": "tool_result", call_id: string, summary: string, content?: string | null, is_error: boolean, attachments?: Array<SessionToolAttachment>, } | { "kind": "system_item", item_kind: string, content: string, data?: unknown, } | { "kind": "run_yielded", reason: RunYieldReason, active_run_turn_count: number, } | { "kind": "run_resumed", source: RunResumeSource, active_run_turn_count: number, } | { "kind": "run_error", message: string, failure?: RunFailureKind | null, });
timestamp: number, provenance: SessionEntryProvenance, derived_from?: Array<string>, } & ({ "kind": "user_input", segments: Array<Segment>, } | { "kind": "message", role: SessionMessageRole, content: Array<SessionContentPart>, } | { "kind": "tool_call", call_id: string, name: string, arguments: string, } | { "kind": "tool_result", call_id: string, summary: string, content?: string | null, is_error: boolean, attachments?: Array<SessionToolAttachment>, } | { "kind": "system_item", item_kind: string, content: string, data?: unknown, } | { "kind": "run_yielded", reason: RunYieldReason, active_run_turn_count: number, } | { "kind": "run_resumed", source: RunResumeSource, active_run_turn_count: number, } | { "kind": "run_cancelled" } | { "kind": "run_error", message: string, failure?: RunFailureKind | null, });
export type SessionConversationTurn = { turn_id: string, entries: Array<SessionSnapshotEntry>, };
@@ -134,6 +134,7 @@ function canonicalSession(logEntries: unknown[]): Event extends {
}
case "run_yielded":
case "run_resumed":
case "run_cancelled":
entries.push({
entry_id: `legacy-test-${sequence++}`,
provenance: "legacy_unknown",
@@ -395,6 +396,43 @@ Deno.test("snapshot restores compaction terminal without reviving transition pro
assertEquals(projection.lines[0].kind, "error");
});
Deno.test("intentional cancellation stays non-error in live and snapshot projections", () => {
const live = projectConsole([
{
eventId: "cancelled-terminal",
event: {
event: "session_entry_committed",
data: {
entry: {
entry_id: "cancelled-entry",
timestamp: 10,
provenance: "legacy_unknown",
kind: "run_cancelled",
},
},
} as Event,
},
{
eventId: "cancelled-run-end",
event: {
event: "run_end",
data: { result: "cancelled" },
} as Event,
},
]);
assertEquals(live.lines.some((line) => line.kind === "error"), false);
const restored = projectConsole([{
eventId: "snapshot",
event: snapshotEvent("/repo", [{
kind: "run_cancelled",
timestamp: 10,
}]),
}]);
assertEquals(restored.lines.some((line) => line.kind === "error"), false);
assertEquals(restored.compaction, null);
});
Deno.test("snapshot replaces a live error with one durable run_errored row", () => {
const projector = createConsoleProjector();
let projection = projector.append([
@@ -2190,8 +2190,9 @@ function applySessionEntry(
}
case "run_yielded":
case "run_resumed":
// Durable logical-Run transitions are replayable state, not Console
// progress rows. Runtime compaction progress remains the spinner authority.
case "run_cancelled":
// Durable logical-Run transitions and intentional cancellation are
// replayable lifecycle state, not errors or Console progress rows.
break;
case "run_error":
appendDurableRunFailure(