feat(graph): preserve integrity and temporal evidence

.github/workflows/validate-graph.yml

@@ -0,0 +1,34 @@
+name: validate / graph
+
+on:
+  pull_request:
+    paths:
+      - "apps/python-app/libs/cssc_graph/**"
+      - "apps/python-app/services/graph-service/**"
+      - "supply-chain-graph/**"
+      - ".github/workflows/validate-graph.yml"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  graph:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5.0.0
+      - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6
+        with:
+          python-version: "3.12"
+      - name: Install declared graph test dependencies
+        run: |
+          python -m pip install -e 'apps/python-app/libs/cssc_graph[test]' \
+            -e 'apps/python-app/services/graph-service[test]'
+      - name: Require the real graph engine and run library and HTTP tests
+        run: |
+          python -c 'import ladybug; print("Native LadybugDB engine available")'
+          python -m pytest --import-mode=importlib -q \
+            apps/python-app/libs/cssc_graph/tests \
+            apps/python-app/services/graph-service/tests
+      - name: Validate committed example schemas
+        run: cssc-graph validate supply-chain-graph

apps/python-app/libs/cssc_graph/README.md

@@ -25,6 +25,14 @@ cssc-graph id supply-chain-graph/examples/artifact-mirrored.yaml
 
 `validate` exits non-zero on any error, so it fits pre-commit and CI.
 
+For current indexing, query and integrity behavior, see the
+[query guide](../../../../docs/guides/observability/querying-the-supply-chain-graph.md).
+`index` emits unresolved-reference warnings by default;
+`validate --strict-references` and `index --strict-references` reject missing
+independent occurrence evidence. Deployment observations are available through
+`deployments --ref`, and `tag-history --format json` includes explicit
+observation-derived intervals and ambiguity.
+
 ## Concepts
 
 - **Envelope** — the common wrapper on every record (`schemaVersion`, `kind`,

apps/python-app/libs/cssc_graph/cssc_graph/cli.py

@@ -57,13 +57,19 @@ def cli() -> None:
     default="text",
     help="Output format.",
 )
-def validate(root: Path, schema_dir: Path | None, output_format: str) -> None:
-    """Validate every record under ROOT against the JSON Schemas."""
+@click.option("--strict-references", is_flag=True, help="Also reject references without independent input evidence.")
+def validate(root: Path, schema_dir: Path | None, output_format: str, strict_references: bool) -> None:
+    """Validate schemas, optionally requiring all occurrence references to resolve."""
 
     resolved_schema_dir = schema_dir or default_schema_dir(root)
     try:
         files = discover_record_files(root, resolved_schema_dir)
         diagnostics = validate_data(root, resolved_schema_dir)
+        if strict_references and not diagnostics:
+            from .records import iter_records
+            from .references import reference_diagnostics
+
+            diagnostics = reference_diagnostics(iter_records(root, resolved_schema_dir), strict=True)
     except (FileNotFoundError, ValueError) as exc:
         raise click.ClickException(str(exc))
 
@@ -116,12 +122,18 @@ def validate(root: Path, schema_dir: Path | None, output_format: str) -> None:
     is_flag=True,
     help="Delete any existing database first for a clean rebuild.",
 )
-def index(root: Path, database: Path, schema_dir: Path | None, rebuild: bool) -> None:
-    """Validate ROOT, then index its records into a LadybugDB graph."""
+@click.option("--strict-references", is_flag=True, help="Reject unresolved references before changing the database.")
+def index(root: Path, database: Path, schema_dir: Path | None, rebuild: bool, strict_references: bool) -> None:
+    """Validate and index ROOT; default reference warnings preserve permissive behavior."""
 
     resolved_schema_dir = schema_dir or default_schema_dir(root)
     try:
         diagnostics = validate_data(root, resolved_schema_dir)
+        if strict_references and not diagnostics:
+            from .records import iter_records
+            from .references import reference_diagnostics
+
+            diagnostics = reference_diagnostics(iter_records(root, resolved_schema_dir), strict=True)
     except (FileNotFoundError, ValueError) as exc:
         raise click.ClickException(str(exc))
     if diagnostics:
@@ -135,10 +147,15 @@ def index(root: Path, database: Path, schema_dir: Path | None, rebuild: bool) ->
 
     if rebuild:
         GraphStore.destroy(database)
-    with GraphStore(database) as store:
-        store.init_schema()
-        stats = index_data(store, root, resolved_schema_dir)
+    try:
+        with GraphStore(database) as store:
+            store.init_schema()
+            stats = index_data(store, root, resolved_schema_dir)
+    except ValueError as exc:
+        raise click.ClickException(str(exc)) from exc
 
+    for diagnostic in stats.diagnostics:
+        click.echo(diagnostic.format(root), err=True)
     click.echo(f"Indexed {stats.records} record(s) into {database}.")
     for kind, count in sorted(stats.by_kind.items()):
         click.echo(f"  {kind}: {count}")
@@ -168,9 +185,13 @@ def _format_option(func):
 
 
 def _open_store(database: Path):
+    """Open a compatible graph for queries, with actionable legacy-rebuild errors."""
     from .graph import GraphStore
 
-    return GraphStore(database).connect()
+    try:
+        return GraphStore(database).connect()
+    except ValueError as exc:
+        raise click.ClickException(str(exc)) from exc
 
 
 def _emit_subgraph(subgraph: dict, output_format: str) -> None:
@@ -215,7 +236,7 @@ def path(database: Path, digest: str | None, ref: str | None, depth: int, output
 @click.option("--tag", required=True)
 @_format_option
 def tag_history(database: Path, repo: str, tag: str, output_format: str) -> None:
-    """List every digest a tag pointed to over time, chronologically."""
+    """List tag observations and evidence-derived intervals, with ambiguity explicit."""
 
     from . import queries
 
@@ -229,9 +250,38 @@ def tag_history(database: Path, repo: str, tag: str, output_format: str) -> None
         return
     for entry in history:
         click.echo(f"  {entry['observedAt']}  {entry['digest']}  {entry.get('runUrl', '')}".rstrip())
+        click.echo(
+            f"    observed interval: [{entry['validFrom'] or 'unknown'}, "
+            f"{entry['validUntil'] or 'unknown'}); ambiguous={entry['ambiguous']}"
+        )
     click.echo(f"{len(history)} observation(s).")
 
 
+@cli.command(name="deployments")
+@_database_option
+@click.option("--ref", required=True, help="The deployed registry/repository[@digest|:tag].")
+@_format_option
+def deployments(database: Path, ref: str, output_format: str) -> None:
+    """List distinct deployment observations in chronological order."""
+    from . import queries
+
+    store = _open_store(database)
+    try:
+        history = queries.deployment_history(store, ref)
+    finally:
+        store.close()
+    if output_format == "json":
+        click.echo(json.dumps(history, indent=2))
+    else:
+        for entry in history:
+            click.echo(
+                f"{entry['recordedAt']}  {entry['cluster']}/{entry['namespace']}  "
+                f"{entry['occurrence']}  chart={entry['chart']}:{entry['chartVersion']}  "
+                f"event={entry['eventId']}"
+            )
+        click.echo(f"{len(history)} deployment observation(s).")
+
+
 @cli.command()
 @_database_option
 @click.option("--ref", required=True, help="The built image (registry/repository[@digest|:tag]).")

apps/python-app/libs/cssc_graph/cssc_graph/graph.py

@@ -40,7 +40,7 @@
     "CREATE REL TABLE IF NOT EXISTS POINTED_TO("
     "FROM Tag TO Occurrence, observedAt STRING, digest STRING, runUrl STRING)",
     "CREATE REL TABLE IF NOT EXISTS RUNS("
-    "FROM Deployment TO Occurrence, chart STRING, chartVersion STRING, "
+    "FROM Deployment TO Occurrence, eventId STRING, sourceJson STRING, chart STRING, chartVersion STRING, "
     "runUrl STRING, recordedAt STRING)",
     "CREATE REL TABLE IF NOT EXISTS REFERS_TO("
     "FROM Occurrence TO Occurrence, artifactType STRING, observedAt STRING, "
@@ -62,13 +62,32 @@ def __init__(self, database_path: str | Path) -> None:
     # -- lifecycle ----------------------------------------------------------
 
     def connect(self) -> "GraphStore":
+        """Open the database and reject incompatible history schemas without altering them."""
         import ladybug as lb  # lazy: only the graph commands need the native dep
 
         self._db = lb.Database(str(self._path))
         self._conn = lb.Connection(self._db)
+        try:
+            self.check_history_schema()
+        except ValueError:
+            self.close()
+            raise
         return self
 
+    def check_history_schema(self) -> None:
+        """Read-only compatibility guard shared by query opens and schema initialization."""
+        existing = self.query("CALL show_tables() RETURN name")
+        if any(row["name"] == "RUNS" for row in existing):
+            columns = self.query("CALL table_info('RUNS') RETURN name")
+            if not {"eventId", "sourceJson"}.issubset({row["name"] for row in columns}):
+                raise ValueError(
+                    "graph schema predates deployment event history; rebuild this derived "
+                    "database with cssc-graph index --rebuild (source files are unchanged)"
+                )
+
     def init_schema(self) -> None:
+        """Create a fresh schema or reject an old deployment table without data loss."""
+        self.check_history_schema()
         for ddl in SCHEMA_DDL:
             self.execute(ddl)
 

apps/python-app/libs/cssc_graph/cssc_graph/indexer.py

@@ -2,26 +2,31 @@
 
 from __future__ import annotations
 
+import hashlib
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Mapping
 
 from .graph import GraphStore
 from .identity import (
+    canonical_json,
     make_occurrence_key,
     make_ref,
     split_ref,
     tag_key,
 )
 from .records import iter_records
+from .references import UnresolvedReferences, reference_diagnostics
+from .validate import Diagnostic
 
 
 @dataclass
 class IndexStats:
-    """Counts produced by an indexing run."""
+    """Indexed record counts and non-fatal reference diagnostics for one input set."""
 
     records: int = 0
     by_kind: dict[str, int] = field(default_factory=dict)
+    diagnostics: list[Diagnostic] = field(default_factory=list)
 
     def bump(self, kind: str) -> None:
         self.records += 1
@@ -175,17 +180,23 @@ def _index_artifact_built(self, record: Mapping[str, Any]) -> None:
         )
 
     def _index_artifact_deployed(self, record: Mapping[str, Any]) -> None:
+        """Preserve each full deployment observation; replaying it is idempotent."""
         image_key = self._merge_occurrence_from(record["image"])
         env = record.get("environment", {})
         dep_key = self._merge_deployment(env.get("cluster", ""), env.get("namespace", ""))
         chart = record.get("chart", {})
+        event_id = "sha256:" + hashlib.sha256(
+            canonical_json({key: value for key, value in record.items() if key != "id"}).encode("utf-8")
+        ).hexdigest()
         self._store.execute(
             "MATCH (d:Deployment {key: $dep}), (o:Occurrence {key: $image}) "
-            "MERGE (d)-[r:RUNS]->(o) "
-            "SET r.chart = $chart, r.chartVersion = $chartVersion, "
+            "MERGE (d)-[r:RUNS {eventId: $eventId}]->(o) "
+            "SET r.sourceJson = $sourceJson, r.chart = $chart, r.chartVersion = $chartVersion, "
             "r.runUrl = $runUrl, r.recordedAt = $recordedAt",
             {
                 "dep": dep_key,
+                "eventId": event_id,
+                "sourceJson": canonical_json(record.get("source", {})),
                 "image": image_key,
                 "chart": chart.get("name", ""),
                 "chartVersion": chart.get("version", ""),
@@ -300,16 +311,30 @@ def _run_url(record: Mapping[str, Any]) -> str:
 
 
 def _snake(kind: str) -> str:
-    out = []
+    """Map a record kind to its per-kind indexer method suffix."""
+    out: list[str] = []
     for ch in kind:
         if ch.isupper() and out:
             out.append("_")
         out.append(ch.lower())
     return "".join(out)
 
 
-def index_data(store: GraphStore, root: Path, schema_dir: Path | None = None) -> IndexStats:
-    """Index every record under *root* into *store*."""
+def index_data(
+    store: GraphStore, root: Path, schema_dir: Path | None = None,
+    *, strict_references: bool = False,
+) -> IndexStats:
+    """Index a validated input set, returning reference diagnostics.
 
+    Strict unresolved references raise before any record writes. Declarations
+    anywhere in this input set resolve references regardless of file order.
+    """
+
+    records = list(iter_records(root, schema_dir))
+    diagnostics = reference_diagnostics(records, strict=strict_references)
+    if strict_references and diagnostics:
+        raise UnresolvedReferences(diagnostics)
     indexer = Indexer(store)
-    return indexer.index_records(data for _, data in iter_records(root, schema_dir))
+    stats = indexer.index_records(data for _, data in records)
+    stats.diagnostics = diagnostics
+    return stats