# Code Gen > Code generation quality principles — TDD, typing, error handling, logging, API integration, LLM integration. - Skill: `cwijayasundara/code-gen-4` (Agent Skill, multi-file: 12 files) - Install (CLI): `npx skillmds@latest add cwijayasundara/code-gen-4` - Raw SKILL.md: https://api.skillmd.com/api/skills/cwijayasundara/code-gen-4/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - Author: cwijayasundara (https://skillmd.com/u/cwijayasundara) - Updated: 2026-09-22 - Page: https://skillmd.com/skills/cwijayasundara/code-gen-4 --- # 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` 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 ```typescript 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. ```python # 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. ```python # 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. ```python # 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. ```python @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. ```python @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: ```python from pydantic import BaseModel from typing import Literal class ClassificationResult(BaseModel): category: str confidence: Literal["high", "medium", "low"] reasoning: str ``` ```typescript 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: ```python # 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: ```python # 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: ```python 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). ```python # 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. ```python # 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: ```python 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: ```python 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") ``` ```typescript // 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 ```python # 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: ```json { "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.