feat(fleet): module-owned publisher lineage; reject expressions in Dagger args

.dagger/src/ci/fleet.py

@@ -5,8 +5,11 @@
 from dataclasses import dataclass
 from datetime import date
 
-from ci.fleet_policy import PolicyFinding, RepositoryExpectation, validate_repository
-from ci.github_fleet import GitHubHttpTransport, read_repository
+from ci.fleet_coverage import coverage_results
+from ci.fleet_policy import PolicyFinding, RepositoryExpectation, finding, validate_repository
+from ci.github_fleet import FleetAccessError, GitHubHttpTransport, GitHubTransport, read_repository
+
+OWNER = "hseshadr"
 
 
 @dataclass(frozen=True)
@@ -21,6 +24,8 @@ class RepositoryResult:
 def repository_expectations(include_central: bool) -> tuple[RepositoryExpectation, ...]:
     """Return the reviewed fleet contract for the current rollout phase."""
     consumers = (
+        expectation("agentic-context-service"),
+        expectation("agentic-saga"),
         expectation("almamesh", grandfathered_until=date(2026, 12, 15)),
         expectation("aml-filter", grandfathered_until=date(2026, 12, 31)),
         expectation("assay", linear_history=True, grandfathered_until=date(2026, 11, 30)),
@@ -57,15 +62,28 @@ def expectation_for(name: str) -> RepositoryExpectation:
 
 def scan_fleet(token: str, include_central: bool) -> tuple[RepositoryResult, ...]:
     """Read and evaluate each repository from authoritative exact-main evidence."""
-    transport = GitHubHttpTransport(token)
+    return scan_fleet_with(GitHubHttpTransport(token), include_central)
+
+
+def scan_fleet_with(
+    transport: GitHubTransport, include_central: bool
+) -> tuple[RepositoryResult, ...]:
+    """Prove coverage of every discovered consumer, then evaluate each reviewed one."""
+    reviewed = tuple(item.name for item in repository_expectations(True))
+    uncovered = coverage_results(transport, OWNER, reviewed)
+    coverage = tuple(RepositoryResult(item.name, "", item.findings) for item in uncovered)
     expectations = repository_expectations(include_central)
-    return tuple(scan_repository(transport, item) for item in expectations)
+    return coverage + tuple(scan_repository(transport, item) for item in expectations)
 
 
 def scan_repository(
-    transport: GitHubHttpTransport, expectation_: RepositoryExpectation
+    transport: GitHubTransport, expectation_: RepositoryExpectation
 ) -> RepositoryResult:
-    """Evaluate one exact-main repository against its reviewed contract."""
-    snapshot = read_repository(transport, "hseshadr", expectation_.name)
+    """Evaluate one repository, turning unreadable evidence into a failing finding."""
+    try:
+        snapshot = read_repository(transport, OWNER, expectation_.name)
+    except FleetAccessError as error:
+        unreadable = finding("evidence-unreadable", "github", str(error))
+        return RepositoryResult(expectation_.name, "", (unreadable,))
     findings = validate_repository(snapshot, expectation_)
     return RepositoryResult(snapshot.name, snapshot.sha, findings)

.dagger/src/ci/fleet_coverage.py

@@ -0,0 +1,101 @@
+"""Discover every Dagger consumer of the central modules and prove fleet coverage.
+
+The reviewed fleet contract (`repository_expectations`) is a hand-maintained list, so a
+new consumer silently escapes every fleet check until someone remembers to add it. This
+module closes that gap: it lists the owner's repositories, reads each default-branch
+`dagger.json`, and fails the scan for any consumer the contract does not name.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Final
+
+from pydantic.dataclasses import dataclass as validated_dataclass
+
+from ci.fleet_policy import PolicyFinding, finding
+from ci.github_fleet import (
+    BOUNDARY_CONFIG,
+    HTTP_NOT_FOUND,
+    GitHubTransport,
+    decode_source,
+    parse_content_response,
+    parse_dagger_config,
+    read_model,
+)
+
+UNCOVERED_CODE: Final = "uncovered-consumer"
+CENTRAL_PREFIXES: Final = ("github.com/hseshadr/ci/", "github.com/hseshadr/ci@")
+PAGE_SIZE: Final = 100
+
+
+@validated_dataclass(config=BOUNDARY_CONFIG)
+class OwnedRepositoryPayload:
+    """One entry of the owner's public repository listing."""
+
+    name: str
+    archived: bool
+    default_branch: str
+
+
+@dataclass(frozen=True)
+class UncoveredConsumer:
+    """One discovered consumer that the reviewed fleet contract does not name."""
+
+    name: str
+    findings: tuple[PolicyFinding, ...]
+
+
+def discover_consumers(transport: GitHubTransport, owner: str) -> tuple[str, ...]:
+    """Return every active repository whose default-branch dagger.json pins hseshadr/ci."""
+    return tuple(
+        repository.name
+        for repository in list_repositories(transport, owner)
+        if not repository.archived and consumes_central(transport, owner, repository)
+    )
+
+
+def list_repositories(transport: GitHubTransport, owner: str) -> tuple[OwnedRepositoryPayload, ...]:
+    """Read every page of the owner's repository listing, failing closed on any error."""
+    collected: list[OwnedRepositoryPayload] = []
+    page = 1
+    while True:
+        path = f"users/{owner}/repos?type=owner&per_page={PAGE_SIZE}&page={page}"
+        batch = read_model(transport, path, tuple[OwnedRepositoryPayload, ...])
+        collected.extend(batch)
+        if len(batch) < PAGE_SIZE:
+            return tuple(collected)
+        page += 1
+
+
+def consumes_central(
+    transport: GitHubTransport, owner: str, repository: OwnedRepositoryPayload
+) -> bool:
+    """Return whether one default-branch dagger.json declares a central module."""
+    base = f"repos/{owner}/{repository.name}"
+    path = f"{base}/contents/dagger.json?ref={repository.default_branch}"
+    response = transport.get(path)
+    if response.status == HTTP_NOT_FOUND:
+        return False
+    source = decode_source(parse_content_response(response, path), "dagger.json", base)
+    config = parse_dagger_config(source)
+    return any(item.source.startswith(CENTRAL_PREFIXES) for item in config.dependencies)
+
+
+def uncovered_consumers(discovered: tuple[str, ...], reviewed: tuple[str, ...]) -> tuple[str, ...]:
+    """Return discovered consumers absent from the reviewed fleet contract."""
+    return tuple(name for name in discovered if name not in reviewed)
+
+
+def coverage_results(
+    transport: GitHubTransport, owner: str, reviewed: tuple[str, ...]
+) -> tuple[UncoveredConsumer, ...]:
+    """Build one failing result per discovered consumer the fleet scan would skip."""
+    missing = uncovered_consumers(discover_consumers(transport, owner), reviewed)
+    return tuple(UncoveredConsumer(name, (uncovered_finding(name),)) for name in missing)
+
+
+def uncovered_finding(name: str) -> PolicyFinding:
+    """Name the exact fix for one consumer that escapes the fleet scan."""
+    message = f"{name} pins hseshadr/ci modules but is missing from repository_expectations"
+    return finding(UNCOVERED_CODE, "dagger.json", message)