feat(architect): generate → run → judge → refine loop that ships Proven templates

README.md

@@ -68,6 +68,7 @@
 - [Directory Input & CSV Export](#directory-input--csv-export)
 - [Community Hub](#community-hub)
 - [Verified Templates (.sctpl)](#verified-templates-sctpl)
+  - [The Architect](#the-architect)
 - [Real-time Event Stream](#real-time-event-stream)
 - [Run Time Machine](#run-time-machine)
 - [Budget Guardrails](#budget-guardrails)
@@ -1530,6 +1531,20 @@ The verification is mechanical, not a badge. The replay runs in strict mode: eve
 
 Installed bundles land in the community templates directory, so they appear in the Lite wizard and dashboard immediately - with the proof cassette kept alongside for re-verification anytime. A working example ships in the repo: `examples/templates/text-summarizer-1.0.0.sctpl`. Format details: [docs/verified-templates.md](docs/verified-templates.md).
 
+### The Architect
+
+Describe a workflow and get back a proven one. The Architect runs an autonomous loop - generate, run, evaluate, refine - until the workflow actually works: it generates YAML from your description, executes it live with cassette recording on, hard-checks the run (completed, no failures, real output), scores the output against your description with an LLM judge, and feeds any shortfall back into the generator for the next iteration. On success it packs the workflow and its freshly recorded cassette into a verified `.sctpl` bundle and installs it - a template born with its proof attached.
+
+```bash
+sandcastle architect "Summarize a customer email and draft a polite reply" \
+    --input email="Hi, my order arrived damaged..." --budget 0.50
+#   iteration 1: completed  cost $0.0214  score 0.86
+#   PROVEN  email-reply-assistant
+#   Bundle: email-reply-assistant-1.0.0.sctpl
+```
+
+The whole session is budget-capped (`ARCHITECT_BUDGET_USD`, default $1.00) and bounded (`ARCHITECT_MAX_ITERATIONS`, default 3; `ARCHITECT_SCORE_THRESHOLD`, default 0.7). Test inputs come from you (`--input`), the generated input schema, or a plausible set the advisor derives. Also available as an async API job: `POST /api/architect` starts a session, `GET /api/architect/{job_id}` streams the per-iteration log, scores, and the final bundle path.
+
 ---
 
 ## Real-time Event Stream

src/sandcastle/__main__.py

@@ -3554,6 +3554,79 @@ def _cmd_template(args: argparse.Namespace) -> None:
         sys.exit(1)
 
 
+# ---------------------------------------------------------------------------
+# The Architect - NL description -> proven workflow template
+# ---------------------------------------------------------------------------
+
+
+def _cmd_architect(args: argparse.Namespace) -> None:
+    """Run the Architect loop synchronously: generate -> run -> evaluate -> refine.
+
+    Prints per-iteration progress; on success exits 0 with the bundle path of
+    the freshly proven template. Exits 2 when the loop ends without a proof.
+    """
+    import asyncio
+
+    from sandcastle.engine.architect import design_workflow
+
+    if args.budget is not None and args.budget <= 0:
+        print("Error: --budget must be positive.", file=sys.stderr)
+        sys.exit(1)
+
+    test_input: dict[str, Any] = {}
+    if args.input_file:
+        test_input = _load_input_file(args.input_file)
+    test_input.update(_parse_input_pairs(args.input))
+
+    def _progress(msg: str) -> None:
+        print(_color(f"  {msg}", _C.CYAN))
+
+    print(_color(f"The Architect: {args.description!r}", _C.BOLD))
+    try:
+        result = asyncio.run(
+            design_workflow(
+                args.description,
+                test_input=test_input or None,
+                budget_usd=args.budget,
+                max_iterations=args.max_iterations,
+                score_threshold=args.threshold,
+                output_dir=args.output,
+                install=not args.no_install,
+                progress=_progress,
+            )
+        )
+    except Exception as exc:
+        print(_format_cli_error(exc), file=sys.stderr)
+        sys.exit(1)
+
+    if getattr(args, "json", False):
+        print(json.dumps(result.to_dict(), indent=2, default=str))
+    else:
+        for it in result.iterations:
+            score = "-" if it.judge_score is None else f"{it.judge_score:.2f}"
+            line = (
+                f"  iteration {it.iteration}: {it.run_status or 'not run'}"
+                f"  cost ${it.run_cost_usd:.4f}  score {score}"
+            )
+            print(line)
+            for failure in it.hard_check_failures:
+                print(_color(f"    ! {failure}", _C.YELLOW))
+        print(f"  total live-run cost: ${result.total_cost_usd:.4f}")
+
+    if result.proven:
+        print(_color(f"PROVEN  {result.template_name}", _C.GREEN))
+        print(f"  Bundle: {result.bundle_path}")
+        if result.installed_path:
+            print(f"  Installed: {result.installed_path}")
+            print(f"  Run it: sandcastle run --local {result.installed_path}")
+        return
+
+    print(_color(f"NOT PROVEN ({result.status})", _C.RED), file=sys.stderr)
+    if result.error:
+        print(_color(f"  {result.error}", _C.RED), file=sys.stderr)
+    sys.exit(2)
+
+
 # ---------------------------------------------------------------------------
 # API helpers (shared by new command groups)
 # ---------------------------------------------------------------------------
@@ -5383,6 +5456,42 @@ def _build_parser() -> argparse.ArgumentParser:
     t_search.add_argument("query", help="Search query")
     t_search.add_argument("--json", action="store_true", help="JSON output")
 
+    # --- architect ---
+    p_architect = subparsers.add_parser(
+        "architect",
+        help="Turn a description into a proven workflow (generate -> run -> refine loop)",
+    )
+    p_architect.add_argument(
+        "description", help="Natural language description of the workflow to build"
+    )
+    p_architect.add_argument(
+        "--input", "-i", action="append",
+        help="Test input key=value for the proof run (repeatable)",
+    )
+    p_architect.add_argument(
+        "--input-file", "-f", help="JSON file with the test inputs for the proof run"
+    )
+    p_architect.add_argument(
+        "--budget", type=float, default=None,
+        help="Live-run spend cap in USD (default: architect_budget_usd)",
+    )
+    p_architect.add_argument(
+        "--max-iterations", type=int, default=None,
+        help="Loop bound (default: architect_max_iterations)",
+    )
+    p_architect.add_argument(
+        "--threshold", type=float, default=None,
+        help="Minimum judge score 0-1 (default: architect_score_threshold)",
+    )
+    p_architect.add_argument(
+        "--output", "-o", default=None, help="Directory for the .sctpl bundle"
+    )
+    p_architect.add_argument(
+        "--no-install", action="store_true",
+        help="Do not install the proven template into the community templates dir",
+    )
+    p_architect.add_argument("--json", action="store_true", help="JSON output")
+
     # --- describe ---
     p_describe = subparsers.add_parser(
         "describe", help="Print workflow summary with responsibilities"
@@ -5483,6 +5592,7 @@ def main() -> None:
         "hub": _cmd_hub,
         "pack": _cmd_pack,
         "template": _cmd_template,
+        "architect": _cmd_architect,
         "keys": _cmd_keys,
         "dlq": _cmd_dlq,
         "violations": _cmd_violations,

src/sandcastle/api/routes.py

@@ -39,6 +39,7 @@
     ApiResponse,
     ApprovalRespondRequest,
     ApprovalResponse,
+    ArchitectRequest,
     AuditEventResponse,
     AuditVerifyResponse,
     AutoPilotStatsResponse,
@@ -3720,6 +3721,113 @@ async def generate_chat(req: Request, request: GenerateChatRequest) -> ApiRespon
     return ApiResponse(data=result)
 
 
+# --- The Architect: autonomous generate -> run -> evaluate -> refine ---
+
+# In-memory job store, same pattern as the batch store. Architect sessions are
+# operator-initiated and few; a bounded dict keeps the endpoint dependency-free.
+_architect_jobs: dict[str, dict[str, Any]] = {}
+_ARCHITECT_JOBS_MAX_SIZE = 100
+
+
+def _prune_architect_jobs() -> None:
+    """Drop the oldest finished jobs when the store grows past its cap."""
+    if len(_architect_jobs) <= _ARCHITECT_JOBS_MAX_SIZE:
+        return
+    finished = [
+        jid for jid, j in _architect_jobs.items()
+        if j.get("status") in ("completed", "failed")
+    ]
+    for jid in finished[: len(_architect_jobs) - _ARCHITECT_JOBS_MAX_SIZE]:
+        _architect_jobs.pop(jid, None)
+
+
+async def _run_architect_job(job_id: str, request: ArchitectRequest) -> None:
+    """Drive one Architect session and mirror its progress into the job store."""
+    from sandcastle.engine.architect import design_workflow
+
+    job = _architect_jobs.get(job_id)
+    if job is None:
+        return
+    job["status"] = "running"
+
+    def _progress(msg: str) -> None:
+        job["log"].append(msg)
+
+    try:
+        result = await design_workflow(
+            request.description,
+            test_input=request.test_input,
+            budget_usd=request.budget_usd,
+            max_iterations=request.max_iterations,
+            score_threshold=request.score_threshold,
+            progress=_progress,
+        )
+        job["result"] = result.to_dict()
+        job["status"] = "completed"
+    except Exception as exc:
+        logger.error("Architect job %s failed: %s", job_id, exc)
+        job["status"] = "failed"
+        job["error"] = str(exc)
+    finally:
+        job["completed_at"] = datetime.now(timezone.utc).isoformat()
+
+
+@router.post("/architect")
+async def start_architect(req: Request, request: ArchitectRequest) -> ApiResponse:
+    """Start an Architect session: NL description -> proven workflow template.
+
+    Runs asynchronously; poll GET /api/architect/{job_id} for the per-iteration
+    log and the final bundle path. Admin-gated - the loop executes live runs.
+    """
+    _require_admin(req)
+    await execution_limiter.check(req)
+
+    if not settings.anthropic_api_key and not os.environ.get("ANTHROPIC_API_KEY"):
+        raise HTTPException(
+            status_code=400,
+            detail=ApiResponse(
+                error=ErrorResponse(
+                    code="MISSING_API_KEY",
+                    message="An advisor API key is required for the Architect",
+                )
+            ).model_dump(),
+        )
+
+    _prune_architect_jobs()
+    job_id = str(uuid.uuid4())
+    _architect_jobs[job_id] = {
+        "job_id": job_id,
+        "status": "queued",
+        "description": request.description,
+        "created_at": datetime.now(timezone.utc).isoformat(),
+        "completed_at": None,
+        "log": [],
+        "result": None,
+        "error": None,
+    }
+    asyncio.create_task(_run_architect_job(job_id, request))
+
+    return ApiResponse(data={"job_id": job_id, "status": "queued"})
+
+
+@router.get("/architect/{job_id}")
+async def get_architect_job(job_id: str, req: Request) -> ApiResponse:
+    """Status of an Architect job: per-iteration log, scores, final bundle."""
+    _require_admin(req)
+    job = _architect_jobs.get(job_id)
+    if job is None:
+        raise HTTPException(
+            status_code=404,
+            detail=ApiResponse(
+                error=ErrorResponse(
+                    code="NOT_FOUND",
+                    message=f"Architect job '{job_id}' not found",
+                )
+            ).model_dump(),
+        )
+    return ApiResponse(data=job)
+
+
 @router.post("/advisor/explain")
 async def advisor_explain_error(req: Request, request: ExplainErrorRequest) -> ApiResponse:
     """Explain a step failure using AI and suggest a fix."""

src/sandcastle/api/schemas.py

@@ -219,6 +219,33 @@ class WorkflowGenerateRequest(BaseModel):
     )
 
 
+class ArchitectRequest(BaseModel):
+    """Request to start an Architect session (generate -> run -> evaluate -> refine)."""
+
+    description: str = Field(
+        ...,
+        description="Natural language description of the workflow to build and prove",
+        min_length=1,
+        max_length=10000,
+    )
+    test_input: dict | None = Field(
+        None,
+        description="Inputs for the proof run (derived from the input schema if omitted)",
+    )
+    budget_usd: float | None = Field(
+        None, gt=0, le=1000,
+        description="Live-run spend cap (default: architect_budget_usd)",
+    )
+    max_iterations: int | None = Field(
+        None, ge=1, le=10,
+        description="Loop bound (default: architect_max_iterations)",
+    )
+    score_threshold: float | None = Field(
+        None, ge=0.0, le=1.0,
+        description="Minimum judge score (default: architect_score_threshold)",
+    )
+
+
 class GenerateChatMessage(BaseModel):
     """A single message in a chat-based generation conversation."""
 

src/sandcastle/config.py

@@ -110,6 +110,11 @@ class Settings(BaseSettings):
     # Budget
     default_max_cost_usd: float = 0.0  # 0 = no limit (must be >= 0)
 
+    # The Architect - autonomous generate->run->evaluate->refine loop
+    architect_max_iterations: int = 3  # loop bound per session
+    architect_budget_usd: float = 1.0  # total live-run spend cap per session
+    architect_score_threshold: float = 0.7  # minimum LLM-judge score to accept
+
     # Workflows directory (default: ~/.sandcastle/workflows)
     workflows_dir: str = _DEFAULT_WORKFLOWS_DIR
 
@@ -336,6 +341,43 @@ def _validate_default_max_cost(cls, v: float) -> float:
             return 0.0
         return v
 
+    @field_validator("architect_max_iterations", mode="after")
+    @classmethod
+    def _validate_architect_max_iterations(cls, v: int) -> int:
+        """Ensure architect_max_iterations is at least 1."""
+        if v < 1:
+            _logger.warning(
+                "ARCHITECT_MAX_ITERATIONS=%d is invalid (must be >= 1), using 3", v
+            )
+            return 3
+        return v
+
+    @field_validator("architect_budget_usd", mode="after")
+    @classmethod
+    def _validate_architect_budget(cls, v: float) -> float:
+        """Ensure architect_budget_usd is positive."""
+        if v <= 0:
+            _logger.warning(
+                "ARCHITECT_BUDGET_USD=%.2f is invalid (must be > 0), using 1.0", v
+            )
+            return 1.0
+        return v
+
+    @field_validator("architect_score_threshold", mode="after")
+    @classmethod
+    def _validate_architect_threshold(cls, v: float) -> float:
+        """Clamp architect_score_threshold to [0.0, 1.0]."""
+        if v < 0.0 or v > 1.0:
+            clamped = max(0.0, min(1.0, v))
+            _logger.warning(
+                "ARCHITECT_SCORE_THRESHOLD=%.2f is out of range [0.0, 1.0], "
+                "clamping to %.2f",
+                v,
+                clamped,
+            )
+            return clamped
+        return v
+
     @field_validator("tool_smtp_port", mode="after")
     @classmethod
     def _validate_tool_smtp_port(cls, v: int) -> int: