@@ -0,0 +1,117 @@
+import threading
+import time
+from dataclasses import dataclass
+
+from robot_orchestrator.bootlog import BOOT_LOG, BootEventLog
+
+PENDING = "pending"
+RUNNING = "running"
+OK = "ok"
+WARN = "warn"
+FAILED = "failed"
+SKIPPED = "skipped"
+
+TERMINAL = {OK, WARN, FAILED, SKIPPED}
+
+DEFAULT_STEPS: list[tuple[str, str]] = [
+    ("init", "Инициализация"),
+    ("recover", "Восстановление журнала"),
+    ("gpio", "Перемычка GPIO"),
+    ("secrets", "Секреты"),
+    ("qr", "Сканирование QR-кода"),
+    ("camera", "Камера"),
+    ("microphone", "Микрофон"),
+    ("network", "Сеть и интернет"),
+    ("tooling", "Инструменты прошивки"),
+    ("mcu", "ST-Link и микроконтроллер"),
+    ("repos", "Обновление репозиториев"),
+    ("firmware", "Прошивка микроконтроллера"),
+    ("mode", "Определение режима"),
+    ("services", "Запуск сервисов"),
+    ("handoff", "Запуск интерфейса"),
+]
+
+_LEVEL_FOR_STATUS = {OK: "ok", WARN: "warn", FAILED: "error", SKIPPED: "dim"}
+_MARK_FOR_STATUS = {OK: "✔", WARN: "⚠", FAILED: "✖", SKIPPED: "–"}
+
+
+@dataclass
+class BootStep:
+    id: str
+    title: str
+    status: str = PENDING
+    detail: str = ""
+    started_at: float | None = None
+    finished_at: float | None = None
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id, "title": self.title, "status": self.status, "detail": self.detail,
+            "started_at": self.started_at, "finished_at": self.finished_at,
+        }
+
+
+class BootProgress:
+    def __init__(self, log: BootEventLog | None = None, steps: list[tuple[str, str]] | None = None):
+        self.log = log or BOOT_LOG
+        self.steps = [BootStep(step_id, title) for step_id, title in (steps or DEFAULT_STEPS)]
+        self._by_id = {step.id: step for step in self.steps}
+        self.step = "starting"
+        self.started_at = time.time()
+        self.finished_at: float | None = None
+        self._lock = threading.Lock()
+
+    def begin(self, step_id: str, detail: str = "") -> None:
+        with self._lock:
+            step = self._by_id[step_id]
+            step.status = RUNNING
+            step.detail = detail
+            step.started_at = time.time()
+            step.finished_at = None
+            self.step = step_id
+            title = step.title
+        self.log.emit("boot", f"▶ {title}" + (f" — {detail}" if detail else ""), "step")
+
+    def update(self, step_id: str, detail: str) -> None:
+        with self._lock:
+            self._by_id[step_id].detail = detail
+
+    def finish(self, step_id: str, status: str = OK, detail: str = "") -> None:
+        with self._lock:
+            step = self._by_id[step_id]
+            step.status = status
+            if detail:
+                step.detail = detail
+            now = time.time()
+            step.started_at = step.started_at or now
+            step.finished_at = now
+            title, final_detail = step.title, step.detail
+        text = f"{_MARK_FOR_STATUS.get(status, '•')} {title}" + (f" — {final_detail}" if final_detail else "")
+        self.log.emit("boot", text, _LEVEL_FOR_STATUS.get(status, "info"))
+
+    def skip(self, step_id: str, detail: str = "") -> None:
+        self.finish(step_id, SKIPPED, detail)
+
+    def note(self, text: str, level: str = "info", source: str = "boot") -> None:
+        self.log.emit(source, text, level)
+
+    def complete(self) -> None:
+        with self._lock:
+            self.finished_at = time.time()
+            self.step = "done"
+
+    def snapshot(self) -> dict:
+        with self._lock:
+            steps = [step.to_dict() for step in self.steps]
+            now = time.time()
+            end = self.finished_at or now
+            finished = sum(1 for step in self.steps if step.status in TERMINAL)
+            running = sum(1 for step in self.steps if step.status == RUNNING)
+            percent = round(100 * (finished + 0.5 * running) / len(self.steps))
+            return {
+                "step": self.step,
+                "steps": steps,
+                "started_at": self.started_at,
+                "elapsed": round(end - self.started_at, 1),
+                "percent": 100 if self.finished_at else min(percent, 99),
+            }
@@ -0,0 +1,212 @@
+import collections
+import logging
+import re
+import subprocess
+import threading
+import time
+import uuid
+from typing import Callable
+
+_ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[a-zA-Z]")
+_ERROR_RE = re.compile(r"\b(error|traceback|fatal|exception|failed|segmentation fault|critical)\b", re.IGNORECASE)
+_WARN_RE = re.compile(r"\b(warn|warning|deprecated|retry|retrying)\b", re.IGNORECASE)
+_MAX_LINE = 600
+
+
+_URL_USERINFO_RE = re.compile(r"(\b[a-zA-Z][a-zA-Z0-9+.\-]*://)([^/\s:@]+):([^@\s/]+)@")
+_QUERY_SECRET_RE = re.compile(
+    r"([?&](?:token|key|api_key|apikey|access_token|password|passwd|psk|secret)=)[^&\s\"']+", re.IGNORECASE
+)
+_KV_SECRET_RE = re.compile(
+    r"\b(password|passwd|psk|secret|token|api_key|apikey)\b(\s*[=:]\s*)(\"?)[^\s\"',;]+", re.IGNORECASE
+)
+_BEARER_RE = re.compile(r"(\bBearer\s+)[A-Za-z0-9._~+/=\-]+", re.IGNORECASE)
+_MASK = "••••"
+
+
+def strip_ansi(text: str) -> str:
+    return _ANSI_RE.sub("", text)
+
+
+def redact(text: str, known_secrets: tuple[str, ...] = ()) -> str:
+    """Hide credentials before a line can reach a screen or a network client."""
+    for secret in known_secrets:
+        if secret in text:
+            text = text.replace(secret, _MASK)
+    text = _URL_USERINFO_RE.sub(r"\1***:***@", text)
+    text = _QUERY_SECRET_RE.sub(rf"\1{_MASK}", text)
+    text = _KV_SECRET_RE.sub(rf"\1\2\3{_MASK}", text)
+    return _BEARER_RE.sub(rf"\1{_MASK}", text)
+
+
+def classify_level(text: str, default: str = "info") -> str:
+    if _ERROR_RE.search(text):
+        return "error"
+    if _WARN_RE.search(text):
+        return "warn"
+    return default
+
+
+class BootEventLog:
+    """Thread-safe ring buffer of boot events that any number of consumers can follow live."""
+
+    def __init__(self, max_events: int = 4000):
+        self.epoch = uuid.uuid4().hex[:12]
+        self._events: collections.deque[dict] = collections.deque(maxlen=max_events)
+        self._seq = 0
+        self._lock = threading.Lock()
+        self._subscribers: list[Callable[[dict], None]] = []
+        self._known_secrets: tuple[str, ...] = ()
+
+    def set_known_secrets(self, values) -> None:
+        cleaned = {str(v) for v in values if v and len(str(v)) >= 6}
+        self._known_secrets = tuple(sorted(cleaned, key=len, reverse=True))
+
+    def emit(self, source: str, text: str, level: str = "info") -> dict | None:
+        text = redact(strip_ansi(text).rstrip(), self._known_secrets)
+        if not text:
+            return None
+        if len(text) > _MAX_LINE:
+            text = text[:_MAX_LINE] + "…"
+        with self._lock:
+            self._seq += 1
+            event = {"seq": self._seq, "ts": time.time(), "source": source, "level": level, "text": text}
+            self._events.append(event)
+            subscribers = list(self._subscribers)
+        for callback in subscribers:
+            try:
+                callback(event)
+            except Exception:
+                pass
+        return event
+
+    def tail(self, after_seq: int = 0, limit: int = 500) -> list[dict]:
+        with self._lock:
+            events = [e for e in self._events if e["seq"] > after_seq]
+        return events[-limit:]
+
+    def subscribe(self, callback: Callable[[dict], None]) -> None:
+        with self._lock:
+            self._subscribers.append(callback)
+
+    def unsubscribe(self, callback: Callable[[dict], None]) -> None:
+        with self._lock:
+            if callback in self._subscribers:
+                self._subscribers.remove(callback)
+
+    @property
+    def last_seq(self) -> int:
+        with self._lock:
+            return self._seq
+
+
+BOOT_LOG = BootEventLog()
+
+
+def forward_service_line(name: str, raw: str, log: BootEventLog | None = None) -> None:
+    """Mirror a LogSink line ('<ts> <stream> <text>') into the boot event log."""
+    from robot_orchestrator.supervisor.logsink import parse_line
+
+    parsed = parse_line(name, raw)
+    (log or BOOT_LOG).emit(name, parsed["text"], classify_level(parsed["text"]))
+
+
+class BootLogHandler(logging.Handler):
+    """Forwards orchestrator log records (not uvicorn/asyncio chatter) to the boot event log."""
+
+    def __init__(self, log: BootEventLog | None = None):
+        super().__init__(level=logging.INFO)
+        self._log = log or BOOT_LOG
+
+    def emit(self, record: logging.LogRecord) -> None:
+        if record.name.startswith(("uvicorn", "asyncio", "http")):
+            return
+        try:
+            level = {"WARNING": "warn", "ERROR": "error", "CRITICAL": "error"}.get(record.levelname, "info")
+            self._log.emit("orchestrator", record.getMessage(), level)
+        except Exception:
+            pass
+
+
+def _mask_values(text: str, mask: tuple[str, ...]) -> str:
+    for value in mask:
+        if value:
+            text = text.replace(value, _MASK)
+    return text
+
+
+def _pump(stream, source: str, sink: list[str], log: BootEventLog, mask: tuple[str, ...]) -> None:
+    last = None
+    try:
+        for raw in stream:
+            sink.append(raw)
+            line = _mask_values(strip_ansi(raw).rstrip(), mask)
+            if not line or line == last:
+                continue
+            last = line
+            log.emit(source, line, classify_level(line))
+    except (ValueError, OSError):
+        pass
+
+
+def _describe(argv, mask: tuple[str, ...] = ()) -> str:
+    text = argv if isinstance(argv, str) else " ".join(str(part) for part in argv)
+    text = _mask_values(text, mask)
+    return text if len(text) <= 240 else text[:237] + "..."
+
+
+def logged_run(
+    argv,
+    *,
+    source: str,
+    cwd=None,
+    env=None,
+    timeout: float | None = None,
+    check: bool = False,
+    shell: bool = False,
+    announce: bool = True,
+    mask: tuple[str, ...] = (),
+    log: BootEventLog | None = None,
+) -> subprocess.CompletedProcess:
+    """Drop-in for subprocess.run(capture_output=True, text=True) that also streams every output line live."""
+    log = log or BOOT_LOG
+    if announce:
+        log.emit(source, f"$ {_describe(argv, mask)}", "cmd")
+
+    process = subprocess.Popen(
+        argv, cwd=cwd, env=env, shell=shell,
+        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        text=True, encoding="utf-8", errors="replace", bufsize=1,
+    )
+    out_lines: list[str] = []
+    err_lines: list[str] = []
+    threads = [
+        threading.Thread(target=_pump, args=(process.stdout, source, out_lines, log, mask), daemon=True),
+        threading.Thread(target=_pump, args=(process.stderr, source, err_lines, log, mask), daemon=True),
+    ]
+    for thread in threads:
+        thread.start()
+
+    try:
+        process.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        process.kill()
+        process.wait()
+        for thread in threads:
+            thread.join(timeout=2.0)
+        log.emit(source, f"таймаут {timeout}s: процесс убит", "error")
+        raise subprocess.TimeoutExpired(argv, timeout, output="".join(out_lines), stderr="".join(err_lines))
+    except BaseException:
+        process.kill()
+        process.wait()
+        raise
+
+    for thread in threads:
+        thread.join(timeout=5.0)
+
+    stdout, stderr = "".join(out_lines), "".join(err_lines)
+    if announce and process.returncode != 0:
+        log.emit(source, f"завершилось с кодом {process.returncode}", "error")
+    if check and process.returncode != 0:
+        raise subprocess.CalledProcessError(process.returncode, argv, output=stdout, stderr=stderr)
+    return subprocess.CompletedProcess(argv, process.returncode, stdout, stderr)