//! A concrete vertical slice: durable cells, explicit local delivery through a
//! separate policy worker, terminal waits, and hook-only revision replacement.
//! This is not a model loop or a Discord adapter.
use crate::{
    behavior::Release,
    hooks::Policy,
    protocol::*,
    remote::{
        callback::CallbackContext,
        exec::RemoteManager,
        lifecycle::{
            BootstrapRepair, ConnectionScopedLifecycle, ConnectionState, LifecycleSnapshot,
            PersistenceMode, RepairReceipt, RepairRequest,
        },
    },
    store::{PreparedEffect, Store},
    target::{KernelTarget, NamespaceGeneration, NamespaceId, RemoteTarget, SessionPlacement},
    work::{ConsequenceMetadata, Impact, Provenance, RunCause, WorkItem},
    worker::{
        BehaviorLocation, BehaviorSource, HostCall, HostService, WorkCommand, Worker,
        WorkerBackend, WorkerConfig,
    },
};
use anyhow::{anyhow, bail, ensure, Context, Result};
use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
use serde::Serialize;
use serde_json::{json, Value};
use std::{
    path::{Component, Path},
    process::Stdio,
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    },
    time::Duration,
};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    process::{Child, Command},
    sync::{Mutex, RwLock, Semaphore},
};

#[cfg(unix)]
use std::os::unix::process::CommandExt;

#[derive(Debug)]
struct RoutedHostError {
    code: String,
    message: String,
}

impl std::fmt::Display for RoutedHostError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.message)
    }
}

impl std::error::Error for RoutedHostError {}

struct ProcessGroup {
    child: Option<Child>,
    pid: i32,
    finished: bool,
}

impl ProcessGroup {
    fn new(child: Child) -> Result<Self> {
        let pid = child.id().context("workspace command has no process id")? as i32;
        Ok(Self {
            child: Some(child),
            pid,
            finished: false,
        })
    }

    fn child(&mut self) -> &mut Child {
        self.child.as_mut().expect("live process group has a child")
    }

    async fn terminate(&mut self) -> Result<std::process::ExitStatus> {
        terminate_process_group(self.pid);
        let waited = {
            let child = self.child();
            tokio::time::timeout(Duration::from_secs(2), child.wait()).await
        };
        match waited {
            Ok(status) => status.context("waiting for terminated workspace command"),
            Err(_) => {
                kill_process_group(self.pid);
                self.child()
                    .wait()
                    .await
                    .context("waiting for killed workspace command")
            }
        }
    }

    fn finish(&mut self) {
        self.finished = true;
    }
}

impl Drop for ProcessGroup {
    fn drop(&mut self) {
        if self.finished || self.child.is_none() {
            return;
        }
        terminate_process_group(self.pid);
        let pid = self.pid;
        let mut child = self.child.take().expect("checked above");
        if let Ok(runtime) = tokio::runtime::Handle::try_current() {
            runtime.spawn(async move {
                if tokio::time::timeout(Duration::from_secs(2), child.wait())
                    .await
                    .is_err()
                {
                    kill_process_group(pid);
                    let _ = child.wait().await;
                }
            });
        } else {
            kill_process_group(pid);
            let _ = child.start_kill();
        }
    }
}

#[cfg(unix)]
fn signal_process_group(pid: i32, signal: rustix::process::Signal) {
    // The child is the process-group leader. ESRCH only means it won the exit race.
    if let Some(pid) = rustix::process::Pid::from_raw(pid) {
        let _ = rustix::process::kill_process_group(pid, signal);
    }
}

#[cfg(unix)]
fn terminate_process_group(pid: i32) {
    signal_process_group(pid, rustix::process::Signal::TERM);
}
#[cfg(unix)]
fn kill_process_group(pid: i32) {
    signal_process_group(pid, rustix::process::Signal::KILL);
}
#[cfg(not(unix))]
fn terminate_process_group(_pid: i32) {}
#[cfg(not(unix))]
fn kill_process_group(_pid: i32) {}

#[derive(Default)]
struct CommandCapture {
    stdout: Vec<u8>,
    stderr: Vec<u8>,
    stored: usize,
    stdout_truncated: bool,
    stderr_truncated: bool,
}

async fn drain_command_output(
    mut reader: impl tokio::io::AsyncRead + Unpin,
    stdout: bool,
    limit: usize,
    capture: Arc<Mutex<CommandCapture>>,
) {
    let mut chunk = [0; 65_536];
    loop {
        let read = match reader.read(&mut chunk).await {
            Ok(0) | Err(_) => return,
            Ok(read) => read,
        };
        let mut capture = capture.lock().await;
        let keep = (limit - capture.stored).min(read);
        let truncated = keep != read;
        if stdout {
            capture.stdout.extend_from_slice(&chunk[..keep]);
            capture.stdout_truncated |= truncated;
        } else {
            capture.stderr.extend_from_slice(&chunk[..keep]);
            capture.stderr_truncated |= truncated;
        }
        capture.stored += keep;
    }
}

fn workspace_cwd(root: &Path, cwd: &Path) -> Result<std::path::PathBuf> {
    ensure!(!cwd.is_absolute(), "workspace cwd must be relative");
    ensure!(
        !cwd.components()
            .any(|part| matches!(part, Component::ParentDir | Component::Prefix(_))),
        "workspace cwd escapes source root"
    );
    let root = std::fs::canonicalize(root).context("cannot resolve workspace source root")?;
    let cwd = std::fs::canonicalize(root.join(cwd)).context("workspace cwd does not exist")?;
    ensure!(cwd.is_dir(), "workspace cwd is not a directory");
    ensure!(
        cwd.starts_with(&root),
        "workspace cwd escapes source root through a symlink"
    );
    Ok(cwd)
}

/// Session placement: which workbench executes `python(code)`, resolved once
/// per workbench generation. A remote placement binds a namespace stamp that
/// every envelope must echo; there is no fallback to local execution, home
/// directories, or another target.
#[derive(Clone)]
pub enum Placement {
    Local,
    Remote {
        target: RemoteTarget,
        /// The host-owned backend for the target, supplied by the target manager. Python never
        /// launches SSH itself.
        backend: Arc<dyn WorkerBackend>,
        /// Namespace binding issued at launch. The worker must echo it in its hello and every
        /// envelope; anything else is fatal.
        binding: TargetStamp,
        /// Target-side directory of the active behavior release.
        behavior_bundle: String,
    },
}

impl Placement {
    #[must_use]
    pub fn local() -> Self {
        Self::Local
    }

    #[must_use]
    pub fn remote(
        target: RemoteTarget,
        backend: Arc<dyn WorkerBackend>,
        namespace: NamespaceId,
        behavior_bundle: String,
    ) -> Self {
        Self::Remote {
            binding: TargetStamp {
                target_id: target.id.clone(),
                namespace_id: namespace,
                namespace_generation: NamespaceGeneration::fresh(),
            },
            backend,
            behavior_bundle,
            target,
        }
    }

    /// The kind of target this placement names.
    #[must_use]
    pub fn kernel_target(&self) -> KernelTarget {
        match self {
            Self::Local => KernelTarget::Local,
            Self::Remote { target, .. } => KernelTarget::Remote(target.clone()),
        }
    }

    /// The durable placement descriptor persisted under this generation.
    fn descriptor(&self, generation: &Generation) -> Result<SessionPlacement> {
        Ok(match self {
            Self::Local => SessionPlacement {
                kernel_target: KernelTarget::Local,
                namespace_id: NamespaceId::parse("main")?,
                namespace_generation: NamespaceGeneration::parse(generation.as_str())?,
            },
            Self::Remote {
                target, binding, ..
            } => SessionPlacement {
                kernel_target: KernelTarget::Remote(target.clone()),
                namespace_id: binding.namespace_id.clone(),
                namespace_generation: binding.namespace_generation.clone(),
            },
        })
    }
}

/// The target/workspace/interpreter identity the agent is actually using,
/// resolved ONCE per workbench generation - never flooded into prompts.
#[derive(Clone, Debug, Serialize)]
pub struct TargetIdentitySnapshot {
    pub kind: String,
    pub target: String,
    pub workspace: String,
    pub platform: String,
    pub arch: String,
    pub interpreter_path: String,
    pub interpreter_version: String,
    pub environment_fingerprint: Option<String>,
    pub capability_revision: u16,
    pub generation: String,
}

impl TargetIdentitySnapshot {
    fn resolve(
        placement: &Placement,
        config: &WorkerConfig,
        generation: &Generation,
        identity: Option<WorkerIdentity>,
    ) -> Result<Self> {
        match (placement, identity) {
            (Placement::Remote { target: remote, .. }, Some(worker_identity)) => {
                let runtime = &worker_identity.runtime;
                let environment_fingerprint = runtime
                    .environment_fingerprint
                    .as_ref()
                    .map(|f| f.as_str().to_string());
                Ok(Self {
                    kind: "remote".into(),
                    target: remote.id.to_string(),
                    workspace: remote.workspace.path.clone(),
                    platform: runtime.platform.clone(),
                    arch: runtime.arch.clone(),
                    interpreter_path: runtime.interpreter_path.clone(),
                    interpreter_version: runtime.interpreter_version.clone(),
                    environment_fingerprint,
                    capability_revision: worker_identity
                        .target
                        .as_ref()
                        .and_then(|binding| binding.identity.as_ref())
                        .map(|identity| identity.capability_revision)
                        .unwrap_or(0),
                    generation: generation.as_str().to_string(),
                })
            }
            (Placement::Remote { target, .. }, None) => bail!(
                "target {}: remote worker did not report its runtime identity",
                target.id
            ),
            (Placement::Local, _) => Ok(Self {
                kind: "local".into(),
                target: "local".into(),
                workspace: config.cwd.display().to_string(),
                platform: std::env::consts::OS.to_string(),
                arch: std::env::consts::ARCH.to_string(),
                interpreter_path: config.python.display().to_string(),
                interpreter_version: local_interpreter_version(config)?,
                environment_fingerprint: None,
                capability_revision: 0,
                generation: generation.as_str().to_string(),
            }),
        }
    }
}

/// One interpreter probe per workbench generation, not per prompt.
fn local_interpreter_version(config: &WorkerConfig) -> Result<String> {
    let output = std::process::Command::new(&config.python)
        .arg("-c")
        .arg("import sys;print(sys.version.split()[0])")
        .output()
        .context("local interpreter probe failed")?;
    ensure!(output.status.success(), "local interpreter probe failed");
    let version = String::from_utf8_lossy(&output.stdout).trim().to_string();
    ensure!(!version.is_empty(), "local interpreter reported no version");
    Ok(version)
}

struct ActivePolicy {
    epoch: i64,
    policy: Arc<Policy>,
}

struct PlacementState {
    placement: Placement,
    identity: TargetIdentitySnapshot,
}

pub struct SessionRuntime {
    pub id: SessionId,
    pub store: Store,
    config: WorkerConfig,
    workbench: Mutex<Worker>,
    policy: RwLock<ActivePolicy>,
    placement: RwLock<PlacementState>,
    /// Remote namespaces are connection-scoped in M2. `None` is the local
    /// adapter; `Some` becomes terminally lost on transport disconnect.
    lifecycle: RwLock<Option<ConnectionScopedLifecycle>>,
    remote: RemoteManager,
    foreground: Arc<Semaphore>,
    activation_gate: Arc<Semaphore>,
    stopped: AtomicBool,
    storage_failed: AtomicBool,
    pub orientation_host: Arc<std::sync::RwLock<Option<Arc<dyn OrientationHost>>>>,
    pub peer_host: Arc<std::sync::RwLock<Option<Arc<dyn PeerHost>>>>,
}

#[derive(Clone, Debug, Serialize)]
pub struct ExecutionReport {
    pub id: OperationId,
    pub generation: Generation,
    pub hook_revision: String,
    pub outcome: Outcome,
}

/// Spawn the workbench under the placement's backend. Local is the unchanged
/// adapter path; a remote launch goes through the host-owned backend and
/// carries the issued namespace binding. Returns the worker and its resolved
/// target identity snapshot.
async fn spawn_workbench(
    placement: &Placement,
    config: &WorkerConfig,
    session: &SessionId,
    release: &Release,
) -> Result<(Worker, TargetIdentitySnapshot)> {
    let worker = match placement {
        Placement::Local => {
            Worker::spawn(config, session.clone(), Role::Workbench, Some(release)).await?
        }
        Placement::Remote {
            backend,
            binding,
            behavior_bundle,
            ..
        } => {
            let behavior = BehaviorSource {
                revision: release.id.clone(),
                location: BehaviorLocation::RemoteBundle(behavior_bundle.clone()),
            };
            Worker::launch_with(
                backend.as_ref(),
                config.startup_timeout,
                session.clone(),
                Role::Workbench,
                Some(behavior),
                Some(binding.clone()),
            )
            .await?
        }
    };
    let identity = TargetIdentitySnapshot::resolve(
        placement,
        config,
        worker.generation(),
        worker.worker_identity(),
    )?;
    Ok((worker, identity))
}

fn lifecycle_for_placement(placement: &Placement) -> Result<Option<ConnectionScopedLifecycle>> {
    Ok(match placement {
        Placement::Local => None,
        Placement::Remote { binding, .. } => Some(ConnectionScopedLifecycle::new(
            binding.namespace_id.clone(),
            binding.namespace_generation.clone(),
        )),
    })
}

async fn run_workspace_command(
    root: &Path,
    argv: Vec<String>,
    cwd: std::path::PathBuf,
    env: std::collections::BTreeMap<String, String>,
    stdin_base64: Option<String>,
    timeout_ms: u64,
    max_output_bytes: u64,
) -> Result<Value> {
    ensure!(
        !argv.is_empty() && argv.iter().all(|part| !part.is_empty()),
        "argv must contain non-empty strings"
    );
    ensure!(
        argv.iter().all(|part| !part.contains('\0')),
        "argv contains NUL"
    );
    ensure!(
        timeout_ms > 0 && timeout_ms <= 3_600_000,
        "invalid workspace command timeout"
    );
    // Base64 expands by 4/3 and the reply carries metadata too; leave room for
    // its JSON envelope instead of accepting an output that cannot be framed.
    let output_limit =
        usize::try_from(max_output_bytes).context("workspace output limit is too large")?;
    ensure!(
        output_limit <= 750_000,
        "workspace output limit exceeds reply frame capacity"
    );
    ensure!(
        env.iter().all(|(key, value)| !key.is_empty()
            && !key.contains('\0')
            && !key.contains('=')
            && !value.contains('\0')),
        "workspace environment contains an invalid key or value"
    );
    let stdin = stdin_base64
        .map(|encoded| {
            BASE64
                .decode(encoded)
                .context("invalid base64 workspace stdin")
        })
        .transpose()?;
    let cwd = workspace_cwd(root, &cwd)?;
    let mut command = Command::new(&argv[0]);
    command
        .args(&argv[1..])
        .current_dir(&cwd)
        // This is an overlay on the daemon's already-sanitized environment;
        // it is not a reconstructed shell environment.
        .envs(&env)
        .stdin(if stdin.is_some() {
            Stdio::piped()
        } else {
            Stdio::null()
        })
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(false);
    #[cfg(unix)]
    command.as_std_mut().process_group(0);
    let started = tokio::time::Instant::now();
    let mut group = ProcessGroup::new(command.spawn().context("spawning workspace command")?)?;
    let capture = Arc::new(Mutex::new(CommandCapture::default()));
    let stdout = group
        .child()
        .stdout
        .take()
        .context("workspace command has no stdout")?;
    let stderr = group
        .child()
        .stderr
        .take()
        .context("workspace command has no stderr")?;
    let stdout_reader = tokio::spawn(drain_command_output(
        stdout,
        true,
        output_limit,
        capture.clone(),
    ));
    let stderr_reader = tokio::spawn(drain_command_output(
        stderr,
        false,
        output_limit,
        capture.clone(),
    ));
    let writer = group
        .child()
        .stdin
        .take()
        .zip(stdin)
        .map(|(mut pipe, input)| {
            tokio::spawn(async move {
                let _ = pipe.write_all(&input).await;
                let _ = pipe.shutdown().await;
            })
        });
    let (status, timed_out) =
        match tokio::time::timeout(Duration::from_millis(timeout_ms), group.child().wait()).await {
            Ok(status) => (status.context("waiting for workspace command")?, false),
            Err(_) => (group.terminate().await?, true),
        };
    group.finish();
    if let Some(writer) = writer {
        let _ = writer.await;
    }
    let _ = stdout_reader.await;
    let _ = stderr_reader.await;
    let capture = capture.lock().await;
    Ok(json!({
        "argv": argv,
        "cwd": cwd,
        "exit_code": status.code(),
        "stdout_base64": BASE64.encode(&capture.stdout),
        "stderr_base64": BASE64.encode(&capture.stderr),
        "duration_ms": started.elapsed().as_millis() as u64,
        "completed": !timed_out,
        "timed_out": timed_out,
        "cancelled": false,
        "stdout_truncated": capture.stdout_truncated,
        "stderr_truncated": capture.stderr_truncated,
    }))
}

impl SessionRuntime {
    pub async fn start(
        store: Store,
        id: SessionId,
        config: WorkerConfig,
        initial: Release,
    ) -> Result<Arc<Self>> {
        Self::start_with_placement(store, id, config, initial, Placement::local()).await
    }

    /// Placement-aware start. A remote placement launches the workbench on the
    /// configured target's backend and binds its namespace stamp; refusal at
    /// any point is typed and precedes execution. There is no local fallback.
    pub async fn start_with_placement(
        store: Store,
        id: SessionId,
        config: WorkerConfig,
        initial: Release,
        placement: Placement,
    ) -> Result<Arc<Self>> {
        store.ensure_session(&id).await?;
        let (epoch, release) = match store.activation(&id).await? {
            Some(active) => {
                let release = Release::load(&active.path)?;
                ensure!(
                    release.id == active.revision,
                    "active release bytes have changed"
                );
                (active.epoch, release)
            }
            None => (0, initial.clone()),
        };
        let policy = Policy::start(&config, id.clone(), release).await?;
        let epoch = if epoch == 0 {
            store.activate(&id, 0, policy.release.clone()).await?
        } else {
            epoch
        };
        let (workbench, identity) =
            spawn_workbench(&placement, &config, &id, &policy.release).await?;
        let lifecycle = lifecycle_for_placement(&placement)?;
        let remote = RemoteManager::new();
        if let Placement::Remote {
            target, backend, ..
        } = &placement
        {
            remote.register_relay_target(
                target.clone(),
                backend.clone(),
                config.startup_timeout,
            )?;
        }
        store
            .set_session_placement(&id, &placement.descriptor(workbench.generation())?)
            .await
            .context("failed to persist session placement")?;
        if let Placement::Remote { binding, .. } = &placement {
            store
                .append(
                    &id,
                    "kernel.placement",
                    json!({
                        "generation": workbench.generation().as_str(),
                        "binding": binding,
                        "reason": "workbench placed on configured target; namespace is fresh; heap not restored",
                    }),
                )
                .await?;
        }
        store
            .append(
                &id,
                "kernel.reset",
                json!({
                    "generation": workbench.generation(),
                    "reason": "session start; heap not restored",
                    // The instruction set this generation was spawned with, echoed by the
                    // worker. Without it the record would only say what the roots hold now.
                    "skills": workbench.skills_revision(),
                }),
            )
            .await?;
        Ok(Arc::new(Self {
            id,
            store,
            config,
            workbench: Mutex::new(workbench),
            policy: RwLock::new(ActivePolicy { epoch, policy }),
            placement: RwLock::new(PlacementState {
                placement,
                identity,
            }),
            lifecycle: RwLock::new(lifecycle),
            remote,
            foreground: Arc::new(Semaphore::new(1)),
            activation_gate: Arc::new(Semaphore::new(1)),
            stopped: AtomicBool::new(false),
            storage_failed: AtomicBool::new(false),
            orientation_host: Arc::new(std::sync::RwLock::new(None)),
            peer_host: Arc::new(std::sync::RwLock::new(Some(Arc::new(
                crate::peer::LocalPeerHost,
            )))),
        }))
    }

    pub fn set_orientation_host(&self, host: Arc<dyn OrientationHost>) {
        *self.orientation_host.write().unwrap() = Some(host);
    }

    pub fn set_peer_host(&self, host: Arc<dyn PeerHost>) {
        *self.peer_host.write().unwrap() = Some(host);
    }

    pub async fn hook_epoch(&self) -> i64 {
        self.policy.read().await.epoch
    }
    pub async fn hooks(&self) -> Value {
        self.policy.read().await.policy.list()
    }

    pub async fn policy(&self) -> Arc<Policy> {
        self.policy.read().await.policy.clone()
    }

    /// Returns the advertised persistence mode for a remote workbench.
    /// `None` denotes the unchanged local adapter.
    pub async fn persistence_mode(&self) -> Option<PersistenceMode> {
        self.lifecycle
            .read()
            .await
            .as_ref()
            .map(ConnectionScopedLifecycle::mode)
    }

    /// Returns the remote connection state. A lost state is terminal for this
    /// handle; only [`Self::reconnect`] may create a new namespace owner.
    pub async fn connection_state(&self) -> Option<ConnectionState> {
        self.lifecycle
            .read()
            .await
            .as_ref()
            .map(|lifecycle| lifecycle.snapshot().state)
    }

    /// The target, namespace, owner and state visible to the SDK.
    pub async fn lifecycle_snapshot(&self) -> Option<LifecycleSnapshot> {
        self.lifecycle
            .read()
            .await
            .as_ref()
            .map(ConnectionScopedLifecycle::snapshot)
    }

    pub fn workspace(&self) -> &std::path::Path {
        &self.config.cwd
    }

    /// Every id this session offers, read from disk on each call: a skill an agent
    /// just wrote is offered to the next load, not to the next restart.
    pub fn skill_catalog(&self) -> crate::skills::Catalog {
        self.config.skills.catalog()
    }

    /// Where those ids come from. Reported without the index, which has its own
    /// call: this is a location, not a listing.
    fn skill_roots_report(&self) -> Value {
        let catalog = self.skill_catalog();
        json!({
            "builtin_root": self.config.skills.builtin_root().display().to_string(),
            "managed_root": self.config.skills.managed_root().map(|path| path.display().to_string()),
            "ids": catalog.ids(),
            "refused_count": catalog.refused.len(),
            "revision": catalog.revision(),
        })
    }

    /// Candidate import/handshake precedes activation. Existing cells retain the
    /// Arc they pinned; this operation does not replace their workbench.
    /// Once accepted, caller cancellation does not interrupt the DB/cache switch.
    pub async fn activate_candidate(
        self: &Arc<Self>,
        expected_epoch: i64,
        release: Release,
    ) -> Result<i64> {
        self.activate_candidate_with_work(expected_epoch, release, None)
            .await
    }

    pub async fn activate_candidate_with_work(
        self: &Arc<Self>,
        expected_epoch: i64,
        release: Release,
        work_item_id: Option<String>,
    ) -> Result<i64> {
        self.activate_hooks_internal(expected_epoch, release, true, work_item_id)
            .await
    }

    pub async fn activate_hooks(
        self: &Arc<Self>,
        expected_epoch: i64,
        release: Release,
    ) -> Result<i64> {
        self.activate_hooks_internal(expected_epoch, release, false, None)
            .await
    }

    async fn activate_hooks_internal(
        self: &Arc<Self>,
        expected_epoch: i64,
        release: Release,
        validate_synthetic: bool,
        work_item_id: Option<String>,
    ) -> Result<i64> {
        ensure!(
            !self.stopped.load(Ordering::Acquire),
            "session has shut down"
        );
        // Verify release integrity from disk before anything else
        let rechecked = Release::load(&release.path).context("cannot reload release from disk")?;
        ensure!(
            rechecked.id == release.id,
            "release bytes have changed on disk: expected {}, got {}",
            release.id,
            rechecked.id
        );
        let permit = self
            .activation_gate
            .clone()
            .try_acquire_owned()
            .context("another hook activation is in progress")?;
        let this = self.clone();
        tokio::spawn(async move {
            let _permit = permit;
            let candidate = Policy::start(&this.config, this.id.clone(), release).await?;
            if validate_synthetic {
                if let Err(err) = candidate.validate_synthetic().await {
                    candidate.worker.stop().await;
                    return Err(err);
                }
            }
            let mut active = this.policy.write().await;
            ensure!(
                !this.stopped.load(Ordering::Acquire),
                "session has shut down"
            );
            if active.epoch != expected_epoch {
                candidate.worker.stop().await;
                bail!(
                    "activation conflict: expected epoch {expected_epoch}, current epoch {}",
                    active.epoch
                );
            }
            let epoch = match this
                .store
                .activate_with_work(
                    &this.id,
                    expected_epoch,
                    candidate.release.clone(),
                    work_item_id,
                )
                .await
            {
                Ok(ep) => ep,
                Err(err) => {
                    candidate.worker.stop().await;
                    return Err(err);
                }
            };
            *active = ActivePolicy {
                epoch,
                policy: candidate,
            };
            Ok::<_, anyhow::Error>(epoch)
        })
        .await
        .context("activation task panicked")?
    }

    /// The host task owns accepted execution, not the caller's connection.
    /// Dropping this future does not replay or abandon the cell. Interrupt through
    /// the independent stop path; every cell also has a host-owned deadline.
    pub async fn execute(
        self: &Arc<Self>,
        code: String,
        budget: Duration,
    ) -> Result<ExecutionReport> {
        self.execute_with_purpose(code, budget, Purpose::Interactive)
            .await
    }

    pub async fn execute_with_purpose(
        self: &Arc<Self>,
        code: String,
        budget: Duration,
        purpose: Purpose,
    ) -> Result<ExecutionReport> {
        ensure!(
            !self.stopped.load(Ordering::Acquire),
            "session has shut down"
        );
        ensure!(code.len() <= MAX_CODE, "cell is too large");
        ensure!(
            budget > Duration::ZERO && budget <= Duration::from_secs(3600),
            "invalid cell deadline"
        );
        ensure!(
            !self.storage_failed.load(Ordering::Acquire),
            "session paused after storage failure"
        );
        let permit = self
            .foreground
            .clone()
            .try_acquire_owned()
            .context("session already has a foreground cell")?;
        let this = self.clone();
        tokio::spawn(async move {
            let _permit = permit;
            this.execute_owned(code, budget, purpose).await
        })
        .await
        .context("execution task panicked")?
    }

    async fn execute_owned(
        self: Arc<Self>,
        code: String,
        budget: Duration,
        purpose: Purpose,
    ) -> Result<ExecutionReport> {
        let policy = self.policy.read().await.policy.clone();
        let placement = self.placement.read().await.placement.clone();
        let remote_lifecycle = self.lifecycle.read().await.clone();
        if let Some(lifecycle) = &remote_lifecycle {
            lifecycle.ensure_connected()?;
        }
        let worker = {
            let mut worker = self.workbench.lock().await;
            ensure!(
                !self.stopped.load(Ordering::Acquire),
                "session has shut down"
            );
            if !worker.is_alive() {
                if remote_lifecycle.is_some() {
                    // A remote generation is not a reconnectable heap. Refuse
                    // this cell until the caller explicitly opens a new owner.
                    bail!("remote connection is lost; reconnect explicitly before executing");
                }
                // The local adapter may retain its historical restart behavior.
                // This branch is never used for a remote placement.
                let (spawned, identity) =
                    spawn_workbench(&placement, &self.config, &self.id, &policy.release).await?;
                *worker = spawned;
                *self.placement.write().await = PlacementState {
                    placement: placement.clone(),
                    identity,
                };
                self.store
                    .append(
                        &self.id,
                        "kernel.reset",
                        json!({
                            "generation": worker.generation(),
                            "reason": "previous local worker ended; code was not replayed",
                            "skills": worker.skills_revision(),
                        }),
                    )
                    .await?;
            }
            worker.clone()
        };
        let id = OperationId::fresh();
        let generation = worker.generation().clone();
        let revision = policy.release.id.clone();
        self.store
            .begin(&self.id, &id, &generation, revision.clone(), code.clone())
            .await?;
        let service = self.service(
            generation.clone(),
            policy,
            placement.clone(),
            purpose,
            budget,
        );
        let outcome = match worker
            .call(
                WorkCommand::Execute {
                    id: id.clone(),
                    code,
                    purpose,
                },
                budget,
                service,
            )
            .await
        {
            Ok(outcome) => outcome,
            Err(error)
                if remote_lifecycle
                    .as_ref()
                    .is_some_and(|lifecycle| !lifecycle.is_connected())
                    || (remote_lifecycle.is_some() && !worker.is_alive()) =>
            {
                let reason = format!("remote transport lost before completion: {error:#}");
                if let Some(lifecycle) = &remote_lifecycle {
                    lifecycle.mark_lost(reason.clone())?;
                }
                // The operation may already have mutated the remote process.
                // Persist UNKNOWN and return without issuing a second call.
                if let Err(storage_error) = self.store.finish_unknown(&id, reason.clone()).await {
                    self.storage_failed.store(true, Ordering::Release);
                    return Err(storage_error
                        .context("cannot persist unknown remote completion; session paused"));
                }
                Outcome::Unknown { reason }
            }
            Err(error) => Outcome::Failed {
                error: format!("{error:#}"),
            },
        };
        if let Err(error) = self.store.finish(&id, outcome.clone()).await {
            self.storage_failed.store(true, Ordering::Release);
            worker.stop().await;
            return Err(error.context("cannot persist execution completion; session paused"));
        }
        Ok(ExecutionReport {
            id,
            generation,
            hook_revision: revision,
            outcome,
        })
    }

    fn service(
        self: &Arc<Self>,
        generation: Generation,
        policy: Arc<Policy>,
        placement: Placement,
        purpose: Purpose,
        budget: Duration,
    ) -> HostService {
        let this = self.clone();
        let deadline = tokio::time::Instant::now() + budget;
        let behavior_path = match &placement {
            Placement::Local => Some(policy.release.path.to_string_lossy().into_owned()),
            Placement::Remote { .. } => None,
        };
        Arc::new(move |call: HostCall| {
            let behavior_path = behavior_path.clone();
            let (this, generation, policy, target, binding) = (
                this.clone(),
                generation.clone(),
                policy.clone(),
                placement.kernel_target(),
                match &placement {
                    Placement::Remote { binding, .. } => Some(binding.clone()),
                    Placement::Local => None,
                },
            );
            Box::pin(async move {
                let result = async {
                    match call.request {
                        HostRequest::LocalSend { text } => {
                            let receipt = match this
                                .store
                                .prepare_send(
                                    &this.id,
                                    &call.execution_id,
                                    &generation,
                                    call.id,
                                    text.clone(),
                                )
                                .await?
                            {
                                PreparedEffect::Existing(receipt) => receipt,
                                PreparedEffect::New(effect) => {
                                    // No database transaction or session lock is
                                    // held while awaiting the separate callback lane.
                                    let context = match (&target, &binding) {
                                        (KernelTarget::Local, None) => CallbackContext::local(
                                            this.id.clone(),
                                            call.execution_id.clone(),
                                            generation.clone(),
                                            policy.release.id.clone(),
                                        ),
                                        (KernelTarget::Remote(_), Some(stamp)) => {
                                            CallbackContext::remote(
                                                this.id.clone(),
                                                call.execution_id.clone(),
                                                stamp.clone(),
                                                generation.clone(),
                                                policy.release.id.clone(),
                                            )
                                        }
                                        (KernelTarget::Local, Some(_)) => {
                                            return Err(anyhow!(
                                                "local callback cannot carry a remote binding"
                                            ));
                                        }
                                        (KernelTarget::Remote(remote), None) => {
                                            return Err(anyhow!(
                                                "remote target {} has no callback binding",
                                                remote.id
                                            ));
                                        }
                                    };
                                    let review =
                                        policy.review_with_context(context, &effect, &text).await;
                                    this.store
                                        .publish_send(
                                            &this.id,
                                            &call.execution_id,
                                            &generation,
                                            effect,
                                            policy.release.id.clone(),
                                            review,
                                        )
                                        .await?
                                }
                            };
                            Ok(serde_json::to_value(receipt)?)
                        }
                        HostRequest::Wait { seconds, reason } => {
                            this.store
                                .wait(&this.id, &call.execution_id, &generation, seconds, reason)
                                .await
                        }
                        HostRequest::HooksList => Ok(policy.list()),
                        HostRequest::HooksDescribe { slot } => Ok(policy.describe(slot)),
                        HostRequest::BehaviorActivate {
                            path,
                            expected_epoch,
                            revision,
                            work_item_id,
                        } => {
                            let release =
                                Release::load(&path).context("failed to load release from disk")?;
                            if let Some(expected_rev) = &revision {
                                ensure!(
                                    &release.id == expected_rev,
                                    "release revision mismatch: expected {}, got {}",
                                    expected_rev,
                                    release.id
                                );
                            }
                            let new_epoch = this
                                .activate_candidate_with_work(expected_epoch, release.clone(), work_item_id.clone())
                                .await?;
                            Ok(json!({
                                "epoch": new_epoch,
                                "revision": release.id,
                                "work_item_id": work_item_id,
                            }))
                        }
                        HostRequest::MaintenanceAssess {
                            work_item_id,
                            outcome,
                            reason,
                            missing_refs,
                            revisit_condition,
                            revisit_deadline_ms,
                            trigger_ms,
                            candidate_revision,
                            check_receipts,
                        } => {
                            // Closed outcome parse at the host boundary: an
                            // unknown tag, a missing variant field or a
                            // contradictory reference is a typed error to the
                            // caller, never a silently accepted payload.
                            let assessment_fields = json!({
                                "reason": reason,
                                "missing_refs": missing_refs,
                                "revisit_condition": revisit_condition,
                                "revisit_deadline_ms": revisit_deadline_ms,
                                "trigger_ms": trigger_ms,
                                "candidate_revision": candidate_revision,
                                "check_receipts": check_receipts,
                            });
                            let assessment =
                                MaintenanceAssessment::parse(&outcome, &assessment_fields)?;

                            let request_op_str = format!("{}_{}", call.execution_id, call.id);
                            let request_op = if request_op_str.len() <= 128
                                && request_op_str
                                    .bytes()
                                    .all(|b| b.is_ascii_alphanumeric() || b"_.-".contains(&b))
                            {
                                OperationId::try_from(request_op_str).map_err(|e| anyhow!(e))?
                            } else {
                                use sha2::{Digest, Sha256};
                                let mut hasher = Sha256::new();
                                hasher.update(request_op_str.as_bytes());
                                let hash = format!("{:x}", hasher.finalize());
                                OperationId::try_from(format!("req_{}", hash))
                                    .map_err(|e| anyhow!(e))?
                            };

                            let seq = this
                                .store
                                .commit_maintenance_assessment(
                                    &this.id,
                                    &request_op,
                                    &work_item_id,
                                    &outcome,
                                    assessment_fields,
                                )
                                .await?;

                            Ok(json!({
                                "status": "recorded",
                                "work_item_id": work_item_id,
                                "outcome": assessment.outcome_tag(),
                                "seq": seq,
                            }))
                        }
                        HostRequest::MaintenanceOriginate {
                            item_id,
                            purpose,
                            reason,
                            evidence_refs,
                            impact,
                        } => {
                            let now = crate::store::now_ms()?;
                            let work_id = item_id
                                .unwrap_or_else(|| format!("work-{}", call.id));
                            let work_item = WorkItem::new(
                                &work_id,
                                &purpose,
                                RunCause::MaintenanceReady {
                                    item_id: work_id.clone(),
                                    purpose: purpose.clone(),
                                },
                                ConsequenceMetadata::new(impact.unwrap_or(Impact::High)),
                                Provenance::agent("runtime.origination", now),
                                now,
                            )
                            .with_session(this.id.to_string())
                            .with_evidence_refs(evidence_refs.clone());

                            let origination_payload = json!({
                                "type": "maintenance.originated",
                                "id": work_id,
                                "purpose": purpose,
                                "reason": reason,
                                "evidence_refs": evidence_refs,
                                "impact": impact,
                            });

                            let request_op_str = format!("{}_{}", call.execution_id, call.id);
                            let request_op = if request_op_str.len() <= 128
                                && request_op_str
                                    .bytes()
                                    .all(|b| b.is_ascii_alphanumeric() || b"_.-".contains(&b))
                            {
                                OperationId::try_from(request_op_str).map_err(|e| anyhow!(e))?
                            } else {
                                use sha2::{Digest, Sha256};
                                let mut hasher = Sha256::new();
                                hasher.update(request_op_str.as_bytes());
                                let hash = format!("{:x}", hasher.finalize());
                                OperationId::try_from(format!("req_{}", hash))
                                    .map_err(|e| anyhow!(e))?
                            };

                            let (runnable_id, seq) = this
                                .store
                                .commit_origination_with_work(
                                    &this.id,
                                    &request_op,
                                    work_item,
                                    origination_payload,
                                )
                                .await?;

                            Ok(json!({
                                "item_id": runnable_id.as_str(),
                                "work_item_id": runnable_id.as_str(),
                                "seq": seq,
                            }))
                        }
                        HostRequest::EvidenceResolve { refs } => {
                            let mut resolved = Vec::new();
                            let mut unavailable = Vec::new();
                            for r in refs {
                                if let Some(rest) = r.strip_prefix("event:") {
                                    if let Ok(seq) = rest.parse::<i64>() {
                                        match this.store.get_event(&this.id, seq).await {
                                            Ok(Some(event)) => {
                                                let operation = event
                                                    .payload
                                                    .get("operation")
                                                    .and_then(|v| v.as_str())
                                                    .map(ToString::to_string);
                                                let signature = event
                                                    .payload
                                                    .get("signature")
                                                    .and_then(|v| v.as_str())
                                                    .map(ToString::to_string);
                                                let error = event
                                                    .payload
                                                    .get("error")
                                                    .and_then(|v| v.as_str())
                                                    .or_else(|| {
                                                        event
                                                            .payload
                                                            .get("outcome")
                                                            .and_then(|o| o.get("error"))
                                                            .and_then(|v| v.as_str())
                                                    })
                                                    .map(ToString::to_string);
                                                let active_epoch = this.hook_epoch().await;
                                                resolved.push(json!({
                                                    "ref": r,
                                                    "kind": "event",
                                                    "operation": operation,
                                                    "signature": signature,
                                                    "error": error,
                                                    "event": {
                                                        "id": event.id,
                                                        "kind": event.kind,
                                                        "payload": event.payload,
                                                        "created_ms": event.created_ms,
                                                    },
                                                    "current_lookup_context": {
                                                        "workspace_path": this.workspace().display().to_string(),
                                                        "session_id": this.id.to_string(),
                                                        "active_release_id": policy.release.id.clone(),
                                                        "active_epoch": active_epoch,
                                                    },
                                                    "workspace": {
                                                        "path": this.workspace().display().to_string(),
                                                        "session_id": this.id.to_string(),
                                                    },
                                                    "revision": {
                                                        "id": policy.release.id.clone(),
                                                        "epoch": active_epoch,
                                                    }
                                                }));
                                            }
                                            Ok(None) => {
                                                unavailable.push(json!({"ref": r, "reason": "not_found"}));
                                            }
                                            Err(e) => {
                                                unavailable.push(json!({"ref": r, "reason": format!("{e}")}));
                                            }
                                        }
                                    } else {
                                        unavailable.push(json!({"ref": r, "reason": "invalid_seq"}));
                                    }
                                } else if let Some(case_id) = r.strip_prefix("case:") {
                                    match this.store.get_observer_case(case_id).await {
                                        Ok(Some(case)) => {
                                            resolved.push(json!({
                                                "ref": r,
                                                "kind": "case",
                                                "operation": case.operation,
                                                "signature": case.signature,
                                                "component_revision": case.component_revision,
                                                "workspace": {
                                                    "path": case.workspace,
                                                    "session_id": case.session_id,
                                                },
                                                "count": case.count,
                                                "first_seen_ms": case.first_seen_ms,
                                                "last_seen_ms": case.last_seen_ms,
                                                "evidence_refs": case.evidence_refs,
                                                "work_item_id": case.work_item_id,
                                            }));
                                        }
                                        Ok(None) => {
                                            unavailable.push(json!({"ref": r, "reason": "not_found"}));
                                        }
                                        Err(e) => {
                                            unavailable.push(json!({"ref": r, "reason": format!("{e}")}));
                                        }
                                    }
                                } else if let Some(work_id) = r.strip_prefix("work:") {
                                    match this.store.get_work_item(work_id).await {
                                        Ok(Some(item)) => {
                                            resolved.push(json!({
                                                "ref": r,
                                                "kind": "work_item",
                                                "id": item.id,
                                                "session_id": item.session_id,
                                                "purpose": item.purpose,
                                                "cause": format!("{:?}", item.cause),
                                                "evidence_refs": item.provenance.source_refs,
                                                "status": format!("{:?}", item.status),
                                                "created_ms": item.created_ms,
                                            }));
                                        }
                                        Ok(None) => {
                                            unavailable.push(json!({"ref": r, "reason": "not_found"}));
                                        }
                                        Err(e) => {
                                            unavailable.push(json!({"ref": r, "reason": format!("{e}")}));
                                        }
                                    }
                                } else if let Some(rel_path) = r.strip_prefix("source:") {
                                    let path_obj = std::path::Path::new(rel_path);
                                    if path_obj.is_absolute()
                                        || path_obj
                                            .components()
                                            .any(|c| matches!(c, std::path::Component::ParentDir | std::path::Component::Prefix(_)))
                                    {
                                        unavailable.push(json!({
                                            "ref": r,
                                            "reason": "path escapes declared workspace root"
                                        }));
                                        continue;
                                    }
                                    let ws_root = match std::fs::canonicalize(this.workspace()) {
                                        Ok(p) => p,
                                        Err(_) => this.workspace().to_path_buf(),
                                    };
                                    let ws_candidate = this.workspace().join(rel_path);
                                    let rel_root = match std::fs::canonicalize(&policy.release.path) {
                                        Ok(p) => p,
                                        Err(_) => policy.release.path.clone(),
                                    };
                                    let rel_candidate = policy.release.path.join(rel_path);

                                    let candidate = if ws_candidate.is_file() {
                                        if let Ok(canon) = std::fs::canonicalize(&ws_candidate) {
                                            if canon.starts_with(&ws_root) {
                                                Some(canon)
                                            } else {
                                                None
                                            }
                                        } else {
                                            None
                                        }
                                    } else if rel_candidate.is_file() {
                                        if let Ok(canon) = std::fs::canonicalize(&rel_candidate) {
                                            if canon.starts_with(&rel_root) {
                                                Some(canon)
                                            } else {
                                                None
                                            }
                                        } else {
                                            None
                                        }
                                    } else {
                                        None
                                    };
                                    match candidate {
                                        Some(p) => match std::fs::read_to_string(&p) {
                                            Ok(content) => {
                                                resolved.push(json!({
                                                    "ref": r,
                                                    "kind": "source",
                                                    "path": rel_path,
                                                    "content": content,
                                                    "revision": policy.release.id.clone(),
                                                }));
                                            }
                                            Err(e) => {
                                                unavailable.push(json!({"ref": r, "reason": format!("read error: {e}")}));
                                            }
                                        },
                                        None => {
                                            unavailable.push(json!({"ref": r, "reason": "file not found in workspace or release"}));
                                        }
                                    }
                                } else {
                                    unavailable.push(json!({"ref": r, "reason": "unknown_ref_type"}));
                                }
                            }
                            Ok(json!({
                                "resolved": resolved,
                                "unavailable": unavailable,
                            }))
                        }
                        HostRequest::RuntimeInstructions => {
                            // The exact instruction text this transcript is running under, so a
                            // reader that received a patch does not have to apply it by hand
                            // and cannot silently apply it wrongly. A patch is a change; this
                            // is the state the change lands in.
                            let mut events = Vec::new();
                            let mut after = 0i64;
                            loop {
                                let page = this.store.history(&this.id, after, 1024).await?;
                                if page.is_empty() {
                                    break;
                                }
                                after = page.last().map(|event| event.id).unwrap_or(after);
                                let last = page.len();
                                events.extend(page);
                                if last < 1024 {
                                    break;
                                }
                            }
                            let known = crate::prompt_header::known_snapshot(
                                events
                                    .iter()
                                    .map(|event| (event.id, event.kind.as_str(), &event.payload)),
                            );
                            Ok(match known {
                                Some(snapshot) => json!({
                                    "resolved": true,
                                    "digest": snapshot.digest,
                                    "text": snapshot.text,
                                    "sources": snapshot
                                        .sources
                                        .iter()
                                        .map(|source| json!({
                                            "name": source.name,
                                            "digest": source.digest,
                                        }))
                                        .collect::<Vec<_>>(),
                                }),
                                None => json!({
                                    "resolved": false,
                                    "reason": "this transcript has no pinned instructions yet",
                                }),
                            })
                        }
                        HostRequest::SkillsIndex => Ok(this.skill_catalog().report()),
                        HostRequest::SkillsLoad { id } => {
                            let body = this.config.skills.load(&id)?;
                            Ok(body.report())
                        }
                        HostRequest::RuntimeInspect => {
                            let identity = this.target_identity().await;
                            // The header records are read here rather than passed in: the pin and
                            // the chain are durable facts about this transcript, and a value the
                            // caller had to remember to supply would be absent exactly when it
                            // matters most.
                            let mut header_events = Vec::new();
                            let mut after = 0i64;
                            loop {
                                let page = this.store.history(&this.id, after, 1024).await?;
                                if page.is_empty() {
                                    break;
                                }
                                after = page.last().map(|event| event.id).unwrap_or(after);
                                let last = page.len();
                                header_events.extend(page);
                                if last < 1024 {
                                    break;
                                }
                            }
                            let records = || {
                                header_events.iter().map(|event| {
                                    (event.id, event.kind.as_str(), &event.payload)
                                })
                            };
                            let head = crate::prompt_header::newest_pin(records());
                            let revisions_since_head = head
                                .as_ref()
                                .map(|(pin_seq, _)| {
                                    header_events
                                        .iter()
                                        .filter(|event| {
                                            event.kind == crate::prompt_header::HEADER_REVISION_EVENT
                                                && event.id > *pin_seq
                                        })
                                        .count()
                                })
                                .unwrap_or(0);
                            let known = crate::prompt_header::known_snapshot(records());
                            let selected = this.store.activation(&this.id).await?;
                            // The roots below are derived from THIS session's workspace
                            // and therefore follow the target: a peer or remote session
                            // reports its own tree, never the daemon's. Reported paths
                            // are real paths even when a directory has not been created
                            // yet, so a model does not have to walk the host to find out
                            // where it is or guess from the process cwd.
                            let workspace = this.workspace();
                            let draft_root = workspace.join(".klbr").join("drafts");
                            // Host-sourced time, so a claim about the world can be
                            // placed against the moment it was made rather than
                            // reconstructed from a transcript that carries no clock.
                            let now_ms = crate::store::now_ms().unwrap_or(0);
                            Ok(json!({
                                "session_id": this.id,
                                "execution_id": call.execution_id,
                                "purpose": purpose,
                                "target": identity,
                                "now_ms": now_ms,
                                // Same shape as the stamps the transcript carries, so the
                                // clock and an observation can be compared by eye.
                                "now": chrono::DateTime::from_timestamp_millis(now_ms)
                                    .map(|stamp| {
                                        stamp
                                            .to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
                                    })
                                    .unwrap_or_else(|| "unknown".into()),
                                "source_root": {
                                    "path": workspace.display().to_string(),
                                    "exists": workspace.is_dir(),
                                    "writable": false,
                                },
                                "draft_root": {
                                    "path": draft_root.display().to_string(),
                                    "target": identity.target,
                                    "exists": draft_root.is_dir(),
                                },
                                // The revision the worker is actually running under is
                                // the applied one; the store's selection is the chosen
                                // one. They are reported separately because a selection
                                // that has not activated is not in force.
                                "applied_behavior_revision": policy.release.id,
                                "pinned_behavior_revision": policy.release.id,
                                "selected_behavior_revision": selected.as_ref().map(|s| &s.revision),
                                "selected_behavior_epoch": selected.as_ref().map(|s| s.epoch),
                                "behavior_source": behavior_path.map(|path| json!({"target": identity.target, "path": path, "writable": false})),
                                "skills": this.skill_roots_report(),
                                // Which instructions this transcript is running under, from its
                                // own records. The head is a pinned revision and the newest text
                                // the transcript can reconstruct, so a reader can name the head it
                                // ran under without hashing a context it cannot reproduce byte for
                                // byte, and can see whether a change has been observed since.
                                "instructions": {
                                    "head_digest": head.as_ref().map(|(_, pin)| pin.digest.clone()),
                                    "head_pinned_at_ms": head.as_ref().map(|(_, pin)| pin.pinned_at_ms),
                                    "head_reason": head.as_ref().map(|(_, pin)| pin.reason.as_str()),
                                    "revisions_since_head": revisions_since_head,
                                    "known_digest": known.as_ref().map(|snapshot| snapshot.digest.clone()),
                                },
                                "cell_budget": {"limit_ms": budget.as_millis() as u64,
                                    "remaining_ms": deadline.saturating_duration_since(tokio::time::Instant::now()).as_millis() as u64},
                                "unavailable": {
                                    "agent_id": "session-to-agent association not configured",
                                    "attempt_id": "model attempt context not supplied to workbench",
                                    "remote_behavior_source": "remote bundle has no verified mapping to the pinned host policy revision",
                                    "cumulative_model_budget": "model budget accounting not supplied to workbench"
                                }
                            }))
                        }
                        HostRequest::WorkspaceRun {
                            argv,
                            cwd,
                            env,
                            stdin_base64,
                            timeout_ms,
                            max_output_bytes,
                        } => {
                            ensure!(
                                matches!(target, KernelTarget::Local),
                                "workspace.run has no host executor for a remote placement"
                            );
                            run_workspace_command(
                                this.workspace(),
                                argv,
                                cwd,
                                env,
                                stdin_base64,
                                timeout_ms,
                                max_output_bytes,
                            )
                            .await
                        }
                        HostRequest::WorkspaceInfo => {
                            let epoch = this.hook_epoch().await;
                            let identity = this.target_identity().await;
                            Ok(json!({
                                "path": this.workspace().display().to_string(),
                                "session_id": this.id.to_string(),
                                "active_revision": policy.release.id.clone(),
                                "epoch": epoch,
                                "target": serde_json::to_value(&identity)?,
                            }))
                        }
                        HostRequest::OrientationStage(req) => {
                            let host = this.orientation_host.read().unwrap().clone()
                                .context("unconfigured_orientation_host")?;
                            host.stage_orientation(&this.store, &this.id, req).await
                        }
                        HostRequest::OrientationCurrent(req) => {
                            let host = this.orientation_host.read().unwrap().clone()
                                .context("unconfigured_orientation_host")?;
                            host.current_orientation(&this.store, &this.id, req).await
                        }
                        HostRequest::OrientationAdopt(req) => {
                            let host = this.orientation_host.read().unwrap().clone()
                                .context("unconfigured_orientation_host")?;
                            host.adopt_orientation(&this.store, &this.id, req).await
                        }
                        HostRequest::FrontierSelect(req) => {
                            let host = this.orientation_host.read().unwrap().clone()
                                .context("unconfigured_orientation_host")?;
                            host.select_frontier(&this.store, &this.id, req).await
                        }
                        HostRequest::PeerOpen {
                            peer_id,
                            mode,
                            profile,
                            workspace,
                            kernel_target,
                            budget,
                            branch_parent_revision,
                            evidence,
                        } => {
                            let host_opt = this.peer_host.read().unwrap().clone();
                            let Some(host) = host_opt else {
                                bail!("unconfigured_peer_host: peer host not configured for session {}", this.id);
                            };
                            let req = PeerOpenReq {
                                peer_id,
                                mode,
                                profile,
                                workspace,
                                kernel_target,
                                budget,
                                branch_parent_revision,
                                evidence,
                            };
                            host.open_peer(&this.store, &this.id, req).await
                        }
                        HostRequest::PeerSend {
                            peer_id,
                            request_id,
                            content,
                            evidence,
                        } => {
                            let host_opt = this.peer_host.read().unwrap().clone();
                            let Some(host) = host_opt else {
                                bail!("unconfigured_peer_host: peer host not configured for session {}", this.id);
                            };
                            let req = PeerSendReq {
                                peer_id,
                                request_id,
                                content,
                                evidence,
                            };
                            host.send_peer_message(&this.store, &this.id, req).await
                        }
                        HostRequest::PeerResult {
                            request_id,
                            peer_id,
                            wait_ms,
                        } => {
                            let host_opt = this.peer_host.read().unwrap().clone();
                            let Some(host) = host_opt else {
                                bail!("unconfigured_peer_host: peer host not configured for session {}", this.id);
                            };
                            let req = PeerResultReq {
                                request_id,
                                peer_id,
                                wait_ms,
                            };
                            host.get_peer_result(&this.store, &this.id, req).await
                        }
                        HostRequest::PeerSubmitResult {
                            request_id,
                            peer_id,
                            result,
                        } => {
                            let host_opt = this.peer_host.read().unwrap().clone();
                            let Some(host) = host_opt else {
                                bail!("unconfigured_peer_host: peer host not configured for session {}", this.id);
                            };
                            let req = PeerSubmitResultReq {
                                request_id,
                                peer_id,
                                result,
                            };
                            host.submit_peer_result(&this.store, &this.id, req).await
                        }
                        HostRequest::PeerCancel {
                            request_id,
                            peer_id,
                        } => {
                            let host_opt = this.peer_host.read().unwrap().clone();
                            let Some(host) = host_opt else {
                                bail!("unconfigured_peer_host: peer host not configured for session {}", this.id);
                            };
                            let req = PeerCancelReq {
                                request_id,
                                peer_id,
                            };
                            host.cancel_peer_request(&this.store, &this.id, req).await
                        }
                        HostRequest::PeerGet { peer_id } => {
                            let host_opt = this.peer_host.read().unwrap().clone();
                            let Some(host) = host_opt else {
                                bail!("unconfigured_peer_host: peer host not configured for session {}", this.id);
                            };
                            let req = PeerGetReq { peer_id };
                            host.get_peer(&this.store, &this.id, req).await
                        }
                        HostRequest::PeerList => {
                            let host_opt = this.peer_host.read().unwrap().clone();
                            let Some(host) = host_opt else {
                                bail!("unconfigured_peer_host: peer host not configured for session {}", this.id);
                            };
                            host.list_peers(&this.store, &this.id).await
                        }
                        HostRequest::PeerStop { peer_id, drain } => {
                            let host_opt = this.peer_host.read().unwrap().clone();
                            let Some(host) = host_opt else {
                                bail!("unconfigured_peer_host: peer host not configured for session {}", this.id);
                            };
                            let req = PeerStopReq { peer_id, drain };
                            host.stop_peer(&this.store, &this.id, req).await
                        }
                        request @ (HostRequest::RemoteEnsure { .. }
                        | HostRequest::RemoteCall { .. }
                        | HostRequest::RemoteCancel { .. }
                        | HostRequest::RemoteClose { .. }) => {
                            match this
                                .remote
                                .handle(&this.id, &call.execution_id, request)
                                .await
                            {
                                HostReply::Ok { value } => Ok(value),
                                HostReply::Error { code, message } => {
                                    Err(RoutedHostError { code, message }.into())
                                }
                            }
                        }
                    }
                }
                .await;
                match result {
                    Ok(value) => HostReply::Ok { value },
                    Err(error) => {
                        if let Some(routed) = error.downcast_ref::<RoutedHostError>() {
                            return HostReply::error(&routed.code, routed.message.clone());
                        }
                        let msg = format!("{error:#}");
                        let code = if msg.contains("activation conflict") {
                            "cas_conflict"
                        } else if msg.contains("validation")
                            || msg.contains("bytes have changed")
                            || msg.contains("revision mismatch")
                        {
                            "validation_error"
                        } else {
                            "host_operation_failed"
                        };
                        HostReply::error(code, msg)
                    }
                }
            })
        })
    }

    /// Management bypasses policy hooks. A broken hook cannot veto its repair.
    pub async fn interrupt(&self) {
        let worker = self.workbench.lock().await.clone();
        worker.stop().await;
        if let Some(lifecycle) = self.lifecycle.read().await.clone() {
            let _ = lifecycle.mark_lost("connection interrupted by host");
        }
    }

    /// Wait until any active foreground cell completes execution.
    pub async fn wait_for_idle(&self) {
        if let Ok(_permit) = self.foreground.acquire().await {
            // Permit successfully acquired and dropped: in-flight cell finished.
        }
    }

    /// The resolved target identity for the CURRENT workbench generation.
    /// Resolved once per generation; consumers must not re-derive per prompt.
    pub async fn target_identity(&self) -> TargetIdentitySnapshot {
        self.placement.read().await.identity.clone()
    }

    /// Change the session's placement. This is a VISIBLE NEW GENERATION with
    /// an explicit transition: the old workbench is stopped (an in-flight cell
    /// is interrupted honestly), the namespace is NOT migrated, and no global
    /// moves. Refused while a foreground cell is running.
    pub async fn relocate(self: &Arc<Self>, placement: Placement) -> Result<Generation> {
        ensure!(
            !self.stopped.load(Ordering::Acquire),
            "session has shut down"
        );
        let permit = self
            .foreground
            .clone()
            .try_acquire_owned()
            .context("foreground cell is running; placement change refused")?;
        let policy = self.policy.read().await.policy.clone();
        // Reconnect and relocation invalidate every auxiliary namespace owned by
        // the old connection before a fresh endpoint is admitted.
        self.remote.shutdown().await?;
        let next_lifecycle = lifecycle_for_placement(&placement)?;
        let old_worker = { self.workbench.lock().await.clone() };
        let old_generation = old_worker.generation().clone();
        let old_identity = self.target_identity().await;
        old_worker.stop().await;
        if let Some(old_lifecycle) = self.lifecycle.read().await.clone() {
            old_lifecycle.mark_lost("placement changed; old connection closed")?;
        }
        let (worker, identity) =
            spawn_workbench(&placement, &self.config, &self.id, &policy.release).await?;
        if let Placement::Remote {
            target, backend, ..
        } = &placement
        {
            self.remote.register_relay_target(
                target.clone(),
                backend.clone(),
                self.config.startup_timeout,
            )?;
        }
        let generation = worker.generation().clone();
        self.store
            .append(
                &self.id,
                "kernel.placement",
                json!({
                    "generation": generation.as_str(),
                    "previous_generation": old_generation.as_str(),
                    "from": {"kind": old_identity.kind, "target": old_identity.target,
                             "generation": old_identity.generation},
                    "to": {"kind": identity.kind, "target": identity.target,
                           "generation": identity.generation},
                    "reason": "placement changed; namespace did not migrate; globals did not move",
                }),
            )
            .await?;
        self.store
            .set_session_placement(&self.id, &placement.descriptor(&generation)?)
            .await
            .context("failed to persist placement change")?;
        *self.workbench.lock().await = worker;
        *self.placement.write().await = PlacementState {
            placement,
            identity,
        };
        *self.lifecycle.write().await = next_lifecycle;
        drop(permit);
        Ok(generation)
    }

    /// Explicitly opens a fresh connection-scoped remote namespace. The old
    /// owner remains lost, and no operation from it is replayed.
    pub async fn reconnect(self: &Arc<Self>) -> Result<Generation> {
        ensure!(
            !self.stopped.load(Ordering::Acquire),
            "session has shut down"
        );
        let old_lifecycle = self
            .lifecycle
            .read()
            .await
            .clone()
            .context("local sessions do not have a remote connection")?;
        let next = old_lifecycle.reconnect()?;
        let current = self.placement.read().await.placement.clone();
        let (target, backend, behavior_bundle) = match current {
            Placement::Remote {
                target,
                backend,
                behavior_bundle,
                ..
            } => (target, backend, behavior_bundle),
            Placement::Local => bail!("remote connection-scoped placement is unavailable"),
        };
        let placement = Placement::remote(
            target,
            backend,
            next.namespace_id().clone(),
            behavior_bundle,
        );
        self.relocate(placement).await
    }

    /// Runs the explicitly diagnosed production bootstrap repair and then
    /// returns ordinary execution to a fresh connection-scoped kernel.
    pub async fn repair_bootstrap(
        self: &Arc<Self>,
        repair: BootstrapRepair,
        request: RepairRequest,
    ) -> Result<RepairReceipt> {
        ensure!(
            !self.stopped.load(Ordering::Acquire),
            "session has shut down"
        );
        let lifecycle = self
            .lifecycle
            .read()
            .await
            .clone()
            .context("local sessions do not have a remote bootstrap repair lane")?;
        lifecycle.mark_lost("explicit bootstrap repair closed the old owner")?;
        let receipt = repair.repair(&self.store, &self.id, request).await?;
        // The repair action is complete; ordinary work resumes only through a
        // newly admitted namespace, never through the repair transport.
        self.reconnect().await?;
        Ok(receipt)
    }

    pub async fn restart_workbench(&self) -> Result<()> {
        let worker = self.workbench.lock().await.clone();
        if worker.is_alive() {
            worker.stop().await;
        }
        if let Some(lifecycle) = self.lifecycle.read().await.clone() {
            lifecycle.mark_lost("connection closed by host")?;
        }
        Ok(())
    }

    pub async fn shutdown(&self) {
        self.stopped.store(true, Ordering::Release);
        let _ = self.remote.shutdown().await;
        self.interrupt().await;
        let policy = self.policy.read().await.policy.clone();
        policy.worker.stop().await;
    }
}