huskies/server/src/crdt_state/write/item.rs

//! CRDT write operations for individual pipeline items.
//!
//! Provides typed setters for agent, QA mode, retry count, and the primary
//! `write_item` function that inserts or updates pipeline items in the CRDT.

use bft_json_crdt::json_crdt::{CrdtNode, JsonValue};
use bft_json_crdt::lww_crdt::LwwRegisterCrdt;
use bft_json_crdt::op::ROOT_ID;
use serde_json::json;

use super::super::state::{apply_and_persist, emit_event, get_crdt, rebuild_index};
use super::super::types::CrdtEvent;
use crate::io::story_metadata::QaMode;

/// Set the typed `depends_on` CRDT register for a pipeline item.
///
/// Encodes `deps` as a compact JSON array string (e.g. `"[837]"`) and writes it
/// to the item's `depends_on` register.  An empty slice clears the register to an
/// empty string, which means "no dependencies".
///
/// Returns `true` if the item was found and the op was applied, `false` otherwise.
pub fn set_depends_on(story_id: &str, deps: &[u32]) -> bool {
    let Some(state_mutex) = get_crdt() else {
        return false;
    };
    let Ok(mut state) = state_mutex.lock() else {
        return false;
    };
    let Some(&idx) = state.index.get(story_id) else {
        return false;
    };
    let value = if deps.is_empty() {
        String::new()
    } else {
        serde_json::to_string(deps).unwrap_or_default()
    };
    apply_and_persist(&mut state, |s| s.crdt.doc.items[idx].depends_on.set(value));
    true
}

/// Set the `mergemaster_attempted` CRDT flag for a pipeline item.
///
/// Passing `true` records that a mergemaster session has been spawned for this
/// item, preventing repeated auto-spawns across restarts.
/// Passing `false` explicitly writes `false` (does not remove the register) so
/// the cleared state is distinguishable from an unset register and survives
/// CRDT replay correctly.
///
/// Returns `true` if the item was found and the op was applied, `false` otherwise.
pub fn set_mergemaster_attempted(story_id: &str, value: bool) -> bool {
    let Some(state_mutex) = get_crdt() else {
        return false;
    };
    let Ok(mut state) = state_mutex.lock() else {
        return false;
    };
    let Some(&idx) = state.index.get(story_id) else {
        return false;
    };
    apply_and_persist(&mut state, |s| {
        s.crdt.doc.items[idx].mergemaster_attempted.set(value)
    });
    true
}

/// Set the `agent` field for a pipeline item by its story ID.
///
/// `Some(name)` writes the agent name into the CRDT register.
/// `None` clears the register by writing an empty string — use this
/// to unpin an agent without touching the surrounding item.
///
/// This is the typed setter counterpart to [`write_item`]'s `agent` parameter.
/// Callers that only need to update the agent (e.g. the `update_story` MCP tool
/// and the Matrix `!assign` command) should prefer this function over
/// passing `agent` through the full [`write_item`] call, which requires all
/// other fields to be known.
///
/// Returns `true` if the item was found and the write was performed.
pub fn set_agent(story_id: &str, agent: Option<&str>) -> bool {
    let Some(state_mutex) = get_crdt() else {
        return false;
    };
    let Ok(mut state) = state_mutex.lock() else {
        return false;
    };
    let Some(&idx) = state.index.get(story_id) else {
        return false;
    };
    let value = agent.unwrap_or("").to_string();
    apply_and_persist(&mut state, |s| {
        s.crdt.doc.items[idx].agent.set(value.clone())
    });
    true
}

/// Set the typed `qa_mode` CRDT register for a pipeline item.
///
/// Passing `Some(mode)` writes the mode string (e.g. `"server"`, `"agent"`, `"human"`)
/// to the item's `qa_mode` register and persists a signed op.
/// Passing `None` clears the register to an empty string, which means
/// "use the project default" (same as if the field was never set).
///
/// Returns `true` if the item was found and the op was applied, `false` otherwise.
pub fn set_qa_mode(story_id: &str, mode: Option<QaMode>) -> bool {
    let Some(state_mutex) = get_crdt() else {
        return false;
    };
    let Ok(mut state) = state_mutex.lock() else {
        return false;
    };
    let Some(&idx) = state.index.get(story_id) else {
        return false;
    };
    let value = mode.map(|m| m.as_str().to_string()).unwrap_or_default();
    apply_and_persist(&mut state, |s| s.crdt.doc.items[idx].qa_mode.set(value));
    true
}

/// Write a pipeline item state through CRDT operations.
///
/// If the item exists, updates its registers.  If not, inserts a new item
/// into the list.  All ops are signed and persisted to SQLite.
///
/// When the stage changes (or a new item is created), a [`CrdtEvent`] is
/// broadcast so subscribers can react to the transition.
#[allow(clippy::too_many_arguments)]
pub fn write_item(
    story_id: &str,
    stage: &str,
    name: Option<&str>,
    agent: Option<&str>,
    retry_count: Option<i64>,
    blocked: Option<bool>,
    depends_on: Option<&str>,
    claimed_by: Option<&str>,
    claimed_at: Option<f64>,
    merged_at: Option<f64>,
) {
    let Some(state_mutex) = get_crdt() else {
        return;
    };
    let Ok(mut state) = state_mutex.lock() else {
        return;
    };

    // Reject any write (insert or update) for a tombstoned story_id.
    // This prevents a concurrent or late-arriving write from resurrecting
    // a story that was permanently deleted via evict_item.
    if state.tombstones.contains(story_id) {
        return;
    }

    if let Some(&idx) = state.index.get(story_id) {
        // Capture the old stage before updating so we can detect transitions.
        let old_stage = match state.crdt.doc.items[idx].stage.view() {
            JsonValue::String(s) => Some(s),
            _ => None,
        };

        // Update existing item registers.
        apply_and_persist(&mut state, |s| {
            s.crdt.doc.items[idx].stage.set(stage.to_string())
        });

        if let Some(n) = name {
            apply_and_persist(&mut state, |s| {
                s.crdt.doc.items[idx].name.set(n.to_string())
            });
        }
        if let Some(a) = agent {
            apply_and_persist(&mut state, |s| {
                s.crdt.doc.items[idx].agent.set(a.to_string())
            });
        }
        if let Some(rc) = retry_count {
            apply_and_persist(&mut state, |s| {
                s.crdt.doc.items[idx].retry_count.set(rc as f64)
            });
        }
        if let Some(b) = blocked {
            apply_and_persist(&mut state, |s| s.crdt.doc.items[idx].blocked.set(b));
        }
        if let Some(d) = depends_on {
            apply_and_persist(&mut state, |s| {
                s.crdt.doc.items[idx].depends_on.set(d.to_string())
            });
        }
        if let Some(cb) = claimed_by {
            apply_and_persist(&mut state, |s| {
                s.crdt.doc.items[idx].claimed_by.set(cb.to_string())
            });
        }
        if let Some(ca) = claimed_at {
            apply_and_persist(&mut state, |s| s.crdt.doc.items[idx].claimed_at.set(ca));
        }
        if let Some(ma) = merged_at {
            apply_and_persist(&mut state, |s| s.crdt.doc.items[idx].merged_at.set(ma));
        }

        // Broadcast a CrdtEvent if the stage actually changed.
        let stage_changed = old_stage.as_deref() != Some(stage);
        if stage_changed {
            // Read the current name from the CRDT document for the event.
            let current_name = match state.crdt.doc.items[idx].name.view() {
                JsonValue::String(s) if !s.is_empty() => Some(s),
                _ => None,
            };
            emit_event(CrdtEvent {
                story_id: story_id.to_string(),
                from_stage: old_stage,
                to_stage: stage.to_string(),
                name: current_name,
            });
        }
    } else {
        // Insert new item.
        let item_json: JsonValue = json!({
            "story_id": story_id,
            "stage": stage,
            "name": name.unwrap_or(""),
            "agent": agent.unwrap_or(""),
            "retry_count": retry_count.unwrap_or(0) as f64,
            "blocked": blocked.unwrap_or(false),
            "depends_on": depends_on.unwrap_or(""),
            "claimed_by": claimed_by.unwrap_or(""),
            "claimed_at": claimed_at.unwrap_or(0.0),
            "merged_at": merged_at.unwrap_or(0.0),
            "qa_mode": "",
            "mergemaster_attempted": false,
        })
        .into();

        apply_and_persist(&mut state, |s| s.crdt.doc.items.insert(ROOT_ID, item_json));

        // Rebuild index after insertion (indices may shift).
        state.index = rebuild_index(&state.crdt);

        // Advance the inner registers of the newly-created item to the Lamport
        // floor so their first local ops don't re-emit low sequence numbers.
        let floor = state.lamport_floor;
        if floor > 0
            && let Some(&idx) = state.index.get(story_id)
        {
            let item = &mut state.crdt.doc.items[idx];
            item.story_id.advance_seq(floor);
            item.stage.advance_seq(floor);
            item.name.advance_seq(floor);
            item.agent.advance_seq(floor);
            item.retry_count.advance_seq(floor);
            item.blocked.advance_seq(floor);
            item.depends_on.advance_seq(floor);
            item.claimed_by.advance_seq(floor);
            item.claimed_at.advance_seq(floor);
            item.merged_at.advance_seq(floor);
            item.qa_mode.advance_seq(floor);
            item.mergemaster_attempted.advance_seq(floor);
        }

        // Broadcast a CrdtEvent for the new item.
        emit_event(CrdtEvent {
            story_id: story_id.to_string(),
            from_stage: None,
            to_stage: stage.to_string(),
            name: name.map(String::from),
        });
    }
}

/// Set `retry_count` to an explicit value for a pipeline item.
///
/// Pure metadata operation — the item's stage is not changed.
/// Call `set_retry_count(story_id, 0)` to reset the counter after a
/// stage transition or an explicit unblock.
pub fn set_retry_count(story_id: &str, count: i64) {
    let Some(state_mutex) = get_crdt() else {
        return;
    };
    let Ok(mut state) = state_mutex.lock() else {
        return;
    };
    if let Some(&idx) = state.index.get(story_id) {
        apply_and_persist(&mut state, |s| {
            s.crdt.doc.items[idx].retry_count.set(count as f64)
        });
    }
}

/// Set the `blocked` register on a story to the given value.
///
/// Pure metadata operation — the item's stage is not changed.
/// Use this alongside a state-machine transition out of `Blocked` /
/// `MergeFailure` to keep the legacy `blocked` register in sync with the
/// typed stage post-865 (where YAML side-effects no longer clear the
/// register on their own).
pub fn set_blocked(story_id: &str, blocked: bool) {
    let Some(state_mutex) = get_crdt() else {
        return;
    };
    let Ok(mut state) = state_mutex.lock() else {
        return;
    };
    if let Some(&idx) = state.index.get(story_id) {
        apply_and_persist(&mut state, |s| s.crdt.doc.items[idx].blocked.set(blocked));
    }
}

/// Increment `retry_count` by 1 and return the new value.
///
/// Pure metadata operation — the item's stage is not changed.
/// Returns 0 if the item is not found in the CRDT (no-op in that case).
/// Use the returned value to decide whether the story should be blocked.
pub fn bump_retry_count(story_id: &str) -> i64 {
    let Some(state_mutex) = get_crdt() else {
        return 0;
    };
    let Ok(mut state) = state_mutex.lock() else {
        return 0;
    };
    let Some(&idx) = state.index.get(story_id) else {
        return 0;
    };
    let current = match state.crdt.doc.items[idx].retry_count.view() {
        JsonValue::Number(n) => n as i64,
        _ => 0,
    };
    let new_count = current + 1;
    apply_and_persist(&mut state, |s| {
        s.crdt.doc.items[idx].retry_count.set(new_count as f64)
    });
    new_count
}