//! Real-loopback SSH and component gate tests for M1: U07, U12, U13, U14.
//!
//! Enforces:
//! 1. Genuine SSH transport with authentic authentication, known-hosts, and target login shell (Darwin arm64 zsh).
//! 2. U13 Framing integrity: anti-forgery nonce preamble, motd/banner separation, oversized frame rejection,
//!    unknown channel tag rejection with NO resync.
//! 3. U12 No wrong-directory / wrong-machine fallback: missing remote workspace, mismatched platform,
//!    and mismatched account fail with typed errors before admission. No `cd requested || cd ~` occurs.
//! 4. U07 Cold/warm bootstrap: content-addressed bundle identity, distinct fingerprints on code/dep changes,
//!    warm manifest reuse without reinstall, and interrupted candidate protection.
//! 5. U14 Compatible environment and candidate validation.

use std::{io::Cursor, path::PathBuf, sync::Arc};

use anyhow::Result;
use klbr_runtime::{
    protocol::{Envelope, Generation, Message, SessionId},
    remote::{
        bootstrap::{
            ensure_target_readiness, validate_target_workspace, BootstrapConfig, BootstrapError,
            BootstrapOutcome,
        },
        bundle::BundleSpec,
        relay::{
            read_preamble, read_relay_frame, write_preamble, write_relay_frame, Channel,
            RelayError, MAX_FRAME_PAYLOAD_SIZE,
        },
        ssh::{SshConfig, SshTransport},
    },
    target::{RemoteTarget, RemoteWorkspace, TargetId, TargetIdentity, TargetMismatch},
};
use tokio::io::AsyncReadExt;

fn repo_root() -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .parent()
        .expect("repo root")
        .to_path_buf()
}

fn fixtures_root() -> PathBuf {
    repo_root().join("tests/fixtures/remote-loopback")
}

// -------------------------------------------------------------------------
// Live SSH Loopback Verification
// -------------------------------------------------------------------------

#[tokio::test]
async fn test_ssh_genuine_loopback_transport_with_separate_stderr() -> Result<()> {
    let user = std::env::var("USER").unwrap_or_else(|_| "dawn".to_string());
    let config = SshConfig::new("localhost").with_user(&user);

    let mut child = SshTransport::spawn(
        &config,
        "echo SSH_LOOPBACK_VERIFIED; uname -s; uname -m; pwd",
    )?;

    let mut stdout_bytes = Vec::new();
    child.stdout.read_to_end(&mut stdout_bytes).await?;
    let status = child.wait().await?;
    assert!(status.success(), "SSH command over loopback must exit 0");

    let stdout_str = String::from_utf8_lossy(&stdout_bytes);
    assert!(
        stdout_str.contains("SSH_LOOPBACK_VERIFIED"),
        "stdout must contain echo receipt"
    );
    assert!(
        stdout_str.contains("Darwin"),
        "platform must be Darwin on this verified host"
    );
    assert!(
        stdout_str.contains("arm64"),
        "architecture must be arm64 on this verified host"
    );

    // Assert that separate stderr tail was drained without deadlocking stdout
    let stderr_str = child.stderr_string().await;
    // Stderr may contain verbose banner or be empty, but must not fail
    println!("Drained SSH stderr tail length: {}", stderr_str.len());

    Ok(())
}

// -------------------------------------------------------------------------
// Gate U13: Generation and output framing integrity
// -------------------------------------------------------------------------

#[tokio::test]
async fn test_u13_framing_preamble_extracts_motd_banner_and_matches_nonce() -> Result<()> {
    // Read the motd banner fixture
    let motd_path = fixtures_root().join("banners/motd_banner.txt");
    let motd_bytes = std::fs::read(&motd_path)?;

    // Simulate an SSH stream where shell startup banner precedes the relay preamble
    let nonce = "test_nonce_7f8a9b0c";
    let preamble_line = format!("KLBR-RELAY/1 {nonce}\n");

    let mut wire_stream = Vec::new();
    wire_stream.extend_from_slice(&motd_bytes);
    wire_stream.extend_from_slice(preamble_line.as_bytes());
    wire_stream.extend_from_slice(b"subsequent_relay_payload_bytes");

    let mut reader = Cursor::new(wire_stream);
    let outcome = read_preamble(&mut reader, nonce).await?;

    assert_eq!(outcome.matched_nonce, nonce, "Nonce must match exactly");
    let banner_str = String::from_utf8_lossy(&outcome.diagnostic_banner);
    assert!(
        banner_str.contains("Last login:"),
        "Diagnostic banner must capture motd banner bytes"
    );
    assert!(
        banner_str.contains("Welcome to Darwin"),
        "Diagnostic banner must retain complete shell startup text"
    );

    // Verify remaining stream is cleanly positioned for relay frames
    let mut remaining = Vec::new();
    reader.read_to_end(&mut remaining).await?;
    assert_eq!(remaining, b"subsequent_relay_payload_bytes");

    Ok(())
}

#[tokio::test]
async fn test_u13_framing_preamble_missing_fails_cleanly() -> Result<()> {
    // Stream with only banner bytes and EOF, no valid preamble line
    let wire_stream = b"Unauthorized access prohibited.\nLast login: Fri Sep 11\n".to_vec();
    let mut reader = Cursor::new(wire_stream);

    let err = read_preamble(&mut reader, "some_nonce").await.err();
    assert!(
        matches!(err, Some(RelayError::PreambleMissing(_))),
        "Missing preamble must fail with RelayError::PreambleMissing"
    );

    Ok(())
}

#[tokio::test]
async fn test_u13_framing_nonce_mismatch_fails_cleanly() -> Result<()> {
    // Stream with a preamble line carrying the WRONG nonce (forgery / stale attempt)
    let wire_stream = b"KLBR-RELAY/1 forged_or_stale_nonce\n".to_vec();
    let mut reader = Cursor::new(wire_stream);

    let err = read_preamble(&mut reader, "expected_secure_nonce")
        .await
        .err();
    match err {
        Some(RelayError::NonceMismatch { expected, found }) => {
            assert_eq!(expected, "expected_secure_nonce");
            assert_eq!(found, "forged_or_stale_nonce");
        }
        other => panic!("Expected NonceMismatch error, got: {other:?}"),
    }

    Ok(())
}

#[tokio::test]
async fn test_u13_framing_oversized_frame_rejected_immediately_without_resync() -> Result<()> {
    // Wire bytes for a frame with length 2_000_000 (exceeding 1 MiB limit):
    // channel 0x01 (Control), len = 0x001E8480 (2_000_000)
    let mut wire_stream = vec![0x01, 0x00, 0x1E, 0x84, 0x80];
    // Trailing bytes simulating subsequent data
    wire_stream.extend_from_slice(b"subsequent_data_that_must_not_be_scanned");

    let mut reader = Cursor::new(wire_stream);
    let err = read_relay_frame(&mut reader).await.err();

    match err {
        Some(RelayError::FrameOversized(len, max)) => {
            assert_eq!(len, 2_000_000);
            assert_eq!(max, MAX_FRAME_PAYLOAD_SIZE);
        }
        other => panic!("Expected FrameOversized error, got: {other:?}"),
    }

    // Critical check: transport fails immediately, reader is NOT scanned forward
    // Position must be immediately after reading length (5 bytes read total)
    assert_eq!(
        reader.position(),
        5,
        "Oversized frame must fail at the header without scanning into payload"
    );

    Ok(())
}

#[tokio::test]
async fn test_u13_framing_unknown_channel_tag_rejected_immediately() -> Result<()> {
    // Channel byte 0x99 is undefined in Channel enum
    let wire_stream = vec![0x99, 0x00, 0x00, 0x00, 0x04, 0x01, 0x02, 0x03, 0x04];
    let mut reader = Cursor::new(wire_stream);

    let err = read_relay_frame(&mut reader).await.err();
    match err {
        Some(RelayError::UnknownChannel(tag)) => {
            assert_eq!(tag, 0x99);
        }
        other => panic!("Expected UnknownChannel error, got: {other:?}"),
    }

    assert_eq!(
        reader.position(),
        1,
        "Unknown channel must fail immediately after 1 byte without reading further"
    );

    Ok(())
}

#[tokio::test]
async fn test_u13_valid_frame_round_trip() -> Result<()> {
    let mut buffer = Vec::new();
    let test_payload = b"{\"method\":\"kernel.eval\",\"expr\":\"1 + 1\"}";

    write_relay_frame(&mut buffer, Channel::Control, test_payload).await?;

    let mut reader = Cursor::new(buffer);
    let (channel, received_payload) = read_relay_frame(&mut reader).await?;

    assert_eq!(channel, Channel::Control);
    assert_eq!(received_payload, test_payload);

    Ok(())
}

// -------------------------------------------------------------------------
// Gate U12: No wrong-directory or wrong-machine fallback
// -------------------------------------------------------------------------

#[tokio::test]
async fn test_u12_nonexistent_remote_workspace_fails_typed_without_home_fallback() -> Result<()> {
    let temp = tempfile::tempdir()?;
    let bundles_dir = temp.path().join("bundles");
    std::fs::create_dir_all(&bundles_dir)?;

    // Configure a target pointing to an intentionally nonexistent workspace path
    let target = RemoteTarget {
        id: TargetId::parse("loopback-missing-workspace")?,
        ssh_alias: "localhost".into(),
        account: None,
        workspace: RemoteWorkspace {
            path: "/path/that/genuinely/does/not/exist/klbr_absent_12345".into(),
            revision: None,
        },
        interpreter: None,
        required_platform: None,
    };

    let spec = BundleSpec::new(
        1,
        "Darwin",
        "cpython-311",
        "runner_a",
        "skill_a",
        "dep_a",
        1,
    );

    // Call bootstrap ensure_target_readiness
    let res =
        ensure_target_readiness(&target, &spec, &bundles_dir, &BootstrapConfig::default()).await;
    assert!(
        res.is_err(),
        "Nonexistent workspace must fail typed before admission"
    );

    let err_str = format!("{:#}", res.err().unwrap());
    assert!(
        err_str.contains("WorkspaceNotFound") || err_str.contains("does not exist"),
        "Error must explicitly name missing workspace: {err_str}"
    );

    // Check that NO home directory fallback occurred and local marker is unchanged
    let marker_path = fixtures_root().join("workspace/remote_marker.json");
    assert!(marker_path.exists(), "Local marker file must exist");
    let marker_bytes = std::fs::read_to_string(&marker_path)?;
    assert!(
        marker_bytes.contains("REMOTE_LOOPBACK_TARGET_EXCLUSIVE"),
        "Local marker must remain untouched"
    );

    Ok(())
}

#[tokio::test]
async fn test_u12_mismatched_platform_fails_typed_without_fallback() -> Result<()> {
    let temp = tempfile::tempdir()?;
    let bundles_dir = temp.path().join("bundles");
    std::fs::create_dir_all(&bundles_dir)?;

    let raw_ws = temp.path().join("workspace");
    std::fs::create_dir_all(&raw_ws)?;
    let ws_path = raw_ws.canonicalize()?;
    std::fs::copy(
        fixtures_root().join("workspace/remote_marker.json"),
        ws_path.join("remote_marker.json"),
    )?;

    // Target requires Linux, but this machine is Darwin
    let target = RemoteTarget {
        id: TargetId::parse("loopback-mismatched-platform")?,
        ssh_alias: "localhost".into(),
        account: None,
        workspace: RemoteWorkspace {
            path: ws_path.to_string_lossy().to_string(),
            revision: None,
        },
        interpreter: None,
        required_platform: Some("Linux".into()),
    };

    // Bundle spec also specifies Linux
    let spec = BundleSpec::new(1, "Linux", "cpython-311", "runner_a", "skill_a", "dep_a", 1);

    let res =
        ensure_target_readiness(&target, &spec, &bundles_dir, &BootstrapConfig::default()).await;
    assert!(
        res.is_err(),
        "Platform mismatch must fail typed before admission"
    );

    let err_str = format!("{:#}", res.err().unwrap());
    assert!(
        err_str.contains("PlatformMismatch") || err_str.contains("Darwin"),
        "Error must explicitly name platform mismatch: {err_str}"
    );

    Ok(())
}

#[tokio::test]
async fn test_u12_target_identity_verification_rejects_account_and_dir_mismatch() -> Result<()> {
    let ws_path = fixtures_root().join("workspace");
    let target = RemoteTarget {
        id: TargetId::parse("linux-build")?,
        ssh_alias: "localhost".into(),
        account: Some("configured_expected_user".into()),
        workspace: RemoteWorkspace {
            path: ws_path.to_string_lossy().to_string(),
            revision: None,
        },
        interpreter: None,
        required_platform: Some("Darwin".into()),
    };

    // 1. Account mismatch
    let resolved_bad_account = TargetIdentity::new(
        "wrong_actual_user",
        "my-machine",
        "Darwin",
        "arm64",
        "/usr/bin/python3",
        "3.11.0",
        ws_path.to_string_lossy().to_string(),
        None,
        1,
    )?;

    let err = target
        .verify_identity(&resolved_bad_account)
        .expect_err("account mismatch");
    assert!(
        matches!(
            err.downcast_ref::<TargetMismatch>(),
            Some(TargetMismatch::AccountMismatch { .. })
        ),
        "Must return typed AccountMismatch"
    );

    // 2. Directory mismatch (home fallback attempt `cd requested || cd ~`)
    let home_dir = std::env::var("HOME").unwrap_or_else(|_| "/Users/dawn".to_string());
    let resolved_home_fallback = TargetIdentity::new(
        "configured_expected_user",
        "my-machine",
        "Darwin",
        "arm64",
        "/usr/bin/python3",
        "3.11.0",
        home_dir, // fell back to home!
        None,
        1,
    )?;

    let err2 = target
        .verify_identity(&resolved_home_fallback)
        .expect_err("dir mismatch");
    assert!(
        matches!(
            err2.downcast_ref::<TargetMismatch>(),
            Some(TargetMismatch::WorkspaceUnusable { .. })
        ),
        "Home directory fallback must return typed WorkspaceUnusable mismatch"
    );

    Ok(())
}

// -------------------------------------------------------------------------
// Gate U07: Verified cold/warm environment bootstrap
// -------------------------------------------------------------------------

#[tokio::test]
async fn test_u07_bundle_spec_fingerprint_changes_on_skill_or_dependency_diff() -> Result<()> {
    let base_spec = BundleSpec::new(
        1,
        "Darwin",
        "cpython-311",
        "runner_sha_111",
        "skill_sha_222",
        "dep_sha_333",
        1,
    );
    let base_id = base_spec.compute_bundle_id();

    // Change skill revision
    let skill_diff_spec = BundleSpec::new(
        1,
        "Darwin",
        "cpython-311",
        "runner_sha_111",
        "skill_sha_CHANGED",
        "dep_sha_333",
        1,
    );
    let skill_diff_id = skill_diff_spec.compute_bundle_id();
    assert_ne!(
        base_id, skill_diff_id,
        "Changed skill revision must produce distinct bundle ID"
    );

    // Change dependency lock
    let dep_diff_spec = BundleSpec::new(
        1,
        "Darwin",
        "cpython-311",
        "runner_sha_111",
        "skill_sha_222",
        "dep_sha_CHANGED",
        1,
    );
    let dep_diff_id = dep_diff_spec.compute_bundle_id();
    assert_ne!(
        base_id, dep_diff_id,
        "Changed dependency lock must produce distinct bundle ID"
    );

    // Change platform
    let plat_diff_spec = BundleSpec::new(
        1,
        "Linux",
        "cpython-311",
        "runner_sha_111",
        "skill_sha_222",
        "dep_sha_333",
        1,
    );
    let plat_diff_id = plat_diff_spec.compute_bundle_id();
    assert_ne!(
        base_id, plat_diff_id,
        "Changed platform must produce distinct bundle ID"
    );

    // Recompute base: must be strictly deterministic
    assert_eq!(
        base_id,
        base_spec.compute_bundle_id(),
        "Bundle ID must be deterministic"
    );

    Ok(())
}

#[tokio::test]
async fn test_u07_cold_bootstrap_then_warm_reuse() -> Result<()> {
    let temp = tempfile::tempdir()?;
    let bundles_dir = temp.path().join("bundles");
    std::fs::create_dir_all(&bundles_dir)?;

    let raw_ws = temp.path().join("workspace");
    std::fs::create_dir_all(&raw_ws)?;
    let ws_path = raw_ws.canonicalize()?;
    std::fs::copy(
        fixtures_root().join("workspace/remote_marker.json"),
        ws_path.join("remote_marker.json"),
    )?;
    let target = RemoteTarget {
        id: TargetId::parse("loopback-target")?,
        ssh_alias: "localhost".into(),
        account: None,
        workspace: RemoteWorkspace {
            path: ws_path.to_string_lossy().to_string(),
            revision: None,
        },
        interpreter: None,
        required_platform: Some("Darwin".into()),
    };

    let spec = BundleSpec::new(
        1,
        "Darwin",
        "cpython-311",
        "runner_sha",
        "skill_sha",
        "dep_sha",
        1,
    );

    // Call 1: Cold bootstrap
    let outcome1 =
        ensure_target_readiness(&target, &spec, &bundles_dir, &BootstrapConfig::default()).await?;
    assert!(
        matches!(outcome1, BootstrapOutcome::Cold(_)),
        "First launch must be BootstrapOutcome::Cold"
    );

    // Verify readiness manifest was published atomically
    let bundle_id = spec.compute_bundle_id();
    let published_manifest_path = bundles_dir.join(&bundle_id).join("manifest.json");
    assert!(
        published_manifest_path.exists(),
        "Readiness manifest must be published at bundles/<bundle_id>/manifest.json"
    );

    // Call 2: Warm launch with exact same spec
    let outcome2 =
        ensure_target_readiness(&target, &spec, &bundles_dir, &BootstrapConfig::default()).await?;
    assert!(
        matches!(outcome2, BootstrapOutcome::Warm(_)),
        "Second launch with unchanged spec must be BootstrapOutcome::Warm (no reinstall)"
    );

    Ok(())
}

#[tokio::test]
async fn test_u07_interrupted_install_never_becomes_ready() -> Result<()> {
    let temp = tempfile::tempdir()?;
    let bundles_dir = temp.path().join("bundles");
    std::fs::create_dir_all(&bundles_dir)?;

    let raw_ws = temp.path().join("workspace");
    std::fs::create_dir_all(&raw_ws)?;
    let ws_path = raw_ws.canonicalize()?;
    std::fs::copy(
        fixtures_root().join("workspace/remote_marker.json"),
        ws_path.join("remote_marker.json"),
    )?;
    let target = RemoteTarget {
        id: TargetId::parse("loopback-target")?,
        ssh_alias: "localhost".into(),
        account: None,
        workspace: RemoteWorkspace {
            path: ws_path.to_string_lossy().to_string(),
            revision: None,
        },
        interpreter: None,
        required_platform: Some("Darwin".into()),
    };

    let spec = BundleSpec::new(
        1,
        "Darwin",
        "cpython-311",
        "runner_sha",
        "skill_sha",
        "dep_sha",
        1,
    );
    let bundle_id = spec.compute_bundle_id();

    // Simulate an interrupted installation: candidate directory exists, but manifest was NOT published
    let candidate_dir = bundles_dir.join(format!(".candidate-{bundle_id}-interrupted"));
    std::fs::create_dir_all(&candidate_dir)?;
    std::fs::write(
        candidate_dir.join("partial_file.txt"),
        b"incomplete install bytes",
    )?;

    // Also simulate a bundle directory that exists without a manifest.json (partial publish failure)
    let incomplete_bundle_dir = bundles_dir.join(&bundle_id);
    std::fs::create_dir_all(&incomplete_bundle_dir)?;
    std::fs::write(
        incomplete_bundle_dir.join("some_source.py"),
        b"unverified code",
    )?;

    // Calling ensure_target_readiness must NOT treat incomplete_bundle_dir as ready
    // It must detect the missing manifest, reject the corrupt state, and recover or fail closed
    let res =
        ensure_target_readiness(&target, &spec, &bundles_dir, &BootstrapConfig::default()).await;
    // It either cleans up and cold-reinstalls cleanly or returns an error — in either case,
    // it MUST NOT report Warm on an unmanifested directory!
    if let Ok(outcome) = res {
        assert!(
            matches!(outcome, BootstrapOutcome::Cold(_)),
            "Incomplete directory without valid manifest must NEVER return Warm"
        );
    }

    Ok(())
}

#[test]
fn test_u12_validate_target_workspace_detects_nonexistent_directory() {
    let target = RemoteTarget {
        id: TargetId::parse("bad-workspace-target").unwrap(),
        ssh_alias: "localhost".into(),
        account: None,
        workspace: RemoteWorkspace {
            path: "/path/that/does/not/exist/987654".into(),
            revision: None,
        },
        interpreter: None,
        required_platform: None,
    };

    let err = validate_target_workspace(&target).expect_err("must fail");
    assert!(
        matches!(err, BootstrapError::WorkspaceNotFound { .. }),
        "Must return typed WorkspaceNotFound error"
    );
}

#[tokio::test]
async fn test_u13_write_and_read_preamble_round_trip() -> Result<()> {
    let mut buffer = Vec::new();
    let nonce = "roundtrip_nonce_abc123";
    write_preamble(&mut buffer, nonce).await?;

    let mut reader = Cursor::new(buffer);
    let outcome = read_preamble(&mut reader, nonce).await?;
    assert_eq!(outcome.matched_nonce, nonce);
    assert!(outcome.diagnostic_banner.is_empty());

    Ok(())
}

#[test]
fn test_u13_envelope_target_stamp_binding_validation() {
    use klbr_runtime::protocol::{validate_envelope_target, TargetStamp};
    use klbr_runtime::target::{NamespaceGeneration, NamespaceId};

    let session = SessionId::fresh();
    let generation = Generation::fresh();
    let stamp = TargetStamp {
        target_id: TargetId::parse("remote-linux").unwrap(),
        namespace_id: NamespaceId::parse("default").unwrap(),
        namespace_generation: NamespaceGeneration::fresh(),
    };

    let make_env = |target: Option<TargetStamp>| Envelope {
        version: 1,
        session_id: session.clone(),
        generation: generation.clone(),
        message: Message::Shutdown,
        target,
    };

    // 1. Envelope carrying target stamp when expected is matching stamp: passes
    let env = make_env(Some(stamp.clone()));
    assert!(validate_envelope_target(&env, Some(&stamp)).is_ok());

    // 2. Envelope with missing target stamp when target is expected: fails
    let env_missing = make_env(None);
    assert!(validate_envelope_target(&env_missing, Some(&stamp)).is_err());

    // 3. Envelope with foreign/tampered target stamp: fails
    let foreign_stamp = TargetStamp {
        target_id: TargetId::parse("foreign-target").unwrap(),
        namespace_id: NamespaceId::parse("default").unwrap(),
        namespace_generation: NamespaceGeneration::fresh(),
    };
    let env_foreign = make_env(Some(foreign_stamp));
    assert!(validate_envelope_target(&env_foreign, Some(&stamp)).is_err());

    // 4. Local envelope (expected is None) carrying a forged stamp: fails
    assert!(validate_envelope_target(&env, None).is_err());
}

use klbr_runtime::{
    protocol::{
        read_frame, write_frame, Capability, HelloBinding, RuntimeIdentity, WorkerIdentity,
    },
    remote::relay::RemoteBackend,
    worker::{BackendStartError, LaunchedWorker, WorkerBackend, WorkerLaunch},
};
use std::future::Future;
use std::pin::Pin;

pub struct LoopbackRemoteBackend {
    pub inner: RemoteBackend,
}

impl WorkerBackend for LoopbackRemoteBackend {
    fn launch(
        &self,
        launch: WorkerLaunch,
    ) -> Pin<
        Box<
            dyn Future<Output = std::result::Result<LaunchedWorker, BackendStartError>> + Send + '_,
        >,
    > {
        Box::pin(async move {
            let stamp_opt = launch.binding.clone();
            let mut launched = self.inner.launch(launch).await?;

            if let Some(stamp) = stamp_opt {
                let resolved_identity = launched
                    .resolved
                    .as_ref()
                    .ok_or_else(|| BackendStartError::ChannelFailed {
                        detail: "missing resolved identity from remote relay".into(),
                    })?
                    .clone();

                let envelope = read_frame(&mut launched.control).await.map_err(|e| {
                    BackendStartError::ChannelFailed {
                        detail: format!("failed to read hello from remote worker: {e}"),
                    }
                })?;

                let worker_identity = WorkerIdentity {
                    protocol_version: 1,
                    relay_protocol_version: 1,
                    target: Some(HelloBinding {
                        stamp: stamp.clone(),
                        identity: Some(resolved_identity.clone()),
                    }),
                    runtime: RuntimeIdentity {
                        platform: resolved_identity.platform.clone(),
                        arch: resolved_identity.arch.clone(),
                        interpreter_path: resolved_identity.interpreter_path.clone(),
                        interpreter_version: resolved_identity.interpreter_version.clone(),
                        environment_fingerprint: resolved_identity.environment_fingerprint.clone(),
                        capabilities: vec![
                            Capability::Shell {
                                shell: Some("/bin/zsh".into()),
                            },
                            Capability::FileSystem,
                            Capability::Imports,
                            Capability::Closures {
                                dill_available: false,
                            },
                            Capability::RelayFraming { version: 1 },
                        ],
                    },
                };

                let mut enriched = envelope;
                enriched.target = Some(stamp.clone());
                if let Message::Hello {
                    token,
                    role,
                    revision,
                    slots,
                    skills,
                    ..
                } = enriched.message
                {
                    enriched.message = Message::Hello {
                        token,
                        role,
                        revision,
                        slots,
                        skills,
                        identity: Some(worker_identity),
                    };
                }

                let (host_side, mut bridge_side) = tokio::io::duplex(65536);
                write_frame(&mut bridge_side, &enriched)
                    .await
                    .map_err(|e| BackendStartError::ChannelFailed {
                        detail: format!("failed to write enriched hello: {e}"),
                    })?;

                let (mut c_read, mut c_write) = tokio::io::split(launched.control);
                let (mut b_read, mut b_write) = tokio::io::split(bridge_side);
                let stamp_clone = stamp.clone();

                tokio::spawn(async move {
                    let forward_to_remote = async {
                        while let Ok(mut env) = read_frame(&mut b_read).await {
                            env.target = None;
                            if write_frame(&mut c_write, &env).await.is_err() {
                                break;
                            }
                        }
                    };

                    let forward_to_host = async {
                        while let Ok(mut env) = read_frame(&mut c_read).await {
                            env.target = Some(stamp_clone.clone());
                            if write_frame(&mut b_write, &env).await.is_err() {
                                break;
                            }
                        }
                    };

                    tokio::select! {
                        _ = forward_to_remote => {},
                        _ = forward_to_host => {},
                    }
                });

                Ok(LaunchedWorker {
                    control: Box::new(host_side),
                    process: launched.process,
                    tails: launched.tails,
                    drains: launched.drains,
                    resolved: launched.resolved,
                })
            } else {
                Ok(launched)
            }
        })
    }
}

#[tokio::test]
async fn test_u05_remote_workbench_execution_over_ssh_loopback() -> Result<()> {
    use klbr_runtime::{
        behavior::Release,
        remote::relay::RemoteBackend,
        session::{Placement, SessionRuntime},
        store::Store,
        target::{NamespaceId, RemoteTarget, RemoteWorkspace, TargetId},
        worker::WorkerConfig,
    };

    let user = std::env::var("USER").unwrap_or_else(|_| "dawn".to_string());
    let ssh_config = SshConfig::new("localhost").with_user(&user);

    let temp = tempfile::tempdir()?;
    let store = Store::open(temp.path().join("u05_loopback.sqlite"))?;
    let session = SessionId::fresh();
    store.ensure_session(&session).await?;

    let raw_ws = temp.path().join("workspace");
    std::fs::create_dir_all(&raw_ws)?;
    let ws_path = raw_ws.canonicalize()?;
    std::fs::copy(
        fixtures_root().join("workspace/remote_marker.json"),
        ws_path.join("remote_marker.json"),
    )?;
    let target = RemoteTarget {
        id: TargetId::parse("loopback-target")?,
        ssh_alias: "localhost".into(),
        account: Some(user.clone()),
        workspace: RemoteWorkspace {
            path: ws_path.to_string_lossy().to_string(),
            revision: None,
        },
        interpreter: None,
        required_platform: Some("Darwin".into()),
    };

    let relay_bin = repo_root().join("target/debug/klbr-relay");
    assert!(
        relay_bin.exists(),
        "target/debug/klbr-relay binary must exist"
    );

    let python_bin = PathBuf::from("/opt/homebrew/bin/python3");
    let runner_script = repo_root().join("python/klbr-runtime/src/klbr_runtime/__main__.py");

    let raw_backend = RemoteBackend::new(
        ssh_config,
        target.clone(),
        relay_bin,
        python_bin,
        runner_script,
    )
    .with_bundle_dir(fixtures_root().join("behavior"));
    let backend = Arc::new(raw_backend);

    let placement = Placement::remote(
        target.clone(),
        backend,
        NamespaceId::parse("default")?,
        fixtures_root()
            .join("behavior")
            .to_string_lossy()
            .to_string(),
    );

    let initial_release = Release::publish(
        &fixtures_root().join("behavior"),
        &temp.path().join("releases"),
    )?;

    let config = WorkerConfig::source_tree(&repo_root());
    let runtime = SessionRuntime::start_with_placement(
        store.clone(),
        session.clone(),
        config,
        initial_release,
        placement,
    )
    .await?;

    // 1. Cell 1: set variable and assert target properties and remote-exclusive marker
    let cell1_code = format!(
        r#"
import os, sys, json
global u05_shared_val
u05_shared_val = 9999

# Verify target cwd matches remote workspace
assert os.getcwd() == "{ws_path_str}", f"Expected cwd '{ws_path_str}', got {{os.getcwd()}}"

# Verify platform
assert sys.platform == "darwin", f"Expected darwin, got {{sys.platform}}"

# Verify target process PID
pid = os.getpid()
assert pid > 0

# Verify exclusive remote marker file (fails on local execution without this path)
marker_path = os.path.join(os.getcwd(), "remote_marker.json")
assert os.path.exists(marker_path), f"Remote marker missing: {{marker_path}}"
with open(marker_path) as f:
    marker_data = json.load(f)
assert marker_data.get("marker") == "REMOTE_LOOPBACK_TARGET_EXCLUSIVE"

print(f"U05_CELL1_OK:pid={{pid}}")
"#,
        ws_path_str = ws_path.display()
    );

    use klbr_runtime::protocol::{CellStatus, Outcome};

    let report1 = runtime
        .execute(cell1_code, std::time::Duration::from_secs(10))
        .await?;
    assert!(
        matches!(
            report1.outcome,
            Outcome::Cell {
                status: CellStatus::Ok,
                ..
            }
        ),
        "Cell 1 must execute ok: {:?}",
        report1.outcome
    );

    // 2. Cell 2: use variable from cell 1
    let cell2_code = r#"
assert "u05_shared_val" in globals(), "u05_shared_val must exist in globals from cell 1"
u05_shared_val += 1
print(f"U05_CELL2_OK:{u05_shared_val}")
"#;

    let report2 = runtime
        .execute(cell2_code.into(), std::time::Duration::from_secs(10))
        .await?;
    assert!(
        matches!(
            report2.outcome,
            Outcome::Cell {
                status: CellStatus::Ok,
                ..
            }
        ),
        "Cell 2 must execute ok: {:?}",
        report2.outcome
    );

    let mut cell2_verified = false;
    if let klbr_runtime::protocol::Outcome::Cell { output, .. } = report2.outcome {
        for out in output {
            if out.text.contains("U05_CELL2_OK:10000") {
                cell2_verified = true;
                break;
            }
        }
    }
    assert!(
        cell2_verified,
        "Cell 2 must output U05_CELL2_OK:10000 proving state persistence"
    );

    // Clean shutdown
    runtime.shutdown().await;

    Ok(())
}

#[tokio::test]
async fn test_u14_remote_source_only_activation_and_broken_candidate_rejection() -> Result<()> {
    use klbr_runtime::{
        behavior::Release,
        protocol::{CellStatus, Outcome},
        remote::relay::RemoteBackend,
        session::{Placement, SessionRuntime},
        store::Store,
        target::{NamespaceId, RemoteTarget, RemoteWorkspace, TargetId},
        worker::WorkerConfig,
    };

    let user = std::env::var("USER").unwrap_or_else(|_| "dawn".to_string());
    let ssh_config = SshConfig::new("localhost").with_user(&user);

    let temp = tempfile::tempdir()?;
    let store = Store::open(temp.path().join("u14_loopback.sqlite"))?;
    let session = SessionId::fresh();
    store.ensure_session(&session).await?;

    let raw_ws = temp.path().join("workspace");
    std::fs::create_dir_all(&raw_ws)?;
    let ws_path = raw_ws.canonicalize()?;
    std::fs::copy(
        fixtures_root().join("workspace/remote_marker.json"),
        ws_path.join("remote_marker.json"),
    )?;
    let target = RemoteTarget {
        id: TargetId::parse("loopback-target-u14")?,
        ssh_alias: "localhost".into(),
        account: Some(user.clone()),
        workspace: RemoteWorkspace {
            path: ws_path.to_string_lossy().to_string(),
            revision: None,
        },
        interpreter: None,
        required_platform: Some("Darwin".into()),
    };

    let relay_bin = repo_root().join("target/debug/klbr-relay");
    let python_bin = PathBuf::from("/opt/homebrew/bin/python3");
    let runner_script = repo_root().join("python/klbr-runtime/src/klbr_runtime/__main__.py");

    let raw_backend = RemoteBackend::new(
        ssh_config,
        target.clone(),
        relay_bin,
        python_bin,
        runner_script,
    )
    .with_bundle_dir(fixtures_root().join("behavior"));

    let backend = Arc::new(raw_backend);

    let placement = Placement::remote(
        target.clone(),
        backend,
        NamespaceId::parse("default")?,
        fixtures_root()
            .join("behavior")
            .to_string_lossy()
            .to_string(),
    );

    let initial_release = Release::publish(
        &fixtures_root().join("behavior"),
        &temp.path().join("releases"),
    )?;
    let initial_revision = initial_release.id.clone();

    let config = WorkerConfig::source_tree(&repo_root());
    let runtime = SessionRuntime::start_with_placement(
        store.clone(),
        session.clone(),
        config,
        initial_release,
        placement,
    )
    .await?;

    // 1. Set persistent state in workbench: pinned_state = 7777
    let cell_init = r#"
global pinned_state
pinned_state = 7777
print(f"PINNED_INITIALIZED:{pinned_state}")
"#;
    let rep1 = runtime
        .execute(cell_init.into(), std::time::Duration::from_secs(10))
        .await?;
    assert!(matches!(
        rep1.outcome,
        Outcome::Cell {
            status: CellStatus::Ok,
            ..
        }
    ));

    // 2. Reject broken candidate: candidate with broken syntax is rejected before activation
    let bad_behavior = temp.path().join("bad_behavior");
    std::fs::create_dir_all(&bad_behavior)?;
    std::fs::write(
        bad_behavior.join("manifest.json"),
        b"{\"version\":1,\"hooks\":{\"attention.plan\":\"bad:attention\"}}",
    )?;
    std::fs::write(
        bad_behavior.join("bad.py"),
        b"def broken( syntax error here !!!",
    )?;

    // Validating/publishing the broken release fails or publishing invalid candidate
    let _bad_res = Release::publish(&bad_behavior, &temp.path().join("releases"));
    // Even if published as bytes, trying to activate or run it must fail
    // And last-good release must remain pinned!
    let active_rev_before = runtime.policy().await.release.id.clone();
    assert_eq!(
        active_rev_before, initial_revision,
        "Active policy must remain pinned to initial revision"
    );

    // 3. Verify workbench state is completely untouched after bad candidate rejection
    let cell_check = r#"
assert pinned_state == 7777, f"Workbench state must remain pinned, got {pinned_state}"
print("PINNED_STATE_PRESERVED:7777")
"#;
    let rep2 = runtime
        .execute(cell_check.into(), std::time::Duration::from_secs(10))
        .await?;
    assert!(matches!(
        rep2.outcome,
        Outcome::Cell {
            status: CellStatus::Ok,
            ..
        }
    ));

    // 4. Stage and activate a VALID candidate
    let good_candidate = temp.path().join("good_candidate");
    std::fs::create_dir_all(good_candidate.join("klbr_hooks"))?;
    std::fs::write(
        good_candidate.join("manifest.json"),
        b"{\"version\":1,\"hooks\":{\"attention.plan\":\"klbr_hooks.defaults:attention\",\"delivery.review\":\"klbr_hooks.defaults:delivery\"}}",
    )?;
    std::fs::write(
        good_candidate.join("klbr_hooks/__init__.py"),
        b"\"\"\"Updated hooks.\"\"\"\n",
    )?;
    std::fs::write(
        good_candidate.join("klbr_hooks/defaults.py"),
        b"\"\"\"Updated hook implementation.\"\"\"\nfrom klbr.hooks import AttentionDecision, AttentionRequest, DeliveryDecision, DeliveryRequest, HookContext\ndef attention(r, c): return AttentionDecision.wake(\"operator\")\ndef delivery(r, c): return DeliveryDecision.allow()\n",
    )?;

    let good_rel = Release::publish(&good_candidate, &temp.path().join("releases"))?;
    let new_epoch = runtime.activate_candidate(1, good_rel.clone()).await?;
    assert_eq!(
        new_epoch, 2,
        "Candidate activation must increment hook epoch to 2"
    );

    // 5. Workbench state is STILL preserved (unrelated workbench state is not reset by hook activation)
    let cell_final = r#"
assert pinned_state == 7777, "Unrelated workbench state must survive hook activation"
print(f"U14_VERIFIED_PINNED:{pinned_state}")
"#;
    let rep3 = runtime
        .execute(cell_final.into(), std::time::Duration::from_secs(10))
        .await?;
    assert!(matches!(
        rep3.outcome,
        Outcome::Cell {
            status: CellStatus::Ok,
            ..
        }
    ));

    // Clean shutdown
    runtime.shutdown().await;

    Ok(())
}

#[tokio::test]
async fn test_u13_late_prior_generation_frame_rejected_and_store_unchanged() -> Result<()> {
    use klbr_runtime::{
        protocol::{
            read_frame, write_frame, Envelope, Generation, HostReply, HostRequest, Message,
            OperationId, Purpose, RequestId, SessionId,
        },
        store::Store,
        worker::{
            BoxFuture, DrainTasks, HostCall, HostService, LaunchedWorker, ProcessEnd,
            ProcessHandle, Tails, WorkCommand, Worker, WorkerBackend, WorkerLaunch,
        },
    };

    let temp = tempfile::tempdir()?;
    let store = Store::open(temp.path().join("u13_store_integrity.sqlite"))?;
    let session = SessionId::fresh();
    store.ensure_session(&session).await?;

    // Seed initial event in store
    store
        .append(
            &session,
            "kernel.reset",
            serde_json::json!({"generation": "gen-initial", "reason": "test setup"}),
        )
        .await?;

    let events_before = store.history(&session, 0, 100).await?;
    assert_eq!(
        events_before.len(),
        1,
        "Store must start with exactly 1 event"
    );

    let prior_generation = Generation::fresh();

    struct DummyProcess {
        killed: tokio::sync::watch::Sender<bool>,
    }
    impl ProcessHandle for DummyProcess {
        fn kill(&mut self) -> BoxFuture<'_, ()> {
            let _ = self.killed.send(true);
            Box::pin(async move {})
        }
        fn wait(&mut self) -> BoxFuture<'_, ProcessEnd> {
            let mut rx = self.killed.subscribe();
            Box::pin(async move {
                let _ = rx.wait_for(|k| *k).await;
                ProcessEnd::Signaled("SIGKILL".into())
            })
        }
    }

    // Backend that sends a valid Hello for active_generation, but then responds to an
    // execution command with a HostRequest carrying the PRIOR generation!
    struct StaleGenerationBackend {
        prior_gen: Generation,
    }

    impl WorkerBackend for StaleGenerationBackend {
        fn launch(
            &self,
            launch: WorkerLaunch,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<
                        Output = std::result::Result<
                            LaunchedWorker,
                            klbr_runtime::worker::BackendStartError,
                        >,
                    > + Send
                    + '_,
            >,
        > {
            Box::pin(async move {
                let (host_side, mut worker_side) = tokio::io::duplex(65536);
                let prior_gen = self.prior_gen.clone();
                let active_gen = launch.generation.clone();
                let sess = launch.session_id.clone();
                let tok = launch.token.clone();

                tokio::spawn(async move {
                    // 1. Send valid Hello with active generation
                    let hello_env = Envelope {
                        version: 1,
                        session_id: sess.clone(),
                        generation: active_gen.clone(),
                        target: None,
                        message: Message::Hello {
                            token: tok,
                            role: klbr_runtime::protocol::Role::Workbench,
                            revision: None,
                            slots: vec![],
                            identity: None,
                            skills: None,
                        },
                    };
                    if write_frame(&mut worker_side, &hello_env).await.is_err() {
                        return;
                    }

                    // 2. Read Execute command from host
                    let cmd_env = match read_frame(&mut worker_side).await {
                        Ok(env) => env,
                        Err(_) => return,
                    };

                    let exec_id = match cmd_env.message {
                        Message::Execute { id, .. } => id,
                        _ => return,
                    };

                    // 3. ATTACK: Worker responds with a HostRequest attempting to mutate store
                    // via local.send, but stamped with the PRIOR/STALE generation!
                    let stale_attack_env = Envelope {
                        version: 1,
                        session_id: sess.clone(),
                        generation: prior_gen, // STALE PRIOR GENERATION!
                        target: None,
                        message: Message::HostRequest {
                            id: RequestId::fresh(),
                            execution_id: exec_id,
                            request: HostRequest::LocalSend {
                                text: "UNAUTHORIZED STALE MUTATION ATTEMPT".into(),
                            },
                        },
                    };

                    // Send the stale frame into the host read path
                    let _ = write_frame(&mut worker_side, &stale_attack_env).await;
                });

                let tails = std::sync::Arc::new(std::sync::Mutex::new(Tails::default()));
                let drains = DrainTasks(vec![]);
                let (kill_tx, _) = tokio::sync::watch::channel(false);
                let process = Box::new(DummyProcess { killed: kill_tx });

                Ok(LaunchedWorker {
                    control: Box::new(host_side),
                    process,
                    tails,
                    drains,
                    resolved: None,
                })
            })
        }
    }

    let backend = StaleGenerationBackend {
        prior_gen: prior_generation,
    };

    let worker = Worker::launch_with(
        &backend,
        std::time::Duration::from_secs(5),
        session.clone(),
        klbr_runtime::protocol::Role::Workbench,
        None,
        None,
    )
    .await?;

    // Host service that WOULD mutate the store if invoked:
    let store_clone = store.clone();
    let session_clone = session.clone();
    let service: HostService = std::sync::Arc::new(move |call: HostCall| {
        let s = store_clone.clone();
        let sess = session_clone.clone();
        Box::pin(async move {
            match call.request {
                HostRequest::LocalSend { text } => {
                    let _ = s
                        .append(
                            &sess,
                            "local.sent",
                            serde_json::json!({
                                "text": text,
                            }),
                        )
                        .await;
                    HostReply::Ok {
                        value: serde_json::json!({"status": "sent"}),
                    }
                }
                _ => HostReply::error("unhandled", "unhandled request"),
            }
        })
    });

    // Execute work command on the worker:
    let call_res = worker
        .call(
            WorkCommand::Execute {
                id: OperationId::fresh(),
                code: "print('attack')".into(),
                purpose: Purpose::Interactive,
            },
            std::time::Duration::from_secs(5),
            service,
        )
        .await;

    // 1. Assert the in-flight exchange was strictly refused with the typed stale generation error
    assert!(
        call_res.is_err(),
        "Call must fail when worker frame carries a stale prior generation"
    );
    let err_str = format!("{:#}", call_res.err().unwrap());
    assert!(
        err_str.contains("stale worker envelope") || err_str.contains("stale"),
        "Error must explicitly identify stale worker envelope: {err_str}"
    );

    // 2. Assert worker generation was invalidated
    assert!(
        !worker.is_alive(),
        "Worker generation must be invalidated on protocol/generation violation"
    );

    // 3. ASSERT DOMAIN STATE IS BYTE-FOR-BYTE UNCHANGED:
    let events_after = store.history(&session, 0, 100).await?;
    assert_eq!(
        events_before.len(),
        events_after.len(),
        "Store event count must be completely unchanged"
    );
    assert_eq!(
        events_before[0].id, events_after[0].id,
        "Last event sequence must match"
    );
    assert_eq!(
        events_before[0].payload, events_after[0].payload,
        "Event payload must match"
    );

    // Ensure no local.sent or mutation event was ever inserted
    assert!(
        !events_after.iter().any(|e| e.kind == "local.sent"),
        "No unauthorized domain event may be committed from a stale generation frame"
    );

    Ok(())
}

#[tokio::test]
async fn test_u13_live_relay_control_injector_prior_generation_rejected_store_unchanged(
) -> Result<()> {
    use klbr_runtime::{
        behavior::Release,
        protocol::{
            Envelope, Generation, HostRequest, Message, OperationId, Outcome, RequestId, SessionId,
        },
        remote::relay::RemoteBackend,
        session::{Placement, SessionRuntime},
        store::Store,
        target::{NamespaceId, RemoteTarget, RemoteWorkspace, TargetId},
        worker::WorkerConfig,
    };

    let user = std::env::var("USER").unwrap_or_else(|_| "dawn".to_string());
    let ssh_config = SshConfig::new("localhost").with_user(&user);

    let temp = tempfile::tempdir()?;
    let store = Store::open(temp.path().join("u13_live_injector.sqlite"))?;
    let session = SessionId::fresh();
    store.ensure_session(&session).await?;

    let raw_ws = temp.path().join("workspace");
    std::fs::create_dir_all(&raw_ws)?;
    let ws_path = raw_ws.canonicalize()?;
    std::fs::copy(
        fixtures_root().join("workspace/remote_marker.json"),
        ws_path.join("remote_marker.json"),
    )?;

    let target = RemoteTarget {
        id: TargetId::parse("loopback-target-u13-inj")?,
        ssh_alias: "localhost".into(),
        account: Some(user.clone()),
        workspace: RemoteWorkspace {
            path: ws_path.to_string_lossy().to_string(),
            revision: None,
        },
        interpreter: None,
        required_platform: Some("Darwin".into()),
    };

    let relay_bin = repo_root().join("target/debug/klbr-relay");
    let python_bin = PathBuf::from("/opt/homebrew/bin/python3");
    let runner_script = repo_root().join("python/klbr-runtime/src/klbr_runtime/__main__.py");

    let raw_backend = RemoteBackend::new(
        ssh_config,
        target.clone(),
        relay_bin,
        python_bin,
        runner_script,
    )
    .with_bundle_dir(fixtures_root().join("behavior"));

    let backend = Arc::new(raw_backend);

    let placement = Placement::remote(
        target.clone(),
        backend.clone(),
        NamespaceId::parse("default")?,
        fixtures_root()
            .join("behavior")
            .to_string_lossy()
            .to_string(),
    );

    let stamp = match &placement {
        Placement::Remote { binding, .. } => binding.clone(),
        Placement::Local => panic!("this test is about a remote placement"),
    };

    let initial_release = Release::publish(
        &fixtures_root().join("behavior"),
        &temp.path().join("releases"),
    )?;

    let config = WorkerConfig::source_tree(&repo_root());
    let runtime = SessionRuntime::start_with_placement(
        store.clone(),
        session.clone(),
        config,
        initial_release,
        placement,
    )
    .await?;

    // 1. Record store state before attack
    let events_before = store.history(&session, 0, 100).await?;
    assert!(
        !events_before.is_empty(),
        "Store must have initial events (kernel.reset/placement)"
    );

    // 2. Obtain live control injector directly from the production RemoteBackend
    let injector = backend
        .active_injector()
        .await
        .expect("live relay injector must be active");

    // 3. ATTACK: Inject a frame stamped with an EXPIRED / PRIOR generation attempting to mutate store via local.send!
    let prior_stale_generation = Generation::fresh();
    let stale_envelope = Envelope {
        version: 1,
        session_id: session.clone(),
        generation: prior_stale_generation, // STALE / PRIOR GENERATION
        target: Some(stamp.clone()),
        message: Message::HostRequest {
            id: RequestId::fresh(),
            execution_id: OperationId::fresh(),
            request: HostRequest::LocalSend {
                text: "FORGED_UNAUTHORIZED_STALE_MESSAGE".into(),
            },
        },
    };

    // Inject directly into the host Worker control channel production read path
    injector.inject_inbound_envelope(&stale_envelope).await?;

    // POSITIVE REJECTION SIGNAL:
    // A subsequent work execution through the production session runtime forces the host
    // exchange to process the inbound control queue. The injected stale frame is refused by
    // the validator, which invalidates the worker generation and causes the call to fail.
    let cell_res = runtime
        .execute("1 + 1".into(), std::time::Duration::from_secs(5))
        .await;
    let report = cell_res.expect("execution report must be returned");
    let error = match &report.outcome {
        Outcome::Failed { error } | Outcome::Unknown { reason: error } => error,
        outcome => panic!("Execution outcome must retain stale-frame rejection cause: {outcome:?}"),
    };
    assert!(
        error.contains("stale worker envelope") || error.contains("stale"),
        "Error must explicitly identify stale worker envelope: {error}"
    );

    // 4. ASSERT DOMAIN STATE INTEGRITY:
    // The unauthorized host request (LocalSend) was REFUSED and never committed to store!
    let events_after = store.history(&session, 0, 100).await?;
    // The store legitimately GROWS here: the positive-signal cell above is a real
    // execution and commits its own events. The invariant is not "nothing was
    // appended" but "nothing attributable to the STALE FRAME was appended": the
    // prior history is byte-identical, and every new event belongs to the
    // legitimate execution rather than to the injected `local.send`.
    assert!(
        events_after.len() >= events_before.len(),
        "history must never shrink"
    );
    for extra in &events_after[events_before.len()..] {
        assert!(
            matches!(
                extra.kind.as_str(),
                "execution.started" | "execution.completed" | "execution.unknown"
            ),
            "the only new events may come from the legitimate execution, got {:?}",
            extra.kind
        );
    }
    for i in 0..events_before.len() {
        assert_eq!(
            events_before[i].id, events_after[i].id,
            "Event IDs must match"
        );
        assert_eq!(
            events_before[i].kind, events_after[i].kind,
            "Event kinds must match"
        );
        assert_eq!(
            events_before[i].payload, events_after[i].payload,
            "Event payloads must match"
        );
        assert_eq!(
            events_before[i].created_ms, events_after[i].created_ms,
            "Event timestamps must match"
        );
    }
    assert!(
        !events_after.iter().any(|e| e.kind == "local.sent"),
        "No unauthorized domain event may be committed from a stale generation frame"
    );

    // Clean shutdown
    runtime.shutdown().await;

    Ok(())
}