trusty-mpm 1.0.2

//! Transport-agnostic session-lifecycle core shared by the HTTP routes and the
//! MCP tools (#1221).
//!
//! Why: both the HTTP `…/managed/*` handlers (in this module's `mod.rs`) and the
//! MCP session-lifecycle tools (in `crate::daemon::mcp_session`) must spawn and
//! resume managed sessions with IDENTICAL semantics. Keeping the spawn/resume
//! flow here — rather than duplicating it per transport — guarantees they cannot
//! diverge, and keeps the route file under the 500-SLOC production cap.
//! What: the `SpawnParams` input struct, the `spawn_managed` flow
//! (provision → create tmux host → launch harness), the typed `ResumeManagedError`
//! (so callers map failures to HTTP 404/409/500 by VARIANT, never by `Display`
//! substring), and the `resume_managed` flow (single-round-trip resume + respawn,
//! no TOCTOU pre-flight `get`).
//! Test: `spawn_managed`/`resume_managed` are exercised by the HTTP spawn/resume
//! handler tests and the MCP `session_new_invalid_runtime_errors` /
//! `session_resume_unknown_id_errors` tests, plus the typed-error regression
//! tests `resume_managed_typed_*` in tests/session_manager_mvp.rs.

use std::sync::Arc;

use tracing::{info, warn};

use super::deployment_check::ensure_deployment_complete;
// `carrier_reachable`/`warn_if_no_persona_carrier` are only called from
// within `deployment_check` itself in non-test code; `lifecycle_tests.rs`
// (a child module, `use super::*`) still exercises them directly, so they
// are imported here ONLY under `#[cfg(test)]` to avoid an unused-import
// warning on the production build.
#[cfg(test)]
use super::deployment_check::{carrier_reachable, warn_if_no_persona_carrier};
use super::inproject::try_inproject_spawn;
use super::resume_error::ResumeManagedError;
use crate::core::git_identity::resolve_for_config_enforced;
use crate::daemon::state::DaemonState;
use crate::provisioner::WorkspaceProvisioner;
use crate::runtime::{RuntimeKind, build_adapter};
use crate::session_manager::{ManagedSessionId, ManagedSessionState, SessionRecord};
// `ManagedError` moved out of this file's non-test code with the #2577 review
// enum split (see `resume_error.rs`); it is still used by this module's own
// `#[cfg(test)]` `StubTmux` double below.
#[cfg(test)]
use crate::session_manager::ManagedError;

/// Resolve the #3025 `GH_TOKEN`/`GH_USER` spawn-env override for `cwd`,
/// consulting the daemon's shared `ProjectRegistry` (review follow-up — the
/// actual persistence target for a pinned `gh_account`, not the static
/// config file).
///
/// Why: every `adapter.spawn`/`spawn_resume` call site below needs the
/// identical "get the registry, resolve for this workspace" sequence;
/// centralising it here keeps each call site a single `.await` line and
/// guarantees they cannot diverge.
/// What: delegates to `core::gh_account::resolve_gh_account_env_for_registry`,
/// which runs its blocking work (git origin probe, `gh auth token`) via
/// `tokio::task::spawn_blocking` — safe to call directly from these async
/// handlers without stalling the executor.
/// Test: exercised indirectly via the existing `handler_spawn_*`/
/// `resume_managed_*` tests in `tests/session_manager_mvp.rs` (empty-registry
/// case — no regression); the registry-matching logic itself is unit-tested
/// in `core::gh_account_spawn_env_tests`.
pub(super) async fn resolve_gh_env(
    state: &Arc<DaemonState>,
    cwd: &std::path::Path,
) -> Vec<(String, String)> {
    let registry = state.project_registry().await;
    crate::core::gh_account::resolve_gh_account_env_for_registry(&registry, cwd).await
}

/// Transport-agnostic inputs for spawning a managed session.
///
/// Why: both the HTTP `POST /…/managed` handler and the MCP `session_new` tool
/// need to spawn a session with the same semantics; a shared struct lets one
/// [`spawn_managed`] function serve both without the MCP path re-implementing
/// the provision→create→spawn ritual.
/// What: the same fields as `super::SpawnRequest` but plain owned types (no
/// axum/serde extraction), so non-HTTP callers can build it directly.
/// Test: `spawn_managed` is exercised via `crate::daemon::mcp_session`'s
/// `session_new_invalid_runtime_errors` and the HTTP spawn tests.
#[derive(Debug, Clone)]
pub struct SpawnParams {
    /// Repository URL to provision the session workspace from.
    pub repo_url: String,
    /// Git branch or ref to check out.
    pub git_ref: String,
    /// Human-readable task description for the session.
    pub task: String,
    /// Optional name hint overriding the auto-generated tmux session name.
    pub name_hint: Option<String>,
    /// Optional runtime selector (`"claude-code"` | `"tcode"`).
    pub runtime: Option<String>,
    /// Whether the spawned session is EPHEMERAL (a test/throwaway session) (#1508).
    ///
    /// Why: e2e harnesses (and any caller that knows it is creating a disposable
    /// session) set this so the bulk-teardown and age-based reap paths may clean
    /// the session up automatically. `None`/`Some(false)` → a normal, durable
    /// session that the automatic paths never touch.
    pub ephemeral: Option<bool>,
    /// Whether this spawn originates from the MCP tool surface (#1836, #1837).
    ///
    /// Why: an MCP-triggered `session_new` call can mint real infrastructure
    /// for any repo an LLM caller names, with zero operator confirmation (the
    /// ARIA incident). `true` subjects the spawn to the two-layer MCP spawn
    /// gate (off-by-default + registry allowlist,
    /// [`super::mcp_spawn_gate::ensure_mcp_spawn_allowed`]) BEFORE any
    /// provisioning begins. `false` — set by the HTTP route (`tm launch`/`tm
    /// ticket` clients) and the SM-STDIO adapter (`sm.sessions.launch`) — never
    /// gates; those are trusted, explicitly-operator-driven paths.
    pub mcp_initiated: bool,
    /// Whether to AUTO-INJECT `task` into the spawned pane once the runtime is
    /// ready (issues #1903 / #1299).
    ///
    /// Why: `--task` is meant to be turnkey — the session should start working
    /// on the task immediately after spawn. `None`/`Some(true)` inject (the
    /// default, so `tm session new`, `tm ticket`, and MCP/SM callers are all
    /// turnkey without opting in). `Some(false)` selects the legacy
    /// metadata-only behavior (`tm session new --no-inject`), for callers that
    /// deliver the task themselves via `tm session send`. Injection is
    /// additionally gated by [`crate::session_manager::should_inject_task`]
    /// (non-empty task, Claude Code runtime, session reached `Active`).
    pub inject_task: Option<bool>,
    /// Optional Deliverable id to bind this session to (DOC-35 §10.6, #2379).
    ///
    /// Why: `tm sessions new --deliverable <id>` links a session to a
    /// Deliverable so its status/ls output can surface which unit of work the
    /// session is advancing (§10.6: 1 Deliverable ↔ many Sessions). Carried as
    /// the raw stringified id here — the HTTP route (`spawn_session`) is
    /// responsible for validating the Deliverable exists AND belongs to the
    /// spawning project BEFORE any provisioning side effect (a 404-style
    /// [`crate::daemon::error::DaemonError::DeliverableNotFound`] otherwise);
    /// [`spawn_managed`] trusts that pre-validation and just persists the
    /// pointer post-creation via
    /// [`crate::session_manager::SessionManager::set_deliverable_id`]. A
    /// malformed or absent id is silently a no-op link (`None`) — never a
    /// reason to fail an otherwise-successful spawn.
    pub deliverable_id: Option<String>,
    /// Whether to FORCE a brand-new session even when a live in-project session
    /// for the same project already exists (#2450).
    ///
    /// Why: the #1707 in-project reconnect pre-flight silently ADOPTS an
    /// existing live session for the same `source_id` (returning it instead of
    /// spawning). A surface that explicitly means "launch a NEW session" — the
    /// `tm` picker's "[N] launch new session" choice, `tm session new`/`session
    /// start` — would otherwise have its `--task` injected into an unrelated
    /// live session. `true` SKIPS the reconnect pre-flight (always spawns
    /// fresh); `false` (the default, so programmatic/idempotent callers — MCP
    /// `session_new`, the SM-STDIO adapter, the chat surfaces — keep
    /// reconnecting) preserves #1707.
    pub force_new: bool,
}

/// Spawn a managed session, shared by the HTTP handler and the MCP tool.
///
/// Why: the spawn flow (resolve runtime → provision workspace → create tmux host
/// → launch harness) must be identical across transports; centralising it here
/// means the MCP `session_new` tool is a true thin wrapper rather than a
/// divergent copy.
/// What: in order — (0) parses the runtime selector (an unknown value is an early
/// `Err` before any side effect); (1) provisions an isolated workspace; (2)
/// creates the tmux session rooted at that workspace; (3) spawns the selected
/// runtime in the pane (a spawn failure marks the record errored but is not
/// fatal — the record still exists). Returns the final [`SessionRecord`].
/// Test: `crate::daemon::mcp_session::tests::session_new_invalid_runtime_errors`
/// covers the early runtime-rejection path; the HTTP spawn tests cover the
/// provision/create/spawn path.
///
/// #2605: the session id is now a CALLER-SUPPLIED parameter rather than minted
/// internally, so the asynchronous-spawn path (`background: true`) can learn it
/// BEFORE provisioning starts — it returns that id to the client immediately
/// (to poll `provision-status`), keys the
/// [`crate::daemon::provisioning::ProvisioningRegistry`] on it, and correlates
/// the `provisioning_stage` SSE frames the background task streams by it. NOTE:
/// on the reconnect-to-existing-session outcome (#1707) the returned record's
/// id may DIFFER from `session_id`; callers that keyed anything on `session_id`
/// must read the final id off the returned record. Synchronous callers simply
/// pass a fresh [`ManagedSessionId::new()`].
pub async fn spawn_managed(
    state: &Arc<DaemonState>,
    session_id: ManagedSessionId,
    params: SpawnParams,
) -> Result<SessionRecord, String> {
    // Config is loaded ONCE here and reused below for both the MCP spawn gate
    // and the workspace-root resolution, so the two cannot read divergent
    // snapshots of `config.yaml` within a single spawn.
    let config = crate::core::trusty_tools_config::TrustyToolsConfig::load();

    // Step 0: resolve the runtime backend (default claude-code). Reject unknown
    // selectors BEFORE any provisioning so a typo never leaves an orphan
    // workspace.
    let runtime = match params.runtime.as_deref() {
        None => RuntimeKind::default(),
        Some(raw) => raw.parse::<RuntimeKind>().map_err(|e| e.to_string())?,
    };

    // Step 0.4 — MCP SPAWN GATE (#1836, #1837): an MCP-triggered spawn (never a
    // `tm launch`/`tm ticket`/SM-STDIO call — see `SpawnParams::mcp_initiated`)
    // must be refused BEFORE any side effect when MCP spawning is disabled
    // (the default) or the target repo is not an already-known project. This
    // runs before `ManagedSessionId::new()` so a refusal mints nothing.
    if params.mcp_initiated {
        let registry = state.project_registry().await;
        super::mcp_spawn_gate::ensure_mcp_spawn_allowed(&registry, &config, &params.repo_url)
            .await?;
    }

    // Wrap the ENTIRE spawn dispatch — in-project, local-path, AND clone-based
    // — in a `provisioning_stage` scope (issue #1904 stretch goal; #1919 fix).
    // Before #1919 only the clone-based tail (`spawn_managed_cloned`) was
    // wrapped here, so the `is_local_workdir` branch below — covering BOTH
    // `spawn_managed_inproject` and `spawn_managed_local` — returned before
    // this scope was ever installed. Since #1916 unified `tm session start`
    // onto `spawn_managed_inproject`, that branch is now the dominant path,
    // so every `emit(...)` call anywhere in ITS call tree (including
    // `try_inproject_spawn`'s `ensure_base_clone` first-run clone, several
    // layers deep in `inproject.rs`) now also broadcasts on the daemon's
    // existing SSE channel — no signature changes needed anywhere in that
    // call tree, per `core::provisioning_stage`'s module doc. The client
    // correlates by `repo_url` (it cannot know `session_id` until this whole
    // call returns).
    let emitter = crate::core::provisioning_stage::StageEmitter::new(
        session_id.to_string(),
        params.repo_url.clone(),
        state.event_tx.clone(),
    );
    // Capture the injection opt-out and the (already HTTP-layer-validated)
    // Deliverable link before `params` is moved into the routed dispatch.
    // inject_task: `None`/`Some(true)` → turnkey (inject); `Some(false)` →
    // metadata-only (`--no-inject`).
    let inject_flag = params.inject_task != Some(false);
    let deliverable_id = params.deliverable_id.clone();
    let mut record = crate::core::provisioning_stage::scoped(
        emitter,
        spawn_managed_routed(state, session_id, params, runtime, config),
    )
    .await?;

    // Deliverable linkage (DOC-35 §10.6, #2379): the HTTP route already
    // validated `deliverable_id` exists and belongs to this project BEFORE
    // provisioning started, so this is a trusting, non-fatal persist of a
    // PURE POINTER — never a reason to fail an otherwise-successful spawn,
    // and never a mutation of the Deliverable itself (§11 forbids
    // auto-transitions). The local `record` is updated too so the immediate
    // spawn response already reflects the link, not just later `GET`s.
    if let Some(raw_id) = deliverable_id {
        match raw_id.parse::<crate::deliverable::DeliverableId>() {
            Ok(did) => match state
                .session_manager()
                .await
                .set_deliverable_id(&record.id, did)
                .await
            {
                Ok(()) => record.deliverable_id = Some(did),
                Err(e) => warn!(
                    id = %record.id,
                    "spawn_managed: set_deliverable_id failed: {e}; deliverable link not recorded"
                ),
            },
            Err(e) => warn!(
                id = %record.id,
                "spawn_managed: deliverable_id {raw_id:?} failed to re-parse (already validated by the HTTP layer): {e}"
            ),
        }
    }

    // Turnkey task injection (#1903 / #1299): once the pane is up and the
    // runtime is ready, deliver the task through the same seam `tm session send`
    // uses, so `tm session new --task` / `tm ticket` start working immediately
    // instead of sitting idle at an empty prompt.
    spawn_task_injection(state.clone(), record.clone(), inject_flag);

    Ok(record)
}

/// Kick off turnkey `--task` injection for a freshly-spawned session (#1903 / #1299).
///
/// Why: `--task` must be turnkey — after the runtime launches, the task is typed
/// into the pane so the session starts working immediately. Delivery has to WAIT
/// for the runtime to be ready (keystrokes sent before `claude` execs are lost),
/// which can take a few seconds; blocking the spawn response on that would make
/// every caller wait, so this runs in a detached background task (mirroring
/// `daemon::services::session_service::spawn_pid_capture`).
/// What: gates on [`crate::session_manager::should_inject_task`] (opt-out flag,
/// non-empty task, Claude Code runtime — `tcode` injects the task in its own
/// launch command — and a session that reached `Active`); when it passes, spawns
/// a Tokio task that awaits
/// [`crate::session_manager::SessionManager::inject_task_when_ready`]. A skip or
/// failure is non-fatal: the task remains on the record as metadata, deliverable
/// via `tm session send`.
/// Test: the gate is unit-tested in `session_manager::task_inject`; the delivery
/// via the fake seam is covered there too. This wiring is side-effect-only
/// (fire-and-forget) and exercised by the live `#[ignore]` e2e flow.
fn spawn_task_injection(state: Arc<DaemonState>, record: SessionRecord, inject_flag: bool) {
    if !crate::session_manager::should_inject_task(
        inject_flag,
        &record.task,
        record.runtime,
        &record.state,
    ) {
        return;
    }
    tokio::spawn(async move {
        let mgr = state.session_manager().await;
        match mgr.inject_task_when_ready(&record.id, &record.task).await {
            Ok(true) => {}
            Ok(false) => warn!(
                id = %record.id,
                name = %record.tmux_name,
                "turnkey task injection skipped (runtime never became ready); task retained as metadata"
            ),
            Err(e) => warn!(
                id = %record.id,
                name = %record.tmux_name,
                "turnkey task injection failed: {e}; task retained as metadata"
            ),
        }
    });
}

/// Route a spawn to the in-project, local-path, or clone-based branch (#1919).
///
/// Why: extracted from [`spawn_managed`] so the `provisioning_stage::scoped`
/// wrapper installed there covers ALL THREE spawn branches uniformly — this
/// function is exactly the routing logic `spawn_managed` used to run inline
/// after `is_local_workdir` detection, before #1919 moved the scope to cover it.
/// What: local-path fast path (#1433) — if `repo_url` is an existing local
/// absolute directory, tries in-project DETECTION (#1706, via
/// `try_inproject_spawn`, which since #2032 only ensures the base clone and
/// returns `(base_path, owner, repo)` — it no longer creates the worktree
/// itself). On a match, this function resolves the SEMANTIC tmux name here —
/// BEFORE the per-session worktree exists — via
/// `SessionManager::resolve_session_name`, folding in
/// `inproject::worktree_name_collides` as the extra collision predicate so a
/// name whose worktree dir/branch already exists is retried with the next
/// free serial instead of silently colliding. It then creates the worktree at
/// that resolved name (`inproject::create_session_worktree`) and dispatches to
/// `spawn_managed_inproject`, threading the resolved name through so
/// `create_with_id` never re-derives it (issue #2032 — a session's tmux name
/// is derived in exactly ONE place). Any failure in detection, name
/// resolution, or worktree creation falls through to `spawn_managed_local`. A
/// remote repo URL (the common case) falls straight through to
/// `spawn_managed_cloned`.
/// Test: exercised transitively by the same tests that covered the inline
/// version before extraction (HTTP spawn tests, MCP session tests); the
/// stage-emission behaviour added by #1919 is covered by
/// `prepare_inproject_session_emits_stage_events_in_order` in this module's
/// `tests` submodule and `ensure_base_clone_emits_cloning_repo_only_on_fresh_clone`
/// in `inproject::tests`.
async fn spawn_managed_routed(
    state: &Arc<DaemonState>,
    session_id: ManagedSessionId,
    params: SpawnParams,
    runtime: RuntimeKind,
    config: crate::core::trusty_tools_config::TrustyToolsConfig,
) -> Result<SessionRecord, String> {
    // Detection heuristic (documented): the string is an ABSOLUTE path that EXISTS
    // and is a DIRECTORY on the daemon host. Anything else (a `https://…` /
    // `git@…` URL, a relative path, a non-existent path) falls through to the
    // existing clone-based provisioning below, so remote-URL callers are unaffected.
    if is_local_workdir(&params.repo_url) {
        // In-project path (#1706): if the local directory is a git repo with a
        // GitHub remote, spawn against a per-session worktree of a protected base
        // clone rather than using the directory directly. If it is NOT a git repo
        // with a GitHub remote (no `.git`, no remote origin, non-GitHub URL), fall
        // through to the existing local-path spawn.
        let local_path = std::path::Path::new(&params.repo_url);

        // #3455 "launch on main" opt-out: checked BEFORE `try_inproject_spawn`
        // (which would otherwise establish a base clone as a side effect even
        // when it's never going to be used) so a project with worktree
        // isolation disabled never gets a base clone OR a worktree at all —
        // the exact wasted-disk-space complaint the issue raises. Only a
        // GitHub-remote-having local checkout can match a registered project
        // (mirrors `try_inproject_spawn`'s own detection).
        if let Some(origin_url) = super::inproject::get_origin_url(local_path)
            && let Some(gh) = trusty_common::github_path::parse_github_path(&origin_url)
        {
            let registry = state.project_registry().await;
            if !super::launch_on_main::worktree_enabled_for_origin(&registry, &origin_url).await {
                info!(
                    id = %session_id,
                    path = %local_path.display(),
                    "spawn_managed: project has worktree isolation disabled (#3455); \
                     launching directly in the main checkout"
                );
                // Same reconnect pre-flight the worktree branch runs below
                // (#1707 + `force_new` opt-out, #2450): a live session for
                // this repo reconnects instead of spawning a duplicate.
                {
                    let mgr = state.session_manager().await;
                    let existing = mgr.list().await;
                    let source_id = format!("{}/{}", gh.owner, gh.repo);
                    if let Some(live) = reconnect_candidate(
                        params.force_new,
                        &existing,
                        &source_id,
                        &*mgr.tmux_driver(),
                    ) {
                        info!(
                            id = %live.id,
                            source_id = %source_id,
                            "spawn_managed (launch-on-main): reconnecting to existing live session"
                        );
                        return Ok(live);
                    }
                }
                return super::launch_on_main::spawn_managed_on_main(
                    state,
                    &session_id,
                    &params,
                    runtime,
                    local_path,
                    &gh.owner,
                    &gh.repo,
                )
                .await;
            }
        }

        match try_inproject_spawn(local_path) {
            Ok(Some((base, owner, repo))) => {
                // Reconnect pre-flight (#1707), HOISTED AHEAD of worktree
                // reservation (#2450): the reconnect used to live inside
                // `spawn_managed_inproject`, which meant `reserve_inproject_worktree`
                // had already created (then left for pruning) a fresh
                // `.worktrees/<name>` slice before the reconnect returned the
                // existing session — worktree litter, including a locked orphan.
                // Running the check here, before any worktree is reserved, means
                // a reconnect creates ZERO worktrees. Gated by `force_new` so an
                // explicit "launch new" surface always spawns fresh instead of
                // adopting an unrelated live session for the same project.
                {
                    let mgr = state.session_manager().await;
                    let existing = mgr.list().await;
                    let source_id = format!("{owner}/{repo}");
                    if let Some(live) = reconnect_candidate(
                        params.force_new,
                        &existing,
                        &source_id,
                        &*mgr.tmux_driver(),
                    ) {
                        info!(
                            id = %live.id,
                            source_id = %source_id,
                            "spawn_managed (inproject): reconnecting to existing live session"
                        );
                        return Ok(live);
                    }
                }
                match reserve_inproject_worktree(
                    state,
                    &session_id,
                    &params,
                    local_path,
                    &base,
                    &repo,
                    &config,
                )
                .await
                {
                    Ok((worktree, reserved_name)) => {
                        return spawn_managed_inproject(
                            state,
                            &session_id,
                            &params,
                            runtime,
                            worktree,
                            owner,
                            repo,
                            reserved_name,
                        )
                        .await;
                    }
                    Err(e) => {
                        tracing::warn!(
                            id = %session_id,
                            "in-project spawn: {e}; falling back to local-path spawn"
                        );
                    }
                }
            }
            Ok(None) => {
                // Not a git repo with a GitHub remote — use local-path fast path.
            }
            Err(e) => {
                tracing::warn!(
                    id = %session_id,
                    "in-project spawn failed: {e}; falling back to local-path spawn"
                );
            }
        }
        return spawn_managed_local(state, &session_id, &params, runtime).await;
    }

    spawn_managed_cloned(state, session_id, params, runtime, config).await
}

/// Resolve the semantic tmux name and create the per-session worktree for the
/// in-project spawn path (issue #2032).
///
/// Why: split out of [`spawn_managed_routed`] so the name-resolution-then-
/// worktree-creation sequence — the actual #2032 hoist — reads as one clear
/// step, and so its two fallible sub-steps (name resolution, worktree
/// creation) share one error-formatting path back to the caller's
/// fall-through-to-local-path handling.
/// What: (1) resolves a collision-free tmux name via
/// `SessionManager::resolve_session_name`, using `params.name_hint`, the
/// detected repo name as the GitHub-project fallback (`cwd` is passed only as
/// the third-priority basename fallback — it is never actually reached here
/// since the repo name is always `Some`), and
/// `inproject::worktree_name_collides(base, candidate)` as the extra
/// collision predicate (so a name whose `.worktrees/<name>` dir or
/// `session/<name>` branch already exists is retried with the next serial,
/// not silently reused); (2) creates the worktree at that name via
/// `inproject::create_session_worktree`; (3) best-effort syncs the
/// operator's allowlisted untracked/secret files (`.env*` by default, #2196)
/// from `local_path` (the operator's live checkout) into the new worktree via
/// [`super::inproject::untracked_sync::sync_untracked_files`] — resolved
/// per-project-over-global-over-default via
/// `trusty_tools_config::resolve_untracked_sync(config, Some(repo))`, and
/// SKIPPED entirely (not even attempted) when resolution says `enabled ==
/// false`. Step 3 can never fail this function: sync failures are `warn!`-
/// logged inside `sync_untracked_files` itself and never propagate. Returns
/// `(worktree_path, reserved_name)` so the caller can create the tmux
/// session under the SAME name without re-deriving it.
/// Test: exercised transitively by the in-project spawn integration tests;
/// the sync step's own behaviour (matching, size cap, path-escape guard,
/// `.git/info/exclude` append) is unit-tested directly in
/// `inproject::untracked_sync::tests`.
async fn reserve_inproject_worktree(
    state: &Arc<DaemonState>,
    session_id: &ManagedSessionId,
    params: &SpawnParams,
    local_path: &std::path::Path,
    base: &std::path::Path,
    repo: &str,
    config: &crate::core::trusty_tools_config::TrustyToolsConfig,
) -> Result<(std::path::PathBuf, String), String> {
    let mgr = state.session_manager().await;
    let reserved_name = mgr
        .resolve_session_name(
            params.name_hint.as_deref(),
            Some(repo),
            local_path,
            |candidate| super::inproject::worktree_name_collides(base, candidate),
        )
        .await
        .map_err(|e| format!("name resolution failed for session {session_id}: {e}"))?;

    let worktree = super::inproject::create_session_worktree(base, &reserved_name, session_id)
        .map_err(|e| format!("worktree creation failed for session {session_id}: {e}"))?;

    // #2196: best-effort sync of the operator's allowlisted untracked/secret
    // files (default `.env*`) from the live checkout into the fresh
    // worktree. Never fails the spawn — see the doc above.
    let resolved_sync =
        crate::core::trusty_tools_config::resolve_untracked_sync(config, Some(repo));
    if resolved_sync.enabled {
        super::inproject::untracked_sync::sync_untracked_files(
            local_path,
            &worktree,
            &resolved_sync.patterns,
        );
    }

    Ok((worktree, reserved_name))
}

/// The clone-based provisioning tail of [`spawn_managed_routed`] (issue #1904).
///
/// Why: split out so the (now-shared, #1919) stage-observer scope installed
/// in `spawn_managed` has a clean per-branch extraction to call — mirroring
/// `spawn_managed_inproject`/`spawn_managed_local`, which get the identical
/// scope for free since #1919 moved the `scoped(...)` wrapper up to cover all
/// three branches uniformly (before #1919 this was the ONLY branch it wrapped).
/// What: provisions the workspace (emits `CloningRepo`/`DeployingAgents`/
/// `DeployingSkills`/`BuildingInstructions`/`ConfiguringMcp` from deep inside
/// `provision`/`provision_in`/`prepare_session_inner`), creates the tmux
/// session (`CreatingTmuxSession`), runs the FRONT gate, spawns the runtime
/// (`LaunchingRuntime`), and emits `Complete`. Identical behaviour to the
/// pre-#1904 inline tail of `spawn_managed` — this is a pure extraction plus
/// `emit(...)` calls.
/// Test: `handler_spawn_wires_provision_and_spawn` /
/// `handler_spawn_creates_tmux_at_workspace_cwd` in tests/session_manager_mvp.rs
/// cover the behaviour; the stage-emission itself is unit-tested in
/// `core::provisioning_stage` and `provisioner::workspace`/`session_launch`
/// (no live daemon needed for the emission tests).
async fn spawn_managed_cloned(
    state: &Arc<DaemonState>,
    session_id: ManagedSessionId,
    params: SpawnParams,
    runtime: RuntimeKind,
    config: crate::core::trusty_tools_config::TrustyToolsConfig,
) -> Result<SessionRecord, String> {
    use crate::core::provisioning_stage::{ProvisioningStage, emit};

    // #2184: resolve the per-project gh identity + commit identity ONCE for
    // this spawn (project `github:`/commit fields in `config.projects` >
    // global `github:` > fully ambient) and apply it to every git subprocess
    // the provisioner runs below. #3312: also verifies/self-heals the #2081
    // account pairing inside the resolved `config_dir` — see
    // `core::git_identity` module docs. Surfaced as a spawn failure BEFORE
    // any provisioning side effect, mirroring how the MCP spawn gate above
    // already fails closed before touching disk.
    let git_identity = resolve_for_config_enforced(&config, &params.repo_url)
        .await
        .map_err(|e| {
            warn!(id = %session_id, "spawn_managed: git identity resolution failed: {e}");
            format!("git identity resolution failed: {e}")
        })?;

    // Provision an isolated workspace under the pre-generated `session_id` (the id
    // is generated ONCE in `spawn_managed`, before the local-path/clone branch
    // split, so both branches register the same id).
    //
    // #1220: the workspace root defaults to `~/trusty-mpm-projects/` (overridable
    // via the `TRUSTY_MPM_WORKSPACE_ROOT` env var or the
    // `~/.trusty-tools/trusty-mpm/config.yaml` `workspace_root_template`), and the
    // session nests under the target repo's GitHub `<owner>/<repo>` identity:
    // `<root>/<owner>/<repo>/<session-id>/`. When the repo URL has no parseable
    // GitHub identity we fall back to the legacy single-slug `provision` path so a
    // bare/non-GitHub URL still provisions cleanly. `config` was already loaded
    // in `spawn_managed` (before the MCP spawn gate); reused here for the
    // workspace root. `CloningRepo`/`DeployingAgents`/`DeployingSkills`/
    // `BuildingInstructions`/`ConfiguringMcp` are emitted from inside `provision`/
    // `provision_in`/`prepare_session_inner` — not here — since those are the
    // functions that actually perform each step.
    let prepared = match trusty_common::github_path::parse_github_path(&params.repo_url) {
        Some(gh) => {
            let project_dir = crate::core::trusty_tools_config::workspace_subpath(&config, &gh);
            // `provision_in` only appends the session id; pass an empty workspace
            // root because the project dir is already absolute.
            let provisioner = WorkspaceProvisioner::new(
                crate::provisioner::RealGitBackend::new(git_identity),
                std::path::PathBuf::new(),
            );
            provisioner.provision_in(
                &project_dir,
                &session_id,
                &params.repo_url,
                &params.git_ref,
                &params.task,
            )
        }
        None => {
            let workspace_root = crate::core::trusty_tools_config::workspace_root(&config);
            let provisioner = WorkspaceProvisioner::new(
                crate::provisioner::RealGitBackend::new(git_identity),
                workspace_root,
            );
            provisioner.provision(&session_id, &params.repo_url, &params.git_ref, &params.task)
        }
    }
    .map_err(|e| {
        warn!(id = %session_id, "spawn_managed: provision failed: {e}");
        format!("workspace provisioning failed: {e}")
    })?;

    // Step 2: create the tmux session rooted at the provisioned workspace.
    // #1935: `owned=false` — the workspace is now a `git worktree` slice of a
    // shared, persistent base checkout (`<project_dir>/.base/`), not an
    // independently-owned full clone. Bulk `remove_dir_all` would leave the
    // base checkout's git worktree metadata and session branch ref dangling;
    // `session_manager::decommission` instead detects the `.worktrees/<id>`
    // shape (`is_session_worktree`) and runs `git worktree remove --force` +
    // branch cleanup via `remove_session_worktree`, mirroring exactly how the
    // in-project spawn path (`spawn_managed_inproject`, below) already handles
    // its own per-session worktrees.
    emit(ProvisioningStage::CreatingTmuxSession);
    let mgr = state.session_manager().await;
    let record = mgr
        .create_with_id(
            session_id,
            params.task.clone(),
            Some(prepared.path.clone()),
            params.name_hint,
            Some(prepared.path.clone()),
            Some(params.repo_url.clone()),
            Some(params.git_ref.clone()),
            runtime,
            params.ephemeral.unwrap_or(false),
            false, // owned: false — worktree of a shared base checkout, not a full clone
        )
        .await
        .map_err(|e| {
            warn!(id = %session_id, "spawn_managed: session create failed: {e}");
            e.to_string()
        })?;

    // #3649: best-effort; see `worktree_ownership::set_worktree_owner_best_effort`.
    mgr.set_worktree_owner_best_effort(&session_id, session_id)
        .await;

    // #3822: explicit spawn-time project registration — the one-shot
    // `auto_register_from_sessions` boot pass (`ProjectRegistry::
    // auto_register_from_sessions`) never sees a session created after the
    // registry's first touch, so `tm project list` reported this project
    // missing forever without this. Best-effort; never blocks the spawn.
    state
        .project_registry()
        .await
        .register_from_session(&record)
        .await;

    // Step 2.5 — INTENT-CONFORMANCE FRONT GATE (#1360, spec §5.1).
    //
    // Between record-creation and `adapter.spawn`, resolve the ticket+spec intent
    // for this task and decide whether to auto-proceed or escalate BEFORE any code
    // is written. The gate is fail-open (non-ticketed work, an unresolved ISR, or
    // a gap all auto-proceed) so it can never be the reason a spawn stalls. On an
    // escalation it withholds the spawn, writes `pending_decision`/`proposed_default`
    // (surfaced through every existing channel), and leaves the session in its
    // pre-spawn `Provisioning` state until a human resolves it via `POST …/answer`.
    if let Some(record) =
        front_gate_or_escalate(&mgr, &record, &params.repo_url, &params.task).await?
    {
        return Ok(record);
    }

    if let Err(e) = mgr
        .set_workspace(
            &record.id,
            prepared.path.clone(),
            ManagedSessionState::Active,
        )
        .await
    {
        warn!(id = %record.id, "spawn_managed: set_workspace failed: {e}");
    }

    // Deployment-completeness check (#2158, made non-blocking by #2172): best-
    // effort auto-repair of an incomplete `.claude/` payload. #2171 tracks the
    // validator over-reporting INCOMPLETE; until that lands, a false positive
    // here must never withhold the runtime launch (P0: it was leaving every
    // session at a bare shell). Any gap — real or falsely reported — is now
    // logged and the session proceeds to `adapter.spawn` regardless.
    let fw = crate::core::paths::FrameworkPaths::for_managed_workspace(&prepared.path);
    if let Err(reason) =
        ensure_deployment_complete(&fw, &prepared.path, record.repo_url.as_deref(), &record.id)
    {
        warn!(id = %record.id, "spawn_managed: deployment incomplete after auto-repair (non-blocking, launch proceeds): {reason}");
    }

    // Workstream-activity label ensure (issue #3726): idempotent, best-effort,
    // detached — see `session_launch::spawn_workstream_label_ensure`.
    crate::core::session_launch::spawn_workstream_label_ensure(
        record.repo_url.clone(),
        prepared.path.clone(),
        record.tmux_name.clone(),
    );

    // Step 3: spawn the selected runtime in the pane. A spawn failure is recorded
    // (the record is marked errored) but is not fatal — the record exists and the
    // caller still gets it back.
    emit(ProvisioningStage::LaunchingRuntime);
    let tmux_arc = mgr.tmux_driver();
    let adapter = build_adapter(record.runtime, tmux_arc);
    let gh_env = resolve_gh_env(state, &prepared.path).await;
    if let Err(e) = adapter.spawn(
        &record.tmux_name,
        &prepared.path,
        &params.task,
        &record.id.to_string(),
        &gh_env,
    ) {
        warn!(
            id = %record.id,
            name = %record.tmux_name,
            runtime = %record.runtime.as_str(),
            "spawn_managed: runtime adapter spawn failed: {e}"
        );
        let _ = mgr
            .mark_errored(&record.id, &format!("spawn failed: {e}"))
            .await;
    } else {
        info!(
            id = %record.id,
            name = %record.tmux_name,
            path = %prepared.path.display(),
            "managed session spawned successfully"
        );
    }

    emit(ProvisioningStage::Complete);
    Ok(mgr.get(&record.id).await.unwrap_or(record))
}

/// Whether `s` names an EXISTING local directory usable as a session workspace
/// directly, i.e. without a git clone (#1433).
///
/// Why: the local-path spawn fast path must reliably distinguish "an absolute
/// directory the operator already has on disk" from "a remote repo URL to clone".
/// A URL (`https://…`, `git@…:…`) is never an absolute filesystem path, a relative
/// path is rejected (ambiguous against the daemon's cwd), and a non-existent path
/// falls through to clone — so this errs toward the safe existing behaviour.
/// What: returns `true` iff `s` is an ABSOLUTE path that is a directory. `is_dir()`
/// already implies existence in a single `stat` syscall (a missing path is not a
/// dir) and follows symlinks, so a separate `exists()` probe is redundant.
/// Test: `is_local_workdir_detects_absolute_dir`,
/// `is_local_workdir_rejects_url_relative_and_missing` in tests/local_spawn.rs.
pub fn is_local_workdir(s: &str) -> bool {
    let p = std::path::Path::new(s);
    p.is_absolute() && p.is_dir()
}

/// Write the task description to `TASK.md` in a workspace directory (both paths).
///
/// Why: the agent's initial brief must be available as a file in the workspace
/// so it can be read without interactive input (closes #1693). This helper is
/// shared by both the clone path (via `WorkspaceProvisioner::provision_in`) and
/// the local-path fast path (`spawn_managed_local`) so the two call sites cannot
/// diverge. Writing is non-fatal: a failed write is logged but never aborts the
/// spawn. Overwrite semantics are intentional — the caller's task always wins.
/// What: when `task` is non-empty, writes `task` to `<workspace>/TASK.md` and
/// logs a warning on I/O failure. When `task` is empty, does nothing (avoids
/// writing an empty file that would mislead the agent).
/// Test: `local_path_spawn_writes_task_md` in tests/local_spawn.rs.
pub fn write_task_md(workspace: &std::path::Path, task: &str, session_id: &ManagedSessionId) {
    if task.is_empty() {
        return;
    }
    let task_file = workspace.join("TASK.md");
    if let Err(e) = std::fs::write(&task_file, task) {
        tracing::warn!(
            session = %session_id,
            path = %task_file.display(),
            "failed to write TASK.md (local-path spawn): {e}"
        );
    }
}

/// Find an existing Active managed session for `source_id` whose tmux session
/// is still live, so [`spawn_managed_inproject`] can reconnect instead of
/// provisioning a duplicate worktree (#1707).
///
/// Why (issue #1931): extracted out of `spawn_managed_inproject` so the
/// reconnect PREDICATE — the exact rule the operator relies on to avoid
/// duplicate clones/worktrees for the same project — is unit-testable without
/// a live `DaemonState`, tmux, or git worktree. `find` (not the async
/// `SessionManager`/tmux calls themselves) is the only logic that can drift
/// and cause the "existing clone not detected" symptom, so this is the seam
/// worth pinning down with a hermetic regression test.
/// What: returns the first record in `records` whose `source_id` matches
/// `source_id` exactly, whose `state` is [`ManagedSessionState::Active`], and
/// whose `tmux_name` reports alive via `tmux.session_exists(...)`. `None` when
/// no record satisfies all three.
/// Test: `find_reusable_inproject_session_matches_active_live_session`,
/// `find_reusable_inproject_session_ignores_stopped_or_dead_or_other_project`.
fn find_reusable_inproject_session(
    records: &[SessionRecord],
    source_id: &str,
    tmux: &dyn crate::session_manager::ManagedTmuxDriver,
) -> Option<SessionRecord> {
    records
        .iter()
        .find(|r| {
            r.source_id.as_deref() == Some(source_id)
                && r.state == ManagedSessionState::Active
                && tmux.session_exists(&r.tmux_name)
        })
        .cloned()
}

/// Decide whether an in-project spawn should RECONNECT to an existing live
/// session rather than provision a fresh worktree (#1707 + `force_new` opt-out).
///
/// Why (issue #2450): the reconnect decision is exactly what the `force_new`
/// opt-out gates — an explicit "launch new" surface (the `tm` picker's "launch
/// new session" choice) must be able to bypass it, otherwise its task is
/// injected into an unrelated live session for the same project. Keeping the
/// decision a pure function (the `force_new` short-circuit composed with
/// [`find_reusable_inproject_session`]) makes BOTH branches — skip-when-forced
/// AND still-reconnect-when-not-forced — unit-testable without a live
/// `DaemonState`, tmux, or git worktree.
/// What: returns `None` immediately when `force_new` is set (never reconnect);
/// otherwise delegates to [`find_reusable_inproject_session`] (the unchanged
/// #1707 predicate).
/// Test: `reconnect_candidate_none_when_force_new`,
/// `reconnect_candidate_reconnects_when_not_forced`.
fn reconnect_candidate(
    force_new: bool,
    records: &[SessionRecord],
    source_id: &str,
    tmux: &dyn crate::session_manager::ManagedTmuxDriver,
) -> Option<SessionRecord> {
    if force_new {
        return None;
    }
    find_reusable_inproject_session(records, source_id, tmux)
}

/// Spawn a managed session rooted at a per-session git worktree (#1706).
///
/// Why: the in-project spawn path gives every managed session its own isolated
/// branch worktree of a protected base clone, rather than operating directly on
/// the operator's working directory. This mirrors `spawn_managed_local` but uses
/// the `worktree` path as both the session workspace and cwd, records the
/// `source_id` (`owner/repo`) on the record so `tm` can reconnect to existing
/// sessions for the same project, and sets `workspace_owned = false` (the
/// worktree is inside the base clone dir, which the operator should manage; the
/// session does NOT own it for decommission purposes).
///
/// #1913: unlike `spawn_managed`'s clone path and `spawn_managed_local`, this
/// path does not go through `WorkspaceProvisioner::provision_in` (there is no
/// clone step — the worktree already exists via `try_inproject_spawn` +
/// `reserve_inproject_worktree`), so it must call
/// [`crate::core::session_launch::prepare_session_with_repo_url`] itself.
/// Before this fix it never did, so every in-project session silently got no
/// statusline, no deployed agents/skills, no injected trusty-memory/
/// trusty-search MCP config, and no merged CLAUDE.md.
/// What: in order — (1) writes `TASK.md` into the worktree; (2) runs
/// [`prepare_inproject_session`] (best-effort, mirrors `provision_in`'s
/// non-fatal error handling); (3) creates the tmux session rooted at the
/// worktree via `create_with_reserved_name` (issue #2032 — `reserved_name` was
/// already resolved by `reserve_inproject_worktree` and used to name the
/// worktree/branch, so this step reuses it verbatim instead of re-deriving);
/// (4) sets `source_id` via `set_source_id`; (5) runs the FRONT gate
/// (fail-open); (6) marks `Active`; (7) spawns the runtime. A spawn failure
/// marks the record errored (non-fatal).
/// Test: covered transitively by the in-project spawn integration tests;
/// `prepare_inproject_session_writes_statusline` in this module's `tests`
/// submodule exercises the new prep-call in isolation (hermetic — no daemon,
/// tmux, or git required).
#[allow(clippy::too_many_arguments)]
async fn spawn_managed_inproject(
    state: &std::sync::Arc<crate::daemon::state::DaemonState>,
    session_id: &crate::session_manager::ManagedSessionId,
    params: &SpawnParams,
    runtime: crate::runtime::RuntimeKind,
    worktree: std::path::PathBuf,
    owner: String,
    repo: String,
    reserved_name: String,
) -> Result<crate::session_manager::SessionRecord, String> {
    use crate::core::provisioning_stage::{ProvisioningStage, emit};
    use crate::session_manager::ManagedSessionState;

    // NOTE (#2450): the #1707 reconnect pre-flight that used to run here has
    // been hoisted into `spawn_managed_routed`, AHEAD of the worktree
    // reservation, so a reconnect no longer creates-then-prunes a worktree and
    // the `force_new` opt-out can skip it before any side effect. By the time
    // this function runs, the caller has already decided a fresh session is
    // wanted and reserved its worktree.

    info!(
        id = %session_id,
        worktree = %worktree.display(),
        owner = %owner,
        repo = %repo,
        "spawn_managed: in-project worktree spawn"
    );

    write_task_md(&worktree, &params.task, session_id);

    // Pass a canonical GitHub HTTPS URL as repo_url so the session-manager can
    // derive the project name (`tm-<repo>-NN`, issue #1955, formerly
    // `tmpm-<repo>-<8hex>` per #1789) for the tmux session name. Using a
    // synthetic HTTPS URL is safe: `parse_github_path`
    // normalises both SSH and HTTPS forms to the same `{ owner, repo }` pair, and
    // this URL is the real origin of the base clone (it was derived from the
    // operator's local `remote.origin.url`). The record stores it as `repo_url`
    // which gives `tm session ls` useful project context even for in-project
    // sessions that did not clone a fresh workspace. It also doubles as the
    // `repo_url` threaded into `prepare_inproject_session` below, for the same
    // trusty-memory palace-pinning reason `provision_in` threads its `repo_url`.
    let synthetic_repo_url = format!("https://github.com/{owner}/{repo}");

    // Prepare the session BEFORE spawning the runtime (#1913). See the
    // function-level doc for why this call is required here specifically (no
    // clone step wraps it, unlike the other two spawn paths).
    //
    // #1931: use `for_managed_workspace(&worktree)`, NOT `default()` — the
    // harness cwd for this session IS `worktree`, so deployed agents/skills
    // must land in `<worktree>/.claude/{agents,skills}` (where Claude Code's
    // project-skill discovery looks), not the real `$HOME/.claude`.
    let fw = crate::core::paths::FrameworkPaths::for_managed_workspace(&worktree);
    prepare_inproject_session(&fw, session_id, &worktree, &synthetic_repo_url);

    // #1919: mirrors `spawn_managed_cloned`'s placement — announce the tmux
    // stage right before the record (and its tmux session name) is created.
    emit(ProvisioningStage::CreatingTmuxSession);
    let mgr = state.session_manager().await;
    let record = mgr
        .create_with_reserved_name(
            *session_id,
            reserved_name,
            params.task.clone(),
            Some(worktree.clone()),
            Some(worktree.clone()),
            Some(synthetic_repo_url),
            None,
            runtime,
            params.ephemeral.unwrap_or(false),
            false, // workspace_owned: worktrees are inside the base clone; not auto-deletable
        )
        .await
        .map_err(|e| {
            warn!(id = %session_id, "spawn_managed (inproject): create failed: {e}");
            e.to_string()
        })?;

    // Record the source project identity so callers can reconnect later.
    let source_id = format!("{owner}/{repo}");
    if let Err(e) = mgr.set_source_id(session_id, &source_id).await {
        warn!(id = %session_id, "spawn_managed (inproject): set_source_id failed: {e}");
    }

    // #3649: mirrors `spawn_managed_cloned`'s identical call.
    mgr.set_worktree_owner_best_effort(session_id, *session_id)
        .await;

    // #3822: mirrors `spawn_managed_cloned`'s identical explicit
    // spawn-time project registration.
    state
        .project_registry()
        .await
        .register_from_session(&record)
        .await;

    // Front gate with the original repo_url so GitHub identity is parseable.
    if let Some(record) =
        front_gate_or_escalate(&mgr, &record, &params.repo_url, &params.task).await?
    {
        return Ok(record);
    }

    if let Err(e) = mgr
        .set_workspace(session_id, worktree.clone(), ManagedSessionState::Active)
        .await
    {
        warn!(id = %session_id, "spawn_managed (inproject): set_workspace failed: {e}");
    }

    // Deployment-completeness check (#2158, made non-blocking by #2172): see
    // `spawn_managed_cloned`'s identical check for the full rationale. Reuses
    // the `fw` already resolved above for `prepare_inproject_session`.
    if let Err(reason) =
        ensure_deployment_complete(&fw, &worktree, record.repo_url.as_deref(), session_id)
    {
        warn!(id = %session_id, "spawn_managed (inproject): deployment incomplete after auto-repair (non-blocking, launch proceeds): {reason}");
    }

    // Workstream-activity label ensure (issue #3726): idempotent, best-effort,
    // detached — see `session_launch::spawn_workstream_label_ensure`.
    crate::core::session_launch::spawn_workstream_label_ensure(
        record.repo_url.clone(),
        worktree.clone(),
        record.tmux_name.clone(),
    );

    emit(ProvisioningStage::LaunchingRuntime);
    let tmux_arc = mgr.tmux_driver();
    let adapter = crate::runtime::build_adapter(record.runtime, tmux_arc);
    let gh_env = resolve_gh_env(state, &worktree).await;
    if let Err(e) = adapter.spawn(
        &record.tmux_name,
        &worktree,
        &params.task,
        &record.id.to_string(),
        &gh_env,
    ) {
        warn!(
            id = %record.id,
            name = %record.tmux_name,
            "spawn_managed (inproject): runtime adapter spawn failed: {e}"
        );
        let _ = mgr
            .mark_errored(&record.id, &format!("spawn failed: {e}"))
            .await;
    } else {
        info!(
            id = %record.id,
            name = %record.tmux_name,
            worktree = %worktree.display(),
            "managed session spawned successfully (in-project worktree)"
        );
    }

    emit(ProvisioningStage::Complete);
    Ok(mgr.get(&record.id).await.unwrap_or(record))
}

/// Run the session-preparation pipeline for an in-project worktree, logging
/// (never propagating) any failure.
///
/// Why (#1913): [`spawn_managed_inproject`] has no clone step to wrap this call
/// in — unlike `spawn_managed`'s clone branch and `spawn_managed_local`, which
/// both get preparation "for free" as part of `WorkspaceProvisioner::provision_in`
/// — so it must invoke [`crate::core::session_launch::prepare_session_with_repo_url`]
/// directly. Extracted to a named, `fw`-parameterised function (rather than
/// inlined) so it is unit-testable against a hermetic [`crate::core::paths::FrameworkPaths::under`]
/// tempdir without touching the operator's real `~/.trusty-mpm`/`~/.claude`.
/// What: calls `prepare_session_with_repo_url(fw, worktree, Some(repo_url))`.
/// On success, logs the deployed-agent count AND the deployed-skill count
/// (`report.skill_deploy.deployed.len()` — #1917; previously only the agent
/// count was logged, so a skill-deploy no-op was invisible here too). On
/// failure, logs a `tracing::warn!` and returns — mirroring
/// `WorkspaceProvisioner::provision_in`'s non-fatal handling of the identical
/// call, so a prep failure never blocks the session from spawning.
/// Test: `prepare_inproject_session_writes_statusline` in this module's `tests`
/// submodule.
pub(super) fn prepare_inproject_session(
    fw: &crate::core::paths::FrameworkPaths,
    session_id: &ManagedSessionId,
    worktree: &std::path::Path,
    repo_url: &str,
) {
    match crate::core::session_launch::prepare_session_with_repo_url(fw, worktree, Some(repo_url)) {
        Ok(report) => {
            info!(
                id = %session_id,
                deployed = report.deploy.deployed.len(),
                skills_deployed = report.skill_deploy.deployed.len(),
                worktree = %worktree.display(),
                "spawn_managed (inproject): session prepared"
            );
            // Issue #2149: a roster-deploy failure no longer aborts
            // preparation, so surface it loudly here rather than letting it
            // hide behind a low `deployed`/`skills_deployed` count.
            for err in &report.roster_errors {
                tracing::error!(
                    id = %session_id,
                    worktree = %worktree.display(),
                    "spawn_managed (inproject): roster provisioning gap (session \
                     still launches with its trusty-mpm identity): {err}"
                );
            }
        }
        Err(e) => {
            warn!(
                id = %session_id,
                worktree = %worktree.display(),
                "spawn_managed (inproject): session prep failed (non-fatal): {e}"
            );
        }
    }
}

/// Spawn a managed session rooted at a local directory, redirecting to a managed
/// clone when the directory has a parseable GitHub remote (#1590).
///
/// Why: the local-path fast path of [`spawn_managed`]. Before #1590 this function
/// used the live checkout directly; now it checks for a GitHub remote and, when
/// found, provisions a managed clone under the canonical
/// `~/trusty-mpm-projects/<owner>/<repo>/<session-id>/` path — keeping the live
/// checkout untouched. A local directory with NO parseable GitHub remote is an
/// error: managed sessions always operate on a remote clone so concurrent sessions
/// are isolated from the operator's working tree.
/// What: in order — (0) reads `remote.origin.url` from the local directory; if
/// absent or unparseable, returns `Err` (the `tm connect` path handles remotes-less
/// directories); (1) provisions a managed clone via `provision_in` and reassigns
/// `workspace` to the clone path; (2) creates the tmux session record via
/// `create_with_id` (with `cwd = workspace_path = <managed clone>`,
/// `repo_url = Some(<origin_url>)`, `owned = true`); (3) sets `source_id` on the
/// record; (4) runs the FRONT gate (fail-open); (5) marks the record `Active`;
/// (6) spawns the runtime. A spawn failure marks the record errored (non-fatal).
/// Returns the final record.
/// Test: `spawn_managed_local_redirects_to_managed_clone` and
/// `spawn_managed_local_errors_on_no_remote` in tests/local_spawn.rs cover the
/// two key branches. The clone path assertions live in the existing
/// `local_path_spawn_*` test suite.
async fn spawn_managed_local(
    state: &Arc<DaemonState>,
    session_id: &ManagedSessionId,
    params: &SpawnParams,
    runtime: RuntimeKind,
) -> Result<SessionRecord, String> {
    use crate::core::provisioning_stage::{ProvisioningStage, emit};

    let local_dir = std::path::PathBuf::from(&params.repo_url);

    // Step 0 (#1590): managed-path redirect.
    //
    // Check whether the local directory has a parseable GitHub remote. If it does,
    // provision a managed clone and operate in that clone instead of the live
    // checkout. If it does not, the managed path cannot be established — error so
    // the caller (or the operator via `tm connect`) handles the no-remote case.
    let origin_url = super::inproject::get_origin_url(&local_dir).ok_or_else(|| {
        format!(
            "spawn failed: '{}' has no git origin remote; \
                 managed sessions require a GitHub remote. \
                 Use `tm connect` / `tm launch --live` to run in the live checkout.",
            local_dir.display()
        )
    })?;

    let gh = trusty_common::github_path::parse_github_path(&origin_url).ok_or_else(|| {
        format!(
            "spawn failed: could not parse a GitHub owner/repo from origin remote \
             '{origin_url}' for '{}'. \
             Use `tm connect` to run in the live checkout instead.",
            local_dir.display()
        )
    })?;

    let source_id_str = format!("{}/{}", gh.owner, gh.repo);
    let config = crate::core::trusty_tools_config::TrustyToolsConfig::load();
    let project_dir = crate::core::trusty_tools_config::workspace_subpath(&config, &gh);
    // #2184: same per-project identity resolution as the clone-based path
    // (`spawn_managed_cloned`) — this branch also provisions a managed clone
    // (of `origin_url`), so it must honour the same binding, including the
    // #3312 account-pairing self-heal (see `core::git_identity` module docs).
    let git_identity = resolve_for_config_enforced(&config, &origin_url)
        .await
        .map_err(|e| {
            warn!(id = %session_id, "spawn_managed (local→managed): git identity resolution failed: {e}");
            format!("git identity resolution failed: {e}")
        })?;
    let provisioner = crate::provisioner::WorkspaceProvisioner::new(
        crate::provisioner::RealGitBackend::new(git_identity),
        std::path::PathBuf::new(),
    );
    let prepared = provisioner
        .provision_in(
            &project_dir,
            session_id,
            &origin_url,
            &params.git_ref,
            &params.task,
        )
        .map_err(|e| {
            warn!(id = %session_id, "spawn_managed (local→managed): provision failed: {e}");
            format!("workspace provisioning failed: {e}")
        })?;

    // `workspace` now points at the MANAGED clone, not the live checkout.
    let workspace = prepared.path;
    info!(
        id = %session_id,
        live = %local_dir.display(),
        managed = %workspace.display(),
        source_id = %source_id_str,
        "spawn_managed: local-path redirected to managed clone (#1590)"
    );

    // #1919: mirrors `spawn_managed_cloned`'s placement — the clone/prepare
    // stages above already fired inside `provision_in`; announce the tmux
    // stage right before the record (and its tmux session name) is created.
    emit(ProvisioningStage::CreatingTmuxSession);
    let mgr = state.session_manager().await;
    let record = mgr
        .create_with_id(
            *session_id,
            params.task.clone(),
            Some(workspace.clone()),
            params.name_hint.clone(),
            Some(workspace.clone()),
            Some(origin_url.clone()),
            if params.git_ref.is_empty() {
                None
            } else {
                Some(params.git_ref.clone())
            },
            runtime,
            params.ephemeral.unwrap_or(false),
            true, // owned: we provisioned a fresh clone; decommission may remove it
        )
        .await
        .map_err(|e| {
            warn!(id = %session_id, "spawn_managed (local→managed): create failed: {e}");
            e.to_string()
        })?;

    // Record the source project identity so callers can reconnect by project.
    if let Err(e) = mgr.set_source_id(session_id, &source_id_str).await {
        warn!(id = %session_id, "spawn_managed (local→managed): set_source_id failed: {e}");
    }

    // #3822: mirrors `spawn_managed_cloned`'s identical explicit
    // spawn-time project registration.
    state
        .project_registry()
        .await
        .register_from_session(&record)
        .await;

    // FRONT gate: origin_url is a real GitHub URL so the gate is active.
    if let Some(record) = front_gate_or_escalate(&mgr, &record, &origin_url, &params.task).await? {
        return Ok(record);
    }

    if let Err(e) = mgr
        .set_workspace(&record.id, workspace.clone(), ManagedSessionState::Active)
        .await
    {
        warn!(id = %record.id, "spawn_managed (local→managed): set_workspace failed: {e}");
    }

    // Deployment-completeness check (#2158, made non-blocking by #2172): see
    // `spawn_managed_cloned`'s identical check for the full rationale.
    let fw = crate::core::paths::FrameworkPaths::for_managed_workspace(&workspace);
    if let Err(reason) =
        ensure_deployment_complete(&fw, &workspace, record.repo_url.as_deref(), &record.id)
    {
        warn!(id = %record.id, "spawn_managed (local→managed): deployment incomplete after auto-repair (non-blocking, launch proceeds): {reason}");
    }

    // Workstream-activity label ensure (issue #3726): idempotent, best-effort,
    // detached — see `session_launch::spawn_workstream_label_ensure`.
    crate::core::session_launch::spawn_workstream_label_ensure(
        record.repo_url.clone(),
        workspace.clone(),
        record.tmux_name.clone(),
    );

    emit(ProvisioningStage::LaunchingRuntime);
    let tmux_arc = mgr.tmux_driver();
    let adapter = build_adapter(record.runtime, tmux_arc);
    let gh_env = resolve_gh_env(state, &workspace).await;
    if let Err(e) = adapter.spawn(
        &record.tmux_name,
        &workspace,
        &params.task,
        &record.id.to_string(),
        &gh_env,
    ) {
        warn!(
            id = %record.id,
            name = %record.tmux_name,
            runtime = %record.runtime.as_str(),
            "spawn_managed (local→managed): runtime adapter spawn failed: {e}"
        );
        let _ = mgr
            .mark_errored(&record.id, &format!("spawn failed: {e}"))
            .await;
    } else {
        info!(
            id = %record.id,
            name = %record.tmux_name,
            path = %workspace.display(),
            "managed session spawned successfully (local→managed clone)"
        );
    }

    emit(ProvisioningStage::Complete);
    Ok(mgr.get(&record.id).await.unwrap_or(record))
}

/// Perform the withheld spawn (Step 3) for a FRONT-gate-escalated session.
///
/// Why: when the FRONT gate escalates (#1360), the spawn is WITHHELD and the
/// session sits in `Provisioning` with a `pending_decision`. After a human
/// resolves it via `POST …/answer`, the runtime must actually start — otherwise
/// the answer clears the decision but the agent never launches (AC-15). This is
/// the exact Step 3 from [`spawn_managed`], lifted so the answer path can invoke
/// it on demand.
/// What: transitions the record to `Active` (persisting the workspace), builds
/// the runtime adapter, and spawns the harness in the pane — marking the record
/// errored on spawn failure (non-fatal, mirroring `spawn_managed`). Idempotent
/// guard: callers should only invoke this for a session still in `Provisioning`
/// (never spawned), so an already-live session is not double-spawned.
/// Test: `front_gate_answer_unblocks_spawn` in tests/session_manager_mvp.rs.
pub async fn spawn_runtime_for(
    state: &Arc<DaemonState>,
    record: &SessionRecord,
) -> Result<(), String> {
    let mgr = state.session_manager().await;
    let workspace = record
        .workspace_path
        .clone()
        .unwrap_or_else(|| record.cwd.clone());

    if let Err(e) = mgr
        .set_workspace(&record.id, workspace.clone(), ManagedSessionState::Active)
        .await
    {
        warn!(id = %record.id, "spawn_runtime_for: set_workspace failed: {e}");
    }

    let tmux_arc = mgr.tmux_driver();
    let adapter = build_adapter(record.runtime, tmux_arc);
    let gh_env = resolve_gh_env(state, &workspace).await;
    if let Err(e) = adapter.spawn(
        &record.tmux_name,
        &workspace,
        &record.task,
        &record.id.to_string(),
        &gh_env,
    ) {
        warn!(
            id = %record.id,
            name = %record.tmux_name,
            "spawn_runtime_for: runtime adapter spawn failed: {e}"
        );
        let _ = mgr
            .mark_errored(&record.id, &format!("spawn failed: {e}"))
            .await;
        return Err(e.to_string());
    }
    info!(
        id = %record.id,
        name = %record.tmux_name,
        "FRONT-gate-escalated session spawned after human approval"
    );
    Ok(())
}

/// Run the intent-conformance FRONT gate for a freshly-created session record.
///
/// Why: factored out of [`spawn_managed`] so the gate's escalate-vs-proceed
/// decision (and the #1269 degraded operator-confirm branch) is a single, named,
/// testable step rather than inlined in the spawn ritual (spec §5.1, #1360). It
/// keeps `spawn_managed` linear and under the 500-SLOC production cap.
/// What: derives owner/repo from the repo URL, builds the production
/// [`IsrConformanceGate`] rooted at the provisioned workspace, composes the
/// conformance disposition with the pre-work autonomy disposition (stricter-wins,
/// via [`run_front_gate`]), and —
/// - on **auto-accept** → returns `Ok(None)`; the caller proceeds to spawn;
/// - on **escalate** → writes `pending_decision`/`proposed_default` (so the
///   escalation surfaces through `…/activity`, MCP `session_status`, the
///   supervisor, and the `tm` CLI), withholds the spawn, logs the divergence,
///   and returns `Ok(Some(record))` so the caller returns early WITHOUT spawning.
///
/// #1269 degradation: until the headless auto-spawn approval path lands
/// ([`HeadlessApproval::current`] returns `OperatorConfirm`), an escalation
/// degrades to the operator-confirm path — the pending decision is recorded and a
/// human resolves it via `POST …/answer` (which then unblocks the spawn). The
/// gate itself always functions; only the *resolution channel* is degraded.
///
/// Fail-open: the gate never `Err`s on its own resolution failures (it returns an
/// `AutoAccept`); this function only returns `Err` if persisting the escalation
/// fails — in which case the caller surfaces the store error rather than silently
/// spawning.
/// Test: `front_gate_escalation_sets_pending_decision` /
/// `front_gate_clean_match_spawns` in tests/session_manager_mvp.rs, plus the unit
/// matrix in `managed_routes::front_gate::tests`.
pub(super) async fn front_gate_or_escalate(
    mgr: &Arc<crate::session_manager::SessionManager>,
    record: &SessionRecord,
    repo_url: &str,
    task: &str,
) -> Result<Option<SessionRecord>, String> {
    use super::front_gate::{
        HeadlessApproval, IsrConformanceGate, prework_autonomy_disposition, run_front_gate,
    };

    let (owner, repo) = match trusty_common::github_path::parse_github_path(repo_url) {
        Some(gh) => (gh.owner, gh.repo),
        // No GitHub identity → no ticket backend to resolve against; fail-open.
        None => return Ok(None),
    };

    let repo_root = record
        .workspace_path
        .clone()
        .unwrap_or_else(|| record.cwd.clone());
    let gate = IsrConformanceGate::new(repo_root);
    let autonomy = prework_autonomy_disposition(task);
    let approval = HeadlessApproval::current();

    let outcome = run_front_gate(&gate, &owner, &repo, task, autonomy, approval).await;

    if outcome.may_spawn() {
        return Ok(None);
    }

    let reason = outcome
        .escalation_reason()
        .unwrap_or_else(|| "conformance escalation".to_string());
    warn!(
        id = %record.id,
        approval = ?outcome.approval,
        "spawn_managed: FRONT gate escalated; withholding spawn: {reason}"
    );
    mgr.set_pending_decision(&record.id, &reason, outcome.proposed_default.as_deref())
        .await
        .map_err(|e| {
            warn!(id = %record.id, "spawn_managed: set_pending_decision failed: {e}");
            e.to_string()
        })?;

    // Return the updated record (now carrying the pending decision), withholding
    // the spawn. The session stays in `Provisioning` until a human answers.
    Ok(Some(
        mgr.get(&record.id).await.unwrap_or_else(|_| record.clone()),
    ))
}

/// Resume a stopped session and re-spawn its runtime, shared across transports.
///
/// Why: the HTTP resume handler and the MCP `session_resume` tool must both
/// resume the record AND re-spawn the runtime so the session is actually live;
/// centralising avoids the MCP path silently resuming without re-spawning.
/// What: calls [`crate::session_manager::SessionManager::resume`] (which performs
/// the existence + state check in a SINGLE round-trip — no pre-flight `get`, so
/// no TOCTOU window) and maps its typed [`ManagedError`] into a typed
/// [`ResumeManagedError`] (`NotFound`/`InvalidState`/`Other`). It then re-spawns
/// the SAME runtime backend in the fresh tmux session (no re-clone) and returns
/// the final record.
///
/// #1913 self-heal: sessions spawned via the (now-fixed) in-project worktree
/// path before this fix landed never ran `prepare_session` at all, so their
/// workspace may be permanently missing the `statusLine` config key. Every
/// resume defensively re-applies [`crate::core::session_launch::ensure_status_line`]
/// — the ONE prep step confirmed idempotent/non-clobbering by its own doc
/// comment — so such a session self-heals the next time it is resumed. The
/// broader prep pipeline (agent/skill redeploy, CLAUDE.md merge, MCP injection)
/// is intentionally NOT re-run here: those steps are not all confirmed safe to
/// repeat against an already-running workspace, so re-running them on every
/// resume risks a different class of bug for a narrower payoff.
///
/// #2647 worktree/upstream sync: before the statusline self-heal, every
/// resume also calls
/// [`crate::core::session_launch::sync_worktree_with_upstream`] (fetch +
/// fast-forward-only merge — never resets the branch, never touches
/// uncommitted changes) and
/// [`crate::core::session_launch::self_heal_claude_md`] (strips legacy #2170
/// delegation-directive pollution from `CLAUDE.md`, including when it is an
/// uncommitted diff the fast-forward alone cannot touch). Both are
/// best-effort and never block the resume — a long-lived session worktree
/// that was previously frozen at its creation commit now catches up to
/// `origin/main` on every resume instead of silently drifting forever.
/// Test: covered by the HTTP `resume_managed_session` tests and the MCP
/// `session_resume_unknown_id_errors` test;
/// `resume_managed_backfills_missing_status_line` in
/// `tests/session_manager_mvp.rs` covers the self-heal call added here;
/// `core::session_launch::worktree_sync`'s own unit tests cover the sync/
/// self-heal primitives.
pub async fn resume_managed(
    state: &Arc<DaemonState>,
    id: &ManagedSessionId,
) -> Result<SessionRecord, ResumeManagedError> {
    let mgr = state.session_manager().await;
    let record = mgr.resume(id).await.map_err(ResumeManagedError::from)?;

    let workspace = record
        .workspace_path
        .clone()
        .unwrap_or_else(|| record.cwd.clone());

    // Worktree/upstream sync + legacy CLAUDE.md self-heal (#2647): fast-forward
    // -only, best-effort, never blocks the resume — see
    // `core::session_launch::resume_self_heal` for the full safety contract
    // (never resets the branch, never touches uncommitted changes). This is
    // the resume-time counterpart to `WorkspaceProvisioner::fetch_and_reset`
    // for the shared `.base` checkout, closing the gap that let a session
    // worktree silently drift from `origin/main` forever.
    crate::core::session_launch::resume_self_heal(&workspace, &record.id.to_string()).await;

    // Defensive self-heal (#1913): best-effort, never blocks the resume.
    if let Err(e) = crate::core::session_launch::ensure_status_line(&workspace) {
        warn!(
            id = %record.id,
            "resume_managed: statusline self-heal failed (non-fatal): {e}"
        );
    }

    // Deployment-completeness check (#2158, made non-blocking by #2172): see
    // `spawn_managed_cloned`'s identical check for the full rationale.
    // `ensure_deployment_complete` itself no-ops for an unresolved (`/unknown`)
    // workspace — an adopted session with no known cwd is handled separately by
    // the reconcile-on-boot fix, not here.
    let fw = crate::core::paths::FrameworkPaths::for_managed_workspace(&workspace);
    if let Err(reason) =
        ensure_deployment_complete(&fw, &workspace, record.repo_url.as_deref(), &record.id)
    {
        warn!(id = %record.id, "resume_managed: deployment incomplete after auto-repair (non-blocking, launch proceeds): {reason}");
    }

    let tmux_arc = mgr.tmux_driver();
    let adapter = build_adapter(record.runtime, tmux_arc);
    // #1744: prefer --resume <id> when a claude_session_id was captured at
    // SessionStart; fall back to --continue (most-recent conversation in the
    // workspace) when the id is absent. ClaudeCodeAdapter overrides spawn_resume
    // to implement this; TcodeAdapter's default delegates to plain spawn.
    // Sibling-window hijack fix (follow-up to #2456): pass the record's OWN
    // `record.pane_id` through so the adapter targets that SPECIFIC pane
    // (`SessionManager::resume`, just above, already confirmed it — or a
    // freshly recreated one — still exists) instead of a session-scoped
    // target that tmux could resolve to an unrelated active sibling pane.
    let gh_env = resolve_gh_env(state, &workspace).await;
    if let Err(e) = adapter.spawn_resume(
        &record.tmux_name,
        record.pane_id.as_deref(),
        &workspace,
        &record.task,
        record.claude_session_id.as_deref(),
        &record.id.to_string(),
        &gh_env,
    ) {
        warn!(
            id = %record.id,
            name = %record.tmux_name,
            runtime = %record.runtime.as_str(),
            "resume_managed: runtime adapter spawn_resume failed: {e}"
        );
        let _ = mgr
            .mark_errored(&record.id, &format!("resume spawn failed: {e}"))
            .await;
    } else {
        info!(
            id = %record.id,
            name = %record.tmux_name,
            workspace = %workspace.display(),
            "managed session resumed and runtime respawned"
        );
    }

    Ok(mgr.get(id).await.unwrap_or(record))
}

#[cfg(test)]
#[path = "lifecycle_tests.rs"]
mod tests;