Add ability to edit safe .env.example options to Settings

src/transcription/ui/pages/settings_page.py

@@ -21,6 +21,8 @@ from transcription.ui.components.primitives import section_header_row
 from transcription.ui.components.table.registry import render_registry_table
 from transcription.ui.homepage_store import read_homepage_markdown
 from transcription.ui.homepage_store import save_homepage_markdown
+from transcription.ui.runtime_settings_store import read_runtime_settings_snapshot
+from transcription.ui.runtime_settings_store import save_runtime_settings
 from transcription.ui.theme import page_header
 
 from ...db.session import SessionFactoryDep
@@ -497,14 +499,78 @@ def register_page(*, settings: Settings) -> None:  # noqa: PLR0915
                     with ui.row().classes("items-center gap-2"):
                         ui.button("Save home text", icon="save", on_click=save_home_text).classes("ui-btn-primary")
 
+            @ui.refreshable
+            async def render_runtime_settings() -> None:
+                with archival_card("Runtime Settings"):
+                    ui.label("Edit non-secret .env settings. Secret fields are intentionally excluded.").classes(
+                        "text-xs ui-text-muted"
+                    )
+                    ui.label("Changes are persisted to .env and apply after restart.").classes(
+                        "text-xs ui-text-muted mb-3"
+                    )
+                    snapshot_outcome = await run_ui_action(
+                        operation="settings.runtime.read",
+                        title="Runtime settings unavailable",
+                        action=lambda: _read_runtime_settings_snapshot(settings),
+                    )
+                    if not snapshot_outcome.ok or snapshot_outcome.value is None:
+                        return
+                    snapshot = snapshot_outcome.value
+                    field_controls: dict[str, Any] = {}
+                    for field in snapshot.fields:
+                        with ui.column().classes("w-full gap-1 py-2 ui-header-divider"):
+                            ui.label(field.label).classes("text-sm font-semibold")
+                            ui.label(field.description).classes("text-xs ui-text-muted")
+                            if field.control == "bool":
+                                control = ui.checkbox(field.env_key, value=bool(field.value))
+                            elif field.control == "select":
+                                control = (
+                                    ui.select(
+                                        list(field.options),
+                                        label=field.env_key,
+                                        value=str(field.value),
+                                    )
+                                    .props("outlined")
+                                    .classes("w-full")
+                                )
+                            else:
+                                control = (
+                                    ui.input(field.env_key, value=str(field.value)).props("outlined").classes("w-full")
+                                )
+                            field_controls[field.field_name] = control
+
+                    async def save_runtime() -> None:
+                        updates: dict[str, str | bool] = {}
+                        for field in snapshot.fields:
+                            control = field_controls[field.field_name]
+                            if field.control == "bool":
+                                updates[field.field_name] = bool(control.value)
+                            else:
+                                updates[field.field_name] = str(control.value or "")
+                        save_outcome = await run_ui_action(
+                            operation="settings.runtime.write",
+                            title="Runtime settings save failed",
+                            action=lambda: _write_runtime_settings(settings=settings, updates=updates),
+                        )
+                        if not save_outcome.ok:
+                            return
+                        ui.notify("Runtime settings saved to .env", type="positive")
+                        render_runtime_settings.refresh()
+
+                    with ui.row().classes("items-center gap-2 mt-3"):
+                        ui.button("Save runtime settings", icon="save", on_click=save_runtime).classes("ui-btn-primary")
+
             with ui.tabs().classes("w-full") as tabs:
+                runtime_settings_tab = ui.tab("Runtime Settings")
                 document_types_tab = ui.tab("Document Types")
                 person_roles_tab = ui.tab("Person Roles")
                 tags_tab = ui.tab("Tags")
                 prompts_tab = ui.tab("Prompts")
                 home_page_text_tab = ui.tab("Home Page Text")
 
-            with ui.tab_panels(tabs, value=document_types_tab).classes("w-full"):
+            with ui.tab_panels(tabs, value=runtime_settings_tab).classes("w-full"):
+                with ui.tab_panel(runtime_settings_tab):
+                    await render_runtime_settings()
                 with ui.tab_panel(document_types_tab):
                     await render_document_types()
                 with ui.tab_panel(person_roles_tab):
@@ -546,3 +612,11 @@ async def _read_home_page_text(settings: Settings) -> str:
 
 async def _write_home_page_text(settings: Settings, markdown_text: str) -> None:
     await run_blocking(save_homepage_markdown, markdown_text, settings=settings)
+
+
+async def _read_runtime_settings_snapshot(settings: Settings):
+    return await run_blocking(read_runtime_settings_snapshot, settings=settings)
+
+
+async def _write_runtime_settings(*, settings: Settings, updates: dict[str, str | bool]):
+    return await run_blocking(save_runtime_settings, settings=settings, updates=updates)

src/transcription/ui/runtime_settings_store.py

@@ -0,0 +1,465 @@
+"""Safe runtime settings catalog and .env persistence helpers."""
+
+from __future__ import annotations
+
+import re
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal
+
+from pydantic import ValidationError
+
+from transcription.config import Provider
+from transcription.config import Settings
+from transcription.errors import AppError
+from transcription.errors import ErrorCategory
+
+FieldControl = Literal["text", "bool", "select"]
+
+
+@dataclass(frozen=True, slots=True)
+class RuntimeSettingField:
+    """UI metadata and value for a safe, editable runtime setting."""
+
+    field_name: str
+    env_key: str
+    label: str
+    description: str
+    control: FieldControl
+    value: str | bool
+    options: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True, slots=True)
+class RuntimeSettingsSnapshot:
+    """Typed snapshot of runtime settings editable in UI."""
+
+    env_file_path: Path
+    fields: tuple[RuntimeSettingField, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class RuntimeSettingDescriptor:
+    field_name: str
+    env_key: str
+    label: str
+    description: str
+    control: FieldControl
+    options: tuple[str, ...] = ()
+
+
+SETTINGS_UI_EXCLUDED_FIELDS = frozenset(
+    {
+        "openrouter_api_key",
+        "database",
+    }
+)
+
+RUNTIME_SETTINGS_CATALOG: tuple[RuntimeSettingDescriptor, ...] = (
+    RuntimeSettingDescriptor("host", "HOST", "Host", "Server bind host.", "text"),
+    RuntimeSettingDescriptor("port", "PORT", "Port", "Server bind port.", "text"),
+    RuntimeSettingDescriptor(
+        "log_level",
+        "LOG_LEVEL",
+        "Log level",
+        "Application log verbosity.",
+        "select",
+        options=("critical", "error", "warning", "info", "debug", "trace"),
+    ),
+    RuntimeSettingDescriptor("reload", "RELOAD", "Reload", "Auto-reload on code changes.", "bool"),
+    RuntimeSettingDescriptor("log_dir", "LOG_DIR", "Log directory", "Directory for rotating log files.", "text"),
+    RuntimeSettingDescriptor(
+        "log_file_name",
+        "LOG_FILE_NAME",
+        "Log file name",
+        "Active rotating log filename.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "log_file_max_bytes",
+        "LOG_FILE_MAX_BYTES",
+        "Log file max bytes",
+        "Maximum bytes per log file before rotation.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "log_file_backup_count",
+        "LOG_FILE_BACKUP_COUNT",
+        "Log file backup count",
+        "Number of rotated log files to keep.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "provider",
+        "PROVIDER",
+        "Provider",
+        "Transcription provider key.",
+        "select",
+        options=tuple(member.value for member in Provider),
+    ),
+    RuntimeSettingDescriptor(
+        "provider_model",
+        "PROVIDER_MODEL",
+        "Provider model",
+        "Default model id used for transcription jobs.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "provider_models",
+        "PROVIDER_MODELS",
+        "Provider models (JSON array)",
+        "Allowed model ids as a JSON array.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "openrouter_http_referer",
+        "OPENROUTER_HTTP_REFERER",
+        "OpenRouter HTTP referer",
+        "Optional header value sent to OpenRouter.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "openrouter_app_title",
+        "OPENROUTER_APP_TITLE",
+        "OpenRouter app title",
+        "Optional app title sent to OpenRouter.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "default_prompt_name",
+        "DEFAULT_PROMPT_NAME",
+        "Default prompt file",
+        "Prompt filename used for new transcription requests.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "transcription_temperature",
+        "TRANSCRIPTION_TEMPERATURE",
+        "Temperature",
+        "Optional sampling temperature (0.0 to 2.0).",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "transcription_top_p",
+        "TRANSCRIPTION_TOP_P",
+        "Top P",
+        "Optional nucleus sampling probability (0.0 to 1.0).",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "environment",
+        "ENVIRONMENT",
+        "Environment",
+        "Runtime environment mode.",
+        "select",
+        options=("development", "test", "production"),
+    ),
+    RuntimeSettingDescriptor(
+        "transcription_commit",
+        "TRANSCRIPTION_COMMIT",
+        "Transcription commit",
+        "Optional build or commit identifier for evidence provenance.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "bootstrap_schema_on_startup",
+        "BOOTSTRAP_SCHEMA_ON_STARTUP",
+        "Bootstrap schema on startup",
+        "Create schema on startup when enabled.",
+        "bool",
+    ),
+    RuntimeSettingDescriptor(
+        "sqlite_check_same_thread",
+        "SQLITE_CHECK_SAME_THREAD",
+        "SQLite check same thread",
+        "SQLite engine same-thread flag.",
+        "bool",
+    ),
+    RuntimeSettingDescriptor(
+        "upload_dir",
+        "UPLOAD_DIR",
+        "Upload directory",
+        "Root directory for uploaded media and generated files.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "prompt_dir",
+        "PROMPT_DIR",
+        "Prompt directory",
+        "Directory containing editable markdown prompts.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "database_backup_dir",
+        "DATABASE_BACKUP_DIR",
+        "Database backup directory",
+        "Directory for generated database backups.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "worker_max_retries",
+        "WORKER_MAX_RETRIES",
+        "Worker max retries",
+        "Maximum retries per source attempt.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "worker_provider_timeout_seconds",
+        "WORKER_PROVIDER_TIMEOUT_SECONDS",
+        "Worker provider timeout seconds",
+        "Provider call timeout budget in seconds.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "worker_stale_job_seconds",
+        "WORKER_STALE_JOB_SECONDS",
+        "Worker stale job seconds",
+        "Seconds before queued jobs are considered stale.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "worker_retry_backoff_seconds",
+        "WORKER_RETRY_BACKOFF_SECONDS",
+        "Worker retry backoff seconds",
+        "Base backoff delay between retries.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "worker_shutdown_grace_seconds",
+        "WORKER_SHUTDOWN_GRACE_SECONDS",
+        "Worker shutdown grace seconds",
+        "Grace period before forced worker shutdown.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "worker_poll_interval_seconds",
+        "WORKER_POLL_INTERVAL_SECONDS",
+        "Worker poll interval seconds",
+        "Worker polling interval for queued jobs.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "worker_min_transcription_chars",
+        "WORKER_MIN_TRANSCRIPTION_CHARS",
+        "Worker min transcription chars",
+        "Minimum character threshold for successful transcription.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "worker_min_transcription_lines",
+        "WORKER_MIN_TRANSCRIPTION_LINES",
+        "Worker min transcription lines",
+        "Minimum line threshold for successful transcription.",
+        "text",
+    ),
+    RuntimeSettingDescriptor(
+        "worker_fail_on_finish_reason_length",
+        "WORKER_FAIL_ON_FINISH_REASON_LENGTH",
+        "Fail on finish reason length",
+        "Treat provider finish_reason=length as failure when enabled.",
+        "bool",
+    ),
+)
+
+
+def settings_catalog_field_names() -> frozenset[str]:
+    return frozenset(item.field_name for item in RUNTIME_SETTINGS_CATALOG)
+
+
+def read_runtime_settings_snapshot(*, settings: Settings, env_file_path: Path | None = None) -> RuntimeSettingsSnapshot:
+    resolved_env_path = _resolve_env_file_path(settings=settings, env_file_path=env_file_path)
+    fields: list[RuntimeSettingField] = []
+    for descriptor in RUNTIME_SETTINGS_CATALOG:
+        value = getattr(settings, descriptor.field_name)
+        fields.append(
+            RuntimeSettingField(
+                field_name=descriptor.field_name,
+                env_key=descriptor.env_key,
+                label=descriptor.label,
+                description=descriptor.description,
+                control=descriptor.control,
+                value=_display_value(value),
+                options=descriptor.options,
+            )
+        )
+    return RuntimeSettingsSnapshot(
+        env_file_path=resolved_env_path,
+        fields=tuple(fields),
+    )
+
+
+def save_runtime_settings(
+    *,
+    settings: Settings,
+    updates: dict[str, str | bool],
+    env_file_path: Path | None = None,
+) -> RuntimeSettingsSnapshot:
+    descriptors = {item.field_name: item for item in RUNTIME_SETTINGS_CATALOG}
+    unknown_keys = sorted(set(updates) - set(descriptors))
+    if unknown_keys:
+        raise AppError(
+            "One or more settings fields cannot be edited from this page.",
+            category=ErrorCategory.VALIDATION,
+            suggestion="Refresh the page and submit only editable fields.",
+            detail=f"Unknown settings keys: {unknown_keys}",
+        )
+
+    resolved_env_path = _resolve_env_file_path(settings=settings, env_file_path=env_file_path)
+    original_lines = _read_env_lines(resolved_env_path)
+    env_map = _parse_env_map(original_lines)
+    mutable_lines = list(original_lines)
+
+    for field_name, raw_value in updates.items():
+        descriptor = descriptors[field_name]
+        encoded = _encode_field_value(raw_value)
+        if encoded == "":
+            _delete_env_key(mutable_lines, descriptor.env_key)
+            env_map.pop(descriptor.env_key, None)
+            continue
+        _upsert_env_key(mutable_lines, descriptor.env_key, encoded)
+        env_map[descriptor.env_key] = encoded
+
+    _validate_candidate_env(env_map=env_map, env_file_path=resolved_env_path)
+    _write_env_lines_atomic(path=resolved_env_path, lines=mutable_lines)
+    refreshed = Settings(_env_file=resolved_env_path, _cli_parse_args=False)
+    return read_runtime_settings_snapshot(settings=refreshed, env_file_path=resolved_env_path)
+
+
+def _resolve_env_file_path(*, settings: Settings, env_file_path: Path | None) -> Path:
+    if env_file_path is not None:
+        return env_file_path
+
+    configured = settings.model_config.get("env_file")
+    if configured is None:
+        return Path(".env")
+    if isinstance(configured, Path):
+        return Path(configured)
+    if isinstance(configured, str):
+        return Path(configured)
+    if isinstance(configured, (list, tuple)) and configured:
+        first = configured[0]
+        if isinstance(first, Path):
+            return Path(first)
+        if isinstance(first, str):
+            return Path(first)
+    return Path(".env")
+
+
+def _display_value(value: object) -> str | bool:
+    if isinstance(value, bool):
+        return value
+    if value is None:
+        return ""
+    if isinstance(value, tuple):
+        values = [str(item) for item in value]
+        escaped = ",".join(f'"{_escape_json_string(item)}"' for item in values)
+        return f"[{escaped}]"
+    if isinstance(value, Path):
+        return value.as_posix()
+    if isinstance(value, Provider):
+        return value.value
+    return str(value)
+
+
+def _encode_field_value(raw_value: str | bool) -> str:
+    if isinstance(raw_value, bool):
+        return "true" if raw_value else "false"
+    text = str(raw_value).strip()
+    if text == "":
+        return ""
+    if text.startswith("[") and text.endswith("]"):
+        return text
+    if " " in text or "#" in text:
+        escaped = text.replace("\\", "\\\\").replace('"', '\\"')
+        return f'"{escaped}"'
+    return text
+
+
+def _validate_candidate_env(*, env_map: dict[str, str], env_file_path: Path) -> None:
+    serialized = "\n".join(f"{key}={value}" for key, value in env_map.items()) + "\n"
+    with tempfile.NamedTemporaryFile(mode="w", delete=False, encoding="utf-8", newline="\n", suffix=".env") as handle:
+        temp_path = Path(handle.name)
+        handle.write(serialized)
+    try:
+        Settings(_env_file=temp_path, _cli_parse_args=False)
+    except ValidationError as exc:
+        raise AppError(
+            "One or more settings values are invalid.",
+            category=ErrorCategory.VALIDATION,
+            suggestion="Review the highlighted values and use the same formats shown in .env.example.",
+            detail=f"Invalid runtime settings for {env_file_path}: {exc}",
+        ) from exc
+    finally:
+        temp_path.unlink(missing_ok=True)
+
+
+def _read_env_lines(path: Path) -> list[str]:
+    if not path.exists():
+        return []
+    return path.read_text(encoding="utf-8").splitlines()
+
+
+def _parse_env_map(lines: list[str]) -> dict[str, str]:
+    env_map: dict[str, str] = {}
+    for line in lines:
+        parsed = _parse_env_line(line)
+        if parsed is None:
+            continue
+        key, value = parsed
+        env_map[key] = value
+    return env_map
+
+
+def _parse_env_line(line: str) -> tuple[str, str] | None:
+    if not line or line.lstrip().startswith("#"):
+        return None
+    match = re.match(r"^\s*([A-Za-z_][A-Za-z0-9_]*(?:__[A-Za-z0-9_]+)*)\s*=\s*(.*)$", line)
+    if not match:
+        return None
+    return match.group(1), match.group(2)
+
+
+def _delete_env_key(lines: list[str], key: str) -> None:
+    for index, line in enumerate(lines):
+        parsed = _parse_env_line(line)
+        if parsed is None:
+            continue
+        current_key, _ = parsed
+        if current_key == key:
+            del lines[index]
+            return
+
+
+def _upsert_env_key(lines: list[str], key: str, value: str) -> None:
+    for index, line in enumerate(lines):
+        parsed = _parse_env_line(line)
+        if parsed is None:
+            continue
+        current_key, _ = parsed
+        if current_key == key:
+            lines[index] = f"{key}={value}"
+            return
+    lines.append(f"{key}={value}")
+
+
+def _write_env_lines_atomic(*, path: Path, lines: list[str]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    content = "\n".join(lines).rstrip("\n")
+    if content:
+        content += "\n"
+    with tempfile.NamedTemporaryFile(
+        mode="w",
+        delete=False,
+        dir=path.parent,
+        encoding="utf-8",
+        newline="\n",
+        suffix=".env.tmp",
+    ) as handle:
+        temp_path = Path(handle.name)
+        handle.write(content)
+    temp_path.replace(path)
+
+
+def _escape_json_string(value: str) -> str:
+    return value.replace("\\", "\\\\").replace('"', '\\"')