From e574dab3c540590dc94e4589bf6f21ccd3f7ec64 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 17 Aug 2026 09:28:44 +0000 Subject: [PATCH] feat: show RankWeave rankings fail-closed (v0.75.0) After login, Rankings names rankweave_not_available when the port is down and lists accepted fused hits when RankWeave ranks visible posts. A hidden post is omitted. Never invent a fused score or a theta. Independent exact-head APPROVE required. Do not mix into #74 or #92. --- AGENTS.md | 4 +- ARCHITECTURE.md | 6 + .../0.75.0-rankweave-fusion-fail-closed.md | 9 + CHANGELOG.md | 10 + README.md | 2 +- backend/app/config.py | 8 + backend/app/main.py | 28 ++ backend/app/ranking_ingestion.py | 26 ++ backend/tests/test_config.py | 10 + docs/adr/0024-rankweave-fusion-fail-closed.md | 60 ++++ frontend/package.json | 2 +- frontend/src/App.test.tsx | 66 ++++ frontend/src/App.tsx | 61 ++++ frontend/src/api.ts | 17 + lineageweave/__init__.py | 2 +- lineageweave/rankweave_client.py | 304 ++++++++++++++++++ pyproject.toml | 2 +- tests/test_rankweave_client.py | 208 ++++++++++++ uv.lock | 2 +- 19 files changed, 821 insertions(+), 6 deletions(-) create mode 100644 CHANGELOG.d/0.75.0-rankweave-fusion-fail-closed.md create mode 100644 backend/app/ranking_ingestion.py create mode 100644 docs/adr/0024-rankweave-fusion-fail-closed.md create mode 100644 lineageweave/rankweave_client.py create mode 100644 tests/test_rankweave_client.py diff --git a/AGENTS.md b/AGENTS.md index e474b9ded..c790995c1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -30,7 +30,9 @@ reimplementing them: - [ThreadWeave](https://github.com/ContextualWisdomLab/ThreadWeave) for tree assembly (`reconstruct.py`'s `_walk`/`thread_messages` calls). - [RankWeave](https://github.com/ContextualWisdomLab/RankWeave) for - multi-channel score fusion (`weighted_convex_fuse`). + multi-channel score fusion (`weighted_convex_fuse` in + `reconstruct.py`) and the buyer-facing Rankings port + (`rankweave_client.py`) -- never invent a fused score or a theta. - [TEPP](https://github.com/ContextualWisdomLab/TEPP)'s published wire contract for calibrated measurement (`tepp_client.py`) -- never reimplement TEPP's model here. diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 5166e1e72..f8a83ceb1 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -66,6 +66,7 @@ flowchart LR | `adjudication_client.py` | Pluggable LLM-judgment channel (`Null` default, `ContextualOrchestrator` real impl) | | `image_content.py` | Pluggable vision channel: OCR + object recognition/tagging for embedded images (`Null` default, `OpenAiCompatibleVisionClient` real impl) | | `tepp_client.py` | TEPP's published `AnalysisRunRequest` wire contract, pluggable transport | +| `rankweave_client.py` | Fail-closed RankWeave ranking port (`weighted_reciprocal_rank_fuse` in-process; never invent a fused score or a theta) | | `reconstruct.py` | The pipeline: group → candidate window → score → fuse → thread | | `lineage_persistence.py` | Flattens reconstruct trees into `post_lineage_edge` row specs (parent, child, fused_score) | | `knowledge_graph.py` | Random-walk-with-restart relevance + per-node adaptive related-node cutoff (Tong et al., 2006) -- pure graph math, no Postgres | @@ -117,6 +118,11 @@ flowchart LR (`AnalysisRunRequest.to_json()` mirrors TEPP's published JSON Schema exactly, `additionalProperties: false` and all) so wiring in a real transport is additive, not a rewrite. +- **RankWeave is an in-process library, not an HTTP host.** + `rankweave_client.py`'s default transport raises + `RankWeaveNotAvailable`. `GET /api/rankings` then returns + `rankweave_not_available` and an empty ranking list. Hidden posts + are omitted from every channel. See ADR 0024. ## Standards and citations diff --git a/CHANGELOG.d/0.75.0-rankweave-fusion-fail-closed.md b/CHANGELOG.d/0.75.0-rankweave-fusion-fail-closed.md new file mode 100644 index 000000000..073075cdb --- /dev/null +++ b/CHANGELOG.d/0.75.0-rankweave-fusion-fail-closed.md @@ -0,0 +1,9 @@ +# 0.75.0 — Fail-closed RankWeave rankings + +## Added + +- Home Rankings panel fuses visible posts through `RankWeaveClient`. + After login with the port disabled or the library missing, Demo + Analyst sees **Rankings · RankWeave not available**. An accepted hit + lists the title; click opens that post. A hidden post is omitted. + Never invent a fused score or a theta. diff --git a/CHANGELOG.md b/CHANGELOG.md index e7ffaa1f3..6bfcaa28f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,16 @@ All notable changes to this project are documented here. Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.75.0] - 2026-08-17 + +### Added + +- Home Rankings panel fuses visible posts through `RankWeaveClient` + (ADR 0024). After login with the port disabled or the library + missing, Demo Analyst sees **Rankings · RankWeave not available**. + An accepted hit lists the title; click opens that post. A hidden + post is omitted. Never invent a fused score or a theta. + ## [0.71.2] - 2026-08-17 ### Added diff --git a/README.md b/README.md index 2f963c7f0..a3626d9f2 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ The optional LLM-adjudication channel calls [ThreadWeave](https://github.com/ContextualWisdomLab/ThreadWeave) (JWZ message threading) and channel fusion reuses [RankWeave](https://github.com/ContextualWisdomLab/RankWeave) (weighted -score fusion) -- both real dependencies, not reimplemented here. +score fusion for reconstruction and the fail-closed Rankings port) -- both real dependencies, not reimplemented here. ## Run it diff --git a/backend/app/config.py b/backend/app/config.py index 4202136a6..68cad3439 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -48,6 +48,10 @@ class Settings: # means the verification channel is unavailable, same "no fake # channel" discipline as every other pluggable client. searxng_base_url: str + # RankWeave ranking port (ADR 0024). True = fail-closed + # RankWeaveNotAvailable -- never invent a fused score. Default false + # uses the in-process library already required by reconstruct.py. + rankweave_disabled: bool @property def keycloak_jwks_uri(self) -> str: @@ -80,4 +84,8 @@ def load_settings() -> Settings: vision_model=os.environ.get("VISION_MODEL", ""), valkey_url=os.environ.get("VALKEY_URL", "redis://localhost:16379/0"), searxng_base_url=os.environ.get("SEARXNG_BASE_URL", ""), + rankweave_disabled=os.environ.get("RANKWEAVE_DISABLED", "") + .strip() + .lower() + in {"1", "true", "yes", "on"}, ) diff --git a/backend/app/main.py b/backend/app/main.py index 91ad406b0..27f679119 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -54,6 +54,7 @@ ) from lineageweave.post_summary import ContextualOrchestratorPostSummaryClient, NullPostSummaryClient from lineageweave.relation_verification import NullRelationVerificationClient, SearxngRelationVerificationClient +from lineageweave.rankweave_client import build_rankweave_client from backend.app.activity_stream import ( create_valkey_client, @@ -72,6 +73,7 @@ ingest_post_entity_relationships, ) from backend.app.post_evaluation_ingestion import fetch_post_evaluation, ingest_post_evaluation +from backend.app.ranking_ingestion import load_visible_ranking_posts from backend.app.report_ingestion import ( GROUPING_KINDS, fetch_period_comparison, @@ -235,6 +237,11 @@ def _post_evaluation_client(): ) +def _rankweave_client(): + """In-process RankWeave unless RANKWEAVE_DISABLED=1 (ADR 0024).""" + return build_rankweave_client(disabled=load_settings().rankweave_disabled) + + def _can_see_post(account: CurrentAccount, post: asyncpg.Record) -> bool: """ABAC: public rows are visible; private rows require same-corp affiliation.""" if post["visibility_code"] == "public": @@ -1122,3 +1129,24 @@ async def read_calendar( for c in visible: del c["visibility_code"], c["corporate_entity_id"] return {"commitments": visible} + + +@app.get("/api/rankings") +async def read_rankings( + account: CurrentAccount = Depends(get_current_account), + pool: asyncpg.Pool = Depends(get_pool), +) -> dict[str, Any]: + """RankWeave fusion of ABAC-visible posts (ADR 0024). + + Hidden posts are omitted from every channel. Never invents a fused + score or a theta. Fail-closed when RankWeave is disabled or the + library is missing. + """ + _require_post_read(account) + async with pool.acquire() as conn: + posts = await load_visible_ranking_posts( + conn, lambda row: _can_see_post(account, row) + ) + return _rankweave_client().as_api_payload( + posts, can_see_post=lambda _row: True + ) diff --git a/backend/app/ranking_ingestion.py b/backend/app/ranking_ingestion.py new file mode 100644 index 000000000..512b71273 --- /dev/null +++ b/backend/app/ranking_ingestion.py @@ -0,0 +1,26 @@ +"""Load ABAC-visible posts for the RankWeave ranking port. + +A hidden post is omitted from every channel. This module never invents +a fused score or a theta. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Callable, Mapping + +if TYPE_CHECKING: + import asyncpg + +__all__ = ["load_visible_ranking_posts"] + + +async def load_visible_ranking_posts( + conn: "asyncpg.Connection", + can_see_post: Callable[[Mapping[str, Any]], bool], +) -> list[dict[str, Any]]: + """Read ``source_post`` rows the buyer may rank.""" + posts = await conn.fetch( + "select post_id, post_title, created_at, visibility_code, " + "corporate_entity_id from source_post" + ) + return [dict(row) for row in posts if can_see_post(row)] diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py index 655f51654..c2f3994d5 100644 --- a/backend/tests/test_config.py +++ b/backend/tests/test_config.py @@ -21,3 +21,13 @@ def test_frontend_origins_are_parsed_from_comma_separated_env(monkeypatch) -> No def test_frontend_origins_drop_blank_entries(monkeypatch) -> None: monkeypatch.setenv("FRONTEND_ORIGINS", "http://localhost:5173,,") assert load_settings().frontend_origins == ["http://localhost:5173"] + + +def test_rankweave_disabled_defaults_off(monkeypatch) -> None: + monkeypatch.delenv("RANKWEAVE_DISABLED", raising=False) + assert load_settings().rankweave_disabled is False + + +def test_rankweave_disabled_flag_is_opt_in(monkeypatch) -> None: + monkeypatch.setenv("RANKWEAVE_DISABLED", "1") + assert load_settings().rankweave_disabled is True diff --git a/docs/adr/0024-rankweave-fusion-fail-closed.md b/docs/adr/0024-rankweave-fusion-fail-closed.md new file mode 100644 index 000000000..af1de902e --- /dev/null +++ b/docs/adr/0024-rankweave-fusion-fail-closed.md @@ -0,0 +1,60 @@ +# ADR 0024 — Fail-closed RankWeave ranking port + +**Decision status:** Accepted +**Date:** 2026-08-17 + +## Context + +LineageWeave already calls RankWeave inside `reconstruct.py` to fuse +per-candidate channel scores into a parent choice. Demo Analyst had no +buyer-facing Rankings surface over the same visible `source_post` rows. +RankWeave is an in-process library +([README](https://github.com/ContextualWisdomLab/RankWeave)): it does +not define HTTP, a mailbox host, or authentication. A missing package +or a disabled port must not become an invented fused score or a +calibrated theta (TEPP owns theta; see ADR 0022 on #214). + +This ADR does not replace `reconstruct.py`, does not read naruon +tables, and does not bind the demo IdP to production Keyverse. + +## Decision + +1. Consume RankWeave only through `RankWeaveClient`. The default + transport raises `RankWeaveNotAvailable`. `build_rankweave_client + (disabled=False)` uses `LibraryRankWeaveTransport`, which imports + `weighted_reciprocal_rank_fuse` inside the call so a missing + package fail-closes. +2. `GET /api/rankings` (`post_read`) loads ABAC-visible posts as two + rank-only channels: temporal (newest first) and lexical (token + overlap with the synthetic demo query `pricing quote delivery`). + Hidden posts are omitted from every channel. Never invent a score. +3. Fusion is weighted RRF with Cormack et al. (2009) η = 60 and + Samuel et al. (2025) unequal-channel weights (`temporal` 0.25, + `lexical` 0.75). The buyer sees 1-based `fused_rank` and the post + title — not a TEPP theta. +4. After login, Rankings sits above Calendar. Unavailable copy is + **Rankings · RankWeave not available**. An accepted hit lists the + title; click opens that `source_post`. + +## Consequences + +`RANKWEAVE_DISABLED=1` keeps the fail-closed transport. The default +seeded stack uses the in-process library already required by +`reconstruct.py`. Mailbox stays on ADR 0020 / #217. Conversations stay +on ADR 0021 / #219. Leftover pairs stay on #211. TEPP stays on #214. +Keyverse IdP remains a later slice. + +## References + +Cormack, G. V., Clarke, C. L. A., & Buettcher, S. (2009). Reciprocal +rank fusion outperforms condorcet and individual rank learning +methods. In *Proceedings of the 32nd international ACM SIGIR +conference on Research and development in information retrieval* +(pp. 758–759). ACM. https://doi.org/10.1145/1571941.1572114 + +Samuel, D., MacAvaney, S., Yates, A., Zhang, E., Zhang, S., +Macdonald, C., & Ounis, I. (2025). *Weighted reciprocal rank fusion +for multi-channel retrieval* [Preprint]. + +Contextual Wisdom Lab. (2026). *RankWeave* [Software documentation]. +https://github.com/ContextualWisdomLab/RankWeave diff --git a/frontend/package.json b/frontend/package.json index 49535e112..575b7c586 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "0.71.2", + "version": "0.75.0", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index bc84dd3fa..a32a26403 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -56,6 +56,15 @@ describe("App, authenticated", () => { function stubBackend(options?: { admin?: boolean; calendarCommitments?: unknown[]; + rankings?: { + status?: "accepted" | "unavailable"; + status_reason?: string | null; + rankings?: { + post_id: string; + post_title: string; + fused_rank: number; + }[]; + }; chatUnavailable?: boolean; searchUnavailable?: boolean; verificationEvidenceUrl?: string | null; @@ -204,6 +213,21 @@ describe("App, authenticated", () => { }), ); } + if (url.endsWith("/api/rankings")) { + const rankings = options?.rankings ?? { + status: "unavailable" as const, + status_reason: "rankweave_not_available", + rankings: [], + }; + return Promise.resolve( + jsonResponse({ + port: "rankweave", + status: rankings.status, + status_reason: rankings.status_reason, + rankings: rankings.rankings ?? [], + }), + ); + } if (url.includes("/api/reports/compare/") && method === "GET") { return Promise.resolve( jsonResponse({ @@ -1241,6 +1265,48 @@ describe("App, authenticated", () => { ); }); + it("names RankWeave unavailability on home rankings instead of inventing a fused score", async () => { + stubBackend(); + render(); + + expect(await screen.findByText("Rankings · RankWeave not available")).toBeInTheDocument(); + expect(screen.queryByText("Pricing renegotiation: revised quote sent")).not.toBeInTheDocument(); + }); + + it("opens an accepted ranking hit without inventing a fused score", async () => { + stubBackend({ + rankings: { + status: "accepted", + status_reason: null, + rankings: [ + { + post_id: "post-1", + post_title: "Public post", + fused_rank: 1, + }, + { + post_id: "post-2", + post_title: "Pricing renegotiation: revised quote sent", + fused_rank: 2, + }, + ], + }, + }); + render(); + + const rankingButton = await screen.findByRole("button", { + name: /open ranking: public post/i, + }); + expect(rankingButton).toHaveTextContent("Public post"); + expect(rankingButton).toHaveTextContent("Rankings · rankweave"); + expect(rankingButton).toHaveTextContent("rank 1"); + expect(screen.queryByRole("button", { name: /open ranking: private parent/i })).not.toBeInTheDocument(); + + await userEvent.click(rankingButton); + + await waitFor(() => expect(screen.getByText("The full body text.")).toBeInTheDocument()); + }); + it("shows upcoming commitments on the home page calendar and opens the post on click", async () => { stubBackend(); render(); diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index ff11eb163..6056e5eb4 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -25,6 +25,7 @@ import { fetchPeriodReportIndex, fetchPeriodReports, fetchPosts, + fetchRankings, fetchRelatedEntity, fetchRelatedKeymen, rebuildLineage, @@ -49,6 +50,7 @@ import { type PeriodReports, type PostLineage, type PostSummary, + type RankingList, type RelatedNode, type VocEvidence, } from "./api"; @@ -1311,6 +1313,64 @@ function PostDetailPopup({ ); } +function RankingsPanel({ + accessToken, + onSelectPost, +}: { + accessToken: string; + onSelectPost: (postId: string) => void; +}) { + const [ranking, setRanking] = useState(null); + const [error, setError] = useState(null); + + useEffect(() => { + setError(null); + fetchRankings(accessToken) + .then(setRanking) + .catch((err) => setError(String(err))); + }, [accessToken]); + + return ( +
+
+

Rankings

+ {ranking && ( + + {ranking.status === "accepted" + ? "rankweave" + : `rankweave · ${ranking.status_reason ?? "unavailable"}`} + + )} +
+ {error &&

{error}

} + {ranking === null && !error &&

Loading rankings...

} + {ranking && ranking.status === "unavailable" && ( +

Rankings · RankWeave not available

+ )} + {ranking && ranking.status === "accepted" && ranking.rankings.length === 0 && ( +

No fused rankings from RankWeave.

+ )} + {ranking && ranking.rankings.length > 0 && ( +
    + {ranking.rankings.map((hit) => ( +
  • + +
    • + ))} +
+ )} +
+ ); +} + function CalendarPanel({ accessToken, onSelectPost, @@ -1629,6 +1689,7 @@ function PostList({ accessToken }: { accessToken: string }) { return ( <> +
diff --git a/frontend/src/api.ts b/frontend/src/api.ts index a28fad274..e6dfcbad2 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -498,3 +498,20 @@ export function deriveCommitment(accessToken: string, postId: string): Promise { return backendFetch("/api/calendar", accessToken); } + +export interface RankedPost { + post_id: string; + post_title: string; + fused_rank: number; +} + +export interface RankingList { + port: string; + status: "accepted" | "unavailable"; + status_reason: string | null; + rankings: RankedPost[]; +} + +export function fetchRankings(accessToken: string): Promise { + return backendFetch("/api/rankings", accessToken); +} diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index 90076db27..1710c009e 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -35,4 +35,4 @@ "sentence_excerpts", ] -__version__ = "0.71.2" +__version__ = "0.75.0" diff --git a/lineageweave/rankweave_client.py b/lineageweave/rankweave_client.py new file mode 100644 index 000000000..eb0b3358b --- /dev/null +++ b/lineageweave/rankweave_client.py @@ -0,0 +1,304 @@ +"""Fail-closed adapter for RankWeave's in-process weighted RRF fusion. + +`RankWeave `_ is a +library, not an HTTP service. Reconstruction already calls +``weighted_convex_fuse`` inside ``reconstruct.py``; this module is the +only LineageWeave port that may call ``weighted_reciprocal_rank_fuse`` +for the buyer-facing Rankings surface. It never invents a fused score, +a theta, or a hidden post. + +The default transport raises :class:`RankWeaveNotAvailable` so a +disabled or missing library is fail-closed, the same discipline as +:class:`lineageweave.threadweave_client.ThreadWeaveNotAvailable`. +Wiring the in-process library is additive +(``LibraryRankWeaveTransport``), not a redesign. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Callable, Mapping, Sequence + +# Cormack et al. (2009) reciprocal-rank fusion constant. +DEFAULT_RANK_CONSTANT_ETA = 60 +DEFAULT_RANKING_LIMIT = 20 +DEFAULT_CHANNEL_WEIGHTS = {"temporal": 0.25, "lexical": 0.75} +# Seeded A-100 titles mention pricing, quote, and delivery. This is a +# synthetic demo query, not a customer string. +DEFAULT_RANKING_QUERY = "pricing quote delivery" + + +class RankWeaveNotAvailable(RuntimeError): + """Raised when the RankWeave ranking port is down or disabled.""" + + reason = "rankweave_not_available" + + +def _no_transport( + _channels: dict[str, list[str]], + _weights: dict[str, float], +) -> list[dict[str, Any]]: + raise RankWeaveNotAvailable( + "rankweave_not_available: RankWeave ranking port is not configured. " + "Pass RANKWEAVE_DISABLED=0 (default) or a transport= callable. " + "Never invent a fused score." + ) + + +def _import_rankweave() -> Any: + """Import RankWeave at call time so a missing package fail-closes.""" + import rankweave as rw + + return rw + + +def _as_datetime(value: object) -> datetime: + if isinstance(value, datetime): + return value + text = str(value or "").strip() + if not text: + return datetime.min + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + return datetime.fromisoformat(text) + except ValueError: + return datetime.min + + +def _token_overlap(title: str, query: str) -> int: + query_tokens = {token for token in query.casefold().split() if token} + title_tokens = {token for token in title.casefold().split() if token} + return len(query_tokens & title_tokens) + + +def ranking_channels_from_rows( + posts: Sequence[Mapping[str, Any]], + can_see_post: Callable[[Mapping[str, Any]], bool], + query: str = DEFAULT_RANKING_QUERY, +) -> dict[str, list[str]]: + """Project ABAC-visible posts into rank-only RankWeave channels. + + Hidden posts are omitted from every channel. Missing cells stay + missing: a post never receives an invented score. Temporal ranks + newest first; lexical ranks by token overlap with the demo query. + """ + visible: list[Mapping[str, Any]] = [] + for row in posts: + if not can_see_post(row): + continue + title = str(row.get("post_title") or "").strip() + post_id = str(row.get("post_id") or "").strip() + if not title or not post_id: + continue + visible.append(row) + temporal = sorted( + visible, + key=lambda row: ( + _as_datetime(row.get("created_at")), + str(row.get("post_id") or ""), + ), + reverse=True, + ) + lexical = sorted( + visible, + key=lambda row: ( + -_token_overlap(str(row.get("post_title") or ""), query), + str(row.get("post_title") or "").casefold(), + str(row.get("post_id") or ""), + ), + ) + return { + "temporal": [str(row["post_id"]) for row in temporal], + "lexical": [str(row["post_id"]) for row in lexical], + } + + +@dataclass(frozen=True) +class RankedPost: + """One visible fused hit. Rank is 1-based position, never a theta.""" + + post_id: str + post_title: str + fused_rank: int + + def to_json(self) -> dict[str, Any]: + return { + "post_id": self.post_id, + "post_title": self.post_title, + "fused_rank": self.fused_rank, + } + + +@dataclass(frozen=True) +class RankingList: + """Accepted ranking projection. Empty when RankWeave returned no hits.""" + + items: tuple[RankedPost, ...] + + def to_json(self) -> list[dict[str, Any]]: + return [item.to_json() for item in self.items] + + +def _item_id_from_hit(hit: object) -> str: + if isinstance(hit, Mapping): + return str(hit.get("item_id") or hit.get("post_id") or "").strip() + item_id = getattr(hit, "item_id", None) + if item_id is not None: + return str(item_id).strip() + if isinstance(hit, (tuple, list)) and hit: + return str(hit[0]).strip() + return "" + + +def project_ranking_list( + raw: object, + titles_by_id: Mapping[str, str], +) -> RankingList: + """Accept transport output. Unknown shapes fail closed. Hidden ids drop.""" + if not isinstance(raw, list): + raise RankWeaveNotAvailable( + "rankweave_not_available: ranking envelope is not a hit list" + ) + items: list[RankedPost] = [] + seen: set[str] = set() + for hit in raw: + post_id = _item_id_from_hit(hit) + title = str(titles_by_id.get(post_id) or "").strip() + if not post_id or not title or post_id in seen: + continue + seen.add(post_id) + items.append( + RankedPost( + post_id=post_id, + post_title=title, + fused_rank=len(items) + 1, + ) + ) + return RankingList(items=tuple(items)) + + +class LibraryRankWeaveTransport: + """Call RankWeave ``weighted_reciprocal_rank_fuse`` in-process.""" + + def __call__( + self, + channels: dict[str, list[str]], + weights: dict[str, float], + ) -> list[dict[str, Any]]: + try: + rw = _import_rankweave() + except ImportError as exc: + raise RankWeaveNotAvailable( + "rankweave_not_available: rankweave package is not installed. " + "Never invent a fused score." + ) from exc + usable = { + name: [item_id for item_id in ranks if str(item_id).strip()] + for name, ranks in channels.items() + if ranks + } + if not usable: + return [] + active_weights = { + name: weights[name] for name in usable if name in weights and weights[name] > 0 + } + if not active_weights: + raise RankWeaveNotAvailable( + "rankweave_not_available: no positive channel weights remain" + ) + try: + hits = rw.weighted_reciprocal_rank_fuse( + usable, + active_weights, + limit=DEFAULT_RANKING_LIMIT, + rank_constant_eta=DEFAULT_RANK_CONSTANT_ETA, + ) + except TypeError: + try: + hits = rw.weighted_reciprocal_rank_fuse( + usable, + active_weights, + limit=DEFAULT_RANKING_LIMIT, + ) + except Exception as exc: + raise RankWeaveNotAvailable( + f"rankweave_not_available: weighted_reciprocal_rank_fuse failed ({exc})" + ) from exc + except Exception as exc: + raise RankWeaveNotAvailable( + f"rankweave_not_available: weighted_reciprocal_rank_fuse failed ({exc})" + ) from exc + projected: list[dict[str, Any]] = [] + for hit in hits: + item_id = _item_id_from_hit(hit) + if item_id: + projected.append({"item_id": item_id}) + return projected + + +def build_rankweave_client(disabled: bool = False) -> "RankWeaveClient": + """``disabled=True`` keeps the default fail-closed transport.""" + if disabled: + return RankWeaveClient() + return RankWeaveClient(transport=LibraryRankWeaveTransport()) + + +class RankWeaveClient: + """Fuses visible-post channels through a pluggable RankWeave transport.""" + + def __init__( + self, + transport: Callable[ + [dict[str, list[str]], dict[str, float]], list[dict[str, Any]] + ] = _no_transport, + ) -> None: + self._transport = transport + + def fuse_rankings( + self, + channels: dict[str, list[str]], + titles_by_id: Mapping[str, str], + weights: dict[str, float] | None = None, + ) -> RankingList: + try: + raw = self._transport(channels, weights or DEFAULT_CHANNEL_WEIGHTS) + except RankWeaveNotAvailable: + raise + except Exception as exc: + raise RankWeaveNotAvailable( + f"rankweave_not_available: ranking transport failed ({exc})" + ) from exc + return project_ranking_list(raw, titles_by_id) + + def as_api_payload( + self, + posts: Sequence[Mapping[str, Any]], + can_see_post: Callable[[Mapping[str, Any]], bool], + query: str = DEFAULT_RANKING_QUERY, + ) -> dict[str, Any]: + """Buyer-visible ranking status. Never invents a fused score.""" + channels = ranking_channels_from_rows(posts, can_see_post, query=query) + titles_by_id = { + post_id: str(row.get("post_title") or "").strip() + for row in posts + if can_see_post(row) + for post_id in [str(row.get("post_id") or "").strip()] + if post_id and str(row.get("post_title") or "").strip() + } + try: + ranking = self.fuse_rankings(channels, titles_by_id) + except RankWeaveNotAvailable: + return { + "port": "rankweave", + "status": "unavailable", + "status_reason": RankWeaveNotAvailable.reason, + "rankings": [], + } + return { + "port": "rankweave", + "status": "accepted", + "status_reason": None, + "rankings": ranking.to_json(), + } diff --git a/pyproject.toml b/pyproject.toml index 63d236886..764ebad72 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "0.71.2" +version = "0.75.0" description = "Reconstructs git-branch-style lineage DAGs from scattered short records using multi-channel score fusion and LLM adjudication." readme = "README.md" license = { text = "MIT" } diff --git a/tests/test_rankweave_client.py b/tests/test_rankweave_client.py new file mode 100644 index 000000000..64c5e70b1 --- /dev/null +++ b/tests/test_rankweave_client.py @@ -0,0 +1,208 @@ +"""Fail-closed RankWeave ranking port. + +RankWeave is an in-process weighted-RRF library. LineageWeave fuses +only visible posts. A hidden post is omitted from every channel. The +client never invents a fused score or a theta. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from lineageweave.rankweave_client import ( + LibraryRankWeaveTransport, + RankWeaveClient, + RankWeaveNotAvailable, + build_rankweave_client, + project_ranking_list, + ranking_channels_from_rows, +) + + +PUBLIC = { + "post_id": "post-1", + "post_title": "Public post", + "created_at": "2026-01-05T00:00:00Z", +} +QUOTE = { + "post_id": "post-2", + "post_title": "Pricing renegotiation: revised quote sent", + "created_at": "2026-01-09T00:00:00Z", +} +HIDDEN = { + "post_id": "hidden-parent", + "post_title": "Private parent", + "created_at": "2026-01-10T00:00:00Z", +} + + +def test_default_transport_fails_closed() -> None: + client = RankWeaveClient() + with pytest.raises(RankWeaveNotAvailable, match="rankweave_not_available"): + client.fuse_rankings( + {"temporal": ["post-1"], "lexical": ["post-1"]}, + {"post-1": "Public post"}, + ) + + +def test_default_payload_never_invents_a_fused_score() -> None: + payload = RankWeaveClient().as_api_payload( + [PUBLIC, QUOTE], + can_see_post=lambda _row: True, + ) + + assert payload == { + "port": "rankweave", + "status": "unavailable", + "status_reason": "rankweave_not_available", + "rankings": [], + } + + +def test_disabled_factory_fails_closed() -> None: + client = build_rankweave_client(disabled=True) + payload = client.as_api_payload([PUBLIC], can_see_post=lambda _row: True) + assert payload["status"] == "unavailable" + assert payload["rankings"] == [] + + +def test_library_transport_fails_closed_when_rankweave_missing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def boom() -> object: + raise ImportError("No module named 'rankweave'") + + monkeypatch.setattr("lineageweave.rankweave_client._import_rankweave", boom) + client = RankWeaveClient(transport=LibraryRankWeaveTransport()) + with pytest.raises(RankWeaveNotAvailable, match="rankweave_not_available"): + client.fuse_rankings( + {"temporal": ["post-1"], "lexical": ["post-1"]}, + {"post-1": "Public post"}, + ) + assert ( + client.as_api_payload([PUBLIC], can_see_post=lambda _row: True)["rankings"] + == [] + ) + + +def test_library_transport_fails_closed_when_fuse_raises( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class FakeRw: + @staticmethod + def weighted_reciprocal_rank_fuse(*_args: object, **_kwargs: object) -> list: + raise RuntimeError("duplicate identifiers") + + monkeypatch.setattr( + "lineageweave.rankweave_client._import_rankweave", lambda: FakeRw + ) + client = RankWeaveClient(transport=LibraryRankWeaveTransport()) + with pytest.raises(RankWeaveNotAvailable, match="rankweave_not_available"): + client.fuse_rankings( + {"temporal": ["post-1"], "lexical": ["post-1"]}, + {"post-1": "Public post"}, + ) + + +def test_injected_transport_returns_accepted_hits() -> None: + def fake_transport( + _channels: dict[str, list[str]], + _weights: dict[str, float], + ) -> list[dict]: + return [{"item_id": "post-2"}, {"item_id": "post-1"}] + + payload = RankWeaveClient(transport=fake_transport).as_api_payload( + [PUBLIC, QUOTE], + can_see_post=lambda _row: True, + ) + + assert payload["status"] == "accepted" + assert payload["status_reason"] is None + assert payload["rankings"] == [ + { + "post_id": "post-2", + "post_title": "Pricing renegotiation: revised quote sent", + "fused_rank": 1, + }, + { + "post_id": "post-1", + "post_title": "Public post", + "fused_rank": 2, + }, + ] + + +def test_library_transport_projects_monkeypatched_rrf( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, object] = {} + + class FakeRw: + @staticmethod + def weighted_reciprocal_rank_fuse( + channels: dict[str, list[str]], + weights: dict[str, float], + limit: int = 20, + rank_constant_eta: int = 60, + ) -> list: + captured["channels"] = channels + captured["weights"] = weights + captured["limit"] = limit + captured["eta"] = rank_constant_eta + return [ + SimpleNamespace(item_id="post-2"), + SimpleNamespace(item_id="post-1"), + ] + + monkeypatch.setattr( + "lineageweave.rankweave_client._import_rankweave", lambda: FakeRw + ) + payload = RankWeaveClient(transport=LibraryRankWeaveTransport()).as_api_payload( + [PUBLIC, QUOTE], + can_see_post=lambda _row: True, + ) + + assert captured["eta"] == 60 + assert captured["weights"]["lexical"] == 0.75 + assert payload["rankings"][0]["post_title"] == ( + "Pricing renegotiation: revised quote sent" + ) + assert payload["rankings"][0]["fused_rank"] == 1 + + +def test_hidden_post_is_omitted_from_every_channel() -> None: + channels = ranking_channels_from_rows( + [PUBLIC, QUOTE, HIDDEN], + can_see_post=lambda row: str(row["post_id"]) != "hidden-parent", + ) + + assert "hidden-parent" not in channels["temporal"] + assert "hidden-parent" not in channels["lexical"] + assert channels["temporal"][0] == "post-2" + assert channels["lexical"][0] == "post-2" + + +def test_lexical_channel_ranks_quote_ahead_of_generic_title() -> None: + channels = ranking_channels_from_rows( + [PUBLIC, QUOTE], + can_see_post=lambda _row: True, + query="pricing quote delivery", + ) + assert channels["lexical"][0] == "post-2" + + +def test_unknown_envelope_fails_closed() -> None: + with pytest.raises(RankWeaveNotAvailable, match="rankweave_not_available"): + project_ranking_list({"hits": [{"item_id": "spoofed"}]}, {"spoofed": "x"}) + + +def test_unknown_hit_id_is_dropped_not_repaired() -> None: + ranking = project_ranking_list( + [{"item_id": "invented"}, {"item_id": "post-2"}], + {"post-2": "Pricing renegotiation: revised quote sent"}, + ) + assert [item.post_id for item in ranking.items] == ["post-2"] + assert all(item.post_id != "invented" for item in ranking.items) + assert ranking.items[0].fused_rank == 1 diff --git a/uv.lock b/uv.lock index 76a4a7d6a..08eab7768 100644 --- a/uv.lock +++ b/uv.lock @@ -355,7 +355,7 @@ wheels = [ [[package]] name = "lineageweave" -version = "0.71.2" +version = "0.75.0" source = { virtual = "." } dependencies = [ { name = "certifi" },