p1-1: scaffold kb-source-fs crate (FsSourceConnector)

Walk config.workspace.root, apply gitignore-style filters
(config.workspace.exclude ∪ .kbignore ∪ baked-in defaults for
.DS_Store / ._*), stream BLAKE3 over each file, and emit a
deterministic Vec<RawAsset> sorted by workspace_path.

Modules:
  - hash:      streaming blake3::Hasher + 64 KiB read buffer (no whole-file
               loads); pinned digests for empty input and "hello world".
  - media:     extension → MediaType (markdown/pdf/image/audio/other).
  - walker:    ignore::OverrideBuilder for filter union; walkdir with
               manual visited-set cycle protection on top of follow_links.
  - connector: public FsSourceConnector::new(&Config) +
               SourceConnector::scan(&SourceScope) impl. Uses
               kb_core::to_posix for WorkspacePath construction (carries
               P0-1 # rejection through unchanged) and kb_core::id_for_asset
               for AssetId derivation. Storage variant signals intent only;
               actual byte copy is P1-6's responsibility.

Per design §3.3, §6.2, §6.6, §7.1, §7.2, §8.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/kb-source-fs/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name          = "kb-source-fs"
+version       = { workspace = true }
+edition       = { workspace = true }
+rust-version  = { workspace = true }
+license       = { workspace = true }
+repository    = { workspace = true }
+description   = "Local filesystem SourceConnector — walks workspace.root + applies gitignore filters"
+
+[dependencies]
+kb-core      = { path = "../kb-core" }
+kb-config    = { path = "../kb-config" }
+anyhow       = { workspace = true }
+serde        = { workspace = true }
+time         = { workspace = true }
+blake3       = { workspace = true }
+tracing      = { workspace = true }
+walkdir      = "2"
+ignore       = "0.4"
+
+[dev-dependencies]
+serde_json   = { workspace = true }
+tempfile     = "3"

crates/kb-source-fs/src/connector.rs

@@ -0,0 +1,411 @@
+//! `FsSourceConnector` — public surface for the crate.
+//!
+//! ```ignore
+//! pub struct FsSourceConnector { /* internal */ }
+//! impl FsSourceConnector {
+//!     pub fn new(config: &kb_config::Config) -> anyhow::Result<Self>;
+//! }
+//! impl kb_core::SourceConnector for FsSourceConnector {
+//!     fn scan(&self, scope: &kb_core::SourceScope) -> anyhow::Result<Vec<kb_core::RawAsset>>;
+//! }
+//! ```
+
+use std::path::PathBuf;
+
+use anyhow::{Context, Result};
+use time::OffsetDateTime;
+
+use kb_config::Config;
+use kb_core::{
+    AssetStorage, Checksum, RawAsset, SourceConnector, SourceScope, SourceUri,
+    id_for_asset, to_posix,
+};
+
+use crate::hash::hash_file;
+use crate::media::media_type_for;
+use crate::walker::{build_overrides, read_kbignore, walk_files};
+
+/// Local-filesystem `SourceConnector`. Constructed once from `Config`,
+/// reused across `scan` calls.
+///
+/// State carried between `new` and `scan`:
+///   - `default_root`: `config.workspace.root` resolved to a `PathBuf`. Used
+///     only when `SourceScope::root` is empty (i.e. the caller did not
+///     override the root).
+///   - `default_exclude`: snapshot of `config.workspace.exclude` at
+///     construction time.
+///   - `copy_threshold_bytes`: `config.storage.copy_threshold_mb * 1 MiB`
+///     pre-multiplied so we don't recompute per file.
+pub struct FsSourceConnector {
+    default_root: PathBuf,
+    default_exclude: Vec<String>,
+    copy_threshold_bytes: u64,
+}
+
+impl FsSourceConnector {
+    pub fn new(config: &Config) -> Result<Self> {
+        // `config.workspace.root` is a String that may contain `~` or env
+        // expansions. P0-* did not yet provide a path-expansion helper in
+        // kb-config; for P1-1 we expand `~` ourselves and leave `${VAR}`
+        // for a follow-up. The vast majority of users hit the `~` case.
+        let root = expand_tilde(&config.workspace.root);
+
+        let copy_threshold_bytes = config
+            .storage
+            .copy_threshold_mb
+            .saturating_mul(1024 * 1024);
+
+        Ok(Self {
+            default_root: root,
+            default_exclude: config.workspace.exclude.clone(),
+            copy_threshold_bytes,
+        })
+    }
+}
+
+impl SourceConnector for FsSourceConnector {
+    fn scan(&self, scope: &SourceScope) -> Result<Vec<RawAsset>> {
+        // `SourceScope::root` overrides config root when non-empty. This
+        // matches the design's "scope is the per-call lens; config is the
+        // default" split (§7.1).
+        let root = if scope.root.as_os_str().is_empty() {
+            self.default_root.clone()
+        } else {
+            scope.root.clone()
+        };
+
+        // Union: config.workspace.exclude ∪ scope.exclude ∪ .kbignore.
+        // Per §6.2 the union of `.kbignore` and `config.workspace.exclude`
+        // is the filter set. `scope.exclude` is added on top so a caller
+        // can layer a per-call narrowing.
+        let mut excludes = self.default_exclude.clone();
+        excludes.extend(scope.exclude.iter().cloned());
+        let kbignore = read_kbignore(&root)?;
+
+        let overrides = build_overrides(&root, &excludes, &kbignore)?;
+
+        // `scope.include` is intentionally ignored at this stage of the
+        // pipeline: per §6.2 the workspace-level include lives in
+        // `WorkspaceCfg` and is enforced by the asset writer / extractors.
+        // Surfacing it here would double-filter Markdown vs PDF before the
+        // extractor router gets to see them.
+        if !scope.include.is_empty() {
+            tracing::debug!(
+                count = scope.include.len(),
+                "FsSourceConnector ignores scope.include — handled by extractor router"
+            );
+        }
+
+        let files = walk_files(&root, &overrides)?;
+
+        let mut assets = Vec::with_capacity(files.len());
+        for abs in &files {
+            // `to_posix` does NFC + leading `./` strip + `#` rejection.
+            // Compute the workspace-relative path before handing to it so
+            // emitted `WorkspacePath` is always relative.
+            let rel = abs.strip_prefix(&root).unwrap_or(abs);
+            let workspace_path = match to_posix(rel) {
+                Ok(p) => p,
+                Err(e) => {
+                    // A path containing `#` is the only documented reason
+                    // `to_posix` fails today. Drop the file with a warning
+                    // rather than aborting the entire scan — a single bad
+                    // filename should not nuke a 10 000-file ingest.
+                    tracing::warn!(
+                        path = %abs.display(),
+                        error = %e,
+                        "skipping file: path is not a valid WorkspacePath",
+                    );
+                    continue;
+                }
+            };
+
+            let media_type = media_type_for(abs);
+            let (byte_len, full_hex) = hash_file(abs)
+                .with_context(|| format!("hashing {}", abs.display()))?;
+            let checksum = Checksum(full_hex.clone());
+            let asset_id = id_for_asset(&full_hex);
+
+            // Storage variant signals *intent*, not an actual copy.
+            // P1-6 (asset writer) is responsible for the on-disk copy.
+            let stored = if byte_len > self.copy_threshold_bytes {
+                AssetStorage::Reference {
+                    path: abs.clone(),
+                    sha: checksum.clone(),
+                }
+            } else {
+                AssetStorage::Copied { path: abs.clone() }
+            };
+
+            assets.push(RawAsset {
+                asset_id,
+                source_uri: SourceUri::File(abs.clone()),
+                workspace_path,
+                media_type,
+                byte_len,
+                checksum,
+                discovered_at: OffsetDateTime::now_utc(),
+                stored,
+            });
+        }
+
+        // Determinism: sort by workspace_path. WorkspacePath is a String
+        // newtype with stable lexicographic ordering. Two scans of the
+        // same tree must produce identical Vec<RawAsset> modulo the
+        // wall-clock `discovered_at` field.
+        assets.sort_by(|a, b| a.workspace_path.0.cmp(&b.workspace_path.0));
+
+        Ok(assets)
+    }
+}
+
+/// Expand a leading `~` to the current user's home directory. No-op for
+/// any other shape (absolute, relative, `${VAR}`-style).
+fn expand_tilde(s: &str) -> PathBuf {
+    if let Some(rest) = s.strip_prefix("~/") {
+        if let Some(home) = dirs_home() {
+            return home.join(rest);
+        }
+    } else if s == "~" {
+        if let Some(home) = dirs_home() {
+            return home;
+        }
+    }
+    PathBuf::from(s)
+}
+
+/// Tiny `dirs::home_dir`-compat shim that does NOT add the `dirs` crate to
+/// our dep set (we explicitly enumerate allowed deps in the task spec).
+/// Reads `$HOME` directly.
+fn dirs_home() -> Option<PathBuf> {
+    std::env::var_os("HOME").map(PathBuf::from)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use kb_config::Config;
+
+    fn cfg_with_root(root: &str) -> Config {
+        let mut c = Config::defaults();
+        c.workspace.root = root.to_string();
+        c.workspace.exclude.clear();
+        c
+    }
+
+    #[test]
+    fn scan_empty_dir_yields_empty_vec() {
+        let dir = tempfile::tempdir().unwrap();
+        let conn = FsSourceConnector::new(&cfg_with_root(
+            dir.path().to_str().unwrap(),
+        ))
+        .unwrap();
+        let scope = SourceScope::default();
+        let v = conn.scan(&scope).unwrap();
+        assert!(v.is_empty());
+    }
+
+    #[test]
+    fn scan_emits_sorted_workspace_paths() {
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+        std::fs::create_dir_all(root.join("notes")).unwrap();
+        std::fs::write(root.join("README.md"), b"hi").unwrap();
+        std::fs::write(root.join("notes/beta.md"), b"b").unwrap();
+        std::fs::write(root.join("notes/alpha.md"), b"a").unwrap();
+
+        let conn =
+            FsSourceConnector::new(&cfg_with_root(root.to_str().unwrap()))
+                .unwrap();
+        let v = conn.scan(&SourceScope::default()).unwrap();
+        let names: Vec<_> = v.iter().map(|a| a.workspace_path.0.clone()).collect();
+        assert_eq!(
+            names,
+            vec![
+                "README.md".to_string(),
+                "notes/alpha.md".to_string(),
+                "notes/beta.md".to_string(),
+            ]
+        );
+    }
+
+    #[test]
+    fn scan_filters_by_kbignore() {
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+        std::fs::write(root.join(".kbignore"), "*.tmp\n").unwrap();
+        std::fs::write(root.join("a.md"), b"x").unwrap();
+        std::fs::write(root.join("b.tmp"), b"x").unwrap();
+
+        let conn =
+            FsSourceConnector::new(&cfg_with_root(root.to_str().unwrap()))
+                .unwrap();
+        let v = conn.scan(&SourceScope::default()).unwrap();
+        let names: Vec<_> = v.iter().map(|a| a.workspace_path.0.clone()).collect();
+        // .kbignore itself starts with `.` and is not in DEFAULT_EXCLUDES,
+        // so it is *not* automatically hidden — but the task spec only
+        // requires `*.tmp` and `.DS_Store` / `._*` filtering, and the
+        // `.kbignore` file is a legitimate "Other(\"\")" asset. Either
+        // present-or-absent is acceptable; the assertion below pins
+        // current behaviour: .kbignore appears, b.tmp does not.
+        assert!(names.contains(&".kbignore".to_string()));
+        assert!(names.contains(&"a.md".to_string()));
+        assert!(!names.contains(&"b.tmp".to_string()));
+    }
+
+    #[test]
+    fn scan_filters_default_excludes() {
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+        std::fs::write(root.join("a.md"), b"x").unwrap();
+        std::fs::write(root.join(".DS_Store"), b"\0\0").unwrap();
+        std::fs::write(root.join("._sidecar"), b"\0\0").unwrap();
+
+        let conn =
+            FsSourceConnector::new(&cfg_with_root(root.to_str().unwrap()))
+                .unwrap();
+        let v = conn.scan(&SourceScope::default()).unwrap();
+        let names: Vec<_> = v.iter().map(|a| a.workspace_path.0.clone()).collect();
+        assert_eq!(names, vec!["a.md".to_string()]);
+    }
+
+    #[test]
+    fn scan_unions_config_exclude_and_kbignore() {
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+        std::fs::write(root.join(".kbignore"), "*.tmp\n").unwrap();
+        std::fs::write(root.join("a.md"), b"x").unwrap();
+        std::fs::write(root.join("b.tmp"), b"x").unwrap();
+        std::fs::write(root.join("c.log"), b"x").unwrap();
+
+        let mut cfg = cfg_with_root(root.to_str().unwrap());
+        cfg.workspace.exclude.push("*.log".to_string());
+
+        let conn = FsSourceConnector::new(&cfg).unwrap();
+        let v = conn.scan(&SourceScope::default()).unwrap();
+        let names: Vec<_> = v.iter().map(|a| a.workspace_path.0.clone()).collect();
+        assert!(names.contains(&"a.md".to_string()));
+        assert!(!names.contains(&"b.tmp".to_string()), "kbignore should drop *.tmp");
+        assert!(!names.contains(&"c.log".to_string()), "config.exclude should drop *.log");
+    }
+
+    #[test]
+    fn scan_blake3_pinned_for_known_file() {
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+        std::fs::write(root.join("hello.md"), b"hello world").unwrap();
+
+        let conn =
+            FsSourceConnector::new(&cfg_with_root(root.to_str().unwrap()))
+                .unwrap();
+        let v = conn.scan(&SourceScope::default()).unwrap();
+        assert_eq!(v.len(), 1);
+        let asset = &v[0];
+        assert_eq!(
+            asset.checksum.0,
+            "d74981efa70a0c880b8d8c1985d075dbcbf679b99a5f9914e5aaf96b831a9e24"
+        );
+        assert_eq!(asset.byte_len, 11);
+        // asset_id is derived from the full hex via id_for_asset.
+        assert_eq!(asset.asset_id, id_for_asset(&asset.checksum.0));
+    }
+
+    #[test]
+    fn scan_idempotent_modulo_timestamp() {
+        // Same filesystem state → identical Vec<RawAsset> *modulo*
+        // discovered_at. Strip that field and compare.
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+        std::fs::create_dir_all(root.join("notes")).unwrap();
+        std::fs::write(root.join("notes/a.md"), b"alpha").unwrap();
+        std::fs::write(root.join("notes/b.md"), b"beta").unwrap();
+
+        let conn =
+            FsSourceConnector::new(&cfg_with_root(root.to_str().unwrap()))
+                .unwrap();
+        let v1 = conn.scan(&SourceScope::default()).unwrap();
+        let v2 = conn.scan(&SourceScope::default()).unwrap();
+        assert_eq!(v1.len(), v2.len());
+        for (a, b) in v1.iter().zip(v2.iter()) {
+            assert_eq!(a.asset_id, b.asset_id);
+            assert_eq!(a.workspace_path, b.workspace_path);
+            assert_eq!(a.checksum, b.checksum);
+            assert_eq!(a.byte_len, b.byte_len);
+            assert_eq!(a.media_type, b.media_type);
+            assert_eq!(a.source_uri, b.source_uri);
+            assert_eq!(a.stored, b.stored);
+            // discovered_at intentionally NOT compared
+        }
+    }
+
+    #[test]
+    fn scan_emits_posix_normalized_paths() {
+        // End-to-end: the connector must produce POSIX-normalized
+        // workspace paths via `kb_core::to_posix`. We can't construct an
+        // input with literal `./` / `//` segments via the filesystem (the
+        // OS won't let us), so instead we assert the resulting strings
+        // are already POSIX-clean (no leading `./`, no `//`, forward
+        // slashes only) — which is the post-conditions side of the
+        // round-trip the unit tests in `kb-core::normalize` cover.
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+        std::fs::create_dir_all(root.join("a/b/c")).unwrap();
+        std::fs::write(root.join("a/b/c/d.md"), b"x").unwrap();
+
+        let conn =
+            FsSourceConnector::new(&cfg_with_root(root.to_str().unwrap()))
+                .unwrap();
+        let v = conn.scan(&SourceScope::default()).unwrap();
+        assert_eq!(v.len(), 1);
+        let p = &v[0].workspace_path.0;
+        assert_eq!(p, "a/b/c/d.md");
+        assert!(!p.starts_with("./"));
+        assert!(!p.contains("//"));
+        assert!(!p.contains('\\'));
+    }
+
+    #[test]
+    fn scan_skips_files_whose_name_contains_hash() {
+        // `WorkspacePath` rejects `#` (collides with the W3C-Media-Fragments
+        // separator used by `Citation`). The connector must drop such
+        // files with a warning rather than aborting the scan.
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+        std::fs::write(root.join("ok.md"), b"x").unwrap();
+        std::fs::write(root.join("has#hash.md"), b"y").unwrap();
+
+        let conn =
+            FsSourceConnector::new(&cfg_with_root(root.to_str().unwrap()))
+                .unwrap();
+        let v = conn.scan(&SourceScope::default()).unwrap();
+        let names: Vec<_> = v.iter().map(|a| a.workspace_path.0.clone()).collect();
+        assert_eq!(names, vec!["ok.md".to_string()]);
+    }
+
+    #[test]
+    fn copy_vs_reference_threshold_signals_intent() {
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+        std::fs::write(root.join("small.md"), b"hi").unwrap();
+
+        let mut cfg = cfg_with_root(root.to_str().unwrap());
+        // Threshold = 0 MiB ⇒ even a 2-byte file becomes Reference.
+        cfg.storage.copy_threshold_mb = 0;
+        let conn = FsSourceConnector::new(&cfg).unwrap();
+        let v = conn.scan(&SourceScope::default()).unwrap();
+        assert_eq!(v.len(), 1);
+        match &v[0].stored {
+            AssetStorage::Reference { sha, .. } => {
+                assert_eq!(sha, &v[0].checksum);
+            }
+            other => panic!("expected Reference, got {other:?}"),
+        }
+
+        // Threshold high (default 100 MiB) ⇒ Copied.
+        let mut cfg2 = cfg_with_root(root.to_str().unwrap());
+        cfg2.storage.copy_threshold_mb = 100;
+        let conn2 = FsSourceConnector::new(&cfg2).unwrap();
+        let v2 = conn2.scan(&SourceScope::default()).unwrap();
+        assert!(matches!(v2[0].stored, AssetStorage::Copied { .. }));
+    }
+}

crates/kb-source-fs/src/hash.rs

@@ -0,0 +1,92 @@
+//! Streaming BLAKE3 over a file path. Per task spec, files MUST NOT be
+//! loaded fully into memory: `blake3::Hasher::update_reader` reads through a
+//! 64 KiB internal buffer, which keeps memory bounded for any size of file.
+//!
+//! Returns `(byte_len, full_hex)`:
+//!   - `byte_len` is the total bytes hashed (== file size after follow).
+//!   - `full_hex` is the canonical lowercase hex (64 chars) of the full
+//!     blake3 digest. The `kb-core::Checksum` invariant is "full hex"; the
+//!     32-char prefix is reserved for `AssetId` derivation via
+//!     `kb_core::id_for_asset`.
+
+use std::fs::File;
+use std::io::{self, Read};
+use std::path::Path;
+
+use anyhow::{Context, Result};
+
+const READ_BUFFER_BYTES: usize = 64 * 1024;
+
+/// Stream-hash a file with blake3. Returns `(byte_len, full_hex_64)`.
+///
+/// `byte_len` is computed during streaming so callers do not need a separate
+/// `metadata().len()` call (which can disagree with hashed bytes if the file
+/// is rewritten mid-scan, but blake3-of-stream is the source of truth for
+/// `RawAsset.checksum`).
+pub(crate) fn hash_file(path: &Path) -> Result<(u64, String)> {
+    let file = File::open(path)
+        .with_context(|| format!("failed to open {} for hashing", path.display()))?;
+    hash_reader(file).with_context(|| format!("failed to hash {}", path.display()))
+}
+
+fn hash_reader<R: Read>(mut reader: R) -> Result<(u64, String)> {
+    let mut hasher = blake3::Hasher::new();
+    let mut buf = vec![0u8; READ_BUFFER_BYTES];
+    let mut total: u64 = 0;
+    loop {
+        match reader.read(&mut buf) {
+            Ok(0) => break,
+            Ok(n) => {
+                hasher.update(&buf[..n]);
+                total = total.saturating_add(n as u64);
+            }
+            Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
+            Err(e) => return Err(e.into()),
+        }
+    }
+    Ok((total, hasher.finalize().to_hex().to_string()))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// blake3 of the empty input is the well-known "official empty hash"
+    /// from the blake3 spec. Pinned so that swapping the hash crate or the
+    /// streaming implementation can never silently produce a different
+    /// digest for a known input.
+    #[test]
+    fn empty_blake3_pinned() {
+        let (n, hex) = hash_reader(std::io::empty()).unwrap();
+        assert_eq!(n, 0);
+        assert_eq!(
+            hex,
+            "af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262"
+        );
+    }
+
+    /// `b"hello world"` blake3 (full 64 hex). Computed independently with
+    /// `b3sum`; pinning here detects any drift in the streaming pipeline.
+    #[test]
+    fn known_bytes_blake3_pinned() {
+        let bytes = b"hello world";
+        let (n, hex) = hash_reader(&bytes[..]).unwrap();
+        assert_eq!(n, 11);
+        assert_eq!(
+            hex,
+            "d74981efa70a0c880b8d8c1985d075dbcbf679b99a5f9914e5aaf96b831a9e24"
+        );
+    }
+
+    /// Streaming a buffer larger than `READ_BUFFER_BYTES` must produce the
+    /// same digest as a single-shot blake3 over the same bytes — i.e. the
+    /// chunk boundary is invisible.
+    #[test]
+    fn streaming_matches_oneshot_over_buffer_boundary() {
+        let bytes: Vec<u8> = (0u8..=255u8).cycle().take(READ_BUFFER_BYTES * 3 + 17).collect();
+        let (n, streamed) = hash_reader(&bytes[..]).unwrap();
+        assert_eq!(n, bytes.len() as u64);
+        let oneshot = blake3::hash(&bytes).to_hex().to_string();
+        assert_eq!(streamed, oneshot);
+    }
+}

crates/kb-source-fs/src/lib.rs

@@ -0,0 +1,16 @@
+//! `kb-source-fs` — local filesystem `SourceConnector`.
+//!
+//! Walks `config.workspace.root`, applies gitignore-style filters from
+//! `config.workspace.exclude` ∪ `.kbignore`, computes BLAKE3 of every file,
+//! and emits `Vec<RawAsset>` sorted by `workspace_path` for determinism.
+//!
+//! Per design §3.3 (RawAsset), §6.2 (workspace + .kbignore), §6.6 (POSIX
+//! normalization), §7.1 (SourceScope), §7.2 (SourceConnector), §8 (module
+//! boundaries).
+
+mod connector;
+mod hash;
+mod media;
+mod walker;
+
+pub use connector::FsSourceConnector;

crates/kb-source-fs/src/media.rs

@@ -0,0 +1,85 @@
+//! Media-type detection by extension. Per P1-1 task spec we do NOT do
+//! libmagic-style sniffing; extension is enough for P1. Unknown / missing
+//! extensions fall through to `MediaType::Other(ext.to_string())` (empty
+//! string when the file has no extension at all).
+
+use std::path::Path;
+
+use kb_core::{AudioType, ImageType, MediaType};
+
+/// Return `MediaType` for `path` based purely on its lowercased extension.
+/// `.md` → Markdown, `.pdf` → Pdf, image and audio extensions map onto
+/// `MediaType::Image(_)` / `MediaType::Audio(_)`. Anything else (including
+/// missing extension) → `MediaType::Other(ext)`.
+pub(crate) fn media_type_for(path: &Path) -> MediaType {
+    let ext = path
+        .extension()
+        .and_then(|s| s.to_str())
+        .map(|s| s.to_ascii_lowercase())
+        .unwrap_or_default();
+
+    match ext.as_str() {
+        "md" => MediaType::Markdown,
+        "pdf" => MediaType::Pdf,
+
+        "png" => MediaType::Image(ImageType::Png),
+        "jpg" | "jpeg" => MediaType::Image(ImageType::Jpeg),
+        "webp" => MediaType::Image(ImageType::Webp),
+        "gif" => MediaType::Image(ImageType::Gif),
+        "tiff" | "tif" => MediaType::Image(ImageType::Tiff),
+
+        "m4a" => MediaType::Audio(AudioType::M4a),
+        "mp3" => MediaType::Audio(AudioType::Mp3),
+        "wav" => MediaType::Audio(AudioType::Wav),
+        "flac" => MediaType::Audio(AudioType::Flac),
+        "ogg" => MediaType::Audio(AudioType::Ogg),
+
+        // Empty string (no extension) and any other extension: bucket as
+        // Other and let downstream extractors decide if they support it.
+        _ => MediaType::Other(ext),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn markdown_and_pdf() {
+        assert_eq!(media_type_for(Path::new("a/b.md")), MediaType::Markdown);
+        assert_eq!(media_type_for(Path::new("a/b.MD")), MediaType::Markdown);
+        assert_eq!(media_type_for(Path::new("a/b.pdf")), MediaType::Pdf);
+    }
+
+    #[test]
+    fn images_and_audio() {
+        assert_eq!(
+            media_type_for(Path::new("p.jpg")),
+            MediaType::Image(ImageType::Jpeg)
+        );
+        assert_eq!(
+            media_type_for(Path::new("p.JPEG")),
+            MediaType::Image(ImageType::Jpeg)
+        );
+        assert_eq!(
+            media_type_for(Path::new("a.M4A")),
+            MediaType::Audio(AudioType::M4a)
+        );
+        assert_eq!(
+            media_type_for(Path::new("a.flac")),
+            MediaType::Audio(AudioType::Flac)
+        );
+    }
+
+    #[test]
+    fn unknown_and_missing_extension() {
+        assert_eq!(
+            media_type_for(Path::new("notes/x.weird")),
+            MediaType::Other("weird".to_string())
+        );
+        assert_eq!(
+            media_type_for(Path::new("README")),
+            MediaType::Other(String::new())
+        );
+    }
+}

crates/kb-source-fs/src/walker.rs

@@ -0,0 +1,247 @@
+//! Directory walker with gitignore-style filtering and symlink-cycle
+//! protection.
+//!
+//! Filter set (per task spec, design §6.2):
+//!   - `config.workspace.exclude` (passed in by `FsSourceConnector`)
+//!   - `<root>/.kbignore` (optional file at workspace root)
+//!   - default-excludes for `.DS_Store` and macOS resource forks (`._*`)
+//!
+//! All three are merged via `ignore::overrides::OverrideBuilder`, which
+//! gives full gitignore semantics (anchors, `!` negation, `**`, etc.). We
+//! prepend `!` to each pattern because `OverrideBuilder` treats positive
+//! patterns as "include" and negative as "exclude" — see §"Filter set"
+//! comment in `build_walker` for the full reasoning.
+//!
+//! Symlink handling: we want to follow links (so a workspace using a
+//! symlinked `notes/` directory works), but we must NOT loop forever on
+//! `a -> b -> a`. `walkdir` does NOT detect cycles for us when
+//! `follow_links(true)`; we layer our own visited-set on top, keyed by the
+//! canonical path of every entry, and skip any entry we've already seen.
+
+use std::collections::HashSet;
+use std::path::{Path, PathBuf};
+
+use anyhow::{Context, Result};
+use ignore::overrides::{Override, OverrideBuilder};
+use walkdir::{DirEntry, WalkDir};
+
+/// Default-excludes baked into the connector. These are NOT configurable;
+/// they cover noise that is never useful to ingest and would otherwise need
+/// to appear in every user's `.kbignore`.
+const DEFAULT_EXCLUDES: &[&str] = &[
+    // Finder metadata
+    ".DS_Store",
+    "**/.DS_Store",
+    // macOS resource forks (AppleDouble files)
+    "._*",
+    "**/._*",
+];
+
+/// Build the merged `Override` from `config.workspace.exclude` ∪ `.kbignore`
+/// ∪ baked-in default excludes.
+///
+/// Each input pattern is registered as an *exclude* (gitignore-style: a
+/// leading `!` flips a positive match to a negative one in the
+/// `OverrideBuilder` API). Order doesn't matter — the union is computed by
+/// the underlying gitignore engine.
+pub(crate) fn build_overrides(
+    root: &Path,
+    config_exclude: &[String],
+    kbignore_patterns: &[String],
+) -> Result<Override> {
+    let mut builder = OverrideBuilder::new(root);
+
+    for pat in DEFAULT_EXCLUDES {
+        builder
+            .add(&format!("!{pat}"))
+            .with_context(|| format!("invalid default-exclude pattern: {pat}"))?;
+    }
+    for pat in config_exclude {
+        builder
+            .add(&format!("!{pat}"))
+            .with_context(|| format!("invalid workspace.exclude pattern: {pat}"))?;
+    }
+    for pat in kbignore_patterns {
+        builder
+            .add(&format!("!{pat}"))
+            .with_context(|| format!("invalid .kbignore pattern: {pat}"))?;
+    }
+
+    builder.build().context("failed to compile override set")
+}
+
+/// Read `<root>/.kbignore` if it exists. Each non-blank, non-comment line is
+/// a gitignore pattern. Missing file → empty Vec (not an error).
+pub(crate) fn read_kbignore(root: &Path) -> Result<Vec<String>> {
+    let path = root.join(".kbignore");
+    if !path.exists() {
+        return Ok(Vec::new());
+    }
+    let text = std::fs::read_to_string(&path)
+        .with_context(|| format!("failed to read {}", path.display()))?;
+    Ok(text
+        .lines()
+        .map(|l| l.trim())
+        .filter(|l| !l.is_empty() && !l.starts_with('#'))
+        .map(|l| l.to_string())
+        .collect())
+}
+
+/// Iterate every regular file under `root`, applying `overrides` and
+/// detecting symlink cycles. Returns absolute file paths.
+///
+/// Strategy:
+///   - `walkdir::WalkDir::follow_links(true)` to traverse symlinks.
+///   - Maintain `visited: HashSet<PathBuf>` of *canonical* paths. Before
+///     descending into a directory entry, canonicalize and check the set;
+///     if already present, skip. This breaks `a -> b -> a` cycles in O(n)
+///     per entry without a custom recursive walker.
+///   - For each yielded entry, ask `overrides` whether it is excluded; if
+///     so, drop it. If the entry is a directory, also short-circuit
+///     `WalkDir`'s descent via `it.skip_current_dir()`.
+pub(crate) fn walk_files(root: &Path, overrides: &Override) -> Result<Vec<PathBuf>> {
+    let mut out = Vec::new();
+    let mut visited: HashSet<PathBuf> = HashSet::new();
+
+    let walker = WalkDir::new(root).follow_links(true).into_iter();
+    let mut it = walker.filter_entry(|e| !is_excluded(e, root, overrides));
+
+    while let Some(res) = it.next() {
+        let entry = match res {
+            Ok(e) => e,
+            Err(err) => {
+                // `walkdir` surfaces I/O errors AND its own cycle detector
+                // (when follow_links is on it sometimes catches them).
+                // Either way: log and skip; do not abort the whole scan.
+                tracing::warn!(error = %err, "walkdir entry error; skipping");
+                continue;
+            }
+        };
+
+        let path = entry.path();
+
+        // Cycle guard: only canonicalize symlinks (cheap on the common case
+        // of plain files/dirs) and on directories that are followed via a
+        // symlink. `walkdir`'s `path_is_symlink()` is true when the entry's
+        // *original* path is a symlink (it returns true for the link, not
+        // for the resolved target). For non-symlinked directories we still
+        // record the canonical path so a *later* symlink that points back
+        // to one of them is detected.
+        if entry.file_type().is_dir() {
+            match std::fs::canonicalize(path) {
+                Ok(canon) => {
+                    if !visited.insert(canon) {
+                        // Already visited via another path → break cycle.
+                        it.skip_current_dir();
+                        continue;
+                    }
+                }
+                Err(_) => {
+                    // Broken symlink etc. — skip silently.
+                    continue;
+                }
+            }
+        }
+
+        if entry.file_type().is_file() {
+            out.push(path.to_path_buf());
+        }
+    }
+
+    Ok(out)
+}
+
+fn is_excluded(entry: &DirEntry, root: &Path, overrides: &Override) -> bool {
+    // `Override::matched(path, is_dir)` uses the path *relative to* the
+    // override builder's root. `walkdir` gives absolute paths when
+    // `WalkDir::new` was given an absolute path — strip the root prefix
+    // before consulting the override.
+    let rel = match entry.path().strip_prefix(root) {
+        Ok(p) => p,
+        Err(_) => entry.path(),
+    };
+    overrides
+        .matched(rel, entry.file_type().is_dir())
+        .is_ignore()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn empty_inputs_compile_into_an_override() {
+        let dir = tempfile::tempdir().unwrap();
+        let ov = build_overrides(dir.path(), &[], &[]).unwrap();
+        // Default-excludes only; non-special files should not match.
+        let m = ov.matched(Path::new("notes/alpha.md"), false);
+        assert!(!m.is_ignore());
+    }
+
+    #[test]
+    fn default_excludes_ds_store_and_resource_forks() {
+        let dir = tempfile::tempdir().unwrap();
+        let ov = build_overrides(dir.path(), &[], &[]).unwrap();
+        assert!(ov.matched(Path::new(".DS_Store"), false).is_ignore());
+        assert!(
+            ov.matched(Path::new("notes/.DS_Store"), false).is_ignore()
+        );
+        assert!(ov.matched(Path::new("._foo.md"), false).is_ignore());
+        assert!(
+            ov.matched(Path::new("notes/._sidecar"), false).is_ignore()
+        );
+    }
+
+    #[test]
+    fn config_exclude_filters_tmp_and_node_modules() {
+        let dir = tempfile::tempdir().unwrap();
+        let ov = build_overrides(
+            dir.path(),
+            &["*.tmp".to_string(), "node_modules/**".to_string()],
+            &[],
+        )
+        .unwrap();
+        assert!(ov.matched(Path::new("a.tmp"), false).is_ignore());
+        assert!(ov.matched(Path::new("notes/x.tmp"), false).is_ignore());
+        assert!(
+            ov.matched(Path::new("node_modules/foo/bar.js"), false)
+                .is_ignore()
+        );
+        assert!(!ov.matched(Path::new("alpha.md"), false).is_ignore());
+    }
+
+    #[test]
+    fn kbignore_union_with_config_exclude() {
+        // "either set excluding it ⇒ excluded"
+        let dir = tempfile::tempdir().unwrap();
+        let ov = build_overrides(
+            dir.path(),
+            &["*.tmp".to_string()],
+            &["secret/**".to_string()],
+        )
+        .unwrap();
+        assert!(ov.matched(Path::new("a.tmp"), false).is_ignore());
+        assert!(
+            ov.matched(Path::new("secret/key.md"), false).is_ignore()
+        );
+    }
+
+    #[test]
+    fn read_kbignore_missing_returns_empty() {
+        let dir = tempfile::tempdir().unwrap();
+        let v = read_kbignore(dir.path()).unwrap();
+        assert!(v.is_empty());
+    }
+
+    #[test]
+    fn read_kbignore_strips_blanks_and_comments() {
+        let dir = tempfile::tempdir().unwrap();
+        std::fs::write(
+            dir.path().join(".kbignore"),
+            "# comment\n*.tmp\n\nignored/**\n",
+        )
+        .unwrap();
+        let v = read_kbignore(dir.path()).unwrap();
+        assert_eq!(v, vec!["*.tmp".to_string(), "ignored/**".to_string()]);
+    }
+}