diff --git a/src/operator_os_seam_linter.py b/src/operator_os_seam_linter.py index 1f8db99..da4b119 100644 --- a/src/operator_os_seam_linter.py +++ b/src/operator_os_seam_linter.py @@ -2,6 +2,8 @@ import argparse import json +import re +import sqlite3 from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path @@ -9,14 +11,25 @@ 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: diff --git a/src/project_registry.py b/src/project_registry.py index ea925e3..63db72d 100644 --- a/src/project_registry.py +++ b/src/project_registry.py @@ -27,6 +27,46 @@ SCHEMA_VERSION = "1.0" NOTION_PROJECTION_POLICY_SCHEMA_VERSION = "notion_projection_policy.v2" +IDENTITY_ALIAS_MAP_DEPRECATES_AFTER = "2026-09-30" + +# Migration-only alias map for Operator-OS identity dialects found by the +# 2026-07-03 dialect census. Canonical values are repo_full_name, not +# project_key; do not grow this into a permanent second identity system. +IDENTITY_ALIAS_MAP: dict[str, str] = { + "OPERANT": "saagpatel/operant", + "CryptForge": "saagpatel/CryptForge", + "Fun:GamePrjs/ CryptForge": "saagpatel/CryptForge", + "Devil-s-Advocate": "saagpatel/devils-advocate", + "devils-advocate": "saagpatel/devils-advocate", + "GithubRepoAuditor": "saagpatel/GithubRepoAuditor", + "github-repo-auditor": "saagpatel/GithubRepoAuditor", + "MCPAudit": "saagpatel/MCPAudit", + "mcpaudit-web": "saagpatel/MCPAudit", + "Notion": "saagpatel/Notion", + "notion_os": "saagpatel/Notion", + "fable-outputs": "saagpatel/fable-outputs", + "fable-second-sprint": "saagpatel/fable-second-sprint", + "portfolio-mcp": "saagpatel/portfolio-mcp", + "mcp-trust": "saagpatel/mcp-trust", + "agent-bridge": "saagpatel/agent-bridge", + "bridge-db": "saagpatel/bridge-db", + "cost-tracker": "saagpatel/cost-tracker", + "notification-hub": "saagpatel/notification-hub", + "portfolio-code-health": "saagpatel/portfolio-code-health", + "portfolio-index": "saagpatel/portfolio-index", + "AIWorkFlow": "saagpatel/AIWorkFlow", + "ApplyKit": "saagpatel/ApplyKit", + "Signal & Noise": "saagpatel/SignalAndNoise", + "Signal---Noise": "saagpatel/SignalAndNoise", +} + +BRIDGE_CANONICAL_KEY_DISAGREEMENTS: dict[str, str] = { + "operant-public": "Bridge canonical_key should reconcile to OPERANT/saagpatel/operant.", + "portfolio-health": ( + "Bridge canonical_key should reconcile to " + "portfolio-code-health/saagpatel/portfolio-code-health." + ), +} # Built-in fallbacks, mirrored by config/project-registry-overrides.json. # Hard normalization failures: drifted identifier -> canonical project_key. diff --git a/tests/test_operator_os_seam_linter.py b/tests/test_operator_os_seam_linter.py index 7991689..c11a7ee 100644 --- a/tests/test_operator_os_seam_linter.py +++ b/tests/test_operator_os_seam_linter.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import sqlite3 from datetime import UTC, datetime from pathlib import Path @@ -21,10 +22,11 @@ def _write_truth( *, generated_at: str = "2026-07-03T09:00:00+00:00", schema_version: str | None = SCHEMA_VERSION, + projects: list[dict] | None = None, ) -> None: payload = { "generated_at": generated_at, - "projects": [], + "projects": projects or [], } if schema_version is not None: payload["schema_version"] = schema_version @@ -46,6 +48,62 @@ def _passing_paths(tmp_path: Path) -> tuple[Path, list[Path]]: return truth, [registry, report] +def _write_identity_truth(path: Path) -> None: + _write_truth( + path, + projects=[ + { + "identity": { + "project_key": "Fun:GamePrjs/ CryptForge", + "display_name": "CryptForge", + "repo_full_name": "saagpatel/CryptForge", + } + }, + { + "identity": { + "project_key": "operant-public", + "display_name": "operant-public", + "repo_full_name": "saagpatel/operant", + } + }, + ], + ) + + +def _write_bridge_db( + path: Path, + *, + activity_rows: list[tuple[str, str | None]] | None = None, + session_cost_names: list[str | None] | None = None, +) -> None: + with sqlite3.connect(path) as conn: + conn.execute( + "CREATE TABLE activity_log (project_name TEXT NOT NULL, canonical_key TEXT)" + ) + conn.execute("CREATE TABLE session_costs (project_name TEXT)") + conn.executemany( + "INSERT INTO activity_log (project_name, canonical_key) VALUES (?, ?)", + activity_rows or [], + ) + conn.executemany( + "INSERT INTO session_costs (project_name) VALUES (?)", + [(name,) for name in (session_cost_names or [])], + ) + + +def _write_notification_db(path: Path, projects: list[str | None]) -> None: + with sqlite3.connect(path) as conn: + conn.execute("CREATE TABLE durable_events (project TEXT)") + conn.executemany( + "INSERT INTO durable_events (project) VALUES (?)", + [(project,) for project in projects], + ) + + +def _write_notion_snapshot(path: Path, titles: list[str]) -> None: + path.write_text(json.dumps({"projects": [{"title": title} for title in titles]})) + + def test_fresh_artifact_passes(tmp_path: Path) -> None: truth, markdown = _passing_paths(tmp_path) @@ -138,6 +196,143 @@ def test_worklist_payload_uses_existing_attention_item_shape(tmp_path: Path) -> assert payload["items"][0]["target_type"] == "artifact" +def test_identity_resolution_known_aliases_pass(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_identity_truth(truth) + bridge_db = tmp_path / "bridge.db" + notification_db = tmp_path / "notification.sqlite3" + notion_snapshot = tmp_path / "notion.json" + _write_bridge_db( + bridge_db, + activity_rows=[("CryptForge", "saagpatel/CryptForge")], + session_cost_names=["Fun-GamePrjs-CryptForge"], + ) + _write_notification_db(notification_db, ["CryptForge"]) + _write_notion_snapshot(notion_snapshot, ["OPERANT"]) + + result = lint_operator_os_seams( + truth_path=truth, + markdown_paths=markdown, + bridge_db_path=bridge_db, + notification_db_path=notification_db, + notion_snapshot_path=notion_snapshot, + now=NOW, + ) + + assert result.passed + + +def test_identity_resolution_explicit_home_adhoc_passes(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_identity_truth(truth) + bridge_db = tmp_path / "bridge.db" + _write_bridge_db(bridge_db, session_cost_names=["home-adhoc"]) + + result = lint_operator_os_seams( + truth_path=truth, + markdown_paths=markdown, + bridge_db_path=bridge_db, + now=NOW, + ) + + assert result.passed + + +def test_identity_resolution_minted_dialect_fails(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_identity_truth(truth) + notification_db = tmp_path / "notification.sqlite3" + _write_notification_db(notification_db, ["task-clear-my-portfolio-s-dependabot"]) + + result = lint_operator_os_seams( + truth_path=truth, + markdown_paths=markdown, + notification_db_path=notification_db, + now=NOW, + ) + + assert not result.passed + assert result.findings[0].check == "identity_resolution" + assert result.findings[0].violation == "minted identity dialect" + + +def test_identity_resolution_hex_fragment_fails(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_identity_truth(truth) + bridge_db = tmp_path / "bridge.db" + _write_bridge_db(bridge_db, session_cost_names=["085"]) + + result = lint_operator_os_seams( + truth_path=truth, + markdown_paths=markdown, + bridge_db_path=bridge_db, + now=NOW, + ) + + assert not result.passed + assert result.findings[0].check == "identity_resolution" + assert result.findings[0].violation == "silent unresolved identity" + + +def test_identity_resolution_bridge_canonical_key_disagreement_fails( + tmp_path: Path, +) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_identity_truth(truth) + bridge_db = tmp_path / "bridge.db" + _write_bridge_db(bridge_db, activity_rows=[("OPERANT", "operant-public")]) + + result = lint_operator_os_seams( + truth_path=truth, + markdown_paths=markdown, + bridge_db_path=bridge_db, + now=NOW, + ) + + assert not result.passed + assert result.findings[0].check == "identity_resolution" + assert result.findings[0].violation == "bridge canonical_key disagrees with alias map" + + +def test_cli_identity_resolution_is_opt_in(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + bridge_db = tmp_path / "bridge.db" + _write_bridge_db(bridge_db, session_cost_names=["085"]) + + code = main( + [ + "--truth", + str(truth), + "--markdown", + str(markdown[0]), + "--markdown", + str(markdown[1]), + "--bridge-db", + str(bridge_db), + "--json", + ] + ) + + assert code == 0 + + code = main( + [ + "--truth", + str(truth), + "--markdown", + str(markdown[0]), + "--markdown", + str(markdown[1]), + "--identity-resolution", + "--bridge-db", + str(bridge_db), + "--json", + ] + ) + + assert code == 1 + + def test_cli_exits_nonzero_and_writes_worklist_on_failure(tmp_path: Path) -> None: truth, markdown = _passing_paths(tmp_path) _write_truth(truth, schema_version="0.6.0")