[cross-repo from server#230] TD-S114: PHP activity failures leak PHP internals into Python-facing failure payloads (#67)

Author: durable-workflow-ops

README.md

@@ -70,6 +70,17 @@ Retry and timeout settings are scoped to the layer where you configure them:
 
 Timeout names are also layer-specific. `start_to_close_timeout` limits one activity attempt, `schedule_to_start_timeout` limits queue wait before an activity starts, `schedule_to_close_timeout` limits the whole activity execution including retries, and `heartbeat_timeout` limits the gap between activity heartbeats. For child workflows, `execution_timeout_seconds` covers the overall child workflow execution and `run_timeout_seconds` covers one run.
 
+## Activity failure payloads
+
+When replay raises `ActivityFailed`, the top-level attributes expose the
+stable cross-language fields: `activity_type`, `failure_category`,
+`exception_type`, `message`, `non_retryable`, and `code`. The
+`exception_payload` dictionary is filtered to language-neutral keys such as
+`type`, `message`, `details`, `details_payload_codec`, and `non_retryable`.
+Runtime diagnostics like PHP or Python exception classes, source file paths,
+line numbers, and traces are not included by default unless the history event
+contains an explicit `diagnostics` or `runtime_diagnostics` envelope.
+
 ## Activity retries and timeouts
 
 Configure per-call activity retries and deadlines from workflow code:

src/durable_workflow/workflow.py

@@ -1845,9 +1845,35 @@ def _optional_str(value: Any) -> str | None:
     return value if isinstance(value, str) and value != "" else None
 
 
+_NEUTRAL_EXCEPTION_PAYLOAD_KEYS = (
+    "type",
+    "message",
+    "code",
+    "details",
+    "details_payload_codec",
+    "non_retryable",
+)
+_EXPLICIT_DIAGNOSTICS_KEYS = ("diagnostics", "runtime_diagnostics")
+
+
+def _neutral_exception_payload(payload: Any) -> dict[str, Any] | None:
+    if not isinstance(payload, Mapping):
+        return None
+
+    neutral = {key: payload[key] for key in _NEUTRAL_EXCEPTION_PAYLOAD_KEYS if key in payload}
+
+    for key in _EXPLICIT_DIAGNOSTICS_KEYS:
+        diagnostics = payload.get(key)
+        if isinstance(diagnostics, Mapping):
+            neutral[key] = dict(diagnostics)
+
+    return neutral or None
+
+
 def _activity_failed_from_payload(payload: Mapping[str, Any]) -> ActivityFailed:
     exception_payload = payload.get("exception")
     exception = dict(exception_payload) if isinstance(exception_payload, Mapping) else None
+    exposed_exception = _neutral_exception_payload(exception_payload)
     activity_payload = payload.get("activity")
     activity = dict(activity_payload) if isinstance(activity_payload, Mapping) else None
 
@@ -1872,7 +1898,7 @@ def _activity_failed_from_payload(payload: Mapping[str, Any]) -> ActivityFailed:
         exception_class=exception_class,
         non_retryable=payload.get("non_retryable") is True,
         code=payload.get("code"),
-        exception_payload=exception,
+        exception_payload=exposed_exception,
         activity=activity,
     )
 

tests/test_replay.py

@@ -69,6 +69,23 @@ def run(self, ctx: WorkflowContext, order_id: str):  # type: ignore[no-untyped-d
         return {"charged": True}
 
 
+@workflow.defn(name="activity-failure-payload-inspector")
+class ActivityFailurePayloadInspector:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        try:
+            yield ctx.schedule_activity("polyglot.php.fail", [])
+        except ActivityFailed as exc:
+            return {
+                "activity_type": exc.activity_type,
+                "failure_category": exc.failure_category,
+                "exception_type": exc.exception_type,
+                "non_retryable": exc.non_retryable,
+                "message": str(exc),
+                "exception_payload": exc.exception_payload,
+            }
+        return {"ok": True}
+
+
 @workflow.defn(name="two-activities")
 class TwoActivities:
     def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
@@ -233,6 +250,92 @@ def test_failed_activity_compensation_can_complete(self) -> None:
             "exception_class": "payments.PaymentDeclined",
         }
 
+    def test_php_activity_failure_payload_omits_runtime_diagnostics_by_default(self) -> None:
+        history = [
+            {
+                "event_type": "ActivityFailed",
+                "payload": {
+                    "activity_type": "polyglot.php.fail",
+                    "failure_category": "activity",
+                    "exception_type": "PolyglotPhpPlannedFailure",
+                    "exception_class": "RuntimeException",
+                    "message": "php activity planned failure",
+                    "non_retryable": True,
+                    "exception": {
+                        "class": "RuntimeException",
+                        "type": "PolyglotPhpPlannedFailure",
+                        "message": "php activity planned failure",
+                        "file": "/app/vendor/durable-workflow/workflow/src/V2/Support/FailureFactory.php",
+                        "line": 571,
+                        "trace": [
+                            {
+                                "file": "/app/vendor/durable-workflow/workflow/src/V2/Support/FailureFactory.php"
+                            }
+                        ],
+                        "properties": [],
+                        "details": "encoded-details",
+                        "details_payload_codec": "avro",
+                        "non_retryable": True,
+                    },
+                },
+            }
+        ]
+
+        outcome = replay(ActivityFailurePayloadInspector, history, [])
+
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, CompleteWorkflow)
+        assert cmd.result == {
+            "activity_type": "polyglot.php.fail",
+            "failure_category": "activity",
+            "exception_type": "PolyglotPhpPlannedFailure",
+            "non_retryable": True,
+            "message": "php activity planned failure",
+            "exception_payload": {
+                "type": "PolyglotPhpPlannedFailure",
+                "message": "php activity planned failure",
+                "details": "encoded-details",
+                "details_payload_codec": "avro",
+                "non_retryable": True,
+            },
+        }
+
+    def test_activity_failure_payload_preserves_explicit_diagnostics_envelope(self) -> None:
+        history = [
+            {
+                "event_type": "ActivityFailed",
+                "payload": {
+                    "activity_type": "polyglot.php.fail",
+                    "failure_category": "activity",
+                    "exception_type": "PolyglotPhpPlannedFailure",
+                    "message": "php activity planned failure",
+                    "exception": {
+                        "type": "PolyglotPhpPlannedFailure",
+                        "message": "php activity planned failure",
+                        "runtime_diagnostics": {
+                            "class": "RuntimeException",
+                            "file": "/app/src/TimeoutActivity.php",
+                        },
+                    },
+                },
+            }
+        ]
+
+        outcome = replay(ActivityFailurePayloadInspector, history, [])
+
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, CompleteWorkflow)
+        assert cmd.result["exception_payload"] == {
+            "type": "PolyglotPhpPlannedFailure",
+            "message": "php activity planned failure",
+            "runtime_diagnostics": {
+                "class": "RuntimeException",
+                "file": "/app/src/TimeoutActivity.php",
+            },
+        }
+
     def test_server_command_shape(self) -> None:
         outcome = replay(OneActivity, [], ["world"])
         cmd = outcome.commands[0]