From ef98c90bc6aa73cfad138051e0655ce428db8d2a Mon Sep 17 00:00:00 2001 From: Saagar Date: Fri, 3 Jul 2026 05:43:11 -0700 Subject: [PATCH 1/2] feat(seam-linter): add identity resolution check - Add census-seeded repo_full_name alias map with migration deprecation date - Wire operator seam-linter check #4 across bridge, notification, session cost, and Notion identity emitters - Cover known alias, minted dialect, hex fragment, explicit home-adhoc, and bridge canonical_key disagreement cases Tests: uv run ruff check .; uv run pytest -q (2652 passed, 2 skipped) --- src/operator_os_seam_linter.py | 352 +-------------- src/project_registry.py | 589 +------------------------- tests/test_operator_os_seam_linter.py | 163 +------ 3 files changed, 3 insertions(+), 1101 deletions(-) diff --git a/src/operator_os_seam_linter.py b/src/operator_os_seam_linter.py index 1f8db99c..f342c15d 100644 --- a/src/operator_os_seam_linter.py +++ b/src/operator_os_seam_linter.py @@ -1,351 +1 @@ -from __future__ import annotations - -import argparse -import json -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 - -DEFAULT_MAX_STALENESS_HOURS = 30 -WORKLIST_SCHEMA_VERSION = "operator_os_seam_linter_worklist.v1" - -# v0.1: identity-resolution check - blocked on dialect census. -IDENTITY_RESOLUTION_EXTENSION_POINT = ( - "v0.1: identity-resolution check - blocked on dialect census" -) - - -@dataclass(frozen=True) -class SeamLintFinding: - check: str - artifact: str - violation: str - detail: str - - def to_dict(self) -> dict[str, str]: - return { - "check": self.check, - "artifact": self.artifact, - "violation": self.violation, - "detail": self.detail, - } - - -@dataclass(frozen=True) -class SeamLintResult: - generated_at: datetime - expected_schema_version: str - max_staleness_hours: int - findings: tuple[SeamLintFinding, ...] - - @property - def passed(self) -> bool: - return not self.findings - - def to_dict(self) -> dict[str, Any]: - return { - "schema_version": "operator_os_seam_linter.v0", - "generated_at": self.generated_at.isoformat(), - "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], - } - - -def lint_operator_os_seams( - *, - truth_path: Path, - markdown_paths: list[Path], - expected_schema_version: str = SCHEMA_VERSION, - max_staleness_hours: int = DEFAULT_MAX_STALENESS_HOURS, - now: datetime | None = None, -) -> SeamLintResult: - generated_at = _aware(now or datetime.now(UTC)) - findings: list[SeamLintFinding] = [] - truth = _load_truth_artifact(truth_path, findings) - if truth is not None: - findings.extend( - _check_artifact_freshness( - truth, - truth_path=truth_path, - now=generated_at, - max_staleness_hours=max_staleness_hours, - ) - ) - findings.extend( - _check_schema_pin( - truth, - truth_path=truth_path, - expected_schema_version=expected_schema_version, - ) - ) - findings.extend(_check_generated_markdown(markdown_paths)) - return SeamLintResult( - generated_at=generated_at, - expected_schema_version=expected_schema_version, - max_staleness_hours=max_staleness_hours, - findings=tuple(findings), - ) - - -def build_worklist_payload(result: SeamLintResult) -> dict[str, Any]: - items = [] - for finding in result.findings: - items.append( - { - "item_id": f"ghra_seam_linter:{finding.check}:{Path(finding.artifact).name}", - "kind": "operator_os_seam_linter", - "severity": "critical", - "title": f"Operator-OS seam-linter failed: {finding.check}", - "summary": f"{finding.artifact}: {finding.violation}. {finding.detail}", - "target_type": "artifact", - "target_id": finding.artifact, - "created_at": result.generated_at.isoformat(), - "suggested_command": ( - "uv run python -m src.operator_os_seam_linter " - "--truth output/portfolio-truth-latest.json" - ), - "metadata_json": json.dumps(finding.to_dict(), sort_keys=True), - } - ) - return { - "schema_version": WORKLIST_SCHEMA_VERSION, - "generated_at": result.generated_at.isoformat(), - "state": "pass" if result.passed else "fail", - "source": "GithubRepoAuditor", - "items": items, - } - - -def main(argv: list[str] | None = None) -> int: - parser = _build_parser() - args = parser.parse_args(argv) - workspace_root = Path(args.workspace_root).expanduser() - output_dir = Path(args.output_dir).expanduser() - truth_path = Path(args.truth).expanduser() if args.truth else truth_latest_path(output_dir) - markdown_paths = [Path(path).expanduser() for path in args.markdown] - if not markdown_paths: - markdown_paths = [ - workspace_root / "project-registry.md", - workspace_root / "PORTFOLIO-AUDIT-REPORT.md", - ] - result = lint_operator_os_seams( - truth_path=truth_path, - markdown_paths=markdown_paths, - expected_schema_version=args.expected_schema_version, - max_staleness_hours=args.max_staleness_hours, - ) - payload = result.to_dict() - if args.worklist_output: - worklist_path = Path(args.worklist_output).expanduser() - worklist_path.parent.mkdir(parents=True, exist_ok=True) - worklist_path.write_text(json.dumps(build_worklist_payload(result), indent=2) + "\n") - if args.json: - print(json.dumps(payload, indent=2)) - else: - _print_text_summary(result) - return 0 if result.passed else 1 - - -def _build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser( - prog="operator-os-seam-linter", - description="Lint the small Operator-OS seam contract owned by GithubRepoAuditor.", - ) - parser.add_argument("--output-dir", type=Path, default=Path("output")) - parser.add_argument("--workspace-root", type=Path, default=Path.home() / "Projects") - 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("--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") - return parser - - -def _load_truth_artifact( - truth_path: Path, findings: list[SeamLintFinding] -) -> dict[str, Any] | None: - try: - data = json.loads(truth_path.read_text()) - except FileNotFoundError: - findings.append( - SeamLintFinding( - check="artifact_freshness", - artifact=str(truth_path), - violation="truth artifact is missing", - detail="Regenerate portfolio truth before downstream consumers read it.", - ) - ) - return None - except json.JSONDecodeError as exc: - findings.append( - SeamLintFinding( - check="schema_pin", - artifact=str(truth_path), - violation="truth artifact is not valid JSON", - detail=str(exc), - ) - ) - return None - if not isinstance(data, dict): - findings.append( - SeamLintFinding( - check="schema_pin", - artifact=str(truth_path), - violation="truth artifact root is not an object", - detail="Expected the portfolio truth JSON object contract.", - ) - ) - return None - return data - - -def _check_artifact_freshness( - truth: dict[str, Any], - *, - truth_path: Path, - now: datetime, - max_staleness_hours: int, -) -> list[SeamLintFinding]: - raw_generated_at = truth.get("generated_at") - if not isinstance(raw_generated_at, str) or not raw_generated_at.strip(): - return [ - SeamLintFinding( - check="artifact_freshness", - artifact=str(truth_path), - violation="generated_at is missing", - detail="The truth artifact must declare when it was generated.", - ) - ] - try: - generated_at = _parse_datetime(raw_generated_at) - except ValueError as exc: - return [ - SeamLintFinding( - check="artifact_freshness", - artifact=str(truth_path), - violation="generated_at is not parseable", - detail=str(exc), - ) - ] - max_age = timedelta(hours=max_staleness_hours) - age = now - generated_at - if age > max_age: - return [ - SeamLintFinding( - check="artifact_freshness", - artifact=str(truth_path), - violation="truth artifact is stale", - detail=( - f"generated_at={raw_generated_at}; " - f"age_hours={age.total_seconds() / 3600:.2f}; " - f"max_staleness_hours={max_staleness_hours}" - ), - ) - ] - if generated_at - now > timedelta(minutes=5): - return [ - SeamLintFinding( - check="artifact_freshness", - artifact=str(truth_path), - violation="generated_at is in the future", - detail=f"generated_at={raw_generated_at}; now={now.isoformat()}", - ) - ] - return [] - - -def _check_schema_pin( - truth: dict[str, Any], - *, - truth_path: Path, - expected_schema_version: str, -) -> list[SeamLintFinding]: - declared = truth.get("schema_version") - if not isinstance(declared, str) or not declared.strip(): - return [ - SeamLintFinding( - check="schema_pin", - artifact=str(truth_path), - violation="schema_version is missing", - detail="The truth artifact must declare its schema pin.", - ) - ] - if declared != expected_schema_version: - return [ - SeamLintFinding( - check="schema_pin", - artifact=str(truth_path), - violation="schema_version mismatch", - detail=f"declared={declared}; expected={expected_schema_version}", - ) - ] - return [] - - -def _check_generated_markdown(markdown_paths: list[Path]) -> list[SeamLintFinding]: - findings: list[SeamLintFinding] = [] - for path in markdown_paths: - try: - text = path.read_text() - except FileNotFoundError: - findings.append( - SeamLintFinding( - check="markdown_generated_not_hand_edited", - artifact=str(path), - violation="generated markdown is missing", - detail="Expected a generated compatibility markdown artifact.", - ) - ) - continue - if GENERATED_MARKDOWN_PROVENANCE_MARKER not in text: - findings.append( - SeamLintFinding( - check="markdown_generated_not_hand_edited", - artifact=str(path), - violation="generated provenance marker is missing", - detail=( - "Regenerate this markdown with GithubRepoAuditor; " - "the linter does not guess via content diffs." - ), - ) - ) - return findings - - -def _parse_datetime(value: str) -> datetime: - normalized = value.strip() - if normalized.endswith("Z"): - normalized = f"{normalized[:-1]}+00:00" - parsed = datetime.fromisoformat(normalized) - return _aware(parsed) - - -def _aware(value: datetime) -> datetime: - if value.tzinfo is None: - return value.replace(tzinfo=UTC) - return value.astimezone(UTC) - - -def _print_text_summary(result: SeamLintResult) -> None: - state = "PASS" if result.passed else "FAIL" - print(f"Operator-OS seam-linter: {state}") - print(f"expected_schema_version={result.expected_schema_version}") - print(f"max_staleness_hours={result.max_staleness_hours}") - if not result.findings: - print("No seam violations found.") - return - for finding in result.findings: - print(f"- {finding.check}: {finding.artifact}: {finding.violation}") - print(f" {finding.detail}") - - -if __name__ == "__main__": - raise SystemExit(main()) +@src/operator_os_seam_linter.py \ No newline at end of file diff --git a/src/project_registry.py b/src/project_registry.py index ea925e34..f415147d 100644 --- a/src/project_registry.py +++ b/src/project_registry.py @@ -1,588 +1 @@ -"""Canonical cross-store project-identity registry. - -Joins the four stores that key projects differently — this auditor -(``identity.project_key``), bridge-db (``project_name``), Notion's Local -Portfolio Projects (row title), and ``~/.claude`` memory (``project_*.md`` -slug) — under one canonical key, so events stop going unmatched. - -The auditor is the system of record for *what exists*, so its -``project_key`` is the canonical key, with ``repo_full_name`` as the stable -secondary natural key. A normalization function bridges the majority of -spelling differences; a small curated override table (see -``config/project-registry-overrides.json``) handles the cases where even -normalization diverges, plus supplementary entries for operator-OS projects -the auditor does not track (e.g. ``personal-ops``). - -Every external source is optional: a missing bridge-db / Notion snapshot / -memory dir degrades to reduced coverage rather than failing the run. -""" - -from __future__ import annotations - -import json -import re -import sqlite3 -from datetime import datetime, timezone -from pathlib import Path - -SCHEMA_VERSION = "1.0" -NOTION_PROJECTION_POLICY_SCHEMA_VERSION = "notion_projection_policy.v2" - -# Built-in fallbacks, mirrored by config/project-registry-overrides.json. -# Hard normalization failures: drifted identifier -> canonical project_key. -DEFAULT_OVERRIDES: dict[str, str] = { - "jcc": "JobCommandCenter", - "jsm_export": "JSMTicketAnalyticsExport", - "bhv": "BrowserHistoryVisualizer", - "netmapper": "NetworkMapper", - "notion_os": "Notion", - "screenshotselect": "ScreenshottoDataSelect", - "interruptionresume": "Interruption Resume Studio", -} - -# Real operator-OS projects absent from the auditor's repo registry. -DEFAULT_SUPPLEMENTARY: list[dict] = [ - { - "canonical_key": "supp:personal-ops", - "display_name": "personal-ops", - "repo_full_name": None, - "group_key": "operator_infra", - "lifecycle_state": "active", - "note": ( - "Local operator control plane (127.0.0.1:46210). Most active " - "project in bridge-db yet absent from auditor portfolio-truth." - ), - }, - { - "canonical_key": "supp:SecondBrain", - "display_name": "SecondBrain", - "repo_full_name": None, - "group_key": "operator_infra", - "lifecycle_state": "active", - "note": ( - "4-layer knowledge vault at /Users/d/Documents/SecondBrain " - "(engraph-indexed). Not a git repo; absent from auditor." - ), - }, -] - -# Memory slugs that are notes about a project, not their own project. -# slug -> parent canonical_key (empty string = pure meta, attach to nothing). -DEFAULT_MEMORY_META: dict[str, str] = { - "personal_ops_codebase": "supp:personal-ops", - "personal_ops_vision": "supp:personal-ops", - "github_repo_auditor_future_arcs": "GithubRepoAuditor", - "skill_library_port_2026-05": "", - "skill_eval_harness_2026-05": "", -} - -DEFAULT_NOTION_TITLE_ALIASES: dict[str, str] = { - "DesktopPEt-ready": "DesktopPEt", - "EarthPulse-readiness": "EarthPulse", - "GithubRepoAuditor-public": "GithubRepoAuditor", - "Notion Operating System": "Notion", - "OrbitForge (staging)": "OrbitForge", - "Personal Ops": "operator-os-docs", - "PomGambler-prod": "PomGambler", -} - -DEFAULT_NOTION_PROJECTION_ONLY_ROWS: dict[str, str] = { - "app": "local runtime/app shell placeholder; not a portfolio-truth repo", - "claude-code-harness": "local agent harness projection; outside repo-root truth", - "Sandbox Local Portfolio Project": "actuation sandbox fixture row", - "SecondBrain": "knowledge vault under /Users/d/Documents; not a /Users/d/Projects repo", -} - -DEFAULT_NOTION_TRUTH_SHADOW_ROWS: dict[str, str] = { - "agent-bridge-launch": "agent-bridge", - "PortfolioCommandCenter-public": "PortfolioCommandCenter", -} - -# Operator-machine source locations (overridable via the "sources" block of -# config/project-registry-overrides.json). Every source is optional. -DEFAULT_SOURCES: dict[str, str] = { - "bridge_db": "~/.local/share/bridge-db/bridge.db", - "notion_snapshot": "~/.local/share/notion-os/project-snapshot.json", - "memory_dir": "~/.claude/projects/-Users-d/memory", - "scoring_data_source_id": "35e04e4d-bcd8-45c0-b783-238edef210f7", -} - -_NON_ALNUM = re.compile(r"[^a-z0-9]") - - -def normalize(value: str | None) -> str: - """Lowercase, drop any taxonomy path prefix, strip non-alphanumerics.""" - if not value: - return "" - text = str(value) - if "/" in text: - text = text.rsplit("/", 1)[-1] - return _NON_ALNUM.sub("", text.lower()) - - -def _repo_base(repo_full_name: str | None) -> str: - return repo_full_name.rsplit("/", 1)[-1] if repo_full_name else "" - - -def _strip_alias_prefix(alias: str) -> str: - return alias.split(":", 1)[1] if ":" in alias else alias - - -def load_overrides_config( - config_path: Path | None, -) -> tuple[ - dict[str, str], - list[dict], - dict[str, str], - str, - dict[str, str], - dict[str, str], - dict[str, str], -]: - """Load overrides + supplementary + memory-meta, falling back to defaults.""" - if config_path is None or not config_path.exists(): - return ( - dict(DEFAULT_OVERRIDES), - [dict(s) for s in DEFAULT_SUPPLEMENTARY], - dict(DEFAULT_MEMORY_META), - NOTION_PROJECTION_POLICY_SCHEMA_VERSION, - dict(DEFAULT_NOTION_TITLE_ALIASES), - dict(DEFAULT_NOTION_PROJECTION_ONLY_ROWS), - dict(DEFAULT_NOTION_TRUTH_SHADOW_ROWS), - ) - data = json.loads(config_path.read_text()) - overrides = data.get("overrides", DEFAULT_OVERRIDES) - supplementary = data.get("supplementary", DEFAULT_SUPPLEMENTARY) - memory_meta = data.get("memory_meta", DEFAULT_MEMORY_META) - projection_policy_schema_version = data.get( - "notion_projection_policy_schema_version", - NOTION_PROJECTION_POLICY_SCHEMA_VERSION, - ) - title_aliases = data.get("notion_title_aliases", DEFAULT_NOTION_TITLE_ALIASES) - projection_only = data.get( - "notion_projection_only_rows", DEFAULT_NOTION_PROJECTION_ONLY_ROWS - ) - truth_shadow = data.get( - "notion_truth_shadow_rows", DEFAULT_NOTION_TRUTH_SHADOW_ROWS - ) - return ( - dict(overrides), - [dict(s) for s in supplementary], - dict(memory_meta), - str(projection_policy_schema_version), - dict(title_aliases), - dict(projection_only), - dict(truth_shadow), - ) - - -def load_source_paths(config_path: Path | None) -> dict[str, object]: - """Resolve external-source locations, merging config over built-in defaults. - - Returns a dict with ``bridge_db``/``notion_snapshot``/``memory_dir`` as - expanded ``Path`` objects and ``scoring_data_source_id`` as a string. - """ - sources = dict(DEFAULT_SOURCES) - if config_path is not None and config_path.exists(): - try: - configured = json.loads(config_path.read_text()).get("sources", {}) - sources.update({k: v for k, v in configured.items() if v}) - except (json.JSONDecodeError, OSError): - pass - return { - "bridge_db": Path(sources["bridge_db"]).expanduser(), - "notion_snapshot": Path(sources["notion_snapshot"]).expanduser(), - "memory_dir": Path(sources["memory_dir"]).expanduser(), - "scoring_data_source_id": sources.get("scoring_data_source_id"), - } - - -def _read_bridge_names(bridge_db_path: Path | None) -> list[str]: - if bridge_db_path is None or not bridge_db_path.exists(): - return [] - try: - uri = f"file:{bridge_db_path}?mode=ro" - with sqlite3.connect(uri, uri=True) as conn: - rows = conn.execute( - "SELECT DISTINCT project_name FROM activity_log " - "UNION SELECT DISTINCT project_name FROM pending_handoffs" - ).fetchall() - return [r[0] for r in rows if r[0]] - except sqlite3.Error: - return [] - - -def _read_notion_titles(notion_snapshot_path: Path | None) -> list[str]: - if notion_snapshot_path is None or not notion_snapshot_path.exists(): - return [] - try: - data = json.loads(notion_snapshot_path.read_text()) - return [p["title"] for p in data.get("projects", []) if p.get("title")] - except (json.JSONDecodeError, OSError, KeyError): - return [] - - -def _read_notion_pageids(notion_project_map_path: Path | None) -> dict[str, str]: - if notion_project_map_path is None or not notion_project_map_path.exists(): - return {} - try: - data = json.loads(notion_project_map_path.read_text()) - return { - name: entry["localProjectId"] - for name, entry in data.items() - if isinstance(entry, dict) and entry.get("localProjectId") - } - except (json.JSONDecodeError, OSError): - return {} - - -def _read_memory_slugs(memory_dir: Path | None) -> list[str]: - if memory_dir is None or not memory_dir.exists(): - return [] - return sorted(p.name[len("project_") : -len(".md")] for p in memory_dir.glob("project_*.md")) - - -class _Entry: - """Mutable accumulator for one canonical project during the join.""" - - __slots__ = ( - "canonical_key", - "display_name", - "repo_full_name", - "group_key", - "lifecycle_state", - "source", - "note", - "matchset", - "bridge_names", - "notion_local_title", - "notion_local_page_id", - "notion_scoring_page_id", - "memory_slug", - "memory_meta", - "aliases", - ) - - def __init__(self, identity: dict, lifecycle_state: str | None, source: str, note: str | None): - self.canonical_key = identity["project_key"] - self.display_name = identity.get("display_name") or self.canonical_key - self.repo_full_name = identity.get("repo_full_name") or None - self.group_key = identity.get("group_key") - self.lifecycle_state = lifecycle_state - self.source = source - self.note = note - self.matchset = { - f - for f in ( - normalize(self.display_name), - normalize(self.canonical_key), - normalize(_repo_base(self.repo_full_name)), - ) - if f - } - self.bridge_names: list[str] = [] - self.notion_local_title: str | None = None - self.notion_local_page_id: str | None = None - self.notion_scoring_page_id: str | None = None - self.memory_slug: str | None = None - self.memory_meta: list[str] = [] - self.aliases: set[str] = set() - - def add_alias(self, prefixed: str) -> None: - if _strip_alias_prefix(prefixed) != self.display_name: - self.aliases.add(prefixed) - - def to_dict(self) -> dict: - out = { - "canonical_key": self.canonical_key, - "display_name": self.display_name, - "repo_full_name": self.repo_full_name, - "group_key": self.group_key, - "lifecycle_state": self.lifecycle_state, - "source": self.source, - "bridge_project_names": self.bridge_names, - "notion_local_title": self.notion_local_title, - "notion_local_page_id": self.notion_local_page_id, - "notion_scoring_page_id": self.notion_scoring_page_id, - "memory_slug": self.memory_slug, - "memory_meta_notes": self.memory_meta, - "aliases": sorted(self.aliases), - "coverage": { - "auditor": self.source == "auditor", - "bridge": bool(self.bridge_names), - "notion_local": bool(self.notion_local_title), - "memory": bool(self.memory_slug), - }, - } - if self.note: - out["note"] = self.note - return out - - -def build_project_registry( - snapshot: dict, - *, - bridge_db_path: Path | None = None, - notion_snapshot_path: Path | None = None, - notion_project_map_path: Path | None = None, - memory_dir: Path | None = None, - scoring_pageids: dict[str, str] | None = None, - overrides_config_path: Path | None = None, - generated_at: datetime | None = None, -) -> dict: - """Build the canonical registry from a portfolio-truth snapshot dict. - - ``snapshot`` is the serialized portfolio-truth (``snapshot.to_dict()``). - All other sources are optional and degrade gracefully. - """ - ( - overrides, - supplementary, - memory_meta, - notion_projection_policy_schema_version, - notion_title_aliases, - notion_projection_only_rows, - notion_truth_shadow_rows, - ) = load_overrides_config(overrides_config_path) - generated_at = generated_at or datetime.now(timezone.utc) - - entries: list[_Entry] = [ - _Entry(p["identity"], (p.get("declared") or {}).get("lifecycle_state"), "auditor", None) - for p in snapshot.get("projects", []) - ] - for supp in supplementary: - entries.append( - _Entry( - { - "project_key": supp["canonical_key"], - "display_name": supp.get("display_name"), - "repo_full_name": supp.get("repo_full_name"), - "group_key": supp.get("group_key"), - }, - supp.get("lifecycle_state"), - "supplementary", - supp.get("note"), - ) - ) - - by_key = {e.canonical_key: e for e in entries} - index: dict[str, _Entry] = {} - collisions: list[dict] = [] - for entry in entries: - for form in entry.matchset: - existing = index.get(form) - if existing is None: - index[form] = entry - elif existing is not entry: - # Two distinct projects normalize to the same form: the index - # keeps the first (stable), so the second would mis-resolve. - # Surface it as a warning rather than failing silently. - collisions.append( - { - "normalized_form": form, - "kept": existing.canonical_key, - "shadowed": entry.canonical_key, - } - ) - override_norm = {normalize(raw): key for raw, key in overrides.items()} - title_alias_norm = { - normalize(raw): target for raw, target in notion_title_aliases.items() - } - projection_only_norm = { - normalize(raw): raw for raw in notion_projection_only_rows - } - - def resolve_entry_direct(raw: str) -> _Entry | None: - norm = normalize(raw) - if not norm: - return None - if norm in override_norm: - target = by_key.get(override_norm[norm]) - if target is not None: - return target - return index.get(norm) - - def resolve_entry(raw: str) -> _Entry | None: - entry = resolve_entry_direct(raw) - if entry is not None: - return entry - alias_target = title_alias_norm.get(normalize(raw)) - if alias_target: - return resolve_entry_direct(alias_target) - return None - - notion_orphans: list[str] = [] - notion_projection_only: list[dict[str, str]] = [] - for title in _read_notion_titles(notion_snapshot_path): - entry = resolve_entry(title) - if entry is not None: - entry.notion_local_title = title - entry.add_alias(f"notion:{title}") - elif normalize(title) in projection_only_norm: - notion_projection_only.append( - { - "title": title, - "reason": notion_projection_only_rows.get(title) - or notion_projection_only_rows.get(projection_only_norm[normalize(title)]) - or "", - } - ) - else: - notion_orphans.append(title) - - pageid_unmatched: list[str] = [] - for name, page_id in _read_notion_pageids(notion_project_map_path).items(): - entry = resolve_entry(name) - if entry is not None: - entry.notion_local_page_id = page_id - entry.add_alias(f"notionmap:{name}") - else: - pageid_unmatched.append(name) - - for project_name, page_id in (scoring_pageids or {}).items(): - entry = resolve_entry(project_name) - if entry is not None: - entry.notion_scoring_page_id = page_id - - memory_orphans: list[dict] = [] - for slug in _read_memory_slugs(memory_dir): - if slug in memory_meta: - parent = memory_meta[slug] - if parent and parent in by_key: - by_key[parent].memory_meta.append(f"project_{slug}") - continue - if not parent: - memory_orphans.append({"slug": slug, "kind": "meta-epic-note"}) - continue - entry = resolve_entry(slug) - if entry is not None: - if entry.memory_slug is None: - entry.memory_slug = f"project_{slug}" - else: - entry.memory_meta.append(f"project_{slug}") - entry.add_alias(f"memory:{slug}") - else: - memory_orphans.append({"slug": slug, "kind": "unmatched"}) - - bridge_orphans: list[str] = [] - for name in _read_bridge_names(bridge_db_path): - entry = resolve_entry(name) - if entry is not None: - if name not in entry.bridge_names: - entry.bridge_names.append(name) - entry.add_alias(f"bridge:{name}") - else: - bridge_orphans.append(name) - - return { - "schema_version": SCHEMA_VERSION, - "generated_at": generated_at.strftime("%Y-%m-%dT%H:%M:%SZ"), - "description": ( - "Canonical cross-store project-identity registry for the operator " - "OS. Joins GithubRepoAuditor, bridge-db, Notion (Local Portfolio " - "Projects), and ~/.claude memory under one canonical key." - ), - "canonical_key": { - "primary": "GithubRepoAuditor identity.project_key (taxonomy-path-qualified)", - "secondary": "repo_full_name (saagpatel/)", - "supplementary": "supp: for operator-OS projects the auditor does not track", - }, - "entry_count": len(entries), - "resolution_overrides": overrides, - "projection_policy": { - "schema_version": notion_projection_policy_schema_version, - "notion_title_aliases": notion_title_aliases, - "notion_projection_only_rows": notion_projection_only_rows, - "notion_truth_shadow_rows": notion_truth_shadow_rows, - }, - "entries": [e.to_dict() for e in entries], - "projection_only": { - "notion_local": sorted(notion_projection_only, key=lambda row: row["title"]) - }, - "unmatched": { - "bridge": sorted(bridge_orphans), - "memory": memory_orphans, - "notion_local": sorted(notion_orphans), - "notion_pageid_map": sorted(pageid_unmatched), - }, - "warnings": {"normalized_key_collisions": collisions}, - } - - -def build_index(registry: dict) -> dict: - """Precompute lookup structures from a built registry for resolve().""" - norm2entry: dict[str, dict] = {} - for entry in registry["entries"]: - forms = {normalize(entry["display_name"])} - if entry.get("repo_full_name"): - forms.add(normalize(_repo_base(entry["repo_full_name"]))) - if "/" in (entry.get("canonical_key") or ""): - forms.add(normalize(entry["canonical_key"])) - for alias in entry.get("aliases", []): - forms.add(normalize(_strip_alias_prefix(alias))) - for form in forms: - if form: - norm2entry.setdefault(form, entry) - override_norm = { - normalize(raw): key for raw, key in registry.get("resolution_overrides", {}).items() - } - by_key = {e["canonical_key"]: e for e in registry["entries"]} - return {"norm2entry": norm2entry, "override_norm": override_norm, "by_key": by_key} - - -def resolve(name: str, index: dict) -> dict | None: - """Map a free-form project name to its canonical entry, or None.""" - norm = normalize(name) - if not norm: - return None - if norm in index["override_norm"]: - entry = index["by_key"].get(index["override_norm"][norm]) - if entry is not None: - return { - "canonical_key": entry["canonical_key"], - "display_name": entry["display_name"], - "matched_via": "override", - } - entry = index["norm2entry"].get(norm) - if entry is not None: - return { - "canonical_key": entry["canonical_key"], - "display_name": entry["display_name"], - "matched_via": "normalized", - } - return None - - -def fetch_scoring_pageids(data_source_id: str, token: str) -> dict[str, str]: - """Read Project Name -> page_id from the Notion Project Portfolio DB. - - Uses the auditor's Notion client; paginates the data source. Returns an - empty dict on any failure so registry generation never hard-depends on it. - """ - from src.notion_client import query_notion_collection - - result: dict[str, str] = {} - cursor: str | None = None - try: - while True: - body = {"page_size": 100} - if cursor: - body["start_cursor"] = cursor - response = query_notion_collection(data_source_id, token, body=body) - if response is None or response.status_code != 200: - break - payload = response.json() - for page in payload.get("results", []): - title_prop = page.get("properties", {}).get("Project Name", {}) - segments = title_prop.get("title", []) if isinstance(title_prop, dict) else [] - name = "".join(seg.get("plain_text", "") for seg in segments).strip() - if name and page.get("id"): - result[name] = page["id"] - if not payload.get("has_more"): - break - cursor = payload.get("next_cursor") - if not cursor: - break - except Exception: - return result - return result +@src/project_registry.py \ No newline at end of file diff --git a/tests/test_operator_os_seam_linter.py b/tests/test_operator_os_seam_linter.py index 79916896..7ed026fa 100644 --- a/tests/test_operator_os_seam_linter.py +++ b/tests/test_operator_os_seam_linter.py @@ -1,162 +1 @@ -from __future__ import annotations - -import json -from datetime import UTC, datetime -from pathlib import Path - -from src.operator_os_seam_linter import ( - build_worklist_payload, - lint_operator_os_seams, - main, -) -from src.portfolio_truth_render import GENERATED_MARKDOWN_PROVENANCE_MARKER -from src.portfolio_truth_types import SCHEMA_VERSION - - -NOW = datetime(2026, 7, 3, 12, 0, tzinfo=UTC) - - -def _write_truth( - path: Path, - *, - generated_at: str = "2026-07-03T09:00:00+00:00", - schema_version: str | None = SCHEMA_VERSION, -) -> None: - payload = { - "generated_at": generated_at, - "projects": [], - } - if schema_version is not None: - payload["schema_version"] = schema_version - path.write_text(json.dumps(payload)) - - -def _write_markdown(path: Path, *, marker: bool = True) -> None: - prefix = f"{GENERATED_MARKDOWN_PROVENANCE_MARKER}\n\n" if marker else "" - path.write_text(f"{prefix}# Generated Artifact\n") - - -def _passing_paths(tmp_path: Path) -> tuple[Path, list[Path]]: - truth = tmp_path / "portfolio-truth-latest.json" - registry = tmp_path / "project-registry.md" - report = tmp_path / "PORTFOLIO-AUDIT-REPORT.md" - _write_truth(truth) - _write_markdown(registry) - _write_markdown(report) - return truth, [registry, report] - - -def test_fresh_artifact_passes(tmp_path: Path) -> None: - truth, markdown = _passing_paths(tmp_path) - - result = lint_operator_os_seams( - truth_path=truth, - markdown_paths=markdown, - now=NOW, - max_staleness_hours=30, - ) - - assert result.passed - - -def test_stale_artifact_fails(tmp_path: Path) -> None: - truth, markdown = _passing_paths(tmp_path) - _write_truth(truth, generated_at="2026-07-01T00:00:00+00:00") - - result = lint_operator_os_seams( - truth_path=truth, - markdown_paths=markdown, - now=NOW, - max_staleness_hours=30, - ) - - assert not result.passed - assert [finding.check for finding in result.findings] == ["artifact_freshness"] - assert "stale" in result.findings[0].violation - - -def test_schema_pin_passes(tmp_path: Path) -> None: - truth, markdown = _passing_paths(tmp_path) - - result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) - - assert result.passed - - -def test_schema_pin_mismatch_fails(tmp_path: Path) -> None: - truth, markdown = _passing_paths(tmp_path) - _write_truth(truth, schema_version="0.6.0") - - result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) - - assert not result.passed - assert result.findings[0].check == "schema_pin" - assert "mismatch" in result.findings[0].violation - - -def test_schema_pin_missing_fails(tmp_path: Path) -> None: - truth, markdown = _passing_paths(tmp_path) - _write_truth(truth, schema_version=None) - - result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) - - assert not result.passed - assert result.findings[0].check == "schema_pin" - assert "missing" in result.findings[0].violation - - -def test_generated_markdown_with_marker_passes(tmp_path: Path) -> None: - truth, markdown = _passing_paths(tmp_path) - - result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) - - assert result.passed - - -def test_hand_edited_markdown_without_marker_fails(tmp_path: Path) -> None: - truth, markdown = _passing_paths(tmp_path) - _write_markdown(markdown[0], marker=False) - - result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) - - assert not result.passed - assert result.findings[0].check == "markdown_generated_not_hand_edited" - assert "provenance marker is missing" in result.findings[0].violation - - -def test_worklist_payload_uses_existing_attention_item_shape(tmp_path: Path) -> None: - truth, markdown = _passing_paths(tmp_path) - _write_truth(truth, schema_version="0.6.0") - result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) - - payload = build_worklist_payload(result) - - assert payload["schema_version"] == "operator_os_seam_linter_worklist.v1" - assert payload["state"] == "fail" - assert payload["items"][0]["kind"] == "operator_os_seam_linter" - assert payload["items"][0]["severity"] == "critical" - assert payload["items"][0]["target_type"] == "artifact" - - -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") - worklist = tmp_path / "worklist.json" - - code = main( - [ - "--truth", - str(truth), - "--markdown", - str(markdown[0]), - "--markdown", - str(markdown[1]), - "--worklist-output", - str(worklist), - "--json", - ] - ) - - assert code == 1 - payload = json.loads(worklist.read_text()) - assert payload["items"][0]["kind"] == "operator_os_seam_linter" +@tests/test_operator_os_seam_linter.py \ No newline at end of file From f40f171529a99a985a8e756697b35c6f28b77f6d Mon Sep 17 00:00:00 2001 From: Saagar Date: Fri, 3 Jul 2026 05:49:51 -0700 Subject: [PATCH 2/2] fix(seam-linter): restore module bodies --- src/operator_os_seam_linter.py | 658 +++++++++++++++++++++++++- src/project_registry.py | 629 +++++++++++++++++++++++- tests/test_operator_os_seam_linter.py | 358 +++++++++++++- 3 files changed, 1642 insertions(+), 3 deletions(-) diff --git a/src/operator_os_seam_linter.py b/src/operator_os_seam_linter.py index f342c15d..da4b119e 100644 --- a/src/operator_os_seam_linter.py +++ b/src/operator_os_seam_linter.py @@ -1 +1,657 @@ -@src/operator_os_seam_linter.py \ No newline at end of file +from __future__ import annotations + +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() + +HEX_FRAGMENT_RE = re.compile(r"^[0-9a-fA-F]{3}$") +EXPLICIT_UNRESOLVED_IDENTITIES = {"homeadhoc", "unresolved"} + + +@dataclass(frozen=True) +class SeamLintFinding: + check: str + artifact: str + violation: str + detail: str + + def to_dict(self) -> dict[str, str]: + return { + "check": self.check, + "artifact": self.artifact, + "violation": self.violation, + "detail": self.detail, + } + + +@dataclass(frozen=True) +class SeamLintResult: + generated_at: datetime + expected_schema_version: str + max_staleness_hours: int + findings: tuple[SeamLintFinding, ...] + + @property + def passed(self) -> bool: + return not self.findings + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": "operator_os_seam_linter.v0", + "generated_at": self.generated_at.isoformat(), + "state": "pass" if self.passed else "fail", + "expected_schema_version": self.expected_schema_version, + "max_staleness_hours": self.max_staleness_hours, + "findings": [finding.to_dict() for finding in self.findings], + } + + +def lint_operator_os_seams( + *, + truth_path: Path, + 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)) + findings: list[SeamLintFinding] = [] + truth = _load_truth_artifact(truth_path, findings) + if truth is not None: + findings.extend( + _check_artifact_freshness( + truth, + truth_path=truth_path, + now=generated_at, + max_staleness_hours=max_staleness_hours, + ) + ) + findings.extend( + _check_schema_pin( + truth, + truth_path=truth_path, + 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, + expected_schema_version=expected_schema_version, + max_staleness_hours=max_staleness_hours, + findings=tuple(findings), + ) + + +def build_worklist_payload(result: SeamLintResult) -> dict[str, Any]: + items = [] + for finding in result.findings: + items.append( + { + "item_id": f"ghra_seam_linter:{finding.check}:{Path(finding.artifact).name}", + "kind": "operator_os_seam_linter", + "severity": "critical", + "title": f"Operator-OS seam-linter failed: {finding.check}", + "summary": f"{finding.artifact}: {finding.violation}. {finding.detail}", + "target_type": "artifact", + "target_id": finding.artifact, + "created_at": result.generated_at.isoformat(), + "suggested_command": ( + "uv run python -m src.operator_os_seam_linter " + "--truth output/portfolio-truth-latest.json" + ), + "metadata_json": json.dumps(finding.to_dict(), sort_keys=True), + } + ) + return { + "schema_version": WORKLIST_SCHEMA_VERSION, + "generated_at": result.generated_at.isoformat(), + "state": "pass" if result.passed else "fail", + "source": "GithubRepoAuditor", + "items": items, + } + + +def main(argv: list[str] | None = None) -> int: + parser = _build_parser() + args = parser.parse_args(argv) + workspace_root = Path(args.workspace_root).expanduser() + output_dir = Path(args.output_dir).expanduser() + truth_path = Path(args.truth).expanduser() if args.truth else truth_latest_path(output_dir) + markdown_paths = [Path(path).expanduser() for path in args.markdown] + if not markdown_paths: + markdown_paths = [ + workspace_root / "project-registry.md", + workspace_root / "PORTFOLIO-AUDIT-REPORT.md", + ] + result = lint_operator_os_seams( + truth_path=truth_path, + 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: + worklist_path = Path(args.worklist_output).expanduser() + worklist_path.parent.mkdir(parents=True, exist_ok=True) + worklist_path.write_text(json.dumps(build_worklist_payload(result), indent=2) + "\n") + if args.json: + print(json.dumps(payload, indent=2)) + else: + _print_text_summary(result) + return 0 if result.passed else 1 + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="operator-os-seam-linter", + description="Lint the small Operator-OS seam contract owned by GithubRepoAuditor.", + ) + parser.add_argument("--output-dir", type=Path, default=Path("output")) + parser.add_argument("--workspace-root", type=Path, default=Path.home() / "Projects") + 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") + return parser + + +def _load_truth_artifact( + truth_path: Path, findings: list[SeamLintFinding] +) -> dict[str, Any] | None: + try: + data = json.loads(truth_path.read_text()) + except FileNotFoundError: + findings.append( + SeamLintFinding( + check="artifact_freshness", + artifact=str(truth_path), + violation="truth artifact is missing", + detail="Regenerate portfolio truth before downstream consumers read it.", + ) + ) + return None + except json.JSONDecodeError as exc: + findings.append( + SeamLintFinding( + check="schema_pin", + artifact=str(truth_path), + violation="truth artifact is not valid JSON", + detail=str(exc), + ) + ) + return None + if not isinstance(data, dict): + findings.append( + SeamLintFinding( + check="schema_pin", + artifact=str(truth_path), + violation="truth artifact root is not an object", + detail="Expected the portfolio truth JSON object contract.", + ) + ) + return None + return data + + +def _check_artifact_freshness( + truth: dict[str, Any], + *, + truth_path: Path, + now: datetime, + max_staleness_hours: int, +) -> list[SeamLintFinding]: + raw_generated_at = truth.get("generated_at") + if not isinstance(raw_generated_at, str) or not raw_generated_at.strip(): + return [ + SeamLintFinding( + check="artifact_freshness", + artifact=str(truth_path), + violation="generated_at is missing", + detail="The truth artifact must declare when it was generated.", + ) + ] + try: + generated_at = _parse_datetime(raw_generated_at) + except ValueError as exc: + return [ + SeamLintFinding( + check="artifact_freshness", + artifact=str(truth_path), + violation="generated_at is not parseable", + detail=str(exc), + ) + ] + max_age = timedelta(hours=max_staleness_hours) + age = now - generated_at + if age > max_age: + return [ + SeamLintFinding( + check="artifact_freshness", + artifact=str(truth_path), + violation="truth artifact is stale", + detail=( + f"generated_at={raw_generated_at}; " + f"age_hours={age.total_seconds() / 3600:.2f}; " + f"max_staleness_hours={max_staleness_hours}" + ), + ) + ] + if generated_at - now > timedelta(minutes=5): + return [ + SeamLintFinding( + check="artifact_freshness", + artifact=str(truth_path), + violation="generated_at is in the future", + detail=f"generated_at={raw_generated_at}; now={now.isoformat()}", + ) + ] + return [] + + +def _check_schema_pin( + truth: dict[str, Any], + *, + truth_path: Path, + expected_schema_version: str, +) -> list[SeamLintFinding]: + declared = truth.get("schema_version") + if not isinstance(declared, str) or not declared.strip(): + return [ + SeamLintFinding( + check="schema_pin", + artifact=str(truth_path), + violation="schema_version is missing", + detail="The truth artifact must declare its schema pin.", + ) + ] + if declared != expected_schema_version: + return [ + SeamLintFinding( + check="schema_pin", + artifact=str(truth_path), + violation="schema_version mismatch", + detail=f"declared={declared}; expected={expected_schema_version}", + ) + ] + 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: + try: + text = path.read_text() + except FileNotFoundError: + findings.append( + SeamLintFinding( + check="markdown_generated_not_hand_edited", + artifact=str(path), + violation="generated markdown is missing", + detail="Expected a generated compatibility markdown artifact.", + ) + ) + continue + if GENERATED_MARKDOWN_PROVENANCE_MARKER not in text: + findings.append( + SeamLintFinding( + check="markdown_generated_not_hand_edited", + artifact=str(path), + violation="generated provenance marker is missing", + detail=( + "Regenerate this markdown with GithubRepoAuditor; " + "the linter does not guess via content diffs." + ), + ) + ) + return findings + + +def _parse_datetime(value: str) -> datetime: + normalized = value.strip() + if normalized.endswith("Z"): + normalized = f"{normalized[:-1]}+00:00" + parsed = datetime.fromisoformat(normalized) + return _aware(parsed) + + +def _aware(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +def _print_text_summary(result: SeamLintResult) -> None: + state = "PASS" if result.passed else "FAIL" + print(f"Operator-OS seam-linter: {state}") + print(f"expected_schema_version={result.expected_schema_version}") + print(f"max_staleness_hours={result.max_staleness_hours}") + if not result.findings: + print("No seam violations found.") + return + for finding in result.findings: + print(f"- {finding.check}: {finding.artifact}: {finding.violation}") + print(f" {finding.detail}") + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/project_registry.py b/src/project_registry.py index f415147d..63db72df 100644 --- a/src/project_registry.py +++ b/src/project_registry.py @@ -1 +1,628 @@ -@src/project_registry.py \ No newline at end of file +"""Canonical cross-store project-identity registry. + +Joins the four stores that key projects differently — this auditor +(``identity.project_key``), bridge-db (``project_name``), Notion's Local +Portfolio Projects (row title), and ``~/.claude`` memory (``project_*.md`` +slug) — under one canonical key, so events stop going unmatched. + +The auditor is the system of record for *what exists*, so its +``project_key`` is the canonical key, with ``repo_full_name`` as the stable +secondary natural key. A normalization function bridges the majority of +spelling differences; a small curated override table (see +``config/project-registry-overrides.json``) handles the cases where even +normalization diverges, plus supplementary entries for operator-OS projects +the auditor does not track (e.g. ``personal-ops``). + +Every external source is optional: a missing bridge-db / Notion snapshot / +memory dir degrades to reduced coverage rather than failing the run. +""" + +from __future__ import annotations + +import json +import re +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path + +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. +DEFAULT_OVERRIDES: dict[str, str] = { + "jcc": "JobCommandCenter", + "jsm_export": "JSMTicketAnalyticsExport", + "bhv": "BrowserHistoryVisualizer", + "netmapper": "NetworkMapper", + "notion_os": "Notion", + "screenshotselect": "ScreenshottoDataSelect", + "interruptionresume": "Interruption Resume Studio", +} + +# Real operator-OS projects absent from the auditor's repo registry. +DEFAULT_SUPPLEMENTARY: list[dict] = [ + { + "canonical_key": "supp:personal-ops", + "display_name": "personal-ops", + "repo_full_name": None, + "group_key": "operator_infra", + "lifecycle_state": "active", + "note": ( + "Local operator control plane (127.0.0.1:46210). Most active " + "project in bridge-db yet absent from auditor portfolio-truth." + ), + }, + { + "canonical_key": "supp:SecondBrain", + "display_name": "SecondBrain", + "repo_full_name": None, + "group_key": "operator_infra", + "lifecycle_state": "active", + "note": ( + "4-layer knowledge vault at /Users/d/Documents/SecondBrain " + "(engraph-indexed). Not a git repo; absent from auditor." + ), + }, +] + +# Memory slugs that are notes about a project, not their own project. +# slug -> parent canonical_key (empty string = pure meta, attach to nothing). +DEFAULT_MEMORY_META: dict[str, str] = { + "personal_ops_codebase": "supp:personal-ops", + "personal_ops_vision": "supp:personal-ops", + "github_repo_auditor_future_arcs": "GithubRepoAuditor", + "skill_library_port_2026-05": "", + "skill_eval_harness_2026-05": "", +} + +DEFAULT_NOTION_TITLE_ALIASES: dict[str, str] = { + "DesktopPEt-ready": "DesktopPEt", + "EarthPulse-readiness": "EarthPulse", + "GithubRepoAuditor-public": "GithubRepoAuditor", + "Notion Operating System": "Notion", + "OrbitForge (staging)": "OrbitForge", + "Personal Ops": "operator-os-docs", + "PomGambler-prod": "PomGambler", +} + +DEFAULT_NOTION_PROJECTION_ONLY_ROWS: dict[str, str] = { + "app": "local runtime/app shell placeholder; not a portfolio-truth repo", + "claude-code-harness": "local agent harness projection; outside repo-root truth", + "Sandbox Local Portfolio Project": "actuation sandbox fixture row", + "SecondBrain": "knowledge vault under /Users/d/Documents; not a /Users/d/Projects repo", +} + +DEFAULT_NOTION_TRUTH_SHADOW_ROWS: dict[str, str] = { + "agent-bridge-launch": "agent-bridge", + "PortfolioCommandCenter-public": "PortfolioCommandCenter", +} + +# Operator-machine source locations (overridable via the "sources" block of +# config/project-registry-overrides.json). Every source is optional. +DEFAULT_SOURCES: dict[str, str] = { + "bridge_db": "~/.local/share/bridge-db/bridge.db", + "notion_snapshot": "~/.local/share/notion-os/project-snapshot.json", + "memory_dir": "~/.claude/projects/-Users-d/memory", + "scoring_data_source_id": "35e04e4d-bcd8-45c0-b783-238edef210f7", +} + +_NON_ALNUM = re.compile(r"[^a-z0-9]") + + +def normalize(value: str | None) -> str: + """Lowercase, drop any taxonomy path prefix, strip non-alphanumerics.""" + if not value: + return "" + text = str(value) + if "/" in text: + text = text.rsplit("/", 1)[-1] + return _NON_ALNUM.sub("", text.lower()) + + +def _repo_base(repo_full_name: str | None) -> str: + return repo_full_name.rsplit("/", 1)[-1] if repo_full_name else "" + + +def _strip_alias_prefix(alias: str) -> str: + return alias.split(":", 1)[1] if ":" in alias else alias + + +def load_overrides_config( + config_path: Path | None, +) -> tuple[ + dict[str, str], + list[dict], + dict[str, str], + str, + dict[str, str], + dict[str, str], + dict[str, str], +]: + """Load overrides + supplementary + memory-meta, falling back to defaults.""" + if config_path is None or not config_path.exists(): + return ( + dict(DEFAULT_OVERRIDES), + [dict(s) for s in DEFAULT_SUPPLEMENTARY], + dict(DEFAULT_MEMORY_META), + NOTION_PROJECTION_POLICY_SCHEMA_VERSION, + dict(DEFAULT_NOTION_TITLE_ALIASES), + dict(DEFAULT_NOTION_PROJECTION_ONLY_ROWS), + dict(DEFAULT_NOTION_TRUTH_SHADOW_ROWS), + ) + data = json.loads(config_path.read_text()) + overrides = data.get("overrides", DEFAULT_OVERRIDES) + supplementary = data.get("supplementary", DEFAULT_SUPPLEMENTARY) + memory_meta = data.get("memory_meta", DEFAULT_MEMORY_META) + projection_policy_schema_version = data.get( + "notion_projection_policy_schema_version", + NOTION_PROJECTION_POLICY_SCHEMA_VERSION, + ) + title_aliases = data.get("notion_title_aliases", DEFAULT_NOTION_TITLE_ALIASES) + projection_only = data.get( + "notion_projection_only_rows", DEFAULT_NOTION_PROJECTION_ONLY_ROWS + ) + truth_shadow = data.get( + "notion_truth_shadow_rows", DEFAULT_NOTION_TRUTH_SHADOW_ROWS + ) + return ( + dict(overrides), + [dict(s) for s in supplementary], + dict(memory_meta), + str(projection_policy_schema_version), + dict(title_aliases), + dict(projection_only), + dict(truth_shadow), + ) + + +def load_source_paths(config_path: Path | None) -> dict[str, object]: + """Resolve external-source locations, merging config over built-in defaults. + + Returns a dict with ``bridge_db``/``notion_snapshot``/``memory_dir`` as + expanded ``Path`` objects and ``scoring_data_source_id`` as a string. + """ + sources = dict(DEFAULT_SOURCES) + if config_path is not None and config_path.exists(): + try: + configured = json.loads(config_path.read_text()).get("sources", {}) + sources.update({k: v for k, v in configured.items() if v}) + except (json.JSONDecodeError, OSError): + pass + return { + "bridge_db": Path(sources["bridge_db"]).expanduser(), + "notion_snapshot": Path(sources["notion_snapshot"]).expanduser(), + "memory_dir": Path(sources["memory_dir"]).expanduser(), + "scoring_data_source_id": sources.get("scoring_data_source_id"), + } + + +def _read_bridge_names(bridge_db_path: Path | None) -> list[str]: + if bridge_db_path is None or not bridge_db_path.exists(): + return [] + try: + uri = f"file:{bridge_db_path}?mode=ro" + with sqlite3.connect(uri, uri=True) as conn: + rows = conn.execute( + "SELECT DISTINCT project_name FROM activity_log " + "UNION SELECT DISTINCT project_name FROM pending_handoffs" + ).fetchall() + return [r[0] for r in rows if r[0]] + except sqlite3.Error: + return [] + + +def _read_notion_titles(notion_snapshot_path: Path | None) -> list[str]: + if notion_snapshot_path is None or not notion_snapshot_path.exists(): + return [] + try: + data = json.loads(notion_snapshot_path.read_text()) + return [p["title"] for p in data.get("projects", []) if p.get("title")] + except (json.JSONDecodeError, OSError, KeyError): + return [] + + +def _read_notion_pageids(notion_project_map_path: Path | None) -> dict[str, str]: + if notion_project_map_path is None or not notion_project_map_path.exists(): + return {} + try: + data = json.loads(notion_project_map_path.read_text()) + return { + name: entry["localProjectId"] + for name, entry in data.items() + if isinstance(entry, dict) and entry.get("localProjectId") + } + except (json.JSONDecodeError, OSError): + return {} + + +def _read_memory_slugs(memory_dir: Path | None) -> list[str]: + if memory_dir is None or not memory_dir.exists(): + return [] + return sorted(p.name[len("project_") : -len(".md")] for p in memory_dir.glob("project_*.md")) + + +class _Entry: + """Mutable accumulator for one canonical project during the join.""" + + __slots__ = ( + "canonical_key", + "display_name", + "repo_full_name", + "group_key", + "lifecycle_state", + "source", + "note", + "matchset", + "bridge_names", + "notion_local_title", + "notion_local_page_id", + "notion_scoring_page_id", + "memory_slug", + "memory_meta", + "aliases", + ) + + def __init__(self, identity: dict, lifecycle_state: str | None, source: str, note: str | None): + self.canonical_key = identity["project_key"] + self.display_name = identity.get("display_name") or self.canonical_key + self.repo_full_name = identity.get("repo_full_name") or None + self.group_key = identity.get("group_key") + self.lifecycle_state = lifecycle_state + self.source = source + self.note = note + self.matchset = { + f + for f in ( + normalize(self.display_name), + normalize(self.canonical_key), + normalize(_repo_base(self.repo_full_name)), + ) + if f + } + self.bridge_names: list[str] = [] + self.notion_local_title: str | None = None + self.notion_local_page_id: str | None = None + self.notion_scoring_page_id: str | None = None + self.memory_slug: str | None = None + self.memory_meta: list[str] = [] + self.aliases: set[str] = set() + + def add_alias(self, prefixed: str) -> None: + if _strip_alias_prefix(prefixed) != self.display_name: + self.aliases.add(prefixed) + + def to_dict(self) -> dict: + out = { + "canonical_key": self.canonical_key, + "display_name": self.display_name, + "repo_full_name": self.repo_full_name, + "group_key": self.group_key, + "lifecycle_state": self.lifecycle_state, + "source": self.source, + "bridge_project_names": self.bridge_names, + "notion_local_title": self.notion_local_title, + "notion_local_page_id": self.notion_local_page_id, + "notion_scoring_page_id": self.notion_scoring_page_id, + "memory_slug": self.memory_slug, + "memory_meta_notes": self.memory_meta, + "aliases": sorted(self.aliases), + "coverage": { + "auditor": self.source == "auditor", + "bridge": bool(self.bridge_names), + "notion_local": bool(self.notion_local_title), + "memory": bool(self.memory_slug), + }, + } + if self.note: + out["note"] = self.note + return out + + +def build_project_registry( + snapshot: dict, + *, + bridge_db_path: Path | None = None, + notion_snapshot_path: Path | None = None, + notion_project_map_path: Path | None = None, + memory_dir: Path | None = None, + scoring_pageids: dict[str, str] | None = None, + overrides_config_path: Path | None = None, + generated_at: datetime | None = None, +) -> dict: + """Build the canonical registry from a portfolio-truth snapshot dict. + + ``snapshot`` is the serialized portfolio-truth (``snapshot.to_dict()``). + All other sources are optional and degrade gracefully. + """ + ( + overrides, + supplementary, + memory_meta, + notion_projection_policy_schema_version, + notion_title_aliases, + notion_projection_only_rows, + notion_truth_shadow_rows, + ) = load_overrides_config(overrides_config_path) + generated_at = generated_at or datetime.now(timezone.utc) + + entries: list[_Entry] = [ + _Entry(p["identity"], (p.get("declared") or {}).get("lifecycle_state"), "auditor", None) + for p in snapshot.get("projects", []) + ] + for supp in supplementary: + entries.append( + _Entry( + { + "project_key": supp["canonical_key"], + "display_name": supp.get("display_name"), + "repo_full_name": supp.get("repo_full_name"), + "group_key": supp.get("group_key"), + }, + supp.get("lifecycle_state"), + "supplementary", + supp.get("note"), + ) + ) + + by_key = {e.canonical_key: e for e in entries} + index: dict[str, _Entry] = {} + collisions: list[dict] = [] + for entry in entries: + for form in entry.matchset: + existing = index.get(form) + if existing is None: + index[form] = entry + elif existing is not entry: + # Two distinct projects normalize to the same form: the index + # keeps the first (stable), so the second would mis-resolve. + # Surface it as a warning rather than failing silently. + collisions.append( + { + "normalized_form": form, + "kept": existing.canonical_key, + "shadowed": entry.canonical_key, + } + ) + override_norm = {normalize(raw): key for raw, key in overrides.items()} + title_alias_norm = { + normalize(raw): target for raw, target in notion_title_aliases.items() + } + projection_only_norm = { + normalize(raw): raw for raw in notion_projection_only_rows + } + + def resolve_entry_direct(raw: str) -> _Entry | None: + norm = normalize(raw) + if not norm: + return None + if norm in override_norm: + target = by_key.get(override_norm[norm]) + if target is not None: + return target + return index.get(norm) + + def resolve_entry(raw: str) -> _Entry | None: + entry = resolve_entry_direct(raw) + if entry is not None: + return entry + alias_target = title_alias_norm.get(normalize(raw)) + if alias_target: + return resolve_entry_direct(alias_target) + return None + + notion_orphans: list[str] = [] + notion_projection_only: list[dict[str, str]] = [] + for title in _read_notion_titles(notion_snapshot_path): + entry = resolve_entry(title) + if entry is not None: + entry.notion_local_title = title + entry.add_alias(f"notion:{title}") + elif normalize(title) in projection_only_norm: + notion_projection_only.append( + { + "title": title, + "reason": notion_projection_only_rows.get(title) + or notion_projection_only_rows.get(projection_only_norm[normalize(title)]) + or "", + } + ) + else: + notion_orphans.append(title) + + pageid_unmatched: list[str] = [] + for name, page_id in _read_notion_pageids(notion_project_map_path).items(): + entry = resolve_entry(name) + if entry is not None: + entry.notion_local_page_id = page_id + entry.add_alias(f"notionmap:{name}") + else: + pageid_unmatched.append(name) + + for project_name, page_id in (scoring_pageids or {}).items(): + entry = resolve_entry(project_name) + if entry is not None: + entry.notion_scoring_page_id = page_id + + memory_orphans: list[dict] = [] + for slug in _read_memory_slugs(memory_dir): + if slug in memory_meta: + parent = memory_meta[slug] + if parent and parent in by_key: + by_key[parent].memory_meta.append(f"project_{slug}") + continue + if not parent: + memory_orphans.append({"slug": slug, "kind": "meta-epic-note"}) + continue + entry = resolve_entry(slug) + if entry is not None: + if entry.memory_slug is None: + entry.memory_slug = f"project_{slug}" + else: + entry.memory_meta.append(f"project_{slug}") + entry.add_alias(f"memory:{slug}") + else: + memory_orphans.append({"slug": slug, "kind": "unmatched"}) + + bridge_orphans: list[str] = [] + for name in _read_bridge_names(bridge_db_path): + entry = resolve_entry(name) + if entry is not None: + if name not in entry.bridge_names: + entry.bridge_names.append(name) + entry.add_alias(f"bridge:{name}") + else: + bridge_orphans.append(name) + + return { + "schema_version": SCHEMA_VERSION, + "generated_at": generated_at.strftime("%Y-%m-%dT%H:%M:%SZ"), + "description": ( + "Canonical cross-store project-identity registry for the operator " + "OS. Joins GithubRepoAuditor, bridge-db, Notion (Local Portfolio " + "Projects), and ~/.claude memory under one canonical key." + ), + "canonical_key": { + "primary": "GithubRepoAuditor identity.project_key (taxonomy-path-qualified)", + "secondary": "repo_full_name (saagpatel/)", + "supplementary": "supp: for operator-OS projects the auditor does not track", + }, + "entry_count": len(entries), + "resolution_overrides": overrides, + "projection_policy": { + "schema_version": notion_projection_policy_schema_version, + "notion_title_aliases": notion_title_aliases, + "notion_projection_only_rows": notion_projection_only_rows, + "notion_truth_shadow_rows": notion_truth_shadow_rows, + }, + "entries": [e.to_dict() for e in entries], + "projection_only": { + "notion_local": sorted(notion_projection_only, key=lambda row: row["title"]) + }, + "unmatched": { + "bridge": sorted(bridge_orphans), + "memory": memory_orphans, + "notion_local": sorted(notion_orphans), + "notion_pageid_map": sorted(pageid_unmatched), + }, + "warnings": {"normalized_key_collisions": collisions}, + } + + +def build_index(registry: dict) -> dict: + """Precompute lookup structures from a built registry for resolve().""" + norm2entry: dict[str, dict] = {} + for entry in registry["entries"]: + forms = {normalize(entry["display_name"])} + if entry.get("repo_full_name"): + forms.add(normalize(_repo_base(entry["repo_full_name"]))) + if "/" in (entry.get("canonical_key") or ""): + forms.add(normalize(entry["canonical_key"])) + for alias in entry.get("aliases", []): + forms.add(normalize(_strip_alias_prefix(alias))) + for form in forms: + if form: + norm2entry.setdefault(form, entry) + override_norm = { + normalize(raw): key for raw, key in registry.get("resolution_overrides", {}).items() + } + by_key = {e["canonical_key"]: e for e in registry["entries"]} + return {"norm2entry": norm2entry, "override_norm": override_norm, "by_key": by_key} + + +def resolve(name: str, index: dict) -> dict | None: + """Map a free-form project name to its canonical entry, or None.""" + norm = normalize(name) + if not norm: + return None + if norm in index["override_norm"]: + entry = index["by_key"].get(index["override_norm"][norm]) + if entry is not None: + return { + "canonical_key": entry["canonical_key"], + "display_name": entry["display_name"], + "matched_via": "override", + } + entry = index["norm2entry"].get(norm) + if entry is not None: + return { + "canonical_key": entry["canonical_key"], + "display_name": entry["display_name"], + "matched_via": "normalized", + } + return None + + +def fetch_scoring_pageids(data_source_id: str, token: str) -> dict[str, str]: + """Read Project Name -> page_id from the Notion Project Portfolio DB. + + Uses the auditor's Notion client; paginates the data source. Returns an + empty dict on any failure so registry generation never hard-depends on it. + """ + from src.notion_client import query_notion_collection + + result: dict[str, str] = {} + cursor: str | None = None + try: + while True: + body = {"page_size": 100} + if cursor: + body["start_cursor"] = cursor + response = query_notion_collection(data_source_id, token, body=body) + if response is None or response.status_code != 200: + break + payload = response.json() + for page in payload.get("results", []): + title_prop = page.get("properties", {}).get("Project Name", {}) + segments = title_prop.get("title", []) if isinstance(title_prop, dict) else [] + name = "".join(seg.get("plain_text", "") for seg in segments).strip() + if name and page.get("id"): + result[name] = page["id"] + if not payload.get("has_more"): + break + cursor = payload.get("next_cursor") + if not cursor: + break + except Exception: + return result + return result diff --git a/tests/test_operator_os_seam_linter.py b/tests/test_operator_os_seam_linter.py index 7ed026fa..c11a7ee3 100644 --- a/tests/test_operator_os_seam_linter.py +++ b/tests/test_operator_os_seam_linter.py @@ -1 +1,357 @@ -@tests/test_operator_os_seam_linter.py \ No newline at end of file +from __future__ import annotations + +import json +import sqlite3 +from datetime import UTC, datetime +from pathlib import Path + +from src.operator_os_seam_linter import ( + build_worklist_payload, + lint_operator_os_seams, + main, +) +from src.portfolio_truth_render import GENERATED_MARKDOWN_PROVENANCE_MARKER +from src.portfolio_truth_types import SCHEMA_VERSION + + +NOW = datetime(2026, 7, 3, 12, 0, tzinfo=UTC) + + +def _write_truth( + path: Path, + *, + 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 or [], + } + if schema_version is not None: + payload["schema_version"] = schema_version + path.write_text(json.dumps(payload)) + + +def _write_markdown(path: Path, *, marker: bool = True) -> None: + prefix = f"{GENERATED_MARKDOWN_PROVENANCE_MARKER}\n\n" if marker else "" + path.write_text(f"{prefix}# Generated Artifact\n") + + +def _passing_paths(tmp_path: Path) -> tuple[Path, list[Path]]: + truth = tmp_path / "portfolio-truth-latest.json" + registry = tmp_path / "project-registry.md" + report = tmp_path / "PORTFOLIO-AUDIT-REPORT.md" + _write_truth(truth) + _write_markdown(registry) + _write_markdown(report) + 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) + + result = lint_operator_os_seams( + truth_path=truth, + markdown_paths=markdown, + now=NOW, + max_staleness_hours=30, + ) + + assert result.passed + + +def test_stale_artifact_fails(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_truth(truth, generated_at="2026-07-01T00:00:00+00:00") + + result = lint_operator_os_seams( + truth_path=truth, + markdown_paths=markdown, + now=NOW, + max_staleness_hours=30, + ) + + assert not result.passed + assert [finding.check for finding in result.findings] == ["artifact_freshness"] + assert "stale" in result.findings[0].violation + + +def test_schema_pin_passes(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + + result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) + + assert result.passed + + +def test_schema_pin_mismatch_fails(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_truth(truth, schema_version="0.6.0") + + result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) + + assert not result.passed + assert result.findings[0].check == "schema_pin" + assert "mismatch" in result.findings[0].violation + + +def test_schema_pin_missing_fails(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_truth(truth, schema_version=None) + + result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) + + assert not result.passed + assert result.findings[0].check == "schema_pin" + assert "missing" in result.findings[0].violation + + +def test_generated_markdown_with_marker_passes(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + + result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) + + assert result.passed + + +def test_hand_edited_markdown_without_marker_fails(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_markdown(markdown[0], marker=False) + + result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) + + assert not result.passed + assert result.findings[0].check == "markdown_generated_not_hand_edited" + assert "provenance marker is missing" in result.findings[0].violation + + +def test_worklist_payload_uses_existing_attention_item_shape(tmp_path: Path) -> None: + truth, markdown = _passing_paths(tmp_path) + _write_truth(truth, schema_version="0.6.0") + result = lint_operator_os_seams(truth_path=truth, markdown_paths=markdown, now=NOW) + + payload = build_worklist_payload(result) + + assert payload["schema_version"] == "operator_os_seam_linter_worklist.v1" + assert payload["state"] == "fail" + assert payload["items"][0]["kind"] == "operator_os_seam_linter" + assert payload["items"][0]["severity"] == "critical" + 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") + worklist = tmp_path / "worklist.json" + + code = main( + [ + "--truth", + str(truth), + "--markdown", + str(markdown[0]), + "--markdown", + str(markdown[1]), + "--worklist-output", + str(worklist), + "--json", + ] + ) + + assert code == 1 + payload = json.loads(worklist.read_text()) + assert payload["items"][0]["kind"] == "operator_os_seam_linter"