plugins/agent/engine/tests/integration/model_interruption.rs

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

use crate::support::invocation;
mod recovery;
mod websocket;

use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;

use maka_agent_engine::{Engine, RunError, RunInput};
use maka_agent_facts::event::Fact;
use maka_agent_facts::event::Invocation;
use maka_agent_facts::event::InvocationOutcome;
use maka_agent_facts::event::ModelInterruption;
use maka_agent_facts::event::TerminalStatus;
use maka_agent_facts::model::ModelEvent;
use maka_agent_facts::model::error::ModelFailureCause;
use maka_code_runtime::CellLimits;
use maka_event_log::EventLog;
use maka_model::{ModelError, ModelExecutor, ProviderConfig, ProviderKind};
use maka_tools::CodeExecutor;
use maka_tools::ToolMode;
use serde_json::json;
use sha2::{Digest, Sha256};
use tokio::io::AsyncWriteExt;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Notify;
use tokio_util::sync::CancellationToken;

fn input(base: &str, suffix: &str) -> RunInput {
    RunInput {
        model_source: None,
        model_revision: None,
        provider_id: "fixture".into(),
        main_output_limit: None,
        context: None,
        invocation: Invocation {
            session_id: "session".into(),
            turn_id: format!("turn-{suffix}"),
            run_id: format!("run-{suffix}"),
            invocation_id: format!("invocation-{suffix}"),
        },
        request_fingerprint: None,
        provider: ProviderConfig {
            adapter: None,
            capabilities: Default::default(),
            kind: ProviderKind::OpenaiChatCompletions,
            model: "test".into(),
            base_url: base.into(),
            auth: maka_model::ProviderAuth::ApiKey("test-key".into()),
            headers: BTreeMap::new(),
            network: Default::default(),
            body_overlay: None,
        },
        provider_options: json!({}),
        supports_vision: false,
        configuration: invocation::configuration(ToolMode::Direct),
        work: maka_agent_engine::RunWork::Message {
            allow_prior_unknown: false,
            source_messages: Vec::new(),
            message: format!("question {suffix}").into(),
            tools: Default::default(),
        },
    }
}

use crate::support::http;
use http::read_request;
async fn respond(socket: &mut TcpStream, partial: bool) {
    let text = if partial {
        "unfinished-secret-fragment"
    } else {
        "done"
    };
    let chunk = json!({"id":"reply","object":"chat.completion.chunk","created":1,"model":"test","choices":[{"index":0,"delta":{"content":text},"finish_reason":null}]});
    socket
        .write_all(
            b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\n",
        )
        .await
        .unwrap();
    socket
        .write_all(format!("data: {chunk}\n\n").as_bytes())
        .await
        .unwrap();
    if !partial {
        let end = json!({"id":"reply","object":"chat.completion.chunk","created":1,"model":"test","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]});
        socket
            .write_all(format!("data: {end}\n\ndata: [DONE]\n\n").as_bytes())
            .await
            .unwrap();
    }
}

fn engine(log: Arc<EventLog>) -> Engine {
    Engine::new(
        log,
        ModelExecutor::new(Duration::from_secs(20)).unwrap(),
        CodeExecutor::new(CellLimits::default()).unwrap(),
    )
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn safe_retry_preserves_failed_evidence_and_cancellation_stops_backoff() {
    tokio::time::timeout(Duration::from_secs(15), async {
        for (cancel, reset) in [(false, false), (true, false), (false, true)] {
            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
            let base = format!("http://{}/v1", listener.local_addr().unwrap());
            let observed = Arc::new(Notify::new());
            let reset_ready = observed.clone();
            let server = tokio::spawn(async move {
                let (mut socket, _) = listener.accept().await.unwrap();
                let mut requests = vec![read_request(&mut socket).await];
                respond(&mut socket, true).await;
                let intent = json!({"id":"reply","object":"chat.completion.chunk","created":1,"model":"test",
                    "choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"unexecuted","type":"function",
                        "function":{"name":"Read","arguments":"{\"path\":\"never-read\"}"}}]},"finish_reason":null}]});
                socket.write_all(format!("data: {intent}\n\n").as_bytes()).await.unwrap();
                if reset {
                    reset_ready.notified().await;
                    socket.set_zero_linger().unwrap();
                }
                drop(socket); // Real EOF or RST, never a fabricated model error.
                if !cancel {
                    let (mut socket, _) = listener.accept().await.unwrap();
                    requests.push(read_request(&mut socket).await);
                    respond(&mut socket, false).await;
                }
                (requests, listener)
            });
            let directory = tempfile::tempdir().unwrap();
            let log = Arc::new(EventLog::open(&directory.path().join("events.sqlite")).await.unwrap());
            let worker = engine(log.clone());
            let cancellation = CancellationToken::new();
            let mut commits = log.subscribe_commits();
            let stop = async {
                if reset {
                    loop {
                        let prefix = log.prefix(100, 128 * 1024).await.unwrap();
                        if prefix.events.iter().any(|event| matches!(&event.event.fact,
                            Fact::ModelObserved { event: ModelEvent::PartDelta { text, .. }, .. }
                                if text == "unfinished-secret-fragment")) { break; }
                        commits.changed().await.unwrap();
                    }
                    observed.notify_one();
                }
                if cancel {
                    loop {
                        let prefix = log.prefix(100, 128 * 1024).await.unwrap();
                        if prefix.events.iter().any(|event| matches!(event.event.fact,
                            Fact::ModelInterrupted { status: ModelInterruption::RetryableFailure, .. })) { break; }
                        commits.changed().await.unwrap();
                    }
                    cancellation.cancel();
                }
            };
            let (result, ()) = tokio::join!(worker.run(input(&base, "retry"), cancellation.clone()), stop);
            if cancel { assert!(matches!(result, Err(RunError::Cancelled))); }
            else { result.unwrap(); }
            worker.drain().await;
            let (requests, listener) = server.await.unwrap();
            assert_eq!(requests.len(), if cancel { 1 } else { 2 });
            if !cancel { assert_eq!(requests[0], requests[1], "retry must use the identical frozen prompt, not failed text or tool intent"); }
            assert!(tokio::time::timeout(Duration::from_millis(20), listener.accept()).await.is_err());
            let prefix = log.prefix(100, 128 * 1024).await.unwrap();
            let requested: Vec<_> = prefix.events.iter().filter_map(|event| match &event.event.fact {
                Fact::ModelRequested { step_id, input_digest, .. } => Some((step_id, input_digest)), _ => None,
            }).collect();
            assert_eq!(requested.len(), requests.len());
            if !cancel { assert_ne!(requested[0].0, requested[1].0); assert_eq!(requested[0].1, requested[1].1); }
            assert_eq!(prefix.events.iter().filter(|event| matches!(event.event.fact,
                Fact::ModelInterrupted { status: ModelInterruption::RetryableFailure, .. })).count(), 1);
            assert!(prefix.events.iter().any(|event| matches!(&event.event.fact,
                Fact::ModelInterrupted { diagnostic: Some(message), .. } if !message.is_empty())),
                "a recovered request must retain its original failure cause");
            assert!(prefix.events.iter().any(|event| matches!(&event.event.fact,
                Fact::ModelObserved { event: ModelEvent::PartDelta { text, .. }, .. } if text == "unfinished-secret-fragment")));
            assert!(!prefix.events.iter().any(|event| matches!(event.event.fact,
                Fact::ToolDispatched { .. } | Fact::ToolRejected { .. })));
        }
    }).await.expect("retry and cancellation must settle within their bounds");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn continuation_claim_replays_only_its_lineage_and_retries_with_frozen_input() {
    use maka_agent_facts::event::EventWrite;
    use maka_agent_facts::event::LogScope;
    use maka_agent_facts::event::RuntimeEvent;
    use maka_agent_facts::input::InvocationInput;
    tokio::time::timeout(Duration::from_secs(20), async {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base = format!("http://{}/v1", listener.local_addr().unwrap());
        let directory = tempfile::tempdir().unwrap();
        let log = Arc::new(EventLog::open(&directory.path().join("events.sqlite")).await.unwrap());
        let mut source = input(&base, "source");
        source.configuration.workspace_identity = Some(
            maka_agent_facts::execution::WorkspaceIdentity::from_marker_id("c193cd58-f929-4ba2-bfb5-6887beaf8132").unwrap());
        let opening = |text: &str| Fact::InvocationOpened {
            configuration: Some(Box::new(source.configuration.clone())),
            input: InvocationInput::Message { content: text.into(), request_fingerprint: None,
                source_messages: Vec::new(), },
        };
        let route = format!("sha256:{}", hex::encode(Sha256::digest(serde_json::to_vec(&source.provider).unwrap())));
        for fact in [
            opening("question source"),
            Fact::ModelRequested { item_acceptance: false, step_id: "old".into(), model_id: "test".into(), purpose: maka_agent_facts::context::ModelPurpose::Main, context: None, context_breakdown: None,
                source_scope: LogScope::Session { id: "session".into() }, source_high_water: 1,
                source_digest: "fixture".into(), input_digest: "fixture".into(), route_identity: route,
                checkpoint_event_id: None, effective_source_digest: None },
            Fact::ModelCompleted { step_id: "old".into(), output: serde_json::from_value(json!({
                "parts":[{"kind":"text","text_kind":"text","text":"trimmed old tail","provider_options":null}],
                "finish_reason":"stop","usage":{}
            })).unwrap() },
            Fact::InvocationEnded { outcome: InvocationOutcome::Cancelled { source: "user".into() } },
        ] {
            log.append(&EventWrite::plain(RuntimeEvent::new(source.invocation.clone(), fact)).unwrap()).await.unwrap();
        }
        let prefix = log.run_source("session", &source.invocation.run_id, None).await.unwrap().unwrap();
        let boundary = prefix.boundary;
        for fact in [opening("unrelated later branch"), Fact::InvocationEnded { outcome: InvocationOutcome::Completed }] {
            log.append(&EventWrite::plain(RuntimeEvent::new(input(&base, "branch").invocation, fact)).unwrap()).await.unwrap();
        }
        let continuation = |id: &str| {
            let mut child = input(&base, id);
            child.configuration = source.configuration.clone();
            child.request_fingerprint = Some(maka_agent_facts::artifact::content_digest(id.as_bytes()));
            child.work = maka_agent_engine::RunWork::Continuation { source: boundary.clone(),tools: Default::default() };
            child
        };
        let server = tokio::spawn(async move {
            let mut requests = Vec::new();
            for partial in [true, false] {
                let (mut socket, _) = listener.accept().await.unwrap();
                requests.push(read_request(&mut socket).await);
                respond(&mut socket, partial).await;
            }
            (requests, listener)
        });
        let worker = engine(log.clone());
        let before = log.prefix(100, 128 * 1024).await.unwrap();
        worker.check_continuation(&continuation("preview"), &CancellationToken::new()).await.unwrap();
        assert_eq!(log.prefix(100, 128 * 1024).await.unwrap().digest, before.digest, "query cannot claim or mutate the source");
        worker.run(continuation("child"), CancellationToken::new()).await.unwrap();
        let (requests, listener) = server.await.unwrap();
        assert_eq!(requests[0], requests[1]);
        let prompt = serde_json::to_string(&requests[0]["messages"]).unwrap();
        assert!(prompt.contains("question source"));
        for excluded in ["trimmed old tail", "unrelated later branch", "question child", "unfinished-secret-fragment"] {
            assert!(!prompt.contains(excluded), "{excluded}: {prompt}");
        }
        let mut moved = continuation("moved");
        moved.configuration.workspace_identity = None;
        assert!(worker.run(moved, CancellationToken::new()).await.is_err());
        assert!(worker.run(continuation("duplicate"), CancellationToken::new()).await.is_err());
        worker.drain().await;
        assert!(tokio::time::timeout(Duration::from_millis(20), listener.accept()).await.is_err());
        let facts = log.prefix(100, 128 * 1024).await.unwrap();
        let claims: Vec<_> = facts.events.iter().filter_map(|s| match &s.event.fact {
            Fact::InvocationOpened { input: InvocationInput::Continuation { claim, .. }, .. } => Some(claim),
            _ => None,
        }).collect();
        assert_eq!(claims.len(), 1, "only one canonical claim may acquire the sealed source");
        let attempts: Vec<_> = facts.events.iter().filter_map(|s| match &s.event.fact {
            Fact::ModelRequested { step_id, source_high_water, source_digest, input_digest, .. } if s.event.invocation.run_id == "run-child" =>
                Some(((*source_high_water, source_digest), input_digest, step_id)),
            _ => None,
        }).collect();
        assert_eq!(attempts.len(), 2);
        assert_eq!(attempts[0].0, attempts[1].0, "retry preserves the original canonical cut");
        assert_eq!(attempts[0].1, &claims[0].replay.digest);
        assert_eq!(attempts[0].1, attempts[1].1);
        assert_ne!(attempts[0].2, attempts[1].2, "each physical attempt has its own identity");
        assert_eq!(facts.events.iter().filter(|s| matches!(s.event.fact,
            Fact::ModelInterrupted { status: ModelInterruption::RetryableFailure, .. })).count(), 1);
        for fact in [
            opening("later unknown effect"),
            Fact::ToolDispatched { presentation: None,
        title: None, operation_id: "unknown".into(),
                call: maka_agent_facts::tool_call::ToolCallIdentity::standalone("unknown-call".into()),
                name: "write".into(), input: json!({}) },
            Fact::InvocationEnded { outcome: InvocationOutcome::Failed { class: "outcome_unknown".into(), message: None, model_failure: None } },
        ] {
            log.append(&EventWrite::plain(RuntimeEvent::new(input(&base, "unknown").invocation, fact)).unwrap()).await.unwrap();
        }
        let unsafe_preview = worker.check_continuation(&continuation("unsafe-preview"), &CancellationToken::new()).await;
        assert!(matches!(&unsafe_preview,
            Err(RunError::Store(maka_event_log::StoreError::HistoryUnsettled))),
            "a safe historical cut cannot hide a later unresolved Session effect: {unsafe_preview:?}");
    }).await.expect("continuation must settle");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn model_timeout_cause_survives_reopening_the_log() {
    tokio::time::timeout(Duration::from_secs(10), async {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base = format!("http://{}/v1", listener.local_addr().unwrap());
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            read_request(&mut socket).await;
            std::future::pending::<()>().await;
        });
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("events.sqlite");
        {
            let log = Arc::new(EventLog::open(&path).await.unwrap());
            let engine = Engine::new(
                log.clone(),
                ModelExecutor::new(Duration::from_millis(250)).unwrap(),
                CodeExecutor::new(CellLimits::default()).unwrap(),
            );
            assert!(matches!(
                engine
                    .run(input(&base, "timeout"), CancellationToken::new())
                    .await,
                Err(RunError::Model(ModelError::TimedOut))
            ));
            drop(engine);
            Arc::try_unwrap(log).ok().unwrap().close().await.unwrap();
        }
        server.abort();
        let reopened = EventLog::open(&path).await.unwrap();
        let prefix = reopened.prefix(100, 128 * 1024).await.unwrap();
        assert!(matches!(
            &prefix.events.last().unwrap().event.fact,
            Fact::InvocationEnded {
                outcome: InvocationOutcome::Failed { class, message: Some(message), model_failure: Some(ModelFailureCause::TimedOut) }
            } if class == "model_timeout" && message == &ModelError::TimedOut.to_string()
        ));
        assert!(prefix.events.iter().any(|event| matches!(&event.event.fact,
            Fact::ModelInterrupted { model_failure: Some(ModelFailureCause::TimedOut), .. })));
        assert_eq!(
            prefix.project_invocation("invocation-timeout").terminal,
            Some(TerminalStatus::Failed)
        );
    })
    .await
    .expect("model timeout must durably settle");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn partial_text_survives_cancellation_without_becoming_replayed_model_history() {
    tokio::time::timeout(Duration::from_secs(30), async {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base = format!("http://{}/v1", listener.local_addr().unwrap());
        let release = Arc::new(Notify::new());
        let server_release = release.clone();
        let server = tokio::spawn(async move {
            let mut requests = Vec::new();
            for index in 0..3 {
                let (mut socket, _) = listener.accept().await.unwrap();
                requests.push(read_request(&mut socket).await);
                respond(&mut socket, index == 0).await;
                if index == 0 {
                    // Keep the first response incomplete until cancellation has
                    // settled; observing bytes sent is not a durability barrier.
                    server_release.notified().await;
                }
            }
            requests
        });
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("events.sqlite");
        {
            let log = Arc::new(EventLog::open(&path).await.unwrap());
            let engine = engine(log.clone());
            let cancellation = CancellationToken::new();
            let running_engine = engine.clone();
            let running_token = cancellation.clone();
            let first = input(&base, "cancelled");
            let running = tokio::spawn(async move {
                running_engine.run(first, running_token).await
            });
            let observed = loop {
                let prefix = log.prefix(100, 128 * 1024).await.unwrap();
                if prefix.events.iter().any(|stored| matches!(
                    &stored.event.fact,
                    Fact::ModelObserved { event: ModelEvent::PartDelta { text, .. }, .. }
                        if text == "unfinished-secret-fragment"
                )) {
                    break prefix;
                }
                assert!(!running.is_finished(), "model ended before durable partial text");
                tokio::task::yield_now().await;
            };
            let active = observed.project_invocation("invocation-cancelled");
            assert_eq!(active.terminal, None);
            assert_eq!(active.unfinished_model_steps.len(), 1);
            cancellation.cancel();
            assert!(matches!(running.await.unwrap(), Err(RunError::Model(ModelError::Cancelled))));
            let settled = log.prefix(100, 128 * 1024).await.unwrap();
            let tail = &settled.events[observed.events.len()..];
            let interrupted = tail.iter().position(|stored| matches!(
                &stored.event.fact,
                Fact::ModelInterrupted { step_id, status: ModelInterruption::Cancelled, .. }
                    if step_id == &active.unfinished_model_steps[0]
            )).unwrap();
            assert!(matches!(&tail.last().unwrap().event.fact,
                Fact::InvocationEnded { outcome: InvocationOutcome::Cancelled { source } }
                    if source == "runtime_cancellation"));
            assert!(interrupted < tail.len() - 1);
            assert!(!settled.events.iter().any(|stored| matches!(
                stored.event.fact, Fact::ModelCompleted { .. }
            )));
            let state = settled.project_invocation("invocation-cancelled");
            assert_eq!(state.terminal, Some(TerminalStatus::Cancelled));
            assert!(state.unfinished_model_steps.is_empty());
            release.notify_one();
            engine.run(input(&base, "next"), CancellationToken::new()).await.unwrap();
            drop(engine);
            Arc::try_unwrap(log).ok().unwrap().close().await.unwrap();
        }
        let log = Arc::new(EventLog::open(&path).await.unwrap());
        engine(log.clone()).run(input(&base, "reopened"), CancellationToken::new()).await.unwrap();
        let prefix = log.prefix(100, 128 * 1024).await.unwrap();
        assert!(prefix.events.iter().any(|stored| matches!(
            &stored.event.fact,
            Fact::InvocationEnded { outcome: InvocationOutcome::Cancelled { source } }
                if stored.event.invocation.invocation_id == "invocation-cancelled"
                    && source == "runtime_cancellation"
        )));
        assert!(prefix.events.iter().any(|stored| matches!(
            &stored.event.fact,
            Fact::ModelObserved { event: ModelEvent::PartDelta { text, .. }, .. }
                if text == "unfinished-secret-fragment"
        )));
        for suffix in ["cancelled", "next", "reopened"] {
            let state = prefix.project_invocation(&format!("invocation-{suffix}"));
            assert!(state.unfinished_model_steps.is_empty());
            assert_eq!(state.terminal, Some(if suffix == "cancelled" {
                TerminalStatus::Cancelled
            } else {
                TerminalStatus::Completed
            }));
        }
        let requests = server.await.unwrap();
        // This test owns conversation replay; system instructions have their own contract.
        let conversation = |request: &serde_json::Value| {
            request["messages"].as_array().unwrap().iter()
                .filter(|message| message["role"] != "system")
                .cloned().collect::<Vec<_>>()
        };
        assert_eq!(conversation(&requests[0]), vec![json!({"role":"user","content":"question cancelled"})]);
        // Couple the digest to the system bytes actually sent, without fixing their wording.
        let mut expected_prompt: Vec<_> = requests[0]["messages"].as_array().unwrap().iter()
            .filter(|message| message["role"] == "system").cloned().collect();
        expected_prompt.push(json!({"role":"user","content":[{"type":"text","text":"question cancelled"}]}));
        // Rebuild each exact pre-request prefix in an independent log. Its
        // digest includes original event identities and sequence, not history.
        let replica = EventLog::open(&directory.path().join("replica.sqlite")).await.unwrap();
        for stored in &prefix.events {
            if let Fact::ModelRequested { source_scope, source_high_water, source_digest, input_digest, .. } = &stored.event.fact {
                let source = replica.scoped_prefix(source_scope.clone(), 100, 128 * 1024).await.unwrap();
                assert_eq!(*source_scope, maka_agent_facts::event::LogScope::Session {
                    id: stored.event.invocation.session_id.clone(),
                });
                assert_eq!(*source_high_water, source.high_water);
                assert_eq!(*source_digest, source.digest);
                if stored.event.invocation.invocation_id == "invocation-cancelled" {
                    let expected = serde_json::to_vec(&json!({
                        "projection":"maka.model-history.v1",
                        "prompt": expected_prompt,
                        "tools":[], "providerOptions":{}, "maxOutputTokens":requests[0]["max_tokens"]
                    })).unwrap();
                    assert_eq!(*input_digest, format!("sha256:{}", hex::encode(Sha256::digest(expected))));
                }
            }
            assert_eq!(replica.append(&maka_agent_facts::event::EventWrite::plain(stored.event.clone()).unwrap()).await.unwrap(), stored.sequence);
        }
        assert_eq!(conversation(&requests[1]), vec![
            json!({"role":"user","content":"question cancelled"}),
            json!({"role":"user","content":"question next"})
        ]);
        assert_eq!(conversation(&requests[2]), vec![
            json!({"role":"user","content":"question cancelled"}),
            json!({"role":"user","content":"question next"}),
            json!({"role":"assistant","content":"done"}),
            json!({"role":"user","content":"question reopened"})
        ]);
    }).await.expect("partial model cancellation and recovery must make bounded progress");
}

#[tokio::test]
async fn captured_language_is_evidenced_without_becoming_a_provider_parameter() {
    use maka_agent_facts::language::Language;
    // Independent wire-evidence types preserve canonical field order at every level.
    #[derive(serde::Serialize)]
    struct Text<'a> {
        #[serde(rename = "type")]
        kind: &'static str,
        text: &'a str,
    }
    #[derive(serde::Serialize)]
    #[serde(tag = "role", rename_all = "lowercase")]
    enum Prompt<'a> {
        System { content: &'a str },
        User { content: [Text<'a>; 1] },
    }
    #[derive(serde::Serialize)]
    #[serde(rename_all = "camelCase")]
    struct Evidence<'a> {
        projection: &'static str,
        #[serde(skip_serializing_if = "Option::is_none")]
        language: Option<&'static str>,
        prompt: [Prompt<'a>; 2],
        tools: [(); 0],
        provider_options: serde_json::Value,
        max_output_tokens: u64,
    }
    for (language, tag, context_label) in [
        (Language::En, None, "Session context"),
        (Language::ZhCn, Some("zh-CN"), "会话上下文"),
        (Language::ZhTw, Some("zh-TW"), "工作階段上下文"),
    ] {
        let directory = tempfile::tempdir().unwrap();
        let log = Arc::new(
            EventLog::open(&directory.path().join("events.sqlite"))
                .await
                .unwrap(),
        );
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base = format!("http://{}/v1", listener.local_addr().unwrap());
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let request = read_request(&mut socket).await;
            respond(&mut socket, false).await;
            request
        });
        let mut run = input(&base, "language");
        run.configuration.language = language;
        let maka_agent_engine::RunWork::Message { message, .. } = &mut run.work else {
            unreachable!()
        };
        *message = "Keep 原文 /tmp/exact.rs unchanged".into();
        let worker = engine(log.clone());
        worker.run(run, CancellationToken::new()).await.unwrap();
        worker.drain().await;
        let request = server.await.unwrap();
        assert!(request.get("language").is_none());
        let system = request["messages"][0].clone();
        assert_eq!(system["role"], "system");
        assert!(system["content"].as_str().unwrap().contains(context_label));
        assert_eq!(
            request["messages"][1],
            json!({"role":"user","content":"Keep 原文 /tmp/exact.rs unchanged"})
        );
        let evidence = Evidence {
            projection: "maka.model-history.v1",
            language: tag,
            prompt: [
                Prompt::System {
                    content: system["content"].as_str().unwrap(),
                },
                Prompt::User {
                    content: [Text {
                        kind: "text",
                        text: "Keep 原文 /tmp/exact.rs unchanged",
                    }],
                },
            ],
            tools: [],
            provider_options: json!({}),
            max_output_tokens: request["max_tokens"].as_u64().unwrap(),
        };
        let expected_digest = format!(
            "sha256:{}",
            hex::encode(Sha256::digest(serde_json::to_vec(&evidence).unwrap()))
        );
        let prefix = log.prefix(100, 128 * 1024).await.unwrap();
        let opening = prefix
            .events
            .iter()
            .find_map(|event| match &event.event.fact {
                Fact::InvocationOpened {
                    configuration: Some(configuration),
                    ..
                } => Some(configuration),
                _ => None,
            })
            .unwrap();
        assert_eq!(opening.language, language);
        let recorded = serde_json::to_value(opening).unwrap();
        assert_eq!(
            recorded.get("language").and_then(serde_json::Value::as_str),
            tag
        );
        let digest = prefix
            .events
            .iter()
            .find_map(|event| match &event.event.fact {
                Fact::ModelRequested { input_digest, .. } => Some(input_digest),
                _ => None,
            })
            .unwrap();
        assert_eq!(digest, &expected_digest);
        log.shutdown().await.unwrap();
    }
}