synaps-engine 0.10.1

//! §5.2 — `SessionActor` unit tests through `EngineHost::create_session` +
//! `LocalTransport`, against a loopback Anthropic SSE stub.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use agent_engine::session::{
    ClientKind, ClientMeta, ClientTransport, EndReason, Envelope, LocalTransport, SessionCommand,
    SessionConfig, SessionEventWire, SessionSetting,
};
use agent_engine::{EngineHost, HostOpts, SessionEvent, StreamEvent};
use axum::response::IntoResponse;
use serial_test::serial;

const MODEL: &str = "claude-sonnet-4-5";

const SSE_HI: &str = concat!(
    "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_01\",\"type\":\"message\",",
    "\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4-5\",\"stop_reason\":null,",
    "\"stop_sequence\":null,\"usage\":{\"input_tokens\":10,\"output_tokens\":0,",
    "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0}}}\n\n",
    "data: {\"type\":\"content_block_start\",\"index\":0,",
    "\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
    "data: {\"type\":\"content_block_delta\",\"index\":0,",
    "\"delta\":{\"type\":\"text_delta\",\"text\":\"hi\"}}\n\n",
    "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
    "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",",
    "\"stop_sequence\":null},\"usage\":{\"input_tokens\":10,\"output_tokens\":1,",
    "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0}}\n\n",
    "data: {\"type\":\"message_stop\"}\n\n",
);

/// Started stream with one text delta and NO terminal event.
const SSE_PREFIX: &str = concat!(
    "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_01\",\"type\":\"message\",",
    "\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4-5\",\"stop_reason\":null,",
    "\"stop_sequence\":null,\"usage\":{\"input_tokens\":10,\"output_tokens\":0,",
    "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0}}}\n\n",
    "data: {\"type\":\"content_block_start\",\"index\":0,",
    "\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
    "data: {\"type\":\"content_block_delta\",\"index\":0,",
    "\"delta\":{\"type\":\"text_delta\",\"text\":\"hi\"}}\n\n",
);

struct Home {
    _dir: tempfile::TempDir,
}

impl Home {
    fn new() -> Self {
        let dir = tempfile::TempDir::new().unwrap();
        let base = dir.path().join(".synaps-cli");
        std::fs::create_dir_all(&base).unwrap();
        std::fs::write(base.join("config"), "").unwrap();
        std::fs::write(
            base.join("auth.json"),
            "{\"anthropic\": {\"type\": \"oauth\", \"refresh\": \"r\", \"access\": \"synthetic-token\", \"expires\": 9999999999999}}",
        )
        .unwrap();
        for k in ["ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GOOGLE_API_KEY"] {
            std::env::remove_var(k);
        }
        std::env::set_var("HOME", dir.path());
        std::env::set_var("SYNAPS_BASE_DIR", &base);
        Self { _dir: dir }
    }
}

/// `endless`: after the prefix, keep-alive comments forever (cancel fixture).
async fn stub(body: &'static str, endless: bool) -> (String, Arc<AtomicUsize>) {
    let hits = Arc::new(AtomicUsize::new(0));
    let hits_c = Arc::clone(&hits);
    let app = axum::Router::new().fallback(move || {
        let hits = Arc::clone(&hits_c);
        async move {
            hits.fetch_add(1, Ordering::SeqCst);
            if endless {
                let stream = futures::stream::once(async move {
                    Ok::<_, std::io::Error>(axum::body::Bytes::from(body))
                })
                .chain(futures::stream::unfold((), |()| async {
                    tokio::time::sleep(Duration::from_millis(200)).await;
                    Some((Ok(axum::body::Bytes::from(": keep-alive\n\n")), ()))
                }));
                axum::response::Response::builder()
                    .status(200)
                    .header("content-type", "text/event-stream")
                    .body(axum::body::Body::from_stream(stream))
                    .unwrap()
            } else {
                (
                    axum::http::StatusCode::OK,
                    [("content-type", "text/event-stream")],
                    body.to_string(),
                )
                    .into_response()
            }
        }
    });
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
    (format!("http://{addr}"), hits)
}

use futures::StreamExt;

/// Sequenced stub: hit `i` serves `bodies[min(i, len-1)]` (all terminal).
async fn stub_seq(bodies: &'static [&'static str]) -> (String, Arc<AtomicUsize>) {
    let hits = Arc::new(AtomicUsize::new(0));
    let hits_c = Arc::clone(&hits);
    let app = axum::Router::new().fallback(move || {
        let hits = Arc::clone(&hits_c);
        async move {
            let i = hits.fetch_add(1, Ordering::SeqCst);
            let body = bodies[i.min(bodies.len() - 1)];
            (
                axum::http::StatusCode::OK,
                [("content-type", "text/event-stream")],
                body.to_string(),
            )
                .into_response()
        }
    });
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
    (format!("http://{addr}"), hits)
}

async fn host() -> Arc<EngineHost> {
    EngineHost::boot(HostOpts {
        profile: None,
        no_extensions: true,
    })
    .await
    .unwrap()
}

fn cfg() -> SessionConfig {
    SessionConfig {
        model_override: Some(MODEL.into()),
        persist: false,
        ..SessionConfig::default()
    }
}

async fn next(t: &mut LocalTransport) -> Envelope {
    tokio::time::timeout(Duration::from_secs(30), t.next_event())
        .await
        .expect("timely")
        .expect("alive")
}

async fn until(t: &mut LocalTransport, pred: impl Fn(&SessionEventWire) -> bool) -> Vec<Envelope> {
    let mut out = Vec::new();
    loop {
        let env = next(t).await;
        let done = pred(&env.event);
        out.push(env);
        if done {
            return out;
        }
    }
}

fn submit(text: &str) -> SessionCommand {
    SessionCommand::Submit {
        text: text.into(),
        attachments: vec![],
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn attach_detach_counts_and_seq_is_gapless() {
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    assert_eq!(host.sessions().len(), 1);
    assert!(host.attach(&handle.id).is_some());

    let (mut a, snap_a) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    assert_eq!(snap_a.clients.len(), 1);
    assert_eq!(snap_a.view.model, MODEL);
    let (mut b, snap_b) =
        LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Attach))
            .await
            .unwrap();
    assert_eq!(snap_b.clients.len(), 2);
    assert_ne!(a.client_id(), b.client_id());

    a.send(submit("hello")).await.unwrap();
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Idle)).await;
    let seqs: Vec<u64> = seen.iter().map(|e| e.seq).collect();
    assert!(seqs.windows(2).all(|w| w[1] == w[0] + 1), "gapless: {seqs:?}");
    assert!(seen
        .iter()
        .any(|e| matches!(e.event, SessionEventWire::TurnStarted { .. })));
    assert!(seen.iter().any(|e| matches!(
        e.event,
        SessionEventWire::Stream(StreamEvent::Session(SessionEvent::Done))
    )));
    let conv = seen
        .iter()
        .rev()
        .find_map(|e| match &e.event {
            SessionEventWire::Conversation(c) => Some(c.clone()),
            _ => None,
        })
        .unwrap();
    assert_eq!(conv.api_messages.len(), 2);
    assert_eq!(conv.tokens.input, 10);

    // Both clients saw the same stream.
    let seen_b = until(&mut b, |e| matches!(e, SessionEventWire::Idle)).await;
    let sa: Vec<String> = seen
        .iter()
        .filter(|e| matches!(e.event, SessionEventWire::Stream(_)))
        .map(|e| format!("{:?}", e.event))
        .collect();
    let sb: Vec<String> = seen_b
        .iter()
        .filter(|e| matches!(e.event, SessionEventWire::Stream(_)))
        .map(|e| format!("{:?}", e.event))
        .collect();
    assert_eq!(sa, sb);

    b.send(SessionCommand::Detach {
        client: b.client_id(),
    })
    .await
    .unwrap();
    let left = until(&mut a, |e| matches!(e, SessionEventWire::ClientLeft { .. })).await;
    assert!(!left.is_empty());
    assert!(handle.is_alive());

    a.send(SessionCommand::End {
        reason: EndReason::ClientQuit,
    })
    .await
    .unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::Ended { .. })).await;
    handle.closed().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    assert_eq!(host.sessions().len(), 0, "actor removed itself from the host map");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn detach_does_not_cancel_stream_end_does() {
    let _h = Home::new();
    let (url, _) = stub(SSE_PREFIX, true).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    a.send(submit("go")).await.unwrap();
    until(&mut a, |e| {
        matches!(e, SessionEventWire::Stream(StreamEvent::Llm(agent_engine::LlmEvent::Text(_))))
    })
    .await;

    // Detach every client mid-turn: the stream keeps running.
    a.send(SessionCommand::Detach {
        client: a.client_id(),
    })
    .await
    .unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::ClientLeft { .. })).await;
    tokio::time::sleep(Duration::from_millis(300)).await;
    assert!(handle.is_alive());

    // A fresh client attaches mid-turn: streaming=true, replay non-empty.
    let (mut b, snap) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Attach))
        .await
        .unwrap();
    assert!(snap.streaming, "turn still running after every client detached");
    assert!(snap
        .replay
        .iter()
        .any(|e| matches!(e.event, SessionEventWire::Stream(_))));

    // End cancels the turn first, then finishes.
    b.send(SessionCommand::End {
        reason: EndReason::HostShutdown,
    })
    .await
    .unwrap();
    let tail = until(&mut b, |e| matches!(e, SessionEventWire::Ended { .. })).await;
    assert!(tail
        .iter()
        .any(|e| matches!(e.event, SessionEventWire::Conversation(_))));
    handle.closed().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn set_applies_and_publishes_view() {
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    a.send(SessionCommand::Set {
        id: 42,
        setting: SessionSetting::ContextWindow { tokens: Some(4242) },
    })
    .await
    .unwrap();
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::SettingChanged(_))).await;
    match &seen.last().unwrap().event {
        SessionEventWire::SettingChanged(s) => {
            assert!(s.ok);
            assert_eq!(s.id, 42);
            assert!(s.clamp.is_none());
            assert_eq!(s.setting, "context_window");
            assert_eq!(s.view.context_window, 4242);
        }
        _ => unreachable!(),
    }
    assert_eq!(a.view().context_window, 4242);
    a.send(SessionCommand::End {
        reason: EndReason::ClientQuit,
    })
    .await
    .unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::Ended { .. })).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn create_session_sets_cwd_and_session_id() {
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let tmp = tempfile::TempDir::new().unwrap();
    let handle = host
        .create_session(SessionConfig {
            cwd: Some(tmp.path().to_path_buf()),
            env: None,
            env_stripped: Vec::new(),
            ..cfg()
        })
        .await
        .unwrap();
    assert_eq!(handle.meta().cwd.as_deref(), Some(tmp.path()));
    assert_eq!(handle.meta().host_pid, std::process::id());
    // Registered under the real session registry (synaps send path).
    let reg = agent_engine::events::registry::find_session_registration(handle.id.as_str());
    assert!(reg.is_some(), "session registered");
    let (mut a, snap) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    assert_eq!(snap.meta.id, handle.id);
    a.send(SessionCommand::End {
        reason: EndReason::ClientQuit,
    })
    .await
    .unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::Ended { .. })).await;
    handle.closed().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    assert!(
        agent_engine::events::registry::find_session_registration(handle.id.as_str()).is_none(),
        "unregistered at End"
    );
}

fn last_conversation(seen: &[Envelope]) -> agent_engine::session::ConversationSnapshot {
    seen.iter()
        .rev()
        .find_map(|e| match &e.event {
            SessionEventWire::Conversation(c) => Some(c.clone()),
            _ => None,
        })
        .expect("a Conversation envelope")
}

async fn end(t: &mut LocalTransport) {
    t.send(SessionCommand::End {
        reason: EndReason::ClientQuit,
    })
    .await
    .unwrap();
    until(t, |e| matches!(e, SessionEventWire::Ended { .. })).await;
}

/// §11 #1: `Cancel` while idle is a no-op — at most an `Idle` echo. No
/// "aborted" notice, no Conversation (= no interruption marker, no save),
/// and the next Submit is appended verbatim.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn cancel_while_idle_is_a_noop() {
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();

    until(&mut a, |e| matches!(e, SessionEventWire::ClientJoined { .. })).await;

    // Never streamed: Cancel on a fresh session.
    a.send(SessionCommand::Cancel).await.unwrap();
    let env = next(&mut a).await;
    assert!(matches!(env.event, SessionEventWire::Idle), "got {:?}", env.event);

    a.send(submit("hello")).await.unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::Idle)).await;

    a.send(SessionCommand::Cancel).await.unwrap();
    let env = next(&mut a).await;
    assert!(matches!(env.event, SessionEventWire::Idle), "got {:?}", env.event);

    a.send(submit("again")).await.unwrap();
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Idle)).await;
    assert!(
        !seen
            .iter()
            .any(|e| matches!(&e.event, SessionEventWire::Aborted { .. })),
        "no Aborted event"
    );
    let conv = last_conversation(&seen);
    assert_eq!(conv.api_messages.len(), 4);
    assert_eq!(conv.api_messages[2]["content"], "again");
    assert!(conv.abort_context.is_none());
    assert!(
        !conv.api_messages.iter().any(|m| m["content"]
            .as_str()
            .is_some_and(agent_engine::engine::interrupt::is_interruption_marker)),
        "an idle Cancel records no interruption"
    );
    end(&mut a).await;
}

/// §11 #1: a `Cancel` that lands after `Done` sees `streaming=false`: it
/// must not mark the finished turn as interrupted.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn cancel_after_done_does_not_scrape_previous_turn() {
    let _h = Home::new();
    let (url, hits) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();

    a.send(submit("first")).await.unwrap();
    // Fire Cancel the moment Done is forwarded; it queues behind the
    // actor's Done handling and must see streaming=false.
    until(&mut a, |e| {
        matches!(
            e,
            SessionEventWire::Stream(StreamEvent::Session(SessionEvent::Done))
        )
    })
    .await;
    a.send(SessionCommand::Cancel).await.unwrap();
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Idle)).await;
    assert!(last_conversation(&seen).abort_context.is_none());
    // Drain the Cancel's own Idle echo if it was not already consumed.
    let mut idles = seen.iter().filter(|e| matches!(e.event, SessionEventWire::Idle)).count();
    while idles < 2 {
        let env = next(&mut a).await;
        assert!(
            matches!(env.event, SessionEventWire::Idle),
            "only Idle may follow an idle Cancel, got {:?}",
            env.event
        );
        idles += 1;
    }

    a.send(submit("second")).await.unwrap();
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Idle)).await;
    let conv = last_conversation(&seen);
    assert_eq!(hits.load(Ordering::SeqCst), 2);
    assert_eq!(conv.api_messages[2]["content"], "second");
    assert_eq!(conv.api_messages.len(), 4, "no marker between the turns");
    assert!(!conv.api_messages[2]["content"]
        .as_str()
        .unwrap()
        .contains("ABORT CONTEXT"));
    end(&mut a).await;
}

/// §11 #2: `--continue X` while X is live on this host attaches to the
/// running actor instead of building a second one on the same session
/// file / per-session UDS / registry entry.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn duplicate_continue_attaches_to_live_session() {
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;

    // A persisted session on disk, then ended.
    let persisted = SessionConfig {
        persist: true,
        ..cfg()
    };
    let h0 = host.create_session(persisted.clone()).await.unwrap();
    let (mut a, _) = LocalTransport::attach(h0.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    a.send(submit("hello")).await.unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::Idle)).await;
    end(&mut a).await;
    h0.closed().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    assert_eq!(host.sessions().len(), 0);
    let id = h0.id.clone();

    // --continue X, twice, plus --continue latest: one actor.
    let cont = SessionConfig {
        continue_session: Some(Some(id.as_str().to_string())),
        ..persisted.clone()
    };
    let h1 = host.create_session(cont.clone()).await.unwrap();
    assert_eq!(h1.id, id);
    let reg1 = agent_engine::events::registry::find_session_registration(id.as_str())
        .expect("registered");
    let h2 = host.create_session(cont.clone()).await.unwrap();
    let h3 = host
        .create_session(SessionConfig {
            continue_session: Some(None),
            ..persisted.clone()
        })
        .await
        .unwrap();
    assert_eq!(h2.id, id);
    assert_eq!(h3.id, id);
    assert_eq!(host.sessions().len(), 1, "one actor");
    let reg2 = agent_engine::events::registry::find_session_registration(id.as_str())
        .expect("still registered");
    assert_eq!(reg1.socket_path, reg2.socket_path);
    assert!(
        tokio::net::UnixStream::connect(&reg1.socket_path).await.is_ok(),
        "first actor's UDS still bound"
    );

    // Same actor: a client on h2 sees the turn started via h1.
    let (mut b, snap) = LocalTransport::attach(h2.clone(), ClientMeta::new(ClientKind::Attach))
        .await
        .unwrap();
    assert_eq!(snap.conversation.api_messages.len(), 2, "continued history");
    let (mut c, _) = LocalTransport::attach(h1.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    c.send(submit("again")).await.unwrap();
    let seen = until(&mut b, |e| matches!(e, SessionEventWire::Idle)).await;
    assert_eq!(last_conversation(&seen).api_messages.len(), 4);

    end(&mut c).await;
    h1.closed().await;
    assert!(!h2.is_alive());
    tokio::time::sleep(Duration::from_millis(50)).await;
    assert_eq!(host.sessions().len(), 0);
}

// ── prompts + lifecycle (§11 #4) ──────────────────────────────────────────

/// Turn requesting the prompting fixture tool; the stub then serves SSE_HI
/// for the follow-up round.
const SSE_PROMPT_TOOL_USE: &str = concat!(
    "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_p1\",\"type\":\"message\",",
    "\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4-5\",\"stop_reason\":null,",
    "\"stop_sequence\":null,\"usage\":{\"input_tokens\":10,\"output_tokens\":0,",
    "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0}}}\n\n",
    "data: {\"type\":\"content_block_start\",\"index\":0,",
    "\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_prompt\",\"name\":\"prompt_fixture\"}}\n\n",
    "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
    "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\",",
    "\"stop_sequence\":null},\"usage\":{\"input_tokens\":10,\"output_tokens\":5,",
    "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0}}\n\n",
    "data: {\"type\":\"message_stop\"}\n\n",
);

/// Builtin-origin tool that asks for a secret through the stream's
/// `SecretPromptHandle` and reports only its length (never the value).
struct PromptFixtureTool;

#[async_trait::async_trait]
impl agent_engine::Tool for PromptFixtureTool {
    fn name(&self) -> &str {
        "prompt_fixture"
    }
    fn description(&self) -> &str {
        "prompts for a secret"
    }
    fn parameters(&self) -> agent_engine::Value {
        serde_json::json!({"type": "object"})
    }
    fn origin(&self) -> agent_engine::tools::ToolOrigin {
        agent_engine::tools::ToolOrigin::Builtin
    }
    async fn execute(
        &self,
        _params: agent_engine::Value,
        ctx: agent_engine::ToolContext,
    ) -> agent_engine::Result<String> {
        let handle = ctx
            .capabilities
            .secret_prompt
            .expect("actor passes its SecretPromptHandle to the stream");
        Ok(match handle.prompt("Secret".into(), "enter secret".into()).await {
            Some(v) => format!("answered:{}", v.len()),
            None => "cancelled".to_string(),
        })
    }
}

async fn prompt_host() -> Arc<EngineHost> {
    let host = host().await;
    host.parts()
        .tools
        .write()
        .await
        .register(Arc::new(PromptFixtureTool));
    host
}

fn tool_results(snap: &agent_engine::session::ConversationSnapshot) -> Vec<String> {
    snap.api_messages
        .iter()
        .filter_map(|m| m["content"].as_array())
        .flat_map(|b| b.iter())
        .filter(|b| b["type"] == "tool_result")
        .map(|b| b["content"].as_str().unwrap_or("").to_string())
        .collect()
}

fn prompt_id(env: &Envelope) -> Option<u64> {
    match &env.event {
        SessionEventWire::Prompt(p) => Some(p.id),
        _ => None,
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn pending_prompt_survives_detach_and_replays_on_attach() {
    let _h = Home::new();
    let (url, _) = stub_seq(&[SSE_PROMPT_TOOL_USE, SSE_HI]).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = prompt_host().await;
    let handle = host.create_session(cfg()).await.unwrap();

    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    a.send(submit("go")).await.unwrap();
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Prompt(_))).await;
    let pid = prompt_id(seen.last().unwrap()).unwrap();
    // Never replayed: prompts are not in the turn ring.
    a.send(SessionCommand::Detach {
        client: a.client_id(),
    })
    .await
    .unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::ClientLeft { .. })).await;
    tokio::time::sleep(Duration::from_millis(200)).await;
    assert!(handle.is_alive(), "tool blocks on the prompt; session stays live");

    let (mut b, snap) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Attach))
        .await
        .unwrap();
    assert!(snap.streaming);
    assert_eq!(snap.pending_prompts.len(), 1);
    assert_eq!(snap.pending_prompts[0].id, pid);
    assert_eq!(snap.pending_prompts[0].title, "Secret");
    assert!(
        !snap.replay.iter().any(|e| matches!(e.event, SessionEventWire::Prompt(_))),
        "prompt is in pending_prompts, not in replay"
    );

    b.send(SessionCommand::Answer {
        prompt_id: pid,
        value: Some("s3cret".into()),
    })
    .await
    .unwrap();
    let seen = until(&mut b, |e| matches!(e, SessionEventWire::Idle)).await;
    assert!(seen.iter().any(|e| matches!(
        e.event,
        SessionEventWire::PromptResolved { prompt_id } if prompt_id == pid
    )));
    let conv = last_conversation(&seen);
    assert_eq!(tool_results(&conv), vec!["answered:6".to_string()]);
    assert!(
        !seen.iter().any(|e| format!("{:?}", e.event).contains("s3cret")),
        "the answer never appears on the event stream"
    );
    end(&mut b).await;
}

/// P11: a pending prompt on a NON-driver session survives the last-client
/// detach (the reattach-to-answer contract) AND a fast re-attach cancels the
/// abandonment deadline — the prompt is NOT auto-denied while a client is back,
/// so the user answers it normally even past the original deadline.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn pending_prompt_reattach_cancels_abandon_deadline() {
    let _h = Home::new();
    // Tiny 1 s deadline so the test is fast; re-attach well within it.
    std::env::set_var("SYNAPS_DAEMON_PROMPT_ABANDON_SECS", "1");
    let (url, _) = stub_seq(&[SSE_PROMPT_TOOL_USE, SSE_HI]).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = prompt_host().await;
    let handle = host.create_session(cfg()).await.unwrap();

    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    a.send(submit("go")).await.unwrap();
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Prompt(_))).await;
    let pid = prompt_id(seen.last().unwrap()).unwrap();

    // Detach the only client → the abandonment deadline arms (1 s).
    a.send(SessionCommand::Detach { client: a.client_id() })
        .await
        .unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::ClientLeft { .. })).await;
    drop(a);

    // Re-attach quickly (< 1 s): the prompt survived detach and the re-attach
    // cancels the deadline.
    let (mut b, snap) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Attach))
        .await
        .unwrap();
    assert_eq!(
        snap.pending_prompts.len(),
        1,
        "prompt must survive detach on a non-driver session"
    );
    assert_eq!(snap.pending_prompts[0].id, pid);

    // Wait PAST the original 1 s deadline while attached: the prompt must NOT be
    // auto-denied — the deadline was cancelled by the re-attach.
    tokio::time::sleep(Duration::from_millis(1400)).await;
    assert!(handle.is_alive(), "session stays live with the prompt pending");

    // Answer succeeds → the tool sees the real value (`answered:6`), proving the
    // prompt was never fail-closed to `None`.
    b.send(SessionCommand::Answer {
        prompt_id: pid,
        value: Some("s3cret".into()),
    })
    .await
    .unwrap();
    let seen = until(&mut b, |e| matches!(e, SessionEventWire::Idle)).await;
    assert!(seen.iter().any(|e| matches!(
        e.event,
        SessionEventWire::PromptResolved { prompt_id } if prompt_id == pid
    )));
    assert_eq!(tool_results(&last_conversation(&seen)), vec!["answered:6".to_string()]);
    end(&mut b).await;
    std::env::remove_var("SYNAPS_DAEMON_PROMPT_ABANDON_SECS");
}

/// P11: when NObody re-attaches before the abandonment deadline, the pending
/// prompt is fail-closed (`None` / denied) so `can_park()` unblocks and the
/// session parks normally — bounding the abandoned-prompt resource pin without
/// punishing a transient detach.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn pending_prompt_denied_and_parks_after_abandon_deadline() {
    let _h = Home::new();
    // 1 s abandonment deadline, and park immediately once the turn drains.
    std::env::set_var("SYNAPS_DAEMON_PROMPT_ABANDON_SECS", "1");
    std::env::set_var("SYNAPS_DAEMON_PARK_GRACE_SECS", "0");
    let (url, _) = stub_seq(&[SSE_PROMPT_TOOL_USE, SSE_HI]).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = prompt_host().await;
    let handle = host
        .create_session(SessionConfig { persist: true, ..cfg() })
        .await
        .unwrap();

    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    a.send(submit("go")).await.unwrap();
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Prompt(_))).await;
    let pid = prompt_id(seen.last().unwrap()).unwrap();

    // Detach the only client and stay gone → the deadline (1 s) expires with
    // zero clients → the prompt is denied (`None`) → the tool returns
    // "cancelled" → the turn drains → `can_park()` becomes true → the session
    // parks (grace 0).
    a.send(SessionCommand::Detach { client: a.client_id() })
        .await
        .unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::ClientLeft { .. })).await;
    drop(a);

    // Deadline (1 s) + turn drain + park grace (0). Poll up to a few seconds.
    let parked = tokio::time::timeout(Duration::from_secs(6), async {
        loop {
            if matches!(
                handle.lifecycle(),
                agent_engine::session::SessionLifecycle::Parked
            ) {
                break true;
            }
            if !handle.is_alive() {
                break false; // Ended is also acceptable (still unpinned)
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
    })
    .await
    .expect("session must unpin (park/end) after the abandonment deadline");
    assert!(parked, "abandoned prompt did not unblock parking");

    // Re-attach: the prompt is gone (denied), not stuck pending.
    let (mut c, snap) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    assert!(
        snap.pending_prompts.is_empty(),
        "abandoned prompt was not denied at the deadline: {:?}",
        snap.pending_prompts
    );
    end(&mut c).await;
    handle.closed().await;
    std::env::remove_var("SYNAPS_DAEMON_PROMPT_ABANDON_SECS");
    std::env::remove_var("SYNAPS_DAEMON_PARK_GRACE_SECS");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn answer_dedup_on_prompt_id() {
    let _h = Home::new();
    let (url, _) = stub_seq(&[SSE_PROMPT_TOOL_USE, SSE_HI]).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = prompt_host().await;
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    a.send(submit("go")).await.unwrap();
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Prompt(_))).await;
    let pid = prompt_id(seen.last().unwrap()).unwrap();

    // Unknown id: ignored. Same id twice: second ignored.
    for (id, v) in [(pid + 100, "zzz"), (pid, "aaaaaa"), (pid, "bb")] {
        a.send(SessionCommand::Answer {
            prompt_id: id,
            value: Some(v.into()),
        })
        .await
        .unwrap();
    }
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Idle)).await;
    let resolved = seen
        .iter()
        .filter(|e| matches!(e.event, SessionEventWire::PromptResolved { .. }))
        .count();
    assert_eq!(resolved, 1, "exactly one PromptResolved");
    assert_eq!(tool_results(&last_conversation(&seen)), vec!["answered:6".to_string()]);
    end(&mut a).await;
}

/// on_session_end spy: records whether the session file existed when the
/// hook fired (save must precede it).
struct SessionEndSpy {
    file_present_at_hook: Arc<std::sync::Mutex<Option<bool>>>,
}

#[async_trait::async_trait]
impl agent_engine::extensions::runtime::ExtensionHandler for SessionEndSpy {
    fn id(&self) -> &str {
        "session-end-spy"
    }
    async fn handle(
        &self,
        event: &agent_engine::extensions::hooks::events::HookEvent,
    ) -> agent_engine::extensions::hooks::events::HookResult {
        if let Some(id) = &event.session_id {
            let present = agent_engine::core::session::Session::load(id).is_ok();
            *self.file_present_at_hook.lock().unwrap() = Some(present);
        }
        agent_engine::extensions::hooks::events::HookResult::Continue
    }
    async fn shutdown(&self) {}
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn end_saves_then_emits_session_end_then_ended() {
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    let at_hook = Arc::new(std::sync::Mutex::new(None));
    host.parts()
        .hook_bus
        .subscribe(
            agent_engine::extensions::hooks::events::HookKind::OnSessionEnd,
            Arc::new(SessionEndSpy {
                file_present_at_hook: Arc::clone(&at_hook),
            }),
            None,
            None,
            agent_engine::extensions::permissions::PermissionSet::from_strings(&[
                "session.lifecycle".to_string(),
            ]),
        )
        .await
        .unwrap();

    let handle = host
        .create_session(SessionConfig {
            persist: true,
            ..cfg()
        })
        .await
        .unwrap();
    let id = handle.id.clone();
    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    a.send(submit("hello")).await.unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::Idle)).await;

    a.send(SessionCommand::End {
        reason: EndReason::ClientQuit,
    })
    .await
    .unwrap();
    let tail = until(&mut a, |e| matches!(e, SessionEventWire::Ended { .. })).await;
    // Ended is the last envelope; the hook already ran and saw the file.
    assert_eq!(
        *at_hook.lock().unwrap(),
        Some(true),
        "on_session_end fired after save, before Ended"
    );
    assert!(matches!(
        tail.last().unwrap().event,
        SessionEventWire::Ended {
            reason: EndReason::ClientQuit
        }
    ));
    assert!(a.next_event().await.is_none(), "nothing after Ended");
    let saved = agent_engine::core::session::Session::load(id.as_str()).expect("on disk");
    assert_eq!(saved.api_messages.len(), 2);
    handle.closed().await;
}

/// Host-level `daemon stop`: End{HostShutdown} to N live sessions
/// concurrently → N session files on disk.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn host_shutdown_saves_every_session() {
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    const N: usize = 3;
    let mut clients = Vec::new();
    for i in 0..N {
        let handle = host
            .create_session(SessionConfig {
                persist: true,
                ..cfg()
            })
            .await
            .unwrap();
        let (mut t, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
            .await
            .unwrap();
        t.send(submit(&format!("hello {i}"))).await.unwrap();
        until(&mut t, |e| matches!(e, SessionEventWire::Idle)).await;
        clients.push((handle, t));
    }
    assert_eq!(host.sessions().len(), N);

    let ends = clients.into_iter().map(|(handle, mut t)| async move {
        t.send(SessionCommand::End {
            reason: EndReason::HostShutdown,
        })
        .await
        .unwrap();
        until(&mut t, |e| matches!(e, SessionEventWire::Ended { .. })).await;
        handle.closed().await;
        handle.id.clone()
    });
    let ids = tokio::time::timeout(
        Duration::from_secs(agent_engine::session::budgets::TEARDOWN_TIMEOUT_SECS),
        futures::future::join_all(ends),
    )
    .await
    .expect("all sessions end within the teardown budget");
    for (i, id) in ids.iter().enumerate() {
        let saved = agent_engine::core::session::Session::load(id.as_str())
            .unwrap_or_else(|e| panic!("session {id:?} on disk: {e}"));
        assert_eq!(saved.api_messages.len(), 2);
        assert_eq!(saved.api_messages[0]["content"], format!("hello {i}"));
    }
    tokio::time::sleep(Duration::from_millis(50)).await;
    assert_eq!(host.sessions().len(), 0);
}

/// Bug: the session name was frozen at create — `SessionMeta.name` and the
/// `run/<sid>.json` registry entry were written once, so `synaps send
/// --session <name>` only resolved sessions created via `--continue <name>`.
/// Now `SessionConfig.name` names the session at create and `saveas` updates
/// both the handle and the registry.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn name_at_create_and_saveas_reach_registry_and_meta() {
    use agent_engine::events::registry::find_session_registration;
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;

    let handle = host
        .create_session(SessionConfig {
            name: Some("ambient".into()),
            persist: true,
            ..cfg()
        })
        .await
        .unwrap();
    // Registry + listing carry the name from the first second (no turn, no --continue).
    let reg = find_session_registration("ambient").expect("send --session ambient resolves");
    assert_eq!(reg.session_id, handle.id.as_str());
    assert_eq!(handle.name().as_deref(), Some("ambient"));
    assert_eq!(host.sessions()[0].name.as_deref(), Some("ambient"));
    // ...and it is on disk, so `--continue ambient` resolves via the journal too.
    let on_disk = agent_engine::core::session::resolve_session("ambient").expect("saved at create");
    assert_eq!(on_disk.id, handle.id.as_str());

    // A second create with the same name is refused (saveas rules), not silently duplicated.
    let dup = host
        .create_session(SessionConfig {
            name: Some("ambient".into()),
            persist: true,
            ..cfg()
        })
        .await;
    assert!(dup.is_err(), "duplicate name must be refused");
    assert_eq!(host.sessions().len(), 1);

    // Rename via `/cmd saveas` → handle, listing and registry all follow.
    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    a.send(SessionCommand::EngineCommand {
        id: 1,
        name: "saveas".into(),
        arg: "ambient2".into(),
    })
    .await
    .unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::QueryResult { id: 1, .. })).await;
    assert_eq!(handle.name().as_deref(), Some("ambient2"));
    assert_eq!(host.sessions()[0].name.as_deref(), Some("ambient2"));
    assert!(find_session_registration("ambient").is_none(), "old name gone");
    assert_eq!(
        find_session_registration("ambient2").expect("new name resolves").session_id,
        handle.id.as_str()
    );

    // `--continue ambient2` while live → the same actor (name resolved in
    // memory first, then disk; either way never a second actor).
    let again = host
        .create_session(SessionConfig {
            continue_session: Some(Some("ambient2".into())),
            persist: true,
            ..cfg()
        })
        .await
        .unwrap();
    assert_eq!(again.id, handle.id);
    assert_eq!(host.sessions().len(), 1);
    // Non-persisted session: its name exists ONLY in memory — still attaches.
    let eph = host
        .create_session(SessionConfig {
            name: Some("ephemeral".into()),
            persist: false,
            ..cfg()
        })
        .await
        .unwrap();
    assert!(agent_engine::core::session::resolve_session("ephemeral").is_err(), "not on disk");
    let again = host
        .create_session(SessionConfig {
            continue_session: Some(Some("ephemeral".into())),
            ..cfg()
        })
        .await
        .unwrap();
    assert_eq!(again.id, eph.id);
    assert_eq!(host.sessions().len(), 2);
    eph.send(SessionCommand::End { reason: EndReason::HostShutdown }).await.unwrap();
    eph.closed().await;

    // Clear: `saveas` with no arg.
    a.send(SessionCommand::EngineCommand {
        id: 2,
        name: "saveas".into(),
        arg: String::new(),
    })
    .await
    .unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::QueryResult { id: 2, .. })).await;
    assert_eq!(handle.name(), None);
    assert!(find_session_registration("ambient2").is_none());
    assert!(find_session_registration(handle.id.as_str()).is_some(), "still registered by id");
    end(&mut a).await;
}

/// Bug: `events.auto_turn = false` was parsed but the actor hardcoded
/// `auto_turn_enabled = true` — in daemon mode the opt-out did nothing. The
/// event must still be injected and forwarded; no EventAuto turn may run.
/// Flipping the key back on (host.reload_config) re-enables auto turns.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn events_auto_turn_false_injects_without_a_turn() {
    use tokio::io::AsyncWriteExt;
    let h = Home::new();
    let config_path = h._dir.path().join(".synaps-cli").join("config");
    std::fs::write(&config_path, "events.auto_turn = false\n").unwrap();
    let (url, hits) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;
    assert!(!host.config().events.auto_turn, "config key read at boot");
    let handle = host.create_session(cfg()).await.unwrap();
    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();

    let push = |text: &'static str| {
        let reg = agent_engine::events::registry::find_session_registration(handle.id.as_str())
            .expect("registered");
        async move {
            let event = agent_engine::events::types::Event::simple(
                "test",
                text,
                Some(agent_engine::events::types::Severity::High),
            );
            let mut c = tokio::net::UnixStream::connect(&reg.socket_path).await.unwrap();
            c.write_all(&serde_json::to_vec(&event).unwrap()).await.unwrap();
            c.shutdown().await.unwrap();
        }
    };

    push("no turn please").await;
    // Injected + forwarded, then nothing: no TurnStarted within the window.
    let external = until(&mut a, |e| matches!(e, SessionEventWire::External(_))).await;
    assert!(external.iter().all(|e| !matches!(e.event, SessionEventWire::TurnStarted { .. })));
    let quiet = tokio::time::timeout(Duration::from_millis(1500), async {
        loop {
            let e = next(&mut a).await;
            if matches!(e.event, SessionEventWire::TurnStarted { .. }) {
                return e.event;
            }
        }
    })
    .await;
    assert!(quiet.is_err(), "auto_turn=false must not start a turn: {quiet:?}");
    assert_eq!(hits.load(Ordering::SeqCst), 0, "provider never called");

    // Opt back in without a restart: the actor reads the host config per wake.
    std::fs::write(&config_path, "events.auto_turn = true\n").unwrap();
    host.reload_config();
    assert!(host.config().events.auto_turn);
    push("now turn").await;
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Idle)).await;
    assert!(seen.iter().any(|e| matches!(
        e.event,
        SessionEventWire::TurnStarted { trigger: agent_engine::session::TurnTrigger::EventAuto, .. }
    )));
    assert_eq!(hits.load(Ordering::SeqCst), 1);
    end(&mut a).await;
}

/// T5: env + env_stripped survive park → unpark.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn env_survives_park_unpark() {
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    std::env::set_var("SYNAPS_DAEMON_PARK_GRACE_SECS", "0");
    let host = host().await;
    let env_pairs: Vec<(String, String)> = vec![
        ("PATH".into(), "/usr/bin".into()),
        ("CUSTOM_VAR".into(), "myvalue".into()),
    ];
    let handle = host
        .create_session(SessionConfig {
            env: Some(env_pairs.clone()),
            env_stripped: vec!["GH_TOKEN".into()],
            persist: true,
            ..cfg()
        })
        .await
        .unwrap();
    let (mut a, _snap) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    // Run one turn so the session has history.
    a.send(submit("hi")).await.unwrap();
    until(&mut a, |e| matches!(e, SessionEventWire::Idle)).await;
    // Detach → parks (grace=0).
    a.send(SessionCommand::Detach { client: a.client_id() }).await.unwrap();
    // Drop the transport so we can re-attach.
    drop(a);
    // Wait for park.
    tokio::time::sleep(Duration::from_millis(500)).await;
    // Re-attach → unparks.
    let (mut b, snap) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    // Conversation history survived.
    assert!(
        snap.conversation.api_messages.len() >= 2,
        "history survived park/unpark"
    );
    // Session env should be carried in the config (verified via view or
    // the actor's runtime). Since we can't directly inspect the runtime
    // from here, we verify the session handle meta is still valid.
    assert_eq!(handle.meta().id, snap.meta.id, "same session after unpark");
    end(&mut b).await;
    handle.closed().await;
}

/// F18: a session with zero turns and zero clients cannot park (nothing to
/// journal), so at the park deadline it must END — not sit Live with a
/// Runtime resident forever. A session WITH history parks as before.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn zero_turn_session_ends_idle_at_park_deadline_while_one_turn_parks() {
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    std::env::set_var("SYNAPS_DAEMON_PARK_GRACE_SECS", "0");
    std::env::set_var("SYNAPS_DAEMON_IDLE_END_GRACE_SECS", "0");
    let host = host().await;

    // (a) zero turns → Ended(Idle), handle closes.
    let handle = host
        .create_session(SessionConfig { persist: true, ..cfg() })
        .await
        .unwrap();
    let (mut a, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    a.send(SessionCommand::Detach { client: a.client_id() }).await.unwrap();
    let ended = tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            match a.next_event().await {
                Some(e) => {
                    if let SessionEventWire::Ended { reason } = e.event {
                        break Some(reason);
                    }
                }
                None => break None,
            }
        }
    })
    .await
    .expect("zero-turn session must end at the park deadline");
    assert!(
        matches!(ended, Some(EndReason::Idle) | None),
        "expected Ended(Idle) (or the stream closing on end), got {ended:?}"
    );
    tokio::time::timeout(Duration::from_secs(5), handle.closed())
        .await
        .expect("handle must close — the actor must not stay Live");

    // (b) one turn → Parked, still resumable.
    let handle = host
        .create_session(SessionConfig { persist: true, ..cfg() })
        .await
        .unwrap();
    let (mut b, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    b.send(submit("hi")).await.unwrap();
    until(&mut b, |e| matches!(e, SessionEventWire::Idle)).await;
    b.send(SessionCommand::Detach { client: b.client_id() }).await.unwrap();
    drop(b);
    tokio::time::sleep(Duration::from_millis(500)).await;
    assert_eq!(handle.lifecycle(), agent_engine::session::SessionLifecycle::Parked, "one-turn session parks");
    let (mut c, snap) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Test))
        .await
        .unwrap();
    assert!(snap.conversation.api_messages.len() >= 2, "history survived park");
    end(&mut c).await;
    handle.closed().await;
}

/// Track F: attachments ride `Submit` as content blocks. A model that cannot
/// take images gets a typed `Refused` addressed to the submitter (so the TUI
/// restores the editor and keeps the drafts) and history is untouched; a
/// model that can takes the turn with the blocks in the user message.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn submit_with_image_is_refused_on_text_model_and_accepted_on_image_model() {
    const PNG: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aX1sAAAAASUVORK5CYII=";
    let image = serde_json::json!({"type":"image","source":{"type":"base64","media_type":"image/png","data":PNG}});
    let _h = Home::new();
    let (url, _) = stub(SSE_HI, false).await;
    std::env::set_var("SYNAPS_ANTHROPIC_BASE_URL", &url);
    let host = host().await;

    // (a) text-only model → Refused to the submitter, history unchanged.
    let handle = host
        .create_session(SessionConfig {
            model_override: Some("openai-codex/gpt-5.3-codex-spark".into()),
            ..cfg()
        })
        .await
        .unwrap();
    let (mut a, snap) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    let before = snap.conversation.api_messages.len();
    // Client-addressed (what the TUI does) — a host-originated `send` has
    // no client to refuse to and gets a `SystemNotice` instead.
    a.send_from_self(SessionCommand::Submit { text: "what is this".into(), attachments: vec![image.clone()] })
        .await
        .unwrap();
    let seen = until(&mut a, |e| matches!(e, SessionEventWire::Refused { .. })).await;
    let refused = seen
        .iter()
        .find_map(|e| match &e.event {
            SessionEventWire::Refused { client, command, reason } => Some((*client, command.clone(), reason.clone())),
            _ => None,
        })
        .expect("Refused event");
    assert_eq!(refused.0, a.client_id(), "refusal is addressed to the submitter");
    assert_eq!(refused.1, "submit");
    assert!(refused.2.contains("attachments rejected"), "{}", refused.2);
    assert!(
        !seen.iter().any(|e| matches!(e.event, SessionEventWire::TurnStarted { .. })),
        "no turn may start on a refused submit"
    );
    let (mut a2, snap2) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    assert_eq!(snap2.conversation.api_messages.len(), before, "history untouched");
    end(&mut a2).await;
    drop(a);
    handle.closed().await;

    // (b) image-capable model → turn starts with the block in the user message.
    // Provider-prefixed id: the validator's static modality table is keyed
    // that way (an unprefixed `claude-sonnet-4-5` has no image evidence in a
    // test process with an empty capability cache and is refused too).
    let handle = host
        .create_session(SessionConfig {
            model_override: Some("anthropic/claude-sonnet-4-6".into()),
            ..cfg()
        })
        .await
        .unwrap();
    let (mut b, _) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    b.send_from_self(SessionCommand::Submit { text: "what is this".into(), attachments: vec![image.clone()] })
        .await
        .unwrap();
    let seen = until(&mut b, |e| matches!(e, SessionEventWire::Idle)).await;
    assert!(seen.iter().any(|e| matches!(e.event, SessionEventWire::TurnStarted { .. })));
    let (mut c, snap3) = LocalTransport::attach(handle.clone(), ClientMeta::new(ClientKind::Tui))
        .await
        .unwrap();
    let user = snap3.conversation.api_messages.first().expect("user message");
    let blocks = user["content"].as_array().expect("multipart content");
    assert!(blocks.iter().any(|b| b["type"] == "image"), "image block on the wire: {user}");
    assert!(blocks.iter().any(|b| b["type"] == "text" && b["text"] == "what is this"));
    end(&mut c).await;
    drop(b);
    handle.closed().await;
}