feat: extract sensitive field redaction logic into a dedicated utility module

Author: tercel

src/apcore/builtin_steps.py

@@ -262,7 +262,10 @@ async def execute(self, ctx: PipelineContext) -> StepResult:
             result = await self._handler.check_approval(token)
         elif self._executor is not None and hasattr(self._executor, "_check_approval_async"):
             await self._executor._check_approval_async(
-                module, ctx.module_id, ctx.inputs, ctx.context,
+                module,
+                ctx.module_id,
+                ctx.inputs,
+                ctx.context,
             )
             return StepResult(action="continue")
         else:
@@ -317,11 +320,15 @@ async def execute(self, ctx: PipelineContext) -> StepResult:
             try:
                 if hasattr(self._middleware_manager, "execute_before_async"):
                     inputs, executed = await self._middleware_manager.execute_before_async(
-                        ctx.module_id, ctx.inputs, ctx.context,
+                        ctx.module_id,
+                        ctx.inputs,
+                        ctx.context,
                     )
                 else:
                     inputs, executed = self._middleware_manager.execute_before(
-                        ctx.module_id, ctx.inputs, ctx.context,
+                        ctx.module_id,
+                        ctx.inputs,
+                        ctx.context,
                     )
                 ctx.inputs = inputs
                 ctx.executed_middlewares = list(executed)
@@ -333,11 +340,19 @@ async def execute(self, ctx: PipelineContext) -> StepResult:
                 # on_error recovery for non-chain errors
                 if hasattr(self._middleware_manager, "execute_on_error_async"):
                     recovery = await self._middleware_manager.execute_on_error_async(
-                        ctx.module_id, ctx.inputs, exc, ctx.context, ctx.executed_middlewares,
+                        ctx.module_id,
+                        ctx.inputs,
+                        exc,
+                        ctx.context,
+                        ctx.executed_middlewares,
                     )
                 elif hasattr(self._middleware_manager, "execute_on_error"):
                     recovery = self._middleware_manager.execute_on_error(
-                        ctx.module_id, ctx.inputs, exc, ctx.context, ctx.executed_middlewares,
+                        ctx.module_id,
+                        ctx.inputs,
+                        exc,
+                        ctx.context,
+                        ctx.executed_middlewares,
                     )
                 else:
                     recovery = None
@@ -415,7 +430,7 @@ async def execute(self, ctx: PipelineContext) -> StepResult:
         # Redact sensitive fields after successful validation
         schema_dict_fn = getattr(input_schema, "model_json_schema", None)
         if schema_dict_fn is not None and callable(schema_dict_fn):
-            from apcore.executor import redact_sensitive
+            from apcore.utils.redaction import redact_sensitive
 
             schema = schema_dict_fn()
             redacted = redact_sensitive(ctx.inputs, schema)
@@ -512,7 +527,8 @@ async def execute(self, ctx: PipelineContext) -> StepResult:
         except asyncio.TimeoutError:
             timeout_ms = int((timeout_s or 0) * 1000)
             raise ModuleTimeoutError(
-                module_id=ctx.module_id, timeout_ms=timeout_ms,
+                module_id=ctx.module_id,
+                timeout_ms=timeout_ms,
             ) from None
         except (ExecutionCancelledError, ModuleTimeoutError, InvalidInputError):
             raise
@@ -568,7 +584,7 @@ async def execute(self, ctx: PipelineContext) -> StepResult:
         if ctx.context is not None and hasattr(ctx.context, "data"):
             schema_dict_fn = getattr(output_schema, "model_json_schema", None)
             if schema_dict_fn is not None and callable(schema_dict_fn):
-                from apcore.executor import redact_sensitive
+                from apcore.utils.redaction import redact_sensitive
 
                 schema = schema_dict_fn()
                 redacted = redact_sensitive(ctx.output, schema)
@@ -605,11 +621,17 @@ async def execute(self, ctx: PipelineContext) -> StepResult:
         if self._middleware_manager is not None:
             if hasattr(self._middleware_manager, "execute_after_async"):
                 output = await self._middleware_manager.execute_after_async(
-                    ctx.module_id, ctx.inputs, ctx.output or {}, ctx.context,
+                    ctx.module_id,
+                    ctx.inputs,
+                    ctx.output or {},
+                    ctx.context,
                 )
             else:
                 output = self._middleware_manager.execute_after(
-                    ctx.module_id, ctx.inputs, ctx.output or {}, ctx.context,
+                    ctx.module_id,
+                    ctx.inputs,
+                    ctx.output or {},
+                    ctx.context,
                 )
             ctx.output = output
             return StepResult(action="continue")

src/apcore/executor.py

@@ -8,19 +8,15 @@
 from __future__ import annotations
 
 import asyncio
-import copy
 import dataclasses
-import inspect
 import logging
 import threading
-import time
 from collections.abc import AsyncIterator
 from typing import Any, Callable
 
 import pydantic
 
 from apcore.acl import ACL
-from apcore.context_keys import REDACTED_OUTPUT
 from apcore.approval import ApprovalHandler, ApprovalRequest, ApprovalResult
 from apcore.cancel import ExecutionCancelledError
 from apcore.config import Config
@@ -51,11 +47,10 @@
     StrategyNotFoundError,
 )
 from apcore.registry import MODULE_ID_PATTERN, Registry
-from apcore.utils.call_chain import guard_call_chain
 
-__all__ = ["redact_sensitive", "REDACTED_VALUE", "Executor"]
+from apcore.utils.redaction import REDACTED_VALUE, redact_sensitive
 
-REDACTED_VALUE: str = "***REDACTED***"
+__all__ = ["redact_sensitive", "REDACTED_VALUE", "Executor"]
 
 _logger = logging.getLogger(__name__)
 
@@ -123,71 +118,8 @@ def _convert_validation_errors(error: pydantic.ValidationError) -> list[dict[str
 # =============================================================================
 
 
-def redact_sensitive(data: dict[str, Any], schema_dict: dict[str, Any]) -> dict[str, Any]:
-    """Redact fields marked with x-sensitive in the schema.
-
-    Implements Algorithm A13 from PROTOCOL_SPEC section 9.5.
-    Returns a deep copy of data with sensitive values replaced by "***REDACTED***".
-    Also redacts any keys starting with "_secret_" regardless of schema.
-
-    Args:
-        data: The data dict to redact.
-        schema_dict: A JSON Schema dict that may contain "x-sensitive": true
-            on individual properties.
-
-    Returns:
-        A new dict with sensitive values replaced. Original data is not modified.
-    """
-    redacted = copy.deepcopy(data)
-    _redact_fields(redacted, schema_dict)
-    _redact_secret_prefix(redacted)
-    return redacted
-
-
-def _redact_fields(data: dict[str, Any], schema_dict: dict[str, Any]) -> None:
-    """In-place redaction based on schema x-sensitive markers."""
-    properties = schema_dict.get("properties")
-    if not properties:
-        return
-
-    for field_name, field_schema in properties.items():
-        if field_name not in data:
-            continue
-
-        value = data[field_name]
-
-        # x-sensitive: true on this property
-        if field_schema.get("x-sensitive") is True:
-            if value is not None:
-                data[field_name] = REDACTED_VALUE
-            continue
-
-        # Nested object: recurse
-        if field_schema.get("type") == "object" and "properties" in field_schema and isinstance(value, dict):
-            _redact_fields(value, field_schema)
-            continue
-
-        # Array: redact items
-        if field_schema.get("type") == "array" and "items" in field_schema and isinstance(value, list):
-            items_schema = field_schema["items"]
-            if items_schema.get("x-sensitive") is True:
-                for i, item in enumerate(value):
-                    if item is not None:
-                        value[i] = REDACTED_VALUE
-            elif items_schema.get("type") == "object" and "properties" in items_schema:
-                for item in value:
-                    if isinstance(item, dict):
-                        _redact_fields(item, items_schema)
-
-
-def _redact_secret_prefix(data: dict[str, Any]) -> None:
-    """In-place redaction of keys starting with _secret_."""
-    for key in data:
-        value = data[key]
-        if key.startswith("_secret_") and value is not None:
-            data[key] = REDACTED_VALUE
-        elif isinstance(value, dict):
-            _redact_secret_prefix(value)
+# redact_sensitive and REDACTED_VALUE moved to apcore.utils.redaction in v0.17
+# Re-exported here for backward compatibility.
 
 
 # =============================================================================
@@ -480,12 +412,8 @@ def validate(
         if loop is None:
             if self._sync_loop is None or self._sync_loop.is_closed():
                 self._sync_loop = asyncio.new_event_loop()
-            return self._sync_loop.run_until_complete(
-                self._validate_async(module_id, inputs, context)
-            )
-        return self._run_in_new_thread(
-            self._validate_async(module_id, inputs, context), module_id, None
-        )
+            return self._sync_loop.run_until_complete(self._validate_async(module_id, inputs, context))
+        return self._run_in_new_thread(self._validate_async(module_id, inputs, context), module_id, None)
 
     async def _validate_async(
         self,
@@ -550,7 +478,11 @@ async def _validate_async(
             requires_approval = self._needs_approval(pipe_ctx.module)
 
         # Module-level preflight (optional)
-        if pipe_ctx.module is not None and hasattr(pipe_ctx.module, "preflight") and callable(pipe_ctx.module.preflight):
+        if (
+            pipe_ctx.module is not None
+            and hasattr(pipe_ctx.module, "preflight")
+            and callable(pipe_ctx.module.preflight)
+        ):
             try:
                 preflight_warnings = pipe_ctx.module.preflight(inputs, pipe_ctx.context)
                 if isinstance(preflight_warnings, list) and preflight_warnings:
@@ -719,6 +651,7 @@ def _translate_abort(self, abort: PipelineAbortError) -> ModuleError:
         if step == "execute":
             if "cancelled" in explanation.lower():
                 from apcore.cancel import ExecutionCancelledError
+
                 return ExecutionCancelledError()
             if "deadline" in explanation.lower() or "timed out" in explanation.lower():
                 return ModuleTimeoutError(module_id="", timeout_ms=0)
@@ -876,7 +809,10 @@ async def stream(
             wrapped = propagate_error(exc, module_id, ctx_obj) if ctx_obj else exc
             if pipe_ctx.executed_middlewares:
                 recovery = await self._middleware_manager.execute_on_error_async(
-                    module_id, pipe_ctx.inputs, wrapped, ctx_obj,
+                    module_id,
+                    pipe_ctx.inputs,
+                    wrapped,
+                    ctx_obj,
                     pipe_ctx.executed_middlewares,
                 )
                 if recovery is not None:
@@ -902,7 +838,10 @@ async def stream(
             wrapped = propagate_error(exc, module_id, ctx_obj) if ctx_obj else exc
             if pipe_ctx.executed_middlewares:
                 recovery = await self._middleware_manager.execute_on_error_async(
-                    module_id, pipe_ctx.inputs, wrapped, ctx_obj,
+                    module_id,
+                    pipe_ctx.inputs,
+                    wrapped,
+                    ctx_obj,
                     pipe_ctx.executed_middlewares,
                 )
                 if recovery is not None:
@@ -912,8 +851,9 @@ async def stream(
 
         # Phase 3: Output validation + middleware_after on accumulated result
         pipe_ctx.output = accumulated
-        post_steps = [s for s in self._strategy.steps
-                      if s.name in ("output_validation", "middleware_after", "return_result")]
+        post_steps = [
+            s for s in self._strategy.steps if s.name in ("output_validation", "middleware_after", "return_result")
+        ]
         if post_steps:
             post_strategy = ExecutionStrategy("post_stream", post_steps)
             try:

src/apcore/pipeline.py

@@ -226,9 +226,7 @@ async def run(
             timeout_ms = getattr(step, "timeout_ms", 0)
 
             # (1) match_modules filter
-            if match_modules is not None and not _any_match(
-                match_modules, ctx.module_id
-            ):
+            if match_modules is not None and not _any_match(match_modules, ctx.module_id):
                 trace.steps.append(
                     StepTrace(
                         name=step.name,
@@ -306,9 +304,7 @@ async def run(
                 duration = (time.monotonic() - step_start) * 1000
                 # (4) ignore_errors: log and continue
                 if ignore_errors:
-                    _logger.warning(
-                        "Step '%s' failed (ignored): %s", step.name, exc
-                    )
+                    _logger.warning("Step '%s' failed (ignored): %s", step.name, exc)
                     trace.steps.append(
                         StepTrace(
                             name=step.name,
@@ -327,9 +323,7 @@ async def run(
                     StepTrace(
                         name=step.name,
                         duration_ms=duration,
-                        result=StepResult(
-                            action="abort", explanation=str(exc)
-                        ),
+                        result=StepResult(action="abort", explanation=str(exc)),
                     )
                 )
                 trace.total_duration_ms = (time.monotonic() - start) * 1000

src/apcore/pipeline_config.py

@@ -115,9 +115,7 @@ def _resolve_step(step_def: dict[str, Any]) -> BaseStep:
 def _import_step(handler_path: str, name: str, config: dict[str, Any]) -> BaseStep:
     """Import a step class from a handler path like 'myapp.steps:RateLimitStep'."""
     if ":" not in handler_path:
-        raise ValueError(
-            f"Invalid handler path '{handler_path}'. Expected format: 'module.path:ClassName'"
-        )
+        raise ValueError(f"Invalid handler path '{handler_path}'. Expected format: 'module.path:ClassName'")
     module_path, class_name = handler_path.split(":", 1)
 
     import importlib
@@ -193,9 +191,7 @@ def build_strategy_from_config(
                     if hasattr(step, key):
                         setattr(step, key, value)
                     else:
-                        _logger.warning(
-                            "Step '%s' has no field '%s'", step_name, key
-                        )
+                        _logger.warning("Step '%s' has no field '%s'", step_name, key)
                 break
 
     # (3) Resolve and insert custom steps