[NOTICKET] fix: F-9 — redact PII in the two persisted artifacts

`pii_flag` was an ingestion-time control only. It nulls `sample_values` going into the
planner prompt, and REPO_STATUS §8 claims "real values never enter prompts" — true of
samples, false of results. Nothing stops the planner SELECTing a flagged column, and
it should not: "list our top 20 customers by revenue" legitimately selects
`customer_name` and `email`.

Those real values then reached two places that KEEP them:
1. `message_traceability.data` — persisted, and served by a GET with no auth (F-3
was declined, so the service secret is the only control there).
2. Report evidence tables — rendered into `reports.content`, permanently, versioned.

`retrieve_data` now reports `meta.pii_columns`, resolved through the IR select list so
aliases are honoured, and both sinks redact those cells to "[redacted]".

Per the lead's 2026-07-23 decision the ASSEMBLER still receives the real rows, so the
answer prose is unchanged and the question stays answerable. Verified end-to-end
through the real `_retrieve_data` path: the assembler input kept "Ada Lovelace" while
the persisted preview showed "[redacted]".

Deliberate choices:
- Aggregates are NOT masked, except `min`/`max`. `sum(salary)` and `count(email)`
are derived numbers identifying nobody; masking them would destroy a legitimate
answer for no privacy gain. `min`/`max` are different — they return an actual
member value, so `max(email)` really is one customer's address.
- Column HEADERS are kept. The column name is not the secret, and keeping it is what
makes the redaction legible rather than looking like missing data. The traceability
record also carries `pii_masked` naming the affected columns.
- A fixed marker, not a partial reveal: "j***@acme.com" still leaks the domain and
the name shape.
- Fails OPEN. An output name that cannot be resolved is left unmasked rather than
blanking a legitimate column, so the worst case is today's behaviour. This is a
mitigation, not a guarantee — the guarantee would have to come from not selecting
the column, which is the product decision already taken the other way.
- `_PII_MASK` and `_pii_indexes` are imported by the report generator rather than
re-implemented; a second copy is the drift CODE_REVIEW F-27 warns about.

Older persisted rows carry no `pii_columns` key and are returned exactly as before.

15 new tests. Suite 456 passed / 0 failed / 7 skipped. Ruff clean on touched paths.
Contract updated for both surfaces.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/agents/report/generator.py

@@ -25,6 +25,11 @@ from langchain_openai import AzureChatOpenAI
 
 from src.middlewares.logging import get_logger
 
+# Reused, not re-implemented: the traceability preview and the report evidence table
+# are the two persisted sinks F-9 masks, and a second copy of the mask marker or the
+# index lookup would be the exact drift CODE_REVIEW F-27 warns about.
+from src.traceability.scratchpad import _PII_MASK, _pii_indexes
+
 from ..language import detect_reply_language
 from ..slow_path.schemas import AnalysisRecord, TaskSummary
 from .errors import ReportError
@@ -186,12 +191,25 @@ def _collect_evidence(records: list[AnalysisRecord]) -> dict[str, list[EvidenceT
                     continue
                 if len(output.columns) > _EVIDENCE_MAX_COLS:
                     continue
+                # Mask PII cells before they are frozen into `reports.content`
+                # (F-9, 2026-07-24). A report is a permanent, versioned artifact, so
+                # this is the sink where an unmasked customer name or email lasts
+                # longest. The assembler's FINDINGS are untouched — the lead's
+                # 2026-07-23 decision keeps real values in the answer prose; this
+                # redacts only the raw evidence dump beneath it. `pii_columns` is
+                # absent on records persisted before F-9, which yields no masking.
+                pii_idx = _pii_indexes(
+                    (output.meta or {}).get("pii_columns"), output.columns
+                )
                 tables.append(
                     EvidenceTable(
                         title=result.objective,
                         columns=[str(c) for c in output.columns],
                         rows=[
-                            [_fmt_cell(v) for v in row]
+                            [
+                                _PII_MASK if i in pii_idx else _fmt_cell(v)
+                                for i, v in enumerate(row)
+                            ]
                             for row in output.rows[:_EVIDENCE_MAX_ROWS]
                         ],
                         truncated=len(output.rows) > _EVIDENCE_MAX_ROWS,

src/tools/data_access.py

@@ -281,6 +281,15 @@ class DataAccessToolInvoker:
                 "elapsed_ms": result.elapsed_ms,
                 # Executed query for traceability (KM-691); None if unavailable.
                 "query": result.query,
+                # Which of `columns` hold PII, by NAME (F-9, 2026-07-24). `pii_flag`
+                # was an ingestion-time control only: it nulls `sample_values` into
+                # the prompt, but nothing stops the planner SELECTing a flagged column
+                # — and "list our top 20 customers" legitimately selects one. The real
+                # values then flowed into the traceability preview and the report
+                # evidence tables, both PERSISTED. Carrying the flags here is what lets
+                # those two sinks mask without re-reading the catalog. Consumers must
+                # tolerate its absence: older persisted rows have no such key.
+                "pii_columns": _pii_column_names(catalog, ir, result.columns),
             },
         )
 
@@ -345,6 +354,52 @@ class DataAccessToolInvoker:
         )
 
 
+def _pii_column_names(
+    catalog: Catalog, ir: QueryIR, output_columns: list[str]
+) -> list[str]:
+    """Which of `output_columns` carry PII, resolved through the IR's select list.
+
+    Returns names, not indexes, because the two compilers do not agree on column
+    ORDER: the grouped pandas path builds its frame with `reset_index()`, which emits
+    the group columns first regardless of where they sat in the select list. Names
+    survive that; positions do not.
+
+    Aggregates are deliberately NOT flagged, with two exceptions. `count`, `sum`,
+    `avg` and `count_distinct` return derived numbers that identify nobody — masking
+    `sum(salary)` would destroy a legitimate answer for no privacy gain. `min` and
+    `max` are different: they return an actual member value, so `max(email)` really is
+    one customer's email.
+
+    Fails OPEN by design — an output name this cannot resolve is simply not listed, so
+    the worst case is today's behaviour (unmasked), never a legitimate column wrongly
+    blanked. That makes this a mitigation, not a guarantee; the guarantee has to come
+    from not selecting the column, which is a product decision (the lead's 2026-07-23
+    call was that the assembler still sees real values, so "list our top customers"
+    stays answerable).
+    """
+    cols_by_id = {
+        c.column_id: c
+        for s in catalog.sources
+        for t in s.tables
+        for c in t.columns
+    }
+    present = set(output_columns)
+    flagged: list[str] = []
+    for item in ir.select:
+        col = cols_by_id.get(getattr(item, "column_id", None) or "")
+        if col is None or not col.pii_flag:
+            continue
+        if item.kind == "column":
+            name = item.alias or col.name
+        elif item.fn in ("min", "max"):
+            name = item.alias or f"{item.fn}_{col.name}"
+        else:
+            continue  # count/sum/avg/count_distinct — derived, identifies nobody
+        if name in present and name not in flagged:
+            flagged.append(name)
+    return flagged
+
+
 def _json_safe(value: Any) -> Any:
     """Coerce DB scalar types that JSON can't represent into plain Python.
 

src/traceability/scratchpad.py

@@ -29,6 +29,27 @@ CAP_STR = 300
 # the feature, and 300 chars mangles all but the smallest statements.
 CAP_QUERY = 2000
 
+# What a masked PII cell shows in the persisted preview (F-9). A fixed marker, not a
+# partial reveal: showing "j***@acme.com" still leaks the domain and the name shape.
+_PII_MASK = "[redacted]"
+
+
+def _pii_indexes(pii_columns: Any, columns: Any) -> set[int]:
+    """Positions in `columns` named by `pii_columns` (from `ToolOutput.meta`).
+
+    Tolerant on purpose: the key is absent on any output persisted before F-9 and on
+    synth dicts, and a name that no longer matches a column is skipped rather than
+    raising. This runs inside the traceability scratchpad, which is a never-throw seam
+    — a masking slip must never break the user's answer.
+    """
+    if not pii_columns or not columns:
+        return set()
+    try:
+        wanted = set(pii_columns)
+        return {i for i, name in enumerate(columns) if name in wanted}
+    except TypeError:  # non-iterable payload from an unexpected shape
+        return set()
+
 
 def _truncate(obj: Any) -> Any:
     """Recursively cap strings to CAP_STR and summarize embedded tool results.
@@ -66,9 +87,26 @@ def _output_to_dict(output: Any) -> dict[str, Any]:
         columns = getattr(output, "columns", None)
         if columns is not None:
             result["columns"] = list(columns)
+        # Mask PII cells before the preview is PERSISTED (F-9, 2026-07-24). This row
+        # goes to `message_traceability.data` and is served by an unauthenticated GET,
+        # so real customer names/emails would otherwise sit in the database
+        # indefinitely. The assembler still receives the unmasked rows — the lead's
+        # 2026-07-23 decision — so the ANSWER is unaffected; only the stored artifact
+        # is redacted. `pii_columns` is absent on older outputs, which yields no
+        # masking, exactly as before.
+        meta = getattr(output, "meta", None) or {}
+        pii_idx = _pii_indexes(meta.get("pii_columns"), columns)
         result["preview"] = [
-            [_truncate(cell) for cell in row] for row in rows[:CAP_PREVIEW_ROWS]
+            [
+                _PII_MASK if i in pii_idx else _truncate(cell)
+                for i, cell in enumerate(row)
+            ]
+            for row in rows[:CAP_PREVIEW_ROWS]
         ]
+        if pii_idx:
+            result["pii_masked"] = sorted(
+                columns[i] for i in pii_idx if i < len(columns)
+            )
     value = getattr(output, "value", None)
     if value is not None:
         if kind == "chart" and isinstance(value, dict):