1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
|
--- a/crates/airouter-agentd/src/session_service/driver.rs
+++ b/crates/airouter-agentd/src/session_service/driver.rs
@@ -1,6 +1,6 @@
use super::*;
use airouter_agent_client::journal::Journal;
-use airouter_agent_protocol::{Command, task};
+use airouter_agent_protocol::{Command, Phase, Usage, controller, session, task};
use std::time::Duration;
impl<B: Backend> Service<B> {
@@ -37,6 +37,13 @@
}
let mut run = prepared.run.clone();
let mut error = None;
+ if saved.status.cancel_requested && !saved.attempted && run.phase == Phase::Uncertain {
+ // Reservation can precede service admission. The fsynced cancel
+ // record and never-attempted marker prove no host was polled, so
+ // they are sufficient to redeliver this exact zero-effect outcome.
+ // Attempted jobs must still recover their original host receipts.
+ run = self.finish_unattempted_cancellation(&prepared).await?;
+ }
if !run.phase.terminal() {
let journal = Journal::new(self.store.root())?;
match managed_session::reconcile_saved(&self.client, &prepared, &journal).await {
@@ -68,6 +75,52 @@
})?;
Ok(record.status)
}
+
+ async fn finish_unattempted_cancellation(
+ &self,
+ prepared: &controller::Prepared,
+ ) -> Result<Run, String> {
+ let run = &prepared.run;
+ let checkpoint = run.session.as_ref().ok_or("session checkpoint missing")?;
+ if run.failure.as_deref() != Some("cancel_requested")
+ || !checkpoint.responses.is_empty()
+ || checkpoint.outcome.is_some()
+ || run.usage.model_calls != 0
+ || run.usage.tool_calls != 0
+ || run.usage.input_tokens != 0
+ || run.usage.output_tokens != 0
+ || run.usage.cost_micro_usd != 0
+ || run.usage.cached_input_tokens.unwrap_or(0) != 0
+ || run.usage.cache_write_input_tokens.unwrap_or(0) != 0
+ {
+ return Err("unattempted cancellation conflicts with execution evidence".into());
+ }
+ self.client.observe_host(run.id, &controller::Observe {
+ lease_id: prepared.binding.lease_id,
+ host: prepared.binding.host.clone(),
+ expected_revision: run.revision,
+ operation_id: checkpoint.operation.id,
+ digest: checkpoint.operation.digest.clone(),
+ observation: controller::Observation::Session {
+ observation: session::Observation::Finished {
+ outcome: session::Outcome {
+ final_response_id: None,
+ error: Some("session cancelled before execution".into()),
+ usage: Usage {
+ coverage: Some(session::AccountingCoverage {
+ model_calls: true, tool_calls: true, input_tokens: true,
+ output_tokens: true, cost: true,
+ }),
+ cached_input_tokens: Some(0),
+ cache_write_input_tokens: Some(0),
+ ..Usage::default()
+ },
+ },
+ },
+ },
+ }).await
+ }
+
pub(super) async fn drive(
&self,
id: Uuid,
--- a/crates/airouter-agentd/tests/session_service/mod.rs
+++ b/crates/airouter-agentd/tests/session_service/mod.rs
@@ -303,6 +303,7 @@
async fn restart_with_unfinished_claim_preserves_uncertainty_and_never_executes() {
let f = Fixture::new().await;
let prepared = f.managed_session().await;
+ wait_for(async || f.client.get(prepared.run.id).await.unwrap().phase == Phase::SessionInFlight).await;
let dir = tempfile::Builder::new().permissions(std::fs::Permissions::from_mode(0o700)).tempdir().unwrap();
let journal = Journal::new(dir.path()).unwrap();
journal
@@ -327,6 +328,10 @@
reopened.status(request.binding.run_id).unwrap().state,
State::ReconciliationRequired
);
+ let cancelled = reopened.cancel(request.binding.run_id).await.unwrap();
+ assert_eq!(cancelled.state, State::ReconciliationRequired);
+ assert_eq!(cancelled.phase, Phase::Uncertain);
+ assert!(!cancelled.activity_complete);
assert!(
reopened
.status(request.binding.run_id)
@@ -450,6 +455,7 @@
async fn recovered_cancel_before_attempt_has_complete_empty_activity() {
let f = Fixture::new().await;
let prepared = f.managed_session().await;
+ wait_for(async || f.client.get(prepared.run.id).await.unwrap().phase == Phase::SessionInFlight).await;
let dir = tempfile::Builder::new().permissions(std::fs::Permissions::from_mode(0o700)).tempdir().unwrap();
let request = submit(&prepared,dir.path());
// Durable accepted admission recovered before the first driver poll.
@@ -462,6 +468,13 @@
})).unwrap()).unwrap();
std::fs::set_permissions(&path,std::fs::Permissions::from_mode(0o600)).unwrap();
let host=adapter(None);
+ let service=Service::open(f.client.clone(),dir.path(),host.clone(),options()).unwrap();
+ f.unavailable.store(1, Ordering::SeqCst);
+ service.recover().await.unwrap();
+ assert_eq!(service.status(prepared.run.id).unwrap().state, State::ReconciliationRequired);
+ service.shutdown().await;
+ drop(service);
+ f.unavailable.store(0, Ordering::SeqCst);
let service=Service::open(f.client.clone(),dir.path(),host.clone(),options()).unwrap();
service.recover().await.unwrap();
wait_for(async || service.status(prepared.run.id).unwrap().state==State::Finished).await;
@@ -496,3 +509,63 @@
assert!(observer.events(prepared.run.id,0).await.unwrap().events.is_empty());
service.shutdown().await; server.abort(); let _=server.await;
}
+
+#[tokio::test]
+async fn cancellation_after_reservation_before_admission_finishes_without_execution() {
+ let f = Fixture::new().await;
+ let prepared = f.managed_session().await;
+ // Force the ordering that races the controller's periodic reservation.
+ wait_for(async || f.client.get(prepared.run.id).await.unwrap().phase == Phase::SessionInFlight).await;
+ let dir = tempfile::Builder::new().permissions(std::fs::Permissions::from_mode(0o700)).tempdir().unwrap();
+ let host = adapter(None);
+ let request = submit(&prepared, dir.path());
+ let service = Service::open(f.client.clone(), dir.path(), host.clone(), options()).unwrap();
+ let cancelled = service.cancel_submission(request.clone()).await.unwrap();
+ assert_eq!(cancelled.phase, Phase::Cancelled);
+ assert_eq!(cancelled.state, State::Finished);
+ assert!(cancelled.activity_complete);
+ let run = f.client.get(prepared.run.id).await.unwrap();
+ assert_eq!(run.phase, Phase::Cancelled);
+ assert!(!run.usage.usage_unknown);
+ assert!(run.usage.coverage.as_ref().unwrap().complete());
+ assert_eq!(run.usage.model_calls, 0);
+ assert_eq!(run.usage.input_tokens, 0);
+ assert_eq!(run.usage.output_tokens, 0);
+ assert_eq!(run.usage.cost_micro_usd, 0);
+ assert!(service.events(run.id, 0).unwrap().events.is_empty());
+ service.shutdown().await;
+ drop(service);
+ let reopened = Service::open(f.client.clone(), dir.path(), host.clone(), options()).unwrap();
+ reopened.recover().await.unwrap();
+ assert_eq!(reopened.submit(request).await.unwrap(), cancelled);
+ assert_eq!(host.0.calls.load(Ordering::SeqCst), 0);
+ reopened.shutdown().await;
+}
+
+#[tokio::test]
+async fn unattempted_cancel_rejects_contradictory_execution_evidence() {
+ let f = Fixture::new().await;
+ let prepared = f.managed_session().await;
+ wait_for(async || f.client.get(prepared.run.id).await.unwrap().phase == Phase::SessionInFlight).await;
+ let run = f.client.get(prepared.run.id).await.unwrap();
+ let operation = &run.session.as_ref().unwrap().operation;
+ f.client.observe_host(run.id, &controller::Observe {
+ lease_id: prepared.binding.lease_id, host: prepared.binding.host.clone(),
+ expected_revision: run.revision, operation_id: operation.id, digest: operation.digest.clone(),
+ observation: controller::Observation::Session {
+ observation: session::Observation::ResponseStarted { id: "existing-effect".into() },
+ },
+ }).await.unwrap();
+ let dir = tempfile::Builder::new().permissions(std::fs::Permissions::from_mode(0o700)).tempdir().unwrap();
+ let host = adapter(None);
+ let service = Service::open(f.client.clone(), dir.path(), host.clone(), options()).unwrap();
+ let error = service.cancel_submission(submit(&prepared, dir.path())).await.unwrap_err();
+ assert_eq!(error, "unattempted cancellation conflicts with execution evidence");
+ let run = f.client.get(run.id).await.unwrap();
+ assert_eq!(run.phase, Phase::Uncertain);
+ assert!(run.usage.usage_unknown);
+ assert_eq!(run.session.as_ref().unwrap().responses[0].id, "existing-effect");
+ assert!(!service.status(run.id).unwrap().activity_complete);
+ assert_eq!(host.0.calls.load(Ordering::SeqCst), 0);
+ service.shutdown().await;
+}
--- a/crates/airouter-agentd/SESSION_SERVICE.md
+++ b/crates/airouter-agentd/SESSION_SERVICE.md
@@ -33,6 +33,14 @@
share the same lock. This endpoint saves a cancellation without starting a
process or reading attachment sources; repeating the original submission returns
the cancelled record. Changed bodies and foreign bindings are rejected.
+
+The controller may already have reserved a session when this cancellation
+arrives. A durable cancellation record with `attempted = false` proves that
+the session service has never polled the executor. It reports a terminal
+cancelled outcome with known zero usage under the original binding, including
+after restart or a framework outage. Conflicting response or usage evidence
+is rejected. An attempted execution still requires its original cleanup and
+activity receipts; cancellation does not turn missing evidence into success.
The spool is a private canonical directory outside the model workspace; the
service rejects a spool beneath the submitted working directory. A
|
