code-gen

name	code-gen
description	Code generation quality principles — TDD, typing, error handling, logging, API integration, LLM integration.

Code Generation Skill

Reference skill for generator teammates. Read this before writing any code.

---

Six Quality Principles

1. Small Modules — One File, One Responsibility

• Each file must have a single, clearly named responsibility.
• Warning threshold: 200 lines — add a comment noting the file is growing large.
• Block threshold: 300 lines — do not submit. Split before opening a PR.
• If you hit 300 lines, decompose into sub-modules and re-export from an index file.

2. Static Typing — Annotate Everything

• Every function parameter, return value, and variable must have an explicit type.
• TypeScript: Zero any. Use unknown + type guard if the shape is truly unknown.
• Python: Full type hints on all functions. Use TypeVar, Generic, Protocol where appropriate.
• Type aliases for domain concepts (UserId = str, type OrderId = string).

3. Functions Under 50 Lines

• If a function body exceeds 50 lines, decompose it into named sub-functions.
• Each sub-function should be testable in isolation.
• Use descriptive names that read as a sentence: validateOrderItems, buildPaymentPayload.
• Avoid deeply nested control flow — extract branches into named helpers.

4. Single Owner for State Mutations

• Every state-creating operation (DB insert, file write, queue publish) must have exactly ONE call site.
• If a route handler creates a record, the service it calls must receive the ID — not create a second record.
• Anti-pattern: Route creates Task, then calls service.start(query) which also creates a Task → duplicate records.
• Correct pattern: Route creates Task with ID, then calls service.start(task_id) which operates on the existing record.
• When a background task or async flow needs a resource, pass the ID, don't re-create it.
• Test this explicitly: after calling the endpoint, assert the exact count of records created (e.g., assert db.query(Task).count() == 1).

5. Explicit Error Handling

• Define typed error classes per domain (e.g., class OrderNotFoundError extends AppError).
• Never use bare except Exception or catch (e: any).
• All error paths must be covered by tests.
• Propagate errors up with context; do not swallow silently.
• In TypeScript: use Result<T, E> or typed throws with JSDoc @throws.

6. No Dead Code

• Every line of code must trace to a user story or a technical requirement.
• Do not leave commented-out code in PRs.
• Remove unused imports, variables, and parameters immediately.
• If code is speculative ("might need later"), do not include it.

7. Self-Documenting — Names Over Comments

• Variable and function names should make comments unnecessary.
• Types act as documentation — a well-typed function signature is its own doc.
• Use comments only for non-obvious decisions (algorithm choice, regulatory constraints).
• Avoid // TODO in submitted code — file a story instead.

---

Code Patterns

Test Structure: Arrange → Act → Assert

// Arrange
const order = buildOrder({ status: "pending" });
// Act
const result = processOrder(order);
// Assert
expect(result.status).toBe("confirmed");

Typed Error Classes

class DomainError extends Error {
  constructor(message: string, public readonly code: string) {
    super(message);
    this.name = this.constructor.name;
  }
}
class OrderNotFoundError extends DomainError {
  constructor(orderId: OrderId) {
    super(`Order ${orderId} not found`, "ORDER_NOT_FOUND");
  }
}

Naming Conventions

• Files: kebab-case.ts for TypeScript, snake_case.py for Python.
• Functions/methods: camelCase (TS), snake_case (Python).
• Types/classes: PascalCase in both languages.
• Constants: UPPER_SNAKE_CASE.
• Booleans: prefix with is, has, can, should.

Database Session Lifecycle (Python/FastAPI)

• Use async_sessionmaker with FastAPI Depends() for dependency injection — never create sessions manually per request.
• Engine and session factory created once in lifespan, stored on app.state, disposed on shutdown.
• Sessions auto-close via async with — never leave sessions open.
• Tests override the session dependency to use a test DB, not the dev DB.

# CORRECT — lifespan manages engine lifecycle
@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine(settings.DATABASE_URL)
    app.state.session_factory = async_sessionmaker(engine)
    yield
    await engine.dispose()

# CORRECT — session via dependency injection
async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]:
    async with request.app.state.session_factory() as session:
        yield session

DbSession = Annotated[AsyncSession, Depends(get_db)]

Background Tasks Must Be Tracked

• Every background task must have a corresponding DB record with a status field (pending, running, completed, failed).
• The route creates the record (status=pending), then starts the background task with the record's ID.
• The background task updates the record on completion or failure — never fire-and-forget.
• Wrap the entire background task body in try/except: on failure, update the record to failed with error details and log the exception.

# CORRECT — tracked background task
@router.post("/tasks")
async def create_task(request: TaskRequest, db: DbSession, background_tasks: BackgroundTasks):
    task = Task(query=request.query, status="pending")
    db.add(task)
    await db.commit()
    background_tasks.add_task(run_task, task.id)  # pass ID, not the object
    return {"task_id": task.id, "status": "pending"}

async def run_task(task_id: UUID) -> None:
    async with get_session() as db:
        task = await db.get(Task, task_id)
        try:
            task.status = "running"
            await db.commit()
            result = await do_work(task.query)
            task.status = "completed"
            task.result = result
        except Exception as e:
            logger.exception("Task failed", extra={"task_id": str(task_id)})
            task.status = "failed"
            task.error = str(e)
        finally:
            await db.commit()

CORS Must Be Environment-Configured

• Never use allow_origins=["*"] with allow_credentials=True — this is a security vulnerability.
• Read allowed origins from an environment variable, default to localhost only.

# CORRECT — env-configured CORS
allowed_origins = settings.ALLOWED_ORIGINS.split(",")  # from .env
app.add_middleware(
    CORSMiddleware,
    allow_origins=allowed_origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

Health Check Must Verify Dependencies

• Health endpoint must test critical dependencies (DB, cache, external APIs), not just return {"status": "ok"}.
• Return 503 if any dependency is unreachable.

@router.get("/health")
async def health(db: DbSession) -> dict:
    try:
        await db.execute(text("SELECT 1"))
        return {"status": "ok", "db": "connected"}
    except Exception:
        raise HTTPException(status_code=503, detail="Database unreachable")

Request ID Middleware

• Add middleware that generates a unique request_id per request and includes it in all log entries and error responses.
• This enables tracing a single user request across all service logs.

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid4())
    request.state.request_id = request_id
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

---

Testing Rules — TDD Mandatory

"Coverage isn't about bug prevention — it's about guaranteeing the agent has double-checked the behavior of every line of code it wrote." — Steve Krenzel

1. Tests FIRST, then code (TDD):
   • Write a failing test that defines expected behavior
   • Run it — verify it fails for the right reason
   • Write the minimum code to make it pass
   • Run it — verify it passes
   • Refactor if needed, re-run tests
   • Commit
2. 100% meaningful coverage — every branch, every error path. At 100%, any uncovered line is an immediate signal of missing verification. The ratchet gate BLOCKS below 80%.
3. Only mock external boundaries: databases, third-party APIs, file I/O, clocks.
4. Never mock business logic — if you mock a service to test another service, you are hiding bugs.
5. Isolate tests from .env files: When testing settings/config that uses pydantic-settings or dotenv, pass _env_file=None (pydantic) or mock dotenv.load_dotenv to prevent the developer's .env from leaking into tests. Tests must be self-contained — they must pass regardless of what's in the local .env.
6. Use async-compatible connection strings: When using async frameworks (SQLAlchemy async, asyncpg), defaults must use the async driver scheme (e.g., postgresql+asyncpg:// not postgresql://). The sync scheme will fail at runtime with a cryptic driver error.
7. Realistic test data — use domain-representative values (real-looking emails, valid UUIDs, plausible amounts). Never "foo", 123, or "test".
8. Test names describe behavior: "returns 404 when order does not exist", not "test order".
9. Integration tests for multi-step flows: When a route triggers a background task or async flow (e.g., POST creates a record then starts processing), write a test that calls the endpoint and asserts the FINAL state — not just that each unit works alone. Assert exact record counts: assert db.query(Task).count() == 1 after one API call.

---

LLM Integration — Structured Output Mandatory

When generated code calls any LLM (Claude, GPT, or other), follow these rules:

1. Always Use Structured Output

Use tool_use / function_calling / response_format: { type: "json_schema", json_schema: ... } for every LLM call. Never parse free-text responses with regex or string splitting.

2. Define a Response Schema

Every LLM call must have a typed model for the expected response:

from pydantic import BaseModel
from typing import Literal

class ClassificationResult(BaseModel):
    category: str
    confidence: Literal["high", "medium", "low"]
    reasoning: str

interface ClassificationResult {
  category: string;
  confidence: "high" | "medium" | "low";
  reasoning: string;
}

3. Validate Before Using

Parse the LLM response through the schema. If validation fails:

1. Retry once with an explicit correction prompt: "Your response did not match the required schema. Required: {schema}. Please respond again."
2. If second attempt fails, raise a typed error — do not fall back to a default value.

4. No Silent Fallbacks

Never write:

# WRONG — hides bugs that compound
try:
    result = await call_llm(prompt)
    parsed = ResponseModel.model_validate_json(result)
except Exception:
    parsed = ResponseModel(category="unknown", confidence="low", reasoning="")

Instead:

# CORRECT — caller decides how to handle failure
class LLMResponseError(Exception):
    def __init__(self, raw_response: str, validation_error: str):
        self.raw_response = raw_response
        self.validation_error = validation_error
        super().__init__(f"LLM response validation failed: {validation_error}")

try:
    result = await call_llm(prompt)
    parsed = ResponseModel.model_validate_json(result)
except ValidationError as e:
    raise LLMResponseError(raw_response=result, validation_error=str(e))

5. Log Raw Responses

Always log the raw LLM response at DEBUG level before parsing:

logger.debug(
    "LLM response received",
    extra={
        "provider": self._provider_name,
        "model": self._model,
        "prompt_tokens": response.usage.input_tokens,
        "completion_tokens": response.usage.output_tokens,
        "raw_content": response.content[:1000],
        "latency_ms": round(elapsed_ms, 2),
    },
)

---

External API Integration

When generated code calls any external API (third-party services, partner APIs, cloud services), follow these rules. See .claude/skills/code-gen/references/api-integration-patterns.md for full templates.

Service Wrapper Pattern (Mandatory)

Every external API gets a dedicated wrapper class. This is the ONLY file that imports the SDK or makes HTTP calls to that service.

Business Logic (process_service.py)
    ↓ calls typed methods
API Wrapper (external_client.py)    ← only file that imports SDK / makes HTTP calls
    ↓ calls
External API

Rules:

• One wrapper class per external API
• Wrapper exposes project-internal typed models, not SDK types
• Business logic never sees SDK response objects — only your domain types
• The wrapper is the mock boundary in tests

Error Taxonomy (Mandatory)

Every wrapper classifies errors into typed categories:

class ApiTransientError(Exception):
    """Retryable: 429, 502, 503, timeout, connection reset."""
    pass

class ApiPermanentError(Exception):
    """Not retryable: 400, 401, 403, 404, schema mismatch."""
    pass

class ApiRateLimitError(ApiTransientError):
    """Rate limited with backoff hint."""
    def __init__(self, message: str, retry_after: float | None = None):
        super().__init__(message)
        self.retry_after = retry_after

• Business logic catches ApiTransientError to retry/degrade, ApiPermanentError to fail fast
• No bare except Exception in any API-calling code
• All exceptions carry HTTP status code and response body for debugging

Retry and Rate Limiting

• Retry config lives in config.yml under external_apis.{service_name}.retry, not hardcoded
• Wrapper applies exponential backoff internally — business logic is unaware of retries
• Respect Retry-After headers when present
• Log every retry attempt at WARNING level

Async Bridging

When an SDK is synchronous but the backend is async:

• Use asyncio.to_thread() only inside the wrapper class
• Never bridge in business logic
• Prefer async SDKs or HTTP clients when available

Secrets

• API keys in .env only, loaded via config layer
• Wrapper reads from injected config, never from os.environ directly
• .env.example committed with placeholder values

---

Production Standards

These standards apply to ALL generated code, not just API wrappers or LLM calls.

Structured Logging

All generated services must use structured logging with extra dicts:

import logging

logger = logging.getLogger(__name__)

# CORRECT — structured fields for JSON log formatters
logger.info("Document processed", extra={
    "document_id": doc.id,
    "processing_time_ms": round(elapsed_ms, 2),
    "output_size_bytes": len(result),
})

# WRONG — data interpolated into message string
logger.info(f"Document {doc.id} processed in {elapsed_ms}ms")

// CORRECT — structured logger
logger.info("Document processed", {
  documentId: doc.id,
  processingTimeMs: Math.round(elapsedMs),
  outputSizeBytes: result.length,
});

// WRONG — template literal message
logger.info(`Document ${doc.id} processed in ${elapsedMs}ms`);

Rules:

• Use logging.getLogger(__name__) (Python) or scoped logger (TypeScript) at module level
• INFO for business events (request received, document processed, job completed)
• WARNING for recoverable issues (retry triggered, fallback used, slow response)
• ERROR for failures requiring attention (unhandled exception, data corruption, external service down)
• DEBUG for troubleshooting data (raw payloads, intermediate state, timing breakdowns)
• Never log secrets, tokens, passwords, or PII
• Log at service boundaries: incoming requests, outgoing calls, business decisions

Exception Handling

# CORRECT — typed exception with context
class DocumentProcessingError(Exception):
    def __init__(self, document_id: str, stage: str, cause: Exception):
        self.document_id = document_id
        self.stage = stage
        self.cause = cause
        super().__init__(f"Failed at {stage} for document {document_id}: {cause}")

# WRONG — bare except swallowing the error
try:
    result = process(doc)
except Exception:
    result = default_value

Rules:

• Define typed exception classes per domain (not per function)
• Every exception carries enough context to debug without the stack trace
• Never catch Exception or BaseException unless re-raising or logging at a top-level boundary
• No silent fallbacks — if an operation fails, the caller must know
• API route handlers catch domain exceptions and map to HTTP error responses

Structured Error Responses

All API error responses follow a consistent envelope:

{
  "error": {
    "code": "DOCUMENT_NOT_FOUND",
    "message": "Document with ID abc123 does not exist",
    "details": {}
  }
}

Rules:

• code is a machine-readable UPPER_SNAKE_CASE string enum
• message is human-readable
• details is optional structured context
• HTTP status mapping: 400 validation, 404 not found, 409 conflict, 422 processing error, 500 internal

Request/Response Validation

• All API inputs validated via Pydantic models (Python) or Zod schemas (TypeScript)
• Validation errors return 400 with field-level messages
• All API outputs serialized through response models — never return raw dicts or ORM objects

Configuration

• All configurable values in config.yml or environment variables
• No magic numbers or hardcoded strings in business logic
• Config loaded once at startup, injected into services via constructor
• Defaults provided for all non-secret config values

---

Parallel Execution

• File ownership: consult component-map.md before touching any file.
• Plan approval required before starting parallel work.
• Shared interfaces: message teammates before changing a type or API contract that crosses boundaries.
• Task sizing: aim for 5–6 discrete tasks per teammate per sprint cycle.
• Conflicts: if two teammates need the same file, one blocks; do not merge partial changes.

---

Gotchas (Things That Cause Review Failures)

• Importing upward across layers (UI importing from repository layer)
• Functions exceeding 50 lines without decomposition
• Untyped values — any, missing return types, unannotated parameters
• Broad exception catches without re-raise or typed handling
• Mocking business logic in unit tests
• Generic test data ("test", 0, null as stand-ins for real domain values)
• Commented-out code in the submitted diff
• Missing error-path test coverage
• Teammates editing the same file in the same sprint without coordination
• Free-text LLM parsing — Never use regex to parse LLM output. Use structured output (tool_use / JSON mode).
• Silent fallback on LLM error — except Exception: return default hides compounding bugs. Raise typed errors.
• Missing raw response logging — Always log raw LLM response at DEBUG before parsing. This is the debugging ground truth.
• Direct SDK imports outside wrapper — All SDK imports must be inside the wrapper class file. Business logic imports your wrapper, not the SDK.
• Bare except on API calls — Catch ApiTransientError and ApiPermanentError specifically. Never except Exception.
• Hardcoded retry config — Retry attempts, backoff, and timeout belong in config.yml, not in code.
• Missing structured logging in API wrapper — Every request/response/error must be logged with structured fields (service, operation, attempt, latency_ms).
• f-string log messages — Use extra dict for structured fields, not string interpolation. Structured logs are searchable; f-strings are not.
• Missing logging at service boundaries — Every incoming request and outgoing call must be logged with timing and status.
• Raw dict API responses — Always serialize through a response model. Raw dicts bypass validation and leak internal structure.
• Magic numbers — All thresholds, limits, timeouts, and configuration belong in config.yml.
• .env leaking into tests — Tests that validate "missing config raises error" will pass in CI but fail locally if .env has the value. Always pass _env_file=None in pydantic-settings tests.
• Sync DB driver in async app — postgresql:// uses psycopg2 (sync). Async SQLAlchemy needs postgresql+asyncpg://. Always match the driver scheme to the engine type.
• Duplicate record creation — Route creates a record, then calls a service that creates the same record again. Pass the ID, don't re-create. Test with assert count == 1 after one API call.
• Manual session creation — Never create DB sessions manually per request. Use Depends(get_db) with async_sessionmaker. Manual sessions leak connections.
• Fire-and-forget background tasks — Every background task must update a DB record on completion or failure. No background_tasks.add_task(fn) without status tracking.
• CORS allow_origins=["*"] — Never use wildcard origins with allow_credentials=True. Read origins from env var, default to localhost.
• Health check returns OK without checking DB — Health endpoint must SELECT 1 against the database. A healthy HTTP server with a dead DB is not healthy.
• Engine not disposed on shutdown — Always await engine.dispose() in the lifespan's teardown. Leaked connections exhaust the pool.
• No request ID tracing — Add middleware that generates a UUID per request, injects into logs and response headers. Without it, errors can't be traced to requests.
• Deprecated startup/shutdown events — Use @asynccontextmanager lifespan, not @app.on_event("startup"). The event-based API is deprecated in FastAPI.
• Thread pool exhaustion — asyncio.to_thread() uses a default pool of ~5 workers. Under concurrent load, blocking SDK calls exhaust the pool. Set loop.set_default_executor(ThreadPoolExecutor(max_workers=20)) or use async clients.