brokk-mj-controller 2.17.0

use super::*;

use crate::server_runtime::profile_catalog::{ProfileCatalog, counting_probe, test_config};
use mj_core::state::SessionRecord;

fn quota(profile_id: &str, harness: HarnessKind, remaining: &[u8]) -> ProfileQuota {
    ProfileQuota {
        profile_id: profile_id.into(),
        harness,
        windows: remaining
            .iter()
            .map(|remaining_percent| crate::quota::QuotaWindow {
                label: "window".into(),
                remaining_percent: Some(*remaining_percent),
                used: None,
                limit: None,
                resets: None,
                resets_at_epoch_seconds: None,
            })
            .collect(),
        extra: None,
        error: None,
        refreshed_at_epoch_seconds: 0,
    }
}

#[test]
fn subagent_profiles_choose_the_most_remaining_quota_per_harness() {
    let candidates = vec![
        ("codex-low".into(), HarnessKind::Codex),
        ("claude-only".into(), HarnessKind::Claude),
        ("codex-high".into(), HarnessKind::Codex),
    ];
    let reports = BTreeMap::from([
        (
            "codex-low".into(),
            quota("codex-low", HarnessKind::Codex, &[80, 15]),
        ),
        (
            "codex-high".into(),
            quota("codex-high", HarnessKind::Codex, &[60, 55]),
        ),
    ]);

    assert_eq!(
        select_profile_per_harness(candidates, &reports),
        vec![
            ("codex-high".into(), HarnessKind::Codex),
            ("claude-only".into(), HarnessKind::Claude),
        ]
    );
}

#[test]
fn subagent_profile_selection_puts_unknown_quota_last_and_breaks_ties_by_id() {
    let candidates = vec![
        ("codex-unknown".into(), HarnessKind::Codex),
        ("codex-b".into(), HarnessKind::Codex),
        ("codex-a".into(), HarnessKind::Codex),
    ];
    let reports = BTreeMap::from([
        (
            "codex-a".into(),
            quota("codex-a", HarnessKind::Codex, &[50]),
        ),
        (
            "codex-b".into(),
            quota("codex-b", HarnessKind::Codex, &[50]),
        ),
    ]);

    assert_eq!(
        select_profile_per_harness(candidates, &reports),
        vec![("codex-a".into(), HarnessKind::Codex)]
    );
}

#[test]
fn failed_subagent_followup_is_terminal_error_with_its_cause() {
    let status = StartStatus::Failed {
        message: "model is unavailable".into(),
    };
    assert_eq!(
        subagent_status(None, None, Some(&status), None, false),
        ("error".into(), Some("model is unavailable".into()), true)
    );
}

/// A close is accepted long before the child is gone: it cancels whatever owned
/// the session, checkpoints, seals the relay and tears the process tree down,
/// and the record only says `Closing` once that is under way. A child that had
/// finished its turn therefore reported `completed` the moment its close was
/// admitted, so a parent that closed a child and spawned its replacement stacked
/// both process trees inside one container (#1087). A wait must follow the close
/// instead, the way a session-level wait does.
#[test]
fn a_child_whose_close_is_running_is_not_finished_until_the_close_is() {
    let mut record = crate::controller::test_support::checkpoint_test_session("child");
    record.state = SessionState::Running;
    let summary = mj_core::state::MaterializedSessionSummary {
        session_id: "child".into(),
        applied_event_ordinal: 4,
        last_activity_at_ms: None,
        execution: MaterializedExecutionState::Idle,
        session_title: None,
        last_agent_message: Some("the child's report".into()),
        last_user_message: None,
        last_agent_message_follows_last_user: true,
        agent_message_latest_content_ordinals: Vec::new(),
        interruption_event_ordinals: Vec::new(),
    };

    assert_eq!(
        subagent_status(Some(&record), Some(&summary), None, None, false),
        ("completed".into(), Some("the child's report".into()), true),
        "an idle child nobody is closing is finished"
    );

    let (state, output, finished) =
        subagent_status(Some(&record), Some(&summary), None, None, true);
    assert_eq!(state, "stopping");
    assert_eq!(output, None);
    assert!(
        !finished,
        "a wait must keep following a child whose close is still running"
    );

    // A child whose start failed is terminal, but it is not gone either while
    // its close runs; its cause was already reported to the parent before it
    // asked for the close.
    let failed = StartStatus::Failed {
        message: "model is unavailable".into(),
    };
    assert!(
        !subagent_status(Some(&record), Some(&summary), Some(&failed), None, true).2,
        "a failed child is still being torn down while its close runs"
    );

    // The close finished: the record settled, and the flag the daemon clears
    // just after it is no longer allowed to hold the wait open.
    record.state = SessionState::Stopped;
    assert_eq!(
        subagent_status(Some(&record), Some(&summary), None, None, true),
        ("stopped".into(), None, true),
        "a record that already settled to stopped ends the wait"
    );
}

/// When a child's startup failed, the record holds the reason and the
/// follow-up holds only the symptom: that the session would not take a first
/// prompt. The parent model must read the reason, which is what #1065 could
/// not do.
#[test]
fn a_failed_child_reports_the_startup_cause_rather_than_the_symptom() {
    let mut record = crate::controller::test_support::checkpoint_test_session("child");
    record.state = SessionState::Error;
    record.last_error = Some(
        "sub-agent startup failed: the worker process is gone; it reached the startup step \
         \"review-baseline\""
            .into(),
    );
    let status = StartStatus::Failed {
        message: "session child is Error and will not take a first prompt".into(),
    };

    let (state, output, terminal) =
        subagent_status(Some(&record), None, Some(&status), None, false);

    assert_eq!(state, "error");
    assert!(terminal);
    assert_eq!(output.as_deref(), record.last_error.as_deref());
}
use mj_client::session::{
    ManagedSessionView, PendingRelaySubmit, PendingRelaySync, SessionControlBackend,
    SessionHandleBackend,
};
use tokio::sync::mpsc;

/// A session actor that records what it was asked to submit and accepts it
/// at a fixed ordinal. Hand-written rather than mocked so the test proves
/// the exact relay command the API sends.
#[derive(Clone)]
struct FakeSession {
    session_id: String,
    accepted_ordinal: u64,
    submitted: mpsc::UnboundedSender<(String, RelayCommand)>,
    /// What `view()` reports. `None` is the default empty view, which is
    /// what a session that has not connected yet looks like.
    view: Option<ManagedSessionView>,
}

impl SessionHandleBackend for FakeSession {
    fn search_prompts(
        &self,
        _bundle_id: String,
        _scope: mj_core::storage::HistoryScope,
        _query: String,
    ) -> mj_client::session::BoxFuture<'_, Result<Vec<mj_core::storage::PromptHistoryEntry>>> {
        Box::pin(async { Ok(Vec::new()) })
    }
    fn review_state(
        &self,
    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::ReviewState>> {
        Box::pin(async { Ok(Default::default()) })
    }

    fn config_result(&self, _command_id: String) -> BoxFuture<'_, Result<Option<Option<String>>>> {
        Box::pin(async { Ok(Some(None)) })
    }
    fn clone_box(&self) -> Box<dyn SessionHandleBackend> {
        Box::new(self.clone())
    }
    fn session_id(&self) -> &str {
        &self.session_id
    }
    fn view(&self) -> ManagedSessionView {
        self.view.clone().unwrap_or_default()
    }
    fn is_stopped(&self) -> bool {
        false
    }
    fn has_changed(&self) -> Result<bool> {
        Ok(false)
    }
    fn changed(&mut self) -> BoxFuture<'_, Result<ManagedSessionView>> {
        Box::pin(std::future::pending())
    }
    fn enqueue_submit(
        &self,
        command_id: String,
        command: RelayCommand,
    ) -> BoxFuture<'_, Result<PendingRelaySubmit>> {
        let _ = self.submitted.send((command_id, command));
        let ordinal = self.accepted_ordinal;
        Box::pin(async move {
            Ok(PendingRelaySubmit::new(Box::pin(
                async move { Ok(ordinal) },
            )))
        })
    }
    fn enqueue_sync(&self) -> BoxFuture<'_, Result<PendingRelaySync>> {
        Box::pin(async move { Ok(PendingRelaySync::new(Box::pin(async { Ok(()) }))) })
    }
    fn respond_elicitation(
        &self,
        _elicitation_id: String,
        _response: mj_core::elicitation::ElicitationResponse,
    ) -> BoxFuture<'_, Result<()>> {
        Box::pin(async { Ok(()) })
    }
    fn stop_background_task(&self, _background_task_id: String) -> BoxFuture<'_, Result<()>> {
        Box::pin(async { Ok(()) })
    }
    fn reviewer(
        &self,
        _role: Option<String>,
        _action: mj_client::session::ReviewerAction,
    ) -> BoxFuture<'_, Result<mj_client::session::ReviewerOutcome>> {
        Box::pin(async { anyhow::bail!("no reviewer in this fake") })
    }
}

/// Every session this daemon is asked about is up and running.
fn running_states() -> SessionStateSource {
    Arc::new(|_| Some(SessionState::Running))
}

#[test]
fn checkpoint_export_retains_typed_deferrals_and_real_failures() {
    let error = anyhow!("disk failed");
    assert!(matches!(
        checkpoint_export_error(error),
        ExportError::Failed(_)
    ));
    let error = anyhow::Error::new(crate::controller::CheckpointDeferred::harness_busy())
        .context("capture bundle");
    let ExportError::Refused(message) = checkpoint_export_error(error) else {
        panic!("expected a deferred export");
    };
    assert!(message.contains("capture bundle"));
    assert!(message.contains("agent is working"));
}

/// An export runtime with nothing in it. The follow-up tests never export;
/// the export path's own behavior is proved by the API handler tests.
struct NoExports;

impl ExportRuntime for NoExports {
    fn session_record(&self, _session_id: &str) -> Option<mj_core::state::SessionRecord> {
        None
    }
    fn checkpoint_now(
        &self,
        session_id: String,
    ) -> BoxFuture<'_, Result<mj_core::state::CheckpointMetadata>> {
        Box::pin(async move { bail!("session {session_id} cannot be checkpointed in a test") })
    }
}

/// A connected view whose harness is ready and offers one model.
fn ready_view(model: &str) -> ManagedSessionView {
    let materialized = mj_core::state::MaterializedSession::empty("session-1");
    let option: agent_client_protocol::schema::v1::SessionConfigOption =
        serde_json::from_value(serde_json::json!({
            "id": "model",
            "name": "Model",
            "category": "model",
            "type": "select",
            "currentValue": model,
            "options": [{"value": model, "name": model}],
        }))
        .expect("the fixture describes a select the schema accepts");
    let operational = mj_core::relay::RelayOperationalState {
        continuation: Default::default(),
        relay_protocol_version: Some(mj_core::relay::RELAY_PROTOCOL_VERSION),
        native_agents: Vec::new(),
        steering: None,
        cancelling_prompt_id: None,
        clear_context: false,
        clear_context_started_at_ms: None,
        native_agent_count: 0,
        expected_continuation: None,
        inferred_idle_since_ms: None,
        goal: Default::default(),
        capacity_retry: None,
        activity_turn_started_at_ms: None,
        idle_since_ms: None,
        store_id: None,
        session_id: "session-1".into(),
        execution: mj_core::relay::RelayExecutionState::Idle,
        latest_ordinal: 0,
        latest_digest: mj_core::relay::RELAY_EVENT_GENESIS_DIGEST.into(),
        acknowledged_through: 0,
        acknowledged_digest: mj_core::relay::RELAY_EVENT_GENESIS_DIGEST.into(),
        recovery_floor_ordinal: 0,
        recovery_floor_digest: mj_core::relay::RELAY_EVENT_GENESIS_DIGEST.into(),
        native_session_id: Some("native-1".into()),
        native_continuity_lost: false,
        checkpoint_only: false,
        acp_ready: Some(true),
        agent_capabilities: None,
        agent_info: None,
        steering_supported: None,
        config_options: vec![option],
        modes: None,
        available_commands: Vec::new(),
        config: BTreeMap::new(),
        active_prompt: None,
        queued_prompts: Vec::new(),
        active_user_shells: Vec::new(),
        active_agent_terminals: Vec::new(),
        checkpoint_barrier: None,
        checkpoint_ready: None,
        last_acp_activity_at_ms: None,
        current_step_started_at_ms: None,
        foreground_tool_started_at_ms: None,
        tools_in_flight: Vec::new(),
        activity: None,
        harness_turn: None,
        last_harness_turn_started_ordinal: None,
        background_commands: Vec::new(),
        background_work_known: None,
    };
    ManagedSessionView {
        snapshot: Some(mj_core::state::ManagedSessionSnapshot {
            subagent_requests: Vec::new(),
            subagent_results: Vec::new(),
            window: mj_core::state::ProjectionWindow::of(&materialized),
            materialized,
            operational,
            latest_credential_sync_signal: None,
            worker_build: None,
        }),
        connected: true,
        error: None,
    }
}

struct FakeControl(FakeSession);

impl SessionControlBackend for FakeControl {
    fn session(&self, session_id: String) -> BoxFuture<'_, Result<SessionHandle>> {
        let session = self.0.clone();
        Box::pin(async move {
            anyhow::ensure!(session_id == session.session_id, "unknown session");
            Ok(SessionHandle::new(session))
        })
    }
}

/// A running parent session, as the durable record the tool path reads it
/// from. Only the fields the tool path uses carry meaning.
struct ParentExports(SessionRecord);

impl ExportRuntime for ParentExports {
    fn session_record(&self, session_id: &str) -> Option<SessionRecord> {
        (session_id == self.0.id).then(|| self.0.clone())
    }
    fn checkpoint_now(
        &self,
        session_id: String,
    ) -> BoxFuture<'_, Result<mj_core::state::CheckpointMetadata>> {
        Box::pin(async move { bail!("session {session_id} cannot be checkpointed in a test") })
    }
}

fn parent_record(id: &str, profile: &str) -> SessionRecord {
    SessionRecord {
        mjolnir_subagents: None,
        create_managed_worktree: None,
        container_workspace: None,
        build_cache: None,
        workspace_id: mj_core::workspace::DEFAULT_WORKSPACE_ID.to_owned(),
        archived: false,
        container_cpus: None,
        container_memory: None,
        id: id.to_owned(),
        title: "parent".into(),
        harness_kind: HarnessKind::Codex,
        last_profile: profile.to_owned(),
        bundle_id: "hel".into(),
        project_directory: None,
        managed_worktree: None,
        target_template_id: "podman".into(),
        resource_allocation: None,
        additional_mounts: Vec::new(),
        state: SessionState::Running,
        target: None,
        native_session_id: None,
        acp_session_title: None,
        session_title_override: None,
        created_at: "2026-08-09T12:00:00Z".into(),
        updated_at: "2026-08-09T12:01:00Z".into(),
        viewed_through_event_ordinal: 0,
        draft_input: String::new(),
        last_error: None,
        last_checkpoint_error: None,
        checkpoint: None,
    }
}

/// The behaviour the whole change exists for: once the background pass has
/// discovered the profiles, the tool call itself discovers nothing.
#[tokio::test]
async fn list_profiles_answers_from_the_warm_catalogue_without_probing_again() {
    let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let catalog = ProfileCatalog::with_probe(counting_probe(calls.clone()));
    let config = test_config(
        &[
            ("parent", HarnessKind::Codex),
            ("helper", HarnessKind::Claude),
        ],
        &["helper"],
    );
    catalog.sync_now(&config).await;
    let warmed = calls.load(std::sync::atomic::Ordering::SeqCst);
    assert_eq!(warmed, 2, "the pass discovers every enabled profile once");

    let backend = Arc::new(
        ApiBackend::new(
            SessionControl::new(FakeControl(FakeSession {
                session_id: "parent-1".into(),
                accepted_ordinal: 1,
                submitted: mpsc::unbounded_channel().0,
                view: None,
            })),
            running_states(),
            Arc::new(ParentExports(parent_record("parent-1", "parent"))),
        )
        .with_profile_catalog(catalog),
    );

    let result = backend
        .execute_subagent_tool(
            "parent-1".into(),
            mj_core::subagent::SubagentToolRequest {
                request_id: "request-1".into(),
                created_at_ms: 0,
                action: mj_core::subagent::SubagentToolAction::ListProfiles,
            },
        )
        .await;

    assert!(!result.is_error, "the tool call failed: {}", result.message);
    let answer: serde_json::Value =
        serde_json::from_str(&result.message).expect("the answer is JSON");
    let profiles = answer["profiles"]
        .as_array()
        .expect("the answer names profiles")
        .clone();
    assert_eq!(
        profiles
            .iter()
            .map(|profile| (
                profile["profile_id"].as_str().unwrap().to_owned(),
                profile["harness"].as_str().unwrap().to_owned(),
            ))
            .collect::<Vec<_>>(),
        vec![
            ("parent".to_owned(), "codex".to_owned()),
            ("helper".to_owned(), "claude".to_owned()),
        ],
        "the parent's own profile and the eligible one are offered, once per harness"
    );
    for profile in &profiles {
        let id = profile["profile_id"].as_str().unwrap();
        assert_eq!(
            profile["default_model"].as_str(),
            Some(format!("{id}-model").as_str()),
            "the answer quotes the discovered default model"
        );
    }
    assert_eq!(
        calls.load(std::sync::atomic::Ordering::SeqCst),
        warmed,
        "the call must not discover anything the background pass already did"
    );
}

/// A backend whose parent session is the `parent` profile, over the given
/// catalogue, so a spawn can be driven against a warm or a cold one.
fn spawn_backend(catalog: Arc<ProfileCatalog>) -> Arc<ApiBackend> {
    Arc::new(
        ApiBackend::new(
            SessionControl::new(FakeControl(FakeSession {
                session_id: "parent-1".into(),
                accepted_ordinal: 1,
                submitted: mpsc::unbounded_channel().0,
                view: None,
            })),
            running_states(),
            Arc::new(ParentExports(parent_record("parent-1", "parent"))),
        )
        .with_profile_catalog(catalog),
    )
}

fn spawn_request(model: &str) -> mj_core::subagent::SubagentToolRequest {
    mj_core::subagent::SubagentToolRequest {
        request_id: "request-1".into(),
        created_at_ms: 0,
        action: mj_core::subagent::SubagentToolAction::Spawn {
            task_name: "audit deps".into(),
            instructions: "check the lockfile".into(),
            profile_id: None,
            model: Some(model.into()),
            effort: None,
            working_directory: Default::default(),
            context: None,
            files: Vec::new(),
        },
    }
}

#[tokio::test]
async fn spawn_rejects_a_model_the_warm_catalogue_says_is_not_offered() {
    let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let catalog = ProfileCatalog::with_probe(counting_probe(calls.clone()));
    catalog
        .sync_now(&test_config(&[("parent", HarnessKind::Codex)], &[]))
        .await;
    let warmed = calls.load(std::sync::atomic::Ordering::SeqCst);

    let result = spawn_backend(catalog)
        .execute_subagent_tool("parent-1".into(), spawn_request("no-such-model"))
        .await;

    assert!(result.is_error, "the spawn should have been refused");
    assert!(
        result.message.contains("does not offer \"no-such-model\""),
        "{}",
        result.message
    );
    assert_eq!(
        calls.load(std::sync::atomic::Ordering::SeqCst),
        warmed,
        "the check must not launch a discovery harness"
    );
}

/// With nothing published, spawn does not discover and does not refuse the
/// selector: the child's start follow-up checks it against the live
/// harness. Here the spawn gets past validation and fails on this fake's
/// missing sub-agent runtime instead.
#[tokio::test]
async fn spawn_over_a_cold_catalogue_does_not_discover_or_refuse_the_model() {
    let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let catalog = ProfileCatalog::with_probe(counting_probe(calls.clone()));

    let result = spawn_backend(catalog)
        .execute_subagent_tool("parent-1".into(), spawn_request("no-such-model"))
        .await;

    assert_eq!(
        calls.load(std::sync::atomic::Ordering::SeqCst),
        0,
        "a cold catalogue must not be warmed by a spawn"
    );
    assert!(
        !result.message.contains("does not offer"),
        "an unchecked selector must not be refused here: {}",
        result.message
    );
    assert!(
        result.message.contains("sub-agent creation is unavailable"),
        "the spawn should have reached the start: {}",
        result.message
    );
}

#[tokio::test]
async fn prompt_submits_one_text_block_and_returns_its_acceptance_ordinal() {
    let (submitted_tx, mut submitted) = mpsc::unbounded_channel();
    let backend = ApiBackend::new(
        SessionControl::new(FakeControl(FakeSession {
            session_id: "session-1".into(),
            accepted_ordinal: 42,
            submitted: submitted_tx,
            view: None,
        })),
        running_states(),
        Arc::new(NoExports),
    );

    let turn_id = backend
        .prompt("session-1".into(), "add a README line".into())
        .await
        .unwrap();
    assert_eq!(turn_id, 42);

    let (command_id, command) = submitted.recv().await.unwrap();
    assert!(
        command_id.starts_with("api-"),
        "command id {command_id} should name the API as its source"
    );
    let RelayCommand::Prompt { prompt } = command else {
        panic!("the API must submit a prompt command");
    };
    assert_eq!(prompt.len(), 1);
    let ContentBlock::Text(text) = &prompt[0] else {
        panic!("the API must submit the prompt as one text block");
    };
    assert_eq!(text.text, "add a README line");
}

#[tokio::test]
async fn a_finished_subagent_is_recorded_as_a_notice_not_a_prompt() {
    let (submitted_tx, mut submitted) = mpsc::unbounded_channel();
    let backend = ApiBackend::new(
        SessionControl::new(FakeControl(FakeSession {
            session_id: "parent-1".into(),
            accepted_ordinal: 7,
            submitted: submitted_tx,
            view: None,
        })),
        running_states(),
        Arc::new(NoExports),
    );

    backend
        .record_subagent_completion_notice(
            "parent-1".into(),
            "child-abcdef012345",
            "audit deps",
            3,
            "completed",
        )
        .await
        .unwrap();

    let (command_id, command) = submitted.recv().await.unwrap();
    assert!(
        command_id.starts_with("subagent-"),
        "notice command id {command_id} should name the sub-agent path"
    );
    let RelayCommand::RecordNotice { text } = command else {
        panic!("a finished sub-agent must be a notice, not {command:?}");
    };
    assert!(
        text.contains("audit deps") && text.contains("finished turn 3"),
        "unexpected notice text: {text}"
    );
    assert!(
        !text.contains("full child output"),
        "the notice must stay terse, not paste the child output: {text}"
    );
}

#[tokio::test]
async fn the_start_follow_up_configures_the_model_before_it_prompts() {
    let (submitted_tx, mut submitted) = mpsc::unbounded_channel();
    let backend = ApiBackend::new(
        SessionControl::new(FakeControl(FakeSession {
            session_id: "session-1".into(),
            accepted_ordinal: 12,
            submitted: submitted_tx,
            view: Some(ready_view("gpt-5-codex")),
        })),
        running_states(),
        Arc::new(NoExports),
    );

    backend
        .start_followup(
            "session-1".into(),
            StartFollowup {
                model: Some("gpt-5-codex".into()),
                effort: None,
                prompt: Some("add a README line".into()),
            },
        )
        .await
        .unwrap();

    let (_, first) = submitted.recv().await.unwrap();
    assert_eq!(
        first,
        RelayCommand::SetConfig {
            key: "model".into(),
            value: "gpt-5-codex".into(),
        },
        "the model must be set before the prompt runs under the old one"
    );
    let (_, second) = submitted.recv().await.unwrap();
    assert!(matches!(second, RelayCommand::Prompt { .. }));

    // The status the follow-up records is what a wait blocks on, so it has
    // to name the turn the prompt was accepted as.
    let status = loop {
        match backend.start_status("session-1".into()).await.unwrap() {
            Some(StartStatus::Pending) | None => tokio::task::yield_now().await,
            Some(status) => break status,
        }
    };
    assert_eq!(status, StartStatus::Submitted { turn_id: 12 });
}

#[tokio::test]
async fn the_start_follow_up_refuses_a_model_the_agent_does_not_offer() {
    let (submitted_tx, mut submitted) = mpsc::unbounded_channel();
    let backend = ApiBackend::new(
        SessionControl::new(FakeControl(FakeSession {
            session_id: "session-1".into(),
            accepted_ordinal: 12,
            submitted: submitted_tx,
            view: Some(ready_view("gpt-5-codex")),
        })),
        running_states(),
        Arc::new(NoExports),
    );

    backend
        .start_followup(
            "session-1".into(),
            StartFollowup {
                model: Some("no-such-model".into()),
                effort: None,
                prompt: Some("add a README line".into()),
            },
        )
        .await
        .unwrap();

    let status = loop {
        match backend.start_status("session-1".into()).await.unwrap() {
            Some(StartStatus::Pending) | None => tokio::task::yield_now().await,
            Some(status) => break status,
        }
    };
    let StartStatus::Failed { message } = status else {
        panic!("an unavailable model must fail the start, not submit the prompt");
    };
    assert!(message.contains("no-such-model"), "unexpected: {message}");
    assert!(
        submitted.try_recv().is_err(),
        "nothing may be submitted once the configuration is refused"
    );
    backend
        .set_config("session-1".into(), "model".into(), "gpt-5-codex".into())
        .await
        .unwrap();
    assert!(
        backend
            .start_status("session-1".into())
            .await
            .unwrap()
            .is_none()
    );
    assert!(matches!(
        submitted.recv().await.unwrap().1,
        RelayCommand::SetConfig { .. }
    ));
    assert!(
        submitted.try_recv().is_err(),
        "repair must not replay the abandoned initial prompt"
    );
    assert_eq!(
        backend
            .prompt("session-1".into(), "repaired prompt".into())
            .await
            .unwrap(),
        12
    );
    assert!(matches!(
        submitted.recv().await.unwrap().1,
        RelayCommand::Prompt { .. }
    ));
}

#[tokio::test]
async fn closing_cancels_the_supervised_start_before_any_prompt() {
    let (submitted_tx, mut submitted) = mpsc::unbounded_channel();
    let backend = ApiBackend::new(
        SessionControl::new(FakeControl(FakeSession {
            session_id: "session-1".into(),
            accepted_ordinal: 1,
            submitted: submitted_tx,
            view: None,
        })),
        running_states(),
        Arc::new(NoExports),
    );
    backend
        .start_followup(
            "session-1".into(),
            StartFollowup {
                prompt: Some("must not run".into()),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let task = backend
        .starts
        .lock()
        .unwrap()
        .get_mut("session-1")
        .unwrap()
        .task
        .take()
        .unwrap();
    backend.cancel_start("session-1".into()).await.unwrap();
    tokio::time::timeout(Duration::from_secs(2), task)
        .await
        .unwrap()
        .unwrap();
    assert!(submitted.try_recv().is_err());
    assert!(
        backend
            .start_status("session-1".into())
            .await
            .unwrap()
            .is_none()
    );
}

#[tokio::test]
async fn prompt_reports_a_session_the_manager_does_not_hold() {
    let (submitted_tx, _submitted) = mpsc::unbounded_channel();
    let backend = ApiBackend::new(
        SessionControl::new(FakeControl(FakeSession {
            session_id: "session-1".into(),
            accepted_ordinal: 1,
            submitted: submitted_tx,
            view: None,
        })),
        running_states(),
        Arc::new(NoExports),
    );
    let error = backend
        .prompt("session-2".into(), "hello".into())
        .await
        .unwrap_err();
    assert!(
        format!("{error:#}").contains("session-2 is not running"),
        "unexpected error: {error:#}"
    );
}

/// A checkpoint that finishes only after its requester has given up, so a
/// dropped request can be told apart from a cancelled checkpoint.
struct SlowCheckpoint {
    started: Arc<std::sync::atomic::AtomicBool>,
    finished: Arc<std::sync::atomic::AtomicBool>,
}

impl ExportRuntime for SlowCheckpoint {
    fn session_record(&self, _session_id: &str) -> Option<SessionRecord> {
        None
    }
    fn checkpoint_now(
        &self,
        _session_id: String,
    ) -> BoxFuture<'_, Result<mj_core::state::CheckpointMetadata>> {
        use std::sync::atomic::Ordering;
        Box::pin(async move {
            self.started.store(true, Ordering::Release);
            tokio::time::sleep(Duration::from_millis(50)).await;
            self.finished.store(true, Ordering::Release);
            bail!("the archive is not built in this test")
        })
    }
}

/// A client that times out drops the request future. The checkpoint behind a
/// bundle export must run to its end anyway: it has already marked the session
/// `Checkpointing` and holds a barrier on the worker, so abandoning it midway
/// left the session busy with nothing to finish or fail it, and every retry
/// refused for minutes (#1010).
#[tokio::test]
async fn a_dropped_bundle_export_request_does_not_abandon_its_checkpoint() {
    use std::sync::atomic::Ordering;

    let started = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let finished = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let exports = Arc::new(SlowCheckpoint {
        started: started.clone(),
        finished: finished.clone(),
    });

    let request = supervised_checkpoint(exports, "session-1".into());
    // Let the checkpoint start, then abandon the request the way a timed-out
    // client does.
    let abandoned = tokio::time::timeout(Duration::from_millis(5), request).await;
    assert!(abandoned.is_err(), "the checkpoint should still be running");
    assert!(started.load(Ordering::Acquire), "the checkpoint started");
    assert!(
        !finished.load(Ordering::Acquire),
        "the checkpoint was still running when the request was dropped"
    );

    tokio::time::sleep(Duration::from_millis(200)).await;
    assert!(
        finished.load(Ordering::Acquire),
        "the supervised checkpoint ran to its end without its requester"
    );
}

/// A relative export path resolves against the directory the agent runs in,
/// not the workspace root above it. For a bare project session those differ by
/// one level, which is why a file the agent had just written was refused as
/// "not in the session workspace" (#1079).
#[test]
fn a_file_export_resolves_a_relative_path_against_the_agents_directory() {
    use mj_checkpoint::checkpoint::{CheckpointRepositoryCapture, CheckpointRepositorySpec};

    // The shape `session_export_layout` builds for a bare project session whose
    // agent is launched in `/home/dev/project`.
    let bare = SessionExportLayout {
        backend: targets::TargetLocator::LocalBare {
            worker_root: "/home/dev/.local/share/hel/workers/session".into(),
        },
        workspace_root: "/home/dev".into(),
        primary_repository: "project".into(),
        repositories: vec![CheckpointRepositorySpec {
            id: "project".into(),
            relative_destination: PathBuf::from("project"),
            capture: CheckpointRepositoryCapture::MetadataOnly,
            origin_override: None,
        }],
        managed_worktree: None,
    };
    assert_eq!(
        agent_working_directory(&bare).unwrap(),
        "/home/dev/project",
        "the workspace root is the project's parent, not where the agent runs"
    );

    // The shape it builds for a bundle session, whose agent is launched in the
    // primary repository under the target's workspace directory.
    let bundle = SessionExportLayout {
        backend: targets::TargetLocator::LocalPodman {
            container_id: "hel-session".into(),
            workspace_storage: Default::default(),
            borrowed_from: None,
        },
        workspace_root: "/workspace".into(),
        primary_repository: "app".into(),
        repositories: vec![
            CheckpointRepositorySpec {
                id: "app".into(),
                relative_destination: PathBuf::from("app"),
                capture: CheckpointRepositoryCapture::RemoteWorkspace,
                origin_override: None,
            },
            CheckpointRepositorySpec {
                id: "lib".into(),
                relative_destination: PathBuf::from("lib"),
                capture: CheckpointRepositoryCapture::RemoteWorkspace,
                origin_override: None,
            },
        ],
        managed_worktree: None,
    };
    assert_eq!(agent_working_directory(&bundle).unwrap(), "/workspace/app");

    // What the target is actually asked to read. The one-repository layout is
    // bounded by the agent's own directory, so the path it is handed is the one
    // the caller typed.
    assert_eq!(
        export_root_and_path(&bare, Path::new("secret.txt")).unwrap(),
        ("/home/dev/project".to_owned(), "secret.txt".to_owned())
    );
    // The bundle is bounded by the workspace root the repositories share, so
    // the path is rewritten to start at the primary repository.
    assert_eq!(
        export_root_and_path(&bundle, Path::new("src/main.rs")).unwrap(),
        ("/workspace".to_owned(), "app/src/main.rs".to_owned())
    );

    // A secondary repository sits beside the primary one under the workspace
    // root, so `..` reaches it. Before paths resolved in the agent's directory
    // this was `lib/README.md`; it must not have become unreachable (#1079).
    assert_eq!(
        export_root_and_path(&bundle, Path::new("../lib/README.md")).unwrap(),
        ("/workspace".to_owned(), "lib/README.md".to_owned())
    );
    assert_eq!(
        export_root_and_path(&bundle, Path::new("./src/../src/main.rs")).unwrap(),
        ("/workspace".to_owned(), "app/src/main.rs".to_owned())
    );

    // Above the workspace root is refused, and the refusal names both the
    // boundary and the directory the path was resolved in.
    for escape in ["../../etc/passwd", "../.."] {
        let ExportError::Refused(message) =
            export_root_and_path(&bundle, Path::new(escape)).unwrap_err()
        else {
            panic!("{escape} must be refused, not attempted");
        };
        assert!(
            message.contains("/workspace") && message.contains("/workspace/app"),
            "the refusal names the boundary and the directory searched: {message}"
        );
    }

    // A one-repository layout has no sibling to reach, and its workspace root
    // is the parent directory holding the user's other projects, not a boundary
    // Hel owns. `..` stops at the agent's directory there.
    let ExportError::Refused(message) =
        export_root_and_path(&bare, Path::new("../other-project/.env")).unwrap_err()
    else {
        panic!("a single-repository layout must not reach outside its own directory");
    };
    assert!(
        message.contains("/home/dev/project"),
        "the refusal names the boundary: {message}"
    );
    assert!(
        !message.contains("/home/dev/other-project"),
        "the refusal does not suggest the path was looked for: {message}"
    );

    // The boundary itself is not a file inside it.
    for (layout, path) in [(&bundle, ".."), (&bare, ".")] {
        assert!(matches!(
            export_root_and_path(layout, Path::new(path)),
            Err(ExportError::Refused(_))
        ));
    }
}