Edition 3: Debug an AI workflow without logging customer secrets

Your support-draft workflow fails at 2:10 p.m. The dashboard says the model timed out, but it cannot show which step failed without revealing the customer's email, account reference and full message. Do you keep the trace and accept the privacy risk, delete it and debug blind, or build a smaller record that answers the operating question without storing the conversation?

This lab builds the third option. You will produce a structural trace for a fictional customer-support drafting workflow, reject unexpected content fields before persistence, export a JSONL event file, create an incident summary and prove that three seeded sensitive values never reach either artifact.

The intended reader is a founder or small operations team that needs to diagnose an AI workflow but does not have a dedicated observability or privacy engineering group. Allow 45 to 60 minutes. You need Python 3.12 or later, a terminal and an empty directory. The exercise uses only the standard library, makes no network calls and costs no API money.

Why this problem deserves a build

The joint government guide Careful Adoption of Agentic AI Services, published May 1, 2026, identifies obscure event records and expanding data access as operating risks. That establishes a recognized security problem. It does not measure how often small businesses leak data through traces.

A TraceRoot issue opened September 24, 2026 reported that the documentation did not clearly disclose that tracing could capture prompts, completions, function arguments, return values and source context. The issue says teams handling confidential prompts need capture controls before adoption. A documentation change closed the issue on October 12. This is one public implementer report and repair, not prevalence or search-volume evidence.

A bounded guidance sample was checked on October 18. OpenTelemetry's sensitive-data guidance recommends data minimization and describes attribute, filter, redaction and transform processors. It also warns that hashing predictable identifiers may not anonymize them. OpenTelemetry's observability primer explains why traces need enough context to follow work across steps. The TraceRoot repair documents product capture controls. Those guides are useful, but a small team still needs a vendor-neutral exercise that declares a safe event contract, rejects drift, runs canary leak tests and keeps exception effort visible. This article supplies that bounded step. It does not claim exclusive coverage.

If you only need to choose a first workflow, begin with Edition 1. If uncertain external actions are your main problem, use Edition 2's recovery ledger. This lesson handles a different problem: retaining enough evidence to diagnose work while minimizing stored customer content.

1. Define the operating question before the fields

The question is not "What did the customer say?" It is "Which step failed, how long did it run, how much model work occurred, and does a person need to intervene?"

For this fictional workflow, each trace may retain:

1. A random or application-generated trace identifier.
2. A step name from a short approved set.
3. Status, duration, token counts and input or output lengths.
4. A low-cardinality error class.
5. A Boolean review flag.
6. A keyed pseudonym when repeated work for the same actor must be correlated.

It may not retain prompts, responses, tool arguments, contact values, account references or arbitrary metadata. An allowlist is safer here than a blacklist because a new field fails closed. A regular expression that removes known email formats would still miss unanticipated secrets, free-form names, private business text and new payload shapes.

The keyed actor reference is pseudonymous, not anonymous. Anyone with the key and a small candidate set may test guesses. Production keys belong in a secret manager, access must be restricted, retention must be bounded, and a team should omit the field entirely when cross-event correlation is unnecessary.

2. Freeze the baseline and acceptance criteria

The synthetic baseline has three support cases and nine raw content fields: contact, message and draft for each case. It contains three canaries representing an email, a phone number and a token-shaped secret.

Declare these acceptance criteria before running the code:

• The trace reconstructs each request's intake, generation and policy-check path.
• No event captures raw content.
• Neither output artifact contains any seeded canary.
• An unexpected raw_prompt field is rejected.
• Turning content capture on is rejected.
• A malformed actor reference is rejected.
• Review cases remain visible with their time and illustrative labor cost.

The lab assigns five review minutes to a policy conflict and two minutes to a timeout, then values seven minutes at an illustrative $30 per hour. The resulting $3.50 is synthetic. It excludes implementation, storage, incident response, legal review and any provider charge.

3. Run the complete privacy-safe trace lab

Create an empty directory. Save the following as privacy_safe_trace.py:

!/usr/bin/env python3

"""Dependency-free lab for useful AI workflow traces without stored content.""" from future import annotations

import hashlib import hmac import json import re import sys from collections import Counter from decimal import Decimal from pathlib import Path

DEMO_HMAC_KEY = b"training-key-not-for-production" ALLOWED_FIELDS = { "schema_version", "event_id", "trace_id", "actor_ref", "step", "status", "duration_ms", "input_chars", "output_chars", "input_tokens", "output_tokens", "error_code", "review_required", "content_capture", } STEPS = {"intake", "generate", "policy_check"} STATUSES = {"OK", "ERROR", "REVIEW"} ERROR_CODES = {"NONE", "POLICY_CONFLICT", "PROVIDER_TIMEOUT"} PATTERNS = { "event_id": re.compile(r"EVT-[0-9]{3}"), "trace_id": re.compile(r"TR-[0-9]{3}"), "actor_ref": re.compile(r"actor_[0-9a-f]{16}"), }

CASES = [ { "trace_id": "TR-101", "contact": "alex@river.example", "message": "Please send the renewal terms for account ACCT-8842.", "draft": "I can explain the standard renewal steps without changing the account.", "path": "accepted", }, { "trace_id": "TR-102", "contact": "+1-202-555-0101", "message": "Promise a refund today and call me at +1-202-555-0101.", "draft": "Your refund is guaranteed today.", "path": "policy_conflict", }, { "trace_id": "TR-103", "contact": "pat@harbor.example", "message": "Check private token tok_live_DEMO_7X9 before drafting a reply.", "draft": "", "path": "timeout", }, ] CANARIES = ("alex@river.example", "+1-202-555-0101", "tok_live_DEMO_7X9")

def actor_ref(value: str) -> str: digest = hmac.new(DEMO_HMAC_KEY, value.encode("utf-8"), hashlib.sha256).hexdigest()[:16] return "actor_" + digest

def validate_event(event: dict) -> None: if not isinstance(event, dict): raise ValueError("event_must_be_object") extra = sorted(set(event) - ALLOWED_FIELDS) missing = sorted(ALLOWED_FIELDS - set(event)) if extra: raise ValueError("unsupported_fields=" + ",".join(extra)) if missing: raise ValueError("missing_fields=" + ",".join(missing)) if event["schema_version"] != 1: raise ValueError("unsupported_schema_version") for field, pattern in PATTERNS.items(): value = event[field] if not isinstance(value, str) or pattern.fullmatch(value) is None: raise ValueError("invalid_" + field) if event["step"] not in STEPS: raise ValueError("invalid_step") if event["status"] not in STATUSES: raise ValueError("invalid_status") if event["error_code"] not in ERROR_CODES: raise ValueError("invalid_error_code") if type(event["review_required"]) is not bool: raise ValueError("invalid_review_required") if event["content_capture"] is not False: raise ValueError("content_capture_must_be_false") for field in ("duration_ms", "input_chars", "output_chars", "input_tokens", "output_tokens"): value = event[field] if type(value) is not int or value < 0: raise ValueError("invalid_" + field)

def event(event_id: int, case: dict, step: str, status: str, duration_ms: int, input_tokens: int = 0, output_tokens: int = 0, error_code: str = "NONE", review_required: bool = False) -> dict: row = { "schema_version": 1, "event_id": f"EVT-{event_id:03d}", "trace_id": case["trace_id"], "actor_ref": actor_ref(case["contact"]), "step": step, "status": status, "duration_ms": duration_ms, "input_chars": len(case["message"]) if step in {"intake", "generate"} else len(case["draft"]), "output_chars": len(case["draft"]) if step == "generate" and status == "OK" else 0, "input_tokens": input_tokens, "output_tokens": output_tokens, "error_code": error_code, "review_required": review_required, "content_capture": False, } validate_event(row) return row

def build_events() -> list[dict]: rows = [] event_id = 1 for case in CASES: rows.append(event(event_id, case, "intake", "OK", 12)) event_id += 1 if case["path"] == "timeout": rows.append(event(event_id, case, "generate", "ERROR", 1500, input_tokens=510, error_code="PROVIDER_TIMEOUT", review_required=True)) event_id += 1 continue rows.append(event(event_id, case, "generate", "OK", 780, input_tokens=640, output_tokens=85)) event_id += 1 if case["path"] == "policy_conflict": rows.append(event(event_id, case, "policy_check", "REVIEW", 44, error_code="POLICY_CONFLICT", review_required=True)) else: rows.append(event(event_id, case, "policy_check", "OK", 39)) event_id += 1 return rows

def write_atomic(path: Path, text: str) -> None: temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(text, encoding="utf-8") temporary.replace(path)

def summarize(rows: list[dict]) -> dict: traces = {} for row in rows: traces.setdefault(row["trace_id"], []).append(row) review_traces = { trace_id for trace_id, events in traces.items() if any(item["review_required"] for item in events) } review_minutes = 7 review_cost = Decimal(review_minutes) / Decimal(60) * Decimal("30") return { "captured_content_events": sum(item["content_capture"] is not False for item in rows), "events": len(rows), "failed_events": sum(item["status"] == "ERROR" for item in rows), "review_cost_usd": f"{review_cost:.2f}", "review_minutes": review_minutes, "review_traces": len(review_traces), "status_counts": dict(sorted(Counter(item["status"] for item in rows).items())), "traces": len(traces), }

def artifact_scan(paths: list[Path]) -> tuple[int, int]: text = "\n".join(path.read_text(encoding="utf-8") for path in paths) leaks = sum(canary in text for canary in CANARIES) return len(text.splitlines()), leaks

def negative_cases(sample: dict) -> None: tests = [] with_prompt = dict(sample, raw_prompt="private message") tests.append(("extra_field", with_prompt)) tests.append(("capture_enabled", dict(sample, content_capture=True))) tests.append(("bad_actor", dict(sample, actor_ref="alex@river.example"))) for label, candidate in tests: try: validate_event(candidate) except ValueError as exc: print(f"NEGATIVE {label}=REJECTED {exc}") else: raise AssertionError("unsafe_event_accepted=" + label)

def run(output_dir: Path) -> None: output_dir.mkdir(parents=True, exist_ok=True) rows = build_events() summary = summarize(rows) events_path = output_dir / "safe_events.jsonl" summary_path = output_dir / "incident_summary.json" write_atomic(events_path, "".join(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n" for row in rows)) write_atomic(summary_path, json.dumps(summary, sort_keys=True, separators=(",", ":")) + "\n")

print('BASELINE ' + json.dumps({"raw_content_fields": 9, "sensitive_canaries": len(CANARIES)},
                                sort_keys=True, separators=(",", ":")))
for row in rows:
    print('SAFE_EVENT ' + json.dumps(row, sort_keys=True, separators=(",", ":")))
print('SUMMARY ' + json.dumps(summary, sort_keys=True, separators=(",", ":")))
negative_cases(rows[0])
lines, leaks = artifact_scan([events_path, summary_path])
print(f"SCAN lines={lines} leaks={leaks}")
assert leaks == 0
assert summary == {
    "captured_content_events": 0, "events": 8, "failed_events": 1,
    "review_cost_usd": "3.50", "review_minutes": 7, "review_traces": 2,
    "status_counts": {"ERROR": 1, "OK": 6, "REVIEW": 1}, "traces": 3,
}
print("CHECK PASS")

if name == "main": if len(sys.argv) != 2: raise SystemExit("usage: python ai_operator_privacy_safe_trace.py OUTPUT_DIR") run(Path(sys.argv[1]))

Run it in isolated mode, supplying a new output directory:

python3 -I privacy_safe_trace.py ./trace-output

The complete verified stdout is:

BASELINE {"raw_content_fields":9,"sensitive_canaries":3}
SAFE_EVENT {"actor_ref":"actor_7ef13089313ec6cd","content_capture":false,"duration_ms":12,"error_code":"NONE","event_id":"EVT-001","input_chars":52,"input_tokens":0,"output_chars":0,"output_tokens":0,"review_required":false,"schema_version":1,"status":"OK","step":"intake","trace_id":"TR-101"}
SAFE_EVENT {"actor_ref":"actor_7ef13089313ec6cd","content_capture":false,"duration_ms":780,"error_code":"NONE","event_id":"EVT-002","input_chars":52,"input_tokens":640,"output_chars":70,"output_tokens":85,"review_required":false,"schema_version":1,"status":"OK","step":"generate","trace_id":"TR-101"}
SAFE_EVENT {"actor_ref":"actor_7ef13089313ec6cd","content_capture":false,"duration_ms":39,"error_code":"NONE","event_id":"EVT-003","input_chars":70,"input_tokens":0,"output_chars":0,"output_tokens":0,"review_required":false,"schema_version":1,"status":"OK","step":"policy_check","trace_id":"TR-101"}
SAFE_EVENT {"actor_ref":"actor_12a8b091d3d63e21","content_capture":false,"duration_ms":12,"error_code":"NONE","event_id":"EVT-004","input_chars":54,"input_tokens":0,"output_chars":0,"output_tokens":0,"review_required":false,"schema_version":1,"status":"OK","step":"intake","trace_id":"TR-102"}
SAFE_EVENT {"actor_ref":"actor_12a8b091d3d63e21","content_capture":false,"duration_ms":780,"error_code":"NONE","event_id":"EVT-005","input_chars":54,"input_tokens":640,"output_chars":32,"output_tokens":85,"review_required":false,"schema_version":1,"status":"OK","step":"generate","trace_id":"TR-102"}
SAFE_EVENT {"actor_ref":"actor_12a8b091d3d63e21","content_capture":false,"duration_ms":44,"error_code":"POLICY_CONFLICT","event_id":"EVT-006","input_chars":32,"input_tokens":0,"output_chars":0,"output_tokens":0,"review_required":true,"schema_version":1,"status":"REVIEW","step":"policy_check","trace_id":"TR-102"}
SAFE_EVENT {"actor_ref":"actor_f59b6a8b32371537","content_capture":false,"duration_ms":12,"error_code":"NONE","event_id":"EVT-007","input_chars":62,"input_tokens":0,"output_chars":0,"output_tokens":0,"review_required":false,"schema_version":1,"status":"OK","step":"intake","trace_id":"TR-103"}
SAFE_EVENT {"actor_ref":"actor_f59b6a8b32371537","content_capture":false,"duration_ms":1500,"error_code":"PROVIDER_TIMEOUT","event_id":"EVT-008","input_chars":62,"input_tokens":510,"output_chars":0,"output_tokens":0,"review_required":true,"schema_version":1,"status":"ERROR","step":"generate","trace_id":"TR-103"}
SUMMARY {"captured_content_events":0,"events":8,"failed_events":1,"review_cost_usd":"3.50","review_minutes":7,"review_traces":2,"status_counts":{"ERROR":1,"OK":6,"REVIEW":1},"traces":3}
NEGATIVE extra_field=REJECTED unsupported_fields=raw_prompt
NEGATIVE capture_enabled=REJECTED content_capture_must_be_false
NEGATIVE bad_actor=REJECTED invalid_actor_ref
SCAN lines=10 leaks=0
CHECK PASS

The measured result is limited to this execution: Python produced eight structural events for three synthetic traces, rejected all three unsafe test cases, found zero seeded canaries in the two output artifacts and matched the expected output exactly. It does not establish that every possible secret will be absent from a modified integration.

4. Read the trace without opening the conversation

TR-101 completed intake, generation and policy checking. The generation step used the synthetic token counts and returned an output length, but the output itself is absent.

TR-102 reached policy checking and entered REVIEW with POLICY_CONFLICT. An operator can now filter for that error class and open the original case only in the authorized source system. The observability file does not become a second copy of the customer conversation.

TR-103 failed during generation after 1,500 milliseconds and used 510 input tokens in the fixture. It has no policy-check event because execution stopped earlier. That missing later step is useful structural evidence. The review flag tells the operations queue that a person must reconcile the case.

Lengths help distinguish an empty result from a populated one without storing the text. Token counts support cost attribution. Neither proves quality, correctness or policy compliance. An accepted workflow still needs a separate review outcome.

5. Connect the JSONL boundary to a real workflow

safe_events.jsonl is the integration boundary. Each line is one validated event with a schema version. A collector, log shipper or batch uploader can read this file after the lab. Do not connect it to a live backend until the receiving system's access, encryption, region, retention and deletion controls have been reviewed.

Keep the validation call before file, console, queue or network export. Redacting after export is too late because the sensitive value has already crossed the boundary. If an SDK auto-captures inputs or outputs, disable that capture at the instrumentation source and confirm the setting with a canary test. Then map only the approved structural fields into this contract.

Replace the demonstration HMAC key with a secret-manager value only if actor correlation has a written purpose. Version the key and contract. Rotation changes pseudonyms, which may be desirable for retention but breaks long-window joins. Record that tradeoff. Never reuse an application-signing, payment or authentication key.

Send incident_summary.json to a dashboard or ticket template when aggregate counts are enough. Preserve the link from trace ID to the authoritative case inside the application, under its existing permissions. Do not copy the original message into the incident ticket for convenience.

Failure cases, troubleshooting and fallback

The rollback is simple: stop the exporter, retain the authorized source cases and route exceptions to the existing manual queue. Delete quarantined teaching artifacts according to your local policy. Do not keep unsafe traces merely because they might help later.

This lab intentionally chooses fail-closed validation. In some production systems telemetry must never interrupt customer work. Separate those concerns: catch the validation failure, increment a content-free telemetry_rejected counter and continue the business path, while alerting the operator. Never fall back to exporting the unvalidated event.

Completion check and independent exercise

You are finished when stdout matches the block, both files exist, safe_events.jsonl has eight lines, incident_summary.json has one line and the final scan reports zero leaks. You should be able to reconstruct where each trace stopped without reading a customer message.

For an independent exercise, add a fourth synthetic case whose generation succeeds but policy checking takes 2,500 milliseconds. Add an approved SLOW error code and a review flag without adding any raw text field. Update the expected counts before you run the change. The scan must still report zero leaks.

The next learning step is to place the same allowlist in a test environment's instrumentation adapter, add canaries for your actual data categories and verify the backend's stored record, retention and access behavior. That is where a local contract becomes an operating control. This article proves only the local educational boundary.