Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .verlyn/workflow_pack.json
Original file line number Diff line number Diff line change
@@ -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"
}
2 changes: 1 addition & 1 deletion AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 <change-id>` for full hosted closeout when a change is ready to land.
- Use `verlyn changes deliver <change-id>` for hosted source-control closeout when a change is ready to land; use `verlyn changes deploy <change-id>` 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.
Expand Down
15 changes: 9 additions & 6 deletions Documentation/guides/VERLYN_AGENT_WORKFLOW.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 <change-id> --merge-method squash`.
9. When the delivered change should roll out to a configured deployment provider, use `verlyn changes deploy <change-id>`.

`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 <change-id>`
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`,
Expand Down
21 changes: 21 additions & 0 deletions backend/api/assembly_plans.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down
152 changes: 152 additions & 0 deletions backend/services/fabrication_export.py
Original file line number Diff line number Diff line change
@@ -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)
140 changes: 140 additions & 0 deletions tests/test_fabrication_export.py
Original file line number Diff line number Diff line change
@@ -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"