diff --git a/Cargo.lock b/Cargo.lock
index ef1d040..479945c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -238,6 +238,27 @@ dependencies = [
"crypto-common 0.2.2",
]
+[[package]]
+name = "dirs"
+version = "5.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "44c45a9d03d6676652bcb5e724c7e988de1acad23a711b5217ab9cbecbec2225"
+dependencies = [
+ "dirs-sys",
+]
+
+[[package]]
+name = "dirs-sys"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "520f05a5cbd335fae5a99ff7a6ab8627577660ee5cfd6a94a6a929b52ff0321c"
+dependencies = [
+ "libc",
+ "option-ext",
+ "redox_users",
+ "windows-sys 0.48.0",
+]
+
[[package]]
name = "displaydoc"
version = "0.2.6"
@@ -511,6 +532,12 @@ version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c"
+[[package]]
+name = "hex"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
+
[[package]]
name = "http"
version = "1.4.2"
@@ -820,6 +847,15 @@ dependencies = [
"pkg-config",
]
+[[package]]
+name = "libredox"
+version = "0.1.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2026a5056764a10b2bf5d56488cba40da507f5493a6a429340e2004d9ed085fa"
+dependencies = [
+ "libc",
+]
+
[[package]]
name = "libsqlite3-sys"
version = "0.38.1"
@@ -918,6 +954,12 @@ version = "1.21.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
+[[package]]
+name = "option-ext"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d"
+
[[package]]
name = "parking_lot"
version = "0.12.5"
@@ -991,7 +1033,7 @@ dependencies = [
"rustc-hash",
"rustls",
"socket2",
- "thiserror",
+ "thiserror 2.0.18",
"tokio",
"tracing",
"web-time",
@@ -1013,7 +1055,7 @@ dependencies = [
"rustls",
"rustls-pki-types",
"slab",
- "thiserror",
+ "thiserror 2.0.18",
"tinyvec",
"tracing",
"web-time",
@@ -1083,6 +1125,17 @@ dependencies = [
"bitflags",
]
+[[package]]
+name = "redox_users"
+version = "0.4.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43"
+dependencies = [
+ "getrandom 0.2.17",
+ "libredox",
+ "thiserror 1.0.69",
+]
+
[[package]]
name = "ref-cast"
version = "1.0.25"
@@ -1194,7 +1247,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c51c9ae4df8a7fba42103df5c621fa3c37eccf3a3c650879e90fc48b11cc192c"
dependencies = [
"hashbrown 0.16.1",
- "thiserror",
+ "thiserror 2.0.18",
]
[[package]]
@@ -1421,6 +1474,16 @@ version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba"
+[[package]]
+name = "signal-hook-registry"
+version = "1.4.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b"
+dependencies = [
+ "errno",
+ "libc",
+]
+
[[package]]
name = "slab"
version = "0.4.12"
@@ -1522,13 +1585,33 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "thiserror"
+version = "1.0.69"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
+dependencies = [
+ "thiserror-impl 1.0.69",
+]
+
[[package]]
name = "thiserror"
version = "2.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4"
dependencies = [
- "thiserror-impl",
+ "thiserror-impl 2.0.18",
+]
+
+[[package]]
+name = "thiserror-impl"
+version = "1.0.69"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
]
[[package]]
@@ -1554,7 +1637,7 @@ dependencies = [
"serde",
"serde_json",
"sha2 0.11.0",
- "thiserror",
+ "thiserror 2.0.18",
"tokio",
"tracing",
]
@@ -1566,9 +1649,11 @@ dependencies = [
"anyhow",
"async-trait",
"chrono",
+ "dirs",
"dotenvy",
"futures",
"git2",
+ "hex",
"log",
"parking_lot",
"rand",
@@ -1580,7 +1665,7 @@ dependencies = [
"serde_json",
"sha2 0.10.9",
"tempfile",
- "thiserror",
+ "thiserror 2.0.18",
"tinyagents",
"tinycortex-api",
"tokio",
@@ -1601,7 +1686,7 @@ dependencies = [
"serde",
"serde_json",
"sha2 0.10.9",
- "thiserror",
+ "thiserror 2.0.18",
"uuid",
]
@@ -1640,6 +1725,7 @@ dependencies = [
"libc",
"mio",
"pin-project-lite",
+ "signal-hook-registry",
"socket2",
"tokio-macros",
"windows-sys 0.61.2",
@@ -2052,13 +2138,22 @@ dependencies = [
"windows-link",
]
+[[package]]
+name = "windows-sys"
+version = "0.48.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
+dependencies = [
+ "windows-targets 0.48.5",
+]
+
[[package]]
name = "windows-sys"
version = "0.52.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
dependencies = [
- "windows-targets",
+ "windows-targets 0.52.6",
]
[[package]]
@@ -2070,34 +2165,67 @@ dependencies = [
"windows-link",
]
+[[package]]
+name = "windows-targets"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c"
+dependencies = [
+ "windows_aarch64_gnullvm 0.48.5",
+ "windows_aarch64_msvc 0.48.5",
+ "windows_i686_gnu 0.48.5",
+ "windows_i686_msvc 0.48.5",
+ "windows_x86_64_gnu 0.48.5",
+ "windows_x86_64_gnullvm 0.48.5",
+ "windows_x86_64_msvc 0.48.5",
+]
+
[[package]]
name = "windows-targets"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973"
dependencies = [
- "windows_aarch64_gnullvm",
- "windows_aarch64_msvc",
- "windows_i686_gnu",
+ "windows_aarch64_gnullvm 0.52.6",
+ "windows_aarch64_msvc 0.52.6",
+ "windows_i686_gnu 0.52.6",
"windows_i686_gnullvm",
- "windows_i686_msvc",
- "windows_x86_64_gnu",
- "windows_x86_64_gnullvm",
- "windows_x86_64_msvc",
+ "windows_i686_msvc 0.52.6",
+ "windows_x86_64_gnu 0.52.6",
+ "windows_x86_64_gnullvm 0.52.6",
+ "windows_x86_64_msvc 0.52.6",
]
+[[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8"
+
[[package]]
name = "windows_aarch64_gnullvm"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc"
+
[[package]]
name = "windows_aarch64_msvc"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
+[[package]]
+name = "windows_i686_gnu"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e"
+
[[package]]
name = "windows_i686_gnu"
version = "0.52.6"
@@ -2110,24 +2238,48 @@ version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
+[[package]]
+name = "windows_i686_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406"
+
[[package]]
name = "windows_i686_msvc"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e"
+
[[package]]
name = "windows_x86_64_gnu"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
+[[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc"
+
[[package]]
name = "windows_x86_64_gnullvm"
version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538"
+
[[package]]
name = "windows_x86_64_msvc"
version = "0.52.6"
diff --git a/Cargo.toml b/Cargo.toml
index 216d105..0c046ef 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -63,6 +63,22 @@ sync = ["dep:reqwest", "dep:tracing", "tokio"]
# the OpenRouter reference provider, but the pipeline depends only on the
# `ChatProvider` / `Summariser` / `EmbeddingBackend` traits.
persona = []
+
+# Git-backed wiki mirror of derived summary nodes
+# (`memory::store::content::wiki_git`): initialises `/wiki/.git`,
+# commits `summaries/**`, and stores read high-water marks as lightweight
+# `refs/tags/read/*` pointers. Enables `git2` directly rather than implying
+# `git-diff`, which would additionally compile in the unrelated `memory::diff`
+# module — the same posture as `sync` and `providers-http` both enabling
+# `dep:reqwest` independently.
+wiki-git = ["dep:git2", "dep:hex"]
+
+# Obsidian vault interop (`memory::store::content::{obsidian,obsidian_registry}`):
+# stages the bundled `.obsidian/` defaults into the content root, and
+# best-effort detection of whether that root is a vault Obsidian already knows
+# about (its `obsidian.json` registry).
+obsidian = ["dep:dirs"]
+
[dependencies]
anyhow = "1"
log = "0.4"
@@ -103,8 +119,15 @@ git2 = { version = "0.21", default-features = false, features = [
reqwest = { version = "0.12", default-features = false, features = [
"json",
"rustls-tls",
+ # `Response::bytes_stream()` — lets the network readers cap body size while
+ # streaming instead of after buffering the whole response.
+ "stream",
], optional = true }
tracing = { version = "0.1", optional = true }
+# Per-OS config/home dir probing for the Obsidian vault registry (`obsidian`).
+dirs = { version = "5", optional = true }
+# Hex-encodes summary read-pointer ids into git tag names (`wiki-git`).
+hex = { version = "0.4", optional = true }
# tokio powers the optional background worker loops (`tokio` feature). It is
# also listed under dev-dependencies so the async test suite always compiles.
tokio = { version = "1", features = [
@@ -113,6 +136,12 @@ tokio = { version = "1", features = [
"macros",
"time",
"sync",
+ # `process` powers the GitHub source reader's `gh` / `git` subprocess calls
+ # (`memory::sources::readers::github`, behind the `sync` feature).
+ "process",
+ # `net` powers the web-page reader's DNS-resolution SSRF guard
+ # (`tokio::net::lookup_host`, behind the `sync` feature).
+ "net",
], optional = true }
[dev-dependencies]
diff --git a/src/memory/health.rs b/src/memory/health.rs
new file mode 100644
index 0000000..3bf1fa8
--- /dev/null
+++ b/src/memory/health.rs
@@ -0,0 +1,350 @@
+//! Typed failure + degradation model for the memory pipeline.
+//!
+//! The chunk→wiki pipeline and the time-tree summarizer fail in several
+//! distinct ways (budget exhausted, missing/invalid key, missing local
+//! model, dimension mismatch, extraction timeout, transient network).
+//! Historically these all collapsed into an opaque error string and were
+//! retried identically — so a hard "Insufficient budget" 4xx burned the
+//! retry budget and the user saw a generic `error: N failed jobs`.
+//!
+//! This module is the single source of truth that fixes that:
+//!
+//! - [`FailureCode`] enumerates every distinguishable cause.
+//! - Each code maps to a [`FailureClass`] (`Transient` ⇒ retry with
+//! backoff, `Unrecoverable` ⇒ fail fast) and a stable i18n
+//! `remediation_key` so the status surface / doctor / job row all show
+//! consistent, actionable text. Embeddings remediation leads with the
+//! local-Ollama path (the steered primary fix), with BYO key secondary.
+//! - [`PipelineFailure`] is a `std::error::Error`, so it can be wrapped in
+//! `anyhow` and propagated up through the job processor, then downcast in
+//! the queue worker to decide retry-vs-fail.
+//! - [`DegradedState`] captures "the pipeline ran but recall/structure is
+//! reduced" — surfaced so degraded output is never presented as success.
+//!
+//! ## Scope: taxonomy only
+//!
+//! This is the engine's own failure vocabulary and nothing else. The
+//! *process-global degradation flags* that the embed/extract stages set and the
+//! `pipeline_status` RPC reads stay in the embedding host, because they are
+//! coupled to a host socket broadcast and to host plumbing. So does the
+//! `doctor` report (it reads the host's scheduler-gate config) and the
+//! user-error publisher (its `kind` string is a pinned frontend contract).
+//!
+//! Note the name: `tinycortex_api::health` is a *different* thing — driver
+//! liveness (`MemoryHealth`). This module is pipeline failure classification.
+//!
+//! The `remediation_key` values are i18n keys resolved by the host's frontend.
+//! They travel with [`FailureCode`] because `PipelineFailure::remediation_key`
+//! is a serialized wire field populated by `PipelineFailure::new`; splitting the
+//! table out would change the type's shape. The emitted strings are unchanged.
+
+use std::fmt;
+
+mod types;
+
+pub use types::{DegradedState, FailureClass, FailureCode, PipelineFailure};
+
+impl FailureClass {
+ pub fn as_str(self) -> &'static str {
+ match self {
+ Self::Transient => "transient",
+ Self::Unrecoverable => "unrecoverable",
+ }
+ }
+}
+
+impl FailureCode {
+ /// Stable wire string.
+ pub fn as_str(self) -> &'static str {
+ match self {
+ Self::BudgetExhausted => "budget_exhausted",
+ Self::AuthMissing => "auth_missing",
+ Self::AuthInvalid => "auth_invalid",
+ Self::EmbeddingsUnconfigured => "embeddings_unconfigured",
+ Self::EmbeddingDimMismatch => "embedding_dim_mismatch",
+ Self::LocalModelUnavailable => "local_model_unavailable",
+ Self::ExtractionTimeout => "extraction_timeout",
+ Self::SummarizerUnavailable => "summarizer_unavailable",
+ Self::EmptyInputRefused => "empty_input_refused",
+ Self::StorageUnavailable => "storage_unavailable",
+ Self::Transient => "transient",
+ }
+ }
+
+ /// Parses the stable wire string produced by [`Self::as_str`].
+ ///
+ /// Deliberately an inherent method returning `Option`, not a
+ /// [`std::str::FromStr`] impl: the trait must return `Result`, and this
+ /// arrived here as a verbatim relocation from the OpenHuman host. Changing
+ /// the signature would be an API change smuggled inside a move, which is
+ /// exactly what the port was structured to avoid. Revisit as its own
+ /// change if a `FromStr` impl is ever wanted.
+ #[allow(clippy::should_implement_trait)]
+ pub fn from_str(s: &str) -> Option {
+ Some(match s {
+ "budget_exhausted" => Self::BudgetExhausted,
+ "auth_missing" => Self::AuthMissing,
+ "auth_invalid" => Self::AuthInvalid,
+ "embeddings_unconfigured" => Self::EmbeddingsUnconfigured,
+ "embedding_dim_mismatch" => Self::EmbeddingDimMismatch,
+ "local_model_unavailable" => Self::LocalModelUnavailable,
+ "extraction_timeout" => Self::ExtractionTimeout,
+ "summarizer_unavailable" => Self::SummarizerUnavailable,
+ "empty_input_refused" => Self::EmptyInputRefused,
+ "storage_unavailable" => Self::StorageUnavailable,
+ "transient" => Self::Transient,
+ _ => return None,
+ })
+ }
+
+ /// Retry policy for this cause.
+ ///
+ /// [`LocalModelUnavailable`](Self::LocalModelUnavailable) is deliberately
+ /// **transient** even though the user has to act: the condition (Ollama
+ /// daemon stopped, model not pulled) clears from outside the app, and only
+ /// transient rows are picked up by `requeue_transient_failed` — the
+ /// automatic self-healing requeue. Classifying it unrecoverable would park
+ /// every affected job until someone clicks "Retry failed" by hand, so a
+ /// user who simply restarts Ollama would never see ingestion resume.
+ pub fn class(self) -> FailureClass {
+ match self {
+ Self::Transient | Self::ExtractionTimeout | Self::LocalModelUnavailable => {
+ FailureClass::Transient
+ }
+ _ => FailureClass::Unrecoverable,
+ }
+ }
+
+ /// i18n key for the user-facing remediation. Embeddings causes lead
+ /// with the local-Ollama path (the steered primary fix per spec FR-015).
+ pub fn remediation_key(self) -> &'static str {
+ match self {
+ Self::BudgetExhausted => "memory.health.remediation.budget_exhausted",
+ Self::AuthMissing => "memory.health.remediation.auth_missing",
+ Self::AuthInvalid => "memory.health.remediation.auth_invalid",
+ Self::EmbeddingsUnconfigured => "memory.health.remediation.embeddings_unconfigured",
+ Self::EmbeddingDimMismatch => "memory.health.remediation.embedding_dim_mismatch",
+ Self::LocalModelUnavailable => "memory.health.remediation.local_model_unavailable",
+ Self::ExtractionTimeout => "memory.health.remediation.extraction_timeout",
+ Self::SummarizerUnavailable => "memory.health.remediation.summarizer_unavailable",
+ Self::EmptyInputRefused => "memory.health.remediation.empty_input_refused",
+ Self::StorageUnavailable => "memory.health.remediation.storage_unavailable",
+ Self::Transient => "memory.health.remediation.transient",
+ }
+ }
+}
+
+impl PipelineFailure {
+ /// Build a failure from a code, deriving class + remediation key.
+ pub fn new(code: FailureCode) -> Self {
+ Self {
+ code,
+ class: code.class(),
+ remediation_key: code.remediation_key().to_string(),
+ detail: None,
+ }
+ }
+
+ /// Attach a non-localized detail string (bounded by `truncate_detail`;
+ /// never log secrets).
+ pub fn with_detail(mut self, detail: impl Into) -> Self {
+ let detail = detail.into();
+ self.detail = Some(truncate_detail(&detail));
+ self
+ }
+
+ /// True when this failure should fail fast (no retry budget).
+ pub fn is_unrecoverable(&self) -> bool {
+ self.class == FailureClass::Unrecoverable
+ }
+}
+
+/// Classify an embedding-stage error into a typed [`PipelineFailure`].
+///
+/// The embed path bottoms out in `embeddings::openai::OpenAiEmbedding::embed`,
+/// which on a non-2xx response bails with the message
+/// `"Embedding API error (): "` (status is reqwest's
+/// `StatusCode` Display, e.g. `402 Payment Required`). Dimension mismatches
+/// surface from the memory-tree `CloudEmbedder`/trait validator as
+/// `"... returned N dims, expected M"` or `"... dims, expected ..."`. We
+/// parse those shapes to decide retry-vs-fail:
+///
+/// - `401` / `403` → `auth_invalid` (a bearer was sent but rejected).
+/// - `402` / `429` / a body mentioning budget/quota/insufficient →
+/// `budget_exhausted` (the managed Voyage route is out of budget; the
+/// user must bring their own key or top up — retrying won't help).
+/// - dimension-mismatch text → `embedding_dim_mismatch`.
+/// - Ollama daemon-unreachable / model-not-pulled text →
+/// `local_model_unavailable`, so the panel names the local-runtime fix.
+/// - everything else (5xx, timeouts, transport, unparseable) → `transient`,
+/// so the worker's existing retry-with-backoff still applies.
+///
+/// Operates on the flattened `anyhow` chain (`{err:#}`) so it still matches
+/// when the embed error has been `.context()`-wrapped on the way up.
+pub fn classify_embed_error(err: &anyhow::Error) -> PipelineFailure {
+ let msg = format!("{err:#}");
+ classify_embed_error_str(&msg)
+}
+
+/// String-level core of [`classify_embed_error`], split out so unit tests can
+/// exercise the mapping without constructing reqwest errors.
+pub fn classify_embed_error_str(msg: &str) -> PipelineFailure {
+ let lower = msg.to_ascii_lowercase();
+
+ // #13021: client-side refusal from the provider pre-flight guard fires
+ // *before* any HTTP round-trip, so it carries no `Embedding API error
+ // ()` shape. Without an explicit match it would fall through to
+ // `Transient` and the `reembed_backfill` worker would retry the same
+ // un-embeddable row forever (and eventually fail the whole job).
+ // Classify as unrecoverable per-row so the worker tombstones the chunk /
+ // summary instead. Both `OpenAiEmbedding::embed` and
+ // `OpenHumanCloudEmbedding::embed` use the literal phrase
+ // "refusing empty/whitespace".
+ if lower.contains("refusing empty/whitespace") {
+ return PipelineFailure::new(FailureCode::EmptyInputRefused)
+ .with_detail(truncate_detail(msg));
+ }
+
+ // Sibling of the #13021 case above: `OpenHumanCloudEmbedding::resolve_bearer`
+ // bails *before any HTTP round-trip* when the desktop/backend session
+ // bearer is absent (user signed out), with the literal phrase
+ // "No backend session for cloud embeddings ..." (see
+ // `src/openhuman/inference/embeddings/cloud.rs`). Being a client-side bail it carries
+ // no `Embedding API error ()` shape, so without this match it falls
+ // through to `Transient` — the Memory Tree then shows "temporary error…
+ // will retry automatically" and the worker retries an auth failure that a
+ // retry can never fix. Classify as `AuthMissing` so the health banner
+ // surfaces the "log in to OpenHuman" remediation and the job fails fast.
+ if lower.contains("no backend session") {
+ return PipelineFailure::new(FailureCode::AuthMissing).with_detail(truncate_detail(msg));
+ }
+
+ // #5354 — the local Ollama runtime is not usable: the daemon is not
+ // listening, or the configured embedding model was never pulled. Both are
+ // emitted by `tinyagents::harness::embeddings::ollama` with the fix already
+ // in the text:
+ //
+ // "ollama embed request failed (is Ollama running at ?): …"
+ // "Ollama embedding model `` is not installed at . Run `ollama pull ` …"
+ //
+ // Neither carries an `Embedding API error ()` shape — the first is a
+ // transport bail, the second a rewritten 404 — so both used to fall through
+ // to `Transient` and surface as "a temporary error … will retry
+ // automatically". That is the wrong remediation: retrying cannot start a
+ // daemon or pull a model, and the user was never told what to do. Match the
+ // two shapes explicitly so the status panel renders the
+ // `local_model_unavailable` remediation instead. The class stays transient
+ // (see `FailureCode::class`) so jobs auto-resume once Ollama is back.
+ //
+ // Anchored on Ollama-specific wording so a generic cloud-embedder transport
+ // failure ("error sending request for url …") keeps its `Transient` code.
+ if lower.contains("is ollama running at")
+ || (lower.contains("ollama embedding model") && lower.contains("is not installed at"))
+ {
+ return PipelineFailure::new(FailureCode::LocalModelUnavailable)
+ .with_detail(truncate_detail(msg));
+ }
+
+ // Dimension mismatch — the trait validator / CloudEmbedder rejects a
+ // vector whose length isn't EMBEDDING_DIM. Check before status parsing:
+ // it's a 2xx-but-wrong-shape case with no HTTP status to match.
+ if lower.contains("dims, expected") || lower.contains("dimensions, expected") {
+ return PipelineFailure::new(FailureCode::EmbeddingDimMismatch)
+ .with_detail(truncate_detail(msg));
+ }
+
+ // Budget/quota wording wins regardless of the numeric status — the
+ // managed backend may surface budget exhaustion as 4xx with an explicit
+ // body, and we always want the BYO-key remediation here.
+ if lower.contains("insufficient budget")
+ || lower.contains("budget")
+ || lower.contains("quota")
+ || lower.contains("payment required")
+ {
+ return PipelineFailure::new(FailureCode::BudgetExhausted)
+ .with_detail(truncate_detail(msg));
+ }
+
+ // Parse the HTTP status out of the `Embedding API error (): ...`
+ // shape. reqwest renders e.g. `402 Payment Required`, so the first
+ // 3-digit run after the opening paren is the code.
+ if let Some(code) = parse_http_status(msg) {
+ return match code {
+ 401 | 403 => {
+ PipelineFailure::new(FailureCode::AuthInvalid).with_detail(truncate_detail(msg))
+ }
+ 402 => {
+ PipelineFailure::new(FailureCode::BudgetExhausted).with_detail(truncate_detail(msg))
+ }
+ 429 => PipelineFailure::new(FailureCode::Transient).with_detail(truncate_detail(msg)),
+ // 4xx other than the above is a hard client error retrying won't
+ // fix (malformed request, model not found); fail fast but tag it
+ // generically as auth_invalid's sibling — use Transient only for
+ // 5xx/unknown. We treat unknown 4xx as unrecoverable via
+ // budget? No — be conservative: only the known codes above are
+ // unrecoverable; other 4xx fall through to transient so we don't
+ // wedge on a transient 408/425.
+ 500..=599 => {
+ PipelineFailure::new(FailureCode::Transient).with_detail(truncate_detail(msg))
+ }
+ _ => PipelineFailure::new(FailureCode::Transient).with_detail(truncate_detail(msg)),
+ };
+ }
+
+ // No recognizable status — transport error, timeout, connection reset,
+ // or an unparseable message. Treat as transient so retry/backoff applies.
+ PipelineFailure::new(FailureCode::Transient).with_detail(truncate_detail(msg))
+}
+
+/// Extract the HTTP status code from an `Embedding API error ()`
+/// message. Anchors on the `Embedding API error (` marker rather than the
+/// first `(` in the flattened anyhow chain, so a wrapper context that happens
+/// to contain parentheses before the real error cannot break the parse.
+fn parse_http_status(msg: &str) -> Option {
+ let (_, rest) = msg.split_once("Embedding API error (")?;
+ let digits: String = rest
+ .trim_start()
+ .chars()
+ .take_while(|c| c.is_ascii_digit())
+ .collect();
+ if digits.len() == 3 {
+ digits.parse().ok()
+ } else {
+ None
+ }
+}
+
+/// Cap a detail string so we never balloon logs / wire payloads with a full
+/// provider response body. Never contains a secret (it's an error body), but
+/// keep it short anyway.
+fn truncate_detail(s: &str) -> String {
+ const MAX: usize = 200;
+ if s.chars().count() <= MAX {
+ return s.to_string();
+ }
+ let truncated: String = s.chars().take(MAX).collect();
+ format!("{truncated}…")
+}
+
+impl fmt::Display for PipelineFailure {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ write!(f, "{} ({})", self.code.as_str(), self.class.as_str())?;
+ if let Some(detail) = &self.detail {
+ write!(f, ": {detail}")?;
+ }
+ Ok(())
+ }
+}
+
+impl std::error::Error for PipelineFailure {}
+
+impl DegradedState {
+ /// True when any degradation is present.
+ pub fn is_degraded(&self) -> bool {
+ self.semantic_recall || self.structure || self.storage
+ }
+}
+
+#[cfg(test)]
+#[path = "health_tests.rs"]
+mod tests;
diff --git a/src/memory/health/types.rs b/src/memory/health/types.rs
new file mode 100644
index 0000000..bacf72f
--- /dev/null
+++ b/src/memory/health/types.rs
@@ -0,0 +1,107 @@
+//! Type definitions for the pipeline failure + degradation taxonomy.
+//!
+//! The taxonomy itself — `FailureClass`, `FailureCode`, `PipelineFailure`,
+//! `DegradedState` — is data: serde types carried on the wire and embedded in
+//! `PipelineFailure`. The classification logic (`classify_embed_error`), the
+//! `std::error::Error` plumbing, and the remediation-key/class derivation live
+//! in the parent `health` module, so this file stays pure type definitions.
+
+use serde::{Deserialize, Serialize};
+
+/// Whether a failure should be retried (`Transient`) or fail fast
+/// (`Unrecoverable`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FailureClass {
+ /// Retry with backoff up to `max_attempts` (network 5xx, timeouts,
+ /// truncated streams).
+ Transient,
+ /// Stop immediately — retrying the same input cannot succeed (budget
+ /// exhausted, bad/missing key, missing local model, dim mismatch).
+ Unrecoverable,
+}
+
+/// A distinguishable pipeline failure cause. Each variant carries a fixed
+/// [`FailureClass`] and i18n remediation key.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FailureCode {
+ /// Managed embeddings route returned an out-of-budget error (4xx).
+ BudgetExhausted,
+ /// No auth/session available for the embeddings provider.
+ AuthMissing,
+ /// Auth present but rejected (expired/invalid key or JWT).
+ AuthInvalid,
+ /// No embeddings provider is configured at all.
+ EmbeddingsUnconfigured,
+ /// Provider returned vectors of an unexpected dimensionality.
+ EmbeddingDimMismatch,
+ /// A required local model (Ollama) is not available.
+ LocalModelUnavailable,
+ /// The extraction model timed out / exhausted retries.
+ ExtractionTimeout,
+ /// No summarization provider could be resolved for "Build Summary Trees"
+ /// — neither local AI nor a configured cloud chat provider. Distinct from
+ /// [`LocalModelUnavailable`](Self::LocalModelUnavailable), which implies the
+ /// local path was selected; this covers the cloud-only setup whose provider
+ /// failed to resolve, so the remediation names both paths.
+ SummarizerUnavailable,
+ /// The embedding provider refused an empty/whitespace input at the
+ /// pre-flight guard (#13021). Unrecoverable per-row: the offending row
+ /// will never become embeddable, so the worker must tombstone it instead
+ /// of retrying. Bail wording for both `OpenAiEmbedding::embed` and
+ /// `OpenHumanCloudEmbedding::embed` starts with
+ /// `" embed: refusing empty/whitespace input ..."`.
+ EmptyInputRefused,
+ /// The host filesystem cannot service the memory_tree path — `create_dir`
+ /// / DB open returned a persistent OS-level I/O error (EIO `5`, ENOSPC
+ /// `28`, EROFS `30`), e.g. a failing/disconnected SD card or a volume the
+ /// kernel remounted read-only. Unrecoverable from inside the app: only the
+ /// user can reseat/replace/free the storage. Distinct from the embeddings
+ /// provider faults above and from the SQLite-level `SQLITE_FULL` /
+ /// `SQLITE_CORRUPT` handled in the queue worker — this is the
+ /// directory/DB-init layer below them.
+ StorageUnavailable,
+ /// Catch-all transient failure (network 5xx, timeout, truncated JSON).
+ Transient,
+}
+
+/// A typed pipeline failure: a [`FailureCode`] plus the derived class +
+/// remediation key (carried on the wire so the frontend stays
+/// presentational) and an optional human-readable detail for logs/diagnosis.
+///
+/// Implements [`std::error::Error`] so it can be `anyhow`-wrapped at the
+/// embed/extract/summarize boundary, propagated through the job processor,
+/// and downcast in the queue worker to drive retry-vs-fail.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PipelineFailure {
+ pub code: FailureCode,
+ pub class: FailureClass,
+ /// i18n key — the frontend resolves this to localized remediation text.
+ pub remediation_key: String,
+ /// Optional non-localized detail for logs/diagnosis (never a secret).
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub detail: Option,
+}
+
+/// "The pipeline ran, but output quality is reduced." Surfaced so degraded
+/// results are never presented as success.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct DegradedState {
+ /// True when embeddings were skipped (no usable provider) so semantic
+ /// recall falls back to recency-only.
+ pub semantic_recall: bool,
+ /// True when extraction yielded empty across the board so the wiki has
+ /// no entity/topic structure.
+ pub structure: bool,
+ /// True when the memory_tree's own storage path is unusable — the host
+ /// filesystem returned a persistent I/O error on dir-create / DB open
+ /// (EIO/ENOSPC/EROFS). This is the most severe degradation: the pipeline
+ /// can't even open its DB, so nothing else runs. `#[serde(default)]` keeps
+ /// the wire format backward-compatible (older clients omit it → `false`).
+ #[serde(default)]
+ pub storage: bool,
+ /// The cause of the most significant degradation, when known.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub cause: Option,
+}
diff --git a/src/memory/health_tests.rs b/src/memory/health_tests.rs
new file mode 100644
index 0000000..18c4392
--- /dev/null
+++ b/src/memory/health_tests.rs
@@ -0,0 +1,372 @@
+//! Tests for the pipeline failure taxonomy and the embed-error classifier.
+
+use super::*;
+
+const ALL_CODES: [FailureCode; 11] = [
+ FailureCode::BudgetExhausted,
+ FailureCode::AuthMissing,
+ FailureCode::AuthInvalid,
+ FailureCode::EmbeddingsUnconfigured,
+ FailureCode::EmbeddingDimMismatch,
+ FailureCode::LocalModelUnavailable,
+ FailureCode::ExtractionTimeout,
+ FailureCode::SummarizerUnavailable,
+ FailureCode::EmptyInputRefused,
+ FailureCode::StorageUnavailable,
+ FailureCode::Transient,
+];
+
+#[test]
+fn every_code_has_class_and_nonempty_remediation_key() {
+ for code in ALL_CODES {
+ let key = code.remediation_key();
+ assert!(
+ !key.is_empty(),
+ "{} has empty remediation key",
+ code.as_str()
+ );
+ assert!(
+ key.starts_with("memory.health.remediation."),
+ "{} remediation key has unexpected prefix: {key}",
+ code.as_str()
+ );
+ // class() must be total (no panic); Transient, ExtractionTimeout
+ // and LocalModelUnavailable are retryable, everything else is
+ // unrecoverable.
+ let class = code.class();
+ match code {
+ FailureCode::Transient
+ | FailureCode::ExtractionTimeout
+ | FailureCode::LocalModelUnavailable => {
+ assert_eq!(
+ class,
+ FailureClass::Transient,
+ "{} should be transient",
+ code.as_str()
+ );
+ }
+ _ => {
+ assert_eq!(
+ class,
+ FailureClass::Unrecoverable,
+ "{} should be unrecoverable",
+ code.as_str()
+ );
+ }
+ }
+ }
+}
+
+#[test]
+fn code_str_roundtrips() {
+ for code in ALL_CODES {
+ assert_eq!(FailureCode::from_str(code.as_str()), Some(code));
+ }
+ assert_eq!(FailureCode::from_str("nonsense"), None);
+}
+
+#[test]
+fn new_fills_class_and_remediation_from_code() {
+ let f = PipelineFailure::new(FailureCode::BudgetExhausted);
+ assert_eq!(f.code, FailureCode::BudgetExhausted);
+ assert_eq!(f.class, FailureClass::Unrecoverable);
+ assert_eq!(
+ f.remediation_key,
+ "memory.health.remediation.budget_exhausted"
+ );
+ assert!(f.detail.is_none());
+ assert!(f.is_unrecoverable());
+}
+
+#[test]
+fn with_detail_and_display() {
+ let f = PipelineFailure::new(FailureCode::Transient).with_detail("HTTP 503");
+ assert_eq!(f.detail.as_deref(), Some("HTTP 503"));
+ assert!(!f.is_unrecoverable());
+ assert_eq!(f.to_string(), "transient (transient): HTTP 503");
+}
+
+#[test]
+fn with_detail_truncates_long_input() {
+ // `with_detail` must bound the stored detail itself, not rely on every
+ // caller remembering to pre-truncate — an unbounded detail would balloon
+ // logs / wire payloads with a full provider response body.
+ let f = PipelineFailure::new(FailureCode::Transient).with_detail("x".repeat(500));
+ let detail = f.detail.expect("detail must be set");
+ assert!(
+ detail.chars().count() <= 201,
+ "detail not bounded, got {} chars",
+ detail.chars().count()
+ );
+ assert!(detail.ends_with('…'));
+}
+
+#[test]
+fn pipeline_failure_serde_roundtrips() {
+ let f = PipelineFailure::new(FailureCode::EmbeddingDimMismatch).with_detail("got 3072");
+ let json = serde_json::to_string(&f).unwrap();
+ let back: PipelineFailure = serde_json::from_str(&json).unwrap();
+ assert_eq!(f, back);
+ // detail omitted when None.
+ let none = PipelineFailure::new(FailureCode::AuthMissing);
+ assert!(!serde_json::to_string(&none).unwrap().contains("detail"));
+}
+
+#[test]
+fn degraded_state_default_is_healthy() {
+ let d = DegradedState::default();
+ assert!(!d.is_degraded());
+ let d2 = DegradedState {
+ structure: true,
+ ..Default::default()
+ };
+ assert!(d2.is_degraded());
+}
+
+#[test]
+fn pipeline_failure_is_error_and_downcasts_from_anyhow() {
+ let err: anyhow::Error = anyhow::Error::new(PipelineFailure::new(FailureCode::BudgetExhausted));
+ let downcast = err.downcast_ref::();
+ assert!(downcast.is_some());
+ assert!(downcast.unwrap().is_unrecoverable());
+}
+
+// ── classify_embed_error (T008) ──────────────────────────────────────
+
+#[test]
+fn classify_budget_from_body_wording() {
+ // The managed Voyage route surfaces budget exhaustion in the body.
+ let f = classify_embed_error_str(
+ "Embedding API error (400 Bad Request): {\"error\":\"Insufficient budget\"}",
+ );
+ assert_eq!(f.code, FailureCode::BudgetExhausted);
+ assert!(f.is_unrecoverable());
+}
+
+#[test]
+fn classify_budget_from_402() {
+ let f = classify_embed_error_str("Embedding API error (402 Payment Required): nope");
+ assert_eq!(f.code, FailureCode::BudgetExhausted);
+ assert!(f.is_unrecoverable());
+}
+
+#[test]
+fn classify_429_rate_limit_as_transient() {
+ let f = classify_embed_error_str("Embedding API error (429 Too Many Requests): nope");
+ assert_eq!(f.code, FailureCode::Transient);
+ assert!(!f.is_unrecoverable());
+}
+
+#[test]
+fn classify_auth_from_401_403() {
+ for status in ["401 Unauthorized", "403 Forbidden"] {
+ let f = classify_embed_error_str(&format!("Embedding API error ({status}): denied"));
+ assert_eq!(f.code, FailureCode::AuthInvalid, "status {status}");
+ assert!(f.is_unrecoverable());
+ }
+}
+
+#[test]
+fn classify_dim_mismatch() {
+ let f = classify_embed_error_str("cloud embedder returned 3072 dims, expected 1024");
+ assert_eq!(f.code, FailureCode::EmbeddingDimMismatch);
+ assert!(f.is_unrecoverable());
+}
+
+/// #13021: the provider pre-flight bail wording from both OpenAI and the
+/// cloud wrapper must classify as `EmptyInputRefused` (unrecoverable) so
+/// `reembed_backfill` tombstones the offending row instead of retrying
+/// the same blank input forever and eventually failing the job.
+#[test]
+fn classify_empty_input_refusal_as_unrecoverable() {
+ for msg in [
+ "openai embed: refusing empty/whitespace input at index 0 of 1 (model=text-embedding-3-small)",
+ "cloud embed: refusing empty/whitespace input at index 2 of 5 (model=embedding-v1)",
+ ] {
+ let f = classify_embed_error_str(msg);
+ assert_eq!(
+ f.code,
+ FailureCode::EmptyInputRefused,
+ "expected EmptyInputRefused for {msg:?}"
+ );
+ assert!(
+ f.is_unrecoverable(),
+ "EmptyInputRefused must be unrecoverable for {msg:?}"
+ );
+ }
+}
+
+/// The refusal must out-rank the dim-mismatch and budget rules even when
+/// the wrapped error happens to contain those tokens — the refusal phrase
+/// is the most specific signal and the only one that means "this row is
+/// permanently un-embeddable", not "the provider is misbehaving".
+#[test]
+fn classify_empty_input_refusal_through_anyhow_context_chain() {
+ let base = anyhow::anyhow!(
+ "openai embed: refusing empty/whitespace input at index 0 of 1 (model=embedding-v1)"
+ );
+ let wrapped = base
+ .context("embed summary during seal tree_id=t level=0")
+ .context("reembed_backfill chunk_id=c");
+ let f = classify_embed_error(&wrapped);
+ assert_eq!(f.code, FailureCode::EmptyInputRefused);
+ assert!(f.is_unrecoverable());
+}
+
+/// #4359: `OpenHumanCloudEmbedding::resolve_bearer` bails with "No backend
+/// session for cloud embeddings ..." *before any HTTP call* when the user
+/// is signed out. This must classify as `AuthMissing` (unrecoverable, "log
+/// in to OpenHuman" remediation) rather than falling through to `Transient`
+/// ("will retry automatically" — a loop that an auth failure can never win).
+#[test]
+fn classify_no_backend_session_as_auth_missing() {
+ let msg = "No backend session for cloud embeddings: log in to OpenHuman, or set \
+ memory.embedding_provider to \"ollama\" / \"none\" in config.toml";
+ let f = classify_embed_error_str(msg);
+ assert_eq!(
+ f.code,
+ FailureCode::AuthMissing,
+ "expected AuthMissing for {msg:?}"
+ );
+ assert_eq!(f.class, FailureClass::Unrecoverable);
+ assert_eq!(f.remediation_key, "memory.health.remediation.auth_missing");
+ assert!(f.is_unrecoverable());
+}
+
+/// The match must be case-insensitive and survive `anyhow` context wrapping:
+/// the bail is `.context()`-wrapped on its way up through the embed pipeline
+/// (e.g. `embed_each_via_provider` adds "cloud embeddings failed"), and
+/// `classify_embed_error` flattens the chain via `{err:#}`.
+#[test]
+fn classify_no_backend_session_through_anyhow_context_chain() {
+ let base = anyhow::anyhow!(
+ "No backend session for cloud embeddings: log in to OpenHuman, or set \
+ memory.embedding_provider to \"ollama\" / \"none\" in config.toml"
+ );
+ let wrapped = base
+ .context("cloud embeddings failed")
+ .context("reembed_backfill chunk_id=c");
+ let f = classify_embed_error(&wrapped);
+ assert_eq!(f.code, FailureCode::AuthMissing);
+ assert!(f.is_unrecoverable());
+}
+
+#[test]
+fn classify_5xx_is_transient() {
+ let f = classify_embed_error_str("Embedding API error (503 Service Unavailable): retry");
+ assert_eq!(f.code, FailureCode::Transient);
+ assert!(!f.is_unrecoverable());
+}
+
+#[test]
+fn classify_transport_error_is_transient() {
+ let f = classify_embed_error_str("error sending request for url (...): connection reset");
+ assert_eq!(f.code, FailureCode::Transient);
+ assert!(!f.is_unrecoverable());
+}
+
+/// #5354 — the Ollama daemon is not listening. Verbatim wording from
+/// `tinyagents::harness::embeddings::ollama::OllamaEmbeddingModel::request`.
+/// Note the parenthesised hint: `parse_http_status` reads the first `(`, so
+/// without an explicit match this fell through to `Transient` and the panel
+/// told the user to wait for a retry that can never start their daemon.
+#[test]
+fn classify_ollama_daemon_down_as_local_model_unavailable() {
+ let f = classify_embed_error_str(
+ "ollama embed request failed (is Ollama running at http://localhost:11434?): \
+ error sending request for url (http://localhost:11434/api/embed)",
+ );
+ assert_eq!(f.code, FailureCode::LocalModelUnavailable);
+ assert_eq!(
+ f.remediation_key,
+ "memory.health.remediation.local_model_unavailable"
+ );
+ // Transient so `requeue_transient_failed` resumes ingestion by itself
+ // once the user starts Ollama again.
+ assert!(!f.is_unrecoverable());
+}
+
+/// #5354 — the model was never pulled. `ollama_http_error` rewrites the
+/// 404 into remediation prose, so the `Embedding API error ()`
+/// shape the status parser looks for is gone.
+#[test]
+fn classify_ollama_model_not_pulled_as_local_model_unavailable() {
+ let f = classify_embed_error_str(
+ "Ollama embedding model `bge-m3` is not installed at http://localhost:11434. \
+ Run `ollama pull bge-m3` or choose an installed embedding model",
+ );
+ assert_eq!(f.code, FailureCode::LocalModelUnavailable);
+ assert!(!f.is_unrecoverable());
+}
+
+/// The real call path wraps the provider error twice (`ProviderEmbedder`
+/// adds "ollama embeddings failed", then the seal/reembed site adds its
+/// own context), so the matcher must survive the flattened chain.
+#[test]
+fn classify_ollama_daemon_down_through_anyhow_context_chain() {
+ let base = anyhow::anyhow!(
+ "ollama embed request failed (is Ollama running at http://127.0.0.1:11434?): \
+ tcp connect error: Connection refused (os error 61)"
+ );
+ let wrapped = base
+ .context("ollama embeddings failed")
+ .context("seal embedding failed");
+ let f = classify_embed_error(&wrapped);
+ assert_eq!(f.code, FailureCode::LocalModelUnavailable);
+}
+
+/// Regression guard for the matcher's blast radius: a cloud-embedder
+/// transport failure carries no Ollama wording and must keep its generic
+/// `Transient` code, or every network blip would start telling users to
+/// install Ollama.
+#[test]
+fn classify_non_ollama_transport_error_stays_transient() {
+ let f = classify_embed_error_str(
+ "cloud embeddings failed: error sending request for url \
+ (https://api.tinyhumans.ai/openai/v1/embeddings): connection reset",
+ );
+ assert_eq!(f.code, FailureCode::Transient);
+}
+
+#[test]
+fn classify_through_anyhow_context_chain() {
+ // The embed error is commonly `.context()`-wrapped on the way up;
+ // the flattened `{err:#}` must still classify.
+ let base = anyhow::anyhow!("Embedding API error (402 Payment Required): out of budget");
+ let wrapped = base
+ .context("cloud embeddings failed")
+ .context("seal embed");
+ let f = classify_embed_error(&wrapped);
+ assert_eq!(f.code, FailureCode::BudgetExhausted);
+}
+
+#[test]
+fn parse_http_status_extracts_leading_code() {
+ assert_eq!(
+ parse_http_status("Embedding API error (402 Payment Required): x"),
+ Some(402)
+ );
+ assert_eq!(parse_http_status("no parens here"), None);
+ assert_eq!(parse_http_status("(not a status): x"), None);
+}
+
+#[test]
+fn classify_embed_error_survives_context_with_parens() {
+ // `parse_http_status` must anchor on the `Embedding API error (` marker,
+ // not the first `(` in the flattened anyhow chain. A wrapper context
+ // containing parentheses before the real error used to make the parse
+ // return `None`, demoting a hard auth failure to `Transient`.
+ let base = anyhow::anyhow!("Embedding API error (401 Unauthorized): bad bearer");
+ let wrapped = base.context("provider rejected the request (see logs for details)");
+ let f = classify_embed_error(&wrapped);
+ assert_eq!(f.code, FailureCode::AuthInvalid);
+ assert!(f.is_unrecoverable());
+}
+
+#[test]
+fn truncate_detail_caps_length() {
+ let long = "x".repeat(500);
+ let out = truncate_detail(&long);
+ assert!(out.chars().count() <= 201, "got {}", out.chars().count());
+ assert!(out.ends_with('…'));
+}
diff --git a/src/memory/mod.rs b/src/memory/mod.rs
index 531c56e..6614394 100644
--- a/src/memory/mod.rs
+++ b/src/memory/mod.rs
@@ -73,6 +73,7 @@ pub mod entities;
pub mod fsutil;
pub mod goals;
pub mod graph;
+pub mod health;
pub mod ingest;
pub mod queue;
pub mod retrieval;
diff --git a/src/memory/queue/worker.rs b/src/memory/queue/worker.rs
index 6c1291a..6db7d44 100644
--- a/src/memory/queue/worker.rs
+++ b/src/memory/queue/worker.rs
@@ -189,11 +189,24 @@ fn settle_planned_job(
}
Err(err) => {
// Preserve the full anyhow cause chain in `last_error` so a reader
- // can see the root cause. If the chain carries a typed `JobFailure`,
+ // can see the root cause. If the chain carries a typed failure,
// pass it through so an unrecoverable cause fails fast instead of
// burning the retry budget.
let message = format!("{err:#}");
- let typed = err.downcast_ref::();
+ // The pipeline's own failure taxonomy (`health::PipelineFailure`) is
+ // the type `classify_embed_error` returns; handlers wrap it in the
+ // anyhow chain with `.context(...)`. Map it onto the queue's
+ // settlement type so unrecoverable codes (`budget_exhausted`,
+ // `auth_invalid`, …) still fail fast.
+ let converted_pipeline_failure = err
+ .downcast_ref::()
+ .map(|f| JobFailure {
+ code: f.code.as_str(),
+ class: f.class.as_str(),
+ });
+ let typed = err
+ .downcast_ref::()
+ .or(converted_pipeline_failure.as_ref());
mark_failed_typed(config, job, &message, typed)?;
// The handler clears this flag on every successful terminal
diff --git a/src/memory/queue/worker_tests.rs b/src/memory/queue/worker_tests.rs
index ea7d328..73675ce 100644
--- a/src/memory/queue/worker_tests.rs
+++ b/src/memory/queue/worker_tests.rs
@@ -73,6 +73,45 @@ async fn run_once_parks_unparseable_payload_as_unrecoverable() {
);
}
+/// A handler that propagates the pipeline taxonomy
+/// (`health::PipelineFailure`) must still fail fast: the worker downcasts it
+/// and maps `code`/`class` onto the queue settlement type.
+#[test]
+fn pipeline_failure_downcast_fails_fast_as_unrecoverable() {
+ use crate::memory::health::{FailureCode, PipelineFailure};
+ use crate::memory::queue::types::JobKind;
+
+ let (_tmp, cfg) = test_config();
+ let poison = NewJob {
+ kind: JobKind::FlushStale,
+ payload_json: "{}".into(),
+ dedupe_key: None,
+ available_at_ms: None,
+ max_attempts: Some(5),
+ };
+ let id = enqueue(&cfg, &poison).unwrap().expect("enqueued");
+ let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+
+ // The production shape: a handler returns a `PipelineFailure` and the
+ // processor wraps it in anyhow context on the way up. The worker's
+ // downcast must still find it through the chain.
+ let err = anyhow::Error::new(
+ PipelineFailure::new(FailureCode::BudgetExhausted)
+ .with_detail("managed voyage route exhausted"),
+ )
+ .context("flush stale handler failed");
+ settle_job(&cfg, &claimed, Err(err)).unwrap();
+
+ let job = get_job(&cfg, &id).unwrap().unwrap();
+ assert_eq!(
+ job.status,
+ JobStatus::Failed,
+ "budget_exhausted PipelineFailure must fail fast, not retry"
+ );
+ assert_eq!(job.failure_class.as_deref(), Some("unrecoverable"));
+ assert_eq!(job.failure_reason.as_deref(), Some("budget_exhausted"));
+}
+
#[tokio::test]
async fn run_once_claims_and_completes_a_flush_stale_job() {
let (_tmp, cfg) = test_config();
diff --git a/src/memory/sources/mod.rs b/src/memory/sources/mod.rs
index f013a1d..85965e9 100644
--- a/src/memory/sources/mod.rs
+++ b/src/memory/sources/mod.rs
@@ -17,11 +17,12 @@
//!
//! ## Ownership boundary
//!
-//! Per the engine spec, TinyCortex does **not** own live sync, polling, or
-//! OAuth. Network-backed kinds keep their type contracts and validation here,
-//! but their live fetchers are host-owned. Only the local kinds — `folder` and
-//! `conversation` — ship real readers (see [`readers::reader_for`]). The host's
-//! sync runner consumes this registry to decide what to sync and when.
+//! TinyCortex owns fetching and parsing — `github_repo`, `rss_feed`, and
+//! `web_page` ship readers here behind the `sync` feature — but it does **not**
+//! own live sync scheduling, polling cadence, OAuth, or credentials. The host's
+//! sync runner consumes this registry to decide what to sync and when, and
+//! [`readers::reader_for`] hands out only the two kinds (`folder`,
+//! `conversation`) that are safe to read on a timer with no network egress.
pub mod readers;
pub mod registry;
diff --git a/src/memory/sources/readers/github.rs b/src/memory/sources/readers/github.rs
new file mode 100644
index 0000000..7f02810
--- /dev/null
+++ b/src/memory/sources/readers/github.rs
@@ -0,0 +1,344 @@
+//! GitHub repo source reader.
+//!
+//! Pulls **project activity** (commits, issues, PRs) from a GitHub
+//! repository — not source code. Uses the `gh` CLI when available for
+//! authenticated, higher-rate-limit access; falls back to the public
+//! GitHub REST API for unauthenticated reads.
+//!
+//! ## Module layout
+//!
+//! - [`self`] — [`GithubReader`] orchestration: item listing/reading, URL
+//! parsing, raw-archive coordinates, shared utilities, and the cached
+//! `gh`-availability probe.
+//! - `types` — API response models and the `gh`-fallback list cache.
+//! - `git` — local bare-clone + `git log` / `git show` helpers.
+//! - `api` — `gh api` / REST transport plus commit list/read helpers.
+//! - `issues` — issue and pull-request list/read helpers.
+
+mod api;
+mod git;
+mod issues;
+mod types;
+
+#[cfg(test)]
+#[path = "github_tests.rs"]
+mod tests;
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+
+use crate::memory::config::MemoryConfig;
+use crate::memory::error::MemoryEngineResult;
+use crate::memory::sources::types::{MemorySourceEntry, SourceContent, SourceItem, SourceKind};
+use crate::memory::store::content::raw::RawKind;
+
+use super::{into_engine_error, SourceReader};
+
+// Re-export for the sibling submodules and the test module.
+pub(crate) use types::{ItemKind, LIST_CACHE};
+
+/// Default number of items of **each** type (commits, issues, PRs) to pull
+/// when the source entry doesn't override it. Tunable per-source via
+/// `max_commits` / `max_issues` / `max_prs` on [`MemorySourceEntry`].
+pub(crate) const DEFAULT_GITHUB_ITEM_LIMIT: u32 = 1000;
+
+/// Timeout for a single `gh` CLI invocation (including the availability
+/// probe).
+const GH_CLI_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Whether the `gh` CLI is on PATH and runs. Probed once per process and
+/// cached: `gh api` is the preferred transport for authenticated,
+/// higher-rate-limit access, and re-probing on every item read is wasteful.
+static GH_AVAILABLE: tokio::sync::OnceCell = tokio::sync::OnceCell::const_new();
+
+/// Probe `gh --version` (async, so a stuck `gh` cannot block a worker
+/// thread) and cache the result for the process lifetime.
+async fn gh_available() -> bool {
+ *GH_AVAILABLE
+ .get_or_init(|| async {
+ let status = tokio::time::timeout(
+ GH_CLI_TIMEOUT,
+ tokio::process::Command::new("gh")
+ .arg("--version")
+ .stdout(std::process::Stdio::null())
+ .stderr(std::process::Stdio::null())
+ .status(),
+ )
+ .await;
+ status
+ .map(|s| s.map(|st| st.success()).unwrap_or(false))
+ .unwrap_or(false)
+ })
+ .await
+}
+
+pub struct GithubReader;
+
+/// Parse `owner` and `repo` from a GitHub URL.
+///
+/// Accepts only the canonical `https://github.com//[.git][/]`
+/// shape — extra segments like `/tree/main` or `/blob/...` are rejected
+/// so callers can't accidentally derive the wrong owner/repo from a
+/// deep link.
+pub(crate) fn parse_github_url(url: &str) -> Result<(String, String), String> {
+ let trimmed = url.trim();
+ let rest = trimmed
+ .strip_prefix("https://github.com/")
+ .or_else(|| trimmed.strip_prefix("http://github.com/"))
+ .or_else(|| trimmed.strip_prefix("git@github.com:"))
+ .ok_or_else(|| format!("not a GitHub URL: {url}"))?;
+ let cleaned = rest.trim_end_matches('/').trim_end_matches(".git");
+ let parts: Vec<&str> = cleaned.split('/').collect();
+ if parts.len() != 2 || parts[0].is_empty() || parts[1].is_empty() {
+ return Err(format!(
+ "expected https://github.com//, got: {url}"
+ ));
+ }
+ Ok((parts[0].to_string(), parts[1].to_string()))
+}
+
+// ── Raw-archive coordinates ─────────────────────────────────────────
+
+/// Slugifiable raw-archive source id for a repo URL.
+///
+/// Returns `github.com//`, which slugifies (via
+/// `slugify_source_id`) to `github-com--` so a source's
+/// commits/issues/PRs land under
+/// `raw/github-com--/{commits,issues,prs}/`.
+pub fn repo_archive_source_id(url: &str) -> Option {
+ let (owner, repo) = parse_github_url(url).ok()?;
+ Some(format!("github.com/{owner}/{repo}"))
+}
+
+/// Chunk-store source id for a single repo item (dedup key).
+///
+/// `github:/:` keeps per-item uniqueness for the
+/// `mem_tree_ingested_sources` dedup table while the separate
+/// [`repo_chunk_scope`] drives a shared directory.
+pub fn chunk_source_id(url: &str, item_id: &str) -> Option {
+ let (owner, repo) = parse_github_url(url).ok()?;
+ Some(format!("github:{owner}/{repo}:{item_id}"))
+}
+
+/// Repo-scoped chunk path scope so all items from one repo share a
+/// single directory in the content store (e.g. `document/github-org-repo/`).
+pub fn repo_chunk_scope(url: &str) -> Option {
+ let (owner, repo) = parse_github_url(url).ok()?;
+ Some(format!("github:{owner}/{repo}"))
+}
+
+/// Map a [`SourceItem`] id (`commit:`, `issue:`, `pr:`) to its
+/// raw-archive [`RawKind`] and the clean uid used as the filename suffix.
+pub fn raw_archive_coords(item_id: &str) -> Option<(RawKind, String)> {
+ let (kind, rest) = ItemKind::from_id(item_id)?;
+ let raw_kind = match kind {
+ ItemKind::Commit => RawKind::Commit,
+ ItemKind::Issue => RawKind::Issue,
+ ItemKind::PullRequest => RawKind::PullRequest,
+ };
+ Some((raw_kind, rest.to_string()))
+}
+
+// ── Reader implementation ───────────────────────────────────────────
+
+#[async_trait]
+impl SourceReader for GithubReader {
+ fn kind(&self) -> SourceKind {
+ SourceKind::GithubRepo
+ }
+
+ async fn list_items(
+ &self,
+ source: &MemorySourceEntry,
+ config: &MemoryConfig,
+ ) -> MemoryEngineResult> {
+ self.list_items_inner(source, config)
+ .await
+ .map_err(into_engine_error)
+ }
+
+ async fn read_item(
+ &self,
+ source: &MemorySourceEntry,
+ item_id: &str,
+ config: &MemoryConfig,
+ ) -> MemoryEngineResult {
+ self.read_item_inner(source, item_id, config)
+ .await
+ .map_err(into_engine_error)
+ }
+}
+
+impl GithubReader {
+ async fn list_items_inner(
+ &self,
+ source: &MemorySourceEntry,
+ config: &MemoryConfig,
+ ) -> Result, String> {
+ let url = source
+ .url
+ .as_deref()
+ .ok_or("github source requires a url")?;
+ let (owner, repo) = parse_github_url(url)?;
+ let use_gh = gh_available().await;
+
+ let max_commits = source.max_commits.unwrap_or(DEFAULT_GITHUB_ITEM_LIMIT);
+ let max_issues = source.max_issues.unwrap_or(DEFAULT_GITHUB_ITEM_LIMIT);
+ let max_prs = source.max_prs.unwrap_or(DEFAULT_GITHUB_ITEM_LIMIT);
+ // A configured branch narrows commits to that ref; configured paths
+ // narrow them to the touched files. Both fall through to the API
+ // fallback so the two transports agree on scope.
+ let branch = source.branch.as_deref();
+ let paths = source.paths.as_slice();
+
+ let cache_dir = git::git_cache_dir(&config.workspace, &owner, &repo);
+
+ tracing::debug!(
+ owner = %owner,
+ repo = %repo,
+ use_gh = use_gh,
+ branch = %branch.unwrap_or("(all)"),
+ max_commits,
+ max_issues,
+ max_prs,
+ cache = %cache_dir.display(),
+ "[memory_sources:github] listing items"
+ );
+
+ // Clear the list cache so stale data from a prior sync doesn't
+ // leak into this run.
+ if let Ok(mut cache) = LIST_CACHE.lock() {
+ cache.clear();
+ }
+
+ let mut items = Vec::new();
+ let mut errors = Vec::new();
+
+ // Commits via local git (clone/fetch bare repo, then git log)
+ match git::list_commits_git(&owner, &repo, max_commits, &cache_dir, branch, paths).await {
+ Ok(commits) => items.extend(commits),
+ Err(e) => {
+ tracing::warn!(error = %e, "[memory_sources:github] git commit list failed, falling back to API");
+ match api::list_commits_api(&owner, &repo, max_commits, use_gh, branch, paths).await
+ {
+ Ok(commits) => items.extend(commits),
+ Err(e2) => {
+ tracing::warn!(error = %e2, "[memory_sources:github] API commit list also failed");
+ errors.push(e2);
+ }
+ }
+ }
+ }
+
+ // Issues and PRs via gh CLI / API (no local equivalent)
+ match issues::list_issues(&owner, &repo, max_issues, use_gh).await {
+ Ok(issues) => items.extend(issues),
+ Err(e) => {
+ tracing::warn!(error = %e, "[memory_sources:github] failed to list issues");
+ errors.push(e);
+ }
+ }
+
+ match issues::list_prs(&owner, &repo, max_prs, use_gh).await {
+ Ok(prs) => items.extend(prs),
+ Err(e) => {
+ tracing::warn!(error = %e, "[memory_sources:github] failed to list PRs");
+ errors.push(e);
+ }
+ }
+
+ if items.is_empty() && !errors.is_empty() {
+ return Err(format!(
+ "all GitHub API calls failed: {}",
+ errors.join("; ")
+ ));
+ }
+
+ tracing::debug!(count = items.len(), "[memory_sources:github] found items");
+ Ok(items)
+ }
+
+ async fn read_item_inner(
+ &self,
+ source: &MemorySourceEntry,
+ item_id: &str,
+ config: &MemoryConfig,
+ ) -> Result {
+ let url = source
+ .url
+ .as_deref()
+ .ok_or("github source requires a url")?;
+ let (owner, repo) = parse_github_url(url)?;
+ let use_gh = gh_available().await;
+
+ let (kind, ref_id) =
+ ItemKind::from_id(item_id).ok_or_else(|| format!("invalid item id: {item_id}"))?;
+
+ tracing::debug!(
+ item_id = %item_id,
+ kind = ?kind,
+ "[memory_sources:github] reading item"
+ );
+
+ match kind {
+ ItemKind::Commit => {
+ let cache_dir = git::git_cache_dir(&config.workspace, &owner, &repo);
+ match git::read_commit_git(&owner, &repo, ref_id, &cache_dir).await {
+ Ok(content) => Ok(content),
+ Err(e) => {
+ tracing::debug!(
+ sha = %ref_id,
+ error = %e,
+ "[memory_sources:github] git read_commit failed, falling back to API"
+ );
+ api::read_commit_api(&owner, &repo, ref_id, use_gh).await
+ }
+ }
+ }
+ ItemKind::Issue => {
+ let num: u64 = ref_id
+ .parse()
+ .map_err(|_| format!("invalid issue number: {ref_id}"))?;
+ issues::read_issue(&owner, &repo, num, use_gh).await
+ }
+ ItemKind::PullRequest => {
+ let num: u64 = ref_id
+ .parse()
+ .map_err(|_| format!("invalid PR number: {ref_id}"))?;
+ issues::read_pr(&owner, &repo, num, use_gh).await
+ }
+ }
+ }
+}
+
+// ── Utilities ───────────────────────────────────────────────────────
+
+fn parse_iso_ts(s: &str) -> Option {
+ chrono::DateTime::parse_from_rfc3339(s)
+ .ok()
+ .map(|dt| dt.timestamp_millis())
+}
+
+/// Render GitHub logins as a deduped, order-preserving, space-separated
+/// list of `@handle`s. Empty / `unknown` logins are skipped; an empty
+/// result renders as `none`. Used so unique committers/commenters surface
+/// as `handle:` entities in the memory tree.
+fn unique_handles<'a>(logins: impl Iterator) -> String {
+ let mut seen = std::collections::HashSet::new();
+ let mut out: Vec = Vec::new();
+ for login in logins {
+ let l = login.trim();
+ if l.is_empty() || l == "unknown" {
+ continue;
+ }
+ if seen.insert(l.to_string()) {
+ out.push(format!("@{l}"));
+ }
+ }
+ if out.is_empty() {
+ "none".to_string()
+ } else {
+ out.join(" ")
+ }
+}
diff --git a/src/memory/sources/readers/github/api.rs b/src/memory/sources/readers/github/api.rs
new file mode 100644
index 0000000..0f9a281
--- /dev/null
+++ b/src/memory/sources/readers/github/api.rs
@@ -0,0 +1,360 @@
+//! `gh` CLI + REST API helpers for the GitHub reader.
+//!
+//! [`fetch_github`] prefers the authenticated `gh api` path and falls back to
+//! the unauthenticated REST API. Commit list/read helpers live here; issue and
+//! pull-request list/read helpers live in the sibling `super::issues` module,
+//! and commit reads additionally have a local `git` path in the sibling
+//! `super::git` module.
+//!
+//! Branch/path filters are honored on the commits list: `sha=` and
+//! `path=` query params narrow what the API returns to the configured
+//! scope.
+
+use std::collections::HashSet;
+
+use crate::memory::sources::types::{ContentType, SourceContent, SourceItem};
+
+use super::types::GhCommit;
+use super::{parse_iso_ts, GH_CLI_TIMEOUT};
+
+/// GitHub REST API maximum page size (`per_page`).
+pub(super) const GH_PAGE_SIZE: u32 = 100;
+
+/// Hard ceiling on pagination loops so a misbehaving API (always returning a
+/// full page) can never spin forever even if `max` is enormous.
+pub(super) const GH_MAX_PAGES: u32 = 1000;
+
+/// Run `gh ` and return stdout as UTF-8.
+pub(super) async fn gh_json(args: &[&str]) -> Result {
+ let output = tokio::time::timeout(
+ GH_CLI_TIMEOUT,
+ tokio::process::Command::new("gh").args(args).output(),
+ )
+ .await
+ .map_err(|_| format!("gh command timed out after {}s", GH_CLI_TIMEOUT.as_secs()))?
+ .map_err(|e| format!("gh command failed: {e}"))?;
+
+ if !output.status.success() {
+ let stderr = String::from_utf8_lossy(&output.stderr);
+ return Err(format!("gh exited {}: {stderr}", output.status));
+ }
+
+ String::from_utf8(output.stdout).map_err(|e| format!("gh output not utf8: {e}"))
+}
+
+/// Unauthenticated GET against the GitHub REST API.
+pub(super) async fn api_get(path: &str) -> Result {
+ let url = format!("https://api.github.com{path}");
+ let client = reqwest::Client::builder()
+ .timeout(std::time::Duration::from_secs(20))
+ .build()
+ .map_err(|e| format!("failed to build GitHub client: {e}"))?;
+ let resp = client
+ .get(&url)
+ .header("User-Agent", "openhuman")
+ .header("Accept", "application/vnd.github.v3+json")
+ .send()
+ .await
+ .map_err(|e| format!("GitHub API request failed: {e}"))?;
+
+ if !resp.status().is_success() {
+ let status = resp.status();
+ let body = resp.text().await.unwrap_or_default();
+ return Err(format!("GitHub API returned {status}: {body}"));
+ }
+
+ resp.text()
+ .await
+ .map_err(|e| format!("failed to read response: {e}"))
+}
+
+/// Try `gh api` first, fall back to unauthenticated REST API.
+pub(super) async fn fetch_github(api_path: &str, use_gh: bool) -> Result {
+ if use_gh {
+ match gh_json(&["api", api_path]).await {
+ Ok(s) => return Ok(s),
+ Err(e) => {
+ tracing::debug!(
+ error = %e,
+ path = %api_path,
+ "[memory_sources:github] gh failed, falling back to API"
+ );
+ }
+ }
+ }
+ api_get(&format!("/{api_path}")).await
+}
+
+/// Fetch up to `max` rows from a paginated GitHub list endpoint.
+///
+/// Walks `?per_page=100&page=N` with a constant page size — GitHub's
+/// offset-based pagination is per_page-relative, so shrinking the page size
+/// mid-walk would re-window the offsets and silently skip rows (e.g. `max=150`
+/// would fetch items 51-100 a second time instead of 101-150). Iteration stops
+/// once `max` rows are collected or the API returns a short page (the last
+/// page); `extra_query` is appended verbatim (e.g. `"state=all"`). The result
+/// is truncated to exactly `max`.
+pub(super) async fn fetch_all_pages(
+ owner: &str,
+ repo: &str,
+ resource: &str,
+ extra_query: &str,
+ max: u32,
+ use_gh: bool,
+) -> Result, String> {
+ let fetch = |page: u32| async_fetch_page(page, owner, repo, resource, extra_query, use_gh);
+ collect_pages(resource, max, fetch).await
+}
+
+/// Fetch one page's raw JSON at a constant [`GH_PAGE_SIZE`].
+async fn async_fetch_page(
+ page: u32,
+ owner: &str,
+ repo: &str,
+ resource: &str,
+ extra_query: &str,
+ use_gh: bool,
+) -> Result {
+ let mut path = format!("repos/{owner}/{repo}/{resource}?per_page={GH_PAGE_SIZE}&page={page}");
+ if !extra_query.is_empty() {
+ path.push('&');
+ path.push_str(extra_query);
+ }
+ fetch_github(&path, use_gh).await
+}
+
+/// Core pagination walk, split out from [`fetch_all_pages`] so the loop is
+/// unit-testable with a fake fetch instead of a live GitHub API.
+///
+/// `fetch` maps a 1-based page number to the raw JSON for that page. The page
+/// size the fetch encodes must stay constant across pages — see
+/// [`fetch_all_pages`] for why shrinking it mid-walk skips rows.
+pub(super) async fn collect_pages(
+ label: &str,
+ max: u32,
+ mut fetch: F,
+) -> Result, String>
+where
+ T: serde::de::DeserializeOwned,
+ F: FnMut(u32) -> Fut,
+ Fut: std::future::Future
")
+ );
+}
+
+#[test]
+fn parse_rss_description_with_cdata_is_clean() {
+ let xml = r#"
+
+ Post
+ 1
+ Hello world]]>
+
+ "#;
+
+ let entries = parse_rss(xml).unwrap();
+ assert_eq!(entries.len(), 1);
+ assert_eq!(entries[0].body, "
Hello world
");
+ assert!(!entries[0].body.contains("CDATA"));
+}
+
+#[test]
+fn parse_rss_empty_description_falls_back_to_encoded_content() {
+ // A present-but-empty `` must not block the
+ // `content:encoded` fallback — the item carries its body there instead.
+ let xml = r#"
+
+ Post
+ 1
+
+ Full body from content:encoded]]>
+
+ "#;
+
+ let entries = parse_rss(xml).unwrap();
+ assert_eq!(entries.len(), 1);
+ assert_eq!(entries[0].body, "
Full body from content:encoded
");
+}
+
+#[test]
+fn parse_atom_empty_content_falls_back_to_summary() {
+ // Mirrors the RSS `description`/`content:encoded` pair: an empty
+ // `` must fall through to a populated ``.
+ let xml = r#"
+
+ Atom entry
+ urn:entry:1
+
+ Summary body
+
+ "#;
+
+ let entries = parse_atom(xml).unwrap();
+ assert_eq!(entries.len(), 1);
+ assert_eq!(entries[0].body, "Summary body");
+}
+
+// ── URL host redaction ──────────────────────────────────────────────
+
+#[test]
+fn url_host_redacts_userinfo() {
+ // Credentials embedded in a source URL must never reach debug traces —
+ // only the host is logged.
+ assert_eq!(
+ url_host("https://alice:secret@example.com/feed.xml"),
+ "example.com"
+ );
+}
+
+#[test]
+fn url_host_drops_path_query_and_fragment() {
+ assert_eq!(
+ url_host("https://example.com/feed?token=abc#top"),
+ "example.com"
+ );
+}
+
+#[test]
+fn url_host_fallback_strips_userinfo_without_scheme() {
+ // Scheme-less values are unparseable by reqwest; the textual fallback
+ // must still drop the `user:pass@` prefix and the path.
+ assert_eq!(url_host("alice:secret@example.com/feed"), "example.com");
+ assert_eq!(url_host("example.com/feed"), "example.com");
+}
+
+// ── Entity decoding ─────────────────────────────────────────────────
+
+#[test]
+fn decode_xml_entities_decodes_amp_last() {
+ // `<` is the escaped form of `<`; it must decode once to `<`,
+ // not twice to `<`.
+ assert_eq!(decode_xml_entities("<"), "<");
+ assert_eq!(decode_xml_entities("&"), "&");
+}
+
+#[test]
+fn decode_xml_entities_handles_all_named() {
+ assert_eq!(
+ decode_xml_entities("<b> "q" 'a' & more"),
+ " \"q\" 'a' & more"
+ );
+}
diff --git a/src/memory/sources/readers/ssrf.rs b/src/memory/sources/readers/ssrf.rs
new file mode 100644
index 0000000..85a8e96
--- /dev/null
+++ b/src/memory/sources/readers/ssrf.rs
@@ -0,0 +1,226 @@
+//! Shared SSRF guard and fetch hygiene for the network source readers.
+//!
+//! The web-page and RSS readers both fetch user-configured URLs, so they share
+//! the policy in this module.
+//!
+//! The hostname *text* check (`is_blocked_host`) rejects private IP literals
+//! (including their IPv4-mapped IPv6 forms, e.g. `::ffff:127.0.0.1`),
+//! `localhost`, `.local` / `.internal` names, and single-label hostnames, but
+//! a public-looking name can resolve to a loopback / private / link-local
+//! address (including the cloud-metadata `169.254.169.254`) at lookup time.
+//! `PublicOnlyResolver` therefore vets the resolved addresses and only lets the
+//! connection proceed to a globally routable IP, so the request is pinned to an
+//! address we have already allowed (no re-resolution between the check and the
+//! connect). Redirects are re-checked through `is_url_allowed` so a public URL
+//! cannot bounce the fetch onto an internal host.
+//!
+//! `read_body_capped` streams a response body and stops at a byte cap, so a
+//! hostile or gigantic page/feed cannot OOM the process before the size check
+//! runs.
+
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::sync::Arc;
+
+use futures::stream::StreamExt;
+use reqwest::dns::{Addrs, Name, Resolve, Resolving};
+
+/// Build an HTTP client with a redirect policy that re-applies the SSRF
+/// host/scheme check to every redirect hop, and a DNS resolver that only
+/// yields globally routable addresses.
+pub(super) fn build_client() -> Result {
+ reqwest::Client::builder()
+ .timeout(std::time::Duration::from_secs(20))
+ .redirect(reqwest::redirect::Policy::custom(|attempt| {
+ if is_url_allowed(attempt.url()) {
+ attempt.follow()
+ } else {
+ // `stop` returns the redirect response to the caller instead
+ // of following it; the read then fails on the non-2xx status.
+ attempt.stop()
+ }
+ }))
+ .dns_resolver(Arc::new(PublicOnlyResolver))
+ .build()
+ .map_err(|e| format!("failed to build http client: {e}"))
+}
+
+/// Stream a response body, failing once it exceeds `max` bytes.
+///
+/// `Response::bytes()` buffers the entire body before any size check, so a
+/// server that omits or understates `Content-Length` (for example a chunked
+/// response) could OOM the process despite the cap. Reading incrementally
+/// enforces the limit while the bytes arrive.
+pub(super) async fn read_body_capped(resp: reqwest::Response, max: u64) -> Result, String> {
+ // Trust a truthful Content-Length up front so a known-huge body is
+ // rejected before the first byte is read.
+ if let Some(len) = resp.content_length() {
+ if len > max {
+ return Err(format!(
+ "response body exceeds {max}-byte limit (Content-Length={len})"
+ ));
+ }
+ }
+
+ let mut body = Vec::new();
+ let mut stream = resp.bytes_stream();
+ while let Some(chunk) = stream.next().await {
+ let chunk = chunk.map_err(|e| format!("failed to read response body: {e}"))?;
+ body.extend_from_slice(&chunk);
+ if body.len() as u64 > max {
+ return Err(format!(
+ "response body exceeds {max}-byte limit (read {} bytes)",
+ body.len()
+ ));
+ }
+ }
+ Ok(body)
+}
+
+/// A DNS resolver that only yields globally routable addresses.
+///
+/// The text-based `is_blocked_host` check rejects private IP *literals* and
+/// local hostnames, but a public-looking hostname can resolve to a loopback,
+/// private, link-local, or cloud-metadata address (`169.254.169.254`) at
+/// lookup time. Installing this resolver means reqwest connects to addresses
+/// we have already vetted: a hostname whose current resolution is non-public
+/// fails the request instead of silently reaching an internal service, and the
+/// validated address is the one the connection is pinned to (no re-resolution
+/// between the check and the connect).
+#[derive(Debug, Default)]
+struct PublicOnlyResolver;
+
+impl Resolve for PublicOnlyResolver {
+ fn resolve(&self, name: Name) -> Resolving {
+ let host = name.as_str().to_string();
+ Box::pin(async move {
+ let addrs: Vec = tokio::net::lookup_host((host.as_str(), 0))
+ .await
+ .map_err(box_err)?
+ .filter(|addr| is_public_ip(addr.ip()))
+ .collect();
+ if addrs.is_empty() {
+ return Err(box_err(std::io::Error::new(
+ std::io::ErrorKind::AddrNotAvailable,
+ format!("host {host} resolved to no public addresses"),
+ )));
+ }
+ Ok(Box::new(addrs.into_iter()) as Addrs)
+ })
+ }
+}
+
+fn box_err(
+ e: impl std::error::Error + Send + Sync + 'static,
+) -> Box {
+ Box::new(e)
+}
+
+/// Whether `ip` is a globally routable address — the resolved-address half of
+/// the SSRF guard. Mirrors the literal/name policy in `is_blocked_host`:
+/// loopback, private, link-local, unique-local, multicast, broadcast,
+/// unspecified, and documentation/reserved ranges are not fetchable.
+fn is_public_ip(ip: IpAddr) -> bool {
+ match ip {
+ IpAddr::V4(v4) => is_public_ipv4(v4),
+ IpAddr::V6(v6) => is_public_ipv6(v6),
+ }
+}
+
+fn is_public_ipv4(ip: Ipv4Addr) -> bool {
+ if is_private_ipv4(ip) || ip.is_multicast() || ip.is_broadcast() {
+ return false;
+ }
+ let o = ip.octets();
+ // Documentation (192.0.2.0/24, 198.51.100.0/24, 203.0.113.0/24),
+ // benchmarking (198.18.0.0/15), and reserved (240.0.0.0/4) ranges are not
+ // globally routable.
+ !((o[0] == 192 && o[1] == 0 && o[2] == 2)
+ || (o[0] == 198 && o[1] == 51 && o[2] == 100)
+ || (o[0] == 203 && o[1] == 0 && o[2] == 113)
+ || (o[0] == 198 && o[1] == 18)
+ || o[0] >= 240)
+}
+
+fn is_public_ipv6(ip: Ipv6Addr) -> bool {
+ if is_private_ipv6(ip) || ip.is_multicast() {
+ return false;
+ }
+ let o = ip.octets();
+ // Documentation prefix 2001:db8::/32.
+ if o[0] == 0x20 && o[1] == 0x01 && o[2] == 0x0d && o[3] == 0xb8 {
+ return false;
+ }
+ // IPv4-mapped (`::ffff:a.b.c.d`) delegate to the embedded IPv4, so a
+ // mapped loopback/private address stays blocked.
+ if let Some(v4) = ip.to_ipv4_mapped() {
+ return is_public_ipv4(v4);
+ }
+ true
+}
+
+/// Whether a URL may be fetched: `http(s)` scheme against a public host.
+pub(super) fn is_url_allowed(url: &reqwest::Url) -> bool {
+ match url.scheme() {
+ "http" | "https" => {}
+ _ => return false,
+ }
+ let Some(host) = url.host_str() else {
+ return false;
+ };
+ !is_blocked_host(host)
+}
+
+/// Reject hosts that could target non-public resources: IP literals in
+/// loopback / private / link-local / unique-local / unspecified ranges (and
+/// their IPv4-mapped IPv6 forms), plus `localhost`, `.local` / `.internal`
+/// names, and single-label hostnames (internal service names such as `mongo`
+/// or `redis`).
+fn is_blocked_host(host: &str) -> bool {
+ let host = host.trim().trim_end_matches('.').to_ascii_lowercase();
+ if host.is_empty() {
+ return true;
+ }
+ if let Ok(ip) = host.parse::() {
+ // Use the same public-address classification as the resolved-address
+ // guard (and the IPv6 literal branch) so reserved/multicast/broadcast/
+ // documentation/benchmarking literals are rejected too. A literal never
+ // goes through DNS resolution, so the `PublicOnlyResolver` never sees
+ // it — this text check is the only line of defense for it.
+ return !is_public_ipv4(ip);
+ }
+ if let Ok(ip) = host.parse::() {
+ // Use the same public-address classification as the resolved-address
+ // guard so an IPv4-mapped literal (`::ffff:127.0.0.1`,
+ // `::ffff:10.0.0.1`) is rejected like its bare IPv4 counterpart. A
+ // literal never goes through DNS resolution, so the `PublicOnlyResolver`
+ // never sees it — this text check is the only line of defense for it.
+ return !is_public_ipv6(ip);
+ }
+ if host == "localhost" || host.ends_with(".local") || host.ends_with(".internal") {
+ return true;
+ }
+ // A single-label name is an internal-service name, not a public domain.
+ !host.contains('.')
+}
+
+fn is_private_ipv4(ip: std::net::Ipv4Addr) -> bool {
+ if ip.is_loopback() || ip.is_private() || ip.is_link_local() || ip.is_unspecified() {
+ return true;
+ }
+ let o = ip.octets();
+ // 100.64.0.0/10 CGNAT and 192.0.0.0/24 (IETF protocol assignments).
+ (o[0] == 100 && o[1] & 0xc0 == 0x40) || (o[0] == 192 && o[1] == 0)
+}
+
+fn is_private_ipv6(ip: std::net::Ipv6Addr) -> bool {
+ if ip.is_loopback() || ip.is_unspecified() {
+ return true;
+ }
+ let o = ip.octets();
+ // Unique-local fc00::/7 and link-local fe80::/10.
+ (o[0] == 0xfc || o[0] == 0xfd) || (o[0] == 0xfe && o[1] & 0xc0 == 0x80)
+}
+
+#[cfg(test)]
+#[path = "ssrf_tests.rs"]
+mod tests;
diff --git a/src/memory/sources/readers/ssrf_tests.rs b/src/memory/sources/readers/ssrf_tests.rs
new file mode 100644
index 0000000..4f8189a
--- /dev/null
+++ b/src/memory/sources/readers/ssrf_tests.rs
@@ -0,0 +1,165 @@
+use super::*;
+
+// ── SSRF guard ──────────────────────────────────────────────────────
+
+#[test]
+fn is_url_allowed_accepts_public_http_urls() {
+ assert!(is_url_allowed(
+ &reqwest::Url::parse("https://example.com").unwrap()
+ ));
+ assert!(is_url_allowed(
+ &reqwest::Url::parse("http://example.com/x").unwrap()
+ ));
+ assert!(is_url_allowed(
+ &reqwest::Url::parse("https://sub.example.com").unwrap()
+ ));
+ assert!(is_url_allowed(
+ &reqwest::Url::parse("https://8.8.8.8").unwrap()
+ ));
+}
+
+#[test]
+fn is_url_allowed_rejects_private_and_internal_targets() {
+ // Private / loopback / link-local IP literals.
+ assert!(!is_url_allowed(
+ &reqwest::Url::parse("http://127.0.0.1").unwrap()
+ ));
+ assert!(!is_url_allowed(
+ &reqwest::Url::parse("http://10.0.0.1").unwrap()
+ ));
+ assert!(!is_url_allowed(
+ &reqwest::Url::parse("http://192.168.1.1").unwrap()
+ ));
+ assert!(!is_url_allowed(
+ &reqwest::Url::parse("http://169.254.169.254").unwrap()
+ ));
+ assert!(!is_url_allowed(
+ &reqwest::Url::parse("http://[::1]").unwrap()
+ ));
+ // Internal service names and local-only names.
+ assert!(!is_url_allowed(
+ &reqwest::Url::parse("http://localhost").unwrap()
+ ));
+ assert!(!is_url_allowed(
+ &reqwest::Url::parse("http://mongo").unwrap()
+ ));
+ assert!(!is_url_allowed(
+ &reqwest::Url::parse("http://service.internal").unwrap()
+ ));
+ // Non-http scheme.
+ assert!(!is_url_allowed(
+ &reqwest::Url::parse("ftp://example.com").unwrap()
+ ));
+ assert!(!is_url_allowed(
+ &reqwest::Url::parse("file:///etc/passwd").unwrap()
+ ));
+}
+
+#[test]
+fn is_blocked_host_rejects_ip_ranges_and_local_names() {
+ let blocked = [
+ "127.0.0.1",
+ "0.0.0.0",
+ "10.0.0.1",
+ "172.16.0.1",
+ "192.168.0.1",
+ "169.254.169.254",
+ "100.64.0.1", // CGNAT
+ "192.0.0.1", // IETF protocol assignments
+ "localhost",
+ "foo.local",
+ "bar.internal",
+ "mongo",
+ "::1",
+ "fc00::1", // unique-local
+ "fe80::1", // link-local
+ "::ffff:127.0.0.1", // IPv4-mapped loopback literal
+ "::ffff:10.0.0.1", // IPv4-mapped private literal
+ "::ffff:169.254.169.254", // IPv4-mapped link-local / cloud metadata
+ // Special IPv4 literals that are not globally routable: multicast,
+ // broadcast, documentation, benchmarking, and reserved ranges. A
+ // literal never goes through DNS resolution, so the text check is the
+ // only line of defense for these.
+ "224.0.0.1", // multicast
+ "255.255.255.255", // broadcast
+ "192.0.2.1", // documentation
+ "198.51.100.1", // documentation
+ "203.0.113.1", // documentation
+ "198.18.0.1", // benchmarking
+ "240.0.0.1", // reserved
+ ];
+ for host in blocked {
+ assert!(is_blocked_host(host), "expected {host:?} to be blocked");
+ }
+}
+
+#[test]
+fn is_blocked_host_accepts_public_hosts() {
+ let allowed = [
+ "8.8.8.8",
+ "1.1.1.1",
+ "example.com",
+ "sub.example.com",
+ "example.co.uk",
+ "8.8.8.8.", // trailing dot is normalized away
+ "EXAMPLE.com", // case-insensitive
+ "2001:4860:4860::8888",
+ "::ffff:8.8.8.8", // IPv4-mapped public literal
+ ];
+ for host in allowed {
+ assert!(!is_blocked_host(host), "expected {host:?} to be allowed");
+ }
+}
+
+// ── resolved-address (DNS) SSRF classification ──────────────────────
+
+fn public_ip(s: &str) -> IpAddr {
+ s.parse().expect("valid ip literal")
+}
+
+#[test]
+fn is_public_ip_rejects_internal_and_special_ranges() {
+ let blocked = [
+ "127.0.0.1", // loopback
+ "0.0.0.0", // unspecified
+ "10.0.0.1", // private
+ "172.16.0.1", // private
+ "192.168.1.1", // private
+ "169.254.169.254", // link-local / cloud metadata
+ "100.64.0.1", // CGNAT
+ "192.0.0.1", // IETF protocol assignments
+ "224.0.0.1", // multicast
+ "255.255.255.255", // broadcast
+ "192.0.2.1", // documentation
+ "198.51.100.1", // documentation
+ "203.0.113.1", // documentation
+ "198.18.0.1", // benchmarking
+ "240.0.0.1", // reserved
+ "::1", // loopback
+ "::", // unspecified
+ "fc00::1", // unique-local
+ "fe80::1", // link-local
+ "ff00::1", // multicast
+ "2001:db8::1", // documentation
+ "::ffff:127.0.0.1", // IPv4-mapped loopback
+ "::ffff:169.254.169.254", // IPv4-mapped link-local
+ ];
+ for s in blocked {
+ assert!(!is_public_ip(public_ip(s)), "expected {s:?} to be rejected");
+ }
+}
+
+#[test]
+fn is_public_ip_accepts_global_addresses() {
+ let allowed = [
+ "8.8.8.8",
+ "1.1.1.1",
+ "93.184.216.34",
+ "2001:4860:4860::8888",
+ "2606:4700:4700::1111",
+ "::ffff:8.8.8.8", // IPv4-mapped public
+ ];
+ for s in allowed {
+ assert!(is_public_ip(public_ip(s)), "expected {s:?} to be allowed");
+ }
+}
diff --git a/src/memory/sources/readers/web_page.rs b/src/memory/sources/readers/web_page.rs
new file mode 100644
index 0000000..12721c1
--- /dev/null
+++ b/src/memory/sources/readers/web_page.rs
@@ -0,0 +1,483 @@
+//! Web page source reader.
+//!
+//! Fetches a single URL and extracts its text content. When a CSS
+//! `selector` is configured, only matching elements are included;
+//! otherwise the full page body is returned.
+//!
+//! The fetch-side SSRF guard (scheme/host policy plus a DNS resolver that
+//! pins connections to globally routable addresses) lives in the shared
+//! `ssrf` module, which the RSS reader uses too.
+
+mod types;
+
+use async_trait::async_trait;
+
+use super::ssrf::{build_client, is_url_allowed, read_body_capped};
+use types::SelectorSpec;
+
+use crate::memory::config::MemoryConfig;
+use crate::memory::error::MemoryEngineResult;
+use crate::memory::sources::types::{
+ ContentType, MemorySourceEntry, SourceContent, SourceItem, SourceKind,
+};
+
+use super::{into_engine_error, SourceReader};
+
+pub struct WebPageReader;
+
+#[async_trait]
+impl SourceReader for WebPageReader {
+ fn kind(&self) -> SourceKind {
+ SourceKind::WebPage
+ }
+
+ async fn list_items(
+ &self,
+ source: &MemorySourceEntry,
+ config: &MemoryConfig,
+ ) -> MemoryEngineResult> {
+ self.list_items_inner(source, config)
+ .await
+ .map_err(into_engine_error)
+ }
+
+ async fn read_item(
+ &self,
+ source: &MemorySourceEntry,
+ item_id: &str,
+ config: &MemoryConfig,
+ ) -> MemoryEngineResult {
+ self.read_item_inner(source, item_id, config)
+ .await
+ .map_err(into_engine_error)
+ }
+}
+
+impl WebPageReader {
+ async fn list_items_inner(
+ &self,
+ source: &MemorySourceEntry,
+ _config: &MemoryConfig,
+ ) -> Result, String> {
+ let url = source
+ .url
+ .as_deref()
+ .ok_or("web_page source requires a url")?;
+
+ Ok(vec![SourceItem {
+ id: url.to_string(),
+ title: source.label.clone(),
+ updated_at_ms: None,
+ }])
+ }
+
+ async fn read_item_inner(
+ &self,
+ source: &MemorySourceEntry,
+ item_id: &str,
+ _config: &MemoryConfig,
+ ) -> Result {
+ let url = if item_id.starts_with("http") {
+ item_id.to_string()
+ } else {
+ source.url.clone().ok_or("web_page source requires a url")?
+ };
+
+ // SSRF guard: validate scheme and host, reject private/internal
+ // targets, and refuse redirects that would escape that policy.
+ let parsed = reqwest::Url::parse(&url).map_err(|e| format!("invalid URL: {e}"))?;
+ if !is_url_allowed(&parsed) {
+ return Err(format!(
+ "web_page source requires an http(s) URL to a public host, got: {}",
+ url.chars().take(64).collect::()
+ ));
+ }
+
+ tracing::debug!(
+ host = %parsed.host_str().unwrap_or(""),
+ selector = ?source.selector,
+ "[memory_sources:web_page] reading item"
+ );
+
+ let client = build_client()?;
+ let resp = client
+ .get(parsed)
+ .header("User-Agent", "openhuman")
+ .send()
+ .await
+ .map_err(|e| format!("failed to fetch page: {e}"))?;
+
+ if !resp.status().is_success() {
+ return Err(format!("page returned {}", resp.status()));
+ }
+
+ // Cap response body to 10 MiB so a hostile/giant page can't OOM us.
+ // The read is streamed so the cap is enforced while downloading, not
+ // after the whole body has been buffered into memory.
+ const MAX_BODY_BYTES: u64 = 10 * 1024 * 1024;
+ let bytes = read_body_capped(resp, MAX_BODY_BYTES).await?;
+ let body = String::from_utf8_lossy(&bytes).into_owned();
+
+ let extracted = if let Some(selector) = source.selector.as_deref() {
+ extract_by_selector(&body, selector)
+ } else {
+ strip_html_tags(&body)
+ };
+
+ Ok(SourceContent {
+ id: url.clone(),
+ title: extract_title(&body).unwrap_or_else(|| url.clone()),
+ body: extracted,
+ content_type: ContentType::Plaintext,
+ metadata: serde_json::json!({ "url": url }),
+ })
+ }
+}
+
+// ── Text extraction ─────────────────────────────────────────────────
+
+fn extract_title(html: &str) -> Option {
+ let start = html.find("')? + start + 1;
+ let end = html[content_start..].find("")? + content_start;
+ Some(html[content_start..end].trim().to_string())
+}
+
+fn parse_selector(selector: &str) -> Option {
+ let last = selector
+ .trim()
+ .rsplit(char::is_whitespace)
+ .next()
+ .unwrap_or("")
+ .trim();
+ if last.is_empty() {
+ return None;
+ }
+
+ let mut spec = SelectorSpec {
+ tag: None,
+ id: None,
+ classes: Vec::new(),
+ };
+ let mut part = String::new();
+ let mut sep = ' '; // leading bare token is the tag
+ for ch in last.chars() {
+ match ch {
+ '.' | '#' => {
+ push_selector_part(&mut spec, &mut part, sep);
+ sep = ch;
+ }
+ _ => part.push(ch),
+ }
+ }
+ push_selector_part(&mut spec, &mut part, sep);
+
+ if spec.tag.is_none() && spec.id.is_none() && spec.classes.is_empty() {
+ None
+ } else {
+ Some(spec)
+ }
+}
+
+fn push_selector_part(spec: &mut SelectorSpec, part: &mut String, sep: char) {
+ let part = std::mem::take(part);
+ if part.is_empty() {
+ return;
+ }
+ match sep {
+ '#' => spec.id = Some(part),
+ '.' => spec.classes.push(part),
+ _ => {
+ if spec.tag.is_none() {
+ spec.tag = Some(part);
+ } else {
+ spec.classes.push(part);
+ }
+ }
+ }
+}
+
+/// Extract text from elements matching a simple CSS selector.
+///
+/// Falls back to the whole stripped page when the selector never matches
+/// (rather than erroring), mirroring the reader's lenient posture for pages
+/// whose structure changes between list and read time.
+fn extract_by_selector(html: &str, selector: &str) -> String {
+ let Some(spec) = parse_selector(selector) else {
+ return strip_html_tags(html);
+ };
+ // Match against a script/style-stripped copy so JS strings and CSS rules
+ // cannot be mistaken for nested elements, and so a selector that lands on
+ // a `