feat(tasks): answer "what's next" by priority, ten at a time, and say when there is more

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)

crates/cli/src/main.rs

@@ -3147,15 +3147,15 @@ fn build_mcp_server() -> anyhow::Result<forge_mcp_server::McpServer> {
     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": {
                 "status":       {"type": "string",  "description": "Filter by status: triage, ready, in_progress, blocked, complete, archived"},
                 "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<forge_mcp_server::McpServer> {
                 .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<forge_mcp_server::McpServer> {
             // 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<FerrosaMemoryConfig> {
+pub(crate) fn ferrosa_memory_config() -> Option<FerrosaMemoryConfig> {
     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<std::path::PathBuf>,
     config: &Option<FerrosaMemoryConfig>,
 ) -> anyhow::Result<ResolvedTransport> {

crates/fmem-client/src/tools/ingest_skill.rs

@@ -53,6 +53,13 @@ pub struct IngestSkillArgs {
     pub completion_criteria: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub content_hash: Option<String>,
+    /// 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<String>,
     /// 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<IngestSkillResponse, Error>
 #[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(),

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<Section>,
 }
@@ -84,6 +87,14 @@ fn parse_corpus_file(path: &Path) -> Result<Option<CorpusDoc>> {
         .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<Option<CorpusDoc>> {
         publisher,
         category,
         rel_path,
+        source_path,
         summary_section,
         sections,
     }))
@@ -204,6 +216,7 @@ fn build_doc_graph(doc: &CorpusDoc) -> (Vec<Entity>, Vec<Edge>) {
         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<Entity>, Vec<Edge>) {
         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<Entity>, Vec<Edge>) {
             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(),

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<String>,
+
     /// Full file body for `file` entities; `None` for all other entity types.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub source_text: Option<String>,
@@ -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(),

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

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");

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(),

crates/ingest/src/skill_ingest/parse.rs

@@ -43,6 +43,12 @@ pub struct Skill {
     pub supplementary_files: Vec<String>,
     /// Parsed steps from the body.
     pub steps: Vec<Step>,
+    /// 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<String>,
     /// Raw frontmatter YAML bytes, captured for hashing.
     pub frontmatter_bytes: Vec<u8>,
     /// 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<Skill, ParseError> {
+    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<String>,
+) -> Result<Skill, ParseError> {
     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<Skill, ParseError> {
 
     Ok(Skill {
         name,
+        source_path,
         category: category.to_string(),
         description,
         argument_hint: fm.argument_hint.map(ArgumentHint::into_string),