fix: keep qualified bundles consistent through sanitization

openadapt_flow/qualification.py

@@ -5478,6 +5478,11 @@ def save_qualified_workflow(
             raise QualificationError(
                 "governed authorization template changed the bundle content digest"
             )
+    workflow_py = Path(bundle_dir) / "workflow.py"
+    if workflow_py.is_file():
+        from openadapt_flow.compiler.codegen import render_workflow_py
+
+        workflow_py.write_text(render_workflow_py(workflow), encoding="utf-8")
     return path
 
 

openadapt_flow/sanitized_artifact.py

@@ -310,9 +310,21 @@ def _bundle_string_is_structural(path: tuple[str | int, ...], value: str) -> boo
     key = path[-1] if path and isinstance(path[-1], str) else ""
     if key.endswith(("_sha256", "_digest", "_hash", "_version")):
         return True
+    if key == "sha256" and re.fullmatch(r"[0-9a-f]{64}", value):
+        return True
     if key.endswith("_at") and value:
         return True
-    if key in {"recording_id", "id"} and _OPAQUE_MACHINE_ID.fullmatch(value):
+    if key in {"recording_id", "id", "step_id"} and _OPAQUE_MACHINE_ID.fullmatch(value):
+        return True
+    if key.endswith("_id") and re.fullmatch(
+        r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}", value, re.IGNORECASE
+    ):
+        return True
+    if key == "environment_observer_id" and re.fullmatch(
+        r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", value
+    ):
+        return True
+    if key == "relation" and value in {"left_of", "right_of", "above", "below"}:
         return True
     if key in {"c", "r", "phash", "salt", "structured"} and re.fullmatch(
         r"[0-9a-f]{16,}", value

tests/test_qualification_project.py

@@ -2448,11 +2448,14 @@ def test_minimum_effect_tier_versions_round_trips_and_invalidates_certification(
 def test_full_campaign_certifies_through_existing_policy_and_round_trips(
     tmp_path: Path,
 ) -> None:
+    from openadapt_flow.compiler.codegen import render_workflow_py
+
     workflow = _workflow()
     bundle = tmp_path / "bundle"
     (bundle / "templates").mkdir(parents=True)
     (bundle / "templates" / "save.png").write_bytes(_qualification_visual_fixture()[1])
     workflow.save(bundle)
+    (bundle / "workflow.py").write_text(render_workflow_py(workflow), encoding="utf-8")
     workflow = Workflow.load(bundle)
     _configure(workflow, tier=VerificationTier.INDEPENDENT_SYSTEM)
     evidence_root = tmp_path / "evidence"
@@ -2472,6 +2475,9 @@ def test_full_campaign_certifies_through_existing_policy_and_round_trips(
 
     save_qualified_workflow(workflow, bundle)
     loaded = Workflow.load(bundle)
+    assert (bundle / "workflow.py").read_text(encoding="utf-8") == (
+        render_workflow_py(loaded)
+    )
     assert loaded.qualification is not None
     assert loaded.qualification.schema_version == "openadapt.qualification-project/v1"
     assert loaded.qualification.last_certification is not None

tests/test_sanitized_artifact.py

@@ -32,6 +32,7 @@
     APPROVAL_NAME,
     MANIFEST_NAME,
     SanitizationError,
+    _bundle_string_is_structural,
     _review_content_length,
     _valid_review_host,
     add_manual_image_redaction,
@@ -81,6 +82,24 @@ def scrub_image(self, image, fill_color=None):
         return image
 
 
+@pytest.mark.parametrize(
+    ("path", "value"),
+    [
+        (("qualification", "cases", 0, "evidence", 0, "sha256"), "a" * 64),
+        (("steps", 0, "id"), "step_000"),
+        (("qualification", "identity_policies", "step_000", "step_id"), "step_000"),
+        (("qualification", "project_id"), "f6d239d0-0b5e-440f-b794-bbe69cc01115"),
+        (
+            ("qualification", "environment", "environment_observer_id"),
+            "backend:PlaywrightBackend",
+        ),
+        (("steps", 0, "anchor", "landmarks", 0, "relation"), "left_of"),
+    ],
+)
+def test_bundle_machine_contract_strings_are_not_sent_to_phi_ner(path, value):
+    assert _bundle_string_is_structural(path, value)
+
+
 @pytest.fixture(autouse=True)
 def _scrubber(tmp_path, monkeypatch):
     privacy.set_text_scrubber(FakeScrubber())