"""Native SHiRE OS Forge with the complete Academy Safe Plan Fast Builder flow. Stage 4A keeps the existing PyVista viewport and trusted Forge workspace while making the Stage 3C.7 Academy API the single source of truth for Plain-English plans, immutable revisions, exact-hash generation, cancellation and evidence. """ from __future__ import annotations import html from pathlib import Path import traceback from typing import Any, Callable from qtpy.QtCore import Qt, QThread, QTimer, Signal from qtpy.QtGui import QFont from qtpy.QtWidgets import ( QApplication, QComboBox, QFrame, QGridLayout, QHBoxLayout, QLabel, QLineEdit, QListWidget, QPushButton, QScrollArea, QSizePolicy, QSplitter, QTextBrowser, QTextEdit, QVBoxLayout, QWidget, ) from pyvistaqt import QtInteractor from shireforge.forge_engine.build_service import build_calibration_cube from shireforge.forge_engine.network_export import export_validated_build from shireforge.forge_engine.safe_plan_client import ( DownloadedSafePlanArtifacts, PROMPT_LIMIT, SafePlanClient, ) from shireforge.forge_engine.safe_plan_export import export_safe_plan_build from shireforge.generators.calibration_cube import CubeParameters from shireforge.validation.mesh_validator import validate_stl from shireforge.viewer.mesh_loader import load_preview_mesh STAGE4A_VERSION = "0.4.0-stage4a-native-forge-fast-builder" GREEN = "#00ff72" PURPLE = "#c02cff" RED = "#ff315f" WHITE = "#eafff1" MUTED = "#9bb3a5" DARK = "#020406" PANEL = "#07100d" PANEL_ALT = "#100817" class TaskWorker(QThread): succeeded = Signal(object) failed = Signal(str) def __init__(self, operation: Callable[[], Any], parent: QWidget | None = None) -> None: super().__init__(parent) self.operation = operation def run(self) -> None: try: self.succeeded.emit(self.operation()) except Exception as exc: # GUI boundary: surface a safe message, retain traceback in logs. traceback.print_exc() self.failed.emit(str(exc)) class SafePlanCard(QFrame): generate_requested = Signal(object) rethink_requested = Signal(object) alternate_requested = Signal(object) cancel_plan_requested = Signal(object) cancel_build_requested = Signal(object, object) close_requested = Signal(object) def __init__(self, plan: dict[str, Any], parent: QWidget | None = None) -> None: super().__init__(parent) self.plan = plan self.job: dict[str, Any] | None = None self._collapsed = False self.setObjectName("safePlanCard") self.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Maximum) root = QVBoxLayout(self) root.setContentsMargins(12, 12, 12, 12) root.setSpacing(8) heading = QHBoxLayout() title_box = QVBoxLayout() self.title = QLabel() self.title.setObjectName("planTitle") self.identity = QLabel() self.identity.setObjectName("planIdentity") self.identity.setTextInteractionFlags(Qt.TextSelectableByMouse) title_box.addWidget(self.title) title_box.addWidget(self.identity) heading.addLayout(title_box, 1) self.status = QLabel() self.status.setObjectName("planStatus") heading.addWidget(self.status, 0, Qt.AlignTop) root.addLayout(heading) self.body = QFrame() self.body.setObjectName("planBody") body_layout = QVBoxLayout(self.body) body_layout.setContentsMargins(9, 9, 9, 9) body_layout.setSpacing(7) self.summary = QGridLayout() self.summary.setHorizontalSpacing(12) self.summary.setVerticalSpacing(5) body_layout.addLayout(self.summary) body_layout.addWidget(QLabel("FULL STORED PROMPT")) self.prompt = QTextEdit() self.prompt.setReadOnly(True) self.prompt.setMaximumHeight(92) self.prompt.setObjectName("storedPrompt") body_layout.addWidget(self.prompt) self.evidence = QLabel() self.evidence.setWordWrap(True) self.evidence.setTextInteractionFlags(Qt.TextSelectableByMouse) self.evidence.setObjectName("evidencePath") body_layout.addWidget(self.evidence) self.job_box = QLabel("No build started. No downloadable output is exposed until validation passes.") self.job_box.setWordWrap(True) self.job_box.setObjectName("jobBox") self.job_box.setTextInteractionFlags(Qt.TextSelectableByMouse) body_layout.addWidget(self.job_box) root.addWidget(self.body) buttons = QGridLayout() buttons.setHorizontalSpacing(6) buttons.setVerticalSpacing(6) self.generate_button = QPushButton("GENERATE REAL CADQUERY DRAFT") self.generate_button.setObjectName("primaryAction") self.rethink_button = QPushButton("RETHINK PLAN") self.alternate_button = QPushButton("TRY ANOTHER APPROACH") self.cancel_button = QPushButton("CANCEL PLAN") self.cancel_button.setObjectName("dangerAction") self.collapse_button = QPushButton("COLLAPSE") self.close_button = QPushButton("CLOSE PLAN") buttons.addWidget(self.generate_button, 0, 0, 1, 2) buttons.addWidget(self.rethink_button, 1, 0) buttons.addWidget(self.alternate_button, 1, 1) buttons.addWidget(self.cancel_button, 2, 0) buttons.addWidget(self.collapse_button, 2, 1) buttons.addWidget(self.close_button, 3, 0, 1, 2) root.addLayout(buttons) self.generate_button.clicked.connect(lambda: self.generate_requested.emit(self.plan)) self.rethink_button.clicked.connect(lambda: self.rethink_requested.emit(self.plan)) self.alternate_button.clicked.connect(lambda: self.alternate_requested.emit(self.plan)) self.cancel_button.clicked.connect(self._cancel_clicked) self.collapse_button.clicked.connect(self._toggle_collapse) self.close_button.clicked.connect(lambda: self.close_requested.emit(self)) self.update_plan(plan) @staticmethod def _friendly(value: Any) -> str: text = str(value if value not in (None, "") else "Not specified") return text.replace("_", " ").strip().title() if len(text) < 80 else text def _row(self, row: int, label: str, value: Any) -> None: name = QLabel(label.upper()) name.setObjectName("planFieldName") detail = QLabel(str(value if value not in (None, "") else "Not specified")) detail.setWordWrap(True) detail.setTextInteractionFlags(Qt.TextSelectableByMouse) detail.setObjectName("planFieldValue") self.summary.addWidget(name, row, 0, Qt.AlignTop) self.summary.addWidget(detail, row, 1) def update_plan(self, plan: dict[str, Any]) -> None: self.plan = plan spec = plan.get("spec") or {} parameters = spec.get("parameters") or {} project = spec.get("project_name") or spec.get("object_name") or "CadQuery Draft" self.title.setText(str(project)) self.identity.setText( f"Revision {plan.get('revision_id')} • Parent {plan.get('parent_revision_id') or 'None'}\n" f"Specification SHA-256: {plan.get('specification_hash')}" ) self.status.setText(str(plan.get("status") or "planned").upper()) self.prompt.setPlainText(str(plan.get("full_prompt") or "")) self.evidence.setText(f"SHiREVault evidence: {plan.get('evidence_path') or 'Not available'}") while self.summary.count(): item = self.summary.takeAt(0) widget = item.widget() if widget is not None: widget.deleteLater() fields = [ ("Family", self._friendly(spec.get("object_family") or "Unknown")), ("Mechanical contract", parameters.get("mechanical_contract") or "Not applicable"), ("Theme", self._friendly(spec.get("design_theme") or parameters.get("theme_name") or "Automatic")), ("Façade style", parameters.get("gothic_facade_style") or "Not applicable"), ("Buttresses", parameters.get("gothic_buttresses") or "Not applicable"), ("Lancet details", parameters.get("gothic_lancet_details") or "Not applicable"), ("Tracery", parameters.get("gothic_tracery") or "Not applicable"), ("Rose window", parameters.get("gothic_rose_window") or "Not applicable"), ("Crown detail", parameters.get("gothic_crown_detail") or "Not applicable"), ("Tray integration", parameters.get("gothic_tray_integration") or "Not applicable"), ("Protected path intrusion", plan.get("protected_path_intrusion") or parameters.get("protected_path_intrusion") or "Not applicable"), ] for row, (label, value) in enumerate(fields): self._row(row, label, value) self._refresh_actions() def update_job(self, job: dict[str, Any]) -> None: self.job = job phase = str(job.get("phase") or "unknown") message = str(job.get("message") or phase) self.job_box.setText( f"Build {job.get('job_id')} • {phase.upper()}\n" f"Exact spec hash: {job.get('specification_hash')}\n{message}" ) self._refresh_actions() def _refresh_actions(self) -> None: plan_status = str(self.plan.get("status") or "planned") phase = str((self.job or {}).get("phase") or "") building = phase == "building" or plan_status == "building" planned = plan_status == "planned" and not building self.generate_button.setEnabled(planned) self.rethink_button.setEnabled(planned) self.alternate_button.setEnabled(planned) self.cancel_button.setEnabled(planned or building) self.cancel_button.setText("CANCEL BUILD" if building else "CANCEL PLAN") self.status.setText((phase if phase else plan_status).upper()) def _cancel_clicked(self) -> None: if self.job and self.job.get("phase") == "building": self.cancel_build_requested.emit(self.plan, self.job) else: self.cancel_plan_requested.emit(self.plan) def _toggle_collapse(self) -> None: self._collapsed = not self._collapsed self.body.setVisible(not self._collapsed) self.collapse_button.setText("EXPAND" if self._collapsed else "COLLAPSE") class ForgePage(QWidget): """Product-building Forge embedded inside SHiRE OS.""" def __init__(self, *, embedded: bool = False, auto_build: bool = True) -> None: super().__init__() self.embedded = bool(embedded) self.auto_build = bool(auto_build) self.api = SafePlanClient() self.current_build = None self.current_plan: dict[str, Any] | None = None self.current_job: dict[str, Any] | None = None self.current_artifacts: DownloadedSafePlanArtifacts | None = None self.current_safe_validation = None self._workers: set[TaskWorker] = set() self._cards: dict[str, SafePlanCard] = {} self._jobs: dict[str, dict[str, Any]] = {} self._activated = False root = QVBoxLayout(self) root.setContentsMargins(10, 8, 10, 8) root.setSpacing(7) root.addLayout(self._header()) splitter = QSplitter(Qt.Horizontal) splitter.setChildrenCollapsible(False) splitter.addWidget(self._request_panel()) splitter.addWidget(self._viewport_panel()) splitter.addWidget(self._plans_panel()) splitter.setSizes([390, 850, 500]) root.addWidget(splitter, 1) if not self.embedded: nav = QHBoxLayout() for name in ("SHiRE", "FORGE", "ACADEMY", "SENTINEL", "STATIONS"): button = QPushButton(name) button.setEnabled(name == "FORGE") nav.addWidget(button) root.addLayout(nav) self.poll_timer = QTimer(self) self.poll_timer.setInterval(1200) self.poll_timer.timeout.connect(self._poll_jobs) self.poll_timer.start() self.setStyleSheet(self._stylesheet()) if self.auto_build: self.activate() def _header(self) -> QHBoxLayout: header = QHBoxLayout() title_box = QVBoxLayout() title = QLabel("FORGE" if self.embedded else "SHiRE • FORGE") title.setObjectName("title") subtitle = QLabel("NATIVE FAST BUILDER • IMMUTABLE SAFE PLANS • REAL CADQUERY") subtitle.setObjectName("subtitle") title_box.addWidget(title) title_box.addWidget(subtitle) header.addLayout(title_box) header.addStretch() self.academy_badge = QLabel("● ACADEMY CHECKING") self.academy_badge.setObjectName("academyBadge") self.status = QLabel("● LOCAL MODE") self.status.setObjectName("status") header.addWidget(self.academy_badge) header.addSpacing(12) header.addWidget(self.status) return header def _request_panel(self) -> QWidget: panel = QFrame() panel.setObjectName("majorPanel") box = QVBoxLayout(panel) box.setContentsMargins(12, 12, 12, 12) box.setSpacing(8) box.addWidget(self._section_label("FAST BUILDER REQUEST")) self.activity = QTextBrowser() self.activity.setObjectName("activityLog") self.activity.setMaximumHeight(150) self.activity.setHtml( f"SHiRE
" "Native Forge is linked to Academy Safe Plans. Describe the complete object; " "SHiRE will preserve the exact accepted prompt, revision and specification hash." ) box.addWidget(self.activity) self.project_name = QLineEdit() self.project_name.setPlaceholderText("Project name (optional)") self.project_name.setMaxLength(80) box.addWidget(self.project_name) self.prompt = QTextEdit() self.prompt.setAcceptRichText(False) self.prompt.setPlaceholderText( "Describe what you want to create. Include dimensions, purpose, material, style, " "openings, clearances, assembly needs and printer constraints." ) self.prompt.setMinimumHeight(205) self.prompt.textChanged.connect(self._update_counter) box.addWidget(self.prompt) counter_row = QHBoxLayout() self.prompt_message = QLabel("Line breaks are preserved. Nothing is silently truncated.") self.prompt_message.setObjectName("promptHint") self.counter = QLabel(f"0 / {PROMPT_LIMIT}") self.counter.setObjectName("counter") counter_row.addWidget(self.prompt_message, 1) counter_row.addWidget(self.counter) box.addLayout(counter_row) options = QGridLayout() options.addWidget(QLabel("PRINTER"), 0, 0) options.addWidget(QLabel("MATERIAL"), 0, 1) options.addWidget(QLabel("OUTPUT"), 2, 0) options.addWidget(QLabel("FAMILY"), 2, 1) self.printer = QComboBox() self.printer.addItem("Bambu Lab P1S", "bambu_p1s") self.printer.addItem("Creality K2 Plus", "creality_k2_plus") self.printer.addItem("Snapmaker U1", "snapmaker_u1") self.material = QComboBox() self.material.addItems(["PLA", "PETG", "TPU"]) self.output_format = QComboBox() self.output_format.addItem("STEP + STL", "both") self.output_format.addItem("STEP", "step") self.output_format.addItem("STL", "stl") self.object_family = QComboBox() self.object_family.addItem("Automatic detection", "auto") self.object_family.addItem("Dice Tower", "dice_tower") self.object_family.addItem("Desk Organiser", "desk_organiser") self.object_family.addItem("Parts Tray", "parts_tray") self.object_family.addItem("Mounting Block", "mounting_block") self.object_family.addItem("Nameplate", "nameplate") options.addWidget(self.printer, 1, 0) options.addWidget(self.material, 1, 1) options.addWidget(self.output_format, 3, 0) options.addWidget(self.object_family, 3, 1) box.addLayout(options) self.create_button = QPushButton("CREATE SAFE DRAFT PLAN") self.create_button.setObjectName("primaryAction") self.create_button.clicked.connect(self.create_safe_plan) box.addWidget(self.create_button) certified = QPushButton("BUILD CERTIFIED CALIBRATION CUBE") certified.setToolTip("Retains the original trusted Forge recovery tool.") certified.clicked.connect(self.build_certified_cube) box.addWidget(certified) box.addStretch() return panel def _viewport_panel(self) -> QWidget: panel = QFrame() panel.setObjectName("majorPanel") box = QVBoxLayout(panel) box.setContentsMargins(10, 10, 10, 10) controls = QHBoxLayout() controls.addWidget(self._section_label("REAL 3D PREVIEW")) controls.addStretch() fit_button = QPushButton("FIT MODEL") fit_button.clicked.connect(self.fit_model) reset_button = QPushButton("RESET CAMERA") reset_button.clicked.connect(self.reset_camera) controls.addWidget(fit_button) controls.addWidget(reset_button) box.addLayout(controls) self.viewer = QtInteractor(panel) self.viewer.set_background("#010302", top="#180820") box.addWidget(self.viewer.interactor, 1) self.stats = QLabel("No validated model loaded.") self.stats.setObjectName("modelStats") self.stats.setAlignment(Qt.AlignCenter) self.stats.setWordWrap(True) box.addWidget(self.stats) self.validation_list = QListWidget() self.validation_list.setMaximumHeight(150) self.validation_list.addItems([ "○ Safe Plan not created", "○ Exact specification hash not generated", "○ Model validation pending", "○ SHiREVault export requires Ray approval", ]) box.addWidget(self.validation_list) self.export_button = QPushButton("APPROVE COPY TO SHiREVAULT READY") self.export_button.setObjectName("approveAction") self.export_button.setEnabled(False) self.export_button.clicked.connect(self.export_current_model) box.addWidget(self.export_button) return panel def _plans_panel(self) -> QWidget: panel = QFrame() panel.setObjectName("majorPanel") box = QVBoxLayout(panel) box.setContentsMargins(10, 10, 10, 10) heading = QHBoxLayout() heading.addWidget(self._section_label("SAFE PLAN HISTORY")) heading.addStretch() self.plan_count = QLabel("0 REVISIONS") self.plan_count.setObjectName("planCount") heading.addWidget(self.plan_count) box.addLayout(heading) note = QLabel( "Rethink and Try Another Approach create new revisions. Older revisions remain visible and are never overwritten." ) note.setWordWrap(True) note.setObjectName("promptHint") box.addWidget(note) scroll = QScrollArea() scroll.setWidgetResizable(True) scroll.setFrameShape(QFrame.NoFrame) self.plan_host = QWidget() self.plan_layout = QVBoxLayout(self.plan_host) self.plan_layout.setContentsMargins(0, 0, 0, 0) self.plan_layout.setSpacing(9) self.empty_plan = QLabel("No Safe Draft Plan yet.\nCreate one from the Plain-English request panel.") self.empty_plan.setAlignment(Qt.AlignCenter) self.empty_plan.setWordWrap(True) self.empty_plan.setObjectName("emptyPlan") self.plan_layout.addWidget(self.empty_plan) self.plan_layout.addStretch() scroll.setWidget(self.plan_host) box.addWidget(scroll, 1) return panel @staticmethod def _section_label(text: str) -> QLabel: label = QLabel(text) label.setObjectName("sectionTitle") return label @staticmethod def _stylesheet() -> str: return f""" QWidget {{ background: {DARK}; color: {WHITE}; font-family: 'DejaVu Sans', Sans Serif; font-size: 13px; }} QFrame#majorPanel {{ background: {PANEL}; border: 1px solid rgba(0,255,114,120); border-radius: 14px; }} QFrame#safePlanCard {{ background: {PANEL_ALT}; border: 1px solid {PURPLE}; border-radius: 12px; }} QFrame#planBody {{ background: rgba(0,0,0,90); border: 1px solid rgba(0,255,114,65); border-radius: 8px; }} QLabel#title {{ color: {GREEN}; font-size: 25px; font-weight: 900; }} QLabel#subtitle {{ color: {PURPLE}; font-size: 10px; font-weight: 800; letter-spacing: 1px; }} QLabel#status, QLabel#academyBadge {{ color: {GREEN}; font-weight: 900; }} QLabel#sectionTitle {{ color: {PURPLE}; font-size: 13px; font-weight: 900; letter-spacing: 1px; }} QLabel#promptHint, QLabel#planIdentity, QLabel#evidencePath {{ color: {MUTED}; font-size: 10px; }} QLabel#counter {{ color: {GREEN}; font-family: monospace; font-weight: 900; }} QLabel#counter[overLimit='true'] {{ color: {RED}; }} QLabel#planTitle {{ color: {GREEN}; font-size: 16px; font-weight: 900; }} QLabel#planStatus {{ color: {PURPLE}; font-weight: 900; border: 1px solid {PURPLE}; border-radius: 7px; padding: 4px 7px; }} QLabel#planFieldName {{ color: {PURPLE}; font-size: 9px; font-weight: 900; }} QLabel#planFieldValue {{ color: {WHITE}; font-size: 10px; }} QLabel#jobBox {{ color: {GREEN}; background: rgba(0,255,114,20); border: 1px solid rgba(0,255,114,80); border-radius: 7px; padding: 7px; }} QLabel#modelStats {{ color: {WHITE}; background: rgba(192,44,255,25); border: 1px solid rgba(192,44,255,100); border-radius: 7px; padding: 7px; }} QLabel#emptyPlan {{ color: {MUTED}; padding: 28px; }} QLabel#planCount {{ color: {GREEN}; font-size: 10px; font-weight: 900; }} QPushButton {{ background: #0b1711; color: {WHITE}; border: 1px solid rgba(0,255,114,135); border-radius: 7px; padding: 7px 9px; font-weight: 800; }} QPushButton:hover {{ background: #13271c; border-color: {GREEN}; }} QPushButton:pressed {{ background: #1c0c25; border-color: {PURPLE}; }} QPushButton:disabled {{ color: #526158; border-color: #27342c; background: #050806; }} QPushButton#primaryAction {{ background: #10231a; color: {GREEN}; border: 2px solid {GREEN}; }} QPushButton#approveAction {{ background: #1c0b25; color: {WHITE}; border: 2px solid {PURPLE}; }} QPushButton#dangerAction {{ color: {RED}; border-color: {RED}; }} QLineEdit, QTextEdit, QTextBrowser, QListWidget, QComboBox {{ background: #010302; color: {WHITE}; border: 1px solid rgba(192,44,255,130); border-radius: 7px; padding: 6px; selection-background-color: {PURPLE}; }} QTextEdit#storedPrompt {{ font-family: monospace; font-size: 10px; }} QComboBox::drop-down {{ border: none; width: 18px; }} QScrollArea {{ background: transparent; border: none; }} QSplitter::handle {{ background: rgba(192,44,255,60); width: 4px; }} QScrollBar:vertical {{ background: #030604; width: 10px; margin: 0; }} QScrollBar::handle:vertical {{ background: {PURPLE}; min-height: 30px; border-radius: 5px; }} """ def activate(self) -> None: if self._activated: return self._activated = True self._run_task("Academy status", self.api.status, self._academy_ready) def _academy_ready(self, status: dict[str, Any]) -> None: limit = status.get("prompt_character_limit") contract = status.get("mechanical_contract") self.academy_badge.setText("● ACADEMY LINK READY") self.status.setText("● FAST BUILDER READY") self._append_activity( "SHiRE", f"Academy Safe Plans are ready. Prompt limit: {limit}. Dice Tower contract: {contract}.", GREEN, ) def _update_counter(self) -> None: count = len(self.prompt.toPlainText()) over = count > PROMPT_LIMIT self.counter.setText(f"{count} / {PROMPT_LIMIT}") self.counter.setProperty("overLimit", over) self.counter.style().unpolish(self.counter) self.counter.style().polish(self.counter) self.create_button.setEnabled(bool(self.prompt.toPlainText().strip()) and not over) if over: self.prompt_message.setText( f"Request exceeds the limit by {count - PROMPT_LIMIT} characters. Nothing was truncated." ) self.prompt_message.setStyleSheet(f"color:{RED};") else: self.prompt_message.setText("Line breaks are preserved. Nothing is silently truncated.") self.prompt_message.setStyleSheet("") def _append_activity(self, speaker: str, message: str, colour: str) -> None: self.activity.append( f"
{html.escape(speaker)}
{html.escape(message)}" ) def _run_task( self, label: str, operation: Callable[[], Any], success: Callable[[Any], None], failure: Callable[[str], None] | None = None, ) -> None: worker = TaskWorker(operation, self) self._workers.add(worker) self.status.setText(f"● {label.upper()}") def done(result: Any) -> None: try: success(result) finally: self._workers.discard(worker) worker.deleteLater() def failed(message: str) -> None: try: if failure is not None: failure(message) else: self._task_failed(label, message) finally: self._workers.discard(worker) worker.deleteLater() worker.succeeded.connect(done) worker.failed.connect(failed) worker.start() def _task_failed(self, label: str, message: str) -> None: self.status.setText("● SAFE STOP") self._append_activity("SAFE STOP", f"{label}: {message}", RED) def create_safe_plan(self) -> None: prompt = self.prompt.toPlainText() if not prompt.strip() or len(prompt) > PROMPT_LIMIT: self._update_counter() return self.create_button.setEnabled(False) self._append_activity("You", prompt, WHITE) operation = lambda: self.api.create_plan( prompt=prompt, project_name=self.project_name.text(), output_format=str(self.output_format.currentData()), object_family=str(self.object_family.currentData()), printer_profile=str(self.printer.currentData()), material=self.material.currentText(), ) def success(result: dict[str, Any]) -> None: self.create_button.setEnabled(True) plan = result["plan"] self._add_plan(plan) self.status.setText("● SAFE PLAN READY") self._append_activity( "SHiRE", f"Safe Plan {plan['revision_id']} created and frozen as {plan['specification_hash']}.", GREEN, ) def failure(message: str) -> None: self.create_button.setEnabled(True) self._task_failed("Create Safe Draft Plan", message) self._run_task("Creating Safe Plan", operation, success, failure) def _add_plan(self, plan: dict[str, Any]) -> SafePlanCard: revision_id = str(plan.get("revision_id")) existing = self._cards.get(revision_id) if existing is not None: existing.update_plan(plan) return existing self.empty_plan.hide() card = SafePlanCard(plan) card.generate_requested.connect(self.generate_plan) card.rethink_requested.connect(self.rethink_plan) card.alternate_requested.connect(self.alternate_plan) card.cancel_plan_requested.connect(self.cancel_plan) card.cancel_build_requested.connect(self.cancel_build) card.close_requested.connect(self.close_plan_card) self.plan_layout.insertWidget(max(0, self.plan_layout.count() - 1), card) self._cards[revision_id] = card self.plan_count.setText(f"{len(self._cards)} REVISIONS") self.current_plan = plan self.validation_list.item(0).setText("✓ Safe Plan revision stored with exact hash") return card def _card_for(self, plan: dict[str, Any]) -> SafePlanCard | None: return self._cards.get(str(plan.get("revision_id") or "")) def rethink_plan(self, plan: dict[str, Any]) -> None: self._run_task( "Rethinking Plan", lambda: self.api.rethink(plan), lambda result: self._revision_created(result, "Rethink created a new revision; the parent remains preserved."), ) def alternate_plan(self, plan: dict[str, Any]) -> None: self._run_task( "Trying Another Approach", lambda: self.api.alternate(plan), lambda result: self._revision_created(result, "A genuinely different valid plan was created; the previous revision remains preserved."), ) def _revision_created(self, result: dict[str, Any], message: str) -> None: new_plan = result["plan"] self._add_plan(new_plan) self.status.setText("● NEW REVISION READY") self._append_activity("SHiRE", message, GREEN) def cancel_plan(self, plan: dict[str, Any]) -> None: def success(result: dict[str, Any]) -> None: cancelled = result["plan"] card = self._card_for(cancelled) if card: card.update_plan(cancelled) self.status.setText("● PLAN CANCELLED") self._append_activity("SHiRE", f"Plan {cancelled['revision_id']} cancelled before generation.", RED) self._run_task("Cancelling Plan", lambda: self.api.cancel_plan(plan), success) def generate_plan(self, plan: dict[str, Any]) -> None: card = self._card_for(plan) if card: card.job_box.setText("Starting the exact displayed Safe Plan revision…") self.current_artifacts = None self.current_safe_validation = None self.export_button.setEnabled(False) def success(result: dict[str, Any]) -> None: job = result["job"] job_id = str(job["job_id"]) self._jobs[job_id] = { "plan": plan, "job": job, "card": card, "polling": False, "download_started": False, } if card: card.update_job(job) self.current_plan = plan self.current_job = job self.status.setText("● CADQUERY BUILDING") self.validation_list.item(1).setText("◉ Generating the exact displayed specification hash") self._append_activity( "SHiRE", f"CadQuery build {job_id} started from exact hash {plan['specification_hash']}.", GREEN, ) self._run_task("Starting CadQuery Build", lambda: self.api.generate(plan), success) def _poll_jobs(self) -> None: for job_id, state in tuple(self._jobs.items()): if state.get("polling") or (state.get("job") or {}).get("phase") != "building": continue state["polling"] = True def success(result: dict[str, Any], *, jid: str = job_id) -> None: current = self._jobs.get(jid) if not current: return current["polling"] = False job = result["job"] current["job"] = job card = current.get("card") if card: card.update_job(job) if self.current_job and self.current_job.get("job_id") == jid: self.current_job = job phase = str(job.get("phase") or "") if phase == "complete" and job.get("valid_downloadable_output") is True: self.validation_list.item(1).setText("✓ Exact specification hash generated") self._download_job(jid) elif phase in {"failed", "cancelled"}: self.status.setText(f"● BUILD {phase.upper()}") self.validation_list.item(1).setText(f"✕ Build {phase}; no output exposed") self._append_activity("SHiRE", str(job.get("message") or phase), RED) def failure(message: str, *, jid: str = job_id) -> None: current = self._jobs.get(jid) if current: current["polling"] = False self._task_failed("Read Build Status", message) self._run_task("Reading Build Status", lambda jid=job_id: self.api.job(jid), success, failure) def _download_job(self, job_id: str) -> None: state = self._jobs.get(job_id) if not state or state.get("download_started"): return state["download_started"] = True job = state["job"] plan = state["plan"] def success(artifacts: DownloadedSafePlanArtifacts) -> None: self.current_plan = plan self.current_job = job self.current_artifacts = artifacts try: validation = validate_stl(artifacts.stl_path) if not validation.passed: raise ValueError("Downloaded STL failed native Forge validation.") self.current_safe_validation = validation self._show_mesh(artifacts.stl_path, validation) self.validation_list.item(2).setText("✓ Native Forge STL validation passed") self.validation_list.item(3).setText("○ Awaiting Ray approval for SHiREVault Ready export") self.export_button.setEnabled(True) self.status.setText("● VALIDATED PREVIEW READY") self._append_activity( "SHiRE", f"Validated STEP and STL downloaded into the protected Forge workspace for {job_id}.", GREEN, ) except Exception as exc: self._task_failed("Preview Validated Build", str(exc)) self._run_task( "Downloading Validated Artifacts", lambda: self.api.download_validated_artifacts(job), success, ) def cancel_build(self, plan: dict[str, Any], job: dict[str, Any]) -> None: def success(result: dict[str, Any]) -> None: cancelled = result["job"] state = self._jobs.get(str(cancelled.get("job_id"))) if state: state["job"] = cancelled card = state.get("card") if card: card.update_job(cancelled) self.status.setText("● BUILD CANCELLED") self.export_button.setEnabled(False) self._append_activity("SHiRE", "Build cancelled. Partial files remain quarantined and are not downloadable.", RED) self._run_task("Cancelling Build", lambda: self.api.cancel_build(plan, job), success) def close_plan_card(self, card: SafePlanCard) -> None: revision_id = str(card.plan.get("revision_id") or "") self._cards.pop(revision_id, None) self.plan_layout.removeWidget(card) card.deleteLater() self.plan_count.setText(f"{len(self._cards)} REVISIONS") if not self._cards: self.empty_plan.show() def _show_mesh(self, stl_path: Path, validation: Any) -> None: mesh = load_preview_mesh(stl_path) self.viewer.clear() self.viewer.add_mesh( mesh, color="#6fdb98", ambient=0.32, diffuse=0.78, specular=0.18, specular_power=14, smooth_shading=True, show_edges=True, edge_color="#3a1748", line_width=1, ) self.viewer.show_grid(color="#8b3da8") self.viewer.add_axes() self.viewer.view_isometric() self.viewer.reset_camera() width, depth, height = validation.dimensions self.stats.setText( f"{width:.1f} × {depth:.1f} × {height:.1f} mm • " f"{validation.triangle_count} triangles • Volume {validation.volume:.1f} mm³ • " f"Watertight {validation.watertight}" ) def build_certified_cube(self) -> None: self.export_button.setEnabled(False) parameters = CubeParameters(width=20.0, depth=20.0, height=20.0) def success(result: Any) -> None: self.current_build = result self.current_plan = None self.current_job = None self.current_artifacts = None self._show_mesh(result.stl_path, result.validation) self.export_button.setEnabled(result.validation.passed) self.status.setText("● CERTIFIED CUBE VALIDATED") self.validation_list.item(0).setText("✓ Certified calibration-cube skill") self.validation_list.item(1).setText("✓ Trusted parametric generator completed") self.validation_list.item(2).setText("✓ Native Forge STL validation passed") self.validation_list.item(3).setText("○ Awaiting Ray approval for SHiREVault Ready export") self._run_task("Building Certified Cube", lambda: build_calibration_cube(parameters), success) def export_current_model(self) -> None: self.export_button.setEnabled(False) if self.current_artifacts and self.current_plan and self.current_job: operation = lambda: export_safe_plan_build( plan=self.current_plan, job=self.current_job, artifacts=self.current_artifacts, ) def success(result: Any) -> None: self.status.setText("● SHIREVAULT EXPORT COMPLETE") self.validation_list.item(3).setText("✓ Ray-approved SHiREVault Ready export complete") self._append_activity("SHiRE", f"Validated build exported to {result.export_directory}", GREEN) def failure(message: str) -> None: self.export_button.setEnabled(True) self._task_failed("SHiREVault Export", message) self._run_task("Exporting to SHiREVault", operation, success, failure) return if self.current_build is not None and self.current_build.validation.passed: def legacy_success(result: Any) -> None: self.status.setText("● SHIREVAULT EXPORT COMPLETE") self.validation_list.item(3).setText("✓ Ray-approved SHiREVault Ready export complete") self._append_activity("SHiRE", f"Validated build exported to {result.export_directory}", GREEN) def legacy_failure(message: str) -> None: self.export_button.setEnabled(True) self._task_failed("SHiREVault Export", message) self._run_task( "Exporting to SHiREVault", lambda: export_validated_build(self.current_build), legacy_success, legacy_failure, ) return self.status.setText("● EXPORT BLOCKED") def submit_prompt(self) -> None: """Compatibility alias used by older Forge wrappers.""" self.create_safe_plan() def rebuild_model(self) -> None: """Compatibility alias retaining the certified recovery build.""" self.build_certified_cube() def fit_model(self) -> None: if self.viewer is not None: self.viewer.reset_camera() def reset_camera(self) -> None: if self.viewer is not None: self.viewer.view_isometric() self.viewer.reset_camera() def shutdown(self) -> None: self.poll_timer.stop() for worker in tuple(self._workers): worker.requestInterruption() viewer = getattr(self, "viewer", None) if viewer is not None: viewer.close() self.viewer = None def closeEvent(self, event) -> None: self.shutdown() super().closeEvent(event)