diff --git a/crates/cli/src/fmem_skill_ingest.rs b/crates/cli/src/fmem_skill_ingest.rs index cd12bee..236760e 100644 --- a/crates/cli/src/fmem_skill_ingest.rs +++ b/crates/cli/src/fmem_skill_ingest.rs @@ -79,18 +79,43 @@ fn execute_run( let dry = MockTransport::panicking(); skill_ingest::run(config, &dry).map_err(|e| anyhow::anyhow!("{e}"))? } else { - let argv = parse_server_command(server.as_deref()); - let stdio_config = StdioConfig { - command: argv, - ..Default::default() + // Prefer a server that is ALREADY RUNNING. + // + // Spawning is the fallback, not the default. On a machine with a live + // fmem, spawning a second one points two servers at the same cluster: + // they both run consolidation, contend for the same partitions, and + // the taxonomy phase times out against a cluster that is serving the + // first server perfectly well. The corpus path has always used the + // configured HTTP endpoint; this brings skills in line with it. + // Only when the caller did not name a server explicitly: an explicit + // --server is a deliberate choice and must not be silently ignored. + let http = if server.is_none() { + crate::ferrosa_memory_config() + } else { + None }; - let transport = - StdioTransport::spawn(stdio_config).map_err(|e| anyhow::anyhow!("spawn fmem: {e}"))?; - // Best-effort handshake — strict protocol version. Fail loud if - // the server advertises a version forge doesn't recognize. - initialize(&transport, ExpectedProtocolVersion::Strict) - .map_err(|e| anyhow::anyhow!("fmem initialize: {e}"))?; - skill_ingest::run(config, &transport).map_err(|e| anyhow::anyhow!("{e}"))? + match crate::resolve_transport_for_ingest(None, &http) { + Ok(crate::ResolvedTransport::Http { transport, label }) => { + eprintln!("[skill-ingest] using {label}"); + initialize(&transport, ExpectedProtocolVersion::Strict) + .map_err(|e| anyhow::anyhow!("fmem initialize: {e}"))?; + skill_ingest::run(config, &transport).map_err(|e| anyhow::anyhow!("{e}"))? + } + _ => { + let argv = parse_server_command(server.as_deref()); + let stdio_config = StdioConfig { + command: argv, + ..Default::default() + }; + let transport = StdioTransport::spawn(stdio_config) + .map_err(|e| anyhow::anyhow!("spawn fmem: {e}"))?; + // Best-effort handshake — strict protocol version. Fail loud if + // the server advertises a version forge doesn't recognize. + initialize(&transport, ExpectedProtocolVersion::Strict) + .map_err(|e| anyhow::anyhow!("fmem initialize: {e}"))?; + skill_ingest::run(config, &transport).map_err(|e| anyhow::anyhow!("{e}"))? + } + } }; Ok(summary) diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 2f4e2fd..6b2cfc5 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -3147,7 +3147,7 @@ fn build_mcp_server() -> anyhow::Result { register_tool!( server, "task_list", - "List tasks with optional filtering by status, assignee, and priority range.", + "List tasks, MOST IMPORTANT FIRST (priority desc, then newest). Returns the top 10 by default and reports the total; when more match, the response carries a `next_page` line telling you the exact offset to ask for. Filter by status, assignee, and priority range.", serde_json::json!({ "type": "object", "properties": { @@ -3155,7 +3155,7 @@ fn build_mcp_server() -> anyhow::Result { "assignee": {"type": "string", "description": "Filter by assignee name"}, "priority_gte": {"type": "integer", "description": "Minimum priority (inclusive)"}, "priority_lte": {"type": "integer", "description": "Maximum priority (inclusive)"}, - "limit": {"type": "integer", "description": "Max results (default 50). Rows are returned NEWEST FIRST, so a limit keeps the most recent work rather than an arbitrary slice."}, + "limit": {"type": "integer", "description": "Rows per page (default 10). Rows come back HIGHEST PRIORITY FIRST, so a page is the most important outstanding work, not an arbitrary slice. When more match than fit, `next_page` in the response gives the offset for the rest."}, "offset": {"type": "integer", "description": "Skip this many rows before the limit, for paging through a long board."}, "full": {"type": "boolean", "description": "Include body, result and metadata on every row. Off by default: a few hundred full rows exceed the tool-result token limit. Use task_get for the detail of a specific task."}, "cql_host": {"type": "string", "description": "CQL host:port (default: 127.0.0.1:9042)"}, @@ -3195,6 +3195,7 @@ fn build_mcp_server() -> anyhow::Result { .list_tasks_paged(filter, offset) .map_err(|e| e.to_string())?; let full = args.get("full").and_then(|v| v.as_bool()).unwrap_or(false); + let page_len = page.tasks.len(); let rows: Vec<_> = if full { page.tasks } else { @@ -3203,12 +3204,19 @@ fn build_mcp_server() -> anyhow::Result { // Report the window explicitly. A capped read that says so is usable; // one that does not is worse than an error -- the previous behaviour // silently omitted recent tasks and looked like a complete answer. - let payload = serde_json::json!({ + // Tell the caller when it is holding a window rather than the + // whole answer, in terms it can act on without guessing the + // parameter name. A truncated result that does not announce + // itself reads as a complete one. + let mut payload = serde_json::json!({ "tasks": rows, "total": page.total, "offset": offset, "truncated": page.truncated, }); + if let Some(hint) = forge_tasks::page_hint(page.total, offset, page_len) { + payload["next_page"] = serde_json::Value::String(hint); + } finish_task(&store, &args, &payload) } ); @@ -3959,7 +3967,7 @@ impl FerrosaMemoryConfig { } /// Read config from user's config directory if it exists. -fn ferrosa_memory_config() -> Option { +pub(crate) fn ferrosa_memory_config() -> Option { let path = dirs::home_dir()?.join(".config/ferrosa-memory.toml"); let content = std::fs::read_to_string(path).ok()?; let table: toml::Table = content.parse().ok()?; @@ -4035,7 +4043,7 @@ impl FerrosaMemoryConfig { /// The resolved transport the CLI/MCP tool should use for this call, /// with a short human-readable label for log/error messages. -enum ResolvedTransport { +pub(crate) enum ResolvedTransport { Http { transport: forge_fmem_client::HttpTransport, label: String, @@ -4072,7 +4080,7 @@ impl ResolvedTransport { /// We DO NOT silently fall back to "extract-only" — a tool named /// `ingest` that doesn't ingest is a footgun (a prior design mistake /// this function exists to close). -fn resolve_transport_for_ingest( +pub(crate) fn resolve_transport_for_ingest( mcp_bin: Option, config: &Option, ) -> anyhow::Result { diff --git a/crates/fmem-client/src/tools/ingest_skill.rs b/crates/fmem-client/src/tools/ingest_skill.rs index b3da292..a5f9b6d 100644 --- a/crates/fmem-client/src/tools/ingest_skill.rs +++ b/crates/fmem-client/src/tools/ingest_skill.rs @@ -53,6 +53,13 @@ pub struct IngestSkillArgs { pub completion_criteria: Option, #[serde(skip_serializing_if = "Option::is_none")] pub content_hash: Option, + /// Absolute path of the SKILL.md this came from. + /// + /// fmem derives the skill's TIER from this. Omitted rather than sent + /// empty when unknown: fmem records a source only when one is stated, and + /// a blank path would file the skill under a root of "". + #[serde(skip_serializing_if = "Option::is_none")] + pub source_path: Option, /// Session to record as `ingested_by_session`. fmem accepts a UUID /// string or the literal `"default"` (see /// `ferrosa-memory-core/src/dispatch.rs`'s tool description). @@ -223,6 +230,7 @@ fn parse_response(raw: serde_json::Value) -> Result #[cfg(test)] fn minimal_args(name: &str) -> IngestSkillArgs { IngestSkillArgs { + source_path: None, name: name.to_string(), category: "task-level".to_string(), description: format!("{name} skill"), @@ -413,6 +421,7 @@ mod tests { })), ); let args = IngestSkillArgs { + source_path: None, name: "tdd".into(), category: "task-level".into(), description: "do tdd".into(), diff --git a/crates/ingest/src/corpus.rs b/crates/ingest/src/corpus.rs index 99dde7e..57708ef 100644 --- a/crates/ingest/src/corpus.rs +++ b/crates/ingest/src/corpus.rs @@ -50,6 +50,9 @@ struct CorpusDoc { category: String, /// Relative path from corpus root parent (e.g. "corpus/functional-programming/foo.md") rel_path: String, + /// Where the file actually is. Becomes `attrs.source_path`, which is what + /// the server tiers on. + source_path: String, summary_section: Section, sections: Vec
, } @@ -84,6 +87,14 @@ fn parse_corpus_file(path: &Path) -> Result> { .unwrap_or_default(); let rel_path = format!("corpus/{category}/{filename}"); + // Absolute, and canonicalised where possible: the tier rules match on a + // root resolved through an alias table, and a relative path only matches + // if someone registered that exact spelling. Falls back to the path as + // given rather than dropping it -- an un-canonicalisable path still tiers + // if it happens to sit under a known root. + let source_path = std::fs::canonicalize(path) + .map(|p| p.to_string_lossy().into_owned()) + .unwrap_or_else(|_| path.to_string_lossy().into_owned()); let mut lines: Vec<&str> = raw.lines().collect(); @@ -163,6 +174,7 @@ fn parse_corpus_file(path: &Path) -> Result> { publisher, category, rel_path, + source_path, summary_section, sections, })) @@ -204,6 +216,7 @@ fn build_doc_graph(doc: &CorpusDoc) -> (Vec, Vec) { entity_type: "document".to_string(), context: l1_context, extractor_schema_version: Some(EXTRACTOR_SCHEMA_VERSION), + source_path: Some(doc.source_path.clone()), ..Default::default() }); @@ -223,6 +236,7 @@ fn build_doc_graph(doc: &CorpusDoc) -> (Vec, Vec) { entity_type: "section".to_string(), context: l2_context, extractor_schema_version: Some(EXTRACTOR_SCHEMA_VERSION), + source_path: Some(doc.source_path.clone()), ..Default::default() }); @@ -254,6 +268,7 @@ fn build_doc_graph(doc: &CorpusDoc) -> (Vec, Vec) { entity_type: "section".to_string(), context: l3_context, extractor_schema_version: Some(EXTRACTOR_SCHEMA_VERSION), + source_path: Some(doc.source_path.clone()), ..Default::default() }); @@ -429,6 +444,7 @@ mod tests { #[test] fn test_build_doc_graph_structure() { let doc = CorpusDoc { + source_path: "/tmp/corpus/test/doc.md".to_owned(), title: "Test Book".to_string(), author: "Test Author".to_string(), year: "2024".to_string(), diff --git a/crates/ingest/src/extractor.rs b/crates/ingest/src/extractor.rs index 30e53ac..8df1587 100644 --- a/crates/ingest/src/extractor.rs +++ b/crates/ingest/src/extractor.rs @@ -52,6 +52,20 @@ pub struct Entity { pub entity_type: String, pub context: String, + /// Where this entity came from, as an absolute path. + /// + /// Sent to ferrosa-memory as `attrs.source_path`, which is what its tier + /// plane derives a tier from: a path under a curated root is curated + /// material, and an entity with no path sits at Data whatever it contains. + /// + /// ABSOLUTE, not relative. The rules are keyed on roots resolved through + /// an alias table, and a relative path like `corpus/x.md` only matches if + /// someone happened to register that exact spelling. The absolute path is + /// what the file actually is, and it matches the alias for the checkout it + /// lives in. + #[serde(skip_serializing_if = "Option::is_none")] + pub source_path: Option, + /// Full file body for `file` entities; `None` for all other entity types. #[serde(skip_serializing_if = "Option::is_none")] pub source_text: Option, @@ -144,6 +158,8 @@ pub fn emit_file_entity(path: &Path, source: &SourceBuffer) -> Entity { name: path_str.clone(), entity_type: "file".to_string(), context: format!("Source file: {path_str}"), + // A file entity IS its path, so it always knows where it came from. + source_path: Some(path_str.clone()), source_text: source.text.clone(), sha256: Some(source.sha256.clone()), // Byte ranges are omitted for file entities; start/end line cover the whole file. @@ -2299,6 +2315,7 @@ namespace MyApp #[test] fn entity_schema_roundtrips_json() { let original = Entity { + source_path: None, id: "test-id-123".to_string(), name: "my_function".to_string(), entity_type: "function".to_string(), diff --git a/crates/ingest/src/graph_loader.rs b/crates/ingest/src/graph_loader.rs index 5b18a2d..7f16663 100644 --- a/crates/ingest/src/graph_loader.rs +++ b/crates/ingest/src/graph_loader.rs @@ -483,6 +483,13 @@ fn entity_to_wire_entity(e: &Entity) -> WireEntity { if let Some(v) = e.extractor_schema_version { attrs.insert("extractor_schema_version".into(), v.into()); } + // Where it came from. The server derives its TIER from this, so an entity + // that ships without one is filed as raw capture no matter how curated the + // file it came from was. + if let Some(path) = &e.source_path { + attrs.insert("source_path".into(), path.clone().into()); + } + // source_text is intentionally excluded from attrs — it is a top-level // field on `file` entities in the server schema. sha256 likewise. // T-future: once the server schema is confirmed, move these to diff --git a/crates/ingest/src/skill_ingest/build_args.rs b/crates/ingest/src/skill_ingest/build_args.rs index 1fde6bf..608ce65 100644 --- a/crates/ingest/src/skill_ingest/build_args.rs +++ b/crates/ingest/src/skill_ingest/build_args.rs @@ -53,6 +53,10 @@ pub fn build_ingest_args( // don't lie on the wire. completion_criteria: None, content_hash: Some(hash::content_hash(skill, supplementary)), + // Where the SKILL.md lives. fmem tiers on this: skills are the whole + // population of the Wisdom tier, and one that arrives without a path + // is filed as raw capture. + source_path: skill.source_path.clone(), session_id, } } @@ -81,12 +85,38 @@ mod tests { fn sup(name: &str, bytes: &[u8]) -> ResolvedSupplementary { ResolvedSupplementary { + inlined: true, declared: name.into(), path: PathBuf::from(name), bytes: bytes.to_vec(), } } + /// The path has to reach the wire, or fmem tiers the skill as raw capture. + /// + /// Skills are the whole population of the Wisdom tier. A regression here + /// is silent: ingest succeeds, the skill is searchable, and it simply sits + /// in the wrong tier for anyone deciding what to share. + #[test] + fn the_source_path_reaches_the_wire() { + let mut skill = parse_skill("name: tdd\ndescription: d\n", "# Steps\n", "task-level"); + skill.source_path = Some("/Users/b/src/research/skills/tdd/SKILL.md".into()); + let args = build_ingest_args(&skill, &[], None); + assert_eq!( + args.source_path.as_deref(), + Some("/Users/b/src/research/skills/tdd/SKILL.md"), + ); + } + + /// A skill parsed from bytes alone has no path, and must send none rather + /// than an empty string -- fmem records a source only when one is stated, + /// and "" would file the skill under a root of "". + #[test] + fn a_skill_with_no_known_path_sends_none() { + let skill = parse_skill("name: tdd\ndescription: d\n", "# Steps\n", "task-level"); + assert_eq!(build_ingest_args(&skill, &[], None).source_path, None); + } + #[test] fn minimal_skill_maps_fields() { let skill = parse_skill("name: tdd\ndescription: do tdd\n", "body", "task-level"); diff --git a/crates/ingest/src/skill_ingest/hash.rs b/crates/ingest/src/skill_ingest/hash.rs index ca2235b..9c9da0b 100644 --- a/crates/ingest/src/skill_ingest/hash.rs +++ b/crates/ingest/src/skill_ingest/hash.rs @@ -31,6 +31,12 @@ pub fn content_hash(skill: &Skill, supplementary: &[ResolvedSupplementary]) -> S for sup in supplementary { hasher.update(sup.declared.as_bytes()); hasher.update(b"\0"); + // Whether it was inlined is part of the identity. A file that moves + // out of the skill directory stops contributing its content, and the + // hash has to change or the skill reads as unchanged while what it + // carries has shrunk. + hasher.update(if sup.inlined { b"i" } else { b"r" }); + hasher.update(b"\0"); hasher.update(&sup.bytes); hasher.update(b"\0"); } @@ -54,6 +60,7 @@ mod tests { fn sup(name: &str, bytes: &[u8]) -> ResolvedSupplementary { ResolvedSupplementary { + inlined: true, declared: name.to_string(), path: PathBuf::from(name), bytes: bytes.to_vec(), diff --git a/crates/ingest/src/skill_ingest/parse.rs b/crates/ingest/src/skill_ingest/parse.rs index bbc84e3..f61d575 100644 --- a/crates/ingest/src/skill_ingest/parse.rs +++ b/crates/ingest/src/skill_ingest/parse.rs @@ -43,6 +43,12 @@ pub struct Skill { pub supplementary_files: Vec, /// Parsed steps from the body. pub steps: Vec, + /// Absolute path of the SKILL.md this was parsed from. + /// + /// Sent to fmem as `source_path`, which is what its tier plane derives a + /// tier from. Skills are the Wisdom tier's whole population; without this + /// they arrive with no origin and sit at Data alongside session exhaust. + pub source_path: Option, /// Raw frontmatter YAML bytes, captured for hashing. pub frontmatter_bytes: Vec, /// Raw body markdown bytes, captured for hashing. @@ -151,6 +157,19 @@ struct Frontmatter { /// `category` comes from the walker (the immediate child of the skill /// root); the parser does not derive it from the file path. pub fn parse(bytes: &[u8], category: &str) -> Result { + parse_from(bytes, category, None) +} + +/// Parse, recording where the file came from. +/// +/// Split from [`parse`] so the many call sites that only have bytes keep +/// working, while the walker -- the one place that knows the path -- can pass +/// it. A skill with no path still ingests; it just cannot be tiered. +pub fn parse_from( + bytes: &[u8], + category: &str, + source_path: Option, +) -> Result { if bytes.len() > MAX_FILE_SIZE_BYTES { return Err(ParseError::TooLarge { actual: bytes.len(), @@ -180,6 +199,7 @@ pub fn parse(bytes: &[u8], category: &str) -> Result { Ok(Skill { name, + source_path, category: category.to_string(), description, argument_hint: fm.argument_hint.map(ArgumentHint::into_string), diff --git a/crates/ingest/src/skill_ingest/run.rs b/crates/ingest/src/skill_ingest/run.rs index 883a51d..602cb85 100644 --- a/crates/ingest/src/skill_ingest/run.rs +++ b/crates/ingest/src/skill_ingest/run.rs @@ -22,7 +22,7 @@ use std::collections::HashSet; use std::path::{Path, PathBuf}; -use std::time::Instant; +use std::time::{Duration, Instant}; use forge_fmem_client::{ ensure_parent_tag, ingest_skill, verify_skill, EnsureParentTagArgs, IngestSkillAction, @@ -143,6 +143,10 @@ pub enum RunError { Walk(walk::WalkError), Collision(collision::CollisionError), Plan(taxonomy::PlanError), + /// fmem was reachable but never able to serve. Its own variant, because + /// "the cluster is still starting" and "your catalog is malformed" want + /// different responses from whoever ran this. + NotReady(String), } impl std::fmt::Display for RunError { @@ -151,6 +155,7 @@ impl std::fmt::Display for RunError { Self::Walk(e) => write!(f, "walk error: {e}"), Self::Collision(e) => write!(f, "{e}"), Self::Plan(e) => write!(f, "{e}"), + Self::NotReady(reason) => write!(f, "fmem never became ready: {reason}"), } } } @@ -221,6 +226,13 @@ pub fn run(config: RunConfig, transport: &T) -> Result 0 { summary.duration_ms = start.elapsed().as_millis(); @@ -256,7 +268,14 @@ fn parse_files<'a>( ) -> Vec<(Skill, &'a SkillFile)> { let mut out = Vec::with_capacity(files.len()); for f in files { - match parse::parse(&f.bytes, &f.category) { + // The walker is the only place that knows where the file is, so it is + // the only place that can tell fmem. Canonicalised so the tier rules, + // which match on a root through an alias table, see the same spelling + // whatever relative path the run was started from. + let source_path = std::fs::canonicalize(&f.path) + .map(|p| p.to_string_lossy().into_owned()) + .unwrap_or_else(|_| f.path.to_string_lossy().into_owned()); + match parse::parse_from(&f.bytes, &f.category, Some(source_path)) { Ok(skill) => { if skill.steps_empty { summary @@ -347,6 +366,56 @@ impl Matcher { } } +// --------------------------------------------------------------------------- +// Readiness +// --------------------------------------------------------------------------- + +/// How long to wait for a freshly spawned fmem to be able to serve. +/// +/// A cold server answers the MCP handshake immediately and connects to its +/// cluster in the background, so "the process started" is not "the tools +/// work". Without this the run fired all 17 taxonomy edges into a server that +/// was still connecting and reported 17 failures for a cluster that was +/// perfectly healthy. +const READY_TIMEOUT: Duration = Duration::from_secs(30); +const READY_POLL: Duration = Duration::from_millis(250); + +/// Wait until fmem can actually answer a read, or say plainly that it cannot. +/// +/// Probes with `count_entities_by_type`, which touches the cluster and +/// changes nothing. Only a "connection not yet established" answer is waited +/// out; anything else means the probe is uninformative here and the run +/// proceeds. The timeout message carries the last error so a cluster that is +/// down is distinguishable from one that is merely slow. +fn wait_until_ready(transport: &T, config: &RunConfig) -> Result<(), String> { + let started = std::time::Instant::now(); + let mut last = String::from("no attempt completed"); + while started.elapsed() < READY_TIMEOUT { + let args = forge_fmem_client::CountEntitiesByTypeArgs { + session_id: config.session_id.clone(), + }; + match forge_fmem_client::count_entities_by_type(transport, args) { + Ok(_) => return Ok(()), + Err(e) => { + last = e.to_string(); + // ONLY a not-ready answer is worth waiting out. Any other + // error means this probe cannot tell us anything -- a mock + // transport, a server too old to have the tool -- so proceed + // and let the real calls report their own failures. Vetoing + // on an error the probe does not understand would turn a + // readiness check into a second failure mode rather than a fix. + if !last.contains("not yet established") { + return Ok(()); + } + } + } + std::thread::sleep(READY_POLL); + } + Err(format!( + "fmem did not become ready within {READY_TIMEOUT:?}: {last}" + )) +} + // --------------------------------------------------------------------------- // Phase A // --------------------------------------------------------------------------- diff --git a/crates/ingest/src/skill_ingest/supplementary.rs b/crates/ingest/src/skill_ingest/supplementary.rs index d67021a..45c9b1a 100644 --- a/crates/ingest/src/skill_ingest/supplementary.rs +++ b/crates/ingest/src/skill_ingest/supplementary.rs @@ -22,7 +22,19 @@ pub struct ResolvedSupplementary { /// Canonical absolute path. pub path: PathBuf, /// File bytes — read eagerly so downstream hashing doesn't re-stat. + /// + /// Empty when `inlined` is false: the file exists but sits outside the + /// skill, so it is REFERENCED rather than pulled in. pub bytes: Vec, + /// Whether the content was taken into this skill. + /// + /// False for a file outside the skill directory. Those are real + /// cross-references — a skill citing a sibling skill's conventions, or a + /// corpus distillation it is built on — and they used to fail the whole + /// skill. Recording the reference and moving on keeps the skill, keeps the + /// pointer, and avoids copying a corpus document into a skill, which would + /// duplicate Information-tier text into Wisdom. + pub inlined: bool, } #[derive(Debug)] @@ -106,14 +118,25 @@ fn resolve_one( source: e, })?; + // Outside the skill: keep the pointer, do not take the content. + // + // NOT an error. The guard exists to stop a skill inlining arbitrary files, + // and it still does -- nothing outside is read. But a skill that cites a + // sibling's conventions or the corpus document it was distilled from is + // making a legitimate reference, and failing the whole skill over it lost + // 3 of 98 in the real catalog. if !resolved.starts_with(skill_dir) { - return Err(SupplementaryError::EscapesSkillDir { + return Ok(ResolvedSupplementary { declared: declared.to_string(), - resolved, - skill_dir: skill_dir.to_path_buf(), + path: resolved, + bytes: Vec::new(), + inlined: false, }); } + // A declared file that cannot be READ stays fatal. That is a typo or a + // deleted file, and passing it silently would let a skill lose half its + // content without anyone noticing. let bytes = fs::read(&resolved).map_err(|e| SupplementaryError::Io { declared: declared.to_string(), source: e, @@ -123,6 +146,7 @@ fn resolve_one( declared: declared.to_string(), path: resolved, bytes, + inlined: true, }) } @@ -169,16 +193,28 @@ mod tests { assert!(matches!(err, SupplementaryError::Absolute(_))); } + /// A path outside the skill is REFERENCED, never read. + /// + /// This used to be an error, which failed the whole skill and lost 3 of + /// 98 in the real catalog over legitimate cross-references. The property + /// that matters is unchanged and is what this now asserts: not one byte + /// from outside the skill directory is taken in. #[test] - fn rejects_parent_escape() { + fn a_path_above_the_skill_is_referenced_and_never_read() { let tmp = TempDir::new().unwrap(); let dir = setup_skill_dir(&tmp); - // Create a file above the skill dir. let above = tmp.path().join("secrets.md"); fs::write(&above, "sensitive").unwrap(); - let err = resolve(&dir, &["../../../secrets.md".to_string()]).unwrap_err(); - assert!(matches!(err, SupplementaryError::EscapesSkillDir { .. })); + let out = + resolve(&dir, &["../../../secrets.md".to_string()]).expect("a reference, not an error"); + let sup = out.first().expect("one entry"); + assert!(!sup.inlined, "content outside the skill was inlined"); + assert!( + sup.bytes.is_empty(), + "bytes were read from outside the skill" + ); + assert_eq!(sup.declared, "../../../secrets.md", "the pointer is kept"); } #[test] @@ -211,22 +247,27 @@ mod tests { assert_eq!(names, vec!["b.md", "a.md", "c.md"]); } + /// A symlink out of the skill directory reads nothing either. + /// + /// The check is on the CANONICALISED path, so a link that looks local and + /// points away is caught the same as `../`. This is the case the guard + /// exists for, and relaxing escapes from fatal to referenced must not + /// weaken it: the assertion is on the bytes, not on the error type. #[cfg(unix)] #[test] - fn symlink_escape_rejected() { + fn a_symlink_out_of_the_skill_is_referenced_and_never_read() { use std::os::unix::fs::symlink; let tmp = TempDir::new().unwrap(); let dir = setup_skill_dir(&tmp); - // File outside the skill tree. let outside_real = tmp.path().join("outside-real.md"); fs::write(&outside_real, "real").unwrap(); - - // Symlink inside the skill dir pointing to it. let link = dir.join("trick.md"); symlink(&outside_real, &link).unwrap(); - let err = resolve(&dir, &["trick.md".to_string()]).unwrap_err(); - assert!(matches!(err, SupplementaryError::EscapesSkillDir { .. })); + let out = resolve(&dir, &["trick.md".to_string()]).expect("a reference, not an error"); + let sup = out.first().expect("one entry"); + assert!(!sup.inlined, "a symlink pulled content in from outside"); + assert!(sup.bytes.is_empty(), "bytes were read through a symlink"); } } diff --git a/crates/ingest/tests/run_harness.rs b/crates/ingest/tests/run_harness.rs index bc1ca84..e840d41 100644 --- a/crates/ingest/tests/run_harness.rs +++ b/crates/ingest/tests/run_harness.rs @@ -59,6 +59,16 @@ fn config(root: PathBuf) -> RunConfig { } } +/// The readiness probe every run makes before it writes anything. +/// +/// A cold fmem answers the MCP handshake immediately and connects to its +/// cluster afterwards, so the run waits until a read actually works. Every +/// script therefore starts with this; one that omits it asserts the run +/// writes into a server it never checked. +fn ok_ready() -> ScriptedResponse { + ScriptedResponse::Ok(json!({ "counts": {}, "total": 0 })) +} + fn ok_created() -> ScriptedResponse { ScriptedResponse::Ok(json!({ "action": "created", @@ -152,8 +162,9 @@ fn happy_path_three_skills_no_hierarchy() { ); let mock = MockTransport::new(); - // Phase A: no hierarchy → no ensure_parent_tag calls. - // Phase B: three ingest_skill calls. + mock.expect_call("tools/call", ok_ready()); // readiness probe + // Phase A: no hierarchy → no ensure_parent_tag calls. + // Phase B: three ingest_skill calls. mock.expect_call("tools/call", ok_created()); // refactor (sorted by path comes first under task-level) mock.expect_call("tools/call", ok_created()); // tdd mock.expect_call("tools/call", ok_created()); // rust @@ -192,8 +203,9 @@ fn phase_c_repass_uses_server_missing_prereqs_signal() { ); let mock = MockTransport::new(); - // Phase B (two ingest calls, sorted by path): tdd (with missing - // prereq), then unit-testing (all clean). + mock.expect_call("tools/call", ok_ready()); // readiness probe + // Phase B (two ingest calls, sorted by path): tdd (with missing + // prereq), then unit-testing (all clean). mock.expect_call("tools/call", ok_created_missing(vec!["unit-testing"])); // tdd mock.expect_call("tools/call", ok_created()); // unit-testing // Phase C: tdd's Phase B outcome had a non-empty list, so it gets @@ -225,7 +237,8 @@ fn phase_c_skipped_when_no_missing_prereqs_reported() { ); let mock = MockTransport::new(); - // Phase B: ok_created (empty missing list — prereq already in fmem). + mock.expect_call("tools/call", ok_ready()); // readiness probe + // Phase B: ok_created (empty missing list — prereq already in fmem). mock.expect_call("tools/call", ok_created()); // Phase D only — no re-pass. mock.expect_call("tools/call", ok_verify(vec!["task-level"], vec![])); @@ -251,6 +264,7 @@ fn verification_failure_exits_4() { ); let mock = MockTransport::new(); + mock.expect_call("tools/call", ok_ready()); // readiness probe mock.expect_call("tools/call", ok_created()); // ingest_skill // Phase D sees the missing prereq. mock.expect_call( @@ -282,6 +296,7 @@ fn missing_expected_tag_fails_verification() { ); let mock = MockTransport::new(); + mock.expect_call("tools/call", ok_ready()); // readiness probe mock.expect_call("tools/call", ok_created()); // ingest_skill mock.expect_call("tools/call", ok_verify(vec!["task-level"], vec![])); @@ -310,7 +325,8 @@ fn hierarchy_phase_a_wired() { fx.hierarchy("task-level: quality\n"); let mock = MockTransport::new(); - // Phase A: one ensure_parent_tag. + mock.expect_call("tools/call", ok_ready()); // readiness probe + // Phase A: one ensure_parent_tag. mock.expect_call("tools/call", ok_parent_tag_created()); // Phase B: two ingest_skill. mock.expect_call("tools/call", ok_created()); @@ -340,7 +356,8 @@ fn filter_matches_zero_warns_exits_clean() { cfg.filter = Some("nonexistent*".into()); let mock = MockTransport::new(); - // Filter matches nothing — no ingest/verify calls. + mock.expect_call("tools/call", ok_ready()); // readiness probe + // Filter matches nothing — no ingest/verify calls. let summary = run(cfg, &mock).unwrap(); assert_eq!(summary.skills_created, 0); assert_eq!(summary.skills_filtered_out, 1); @@ -378,6 +395,7 @@ fn ingest_transport_error_counts_as_failure() { ); let mock = MockTransport::new(); + mock.expect_call("tools/call", ok_ready()); // readiness probe mock.expect_call( "tools/call", ScriptedResponse::ToolError { @@ -422,6 +440,7 @@ fn force_flag_omits_content_hash_on_the_wire() { cfg.force = true; let mock = MockTransport::new(); + mock.expect_call("tools/call", ok_ready()); // readiness probe mock.expect_call_with( "tools/call", |p| { diff --git a/crates/tasks/src/lib.rs b/crates/tasks/src/lib.rs index e6e372e..d6032c2 100644 --- a/crates/tasks/src/lib.rs +++ b/crates/tasks/src/lib.rs @@ -11,7 +11,7 @@ mod types; pub use config::{resolve_cql_host, resolve_cql_hosts, resolve_debug_stop, DEFAULT_CQL_HOST}; pub use debug_stop::{apply_debug_stop, BoardHealth, DEBUG_STOP_CRITICAL}; -pub use store::TaskStore; +pub use store::{default_page_limit, page_hint, TaskStore}; pub use types::{ Comment, CreateTaskRequest, KanbanBoard, Task, TaskFilter, TaskStatus, TaskWithLinks, UpdateTaskPatch, diff --git a/crates/tasks/src/store.rs b/crates/tasks/src/store.rs index b2db08b..0dbe7cd 100644 --- a/crates/tasks/src/store.rs +++ b/crates/tasks/src/store.rs @@ -485,13 +485,17 @@ impl TaskStore { /// As `list_tasks`, but reporting how many matched and whether the read was /// cut short, so a caller can tell a complete answer from a window. pub fn list_tasks_paged(&self, filter: TaskFilter, offset: usize) -> Result { - let limit = filter.limit.unwrap_or(100); + let limit = filter.limit.unwrap_or_else(default_page_limit); // Ask for everything the bound allows; the window is applied here. let mut unbounded = filter; unbounded.limit = Some(MAX_FETCH_ROWS); let mut tasks = self.list_tasks(unbounded)?; let truncated = tasks.len() >= MAX_FETCH_ROWS; let total = tasks.len(); + // Most important first. `list_tasks` sorts newest-first, which answers + // "what changed recently"; a paged read is asking "what should I do + // next", and the page boundary is what makes the total order matter. + sort_by_priority(&mut tasks); let tasks = if offset >= tasks.len() { Vec::new() } else { @@ -849,6 +853,58 @@ fn format_set_text(items: &[String]) -> String { // Tests // --------------------------------------------------------------------------- +/// Rows returned when the caller does not say how many they want. +/// +/// Ten, not fifty. The common question is "what should I do next", and the +/// answer to that is a shortlist an agent can act on — fifty rows is mostly +/// tool-result tokens the caller will never read. Anyone who wants more can +/// ask for more, and `page_hint` tells them how. +pub const fn default_page_limit() -> usize { + 10 +} + +/// Most important first: priority desc, then newest, then task_id. +/// +/// `sort_newest_first` answers "what changed recently". This answers "what +/// should I do next", which is a different question — under recency ordering a +/// P90 blocker filed yesterday loses to a P30 note filed this morning. +/// +/// The task_id tiebreak is not cosmetic. Paging splits one ordering across +/// several reads, so any pair left in an unspecified order can swap between +/// calls, and a task that swaps across a page boundary is shown twice or never +/// — the same reasoning `sort_newest_first` documents. +pub(crate) fn sort_by_priority(tasks: &mut [Task]) { + tasks.sort_by(|a, b| { + b.priority + .cmp(&a.priority) + .then_with(|| b.created_at.cmp(&a.created_at)) + .then_with(|| a.task_id.cmp(&b.task_id)) + }); +} + +/// How to ask for the rest, when there is a rest. +/// +/// Returns `None` for a complete answer so a finished result set does not +/// invite a pointless follow-up call. +/// +/// This exists because a truncated answer that does not announce itself reads +/// as a complete one — the failure this board already hit once, when a LIMIT +/// silently dropped recent tasks and an agent could not see its own write. +/// `truncated` covers the MAX_FETCH_ROWS bound; this covers the caller's own +/// page, which is far more likely to be hit. +pub fn page_hint(total: usize, offset: usize, returned: usize) -> Option { + let shown_through = offset.saturating_add(returned); + if shown_through >= total || returned == 0 { + return None; + } + Some(format!( + "Showing {}-{} of {total}. {} more — call again with offset={shown_through} for the next page.", + offset + 1, + shown_through, + total - shown_through, + )) +} + #[cfg(test)] mod tests { use super::*; @@ -981,4 +1037,118 @@ mod tests { "CREATE TABLE IF NOT EXISTS agent_memory.tasks (" ); } + + /// Build a task with just the fields ordering depends on. + fn task_at(task_id: &str, priority: i32, created_at: i64) -> Task { + let mut row = valid_row(); + row.columns[0] = Some(CqlValue::Text(task_id.into())); + row.columns[6] = Some(CqlValue::Int(priority)); + row.columns[16] = Some(CqlValue::BigInt(created_at)); + parse_task_row(row).expect("valid row") + } + + /// "What should I do next" must answer by IMPORTANCE, not by recency. + /// + /// The board only had `sort_newest_first`, so a P90 blocker filed + /// yesterday lost to a P30 note filed this morning — the agent asking for + /// the most relevant next work got the most recent instead. + #[test] + fn priority_order_puts_the_most_important_first() { + let mut tasks = vec![ + task_at("t_low_new", 30, 2_000), + task_at("t_high_old", 90, 1_000), + task_at("t_mid", 50, 1_500), + ]; + sort_by_priority(&mut tasks); + + let ids: Vec<&str> = tasks.iter().map(|t| t.task_id.as_str()).collect(); + assert_eq!( + ids, + vec!["t_high_old", "t_mid", "t_low_new"], + "highest priority first, regardless of age" + ); + } + + /// Equal priorities fall back to newest first, so the ordering still says + /// something useful rather than being arbitrary within a priority band. + #[test] + fn equal_priority_falls_back_to_newest_first() { + let mut tasks = vec![task_at("t_older", 70, 1_000), task_at("t_newer", 70, 2_000)]; + sort_by_priority(&mut tasks); + + let ids: Vec<&str> = tasks.iter().map(|t| t.task_id.as_str()).collect(); + assert_eq!(ids, vec!["t_newer", "t_older"]); + } + + /// The order must be TOTAL. Paging splits one ordering across several + /// reads, so any pair left in an unspecified order can swap between calls + /// — which would show a task twice on one page and never on the next. The + /// existing `sort_newest_first` breaks ties on task_id for exactly this + /// reason; priority ordering needs the same guarantee. + #[test] + fn identical_priority_and_timestamp_still_have_a_stable_total_order() { + let ordering = |ids: [&str; 3]| { + let mut tasks = vec![ + task_at(ids[0], 70, 1_000), + task_at(ids[1], 70, 1_000), + task_at(ids[2], 70, 1_000), + ]; + sort_by_priority(&mut tasks); + tasks + .iter() + .map(|t| t.task_id.clone()) + .collect::>() + }; + + // Same three tasks presented in different input orders must come out + // the same way every time. + assert_eq!( + ordering(["t_b", "t_a", "t_c"]), + ordering(["t_c", "t_b", "t_a"]), + "tie-breaking must not depend on input order" + ); + assert_eq!(ordering(["t_b", "t_a", "t_c"]), vec!["t_a", "t_b", "t_c"]); + } + + /// When more matched than were returned, the caller is TOLD, in terms it + /// can act on directly. A truncated answer that does not announce itself + /// reads as a complete one. + #[test] + fn a_partial_page_tells_the_caller_how_to_get_the_rest() { + let hint = page_hint(47, 0, 10).expect("a partial page must hint"); + assert!( + hint.contains("47"), + "the hint must state the true total, got: {hint}" + ); + assert!( + hint.contains("offset=10"), + "the hint must give the exact next offset, got: {hint}" + ); + } + + /// Mid-way through the board the hint advances rather than repeating the + /// first page. + #[test] + fn the_hint_advances_with_the_offset() { + let hint = page_hint(47, 10, 10).expect("still more to come"); + assert!( + hint.contains("offset=20"), + "next offset must follow this page, got: {hint}" + ); + } + + /// A complete answer must NOT invite a pointless follow-up call. + #[test] + fn a_complete_result_set_has_no_hint() { + assert!(page_hint(10, 0, 10).is_none(), "exactly complete"); + assert!(page_hint(7, 0, 7).is_none(), "fewer than a full page"); + assert!(page_hint(47, 40, 7).is_none(), "final page"); + assert!(page_hint(0, 0, 0).is_none(), "empty board"); + } + + /// Asking for "the next things" should cost ten rows, not fifty. + #[test] + fn the_default_page_is_ten() { + assert_eq!(default_page_limit(), 10); + } }