logger: add pre-print callback hook + PII redaction filter

.agents/AGENTS.md

@@ -82,6 +82,15 @@ Tips when editing event modules:
 - `acai_aws.common.logger` logs structured JSON by default. Switching `LOG_FORMAT=INLINE` helps during local dev while `LOG_FORMAT=JSON` keeps CloudWatch-friendly output. `LOG_LEVEL` gates log emission (`INFO`, `DEBUG`, `WARN`, `ERROR`).
 - The `@log` decorator wraps any function, optionally gating logs with a boolean `condition`. Maintain argument pass-through so debugging remains straightforward.
 - Error traces should include the stack plus the high-level message; tests assert the JSON keys stay consistent (`level`, `time`, `error_trace`, `log`).
+- Records pass through callbacks registered with `CommonLogger.register_callback(callback)` before they print; each callback receives the `{level, time, trace, log}` record and returns the record to emit (redaction, enrichment, routing). `CommonLogger.reset_callbacks()` clears them.
+- `RedactionFilter(keys=[...], patterns=[...], redact_with='[REDACTED]')` is a callback that scrubs matching field names (case-insensitive, any depth) and regex value matches. It is opt-in: register it with `CommonLogger.register_callback(...)` to apply it.
+
+```python
+from acai_aws.common.logger.common_logger import CommonLogger
+from acai_aws.common.logger.redaction import RedactionFilter
+
+CommonLogger.register_callback(RedactionFilter(keys=['account_number']))
+```
 
 ## Agent Checklist
 1. **Understand the route or event** you’re touching—confirm how the filesystem maps to the API or which event module processes the payload.

README.md

@@ -440,11 +440,59 @@ The old `raise_body_error=True` and `raise_operation_error=True` kwargs still wo
 
 ---
 
+## 🪵 Logging & Redaction
+
+`acai_aws.common.logger` emits structured records to stdout. `LOG_FORMAT` selects `JSON` (default), `PRETTY`, or `INLINE`; `LOG_LEVEL` (`DEBUG`, `INFO`, `WARN`, `ERROR`) gates emission.
+
+```python
+from acai_aws.common import logger
+
+logger.log(level='INFO', log={'message': 'worker created', 'worker_id': worker_id})
+```
+
+### Callbacks
+
+Every record (`{level, time, trace, log}`) passes through registered callbacks before it is printed. A callback receives the record and returns the record to emit, so it can redact, enrich, or route. Callbacks run in registration order.
+
+```python
+from acai_aws.common.logger.common_logger import CommonLogger
+
+
+def add_service(record):
+    record['log']['service'] = 'payroll'
+    return record
+
+
+CommonLogger.register_callback(add_service)
+```
+
+### Redaction
+
+`RedactionFilter` is a callback that scrubs sensitive data: field names (case-insensitive, at any depth) and regex matches in string values. `redact_with` overrides the `[REDACTED]` default.
+
+```python
+from acai_aws.common.logger.common_logger import CommonLogger
+from acai_aws.common.logger.redaction import RedactionFilter
+
+CommonLogger.register_callback(RedactionFilter(
+    keys=['account_number'],
+    patterns=[r'\b\d{3}-\d{2}-\d{4}\b'],
+    redact_with='***',
+))
+
+logger.log(level='INFO', log={'account_number': '987654321', 'note': 'ssn 123-45-6789'})
+# emitted log -> {'account_number': '***', 'note': 'ssn ***'}
+```
+
+Filters are opt-in: register the ones you want at startup and they apply to every subsequent log.
+
+---
+
 ## 🧰 Tooling & Development Experience
 
 - **OpenAPI Generator** – CLI (`python -m acai_aws.apigateway generate-openapi`) scans handlers and updates schema docs
 - **Request/Response Helpers** – Access JSON, GraphQL, form, XML, or raw bodies via `Request.json`, `Request.form`, etc.
-- **Logging** – Configurable JSON/inline logging via `acai_aws.common.logger`
+- **Logging** – Structured JSON/inline logging via `acai_aws.common.logger`, with pluggable pre-print callbacks and an opt-in PII redaction filter
 - **Validation** – JSON Schema (Draft 7) and Pydantic support with helpful error messages
 
 ---

acai_aws/common/logger/__init__.py

@@ -1,6 +1,9 @@
 import logging
 
 from acai_aws.common.logger.common_logger import CommonLogger
+from acai_aws.common.logger.redaction import RedactionFilter
+
+__all__ = ['log', 'CommonLogger', 'RedactionFilter']
 
 
 def log(**kwargs):

acai_aws/common/logger/common_logger.py

@@ -7,6 +7,8 @@
 
 class CommonLogger:
 
+    _callbacks = []
+
     def __init__(self):
         self.__json = jsonpickle
         env_format = os.getenv('LOG_FORMAT', 'JSON') or 'JSON'
@@ -26,14 +28,32 @@ def __init__(self):
         if self.__format not in ['JSON', 'PRETTY', 'INLINE']:
             raise ValueError(f'LOG_FORMAT ENV must be either `JSON`, `PRETTY`, or `INLINE`, recieved: {self.__format}')
 
+    @classmethod
+    def register_callback(cls, callback):
+        cls._callbacks.append(callback)
+
+    @classmethod
+    def reset_callbacks(cls):
+        cls._callbacks = []
+
     def log(self, **kwargs):
-        default_log = {'level': kwargs.get('level', 'INFO'), 'log': kwargs.get('log', {})}
-        if self.__should_log(default_log['level']) and self.__format == 'JSON':
-            self.__log_json(**kwargs)
-        elif self.__should_log(default_log['level']) and self.__format == 'PRETTY':
-            self.__log_json(pretty=True, **kwargs)
-        elif self.__should_log(default_log['level']) and self.__format == 'INLINE':
-            self.__log_inline(**kwargs)
+        level = kwargs.get('level', 'INFO')
+        if not self.__should_log(level):
+            return
+        record = {
+            'level': level,
+            'time': datetime.datetime.now(datetime.timezone.utc).isoformat(),
+            'trace': [trace.strip() for trace in self.__get_traceback().split('\n') if trace],
+            'log': kwargs.get('log', {})
+        }
+        for callback in CommonLogger._callbacks:
+            record = callback(record)
+        if self.__format == 'JSON':
+            self.__log_json(record)
+        elif self.__format == 'PRETTY':
+            self.__log_json(record, pretty=True)
+        elif self.__format == 'INLINE':
+            self.__log_inline(record)
 
     def __get_traceback(self):
         trace = traceback.format_exc()
@@ -46,21 +66,15 @@ def __should_log(self, level):
         log_level_setting = self.log_levels[self.__log_level]
         return current_log_level >= log_level_setting
 
-    def __log_json(self, pretty=False, **kwargs):
-        print(self.__json.encode({
-            'level': kwargs['level'],
-            'time': datetime.datetime.now(datetime.timezone.utc).isoformat(),
-            'trace': [trace.strip() for trace in self.__get_traceback().split('\n') if trace],
-            'log': kwargs['log']
-        }, indent=4 if pretty else None))
+    def __log_json(self, record, pretty=False):
+        print(self.__json.encode(record, indent=4 if pretty else None))
 
-    def __log_inline(self, **kwargs):
-        timestamp = datetime.datetime.now(datetime.timezone.utc).isoformat()
-        trace = self.__get_traceback().strip().replace('\n', ' | ')
-        log_value = kwargs['log']
+    def __log_inline(self, record):
+        log_value = record['log']
         if not isinstance(log_value, str):
             log_value = str(log_value)
-        inline_message = f"{kwargs['level']}|time={timestamp} log={log_value}"
+        inline_message = f"{record['level']}|time={record['time']} log={log_value}"
+        trace = ' | '.join(record['trace'])
         if trace:
             inline_message = f"{inline_message} trace={trace}"
         print(inline_message)

acai_aws/common/logger/redaction.py

@@ -0,0 +1,35 @@
+import re
+
+
+class RedactionFilter:
+
+    DEFAULT_REDACTION = '[REDACTED]'
+
+    def __init__(self, **kwargs):
+        self._keys = {str(key).lower() for key in kwargs.get('keys') or []}
+        self._patterns = [re.compile(pattern) for pattern in kwargs.get('patterns') or []]
+        self._redact_with = kwargs.get('redact_with', self.DEFAULT_REDACTION)
+
+    def __call__(self, record):
+        if isinstance(record, dict) and 'log' in record:
+            record['log'] = self._scrub(record['log'])
+        return record
+
+    def _scrub(self, value):
+        if isinstance(value, dict):
+            return {key: self._scrub_field(key, item) for key, item in value.items()}
+        if isinstance(value, (list, tuple)):
+            return [self._scrub(item) for item in value]
+        if isinstance(value, str):
+            return self._scrub_text(value)
+        return value
+
+    def _scrub_field(self, key, value):
+        if str(key).lower() in self._keys:
+            return self._redact_with
+        return self._scrub(value)
+
+    def _scrub_text(self, value):
+        for pattern in self._patterns:
+            value = pattern.sub(self._redact_with, value)
+        return value

tests/acai_aws/common/test_logger.py

@@ -6,6 +6,8 @@
 
 from acai_aws.common import logger
 from acai_aws.common.logger.decorator import log
+from acai_aws.common.logger.common_logger import CommonLogger
+from acai_aws.common.logger.redaction import RedactionFilter
 
 def some_log_condition(*args, **_):
     if args[0] == 1:
@@ -164,3 +166,48 @@ def test_logger_accepts_critical_level(self):
     @mock.patch.dict(os.environ, {'RUN_MODE': 'SEE-LOGS', 'LOG_STAGE_VARIABLE': 'STAGE', 'STAGE': 'local', 'LOG_LEVEL': 'ERROR', 'LOG_FORMAT': 'BAD'})
     def test_logger_handles_bad_format(self):
         logger.log(level='INFO', log={'INFO': 'ignore'})
+
+
+class LoggerCallbackTest(TestCase):
+
+    def setUp(self):
+        self._saved_callbacks = list(CommonLogger._callbacks)
+        CommonLogger.reset_callbacks()
+
+    def tearDown(self):
+        CommonLogger._callbacks = self._saved_callbacks
+
+    @mock.patch.dict(os.environ, {'LOG_FORMAT': 'JSON', 'LOG_LEVEL': 'INFO'})
+    def test_callback_runs_before_print(self):
+        CommonLogger.register_callback(lambda record: {**record, 'log': {'replaced': True}})
+        buffer = io.StringIO()
+        with redirect_stdout(buffer):
+            logger.log(level='INFO', log={'email': 'ada@example.com'})
+        parsed = json.loads(buffer.getvalue().strip())
+        self.assertEqual({'replaced': True}, parsed['log'])
+
+    @mock.patch.dict(os.environ, {'LOG_FORMAT': 'JSON', 'LOG_LEVEL': 'INFO'})
+    def test_redaction_filter_runs_through_logger(self):
+        CommonLogger.register_callback(RedactionFilter(keys=['email']))
+        buffer = io.StringIO()
+        with redirect_stdout(buffer):
+            logger.log(level='INFO', log={'email': 'ada@example.com', 'id': 1})
+        parsed = json.loads(buffer.getvalue().strip())
+        self.assertEqual('[REDACTED]', parsed['log']['email'])
+        self.assertEqual(1, parsed['log']['id'])
+
+    @mock.patch.dict(os.environ, {'LOG_FORMAT': 'JSON', 'LOG_LEVEL': 'INFO'})
+    def test_callbacks_run_in_registration_order(self):
+        CommonLogger.register_callback(lambda record: {**record, 'log': {'step': 'one'}})
+        CommonLogger.register_callback(lambda record: {**record, 'log': {'step': 'two'}})
+        buffer = io.StringIO()
+        with redirect_stdout(buffer):
+            logger.log(level='INFO', log={'step': 'zero'})
+        parsed = json.loads(buffer.getvalue().strip())
+        self.assertEqual({'step': 'two'}, parsed['log'])
+
+    def test_reset_callbacks_clears_registry(self):
+        CommonLogger.register_callback(lambda record: record)
+        self.assertEqual(1, len(CommonLogger._callbacks))
+        CommonLogger.reset_callbacks()
+        self.assertEqual(0, len(CommonLogger._callbacks))

tests/acai_aws/common/test_redaction.py

@@ -0,0 +1,62 @@
+from unittest import TestCase
+
+from acai_aws.common.logger.redaction import RedactionFilter
+
+
+class RedactionFilterTest(TestCase):
+
+    def test_redacts_matching_keys(self):
+        redact = RedactionFilter(keys=['email', 'last_name'])
+        record = redact({'log': {'first_name': 'Ada', 'last_name': 'Lovelace', 'email': 'ada@example.com'}})
+        self.assertEqual('Ada', record['log']['first_name'])
+        self.assertEqual('[REDACTED]', record['log']['last_name'])
+        self.assertEqual('[REDACTED]', record['log']['email'])
+
+    def test_key_match_is_case_insensitive(self):
+        redact = RedactionFilter(keys=['email'])
+        record = redact({'log': {'Email': 'ada@example.com'}})
+        self.assertEqual('[REDACTED]', record['log']['Email'])
+
+    def test_redacts_nested_keys_in_dicts_and_lists(self):
+        redact = RedactionFilter(keys=['ssn'])
+        payload = {'worker': {'ssn': '123-45-6789', 'id': 5}, 'items': [{'ssn': '111-22-3333'}]}
+        record = redact({'log': payload})
+        self.assertEqual('[REDACTED]', record['log']['worker']['ssn'])
+        self.assertEqual(5, record['log']['worker']['id'])
+        self.assertEqual('[REDACTED]', record['log']['items'][0]['ssn'])
+
+    def test_redacts_value_patterns(self):
+        redact = RedactionFilter(patterns=[r'\b\d{3}-\d{2}-\d{4}\b'])
+        record = redact({'log': {'note': 'ssn is 123-45-6789 today'}})
+        self.assertEqual('ssn is [REDACTED] today', record['log']['note'])
+
+    def test_custom_redact_with(self):
+        redact = RedactionFilter(keys=['email'], redact_with='')
+        record = redact({'log': {'email': 'ada@example.com'}})
+        self.assertEqual('', record['log']['email'])
+
+    def test_leaves_non_matching_data_untouched(self):
+        redact = RedactionFilter(keys=['email'], patterns=[r'\b\d{3}-\d{2}-\d{4}\b'])
+        record = redact({'log': {'count': 3, 'ok': True, 'name': 'service-payroll'}})
+        self.assertEqual({'count': 3, 'ok': True, 'name': 'service-payroll'}, record['log'])
+
+    def test_handles_record_without_log_key(self):
+        redact = RedactionFilter(keys=['email'])
+        record = redact({'level': 'INFO'})
+        self.assertEqual({'level': 'INFO'}, record)
+
+    def test_keys_and_patterns_together(self):
+        redact = RedactionFilter(
+            keys=['first_name', 'email'],
+            patterns=[r'\b\d{3}-\d{2}-\d{4}\b', r'[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}'],
+        )
+        record = redact({'log': {
+            'first_name': 'Ada',
+            'email': 'ada@example.com',
+            'message': 'ssn 123-45-6789 mailto ada@example.com',
+        }})
+        log = record['log']
+        self.assertEqual('[REDACTED]', log['first_name'])
+        self.assertEqual('[REDACTED]', log['email'])
+        self.assertNotIn('123-45-6789', log['message'])
+        self.assertNotIn('ada@example.com', log['message'])