From cbf74ca12e8b59b318235ad19c4f6e97f25533a4 Mon Sep 17 00:00:00 2001 From: Romonaga Date: Sun, 14 Jun 2026 14:32:10 -0400 Subject: [PATCH] Generate PCB fabrication package preflight --- .verlyn/workflow_pack.json | 4 +- AGENTS.md | 2 +- Documentation/guides/VERLYN_AGENT_WORKFLOW.md | 15 +- backend/api/assembly_plans.py | 21 +++ backend/services/fabrication_export.py | 152 ++++++++++++++++++ tests/test_fabrication_export.py | 140 ++++++++++++++++ 6 files changed, 325 insertions(+), 9 deletions(-) create mode 100644 backend/services/fabrication_export.py create mode 100644 tests/test_fabrication_export.py diff --git a/.verlyn/workflow_pack.json b/.verlyn/workflow_pack.json index 7b329ff..4d4a442 100644 --- a/.verlyn/workflow_pack.json +++ b/.verlyn/workflow_pack.json @@ -1,7 +1,7 @@ { "product": "Verlyn", - "workflow_pack_version": "2026.06.14.1", + "workflow_pack_version": "2026.06.14.2", "repo_name": "CircuitShelf", "install_mode": "governance", - "installed_at": "2026-06-14T14:57:44.085895+00:00" + "installed_at": "2026-06-14T18:03:14.283913+00:00" } diff --git a/AGENTS.md b/AGENTS.md index b53edaf..892f56a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -59,7 +59,7 @@ After session compaction, summary recovery, or any other context-compressed resu - Once a Verlyn change is active, use the installed public `verlyn` CLI path first for branch repair, PR delivery, merge, and workflow state changes. If the product path is missing or blocked, record the blocker as Verlyn workflow feedback instead of bypassing it. - When a workflow mutation or delivery action exists in Verlyn, use the installed public CLI path before shell fallbacks such as `gh`. - For hosted PR closeout, prefer the Verlyn hosted delivery path from a repo-visible owner session. If the current session cannot see the repo, treat that as a scope/access gap to fix or switch sessions for, not as permission to bypass Verlyn's hosted workflow or fall back to `gh`. -- Use `verlyn changes deliver ` for full hosted closeout when a change is ready to land. +- Use `verlyn changes deliver ` for hosted source-control closeout when a change is ready to land; use `verlyn changes deploy ` afterward only when the operator wants deployment. - Keep change, work-item, review, and handoff records current while work is in progress. - Treat workflow friction as product feedback: if the process is confusing or blocked, record it as a change or work item instead of bypassing it. - Use subagents only for narrow, bounded, low-coupling parallel work. Avoid delegating tightly coupled workflow logic, central orchestration, or state-heavy UI changes unless there is a strong reason the speedup outweighs reintegration cost. diff --git a/Documentation/guides/VERLYN_AGENT_WORKFLOW.md b/Documentation/guides/VERLYN_AGENT_WORKFLOW.md index 12f5505..d02a822 100644 --- a/Documentation/guides/VERLYN_AGENT_WORKFLOW.md +++ b/Documentation/guides/VERLYN_AGENT_WORKFLOW.md @@ -123,16 +123,19 @@ Do not create ad hoc local workflow files as durable truth. 6. Do the implementation and verification. 7. Update work items, review notes, and risks through Verlyn. 8. When the change is ready to land, use `verlyn changes deliver --merge-method squash`. +9. When the delivered change should roll out to a configured deployment provider, use `verlyn changes deploy `. -`verlyn changes deliver` is the normal hosted closeout command. It commits local +`verlyn changes deliver` is the normal hosted source-control closeout command. It commits local dirty work when `--commit-message` is supplied, pushes with Verlyn-managed provider credentials, opens or updates the PR, merges it, switches the local checkout back to the delivery base branch when local checkout context exists, -and records closeout. It does not deploy by default; pass `--deploy` only when -the operator explicitly wants the change closeout to also trigger or monitor a -configured deployment provider. Credential issuance is gated by Verlyn repo write -access, release operations entitlement, exact client-remote matching, and -redacted audit recording; the CLI must not print provider tokens in JSON output. +and records closeout. It does not deploy. `verlyn changes deploy ` +is the normal change-aware deployment command; it resolves the delivered merge +point from Verlyn, triggers or monitors the configured provider, records +deployment evidence, and keeps provider credentials server-side. Credential +issuance is gated by Verlyn repo write access, release operations entitlement, +exact client-remote matching, and redacted audit recording; the CLI must not +print provider tokens in JSON output. If local checkout cleanup is blocked or unsafe, the hosted merge remains complete and the CLI reports the blocker, `repair_status`, `unsafe`, diff --git a/backend/api/assembly_plans.py b/backend/api/assembly_plans.py index 19e853b..be1e88c 100644 --- a/backend/api/assembly_plans.py +++ b/backend/api/assembly_plans.py @@ -17,6 +17,7 @@ from backend.services.circuit_graph import build_circuit_graph from backend.services.circuit_graph_ai import CircuitGraphAiEnrichmentService from backend.services.conversation_bench_plan_service import ConversationBenchPlanService +from backend.services.fabrication_export import build_fabrication_package from backend.services.ingestion_ai_review_service import estimate_local_tokens from backend.services.kicad_export import build_kicad_project_package from backend.services.openai_assist_utils import parse_json_object @@ -294,6 +295,26 @@ async def assembly_plan_kicad_project(plan_id: str, req: Request): ) return {"package": package} + @router.get("/api/assembly-plans/{plan_id}/fabrication-package") + async def assembly_plan_fabrication_package(plan_id: str, req: Request): + user, error = deps.require_authenticated_user(req) + if error: + return error + plan = assembly_plan_store.get(plan_id, deps.user_id_for_user(user)) + if not plan: + return JSONResponse({"error": "Assembly plan not found."}, status_code=404) + graph = build_circuit_graph(plan) + package = build_fabrication_package(plan, graph) + if not package.get("generated"): + return JSONResponse( + { + "error": "Fabrication package is not ready.", + "package": package, + }, + status_code=422, + ) + return {"package": package} + @router.get("/api/assembly-plans/{plan_id}/learning") async def assembly_learning_get(plan_id: str, req: Request): user, error = deps.require_authenticated_user(req) diff --git a/backend/services/fabrication_export.py b/backend/services/fabrication_export.py new file mode 100644 index 0000000..9e0c439 --- /dev/null +++ b/backend/services/fabrication_export.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import base64 +import io +import os +import shutil +import subprocess +import tempfile +import zipfile +from pathlib import Path +from typing import Any, Callable + +from backend.services.kicad_export import build_kicad_project_package, zip_files_base64 + + +Runner = Callable[[list[str], Path, Path], None] + + +def build_fabrication_package( + plan: dict[str, Any], + graph: dict[str, Any], + *, + kicad_cli_path: str | None = None, + runner: Runner | None = None, +) -> dict[str, Any]: + kicad_package = build_kicad_project_package(plan, graph) + project_name = kicad_package["projectName"] + manifest = { + "schemaVersion": 1, + "projectName": project_name, + "sourcePlanId": graph.get("planId") or plan.get("id"), + "status": "pending", + "files": [], + "checks": [], + } + if not kicad_package.get("exportable"): + manifest["status"] = "blocked" + manifest["checks"].append({"status": "blocking", "code": "kicad_project_not_exportable", "message": "KiCad project package is not exportable."}) + return package_result(False, manifest, [], kicad_package) + + pcb_file = find_file(kicad_package.get("files") or [], ".kicad_pcb") + if not pcb_file: + manifest["status"] = "needs_layout" + manifest["checks"].append({"status": "blocking", "code": "pcb_layout_missing", "message": "No KiCad PCB layout file is available yet."}) + manifest["files"].append({"path": f"{project_name}-kicad-project.zip", "kind": "kicad_project", "bytes": len(kicad_package.get("zipBase64") or "")}) + return package_result(False, manifest, [], kicad_package) + + cli = kicad_cli_path or shutil.which("kicad-cli") + if not cli: + manifest["status"] = "tool_unavailable" + manifest["checks"].append({"status": "blocking", "code": "kicad_cli_missing", "message": "kicad-cli is required to export Gerber and drill files."}) + manifest["files"].append({"path": f"{project_name}-kicad-project.zip", "kind": "kicad_project", "bytes": len(kicad_package.get("zipBase64") or "")}) + return package_result(False, manifest, [], kicad_package) + + with tempfile.TemporaryDirectory(prefix="circuitshelf-fab-") as tmp: + root = Path(tmp) + project_dir = root / project_name + output_dir = root / "fabrication" + project_dir.mkdir() + output_dir.mkdir() + write_kicad_files(project_dir, kicad_package.get("files") or []) + pcb_path = project_dir / pcb_file["path"] + run = runner or run_kicad_cli + try: + run([cli, "pcb", "export", "gerbers", "--output", str(output_dir), str(pcb_path)], project_dir, output_dir) + run([cli, "pcb", "export", "drill", "--output", str(output_dir), str(pcb_path)], project_dir, output_dir) + except Exception as exc: + manifest["status"] = "failed" + manifest["checks"].append({"status": "blocking", "code": "kicad_cli_failed", "message": str(exc)[:500]}) + manifest["files"].append({"path": f"{project_name}-kicad-project.zip", "kind": "kicad_project", "bytes": len(kicad_package.get("zipBase64") or "")}) + return package_result(False, manifest, [], kicad_package) + + fabrication_files = collect_output_files(output_dir) + if not fabrication_files: + manifest["status"] = "failed" + manifest["checks"].append({"status": "blocking", "code": "fabrication_files_missing", "message": "KiCad export completed without Gerber or drill files."}) + return package_result(False, manifest, [], kicad_package) + + manifest["status"] = "generated" + manifest["checks"].append({"status": "pass", "code": "fabrication_files_generated", "message": "Gerber/drill output files were generated."}) + manifest["files"] = [ + {"path": item["path"], "kind": item["kind"], "bytes": len(base64.b64decode(item["base64"]))} + for item in fabrication_files + ] + manifest["files"].append({"path": f"{project_name}-kicad-project.zip", "kind": "kicad_project", "bytes": len(kicad_package.get("zipBase64") or "")}) + return package_result(True, manifest, fabrication_files, kicad_package) + + +def package_result(generated: bool, manifest: dict[str, Any], files: list[dict[str, str]], kicad_package: dict[str, Any]) -> dict[str, Any]: + package_files = [ + {"path": item["path"], "mimeType": item["mimeType"], "content": item["content"]} + for item in kicad_package.get("files") or [] + ] + package_files.append({"path": "fabrication-manifest.json", "mimeType": "application/json", "content": _json_manifest(manifest)}) + for item in files: + package_files.append({"path": item["path"], "mimeType": item["mimeType"], "content": base64.b64decode(item["base64"]).decode("latin1")}) + return { + "generated": generated, + "status": manifest["status"], + "manifest": manifest, + "kicadProject": kicad_package, + "files": files, + "zipBase64": zip_files_base64(package_files), + } + + +def find_file(files: list[dict[str, Any]], suffix: str) -> dict[str, Any] | None: + for item in files: + if str(item.get("path") or "").endswith(suffix): + return item + return None + + +def write_kicad_files(project_dir: Path, files: list[dict[str, Any]]) -> None: + for item in files: + path = project_dir / str(item.get("path") or "") + if not path.name: + continue + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(str(item.get("content") or ""), encoding="utf-8") + + +def run_kicad_cli(command: list[str], _project_dir: Path, _output_dir: Path) -> None: + completed = subprocess.run(command, check=False, capture_output=True, text=True, timeout=120) + if completed.returncode: + message = completed.stderr.strip() or completed.stdout.strip() or f"kicad-cli exited with {completed.returncode}" + raise RuntimeError(message) + + +def collect_output_files(output_dir: Path) -> list[dict[str, str]]: + files = [] + for path in sorted(output_dir.rglob("*")): + if not path.is_file(): + continue + suffix = path.suffix.lower() + kind = "drill" if suffix in {".drl", ".xln"} else "gerber" if suffix.startswith(".g") else "fabrication" + data = path.read_bytes() + files.append( + { + "path": path.relative_to(output_dir).as_posix(), + "kind": kind, + "mimeType": "application/octet-stream", + "base64": base64.b64encode(data).decode("ascii"), + } + ) + return files + + +def _json_manifest(manifest: dict[str, Any]) -> str: + import json + + return json.dumps(manifest, indent=2, sort_keys=True) diff --git a/tests/test_fabrication_export.py b/tests/test_fabrication_export.py new file mode 100644 index 0000000..7cc3ed1 --- /dev/null +++ b/tests/test_fabrication_export.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +import backend.services.fabrication_export as fabrication_export +from backend.api.assembly_plans import create_router +from backend.api.dependencies import ApiDependencies +from backend.services.circuit_graph import build_circuit_graph +from backend.services.fabrication_export import build_fabrication_package + + +def exportable_plan() -> dict: + return { + "id": "plan-1", + "title": "NE555 Rail Check", + "objective": "Wire 555 power rails.", + "componentName": "NE555", + "componentType": "timer", + "summary": "Wire power and ground.", + "confidence": 0.84, + "status": "active", + "parts": [{"id": "part-1", "name": "NE555 timer IC", "detail": "DIP"}], + "power": [{"id": "power-1", "note": "Use regulated 5V and common ground."}], + "steps": [ + {"id": "step-1", "ordinal": 1, "type": "wiring", "title": "Pin 1 GND", "instruction": "Ground rail", "note": "Ground"}, + {"id": "step-2", "ordinal": 2, "type": "wiring", "title": "Pin 8 VCC", "instruction": "5V positive rail", "note": "Power"}, + ], + "sources": [], + "notes": [], + } + + +def fake_kicad_package(): + return { + "exportable": True, + "projectName": "fake-project", + "manifest": {"projectName": "fake-project"}, + "files": [ + {"path": "fake-project.kicad_pro", "mimeType": "application/json", "content": "{}"}, + {"path": "fake-project.kicad_pcb", "mimeType": "text/plain", "content": "(kicad_pcb)"}, + ], + "zipBase64": "ZmFrZQ==", + "validation": {"blocking": [], "warnings": []}, + } + + +def test_fabrication_package_needs_layout_when_project_has_no_pcb(): + plan = exportable_plan() + graph = build_circuit_graph(plan) + + package = build_fabrication_package(plan, graph) + + assert package["generated"] is False + assert package["status"] == "needs_layout" + assert package["manifest"]["checks"][0]["code"] == "pcb_layout_missing" + + +def test_fabrication_package_runs_kicad_cli_and_collects_outputs(monkeypatch): + monkeypatch.setattr(fabrication_export, "build_kicad_project_package", lambda _plan, _graph: fake_kicad_package()) + + def runner(command: list[str], _project_dir: Path, output_dir: Path): + if "gerbers" in command: + (output_dir / "fake-F_Cu.gbr").write_text("G04 copper*", encoding="utf-8") + if "drill" in command: + (output_dir / "fake.drl").write_text("M48", encoding="utf-8") + + package = build_fabrication_package(exportable_plan(), build_circuit_graph(exportable_plan()), kicad_cli_path="/usr/bin/kicad-cli", runner=runner) + + assert package["generated"] is True + assert package["status"] == "generated" + assert {item["kind"] for item in package["files"]} == {"gerber", "drill"} + assert package["manifest"]["checks"][0]["code"] == "fabrication_files_generated" + + +def test_fabrication_package_reports_kicad_cli_failure(monkeypatch): + monkeypatch.setattr(fabrication_export, "build_kicad_project_package", lambda _plan, _graph: fake_kicad_package()) + + def runner(_command: list[str], _project_dir: Path, _output_dir: Path): + raise RuntimeError("bad board") + + package = build_fabrication_package(exportable_plan(), build_circuit_graph(exportable_plan()), kicad_cli_path="/usr/bin/kicad-cli", runner=runner) + + assert package["generated"] is False + assert package["status"] == "failed" + assert package["manifest"]["checks"][0]["code"] == "kicad_cli_failed" + + +def test_fabrication_endpoint_returns_preflight_package_for_owned_plan(): + plan = exportable_plan() + + class Store: + def get(self, plan_id: str, user_id: int | None = None): + assert user_id == 7 + return plan if plan_id == "plan-1" else None + + deps = ApiDependencies( + require_authenticated_user=lambda req: (SimpleNamespace(username="tester", id=7), None), + require_entity_member=lambda req: (None, None, None), + require_entity_admin=lambda req: (None, None, None), + require_system_admin_user=lambda req: (None, None), + bearer_token_from_request=lambda req: "", + session_timeout_seconds=lambda: 300, + user_payload=lambda user: {}, + user_id_for_user=lambda user: user.id, + verify_user=lambda username, password: None, + user_store=None, + user_preferences_store=None, + account_profile_store=None, + entity_store=None, + password_policy_store=None, + ai_provider_store=None, + ) + app = FastAPI() + app.include_router( + create_router( + deps, + assembly_plan_store=Store(), + conversation_store=None, + bench_tools=SimpleNamespace(build_assembly_export=lambda plan, format: {}), + openai_assist_service=None, + get_rag_response=lambda **kwargs: None, + query_ollama_chat_with_retry=lambda *args, **kwargs: None, + normalize_sources_for_api=lambda sources: sources, + build_recovery_prompt=lambda question, answer, sources: "", + parse_recovered_build_card=lambda raw, sources: None, + recovery_system_prompt="", + default_model="local", + username_for_user=lambda user: user.username, + ) + ) + + response = TestClient(app).get("/api/assembly-plans/plan-1/fabrication-package") + + assert response.status_code == 422 + package = response.json()["package"] + assert package["status"] == "needs_layout"