diff --git a/docs/ui/pages/jobs.md b/docs/ui/pages/jobs.md index 36b4c53..b5433e3 100644 --- a/docs/ui/pages/jobs.md +++ b/docs/ui/pages/jobs.md @@ -46,7 +46,7 @@ Jobs manages transcription processing runs. A Job belongs to one Document, links - Polling stops when the Job becomes terminal or a refresh fails. - Queued and processing Jobs expose **Cancel**. - Jobs other than `transcribed` expose **Resubmit** under the current UI rule. The service blocks resubmission while processing is active or when no failed Sources exist. -- All Jobs expose **Delete Job**, subject to delete guardrails. +- All Jobs expose **Delete Job**, subject to explicit evidence-deletion guardrails. - Invalid and missing IDs produce explicit states. ## Cancel Behavior @@ -67,7 +67,10 @@ Jobs manages transcription processing runs. A Job belongs to one Document, links ## Delete Behavior - Deletion is blocked while status is `processing`. -- Allowed deletion warns that related JobSource links are removed. +- Allowed deletion explicitly warns that related `JobSource` projections, + immutable execution attempts, captured transport responses, and attempt-owned + artifacts are permanently removed. +- Source records and source files remain available for separate deletion. - Success returns to the Jobs list. ## Acceptance Checklist diff --git a/src/transcription/db/runtime.py b/src/transcription/db/runtime.py index 7b1356c..3557e52 100644 --- a/src/transcription/db/runtime.py +++ b/src/transcription/db/runtime.py @@ -43,12 +43,18 @@ async def dispose_database_runtime() -> None: def initialize_database_runtime(*, settings: Settings | None = None) -> DatabaseRuntime: """Initialize lifespan-owned async DB resources once per process.""" global _runtime - runtime = _runtime - if runtime is not None: - return runtime - active_settings = settings or get_settings() database_url = get_database_url(active_settings) + runtime = _runtime + if runtime is not None: + runtime_url = runtime.engine.url.render_as_string(hide_password=False) + if runtime_url != database_url: + raise RuntimeError( + "Database runtime is already initialized for a different database: " + f"{runtime_url!r} != {database_url!r}" + ) + return runtime + engine = get_engine(database_url) session_factory = get_session_factory(database_url) runtime = DatabaseRuntime(engine=engine, session_factory=session_factory) diff --git a/src/transcription/providers/openrouter.py b/src/transcription/providers/openrouter.py index 6250159..a07f6d2 100644 --- a/src/transcription/providers/openrouter.py +++ b/src/transcription/providers/openrouter.py @@ -4,7 +4,10 @@ from __future__ import annotations import base64 import hashlib +import json import logging +from collections.abc import AsyncIterator +from collections.abc import Callable from typing import Annotated from typing import Any from typing import Literal @@ -36,7 +39,25 @@ from transcription.providers.evidence import filter_safe_response_headers logger = logging.getLogger(__name__) DEFAULT_OPENROUTER_MODEL = "google/gemini-2.5-flash" -OPENROUTER_ADAPTER_VERSION = "1" +OPENROUTER_ADAPTER_VERSION = "2" + + +class _CapturingAsyncByteStream(httpx.AsyncByteStream): + """Copy streamed response bytes without changing what the SDK consumes.""" + + def __init__(self, stream: httpx.AsyncByteStream, on_complete: Callable[[bytes], None]): + self._stream = stream + self._on_complete = on_complete + + async def __aiter__(self) -> AsyncIterator[bytes]: + content = bytearray() + async for chunk in self._stream: + content.extend(chunk) + yield chunk + self._on_complete(bytes(content)) + + async def aclose(self) -> None: + await self._stream.aclose() class _CapturingAsyncClient: @@ -45,10 +66,15 @@ class _CapturingAsyncClient: def __init__(self, client: httpx.AsyncClient): self._client = client self.last_response: httpx.Response | None = None + self.last_body: bytes | None = None async def send(self, request: httpx.Request, **kwargs: Any) -> httpx.Response: response = await self._client.send(request, **kwargs) self.last_response = response + try: + self.last_body = response.content + except httpx.ResponseNotRead: + response.stream = _CapturingAsyncByteStream(response.stream, self._capture_body) return response def build_request(self, *args: Any, **kwargs: Any) -> httpx.Request: @@ -59,6 +85,10 @@ class _CapturingAsyncClient: def reset(self) -> None: self.last_response = None + self.last_body = None + + def _capture_body(self, body: bytes) -> None: + self.last_body = body class _ProviderModel(BaseModel): @@ -245,7 +275,7 @@ class OpenRouterTranscriptionProvider: else "connection" ) raise ProviderError( - "OpenRouter request failed", + self._transport_error_message(transport), request_manifest=manifest, transport_evidence=transport, failure_phase=failure_phase, @@ -354,10 +384,7 @@ class OpenRouterTranscriptionProvider: if response is None: return TransportEvidence(response_received=False) headers = filter_safe_response_headers(response.headers) - try: - body = response.content - except httpx.ResponseNotRead: - body = None + body = self._capturing_client.last_body if self._capturing_client is not None else None return TransportEvidence( response_received=True, status_code=response.status_code, @@ -369,6 +396,25 @@ class OpenRouterTranscriptionProvider: generation_id=headers.get("x-openrouter-generation-id"), ) + @staticmethod + def _transport_error_message(transport: TransportEvidence) -> str: + message = "OpenRouter request failed" + if transport.status_code is not None: + message += f" with HTTP {transport.status_code}" + if transport.body is None: + return message + try: + payload = json.loads(transport.body) + except (UnicodeDecodeError, json.JSONDecodeError): + return message + if not isinstance(payload, dict): + return message + error = payload.get("error") + detail = error.get("message") if isinstance(error, dict) else None + if isinstance(detail, str) and detail.strip(): + return f"{message}: {detail.strip()[:500]}" + return message + def _build_metadata(self, response: OpenRouterResponse) -> TranscriptionMetadata: choice = response.choices[0] finish_reason = choice.finish_reason.strip() if choice.finish_reason and choice.finish_reason.strip() else None diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py index b9db18e..dee7765 100644 --- a/src/transcription/services/jobs.py +++ b/src/transcription/services/jobs.py @@ -1,6 +1,8 @@ +import logging from collections.abc import Sequence from datetime import UTC from datetime import datetime +from pathlib import Path from uuid import UUID from sqlalchemy import func @@ -13,11 +15,14 @@ from ..db.models import Job from ..db.models import JobSource from ..db.models import JobSourceStatus from ..db.models import JobStatus +from ..db.models import ProcessingArtifact from ..db.models import Source from ..errors import AppError from ..errors import ErrorCategory from .base import ServiceBase +logger = logging.getLogger(__name__) + class JobDeleteBlockedError(AppError): """Raised when a job delete operation is blocked by lifecycle policy.""" @@ -253,6 +258,81 @@ class JobService(ServiceBase): await _session.delete(job) await self._finalize(session=_session, caller_session=session) + async def delete_job_and_evidence(self, *, job_id: UUID) -> None: + """Explicitly delete a terminal job and all evidence owned by its attempts.""" + external_references: list[str] = [] + async with self._session_scope() as session: + job = ( + await session.exec( + select(Job) + .options(selectinload(Job.job_sources)) + .where(Job.id == job_id) + .execution_options(populate_existing=True) + ) + ).first() + if job is None: + raise self._not_found(job_id) + if job.status == JobStatus.PROCESSING: + raise JobDeleteBlockedError( + "Job delete blocked while status is processing", + category=ErrorCategory.VALIDATION, + suggestion="Wait for processing to complete, or cancel it before deleting evidence.", + ) + + attempts = list( + ( + await session.exec( + select(ExecutionAttempt).where(ExecutionAttempt.job_id == job_id) + ) + ).all() + ) + if attempts: + attempt_ids = [attempt.id for attempt in attempts] + artifacts = list( + ( + await session.exec( + select(ProcessingArtifact).where( + ProcessingArtifact.execution_attempt_id.in_(attempt_ids) + ) + ) + ).all() + ) + external_references = [ + artifact.external_reference + for artifact in artifacts + if artifact.external_reference is not None + ] + for artifact in artifacts: + await session.delete(artifact) + await session.flush() + for attempt in attempts: + await session.delete(attempt) + await session.flush() + + for job_source in list(job.job_sources): + await session.delete(job_source) + await session.flush() + await session.delete(job) + await self._finalize(session=session, caller_session=None) + + for external_reference in external_references: + self._delete_external_artifact(external_reference) + + def _delete_external_artifact(self, external_reference: str) -> None: + relative_path = Path(external_reference) + if relative_path.is_absolute() or ".." in relative_path.parts: + logger.warning("Skipped unsafe external artifact reference during job deletion: %s", external_reference) + return + artifact_root = self.settings.artifact_dir.resolve() + artifact_path = (artifact_root / relative_path).resolve() + if artifact_root not in artifact_path.parents: + logger.warning("Skipped external artifact outside configured root: %s", external_reference) + return + try: + artifact_path.unlink(missing_ok=True) + except OSError: + logger.warning("Failed to delete external artifact: %s", artifact_path) + async def cancel_job(self, *, job_id: UUID, session: AsyncSession | None = None) -> Job: """Cancel a queued/processing job and stop remaining source work.""" async with self._session_scope(session) as _session: diff --git a/src/transcription/services/sources.py b/src/transcription/services/sources.py index f9f315b..430f2ef 100644 --- a/src/transcription/services/sources.py +++ b/src/transcription/services/sources.py @@ -1155,7 +1155,7 @@ def handle_transcription_errors(): ) from exc except ProviderError as exc: raise TranscriptionError( - "Provider transcription failed", + f"Provider transcription failed: {exc}", category=ErrorCategory.EXTERNAL_PROVIDER, suggestion="Retry the transcription from jobs. If repeated, check provider availability.", retriable=True, diff --git a/src/transcription/ui/pages/jobs_page.py b/src/transcription/ui/pages/jobs_page.py index 535f2c1..2a89770 100644 --- a/src/transcription/ui/pages/jobs_page.py +++ b/src/transcription/ui/pages/jobs_page.py @@ -330,15 +330,20 @@ def register_page() -> None: # noqa: PLR0915 ) return - ui.label("This action permanently deletes the job.").classes("text-xs ui-text-danger font-medium") + ui.label("This action permanently deletes the job and its immutable execution evidence.").classes( + "text-xs ui-text-danger font-medium" + ) if job.job_sources: - ui.label("Related JobSource links will be removed as part of delete.").classes( + ui.label( + "Related JobSource links, execution attempts, transport responses, and attempt artifacts " + "will be removed. Source records and files remain until deleted separately." + ).classes( "text-xs ui-text-muted" ) async def submit_delete() -> None: try: - await jobs_service.delete_job_with_guardrails(job_id=job.id) + await jobs_service.delete_job_and_evidence(job_id=job.id) except JobDeleteBlockedError as exc: ui.notify(exc.message, type="warning") return @@ -355,7 +360,7 @@ def register_page() -> None: # noqa: PLR0915 with ui.row().classes("w-full items-center gap-2 mt-2"): destructive_button( - "Delete job permanently", on_click=submit_delete, icon="delete_forever", variant="solid" + "Delete job and evidence", on_click=submit_delete, icon="delete_forever", variant="solid" ) ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}"), icon="arrow_back").props("flat") diff --git a/tests/conftest.py b/tests/conftest.py index 7a47979..649f4f0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,6 +4,8 @@ Every test gets a fresh in-memory SQLite database so tests are isolated, fast, and leave no artifacts on disk. """ +from pathlib import Path + import pytest import pytest_asyncio from sqlmodel import Session @@ -12,7 +14,7 @@ from sqlmodel import create_engine from sqlmodel.pool import StaticPool from transcription.config import Settings -from transcription.config import get_settings +from transcription.config import SqliteSettings from transcription.db.engine import get_database_url from transcription.db.engine import get_engine from transcription.db.session import dispose_session_factory @@ -37,10 +39,17 @@ def session(): @pytest_asyncio.fixture -async def default_settings(): +async def default_settings(tmp_path): """Provide default settings for tests.""" - settings = get_settings(database_url="sqlite:///:memory:") + database_path = tmp_path / "tests.db" + settings = Settings( + openrouter_api_key="test-key", + database=SqliteSettings(path=str(database_path)), + environment="test", + ) db_url = get_database_url(settings) + if Path(str(get_engine(database_url=db_url).url.database)).resolve() != database_path.resolve(): + raise RuntimeError(f"Refusing to initialize destructive test fixtures against {db_url}") await dispose_session_factory(db_url) engine = get_engine(database_url=db_url) diff --git a/tests/services/test_job_service.py b/tests/services/test_job_service.py index 0855257..966cfa2 100644 --- a/tests/services/test_job_service.py +++ b/tests/services/test_job_service.py @@ -17,6 +17,7 @@ from transcription.services.jobs import JobDeleteBlockedError from transcription.services.jobs import JobNotFoundError from transcription.services.jobs import JobResubmitBlockedError from transcription.services.jobs import JobService +from transcription.services.sources import SourceService class TestJobService: @@ -232,6 +233,41 @@ class TestJobService: with pytest.raises(JobNotFoundError): await job_service.read_job(job_id=job.id) + @pytest.mark.asyncio + async def test_delete_job_and_evidence_removes_attempts_but_preserves_source( + self, + job_service: JobService, + document_service: DocumentService, + ): + source_service = SourceService(session_factory=job_service.session_factory) + document = await document_service.create_document(Document(name="evidence-delete-doc")) + job = await job_service.create_job(Job(document_id=document.id, status=JobStatus.FAILED)) + source = await source_service.create_source( + Source( + document_id=document.id, + page_number=1, + upload_name="evidence.jpg", + filename="evidence.jpg", + file_path="/uploads/evidence.jpg", + file_hash="d" * 64, + file_size_bytes=1, + ) + ) + await source_service.create_job_source(JobSource(job_id=job.id, source_id=source.id)) + await source_service.update_job_source_transcription( + job_id=job.id, + source_id=source.id, + text=None, + error_detail="fixture failure", + ) + + await job_service.delete_job_and_evidence(job_id=job.id) + + with pytest.raises(JobNotFoundError): + await job_service.read_job(job_id=job.id) + assert await source_service.list_execution_attempts(job_id=job.id) == [] + assert (await source_service.read_source(source.id)).id == source.id + @pytest.mark.asyncio async def test_cancel_job_marks_non_transcribed_sources_failed( self, diff --git a/tests/test_db.py b/tests/test_db.py index baf4998..7186fd9 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -62,6 +62,28 @@ async def test_get_session_yields_async_session(tmp_path): await dispose_database_runtime() +@pytest.mark.asyncio +async def test_runtime_rejects_reinitialization_for_different_database(tmp_path): + """An existing process runtime cannot silently switch database targets.""" + first_settings = Settings( + openrouter_api_key="test-key", + database=SqliteSettings(path=str(tmp_path / "first.db")), + environment="test", + ) + second_settings = Settings( + openrouter_api_key="test-key", + database=SqliteSettings(path=str(tmp_path / "second.db")), + environment="test", + ) + initialize_database_runtime(settings=first_settings) + + try: + with pytest.raises(RuntimeError, match="already initialized for a different database"): + initialize_database_runtime(settings=second_settings) + finally: + await dispose_database_runtime() + + @pytest.mark.asyncio async def test_create_all_seeds_default_registry_rows(tmp_path): settings = Settings( diff --git a/tests/test_v42_evidence.py b/tests/test_v42_evidence.py index 6eedd10..e594156 100644 --- a/tests/test_v42_evidence.py +++ b/tests/test_v42_evidence.py @@ -33,6 +33,18 @@ from transcription.services.sources import TranscriptionError from transcription.services.sources import transcribe_document_image +class _ChunkedAsyncStream(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]): + self._chunks = chunks + + async def __aiter__(self): + for chunk in self._chunks: + yield chunk + + async def aclose(self): + return + + @pytest.mark.asyncio async def test_openrouter_captures_exact_transport_and_secret_safe_manifest(): response_body = ( @@ -87,6 +99,39 @@ async def test_openrouter_captures_exact_transport_and_secret_safe_manifest(): assert result.request_manifest.omitted_optional_parameters == ("temperature", "top_p") +@pytest.mark.asyncio +async def test_openrouter_captures_body_consumed_as_sdk_stream(): + response_body = ( + b'{"id":"gen-2","created":1,"model":"vendor/model","object":"chat.completion",' + b'"system_fingerprint":null,"choices":[{"index":0,"finish_reason":"stop",' + b'"message":{"role":"assistant","content":"Transcript"}}],' + b'"unknown_streamed_field":true}' + ) + + async def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + stream=_ChunkedAsyncStream([response_body[:23], response_body[23:61], response_body[61:]]), + headers={"Content-Type": "application/json"}, + request=request, + ) + + provider = OpenRouterTranscriptionProvider( + settings=Settings(openrouter_api_key="test-key"), + async_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) + result = await provider.transcribe( + prompt_text="Literal prompt", + image_bytes=b"source-bytes", + mime_type="image/png", + ) + + assert result.transport_evidence is not None + assert result.transport_evidence.body == response_body + assert b'"unknown_streamed_field":true' in result.transport_evidence.body + await provider.aclose() + + @pytest.mark.asyncio async def test_openrouter_failure_retains_safe_response_evidence(): async def handler(request: httpx.Request) -> httpx.Response: @@ -113,6 +158,7 @@ async def test_openrouter_failure_retains_safe_response_evidence(): assert evidence.status_code == 500 assert evidence.body == b'{"error":{"message":"provider unavailable"}}' assert evidence.safe_headers == {"content-type": "application/json", "retry-after": "2"} + assert str(failure.value) == "OpenRouter request failed with HTTP 500: provider unavailable" @pytest.mark.asyncio diff --git a/tests/ui/conftest.py b/tests/ui/conftest.py index 10566d2..57a34c9 100644 --- a/tests/ui/conftest.py +++ b/tests/ui/conftest.py @@ -47,6 +47,13 @@ def app_client(tmp_path_factory: pytest.TempPathFactory) -> AsyncGenerator[tuple app = create_app(settings=settings) with TestClient(app) as client: + runtime_path = Path(str(app.state.runtime.engine.url.database)).resolve() + expected_path = Path(settings.database.path).resolve() + if runtime_path != expected_path: + raise RuntimeError( + "Refusing to initialize destructive UI fixtures against " + f"{runtime_path}; expected {expected_path}" + ) yield app, client diff --git a/tests/ui/test_jobs_page.py b/tests/ui/test_jobs_page.py index be71ea7..0743002 100644 --- a/tests/ui/test_jobs_page.py +++ b/tests/ui/test_jobs_page.py @@ -164,5 +164,6 @@ class TestJobsPageRendering: assert response.status_code == 200 assert "Delete Processing Job" in response.text - assert "Delete job permanently" in response.text + assert "Delete job and evidence" in response.text + assert "immutable execution evidence" in response.text assert "Delete is blocked" not in response.text