diff --git a/CHANGELOG.d/external-lineage-contract.md b/CHANGELOG.d/external-lineage-contract.md new file mode 100644 index 000000000..835bacaf6 --- /dev/null +++ b/CHANGELOG.d/external-lineage-contract.md @@ -0,0 +1,12 @@ +# External email/project lineage contract + +- Add a strict, versioned external analysis contract for future Naruon and separately governed consumer use. +- Export immutable request/result types, strict parsing, canonical serialization, deterministic digests, stable errors, and the store-agnostic `analyze_external_lineage` package entry point. +- Accept only bounded caller-authorized opaque evidence references; no provider credentials, mailbox access, persistence, provider mutation, or direct application-database integration is introduced. +- Preserve caller-observed RFC/provider/manual parent relations separately from inferred reconstructed continuation. +- Exclude caller-observed children from alternative inferred-parent scoring, optional model disclosure, and inferred-pair budget while retaining them as candidate history for later records. +- Enforce available-time knowledge cutoffs and disclose excluded evidence without substituting later facts. +- Reject explicit-parent cycles and candidate-pair work above the caller-approved limit before optional LLM/provider activity. +- Expose exact active channel scores, weights, contributions, LLM availability state, proposed project groupings, and deterministic result digests. +- Require a provenance-bearing fast-mlsirm channel-weight estimate for inferred edges; without it, return observed edges plus an explicit unavailable limitation. +- Add JSON Schema Draft 2020-12, union-free ADR 0239, APA 7th doctoring, and focused TDD coverage. diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index c08810078..f8f348ec0 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -24,7 +24,6 @@ AnalysisRunCreateError, fetch_visible_analysis_run, ) -from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL from backend.app.analysis_run_outbox import ( latest_outbox_delivery_is_claimed, latest_outbox_delivery_is_delivered, @@ -34,7 +33,11 @@ load_estimated_channel_weights, records_from_source_posts, ) -from lineageweave.adjudication_client import AdjudicationClient +from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL +from lineageweave.adjudication_client import ( + AdjudicationClient, + AdjudicationClientError, +) from lineageweave.http_client import HttpClientError, post_json from lineageweave.lineage_persistence import lineage_edge_specs from lineageweave.models import Edge @@ -83,7 +86,13 @@ 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: + except ( + AdjudicationClientError, + HttpClientError, + OSError, + ValueError, + TypeError, + ) as exc: raise _AdjudicationProviderError(str(exc)) from exc diff --git a/backend/app/main.py b/backend/app/main.py index 6457bbde1..a407af561 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -165,6 +165,7 @@ ) from backend.app.source_post_revision import fetch_known_at_revision, parse_as_of_clock from lineageweave.adjudication_client import ( + AdjudicationClientError, ContextualOrchestratorAdjudicationClient, NullAdjudicationClient, ) @@ -1311,7 +1312,7 @@ async def rebuild_lineage_graph( "Channel weights are not estimated yet. Run " "scripts/estimate_channel_weights.py, then rebuild again.", ) from exc - except (HttpClientError, OSError) as exc: + except (AdjudicationClientError, HttpClientError, OSError) as exc: # This can issue up to MAXIMUM_LIVE_LLM_PAIR_EVALUATIONS sequential # adjudication calls across the whole corpus (lineage_ingestion.py); # a transient orchestrator hiccup on any one of them must not diff --git a/docs/adr/0239-external-email-project-lineage-contract.md b/docs/adr/0239-external-email-project-lineage-contract.md new file mode 100644 index 000000000..04d9070ff --- /dev/null +++ b/docs/adr/0239-external-email-project-lineage-contract.md @@ -0,0 +1,59 @@ +# ADR 0239: Publish a bounded external email/project lineage contract + +- Status: Accepted +- Date: 2026-08-21 + +## Context + +Naruon owns customer mail/calendar/file access, canonical message/thread identities, projects, tasks, commitments, provider credentials, authorization, and provider mutations. LineageWeave owns evidence-fused lineage reconstruction and the provenance explaining that reconstruction. Future integration must not give either product direct SQL access to the other's application database, duplicate source authority, or depend on a mutable branch/submodule. + +Email thread facts also have different truth semantics from reconstructed semantic continuation. RFC `Message-ID`, `References`, and `In-Reply-To` evidence may establish a caller-observed reply relation, while LineageWeave text/temporal/project signals produce an inferred relation. Flattening both into one unexplained score would make buyer correction and audit impossible. + +## Decision + +LineageWeave publishes contract version `1.0.0` through: + +- `lineageweave.external_lineage_contract` for strict immutable request/result shapes, canonical serialization, bounds, and deterministic digests; +- `lineageweave.external_lineage_analysis` for adapting caller-authorized evidence to the existing reconstruction kernel. + +The initial implementation is a store-agnostic Python package boundary. It performs no database, mailbox, provider, or network operation. A later service or Naruon plugin adapter must preserve the same JSON Schema and truth boundaries. + +Inferred edges additionally require a provenance-bearing +`ChannelWeightEstimate` produced by the repository's fast-mlsirm measurement +boundary. The estimate is an injected execution dependency, not caller JSON: +the external evidence contract cannot assert its own fusion weights. When no +estimate is available, the adapter still returns caller-observed edges and an +explicit `channel_weights_unavailable` limitation, but produces no inferred +edge. The LLM channel is active only when the estimate explicitly includes an +`llm` item; an available model without such measurement remains unavailable +for this run. + +The caller supplies opaque evidence references, bounded text labels, occurrence and availability clocks, an optional secondary key, an optional project reference, and an optional caller-observed parent relation. Explicit observed parent relations replace an inferred parent for the same child and must form an acyclic graph. Reconstructed continuation remains `inferred`. Project groupings remain `proposed`. + +An admitted child with an explicit observed parent is not rescored for an alternative inferred parent and consumes no optional LLM/provider call or inferred-pair budget. The record remains in temporal history and may still be an eligible candidate parent for a later record. This preserves observed authority without weakening downstream lineage reconstruction. + +The caller also supplies `maximum_pair_evaluations` in the bounded policy. The package computes the exact inferred candidate-parent pair count after knowledge-cutoff filtering, excluding children whose parent is already caller-observed, and rejects work above the declared budget before any optional LLM/provider call. Contract v1 caps the declared budget at 5,000 pairs. + +Historical requests include evidence only when: + +```text +available_at <= knowledge_cutoff +``` + +Evidence becoming available after the cutoff is excluded even when it describes an earlier occurrence. + +## Consequences + +- Naruon can eventually consume a released artifact without exposing credentials or application tables. +- RFC reply/thread evidence stays distinguishable from semantic lineage. +- Caller-observed children are never disclosed to an optional model merely to calculate an inferred edge that would be discarded. +- The optional LLM channel is explicit as `not_requested`, `unavailable`, or `completed`; missing output is never zero. +- Missing or malformed psychometric weight provenance yields no inferred edge; no default, equal, or caller-authored weight is substituted. +- Canonical serialization and SHA-256 digesting are deterministic for a given request or result. Repeatability of model-backed scores additionally requires a pinned LineageWeave release, adjudicator implementation, provider/model revision, and model-side determinism policy. +- Explicit parent cycles and analysis work above the caller-approved pair budget fail closed before inference. +- Project evidence can inform Naruon without mutating authoritative project/task/provider state. +- The single generic secondary key reflects the current core kernel. Multiple independent typed secondary-key channels remain a future contract revision rather than being silently flattened. + +## References + +See `docs/doctoring/EXTERNAL_LINEAGE_CONTRACT_REFERENCES.md`. diff --git a/docs/contracts/README.md b/docs/contracts/README.md new file mode 100644 index 000000000..c3e937193 --- /dev/null +++ b/docs/contracts/README.md @@ -0,0 +1,13 @@ +# Integration contracts + +LineageWeave publishes strict, versioned contracts for separately governed consumers. These contracts do not grant source access and do not replace each consumer's authorization, persistence, provider, or audit authority. + +## External lineage analysis v1 + +- JSON Schema: `external-lineage-analysis-v1.schema.json` +- Synthetic request: `external-lineage-analysis-v1.example.json` +- Python parser and immutable types: `lineageweave.external_lineage_contract` +- Store-agnostic execution adapter: `lineageweave.external_lineage_analysis` +- Decision record: `docs/adr/0239-external-email-project-lineage-contract.md` + +A consumer must submit only bounded evidence it is already authorized to disclose. Outputs retain opaque caller references and explicit `observed`, `inferred`, or `proposed` truth boundaries. The contract performs no source-system access or provider mutation. diff --git a/docs/contracts/external-lineage-analysis-v1.authorization.md b/docs/contracts/external-lineage-analysis-v1.authorization.md new file mode 100644 index 000000000..44216d3d2 --- /dev/null +++ b/docs/contracts/external-lineage-analysis-v1.authorization.md @@ -0,0 +1,5 @@ +# External lineage analysis v1 authorization contract + +LineageWeave does not infer authorization from an opaque reference, source kind, group, project, or caller identity. The caller must authorize evidence before projection and must reauthorize any source drill-through after receiving a result. + +The package does not accept provider bearer tokens, browser cookies, mailbox credentials, database DSNs, or caller SQL. A future remote service must use its own audience-scoped service credential and may not forward an end-user token to model providers. diff --git a/docs/contracts/external-lineage-analysis-v1.consumer-checklist.md b/docs/contracts/external-lineage-analysis-v1.consumer-checklist.md new file mode 100644 index 000000000..9c89e17cb --- /dev/null +++ b/docs/contracts/external-lineage-analysis-v1.consumer-checklist.md @@ -0,0 +1,11 @@ +# External lineage analysis v1 consumer checklist + +- Validate the published JSON Schema before sending or accepting payloads. +- Submit only evidence the calling principal is authorized to disclose for the declared purpose. +- Use opaque caller-owned references; never send provider credentials or database locators. +- Bind historical work to a knowledge cutoff and preserve each record's availability time. +- Keep RFC/provider thread observations separate from inferred semantic/project lineage. +- Treat project projections as proposals until the caller's own policy or reviewer accepts them. +- Preserve the returned artifact digest, LineageWeave version, limitations, and channel evidence. +- Fail closed on incompatible contract versions. +- Keep normal caller operation available when LineageWeave is unavailable. diff --git a/docs/contracts/external-lineage-analysis-v1.data-minimization.md b/docs/contracts/external-lineage-analysis-v1.data-minimization.md new file mode 100644 index 000000000..522c55e2a --- /dev/null +++ b/docs/contracts/external-lineage-analysis-v1.data-minimization.md @@ -0,0 +1,12 @@ +# External lineage analysis v1 data minimization + +Consumers should prefer the minimum evidence needed for a declared analysis scope: + +- opaque evidence and grouping references; +- offset-aware occurrence and availability times; +- RFC/provider relation evidence when present; +- bounded subject/title labels or caller-computed text features; +- optional project or secondary-key references; +- optional participant, body, or attachment evidence only when the caller's purpose and policy explicitly permit it. + +The contract does not require a mailbox dump, full thread body, recipient list, provider URL, or attachment bytes. Omitted evidence is unavailable and cannot appear in output. diff --git a/docs/contracts/external-lineage-analysis-v1.example.json b/docs/contracts/external-lineage-analysis-v1.example.json new file mode 100644 index 000000000..b9b90bdce --- /dev/null +++ b/docs/contracts/external-lineage-analysis-v1.example.json @@ -0,0 +1,41 @@ +{ + "contract_version": "1.0.0", + "analysis_id": "analysis:synthetic-email-lineage-001", + "analysis_scope_code": "email_lineage", + "knowledge_cutoff": "2026-08-20T09:30:00Z", + "policy": { + "candidate_window": 50, + "maximum_pair_evaluations": 1000, + "minimum_fused_score": 0.3, + "allow_llm": false + }, + "records": [ + { + "evidence_ref": "email:synthetic-001", + "group_ref": "workspace:synthetic", + "source_kind_code": "email", + "truth_status_code": "observed", + "label": "Synthetic proposal review", + "occurred_at": "2026-08-20T09:00:00Z", + "available_at": "2026-08-20T09:01:00Z", + "secondary_key": "provider-thread:synthetic", + "project_ref": "project:synthetic", + "explicit_parent": null + }, + { + "evidence_ref": "email:synthetic-002", + "group_ref": "workspace:synthetic", + "source_kind_code": "email", + "truth_status_code": "observed", + "label": "Re: Synthetic proposal review", + "occurred_at": "2026-08-20T09:05:00Z", + "available_at": "2026-08-20T09:06:00Z", + "secondary_key": "provider-thread:synthetic", + "project_ref": "project:synthetic", + "explicit_parent": { + "evidence_ref": "email:synthetic-001", + "relation_code": "rfc_reply" + } + } + ] +} diff --git a/docs/contracts/external-lineage-analysis-v1.limitations.md b/docs/contracts/external-lineage-analysis-v1.limitations.md new file mode 100644 index 000000000..fd3864f77 --- /dev/null +++ b/docs/contracts/external-lineage-analysis-v1.limitations.md @@ -0,0 +1,11 @@ +# External lineage analysis v1 limitations + +- The contract does not read IMAP, JMAP, CalDAV, Naruon, or other provider systems. +- It does not authenticate users, authorize tenant access, persist jobs, or retry remote work. +- It does not make semantic lineage equivalent to RFC reply/thread identity. +- It does not turn project groupings, responsibility context, or reconstructed edges into authoritative caller facts. +- It does not infer unavailable evidence as a zero-valued channel. +- It does not guarantee causal relations; reconstructed continuation is an evidence-weighted related-history hypothesis. +- Canonical request/result serialization and digests are deterministic, but an optional remote adjudication channel is not automatically repeatable unless the consumer pins the LineageWeave artifact, adjudicator, provider/model revision, and determinism policy. +- Contract v1 does not carry a remote provider/model receipt inside the result; production wrappers must retain that provenance alongside the result digest before model-backed integration is enabled. +- It does not replace Naruon's canonical email identity, project/task/commitment state, provider mutation, or reconciliation authority. diff --git a/docs/contracts/external-lineage-analysis-v1.operability.md b/docs/contracts/external-lineage-analysis-v1.operability.md new file mode 100644 index 000000000..7e3de54f6 --- /dev/null +++ b/docs/contracts/external-lineage-analysis-v1.operability.md @@ -0,0 +1,5 @@ +# External lineage analysis v1 operability boundary + +The pure package entry point is synchronous and bounded. Remote or model-backed production use must wrap it in a separately reviewed service or plugin lifecycle with durable idempotency, cancellation, timeout, retry classification, rate limiting, resource budgets, artifact retention, OpenTelemetry signals, and user-visible degraded states. + +A consumer must not call optional model-backed pair adjudication directly on an unbounded web request path. LineageWeave #289 tracks the durable asynchronous reconstruction requirement for product persistence, and Naruon #1437 requires an equivalent consumer-side job receipt before integration is enabled. diff --git a/docs/contracts/external-lineage-analysis-v1.schema.json b/docs/contracts/external-lineage-analysis-v1.schema.json new file mode 100644 index 000000000..babf41835 --- /dev/null +++ b/docs/contracts/external-lineage-analysis-v1.schema.json @@ -0,0 +1,255 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://contextualwisdomlab.org/schemas/external-lineage-analysis-v1.schema.json", + "title": "LineageWeave External Lineage Analysis Request v1", + "description": "Bounded caller-authorized evidence for store-agnostic lineage analysis. The response shape is available as $defs.LineageAnalysisResult.", + "type": "object", + "additionalProperties": false, + "required": [ + "contract_version", + "analysis_id", + "analysis_scope_code", + "policy", + "records" + ], + "properties": { + "contract_version": {"const": "1.0.0"}, + "analysis_id": {"$ref": "#/$defs/OpaqueReference"}, + "analysis_scope_code": { + "type": "string", + "enum": ["email_lineage", "project_history", "generic_lineage"] + }, + "knowledge_cutoff": { + "anyOf": [ + {"type": "string", "format": "date-time"}, + {"type": "null"} + ] + }, + "policy": {"$ref": "#/$defs/LineageAnalysisPolicy"}, + "records": { + "type": "array", + "minItems": 1, + "maxItems": 500, + "items": {"$ref": "#/$defs/LineageEvidenceRecord"} + } + }, + "$defs": { + "OpaqueReference": { + "type": "string", + "minLength": 1, + "maxLength": 160, + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:@+\\-]*$" + }, + "NullableOpaqueReference": { + "anyOf": [ + {"$ref": "#/$defs/OpaqueReference"}, + {"type": "null"} + ] + }, + "ExplicitParent": { + "type": "object", + "additionalProperties": false, + "required": ["evidence_ref", "relation_code"], + "properties": { + "evidence_ref": {"$ref": "#/$defs/OpaqueReference"}, + "relation_code": { + "type": "string", + "enum": ["rfc_reply", "provider_reply", "manual_parent"] + } + } + }, + "LineageAnalysisPolicy": { + "type": "object", + "additionalProperties": false, + "required": [ + "candidate_window", + "maximum_pair_evaluations", + "minimum_fused_score", + "allow_llm" + ], + "properties": { + "candidate_window": { + "type": "integer", + "minimum": 1, + "maximum": 200 + }, + "maximum_pair_evaluations": { + "type": "integer", + "minimum": 1, + "maximum": 5000 + }, + "minimum_fused_score": { + "type": "number", + "minimum": 0, + "maximum": 1 + }, + "allow_llm": {"type": "boolean"} + } + }, + "LineageEvidenceRecord": { + "type": "object", + "additionalProperties": false, + "required": [ + "evidence_ref", + "group_ref", + "source_kind_code", + "truth_status_code", + "label", + "occurred_at", + "available_at" + ], + "properties": { + "evidence_ref": {"$ref": "#/$defs/OpaqueReference"}, + "group_ref": {"$ref": "#/$defs/OpaqueReference"}, + "source_kind_code": { + "type": "string", + "enum": ["email", "task", "commitment", "project_event", "generic"] + }, + "truth_status_code": { + "type": "string", + "enum": ["observed", "authoritative_in_caller"] + }, + "label": { + "type": "string", + "minLength": 1, + "maxLength": 2000 + }, + "occurred_at": {"type": "string", "format": "date-time"}, + "available_at": {"type": "string", "format": "date-time"}, + "secondary_key": {"$ref": "#/$defs/NullableOpaqueReference"}, + "project_ref": {"$ref": "#/$defs/NullableOpaqueReference"}, + "explicit_parent": { + "anyOf": [ + {"$ref": "#/$defs/ExplicitParent"}, + {"type": "null"} + ] + } + } + }, + "ChannelEvidence": { + "type": "object", + "additionalProperties": false, + "required": ["channel_code", "score", "weight", "contribution"], + "properties": { + "channel_code": {"type": "string", "minLength": 1, "maxLength": 64}, + "score": {"type": "number", "minimum": 0, "maximum": 1}, + "weight": {"type": "number", "minimum": 0, "maximum": 1}, + "contribution": {"type": "number", "minimum": 0, "maximum": 1} + } + }, + "LineageEdgeResult": { + "type": "object", + "additionalProperties": false, + "required": [ + "parent_evidence_ref", + "child_evidence_ref", + "relation_type_code", + "truth_status_code", + "fused_score", + "channel_evidence" + ], + "properties": { + "parent_evidence_ref": {"$ref": "#/$defs/OpaqueReference"}, + "child_evidence_ref": {"$ref": "#/$defs/OpaqueReference"}, + "relation_type_code": {"type": "string", "minLength": 1, "maxLength": 64}, + "truth_status_code": { + "type": "string", + "enum": ["observed", "inferred"] + }, + "fused_score": {"type": "number", "minimum": 0, "maximum": 1}, + "channel_evidence": { + "type": "array", + "minItems": 1, + "items": {"$ref": "#/$defs/ChannelEvidence"} + } + } + }, + "ProjectProjection": { + "type": "object", + "additionalProperties": false, + "required": ["group_ref", "project_ref", "evidence_refs", "truth_status_code"], + "properties": { + "group_ref": {"$ref": "#/$defs/OpaqueReference"}, + "project_ref": {"$ref": "#/$defs/OpaqueReference"}, + "evidence_refs": { + "type": "array", + "minItems": 1, + "items": {"$ref": "#/$defs/OpaqueReference"}, + "uniqueItems": true + }, + "truth_status_code": {"const": "proposed"} + } + }, + "LineageLimitation": { + "type": "object", + "additionalProperties": false, + "required": ["limitation_code", "evidence_ref", "message"], + "properties": { + "limitation_code": {"type": "string", "minLength": 1, "maxLength": 96}, + "evidence_ref": {"$ref": "#/$defs/NullableOpaqueReference"}, + "message": {"type": "string", "minLength": 1, "maxLength": 500} + } + }, + "LineageAnalysisResult": { + "type": "object", + "additionalProperties": false, + "required": [ + "contract_version", + "analysis_id", + "analysis_scope_code", + "knowledge_cutoff", + "included_evidence_refs", + "excluded_evidence_refs", + "llm_status_code", + "edges", + "project_projections", + "limitations", + "result_digest" + ], + "properties": { + "contract_version": {"const": "1.0.0"}, + "analysis_id": {"$ref": "#/$defs/OpaqueReference"}, + "analysis_scope_code": { + "type": "string", + "enum": ["email_lineage", "project_history", "generic_lineage"] + }, + "knowledge_cutoff": { + "anyOf": [ + {"type": "string", "format": "date-time"}, + {"type": "null"} + ] + }, + "included_evidence_refs": { + "type": "array", + "items": {"$ref": "#/$defs/OpaqueReference"}, + "uniqueItems": true + }, + "excluded_evidence_refs": { + "type": "array", + "items": {"$ref": "#/$defs/OpaqueReference"}, + "uniqueItems": true + }, + "llm_status_code": { + "type": "string", + "enum": ["not_requested", "unavailable", "completed"] + }, + "edges": { + "type": "array", + "items": {"$ref": "#/$defs/LineageEdgeResult"} + }, + "project_projections": { + "type": "array", + "items": {"$ref": "#/$defs/ProjectProjection"} + }, + "limitations": { + "type": "array", + "items": {"$ref": "#/$defs/LineageLimitation"} + }, + "result_digest": { + "type": "string", + "pattern": "^sha256:[0-9a-f]{64}$" + } + } + } + } +} diff --git a/docs/contracts/external-lineage-analysis-v1.security.md b/docs/contracts/external-lineage-analysis-v1.security.md new file mode 100644 index 000000000..c87ffbab2 --- /dev/null +++ b/docs/contracts/external-lineage-analysis-v1.security.md @@ -0,0 +1,22 @@ +# External lineage analysis v1 security boundary + +The contract is an analysis interface, not an authorization interface. + +## Caller responsibilities + +- authenticate the caller and authorize every submitted evidence record; +- enforce tenant, workspace, purpose, retention, and export policy; +- minimize text and participant evidence according to data classification; +- retain provider credentials, raw access tokens, browser sessions, and unrelated mailbox content inside the caller boundary; +- pin and record the immutable LineageWeave artifact used for an analysis; +- retain adjudicator and provider/model provenance beside any model-backed result; +- verify the returned contract version and result digest before persistence or display. + +## LineageWeave boundary + +- rejects unsafe opaque references, unknown fields, invalid timestamps, duplicate evidence, and over-budget inferred work; +- returns only references present in the admitted request from the supported analysis adapter; +- distinguishes observed caller relations from inferred reconstruction; +- does not rescore or disclose a caller-observed child to the optional LLM merely to generate an alternative edge that would be discarded; +- never promotes a proposed project projection to caller authority; +- performs no provider mutation and receives no provider credential through this contract. diff --git a/docs/contracts/external-lineage-analysis-v1.versioning.md b/docs/contracts/external-lineage-analysis-v1.versioning.md new file mode 100644 index 000000000..9b0640ee3 --- /dev/null +++ b/docs/contracts/external-lineage-analysis-v1.versioning.md @@ -0,0 +1,10 @@ +# External lineage analysis versioning policy + +- `contract_version` follows semantic versioning independently from the LineageWeave package version. +- Unknown major versions fail closed. +- Additive optional fields require a new minor contract revision and corresponding consumer fixtures. +- Vocabulary changes, field semantic changes, required-field changes, digest changes, or truth-status changes require a new major contract version. +- A released schema, example, parser, serializer, digest algorithm, and consumer fixtures remain immutable for that contract version. +- Consumers must record both the contract version and immutable LineageWeave package/service artifact identity. The contract version alone does not identify the reconstruction implementation. +- Model-backed runs must additionally retain the adjudicator implementation and provider/model revision outside the v1 result payload; canonical digest determinism must not be described as provider repeatability. +- Naruon and other consumers pin an immutable LineageWeave release or service artifact and verify compatibility before enabling the integration. diff --git a/docs/doctoring/EXTERNAL_LINEAGE_CONTRACT_REFERENCES.md b/docs/doctoring/EXTERNAL_LINEAGE_CONTRACT_REFERENCES.md new file mode 100644 index 000000000..0b88b76e3 --- /dev/null +++ b/docs/doctoring/EXTERNAL_LINEAGE_CONTRACT_REFERENCES.md @@ -0,0 +1,23 @@ +# External Lineage Contract References + +## Product traceability + +| Source | Product decision | +|---|---| +| RFC 3339 | Require offset-aware occurrence, availability, and knowledge-cutoff timestamps. | +| RFC 5322 | Preserve Internet-message identity and reply metadata as caller-observed evidence rather than semantic inference. | +| RFC 5256 | Keep standards-based email threading evidence distinct from LineageWeave reconstruction. | +| W3C PROV-O | Return evidence references, truth status, analysis identity, and provenance-friendly result artifacts. | +| W3C OWL-Time | Separate occurrence time from evidence availability and enforce cutoff safety by availability. | + +## References — APA 7th + +Klyne, G., & Newman, C. (2002). *Date and time on the Internet: Timestamps* (RFC 3339). RFC Editor. https://doi.org/10.17487/RFC3339 + +Crispin, M., & Murchison, K. (2008). *Internet Message Access Protocol—SORT and THREAD extensions* (RFC 5256). RFC Editor. https://doi.org/10.17487/RFC5256 + +Resnick, P. W. (2008). *Internet message format* (RFC 5322). RFC Editor. https://doi.org/10.17487/RFC5322 + +World Wide Web Consortium. (2013). *PROV-O: The PROV ontology*. https://www.w3.org/TR/prov-o/ + +World Wide Web Consortium. (2017). *Time ontology in OWL*. https://www.w3.org/TR/owl-time/ diff --git a/docs/doctoring/EXTERNAL_LINEAGE_CONTRACT_TRACEABILITY.md b/docs/doctoring/EXTERNAL_LINEAGE_CONTRACT_TRACEABILITY.md new file mode 100644 index 000000000..b46360edf --- /dev/null +++ b/docs/doctoring/EXTERNAL_LINEAGE_CONTRACT_TRACEABILITY.md @@ -0,0 +1,12 @@ +# External Lineage Contract Traceability + +| Requirement | Product decision | Implementation | Evidence | +|---|---|---|---| +| Caller authorization remains authoritative | Accept only caller-projected evidence and opaque references | `lineageweave.external_lineage_contract` | strict parser and hostile-input tests | +| Historical answers exclude future evidence | Filter by `available_at <= knowledge_cutoff` | `lineageweave.external_lineage_analysis` | cutoff inclusion/exclusion tests | +| RFC relations remain distinct | Explicit parent relations serialize as observed relation codes | execution adapter | observed-parent precedence tests | +| Semantic lineage remains inferred | Reconstructed edges use `truth_status_code=inferred` | execution adapter | result contract tests | +| Optional LLM absence is honest | Return `not_requested` or `unavailable`; do not fabricate a score | execution adapter | LLM policy tests | +| Work is bounded before provider calls | Enforce record count, candidate window, and maximum pair evaluations | parser and execution adapter | pair-budget tests | +| Project state is not silently mutated | Return only `proposed` project projections | contract/result validator | project truth-status tests | +| Consumer compatibility is machine-checkable | Publish JSON Schema and canonical request/result digests | schema and contract module | schema drift and digest tests | diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 7704fa748..4d42ca5c3 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -354,7 +354,7 @@ this file per §3.5 of the prior snapshot). | #277 | TEPP: persist accepted receipts, poll completed results, keep measurement authority distinct | #657 consumer lifecycle; executable producer route remains unavailable | | #280 | Full project-lifecycle history and handover intervals | #640 adds case/project journeys and #663 adds evidence-backed Project exploration; authoritative lifecycle reconciliation remains #284 | | #284 | Authoritative lifecycle ingestion and idempotent reconciliation | No active delivery PR confirmed | -| #338 | Evidence-bounded email/project lineage contract for Naruon consumption | Missing on protected `main`; #343 merged only into a non-default stack, while #355 is a distinct calendar-consumer contract and is not delivery evidence for email/project lineage | +| #338 | Evidence-bounded email/project lineage contract for Naruon consumption | #704 recreates the provider-side contract on current `main` without arbitrary fusion weights; #343 remains only a non-default-stack merge and #355 is a distinct calendar contract | | #611 | Decompose closed PR #490 ADR 0133–0137 evidence without transferring stale branch state | #631 supplies the current-main inventory only; focused implementation PRs and tests for every unmet criterion are still required | ## 5. Open product and technical gaps @@ -383,6 +383,7 @@ this file per §3.5 of the prior snapshot). | Design tokens and repeated objects | Token extraction started; sanitized Figma Event Lineage desktop/mobile frames exist, while other repeated product surfaces remain incomplete | Tokens in CSS + Storybook stories for board, popup, DAG, Ask, calendar, forms, charts; same-viewport Figma/runtime visual comparison before release | | Frontend delivery performance | #644 implements a native dynamic-import boundary for conditional workspace surfaces and retains accessible loading/error states; exact-head checks passed but the PR is not protected-main evidence | Merge #644 normally, rebuild the protected-main production bundle, and retain the measured chunk inventory rather than raising the warning limit | | External integrations | Search, Zotero, calendar, Keyverse, orchestrator, RankWeave, ThreadWeave, TEPP, DiskSage, wardnet | Provider conformance, failure/reconciliation behavior, and provenance-bearing integration evidence | +| Naruon email/project lineage | #704 provides a strict store-agnostic v1 contract, opaque evidence references, observed/inferred truth separation, knowledge-cutoff admission, and explicit unavailable states. Inferred edges require an injected provenance-bearing fast-mlsirm estimate; no local default weight exists | Merge #704 through protected `main`, publish an immutable attested artifact, then enable the Naruon consumer only against that released version and its contract fixtures | | MSA / modular reuse | LineageWeave must run standalone and as a consumer of org packages | Do not reimplement RankWeave/TEPP/orchestrator/ThreadWeave/Keyverse; fix upstream and PR there | | Product contract authority | The current LineageWeave PRD records exact-case ecosystem authorities. TEPP, fast-mlsirm, keyverse, ThreadWeave, and RankWeave PR #41 have standalone PRDs; RankWeave's remains unmerged. contextual-orchestrator, disksage, and wardnet still rely on product/architecture documents, and naruon has only a scoped Topic Intelligence PRD | Keep ADRs normative, preserve canonical repository case in machine references, land the pending PRDs, and add standalone PRDs in each remaining owning repository before cross-product release claims exceed its documented boundary | | Release quality | PR #660 is now on protected `main`; its pre-merge full Python suite passed 1,352 tests with 17 skips, but release-wide frontend, Storybook, security, browser, and runtime acceptance remain unproven on one exact protected head | Repository-wide coverage, docstrings, Storybook, security, browser, and release evidence on one exact head | diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index c86da1e05..aedeb1aa7 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -9,6 +9,25 @@ from .affiliate_tree import build_affiliate_forest from .corporate_hierarchy_resolution import resolve_corporate_entity from .entity_relationship_classification import OrganizationRelationship +from .external_lineage_analysis import analyze_external_lineage +from .external_lineage_contract import ( + CONTRACT_VERSION, + ChannelEvidence, + ExplicitParent, + LineageAnalysisPolicy, + LineageAnalysisRequest, + LineageAnalysisResult, + LineageContractError, + LineageEdgeResult, + LineageEvidenceRecord, + LineageLimitation, + ProjectProjection, + parse_lineage_analysis_request, + request_digest, + result_digest, + serialize_lineage_analysis_request, + serialize_lineage_analysis_result, +) from .knowledge_graph import random_walk_with_restart, select_related_nodes from .lineage_persistence import ( CHANNEL_EVIDENCE_TOLERANCE, @@ -53,6 +72,16 @@ __all__ = [ "CHANNEL_EVIDENCE_TOLERANCE", + "CONTRACT_VERSION", + "ChannelEvidence", + "ExplicitParent", + "LineageAnalysisPolicy", + "LineageAnalysisRequest", + "LineageAnalysisResult", + "LineageContractError", + "LineageEdgeResult", + "LineageEvidenceRecord", + "LineageLimitation", "NARUON_CALENDAR_MEDIA_TYPE", "NARUON_CALENDAR_SCHEMA_VERSION", "NARUON_CALENDAR_UNAVAILABLE_NEXT_ACTION", @@ -71,12 +100,14 @@ "Edge", "OrganizationRelationship", "PostSummary", + "ProjectProjection", "ProvAssertion", "ProvGraph", "ProvLiteral", "ProvValidationError", "Record", "Tree", + "analyze_external_lineage", "build_affiliate_forest", "build_workspace_naruon_client", "cited_post_summaries", @@ -84,14 +115,19 @@ "lineage_edge_specs", "load_observed_calendar_events", "occurrence_to_workspace_event", + "parse_lineage_analysis_request", "parse_naruon_calendar_page", "random_walk_with_restart", "rank_channel_evidence", "reconstruct", "reconstruction_version", + "request_digest", "resolve_corporate_entity", + "result_digest", "select_related_nodes", "sentence_excerpts", + "serialize_lineage_analysis_request", + "serialize_lineage_analysis_result", ] __version__ = "2.17.0" diff --git a/lineageweave/adjudication_client.py b/lineageweave/adjudication_client.py index 60cfd61ce..646fab91c 100644 --- a/lineageweave/adjudication_client.py +++ b/lineageweave/adjudication_client.py @@ -28,6 +28,10 @@ def judge(self, candidate_label: str, record_label: str) -> float: raise NotImplementedError +class AdjudicationClientError(RuntimeError): + """The provider returned an unusable adjudication response.""" + + class NullAdjudicationClient: """No LLM orchestrator configured -- the llm channel is skipped.""" @@ -39,6 +43,20 @@ def judge(self, candidate_label: str, record_label: str) -> float: # pragma: no _CONFIDENCE_PATTERN = re.compile(r"([01](?:\.\d+)?)") +_STRICT_CONFIDENCE_PATTERN = re.compile(r"(?:0(?:\.\d+)?|1(?:\.0+)?)") + + +def parse_confidence_response(content: object) -> float: + """Parse the provider's number-only confidence response strictly.""" + + if not isinstance(content, str): + raise AdjudicationClientError("provider confidence response was not text") + normalized = content.strip() + if _STRICT_CONFIDENCE_PATTERN.fullmatch(normalized) is None: + raise AdjudicationClientError( + "provider confidence response was not a number in 0..1" + ) + return float(normalized) def judge_prompt(candidate_label: str, record_label: str) -> str: @@ -107,5 +125,7 @@ def judge(self, candidate_label: str, record_label: str) -> float: try: content = chat_completion_content(body) except (TypeError, ValueError) as exc: - raise HttpClientError("adjudication response did not contain text") from exc - return parse_confidence(content) + raise AdjudicationClientError( + "provider response did not contain one chat message" + ) from exc + return parse_confidence_response(content) diff --git a/lineageweave/external_lineage.py b/lineageweave/external_lineage.py new file mode 100644 index 000000000..6a875ff17 --- /dev/null +++ b/lineageweave/external_lineage.py @@ -0,0 +1,41 @@ +"""Stable public package surface for external lineage consumers.""" + +from .external_lineage_analysis import analyze_external_lineage +from .external_lineage_contract import ( + CONTRACT_VERSION, + ChannelEvidence, + ExplicitParent, + LineageAnalysisPolicy, + LineageAnalysisRequest, + LineageAnalysisResult, + LineageContractError, + LineageEdgeResult, + LineageEvidenceRecord, + LineageLimitation, + ProjectProjection, + parse_lineage_analysis_request, + request_digest, + result_digest, + serialize_lineage_analysis_request, + serialize_lineage_analysis_result, +) + +__all__ = [ + "CONTRACT_VERSION", + "ChannelEvidence", + "ExplicitParent", + "LineageAnalysisPolicy", + "LineageAnalysisRequest", + "LineageAnalysisResult", + "LineageContractError", + "LineageEdgeResult", + "LineageEvidenceRecord", + "LineageLimitation", + "ProjectProjection", + "analyze_external_lineage", + "parse_lineage_analysis_request", + "request_digest", + "result_digest", + "serialize_lineage_analysis_request", + "serialize_lineage_analysis_result", +] diff --git a/lineageweave/external_lineage_analysis.py b/lineageweave/external_lineage_analysis.py new file mode 100644 index 000000000..1ac457afc --- /dev/null +++ b/lineageweave/external_lineage_analysis.py @@ -0,0 +1,601 @@ +"""Execute the external lineage contract through the core reconstruction kernel. + +This adapter is deliberately store-agnostic. It accepts an already parsed, +caller-authorized request, applies available-time cutoff rules, invokes the +existing deterministic/optional-LLM reconstruction kernel, and returns only +opaque caller references plus evidence-bounded result metadata. +""" + +from __future__ import annotations + +import math +from collections import defaultdict +from dataclasses import replace + +from .adjudication_client import ( + AdjudicationClient, + NullAdjudicationClient, +) +from .channel_weight_estimation import ChannelWeightEstimate +from .external_lineage_contract import ( + CONTRACT_VERSION, + ChannelEvidence, + LineageAnalysisRequest, + LineageAnalysisResult, + LineageContractError, + LineageEdgeResult, + LineageEvidenceRecord, + LineageLimitation, + ProjectProjection, + parse_lineage_analysis_request, + result_digest, + serialize_lineage_analysis_request, +) +from .models import Record +from .reconstruct import _best_parent, active_weights + + +def _contract_error(code: str, message: str, field: str | None = None) -> None: + """Raise a stable execution-time contract error.""" + + raise LineageContractError(code, message, field=field) + + +class _BoundedAdjudicationClient: + """Keep provider channel scores inside the fusion contract boundary.""" + + available = True + + def __init__(self, client: AdjudicationClient) -> None: + """Wrap one available client without changing its provider behavior.""" + + self._client = client + + def judge(self, candidate_label: str, record_label: str) -> float: + """Return one finite unit-interval score or fail with a stable code.""" + + try: + score = self._client.judge(candidate_label, record_label) + except Exception as exc: + raise LineageContractError( + "llm_channel_error", + "LLM channel returned an unusable provider response", + field="llm", + ) from exc + if isinstance(score, bool) or not isinstance(score, (int, float)): + _contract_error( + "channel_score_out_of_bounds", + "LLM channel score must be finite and within 0..1", + "llm", + ) + number = float(score) + if not math.isfinite(number) or not 0.0 <= number <= 1.0: + _contract_error( + "channel_score_out_of_bounds", + "LLM channel score must be finite and within 0..1", + "llm", + ) + return number + + +def _validated_request(request: LineageAnalysisRequest) -> LineageAnalysisRequest: + """Round-trip a dataclass through the public parser before execution.""" + + return parse_lineage_analysis_request( + serialize_lineage_analysis_request(request) + ) + + +def _validate_explicit_parent_relations( + records: tuple[LineageEvidenceRecord, ...], +) -> None: + """Validate caller-observed parent relations before cutoff filtering.""" + + by_ref = {record.evidence_ref: record for record in records} + for child in records: + explicit = child.explicit_parent + if explicit is None: + continue + if explicit.evidence_ref == child.evidence_ref: + _contract_error( + "explicit_parent_self_reference", + "an evidence record cannot be its own parent", + child.evidence_ref, + ) + parent = by_ref.get(explicit.evidence_ref) + if parent is None: + _contract_error( + "explicit_parent_missing", + "explicit parent is absent from the request", + child.evidence_ref, + ) + if parent.group_ref != child.group_ref: + _contract_error( + "explicit_parent_group_mismatch", + "explicit parent and child must share one group", + child.evidence_ref, + ) + if parent.occurred_at > child.occurred_at: + _contract_error( + "explicit_parent_after_child", + "explicit parent occurs after the child", + child.evidence_ref, + ) + + parent_by_child = { + child.evidence_ref: child.explicit_parent.evidence_ref + for child in records + if child.explicit_parent is not None + } + for start_ref in parent_by_child: + current_ref = start_ref + visited: set[str] = set() + while current_ref in parent_by_child: + if current_ref in visited: + _contract_error( + "explicit_parent_cycle", + "explicit parent relations must form an acyclic graph", + start_ref, + ) + visited.add(current_ref) + current_ref = parent_by_child[current_ref] + + +def _selected_llm( + request: LineageAnalysisRequest, + llm: AdjudicationClient | None, + weight_estimate: ChannelWeightEstimate | None, +) -> tuple[AdjudicationClient, str]: + """Apply the explicit LLM admission policy and return its result status.""" + + if not request.policy.allow_llm: + return NullAdjudicationClient(), "not_requested" + if ( + llm is None + or not getattr(llm, "available", False) + or weight_estimate is None + or "llm" not in weight_estimate.weights + ): + return NullAdjudicationClient(), "unavailable" + return _BoundedAdjudicationClient(llm), "completed" + + +def _included_records( + request: LineageAnalysisRequest, +) -> tuple[ + tuple[LineageEvidenceRecord, ...], + tuple[LineageEvidenceRecord, ...], +]: + """Partition evidence by available time, not occurrence time.""" + + if request.knowledge_cutoff is None: + return request.records, () + included = tuple( + record + for record in request.records + if record.available_at <= request.knowledge_cutoff + ) + excluded = tuple( + record + for record in request.records + if record.available_at > request.knowledge_cutoff + ) + return included, excluded + + +def _ordered_contract_groups( + records: tuple[LineageEvidenceRecord, ...], +) -> tuple[tuple[LineageEvidenceRecord, ...], ...]: + """Return deterministic groups ordered by time and opaque reference.""" + + grouped: dict[str, list[LineageEvidenceRecord]] = defaultdict(list) + for record in records: + grouped[record.group_ref].append(record) + return tuple( + tuple( + sorted( + grouped[group_ref], + key=lambda item: (item.occurred_at, item.evidence_ref), + ) + ) + for group_ref in sorted(grouped) + ) + + +def _has_inference_candidate( + records: tuple[LineageEvidenceRecord, ...], +) -> bool: + """Return whether a same-group predecessor could support inference.""" + + return any( + index > 0 and record.explicit_parent is None + for group_records in _ordered_contract_groups(records) + for index, record in enumerate(group_records) + ) + + +def _pair_evaluation_count( + records: tuple[LineageEvidenceRecord, ...], + candidate_window: int, +) -> int: + """Count only candidate pairs that require inferred parent selection.""" + + included_refs = {record.evidence_ref for record in records} + explicit_children_by_parent: dict[str, set[str]] = defaultdict(set) + for record in records: + if ( + record.explicit_parent is not None + and record.explicit_parent.evidence_ref in included_refs + ): + explicit_children_by_parent[ + record.explicit_parent.evidence_ref + ].add(record.evidence_ref) + + def explicit_descendants(evidence_ref: str) -> set[str]: + """Return observed descendants excluded from the pair-work budget.""" + + descendants: set[str] = set() + pending = list(explicit_children_by_parent.get(evidence_ref, ())) + while pending: + descendant = pending.pop() + if descendant in descendants: + continue + descendants.add(descendant) + pending.extend(explicit_children_by_parent.get(descendant, ())) + return descendants + + pair_count = 0 + for group_records in _ordered_contract_groups(records): + for index, record in enumerate(group_records): + if record.explicit_parent is not None: + continue + candidates = group_records[max(0, index - candidate_window) : index] + descendants = explicit_descendants(record.evidence_ref) + pair_count += sum( + candidate.evidence_ref not in descendants + for candidate in candidates + ) + return pair_count + + +def _enforce_pair_budget( + records: tuple[LineageEvidenceRecord, ...], + request: LineageAnalysisRequest, +) -> int: + """Reject excess pair work before optional LLM/provider activity.""" + + pair_count = _pair_evaluation_count( + records, + request.policy.candidate_window, + ) + if pair_count > request.policy.maximum_pair_evaluations: + _contract_error( + "pair_evaluation_budget_exceeded", + "candidate-pair work exceeds the declared maximum", + "policy.maximum_pair_evaluations", + ) + return pair_count + + +def _core_record(record: LineageEvidenceRecord) -> Record: + """Convert one contract record to the core reconstruction shape.""" + + return Record( + record_id=record.evidence_ref, + group_key=record.group_ref, + label=record.label, + occurred_at=record.occurred_at, + secondary_key=record.secondary_key or "", + ) + + +def _channel_evidence( + channel_scores: dict[str, float], + weights: dict[str, float], +) -> tuple[ChannelEvidence, ...]: + """Project finite active scores with their normalized contributions.""" + + projected: list[ChannelEvidence] = [] + for channel_code in sorted(channel_scores): + score = float(channel_scores[channel_code]) + weight = float(weights[channel_code]) + contribution = score * weight + values = (score, weight, contribution) + if not all( + math.isfinite(value) and 0.0 <= value <= 1.0 + for value in values + ): + _contract_error( + "channel_score_out_of_bounds", + "channel values must be finite within 0..1", + channel_code, + ) + projected.append( + ChannelEvidence( + channel_code, + score, + weight, + contribution, + ) + ) + return tuple(projected) + + +def _inferred_edges( + records: tuple[LineageEvidenceRecord, ...], + llm: AdjudicationClient, + request: LineageAnalysisRequest, + weight_estimate: ChannelWeightEstimate, +) -> list[LineageEdgeResult]: + """Select inferred parents without rescoring explicit observed children.""" + + if not records: + return [] + if not weight_estimate.estimation_method_code.strip(): + _contract_error( + "weight_provenance_missing", + "channel weights require an estimation method code", + "weight_estimate.estimation_method_code", + ) + if weight_estimate.sample_pair_count < 1: + _contract_error( + "weight_provenance_missing", + "channel weights require a positive estimation sample count", + "weight_estimate.sample_pair_count", + ) + required_channels = {"temporal", "secondary_key", "text"} + if not required_channels.issubset(weight_estimate.weights): + _contract_error( + "weight_channels_missing", + "the estimate must cover every deterministic reconstruction channel", + "weight_estimate.weights", + ) + weights = active_weights(llm, weight_estimate.weights) + if not weights or not math.isclose(sum(weights.values()), 1.0, abs_tol=1e-9): + _contract_error( + "weight_sum_mismatch", + "active estimated channel weights must normalize to one", + "weight_estimate.weights", + ) + included_refs = {record.evidence_ref for record in records} + explicit_children_by_parent: dict[str, set[str]] = defaultdict(set) + for record in records: + if ( + record.explicit_parent is not None + and record.explicit_parent.evidence_ref in included_refs + ): + explicit_children_by_parent[ + record.explicit_parent.evidence_ref + ].add(record.evidence_ref) + + def explicit_descendants(evidence_ref: str) -> set[str]: + """Return observed descendants that cannot become inferred parents.""" + + descendants: set[str] = set() + pending = list(explicit_children_by_parent.get(evidence_ref, ())) + while pending: + descendant = pending.pop() + if descendant in descendants: + continue + descendants.add(descendant) + pending.extend(explicit_children_by_parent.get(descendant, ())) + return descendants + + edges: list[LineageEdgeResult] = [] + for group_records in _ordered_contract_groups(records): + core_records = [_core_record(record) for record in group_records] + for index, source_record in enumerate(group_records): + if source_record.explicit_parent is not None: + continue + candidates = core_records[ + max(0, index - request.policy.candidate_window) : index + ] + cycle_forming_parents = explicit_descendants( + source_record.evidence_ref + ) + candidates = [ + candidate + for candidate in candidates + if candidate.record_id not in cycle_forming_parents + ] + parent_choice = _best_parent( + core_records[index], + candidates, + llm, + weights, + request.policy.minimum_fused_score, + ) + if parent_choice is None: + continue + parent, fused_score, channel_scores = parent_choice + edges.append( + LineageEdgeResult( + parent_evidence_ref=parent.record_id, + child_evidence_ref=source_record.evidence_ref, + relation_type_code="reconstructed_continuation", + truth_status_code="inferred", + fused_score=float(fused_score), + channel_evidence=_channel_evidence( + channel_scores, + weights, + ), + ) + ) + return edges + + +def _explicit_edges( + included: tuple[LineageEvidenceRecord, ...], +) -> tuple[ + list[LineageEdgeResult], + set[str], + list[LineageLimitation], +]: + """Project included caller-observed parent relations ahead of inference.""" + + included_refs = {record.evidence_ref for record in included} + edges: list[LineageEdgeResult] = [] + explicit_children: set[str] = set() + limitations: list[LineageLimitation] = [] + for child in included: + explicit = child.explicit_parent + if explicit is None: + continue + explicit_children.add(child.evidence_ref) + if explicit.evidence_ref not in included_refs: + limitations.append( + LineageLimitation( + "explicit_parent_after_cutoff", + child.evidence_ref, + ( + "The caller-observed parent was unavailable at " + "the requested cutoff." + ), + ) + ) + continue + edges.append( + LineageEdgeResult( + parent_evidence_ref=explicit.evidence_ref, + child_evidence_ref=child.evidence_ref, + relation_type_code=explicit.relation_code, + truth_status_code="observed", + fused_score=1.0, + channel_evidence=( + ChannelEvidence( + explicit.relation_code, + 1.0, + 1.0, + 1.0, + ), + ), + ) + ) + return edges, explicit_children, limitations + + +def _project_groups( + records: tuple[LineageEvidenceRecord, ...], +) -> tuple[ProjectProjection, ...]: + """Group included project evidence without crossing caller groups.""" + + grouped: dict[tuple[str, str], list[str]] = defaultdict(list) + for record in records: + if record.project_ref is not None: + grouped[(record.group_ref, record.project_ref)].append( + record.evidence_ref + ) + return tuple( + ProjectProjection( + group_ref, + project_ref, + tuple(sorted(evidence_refs)), + "proposed", + ) + for (group_ref, project_ref), evidence_refs in sorted( + grouped.items() + ) + ) + + +def analyze_external_lineage( + request: LineageAnalysisRequest, + *, + llm: AdjudicationClient | None = None, + weight_estimate: ChannelWeightEstimate | None = None, +) -> LineageAnalysisResult: + """Analyze bounded caller evidence and return a deterministic result. + + The function performs no persistence or network access itself. An optional + client is used only when ``request.policy.allow_llm`` is true and the + supplied client explicitly reports availability. + """ + + validated = _validated_request(request) + _validate_explicit_parent_relations(validated.records) + included, excluded = _included_records(validated) + _enforce_pair_budget(included, validated) + selected_llm, llm_status = _selected_llm(validated, llm, weight_estimate) + + inferred = ( + _inferred_edges(included, selected_llm, validated, weight_estimate) + if weight_estimate is not None + else [] + ) + explicit, explicit_children, explicit_limitations = _explicit_edges( + included + ) + edges = [ + edge + for edge in inferred + if edge.child_evidence_ref not in explicit_children + ] + edges.extend(explicit) + + limitations = [ + LineageLimitation( + "evidence_after_cutoff_excluded", + record.evidence_ref, + ( + "Evidence was first available after the requested " + "knowledge cutoff." + ), + ) + for record in excluded + ] + if weight_estimate is None and _has_inference_candidate(included): + limitations.append( + LineageLimitation( + "channel_weights_unavailable", + None, + ( + "No provenance-bearing psychometric channel-weight estimate " + "was supplied, so inferred continuation edges are unavailable." + ), + ) + ) + limitations.extend(explicit_limitations) + + edge_order = { + record.evidence_ref: (record.group_ref, record.occurred_at, record.evidence_ref) + for record in included + } + result = LineageAnalysisResult( + contract_version=CONTRACT_VERSION, + analysis_id=validated.analysis_id, + analysis_scope_code=validated.analysis_scope_code, + knowledge_cutoff=validated.knowledge_cutoff, + included_evidence_refs=tuple( + sorted(record.evidence_ref for record in included) + ), + excluded_evidence_refs=tuple( + sorted(record.evidence_ref for record in excluded) + ), + llm_status_code=llm_status, # type: ignore[arg-type] + edges=tuple( + sorted( + edges, + key=lambda item: ( + edge_order[item.child_evidence_ref], + item.parent_evidence_ref, + item.relation_type_code, + ), + ) + ), + project_projections=_project_groups(included), + limitations=tuple( + sorted( + limitations, + key=lambda item: ( + item.limitation_code, + item.evidence_ref or "", + item.message, + ), + ) + ), + result_digest="", + ) + return replace( + result, + result_digest=result_digest(result), + ) diff --git a/lineageweave/external_lineage_contract.py b/lineageweave/external_lineage_contract.py new file mode 100644 index 000000000..aa926e509 --- /dev/null +++ b/lineageweave/external_lineage_contract.py @@ -0,0 +1,949 @@ +"""Versioned store-agnostic contract for external lineage consumers. + +The contract accepts only bounded caller-authorized evidence references. It +contains no provider credential, database, mailbox, or network behavior. A +consumer such as Naruon can therefore submit a minimized evidence projection +without granting LineageWeave authority over the consumer's source records. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +from dataclasses import dataclass, replace +from datetime import UTC, datetime +from typing import Final, Literal, cast + +CONTRACT_VERSION: Final = "1.0.0" +MAX_RECORD_COUNT: Final = 500 +MAX_REFERENCE_LENGTH: Final = 160 +MAX_LABEL_LENGTH: Final = 2_000 +MAX_CANDIDATE_WINDOW: Final = 200 +MAX_PAIR_EVALUATIONS: Final = 5_000 + +AnalysisScopeCode = Literal["email_lineage", "project_history", "generic_lineage"] +SourceKindCode = Literal["email", "task", "commitment", "project_event", "generic"] +CallerTruthStatusCode = Literal["observed", "authoritative_in_caller"] +ExplicitRelationCode = Literal["rfc_reply", "provider_reply", "manual_parent"] +ResultTruthStatusCode = Literal["observed", "inferred", "proposed"] +LlmStatusCode = Literal["not_requested", "unavailable", "completed"] + +_ANALYSIS_SCOPES = frozenset({"email_lineage", "project_history", "generic_lineage"}) +_SOURCE_KINDS = frozenset({"email", "task", "commitment", "project_event", "generic"}) +_CALLER_TRUTH_STATUSES = frozenset({"observed", "authoritative_in_caller"}) +_EXPLICIT_RELATIONS = frozenset({"rfc_reply", "provider_reply", "manual_parent"}) +_EDGE_TRUTH_STATUSES = frozenset({"observed", "inferred"}) +_LLM_STATUSES = frozenset({"not_requested", "unavailable", "completed"}) +_OPAQUE_REFERENCE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+\-]*$") +_RESULT_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") +_SCORE_TOLERANCE: Final = 1e-9 + + +class LineageContractError(ValueError): + """A fail-closed request or result contract violation. + + Attributes: + code: Stable machine-readable reason code. + field: Optional dotted field path associated with the violation. + """ + + def __init__(self, code: str, message: str, *, field: str | None = None) -> None: + """Initialize one stable contract error without embedding source evidence.""" + + self.code = code + self.field = field + suffix = f" ({field})" if field else "" + super().__init__(f"{message}{suffix}") + + +@dataclass(frozen=True) +class ExplicitParent: + """One caller-observed immediate parent relation.""" + + evidence_ref: str + relation_code: ExplicitRelationCode + + +@dataclass(frozen=True) +class LineageEvidenceRecord: + """One bounded caller-owned evidence record admitted for analysis.""" + + evidence_ref: str + group_ref: str + source_kind_code: SourceKindCode + truth_status_code: CallerTruthStatusCode + label: str + occurred_at: datetime + available_at: datetime + secondary_key: str | None = None + project_ref: str | None = None + explicit_parent: ExplicitParent | None = None + + +@dataclass(frozen=True) +class LineageAnalysisPolicy: + """Bounded reconstruction policy selected by the caller.""" + + candidate_window: int + maximum_pair_evaluations: int + minimum_fused_score: float + allow_llm: bool + + +@dataclass(frozen=True) +class LineageAnalysisRequest: + """Strict versioned request for external lineage reconstruction.""" + + contract_version: str + analysis_id: str + analysis_scope_code: AnalysisScopeCode + knowledge_cutoff: datetime | None + policy: LineageAnalysisPolicy + records: tuple[LineageEvidenceRecord, ...] + + +@dataclass(frozen=True) +class ChannelEvidence: + """One active reconstruction channel's exact normalized contribution.""" + + channel_code: str + score: float + weight: float + contribution: float + + +@dataclass(frozen=True) +class LineageEdgeResult: + """One observed or inferred edge between caller-owned evidence records.""" + + parent_evidence_ref: str + child_evidence_ref: str + relation_type_code: str + truth_status_code: Literal["observed", "inferred"] + fused_score: float + channel_evidence: tuple[ChannelEvidence, ...] + + +@dataclass(frozen=True) +class ProjectProjection: + """A proposed project grouping bounded to one caller group.""" + + group_ref: str + project_ref: str + evidence_refs: tuple[str, ...] + truth_status_code: Literal["proposed"] = "proposed" + + +@dataclass(frozen=True) +class LineageLimitation: + """A machine-readable limitation disclosed with an analysis result.""" + + limitation_code: str + evidence_ref: str | None + message: str + + +@dataclass(frozen=True) +class LineageAnalysisResult: + """Deterministic external lineage result containing no caller credential.""" + + contract_version: str + analysis_id: str + analysis_scope_code: AnalysisScopeCode + knowledge_cutoff: datetime | None + included_evidence_refs: tuple[str, ...] + excluded_evidence_refs: tuple[str, ...] + llm_status_code: LlmStatusCode + edges: tuple[LineageEdgeResult, ...] + project_projections: tuple[ProjectProjection, ...] + limitations: tuple[LineageLimitation, ...] + result_digest: str + + +def _raise(code: str, message: str, field: str | None = None) -> None: + """Raise one stable contract error.""" + + raise LineageContractError(code, message, field=field) + + +def _object( + value: object, + *, + field: str, + allowed: frozenset[str], + required: frozenset[str], +) -> dict[str, object]: + """Validate a strict object and reject unknown or missing fields.""" + + if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): + _raise("invalid_field_type", "expected an object", field) + typed = cast(dict[str, object], value) + unknown = sorted(set(typed) - allowed) + if unknown: + _raise("unknown_field", f"unknown field {unknown[0]!r}", f"{field}.{unknown[0]}") + missing = sorted(required - set(typed)) + if missing: + _raise("missing_field", f"missing required field {missing[0]!r}", f"{field}.{missing[0]}") + return typed + + +def _string(value: object, *, field: str, minimum: int = 1, maximum: int) -> str: + """Return one trimmed bounded string or fail closed.""" + + if not isinstance(value, str): + _raise("invalid_field_type", "expected a string", field) + normalized = value.strip() + if not minimum <= len(normalized) <= maximum: + _raise( + "text_length_out_of_bounds", + f"length must be {minimum}..{maximum}", + field, + ) + return normalized + + +def _opaque_reference( + value: object, + *, + field: str, + optional: bool = False, +) -> str | None: + """Validate one bounded opaque identifier that cannot be a URL.""" + + if value is None and optional: + return None + normalized = _string(value, field=field, maximum=MAX_REFERENCE_LENGTH) + if "://" in normalized or not _OPAQUE_REFERENCE.fullmatch(normalized): + _raise( + "unsafe_opaque_reference", + "reference must be opaque and whitespace-free", + field, + ) + return normalized + + +def _timestamp(value: object, *, field: str, optional: bool = False) -> datetime | None: + """Parse an offset-aware RFC 3339 timestamp and normalize it to UTC.""" + + if value is None and optional: + return None + if not isinstance(value, str): + _raise("invalid_field_type", "expected an RFC 3339 string", field) + candidate = value[:-1] + "+00:00" if value.endswith("Z") else value + try: + parsed = datetime.fromisoformat(candidate) + except ValueError as exc: + raise LineageContractError( + "invalid_timestamp", + "invalid RFC 3339 timestamp", + field=field, + ) from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + _raise( + "timestamp_must_be_offset_aware", + "timestamp must carry an offset", + field, + ) + return parsed.astimezone(UTC) + + +def _enum( + value: object, + *, + field: str, + allowed: frozenset[str], + code: str, +) -> str: + """Validate one controlled vocabulary value.""" + + if not isinstance(value, str): + _raise("invalid_field_type", "expected a controlled string", field) + if value not in allowed: + _raise(code, f"unsupported value {value!r}", field) + return value + + +def _integer(value: object, *, field: str, minimum: int, maximum: int) -> int: + """Validate one integer policy value within an inclusive range.""" + + if isinstance(value, bool) or not isinstance(value, int): + _raise("invalid_field_type", "expected an integer", field) + if not minimum <= value <= maximum: + _raise( + "policy_value_out_of_bounds", + f"value must be {minimum}..{maximum}", + field, + ) + return value + + +def _number(value: object, *, field: str, minimum: float, maximum: float) -> float: + """Validate one finite numeric policy value within an inclusive range.""" + + if isinstance(value, bool) or not isinstance(value, (int, float)): + _raise("invalid_field_type", "expected a finite number", field) + number = float(value) + if not math.isfinite(number) or not minimum <= number <= maximum: + _raise( + "policy_value_out_of_bounds", + f"value must be {minimum}..{maximum}", + field, + ) + return number + + +def _boolean(value: object, *, field: str) -> bool: + """Validate a real boolean without accepting integer substitutes.""" + + if not isinstance(value, bool): + _raise("invalid_field_type", "expected a boolean", field) + return value + + +def _parse_explicit_parent(value: object, *, field: str) -> ExplicitParent | None: + """Parse one optional caller-observed parent relation.""" + + if value is None: + return None + payload = _object( + value, + field=field, + allowed=frozenset({"evidence_ref", "relation_code"}), + required=frozenset({"evidence_ref", "relation_code"}), + ) + reference = _opaque_reference(payload["evidence_ref"], field=f"{field}.evidence_ref") + relation = _enum( + payload["relation_code"], + field=f"{field}.relation_code", + allowed=_EXPLICIT_RELATIONS, + code="unknown_explicit_relation", + ) + return ExplicitParent( + cast(str, reference), + cast(ExplicitRelationCode, relation), + ) + + +def _parse_record(value: object, *, index: int) -> LineageEvidenceRecord: + """Parse one bounded evidence record from the request array.""" + + field = f"records[{index}]" + payload = _object( + value, + field=field, + allowed=frozenset( + { + "evidence_ref", + "group_ref", + "source_kind_code", + "truth_status_code", + "label", + "occurred_at", + "available_at", + "secondary_key", + "project_ref", + "explicit_parent", + } + ), + required=frozenset( + { + "evidence_ref", + "group_ref", + "source_kind_code", + "truth_status_code", + "label", + "occurred_at", + "available_at", + } + ), + ) + return LineageEvidenceRecord( + evidence_ref=cast( + str, + _opaque_reference(payload["evidence_ref"], field=f"{field}.evidence_ref"), + ), + group_ref=cast( + str, + _opaque_reference(payload["group_ref"], field=f"{field}.group_ref"), + ), + source_kind_code=cast( + SourceKindCode, + _enum( + payload["source_kind_code"], + field=f"{field}.source_kind_code", + allowed=_SOURCE_KINDS, + code="unknown_source_kind", + ), + ), + truth_status_code=cast( + CallerTruthStatusCode, + _enum( + payload["truth_status_code"], + field=f"{field}.truth_status_code", + allowed=_CALLER_TRUTH_STATUSES, + code="unknown_caller_truth_status", + ), + ), + label=_string( + payload["label"], + field=f"{field}.label", + maximum=MAX_LABEL_LENGTH, + ), + occurred_at=cast( + datetime, + _timestamp(payload["occurred_at"], field=f"{field}.occurred_at"), + ), + available_at=cast( + datetime, + _timestamp(payload["available_at"], field=f"{field}.available_at"), + ), + secondary_key=_opaque_reference( + payload.get("secondary_key"), + field=f"{field}.secondary_key", + optional=True, + ), + project_ref=_opaque_reference( + payload.get("project_ref"), + field=f"{field}.project_ref", + optional=True, + ), + explicit_parent=_parse_explicit_parent( + payload.get("explicit_parent"), + field=f"{field}.explicit_parent", + ), + ) + + +def _parse_policy(value: object) -> LineageAnalysisPolicy: + """Parse the bounded reconstruction policy.""" + + payload = _object( + value, + field="policy", + allowed=frozenset( + { + "candidate_window", + "maximum_pair_evaluations", + "minimum_fused_score", + "allow_llm", + } + ), + required=frozenset( + { + "candidate_window", + "maximum_pair_evaluations", + "minimum_fused_score", + "allow_llm", + } + ), + ) + return LineageAnalysisPolicy( + candidate_window=_integer( + payload["candidate_window"], + field="policy.candidate_window", + minimum=1, + maximum=MAX_CANDIDATE_WINDOW, + ), + maximum_pair_evaluations=_integer( + payload["maximum_pair_evaluations"], + field="policy.maximum_pair_evaluations", + minimum=1, + maximum=MAX_PAIR_EVALUATIONS, + ), + minimum_fused_score=_number( + payload["minimum_fused_score"], + field="policy.minimum_fused_score", + minimum=0.0, + maximum=1.0, + ), + allow_llm=_boolean(payload["allow_llm"], field="policy.allow_llm"), + ) + + +def parse_lineage_analysis_request(payload: object) -> LineageAnalysisRequest: + """Parse and strictly validate one external lineage analysis request.""" + + data = _object( + payload, + field="request", + allowed=frozenset( + { + "contract_version", + "analysis_id", + "analysis_scope_code", + "knowledge_cutoff", + "policy", + "records", + } + ), + required=frozenset( + { + "contract_version", + "analysis_id", + "analysis_scope_code", + "policy", + "records", + } + ), + ) + version = _string( + data["contract_version"], + field="contract_version", + maximum=16, + ) + if version != CONTRACT_VERSION: + _raise( + "unsupported_contract_version", + f"only contract version {CONTRACT_VERSION!r} is accepted", + "contract_version", + ) + records_payload = data["records"] + if not isinstance(records_payload, list): + _raise("invalid_field_type", "records must be an array", "records") + if not 1 <= len(records_payload) <= MAX_RECORD_COUNT: + _raise( + "record_count_out_of_bounds", + f"records must contain 1..{MAX_RECORD_COUNT} entries", + "records", + ) + records = tuple( + _parse_record(value, index=index) + for index, value in enumerate(records_payload) + ) + seen: set[str] = set() + for record in records: + if record.evidence_ref in seen: + _raise( + "duplicate_evidence_ref", + f"duplicate evidence reference {record.evidence_ref!r}", + "records", + ) + seen.add(record.evidence_ref) + return LineageAnalysisRequest( + contract_version=version, + analysis_id=cast( + str, + _opaque_reference(data["analysis_id"], field="analysis_id"), + ), + analysis_scope_code=cast( + AnalysisScopeCode, + _enum( + data["analysis_scope_code"], + field="analysis_scope_code", + allowed=_ANALYSIS_SCOPES, + code="unknown_analysis_scope", + ), + ), + knowledge_cutoff=_timestamp( + data.get("knowledge_cutoff"), + field="knowledge_cutoff", + optional=True, + ), + policy=_parse_policy(data["policy"]), + records=records, + ) + + +def _time_text(value: datetime | None) -> str | None: + """Serialize an aware timestamp canonically in UTC with a ``Z`` suffix.""" + + if value is None: + return None + if value.tzinfo is None or value.utcoffset() is None: + _raise( + "timestamp_must_be_offset_aware", + "result timestamp must carry an offset", + ) + utc = value.astimezone(UTC) + text = utc.isoformat(timespec="microseconds").replace( + ".000000+00:00", + "Z", + ) + return text.replace("+00:00", "Z") + + +def _record_dict(record: LineageEvidenceRecord) -> dict[str, object]: + """Serialize one evidence record without adding derived authority.""" + + explicit_parent: dict[str, object] | None = None + if record.explicit_parent is not None: + explicit_parent = { + "evidence_ref": record.explicit_parent.evidence_ref, + "relation_code": record.explicit_parent.relation_code, + } + return { + "evidence_ref": record.evidence_ref, + "group_ref": record.group_ref, + "source_kind_code": record.source_kind_code, + "truth_status_code": record.truth_status_code, + "label": record.label, + "occurred_at": _time_text(record.occurred_at), + "available_at": _time_text(record.available_at), + "secondary_key": record.secondary_key, + "project_ref": record.project_ref, + "explicit_parent": explicit_parent, + } + + +def serialize_lineage_analysis_request( + request: LineageAnalysisRequest, +) -> dict[str, object]: + """Serialize a request canonically with records ordered by evidence reference.""" + + return { + "contract_version": request.contract_version, + "analysis_id": request.analysis_id, + "analysis_scope_code": request.analysis_scope_code, + "knowledge_cutoff": _time_text(request.knowledge_cutoff), + "policy": { + "candidate_window": request.policy.candidate_window, + "maximum_pair_evaluations": request.policy.maximum_pair_evaluations, + "minimum_fused_score": request.policy.minimum_fused_score, + "allow_llm": request.policy.allow_llm, + }, + "records": [ + _record_dict(record) + for record in sorted( + request.records, + key=lambda item: item.evidence_ref, + ) + ], + } + + +def _score(value: float, *, field: str) -> float: + """Validate and canonically round one result score in ``[0, 1]``.""" + + if isinstance(value, bool) or not isinstance(value, (int, float)): + _raise("invalid_field_type", "score must be numeric", field) + number = float(value) + if not math.isfinite(number) or not 0.0 <= number <= 1.0: + _raise( + "score_out_of_bounds", + "score must be finite and within 0..1", + field, + ) + return round(number, 12) + + +def _validated_reference_partition( + values: tuple[str, ...], + *, + field: str, +) -> frozenset[str]: + """Validate one unique result evidence-reference partition.""" + + if len(set(values)) != len(values): + _raise( + "duplicate_evidence_ref", + "result partition contains duplicate references", + field, + ) + for value in values: + _opaque_reference(value, field=field) + return frozenset(values) + + +def _channel_dict(channel: ChannelEvidence) -> dict[str, object]: + """Serialize one exact active-channel contribution.""" + + return { + "channel_code": _string( + channel.channel_code, + field="channel.channel_code", + maximum=64, + ), + "score": _score(channel.score, field="channel.score"), + "weight": _score(channel.weight, field="channel.weight"), + "contribution": _score( + channel.contribution, + field="channel.contribution", + ), + } + + +def _edge_dict( + edge: LineageEdgeResult, + *, + included_refs: frozenset[str], +) -> dict[str, object]: + """Serialize one edge and verify its evidence math and references.""" + + _opaque_reference( + edge.parent_evidence_ref, + field="edge.parent_evidence_ref", + ) + _opaque_reference( + edge.child_evidence_ref, + field="edge.child_evidence_ref", + ) + if edge.parent_evidence_ref == edge.child_evidence_ref: + _raise("self_lineage_edge", "lineage edge cannot reference itself", "edge") + if ( + edge.parent_evidence_ref not in included_refs + or edge.child_evidence_ref not in included_refs + ): + _raise( + "edge_reference_not_included", + "edge references evidence outside the included partition", + "edge", + ) + _enum( + edge.truth_status_code, + field="edge.truth_status_code", + allowed=_EDGE_TRUTH_STATUSES, + code="unknown_result_truth_status", + ) + fused_score = _score(edge.fused_score, field="edge.fused_score") + channels = tuple(edge.channel_evidence) + if not channels: + _raise( + "missing_channel_evidence", + "edge must disclose at least one channel", + "edge.channel_evidence", + ) + channel_codes = [channel.channel_code for channel in channels] + if len(set(channel_codes)) != len(channel_codes): + _raise( + "duplicate_channel_code", + "edge contains duplicate channel codes", + "edge.channel_evidence", + ) + serialized_channels = [_channel_dict(channel) for channel in channels] + weight_sum = sum(float(item["weight"]) for item in serialized_channels) + if not math.isclose(weight_sum, 1.0, abs_tol=_SCORE_TOLERANCE): + _raise( + "channel_weight_sum_mismatch", + "active channel weights must sum to one", + "edge.channel_evidence", + ) + for item in serialized_channels: + expected = float(item["score"]) * float(item["weight"]) + if not math.isclose( + float(item["contribution"]), + expected, + abs_tol=_SCORE_TOLERANCE, + ): + _raise( + "channel_contribution_mismatch", + "each contribution must equal score multiplied by weight", + str(item["channel_code"]), + ) + contribution_sum = sum( + float(item["contribution"]) + for item in serialized_channels + ) + if not math.isclose( + contribution_sum, + fused_score, + abs_tol=_SCORE_TOLERANCE, + ): + _raise( + "channel_contribution_mismatch", + "channel contributions must reconcile to the fused score", + "edge.channel_evidence", + ) + return { + "parent_evidence_ref": edge.parent_evidence_ref, + "child_evidence_ref": edge.child_evidence_ref, + "relation_type_code": _string( + edge.relation_type_code, + field="edge.relation_type_code", + maximum=64, + ), + "truth_status_code": edge.truth_status_code, + "fused_score": fused_score, + "channel_evidence": sorted( + serialized_channels, + key=lambda item: cast(str, item["channel_code"]), + ), + } + + +def _project_dict( + project: ProjectProjection, + *, + included_refs: frozenset[str], +) -> dict[str, object]: + """Serialize one proposed project grouping and validate its references.""" + + _opaque_reference(project.group_ref, field="project.group_ref") + _opaque_reference(project.project_ref, field="project.project_ref") + if project.truth_status_code != "proposed": + _raise( + "unknown_result_truth_status", + "project projection must remain proposed", + "project.truth_status_code", + ) + evidence_refs = tuple(project.evidence_refs) + if len(set(evidence_refs)) != len(evidence_refs): + _raise( + "duplicate_evidence_ref", + "project projection contains duplicate evidence references", + "project.evidence_refs", + ) + for evidence_ref in evidence_refs: + _opaque_reference(evidence_ref, field="project.evidence_refs") + if evidence_ref not in included_refs: + _raise( + "project_reference_not_included", + "project projection references evidence outside the included partition", + evidence_ref, + ) + return { + "group_ref": project.group_ref, + "project_ref": project.project_ref, + "evidence_refs": sorted(evidence_refs), + "truth_status_code": project.truth_status_code, + } + + +def _limitation_dict(limitation: LineageLimitation) -> dict[str, object]: + """Serialize one bounded machine-readable limitation.""" + + if limitation.evidence_ref is not None: + _opaque_reference( + limitation.evidence_ref, + field="limitation.evidence_ref", + ) + return { + "limitation_code": _string( + limitation.limitation_code, + field="limitation.limitation_code", + maximum=96, + ), + "evidence_ref": limitation.evidence_ref, + "message": _string( + limitation.message, + field="limitation.message", + maximum=500, + ), + } + + +def serialize_lineage_analysis_result( + result: LineageAnalysisResult, + *, + include_digest: bool = True, +) -> dict[str, object]: + """Serialize a result with deterministic ordering and full invariants.""" + + if result.contract_version != CONTRACT_VERSION: + _raise( + "unsupported_contract_version", + "result contract version is unsupported", + "contract_version", + ) + _opaque_reference(result.analysis_id, field="analysis_id") + _enum( + result.analysis_scope_code, + field="analysis_scope_code", + allowed=_ANALYSIS_SCOPES, + code="unknown_analysis_scope", + ) + _enum( + result.llm_status_code, + field="llm_status_code", + allowed=_LLM_STATUSES, + code="unknown_llm_status", + ) + included_refs = _validated_reference_partition( + result.included_evidence_refs, + field="included_evidence_refs", + ) + excluded_refs = _validated_reference_partition( + result.excluded_evidence_refs, + field="excluded_evidence_refs", + ) + if included_refs & excluded_refs: + _raise( + "evidence_partition_overlap", + "included and excluded evidence partitions must be disjoint", + "evidence_refs", + ) + payload: dict[str, object] = { + "contract_version": result.contract_version, + "analysis_id": result.analysis_id, + "analysis_scope_code": result.analysis_scope_code, + "knowledge_cutoff": _time_text(result.knowledge_cutoff), + "included_evidence_refs": sorted(included_refs), + "excluded_evidence_refs": sorted(excluded_refs), + "llm_status_code": result.llm_status_code, + "edges": [ + _edge_dict(edge, included_refs=included_refs) + for edge in sorted( + result.edges, + key=lambda item: ( + item.child_evidence_ref, + item.parent_evidence_ref, + item.relation_type_code, + ), + ) + ], + "project_projections": [ + _project_dict(project, included_refs=included_refs) + for project in sorted( + result.project_projections, + key=lambda item: (item.group_ref, item.project_ref), + ) + ], + "limitations": [ + _limitation_dict(limitation) + for limitation in sorted( + result.limitations, + key=lambda item: ( + item.limitation_code, + item.evidence_ref or "", + item.message, + ), + ) + ], + } + if include_digest: + if not _RESULT_DIGEST.fullmatch(result.result_digest): + _raise( + "invalid_result_digest", + "result digest must be a lowercase SHA-256 identifier", + "result_digest", + ) + expected_digest = _digest(payload) + if result.result_digest != expected_digest: + _raise( + "result_digest_mismatch", + "result digest does not match canonical result content", + "result_digest", + ) + payload["result_digest"] = result.result_digest + return payload + + +def _digest(payload: dict[str, object]) -> str: + """Return a SHA-256 digest over canonical UTF-8 JSON.""" + + canonical = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ) + return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def request_digest(request: LineageAnalysisRequest) -> str: + """Return the deterministic digest of one semantic request.""" + + return _digest(serialize_lineage_analysis_request(request)) + + +def result_digest(result: LineageAnalysisResult) -> str: + """Return the deterministic digest of a result excluding its digest field.""" + + without_digest = replace(result, result_digest="") + return _digest( + serialize_lineage_analysis_result( + without_digest, + include_digest=False, + ) + ) diff --git a/scripts/import_postgresql_posts.py b/scripts/import_postgresql_posts.py index 26de5f856..785d9d1a3 100644 --- a/scripts/import_postgresql_posts.py +++ b/scripts/import_postgresql_posts.py @@ -28,6 +28,7 @@ from backend.app.lineage_ingestion import ChannelWeightsNotEstimated, rebuild_lineage from lineageweave.adjudication_client import ( + AdjudicationClientError, ContextualOrchestratorAdjudicationClient, NullAdjudicationClient, ) @@ -52,6 +53,32 @@ } +async def _lineage_rebuild_summary(target: Any, adjudication_client: Any) -> dict[str, object]: + """Rebuild lineage without turning optional provider output into import loss.""" + + try: + edges = await rebuild_lineage(target, llm=adjudication_client) + return {"lineage_edges": len(edges)} + except ChannelWeightsNotEstimated as exc: + return { + "lineage_edges": None, + "lineage_rebuild_skipped": ( + f"{exc} -- after this import, run " + "scripts/estimate_channel_weights.py and then " + "POST /api/lineage/rebuild" + ), + } + except AdjudicationClientError as exc: + return { + "lineage_edges": None, + "lineage_rebuild_unavailable": ( + f"{exc} -- imported source rows remain persisted; retry " + "POST /api/lineage/rebuild after the orchestrator returns " + "a valid confidence response" + ), + } + + def _normalize_voc_type(value: Any, *, mapped: bool) -> str: """Preserve the governed source VOC vocabulary as canonical target codes.""" if value is None or not str(value).strip(): @@ -648,18 +675,9 @@ async def import_rows(args: argparse.Namespace) -> dict[str, object]: # 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, llm=adjudication_client) - 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" - ), - } + lineage_summary = await _lineage_rebuild_summary( + target, adjudication_client + ) summary: dict[str, object] = { "source_rows": len(rows), "imported_rows": imported, diff --git a/tests/test_adjudication_client.py b/tests/test_adjudication_client.py index 0ea9f6e6c..64490eb54 100644 --- a/tests/test_adjudication_client.py +++ b/tests/test_adjudication_client.py @@ -2,8 +2,68 @@ import pytest -from lineageweave.adjudication_client import ContextualOrchestratorAdjudicationClient -from lineageweave.http_client import HttpClientError +from lineageweave.adjudication_client import ( + AdjudicationClientError, + ContextualOrchestratorAdjudicationClient, + parse_confidence_response, +) + + +@pytest.mark.parametrize("content", ["0", "0.75", "1", "1.000"]) +def test_parse_confidence_response_accepts_only_bounded_numbers(content: str) -> None: + """A compliant number-only response becomes its exact unit score.""" + + assert parse_confidence_response(content) == float(content) + + +@pytest.mark.parametrize("content", ["", "maybe 0.75", "2.0", "0.75 extra", ".5"]) +def test_parse_confidence_response_rejects_malformed_or_out_of_range_text( + content: str, +) -> None: + """Malformed provider output is not silently converted to confidence zero.""" + + with pytest.raises(AdjudicationClientError): + parse_confidence_response(content) + + +def test_parse_confidence_response_rejects_non_text_payload() -> None: + """A structured provider payload cannot masquerade as a score.""" + + with pytest.raises(AdjudicationClientError, match="not text"): + parse_confidence_response({"score": 0.5}) + + +def test_adjudication_client_rejects_malformed_provider_shape(monkeypatch) -> None: + """A provider response without one chat message fails explicitly.""" + + monkeypatch.setattr( + "lineageweave.adjudication_client.post_json", lambda *args, **kwargs: {} + ) + client = ContextualOrchestratorAdjudicationClient( + "https://orchestrator.invalid", + "synthetic-key", + ) + + with pytest.raises(AdjudicationClientError, match="one chat message"): + client.judge("Parent", "Child") + + +def test_adjudication_client_rejects_provider_score_outside_unit_interval( + monkeypatch, +) -> None: + """A raw out-of-range score is rejected instead of being clamped.""" + + monkeypatch.setattr( + "lineageweave.adjudication_client.post_json", + lambda *args, **kwargs: {"choices": [{"message": {"content": "1.2"}}]}, + ) + client = ContextualOrchestratorAdjudicationClient( + "https://orchestrator.invalid", + "synthetic-key", + ) + + with pytest.raises(AdjudicationClientError, match="0..1"): + client.judge("Parent", "Child") def test_adjudication_uses_supported_auto_mode_and_long_local_timeout(monkeypatch) -> None: @@ -43,5 +103,5 @@ def test_adjudication_fails_closed_for_unscoreable_responses(monkeypatch, body) base_url="http://orchestrator:8000", api_key="synthetic-token" ) - with pytest.raises(HttpClientError): + with pytest.raises(AdjudicationClientError): client.judge("workshop", "follow-up bid") diff --git a/tests/test_analysis_run_start.py b/tests/test_analysis_run_start.py index 8f508e954..c6d6881b4 100644 --- a/tests/test_analysis_run_start.py +++ b/tests/test_analysis_run_start.py @@ -22,12 +22,30 @@ topic_lineage_submit_outcome, ) from backend.app.lineage_ingestion import records_from_source_posts +from lineageweave.adjudication_client import AdjudicationClientError from lineageweave.channel_weight_estimation import estimate_fixture_channel_weights from lineageweave.fixtures import sample_records from lineageweave.http_client import HttpClientError from lineageweave.lineage_persistence import lineage_edge_specs from lineageweave.tepp_client import AnalysisRunRequest, TeppClient, TeppNotAvailable + +def test_adjudication_boundary_types_malformed_provider_reply() -> None: + """Malformed orchestrator output becomes a controlled provider failure.""" + + class MalformedAdjudication: + available = True + + def judge(self, candidate_label: str, record_label: str) -> float: + raise AdjudicationClientError("synthetic malformed reply") + + client = analysis_run_start._ProviderBoundaryAdjudication( + MalformedAdjudication() + ) + + with pytest.raises(analysis_run_start._AdjudicationProviderError): + client.judge("synthetic parent", "synthetic child") + @lru_cache(maxsize=1) def _estimated_fixture_weights() -> dict[str, float]: """Return the fast-mlsirm estimate or fail the test closed.""" diff --git a/tests/test_external_lineage_analysis.py b/tests/test_external_lineage_analysis.py new file mode 100644 index 000000000..4c84516b0 --- /dev/null +++ b/tests/test_external_lineage_analysis.py @@ -0,0 +1,817 @@ +"""Execution tests for the external Naruon-facing lineage adapter.""" + +from __future__ import annotations + +from dataclasses import replace +from functools import lru_cache + +import pytest + +from lineageweave.channel_weight_estimation import ( + ChannelWeightEstimate, + estimate_channel_weights, + simulate_fixture_pair_scores, +) +from lineageweave.external_lineage_analysis import ( + _BoundedAdjudicationClient, + _channel_evidence, + analyze_external_lineage, +) +from lineageweave.external_lineage_contract import ( + LineageContractError, + parse_lineage_analysis_request, + request_digest, + result_digest, +) + + +@lru_cache(maxsize=1) +def _estimated_weights() -> ChannelWeightEstimate: + """Fit real fast-mlsirm weights over a deterministic synthetic design.""" + + pair_scores, group_ids = simulate_fixture_pair_scores() + estimate = estimate_channel_weights(pair_scores, group_ids) + assert estimate is not None + return estimate + + +def _analyze(request, *, llm=None): + """Analyze with psychometrically estimated synthetic-fixture weights.""" + + return analyze_external_lineage( + request, + llm=llm, + weight_estimate=_estimated_weights(), + ) + + +class AvailableLlm: + """Deterministic available adjudication channel for contract tests.""" + + available = True + + def judge(self, candidate_label: str, record_label: str) -> float: + """Return a high score for labels sharing their first token.""" + + return ( + 0.9 + if candidate_label.split()[0] == record_label.split()[0] + else 0.1 + ) + + +class InvalidLlm: + """Available client returning an invalid score for fail-closed coverage.""" + + available = True + + def judge(self, candidate_label: str, record_label: str) -> float: + """Return an intentionally invalid value.""" + + return 2.0 + + +class TextLlm: + """Available client returning a non-numeric score.""" + + available = True + + def judge(self, candidate_label: str, record_label: str) -> str: + """Return an intentionally malformed score.""" + + return "unknown" + + +class BrokenProviderLlm: + """Available client surfacing an unexpected raw provider failure.""" + + available = True + + def judge(self, candidate_label: str, record_label: str) -> float: + """Raise a raw provider message that must not cross the contract.""" + + raise RuntimeError("provider secret response body") + + +class CountingLlm: + """Available client recording calls for pre-provider budget tests.""" + + available = True + + def __init__(self) -> None: + """Initialize an empty call counter.""" + + self.call_count = 0 + + def judge(self, candidate_label: str, record_label: str) -> float: + """Count one call and return a bounded score.""" + + self.call_count += 1 + return 0.5 + + +def _record( + evidence_ref: str, + label: str, + occurred_at: str, + *, + available_at: str | None = None, + secondary_key: str | None = "thread:opaque", + project_ref: str | None = "project:opaque", + explicit_parent: dict[str, str] | None = None, + group_ref: str = "workspace:demo", +) -> dict[str, object]: + return { + "evidence_ref": evidence_ref, + "group_ref": group_ref, + "source_kind_code": "email", + "truth_status_code": "observed", + "label": label, + "occurred_at": occurred_at, + "available_at": available_at or occurred_at, + "secondary_key": secondary_key, + "project_ref": project_ref, + "explicit_parent": explicit_parent, + } + + +def _request( + records: list[dict[str, object]], + *, + cutoff: str | None = None, + allow_llm: bool = False, + scope: str = "email_lineage", +): + return parse_lineage_analysis_request( + { + "contract_version": "1.0.0", + "analysis_id": "analysis:integration-001", + "analysis_scope_code": scope, + "knowledge_cutoff": cutoff, + "policy": { + "candidate_window": 50, + "maximum_pair_evaluations": 1000, + "minimum_fused_score": 0.1, + "allow_llm": allow_llm, + }, + "records": records, + } + ) + + +def test_missing_weight_estimate_keeps_observed_truth_and_blocks_inference() -> None: + """No inferred edge is invented when psychometric weights are unavailable.""" + + request = _request( + [ + _record("email:one", "Project update", "2026-08-18T09:00:00Z"), + _record("email:two", "Project follow-up", "2026-08-18T09:01:00Z"), + ] + ) + + result = analyze_external_lineage(request) + + assert result.edges == () + assert [item.limitation_code for item in result.limitations] == [ + "channel_weights_unavailable" + ] + + +def test_missing_weights_do_not_warn_for_unrelated_group_singletons() -> None: + """Independent singleton groups need no unavailable inference weights.""" + + request = _request( + [ + _record( + "email:one", + "First independent record", + "2026-08-18T09:00:00Z", + group_ref="workspace:one", + ), + _record( + "email:two", + "Second independent record", + "2026-08-18T09:01:00Z", + group_ref="workspace:two", + ), + ] + ) + + result = analyze_external_lineage(request) + + assert result.edges == () + assert result.limitations == () + + +def test_cutoff_uses_available_time_and_discloses_excluded_evidence() -> None: + request = _request( + [ + _record( + "email:early", + "Project update", + "2026-08-18T09:00:00Z", + available_at="2026-08-18T09:01:00Z", + ), + _record( + "email:late", + "Earlier event reported late", + "2026-08-17T09:00:00Z", + available_at="2026-08-20T09:00:00Z", + ), + ], + cutoff="2026-08-19T00:00:00Z", + ) + + result = _analyze(request) + + assert result.included_evidence_refs == ("email:early",) + assert result.excluded_evidence_refs == ("email:late",) + assert result.edges == () + assert [ + (item.limitation_code, item.evidence_ref) + for item in result.limitations + ] == [ + ("evidence_after_cutoff_excluded", "email:late"), + ] + + +def test_explicit_rfc_reply_overrides_semantic_parent_and_remains_observed() -> None: + request = _request( + [ + _record( + "email:observed-parent", + "Unrelated root", + "2026-08-20T09:00:00Z", + ), + _record( + "email:semantic-parent", + "Phoenix status", + "2026-08-20T09:01:00Z", + ), + _record( + "email:child", + "Phoenix status follow-up", + "2026-08-20T09:02:00Z", + explicit_parent={ + "evidence_ref": "email:observed-parent", + "relation_code": "rfc_reply", + }, + ), + ] + ) + + result = _analyze(request) + child_edges = [ + edge + for edge in result.edges + if edge.child_evidence_ref == "email:child" + ] + + assert len(child_edges) == 1 + assert child_edges[0].parent_evidence_ref == "email:observed-parent" + assert child_edges[0].relation_type_code == "rfc_reply" + assert child_edges[0].truth_status_code == "observed" + assert child_edges[0].channel_evidence[0].channel_code == "rfc_reply" + + +def test_inferred_edge_exposes_active_channel_weights_and_contributions() -> None: + request = _request( + [ + _record( + "email:001", + "Phoenix delivery status", + "2026-08-20T09:00:00Z", + ), + _record( + "email:002", + "Phoenix delivery status update", + "2026-08-20T09:05:00Z", + ), + ] + ) + + result = _analyze(request) + + assert len(result.edges) == 1 + edge = result.edges[0] + assert edge.truth_status_code == "inferred" + assert edge.relation_type_code == "reconstructed_continuation" + assert {item.channel_code for item in edge.channel_evidence} == { + "temporal", + "secondary_key", + "text", + } + assert sum(item.weight for item in edge.channel_evidence) == pytest.approx( + 1.0 + ) + assert sum( + item.contribution + for item in edge.channel_evidence + ) == pytest.approx(edge.fused_score) + + +@pytest.mark.parametrize( + ("allow_llm", "client", "expected_status", "llm_present"), + [ + (False, AvailableLlm(), "not_requested", False), + (True, None, "unavailable", False), + (True, AvailableLlm(), "unavailable", False), + ], +) +def test_llm_policy_is_explicit_and_never_fabricates_absent_scores( + allow_llm: bool, + client, + expected_status: str, + llm_present: bool, +) -> None: + request = _request( + [ + _record( + "email:001", + "Phoenix delivery status", + "2026-08-20T09:00:00Z", + ), + _record( + "email:002", + "Phoenix delivery status update", + "2026-08-20T09:05:00Z", + ), + ], + allow_llm=allow_llm, + ) + + result = _analyze(request, llm=client) + + assert result.llm_status_code == expected_status + channels = { + channel.channel_code + for channel in result.edges[0].channel_evidence + } + assert ("llm" in channels) is llm_present + + +def test_project_projection_is_proposed_and_uses_only_included_evidence() -> None: + request = _request( + [ + _record( + "email:001", + "One", + "2026-08-20T09:00:00Z", + ), + _record( + "email:002", + "Two", + "2026-08-20T09:01:00Z", + ), + _record( + "email:003", + "Late", + "2026-08-18T09:00:00Z", + available_at="2026-08-22T09:00:00Z", + ), + ], + cutoff="2026-08-21T00:00:00Z", + scope="project_history", + ) + + result = _analyze(request) + + assert result.project_projections[0].project_ref == "project:opaque" + assert result.project_projections[0].evidence_refs == ( + "email:001", + "email:002", + ) + assert result.project_projections[0].truth_status_code == "proposed" + + +def test_analysis_is_deterministic_for_reordered_input_and_has_digest() -> None: + records = [ + _record( + "email:001", + "Phoenix delivery status", + "2026-08-20T09:00:00Z", + ), + _record( + "email:002", + "Phoenix delivery status update", + "2026-08-20T09:05:00Z", + ), + ] + first_request = _request(records) + second_request = _request(list(reversed(records))) + + first = _analyze(first_request) + second = _analyze(second_request) + + assert request_digest(first_request) == request_digest(second_request) + assert first == second + assert first.result_digest.startswith("sha256:") + assert result_digest(first) == first.result_digest + + +@pytest.mark.parametrize( + ("records", "expected_code"), + [ + ( + [ + _record( + "email:child", + "Child", + "2026-08-20T09:00:00Z", + explicit_parent={ + "evidence_ref": "email:missing", + "relation_code": "rfc_reply", + }, + ) + ], + "explicit_parent_missing", + ), + ( + [ + _record( + "email:child", + "Child", + "2026-08-20T09:00:00Z", + explicit_parent={ + "evidence_ref": "email:child", + "relation_code": "rfc_reply", + }, + ) + ], + "explicit_parent_self_reference", + ), + ( + [ + _record( + "email:parent", + "Parent", + "2026-08-20T10:00:00Z", + ), + _record( + "email:child", + "Child", + "2026-08-20T09:00:00Z", + explicit_parent={ + "evidence_ref": "email:parent", + "relation_code": "rfc_reply", + }, + ), + ], + "explicit_parent_after_child", + ), + ( + [ + _record( + "email:parent", + "Parent", + "2026-08-20T09:00:00Z", + group_ref="workspace:one", + ), + _record( + "email:child", + "Child", + "2026-08-20T10:00:00Z", + group_ref="workspace:two", + explicit_parent={ + "evidence_ref": "email:parent", + "relation_code": "rfc_reply", + }, + ), + ], + "explicit_parent_group_mismatch", + ), + ], +) +def test_invalid_explicit_parent_semantics_fail_closed( + records: list[dict[str, object]], + expected_code: str, +) -> None: + request = _request(records) + + with pytest.raises(LineageContractError) as captured: + _analyze(request) + + assert captured.value.code == expected_code + + +def test_explicit_parent_cycle_fails_closed_even_when_timestamps_tie() -> None: + request = _request( + [ + _record( + "email:one", + "One", + "2026-08-20T09:00:00Z", + explicit_parent={ + "evidence_ref": "email:two", + "relation_code": "rfc_reply", + }, + ), + _record( + "email:two", + "Two", + "2026-08-20T09:00:00Z", + explicit_parent={ + "evidence_ref": "email:one", + "relation_code": "rfc_reply", + }, + ), + ] + ) + + with pytest.raises(LineageContractError) as captured: + _analyze(request) + + assert captured.value.code == "explicit_parent_cycle" + + +def test_inference_cannot_reverse_an_observed_edge_on_tied_timestamps() -> None: + """A tied-time observed edge excludes its child as an inferred parent.""" + + request = _request( + [ + _record( + "email:z-parent", + "Shared update", + "2026-08-20T09:00:00Z", + ), + _record( + "email:a-child", + "Shared update follow-up", + "2026-08-20T09:00:00Z", + explicit_parent={ + "evidence_ref": "email:z-parent", + "relation_code": "rfc_reply", + }, + ), + ] + ) + + result = _analyze(request) + + assert [ + ( + edge.parent_evidence_ref, + edge.child_evidence_ref, + edge.truth_status_code, + ) + for edge in result.edges + ] == [("email:z-parent", "email:a-child", "observed")] + + +def test_pair_budget_excludes_observed_descendants_never_sent_to_provider() -> None: + """The declared budget counts the exact cycle-safe provider work.""" + + request = _request( + [ + _record( + "email:a-child", + "Observed child", + "2026-08-20T09:00:00Z", + explicit_parent={ + "evidence_ref": "email:z-parent", + "relation_code": "rfc_reply", + }, + ), + _record("email:m-other", "Other", "2026-08-20T09:00:00Z"), + _record("email:z-parent", "Parent", "2026-08-20T09:00:00Z"), + ] + ) + request = replace( + request, + policy=replace(request.policy, maximum_pair_evaluations=2), + ) + + _analyze(request) + + +def test_cutoff_excluded_explicit_parent_creates_limitation_not_edge() -> None: + request = _request( + [ + _record( + "email:parent", + "Parent", + "2026-08-18T09:00:00Z", + available_at="2026-08-22T09:00:00Z", + ), + _record( + "email:child", + "Child", + "2026-08-20T09:00:00Z", + available_at="2026-08-20T09:01:00Z", + explicit_parent={ + "evidence_ref": "email:parent", + "relation_code": "rfc_reply", + }, + ), + ], + cutoff="2026-08-21T00:00:00Z", + ) + + result = _analyze(request) + + assert all( + edge.relation_type_code != "rfc_reply" + for edge in result.edges + ) + assert any( + item.limitation_code == "explicit_parent_after_cutoff" + and item.evidence_ref == "email:child" + for item in result.limitations + ) + + +def test_all_evidence_after_cutoff_returns_empty_bounded_result() -> None: + request = _request( + [ + _record( + "email:late", + "Late", + "2026-08-18T09:00:00Z", + available_at="2026-08-22T09:00:00Z", + project_ref=None, + ) + ], + cutoff="2026-08-21T00:00:00Z", + ) + + result = _analyze(request) + + assert result.included_evidence_refs == () + assert result.edges == () + assert result.project_projections == () + + +def test_invalid_llm_score_fails_closed_at_provider_boundary() -> None: + with pytest.raises(LineageContractError) as captured: + _BoundedAdjudicationClient(InvalidLlm()).judge("Phoenix one", "Phoenix two") + + assert captured.value.code == "channel_score_out_of_bounds" + + +def test_non_numeric_llm_score_fails_closed_at_provider_boundary() -> None: + """A provider score with the wrong type becomes a stable contract error.""" + + with pytest.raises(LineageContractError) as captured: + _BoundedAdjudicationClient(TextLlm()).judge("Phoenix one", "Phoenix two") + + assert captured.value.code == "channel_score_out_of_bounds" + + +def test_raw_provider_response_error_is_stable_at_provider_boundary() -> None: + """A raw provider failure is not exposed as an arbitrary exception.""" + + with pytest.raises(LineageContractError) as captured: + _BoundedAdjudicationClient(BrokenProviderLlm()).judge("Phoenix one", "Phoenix two") + + assert captured.value.code == "llm_channel_error" + assert "provider secret" not in str(captured.value) + + +def test_channel_evidence_rejects_invalid_score_before_serialization() -> None: + """Defense in depth keeps direct channel projection fail-closed.""" + + with pytest.raises(LineageContractError) as captured: + _channel_evidence({"text": 2.0}, {"text": 1.0}) + + assert captured.value.code == "channel_score_out_of_bounds" + + +def test_records_without_project_reference_are_not_projected() -> None: + request = _request( + [ + _record( + "email:001", + "No project", + "2026-08-20T09:00:00Z", + project_ref=None, + ) + ] + ) + + result = _analyze(request) + + assert result.project_projections == () + + +def test_cutoff_excluded_explicit_parent_suppresses_alternative_inference() -> None: + request = _request( + [ + _record( + "email:alternative", + "Phoenix child", + "2026-08-20T08:00:00Z", + available_at="2026-08-20T08:01:00Z", + ), + _record( + "email:observed-parent", + "Observed parent", + "2026-08-18T09:00:00Z", + available_at="2026-08-22T09:00:00Z", + ), + _record( + "email:child", + "Phoenix child", + "2026-08-20T09:00:00Z", + available_at="2026-08-20T09:01:00Z", + explicit_parent={ + "evidence_ref": "email:observed-parent", + "relation_code": "rfc_reply", + }, + ), + ], + cutoff="2026-08-21T00:00:00Z", + ) + + result = _analyze(request) + + assert all( + edge.child_evidence_ref != "email:child" + for edge in result.edges + ) + + +def test_project_projections_do_not_merge_across_groups() -> None: + request = _request( + [ + _record( + "email:one", + "One", + "2026-08-20T09:00:00Z", + group_ref="workspace:one", + ), + _record( + "email:two", + "Two", + "2026-08-20T09:00:00Z", + group_ref="workspace:two", + ), + ], + scope="project_history", + ) + + result = _analyze(request) + projections = [ + (item.group_ref, item.project_ref, item.evidence_refs) + for item in result.project_projections + ] + + assert projections == [ + ("workspace:one", "project:opaque", ("email:one",)), + ("workspace:two", "project:opaque", ("email:two",)), + ] + + +def test_pair_budget_rejects_before_any_optional_llm_call() -> None: + records = [ + _record( + f"email:{index}", + f"Message {index}", + f"2026-08-20T09:0{index}:00Z", + ) + for index in range(4) + ] + payload = { + "contract_version": "1.0.0", + "analysis_id": "analysis:pair-budget", + "analysis_scope_code": "email_lineage", + "knowledge_cutoff": None, + "policy": { + "candidate_window": 50, + "maximum_pair_evaluations": 2, + "minimum_fused_score": 0.1, + "allow_llm": True, + }, + "records": records, + } + request = parse_lineage_analysis_request(payload) + client = CountingLlm() + + with pytest.raises(LineageContractError) as captured: + _analyze(request, llm=client) + + assert captured.value.code == "pair_evaluation_budget_exceeded" + assert client.call_count == 0 + + +def test_missing_cutoff_includes_all_records() -> None: + request = _request( + [ + _record( + "email:one", + "One", + "2026-08-20T09:00:00Z", + ), + _record( + "email:two", + "Two", + "2026-08-21T09:00:00Z", + available_at="2026-09-01T09:00:00Z", + ), + ], + cutoff=None, + ) + + result = _analyze(request) + + assert result.included_evidence_refs == ("email:one", "email:two") + assert result.excluded_evidence_refs == () diff --git a/tests/test_external_lineage_contract.py b/tests/test_external_lineage_contract.py new file mode 100644 index 000000000..2fc7a4f21 --- /dev/null +++ b/tests/test_external_lineage_contract.py @@ -0,0 +1,715 @@ +"""Contract tests for the future Naruon-facing LineageWeave boundary.""" + +from __future__ import annotations + +import json +from dataclasses import replace +from datetime import UTC, datetime +from pathlib import Path + +import pytest + +from lineageweave.external_lineage_contract import ( + CONTRACT_VERSION, + ChannelEvidence, + ExplicitParent, + LineageAnalysisResult, + LineageContractError, + LineageEdgeResult, + LineageLimitation, + ProjectProjection, + parse_lineage_analysis_request, + request_digest, + result_digest, + serialize_lineage_analysis_request, + serialize_lineage_analysis_result, +) + +_ROOT = Path(__file__).resolve().parents[1] + + +def _record( + evidence_ref: str, + *, + occurred_at: str = "2026-08-20T09:00:00Z", + available_at: str = "2026-08-20T09:01:00Z", + explicit_parent: dict[str, str] | None = None, +) -> dict[str, object]: + return { + "evidence_ref": evidence_ref, + "group_ref": "workspace:demo", + "source_kind_code": "email", + "truth_status_code": "observed", + "label": f"Subject {evidence_ref}", + "occurred_at": occurred_at, + "available_at": available_at, + "secondary_key": "provider-thread:opaque", + "project_ref": "project:opaque", + "explicit_parent": explicit_parent, + } + + +def _payload() -> dict[str, object]: + return { + "contract_version": "1.0.0", + "analysis_id": "analysis:demo-001", + "analysis_scope_code": "email_lineage", + "knowledge_cutoff": "2026-08-20T18:00:00+09:00", + "policy": { + "candidate_window": 50, + "maximum_pair_evaluations": 1000, + "minimum_fused_score": 0.3, + "allow_llm": False, + }, + "records": [ + _record("email:001"), + _record( + "email:002", + occurred_at="2026-08-20T09:05:00Z", + available_at="2026-08-20T09:06:00Z", + explicit_parent={ + "evidence_ref": "email:001", + "relation_code": "rfc_reply", + }, + ), + ], + } + + +def _result_fixture() -> LineageAnalysisResult: + return LineageAnalysisResult( + contract_version=CONTRACT_VERSION, + analysis_id="analysis:fixture", + analysis_scope_code="generic_lineage", + knowledge_cutoff=None, + included_evidence_refs=("record:001",), + excluded_evidence_refs=(), + llm_status_code="not_requested", + edges=(), + project_projections=(), + limitations=(), + result_digest="", + ) + + +def test_parse_request_is_strict_immutable_and_canonicalizes_timestamps() -> None: + request = parse_lineage_analysis_request(_payload()) + + assert request.contract_version == CONTRACT_VERSION + assert request.analysis_id == "analysis:demo-001" + assert request.analysis_scope_code == "email_lineage" + assert request.knowledge_cutoff == datetime( + 2026, + 8, + 20, + 9, + 0, + tzinfo=UTC, + ) + assert request.records[1].explicit_parent == ExplicitParent( + evidence_ref="email:001", + relation_code="rfc_reply", + ) + assert serialize_lineage_analysis_request(request)[ + "knowledge_cutoff" + ] == "2026-08-20T09:00:00Z" + with pytest.raises(AttributeError): + request.analysis_id = "changed" # type: ignore[misc] + + +def test_request_digest_is_stable_when_keys_and_records_are_reordered() -> None: + payload = _payload() + reordered = { + "records": list(reversed(payload["records"])), # type: ignore[arg-type] + "policy": { + "allow_llm": False, + "minimum_fused_score": 0.3, + "maximum_pair_evaluations": 1000, + "candidate_window": 50, + }, + "knowledge_cutoff": payload["knowledge_cutoff"], + "analysis_scope_code": payload["analysis_scope_code"], + "analysis_id": payload["analysis_id"], + "contract_version": payload["contract_version"], + } + + assert request_digest( + parse_lineage_analysis_request(payload) + ) == request_digest(parse_lineage_analysis_request(reordered)) + + +@pytest.mark.parametrize( + ("mutator", "expected_code"), + [ + (lambda payload: payload.update({"unexpected": True}), "unknown_field"), + ( + lambda payload: payload["policy"].update( # type: ignore[union-attr] + {"unexpected": True} + ), + "unknown_field", + ), + ( + lambda payload: payload["records"][0].update( # type: ignore[index,union-attr] + {"unexpected": True} + ), + "unknown_field", + ), + ( + lambda payload: payload.update({"contract_version": "2.0.0"}), + "unsupported_contract_version", + ), + ( + lambda payload: payload.update( + {"analysis_scope_code": "mailbox_dump"} + ), + "unknown_analysis_scope", + ), + ], +) +def test_parser_rejects_unknown_fields_and_vocabularies( + mutator, + expected_code: str, +) -> None: + payload = _payload() + mutator(payload) + + with pytest.raises(LineageContractError) as captured: + parse_lineage_analysis_request(payload) + + assert captured.value.code == expected_code + + +def test_parser_rejects_duplicate_references_and_record_count_bounds() -> None: + payload = _payload() + payload["records"] = [_record("email:001"), _record("email:001")] + with pytest.raises(LineageContractError) as duplicate: + parse_lineage_analysis_request(payload) + assert duplicate.value.code == "duplicate_evidence_ref" + + payload["records"] = [] + with pytest.raises(LineageContractError) as empty: + parse_lineage_analysis_request(payload) + assert empty.value.code == "record_count_out_of_bounds" + + payload["records"] = [ + _record(f"email:{index:03d}") + for index in range(501) + ] + with pytest.raises(LineageContractError) as oversized: + parse_lineage_analysis_request(payload) + assert oversized.value.code == "record_count_out_of_bounds" + + +@pytest.mark.parametrize( + ("field_name", "value", "expected_code"), + [ + ( + "occurred_at", + "2026-08-20T09:00:00", + "timestamp_must_be_offset_aware", + ), + ("available_at", "not-a-time", "invalid_timestamp"), + ( + "evidence_ref", + "https://mail.example/message/1", + "unsafe_opaque_reference", + ), + ("evidence_ref", "contains whitespace", "unsafe_opaque_reference"), + ("label", "", "text_length_out_of_bounds"), + ("label", "x" * 2001, "text_length_out_of_bounds"), + ], +) +def test_parser_rejects_unsafe_identifiers_timestamps_and_text( + field_name: str, + value: str, + expected_code: str, +) -> None: + payload = _payload() + payload["records"][0][field_name] = value # type: ignore[index] + + with pytest.raises(LineageContractError) as captured: + parse_lineage_analysis_request(payload) + + assert captured.value.code == expected_code + + +@pytest.mark.parametrize( + ("field_name", "value", "expected_code"), + [ + ("candidate_window", 0, "policy_value_out_of_bounds"), + ("candidate_window", 201, "policy_value_out_of_bounds"), + ("maximum_pair_evaluations", 0, "policy_value_out_of_bounds"), + ("maximum_pair_evaluations", 5_001, "policy_value_out_of_bounds"), + ("minimum_fused_score", -0.1, "policy_value_out_of_bounds"), + ("minimum_fused_score", 1.1, "policy_value_out_of_bounds"), + ("allow_llm", "yes", "invalid_field_type"), + ], +) +def test_parser_rejects_invalid_policy_values( + field_name: str, + value: object, + expected_code: str, +) -> None: + payload = _payload() + payload["policy"][field_name] = value # type: ignore[index] + + with pytest.raises(LineageContractError) as captured: + parse_lineage_analysis_request(payload) + + assert captured.value.code == expected_code + + +def test_result_serialization_is_deterministic_and_digest_is_external() -> None: + edge = LineageEdgeResult( + parent_evidence_ref="email:001", + child_evidence_ref="email:002", + relation_type_code="reconstructed_continuation", + truth_status_code="inferred", + fused_score=0.75, + channel_evidence=( + ChannelEvidence("text", 0.8, 0.5, 0.4), + ChannelEvidence("temporal", 0.7, 0.5, 0.35), + ), + ) + result = LineageAnalysisResult( + contract_version=CONTRACT_VERSION, + analysis_id="analysis:demo-001", + analysis_scope_code="email_lineage", + knowledge_cutoff=datetime( + 2026, + 8, + 20, + 9, + 0, + tzinfo=UTC, + ), + included_evidence_refs=("email:001", "email:002"), + excluded_evidence_refs=(), + llm_status_code="not_requested", + edges=(edge,), + project_projections=( + ProjectProjection( + "workspace:demo", + "project:opaque", + ("email:001", "email:002"), + "proposed", + ), + ), + limitations=( + LineageLimitation("none", None, "No material limitation."), + ), + result_digest="", + ) + digest = result_digest(result) + finalized = replace(result, result_digest=digest) + + serialized = serialize_lineage_analysis_result(finalized) + assert serialized["result_digest"] == digest + assert serialized["knowledge_cutoff"] == "2026-08-20T09:00:00Z" + assert result_digest(finalized) == digest + assert json.dumps(serialized, sort_keys=True, separators=(",", ":")) + + +def test_public_schema_exists_and_mirrors_contract_vocabularies() -> None: + schema = json.loads( + ( + _ROOT + / "docs" + / "contracts" + / "external-lineage-analysis-v1.schema.json" + ).read_text(encoding="utf-8") + ) + + assert schema["$schema"] == ( + "https://json-schema.org/draft/2020-12/schema" + ) + assert schema["properties"]["contract_version"]["const"] == ( + CONTRACT_VERSION + ) + assert set( + schema["properties"]["analysis_scope_code"]["enum"] + ) == { + "email_lineage", + "project_history", + "generic_lineage", + } + assert schema["additionalProperties"] is False + pair_budget = schema["$defs"]["LineageAnalysisPolicy"][ + "properties" + ]["maximum_pair_evaluations"] + assert pair_budget == { + "type": "integer", + "minimum": 1, + "maximum": 5000, + } + + +def test_parser_rejects_non_object_and_missing_required_field() -> None: + with pytest.raises(LineageContractError) as non_object: + parse_lineage_analysis_request([]) + assert non_object.value.code == "invalid_field_type" + + payload = _payload() + del payload["analysis_id"] + with pytest.raises(LineageContractError) as missing: + parse_lineage_analysis_request(payload) + assert missing.value.code == "missing_field" + + +def test_parser_rejects_wrong_scalar_types_and_non_array_records() -> None: + mutations = [ + ("contract_version", 1, "invalid_field_type"), + ("knowledge_cutoff", 1, "invalid_field_type"), + ("analysis_scope_code", 1, "invalid_field_type"), + ] + for field, value, expected in mutations: + payload = _payload() + payload[field] = value + with pytest.raises(LineageContractError) as captured: + parse_lineage_analysis_request(payload) + assert captured.value.code == expected + + payload = _payload() + payload["policy"]["minimum_fused_score"] = "0.3" # type: ignore[index] + with pytest.raises(LineageContractError) as number: + parse_lineage_analysis_request(payload) + assert number.value.code == "invalid_field_type" + + payload = _payload() + payload["policy"]["candidate_window"] = 50.0 # type: ignore[index] + with pytest.raises(LineageContractError) as integer: + parse_lineage_analysis_request(payload) + assert integer.value.code == "invalid_field_type" + + payload = _payload() + payload["records"] = tuple(payload["records"]) # type: ignore[arg-type] + with pytest.raises(LineageContractError) as records: + parse_lineage_analysis_request(payload) + assert records.value.code == "invalid_field_type" + + +def test_optional_references_may_be_omitted() -> None: + payload = _payload() + record = payload["records"][0] # type: ignore[index] + del record["secondary_key"] + del record["project_ref"] + del record["explicit_parent"] + + parsed = parse_lineage_analysis_request(payload) + + assert parsed.records[0].secondary_key is None + assert parsed.records[0].project_ref is None + assert parsed.records[0].explicit_parent is None + + +def test_result_serializer_rejects_naive_timestamp_and_invalid_scores() -> None: + result = replace( + _result_fixture(), + knowledge_cutoff=datetime(2026, 8, 20, 9, 0), # noqa: DTZ001 - rejection fixture + ) + with pytest.raises(LineageContractError) as naive: + serialize_lineage_analysis_result(result) + assert naive.value.code == "timestamp_must_be_offset_aware" + + invalid_type_edge = LineageEdgeResult( + "record:001", + "record:002", + "reconstructed_continuation", + "inferred", + True, # type: ignore[arg-type] + (ChannelEvidence("text", 0.5, 1.0, 0.5),), + ) + result_with_two_records = replace( + _result_fixture(), + included_evidence_refs=("record:001", "record:002"), + ) + with pytest.raises(LineageContractError) as score_type: + serialize_lineage_analysis_result( + replace( + result_with_two_records, + edges=(invalid_type_edge,), + ) + ) + assert score_type.value.code == "invalid_field_type" + + invalid_range_edge = replace(invalid_type_edge, fused_score=1.1) + with pytest.raises(LineageContractError) as score_range: + serialize_lineage_analysis_result( + replace( + result_with_two_records, + edges=(invalid_range_edge,), + ) + ) + assert score_range.value.code == "score_out_of_bounds" + + +def test_result_serializer_rejects_non_proposed_project_and_wrong_version() -> None: + project = ProjectProjection( + "workspace:one", + "project:one", + ("record:001",), + "observed", + ) # type: ignore[arg-type] + with pytest.raises(LineageContractError) as truth: + serialize_lineage_analysis_result( + replace( + _result_fixture(), + project_projections=(project,), + ) + ) + assert truth.value.code == "unknown_result_truth_status" + + with pytest.raises(LineageContractError) as version: + serialize_lineage_analysis_result( + replace(_result_fixture(), contract_version="2.0.0") + ) + assert version.value.code == "unsupported_contract_version" + + +def test_result_requires_a_valid_digest_for_transport() -> None: + with pytest.raises(LineageContractError) as captured: + serialize_lineage_analysis_result(_result_fixture()) + + assert captured.value.code == "invalid_result_digest" + + +def test_result_rejects_overlapping_or_duplicate_partitions() -> None: + overlap = replace( + _result_fixture(), + included_evidence_refs=("record:001",), + excluded_evidence_refs=("record:001",), + result_digest="sha256:" + "0" * 64, + ) + with pytest.raises(LineageContractError) as captured: + serialize_lineage_analysis_result(overlap) + assert captured.value.code == "evidence_partition_overlap" + + duplicate = replace( + _result_fixture(), + included_evidence_refs=("record:001", "record:001"), + result_digest="sha256:" + "0" * 64, + ) + with pytest.raises(LineageContractError) as duplicate_error: + serialize_lineage_analysis_result(duplicate) + assert duplicate_error.value.code == "duplicate_evidence_ref" + + +def test_result_rejects_unincluded_edge_or_project_references() -> None: + edge = LineageEdgeResult( + "record:001", + "record:missing", + "reconstructed_continuation", + "inferred", + 0.5, + (ChannelEvidence("text", 0.5, 1.0, 0.5),), + ) + result = replace( + _result_fixture(), + edges=(edge,), + result_digest="sha256:" + "0" * 64, + ) + with pytest.raises(LineageContractError) as edge_error: + serialize_lineage_analysis_result(result) + assert edge_error.value.code == "edge_reference_not_included" + + project = ProjectProjection( + "workspace:one", + "project:one", + ("record:missing",), + "proposed", + ) + result = replace( + _result_fixture(), + project_projections=(project,), + result_digest="sha256:" + "0" * 64, + ) + with pytest.raises(LineageContractError) as project_error: + serialize_lineage_analysis_result(result) + assert project_error.value.code == "project_reference_not_included" + + +def test_result_rejects_self_edges_and_channel_math_errors() -> None: + base = replace( + _result_fixture(), + included_evidence_refs=("record:001", "record:002"), + ) + self_edge = LineageEdgeResult( + "record:001", + "record:001", + "reconstructed_continuation", + "inferred", + 0.5, + (ChannelEvidence("text", 0.5, 1.0, 0.5),), + ) + with pytest.raises(LineageContractError) as self_error: + serialize_lineage_analysis_result( + replace( + base, + edges=(self_edge,), + result_digest="sha256:" + "0" * 64, + ) + ) + assert self_error.value.code == "self_lineage_edge" + + duplicate_channels = replace( + self_edge, + parent_evidence_ref="record:002", + channel_evidence=( + ChannelEvidence("text", 0.5, 0.5, 0.25), + ChannelEvidence("text", 0.5, 0.5, 0.25), + ), + ) + with pytest.raises(LineageContractError) as duplicate_error: + serialize_lineage_analysis_result( + replace( + base, + edges=(duplicate_channels,), + result_digest="sha256:" + "0" * 64, + ) + ) + assert duplicate_error.value.code == "duplicate_channel_code" + + bad_weights = replace( + duplicate_channels, + channel_evidence=( + ChannelEvidence("text", 0.5, 0.4, 0.2), + ChannelEvidence("temporal", 0.5, 0.4, 0.2), + ), + ) + with pytest.raises(LineageContractError) as weight_error: + serialize_lineage_analysis_result( + replace( + base, + edges=(bad_weights,), + result_digest="sha256:" + "0" * 64, + ) + ) + assert weight_error.value.code == "channel_weight_sum_mismatch" + + bad_contribution = replace( + bad_weights, + channel_evidence=( + ChannelEvidence("text", 0.5, 0.5, 0.2), + ChannelEvidence("temporal", 0.5, 0.5, 0.2), + ), + ) + with pytest.raises(LineageContractError) as contribution_error: + serialize_lineage_analysis_result( + replace( + base, + edges=(bad_contribution,), + result_digest="sha256:" + "0" * 64, + ) + ) + assert contribution_error.value.code == ( + "channel_contribution_mismatch" + ) + + +def test_result_rejects_unsafe_analysis_identifier() -> None: + result = replace( + _result_fixture(), + analysis_id="https://unsafe.example/run", + result_digest="sha256:" + "0" * 64, + ) + with pytest.raises(LineageContractError) as captured: + serialize_lineage_analysis_result(result) + assert captured.value.code == "unsafe_opaque_reference" + + +def test_result_rejects_missing_channels_and_contribution_mismatch() -> None: + base = replace( + _result_fixture(), + included_evidence_refs=("record:001", "record:002"), + ) + missing_channels = LineageEdgeResult( + "record:001", + "record:002", + "reconstructed_continuation", + "inferred", + 0.5, + (), + ) + with pytest.raises(LineageContractError) as missing: + serialize_lineage_analysis_result( + replace( + base, + edges=(missing_channels,), + result_digest="sha256:" + "0" * 64, + ) + ) + assert missing.value.code == "missing_channel_evidence" + + inconsistent = replace( + missing_channels, + channel_evidence=( + ChannelEvidence("text", 0.5, 0.5, 0.3), + ChannelEvidence("temporal", 0.5, 0.5, 0.2), + ), + ) + with pytest.raises(LineageContractError) as mismatch: + serialize_lineage_analysis_result( + replace( + base, + edges=(inconsistent,), + result_digest="sha256:" + "0" * 64, + ) + ) + assert mismatch.value.code == "channel_contribution_mismatch" + + +def test_result_rejects_channel_sum_that_does_not_equal_fused_score() -> None: + """The fused score must reconcile with all otherwise valid contributions.""" + + edge = LineageEdgeResult( + "record:001", + "record:002", + "reconstructed_continuation", + "inferred", + 0.5, + ( + ChannelEvidence("text", 0.2, 0.5, 0.1), + ChannelEvidence("temporal", 0.2, 0.5, 0.1), + ), + ) + with pytest.raises(LineageContractError) as captured: + serialize_lineage_analysis_result( + replace( + _result_fixture(), + included_evidence_refs=("record:001", "record:002"), + edges=(edge,), + result_digest="sha256:" + "0" * 64, + ) + ) + + assert captured.value.code == "channel_contribution_mismatch" + + +def test_result_rejects_duplicate_project_evidence_references() -> None: + project = ProjectProjection( + "workspace:one", + "project:one", + ("record:001", "record:001"), + "proposed", + ) + with pytest.raises(LineageContractError) as captured: + serialize_lineage_analysis_result( + replace( + _result_fixture(), + project_projections=(project,), + result_digest="sha256:" + "0" * 64, + ) + ) + assert captured.value.code == "duplicate_evidence_ref" + + +def test_result_rejects_digest_not_matching_canonical_content() -> None: + result = replace( + _result_fixture(), + result_digest="sha256:" + "0" * 64, + ) + + with pytest.raises(LineageContractError) as captured: + serialize_lineage_analysis_result(result) + + assert captured.value.code == "result_digest_mismatch" diff --git a/tests/test_external_lineage_explicit_parent_budget.py b/tests/test_external_lineage_explicit_parent_budget.py new file mode 100644 index 000000000..27105bf18 --- /dev/null +++ b/tests/test_external_lineage_explicit_parent_budget.py @@ -0,0 +1,166 @@ +"""Regression tests for explicit-parent budget and provider minimization.""" + +from __future__ import annotations + +from lineageweave.channel_weight_estimation import estimate_fixture_channel_weights +from lineageweave.external_lineage_analysis import analyze_external_lineage +from lineageweave.external_lineage_contract import parse_lineage_analysis_request + + +def _analyze(request, *, llm=None): + """Analyze with the real synthetic-fixture fast-mlsirm estimate.""" + + estimate = estimate_fixture_channel_weights() + assert estimate is not None + return analyze_external_lineage(request, llm=llm, weight_estimate=estimate) + + +class CountingLlm: + """Available adjudication client that records every disclosed label pair.""" + + available = True + + def __init__(self) -> None: + """Initialize an empty provider-call ledger.""" + + self.calls: list[tuple[str, str]] = [] + + def judge(self, candidate_label: str, record_label: str) -> float: + """Record one adjudication pair and return a bounded score.""" + + self.calls.append((candidate_label, record_label)) + return 0.5 + + +def _record( + evidence_ref: str, + label: str, + occurred_at: str, + *, + explicit_parent: str | None = None, +) -> dict[str, object]: + """Build one synthetic authorized email evidence record.""" + + return { + "evidence_ref": evidence_ref, + "group_ref": "workspace:synthetic", + "source_kind_code": "email", + "truth_status_code": "observed", + "label": label, + "occurred_at": occurred_at, + "available_at": occurred_at, + "secondary_key": "thread:synthetic", + "project_ref": "project:synthetic", + "explicit_parent": ( + { + "evidence_ref": explicit_parent, + "relation_code": "rfc_reply", + } + if explicit_parent is not None + else None + ), + } + + +def _request( + records: list[dict[str, object]], + *, + allow_llm: bool, + maximum_pair_evaluations: int, +): + """Parse one strict external-lineage request for the regression cases.""" + + return parse_lineage_analysis_request( + { + "contract_version": "1.0.0", + "analysis_id": "analysis:explicit-parent-budget", + "analysis_scope_code": "email_lineage", + "knowledge_cutoff": None, + "policy": { + "candidate_window": 50, + "maximum_pair_evaluations": maximum_pair_evaluations, + "minimum_fused_score": 0.1, + "allow_llm": allow_llm, + }, + "records": records, + } + ) + + +def test_explicit_parent_chain_spends_no_inference_budget_or_llm_calls() -> None: + """Caller-observed edges must not be rescored or charged as inferred work.""" + + request = _request( + [ + _record("email:one", "One", "2026-08-21T09:00:00Z"), + _record( + "email:two", + "Two", + "2026-08-21T09:01:00Z", + explicit_parent="email:one", + ), + _record( + "email:three", + "Three", + "2026-08-21T09:02:00Z", + explicit_parent="email:two", + ), + _record( + "email:four", + "Four", + "2026-08-21T09:03:00Z", + explicit_parent="email:three", + ), + ], + allow_llm=True, + maximum_pair_evaluations=1, + ) + client = CountingLlm() + + result = _analyze(request, llm=client) + + assert client.calls == [] + assert [ + ( + edge.parent_evidence_ref, + edge.child_evidence_ref, + edge.truth_status_code, + ) + for edge in result.edges + ] == [ + ("email:one", "email:two", "observed"), + ("email:two", "email:three", "observed"), + ("email:three", "email:four", "observed"), + ] + + +def test_explicit_child_remains_available_as_a_later_inference_candidate() -> None: + """Skipping its own scoring must not remove an explicit child from history.""" + + request = _request( + [ + _record("email:root", "Root", "2026-08-21T09:00:00Z"), + _record( + "email:observed-child", + "Phoenix delivery update", + "2026-08-21T09:01:00Z", + explicit_parent="email:root", + ), + _record( + "email:later-child", + "Phoenix delivery update", + "2026-08-21T09:02:00Z", + ), + ], + allow_llm=False, + maximum_pair_evaluations=2, + ) + + result = _analyze(request) + + assert any( + edge.parent_evidence_ref == "email:observed-child" + and edge.child_evidence_ref == "email:later-child" + and edge.truth_status_code == "inferred" + for edge in result.edges + ) diff --git a/tests/test_external_lineage_public_api.py b/tests/test_external_lineage_public_api.py new file mode 100644 index 000000000..d30300efc --- /dev/null +++ b/tests/test_external_lineage_public_api.py @@ -0,0 +1,16 @@ +"""Public import-surface tests for external lineage consumers.""" + +from __future__ import annotations + +from lineageweave import external_lineage + + +def test_external_lineage_module_exports_the_versioned_contract() -> None: + assert external_lineage.CONTRACT_VERSION == "1.0.0" + assert callable(external_lineage.parse_lineage_analysis_request) + assert callable(external_lineage.analyze_external_lineage) + assert callable(external_lineage.request_digest) + assert callable(external_lineage.result_digest) + assert external_lineage.LineageContractError.__name__ == ( + "LineageContractError" + ) diff --git a/tests/test_import_postgresql_posts.py b/tests/test_import_postgresql_posts.py index cd35d4f57..5f38b7193 100644 --- a/tests/test_import_postgresql_posts.py +++ b/tests/test_import_postgresql_posts.py @@ -6,8 +6,10 @@ import pytest +from lineageweave.adjudication_client import AdjudicationClientError from scripts.import_postgresql_posts import ( _lineage_grouping_values, + _lineage_rebuild_summary, _normalize_voc_type, _parser, _source_code_matches, @@ -19,6 +21,26 @@ ) +def test_importer_keeps_rows_when_adjudication_response_is_unusable( + monkeypatch, +) -> None: + """A malformed optional score makes lineage unavailable, not the import lost.""" + + async def malformed_provider(_target, *, llm=None): + raise AdjudicationClientError("synthetic malformed confidence") + + monkeypatch.setattr( + "scripts.import_postgresql_posts.rebuild_lineage", malformed_provider + ) + + summary = asyncio.run(_lineage_rebuild_summary(object(), object())) + + assert summary["lineage_edges"] is None + assert "imported source rows remain persisted" in str( + summary["lineage_rebuild_unavailable"] + ) + + def test_placeholder_grouping_is_derived_without_losing_raw_source_values() -> None: assert callable(_lineage_grouping_values) mapping = SimpleNamespace(