direkt-vermittlung-de/CLAUDE.md

CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

Project Overview

DirektVermittlungDe (DVD) is a document-centric communication platform between citizens and German authorities. It eliminates the "phone hunt" by:

1. Citizens upload a document or provide an Aktenzeichen (reference number)
2. The system auto-routes to the responsible unit
3. Opens an interaction thread for direct clarification

Current Development Status

⚠️ IMPORTANT: This repository is currently in transition from prototype phase to main production codebase.

• Current State: Three prototype implementations (chatgpt5, geminiNbt3pro, grok4.1) in /prototype-* directories
• Target State: Unified production codebase in /src with TDD-driven development
• Active Workplan: See docs/WORKPLAN_MainCodebase_Integration.md for detailed migration plan

When working on this codebase:

• Check the workplan to understand which phase we're in
• Follow TDD practices: write tests first, then implementation
• Reference ADRs for architectural decisions (see below)
• New code goes in /src, not prototypes

Repository Structure (Target State)

/
├── docs/
│   ├── architecture/
│   │   ├── adr/                    # Individual ADR files (0001-*.md)
│   │   │   ├── 0001-split-payload-model.md
│   │   │   ├── 0002-stateless-authentication.md
│   │   │   └── ...
│   │   ├── architecture-overview.md
│   │   └── design-patterns.md
│   ├── api/
│   │   ├── openapi.yaml
│   │   └── api-scenarios.md
│   ├── development/
│   │   ├── testing-strategy.md
│   │   ├── agentic-coding-guide.md
│   │   └── setup-guide.md
│   └── WORKPLAN_MainCodebase_Integration.md
├── src/
│   ├── domain/          # Pure business logic (TDD tested)
│   ├── adapters/        # External integrations (mocked in tests)
│   ├── service/         # Application services (TDD tested)
│   ├── api/             # FastAPI routes (integration tested)
│   └── workers/         # Background jobs
├── tests/
│   ├── unit/            # TDD unit tests
│   ├── integration/     # Integration tests
│   └── fixtures/        # Test data and mocks
├── scripts/             # Utility scripts
├── prototype-*/         # ARCHIVED prototypes (reference only)
├── pyproject.toml       # Dependencies & tool config
└── pytest.ini           # Test configuration

Development Commands

Setup and Run (Main Codebase)

# Install dependencies (from repo root)
pip install -e ".[dev]"

# Initialize database metadata
python -m scripts.init_db

# Run development server
uvicorn src.main:app --reload

# View API docs
# http://localhost:8000/docs

Testing (TDD Workflow)

# Run all tests
pytest

# Run with coverage report
pytest --cov=src --cov-report=html

# Run only unit tests
pytest tests/unit/

# Run only integration tests
pytest tests/integration/

# Run specific test file
pytest tests/unit/domain/test_document_metadata.py -v

# Watch mode (requires pytest-watch)
ptw

Code Quality

# Lint and format with ruff
ruff check src/ tests/
ruff format src/ tests/

# Type checking with mypy
mypy src/

# Run pre-commit hooks manually
pre-commit run --all-files

Key Architectural Patterns

1. Split-Payload Model (ADR-001)

Critical: Documents use a split design to enable both encryption and routing:

• metadata (plaintext JSON): authorityId, referenceNumber, docType, issuedAt
• encryptedPayload (encrypted blob): actual PDF/scan content

The backend can route based on metadata without decrypting the payload. This is the core architectural constraint.

When working with documents:

• NEVER attempt to decrypt encryptedPayload in the main service
• The backend treats encrypted content as opaque
• Routing logic operates only on metadata fields

2. Hybrid Concurrency Pattern (ADR-007)

Python's GIL requires careful concurrency handling:

• I/O operations (DB, network): Use native async/await
• CPU operations (crypto, PDF): Offload to ProcessPoolExecutor

Example pattern:

cpu_pool = ProcessPoolExecutor(max_workers=4)

async def handler():
    # I/O: async/await
    data = await db.fetch()

    # CPU: executor
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(cpu_pool, heavy_function, data)

3. Cursor-Based Pagination (ADR-003)

For message threads, use cursor-based pagination (timestamp) instead of offset-based:

• Ensures consistent performance regardless of thread length
• Prevents duplicate/missing messages during real-time updates
• Target: <300ms response time (NFR-1)

4. Async Export Pattern (ADR-004)

Data exports use async request-reply:

1. POST /exports → 202 Accepted + jobId
2. Background worker processes via message queue
3. Client polls status endpoint or receives webhook

5. Stateless Authentication (ADR-002)

OAuth2/JWT with scopes:

• citizen:write: Document submission, thread creation
• official:read: View assigned cases
• official:write: Respond to inquiries

JWTs enable horizontal scaling without session affinity.

6. GDPR-Compliant Retention (ADR-006)

Documents have retention_date field:

• Default: closedAt + gracePeriod
• Personal archive: null or extended date
• Automated cleanup via TTL engine

API Structure

The API follows document-centric REST hierarchy:

• /documents - Root envelope (FR-1: Document intake)
• /documents/{id}/threads - Sub-resource for communication (FR-3)
• /threads/{threadId}/messages - Message exchange
• /exports - Async data export for authority systems (FR-7)

Critical Domain Models

Located in src/domain/models.py (target) or prototype-chatgpt5/src/app/domain/models.py (reference):

Enums:

• ThreadType: TEXT_CHAT, CALLBACK_REQUEST, APPOINTMENT
• SenderRole: CITIZEN, OFFICIAL, SYSTEM
• ExportJobStatus: QUEUED, RUNNING, COMPLETED, FAILED

Core Models:

• DocumentMetadata: authorityId, referenceNumber, docType, issuedAt
  • Validation: authorityId (1-50 chars), referenceNumber required
• DocumentCreateRequest: metadata, encryptedPayload (base64)
• DocumentCreatedResponse: id, status, assignedUnit
• ThreadCreateRequest: type, initialMessage, preferredTimeSlot
• MessageDto: id, senderRole, content (encrypted), timestamp

Performance Targets (NFR)

• Core operations: <300ms response time
• Document upload routing: <500ms
• Concurrent sessions: 10k+ per region
• Availability: ≥99.5%

Security Constraints

• E2E encryption required for sensitive data (NFR-5)
• Backend cannot decrypt document payloads
• Least-privilege access via OAuth scopes (NFR-6)
• No PII in plaintext application logs

TDD Development Workflow

CRITICAL: This project follows a strict Test-Driven Development approach.

The Red-Green-Refactor Cycle

1. RED: Write a failing test that defines desired behavior
2. GREEN: Write minimal code to make the test pass
3. REFACTOR: Improve code while keeping tests green

Example TDD Workflow

# Step 1: Write failing test (tests/unit/domain/test_document_metadata.py)
def test_document_metadata_validates_authority_id():
    with pytest.raises(ValidationError):
        DocumentMetadata(
            authorityId="",  # Empty should fail
            referenceNumber="123/456",
            docType="NOTICE",
            issuedAt=datetime.now()
        )

# Step 2: Run test (it fails)
# $ pytest tests/unit/domain/test_document_metadata.py

# Step 3: Implement validation (src/domain/models.py)
class DocumentMetadata(BaseModel):
    authorityId: str = Field(..., min_length=1, max_length=50)
    # ... rest of fields

# Step 4: Run test again (it passes)

# Step 5: Refactor if needed, ensure tests still pass

Agentic TDD Pattern

When asking Claude to implement features, use this pattern:

"Write a test for [feature] that verifies [specific behavior].
Use pytest and follow the pattern in tests/unit/[module]/.
Reference ADR-[number] for architectural constraints."

Example:

"Write a test for document routing that verifies documents with docType='NOTICE'
are routed to the NoticeTeam. Mock the routing adapter using the protocol defined
in src/adapters/protocols.py. Follow ADR-0001 split-payload constraints."

See docs/development/agentic-coding-guide.md for comprehensive examples.

Documentation References

Architecture & Decisions

• Workplan: docs/WORKPLAN_MainCodebase_Integration.md - Current migration plan
• ADRs: docs/architecture/adr/ - Individual decision records
  • 0001-split-payload-model.md - Core constraint: metadata vs encrypted payload
  • 0002-stateless-authentication.md - OAuth2/JWT architecture
  • 0003-cursor-based-pagination.md - Performance-optimized pagination
  • 0004-async-export-workflow.md - Background job pattern
  • 0005-rest-resource-structure.md - API design hierarchy
  • 0006-gdpr-retention-model.md - Data lifecycle management
  • 0007-hybrid-concurrency-pattern.md - Python GIL mitigation
  • 0008-agentic-tdd-workflow.md - LLM-driven development process
• Overview: docs/architecture/architecture-overview.md - System design
• Patterns: docs/architecture/design-patterns.md - Common patterns

API Documentation

• OpenAPI: docs/api/openapi.yaml - Full API specification
• Scenarios: docs/api/api-scenarios.md - Real-world usage examples

Development Guides

• Testing: docs/development/testing-strategy.md - TDD practices
• Agentic Coding: docs/development/agentic-coding-guide.md - AI-assisted development
• Setup: docs/development/setup-guide.md - Environment setup

Legacy (Prototypes)

• prototype-chatgpt5/README.md - Reference implementation (ARCHIVED)
• docs/decisions.md - Original ADRs (DEPRECATED, see /docs/architecture/adr/)

Database Schema

Key tables (PostgreSQL):

• documents: id, reference_number, authority_id, status, storage_path, retention_date
• threads: id, document_id, type, assigned_official_id, last_activity_at
• messages: id, thread_id, sender_role, content_blob, created_at

Indexes:

• idx_docs_authority on documents(authority_id, status)
• idx_msgs_thread_time on messages(thread_id, created_at DESC)

Technology Stack

• Language: Python 3.11+
• Framework: FastAPI + Uvicorn
• ORM: SQLAlchemy (async)
• Database: PostgreSQL 15+
• Blob Storage: S3-compatible (MinIO/AWS S3)
• Task Queue: ARQ (Redis-based, async) or Celery
• Auth: OAuth2/JWT (stateless)
• Testing: pytest + pytest-asyncio + pytest-cov
• Linting: ruff (fast, replaces flake8/isort/pyupgrade)
• Type Checking: mypy (strict mode)
• CI/CD: GitHub Actions

Workplan & Phase Tracking

Active Workplan: docs/WORKPLAN_MainCodebase_Integration.md

Current Phase Status

To check which phase we're in:

1. Open docs/WORKPLAN_MainCodebase_Integration.md
2. Look for checked [x] items in each phase section
3. Focus development on uncompleted [ ] items in the current phase

How to Contribute

Before implementing any feature:

1. Check if it's in the current phase of the workplan
2. Read the relevant ADR(s) in docs/architecture/adr/
3. Write tests first (TDD approach)
4. Implement to make tests pass
5. Ensure all quality gates pass (pytest, mypy, ruff)

When adding new architectural decisions:

1. Create a new ADR file in docs/architecture/adr/
2. Use the template in 0000-template.md
3. Number sequentially (next available number)
4. Update the ADR index in docs/architecture/adr/README.md

Common Gotchas

1. Stream large files: Use aiobotocore to stream uploads to S3, don't load entire payloads into memory
2. ProcessPoolExecutor: CPU-heavy operations MUST be offloaded to avoid blocking the event loop
3. Metadata separation: The routing engine needs plaintext metadata - design API contracts accordingly
4. Retention dates: Always set retention_date on document creation for GDPR compliance
5. Cursor pagination: Use timestamp-based cursors for message history, not offset-based