agentmux_srv/

persist_subscriber.rs

// Copyright 2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0
//
// Phase E.2c.1+E.2c.2 — persist subscriber. A tokio task that
// consumes the srv reducer's broadcast bus and mirrors
// workspace/tab/block lifecycle events back to SQLite via wcore.
//
// Status:
//   * E.2c.1 (#614) shipped the per-event apply path. Subscriber was
//     dead-code in production because no producer existed.
//   * E.2c.2 (this) wires the workspace HTTP/WS RPC handlers through
//     the reducer, making the subscriber LIVE for workspace events.
//     Adds full-resync-on-`Lagged` for workspaces (RPC-migrated
//     entities only — tab/block resync lands in E.2c.3 / E.2c.4).
//
// Bus-lag handling:
//   On RecvError::Lagged(n) the subscriber drops `n` events.
//   Naive HWM advancement past dropped events would permanently
//   diverge SQLite from reducer state. Instead we do a workspace-
//   scoped resync: snapshot `state.workspaces` under the reducer
//   mutex, write each workspace into SQLite (idempotent insert /
//   no-op / update). The resync is INSERT/UPDATE only — no deletes —
//   because workspaces can also be created OUTSIDE the reducer
//   during the migration window (e.g., the still-wcore-direct
//   `CreateWindow` flow calls `wcore::create_window_full` which
//   creates a workspace under the hood). Deleting workspaces
//   missing from the reducer snapshot would lose those legitimate
//   rows. Stale workspaces deleted-via-reducer that the subscriber
//   missed on Lagged linger on disk until the next user-driven
//   DeleteWorkspace cleans them up via the wcore fallback in
//   service.rs. (codex P1 #615.)
//
//   IMPORTANT: tab/block resync is NOT done here yet. Tabs and
//   blocks remain RPC-direct via wcore; the reducer's view of them
//   is FROZEN at bootstrap. Doing tab/block resync before E.2c.3
//   and E.2c.4 land would clobber RPC-driven writes the reducer
//   doesn't see. Resync expands as each entity migrates.

use std::sync::Arc;

use agentmux_common::ipc::Event;
use tokio::sync::{broadcast, Mutex};

use crate::backend::obj::{
    Block, Client, LayoutActionData, LayoutState as PersistedLayoutState, LeafOrderEntry, Tab,
    Window, Workspace,
};
use crate::backend::storage::store::Store;
use crate::backend::wcore;
use crate::state::State;

/// Spawn the persist subscriber task. Runs until the broadcast
/// channel closes (i.e., the reducer's bus is dropped at process
/// shutdown). The `state` handle is used for workspace-scoped
/// full-resync after a `RecvError::Lagged`.
pub fn spawn_persist_subscriber(
    events_rx: broadcast::Receiver<Event>,
    mstore: Arc<Store>,
    state: Arc<Mutex<State>>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(run_persist_subscriber(events_rx, mstore, state))
}

async fn run_persist_subscriber(
    mut events_rx: broadcast::Receiver<Event>,
    mstore: Arc<Store>,
    state: Arc<Mutex<State>>,
) {
    tracing::info!(target: "srv-persist-subscriber", "[srv-persist-subscriber] started");
    loop {
        match events_rx.recv().await {
            Ok(event) => {
                if let Err(e) = apply_event_to_mstore(&event, &mstore) {
                    tracing::warn!(
                        target: "srv-persist-subscriber",
                        "[srv-persist-subscriber] apply failed for event {:?}: {}",
                        event_kind(&event),
                        e
                    );
                }
            }
            Err(broadcast::error::RecvError::Lagged(n)) => {
                // Workspace-scoped full resync. Subscriber dropped
                // `n` events; rather than guess which entities were
                // affected, snapshot the reducer's workspace map and
                // reconcile SQLite against it. RPC-migrated entities
                // (workspaces in E.2c.2) are correctly recovered;
                // not-yet-migrated entities (tab/block) are left
                // alone — see module-level docs.
                tracing::warn!(
                    target: "srv-persist-subscriber",
                    "[srv-persist-subscriber] dropped {} event(s) — running workspace resync",
                    n
                );
                if let Err(e) = resync_workspaces(&state, &mstore).await {
                    tracing::error!(
                        target: "srv-persist-subscriber",
                        "[srv-persist-subscriber] resync failed: {} — SQLite may diverge from reducer until next event",
                        e
                    );
                }
            }
            Err(broadcast::error::RecvError::Closed) => {
                tracing::info!(target: "srv-persist-subscriber", "[srv-persist-subscriber] bus closed — exiting");
                return;
            }
        }
    }
}

/// Snapshot `state.workspaces` and reconcile against SQLite —
/// INSERT/UPDATE only, no deletes. Workspace-scoped (tab/block
/// resync expands here as each entity's RPC layer migrates).
///
/// Why no delete-phase: during the migration window, workspaces
/// can be created OUTSIDE the reducer (e.g., the still-wcore-direct
/// `CreateWindow` flow calls `wcore::create_window_full` which
/// creates a workspace under the hood). Those workspaces aren't in
/// `state.workspaces`. If the subscriber lagged and we did a
/// delete-not-in-snapshot pass, we'd cascade-delete legitimate
/// workspaces — data loss triggered by bus pressure rather than a
/// user action. Far better for a stale workspace deleted-via-reducer
/// to linger on disk after a Lagged event than to lose a real one.
/// (codex P1 #615.)
///
/// Stale-after-Lagged scenario: reducer fires WorkspaceDeleted, the
/// subscriber drops it on Lagged, resync runs but only insert/updates.
/// SQLite still has the deleted workspace's row. The next user-driven
/// DeleteWorkspace (which falls through to `wcore::delete_workspace`
/// for unknown-to-reducer rows) cleans it up. Acceptable behaviour
/// during the migration; tightens once all RPC is reducer-driven.
///
/// Strategy:
///   1. Lock state, snapshot the workspace map (ids + names),
///      release the lock.
///   2. For each workspace in the snapshot: insert if missing,
///      no-op if name matches, update if name differs.
async fn resync_workspaces(
    state: &Arc<Mutex<State>>,
    mstore: &Store,
) -> Result<(), Box<dyn std::error::Error>> {
    // Snapshot under lock; release before any I/O.
    let snapshot: Vec<(String, String)> = {
        let s = state.lock().await;
        s.workspaces
            .values()
            .map(|w| (w.workspace_id.clone(), w.name.clone()))
            .collect()
    };

    for (workspace_id, name) in &snapshot {
        match mstore.get::<Workspace>(workspace_id)? {
            Some(existing) if existing.name == *name => {
                // Already in sync.
            }
            Some(mut existing) => {
                existing.name = name.clone();
                mstore.update(&mut existing)?;
            }
            None => {
                let mut ws = Workspace {
                    oid: workspace_id.clone(),
                    name: name.clone(),
                    ..Default::default()
                };
                mstore.insert(&mut ws)?;
            }
        }
    }
    Ok(())
}

/// Apply one reducer event to the on-disk store. Idempotent: each
/// arm checks for the entity's current SQLite state before writing
/// so duplicate events (from at-least-once delivery semantics) don't
/// produce duplicate rows or wcore errors.
///
/// Phase E.2c.2 — exposed at crate visibility so RPC handlers in
/// `service.rs` can apply events synchronously after dispatching
/// through the reducer. This closes the race where the async
/// subscriber hadn't yet written SQLite by the time a follow-up
/// RPC (e.g., `CreateTab` against a just-created workspace) tried
/// to read it. Calling this from the RPC handler followed by the
/// subscriber receiving the broadcast event is safe because the
/// arms are idempotent — the subscriber's later apply is a no-op.
pub(crate) fn apply_event_to_mstore(
    event: &Event,
    mstore: &Store,
) -> Result<(), Box<dyn std::error::Error>> {
    match event {
        Event::WorkspaceCreated {
            workspace_id, name, ..
        } => apply_workspace_created(mstore, workspace_id, name),
        Event::WorkspaceDeleted { workspace_id, .. } => {
            apply_workspace_deleted(mstore, workspace_id)
        }
        Event::TabCreated {
            workspace_id,
            tab_id,
            name,
            ..
        } => apply_tab_created(mstore, workspace_id, tab_id, name),
        Event::TabDeleted {
            workspace_id,
            tab_id,
            ..
        } => apply_tab_deleted(mstore, workspace_id, tab_id),
        Event::ActiveTabChanged {
            workspace_id,
            tab_id,
            ..
        } => apply_active_tab_changed(mstore, workspace_id, tab_id.as_deref()),
        Event::TabReordered {
            workspace_id,
            tab_id,
            new_index,
            ..
        } => apply_tab_reordered(mstore, workspace_id, tab_id, *new_index),
        Event::BlockCreated {
            tab_id,
            block_id,
            meta,
            ..
        } => apply_block_created(mstore, tab_id, block_id, meta),
        Event::BlockDeleted {
            tab_id, block_id, ..
        } => apply_block_deleted(mstore, tab_id, block_id),
        Event::SrvWindowOpened {
            window_id,
            workspace_id,
            ..
        } => apply_srv_window_opened(mstore, window_id, workspace_id),
        Event::SrvWindowClosed { window_id, .. } => apply_srv_window_closed(mstore, window_id),
        Event::SrvWindowWorkspaceChanged {
            window_id,
            workspace_id,
            ..
        } => apply_srv_window_workspace_changed(mstore, window_id, workspace_id),
        Event::TabsReorderedBulk {
            workspace_id,
            tab_ids,
            ..
        } => apply_tabs_reordered_bulk(mstore, workspace_id, tab_ids),
        Event::WorkspaceRenamed {
            workspace_id, name, ..
        } => apply_workspace_renamed(mstore, workspace_id, name),
        Event::TabRenamed { tab_id, name, .. } => apply_tab_renamed(mstore, tab_id, name),
        Event::WorkspaceMetaUpdated {
            workspace_id,
            meta_patch,
            ..
        } => apply_workspace_meta_updated(mstore, workspace_id, meta_patch),
        Event::WindowMetaUpdated {
            window_id,
            meta_patch,
            ..
        } => apply_window_meta_updated(mstore, window_id, meta_patch),
        Event::TabMetaUpdated {
            tab_id,
            meta_patch,
            ..
        } => apply_tab_meta_updated(mstore, tab_id, meta_patch),
        Event::BlockMetaUpdated {
            block_id,
            meta_patch,
            ..
        } => apply_block_meta_updated(mstore, block_id, meta_patch),
        Event::TabMoved {
            tab_id,
            src_workspace_id,
            dst_workspace_id,
            dst_index,
            new_src_active_tab_id,
            new_dst_active_tab_id,
            ..
        } => apply_tab_moved(
            mstore,
            tab_id,
            src_workspace_id,
            dst_workspace_id,
            *dst_index,
            new_src_active_tab_id.as_deref(),
            new_dst_active_tab_id.as_deref(),
        ),
        Event::BlockMoved {
            block_id,
            src_tab_id,
            dst_tab_id,
            dst_index,
            ..
        } => apply_block_moved(mstore, block_id, src_tab_id, dst_tab_id, *dst_index),
        Event::FocusedNodeChanged { tab_id, node_id, .. } => {
            apply_focused_node_changed(mstore, tab_id, node_id)
        }
        Event::MagnifiedNodeChanged { tab_id, node_id, .. } => {
            apply_magnified_node_changed(mstore, tab_id, node_id)
        }
        // Phase E.4.B — layout-tree persistence. These two are the
        // first tree-mutating layout events to persist through the
        // subscriber (rather than wcore-direct). They are behaviour-
        // neutral until a production caller dispatches LayoutClear /
        // LayoutSetTree (the Layout* arms are still dormant per
        // reducer.rs §"4 of 11"); they land the persistence path that
        // the UpdateObject→LayoutSetTree reroute (next phase) needs.
        // Insert is deliberately NOT here yet — its event carries op
        // params, not the resulting tree, so persisting it would require
        // re-running the algebra in the subscriber (a new divergence
        // surface). Delete WAS in the same boat until SPEC_864 site #6
        // gave `LayoutNodeDeleted` a `new_tree` (+ `tree_cleared`) and its
        // own arm below; Insert follows when a production caller needs it.
        Event::LayoutCleared { tab_id, .. } => apply_layout_cleared(mstore, tab_id),
        Event::LayoutTreeReplaced {
            tab_id,
            new_tree,
            slices,
            ..
        } => apply_layout_tree_replaced(mstore, tab_id, new_tree, slices.as_ref()),
        // SPEC_864 site #6 — LayoutNodeDeleted gets its own arm (not the
        // group below) because a delete is the one structural op that can
        // legitimately EMPTY the tree (root-orphan case). `tree_cleared`
        // disambiguates a real root-delete clear from a version-skewed
        // sender's absent `new_tree`, which must be ignored rather than
        // applied as an erase (same skew concern as the group below).
        // Leaforder handling follows the granular-arm convention: untouched
        // for a non-empty tree (frontend-recomputed geometry cache, not
        // read on reproject), cleared with everything else on a real
        // empty-tree clear (apply_layout_tree_replaced's clears path).
        Event::LayoutNodeDeleted {
            tab_id,
            new_tree,
            tree_cleared,
            ..
        } => match (new_tree.is_some(), tree_cleared) {
            (true, _) => apply_layout_tree_replaced(mstore, tab_id, new_tree, None),
            (false, true) => apply_layout_tree_replaced(mstore, tab_id, &None, None),
            (false, false) => Ok(()),
        },
        // The 7 granular structural arms each carry the reducer's resulting
        // tree in `new_tree` (post-op, post-balance), so persistence is the
        // same rootnode write as LayoutTreeReplaced — no algebra re-run in the
        // subscriber (the divergence surface we're avoiding). Focus/magnify
        // changes from these ops are not persisted here yet (the frontend
        // reconciles focus on load); that's a tracked follow-up.
        Event::LayoutNodeMoved { tab_id, new_tree, .. }
        | Event::LayoutNodesSwapped { tab_id, new_tree, .. }
        | Event::LayoutNodesResized { tab_id, new_tree, .. }
        | Event::LayoutNodeReplaced { tab_id, new_tree, .. }
        | Event::LayoutSplitHorizontalApplied { tab_id, new_tree, .. }
        | Event::LayoutSplitVerticalApplied { tab_id, new_tree, .. }
        | Event::LayoutNodeInsertedAtIndex { tab_id, new_tree, .. } => {
            // These structural ops NEVER legitimately produce an empty tree,
            // so a `None` here is only a version-skewed / pre-change sender
            // (`new_tree` is `#[serde(default)]`). Ignore it rather than
            // passing it to `apply_layout_tree_replaced`, which would treat
            // `None` as an intentional empty-tree clear and ERASE the
            // persisted layout (codex P2 on #1883). Persist only a real tree.
            match new_tree {
                // Granular events carry no client slices — tree-only write.
                Some(_) => apply_layout_tree_replaced(mstore, tab_id, new_tree, None),
                None => Ok(()),
            }
        }
        // SPEC_864 Phase 4 — APPEND to the pending queue (unlike the
        // slices path above, which REPLACEs it: pushing a slice without
        // processed actions is how the frontend ACKS the queue; the
        // backend enqueues by appending). Idempotent under the
        // double-apply contract (sync from the dispatch helper, again
        // from broadcast) by deduping on `actionid` — every writer
        // stamps a fresh UUID per action.
        Event::LayoutBackendActionsQueued { tab_id, actions, .. } => {
            apply_layout_backend_actions_queued(mstore, tab_id, actions)
        }
        // All other event variants are not domain-state mutations
        // (lifecycle, errors, snapshots). The subscriber ignores them.
        _ => Ok(()),
    }
}

fn apply_workspace_created(
    mstore: &Store,
    workspace_id: &str,
    name: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    if mstore.get::<Workspace>(workspace_id)?.is_some() {
        return Ok(()); // already present
    }
    let mut ws = Workspace {
        oid: workspace_id.to_string(),
        name: name.to_string(),
        ..Default::default()
    };
    mstore.insert(&mut ws)?;
    Ok(())
}

fn apply_workspace_deleted(
    mstore: &Store,
    workspace_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    if mstore.get::<Workspace>(workspace_id)?.is_none() {
        return Ok(()); // already gone
    }
    wcore::delete_workspace(mstore, workspace_id)?;
    Ok(())
}

fn apply_tab_created(
    mstore: &Store,
    workspace_id: &str,
    tab_id: &str,
    name: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    // F1.A — wrap the insert-layout + insert-tab + update-workspace
    // sequence in a single SQLite transaction so a partial failure
    // can't leave a tab without a layout or a workspace whose
    // tab_ids points at a tab that doesn't exist on disk.
    mstore.with_tx(|tx| {
        if tx.get::<Tab>(tab_id)?.is_none() {
            // Phase E.2c.2 — create a LayoutState row alongside the
            // Tab so reducer-originated tabs have a `Tab.layoutstate`
            // to seed (e.g. `setup_torn_off_block_layout`'s
            // `seed_layout_via_reducer` call) just like wcore-created
            // tabs. Mirrors the wcore::create_tab pattern. (codex P2 #614.)
            let mut layout = PersistedLayoutState {
                oid: uuid::Uuid::new_v4().to_string(),
                rootnode: None,
                magnifiednodeid: String::new(),
                focusednodeid: String::new(),
                leaforder: None,
                pendingbackendactions: None,
                meta: None,
                ..Default::default()
            };
            tx.insert(&mut layout)?;
            let mut tab = Tab {
                oid: tab_id.to_string(),
                name: name.to_string(),
                layoutstate: layout.oid.clone(),
                ..Default::default()
            };
            tx.insert(&mut tab)?;
        }
        // Idempotently link tab into the workspace's tabids.
        if let Some(mut ws) = tx.get::<Workspace>(workspace_id)? {
            let already_present = ws.tabids.iter().any(|t| t == tab_id)
                || ws.pinnedtabids.iter().any(|t| t == tab_id);
            if !already_present {
                ws.tabids.push(tab_id.to_string());
                tx.update(&mut ws)?;
            }
        }
        Ok(())
    })?;
    Ok(())
}

fn apply_tab_deleted(
    mstore: &Store,
    workspace_id: &str,
    tab_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    if mstore.get::<Tab>(tab_id)?.is_none() {
        return Ok(()); // already gone (or never existed)
    }
    // wcore::delete_tab handles unlinking from the workspace's
    // tabids, deleting the tab's blocks + layout, and the tab row
    // itself. Returns NotFound errors if any of those don't exist —
    // we surface those as the operation having already happened.
    match wcore::delete_tab(mstore, workspace_id, tab_id) {
        Ok(()) => Ok(()),
        Err(crate::backend::storage::StoreError::NotFound) => Ok(()),
        Err(e) => Err(Box::new(e)),
    }
}

fn apply_active_tab_changed(
    mstore: &Store,
    workspace_id: &str,
    tab_id: Option<&str>,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(mut ws) = mstore.get::<Workspace>(workspace_id)? else {
        return Ok(()); // workspace already gone — nothing to update
    };
    let new = tab_id.unwrap_or("").to_string();
    if ws.activetabid == new {
        return Ok(());
    }
    ws.activetabid = new;
    mstore.update(&mut ws)?;
    Ok(())
}

fn apply_tab_reordered(
    mstore: &Store,
    workspace_id: &str,
    tab_id: &str,
    new_index: u32,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(mut ws) = mstore.get::<Workspace>(workspace_id)? else {
        return Ok(());
    };
    // Pinning was removed from AgentMux but legacy SQLite databases
    // can still have entries in `Workspace.pinnedtabids`; bootstrap
    // surfaces those as regular tabs in the reducer's `tab_ids`. The
    // reducer can therefore emit a TabReordered for a tab that on
    // disk still lives in `pinnedtabids`. Search both lists so the
    // reorder lands wherever the tab actually is. (codex P2 #617.)
    let target_index = new_index as usize;
    if let Some(current_pos) = ws.tabids.iter().position(|t| t == tab_id) {
        let len = ws.tabids.len();
        let target = target_index.min(len.saturating_sub(1));
        if current_pos == target {
            return Ok(());
        }
        let id = ws.tabids.remove(current_pos);
        ws.tabids.insert(target, id);
        mstore.update(&mut ws)?;
    } else if let Some(current_pos) = ws.pinnedtabids.iter().position(|t| t == tab_id) {
        let len = ws.pinnedtabids.len();
        let target = target_index.min(len.saturating_sub(1));
        if current_pos == target {
            return Ok(());
        }
        let id = ws.pinnedtabids.remove(current_pos);
        ws.pinnedtabids.insert(target, id);
        mstore.update(&mut ws)?;
    }
    // Tab not in either list — silent no-op (idempotent).
    Ok(())
}

fn apply_block_created(
    mstore: &Store,
    tab_id: &str,
    block_id: &str,
    meta: &serde_json::Value,
) -> Result<(), Box<dyn std::error::Error>> {
    // F1.A — block insert + tab.blockids update wrapped in one tx.
    let meta_map: crate::backend::obj::MetaMapType = match meta {
        serde_json::Value::Object(_) => {
            serde_json::from_value(meta.clone()).unwrap_or_default()
        }
        _ => Default::default(),
    };
    mstore.with_tx(|tx| {
        if tx.get::<Block>(block_id)?.is_none() {
            // Phase E.2c.4 — write the meta map carried in the event
            // (`view`, layout hints, etc.) so reducer-routed
            // CreateBlock RPC produces blocks with valid view/meta in
            // SQLite. Without this, the frontend sees a block with
            // empty meta and renders a blank pane.
            let mut block = Block {
                oid: block_id.to_string(),
                parentoref: format!("tab:{}", tab_id),
                meta: meta_map,
                ..Default::default()
            };
            tx.insert(&mut block)?;
        }
        if let Some(mut tab) = tx.get::<Tab>(tab_id)? {
            if !tab.blockids.iter().any(|b| b == block_id) {
                tab.blockids.push(block_id.to_string());
                tx.update(&mut tab)?;
            }
        }
        Ok(())
    })?;
    Ok(())
}

fn apply_block_deleted(
    mstore: &Store,
    tab_id: &str,
    block_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    if mstore.get::<Block>(block_id)?.is_none() {
        return Ok(()); // already gone
    }
    match wcore::delete_block(mstore, tab_id, block_id) {
        Ok(()) => Ok(()),
        Err(crate::backend::storage::StoreError::NotFound) => Ok(()),
        Err(e) => Err(Box::new(e)),
    }
}

/// Phase E.5 — record/update a window's workspace pointer on the
/// persisted Window row. Idempotent: inserts if missing, updates
/// if `workspaceid` differs, no-ops if identical.
///
/// Also keeps `Client.windowids` in sync — appends the new window_id
/// when the Window row is freshly created. The legacy
/// `wcore::create_window_full` path does this too; without it, the
/// Window row exists but `GetClientData` / focus-order logic can't
/// see it. (codex P1 #619.)
fn apply_srv_window_opened(
    mstore: &Store,
    window_id: &str,
    workspace_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    // F1.A — Window upsert + Client.windowids update wrapped in one
    // tx so a partial failure can't leave a Window row that
    // `GetClientData` doesn't see (or a `windowids` entry pointing
    // at a Window row that wasn't written).
    mstore.with_tx(|tx| {
        let was_new = match tx.get::<Window>(window_id)? {
            Some(existing) if existing.workspaceid == workspace_id => false,
            Some(mut existing) => {
                existing.workspaceid = workspace_id.to_string();
                tx.update(&mut existing)?;
                false
            }
            None => {
                let mut window = Window {
                    oid: window_id.to_string(),
                    workspaceid: workspace_id.to_string(),
                    ..Default::default()
                };
                tx.insert(&mut window)?;
                true
            }
        };
        if was_new {
            // Inline `wcore::get_client` against the tx-handle to keep
            // the Client.windowids update in the same transaction.
            let clients = tx.get_all::<Client>()?;
            if let Some(mut client) = clients.into_iter().next() {
                if !client.windowids.iter().any(|id| id == window_id) {
                    client.windowids.push(window_id.to_string());
                    tx.update(&mut client)?;
                }
            }
        }
        Ok(())
    })?;
    Ok(())
}

/// Phase E.5 — delete the persisted Window row + remove the id
/// from `Client.windowids`. Mirrors `wcore::close_window`'s order:
/// prune client.windowids FIRST (so any read between the two ops
/// doesn't see a dangling id), then delete the Window row.
/// (codex P1 #619.)
///
/// `pub(crate)`: besides the `SrvWindowClosed` subscriber arm, the
/// `CloseWindow` handler calls this directly on its divergence path —
/// when the reducer never knew the window (silent no-op, no event) but
/// the store still holds it (#2051). Deliberately defensive/idempotent:
/// both callers rely on the already-gone cases being safe.
pub(crate) fn apply_srv_window_closed(
    mstore: &Store,
    window_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    // F1.A — Client.windowids prune + Window-row delete in one tx.
    // Order matters (mirrors `wcore::close_window`): prune client
    // FIRST so any read between the two ops doesn't see a dangling
    // id pointing at a Window row that's already been removed.
    mstore.with_tx(|tx| {
        let window_exists = tx.get::<Window>(window_id)?.is_some();
        // Always defensively prune the client list — handles the
        // already-gone case where divergence left a stale entry.
        let clients = tx.get_all::<Client>()?;
        if let Some(mut client) = clients.into_iter().next() {
            if client.windowids.iter().any(|id| id == window_id) {
                client.windowids.retain(|id| id != window_id);
                tx.update(&mut client)?;
            }
        }
        if window_exists {
            tx.delete::<Window>(window_id)?;
        }
        Ok(())
    })?;
    Ok(())
}

/// Phase E.5 — same shape as `apply_srv_window_opened` for the
/// upsert behavior; separate function for log-clarity since the
/// emitted event is distinct.
fn apply_srv_window_workspace_changed(
    mstore: &Store,
    window_id: &str,
    workspace_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    apply_srv_window_opened(mstore, window_id, workspace_id)
}

/// Phase E.5.3 — replace the workspace's `tabids` with the new
/// list and drain any legacy `pinnedtabids` rows. Pinning was a
/// Waveterm feature removed from AgentMux; bootstrap merges legacy
/// pinned tabs into the reducer's `tab_ids`, so the next reorder
/// is the canonical full ordering. Leaving stale `pinnedtabids`
/// in SQLite would cause UI double-insertion (`workspace.tsx`
/// builds the displayed tab list as `[...pinnedtabids, ...tabids]`).
fn apply_tabs_reordered_bulk(
    mstore: &Store,
    workspace_id: &str,
    tab_ids: &[String],
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(mut ws) = mstore.get::<Workspace>(workspace_id)? else {
        return Ok(());
    };
    let tabids_match = ws.tabids == tab_ids;
    let pinned_clear = ws.pinnedtabids.is_empty();
    if tabids_match && pinned_clear {
        return Ok(());
    }
    ws.tabids = tab_ids.to_vec();
    ws.pinnedtabids.clear();
    mstore.update(&mut ws)?;
    Ok(())
}

/// Phase E.5.3 — rename a persisted workspace.
fn apply_workspace_renamed(
    mstore: &Store,
    workspace_id: &str,
    name: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(mut ws) = mstore.get::<Workspace>(workspace_id)? else {
        return Ok(());
    };
    if ws.name == name {
        return Ok(());
    }
    ws.name = name.to_string();
    mstore.update(&mut ws)?;
    Ok(())
}

/// Phase E.5.3 — rename a persisted tab.
fn apply_tab_renamed(
    mstore: &Store,
    tab_id: &str,
    name: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(mut tab) = mstore.get::<Tab>(tab_id)? else {
        return Ok(());
    };
    if tab.name == name {
        return Ok(());
    }
    tab.name = name.to_string();
    mstore.update(&mut tab)?;
    Ok(())
}

/// Phase E.4 (Option A) — write the reducer-emitted focused node id
/// onto the tab's `LayoutState.focusednodeid` column. Tab→layout is
/// a join through `Tab.layoutstate`. Idempotent: silent no-op when the
/// tab is unknown (deleted concurrently), the layout row is missing
/// (legacy tab without one), or the value already matches. Other
/// `LayoutState` fields stay on their existing wcore-direct path until
/// Option B lands.
fn apply_focused_node_changed(
    mstore: &Store,
    tab_id: &str,
    node_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(tab) = mstore.get::<Tab>(tab_id)? else {
        return Ok(());
    };
    if tab.layoutstate.is_empty() {
        return Ok(());
    }
    let Some(mut layout) = mstore.get::<PersistedLayoutState>(&tab.layoutstate)? else {
        return Ok(());
    };
    if layout.focusednodeid == node_id {
        return Ok(());
    }
    layout.focusednodeid = node_id.to_string();
    mstore.update(&mut layout)?;
    Ok(())
}

/// Phase E.4 (Option A) — write the reducer-emitted magnified node id
/// onto the tab's `LayoutState.magnifiednodeid` column. Same shape as
/// `apply_focused_node_changed`.
fn apply_magnified_node_changed(
    mstore: &Store,
    tab_id: &str,
    node_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(tab) = mstore.get::<Tab>(tab_id)? else {
        return Ok(());
    };
    if tab.layoutstate.is_empty() {
        return Ok(());
    }
    let Some(mut layout) = mstore.get::<PersistedLayoutState>(&tab.layoutstate)? else {
        return Ok(());
    };
    if layout.magnifiednodeid == node_id {
        return Ok(());
    }
    layout.magnifiednodeid = node_id.to_string();
    mstore.update(&mut layout)?;
    Ok(())
}

/// Phase E.4.B — apply a `LayoutCleared` event: wipe the tab's persisted
/// layout tree and the focus/magnify ids (and the leaforder cache — an
/// empty tree has no leaves). Mirrors
/// `reducer::layout::handle_layout_clear`. Idempotent: silent no-op when
/// the tab is unknown, the layout row is missing, or it is already clear
/// (so a double-apply — once synchronously from the RPC handler, once
/// from the broadcast — does not churn the version).
fn apply_layout_cleared(
    mstore: &Store,
    tab_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(tab) = mstore.get::<Tab>(tab_id)? else {
        return Ok(());
    };
    if tab.layoutstate.is_empty() {
        return Ok(());
    }
    let Some(mut layout) = mstore.get::<PersistedLayoutState>(&tab.layoutstate)? else {
        return Ok(());
    };
    if layout.rootnode.is_none()
        && layout.focusednodeid.is_empty()
        && layout.magnifiednodeid.is_empty()
        && layout.leaforder.is_none()
    {
        return Ok(());
    }
    layout.rootnode = None;
    layout.focusednodeid = String::new();
    layout.magnifiednodeid = String::new();
    layout.leaforder = None;
    mstore.update(&mut layout)?;
    Ok(())
}

/// Phase E.4.B — apply a `LayoutTreeReplaced` event: write the reducer-
/// supplied tree onto `LayoutState.rootnode`. Mirrors
/// `reducer::layout::handle_layout_set_tree` — when the new tree is
/// empty, also clear the focus/magnify ids (and leaforder) so they don't
/// dangle at nodes that no longer exist.
///
/// SPEC_864 Phase 2 — when the event carries `slices` (the
/// `UpdateObject`→`LayoutSetTree` full-row push), the frontend-owned
/// columns are REPLACED to mirror the legacy `update_raw` whole-row write
/// this route retires: leaforder / pendingbackendactions are written
/// verbatim (absent = clear — pushing without processed actions is how
/// the frontend acks its queue); focus/magnify strings overwrite (empty =
/// clear). `slices: None` (granular events, tree-only callers) leaves
/// those columns untouched — leaforder is a frontend-recomputed geometry
/// cache (`getLeafOrder` in `layoutGeometry.ts`) and is NOT read on
/// reproject (`persist::bootstrap_state_from_mstore` reads
/// rootnode/focus/magnify only). Idempotent: no-op (no version bump)
/// when nothing changes.
fn apply_layout_tree_replaced(
    mstore: &Store,
    tab_id: &str,
    new_tree: &Option<agentmux_common::LayoutNode>,
    slices: Option<&agentmux_common::LayoutClientSlices>,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(tab) = mstore.get::<Tab>(tab_id)? else {
        return Ok(());
    };
    if tab.layoutstate.is_empty() {
        return Ok(());
    }
    let Some(mut layout) = mstore.get::<PersistedLayoutState>(&tab.layoutstate)? else {
        return Ok(());
    };
    let clears = new_tree.is_none();

    // Compute the candidate values for every column this event owns, then
    // write once only if something actually changed (idempotency contract:
    // a double-apply — sync from the RPC handler, again from broadcast —
    // must not churn the version).
    let new_focused = match (slices, clears) {
        (_, true) => String::new(),
        (Some(s), false) => s.focused_node_id.clone(),
        (None, false) => layout.focusednodeid.clone(),
    };
    let new_magnified = match (slices, clears) {
        (_, true) => String::new(),
        (Some(s), false) => s.magnified_node_id.clone(),
        (None, false) => layout.magnifiednodeid.clone(),
    };
    let new_leaforder: Option<Vec<LeafOrderEntry>> = match (slices, clears) {
        (_, true) => None,
        (Some(s), false) => match &s.leaforder {
            Some(v) if !v.is_null() => Some(serde_json::from_value(v.clone())?),
            _ => None,
        },
        (None, false) => layout.leaforder.clone(),
    };
    let new_pending: Option<Vec<LayoutActionData>> = match slices {
        // The pending queue is frontend-acked state, not tree geometry —
        // an empty-tree push still replaces it (unlike focus/leaforder,
        // which dangle without a tree).
        Some(s) => match &s.pending_backend_actions {
            Some(v) if !v.is_null() => Some(serde_json::from_value(v.clone())?),
            _ => None,
        },
        None => layout.pendingbackendactions.clone(),
    };

    let already = layout.rootnode == *new_tree
        && layout.focusednodeid == new_focused
        && layout.magnifiednodeid == new_magnified
        && layout.leaforder == new_leaforder
        && layout.pendingbackendactions == new_pending;
    if already {
        return Ok(());
    }
    layout.rootnode = new_tree.clone();
    layout.focusednodeid = new_focused;
    layout.magnifiednodeid = new_magnified;
    layout.leaforder = new_leaforder;
    layout.pendingbackendactions = new_pending;
    mstore.update(&mut layout)?;
    Ok(())
}

/// SPEC_864 Phase 4 — apply a `LayoutBackendActionsQueued` event:
/// APPEND the reducer-validated actions to
/// `LayoutState.pendingbackendactions`. The queue is a backend→frontend
/// mailbox: the backend appends here; the frontend drains it by pushing
/// a full slice without the processed actions (the REPLACE path in
/// `apply_layout_tree_replaced`).
///
/// Idempotency: the event is applied twice on the happy path (sync in
/// `queue_layout_actions_via_reducer`, again by the subscriber task
/// consuming the broadcast), so actions whose `actionid` is already in
/// the queue are skipped. Every writer stamps a fresh UUID per action;
/// an (out-of-contract) empty `actionid` dedupes against other empties,
/// which is the safe direction — skip rather than duplicate.
fn apply_layout_backend_actions_queued(
    mstore: &Store,
    tab_id: &str,
    actions: &serde_json::Value,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(tab) = mstore.get::<Tab>(tab_id)? else {
        return Ok(());
    };
    if tab.layoutstate.is_empty() {
        return Ok(());
    }
    let Some(mut layout) = mstore.get::<PersistedLayoutState>(&tab.layoutstate)? else {
        return Ok(());
    };
    let incoming: Vec<LayoutActionData> = serde_json::from_value(actions.clone())?;
    let mut pending = layout.pendingbackendactions.take().unwrap_or_default();
    let before = pending.len();
    for action in incoming {
        if !pending.iter().any(|p| p.actionid == action.actionid) {
            pending.push(action);
        }
    }
    if pending.len() == before {
        // Every action already queued (double-apply) — no version churn.
        layout.pendingbackendactions = Some(pending);
        return Ok(());
    }
    layout.pendingbackendactions = Some(pending);
    mstore.update(&mut layout)?;
    Ok(())
}

/// Phase E.5.3 — apply a meta-patch to a workspace's `meta` map.
/// Reducer doesn't track meta in `WorkspaceRecord`; this subscriber
/// is the sole authority that mutates persisted meta. Patch is a
/// JSON object that merges shallow-key-by-shallow-key on top of the
/// existing meta. `null` values in the patch delete the key.
fn apply_workspace_meta_updated(
    mstore: &Store,
    workspace_id: &str,
    meta_patch: &serde_json::Value,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(mut ws) = mstore.get::<Workspace>(workspace_id)? else {
        return Ok(());
    };
    if merge_meta_patch(&mut ws.meta, meta_patch) {
        mstore.update(&mut ws)?;
    }
    Ok(())
}

/// Phase E.5.x (issue #855) — apply a meta-patch to a window's `meta`
/// map. Same shape as `apply_workspace_meta_updated`. Silent no-op if
/// the window doesn't exist in mstore (preserves the idempotency
/// contract — duplicate or stale events fold to no-op).
fn apply_window_meta_updated(
    mstore: &Store,
    window_id: &str,
    meta_patch: &serde_json::Value,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(mut window) = mstore.get::<Window>(window_id)? else {
        return Ok(());
    };
    if merge_meta_patch(&mut window.meta, meta_patch) {
        mstore.update(&mut window)?;
    }
    Ok(())
}

/// Phase E.5.3 — apply a meta-patch to a tab's `meta` map.
fn apply_tab_meta_updated(
    mstore: &Store,
    tab_id: &str,
    meta_patch: &serde_json::Value,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(mut tab) = mstore.get::<Tab>(tab_id)? else {
        return Ok(());
    };
    if merge_meta_patch(&mut tab.meta, meta_patch) {
        mstore.update(&mut tab)?;
    }
    Ok(())
}

/// Phase E.5.3 — apply a meta-patch to a block's `meta` map.
fn apply_block_meta_updated(
    mstore: &Store,
    block_id: &str,
    meta_patch: &serde_json::Value,
) -> Result<(), Box<dyn std::error::Error>> {
    let Some(mut block) = mstore.get::<Block>(block_id)? else {
        return Ok(());
    };
    if merge_meta_patch(&mut block.meta, meta_patch) {
        mstore.update(&mut block)?;
    }
    Ok(())
}

/// Phase E.5.5 — apply a `TabMoved` event to SQLite.
/// Idempotent: detects already-moved state (tab in dst, not in src)
/// and returns `Ok(())` without re-mutating.
///
/// What this writes:
/// * Removes `tab_id` from `src_workspace.tabids` (and `pinnedtabids`
///   for legacy rows — bootstrap merges them into reducer state, so
///   any leftover here is a stray legacy entry that should be drained).
/// * Updates `src_workspace.activetabid` to `new_src_active_tab_id`
///   (which may be empty when the source emptied).
/// * Inserts `tab_id` at `dst_index` in `dst_workspace.tabids`,
///   clamping to the dst list length.
/// * Updates the `Tab` row's parent ref so loaders find it under
///   the new workspace.
fn apply_tab_moved(
    mstore: &Store,
    tab_id: &str,
    src_workspace_id: &str,
    dst_workspace_id: &str,
    dst_index: u32,
    new_src_active_tab_id: Option<&str>,
    new_dst_active_tab_id: Option<&str>,
) -> Result<(), Box<dyn std::error::Error>> {
    // F1.A — both workspaces' updates wrapped in one tx so a partial
    // failure can't leave the tab in both src.tabids and dst.tabids
    // simultaneously, or in neither.
    mstore.with_tx(|tx| {
        // Source workspace: remove the tab and update activetabid.
        if let Some(mut src_ws) = tx.get::<Workspace>(src_workspace_id)? {
            let len_before_tabids = src_ws.tabids.len();
            let len_before_pinned = src_ws.pinnedtabids.len();
            src_ws.tabids.retain(|id| id != tab_id);
            src_ws.pinnedtabids.retain(|id| id != tab_id);
            let new_active = new_src_active_tab_id.unwrap_or("").to_string();
            let active_changed = src_ws.activetabid != new_active;
            if src_ws.tabids.len() != len_before_tabids
                || src_ws.pinnedtabids.len() != len_before_pinned
                || active_changed
            {
                src_ws.activetabid = new_active;
                tx.update(&mut src_ws)?;
            }
        }

        // Dest workspace: insert at clamped index + update
        // active_tab_id if the event carries one (codex P2 #621).
        // Skip insert if the tab is already present (idempotent on
        // duplicate delivery).
        if let Some(mut dst_ws) = tx.get::<Workspace>(dst_workspace_id)? {
            let mut changed = false;
            if !dst_ws.tabids.iter().any(|id| id == tab_id) {
                let clamped = (dst_index as usize).min(dst_ws.tabids.len());
                dst_ws.tabids.insert(clamped, tab_id.to_string());
                changed = true;
            }
            if let Some(new_active) = new_dst_active_tab_id {
                if dst_ws.activetabid != new_active {
                    dst_ws.activetabid = new_active.to_string();
                    changed = true;
                }
            }
            if changed {
                tx.update(&mut dst_ws)?;
            }
        }

        // No per-Tab parent ref to update — the workspace owns the
        // parentage relationship via its `tabids` list.
        Ok(())
    })?;
    Ok(())
}

/// Phase E.5.5 — apply a `BlockMoved` event to SQLite.
/// Handles both cross-tab moves and intra-tab repositioning.
/// Idempotent on re-delivery (checks current parent before mutating).
fn apply_block_moved(
    mstore: &Store,
    block_id: &str,
    src_tab_id: &str,
    dst_tab_id: &str,
    dst_index: u32,
) -> Result<(), Box<dyn std::error::Error>> {
    // F1.A — src.blockids + dst.blockids + block.parentoref updates
    // wrapped in one tx so a partial failure can't leave the block in
    // both tabs' blockids or with parentoref pointing at the wrong
    // tab.
    mstore.with_tx(|tx| {
        if src_tab_id == dst_tab_id {
            // Intra-tab reposition: remove + re-insert in the same tab.
            if let Some(mut tab) = tx.get::<Tab>(src_tab_id)? {
                tab.blockids.retain(|id| id != block_id);
                let clamped = (dst_index as usize).min(tab.blockids.len());
                tab.blockids.insert(clamped, block_id.to_string());
                tx.update(&mut tab)?;
            }
            return Ok(());
        }

        // Cross-tab: remove from src.
        if let Some(mut src_tab) = tx.get::<Tab>(src_tab_id)? {
            let before = src_tab.blockids.len();
            src_tab.blockids.retain(|id| id != block_id);
            if src_tab.blockids.len() != before {
                tx.update(&mut src_tab)?;
            }
        }

        // Insert into dst (skip if already there).
        if let Some(mut dst_tab) = tx.get::<Tab>(dst_tab_id)? {
            if !dst_tab.blockids.iter().any(|id| id == block_id) {
                let clamped = (dst_index as usize).min(dst_tab.blockids.len());
                dst_tab.blockids.insert(clamped, block_id.to_string());
                tx.update(&mut dst_tab)?;
            }
        }

        // Block row: update parent.
        if let Some(mut block) = tx.get::<Block>(block_id)? {
            let new_parent = format!("tab:{}", dst_tab_id);
            if block.parentoref != new_parent {
                block.parentoref = new_parent;
                tx.update(&mut block)?;
            }
        }

        Ok(())
    })?;
    Ok(())
}

/// Phase E.5.3 — shallow merge a JSON object patch into a
/// `MetaMapType`. Mirrors `backend::obj::merge_meta` semantics so
/// `UpdateObjectMeta` keeps behaving the same way after the reducer
/// migration:
/// - Keys ending in `:*` with a `true` value clear all keys with
///   that prefix (e.g. `{"term:*": true}` removes every `term*`
///   key) before regular merging.
/// - `null` patch values delete the corresponding key.
/// - Other values replace the key.
/// Returns `true` if anything actually changed.
fn merge_meta_patch(
    meta: &mut crate::backend::obj::MetaMapType,
    patch: &serde_json::Value,
) -> bool {
    let serde_json::Value::Object(patch_map) = patch else {
        return false;
    };
    let mut changed = false;
    // First pass: section-clear keys (`prefix:*` with `true`).
    for (k, v) in patch_map {
        if !k.ends_with(":*") {
            continue;
        }
        if !matches!(v, serde_json::Value::Bool(true)) {
            continue;
        }
        let prefix = k.trim_end_matches(":*");
        if prefix.is_empty() {
            continue;
        }
        let prefix_colon = format!("{prefix}:");
        let before = meta.len();
        meta.retain(|k2, _| k2 != prefix && !k2.starts_with(&prefix_colon));
        if meta.len() != before {
            changed = true;
        }
    }
    // Second pass: regular merges and null deletes.
    for (k, v) in patch_map {
        if k.ends_with(":*") {
            continue;
        }
        if v.is_null() {
            if meta.remove(k).is_some() {
                changed = true;
            }
            continue;
        }
        match meta.get(k) {
            Some(existing) if existing == v => {}
            _ => {
                meta.insert(k.clone(), v.clone());
                changed = true;
            }
        }
    }
    changed
}

/// Compact textual identifier for an event (debug logging only).
fn event_kind(event: &Event) -> &'static str {
    match event {
        Event::WorkspaceCreated { .. } => "WorkspaceCreated",
        Event::WorkspaceDeleted { .. } => "WorkspaceDeleted",
        Event::TabCreated { .. } => "TabCreated",
        Event::TabDeleted { .. } => "TabDeleted",
        Event::ActiveTabChanged { .. } => "ActiveTabChanged",
        Event::TabReordered { .. } => "TabReordered",
        Event::SrvWindowOpened { .. } => "SrvWindowOpened",
        Event::SrvWindowClosed { .. } => "SrvWindowClosed",
        Event::SrvWindowWorkspaceChanged { .. } => "SrvWindowWorkspaceChanged",
        Event::TabsReorderedBulk { .. } => "TabsReorderedBulk",
        Event::WorkspaceRenamed { .. } => "WorkspaceRenamed",
        Event::TabRenamed { .. } => "TabRenamed",
        Event::WorkspaceMetaUpdated { .. } => "WorkspaceMetaUpdated",
        Event::WindowMetaUpdated { .. } => "WindowMetaUpdated",
        Event::TabMetaUpdated { .. } => "TabMetaUpdated",
        Event::BlockMetaUpdated { .. } => "BlockMetaUpdated",
        Event::TabMoved { .. } => "TabMoved",
        Event::BlockMoved { .. } => "BlockMoved",
        Event::BlockCreated { .. } => "BlockCreated",
        Event::BlockDeleted { .. } => "BlockDeleted",
        Event::FocusedNodeChanged { .. } => "FocusedNodeChanged",
        Event::MagnifiedNodeChanged { .. } => "MagnifiedNodeChanged",
        _ => "Other",
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::storage::store::Store;

    fn store() -> Arc<Store> {
        // In-memory SQLite for tests (matches existing mstore test pattern).
        let store = Store::open_in_memory().expect("in-memory mstore");
        Arc::new(store)
    }

    #[test]
    fn workspace_created_inserts_row() {
        let s = store();
        apply_event_to_mstore(
            &Event::WorkspaceCreated {
                workspace_id: "ws-1".into(),
                name: "Alpha".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        let ws = s.get::<Workspace>("ws-1").unwrap().unwrap();
        assert_eq!(ws.name, "Alpha");
    }

    #[test]
    fn workspace_created_idempotent_on_duplicate() {
        let s = store();
        let ev = Event::WorkspaceCreated {
            workspace_id: "ws-1".into(),
            name: "Alpha".into(),
            version: 1,
        };
        apply_event_to_mstore(&ev, &s).unwrap();
        // Second application should not error.
        apply_event_to_mstore(&ev, &s).unwrap();
        // Original name preserved (no overwrite).
        let ws = s.get::<Workspace>("ws-1").unwrap().unwrap();
        assert_eq!(ws.name, "Alpha");
    }

    #[test]
    fn workspace_deleted_silent_when_missing() {
        let s = store();
        apply_event_to_mstore(
            &Event::WorkspaceDeleted {
                workspace_id: "ghost".into(),
                block_ids: Vec::new(),
                version: 1,
            },
            &s,
        )
        .unwrap();
    }

    #[test]
    fn tab_created_links_into_workspace() {
        let s = store();
        apply_event_to_mstore(
            &Event::WorkspaceCreated {
                workspace_id: "ws-1".into(),
                name: "Alpha".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::TabCreated {
                workspace_id: "ws-1".into(),
                tab_id: "tab-1".into(),
                name: "Tab".into(),
                version: 2,
            },
            &s,
        )
        .unwrap();
        let ws = s.get::<Workspace>("ws-1").unwrap().unwrap();
        assert_eq!(ws.tabids, vec!["tab-1".to_string()]);
        let tab = s.get::<Tab>("tab-1").unwrap().unwrap();
        assert_eq!(tab.name, "Tab");
    }

    #[test]
    fn tab_created_idempotent_on_duplicate_link() {
        let s = store();
        apply_event_to_mstore(
            &Event::WorkspaceCreated {
                workspace_id: "ws-1".into(),
                name: "Alpha".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        let ev = Event::TabCreated {
            workspace_id: "ws-1".into(),
            tab_id: "tab-1".into(),
            name: "Tab".into(),
            version: 2,
        };
        apply_event_to_mstore(&ev, &s).unwrap();
        apply_event_to_mstore(&ev, &s).unwrap();
        let ws = s.get::<Workspace>("ws-1").unwrap().unwrap();
        assert_eq!(ws.tabids.len(), 1);
    }

    #[test]
    fn active_tab_changed_updates_workspace() {
        let s = store();
        apply_event_to_mstore(
            &Event::WorkspaceCreated {
                workspace_id: "ws-1".into(),
                name: "Alpha".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::TabCreated {
                workspace_id: "ws-1".into(),
                tab_id: "tab-1".into(),
                name: "Tab".into(),
                version: 2,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::ActiveTabChanged {
                workspace_id: "ws-1".into(),
                tab_id: Some("tab-1".into()),
                version: 3,
            },
            &s,
        )
        .unwrap();
        let ws = s.get::<Workspace>("ws-1").unwrap().unwrap();
        assert_eq!(ws.activetabid, "tab-1");
    }

    #[test]
    fn active_tab_changed_to_none_clears_activetabid() {
        let s = store();
        apply_event_to_mstore(
            &Event::WorkspaceCreated {
                workspace_id: "ws-1".into(),
                name: "Alpha".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::TabCreated {
                workspace_id: "ws-1".into(),
                tab_id: "tab-1".into(),
                name: "Tab".into(),
                version: 2,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::ActiveTabChanged {
                workspace_id: "ws-1".into(),
                tab_id: Some("tab-1".into()),
                version: 3,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::ActiveTabChanged {
                workspace_id: "ws-1".into(),
                tab_id: None,
                version: 4,
            },
            &s,
        )
        .unwrap();
        let ws = s.get::<Workspace>("ws-1").unwrap().unwrap();
        assert_eq!(ws.activetabid, "");
    }

    #[test]
    fn tab_created_provisions_layoutstate() {
        let s = store();
        apply_event_to_mstore(
            &Event::WorkspaceCreated {
                workspace_id: "ws-1".into(),
                name: "Alpha".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::TabCreated {
                workspace_id: "ws-1".into(),
                tab_id: "tab-1".into(),
                name: "Tab".into(),
                version: 2,
            },
            &s,
        )
        .unwrap();
        let tab = s.get::<Tab>("tab-1").unwrap().unwrap();
        // codex P2 #614: tab must reference a real LayoutState row,
        // not be left with empty layoutstate.
        assert!(!tab.layoutstate.is_empty());
        let layout = s
            .get::<PersistedLayoutState>(&tab.layoutstate)
            .unwrap()
            .expect("layout row must exist");
        assert_eq!(layout.oid, tab.layoutstate);
    }

    // ── Layout-tree persistence (Phase E.4.B) ───────────────────────────
    fn leaf(id: &str, block: &str) -> agentmux_common::LayoutNode {
        agentmux_common::LayoutNode {
            id: id.into(),
            data: Some(agentmux_common::LayoutNodeData {
                block_id: block.into(),
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    /// Create ws-1 + tab-1 (which provisions a real LayoutState row) and
    /// return the tab id.
    fn ws_tab(s: &Store) -> String {
        apply_event_to_mstore(
            &Event::WorkspaceCreated {
                workspace_id: "ws-1".into(),
                name: "A".into(),
                version: 1,
            },
            s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::TabCreated {
                workspace_id: "ws-1".into(),
                tab_id: "tab-1".into(),
                name: "T".into(),
                version: 2,
            },
            s,
        )
        .unwrap();
        "tab-1".to_string()
    }

    fn layout_of(s: &Store, tab_id: &str) -> PersistedLayoutState {
        let tab = s.get::<Tab>(tab_id).unwrap().unwrap();
        s.get::<PersistedLayoutState>(&tab.layoutstate)
            .unwrap()
            .unwrap()
    }

    #[test]
    fn layout_tree_replaced_persists_rootnode() {
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: Some(leaf("n1", "b1")),
                correlation_id: "c1".into(),
                slices: None,
                version: 3,
            },
            &s,
        )
        .unwrap();
        let root = layout_of(&s, &tab).rootnode.expect("rootnode set");
        assert_eq!(root.id, "n1");
        assert_eq!(root.data.unwrap().block_id, "b1");
    }

    #[test]
    fn layout_granular_event_persists_new_tree() {
        // The 7 granular structural events each carry the reducer's resulting
        // tree in `new_tree`; the unified subscriber arm writes it to
        // db_layout (same path as LayoutTreeReplaced). Representative check
        // via LayoutNodeMoved — the |-pattern routes all 7 identically.
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutNodeMoved {
                tab_id: tab.clone(),
                new_tree: Some(leaf("moved-root", "bm")),
                node_id: "x".into(),
                new_parent_id: "y".into(),
                index: 0,
                correlation_id: "c".into(),
                version: 3,
            },
            &s,
        )
        .unwrap();
        let root = layout_of(&s, &tab)
            .rootnode
            .expect("granular event persisted rootnode");
        assert_eq!(root.id, "moved-root");
    }

    #[test]
    fn layout_granular_event_with_none_tree_does_not_clear() {
        // A version-skewed granular event (new_tree defaulted to None) must NOT
        // erase the persisted layout — these ops never produce an empty tree.
        // (codex P2 #1883.)
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: Some(leaf("keep", "bk")),
                correlation_id: "seed".into(),
                slices: None,
                version: 2,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::LayoutNodeMoved {
                tab_id: tab.clone(),
                new_tree: None, // version skew — must be a no-op, not a clear
                node_id: "x".into(),
                new_parent_id: "y".into(),
                index: 0,
                correlation_id: "c".into(),
                version: 3,
            },
            &s,
        )
        .unwrap();
        let root = layout_of(&s, &tab)
            .rootnode
            .expect("tree preserved, not cleared by a None granular event");
        assert_eq!(root.id, "keep");
    }

    // ── SPEC_864 site #6 — LayoutNodeDeleted persistence ────────────────────

    #[test]
    fn layout_node_deleted_persists_new_tree() {
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutNodeDeleted {
                tab_id: tab.clone(),
                node_id: "gone".into(),
                new_tree: Some(leaf("survivor", "bs")),
                tree_cleared: false,
                was_focused: false,
                was_magnified: false,
                correlation_id: "c".into(),
                version: 3,
            },
            &s,
        )
        .unwrap();
        let root = layout_of(&s, &tab)
            .rootnode
            .expect("delete event persisted the resulting tree");
        assert_eq!(root.id, "survivor");
    }

    #[test]
    fn layout_node_deleted_none_without_cleared_is_ignored() {
        // A version-skewed LayoutNodeDeleted (pre-site-#6 JSON: no
        // new_tree, tree_cleared defaults false) must NOT erase the
        // persisted layout — same skew rule as the granular group.
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: Some(leaf("keep", "bk")),
                correlation_id: "seed".into(),
                slices: None,
                version: 2,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::LayoutNodeDeleted {
                tab_id: tab.clone(),
                node_id: "x".into(),
                new_tree: None,
                tree_cleared: false, // skewed sender — not a real clear
                was_focused: false,
                was_magnified: false,
                correlation_id: "c".into(),
                version: 3,
            },
            &s,
        )
        .unwrap();
        let root = layout_of(&s, &tab)
            .rootnode
            .expect("tree preserved — skewed delete event must not clear");
        assert_eq!(root.id, "keep");
    }

    #[test]
    fn layout_node_deleted_tree_cleared_erases_tree() {
        // Root-orphan delete: new_tree=None + tree_cleared=true is a REAL
        // clear (the deleted node was the root) and must wipe rootnode.
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: Some(leaf("doomed-root", "bd")),
                correlation_id: "seed".into(),
                slices: None,
                version: 2,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::LayoutNodeDeleted {
                tab_id: tab.clone(),
                node_id: "doomed-root".into(),
                new_tree: None,
                tree_cleared: true,
                was_focused: true,
                was_magnified: false,
                correlation_id: "c".into(),
                version: 3,
            },
            &s,
        )
        .unwrap();
        assert!(
            layout_of(&s, &tab).rootnode.is_none(),
            "tree_cleared delete must wipe the persisted tree"
        );
    }

    #[test]
    fn layout_cleared_wipes_tree_and_ids() {
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: Some(leaf("n1", "b1")),
                correlation_id: "c".into(),
                slices: None,
                version: 3,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::FocusedNodeChanged {
                tab_id: tab.clone(),
                node_id: "n1".into(),
                version: 4,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::MagnifiedNodeChanged {
                tab_id: tab.clone(),
                node_id: "n1".into(),
                version: 5,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::LayoutCleared {
                tab_id: tab.clone(),
                correlation_id: "c".into(),
                version: 6,
            },
            &s,
        )
        .unwrap();
        let layout = layout_of(&s, &tab);
        assert!(layout.rootnode.is_none());
        assert!(layout.focusednodeid.is_empty());
        assert!(layout.magnifiednodeid.is_empty());
        assert!(layout.leaforder.is_none());
    }

    #[test]
    fn layout_tree_replaced_empty_clears_focus_magnify() {
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: Some(leaf("n1", "b1")),
                correlation_id: "c".into(),
                slices: None,
                version: 3,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::FocusedNodeChanged {
                tab_id: tab.clone(),
                node_id: "n1".into(),
                version: 4,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: None,
                correlation_id: "c".into(),
                slices: None,
                version: 5,
            },
            &s,
        )
        .unwrap();
        let layout = layout_of(&s, &tab);
        assert!(layout.rootnode.is_none());
        assert!(layout.focusednodeid.is_empty());
        assert!(layout.magnifiednodeid.is_empty());
    }

    #[test]
    fn layout_tree_replaced_idempotent_no_version_churn() {
        let s = store();
        let tab = ws_tab(&s);
        let ev = Event::LayoutTreeReplaced {
            tab_id: tab.clone(),
            new_tree: Some(leaf("n1", "b1")),
            correlation_id: "c".into(),
            slices: None,
            version: 3,
        };
        apply_event_to_mstore(&ev, &s).unwrap();
        let v1 = layout_of(&s, &tab).version;
        apply_event_to_mstore(&ev, &s).unwrap();
        let v2 = layout_of(&s, &tab).version;
        assert_eq!(v1, v2, "idempotent re-apply must not bump version");
    }

    // ── SPEC_864 Phase 2 — client slices on LayoutTreeReplaced ─────────

    fn slices(
        leaforder: Option<serde_json::Value>,
        focused: &str,
        magnified: &str,
        pending: Option<serde_json::Value>,
    ) -> agentmux_common::LayoutClientSlices {
        agentmux_common::LayoutClientSlices {
            leaforder,
            focused_node_id: focused.into(),
            magnified_node_id: magnified.into(),
            pending_backend_actions: pending,
        }
    }

    #[test]
    fn layout_tree_replaced_with_slices_writes_full_row() {
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: Some(leaf("n1", "b1")),
                correlation_id: "c".into(),
                slices: Some(slices(
                    Some(serde_json::json!([{ "nodeid": "n1", "blockid": "b1" }])),
                    "n1",
                    "n1",
                    None,
                )),
                version: 3,
            },
            &s,
        )
        .unwrap();
        let layout = layout_of(&s, &tab);
        assert_eq!(layout.rootnode.unwrap().id, "n1");
        assert_eq!(layout.focusednodeid, "n1");
        assert_eq!(layout.magnifiednodeid, "n1");
        let lo = layout.leaforder.expect("leaforder written");
        assert_eq!(lo.len(), 1);
        assert_eq!(lo[0].nodeid, "n1");
        assert_eq!(lo[0].blockid, "b1");
    }

    #[test]
    fn layout_tree_replaced_slices_ack_clears_pending_actions() {
        // Pushing the row without pendingbackendactions is how the frontend
        // acks its processed backend-action queue — the slice-carrying write
        // must CLEAR the column, not leave it (else actions re-apply on the
        // next reload).
        let s = store();
        let tab = ws_tab(&s);
        // Seed a pending action directly (as queue_target_layout_* would).
        {
            let t = s.get::<Tab>(&tab).unwrap().unwrap();
            let mut layout = s
                .get::<PersistedLayoutState>(&t.layoutstate)
                .unwrap()
                .unwrap();
            layout.pendingbackendactions = Some(vec![LayoutActionData {
                actiontype: "insert".into(),
                actionid: "a1".into(),
                blockid: "b-new".into(),
                nodesize: None,
                nodesizefraction: None,
                indexarr: None,
                focused: true,
                magnified: false,
                ephemeral: false,
                targetblockid: String::new(),
                position: String::new(),
            }]);
            s.update(&mut layout).unwrap();
        }
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: Some(leaf("n1", "b1")),
                correlation_id: "c".into(),
                slices: Some(slices(None, "n1", "", None)),
                version: 4,
            },
            &s,
        )
        .unwrap();
        assert!(
            layout_of(&s, &tab).pendingbackendactions.is_none(),
            "slice write with absent pending queue must clear the column (ack path)"
        );
    }

    #[test]
    fn layout_tree_replaced_without_slices_leaves_client_columns_untouched() {
        // Tree-only events (granular arms, dormant callers) must not clobber
        // focus/magnify/leaforder/pending — only a slice-carrying full-row
        // push owns those columns.
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: Some(leaf("n1", "b1")),
                correlation_id: "seed".into(),
                slices: Some(slices(
                    Some(serde_json::json!([{ "nodeid": "n1", "blockid": "b1" }])),
                    "n1",
                    "",
                    Some(serde_json::json!([{
                        "actiontype": "insert",
                        "actionid": "a1",
                        "blockid": "b2",
                        "focused": false,
                        "magnified": false,
                        "ephemeral": false,
                    }])),
                )),
                version: 3,
            },
            &s,
        )
        .unwrap();
        // Now a tree-only replace (slices: None) with a different tree.
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: tab.clone(),
                new_tree: Some(leaf("n2", "b2")),
                correlation_id: "c".into(),
                slices: None,
                version: 4,
            },
            &s,
        )
        .unwrap();
        let layout = layout_of(&s, &tab);
        assert_eq!(layout.rootnode.unwrap().id, "n2", "tree replaced");
        assert_eq!(layout.focusednodeid, "n1", "focus untouched");
        assert!(layout.leaforder.is_some(), "leaforder untouched");
        assert!(
            layout.pendingbackendactions.is_some(),
            "pending queue untouched by tree-only write"
        );
    }

    #[test]
    fn layout_events_silent_when_tab_missing() {
        let s = store();
        apply_event_to_mstore(
            &Event::LayoutCleared {
                tab_id: "ghost".into(),
                correlation_id: "c".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::LayoutTreeReplaced {
                tab_id: "ghost".into(),
                new_tree: Some(leaf("n", "b")),
                correlation_id: "c".into(),
                slices: None,
                version: 1,
            },
            &s,
        )
        .unwrap();
    }

    #[test]
    fn workspace_deleted_cascades_pinned_tabs() {
        let s = store();
        // Build a workspace with a pinned tab via wcore (the bug
        // path: pinned tab created via wcore::create_tab_with_opts).
        let ws = wcore::create_workspace(&s, "Alpha").unwrap();
        let pinned_tab =
            wcore::create_tab_with_opts(&s, &ws.oid, "PinnedTab", true).unwrap();
        // Verify pinned tab is in pinnedtabids (sanity).
        let ws_loaded = s.get::<Workspace>(&ws.oid).unwrap().unwrap();
        assert!(ws_loaded.pinnedtabids.contains(&pinned_tab.oid));
        // Delete via the subscriber's WorkspaceDeleted handler.
        apply_event_to_mstore(
            &Event::WorkspaceDeleted {
                workspace_id: ws.oid.clone(),
                block_ids: Vec::new(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        // codex P1 #614: pinned tab must be cascade-deleted, not
        // orphaned.
        assert!(s.get::<Tab>(&pinned_tab.oid).unwrap().is_none());
        assert!(s.get::<Workspace>(&ws.oid).unwrap().is_none());
    }

    #[test]
    fn block_created_links_into_tab() {
        let s = store();
        apply_event_to_mstore(
            &Event::WorkspaceCreated {
                workspace_id: "ws-1".into(),
                name: "Alpha".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::TabCreated {
                workspace_id: "ws-1".into(),
                tab_id: "tab-1".into(),
                name: "Tab".into(),
                version: 2,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::BlockCreated { tab_id: "tab-1".into(), block_id: "block-1".into(), meta: serde_json::Value::Null, version: 3 },
            &s,
        )
        .unwrap();
        let tab = s.get::<Tab>("tab-1").unwrap().unwrap();
        assert_eq!(tab.blockids, vec!["block-1".to_string()]);
        let block = s.get::<Block>("block-1").unwrap().unwrap();
        assert_eq!(block.parentoref, "tab:tab-1");
    }

    #[test]
    fn block_deleted_unlinks_from_tab() {
        let s = store();
        apply_event_to_mstore(
            &Event::WorkspaceCreated {
                workspace_id: "ws-1".into(),
                name: "Alpha".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::TabCreated {
                workspace_id: "ws-1".into(),
                tab_id: "tab-1".into(),
                name: "Tab".into(),
                version: 2,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::BlockCreated { tab_id: "tab-1".into(), block_id: "block-1".into(), meta: serde_json::Value::Null, version: 3 },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::BlockDeleted {
                tab_id: "tab-1".into(),
                block_id: "block-1".into(),
                version: 4,
            },
            &s,
        )
        .unwrap();
        assert!(s.get::<Block>("block-1").unwrap().is_none());
        let tab = s.get::<Tab>("tab-1").unwrap().unwrap();
        assert!(tab.blockids.is_empty());
    }

    /// codex P1 #620: a workspace with legacy `pinnedtabids` rows must
    /// have those drained the first time `TabsReorderedBulk` writes
    /// the workspace; otherwise the UI's
    /// `[...pinnedtabids, ...tabids]` combine duplicates the pinned
    /// tab IDs once they have been merged into the reducer's
    /// `tab_ids` at bootstrap.
    #[test]
    fn tabs_reordered_bulk_drains_legacy_pinned_tabids() {
        let s = store();
        let ws = wcore::create_workspace(&s, "Alpha").unwrap();
        let pinned_tab =
            wcore::create_tab_with_opts(&s, &ws.oid, "Pinned", true).unwrap();
        let regular_tab =
            wcore::create_tab_with_opts(&s, &ws.oid, "Regular", false).unwrap();
        // Sanity: pinned tab is in `pinnedtabids` on disk.
        let ws_before = s.get::<Workspace>(&ws.oid).unwrap().unwrap();
        assert!(ws_before.pinnedtabids.contains(&pinned_tab.oid));

        // Reducer-driven bulk reorder treating the pinned tab as a
        // regular tab (mirrors what bootstrap-merge produces).
        apply_event_to_mstore(
            &Event::TabsReorderedBulk {
                workspace_id: ws.oid.clone(),
                tab_ids: vec![pinned_tab.oid.clone(), regular_tab.oid.clone()],
                version: ws_before.version as u64 + 1,
            },
            &s,
        )
        .unwrap();
        let ws_after = s.get::<Workspace>(&ws.oid).unwrap().unwrap();
        assert_eq!(
            ws_after.tabids,
            vec![pinned_tab.oid.clone(), regular_tab.oid.clone()]
        );
        assert!(
            ws_after.pinnedtabids.is_empty(),
            "pinnedtabids must be drained, was {:?}",
            ws_after.pinnedtabids
        );
    }

    /// codex P2 #620: `merge_meta_patch` must honour the existing
    /// `section:*` clear-prefix semantics so `UpdateObjectMeta`
    /// behaviour stays the same after the reducer migration.
    #[test]
    fn meta_updated_clears_section_prefix() {
        let s = store();
        apply_event_to_mstore(
            &Event::TabCreated {
                workspace_id: "ws-1".into(),
                tab_id: "tab-1".into(),
                name: "Tab".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::WorkspaceCreated {
                workspace_id: "ws-1".into(),
                name: "Alpha".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
        // Seed the tab with grouped meta entries.
        let mut tab = s.get::<Tab>("tab-1").unwrap().unwrap();
        tab.meta
            .insert("term:fontsize".into(), serde_json::json!(14));
        tab.meta
            .insert("term:theme".into(), serde_json::json!("solarized"));
        tab.meta.insert("name".into(), serde_json::json!("keep"));
        s.update(&mut tab).unwrap();
        // Patch with `term:*` clear plus a single replacement key.
        apply_event_to_mstore(
            &Event::TabMetaUpdated {
                tab_id: "tab-1".into(),
                meta_patch: serde_json::json!({
                    "term:*": true,
                    "term:fontsize": 18,
                }),
                version: 2,
            },
            &s,
        )
        .unwrap();
        let after = s.get::<Tab>("tab-1").unwrap().unwrap();
        assert!(!after.meta.contains_key("term:theme"),
            "term:theme should be cleared by `term:*` patch");
        assert_eq!(
            after.meta.get("term:fontsize"),
            Some(&serde_json::json!(18)),
            "term:fontsize replacement must take effect after the section clear"
        );
        assert_eq!(after.meta.get("name"), Some(&serde_json::json!("keep")));
    }

    // ---- Phase E.5.5 — TabMoved / BlockMoved subscriber tests ----

    #[test]
    fn tab_moved_cross_workspace_rewrites_both_tabids() {
        let s = store();
        // Two workspaces, each pre-existing in SQLite.
        for (id, name) in &[("src-ws", "Src"), ("dst-ws", "Dst")] {
            apply_event_to_mstore(
                &Event::WorkspaceCreated {
                    workspace_id: id.to_string(),
                    name: name.to_string(),
                    version: 1,
                },
                &s,
            )
            .unwrap();
        }
        // Tab in src.
        apply_event_to_mstore(
            &Event::TabCreated {
                workspace_id: "src-ws".into(),
                tab_id: "tab-1".into(),
                name: "T".into(),
                version: 2,
            },
            &s,
        )
        .unwrap();
        let src = s.get::<Workspace>("src-ws").unwrap().unwrap();
        assert_eq!(src.tabids, vec!["tab-1".to_string()]);

        // Move it. Set dst active to the moved tab (per reducer
        // semantics + codex P2 #621).
        apply_event_to_mstore(
            &Event::TabMoved {
                tab_id: "tab-1".into(),
                src_workspace_id: "src-ws".into(),
                dst_workspace_id: "dst-ws".into(),
                dst_index: 0,
                new_src_active_tab_id: None,
                new_dst_active_tab_id: Some("tab-1".into()),
                version: 3,
            },
            &s,
        )
        .unwrap();
        let src = s.get::<Workspace>("src-ws").unwrap().unwrap();
        let dst = s.get::<Workspace>("dst-ws").unwrap().unwrap();
        assert!(src.tabids.is_empty());
        assert_eq!(src.activetabid, "");
        assert_eq!(dst.tabids, vec!["tab-1".to_string()]);
        assert_eq!(dst.activetabid, "tab-1", "dst active should be the moved tab");
    }

    #[test]
    fn tab_moved_idempotent_on_re_delivery() {
        let s = store();
        for (id, name) in &[("src-ws", "Src"), ("dst-ws", "Dst")] {
            apply_event_to_mstore(
                &Event::WorkspaceCreated {
                    workspace_id: id.to_string(),
                    name: name.to_string(),
                    version: 1,
                },
                &s,
            )
            .unwrap();
        }
        apply_event_to_mstore(
            &Event::TabCreated {
                workspace_id: "src-ws".into(),
                tab_id: "tab-1".into(),
                name: "T".into(),
                version: 2,
            },
            &s,
        )
        .unwrap();
        let ev = Event::TabMoved {
            tab_id: "tab-1".into(),
            src_workspace_id: "src-ws".into(),
            dst_workspace_id: "dst-ws".into(),
            dst_index: 0,
            new_src_active_tab_id: None,
            new_dst_active_tab_id: Some("tab-1".into()),
            version: 3,
        };
        apply_event_to_mstore(&ev, &s).unwrap();
        // Re-deliver the same event.
        apply_event_to_mstore(&ev, &s).unwrap();
        let dst = s.get::<Workspace>("dst-ws").unwrap().unwrap();
        // Still exactly one entry — no duplicate insert.
        assert_eq!(dst.tabids, vec!["tab-1".to_string()]);
    }

    #[test]
    fn block_moved_cross_tab_updates_block_lists_and_parent() {
        let s = store();
        let ws = wcore::create_workspace(&s, "W").unwrap();
        let src_tab = wcore::create_tab_with_opts(&s, &ws.oid, "src", false).unwrap();
        let dst_tab = wcore::create_tab_with_opts(&s, &ws.oid, "dst", false).unwrap();
        // Create a block in src via the subscriber path.
        apply_event_to_mstore(
            &Event::BlockCreated {
                tab_id: src_tab.oid.clone(),
                block_id: "blk-1".into(),
                meta: serde_json::Value::Null,
                version: 1,
            },
            &s,
        )
        .unwrap();
        let src_before = s.get::<Tab>(&src_tab.oid).unwrap().unwrap();
        assert_eq!(src_before.blockids, vec!["blk-1".to_string()]);

        apply_event_to_mstore(
            &Event::BlockMoved {
                block_id: "blk-1".into(),
                src_tab_id: src_tab.oid.clone(),
                dst_tab_id: dst_tab.oid.clone(),
                dst_index: 0,
                version: 2,
            },
            &s,
        )
        .unwrap();
        let src_after = s.get::<Tab>(&src_tab.oid).unwrap().unwrap();
        let dst_after = s.get::<Tab>(&dst_tab.oid).unwrap().unwrap();
        let block = s.get::<Block>("blk-1").unwrap().unwrap();
        assert!(src_after.blockids.is_empty());
        assert_eq!(dst_after.blockids, vec!["blk-1".to_string()]);
        assert_eq!(block.parentoref, format!("tab:{}", dst_tab.oid));
    }

    #[test]
    fn block_moved_intra_tab_repositions() {
        let s = store();
        let ws = wcore::create_workspace(&s, "W").unwrap();
        let tab = wcore::create_tab_with_opts(&s, &ws.oid, "t", false).unwrap();
        for id in &["b1", "b2", "b3"] {
            apply_event_to_mstore(
                &Event::BlockCreated {
                    tab_id: tab.oid.clone(),
                    block_id: id.to_string(),
                    meta: serde_json::Value::Null,
                    version: 1,
                },
                &s,
            )
            .unwrap();
        }
        // Move b1 to position 2 (post-removal end).
        apply_event_to_mstore(
            &Event::BlockMoved {
                block_id: "b1".into(),
                src_tab_id: tab.oid.clone(),
                dst_tab_id: tab.oid.clone(),
                dst_index: 2,
                version: 2,
            },
            &s,
        )
        .unwrap();
        let tab_after = s.get::<Tab>(&tab.oid).unwrap().unwrap();
        assert_eq!(tab_after.blockids, vec!["b2".to_string(), "b3".to_string(), "b1".to_string()]);
    }

    // ── SPEC_864 Phase 4 — pendingbackendactions APPEND semantics ──────────

    fn action(id: &str, block: &str) -> LayoutActionData {
        LayoutActionData {
            actiontype: "insert".into(),
            actionid: id.into(),
            blockid: block.into(),
            nodesize: None,
            nodesizefraction: None,
            indexarr: None,
            focused: true,
            magnified: false,
            ephemeral: false,
            targetblockid: String::new(),
            position: String::new(),
        }
    }

    #[test]
    fn layout_backend_actions_queued_appends_not_replaces() {
        let s = store();
        let tab = ws_tab(&s);
        apply_event_to_mstore(
            &Event::LayoutBackendActionsQueued {
                tab_id: tab.clone(),
                actions: serde_json::to_value(vec![action("a1", "b1")]).unwrap(),
                correlation_id: "c1".into(),
                version: 3,
            },
            &s,
        )
        .unwrap();
        apply_event_to_mstore(
            &Event::LayoutBackendActionsQueued {
                tab_id: tab.clone(),
                actions: serde_json::to_value(vec![action("a2", "b2")]).unwrap(),
                correlation_id: "c2".into(),
                version: 4,
            },
            &s,
        )
        .unwrap();
        let pending = layout_of(&s, &tab).pendingbackendactions.unwrap();
        assert_eq!(pending.len(), 2, "second event must APPEND, not replace");
        assert_eq!(pending[0].actionid, "a1");
        assert_eq!(pending[1].actionid, "a2");
    }

    #[test]
    fn layout_backend_actions_queued_dedupes_on_double_apply() {
        // Same event applied twice (sync dispatch + broadcast redelivery)
        // must not duplicate the queued action.
        let s = store();
        let tab = ws_tab(&s);
        let ev = Event::LayoutBackendActionsQueued {
            tab_id: tab.clone(),
            actions: serde_json::to_value(vec![action("a1", "b1")]).unwrap(),
            correlation_id: "c1".into(),
            version: 3,
        };
        apply_event_to_mstore(&ev, &s).unwrap();
        apply_event_to_mstore(&ev, &s).unwrap();
        let pending = layout_of(&s, &tab).pendingbackendactions.unwrap();
        assert_eq!(pending.len(), 1, "double-apply must not duplicate by actionid");
    }

    #[test]
    fn layout_backend_actions_queued_silent_when_tab_missing() {
        let s = store();
        apply_event_to_mstore(
            &Event::LayoutBackendActionsQueued {
                tab_id: "ghost".into(),
                actions: serde_json::to_value(vec![action("a1", "b1")]).unwrap(),
                correlation_id: "c1".into(),
                version: 1,
            },
            &s,
        )
        .unwrap();
    }
}