developer-farm AGENTS.md

CONTEXT FOR LANGGRAPH CODE GENERATION

1. Environment

• LangGraph version: 1.2.1 (CHECK with pip show langgraph before generating)
• Python: 3.11
• Pydantic: v2 (use from pydantic import BaseModel, NOT v1)
• Checkpointer: SqliteSaver (local file, no Postgres)

2. Architecture Constraints (GOODHART-PROOF ISOLATION)

This system implements 4-layer isolation. NEVER violate these boundaries:

• PLANNING nodes receive ONLY: user_spec, tech_spec, codebase_index
• EXECUTION nodes receive ONLY: task_description, context_files, git_worktree_path ❌ NEVER pass: acceptance_criteria, test_files, verification_rubric
• VERIFICATION nodes receive ONLY: SealedArtifact (git_diff + logs), rubric ❌ NEVER pass: worker_id, original_task_prompt, chat_history, planning_context
• OPTIMIZATION nodes receive ONLY: aggregated verdicts, metrics summary ❌ NEVER pass: artifact contents, current graph state, raw logs

All state schemas MUST use TypedDict with explicit field lists. If a field is not in the TypedDict, it CANNOT be passed between nodes.

3. Required Patterns

• Use StateGraph (not MessageGraph)
• Use add_conditional_edges for validation loops (max 3 iterations)
• Use interrupt() from langgraph.types for human-in-the-loop (NOT deprecated NodeInterrupt)
• Use Send() API for parallel fan-out in execution wave
• Always compile with checkpointer: graph.compile(checkpointer=checkpointer)
• Stream via graph.astream_events(config, version="v2")

4. Working Example Reference

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Send

class ExecutionInput(TypedDict):
    task_description: str
    context_files: list[str]
    worktree_path: str
    # ❌ NO acceptance_criteria, NO rubric

class SealedArtifact(TypedDict):
    git_diff: str
    logs: str
    # ❌ NO worker_id, NO task_description

async def code_worker(state: ExecutionInput) -> dict:
    # Call local vLLM or API here
    artifact = await generate_code(state)
    return {"sealed_artifact": artifact}

async def blind_verifier(state: dict) -> dict:
    artifact = state["sealed_artifact"]
    rubric = state["rubric"]
    # ❌ Cannot access state["task_description"] or state["worker_id"]
    verdict = await verify(artifact, rubric)
    return {"verdict": verdict}

# Conditional edge for retry loop
def should_retry(state: dict) -> str:
    if state["verdict"]["passed"] or state["iteration"] >= 3:
        return "approved"
    return "revise"

builder = StateGraph(dict)
builder.add_node("code_worker", code_worker)
builder.add_node("blind_verifier", blind_verifier)
builder.add_conditional_edges("blind_verifier", should_retry, {
    "approved": END,
    "revise": "code_worker"
})
checkpointer = SqliteSaver.from_conn_string("./checkpoints.db")
graph = builder.compile(checkpointer=checkpointer)