test: cover functions regen-prod already ships untested

tests/test_project_bridge_notes.py

@@ -0,0 +1,236 @@
+import asyncio
+import importlib.util
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / "scripts" / "project_bridge_notes.py"
+
+
+def load_projector():
+    spec = importlib.util.spec_from_file_location("project_bridge_notes", SCRIPT)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def write_note(tmp_path, body):
+    path = tmp_path / "bridge-note.md"
+    path.write_text(
+        """---
+doc_id: test.connection.note
+doc_kind: research
+status: draft
+research_subkind: bridge_note
+disposition: implementation hypothesis
+concepts:
+  - memory-governance
+depends_on:
+  - ic.intelligence-primitives
+relates_to: []
+---
+
+"""
+        + body
+    )
+    return path
+
+
+def test_parse_v2_review_claims_and_stop_c_claim_body(tmp_path):
+    projector = load_projector()
+    path = write_note(
+        tmp_path,
+        """# Test
+
+## Claim Register
+
+**C1** [confidence: high] [anchor: section 1]
+The source claim should not absorb the following review claim.
+
+**C2** [confidence: medium] [anchor: section 2]
+The second source claim supports the same review target.
+
+- **R1**: Add a transformation-edge audit lane. [target: ic.project-learning-membrane] [concept: memory-governance] TODO: slug-deferred
+  supported_by: C1, C2.
+""",
+    )
+
+    note = projector.parse_bridge_note(path, "ic")
+    report = projector.build_parse_report(note, path.read_text())
+
+    assert [claim.c_id for claim in note.claims] == ["C1", "C2"]
+    assert "**R1**" not in note.claims[-1].statement
+    assert len(note.review_directives) == 1
+    directive = note.review_directives[0]
+    assert directive.r_id == "R1"
+    assert directive.target_doc == "ic.project-learning-membrane"
+    assert directive.concept == "memory-governance"
+    assert directive.supported_by == ["C1", "C2"]
+    assert directive.statement == "Add a transformation-edge audit lane."
+    assert report.issues == []
+
+
+def test_parse_legacy_review_claims(tmp_path):
+    projector = load_projector()
+    path = write_note(
+        tmp_path,
+        """# Test
+
+## Claim Register
+
+**C1** [confidence: high] [anchor: theorem]
+Legacy source claim.
+
+**R1** [review claim] [target: ic.intelligence-primitives] [concept: memory-governance]
+The Memory primitive should mention freshness budgets.
+*R1 is supported by C1.*
+""",
+    )
+
+    note = projector.parse_bridge_note(path, "ic")
+    report = projector.build_parse_report(note, path.read_text())
+
+    assert len(note.claims) == 1
+    assert len(note.review_directives) == 1
+    directive = note.review_directives[0]
+    assert directive.r_id == "R1"
+    assert directive.target_doc == "ic.intelligence-primitives"
+    assert directive.concept == "memory-governance"
+    assert directive.supported_by == ["C1"]
+    assert directive.statement == "The Memory primitive should mention freshness budgets."
+    assert report.issues == []
+
+
+def test_parse_report_flags_missing_concept(tmp_path):
+    projector = load_projector()
+    path = write_note(
+        tmp_path,
+        """# Test
+
+## Claim Register
+
+**C1** [confidence: high] [anchor: section]
+Source claim.
+
+- **R1**: Add an audit lane. [target: ic.project-learning-membrane]
+  supported_by: C1.
+""",
+    )
+
+    note = projector.parse_bridge_note(path, "ic")
+    report = projector.build_parse_report(note, path.read_text())
+
+    assert note.review_directives == []
+    assert any(issue.code == "missing_concept" for issue in report.issues)
+    assert any(issue.severity == "error" for issue in report.issues)
+
+
+def test_parse_report_flags_unknown_support_ref(tmp_path):
+    projector = load_projector()
+    path = write_note(
+        tmp_path,
+        """# Test
+
+## Claim Register
+
+**C1** [confidence: high] [anchor: section]
+Source claim.
+
+- **R1**: Add an audit lane. [target: ic.project-learning-membrane] [concept: memory-governance]
+  supported_by: C9.
+""",
+    )
+
+    note = projector.parse_bridge_note(path, "ic")
+    report = projector.build_parse_report(note, path.read_text())
+
+    assert len(note.review_directives) == 1
+    assert any(issue.code == "unknown_support_ref" for issue in report.issues)
+
+
+def test_parse_report_cli_does_not_require_db(tmp_path):
+    path = write_note(
+        tmp_path,
+        """# Test
+
+## Claim Register
+
+**C1** [confidence: high] [anchor: section]
+Source claim.
+
+- **R1**: Add an audit lane. [target: ic.project-learning-membrane] [concept: memory-governance]
+  supported_by: C1.
+""",
+    )
+
+    result = subprocess.run(
+        [sys.executable, str(SCRIPT), "--parse-report", "--note", str(path)],
+        cwd=ROOT,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+
+    assert result.returncode == 0
+    assert "C-claims parsed: 1" in result.stdout
+    assert "R-claims parsed: 1" in result.stdout
+    assert "Issues: none" in result.stdout
+
+
+def test_claims_service_token_prefers_env(monkeypatch, tmp_path):
+    projector = load_projector()
+    monkeypatch.setenv("KOI_CLAIMS_SERVICE_TOKEN", "env-token")
+    monkeypatch.setenv("HOME", str(tmp_path))
+
+    assert projector._claims_service_token() == "env-token"
+
+
+def test_claims_service_token_reads_state_file(monkeypatch, tmp_path):
+    projector = load_projector()
+    monkeypatch.delenv("KOI_CLAIMS_SERVICE_TOKEN", raising=False)
+    monkeypatch.setenv("HOME", str(tmp_path))
+    token_path = tmp_path / ".config/personal-koi/koi-state/claims_service_token"
+    token_path.parent.mkdir(parents=True)
+    token_path.write_text("state-token\n")
+
+    assert projector._claims_service_token() == "state-token"
+
+
+def test_projector_knows_current_intake_dispositions():
+    projector = load_projector()
+
+    assert projector.DISPOSITION_SLUG["candidate protocol"] == "propose-protocol"
+    assert projector.DISPOSITION_SLUG["novel synthesis"] == "synthesize"
+
+
+def test_create_review_claim_raises_on_auth_failure():
+    projector = load_projector()
+
+    class Response:
+        status_code = 401
+        text = "auth required"
+
+    class Client:
+        async def post(self, *args, **kwargs):
+            return Response()
+
+    async def run():
+        with pytest.raises(RuntimeError, match="/claims auth failed"):
+            await projector.create_review_claim(
+                Client(),
+                conn=object(),
+                claimant_uri="org:ic-learning-field",
+                concept_name="memory-governance",
+                about_uri="concept:memory-governance",
+                target_spec_doc="ic.memory-layers",
+                disposition_slug="implementation-hypothesis",
+                project_uri="project:intelligence-commons",
+                projection_batch="test",
+            )
+
+    asyncio.run(run())

tests/test_research_author_sensor.py

@@ -0,0 +1,150 @@
+import importlib.util
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / "scripts" / "research_author_sensor.py"
+
+
+def load_sensor():
+    spec = importlib.util.spec_from_file_location("research_author_sensor", SCRIPT)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def make_author(tmp_path):
+    sensor = load_sensor()
+    return sensor.AuthorConfig(
+        author_id="ghrist-robert",
+        canonical_name="Robert Ghrist",
+        aliases=["R. Ghrist"],
+        corpus_root=tmp_path / "Papers",
+        author_dir=tmp_path / "Papers/authors/ghrist-robert",
+        official_preprints="https://www2.math.upenn.edu/~ghrist/preprints.html",
+        arxiv_queries=['au:"Robert Ghrist"'],
+        project_tags=["sheaf-explorer", "spore"],
+        corpus_tags=["sheaf-theory", "applied-topology"],
+        direct_patterns=["sheaf", "cohomology", "discourse"],
+        project_patterns=["network", "robot", "persistence"],
+    )
+
+
+def test_parse_official_preprints(tmp_path):
+    sensor = load_sensor()
+    author = make_author(tmp_path)
+    html = """
+    <html><body><ul>
+      <li>[2026] <a href="preprints/localglobal.pdf">Neural Networks as Local-to-Global Computations</a></li>
+      <li>[2022] Network Sheaf Models for Social Information Systems</li>
+    </ul></body></html>
+    """
+
+    records = sensor.parse_official_preprints(html, author)
+
+    assert len(records) == 2
+    assert records[0].year == 2026
+    assert records[0].source_url == "https://www2.math.upenn.edu/~ghrist/preprints/localglobal.pdf"
+    assert records[1].title == "Network Sheaf Models for Social Information Systems"
+
+
+def test_parse_arxiv_xml(tmp_path):
+    sensor = load_sensor()
+    author = make_author(tmp_path)
+    xml = """<?xml version="1.0" encoding="UTF-8"?>
+    <feed xmlns="http://www.w3.org/2005/Atom">
+      <entry>
+        <id>http://arxiv.org/abs/2603.14831v3</id>
+        <updated>2026-03-20T00:00:00Z</updated>
+        <published>2026-03-18T00:00:00Z</published>
+        <title>Neural Networks as Local-to-Global Computations</title>
+        <summary>We study sheaf-style local-to-global computation.</summary>
+        <author><name>Victor Bosca</name></author>
+        <author><name>Robert Ghrist</name></author>
+      </entry>
+    </feed>
+    """
+
+    records = sensor.parse_arxiv_xml(xml, author)
+
+    assert len(records) == 1
+    record = records[0]
+    assert record.year == 2026
+    assert record.arxiv_id == "2603.14831"
+    assert record.arxiv_version == "v3"
+    assert record.pdf_url == "https://arxiv.org/pdf/2603.14831"
+    assert record.authors == ["Victor Bosca", "Robert Ghrist"]
+
+
+def test_merge_scores_and_keeps_official_source(tmp_path):
+    sensor = load_sensor()
+    author = make_author(tmp_path)
+    official = sensor.PaperRecord(
+        author_id=author.author_id,
+        canonical_author=author.canonical_name,
+        title="Neural Networks as Local-to-Global Computations",
+        year=2026,
+        source_url="https://www2.math.upenn.edu/~ghrist/preprints/localglobal.pdf",
+        source_kinds=["official_preprints"],
+    )
+    arxiv = sensor.PaperRecord(
+        author_id=author.author_id,
+        canonical_author=author.canonical_name,
+        title="Neural Networks as Local-to-Global Computations",
+        year=2026,
+        abstract="This paper uses sheaf cohomology language for neural network computation.",
+        source_url="https://arxiv.org/abs/2603.14831v3",
+        pdf_url="https://arxiv.org/pdf/2603.14831",
+        arxiv_id="2603.14831",
+        source_kinds=["arxiv"],
+    )
+
+    merged = sensor.merge_records([official, arxiv])
+    sensor.apply_record_ids(merged)
+    score, matches, decision = sensor.score_record(merged[0], author)
+
+    assert len(merged) == 1
+    assert merged[0].official_pdf_or_page == "https://www2.math.upenn.edu/~ghrist/preprints/localglobal.pdf"
+    assert merged[0].source_url == "https://arxiv.org/abs/2603.14831v3"
+    assert merged[0].paper_id == "ghrist-robert/2026-neural-networks-as-local-to-global-computations"
+    assert score >= 6
+    assert "sheaf" in matches
+    assert decision == "download_now"
+
+
+def test_existing_index_uses_manifest_and_metadata_titles(tmp_path):
+    sensor = load_sensor()
+    author = make_author(tmp_path)
+    author.author_dir.mkdir(parents=True)
+    paper_dir = author.author_dir / "2026-existing-paper"
+    paper_dir.mkdir()
+    (paper_dir / "metadata.yaml").write_text('title: "[2026] Existing Paper Title"\n')
+    manifest = author.corpus_root / "manifest.jsonl"
+    manifest.write_text(
+        '{"paper_id":"ghrist-robert/2025-manifest-paper","title":"Manifest Paper"}\n'
+    )
+    records = [
+        sensor.PaperRecord(
+            author_id=author.author_id,
+            canonical_author=author.canonical_name,
+            title="Manifest Paper",
+            year=2025,
+        ),
+        sensor.PaperRecord(
+            author_id=author.author_id,
+            canonical_author=author.canonical_name,
+            title="Existing Paper Title",
+            year=2026,
+        ),
+    ]
+    sensor.apply_record_ids(records)
+
+    ids, titles = sensor.load_existing_index(author)
+    sensor.mark_existing(records, ids, titles)
+
+    assert records[0].existing is True
+    assert records[0].existing_reason == "paper_id"
+    assert records[1].existing is True
+    assert records[1].existing_reason == "title"