huskies: merge 1088

server/src/crdt_state/mod.rs

@@ -56,7 +56,8 @@ pub use write::{
     bump_retry_count, migrate_legacy_stage_strings, migrate_merge_job, migrate_names_from_slugs,
     migrate_node_claims_to_agent_claims, migrate_story_ids_to_numeric, name_from_story_id,
     purge_done_stage_merge_jobs, set_agent, set_depends_on, set_epic, set_item_type, set_name,
-    set_plan_state, set_qa_mode, set_resume_to, set_resume_to_raw, set_retry_count, write_item,
+    set_origin, set_plan_state, set_qa_mode, set_resume_to, set_resume_to_raw, set_retry_count,
+    write_item,
 };
 
 #[cfg(test)]

server/src/crdt_state/read.rs

@@ -29,6 +29,8 @@ pub struct CrdtItemDump {
     /// Hex-encoded OpId of the list insert op — cross-reference with `crdt_ops`.
     pub content_index: String,
     pub is_deleted: bool,
+    /// Origin JSON string, or `None` for items that pre-date story 1088.
+    pub origin: Option<String>,
 }
 
 /// Top-level debug dump of the in-memory CRDT state.
@@ -149,6 +151,10 @@ pub fn dump_crdt_state(story_id_filter: Option<&str>) -> CrdtStateDump {
             JsonValue::Number(n) if n > 0.0 => Some(n),
             _ => None,
         };
+        let origin = match item_crdt.origin.view() {
+            JsonValue::String(s) if !s.is_empty() => Some(s),
+            _ => None,
+        };
 
         let content_index = op.id.iter().map(|b| format!("{b:02x}")).collect::<String>();
 
@@ -163,6 +169,7 @@ pub fn dump_crdt_state(story_id_filter: Option<&str>) -> CrdtStateDump {
             claim_ts,
             content_index,
             is_deleted: op.is_deleted,
+            origin,
         });
     }
 
@@ -408,6 +415,11 @@ pub(super) fn extract_item_view(item: &PipelineItemCrdt) -> Option<PipelineItemV
         _ => None,
     };
 
+    let origin = match item.origin.view() {
+        JsonValue::String(s) if !s.is_empty() => Some(s),
+        _ => None,
+    };
+
     let stage = project_stage_for_view(
         &stage_str,
         &story_id,
@@ -429,6 +441,7 @@ pub(super) fn extract_item_view(item: &PipelineItemCrdt) -> Option<PipelineItemV
         qa_mode,
         item_type,
         epic,
+        origin,
     })
 }
 

server/src/crdt_state/types.rs

@@ -105,6 +105,14 @@ pub struct PipelineItemCrdt {
     /// 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>,
+    /// Story 1088: origin of the work item — who or what created it.
+    ///
+    /// Stored as a compact JSON string, e.g.
+    /// `{"kind":"user","id":"","ts":1716768000.0}` or
+    /// `{"kind":"agent","id":"coder-1","ts":1716768000.0}`.
+    /// Empty string on older items that pre-date this register; the typed
+    /// read path surfaces those as `None`, which the UI renders as `"unknown"`.
+    pub origin: LwwRegisterCrdt<String>,
 }
 
 /// CRDT node that holds a single peer's presence entry.
@@ -203,6 +211,9 @@ pub struct WorkItem {
     pub(super) item_type: Option<crate::io::story_metadata::ItemType>,
     /// Epic this item belongs to. `None` when the item has no parent epic.
     pub(super) epic: Option<EpicId>,
+    /// Origin of the work item (story 1088).  `None` for items created before
+    /// the origin register was introduced; those display as `"unknown"`.
+    pub(super) origin: Option<String>,
 }
 
 impl WorkItem {
@@ -261,6 +272,12 @@ impl WorkItem {
         self.epic
     }
 
+    /// Origin of the work item (story 1088), or `None` for items created before
+    /// the origin register was introduced.
+    pub fn origin(&self) -> Option<&str> {
+        self.origin.as_deref()
+    }
+
     /// Construct a `WorkItem` for use in tests outside `crdt_state::*`.
     ///
     /// Within `crdt_state` use a struct literal directly (fields are `pub(super)`).
@@ -286,6 +303,7 @@ impl WorkItem {
             qa_mode,
             item_type,
             epic,
+            origin: None,
         }
     }
 }

server/src/crdt_state/write/item.rs

@@ -235,6 +235,31 @@ pub fn set_plan_state(story_id: &str, state: crate::pipeline_state::PlanState) -
     true
 }
 
+/// Set the `origin` CRDT register for a pipeline item (story 1088).
+///
+/// Writes a compact JSON string describing who or what created the item, e.g.
+/// `{"kind":"user","id":"","ts":1716768000.0}` or
+/// `{"kind":"agent","id":"coder-1","ts":1716768000.0}`.
+///
+/// Passing an empty string is treated as "no origin set" (equivalent to the
+/// pre-1088 state for older items).  Returns `true` if the item was found and
+/// the op was applied, `false` otherwise.
+pub fn set_origin(story_id: &str, origin: &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;
+    };
+    apply_and_persist(&mut state, |s| {
+        s.crdt.doc.items[idx].origin.set(origin.to_string())
+    });
+    true
+}
+
 /// Write a pipeline item state through CRDT operations.
 ///
 /// If the item exists, updates its registers.  If not, inserts a new item
@@ -394,6 +419,7 @@ pub fn write_item(
             "resume_to": "",
             "plan_state": "",
             "merge_server_start": merge_server_start_val,
+            "origin": "",
         })
         .into();
 
@@ -424,6 +450,7 @@ pub fn write_item(
             item.resume_to.advance_seq(floor);
             item.plan_state.advance_seq(floor);
             item.merge_server_start.advance_seq(floor);
+            item.origin.advance_seq(floor);
         }
 
         // Broadcast a CrdtEvent for the new item.

server/src/crdt_state/write/mod.rs

@@ -10,8 +10,8 @@ mod migrations;
 mod tests;
 
 pub use item::{
-    bump_retry_count, set_agent, set_depends_on, set_epic, set_item_type, set_name, set_plan_state,
-    set_qa_mode, set_resume_to, set_resume_to_raw, set_retry_count, write_item,
+    bump_retry_count, set_agent, set_depends_on, set_epic, set_item_type, set_name, set_origin,
+    set_plan_state, set_qa_mode, set_resume_to, set_resume_to_raw, set_retry_count, write_item,
 };
 
 #[cfg(test)]