Sync from RecastEngine-Pro: lint and types for the carried forward-workload library

src/recast/phases.py

@@ -35,7 +35,6 @@
 from recast.observe import RunEventAction, RunEventEntity, RunObserver
 from recast.plugins.recipe import Recipe, Stage
 from recast.registry import REGISTRY, Registry
-from recast.workspace_resources import resolve_workspace_resources
 from recast.run import (
     StageOutcome,
     UnitRun,
@@ -46,6 +45,7 @@
     output_root,
     run_recipe,
 )
+from recast.workspace_resources import resolve_workspace_resources
 
 __all__ = [
     "BindingCheck",

src/recast/verify/forward_workload.py

@@ -118,16 +118,16 @@
     "ChainIdentity",
     "ForwardWorkloadReport",
     "InheritedObjectives",
-    "TimingBaselineReference",
-    "TransformationBoundaryReceipt",
     "RecordedSourceBoundaryReceipt",
     "RecordedSourceEvidence",
     "RecordedSourceFile",
-    "recorded_source_coverage_digest",
+    "TimingBaselineReference",
+    "TransformationBoundaryReceipt",
     "WorkloadSubject",
     # A caller building an adapter configuration needs to compute the same
     # source digests the engine will check it against.
     "adapter_source_digest",
+    "recorded_source_coverage_digest",
     "verify_bundle_forward_objectives",
 ]
 
@@ -1459,11 +1459,13 @@ def _require_anchor(
 
     if anchor.kind == "recorded-tape" and anchor.tape_digest != declaration.fixtures.tape_digest:
         raise ConfigError("workload anchor tape differs from the declared observation authority")
-    if type(receipt) is RecordedSourceBoundaryReceipt:
+    if isinstance(receipt, RecordedSourceBoundaryReceipt):
         if links:
             raise ConfigError("a recorded source boundary cannot claim intermediate links")
         if declaration.boundary_receipt_digest != receipt.digest():
-            raise ConfigError("the recorded source receipt is not the one the declaration authorized")
+            raise ConfigError(
+                "the recorded source receipt is not the one the declaration authorized"
+            )
         if source_root is None:
             raise ConfigError("a recorded source boundary needs the actual source root")
         engine = candidate.saved.engine
@@ -1912,7 +1914,11 @@ def verify_bundle_forward_objectives(
             engine=installed_engine,
         )
         _require_anchor(
-            original_anchor, boundary_receipt, candidate, declaration, tuple(boundary_links),
+            original_anchor,
+            boundary_receipt,
+            candidate,
+            declaration,
+            tuple(boundary_links),
             recorded_source_evidence=recorded_source_evidence,
             source_root=Path(root),
         )

src/recast/verify/recorded_source.py

@@ -66,8 +66,13 @@ def __post_init__(self) -> None:
         for name in ("downstream_engine_id", "provider_id"):
             _identifier(getattr(self, name), name)
         for name in (
-            "anchor_digest", "downstream_source_artifact_digest", "recording_provenance_digest",
-            "tape_digest", "policy_digest", "observation_keys_digest", "provider_digest",
+            "anchor_digest",
+            "downstream_source_artifact_digest",
+            "recording_provenance_digest",
+            "tape_digest",
+            "policy_digest",
+            "observation_keys_digest",
+            "provider_digest",
             "profile_digest",
         ):
             _digest(getattr(self, name), name)
@@ -135,8 +140,12 @@ def __post_init__(self) -> None:
         if sum(item.size for item in self.source_files) > MAX_SOURCE_BYTES:
             raise ConfigError("recorded source evidence files exceed the total bound")
         for name in (
-            "recording_manifest_digest", "recording_run_digest", "origin_source_digest", "producer_digest",
-            "checker_digest", "coverage_digest",
+            "recording_manifest_digest",
+            "recording_run_digest",
+            "origin_source_digest",
+            "producer_digest",
+            "checker_digest",
+            "coverage_digest",
         ):
             _digest(getattr(self, name), name)
 
@@ -148,22 +157,24 @@ def recorded_source_coverage_digest(declaration: WorkloadDeclaration) -> str:
     """Scientific extent, independent of candidate entry-point reconstruction."""
 
     fixture = declaration.fixtures
-    return _hash({
-        "case": fixture.case,
-        "probe": fixture.probe,
-        "first_sequence": fixture.first_sequence,
-        "final_sequence": fixture.final_sequence,
-        "steps": fixture.steps,
-        "required_inputs": list(fixture.required_inputs),
-        "required_outputs": list(fixture.required_outputs),
-        "outputs": list(declaration.interface.outputs),
-        "static_signatures": list(declaration.interface.static_signatures),
-        "coverage_labels": list(fixture.coverage_labels),
-        "states": list(fixture.states),
-        "state_digests": dict(fixture.state_digests),
-        "observation_keys_digest": fixture.observation_keys_digest,
-        "observation_key_count": fixture.observation_key_count,
-    })
+    return _hash(
+        {
+            "case": fixture.case,
+            "probe": fixture.probe,
+            "first_sequence": fixture.first_sequence,
+            "final_sequence": fixture.final_sequence,
+            "steps": fixture.steps,
+            "required_inputs": list(fixture.required_inputs),
+            "required_outputs": list(fixture.required_outputs),
+            "outputs": list(declaration.interface.outputs),
+            "static_signatures": list(declaration.interface.static_signatures),
+            "coverage_labels": list(fixture.coverage_labels),
+            "states": list(fixture.states),
+            "state_digests": dict(fixture.state_digests),
+            "observation_keys_digest": fixture.observation_keys_digest,
+            "observation_key_count": fixture.observation_key_count,
+        }
+    )
 
 
 def require_recorded_source_files(root: Path, evidence: RecordedSourceEvidence) -> None:
@@ -180,7 +191,8 @@ def require_recorded_source_files(root: Path, evidence: RecordedSourceEvidence)
         if not path.is_file() or path.stat().st_size != item.size:
             raise ConfigError(f"recorded source file {item.path!r} differs from its inventory")
         payload = path.read_bytes()
-        if len(payload) != item.size or "sha256:" + hashlib.sha256(payload).hexdigest() != item.digest:
+        digest = "sha256:" + hashlib.sha256(payload).hexdigest()
+        if len(payload) != item.size or digest != item.digest:
             raise ConfigError(f"recorded source file {item.path!r} differs from its content digest")
 
 
@@ -205,7 +217,9 @@ def require_recorded_source_boundary(
         raise ConfigError("recorded source boundary names a different actual source artifact")
     if engine_id is None or receipt.downstream_engine_id != engine_id:
         raise ConfigError("recorded source boundary names a different installed engine")
-    if receipt.tape_digest != declaration.fixtures.tape_digest or receipt.tape_digest != anchor.tape_digest:
+    if receipt.tape_digest != declaration.fixtures.tape_digest or (
+        receipt.tape_digest != anchor.tape_digest
+    ):
         raise ConfigError("recorded source boundary names a different physical tape")
     if receipt.policy_digest != declaration.policy_digest:
         raise ConfigError("recorded source boundary names a different numerical policy")

src/recast/workspace_resources.py

@@ -36,7 +36,7 @@
 
 from recast.errors import ConfigError
 
-__all__ = ["WORKSPACE_MARKER", "workspace_resource", "resolve_workspace_resources"]
+__all__ = ["WORKSPACE_MARKER", "resolve_workspace_resources", "workspace_resource"]
 
 #: The single key that marks a workspace-relative declaration.
 WORKSPACE_MARKER = "$workspace"

tests/test_workspace_resources.py

@@ -30,7 +30,6 @@
     resolve_workspace_resources,
     workspace_resource,
 )
-
 from tests.test_phases import (  # the existing minimal plugins
     PhaseTransform,
     _engine,
@@ -103,10 +102,15 @@ def materialize(self, unit, facts, workspace, executor, config) -> OracleRef:
         digest = hashlib.sha256(payloads[0].read_bytes()).hexdigest()
         if digest != config["pin"]:
             raise ConfigError(
-                f"the recording at {directory} is {digest[:16]}, not the pinned {config['pin'][:16]}"
+                f"the recording at {directory} is {digest[:16]}, "
+                f"not the pinned {config['pin'][:16]}"
             )
-        return OracleRef(unit=unit.uid, oracle=self.name, key=self.key(unit, None, config),
-                         handle={"input_source": "recorded", "samples": []})
+        return OracleRef(
+            unit=unit.uid,
+            oracle=self.name,
+            key=self.key(unit, None, config),
+            handle={"input_source": "recorded", "samples": []},
+        )
 
 
 class RecordingRecipe(Recipe):
@@ -194,11 +198,11 @@ def test_only_the_operational_workspace_changes_nothing_portable(tmp_path: Path)
     assert first.digest() == second.digest()
     assert first.semantic_config_digest == second.semantic_config_digest
     assert first.verification_plan == second.verification_plan
-    plan_config = [
+    plan_config = next(
         dict(stage.config)
         for stage in first.verification_plan.stages
         if stage.plugin == "phase.recording"
-    ][0]
+    )
     assert dict(plan_config["dumps"]) == {WORKSPACE_MARKER: RECORDING}, (
         "the frozen plan keeps the declaration, not a path"
     )
@@ -247,7 +251,9 @@ def test_a_missing_recording_is_refused(tmp_path: Path) -> None:
 
 def test_a_different_recording_is_refused(tmp_path: Path) -> None:
     bundle, _ = transform_at(tmp_path, "lease-a")
-    report = verify_at(tmp_path, bundle, "verify-other", payload=b"# PROBE example.unit:\nINPUT: 2\n")
+    report = verify_at(
+        tmp_path, bundle, "verify-other", payload=b"# PROBE example.unit:\nINPUT: 2\n"
+    )
     assert report.accepted is False
 
 
@@ -274,8 +280,11 @@ class ExecutorResourceRecipe(RecordingRecipe):
     def stages(self, config: dict[str, Any]) -> list[Stage]:
         stages = super().stages(config)
         return [
-            Stage("executor", "phase.resource-executor",
-                  config={"scratch": workspace_resource("vendor/executor")})
+            Stage(
+                "executor",
+                "phase.resource-executor",
+                config={"scratch": workspace_resource("vendor/executor")},
+            )
             if stage.kind == "executor"
             else stage
             for stage in stages