meerkat-runtime 0.7.21

#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
//! Phase 1 chokepoint tests for runtime ingress and control wiring.

use std::sync::Arc;

use chrono::Utc;
use meerkat_core::lifecycle::core_executor::{CoreApplyOutput, CoreExecutor, CoreExecutorError};
use meerkat_core::lifecycle::run_primitive::{
    ConversationAppend, ConversationAppendRole, CoreRenderable, RunApplyBoundary, RunPrimitive,
};
use meerkat_core::lifecycle::run_receipt::RunBoundaryReceiptDraft;
use meerkat_core::lifecycle::{InputId, RunId, RuntimeExecutionKind};
use meerkat_core::ops::{OpEvent, OperationId};
use meerkat_core::service::TurnToolOverlay;
use meerkat_core::types::{RunResult, SessionId, Usage};
use meerkat_runtime::completion::CompletionOutcome;
use meerkat_runtime::input::{
    ContinuationInput, Input, InputDurability, InputHeader, InputOrigin, InputVisibility,
    OperationInput, PromptInput,
};
use meerkat_runtime::{
    ApplyMode, DrainPolicy, InputAbandonReason, InputLifecycleState, InputTerminalOutcome,
    MeerkatMachine, PolicyDecision, QueueMode, RoutingDisposition, RuntimeState,
    SessionServiceRuntimeExt, WakeMode,
};

fn make_prompt(text: &str) -> Input {
    Input::Prompt(PromptInput {
        injected_context: Vec::new(),
        header: InputHeader {
            id: InputId::new(),
            timestamp: Utc::now(),
            source: InputOrigin::Operator,
            durability: InputDurability::Durable,
            visibility: InputVisibility::default(),
            idempotency_key: None,
            supersession_key: None,
            correlation_id: None,
        },
        content: text.into(),
        typed_turn_appends: Vec::new(),
        turn_metadata: None,
    })
}

fn make_run_result(text: &str) -> RunResult {
    RunResult {
        text: text.into(),
        session_id: SessionId::new(),
        usage: Usage::default(),
        turns: 1,
        tool_calls: 0,
        terminal_cause_kind: None,
        structured_output: None,
        extraction_error: None,
        schema_warnings: None,
        skill_diagnostics: None,
    }
}

struct ResultExecutor;

#[async_trait::async_trait]
impl CoreExecutor for ResultExecutor {
    async fn apply(
        &mut self,
        run_id: RunId,
        primitive: RunPrimitive,
    ) -> Result<CoreApplyOutput, CoreExecutorError> {
        Ok(CoreApplyOutput::with_run_result(
            RunBoundaryReceiptDraft {
                run_id,
                boundary: RunApplyBoundary::RunStart,
                contributing_input_ids: primitive.contributing_input_ids().to_vec(),
                conversation_digest: None,
                message_count: 0,
            },
            None,
            make_run_result("runtime ingress ok"),
        ))
    }

    async fn cancel_after_boundary(&mut self, _reason: String) -> Result<(), CoreExecutorError> {
        Ok(())
    }

    async fn stop_runtime_executor(&mut self, _reason: String) -> Result<(), CoreExecutorError> {
        Ok(())
    }
}

#[derive(Default)]
struct RecordingBatchExecutor {
    seen_contributors: Arc<tokio::sync::Mutex<Vec<Vec<InputId>>>>,
}

#[async_trait::async_trait]
impl CoreExecutor for RecordingBatchExecutor {
    async fn apply(
        &mut self,
        run_id: RunId,
        primitive: RunPrimitive,
    ) -> Result<CoreApplyOutput, CoreExecutorError> {
        self.seen_contributors
            .lock()
            .await
            .push(primitive.contributing_input_ids().to_vec());
        Ok(CoreApplyOutput::with_run_result(
            RunBoundaryReceiptDraft {
                run_id,
                boundary: RunApplyBoundary::RunStart,
                contributing_input_ids: primitive.contributing_input_ids().to_vec(),
                conversation_digest: None,
                message_count: 0,
            },
            None,
            make_run_result("batched runtime ingress ok"),
        ))
    }

    async fn cancel_after_boundary(&mut self, _reason: String) -> Result<(), CoreExecutorError> {
        Ok(())
    }

    async fn stop_runtime_executor(&mut self, _reason: String) -> Result<(), CoreExecutorError> {
        Ok(())
    }
}

#[tokio::test]
async fn runtime_ingress_control_red_ok_accepts_prompt_and_resolves_completion_handle() {
    let adapter = Arc::new(MeerkatMachine::ephemeral());
    let runtime: &dyn SessionServiceRuntimeExt = &*adapter;
    let sid = SessionId::new();
    adapter
        .register_session_with_executor(sid.clone(), Box::new(ResultExecutor))
        .await
        .expect("runtime executor registration should succeed");

    let input = make_prompt("phase 1 runtime ingress");
    let input_id = input.id().clone();
    let (outcome, handle) = runtime
        .accept_input_with_completion(&sid, input)
        .await
        .expect("accept input");
    assert!(outcome.is_accepted(), "prompt should be admitted");

    let completion = handle.expect("accepted input should expose a wait handle");
    let result = completion
        .wait()
        .await
        .expect("completion waiter should resolve");
    assert!(
        matches!(result, CompletionOutcome::Completed(ref run) if run.text == "runtime ingress ok"),
        "runtime-backed ingress should resolve through the completion handle"
    );

    let stored = runtime
        .input_state(&sid, &input_id)
        .await
        .expect("input state")
        .expect("input record");
    assert_eq!(stored.seed.phase, InputLifecycleState::Consumed);
    assert_eq!(
        runtime.runtime_state(&sid).await.expect("runtime state"),
        RuntimeState::Attached
    );
    assert!(
        runtime
            .list_active_inputs(&sid)
            .await
            .expect("active inputs")
            .is_empty(),
        "completed ingress should leave no active inputs behind"
    );
}

#[tokio::test]
async fn runtime_ingress_control_red_ok_reset_preempts_queued_input_once() {
    let adapter = Arc::new(MeerkatMachine::ephemeral());
    let runtime: &dyn SessionServiceRuntimeExt = &*adapter;
    let sid = SessionId::new();
    adapter
        .register_session(sid.clone())
        .await
        .expect("register session");

    let input = make_prompt("queued before reset");
    let input_id = input.id().clone();
    let (outcome, handle) = runtime
        .accept_input_with_completion(&sid, input)
        .await
        .expect("accept queued input");
    assert!(outcome.is_accepted());

    runtime.reset_runtime(&sid).await.expect("reset runtime");

    let result = handle
        .expect("queued input should expose a handle")
        .wait()
        .await
        .expect("completion waiter should resolve");
    assert!(
        matches!(result, CompletionOutcome::RuntimeTerminated { .. }),
        "queued ingress should resolve as terminated when control-plane reset wins"
    );

    let stored = runtime
        .input_state(&sid, &input_id)
        .await
        .expect("input state")
        .expect("input record");
    assert_eq!(stored.seed.phase, InputLifecycleState::Abandoned);
    assert!(matches!(
        stored.seed.terminal_outcome,
        Some(InputTerminalOutcome::Abandoned {
            reason: InputAbandonReason::Reset,
        })
    ));
}

#[tokio::test]
async fn runtime_ingress_control_closed_taxonomy_uses_explicit_continuation_and_operation_inputs() {
    let continuation = Input::Continuation(ContinuationInput::detached_background_op_completed());
    let continuation_policy = meerkat_runtime::DefaultPolicyTable::resolve(&continuation, true);
    assert_eq!(
        continuation.kind(),
        meerkat_runtime::InputKind::Continuation
    );
    assert_eq!(continuation_policy.apply_mode, ApplyMode::StageRunBoundary);
    assert_eq!(continuation_policy.wake_mode, WakeMode::WakeIfIdle);
    assert_eq!(continuation_policy.drain_policy, DrainPolicy::SteerBatch);
    assert_eq!(
        continuation_policy.routing_disposition,
        RoutingDisposition::Steer
    );

    let mut attention_continuation =
        Input::Continuation(ContinuationInput::detached_background_op_completed());
    if let Input::Continuation(continuation) = &mut attention_continuation {
        // The producer declares the typed continuation kind; admission routes
        // the lane and runtime semantics from it, not from overlay strings.
        continuation.continuation_kind = meerkat_runtime::ContinuationKind::WorkgraphAttention;
        continuation.turn_append = Some(ConversationAppend {
            role: ConversationAppendRole::SystemNotice,
            content: CoreRenderable::Text {
                text: "attention turn".to_string(),
            },
        });
        continuation.flow_tool_overlay = Some(TurnToolOverlay {
            dispatch_context: [(
                "workgraph.attention_projection".to_string(),
                serde_json::json!(true),
            )]
            .into_iter()
            .collect(),
            ..TurnToolOverlay::default()
        });
    }
    let attention_policy =
        meerkat_runtime::DefaultPolicyTable::resolve(&attention_continuation, true);
    assert_eq!(attention_policy.apply_mode, ApplyMode::StageRunStart);
    assert_eq!(attention_policy.drain_policy, DrainPolicy::QueueNextTurn);
    let attention_semantics =
        meerkat_runtime::ingress_types::RuntimeInputSemantics::try_from_generated_admission(
            &attention_continuation,
            true,
        )
        .expect("generated admission should project attention continuation semantics");
    assert_eq!(attention_semantics.boundary(), RunApplyBoundary::RunStart);
    assert_eq!(
        attention_semantics.execution_kind(),
        RuntimeExecutionKind::ContentTurn
    );

    let operation = Input::Operation(OperationInput {
        header: InputHeader {
            id: InputId::new(),
            timestamp: Utc::now(),
            source: InputOrigin::System,
            durability: InputDurability::Derived,
            visibility: InputVisibility {
                transcript_eligible: false,
                operator_eligible: false,
            },
            idempotency_key: None,
            supersession_key: None,
            correlation_id: None,
        },
        operation_id: OperationId::new(),
        event: OpEvent::Cancelled {
            id: OperationId::new(),
        },
    });
    let operation_policy = meerkat_runtime::DefaultPolicyTable::resolve(&operation, true);
    assert_eq!(operation.kind(), meerkat_runtime::InputKind::Operation);
    assert_eq!(operation_policy.apply_mode, ApplyMode::Ignore);
    assert_eq!(operation_policy.queue_mode, QueueMode::Priority);
    assert_eq!(
        operation_policy,
        PolicyDecision {
            apply_mode: ApplyMode::Ignore,
            wake_mode: WakeMode::None,
            queue_mode: QueueMode::Priority,
            consume_point: meerkat_runtime::ConsumePoint::OnAccept,
            drain_policy: DrainPolicy::Ignore,
            routing_disposition: RoutingDisposition::Drop,
            record_transcript: false,
            emit_operator_content: false,
            policy_version: meerkat_runtime::generated_default_policy_version(),
        }
    );
}

#[tokio::test]
async fn runtime_ingress_control_batches_same_boundary_contributors_in_runtime_order() {
    let adapter = Arc::new(MeerkatMachine::ephemeral());
    let runtime: &dyn SessionServiceRuntimeExt = &*adapter;
    let sid = SessionId::new();
    let seen = Arc::new(tokio::sync::Mutex::new(Vec::<Vec<InputId>>::new()));

    adapter
        .register_session_with_executor(
            sid.clone(),
            Box::new(RecordingBatchExecutor {
                seen_contributors: Arc::clone(&seen),
            }),
        )
        .await
        .expect("runtime executor registration should succeed");

    let first = make_prompt("batched one");
    let first_id = first.id().clone();
    let second = make_prompt("batched two");
    let second_id = second.id().clone();

    let (_, first_handle) = runtime
        .accept_input_with_completion(&sid, first)
        .await
        .expect("accept first input");
    let (_, second_handle) = runtime
        .accept_input_with_completion(&sid, second)
        .await
        .expect("accept second input");

    let first_result = first_handle
        .expect("first handle")
        .wait()
        .await
        .expect("first completion waiter should resolve");
    let second_result = second_handle
        .expect("second handle")
        .wait()
        .await
        .expect("second completion waiter should resolve");
    assert!(matches!(first_result, CompletionOutcome::Completed(_)));
    assert!(matches!(second_result, CompletionOutcome::Completed(_)));

    let batches = seen.lock().await;
    let flattened: Vec<InputId> = batches
        .iter()
        .flat_map(|batch| batch.iter().cloned())
        .collect();
    assert_eq!(
        flattened,
        vec![first_id, second_id],
        "runtime ingress should preserve contributor order even when attached execution materializes separate runs"
    );
}