The Golden-Path Concierge — natural language in, paved-path infrastructure out

mkdir concierge-lab && cd concierge-lab
git init -b main
python3 -m venv .venv && source .venv/bin/activate
pip install anthropic pydantic jsonschema
export ANTHROPIC_API_KEY=sk-ant-...
printf ".venv/\nproposals/\n" > .gitignore

ollama pull llama3.1          # fine for a five-field extraction; qwen2.5-coder also works

# Generated from the microservice golden path - DO NOT hand-edit fields
name: ${name}
team: ${team}
tier: ${tier}
port: ${port}
alert_route: team-${team}-oncall
base_image: registry.internal/base/python:3.12-hardened

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ${name}
  labels: {team: ${team}, tier: ${tier}}
spec:
  replicas: ${replicas}
  template:
    spec:
      containers:
        - name: ${name}
          image: registry.internal/${team}/${name}:latest
          ports: [{containerPort: ${port}}]
          resources:
            limits: {cpu: "500m", memory: 256Mi}

# ${name}

Owned by **${team}** · tier: ${tier} · scaffolded from the microservice golden path.

{
  "type": "object",
  "additionalProperties": false,
  "required": ["name", "team", "tier", "replicas", "port"],
  "properties": {
    "name":     {"type": "string", "pattern": "^[a-z][a-z0-9-]{2,29}$"},
    "team":     {"enum": ["payments", "identity", "search"]},
    "tier":     {"enum": ["internal", "partner"]},
    "replicas": {"type": "integer", "minimum": 1, "maximum": 4},
    "port":     {"type": "integer", "minimum": 3000, "maximum": 9999}
  }
}

#!/usr/bin/env python3
"""Golden-Path Concierge: NL request -> validated template invocation.

The model only ever fills parameters of an EXISTING template. Layers:
schema-constrained extraction -> code re-validation -> policy gate ->
deterministic render -> branch/PR output. No cloud credentials anywhere.
"""
import json, pathlib, string, subprocess, sys, time
from typing import Literal, Optional
from anthropic import Anthropic
from jsonschema import validate, ValidationError
from pydantic import BaseModel, ConfigDict, Field

TEMPLATE_DIR = pathlib.Path("templates/microservice")
SCHEMA = json.loads((TEMPLATE_DIR / "params.schema.json").read_text())
REGISTRY = pathlib.Path("registry.jsonl")
TEAM_QUOTA = 3
RESERVED_PREFIXES = ("prod-", "admin-", "infra-")

# --- Pydantic mirror of params.schema.json (keep in lockstep) ---
class ServiceParams(BaseModel):
    model_config = ConfigDict(extra="forbid")
    name: str = Field(pattern=r"^[a-z][a-z0-9-]{2,29}$")
    team: Literal["payments", "identity", "search"]
    tier: Literal["internal", "partner"]
    replicas: int = Field(ge=1, le=4)
    port: int = Field(ge=3000, le=9999)

class ConciergeIntake(BaseModel):
    request_supported: bool          # false if ANY part is off the paved path
    params: Optional[ServiceParams]  # filled only when fully supported
    unsupported_asks: list[str]      # every ask the template cannot express

def policy_gate(p: ServiceParams) -> Optional[str]:
    """Return a denial reason, or None if allowed. Deterministic. No AI."""
    if any(p.name.startswith(x) for x in RESERVED_PREFIXES):
        return f"name '{p.name}' uses a reserved prefix {RESERVED_PREFIXES}"
    existing = [json.loads(l) for l in REGISTRY.read_text().splitlines()] \
        if REGISTRY.exists() else []
    if any(e["name"] == p.name for e in existing):
        return f"service '{p.name}' already exists in the registry"
    count = sum(1 for e in existing if e["team"] == p.team)
    if count >= TEAM_QUOTA:
        return (f"team '{p.team}' is at its quota ({TEAM_QUOTA} services); "
                f"request an increase from the platform team")
    return None

def main():
    request = " ".join(sys.argv[1:]) or sys.exit("usage: concierge.py <request>")

    # 1) Extraction: the model fills parameters, nothing more.
    client = Anthropic()
    response = client.messages.parse(
        model="claude-opus-4-8",
        max_tokens=4096,
        thinking={"type": "adaptive"},
        system=(
            "You are a golden-path concierge. You take a natural-language "
            "request for a new microservice and fill ONLY the parameters of "
            "the existing template: name, team, tier (internal|partner), "
            "replicas (1-4; low traffic=1, default 2), port (default 8080). "
            "You never design infrastructure. If the request asks for "
            "ANYTHING the template cannot express (storage, buckets, "
            "databases, queues, public exposure, custom images, IAM), set "
            "request_supported=false, params=null, and list each such ask "
            "verbatim in unsupported_asks."),
        messages=[{"role": "user", "content": request}],
        output_format=ConciergeIntake,
    )
    intake = response.parsed_output

    if not intake.request_supported or intake.params is None:
        print("DENIED - request is off the paved path:")
        for ask in intake.unsupported_asks:
            print(f"  - '{ask}' is not a parameter of the microservice "
                  f"golden path")
        print("Policy: off-path infrastructure requires a platform request "
              "(see platform/requests). Drop the off-path ask and retry, "
              "or file the request.")
        return

    # 2) Never trust, always verify: re-validate against the JSON schema.
    try:
        validate(instance=intake.params.model_dump(), schema=SCHEMA)
    except ValidationError as e:
        print(f"DENIED - schema re-validation failed: {e.message}"); return

    # 3) Policy gate: deterministic org rules.
    reason = policy_gate(intake.params)
    if reason:
        print(f"DENIED - policy: {reason}"); return

    # 4) Deterministic render of the blessed template.
    p = intake.params
    outdir = pathlib.Path("services") / p.name
    outdir.mkdir(parents=True, exist_ok=True)
    for tmpl in TEMPLATE_DIR.glob("*.tmpl"):
        rendered = string.Template(tmpl.read_text()).substitute(p.model_dump())
        (outdir / tmpl.name.removesuffix(".tmpl")).write_text(rendered)

    # 5) Output as a branch (the Everything-Ends-in-a-PR adapter).
    base = subprocess.run(["git", "rev-parse", "--abbrev-ref", "HEAD"],
                          capture_output=True, text=True).stdout.strip()
    branch = f"agent/concierge-{p.name}-{int(time.time())}"
    subprocess.run(["git", "switch", "-c", branch], check=True)
    subprocess.run(["git", "add", str(outdir)], check=True)
    subprocess.run(["git", "-c", "user.name=concierge",
                    "-c", "user.email=concierge@agents.local", "commit", "-m",
                    f"scaffold {p.name} for {p.team} via golden path\n\n"
                    f"Request: {request}\nParams: {p.model_dump()}"],
                   check=True)
    subprocess.run(["git", "switch", base], check=True)

    with REGISTRY.open("a") as f:
        f.write(json.dumps({"name": p.name, "team": p.team, "tier": p.tier,
                            "branch": branch, "request": request,
                            "ts": time.strftime("%Y-%m-%dT%H:%M:%S")}) + "\n")
    print(f"APPROVED - rendered golden path for '{p.name}' on {branch}")
    print(f"Params: {p.model_dump()}")
    print("Next: human review + merge of the branch (nothing provisioned yet).")

if __name__ == "__main__":
    main()

python concierge.py "Need a returns-eligibility API for payments, internal only, low traffic"
git branch --list 'agent/*'
git show $(git branch --list 'agent/*' --format='%(refname:short)' | head -1) --stat
tail -1 registry.jsonl

python - <<'EOF'
import json
from jsonschema import validate, ValidationError
schema = json.load(open("templates/microservice/params.schema.json"))
bad = {"name": "returns-v2", "team": "payments", "tier": "internal",
       "replicas": 12, "port": 8080}          # 12 replicas: over the max
try:
    validate(instance=bad, schema=schema)
except ValidationError as e:
    print("re-validation caught it:", e.message)
EOF

python concierge.py "Spin up a checkout-exporter service for payments, and give it a public S3 bucket for the logs"