from gewu.adapters.text_extractor import TextMarkdownExtractor from gewu.domain.contracts import ( EdgeEndpoint, KnowledgeNode, Locator, NodeType, ProjectStatus, ProposalOperation, ProposalRevision, Scope, SupportStatus, TypedEdge, ) from gewu.domain.validation import ( ValidationReport, assertion_digest, edge_assertion_digest, evidence_digest, prepare_proposal_digests, proposal_payload_digest, validate_proposal, ) from tests.domain.proposal_fixtures import ( BASE_COMMIT, CLAIM, EVIDENCE, EXTRACTION, NODE, POLICIES, PROJECTS, SOURCE_REVISION, proposal, ) def _validate(candidate: ProposalRevision) -> ValidationReport: return validate_proposal( candidate, projects=PROJECTS, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) def test_supported_exact_citation_and_project_policy_are_eligible() -> None: report = _validate(proposal()) assert report.status == "eligible", report.findings assert report.eligible_for_auto_accept assert set(report.affected_policy_ids) == {"dome-policy"} def test_project_only_source_registration_does_not_write_shared_knowledge() -> None: candidate = proposal( node=None, scope=Scope(projects=("dome-performance",), include_shared=False), ) report = _validate(candidate) assert report.status == "eligible", report.findings assert report.eligible_for_auto_accept assert set(report.affected_policy_ids) == {"dome-policy"} def test_proposal_digest_preparation_fills_all_dependent_claim_digests() -> None: draft = proposal().model_dump(mode="json") draft.pop("payload_digest") for operation in draft["operations"]: node = operation.get("node") if node is None: continue for group in ("claims", "assumptions", "limitations"): for claim in node.get(group, []): claim["assessment"].pop("assertion_digest") claim["assessment"].pop("evidence_digests") for evidence in claim["evidence"]: evidence.pop("assertion_path") evidence.pop("assertion_digest") prepared = prepare_proposal_digests(draft) assert prepared.payload_digest == proposal_payload_digest(prepared) assert prepared.operations[1].node is not None claim = prepared.operations[1].node.claims[0] assert claim.evidence[0].assertion_path == "claims[0].text" assert claim.evidence[0].assertion_digest == assertion_digest(claim.text) assert claim.assessment.assertion_digest == assertion_digest(claim.text) assert claim.assessment.evidence_digests == (evidence_digest(claim.evidence[0]),) assert _validate(prepared).status == "eligible" def test_transcript_timestamp_locator_must_match_the_cited_cue() -> None: transcript_bytes = ( b"WEBVTT\n\n" b"cue-1\n00:14:32.000 --> 00:14:36.500\n" b"The installation hosted 400 visitors during the 2026 season.\n" ) source = SOURCE_REVISION.model_copy( update={ "media_type": "text/vtt", "byte_length": len(transcript_bytes), "artifact_sha256": __import__("hashlib").sha256(transcript_bytes).hexdigest(), } ) extraction = TextMarkdownExtractor().extract(source.revision_id, transcript_bytes, "text/vtt") cue = next(block for block in extraction.blocks if block.time_start_ms is not None) quote = CLAIM.text start = cue.normalized_text.index(quote) evidence = EVIDENCE.model_copy( update={ "extraction_id": extraction.extraction_id, "quote": quote, "locator": Locator( kind="transcript_timestamp_span", block_id=cue.block_id, start=start, end=start + len(quote), time_start_ms=cue.time_start_ms, time_end_ms=cue.time_end_ms, ), } ) assessment = CLAIM.assessment.model_copy( update={"evidence_digests": (evidence_digest(evidence),)} ) claim = CLAIM.model_copy(update={"evidence": (evidence,), "assessment": assessment}) node = NODE.model_copy(update={"claims": (claim,)}) eligible = validate_proposal( proposal(node=node, source=source), projects=PROJECTS, policies=POLICIES, sources={}, extractions={extraction.extraction_id: extraction}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) assert eligible.status == "eligible", eligible.findings wrong_time = evidence.model_copy( update={ "locator": evidence.locator.model_copy( update={"time_start_ms": evidence.locator.time_start_ms + 1} ) } ) wrong_assessment = assessment.model_copy( update={"evidence_digests": (evidence_digest(wrong_time),)} ) wrong_claim = claim.model_copy( update={"evidence": (wrong_time,), "assessment": wrong_assessment} ) rejected = validate_proposal( proposal(node=NODE.model_copy(update={"claims": (wrong_claim,)}), source=None), projects=PROJECTS, policies=POLICIES, sources={source.revision_id: source}, extractions={extraction.extraction_id: extraction}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) assert rejected.status == "invalid" assert "EVIDENCE_LOCATION_MISMATCH" in {finding.code for finding in rejected.findings} def test_paused_or_archived_projects_reject_new_acceptance() -> None: for status in (ProjectStatus.PAUSED, ProjectStatus.ARCHIVED): project = PROJECTS["dome-performance"].model_copy(update={"status": status}) report = validate_proposal( proposal(), projects={**PROJECTS, "dome-performance": project}, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) assert not report.eligible_for_auto_accept assert any(item.code == "PROJECT_NOT_ACTIVE" for item in report.findings) def test_cross_project_source_bundle_requires_every_affected_policy() -> None: shared_source = SOURCE_REVISION.model_copy( update={"ingested_for": ("dome-performance", "interactive-video")} ) candidate = proposal( source=shared_source, scope=Scope(projects=("dome-performance", "interactive-video"), include_shared=True), ) eligible = validate_proposal( candidate, projects=PROJECTS, policies=POLICIES, sources={}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) assert eligible.status == "eligible", eligible.findings assert set(eligible.affected_policy_ids) == {"dome-policy", "video-policy"} restrictive_video_policy = POLICIES["video-policy"].model_copy( update={"automatic_source_registration": False} ) blocked = validate_proposal( candidate, projects=PROJECTS, policies={**POLICIES, "video-policy": restrictive_video_policy}, sources={}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) assert blocked.status == "needs-review" assert not blocked.eligible_for_auto_accept assert "AUTO_ACCEPT_UNSPECIFIED" in {finding.code for finding in blocked.findings} def test_matching_quote_at_wrong_location_is_rejected() -> None: evidence = EVIDENCE.model_copy( update={ "quote": "400 visitors", "locator": EVIDENCE.locator.model_copy(update={"start": 0, "end": 13}), } ) changed_claim = CLAIM.model_copy(update={"evidence": (evidence,)}) changed_node = NODE.model_copy(update={"claims": (changed_claim,)}) report = _validate(proposal(node=changed_node, source=None)) assert report.status == "invalid" assert "EVIDENCE_MISMATCH" in {finding.code for finding in report.findings} def test_reference_only_source_cannot_support_accepted_evidence() -> None: reference = SOURCE_REVISION.model_copy( update={ "artifact_sha256": None, "media_type": "application/octet-stream", "byte_length": 0, "original_url": "https://research.example/missing", "archive_status": "reference-only", } ) report = validate_proposal( proposal(source=None), projects=PROJECTS, policies=POLICIES, sources={reference.revision_id: reference}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) assert report.status == "invalid" assert "SOURCE_NOT_ARCHIVED" in {finding.code for finding in report.findings} def test_correct_quote_with_unsupported_assessment_is_rejected() -> None: assessment = CLAIM.assessment.model_copy(update={"status": SupportStatus.UNSUPPORTED}) changed_claim = CLAIM.model_copy(update={"assessment": assessment}) changed_node = NODE.model_copy(update={"claims": (changed_claim,)}) report = _validate(proposal(node=changed_node, source=None)) assert report.status == "invalid" assert "SUPPORT_UNAVAILABLE" in {finding.code for finding in report.findings} def test_node_and_assessor_attribution_cannot_impersonate_another_agent() -> None: other_assessment = CLAIM.assessment.model_copy(update={"assessor_id": "agent-other"}) assessment_claim = CLAIM.model_copy(update={"assessment": other_assessment}) assessment_node = NODE.model_copy(update={"claims": (assessment_claim,)}) assessment_report = _validate(proposal(node=assessment_node, source=None)) assert "ASSESSOR_ATTRIBUTION_MISMATCH" in { finding.code for finding in assessment_report.findings } other_node = NODE.model_copy(update={"agent_id": "agent-other"}) node_report = _validate(proposal(node=other_node, source=None)) assert "AGENT_ATTRIBUTION_MISMATCH" in {finding.code for finding in node_report.findings} def test_existing_node_can_gain_facets_without_replacing_its_assertion() -> None: from datetime import timedelta from tests.domain.proposal_fixtures import NOW existing = NODE enriched = NODE.model_copy( update={ "updated": NOW + timedelta(minutes=1), "tags": (*NODE.tags, "interactive-storytelling"), }, ) draft = ProposalRevision( proposal_id="proposal-enrich-facets", revision=1, initiating_project_id="dome-performance", scope=Scope(projects=("dome-performance",), include_shared=False), base_commit=BASE_COMMIT, payload_digest="0" * 64, proposer_id="agent-pilot", run_id="run-pilot-2", operations=( ProposalOperation(operation="update_node", target_id=NODE.node_id, node=enriched), ), ) candidate = draft.model_copy(update={"payload_digest": proposal_payload_digest(draft)}) report = validate_proposal( candidate, projects=PROJECTS, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={NODE.node_id: existing}, current_base_commit=BASE_COMMIT, ) assert report.status == "eligible", report.findings def test_existing_node_enrichment_rejects_removed_evidence() -> None: from datetime import timedelta from tests.domain.proposal_fixtures import NOW changed_claim = CLAIM.model_copy(update={"text": "The accepted count was changed."}) enriched = NODE.model_copy( update={"updated": NOW + timedelta(minutes=1), "claims": (changed_claim,)} ) draft = ProposalRevision( proposal_id="proposal-enrich-removal", revision=1, initiating_project_id="dome-performance", scope=Scope(projects=("dome-performance",), include_shared=False), base_commit=BASE_COMMIT, payload_digest="0" * 64, proposer_id="agent-pilot", run_id="run-pilot-2", operations=( ProposalOperation(operation="update_node", target_id=NODE.node_id, node=enriched), ), ) candidate = draft.model_copy(update={"payload_digest": proposal_payload_digest(draft)}) report = validate_proposal( candidate, projects=PROJECTS, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={NODE.node_id: NODE}, current_base_commit=BASE_COMMIT, ) assert report.status == "invalid" assert "NODE_ENRICHMENT_CHANGED_CLAIM" in {finding.code for finding in report.findings} def test_entity_name_collision_requires_review_without_creating_a_link() -> None: entity = NODE.model_copy( update={"node_type": NodeType.ENTITY, "title": "The Dome", "aliases": ("Planetarium",)} ) existing = NODE.model_copy( update={ "node_type": NodeType.ENTITY, "node_id": "dome:existing-entity", "title": " planetarium ", } ) report = validate_proposal( proposal( node=entity, source=None, scope=Scope(projects=("dome-performance",), include_shared=False), ), projects=PROJECTS, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={existing.node_id: existing}, current_base_commit=BASE_COMMIT, ) assert report.status == "needs-review" assert not report.eligible_for_auto_accept assert "POSSIBLE_DUPLICATE_ENTITY" in {finding.code for finding in report.findings} def test_entity_duplicate_detection_respects_project_scope() -> None: entity = NODE.model_copy( update={"node_type": NodeType.ENTITY, "title": "The Dome", "aliases": ("Planetarium",)} ) unrelated = NODE.model_copy( update={ "node_type": NodeType.ENTITY, "node_id": "video:existing-entity", "project_id": "interactive-video", "title": "Planetarium", } ) report = validate_proposal( proposal(node=entity, source=None), projects=PROJECTS, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={unrelated.node_id: unrelated}, current_base_commit=BASE_COMMIT, ) assert "POSSIBLE_DUPLICATE_ENTITY" not in {finding.code for finding in report.findings} def test_entity_duplicate_review_uses_only_readable_projects_in_explicit_scope() -> None: entity = NODE.model_copy( update={"node_type": NodeType.ENTITY, "title": "The Dome", "aliases": ("Planetarium",)} ) other_entity = NODE.model_copy( update={ "node_type": NodeType.ENTITY, "node_id": "video:existing-entity", "project_id": "interactive-video", "title": "Planetarium", } ) existing = {other_entity.node_id: other_entity} candidate = proposal( node=entity, source=None, scope=Scope(projects=("dome-performance", "interactive-video")), ) readable = validate_proposal( candidate, projects=PROJECTS, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes=existing, current_base_commit=BASE_COMMIT, readable_project_ids=frozenset({"interactive-video"}), ) unreadable = validate_proposal( candidate, projects=PROJECTS, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes=existing, current_base_commit=BASE_COMMIT, ) assert "POSSIBLE_DUPLICATE_ENTITY" in {finding.code for finding in readable.findings} assert "POSSIBLE_DUPLICATE_ENTITY" not in {finding.code for finding in unreadable.findings} def test_entity_duplicate_review_does_not_expand_beyond_proposal_scope() -> None: entity = NODE.model_copy( update={"node_type": NodeType.ENTITY, "title": "The Dome", "aliases": ("Planetarium",)} ) shared_entity = NODE.model_copy( update={ "node_type": NodeType.ENTITY, "node_id": "shared:planetarium-entity", "project_id": "shared", "title": "Planetarium", } ) report = validate_proposal( proposal( node=entity, source=None, scope=Scope(projects=("dome-performance",), include_shared=False), ), projects=PROJECTS, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={shared_entity.node_id: shared_entity}, current_base_commit=BASE_COMMIT, readable_project_ids=frozenset({"shared"}), ) assert "POSSIBLE_DUPLICATE_ENTITY" not in {finding.code for finding in report.findings} def test_requested_shared_scope_does_not_reveal_unreadable_shared_entity_names() -> None: entity = NODE.model_copy( update={"node_type": NodeType.ENTITY, "title": "The Dome", "aliases": ("Planetarium",)} ) shared_entity = NODE.model_copy( update={ "node_type": NodeType.ENTITY, "node_id": "shared:planetarium-entity", "project_id": "shared", "title": "Planetarium", } ) report = validate_proposal( proposal(node=entity, source=None), projects=PROJECTS, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={shared_entity.node_id: shared_entity}, current_base_commit=BASE_COMMIT, ) assert "POSSIBLE_DUPLICATE_ENTITY" not in {finding.code for finding in report.findings} explicitly_readable = validate_proposal( proposal(node=entity, source=None, scope=Scope(projects=("dome-performance", "shared"))), projects=PROJECTS, policies=POLICIES, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={shared_entity.node_id: shared_entity}, current_base_commit=BASE_COMMIT, readable_project_ids=frozenset({"shared"}), ) assert "POSSIBLE_DUPLICATE_ENTITY" in {finding.code for finding in explicitly_readable.findings} def test_changed_assertion_and_old_assessment_are_rejected() -> None: changed_text = "The installation hosted 401 visitors during the 2026 season." stale_assessment = CLAIM.assessment evidence = CLAIM.evidence[0].model_copy( update={"assertion_digest": assertion_digest(changed_text)} ) changed_claim = CLAIM.model_copy( update={"text": changed_text, "evidence": (evidence,), "assessment": stale_assessment} ) changed_node = NODE.model_copy(update={"claims": (changed_claim,)}) report = _validate(proposal(node=changed_node, source=None)) assert report.status == "invalid" assert "ASSESSMENT_STALE" in {finding.code for finding in report.findings} def test_exact_evidence_quote_does_not_validate_mutated_table_number() -> None: """A structurally valid quote cannot establish that a changed claim is true.""" original_text = "The installation hosted 400 visitors during the 2026 season." mutated_text = "The installation hosted 999 visitors during the 2026 season." digest = assertion_digest(original_text) evidence = EVIDENCE.model_copy( update={ "assertion_path": "claims[0].text", "assertion_digest": digest, } ) claim = CLAIM.model_copy( update={ "text": original_text, "evidence": (evidence,), "assessment": CLAIM.assessment.model_copy( update={ "assertion_digest": digest, "evidence_digests": (evidence_digest(evidence),), } ), } ) candidate = proposal( node=NODE.model_copy(update={"claims": (claim,)}), source=None, ) candidate = candidate.model_copy(update={"payload_digest": proposal_payload_digest(candidate)}) structurally_valid = _validate(candidate) mutated_claim = claim.model_copy( update={ "text": mutated_text, "evidence": ( evidence.model_copy(update={"assertion_digest": assertion_digest(mutated_text)}), ), "assessment": claim.assessment.model_copy( update={ "assertion_digest": assertion_digest(mutated_text), "evidence_digests": ( evidence_digest( evidence.model_copy( update={"assertion_digest": assertion_digest(mutated_text)} ) ), ), } ), } ) mutant = proposal(node=NODE.model_copy(update={"claims": (mutated_claim,)}), source=None) mutant = mutant.model_copy(update={"payload_digest": proposal_payload_digest(mutant)}) structurally_valid_mutant = _validate(mutant) assert structurally_valid.status == "eligible", structurally_valid.findings assert structurally_valid_mutant.status == "needs-review" assert "NUMERIC_CLAIM_UNGROUNDED" in { finding.code for finding in structurally_valid_mutant.findings } assert evidence.quote == EVIDENCE.quote assert "999" in mutated_claim.text and "999" not in evidence.quote def test_assumption_and_limitation_claims_use_the_same_evidence_contract() -> None: assumption_text = CLAIM.text assumption_digest = assertion_digest(assumption_text) assumption_evidence = EVIDENCE.model_copy( update={ "assertion_path": "assumptions[0].text", "assertion_digest": assumption_digest, } ) assumption = CLAIM.model_copy( update={ "claim_id": "dome:assumption-seasonality", "text": assumption_text, "evidence": (assumption_evidence,), "assessment": CLAIM.assessment.model_copy( update={ "assertion_digest": assumption_digest, "evidence_digests": (evidence_digest(assumption_evidence),), } ), } ) limitation_evidence = EVIDENCE.model_copy( update={ "assertion_path": "limitations[0].text", "assertion_digest": assumption_digest, } ) limitation = assumption.model_copy( update={ "claim_id": "dome:limitation-seasonality", "evidence": (limitation_evidence,), "assessment": assumption.assessment.model_copy( update={"evidence_digests": (evidence_digest(limitation_evidence),)} ), } ) node = NODE.model_copy(update={"assumptions": (assumption,), "limitations": (limitation,)}) report = _validate(proposal(node=node, source=None)) assert report.eligible_for_auto_accept, report.findings wrong_location = assumption_evidence.model_copy( update={ "quote": "400 visitors", "locator": assumption_evidence.locator.model_copy(update={"start": 0, "end": 13}), } ) invalid_assumption = assumption.model_copy( update={ "evidence": (wrong_location,), "assessment": assumption.assessment.model_copy( update={"evidence_digests": (evidence_digest(wrong_location),)} ), } ) invalid = _validate( proposal(node=NODE.model_copy(update={"assumptions": (invalid_assumption,)}), source=None) ) assert "EVIDENCE_MISMATCH" in {finding.code for finding in invalid.findings} assert any( finding.path == "operations[0].node.assumptions[0].evidence[0]" for finding in invalid.findings ) def test_missing_policy_and_explicit_scope_escape_do_not_accept() -> None: no_shared_policy = {key: value for key, value in POLICIES.items() if key != "shared"} shared_source = SOURCE_REVISION.model_copy(update={"ingested_for": ("shared",)}) candidate = proposal( node=None, source=shared_source, scope=Scope(projects=("dome-performance",), include_shared=True), ) report = validate_proposal( candidate, projects=PROJECTS, policies=no_shared_policy, sources={}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) assert report.status == "needs-review" assert "POLICY_MISSING" in {finding.code for finding in report.findings} outside_scope = proposal( node=NODE.model_copy(update={"project_id": "interactive-video"}), source=None, scope=Scope(projects=("dome-performance",)), ) report = _validate(outside_scope) assert report.status == "invalid" assert "OUT_OF_SCOPE" in {finding.code for finding in report.findings} def test_policy_cannot_waive_required_support_review() -> None: weakened_policy = POLICIES["dome-policy"].model_copy( update={"require_supported_assessment": False} ) report = validate_proposal( proposal(), projects=PROJECTS, policies={**POLICIES, "dome-policy": weakened_policy}, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) assert report.status == "invalid" assert not report.eligible_for_auto_accept assert "POLICY_WEAKENING" in {finding.code for finding in report.findings} def test_changed_base_is_a_conflict_and_edge_policy_is_explicit() -> None: report = _validate(proposal(base_commit="f" * 64)) assert report.status == "conflict" assert "BASE_CONFLICT" in {finding.code for finding in report.findings} def test_relationship_evidence_is_bound_to_the_edge_and_exact_source_span() -> None: policies = { **POLICIES, "dome-policy": POLICIES["dome-policy"].model_copy( update={"automatic_edge_types": ("causes",)} ), } edge = TypedEdge( edge_id="edge-pilot", edge_type="causes", source=EdgeEndpoint(kind="node", id=NODE.node_id), target=EdgeEndpoint(kind="node", id="dome:other-node"), evidence=(EVIDENCE,), ) bound_evidence = EVIDENCE.model_copy(update={"assertion_digest": edge_assertion_digest(edge)}) edge = edge.model_copy(update={"evidence": (bound_evidence,)}) base = proposal(source=None) candidate = base.model_copy( update={ "operations": (*base.operations, ProposalOperation(operation="add_edge", edge=edge)), "payload_digest": "0" * 64, } ) candidate = candidate.model_copy(update={"payload_digest": proposal_payload_digest(candidate)}) report = validate_proposal( candidate, projects=PROJECTS, policies=policies, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={ "dome:other-node": NODE.model_copy(update={"node_id": "dome:other-node"}), }, current_base_commit=BASE_COMMIT, ) assert report.status == "eligible", report.findings tampered = bound_evidence.model_copy( update={"assertion_digest": assertion_digest("unrelated claim")} ) invalid_edge = edge.model_copy(update={"evidence": (tampered,)}) invalid = candidate.model_copy( update={ "operations": ( *base.operations, ProposalOperation(operation="add_edge", edge=invalid_edge), ), "payload_digest": "0" * 64, } ) invalid = invalid.model_copy(update={"payload_digest": proposal_payload_digest(invalid)}) invalid_report = validate_proposal( invalid, projects=PROJECTS, policies=policies, sources={SOURCE_REVISION.revision_id: SOURCE_REVISION}, extractions={EXTRACTION.extraction_id: EXTRACTION}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) assert invalid_report.status == "invalid" assert "EVIDENCE_ASSERTION_MISMATCH" in {finding.code for finding in invalid_report.findings} def test_proposal_preparation_accepts_json_edge_evidence_arrays() -> None: draft = proposal().model_dump(mode="json") draft.pop("payload_digest") draft["operations"].append( { "operation": "add_edge", "edge": { "edge_id": "edge-preparation-test", "edge_type": "supports", "source": {"kind": "node", "id": NODE.node_id}, "target": {"kind": "node", "id": "dome:other-node"}, "evidence": [EVIDENCE.model_dump(mode="json")], }, } ) prepared = prepare_proposal_digests(draft) edge = next(item.edge for item in prepared.operations if item.edge is not None) assert edge is not None assert edge.evidence[0].assertion_path == "edge" assert edge.evidence[0].assertion_digest == edge_assertion_digest(edge) def test_contradiction_bundle_preserves_both_supported_claims() -> None: import hashlib from gewu.domain.contracts import EdgeEndpoint first_quote = "The seasonal audience count was 400." second_quote = "The seasonal audience count was 450." content = f"Field report: {first_quote}\n\nVendor report: {second_quote}\n".encode() source = SOURCE_REVISION.model_copy( update={ "artifact_sha256": hashlib.sha256(content).hexdigest(), "byte_length": len(content), } ) extraction = TextMarkdownExtractor().extract(source.revision_id, content, "text/markdown") def node_with_claim(node_id: str, claim_id: str, quote: str, title: str) -> KnowledgeNode: block = next(item for item in extraction.blocks if quote in item.normalized_text) start = block.normalized_text.index(quote) evidence = EVIDENCE.model_copy( update={ "source_revision_id": source.revision_id, "extraction_id": extraction.extraction_id, "quote": quote, "locator": Locator( kind="text_span", block_id=block.block_id, start=start, end=start + len(quote) ), "assertion_digest": assertion_digest(quote), } ) assessment = CLAIM.assessment.model_copy( update={ "assertion_digest": assertion_digest(quote), "evidence_digests": (evidence_digest(evidence),), } ) claim = CLAIM.model_copy( update={ "claim_id": claim_id, "text": quote, "evidence": (evidence,), "assessment": assessment, } ) return NODE.model_copy( update={ "node_id": node_id, "title": title, "summary": quote, "claims": (claim,), "value": 400 if "400" in quote else 450, } ) first = node_with_claim( "dome:count-field-report", "dome:claim-count-400", first_quote, "Field count" ) second = node_with_claim( "dome:count-vendor-report", "dome:claim-count-450", second_quote, "Vendor count" ) contradiction = TypedEdge( edge_id="dome:counts-conflict", edge_type="contradicts", source=EdgeEndpoint(kind="node", id=first.node_id), target=EdgeEndpoint(kind="node", id=second.node_id), conditions="Both reports refer to the same seasonal audience count.", interpretation_rationale="The source-backed counts differ for the same reporting period.", ) base = proposal(node=None, source=source) candidate = base.model_copy( update={ "operations": ( *base.operations, ProposalOperation(operation="upsert_node", node=first), ProposalOperation(operation="upsert_node", node=second), ProposalOperation(operation="add_edge", edge=contradiction), ), "payload_digest": "0" * 64, } ) candidate = candidate.model_copy(update={"payload_digest": proposal_payload_digest(candidate)}) report = validate_proposal( candidate, projects=PROJECTS, policies={ **POLICIES, "dome-policy": POLICIES["dome-policy"].model_copy( update={"automatic_edge_types": ("contradicts",)} ), }, sources={}, extractions={extraction.extraction_id: extraction}, existing_nodes={}, current_base_commit=BASE_COMMIT, ) assert report.status == "eligible", report.findings assert tuple( operation.node.node_id for operation in candidate.operations if operation.node is not None ) == (first.node_id, second.node_id) assert first.claims[0].text == first_quote assert second.claims[0].text == second_quote assert any(operation.edge == contradiction for operation in candidate.operations)