From cea0882cf3450b3422175ef4c21eb2f3251ebc0a Mon Sep 17 00:00:00 2001 From: Tomas Pflanzer Date: Wed, 10 Jun 2026 14:19:35 +0200 Subject: [PATCH 1/3] feat(architect): autonomous generate->run->evaluate->refine loop with proof bundles engine/architect.py drives the loop: generate a workflow from a NL description via the existing generator, execute it live with cassette recording on (budget- and time-capped, admin_trusted=False), hard-check the run and LLM-judge the output against the description, and feed failures back through the generator's refine path. On success the workflow + recorded cassette are packed into a .sctpl bundle, verified with the same verify_bundle that gates template installs, and installed into the community templates dir. Config: architect_max_iterations (3), architect_budget_usd (1.0), architect_score_threshold (0.7) with clamping validators. Co-Authored-By: Claude Fable 5 --- src/sandcastle/config.py | 42 +++ src/sandcastle/engine/architect.py | 492 +++++++++++++++++++++++++++++ 2 files changed, 534 insertions(+) create mode 100644 src/sandcastle/engine/architect.py diff --git a/src/sandcastle/config.py b/src/sandcastle/config.py index 531194fa..419dc3c3 100644 --- a/src/sandcastle/config.py +++ b/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: diff --git a/src/sandcastle/engine/architect.py b/src/sandcastle/engine/architect.py new file mode 100644 index 00000000..3cc4c61f --- /dev/null +++ b/src/sandcastle/engine/architect.py @@ -0,0 +1,492 @@ +"""The Architect: an autonomous generate -> run -> evaluate -> refine loop. + +Give it a natural-language description and it returns a WORKING workflow with a +recorded cassette as proof. Each iteration: (1) generate (or refine) a workflow +via the existing NL generator, (2) execute it live with cassette RECORDING on, +budget- and time-capped, (3) evaluate the run - hard checks (completed, no +errors, non-empty output) plus an LLM judge scoring the output against the +original description, (4) below threshold, feed the failure back into the +generator's refine path and try again. On success the workflow and its freshly +recorded cassette are packed into a .sctpl bundle (the same proof format +``sandcastle template verify`` replays offline at $0) and installed where the +template hub surfaces it. + +Costs: live run spend is tracked against ``architect_budget_usd`` and each run +is additionally capped at the remaining budget, so the loop can never overshoot +by more than one step. Advisor calls (generation/judge) are routed through the +same failover stack the generator uses. + +Generated workflows are executed with ``admin_trusted=False`` - the loop never +runs ``code`` steps from model output - and the generator is instructed to emit +only replay-safe ``standard`` steps so the resulting bundle verifies. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import re +import tempfile +from collections.abc import Callable +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +# Appended to the user's description so the generator emits workflows whose +# bundle proof verifies offline (only `standard` steps are cassette-covered). +_REPLAY_SAFE_INSTRUCTION = ( + "\n\nConstraints: use ONLY standard prompt steps (no code, http, browser, " + "tool, parse, or approval steps) so the workflow is fully reproducible " + "from a recorded cassette." +) + +_JUDGE_SYSTEM = ( + "You are a strict quality evaluator. Respond with ONLY a number between " + "0.0 and 1.0." +) + +_INPUT_GEN_SYSTEM = ( + "You generate realistic test inputs for workflows. Respond with ONLY a " + "JSON object, no explanations and no markdown fencing." +) + +_PLACEHOLDERS: dict[str, Any] = { + "string": "example", + "number": 1, + "integer": 1, + "boolean": True, + "array": [], + "object": {}, +} + + +@dataclass +class IterationLog: + """One pass of the generate -> run -> evaluate -> refine loop.""" + + iteration: int + generated: bool = False + validation_errors: list[str] = field(default_factory=list) + run_status: str = "" + run_cost_usd: float = 0.0 + hard_check_failures: list[str] = field(default_factory=list) + judge_score: float | None = None + refinement_note: str = "" + + +@dataclass +class ArchitectResult: + """Final outcome of an Architect session.""" + + status: str # "proven" | "verify_failed" | "max_iterations" | "budget_exceeded" | "error" + proven: bool + description: str + yaml_content: str = "" + workflow_name: str = "" + test_input: dict[str, Any] = field(default_factory=dict) + bundle_path: str | None = None + installed_path: str | None = None + template_name: str | None = None + total_cost_usd: float = 0.0 + best_score: float = 0.0 + iterations: list[IterationLog] = field(default_factory=list) + error: str | None = None + + def to_dict(self) -> dict[str, Any]: + """JSON-serializable view (for the API job store and CLI --json).""" + data = asdict(self) + # The full YAML of every iteration would bloat the job payload; the + # final YAML is what callers act on. + return data + + +def _sanitize_stem(name: str) -> str: + """Sanitize a workflow name to a safe filename stem (no traversal).""" + base = name.split("/")[-1] + safe = re.sub(r"[^a-zA-Z0-9_\-]", "_", base).lstrip(".") + return safe or "workflow" + + +async def _judge_output( + description: str, outputs: dict[str, Any] +) -> tuple[float, str]: + """Score the run output 0.0-1.0 against the original description via LLM. + + Mirrors the AutoPilot judge: a single cheap call that must return a bare + number. On any failure returns (0.5, note) so one flaky judge call neither + blesses nor kills an iteration. + """ + from sandcastle.engine.generator import _call_advisor_llm + + output_str = json.dumps(outputs, default=str)[:4000] + user_msg = ( + "A workflow was built from this request:\n" + f"{description}\n\n" + "Its run produced these outputs:\n" + f"{output_str}\n\n" + "Rate from 0.0 (output does not satisfy the request at all) to 1.0 " + "(output fully satisfies the request). Respond with ONLY the number." + ) + try: + raw = await _call_advisor_llm( + system=_JUDGE_SYSTEM, user=user_msg, max_tokens=16, purpose="judge" + ) + score = max(0.0, min(1.0, float(raw.strip()))) + return score, f"judge scored {score:.2f}" + except Exception as exc: # noqa: BLE001 - judge must never crash the loop + logger.warning("Architect judge call failed: %s", exc) + return 0.5, f"judge unavailable ({exc}); neutral 0.5 assumed" + + +async def _derive_test_input( + description: str, input_schema: dict[str, Any] | None +) -> dict[str, Any]: + """Build a plausible test input for the generated workflow. + + Order: schema defaults/examples -> one LLM call for the leftovers -> + type-based placeholders. Always returns a dict covering every declared + property, so the run never fails on missing input. + """ + props = (input_schema or {}).get("properties", {}) or {} + test_input: dict[str, Any] = {} + missing: dict[str, Any] = {} + for key, spec in props.items(): + spec = spec if isinstance(spec, dict) else {} + if "default" in spec: + test_input[key] = spec["default"] + elif "example" in spec: + test_input[key] = spec["example"] + else: + missing[key] = spec + if not missing: + return test_input + + from sandcastle.engine.generator import _call_advisor_llm + + try: + user_msg = ( + f"Workflow purpose: {description}\n\n" + f"Produce a realistic test value for each input property:\n" + f"{json.dumps(missing, default=str)}\n\n" + "Return a single JSON object mapping property name to value." + ) + raw = await _call_advisor_llm( + system=_INPUT_GEN_SYSTEM, user=user_msg, max_tokens=512, + purpose="generation", + ) + raw = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw.strip()) + generated = json.loads(raw) + if isinstance(generated, dict): + for key in missing: + if key in generated: + test_input[key] = generated[key] + except Exception as exc: # noqa: BLE001 - fall through to placeholders + logger.warning("Architect test-input generation failed: %s", exc) + + for key, spec in missing.items(): + if key not in test_input: + test_input[key] = _PLACEHOLDERS.get(str(spec.get("type", "string")), "example") + return test_input + + +def install_bundle( + bundle_path: str | Path, target_dir: str | Path | None = None +) -> Path: + """Install a bundle's workflow + proof cassettes like ``template install``. + + Writes into the community templates directory (where the dashboard hub and + Lite wizard look) unless *target_dir* is given. Returns the installed + workflow path. + """ + from sandcastle.engine.bundle import read_bundle + + manifest, workflow_yaml, cassettes = read_bundle(bundle_path) + if target_dir is None: + from sandcastle.templates import _TEMPLATES_DIR + + target_dir = _TEMPLATES_DIR / "community" + target_dir = Path(target_dir) + target_dir.mkdir(parents=True, exist_ok=True) + + stem = _sanitize_stem(str(manifest["name"])) + target_path = (target_dir / f"{stem}.yaml").resolve() + if not str(target_path).startswith(str(target_dir.resolve())): + raise ValueError("unsafe template name in manifest") + target_path.write_text(workflow_yaml) + for arcname, blob in cassettes.items(): + cassette_name = _sanitize_stem(Path(arcname).stem) + ".cassette.json" + (target_dir / f"{stem}.{cassette_name}").write_bytes(blob) + return target_path + + +async def design_workflow( + description: str, + *, + test_input: dict[str, Any] | None = None, + budget_usd: float | None = None, + max_iterations: int | None = None, + score_threshold: float | None = None, + run_timeout_seconds: float = 600.0, + output_dir: str | Path | None = None, + install: bool = True, + install_dir: str | Path | None = None, + tenant_id: str | None = None, + progress: Callable[[str], None] | None = None, +) -> ArchitectResult: + """Turn a natural-language description into a proven workflow template. + + Runs the generate -> run -> evaluate -> refine loop until a run completes + with a judge score at or above the threshold, then packs the workflow and + its recorded cassette into a verified .sctpl bundle and installs it. + + Args: + description: What the workflow should do (natural language). + test_input: Inputs for the proof run; derived from the generated + input schema (defaults -> LLM -> placeholders) when omitted. + budget_usd: Total live-run spend cap (default: settings.architect_budget_usd). + max_iterations: Loop bound (default: settings.architect_max_iterations). + score_threshold: Minimum judge score (default: settings.architect_score_threshold). + run_timeout_seconds: Wall-clock cap per proof run. + output_dir: Where the .sctpl bundle is written (default: temp dir). + install: Install the proven bundle like ``template install`` does. + install_dir: Override the install target (default: community templates dir). + tenant_id: Calling tenant, forwarded to the generator. + progress: Optional callback receiving human-readable progress lines. + + Returns: + ArchitectResult - ``proven=True`` only when the packed bundle passes + the same ``verify_bundle`` check that gates template installs. + """ + from sandcastle.config import settings + from sandcastle.engine import generator as _generator + from sandcastle.engine.cassette import CassetteStore + from sandcastle.engine.dag import build_plan, parse_yaml_string + from sandcastle.engine.executor import execute_workflow + + budget = settings.architect_budget_usd if budget_usd is None else float(budget_usd) + iterations = ( + settings.architect_max_iterations if max_iterations is None else int(max_iterations) + ) + threshold = ( + settings.architect_score_threshold if score_threshold is None else float(score_threshold) + ) + iterations = max(1, iterations) + + def _say(msg: str) -> None: + logger.info("Architect: %s", msg) + if progress is not None: + try: + progress(msg) + except Exception: # noqa: BLE001 - progress is cosmetic + pass + + result = ArchitectResult(status="error", proven=False, description=description) + workdir = Path(output_dir) if output_dir else Path( + tempfile.mkdtemp(prefix="sandcastle-architect-") + ) + workdir.mkdir(parents=True, exist_ok=True) + + gen_description = description + _REPLAY_SAFE_INSTRUCTION + yaml_current: str | None = None + feedback = "" + spent = 0.0 + best_score = -1.0 + best_yaml = "" + + for i in range(1, iterations + 1): + log = IterationLog(iteration=i) + result.iterations.append(log) + + # --- 1. generate / refine ----------------------------------------- + try: + if yaml_current is None: + _say(f"iteration {i}: generating workflow from description") + gen = await _generator.generate_workflow( + gen_description, tenant_id=tenant_id + ) + else: + _say(f"iteration {i}: refining workflow ({feedback[:120]})") + log.refinement_note = feedback + gen = await _generator.generate_workflow( + gen_description, + refine_from=yaml_current, + refine_instruction=feedback, + tenant_id=tenant_id, + ) + except Exception as exc: # noqa: BLE001 - report, don't crash the job + result.error = f"generation failed: {exc}" + result.status = "error" + break + log.generated = True + yaml_current = gen.yaml_content + result.yaml_content = yaml_current + log.validation_errors = list(gen.validation_errors) + if gen.validation_errors: + feedback = "Fix these validation errors: " + "; ".join(gen.validation_errors) + log.run_status = "skipped (validation errors)" + continue + + try: + wf = parse_yaml_string(yaml_current) + plan = build_plan(wf) + except Exception as exc: # noqa: BLE001 - feed parse failure back + feedback = f"The YAML does not parse/plan: {exc}. Fix it." + log.run_status = "skipped (parse error)" + continue + result.workflow_name = wf.name + + # --- 2. test input (once) ------------------------------------------ + if test_input is None: + _say(f"iteration {i}: deriving test inputs from the input schema") + test_input = await _derive_test_input(description, gen.input_schema) + result.test_input = dict(test_input) + + # --- 3. run with cassette recording, budget- and time-capped -------- + remaining = budget - spent + if remaining <= 0: + result.status = "budget_exceeded" + result.error = f"budget ${budget:.2f} exhausted before iteration {i} could run" + break + cassette_path = workdir / f"iter{i}.cassette.json" + store = CassetteStore(cassette_path, "record") + _say(f"iteration {i}: running '{wf.name}' (budget left ${remaining:.4f})") + try: + wf_result = await asyncio.wait_for( + execute_workflow( + workflow=wf, + plan=plan, + input_data=dict(test_input), + max_cost_usd=remaining, + admin_trusted=False, # never execute code steps from model output + tenant_id=tenant_id, + cassette=store, + cassette_mode="record", + ), + timeout=run_timeout_seconds, + ) + except asyncio.TimeoutError: + feedback = ( + f"The run timed out after {run_timeout_seconds:.0f}s. Make the " + "workflow simpler/faster." + ) + log.run_status = "timeout" + continue + except Exception as exc: # noqa: BLE001 - feed run crash back + feedback = f"The run crashed: {exc}. Fix the workflow." + log.run_status = "crashed" + continue + + run_cost = float(getattr(wf_result, "total_cost_usd", 0.0) or 0.0) + spent += run_cost + result.total_cost_usd = round(spent, 6) + log.run_status = str(getattr(wf_result, "status", "unknown")) + log.run_cost_usd = round(run_cost, 6) + + if log.run_status == "budget_exceeded": + result.status = "budget_exceeded" + result.error = f"run hit the architect budget cap (${budget:.2f} total)" + break + + # --- 4. evaluate: hard checks + LLM judge --------------------------- + outputs = getattr(wf_result, "outputs", {}) or {} + hard_failures: list[str] = [] + if log.run_status != "completed": + hard_failures.append(f"run finished with status '{log.run_status}'") + run_error = getattr(wf_result, "error", None) + if run_error: + hard_failures.append(f"run error: {run_error}") + if not outputs or all(v in (None, "", {}, []) for v in outputs.values()): + hard_failures.append("run produced empty output") + log.hard_check_failures = hard_failures + + if hard_failures: + score = 0.0 + judge_note = "skipped (hard checks failed)" + else: + _say(f"iteration {i}: judging output against the description") + score, judge_note = await _judge_output(description, outputs) + log.judge_score = score + if score > best_score: + best_score, best_yaml = score, yaml_current + result.best_score = max(0.0, best_score) + + # --- 5. success: pack the proof ------------------------------------ + if not hard_failures and score >= threshold: + store.save() + stem = _sanitize_stem(wf.name) + wf_file = workdir / f"{stem}.yaml" + wf_file.write_text(yaml_current) + bundle_path = workdir / f"{stem}-1.0.0.sctpl" + _say(f"iteration {i}: PASS (score {score:.2f}) - packing proof bundle") + try: + from sandcastle.engine.bundle import create_bundle, verify_bundle + + create_bundle( + wf_file, + [cassette_path], + bundle_path, + name=wf.name, + description=wf.description, + author="the-architect", + example_inputs=dict(test_input), + ) + result.bundle_path = str(bundle_path) + # verify_bundle drives its own event loop - keep it off ours. + verify = await asyncio.to_thread(verify_bundle, bundle_path) + except Exception as exc: # noqa: BLE001 - packing must not lose the win + result.status = "verify_failed" + result.error = f"bundle packing/verification failed: {exc}" + break + if not verify.ok: + result.status = "verify_failed" + result.error = "bundle did not verify: " + "; ".join( + verify.errors + + [c.detail for c in verify.cassette_results if not c.passed] + ) + break + result.proven = True + result.status = "proven" + result.template_name = wf.name + if install: + try: + installed = install_bundle(bundle_path, target_dir=install_dir) + result.installed_path = str(installed) + _say(f"installed proven template -> {installed}") + except Exception as exc: # noqa: BLE001 - install is best-effort + logger.warning("Architect install failed: %s", exc) + break + + # --- 6. refine feedback for the next pass --------------------------- + if hard_failures: + feedback = ( + "The workflow ran but failed hard checks: " + + "; ".join(hard_failures) + + ". Fix the workflow so the run completes with useful output." + ) + else: + feedback = ( + f"The run completed but the output only scored {score:.2f} " + f"(threshold {threshold:.2f}) against the request. Improve the " + "prompts so the output better satisfies: " + description + ) + log.refinement_note = log.refinement_note or judge_note + + if spent >= budget: + result.status = "budget_exceeded" + result.error = ( + f"budget ${budget:.2f} exhausted after iteration {i} (spent ${spent:.4f})" + ) + break + else: + result.status = "max_iterations" + result.error = ( + f"no iteration reached the {threshold:.2f} threshold within " + f"{iterations} iteration(s); returning the best attempt" + ) + + if result.status in ("max_iterations", "budget_exceeded") and best_yaml: + result.yaml_content = best_yaml + return result From 6cf5ab4e1a4137b026eb89d417a132bd65502c51 Mon Sep 17 00:00:00 2001 From: Tomas Pflanzer Date: Wed, 10 Jun 2026 14:19:45 +0200 Subject: [PATCH 2/3] feat(architect): async API job endpoints and 'sandcastle architect' CLI POST /api/architect starts an admin-gated async job (in-memory job store, same pattern as batch runs); GET /api/architect/{job_id} returns status, the per-iteration progress log, and the final result with bundle path. CLI: sandcastle architect "" [--input k=v] [--budget] [--max-iterations] [--threshold] runs the loop synchronously, prints iteration progress, exits 0 with the bundle path on success and 2 when the loop ends unproven. Co-Authored-By: Claude Fable 5 --- src/sandcastle/__main__.py | 110 ++++++++++++++++++++++++++++++++++ src/sandcastle/api/routes.py | 108 +++++++++++++++++++++++++++++++++ src/sandcastle/api/schemas.py | 27 +++++++++ 3 files changed, 245 insertions(+) diff --git a/src/sandcastle/__main__.py b/src/sandcastle/__main__.py index b588b445..f31439e1 100644 --- a/src/sandcastle/__main__.py +++ b/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, diff --git a/src/sandcastle/api/routes.py b/src/sandcastle/api/routes.py index a5f34eec..a60c1f39 100644 --- a/src/sandcastle/api/routes.py +++ b/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.""" diff --git a/src/sandcastle/api/schemas.py b/src/sandcastle/api/schemas.py index e9931686..0a6f5725 100644 --- a/src/sandcastle/api/schemas.py +++ b/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.""" From 5a41c221fc11e8cf9e5961d5f205d2b7a16a37a3 Mon Sep 17 00:00:00 2001 From: Tomas Pflanzer Date: Wed, 10 Jun 2026 14:19:45 +0200 Subject: [PATCH 3/3] test(architect): loop, budget, refine, API job lifecycle, CLI; README section Tests mock the LLM surfaces (generator + judge) but run the real executor against a mocked provider runtime, so the recorded cassette and packed bundle are genuine - the success test proves the bundle with verify_bundle. Covers: first-iteration success, refine-then-succeed, validation-error feedback, budget exceeded aborts cleanly, max iterations returns best-effort proven=false, hard-check failures skip the judge, test-input derivation, config defaults, API job lifecycle (start/poll/404/failure), CLI exit codes. README: 'The Architect' subsection under Verified Templates. Co-Authored-By: Claude Fable 5 --- README.md | 15 ++ tests/test_architect.py | 465 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 480 insertions(+) create mode 100644 tests/test_architect.py diff --git a/README.md b/README.md index 7bafa63d..0f90116b 100644 --- a/README.md +++ b/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 diff --git a/tests/test_architect.py b/tests/test_architect.py new file mode 100644 index 00000000..9ca8863e --- /dev/null +++ b/tests/test_architect.py @@ -0,0 +1,465 @@ +"""Tests for The Architect - the autonomous generate -> run -> evaluate -> refine loop. + +The LLM surfaces (generator, judge) are mocked; the runs are real executor runs +against a mocked provider runtime, so the recorded cassette and the packed +.sctpl bundle are genuine - the success test proves the bundle with the same +``verify_bundle`` that gates template installs. +""" + +from __future__ import annotations + +import asyncio +import uuid +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi.testclient import TestClient + +from sandcastle.engine.architect import ( + ArchitectResult, + _derive_test_input, + design_workflow, +) +from sandcastle.engine.bundle import verify_bundle +from sandcastle.engine.dag import parse_yaml_string +from sandcastle.engine.generator import GenerateResult +from sandcastle.engine.sandshore import SandshoreResult, SandshoreRuntime +from sandcastle.main import app + +client = TestClient(app) + +GEN_PATCH = "sandcastle.engine.generator.generate_workflow" +JUDGE_PATCH = "sandcastle.engine.architect._judge_output" +RUNTIME_PATCH = "sandcastle.engine.executor.get_sandshore_runtime" + + +def _wf(marker: str, name: str = "architect-test") -> str: + """A single standard-step workflow; *marker* keeps step cache keys unique.""" + return f"""name: {name} +description: Summarize the given text. +default_model: sonnet +input_schema: + required: [] + properties: + q: {{ type: string, default: "hi" }} +steps: + - id: summarize + prompt: "Summarize ({marker}): {{input.q}}" +""" + + +def _gen_result(yaml_text: str, validation_errors: list[str] | None = None) -> GenerateResult: + """Build a GenerateResult the way the real generator would for valid YAML.""" + if validation_errors: + return GenerateResult(yaml_content=yaml_text, validation_errors=validation_errors) + wf = parse_yaml_string(yaml_text) + return GenerateResult( + yaml_content=yaml_text, + name=wf.name, + description=wf.description, + steps_count=len(wf.steps), + validation_errors=[], + input_schema=wf.input_schema, + ) + + +def _sandbox(text: str, cost: float = 0.02) -> MagicMock: + """A mock SandshoreRuntime whose query returns a fixed result.""" + sb = MagicMock(spec=SandshoreRuntime) + + async def _query(request): + return SandshoreResult( + text=text, structured_output=None, total_cost_usd=cost, + input_tokens=5, output_tokens=5, + ) + + sb.query = _query + return sb + + +def _unique_input() -> dict: + """Unique test input so the shared DB step cache never short-circuits a run.""" + return {"q": f"hi-{uuid.uuid4().hex}"} + + +def _design(description: str = "Summarize text", **kwargs) -> ArchitectResult: + return asyncio.run(design_workflow(description, **kwargs)) + + +# --------------------------------------------------------------------------- +# the loop +# --------------------------------------------------------------------------- + + +def test_success_on_first_iteration_produces_verified_bundle(tmp_path): + """One good generation -> run -> judge pass yields a proven, installed bundle + whose proof passes the same verify_bundle used by template install.""" + yaml_text = _wf(uuid.uuid4().hex) + gen = AsyncMock(return_value=_gen_result(yaml_text)) + judge = AsyncMock(return_value=(0.9, "judge scored 0.90")) + + with patch(GEN_PATCH, gen), patch(JUDGE_PATCH, judge), \ + patch(RUNTIME_PATCH, return_value=_sandbox("A FINE SUMMARY")): + result = _design( + test_input=_unique_input(), + output_dir=tmp_path / "out", + install_dir=tmp_path / "installed", + ) + + assert result.status == "proven" + assert result.proven is True + assert len(result.iterations) == 1 + it = result.iterations[0] + assert it.run_status == "completed" + assert it.judge_score == 0.9 + assert it.hard_check_failures == [] + assert result.total_cost_usd == pytest.approx(0.02) + gen.assert_awaited_once() + + # The bundle is real and passes the install-gating verification. + assert result.bundle_path and Path(result.bundle_path).exists() + verify = verify_bundle(result.bundle_path) + assert verify.ok, f"{verify.errors} {[c.detail for c in verify.cassette_results]}" + assert verify.manifest["author"] == "the-architect" + + # Installed like `template install`: workflow + proof cassette side by side. + assert result.installed_path and Path(result.installed_path).exists() + assert Path(result.installed_path).read_text() == yaml_text + assert list((tmp_path / "installed").glob("*.cassette.json")) + + +def test_refine_then_succeed(tmp_path): + """A low judge score feeds back into the generator's refine path; the + refined workflow passes on the second iteration.""" + yaml_v1 = _wf(uuid.uuid4().hex) + yaml_v2 = _wf(uuid.uuid4().hex) + gen = AsyncMock(side_effect=[_gen_result(yaml_v1), _gen_result(yaml_v2)]) + judge = AsyncMock(side_effect=[(0.3, "weak"), (0.95, "strong")]) + + with patch(GEN_PATCH, gen), patch(JUDGE_PATCH, judge), \ + patch(RUNTIME_PATCH, return_value=_sandbox("BETTER SUMMARY")): + result = _design( + test_input=_unique_input(), + output_dir=tmp_path / "out", + install=False, + ) + + assert result.status == "proven" + assert result.proven is True + assert len(result.iterations) == 2 + assert result.iterations[0].judge_score == 0.3 + assert result.iterations[1].judge_score == 0.95 + assert result.yaml_content == yaml_v2 + + # Second generation went through refine_from/refine_instruction. + assert gen.await_count == 2 + refine_kwargs = gen.await_args_list[1].kwargs + assert refine_kwargs["refine_from"] == yaml_v1 + assert "0.30" in refine_kwargs["refine_instruction"] + # The judge feedback for the next pass is recorded in the iteration log. + assert result.iterations[1].refinement_note + + +def test_validation_errors_consume_an_iteration_and_feed_back(tmp_path): + """A generation with validation errors is never run; the errors become the + refine instruction for the next iteration.""" + good_yaml = _wf(uuid.uuid4().hex) + gen = AsyncMock(side_effect=[ + _gen_result("name: broken", validation_errors=["step list is empty"]), + _gen_result(good_yaml), + ]) + judge = AsyncMock(return_value=(0.9, "ok")) + + with patch(GEN_PATCH, gen), patch(JUDGE_PATCH, judge), \ + patch(RUNTIME_PATCH, return_value=_sandbox("OUT")): + result = _design( + test_input=_unique_input(), + output_dir=tmp_path / "out", + install=False, + ) + + assert result.proven is True + assert len(result.iterations) == 2 + assert result.iterations[0].validation_errors == ["step list is empty"] + assert result.iterations[0].run_status == "skipped (validation errors)" + assert "step list is empty" in gen.await_args_list[1].kwargs["refine_instruction"] + + +def test_budget_exceeded_aborts_cleanly(tmp_path): + """When live-run spend reaches the budget the loop stops with + budget_exceeded instead of burning further iterations.""" + gen = AsyncMock(side_effect=lambda *a, **kw: _gen_result(_wf(uuid.uuid4().hex))) + judge = AsyncMock(return_value=(0.1, "bad")) # never passes the threshold + + with patch(GEN_PATCH, gen), patch(JUDGE_PATCH, judge), \ + patch(RUNTIME_PATCH, return_value=_sandbox("OUT", cost=0.02)): + result = _design( + test_input=_unique_input(), + budget_usd=0.03, + max_iterations=5, + output_dir=tmp_path / "out", + install=False, + ) + + assert result.status == "budget_exceeded" + assert result.proven is False + assert "budget" in (result.error or "") + assert len(result.iterations) < 5 + assert result.total_cost_usd <= 0.03 + 0.02 # never overshoots by more than one run + + +def test_max_iterations_exhausted_returns_best_effort_unproven(tmp_path): + """Exhausting the loop returns proven=False with the best-scoring YAML.""" + yaml_v1 = _wf(uuid.uuid4().hex) + yaml_v2 = _wf(uuid.uuid4().hex) + gen = AsyncMock(side_effect=[_gen_result(yaml_v1), _gen_result(yaml_v2)]) + judge = AsyncMock(side_effect=[(0.4, "meh"), (0.2, "worse")]) + + with patch(GEN_PATCH, gen), patch(JUDGE_PATCH, judge), \ + patch(RUNTIME_PATCH, return_value=_sandbox("OUT")): + result = _design( + test_input=_unique_input(), + max_iterations=2, + output_dir=tmp_path / "out", + install=False, + ) + + assert result.status == "max_iterations" + assert result.proven is False + assert result.bundle_path is None + assert len(result.iterations) == 2 + assert result.best_score == 0.4 + assert result.yaml_content == yaml_v1 # the best attempt, not the last one + + +def test_failed_run_counts_as_hard_check_failure(tmp_path): + """A run that crashes/fails never reaches the judge and feeds the failure back.""" + gen = AsyncMock(side_effect=lambda *a, **kw: _gen_result(_wf(uuid.uuid4().hex))) + judge = AsyncMock(return_value=(0.9, "ok")) + sb = MagicMock(spec=SandshoreRuntime) + + async def _boom(request): + raise RuntimeError("provider exploded") + + sb.query = _boom + with patch(GEN_PATCH, gen), patch(JUDGE_PATCH, judge), \ + patch(RUNTIME_PATCH, return_value=sb): + result = _design( + test_input=_unique_input(), + max_iterations=2, + output_dir=tmp_path / "out", + install=False, + ) + + assert result.proven is False + assert result.status == "max_iterations" + judge.assert_not_awaited() + assert all(it.hard_check_failures for it in result.iterations) + + +# --------------------------------------------------------------------------- +# test input derivation +# --------------------------------------------------------------------------- + + +def test_derive_test_input_uses_schema_defaults_without_llm(): + schema = {"properties": {"q": {"type": "string", "default": "hello"}}} + with patch( + "sandcastle.engine.generator._call_advisor_llm", + AsyncMock(side_effect=AssertionError("LLM must not be called")), + ): + out = asyncio.run(_derive_test_input("anything", schema)) + assert out == {"q": "hello"} + + +def test_derive_test_input_falls_back_to_placeholders_when_llm_fails(): + schema = {"properties": {"text": {"type": "string"}, "count": {"type": "integer"}}} + with patch( + "sandcastle.engine.generator._call_advisor_llm", + AsyncMock(side_effect=RuntimeError("offline")), + ): + out = asyncio.run(_derive_test_input("anything", schema)) + assert out == {"text": "example", "count": 1} + + +def test_derive_test_input_uses_llm_for_missing_values(): + schema = {"properties": {"city": {"type": "string"}}} + with patch( + "sandcastle.engine.generator._call_advisor_llm", + AsyncMock(return_value='{"city": "Prague"}'), + ): + out = asyncio.run(_derive_test_input("weather report", schema)) + assert out == {"city": "Prague"} + + +# --------------------------------------------------------------------------- +# config +# --------------------------------------------------------------------------- + + +def test_architect_settings_defaults(): + from sandcastle.config import settings + + assert settings.architect_max_iterations == 3 + assert settings.architect_budget_usd == 1.0 + assert settings.architect_score_threshold == 0.7 + + +# --------------------------------------------------------------------------- +# API job lifecycle +# --------------------------------------------------------------------------- + + +def _fake_result() -> ArchitectResult: + return ArchitectResult( + status="proven", proven=True, description="d", + bundle_path="/tmp/x.sctpl", template_name="x", + ) + + +class TestArchitectApi: + def test_post_requires_api_key(self): + from sandcastle.config import settings + + with patch.object(settings, "anthropic_api_key", ""), \ + patch.dict("os.environ", {"ANTHROPIC_API_KEY": ""}): + resp = client.post("/api/architect", json={"description": "Build me a thing"}) + assert resp.status_code == 400 + assert resp.json()["detail"]["error"]["code"] == "MISSING_API_KEY" + + def test_post_starts_job_and_get_returns_it(self): + from sandcastle.config import settings + + with patch.object(settings, "anthropic_api_key", "sk-test"), \ + patch( + "sandcastle.engine.architect.design_workflow", + AsyncMock(return_value=_fake_result()), + ): + resp = client.post("/api/architect", json={"description": "Summarize text"}) + assert resp.status_code == 200 + job_id = resp.json()["data"]["job_id"] + assert resp.json()["data"]["status"] == "queued" + + poll = client.get(f"/api/architect/{job_id}") + assert poll.status_code == 200 + job = poll.json()["data"] + assert job["job_id"] == job_id + assert job["status"] in ("queued", "running", "completed") + assert job["description"] == "Summarize text" + assert "log" in job and "result" in job + + def test_get_unknown_job_404(self): + resp = client.get(f"/api/architect/{uuid.uuid4()}") + assert resp.status_code == 404 + + def test_job_runner_completes_and_stores_result(self): + """The job coroutine transitions queued -> completed with the result dict.""" + from sandcastle.api.routes import _architect_jobs, _run_architect_job + from sandcastle.api.schemas import ArchitectRequest + + job_id = f"test-{uuid.uuid4().hex}" + _architect_jobs[job_id] = { + "job_id": job_id, "status": "queued", "description": "d", + "created_at": "now", "completed_at": None, + "log": [], "result": None, "error": None, + } + try: + with patch( + "sandcastle.engine.architect.design_workflow", + AsyncMock(return_value=_fake_result()), + ): + asyncio.run(_run_architect_job(job_id, ArchitectRequest(description="d"))) + job = _architect_jobs[job_id] + assert job["status"] == "completed" + assert job["result"]["proven"] is True + assert job["result"]["bundle_path"] == "/tmp/x.sctpl" + assert job["completed_at"] + finally: + _architect_jobs.pop(job_id, None) + + def test_job_runner_records_failure(self): + from sandcastle.api.routes import _architect_jobs, _run_architect_job + from sandcastle.api.schemas import ArchitectRequest + + job_id = f"test-{uuid.uuid4().hex}" + _architect_jobs[job_id] = { + "job_id": job_id, "status": "queued", "description": "d", + "created_at": "now", "completed_at": None, + "log": [], "result": None, "error": None, + } + try: + with patch( + "sandcastle.engine.architect.design_workflow", + AsyncMock(side_effect=RuntimeError("kaboom")), + ): + asyncio.run(_run_architect_job(job_id, ArchitectRequest(description="d"))) + job = _architect_jobs[job_id] + assert job["status"] == "failed" + assert "kaboom" in job["error"] + finally: + _architect_jobs.pop(job_id, None) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +class TestArchitectCli: + def test_parser_accepts_architect_args(self): + from sandcastle.__main__ import _build_parser + + parser = _build_parser() + args = parser.parse_args( + ["architect", "Summarize my emails", "-i", "q=hi", + "--budget", "0.5", "--max-iterations", "2", "--threshold", "0.8"] + ) + assert args.command == "architect" + assert args.description == "Summarize my emails" + assert args.input == ["q=hi"] + assert args.budget == 0.5 + assert args.max_iterations == 2 + assert args.threshold == 0.8 + + def test_cli_exits_zero_with_bundle_path_on_success(self, capsys): + import argparse + + from sandcastle.__main__ import _cmd_architect + + args = argparse.Namespace( + description="Summarize text", input=None, input_file=None, + budget=None, max_iterations=None, threshold=None, + output=None, no_install=False, json=False, + ) + with patch( + "sandcastle.engine.architect.design_workflow", + AsyncMock(return_value=_fake_result()), + ): + _cmd_architect(args) # must not raise SystemExit + out = capsys.readouterr().out + assert "PROVEN" in out + assert "/tmp/x.sctpl" in out + + def test_cli_exits_2_when_not_proven(self, capsys): + import argparse + + from sandcastle.__main__ import _cmd_architect + + unproven = ArchitectResult( + status="max_iterations", proven=False, description="d", + error="no iteration reached the threshold", + ) + args = argparse.Namespace( + description="Summarize text", input=None, input_file=None, + budget=None, max_iterations=None, threshold=None, + output=None, no_install=False, json=False, + ) + with patch( + "sandcastle.engine.architect.design_workflow", + AsyncMock(return_value=unproven), + ): + with pytest.raises(SystemExit) as exc_info: + _cmd_architect(args) + assert exc_info.value.code == 2 + assert "NOT PROVEN" in capsys.readouterr().err