meerkat-runtime 0.8.18

//! MobRuntimeAdapter — bridges mob provisioning to v9 RuntimeDriver lifecycle.
//!
//! When a mob member is spawned, the adapter registers a RuntimeDriver for that
//! session. When retired, the adapter retires/unregisters the driver. Flow steps
//! are delivered as FlowStepInput through accept_input().
//!
//! This adapter is optional — mob works without it (existing SessionService path).
//! When present, it enables v9 input lifecycle tracking for mob members.

use meerkat_core::lifecycle::InputId;
use meerkat_core::types::ContentInput;
use meerkat_core::types::SessionId;

use crate::MeerkatMachine;
use crate::input::{
    FlowStepInput, Input, InputDurability, InputHeader, InputOrigin, InputVisibility,
};
#[allow(unused_imports)]
use crate::service_ext::SessionServiceRuntimeExt as _;
use crate::traits::{RuntimeControlPlaneError, RuntimeDriverError};

/// Create a FlowStepInput for a mob flow step.
pub fn create_flow_step_input(
    step_id: &str,
    instructions: ContentInput,
    flow_id: &str,
    step_index: usize,
    turn_metadata: Option<meerkat_core::lifecycle::run_primitive::RuntimeTurnMetadata>,
) -> Input {
    Input::FlowStep(FlowStepInput {
        header: InputHeader {
            id: InputId::new(),
            timestamp: chrono::Utc::now(),
            source: InputOrigin::Flow {
                flow_id: flow_id.into(),
                step_index,
            },
            durability: InputDurability::Durable,
            visibility: InputVisibility::default(),
            idempotency_key: None,
            supersession_key: None,
            correlation_id: None,
        },
        step_id: step_id.into(),
        content: instructions,
        directed_interaction_id: None,
        turn_metadata,
    })
}

/// Create the member-side tracked-turn input for a directive-bearing
/// `DeliverMemberInput` (multi-host mobs §18 O1, DEC-P6F-9 step 3): the
/// local flow-step mechanics relocated to the host that runs the loop.
///
/// `idempotency_key` is the delivery `input_id`, preserved EXACTLY so
/// redelivery deduplicates (`AcceptOutcome::Deduplicated`), matching the
/// plain-delivery path's `peer_input_from_delivery_payload` discipline.
pub fn create_tracked_flow_step_input(
    step_id: &str,
    instructions: ContentInput,
    flow_id: &str,
    turn_metadata: Option<meerkat_core::lifecycle::run_primitive::RuntimeTurnMetadata>,
    stable_input_id: &str,
) -> Result<Input, String> {
    let stable_uuid = uuid::Uuid::parse_str(stable_input_id).map_err(|error| {
        format!("tracked flow-step payload input_id '{stable_input_id}' is not a UUID: {error}")
    })?;
    if stable_uuid.is_nil() {
        return Err("tracked flow-step payload input_id must not be the nil UUID".to_string());
    }
    if stable_uuid.to_string() != stable_input_id {
        return Err(format!(
            "tracked flow-step payload input_id '{stable_input_id}' is not a canonical UUID"
        ));
    }
    Ok(Input::FlowStep(FlowStepInput {
        header: InputHeader {
            id: InputId::from_uuid(stable_uuid),
            timestamp: chrono::Utc::now(),
            source: InputOrigin::Flow {
                flow_id: flow_id.into(),
                // Remote steps carry no member-side step ordinal; the
                // controlling host owns flow sequencing (§18.1 — frame
                // engine stays controlling-host-local).
                step_index: 0,
            },
            durability: InputDurability::Durable,
            visibility: InputVisibility::default(),
            idempotency_key: Some(crate::identifiers::IdempotencyKey::new(stable_input_id)),
            supersession_key: None,
            correlation_id: Some(crate::identifiers::CorrelationId::from_uuid(stable_uuid)),
        },
        step_id: step_id.into(),
        content: instructions,
        directed_interaction_id: Some(meerkat_core::interaction::InteractionId(stable_uuid)),
        turn_metadata,
    }))
}

#[cfg(test)]
mod tracked_flow_step_tests {
    use super::*;

    #[test]
    fn payload_uuid_owns_input_id_correlation_and_idempotency() {
        let stable = uuid::Uuid::new_v4();
        let input = create_tracked_flow_step_input(
            "step-1",
            ContentInput::Text("work".to_string()),
            "run-1",
            None,
            &stable.to_string(),
        )
        .expect("UUID lowers to tracked input");
        assert_eq!(input.id().0, stable);
        assert_eq!(
            input.header().correlation_id.as_ref().map(|id| id.0),
            Some(stable)
        );
        assert_eq!(
            input
                .header()
                .idempotency_key
                .as_ref()
                .map(ToString::to_string),
            Some(stable.to_string())
        );
        crate::input::validate_directed_flow_step_correlation(&input)
            .expect("constructor must mint a valid directed correlation");
    }

    #[test]
    fn forged_directed_correlation_is_rejected() {
        let stable = uuid::Uuid::new_v4();
        let mut input = create_tracked_flow_step_input(
            "step-1",
            ContentInput::Text("work".to_string()),
            "run-1",
            None,
            &stable.to_string(),
        )
        .expect("tracked input");
        let Input::FlowStep(flow_step) = &mut input else {
            panic!("flow step");
        };
        flow_step.directed_interaction_id = Some(meerkat_core::interaction::InteractionId(
            uuid::Uuid::new_v4(),
        ));

        assert!(crate::input::validate_directed_flow_step_correlation(&input).is_err());
    }

    #[test]
    fn caller_metadata_cannot_smuggle_directed_interaction_ids() {
        let stable = uuid::Uuid::new_v4();
        let forged = meerkat_core::interaction::InteractionId(uuid::Uuid::new_v4());
        let mut caller_metadata =
            meerkat_core::lifecycle::run_primitive::RuntimeTurnMetadata::default();
        caller_metadata.directed_interaction_ids.push(forged);
        let input = create_tracked_flow_step_input(
            "step-1",
            ContentInput::Text("work".to_string()),
            "run-1",
            Some(caller_metadata),
            &stable.to_string(),
        )
        .expect("tracked input");
        let semantics =
            crate::ingress_types::RuntimeInputSemantics::try_from_generated_admission(&input, true)
                .expect("flow step admission");
        let projected = crate::runtime_loop::for_input(&input, semantics);

        assert_eq!(
            projected.directed_interaction_ids,
            vec![meerkat_core::interaction::InteractionId(stable)],
            "runtime must overwrite caller-supplied directed identities"
        );
    }

    #[test]
    fn non_uuid_payload_id_is_rejected_before_admission() {
        assert!(
            create_tracked_flow_step_input(
                "step-1",
                ContentInput::Text("work".to_string()),
                "run-1",
                None,
                "alias-not-uuid",
            )
            .is_err()
        );
    }

    #[test]
    fn nil_uuid_payload_id_is_rejected_before_admission() {
        assert!(
            create_tracked_flow_step_input(
                "step-1",
                ContentInput::Text("work".to_string()),
                "run-1",
                None,
                &uuid::Uuid::nil().to_string(),
            )
            .expect_err("nil cannot own tracked-turn custody")
            .contains("nil UUID")
        );
    }
}

/// Register a mob member's session with the runtime adapter.
///
/// Registration is a control-plane prerequisite. A failed register is propagated
/// as a typed error rather than swallowed so the mob provisioning caller can abort
/// instead of proceeding as if the member's runtime exists.
pub async fn register_mob_member(
    adapter: &MeerkatMachine,
    session_id: SessionId,
) -> Result<(), RuntimeControlPlaneError> {
    adapter.register_session(session_id).await
}

/// Unregister a mob member's session from the runtime adapter.
pub async fn unregister_mob_member(
    adapter: &MeerkatMachine,
    session_id: &SessionId,
) -> Result<(), RuntimeDriverError> {
    adapter.unregister_session(session_id).await
}

/// Deliver a flow step to a mob member through the runtime path.
pub async fn deliver_flow_step(
    adapter: &MeerkatMachine,
    session_id: &SessionId,
    step_id: &str,
    instructions: impl Into<ContentInput>,
    flow_id: &str,
    step_index: usize,
) -> Result<crate::AcceptOutcome, RuntimeDriverError> {
    let input = create_flow_step_input(step_id, instructions.into(), flow_id, step_index, None);
    adapter.accept_input(session_id, input).await
}

/// Retire a mob member's runtime.
///
/// If the session is attached to a live `RuntimeLoop`, queued inputs remain
/// pending for drain. For plain registered sessions without a loop, retirement
/// abandons queued work because nothing can execute the drain path.
pub async fn retire_mob_member(
    adapter: &MeerkatMachine,
    session_id: &SessionId,
) -> Result<crate::traits::RetireReport, RuntimeDriverError> {
    adapter.retire_runtime(session_id).await
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;
    use crate::policy_table::DefaultPolicyTable;
    use std::sync::Arc;

    #[tokio::test]
    async fn spawn_creates_runtime_driver_session() {
        let adapter = Arc::new(MeerkatMachine::ephemeral());
        let sid = SessionId::new();

        register_mob_member(&adapter, sid.clone()).await.unwrap();

        // Session should have a runtime driver
        let state = adapter.runtime_state(&sid).await.unwrap();
        assert_eq!(state, crate::RuntimeState::Idle);
    }

    #[tokio::test]
    async fn flow_step_delivered_as_input() {
        let adapter = Arc::new(MeerkatMachine::ephemeral());
        let sid = SessionId::new();
        register_mob_member(&adapter, sid.clone()).await.unwrap();

        let outcome = deliver_flow_step(&adapter, &sid, "step-1", "analyze the data", "flow-1", 0)
            .await
            .unwrap();

        assert!(outcome.is_accepted());

        // Verify policy: flow_step → StageRunStart + WakeIfIdle
        let input = create_flow_step_input("s", "i".into(), "f", 0, None);
        let policy = DefaultPolicyTable::resolve(&input, true);
        assert_eq!(policy.apply_mode, crate::ApplyMode::StageRunStart);
        assert_eq!(policy.wake_mode, crate::WakeMode::WakeIfIdle);
    }

    #[tokio::test]
    async fn retire_without_runtime_loop_abandons_pending_inputs() {
        let adapter = Arc::new(MeerkatMachine::ephemeral());
        let sid = SessionId::new();
        register_mob_member(&adapter, sid.clone()).await.unwrap();

        // Accept an input first
        deliver_flow_step(&adapter, &sid, "s1", "do it", "f1", 0)
            .await
            .unwrap();

        // No RuntimeLoop is attached for plain registration, so retirement
        // abandons queued work instead of leaving it pending forever.
        let report = retire_mob_member(&adapter, &sid).await.unwrap();
        assert_eq!(report.inputs_abandoned, 1);
        assert_eq!(report.inputs_pending_drain, 0);
    }

    #[tokio::test]
    async fn create_flow_step_input_preserves_multimodal_blocks() -> Result<(), String> {
        let input = create_flow_step_input(
            "s",
            ContentInput::Blocks(vec![
                meerkat_core::types::ContentBlock::Text {
                    text: "inspect image".into(),
                },
                meerkat_core::types::ContentBlock::Image {
                    media_type: "image/png".into(),
                    data: "abc123".into(),
                },
            ]),
            "f",
            0,
            None,
        );

        let flow_step = match input {
            Input::FlowStep(flow_step) => flow_step,
            other => return Err(format!("expected flow step input, got {other:?}")),
        };
        assert_eq!(
            flow_step.content.text_content(),
            "inspect image\n[image: image/png]"
        );
        assert!(matches!(
            &flow_step.content,
            meerkat_core::types::ContentInput::Blocks(blocks) if blocks.len() == 2
        ));
        Ok(())
    }

    #[tokio::test]
    async fn unregister_removes_driver() {
        let adapter = Arc::new(MeerkatMachine::ephemeral());
        let sid = SessionId::new();
        register_mob_member(&adapter, sid.clone()).await.unwrap();

        unregister_mob_member(&adapter, &sid).await.unwrap();

        // Should fail now
        let result = adapter.runtime_state(&sid).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn register_exposes_driver_state() {
        // Mob-member registration is owned by the runtime control plane:
        // the registered session must have a live driver entry that surfaces
        // through typed runtime-state queries.
        let adapter = Arc::new(MeerkatMachine::ephemeral());
        let sid = SessionId::new();
        register_mob_member(&adapter, sid.clone()).await.unwrap();

        let active = adapter.list_active_inputs(&sid).await.unwrap();
        assert!(active.is_empty()); // No inputs yet
    }

    /// Gate (#99/#277): a control-plane registration that fails (the session was
    /// destroyed, so the RegisterSession command returns `Destroyed`) must surface
    /// a typed `Err` from `register_session` rather than being laundered to success.
    /// Pre-fix the helper dropped the result with `let _ = ...` and returned `()`,
    /// so the failure was invisible to callers.
    #[tokio::test]
    async fn register_session_surfaces_failure_on_destroyed_session() {
        let adapter = Arc::new(MeerkatMachine::ephemeral());
        let sid = SessionId::new();

        // Establish the session, then destroy it so re-registration must fail.
        adapter.register_session(sid.clone()).await.unwrap();
        let runtime_id = MeerkatMachine::logical_runtime_id(&sid);
        crate::traits::RuntimeControlPlane::destroy(&*adapter, &runtime_id)
            .await
            .unwrap();

        let result = adapter.register_session(sid.clone()).await;
        assert!(
            matches!(result, Err(RuntimeControlPlaneError::Internal(_))),
            "register_session on a destroyed session must surface a typed control-plane error, got {result:?}"
        );
    }

    /// Gate (#99): `register_mob_member` must propagate the typed registration
    /// failure rather than swallow it — a mob member whose runtime cannot be
    /// registered must not be reported as provisioned.
    #[tokio::test]
    async fn register_mob_member_propagates_failure_on_destroyed_session() {
        let adapter = Arc::new(MeerkatMachine::ephemeral());
        let sid = SessionId::new();

        register_mob_member(&adapter, sid.clone()).await.unwrap();
        let runtime_id = MeerkatMachine::logical_runtime_id(&sid);
        crate::traits::RuntimeControlPlane::destroy(&*adapter, &runtime_id)
            .await
            .unwrap();

        let result = register_mob_member(&adapter, sid.clone()).await;
        assert!(
            result.is_err(),
            "register_mob_member must propagate the typed registration failure, got {result:?}"
        );
    }

    /// Gate (#41): `set_session_silent_intents` must surface the inner command's
    /// typed `Err` to the caller instead of dropping it with `let _ = ...`. A
    /// destroyed session makes the SetSilentIntents command return `Destroyed`.
    #[tokio::test]
    async fn set_session_silent_intents_surfaces_failure_on_destroyed_session() {
        let adapter = Arc::new(MeerkatMachine::ephemeral());
        let sid = SessionId::new();

        adapter.register_session(sid.clone()).await.unwrap();
        let runtime_id = MeerkatMachine::logical_runtime_id(&sid);
        crate::traits::RuntimeControlPlane::destroy(&*adapter, &runtime_id)
            .await
            .unwrap();

        let result = adapter
            .set_session_silent_intents(&sid, vec!["status".to_string()])
            .await;
        assert!(
            matches!(result, Err(RuntimeDriverError::Destroyed)),
            "set_session_silent_intents must surface the inner command failure, got {result:?}"
        );
    }
}