[codex] Borrow factlog source support signals

tests/test_corroboration.py

@@ -0,0 +1,143 @@
+# SPDX-License-Identifier: MPL-2.0
+from verinote.pipeline.corroboration import (
+    corroboration,
+    functional_relations,
+    single_valued_conflicts,
+    store_corroboration,
+    store_single_valued_conflicts,
+)
+from verinote.store import Store
+
+
+def _store(tmp_path) -> Store:
+    s = Store(tmp_path / "kb.sqlite")
+    s.init_schema()
+    return s
+
+
+def test_functional_relations_parse_policy_declarations():
+    policy = r'''
+.decl functional(rel: symbol)
+functional("established_on").
+functional("quoted \" relation").
+'''
+
+    assert functional_relations(policy) == {
+        "established_on",
+        'quoted " relation',
+    }
+
+
+def test_corroboration_counts_distinct_engine_sources_only():
+    rows = [
+        {
+            "subject": "Acme",
+            "relation": "uses",
+            "object": "FastAPI",
+            "status": "confirmed",
+            "source": "sources/a.md",
+        },
+        {
+            "subject": "Acme",
+            "relation": "uses",
+            "object": "FastAPI",
+            "status": "accepted",
+            "source": "sources/b.md",
+        },
+        {
+            "subject": "Acme",
+            "relation": "uses",
+            "object": "FastAPI",
+            "status": "confirmed",
+            "source": "sources/b.md",
+        },
+        {
+            "subject": "Acme",
+            "relation": "uses",
+            "object": "FastAPI",
+            "status": "candidate",
+            "source": "sources/c.md",
+        },
+        {
+            "subject": "Acme",
+            "relation": "uses",
+            "object": "FastAPI",
+            "status": "confirmed",
+            "source": "",
+        },
+    ]
+
+    support = corroboration(rows)
+
+    assert len(support) == 1
+    assert support[0].source_count == 2
+    assert support[0].sources == ("sources/a.md", "sources/b.md")
+
+
+def test_store_corroboration_uses_joined_source_paths(tmp_path):
+    s = _store(tmp_path)
+    a = s.add_source("sources/a.md")
+    b = s.add_source("sources/b.md")
+    s.add_fact("Acme", "uses", "FastAPI", status="confirmed", source_id=a)
+    s.add_fact("Acme", "uses", "FastAPI", status="confirmed", source_id=b)
+    s.add_fact("Acme", "uses", "FastAPI", status="candidate", source_id=b)
+
+    support = store_corroboration(s)
+
+    assert [(x.subject, x.relation, x.object, x.source_count) for x in support] == [
+        ("Acme", "uses", "FastAPI", 2)
+    ]
+
+
+def test_single_valued_conflicts_include_per_value_source_support():
+    rows = [
+        {
+            "subject": "Org",
+            "relation": "established_on",
+            "object": "2020",
+            "status": "confirmed",
+            "source": "sources/a.md",
+        },
+        {
+            "subject": "Org",
+            "relation": "established_on",
+            "object": "2021",
+            "status": "accepted",
+            "source": "sources/b.md",
+        },
+        {
+            "subject": "Org",
+            "relation": "established_on",
+            "object": "2022",
+            "status": "candidate",
+            "source": "sources/c.md",
+        },
+        {
+            "subject": "Org",
+            "relation": "alias",
+            "object": "Acme",
+            "status": "confirmed",
+            "source": "sources/d.md",
+        },
+    ]
+
+    conflicts = single_valued_conflicts(rows, {"established_on"})
+
+    assert len(conflicts) == 1
+    assert conflicts[0].subject == "Org"
+    assert [(v.object, v.source_count) for v in conflicts[0].values] == [
+        ("2020", 1),
+        ("2021", 1),
+    ]
+
+
+def test_store_single_valued_conflicts_use_default_policy(tmp_path):
+    s = _store(tmp_path)
+    a = s.add_source("sources/a.md")
+    b = s.add_source("sources/b.md")
+    s.add_fact("Org", "established_on", "2020", status="confirmed", source_id=a)
+    s.add_fact("Org", "established_on", "2021", status="confirmed", source_id=b)
+
+    conflicts = store_single_valued_conflicts(s)
+
+    assert [(c.subject, c.relation) for c in conflicts] == [("Org", "established_on")]

tests/test_web.py

@@ -51,6 +51,31 @@ def test_dashboard_renders(tmp_path):
     assert "verinote" in r.text
 
 
+def test_dashboard_shows_factlog_borrowed_source_signals(tmp_path):
+    c = _client(tmp_path)
+    store = c.app.state.store
+    a = store.add_source("sources/a.md")
+    b = store.add_source("sources/b.md")
+    csrc = store.add_source("sources/c.md")
+    store.add_fact("Acme", "uses", "FastAPI", status="confirmed", source_id=a)
+    store.add_fact("Acme", "uses", "FastAPI", status="accepted", source_id=b)
+    store.add_fact("Acme", "uses", "FastAPI", status="candidate", source_id=csrc)
+    store.add_fact("Org", "established_on", "2020", status="confirmed", source_id=a)
+    store.add_fact("Org", "established_on", "2021", status="confirmed", source_id=b)
+
+    body = unescape(c.get("/").text)
+
+    assert "Source corroboration" in body
+    assert "Acme" in body
+    assert "FastAPI" in body
+    assert ">2</td>" in body
+    assert "Single-valued conflicts" in body
+    assert "Org" in body
+    assert "2020" in body
+    assert "2021" in body
+    assert "(1 source)" in body
+
+
 def test_no_active_kb_shows_selector(tmp_path, monkeypatch):
     monkeypatch.delenv("VERINOTE_ROOT", raising=False)
     monkeypatch.setenv("HOME", str(tmp_path / "home"))

verinote/pipeline/corroboration.py

@@ -0,0 +1,137 @@
+# SPDX-License-Identifier: MPL-2.0
+"""Source-support and single-valued conflict views for engine-input facts.
+
+Borrowed from factlog's deterministic trust signals: distinct source support is
+reported separately from LLM confidence, and single-valued conflicts are judged
+only over facts that have crossed the review gate.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import re
+from typing import Any, Iterable, Mapping
+
+from verinote.engine import DEFAULT_POLICY
+from verinote.store import ENGINE_STATUSES, Store
+
+_FUNCTIONAL_RE = re.compile(r'functional\("((?:\\.|[^"\\])*)"\)\.')
+
+
+@dataclass(frozen=True)
+class FactSupport:
+    subject: str
+    relation: str
+    object: str
+    sources: tuple[str, ...]
+
+    @property
+    def source_count(self) -> int:
+        return len(self.sources)
+
+
+@dataclass(frozen=True)
+class CompetingValue:
+    object: str
+    sources: tuple[str, ...]
+
+    @property
+    def source_count(self) -> int:
+        return len(self.sources)
+
+
+@dataclass(frozen=True)
+class SingleValuedConflict:
+    subject: str
+    relation: str
+    values: tuple[CompetingValue, ...]
+
+
+def functional_relations(policy_dl: str | None) -> set[str]:
+    """Parse ``functional("rel").`` declarations from a policy program."""
+    text = DEFAULT_POLICY if policy_dl is None else policy_dl
+    return {_unescape(m.group(1)) for m in _FUNCTIONAL_RE.finditer(text)}
+
+
+def store_functional_relations(store: Store) -> set[str]:
+    """Return the relation names treated as single-valued for this KB."""
+    from verinote.pipeline.verify import load_policy
+
+    return functional_relations(load_policy(store))
+
+
+def corroboration(facts: Iterable[Mapping[str, object]]) -> list[FactSupport]:
+    """Return distinct-source support for confirmed/accepted SPO triples."""
+    sources: dict[tuple[str, str, str], set[str]] = {}
+    for row in facts:
+        if str(_value(row, "status", "")) not in ENGINE_STATUSES:
+            continue
+        source = _source_ref(row)
+        if not source:
+            continue
+        key = (str(row["subject"]), str(row["relation"]), str(row["object"]))
+        sources.setdefault(key, set()).add(source)
+    return [
+        FactSupport(subject=s, relation=r, object=o, sources=tuple(sorted(srcs)))
+        for (s, r, o), srcs in sorted(sources.items())
+    ]
+
+
+def single_valued_conflicts(
+    facts: Iterable[Mapping[str, object]], single_valued: set[str]
+) -> list[SingleValuedConflict]:
+    """Return conflicting values for single-valued relations with source support."""
+    by_subject_relation: dict[tuple[str, str], dict[str, set[str]]] = {}
+    for row in facts:
+        if str(_value(row, "status", "")) not in ENGINE_STATUSES:
+            continue
+        relation = str(row["relation"])
+        if relation not in single_valued:
+            continue
+        source = _source_ref(row)
+        if not source:
+            continue
+        key = (str(row["subject"]), relation)
+        by_subject_relation.setdefault(key, {}).setdefault(
+            str(row["object"]), set()
+        ).add(source)
+
+    conflicts: list[SingleValuedConflict] = []
+    for (subject, relation), values in sorted(by_subject_relation.items()):
+        if len(values) < 2:
+            continue
+        conflicts.append(
+            SingleValuedConflict(
+                subject=subject,
+                relation=relation,
+                values=tuple(
+                    CompetingValue(object=obj, sources=tuple(sorted(srcs)))
+                    for obj, srcs in sorted(values.items())
+                ),
+            )
+        )
+    return conflicts
+
+
+def store_corroboration(store: Store) -> list[FactSupport]:
+    return corroboration(store.facts())
+
+
+def store_single_valued_conflicts(store: Store) -> list[SingleValuedConflict]:
+    return single_valued_conflicts(store.facts(), store_functional_relations(store))
+
+
+def _source_ref(row: Mapping[str, object]) -> str:
+    value = _value(row, "source_path", "") or _value(row, "source", "")
+    return str(value).strip()
+
+
+def _value(row: Mapping[str, object], key: str, default: object = None) -> Any:
+    try:
+        return row[key]
+    except (IndexError, KeyError):
+        return default
+
+
+def _unescape(value: str) -> str:
+    return re.sub(r"\\(.)", r"\1", value)

verinote/web/app.py

@@ -38,6 +38,10 @@
     verify,
     write_query_file,
 )
+from verinote.pipeline.corroboration import (
+    store_corroboration,
+    store_single_valued_conflicts,
+)
 from verinote.engine.terms import StringLit, render_term
 from verinote.store import Store
 from verinote.store.fact_input import structural_term, term_input_kind
@@ -164,6 +168,8 @@ def _dashboard(request: Request, *, error: str | None = None, status_code: int =
                 "total": sum(counts.values()),
                 "sources": store.sources(),
                 "coverage": coverage(store, root=cfg.root),
+                "corroboration": store_corroboration(store),
+                "single_valued_conflicts": store_single_valued_conflicts(store),
                 "provider": app.state.cfg.provider,
                 "provider_label": PROVIDER_LABELS.get(
                     app.state.cfg.provider, app.state.cfg.provider

verinote/web/templates/dashboard.html

@@ -44,6 +44,49 @@ <h2>Coverage <span class="muted">— engine-input facts per source</span></h2>
 <p class="muted">No sources yet.</p>
 {% endif %}
 
+<h2>Source corroboration <span class="muted">- engine-input facts only</span></h2>
+{% if corroboration %}
+<table class="counts">
+  <thead><tr><th>Fact</th><th>Sources</th></tr></thead>
+  <tbody>
+    {% for item in corroboration %}
+    <tr>
+      <td>
+        <code>{{ item.subject }}</code> / <code>{{ item.relation }}</code> /
+        <code>{{ item.object }}</code>
+      </td>
+      <td class="conf">{{ item.source_count }}</td>
+    </tr>
+    {% endfor %}
+  </tbody>
+</table>
+{% else %}
+<p class="muted">No source-backed engine-input facts yet.</p>
+{% endif %}
+
+<h2>Single-valued conflicts <span class="muted">- source support by value</span></h2>
+{% if single_valued_conflicts %}
+<table class="counts">
+  <thead><tr><th>Subject / relation</th><th>Competing values</th></tr></thead>
+  <tbody>
+    {% for conflict in single_valued_conflicts %}
+    <tr>
+      <td><code>{{ conflict.subject }}</code> / <code>{{ conflict.relation }}</code></td>
+      <td>
+        {% for value in conflict.values %}
+          <code>{{ value.object }}</code>
+          ({{ value.source_count }} source{% if value.source_count != 1 %}s{% endif %})
+          {% if not loop.last %}; {% endif %}
+        {% endfor %}
+      </td>
+    </tr>
+    {% endfor %}
+  </tbody>
+</table>
+{% else %}
+<p class="muted">No source-backed single-valued conflicts.</p>
+{% endif %}
+
 <p><a class="btn" href="/review">Go to review queue →</a>
    <a class="btn ghost" href="/sources">Manage sources →</a></p>
 {% endblock %}