brokk-mj-controller 2.23.2

use super::*;

const DEFERRED_SUBMIT_CAPACITY: usize = 32;

/// Whether a session's durable state means its relay actor should stop.
///
/// `Closing` and `Destroying` are still in flight and their owners need the
/// actor. The rest have ended: nothing will ever answer on this session's
/// control socket again.
fn terminal_for_reconnect(state: mj_core::state::SessionState) -> bool {
    use mj_core::state::SessionState;
    matches!(
        state,
        SessionState::Error
            | SessionState::Lost
            | SessionState::Stopped
            | SessionState::Parked
            | SessionState::DestroyedWithDataLoss
    )
}

/// This session's durable state and stored cause, read off the actor's task.
///
/// Read only when the relay has already failed repeatedly, so the database
/// work is rare, and on the blocking pool, because the actor's own task also
/// serves this session's view.
///
/// A record that is gone answers `Lost`: the controller discards the record of
/// a session whose managed target no longer exists, and an actor that read
/// that as "no answer yet" would reconnect forever. Only a store that could
/// not be read leaves the question open.
async fn durable_session_outcome(
    session_id: &str,
) -> Option<(mj_core::state::SessionState, Option<String>)> {
    let session_id = session_id.to_owned();
    tokio::task::spawn_blocking(move || {
        let Some(record) = crate::database::load_session_record(&session_id).ok()? else {
            return Some((mj_core::state::SessionState::Lost, None));
        };
        Some((record.state, record.last_error))
    })
    .await
    .ok()
    .flatten()
}

pub(super) async fn run_session_actor(
    target: RelaySessionTarget,
    mut commands: mpsc::Receiver<ActorCommand>,
    mut releases: mpsc::UnboundedReceiver<ReturnedConnection>,
    mut retirement: watch::Receiver<bool>,
    view_tx: watch::Sender<ManagedSessionView>,
    updates: CoalescedUpdateSender,
) {
    let mut connection: Option<StandaloneSession> = None;
    let mut failures = 0_u32;
    let mut last_recovery_probe = None;
    let mut lifecycle = ActorLifecycle::default();
    let mut deferred_submits: VecDeque<DeferredSubmit> = VecDeque::new();
    let mut next_lease_id = 1_u64;
    let mut reviewer_tasks = tokio::task::JoinSet::new();
    let mut reviewer_connections = BTreeMap::new();
    let mut reviewer_tails = BTreeMap::new();
    let mut reviewer_cancellation = tokio_util::sync::CancellationToken::new();
    let mut interval = tokio::time::interval(SESSION_SYNC_INTERVAL);
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    enum Event {
        Returned(Option<Box<ReturnedConnection>>),
        Reviewer(Option<std::result::Result<(), tokio::task::JoinError>>),
        Tick,
        Command(Option<ActorCommand>),
        Retirement(std::result::Result<(), watch::error::RecvError>),
    }
    loop {
        lifecycle.set_retirement_requested(*retirement.borrow_and_update());
        if lifecycle.should_stop() {
            break;
        }
        let event = tokio::select! {
            // Release and command use separate channels. Apply a queued return
            // before admitting the next lease, even when both are ready.
            biased;
            returned = releases.recv(), if !releases.is_closed() || !releases.is_empty() => Event::Returned(returned.map(Box::new)),
            // Keep the remaining work fair: a slow sync must not starve commands.
            event = async {
                tokio::select! {
                    completed = reviewer_tasks.join_next(), if !reviewer_tasks.is_empty() => Event::Reviewer(completed),
                    _ = interval.tick() => Event::Tick,
                    command = commands.recv() => Event::Command(command),
                    changed = retirement.changed() => Event::Retirement(changed),
                }
            } => event,
        };
        match event {
            Event::Returned(returned) => {
                let Some(returned) = returned else { continue };
                if lifecycle.return_lease(returned.lease_id) {
                    // A dropped lease returns no connection; `submit_actor_command`
                    // reconnects on demand, so the drain needs no special case.
                    connection = returned.connection;
                    failures = 0;
                    interval.reset();
                    // A lease syncs the connection it borrowed, so this actor's
                    // next sync can find nothing left to apply. Publish what the
                    // returned connection already knows or watchers keep reading
                    // pre-lease state.
                    if let Some(returned) = connection.as_ref() {
                        publish_view(
                            &target.session_id,
                            ManagedSessionView {
                                snapshot: Some(returned.snapshot()),
                                connected: true,
                                error: None,
                            },
                            &view_tx,
                            &updates,
                        );
                    }
                    let retiring = *retirement.borrow();
                    while let Some(deferred) = deferred_submits.pop_front() {
                        if retiring {
                            if deferred
                                .reply
                                .send(Err("session target is changing".into()))
                                .is_err()
                            {
                                tracing::debug!(
                                    session_id = %target.session_id,
                                    operation = "submit",
                                    "deferred submit rejection receiver was already closed"
                                );
                            }
                            continue;
                        }
                        deliver_submit(&target, &mut connection, deferred, &view_tx, &updates)
                            .await;
                    }
                }
            }
            Event::Reviewer(completed) => {
                if let Some(Err(error)) = completed {
                    tracing::error!(session_id = %target.session_id, %error, "reviewer operation task failed");
                }
            }
            Event::Tick => {
                lifecycle.set_retirement_requested(*retirement.borrow());
                if lifecycle.should_stop() {
                    break;
                }
                if lifecycle.is_leased() {
                    continue;
                }
                let result = sync_actor_connection(&target, &mut connection).await;
                match result {
                    Ok(snapshot) => {
                        failures = 0;
                        if let Some(snapshot) = snapshot {
                            publish_view(
                                &target.session_id,
                                ManagedSessionView {
                                    snapshot: Some(snapshot),
                                    connected: true,
                                    error: None,
                                },
                                &view_tx,
                                &updates,
                            );
                        }
                    }
                    Err(error) => {
                        connection = None;
                        failures = failures.saturating_add(1);
                        // A projection integrity failure repeats on every
                        // retry, so report it at once rather than waiting for
                        // the unreachable threshold.
                        let integrity = projection_integrity_failure(&error);
                        tracing::warn!(
                            session_id = target.session_id,
                            consecutive_failures = failures,
                            projection_integrity = integrity,
                            transport_dead = worker_connect_needs_restart(&error),
                            "session relay sync failed: {error:#}"
                        );
                        let recovery_due = !integrity
                            && !crate::controller::move_session::move_owns_session(
                                &target.session_id,
                            )
                            && failures >= UNREACHABLE_FAILURE_THRESHOLD
                            && worker_connect_needs_restart(&error)
                            && target.worker_recovery.is_some()
                            && last_recovery_probe.is_none_or(|last: tokio::time::Instant| {
                                last.elapsed() >= WORKER_RESTART_COOLDOWN
                            });
                        if integrity || failures >= UNREACHABLE_FAILURE_THRESHOLD {
                            // Bind the clone first: borrowing inside the call
                            // would hold the watch read guard while
                            // `publish_view` takes the write lock, deadlocking
                            // this actor on its own view.
                            let snapshot = view_tx.borrow().snapshot.clone();
                            let mut detail = format!("{error:#}");
                            if recovery_due {
                                detail.push_str("; checking whether the relay worker is dead");
                            }
                            publish_view(
                                &target.session_id,
                                ManagedSessionView {
                                    snapshot,
                                    connected: false,
                                    error: Some(if integrity {
                                        ViewError::ProjectionIntegrity(detail)
                                    } else {
                                        ViewError::Unreachable(detail)
                                    }),
                                },
                                &view_tx,
                                &updates,
                            );
                        }
                        // A session whose record has reached a terminal state
                        // is never coming back on this actor. Without this the
                        // actor reconnects to a socket that will never exist
                        // for as long as the daemon runs, logging a failure
                        // every backoff period (#1078).
                        if failures >= UNREACHABLE_FAILURE_THRESHOLD
                            && let Some((state, last_error)) =
                                durable_session_outcome(&target.session_id).await
                            && terminal_for_reconnect(state)
                        {
                            tracing::info!(
                                session_id = target.session_id,
                                ?state,
                                "session reached a terminal state; retiring its relay actor"
                            );
                            // A parked sub-agent stopped its worker on
                            // purpose; it is not unreachable.
                            if state == mj_core::state::SessionState::Parked {
                                break;
                            }
                            let snapshot = view_tx.borrow().snapshot.clone();
                            publish_view(
                                &target.session_id,
                                ManagedSessionView {
                                    snapshot,
                                    connected: false,
                                    error: Some(ViewError::Unreachable(match last_error {
                                        Some(cause) => {
                                            format!("this session ended as {state:?}: {cause}")
                                        }
                                        None => format!("this session ended as {state:?}"),
                                    })),
                                },
                                &view_tx,
                                &updates,
                            );
                            break;
                        }
                        if recovery_due {
                            last_recovery_probe = Some(tokio::time::Instant::now());
                            let plan = target
                                .worker_recovery
                                .clone()
                                .expect("recovery eligibility requires a plan");
                            let restart_unresponsive = worker_connect_allows_live_restart(&error);
                            tracing::warn!(
                                session_id = target.session_id,
                                "relay worker is unreachable; probing it before recovery: {error:#}"
                            );
                            match recover_worker_for_session(
                                plan,
                                restart_unresponsive,
                                Some(target.session_id.clone()),
                            )
                            .await
                            {
                                Ok(
                                    outcome @ (WorkerRecoveryOutcome::RestartedDead
                                    | WorkerRecoveryOutcome::RestartedUnresponsive),
                                ) => {
                                    failures = 0;
                                    let snapshot = view_tx.borrow().snapshot.clone();
                                    let recovery = match outcome {
                                        WorkerRecoveryOutcome::RestartedDead => {
                                            "confirmed the relay worker was dead and restarted it"
                                        }
                                        WorkerRecoveryOutcome::RestartedUnresponsive => {
                                            "the relay worker was alive but not serving handshakes, so it was restarted"
                                        }
                                        WorkerRecoveryOutcome::Alive
                                        | WorkerRecoveryOutcome::Starting
                                        | WorkerRecoveryOutcome::TargetMissing
                                        | WorkerRecoveryOutcome::Suppressed
                                        | WorkerRecoveryOutcome::WorkspaceMissing(_) => {
                                            unreachable!()
                                        }
                                    };
                                    publish_view(
                                        &target.session_id,
                                        ManagedSessionView {
                                            snapshot,
                                            connected: false,
                                            error: Some(ViewError::Unreachable(format!(
                                                "{error:#}; {recovery}"
                                            ))),
                                        },
                                        &view_tx,
                                        &updates,
                                    );
                                    interval.reset_after(RECONNECT_INTERVAL);
                                }
                                Ok(WorkerRecoveryOutcome::Alive) => {
                                    tracing::warn!(
                                        session_id = target.session_id,
                                        "relay transport failed but the worker is alive; leaving it running"
                                    );
                                    let snapshot = view_tx.borrow().snapshot.clone();
                                    publish_view(
                                        &target.session_id,
                                        ManagedSessionView {
                                            snapshot,
                                            connected: false,
                                            error: Some(ViewError::Unreachable(format!(
                                                "{error:#}; relay worker is still alive, so it was not restarted"
                                            ))),
                                        },
                                        &view_tx,
                                        &updates,
                                    );
                                    interval.reset_after(reconnect_delay(failures));
                                }
                                Ok(WorkerRecoveryOutcome::Starting) => {
                                    tracing::warn!(
                                        session_id = target.session_id,
                                        "relay worker is still starting; leaving it running"
                                    );
                                    let snapshot = view_tx.borrow().snapshot.clone();
                                    publish_view(
                                        &target.session_id,
                                        ManagedSessionView {
                                            snapshot,
                                            connected: false,
                                            error: Some(ViewError::Unreachable(format!(
                                                "{error:#}; relay worker is still recovering its durable state, so it was not restarted"
                                            ))),
                                        },
                                        &view_tx,
                                        &updates,
                                    );
                                    interval.reset_after(reconnect_delay(failures));
                                }
                                Ok(WorkerRecoveryOutcome::Suppressed) => {
                                    tracing::info!(
                                        session_id = target.session_id,
                                        "automatic worker recovery suppressed by durable lifecycle or target change"
                                    );
                                    // The desired-target refresher will remove or replace this
                                    // stale actor. It must not reconnect in the meantime.
                                    break;
                                }
                                Ok(WorkerRecoveryOutcome::TargetMissing) => {
                                    let snapshot = view_tx.borrow().snapshot.clone();
                                    publish_view(&target.session_id, ManagedSessionView {
                                        snapshot,
                                        connected: false,
                                        error: Some(ViewError::TargetMissing(
                                            "the managed Podman session container no longer exists"
                                                .into(),
                                        )),
                                    }, &view_tx, &updates);
                                    interval.reset_after(RECONNECT_BACKOFF_CEILING);
                                }
                                Ok(WorkerRecoveryOutcome::WorkspaceMissing(directory)) => {
                                    let snapshot = view_tx.borrow().snapshot.clone();
                                    publish_view(
                                        &target.session_id,
                                        ManagedSessionView {
                                            snapshot,
                                            connected: false,
                                            error: Some(ViewError::TargetMissing(format!(
                                                "the worker working directory {} is missing; resume this session from its recovery archive to restore it",
                                                directory.display(),
                                            ))),
                                        },
                                        &view_tx,
                                        &updates,
                                    );
                                    interval.reset_after(RECONNECT_BACKOFF_CEILING);
                                }
                                Err(recovery_error) => {
                                    tracing::warn!(
                                        session_id = target.session_id,
                                        "automatic relay worker recovery failed safely: {recovery_error:#}"
                                    );
                                    let snapshot = view_tx.borrow().snapshot.clone();
                                    publish_view(
                                        &target.session_id,
                                        ManagedSessionView {
                                            snapshot,
                                            connected: false,
                                            error: Some(ViewError::Unreachable(format!(
                                                "{error:#}; could not confirm the relay worker was dead, so it was not restarted: {recovery_error:#}"
                                            ))),
                                        },
                                        &view_tx,
                                        &updates,
                                    );
                                    interval.reset_after(reconnect_delay(failures));
                                }
                            }
                        } else {
                            interval.reset_after(reconnect_delay(failures));
                        }
                    }
                }
            }
            Event::Command(command) => {
                let Some(command) = command else { break };
                lifecycle.set_retirement_requested(*retirement.borrow());
                if !lifecycle.accepts_new_work() {
                    tracing::debug!(
                        session_id = %target.session_id,
                        operation = command.operation_name(),
                        "rejecting relay operation while session target changes"
                    );
                    command.reject(&target.session_id, "session target is changing");
                    continue;
                }
                match command {
                    ActorCommand::Submit {
                        durable,
                        queued_at,
                        command_id,
                        command,
                        admission,
                        reply,
                    } => {
                        if crate::controller::move_session::move_refuses_command(
                            &target.session_id,
                            &command,
                        ) {
                            let _ = reply.send(Err(
                                "session is moving; keep the draft and retry after Move finishes"
                                    .into(),
                            ));
                            continue;
                        }
                        // A turn under review holds its session's prompts. The
                        // sole exception is a capability issued by the review
                        // host for this exact corrective command; ordinary
                        // prompts and controller-authored notices still take
                        // the refusal path below.
                        let admitted = admission.as_ref().is_some_and(|admission| {
                            matches!(&command, RelayCommand::Prompt { .. })
                                && admission.command_id() == command_id
                                && crate::review_host::review_delivery_admitted(
                                    &target.session_id,
                                    admission,
                                )
                        });
                        if admission.is_some() && !admitted {
                            let _ = reply
                                .send(Err("review delivery admission is no longer valid".into()));
                            continue;
                        }
                        if matches!(
                            &command,
                            RelayCommand::Prompt { .. }
                                | RelayCommand::ContinueAuthorizedWork { .. }
                                | RelayCommand::ResumeAfterQuota { .. }
                                | RelayCommand::ClearContext
                        ) && !admitted
                            && let Some(refusal) =
                                crate::review_host::prompt_refusal(&target.session_id)
                        {
                            tracing::debug!(
                                session_id = %target.session_id,
                                %command_id,
                                "refusing a prompt while a turn review is unresolved"
                            );
                            let _ = reply.send(Err(refusal.to_owned().into()));
                            continue;
                        }
                        if lifecycle.is_leased()
                            && (matches!(command, RelayCommand::ClearContext)
                                || matches!(&command, RelayCommand::Prompt { prompt } if matches!(mj_core::acp::context_command(prompt), Some((mj_core::acp::ContextCommand::Clear, _)))))
                        {
                            let _ = reply.send(Err(
                                "/clear requires an idle session; a lifecycle operation is running"
                                    .into(),
                            ));
                            continue;
                        }
                        if lifecycle.is_leased()
                            && matches!(
                                command,
                                RelayCommand::ContinueAuthorizedWork { .. }
                                    | RelayCommand::ResumeAfterQuota { .. }
                            )
                        {
                            let _ = reply.send(Err(
                                "automatic continuation cancelled by lifecycle operation".into(),
                            ));
                            continue;
                        }
                        if lifecycle.is_leased() {
                            // A checkpoint or other lifecycle operation owns the
                            // connection. Hold the prompt instead of rejecting it
                            // and deliver it when the lease comes back.
                            if deferred_submits.len() >= DEFERRED_SUBMIT_CAPACITY {
                                let _ = reply.send(Err(
                                    "session lifecycle queue is full; request was not accepted"
                                        .into(),
                                ));
                                continue;
                            }
                            deferred_submits.push_back(DeferredSubmit {
                                durable,
                                queued_at,
                                command_id,
                                command,
                                admission,
                                reply,
                            });
                            continue;
                        }
                        deliver_submit(
                            &target,
                            &mut connection,
                            DeferredSubmit {
                                durable,
                                queued_at,
                                command_id,
                                command,
                                admission,
                                reply,
                            },
                            &view_tx,
                            &updates,
                        )
                        .await;
                    }
                    ActorCommand::CommandReceipt {
                        command_id,
                        action,
                        reply,
                    } => {
                        let result = async {
                            // A leased actor may still deliver an earlier queued
                            // submit. Neither absence nor release is safe until
                            // its return has drained those deferred commands.
                            ensure!(
                                !lifecycle.is_leased(),
                                "session is reserved for a lifecycle operation"
                            );
                            if connection.is_none() {
                                sync_actor_connection(&target, &mut connection).await?;
                            }
                            let client = connection.as_mut().context("relay is disconnected")?;
                            match action {
                                CommandReceiptAction::Release => {
                                    client.release_command_receipt(command_id).await?;
                                    Ok(None)
                                }
                                CommandReceiptAction::Lookup => {
                                    client.command_receipt(command_id).await
                                }
                                CommandReceiptAction::CancelAdmission => {
                                    client.cancel_command_admission(command_id).await
                                }
                            }
                        }
                        .await;
                        if result
                            .as_ref()
                            .is_err_and(|error| !is_final_rejection(error))
                            && !lifecycle.is_leased()
                        {
                            connection = None;
                        }
                        let _ = reply.send(result);
                    }
                    ActorCommand::Sync { reply } => {
                        if lifecycle.is_leased() {
                            tracing::debug!(
                                session_id = %target.session_id,
                                operation = "sync",
                                "rejecting sync while session is leased"
                            );
                            if reply
                                .send(Err("session is reserved for a lifecycle operation".into()))
                                .is_err()
                            {
                                tracing::debug!(
                                    session_id = %target.session_id,
                                    operation = "sync",
                                    "sync rejection receiver was already closed"
                                );
                            }
                            continue;
                        }
                        let result =
                            sync_actor_connection(&target, &mut connection)
                                .await
                                .map(|snapshot| {
                                    if let Some(snapshot) = snapshot {
                                        publish_view(
                                            &target.session_id,
                                            ManagedSessionView {
                                                snapshot: Some(snapshot),
                                                connected: true,
                                                error: None,
                                            },
                                            &view_tx,
                                            &updates,
                                        );
                                    }
                                });
                        if result.is_err() {
                            connection = None;
                        }
                        if let Err(error) = &result {
                            tracing::warn!(
                                session_id = %target.session_id,
                                operation = "sync",
                                error = %error,
                                "explicit relay synchronization failed"
                            );
                        }
                        if reply
                            .send(result.map_err(|error| format!("{error:#}")))
                            .is_err()
                        {
                            tracing::debug!(
                                session_id = %target.session_id,
                                operation = "sync",
                                "sync result receiver was already closed"
                            );
                        }
                    }
                    ActorCommand::Reviewer {
                        role,
                        action,
                        reply,
                    } => {
                        if lifecycle.is_leased()
                            || crate::controller::move_session::move_owns_session(
                                &target.session_id,
                            )
                        {
                            // A lifecycle operation owns the connection, and a
                            // reviewer action is not worth deferring: the user
                            // is waiting on its answer now.
                            tracing::debug!(
                                session_id = %target.session_id,
                                operation = action.operation_name(),
                                "rejecting a reviewer action while the session is leased"
                            );
                            if reply
                                .send(Err("session is reserved for a lifecycle operation".into()))
                                .is_err()
                            {
                                tracing::debug!(
                                    session_id = %target.session_id,
                                    operation = "reviewer",
                                    "reviewer rejection receiver was already closed"
                                );
                            }
                            continue;
                        }
                        // A slow harness startup or analysis must not occupy
                        // the primary's relay or serialize independent roles.
                        // Cache each role's connection so transcript polling
                        // does not launch a new SSH/Podman proxy every time.
                        let cached = reviewer_connections
                            .entry(role.clone())
                            .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(None)))
                            .clone();
                        let (finished, tail) = oneshot::channel::<()>();
                        let previous = reviewer_tails.insert(role.clone(), tail);
                        let target = target.clone();
                        let cancelled = reviewer_cancellation.clone();
                        reviewer_tasks.spawn(async move {
                            if let Some(previous) = previous {
                                let _ = previous.await;
                            }
                            run_reviewer_operation(target, role, action, reply, cached, cancelled)
                                .await;
                            drop(finished);
                        });
                    }
                    ActorCommand::RespondElicitation {
                        elicitation_id,
                        response,
                        reply,
                    } => {
                        if lifecycle.is_leased() {
                            tracing::debug!(
                                session_id = %target.session_id,
                                operation = "respond_elicitation",
                                "rejecting elicitation response while session is leased"
                            );
                            if reply
                                .send(Err("session is reserved for a lifecycle operation".into()))
                                .is_err()
                            {
                                tracing::debug!(
                                    session_id = %target.session_id,
                                    operation = "respond_elicitation",
                                    "elicitation rejection receiver was already closed"
                                );
                            }
                            continue;
                        }
                        let result = async {
                            sync_actor_connection(&target, &mut connection).await?;
                            let connection =
                                connection.as_mut().context("relay is disconnected")?;
                            connection
                                .respond_elicitation(elicitation_id, response)
                                .await?;
                            Ok::<_, anyhow::Error>(connection.snapshot())
                        }
                        .await;
                        match result {
                            Ok(ref snapshot) => publish_view(
                                &target.session_id,
                                ManagedSessionView {
                                    snapshot: Some(snapshot.clone()),
                                    connected: true,
                                    error: None,
                                },
                                &view_tx,
                                &updates,
                            ),
                            Err(ref error) if !is_final_rejection(error) => connection = None,
                            Err(_) => {}
                        }
                        if let Err(error) = &result {
                            tracing::warn!(
                                session_id = %target.session_id,
                                operation = "respond_elicitation",
                                error = %error,
                                "relay elicitation response failed"
                            );
                        }
                        if reply
                            .send(result.map(|_| ()).map_err(|error| format!("{error:#}")))
                            .is_err()
                        {
                            tracing::debug!(
                                session_id = %target.session_id,
                                operation = "respond_elicitation",
                                "elicitation result receiver was already closed"
                            );
                        }
                    }
                    ActorCommand::InstallPromptContext { text, reply } => {
                        if lifecycle.is_leased() {
                            let _ = reply
                                .send(Err("session is reserved for a lifecycle operation".into()));
                            continue;
                        }
                        let result = async {
                            sync_actor_connection(&target, &mut connection).await?;
                            let connection =
                                connection.as_mut().context("relay is disconnected")?;
                            connection.install_prompt_context(text).await
                        }
                        .await;
                        // Installing context changes nothing the projection
                        // shows, so there is no view to publish; a transport
                        // failure still drops the connection for a reconnect.
                        if let Err(ref error) = result {
                            if !is_final_rejection(error) {
                                connection = None;
                            }
                            tracing::warn!(
                                session_id = %target.session_id,
                                operation = "install_prompt_context",
                                error = %error,
                                "installing relay prompt context failed"
                            );
                        }
                        if reply
                            .send(result.map_err(|error| format!("{error:#}")))
                            .is_err()
                        {
                            tracing::debug!(
                                session_id = %target.session_id,
                                operation = "install_prompt_context",
                                "prompt context result receiver was already closed"
                            );
                        }
                    }
                    ActorCommand::StopBackgroundTask {
                        background_task_id,
                        reply,
                    } => {
                        if lifecycle.is_leased() {
                            let _ = reply
                                .send(Err("session is reserved for a lifecycle operation".into()));
                            continue;
                        }
                        let result = async {
                            sync_actor_connection(&target, &mut connection).await?;
                            let connection =
                                connection.as_mut().context("relay is disconnected")?;
                            connection.stop_background_task(background_task_id).await?;
                            Ok::<_, anyhow::Error>(connection.snapshot())
                        }
                        .await;
                        match result {
                            Ok(ref snapshot) => publish_view(
                                &target.session_id,
                                ManagedSessionView {
                                    snapshot: Some(snapshot.clone()),
                                    connected: true,
                                    error: None,
                                },
                                &view_tx,
                                &updates,
                            ),
                            Err(ref error) if !is_final_rejection(error) => connection = None,
                            Err(_) => {}
                        }
                        if let Err(error) = &result {
                            tracing::warn!(
                                session_id = %target.session_id,
                                operation = "stop_background_task",
                                error = %error,
                                "relay background task stop failed"
                            );
                        }
                        if reply
                            .send(result.map(|_| ()).map_err(|error| format!("{error:#}")))
                            .is_err()
                        {
                            tracing::debug!(
                                session_id = %target.session_id,
                                operation = "stop_background_task",
                                "background task stop result receiver was already closed"
                            );
                        }
                    }
                    ActorCommand::Lease {
                        idle_harness,
                        reply,
                    } => {
                        if idle_harness.is_some()
                            && (lifecycle.is_leased()
                                || !reviewer_tasks.is_empty()
                                || crate::review_host::prompt_refusal(&target.session_id).is_some())
                        {
                            if reply
                                .send(Err(super::handle::SessionNotIdle.into()))
                                .is_err()
                            {
                                tracing::debug!(session_id = %target.session_id, "idle lease requester disconnected");
                            }
                            continue;
                        }
                        if lifecycle.is_leased() {
                            tracing::debug!(
                                session_id = %target.session_id,
                                operation = "lease",
                                "rejecting duplicate session lifecycle lease"
                            );
                            if reply
                                .send(Err(anyhow::anyhow!(
                                    "session already has a lifecycle operation"
                                )))
                                .is_err()
                            {
                                tracing::debug!(
                                    session_id = %target.session_id,
                                    operation = "lease",
                                    "lease rejection receiver was already closed"
                                );
                            }
                            continue;
                        }
                        let lease_id = next_lease_id;
                        let synced = sync_actor_connection(&target, &mut connection).await;
                        if let Err(error) = synced {
                            connection = None;
                            tracing::warn!(session_id = %target.session_id, %error, "could not sync before leasing the session");
                            if reply.send(Err(error)).is_err() {
                                tracing::debug!(session_id = %target.session_id, "lease requester disconnected");
                            }
                            continue;
                        }
                        // Decide before taking the control channel or cancelling
                        // reviewers. A stale idle observation must not hold steering.
                        if idle_harness.is_some_and(|harness| {
                            !connection
                                .as_ref()
                                .expect("synced connection")
                                .snapshot()
                                .operational
                                .safe_to_replace(harness)
                        }) {
                            if reply
                                .send(Err(super::handle::SessionNotIdle.into()))
                                .is_err()
                            {
                                tracing::debug!(session_id = %target.session_id, "idle lease requester disconnected");
                            }
                            continue;
                        }
                        reviewer_cancellation.cancel();
                        reviewer_cancellation = tokio_util::sync::CancellationToken::new();
                        reviewer_connections.clear();
                        reviewer_tails.clear();
                        next_lease_id = next_lease_id.wrapping_add(1).max(1);
                        match reply.send(Ok((
                            lease_id,
                            connection.take().expect("synced connection"),
                        ))) {
                            Ok(()) => lifecycle.activate_lease(lease_id),
                            Err(Ok((_lease_id, returned))) => connection = Some(returned),
                            Err(Err(_)) => {}
                        }
                    }
                }
            }
            Event::Retirement(changed) => {
                if changed.is_err() {
                    break;
                }
            }
        }
    }
    reviewer_cancellation.cancel();
    reviewer_connections.clear();
    reviewer_tails.clear();
    while let Some(completed) = reviewer_tasks.join_next().await {
        if let Err(error) = completed {
            tracing::error!(session_id = %target.session_id, %error, "reviewer operation task failed during shutdown");
        }
    }
    if let Some(connection) = connection.take()
        && let Err(error) = connection.detach().await
    {
        tracing::warn!(
            session_id = %target.session_id,
            %error,
            "could not detach relay connection during session actor shutdown"
        );
    }
    // No caller may wait forever on a submission this actor will never deliver.
    for deferred in deferred_submits {
        if deferred
            .reply
            .send(Err("session manager stopped".into()))
            .is_err()
        {
            tracing::debug!(
                session_id = %target.session_id,
                operation = "submit",
                "deferred submit shutdown receiver was already closed"
            );
        }
    }
    // A submission still in the command queue never reached the relay. Say
    // so, instead of dropping its reply, which its caller would have to read
    // as possibly delivered: a sub-agent's `send_input` resends a prompt only
    // when it knows the first one was not delivered.
    commands.close();
    while let Ok(command) = commands.try_recv() {
        if let ActorCommand::Submit { reply, .. } = command
            && reply.send(Err("session manager stopped".into())).is_err()
        {
            tracing::debug!(
                session_id = %target.session_id,
                operation = "submit",
                "queued submit shutdown receiver was already closed"
            );
        }
    }
}

/// Submit one relay command and publish the resulting snapshot. Live and
/// deferred submissions share this path so both report identical results.
pub(super) async fn deliver_submit(
    target: &RelaySessionTarget,
    connection: &mut Option<StandaloneSession>,
    submission: DeferredSubmit,
    view_tx: &watch::Sender<ManagedSessionView>,
    updates: &CoalescedUpdateSender,
) {
    let DeferredSubmit {
        durable,
        queued_at,
        command_id,
        command,
        admission,
        reply,
    } = submission;
    if crate::controller::move_session::move_refuses_command(&target.session_id, &command) {
        let _ = reply.send(Err(
            "session is moving; keep the draft and retry after Move finishes".into(),
        ));
        return;
    }
    if let Some(admission) = admission.as_ref()
        && (!matches!(&command, RelayCommand::Prompt { .. })
            || admission.command_id() != command_id
            || !crate::review_host::review_delivery_admitted(&target.session_id, admission))
    {
        let _ = reply.send(Err("review delivery admission is no longer valid".into()));
        return;
    }
    if matches!(
        command,
        RelayCommand::ContinueAuthorizedWork { .. } | RelayCommand::ResumeAfterQuota { .. }
    ) {
        if reply.is_closed() {
            return;
        }
        if let Some(refusal) = crate::review_host::prompt_refusal(&target.session_id) {
            let _ = reply.send(Err(refusal.into()));
            return;
        }
    }
    let started = Instant::now();
    tracing::debug!(target: "mj_controller::latency", session_id = %target.session_id,
        %command_id, queue_ms = queued_at.elapsed().as_secs_f64() * 1000.0, "submission dispatched");
    let result =
        submit_actor_command_with_receipt(target, connection, &command_id, &command, durable).await;
    tracing::debug!(target: "mj_controller::latency", session_id = %target.session_id,
        %command_id, elapsed_ms = started.elapsed().as_secs_f64() * 1000.0,
        accepted_ordinal = ?result.as_ref().ok(), "submission answered");
    if let Err(error) = result.as_ref() {
        tracing::warn!(
            session_id = %target.session_id,
            operation = "submit",
            %command_id,
            retryable = !is_final_rejection(error),
            error = %error,
            "relay command submission failed"
        );
    }
    if let Err(error) = result.as_ref()
        && !is_final_rejection(error)
    {
        *connection = None;
    }
    let accepted = result.as_ref().ok().copied();
    // Answer the caller the moment the relay has the command. Catching the
    // local projection up to it is the expensive half and nobody waiting to
    // hear "accepted" needs it first: the caller has an ordinal, and the view
    // it would read is published below anyway.
    if reply
        .send(result.map_err(|error| submit_failure(&error)))
        .is_err()
    {
        tracing::debug!(
            session_id = %target.session_id,
            operation = "submit",
            %command_id,
            "submit result receiver was already closed"
        );
    }
    let Some(ordinal) = accepted else {
        return;
    };
    tracing::trace!(%ordinal, %command_id, "relay command accepted");
    let Some(session) = connection.as_mut() else {
        return;
    };
    // The command landed either way, so a failed catch-up is a connection
    // problem to retire rather than a failed submission: the caller has
    // already been told the relay took it.
    match session.sync().await {
        Ok(snapshot) => publish_view(
            &target.session_id,
            ManagedSessionView {
                snapshot: Some(snapshot),
                connected: true,
                error: None,
            },
            view_tx,
            updates,
        ),
        Err(error) => {
            tracing::warn!(
                session_id = %target.session_id,
                operation = "submit",
                %command_id,
                error = %format!("{error:#}"),
                "projection could not catch up to an accepted command"
            );
            *connection = None;
        }
    }
}

/// Whether the relay refused this request outright.
///
/// A refusal is a completed round trip, so the connection is healthy. Dropping
/// it would discard whatever that connection owns on the worker, including a
/// checkpoint barrier a controller is still holding.
/// What the submitter is told. A final rejection is the relay's own reason,
/// such as "/clear requires an idle session…"; the relay version, operation,
/// and error code stay in the log written above (I1-12).
pub(super) fn submit_failure(error: &anyhow::Error) -> mj_client::session::SubmitFailure {
    match error
        .downcast_ref::<RelayRejected>()
        .filter(|rejected| !rejected.is_retryable())
    {
        Some(rejected) => mj_client::session::SubmitFailure {
            unconfirmed: false,
            message: rejected.0.message.clone(),
        },
        None => mj_client::session::SubmitFailure {
            unconfirmed: true,
            message: format!("{error:#}"),
        },
    }
}

pub(super) fn is_final_rejection(error: &anyhow::Error) -> bool {
    error
        .downcast_ref::<RelayRejected>()
        .is_some_and(|rejected| !rejected.is_retryable())
}

async fn submit_actor_command_with_receipt(
    target: &RelaySessionTarget,
    connection: &mut Option<StandaloneSession>,
    command_id: &str,
    command: &RelayCommand,
    durable: bool,
) -> Result<u64> {
    let mut first_error = None;
    for attempt in 1..=2 {
        if connection.is_none() {
            sync_actor_connection(target, connection).await?;
        }
        let client = connection.as_mut().context("relay is disconnected")?;
        let result = if durable {
            client
                .submit_durable_accepted(command_id.to_owned(), command.clone())
                .await
        } else {
            client
                .submit_accepted(command_id.to_owned(), command.clone())
                .await
        };
        match result {
            Ok(ordinal) => return Ok(ordinal),
            // A final rejection is a completed round trip: the relay read the
            // command and refused it, so retrying would only be refused again.
            // Reconnecting would also cancel any checkpoint barrier this
            // connection owns, which is how a controller probing for a command
            // an older worker does not understand would lose it.
            Err(error) if is_final_rejection(&error) => return Err(error),
            Err(error) => {
                tracing::warn!(
                    session_id = %target.session_id,
                    operation = "submit",
                    %command_id,
                    attempt,
                    retryable = true,
                    error = %error,
                    "retryable relay command failure; reconnecting"
                );
                if first_error.is_none() {
                    first_error = Some(format!("{error:#}"));
                }
                *connection = None;
            }
        }
    }
    let detail = first_error.unwrap_or_else(|| "relay submission failed".into());
    bail!("relay command {command_id} failed after an idempotent reconnect: {detail}")
}

/// Perform one reviewer action on a synchronized relay connection.
///
/// The reviewer's own relay answers most of these, so the outcomes mirror the
/// primary's: an attach page, an acknowledgement cursor, an accepted command.
pub(super) async fn run_reviewer_operation(
    target: RelaySessionTarget,
    role: Option<String>,
    action: ReviewerAction,
    mut reply: oneshot::Sender<std::result::Result<ReviewerOutcome, String>>,
    cached: Arc<tokio::sync::Mutex<Option<RelayClient>>>,
    cancelled: tokio_util::sync::CancellationToken,
) {
    let operation = action.operation_name();
    let keep_connection = !matches!(
        &action,
        ReviewerAction::Pause | ReviewerAction::PauseGeneration { .. }
    );
    let result = tokio::select! {
        biased;
        _ = cancelled.cancelled() => Err(anyhow::anyhow!("reviewer operation cancelled for session lifecycle change")),
        _ = reply.closed() => return,
        result = async {
            let mut cache = cached.lock().await;
            // Take ownership while a request is in flight: dropping this
            // future closes its connection instead of leaving a late reply
            // available for the next request to misinterpret.
            let mut client = match cache.take() {
                Some(client) => client,
                None => RelayClient::connect(&target.spec, &target.session_id).await?,
            };
            let result = drive_reviewer(&mut client, role, action).await;
            if keep_connection && (result.is_ok() || result.as_ref().is_err_and(is_final_rejection)) {
                *cache = Some(client);
            }
            result
        } => result,
    };
    if let Err(error) = &result {
        tracing::warn!(session_id = %target.session_id, %operation, error = %error, "reviewer action failed");
    }
    if reply
        .send(result.map_err(|error| format!("{error:#}")))
        .is_err()
    {
        tracing::debug!(session_id = %target.session_id, %operation, "reviewer result receiver was already closed");
    }
}

pub(super) async fn drive_reviewer(
    client: &mut RelayClient,
    role: Option<String>,
    action: ReviewerAction,
) -> Result<ReviewerOutcome> {
    let role = role.as_deref();
    Ok(match action {
        ReviewerAction::Start { config } => {
            ReviewerOutcome::Started(Box::new(client.start_reviewer(role, *config).await?))
        }
        ReviewerAction::Submit {
            command_id,
            command,
        } => ReviewerOutcome::Accepted {
            ordinal: client.submit_to_reviewer(role, command_id, command).await?,
        },
        ReviewerAction::Attach {
            after_ordinal,
            after_digest,
        } => ReviewerOutcome::Attached(Box::new(
            client
                .attach_reviewer(role, after_ordinal, after_digest)
                .await?,
        )),
        ReviewerAction::Acknowledge {
            through_ordinal,
            through_digest,
        } => ReviewerOutcome::Acknowledged(
            client
                .acknowledge_reviewer(role, through_ordinal, through_digest)
                .await?,
        ),
        ReviewerAction::Status => {
            ReviewerOutcome::Status(Box::new(client.reviewer_status(role).await?))
        }
        ReviewerAction::RespondElicitation {
            elicitation_id,
            response,
        } => {
            client
                .respond_to_reviewer(role, elicitation_id, response)
                .await?;
            ReviewerOutcome::ElicitationResolved
        }
        ReviewerAction::PauseGeneration { generation } => {
            client.pause_reviewer_generation(role, generation).await?;
            ReviewerOutcome::Paused
        }
        ReviewerAction::Pause => {
            client.pause_reviewer(role).await?;
            ReviewerOutcome::Paused
        }
        ReviewerAction::CaptureDelta { baselines } => ReviewerOutcome::Delta {
            repositories: client.capture_review_delta(role, baselines).await?,
        },
        ReviewerAction::AdvanceBaseline { trees } => {
            client.advance_review_baseline(role, trees).await?;
            ReviewerOutcome::BaselineAdvanced
        }
        ReviewerAction::AnalyzeDelta { repositories } => ReviewerOutcome::ChangedFunctions {
            packet: client.analyze_review_delta(role, repositories).await?,
        },
        ReviewerAction::SubmitDurable {
            generation,
            command_id,
            command,
        } => ReviewerOutcome::Accepted {
            ordinal: client
                .submit_reviewer_durable(role, generation, command_id, command)
                .await?,
        },
        ReviewerAction::ReviewerCommandReceipt {
            generation,
            command_id,
        } => {
            match client
                .reviewer_receipt(
                    role,
                    mj_core::relay::ReviewerRequest::CommandReceipt {
                        generation,
                        command_id,
                    },
                )
                .await?
            {
                mj_core::relay::RelayResponsePayload::CommandReceipt { receipt } => {
                    ReviewerOutcome::CommandReceipt {
                        receipt: receipt.map(Box::new),
                    }
                }
                _ => anyhow::bail!("unexpected reviewer command receipt response"),
            }
        }
        ReviewerAction::CancelReviewerCommandAdmission {
            generation,
            command_id,
        } => {
            match client
                .reviewer_receipt(
                    role,
                    mj_core::relay::ReviewerRequest::CancelCommandAdmission {
                        generation,
                        command_id,
                    },
                )
                .await?
            {
                mj_core::relay::RelayResponsePayload::CommandReceipt { receipt } => {
                    ReviewerOutcome::CommandReceipt {
                        receipt: receipt.map(Box::new),
                    }
                }
                _ => anyhow::bail!("unexpected reviewer cancellation receipt response"),
            }
        }
        ReviewerAction::ReleaseReviewerCommandReceipt {
            generation,
            command_id,
        } => {
            match client
                .reviewer_receipt(
                    role,
                    mj_core::relay::ReviewerRequest::ReleaseCommandReceipt {
                        generation,
                        command_id,
                    },
                )
                .await?
            {
                mj_core::relay::RelayResponsePayload::CommandReceiptReleased => {
                    ReviewerOutcome::CommandReceiptReleased
                }
                _ => anyhow::bail!("unexpected reviewer receipt release response"),
            }
        }
        ReviewerAction::ReadLaneDispatches => ReviewerOutcome::PendingLaneDispatches {
            dispatches: client.read_lane_dispatches().await?,
        },
        ReviewerAction::AckLaneDispatches { ids } => {
            client.ack_lane_dispatches(ids).await?;
            ReviewerOutcome::LaneDispatchesAcknowledged
        }
        ReviewerAction::TakeLaneDispatches => ReviewerOutcome::LaneDispatches {
            requests: client.take_lane_dispatches().await?,
        },
    })
}

pub(super) async fn sync_actor_connection(
    target: &RelaySessionTarget,
    connection: &mut Option<StandaloneSession>,
) -> Result<Option<ManagedSessionSnapshot>> {
    if connection.is_none() {
        let fresh = StandaloneSession::connect(target).await?;
        let snapshot = fresh.snapshot();
        if let Some(recovery) = target.worker_recovery.as_ref() {
            let session_id = target.session_id.clone();
            let source_target = recovery.source_target.clone();
            tokio::task::spawn_blocking(move || -> Result<()> {
                if let Some(intent) = crate::database::load_worker_restart(&session_id)?
                    && intent.target == source_target
                {
                    // A successful sync establishes a live owner, including
                    // an older worker that survived a pre-stop interruption or
                    // a newer build recovered by the replacement daemon.
                    // Upgrade policy owns the separate build-version decision.
                    crate::database::finish_worker_restart(&session_id, &intent.operation_id)?;
                }
                Ok(())
            })
            .await
            .context("settle worker restart observation")??;
        }
        *connection = Some(fresh);
        return Ok(Some(snapshot));
    }
    let connection = connection.as_mut().expect("connection was initialized");
    if connection.sync_in_place().await? {
        Ok(Some(connection.snapshot()))
    } else {
        Ok(None)
    }
}

/// Cheap equivalence for published views.
///
/// The materialized projection is a function of the relay event chain, so its
/// transcript can only differ when the applied event frontier differs. Every
/// sync tick would otherwise walk the whole conversation to prove nothing
/// changed. The remaining scalars are compared directly because they are small
/// and bound the projection's non-transcript state.
pub(super) fn view_is_unchanged(current: &ManagedSessionView, next: &ManagedSessionView) -> bool {
    if current.connected != next.connected || current.error != next.error {
        return false;
    }
    match (&current.snapshot, &next.snapshot) {
        (None, None) => true,
        (Some(current), Some(next)) => {
            let (current_session, next_session) = (&current.materialized, &next.materialized);
            current.latest_credential_sync_signal == next.latest_credential_sync_signal
                && current.operational == next.operational
                // Sub-agent requests/results are non-transcript projection state:
                // they come from the separate `subagents.json` poll in
                // `sync_in_place`, not the relay event chain, so they can change
                // while every transcript scalar below stays identical. They must
                // be compared here, or a request that lands without a coincident
                // view change (e.g. one that survives a daemon restart, where the
                // tool-call ordinal is already applied) is never republished to the
                // drain and its `serve_one` waits to the socket ceiling.
                && current.subagent_requests == next.subagent_requests
                && current.subagent_results == next.subagent_results
                && current_session.session_id == next_session.session_id
                && current_session.applied_event_ordinal == next_session.applied_event_ordinal
                && current_session.applied_event_digest == next_session.applied_event_digest
                && current_session.last_activity_at_ms == next_session.last_activity_at_ms
                && current_session.execution == next_session.execution
                && current_session.session_title == next_session.session_title
                && current_session.queued_prompts == next_session.queued_prompts
        }
        (None, Some(_)) | (Some(_), None) => false,
    }
}

pub(super) fn publish_view(
    session_id: &str,
    view: ManagedSessionView,
    watch: &watch::Sender<ManagedSessionView>,
    updates: &CoalescedUpdateSender,
) {
    // Compare and replace under one lock acquisition; a separate
    // `watch.borrow()` check would reacquire the lock and invite the
    // read-then-write deadlock this function's callers must avoid.
    tracing::debug!(target: "mj_controller::latency", %session_id,
        ordinal = ?view.snapshot.as_ref().map(|s| s.materialized.applied_event_ordinal),
        "session view publication");
    let changed = watch.send_if_modified(|current| {
        if view_is_unchanged(current, &view) {
            return false;
        }
        *current = view.clone();
        true
    });
    if changed {
        updates.send(SessionManagerUpdate {
            session_id: session_id.to_owned(),
            view,
        });
    }
}

/// Read a stored projection without blocking the runtime. The rusqlite read
/// and the transcript deserialization behind it are synchronous and grow with
/// the conversation, so a long session must not stall a worker thread that
/// other actors share.
pub(super) async fn load_projection(
    session_id: &str,
) -> Result<(MaterializedSession, mj_core::state::ProjectionWindow)> {
    let session_id = session_id.to_owned();
    tokio::task::spawn_blocking(move || {
        let loaded = crate::database::load_materialized_actor_projection(&session_id)?;
        Ok(loaded.unwrap_or_else(|| {
            let materialized = MaterializedSession::empty(session_id);
            let window = mj_core::state::ProjectionWindow::of(&materialized);
            (materialized, window)
        }))
    })
    .await
    .context("controller projection load task failed")?
}