Implement Rec - Phase 1
Quality Gate / gate (push) Successful in 2m31s

This commit is contained in:
Jim Lancaster
2026-09-02 15:59:56 -05:00
parent 02dca888d2
commit d5e798825d
10 changed files with 258 additions and 17 deletions
@@ -83,11 +83,13 @@ wrong has already caused a real defect in this repository, in both directions.
| Attribute | Audience | Reaches | Rule | | Attribute | Audience | Reaches | Rule |
| :--- | :--- | :--- | :--- | | :--- | :--- | :--- | :--- |
| `message` | User and API clients | `ErrorEnvelope.message`, UI notifications | Stays generic. Never embed exception text, provider payloads, or filesystem paths. | | `message` | User and API clients | `ErrorEnvelope.message`, UI notifications | Stays generic. Never embed exception text, provider payloads, or filesystem paths. |
| `detail` | Internal only | Logs, and `format_error_detail` -> `ExecutionAttempt.error_detail` and `MaintenanceRun.error_detail` | Carries the root cause. Never rendered to users or serialized into an envelope. | | `detail` | Internal only | Logs, and `format_error_detail` -> `ExecutionAttempt.error_detail` and `MaintenanceRun.error_detail` | Carries the root cause. Never rendered to users or serialized into an envelope without a sanitizing projection. |
- Putting root-cause data in `message` leaks infrastructure detail to users. - Putting root-cause data in `message` leaks infrastructure detail to users.
- Omitting it from `detail` silently degrades the provenance record this system exists to preserve — - Omitting it from `detail` silently degrades the provenance record this system exists to preserve —
a failed attempt whose `error_detail` says nothing is an attempt that cannot be diagnosed later. a failed attempt whose `error_detail` says nothing is an attempt that cannot be diagnosed later.
- Any render boundary that displays persisted `error_detail` must apply the same no-local-path rule
as `message`: sanitize machine-local absolute paths before the text becomes user-visible.
- When you raise from a caught exception, populate **both**: a generic `message` and a `detail` - When you raise from a caught exception, populate **both**: a generic `message` and a `detail`
carrying `type(exc).__name__` and the exception text, with `raise ... from exc`. carrying `type(exc).__name__` and the exception text, with `raise ... from exc`.
- `detail` is optional (`None`). A read path that assumes it is populated must handle its absence. - `detail` is optional (`None`). A read path that assumes it is populated must handle its absence.
+7 -3
View File
@@ -107,12 +107,16 @@ user-facing envelopes must not carry it. `AppError` therefore separates the two
| Field | Audience | Carries root cause | Surfaces | | Field | Audience | Carries root cause | Surfaces |
| --- | --- | --- | --- | | --- | --- | --- | --- |
| `message` | User-facing and API-facing | No | `show_error`, `build_error_envelope` | | `message` | User-facing and API-facing | No | `show_error`, `build_error_envelope` |
| `detail` | Internal only | Yes | `format_error_detail` (evidence), logs | | `detail` | Internal only | Yes | `format_error_detail` (evidence), logs, sanitized UI projection only |
`classify_unexpected_error` builds a generic `message` and puts the exception type and `classify_unexpected_error` builds a generic `message` and puts the exception type and
text on `detail`. Anything rendered to a user or serialized into an API envelope must text on `detail`. Anything rendered to a user or serialized into an API envelope must
read `message`; anything persisted as provenance or logged may read `detail`. read `message`; anything persisted as provenance or logged may read `detail`. When a UI
Enforced by `tests/test_errors.py::test_unexpected_error_does_not_leak_filesystem_paths`. surface needs to show persisted `error_detail`, it must route through a sanitizing
projection that preserves the category, suggestion, and error reference while reducing
machine-local absolute paths to basenames only.
Enforced by `tests/test_errors.py::test_unexpected_error_does_not_leak_filesystem_paths`
and `tests/test_error_message_safety.py`.
## Operator Recovery Guidance ## Operator Recovery Guidance
+3 -3
View File
@@ -747,11 +747,11 @@ set-membership.
def exception_detail(exc: BaseException) -> str: def exception_detail(exc: BaseException) -> str:
"""Internal-only root-cause text for AppError.detail. Never user-facing.""" """Internal-only root-cause text for AppError.detail. Never user-facing."""
def filesystem_error[E: AppError](
error_type: type[E], message: str, exc: OSError, *, suggestion: str def filesystem_error[E: AppError](error_type: type[E], message: str, exc: OSError, *, suggestion: str) -> E:
) -> E:
"""Build a filesystem AppError with a generic message and the path on detail.""" """Build a filesystem AppError with a generic message and the path on detail."""
# src/transcription/ui/components/error_presenter.py # src/transcription/ui/components/error_presenter.py
def display_failure_detail(error_detail: str | None) -> str | None: def display_failure_detail(error_detail: str | None) -> str | None:
"""Sanitize persisted failure detail for UI rendering (HIGH-01).""" """Sanitize persisted failure detail for UI rendering (HIGH-01)."""
+4 -2
View File
@@ -96,9 +96,10 @@ class PromptStore:
raise self._filesystem_error("Prompt directory could not be resolved", exc) from exc raise self._filesystem_error("Prompt directory could not be resolved", exc) from exc
if not root.is_dir(): if not root.is_dir():
raise PromptStoreError( raise PromptStoreError(
f"Prompt directory is unavailable: {root}", "Prompt directory is unavailable.",
category=ErrorCategory.INFRA_PERSISTENT, category=ErrorCategory.INFRA_PERSISTENT,
suggestion="Restore the configured prompt directory and its permissions.", suggestion="Restore the configured prompt directory and its permissions.",
detail=f"Prompt directory is unavailable: {root}",
) )
return root return root
@@ -185,7 +186,8 @@ class PromptStore:
@staticmethod @staticmethod
def _filesystem_error(message: str, exc: Exception) -> PromptStoreError: def _filesystem_error(message: str, exc: Exception) -> PromptStoreError:
return PromptStoreError( return PromptStoreError(
f"{message}: {exc}", message,
category=ErrorCategory.INFRA_PERSISTENT, category=ErrorCategory.INFRA_PERSISTENT,
suggestion="Check prompt directory permissions and available disk space, then retry.", suggestion="Check prompt directory permissions and available disk space, then retry.",
detail=f"{type(exc).__name__}: {exc}",
) )
+10 -4
View File
@@ -853,17 +853,19 @@ def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settin
if not prompt_path.exists() or not prompt_path.is_file(): if not prompt_path.exists() or not prompt_path.is_file():
raise PromptLoadError( raise PromptLoadError(
f"Prompt file not found: {prompt_path}", f"Prompt file not found: {prompt_path.name}",
category=ErrorCategory.INFRA_PERSISTENT, category=ErrorCategory.INFRA_PERSISTENT,
suggestion="Verify PROMPT_DIR and prompt file configuration, then retry.", suggestion="Verify PROMPT_DIR and prompt file configuration, then retry.",
detail=f"Prompt file missing at {prompt_path}",
) )
prompt_text = prompt_path.read_text(encoding="utf-8").strip() prompt_text = prompt_path.read_text(encoding="utf-8").strip()
if not prompt_text: if not prompt_text:
raise PromptLoadError( raise PromptLoadError(
f"Prompt file is empty: {prompt_path}", f"Prompt file is empty: {prompt_path.name}",
category=ErrorCategory.VALIDATION, category=ErrorCategory.VALIDATION,
suggestion="Populate the prompt file with valid instructions and retry.", suggestion="Populate the prompt file with valid instructions and retry.",
detail=f"Prompt file is empty at {prompt_path}",
) )
logger.info("Loaded prompt artifact: %s", prompt_path) logger.info("Loaded prompt artifact: %s", prompt_path)
@@ -911,9 +913,10 @@ def load_source_payload(source_path: str | Path) -> tuple[bytes, str]:
if not path.exists() or not path.is_file(): if not path.exists() or not path.is_file():
raise TranscriptionError( raise TranscriptionError(
f"Source file not found: {path}", f"Source file not found: {path.name}",
category=ErrorCategory.NOT_FOUND, category=ErrorCategory.NOT_FOUND,
suggestion="Verify the Source file exists and retry from the jobs page.", suggestion="Verify the Source file exists and retry from the jobs page.",
detail=f"Source file not found at {path}",
) )
content = path.read_bytes() content = path.read_bytes()
@@ -930,6 +933,7 @@ def handle_transcription_errors():
"Provider authentication failed", "Provider authentication failed",
category=ErrorCategory.INFRA_PERSISTENT, category=ErrorCategory.INFRA_PERSISTENT,
suggestion="Verify provider API credentials and retry.", suggestion="Verify provider API credentials and retry.",
detail=f"{type(exc).__name__}: {exc}",
) from exc ) from exc
except ProviderResponseError as exc: except ProviderResponseError as exc:
raise TranscriptionError( raise TranscriptionError(
@@ -937,11 +941,13 @@ def handle_transcription_errors():
category=ErrorCategory.EXTERNAL_PROVIDER, category=ErrorCategory.EXTERNAL_PROVIDER,
suggestion="Retry once. If it persists, try a different provider model or inspect provider status.", suggestion="Retry once. If it persists, try a different provider model or inspect provider status.",
retriable=True, retriable=True,
detail=f"{type(exc).__name__}: {exc}",
) from exc ) from exc
except ProviderError as exc: except ProviderError as exc:
raise TranscriptionError( raise TranscriptionError(
f"Provider transcription failed: {exc}", "Provider transcription failed",
category=ErrorCategory.EXTERNAL_PROVIDER, category=ErrorCategory.EXTERNAL_PROVIDER,
suggestion="Retry the transcription from jobs. If repeated, check provider availability.", suggestion="Retry the transcription from jobs. If repeated, check provider availability.",
retriable=True, retriable=True,
detail=f"{type(exc).__name__}: {exc}",
) from exc ) from exc
@@ -2,9 +2,12 @@
from __future__ import annotations from __future__ import annotations
import re
from collections.abc import Awaitable from collections.abc import Awaitable
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import PurePosixPath
from pathlib import PureWindowsPath
from typing import TypeVar from typing import TypeVar
from nicegui import ui from nicegui import ui
@@ -15,6 +18,12 @@ from transcription.errors import classify_unexpected_error
T = TypeVar("T") T = TypeVar("T")
_QUOTED_ABSOLUTE_PATH_RE = re.compile(
r"""(?P<quote>['"])(?P<path>(?:[A-Za-z]:[\\/][^'"]+|/(?!uploads/)[^'"]+))(?P=quote)"""
)
_UNQUOTED_WINDOWS_PATH_RE = re.compile(r"""(?P<path>[A-Za-z]:[\\/][^\s|]+)""")
_UNQUOTED_POSIX_PATH_RE = re.compile(r"""(?P<path>(?<![A-Za-z0-9:])/(?!uploads/)[^\s|]+)""")
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class UiActionOutcome[T]: class UiActionOutcome[T]:
@@ -66,6 +75,35 @@ def show_error(exc: Exception, *, title: str, operation: str) -> None:
ui.label(f"Category: {display_error_category(error)}").classes("text-caption") ui.label(f"Category: {display_error_category(error)}").classes("text-caption")
def display_failure_detail(error_detail: str | None) -> str | None:
"""Render persisted failure detail without machine-local paths."""
candidate = (error_detail or "").strip()
if not candidate:
return None
sanitized = _QUOTED_ABSOLUTE_PATH_RE.sub(_replace_quoted_absolute_path, candidate)
sanitized = _UNQUOTED_WINDOWS_PATH_RE.sub(_replace_unquoted_absolute_path, sanitized)
sanitized = _UNQUOTED_POSIX_PATH_RE.sub(_replace_unquoted_absolute_path, sanitized)
return sanitized
def display_error_category(error: AppError) -> str: def display_error_category(error: AppError) -> str:
"""Return the canonical UI-facing category label for an AppError.""" """Return the canonical UI-facing category label for an AppError."""
return canonical_error_category(error) return canonical_error_category(error)
def _replace_quoted_absolute_path(match: re.Match[str]) -> str:
quote = match.group("quote")
return f"{quote}{_basename_for_absolute_path(match.group('path'))}{quote}"
def _replace_unquoted_absolute_path(match: re.Match[str]) -> str:
return _basename_for_absolute_path(match.group("path"))
def _basename_for_absolute_path(path: str) -> str:
if path.startswith("/uploads/"):
return path
if re.match(r"^[A-Za-z]:[\\/]", path):
return PureWindowsPath(path).name or "unknown"
return PurePosixPath(path).name or "unknown"
+2 -1
View File
@@ -20,6 +20,7 @@ from transcription.services.people import PeopleService
from transcription.services.prompts import PromptStore from transcription.services.prompts import PromptStore
from transcription.ui.components.app_shell import render_navigation_header from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.cards import archival_card from transcription.ui.components.cards import archival_card
from transcription.ui.components.error_presenter import display_failure_detail
from transcription.ui.components.error_presenter import run_ui_action from transcription.ui.components.error_presenter import run_ui_action
from transcription.ui.components.primitives import destructive_button from transcription.ui.components.primitives import destructive_button
from transcription.ui.components.primitives import render_empty_state from transcription.ui.components.primitives import render_empty_state
@@ -559,7 +560,7 @@ def register_page(*, settings: Settings) -> None: # noqa: PLR0915
"duration": _format_duration(started_at=run.started_at, finished_at=run.finished_at), "duration": _format_duration(started_at=run.started_at, finished_at=run.finished_at),
"summary": run.summary or "-", "summary": run.summary or "-",
"log_path": run.log_path or "", "log_path": run.log_path or "",
"error_detail": run.error_detail or "", "error_detail": display_failure_detail(run.error_detail) or "",
} }
for run in runs for run in runs
] ]
+7 -3
View File
@@ -33,6 +33,7 @@ from transcription.ui.components.confirm_delete import render_delete_blocked_not
from transcription.ui.components.data_display import archival_badge from transcription.ui.components.data_display import archival_badge
from transcription.ui.components.data_display import metadata_row from transcription.ui.components.data_display import metadata_row
from transcription.ui.components.document_panzoom import render_document_panzoom from transcription.ui.components.document_panzoom import render_document_panzoom
from transcription.ui.components.error_presenter import display_failure_detail
from transcription.ui.components.error_presenter import run_ui_action from transcription.ui.components.error_presenter import run_ui_action
from transcription.ui.components.error_presenter import show_error from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.formatters import parse_uuid from transcription.ui.components.formatters import parse_uuid
@@ -118,7 +119,7 @@ def register_page() -> None: # noqa: PLR0915
document_id=source.document_id, document_id=source.document_id,
document_name=source.document_name, document_name=source.document_name,
job_source_status=source.latest_status.value if source.latest_status else "unprocessed", job_source_status=source.latest_status.value if source.latest_status else "unprocessed",
job_source_error_detail=source.latest_error_detail, job_source_error_detail=display_failure_detail(source.latest_error_detail),
) )
for source in sorted(sources, key=lambda item: (item.page_number, item.upload_name.casefold())) for source in sorted(sources, key=lambda item: (item.page_number, item.upload_name.casefold()))
] ]
@@ -478,10 +479,13 @@ def _render_source_job_metadata_zone(
else "unknown", else "unknown",
) )
if latest_attempt is not None and latest_attempt.attempt.error_detail: failure_detail = (
display_failure_detail(latest_attempt.attempt.error_detail) if latest_attempt is not None else None
)
if failure_detail:
with ui.column().classes("w-full mt-2"): with ui.column().classes("w-full mt-2"):
ui.label("Failure Detail:").classes("ui-text-muted text-xs mb-1") ui.label("Failure Detail:").classes("ui-text-muted text-xs mb-1")
ui.label(latest_attempt.attempt.error_detail).classes("p-2 ui-note-box text-xs") ui.label(failure_detail).classes("p-2 ui-note-box text-xs")
_render_provider_evidence(latest_attempt=latest_attempt) _render_provider_evidence(latest_attempt=latest_attempt)
+145
View File
@@ -0,0 +1,145 @@
"""AST guards for user-facing error safety and UI failure-detail rendering."""
from __future__ import annotations
import ast
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
SOURCE_DIR = PROJECT_ROOT / "src" / "transcription"
UI_DIR = SOURCE_DIR / "ui"
_RAW_DETAIL_ATTRIBUTES = frozenset({"error_detail", "latest_error_detail"})
_SUSPICIOUS_FORMATTED_NAMES = frozenset({"path", "root", "dir", "exc", "err", "e"})
def _python_files(root: Path) -> list[Path]:
return sorted(root.rglob("*.py"))
def _parent_map(tree: ast.AST) -> dict[ast.AST, ast.AST]:
parents: dict[ast.AST, ast.AST] = {}
for node in ast.walk(tree):
for child in ast.iter_child_nodes(node):
parents[child] = node
return parents
def _is_wrapped_in_display_failure_detail(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> bool:
current = node
while current in parents:
current = parents[current]
if not isinstance(current, ast.Call):
continue
func = current.func
if isinstance(func, ast.Name) and func.id == "display_failure_detail":
return True
return False
def _ui_raw_detail_reads() -> dict[str, list[int]]:
violations: dict[str, list[int]] = {}
for path in _python_files(UI_DIR):
tree = ast.parse(path.read_text(encoding="utf-8"))
parents = _parent_map(tree)
found = sorted(
node.lineno
for node in ast.walk(tree)
if isinstance(node, ast.Attribute)
and node.attr in _RAW_DETAIL_ATTRIBUTES
and not _is_wrapped_in_display_failure_detail(node, parents)
)
if found:
violations[str(path.relative_to(PROJECT_ROOT)).replace("\\", "/")] = found
return violations
def _class_bases_by_name() -> dict[str, set[str]]:
bases: dict[str, set[str]] = {}
for path in _python_files(SOURCE_DIR):
tree = ast.parse(path.read_text(encoding="utf-8"))
for node in tree.body:
if not isinstance(node, ast.ClassDef):
continue
inherited = set()
for base in node.bases:
if isinstance(base, ast.Name):
inherited.add(base.id)
elif isinstance(base, ast.Attribute):
inherited.add(base.attr)
bases[node.name] = inherited
return bases
def _app_error_subclasses() -> set[str]:
bases = _class_bases_by_name()
subclasses = {"AppError"}
changed = True
while changed:
changed = False
for name, inherited in bases.items():
if name in subclasses:
continue
if inherited & subclasses:
subclasses.add(name)
changed = True
subclasses.remove("AppError")
return subclasses
def _formatted_name_ids(node: ast.AST) -> set[str]:
return {child.id for child in ast.walk(node) if isinstance(child, ast.Name)}
def _is_safe_basename_projection(node: ast.AST) -> bool:
return isinstance(node, ast.Attribute) and node.attr == "name"
def _is_suspicious_name(name: str) -> bool:
if name in _SUSPICIOUS_FORMATTED_NAMES:
return True
return any(name.endswith(f"_{suffix}") for suffix in _SUSPICIOUS_FORMATTED_NAMES - {"e"})
def _user_message_interpolation_violations() -> dict[str, list[str]]:
violations: dict[str, list[str]] = {}
error_types = _app_error_subclasses()
for path in _python_files(SOURCE_DIR):
tree = ast.parse(path.read_text(encoding="utf-8"))
found: list[str] = []
for node in ast.walk(tree):
if not isinstance(node, ast.Raise):
continue
if not isinstance(node.exc, ast.Call):
continue
func = node.exc.func
if not isinstance(func, ast.Name) or func.id not in error_types:
continue
if not node.exc.args:
continue
message = node.exc.args[0]
if not isinstance(message, ast.JoinedStr):
continue
formatted_names = {
name
for value in message.values
if isinstance(value, ast.FormattedValue)
if not _is_safe_basename_projection(value.value)
for name in _formatted_name_ids(value.value)
}
suspicious = sorted(name for name in formatted_names if _is_suspicious_name(name))
if suspicious:
found.append(f"L{node.lineno}: {', '.join(suspicious)}")
if found:
violations[str(path.relative_to(PROJECT_ROOT)).replace("\\", "/")] = found
return violations
def test_ui_modules_only_render_failure_detail_through_projection():
"""HIGH-01: UI must sanitize persisted failure detail before rendering it."""
assert _ui_raw_detail_reads() == {}
def test_user_facing_app_error_messages_do_not_interpolate_paths_or_exceptions():
"""HIGH-02 / MED-07: keep paths and exception text out of AppError.message."""
assert _user_message_interpolation_violations() == {}
+39
View File
@@ -3,6 +3,7 @@ import pytest
from transcription.errors import AppError from transcription.errors import AppError
from transcription.errors import ErrorCategory from transcription.errors import ErrorCategory
from transcription.ui.components.error_presenter import display_error_category from transcription.ui.components.error_presenter import display_error_category
from transcription.ui.components.error_presenter import display_failure_detail
from transcription.ui.components.error_presenter import run_ui_action from transcription.ui.components.error_presenter import run_ui_action
@@ -17,6 +18,44 @@ def test_display_error_category_uses_canonical_taxonomy():
assert display_error_category(AppError("x", category=ErrorCategory.INFRA_PERSISTENT)) == "internal" assert display_error_category(AppError("x", category=ErrorCategory.INFRA_PERSISTENT)) == "internal"
def test_display_failure_detail_returns_none_for_missing_detail():
assert display_failure_detail(None) is None
@pytest.mark.parametrize(
("raw_detail", "expected_basename"),
[
(
(
"[internal_unexpected_error] Unexpected error during worker.process_job. | "
"detail=PermissionError: [Errno 13] Permission denied: '/app/uploads/documents/abc/page-1.jpg' | "
"suggestion=Retry once. | error_id=deadbeef"
),
"page-1.jpg",
),
(
(
"[infrastructure_persistent_error] Prompt file not found: transcribe_document.md | "
"detail=FileNotFoundError: [Errno 2] No such file or directory: "
r"'C:\app\prompts\transcribe_document.md' | "
"suggestion=Verify PROMPT_DIR. | error_id=feedface"
),
"transcribe_document.md",
),
],
)
def test_display_failure_detail_strips_absolute_paths_from_detail_segment(raw_detail, expected_basename):
rendered = display_failure_detail(raw_detail)
assert rendered is not None
assert rendered.startswith("[")
assert "suggestion=" in rendered
assert "error_id=" in rendered
assert expected_basename in rendered
assert "/app/uploads/documents/abc/page-1.jpg" not in rendered
assert r"C:\app\prompts\transcribe_document.md" not in rendered
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_run_ui_action_returns_success_value(): async def test_run_ui_action_returns_success_value():
outcome = await run_ui_action( outcome = await run_ui_action(