meerkat-runtime 0.6.4

use super::*;

impl MeerkatMachine {
    fn classify_ingress_dsl_rejection(state: RuntimeState, reason: String) -> RuntimeDriverError {
        match state {
            RuntimeState::Destroyed => RuntimeDriverError::Destroyed,
            RuntimeState::Retired | RuntimeState::Stopped => RuntimeDriverError::NotReady { state },
            _ => RuntimeDriverError::ValidationFailed { reason },
        }
    }

    fn reject_visible_terminal_ingress(state: RuntimeState) -> Result<(), RuntimeDriverError> {
        match state {
            RuntimeState::Destroyed => Err(RuntimeDriverError::Destroyed),
            RuntimeState::Retired | RuntimeState::Stopped => {
                Err(RuntimeDriverError::NotReady { state })
            }
            _ => Ok(()),
        }
    }

    pub(super) async fn execute_meerkat_machine_ingress_command(
        &self,
        command: MeerkatMachineCommand,
    ) -> Result<MeerkatMachineCommandResult, RuntimeDriverError> {
        match command {
            MeerkatMachineCommand::AcceptWithCompletion { session_id, input } => {
                let (driver, completions, wake_tx, effect_tx, boundary_handle) = {
                    let sessions = self.sessions.read().await;
                    let entry = sessions
                        .get(&session_id)
                        .ok_or(RuntimeDriverError::NotReady {
                            state: RuntimeState::Destroyed,
                        })?;
                    (
                        entry.driver.clone(),
                        entry.completions.clone(),
                        entry.wake_sender(),
                        entry.effect_sender(),
                        entry.boundary_handle(),
                    )
                };

                let gate = self.session_mutation_gate(&session_id).await;
                let _gate_guard = match gate {
                    Some(ref g) => Some(g.lock().await),
                    None => None,
                };

                let state = self
                    .existing_session_runtime_state(&session_id)
                    .await
                    .unwrap_or(RuntimeState::Destroyed);
                let visible_state = self
                    .existing_session_visible_runtime_state(&session_id)
                    .await
                    .unwrap_or(RuntimeState::Destroyed);
                Self::reject_visible_terminal_ingress(visible_state)?;

                let (resolved, outcome, handle, accepted_input_id, signal) = {
                    let mut driver = driver.lock().await;
                    let runtime_idle = state.is_idle_or_attached();
                    let resolved = driver.resolve_admission_for_runtime_idle(&input, runtime_idle);
                    self.preview_session_dsl_input(
                        &session_id,
                        crate::meerkat_machine::dsl::MeerkatMachineInput::AcceptWithCompletion {
                            input_id: crate::meerkat_machine::dsl::InputId::from_domain(
                                &InputId::new(),
                            ),
                            request_immediate_processing: resolved
                                .coarse_flags
                                .request_immediate_processing,
                            interrupt_yielding: resolved.coarse_flags.interrupt_yielding,
                            wake_if_idle: resolved.coarse_flags.wake_if_idle,
                        },
                        "AcceptWithCompletion",
                    )
                    .await
                    .map_err(|reason| {
                        let reason = format!(
                            "{reason}; input_kind={}; immediate={}; interrupt_yielding={}; wake_if_idle={}",
                            input.kind(),
                            resolved.coarse_flags.request_immediate_processing,
                            resolved.coarse_flags.interrupt_yielding,
                            resolved.coarse_flags.wake_if_idle,
                        );
                        Self::classify_ingress_dsl_rejection(state, reason)
                    })?;
                    let result = match driver
                        .accept_resolved_input(input, resolved.clone())
                        .await
                        .map_err(Self::normalize_destroyed_error)
                    {
                        Ok(r) => r,
                        Err(err) => return Err(err),
                    };

                    match &result {
                        AcceptOutcome::Accepted { input_id, .. } => {
                            let accepted_input_id = input_id.clone();
                            let is_terminal = driver
                                .as_driver()
                                .input_phase(&accepted_input_id)
                                .map(|phase| phase.is_terminal())
                                .unwrap_or(true);
                            let handle = if is_terminal {
                                None
                            } else {
                                Some({
                                    let mut completions = completions.lock().await;
                                    completions.register(accepted_input_id.clone())
                                })
                            };
                            (
                                resolved,
                                result,
                                handle,
                                Some(accepted_input_id),
                                crate::driver::ephemeral::PostAdmissionSignal::None,
                            )
                        }
                        AcceptOutcome::Deduplicated { existing_id, .. } => {
                            let is_terminal = driver
                                .as_driver()
                                .input_phase(existing_id)
                                .map(|phase| phase.is_terminal())
                                .unwrap_or(true);

                            if is_terminal {
                                (
                                    resolved,
                                    result,
                                    None,
                                    None,
                                    crate::driver::ephemeral::PostAdmissionSignal::None,
                                )
                            } else {
                                let handle = {
                                    let mut completions = completions.lock().await;
                                    completions.register(existing_id.clone())
                                };
                                (
                                    resolved,
                                    result,
                                    Some(handle),
                                    None,
                                    crate::driver::ephemeral::PostAdmissionSignal::None,
                                )
                            }
                        }
                        AcceptOutcome::Rejected { reason } => {
                            return Err(RuntimeDriverError::ValidationFailed {
                                reason: reason.to_string(),
                            });
                        }
                    }
                };
                let (signal, runtime_effect, effect_previous_dsl_state) = if let Some(input_id) =
                    accepted_input_id
                {
                    let (previous_dsl_state, effects) = self
                        .apply_session_dsl_input(
                            &session_id,
                            crate::meerkat_machine::dsl::MeerkatMachineInput::AcceptWithCompletion {
                                input_id: crate::meerkat_machine::dsl::InputId::from_domain(
                                    &input_id,
                                ),
                                request_immediate_processing: resolved
                                    .coarse_flags
                                    .request_immediate_processing,
                                interrupt_yielding: resolved.coarse_flags.interrupt_yielding,
                                wake_if_idle: resolved.coarse_flags.wake_if_idle,
                            },
                            "AcceptWithCompletion",
                        )
                        .await
                        .map_err(|reason| {
                            RuntimeDriverError::Internal(format!(
                                "canonical AcceptWithCompletion apply failed after admission: {reason}"
                            ))
                        })?;
                    {
                        let mut driver = driver.lock().await;
                        driver.absorb_post_admission_effects(&effects);
                    }
                    let signal = Self::post_admission_signal_from_effects(&effects);
                    let runtime_effect =
                        crate::effect::runtime_effect_projection_optional_from_dsl_effects(
                            &effects,
                        )
                        .map_err(|reason| {
                            RuntimeDriverError::Internal(format!(
                                "canonical AcceptWithCompletion emitted invalid runtime effect facts: {reason}"
                            ))
                        })?;
                    (signal, runtime_effect, Some(previous_dsl_state))
                } else {
                    (signal, None, None)
                };

                if signal.should_wake()
                    && let Some(ref wake_tx) = wake_tx
                {
                    let _ = wake_tx.try_send(());
                }
                if let Some(projected_effect) = runtime_effect
                    && let Err(err) = self
                        .dispatch_interrupt_yielding_runtime_effect(
                            &session_id,
                            effect_tx,
                            boundary_handle,
                            projected_effect,
                            "AcceptWithCompletion",
                        )
                        .await
                {
                    if let Some(previous_dsl_state) = effect_previous_dsl_state {
                        self.restore_session_dsl_state(&session_id, previous_dsl_state)
                            .await;
                    }
                    return Err(err);
                }

                Ok(MeerkatMachineCommandResult::AcceptWithCompletion {
                    outcome,
                    handle,
                    admission_signal: signal,
                })
            }
            MeerkatMachineCommand::AcceptWithoutWake { session_id, input } => {
                let driver = {
                    let sessions = self.sessions.read().await;
                    let entry = sessions
                        .get(&session_id)
                        .ok_or(RuntimeDriverError::NotReady {
                            state: RuntimeState::Destroyed,
                        })?;
                    entry.driver.clone()
                };

                let gate = self.session_mutation_gate(&session_id).await;
                let _gate_guard = match gate {
                    Some(ref g) => Some(g.lock().await),
                    None => None,
                };

                let state = self
                    .existing_session_runtime_state(&session_id)
                    .await
                    .unwrap_or(RuntimeState::Destroyed);
                let visible_state = self
                    .existing_session_visible_runtime_state(&session_id)
                    .await
                    .unwrap_or(RuntimeState::Destroyed);
                Self::reject_visible_terminal_ingress(visible_state)?;

                let (outcome, accepted_input_id) = {
                    let mut driver = driver.lock().await;
                    let runtime_idle = state.is_idle_or_attached();
                    let resolved = driver.resolve_admission_for_runtime_idle(&input, runtime_idle);
                    self.preview_session_dsl_input(
                        &session_id,
                        crate::meerkat_machine::dsl::MeerkatMachineInput::AcceptWithoutWake {
                            input_id: crate::meerkat_machine::dsl::InputId::from_domain(
                                &InputId::new(),
                            ),
                        },
                        "AcceptWithoutWake",
                    )
                    .await
                    .map_err(|reason| Self::classify_ingress_dsl_rejection(state, reason))?;
                    let mut resolved = resolved;
                    resolved.policy.wake_mode = crate::policy::WakeMode::None;
                    let result = match driver
                        .accept_resolved_input(input, resolved)
                        .await
                        .map_err(Self::normalize_destroyed_error)
                    {
                        Ok(r) => r,
                        Err(err) => return Err(err),
                    };
                    if let AcceptOutcome::Rejected { reason } = &result {
                        return Err(RuntimeDriverError::ValidationFailed {
                            reason: reason.to_string(),
                        });
                    }
                    let accepted_input_id = match &result {
                        AcceptOutcome::Accepted { input_id, .. } => Some(input_id.clone()),
                        AcceptOutcome::Deduplicated { .. } => None,
                        AcceptOutcome::Rejected { .. } => unreachable!("handled above"),
                    };
                    (result, accepted_input_id)
                };
                if let Some(input_id) = accepted_input_id {
                    let (_, effects) = self
                        .apply_session_dsl_input(
                            &session_id,
                            crate::meerkat_machine::dsl::MeerkatMachineInput::AcceptWithoutWake {
                                input_id: crate::meerkat_machine::dsl::InputId::from_domain(
                                    &input_id,
                                ),
                            },
                            "AcceptWithoutWake",
                        )
                        .await
                        .map_err(|reason| {
                            RuntimeDriverError::Internal(format!(
                                "canonical AcceptWithoutWake apply failed after admission: {reason}"
                            ))
                        })?;
                    {
                        let mut driver = driver.lock().await;
                        driver.absorb_post_admission_effects(&effects);
                    }
                    let signal = Self::post_admission_signal_from_effects(&effects);
                    debug_assert!(
                        !signal.should_wake()
                            && !signal.should_interrupt_yielding()
                            && !signal.should_process_immediately(),
                        "AcceptWithoutWake unexpectedly emitted a post-admission signal"
                    );
                }

                Ok(MeerkatMachineCommandResult::AcceptOutcome(outcome))
            }
            _ => unreachable!("non-ingress command routed to ingress handler"),
        }
    }

    pub(super) async fn execute_meerkat_machine_legacy_run_command(
        &self,
        command: MeerkatMachineCommand,
    ) -> Result<MeerkatMachineCommandResult, RuntimeDriverError> {
        match command {
            MeerkatMachineCommand::Prepare { session_id, input } => {
                let driver = {
                    let sessions = self.sessions.read().await;
                    sessions
                        .get(&session_id)
                        .ok_or(RuntimeDriverError::NotReady {
                            state: RuntimeState::Destroyed,
                        })?
                        .driver
                        .clone()
                };

                let gate = self.session_mutation_gate(&session_id).await;
                let _gate_guard = match gate {
                    Some(ref g) => Some(g.lock().await),
                    None => None,
                };

                let visible_state = self
                    .existing_session_visible_runtime_state(&session_id)
                    .await
                    .unwrap_or(RuntimeState::Destroyed);
                Self::reject_visible_terminal_ingress(visible_state)?;

                let prepare_precheck_error = {
                    let driver = driver.lock().await;
                    let state = driver.runtime_state();
                    if !driver.is_idle_or_attached() {
                        Some(Self::normalize_destroyed_error(
                            RuntimeDriverError::NotReady { state },
                        ))
                    } else if !driver.as_driver().active_input_ids().is_empty() {
                        let duplicate_active_input = input
                            .header()
                            .idempotency_key
                            .as_ref()
                            .and_then(|key| driver.input_id_for_idempotency_key(key));
                        if let Some(existing_id) = duplicate_active_input {
                            Some(RuntimeDriverError::ValidationFailed {
                                reason: format!(
                                    "accept_input_and_run does not support deduplicated admission; existing input {existing_id} already owns execution"
                                ),
                            })
                        } else {
                            Some(RuntimeDriverError::NotReady { state })
                        }
                    } else {
                        None
                    }
                };
                if let Some(err) = prepare_precheck_error {
                    return Err(err);
                }

                let run_id = RunId::new();
                let prepared = {
                    let mut driver = driver.lock().await;
                    let outcome = match driver
                        .as_driver_mut()
                        .accept_input(input)
                        .await
                        .map_err(Self::normalize_destroyed_error)
                    {
                        Ok(o) => o,
                        Err(err) => return Err(err),
                    };
                    let input_id = match outcome {
                        AcceptOutcome::Accepted { input_id, .. } => input_id,
                        AcceptOutcome::Deduplicated { existing_id, .. } => {
                            return Err(RuntimeDriverError::ValidationFailed {
                                reason: format!(
                                    "accept_input_and_run does not support deduplicated admission; existing input {existing_id} already owns execution"
                                ),
                            });
                        }
                        AcceptOutcome::Rejected { reason } => {
                            return Err(RuntimeDriverError::ValidationFailed {
                                reason: reason.to_string(),
                            });
                        }
                    };

                    let (dequeued_id, dequeued_input) = match driver.dequeue_next() {
                        Some(pair) => pair,
                        None => {
                            return Err(RuntimeDriverError::Internal(
                                "accepted input was not queued for execution".into(),
                            ));
                        }
                    };
                    if dequeued_id != input_id {
                        return Err(Self::normalize_destroyed_error(
                            RuntimeDriverError::NotReady {
                                state: self
                                    .existing_session_runtime_state(&session_id)
                                    .await
                                    .unwrap_or(RuntimeState::Destroyed),
                            },
                        ));
                    }

                    if let Err(err) = machine_begin_run(&mut driver, run_id.clone()) {
                        return Err(RuntimeDriverError::Internal(format!(
                            "failed to start runtime run: {err}"
                        )));
                    }
                    if let Err(err) = driver.stage_input(&dequeued_id, &run_id) {
                        let _ = driver.rollback_staged(std::slice::from_ref(&dequeued_id));
                        let next_phase = crate::runtime_state::run_return_phase_from_pre_run_phase(
                            driver.pre_run_phase(),
                        );
                        if let Err(rollback_err) = machine_apply_run_return_projection(
                            &mut driver,
                            &run_id,
                            crate::meerkat_machine::driver::RunReturnDisposition::Rollback,
                            next_phase,
                        ) {
                            return Err(RuntimeDriverError::Internal(format!(
                                "failed to roll back runtime run after staging failure: {rollback_err}; staging failure: {err}"
                            )));
                        }
                        return Err(RuntimeDriverError::Internal(format!(
                            "failed to stage accepted input: {err}"
                        )));
                    }

                    let primitive_result =
                        match crate::meerkat_machine::machine_batch_runtime_semantics(
                            &driver,
                            std::slice::from_ref(&dequeued_id),
                        ) {
                            Some(mut semantics) if semantics.len() == 1 => {
                                let projection_inputs =
                                    [(dequeued_id.clone(), dequeued_input.clone())];
                                let projections =
                                    crate::meerkat_machine::machine_batch_primitive_projections(
                                        &driver,
                                        &projection_inputs,
                                    );
                                crate::runtime_loop::admitted_input_to_primitive(
                                    &dequeued_input,
                                    dequeued_id.clone(),
                                    projections.into_iter().next().unwrap_or_default(),
                                    semantics.remove(0),
                                )
                            }
                            _ => Err(
                                meerkat_core::lifecycle::run_primitive::TurnMetadataMergeConflict {
                                    field: "execution_kind",
                                    reason: "runtime-stamped execution kind missing for one or more inputs",
                                },
                            ),
                        };

                    let primitive = match primitive_result {
                        Ok(primitive) => primitive,
                        Err(err) => {
                            let _ = driver.rollback_staged(std::slice::from_ref(&dequeued_id));
                            let next_phase =
                                crate::runtime_state::run_return_phase_from_pre_run_phase(
                                    driver.pre_run_phase(),
                                );
                            if let Err(rollback_err) = machine_apply_run_return_projection(
                                &mut driver,
                                &run_id,
                                crate::meerkat_machine::driver::RunReturnDisposition::Rollback,
                                next_phase,
                            ) {
                                return Err(RuntimeDriverError::Internal(format!(
                                    "failed to roll back runtime run after primitive build failure: {rollback_err}; primitive build failure: {err}"
                                )));
                            }
                            return Err(RuntimeDriverError::Internal(format!(
                                "failed to build accepted input primitive: {err}"
                            )));
                        }
                    };

                    MeerkatMachineRunPrepared {
                        input_id,
                        run_id,
                        primitive,
                    }
                };

                Ok(MeerkatMachineCommandResult::Prepared(prepared))
            }
            MeerkatMachineCommand::Commit {
                session_id,
                input_id,
                run_id,
                output,
            } => {
                let driver = {
                    let sessions = self.sessions.read().await;
                    sessions
                        .get(&session_id)
                        .ok_or(RuntimeDriverError::NotReady {
                            state: RuntimeState::Destroyed,
                        })?
                        .driver
                        .clone()
                };

                let gate = self.session_mutation_gate(&session_id).await;
                let _gate_guard = match gate {
                    Some(ref g) => Some(g.lock().await),
                    None => None,
                };

                if let Err(err) = commit_runtime_loop_run(
                    &driver,
                    run_id.clone(),
                    vec![input_id.clone()],
                    output.receipt,
                    output.session_snapshot,
                )
                .await
                {
                    let should_unregister = err.should_unregister_session();
                    let should_unwind_active_run = err.is_boundary_commit();
                    let err = err.into_driver_error();
                    let rollback_err = if should_unwind_active_run {
                        crate::meerkat_machine::rollback_runtime_loop_run_after_boundary_commit_failure(
                            &driver,
                            &run_id,
                            std::slice::from_ref(&input_id),
                        )
                        .await
                        .err()
                    } else {
                        None
                    };
                    if should_unregister {
                        self.unregister_session_inner(&session_id).await;
                    }
                    let message = match rollback_err {
                        Some(rollback_err) => format!(
                            "runtime commit failed: {err}; additionally failed to unwind active run: {rollback_err}"
                        ),
                        None => format!("runtime commit failed: {err}"),
                    };
                    return Err(RuntimeDriverError::Internal(message));
                }

                Ok(MeerkatMachineCommandResult::Unit)
            }
            MeerkatMachineCommand::Fail {
                session_id,
                run_id,
                failure,
            } => {
                let driver = {
                    let sessions = self.sessions.read().await;
                    sessions
                        .get(&session_id)
                        .ok_or(RuntimeDriverError::NotReady {
                            state: RuntimeState::Destroyed,
                        })?
                        .driver
                        .clone()
                };

                let gate = self.session_mutation_gate(&session_id).await;
                let _gate_guard = match gate {
                    Some(ref g) => Some(g.lock().await),
                    None => None,
                };

                if let Err(run_err) = fail_machine_run(&driver, run_id, failure).await {
                    let should_unregister = run_err.should_unregister_session();
                    let run_err = run_err.into_driver_error();
                    if should_unregister {
                        self.unregister_session_inner(&session_id).await;
                    }
                    return Err(RuntimeDriverError::Internal(format!(
                        "failed to persist runtime failure snapshot: {run_err}"
                    )));
                }

                Ok(MeerkatMachineCommandResult::Unit)
            }
            _ => unreachable!("non-legacy-run command routed to legacy-run handler"),
        }
    }
}