feat(continuous-ops): fail loudly on what's required: one preflight, a lean config (XWING-2440)

pyproject.toml

@@ -20,7 +20,7 @@ traust-engine = "traust_engine.ops.cli:main"
 dev = ["pytest>=9.1.1", "pytest-xdist>=3", "ruff>=0.6", "vulture>=2.16"]
 
 [tool.uv.sources]
-traust-core = { git = "https://github.com/traust-security/traust-core.git", rev = "acf67b8a4409ed7afa1b79d7954636b5b169fa64" }
+traust-core = { git = "https://github.com/traust-security/traust-core.git", rev = "282ebbee2ac065ea7f15456d2bafb4f63077c8b2" }
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/traust_engine"]

src/traust_engine/ops/cli.py

@@ -27,7 +27,7 @@
 
 from traust_engine.ops.loop import EngineLoop
 from traust_engine.ops.pack import Pack
-from traust_engine.ops.runner import Runner
+from traust_engine.ops.runner import Runner, runnable
 from traust_engine.ops.shadow import (
     Diff,
     Tick,
@@ -134,7 +134,7 @@ def _diff(args: argparse.Namespace, core: Context, pack: Pack) -> int:
     if args.routes:
         raw = json.loads(args.routes.read_text(encoding="utf-8"))
         rows = raw.get("decisions", []) if isinstance(raw, dict) else raw
-        lanes = (lane for a in pack.assets.values() for lane in a.trigger.lanes)
+        lanes = iter(pack.lanes)
         same, different = compare_routes(ticks, rows, repository_urls(services), lanes)
     report = Diff(
         since=since,
@@ -220,10 +220,21 @@ def main(argv: Sequence[str] | None = None) -> int:
         if args.models:
             core.register_model_catalog(args.models)
         pack = Pack.load(args.pack, *args.pin, *([args.feeds] if args.feeds else []))
+        if args.op in ("run", "shadow", "claim"):
+            # preflight checks tool requirements against core's shipped pins
+            core.register_toolchain()
         if args.op in ("run", "shadow"):
+            if pack.route_on is not None and not args.router:
+                raise ConfigError(
+                    f"pack {pack.pack} sets route_on {pack.route_on}, but no --router was "
+                    "given -- pass --router module:Class, or drop route_on from the pack"
+                )
+            core.preflight(pack.requirements())
             return _passes(args, core, pack, shadow=args.op == "shadow")
-        ops = {"record": _record, "diff": _diff, "claim": _claim}
-        return ops[args.op](args, core, pack)
+        if args.op == "claim":
+            core.preflight(pack.assets[name].requires for name in runnable(pack))
+            return _claim(args, core, pack)
+        return {"record": _record, "diff": _diff}[args.op](args, core, pack)
     except (OSError, TraustError) as e:
         print(f"ERROR: {e}", file=sys.stderr)
         return 2

src/traust_engine/ops/controller.py

@@ -47,7 +47,6 @@ class TickReport(Model):
     refused_again: int = 0
     stale: dict[AssetName, int] = pydantic.Field(default_factory=dict)
     overdue: dict[AssetName, int] = pydantic.Field(default_factory=dict)
-    unrunnable: tuple[AssetName, ...] = ()
 
 
 class Releases(Protocol):
@@ -99,15 +98,12 @@ def due(
         """What a tick would plan (routed orders, then every due asset), and the stale-upstream
         stops it would record on the feed, writing nothing: the shadow run's view of the loop."""
         now = self._clock.now()
-        unrunnable = self._pack.unrunnable()
-        if unrunnable:
-            log.warning("not planned, no run or model.role: %s", ", ".join(unrunnable))
-        requests, routed, unrouted, refused_again = self._route(now, unrunnable)
+        requests, routed, unrouted, refused_again = self._route(now)
         planned, stale, stops = 0, {}, []
         for name in self._pack.topo_order():
             spec = self._pack.assets[name]
             since = spec.trigger.since(now)
-            if since is None or name in unrunnable:
+            if since is None or not spec.enabled:
                 continue
             subjects: Collection[UUID | None] = repositories if spec.partitions else (None,)
             busy = self._ops.subjects(name, OPEN_ORDER_STATES) | self._ops.subjects(
@@ -129,15 +125,14 @@ def due(
                 planned += 1
         for name, count in stale.items():
             log.warning("%s: %d subjects wait on stale upstreams; not planned", name, count)
-        overdue = self._overdue(repositories, now, unrunnable)
+        overdue = self._overdue(repositories, now)
         report = TickReport(
             planned=planned,
             routed=routed,
             unrouted=unrouted,
             refused_again=refused_again,
             stale=stale,
             overdue=overdue,
-            unrunnable=unrunnable,
         )
         return report, requests, stops
 
@@ -157,14 +152,12 @@ def _stopped(
             )
         return cache[dep]
 
-    def _overdue(
-        self, repositories: Collection[UUID], now: datetime, unrunnable: Collection[AssetName]
-    ) -> dict[AssetName, int]:
+    def _overdue(self, repositories: Collection[UUID], now: datetime) -> dict[AssetName, int]:
         """Subjects past `freshness.fail` since their last materialization (or never made);
-        unrunnable assets are reported as such instead."""
+        a disabled asset is not one the deployment is failing to keep fresh."""
         overdue = {}
         for name, spec in self._pack.assets.items():
-            if spec.freshness.fail is None or name in unrunnable:
+            if spec.freshness.fail is None or not spec.enabled:
                 continue
             subjects = set(repositories) if spec.partitions else {None}
             fresh = self._ops.subjects(name, MATERIALIZED, now - spec.freshness.fail)
@@ -173,16 +166,16 @@ def _overdue(
                 log.warning("%s: %d subjects past freshness.fail", name, count)
         return overdue
 
-    def _route(
-        self, now: datetime, unrunnable: Collection[AssetName]
-    ) -> tuple[list[OrderRequest], int, int, int]:
+    def _route(self, now: datetime) -> tuple[list[OrderRequest], int, int, int]:
         """Re-route repositories whose `route_on` asset materialized, whose routed order was
-        refused, or that a recorded release ships, since the last tick; each route with a lane the
-        pack runs becomes an order for that lane's asset, except a route back into a lane that
-        refused it."""
+        refused, or that a recorded release ships, since the last tick; each route with a lane
+        the pack runs becomes an order for that lane's asset, except a route back into a lane
+        that refused it. A lane the pack runs no asset for is a pack error: the pass fails
+        before it plans anything."""
         source = self._pack.route_on
         if self._router is None or self._state_of is None or source is None:
             return [], 0, 0, 0
+        before = self._cursor
         changed, refusals, releases = self._changes(source, now)
         requests: list[OrderRequest] = []
         unknown: set[LaneName] = set()
@@ -209,7 +202,7 @@ def _route(
                 except ConfigError:
                     unknown.add(route.lane)
                     continue
-                if asset not in unrunnable:
+                if self._pack.assets[asset].enabled:
                     requests.append(
                         OrderRequest(
                             asset=asset,
@@ -221,7 +214,11 @@ def _route(
                     )
             routed += 1
         if unknown:
-            log.warning("routed lanes no pack asset runs: %s", ", ".join(sorted(unknown)))
+            self._cursor = before
+            raise ConfigError(
+                f"pack {self._pack.pack} runs no asset for lane(s) {', '.join(sorted(unknown))}: "
+                "give each lane an asset (trigger.lanes), or fix the router"
+            )
         return requests, routed, unrouted, refused_again
 
     def _changes(

src/traust_engine/ops/pack.py

@@ -16,7 +16,7 @@
 from traust_core.v1.models.base import Dto
 from traust_core.v1.models.common import Sha256
 from traust_core.v1.models.integrity import sha256_hex
-from traust_core.v1.models.operations import AssetName, ConfigDigest, LaneName
+from traust_core.v1.models.operations import AssetName, ConfigDigest, LaneName, Requirement
 
 from traust_engine.ops.cron import Cron
 
@@ -85,6 +85,8 @@ class ModelRole(Dto):
 class AssetSpec(Dto):
     run: tuple[str, ...] = ()
     model: ModelRole | None = None
+    enabled: bool = True
+    requires: Requirement = Requirement()
     partitions: Literal["dynamic:repo"] | None = None
     deps: tuple[AssetName, ...] = ()
     priority: int = 100
@@ -97,6 +99,12 @@ class Pack(Dto):
     pack: str
     route_on: AssetName | None = None
     assets: dict[AssetName, AssetSpec]
+
+    @property
+    def lanes(self) -> tuple[LaneName, ...]:
+        """The lanes the pack actually runs: an enabled asset's, never a disabled one's."""
+        return tuple(lane for a in self.assets.values() if a.enabled for lane in a.trigger.lanes)
+
     digest: Sha256 | None = None  # sha256 of the file's bytes, set by `load`: the pack version
     pins: tuple[ConfigDigest, ...] = ()
 
@@ -108,6 +116,28 @@ def _graph(self) -> Pack:
         lanes = [lane for a in self.assets.values() for lane in a.trigger.lanes]
         if len(lanes) != len(set(lanes)):
             raise ValueError(f"a lane belongs to one asset: {sorted(lanes)}")
+        if unmade := sorted(
+            n for n, a in self.assets.items() if a.enabled and not a.run and a.model is None
+        ):
+            raise ValueError(
+                f"assets with neither run nor model.role: {unmade} -- no runner can make them "
+                "(enabled: false is the exemption)"
+            )
+        if self.route_on is not None and not self.assets[self.route_on].enabled:
+            raise ValueError(f"route_on {self.route_on!r} is disabled: routing would never run")
+        if stalled := sorted(
+            {
+                d
+                for a in self.assets.values()
+                if a.enabled
+                for d in a.deps
+                if not self.assets[d].enabled
+            }
+        ):
+            raise ValueError(
+                f"enabled assets depend on disabled ones: {stalled} -- the dependents are "
+                "withheld upstream-stale every window; enable the dependency or the dependent"
+            )
         if self.route_on is not None and (
             self.route_on not in names or self.assets[self.route_on].trigger.every is None
         ):
@@ -118,9 +148,10 @@ def _graph(self) -> Pack:
             raise ValueError(f"asset deps form a cycle: {e.args[1]}") from e
         return self
 
-    def unrunnable(self) -> tuple[AssetName, ...]:
-        """Assets with neither a command nor a model role: no runner could claim their orders."""
-        return tuple(n for n, a in self.assets.items() if not a.run and a.model is None)
+    def requirements(self) -> tuple[Requirement, ...]:
+        """Each enabled asset's requirements, what `Context.preflight` checks before a pass;
+        `enabled: false` is the only exemption."""
+        return tuple(a.requires for a in self.assets.values() if a.enabled)
 
     def topo_order(self) -> list[AssetName]:
         return list(TopologicalSorter({n: a.deps for n, a in self.assets.items()}).static_order())

src/traust_engine/ops/runner.py

@@ -56,9 +56,9 @@ def _by_exit(done: subprocess.CompletedProcess[str]) -> JobResult:
 
 
 def runnable(pack: Pack) -> tuple[AssetName, ...]:
-    """The assets this runner makes: those with a command; a model role without one has nothing
-    to start its session with."""
-    return tuple(name for name, spec in pack.assets.items() if spec.run)
+    """The assets this runner makes: enabled, with a command; a model role without one has
+    nothing to start its session with."""
+    return tuple(name for name, spec in pack.assets.items() if spec.enabled and spec.run)
 
 
 class UnresolvedParam(TraustError):

tests/fixtures/config/traust.yaml

@@ -0,0 +1,9 @@
+# The suite's config home passes core's preflight completeness audit: locations and the
+# ledger key are the fields it warns about (conftest copies this directory to
+# $TRAUST_CONFIG_HOME; preflight promotes the audit's warnings to findings).
+locations:
+  workspace: ws
+  analysis_results: results
+  progress_tracker: tracker
+ledger:
+  public_key_path: ledger-signing-key.pub

tests/test_engine_loop.py

@@ -15,7 +15,7 @@
 import yaml
 from traust_core.v1.context import Config, Context
 from traust_core.v1.context.context import Services
-from traust_core.v1.errors import NotFoundError
+from traust_core.v1.errors import ConfigError, NotFoundError
 from traust_core.v1.models.artifacts import RunManifestArtifact
 from traust_core.v1.models.common import every_item
 from traust_core.v1.models.inventory import (
@@ -258,19 +258,15 @@ def test_routing_the_same_repository_again_plans_no_second_order(tmp_path: Path)
     assert [o[0] for o in first] == ["diff_scan", "iac_lane"]
 
 
-def test_a_route_to_an_unrunnable_lane_plans_nothing_for_it(tmp_path: Path) -> None:
-    pack = _iac_pack(tmp_path, run=None)
-    assert "iac_lane" in pack.unrunnable()
+def test_a_route_to_a_disabled_assets_lane_plans_nothing_for_it(tmp_path: Path) -> None:
+    pack = _iac_pack(tmp_path, enabled=False)
     services, _, repo = _routed_once(tmp_path, pack, TableAndAdditiveRouter())
     assert _routed_orders(services) == [("diff_scan", "diff-scan", "rule-5", repo)]
 
 
-def test_a_route_to_a_lane_no_asset_runs_plans_nothing_for_it(
-    tmp_path: Path, caplog: pytest.LogCaptureFixture
-) -> None:
-    services, _, repo = _routed_once(tmp_path, Pack.load(PACK), TableAndAdditiveRouter())
-    assert _routed_orders(services) == [("diff_scan", "diff-scan", "rule-5", repo)]
-    assert "routed lanes no pack asset runs: iac-lane" in caplog.text
+def test_a_route_to_a_lane_no_asset_runs_fails_the_pass(tmp_path: Path) -> None:
+    with pytest.raises(ConfigError, match="runs no asset for lane"):
+        _routed_once(tmp_path, Pack.load(PACK), TableAndAdditiveRouter())
 
 
 def test_a_repository_the_router_leaves_alone_gets_no_routed_order(tmp_path: Path) -> None:
@@ -548,10 +544,16 @@ def test_admission_records_the_lane_verdict_and_deferred_orders_spend_nothing(
     }
 
 
-def test_an_asset_no_runner_can_make_is_reported_not_planned(tmp_path: Path) -> None:
+def test_a_disabled_asset_is_not_planned_or_reported(tmp_path: Path) -> None:
     raw = yaml.safe_load(PACK.read_text())
-    del raw["assets"]["repo_state"]["run"]
-    raw["assets"]["repo_state"]["freshness"] = {"fail": "48h"}
+    raw["assets"]["repo_state"] |= {
+        "enabled": False,
+        "freshness": {"fail": "48h"},
+        "requires": {"facts": ["sla-policy"]},
+    }
+    raw.pop("route_on")
+    raw["assets"]["dependency_graph"]["enabled"] = False
+    raw["assets"]["diff_scan"]["enabled"] = False
     path = tmp_path / "assets.yaml"
     path.write_text(yaml.safe_dump(raw))
     clock = Clock()
@@ -560,30 +562,11 @@ def test_an_asset_no_runner_can_make_is_reported_not_planned(tmp_path: Path) ->
 
     report = Controller(Pack.load(path), services.ops, clock).tick(repos)
 
-    assert report.unrunnable == ("repo_state",)
     assert report.planned == 1  # feeds_cache only
     assert "repo_state" not in report.overdue
     assert _by_asset(_orders(services, OrderState.PLANNED)) == {"feeds_cache": 1}
 
 
-def test_a_route_to_a_lane_no_runner_can_make_plans_nothing(tmp_path: Path) -> None:
-    raw = yaml.safe_load(PACK.read_text())
-    del raw["assets"]["diff_scan"]["run"]
-    path = tmp_path / "assets.yaml"
-    path.write_text(yaml.safe_dump(raw))
-    clock = Clock()
-    services = _context(tmp_path, clock).services()
-    _repositories(services, 2)
-    loop = EngineLoop(services, clock, Pack.load(path), LaneRouter(), state_of=lambda r: r)
-    loop.run_once()
-    clock.at += timedelta(minutes=10)
-    _drain(services)
-
-    report = loop.run_once()
-    assert report.tick.routed == 2
-    assert "diff_scan" not in _by_asset(_orders(services, OrderState.QUEUED, OrderState.PLANNED))
-
-
 def test_the_run_manifest_names_the_toolchain_pins_next_to_the_pack(tmp_path: Path) -> None:
     pins = tmp_path / "external-tools.yaml"
     pins.write_text("tools: {grype: {version: 0.104.0}}\n")
@@ -820,3 +803,23 @@ def test_a_restart_does_not_replay_releases_routed_before_it(tmp_path: Path) ->
     router.releases.clear()
     restarted = EngineLoop(services, clock, Pack.load(PACK), router, state_of=lambda r: r)
     assert restarted.run_once().tick.routed == 0 and router.releases == {}
+
+
+def test_a_corrected_router_retry_reads_the_events_it_failed_on(tmp_path: Path) -> None:
+    """The failed pass puts the feed cursor back, so a corrected router on the same loop
+    routes the repositories the failed pass read (it does not skip past their events)."""
+    clock = Clock()
+    services = _context(tmp_path, clock).services()
+    _repositories(services, 1)
+    loop = EngineLoop(
+        services, clock, Pack.load(PACK), TableAndAdditiveRouter(), state_of=lambda r: r
+    )
+    loop.run_once()
+    clock.at += timedelta(minutes=10)
+    _drain(services)
+    with pytest.raises(ConfigError, match="runs no asset for lane"):
+        loop.run_once()
+    loop._controller._router = LaneRouter(frozenset())
+    report = loop.run_once()
+    assert report.tick.routed == 1
+    assert _routed_orders(services)