huskies: merge 1204 story Slim CRDT-backed pipeline_query MCP tool (project / stages / fields / include_archived)

server/src/http/mcp/dispatch.rs

@@ -79,6 +79,7 @@ pub async fn dispatch_tool_call(
         "launch_qa_app" => qa_tools::tool_launch_qa_app(&args, ctx).await,
         // Pipeline status
         "get_pipeline_status" => story_tools::tool_get_pipeline_status(ctx),
+        "pipeline_query" => story_tools::tool_pipeline_query(&args),
         // Diagnostics
         "get_server_logs" => diagnostics::tool_get_server_logs(&args),
         "get_version" => diagnostics::tool_get_version(ctx),

server/src/http/mcp/story_tools/mod.rs

@@ -74,6 +74,7 @@ pub(crate) use refactor::{tool_create_refactor, tool_list_refactors};
 pub(crate) use spike::tool_create_spike;
 pub(crate) use story::{
     tool_accept_story, tool_convert_item_type, tool_create_story, tool_delete_story,
-    tool_freeze_story, tool_get_pipeline_status, tool_list_upcoming, tool_purge_story,
+    tool_freeze_story, tool_get_pipeline_status, tool_list_upcoming, tool_pipeline_query,
-    tool_unblock_story, tool_unfreeze_story, tool_update_story, tool_validate_stories,
+    tool_purge_story, tool_unblock_story, tool_unfreeze_story, tool_update_story,
+    tool_validate_stories,
 };

server/src/http/mcp/story_tools/story/mod.rs

@@ -11,5 +11,7 @@ pub(crate) use convert::tool_convert_item_type;
 pub(crate) use create::{tool_create_story, tool_purge_story};
 pub(crate) use delete::{tool_accept_story, tool_delete_story};
 pub(crate) use freeze::{tool_freeze_story, tool_unfreeze_story};
-pub(crate) use query::{tool_get_pipeline_status, tool_list_upcoming, tool_validate_stories};
+pub(crate) use query::{
+    tool_get_pipeline_status, tool_list_upcoming, tool_pipeline_query, tool_validate_stories,
+};
 pub(crate) use update::{tool_unblock_story, tool_update_story};

server/src/http/mcp/story_tools/story/query.rs

@@ -106,6 +106,78 @@ pub(crate) fn tool_get_pipeline_status(ctx: &AppContext) -> Result<String, Strin
     .map_err(|e| format!("Serialization error: {e}"))
 }
 
+/// Valid `stage` filter values for [`tool_pipeline_query`], derived from the
+/// [`crate::pipeline_state::Pipeline`] wire-format strings so the error
+/// message and filter logic can never drift from the enum.
+const PIPELINE_QUERY_VALID_STAGES: &[&str] = &[
+    "backlog", "coding", "qa", "merge", "done", "closed", "archived",
+];
+
+/// Field names [`tool_pipeline_query`] returns when the caller omits `fields`.
+const PIPELINE_QUERY_DEFAULT_FIELDS: &[&str] = &["story_id", "name", "stage"];
+
+/// Slim, filterable projection of pipeline items straight from the CRDT
+/// typed projection ([`crate::pipeline_state::read_all_typed`]) — never
+/// surfaces markdown-migration rows that aren't live CRDT state. Supports
+/// filtering by `stage`, an opt-in `include_archived` flag (archived items
+/// are excluded by default), and a `fields` allowlist so callers can keep
+/// responses far smaller than `get_pipeline_status`.
+pub(crate) fn tool_pipeline_query(args: &Value) -> Result<String, String> {
+    use crate::pipeline_state::Pipeline;
+
+    let stage_filter = args.get("stage").and_then(|v| v.as_str());
+    if let Some(s) = stage_filter
+        && !PIPELINE_QUERY_VALID_STAGES.contains(&s)
+    {
+        return Err(format!(
+            "Unknown stage '{s}'. Valid stages: {}",
+            PIPELINE_QUERY_VALID_STAGES.join(", ")
+        ));
+    }
+
+    let include_archived = args
+        .get("include_archived")
+        .and_then(|v| v.as_bool())
+        .unwrap_or(false);
+
+    let fields: Vec<&str> = match args.get("fields").and_then(|v| v.as_array()) {
+        Some(arr) => arr.iter().filter_map(|v| v.as_str()).collect(),
+        None => PIPELINE_QUERY_DEFAULT_FIELDS.to_vec(),
+    };
+
+    fn slim_name(name: &str) -> &str {
+        crate::chat::util::truncate_at_char_boundary(name, 120)
+    }
+
+    let items: Vec<Value> = crate::pipeline_state::read_all_typed()
+        .into_iter()
+        .filter(|item| include_archived || item.stage.pipeline() != Pipeline::Archived)
+        .filter(|item| stage_filter.is_none_or(|s| item.stage.pipeline().as_str() == s))
+        .map(|item| {
+            let mut obj = serde_json::Map::new();
+            for field in &fields {
+                let value = match *field {
+                    "story_id" => json!(item.story_id.0),
+                    "name" => json!(slim_name(&item.name)),
+                    "stage" => json!(item.stage.pipeline().as_str()),
+                    "status" => json!(item.stage.status().as_str()),
+                    "depends_on" => json!(
+                        item.depends_on
+                            .iter()
+                            .map(|d| d.0.as_str())
+                            .collect::<Vec<_>>()
+                    ),
+                    _ => continue,
+                };
+                obj.insert((*field).to_string(), value);
+            }
+            Value::Object(obj)
+        })
+        .collect();
+
+    serde_json::to_string_pretty(&items).map_err(|e| format!("Serialization error: {e}"))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -352,4 +424,202 @@ mod tests {
             "nameless items must be invisible to tool_validate_stories"
         );
     }
+
+    #[test]
+    fn pipeline_query_filters_by_stage() {
+        crate::db::ensure_content_store();
+        crate::db::write_item_with_content(
+            "9960_story_pq_current",
+            "2_current",
+            "---\nname: \"PQ Current\"\n---\n",
+            crate::db::ItemMeta::named("PQ Current"),
+        );
+        crate::db::write_item_with_content(
+            "9961_story_pq_qa",
+            "3_qa",
+            "---\nname: \"PQ QA\"\n---\n",
+            crate::db::ItemMeta::named("PQ QA"),
+        );
+
+        let result = tool_pipeline_query(&json!({"stage": "coding"})).unwrap();
+        let parsed: Vec<Value> = serde_json::from_str(&result).unwrap();
+        assert!(
+            parsed
+                .iter()
+                .any(|i| i["story_id"] == "9960_story_pq_current")
+        );
+        assert!(
+            parsed.iter().all(|i| i["story_id"] != "9961_story_pq_qa"),
+            "stage=coding must exclude qa items: {parsed:?}"
+        );
+    }
+
+    #[test]
+    fn pipeline_query_excludes_archived_unless_opted_in() {
+        crate::db::ensure_content_store();
+        crate::db::write_item_with_content(
+            "9962_story_pq_archived",
+            "6_archived",
+            "---\nname: \"PQ Archived\"\n---\n",
+            crate::db::ItemMeta::named("PQ Archived"),
+        );
+
+        let default_result = tool_pipeline_query(&json!({})).unwrap();
+        let default_parsed: Vec<Value> = serde_json::from_str(&default_result).unwrap();
+        assert!(
+            default_parsed
+                .iter()
+                .all(|i| i["story_id"] != "9962_story_pq_archived"),
+            "archived items must be excluded by default: {default_parsed:?}"
+        );
+
+        let opted_in_result = tool_pipeline_query(&json!({"include_archived": true})).unwrap();
+        let opted_in_parsed: Vec<Value> = serde_json::from_str(&opted_in_result).unwrap();
+        assert!(
+            opted_in_parsed
+                .iter()
+                .any(|i| i["story_id"] == "9962_story_pq_archived"),
+            "include_archived=true must surface archived items: {opted_in_parsed:?}"
+        );
+    }
+
+    #[test]
+    fn pipeline_query_projects_only_requested_fields() {
+        crate::db::ensure_content_store();
+        crate::db::write_item_with_content(
+            "9963_story_pq_fields",
+            "2_current",
+            "---\nname: \"PQ Fields\"\n---\n",
+            crate::db::ItemMeta::named("PQ Fields"),
+        );
+
+        // Default slim field set.
+        let result = tool_pipeline_query(&json!({"stage": "coding"})).unwrap();
+        let parsed: Vec<Value> = serde_json::from_str(&result).unwrap();
+        let item = parsed
+            .iter()
+            .find(|i| i["story_id"] == "9963_story_pq_fields")
+            .expect("expected item in default field projection");
+        let obj = item.as_object().unwrap();
+        assert_eq!(
+            obj.keys()
+                .map(String::as_str)
+                .collect::<std::collections::BTreeSet<_>>(),
+            ["story_id", "name", "stage"].into_iter().collect()
+        );
+
+        // Explicit narrower field set.
+        let result =
+            tool_pipeline_query(&json!({"stage": "coding", "fields": ["story_id", "status"]}))
+                .unwrap();
+        let parsed: Vec<Value> = serde_json::from_str(&result).unwrap();
+        let item = parsed
+            .iter()
+            .find(|i| i["story_id"] == "9963_story_pq_fields")
+            .expect("expected item in narrow field projection");
+        let obj = item.as_object().unwrap();
+        assert_eq!(
+            obj.keys()
+                .map(String::as_str)
+                .collect::<std::collections::BTreeSet<_>>(),
+            ["story_id", "status"].into_iter().collect()
+        );
+    }
+
+    #[test]
+    fn pipeline_query_caps_item_names() {
+        crate::db::ensure_content_store();
+        let long_name = "N".repeat(500);
+        crate::db::write_item_with_content(
+            "9964_story_pq_longname",
+            "2_current",
+            &format!("---\nname: \"{long_name}\"\n---\n"),
+            crate::db::ItemMeta::named(&long_name),
+        );
+
+        let result = tool_pipeline_query(&json!({"stage": "coding"})).unwrap();
+        let parsed: Vec<Value> = serde_json::from_str(&result).unwrap();
+        let item = parsed
+            .iter()
+            .find(|i| i["story_id"] == "9964_story_pq_longname")
+            .expect("expected long-named item");
+        assert!(
+            item["name"].as_str().unwrap().len() <= 120,
+            "name must be capped at 120 bytes: {} bytes",
+            item["name"].as_str().unwrap().len()
+        );
+    }
+
+    #[test]
+    fn pipeline_query_unknown_stage_lists_valid_stages() {
+        let err = tool_pipeline_query(&json!({"stage": "not-a-real-stage"})).unwrap_err();
+        assert!(err.contains("not-a-real-stage"), "{err}");
+        for stage in PIPELINE_QUERY_VALID_STAGES {
+            assert!(err.contains(stage), "error should list '{stage}': {err}");
+        }
+    }
+
+    #[test]
+    fn pipeline_query_zero_matches_returns_empty_array_not_error() {
+        let result = tool_pipeline_query(&json!({"stage": "done", "fields": ["story_id"]}));
+        assert!(result.is_ok(), "zero matches must not be an error");
+        let parsed: Vec<Value> = serde_json::from_str(&result.unwrap()).unwrap();
+        assert!(parsed.iter().all(|i| i["story_id"] != "no-such-story"));
+    }
+
+    #[test]
+    fn pipeline_query_omits_ghost_items_not_in_crdt() {
+        crate::db::ensure_content_store();
+        crate::db::write_item_with_content(
+            "9965_story_pq_real",
+            "2_current",
+            "---\nname: \"PQ Real\"\n---\n",
+            crate::db::ItemMeta::named("PQ Real"),
+        );
+
+        let result = tool_pipeline_query(&json!({})).unwrap();
+        let parsed: Vec<Value> = serde_json::from_str(&result).unwrap();
+        assert!(
+            parsed.iter().any(|i| i["story_id"] == "9965_story_pq_real"),
+            "real CRDT item must appear"
+        );
+        // A story_id that was never written via the CRDT (no filesystem
+        // markdown-scan path exists in pipeline_query) must never appear —
+        // this is the regression guard for the ghost-item bug class.
+        assert!(
+            parsed
+                .iter()
+                .all(|i| i["story_id"] != "9999_story_never_written_ghost"),
+            "ghost item not present in CRDT must not appear: {parsed:?}"
+        );
+    }
+
+    #[test]
+    fn pipeline_query_response_is_smaller_than_get_pipeline_status() {
+        crate::db::ensure_content_store();
+        let tmp = tempfile::tempdir().unwrap();
+        for i in 0..10 {
+            let id = format!("997{i}0_story_pq_size");
+            let name = "A Reasonably Named Story For Size Comparison";
+            crate::db::write_item_with_content(
+                &id,
+                "2_current",
+                &format!("---\nname: \"{name}\"\n---\n"),
+                crate::db::ItemMeta {
+                    name: Some(name.to_string()),
+                    ..Default::default()
+                },
+            );
+        }
+        let ctx = test_ctx(tmp.path());
+        let full = tool_get_pipeline_status(&ctx).unwrap();
+        let slim = tool_pipeline_query(&json!({})).unwrap();
+        assert!(
+            slim.len() < full.len(),
+            "slim pipeline_query response ({} bytes) must be smaller than \
+             get_pipeline_status ({} bytes)",
+            slim.len(),
+            full.len()
+        );
+    }
 }

server/src/http/mcp/tools_list/mod.rs

@@ -78,6 +78,7 @@ mod tests {
         assert!(names.contains(&"get_server_logs"));
         assert!(names.contains(&"prompt_permission"));
         assert!(names.contains(&"get_pipeline_status"));
+        assert!(names.contains(&"pipeline_query"));
         assert!(names.contains(&"get_token_usage"));
         assert!(names.contains(&"move_story"));
         assert!(names.contains(&"unblock_story"));
@@ -117,7 +118,7 @@ mod tests {
         assert!(names.contains(&"edit"));
         assert!(names.contains(&"write"));
         assert!(names.contains(&"gc"));
-        assert_eq!(tools.len(), 85);
+        assert_eq!(tools.len(), 86);
     }
 
     #[test]

server/src/http/mcp/tools_list/story_tools.rs

@@ -610,6 +610,28 @@ pub(super) fn story_tools() -> Vec<Value> {
                 "properties": {}
             }
         }),
+        json!({
+            "name": "pipeline_query",
+            "description": "Slim, filterable query over the CRDT-backed pipeline. Unlike get_pipeline_status (a full snapshot), this returns only the requested fields for items matching an optional stage filter, keeping responses far smaller. Items are always projected from live CRDT state, never markdown-migration rows. Archived items are excluded unless include_archived=true. An unknown stage value returns an error listing valid stages; a filter with no matches returns an empty array.",
+            "inputSchema": {
+                "type": "object",
+                "properties": {
+                    "stage": {
+                        "type": "string",
+                        "description": "Filter to items in this stage. One of: backlog, coding, qa, merge, done, closed, archived. Omit to return items across all stages (archived still excluded unless include_archived is true)."
+                    },
+                    "include_archived": {
+                        "type": "boolean",
+                        "description": "Include archived items in results. Default false: archived items are excluded unless explicitly opted in."
+                    },
+                    "fields": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Which fields to include per item. One or more of: story_id, name, stage, status, depends_on. Defaults to [\"story_id\", \"name\", \"stage\"] when omitted."
+                    }
+                }
+            }
+        }),
         json!({
             "name": "delete_story",
             "description": "Delete a work item from the pipeline entirely. Stops any running agent, removes the worktree, and deletes the story file. Use only for removing obsolete or duplicate items.",