From 9ac63cde22f948668e9d7b4759a3bc4df048e38f Mon Sep 17 00:00:00 2001 From: Saagar Date: Fri, 3 Jul 2026 02:20:56 -0700 Subject: [PATCH] feat(seam-linter): add Operator-OS conformance checks - Add artifact freshness and schema-pin checks for portfolio truth - Add generated markdown provenance marker validation - Emit worklist-shaped findings for nightly maintenance Tests: uv run pytest -q; uv run ruff check . --- scripts/operator_os_seam_linter.py | 17 ++ src/operator_os_seam_linter.py | 351 ++++++++++++++++++++++++++ src/portfolio_truth_render.py | 8 + tests/test_operator_os_seam_linter.py | 162 ++++++++++++ 4 files changed, 538 insertions(+) create mode 100644 scripts/operator_os_seam_linter.py create mode 100644 src/operator_os_seam_linter.py create mode 100644 tests/test_operator_os_seam_linter.py diff --git a/scripts/operator_os_seam_linter.py b/scripts/operator_os_seam_linter.py new file mode 100644 index 00000000..ed811756 --- /dev/null +++ b/scripts/operator_os_seam_linter.py @@ -0,0 +1,17 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + + +def _main() -> int: + from src.operator_os_seam_linter import main + + return main() + + +if __name__ == "__main__": + raise SystemExit(_main()) diff --git a/src/operator_os_seam_linter.py b/src/operator_os_seam_linter.py new file mode 100644 index 00000000..1f8db99c --- /dev/null +++ b/src/operator_os_seam_linter.py @@ -0,0 +1,351 @@ +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()) diff --git a/src/portfolio_truth_render.py b/src/portfolio_truth_render.py index 8b4dae9e..f8bd68d6 100644 --- a/src/portfolio_truth_render.py +++ b/src/portfolio_truth_render.py @@ -9,6 +9,10 @@ # the digest cap the per-repo security callout list at the same depth so the two # human-facing surfaces stay consistent. MAX_SECURITY_ATTENTION_ITEMS = 5 +GENERATED_MARKDOWN_PROVENANCE_MARKER = ( + "" +) def _security_overview(projects: list[PortfolioTruthProject]) -> dict[str, int]: @@ -59,6 +63,8 @@ def render_registry_markdown(snapshot: PortfolioTruthSnapshot) -> str: grouped = _group_projects(snapshot.projects) project_labels = registry_project_labels(snapshot.projects) lines = [ + GENERATED_MARKDOWN_PROVENANCE_MARKER, + "", "# Project Registry", "", f"> Master index for Cowork automated tasks. Last generated: {generated_date}.", @@ -108,6 +114,8 @@ def render_portfolio_report_markdown( risk_tier_counts = Counter(project.risk.risk_tier for project in snapshot.projects) security_overview = _security_overview(snapshot.projects) lines = [ + GENERATED_MARKDOWN_PROVENANCE_MARKER, + "", "# Portfolio Audit Report", "", f"> Generated: {generated} | Projects in truth snapshot: {len(snapshot.projects)}", diff --git a/tests/test_operator_os_seam_linter.py b/tests/test_operator_os_seam_linter.py new file mode 100644 index 00000000..79916896 --- /dev/null +++ b/tests/test_operator_os_seam_linter.py @@ -0,0 +1,162 @@ +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"