Llm Safety Patterns

LLM Safety Patterns

The Core Principle

Identifiers flow AROUND the LLM, not THROUGH it.

The LLM sees only content. Attribution happens deterministically.

Why This Matters

When identifiers appear in prompts, bad things happen:

1. Hallucination: LLM invents IDs that don't exist 2. Confusion: LLM mixes up which ID belongs where 3. Injection: Attacker manipulates IDs via prompt injection 4. Leakage: IDs appear in logs, caches, traces 5. Cross-tenant: LLM could reference other users' data

The Architecture

┌─────────────────────────────────────────────────────────────────────────┐
│                                                                         │
│   SYSTEM CONTEXT (flows around LLM)                                     │
│   ┌─────────────────────────────────────────────────────────────────┐   │
│   │ user_id │ tenant_id │ analysis_id │ trace_id │ permissions     │   │
│   └─────────────────────────────────────────────────────────────────┘   │
│        │                                                       │        │
│        │                                                       │        │
│        ▼                                                       ▼        │
│   ┌─────────┐                                           ┌─────────┐    │
│   │ PRE-LLM │       ┌─────────────────────┐            │POST-LLM │    │
│   │ FILTER  │──────▶│        LLM          │───────────▶│ATTRIBUTE│    │
│   │         │       │                     │            │         │    │
│   │ Returns │       │ Sees ONLY:          │            │ Adds:   │    │
│   │ CONTENT │       │ - content text      │            │ - IDs   │    │
│   │ (no IDs)│       │ - context text      │            │ - refs  │    │
│   └─────────┘       │ (NO IDs!)           │            └─────────┘    │
│                     └─────────────────────┘                            │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

What NEVER Goes in Prompts

OrchestKit Forbidden Parameters

Detection Pattern

import re

FORBIDDEN_PATTERNS = [
    r'user[_-]?id',
    r'tenant[_-]?id',
    r'analysis[_-]?id',
    r'document[_-]?id',
    r'artifact[_-]?id',
    r'chunk[_-]?id',
    r'session[_-]?id',
    r'trace[_-]?id',
    r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}',
]

def audit_prompt(prompt: str) -> list[str]:
    """Check for forbidden patterns in prompt"""
    violations = []
    for pattern in FORBIDDEN_PATTERNS:
        if re.search(pattern, prompt, re.IGNORECASE):
            violations.append(pattern)
    return violations

The Three-Phase Pattern

Phase 1: Pre-LLM (Filter & Extract)

async def prepare_for_llm(
    query: str,
    ctx: RequestContext,
) -> tuple[str, list[str], SourceRefs]:
    """
    Filter data and extract content for LLM.
    Returns: (content, context_texts, source_references)
    """
    # 1. Retrieve with tenant filter
    documents = await semantic_search(
        query_embedding=embed(query),
        ctx=ctx,  # Filters by tenant_id, user_id
    )

    # 2. Save references for attribution
    source_refs = SourceRefs(
        document_ids=[d.id for d in documents],
        chunk_ids=[c.id for c in chunks],
    )

    # 3. Extract content only (no IDs)
    content_texts = [d.content for d in documents]

    return query, content_texts, source_refs

Phase 2: LLM Call (Content Only)

def build_prompt(content: str, context_texts: list[str]) -> str:
    """
    Build prompt with ONLY content, no identifiers.
    """
    prompt = f"""
    Analyze the following content and provide insights.

    CONTENT:
    {content}

    RELEVANT CONTEXT:
    {chr(10).join(f"- {text}" for text in context_texts)}

    Provide analysis covering:
    1. Key concepts
    2. Prerequisites
    3. Learning objectives
    """

    # AUDIT: Verify no IDs leaked
    violations = audit_prompt(prompt)
    if violations:
        raise SecurityError(f"IDs leaked to prompt: {violations}")

    return prompt

async def call_llm(prompt: str) -> dict:
    """LLM only sees content, never IDs"""
    response = await llm.generate(prompt)
    return parse_response(response)

Phase 3: Post-LLM (Attribute)

async def save_with_attribution(
    llm_output: dict,
    ctx: RequestContext,
    source_refs: SourceRefs,
) -> Analysis:
    """
    Attach context and references to LLM output.
    Attribution is deterministic, not LLM-generated.
    """
    return await Analysis.create(
        # Generated
        id=uuid4(),

        # From RequestContext (system-provided)
        user_id=ctx.user_id,
        tenant_id=ctx.tenant_id,
        analysis_id=ctx.resource_id,
        trace_id=ctx.trace_id,

        # From Pre-LLM refs (deterministic)
        source_document_ids=source_refs.document_ids,
        source_chunk_ids=source_refs.chunk_ids,

        # From LLM (content only)
        content=llm_output["analysis"],
        key_concepts=llm_output["key_concepts"],
        difficulty=llm_output["difficulty"],

        # Metadata
        created_at=datetime.now(timezone.utc),
        model_used=MODEL_NAME,
    )

Output Validation

After LLM returns, validate:

1. Schema: Response matches expected structure 2. Guardrails: No toxic/harmful content 3. Grounding: Claims are supported by provided context 4. No IDs: LLM didn't hallucinate any IDs

async def validate_output(
    llm_output: dict,
    context_texts: list[str],
) -> ValidationResult:
    """Validate LLM output before use"""

    # 1. Schema validation
    try:
        parsed = AnalysisOutput.model_validate(llm_output)
    except ValidationError as e:
        return ValidationResult(valid=False, reason=f"Schema error: {e}")

    # 2. Guardrails
    if await contains_toxic_content(parsed.content):
        return ValidationResult(valid=False, reason="Toxic content detected")

    # 3. Grounding check
    if not is_grounded(parsed.content, context_texts):
        return ValidationResult(valid=False, reason="Ungrounded claims")

    # 4. No hallucinated IDs
    if contains_uuid_pattern(parsed.content):
        return ValidationResult(valid=False, reason="Hallucinated IDs")

    return ValidationResult(valid=True)

Integration Points in OrchestKit

Content Analysis Workflow

backend/app/workflows/
├── agents/
│   ├── execution.py        # Add context separation
│   └── prompts/            # Audit all prompts
├── tasks/
│   └── generate_artifact.py  # Add attribution

Services

backend/app/services/
├── embeddings/            # Pre-LLM filtering
└── analysis/              # Post-LLM attribution

Checklist Before Any LLM Call

• [ ] RequestContext available
• [ ] Data filtered by tenant_id and user_id
• [ ] Content extracted without IDs
• [ ] Source references saved
• [ ] Prompt passes audit (no forbidden patterns)
• [ ] Output validated before use
• [ ] Attribution uses context, not LLM output

---

Related Skills

• input-validation - Input sanitization patterns that complement LLM safety
• rag-retrieval - RAG pipeline patterns requiring tenant-scoped retrieval
• llm-evaluation - Output quality assessment including hallucination detection
• security-scanning - Automated security scanning for LLM integrations
• defense-in-depth - 8-layer security architecture including Tavily prompt injection firewall at Layer 2

Key Decisions

Version: 1.0.0 (December 2025)

Capability Details

context-separation

Keywords: context separation, prompt context, id in prompt, parameterized Solves:

• How do I prevent IDs from leaking into prompts?
• How do I separate system context from prompt content?
• What should never appear in LLM prompts?

pre-llm-filtering

Keywords: pre-llm, rag filter, data filter, tenant filter Solves:

• How do I filter data before sending to LLM?
• How do I ensure tenant isolation in RAG?
• How do I scope retrieval to current user?

post-llm-attribution

Keywords: attribution, source tracking, provenance, citation Solves:

• How do I track which sources the LLM used?
• How do I attribute results correctly?
• How do I avoid LLM-generated IDs?

output-guardrails

Keywords: guardrail, output validation, hallucination, toxicity Solves:

• How do I validate LLM output?
• How do I detect hallucinations?
• How do I prevent toxic content generation?

prompt-audit

Keywords: prompt audit, prompt security, prompt injection Solves:

• How do I verify no IDs leaked to prompts?
• How do I audit prompts for security?
• How do I prevent prompt injection?

Pre-LLM Call Checklist

Before ANY LLM Call in OrchestKit

Use this checklist before sending any prompt to an LLM:

Phase 1: Context Available

• [ ] RequestContext obtained from JWT (not user input)
• [ ] user_id available in context
• [ ] tenant_id available in context
• [ ] trace_id set for observability

Phase 2: Data Isolation

• [ ] Query includes WHERE tenant_id = :tenant_id
• [ ] Query includes WHERE user_id = :user_id (if user-scoped)
• [ ] Vector search filtered by tenant
• [ ] Full-text search filtered by tenant

Phase 3: Source References Captured

• [ ] document_ids saved for attribution
• [ ] chunk_ids saved for attribution
• [ ] Retrieval timestamp recorded
• [ ] Similarity scores captured (for debugging)

Phase 4: Content Extraction

• [ ] Only content text extracted (no metadata with IDs)
• [ ] Content stripped of any embedded UUIDs
• [ ] Content stripped of any ID field names

Phase 5: Prompt Building

• [ ] Prompt contains ONLY content text
• [ ] No user_id in prompt
• [ ] No tenant_id in prompt
• [ ] No analysis_id in prompt
• [ ] No document_id in prompt
• [ ] No UUIDs in prompt
• [ ] No API keys or secrets in prompt

Phase 6: Prompt Audit

• [ ] audit_prompt() called on final prompt
• [ ] No critical violations detected
• [ ] Warnings logged for review

Phase 7: LLM Call

• [ ] Timeout configured
• [ ] Error handling in place
• [ ] Response parsing ready
• [ ] Langfuse trace started

---

Quick Verification Script

from llm_safety import audit_prompt, has_critical_violations

def verify_llm_ready(
    prompt: str,
    ctx: RequestContext,
    source_refs: SourceReference,
) -> bool:
    """Quick verification before LLM call"""

    # Check context
    assert ctx.user_id is not None, "Missing user_id"
    assert ctx.tenant_id is not None, "Missing tenant_id"

    # Check source refs captured
    assert len(source_refs.document_ids) >= 0, "Source refs not captured"

    # Audit prompt
    violations = audit_prompt(prompt)
    if has_critical_violations(violations):
        raise PromptSecurityError(violations)

    return True

---

Post-LLM Attribution Checklist

After LLM returns:

• [ ] Output parsed with schema validation
• [ ] Output checked for hallucinated IDs
• [ ] Output checked for grounding
• [ ] Content safety validated
• [ ] Attribution attached from RequestContext
• [ ] Source links created from captured refs
• [ ] Audit event logged
• [ ] Langfuse trace completed

---

Sign-off: Run verify_llm_ready() before every LLM call

LLM Safety Checklist

Input Safety

• [ ] Validate input length
• [ ] Detect prompt injection attempts
• [ ] Sanitize user content
• [ ] Rate limit requests

Output Safety

• [ ] Content filtering
• [ ] PII detection and redaction
• [ ] Harmful content detection
• [ ] Bias monitoring

System Prompts

• [ ] Clear boundaries
• [ ] Role definition
• [ ] Refusal instructions
• [ ] No secrets in prompts

Guardrails

• [ ] Input guardrails
• [ ] Output guardrails
• [ ] Topic restrictions
• [ ] Sensitive content handling

Monitoring

• [ ] Log flagged content
• [ ] Alert on violations
• [ ] Human review queue
• [ ] Incident response plan

Context Separation Pattern

The Problem

When identifiers appear in LLM prompts, several security issues arise:

┌─────────────────────────────────────────────────────────┐
│  WHAT HAPPENS WHEN IDs GO INTO PROMPTS                  │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  "Analyze document doc_abc123 for user usr_xyz789"      │
│                     │                    │              │
│                     ▼                    ▼              │
│              ┌──────────────────────────────┐           │
│              │           LLM                │           │
│              │                              │           │
│              │  May hallucinate:            │           │
│              │  - doc_abc124 (off by one)   │           │
│              │  - doc_xyz789 (mixed up)     │           │
│              │  - usr_other (cross-tenant)  │           │
│              └──────────────────────────────┘           │
│                                                         │
│  RISKS:                                                 │
│  • Hallucinated IDs don't exist → crashes              │
│  • Mixed IDs → wrong data attribution                  │
│  • Cross-tenant IDs → security breach                  │
│  • IDs in logs/traces → data leakage                   │
│                                                         │
└─────────────────────────────────────────────────────────┘

The Solution: Context Separation

┌─────────────────────────────────────────────────────────┐
│  CORRECT: CONTEXT FLOWS AROUND LLM                      │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  RequestContext ─────────────────────────────────────►  │
│  (user_id, tenant_id, etc.)                    │        │
│         │                                      │        │
│         │   ┌──────────────────────┐          │        │
│         │   │                      │          │        │
│         ▼   │       LLM            │          ▼        │
│  ┌──────────┤                      ├─────────────┐     │
│  │ Content  │  Sees ONLY:          │  Content +  │     │
│  │ (text)   │  - Document text     │  Context    │     │
│  │          │  - Query text        │  (merged)   │     │
│  └──────────┤  - Instructions      ├─────────────┘     │
│             │                      │                    │
│             │  NO IDs!             │                    │
│             └──────────────────────┘                    │
│                                                         │
└─────────────────────────────────────────────────────────┘

Implementation

1. Define What's Forbidden

# OrchestKit parameters that NEVER go in prompts
FORBIDDEN_IN_PROMPTS = {
    # User identity
    "user_id",      # UUID - hallucination risk
    "tenant_id",    # UUID - cross-tenant risk
    "session_id",   # String - auth context

    # Resource references
    "analysis_id",  # UUID - job tracking
    "document_id",  # UUID - source tracking
    "artifact_id",  # UUID - output tracking
    "chunk_id",     # UUID - RAG reference

    # System context
    "trace_id",     # String - observability
    "request_id",   # String - request tracking
    "workflow_run_id",  # UUID - workflow tracking

    # Secrets
    "api_key",      # String - never!
    "token",        # String - never!
}

2. Separate Context from Content

from dataclasses import dataclass
from uuid import UUID

@dataclass
class ContentPayload:
    """What the LLM sees - content only"""
    query: str
    context_texts: list[str]
    instructions: str

@dataclass
class ContextPayload:
    """What flows around the LLM - never in prompt"""
    user_id: UUID
    tenant_id: UUID
    analysis_id: UUID
    source_refs: list[UUID]
    trace_id: str

async def analyze_content(
    content: ContentPayload,
    context: ContextPayload,
) -> AnalysisResult:
    """
    Content goes TO the LLM.
    Context goes AROUND the LLM.
    """
    # Build prompt from content only
    prompt = build_prompt(
        query=content.query,
        context_texts=content.context_texts,
        instructions=content.instructions,
        # NO context payload fields here!
    )

    # LLM sees content only
    llm_output = await llm.generate(prompt)

    # Reattach context to output
    return AnalysisResult(
        content=llm_output,
        user_id=context.user_id,      # From context
        tenant_id=context.tenant_id,   # From context
        analysis_id=context.analysis_id,  # From context
        sources=context.source_refs,   # From context
    )

3. Audit Prompts Before Sending

import re

UUID_PATTERN = r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'

def audit_prompt(prompt: str) -> list[str]:
    """
    Check for forbidden patterns before sending to LLM.
    Raises if any IDs detected.
    """
    violations = []

    # Check for UUIDs
    if re.search(UUID_PATTERN, prompt, re.IGNORECASE):
        violations.append("UUID detected in prompt")

    # Check for ID field names
    for forbidden in FORBIDDEN_IN_PROMPTS:
        pattern = rf'\b{forbidden}\b'
        if re.search(pattern, prompt, re.IGNORECASE):
            violations.append(f"Forbidden field '{forbidden}' in prompt")

    return violations

# Usage in prompt building
def build_safe_prompt(content: ContentPayload) -> str:
    prompt = f"""
    Analyze the following content:

    {content.query}

    Context:
    {chr(10).join(content.context_texts)}
    """

    # Audit before returning
    violations = audit_prompt(prompt)
    if violations:
        raise PromptSecurityError(
            f"Prompt contains forbidden content: {violations}"
        )

    return prompt

OrchestKit Integration Points

Content Analysis Workflow

# backend/app/workflows/agents/content_analyzer.py

async def analyze(state: AnalysisState) -> AnalysisState:
    # Context is in state, but NOT passed to prompt
    ctx = state.request_context

    # Build content-only payload
    content = ContentPayload(
        query=state.analysis_request.query,
        context_texts=[doc.content for doc in state.retrieved_docs],
        instructions=get_analysis_instructions(),
    )

    # Context payload for attribution
    context = ContextPayload(
        user_id=ctx.user_id,
        tenant_id=ctx.tenant_id,
        analysis_id=state.analysis_id,
        source_refs=[doc.id for doc in state.retrieved_docs],
        trace_id=ctx.trace_id,
    )

    result = await analyze_content(content, context)
    return state.with_result(result)

Common Mistakes

# ❌ BAD: ID in prompt
prompt = f"Analyze document {doc_id} for user {user_id}"

# ❌ BAD: ID in f-string
prompt = f"Context from analysis {analysis_id}:\n{context}"

# ❌ BAD: ID in instruction
prompt = f"You are analyzing for tenant {tenant_id}. Be helpful."

# ✅ GOOD: Content only
prompt = f"Analyze the following document:\n{document_content}"

# ✅ GOOD: No IDs visible
prompt = f"""
Analyze this content and provide insights:

{content}

Relevant context:
{context_texts}
"""

Testing Context Separation

import pytest

class TestContextSeparation:

    def test_prompt_contains_no_uuids(self):
        content = ContentPayload(
            query="What are the key concepts?",
            context_texts=["Machine learning basics..."],
            instructions="Provide clear analysis",
        )

        prompt = build_safe_prompt(content)

        assert not re.search(UUID_PATTERN, prompt)

    def test_prompt_contains_no_forbidden_fields(self):
        content = ContentPayload(...)
        prompt = build_safe_prompt(content)

        for forbidden in FORBIDDEN_IN_PROMPTS:
            assert forbidden not in prompt.lower()

    def test_audit_catches_leaked_uuid(self):
        bad_prompt = "Analyze doc 123e4567-e89b-12d3-a456-426614174000"

        violations = audit_prompt(bad_prompt)

        assert len(violations) > 0
        assert "UUID" in violations[0]

Output Guardrails

Purpose

After LLM returns, validate the output before using it:

┌────────────────────────────────────────────────────────────┐
│                  OUTPUT VALIDATION                         │
├────────────────────────────────────────────────────────────┤
│                                                            │
│  LLM Response ──► Guardrails ──► Validated Output          │
│                       │                                    │
│                       ▼                                    │
│              ┌────────────────┐                            │
│              │   VALIDATORS   │                            │
│              ├────────────────┤                            │
│              │ □ Schema       │  Does it match expected?   │
│              │ □ No IDs       │  No hallucinated UUIDs?    │
│              │ □ Grounded     │  Supported by context?     │
│              │ □ Safe         │  No toxic content?         │
│              │ □ Size         │  Within limits?            │
│              └────────────────┘                            │
│                       │                                    │
│           ┌──────────┴──────────┐                         │
│           ▼                     ▼                         │
│    ┌──────────┐          ┌──────────┐                     │
│    │   PASS   │          │   FAIL   │                     │
│    │          │          │          │                     │
│    │ Continue │          │ Retry or │                     │
│    │          │          │ Error    │                     │
│    └──────────┘          └──────────┘                     │
│                                                            │
└────────────────────────────────────────────────────────────┘

Implementation

1. Validation Result Type

from dataclasses import dataclass
from enum import Enum

class ValidationStatus(Enum):
    PASSED = "passed"
    FAILED = "failed"
    WARNING = "warning"

@dataclass
class ValidationResult:
    status: ValidationStatus
    reason: str | None = None
    details: dict | None = None

    @property
    def is_valid(self) -> bool:
        return self.status in (ValidationStatus.PASSED, ValidationStatus.WARNING)

2. Schema Validation

from pydantic import BaseModel, ValidationError
from typing import TypeVar

T = TypeVar("T", bound=BaseModel)

def validate_schema(
    llm_output: dict,
    schema: type[T],
) -> tuple[T | None, ValidationResult]:
    """
    Validate LLM output matches expected schema.
    """
    try:
        parsed = schema.model_validate(llm_output)
        return parsed, ValidationResult(
            status=ValidationStatus.PASSED,
        )
    except ValidationError as e:
        return None, ValidationResult(
            status=ValidationStatus.FAILED,
            reason=f"Schema validation failed: {e.error_count()} errors",
            details={"errors": e.errors()},
        )

# Usage
class AnalysisOutput(BaseModel):
    summary: str
    key_concepts: list[str]
    difficulty: str

parsed, result = validate_schema(llm_response, AnalysisOutput)
if not result.is_valid:
    raise ValidationError(result.reason)

3. No Hallucinated IDs

import re

UUID_PATTERN = r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'

def validate_no_ids(output: str) -> ValidationResult:
    """
    Ensure LLM didn't hallucinate any identifiers.
    """
    # Check for UUIDs
    uuids = re.findall(UUID_PATTERN, output, re.IGNORECASE)
    if uuids:
        return ValidationResult(
            status=ValidationStatus.FAILED,
            reason=f"Found {len(uuids)} hallucinated UUIDs",
            details={"uuids": uuids},
        )

    # Check for ID-like patterns
    id_patterns = [
        r'user_id[:\s]+\S+',
        r'doc_id[:\s]+\S+',
        r'id[:\s]+[a-f0-9]{8,}',
    ]

    for pattern in id_patterns:
        matches = re.findall(pattern, output, re.IGNORECASE)
        if matches:
            return ValidationResult(
                status=ValidationStatus.WARNING,
                reason=f"Found ID-like pattern: {matches[0]}",
                details={"matches": matches},
            )

    return ValidationResult(status=ValidationStatus.PASSED)

4. Grounding Validation

def validate_grounding(
    output: str,
    context_texts: list[str],
    threshold: float = 0.3,
) -> ValidationResult:
    """
    Check if LLM output is grounded in provided context.
    Uses simple keyword overlap for speed.
    """
    # Extract key terms from output
    output_terms = set(extract_key_terms(output))

    # Extract key terms from context
    context_terms = set()
    for text in context_texts:
        context_terms.update(extract_key_terms(text))

    # Calculate overlap
    if not output_terms:
        return ValidationResult(
            status=ValidationStatus.WARNING,
            reason="No key terms in output",
        )

    overlap = len(output_terms & context_terms) / len(output_terms)

    if overlap < threshold:
        return ValidationResult(
            status=ValidationStatus.WARNING,
            reason=f"Low grounding score: {overlap:.2%}",
            details={
                "overlap": overlap,
                "threshold": threshold,
                "ungrounded_terms": list(output_terms - context_terms)[:10],
            },
        )

    return ValidationResult(
        status=ValidationStatus.PASSED,
        details={"grounding_score": overlap},
    )

def extract_key_terms(text: str) -> list[str]:
    """Extract meaningful terms from text"""
    import re
    # Simple: words 4+ chars, lowercased
    words = re.findall(r'\b[a-zA-Z]{4,}\b', text.lower())
    # Filter common words
    stopwords = {'this', 'that', 'with', 'from', 'have', 'been', 'will', 'would'}
    return [w for w in words if w not in stopwords]

5. Content Safety

async def validate_content_safety(
    output: str,
) -> ValidationResult:
    """
    Check for toxic/harmful content.
    Uses simple pattern matching + optional LLM check.
    """
    # Quick pattern check
    toxic_patterns = [
        r'\b(hate|violence|harm|kill)\b',
        r'\b(password|secret|api.?key)\b',
    ]

    for pattern in toxic_patterns:
        if re.search(pattern, output, re.IGNORECASE):
            return ValidationResult(
                status=ValidationStatus.FAILED,
                reason=f"Potentially unsafe content detected",
            )

    # PII detection
    pii_patterns = {
        "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        "phone": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
        "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
    }

    detected_pii = []
    for pii_type, pattern in pii_patterns.items():
        if re.search(pattern, output):
            detected_pii.append(pii_type)

    if detected_pii:
        return ValidationResult(
            status=ValidationStatus.WARNING,
            reason=f"Potential PII detected: {detected_pii}",
            details={"pii_types": detected_pii},
        )

    return ValidationResult(status=ValidationStatus.PASSED)

6. Size Limits

def validate_size(
    output: str,
    max_chars: int = 50000,
    max_tokens: int = 10000,
) -> ValidationResult:
    """
    Ensure output is within size limits.
    """
    if len(output) > max_chars:
        return ValidationResult(
            status=ValidationStatus.FAILED,
            reason=f"Output exceeds {max_chars} chars: {len(output)}",
        )

    # Rough token estimate
    estimated_tokens = len(output) // 4
    if estimated_tokens > max_tokens:
        return ValidationResult(
            status=ValidationStatus.WARNING,
            reason=f"Output may exceed token limit: ~{estimated_tokens}",
        )

    return ValidationResult(status=ValidationStatus.PASSED)

7. Combined Validator

from dataclasses import dataclass

@dataclass
class GuardrailsConfig:
    validate_schema: bool = True
    validate_no_ids: bool = True
    validate_grounding: bool = True
    validate_safety: bool = True
    validate_size: bool = True
    grounding_threshold: float = 0.3
    max_output_chars: int = 50000

async def run_guardrails(
    llm_output: dict,
    context_texts: list[str],
    schema: type[BaseModel],
    config: GuardrailsConfig = GuardrailsConfig(),
) -> tuple[BaseModel | None, list[ValidationResult]]:
    """
    Run all guardrails on LLM output.
    Returns parsed output and all validation results.
    """
    results = []
    parsed = None

    # 1. Schema validation
    if config.validate_schema:
        parsed, result = validate_schema(llm_output, schema)
        results.append(result)
        if not result.is_valid:
            return None, results  # Stop early

    output_str = str(llm_output)

    # 2. No hallucinated IDs
    if config.validate_no_ids:
        result = validate_no_ids(output_str)
        results.append(result)

    # 3. Grounding check
    if config.validate_grounding:
        result = validate_grounding(
            output_str,
            context_texts,
            config.grounding_threshold,
        )
        results.append(result)

    # 4. Content safety
    if config.validate_safety:
        result = await validate_content_safety(output_str)
        results.append(result)

    # 5. Size limits
    if config.validate_size:
        result = validate_size(output_str, config.max_output_chars)
        results.append(result)

    # Check for failures
    failures = [r for r in results if r.status == ValidationStatus.FAILED]
    if failures:
        return None, results

    return parsed, results

OrchestKit Integration

# backend/app/workflows/agents/content_analyzer.py

async def analyze_with_guardrails(state: AnalysisState) -> AnalysisState:
    """Run LLM with output guardrails"""

    # Call LLM
    llm_response = await llm.generate(state.prompt)

    # Run guardrails
    parsed, validations = await run_guardrails(
        llm_output=llm_response,
        context_texts=state.context_texts,
        schema=AnalysisOutput,
    )

    # Log validations
    for v in validations:
        if v.status != ValidationStatus.PASSED:
            logger.warning(
                "guardrail_issue",
                status=v.status.value,
                reason=v.reason,
                trace_id=state.request_context.trace_id,
            )

    if parsed is None:
        raise GuardrailError(
            "LLM output failed validation",
            validations=[v for v in validations if not v.is_valid],
        )

    return state.with_output(parsed)

Common Mistakes

# ❌ BAD: No validation
artifact.content = llm_response["content"]  # Could be anything!

# ❌ BAD: Only schema validation
parsed = AnalysisOutput.parse_obj(response)  # Ignores content issues

# ❌ BAD: Trusting LLM completely
if llm_response.get("is_safe", True):  # LLM said it's safe!
    use_response(llm_response)

# ✅ GOOD: Full guardrail pipeline
parsed, results = await run_guardrails(
    llm_output=response,
    context_texts=context,
    schema=AnalysisOutput,
)

Testing Guardrails

class TestGuardrails:

    def test_detects_hallucinated_uuid(self):
        output = "Analysis for doc 123e4567-e89b-12d3-a456-426614174000"
        result = validate_no_ids(output)
        assert result.status == ValidationStatus.FAILED

    def test_detects_low_grounding(self):
        output = "This is about quantum physics and black holes"
        context = ["Python programming tutorial"]
        result = validate_grounding(output, context)
        assert result.status == ValidationStatus.WARNING

    async def test_detects_pii(self):
        output = "Contact john@example.com for details"
        result = await validate_content_safety(output)
        assert result.status == ValidationStatus.WARNING
        assert "email" in result.details["pii_types"]

    async def test_full_pipeline_passes(self):
        valid_output = {
            "summary": "Introduction to machine learning",
            "key_concepts": ["ML", "training", "models"],
            "difficulty": "intermediate",
        }
        context = ["Machine learning is a subset of AI..."]

        parsed, results = await run_guardrails(
            llm_output=valid_output,
            context_texts=context,
            schema=AnalysisOutput,
        )

        assert parsed is not None
        assert all(r.is_valid for r in results)

Post-LLM Attribution

The Principle

Attribution is DETERMINISTIC, not LLM-generated.

>

The LLM produces content. We attach context from our records.

┌────────────────────────────────────────────────────────────┐
│                   POST-LLM PHASE                           │
├────────────────────────────────────────────────────────────┤
│                                                            │
│                ┌─────────────────────┐                     │
│                │       LLM           │                     │
│                │                     │                     │
│                │  Output: content    │                     │
│                │  (text, analysis)   │                     │
│                └──────────┬──────────┘                     │
│                           │                                │
│                           ▼                                │
│              ┌────────────────────────┐                    │
│              │   ATTRIBUTION LAYER    │                    │
│              │                        │                    │
│  From Pre-LLM:                        From Context:        │
│  ├─ source_refs ─────────────────────► source_ids         │
│  └─ chunk_ids                          │                   │
│                                        │                   │
│  From RequestContext:                  │                   │
│  ├─ user_id ─────────────────────────► user_id            │
│  ├─ tenant_id ───────────────────────► tenant_id          │
│  ├─ trace_id ────────────────────────► trace_id           │
│  └─ analysis_id ─────────────────────► analysis_id        │
│                                        │                   │
│  Generated:                            │                   │
│  ├─ new UUID ────────────────────────► artifact_id        │
│  └─ timestamp ───────────────────────► created_at         │
│              │                        │                    │
│              └────────────┬───────────┘                    │
│                           │                                │
│                           ▼                                │
│              ┌────────────────────────┐                    │
│              │    COMPLETE RESULT     │                    │
│              │                        │                    │
│              │  content + attribution │                    │
│              │  (ready for storage)   │                    │
│              └────────────────────────┘                    │
│                                                            │
└────────────────────────────────────────────────────────────┘

Implementation

1. Attribution Data Structure

from dataclasses import dataclass
from datetime import datetime
from uuid import UUID, uuid4

@dataclass
class AttributedResult:
    """LLM output with deterministic attribution"""

    # Generated identifier
    id: UUID

    # From RequestContext (system-provided)
    user_id: UUID
    tenant_id: UUID
    analysis_id: UUID
    trace_id: str

    # From Pre-LLM refs (deterministic)
    source_document_ids: list[UUID]
    source_chunk_ids: list[UUID]

    # From LLM (content only)
    content: str
    key_concepts: list[str]
    difficulty_level: str
    summary: str

    # Metadata
    created_at: datetime
    model_used: str
    processing_time_ms: float

2. Attribution Function

async def attribute_llm_output(
    llm_output: dict,
    ctx: RequestContext,
    source_refs: SourceReference,
    model_name: str,
    processing_time_ms: float,
) -> AttributedResult:
    """
    Attach context to LLM output.
    All attribution comes from our records, not the LLM.
    """

    # Validate LLM output has no IDs
    if contains_identifiers(llm_output):
        raise SecurityError("LLM output contains identifiers")

    return AttributedResult(
        # New ID for this artifact
        id=uuid4(),

        # From RequestContext (verified from JWT)
        user_id=ctx.user_id,
        tenant_id=ctx.tenant_id,
        analysis_id=ctx.resource_id,
        trace_id=ctx.trace_id,

        # From Pre-LLM capture (deterministic)
        source_document_ids=source_refs.document_ids,
        source_chunk_ids=source_refs.chunk_ids,

        # From LLM (content only)
        content=llm_output["analysis"],
        key_concepts=llm_output.get("key_concepts", []),
        difficulty_level=llm_output.get("difficulty", "intermediate"),
        summary=llm_output.get("summary", ""),

        # Metadata
        created_at=datetime.now(timezone.utc),
        model_used=model_name,
        processing_time_ms=processing_time_ms,
    )

def contains_identifiers(output: dict) -> bool:
    """Check if LLM output contains any identifiers"""
    import re

    output_str = str(output)

    # Check for UUIDs
    if re.search(UUID_PATTERN, output_str):
        return True

    # Check for ID field names in content
    for field in ["user_id", "tenant_id", "document_id"]:
        if field in output_str.lower():
            return True

    return False

3. Storage with Attribution

async def save_attributed_result(
    result: AttributedResult,
    db: AsyncSession,
) -> None:
    """
    Save result with all attribution intact.
    Attribution comes from our context, not LLM.
    """

    # Create artifact record
    artifact = Artifact(
        id=result.id,
        user_id=result.user_id,
        tenant_id=result.tenant_id,
        analysis_id=result.analysis_id,
        content=result.content,
        key_concepts=result.key_concepts,
        difficulty_level=result.difficulty_level,
        summary=result.summary,
        created_at=result.created_at,
        model_used=result.model_used,
    )
    db.add(artifact)

    # Create source links
    for doc_id in result.source_document_ids:
        link = ArtifactSourceLink(
            artifact_id=result.id,
            document_id=doc_id,
            tenant_id=result.tenant_id,  # Denormalized for RLS
        )
        db.add(link)

    await db.commit()

    # Audit log
    logger.audit(
        "artifact.created",
        artifact_id=result.id,
        user_id=result.user_id,
        tenant_id=result.tenant_id,
        source_count=len(result.source_document_ids),
    )

OrchestKit Integration

Content Analysis Workflow

# backend/app/workflows/agents/content_analyzer.py

async def create_analysis_artifact(state: AnalysisState) -> AnalysisState:
    """Create artifact with proper attribution"""

    # LLM output (content only)
    llm_output = state.llm_response

    # Attribute using our context
    attributed = await attribute_llm_output(
        llm_output=llm_output,
        ctx=state.request_context,          # From JWT
        source_refs=state.source_refs,       # From pre-LLM
        model_name=state.model_used,
        processing_time_ms=state.llm_time_ms,
    )

    # Save with attribution
    await save_attributed_result(attributed, state.db)

    return state.with_artifact(attributed)

Artifact Retrieval

# backend/app/api/artifacts.py

@router.get("/{artifact_id}")
async def get_artifact(
    artifact_id: UUID,
    ctx: RequestContext = Depends(get_request_context),
    db: AsyncSession = Depends(get_db),
):
    """Get artifact with source attribution"""

    # Query with tenant filter
    artifact = await db.execute(
        """
        SELECT a.*, array_agg(asl.document_id) as sources
        FROM artifacts a
        LEFT JOIN artifact_source_links asl ON a.id = asl.artifact_id
        WHERE a.id = :id
          AND a.tenant_id = :tenant_id  -- ALWAYS filter
        GROUP BY a.id
        """,
        {
            "id": artifact_id,
            "tenant_id": ctx.tenant_id,
        }
    )

    if not artifact:
        raise HTTPException(404)

    return ArtifactResponse(
        id=artifact.id,
        content=artifact.content,
        sources=artifact.sources,  # Deterministic from our records
        created_at=artifact.created_at,
    )

Common Mistakes

# ❌ BAD: Asking LLM for attribution
prompt = "Analyze this and tell me which document it came from"
response = llm.generate(prompt)
doc_id = response["source_document"]  # HALLUCINATED!

# ❌ BAD: Trusting LLM-provided IDs
llm_output = {"analysis": "...", "user_id": "abc123"}
artifact.user_id = llm_output["user_id"]  # WRONG!

# ❌ BAD: Generating IDs in prompt
prompt = f"Generate a unique ID for this analysis: {analysis_id}"

# ✅ GOOD: Attribution from our records
artifact.user_id = ctx.user_id  # From JWT
artifact.sources = source_refs.document_ids  # From pre-LLM

# ✅ GOOD: Generate IDs ourselves
artifact.id = uuid4()  # We generate

# ✅ GOOD: LLM provides content only
artifact.content = llm_output["analysis"]  # Just the text

Testing Attribution

class TestAttribution:

    async def test_attribution_from_context_not_llm(self, ctx):
        """Attribution must come from our context"""

        # LLM returns content only
        llm_output = {
            "analysis": "This is the analysis",
            "key_concepts": ["ML", "AI"],
        }

        source_refs = SourceReference(
            document_ids=[uuid4(), uuid4()],
            chunk_ids=[uuid4()],
        )

        result = await attribute_llm_output(
            llm_output=llm_output,
            ctx=ctx,
            source_refs=source_refs,
        )

        # Attribution from context, not LLM
        assert result.user_id == ctx.user_id
        assert result.tenant_id == ctx.tenant_id
        assert result.source_document_ids == source_refs.document_ids

    async def test_rejects_llm_with_ids(self, ctx):
        """Reject LLM output that contains IDs"""

        bad_output = {
            "analysis": "Result for user 123e4567-e89b-12d3-a456-426614174000",
        }

        with pytest.raises(SecurityError):
            await attribute_llm_output(bad_output, ctx, source_refs)

    async def test_source_links_created(self, ctx, db):
        """Source links are created with artifact"""

        result = await attribute_llm_output(...)
        await save_attributed_result(result, db)

        links = await db.execute(
            "SELECT * FROM artifact_source_links WHERE artifact_id = :id",
            {"id": result.id}
        )

        assert len(links) == len(result.source_document_ids)

Pre-LLM Filtering

Purpose

Before ANY data reaches the LLM, it must be: 1. Scoped to the current tenant/user 2. Filtered for relevance 3. Stripped of identifiers 4. Captured for later attribution

┌────────────────────────────────────────────────────────────┐
│                    PRE-LLM PHASE                           │
├────────────────────────────────────────────────────────────┤
│                                                            │
│  User Query ──► Tenant Filter ──► Content Extract ──► LLM  │
│       │              │                   │                 │
│       │              │                   │                 │
│       ▼              ▼                   ▼                 │
│  ┌─────────┐   ┌───────────┐     ┌─────────────┐          │
│  │ Query   │   │ Documents │     │ Text Only   │          │
│  │ Text    │   │ for THIS  │     │ (no IDs)    │          │
│  │         │   │ tenant    │     │             │          │
│  └─────────┘   └───────────┘     └─────────────┘          │
│                      │                                     │
│                      ▼                                     │
│              ┌─────────────┐                              │
│              │ Save Refs   │                              │
│              │ for Later   │  ◄── For post-LLM attribution│
│              │ Attribution │                              │
│              └─────────────┘                              │
│                                                            │
└────────────────────────────────────────────────────────────┘

Implementation

1. Tenant-Scoped Retrieval

from uuid import UUID
from dataclasses import dataclass

@dataclass
class SourceReference:
    """Tracks what was retrieved for attribution"""
    document_ids: list[UUID]
    chunk_ids: list[UUID]
    similarity_scores: list[float]
    retrieval_timestamp: datetime

async def retrieve_with_isolation(
    query: str,
    ctx: RequestContext,
    limit: int = 10,
) -> tuple[list[str], SourceReference]:
    """
    Retrieve documents scoped to tenant/user.
    Returns: (content_texts, source_references)
    """
    # Embed query
    query_embedding = await embed(query)

    # Search with MANDATORY tenant filter
    results = await db.execute(
        """
        SELECT id, chunk_id, content,
               1 - (embedding <-> :query) as similarity
        FROM document_chunks
        WHERE tenant_id = :tenant_id    -- REQUIRED
          AND user_id = :user_id        -- REQUIRED
          AND embedding <-> :query < 0.5
        ORDER BY embedding <-> :query
        LIMIT :limit
        """,
        {
            "tenant_id": ctx.tenant_id,  # From JWT
            "user_id": ctx.user_id,       # From JWT
            "query": query_embedding,
            "limit": limit,
        }
    )

    # Separate content from references
    content_texts = [r.content for r in results]
    source_refs = SourceReference(
        document_ids=[r.id for r in results],
        chunk_ids=[r.chunk_id for r in results],
        similarity_scores=[r.similarity for r in results],
        retrieval_timestamp=datetime.now(timezone.utc),
    )

    return content_texts, source_refs

2. Content Extraction (Strip IDs)

def extract_content_only(documents: list[Document]) -> list[str]:
    """
    Extract text content, stripping any embedded IDs.
    """
    contents = []
    for doc in documents:
        # Get content
        text = doc.content

        # Remove any embedded IDs (defensive)
        text = strip_identifiers(text)

        contents.append(text)

    return contents

def strip_identifiers(text: str) -> str:
    """Remove any identifiers that might have leaked into content"""
    import re

    # Remove UUIDs
    text = re.sub(
        r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}',
        '[REDACTED]',
        text,
        flags=re.IGNORECASE
    )

    # Remove common ID patterns
    patterns = [
        r'user_id:\s*\S+',
        r'tenant_id:\s*\S+',
        r'doc_id:\s*\S+',
    ]
    for pattern in patterns:
        text = re.sub(pattern, '[REDACTED]', text, flags=re.IGNORECASE)

    return text

3. Full Pre-LLM Pipeline

@dataclass
class PreLLMResult:
    """Complete pre-LLM preparation result"""
    query: str
    context_texts: list[str]
    source_refs: SourceReference
    preparation_time_ms: float

async def prepare_for_llm(
    query: str,
    ctx: RequestContext,
) -> PreLLMResult:
    """
    Complete pre-LLM preparation:
    1. Retrieve with tenant isolation
    2. Extract content only
    3. Save references for attribution
    """
    start = time.monotonic()

    # Step 1: Tenant-scoped retrieval
    raw_results, source_refs = await retrieve_with_isolation(
        query=query,
        ctx=ctx,
    )

    # Step 2: Extract and clean content
    context_texts = [strip_identifiers(text) for text in raw_results]

    # Step 3: Audit for any remaining IDs
    for text in context_texts:
        violations = audit_prompt(text)
        if violations:
            logger.warning(
                "ID found in content, redacting",
                violations=violations,
            )

    elapsed = (time.monotonic() - start) * 1000

    return PreLLMResult(
        query=query,
        context_texts=context_texts,
        source_refs=source_refs,
        preparation_time_ms=elapsed,
    )

OrchestKit Integration

In Content Analysis Workflow

# backend/app/workflows/agents/retriever.py

async def retrieve_context(state: AnalysisState) -> AnalysisState:
    """RAG retrieval with tenant isolation"""

    ctx = state.request_context

    # Pre-LLM preparation
    pre_llm = await prepare_for_llm(
        query=state.analysis_request.query,
        ctx=ctx,
    )

    # Store for later phases
    return state.copy(
        context_texts=pre_llm.context_texts,
        source_refs=pre_llm.source_refs,
        # NO IDs in state that goes to LLM
    )

In Library Search

# backend/app/services/search.py

async def search_libraries(
    query: str,
    ctx: RequestContext,
) -> SearchResult:
    """Search golden dataset with isolation"""

    # Always filter by tenant
    results = await db.execute(
        """
        SELECT id, title, url, summary, content
        FROM golden_dataset
        WHERE tenant_id = :tenant_id
          AND search_vector @@ plainto_tsquery(:query)
        ORDER BY ts_rank(search_vector, plainto_tsquery(:query)) DESC
        LIMIT 20
        """,
        {
            "tenant_id": ctx.tenant_id,
            "query": query,
        }
    )

    # Return content and refs separately
    return SearchResult(
        items=[r.content for r in results],  # Content for LLM
        refs=[r.id for r in results],         # IDs for attribution
    )

Common Mistakes

# ❌ BAD: Query without tenant filter
results = await db.execute("SELECT * FROM documents")

# ❌ BAD: Tenant filter as optional
async def search(tenant_id: UUID | None = None):
    query = "SELECT * FROM documents"
    if tenant_id:  # Can be bypassed!
        query += f" WHERE tenant_id = '{tenant_id}'"

# ❌ BAD: Trusting client-provided tenant
async def search(request: Request):
    tenant_id = request.query_params["tenant_id"]  # Attacker controls!

# ❌ BAD: Including IDs in content
results = [{"id": doc.id, "content": doc.content} for doc in docs]

# ✅ GOOD: Mandatory tenant filter from context
results = await db.execute(
    "SELECT content FROM documents WHERE tenant_id = :tid",
    {"tid": ctx.tenant_id}  # From verified JWT
)

# ✅ GOOD: Content separate from refs
content = [doc.content for doc in docs]  # For LLM
refs = [doc.id for doc in docs]           # For attribution

Testing Pre-LLM Filtering

class TestPreLLMFiltering:

    async def test_retrieval_respects_tenant(
        self,
        tenant_a_ctx,
        tenant_b_ctx,
    ):
        # Create doc for tenant B
        await create_document(
            tenant_id=tenant_b_ctx.tenant_id,
            content="Secret data",
        )

        # Search as tenant A
        result = await prepare_for_llm(
            query="secret",
            ctx=tenant_a_ctx,
        )

        # Must not find tenant B's data
        assert len(result.context_texts) == 0

    async def test_content_has_no_uuids(self, ctx):
        result = await prepare_for_llm(
            query="test query",
            ctx=ctx,
        )

        for text in result.context_texts:
            assert not re.search(UUID_PATTERN, text)

    async def test_source_refs_captured(self, ctx):
        result = await prepare_for_llm(
            query="test query",
            ctx=ctx,
        )

        # Refs saved for attribution
        assert len(result.source_refs.document_ids) > 0
        assert result.source_refs.retrieval_timestamp is not None

Prompt Audit

Purpose

Before any prompt is sent to an LLM, audit it for forbidden content:

┌────────────────────────────────────────────────────────────┐
│                     PROMPT AUDIT                           │
├────────────────────────────────────────────────────────────┤
│                                                            │
│  Prompt Template + Variables ──► Audit ──► Send to LLM     │
│                                    │                       │
│                                    ▼                       │
│                          ┌──────────────┐                  │
│                          │  FORBIDDEN   │                  │
│                          │  PATTERNS    │                  │
│                          ├──────────────┤                  │
│                          │ • user_id    │                  │
│                          │ • tenant_id  │                  │
│                          │ • UUIDs      │                  │
│                          │ • API keys   │                  │
│                          │ • Tokens     │                  │
│                          │ • Secrets    │                  │
│                          └──────────────┘                  │
│                                    │                       │
│                    ┌───────────────┼───────────────┐       │
│                    ▼               ▼               ▼       │
│              ┌──────────┐   ┌──────────┐   ┌──────────┐   │
│              │  CLEAN   │   │ WARNING  │   │  BLOCK   │   │
│              │          │   │          │   │          │   │
│              │ Proceed  │   │ Log +    │   │ Reject   │   │
│              │          │   │ Proceed  │   │          │   │
│              └──────────┘   └──────────┘   └──────────┘   │
│                                                            │
└────────────────────────────────────────────────────────────┘

OrchestKit Forbidden Patterns

Critical (Block Immediately)

Warning (Log and Review)

Implementation

1. Pattern Definitions

import re
from enum import Enum
from dataclasses import dataclass

class AuditSeverity(Enum):
    CLEAN = "clean"
    WARNING = "warning"
    CRITICAL = "critical"

@dataclass
class AuditViolation:
    pattern: str
    severity: AuditSeverity
    match: str
    position: int

# OrchestKit-specific patterns
CRITICAL_PATTERNS = [
    # UUIDs
    (r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', "UUID"),
    # Secrets
    (r'api[_-]?key\s*[:=]\s*["\']?\S+', "API_KEY"),
    (r'password\s*[:=]\s*["\']?\S+', "PASSWORD"),
    (r'secret\s*[:=]\s*["\']?\S+', "SECRET"),
    (r'token\s*[:=]\s*["\']?\S+', "TOKEN"),
    (r'bearer\s+\S+', "BEARER_TOKEN"),
]

WARNING_PATTERNS = [
    # OrchestKit identifiers
    (r'\buser[_-]?id\b', "USER_ID_FIELD"),
    (r'\btenant[_-]?id\b', "TENANT_ID_FIELD"),
    (r'\banalysis[_-]?id\b', "ANALYSIS_ID_FIELD"),
    (r'\bdocument[_-]?id\b', "DOCUMENT_ID_FIELD"),
    (r'\bartifact[_-]?id\b', "ARTIFACT_ID_FIELD"),
    (r'\bchunk[_-]?id\b', "CHUNK_ID_FIELD"),
    (r'\bsession[_-]?id\b', "SESSION_ID_FIELD"),
    (r'\btrace[_-]?id\b', "TRACE_ID_FIELD"),
    (r'\bworkflow[_-]?run[_-]?id\b', "WORKFLOW_ID_FIELD"),
]

2. Audit Function

def audit_prompt(prompt: str) -> list[AuditViolation]:
    """
    Audit prompt for forbidden patterns.
    Returns list of violations.
    """
    violations = []

    # Check critical patterns
    for pattern, name in CRITICAL_PATTERNS:
        for match in re.finditer(pattern, prompt, re.IGNORECASE):
            violations.append(AuditViolation(
                pattern=name,
                severity=AuditSeverity.CRITICAL,
                match=match.group()[:50],  # Truncate for logging
                position=match.start(),
            ))

    # Check warning patterns
    for pattern, name in WARNING_PATTERNS:
        for match in re.finditer(pattern, prompt, re.IGNORECASE):
            violations.append(AuditViolation(
                pattern=name,
                severity=AuditSeverity.WARNING,
                match=match.group(),
                position=match.start(),
            ))

    return violations

def has_critical_violations(violations: list[AuditViolation]) -> bool:
    """Check if any violations are critical"""
    return any(v.severity == AuditSeverity.CRITICAL for v in violations)

3. Audit Decorator

from functools import wraps
import structlog

logger = structlog.get_logger()

def audit_before_llm(func):
    """
    Decorator that audits prompts before LLM call.
    Blocks on critical violations, logs warnings.
    """
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # Extract prompt from args/kwargs
        prompt = kwargs.get("prompt") or args[0]

        # Audit
        violations = audit_prompt(prompt)

        # Log warnings
        for v in violations:
            if v.severity == AuditSeverity.WARNING:
                logger.warning(
                    "prompt_audit_warning",
                    pattern=v.pattern,
                    position=v.position,
                )

        # Block on critical
        if has_critical_violations(violations):
            critical = [v for v in violations
                       if v.severity == AuditSeverity.CRITICAL]
            raise PromptSecurityError(
                f"Prompt contains forbidden content: {[v.pattern for v in critical]}"
            )

        # Proceed
        return await func(*args, **kwargs)

    return wrapper

# Usage
@audit_before_llm
async def call_llm(prompt: str) -> str:
    return await llm.generate(prompt)

4. Safe Prompt Builder

class SafePromptBuilder:
    """
    Builds prompts with built-in audit.
    Prevents accidental ID inclusion.
    """

    def __init__(self):
        self._parts: list[str] = []
        self._context_ids: dict[str, Any] = {}  # Stored but never in prompt

    def add_instruction(self, text: str) -> "SafePromptBuilder":
        """Add instruction text (audited)"""
        violations = audit_prompt(text)
        if has_critical_violations(violations):
            raise PromptSecurityError("Instruction contains forbidden content")
        self._parts.append(text)
        return self

    def add_content(self, content: str) -> "SafePromptBuilder":
        """Add user content (sanitized)"""
        # Strip any IDs from content
        clean_content = self._sanitize(content)
        self._parts.append(clean_content)
        return self

    def add_context_texts(self, texts: list[str]) -> "SafePromptBuilder":
        """Add context texts (sanitized)"""
        for text in texts:
            clean = self._sanitize(text)
            self._parts.append(f"- {clean}")
        return self

    def store_context_id(self, key: str, value: Any) -> "SafePromptBuilder":
        """Store ID for post-LLM attribution (never in prompt)"""
        self._context_ids[key] = value
        return self

    def _sanitize(self, text: str) -> str:
        """Remove any IDs from text"""
        # Remove UUIDs
        text = re.sub(
            r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}',
            '[ID]',
            text,
            flags=re.IGNORECASE
        )
        return text

    def build(self) -> tuple[str, dict]:
        """
        Build prompt and return with stored context.
        Returns: (prompt, context_ids)
        """
        prompt = "\n\n".join(self._parts)

        # Final audit
        violations = audit_prompt(prompt)
        if violations:
            logger.warning(
                "prompt_audit_final",
                violation_count=len(violations),
            )

        if has_critical_violations(violations):
            raise PromptSecurityError("Built prompt contains forbidden content")

        return prompt, self._context_ids

# Usage
builder = SafePromptBuilder()
prompt, context = (
    builder
    .add_instruction("Analyze the following content:")
    .add_content(user_query)
    .add_context_texts(retrieved_docs)
    .store_context_id("user_id", ctx.user_id)  # Stored, not in prompt
    .store_context_id("sources", source_refs)   # Stored, not in prompt
    .build()
)

OrchestKit Integration

Workflow Integration

# backend/app/workflows/agents/prompts/content_analysis.py

from llm_safety import SafePromptBuilder

def build_analysis_prompt(
    query: str,
    context_texts: list[str],
    ctx: RequestContext,
) -> tuple[str, dict]:
    """
    Build content analysis prompt safely.
    Context IDs stored separately for attribution.
    """
    return (
        SafePromptBuilder()
        .add_instruction("""
        You are an expert content analyzer. Analyze the following
        content and provide insights about:
        1. Key concepts
        2. Difficulty level
        3. Prerequisites
        4. Summary
        """)
        .add_instruction(f"User query: {query}")
        .add_instruction("Relevant context:")
        .add_context_texts(context_texts)
        .store_context_id("user_id", ctx.user_id)
        .store_context_id("tenant_id", ctx.tenant_id)
        .store_context_id("trace_id", ctx.trace_id)
        .build()
    )

CI/CD Integration

#!/bin/bash
# scripts/audit_prompts.sh

echo "Auditing prompt templates..."

# Check for IDs in prompt files
grep -rn \
    "user_id\|tenant_id\|analysis_id\|document_id\|[0-9a-f]\{8\}-[0-9a-f]\{4\}" \
    backend/app/**/prompts/ \
    --include="*.py" \
    --include="*.txt" \
    --include="*.jinja2"

if [ $? -eq 0 ]; then
    echo "❌ Found potential ID leaks in prompts!"
    exit 1
fi

echo "✅ Prompt audit passed"

Testing

class TestPromptAudit:

    def test_detects_uuid(self):
        prompt = "Analyze doc 123e4567-e89b-12d3-a456-426614174000"
        violations = audit_prompt(prompt)

        assert len(violations) == 1
        assert violations[0].severity == AuditSeverity.CRITICAL
        assert violations[0].pattern == "UUID"

    def test_detects_api_key(self):
        prompt = "Use api_key: sk-1234567890abcdef"
        violations = audit_prompt(prompt)

        assert any(v.pattern == "API_KEY" for v in violations)

    def test_warns_on_user_id_field(self):
        prompt = "For user_id please provide analysis"
        violations = audit_prompt(prompt)

        assert len(violations) == 1
        assert violations[0].severity == AuditSeverity.WARNING

    def test_safe_builder_blocks_id(self):
        with pytest.raises(PromptSecurityError):
            (
                SafePromptBuilder()
                .add_instruction("Analyze for user 123e4567-e89b-12d3-a456-426614174000")
                .build()
            )

    def test_safe_builder_sanitizes_content(self):
        prompt, _ = (
            SafePromptBuilder()
            .add_content("Doc ID: 123e4567-e89b-12d3-a456-426614174000")
            .build()
        )

        assert "123e4567" not in prompt
        assert "[ID]" in prompt

    def test_context_ids_not_in_prompt(self):
        from uuid import uuid4

        user_id = uuid4()
        prompt, context = (
            SafePromptBuilder()
            .add_instruction("Analyze this")
            .store_context_id("user_id", user_id)
            .build()
        )

        assert str(user_id) not in prompt
        assert context["user_id"] == user_id