feat(views): add ephemeral PQC worklist contract

README.md

@@ -140,6 +140,30 @@ Use JSON Schema format checking to enforce UUID and date-time fields; the schema
 does not compare window endpoints. The legacy CLI JSON array remains a consumer
 compatibility surface, separate from this `{window, misses}` view contract.
 
+`views_schema/v1/pqc-worklist.schema.json` is the Draft 2020-12 contract for
+Core's ephemeral PQC worklist. Its required wrapper fields are `product_id`,
+`repository_ref_ids`, `chokepoints` and `entries`. A null ref scope means all
+refs in scope; an empty array means none. Core intersects explicit ref scope
+with the selected product's shipped refs. Chokepoints are caller-supplied URL
+fragments, not persisted configuration.
+
+Entries preserve the legacy repository/ref IDs, raw repository and SHA strings,
+priority tier (`chokepoint`, `crypto-cwe`, `tail`), route (`source`, `sbom-only`,
+`access-request`), nested liveness, reason, nullable scan state and cursor.
+Liveness preserves nullable observations and historical `pushed_at`,
+`unobserved` and reason strings. Scan currency and liveness booleans distinguish
+unknown null from measured false; neither establishes readiness. Objects are
+strict and every serialized field is required, including nullable fields.
+Use format checking for UUIDs and observation timestamps; raw repository, SHA,
+scope and historical strings have no invented URL, hash or markup restrictions.
+
+Core owns `PqcService.worklist`, scope intersection, representative-ref choice,
+ranking, routing, cursor ordering, validation and safe human-readable rendering.
+Building this view uses recorded state only: it does not scan, check out code
+or write storage. The legacy CLI JSON array remains separate from this canonical
+wrapper, and raw untrusted strings remain data rather than authored Markdown
+or HTML.
+
 `views_schema/v1/compliance-precision.schema.json` wraps the native
 `DetectionQualityService.calibrate_compliance(framework, gate=...)` result as
 `{calibration: {framework, assessments, checks, gate}}`. Framework IDs reuse the

tests/test_pqc_worklist_view.py

@@ -0,0 +1,239 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+from jsonschema import Draft202012Validator, FormatChecker, ValidationError
+
+SCHEMA = Path(__file__).parents[1] / "views_schema/v1/pqc-worklist.schema.json"
+REPOSITORY_ID = "00000000-0000-4000-8000-000000000001"
+REF_ID = "00000000-0000-4000-8000-000000000002"
+ASSESSMENT_ID = "00000000-0000-4000-8000-000000000003"
+PRODUCT_ID = "00000000-0000-4000-8000-000000000004"
+HOSTILE = '<script>alert("repo")</script>\n| fake | row |\n[jump](javascript:alert(1)) & \u202e'
+
+
+@pytest.fixture
+def validator() -> Draft202012Validator:
+    schema = json.loads(SCHEMA.read_text())
+    Draft202012Validator.check_schema(schema)
+    return Draft202012Validator(schema, format_checker=FormatChecker())
+
+
+@pytest.fixture
+def document() -> dict[str, Any]:
+    return {
+        "product_id": None,
+        "repository_ref_ids": None,
+        "chokepoints": [],
+        "entries": [
+            {
+                "repository_id": REPOSITORY_ID,
+                "repository": "https://example.com/repo",
+                "repository_ref_id": REF_ID,
+                "sha": "legacy-sha",
+                "priority_tier": "tail",
+                "route": "source",
+                "access": {
+                    "repository_id": REPOSITORY_ID,
+                    "url": "https://example.com/repo",
+                    "observed_at": "2026-11-01T09:30:00+02:00",
+                    "archived": False,
+                    "pushed_at": "historical date",
+                    "dormant": False,
+                    "unobserved": None,
+                    "reason": None,
+                },
+                "reason": None,
+                "scan": {
+                    "repository_ref_id": REF_ID,
+                    "assessment_id": ASSESSMENT_ID,
+                    "basis": "recorded-basis",
+                    "current": True,
+                },
+                "cursor": "2|https://example.com/repo",
+            }
+        ],
+    }
+
+
+def target(document: dict[str, Any], path: tuple[str | int, ...]) -> Any:
+    value = document
+    for part in path:
+        value = value[part]
+    return value
+
+
+OBJECT_PATHS = [(), ("entries", 0), ("entries", 0, "access"), ("entries", 0, "scan")]
+
+
+@pytest.mark.parametrize("path", OBJECT_PATHS)
+def test_each_serialized_field_is_required(
+    validator: Draft202012Validator, document: dict[str, Any], path: tuple[str | int, ...]
+) -> None:
+    obj = target(document, path)
+    for field in tuple(obj):
+        saved = obj.pop(field)
+        with pytest.raises(ValidationError):
+            validator.validate(document)
+        obj[field] = saved
+
+
+@pytest.mark.parametrize("path", OBJECT_PATHS)
+def test_every_object_rejects_unknown_fields(
+    validator: Draft202012Validator, document: dict[str, Any], path: tuple[str | int, ...]
+) -> None:
+    target(document, path)["unexpected"] = "not part of the contract"
+    with pytest.raises(ValidationError):
+        validator.validate(document)
+
+
+@pytest.mark.parametrize("product_id", [None, PRODUCT_ID])
+@pytest.mark.parametrize("refs", [None, [], [REF_ID]])
+def test_global_product_and_explicit_empty_scope_remain_distinct(
+    validator: Draft202012Validator, document: dict[str, Any], product_id: Any, refs: Any
+) -> None:
+    document.update(product_id=product_id, repository_ref_ids=refs, entries=[])
+    validator.validate(document)
+    assert json.loads(json.dumps(document))["repository_ref_ids"] == refs
+
+
+@pytest.mark.parametrize("tier", ["chokepoint", "crypto-cwe", "tail"])
+@pytest.mark.parametrize("route", ["source", "sbom-only", "access-request"])
+def test_all_existing_routes_and_priority_tiers(
+    validator: Draft202012Validator, document: dict[str, Any], tier: str, route: str
+) -> None:
+    document["entries"][0].update(priority_tier=tier, route=route)
+    validator.validate(document)
+
+
+def test_unregistered_repository_preserves_null_ref_sha_and_scan(
+    validator: Draft202012Validator, document: dict[str, Any]
+) -> None:
+    entry = document["entries"][0]
+    entry.update(
+        repository_ref_id=None,
+        sha=None,
+        scan=None,
+        route="access-request",
+        reason="no-registered-commit",
+    )
+    entry["access"].update(
+        observed_at=None,
+        archived=None,
+        pushed_at=None,
+        dormant=None,
+        unobserved=None,
+        reason=None,
+    )
+    validator.validate(document)
+
+
+@pytest.mark.parametrize("state", [None, False, True])
+def test_unknown_and_measured_liveness_and_scan_states_are_distinct(
+    validator: Draft202012Validator, document: dict[str, Any], state: bool | None
+) -> None:
+    entry = document["entries"][0]
+    entry["access"].update(archived=state, dormant=state)
+    entry["scan"].update(current=state, assessment_id=None, basis=None)
+    validator.validate(document)
+    restored = json.loads(json.dumps(document))["entries"][0]
+    assert restored["scan"]["current"] is state
+    assert restored["access"]["archived"] is state
+    assert restored["access"]["dormant"] is state
+
+
+@pytest.mark.parametrize("raw", ["", "legacy-value", HOSTILE])
+def test_raw_scope_and_historical_strings_are_not_url_sha_or_markup_restricted(
+    validator: Draft202012Validator, document: dict[str, Any], raw: str
+) -> None:
+    document["chokepoints"] = [raw]
+    entry = document["entries"][0]
+    entry.update(repository=raw, sha=raw, reason=raw, cursor="2|" + raw)
+    entry["access"].update(url=raw, pushed_at=raw, unobserved=raw, reason=raw)
+    entry["scan"]["basis"] = raw
+    validator.validate(document)
+
+
+ID_PATHS = [
+    ("product_id",),
+    ("repository_ref_ids", 0),
+    ("entries", 0, "repository_id"),
+    ("entries", 0, "repository_ref_id"),
+    ("entries", 0, "access", "repository_id"),
+    ("entries", 0, "scan", "repository_ref_id"),
+    ("entries", 0, "scan", "assessment_id"),
+]
+
+
+@pytest.mark.parametrize("path", ID_PATHS)
+@pytest.mark.parametrize("invalid", ["not-a-uuid", "00000000000040008000000000000001", "", 1])
+def test_identifiers_require_uuid_strings(
+    validator: Draft202012Validator,
+    document: dict[str, Any],
+    path: tuple[str | int, ...],
+    invalid: Any,
+) -> None:
+    document["repository_ref_ids"] = [REF_ID]
+    target(document, path[:-1])[path[-1]] = invalid
+    with pytest.raises(ValidationError):
+        validator.validate(document)
+
+
+@pytest.mark.parametrize(
+    "invalid", ["yesterday", "2026-02-30T00:00:00Z", "2026-11-01T09:30:00", False]
+)
+def test_observation_time_requires_nullable_rfc3339(
+    validator: Draft202012Validator, document: dict[str, Any], invalid: Any
+) -> None:
+    document["entries"][0]["access"]["observed_at"] = invalid
+    with pytest.raises(ValidationError):
+        validator.validate(document)
+
+
+@pytest.mark.parametrize(
+    ("path", "invalid"),
+    [
+        ((), None),
+        (("repository_ref_ids",), {}),
+        (("repository_ref_ids",), [None]),
+        (("chokepoints",), None),
+        (("chokepoints",), [1]),
+        (("entries",), None),
+        (("entries",), {}),
+        (("entries",), [None]),
+        (("entries", 0, "repository_id"), None),
+        (("entries", 0, "repository"), None),
+        (("entries", 0, "sha"), False),
+        (("entries", 0, "priority_tier"), "urgent"),
+        (("entries", 0, "route"), "unknown"),
+        (("entries", 0, "reason"), 1),
+        (("entries", 0, "cursor"), None),
+        (("entries", 0, "access"), None),
+        (("entries", 0, "access", "repository_id"), None),
+        (("entries", 0, "access", "url"), False),
+        (("entries", 0, "access", "archived"), 0),
+        (("entries", 0, "access", "dormant"), "false"),
+        (("entries", 0, "access", "pushed_at"), 1),
+        (("entries", 0, "access", "unobserved"), False),
+        (("entries", 0, "access", "reason"), []),
+        (("entries", 0, "scan"), []),
+        (("entries", 0, "scan", "repository_ref_id"), None),
+        (("entries", 0, "scan", "basis"), 1),
+        (("entries", 0, "scan", "current"), 0),
+    ],
+)
+def test_structured_types_and_closed_vocabularies_are_enforced(
+    validator: Draft202012Validator,
+    document: dict[str, Any],
+    path: tuple[str | int, ...],
+    invalid: Any,
+) -> None:
+    if path:
+        target(document, path[:-1])[path[-1]] = invalid
+    else:
+        document = invalid
+    with pytest.raises(ValidationError):
+        validator.validate(document)

views_schema/v1/pqc-worklist.schema.json

@@ -0,0 +1,92 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://example.com/traust-contracts/views_schema/v1/pqc-worklist.schema.json",
+  "title": "PQC worklist view",
+  "description": "Ephemeral derived Insight ViewModel from PqcService.worklist, using recorded repository liveness and scan state. Not a persisted artifact or SQL projection. Scope selection, ranking and routing belong to Core; unknown state does not establish readiness.",
+  "type": "object",
+  "required": ["product_id", "repository_ref_ids", "chokepoints", "entries"],
+  "additionalProperties": false,
+  "properties": {
+    "product_id": {"type": ["string", "null"], "format": "uuid"},
+    "repository_ref_ids": {
+      "description": "Null selects all refs in the product scope; an empty array selects none. Core intersects explicit refs with the product's shipped refs.",
+      "type": ["array", "null"],
+      "items": {"type": "string", "format": "uuid"}
+    },
+    "chokepoints": {
+      "description": "Caller-supplied repository URL fragments, not persisted configuration. Strings remain raw untrusted data, not authored Markdown or HTML.",
+      "type": "array",
+      "items": {"type": "string"}
+    },
+    "entries": {
+      "type": "array",
+      "items": {"$ref": "#/$defs/entry"}
+    }
+  },
+  "$defs": {
+    "entry": {
+      "type": "object",
+      "required": [
+        "repository_id", "repository", "repository_ref_id", "sha", "priority_tier",
+        "route", "access", "reason", "scan", "cursor"
+      ],
+      "additionalProperties": false,
+      "properties": {
+        "repository_id": {"type": "string", "format": "uuid"},
+        "repository": {
+          "type": "string",
+          "description": "Raw untrusted repository value; no new URL restrictions are imposed."
+        },
+        "repository_ref_id": {"type": ["string", "null"], "format": "uuid"},
+        "sha": {"type": ["string", "null"]},
+        "priority_tier": {"type": "string", "enum": ["chokepoint", "crypto-cwe", "tail"]},
+        "route": {"type": "string", "enum": ["source", "sbom-only", "access-request"]},
+        "access": {"$ref": "#/$defs/liveness"},
+        "reason": {"type": ["string", "null"]},
+        "scan": {"anyOf": [{"$ref": "#/$defs/scan_state"}, {"type": "null"}]},
+        "cursor": {
+          "type": "string",
+          "description": "Producer ordering cursor, composed by Core as <tier-index>|<repository-url>; the repository portion remains raw data."
+        }
+      }
+    },
+    "liveness": {
+      "type": "object",
+      "required": [
+        "repository_id", "url", "observed_at", "archived", "pushed_at", "dormant",
+        "unobserved", "reason"
+      ],
+      "additionalProperties": false,
+      "properties": {
+        "repository_id": {"type": "string", "format": "uuid"},
+        "url": {"type": "string"},
+        "observed_at": {"type": ["string", "null"], "format": "date-time"},
+        "archived": {"type": ["boolean", "null"]},
+        "pushed_at": {
+          "type": ["string", "null"],
+          "description": "Raw recorded value, including historical non-RFC3339 strings."
+        },
+        "dormant": {"type": ["boolean", "null"]},
+        "unobserved": {
+          "type": ["string", "null"],
+          "description": "Recorded ComparisonStatus, preserving historical strings; null is unknown."
+        },
+        "reason": {"type": ["string", "null"]}
+      }
+    },
+    "scan_state": {
+      "type": "object",
+      "required": ["repository_ref_id", "assessment_id", "basis", "current"],
+      "additionalProperties": false,
+      "properties": {
+        "repository_ref_id": {"type": "string", "format": "uuid"},
+        "assessment_id": {"type": ["string", "null"], "format": "uuid"},
+        "basis": {"type": ["string", "null"]},
+        "current": {
+          "type": ["boolean", "null"],
+          "description": "Recorded scan currency; null is unknown, distinct from measured false and never inferred readiness."
+        }
+      }
+    }
+  }
+}