huskies: merge 1036

server/src/crdt_state/read.rs

@@ -403,6 +403,11 @@ pub(super) fn extract_item_view(item: &PipelineItemCrdt) -> Option<PipelineItemV
         _ => None,
     };
 
+    let merge_server_start = match item.merge_server_start.view() {
+        JsonValue::Number(n) if n > 0.0 => Some(n),
+        _ => None,
+    };
+
     let stage = project_stage_for_view(
         &stage_str,
         &story_id,
@@ -412,6 +417,7 @@ pub(super) fn extract_item_view(item: &PipelineItemCrdt) -> Option<PipelineItemV
         claim_ts_secs,
         plan_state_str.as_deref(),
         retry_count_register,
+        merge_server_start,
     )?;
 
     Some(PipelineItemView {
@@ -449,6 +455,7 @@ fn project_stage_for_view(
     claim_ts_secs: Option<u64>,
     plan_state_str: Option<&str>,
     retries: u32,
+    merge_server_start: Option<f64>,
 ) -> Option<crate::pipeline_state::Stage> {
     use crate::pipeline_state::{
         AgentClaim, AgentName, ArchiveReason, BranchName, GitSha, PlanState, Stage,
@@ -518,6 +525,7 @@ fn project_stage_for_view(
             commits_ahead: NonZeroU32::new(1).expect("1 is non-zero"),
             claim,
             retries,
+            server_start_time: merge_server_start,
         }),
         "merge_failure" => {
             // Story 986: read the typed kind directly from ContentKey::MergeFailureKind

server/src/crdt_state/types.rs

@@ -100,6 +100,11 @@ pub struct PipelineItemCrdt {
     /// Wire values: `"missing"` (default/empty), `"drafted"`, `"confirmed"`.
     /// Updated by the filesystem watcher on PLAN.md create/modify/remove events.
     pub plan_state: LwwRegisterCrdt<String>,
+    /// Story 1036: Unix timestamp (f64 seconds) of the server process that
+    /// started the currently active merge task for this item.  Zero / absent
+    /// means no merge task is in flight.  Projected into `Stage::Merge {
+    /// server_start_time }` so callers never read this register directly.
+    pub merge_server_start: LwwRegisterCrdt<f64>,
 }
 
 /// CRDT node that holds a single peer's presence entry.

server/src/crdt_state/write/item.rs

@@ -267,6 +267,14 @@ pub fn write_item(
         Stage::Merge { retries, .. } => *retries as f64,
         _ => 0.0,
     };
+    // Extract merge_server_start from Stage::Merge; 0.0 clears the register.
+    let merge_server_start_val: f64 = match stage {
+        Stage::Merge {
+            server_start_time: Some(t),
+            ..
+        } => *t,
+        _ => 0.0,
+    };
     let Some(state_mutex) = get_crdt() else {
         return;
     };
@@ -335,6 +343,11 @@ pub fn write_item(
         apply_and_persist(&mut state, |s| {
             s.crdt.doc.items[idx].claim_ts.set(claim_ts_val)
         });
+        apply_and_persist(&mut state, |s| {
+            s.crdt.doc.items[idx]
+                .merge_server_start
+                .set(merge_server_start_val)
+        });
         if let Some(ma) = merged_at {
             apply_and_persist(&mut state, |s| s.crdt.doc.items[idx].merged_at.set(ma));
         }
@@ -380,6 +393,7 @@ pub fn write_item(
             "epic": "",
             "resume_to": "",
             "plan_state": "",
+            "merge_server_start": merge_server_start_val,
         })
         .into();
 
@@ -409,6 +423,7 @@ pub fn write_item(
             item.epic.advance_seq(floor);
             item.resume_to.advance_seq(floor);
             item.plan_state.advance_seq(floor);
+            item.merge_server_start.advance_seq(floor);
         }
 
         // Broadcast a CrdtEvent for the new item.
@@ -484,11 +499,13 @@ pub fn set_retry_count(story_id: &str, count: i64) {
             commits_ahead,
             claim,
             retries: _,
+            server_start_time,
         } => Stage::Merge {
             feature_branch,
             commits_ahead,
             claim,
             retries: count.max(0) as u32,
+            server_start_time,
         },
         _ => return,
     };
@@ -525,6 +542,7 @@ pub fn bump_retry_count(story_id: &str) -> i64 {
             commits_ahead,
             claim,
             retries,
+            server_start_time,
         } => {
             let n = retries + 1;
             (
@@ -533,6 +551,7 @@ pub fn bump_retry_count(story_id: &str) -> i64 {
                     commits_ahead,
                     claim,
                     retries: n,
+                    server_start_time,
                 },
                 n,
             )

server/src/crdt_state/write/migrations.rs

@@ -391,6 +391,7 @@ mod stage_migration_tests {
                     commits_ahead: NonZeroU32::new(1).unwrap(),
                     claim: None,
                     retries: 0,
+                    server_start_time: None,
                 },
             ),
             (