crashlabs/huskies
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 26 Wiki Activity

huskies: merge 520_story_typed_pipeline_state_machine_in_rust_foundation_replaces_stringly_typed_crdt_views_with_strict_enums_subsumes_436

Browse Source
This commit is contained in:
dave
2026-04-09 21:24:11 +00:00
parent 1d9287389a
commit 84717b04bd
18 changed files with 1569 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/agents/lifecycle.rs
+6 -4
View File
@@ -33,15 +33,17 @@ fn move_item<'a>(
fields_to_clear: &[&str],
) -> Result<Option<&'a str>, String> {
// Check if the item is already in the target stage or a done stage.
if let Some(item) = crate::crdt_state::read_item(story_id) {
if item.stage == target_dir
|| extra_done_dirs.iter().any(|d| item.stage == *d)
// Use the typed projection for compile-safe stage comparison.
if let Ok(Some(typed_item)) = crate::pipeline_state::read_typed(story_id) {
let current_dir = typed_item.stage.dir_name();
if current_dir == target_dir
|| extra_done_dirs.contains(&current_dir)
{
return Ok(None); // Idempotent: already there.
}
// Verify it's in one of the expected source stages.
let src_dir = sources.iter().find(|&&s| item.stage == s).copied();
let src_dir = sources.iter().find(|&&s| current_dir == s).copied();
if src_dir.is_none() && !missing_ok {
let locs = sources
.iter()
server/src/agents/pool/auto_assign/scan.rs
+4 -6
View File
@@ -22,12 +22,10 @@ pub(super) fn scan_stage_items(project_root: &Path, stage_dir: &str) -> Vec<Stri
use std::collections::BTreeSet;
let mut items = BTreeSet::new();
// Include CRDT items — the primary source of truth for pipeline state.
if let Some(all) = crate::crdt_state::read_all_items() {
for item in &all {
if item.stage == stage_dir {
items.insert(item.story_id.clone());
}
// Include CRDT items via the typed projection — the primary source of truth.
for item in crate::pipeline_state::read_all_typed() {
if item.stage.dir_name() == stage_dir {
items.insert(item.story_id.0.clone());
}
}
server/src/agents/pool/auto_assign/story_checks.rs
+2 -2
View File
@@ -77,7 +77,7 @@ pub(super) fn has_unmet_dependencies(
return true;
}
// If the CRDT had the item and returned empty deps, it means all are met.
if crate::crdt_state::read_item(story_id).is_some() {
if crate::pipeline_state::read_typed(story_id).ok().flatten().is_some() {
return false;
}
// Fallback: filesystem check (CRDT not initialised or item not yet in CRDT).
@@ -96,7 +96,7 @@ pub(super) fn check_archived_dependencies(
story_id: &str,
) -> Vec<u32> {
// Prefer CRDT-based check when the item is known to CRDT.
if crate::crdt_state::read_item(story_id).is_some() {
if crate::pipeline_state::read_typed(story_id).ok().flatten().is_some() {
return crate::crdt_state::check_archived_deps_crdt(story_id);
}
// Fallback: filesystem.
server/src/agents/pool/pipeline/advance.rs
+8 -4
View File
@@ -395,8 +395,10 @@ fn write_review_hold_to_store(story_id: &str) {
let updated = crate::io::story_metadata::write_review_hold_in_content(&contents);
crate::db::write_content(story_id, &updated);
// Also persist to SQLite via shadow write.
let stage = crate::crdt_state::read_item(story_id)
.map(|i| i.stage)
let stage = crate::pipeline_state::read_typed(story_id)
.ok()
.flatten()
.map(|i| i.stage.dir_name().to_string())
.unwrap_or_else(|| "3_qa".to_string());
crate::db::write_item_with_content(story_id, &stage, &updated);
} else {
@@ -419,8 +421,10 @@ fn should_block_story(story_id: &str, max_retries: u32, stage_label: &str) -> Op
if let Some(contents) = crate::db::read_content(story_id) {
let (updated, new_count) = increment_retry_count_in_content(&contents);
crate::db::write_content(story_id, &updated);
let stage = crate::crdt_state::read_item(story_id)
.map(|i| i.stage)
let stage = crate::pipeline_state::read_typed(story_id)
.ok()
.flatten()
.map(|i| i.stage.dir_name().to_string())
.unwrap_or_else(|| "2_current".to_string());
crate::db::write_item_with_content(story_id, &stage, &updated);
server/src/agents/pool/worktree.rs
+6 -9
View File
@@ -23,18 +23,15 @@ impl AgentPool {
/// Return the active pipeline stage directory name for `story_id`, or `None` if the
/// story is not in any active stage (`2_current/`, `3_qa/`, `4_merge/`).
pub(super) fn find_active_story_stage(project_root: &Path, story_id: &str) -> Option<&'static str> {
const STAGES: [&str; 3] = ["2_current", "3_qa", "4_merge"];
// Try CRDT first — primary source of truth.
if let Some(item) = crate::crdt_state::read_item(story_id) {
for stage in &STAGES {
if item.stage == *stage {
return Some(stage);
}
}
// Try typed CRDT projection first — primary source of truth.
if let Ok(Some(item)) = crate::pipeline_state::read_typed(story_id)
&& item.stage.is_active()
{
return Some(item.stage.dir_name());
}
// Also check filesystem (backwards compat / tests).
const STAGES: [&str; 3] = ["2_current", "3_qa", "4_merge"];
for stage in &STAGES {
let path = project_root
.join(".huskies")
server/src/chat/commands/depends.rs
+2 -2
View File
@@ -67,12 +67,12 @@ pub(super) fn handle_depends(ctx: &CommandContext) -> Option<String> {
// --- DB-first lookup ---
for id in crate::db::all_content_ids() {
let file_num = id.split('_').next().unwrap_or("");
if file_num == num_str && let Some(item) = crate::crdt_state::read_item(&id) {
if file_num == num_str && let Ok(Some(item)) = crate::pipeline_state::read_typed(&id) {
let path = ctx
.project_root
.join(".huskies")
.join("work")
.join(&item.stage)
.join(item.stage.dir_name())
.join(format!("{id}.md"));
found = Some((path, id));
break;
server/src/chat/commands/unblock.rs
+4 -2
View File
@@ -129,8 +129,10 @@ fn unblock_by_story_id(story_id: &str) -> String {
updated = set_front_matter_field(&updated, "retry_count", "0");
crate::db::write_content(story_id, &updated);
let stage = crate::crdt_state::read_item(story_id)
.map(|i| i.stage)
let stage = crate::pipeline_state::read_typed(story_id)
.ok()
.flatten()
.map(|i| i.stage.dir_name().to_string())
.unwrap_or_else(|| "2_current".to_string());
crate::db::write_item_with_content(story_id, &stage, &updated);
server/src/chat/timer.rs
+9 -7
View File
@@ -207,14 +207,15 @@ async fn tick_once(
// remove the timer before it fires; this guard covers the case where
// cancellation was not yet called or the story raced forward through
// the pipeline while the timer was pending.
if let Some(item) = crate::crdt_state::read_item(&entry.story_id) {
match item.stage.as_str() {
"3_qa" | "4_merge" | "5_done" | "6_archived" => {
if let Ok(Some(item)) = crate::pipeline_state::read_typed(&entry.story_id) {
use crate::pipeline_state::Stage;
match &item.stage {
Stage::Qa | Stage::Merge { .. } | Stage::Done { .. } | Stage::Archived { .. } => {
crate::slog!(
"[timer] Skipping timer for story {} — currently in '{}', \
not in backlog/current; timer is stale",
entry.story_id,
item.stage
item.stage.dir_name()
);
continue;
}
@@ -425,8 +426,9 @@ pub async fn handle_timer_command(
// The story must be in backlog or current. When the timer fires,
// backlog stories are moved to current automatically.
// Check CRDT state first, then fall back to filesystem.
let in_valid_stage = if let Some(item) = crate::crdt_state::read_item(&story_id) {
matches!(item.stage.as_str(), "1_backlog" | "2_current")
let in_valid_stage = if let Ok(Some(item)) = crate::pipeline_state::read_typed(&story_id) {
use crate::pipeline_state::Stage;
matches!(item.stage, Stage::Backlog | Stage::Coding)
} else {
let work_dir = project_root.join(".huskies").join("work");
work_dir.join("1_backlog").join(format!("{story_id}.md")).exists()
@@ -588,7 +590,7 @@ fn resolve_story_id(number_or_id: &str, project_root: &Path) -> Option<String> {
// --- DB-first lookup ---
for id in crate::db::all_content_ids() {
let file_num = id.split('_').next().unwrap_or("");
if file_num == number_or_id && crate::crdt_state::read_item(&id).is_some() {
if file_num == number_or_id && crate::pipeline_state::read_typed(&id).ok().flatten().is_some() {
return Some(id);
}
}
server/src/chat/transport/matrix/assign.rs
+2 -2
View File
@@ -108,11 +108,11 @@ pub async fn handle_assign(
// --- DB-first lookup ---
for id in crate::db::all_content_ids() {
let file_num = id.split('_').next().unwrap_or("");
if file_num == story_number && let Some(item) = crate::crdt_state::read_item(&id) {
if file_num == story_number && let Ok(Some(item)) = crate::pipeline_state::read_typed(&id) {
let path = project_root
.join(".huskies")
.join("work")
.join(&item.stage)
.join(item.stage.dir_name())
.join(format!("{id}.md"));
found = Some((path, id));
break;
server/src/chat/transport/matrix/delete.rs
+3 -3
View File
@@ -76,13 +76,13 @@ pub async fn handle_delete(
// --- DB-first lookup ---
for id in crate::db::all_content_ids() {
let file_num = id.split('_').next().unwrap_or("");
if file_num == story_number && let Some(item) = crate::crdt_state::read_item(&id) {
if file_num == story_number && let Ok(Some(item)) = crate::pipeline_state::read_typed(&id) {
let path = project_root
.join(".huskies")
.join("work")
.join(&item.stage)
.join(item.stage.dir_name())
.join(format!("{id}.md"));
found = Some((path, item.stage, id));
found = Some((path, item.stage.dir_name().to_string(), id));
break;
}
}
server/src/chat/transport/matrix/start.rs
+2 -2
View File
@@ -95,11 +95,11 @@ pub async fn handle_start(
// --- DB-first lookup ---
for id in crate::db::all_content_ids() {
let file_num = id.split('_').next().unwrap_or("");
if file_num == story_number && let Some(item) = crate::crdt_state::read_item(&id) {
if file_num == story_number && let Ok(Some(item)) = crate::pipeline_state::read_typed(&id) {
let path = project_root
.join(".huskies")
.join("work")
.join(&item.stage)
.join(item.stage.dir_name())
.join(format!("{id}.md"));
found = Some((path, id));
break;
server/src/db/mod.rs
+10 -11
View File
@@ -309,17 +309,16 @@ pub fn delete_item(story_id: &str) {
pub fn next_item_number() -> u32 {
let mut max_num: u32 = 0;
// Scan CRDT items.
if let Some(items) = crate::crdt_state::read_all_items() {
for item in &items {
let num_str: String = item
.story_id
.chars()
.take_while(|c| c.is_ascii_digit())
.collect();
if let Ok(n) = num_str.parse::<u32>() && n > max_num {
max_num = n;
}
// Scan CRDT items via typed projection.
for item in crate::pipeline_state::read_all_typed() {
let num_str: String = item
.story_id
.0
.chars()
.take_while(|c| c.is_ascii_digit())
.collect();
if let Ok(n) = num_str.parse::<u32>() && n > max_num {
max_num = n;
}
}
server/src/http/mcp/mod.rs
+1 -1
View File
@@ -1461,7 +1461,7 @@ mod tests {
assert!(names.contains(&"git_log"));
assert!(names.contains(&"status"));
assert!(names.contains(&"loc_file"));
assert_eq!(tools.len(), 57);
assert_eq!(tools.len(), 58);
}
#[test]
server/src/http/mcp/story_tools.rs
+5 -3
View File
@@ -376,8 +376,10 @@ pub(super) fn tool_update_story(args: &Value, ctx: &AppContext) -> Result<String
update_story_in_file(&root, story_id, user_story, description, front_matter_opt)?;
// Bug 503: warn if any depends_on in the (now updated) story points at an archived story.
let stage = crate::crdt_state::read_item(story_id)
.map(|i| i.stage)
let stage = crate::pipeline_state::read_typed(story_id)
.ok()
.flatten()
.map(|i| i.stage.dir_name().to_string())
.unwrap_or_else(|| "1_backlog".to_string());
let archived_deps = check_archived_deps(&root, &stage, story_id);
if !archived_deps.is_empty() {
@@ -525,7 +527,7 @@ pub(super) async fn tool_delete_story(args: &Value, ctx: &AppContext) -> Result<
// 4. Delete from database content store and CRDT.
let found_in_db = crate::db::read_content(story_id).is_some()
|| crate::crdt_state::read_item(story_id).is_some();
|| crate::pipeline_state::read_typed(story_id).ok().flatten().is_some();
crate::db::delete_item(story_id);
server/src/http/workflow/bug_ops.rs
+27 -29
View File
@@ -202,21 +202,20 @@ pub fn list_bug_files(root: &Path) -> Result<Vec<(String, String)>, String> {
let mut bugs = Vec::new();
let mut seen = std::collections::HashSet::new();
// First: CRDT items in backlog that are bugs.
if let Some(items) = crate::crdt_state::read_all_items() {
for item in items {
if item.stage != "1_backlog" || !is_bug_item(&item.story_id) {
continue;
}
let name = item.name.clone()
.or_else(|| {
crate::db::read_content(&item.story_id)
.and_then(|c| extract_bug_name_from_content(&c))
})
.unwrap_or_else(|| item.story_id.clone());
seen.insert(item.story_id.clone());
bugs.push((item.story_id, name));
// First: typed projection items in backlog that are bugs.
for item in crate::pipeline_state::read_all_typed() {
if !matches!(item.stage, crate::pipeline_state::Stage::Backlog) || !is_bug_item(&item.story_id.0) {
continue;
}
let sid = item.story_id.0;
let name = if item.name.is_empty() { None } else { Some(item.name) }
.or_else(|| {
crate::db::read_content(&sid)
.and_then(|c| extract_bug_name_from_content(&c))
})
.unwrap_or_else(|| sid.clone());
seen.insert(sid.clone());
bugs.push((sid, name));
}
// Then: filesystem fallback.
@@ -267,22 +266,21 @@ pub fn list_refactor_files(root: &Path) -> Result<Vec<(String, String)>, String>
let mut refactors = Vec::new();
let mut seen = std::collections::HashSet::new();
// First: CRDT items.
if let Some(items) = crate::crdt_state::read_all_items() {
for item in items {
if item.stage != "1_backlog" || !is_refactor_item(&item.story_id) {
continue;
}
let name = item.name.clone()
.or_else(|| {
crate::db::read_content(&item.story_id)
.and_then(|c| parse_front_matter(&c).ok())
.and_then(|m| m.name)
})
.unwrap_or_else(|| item.story_id.clone());
seen.insert(item.story_id.clone());
refactors.push((item.story_id, name));
// First: typed projection items.
for item in crate::pipeline_state::read_all_typed() {
if !matches!(item.stage, crate::pipeline_state::Stage::Backlog) || !is_refactor_item(&item.story_id.0) {
continue;
}
let sid = item.story_id.0;
let name = if item.name.is_empty() { None } else { Some(item.name) }
.or_else(|| {
crate::db::read_content(&sid)
.and_then(|c| parse_front_matter(&c).ok())
.and_then(|m| m.name)
})
.unwrap_or_else(|| sid.clone());
seen.insert(sid.clone());
refactors.push((sid, name));
}
// Then: filesystem fallback.
server/src/http/workflow/mod.rs
+90 -35
View File
@@ -79,8 +79,11 @@ pub struct PipelineState {
pub fn load_pipeline_state(ctx: &AppContext) -> Result<PipelineState, String> {
let agent_map = build_active_agent_map(ctx);
// Try CRDT-first read.
if let Some(crdt_items) = crate::crdt_state::read_all_items() {
// Try CRDT-first read via the typed projection layer.
let typed_items = crate::pipeline_state::read_all_typed();
if !typed_items.is_empty() {
use crate::pipeline_state::Stage;
let mut state = PipelineState {
backlog: Vec::new(),
current: Vec::new(),
@@ -89,11 +92,12 @@ pub fn load_pipeline_state(ctx: &AppContext) -> Result<PipelineState, String> {
done: Vec::new(),
};
for item in crdt_items {
let agent = agent_map.get(&item.story_id).cloned();
for item in typed_items {
let sid = &item.story_id.0;
let agent = agent_map.get(sid).cloned();
// Enrich with content-derived metadata (merge_failure, review_hold, qa).
let (merge_failure, review_hold, qa) = crate::db::read_content(&item.story_id)
let (merge_failure, review_hold, qa) = crate::db::read_content(sid)
.and_then(|c| parse_front_matter(&c).ok())
.map(|meta| {
(
@@ -105,24 +109,45 @@ pub fn load_pipeline_state(ctx: &AppContext) -> Result<PipelineState, String> {
.unwrap_or((None, None, None));
let story = UpcomingStory {
story_id: item.story_id,
name: item.name,
story_id: sid.clone(),
name: if item.name.is_empty() {
None
} else {
Some(item.name.clone())
},
error: None,
merge_failure,
agent,
review_hold,
qa,
retry_count: item.retry_count.map(|r| r as u32),
blocked: item.blocked,
depends_on: item.depends_on,
retry_count: if item.retry_count > 0 {
Some(item.retry_count)
} else {
None
},
blocked: if item.stage.is_blocked() {
Some(true)
} else {
None
},
depends_on: if item.depends_on.is_empty() {
None
} else {
Some(
item.depends_on
.iter()
.filter_map(|d| d.0.split('_').next()?.parse::<u32>().ok())
.collect(),
)
},
};
match item.stage.as_str() {
"1_backlog" => state.backlog.push(story),
"2_current" => state.current.push(story),
"3_qa" => state.qa.push(story),
"4_merge" => state.merge.push(story),
"5_done" => state.done.push(story),
_ => {} // ignore archived or unknown stages
match &item.stage {
Stage::Backlog => state.backlog.push(story),
Stage::Coding => state.current.push(story),
Stage::Qa => state.qa.push(story),
Stage::Merge { .. } => state.merge.push(story),
Stage::Done { .. } => state.done.push(story),
Stage::Archived { .. } => {} // skip archived
}
}
@@ -256,22 +281,46 @@ fn load_stage_items_from_fs(
}
pub fn load_upcoming_stories(ctx: &AppContext) -> Result<Vec<UpcomingStory>, String> {
// Try CRDT first.
if let Some(crdt_items) = crate::crdt_state::read_all_items() {
let mut stories: Vec<UpcomingStory> = crdt_items
// Try typed projection first.
let typed_items = crate::pipeline_state::read_all_typed();
if !typed_items.is_empty() {
use crate::pipeline_state::Stage;
let mut stories: Vec<UpcomingStory> = typed_items
.into_iter()
.filter(|item| item.stage == "1_backlog")
.filter(|item| matches!(item.stage, Stage::Backlog))
.map(|item| UpcomingStory {
story_id: item.story_id,
name: item.name,
story_id: item.story_id.0,
name: if item.name.is_empty() {
None
} else {
Some(item.name)
},
error: None,
merge_failure: None,
agent: None,
review_hold: None,
qa: None,
retry_count: item.retry_count.map(|r| r as u32),
blocked: item.blocked,
depends_on: item.depends_on,
retry_count: if item.retry_count > 0 {
Some(item.retry_count)
} else {
None
},
blocked: if item.stage.is_blocked() {
Some(true)
} else {
None
},
depends_on: if item.depends_on.is_empty() {
None
} else {
Some(
item.depends_on
.iter()
.filter_map(|d| d.0.split('_').next()?.parse::<u32>().ok())
.collect(),
)
},
})
.collect();
stories.sort_by(|a, b| a.story_id.cmp(&b.story_id));
@@ -295,13 +344,16 @@ pub fn validate_story_dirs(
) -> Result<Vec<StoryValidationResult>, String> {
let mut results = Vec::new();
// Validate from CRDT + content store.
if let Some(crdt_items) = crate::crdt_state::read_all_items() {
for item in crdt_items {
if item.stage != "1_backlog" && item.stage != "2_current" {
// Validate from typed projection + content store.
{
let typed_items = crate::pipeline_state::read_all_typed();
for item in typed_items {
use crate::pipeline_state::Stage;
if !matches!(item.stage, Stage::Backlog | Stage::Coding) {
continue;
}
if let Some(content) = crate::db::read_content(&item.story_id) {
let sid = item.story_id.0.clone();
if let Some(content) = crate::db::read_content(&sid) {
match parse_front_matter(&content) {
Ok(meta) => {
let mut errors = Vec::new();
@@ -310,20 +362,20 @@ pub fn validate_story_dirs(
}
if errors.is_empty() {
results.push(StoryValidationResult {
story_id: item.story_id,
story_id: sid,
valid: true,
error: None,
});
} else {
results.push(StoryValidationResult {
story_id: item.story_id,
story_id: sid,
valid: false,
error: Some(errors.join("; ")),
});
}
}
Err(e) => results.push(StoryValidationResult {
story_id: item.story_id,
story_id: sid,
valid: false,
error: Some(e.to_string()),
}),
@@ -435,7 +487,10 @@ pub(super) fn write_story_content_with_fs(project_root: &Path, story_id: &str, s
/// Determine what stage a story is in (from CRDT).
pub(super) fn story_stage(story_id: &str) -> Option<String> {
crate::crdt_state::read_item(story_id).map(|item| item.stage)
crate::pipeline_state::read_typed(story_id)
.ok()
.flatten()
.map(|item| item.stage.dir_name().to_string())
}
/// Locate a work item file by searching all active pipeline stages on disk.
server/src/main.rs
+1
View File
@@ -17,6 +17,7 @@ pub mod rebuild;
mod state;
mod store;
mod workflow;
pub(crate) mod pipeline_state;
mod worktree;
use crate::agents::AgentPool;
server/src/pipeline_state.rs
+1387
View File
File diff suppressed because it is too large Load Diff