Code Gen

Code Generation Skill

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

Core 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 30 lines, decompose it into named sub-functions (the pre-write-gate hook enforces this limit deterministically).
• 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.

8. Deep Modules — Simple Interface, Useful Behavior

• Prefer modules with small, stable interfaces that hide meaningful complexity.
• A module interface includes its types, invariants, error modes, ordering requirements, and configuration — not just the function signature.
• Apply the deletion test before adding a module: if deleting it removes complexity entirely, it was probably shallow ceremony; if deleting it spreads complexity across callers, it is earning its keep.
• Do not create pass-through services, repositories, hooks, helpers, or adapters just to satisfy a pattern.
• One implementation behind an interface is not proof an abstraction is needed. Introduce interfaces/adapters when there are two real implementations, a test boundary around an external dependency, or a clear domain seam.

9. Public Interface as Test Surface

• Tests should verify observable behavior through public interfaces: API endpoints, CLI commands, UI flows, exported module functions, or documented domain services.
• Do not test private helpers, implementation details, internal call order, or mock interactions unless that is the public contract.
• If a helper is complex enough to need direct tests, consider making it a named domain module with a clear public interface.
• A good test should survive internal refactors when behavior is unchanged.

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. Tracer-bullet TDD: one behavior at a time.
   • Do not write all tests first, then all implementation.
   • Write one failing behavior test through the public interface.
   • Implement the minimum code to pass that test.
   • Repeat for the next behavior.
   • This prevents imagined tests for imagined architecture.
2. 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
3. 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%.
4. Only mock external boundaries: databases, third-party APIs, file I/O, clocks, payment processors, queues.
5. Never mock business logic — if you mock a service to test another service, you are hiding bugs and testing wiring instead of behavior.
6. 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.
7. 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.
8. Realistic test data — use domain-representative values (real-looking emails, valid UUIDs, plausible amounts). Never "foo", 123, or "test".
9. Test names describe behavior: "returns 404 when order does not exist", not "test order".
10. 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),
    },
)

6. Prompt Caching (Anthropic API)

When the generated app calls the Claude API (the anthropic SDK), make it cache-friendly by default. Prompt caching reuses the unchanging prefix of a request, cutting cost and latency on repeated calls. This is an Anthropic-API-specific feature — do not apply it to OpenAI or other providers (OpenAI caches automatically with no breakpoint API).

• Static content first. Order each request so unchanging content comes before variable content: tools, then system, then long-lived context (instructions, docs, examples, schemas), then the moving conversation tail. Caching only helps a prefix that is byte-for-byte identical across calls, so never interleave per-request data (timestamps, user IDs, retrieved snippets) into that prefix.
• One breakpoint on the stable prefix. Put a single cache_control breakpoint ({"type": "ephemeral"}) on the last block of the stable prefix — typically the end of the system prompt or the end of a large tool/context block. Everything up to and including that block becomes the cached prefix.
• Rely on auto-caching for the moving tail. Do not stamp cache_control on every turn. Anthropic automatically extends cache coverage to the longest previously-seen prefix, so the one breakpoint at the end of the stable section is reused by later turns without per-message markers. Reserve extra breakpoints (max 4 total) only for additional large, reused segments.
• Keep the prefix above the cache threshold. Cacheable prefixes have a model-dependent minimum (~1024+ tokens). Tiny system prompts will not cache — do not contort small prompts to force it.
• Verify cache behavior. Confirm the breakpoint lands where intended by inspecting usage: cache_creation_input_tokens on the first call (cache write) and cache_read_input_tokens on later calls (cache hit).

# CORRECT — stable prefix cached, moving tail auto-cached
resp = client.messages.create(
    model="claude-...",
    system=[
        {
            "type": "text",
            "text": LONG_STATIC_INSTRUCTIONS,          # unchanging across calls
            "cache_control": {"type": "ephemeral"},     # breakpoint at end of stable prefix
        }
    ],
    messages=conversation,                              # moving tail: no cache_control needed
)
# resp.usage.cache_creation_input_tokens / resp.usage.cache_read_input_tokens

7. Batch API for Bulk, Non-Interactive Work (Anthropic API)

When the generated app does bulk, one-shot LLM work — classify a backlog, summarize a document set, score a corpus, run an offline backfill — use the Anthropic Message Batches API instead of one synchronous call per item. Batch requests are 50% cheaper than standard calls, and the discount stacks with prompt caching (a cached, batched request pays ~50% of the already-90%-discounted cache-read rate).

• Use it only for non-interactive paths. Batch is asynchronous (results land within ~24h, usually much sooner) and is not conversational — there is no mid-request tool loop. Never use it for a live request/response path, an interactive agent loop, or anything a user waits on. Those stay synchronous.
• Right shape: a queue/cron/worker that has many independent prompts ready at once. Wrong shape: a web handler serving one user, or an agent that needs each result to decide its next step.
• Combine with caching: put the shared instructions/schema in the cached prefix (section 6) so every item in the batch reuses it.
• This is Anthropic-API-specific. For OpenAI/other providers, use their equivalent batch endpoint or skip.

# CORRECT — bulk classification as one batch job, not N synchronous calls
batch = client.messages.batches.create(
    requests=[
        {"custom_id": item.id, "params": {"model": "claude-...", "max_tokens": 256,
         "system": SHARED_PREFIX, "messages": [{"role": "user", "content": item.text}]}}
        for item in backlog          # many independent items, no inter-item dependency
    ]
)
# poll batch.id, then retrieve results — do NOT block a user request on this

8. Output Budgeting

Output tokens are billed at the highest rate and are fully re-read as input on the next turn of any conversation, so unbounded output compounds cost. Constrain it in generated LLM calls:

• Always set an explicit max_tokens sized to the expected response — not the model maximum. A classifier returning a small JSON object needs ~256, not 4096.
• Ask for the smallest sufficient output. Prefer structured output (section 1) and IDs/enums over prose; instruct the model to omit restated context and long explanations unless required.
• Don't request reasoning you discard. Only enable extended thinking when the task needs it; for mechanical/structured tasks, leave it off.

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

The canonical error-envelope shape and the layering rules live in .claude/skills/code-gen/references/architecture.md — defer to it if the two ever differ. Repeated here for convenience: 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 30 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.