Add opt-in Operator OS identity-resolution audit

src/operator_os_seam_linter.py

@@ -2,21 +2,34 @@
 
 import argparse
 import json
+import re
+import sqlite3
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
 
 from src.portfolio_truth_render import GENERATED_MARKDOWN_PROVENANCE_MARKER
 from src.portfolio_truth_types import SCHEMA_VERSION, truth_latest_path
+from src.project_registry import (
+    BRIDGE_CANONICAL_KEY_DISAGREEMENTS,
+    IDENTITY_ALIAS_MAP,
+    IDENTITY_ALIAS_MAP_DEPRECATES_AFTER,
+    normalize,
+)
 
 DEFAULT_MAX_STALENESS_HOURS = 30
 WORKLIST_SCHEMA_VERSION = "operator_os_seam_linter_worklist.v1"
+DEFAULT_BRIDGE_DB_PATH = Path("~/.local/share/bridge-db/bridge.db").expanduser()
+DEFAULT_NOTIFICATION_DB_PATH = Path(
+    "~/.local/share/notification-hub/inbox.sqlite3"
+).expanduser()
+DEFAULT_NOTION_SNAPSHOT_PATH = Path(
+    "~/.local/share/notion-os/project-snapshot.json"
+).expanduser()
 
-# v0.1: identity-resolution check - blocked on dialect census.
-IDENTITY_RESOLUTION_EXTENSION_POINT = (
-    "v0.1: identity-resolution check - blocked on dialect census"
-)
+HEX_FRAGMENT_RE = re.compile(r"^[0-9a-fA-F]{3}$")
+EXPLICIT_UNRESOLVED_IDENTITIES = {"homeadhoc", "unresolved"}
 
 
 @dataclass(frozen=True)
@@ -53,7 +66,6 @@ def to_dict(self) -> dict[str, Any]:
             "state": "pass" if self.passed else "fail",
             "expected_schema_version": self.expected_schema_version,
             "max_staleness_hours": self.max_staleness_hours,
-            "extension_points": [IDENTITY_RESOLUTION_EXTENSION_POINT],
             "findings": [finding.to_dict() for finding in self.findings],
         }
 
@@ -64,6 +76,9 @@ def lint_operator_os_seams(
     markdown_paths: list[Path],
     expected_schema_version: str = SCHEMA_VERSION,
     max_staleness_hours: int = DEFAULT_MAX_STALENESS_HOURS,
+    bridge_db_path: Path | None = None,
+    notification_db_path: Path | None = None,
+    notion_snapshot_path: Path | None = None,
     now: datetime | None = None,
 ) -> SeamLintResult:
     generated_at = _aware(now or datetime.now(UTC))
@@ -85,6 +100,14 @@ def lint_operator_os_seams(
                 expected_schema_version=expected_schema_version,
             )
         )
+        findings.extend(
+            _check_identity_resolution(
+                truth,
+                bridge_db_path=bridge_db_path,
+                notification_db_path=notification_db_path,
+                notion_snapshot_path=notion_snapshot_path,
+            )
+        )
     findings.extend(_check_generated_markdown(markdown_paths))
     return SeamLintResult(
         generated_at=generated_at,
@@ -140,6 +163,9 @@ def main(argv: list[str] | None = None) -> int:
         markdown_paths=markdown_paths,
         expected_schema_version=args.expected_schema_version,
         max_staleness_hours=args.max_staleness_hours,
+        bridge_db_path=args.bridge_db if args.identity_resolution else None,
+        notification_db_path=args.notification_db if args.identity_resolution else None,
+        notion_snapshot_path=args.notion_snapshot if args.identity_resolution else None,
     )
     payload = result.to_dict()
     if args.worklist_output:
@@ -163,6 +189,14 @@ def _build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--truth", type=Path, default=None)
     parser.add_argument("--markdown", action="append", default=[])
     parser.add_argument("--worklist-output", type=Path, default=None)
+    parser.add_argument(
+        "--identity-resolution",
+        action="store_true",
+        help="Also audit cross-store project identity dialects in local operator stores.",
+    )
+    parser.add_argument("--bridge-db", type=Path, default=DEFAULT_BRIDGE_DB_PATH)
+    parser.add_argument("--notification-db", type=Path, default=DEFAULT_NOTIFICATION_DB_PATH)
+    parser.add_argument("--notion-snapshot", type=Path, default=DEFAULT_NOTION_SNAPSHOT_PATH)
     parser.add_argument("--expected-schema-version", default=SCHEMA_VERSION)
     parser.add_argument("--max-staleness-hours", type=int, default=DEFAULT_MAX_STALENESS_HOURS)
     parser.add_argument("--json", action="store_true")
@@ -290,6 +324,278 @@ def _check_schema_pin(
     return []
 
 
+def _check_identity_resolution(
+    truth: dict[str, Any],
+    *,
+    bridge_db_path: Path | None,
+    notification_db_path: Path | None,
+    notion_snapshot_path: Path | None,
+) -> list[SeamLintFinding]:
+    resolver = _build_identity_resolver(truth)
+    findings: list[SeamLintFinding] = []
+    checked = resolved = explicit_unresolved = 0
+
+    for identity in _read_emitted_identities(
+        bridge_db_path=bridge_db_path,
+        notification_db_path=notification_db_path,
+        notion_snapshot_path=notion_snapshot_path,
+    ):
+        checked += 1
+        raw = identity["value"]
+        source = identity["source"]
+        field = identity["field"]
+        artifact = identity["artifact"]
+        normalized = _identity_norm(raw)
+
+        if normalized in EXPLICIT_UNRESOLVED_IDENTITIES:
+            explicit_unresolved += 1
+            continue
+        if _is_silent_unresolved_identity(raw):
+            findings.append(
+                _identity_finding(
+                    artifact=artifact,
+                    violation="silent unresolved identity",
+                    detail=(
+                        f"{source}.{field} emitted {raw!r}; use explicit "
+                        "home-adhoc/unresolved instead of empty, None, or hex fragments."
+                    ),
+                )
+            )
+            continue
+
+        canonical = resolver.get(normalized)
+        if canonical is None:
+            findings.append(
+                _identity_finding(
+                    artifact=artifact,
+                    violation="minted identity dialect",
+                    detail=(
+                        f"{source}.{field} emitted {raw!r}, which is not in the "
+                        f"census-seeded alias map. checked={checked}; "
+                        f"resolved={resolved}; explicit_unresolved={explicit_unresolved}; "
+                        f"deprecates_after={IDENTITY_ALIAS_MAP_DEPRECATES_AFTER}"
+                    ),
+                )
+            )
+            continue
+
+        resolved += 1
+        if source == "bridge" and field == "canonical_key":
+            disagreement = BRIDGE_CANONICAL_KEY_DISAGREEMENTS.get(str(raw))
+            if disagreement:
+                findings.append(
+                    _identity_finding(
+                        artifact=artifact,
+                        violation="bridge canonical_key disagrees with alias map",
+                        detail=(
+                            f"bridge.canonical_key emitted {raw!r}; {disagreement} "
+                            f"resolved_canonical={canonical}"
+                        ),
+                    )
+                )
+
+    if findings:
+        summary = (
+            f" identity_resolution_summary checked={checked}; resolved={resolved}; "
+            f"explicit_unresolved={explicit_unresolved}; findings={len(findings)}"
+        )
+        return [
+            SeamLintFinding(
+                check=finding.check,
+                artifact=finding.artifact,
+                violation=finding.violation,
+                detail=f"{finding.detail};{summary}",
+            )
+            for finding in findings
+        ]
+    return []
+
+
+def _identity_finding(*, artifact: str, violation: str, detail: str) -> SeamLintFinding:
+    return SeamLintFinding(
+        check="identity_resolution",
+        artifact=artifact,
+        violation=violation,
+        detail=detail,
+    )
+
+
+def _build_identity_resolver(truth: dict[str, Any]) -> dict[str, str]:
+    resolver: dict[str, str] = {}
+    for alias, canonical in IDENTITY_ALIAS_MAP.items():
+        _add_identity_alias(resolver, alias, canonical)
+    for project in truth.get("projects", []):
+        if not isinstance(project, dict):
+            continue
+        identity = project.get("identity")
+        if not isinstance(identity, dict):
+            continue
+        canonical = identity.get("repo_full_name")
+        if not isinstance(canonical, str) or "/" not in canonical:
+            continue
+        for value in (
+            identity.get("display_name"),
+            identity.get("repo_full_name"),
+            _repo_name(canonical),
+        ):
+            _add_identity_alias(resolver, value, canonical)
+        _add_identity_alias(resolver, _flatten_project_key(identity.get("project_key")), canonical)
+    return resolver
+
+
+def _add_identity_alias(
+    resolver: dict[str, str], alias: object, canonical: str | None
+) -> None:
+    if not isinstance(alias, str) or not isinstance(canonical, str) or not canonical:
+        return
+    normalized = _identity_norm(alias)
+    if normalized:
+        resolver.setdefault(normalized, canonical)
+
+
+def _read_emitted_identities(
+    *,
+    bridge_db_path: Path | None,
+    notification_db_path: Path | None,
+    notion_snapshot_path: Path | None,
+) -> list[dict[str, str]]:
+    identities: list[dict[str, str]] = []
+    identities.extend(_read_bridge_identities(bridge_db_path))
+    identities.extend(_read_notification_identities(notification_db_path))
+    identities.extend(_read_session_cost_identities(bridge_db_path))
+    identities.extend(_read_notion_title_identities(notion_snapshot_path))
+    return identities
+
+
+def _read_bridge_identities(path: Path | None) -> list[dict[str, str]]:
+    if path is None or not path.exists():
+        return []
+    rows = _sqlite_rows(
+        path,
+        "SELECT DISTINCT project_name, canonical_key FROM activity_log",
+    )
+    identities: list[dict[str, str]] = []
+    for project_name, canonical_key in rows:
+        if project_name is not None:
+            identities.append(
+                {
+                    "source": "bridge",
+                    "field": "project_name",
+                    "value": str(project_name),
+                    "artifact": str(path),
+                }
+            )
+        if canonical_key is not None:
+            identities.append(
+                {
+                    "source": "bridge",
+                    "field": "canonical_key",
+                    "value": str(canonical_key),
+                    "artifact": str(path),
+                }
+            )
+    return identities
+
+
+def _read_session_cost_identities(path: Path | None) -> list[dict[str, str]]:
+    if path is None or not path.exists():
+        return []
+    rows = _sqlite_rows(
+        path,
+        "SELECT DISTINCT project_name FROM session_costs",
+    )
+    return [
+        {
+            "source": "session_costs",
+            "field": "project_name",
+            "value": "" if row[0] is None else str(row[0]),
+            "artifact": str(path),
+        }
+        for row in rows
+    ]
+
+
+def _read_notification_identities(path: Path | None) -> list[dict[str, str]]:
+    if path is None or not path.exists():
+        return []
+    if _sqlite_table_exists(path, "notification_feed"):
+        rows = _sqlite_rows(path, "SELECT DISTINCT project FROM notification_feed")
+    elif _sqlite_table_exists(path, "durable_events"):
+        rows = _sqlite_rows(path, "SELECT DISTINCT project FROM durable_events")
+    else:
+        rows = []
+    return [
+        {
+            "source": "notification_hub",
+            "field": "project",
+            "value": "" if row[0] is None else str(row[0]),
+            "artifact": str(path),
+        }
+        for row in rows
+    ]
+
+
+def _read_notion_title_identities(path: Path | None) -> list[dict[str, str]]:
+    if path is None or not path.exists():
+        return []
+    try:
+        payload = json.loads(path.read_text())
+    except (json.JSONDecodeError, OSError):
+        return []
+    projects = payload.get("projects", []) if isinstance(payload, dict) else []
+    return [
+        {
+            "source": "notion",
+            "field": "title",
+            "value": str(project.get("title", "")),
+            "artifact": str(path),
+        }
+        for project in projects
+        if isinstance(project, dict)
+    ]
+
+
+def _sqlite_rows(path: Path, query: str) -> list[tuple[Any, ...]]:
+    try:
+        uri = f"file:{path}?mode=ro"
+        with sqlite3.connect(uri, uri=True) as conn:
+            return list(conn.execute(query).fetchall())
+    except sqlite3.Error:
+        return []
+
+
+def _sqlite_table_exists(path: Path, table: str) -> bool:
+    safe_table = table.replace("'", "''")
+    rows = _sqlite_rows(
+        path,
+        f"SELECT name FROM sqlite_master WHERE type='table' AND name='{safe_table}'",
+    )
+    return any(row[0] == table for row in rows)
+
+
+def _identity_norm(value: object) -> str:
+    if value is None:
+        return ""
+    return normalize(str(value))
+
+
+def _is_silent_unresolved_identity(value: object) -> bool:
+    if value is None:
+        return True
+    text = str(value).strip()
+    return not text or text.lower() == "none" or bool(HEX_FRAGMENT_RE.fullmatch(text))
+
+
+def _repo_name(repo_full_name: str) -> str:
+    return repo_full_name.rsplit("/", 1)[-1]
+
+
+def _flatten_project_key(value: object) -> str:
+    if not isinstance(value, str):
+        return ""
+    return re.sub(r"[^A-Za-z0-9]+", "-", value).strip("-")
+
+
 def _check_generated_markdown(markdown_paths: list[Path]) -> list[SeamLintFinding]:
     findings: list[SeamLintFinding] = []
     for path in markdown_paths: