use super::*;
impl CanonicalOperationDriver {
pub fn project_logical_state(&self) -> LogicalStateProjection {
LogicalStateProjection {
root_kind: self.root_kind,
focus: self.focus.clone(),
syscall: self.project_syscall_state(),
scheduler: self.project_scheduler_state(),
context_vm: self.project_context_vm_state(),
}
}
pub(super) fn project_syscall_state(&self) -> SyscallState {
SyscallState {
policy_revision: self.policy.as_ref().map(LivePolicyState::revision),
live_config: self.policy.as_ref().map(|policy| policy.config().clone()),
provider_calls: self
.provider_calls
.iter()
.map(|(effect_id, call)| PendingProviderCallState {
effect_id: effect_id.clone(),
task_id: call.task_id.clone(),
exposed_tools: call.exposed_tools.iter().cloned().collect(),
})
.collect(),
consumed_call_ids: self.consumed_calls.iter().cloned().collect(),
authored_memory_writes: self
.pending_memory_writes
.iter()
.map(|(effect_id, write)| AuthoredMemoryWriteState {
effect_id: effect_id.clone(),
binding_id: write.binding_id.clone(),
name: write.name.clone(),
kind: write.kind,
size_bytes: write.size_bytes,
})
.collect(),
authored_memory_queries: self
.pending_memory_queries
.iter()
.map(|(effect_id, query)| AuthoredMemoryQueryState {
effect_id: effect_id.clone(),
binding_id: query.binding_id.clone(),
text: query.text.clone(),
requested_k: query.requested_k,
})
.collect(),
memory_write_window_ms: self
.engine
.as_ref()
.map(|engine| {
engine
.memory_write_window()
.iter()
.copied()
.map(WireU64::new)
.collect()
})
.unwrap_or_default(),
}
}
pub(super) fn project_scheduler_state(&self) -> SchedulerState {
let Some(engine) = self.engine.as_ref() else {
return SchedulerState::default();
};
let (total_tokens, subagents_spawned, rounds_completed) = engine.local_budget_usage();
let signal_state = engine.signal_checkpoint_state();
let entropy_state = engine.entropy_checkpoint_state();
let workflow = engine.workflow_checkpoint_nodes().map(|runtime_nodes| {
let workflow_id = self
.workflow_id
.as_ref()
.expect("an active canonical workflow has a logical identity");
assert_eq!(
runtime_nodes.len(),
self.workflow_nodes.len(),
"the semantic workflow and its canonical source DAG stay index-aligned"
);
WorkflowGraphState {
workflow_id: workflow_id.clone(),
nodes: runtime_nodes
.into_iter()
.zip(self.workflow_nodes.iter())
.map(|(runtime, wire)| WorkflowNodeState {
node_id: wire.node_id.clone(),
task: wire.task.clone(),
depends_on: wire.depends_on.clone(),
run_spec: wire.run_spec.clone(),
kind: workflow_kind_label(&runtime).to_string(),
status: workflow_status_label(runtime.status).to_string(),
active_agent_id: runtime.active_agent_id,
iterations_completed: runtime.iterations_completed as u32,
})
.collect(),
}
});
SchedulerState {
run_spec: engine.run_spec.as_ref().map(logical_agent_run_spec),
advertised_tool_ids: engine.advertised_tool_ids(),
turn: engine.turn,
total_tokens: WireU64::new(total_tokens),
rounds_completed,
subagents_spawned,
started_at_ms: engine.started_at_ms().map(WireU64::new),
wall_budget_ms: engine.wall_budget().map(WireU64::new),
tasks: engine
.task_table()
.all()
.iter()
.map(|tcb| TaskControlState {
task_id: TaskId::new(tcb.id.as_str())
.expect("an internal task id is always a legal branded ref"),
parent_task_id: tcb
.parent
.as_ref()
.and_then(|parent| TaskId::new(parent.as_str()).ok()),
lifecycle: tcb.state.label().to_string(),
runnable_cause: tcb.runnable_cause,
termination: match tcb.state {
TaskLifecycle::Done(reason) => Some(reason.label().to_string()),
_ => None,
},
wait_set: tcb.wait_set.as_ref().map(project_wait_set),
capability_ids: tcb.caps.iter().map(|cap| cap.to_string()).collect(),
capabilities: tcb.capabilities.clone(),
process: tcb.proc.as_ref().map(|process| ChildProcessState {
role: agent_role_label(process.role).to_string(),
isolation: agent_isolation_label(process.isolation).to_string(),
context_inheritance: context_inheritance_label(process.context_inheritance)
.to_string(),
join_result: process.result.as_ref().map(|result| {
super::super::scalar::BoundedJson::new(
serde_json::to_value(result)
.expect("a child join result is serializable"),
)
.expect("a child join result is bounded")
}),
}),
supervision: tcb.supervision.clone(),
supervision_events: tcb.supervision_events.clone(),
tokens_used: WireU64::new(tcb.budget.total_tokens),
turns_used: tcb.budget.turns,
child_budget_remaining: tcb.child_budget_remaining,
budget_grant: tcb.budget_grant.clone(),
mailbox: tcb.mailbox.clone(),
})
.collect(),
attempts: self
.attempts
.iter()
.filter_map(|(task_id, attempt_id)| {
Some(TaskAttemptState {
task_id: TaskId::new(task_id.as_str()).ok()?,
attempt_id: attempt_id.clone(),
})
})
.collect(),
workflow,
queued_signals: signal_state
.queued
.into_iter()
.map(|queued| queued_signal_state(&queued))
.collect(),
signal_dedupe_keys: signal_state
.seen_order
.into_iter()
.map(|key| key.to_string())
.collect(),
milestone: self
.loaded_contract_id
.as_ref()
.map(|contract_id| MilestoneState {
contract_id: contract_id.clone(),
phase_id: engine.current_milestone_phase_id().map(str::to_string),
complete: engine.is_milestone_complete(),
blocked_count: engine.milestone_blocked_count(),
}),
entropy: EntropyState {
window: entropy_state
.window
.into_iter()
.map(|entry| EntropyTurnState {
errored_results: entry.errored_results,
total_results: entry.total_results,
rollbacks: entry.rollbacks,
})
.collect(),
rollbacks_pending: entropy_state.rollbacks_pending,
disarmed: entropy_state.disarmed,
last_alert_turn: entropy_state.last_alert_turn,
},
channels: engine
.task_table()
.channels()
.iter()
.map(|(channel_id, channel)| LocalChannelState {
channel_id: channel_id.0.to_string(),
channel: channel.clone(),
})
.collect(),
objects: engine.task_table().objects().values().cloned().collect(),
}
}
pub(super) fn project_context_vm_state(&self) -> ContextVmState {
let Some(engine) = self.engine.as_ref() else {
return ContextVmState::default();
};
let ctx = &engine.ctx;
ContextVmState {
handles: ctx
.handles
.all()
.iter()
.map(|handle| {
let (payload_ref, digest, original_size) = match &handle.residency {
Residency::External {
payload_ref,
digest,
original_size,
} => (
Some(payload_ref.clone()),
Some(digest.clone()),
Some(WireU64::new(*original_size)),
),
Residency::PagedOut {
payload_ref,
digest,
} => (Some(payload_ref.clone()), Some(digest.clone()), None),
Residency::Resident | Residency::Collapsed => (None, None, None),
};
HandleState {
handle_id: handle.id,
kind: handle_kind_label(&handle.kind).to_string(),
residency: handle.residency.label().to_string(),
payload_ref,
digest,
original_size,
tokens: handle.tokens,
source: handle.source.as_ref().map(|source| source.to_string()),
}
})
.collect(),
next_handle_id: ctx.next_handle_id(),
pending_payload_loads: self
.pending_payload_loads
.iter()
.map(|(effect_id, load)| PendingPayloadLoadState {
effect_id: effect_id.clone(),
handle_id: load.handle_id.clone(),
digest: load.digest.clone(),
original_size: load.original_size.map(WireU64::new),
})
.collect(),
active_skills: ctx
.active_skills
.iter()
.map(|(skill, lease)| SkillLeaseState {
skill: skill.to_string(),
lease_until_turn: *lease,
})
.collect(),
knowledge: ctx
.partitions
.knowledge
.entries
.iter()
.map(|entry| KnowledgeSlotState {
key: entry.key.as_ref().map(|key| key.to_string()),
role: role_label(entry.message.role).to_string(),
body: self.project_body(&entry.message),
tokens: entry.tokens,
pinned: entry.pinned,
evict_at_boundary: entry.evict_at_boundary,
})
.collect(),
signals: ctx.partitions.signals.clone(),
messages: ctx
.partitions
.system
.messages
.iter()
.map(|message| self.project_message(MessagePartition::System, message))
.chain(
ctx.partitions
.history
.messages
.iter()
.map(|message| self.project_message(MessagePartition::History, message)),
)
.collect(),
task_state: project_task_state(&ctx.partitions.task_state),
partition_tokens: PartitionTokenState {
system: ctx.partitions.system.token_count,
knowledge: ctx.partitions.knowledge.token_count,
history: ctx.partitions.history.token_count,
},
history_len: ctx.partitions.history.messages.len() as u32,
frozen_history_len: ctx.frozen_history_len() as u32,
last_activity_ms: WireU64::new(ctx.last_activity_ms),
last_compact_ms: ctx.last_compact_ms.map(WireU64::new),
}
}
pub(super) fn project_message(
&self,
partition: MessagePartition,
message: &Message,
) -> StoredMessageState {
StoredMessageState {
partition,
role: role_label(message.role).to_string(),
body: self.project_body(message),
tool_calls: message
.tool_calls
.iter()
.map(|call| LogicalToolCall {
call_id: call.id.to_string(),
name: call.name.to_string(),
arguments: call.arguments.to_string(),
})
.collect(),
tokens: message.token_count.unwrap_or(0),
}
}
pub(super) fn project_body(&self, message: &Message) -> StoredMessageBody {
let single_tool_result_is_external =
match &message.content {
Content::Parts(parts) if parts.len() == 1 => match &parts[0] {
ContentPart::ToolResult { call_id, .. } => {
self.engine
.as_ref()
.and_then(|engine| {
engine.ctx.handles.all().iter().find(|handle| {
handle.source.as_deref() == Some(call_id.as_str())
})
})
.is_some_and(|handle| handle.residency.digest().is_some())
}
_ => false,
},
_ => false,
};
if !single_tool_result_is_external {
if let Some(results) = durable_tool_results_from_content(&message.content) {
return StoredMessageBody::Structured(StructuredMessageBody {
durable_content: None,
durable_tool_results: results,
});
}
if let Some(result) = durable_tool_result_from_content(&message.content) {
return StoredMessageBody::Structured(StructuredMessageBody {
durable_content: None,
durable_tool_results: vec![result],
});
}
}
let Some((text, tool_call_id, is_error)) = message_body_parts(message) else {
let durable_content = content_to_durable(&message.content)
.expect("every non-tool message uses the canonical durable content model");
return StoredMessageBody::Structured(StructuredMessageBody {
durable_content: Some(durable_content),
durable_tool_results: Vec::new(),
});
};
let Some(call_id) = tool_call_id.as_deref() else {
return StoredMessageBody::Inline(InlineMessageBody {
text,
tool_call_id,
is_error,
});
};
let referenced = self.engine.as_ref().and_then(|engine| {
let handle = engine
.ctx
.handles
.all()
.iter()
.find(|handle| handle.source.as_deref() == Some(call_id))?;
let digest = handle.residency.digest()?;
Some((
handle.id,
digest.to_string(),
matches!(handle.residency, Residency::PagedOut { .. }),
))
});
match referenced {
Some((handle_id, digest, is_paged_out)) => {
StoredMessageBody::Reference(ReferencedMessageBody {
handle_id,
digest,
preview: if is_paged_out {
crate::context::renderer::collapse_preview(&text, call_id)
} else {
truncate_on_char_boundary(&text, self.preview_bytes())
},
tool_call_id,
is_error,
})
}
None => StoredMessageBody::Inline(InlineMessageBody {
text,
tool_call_id,
is_error,
}),
}
}
pub(super) fn preview_bytes(&self) -> usize {
self.policy
.as_ref()
.map(|policy| policy.config().payload_policy.preview_bytes as usize)
.unwrap_or(2 * 1024)
}
pub fn restore_logical_state(
genesis_config: &ResolvedOperationConfig,
state: &LogicalKernelState,
) -> Result<Self, KernelFault> {
let mut driver = Self::new();
let live_config = state
.syscall
.live_config
.clone()
.unwrap_or_else(|| genesis_config.clone());
let mut engine = build_engine(genesis_config);
install_live_policies(&mut engine, &live_config);
engine.set_root_workflow(state.transition.root_kind == Some(RootKind::Workflow));
driver.policy = Some(LivePolicyState::restore(
state.syscall.policy_revision.unwrap_or(WireU64::ZERO),
live_config.clone(),
));
restore_scheduler(&mut engine, &live_config, &state.scheduler)?;
if let Some(preempt) =
state
.transition
.pending_effects
.iter()
.find_map(|effect| match &effect.effect {
EffectKind::PreemptTasks(preempt) => Some(preempt),
_ => None,
})
{
engine.restore_pending_preempt(
preempt
.attempts
.iter()
.map(|attempt| attempt.task_id.as_str().to_string())
.collect(),
preempt.reason.clone(),
);
}
restore_context_vm(&mut engine, &state.context_vm)?;
engine.restore_memory_write_window(
state
.syscall
.memory_write_window_ms
.iter()
.map(|at| at.get())
.collect(),
);
if let Some(milestone) = &state.scheduler.milestone {
let contract = live_config
.verification_contract(&milestone.contract_id)
.ok_or_else(|| {
KernelFault::new(
KernelFaultCode::CheckpointIncompatible,
format!(
"the checkpoint runs verification contract {:?}, which this \
operation's configuration no longer declares",
milestone.contract_id
),
)
})?;
engine.load_milestone_contract(core_milestone_contract(contract, &live_config));
if !engine
.restore_milestone_cursor(milestone.phase_id.as_deref(), milestone.blocked_count)
{
return Err(KernelFault::new(
KernelFaultCode::CheckpointIncompatible,
format!(
"the checkpoint sits on milestone phase {:?} of contract {:?}, which that \
contract does not declare",
milestone.phase_id, milestone.contract_id
),
));
}
driver.loaded_contract_id = Some(milestone.contract_id.clone());
}
driver.engine = Some(engine);
driver.root_kind = state.transition.root_kind;
driver.focus = state.transition.focus.clone();
if let Some(workflow) = &state.scheduler.workflow {
driver.workflow_id = Some(workflow.workflow_id.clone());
driver.node_ids = workflow
.nodes
.iter()
.map(|node| node.node_id.clone())
.collect();
driver.workflow_nodes = workflow
.nodes
.iter()
.map(|node| WireNode {
node_id: node.node_id.clone(),
task: node.task.clone(),
depends_on: node.depends_on.clone(),
run_spec: node.run_spec.clone(),
})
.collect();
}
driver.attempts = state
.scheduler
.attempts
.iter()
.map(|attempt| {
(
attempt.task_id.as_str().to_string(),
attempt.attempt_id.clone(),
)
})
.collect();
driver.provider_calls = state
.syscall
.provider_calls
.iter()
.map(|call| {
(
call.effect_id.clone(),
PendingProviderCall {
task_id: call.task_id.clone(),
exposed_tools: call.exposed_tools.iter().cloned().collect(),
},
)
})
.collect();
driver.consumed_calls = state.syscall.consumed_call_ids.iter().cloned().collect();
driver.pending_memory_writes = state
.syscall
.authored_memory_writes
.iter()
.map(|write| {
(
write.effect_id.clone(),
AuthoredMemoryWrite {
binding_id: write.binding_id.clone(),
name: write.name.clone(),
kind: write.kind,
size_bytes: write.size_bytes,
},
)
})
.collect();
driver.pending_memory_queries = state
.syscall
.authored_memory_queries
.iter()
.map(|query| {
(
query.effect_id.clone(),
AuthoredMemoryQuery {
binding_id: query.binding_id.clone(),
text: query.text.clone(),
requested_k: query.requested_k,
},
)
})
.collect();
driver.pending_payload_loads = state
.context_vm
.pending_payload_loads
.iter()
.map(|load| {
(
load.effect_id.clone(),
PendingPayloadLoad {
handle_id: load.handle_id.clone(),
digest: load.digest.clone(),
original_size: load.original_size.map(WireU64::get),
},
)
})
.collect();
Ok(driver)
}
}