"""Unit tests for scripts/kb_rich_proposal_creation_plan.py.""" from __future__ import annotations import sys from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "scripts")) import kb_rich_proposal_creation_plan as plan # noqa: E402 def _proposal(): return { "id": "14fa5ecc-ac7a-41c1-807d-a2e85b936617", "proposal_type": "attach_evidence", "status": "approved", "proposed_by_agent_id": "11111111-1111-1111-1111-111111111111", "reviewed_by_handle": "m3ta", "review_note": "Approve A and B; keep C low confidence.", "source_ref": "telegram:m3ta:2026-07-05", "payload": { "old_claim": {"id": "d0000000-0000-0000-0000-000000000005"}, "claim_candidates": [ { "claim_key": "capital_allocation_is_civilizational_steering", "headline": "Capital allocation steers futures.", "body": "Full body with the actual caveat.", "type": "structural", "confidence": 0.65, "tags": ["capital-allocation"], "review_note": "Moderate confidence only.", "evidence": [{"source_key": "m3ta_telegram_approval"}], "edges": [ { "edge_type": "supersedes_component_of", "target_claim_id": "d0000000-0000-0000-0000-000000000005", }, {"edge_type": "requires", "target": "internet_finance_is_mechanism_not_telos"}, ], }, { "claim_key": "internet_finance_is_mechanism_not_telos", "headline": "Internet finance is a mechanism, not telos.", "body": "Boundary body.", "type": "normative/strategic-boundary", "confidence": 0.8, "evidence": [{"source_key": "m3ta_telegram_approval"}], "edges": [{"edge_type": "serves", "target": "teleo telos"}], }, ], "source_candidates": [ { "source_key": "m3ta_telegram_approval", "source_type": "dm", "title": "m3ta approval", "source_quality": "operator direction", "usage_limits": "not external evidence", "key_excerpts": ["approved with caveat"], } ], "proposed_concept_map_dependency": {"concept_key": "distributed_market_intelligence"}, }, } def test_build_creation_plan_assigns_stable_ids_and_rows(): result = plan.build_creation_plan(_proposal()) assert result["direct_production_apply_ready"] is False assert result["runtime_change"] is False assert set(result["generated_ids"]["claims"]) == { "capital_allocation_is_civilizational_steering", "internet_finance_is_mechanism_not_telos", } assert len(result["planned_rows"]["claims"]) == 2 assert len(result["planned_rows"]["sources"]) == 3 # one operator source + two body-preservation sources again = plan.build_creation_plan(_proposal()) assert result["generated_ids"] == again["generated_ids"] def test_claim_type_subtypes_normalize_to_live_claim_types(): result = plan.build_creation_plan(_proposal()) rows = {row["claim_key"]: row for row in result["planned_rows"]["claims"]} assert rows["internet_finance_is_mechanism_not_telos"]["type"] == "normative" assert rows["internet_finance_is_mechanism_not_telos"]["original_type"] == "normative/strategic-boundary" assert rows["internet_finance_is_mechanism_not_telos"]["type_normalization"]["to"] == "normative" def test_source_types_normalize_to_live_source_types(): result = plan.build_creation_plan(_proposal()) rows = {row["source_key"]: row for row in result["planned_rows"]["sources"]} assert rows["m3ta_telegram_approval"]["source_type"] == "dm" body_key = "capital_allocation_is_civilizational_steering__proposal_body" assert rows[body_key]["source_type"] == "other" assert rows[body_key]["original_source_type"] == plan.BODY_SOURCE_TYPE assert rows[body_key]["source_type_normalization"]["to"] == "other" def test_body_sources_preserve_review_notes_and_caveats(): result = plan.build_creation_plan(_proposal()) body_sources = [row for row in result["planned_rows"]["sources"] if row["original_source_type"] == plan.BODY_SOURCE_TYPE] assert len(body_sources) == 2 joined = "\n".join(row["excerpt"] for row in body_sources) assert "Approve A and B; keep C low confidence." in joined assert "Moderate confidence only." in joined assert "Full body with the actual caveat." in joined def test_supported_edges_plan_and_unsupported_edges_block(): result = plan.build_creation_plan(_proposal()) edges = result["planned_rows"]["claim_edges"] assert len(edges) == 1 assert edges[0]["edge_type"] == "requires" blocked_reasons = "\n".join(reason for item in result["blocked_fragments"] for reason in item["reasons"]) assert "unsupported edge_type 'supersedes_component_of'" in blocked_reasons assert "unsupported edge_type 'serves'" in blocked_reasons assert "target is external/context text" in blocked_reasons def test_reviewed_mapping_overlay_converts_supported_subset(): result = plan.build_creation_plan( _proposal(), { "edge_type_aliases": { "supersedes_component_of": { "to": "supersedes", "reason": "component split is represented as explicit successor-to-old-claim supersedes edges", }, "serves": "supports", }, "target_claim_mappings": { "teleo telos": { "to": "13100000-0000-0000-0000-000000000001", "reason": "test-only mapping to a UUID-shaped target", } }, }, ) edge_types = [edge["edge_type"] for edge in result["planned_rows"]["claim_edges"]] assert "supersedes" in edge_types assert "supports" in edge_types mapped_edges = [edge for edge in result["planned_rows"]["claim_edges"] if edge["edge_type_mapping"]] assert mapped_edges assert result["mapping_overlay_applied"] is True def test_evidence_rows_include_candidate_sources_and_body_sources(): result = plan.build_creation_plan(_proposal()) evidence = result["planned_rows"]["claim_evidence"] roles = {row["role"] for row in evidence} assert plan.DEFAULT_EVIDENCE_ROLE in roles assert plan.BODY_EVIDENCE_ROLE in roles assert len(evidence) == 4 def test_proposed_claims_and_reasoning_tool_node_plan(): proposal = { "id": "a64df080-8502-42e2-98f4-9bbdecb8da73", "proposal_type": "attach_evidence", "status": "approved", "proposed_by_agent_id": "11111111-1111-1111-1111-111111111111", "reviewed_by_handle": "m3taversal", "review_note": "Approved after claim #6 was removed/deferred.", "source_ref": "telegram:m3taversal:2026-06-29:7-powers-approved-with-changes", "payload": { "proposed_reasoning_tool_node": { "name": "Helmer 7 Powers", "summary": "Benefit/barrier strategy framework.", "core_test": "A Power needs both benefit and barrier.", "review_gate": "Rio should review mechanism/value-capture claims.", }, "seven_powers": [ { "name": "Scale Economies", "definition": "Unit cost declines with volume.", "examples": ["Netflix content costs amortized", "potential Teleo scale candidate"], } ], "power_lifecycle_map": {"take_off": {"stage_question": "Which powers can be established?"}}, "conceptual_neighborhood": { "theme": "competitive advantage and moats", "additional_moat_neighborhood_candidates": [ {"claim_id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", "edge_type": "relates"}, {"claim_id": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", "edge_type": "defer"}, ], }, "approval_change_log": [{"change": "Removed proposed claim #6."}], "proposed_claims": [ { "claim_key": "seven_powers_requires_benefit_and_barrier", "text": "7 Powers diagnoses durable value capture only with benefit and barrier.", "type": "structural", "confidence": 0.8, "tags": ["grand-strategy"], "evidence": ["acquired_7powers_hamilton_helmer"], "edges": [ { "edge_type": "supports", "to_existing_claim": "8f3e773e-4255-47a7-ba8f-61c57a43bebe", } ], } ], "source_candidates": [ { "source_key": "acquired_7powers_hamilton_helmer", "source_type": "podcast_transcript", "title": "7 Powers with Hamilton Helmer | Acquired", "key_excerpt": "Power has two conditions: benefit and barrier.", } ], }, } result = plan.build_creation_plan(proposal) assert result["generated_ids"]["reasoning_tools"]["Helmer 7 Powers"] assert len(result["planned_rows"]["reasoning_tools"]) == 1 assert result["planned_rows"]["reasoning_tools"][0]["category"] == "strategy_framework" description = result["planned_rows"]["reasoning_tools"][0]["description"] assert "Removed proposed claim #6" in description assert "potential Teleo scale candidate" not in description assert '"deferred_application_examples_removed": 1' in description assert '"deferred_neighborhood_candidates_removed": 1' in description assert len(result["planned_rows"]["claims"]) == 1 assert result["planned_rows"]["claims"][0]["text"].startswith("7 Powers diagnoses") rows = {row["source_key"]: row for row in result["planned_rows"]["sources"]} assert rows["acquired_7powers_hamilton_helmer"]["source_type"] == "transcript" assert len(result["planned_rows"]["claim_evidence"]) == 2 assert len(result["planned_rows"]["claim_edges"]) == 1 assert "insert into public.reasoning_tools" in result["rollback_rehearsal_sql"] def test_rollback_sql_is_not_commit_sql(): result = plan.build_creation_plan(_proposal()) sql = result["rollback_rehearsal_sql"] assert "insert into public.claims" in sql assert "insert into public.sources" in sql assert "rollback;" in sql assert "commit;" not in sql.lower()