1use super::*;
4use crate::agent_store::{RecoveryDisposition, RunUsage};
5use serde::{Deserialize, Serialize};
6use serde_json::Value;
7
8const LEGACY_VERSION: i64 = 1;
9const VERSION: i64 = 2;
10const BOUNDED_VERSION: i64 = 3;
11const OUTCOME_SCHEMA_VERSION: u32 = 1;
12const LIMIT: usize = 16 * 1024 * 1024;
13#[derive(Serialize, Deserialize)]
14#[serde(deny_unknown_fields)]
15struct SnapshotV1 {
16 run_id: String,
17 workspace: String,
18 agent_version: String,
19 binding: String,
20 next_turn: u32,
21 request: ModelRequest,
22}
23
24#[derive(Serialize, Deserialize)]
25#[serde(deny_unknown_fields)]
26struct SnapshotV2 {
27 run_id: String,
28 workspace: String,
29 agent_version: String,
30 binding: String,
31 next_turn: u32,
32 request: ModelRequest,
33 budgets: RunBudgets,
34 usage: RunUsage,
35 original_deadline: i64,
36 #[serde(default)]
37 suspended_ms: i64,
38 outcome_schema_version: u32,
39 projection: context::ProjectionMetadata,
40}
41
42#[derive(Serialize, Deserialize)]
43#[serde(deny_unknown_fields)]
44struct SnapshotV3 {
45 run_id: String,
46 workspace: String,
47 agent_version: String,
48 binding: String,
49 next_turn: u32,
50 request: ModelRequest,
51 budgets: RunBudgets,
52 usage: RunUsage,
53 original_deadline: i64,
54 #[serde(default)]
55 suspended_ms: i64,
56 outcome_schema_version: u32,
57 projection: context::ProjectionMetadata,
58 logical_indexes: Vec<u64>,
59 source_message_count: u64,
60 source_exchange_count: u64,
61}
62
63pub(super) struct CheckpointContext<'a> {
64 pub request: &'a ModelRequest,
65 pub projection: &'a context::ProjectionMetadata,
66 pub logical: Option<(&'a [u64], u64, u64)>,
67}
68
69struct RestoredSnapshot {
70 run_id: String,
71 workspace: String,
72 agent_version: String,
73 binding: String,
74 next_turn: u32,
75 request: ModelRequest,
76 v2: Option<SnapshotV2State>,
77 bounded: Option<(Vec<u64>, u64, u64)>,
78}
79
80struct SnapshotV2State {
81 budgets: RunBudgets,
82 usage: RunUsage,
83 original_deadline: i64,
84 suspended_ms: i64,
85 outcome_schema_version: u32,
86 projection: context::ProjectionMetadata,
87}
88
89fn usage_contains(current: &RunUsage, checkpoint: &RunUsage) -> bool {
90 fn contains(current: Option<u64>, checkpoint: Option<u64>) -> bool {
91 match (current, checkpoint) {
92 (_, None) => true,
93 (Some(current), Some(checkpoint)) => current >= checkpoint,
94 (None, Some(_)) => false,
95 }
96 }
97 current.model_turns >= checkpoint.model_turns
98 && contains(current.input_tokens, checkpoint.input_tokens)
99 && contains(current.output_tokens, checkpoint.output_tokens)
100 && contains(current.total_tokens, checkpoint.total_tokens)
101 && current.responses_with_usage >= checkpoint.responses_with_usage
102 && current.responses_without_usage >= checkpoint.responses_without_usage
103 && current.tool_calls >= checkpoint.tool_calls
104}
105impl AgentRuntime {
106 fn uses_unrecoverable_tool(&self, resource: &LoadedAgentResource) -> bool {
107 resource.definition.agent.tools.iter().any(|name| {
108 name.starts_with("mcp.")
109 || name == "agent.invoke"
110 || self.tool_bindings.get(name).is_some_and(|binding| {
111 tool_binding::mcp_reference(&binding.definition.tool.implementation).is_some()
112 })
113 })
114 }
115
116 pub(super) fn selected_binding_metadata(
117 &self,
118 resource: &LoadedAgentResource,
119 ) -> Result<Value> {
120 self.selected_binding_metadata_with(resource, None)
121 }
122
123 pub(super) fn selected_binding_metadata_with(
124 &self,
125 resource: &LoadedAgentResource,
126 external: Option<&ToolRegistry>,
127 ) -> Result<Value> {
128 let mut bindings = serde_json::Map::new();
129 for name in &resource.definition.agent.tools {
130 if let Some(metadata) = self
131 .tools
132 .find(name)
133 .or_else(|| external.and_then(|tools| tools.find(name)))
134 .and_then(|tool| tool.binding_metadata())
135 {
136 bindings.insert(name.clone(), metadata);
137 }
138 }
139 Ok(Value::Object(bindings))
140 }
141
142 fn binding(&self, resource: &LoadedAgentResource, schema_version: i64) -> Result<String> {
143 let alias = &resource.definition.agent.model;
144 let (provider, model) = self.models.identity(alias)?;
145 let binding = if schema_version == LEGACY_VERSION {
146 json!({
147 "runtime_schema": LEGACY_VERSION,
148 "db":self.config.db.path.canonicalize()?,
149 "candidate_k_keyword":self.config.retrieval.candidate_k_keyword.clamp(1,1000),
150 "provider":provider,"model":model,"definition":self.config.models.get(alias),
151 "tools":self.declarations(resource)?,
152 "tool_binding_contract": tool_binding::CATALOG_CONTRACT_VERSION,
153 "tool_bindings": self.selected_binding_metadata(resource)?,
154 "policy_allow":self.policy.allow,"policy_approval":self.policy.require_approval,
155 })
156 } else {
157 let implementation = self.models.implementation_identity(alias)?;
158 json!({
159 "runtime_schema": schema_version,
160 "db":self.config.db.path.canonicalize()?,
161 "candidate_k_keyword":self.config.retrieval.candidate_k_keyword.clamp(1,1000),
162 "provider":provider,"model":model,"definition":self.config.models.get(alias),
163 "provider_implementation":{"id":implementation.id(),"version":implementation.version()},
164 "tools":self.declarations(resource)?,
165 "tool_binding_contract": tool_binding::CATALOG_CONTRACT_VERSION,
166 "tool_bindings": self.selected_binding_metadata(resource)?,
167 "policy_allow":self.policy.allow,"policy_approval":self.policy.require_approval,
168 })
169 };
170 Ok(format!(
171 "{:x}",
172 Sha256::digest(serde_json::to_vec(&binding)?)
173 ))
174 }
175 pub(super) async fn checkpoint(
176 &self,
177 id: &str,
178 resource: &LoadedAgentResource,
179 checkpoint_context: CheckpointContext<'_>,
180 turn: u32,
181 baseline: Option<&AgentRun>,
182 ) -> Result<()> {
183 if self
185 .execution
186 .as_ref()
187 .is_some_and(|context| context.ancestry.len() > 1)
188 {
189 return Ok(());
190 }
191 if self.uses_unrecoverable_tool(resource) {
192 return Ok(());
193 }
194 checkpoint_context.request.validate()?;
195 let current;
196 let run = if let Some(run) = baseline {
197 run
198 } else {
199 current = self
200 .store
201 .get_run_raw(id)
202 .await?
203 .context("run disappeared")?;
204 ¤t
205 };
206 let timeout = run
207 .budgets
208 .timeout_seconds
209 .context("run timeout budget is unavailable")?;
210 let original_deadline = run
211 .created_at
212 .checked_add(
213 i64::try_from(timeout)?
214 .checked_mul(1000)
215 .context("timeout is too large")?,
216 )
217 .and_then(|value| value.checked_add(run.suspended_ms))
218 .context("timeout is too large")?;
219 let common = SnapshotV2 {
220 run_id: id.into(),
221 workspace: workspace_id(&self.root)?,
222 agent_version: resource.version.clone(),
223 binding: self.binding(
224 resource,
225 if checkpoint_context.logical.is_some() {
226 BOUNDED_VERSION
227 } else {
228 VERSION
229 },
230 )?,
231 next_turn: turn,
232 request: checkpoint_context.request.clone(),
233 budgets: run.budgets.0.clone(),
234 usage: run.usage.0.clone(),
235 original_deadline,
236 suspended_ms: run.suspended_ms,
237 outcome_schema_version: OUTCOME_SCHEMA_VERSION,
238 projection: checkpoint_context.projection.clone(),
239 };
240 let (schema_version, value) =
241 if let Some((logical_indexes, source_message_count, source_exchange_count)) =
242 checkpoint_context.logical
243 {
244 let snapshot = SnapshotV3 {
245 run_id: common.run_id,
246 workspace: common.workspace,
247 agent_version: common.agent_version,
248 binding: common.binding,
249 next_turn: common.next_turn,
250 request: common.request,
251 budgets: common.budgets,
252 usage: common.usage,
253 original_deadline: common.original_deadline,
254 suspended_ms: common.suspended_ms,
255 outcome_schema_version: common.outcome_schema_version,
256 projection: common.projection,
257 logical_indexes: logical_indexes.to_vec(),
258 source_message_count,
259 source_exchange_count,
260 };
261 (BOUNDED_VERSION, serde_json::to_value(snapshot)?)
262 } else {
263 (VERSION, serde_json::to_value(common)?)
264 };
265 ensure!(
266 serde_json::to_vec(&value)?.len() <= LIMIT,
267 "checkpoint exceeds 16 MiB"
268 );
269 self.store
270 .save_checkpoint(id, schema_version, i64::from(turn), &value)
271 .await?;
272 Ok(())
273 }
274
275 pub async fn continue_run(
277 &self,
278 id: &str,
279 resource: &LoadedAgentResource,
280 input: &str,
281 cancel: watch::Receiver<bool>,
282 ) -> Result<AgentRun> {
283 ensure!(
284 !input.trim().is_empty() && input.len() <= 8192,
285 "continuation input must contain 1–8192 bytes"
286 );
287 self.store
288 .get_run(id)
289 .await?
290 .context("run not found in workspace")?;
291 ensure!(
292 !self.uses_unrecoverable_tool(resource),
293 "MCP-backed or delegating runs cannot continue"
294 );
295 ensure!(
296 self.store.lineage(id).await?.parent_run_id.is_none(),
297 "delegated child runs cannot continue independently"
298 );
299 let files = files::acquire(&self.root, id)?;
300 let run = self.store.get_run(id).await?.context("run disappeared")?;
301 ensure!(
302 run.lifecycle == crate::agent_store::RunLifecycle::Suspended
303 && matches!(
304 run.outcome,
305 Some(crate::agent_store::RunOutcomeKind::Blocked)
306 | Some(crate::agent_store::RunOutcomeKind::NeedsUserInput)
307 ),
308 "run is not suspended for explicit continuation"
309 );
310 ensure!(
311 run.agent_name == resource.definition.agent.name
312 && run.agent_version == resource.version
313 && resource.definition.version()? == resource.version
314 && run.model == resource.definition.agent.model,
315 "agent resource or model alias changed"
316 );
317 let suspension = self
318 .store
319 .latest_suspension(id)
320 .await?
321 .context("pending suspension state is unavailable")?;
322 ensure!(
323 suspension.continuation_input.is_none()
324 && suspension.continued_at.is_none()
325 && suspension.suspension_sequence == run.last_sequence
326 && Some(suspension.outcome) == run.outcome
327 && Some(suspension.reason_code.as_str()) == run.reason_code.as_deref(),
328 "pending suspension state does not match the run"
329 );
330 let assistant: ModelMessage = serde_json::from_value(
331 suspension
332 .assistant_message
333 .context("historical suspension cannot continue")?
334 .0,
335 )?;
336 let ModelMessage::Assistant { tool_calls, .. } = &assistant else {
337 anyhow::bail!("pending suspension assistant message is invalid")
338 };
339 ensure!(
340 tool_calls.len() == 1,
341 "pending suspension control is invalid"
342 );
343 let call = &tool_calls[0];
344 let call_id = call.id.clone();
345 let expected_name = match run.outcome {
346 Some(crate::agent_store::RunOutcomeKind::Blocked) => control::BLOCKED,
347 Some(crate::agent_store::RunOutcomeKind::NeedsUserInput) => control::REQUEST_USER_INPUT,
348 _ => unreachable!(),
349 };
350 ensure!(
351 call.name == expected_name
352 && resource.definition.agent.tools.contains(&call.name)
353 && control::outcome(
354 match run.outcome {
355 Some(crate::agent_store::RunOutcomeKind::Blocked) => {
356 crate::traits::RunControlKind::Blocked
357 }
358 _ => crate::traits::RunControlKind::RequestUserInput,
359 },
360 call.arguments.clone(),
361 )
362 .is_ok(),
363 "pending suspension control is invalid"
364 );
365 let checkpoint = self
366 .store
367 .latest_checkpoint(id)
368 .await?
369 .context("run has no recovery checkpoint")?;
370 ensure!(
371 matches!(checkpoint.schema_version, VERSION | BOUNDED_VERSION),
372 "historical suspension checkpoint cannot continue"
373 );
374 let mut state = match checkpoint.schema_version {
375 VERSION => {
376 let current: SnapshotV2 = serde_json::from_value(checkpoint.state.0)?;
377 RestoredSnapshot {
378 run_id: current.run_id,
379 workspace: current.workspace,
380 agent_version: current.agent_version,
381 binding: current.binding,
382 next_turn: current.next_turn,
383 request: current.request,
384 v2: Some(SnapshotV2State {
385 budgets: current.budgets,
386 usage: current.usage,
387 original_deadline: current.original_deadline,
388 suspended_ms: current.suspended_ms,
389 outcome_schema_version: current.outcome_schema_version,
390 projection: current.projection,
391 }),
392 bounded: None,
393 }
394 }
395 BOUNDED_VERSION => {
396 let current: SnapshotV3 = serde_json::from_value(checkpoint.state.0)?;
397 RestoredSnapshot {
398 run_id: current.run_id,
399 workspace: current.workspace,
400 agent_version: current.agent_version,
401 binding: current.binding,
402 next_turn: current.next_turn,
403 request: current.request,
404 v2: Some(SnapshotV2State {
405 budgets: current.budgets,
406 usage: current.usage,
407 original_deadline: current.original_deadline,
408 suspended_ms: current.suspended_ms,
409 outcome_schema_version: current.outcome_schema_version,
410 projection: current.projection,
411 }),
412 bounded: Some((
413 current.logical_indexes,
414 current.source_message_count,
415 current.source_exchange_count,
416 )),
417 }
418 }
419 _ => unreachable!(),
420 };
421 ensure!(
422 state.run_id == id
423 && state.workspace == run.workspace_id
424 && state.agent_version == run.agent_version
425 && state.binding == self.binding(resource, checkpoint.schema_version)?,
426 "checkpoint binding changed or is invalid"
427 );
428 let snapshot = state.v2.as_ref().context("checkpoint state is invalid")?;
429 let timeout = run
430 .budgets
431 .timeout_seconds
432 .context("run timeout budget is unavailable")?;
433 let deadline = run
434 .created_at
435 .checked_add(
436 i64::try_from(timeout)?
437 .checked_mul(1000)
438 .context("timeout is too large")?,
439 )
440 .and_then(|value| value.checked_add(run.suspended_ms))
441 .context("timeout is too large")?;
442 ensure!(
443 snapshot.budgets == run.budgets.0
444 && snapshot.suspended_ms == run.suspended_ms
445 && snapshot.original_deadline == deadline
446 && snapshot.outcome_schema_version == OUTCOME_SCHEMA_VERSION
447 && usage_contains(&run.usage, &snapshot.usage)
448 && serde_json::to_value(&state.request.tools)?
449 == serde_json::to_value(self.declarations(resource)?)?,
450 "checkpoint run state changed or is invalid"
451 );
452 state.request.validate()?;
453 let initial = vec![
454 ModelMessage::System {
455 content: resource.definition.prompt.system.clone(),
456 },
457 ModelMessage::User {
458 content: run.input.clone(),
459 },
460 ];
461 ensure!(
462 if checkpoint.schema_version == BOUNDED_VERSION {
463 state.request.messages.len() >= 3
464 && serde_json::to_value([
465 &state.request.messages[0],
466 &state.request.messages[2],
467 ])? == serde_json::to_value(initial)?
468 } else {
469 state.request.messages.len() >= 2
470 && serde_json::to_value(&state.request.messages[..2])?
471 == serde_json::to_value(initial)?
472 },
473 "checkpoint input changed"
474 );
475 let invocations = self.store.tool_invocations(id).await?;
476 let mut restored_results = std::collections::HashMap::new();
477 let mut restored_calls = std::collections::HashMap::new();
478 for message in &state.request.messages {
479 match message {
480 ModelMessage::Tool { call_id, content } => {
481 restored_results.insert(call_id.as_str(), content.as_str());
482 }
483 ModelMessage::Assistant { tool_calls, .. } => {
484 for call in tool_calls {
485 restored_calls.insert(call.id.as_str(), call);
486 }
487 }
488 _ => {}
489 }
490 }
491 for invocation in &invocations {
492 let recoverable = invocation.outcome_class.as_deref() == Some("recoverable_error")
493 && invocation.status == "failed"
494 && invocation.result_content.is_some();
495 ensure!(
496 (invocation.status == "completed" || recoverable)
497 && invocation.requested_sequence < checkpoint.sequence,
498 "run requires reconciliation before continuation"
499 );
500 let call = restored_calls
501 .get(invocation.call_id.as_str())
502 .context("tool call missing from checkpoint; manual reconciliation required")?;
503 ensure!(
504 call.name == invocation.tool_name && call.arguments == invocation.arguments.0,
505 "tool history does not match checkpoint; manual reconciliation required"
506 );
507 let serialized = if recoverable {
508 invocation
509 .result_content
510 .as_deref()
511 .context("recoverable tool result content missing")?
512 .to_owned()
513 } else {
514 serde_json::to_string(
515 &invocation.result.as_ref().context("tool result missing")?.0,
516 )?
517 };
518 ensure!(
519 restored_results.get(invocation.call_id.as_str()) == Some(&serialized.as_str()),
520 "checkpoint tool result changed"
521 );
522 }
523 let execution_stopped_at = run.completed_at.context("suspension timestamp missing")?;
524 ensure!(
525 execution_stopped_at
526 .checked_sub(run.created_at)
527 .and_then(|value| value.checked_sub(run.suspended_ms))
528 .is_some_and(|elapsed| elapsed >= 0
529 && elapsed
530 < i64::try_from(timeout)
531 .unwrap_or(i64::MAX)
532 .saturating_mul(1000)),
533 "duration budget exhausted"
534 );
535 ensure!(
536 run.budgets.max_total_tokens.is_none()
537 || (run.usage.responses_without_usage == 0
538 && run.usage.total_tokens.is_some_and(|total| {
539 run.budgets
540 .max_total_tokens
541 .is_some_and(|limit| total < limit)
542 })),
543 "token budget exhausted or unavailable"
544 );
545 ensure!(
546 run.budgets
547 .max_tool_calls
548 .is_none_or(|limit| run.usage.tool_calls < limit),
549 "tool call budget exhausted"
550 );
551 let outcome =
552 serde_json::to_string(&run.outcome.context("suspended outcome is unavailable")?)?;
553 let reason = serde_json::to_string(
554 run.reason_code
555 .as_deref()
556 .context("suspended reason is unavailable")?,
557 )?;
558 let prompt = serde_json::to_string(&suspension.prompt)?;
559 let supplied = serde_json::to_string(input)?;
560 let content = format!(
561 "{{\"type\":\"run_continuation\",\"outcome\":{outcome},\"reason_code\":{reason},\"prompt\":{prompt},\"input\":{supplied}}}"
562 );
563 state.request.messages.push(assistant);
564 state
565 .request
566 .messages
567 .push(ModelMessage::Tool { call_id, content });
568 state.request.validate()?;
569 let attempts = u32::try_from(run.usage.model_turns)?;
570 ensure!(
571 attempts < resource.definition.agent.execution.max_turns,
572 "model turn budget exhausted"
573 );
574 state.next_turn = attempts;
575 let (mut logical_indexes, source_message_count, source_exchange_count) =
576 if let Some((mut indexes, count, exchanges)) = state.bounded.take() {
577 indexes.push(count);
578 indexes.push(count.checked_add(1).context("message count overflow")?);
579 (Some(indexes), Some(count + 2), Some(exchanges + 1))
580 } else {
581 (None, None, None)
582 };
583 let projection = if checkpoint.schema_version == VERSION {
584 self.context_builder.describe(&state.request)
585 } else {
586 let max_bytes = resource
587 .definition
588 .agent
589 .execution
590 .max_context_bytes
591 .context("bounded context budget is unavailable")?;
592 let mut source = state.request.clone();
593 ensure!(
594 source
595 .messages
596 .get(1)
597 .is_some_and(|message| matches!(message, ModelMessage::System { .. })),
598 "bounded checkpoint capsule is unavailable"
599 );
600 source.messages.remove(1);
601 let working = self.store.working_state(id).await?;
602 ensure!(
603 working.status == WorkingStateStatus::Current,
604 "working state is not current"
605 );
606 let projected = context::bounded_project(
607 &source,
608 max_bytes,
609 &working,
610 logical_indexes
611 .as_deref()
612 .context("logical indexes missing")?,
613 source_message_count.context("source count missing")?,
614 source_exchange_count.context("exchange count missing")?,
615 )?;
616 state.request = projected.request;
617 logical_indexes = projected.retained_indexes;
618 projected.metadata
619 };
620 let checkpoint_state = if checkpoint.schema_version == VERSION {
621 serde_json::to_value(SnapshotV2 {
622 run_id: state.run_id,
623 workspace: state.workspace,
624 agent_version: state.agent_version,
625 binding: state.binding,
626 next_turn: state.next_turn,
627 request: state.request.clone(),
628 budgets: run.budgets.0.clone(),
629 usage: run.usage.0.clone(),
630 original_deadline: snapshot.original_deadline,
631 suspended_ms: run.suspended_ms,
632 outcome_schema_version: OUTCOME_SCHEMA_VERSION,
633 projection,
634 })?
635 } else {
636 serde_json::to_value(SnapshotV3 {
637 run_id: state.run_id,
638 workspace: state.workspace,
639 agent_version: state.agent_version,
640 binding: state.binding,
641 next_turn: state.next_turn,
642 request: state.request.clone(),
643 budgets: run.budgets.0.clone(),
644 usage: run.usage.0.clone(),
645 original_deadline: snapshot.original_deadline,
646 suspended_ms: run.suspended_ms,
647 outcome_schema_version: OUTCOME_SCHEMA_VERSION,
648 projection,
649 logical_indexes: logical_indexes.clone().context("logical indexes missing")?,
650 source_message_count: source_message_count.context("source count missing")?,
651 source_exchange_count: source_exchange_count.context("exchange count missing")?,
652 })?
653 };
654 let continued = self
655 .store
656 .continue_suspended_run(
657 id,
658 run.last_sequence,
659 suspension.suspension_sequence,
660 input,
661 checkpoint.schema_version,
662 i64::from(attempts),
663 &checkpoint_state,
664 )
665 .await?;
666 self.drive(
667 &continued,
668 resource,
669 Some(RestoredRequest {
670 request: state.request,
671 logical_indexes,
672 source_message_count,
673 source_exchange_count,
674 }),
675 attempts,
676 cancel,
677 &files,
678 )
679 .await
680 }
681
682 pub async fn resume(
685 &self,
686 id: &str,
687 resource: &LoadedAgentResource,
688 cancel: watch::Receiver<bool>,
689 ) -> Result<AgentRun> {
690 self.store
692 .get_run(id)
693 .await?
694 .context("run not found in workspace")?;
695 ensure!(
696 !self.uses_unrecoverable_tool(resource),
697 "MCP-backed or delegating runs cannot resume; tree/session recovery is unsupported"
698 );
699 ensure!(
700 self.store.lineage(id).await?.parent_run_id.is_none(),
701 "delegated child runs cannot resume independently"
702 );
703 let files = files::acquire(&self.root, id)?;
704 let run = self.store.get_run(id).await?.context("run disappeared")?;
705 match self.store.recovery_disposition(id).await? {
706 RecoveryDisposition::ResumeEligible => {}
707 RecoveryDisposition::ReconciliationRequired => {
708 anyhow::bail!("run requires reconciliation before resume")
709 }
710 RecoveryDisposition::RestartRequired => anyhow::bail!(
711 "run requires restart because its checkpoint, deadline, delegation, or budget is unavailable"
712 ),
713 RecoveryDisposition::Complete => anyhow::bail!("completed run cannot resume"),
714 RecoveryDisposition::Suspended => {
715 anyhow::bail!("suspended run requires an explicit continuation")
716 }
717 }
718 ensure!(
719 run.agent_name == resource.definition.agent.name
720 && run.agent_version == resource.version
721 && resource.definition.version()? == resource.version
722 && run.model == resource.definition.agent.model,
723 "agent resource or model alias changed"
724 );
725 self.remaining_time(&run, resource)?;
726 let checkpoint = self
727 .store
728 .latest_checkpoint(id)
729 .await?
730 .context("run has no recovery checkpoint")?;
731 ensure!(
732 matches!(
733 checkpoint.schema_version,
734 LEGACY_VERSION | VERSION | BOUNDED_VERSION
735 ),
736 "unsupported checkpoint version"
737 );
738 ensure!(
739 serde_json::to_vec(&checkpoint.state.0)?.len() <= LIMIT,
740 "checkpoint exceeds 16 MiB"
741 );
742 let state = match checkpoint.schema_version {
743 LEGACY_VERSION => {
744 let legacy: SnapshotV1 = serde_json::from_value(checkpoint.state.0)?;
745 RestoredSnapshot {
746 run_id: legacy.run_id,
747 workspace: legacy.workspace,
748 agent_version: legacy.agent_version,
749 binding: legacy.binding,
750 next_turn: legacy.next_turn,
751 request: legacy.request,
752 v2: None,
753 bounded: None,
754 }
755 }
756 VERSION => {
757 let current: SnapshotV2 = serde_json::from_value(checkpoint.state.0)?;
758 RestoredSnapshot {
759 run_id: current.run_id,
760 workspace: current.workspace,
761 agent_version: current.agent_version,
762 binding: current.binding,
763 next_turn: current.next_turn,
764 request: current.request,
765 v2: Some(SnapshotV2State {
766 budgets: current.budgets,
767 usage: current.usage,
768 original_deadline: current.original_deadline,
769 suspended_ms: current.suspended_ms,
770 outcome_schema_version: current.outcome_schema_version,
771 projection: current.projection,
772 }),
773 bounded: None,
774 }
775 }
776 BOUNDED_VERSION => {
777 let current: SnapshotV3 = serde_json::from_value(checkpoint.state.0)?;
778 RestoredSnapshot {
779 run_id: current.run_id,
780 workspace: current.workspace,
781 agent_version: current.agent_version,
782 binding: current.binding,
783 next_turn: current.next_turn,
784 request: current.request,
785 bounded: Some((
786 current.logical_indexes,
787 current.source_message_count,
788 current.source_exchange_count,
789 )),
790 v2: Some(SnapshotV2State {
791 budgets: current.budgets,
792 usage: current.usage,
793 original_deadline: current.original_deadline,
794 suspended_ms: current.suspended_ms,
795 outcome_schema_version: current.outcome_schema_version,
796 projection: current.projection,
797 }),
798 }
799 }
800 _ => unreachable!(),
801 };
802 ensure!(
803 state.run_id == id
804 && state.workspace == run.workspace_id
805 && state.agent_version == run.agent_version
806 && state.binding == self.binding(resource, checkpoint.schema_version)?
807 && i64::from(state.next_turn) == checkpoint.turn,
808 "checkpoint binding changed or is invalid"
809 );
810 if let Some(v2) = &state.v2 {
811 let timeout = run
812 .budgets
813 .timeout_seconds
814 .context("run timeout budget is unavailable")?;
815 let deadline = run
816 .created_at
817 .checked_add(
818 i64::try_from(timeout)?
819 .checked_mul(1000)
820 .context("timeout is too large")?,
821 )
822 .and_then(|value| value.checked_add(run.suspended_ms))
823 .context("timeout is too large")?;
824 let common_valid = v2.outcome_schema_version == OUTCOME_SCHEMA_VERSION
825 && v2.budgets == run.budgets.0
826 && v2.original_deadline == deadline
827 && v2.suspended_ms == run.suspended_ms
828 && usage_contains(&run.usage, &v2.usage);
829 let projection_valid = if checkpoint.schema_version == BOUNDED_VERSION {
830 state
831 .bounded
832 .as_ref()
833 .is_some_and(|(indexes, count, exchanges)| {
834 v2.projection.builder == "bounded_v1"
835 && v2.projection.strategy == "bounded_suffix_v1"
836 && indexes.len() + 1 == state.request.messages.len()
837 && *count >= indexes.len() as u64
838 && *exchanges >= v2.projection.included_exchange_count.unwrap_or(0)
839 })
840 } else {
841 v2.projection.builder == "compatibility_v1"
842 && v2.projection.strategy == "complete_transcript"
843 && v2.projection.message_count == Some(state.request.messages.len())
844 && v2.projection.tool_count == Some(state.request.tools.len())
845 };
846 ensure!(
847 common_valid && projection_valid,
848 "checkpoint run state changed or is invalid"
849 );
850 }
851 state.request.validate()?;
852 ensure!(
853 serde_json::to_value(&state.request.tools)?
854 == serde_json::to_value(self.declarations(resource)?)?,
855 "checkpoint tool definitions changed"
856 );
857 let initial = vec![
858 ModelMessage::System {
859 content: resource.definition.prompt.system.clone(),
860 },
861 ModelMessage::User {
862 content: run.input.clone(),
863 },
864 ];
865 ensure!(
866 if checkpoint.schema_version == BOUNDED_VERSION {
867 state.request.messages.len() >= 3
868 && serde_json::to_value([
869 &state.request.messages[0],
870 &state.request.messages[2],
871 ])? == serde_json::to_value(initial)?
872 } else {
873 state.request.messages.len() >= 2
874 && serde_json::to_value(&state.request.messages[..2])?
875 == serde_json::to_value(initial)?
876 },
877 "checkpoint input changed"
878 );
879 let invocations = self.store.tool_invocations(id).await?;
880 let omitted_exchanges = state
881 .v2
882 .as_ref()
883 .and_then(|snapshot| snapshot.projection.omitted_exchange_count)
884 .unwrap_or(0);
885 let mut restored_results = std::collections::HashMap::new();
886 let mut restored_calls = std::collections::HashMap::new();
887 for message in &state.request.messages {
888 match message {
889 ModelMessage::Tool { call_id, content } => {
890 restored_results.insert(call_id.as_str(), content.as_str());
891 }
892 ModelMessage::Assistant { tool_calls, .. } => {
893 for call in tool_calls {
894 restored_calls.insert(call.id.as_str(), call);
895 }
896 }
897 _ => {}
898 }
899 }
900 for (call_id, call) in &restored_calls {
901 if control::is_reserved(&call.name) {
902 restored_results.remove(call_id);
903 }
904 }
905 let captured_invocations = invocations
906 .iter()
907 .filter(|invocation| !invocation.tool_name.starts_with("runtime."))
908 .count();
909 ensure!(
910 if checkpoint.schema_version == BOUNDED_VERSION {
911 captured_invocations >= restored_results.len()
912 } else {
913 captured_invocations == restored_results.len()
914 },
915 "tool history is not captured by checkpoint; manual reconciliation required"
916 );
917 for invocation in &invocations {
918 if invocation.tool_name.starts_with("runtime.") {
919 continue;
920 }
921 let recoverable = invocation.outcome_class.as_deref() == Some("recoverable_error")
922 && invocation.status == "failed"
923 && invocation.result.is_some()
924 && invocation.result_content.is_some()
925 && invocation.error_code.is_some()
926 && invocation.repetition_key.is_some()
927 && invocation.consecutive_count.is_some_and(|count| count > 0);
928 ensure!(
929 (invocation.status == "completed" || recoverable)
930 && invocation.requested_sequence < checkpoint.sequence,
931 "tool history is unsafe for recovery; manual reconciliation required"
932 );
933 if checkpoint.schema_version == BOUNDED_VERSION
934 && !restored_results.contains_key(invocation.call_id.as_str())
935 {
936 ensure!(
937 omitted_exchanges > 0,
938 "tool history omission is not represented by the projection trace"
939 );
940 continue;
941 }
942 let call = restored_calls
943 .get(invocation.call_id.as_str())
944 .context("tool call missing from checkpoint; manual reconciliation required")?;
945 ensure!(
946 call.name == invocation.tool_name && call.arguments == invocation.arguments.0,
947 "tool history does not match checkpoint; manual reconciliation required"
948 );
949 let result = invocation.result.as_ref().context("tool result missing")?;
950 let serialized = if recoverable {
951 invocation
952 .result_content
953 .clone()
954 .context("recoverable tool result content missing")?
955 } else {
956 serde_json::to_string(&result.0)?
957 };
958 ensure!(
959 restored_results.get(invocation.call_id.as_str()) == Some(&serialized.as_str()),
960 "checkpoint tool result changed"
961 );
962 }
963 let mut cursor = 0;
966 let mut attempts = 0u32;
967 loop {
968 let events = self.store.events(id, cursor, 1000).await?;
969 if events.is_empty() {
970 break;
971 }
972 for event in events {
973 cursor = event.sequence;
974 if event.event_type == "model.requested" {
975 attempts = attempts.checked_add(1).context("too many model attempts")?;
976 }
977 if event.sequence > checkpoint.sequence {
978 ensure!(matches!(event.event_type.as_str(),"model.requested"|"model.responded"|"model.failed"|"model.retrying"|"run.failed"|"run.cancelled"|"run.resumed"|"run.continued"|"context.resolved"|"context.projected"),
979 "run has activity beyond its safe checkpoint; manual reconciliation required");
980 }
981 }
982 }
983 ensure!(
984 attempts >= state.next_turn && attempts < resource.definition.agent.execution.max_turns,
985 "model turn budget exhausted or invalid"
986 );
987 let resumed = self.store.reopen_run(id, run.last_sequence).await?;
988 self.drive(
989 &resumed,
990 resource,
991 Some(RestoredRequest {
992 request: state.request,
993 logical_indexes: state.bounded.as_ref().map(|value| value.0.clone()),
994 source_message_count: state.bounded.as_ref().map(|value| value.1),
995 source_exchange_count: state.bounded.as_ref().map(|value| value.2),
996 }),
997 attempts,
998 cancel,
999 &files,
1000 )
1001 .await
1002 }
1003}
1004
1005#[cfg(test)]
1006mod tests {
1007 use super::*;
1008
1009 #[test]
1010 fn legacy_v1_snapshot_remains_decodable() {
1011 let value = json!({
1012 "run_id":"run",
1013 "workspace":"workspace",
1014 "agent_version":"agent-v1",
1015 "binding":"binding",
1016 "next_turn":1,
1017 "request":{
1018 "messages":[{"role":"user","content":"objective"}],
1019 "tools":[],
1020 "output_schema":null,
1021 "max_output_tokens":null
1022 }
1023 });
1024 let snapshot: SnapshotV1 = serde_json::from_value(value).unwrap();
1025 assert_eq!(snapshot.run_id, "run");
1026 assert_eq!(snapshot.next_turn, 1);
1027 snapshot.request.validate().unwrap();
1028 }
1029}