agentmux_srv/sagas/

delete_block.rs

// Copyright 2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0
//
// Phase E.5.7 (Step 5 PR 1) — DeleteBlock saga.
//
// Replaces the SQLite-first delete pattern in `service.rs`'s
// `("object", "DeleteBlock")` handler with a reducer-driven saga.
// The legacy handler called `wcore::delete_block` first and then
// dispatched `Command::DeleteBlock` to keep the reducer in sync — a
// short-circuit that pre-dates the saga coordinator + persist
// subscriber pattern (closes gap §4 in
// `docs/retro/reducer-architecture-gaps-2026-05-01.md`).
//
// **Steps:**
// 0. `agent_teardown::run` for the block, with the caller's policy
//    (`Policy::close()` by default, `Policy::quit()` from `self_quit`) — the same graceful
//    shutdown the pane's × uses (interrupt a running turn, EOF, force-kill
//    only at the deadline, release its processes, save final state),
//    BEFORE the records go (SPEC_AGENT_PANE_CLOSE_GRACEFUL_SHUTDOWN_2026_09_18.md
//    §4.2).
// 1. `DeleteBlock { tab_id, block_id }` — reducer removes the block
//    from canonical state and emits `Event::BlockDeleted`. The
//    persist subscriber writes SQLite via `wcore::delete_block`.
// 2. `LayoutDeleteNodeByBlock` (SPEC_864 site #6) — reducer prunes the
//    block from `db_layout`, the single writer's own copy of the tree.
//    For one tab of a stacked pane it removes just that member and keeps
//    the pane and its other tabs; the last member takes the leaf.
// 2b. `queue_source_layout_delete` — tells any frontend that already
//    has this tab's tree loaded (via `LayoutState.pendingbackendactions`)
//    to prune the node too. Added after
//    `INVESTIGATION_LAYOUT_DEAD_SPACE_STALE_TREE_RESURRECTION_2026_07_08.md`
//    found that without this, step 2's direct backend write is
//    invisible to an already-loaded frontend, whose next unrelated
//    edit in the tab silently resurrects the just-pruned node as a
//    dangling reference.
//
// **Block controller side-effect:** the process is stopped gracefully by
// step 0. `blockcontroller::delete_controller` then runs AFTER the saga,
// whenever the reducer has removed the block (see the round-3 note in
// `run`); by then it is idempotent cleanup of an already-stopped
// controller, not a kill. (This header used to say the controller was
// killed before the reducer dispatch — the legacy RPC handler's order,
// long since replaced; that stale line misled a reader into thinking this
// saga hard-kills.)
//
// This saga is therefore the right primitive for a close srv starts
// itself on one tab: unlike `close_pane::run` with one id, which queues a
// frontend action only when the whole leaf goes, step 2b updates an
// already-loaded frontend.
//
// **Compensation:** delete sagas are awkward to compensate — once
// the block row + controller are gone, "un-delete" requires
// reconstructing both the SQLite row and the PTY/process subtree,
// neither of which is meaningful from saga state. We follow the
// brief's pragma: log a warning on dispatch failure, no automatic
// re-create. The reducer's `DeleteBlock` is silent-no-op on missing
// inputs (see reducer.rs handle_delete_block), so the only failure
// path is mstore write errors surfaced by the persist subscriber —
// in which case the controller is already gone (intentional; the
// PTY can't be partially-killed) and the SQLite row may or may not
// have been written. PR 2's `compensate_unresolved` resume scan
// surfaces these via the durable saga log for operator follow-up.
//
// **Pre-condition:** the block must exist in the reducer state.
// Without this, the reducer would silently no-op (handle_delete_block
// returns an empty event vec on missing tab/block) and the user
// would see "delete succeeded" while nothing happened. The saga
// surfaces a clear "block not found" error instead. We check the
// reducer state (not SQLite) because `Command::DeleteBlock` carries
// `tab_id` looked up from the block's owning record in srv state
// (service.rs: `s.blocks[block_id].tab_id`) and we want to validate
// against the reducer's view of (tab → blocks) — that's what the
// dispatch will mutate.

use agentmux_common::ipc::Command;
use serde_json::{json, Value};

use super::{
    alloc_saga_id, classify_run_saga_result, emit_saga_started, emit_terminal, run_saga, SagaCtx,
};
use crate::server::AppState;

/// Run the DeleteBlock saga. On success returns
/// `{"block_id": "...", "tab_id": "..."}`.
pub async fn run(
    state: &AppState,
    tab_id: String,
    block_id: String,
) -> Result<Value, String> {
    run_with_policy(state, tab_id, block_id, super::agent_teardown::Policy::close()).await.0
}

/// [`run`] with the teardown policy for the block (`self_quit` passes
/// `Policy::quit()`). Returns the saga's result and the teardown's report;
/// the report is there whenever the teardown ran, even if the close then
/// failed, and `None` only when the close failed before it.
pub async fn run_with_policy(
    state: &AppState,
    tab_id: String,
    block_id: String,
    policy: super::agent_teardown::Policy,
) -> (Result<Value, String>, Option<super::agent_teardown::TeardownReport>) {
    let mut report = None;
    let result = run_inner_with_policy(state, tab_id, block_id, policy, &mut report).await;
    (result, report)
}

async fn run_inner_with_policy(
    state: &AppState,
    tab_id: String,
    block_id: String,
    policy: super::agent_teardown::Policy,
    report_out: &mut Option<super::agent_teardown::TeardownReport>,
) -> Result<Value, String> {
    // Pre-condition: block exists and is in the named tab. Reducer
    // would silent-no-op otherwise (see handle_delete_block); the
    // saga surfaces a clear error instead.
    {
        let s = state.srv_state.lock().await;
        match s.blocks.get(&block_id) {
            None => {
                return Err(format!("DeleteBlock: block not found: {}", block_id));
            }
            Some(block) if block.tab_id != tab_id => {
                return Err(format!(
                    "DeleteBlock: block {} is in tab {}, not {}",
                    block_id, block.tab_id, tab_id
                ));
            }
            _ => {}
        }
        if !s.tabs.contains_key(&tab_id) {
            return Err(format!("DeleteBlock: tab not found: {}", tab_id));
        }
    }

    let saga_id = alloc_saga_id(state);
    if let Err(e) = emit_saga_started(
        state,
        saga_id,
        "delete_block",
        json!({
            "tab_id": &tab_id,
            "block_id": &block_id,
        }),
    )
    .await
    {
        return Err(e);
    }
    // Stop the block's process and save its final state BEFORE its records
    // are deleted — the same per-agent order the pane's × uses
    // (`close_pane`, SPEC_AGENT_PANE_CLOSE_GRACEFUL_SHUTDOWN_2026_09_18.md
    // §4.2). Deleting first left the controller's own exit cleanup writing
    // to a block that no longer existed. After `emit_saga_started`, so a
    // saga-start collision still has no side effect (round 1 below).
    *report_out = Some(super::agent_teardown::run(state, &block_id, policy).await);
    let ctx = SagaCtx::new(state, saga_id);
    let result = run_saga("delete_block", run_inner(ctx, tab_id, block_id.clone())).await;
    // Controller-kill ordering. Three rounds of bot review:
    //   * Round 1 (agent): killed BEFORE emit_saga_started → reagent
    //     P2: side-effect leak if start_saga collides.
    //   * Round 2 (this PR): conditional on result.is_ok() → codex P2
    //     round 2: leaks PTY when reducer succeeds but
    //     `apply_event_to_mstore` fails inside `SagaCtx::dispatch`
    //     (block already removed from reducer state, RPC returns
    //     error, retry pre-check sees "block not found" → controller
    //     never cleaned up).
    //   * Round 3 (this fix): kill controller whenever the reducer
    //     dispatched DeleteBlock — i.e., whenever block was removed
    //     from reducer state. This includes both success and the
    //     reducer-succeeded-mstore-failed cases. We approximate this
    //     by checking reducer state for the block: if it's gone, the
    //     reducer dispatched (regardless of mstore outcome), so kill
    //     the controller. Idempotent on missing controller.
    {
        let block_still_in_reducer =
            state.srv_state.lock().await.blocks.contains_key(&block_id);
        if !block_still_in_reducer {
            crate::sagas::agent_teardown::discard(&block_id);
            // MPS persist_map is keyed by (event, scope) and its key set is
            // never otherwise pruned — a deleted block's persisted history
            // (install_progress, block:activity, ...) would linger for the
            // life of the process. Purge it alongside the controller kill.
            state.broker.purge_scope(&format!("block:{}", block_id));
        }
    }
    if let Err(reason) = &result {
        // The pane is showing the shutdown log and waiting; tell it why.
        super::close_pane::publish_shutdown_error(state, &block_id, reason);
    }
    super::close_pane::finish_close(state, &block_id).await;
    emit_terminal(state, saga_id, classify_run_saga_result(&result)).await;
    result
}

async fn run_inner(
    ctx: SagaCtx<'_>,
    tab_id: String,
    block_id: String,
) -> Result<Value, String> {
    // Step 1: dispatch DeleteBlock through the reducer. The persist
    // subscriber sees the BlockDeleted event and runs
    // `wcore::delete_block` (SQLite row deletes; the layout prune moved
    // to Step 2 — SPEC_864 site #6).
    if let Err(reason) = ctx
        .dispatch(Command::DeleteBlock {
            tab_id: tab_id.clone(),
            block_id: block_id.clone(),
        })
        .await
    {
        // No automatic compensation — un-deleting a block requires
        // reconstructing both SQLite + PTY which we cannot do from
        // saga state. Surface the failure; PR 2's restart-recovery
        // scan picks up the durable log row for operator review.
        tracing::warn!(
            tab_id = %tab_id,
            block_id = %block_id,
            "[saga] DeleteBlock dispatch failed (no automatic compensation): {}",
            reason
        );
        return Err(format!("DeleteBlock: {}", reason));
    }

    // Step 2 (SPEC_864 site #6): prune the deleted block's layout node
    // through the reducer — the single writer of db_layout — instead of
    // the retired wcore-direct prune the persist subscriber used to run.
    // The reducer resolves block→node itself (it owns the tree) and
    // silently no-ops when the block has no node (frontend already
    // pushed a pruned tree, block never laid out, empty tab). The
    // resulting `LayoutNodeDeleted{new_tree, tree_cleared}` event is
    // what the subscriber persists.
    //
    // Best-effort, mirroring the old wcore prune's `let _ =
    // store.update(&mut layout)`: a prune failure must not fail the
    // block deletion itself (the block + controller are already gone).
    //
    // CORRECTION (found via INVESTIGATION_LAYOUT_DEAD_SPACE_STALE_TREE_
    // RESURRECTION_2026_07_08.md): this comment used to claim "an
    // orphaned node is exactly what the frontend's own delete-push
    // converges away" — that's backwards. The frontend's `LayoutModel`
    // doesn't auto-sync from a bare backend rootnode write (see
    // `layout_helpers.rs::queue_target_layout_insert`'s doc comment); a
    // frontend that already has this tab's tree loaded is never told
    // this dispatch happened, so its *next* unrelated edit re-pushes
    // its stale copy — still containing this node — and resurrects the
    // dangling reference instead of converging it away. Step 2b below
    // closes that gap via the same `pendingbackendactions` channel
    // tear-off/redock already use for exactly this reason.
    if let Err(reason) = ctx
        .dispatch(Command::LayoutDeleteNodeByBlock {
            tab_id: tab_id.clone(),
            block_id: block_id.clone(),
            correlation_id: String::new(),
        })
        .await
    {
        tracing::warn!(
            tab_id = %tab_id,
            block_id = %block_id,
            "[saga] LayoutDeleteNodeByBlock dispatch failed (best-effort; layout node may orphan until frontend push): {}",
            reason
        );
    }

    // Step 2b: tell any frontend that already has this tab's layout
    // tree loaded to prune the node too, via the same pendingbackendactions
    // channel `queue_source_layout_delete` already uses for tear-off's
    // source-tab prune. Without this, Step 2's direct reducer write is
    // invisible to an already-loaded frontend, and its next unrelated
    // edit in this tab resurrects the just-deleted node as a dangling
    // reference (empty/dead pane). Best-effort for the same reason as
    // Step 2 — the block is already gone either way.
    if let Err(reason) =
        crate::server::service::layout_helpers::queue_source_layout_delete(
            ctx.state, &tab_id, &block_id,
        )
        .await
    {
        tracing::warn!(
            tab_id = %tab_id,
            block_id = %block_id,
            "[saga] queue_source_layout_delete failed (best-effort; a frontend with a stale loaded copy of this tab may resurrect the node): {}",
            reason
        );
    }

    Ok(json!({
        "tab_id": tab_id,
        "block_id": block_id,
    }))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::obj::Block;
    use crate::server::tests::test_state;
    use agentmux_common::ipc::Event;

    async fn dispatch_apply(
        state: &crate::server::AppState,
        cmd: agentmux_common::ipc::Command,
    ) -> Vec<agentmux_common::ipc::Event> {
        let events = crate::server::service::dispatch_to_reducer(state, cmd).await;
        for ev in &events {
            crate::persist_subscriber::apply_event_to_mstore(ev, &state.mstore).unwrap();
        }
        events
    }

    /// Seed a workspace + tab + block and return their ids.
    async fn seed() -> (
        crate::server::AppState,
        String, // workspace_id
        String, // tab_id
        String, // block_id
    ) {
        let state = test_state();
        let ws_evs = dispatch_apply(
            &state,
            agentmux_common::ipc::Command::CreateWorkspace { name: "w".into() },
        )
        .await;
        let ws_id = ws_evs
            .iter()
            .find_map(|e| match e {
                Event::WorkspaceCreated { workspace_id, .. } => Some(workspace_id.clone()),
                _ => None,
            })
            .unwrap();
        let tab_evs = dispatch_apply(
            &state,
            agentmux_common::ipc::Command::CreateTab {
                workspace_id: ws_id.clone(),
                name: "t".into(),
            },
        )
        .await;
        let tab_id = tab_evs
            .iter()
            .find_map(|e| match e {
                Event::TabCreated { tab_id, .. } => Some(tab_id.clone()),
                _ => None,
            })
            .unwrap();
        let blk_evs = dispatch_apply(
            &state,
            agentmux_common::ipc::Command::CreateBlock {
                tab_id: tab_id.clone(),
                meta: serde_json::Value::Null,
            },
        )
        .await;
        let block_id = blk_evs
            .iter()
            .find_map(|e| match e {
                Event::BlockCreated { block_id, .. } => Some(block_id.clone()),
                _ => None,
            })
            .unwrap();
        (state, ws_id, tab_id, block_id)
    }

    #[tokio::test]
    async fn happy_path_removes_block_from_reducer_and_sqlite() {
        let (state, _ws_id, tab_id, block_id) = seed().await;

        // Sanity: block is present pre-delete.
        {
            let s = state.srv_state.lock().await;
            assert!(s.blocks.contains_key(&block_id));
            assert_eq!(s.tabs[&tab_id].block_ids, vec![block_id.clone()]);
        }
        assert!(state.mstore.get::<Block>(&block_id).unwrap().is_some());

        let result = run(&state, tab_id.clone(), block_id.clone()).await.unwrap();
        assert_eq!(result["block_id"], block_id);
        assert_eq!(result["tab_id"], tab_id);

        // Reducer: block gone, tab's block_ids empty.
        let s = state.srv_state.lock().await;
        assert!(!s.blocks.contains_key(&block_id));
        assert!(s.tabs[&tab_id].block_ids.is_empty());
        drop(s);

        // SQLite: block gone.
        assert!(state.mstore.get::<Block>(&block_id).unwrap().is_none());
    }

    #[tokio::test]
    async fn saga_prunes_layout_through_reducer_coherently() {
        // SPEC_864 site #6 — the layout prune is Step 2 of this saga
        // (Command::LayoutDeleteNodeByBlock), not the retired
        // wcore-direct write in the persist subscriber. Invariant:
        // after the saga, the reducer's TabRecord tree and
        // db_layout's rootnode agree (single-writer coherence), and a
        // single-block tree ends genuinely empty on both sides.
        let (state, _ws_id, tab_id, block_id) = seed().await;

        // Seed a layout tree holding the block, reducer-routed +
        // persisted (same LayoutSetTree path the frontend push uses).
        let tree = agentmux_common::LayoutNode {
            id: "n-solo".into(),
            data: Some(agentmux_common::LayoutNodeData {
                block_id: block_id.clone(),
                ..Default::default()
            }),
            ..Default::default()
        };
        dispatch_apply(
            &state,
            agentmux_common::ipc::Command::LayoutSetTree {
                tab_id: tab_id.clone(),
                new_tree: Some(tree),
                correlation_id: String::new(),
                slices: None,
            },
        )
        .await;
        // Sanity: reducer + SQLite both hold the seeded tree.
        {
            let s = state.srv_state.lock().await;
            assert!(s.tabs[&tab_id].rootnode.is_some());
        }
        let layout_id = state
            .mstore
            .get::<crate::backend::obj::Tab>(&tab_id)
            .unwrap()
            .unwrap()
            .layoutstate;
        assert!(state
            .mstore
            .get::<crate::backend::obj::LayoutState>(&layout_id)
            .unwrap()
            .unwrap()
            .rootnode
            .is_some());

        run(&state, tab_id.clone(), block_id.clone()).await.unwrap();

        // Reducer tree: pruned to empty (block was the sole leaf).
        {
            let s = state.srv_state.lock().await;
            assert!(
                s.tabs[&tab_id].rootnode.is_none(),
                "reducer tree must be pruned by the saga's Step-2 dispatch"
            );
        }
        // db_layout: matches the reducer (tree_cleared persisted the wipe).
        assert!(
            state
                .mstore
                .get::<crate::backend::obj::LayoutState>(&layout_id)
                .unwrap()
                .unwrap()
                .rootnode
                .is_none(),
            "db_layout must agree with the reducer post-saga (single-writer coherence)"
        );
    }

    #[tokio::test]
    async fn saga_queues_a_pendingbackendactions_delete_for_a_stale_frontend() {
        // Regression test for
        // INVESTIGATION_LAYOUT_DEAD_SPACE_STALE_TREE_RESURRECTION_2026_07_08.md:
        // Step 2 (LayoutDeleteNodeByBlock) alone prunes db_layout correctly
        // but is invisible to a frontend that already has this tab's tree
        // loaded — its next unrelated edit re-pushes its stale copy and
        // resurrects the just-deleted node. Step 2b must queue a "delete"
        // pendingbackendactions entry so that frontend converges instead.
        let (state, _ws_id, tab_id, block_id) = seed().await;

        let tree = agentmux_common::LayoutNode {
            id: "n-solo".into(),
            data: Some(agentmux_common::LayoutNodeData {
                block_id: block_id.clone(),
                ..Default::default()
            }),
            ..Default::default()
        };
        dispatch_apply(
            &state,
            agentmux_common::ipc::Command::LayoutSetTree {
                tab_id: tab_id.clone(),
                new_tree: Some(tree),
                correlation_id: String::new(),
                slices: None,
            },
        )
        .await;

        run(&state, tab_id.clone(), block_id.clone()).await.unwrap();

        let tab = state
            .mstore
            .must_get::<crate::backend::obj::Tab>(&tab_id)
            .unwrap();
        let layout = state
            .mstore
            .must_get::<crate::backend::obj::LayoutState>(&tab.layoutstate)
            .unwrap();
        let actions = layout.pendingbackendactions.unwrap_or_default();
        let delete_action = actions
            .iter()
            .find(|a| a.actiontype == "delete" && a.blockid == block_id);
        assert!(
            delete_action.is_some(),
            "expected a queued 'delete' pendingbackendactions entry for the pruned block \
             (matches LayoutTreeActionType.DeleteNode in frontend/layout/lib/types.ts), got: {actions:?}"
        );
    }

    /// SPEC_PANE_TABS_REDUCER_COMMANDS_2026_09_18.md §2.2: deleting one tab
    /// of a stacked pane (e.g. a terminal tab whose shell exited —
    /// close-on-exit runs this saga) keeps the pane and its other tabs. It
    /// used to delete the whole leaf, leaving the sibling (an agent, say)
    /// running with no pane.
    #[tokio::test]
    async fn deleting_one_tab_of_a_stacked_pane_keeps_the_pane_and_its_sibling() {
        let (state, _ws_id, tab_id, agent) = seed().await;
        let term = dispatch_apply(
            &state,
            agentmux_common::ipc::Command::CreateBlock {
                tab_id: tab_id.clone(),
                meta: serde_json::Value::Null,
            },
        )
        .await
        .iter()
        .find_map(|e| match e {
            Event::BlockCreated { block_id, .. } => Some(block_id.clone()),
            _ => None,
        })
        .unwrap();
        dispatch_apply(
            &state,
            agentmux_common::ipc::Command::LayoutSetTree {
                tab_id: tab_id.clone(),
                new_tree: Some(agentmux_common::LayoutNode {
                    id: "pane".into(),
                    data: Some(agentmux_common::LayoutNodeData {
                        block_id: term.clone(),
                        block_stack: vec![agent.clone(), term.clone()],
                        active_block_id: term.clone(),
                        ..Default::default()
                    }),
                    ..Default::default()
                }),
                correlation_id: String::new(),
                slices: None,
            },
        )
        .await;

        run(&state, tab_id.clone(), term.clone()).await.unwrap();

        let s = state.srv_state.lock().await;
        assert!(s.blocks.contains_key(&agent), "sibling block survives");
        let leaf = s.tabs[&tab_id].rootnode.as_ref().expect("the pane survives");
        let data = leaf.data.as_ref().unwrap();
        assert_eq!(data.block_stack, vec![agent.clone()]);
        assert_eq!(data.block_id, agent, "the sibling becomes the visible tab");
    }

    #[tokio::test]
    async fn rejects_when_block_not_found() {
        let (state, _ws_id, tab_id, _block_id) = seed().await;
        let err = run(&state, tab_id, "ghost-block".into()).await.unwrap_err();
        assert!(err.contains("block not found"), "got: {}", err);
    }

    #[tokio::test]
    async fn rejects_when_block_in_different_tab() {
        let (state, _ws_id, tab_id, block_id) = seed().await;
        // Create a second tab; ask to delete block via that tab's id.
        let tab_evs = dispatch_apply(
            &state,
            agentmux_common::ipc::Command::CreateTab {
                workspace_id: _ws_id.clone(),
                name: "other".into(),
            },
        )
        .await;
        let other_tab = tab_evs
            .iter()
            .find_map(|e| match e {
                Event::TabCreated { tab_id, .. } => Some(tab_id.clone()),
                _ => None,
            })
            .unwrap();
        let err = run(&state, other_tab, block_id.clone()).await.unwrap_err();
        assert!(
            err.contains("is in tab") && err.contains(&tab_id),
            "got: {}",
            err
        );
        // Block must still be present (saga rejected pre-dispatch).
        let s = state.srv_state.lock().await;
        assert!(s.blocks.contains_key(&block_id));
    }
}