diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..458a167 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,38 @@ +name: CI + +on: + pull_request: + push: + branches: + - main + +jobs: + test: + name: Test and lint + runs-on: ubuntu-latest + + steps: + - name: Check out repository + uses: actions/checkout@v4 + + - name: Install uv + uses: astral-sh/setup-uv@v5 + with: + enable-cache: true + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version-file: .python-version + + - name: Install dependencies + run: uv sync --dev --locked + + - name: Run tests + run: uv run pytest + + - name: Run ruff lint + run: uv run ruff check . + + - name: Check ruff formatting + run: uv run ruff format --check . diff --git a/.gitignore b/.gitignore index ffeec89..2528009 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,4 @@ data/ logs/ .superpowers/ .codex +.worktrees/ diff --git a/app/core/logs.py b/app/core/logs.py index 773342b..72deaaf 100644 --- a/app/core/logs.py +++ b/app/core/logs.py @@ -40,6 +40,18 @@ def read_log_chunk(path: Path, before: int | None = None, limit: int = 200) -> L return LogChunk(_decode(buffer), None, False) +def read_log_append(path: Path, offset: int = 0) -> dict[str, object]: + if not path.exists(): + return {"text": "", "offset": 0} + + file_size = path.stat().st_size + start = max(0, min(offset, file_size)) + with path.open("rb") as log_file: + log_file.seek(start) + data = log_file.read() + return {"text": _decode(data), "offset": file_size} + + def _chunk_start(buffer: bytearray, limit: int) -> int | None: if not buffer: return 0 diff --git a/app/core/retention.py b/app/core/retention.py index c217846..e5a9b74 100644 --- a/app/core/retention.py +++ b/app/core/retention.py @@ -12,12 +12,18 @@ def prune_logs( ) -> int: if keep_days <= 0: return 0 - current_time = now or datetime.now(UTC) + current_time = _as_utc(now or datetime.now(UTC)) cutoff = current_time - timedelta(days=keep_days) deleted = 0 for path, created_at in log_records: - if created_at >= cutoff or not path.exists(): + if _as_utc(created_at) >= cutoff or not path.exists(): continue path.unlink() deleted += 1 return deleted + + +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) diff --git a/app/core/runner.py b/app/core/runner.py index 3772f43..419001c 100644 --- a/app/core/runner.py +++ b/app/core/runner.py @@ -1,13 +1,16 @@ from __future__ import annotations import asyncio +import json import os from collections.abc import Awaitable, Callable +from contextlib import suppress from datetime import datetime from pathlib import Path from app.core.commands import build_rclone_argv from app.core.models import Job, JobRunResult, JobStep, StepRunResult, utc_now +from app.db import JobRunRecord, JobStepRunRecord Executor = Callable[[list[str], dict[str, str], Path], Awaitable[int]] @@ -23,11 +26,20 @@ async def subprocess_executor(argv: list[str], env: dict[str, str], log_path: Pa env=merged_env, ) assert process.stdout is not None - with log_path.open("wb") as log_file: - async for chunk in process.stdout: - log_file.write(chunk) - log_file.flush() - return await process.wait() + try: + with log_path.open("wb") as log_file: + async for chunk in process.stdout: + log_file.write(chunk) + log_file.flush() + return await process.wait() + except asyncio.CancelledError: + process.terminate() + try: + await asyncio.wait_for(process.wait(), timeout=5) + except TimeoutError: + process.kill() + await process.wait() + raise class JobRunner: @@ -93,6 +105,231 @@ async def run_job( ) +class LiveJobRunner: + def __init__( + self, log_root: Path, session_factory, executor: Executor = subprocess_executor + ) -> None: + self._log_root = log_root + self._session_factory = session_factory + self._executor = executor + self._active_job_ids: set[int] = set() + self._tasks: dict[int, asyncio.Task[None]] = {} + + def start_job( + self, job: Job, trigger: str, dry_run: bool = False, step_id: int | None = None + ) -> JobStepRunRecord | None: + now = utc_now() + selected_steps = _selected_steps(job, step_id) + if job.id in self._active_job_ids or not selected_steps: + status = "skipped" if job.id in self._active_job_ids else "success" + with self._session_factory() as db: + db.add( + JobRunRecord( + job_id=job.id, + job_name=job.name, + trigger=trigger, + status=status, + started_at=now, + ended_at=now, + ) + ) + db.commit() + return None + + self._active_job_ids.add(job.id) + run_stamp = _stamp(now) + first_step = selected_steps[0] + first_argv = build_rclone_argv(first_step.command, _common_args(job, dry_run), job.env) + first_log_path = self._log_root / f"job-{job.id}" / f"{run_stamp}-step-{first_step.id}.log" + first_log_path.parent.mkdir(parents=True, exist_ok=True) + first_log_path.touch(exist_ok=True) + + with self._session_factory() as db: + job_run = JobRunRecord( + job_id=job.id, + job_name=job.name, + trigger=trigger, + status="running", + started_at=now, + ended_at=None, + ) + db.add(job_run) + db.flush() + step_run = JobStepRunRecord( + job_run_id=job_run.id, + step_id=first_step.id, + step_name=first_step.name, + argv_json=json.dumps(first_argv), + status="running", + exit_code=None, + log_path=str(first_log_path), + started_at=now, + ended_at=None, + ) + db.add(step_run) + db.commit() + db.refresh(step_run) + job_run_id = job_run.id + step_run_id = step_run.id + step_run_job_run_id = step_run.job_run_id + step_run_step_id = step_run.step_id + step_run_step_name = step_run.step_name + step_run_argv_json = step_run.argv_json + step_run_status = step_run.status + step_run_exit_code = step_run.exit_code + step_run_log_path = step_run.log_path + step_run_started_at = step_run.started_at + step_run_ended_at = step_run.ended_at + db.expunge(step_run) + + task = asyncio.create_task( + self._run_job_background(job_run_id, job, dry_run, selected_steps, run_stamp) + ) + self._tasks[job_run_id] = task + task.add_done_callback(lambda completed: self._task_done(job.id, job_run_id, completed)) + return JobStepRunRecord( + id=step_run_id, + job_run_id=step_run_job_run_id, + step_id=step_run_step_id, + step_name=step_run_step_name, + argv_json=step_run_argv_json, + status=step_run_status, + exit_code=step_run_exit_code, + log_path=step_run_log_path, + started_at=step_run_started_at, + ended_at=step_run_ended_at, + ) + + def cancel_job_run(self, job_run_id: int) -> bool: + task = self._tasks.get(job_run_id) + if task is None or task.done(): + return False + task.cancel() + return True + + async def wait_for_job_run(self, job_run_id: int) -> None: + task = self._tasks.get(job_run_id) + if task is None: + return + with suppress(asyncio.CancelledError): + await task + + async def _run_job_background( + self, + job_run_id: int, + job: Job, + dry_run: bool, + selected_steps: list[JobStep], + run_stamp: str, + ) -> None: + job_status = "success" + step_run: JobStepRunRecord | None = None + try: + for index, step in enumerate(selected_steps): + if index == 0: + step_run = self._get_first_step_run(job_run_id) + else: + step_run = self._create_step_run(job_run_id, job, step, dry_run, run_stamp) + + exit_code = await self._executor( + json.loads(step_run.argv_json), job.env, Path(step_run.log_path) + ) + step_status = "success" if exit_code == 0 else "failed" + self._finish_step_run(step_run.id, step_status, exit_code) + if exit_code != 0: + job_status = "failed" + break + except asyncio.CancelledError: + job_status = "canceled" + self._cancel_running_step(job_run_id) + raise + except Exception: + job_status = "failed" + if step_run is not None: + self._finish_step_run(step_run.id, "failed", None) + finally: + self._finish_job_run(job_run_id, job_status) + + def _get_first_step_run(self, job_run_id: int) -> JobStepRunRecord: + with self._session_factory() as db: + step_run = ( + db.query(JobStepRunRecord) + .filter_by(job_run_id=job_run_id) + .order_by(JobStepRunRecord.id) + .first() + ) + if step_run is None: + raise RuntimeError(f"missing first step run for job run {job_run_id}") + db.expunge(step_run) + return step_run + + def _create_step_run( + self, job_run_id: int, job: Job, step: JobStep, dry_run: bool, run_stamp: str + ) -> JobStepRunRecord: + now = utc_now() + argv = build_rclone_argv(step.command, _common_args(job, dry_run), job.env) + log_path = self._log_root / f"job-{job.id}" / f"{run_stamp}-step-{step.id}.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.touch(exist_ok=True) + with self._session_factory() as db: + step_run = JobStepRunRecord( + job_run_id=job_run_id, + step_id=step.id, + step_name=step.name, + argv_json=json.dumps(argv), + status="running", + exit_code=None, + log_path=str(log_path), + started_at=now, + ended_at=None, + ) + db.add(step_run) + db.commit() + db.refresh(step_run) + db.expunge(step_run) + return step_run + + def _finish_step_run(self, step_run_id: int, status: str, exit_code: int | None) -> None: + with self._session_factory() as db: + step_run = db.get(JobStepRunRecord, step_run_id) + if step_run is None or step_run.status != "running": + return + step_run.status = status + step_run.exit_code = exit_code + step_run.ended_at = utc_now() + db.commit() + + def _cancel_running_step(self, job_run_id: int) -> None: + with self._session_factory() as db: + step_run = ( + db.query(JobStepRunRecord) + .filter_by(job_run_id=job_run_id, status="running") + .order_by(JobStepRunRecord.started_at.desc()) + .first() + ) + if step_run is None: + return + step_run.status = "canceled" + step_run.exit_code = None + step_run.ended_at = utc_now() + db.commit() + + def _finish_job_run(self, job_run_id: int, status: str) -> None: + with self._session_factory() as db: + job_run = db.get(JobRunRecord, job_run_id) + if job_run is None or job_run.status != "running": + return + job_run.status = status + job_run.ended_at = utc_now() + db.commit() + + def _task_done(self, job_id: int, job_run_id: int, task: asyncio.Task[None]) -> None: + self._active_job_ids.discard(job_id) + self._tasks.pop(job_run_id, None) + if not task.cancelled(): + task.exception() + + def _stamp(value: datetime) -> str: return value.strftime("%Y%m%dT%H%M%SZ") diff --git a/app/db.py b/app/db.py index eab0f0d..088c7d5 100644 --- a/app/db.py +++ b/app/db.py @@ -6,7 +6,7 @@ from typing import Annotated from fastapi import Depends -from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, create_engine +from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, create_engine, inspect from sqlalchemy.orm import ( DeclarativeBase, Mapped, @@ -60,7 +60,7 @@ class JobRunRecord(Base): trigger: Mapped[str] = mapped_column(String(40), nullable=False) status: Mapped[str] = mapped_column(String(40), nullable=False) started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) - ended_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) step_runs: Mapped[list[JobStepRunRecord]] = relationship( back_populates="job_run", cascade="all, delete-orphan" ) @@ -74,10 +74,11 @@ class JobStepRunRecord(Base): step_id: Mapped[int | None] = mapped_column(ForeignKey("job_steps.id"), nullable=True) step_name: Mapped[str] = mapped_column(String(200), nullable=False) argv_json: Mapped[str] = mapped_column(Text, nullable=False) - exit_code: Mapped[int] = mapped_column(Integer, nullable=False) + status: Mapped[str] = mapped_column(String(40), nullable=False, default="success") + exit_code: Mapped[int | None] = mapped_column(Integer, nullable=True) log_path: Mapped[str] = mapped_column(Text, nullable=False) started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) - ended_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) job_run: Mapped[JobRunRecord] = relationship(back_populates="step_runs") @@ -85,12 +86,13 @@ class ConsoleRunRecord(Base): __tablename__ = "console_runs" id: Mapped[int] = mapped_column(Integer, primary_key=True) + status: Mapped[str] = mapped_column(String(40), nullable=False, default="success") command: Mapped[str] = mapped_column(Text, nullable=False) argv_json: Mapped[str] = mapped_column(Text, nullable=False) - exit_code: Mapped[int] = mapped_column(Integer, nullable=False) + exit_code: Mapped[int | None] = mapped_column(Integer, nullable=True) log_path: Mapped[str] = mapped_column(Text, nullable=False) started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) - ended_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) class SettingRecord(Base): @@ -108,6 +110,95 @@ def init_db() -> None: settings.data_dir.mkdir(parents=True, exist_ok=True) settings.log_dir.mkdir(parents=True, exist_ok=True) Base.metadata.create_all(bind=engine) + _migrate_lifecycle_columns() + + +def _migrate_lifecycle_columns() -> None: + inspector = inspect(engine) + existing_tables = set(inspector.get_table_names()) + if engine.dialect.name != "sqlite": + return + rebuild_tables = [ + table + for table in ("job_runs", "job_step_runs", "console_runs") + if table in existing_tables and _needs_lifecycle_rebuild(inspector, table) + ] + if "job_runs" in rebuild_tables and "job_step_runs" in existing_tables: + rebuild_tables = [table for table in rebuild_tables if table != "job_step_runs"] + rebuild_tables.insert(1, "job_step_runs") + if not rebuild_tables: + return + + with engine.begin() as connection: + connection.exec_driver_sql("PRAGMA foreign_keys=OFF") + try: + for table in rebuild_tables: + connection.exec_driver_sql(f"ALTER TABLE {table} RENAME TO {table}_old") + for table in rebuild_tables: + Base.metadata.tables[table].create(bind=connection) + for table in rebuild_tables: + _copy_lifecycle_rows(connection, inspector, table) + connection.exec_driver_sql(f"DROP TABLE {table}_old") + finally: + connection.exec_driver_sql("PRAGMA foreign_keys=ON") + + +def _needs_lifecycle_rebuild(inspector, table: str) -> bool: + columns = {column["name"]: column for column in inspector.get_columns(table)} + if table == "job_runs": + return bool(columns.get("ended_at", {}).get("nullable") is False) + if table in {"job_step_runs", "console_runs"}: + return ( + "status" not in columns + or columns.get("exit_code", {}).get("nullable") is False + or columns.get("ended_at", {}).get("nullable") is False + ) + return False + + +def _copy_lifecycle_rows(connection, inspector, table: str) -> None: + old_columns = {column["name"] for column in inspector.get_columns(f"{table}_old")} + status_expr = _lifecycle_status_expr(old_columns) + if table == "job_runs": + connection.exec_driver_sql( + """ + INSERT INTO job_runs (id, job_id, job_name, trigger, status, started_at, ended_at) + SELECT id, job_id, job_name, trigger, status, started_at, ended_at + FROM job_runs_old + """ + ) + elif table == "job_step_runs": + connection.exec_driver_sql( + f""" + INSERT INTO job_step_runs ( + id, job_run_id, step_id, step_name, argv_json, status, exit_code, log_path, + started_at, ended_at + ) + SELECT + id, job_run_id, step_id, step_name, argv_json, {status_expr}, exit_code, log_path, + started_at, ended_at + FROM job_step_runs_old + """ + ) + elif table == "console_runs": + connection.exec_driver_sql( + f""" + INSERT INTO console_runs ( + id, status, command, argv_json, exit_code, log_path, started_at, ended_at + ) + SELECT + id, {status_expr}, command, argv_json, exit_code, log_path, started_at, ended_at + FROM console_runs_old + """ + ) + + +def _lifecycle_status_expr(old_columns: set[str]) -> str: + if "status" in old_columns: + return "status" + if "exit_code" in old_columns: + return "CASE WHEN exit_code = 0 THEN 'success' ELSE 'failed' END" + return "'success'" def get_db() -> Session: diff --git a/app/main.py b/app/main.py index f63c5d7..3d1c076 100644 --- a/app/main.py +++ b/app/main.py @@ -4,7 +4,9 @@ import json from contextlib import suppress from datetime import UTC, datetime +from math import ceil from pathlib import Path +from urllib.parse import urlencode from zoneinfo import ZoneInfo from fastapi import FastAPI, Form, Request, WebSocket @@ -17,7 +19,7 @@ from app.auth import SESSION_KEY, AuthRequired, verify_password from app.config import settings from app.core.commands import CommandPolicyError, build_rclone_argv, parse_console_command -from app.core.logs import read_log_chunk +from app.core.logs import read_log_append, read_log_chunk from app.core.models import utc_now from app.core.pty_console import bridge_pty, display_argv from app.core.retention import prune_logs @@ -26,6 +28,7 @@ ConsoleRunRecord, DbSession, JobRecord, + JobRunRecord, JobStepRecord, JobStepRunRecord, SessionLocal, @@ -33,7 +36,6 @@ init_db, parse_env_lines, record_to_job, - save_job_run, ) from app.runner_service import runner from app.scheduler import scheduler, sync_schedules @@ -54,8 +56,11 @@ def run_mode_label(trigger: str) -> str: templates.env.globals["run_mode_label"] = run_mode_label templates.env.globals["format_local_time"] = lambda value: _format_local_time(value) templates.env.globals["format_duration"] = lambda start, end: _format_duration(start, end) +templates.env.globals["utc_now"] = utc_now +templates.env.globals["run_exit_label"] = lambda run: _exit_label(run.status, run.exit_code) LOG_CHUNK_LINES = 200 +HISTORY_PAGE_SIZE = 25 def create_app() -> FastAPI: @@ -115,7 +120,31 @@ async def index(_: AuthRequired) -> Response: @app.get("/jobs", response_class=HTMLResponse) async def jobs(request: Request, _: AuthRequired, db: DbSession) -> Response: records = db.query(JobRecord).order_by(JobRecord.name).all() - return templates.TemplateResponse(request, "jobs.html", {"jobs": _job_rows(records)}) + ongoing_job_runs = ( + db.query(JobRunRecord).filter_by(status="running").order_by(JobRunRecord.started_at).all() + ) + ongoing_step_runs = ( + db.query(JobStepRunRecord) + .filter_by(status="running") + .order_by(JobStepRunRecord.started_at) + .all() + ) + ongoing_console_runs = ( + db.query(ConsoleRunRecord) + .filter_by(status="running") + .order_by(ConsoleRunRecord.started_at) + .all() + ) + return templates.TemplateResponse( + request, + "jobs.html", + { + "jobs": _job_rows(records), + "ongoing_job_runs": _ongoing_job_run_rows(ongoing_job_runs), + "ongoing_step_runs": ongoing_step_runs, + "ongoing_console_runs": ongoing_console_runs, + }, + ) @app.get("/jobs/new", response_class=HTMLResponse) @@ -125,9 +154,9 @@ async def new_job(request: Request, _: AuthRequired) -> Response: "job_form.html", { "job": None, - "schedule_summary": cron_summary("0 2 * * *"), - "env_lines": "BW_LIMIT=8M", - "steps_text": "Sync Music|sync /media/Musiikki secret:/Musiikki", + "schedule_summary": cron_summary(""), + "env_lines": "", + "steps_text": "", "command_previews": [], }, ) @@ -164,6 +193,13 @@ async def job_detail(request: Request, job_id: int, _: AuthRequired, db: DbSessi job = db.get(JobRecord, job_id) if job is None: return RedirectResponse("/jobs", status_code=303) + job_runs = ( + db.query(JobRunRecord) + .filter_by(job_id=job_id) + .order_by(JobRunRecord.started_at.desc()) + .limit(20) + .all() + ) runs = ( db.query(JobStepRunRecord) .join(JobStepRunRecord.job_run) @@ -181,6 +217,7 @@ async def job_detail(request: Request, job_id: int, _: AuthRequired, db: DbSessi "env_lines": env_to_lines(job.env_json), "steps_text": _steps_to_text(job.steps), "command_previews": _command_previews(job), + "job_runs": job_runs, "runs": runs, }, ) @@ -239,50 +276,132 @@ async def _run_job( job = db.get(JobRecord, job_id) if job is None: return RedirectResponse("/jobs", status_code=303) - result = await runner.run_job( + first_step_run = runner.start_job( record_to_job(job), trigger=trigger, dry_run=dry_run, step_id=step_id ) - run_record = save_job_run(db, result) - first_step_run = db.query(JobStepRunRecord).filter_by(job_run_id=run_record.id).first() if first_step_run is None: return RedirectResponse("/runs", status_code=303) + if step_id is None: + return RedirectResponse(f"/job-runs/{first_step_run.job_run_id}", status_code=303) return RedirectResponse(f"/runs/{first_step_run.id}", status_code=303) @app.get("/runs", response_class=HTMLResponse) -async def runs(request: Request, _: AuthRequired, db: DbSession) -> Response: - job_runs = db.query(JobRecord).all() - recent = ( - db.query(JobStepRunRecord).order_by(JobStepRunRecord.started_at.desc()).limit(100).all() +async def runs( + request: Request, + _: AuthRequired, + db: DbSession, + job_page: int = 1, + step_page: int = 1, + console_page: int = 1, +) -> Response: + job_query = db.query(JobRunRecord).order_by(JobRunRecord.started_at.desc()) + step_query = db.query(JobStepRunRecord).order_by(JobStepRunRecord.started_at.desc()) + console_query = db.query(ConsoleRunRecord).order_by(ConsoleRunRecord.started_at.desc()) + job_runs, job_pagination = _paginated_history( + job_query, + page=job_page, + page_param="job_page", + other_pages={"step_page": step_page, "console_page": console_page}, + ) + recent, step_pagination = _paginated_history( + step_query, + page=step_page, + page_param="step_page", + other_pages={"job_page": job_page, "console_page": console_page}, ) - console_runs = ( - db.query(ConsoleRunRecord).order_by(ConsoleRunRecord.started_at.desc()).limit(50).all() + console_runs, console_pagination = _paginated_history( + console_query, + page=console_page, + page_param="console_page", + other_pages={"job_page": job_page, "step_page": step_page}, ) return templates.TemplateResponse( request, "runs.html", { - "jobs": job_runs, + "job_runs": job_runs, "step_runs": recent, "console_runs": console_runs, + "job_pagination": job_pagination, + "step_pagination": step_pagination, + "console_pagination": console_pagination, }, ) +@app.get("/job-runs/{job_run_id}", response_class=HTMLResponse) +async def job_run_detail( + request: Request, job_run_id: int, _: AuthRequired, db: DbSession +) -> Response: + job_run = db.get(JobRunRecord, job_run_id) + if job_run is None: + return RedirectResponse("/runs", status_code=303) + active_step_run = _active_step_run(job_run) + context = { + "job_run": job_run, + "step_rows": _job_run_step_rows(job_run, db), + "active_step_run": active_step_run, + "job_run_status_url": f"/job-runs/{job_run.id}/status", + "job_run_cancel_url": f"/job-runs/{job_run.id}/cancel", + } + if active_step_run is not None: + log_path = Path(active_step_run.log_path) + context |= { + "run": active_step_run, + "argv": json.loads(active_step_run.argv_json), + "log_chunk": read_log_chunk(log_path, limit=LOG_CHUNK_LINES), + "log_chunk_url": f"/runs/{active_step_run.id}/log/chunk", + "log_append_url": f"/runs/{active_step_run.id}/log/append", + "log_status_url": f"/runs/{active_step_run.id}/status", + "log_append_offset": _log_size(log_path), + "raw_log_url": f"/runs/{active_step_run.id}/log/raw", + } + return templates.TemplateResponse(request, "job_run_detail.html", context) + + +@app.get("/job-runs/{job_run_id}/status") +async def job_run_status(job_run_id: int, _: AuthRequired, db: DbSession) -> dict[str, object]: + job_run = db.get(JobRunRecord, job_run_id) + if job_run is None: + return {"status": "missing", "can_cancel": False, "steps": []} + return _job_run_status_payload(job_run, db) + + +@app.post("/job-runs/{job_run_id}/cancel") +async def cancel_job_run(job_run_id: int, _: AuthRequired, db: DbSession) -> Response: + if not runner.cancel_job_run(job_run_id): + _mark_job_run_canceled(db, job_run_id) + return RedirectResponse(f"/job-runs/{job_run_id}", status_code=303) + + +@app.post("/job-runs/{job_run_id}/delete") +async def delete_job_run(job_run_id: int, _: AuthRequired, db: DbSession) -> Response: + job_run = db.get(JobRunRecord, job_run_id) + if job_run is not None: + _delete_job_run(db, job_run) + return RedirectResponse("/runs", status_code=303) + + @app.get("/runs/{run_id}", response_class=HTMLResponse) async def run_detail(request: Request, run_id: int, _: AuthRequired, db: DbSession) -> Response: run = db.get(JobStepRunRecord, run_id) if run is None: return RedirectResponse("/runs", status_code=303) log_chunk = read_log_chunk(Path(run.log_path), limit=LOG_CHUNK_LINES) + log_size = _log_size(Path(run.log_path)) return templates.TemplateResponse( request, "run_detail.html", { "run": run, "argv": json.loads(run.argv_json), + "job_run_step_count": len(run.job_run.step_runs), "log_chunk": log_chunk, "log_chunk_url": f"/runs/{run.id}/log/chunk", + "log_append_url": f"/runs/{run.id}/log/append", + "log_status_url": f"/runs/{run.id}/status", + "log_append_offset": log_size, "raw_log_url": f"/runs/{run.id}/log/raw", }, ) @@ -298,6 +417,40 @@ async def run_log_chunk( return _log_chunk_payload(Path(run.log_path), before) +@app.get("/runs/{run_id}/status") +async def run_status(run_id: int, _: AuthRequired, db: DbSession) -> dict[str, object]: + run = db.get(JobStepRunRecord, run_id) + if run is None: + return {"status": "missing", "can_cancel": False} + return _run_status_payload(run) + + +@app.get("/runs/{run_id}/log/append") +async def run_log_append( + run_id: int, _: AuthRequired, db: DbSession, offset: int = 0 +) -> dict[str, object]: + run = db.get(JobStepRunRecord, run_id) + if run is None: + return {"text": "", "offset": 0} + return read_log_append(Path(run.log_path), offset) + + +@app.post("/runs/{run_id}/cancel") +async def cancel_run(run_id: int, _: AuthRequired, db: DbSession) -> Response: + run = db.get(JobStepRunRecord, run_id) + if run is not None and not runner.cancel_job_run(run.job_run_id): + _mark_job_run_canceled(db, run.job_run_id) + return RedirectResponse(f"/runs/{run_id}", status_code=303) + + +@app.post("/runs/{run_id}/delete") +async def delete_run(run_id: int, _: AuthRequired, db: DbSession) -> Response: + run = db.get(JobStepRunRecord, run_id) + if run is not None: + _delete_step_run(db, run) + return RedirectResponse("/runs", status_code=303) + + @app.get("/runs/{run_id}/log/raw") async def raw_run_log(run_id: int, _: AuthRequired, db: DbSession) -> Response: run = db.get(JobStepRunRecord, run_id) @@ -327,6 +480,7 @@ async def console_run_detail( if run is None: return RedirectResponse("/console", status_code=303) log_chunk = read_log_chunk(Path(run.log_path), limit=LOG_CHUNK_LINES) + log_size = _log_size(Path(run.log_path)) return templates.TemplateResponse( request, "console_run_detail.html", @@ -335,6 +489,9 @@ async def console_run_detail( "argv": json.loads(run.argv_json), "log_chunk": log_chunk, "log_chunk_url": f"/console/runs/{run.id}/log/chunk", + "log_append_url": "", + "log_status_url": "", + "log_append_offset": log_size, "raw_log_url": f"/console/runs/{run.id}/log/raw", }, ) @@ -358,6 +515,14 @@ async def raw_console_log(run_id: int, _: AuthRequired, db: DbSession) -> Respon return _raw_log_response(Path(run.log_path)) +@app.post("/console/runs/{run_id}/delete") +async def delete_console_run(run_id: int, _: AuthRequired, db: DbSession) -> Response: + run = db.get(ConsoleRunRecord, run_id) + if run is not None: + _delete_console_run(db, run) + return RedirectResponse("/runs", status_code=303) + + @app.websocket("/console/terminal") async def console_terminal(websocket: WebSocket) -> None: if websocket.session.get(SESSION_KEY) is not True: @@ -390,7 +555,10 @@ async def start_command(command: str) -> None: started_at = utc_now() log_path = _console_log_path(started_at) log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.touch(exist_ok=True) input_queue = asyncio.Queue() + with SessionLocal() as db: + run_id = _start_console_run_record(db, command, argv, log_path, started_at) def write_log(text: str) -> None: with log_path.open("a", encoding="utf-8") as log_file: @@ -404,28 +572,29 @@ async def run() -> None: nonlocal command_task, input_queue await send_state("running") write_log(f"$ {display_argv(argv)}\r\n") - exit_code = await bridge_pty(argv, input_queue.get, send_and_log) - ended_at = utc_now() - exit_line = f"\r\n[process exited with code {exit_code}]\r\n" - write_log(exit_line) - await send_output(exit_line) - with SessionLocal() as db: - record = ConsoleRunRecord( - command=command, - argv_json=json.dumps(argv), - exit_code=exit_code, - log_path=str(log_path), - started_at=started_at, - ended_at=ended_at, - ) - db.add(record) - db.commit() - db.refresh(record) - await websocket.send_json({"type": "history", "run": _console_history_row(record)}) - command_task = None - input_queue = None - await send_output("rclone> ") - await send_state("idle") + try: + exit_code = await bridge_pty(argv, input_queue.get, send_and_log) + ended_at = utc_now() + exit_line = f"\r\n[process exited with code {exit_code}]\r\n" + write_log(exit_line) + await send_output(exit_line) + with SessionLocal() as db: + record = _finish_console_run_record(db, run_id, exit_code, ended_at) + if record is not None: + await websocket.send_json( + {"type": "history", "run": _console_history_row(record)} + ) + await send_output("rclone> ") + await send_state("idle") + except asyncio.CancelledError: + ended_at = utc_now() + write_log("\r\n[process canceled]\r\n") + with SessionLocal() as db: + _cancel_console_run_record(db, run_id, ended_at) + raise + finally: + command_task = None + input_queue = None command_task = asyncio.create_task(run()) @@ -442,10 +611,18 @@ async def run() -> None: except WebSocketDisconnect, RuntimeError, json.JSONDecodeError: if command_task is not None and not command_task.done(): command_task.cancel() + with suppress(asyncio.CancelledError): + await command_task @app.get("/settings", response_class=HTMLResponse) -async def settings_page(request: Request, _: AuthRequired, db: DbSession) -> Response: +async def settings_page( + request: Request, + _: AuthRequired, + db: DbSession, + pruned: int | None = None, + cleared: int | None = None, +) -> Response: count = _prunable_logs(db) return templates.TemplateResponse( request, @@ -453,14 +630,22 @@ async def settings_page(request: Request, _: AuthRequired, db: DbSession) -> Res { "settings": settings, "known_logs": len(count), + "pruned": pruned, + "cleared": cleared, }, ) @app.post("/settings/prune") async def prune(_: AuthRequired, db: DbSession) -> Response: - prune_logs(_prunable_logs(db), keep_days=settings.retention_days) - return RedirectResponse("/settings", status_code=303) + deleted = prune_logs(_prunable_logs(db), keep_days=settings.retention_days) + return RedirectResponse(f"/settings?pruned={deleted}", status_code=303) + + +@app.post("/settings/clear-history") +async def clear_history(_: AuthRequired, db: DbSession) -> Response: + cleared = _clear_history(db) + return RedirectResponse(f"/settings?cleared={cleared}", status_code=303) def _replace_steps(job: JobRecord, steps_text: str) -> None: @@ -516,10 +701,153 @@ def _prunable_logs(db) -> list[tuple[Path, datetime]]: return step_logs + console_logs +def _clear_history(db) -> int: + logs = _prunable_logs(db) + deleted_records = db.query(JobRunRecord).count() + db.query(ConsoleRunRecord).count() + for path, _started_at in logs: + if path.exists(): + path.unlink() + db.query(ConsoleRunRecord).delete(synchronize_session=False) + db.query(JobStepRunRecord).delete(synchronize_session=False) + db.query(JobRunRecord).delete(synchronize_session=False) + db.commit() + return deleted_records + + +def _mark_job_run_canceled(db, job_run_id: int) -> None: + now = utc_now() + job_run = db.get(JobRunRecord, job_run_id) + if job_run is None or job_run.status != "running": + return + job_run.status = "canceled" + job_run.ended_at = now + for step_run in job_run.step_runs: + if step_run.status == "running": + step_run.status = "canceled" + step_run.exit_code = None + step_run.ended_at = now + db.commit() + + +def _delete_job_run(db, job_run: JobRunRecord) -> None: + for step_run in list(job_run.step_runs): + _delete_log_file(Path(step_run.log_path)) + db.delete(job_run) + db.commit() + + +def _delete_step_run(db, step_run: JobStepRunRecord) -> None: + _delete_log_file(Path(step_run.log_path)) + db.delete(step_run) + db.commit() + + +def _delete_console_run(db, run: ConsoleRunRecord) -> None: + _delete_log_file(Path(run.log_path)) + db.delete(run) + db.commit() + + +def _delete_log_file(path: Path) -> None: + if path.exists(): + path.unlink() + + def _console_history_rows(records: list[ConsoleRunRecord]) -> list[dict[str, object]]: return [_console_history_row(record) for record in records] +def _paginated_history(query, page: int, page_param: str, other_pages: dict[str, int]): + current_page = max(1, page) + total_items = query.count() + total_pages = max(1, ceil(total_items / HISTORY_PAGE_SIZE)) + offset = (current_page - 1) * HISTORY_PAGE_SIZE + items = query.offset(offset).limit(HISTORY_PAGE_SIZE).all() + has_next = current_page < total_pages + pagination = { + "page": current_page, + "total_pages": total_pages, + "has_previous": current_page > 1, + "has_next": has_next, + "previous_url": _history_page_url( + page_param, current_page - 1, other_pages, _history_section_id(page_param) + ) + if current_page > 1 + else None, + "next_url": _history_page_url( + page_param, current_page + 1, other_pages, _history_section_id(page_param) + ) + if has_next + else None, + "target": _history_section_id(page_param), + } + return items, pagination + + +def _history_page_url( + page_param: str, page: int, other_pages: dict[str, int], section_id: str | None = None +) -> str: + params = {page_param: max(1, page)} + for name, value in other_pages.items(): + params[name] = max(1, value) + url = "/runs?" + urlencode(params) + if section_id is not None: + url += f"#{section_id}" + return url + + +def _history_section_id(page_param: str) -> str: + return { + "job_page": "job-runs-section", + "step_page": "step-runs-section", + "console_page": "console-runs-section", + }[page_param] + + +def _start_console_run_record( + db, command: str, argv: list[str], log_path: Path, started_at: datetime +) -> int: + record = ConsoleRunRecord( + status="running", + command=command, + argv_json=json.dumps(argv), + exit_code=None, + log_path=str(log_path), + started_at=started_at, + ended_at=None, + ) + db.add(record) + db.commit() + db.refresh(record) + return record.id + + +def _finish_console_run_record( + db, run_id: int, exit_code: int, ended_at: datetime +) -> ConsoleRunRecord | None: + record = db.get(ConsoleRunRecord, run_id) + if record is None: + return None + record.status = "success" if exit_code == 0 else "failed" + record.exit_code = exit_code + record.ended_at = ended_at + db.commit() + db.refresh(record) + return record + + +def _cancel_console_run_record(db, run_id: int, ended_at: datetime) -> ConsoleRunRecord | None: + record = db.get(ConsoleRunRecord, run_id) + if record is None: + return None + record.status = "canceled" + record.exit_code = None + record.ended_at = ended_at + db.commit() + db.refresh(record) + return record + + def _console_history_row(record: ConsoleRunRecord) -> dict[str, object]: return { "id": record.id, @@ -529,6 +857,126 @@ def _console_history_row(record: ConsoleRunRecord) -> dict[str, object]: } +def _ongoing_job_run_rows(records: list[JobRunRecord]) -> list[dict[str, object]]: + return [ + { + "run": record, + "current_step": active_step.step_name if active_step is not None else "Starting", + } + for record in records + for active_step in [_active_step_run(record)] + ] + + +def _job_run_status_payload(job_run: JobRunRecord, db) -> dict[str, object]: + finished_at = _format_local_time(job_run.ended_at) if job_run.ended_at is not None else None + end = job_run.ended_at or utc_now() + elapsed_seconds = _duration_seconds(job_run.started_at, end) + active_step_run = _active_step_run(job_run) + return { + "status": job_run.status, + "started_at": _format_local_time(job_run.started_at), + "finished_at": finished_at, + "elapsed": _format_seconds(elapsed_seconds), + "elapsed_seconds": elapsed_seconds, + "can_cancel": job_run.status == "running", + "active_step_run_id": active_step_run.id if active_step_run is not None else None, + "steps": [_job_run_step_payload(row) for row in _job_run_step_rows(job_run, db)], + } + + +def _job_run_step_payload(row: dict[str, object]) -> dict[str, object]: + step_run = row["run"] + status = str(row["status"]) + return { + "key": row["key"], + "name": row["name"], + "status": status, + "run_id": step_run.id if step_run is not None else None, + "exit_label": _exit_label(status, step_run.exit_code if step_run is not None else None), + } + + +def _job_run_step_rows(job_run: JobRunRecord, db) -> list[dict[str, object]]: + step_runs = sorted(job_run.step_runs, key=lambda item: item.started_at) + step_run_by_step_id = { + step_run.step_id: step_run for step_run in step_runs if step_run.step_id is not None + } + if job_run.job_id is not None and "step" not in job_run.trigger: + job = db.get(JobRecord, job_run.job_id) + if job is not None: + rows = [] + for step in job.steps: + step_run = step_run_by_step_id.get(step.id) + status = ( + step_run.status if step_run is not None else _unstarted_step_status(job_run) + ) + rows.append( + { + "key": f"step-{step.id}", + "name": step.name, + "status": status, + "run": step_run, + } + ) + return rows + + return [ + { + "key": f"step-{step_run.step_id}" + if step_run.step_id is not None + else f"run-{step_run.id}", + "name": step_run.step_name, + "status": step_run.status, + "run": step_run, + } + for step_run in step_runs + ] + + +def _active_step_run(job_run: JobRunRecord) -> JobStepRunRecord | None: + step_runs = sorted(job_run.step_runs, key=lambda item: item.started_at) + running = [step_run for step_run in step_runs if step_run.status == "running"] + if running: + return running[-1] + if step_runs: + return step_runs[-1] + return None + + +def _unstarted_step_status(job_run: JobRunRecord) -> str: + if job_run.status == "canceled": + return "canceled" + return "pending" + + +def _run_status_payload(run: JobStepRunRecord) -> dict[str, object]: + finished_at = _format_local_time(run.ended_at) if run.ended_at is not None else None + end = run.ended_at or utc_now() + elapsed_seconds = _duration_seconds(run.started_at, end) + return { + "status": run.status, + "job_status": run.job_run.status, + "exit_code": run.exit_code, + "exit_label": _exit_label(run.status, run.exit_code), + "started_at": _format_local_time(run.started_at), + "finished_at": finished_at, + "elapsed": _format_seconds(elapsed_seconds), + "elapsed_seconds": elapsed_seconds, + "can_cancel": run.status == "running", + } + + +def _exit_label(status: str, exit_code: int | None) -> str: + if exit_code is not None: + return str(exit_code) + if status == "canceled": + return "Canceled" + if status == "skipped": + return "Skipped" + return "Pending" + + def _console_log_path(started_at: datetime) -> Path: return settings.log_dir / "console" / f"{started_at.strftime('%Y%m%dT%H%M%S%fZ')}.log" @@ -541,7 +989,18 @@ def _format_local_time(value: datetime) -> str: def _format_duration(started_at: datetime, ended_at: datetime) -> str: - total_seconds = max(0.0, (ended_at - started_at).total_seconds()) + return _format_seconds(_duration_seconds(started_at, ended_at)) + + +def _duration_seconds(started_at: datetime, ended_at: datetime) -> float: + if started_at.tzinfo is None and ended_at.tzinfo is not None: + ended_at = ended_at.replace(tzinfo=None) + elif started_at.tzinfo is not None and ended_at.tzinfo is None: + started_at = started_at.replace(tzinfo=None) + return max(0.0, (ended_at - started_at).total_seconds()) + + +def _format_seconds(total_seconds: float) -> str: if total_seconds < 60: return f"{total_seconds:.1f}s" @@ -563,6 +1022,12 @@ def _log_chunk_payload(path: Path, before: int | None) -> dict[str, object]: } +def _log_size(path: Path) -> int: + if not path.exists(): + return 0 + return path.stat().st_size + + def _raw_log_response(path: Path) -> Response: if not path.exists(): return Response("Log file has been pruned or is not available.", status_code=404) diff --git a/app/runner_service.py b/app/runner_service.py index 6513443..a9c3a90 100644 --- a/app/runner_service.py +++ b/app/runner_service.py @@ -1,6 +1,7 @@ from __future__ import annotations from app.config import settings -from app.core.runner import JobRunner +from app.core.runner import LiveJobRunner +from app.db import SessionLocal -runner = JobRunner(settings.log_dir) +runner = LiveJobRunner(settings.log_dir, session_factory=SessionLocal) diff --git a/app/scheduler.py b/app/scheduler.py index 1be6f0f..3fd03b1 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -1,6 +1,5 @@ from __future__ import annotations -import asyncio from zoneinfo import ZoneInfo from apscheduler.schedulers.asyncio import AsyncIOScheduler @@ -9,7 +8,7 @@ from app.config import settings from app.core.schedule import normalize_cron -from app.db import JobRecord, SessionLocal, record_to_job, save_job_run +from app.db import JobRecord, SessionLocal, record_to_job from app.runner_service import runner scheduler = AsyncIOScheduler(timezone=ZoneInfo(settings.timezone)) @@ -33,7 +32,7 @@ def sync_schedules(session: Session) -> None: if not normalize_cron(record.cron): continue scheduler.add_job( - scheduled_run, + _run_and_record, cron_trigger(record.cron), args=[record.id], id=f"job-{record.id}", @@ -43,14 +42,9 @@ def sync_schedules(session: Session) -> None: ) -def scheduled_run(job_id: int) -> None: - asyncio.create_task(_run_and_record(job_id)) - - async def _run_and_record(job_id: int) -> None: with SessionLocal() as session: record = session.get(JobRecord, job_id) if record is None: return - result = await runner.run_job(record_to_job(record), trigger="schedule") - save_job_run(session, result) + runner.start_job(record_to_job(record), trigger="schedule") diff --git a/app/static/favicon.svg b/app/static/favicon.svg new file mode 100644 index 0000000..2fa2eb2 --- /dev/null +++ b/app/static/favicon.svg @@ -0,0 +1,6 @@ + + + + + + diff --git a/app/static/styles.css b/app/static/styles.css index aef4d8f..058ff6f 100644 --- a/app/static/styles.css +++ b/app/static/styles.css @@ -8,6 +8,62 @@ --accent: #2f6f4e; --accent-dark: #23543b; --danger: #a33a32; + --danger-dark: #7f2d27; + --secondary-hover: #f1f3ed; + --input-bg: #ffffff; + --badge-bg: #eef2ea; + --badge-dry-bg: #fff6dc; + --badge-dry-line: #dfc982; + --command-bg: #f1f3ed; + --copy-hover-bg: #ffffff; + --toast-bg: #eff8f1; + --toast-border: #b8d8bf; + --toast-text: #214a2d; + --toast-error-bg: #fff1ef; + --toast-error-border: #e2b5af; + --summary-primary-border: #c7d8ca; + --active-row-bg: #f3f8f1; + --log-bg: #151713; + --log-text: #f0f5eb; + --terminal-bg: #10120f; + --terminal-text: #eef6e8; + --notice-bg: #edf6ec; + --notice-border: #cbdcc9; + --notice-text: var(--accent-dark); +} + +html[data-theme="dark"] { + color-scheme: dark; + --bg: #111410; + --panel: #1a1f18; + --text: #edf2e9; + --muted: #a7b0a1; + --line: #343d31; + --accent: #5da97d; + --accent-dark: #8fd5aa; + --danger: #e06e65; + --danger-dark: #c85c54; + --secondary-hover: #232a20; + --input-bg: #121610; + --badge-bg: #232a20; + --badge-dry-bg: #332c18; + --badge-dry-line: #6f5c25; + --command-bg: #131711; + --copy-hover-bg: #242b21; + --toast-bg: #183323; + --toast-border: #315f42; + --toast-text: #dff3e4; + --toast-error-bg: #3a1d1a; + --toast-error-border: #73433d; + --summary-primary-border: #3d6649; + --active-row-bg: #18251b; + --log-bg: #080a08; + --log-text: #edf7e8; + --terminal-bg: #080a08; + --terminal-text: #edf7e8; + --notice-bg: #17261b; + --notice-border: #31553a; + --notice-text: #bde4c7; } * { @@ -36,11 +92,20 @@ a { } .brand { + align-items: center; color: var(--text); + display: inline-flex; + gap: 8px; font-weight: 700; text-decoration: none; } +.brand-mark { + display: block; + height: 28px; + width: 28px; +} + .topbar nav { display: flex; flex: 1; @@ -52,6 +117,16 @@ a { text-decoration: none; } +.topbar-actions { + align-items: center; + display: flex; + gap: 10px; +} + +.theme-toggle { + min-width: 104px; +} + .page { margin: 0 auto; max-width: 1120px; @@ -67,10 +142,15 @@ a { } h1, -h2 { +h2, +h3 { margin: 0 0 10px; } +h3 { + font-size: 18px; +} + p { color: var(--muted); margin-top: 0; @@ -94,6 +174,21 @@ button:hover, background: var(--accent-dark); } +button.danger, +.button.danger { + background: var(--danger); +} + +button.danger:hover, +.button.danger:hover { + background: var(--danger-dark); +} + +button.small, +.button.small { + padding: 6px 10px; +} + .secondary, .ghost { background: transparent; @@ -104,7 +199,7 @@ button:hover, button.secondary:hover, .button.secondary:hover, .ghost:hover { - background: #f1f3ed; + background: var(--secondary-hover); color: var(--text); } @@ -120,6 +215,42 @@ button.secondary:hover, overflow: auto; } +.table-wrap + h2, +.form-grid + h2, +.terminal-output + h2 { + margin-top: 28px; +} + +.table-wrap + h3 { + margin-top: 20px; +} + +.pagination { + align-items: center; + display: flex; + gap: 10px; + justify-content: flex-end; + margin-top: 10px; +} + +.pagination + h2 { + margin-top: 28px; +} + +.pagination-page { + color: var(--muted); +} + +.disabled { + cursor: default; + opacity: 0.55; + pointer-events: none; +} + +.section-heading { + margin-top: 28px; +} + table { border-collapse: collapse; width: 100%; @@ -130,7 +261,16 @@ td { border-bottom: 1px solid var(--line); padding: 12px 14px; text-align: left; - vertical-align: top; + vertical-align: middle; +} + +.action-column { + width: 96px; +} + +.action-cell { + text-align: right; + white-space: nowrap; } tr:last-child td { @@ -162,6 +302,7 @@ label { input, textarea { + background: var(--input-bg); border: 1px solid var(--line); border-radius: 6px; color: var(--text); @@ -203,7 +344,7 @@ textarea { } .badge { - background: #eef2ea; + background: var(--badge-bg); border: 1px solid var(--line); border-radius: 999px; color: var(--text); @@ -215,8 +356,8 @@ textarea { } .badge.dry-run { - background: #fff6dc; - border-color: #dfc982; + background: var(--badge-dry-bg); + border-color: var(--badge-dry-line); } .command-preview h2 { @@ -237,7 +378,7 @@ textarea { } .command-preview-box { - background: #f1f3ed; + background: var(--command-bg); border-radius: 6px; margin: 8px 0 0; position: relative; @@ -283,7 +424,7 @@ textarea { } .copy-command-button:hover { - background: #fff; + background: var(--copy-hover-bg); color: var(--text); } @@ -309,6 +450,73 @@ textarea { font-weight: 700; } +.toast { + align-items: center; + animation: toast-in 180ms cubic-bezier(0.2, 0.8, 0.2, 1); + background: var(--toast-bg); + border: 1px solid var(--toast-border); + border-radius: 8px; + box-shadow: 0 14px 36px rgb(0 0 0 / 16%); + color: var(--toast-text); + display: inline-flex; + gap: 10px; + font-weight: 700; + padding: 12px 16px; + position: fixed; + right: max(24px, calc((100vw - 1120px) / 2 + 24px)); + top: 82px; + transform-origin: top right; + transition: opacity 160ms ease, transform 160ms ease; + z-index: 20; +} + +.toast[hidden] { + display: none; +} + +.toast.leaving { + opacity: 0; + transform: translateY(-8px) scale(0.98); +} + +.toast-icon { + align-items: center; + background: var(--accent); + border-radius: 999px; + color: #fff; + display: inline-flex; + font-size: 0.88rem; + height: 22px; + justify-content: center; + line-height: 1; + width: 22px; +} + +.toast-message { + line-height: 1.2; +} + +.toast.error { + background: var(--toast-error-bg); + border-color: var(--toast-error-border); + color: var(--danger); +} + +.toast.error .toast-icon { + background: var(--danger); +} + +@keyframes toast-in { + from { + opacity: 0; + transform: translateY(-10px) scale(0.96); + } + to { + opacity: 1; + transform: translateY(0) scale(1); + } +} + .console-form { display: flex; gap: 10px; @@ -328,15 +536,100 @@ textarea { padding: 0; } +.run-summary { + display: grid; + gap: 8px; + grid-template-columns: repeat(3, minmax(150px, 1fr)); + margin-top: 12px; + max-width: 860px; +} + +.run-summary-item { + background: var(--panel); + border: 1px solid var(--line); + border-radius: 8px; + display: grid; + gap: 4px; + min-height: 64px; + padding: 10px 12px; +} + +.run-summary-item.primary { + border-color: var(--summary-primary-border); +} + +.run-summary-label { + color: var(--muted); + font-size: 0.78rem; + font-weight: 700; + text-transform: uppercase; +} + +.run-status { + align-items: center; + display: inline-flex; + font-weight: 800; + gap: 8px; +} + +.run-status::before { + background: currentColor; + border-radius: 999px; + content: ""; + height: 10px; + width: 10px; +} + +.run-status.running { + color: var(--accent-dark); +} + +.run-status.running::before { + animation: pulse 1.2s ease-in-out infinite; +} + +.run-status.failed, +.run-status.canceled { + color: var(--danger); +} + +.run-status.success { + color: var(--accent-dark); +} + +.run-status.skipped { + color: var(--muted); +} + +.run-status.pending { + color: var(--muted); +} + +.active-row { + background: var(--active-row-bg); +} + +@keyframes pulse { + 0%, + 100% { + opacity: 0.35; + transform: scale(0.82); + } + 50% { + opacity: 1; + transform: scale(1); + } +} + .log-viewer { display: grid; gap: 8px; } .log { - background: #151713; + background: var(--log-bg); border-radius: 8px; - color: #f0f5eb; + color: var(--log-text); font-family: "JetBrains Mono", "SFMono-Regular", Consolas, "Liberation Mono", monospace; line-height: 1.35; max-height: 620px; @@ -368,9 +661,9 @@ textarea { } .terminal-output { - background: #10120f; + background: var(--terminal-bg); border-radius: 6px; - color: #eef6e8; + color: var(--terminal-text); font-family: "JetBrains Mono", "SFMono-Regular", Consolas, "Liberation Mono", monospace; font-size: var(--terminal-font-size, 14px); height: 360px; @@ -389,8 +682,16 @@ textarea { } .terminal-caret { - background: #eef6e8; - color: #10120f; + background: var(--terminal-text); + color: var(--terminal-bg); +} + +.notice { + background: var(--notice-bg); + border: 1px solid var(--notice-border); + border-radius: 8px; + color: var(--notice-text); + padding: 12px 14px; } .settings-list { @@ -410,6 +711,12 @@ textarea { padding: 10px 0; } +.danger-zone { + border-top: 1px solid var(--line); + margin-top: 28px; + padding-top: 24px; +} + @media (max-width: 720px) { .topbar, .page-header, diff --git a/app/templates/_job_editor.html b/app/templates/_job_editor.html index d6fb20b..3ed5b3d 100644 --- a/app/templates/_job_editor.html +++ b/app/templates/_job_editor.html @@ -1,16 +1,16 @@ -
- - + + +
Schedule: {{ schedule_summary }}

Full commands

@@ -43,6 +43,10 @@

Full commands

Cancel + {% if job %} {% for preview in command_previews %} @@ -53,8 +57,104 @@

Full commands

diff --git a/app/templates/_log_viewer.html b/app/templates/_log_viewer.html index d812b50..170427a 100644 --- a/app/templates/_log_viewer.html +++ b/app/templates/_log_viewer.html @@ -7,6 +7,10 @@ id="log-output" class="log" data-chunk-url="{{ log_chunk_url }}" + data-append-url="{{ log_append_url }}" + data-status-url="{{ log_status_url }}" + data-append-offset="{{ log_append_offset }}" + data-live="{{ 'true' if run.status == 'running' and log_status_url else 'false' }}" data-next-before="{{ log_chunk.next_before if log_chunk.next_before is not none else '' }}" data-has-more="{{ 'true' if log_chunk.has_more else 'false' }}" >{{ log_chunk.text }} @@ -16,7 +20,14 @@ (() => { const output = document.querySelector("#log-output"); const loadStatus = document.querySelector("#log-load-status"); + const statusText = document.querySelector("#run-status"); + const exitCode = document.querySelector("#run-exit-code"); + const finishedAt = document.querySelector("#run-finished-at"); + const elapsed = document.querySelector("#run-elapsed"); + const cancelForm = document.querySelector("#cancel-run-form"); let loading = false; + let live = output.dataset.live === "true"; + let appendOffset = Number.parseInt(output.dataset.appendOffset || "0", 10); function hasMore() { return output.dataset.hasMore === "true" && output.dataset.nextBefore !== ""; @@ -44,6 +55,63 @@ updateStatus(); } + function isPinnedToBottom() { + return output.scrollHeight - output.scrollTop - output.clientHeight <= 24; + } + + async function refreshStatus() { + if (!output.dataset.statusUrl) { + return; + } + const response = await fetch(output.dataset.statusUrl); + const state = await response.json(); + if (statusText && state.status) { + statusText.textContent = state.status.charAt(0).toUpperCase() + state.status.slice(1); + statusText.className = `run-status ${state.status}`; + } + if (exitCode) { + exitCode.textContent = state.exit_label ?? "Pending"; + } + if (finishedAt) { + finishedAt.textContent = state.finished_at ?? "Running"; + } + if (elapsed && state.elapsed) { + elapsed.textContent = state.elapsed; + } + if (!state.can_cancel && cancelForm) { + cancelForm.remove(); + } + live = state.status === "running"; + } + + async function appendNewOutput() { + if (!output.dataset.appendUrl) { + return; + } + const pinned = isPinnedToBottom(); + const response = await fetch(`${output.dataset.appendUrl}?offset=${appendOffset}`); + const chunk = await response.json(); + appendOffset = chunk.offset ?? appendOffset; + output.dataset.appendOffset = appendOffset; + if (chunk.text) { + output.textContent = `${output.textContent}${chunk.text}`; + if (pinned) { + output.scrollTop = output.scrollHeight; + } + } + } + + async function pollLiveRun() { + if (!live) { + return; + } + await appendNewOutput(); + await refreshStatus(); + if (live) { + window.setTimeout(pollLiveRun, 1000); + } + } + output.addEventListener("scroll", () => { if (output.scrollTop <= 8) { loadOlderLines(); @@ -54,5 +122,6 @@ requestAnimationFrame(() => { output.scrollTop = output.scrollHeight; }); + pollLiveRun(); })(); diff --git a/app/templates/_pagination.html b/app/templates/_pagination.html new file mode 100644 index 0000000..cd90e08 --- /dev/null +++ b/app/templates/_pagination.html @@ -0,0 +1,15 @@ +{% if pagination and (pagination.has_previous or pagination.has_next) %} + +{% endif %} diff --git a/app/templates/base.html b/app/templates/base.html index c6c0a32..4cbda18 100644 --- a/app/templates/base.html +++ b/app/templates/base.html @@ -4,23 +4,66 @@ {% block title %}Rclone Runner{% endblock %} + +
- Rclone Runner + + + Rclone Runner + -
- -
+
+ +
+ +
+
{% block content %}{% endblock %}
+ diff --git a/app/templates/console_run_detail.html b/app/templates/console_run_detail.html index 6a0602b..a1d21fa 100644 --- a/app/templates/console_run_detail.html +++ b/app/templates/console_run_detail.html @@ -5,13 +5,19 @@

{{ run.command }}

- Exit {{ run.exit_code }} · + Status {{ run.status|capitalize }} · + Exit {{ run.exit_code if run.exit_code is not none else "Pending" }} · Started {{ format_local_time(run.started_at) }} · - Finished {{ format_local_time(run.ended_at) }} · - Elapsed {{ format_duration(run.started_at, run.ended_at) }} + Finished {{ format_local_time(run.ended_at) if run.ended_at is not none else "Running" }} · + Elapsed {{ format_duration(run.started_at, run.ended_at) if run.ended_at is not none else format_duration(run.started_at, utc_now()) }}

- Back +
+
+ +
+ Back +
{% include "_executed_command.html" %} {% include "_log_viewer.html" %} diff --git a/app/templates/job_detail.html b/app/templates/job_detail.html index e4dc15d..27a0f9f 100644 --- a/app/templates/job_detail.html +++ b/app/templates/job_detail.html @@ -4,7 +4,7 @@

{{ job.name }}

-

{{ schedule_summary }}{% if job.cron %} · {{ job.cron }}{% endif %} · {{ "Enabled" if job.enabled else "Disabled" }}

+

{{ schedule_summary }} · {{ "Enabled" if job.enabled else "Disabled" }}

@@ -14,21 +14,42 @@

{{ job.name }}

{% include "_job_editor.html" %} +

Recent job runs

+
+ + + + {% for run in job_runs %} + + + + + + + + {% else %} + + {% endfor %} + +
ModeStartedFinishedStatus
{{ run_mode_label(run.trigger) }}{{ format_local_time(run.started_at) }}{{ format_local_time(run.ended_at) if run.ended_at is not none else "Running" }}{{ run.status|capitalize }}Open
No job runs yet.
+
+

Recent step runs

- + {% for run in runs %} - - + + + {% else %} - + {% endfor %}
StepModeStartedExit
StepModeStartedStatusExit
{{ run.step_name }} {{ run_mode_label(run.job_run.trigger) }}{{ run.started_at }}{{ run.exit_code }}{{ format_local_time(run.started_at) }}{{ run.status|capitalize }}{{ run_exit_label(run) }} Log
No runs yet.
No runs yet.
diff --git a/app/templates/job_run_detail.html b/app/templates/job_run_detail.html new file mode 100644 index 0000000..ff7d76c --- /dev/null +++ b/app/templates/job_run_detail.html @@ -0,0 +1,144 @@ +{% extends "base.html" %} +{% block title %}{{ job_run.job_name }} run - Rclone Runner{% endblock %} +{% block content %} +
+
+

{{ job_run.job_name }}

+
+
+ Run status + {{ job_run.status|capitalize }} +
+
+ Mode + {{ run_mode_label(job_run.trigger) }} +
+
+ Current step + {{ active_step_run.step_name if active_step_run else "None" }} +
+
+ Elapsed + {{ format_duration(job_run.started_at, job_run.ended_at) if job_run.ended_at is not none else format_duration(job_run.started_at, utc_now()) }} +
+
+ Started + {{ format_local_time(job_run.started_at) }} +
+
+ Finished + {{ format_local_time(job_run.ended_at) if job_run.ended_at is not none else "Running" }} +
+
+
+
+ {% if job_run.status == "running" %} +
+ +
+ {% endif %} +
+ +
+ History +
+
+ +

Steps

+
+ + + + {% for row in step_rows %} + {% set step_run = row.run %} + + + + + + + {% endfor %} + +
StepStatusExitLog
{{ row.name }}{{ row.status|capitalize }}{{ run_exit_label(step_run) if step_run else "Pending" }} + {% if step_run %} + Open + {% else %} + Pending + {% endif %} +
+
+ +{% if active_step_run %} +

Active log

+{% set run = active_step_run %} +{% include "_executed_command.html" %} +{% include "_log_viewer.html" %} +{% endif %} + + +{% endblock %} diff --git a/app/templates/jobs.html b/app/templates/jobs.html index aa814b8..75a23b0 100644 --- a/app/templates/jobs.html +++ b/app/templates/jobs.html @@ -9,6 +9,74 @@

Jobs

New job
+

Ongoing activity

+ +

Job runs

+
+ + + + {% for item in ongoing_job_runs %} + {% set run = item.run %} + + + + + + + + + + {% else %} + + {% endfor %} + +
JobModeStartedElapsedCurrent stepStatus
{{ run.job_name }}{{ run_mode_label(run.trigger) }}{{ format_local_time(run.started_at) }}{{ format_duration(run.started_at, utc_now()) }}{{ item.current_step }}{{ run.status|capitalize }}Open
No job runs are currently active.
+
+ +

Step runs

+
+ + + + {% for run in ongoing_step_runs %} + + + + + + + + + + {% else %} + + {% endfor %} + +
StepJobModeStartedElapsedStatus
{{ run.step_name }}{{ run.job_run.job_name }}{{ run_mode_label(run.job_run.trigger) }}{{ format_local_time(run.started_at) }}{{ format_duration(run.started_at, utc_now()) }}{{ run.status|capitalize }}Log
No step runs are currently active.
+
+ +

Console commands

+
+ + + + {% for run in ongoing_console_runs %} + + + + + + + + {% else %} + + {% endfor %} + +
CommandStartedElapsedStatus
{{ run.command }}{{ format_local_time(run.started_at) }}{{ format_duration(run.started_at, utc_now()) }}{{ run.status|capitalize }}Log
No console commands are currently active.
+
+ +

Configured jobs

@@ -17,10 +85,7 @@

Jobs

{% set job = item.record %} - + ' in html + assert '
NameScheduleStatusSteps
{{ job.name }} -
{{ item.schedule_summary }}
- {% if job.cron %}{{ job.cron }}{% endif %} - 		{{ item.schedule_summary }} {{ "Enabled" if job.enabled else "Disabled" }} {{ job.steps|length }} diff --git a/app/templates/login.html b/app/templates/login.html index 640f701..db6b877 100644 --- a/app/templates/login.html +++ b/app/templates/login.html @@ -4,6 +4,19 @@ Log in - Rclone Runner + diff --git a/app/templates/run_detail.html b/app/templates/run_detail.html index 7f5952b..3df6eab 100644 --- a/app/templates/run_detail.html +++ b/app/templates/run_detail.html @@ -4,14 +4,45 @@

{{ run.step_name }}

-

- {{ run_mode_label(run.job_run.trigger) }} · Exit {{ run.exit_code }} · - Started {{ format_local_time(run.started_at) }} · - Finished {{ format_local_time(run.ended_at) }} · - Elapsed {{ format_duration(run.started_at, run.ended_at) }} -

+
+
+ Status + {{ run.status|capitalize }} +
+
+ Mode + {{ run_mode_label(run.job_run.trigger) }} +
+
+ Exit + {{ run_exit_label(run) }} +
+
+ Elapsed + {{ format_duration(run.started_at, run.ended_at) if run.ended_at is not none else format_duration(run.started_at, utc_now()) }} +
+
+ Started + {{ format_local_time(run.started_at) }} +
+
+ Finished + {{ format_local_time(run.ended_at) if run.ended_at is not none else "Running" }} +
+
+
+
+ {% if run.status == "running" %} +
+ +
+ {% endif %} +
+ +
+ Full run + Back
- Back
{% include "_executed_command.html" %} {% include "_log_viewer.html" %} diff --git a/app/templates/runs.html b/app/templates/runs.html index bdccd9f..6b8bc9d 100644 --- a/app/templates/runs.html +++ b/app/templates/runs.html @@ -2,42 +2,128 @@ {% block title %}History - Rclone Runner{% endblock %} {% block content %}

History

+
+

Job runs

+
+ + + + {% for run in job_runs %} + + + + + + + + + + {% else %} + + {% endfor %} + +
JobModeStartedFinishedStatusRun
{{ run.job_name }}{{ run_mode_label(run.trigger) }}{{ format_local_time(run.started_at) }}{{ format_local_time(run.ended_at) if run.ended_at is not none else "Running" }}{{ run.status|capitalize }}Open +
+ +
+
No job runs recorded.
+
+{% with pagination=job_pagination, label="Job runs pagination" %} +{% include "_pagination.html" %} +{% endwith %} +
+ +

Job step runs

- + {% for run in step_runs %} - - + + + + {% else %} - + {% endfor %}
StepModeStartedExitLog
StepModeStartedStatusExitLog
{{ run.step_name }} {{ run_mode_label(run.job_run.trigger) }}{{ run.started_at }}{{ run.exit_code }}{{ format_local_time(run.started_at) }}{{ run.status|capitalize }}{{ run_exit_label(run) }} Open +
+ +
+
No job runs recorded.
No job runs recorded.
+{% with pagination=step_pagination, label="Job step runs pagination" %} +{% include "_pagination.html" %} +{% endwith %} +
+

Console runs

- + {% for run in console_runs %} - - + + + + {% else %} - + {% endfor %}
CommandStartedExitLog
CommandStartedStatusExitLog
{{ run.command }}{{ run.started_at }}{{ run.exit_code }}{{ format_local_time(run.started_at) }}{{ run.status|capitalize }}{{ run.exit_code if run.exit_code is not none else "Pending" }} Open +
+ +
+
No console runs recorded.
No console runs recorded.
+{% with pagination=console_pagination, label="Console runs pagination" %} +{% include "_pagination.html" %} +{% endwith %} +
+ + {% endblock %} diff --git a/app/templates/settings.html b/app/templates/settings.html index 851f687..e486920 100644 --- a/app/templates/settings.html +++ b/app/templates/settings.html @@ -2,6 +2,12 @@ {% block title %}Settings - Rclone Runner{% endblock %} {% block content %}

Settings

+{% if pruned is not none %} +

Deleted {{ pruned }} old log {{ "file" if pruned == 1 else "files" }}.

+{% endif %} +{% if cleared is not none %} +

Cleared {{ cleared }} history {{ "record" if cleared == 1 else "records" }} and their known log files.

+{% endif %}
Data directory
{{ settings.data_dir }}
Log directory
{{ settings.log_dir }}
@@ -12,4 +18,12 @@

Settings

+ +
+

History

+

Clear run and console history to start from a clean slate. Job configuration remains intact.

+
+ +
+
{% endblock %} diff --git a/docs/superpowers/plans/2026-04-26-live-run-logs.md b/docs/superpowers/plans/2026-04-26-live-run-logs.md new file mode 100644 index 0000000..f372bb3 --- /dev/null +++ b/docs/superpowers/plans/2026-04-26-live-run-logs.md @@ -0,0 +1,305 @@ +# Live Run Logs Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Start manual job and step runs in the background, open the step log immediately, stream appended output live, update terminal metadata, and support canceling active jobs. + +**Architecture:** Persist `JobRunRecord` and `JobStepRunRecord` rows before subprocess execution starts. A `LiveJobRunner` service owns active asyncio tasks and subprocess cancellation, while existing pages poll JSON endpoints for status and appended log content. + +**Tech Stack:** FastAPI, SQLAlchemy, asyncio subprocesses, Jinja templates, vanilla JavaScript, pytest. + +--- + +### File Structure + +- Modify `app/db.py`: make run fields lifecycle-aware, add status fields, and add startup migration for existing SQLite databases. +- Modify `app/core/runner.py`: add cancelable subprocess handles and a live runner service that persists start/finish state. +- Modify `app/runner_service.py`: expose the live runner instance. +- Modify `app/scheduler.py`: use the live runner path for scheduled jobs. +- Modify `app/main.py`: start manual runs immediately, add status/log append/cancel endpoints, and render running metadata. +- Modify `app/core/logs.py`: add byte-offset append reading. +- Modify `app/templates/run_detail.html`: render status-aware metadata and stop button. +- Modify `app/templates/_log_viewer.html`: poll live output and status while running. +- Modify `app/templates/job_detail.html` and `app/templates/runs.html`: display running/canceled statuses. +- Modify `app/static/styles.css`: style the stop action and running status text. +- Modify `tests/test_runner.py`: cover live persistence and cancellation. +- Modify or create web/log tests: cover status and append payload behavior. + +### Task 1: Lifecycle Columns and Log Append Helper + +**Files:** +- Modify: `app/db.py` +- Modify: `app/core/logs.py` +- Test: `tests/test_log_chunks.py` + +- [ ] **Step 1: Write failing tests** + +Add tests to `tests/test_log_chunks.py`: + +```python +from app.core.logs import read_log_append + + +def test_reads_log_append_from_byte_offset(tmp_path: Path): + log_path = tmp_path / "run.log" + log_path.write_text("one\n", encoding="utf-8") + offset = log_path.stat().st_size + with log_path.open("a", encoding="utf-8") as log_file: + log_file.write("two\n") + + append = read_log_append(log_path, offset=offset) + + assert append == {"text": "two\n", "offset": log_path.stat().st_size} + + +def test_missing_append_log_returns_empty_payload(tmp_path: Path): + append = read_log_append(tmp_path / "missing.log", offset=12) + + assert append == {"text": "", "offset": 0} +``` + +- [ ] **Step 2: Run tests to verify failure** + +Run: `UV_CACHE_DIR=.uv-cache uv run pytest tests/test_log_chunks.py -q` + +Expected: import failure for `read_log_append`. + +- [ ] **Step 3: Implement log append helper and DB lifecycle fields** + +In `app/core/logs.py`, add: + +```python +def read_log_append(path: Path, offset: int = 0) -> dict[str, object]: + if not path.exists(): + return {"text": "", "offset": 0} + file_size = path.stat().st_size + start = max(0, min(offset, file_size)) + with path.open("rb") as log_file: + log_file.seek(start) + data = log_file.read() + return {"text": _decode(data), "offset": file_size} +``` + +In `app/db.py`, change `JobRunRecord.ended_at`, `JobStepRunRecord.exit_code`, `JobStepRunRecord.ended_at`, `ConsoleRunRecord.exit_code`, and `ConsoleRunRecord.ended_at` to nullable where needed for running records. Add `status` columns to job and step run records with defaults. Add a small SQLite migration in `init_db()` that uses `inspect(engine)` and `ALTER TABLE` to add `status` columns if missing. + +- [ ] **Step 4: Run tests** + +Run: `UV_CACHE_DIR=.uv-cache uv run pytest tests/test_log_chunks.py -q` + +Expected: all tests in the file pass. + +### Task 2: Live Runner Service + +**Files:** +- Modify: `app/core/runner.py` +- Modify: `app/runner_service.py` +- Test: `tests/test_runner.py` + +- [ ] **Step 1: Write failing tests** + +Add tests for: + +```python +@pytest.mark.asyncio +async def test_live_runner_creates_step_record_before_executor_finishes(): + started = asyncio.Event() + release = asyncio.Event() + + async def executor(argv, env, log_path): + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text("started\n", encoding="utf-8") + started.set() + await release.wait() + return 0 + + job = Job( + id=42, + name="backup", + common_args="", + env={}, + steps=[JobStep(id=1, name="one", command="lsd secret:")], + ) + + with tempfile.TemporaryDirectory() as tmpdir: + runner = LiveJobRunner(Path(tmpdir), session_factory=SessionLocal, executor=executor) + first_step = runner.start_job(job, trigger="manual") + assert first_step.status == "running" + assert first_step.exit_code is None + await started.wait() + release.set() + await runner.wait_for_job_run(first_step.job_run_id) + + with SessionLocal() as db: + finished = db.get(JobStepRunRecord, first_step.id) + assert finished.status == "success" + assert finished.exit_code == 0 + assert finished.ended_at is not None + +@pytest.mark.asyncio +async def test_live_runner_cancel_marks_job_canceled_and_skips_remaining_steps(): + started = asyncio.Event() + calls = [] + + async def executor(argv, env, log_path): + calls.append(argv) + started.set() + await asyncio.sleep(30) + return 0 + + job = Job( + id=42, + name="backup", + common_args="", + env={}, + steps=[ + JobStep(id=1, name="one", command="lsd secret:"), + JobStep(id=2, name="two", command="lsd secret:/two"), + ], + ) + + with tempfile.TemporaryDirectory() as tmpdir: + runner = LiveJobRunner(Path(tmpdir), session_factory=SessionLocal, executor=executor) + first_step = runner.start_job(job, trigger="manual") + await started.wait() + assert runner.cancel_job_run(first_step.job_run_id) + await runner.wait_for_job_run(first_step.job_run_id) + + with SessionLocal() as db: + job_run = db.get(JobRunRecord, first_step.job_run_id) + step_runs = db.query(JobStepRunRecord).filter_by(job_run_id=job_run.id).all() + assert job_run.status == "canceled" + assert [step.status for step in step_runs] == ["canceled"] + assert len(calls) == 1 +``` + +Use a temporary SQLite database/session or the app session fixture pattern already present in tests. The first test should start a live run, assert a step record exists with `status == "running"` before releasing the executor, then assert it becomes `success` with `exit_code == 0`. The second should start a two-step job, cancel the parent job run while the first executor waits, then assert only the first step ran and both job and step statuses are `canceled`. + +- [ ] **Step 2: Run tests to verify failure** + +Run: `UV_CACHE_DIR=.uv-cache uv run pytest tests/test_runner.py -q` + +Expected: failures because no live runner API exists. + +- [ ] **Step 3: Implement live runner** + +Add `LiveJobRunner` to `app/core/runner.py` with: + +- `start_job(job, trigger, dry_run=False, step_id=None) -> JobStepRunRecord | None` +- `_run_job_background(job_run_id, job, trigger, dry_run, selected_steps, run_stamp) -> None` internal coroutine +- `cancel_job_run(job_run_id) -> bool` +- `wait_for_job_run(job_run_id) -> None` for tests and shutdown-safe coordination +- active maps for `job_id -> job_run_id`, `job_run_id -> task`, and `job_run_id -> active subprocess/process cancel callback` + +Keep the existing `JobRunner.run_job()` API intact for current unit tests. Update `subprocess_executor` or add a live executor path that can terminate the subprocess when cancellation is requested. + +- [ ] **Step 4: Run tests** + +Run: `UV_CACHE_DIR=.uv-cache uv run pytest tests/test_runner.py -q` + +Expected: runner tests pass. + +### Task 3: Web Routes for Immediate Redirect, Status, Append, Cancel + +**Files:** +- Modify: `app/main.py` +- Modify: `app/scheduler.py` +- Test: `tests/test_web.py` or new `tests/test_live_runs.py` + +- [ ] **Step 1: Write failing route tests** + +Add tests that call the manual run route and assert it returns a redirect to `/runs/{id}` before the executor finishes. Add direct tests for: + +- `GET /runs/{run_id}/status` +- `GET /runs/{run_id}/log/append?offset=0` +- `POST /runs/{run_id}/cancel` + +- [ ] **Step 2: Run tests to verify failure** + +Run: `UV_CACHE_DIR=.uv-cache uv run pytest tests/test_web.py tests/test_runner.py -q` + +Expected: route or endpoint failures. + +- [ ] **Step 3: Implement routes** + +Change `_run_job()` in `app/main.py` to call `runner.start_job(...)` and redirect immediately. Add: + +```python +@app.get("/runs/{run_id}/status") +async def run_status(run_id: int, _: AuthRequired, db: DbSession) -> dict[str, object]: + run = db.get(JobStepRunRecord, run_id) + if run is None: + return {"status": "missing", "can_cancel": False} + return _run_status_payload(run) + +@app.get("/runs/{run_id}/log/append") +async def run_log_append(run_id: int, _: AuthRequired, db: DbSession, offset: int = 0) -> dict[str, object]: + run = db.get(JobStepRunRecord, run_id) + if run is None: + return {"text": "", "offset": 0} + return read_log_append(Path(run.log_path), offset) + +@app.post("/runs/{run_id}/cancel") +async def cancel_run(run_id: int, _: AuthRequired, db: DbSession) -> Response: + run = db.get(JobStepRunRecord, run_id) + if run is not None: + runner.cancel_job_run(run.job_run_id) + return RedirectResponse(f"/runs/{run_id}", status_code=303) +``` + +Update `app/scheduler.py` to call the same live runner service without expecting a return value. + +- [ ] **Step 4: Run focused web tests** + +Run: `UV_CACHE_DIR=.uv-cache uv run pytest tests/test_web.py tests/test_runner.py -q` + +Expected: focused tests pass. + +### Task 4: Live Log UI + +**Files:** +- Modify: `app/templates/run_detail.html` +- Modify: `app/templates/_log_viewer.html` +- Modify: `app/templates/job_detail.html` +- Modify: `app/templates/runs.html` +- Modify: `app/static/styles.css` + +- [ ] **Step 1: Update templates** + +Add data attributes for status and append URLs to the log viewer. Render status text instead of assuming exit code exists. Show a stop button only when `run.status == "running"`. + +- [ ] **Step 2: Update JavaScript** + +Extend `_log_viewer.html` JavaScript to poll append and status endpoints every second while running. Preserve older-lines loading and bottom-follow behavior. + +- [ ] **Step 3: Update styles** + +Add minimal styles for `.danger`, `.run-meta`, and disabled stop form state. + +- [ ] **Step 4: Run template-related tests** + +Run: `UV_CACHE_DIR=.uv-cache uv run pytest tests/test_web.py tests/test_console_logs.py -q` + +Expected: tests pass. + +### Task 5: Full Verification + +**Files:** +- All modified files + +- [ ] **Step 1: Run formatter** + +Run: `UV_CACHE_DIR=.uv-cache uv run ruff format app tests` + +Expected: files formatted. + +- [ ] **Step 2: Run full tests and lint** + +Run: `UV_CACHE_DIR=.uv-cache uv run pytest && UV_CACHE_DIR=.uv-cache uv run ruff check . && UV_CACHE_DIR=.uv-cache uv run ruff format --check .` + +Expected: all tests pass and ruff reports no issues. + +- [ ] **Step 3: Review diff** + +Run: `git diff --stat && git diff -- app tests docs/superpowers/plans/2026-04-26-live-run-logs.md` + +Expected: diff is scoped to live run logs, status, cancellation, and tests. diff --git a/docs/superpowers/specs/2026-04-26-live-run-logs-design.md b/docs/superpowers/specs/2026-04-26-live-run-logs-design.md new file mode 100644 index 0000000..85835e3 --- /dev/null +++ b/docs/superpowers/specs/2026-04-26-live-run-logs-design.md @@ -0,0 +1,68 @@ +# Live Run Logs Design + +## Goal + +Manual job and step runs should open their log page immediately, stream visible output while the command runs, update exit code and finish time when complete, show live elapsed time, and allow a running command to be canceled. + +## Current Behavior + +`POST /jobs/{job_id}/run` and step run routes await `runner.run_job(...)` before saving `JobRunRecord` and `JobStepRunRecord` rows. Because records are only persisted after completion, the app has no run detail URL to redirect to while a command is still executing. + +## Architecture + +Persist run records at the start of execution, then run the subprocess in a background task. The log page reads the already-created `JobStepRunRecord`, polls for appended log output and metadata, and displays terminal state once the runner updates the record. + +The runner remains the owner of subprocess lifecycle and overlap prevention. The web layer starts manual runs, redirects to the first persisted step run, and exposes status and cancel endpoints. + +## Data Model + +`JobRunRecord` and `JobStepRunRecord` need lifecycle-friendly fields: + +- `status`: `running`, `success`, `failed`, `canceled`, or `skipped`. +- `exit_code`: nullable for running rows. +- `ended_at`: nullable for running rows. + +Existing completed records continue to render as finished runs. + +## Execution Flow + +When a manual job or step is started: + +1. Validate and load the job. +2. Create a `JobRunRecord` with `status="running"` and a first `JobStepRunRecord` with `status="running"`, nullable `exit_code`, nullable `ended_at`, command argv, timestamps, and log path. +3. Start the actual execution in an asyncio background task. +4. Redirect immediately to `/runs/{step_run_id}`. + +For full jobs, subsequent step rows are created as each step starts. If a step fails, later steps are not started and the job status becomes `failed`. If a run is canceled, the active subprocess is stopped, the active step and parent job are marked `canceled`, and later steps are skipped by omission. + +Scheduled runs use the same live persistence path but do not need an immediate redirect. + +## Live Log Page + +The existing run detail page remains the canonical log page. It renders current metadata from the database and initializes the existing tail chunk. + +For running rows, the page: + +- Polls a metadata endpoint for status, exit code, finish time, and elapsed seconds. +- Polls a log endpoint for bytes appended since the last known offset. +- Keeps the log scrolled to the bottom when the user is already following the tail. +- Stops live polling when the run reaches a terminal status. + +The existing older-lines behavior remains available for completed logs and for users who scroll upward. + +## Cancellation + +Running step pages show a stop button. Posting to the cancel endpoint requests cancellation for the parent job run. The runner terminates the current subprocess, records the active step as canceled, records the parent job as canceled, sets `ended_at`, and releases overlap protection for that job. + +If cancellation is requested after the run has already finished, the endpoint is idempotent and redirects back to the log page without changing terminal metadata. + +## Testing + +Tests should cover: + +- Manual run routes create records and redirect before executor completion. +- Running status responses expose nullable exit code, live elapsed time, and `can_cancel=true`. +- Finishing a run updates exit code, finish time, and terminal status. +- Canceling a full job stops the active step and prevents later steps from starting. +- Log append polling returns only content written after the requested byte offset. +- Existing completed-run rendering and history behavior remain compatible. diff --git a/tests/test_console_logs.py b/tests/test_console_logs.py index c7623e6..ee091e4 100644 --- a/tests/test_console_logs.py +++ b/tests/test_console_logs.py @@ -1,7 +1,16 @@ from datetime import UTC, datetime -from app.db import ConsoleRunRecord -from app.main import _console_history_row, _console_log_path +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from app.db import Base, ConsoleRunRecord +from app.main import ( + _cancel_console_run_record, + _console_history_row, + _console_log_path, + _finish_console_run_record, + _start_console_run_record, +) def test_console_log_path_includes_microseconds_to_avoid_quick_run_collisions(): @@ -32,3 +41,54 @@ def test_console_history_row_links_to_log_without_embedding_output(tmp_path): "started_at": "2026-04-26 15:00:00 EEST", "exit_code": 0, } + + +def test_console_run_is_recorded_while_running_and_finished_later(tmp_path): + session_factory = _session_factory(tmp_path / "runs.db") + started_at = datetime(2026, 4, 26, 12, 0, tzinfo=UTC) + ended_at = datetime(2026, 4, 26, 12, 0, 1, tzinfo=UTC) + with session_factory() as db: + run_id = _start_console_run_record( + db, + "version", + ["rclone", "version"], + tmp_path / "console.log", + started_at, + ) + + running = db.get(ConsoleRunRecord, run_id) + assert running.status == "running" + assert running.exit_code is None + assert running.ended_at is None + + finished = _finish_console_run_record(db, run_id, 0, ended_at) + + assert finished.status == "success" + assert finished.exit_code == 0 + assert finished.ended_at == ended_at.replace(tzinfo=None) + + +def test_console_run_can_be_finalized_as_canceled(tmp_path): + session_factory = _session_factory(tmp_path / "runs.db") + started_at = datetime(2026, 4, 26, 12, 0, tzinfo=UTC) + ended_at = datetime(2026, 4, 26, 12, 0, 1, tzinfo=UTC) + with session_factory() as db: + run_id = _start_console_run_record( + db, + "version", + ["rclone", "version"], + tmp_path / "console.log", + started_at, + ) + + canceled = _cancel_console_run_record(db, run_id, ended_at) + + assert canceled.status == "canceled" + assert canceled.exit_code is None + assert canceled.ended_at == ended_at.replace(tzinfo=None) + + +def _session_factory(database_path): + engine = create_engine(f"sqlite:///{database_path}", connect_args={"check_same_thread": False}) + Base.metadata.create_all(bind=engine) + return sessionmaker(bind=engine, autoflush=False, autocommit=False) diff --git a/tests/test_db_migration.py b/tests/test_db_migration.py new file mode 100644 index 0000000..b81bee3 --- /dev/null +++ b/tests/test_db_migration.py @@ -0,0 +1,72 @@ +from sqlalchemy import create_engine, inspect + +from app.db import Base, _copy_lifecycle_rows + + +def test_lifecycle_migration_derives_missing_status_from_exit_code(tmp_path): + engine = create_engine(f"sqlite:///{tmp_path / 'runs.db'}") + with engine.begin() as connection: + connection.exec_driver_sql( + """ + CREATE TABLE job_step_runs_old ( + id INTEGER PRIMARY KEY, + job_run_id INTEGER NOT NULL, + step_id INTEGER, + step_name VARCHAR(200) NOT NULL, + argv_json TEXT NOT NULL, + exit_code INTEGER NOT NULL, + log_path TEXT NOT NULL, + started_at DATETIME NOT NULL, + ended_at DATETIME NOT NULL + ) + """ + ) + connection.exec_driver_sql( + """ + INSERT INTO job_step_runs_old ( + id, job_run_id, step_id, step_name, argv_json, exit_code, log_path, + started_at, ended_at + ) + VALUES + (1, 1, 1, 'ok', '[]', 0, 'ok.log', '2026-04-26 12:00:00', '2026-04-26 12:00:01'), + (2, 1, 2, 'bad', '[]', 7, 'bad.log', '2026-04-26 12:00:00', '2026-04-26 12:00:01') + """ + ) + connection.exec_driver_sql( + """ + CREATE TABLE console_runs_old ( + id INTEGER PRIMARY KEY, + command TEXT NOT NULL, + argv_json TEXT NOT NULL, + exit_code INTEGER NOT NULL, + log_path TEXT NOT NULL, + started_at DATETIME NOT NULL, + ended_at DATETIME NOT NULL + ) + """ + ) + connection.exec_driver_sql( + """ + INSERT INTO console_runs_old ( + id, command, argv_json, exit_code, log_path, started_at, ended_at + ) + VALUES + (1, 'ok', '[]', 0, 'ok.log', '2026-04-26 12:00:00', '2026-04-26 12:00:01'), + (2, 'bad', '[]', 7, 'bad.log', '2026-04-26 12:00:00', '2026-04-26 12:00:01') + """ + ) + Base.metadata.tables["job_step_runs"].create(bind=connection) + Base.metadata.tables["console_runs"].create(bind=connection) + + _copy_lifecycle_rows(connection, inspect(connection), "job_step_runs") + _copy_lifecycle_rows(connection, inspect(connection), "console_runs") + + step_statuses = connection.exec_driver_sql( + "SELECT id, status FROM job_step_runs ORDER BY id" + ).all() + console_statuses = connection.exec_driver_sql( + "SELECT id, status FROM console_runs ORDER BY id" + ).all() + + assert step_statuses == [(1, "success"), (2, "failed")] + assert console_statuses == [(1, "success"), (2, "failed")] diff --git a/tests/test_job_preview.py b/tests/test_job_preview.py index 8825ccd..159f3ed 100644 --- a/tests/test_job_preview.py +++ b/tests/test_job_preview.py @@ -1,10 +1,121 @@ import json +from datetime import UTC, datetime from html.parser import HTMLParser +from pathlib import Path +from types import SimpleNamespace from app.db import JobRecord, JobStepRecord from app.main import _command_previews, templates +def test_headings_after_large_blocks_have_section_spacing(): + css = Path("app/static/styles.css").read_text() + + assert ".table-wrap + h2" in css + assert ".form-grid + h2" in css + assert ".terminal-output + h2" in css + assert ".section-heading" in css + + +def test_base_template_links_svg_favicon(): + html = templates.get_template("base.html").render() + + assert '' in html + assert '' in html + + +def test_base_template_supports_theme_detection_and_toggle(): + html = templates.get_template("base.html").render() + login_html = templates.get_template("login.html").render() + css = Path("app/static/styles.css").read_text() + + assert 'localStorage.getItem("theme")' in html + assert 'localStorage.getItem("theme")' in login_html + assert 'matchMedia("(prefers-color-scheme: dark)")' in html + assert 'matchMedia("(prefers-color-scheme: dark)")' in login_html + assert "document.documentElement.dataset.theme = theme;" in html + assert "document.documentElement.dataset.theme = theme;" in login_html + assert 'id="theme-toggle"' in html + assert 'aria-label="Switch color theme"' in html + assert 'localStorage.setItem("theme", nextTheme)' in html + assert 'html[data-theme="dark"]' in css + assert "color-scheme: dark" in css + + +def test_jobs_page_renders_ongoing_activity_sections(): + started_at = datetime(2026, 4, 26, 18, 0, tzinfo=UTC) + job_run = SimpleNamespace( + id=3, + job_name="backup", + trigger="manual", + status="running", + started_at=started_at, + ended_at=None, + ) + step_run = SimpleNamespace( + id=4, + step_name="Music", + status="running", + started_at=started_at, + ended_at=None, + job_run=SimpleNamespace(job_name="backup", trigger="manual"), + ) + console_run = SimpleNamespace( + id=5, + command="lsd remote:", + status="running", + started_at=started_at, + ended_at=None, + ) + + html = templates.get_template("jobs.html").render( + jobs=[], + ongoing_job_runs=[{"run": job_run, "current_step": "Music"}], + ongoing_step_runs=[step_run], + ongoing_console_runs=[console_run], + ) + + assert "

Ongoing activity

" in html + assert "

Job runs

" in html + assert 'Open' in html + assert "

Step runs

" in html + assert 'Log' in html + assert "

Console commands

" in html + assert 'Log' in html + assert "2026-04-26 21:00:00 EEST" in html + + +def test_jobs_page_renders_empty_ongoing_activity_tables(): + html = templates.get_template("jobs.html").render( + jobs=[], + ongoing_job_runs=[], + ongoing_step_runs=[], + ongoing_console_runs=[], + ) + + assert "

Ongoing activity

" in html + assert "

Job runs

" in html + assert "No job runs are currently active." in html + assert "

Step runs

" in html + assert "No step runs are currently active." in html + assert "

Console commands

" in html + assert "No console commands are currently active." in html + + +def test_job_detail_history_heading_has_explicit_section_spacing(): + html = templates.get_template("job_detail.html").render( + job=SimpleNamespace(id=1, name="backup", cron="", enabled=True), + schedule_summary="Never", + env_lines="", + steps_text="", + command_previews=[], + job_runs=[], + runs=[], + ) + + assert '

Recent job runs

' in html + + def test_command_preview_uses_common_args_and_job_environment(): job = JobRecord( name="backup", @@ -65,6 +176,311 @@ def test_step_run_buttons_do_not_create_nested_forms(): assert 'id="dry-run-step-2" method="post" action="/jobs/1/steps/2/dry-run"' in html +def test_job_editor_ctrl_s_saves_without_navigating(): + html = templates.get_template("_job_editor.html").render( + job=SimpleNamespace(id=1, name="backup", cron="", enabled=True, common_args=""), + schedule_summary="Never", + env_lines="", + steps_text="", + command_previews=[], + ) + + assert '
Canceled' in history_html + assert 'Success' in history_html + assert 'Canceled' in job_html + + +def test_history_pages_link_to_whole_job_runs(): + job_run = SimpleNamespace( + id=3, + job_name="backup", + trigger="manual", + status="success", + started_at=datetime(2026, 4, 26, 17, 59, tzinfo=UTC), + ended_at=datetime(2026, 4, 26, 18, 0, tzinfo=UTC), + ) + + history_html = templates.get_template("runs.html").render( + job_runs=[job_run], + step_runs=[], + console_runs=[], + ) + job_html = templates.get_template("job_detail.html").render( + job=SimpleNamespace(id=1, name="backup", cron="", enabled=True), + schedule_summary="Never", + env_lines="", + steps_text="", + command_previews=[], + job_runs=[job_run], + runs=[], + ) + + assert "

Job runs

" in history_html + assert 'Open' in history_html + assert '

Recent job runs

' in job_html + assert 'Open' in job_html + + +def test_history_tables_render_pagination_controls(): + job_run = SimpleNamespace( + id=3, + job_name="backup", + trigger="manual", + status="success", + started_at=datetime(2026, 4, 26, 17, 59, tzinfo=UTC), + ended_at=datetime(2026, 4, 26, 18, 0, tzinfo=UTC), + ) + + html = templates.get_template("runs.html").render( + job_runs=[job_run], + step_runs=[], + console_runs=[], + job_pagination={ + "page": 2, + "has_previous": True, + "has_next": True, + "total_pages": 4, + "previous_url": "/runs?job_page=1&step_page=3&console_page=4#job-runs-section", + "next_url": "/runs?job_page=3&step_page=3&console_page=4#job-runs-section", + "target": "job-runs-section", + }, + step_pagination={"page": 3, "has_previous": False, "has_next": False}, + console_pagination={"page": 4, "has_previous": False, "has_next": False}, + ) + + assert '
' in html + assert '
' in html + + +def test_job_history_tables_format_times_in_configured_timezone(): + job_run = SimpleNamespace( + id=3, + job_name="backup", + trigger="manual", + status="success", + started_at=datetime(2026, 4, 26, 18, 0, tzinfo=UTC), + ended_at=datetime(2026, 4, 26, 18, 1, tzinfo=UTC), + ) + step_run = SimpleNamespace( + id=1, + step_name="Music", + status="success", + exit_code=0, + started_at=datetime(2026, 4, 26, 18, 0, tzinfo=UTC), + job_run=SimpleNamespace(trigger="manual"), + ) + console_run = SimpleNamespace( + id=2, + command="lsd remote:", + status="success", + exit_code=0, + started_at=datetime(2026, 4, 26, 18, 0, tzinfo=UTC), + ) + + history_html = templates.get_template("runs.html").render( + job_runs=[job_run], + step_runs=[step_run], + console_runs=[console_run], + ) + job_html = templates.get_template("job_detail.html").render( + job=SimpleNamespace(id=1, name="backup", cron="", enabled=True), + schedule_summary="Never", + env_lines="", + steps_text="", + command_previews=[], + job_runs=[job_run], + runs=[step_run], + ) + + assert "2026-04-26 21:00:00 EEST" in history_html + assert "2026-04-26 21:01:00 EEST" in history_html + assert "2026-04-26 18:00:00+00:00" not in history_html + assert "2026-04-26 21:00:00 EEST" in job_html + assert "2026-04-26 18:00:00+00:00" not in job_html + + +def test_job_summaries_show_human_schedule_without_raw_cron(): + job = SimpleNamespace(id=1, name="backup", cron="0 2 * * *", enabled=True, steps=[]) + + jobs_html = templates.get_template("jobs.html").render( + jobs=[{"record": job, "schedule_summary": "Daily at 02:00"}], + ) + job_html = templates.get_template("job_detail.html").render( + job=job, + schedule_summary="Daily at 02:00", + env_lines="", + steps_text="", + command_previews=[], + job_runs=[], + runs=[], + ) + + assert "Daily at 02:00" in jobs_html + assert "0 2 * * *" not in jobs_html + assert "

Daily at 02:00 · Enabled

" in job_html + + +def test_settings_page_shows_prune_feedback_and_clear_history_action(): + html = templates.get_template("settings.html").render( + settings=SimpleNamespace( + data_dir="/data", + log_dir="/logs", + timezone="Europe/Helsinki", + retention_days=30, + ), + known_logs=3, + pruned=2, + cleared=4, + ) + + assert "Deleted 2 old log files." in html + assert "Cleared 4 history records and their known log files." in html + assert 'action="/settings/clear-history"' in html + assert "Clear all history" in html + assert "Job configuration remains intact." in html + + class NestedFormParser(HTMLParser): def __init__(self) -> None: super().__init__() diff --git a/tests/test_log_chunks.py b/tests/test_log_chunks.py index 5196882..a0a6b75 100644 --- a/tests/test_log_chunks.py +++ b/tests/test_log_chunks.py @@ -1,6 +1,6 @@ from pathlib import Path -from app.core.logs import read_log_chunk +from app.core.logs import read_log_append, read_log_chunk def test_reads_last_lines_without_loading_entire_log(tmp_path: Path): @@ -33,3 +33,21 @@ def test_missing_log_returns_pruned_message(tmp_path: Path): assert chunk.text == "Log file has been pruned or is not available." assert not chunk.has_more assert chunk.next_before is None + + +def test_reads_log_append_from_byte_offset(tmp_path: Path): + log_path = tmp_path / "run.log" + log_path.write_text("one\n", encoding="utf-8") + offset = log_path.stat().st_size + with log_path.open("a", encoding="utf-8") as log_file: + log_file.write("two\n") + + append = read_log_append(log_path, offset=offset) + + assert append == {"text": "two\n", "offset": log_path.stat().st_size} + + +def test_missing_append_log_returns_empty_payload(tmp_path: Path): + append = read_log_append(tmp_path / "missing.log", offset=12) + + assert append == {"text": "", "offset": 0} diff --git a/tests/test_retention.py b/tests/test_retention.py index 0640d9c..74f3cd9 100644 --- a/tests/test_retention.py +++ b/tests/test_retention.py @@ -29,3 +29,27 @@ def test_prunes_old_logs_but_keeps_recent_logs(): assert deleted == 1 assert not old_log.exists() assert recent_log.exists() + + +def test_prune_logs_treats_naive_database_timestamps_as_utc(): + now = datetime(2026, 4, 26, tzinfo=UTC) + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + old_log = root / "old.log" + recent_log = root / "recent.log" + old_log.write_text("old", encoding="utf-8") + recent_log.write_text("recent", encoding="utf-8") + + deleted = prune_logs( + [ + (old_log, datetime(2026, 3, 1)), + (recent_log, datetime(2026, 4, 25)), + ], + keep_days=30, + now=now, + ) + + assert deleted == 1 + assert not old_log.exists() + assert recent_log.exists() diff --git a/tests/test_runner.py b/tests/test_runner.py index af4b696..96d112c 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -3,9 +3,12 @@ from pathlib import Path import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker from app.core.models import Job, JobStep -from app.core.runner import JobRunner +from app.core.runner import JobRunner, LiveJobRunner +from app.db import Base, JobRunRecord, JobStepRunRecord from app.main import run_mode_label @@ -127,3 +130,154 @@ def test_run_mode_labels_distinguish_dry_runs(): assert run_mode_label("schedule") == "Scheduled" assert run_mode_label("manual-dry-run") == "Dry run" assert run_mode_label("manual-step-dry-run") == "Dry run" + + +@pytest.mark.asyncio +async def test_live_runner_creates_step_record_before_executor_finishes(): + started = asyncio.Event() + release = asyncio.Event() + + async def executor(argv, env, log_path): + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text("started\n", encoding="utf-8") + started.set() + await release.wait() + return 0 + + job = Job( + id=42, + name="backup", + common_args="", + env={}, + steps=[JobStep(id=1, name="one", command="lsd secret:")], + ) + + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + runner = LiveJobRunner(Path(tmpdir), session_factory=session_factory, executor=executor) + first_step = runner.start_job(job, trigger="manual") + + assert first_step is not None + assert first_step.status == "running" + assert first_step.exit_code is None + assert first_step.ended_at is None + + await started.wait() + release.set() + await runner.wait_for_job_run(first_step.job_run_id) + + with session_factory() as db: + finished = db.get(JobStepRunRecord, first_step.id) + assert finished is not None + assert finished.status == "success" + assert finished.exit_code == 0 + assert finished.ended_at is not None + + +@pytest.mark.asyncio +async def test_live_runner_start_job_works_with_expiring_sessions(): + started = asyncio.Event() + + async def executor(argv, env, log_path): + started.set() + return 0 + + job = Job( + id=42, + name="backup", + common_args="", + env={}, + steps=[JobStep(id=1, name="one", command="lsd secret:")], + ) + + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db", expire_on_commit=True) + runner = LiveJobRunner(Path(tmpdir), session_factory=session_factory, executor=executor) + + first_step = runner.start_job(job, trigger="manual-step-dry-run", dry_run=True, step_id=1) + + assert first_step is not None + assert first_step.job_run_id == 1 + await started.wait() + await runner.wait_for_job_run(first_step.job_run_id) + + +@pytest.mark.asyncio +async def test_live_runner_cancel_marks_job_canceled_and_skips_remaining_steps(): + started = asyncio.Event() + calls = [] + + async def executor(argv, env, log_path): + calls.append(argv) + started.set() + await asyncio.sleep(30) + return 0 + + job = Job( + id=42, + name="backup", + common_args="", + env={}, + steps=[ + JobStep(id=1, name="one", command="lsd secret:"), + JobStep(id=2, name="two", command="lsd secret:/two"), + ], + ) + + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + runner = LiveJobRunner(Path(tmpdir), session_factory=session_factory, executor=executor) + first_step = runner.start_job(job, trigger="manual") + assert first_step is not None + + await started.wait() + assert runner.cancel_job_run(first_step.job_run_id) + await runner.wait_for_job_run(first_step.job_run_id) + + with session_factory() as db: + job_run = db.get(JobRunRecord, first_step.job_run_id) + step_runs = db.query(JobStepRunRecord).filter_by(job_run_id=job_run.id).all() + assert job_run.status == "canceled" + assert job_run.ended_at is not None + assert [step.status for step in step_runs] == ["canceled"] + assert step_runs[0].exit_code is None + assert len(calls) == 1 + + +@pytest.mark.asyncio +async def test_live_runner_marks_executor_exception_as_failed(): + async def executor(argv, env, log_path): + raise FileNotFoundError("rclone") + + job = Job( + id=42, + name="backup", + common_args="", + env={}, + steps=[JobStep(id=1, name="one", command="lsd secret:")], + ) + + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + runner = LiveJobRunner(Path(tmpdir), session_factory=session_factory, executor=executor) + first_step = runner.start_job(job, trigger="manual") + assert first_step is not None + + await runner.wait_for_job_run(first_step.job_run_id) + + with session_factory() as db: + job_run = db.get(JobRunRecord, first_step.job_run_id) + step_run = db.get(JobStepRunRecord, first_step.id) + assert job_run.status == "failed" + assert job_run.ended_at is not None + assert step_run.status == "failed" + assert step_run.exit_code is None + assert step_run.ended_at is not None + + +def _session_factory(database_path: Path, expire_on_commit: bool = False): + engine = create_engine(f"sqlite:///{database_path}", connect_args={"check_same_thread": False}) + Base.metadata.create_all(bind=engine) + return sessionmaker( + bind=engine, autoflush=False, autocommit=False, expire_on_commit=expire_on_commit + ) diff --git a/tests/test_schedule.py b/tests/test_schedule.py index 79bb448..c72350f 100644 --- a/tests/test_schedule.py +++ b/tests/test_schedule.py @@ -1,4 +1,11 @@ +from pathlib import Path + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +import app.scheduler as scheduler_module from app.core.schedule import cron_summary, normalize_cron +from app.db import Base, JobRecord def test_empty_cron_is_never(): @@ -12,3 +19,32 @@ def test_common_cron_summaries_are_humanized(): assert cron_summary("0 4 1 * *") == "Monthly on day 1 at 04:00" assert cron_summary("*/15 * * * *") == "Every 15 minutes" assert cron_summary("0 */6 * * *") == "Every 6 hours" + + +def test_sync_schedules_registers_async_job_runner(tmp_path: Path, monkeypatch): + class FakeScheduler: + def __init__(self) -> None: + self.jobs = [] + + def remove_all_jobs(self) -> None: + self.jobs.clear() + + def add_job(self, func, trigger, **kwargs) -> None: + self.jobs.append({"func": func, "trigger": trigger, "kwargs": kwargs}) + + fake_scheduler = FakeScheduler() + monkeypatch.setattr(scheduler_module, "scheduler", fake_scheduler) + session_factory = _session_factory(tmp_path / "runs.db") + with session_factory() as session: + session.add(JobRecord(name="backup", cron="*/1 * * * *", enabled=True)) + session.commit() + + scheduler_module.sync_schedules(session) + + assert fake_scheduler.jobs[0]["func"] is scheduler_module._run_and_record + + +def _session_factory(database_path: Path): + engine = create_engine(f"sqlite:///{database_path}", connect_args={"check_same_thread": False}) + Base.metadata.create_all(bind=engine) + return sessionmaker(bind=engine, autoflush=False, autocommit=False) diff --git a/tests/test_web.py b/tests/test_web.py index f37c012..9c8feb5 100644 --- a/tests/test_web.py +++ b/tests/test_web.py @@ -1,5 +1,22 @@ +import asyncio +import tempfile +from pathlib import Path + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker from starlette.requests import Request +import app.main as main +from app.core.runner import LiveJobRunner +from app.db import ( + Base, + ConsoleRunRecord, + JobRecord, + JobRunRecord, + JobStepRecord, + JobStepRunRecord, +) from app.main import login_form @@ -10,3 +27,587 @@ async def test_login_page_renders_with_current_starlette_template_api(): assert response.status_code == 200 assert response.template.name == "login.html" + + +async def test_new_job_page_uses_empty_values_with_hints(): + request = Request({"type": "http", "method": "GET", "path": "/jobs/new", "headers": []}) + + response = await main.new_job(request, None) + html = response.body.decode() + + assert 'name="name" value=""' in html + assert 'name="cron" value=""' in html + assert "Google Drive backups" not in html + assert "--fast-list --transfers=20 --checkers=40 --verbose --bwlimit ${BW_LIMIT}" in html + assert 'placeholder="BW_LIMIT=8M">' in html + assert 'placeholder="Sync Music|sync /media/Musiikki remote:/Musiikki">' in html + assert 'placeholder="Nightly media backup"' in html + + +async def test_jobs_page_receives_ongoing_activity(): + request = Request({"type": "http", "method": "GET", "path": "/jobs", "headers": []}) + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + with session_factory() as db: + job = _create_job(db) + job_run = JobRunRecord( + job_id=job.id, + job_name=job.name, + trigger="manual", + status="running", + started_at=main.utc_now(), + ended_at=None, + ) + db.add(job_run) + db.flush() + step_run = JobStepRunRecord( + job_run_id=job_run.id, + step_id=job.steps[0].id, + step_name=job.steps[0].name, + argv_json='["rclone", "lsd", "secret:"]', + status="running", + exit_code=None, + log_path=str(Path(tmpdir) / "run.log"), + started_at=job_run.started_at, + ended_at=None, + ) + console_run = ConsoleRunRecord( + status="running", + command="lsd remote:", + argv_json='["rclone", "lsd", "remote:"]', + exit_code=None, + log_path=str(Path(tmpdir) / "console.log"), + started_at=job_run.started_at, + ended_at=None, + ) + db.add_all([step_run, console_run]) + db.commit() + + response = await main.jobs(request, None, db) + + assert response.context["ongoing_job_runs"][0]["run"].job_name == "backup" + assert response.context["ongoing_step_runs"][0].step_name == "one" + assert response.context["ongoing_console_runs"][0].command == "lsd remote:" + + +@pytest.mark.asyncio +async def test_manual_full_job_redirects_to_live_job_run_before_executor_finishes(monkeypatch): + started = asyncio.Event() + release = asyncio.Event() + + async def executor(argv, env, log_path): + started.set() + await release.wait() + return 0 + + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + runner = LiveJobRunner(Path(tmpdir), session_factory=session_factory, executor=executor) + monkeypatch.setattr(main, "runner", runner) + with session_factory() as db: + job = _create_job(db) + + response = await main._run_job(job.id, "manual", False, db) + + assert response.status_code == 303 + assert response.headers["location"] == "/job-runs/1" + step_run = db.get(JobStepRunRecord, 1) + assert step_run.status == "running" + assert step_run.exit_code is None + + await started.wait() + release.set() + await runner.wait_for_job_run(1) + + +@pytest.mark.asyncio +async def test_manual_single_step_run_redirects_to_live_step_log(monkeypatch): + started = asyncio.Event() + release = asyncio.Event() + + async def executor(argv, env, log_path): + started.set() + await release.wait() + return 0 + + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + runner = LiveJobRunner(Path(tmpdir), session_factory=session_factory, executor=executor) + monkeypatch.setattr(main, "runner", runner) + with session_factory() as db: + job = _create_job(db) + step_id = job.steps[0].id + + response = await main._run_job(job.id, "manual-step", False, db, step_id=step_id) + + assert response.status_code == 303 + assert response.headers["location"] == "/runs/1" + + await started.wait() + release.set() + await runner.wait_for_job_run(1) + + +@pytest.mark.asyncio +async def test_run_status_reports_running_and_finished_metadata(): + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + with session_factory() as db: + step_run = _create_running_step_run(db) + + running = await main.run_status(step_run.id, None, db) + + assert running["status"] == "running" + assert running["exit_code"] is None + assert running["exit_label"] == "Pending" + assert running["finished_at"] is None + assert running["can_cancel"] is True + assert running["elapsed_seconds"] >= 0 + + step_run.status = "success" + step_run.exit_code = 0 + step_run.ended_at = step_run.started_at + db.commit() + + finished = await main.run_status(step_run.id, None, db) + + assert finished["status"] == "success" + assert finished["exit_code"] == 0 + assert finished["exit_label"] == "0" + assert finished["finished_at"] is not None + assert finished["can_cancel"] is False + + step_run.status = "canceled" + step_run.exit_code = None + db.commit() + + canceled = await main.run_status(step_run.id, None, db) + + assert canceled["status"] == "canceled" + assert canceled["exit_code"] is None + assert canceled["exit_label"] == "Canceled" + assert canceled["can_cancel"] is False + + +async def test_run_log_append_returns_new_bytes(): + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + log_path = Path(tmpdir) / "run.log" + log_path.write_text("first\n", encoding="utf-8") + offset = log_path.stat().st_size + with session_factory() as db: + step_run = _create_running_step_run(db, log_path=log_path) + with log_path.open("a", encoding="utf-8") as log_file: + log_file.write("second\n") + + append = await main.run_log_append(step_run.id, None, db, offset=offset) + + assert append == {"text": "second\n", "offset": log_path.stat().st_size} + + +async def test_job_run_status_includes_running_and_pending_steps(): + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + with session_factory() as db: + job = _create_job(db, step_names=["first", "second"]) + job_run = JobRunRecord( + job_id=job.id, + job_name=job.name, + trigger="manual", + status="running", + started_at=main.utc_now(), + ended_at=None, + ) + db.add(job_run) + db.flush() + first_step = job.steps[0] + step_run = JobStepRunRecord( + job_run_id=job_run.id, + step_id=first_step.id, + step_name=first_step.name, + argv_json='["rclone", "lsd", "secret:"]', + status="running", + exit_code=None, + log_path=str(Path(tmpdir) / "run.log"), + started_at=job_run.started_at, + ended_at=None, + ) + db.add(step_run) + db.commit() + + payload = await main.job_run_status(job_run.id, None, db) + + assert payload["status"] == "running" + assert payload["active_step_run_id"] == step_run.id + assert payload["can_cancel"] is True + assert [ + (step["name"], step["status"], step["run_id"]) for step in payload["steps"] + ] == [ + ("first", "running", step_run.id), + ("second", "pending", None), + ] + + +async def test_job_run_status_marks_unstarted_steps_canceled_after_cancellation(): + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + with session_factory() as db: + job = _create_job(db, step_names=["first", "second"]) + job_run = JobRunRecord( + job_id=job.id, + job_name=job.name, + trigger="manual", + status="canceled", + started_at=main.utc_now(), + ended_at=main.utc_now(), + ) + db.add(job_run) + db.flush() + first_step = job.steps[0] + step_run = JobStepRunRecord( + job_run_id=job_run.id, + step_id=first_step.id, + step_name=first_step.name, + argv_json='["rclone", "lsd", "secret:"]', + status="canceled", + exit_code=None, + log_path=str(Path(tmpdir) / "run.log"), + started_at=job_run.started_at, + ended_at=job_run.ended_at, + ) + db.add(step_run) + db.commit() + + payload = await main.job_run_status(job_run.id, None, db) + + assert [ + (step["name"], step["status"], step["run_id"]) for step in payload["steps"] + ] == [ + ("first", "canceled", step_run.id), + ("second", "canceled", None), + ] + + +async def test_job_run_detail_renders_whole_run_tracker(): + request = Request({"type": "http", "method": "GET", "path": "/job-runs/1", "headers": []}) + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + with session_factory() as db: + job = _create_job(db, step_names=["first", "second"]) + job_run = JobRunRecord( + job_id=job.id, + job_name=job.name, + trigger="manual", + status="running", + started_at=main.utc_now(), + ended_at=None, + ) + db.add(job_run) + db.flush() + db.add( + JobStepRunRecord( + job_run_id=job_run.id, + step_id=job.steps[0].id, + step_name="first", + argv_json='["rclone", "lsd", "secret:"]', + status="running", + exit_code=None, + log_path=str(Path(tmpdir) / "run.log"), + started_at=job_run.started_at, + ended_at=None, + ) + ) + db.commit() + + response = await main.job_run_detail(request, job_run.id, None, db) + html = response.body.decode() + + assert response.template.name == "job_run_detail.html" + assert "Current step" in html + assert "first" in html + assert "second" in html + assert "Active log" in html + + +async def test_history_page_receives_job_runs(): + request = Request({"type": "http", "method": "GET", "path": "/runs", "headers": []}) + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + with session_factory() as db: + job = _create_job(db) + db.add( + JobRunRecord( + job_id=job.id, + job_name=job.name, + trigger="manual", + status="success", + started_at=main.utc_now(), + ended_at=main.utc_now(), + ) + ) + db.commit() + + response = await main.runs(request, None, db) + + assert response.context["job_runs"][0].job_name == "backup" + + +async def test_history_page_paginates_independent_tables(): + request = Request({"type": "http", "method": "GET", "path": "/runs", "headers": []}) + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + with session_factory() as db: + job = _create_job(db) + for index in range(30): + job_run = JobRunRecord( + job_id=job.id, + job_name=f"job-{index}", + trigger="manual", + status="success", + started_at=main.utc_now(), + ended_at=main.utc_now(), + ) + db.add(job_run) + db.flush() + db.add( + JobStepRunRecord( + job_run_id=job_run.id, + step_id=job.steps[0].id, + step_name=f"step-{index}", + argv_json='["rclone", "lsd", "secret:"]', + status="success", + exit_code=0, + log_path=str(Path(tmpdir) / f"run-{index}.log"), + started_at=job_run.started_at, + ended_at=job_run.ended_at, + ) + ) + db.add( + ConsoleRunRecord( + status="success", + command=f"version-{index}", + argv_json='["rclone", "version"]', + exit_code=0, + log_path=str(Path(tmpdir) / f"console-{index}.log"), + started_at=job_run.started_at, + ended_at=job_run.ended_at, + ) + ) + db.commit() + + response = await main.runs(request, None, db, job_page=2, step_page=2, console_page=2) + + assert len(response.context["job_runs"]) == 5 + assert len(response.context["step_runs"]) == 5 + assert len(response.context["console_runs"]) == 5 + assert response.context["job_pagination"]["page"] == 2 + assert response.context["job_pagination"]["total_pages"] == 2 + assert response.context["job_pagination"]["has_previous"] is True + assert response.context["job_pagination"]["has_next"] is False + assert response.context["step_pagination"]["page"] == 2 + assert response.context["step_pagination"]["total_pages"] == 2 + assert response.context["console_pagination"]["page"] == 2 + assert response.context["console_pagination"]["total_pages"] == 2 + + +async def test_settings_prune_reports_deleted_count(): + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + log_path = Path(tmpdir) / "old.log" + log_path.write_text("old", encoding="utf-8") + with session_factory() as db: + step_run = _create_running_step_run(db, log_path=log_path) + step_run.started_at = step_run.started_at.replace(year=2020) + db.commit() + + response = await main.prune(None, db) + + assert response.status_code == 303 + assert response.headers["location"] == "/settings?pruned=1" + + +async def test_clear_history_removes_run_records_and_logs_but_keeps_jobs(): + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + log_path = Path(tmpdir) / "run.log" + console_log_path = Path(tmpdir) / "console.log" + log_path.write_text("run", encoding="utf-8") + console_log_path.write_text("console", encoding="utf-8") + with session_factory() as db: + job = _create_job(db) + step_run = _create_running_step_run(db, log_path=log_path) + step_run.job_run.job_id = job.id + db.add( + ConsoleRunRecord( + status="success", + command="version", + argv_json='["rclone", "version"]', + exit_code=0, + log_path=str(console_log_path), + started_at=main.utc_now(), + ended_at=main.utc_now(), + ) + ) + db.commit() + + response = await main.clear_history(None, db) + + assert response.status_code == 303 + assert response.headers["location"] == "/settings?cleared=2" + assert db.query(JobRecord).count() == 1 + assert db.query(JobStepRecord).count() == 1 + assert db.query(JobRunRecord).count() == 0 + assert db.query(JobStepRunRecord).count() == 0 + assert db.query(ConsoleRunRecord).count() == 0 + assert not log_path.exists() + assert not console_log_path.exists() + + +async def test_cancel_stale_running_job_marks_it_canceled(monkeypatch): + class FakeRunner: + def cancel_job_run(self, job_run_id): + return False + + monkeypatch.setattr(main, "runner", FakeRunner()) + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + with session_factory() as db: + step_run = _create_running_step_run(db) + job_run_id = step_run.job_run_id + + response = await main.cancel_job_run(job_run_id, None, db) + + assert response.status_code == 303 + job_run = db.get(JobRunRecord, job_run_id) + assert job_run.status == "canceled" + assert job_run.ended_at is not None + assert step_run.status == "canceled" + assert step_run.ended_at is not None + + +async def test_delete_individual_job_run_removes_steps_and_logs(): + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + log_path = Path(tmpdir) / "run.log" + log_path.write_text("run", encoding="utf-8") + with session_factory() as db: + step_run = _create_running_step_run(db, log_path=log_path) + job_run_id = step_run.job_run_id + + response = await main.delete_job_run(job_run_id, None, db) + + assert response.status_code == 303 + assert response.headers["location"] == "/runs" + assert db.get(JobRunRecord, job_run_id) is None + assert db.get(JobStepRunRecord, step_run.id) is None + assert not log_path.exists() + + +async def test_delete_individual_step_run_removes_log_only_for_that_step(): + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + log_path = Path(tmpdir) / "run.log" + log_path.write_text("run", encoding="utf-8") + with session_factory() as db: + step_run = _create_running_step_run(db, log_path=log_path) + job_run_id = step_run.job_run_id + + response = await main.delete_run(step_run.id, None, db) + + assert response.status_code == 303 + assert response.headers["location"] == "/runs" + assert db.get(JobRunRecord, job_run_id) is not None + assert db.get(JobStepRunRecord, step_run.id) is None + assert not log_path.exists() + + +async def test_delete_individual_console_run_removes_log(): + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + log_path = Path(tmpdir) / "console.log" + log_path.write_text("console", encoding="utf-8") + with session_factory() as db: + run = ConsoleRunRecord( + status="success", + command="version", + argv_json='["rclone", "version"]', + exit_code=0, + log_path=str(log_path), + started_at=main.utc_now(), + ended_at=main.utc_now(), + ) + db.add(run) + db.commit() + + response = await main.delete_console_run(run.id, None, db) + + assert response.status_code == 303 + assert response.headers["location"] == "/runs" + assert db.get(ConsoleRunRecord, run.id) is None + assert not log_path.exists() + + +@pytest.mark.asyncio +async def test_cancel_run_requests_parent_job_cancellation(monkeypatch): + called = [] + + class FakeRunner: + def cancel_job_run(self, job_run_id): + called.append(job_run_id) + return True + + monkeypatch.setattr(main, "runner", FakeRunner()) + with tempfile.TemporaryDirectory() as tmpdir: + session_factory = _session_factory(Path(tmpdir) / "runs.db") + with session_factory() as db: + step_run = _create_running_step_run(db) + + response = await main.cancel_run(step_run.id, None, db) + + assert response.status_code == 303 + assert response.headers["location"] == f"/runs/{step_run.id}" + assert called == [step_run.job_run_id] + + +def _session_factory(database_path: Path): + engine = create_engine(f"sqlite:///{database_path}", connect_args={"check_same_thread": False}) + Base.metadata.create_all(bind=engine) + return sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False) + + +def _create_job(db, step_names: list[str] | None = None): + job = JobRecord(name="backup", cron="", enabled=True, common_args="", env_json="{}") + for position, name in enumerate(step_names or ["one"], start=1): + job.steps.append(JobStepRecord(position=position, name=name, command="lsd secret:")) + db.add(job) + db.commit() + db.refresh(job) + return job + + +def _create_running_step_run(db, log_path: Path | None = None): + job_run = JobRunRecord( + job_id=1, + job_name="backup", + trigger="manual", + status="running", + started_at=main.utc_now(), + ended_at=None, + ) + db.add(job_run) + db.flush() + step_run = JobStepRunRecord( + job_run_id=job_run.id, + step_id=1, + step_name="one", + argv_json='["rclone", "lsd", "secret:"]', + status="running", + exit_code=None, + log_path=str(log_path or Path("missing.log")), + started_at=job_run.started_at, + ended_at=None, + ) + db.add(step_run) + db.commit() + db.refresh(step_run) + return step_run