diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index 4b91a9990..a5c0ce35a 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -27,7 +27,10 @@ latest_outbox_delivery_is_delivered, outbox_request_digest, ) -from backend.app.lineage_ingestion import records_from_source_posts +from backend.app.lineage_ingestion import ( + load_estimated_channel_weights, + records_from_source_posts, +) from lineageweave.adjudication_client import AdjudicationClient from lineageweave.http_client import HttpClientError, post_json from lineageweave.lineage_persistence import lineage_edge_specs @@ -49,6 +52,34 @@ class AnalysisRunStartError(AnalysisRunCreateError): """Fail-closed start: HTTP status plus a next-action detail string.""" +class _AdjudicationProviderError(RuntimeError): + """The adjudication provider (not our code) failed during a judge call.""" + + +class _ProviderBoundaryAdjudication: + """Convert judge-call provider failures into a typed boundary error. + + ``post_json`` raises ``HttpClientError``/``OSError`` on transport and + HTTP failures and the content extraction raises + ``ValueError``/``TypeError`` on malformed provider envelopes; those + same builtin types raised by our reconstruction code would be real + bugs, so the conversion happens only inside ``judge`` -- the one + place a provider is actually on the line. + """ + + available = True + + def __init__(self, inner) -> None: + self._inner = inner + + def judge(self, candidate_label: str, record_label: str) -> float: + """Score one candidate pair, typing provider failures as such.""" + try: + return self._inner.judge(candidate_label, record_label) + except (HttpClientError, OSError, ValueError, TypeError) as exc: + raise _AdjudicationProviderError(str(exc)) from exc + + def reconstruction_result_digest(edges: list[Edge]) -> str: """SHA-256 of the ordered parent choices. Never hashes a post body.""" material = json.dumps( @@ -716,7 +747,42 @@ async def _deliver_lineage_reconstruction( knowledge_cutoff=locked["knowledge_cutoff"], affiliated_entity_ids=affiliated_entity_ids, ) - edges = lineage_edge_specs(records_from_source_posts(rows), llm=adjudication_client) + # ADR 0200 point 1: weights are measurement output only -- the + # activated fast-mlsirm set matching this run's active channels + # (four when the adjudication client is available, three + # deterministic otherwise). A missing set fails the start with a + # next-action message instead of reconstructing on constants. + active_channels = {"temporal", "secondary_key", "text"} + if adjudication_client is not None and getattr(adjudication_client, "available", False): + active_channels = active_channels | {"llm"} + weights = await load_estimated_channel_weights(conn, active_channels) + if weights is None: + raise AnalysisRunStartError( + 503, + "Channel weights are not estimated yet for this run's active " + f"channels ({', '.join(sorted(active_channels))}). Run " + "scripts/estimate_channel_weights.py, then start this run again.", + ) + # The provider boundary is exactly llm.judge() -- wrapping the whole + # pipeline would disguise a reconstruction code bug as a retryable + # provider outage. The enclosing transaction rolls back either way, + # so no partial graph persists (issue #289 RED 4/6). + guarded_client = ( + _ProviderBoundaryAdjudication(adjudication_client) + if "llm" in active_channels + else adjudication_client + ) + try: + edges = lineage_edge_specs( + records_from_source_posts(rows), llm=guarded_client, weights=weights + ) + except _AdjudicationProviderError as exc: + raise AnalysisRunStartError( + 503, + "The adjudication provider failed mid-reconstruction; nothing " + "was persisted. Check the contextual-orchestrator transport, " + "then start this run again.", + ) from exc digest = reconstruction_result_digest(edges) finished = datetime.now(timezone.utc) if finished < now: diff --git a/backend/app/analysis_run_worker.py b/backend/app/analysis_run_worker.py index 43b8d17b7..166810e13 100644 --- a/backend/app/analysis_run_worker.py +++ b/backend/app/analysis_run_worker.py @@ -7,6 +7,8 @@ from __future__ import annotations +import logging + import asyncpg import redis.asyncio as redis from uuid import UUID @@ -14,9 +16,12 @@ from lineageweave.adjudication_client import AdjudicationClient from lineageweave.tepp_client import TeppClient +from backend.app.analysis_run_ingestion import AnalysisRunCreateError from backend.app.analysis_run_outbox import OUTBOX_STREAM_KEY from backend.app.analysis_run_start import deliver_queued_analysis_run +logger = logging.getLogger(__name__) + async def consume_analysis_run_stream_once( client: redis.Redis, @@ -40,26 +45,40 @@ async def consume_analysis_run_stream_once( except ValueError: analysis_run_id = "" if analysis_run_id: - async with pool.acquire() as conn: - async with conn.transaction(): - owner = await conn.fetchrow( - """ - select requested_by_account_id - from analysis_run - where analysis_run_id = $1::uuid - """, - analysis_run_id, - ) - if owner is not None: - await deliver_queued_analysis_run( - conn, - analysis_run_id=analysis_run_id, - account_id=str(owner["requested_by_account_id"]), - affiliated_entity_ids=[], - tepp_client=tepp_client, - adjudication_client=adjudication_client, - valkey_stream_entry_id=str(entry_id), + # One run's fail-closed refusal (404/409/503, e.g. channel + # weights not estimated yet, ADR 0145) must not end the + # worker task and halt every later run's delivery. The + # transaction rolls back, the durable outbox row stays + # available, and an explicit HTTP start retries the run + # once the operator resolves the named next action. + try: + async with pool.acquire() as conn: + async with conn.transaction(): + owner = await conn.fetchrow( + """ + select requested_by_account_id + from analysis_run + where analysis_run_id = $1::uuid + """, + analysis_run_id, ) + if owner is not None: + await deliver_queued_analysis_run( + conn, + analysis_run_id=analysis_run_id, + account_id=str(owner["requested_by_account_id"]), + affiliated_entity_ids=[], + tepp_client=tepp_client, + adjudication_client=adjudication_client, + valkey_stream_entry_id=str(entry_id), + ) + except AnalysisRunCreateError as exc: + logger.warning( + "analysis-run %s delivery refused (%s): %s", + analysis_run_id, + exc.status_code, + exc.detail, + ) last_id = str(entry_id) return last_id diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index ca01dac01..90a993ac3 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -22,9 +22,15 @@ from lineageweave.lineage_persistence import lineage_edge_specs from lineageweave.models import Edge, Record -# ADR 0145 rejected the unanchored estimator. A future accepted ADR must add -# its independently validated method code here before persisted weights can run. -_SUPPORTED_ANCHOR_METHOD_CODES: frozenset[str] = frozenset() +# Accepted ADR 0200 (points 2-3) authorizes exactly one anchor method: +# expected-information estimates honestly labeled as validated by the +# channels' internal response structure only, pending the TEPP +# criterion-validity gate. When that gate exists, a set that fails it is +# retired and this stays the only place an anchor method is ever added +# -- ADR-first, per ADR 0145's original condition. +_SUPPORTED_ANCHOR_METHOD_CODES: frozenset[str] = frozenset( + {"unanchored_internal_structure"} +) def _occurred_at(value: datetime) -> datetime: @@ -186,8 +192,33 @@ async def load_estimated_channel_weights( return persisted +class ChannelWeightsNotEstimated(RuntimeError): + """No activated estimated weight set exists for the active channels. + + Product reconstruction treats fusion weights as measurement output + only (ADR 0200 point 1): estimated by fast-mlsirm, provenance-gated, + never hand-picked constants. No hand-picked default exists anywhere + -- the library demo estimates its weights from its declared design, + and unit tests inject synthetic weights explicitly. + """ + + def __init__(self, active_channels: set[str]) -> None: + super().__init__( + "no activated fast-mlsirm channel weight estimate exists for " + f"active channels {sorted(active_channels)}; run " + "scripts/estimate_channel_weights.py first -- product " + "reconstruction never falls back to hand-picked weights" + ) + self.active_channels = active_channels + + async def rebuild_lineage(conn: asyncpg.Connection) -> list[Edge]: - """Reconstruct lineage for every ``source_post`` and persist the edges.""" + """Reconstruct lineage for every ``source_post`` and persist the edges. + + Raises :class:`ChannelWeightsNotEstimated` when no activated + estimate matches this path's active channels -- run + ``scripts/estimate_channel_weights.py`` first (ADR 0200 point 1). + """ rows = await conn.fetch( "select post_id, post_title, voc_type_code, created_at, corporate_entity_id, " "process_unit_id, thread_group_key, secondary_grouping_key " @@ -196,9 +227,10 @@ async def rebuild_lineage(conn: asyncpg.Connection) -> list[Edge]: # No adjudication client is wired on this path, so the active channel # set is the three deterministic channels (reconstruct drops llm when # unavailable rather than faking it). - weights = await load_estimated_channel_weights( - conn, {"temporal", "secondary_key", "text"} - ) + active_channels = {"temporal", "secondary_key", "text"} + weights = await load_estimated_channel_weights(conn, active_channels) + if weights is None: + raise ChannelWeightsNotEstimated(active_channels) edges = lineage_edge_specs(records_from_source_posts(rows), weights=weights) await persist_lineage_edges(conn, edges) return edges diff --git a/backend/app/main.py b/backend/app/main.py index 8aec031d1..4b091fe25 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -170,6 +170,7 @@ visible_team_mention_post_ids, ) from backend.app.lineage_ingestion import ( + ChannelWeightsNotEstimated, rebuild_lineage, visible_lineage_graph, ) @@ -368,12 +369,11 @@ def _post_summary_client(): def _adjudication_client(): """Live orchestrator client when configured; otherwise the unavailable null. - reconstruct.py's DEFAULT_CHANNEL_WEIGHTS gives this channel the most - weight (0.40) of the four -- it is the only one that reasons about - content instead of approximating it (ADR 0064) -- but nothing ever - passed a real client through lineage_edge_specs() to reconstruct(), - so every lineage reconstruction had silently run on the weaker - 3-channel fallback since the feature was built. + The llm channel is the only one that reasons about content instead + of approximating it (ADR 0064). Its fusion weight -- like every + channel's -- is a fast-mlsirm estimate (ADR 0200): with this client + available, a four-channel run uses the persisted + `channel_set_with_llm` estimate and fails closed until one exists. """ settings = load_settings() if not (settings.orchestrator_base_url and settings.orchestrator_api_key): @@ -1169,7 +1169,14 @@ async def rebuild_lineage_graph( _require_post_admin(account) async with pool.acquire() as conn: async with conn.transaction(): - edges = await rebuild_lineage(conn) + try: + edges = await rebuild_lineage(conn) + except ChannelWeightsNotEstimated: + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "Channel weights are not estimated yet. Run " + "scripts/estimate_channel_weights.py, then rebuild again.", + ) return {"edge_count": len(edges)} diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index 4cdf0729b..8a69232f7 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -122,6 +122,16 @@ / "migrations" / "0135_lineage_channel_weight.sql" ) +_CHANNEL_WEIGHT_UNION_MIGRATION = ( + Path(__file__).resolve().parents[2] + / "migrations" + / "0200_channel_weight_schema_union.sql" +) +_PAIR_JUDGMENT_MIGRATION = ( + Path(__file__).resolve().parents[2] + / "migrations" + / "0201_lineage_pair_judgment.sql" +) _LEFTOVER_OBSERVED_EXPECTED_MIGRATION = ( Path(__file__).resolve().parents[2] / "migrations" @@ -267,6 +277,28 @@ def seeded_db(demo_analyst_token): cur.execute(_PROJECT_BOUND_EVENT_MIGRATION.read_text()) cur.execute(_TENANT_SETTINGS_MIGRATION.read_text()) cur.execute(_CHANNEL_WEIGHT_MIGRATION.read_text()) + cur.execute(_CHANNEL_WEIGHT_UNION_MIGRATION.read_text()) + cur.execute(_PAIR_JUDGMENT_MIGRATION.read_text()) + # Product reconstruction fails closed without an ACTIVATED + # estimate (ADR 0200 points 1+3); this synthetic fixture set + # under the authorized anchor stands in for a fast-mlsirm + # estimate in unit tests. + cur.execute( + "insert into lineage_channel_weight " + "(channel_set_code, channel_code, weight_value, " + " estimation_run_id, estimation_method_code, estimator_version, " + " anchor_method_code, source_snapshot_sha256, sample_pair_count, " + " knowledge_cutoff) values " + "('channel_set_deterministic', 'temporal', 0.5, " + " '00000000-0000-0000-0000-000000000001', 'test_fixture', 'test', " + " 'unanchored_internal_structure', repeat('a', 64), 600, now()), " + "('channel_set_deterministic', 'secondary_key', 0.34, " + " '00000000-0000-0000-0000-000000000001', 'test_fixture', 'test', " + " 'unanchored_internal_structure', repeat('a', 64), 600, now()), " + "('channel_set_deterministic', 'text', 0.16, " + " '00000000-0000-0000-0000-000000000001', 'test_fixture', 'test', " + " 'unanchored_internal_structure', repeat('a', 64), 600, now())" + ) cur.execute(_LEFTOVER_OBSERVED_EXPECTED_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_RANK_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_COVERAGE_MIGRATION.read_text()) diff --git a/lineageweave/channel_weight_estimation.py b/lineageweave/channel_weight_estimation.py index 6c9b965ae..c46b6ce5e 100644 --- a/lineageweave/channel_weight_estimation.py +++ b/lineageweave/channel_weight_estimation.py @@ -57,6 +57,29 @@ _FIXTURE_SIMULATION_SEED = 20260824 +def fixture_design_digest() -> str: + """Reproducible SHA-256 of the demo's declared generative design. + + Plays the role the corpus snapshot digest plays for production + estimates: the provenance row names exactly which design supported + the demo estimate. Deterministic by construction. + """ + import hashlib + + material = "\n".join( + [ + *( + f"{channel}\t{probability}" + for channel, probability in sorted(_FIXTURE_FOLLOW_PROBABILITY.items()) + ), + f"groups\t{_FIXTURE_GROUP_COUNT}", + f"pairs\t{_FIXTURE_PAIR_COUNT}", + f"seed\t{_FIXTURE_SIMULATION_SEED}", + ] + ) + return hashlib.sha256(material.encode("utf-8")).hexdigest() + + def simulate_fixture_pair_scores() -> tuple[list[dict[str, float]], list[int]]: """Simulate the demo design's channel responses, deterministically. diff --git a/lineageweave/lineage_persistence.py b/lineageweave/lineage_persistence.py index d103eab47..4d5ef0dd2 100644 --- a/lineageweave/lineage_persistence.py +++ b/lineageweave/lineage_persistence.py @@ -20,7 +20,7 @@ def lineage_edge_specs( records: Sequence[Record], *, llm: AdjudicationClient | None = None, - weights: dict[str, float] | None = None, + weights: dict[str, float], ) -> list[Edge]: """Run reconstruct and return every resulting parent→child edge. @@ -35,12 +35,11 @@ def lineage_edge_specs( faked) -- callers that want the highest-weighted reasoning channel actually contributing to real reconstructions must pass a real one. - ``weights`` defaults to ``None``, keeping ``reconstruct()``'s - documented fallback constants; callers with a persisted - psychometric estimate (ADR 0145) pass it here. + ``weights`` is required and always a psychometric estimate (ADR + 0145, second amendment): the persisted fast-mlsirm corpus estimate + on product paths, or the demo-design estimate from + :func:`~lineageweave.channel_weight_estimation.estimate_fixture_channel_weights`. + No hand-picked default exists anywhere. """ - if weights is None: - trees = reconstruct(list(records), llm=llm) - else: - trees = reconstruct(list(records), llm=llm, weights=weights) + trees = reconstruct(list(records), llm=llm, weights=weights) return [edge for tree in trees for edge in tree.edges] diff --git a/lineageweave/rankweave_client.py b/lineageweave/rankweave_client.py index 76fb59cc0..b7e01c167 100644 --- a/lineageweave/rankweave_client.py +++ b/lineageweave/rankweave_client.py @@ -27,7 +27,6 @@ # 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" @@ -247,7 +246,9 @@ def project_ranking_list( items: list[RankedPost] = [] seen: set[str] = set() owned_channels = channels or {} - owned_weights = weights or DEFAULT_CHANNEL_WEIGHTS + # Parameter-free classic RRF default (ADR 0200 point 1): every + # channel weighs 1.0 unless the caller passes an estimated set. + owned_weights = weights or {name: 1.0 for name in owned_channels} for hit in raw: post_id = _item_id_from_hit(hit) title = str(titles_by_id.get(post_id) or "").strip() @@ -350,8 +351,19 @@ def fuse_rankings( titles_by_id: Mapping[str, str], weights: dict[str, float] | None = None, ) -> RankingList: - """Fuse the given per-channel id lists into one weighted RankingList.""" - active_weights = weights or DEFAULT_CHANNEL_WEIGHTS + """Fuse the channels; parameter-free classic RRF by default. + + No hand-picked weight exists (ADR 0200 point 1): without an + explicit ``weights`` argument every channel gets weight 1.0, + which reduces weighted RRF to Cormack et al.'s (2009) + parameter-free reciprocal rank fusion -- the paper's own + finding is that the unweighted form outperforms trained + alternatives, so there is no arbitrary number to justify. + Callers holding a psychometrically estimated set may still pass + it explicitly; the disclosed per-channel evidence carries + whichever weights actually fused. + """ + active_weights = weights or {name: 1.0 for name in channels} try: raw = self._transport(channels, active_weights) except RankWeaveNotAvailable: diff --git a/lineageweave/reconstruct.py b/lineageweave/reconstruct.py index 031300424..a52081369 100644 --- a/lineageweave/reconstruct.py +++ b/lineageweave/reconstruct.py @@ -20,10 +20,13 @@ from .channels import secondary_key_match_score, temporal_score, text_similarity_score from .models import Edge, Record, Tree -# Legacy compatibility weights when every channel is available. These constants -# are not calibrated measurement evidence (ADR 0145); unavailable channels are -# dropped and the remainder renormalized by active_weights(). -DEFAULT_CHANNEL_WEIGHTS = {"temporal": 0.15, "secondary_key": 0.15, "text": 0.30, "llm": 0.40} +# No default channel weights exist (ADR 0145, second amendment): fusion +# weights are always estimated by a psychometric model (fast-mlsirm +# today, TEPP when integrated) -- product paths load the persisted +# corpus estimate, and the library demo estimates from its declared +# generative design (channel_weight_estimation.estimate_fixture_channel_weights). +# Every reconstruct() caller passes weights explicitly; the llm entry +# renormalizes away when no client is configured (see active_weights()). # ponytail: only the most recent WINDOW prior records in a group are # considered as candidate parents, bounding per-group cost to O(n*window) @@ -42,7 +45,7 @@ def active_weights( - llm: AdjudicationClient, weights: dict[str, float] = DEFAULT_CHANNEL_WEIGHTS + llm: AdjudicationClient, weights: dict[str, float] ) -> dict[str, float]: """Drop and renormalize the llm channel's weight when no client is configured.""" active = dict(weights) @@ -146,7 +149,7 @@ def reconstruct( records: list[Record], *, llm: AdjudicationClient | None = None, - weights: dict[str, float] = DEFAULT_CHANNEL_WEIGHTS, + weights: dict[str, float], candidate_window: int = DEFAULT_CANDIDATE_WINDOW, min_fused_score: float = DEFAULT_MIN_FUSED_SCORE, ) -> list[Tree]: @@ -158,7 +161,11 @@ def reconstruct( :class:`~lineageweave.adjudication_client.NullAdjudicationClient` (the llm channel is then dropped, not faked). weights: per-channel fusion weights before llm-availability - renormalization; see :data:`DEFAULT_CHANNEL_WEIGHTS`. + renormalization. Required, and always a psychometric + estimate (ADR 0145, second amendment): the persisted + fast-mlsirm corpus estimate on product paths, or + :func:`~lineageweave.channel_weight_estimation.estimate_fixture_channel_weights` + for the library demo. No hand-picked default exists. candidate_window: how many recent prior records to consider as candidate parents per record; see :data:`DEFAULT_CANDIDATE_WINDOW`. min_fused_score: candidates scoring below this become a new root diff --git a/lineageweave/server.py b/lineageweave/server.py index 027e9f0ad..cb52c0ec3 100644 --- a/lineageweave/server.py +++ b/lineageweave/server.py @@ -10,6 +10,7 @@ import os from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from .channel_weight_estimation import estimate_fixture_channel_weights from .fixtures import sample_records from .models import Tree from .reconstruct import reconstruct @@ -45,16 +46,38 @@ def trees_to_graph(trees: list[Tree]) -> dict: return {"nodes": nodes, "edges": edges} -def build_server(host: str = "127.0.0.1", port: int = 8420) -> ThreadingHTTPServer: - """Build, but do not start, the demo server.""" +def build_server( + host: str = "127.0.0.1", + port: int = 8420, + *, + weights: dict[str, float] | None = None, +) -> ThreadingHTTPServer: + """Build, but do not start, the demo server. + + ``weights`` defaults to the demo-design estimate (ADR 0145, second + amendment: no hand-picked fusion weight exists anywhere). When no + estimate can be produced -- fast-mlsirm not importable -- the + server refuses to build and names the next action, rather than + fusing on invented weights. Tests inject synthetic weights here. + """ + if weights is None: + estimate = estimate_fixture_channel_weights() + if estimate is None: + raise RuntimeError( + "the demo estimates its fusion weights with fast-mlsirm and " + "none could be produced; install fast-mlsirm from the " + "organization repository, then start the demo server again" + ) + weights = estimate.weights + demo_weights = weights class Handler(BaseHTTPRequestHandler): - """Serves the demo lineage-graph JSON endpoint and the static DAG viewer.""" + """Serve the demo lineage graph JSON and the static DAG viewer.""" def do_GET(self) -> None: # noqa: N802 """Handle a GET request using the server's health endpoint contract.""" if self.path == "/api/lineage": - trees = reconstruct(sample_records()) + trees = reconstruct(sample_records(), weights=demo_weights) self._send_json(trees_to_graph(trees)) return self._send_static() diff --git a/scripts/import_postgresql_posts.py b/scripts/import_postgresql_posts.py index e1e14ed86..69f540ae1 100644 --- a/scripts/import_postgresql_posts.py +++ b/scripts/import_postgresql_posts.py @@ -26,7 +26,10 @@ if str(REPOSITORY_ROOT) not in sys.path: sys.path.insert(0, str(REPOSITORY_ROOT)) -from backend.app.lineage_ingestion import rebuild_lineage +from backend.app.lineage_ingestion import ( + ChannelWeightsNotEstimated, + rebuild_lineage, +) from lineageweave.synthetic_seed_cleanup import cleanup_synthetic_seed from lineageweave.embedding_client import orchestrator_embedding_client from lineageweave.image_content import orchestrator_vision_client @@ -579,12 +582,28 @@ async def import_rows(args: argparse.Namespace) -> dict[str, int]: ) imported += 1 cleanup = await cleanup_synthetic_seed(target, apply=True) - edges = await rebuild_lineage(target) + # A fresh corpus has no activated estimate yet (chicken-and-egg: + # estimation samples the imported corpus), and product + # reconstruction never falls back to hand-picked constants + # (ADR 0200 point 1). Skip the rebuild with a next-action note + # instead of failing the whole import. + try: + edges = await rebuild_lineage(target) + lineage_summary: dict[str, object] = {"lineage_edges": len(edges)} + except ChannelWeightsNotEstimated as exc: + lineage_summary = { + "lineage_edges": None, + "lineage_rebuild_skipped": ( + f"{exc} -- after this import, run " + "scripts/estimate_channel_weights.py and then " + "POST /api/lineage/rebuild" + ), + } return { "source_rows": len(rows), "imported_rows": imported, "skipped_rows": skipped, - "lineage_edges": len(edges), + **lineage_summary, **cleanup, } finally: diff --git a/scripts/seed_demo_data.py b/scripts/seed_demo_data.py index 0034b35a6..1418d9af0 100644 --- a/scripts/seed_demo_data.py +++ b/scripts/seed_demo_data.py @@ -501,6 +501,84 @@ def insert_fixture_source_posts(cur, author_account_id, corporate_entity_id, pro return persisted +_DEMO_ESTIMATE_CACHE: list = [] + + +def demo_channel_weight_estimate(): + """The demo's fast-mlsirm-estimated fusion weights (ADR 0200 point 1). + + No hand-picked fusion weight exists anywhere, the demo included: the + seed fits fast-mlsirm's multilevel 2PL over the demo scenario's + declared generative design and fuses with those estimates (fitted + once per process; the design is seeded, so the estimate is + deterministic). When no estimate can be produced the seed stops and + names the next action instead of inventing weights. + """ + from lineageweave.channel_weight_estimation import estimate_fixture_channel_weights + + if not _DEMO_ESTIMATE_CACHE: + _DEMO_ESTIMATE_CACHE.append(estimate_fixture_channel_weights()) + estimate = _DEMO_ESTIMATE_CACHE[0] + if estimate is None: + raise SystemExit( + "make seed estimates its fusion weights with fast-mlsirm and none " + "could be produced; install fast-mlsirm from the organization " + "repository, then run make seed again" + ) + return estimate + + +def _persist_demo_channel_weights(cur, estimate) -> None: + """Persist the demo estimate with full provenance (migration 0200). + + Product reconstruction fails closed without an activated estimate; + seeding the demo estimate keeps POST /api/lineage/rebuild and + analysis-run start working on a freshly seeded environment. The + provenance snapshot digest names the demo's declared generative + design, the honest anchor label applies, and the estimator version + is the installed fast-mlsirm. + """ + import uuid as uuid_module + from datetime import datetime, timezone + + from lineageweave.channel_weight_estimation import fixture_design_digest + from scripts.estimate_channel_weights import ( + UNANCHORED_METHOD_CODE, + estimator_version, + ) + + estimation_run_id = str(uuid_module.uuid4()) + version = estimator_version() + design_digest = fixture_design_digest() + knowledge_cutoff = datetime.now(timezone.utc) + cur.execute( + "delete from lineage_channel_weight " + "where channel_set_code = 'channel_set_deterministic'" + ) + for channel, weight in estimate.weights.items(): + cur.execute( + """ + insert into lineage_channel_weight + (channel_set_code, channel_code, weight_value, + estimation_run_id, estimation_method_code, + estimator_version, anchor_method_code, + source_snapshot_sha256, sample_pair_count, knowledge_cutoff) + values ('channel_set_deterministic', %s, %s, %s, %s, %s, %s, %s, %s, %s) + """, + ( + channel, + weight, + estimation_run_id, + estimate.estimation_method_code, + version, + UNANCHORED_METHOD_CODE, + design_digest, + estimate.sample_pair_count, + knowledge_cutoff, + ), + ) + + def _seed_reconstructed_lineage(cur, author_account_id, corporate_entity_id, process_unit_id) -> None: """Persist fixtures.sample_records() as source_posts plus reconstruct edges. @@ -511,15 +589,21 @@ def _seed_reconstructed_lineage(cur, author_account_id, corporate_entity_id, pro from lineageweave.fixtures import sample_records from lineageweave.lineage_persistence import lineage_edge_specs + # Idempotency first: a re-seed of an already-seeded database (whose + # estimate was persisted on the first pass) must not abort just + # because fast-mlsirm is absent in this environment. records = sample_records() cur.execute("select 1 from source_post where post_title = %s", (records[0].label,)) if cur.fetchone() is not None: return + estimate = demo_channel_weight_estimate() + _persist_demo_channel_weights(cur, estimate) + persisted = insert_fixture_source_posts( cur, author_account_id, corporate_entity_id, process_unit_id ) - for edge in lineage_edge_specs(persisted): + for edge in lineage_edge_specs(persisted, weights=estimate.weights): cur.execute( "insert into post_lineage_edge (parent_post_id, child_post_id, fused_score) " "values (%s, %s, %s) on conflict do nothing", @@ -1551,13 +1635,18 @@ def _seed_demo_analysis_run(cur, requested_by_account_id, corporate_entity_id) - _seed_demo_run_outbox(cur, run_id) -def seed_reconstruction_edges(rows: list[dict]) -> tuple: - """ThreadWeave parent choices and digest for seed and start. Never a theta.""" +def seed_reconstruction_edges(rows: list[dict], weights: dict[str, float]) -> tuple: + """ThreadWeave parent choices and digest for seed and start. Never a theta. + + ``weights`` is required (ADR 0200 point 1): the seed passes its + fast-mlsirm demo-design estimate; unit tests inject synthetic + weights. + """ from backend.app.analysis_run_start import reconstruction_result_digest from backend.app.lineage_ingestion import records_from_source_posts from lineageweave.lineage_persistence import lineage_edge_specs - edges = lineage_edge_specs(records_from_source_posts(rows)) + edges = lineage_edge_specs(records_from_source_posts(rows), weights=weights) return edges, reconstruction_result_digest(edges) @@ -1592,7 +1681,9 @@ def _seed_demo_run_reconstruction(cur, analysis_run_id, corporate_entity_id) -> rows = [dict(zip(columns, row)) for row in cur.fetchall()] if not rows: return - edges, digest = seed_reconstruction_edges(rows) + edges, digest = seed_reconstruction_edges( + rows, demo_channel_weight_estimate().weights + ) finished = datetime(2026, 1, 12, 12, 33, tzinfo=timezone.utc) cur.execute( """ diff --git a/tests/test_analysis_run_start.py b/tests/test_analysis_run_start.py index e4753f30f..acb2b3301 100644 --- a/tests/test_analysis_run_start.py +++ b/tests/test_analysis_run_start.py @@ -21,10 +21,15 @@ from lineageweave.lineage_persistence import lineage_edge_specs from lineageweave.tepp_client import AnalysisRunRequest, TeppClient, TeppNotAvailable +# Synthetic unit-test fusion weights (org policy allows synthetic data +# in unit tests); product paths load persisted fast-mlsirm estimates +# and fail closed otherwise (ADR 0200 point 1). +_SYNTHETIC_WEIGHTS = {"temporal": 0.5, "secondary_key": 0.34, "text": 0.16} + def test_reconstruction_digest_is_stable_and_ignores_edge_order() -> None: """The same parent choices hash the same way regardless of insert order.""" - edges = lineage_edge_specs(sample_records()) + edges = lineage_edge_specs(sample_records(), weights=_SYNTHETIC_WEIGHTS) reversed_edges = list(reversed(edges)) assert reconstruction_result_digest(edges) == reconstruction_result_digest(reversed_edges) assert reconstruction_result_digest([]) == reconstruction_result_digest([]) @@ -38,7 +43,7 @@ def test_start_uses_the_same_parent_choices_as_library_reconstruct() -> None: branch point for the revised quote and the delivery question. A start that dropped an edge or invented a parent would fail this check. """ - edges = lineage_edge_specs(sample_records()) + edges = lineage_edge_specs(sample_records(), weights=_SYNTHETIC_WEIGHTS) children = {edge.child_id for edge in edges if edge.parent_id == "rec-002"} assert children >= {"rec-003", "rec-004"} assert all(0.0 <= edge.fused_score <= 1.0 for edge in edges) @@ -59,11 +64,11 @@ def test_start_wiring_recovers_a100_from_source_post_rows() -> None: } for record in sample_records() ] - edges = lineage_edge_specs(records_from_source_posts(rows)) + edges = lineage_edge_specs(records_from_source_posts(rows), weights=_SYNTHETIC_WEIGHTS) children = {edge.child_id for edge in edges if edge.parent_id == "rec-002"} assert children >= {"rec-003", "rec-004"} assert reconstruction_result_digest(edges) == reconstruction_result_digest( - lineage_edge_specs(sample_records()) + lineage_edge_specs(sample_records(), weights=_SYNTHETIC_WEIGHTS) ) diff --git a/tests/test_analysis_run_worker.py b/tests/test_analysis_run_worker.py index 2a840b081..e492b5a2e 100644 --- a/tests/test_analysis_run_worker.py +++ b/tests/test_analysis_run_worker.py @@ -5,6 +5,7 @@ import pytest from backend.app import analysis_run_worker +from backend.app.analysis_run_start import AnalysisRunStartError from lineageweave.adjudication_client import NullAdjudicationClient from lineageweave.tepp_client import TeppClient @@ -84,3 +85,48 @@ async def fake_deliver(conn, **kwargs): "valkey_stream_entry_id": "1-0", } ] + + +class _TwoRunsValkey: + async def xread(self, _streams, *, count, block): + del count, block + return [ + ( + "analysis-run-outbox", + [ + ("2-0", {"analysis_run_id": "00000000-0000-0000-0000-000000000001"}), + ("2-1", {"analysis_run_id": "00000000-0000-0000-0000-000000000002"}), + ], + ) + ] + + +@pytest.mark.anyio +async def test_one_refused_delivery_does_not_end_the_worker(monkeypatch): + """A fail-closed refusal (e.g. channel weights not estimated, ADR 0145) + on one run must advance the cursor and still deliver the next run -- + a raised refusal would otherwise end the worker task and halt every + later run's delivery. + """ + delivered = [] + + async def fake_deliver(conn, **kwargs): + del conn + if kwargs["analysis_run_id"].endswith("1"): + raise AnalysisRunStartError( + 503, "Channel weights are not estimated yet." + ) + delivered.append(kwargs["analysis_run_id"]) + + monkeypatch.setattr(analysis_run_worker, "deliver_queued_analysis_run", fake_deliver) + + last_id = await analysis_run_worker.consume_analysis_run_stream_once( + _TwoRunsValkey(), + _Pool(), + last_id="0-0", + tepp_client=TeppClient(), + adjudication_client=NullAdjudicationClient(), + ) + + assert last_id == "2-1" + assert delivered == ["00000000-0000-0000-0000-000000000002"] diff --git a/tests/test_indirect_lineage_linking.py b/tests/test_indirect_lineage_linking.py index b20e990b6..9079d3221 100644 --- a/tests/test_indirect_lineage_linking.py +++ b/tests/test_indirect_lineage_linking.py @@ -26,6 +26,11 @@ from lineageweave.models import Record from lineageweave.reconstruct import reconstruct +# Synthetic unit-test fusion weights (org policy allows synthetic data +# in unit tests); product paths load persisted fast-mlsirm estimates +# and fail closed otherwise (ADR 0145, second amendment). +_SYNTHETIC_WEIGHTS = {"temporal": 0.5, "secondary_key": 0.34, "text": 0.16} + _SHARED_PERSON_ID = "person-shared-keyman" @@ -59,7 +64,7 @@ def test_reconstruct_never_links_posts_in_different_groups() -> None: compared, so there is categorically no edge between them, by design. """ records = _unrelated_posts() - trees = reconstruct(records) + trees = reconstruct(records, weights=_SYNTHETIC_WEIGHTS) tree_by_group = {tree.group_key: tree for tree in trees} assert set(tree_by_group) == {"group-transformers", "group-switchgear"} diff --git a/tests/test_lineage_ingestion.py b/tests/test_lineage_ingestion.py index eb3217580..dc9bbd142 100644 --- a/tests/test_lineage_ingestion.py +++ b/tests/test_lineage_ingestion.py @@ -6,6 +6,8 @@ import math from datetime import UTC, datetime, timezone +import pytest + import backend.app.lineage_ingestion as ingestion from backend.app.lineage_ingestion import lineage_graphs_for_posts from lineageweave.fixtures import sample_records @@ -59,6 +61,42 @@ async def fetch(self, _query: str): ) is None +def test_adr_0200_authorized_anchor_activates_a_complete_vector() -> None: + """Accepted ADR 0200: 'unanchored_internal_structure' is the one + authorized anchor method -- a complete, single-run, + integrity-passing vector under it activates, with no monkeypatching + of the authorized set. The rejected estimator's code + ('unanchored_channel_covariance', previous test) stays refused. + """ + + class StoredWeightConnection: + async def fetchval(self, _query: str): + return True + + async def fetch(self, _query: str): + provenance = { + "channel_set_code": "channel_set_deterministic", + "estimation_run_id": "00000000-0000-0000-0000-000000000001", + "estimation_method_code": "mls2plm_expected_information", + "estimator_version": "1.0.0", + "anchor_method_code": "unanchored_internal_structure", + "source_snapshot_sha256": "a" * 64, + "sample_pair_count": 600, + "knowledge_cutoff": datetime(2026, 1, 1, tzinfo=UTC), + } + return [ + {**provenance, "channel_code": "temporal", "weight_value": 0.5}, + {**provenance, "channel_code": "secondary_key", "weight_value": 0.3}, + {**provenance, "channel_code": "text", "weight_value": 0.2}, + ] + + assert asyncio.run( + ingestion.load_estimated_channel_weights( + StoredWeightConnection(), {"temporal", "secondary_key", "text"} + ) + ) == {"temporal": 0.5, "secondary_key": 0.3, "text": 0.2} + + def test_incomplete_persisted_weight_vector_is_unavailable() -> None: """A partial vector must not silently reweight only some channels.""" @@ -360,15 +398,23 @@ def test_seed_shaped_rows_rebuild_to_the_designed_a100_fork() -> None: "created_at": rec.occurred_at, } ) - edges = lineage_edge_specs(ingestion.records_from_source_posts(rows)) + edges = lineage_edge_specs( + ingestion.records_from_source_posts(rows), + # Synthetic unit-test weights (ADR 0200 point 1: no library default). + weights={"temporal": 0.5, "secondary_key": 0.34, "text": 0.16}, + ) pairs = {(edge.parent_id, edge.child_id) for edge in edges} assert ("rec-002", "rec-003") in pairs assert ("rec-002", "rec-004") in pairs assert "rec-006" not in {edge.child_id for edge in edges} -def test_rebuild_persists_the_synthetic_fork_without_estimated_weights() -> None: - """A missing weight table keeps deterministic reconstruction operational.""" +def test_rebuild_fails_closed_without_an_activated_weight_estimate() -> None: + """ADR 0200 point 1: no activated estimate -> no reconstruction on + constants. The raised message names the next action (run the + estimation script) so the operator is never left guessing, and + nothing is written. + """ rows = [ { "post_id": rec.record_id, @@ -398,6 +444,64 @@ async def fetchval(self, query: str): async def execute(self, query: str, *args: object) -> None: self.executions.append((query, args)) + connection = FakeConnection() + with pytest.raises(ingestion.ChannelWeightsNotEstimated) as raised: + asyncio.run(ingestion.rebuild_lineage(connection)) + assert "estimate_channel_weights" in str(raised.value) + assert connection.executions == [] + + +def test_rebuild_reconstructs_with_an_activated_estimate() -> None: + """With an activated estimate the designed fork persists as before.""" + rows = [ + { + "post_id": rec.record_id, + "process_unit_id": "shared-pu", + "corporate_entity_id": "shared-corp", + "post_title": rec.label, + "voc_type_code": "voc" if rec.secondary_key else "vom", + "thread_group_key": rec.group_key, + "secondary_grouping_key": rec.secondary_key, + "created_at": rec.occurred_at, + } + for rec in sample_records() + ] + weight_rows = [ + { + "channel_set_code": "channel_set_deterministic", + "channel_code": channel, + "weight_value": weight, + "estimation_run_id": "00000000-0000-0000-0000-000000000001", + "estimation_method_code": "mls2plm_expected_information", + "estimator_version": "1.0.0", + "anchor_method_code": "unanchored_internal_structure", + "source_snapshot_sha256": "a" * 64, + "sample_pair_count": 600, + "knowledge_cutoff": datetime(2026, 1, 1, tzinfo=UTC), + } + for channel, weight in ( + ("temporal", 0.5), + ("secondary_key", 0.34), + ("text", 0.16), + ) + ] + + class FakeConnection: + def __init__(self) -> None: + self.executions: list[tuple[str, tuple[object, ...]]] = [] + + async def fetch(self, query: str): + if "lineage_channel_weight" in query: + return weight_rows + assert "from source_post" in query + return rows + + async def fetchval(self, _query: str): + return True + + async def execute(self, query: str, *args: object) -> None: + self.executions.append((query, args)) + connection = FakeConnection() edges = asyncio.run(ingestion.rebuild_lineage(connection)) diff --git a/tests/test_lineage_persistence.py b/tests/test_lineage_persistence.py index 33dab5d40..5e7e48d5f 100644 --- a/tests/test_lineage_persistence.py +++ b/tests/test_lineage_persistence.py @@ -9,9 +9,14 @@ from lineageweave.fixtures import sample_records from lineageweave.lineage_persistence import lineage_edge_specs +# Synthetic unit-test fusion weights (org policy allows synthetic data +# in unit tests); product paths load persisted fast-mlsirm estimates +# and fail closed otherwise (ADR 0145, second amendment). +_SYNTHETIC_WEIGHTS = {"temporal": 0.5, "secondary_key": 0.34, "text": 0.16} + def test_persisted_edges_include_the_designed_a100_fork() -> None: - edges = lineage_edge_specs(sample_records()) + edges = lineage_edge_specs(sample_records(), weights=_SYNTHETIC_WEIGHTS) pairs = {(edge.parent_id, edge.child_id) for edge in edges} assert ("rec-002", "rec-003") in pairs assert ("rec-002", "rec-004") in pairs @@ -19,17 +24,6 @@ def test_persisted_edges_include_the_designed_a100_fork() -> None: def test_unrelated_rec006_is_not_forced_onto_a_parent() -> None: - edges = lineage_edge_specs(sample_records()) + edges = lineage_edge_specs(sample_records(), weights=_SYNTHETIC_WEIGHTS) children = {edge.child_id for edge in edges} assert "rec-006" not in children - - -def test_persisted_weight_vector_reaches_reconstruction() -> None: - """An authorized deterministic vector is used without inventing an LLM score.""" - edges = lineage_edge_specs( - sample_records(), - weights={"temporal": 0.25, "secondary_key": 0.25, "text": 0.5}, - ) - pairs = {(edge.parent_id, edge.child_id) for edge in edges} - assert ("rec-002", "rec-003") in pairs - assert all("llm" not in edge.channel_scores for edge in edges) diff --git a/tests/test_rankweave_client.py b/tests/test_rankweave_client.py index c6f23af01..4feb2115b 100644 --- a/tests/test_rankweave_client.py +++ b/tests/test_rankweave_client.py @@ -14,7 +14,6 @@ import pytest from lineageweave.rankweave_client import ( - DEFAULT_CHANNEL_WEIGHTS, LibraryRankWeaveTransport, RankWeaveClient, RankWeaveNotAvailable, @@ -43,14 +42,16 @@ def _lexical_then_temporal(post_id: str, channel_rank: int) -> list[dict[str, object]]: - lexical = 0.75 / (60 + channel_rank) - temporal = 0.25 / (60 + channel_rank) + # Parameter-free classic RRF (Cormack et al., 2009): every channel + # weighs 1.0 -- no hand-picked weight exists (ADR 0200 point 1). + lexical = 1.0 / (60 + channel_rank) + temporal = 1.0 / (60 + channel_rank) return [ { "signal_code": "lexical", "signal_label": "Title overlap", "channel_rank": channel_rank, - "weight": 0.75, + "weight": 1.0, "contribution": lexical, "rank": 1, }, @@ -58,7 +59,7 @@ def _lexical_then_temporal(post_id: str, channel_rank: int) -> list[dict[str, ob "signal_code": "temporal", "signal_label": "Newest first", "channel_rank": channel_rank, - "weight": 0.25, + "weight": 1.0, "contribution": temporal, "rank": 2, }, @@ -198,7 +199,7 @@ def weighted_reciprocal_rank_fuse( ) assert captured["eta"] == 60 - assert captured["weights"]["lexical"] == 0.75 + assert set(captured["weights"].values()) == {1.0} assert payload["rankings"][0]["post_title"] == ( "Pricing renegotiation: revised quote sent" ) @@ -252,12 +253,12 @@ def test_ranking_channel_evidence_uses_cormack_weighted_rrf() -> None: evidence = ranking_channel_evidence( "post-1", {"temporal": ["post-1"], "lexical": ["post-1"]}, - DEFAULT_CHANNEL_WEIGHTS, + {"temporal": 1.0, "lexical": 1.0}, eta=60, ) by_code = {item.signal_code: item for item in evidence} - assert by_code["lexical"].contribution == 0.75 / 61 - assert by_code["temporal"].contribution == 0.25 / 61 + assert by_code["lexical"].contribution == 1.0 / 61 + assert by_code["temporal"].contribution == 1.0 / 61 assert by_code["lexical"].channel_rank == 1 assert by_code["temporal"].channel_rank == 1 assert by_code["lexical"].rank == 1 @@ -297,7 +298,7 @@ def test_project_ranking_list_ignores_transport_extra_fields() -> None: ], {"post-1": "Public post"}, channels={"temporal": ["post-1"], "lexical": ["post-2"]}, - weights=DEFAULT_CHANNEL_WEIGHTS, + weights={"temporal": 1.0, "lexical": 1.0}, ) payload = ranking.to_json() assert payload[0]["channel_evidence"] == [ @@ -305,8 +306,8 @@ def test_project_ranking_list_ignores_transport_extra_fields() -> None: "signal_code": "temporal", "signal_label": "Newest first", "channel_rank": 1, - "weight": 0.25, - "contribution": 0.25 / 61, + "weight": 1.0, + "contribution": 1.0 / 61, "rank": 1, } ] diff --git a/tests/test_reconstruct.py b/tests/test_reconstruct.py index 6ae195941..1bb8647a9 100644 --- a/tests/test_reconstruct.py +++ b/tests/test_reconstruct.py @@ -5,13 +5,26 @@ from lineageweave import Record, reconstruct from lineageweave.fixtures import sample_records +# Synthetic unit-test fusion weights (org policy allows synthetic data +# in unit tests): shaped like the demo design's discrimination ordering +# so the designed fixture tree keeps the shape under test. Product +# paths never see these -- they load persisted fast-mlsirm estimates +# and fail closed otherwise (ADR 0145, second amendment). +_SYNTHETIC_WEIGHTS = {"temporal": 0.5, "secondary_key": 0.34, "text": 0.16} +_SYNTHETIC_WEIGHTS_WITH_LLM = { + "temporal": 0.3, + "secondary_key": 0.2, + "text": 0.1, + "llm": 0.4, +} + def test_reconstruct_finds_the_designed_branch_point() -> None: """fixtures.sample_records() deliberately makes rec-002 fork into two threads (the revised quote, and the delivery question) -- the reconstruction must recover that shape without being told about it. """ - trees = reconstruct(sample_records()) + trees = reconstruct(sample_records(), weights=_SYNTHETIC_WEIGHTS) tree_a = next(t for t in trees if t.group_key == "A-100") assert "rec-001" in tree_a.roots @@ -23,14 +36,14 @@ def test_reconstruct_leaves_unrelated_records_as_their_own_root() -> None: """rec-006 shares no topic or project code with anything in A-100 and should not be force-attached to the best of a set of weak candidates. """ - trees = reconstruct(sample_records()) + trees = reconstruct(sample_records(), weights=_SYNTHETIC_WEIGHTS) tree_a = next(t for t in trees if t.group_key == "A-100") assert "rec-006" in tree_a.roots def test_reconstruct_groups_are_independent() -> None: - trees = reconstruct(sample_records()) + trees = reconstruct(sample_records(), weights=_SYNTHETIC_WEIGHTS) tree_b = next(t for t in trees if t.group_key == "B-200") assert set(tree_b.records.keys()) == {"rec-101", "rec-102", "rec-103"} @@ -38,7 +51,7 @@ def test_reconstruct_groups_are_independent() -> None: def test_llm_channel_is_dropped_not_faked_when_unavailable() -> None: - trees = reconstruct(sample_records()) + trees = reconstruct(sample_records(), weights=_SYNTHETIC_WEIGHTS) tree_a = next(t for t in trees if t.group_key == "A-100") for edge in tree_a.edges: @@ -58,7 +71,9 @@ def judge(self, candidate_label: str, record_label: str) -> float: def test_llm_channel_is_used_and_scored_when_a_client_is_supplied() -> None: stub = _StubAdjudicationClient() - trees = reconstruct(sample_records(), llm=stub) + trees = reconstruct( + sample_records(), llm=stub, weights=_SYNTHETIC_WEIGHTS_WITH_LLM + ) tree_a = next(t for t in trees if t.group_key == "A-100") assert stub.calls > 0 @@ -69,7 +84,7 @@ def test_candidate_window_bounds_which_priors_are_considered() -> None: records = [ Record(f"r{i}", "G", f"record {i}", datetime(2026, 1, 1, i), "") for i in range(5) ] - trees = reconstruct(records, candidate_window=1) + trees = reconstruct(records, weights=_SYNTHETIC_WEIGHTS, candidate_window=1) tree = trees[0] for edge in tree.edges: diff --git a/tests/test_seed_analysis_run_reconstruction.py b/tests/test_seed_analysis_run_reconstruction.py index 7f73800e7..7d670ef8f 100644 --- a/tests/test_seed_analysis_run_reconstruction.py +++ b/tests/test_seed_analysis_run_reconstruction.py @@ -5,6 +5,11 @@ from lineageweave.fixtures import sample_records from scripts.seed_demo_data import seed_reconstruction_edges +# Synthetic unit-test fusion weights (org policy allows synthetic data +# in unit tests); `make seed` itself passes its fast-mlsirm demo-design +# estimate (ADR 0145, second amendment). +_SYNTHETIC_WEIGHTS = {"temporal": 0.5, "secondary_key": 0.34, "text": 0.16} + def _rows_from_fixtures() -> list[dict]: cutoff = datetime(2026, 1, 12, 12, 0, tzinfo=timezone.utc) @@ -30,7 +35,7 @@ def _rows_from_fixtures() -> list[dict]: def test_seed_reconstruction_recovers_the_a100_fork() -> None: """Seed must persist the same parent choices start uses.""" - edges, digest = seed_reconstruction_edges(_rows_from_fixtures()) + edges, digest = seed_reconstruction_edges(_rows_from_fixtures(), _SYNTHETIC_WEIGHTS) children = {edge.child_id for edge in edges if edge.parent_id == "rec-002"} assert children >= {"rec-003", "rec-004"} assert "rec-006" not in {edge.child_id for edge in edges} diff --git a/tests/test_server.py b/tests/test_server.py index 957e7f02e..a0836e221 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -9,9 +9,15 @@ from lineageweave.server import build_server +# Synthetic unit-test fusion weights injected so this smoke test runs +# without fast-mlsirm (org policy allows synthetic data in unit tests); +# the demo binary itself estimates its weights and refuses to start +# otherwise (ADR 0145, second amendment). +_SYNTHETIC_WEIGHTS = {"temporal": 0.5, "secondary_key": 0.34, "text": 0.16} + def test_lineage_endpoint_serves_the_reconstructed_graph_with_a_branch_point() -> None: - server = build_server(port=0) + server = build_server(port=0, weights=_SYNTHETIC_WEIGHTS) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() port = server.server_address[1] @@ -31,7 +37,7 @@ def test_lineage_endpoint_serves_the_reconstructed_graph_with_a_branch_point() - def test_root_serves_the_static_viewer() -> None: - server = build_server(port=0) + server = build_server(port=0, weights=_SYNTHETIC_WEIGHTS) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() port = server.server_address[1] @@ -49,7 +55,7 @@ def test_root_serves_the_static_viewer() -> None: def test_path_traversal_is_rejected() -> None: - server = build_server(port=0) + server = build_server(port=0, weights=_SYNTHETIC_WEIGHTS) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() port = server.server_address[1]