huskies: merge 1019

server/src/io/watcher/mod.rs

@@ -20,6 +20,8 @@ mod events;
 mod sweep;
 
 pub use events::WatcherEvent;
+pub(crate) use sweep::spawn_done_to_archived_subscriber;
+#[cfg(test)]
 pub(crate) use sweep::sweep_done_to_archived;
 
 use crate::slog;

server/src/io/watcher/sweep.rs

@@ -1,12 +1,75 @@
-//! Periodic sweep of completed work items from `done` to `archived`.
+//! Sweep and reactive subscriber for promoting `done` items to `archived`.
 //!
 //! Items in `Stage::Done` whose `merged_at` timestamp exceeds the configured
 //! retention duration are promoted to `Stage::Archived` via the canonical
 //! pipeline state machine (story 934, stage 5).
+//!
+//! `sweep_done_to_archived` is the synchronous one-shot sweep (used in tests
+//! and for backwards-compat call sites). `spawn_done_to_archived_subscriber`
+//! is the reactive replacement for the tick-loop periodic scan.
 
 use crate::slog;
+use crate::slog_warn;
 use std::time::Duration;
 
+/// Spawn a subscriber that archives each `Stage::Done` story after the
+/// configured retention period expires.
+///
+/// Subscribes to the pipeline [`TransitionFired`][crate::pipeline_state::TransitionFired]
+/// broadcast channel. On each `Stage::Done` transition, spawns a short-lived
+/// task that sleeps for the remaining retention time (computed from the story's
+/// `merged_at` timestamp) and then calls
+/// [`apply_transition`][crate::pipeline_state::apply_transition] with
+/// `PipelineEvent::Accepted` to move it to `Stage::Archived`.
+///
+/// Using `merged_at` rather than a fixed sleep means the subscriber correctly
+/// handles stories that have been in `Done` for hours before a server restart:
+/// the computed remaining time will be small (or zero), so archival happens
+/// promptly rather than waiting another full retention period.
+///
+/// Replaces the periodic `sweep_done_to_archived` call from the tick loop.
+pub(crate) fn spawn_done_to_archived_subscriber(done_retention: Duration) {
+    use crate::pipeline_state::{PipelineEvent, Stage, apply_transition, subscribe_transitions};
+
+    let mut rx = subscribe_transitions();
+    tokio::spawn(async move {
+        loop {
+            match rx.recv().await {
+                Ok(fired) => {
+                    if let Stage::Done { merged_at, .. } = fired.after {
+                        let story_id = fired.story_id.0.clone();
+                        let retention = done_retention;
+                        tokio::spawn(async move {
+                            let age = chrono::Utc::now()
+                                .signed_duration_since(merged_at)
+                                .to_std()
+                                .unwrap_or_default();
+                            if age < retention {
+                                tokio::time::sleep(retention - age).await;
+                            }
+                            match apply_transition(&story_id, PipelineEvent::Accepted, None) {
+                                Ok(_) => {
+                                    slog!("[watcher] sweep: promoted {story_id} → archived")
+                                }
+                                Err(e) => {
+                                    slog!("[watcher] sweep: transition error for {story_id}: {e}")
+                                }
+                            }
+                        });
+                    }
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
+                    slog_warn!(
+                        "[done-archive-sub] Lagged, skipped {n} event(s); some Done stories \
+                         may not auto-archive."
+                    );
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            }
+        }
+    });
+}
+
 /// Sweep items in `Stage::Done` whose `merged_at` timestamp exceeds the
 /// retention duration to `Stage::Archived` via the typed transition table.
 ///
@@ -14,6 +77,10 @@ use std::time::Duration;
 /// `Done + Accepted → Archived` transition is validated and a
 /// `TransitionFired` event is emitted to subscribers (worktree pruning,
 /// matrix notifier, etc.).
+///
+/// Used in tests for direct one-shot sweeps; production code uses
+/// [`spawn_done_to_archived_subscriber`] instead.
+#[cfg(test)]
 pub(crate) fn sweep_done_to_archived(done_retention: Duration) {
     use crate::pipeline_state::{PipelineEvent, Stage, apply_transition, read_all_typed};
 

server/src/io/watcher/tests.rs

@@ -259,6 +259,48 @@ fn sweep_uses_crdt_merged_at_not_utc_now() {
     );
 }
 
+// ── spawn_done_to_archived_subscriber (reactive) ─────────────────────────
+//
+// AC4: tests that the TransitionFired subscriber archives Done stories
+// on transition rather than on a periodic tick.
+
+/// Moving a story to Done fires the subscriber; with zero retention the
+/// subscriber archives it immediately.
+#[tokio::test]
+async fn done_to_archived_subscriber_archives_on_transition() {
+    crate::crdt_state::init_for_test();
+    crate::db::ensure_content_store();
+
+    let story_id = "9886_sub_archive_reactive";
+    crate::db::write_item_with_content(
+        story_id,
+        "1_backlog",
+        "---\nname: Reactive archive test\n---\n",
+        crate::db::ItemMeta::named("Reactive archive test"),
+    );
+
+    // Zero retention: archive immediately after the Done transition.
+    spawn_done_to_archived_subscriber(Duration::ZERO);
+
+    // Trigger Done via Close event (valid from Backlog → Done).
+    crate::pipeline_state::apply_transition(
+        story_id,
+        crate::pipeline_state::PipelineEvent::Close,
+        None,
+    )
+    .expect("Close transition must succeed from Backlog");
+
+    // Give the subscriber task time to process and archive.
+    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+
+    let items = crate::pipeline_state::read_all_typed();
+    let item = items.iter().find(|i| i.story_id.0 == story_id);
+    assert!(
+        item.is_some_and(|i| matches!(i.stage, crate::pipeline_state::Stage::Archived { .. })),
+        "story should be archived after Done transition with zero retention"
+    );
+}
+
 /// Prove that an item with merged_at NEWER than done_retention is NOT swept.
 #[test]
 fn sweep_keeps_item_newer_than_retention() {