Enforce bounded operation guardrails

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f

tests/test_bounded_operations.py

@@ -0,0 +1,167 @@
+from __future__ import annotations
+
+import pytest
+
+from activity_core.bounded_operations import (
+    BOUNDED_OPERATION_REGISTRY,
+    normalize_operation_result,
+    operation_sources,
+    validate_bounded_operations,
+)
+
+
+def _evidence_sinks(event_type: str) -> list[dict[str, str]]:
+    return [{"type": "state-hub-progress", "event_type": event_type}]
+
+
+def test_registry_names_only_the_three_accepted_operations() -> None:
+    assert set(BOUNDED_OPERATION_REGISTRY) == {
+        "sbom_nexus_ingest",
+        "forgejo_package_prune",
+        "cnpg_option_a_backup",
+    }
+
+
+def test_operation_result_projection_drops_subprocess_output() -> None:
+    projected = normalize_operation_result(
+        "cnpg_option_a_backup",
+        {
+            "overall": "pass",
+            "dumped": 3,
+            "uploaded": 3,
+            "failed": 0,
+            "script_exit_code": 0,
+            "log_tail": "must not persist",
+            "tool_output": "must not persist",
+            "credential": "must not persist",
+        },
+    )
+
+    assert projected == {
+        "kind": "cnpg_option_a_backup",
+        "overall": "pass",
+        "dry_run": False,
+        "dumped": 3,
+        "uploaded": 3,
+        "failed": 0,
+        "script_exit_code": 0,
+    }
+    assert "must not persist" not in str(projected)
+
+
+def test_valid_prune_operation_is_admitted() -> None:
+    sources = [
+        {
+            "type": "shell",
+            "query": "forgejo_package_prune",
+            "operation": "forgejo_package_prune",
+            "params": {
+                "prune_script": "/opt/railiance-platform/tools/cmd/forgejo-package-prune",
+                "live_images_file": "/opt/railiance-platform/live-images.txt",
+                "apply": True,
+                "max_versions": 3,
+                "evidence_sinks": _evidence_sinks("forgejo_package_prune"),
+            },
+        }
+    ]
+
+    validate_bounded_operations(sources, [])
+    assert operation_sources(sources)[0][1].temporal_max_attempts == 1
+
+
+@pytest.mark.parametrize(
+    ("source", "error"),
+    [
+        (
+            {"type": "shell", "query": "run_whatever", "params": {}},
+            "not registered",
+        ),
+        (
+            {
+                "type": "shell",
+                "query": "forgejo_package_prune",
+                "params": {},
+            },
+            "must declare operation",
+        ),
+        (
+            {
+                "type": "shell",
+                "query": "forgejo_package_prune",
+                "operation": "forgejo_package_prune",
+                "params": {
+                    "prune_script": "/tmp/arbitrary-command",
+                    "apply": False,
+                    "max_versions": 3,
+                    "evidence_sinks": _evidence_sinks("forgejo_package_prune"),
+                },
+            },
+            "canonical path",
+        ),
+    ],
+)
+def test_shell_admission_fails_closed(source: dict, error: str) -> None:
+    with pytest.raises(ValueError, match=error):
+        validate_bounded_operations([source], [])
+
+
+def test_sbom_requires_bounded_limit_and_report_evidence() -> None:
+    source = {
+        "type": "sbom-nexus",
+        "query": "catch_up",
+        "operation": "sbom_nexus_ingest",
+        "params": {"apply": True, "limit": 4},
+    }
+    instruction = {
+        "report_sinks": _evidence_sinks("sbom_catchup"),
+    }
+
+    with pytest.raises(ValueError, match="limit must be in 1..3"):
+        validate_bounded_operations([source], [instruction])
+
+    source["params"]["limit"] = 3
+    with pytest.raises(ValueError, match="requires an instruction report"):
+        validate_bounded_operations([source], [])
+
+    validate_bounded_operations([source], [instruction])
+
+
+def test_backup_requires_explicit_safe_targets_and_evidence() -> None:
+    source = {
+        "type": "shell",
+        "query": "cnpg_option_a_backup",
+        "operation": "cnpg_option_a_backup",
+        "params": {
+            "backup_script": "/opt/railiance-platform/tools/cmd/cnpg-option-a-backup",
+            "dry_run": False,
+            "timeout_seconds": 7200,
+            "targets": "r01-forgejo-db,r01-state-hub-db",
+            "evidence_sinks": _evidence_sinks("cnpg_option_a_backup"),
+        },
+    }
+
+    validate_bounded_operations([source], [])
+    source["params"]["targets"] = "ok,$(unsafe)"
+    with pytest.raises(ValueError, match="safe names"):
+        validate_bounded_operations([source], [])
+
+    source["params"]["targets"] = "r01-forgejo-db"
+    source["params"]["evidence_sinks"] = [{"type": "state-hub-progress"}]
+    with pytest.raises(ValueError, match="event_type: cnpg_option_a_backup"):
+        validate_bounded_operations([source], [])
+
+
+def test_definition_cannot_combine_two_bounded_operations() -> None:
+    prune = {
+        "type": "shell",
+        "query": "forgejo_package_prune",
+        "operation": "forgejo_package_prune",
+    }
+    backup = {
+        "type": "shell",
+        "query": "cnpg_option_a_backup",
+        "operation": "cnpg_option_a_backup",
+    }
+
+    with pytest.raises(ValueError, match="at most one bounded operation"):
+        validate_bounded_operations([prune, backup], [])