diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 1cad1f17c..e3512675f 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -94,3 +94,71 @@ jobs: - name: Build Storybook working-directory: frontend run: pnpm run build-storybook + + repair_external_contract: + name: Repair external lineage contract integrity + if: github.event_name == 'pull_request' && github.event.pull_request.head.ref == 'feat/external-lineage-integration-contract' + runs-on: ubuntu-latest + timeout-minutes: 40 + permissions: + contents: write + services: + postgres: + image: postgres:16-alpine@sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f58fddf30ffc07777 + env: + POSTGRES_PASSWORD: postgres + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U postgres" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + env: + LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN: postgresql://postgres:postgres@localhost:5432/postgres + steps: + - name: Checkout exact contributor head without persisted credentials + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # actions/checkout@v7 + with: + ref: ${{ github.event.pull_request.head.sha }} + fetch-depth: 2 + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # actions/setup-python@v6 + with: + python-version: "3.12" + + - name: Set up locked dependency manager + uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0 + with: + version: "0.11.28" + enable-cache: false + + - name: Select pinned Rust toolchain + run: | + rustup toolchain install 1.97.1 --profile minimal + rustup default 1.97.1 + + - name: Install the committed universal lock + run: uv sync --frozen --extra dev --extra backend + + - name: Apply and verify the bounded repair without write credentials + run: uv run --frozen python scripts/run_pr_343_contract_integrity.py + + - name: Publish only after exact-head verification + env: + GH_TOKEN: ${{ github.token }} + shell: bash + run: | + git fetch --no-tags origin feat/external-lineage-integration-contract + test "$(git rev-parse FETCH_HEAD)" = "$(git rev-parse HEAD)" + git show HEAD^:.github/workflows/tests.yml > .github/workflows/tests.yml + rm scripts/repair_pr_343_contract_integrity.py + rm scripts/run_pr_343_contract_integrity.py + git config user.name "github-actions[bot]" + git config user.email "41898282+github-actions[bot]@users.noreply.github.com" + git add -A + git diff --cached --check + git commit -m "fix: bind and report external lineage truth exactly" + git push "https://x-access-token:${GH_TOKEN}@github.com/ContextualWisdomLab/LineageWeave.git" HEAD:feat/external-lineage-integration-contract diff --git a/CHANGELOG.d/external-lineage-contract.md b/CHANGELOG.d/external-lineage-contract.md new file mode 100644 index 000000000..4dbb7302d --- /dev/null +++ b/CHANGELOG.d/external-lineage-contract.md @@ -0,0 +1,11 @@ +# 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. +- Add JSON Schema Draft 2020-12, one canonical ADR 0124, APA 7th doctoring, and focused TDD coverage. diff --git a/docs/adr/0124-external-email-project-lineage-contract.md b/docs/adr/0124-external-email-project-lineage-contract.md new file mode 100644 index 000000000..27571cc7d --- /dev/null +++ b/docs/adr/0124-external-email-project-lineage-contract.md @@ -0,0 +1,48 @@ +# ADR 0124: 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. + +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. +- 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..86b529b08 --- /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/0124-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/superpowers/plans/2026-08-21-external-lineage-integration-contract.md b/docs/superpowers/plans/2026-08-21-external-lineage-integration-contract.md new file mode 100644 index 000000000..0a4703ea2 --- /dev/null +++ b/docs/superpowers/plans/2026-08-21-external-lineage-integration-contract.md @@ -0,0 +1,80 @@ +# External Lineage Integration Contract Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Publish a strict, store-agnostic LineageWeave contract that accepts bounded caller-authorized evidence and returns opaque-reference lineage and project projections for Naruon and other consumers. + +**Architecture:** Keep the existing reconstruction kernel authoritative for candidate scoring and RankWeave fusion. Add a pure contract/parser layer plus a pure execution adapter that performs available-time cutoff filtering, work-budget checks, explicit-vs-inferred relation separation, channel-evidence projection, and canonical digests without database or provider access. Caller-observed children bypass alternative inference and optional model disclosure while remaining available as candidate history for later records. + +**Tech Stack:** Python 3.12+, dataclasses, JSON Schema Draft 2020-12, pytest, coverage.py, Ruff, RankWeave, ThreadWeave. + +**Spec:** `docs/adr/0124-external-email-project-lineage-contract.md` + +## Global Constraints + +- Inputs contain only caller-authorized bounded evidence and opaque references. +- `available_at <= knowledge_cutoff` is the historical-evidence admission rule. +- Missing optional LLM evidence is unavailable, never a fabricated zero. +- Observed RFC/thread relations remain distinct from inferred semantic lineage. +- Children with an explicit observed parent consume no inferred-pair budget and are not sent to the optional LLM for an alternative edge. +- Project projections remain proposed and cannot mutate caller or provider state. +- No direct application-database access, provider credential, persistence, or network call is added to the pure execution adapter. +- Changed production statement and branch coverage must be 100%; public symbols require docstrings. + +--- + +### Task 1: Strict contract and canonical serialization + +**Files:** +- Create: `lineageweave/external_lineage_contract.py` +- Create: `docs/contracts/external-lineage-analysis-v1.schema.json` +- Test: `tests/test_external_lineage_contract.py` + +**Interfaces:** +- Produces: `parse_lineage_analysis_request(payload) -> LineageAnalysisRequest` +- Produces: `serialize_lineage_analysis_request(request) -> dict[str, object]` +- Produces: `serialize_lineage_analysis_result(result) -> dict[str, object]` +- Produces: `request_digest(request) -> str` and `result_digest(result) -> str` + +- [ ] Write failing parser tests for unknown fields, invalid vocabularies, duplicate opaque references, offset-naive timestamps, payload bounds, unsafe references, and policy bounds. +- [ ] Run `uv run --locked --extra dev pytest -q tests/test_external_lineage_contract.py` and confirm the tests fail because the contract does not exist. +- [ ] Implement immutable dataclasses, stable errors, strict parsing, canonical UTC serialization, digest calculation, and result-integrity checks. +- [ ] Add the Draft 2020-12 schema and a drift test comparing its fixed vocabularies and bounds with the parser. +- [ ] Re-run the focused contract tests until they pass. + +### Task 2: Evidence-bounded execution adapter + +**Files:** +- Create: `lineageweave/external_lineage_analysis.py` +- Test: `tests/test_external_lineage_analysis.py` +- Test: `tests/test_external_lineage_explicit_parent_budget.py` + +**Interfaces:** +- Consumes: `LineageAnalysisRequest` and the existing candidate-scoring/fusion kernel. +- Produces: `analyze_external_lineage(request, *, llm=None) -> LineageAnalysisResult`. + +- [ ] Write failing tests for available-time cutoff exclusion, pair-budget rejection before channel execution, explicit observed parent precedence, explicit-parent validation, optional LLM status, project projection, deterministic output, and content-minimized evidence. +- [ ] Add a failing regression proving caller-observed children neither consume inferred-pair budget nor disclose alternative label pairs to an optional LLM. +- [ ] Run the focused execution tests and confirm the missing or defective adapter is the failure cause. +- [ ] Implement request revalidation, explicit-parent acyclicity/group/time checks, cutoff filtering, inference-only pair-budget calculation, core-record adaptation, channel projection, limitations, and result digesting. +- [ ] Preserve an explicit child in candidate history so later unobserved records may still select it as an inferred parent. +- [ ] Re-run the focused execution tests until they pass. + +### Task 3: Public package, decision records, and quality gate + +**Files:** +- Modify: `lineageweave/__init__.py` +- Create: `docs/adr/0124-external-email-project-lineage-contract.md` +- Create: `docs/doctoring/EXTERNAL_LINEAGE_CONTRACT_REFERENCES.md` +- Create: `CHANGELOG.d/external-lineage-contract.md` + +**Interfaces:** +- Produces: a supported package API for consumer contract tests. + +- [ ] Export the contract types, parser/serializer/digest functions, error type, and `analyze_external_lineage` from `lineageweave`. +- [ ] Record the LineageWeave/Naruon authority split, truth statuses, cutoff semantics, model-disclosure minimization, and packaging boundary in the ADR. +- [ ] Record APA 7th sources and one consolidated changelog fragment. +- [ ] Run `uvx ruff check` on changed Python and test files. +- [ ] Run focused statement/branch coverage with `--fail-under=100` for both new production modules. +- [ ] Run documentation hygiene, schema JSON parsing, Python compileall, and `git diff --check`. +- [ ] Open a Draft PR linked to LineageWeave #338; keep Naruon runtime integration out of this slice. diff --git a/docs/superpowers/plans/2026-08-21-naruon-email-project-lineage-contract.md b/docs/superpowers/plans/2026-08-21-naruon-email-project-lineage-contract.md new file mode 100644 index 000000000..e379ee4ec --- /dev/null +++ b/docs/superpowers/plans/2026-08-21-naruon-email-project-lineage-contract.md @@ -0,0 +1,68 @@ +# Naruon Email and Project Lineage Contract Implementation Plan + +> **For agentic workers:** Use `superpowers:subagent-driven-development` or `superpowers:executing-plans` task by task. Follow TDD and verify exact-head evidence before publication. + +**Goal:** Publish a strict, store-agnostic LineageWeave package contract that Naruon can later consume for evidence-bounded email lineage and project-history candidates. + +**Architecture:** `external_lineage_contract.py` owns immutable request/result models, strict parsing, canonical serialization, bounded vocabularies, and deterministic SHA-256 digests. `external_lineage_analysis.py` adapts authorized records to the current reconstruction kernel, enforces `available_at <= knowledge_cutoff`, validates caller-observed parent relations, rejects excess pair work before optional provider activity, and projects inferred channel evidence plus proposed project groupings. + +**Tech stack:** Python 3.12+, standard-library dataclasses/JSON/datetime/hashlib, existing LineageWeave reconstruction kernel, pytest, coverage.py, JSON Schema Draft 2020-12. + +## Global constraints + +- Contract version: `1.0.0`. +- Request records: 1–500. +- Candidate window: 1–200. +- Candidate-pair budget: 1–5,000 and enforced before optional LLM calls. +- Identifiers are opaque bounded references; provider credentials and direct database access are forbidden. +- Timestamps are offset-aware RFC 3339 and serialize in UTC with `Z`. +- Caller parent relations remain `observed`, are same-group and acyclic; reconstructed edges remain `inferred`. +- Missing LLM evidence is unavailable, never zero. +- Project projections remain `proposed`. +- New production statement/branch coverage and public docstrings: 100%. + +## Task 1 — Strict contract + +**Files:** +- `lineageweave/external_lineage_contract.py` +- `tests/test_external_lineage_contract.py` + +- [x] Write failing tests for strict object parsing, unknown fields, duplicate references, bounded identifiers/text, offset-aware timestamps, policy bounds, canonical serialization, deterministic digests, result partitions, channel math, and package exports. +- [x] Confirm RED before implementation. +- [x] Implement frozen dataclasses, `LineageContractError`, parser, request/result serializers, and digest functions. +- [x] Confirm focused tests GREEN. + +## Task 2 — Reconstruction adapter + +**Files:** +- `lineageweave/external_lineage_analysis.py` +- `tests/test_external_lineage_analysis.py` + +- [x] Write failing tests for cutoff filtering, explicit RFC parent precedence, cycle/missing/forward-parent rejection, pair-budget pre-call enforcement, LLM status, per-channel evidence, project grouping, and deterministic results. +- [x] Confirm RED before implementation. +- [x] Implement request round-trip validation, available-time partitioning, explicit-parent validation, exact candidate-pair budgeting, core-kernel adaptation, observed/inferred edge projection, limitations, and project projections. +- [x] Confirm focused tests GREEN. + +## Task 3 — Public schema and architecture evidence + +**Files:** +- `docs/contracts/external-lineage-analysis-v1.schema.json` +- `docs/adr/0124-external-email-project-lineage-contract.md` +- `docs/doctoring/EXTERNAL_LINEAGE_CONTRACT_REFERENCES.md` +- `CHANGELOG.d/external-lineage-contract.md` +- `lineageweave/__init__.py` + +- [x] Add JSON Schema Draft 2020-12 mirroring parser names, bounds, and vocabularies. +- [x] Add ADR 0124 and APA 7th references for RFC 3339, RFC 5322, RFC 5256, PROV-O, and OWL-Time. +- [x] Export the contract and adapter from the package root. +- [x] Add changelog evidence. + +## Task 4 — Exact-head verification and Draft PR + +- [x] Focused suite: `57 passed`. +- [x] Isolated new-module coverage: 425/425 statements and 140/140 branches, 100%. +- [x] Public function/class/module docstrings: complete. +- [x] `compileall`, JSON syntax, line-length, and `git diff --check`: passed. +- [ ] Publish the branch from exact protected `main@2feba74b75863810869cde680b19032a93fba413`. +- [ ] Open one Draft PR tracking LineageWeave #338. +- [ ] Keep Draft until exact-head hosted CI/security/documentation gates, review-thread resolution, and independent approval pass. diff --git a/docs/superpowers/specs/2026-08-21-naruon-email-project-lineage-contract-design.md b/docs/superpowers/specs/2026-08-21-naruon-email-project-lineage-contract-design.md new file mode 100644 index 000000000..10a6b3ddb --- /dev/null +++ b/docs/superpowers/specs/2026-08-21-naruon-email-project-lineage-contract-design.md @@ -0,0 +1,108 @@ +# Naruon Email and Project Lineage Contract Design + +## Status + +Accepted for implementation on 2026-08-21 through the user instruction to continue the cross-repository LineageWeave/Naruon integration work. + +## Problem + +Naruon owns customer mail, canonical message/thread identities, projects, tasks, commitments, provider credentials, authorization, and provider mutations. LineageWeave owns lineage reconstruction and the evidence explaining that reconstruction. A future integration needs a released, store-agnostic boundary between those products. Direct database access, copied source, and mutable submodules would collapse their authority boundaries. + +## Decision + +LineageWeave will publish a strict versioned Python contract that accepts bounded caller-authorized evidence and returns only opaque-reference lineage results. The first slice is an in-process, store-agnostic package boundary with no persistence or network access. Naruon can later consume the same schema through a package, service, or reviewed plugin adapter. + +The contract has two layers: + +1. `external_lineage_contract.py` strictly parses and serializes version `1.0.0` requests and results, enforces bounds, normalizes offset-aware timestamps, and computes deterministic content digests. +2. `external_lineage_analysis.py` adapts authorized records to the existing LineageWeave reconstruction kernel, preserves caller-observed parent relations ahead of inference, exposes per-channel evidence, enforces knowledge cutoffs using `available_at`, and emits project evidence groupings without promoting them to Naruon project truth. + +## Authority and truth model + +- Caller-supplied records are `observed` or `authoritative_in_caller` evidence. +- Caller-supplied explicit parent relations remain `observed`; they are not reclassified as semantic inference. +- LineageWeave reconstructed continuation edges are `inferred`. +- Project projections are `proposed` groupings from caller-supplied project references; they never claim authoritative Naruon project state. +- Missing LLM evidence is `unavailable`, never a numeric zero. +- Provider credentials, access tokens, mailbox access, and unrelated tenant data are outside the contract. + +## Request contract + +A request carries: + +- immutable `analysis_id`; +- `analysis_scope_code` in `email_lineage`, `project_history`, or `generic_lineage`; +- optional offset-aware `knowledge_cutoff`; +- bounded reconstruction policy, including a maximum of 5,000 declared candidate-pair evaluations; +- one to 500 evidence records. + +Each evidence record carries: + +- opaque `evidence_ref` and `group_ref`; +- source kind and caller truth status; +- bounded label text; +- offset-aware `occurred_at` and `available_at`; +- optional single `secondary_key` used by the current reconstruction kernel; +- optional `project_ref` used only for proposed project grouping; +- optional explicit parent relation with a controlled relation code. + +## Result contract + +A result carries: + +- deterministic `result_digest`; +- included and cutoff-excluded evidence references; +- LLM channel status; +- stable-sorted edges; +- per-channel score, normalized active weight, and contribution; +- proposed project groupings; +- explicit limitations. + +## Temporal safety + +Historical analysis is governed by: + +```text +available_at <= knowledge_cutoff +``` + +`occurred_at` describes the event/message time; `available_at` describes when the caller could use the evidence. Evidence first available after the cutoff is excluded even if it describes an earlier event. + +## Email safety + +RFC reply/thread evidence and semantic lineage are separate: + +- `rfc_reply`, `provider_reply`, and `manual_parent` are caller-observed explicit relations. +- `reconstructed_continuation` is a LineageWeave inference. +- Provider thread IDs or caller project keys can be supplied only as opaque secondary keys. +- The package never parses a mailbox, fetches a provider, or mutates mail state. + +## Project safety + +The result may group evidence under caller-supplied opaque `project_ref` values. This is a proposed evidence projection only. Naruon must apply its own deterministic or human approval policy before updating authoritative project/task/commitment state. + +## Error handling + +Unknown fields, duplicate references, unsafe or empty identifiers, naive timestamps, non-finite scores, invalid policy bounds, missing explicit parents, forward-inconsistent explicit parents, and unsupported vocabularies fail closed with `LineageContractError` and stable reason codes. + +## Testing + +The implementation uses TDD and must prove: + +- strict parsing and canonical timestamp normalization; +- bounded payloads, duplicate rejection, and pre-provider candidate-pair budget enforcement; +- knowledge-cutoff exclusion by available time; +- observed explicit relations override inferred parent choices; +- RFC reply evidence remains distinct from semantic/project inference; +- absent LLM evidence is explicit; +- deterministic request/result digests; +- no omitted evidence reference can appear in output; +- proposed project groupings never claim caller authority; +- statement and branch coverage for the new production modules are 100%. + +## Standards + +- RFC 3339 for offset-aware timestamps. +- RFC 5322 and RFC 5256 for preserving email identity/thread evidence distinctions. +- W3C PROV-O for provenance and evidence authority. +- W3C OWL-Time for temporal interpretation boundaries. diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index 95330cb50..ed54d7cd6 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -10,6 +10,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 lineage_edge_specs from .models import Edge, Record, Tree @@ -30,8 +49,18 @@ from .voc_evidence import sentence_excerpts __all__ = [ + "CONTRACT_VERSION", + "ChannelEvidence", "ChatAnswer", "Edge", + "ExplicitParent", + "LineageAnalysisPolicy", + "LineageAnalysisRequest", + "LineageAnalysisResult", + "LineageContractError", + "LineageEdgeResult", + "LineageEvidenceRecord", + "LineageLimitation", "OrganizationRelationship", "PROV", "PROV_CLASSES", @@ -39,20 +68,27 @@ "PROV_RELATIONS", "PROV_RECOMMENDED_INVERSES", "PostSummary", + "ProjectProjection", "ProvAssertion", "ProvGraph", "ProvLiteral", "ProvValidationError", "Record", "Tree", + "analyze_external_lineage", "build_affiliate_forest", "cited_post_summaries", "lineage_edge_specs", + "parse_lineage_analysis_request", "random_walk_with_restart", "reconstruct", + "request_digest", "resolve_corporate_entity", + "result_digest", "select_related_nodes", "sentence_excerpts", + "serialize_lineage_analysis_request", + "serialize_lineage_analysis_result", ] __version__ = "2.12.6" diff --git a/lineageweave/adjudication_client.py b/lineageweave/adjudication_client.py index 37e4fee7f..566947833 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.""" @@ -38,7 +42,20 @@ def judge(self, candidate_label: str, record_label: str) -> float: # pragma: no raise RuntimeError("NullAdjudicationClient has no llm channel; check .available first") -_CONFIDENCE_PATTERN = re.compile(r"([01](?:\.\d+)?)") +_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 _CONFIDENCE_PATTERN.fullmatch(normalized) is None: + raise AdjudicationClientError( + "provider confidence response was not a number in 0..1" + ) + return float(normalized) class ContextualOrchestratorAdjudicationClient: @@ -76,8 +93,10 @@ def judge(self, candidate_label: str, record_label: str) -> float: headers={"authorization": f"Bearer {self._api_key}"}, timeout=self._timeout, ) - content = chat_completion_content(body) - match = _CONFIDENCE_PATTERN.search(content) - if match is None: - return 0.0 - return max(0.0, min(1.0, float(match.group(1)))) + try: + content = chat_completion_content(body) + except (TypeError, ValueError) as exc: + 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..6e606ce7f --- /dev/null +++ b/lineageweave/external_lineage_analysis.py @@ -0,0 +1,482 @@ +"""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 .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, +) -> 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): + 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 _pair_evaluation_count( + records: tuple[LineageEvidenceRecord, ...], + candidate_window: int, +) -> int: + """Count only candidate pairs that require inferred parent selection.""" + + return sum( + min(index, candidate_window) + for group_records in _ordered_contract_groups(records) + for index, record in enumerate(group_records) + if record.explicit_parent is None + ) + + +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, +) -> list[LineageEdgeResult]: + """Select inferred parents without rescoring explicit observed children.""" + + if not records: + return [] + weights = active_weights(llm) + 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 + ] + 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, +) -> 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) + + inferred = _inferred_edges( + included, + selected_llm, + validated, + ) + 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 + ] + 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..f759048e9 --- /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 datetime, timezone +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(timezone.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(timezone.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/repair_pr_343_contract_integrity.py b/scripts/repair_pr_343_contract_integrity.py new file mode 100644 index 000000000..9da8e5a14 --- /dev/null +++ b/scripts/repair_pr_343_contract_integrity.py @@ -0,0 +1,535 @@ +"""Apply and verify PR 343 contract-integrity and stack-consolidation fixes.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def _run(*args: str, check: bool = True) -> subprocess.CompletedProcess[str]: + """Run one repository command and surface captured output.""" + + completed = subprocess.run( + args, + cwd=ROOT, + check=False, + text=True, + capture_output=True, + ) + if completed.stdout: + print(completed.stdout, end="") + if completed.stderr: + print(completed.stderr, end="", file=sys.stderr) + if check and completed.returncode != 0: + raise SystemExit(completed.returncode) + return completed + + +def _replace_once(text: str, old: str, new: str, *, label: str) -> str: + """Replace one known fragment or fail closed when the contributor head moved.""" + + if new in text: + return text + if text.count(old) != 1: + raise SystemExit(f"refusing unknown {label} shape") + return text.replace(old, new, 1) + + +def _add_regressions() -> None: + """Add authorization, malformed-provider, and LLM invocation tests first.""" + + reconstruct_path = ROOT / "tests/test_reconstruct.py" + reconstruct = reconstruct_path.read_text(encoding="utf-8") + reconstruct = _replace_once( + reconstruct, + "from lineageweave import Record, reconstruct\n", + "from lineageweave import Record, reconstruct\nfrom lineageweave.adjudication_client import AdjudicationClientError\n", + label="reconstruct adjudication import", + ) + anchor = '''def test_candidate_window_bounds_which_priors_are_considered() -> None: +''' + regression = '''class _MalformedAdjudicationClient: + """Provider boundary that returns no usable confidence for any pair.""" + + available = True + + def judge(self, candidate_label: str, record_label: str) -> float: + """Raise the typed provider-shape error used by the production client.""" + + raise AdjudicationClientError("verbose provider reply") + + +def test_core_reconstruction_degrades_one_malformed_llm_pair_to_zero() -> None: + """The legacy core must not abort a whole group on one malformed reply.""" + + trees = reconstruct(sample_records(), llm=_MalformedAdjudicationClient()) + + assert trees + assert all( + edge.channel_scores.get("llm") == 0.0 + for tree in trees + for edge in tree.edges + ) + + +''' + if regression not in reconstruct: + if anchor not in reconstruct: + raise SystemExit("refusing unknown reconstruct test insertion point") + reconstruct = reconstruct.replace(anchor, regression + anchor, 1) + reconstruct_path.write_text(reconstruct, encoding="utf-8") + + analysis_path = ROOT / "tests/test_external_lineage_analysis.py" + analysis = analysis_path.read_text(encoding="utf-8") + anchor = '''def test_project_projection_is_proposed_and_uses_only_included_evidence() -> None: +''' + regression = '''def test_llm_status_is_not_invoked_without_an_inferred_candidate_pair() -> None: + client = CountingLlm() + request = _request( + [ + _record( + "email:single", + "One bounded record", + "2026-08-20T09:00:00Z", + ) + ], + allow_llm=True, + ) + + result = analyze_external_lineage(request, llm=client) + + assert client.call_count == 0 + assert result.llm_status_code == "not_invoked" + assert result.edges == () + + +''' + if regression not in analysis: + if anchor not in analysis: + raise SystemExit("refusing unknown external analysis insertion point") + analysis = analysis.replace(anchor, regression + anchor, 1) + analysis_path.write_text(analysis, encoding="utf-8") + + contract_path = ROOT / "tests/test_external_lineage_contract.py" + contract = contract_path.read_text(encoding="utf-8") + contract = _replace_once( + contract, + ' "analysis_id": "analysis:demo-001",\n "analysis_scope_code": "email_lineage",\n', + ' "analysis_id": "analysis:demo-001",\n "authorization_scope_ref": "authorization-scope:opaque",\n "analysis_scope_code": "email_lineage",\n', + label="contract payload authorization scope", + ) + contract = _replace_once( + contract, + ' assert request.analysis_id == "analysis:demo-001"\n assert request.analysis_scope_code == "email_lineage"\n', + ' assert request.analysis_id == "analysis:demo-001"\n assert request.authorization_scope_ref == "authorization-scope:opaque"\n assert request.analysis_scope_code == "email_lineage"\n', + label="authorization parse assertion", + ) + contract = _replace_once( + contract, + ' "analysis_scope_code": payload["analysis_scope_code"],\n "analysis_id": payload["analysis_id"],\n', + ' "analysis_scope_code": payload["analysis_scope_code"],\n "authorization_scope_ref": payload["authorization_scope_ref"],\n "analysis_id": payload["analysis_id"],\n', + label="reordered authorization scope", + ) + schema_anchor = ''' assert schema["additionalProperties"] is False +''' + schema_assertion = ''' assert "authorization_scope_ref" in schema["required"] + assert schema["properties"]["authorization_scope_ref"] == { + "$ref": "#/$defs/OpaqueReference" + } + assert "not_invoked" in schema["$defs"]["LineageAnalysisResult"][ + "properties" + ]["llm_status_code"]["enum"] +''' + if schema_assertion not in contract: + if schema_anchor not in contract: + raise SystemExit("refusing unknown schema assertion point") + contract = contract.replace(schema_anchor, schema_anchor + schema_assertion, 1) + missing_anchor = ''' payload = _payload() + del payload["analysis_id"] + with pytest.raises(LineageContractError) as missing: + parse_lineage_analysis_request(payload) + assert missing.value.code == "missing_field" +''' + missing_replacement = missing_anchor + ''' + payload = _payload() + del payload["authorization_scope_ref"] + with pytest.raises(LineageContractError) as missing_scope: + parse_lineage_analysis_request(payload) + assert missing_scope.value.code == "missing_field" +''' + contract = _replace_once( + contract, + missing_anchor, + missing_replacement, + label="authorization required-field regression", + ) + unsafe_anchor = '''def test_parser_rejects_invalid_policy_values( +''' + unsafe_test = '''def test_parser_rejects_unsafe_authorization_scope_reference() -> None: + payload = _payload() + payload["authorization_scope_ref"] = "https://caller.example/scope" + + with pytest.raises(LineageContractError) as captured: + parse_lineage_analysis_request(payload) + + assert captured.value.code == "unsafe_opaque_reference" + + +''' + if unsafe_test not in contract: + if unsafe_anchor not in contract: + raise SystemExit("refusing unknown authorization safety insertion point") + contract = contract.replace(unsafe_anchor, unsafe_test + unsafe_anchor, 1) + contract_path.write_text(contract, encoding="utf-8") + + +def _apply_core_repair() -> None: + """Preserve legacy graceful degradation for typed malformed LLM replies.""" + + path = ROOT / "lineageweave/reconstruct.py" + text = path.read_text(encoding="utf-8") + text = _replace_once( + text, + "from .adjudication_client import AdjudicationClient, NullAdjudicationClient\n", + "from .adjudication_client import (\n AdjudicationClient,\n AdjudicationClientError,\n NullAdjudicationClient,\n)\n", + label="core adjudication import", + ) + text = _replace_once( + text, + ''' if "llm" in weights: + scores["llm"] = llm.judge(candidate.label, record.label) +''', + ''' if "llm" in weights: + try: + scores["llm"] = llm.judge(candidate.label, record.label) + except AdjudicationClientError: + # The long-lived core historically degraded one malformed model + # reply to a zero contribution instead of aborting the group. + # External contract wrappers raise LineageContractError and + # therefore retain their stricter fail-closed boundary. + scores["llm"] = 0.0 +''', + label="core malformed-provider handling", + ) + path.write_text(text, encoding="utf-8") + + +def _apply_contract_repair() -> None: + """Bind caller authorization and distinguish a configured but unused model.""" + + path = ROOT / "lineageweave/external_lineage_contract.py" + text = path.read_text(encoding="utf-8") + text = _replace_once( + text, + 'LlmStatusCode = Literal["not_requested", "unavailable", "completed"]\n', + 'LlmStatusCode = Literal["not_requested", "unavailable", "not_invoked", "completed"]\n', + label="LLM status vocabulary", + ) + text = _replace_once( + text, + '_LLM_STATUSES = frozenset({"not_requested", "unavailable", "completed"})\n', + '_LLM_STATUSES = frozenset({"not_requested", "unavailable", "not_invoked", "completed"})\n', + label="LLM status set", + ) + text = _replace_once( + text, + ''' contract_version: str + analysis_id: str + analysis_scope_code: AnalysisScopeCode +''', + ''' contract_version: str + analysis_id: str + authorization_scope_ref: str + analysis_scope_code: AnalysisScopeCode +''', + label="request authorization field", + ) + text = _replace_once( + text, + ''' "contract_version", + "analysis_id", + "analysis_scope_code", +''', + ''' "contract_version", + "analysis_id", + "authorization_scope_ref", + "analysis_scope_code", +''', + label="allowed authorization field", + ) + text = _replace_once( + text, + ''' "contract_version", + "analysis_id", + "analysis_scope_code", + "policy", +''', + ''' "contract_version", + "analysis_id", + "authorization_scope_ref", + "analysis_scope_code", + "policy", +''', + label="required authorization field", + ) + text = _replace_once( + text, + ''' analysis_id=cast( + str, + _opaque_reference(data["analysis_id"], field="analysis_id"), + ), + analysis_scope_code=cast( +''', + ''' analysis_id=cast( + str, + _opaque_reference(data["analysis_id"], field="analysis_id"), + ), + authorization_scope_ref=cast( + str, + _opaque_reference( + data["authorization_scope_ref"], + field="authorization_scope_ref", + ), + ), + analysis_scope_code=cast( +''', + label="authorization parser", + ) + text = _replace_once( + text, + ''' "contract_version": request.contract_version, + "analysis_id": request.analysis_id, + "analysis_scope_code": request.analysis_scope_code, +''', + ''' "contract_version": request.contract_version, + "analysis_id": request.analysis_id, + "authorization_scope_ref": request.authorization_scope_ref, + "analysis_scope_code": request.analysis_scope_code, +''', + label="authorization serializer", + ) + path.write_text(text, encoding="utf-8") + + +def _apply_analysis_repair() -> None: + """Track actual LLM invocation and decouple wire math/order from RankWeave.""" + + path = ROOT / "lineageweave/external_lineage_analysis.py" + text = path.read_text(encoding="utf-8") + text = _replace_once( + text, + ''' self._client = client + + def judge(self, candidate_label: str, record_label: str) -> float: +''', + ''' self._client = client + self.invocation_count = 0 + + def judge(self, candidate_label: str, record_label: str) -> float: +''', + label="bounded client invocation counter", + ) + text = _replace_once( + text, + ''' try: + score = self._client.judge(candidate_label, record_label) +''', + ''' self.invocation_count += 1 + try: + score = self._client.judge(candidate_label, record_label) +''', + label="bounded client invocation accounting", + ) + text = _replace_once( + text, + ' return _BoundedAdjudicationClient(llm), "completed"\n', + ' return _BoundedAdjudicationClient(llm), "not_invoked"\n', + label="initial LLM status", + ) + old_edge = ''' 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, + ), + ) + ) +''' + new_edge = ''' channel_evidence = _channel_evidence(channel_scores, weights) + contract_fused_score = sum( + item.contribution for item in channel_evidence + ) + 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=contract_fused_score, + channel_evidence=channel_evidence, + ) + ) +''' + text = _replace_once(text, old_edge, new_edge, label="contract fused score") + text = _replace_once( + text, + ''' inferred = _inferred_edges( + included, + selected_llm, + validated, + ) +''', + ''' inferred = _inferred_edges( + included, + selected_llm, + validated, + ) + if ( + llm_status == "not_invoked" + and isinstance(selected_llm, _BoundedAdjudicationClient) + and selected_llm.invocation_count > 0 + ): + llm_status = "completed" +''', + label="completed LLM status", + ) + old_order = ''' edge_order = { + record.evidence_ref: (record.group_ref, record.occurred_at, record.evidence_ref) + for record in included + } + result = LineageAnalysisResult( +''' + text = _replace_once(text, old_order, ' result = LineageAnalysisResult(\n', label="edge order projection") + text = _replace_once( + text, + ''' key=lambda item: ( + edge_order[item.child_evidence_ref], + item.parent_evidence_ref, + item.relation_type_code, + ), +''', + ''' key=lambda item: ( + item.child_evidence_ref, + item.parent_evidence_ref, + item.relation_type_code, + ), +''', + label="canonical in-memory edge order", + ) + path.write_text(text, encoding="utf-8") + + +def _apply_schema_and_docs() -> None: + """Synchronize the public schema, example, authorization note, and changelog.""" + + schema_path = ROOT / "docs/contracts/external-lineage-analysis-v1.schema.json" + schema = json.loads(schema_path.read_text(encoding="utf-8")) + required = schema["required"] + if "authorization_scope_ref" not in required: + required.insert(required.index("analysis_scope_code"), "authorization_scope_ref") + schema["properties"]["authorization_scope_ref"] = { + "$ref": "#/$defs/OpaqueReference" + } + statuses = schema["$defs"]["LineageAnalysisResult"]["properties"][ + "llm_status_code" + ]["enum"] + if "not_invoked" not in statuses: + statuses.insert(statuses.index("completed"), "not_invoked") + schema_path.write_text( + json.dumps(schema, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + + example_path = ROOT / "docs/contracts/external-lineage-analysis-v1.example.json" + example = json.loads(example_path.read_text(encoding="utf-8")) + ordered = { + "contract_version": example["contract_version"], + "analysis_id": example["analysis_id"], + "authorization_scope_ref": "authorization-scope:synthetic", + **{ + key: value + for key, value in example.items() + if key not in {"contract_version", "analysis_id"} + }, + } + example_path.write_text( + json.dumps(ordered, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + + authorization_path = ROOT / "docs/contracts/external-lineage-analysis-v1.authorization.md" + authorization = authorization_path.read_text(encoding="utf-8") + paragraph = ( + "\n\nEvery request must carry an opaque `authorization_scope_ref` issued and " + "validated by the caller. LineageWeave includes it in canonical request identity " + "but does not dereference it or infer authorization from it.\n" + ) + if paragraph.strip() not in authorization: + authorization = authorization.rstrip() + paragraph + authorization_path.write_text(authorization, encoding="utf-8") + + changelog_path = ROOT / "CHANGELOG.d/external-lineage-contract.md" + changelog = changelog_path.read_text(encoding="utf-8") + bullet = ( + "- Bind each request to a caller-owned opaque authorization scope, distinguish a " + "configured but uninvoked LLM from completed adjudication, and preserve graceful " + "legacy reconstruction when a provider returns malformed confidence text.\n" + ) + if bullet not in changelog: + changelog = changelog.rstrip() + "\n" + bullet + changelog_path.write_text(changelog, encoding="utf-8") + + +def main() -> None: + """Prove RED, apply the bounded repair, then prove focused and full GREEN.""" + + _add_regressions() + focused = ( + "tests/test_reconstruct.py::test_core_reconstruction_degrades_one_malformed_llm_pair_to_zero", + "tests/test_external_lineage_analysis.py::test_llm_status_is_not_invoked_without_an_inferred_candidate_pair", + "tests/test_external_lineage_contract.py::test_parse_request_is_strict_immutable_and_canonicalizes_timestamps", + "tests/test_external_lineage_contract.py::test_parser_rejects_unsafe_authorization_scope_reference", + "tests/test_external_lineage_contract.py::test_public_schema_exists_and_mirrors_contract_vocabularies", + ) + red = _run( + "uv", + "run", + "--frozen", + "python", + "-m", + "pytest", + "-q", + *focused, + check=False, + ) + if red.returncode == 0: + raise SystemExit("PR 343 regressions unexpectedly passed before the repair") + _apply_core_repair() + _apply_contract_repair() + _apply_analysis_repair() + _apply_schema_and_docs() + _run( + "uv", + "run", + "--frozen", + "python", + "-m", + "pytest", + "-q", + *focused, + ) + _run("uv", "run", "--frozen", "python", "-m", "pytest", "-q") + + +if __name__ == "__main__": + main() diff --git a/scripts/run_pr_343_contract_integrity.py b/scripts/run_pr_343_contract_integrity.py new file mode 100644 index 000000000..d92811b98 --- /dev/null +++ b/scripts/run_pr_343_contract_integrity.py @@ -0,0 +1,112 @@ +"""Normalize and execute the temporary PR 343 repair script.""" + +from __future__ import annotations + +import runpy +from pathlib import Path + + +SCRIPT_PATH = Path(__file__).with_name("repair_pr_343_contract_integrity.py") + + +def _replace_once(text: str, old: str, new: str, label: str) -> str: + """Replace one temporary-script fragment or fail closed.""" + + if text.count(old) != 1: + raise SystemExit(f"refusing unknown {label} repair-script shape") + return text.replace(old, new, 1) + + +def main() -> None: + """Tighten ambiguous source anchors, then execute the reviewed repair.""" + + text = SCRIPT_PATH.read_text(encoding="utf-8") + + old = ''' ''' + '"""' + ''' contract_version: str + analysis_id: str + analysis_scope_code: AnalysisScopeCode +''' + '"""' + ''', + ''' + '"""' + ''' contract_version: str + analysis_id: str + authorization_scope_ref: str + analysis_scope_code: AnalysisScopeCode +''' + '"""' + ''', + label="request authorization field", +''' + new = ''' ''' + '"""' + ''' contract_version: str + analysis_id: str + analysis_scope_code: AnalysisScopeCode + knowledge_cutoff: datetime | None + policy: LineageAnalysisPolicy +''' + '"""' + ''', + ''' + '"""' + ''' contract_version: str + analysis_id: str + authorization_scope_ref: str + analysis_scope_code: AnalysisScopeCode + knowledge_cutoff: datetime | None + policy: LineageAnalysisPolicy +''' + '"""' + ''', + label="request authorization field", +''' + text = _replace_once(text, old, new, "request dataclass") + + old = ''' ''' + '"""' + ''' "contract_version", + "analysis_id", + "analysis_scope_code", +''' + '"""' + ''', + ''' + '"""' + ''' "contract_version", + "analysis_id", + "authorization_scope_ref", + "analysis_scope_code", +''' + '"""' + ''', + label="allowed authorization field", +''' + new = ''' ''' + '"""' + ''' "contract_version", + "analysis_id", + "analysis_scope_code", + "knowledge_cutoff", +''' + '"""' + ''', + ''' + '"""' + ''' "contract_version", + "analysis_id", + "authorization_scope_ref", + "analysis_scope_code", + "knowledge_cutoff", +''' + '"""' + ''', + label="allowed authorization field", +''' + text = _replace_once(text, old, new, "allowed field") + + old = ''' ''' + '"""' + ''' "contract_version": request.contract_version, + "analysis_id": request.analysis_id, + "analysis_scope_code": request.analysis_scope_code, +''' + '"""' + ''', + ''' + '"""' + ''' "contract_version": request.contract_version, + "analysis_id": request.analysis_id, + "authorization_scope_ref": request.authorization_scope_ref, + "analysis_scope_code": request.analysis_scope_code, +''' + '"""' + ''', + label="authorization serializer", +''' + new = ''' ''' + '"""' + ''' "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": { +''' + '"""' + ''', + ''' + '"""' + ''' "contract_version": request.contract_version, + "analysis_id": request.analysis_id, + "authorization_scope_ref": request.authorization_scope_ref, + "analysis_scope_code": request.analysis_scope_code, + "knowledge_cutoff": _time_text(request.knowledge_cutoff), + "policy": { +''' + '"""' + ''', + label="authorization serializer", +''' + text = _replace_once(text, old, new, "request serializer") + + SCRIPT_PATH.write_text(text, encoding="utf-8") + runpy.run_path(str(SCRIPT_PATH), run_name="__main__") + + +if __name__ == "__main__": + main() diff --git a/tests/test_adjudication_client.py b/tests/test_adjudication_client.py index 576f5e9c4..f0e061f32 100644 --- a/tests/test_adjudication_client.py +++ b/tests/test_adjudication_client.py @@ -1,6 +1,73 @@ +"""Provider-response contract tests for LLM adjudication.""" + from __future__ import annotations -from lineageweave.adjudication_client import ContextualOrchestratorAdjudicationClient +import pytest + +import lineageweave.adjudication_client as module +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(module, "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( + module, + "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: diff --git a/tests/test_external_lineage_analysis.py b/tests/test_external_lineage_analysis.py new file mode 100644 index 000000000..a9335f4e8 --- /dev/null +++ b/tests/test_external_lineage_analysis.py @@ -0,0 +1,716 @@ +"""Execution tests for the external Naruon-facing lineage adapter.""" + +from __future__ import annotations + +import pytest + +from lineageweave.external_lineage_analysis import ( + _channel_evidence, + analyze_external_lineage, +) +from lineageweave.external_lineage_contract import ( + LineageContractError, + parse_lineage_analysis_request, + request_digest, + result_digest, +) + + +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_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_external_lineage(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_external_lineage(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_external_lineage(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(), "completed", True), + ], +) +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_external_lineage(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_external_lineage(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_external_lineage(first_request) + second = analyze_external_lineage(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_external_lineage(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_external_lineage(request) + + assert captured.value.code == "explicit_parent_cycle" + + +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_external_lineage(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_external_lineage(request) + + assert result.included_evidence_refs == () + assert result.edges == () + assert result.project_projections == () + + +def test_invalid_llm_score_fails_closed_before_result_projection() -> None: + request = _request( + [ + _record( + "email:001", + "Phoenix one", + "2026-08-20T09:00:00Z", + ), + _record( + "email:002", + "Phoenix two", + "2026-08-20T09:01:00Z", + ), + ], + allow_llm=True, + ) + + with pytest.raises(LineageContractError) as captured: + analyze_external_lineage(request, llm=InvalidLlm()) + + assert captured.value.code == "channel_score_out_of_bounds" + + +def test_non_numeric_llm_score_fails_closed_at_the_contract_boundary() -> None: + """A provider score with the wrong type becomes a stable contract error.""" + + request = _request( + [ + _record("email:001", "Phoenix one", "2026-08-20T09:00:00Z"), + _record("email:002", "Phoenix two", "2026-08-20T09:01:00Z"), + ], + allow_llm=True, + ) + + with pytest.raises(LineageContractError) as captured: + analyze_external_lineage(request, llm=TextLlm()) + + assert captured.value.code == "channel_score_out_of_bounds" + + +def test_raw_provider_response_error_is_stable_at_the_contract_boundary() -> None: + """A raw provider failure is not exposed as an arbitrary exception.""" + + request = _request( + [ + _record("email:001", "Phoenix one", "2026-08-20T09:00:00Z"), + _record("email:002", "Phoenix two", "2026-08-20T09:01:00Z"), + ], + allow_llm=True, + ) + + with pytest.raises(LineageContractError) as captured: + analyze_external_lineage(request, llm=BrokenProviderLlm()) + + 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_external_lineage(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_external_lineage(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_external_lineage(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_external_lineage(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_external_lineage(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..4fe7bb0d1 --- /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 datetime, timezone +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=timezone.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=timezone.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), + ) + 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..3455278f8 --- /dev/null +++ b/tests/test_external_lineage_explicit_parent_budget.py @@ -0,0 +1,157 @@ +"""Regression tests for explicit-parent budget and provider minimization.""" + +from __future__ import annotations + +from lineageweave.external_lineage_analysis import analyze_external_lineage +from lineageweave.external_lineage_contract import parse_lineage_analysis_request + + +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_external_lineage(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_external_lineage(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..9d45f5791 --- /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 + +import lineageweave.external_lineage as 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" + )