huskies: merge 857

server/src/pipeline_state/apply.rs

@@ -0,0 +1,99 @@
+//! Single entry point for pipeline state transitions.
+//!
+//! [`apply_transition`] is the **only** function that should mutate a story's
+//! pipeline stage.  It reads the current typed stage from the CRDT, validates
+//! the transition via the pure [`super::transition()`] function, writes the new
+//! stage back to the CRDT, and returns a [`TransitionFired`] event for
+//! downstream subscribers.
+
+use super::{
+    PipelineEvent, StoryId, TransitionFired, event_label, read_typed, stage_label, transition,
+};
+use chrono::Utc;
+
+/// Error type for [`apply_transition`].
+#[derive(Debug)]
+pub enum ApplyError {
+    /// The story was not found in the CRDT.
+    NotFound(String),
+    /// The CRDT projection failed.
+    Projection(super::ProjectionError),
+    /// The transition was invalid for the current stage.
+    InvalidTransition(super::TransitionError),
+}
+
+impl std::fmt::Display for ApplyError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::NotFound(id) => write!(f, "story '{id}' not found in CRDT"),
+            Self::Projection(e) => write!(f, "projection error: {e}"),
+            Self::InvalidTransition(e) => write!(f, "{e}"),
+        }
+    }
+}
+
+impl std::error::Error for ApplyError {}
+
+impl From<super::ProjectionError> for ApplyError {
+    fn from(e: super::ProjectionError) -> Self {
+        Self::Projection(e)
+    }
+}
+
+impl From<super::TransitionError> for ApplyError {
+    fn from(e: super::TransitionError) -> Self {
+        Self::InvalidTransition(e)
+    }
+}
+
+/// Apply a pipeline event to a story, validating the transition and writing
+/// the new stage to the CRDT.
+///
+/// This is the single canonical entry point for all pipeline stage mutations.
+/// Returns a [`TransitionFired`] describing what happened, suitable for
+/// broadcasting to event-bus subscribers and CRDT log consumers.
+///
+/// An optional `content_transform` allows callers to modify the stored content
+/// (e.g. clearing front-matter fields) atomically with the stage change.
+pub fn apply_transition(
+    story_id: &str,
+    event: PipelineEvent,
+    content_transform: Option<&dyn Fn(&str) -> String>,
+) -> Result<TransitionFired, ApplyError> {
+    let item = read_typed(story_id)?.ok_or_else(|| ApplyError::NotFound(story_id.to_string()))?;
+
+    let before = item.stage.clone();
+    let after = transition(before.clone(), event.clone())?;
+    let new_dir = after.dir_name();
+
+    // Write the new stage to the CRDT (with optional content transform).
+    crate::db::move_item_stage(story_id, new_dir, content_transform);
+
+    let fired = TransitionFired {
+        story_id: StoryId(story_id.to_string()),
+        before,
+        after,
+        event,
+        at: Utc::now(),
+    };
+
+    crate::slog!(
+        "[pipeline/transition] #{}: {} + {} → {}",
+        story_id,
+        stage_label(&fired.before),
+        event_label(&fired.event),
+        stage_label(&fired.after),
+    );
+
+    Ok(fired)
+}
+
+/// Convenience: apply a transition, returning an `Err(String)` on failure
+/// (matches the existing lifecycle function signatures).
+pub fn apply_transition_str(
+    story_id: &str,
+    event: PipelineEvent,
+    content_transform: Option<&dyn Fn(&str) -> String>,
+) -> Result<TransitionFired, String> {
+    apply_transition(story_id, event, content_transform).map_err(|e| e.to_string())
+}

server/src/pipeline_state/mod.rs

@@ -23,6 +23,7 @@
 // the dead_code lint is suppressed for the module.
 #![allow(dead_code)]
 
+mod apply;
 mod events;
 mod projection;
 mod subscribers;
@@ -52,6 +53,9 @@ pub use events::{EventBus, TransitionFired, TransitionSubscriber};
 pub use projection::{ProjectionError, project_stage};
 pub use projection::{read_all_typed, read_typed};
 
+#[allow(unused_imports)]
+pub use apply::{ApplyError, apply_transition, apply_transition_str};
+
 #[allow(unused_imports)]
 pub use subscribers::{
     AutoAssignSubscriber, FileRendererSubscriber, MatrixBotSubscriber, PipelineItemsSubscriber,

server/src/pipeline_state/projection.rs

@@ -70,6 +70,7 @@ impl TryFrom<&PipelineItemView> for PipelineItem {
 /// loose CRDT data becomes typed.
 pub fn project_stage(view: &PipelineItemView) -> Result<Stage, ProjectionError> {
     match view.stage.as_str() {
+        "0_upcoming" => Ok(Stage::Upcoming),
         "1_backlog" => Ok(Stage::Backlog),
         "2_current" => Ok(Stage::Coding),
         "3_qa" => Ok(Stage::Qa),
@@ -192,6 +193,26 @@ mod tests {
         StoryId(s.to_string())
     }
 
+    #[test]
+    fn project_upcoming_item() {
+        let view = PipelineItemView {
+            story_id: "42_story_test".to_string(),
+            stage: "0_upcoming".to_string(),
+            name: Some("Test Story".to_string()),
+            agent: None,
+            retry_count: None,
+            blocked: None,
+            depends_on: None,
+            claimed_by: None,
+            claimed_at: None,
+            merged_at: None,
+            qa_mode: None,
+            mergemaster_attempted: None,
+        };
+        let item = PipelineItem::try_from(&view).unwrap();
+        assert!(matches!(item.stage, Stage::Upcoming));
+    }
+
     #[test]
     fn project_backlog_item() {
         let view = PipelineItemView {
@@ -345,6 +366,7 @@ mod tests {
     #[test]
     fn reverse_projection_stage_dirs() {
         let cases: Vec<(Stage, &str, bool)> = vec![
+            (Stage::Upcoming, "0_upcoming", false),
             (Stage::Backlog, "1_backlog", false),
             (Stage::Coding, "2_current", false),
             (Stage::Qa, "3_qa", false),

server/src/pipeline_state/tests.rs

@@ -406,4 +406,136 @@ fn transition_error_display() {
     assert_eq!(err.to_string(), "invalid transition: Backlog + Accepted");
 }
 
+// ── Upcoming / Triage ──────────────────────────────────────────────
+
+#[test]
+fn triage_upcoming_to_backlog() {
+    let s = Stage::Upcoming;
+    let s = transition(s, PipelineEvent::Triage).unwrap();
+    assert!(matches!(s, Stage::Backlog));
+}
+
+#[test]
+fn cannot_triage_from_backlog() {
+    let result = transition(Stage::Backlog, PipelineEvent::Triage);
+    assert!(matches!(
+        result,
+        Err(TransitionError::InvalidTransition { .. })
+    ));
+}
+
+#[test]
+fn abandon_from_upcoming() {
+    let result = transition(Stage::Upcoming, PipelineEvent::Abandon).unwrap();
+    assert!(matches!(
+        result,
+        Stage::Archived {
+            reason: ArchiveReason::Abandoned,
+            ..
+        }
+    ));
+}
+
+#[test]
+fn supersede_from_upcoming() {
+    let result = transition(
+        Stage::Upcoming,
+        PipelineEvent::Supersede {
+            by: sid("999_story_new"),
+        },
+    )
+    .unwrap();
+    assert!(matches!(
+        result,
+        Stage::Archived {
+            reason: ArchiveReason::Superseded { .. },
+            ..
+        }
+    ));
+}
+
+#[test]
+fn cannot_deps_met_from_upcoming() {
+    let result = transition(Stage::Upcoming, PipelineEvent::DepsMet);
+    assert!(matches!(
+        result,
+        Err(TransitionError::InvalidTransition { .. })
+    ));
+}
+
+// ── Reject ─────────────────────────────────────────────────────────
+
+#[test]
+fn reject_from_active_stages() {
+    for s in [Stage::Backlog, Stage::Coding, Stage::Qa] {
+        let result = transition(
+            s.clone(),
+            PipelineEvent::Reject {
+                reason: "not needed".into(),
+            },
+        );
+        assert!(matches!(
+            result,
+            Ok(Stage::Archived {
+                reason: ArchiveReason::Rejected { .. },
+                ..
+            })
+        ));
+    }
+
+    let m = Stage::Merge {
+        feature_branch: fb("f"),
+        commits_ahead: nz(1),
+    };
+    let result = transition(
+        m,
+        PipelineEvent::Reject {
+            reason: "not needed".into(),
+        },
+    );
+    assert!(matches!(
+        result,
+        Ok(Stage::Archived {
+            reason: ArchiveReason::Rejected { .. },
+            ..
+        })
+    ));
+}
+
+#[test]
+fn cannot_reject_from_done() {
+    let s = Stage::Done {
+        merged_at: chrono::Utc::now(),
+        merge_commit: sha("abc"),
+    };
+    let result = transition(
+        s,
+        PipelineEvent::Reject {
+            reason: "too late".into(),
+        },
+    );
+    assert!(matches!(
+        result,
+        Err(TransitionError::InvalidTransition { .. })
+    ));
+}
+
+#[test]
+fn cannot_reject_from_archived() {
+    let s = Stage::Archived {
+        archived_at: chrono::Utc::now(),
+        reason: ArchiveReason::Completed,
+    };
+    let result = transition(
+        s,
+        PipelineEvent::Reject {
+            reason: "already done".into(),
+        },
+    );
+    assert!(matches!(
+        result,
+        Err(TransitionError::InvalidTransition { .. })
+    ));
+}
+
 // ── ProjectionError Display ─────────────────────────────────────────

server/src/pipeline_state/transition.rs

@@ -47,6 +47,15 @@ pub enum PipelineEvent {
     Supersede { by: StoryId },
     /// Story put on review hold.
     ReviewHold { reason: String },
+    /// Story rejected by QA or reviewer.
+    Reject { reason: String },
+    /// Story triaged from upcoming to backlog.
+    Triage,
+    /// Direct completion — item closed without going through the full pipeline
+    /// (e.g. spike auto-merge, bug closure, manual acceptance).
+    Close,
+    /// Manual demotion back to backlog from an active stage.
+    Demote,
 }
 
 // ── Per-node execution events ───────────────────────────────────────────────
@@ -81,6 +90,10 @@ pub fn event_label(e: &PipelineEvent) -> &'static str {
         PipelineEvent::Abandon => "Abandon",
         PipelineEvent::Supersede { .. } => "Supersede",
         PipelineEvent::ReviewHold { .. } => "ReviewHold",
+        PipelineEvent::Reject { .. } => "Reject",
+        PipelineEvent::Triage => "Triage",
+        PipelineEvent::Close => "Close",
+        PipelineEvent::Demote => "Demote",
     }
 }
 
@@ -103,6 +116,9 @@ pub fn transition(state: Stage, event: PipelineEvent) -> Result<Stage, Transitio
     let now = Utc::now();
 
     match (state, event) {
+        // ── Triage: upcoming → backlog ──────────────────────────────────
+        (Upcoming, Triage) => Ok(Backlog),
+
         // ── Forward path ────────────────────────────────────────────────
         (Backlog, DepsMet) => Ok(Coding),
         (Coding, GatesStarted) => Ok(Qa),
@@ -161,7 +177,8 @@ pub fn transition(state: Stage, event: PipelineEvent) -> Result<Stage, Transitio
         }),
 
         // ── Abandon / supersede from any active or done stage ───────────
-        (Backlog, Abandon)
+        (Upcoming, Abandon)
+        | (Backlog, Abandon)
         | (Coding, Abandon)
         | (Qa, Abandon)
         | (Merge { .. }, Abandon)
@@ -170,7 +187,8 @@ pub fn transition(state: Stage, event: PipelineEvent) -> Result<Stage, Transitio
             reason: ArchiveReason::Abandoned,
         }),
 
-        (Backlog, Supersede { by })
+        (Upcoming, Supersede { by })
+        | (Backlog, Supersede { by })
         | (Coding, Supersede { by })
         | (Qa, Supersede { by })
         | (Merge { .. }, Supersede { by })
@@ -179,13 +197,38 @@ pub fn transition(state: Stage, event: PipelineEvent) -> Result<Stage, Transitio
             reason: ArchiveReason::Superseded { by },
         }),
 
-        // ── Unblock: only from Archived(Blocked) → Backlog ─────────────
+        // ── Reject from any active stage or QA ──────────────────────────
+        (Backlog, Reject { reason })
+        | (Coding, Reject { reason })
+        | (Qa, Reject { reason })
+        | (Merge { .. }, Reject { reason }) => Ok(Archived {
+            archived_at: now,
+            reason: ArchiveReason::Rejected { reason },
+        }),
+
+        // ── Demote: send an active item back to backlog ────────────────
+        (Coding, Demote) | (Qa, Demote) | (Merge { .. }, Demote) => Ok(Backlog),
+
+        // ── Close: direct completion from any active stage ─────────────
+        (Backlog, Close) | (Coding, Close) | (Qa, Close) | (Merge { .. }, Close) => Ok(Done {
+            merged_at: now,
+            merge_commit: GitSha("closed".to_string()),
+        }),
+
+        // ── Unblock: from Archived(Blocked) or Archived(MergeFailed) → Backlog
         (
             Archived {
                 reason: ArchiveReason::Blocked { .. },
                 ..
             },
             Unblock,
+        )
+        | (
+            Archived {
+                reason: ArchiveReason::MergeFailed { .. },
+                ..
+            },
+            Unblock,
         ) => Ok(Backlog),
 
         // ── Everything else is invalid ──────────────────────────────────

server/src/pipeline_state/types.rs

@@ -48,8 +48,30 @@ impl fmt::Display for AgentName {
 ///   - `agent` — local execution state, not pipeline state
 ///   - `retry_count` — also local
 ///   - `blocked` — folded into `Archived { reason: Blocked { .. } }`
+///
+/// ## Canonical state machine (story 857)
+///
+/// The following named lifecycle states map to `Stage` variants:
+///
+/// | Lifecycle state | Stage variant                     |
+/// |-----------------|-----------------------------------|
+/// | upcoming        | `Upcoming`                        |
+/// | backlog         | `Backlog`                         |
+/// | current         | `Coding`                          |
+/// | qa_pending      | `Qa`                              |
+/// | merge_pending   | `Merge { .. }`                    |
+/// | done            | `Done { .. }`                     |
+/// | blocked         | `Archived { Blocked { .. } }`     |
+/// | merge_failure   | `Archived { MergeFailed { .. } }` |
+/// | archived        | `Archived { Completed }`          |
+/// | superseded      | `Archived { Superseded { .. } }`  |
+/// | rejected        | `Archived { Rejected { .. } }`    |
+/// | abandoned       | `Archived { Abandoned }`          |
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub enum Stage {
+    /// Story has been created but not yet triaged into the backlog.
+    Upcoming,
+
     /// Story exists, waiting for dependencies or auto-assign promotion.
     Backlog,
 
@@ -96,6 +118,8 @@ pub enum ArchiveReason {
     MergeFailed { reason: String },
     /// Held in review at human request.
     ReviewHeld { reason: String },
+    /// Story rejected by QA or reviewer with an explanation.
+    Rejected { reason: String },
 }
 
 // ── Stage convenience methods ──────────────────────────────────────────────
@@ -106,6 +130,11 @@ impl Stage {
         matches!(self, Stage::Coding | Stage::Qa | Stage::Merge { .. })
     }
 
+    /// Returns true if this is the Upcoming variant.
+    pub fn is_upcoming(&self) -> bool {
+        matches!(self, Stage::Upcoming)
+    }
+
     /// Returns the filesystem directory name for this stage.
     pub fn dir_name(&self) -> &'static str {
         stage_dir_name(self)
@@ -132,6 +161,7 @@ impl Stage {
     /// accessing the rich metadata fields.
     pub fn from_dir(s: &str) -> Option<Self> {
         match s {
+            "0_upcoming" => Some(Stage::Upcoming),
             "1_backlog" => Some(Stage::Backlog),
             "2_current" => Some(Stage::Coding),
             "3_qa" => Some(Stage::Qa),
@@ -219,6 +249,7 @@ impl std::error::Error for TransitionError {}
 /// Human-readable label for a `Stage` variant.
 pub fn stage_label(s: &Stage) -> &'static str {
     match s {
+        Stage::Upcoming => "Upcoming",
         Stage::Backlog => "Backlog",
         Stage::Coding => "Coding",
         Stage::Qa => "Qa",
@@ -231,6 +262,7 @@ pub fn stage_label(s: &Stage) -> &'static str {
 /// Map a Stage to the filesystem directory name used by the work pipeline.
 pub fn stage_dir_name(s: &Stage) -> &'static str {
     match s {
+        Stage::Upcoming => "0_upcoming",
         Stage::Backlog => "1_backlog",
         Stage::Coding => "2_current",
         Stage::Qa => "3_qa",