chore: commit remaining pre-existing changes
- Agent configs (.opencode/agents/) - Backend: alembic, routes, app, utils, scripts - Frontend: package.json, vite, components, e2e infra - Specs: 028-llm-datasource-supeset updates - Docker e2e config and Playwright setup
This commit is contained in:
@@ -4,21 +4,21 @@
|
||||
# @RELATION DEPENDS_ON -> [ComplianceExecutionService]
|
||||
# @RELATION DEPENDS_ON -> [CleanReleaseRepository]
|
||||
# @INVARIANT: TUI refuses startup in non-TTY environments; headless flow is CLI/API only.
|
||||
|
||||
import contextlib
|
||||
import curses
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
from datetime import UTC, datetime
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
# Standardize sys.path for direct execution from project root or scripts dir.
|
||||
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
BACKEND_ROOT = os.path.abspath(os.path.join(SCRIPT_DIR, "..", ".."))
|
||||
if BACKEND_ROOT not in sys.path:
|
||||
sys.path.insert(0, BACKEND_ROOT)
|
||||
|
||||
from src.core.cot_logger import seed_trace_id
|
||||
from src.models.clean_release import (
|
||||
CandidateArtifact,
|
||||
@@ -43,8 +43,6 @@ from src.services.clean_release.enums import CandidateStatus
|
||||
from src.services.clean_release.manifest_service import build_manifest_snapshot
|
||||
from src.services.clean_release.publication_service import publish_candidate
|
||||
from src.services.clean_release.repository import CleanReleaseRepository
|
||||
|
||||
|
||||
# #region TuiFacadeAdapter [TYPE Class]
|
||||
# @BRIEF Thin TUI adapter that routes business mutations through application services.
|
||||
# @PRE: repository contains candidate and trusted policy/registry snapshots for execution.
|
||||
@@ -52,7 +50,6 @@ from src.services.clean_release.repository import CleanReleaseRepository
|
||||
class TuiFacadeAdapter:
|
||||
def __init__(self, repository: CleanReleaseRepository):
|
||||
self.repository = repository
|
||||
|
||||
def _build_config_manager(self):
|
||||
policy = self.repository.get_active_policy()
|
||||
if policy is None:
|
||||
@@ -64,7 +61,6 @@ class TuiFacadeAdapter:
|
||||
settings = SimpleNamespace(clean_release=clean_release)
|
||||
config = SimpleNamespace(settings=settings)
|
||||
return SimpleNamespace(get_config=lambda: config)
|
||||
|
||||
def run_compliance(self, *, candidate_id: str, actor: str):
|
||||
manifests = self.repository.get_manifests_by_candidate(candidate_id)
|
||||
if not manifests:
|
||||
@@ -81,7 +77,6 @@ class TuiFacadeAdapter:
|
||||
requested_by=actor,
|
||||
manifest_id=latest_manifest.id,
|
||||
)
|
||||
|
||||
def approve_latest(self, *, candidate_id: str, actor: str):
|
||||
reports = [
|
||||
item
|
||||
@@ -98,7 +93,6 @@ class TuiFacadeAdapter:
|
||||
decided_by=actor,
|
||||
comment="Approved from TUI",
|
||||
)
|
||||
|
||||
def publish_latest(self, *, candidate_id: str, actor: str):
|
||||
reports = [
|
||||
item
|
||||
@@ -116,14 +110,12 @@ class TuiFacadeAdapter:
|
||||
target_channel="stable",
|
||||
publication_ref=None,
|
||||
)
|
||||
|
||||
def build_manifest(self, *, candidate_id: str, actor: str):
|
||||
return build_manifest_snapshot(
|
||||
repository=self.repository,
|
||||
candidate_id=candidate_id,
|
||||
created_by=actor,
|
||||
)
|
||||
|
||||
def get_overview(self, *, candidate_id: str) -> dict[str, Any]:
|
||||
candidate = self.repository.get_candidate(candidate_id)
|
||||
manifests = self.repository.get_manifests_by_candidate(candidate_id)
|
||||
@@ -186,11 +178,7 @@ class TuiFacadeAdapter:
|
||||
"policy": policy,
|
||||
"registry": registry,
|
||||
}
|
||||
|
||||
|
||||
# #endregion TuiFacadeAdapter
|
||||
|
||||
|
||||
# #region CleanReleaseTUI [TYPE Class]
|
||||
# @BRIEF Curses-based application for compliance monitoring.
|
||||
# @UX_STATE: READY -> Waiting for operator to start checks (F5).
|
||||
@@ -212,7 +200,6 @@ class CleanReleaseTUI:
|
||||
self.last_error: str | None = None
|
||||
self.overview: dict[str, Any] = {}
|
||||
self.refresh_overview()
|
||||
|
||||
curses.start_color()
|
||||
curses.use_default_colors()
|
||||
curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLUE) # Header/Footer
|
||||
@@ -220,7 +207,6 @@ class CleanReleaseTUI:
|
||||
curses.init_pair(3, curses.COLOR_RED, -1) # FAIL/BLOCKED
|
||||
curses.init_pair(4, curses.COLOR_YELLOW, -1) # RUNNING
|
||||
curses.init_pair(5, curses.COLOR_CYAN, -1) # Text
|
||||
|
||||
def _build_repository(self, mode: str) -> CleanReleaseRepository:
|
||||
repo = CleanReleaseRepository()
|
||||
if mode == "demo":
|
||||
@@ -228,7 +214,6 @@ class CleanReleaseTUI:
|
||||
else:
|
||||
self._bootstrap_real_repository(repo)
|
||||
return repo
|
||||
|
||||
def _bootstrap_demo_repository(self, repository: CleanReleaseRepository) -> None:
|
||||
now = datetime.now(UTC)
|
||||
policy = CleanProfilePolicy(
|
||||
@@ -242,7 +227,6 @@ class CleanReleaseTUI:
|
||||
)
|
||||
policy.immutable = True
|
||||
repository.save_policy(policy)
|
||||
|
||||
registry = ResourceSourceRegistry(
|
||||
registry_id="REG-1",
|
||||
name="Default Internal Registry",
|
||||
@@ -301,15 +285,12 @@ class CleanReleaseTUI:
|
||||
summary = dict(manifest.content_json.get("summary", {}))
|
||||
summary["prohibited_detected_count"] = 1
|
||||
manifest.content_json["summary"] = summary
|
||||
|
||||
def _bootstrap_real_repository(self, repository: CleanReleaseRepository) -> None:
|
||||
bootstrap_path = os.getenv("CLEAN_TUI_BOOTSTRAP_JSON", "").strip()
|
||||
if not bootstrap_path:
|
||||
return
|
||||
|
||||
with open(bootstrap_path, encoding="utf-8") as bootstrap_file:
|
||||
payload = json.load(bootstrap_file)
|
||||
|
||||
now = datetime.now(UTC)
|
||||
candidate = ReleaseCandidate(
|
||||
id=payload.get("candidate_id", "candidate-1"),
|
||||
@@ -329,7 +310,6 @@ class CleanReleaseTUI:
|
||||
if imported_artifacts:
|
||||
candidate.transition_to(CandidateStatus.PREPARED)
|
||||
repository.save_candidate(candidate)
|
||||
|
||||
registry_id = payload.get("registry_id", "REG-1")
|
||||
entries = [
|
||||
ResourceSourceEntry(
|
||||
@@ -353,7 +333,6 @@ class CleanReleaseTUI:
|
||||
status=RegistryStatus.ACTIVE,
|
||||
)
|
||||
)
|
||||
|
||||
if entries:
|
||||
repository.save_policy(
|
||||
CleanProfilePolicy(
|
||||
@@ -372,17 +351,14 @@ class CleanReleaseTUI:
|
||||
effective_from=now,
|
||||
)
|
||||
)
|
||||
|
||||
def _resolve_candidate_id(self) -> str:
|
||||
env_candidate = os.getenv("CLEAN_TUI_CANDIDATE_ID", "").strip()
|
||||
if env_candidate:
|
||||
return env_candidate
|
||||
|
||||
candidate_ids = list(self.repo.candidates.keys())
|
||||
if candidate_ids:
|
||||
return candidate_ids[0]
|
||||
return ""
|
||||
|
||||
def draw_header(self, max_y: int, max_x: int):
|
||||
header_text = " Enterprise Clean Release Validator (TUI) "
|
||||
self.stdscr.attron(curses.color_pair(1) | curses.A_BOLD)
|
||||
@@ -390,7 +366,6 @@ class CleanReleaseTUI:
|
||||
centered = header_text.center(max_x)
|
||||
self.stdscr.addstr(0, 0, centered[:max_x])
|
||||
self.stdscr.attroff(curses.color_pair(1) | curses.A_BOLD)
|
||||
|
||||
candidate = self.overview.get("candidate")
|
||||
candidate_text = self.candidate_id or "not-set"
|
||||
profile_text = "enterprise-clean"
|
||||
@@ -400,7 +375,6 @@ class CleanReleaseTUI:
|
||||
f"Lifecycle: [{lifecycle}] Mode: [{self.mode}]"
|
||||
).ljust(max_x)
|
||||
self.stdscr.addstr(2, 0, info_line_text[:max_x])
|
||||
|
||||
def draw_checks(self):
|
||||
self.stdscr.addstr(4, 3, "Checks:")
|
||||
check_defs = [
|
||||
@@ -412,14 +386,11 @@ class CleanReleaseTUI:
|
||||
(CheckStageName.NO_EXTERNAL_ENDPOINTS, "No External Internet Endpoints"),
|
||||
(CheckStageName.MANIFEST_CONSISTENCY, "Release Manifest Consistency"),
|
||||
]
|
||||
|
||||
row = 5
|
||||
drawn_checks = {c["stage"]: c for c in self.checks_progress}
|
||||
|
||||
for stage, desc in check_defs:
|
||||
status_text = " "
|
||||
color = curses.color_pair(5)
|
||||
|
||||
if stage in drawn_checks:
|
||||
c = drawn_checks[stage]
|
||||
if c["status"] == "RUNNING":
|
||||
@@ -431,12 +402,10 @@ class CleanReleaseTUI:
|
||||
elif c["status"] == CheckStageStatus.FAIL:
|
||||
status_text = "FAIL"
|
||||
color = curses.color_pair(3)
|
||||
|
||||
self.stdscr.addstr(row, 4, f"[{status_text:^4}] {desc}")
|
||||
if status_text != " ":
|
||||
self.stdscr.addstr(row, 50, f"{status_text:>10}", color | curses.A_BOLD)
|
||||
row += 1
|
||||
|
||||
def draw_sources(self):
|
||||
self.stdscr.addstr(12, 3, "Allowed Internal Sources:", curses.A_BOLD)
|
||||
reg = self.overview.get("registry")
|
||||
@@ -447,31 +416,26 @@ class CleanReleaseTUI:
|
||||
row += 1
|
||||
else:
|
||||
self.stdscr.addstr(row, 3, " - (none)")
|
||||
|
||||
def draw_status(self):
|
||||
color = curses.color_pair(5)
|
||||
if self.status == CheckFinalStatus.COMPLIANT:
|
||||
color = curses.color_pair(2)
|
||||
elif self.status == CheckFinalStatus.BLOCKED:
|
||||
color = curses.color_pair(3)
|
||||
|
||||
stat_str = str(
|
||||
self.status.value if hasattr(self.status, "value") else self.status
|
||||
)
|
||||
self.stdscr.addstr(
|
||||
18, 3, f"FINAL STATUS: {stat_str.upper()}", color | curses.A_BOLD
|
||||
)
|
||||
|
||||
if self.report_id:
|
||||
self.stdscr.addstr(19, 3, f"Report ID: {self.report_id}")
|
||||
|
||||
approval = self.overview.get("approval")
|
||||
publication = self.overview.get("publication")
|
||||
if approval:
|
||||
self.stdscr.addstr(20, 3, f"Approval: {approval.decision}")
|
||||
if publication:
|
||||
self.stdscr.addstr(20, 32, f"Publication: {publication.status}")
|
||||
|
||||
if self.violations_list:
|
||||
self.stdscr.addstr(
|
||||
21,
|
||||
@@ -498,15 +462,13 @@ class CleanReleaseTUI:
|
||||
f"Error: {self.last_error}"[:100],
|
||||
curses.color_pair(3) | curses.A_BOLD,
|
||||
)
|
||||
|
||||
def draw_footer(self, max_y: int, max_x: int):
|
||||
footer_text = " F5 Run F6 Manifest F7 Refresh F8 Approve F9 Publish F10 Exit ".center(
|
||||
footer_text = " F4 Build Bundle F5 Run F6 Manifest F7 Refresh F8 Approve F9 Publish F10 Exit ".center(
|
||||
max_x
|
||||
)
|
||||
self.stdscr.attron(curses.color_pair(1))
|
||||
self.stdscr.addstr(max_y - 1, 0, footer_text[:max_x])
|
||||
self.stdscr.attroff(curses.color_pair(1))
|
||||
|
||||
# [DEF:run_checks:Function]
|
||||
# @PURPOSE: Execute compliance run via facade adapter and update UI state.
|
||||
# @PRE: Candidate and policy snapshots are present in repository.
|
||||
@@ -518,7 +480,6 @@ class CleanReleaseTUI:
|
||||
self.checks_progress = []
|
||||
self.last_error = None
|
||||
self.refresh_screen()
|
||||
|
||||
try:
|
||||
result = self.facade.run_compliance(
|
||||
candidate_id=self.candidate_id, actor="operator"
|
||||
@@ -528,7 +489,6 @@ class CleanReleaseTUI:
|
||||
self.last_error = str(exc)
|
||||
self.refresh_screen()
|
||||
return
|
||||
|
||||
self.checks_progress = [
|
||||
{
|
||||
"stage": stage.stage_name,
|
||||
@@ -540,7 +500,6 @@ class CleanReleaseTUI:
|
||||
]
|
||||
self.violations_list = result.violations
|
||||
self.report_id = result.report.id if result.report is not None else None
|
||||
|
||||
final_status = str(result.run.final_status or "").upper()
|
||||
if final_status in {"BLOCKED", CheckFinalStatus.BLOCKED.value}:
|
||||
self.status = CheckFinalStatus.BLOCKED
|
||||
@@ -550,9 +509,7 @@ class CleanReleaseTUI:
|
||||
self.status = CheckFinalStatus.FAILED
|
||||
self.refresh_overview()
|
||||
self.refresh_screen()
|
||||
|
||||
# [/DEF:run_checks:Function]
|
||||
|
||||
def build_manifest(self):
|
||||
try:
|
||||
manifest = self.facade.build_manifest(
|
||||
@@ -567,7 +524,6 @@ class CleanReleaseTUI:
|
||||
self.last_error = str(exc)
|
||||
self.refresh_overview()
|
||||
self.refresh_screen()
|
||||
|
||||
def clear_history(self):
|
||||
self.status = "READY"
|
||||
self.report_id = None
|
||||
@@ -576,7 +532,6 @@ class CleanReleaseTUI:
|
||||
self.last_error = None
|
||||
self.refresh_overview()
|
||||
self.refresh_screen()
|
||||
|
||||
def approve_latest(self):
|
||||
if not self.report_id:
|
||||
self.last_error = "F8 disabled: no compliance report available"
|
||||
@@ -589,7 +544,6 @@ class CleanReleaseTUI:
|
||||
self.last_error = str(exc)
|
||||
self.refresh_overview()
|
||||
self.refresh_screen()
|
||||
|
||||
def publish_latest(self):
|
||||
if not self.report_id:
|
||||
self.last_error = "F9 disabled: no compliance report available"
|
||||
@@ -602,26 +556,173 @@ class CleanReleaseTUI:
|
||||
self.last_error = str(exc)
|
||||
self.refresh_overview()
|
||||
self.refresh_screen()
|
||||
|
||||
def refresh_overview(self):
|
||||
if not self.report_id:
|
||||
self.last_error = "F9 disabled: no compliance report available"
|
||||
self.refresh_screen()
|
||||
# #region bundle_build_mode [C:4] [TYPE Function] [SEMANTICS bundle,build,release]
|
||||
# @BRIEF Interactive bundle build screen — select type, enter tag, watch live build output.
|
||||
# @PRE TTY is available. Bundle type selected (1 or 2). Tag is non-empty.
|
||||
# @POST Subprocess completes (success/failure). Output displayed. User returns via Esc.
|
||||
# @SIDE_EFFECT Runs docker build subprocess, creates dist/docker/ files
|
||||
# @RELATION CALLS -> [EXT:build.sh]
|
||||
# @UX_STATE BUNDLE_MENU -> User selects bundle type and enters release tag
|
||||
# @UX_STATE BUNDLE_BUILDING -> Subprocess running, real-time output displayed
|
||||
# @UX_STATE BUNDLE_DONE -> Build completed with success/failure result
|
||||
def bundle_build_mode(self):
|
||||
"""Enter bundle build sub-loop: choose type, enter tag, watch live output."""
|
||||
max_y, max_x = self.stdscr.getmaxyx()
|
||||
# ── Phase 1: Bundle type selection ──
|
||||
self.stdscr.clear()
|
||||
self.stdscr.attron(curses.color_pair(1) | curses.A_BOLD)
|
||||
self.stdscr.addstr(0, 0, " Bundle Build ".center(max_x)[:max_x])
|
||||
self.stdscr.attroff(curses.color_pair(1) | curses.A_BOLD)
|
||||
self.stdscr.addstr(3, 3, "Select bundle type:", curses.A_BOLD)
|
||||
self.stdscr.addstr(5, 5, "[1] Full Bundle (backend + frontend + postgres — all dependencies)")
|
||||
self.stdscr.addstr(6, 5, "[2] Light Bundle (~200MB, compressed without Playwright)")
|
||||
self.stdscr.addstr(8, 3, "Your choice: ")
|
||||
self.stdscr.addstr(max_y - 1, 0, " Esc: Cancel ".center(max_x)[:max_x],
|
||||
curses.color_pair(1))
|
||||
self.stdscr.refresh()
|
||||
bundle_cmd: str | None = None
|
||||
bundle_label: str | None = None
|
||||
while True:
|
||||
k = self.stdscr.getch()
|
||||
if k == ord("1"):
|
||||
bundle_cmd, bundle_label = "bundle", "Full Bundle"
|
||||
break
|
||||
if k == ord("2"):
|
||||
bundle_cmd, bundle_label = "bundle:light", "Light Bundle"
|
||||
break
|
||||
if k == 27: # Esc
|
||||
return
|
||||
# ── Phase 2: Tag input ──
|
||||
self.stdscr.clear()
|
||||
self.stdscr.attron(curses.color_pair(1) | curses.A_BOLD)
|
||||
self.stdscr.addstr(0, 0, f" Bundle Build — {bundle_label} ".center(max_x)[:max_x])
|
||||
self.stdscr.attroff(curses.color_pair(1) | curses.A_BOLD)
|
||||
self.stdscr.addstr(3, 3, f"Type: {bundle_label}", curses.A_BOLD)
|
||||
self.stdscr.addstr(5, 3, "Release tag (e.g. v1.0.0): ")
|
||||
self.stdscr.addstr(max_y - 1, 0, " Esc: Cancel ".center(max_x)[:max_x],
|
||||
curses.color_pair(1))
|
||||
self.stdscr.refresh()
|
||||
curses.echo()
|
||||
raw_tag = self.stdscr.getstr(5, 33, 30).decode().strip()
|
||||
curses.noecho()
|
||||
tag = raw_tag.strip()
|
||||
if not tag:
|
||||
return
|
||||
# ── Phase 3: Build execution ──
|
||||
build_script = os.path.join(os.path.dirname(BACKEND_ROOT), "build.sh")
|
||||
self.stdscr.clear()
|
||||
self.stdscr.attron(curses.color_pair(1) | curses.A_BOLD)
|
||||
self.stdscr.addstr(0, 0, f" Building {bundle_label} — {tag} ".center(max_x)[:max_x])
|
||||
self.stdscr.attroff(curses.color_pair(1) | curses.A_BOLD)
|
||||
self.stdscr.refresh()
|
||||
try:
|
||||
self.facade.publish_latest(candidate_id=self.candidate_id, actor="operator")
|
||||
self.last_error = None
|
||||
except Exception as exc:
|
||||
self.last_error = str(exc)
|
||||
self.refresh_overview()
|
||||
self.refresh_screen()
|
||||
|
||||
proc = subprocess.Popen(
|
||||
[build_script, bundle_cmd, tag],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
bufsize=1,
|
||||
)
|
||||
except (FileNotFoundError, OSError) as exc:
|
||||
self.stdscr.addstr(3, 3, str(exc), curses.color_pair(3))
|
||||
self.stdscr.addstr(max_y - 1, 0, " Press Esc to return ".center(max_x)[:max_x],
|
||||
curses.color_pair(1))
|
||||
self.stdscr.refresh()
|
||||
while self.stdscr.getch() != 27:
|
||||
pass
|
||||
return
|
||||
output_lines: list[str] = []
|
||||
lock = threading.Lock()
|
||||
build_done = threading.Event()
|
||||
return_code: list[int | None] = [None]
|
||||
def _reader_thread() -> None:
|
||||
"""Read subprocess stdout line by line in a background thread."""
|
||||
try:
|
||||
for line in iter(proc.stdout.readline, ""):
|
||||
with lock:
|
||||
output_lines.append(line.rstrip("\n"))
|
||||
proc.stdout.close()
|
||||
except ValueError:
|
||||
pass
|
||||
return_code[0] = proc.wait()
|
||||
build_done.set()
|
||||
reader = threading.Thread(target=_reader_thread, daemon=True)
|
||||
reader.start()
|
||||
output_top = 2
|
||||
output_height = max_y - 4
|
||||
canceled = False
|
||||
while not build_done.is_set():
|
||||
with lock:
|
||||
visible = output_lines[-(output_height - 1):] if output_lines else []
|
||||
for i, line in enumerate(visible):
|
||||
with contextlib.suppress(curses.error):
|
||||
self.stdscr.addstr(output_top + i, 2, line[:max_x - 4])
|
||||
with contextlib.suppress(curses.error):
|
||||
self.stdscr.addstr(max_y - 2, 2, " Building... (Esc to cancel) ",
|
||||
curses.color_pair(4))
|
||||
self.stdscr.refresh()
|
||||
self.stdscr.timeout(200)
|
||||
k = self.stdscr.getch()
|
||||
self.stdscr.timeout(-1)
|
||||
if k == 27: # Esc
|
||||
proc.terminate()
|
||||
canceled = True
|
||||
build_done.wait(timeout=5.0)
|
||||
if proc.poll() is None:
|
||||
proc.kill()
|
||||
build_done.wait(timeout=2.0)
|
||||
break
|
||||
if not canceled:
|
||||
build_done.wait()
|
||||
try:
|
||||
remaining = proc.stdout.read()
|
||||
if remaining:
|
||||
with lock:
|
||||
for line in remaining.rstrip("\n").split("\n"):
|
||||
if line:
|
||||
output_lines.append(line)
|
||||
except Exception:
|
||||
pass
|
||||
# ── Phase 4: Result summary ──
|
||||
self.stdscr.clear()
|
||||
ret = return_code[0]
|
||||
if ret == 0:
|
||||
title = f" ✅ BUILD COMPLETE — {bundle_label} / {tag} "
|
||||
elif canceled:
|
||||
title = f" ⏹ BUILD CANCELED — {bundle_label} / {tag} "
|
||||
else:
|
||||
title = f" ❌ BUILD FAILED — {bundle_label} / {tag} "
|
||||
self.stdscr.attron(curses.color_pair(1) | curses.A_BOLD)
|
||||
self.stdscr.addstr(0, 0, title.center(max_x)[:max_x])
|
||||
self.stdscr.attroff(curses.color_pair(1) | curses.A_BOLD)
|
||||
if ret == 0:
|
||||
self.stdscr.addstr(2, 3, "Build completed successfully!",
|
||||
curses.color_pair(2) | curses.A_BOLD)
|
||||
self.stdscr.addstr(3, 3, "Output: dist/docker/",
|
||||
curses.color_pair(2))
|
||||
elif canceled:
|
||||
self.stdscr.addstr(2, 3, "Build canceled by user.",
|
||||
curses.color_pair(4) | curses.A_BOLD)
|
||||
else:
|
||||
self.stdscr.addstr(2, 3, f"Build FAILED (exit code: {ret})",
|
||||
curses.color_pair(3) | curses.A_BOLD)
|
||||
self.stdscr.addstr(5, 3, "Last output:", curses.A_BOLD)
|
||||
with lock:
|
||||
tail = output_lines[-12:] if len(output_lines) > 12 else output_lines
|
||||
for i, line in enumerate(tail):
|
||||
with contextlib.suppress(curses.error):
|
||||
self.stdscr.addstr(6 + i, 3, line[:max_x - 6])
|
||||
self.stdscr.addstr(max_y - 1, 0, " Press Esc to return ".center(max_x)[:max_x],
|
||||
curses.color_pair(1))
|
||||
self.stdscr.refresh()
|
||||
while self.stdscr.getch() != 27:
|
||||
pass
|
||||
# #endregion bundle_build_mode
|
||||
def refresh_overview(self):
|
||||
if not self.candidate_id:
|
||||
self.overview = {}
|
||||
return
|
||||
self.overview = self.facade.get_overview(candidate_id=self.candidate_id)
|
||||
|
||||
def refresh_screen(self):
|
||||
max_y, max_x = self.stdscr.getmaxyx()
|
||||
self.stdscr.clear()
|
||||
@@ -634,13 +735,15 @@ class CleanReleaseTUI:
|
||||
except Exception:
|
||||
pass
|
||||
self.stdscr.refresh()
|
||||
|
||||
def loop(self):
|
||||
self.refresh_screen()
|
||||
while True:
|
||||
char = self.stdscr.getch()
|
||||
if char == curses.KEY_F10:
|
||||
break
|
||||
elif char == curses.KEY_F4:
|
||||
self.bundle_build_mode()
|
||||
self.refresh_screen()
|
||||
elif char == curses.KEY_F5:
|
||||
self.run_checks()
|
||||
elif char == curses.KEY_F6:
|
||||
@@ -651,17 +754,11 @@ class CleanReleaseTUI:
|
||||
self.approve_latest()
|
||||
elif char == curses.KEY_F9:
|
||||
self.publish_latest()
|
||||
|
||||
|
||||
# #endregion CleanReleaseTUI
|
||||
|
||||
|
||||
def tui_main(stdscr: curses.window):
|
||||
curses.curs_set(0) # Hide cursor
|
||||
app = CleanReleaseTUI(stdscr)
|
||||
app.loop()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
seed_trace_id()
|
||||
# TUI requires interactive terminal; headless mode must use CLI/API flow.
|
||||
@@ -677,8 +774,6 @@ def main() -> int:
|
||||
except Exception as e:
|
||||
print(f"Error starting TUI: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
# #endregion CleanReleaseTuiScript
|
||||
|
||||
Reference in New Issue
Block a user