Keep backup subprocess output out of workflow evidence

Sanitize failures and returned summaries at the adapter boundary. Allowlist backup status values and preserve failure exit codes. Verify synthetic sensitive output cannot survive Python traceback or Temporal failure serialization.

Decision-Class: strengthen-existing-evidence-boundary

Delegation: explicit-user-task; Emergent-Security-A6

Evidence: full-suite-676-passed-1-skipped; final-focused-48-passed; no-new-lint-findings

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a100de-2e1c-7e23-9ae9-2da0e0499411

tests/test_bounded_operations.py

@@ -165,3 +165,15 @@ def test_definition_cannot_combine_two_bounded_operations() -> None:
 
     with pytest.raises(ValueError, match="at most one bounded operation"):
         validate_bounded_operations([prune, backup], [])
+
+
+@pytest.mark.parametrize("overall", ["sensitive-value", {"secret": "value"}, ["value"], None])
+def test_backup_projection_does_not_copy_untrusted_status(overall):
+    result = normalize_operation_result("cnpg_option_a_backup", {"overall": overall})
+    assert result["overall"] == "unknown"
+
+
+@pytest.mark.parametrize("overall", ["ok", "degraded", "error", "pass", "fail", "unknown"])
+def test_backup_projection_preserves_supported_status(overall):
+    result = normalize_operation_result("cnpg_option_a_backup", {"overall": overall})
+    assert result["overall"] == overall

tests/test_cnpg_option_a_backup_resolver.py

@@ -3,11 +3,12 @@
 from __future__ import annotations
 
 import json
+import subprocess
+import traceback
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
-
 from activity_core.context_resolvers.cnpg_backup import cnpg_option_a_backup
 
 
@@ -60,3 +61,77 @@ def test_degraded_exit_2_still_parses(tmp_path: Path) -> None:
         result = cnpg_option_a_backup({"backup_script": str(script)})
     assert result["overall"] == "degraded"
     assert result["script_exit_code"] == 2
+
+
+CANARY = "synthetic-sensitive-subprocess-value-83917"
+
+
+def assert_safe_failure(error):
+    from temporalio.api.failure.v1 import Failure
+    from temporalio.converter import DefaultFailureConverter, DefaultPayloadConverter
+
+    assert CANARY not in "".join(traceback.format_exception(error))
+    failure = Failure()
+    DefaultFailureConverter().to_failure(error, DefaultPayloadConverter(), failure)
+    assert CANARY.encode() not in failure.SerializeToString()
+
+
+@pytest.mark.parametrize("failure,expected", [
+    (subprocess.TimeoutExpired([CANARY], 5, output=CANARY, stderr=CANARY), "timed out"),
+    (OSError(CANARY), "could not start"),
+    (UnicodeDecodeError("utf-8", CANARY.encode(), 0, 1, CANARY), "invalid output encoding"),
+])
+def test_subprocess_exceptions_are_safe_in_temporal(tmp_path, failure, expected):
+    script = tmp_path / "backup"
+    script.touch()
+    with (
+        patch("activity_core.context_resolvers.cnpg_backup.subprocess.run", side_effect=failure),
+        pytest.raises(RuntimeError, match=expected) as exc,
+    ):
+        cnpg_option_a_backup({"backup_script": str(script)})
+    assert_safe_failure(exc.value)
+
+
+@pytest.mark.parametrize("stdout,exit_code,expected,error_type", [
+    ("", 1, "failed ", RuntimeError),
+    (CANARY, 0, "invalid JSON", RuntimeError),
+    (json.dumps([CANARY]), 0, "non-object summary", TypeError),
+    ("null", 0, "non-object summary", TypeError),
+    (json.dumps(CANARY), 0, "non-object summary", TypeError),
+])
+def test_bad_output_never_reaches_exception_evidence(tmp_path, stdout, exit_code, expected, error_type):
+    script = tmp_path / "backup"
+    script.touch()
+    completed = MagicMock(returncode=exit_code, stdout=stdout, stderr=CANARY)
+    with (
+        patch("activity_core.context_resolvers.cnpg_backup.subprocess.run", return_value=completed),
+        pytest.raises(error_type, match=expected) as exc,
+    ):
+        cnpg_option_a_backup({"backup_script": str(script)})
+    assert_safe_failure(exc.value)
+
+
+@pytest.mark.parametrize("overall", [CANARY, {"secret": CANARY}, [CANARY]])
+def test_successful_json_is_allowlisted_before_return(tmp_path, overall):
+    script = tmp_path / "backup"
+    script.touch()
+    payload = {"overall": overall, "dry_run": False, "dumped": 2, "uploaded": 2,
+               "failed": 0, "targets": [CANARY], "errors": [CANARY], "log_tail": CANARY}
+    completed = MagicMock(returncode=0, stdout=json.dumps(payload), stderr=CANARY)
+    with patch("activity_core.context_resolvers.cnpg_backup.subprocess.run", return_value=completed):
+        result = cnpg_option_a_backup({"backup_script": str(script)})
+    assert result == {"kind": "cnpg_option_a_backup", "overall": "unknown", "dry_run": False,
+                      "dumped": 2, "uploaded": 2, "failed": 0, "script_exit_code": 0}
+    assert CANARY not in json.dumps(result)
+
+
+@pytest.mark.parametrize("exit_code", [1, -9])
+def test_process_failure_cannot_claim_success(tmp_path, exit_code):
+    script = tmp_path / "backup"
+    script.touch()
+    completed = MagicMock(returncode=exit_code, stdout=json.dumps({"overall": "ok"}), stderr=CANARY)
+    with patch("activity_core.context_resolvers.cnpg_backup.subprocess.run", return_value=completed):
+        result = cnpg_option_a_backup({"backup_script": str(script)})
+    assert result["overall"] == "error"
+    assert result["script_exit_code"] == exit_code
+    assert CANARY not in json.dumps(result)