"""Native ARMOR OS dialog for SHiRE Unified Learning 0003.""" from __future__ import annotations import json import re import urllib.error import urllib.request from typing import Any from PyQt5.QtCore import QThread, pyqtSignal, Qt from PyQt5.QtWidgets import ( QDialog, QHBoxLayout, QLabel, QMessageBox, QPlainTextEdit, QPushButton, QTextBrowser, QVBoxLayout, ) SERVICE_URL = "http://127.0.0.1:8792" MAX_PROMPT_CHARS = 1000 def is_unified_learning_command(prompt: str) -> bool: return bool(re.match(r"^\s*shire\s+learn(?:\s|:|-|$)", str(prompt or ""), flags=re.I)) def strip_learning_command(prompt: str) -> str: return re.sub(r"^\s*shire\s+learn\s*[:\-]?\s*", "", str(prompt or ""), flags=re.I).strip() class UnifiedLearningRequestWorker(QThread): completed = pyqtSignal(bool, object) def __init__(self, path: str, payload: dict[str, Any], timeout: int = 330, parent=None): super().__init__(parent) self.path = path self.payload = payload self.timeout = timeout def run(self) -> None: try: request = urllib.request.Request( SERVICE_URL + self.path, data=json.dumps(self.payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(request, timeout=self.timeout) as response: data = json.loads(response.read().decode("utf-8")) self.completed.emit(bool(data.get("ok")), data) except urllib.error.HTTPError as exc: try: data = json.loads(exc.read().decode("utf-8")) except Exception: data = {"ok": False, "error": f"HTTP {exc.code}"} self.completed.emit(False, data) except Exception as exc: self.completed.emit(False, {"ok": False, "error": str(exc)}) class UnifiedLearningDialog(QDialog): """Single plan-and-approval popup shared by ARMOR OS and Academy.""" def __init__(self, prompt: str = "", parent=None): super().__init__(parent) self.plan: dict[str, Any] | None = None self.started_job: dict[str, Any] | None = None self.worker: UnifiedLearningRequestWorker | None = None self.setWindowTitle("SHiRE Unified Learning") self.setModal(True) self.resize(900, 760) self.setMinimumSize(760, 620) self.setStyleSheet(""" QDialog { background: #070910; color: #f3f5fa; } QLabel { color: #d9deea; font-size: 13px; } QPlainTextEdit, QTextBrowser { background: #0d111c; color: #f4f6fb; border: 1px solid #384154; border-radius: 8px; padding: 9px; selection-background-color: #53657f; } QPushButton { background: #151b29; color: #f4f6fb; border: 1px solid #4b566d; border-radius: 7px; padding: 9px 15px; font-weight: 700; } QPushButton:hover { background: #202a3d; } QPushButton:disabled { color: #697386; border-color: #2a3040; } QPushButton#startLearning { background: #17382d; border-color: #3c8d70; } """) root = QVBoxLayout(self) root.setContentsMargins(18, 18, 18, 18) root.setSpacing(10) heading = QLabel("SHiRE Unified Learning") heading.setStyleSheet("font-size: 22px; font-weight: 800; color: #ffffff;") root.addWidget(heading) guidance = QLabel( "Describe what SHiRE should learn. The complete plan appears here before learning starts." ) guidance.setWordWrap(True) root.addWidget(guidance) self.prompt_edit = QPlainTextEdit() self.prompt_edit.setPlaceholderText("Example: video game cosplay outfits") self.prompt_edit.setPlainText(strip_learning_command(prompt)) self.prompt_edit.setMaximumHeight(130) self.prompt_edit.textChanged.connect(self._update_counter) root.addWidget(self.prompt_edit) self.counter = QLabel() self.counter.setAlignment(Qt.AlignRight) root.addWidget(self.counter) self.status = QLabel("Review or revise the request to create the learning plan.") self.status.setWordWrap(True) root.addWidget(self.status) self.plan_view = QTextBrowser() self.plan_view.setOpenExternalLinks(False) self.plan_view.setHtml(self._empty_plan_html()) root.addWidget(self.plan_view, 1) buttons = QHBoxLayout() self.start_button = QPushButton("Start Learning") self.start_button.setObjectName("startLearning") self.revise_button = QPushButton("Revise Plan") self.save_button = QPushButton("Save for Later") self.cancel_button = QPushButton("Cancel") self.start_button.clicked.connect(self.start_learning) self.revise_button.clicked.connect(self.analyse_plan) self.save_button.clicked.connect(self.save_for_later) self.cancel_button.clicked.connect(self.reject) buttons.addWidget(self.start_button) buttons.addWidget(self.revise_button) buttons.addWidget(self.save_button) buttons.addStretch(1) buttons.addWidget(self.cancel_button) root.addLayout(buttons) self._update_counter() self._set_plan_actions(False) if strip_learning_command(prompt): self.analyse_plan() @classmethod def open_for_prompt(cls, prompt: str = "", parent=None) -> int: dialog = cls(prompt, parent) return dialog.exec_() def _empty_plan_html(self) -> str: return ( "

Plan not analysed yet

" "

The plan will show scope, stages, prerequisites, selected SHiRE brains and Agents, " "tools and sources, estimated duration, autonomous versus manual work, safety, intellectual " "property and licensing, blockers, and expected capability.

" ) def _prompt(self) -> str: return self.prompt_edit.toPlainText().strip() def _valid_prompt(self, show_error: bool = True) -> bool: prompt = self._prompt() if not prompt: if show_error: QMessageBox.warning(self, "Learning request required", "Enter what SHiRE should learn.") return False if len(prompt) > MAX_PROMPT_CHARS: if show_error: QMessageBox.warning( self, "Learning request too long", f"The request is {len(prompt)} characters. The limit is {MAX_PROMPT_CHARS}.", ) return False return True def _update_counter(self) -> None: count = len(self.prompt_edit.toPlainText()) self.counter.setText(f"{count} / {MAX_PROMPT_CHARS} characters") self.counter.setStyleSheet("color: #ff8585;" if count > MAX_PROMPT_CHARS else "color: #9aa7ba;") if count > MAX_PROMPT_CHARS: self.start_button.setEnabled(False) self.save_button.setEnabled(False) self.revise_button.setEnabled(False) else: self.revise_button.setEnabled(True) self.start_button.setEnabled(self.plan is not None) self.save_button.setEnabled(self.plan is not None) def _set_busy(self, busy: bool, text: str = "") -> None: self.prompt_edit.setEnabled(not busy) self.start_button.setEnabled(not busy and self.plan is not None) self.revise_button.setEnabled(not busy) self.save_button.setEnabled(not busy and self.plan is not None) self.cancel_button.setEnabled(not busy) if text: self.status.setText(text) def _set_plan_actions(self, enabled: bool) -> None: self.start_button.setEnabled(enabled) self.save_button.setEnabled(enabled) def _request(self, path: str, payload: dict[str, Any], callback, busy_text: str) -> None: if self.worker is not None and self.worker.isRunning(): return self._set_busy(True, busy_text) self.worker = UnifiedLearningRequestWorker(path, payload, parent=self) self.worker.completed.connect(callback) self.worker.finished.connect(lambda: self._set_busy(False)) self.worker.start() def analyse_plan(self) -> None: if not self._valid_prompt(): return self._request( "/api/plan/analyse", {"prompt": self._prompt()}, self._plan_ready, "SHiRE is analysing the request and building the plan...", ) def _plan_ready(self, ok: bool, payload: object) -> None: data = payload if isinstance(payload, dict) else {} if not ok: self.plan = None self.plan_view.setHtml(self._empty_plan_html()) self._set_plan_actions(False) QMessageBox.critical(self, "Plan analysis failed", str(data.get("error") or "Unknown error")) return self.plan = data.get("plan") if isinstance(data.get("plan"), dict) else None if self.plan is None: QMessageBox.critical(self, "Plan analysis failed", "The service returned no plan.") return self.plan_view.setHtml(self._plan_html(self.plan)) self.status.setText("Plan ready. Start, revise, save, or cancel.") self._set_plan_actions(True) def start_learning(self) -> None: if self.plan is None or not self._valid_prompt(): return self.plan["request"] = self._prompt() self._request( "/api/job/start", {"plan": self.plan}, self._job_started, "Creating the persistent learning job...", ) def _job_started(self, ok: bool, payload: object) -> None: data = payload if isinstance(payload, dict) else {} if not ok: QMessageBox.critical(self, "Learning did not start", str(data.get("error") or "Unknown error")) return job = data.get("job") if isinstance(data.get("job"), dict) else {} self.started_job = job self.status.setText( f"Learning job {job.get('job_id', 'created')} was confirmed on SHiREVault. Closing plan." ) # SHIRE-Unified-Learning-0003: close after the persistent job exists. self.accept() def save_for_later(self) -> None: if self.plan is None or not self._valid_prompt(): return self.plan["request"] = self._prompt() self._request( "/api/plan/save", {"plan": self.plan}, self._plan_saved, "Saving the plan to SHiREVault...", ) def _plan_saved(self, ok: bool, payload: object) -> None: data = payload if isinstance(payload, dict) else {} if not ok: QMessageBox.critical(self, "Plan was not saved", str(data.get("error") or "Unknown error")) return self.plan = data.get("plan") if isinstance(data.get("plan"), dict) else self.plan self.status.setText("Plan saved for later on SHiREVault.") QMessageBox.information(self, "Plan saved", "The plan was saved without starting learning.") def _plan_html(self, plan: dict[str, Any]) -> str: def esc(value: Any) -> str: import html return html.escape(str(value)) def list_items(values: Any) -> str: rows = values if isinstance(values, list) else [] return "" scope = plan.get("scope") if isinstance(plan.get("scope"), dict) else {} work = plan.get("work_split") if isinstance(plan.get("work_split"), dict) else {} safety = plan.get("safety") if isinstance(plan.get("safety"), dict) else {} ip = plan.get("ip_and_licensing") if isinstance(plan.get("ip_and_licensing"), dict) else {} duration = plan.get("estimated_duration") if isinstance(plan.get("estimated_duration"), dict) else {} stages = plan.get("stages") if isinstance(plan.get("stages"), list) else [] brains = plan.get("selected_brains_agents") if isinstance(plan.get("selected_brains_agents"), list) else [] tools = plan.get("tools_and_sources") if isinstance(plan.get("tools_and_sources"), list) else [] stage_html = "".join( f"
  • {esc(stage.get('title'))} — {esc(stage.get('objective'))} " f"({esc(stage.get('kind'))})
    • " for stage in stages if isinstance(stage, dict) ) brain_html = "".join( f"
  • {esc(item.get('name'))} — {esc(item.get('role'))}
    • " for item in brains if isinstance(item, dict) ) tool_html = "".join( f"
  • {esc(item.get('name'))}: {esc(item.get('status'))} — {esc(item.get('claim_boundary'))}
    • " for item in tools if isinstance(item, dict) ) return f"""

    {esc(plan.get('title'))}

    Scope

    Included{list_items(scope.get('included'))}Excluded{list_items(scope.get('excluded'))}

    Stages

      {stage_html}

    Prerequisites

    {list_items(plan.get('prerequisites'))}

    Selected SHiRE brains and Agents

    Tools and sources

    Estimated duration

    {esc(duration.get('minutes'))} minutes. {esc(duration.get('basis'))}

    Autonomous versus manual work

    Autonomous{list_items(work.get('autonomous'))}Manual or review{list_items(work.get('manual_or_review'))}

    Safety

    Manual review required: {esc(safety.get('manual_review_required'))}. No protected Academy writes, counted attempts, certification, Forge changes, tool execution or physical testing.

    IP and licensing

    {esc(ip.get('note'))}

    Blockers

    {list_items(plan.get('blockers')) if plan.get('blockers') else '

    None identified for local knowledge stages.

    '}

    Expected capability

    {esc(plan.get('expected_capability'))}

    """