agent-runtime-execution 0.1.0

//! Execution lifecycle coordinator between Outer Shell and Runtime Kernel.

use std::{
    collections::BTreeMap,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};

use async_trait::async_trait;
use runtime_api_contract::{
    CreateExecutionRequest, CreateExecutionResponse, EventPage, EventPayload, ExecutionFailure,
    ExecutionOptions, ExecutionOutcome, ExecutionState, ExecutionView,
    ModelGenerationOptions as ApiModelGenerationOptions, ModelToolChoice as ApiModelToolChoice,
    RuntimeInput, SubmitInputRequest,
};
use runtime_kernel_api::{
    AgentDefinitionResolver, ContextLimits, KernelEvent, KernelEventSink, KernelFailure,
    KernelLimits, KernelSpec, RuntimeKernel,
};
use runtime_link_api::{
    ExecutionJournal, JournalDelegation, JournalError, JournalReservation, LinkError,
    NewJournalExecution, RuntimeLink,
};
use runtime_ports::{
    CommitDisposition, ExecutionDelegationLease, ExecutionSessionFactory, InteractionRequest,
    InteractionResponse, InteractionSession, ModelGenerationOptions, ModelToolChoice,
    OperationContext, PortFailure, PortFailureKind, ResolvedExecutionContext,
    RuntimeInstanceResolver, SessionScope, SubagentOutcome, SubagentRequest, SubagentSession,
    TraceSession,
};
use runtime_types::{ExecutionId, OperationId, RequestAuthority};
use sha2::{Digest, Sha256};
use tokio::sync::{Mutex, OwnedSemaphorePermit, Semaphore};
use tokio_util::{sync::CancellationToken, task::TaskTracker};

#[derive(Clone)]
pub struct ExecutionCoordinator {
    journal: Arc<dyn ExecutionJournal>,
    instance_resolver: Arc<dyn RuntimeInstanceResolver>,
    session_factory: Arc<dyn ExecutionSessionFactory>,
    definition_resolver: Arc<dyn AgentDefinitionResolver>,
    kernel: Arc<dyn RuntimeKernel>,
    context_limits: ContextLimits,
    capacity: Arc<Semaphore>,
    cancellations: Arc<Mutex<BTreeMap<ExecutionId, CancellationToken>>>,
    authorities: Arc<Mutex<BTreeMap<ExecutionId, RequestAuthority>>>,
    accepting: Arc<AtomicBool>,
    tasks: TaskTracker,
    recovery_tasks: TaskTracker,
    recovery_stop: CancellationToken,
    recovery_started: Arc<AtomicBool>,
    worker_id: Arc<str>,
}

impl ExecutionCoordinator {
    pub fn new(
        journal: Arc<dyn ExecutionJournal>,
        instance_resolver: Arc<dyn RuntimeInstanceResolver>,
        session_factory: Arc<dyn ExecutionSessionFactory>,
        definition_resolver: Arc<dyn AgentDefinitionResolver>,
        kernel: Arc<dyn RuntimeKernel>,
        max_concurrent_executions: usize,
    ) -> Result<Self, LinkError> {
        Self::new_with_context_limits(
            journal,
            instance_resolver,
            session_factory,
            definition_resolver,
            kernel,
            max_concurrent_executions,
            ContextLimits::default(),
        )
    }

    #[allow(clippy::too_many_arguments)]
    pub fn new_with_context_limits(
        journal: Arc<dyn ExecutionJournal>,
        instance_resolver: Arc<dyn RuntimeInstanceResolver>,
        session_factory: Arc<dyn ExecutionSessionFactory>,
        definition_resolver: Arc<dyn AgentDefinitionResolver>,
        kernel: Arc<dyn RuntimeKernel>,
        max_concurrent_executions: usize,
        context_limits: ContextLimits,
    ) -> Result<Self, LinkError> {
        if max_concurrent_executions == 0 {
            return Err(LinkError::Invalid(
                "max_concurrent_executions must be positive".into(),
            ));
        }
        Ok(Self {
            journal,
            instance_resolver,
            session_factory,
            definition_resolver,
            kernel,
            context_limits,
            capacity: Arc::new(Semaphore::new(max_concurrent_executions)),
            cancellations: Arc::new(Mutex::new(BTreeMap::new())),
            authorities: Arc::new(Mutex::new(BTreeMap::new())),
            accepting: Arc::new(AtomicBool::new(true)),
            tasks: TaskTracker::new(),
            recovery_tasks: TaskTracker::new(),
            recovery_stop: CancellationToken::new(),
            recovery_started: Arc::new(AtomicBool::new(false)),
            worker_id: format!("runtime-worker-{}", ExecutionId::random().as_str()).into(),
        })
    }

    /// Claim and recover work left by a crashed host. Queued executions with a
    /// durable delegation are resumed; executions interrupted after starting
    /// are safely finalized without replaying unknown side effects.
    pub async fn recover(&self, limit: usize) -> Result<usize, LinkError> {
        let now = chrono::Utc::now().timestamp_millis();
        let records = self
            .journal
            .recoverable(now, limit)
            .await
            .map_err(map_journal)?;
        let mut scheduled = 0;
        for record in records {
            let Ok(permit) = self.capacity.clone().try_acquire_owned() else {
                break;
            };
            scheduled += 1;
            if record.view.state == ExecutionState::Queued && record.delegation.is_some() {
                self.tasks
                    .spawn(self.clone().run(record.view.id.clone(), permit));
            } else {
                self.tasks.spawn(
                    self.clone()
                        .recover_interrupted(record.view.id.clone(), permit),
                );
            }
        }
        Ok(scheduled)
    }

    /// Keep scanning after startup so a row whose prior worker claim had not
    /// yet expired at boot is recovered once the short lease elapses.
    pub fn start_recovery_loop(
        &self,
        interval: Duration,
        batch_size: usize,
    ) -> Result<(), LinkError> {
        if interval.is_zero() || batch_size == 0 || batch_size > 10_000 {
            return Err(LinkError::Invalid(
                "recovery interval and batch size must be bounded and positive".into(),
            ));
        }
        if self
            .recovery_started
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
            .is_err()
        {
            return Ok(());
        }
        let coordinator = self.clone();
        let stop = self.recovery_stop.clone();
        self.recovery_tasks.spawn(async move {
            let mut ticker = tokio::time::interval(interval);
            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            // Wiring performs one synchronous startup scan before admission;
            // the loop begins with the first real interval.
            ticker.tick().await;
            loop {
                tokio::select! {
                    _ = stop.cancelled() => return,
                    _ = ticker.tick() => {
                        if !coordinator.accepting.load(Ordering::SeqCst) {
                            return;
                        }
                        if let Err(error) = coordinator.recover(batch_size).await {
                            tracing::warn!(%error, "periodic execution recovery scan failed");
                        }
                    }
                }
            }
        });
        Ok(())
    }

    /// Stop admission and close the task scope so shutdown can wait for every
    /// execution spawned by this coordinator.
    pub fn begin_shutdown(&self) {
        self.accepting.store(false, Ordering::SeqCst);
        self.recovery_stop.cancel();
        self.recovery_tasks.close();
        self.tasks.close();
    }

    /// Drain active executions within `grace`. On expiry, propagate
    /// cancellation to every registered execution and wait for cleanup.
    pub async fn shutdown(&self, grace: Duration) -> bool {
        self.begin_shutdown();
        if tokio::time::timeout(grace, async {
            tokio::join!(self.recovery_tasks.wait(), self.tasks.wait());
        })
        .await
        .is_ok()
        {
            return true;
        }
        for cancellation in self.cancellations.lock().await.values() {
            cancellation.cancel();
        }
        tokio::time::timeout(Duration::from_secs(5), async {
            tokio::join!(self.recovery_tasks.wait(), self.tasks.wait());
        })
        .await
        .is_ok()
    }

    pub fn active_executions(&self) -> usize {
        self.tasks.len()
    }

    async fn run(self, execution_id: ExecutionId, permit: OwnedSemaphorePermit) {
        let _permit = permit;
        let now = chrono::Utc::now().timestamp_millis();
        let expires = now.saturating_add(30_000);
        if self
            .journal
            .claim(&execution_id, &self.worker_id, now, expires)
            .await
            .is_err()
        {
            return;
        }
        if let Err(error) = self.run_with_claim_heartbeat(&execution_id).await {
            tracing::error!(execution_id=%execution_id, code=%error.code, message=%error.message, "execution failed");
            if error.code == "WORKER_CLAIM_LOST" {
                // This worker no longer owns terminal decisions. Its execution
                // cancellation is local; the next claimant resolves the
                // durable state after the short lease expires.
                self.cancellations.lock().await.remove(&execution_id);
                self.authorities.lock().await.remove(&execution_id);
                return;
            }
            if error.commit == CommitDisposition::Unknown {
                let payload = EventPayload::Warning {
                    code: "COMMIT_DISPOSITION_UNKNOWN".into(),
                    message: format!(
                        "{}: {}; recovery must resolve the committed operation before terminal state",
                        error.code, error.message
                    ),
                };
                let state = self
                    .journal
                    .get(&execution_id)
                    .await
                    .ok()
                    .map(|record| record.view.state);
                let preserved = match state {
                    Some(ExecutionState::Finalizing) => self
                        .journal
                        .append_event(&execution_id, payload)
                        .await
                        .is_ok(),
                    Some(ExecutionState::Running) => self
                        .journal
                        .transition_with_event(
                            &execution_id,
                            &[ExecutionState::Running],
                            ExecutionState::Finalizing,
                            None,
                            None,
                            payload,
                        )
                        .await
                        .is_ok(),
                    _ => false,
                };
                if preserved {
                    self.cancellations.lock().await.remove(&execution_id);
                    self.authorities.lock().await.remove(&execution_id);
                    return;
                }
            }
            if error.code == "CANCELED"
                && self
                    .journal
                    .get(&execution_id)
                    .await
                    .is_ok_and(|record| record.view.state == ExecutionState::Finalizing)
            {
                self.cancellations.lock().await.remove(&execution_id);
                self.authorities.lock().await.remove(&execution_id);
                return;
            }
            let _ = self
                .finalize_failure(
                    &execution_id,
                    ExecutionFailure {
                        code: error.code,
                        message: error.message,
                        retryable: error.retryable,
                    },
                )
                .await;
        }
        self.cleanup_delegation(&execution_id).await;
        self.cancellations.lock().await.remove(&execution_id);
        self.authorities.lock().await.remove(&execution_id);
    }

    async fn run_with_claim_heartbeat(
        &self,
        execution_id: &ExecutionId,
    ) -> Result<(), KernelFailure> {
        let mut execution = Box::pin(self.run_inner(execution_id));
        let mut heartbeat = tokio::time::interval(Duration::from_secs(10));
        // `interval` ticks immediately once; the initial claim was committed
        // by `run`, so only subsequent ticks renew it.
        heartbeat.tick().await;
        loop {
            tokio::select! {
                result = &mut execution => return result,
                _ = heartbeat.tick() => {
                    let now = chrono::Utc::now().timestamp_millis();
                    if self
                        .journal
                        .claim(
                            execution_id,
                            &self.worker_id,
                            now,
                            now.saturating_add(30_000),
                        )
                        .await
                        .is_err()
                    {
                        if let Some(cancellation) = self.cancellations.lock().await.get(execution_id) {
                            cancellation.cancel();
                        }
                        // Give cancellation-aware ports a bounded opportunity
                        // to stop commands/streams before this worker yields
                        // ownership. Durable cleanup remains the next
                        // claimant's responsibility.
                        let _ = tokio::time::timeout(Duration::from_secs(5), &mut execution).await;
                        return Err(KernelFailure {
                            code: "WORKER_CLAIM_LOST".into(),
                            message: "worker could not renew its durable execution claim".into(),
                            retryable: true,
                            commit: CommitDisposition::Unknown,
                        });
                    }
                }
            }
        }
    }

    async fn recover_interrupted(self, execution_id: ExecutionId, permit: OwnedSemaphorePermit) {
        let _permit = permit;
        let now = chrono::Utc::now().timestamp_millis();
        if self
            .journal
            .claim(
                &execution_id,
                &self.worker_id,
                now,
                now.saturating_add(60_000),
            )
            .await
            .is_err()
        {
            return;
        }
        let Ok(mut record) = self.journal.get(&execution_id).await else {
            return;
        };
        if record.view.state.is_terminal() {
            self.cleanup_delegation(&execution_id).await;
            return;
        }
        if matches!(
            record.view.state,
            ExecutionState::Running | ExecutionState::WaitingForInput
        ) {
            let Ok(mutation) = self
                .journal
                .transition_with_event(
                    &execution_id,
                    &[ExecutionState::Running, ExecutionState::WaitingForInput],
                    ExecutionState::Finalizing,
                    None,
                    None,
                    EventPayload::Warning {
                        code: "EXECUTION_INTERRUPTED".into(),
                        message: "the prior worker stopped; recovery will not replay uncheckpointed side effects".into(),
                    },
                )
                .await
            else {
                return;
            };
            record = mutation.execution;
        }
        if record.view.state == ExecutionState::Queued {
            let _ = self
                .journal
                .transition_with_event(
                    &execution_id,
                    &[ExecutionState::Queued],
                    ExecutionState::Failed,
                    None,
                    Some(ExecutionFailure {
                        code: "DELEGATION_NOT_DURABLE".into(),
                        message: "queued execution lost caller authority before delegation attach"
                            .into(),
                        retryable: true,
                    }),
                    EventPayload::ExecutionFailed {
                        code: "DELEGATION_NOT_DURABLE".into(),
                        message: "queued execution cannot be recovered without a delegation".into(),
                    },
                )
                .await;
            return;
        }
        if let Some(stored) = record
            .delegation
            .clone()
            .filter(|delegation| !delegation.cleanup_complete)
        {
            let delegation = delegation_from_journal(stored);
            if self
                .revoke_for_finalization(&execution_id, &record.caller, &delegation)
                .await
                .is_err()
            {
                return;
            }
        }
        let _ = self
            .journal
            .transition_with_event(
                &execution_id,
                &[ExecutionState::Finalizing],
                ExecutionState::Failed,
                None,
                Some(ExecutionFailure {
                    code: "EXECUTION_INTERRUPTED".into(),
                    message: "execution was interrupted before a durable resumable checkpoint"
                        .into(),
                    retryable: true,
                }),
                EventPayload::ExecutionFailed {
                    code: "EXECUTION_INTERRUPTED".into(),
                    message: "execution was safely finalized after worker recovery".into(),
                },
            )
            .await;
    }

    async fn run_inner(&self, execution_id: &ExecutionId) -> Result<(), KernelFailure> {
        let record = self
            .journal
            .get(execution_id)
            .await
            .map_err(journal_kernel_error)?;
        let cancellation = CancellationToken::new();
        self.cancellations
            .lock()
            .await
            .insert(execution_id.clone(), cancellation.clone());
        let remaining_ms = record
            .view
            .created_at_ms
            .saturating_add((record.request.options.deadline_seconds * 1000) as i64)
            .saturating_sub(chrono::Utc::now().timestamp_millis())
            .max(1) as u64;
        let operation = OperationContext {
            id: OperationId::new(format!("{execution_id}:run"))
                .map_err(|error| KernelFailure::new("OPERATION_ID_INVALID", error.to_string()))?,
            execution_id: execution_id.clone(),
            deadline: Instant::now() + Duration::from_millis(remaining_ms),
            cancellation,
        };
        let resolved = self
            .instance_resolver
            .resolve(
                &operation,
                &record.caller,
                &record.request.runtime_instance_id,
            )
            .await
            .map_err(port_kernel_error)?;
        let resolved = apply_request_routing(resolved, &record.request);
        let definition = self
            .definition_resolver
            .resolve(&operation, &resolved)
            .await?;
        let scope = SessionScope {
            execution_id: execution_id.clone(),
            conversation_id: record.request.conversation_id.clone(),
            resolved: resolved.clone(),
        };
        let mut delegation = match record.delegation {
            Some(delegation) if !delegation.cleanup_complete => delegation_from_journal(delegation),
            Some(_) => {
                return Err(KernelFailure::new(
                    "DELEGATION_ALREADY_CLEANED",
                    "execution delegation was already revoked",
                ));
            }
            None => {
                let authority = self
                    .authorities
                    .lock()
                    .await
                    .get(execution_id)
                    .cloned()
                    .ok_or_else(|| KernelFailure::new(
                        "EXECUTION_AUTHORITY_MISSING",
                        "queued execution has no durable delegation and caller authority is unavailable",
                    ))?;
                let delegation = self
                    .session_factory
                    .establish_delegation(&operation, &authority, &scope)
                    .await
                    .map_err(port_kernel_error)?;
                self.journal
                    .attach_delegation(execution_id, journal_delegation(&delegation))
                    .await
                    .map_err(journal_kernel_error)?;
                delegation
            }
        };
        let now_seconds = chrono::Utc::now().timestamp().max(0) as u64;
        if delegation.expires_at_seconds
            <= now_seconds
                .saturating_add(operation.remaining().map_err(port_kernel_error)?.as_secs())
                .saturating_add(30)
        {
            delegation = self
                .session_factory
                .renew_delegation(&operation, &record.caller, &delegation)
                .await
                .map_err(port_kernel_error)?;
            self.journal
                .attach_delegation(execution_id, journal_delegation(&delegation))
                .await
                .map_err(journal_kernel_error)?;
        }
        self.journal
            .transition(
                execution_id,
                &[ExecutionState::Queued],
                ExecutionState::Running,
            )
            .await
            .map_err(journal_kernel_error)?;
        self.journal
            .append_event(execution_id, EventPayload::ExecutionStarted)
            .await
            .map_err(journal_kernel_error)?;
        let mut sessions = self
            .session_factory
            .create(&operation, &record.caller, &delegation, &scope)
            .await
            .map_err(port_kernel_error)?;
        sessions.interaction = Arc::new(JournaledInteractionSession {
            execution_id: execution_id.clone(),
            journal: self.journal.clone(),
            inner: sessions.interaction.clone(),
        });
        sessions.subagent = Arc::new(ChildExecutionSubagentSession {
            coordinator: self.clone(),
            authority: self.authorities.lock().await.get(execution_id).cloned(),
            parent_execution_id: execution_id.clone(),
            runtime_instance_id: record.request.runtime_instance_id.clone(),
            workspace_id: record.request.workspace_id.clone(),
            model: record.request.model.clone(),
            generation: record.request.generation.clone(),
        });
        let prompt = record
            .request
            .input
            .user_text()
            .ok_or_else(|| {
                KernelFailure::new(
                    "INITIAL_INPUT_INVALID",
                    "execution must start with at least one user message",
                )
            })?
            .to_string();
        let request_messages = request_messages(&record.request.input)?;
        let sink: Arc<dyn KernelEventSink> = Arc::new(JournalKernelEventSink {
            execution_id: execution_id.clone(),
            journal: self.journal.clone(),
            trace: sessions.trace.clone(),
            operation: operation.clone(),
        });
        let outcome = self
            .kernel
            .execute(
                operation,
                KernelSpec {
                    execution_id: execution_id.clone(),
                    conversation_id: record.request.conversation_id.clone(),
                    user_prompt: prompt,
                    request_messages,
                    model: resolved.model,
                    generation: model_generation_options(&record.request.generation),
                    definition,
                    granted_capabilities: record.caller.capabilities.iter().cloned().collect(),
                    limits: self.kernel_limits(&record.request.options, &record.request.generation),
                },
                sessions,
                sink,
            )
            .await?;
        let public = ExecutionOutcome {
            answer: outcome.answer.clone(),
            model_turns: outcome.model_turns,
            tool_calls: outcome.tool_calls,
            input_tokens: outcome.usage.input_tokens,
            output_tokens: outcome.usage.output_tokens,
        };
        self.journal
            .transition(
                execution_id,
                &[ExecutionState::Running],
                ExecutionState::Finalizing,
            )
            .await
            .map_err(journal_kernel_error)?;
        self.revoke_for_finalization(execution_id, &record.caller, &delegation)
            .await?;
        self.journal
            .transition_with_event(
                execution_id,
                &[ExecutionState::Finalizing],
                ExecutionState::Completed,
                Some(public),
                None,
                EventPayload::ExecutionCompleted {
                    answer: outcome.answer,
                },
            )
            .await
            .map_err(journal_kernel_error)?;
        Ok(())
    }

    async fn revoke_for_finalization(
        &self,
        execution_id: &ExecutionId,
        caller: &runtime_types::CallerScope,
        delegation: &ExecutionDelegationLease,
    ) -> Result<(), KernelFailure> {
        let operation = OperationContext {
            id: OperationId::new(format!("{execution_id}:cleanup"))
                .unwrap_or_else(|_| OperationId::random()),
            execution_id: execution_id.clone(),
            deadline: Instant::now() + Duration::from_secs(5),
            cancellation: CancellationToken::new(),
        };
        self.session_factory
            .revoke_delegation(&operation, caller, delegation)
            .await
            .map_err(|error| {
                let mut failure = port_kernel_error(error);
                failure.commit = CommitDisposition::Unknown;
                failure
            })?;
        self.journal
            .complete_delegation_cleanup(execution_id, &delegation.lease_ref)
            .await
            .map_err(journal_kernel_error)?;
        Ok(())
    }

    async fn cleanup_delegation(&self, execution_id: &ExecutionId) {
        let Ok(record) = self.journal.get(execution_id).await else {
            return;
        };
        let Some(stored) = record
            .delegation
            .filter(|delegation| !delegation.cleanup_complete)
        else {
            return;
        };
        if !record.view.state.is_terminal() {
            return;
        }
        let delegation = delegation_from_journal(stored);
        let operation = OperationContext {
            id: OperationId::new(format!("{execution_id}:cleanup"))
                .unwrap_or_else(|_| OperationId::random()),
            execution_id: execution_id.clone(),
            deadline: Instant::now() + Duration::from_secs(5),
            cancellation: CancellationToken::new(),
        };
        match self
            .session_factory
            .revoke_delegation(&operation, &record.caller, &delegation)
            .await
        {
            Ok(()) => {
                let _ = self
                    .journal
                    .complete_delegation_cleanup(execution_id, &delegation.lease_ref)
                    .await;
            }
            Err(error) => tracing::warn!(
                execution_id = %execution_id,
                code = %error.code,
                "delegation cleanup deferred to recovery"
            ),
        }
    }

    async fn finalize_failure(
        &self,
        execution_id: &ExecutionId,
        failure: ExecutionFailure,
    ) -> Result<ExecutionView, LinkError> {
        let mut record = self.journal.get(execution_id).await.map_err(map_journal)?;
        if record.view.state.is_terminal() {
            return Ok(record.view);
        }
        if record.view.state != ExecutionState::Finalizing {
            record = self
                .journal
                .transition(
                    execution_id,
                    &[
                        ExecutionState::Queued,
                        ExecutionState::Running,
                        ExecutionState::WaitingForInput,
                    ],
                    ExecutionState::Finalizing,
                )
                .await
                .map_err(map_journal)?;
        }
        if let Some(stored) = record
            .delegation
            .clone()
            .filter(|delegation| !delegation.cleanup_complete)
        {
            self.revoke_for_finalization(
                execution_id,
                &record.caller,
                &delegation_from_journal(stored),
            )
            .await
            .map_err(|error| {
                LinkError::Unavailable(format!("{}: {}", error.code, error.message))
            })?;
        }
        let event = EventPayload::ExecutionFailed {
            code: failure.code.clone(),
            message: failure.message.clone(),
        };
        self.journal
            .transition_with_event(
                execution_id,
                &[ExecutionState::Finalizing],
                ExecutionState::Failed,
                None,
                Some(failure),
                event,
            )
            .await
            .map(|mutation| mutation.execution.view)
            .map_err(map_journal)
    }

    async fn fail_queued(
        &self,
        execution_id: &ExecutionId,
        code: &str,
        message: &str,
        retryable: bool,
    ) -> Result<ExecutionView, LinkError> {
        self.finalize_failure(
            execution_id,
            ExecutionFailure {
                code: code.into(),
                message: message.into(),
                retryable,
            },
        )
        .await
    }

    async fn prepare_delegation(
        &self,
        record: &runtime_link_api::JournalExecution,
        authority: &RequestAuthority,
    ) -> Result<runtime_link_api::JournalExecution, LinkError> {
        if record.delegation.is_some() {
            return Ok(record.clone());
        }
        let remaining_ms = record
            .view
            .created_at_ms
            .saturating_add((record.request.options.deadline_seconds as i64) * 1000)
            .saturating_sub(chrono::Utc::now().timestamp_millis())
            .max(1) as u64;
        let operation = OperationContext {
            id: OperationId::new(format!("{}:delegate", record.view.id))
                .map_err(|error| LinkError::Internal(error.to_string()))?,
            execution_id: record.view.id.clone(),
            deadline: Instant::now() + Duration::from_millis(remaining_ms),
            cancellation: CancellationToken::new(),
        };
        let resolved = self
            .instance_resolver
            .resolve(
                &operation,
                &record.caller,
                &record.request.runtime_instance_id,
            )
            .await
            .map_err(map_port)?;
        let resolved = apply_request_routing(resolved, &record.request);
        let scope = SessionScope {
            execution_id: record.view.id.clone(),
            conversation_id: record.request.conversation_id.clone(),
            resolved,
        };
        let delegation = self
            .session_factory
            .establish_delegation(&operation, authority, &scope)
            .await
            .map_err(map_port)?;
        self.journal
            .attach_delegation(&record.view.id, journal_delegation(&delegation))
            .await
            .map_err(map_journal)
    }
}

#[async_trait]
impl RuntimeLink for ExecutionCoordinator {
    async fn create_execution(
        &self,
        authority: &RequestAuthority,
        idempotency_key: &str,
        request: CreateExecutionRequest,
    ) -> Result<CreateExecutionResponse, LinkError> {
        authority
            .caller
            .validate()
            .map_err(|error| LinkError::Invalid(error.to_string()))?;
        validate_create(idempotency_key, &request)?;
        if matches!(
            &request.input,
            RuntimeInput::Messages { messages }
                if messages.len() >= self.context_limits.max_messages
        ) {
            return Err(LinkError::Invalid(format!(
                "request messages must contain fewer than {} items",
                self.context_limits.max_messages
            )));
        }
        if !self.accepting.load(Ordering::SeqCst) {
            return Err(LinkError::Unavailable("runtime is shutting down".into()));
        }
        let now = chrono::Utc::now().timestamp_millis();
        let view = ExecutionView {
            id: ExecutionId::random(),
            runtime_instance_id: request.runtime_instance_id.clone(),
            conversation_id: request.conversation_id.clone(),
            workspace_id: request.workspace_id.clone(),
            model: request.model.clone(),
            metadata: request.metadata.clone(),
            state: ExecutionState::Queued,
            outcome: None,
            failure: None,
            created_at_ms: now,
            updated_at_ms: now,
        };
        match self
            .journal
            .reserve(NewJournalExecution {
                idempotency_key: idempotency_key.into(),
                caller: authority.caller.clone(),
                request,
                view,
            })
            .await
            .map_err(map_journal)?
        {
            JournalReservation::Existing(record) => {
                ensure_owner(&record.caller, authority)?;
                let record = if record.view.state == ExecutionState::Queued {
                    let prepared = self.prepare_delegation(&record, authority).await?;
                    if let Ok(permit) = self.capacity.clone().try_acquire_owned() {
                        self.authorities
                            .lock()
                            .await
                            .insert(prepared.view.id.clone(), authority.clone());
                        self.tasks
                            .spawn(self.clone().run(prepared.view.id.clone(), permit));
                    }
                    prepared
                } else {
                    record
                };
                Ok(CreateExecutionResponse {
                    execution: record.view,
                    replayed: true,
                })
            }
            JournalReservation::Created(record) => {
                self.journal
                    .append_event(&record.view.id, EventPayload::ExecutionQueued)
                    .await
                    .map_err(map_journal)?;
                let record = self.prepare_delegation(&record, authority).await?;
                let execution = if !self.accepting.load(Ordering::SeqCst) {
                    let failed = self
                        .fail_queued(
                            &record.view.id,
                            "RUNTIME_SHUTTING_DOWN",
                            "runtime stopped accepting work before dispatch",
                            true,
                        )
                        .await?;
                    self.cleanup_delegation(&record.view.id).await;
                    failed
                } else if let Ok(permit) = self.capacity.clone().try_acquire_owned() {
                    self.authorities
                        .lock()
                        .await
                        .insert(record.view.id.clone(), authority.clone());
                    let execution = record.view.clone();
                    self.tasks
                        .spawn(self.clone().run(execution.id.clone(), permit));
                    execution
                } else {
                    let failed = self
                        .fail_queued(
                            &record.view.id,
                            "CAPACITY_EXCEEDED",
                            "runtime execution capacity is exhausted",
                            true,
                        )
                        .await?;
                    self.cleanup_delegation(&record.view.id).await;
                    failed
                };
                Ok(CreateExecutionResponse {
                    execution,
                    replayed: false,
                })
            }
        }
    }

    async fn execution(
        &self,
        authority: &RequestAuthority,
        id: &ExecutionId,
    ) -> Result<ExecutionView, LinkError> {
        let record = self.journal.get(id).await.map_err(map_journal)?;
        ensure_owner(&record.caller, authority)?;
        Ok(record.view)
    }

    async fn events(
        &self,
        authority: &RequestAuthority,
        id: &ExecutionId,
        after: Option<u64>,
        limit: usize,
    ) -> Result<EventPage, LinkError> {
        if limit == 0 || limit > 1000 {
            return Err(LinkError::Invalid(
                "event limit must be within 1..=1000".into(),
            ));
        }
        ensure_owner(
            &self.journal.get(id).await.map_err(map_journal)?.caller,
            authority,
        )?;
        let items = self
            .journal
            .events(id, after, limit)
            .await
            .map_err(map_journal)?;
        let next_after = items.last().map(|event| event.sequence);
        Ok(EventPage { items, next_after })
    }

    async fn submit_input(
        &self,
        authority: &RequestAuthority,
        id: &ExecutionId,
        request: SubmitInputRequest,
    ) -> Result<ExecutionView, LinkError> {
        let record = self.journal.get(id).await.map_err(map_journal)?;
        ensure_owner(&record.caller, authority)?;
        if record.view.state != ExecutionState::WaitingForInput {
            return Err(LinkError::Conflict(
                "execution is not waiting for input".into(),
            ));
        }
        let (request_id, text) = match request.input {
            RuntimeInput::ElicitationResponse { request_id, text } => (request_id, text),
            RuntimeInput::ToolApproval {
                request_id,
                approved,
            } => (
                request_id,
                if approved {
                    "approved".into()
                } else {
                    "denied".into()
                },
            ),
            RuntimeInput::UserMessage { .. } | RuntimeInput::Messages { .. } => {
                return Err(LinkError::Invalid(
                    "waiting input requires tool_approval or elicitation_response".into(),
                ));
            }
        };
        if request_id.is_empty() || request_id.len() > 256 || text.len() > 1024 * 1024 {
            return Err(LinkError::Invalid(
                "interaction request ID must be 1..=256 bytes and response at most 1 MiB".into(),
            ));
        }
        let operation = OperationContext {
            id: request.operation_id.clone(),
            execution_id: id.clone(),
            deadline: Instant::now() + Duration::from_secs(30),
            cancellation: CancellationToken::new(),
        };
        let mutation = self
            .journal
            .commit_interaction_input(id, &request.operation_id, &request_id, &text)
            .await
            .map_err(map_journal)?;
        if let Err(error) = self
            .session_factory
            .submit_input(
                &operation,
                &record.caller,
                id,
                &request.operation_id,
                &request_id,
                InteractionResponse { text },
            )
            .await
        {
            // The durable receipt remains replayable. A retry with the same
            // operation/body can wake a waiter that appeared after this race.
            return Err(map_port(error));
        }
        Ok(mutation.execution.view)
    }

    async fn cancel(
        &self,
        authority: &RequestAuthority,
        id: &ExecutionId,
    ) -> Result<ExecutionView, LinkError> {
        let record = self.journal.get(id).await.map_err(map_journal)?;
        ensure_owner(&record.caller, authority)?;
        if record.view.state.is_terminal() {
            return Ok(record.view);
        }
        if let Some(token) = self.cancellations.lock().await.get(id) {
            token.cancel();
        }
        let record = if record.view.state == ExecutionState::Finalizing {
            record
        } else {
            self.journal
                .transition_with_event(
                    id,
                    &[
                        ExecutionState::Queued,
                        ExecutionState::Running,
                        ExecutionState::WaitingForInput,
                    ],
                    ExecutionState::Finalizing,
                    None,
                    None,
                    EventPayload::Warning {
                        code: "CANCELLATION_REQUESTED".into(),
                        message: "execution cancellation entered cleanup".into(),
                    },
                )
                .await
                .map_err(map_journal)?
                .execution
        };
        if let Some(stored) = record
            .delegation
            .clone()
            .filter(|delegation| !delegation.cleanup_complete)
        {
            self.revoke_for_finalization(id, &record.caller, &delegation_from_journal(stored))
                .await
                .map_err(|error| {
                    LinkError::Unavailable(format!("{}: {}", error.code, error.message))
                })?;
        }
        let updated = self
            .journal
            .transition_with_event(
                id,
                &[ExecutionState::Finalizing],
                ExecutionState::Canceled,
                None,
                None,
                EventPayload::ExecutionCanceled,
            )
            .await
            .map_err(map_journal)?;
        Ok(updated.execution.view)
    }
}

fn journal_delegation(delegation: &ExecutionDelegationLease) -> JournalDelegation {
    JournalDelegation {
        lease_ref: delegation.lease_ref.clone(),
        expires_at_seconds: delegation.expires_at_seconds,
        revision: delegation.revision,
        cleanup_complete: false,
    }
}

fn delegation_from_journal(delegation: JournalDelegation) -> ExecutionDelegationLease {
    ExecutionDelegationLease {
        lease_ref: delegation.lease_ref,
        expires_at_seconds: delegation.expires_at_seconds,
        revision: delegation.revision,
    }
}

struct JournalKernelEventSink {
    execution_id: ExecutionId,
    journal: Arc<dyn ExecutionJournal>,
    trace: Arc<dyn TraceSession>,
    operation: OperationContext,
}
#[async_trait]
impl KernelEventSink for JournalKernelEventSink {
    async fn emit(&self, event: KernelEvent) -> Result<(), KernelFailure> {
        let payload = match event {
            // Link commits Running + ExecutionStarted before acquiring Kernel
            // resources. Kernel's local Started signal must not duplicate it.
            KernelEvent::Started => return Ok(()),
            KernelEvent::ModelStarted {
                turn,
                invocation_id,
            } => EventPayload::ModelStarted {
                turn,
                invocation_id,
            },
            KernelEvent::ModelCompleted {
                turn,
                invocation_id,
                finish,
                usage,
            } => EventPayload::ModelCompleted {
                turn,
                invocation_id,
                finish_reason: match finish {
                    runtime_ports::ModelFinish::Stop => "stop",
                    runtime_ports::ModelFinish::ToolCalls => "tool_calls",
                    runtime_ports::ModelFinish::Length => "length",
                    runtime_ports::ModelFinish::ContentFilter => "content_filter",
                    runtime_ports::ModelFinish::Other => "other",
                }
                .into(),
                input_tokens: usage.input_tokens,
                output_tokens: usage.output_tokens,
            },
            KernelEvent::ToolStarted { call_id, name } => {
                EventPayload::ToolStarted { call_id, name }
            }
            KernelEvent::ToolCompleted {
                call_id,
                name,
                failed,
            } => EventPayload::ToolCompleted {
                call_id,
                name,
                failed,
            },
            KernelEvent::Warning { code, message } => EventPayload::Warning { code, message },
        };
        let event_type = match &payload {
            EventPayload::ModelStarted { .. } => "runtime.model.started",
            EventPayload::ModelCompleted { .. } => "runtime.model.completed",
            EventPayload::ToolStarted { .. } => "runtime.tool.started",
            EventPayload::ToolCompleted { .. } => "runtime.tool.completed",
            EventPayload::Warning { .. } => "runtime.warning",
            _ => "runtime.event",
        };
        let trace_payload = serde_json::to_value(&payload)
            .map_err(|error| KernelFailure::new("EVENT_SERIALIZATION_FAILED", error.to_string()))?;
        self.journal
            .append_event(&self.execution_id, payload)
            .await
            .map_err(journal_kernel_error)?;
        // Trace is an observation projection. A trace outage must not roll back
        // an event that has already become authoritative in the journal.
        if let Err(error) = self
            .trace
            .append(&self.operation, event_type, trace_payload)
            .await
        {
            tracing::warn!(
                execution_id = %self.execution_id,
                code = %error.code,
                "trace projection failed after journal commit"
            );
        }
        Ok(())
    }
}

struct JournaledInteractionSession {
    execution_id: ExecutionId,
    journal: Arc<dyn ExecutionJournal>,
    inner: Arc<dyn InteractionSession>,
}

struct ChildExecutionSubagentSession {
    coordinator: ExecutionCoordinator,
    authority: Option<RequestAuthority>,
    parent_execution_id: ExecutionId,
    runtime_instance_id: runtime_types::RuntimeInstanceId,
    workspace_id: Option<runtime_types::WorkspaceId>,
    model: Option<String>,
    generation: ApiModelGenerationOptions,
}

#[async_trait]
impl SubagentSession for ChildExecutionSubagentSession {
    async fn execute(
        &self,
        operation: &OperationContext,
        request: SubagentRequest,
    ) -> Result<SubagentOutcome, PortFailure> {
        let remaining = operation.remaining()?;
        let authority = self.authority.as_ref().ok_or_else(|| {
            PortFailure::new(
                PortFailureKind::Unavailable,
                "SUBAGENT_AUTHORITY_UNAVAILABLE",
                "recovered execution cannot widen its delegation with a new child execution",
            )
        })?;
        if request.prompt.trim().is_empty() || !(1..=256).contains(&request.max_model_turns) {
            return Err(PortFailure::new(
                PortFailureKind::Invalid,
                "SUBAGENT_REQUEST_INVALID",
                "subagent prompt and max_model_turns within 1..=256 are required",
            ));
        }
        let mut digest = Sha256::new();
        digest.update(self.parent_execution_id.as_str());
        digest.update(b":");
        digest.update(request.request_id.as_bytes());
        let idempotency_key = format!("subagent:{:x}", digest.finalize());
        let deadline_seconds = remaining.as_secs().saturating_add(1).clamp(1, 3600);
        let child = self
            .coordinator
            .create_execution(
                authority,
                &idempotency_key,
                CreateExecutionRequest {
                    runtime_instance_id: self.runtime_instance_id.clone(),
                    conversation_id: None,
                    input: RuntimeInput::UserMessage {
                        text: request.prompt,
                    },
                    workspace_id: self.workspace_id.clone(),
                    model: self.model.clone(),
                    instructions: None,
                    metadata: BTreeMap::new(),
                    generation: self.generation.clone(),
                    options: ExecutionOptions {
                        deadline_seconds,
                        max_model_turns: request.max_model_turns,
                        max_tool_calls: request.max_model_turns.saturating_mul(8).clamp(1, 2048),
                    },
                },
            )
            .await
            .map_err(link_port_error)?
            .execution;
        loop {
            if operation.cancellation.is_cancelled() {
                let _ = self.coordinator.cancel(authority, &child.id).await;
                return Err(PortFailure::canceled());
            }
            operation.remaining()?;
            let view = self
                .coordinator
                .execution(authority, &child.id)
                .await
                .map_err(link_port_error)?;
            match view.state {
                ExecutionState::Completed => {
                    return Ok(SubagentOutcome {
                        answer: view
                            .outcome
                            .map(|outcome| outcome.answer)
                            .unwrap_or_default(),
                    });
                }
                ExecutionState::Failed => {
                    let failure = view.failure.unwrap_or(ExecutionFailure {
                        code: "SUBAGENT_FAILED".into(),
                        message: "subagent failed without a failure payload".into(),
                        retryable: false,
                    });
                    let mut error = PortFailure::new(
                        PortFailureKind::Unavailable,
                        failure.code,
                        failure.message,
                    );
                    error.retryable = failure.retryable;
                    return Err(error);
                }
                ExecutionState::Canceled => return Err(PortFailure::canceled()),
                ExecutionState::WaitingForInput => {
                    let _ = self.coordinator.cancel(authority, &child.id).await;
                    return Err(PortFailure::new(
                        PortFailureKind::Conflict,
                        "SUBAGENT_INTERACTION_UNSUPPORTED",
                        "a nested execution cannot request direct user input",
                    ));
                }
                ExecutionState::Queued | ExecutionState::Running | ExecutionState::Finalizing => {}
            }
            tokio::select! {
                _ = operation.cancellation.cancelled() => {
                    let _ = self.coordinator.cancel(authority, &child.id).await;
                    return Err(PortFailure::canceled());
                }
                _ = tokio::time::sleep(Duration::from_millis(25)) => {}
            }
        }
    }
}
#[async_trait]
impl InteractionSession for JournaledInteractionSession {
    async fn prepare(
        &self,
        _: &OperationContext,
        _: InteractionRequest,
    ) -> Result<(), PortFailure> {
        Err(PortFailure::new(
            PortFailureKind::Internal,
            "INTERACTION_WRAPPER_PROTOCOL",
            "journaled interaction must use request()",
        ))
    }

    async fn wait(
        &self,
        _: &OperationContext,
        _: &str,
    ) -> Result<InteractionResponse, PortFailure> {
        Err(PortFailure::new(
            PortFailureKind::Internal,
            "INTERACTION_WRAPPER_PROTOCOL",
            "journaled interaction must use request()",
        ))
    }

    async fn request(
        &self,
        operation: &OperationContext,
        request: InteractionRequest,
    ) -> Result<InteractionResponse, PortFailure> {
        self.inner.prepare(operation, request.clone()).await?;
        self.journal
            .begin_interaction(&self.execution_id, &request.request_id, &request.prompt)
            .await
            .map_err(journal_port_error)?;
        let result = self.inner.wait(operation, &request.request_id).await;
        match result {
            Ok(response) => {
                self.journal
                    .complete_interaction(&self.execution_id, &request.request_id)
                    .await
                    .map_err(journal_port_error)?;
                Ok(response)
            }
            Err(error) => Err(error),
        }
    }
}

fn validate_create(key: &str, request: &CreateExecutionRequest) -> Result<(), LinkError> {
    if key.is_empty() || key.len() > 256 {
        return Err(LinkError::Invalid(
            "Idempotency-Key must contain 1..=256 bytes".into(),
        ));
    }
    request
        .options
        .validate()
        .map_err(|message| LinkError::Invalid(message.into()))?;
    request
        .generation
        .validate()
        .map_err(|message| LinkError::Invalid(message.into()))?;
    if request
        .model
        .as_ref()
        .is_some_and(|model| model.trim().is_empty())
    {
        return Err(LinkError::Invalid("model must not be empty".into()));
    }
    if request
        .instructions
        .as_ref()
        .is_some_and(|instructions| instructions.len() > 256 * 1024)
    {
        return Err(LinkError::Invalid(
            "instructions must not exceed 256 KiB".into(),
        ));
    }
    let user_metadata_count = request
        .metadata
        .keys()
        .filter(|key| key.as_str() != "requestId")
        .count();
    if user_metadata_count > 16
        || request
            .metadata
            .iter()
            .any(|(key, value)| key.is_empty() || key.len() > 64 || value.len() > 512)
    {
        return Err(LinkError::Invalid(
            "metadata supports at most 16 entries with 1..=64 byte keys and values up to 512 bytes"
                .into(),
        ));
    }
    match &request.input {
        RuntimeInput::UserMessage { text }
            if !text.trim().is_empty() && text.len() <= 1024 * 1024 =>
        {
            Ok(())
        }
        RuntimeInput::Messages { messages }
            if !messages.is_empty()
                && messages.len() <= 256
                && messages.iter().any(|message| {
                    message.role == runtime_api_contract::RuntimeMessageRole::User
                        && !message.text.trim().is_empty()
                })
                && messages
                    .iter()
                    .all(|message| !message.text.trim().is_empty() && message.text.len() <= 1024 * 1024) =>
        {
            Ok(())
        }
        _ => Err(LinkError::Invalid(
            "initial input must contain 1..=256 non-empty text messages of at most 1 MiB each and at least one user role".into(),
        )),
    }
}

fn request_messages(
    input: &RuntimeInput,
) -> Result<Vec<runtime_ports::ConversationMessage>, KernelFailure> {
    use runtime_api_contract::RuntimeMessageRole;
    use runtime_ports::ModelRole;

    match input {
        RuntimeInput::UserMessage { text } => Ok(vec![runtime_ports::ConversationMessage {
            role: ModelRole::User,
            text: text.clone(),
        }]),
        RuntimeInput::Messages { messages } => Ok(messages
            .iter()
            .map(|message| runtime_ports::ConversationMessage {
                role: match message.role {
                    RuntimeMessageRole::System | RuntimeMessageRole::Developer => ModelRole::System,
                    RuntimeMessageRole::User => ModelRole::User,
                    RuntimeMessageRole::Assistant => ModelRole::Assistant,
                },
                text: message.text.clone(),
            })
            .collect()),
        RuntimeInput::ToolApproval { .. } | RuntimeInput::ElicitationResponse { .. } => {
            Err(KernelFailure::new(
                "INITIAL_INPUT_INVALID",
                "execution must start with messages",
            ))
        }
    }
}

/// Apply task-scoped routing only after the immutable Runtime binding has been
/// resolved. Authorization is intentionally not decided here: the Agent Infra
/// delegation is established from the resulting scope and remains the source
/// of truth for model and workspace access.
fn apply_request_routing(
    mut resolved: ResolvedExecutionContext,
    request: &CreateExecutionRequest,
) -> ResolvedExecutionContext {
    if let Some(workspace_id) = &request.workspace_id {
        resolved.workspace_id = workspace_id.clone();
    }
    if let Some(model) = &request.model {
        resolved.model = model.clone();
    }
    if let Some(instructions) = &request.instructions {
        resolved
            .metadata
            .entry("instructions".into())
            .and_modify(|current| {
                current.push_str("\n\n");
                current.push_str(instructions);
            })
            .or_insert_with(|| instructions.clone());
    }
    for (key, value) in &request.metadata {
        resolved
            .metadata
            .insert(format!("task.{key}"), value.clone());
    }
    resolved
}

fn ensure_owner(
    caller: &runtime_types::CallerScope,
    authority: &RequestAuthority,
) -> Result<(), LinkError> {
    if caller.subject == authority.caller.subject
        && caller.tenant_id == authority.caller.tenant_id
        && caller.project_id == authority.caller.project_id
    {
        Ok(())
    } else {
        Err(LinkError::Forbidden)
    }
}

impl ExecutionCoordinator {
    fn kernel_limits(
        &self,
        options: &ExecutionOptions,
        generation: &ApiModelGenerationOptions,
    ) -> KernelLimits {
        let mut context = self.context_limits.clone();
        if let Some(max_output_tokens) = generation.max_output_tokens {
            context.reserved_output_tokens = max_output_tokens;
        }
        KernelLimits {
            max_model_turns: options.max_model_turns,
            max_tool_calls: options.max_tool_calls,
            context,
            max_tool_output_bytes: 256 * 1024,
        }
    }
}

fn model_generation_options(options: &ApiModelGenerationOptions) -> ModelGenerationOptions {
    ModelGenerationOptions {
        max_output_tokens: options.max_output_tokens,
        temperature: options.temperature,
        stop_sequences: options.stop_sequences.clone(),
        response_format: options.response_format.clone(),
        tool_choice: match &options.tool_choice {
            ApiModelToolChoice::Auto => ModelToolChoice::Auto,
            ApiModelToolChoice::None => ModelToolChoice::None,
        },
    }
}
fn map_journal(error: JournalError) -> LinkError {
    match error {
        JournalError::NotFound => LinkError::NotFound,
        JournalError::Conflict => LinkError::Conflict("journal compare-and-set failed".into()),
        JournalError::Unavailable(message) => LinkError::Unavailable(message),
    }
}
fn map_port(error: PortFailure) -> LinkError {
    LinkError::Unavailable(format!("{}: {}", error.code, error.message))
}
fn link_port_error(error: LinkError) -> PortFailure {
    match error {
        LinkError::Invalid(message) => {
            PortFailure::new(PortFailureKind::Invalid, "SUBAGENT_INVALID", message)
        }
        LinkError::NotFound => PortFailure::new(
            PortFailureKind::NotFound,
            "SUBAGENT_NOT_FOUND",
            "nested execution was not found",
        ),
        LinkError::Forbidden => PortFailure::new(
            PortFailureKind::Forbidden,
            "SUBAGENT_FORBIDDEN",
            "nested execution is forbidden",
        ),
        LinkError::Conflict(message) => {
            PortFailure::new(PortFailureKind::Conflict, "SUBAGENT_CONFLICT", message)
        }
        LinkError::Overloaded => PortFailure::new(
            PortFailureKind::Unavailable,
            "SUBAGENT_OVERLOADED",
            "nested execution capacity is exhausted",
        ),
        LinkError::Unavailable(message) => PortFailure::new(
            PortFailureKind::Unavailable,
            "SUBAGENT_UNAVAILABLE",
            message,
        ),
        LinkError::Internal(message) => {
            PortFailure::new(PortFailureKind::Internal, "SUBAGENT_INTERNAL", message)
        }
    }
}
fn journal_kernel_error(error: JournalError) -> KernelFailure {
    KernelFailure::new("JOURNAL_FAILURE", error.to_string())
}
fn port_kernel_error(error: PortFailure) -> KernelFailure {
    KernelFailure {
        code: error.code,
        message: error.message,
        retryable: error.retryable,
        commit: error.commit,
    }
}
fn journal_port_error(error: JournalError) -> PortFailure {
    PortFailure::new(
        runtime_ports::PortFailureKind::Unavailable,
        "JOURNAL_FAILURE",
        error.to_string(),
    )
}

#[cfg(test)]
mod tests {
    use std::{collections::BTreeMap, sync::Arc};

    use agent_runtime_code_agent::definition as code_agent_definition;
    use agent_runtime_testkit::{ScriptedModelSession, sessions};
    use runtime_adapter_execution_journal::MemoryExecutionJournal;
    use runtime_agent_definition_loader::DefinitionCatalogResolver;
    use runtime_api_contract::{
        CreateExecutionRequest, ExecutionOptions, ExecutionState, RuntimeInput,
    };
    use runtime_kernel::AgentKernel;
    use runtime_link_api::RuntimeLink;
    use runtime_ports::{
        CommitDisposition, ExecutionSessions, InteractionResponse, ModelContent, ModelFinish,
        ModelResponse, PortFailureKind, ResolvedExecutionContext, TokenUsage,
    };
    use runtime_types::{
        CallerScope, CredentialHandle, DelegationLeaseRef, RuntimeInstanceId, WorkspaceId,
    };
    use serde_json::json;

    use super::*;

    struct FixtureResolver;
    #[async_trait]
    impl RuntimeInstanceResolver for FixtureResolver {
        async fn resolve(
            &self,
            _: &OperationContext,
            _: &CallerScope,
            id: &RuntimeInstanceId,
        ) -> Result<ResolvedExecutionContext, PortFailure> {
            Ok(ResolvedExecutionContext {
                runtime_instance_id: id.clone(),
                agent_id: "agt_test".into(),
                runtime_type: "code".into(),
                model: "fixture".into(),
                workspace_id: WorkspaceId::new("workspace-1").unwrap(),
                definition_id: "runtime-code-agent".into(),
                definition_version: code_agent_definition()
                    .expect("valid Code Agent definition")
                    .manifest
                    .version
                    .clone(),
                definition_digest: None,
                metadata: BTreeMap::new(),
            })
        }
    }

    fn code_agent_resolver() -> Arc<dyn AgentDefinitionResolver> {
        let definition = code_agent_definition().expect("valid Code Agent definition");
        Arc::new(
            DefinitionCatalogResolver::new([definition])
                .expect("single Code Agent definition catalog"),
        )
    }

    struct FixtureFactory {
        model: Arc<dyn runtime_ports::ModelSession>,
    }

    struct HangingModel;
    #[async_trait]
    impl runtime_ports::ModelSession for HangingModel {
        async fn invoke(
            &self,
            operation: &OperationContext,
            _: runtime_ports::ModelRequest,
        ) -> Result<ModelResponse, PortFailure> {
            operation.cancellation.cancelled().await;
            Err(PortFailure::canceled())
        }
    }

    struct CommitUnknownModel;
    #[async_trait]
    impl runtime_ports::ModelSession for CommitUnknownModel {
        async fn invoke(
            &self,
            _: &OperationContext,
            _: runtime_ports::ModelRequest,
        ) -> Result<ModelResponse, PortFailure> {
            let mut failure = PortFailure::new(
                PortFailureKind::Unavailable,
                "MODEL_TRANSPORT_UNKNOWN",
                "model transport failed after dispatch",
            );
            failure.commit = CommitDisposition::Unknown;
            Err(failure)
        }
    }
    #[async_trait]
    impl ExecutionSessionFactory for FixtureFactory {
        async fn establish_delegation(
            &self,
            _: &OperationContext,
            _: &RequestAuthority,
            _: &SessionScope,
        ) -> Result<ExecutionDelegationLease, PortFailure> {
            Ok(ExecutionDelegationLease {
                lease_ref: DelegationLeaseRef::new("edl_fixture").unwrap(),
                expires_at_seconds: u64::MAX,
                revision: 1,
            })
        }

        async fn create(
            &self,
            _: &OperationContext,
            _: &CallerScope,
            _: &ExecutionDelegationLease,
            _: &SessionScope,
        ) -> Result<ExecutionSessions, PortFailure> {
            Ok(sessions(self.model.clone()))
        }
        async fn submit_input(
            &self,
            _: &OperationContext,
            _: &CallerScope,
            _: &ExecutionId,
            _: &OperationId,
            _: &str,
            _: InteractionResponse,
        ) -> Result<(), PortFailure> {
            Err(PortFailure::new(
                PortFailureKind::Conflict,
                "NO_INTERACTION",
                "fixture has no pending interaction",
            ))
        }

        async fn renew_delegation(
            &self,
            _: &OperationContext,
            _: &CallerScope,
            delegation: &ExecutionDelegationLease,
        ) -> Result<ExecutionDelegationLease, PortFailure> {
            Ok(delegation.clone())
        }

        async fn revoke_delegation(
            &self,
            _: &OperationContext,
            _: &CallerScope,
            _: &ExecutionDelegationLease,
        ) -> Result<(), PortFailure> {
            Ok(())
        }
    }

    #[tokio::test]
    async fn create_runs_code_distribution_to_durable_terminal_state() {
        let model = Arc::new(ScriptedModelSession::new(vec![ModelResponse {
            output: vec![ModelContent::ToolUse {
                id: "call-1".into(),
                name: "runtime.complete".into(),
                arguments: json!({"answer":"complete"}),
            }],
            finish: ModelFinish::ToolCalls,
            usage: TokenUsage::default(),
        }]));
        let coordinator = ExecutionCoordinator::new(
            Arc::new(MemoryExecutionJournal::new(100)),
            Arc::new(FixtureResolver),
            Arc::new(FixtureFactory { model }),
            code_agent_resolver(),
            Arc::new(AgentKernel),
            2,
        )
        .unwrap();
        let authority = RequestAuthority {
            caller: CallerScope {
                subject: "user-1".into(),
                tenant_id: "tenant-1".into(),
                project_id: "project-1".into(),
                capabilities: Vec::new(),
            },
            credential: CredentialHandle::new("test-token").unwrap(),
        };
        let created = coordinator
            .create_execution(
                &authority,
                "test-key",
                CreateExecutionRequest {
                    runtime_instance_id: RuntimeInstanceId::new("runtime-1").unwrap(),
                    conversation_id: None,
                    input: RuntimeInput::UserMessage {
                        text: "do it".into(),
                    },
                    workspace_id: None,
                    model: None,
                    instructions: None,
                    metadata: BTreeMap::new(),
                    generation: ApiModelGenerationOptions::default(),
                    options: ExecutionOptions::default(),
                },
            )
            .await
            .unwrap();
        let mut view = created.execution;
        for _ in 0..100 {
            view = coordinator.execution(&authority, &view.id).await.unwrap();
            if view.state.is_terminal() {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert_eq!(view.state, ExecutionState::Completed, "{view:?}");
        assert_eq!(view.outcome.unwrap().answer, "complete");
        let events = coordinator
            .events(&authority, &view.id, None, 100)
            .await
            .unwrap();
        assert!(matches!(
            events.items.first().unwrap().payload,
            EventPayload::ExecutionQueued
        ));
        assert!(matches!(
            events.items.last().unwrap().payload,
            EventPayload::ExecutionCompleted { .. }
        ));
    }

    #[tokio::test]
    async fn recovery_resumes_queued_execution_from_durable_delegation_without_caller_token() {
        let model = Arc::new(ScriptedModelSession::new(vec![ModelResponse {
            output: vec![ModelContent::ToolUse {
                id: "call-recovered".into(),
                name: "runtime.complete".into(),
                arguments: json!({"answer":"recovered"}),
            }],
            finish: ModelFinish::ToolCalls,
            usage: TokenUsage::default(),
        }]));
        let journal = Arc::new(MemoryExecutionJournal::new(100));
        let request = CreateExecutionRequest {
            runtime_instance_id: RuntimeInstanceId::new("runtime-1").unwrap(),
            conversation_id: None,
            input: RuntimeInput::UserMessage {
                text: "resume".into(),
            },
            workspace_id: None,
            model: None,
            instructions: None,
            metadata: BTreeMap::new(),
            generation: ApiModelGenerationOptions::default(),
            options: ExecutionOptions::default(),
        };
        let now = chrono::Utc::now().timestamp_millis();
        let id = ExecutionId::random();
        journal
            .reserve(NewJournalExecution {
                idempotency_key: "recovery-key".into(),
                caller: CallerScope {
                    subject: "user-1".into(),
                    tenant_id: "tenant-1".into(),
                    project_id: "project-1".into(),
                    capabilities: Vec::new(),
                },
                request: request.clone(),
                view: runtime_api_contract::ExecutionView {
                    id: id.clone(),
                    runtime_instance_id: request.runtime_instance_id.clone(),
                    conversation_id: None,
                    workspace_id: request.workspace_id.clone(),
                    model: request.model.clone(),
                    metadata: request.metadata.clone(),
                    state: ExecutionState::Queued,
                    outcome: None,
                    failure: None,
                    created_at_ms: now,
                    updated_at_ms: now,
                },
            })
            .await
            .unwrap();
        journal
            .attach_delegation(
                &id,
                JournalDelegation {
                    lease_ref: DelegationLeaseRef::new("edl_recovered").unwrap(),
                    expires_at_seconds: u64::MAX,
                    revision: 1,
                    cleanup_complete: false,
                },
            )
            .await
            .unwrap();
        let coordinator = ExecutionCoordinator::new(
            journal.clone(),
            Arc::new(FixtureResolver),
            Arc::new(FixtureFactory { model }),
            code_agent_resolver(),
            Arc::new(AgentKernel),
            2,
        )
        .unwrap();
        assert_eq!(coordinator.recover(10).await.unwrap(), 1);
        for _ in 0..100 {
            let record = journal.get(&id).await.unwrap();
            if record.view.state.is_terminal() {
                assert_eq!(record.view.state, ExecutionState::Completed);
                assert_eq!(record.view.outcome.unwrap().answer, "recovered");
                assert!(record.delegation.unwrap().cleanup_complete);
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("recovered execution did not reach terminal state");
    }

    #[tokio::test]
    async fn periodic_recovery_picks_up_claim_that_expires_after_startup_scan() {
        let model = Arc::new(ScriptedModelSession::new(vec![ModelResponse {
            output: vec![ModelContent::ToolUse {
                id: "call-delayed-recovery".into(),
                name: "runtime.complete".into(),
                arguments: json!({"answer":"delayed recovery"}),
            }],
            finish: ModelFinish::ToolCalls,
            usage: TokenUsage::default(),
        }]));
        let journal = Arc::new(MemoryExecutionJournal::new(100));
        let request = request("resume after claim expiry");
        let now = chrono::Utc::now().timestamp_millis();
        let id = ExecutionId::random();
        journal
            .reserve(NewJournalExecution {
                idempotency_key: "delayed-recovery-key".into(),
                caller: authority().caller,
                request: request.clone(),
                view: ExecutionView {
                    id: id.clone(),
                    runtime_instance_id: request.runtime_instance_id.clone(),
                    conversation_id: None,
                    workspace_id: request.workspace_id.clone(),
                    model: request.model.clone(),
                    metadata: request.metadata.clone(),
                    state: ExecutionState::Queued,
                    outcome: None,
                    failure: None,
                    created_at_ms: now,
                    updated_at_ms: now,
                },
            })
            .await
            .unwrap();
        journal
            .attach_delegation(
                &id,
                JournalDelegation {
                    lease_ref: DelegationLeaseRef::new("edl_delayed_recovery").unwrap(),
                    expires_at_seconds: u64::MAX,
                    revision: 1,
                    cleanup_complete: false,
                },
            )
            .await
            .unwrap();
        journal
            .claim(&id, "dead-worker", now, now + 50)
            .await
            .unwrap();
        let coordinator = ExecutionCoordinator::new(
            journal.clone(),
            Arc::new(FixtureResolver),
            Arc::new(FixtureFactory { model }),
            code_agent_resolver(),
            Arc::new(AgentKernel),
            1,
        )
        .unwrap();
        assert_eq!(coordinator.recover(10).await.unwrap(), 0);
        coordinator
            .start_recovery_loop(Duration::from_millis(10), 10)
            .unwrap();
        for _ in 0..100 {
            let record = journal.get(&id).await.unwrap();
            if record.view.state == ExecutionState::Completed {
                assert_eq!(record.view.outcome.unwrap().answer, "delayed recovery");
                assert!(coordinator.shutdown(Duration::from_secs(1)).await);
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        coordinator.begin_shutdown();
        panic!("periodic recovery did not pick up the expired claim");
    }

    #[tokio::test]
    async fn shutdown_cancels_registered_execution_tasks_after_grace() {
        let journal = Arc::new(MemoryExecutionJournal::new(100));
        let coordinator = ExecutionCoordinator::new(
            journal.clone(),
            Arc::new(FixtureResolver),
            Arc::new(FixtureFactory {
                model: Arc::new(HangingModel),
            }),
            code_agent_resolver(),
            Arc::new(AgentKernel),
            1,
        )
        .unwrap();
        let authority = authority();
        let created = coordinator
            .create_execution(&authority, "shutdown", request("wait"))
            .await
            .unwrap()
            .execution;
        for _ in 0..100 {
            if coordinator
                .execution(&authority, &created.id)
                .await
                .unwrap()
                .state
                == ExecutionState::Running
            {
                break;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        assert!(coordinator.shutdown(Duration::from_millis(1)).await);
        assert_eq!(coordinator.active_executions(), 0);
        let terminal = coordinator
            .execution(&authority, &created.id)
            .await
            .unwrap();
        assert_eq!(terminal.state, ExecutionState::Failed);
        assert_eq!(terminal.failure.unwrap().code, "CANCELED");
        assert!(
            journal
                .get(&created.id)
                .await
                .unwrap()
                .delegation
                .unwrap()
                .cleanup_complete,
            "Failed must imply durable delegation cleanup"
        );
    }

    #[tokio::test]
    async fn commit_unknown_stays_finalizing_for_recovery() {
        let coordinator = ExecutionCoordinator::new(
            Arc::new(MemoryExecutionJournal::new(100)),
            Arc::new(FixtureResolver),
            Arc::new(FixtureFactory {
                model: Arc::new(CommitUnknownModel),
            }),
            code_agent_resolver(),
            Arc::new(AgentKernel),
            1,
        )
        .unwrap();
        let authority = authority();
        let execution = coordinator
            .create_execution(&authority, "commit-unknown", request("invoke"))
            .await
            .unwrap()
            .execution;
        for _ in 0..100 {
            let view = coordinator
                .execution(&authority, &execution.id)
                .await
                .unwrap();
            if view.state == ExecutionState::Finalizing && coordinator.active_executions() == 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        let view = coordinator
            .execution(&authority, &execution.id)
            .await
            .unwrap();
        assert_eq!(view.state, ExecutionState::Finalizing);
        assert!(view.failure.is_none());
        let events = coordinator
            .events(&authority, &execution.id, None, 100)
            .await
            .unwrap();
        assert!(matches!(
            events.items.last().unwrap().payload,
            EventPayload::Warning { ref code, .. } if code == "COMMIT_DISPOSITION_UNKNOWN"
        ));
    }

    fn authority() -> RequestAuthority {
        RequestAuthority {
            caller: CallerScope {
                subject: "user-1".into(),
                tenant_id: "tenant-1".into(),
                project_id: "project-1".into(),
                capabilities: Vec::new(),
            },
            credential: CredentialHandle::new("test-token").unwrap(),
        }
    }

    fn request(prompt: &str) -> CreateExecutionRequest {
        CreateExecutionRequest {
            runtime_instance_id: RuntimeInstanceId::new("runtime-1").unwrap(),
            conversation_id: None,
            input: RuntimeInput::UserMessage {
                text: prompt.into(),
            },
            workspace_id: None,
            model: None,
            instructions: None,
            metadata: BTreeMap::new(),
            generation: ApiModelGenerationOptions::default(),
            options: ExecutionOptions::default(),
        }
    }

    #[test]
    fn task_routing_overrides_workspace_model_and_augments_instructions() {
        let mut execution = request("inspect");
        execution.workspace_id = Some(WorkspaceId::new("workspace-2").unwrap());
        execution.model = Some("model-2".into());
        execution.instructions = Some("request instruction".into());
        execution
            .metadata
            .insert("traceId".into(), "trace-1".into());
        let mut metadata = BTreeMap::new();
        metadata.insert("instructions".into(), "binding instruction".into());
        let resolved = apply_request_routing(
            ResolvedExecutionContext {
                runtime_instance_id: RuntimeInstanceId::new("runtime-1").unwrap(),
                agent_id: "agt_test".into(),
                runtime_type: "test".into(),
                model: "model-1".into(),
                workspace_id: WorkspaceId::new("workspace-1").unwrap(),
                definition_id: "definition-1".into(),
                definition_version: "1".into(),
                definition_digest: None,
                metadata,
            },
            &execution,
        );
        assert_eq!(resolved.workspace_id.as_str(), "workspace-2");
        assert_eq!(resolved.model, "model-2");
        assert_eq!(
            resolved.metadata["instructions"],
            "binding instruction\n\nrequest instruction"
        );
        assert_eq!(resolved.metadata["task.traceId"], "trace-1");
    }
}