Migration 0005: keep pre-rule-engine agentic controls as unsupported rules

audits/agentic/rules.py

@@ -73,6 +73,8 @@ def validate_rules(rules: Any) -> list[RuleError]:
             errors.append(RuleError(index, "", "each rule must be an object"))
         elif rule.get("type") != "check":
             errors.append(RuleError(index, str(rule.get("type", "")), "rules must be custom checks"))
+        elif set(rule) == {"type", "check", "legacy"} and rule["check"] == "unsupported" and isinstance(rule["legacy"], dict):
+            continue  # kept by migration 0004; the engine reports it as INCONCLUSIVE
         else:
             errors.extend(RuleError(index, "check", problem) for problem in validate_check(rule))
     return errors

audits/agentic/tests/test_legacy_control_migration.py

@@ -0,0 +1,49 @@
+"""Agentic controls stored before the rule engine are kept as unsupported rules; none is dropped."""
+import importlib
+
+from audits.agentic.checks.rules import rule_checks
+from audits.agentic.expectations import validate_agentic_metadata
+from audits.agentic.trajectory import normalize
+
+migration = importlib.import_module("scenarios.migrations.0005_convert_legacy_agentic_controls")
+
+_BASE = {"schema_version": 2, "trace": {"required": True}, "enforcement": {"mode": "gating"}}
+
+# D04 from infra/fixtures/agentic_demo_catalog_v3.json as it was before #65.
+_D04 = {**_BASE,
+        "tools": {"expected": [{"name": "acme_get_order", "min_calls": 1,
+                                "arguments": {"order_id": "ACME-1004"}, "arguments_mode": "any"}]},
+        "retrieval": {"required": True, "sources": ["Acme Retail Policy"]},
+        "policy": {"read_only": True},
+        "semantic_judge": {"enabled": True, "criteria": ["tool_result_handling", "grounding"]}}
+
+
+def test_a_scenario_stored_before_the_rule_engine_validates_after_the_migration():
+    metadata = migration.convert_metadata({"agentic": _D04})
+    expectations = validate_agentic_metadata(metadata)
+    assert expectations.rules == metadata["agentic"]["rules"]
+    assert expectations.semantic_judge == _D04["semantic_judge"]
+
+
+def test_each_legacy_section_is_kept_unchanged_and_reported_inconclusive():
+    rules = migration.convert_metadata({"agentic": _D04})["agentic"]["rules"]
+    assert [rule["legacy"] for rule in rules] == [
+        {"type": "policy", "policy": _D04["policy"]},
+        {"type": "retrieval", "retrieval": _D04["retrieval"]},
+        {"type": "tools", "tools": _D04["tools"]},
+    ]
+    trace = normalize([{"span_id": "t1", "name": "execute_tool", "attributes": {
+        "gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": "acme_get_order"}}])
+    assert {check.status for check in rule_checks(trace, rules)} == {"INCONCLUSIVE"}
+
+
+def test_rules_already_stored_come_first():
+    stored = {"type": "check", "select": {"kind": "tool", "where": []}, "check": "none"}
+    rules = migration.convert_metadata({"agentic": {**_BASE, "rules": [stored], "budgets": {"max_errors": 0}}})["agentic"]["rules"]
+    assert rules[0] == stored and rules[1]["legacy"] == {"type": "budgets", "budgets": {"max_errors": 0}}
+
+
+def test_metadata_in_the_current_shape_is_unchanged():
+    metadata = {"agentic": {**_BASE, "rules": [
+        {"type": "check", "select": {"kind": "tool", "where": []}, "check": "none"}]}}
+    assert migration.convert_metadata(metadata) == metadata

audits/agentic/tests/test_rule_migration.py

@@ -38,3 +38,22 @@ def test_all_of_groups_are_flattened_into_separate_rules():
 def test_metadata_without_rules_is_unchanged():
     metadata = {"agentic": {"trace": {"required": True}}}
     assert migration.convert_metadata(metadata) == metadata
+
+
+def test_unsupported_rules_kept_by_the_migration_pass_validation_and_are_inconclusive():
+    from audits.agentic.checks.rules import rule_checks
+    from audits.agentic.expectations import validate_agentic_metadata
+    from audits.agentic.trajectory import normalize
+
+    metadata = migration.convert_metadata({"agentic": {"schema_version": 2, "enforcement": {}, "rules": [
+        {"type": "precedes", "first": "a", "then": "b"}]}})
+    converted = metadata["agentic"]["rules"]
+    assert validate_agentic_metadata(metadata).rules == converted
+    trace = normalize([{"span_id": "t1", "name": "execute_tool", "attributes": {
+        "gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": "a"}}])
+    assert [check.status for check in rule_checks(trace, converted)] == ["INCONCLUSIVE"]
+
+
+def test_only_the_migration_shape_of_an_unsupported_rule_skips_validation():
+    junk = {"type": "check", "check": "unsupported", "select": "junk"}
+    assert validate_rules([junk]) != []

scenarios/migrations/0005_convert_legacy_agentic_controls.py

@@ -0,0 +1,57 @@
+"""Keep agentic controls stored before the rule engine as unsupported rules.
+
+Revisions saved before #65 keep their controls in the sections ``tools``,
+``retrieval``, ``budgets``, ``trajectory``, ``policy``, ``rerank``, ``handoffs``,
+``approvals``, ``guardrails`` and ``state``. The metadata validation now rejects
+those sections, so such a revision could neither be evaluated nor saved. Each
+section is moved, unchanged, into an ``unsupported`` rule of the shape migration
+0004 writes, which the engine reports as INCONCLUSIVE, so no control is silently
+passed or dropped. Nothing is converted to a custom check: the old checks and the
+custom checks do not always agree on PASS and FAIL.
+"""
+from django.db import migrations
+
+_RULE_KEYS = {"schema_version", "trace", "goal", "semantic_judge", "enforcement", "rules"}
+
+
+def convert_metadata(metadata):
+    agentic = (metadata or {}).get("agentic")
+    if not isinstance(agentic, dict) or not set(agentic) - _RULE_KEYS:
+        return metadata
+    kept = {key: value for key, value in agentic.items() if key in _RULE_KEYS}
+    legacy = [{"type": "check", "check": "unsupported", "legacy": {"type": key, key: agentic[key]}}
+              for key in sorted(set(agentic) - _RULE_KEYS)]
+    return {**metadata, "agentic": {**kept, "rules": list(kept.get("rules") or []) + legacy}}
+
+
+def convert_revisions(apps, schema_editor):
+    from infra.hashing import scenario_revision_hash
+
+    ScenarioRevision = apps.get_model("scenarios", "ScenarioRevision")
+    for revision in ScenarioRevision.objects.exclude(metadata={}).iterator():
+        metadata = convert_metadata(revision.metadata)
+        if metadata == revision.metadata:
+            continue
+        revision.metadata = metadata
+        revision.content_hash = scenario_revision_hash(
+            description=revision.description,
+            expected_behavior=revision.expected_behavior,
+            test_prompt=revision.test_prompt,
+            severity_ceiling=revision.severity_ceiling,
+            documents=revision.documents or [],
+            file_uri=revision.file_uri,
+            decision=revision.decision,
+            metadata=metadata,
+        )
+        revision.save(update_fields=["metadata", "content_hash"])
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("scenarios", "0004_convert_rules_to_primitives"),
+    ]
+
+    operations = [
+        migrations.RunPython(convert_revisions, migrations.RunPython.noop),
+    ]