From 078d0763a489cfce73d721580eb239ec2f5aa5ca Mon Sep 17 00:00:00 2001 From: Test Date: Sat, 16 May 2026 22:55:39 -0500 Subject: [PATCH] =?UTF-8?q?feat(genome):=20demand-aligned-recall=20PR-3c?= =?UTF-8?q?=20=E2=80=94=20trait=20impl=20+=20CandidateSource=20seam?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR-3c of demand-aligned-recall. Wires `DemandAlignedRecall` trait impl on `LocalDemandAlignedRecall` + introduces `CandidateSource` trait as the seam between the ranking engine and the substrate candidate sources. PR-3d will wrap the working-set-manager (#1362's bus hook) as a CandidateSource impl; PR-3c stays substrate-agnostic. Why this split PR-3c locks the source seam first. PR-3d adds the working-set walker as one impl; future PRs add the genome catalog walker + federation peer source. Each is independently testable. What lands - CandidateSource trait — async fn fetch(query, context) -> Vec. Send + Sync + async_trait for tokio. Object-safe; PR-3d's working-set walker is one impl. - LocalDemandAlignedRecall.source: Option> — optional injection. None = empty-pool mode (legitimate "no candidates locally; try federation" signal). Some = trait impl's recall() dispatches to source.fetch() then rank(). - with_source(source) constructor. - with_config_and_source(weights, half_life, source) constructor for governor-driven config + source wiring. - DemandAlignedRecall trait impl on LocalDemandAlignedRecall: - recall(query, context) — fetches via source, scores via rank() with SystemTime::now() (rank() stays pure with explicit now_ms threading for replay determinism) - replay(trace) — returns typed RecallError::ScopeUnreachable with "RecallTraceStore (sentinel PR); not yet implemented in PR-3c". Per never-swallow-errors: typed refusal beats silent empty pool. When sentinel ships RecallTraceStore, this test flips to expect Ok(pool). Design choices - Source is Option, not required. The no-source path returns empty — useful for unit tests that don't need substrate + diagnostic tooling that wants a recall engine without candidate plumbing. - `recall()` reads SystemTime::now at the trait entry. The internal rank() still takes explicit now_ms; replay determinism preserved at the pure layer, live recall at the trait layer. This is the cleanest decoupling I could find that satisfies both spec asks. - PR-3c scope: no scope filtering, no freshness enforcement, no budget filtering. The CandidateSource does query-aware pruning in its fetch(); PR-3d's working-set walker filters by RecallScope::Local. Future PRs add the rest. Tests 5 new tests on the PR-3c surface: - recall_dispatches_through_dyn_demand_aligned_recall — Arc object-safety - recall_without_source_returns_empty_pool_not_error — empty-pool contract - recall_with_source_dispatches_to_fetch_and_ranks — fetch call count + candidate-in-pool round-trip - with_config_and_source_preserves_all_three - replay_returns_typed_not_implemented_refusal_in_pr3c — pins the typed refusal so sentinel PR has a regression check to flip 18/18 pass on genome::recall_impl (13 PR-3b + 5 PR-3c). No regressions across other 2802 lib tests. Stack - #1346 / #1353 / #1355 / #1358 / #1362 — my genome stack - #1366 — DAR PR-1: pure types - #1367 + #1370 — DAR PR-2: trait + composite types - #1371 — DAR PR-3a: scoring function + per-factor curves - #1372 — DAR PR-3b: LocalDemandAlignedRecall ranking engine - THIS PR — DAR PR-3c: trait impl + CandidateSource seam - NEXT — DAR PR-3d: WorkingSetCandidateSource wrapping #1362's bus hook + concrete walker for the persona's working set Co-Authored-By: Claude Opus 4.7 (1M context) --- src/workers/continuum-core/src/genome/mod.rs | 2 +- .../continuum-core/src/genome/recall_impl.rs | 296 +++++++++++++++++- 2 files changed, 284 insertions(+), 14 deletions(-) diff --git a/src/workers/continuum-core/src/genome/mod.rs b/src/workers/continuum-core/src/genome/mod.rs index 7f70868cf..f57b950f5 100644 --- a/src/workers/continuum-core/src/genome/mod.rs +++ b/src/workers/continuum-core/src/genome/mod.rs @@ -98,4 +98,4 @@ pub use recall_scoring::{ DEFAULT_RECENCY_HALF_LIFE_MS, }; pub mod recall_impl; -pub use recall_impl::{CandidateArtifact, LocalDemandAlignedRecall}; +pub use recall_impl::{CandidateArtifact, CandidateSource, LocalDemandAlignedRecall}; diff --git a/src/workers/continuum-core/src/genome/recall_impl.rs b/src/workers/continuum-core/src/genome/recall_impl.rs index 0bedee161..3d8d4e1c7 100644 --- a/src/workers/continuum-core/src/genome/recall_impl.rs +++ b/src/workers/continuum-core/src/genome/recall_impl.rs @@ -42,13 +42,16 @@ //! - Embedding model integration (the semantic factor input) — //! separate Lane H slice. +use async_trait::async_trait; use serde::{Deserialize, Serialize}; +use std::sync::Arc; use ts_rs::TS; -use super::recall::{RecallScore, ResidencyHint}; +use super::recall::{RecallError, RecallScore, ResidencyHint}; use super::recall_scoring::{score, DEFAULT_RECENCY_HALF_LIFE_MS}; use super::recall_trait::{ - CompositionHint, EngramRef, LoRALayerRef, MoEExpertRef, RankedPool, RecallScoreWeights, + CapabilityQuery, CompositionHint, DemandAlignedRecall, EngramRef, LoRALayerRef, MoEExpertRef, + RankedPool, RecallContext, RecallScoreWeights, RecallTrace, }; use super::working_set::{ArtifactId, PageKind}; @@ -94,38 +97,102 @@ pub struct CandidateArtifact { pub provenance_trust_factor: f32, } +/// Source of recall candidates. PR-3c introduces the seam between +/// the ranking engine (LocalDemandAlignedRecall) and the substrate +/// sources (working-set-manager, genome catalog, federation peers). +/// PR-3d wraps `LocalWorkingSetManager` as a CandidateSource impl. +/// +/// `Send + Sync + async_trait` for tokio concurrency. The trait +/// takes the query + context so future impls can do query-aware +/// pruning (don't return artifacts that violate scope, exceed +/// budget, fail freshness target). +/// +/// PR-3c's stub impls in tests return canned Vec; +/// PR-3d's working-set walker returns the persona's resident pages +/// translated to candidates. +#[async_trait] +pub trait CandidateSource: Send + Sync { + /// Return all candidates relevant to the query within the + /// persona's context. Pure data — no scoring, no sorting; the + /// ranking engine handles that. + /// + /// May return an empty Vec; recall handles that gracefully + /// (no error, empty pools — caller may try federation). + async fn fetch( + &self, + query: &CapabilityQuery, + context: &RecallContext, + ) -> Vec; +} + /// Per-process implementation of demand-aligned recall ranking. -/// Holds the governor-tunable scoring weights + recency half-life; -/// the actual candidate sourcing is the caller's concern in PR-3b. +/// Holds the governor-tunable scoring weights + recency half-life +/// + an optional CandidateSource for the trait impl. /// /// Thread-safe through immutability: the struct's fields don't /// change after construction. `rank` is pure-function over the -/// candidate set + the engine's config. A future PR may add a -/// `with_weights` constructor for governor-driven weight updates; -/// PR-3b's design keeps weights immutable per instance. +/// candidate set + the engine's config. The DemandAlignedRecall +/// trait impl uses the configured CandidateSource to fetch +/// candidates; if no source is configured, recall returns an empty +/// pool (no error — that's a legitimate "no candidates known" +/// signal callers may use to fall back to federation). pub struct LocalDemandAlignedRecall { weights: RecallScoreWeights, half_life_ms: u64, + source: Option>, } impl LocalDemandAlignedRecall { - /// Construct with default weights (sum-to-1 baseline from - /// GENOME-FOUNDRY-SENTINEL Part 7) and default 24h recency - /// half-life. + /// Construct with default weights, default 24h recency + /// half-life, and no candidate source. The `rank()` method + /// works (caller passes candidates explicitly) but the trait + /// impl returns empty pools. pub fn new() -> Self { Self { weights: RecallScoreWeights::default(), half_life_ms: DEFAULT_RECENCY_HALF_LIFE_MS, + source: None, } } - /// Construct with explicit weights + half-life. Used by tests - /// and by PR-3c when wiring with governor-driven config. + /// Construct with explicit weights + half-life, no source. /// Weights are validated by `RecallScoreWeights::new` at /// construction upstream; this constructor takes them as /// already-valid. pub fn with_config(weights: RecallScoreWeights, half_life_ms: u64) -> Self { - Self { weights, half_life_ms } + Self { + weights, + half_life_ms, + source: None, + } + } + + /// Construct with a candidate source. The trait impl's + /// `recall()` calls `source.fetch()` then `rank()`. Weights + + /// half-life are at defaults; use `with_config_and_source` + /// for explicit values. + pub fn with_source(source: Arc) -> Self { + Self { + weights: RecallScoreWeights::default(), + half_life_ms: DEFAULT_RECENCY_HALF_LIFE_MS, + source: Some(source), + } + } + + /// Construct with explicit weights, half-life, AND a candidate + /// source. PR-3d's working-set walker uses this when wiring + /// LocalDemandAlignedRecall into Runtime with governor-driven + /// config. + pub fn with_config_and_source( + weights: RecallScoreWeights, + half_life_ms: u64, + source: Arc, + ) -> Self { + Self { + weights, + half_life_ms, + source: Some(source), + } } /// Score + partition + sort the candidate set. Returns a fully- @@ -238,6 +305,58 @@ impl Default for LocalDemandAlignedRecall { } } +#[async_trait] +impl DemandAlignedRecall for LocalDemandAlignedRecall { + /// Fetch candidates from the configured CandidateSource, then + /// rank them. If no source is configured (`new()` / + /// `with_config()` constructors), returns an empty pool — no + /// error, because "no candidates known locally" is a + /// legitimate signal callers may use to fall back to + /// federation. + /// + /// `now_ms` is read from `SystemTime::now()` here (the public + /// entry point), then threaded through `rank()` which keeps + /// the explicit-now-ms contract for replay determinism. The + /// trait surface looks "live" but `rank()` stays pure. + /// + /// PR-3c scope: no scope filtering, no freshness enforcement, + /// no budget filtering. The CandidateSource does query-aware + /// pruning in its `fetch()`; PR-3d's working-set walker + /// filters by RecallScope::Local. Future PRs add the rest. + async fn recall( + &self, + query: &CapabilityQuery, + context: &RecallContext, + ) -> Result { + let candidates = match &self.source { + Some(src) => src.fetch(query, context).await, + None => Vec::new(), + }; + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + Ok(self.rank(now_ms, candidates)) + } + + /// Replay support deferred to a sentinel-owned PR. PR-3c + /// returns `RecallError::ScopeUnreachable` with a clear reason + /// so callers see a typed refusal rather than silent empty + /// pool — per Joel's "never swallow errors" rule. The sentinel + /// PR will add a RecallTraceStore that maps RecallTrace → + /// snapshotted (weights, candidate_set, now_ms), then replay + /// re-ranks deterministically. + async fn replay( + &self, + _trace: &super::recall_trait::RecallTrace, + ) -> Result { + Err(RecallError::ScopeUnreachable { + reason: "replay requires RecallTraceStore (sentinel PR); not yet implemented in PR-3c" + .to_string(), + }) + } +} + #[cfg(test)] mod tests { //! Pin the ranking behavior: @@ -518,4 +637,155 @@ mod tests { "different now_ms must yield different trace_ref" ); } + + // ─── PR-3c: trait impl + CandidateSource tests ───────────── + + use crate::genome::recall_trait::{ + CapabilityQuery, DemandAlignedRecall, DomainHint, RecallBudget, RecallContext, RecallTrace, + }; + use crate::genome::recall::{FreshnessTarget, RecallError, RecallScope, TaskKind}; + use crate::genome::working_set::PersonaId; + use parking_lot::Mutex; + + /// Stub CandidateSource: returns a pre-set Vec on every call, + /// records each fetch invocation so tests can assert it ran. + struct StubSource { + canned: Vec, + fetch_calls: Mutex, + } + + impl StubSource { + fn new(canned: Vec) -> Arc { + Arc::new(Self { + canned, + fetch_calls: Mutex::new(0), + }) + } + fn fetch_count(&self) -> u32 { + *self.fetch_calls.lock() + } + } + + #[async_trait] + impl CandidateSource for StubSource { + async fn fetch( + &self, + _query: &CapabilityQuery, + _context: &RecallContext, + ) -> Vec { + *self.fetch_calls.lock() += 1; + self.canned.clone() + } + } + + fn sample_query() -> CapabilityQuery { + CapabilityQuery { + task_kind: TaskKind::Chat, + domain_hints: vec![DomainHint::new("test")], + budget: RecallBudget { + max_bytes: 1_000_000, + max_duration_ms: 100, + }, + must_include: vec![], + prefer_refined: true, + scope: RecallScope::Local, + freshness_target: FreshnessTarget::BestEffort, + } + } + + fn sample_persona() -> PersonaId { + PersonaId::new(Uuid::from_u128(100)) + } + + /// What this catches: trait impl exists + is object-safe. + /// `Arc` dispatch through LocalDemand + /// AlignedRecall works. This is the seam persona-cognition will + /// use. + #[tokio::test] + async fn recall_dispatches_through_dyn_demand_aligned_recall() { + let recall: Arc = + Arc::new(LocalDemandAlignedRecall::new()); + let ctx = RecallContext::cold_start(sample_persona()); + let pool = recall.recall(&sample_query(), &ctx).await.unwrap(); + assert!(pool.layers.is_empty()); + assert!(pool.experts.is_empty()); + assert!(pool.engrams.is_empty()); + } + + /// What this catches: no-source mode returns empty pool, NOT + /// an error. Empty pool is the legitimate "no candidates + /// known locally; caller may try federation" signal. + #[tokio::test] + async fn recall_without_source_returns_empty_pool_not_error() { + let recall = LocalDemandAlignedRecall::new(); + let ctx = RecallContext::cold_start(sample_persona()); + let result = recall.recall(&sample_query(), &ctx).await; + assert!(result.is_ok()); + let pool = result.unwrap(); + assert!(pool.layers.is_empty()); + } + + /// What this catches: with_source dispatches to the source's + /// fetch() — count the calls to prove dispatch happened. The + /// source's canned candidates land in the resulting pool. + #[tokio::test] + async fn recall_with_source_dispatches_to_fetch_and_ranks() { + let hot = ResidencyHint::Hot { role: super::super::tier::TierRole::Fast }; + let cand = CandidateArtifact { + kind: PageKind::LoRALayer, + artifact_id: ArtifactId::new(Uuid::from_u128(42)), + semantic_factor: 0.9, + outcome_history_factor: 0.8, + last_used_ms: 0, + residency: hot, + provenance_trust_factor: 0.7, + }; + let source = StubSource::new(vec![cand]); + let recall = LocalDemandAlignedRecall::with_source(source.clone()); + let ctx = RecallContext::cold_start(sample_persona()); + + let pool = recall.recall(&sample_query(), &ctx).await.unwrap(); + + assert_eq!(source.fetch_count(), 1, "source.fetch must be called once"); + assert_eq!(pool.layers.len(), 1); + assert_eq!(pool.layers[0].0.0.as_uuid(), Uuid::from_u128(42)); + } + + /// What this catches: with_config_and_source preserves all + /// three (weights, half_life, source). PR-3d's working-set + /// walker uses this constructor when wiring with governor- + /// driven config. + #[tokio::test] + async fn with_config_and_source_preserves_all_three() { + let w = RecallScoreWeights::new(0.2, 0.2, 0.2, 0.2, 0.2).unwrap(); + let source = StubSource::new(Vec::new()); + let recall = LocalDemandAlignedRecall::with_config_and_source(w, 12345, source.clone()); + assert_eq!(*recall.weights(), w); + assert_eq!(recall.half_life_ms(), 12345); + + let ctx = RecallContext::cold_start(sample_persona()); + let _ = recall.recall(&sample_query(), &ctx).await.unwrap(); + assert_eq!(source.fetch_count(), 1, "source still wired"); + } + + /// What this catches: replay returns the typed + /// ScopeUnreachable refusal with a clear reason rather than + /// silently returning an empty pool. Per Joel's never-swallow- + /// errors rule — when the sentinel PR adds the RecallTraceStore, + /// this test flips to expect Ok(pool). + #[tokio::test] + async fn replay_returns_typed_not_implemented_refusal_in_pr3c() { + let recall = LocalDemandAlignedRecall::new(); + let trace = RecallTrace(ArtifactId::new(Uuid::nil())); + let result = recall.replay(&trace).await; + match result { + Err(RecallError::ScopeUnreachable { reason }) => { + assert!( + reason.contains("RecallTraceStore") || reason.contains("not yet implemented"), + "expected typed not-implemented reason, got: {reason}" + ); + } + other => panic!("expected ScopeUnreachable, got {other:?}"), + } + } }