feat(*): expose the RRF damping constant and rebalance skill source fusion

CONTEXT.md

@@ -414,7 +414,7 @@ A remote OpenAPI skill marketplace, configured via `skillForge.router.hub` (`end
 `api_key` / `timeout_s` / `min_safety`; `endpoint=None` disables it). `SkillHubClient` offers
 progressive disclosure — `search()` (metadata-only discovery), `get()` (skill body),
 `install()` (download + safe extract); during routing `HubSkillSource` feeds metadata-only
-candidates into the weighted RRF (weight 0.85, below Local 1.0 and Everos 0.9), and the
+candidates into the weighted RRF (weight 0.85, below Local 0.96 and Everos 0.9), and the
 `read_skill` / `use_skill` tools do on-demand body fetch / script materialization. Replaces
 the retired "Mass" source.
 

raven/config/raven.py

@@ -1173,15 +1173,26 @@ class SkillForgeRouterConfig(_Base):
 
     weights: dict[str, float] = Field(
         default_factory=lambda: {
-            "local": 1.0,
+            "local": 0.96,
             "everos": 0.9,
             "hub": 0.85,
         },
     )
     """Per-source RRF weight. Higher = more rank mass when the same skill
     surfaces from multiple sources. Local highest (hand-curated); Hub
     (the remote marketplace, replaces the retired Mass source) lowest as
-    imported/unvalidated; Everos in between (task-specific, auto-evolved)."""
+    imported/unvalidated; Everos in between (task-specific, auto-evolved).
+
+    Only the ratios matter -- scaling all three leaves the order unchanged.
+    Read them together with ``rrf_k``: the spread has to stay well inside
+    the rank ladder that ``rrf_k`` produces, or weight silently overrides
+    rank and each source becomes a strict tier."""
+
+    rrf_k: int = Field(default=10, ge=1)
+    """RRF damping constant, mirroring ``skill_forge.fusion.RRF_K``.
+    Lower = source-internal rank carries more weight relative to
+    ``weights``; higher = flatter, so cross-source agreement and source
+    identity dominate."""
 
     over_fetch_factor: int = 2
     """Each source is asked for ``top_k * factor`` hits before fusion

raven/context_engine/factory.py

@@ -211,6 +211,7 @@ def _build_router(
         sources=sources,
         over_fetch_factor=skill_forge_router_config.over_fetch_factor,
         dedup_by=skill_forge_router_config.dedup_by,
+        rrf_k=skill_forge_router_config.rrf_k,
     )
 
 

raven/memory_engine/skill_forge/fusion.py

@@ -1,15 +1,18 @@
 """Weighted Reciprocal Rank Fusion across heterogeneous skill sources.
 
-The classic RRF formula sums ``1 / (k + rank_i(d))`` over the sources
-that hit document ``d``. Multi-source skill retrieval needs a small
-extension: each source carries a **trust weight** so curated content
-(Local) outranks imported content (Mass) at equal rank. The weighted
-form is::
+The classic RRF formula sums ``1 / (rrf_k + rank_i(d))`` over the
+sources that hit document ``d``. Multi-source skill retrieval needs a
+small extension: each source carries a **trust weight** so curated
+content (Local) outranks imported content (Mass) at equal rank. The
+weighted form is::
 
-    rrf_score(d) = Σ_i  w_i / (k + rank_i(d))
+    rrf_score(d) = Σ_i  w_i / (rrf_k + rank_i(d))
 
-with ``k = 60`` (the long-standing RRF constant) and the per-source
-``w_i`` coming from the source's :attr:`SkillSource.weight` attribute.
+with ``rrf_k`` defaulting to :data:`RRF_K` and overridable per call or
+via ``skillForge.router.rrfK``, and the per-source ``w_i`` coming from
+the source's :attr:`SkillSource.weight` attribute. Note that ``rrf_k``
+is the damping constant, distinct from the ``k`` argument of
+:func:`rrf_merge_weighted`, which caps the output length.
 
 Three additional behaviors are baked in:
 
@@ -37,16 +40,18 @@
 
 from raven.memory_engine.skill_forge.types import RouterHit
 
-# The "60" in classic RRF — dampens rank effects so a #1 at one source
-# doesn't always crowd out top-3 from another. Standard value, kept as
-# a module constant so the rare experiment that wants to tune it can.
-RRF_K: int = 60
+# Classic RRF uses 60, tuned for TREC-scale runs of ~1000 hits. Sources
+# here return ~10, where 60 flattens the whole rank ladder to a 15% score
+# spread -- narrower than the 1.0/0.85 source-weight gap, so weight alone
+# decides the order and rank stops mattering. 10 keeps that ladder at 82%.
+RRF_K: int = 10
 
 
 def rrf_merge_weighted(
     source_results: list[tuple[str, float, list[RouterHit]]],
     k: int,
     dedup_by: str = "name",
+    rrf_k: int | None = None,
 ) -> list[RouterHit]:
     """Fuse per-source ranked lists into one top-K.
 
@@ -60,21 +65,24 @@ def rrf_merge_weighted(
             sources surfacing a skill with the same display name are
             one logical skill. Tests pass ``"qualified_id"`` when they
             want to verify "no dedup happened" on disjoint hits.
+        rrf_k: RRF damping constant. ``None`` uses :data:`RRF_K`. Not to
+            be confused with ``k`` above, which caps the output length.
 
     Returns:
         Up to ``k`` :class:`RouterHit` records, ranked by descending
         RRF score. Each has ``rrf_score`` (float) and
         ``contributing_sources`` (list[str], stable-ordered as
         encountered) added to its ``meta``.
     """
+    damping = RRF_K if rrf_k is None else rrf_k
     rrf_scores: dict[str, float] = defaultdict(float)
     best_hit: dict[str, RouterHit] = {}
     contributing: dict[str, list[str]] = defaultdict(list)
 
     for source_name, weight, hits in source_results:
         for rank, hit in enumerate(hits, start=1):
             key = getattr(hit, dedup_by)
-            rrf_scores[key] += weight / (RRF_K + rank)
+            rrf_scores[key] += weight / (damping + rank)
             contributing[key].append(source_name)
             prev = best_hit.get(key)
             # Keep the hit with the higher per-source ``score`` as the

raven/memory_engine/skill_forge/router.py

@@ -43,6 +43,7 @@ def __init__(
         *,
         over_fetch_factor: int = 2,
         dedup_by: str = "name",
+        rrf_k: int | None = None,
     ) -> None:
         # The list is captured by reference; callers should pass an
         # already-frozen tuple if they want to forbid mutation. We
@@ -52,6 +53,7 @@ def __init__(
         self._sources = sources
         self._over_fetch_factor = max(1, over_fetch_factor)
         self._dedup_by = dedup_by
+        self._rrf_k = rrf_k
 
     async def select(
         self,
@@ -66,6 +68,7 @@ async def select(
             [(s.name, s.weight, hits) for s, hits in zip(self._sources, per_source)],
             k=k,
             dedup_by=self._dedup_by,
+            rrf_k=self._rrf_k,
         )
 
     async def _safe_search(

tests/test_config_raven_sections.py

@@ -46,10 +46,11 @@ def test_memory_backend_none_disables(self) -> None:
     def test_skill_router_defaults(self) -> None:
         c = SkillForgeRouterConfig()
         assert c.enabled is True
-        assert c.weights == {"local": 1.0, "everos": 0.9, "hub": 0.85}
+        assert c.weights == {"local": 0.96, "everos": 0.9, "hub": 0.85}
         assert c.over_fetch_factor == 2
         assert c.dedup_by == "name"
         assert c.top_k == 5
+        assert c.rrf_k == 10
         # Hub is the remote source (replaces the retired Mass source);
         # disabled until an endpoint is set.
         assert isinstance(c.hub, HubSourceConfig)
@@ -58,6 +59,11 @@ def test_skill_router_defaults(self) -> None:
         assert c.hub.timeout_s == pytest.approx(2.0)
         assert c.hub.min_safety == pytest.approx(0.7)
 
+    def test_rrf_k_accepts_camel_case_and_rejects_zero(self) -> None:
+        assert SkillForgeRouterConfig(rrfK=30).rrf_k == 30
+        with pytest.raises(ValidationError):
+            SkillForgeRouterConfig(rrf_k=0)
+
     def test_skill_forge_public_defaults(self) -> None:
         c = SkillForgeConfig()
         assert c.embedding_model == "default"

tests/test_config_update.py

@@ -279,7 +279,7 @@ def test_init_extension_defaults_seeds_safe_subset(cfg_path: Path) -> None:
     assert data["skillForge"]["enabled"] is True
     assert data["skillForge"]["everos"] == {"enabled": True}
     assert data["skillForge"]["router"]["weights"] == {
-        "local": 1.0,
+        "local": 0.96,
         "everos": 0.9,
         "hub": 0.85,
     }

tests/test_context_engine_factory.py

@@ -85,6 +85,7 @@ def _build_engine(
     memory_config: MemoryConfig | None = None,
     model: str = "stub",
     skill_forge_config: SkillForgeConfig | None = None,
+    rrf_k: int | None = None,
 ) -> ContextAssembler:
     builder = ContextBuilder(workspace=tmp_path)
     engine = build_context_engine(
@@ -99,6 +100,7 @@ def _build_engine(
         memory_config=memory_config or MemoryConfig(),
         skill_forge_router_config=SkillForgeRouterConfig(
             hub=HubSourceConfig(endpoint=hub_endpoint),
+            **({} if rrf_k is None else {"rrf_k": rrf_k}),
         ),
         skill_forge_config=skill_forge_config,
     )
@@ -190,6 +192,16 @@ def test_hub_source_present_when_endpoint_set(self, tmp_path: Path) -> None:
         types, _ = _router_sources(_build_engine(tmp_path, backend=_FakeBackend(), hub_endpoint="http://hub.test"))
         assert HubSkillSource in types
 
+    def test_rrf_k_forwarded_from_config(self, tmp_path: Path) -> None:
+        engine = _build_engine(tmp_path, backend=_FakeBackend(), rrf_k=25)
+        skills = next(b for b in engine._builders if isinstance(b, SkillsSegmentBuilder))
+        assert skills._router._rrf_k == 25
+
+    def test_rrf_k_defaults_to_config_default(self, tmp_path: Path) -> None:
+        engine = _build_engine(tmp_path, backend=_FakeBackend())
+        skills = next(b for b in engine._builders if isinstance(b, SkillsSegmentBuilder))
+        assert skills._router._rrf_k == SkillForgeRouterConfig().rrf_k
+
     def test_track_ids_from_memory_config(self, tmp_path: Path) -> None:
         engine = _build_engine(
             tmp_path,

tests/test_skill_router_fusion.py

@@ -6,7 +6,9 @@
 
 import pytest
 
+from raven.config.raven import SkillForgeRouterConfig
 from raven.memory_engine.skill_forge import (
+    RRF_K,
     RouterHit,
     SkillForgeRouter,
     rrf_merge_weighted,
@@ -82,8 +84,25 @@ def test_contributing_sources_recorded(self) -> None:
     def test_rrf_score_written_to_meta(self) -> None:
         hits = [_hit("a/1", "x")]
         out = rrf_merge_weighted([("a", 1.0, hits)], k=5)
-        # RRF score for rank-1 of single source with w=1: 1/(60+1) ~= 0.01639
-        assert out[0].meta["rrf_score"] == pytest.approx(1.0 / 61.0)
+        # RRF score for rank-1 of single source with w=1: 1/(RRF_K+1)
+        assert out[0].meta["rrf_score"] == pytest.approx(1.0 / (RRF_K + 1))
+
+    def test_rrf_k_override_steepens_rank_ladder(self) -> None:
+        """A larger rrf_k flattens rank differences until source weight
+        alone decides the order; the default keeps rank competitive."""
+        local = [
+            _hit("local/a", "alpha"),
+            _hit("local/b", "beta"),
+            _hit("local/c", "gamma"),
+        ]
+        hub = [_hit("hub/d", "delta")]
+        args = [("local", 1.0, local), ("hub", 0.85, hub)]
+
+        names = [h.name for h in rrf_merge_weighted(args, k=5)]
+        assert names.index("delta") < names.index("gamma")
+
+        names = [h.name for h in rrf_merge_weighted(args, k=5, rrf_k=60)]
+        assert names.index("gamma") < names.index("delta")
 
     def test_weight_affects_relative_ranking(self) -> None:
         """A hit ranked #3 in a high-weight source can outrank a hit
@@ -101,7 +120,7 @@ def test_weight_affects_relative_ranking(self) -> None:
             k=5,
         )
         names = [h.name for h in out]
-        # local rank-3 (1/63 = ~0.0159) > mass rank-1 (0.01/61 = ~0.00016)
+        # local rank-3 (1/(RRF_K+3)) > mass rank-1 (0.01/(RRF_K+1)),
         # so 'gamma' must outrank 'delta'.
         assert names.index("gamma") < names.index("delta")
 
@@ -178,6 +197,40 @@ async def test_fans_out_to_all_sources(self) -> None:
         names = {h.name for h in out}
         assert names == {"x", "y"}
 
+    async def test_rrf_k_reaches_the_fusion(self) -> None:
+        """The constructor arg must survive the hop into rrf_merge_weighted.
+        Asserting the exact score catches a dropped pass-through, which
+        would otherwise silently fall back to the module default."""
+        a = _StubSource("local", 1.0, [_hit("local/x", "x")])
+
+        out = await SkillForgeRouter([a], rrf_k=60).select("q", history=[], k=5)
+        assert out[0].meta["rrf_score"] == pytest.approx(1.0 / 61.0)
+
+        out = await SkillForgeRouter([a]).select("q", history=[], k=5)
+        assert out[0].meta["rrf_score"] == pytest.approx(1.0 / (RRF_K + 1))
+
+    async def test_shipped_defaults_interleave_the_three_sources(self) -> None:
+        """The ordering the shipped config produces, asserted rather than
+        described: every source places its top hit and Local keeps slot 1."""
+        cfg = SkillForgeRouterConfig()
+        sources = [
+            _StubSource(
+                name,
+                cfg.weights[name],
+                [_hit(f"{name}/1", f"{name}#1"), _hit(f"{name}/2", f"{name}#2")],
+            )
+            for name in ("local", "everos", "hub")
+        ]
+        router = SkillForgeRouter(sources, rrf_k=cfg.rrf_k)
+        out = await router.select("q", history=[], k=5)
+        assert [h.name for h in out] == [
+            "local#1",
+            "everos#1",
+            "local#2",
+            "hub#1",
+            "everos#2",
+        ]
+
     async def test_over_fetches_per_source(self) -> None:
         """Default over_fetch_factor=2 means each source is asked for k*2."""
         a = _StubSource("local", 1.0, [_hit("local/x", "x")])