V3 step 1 update models.py and step 2 implement service/worker, and raw API response persistence

src/transcription/config.py

@@ -69,6 +69,9 @@ class Settings(BaseSettings):
     provider_model: str | None = None
     openrouter_http_referer: str | None = None
     openrouter_app_title: str | None = None
+    default_prompt_name: str = "transcribe_document.md"
+    transcription_temperature: float | None = None
+    transcription_top_p: float | None = None
 
     # --- runtime environment ---
     environment: Literal["development", "test", "production"] = "development"

src/transcription/db/models.py

@@ -1,4 +1,4 @@
-"""SQLModel domain models for the V2 transcription system."""
+"""SQLModel domain models for the V3 transcription system."""
 
 from datetime import UTC
 from datetime import date
@@ -10,6 +10,7 @@ from uuid import UUID
 from uuid import uuid4
 
 from sqlalchemy import Column
+from sqlalchemy import BigInteger
 from sqlalchemy import JSON
 from sqlalchemy import UniqueConstraint
 from sqlalchemy.dialects.postgresql import JSONB
@@ -125,7 +126,6 @@ class Job(SQLModel, table=True):
     date_updated: datetime = Field(default_factory=lambda: datetime.now(UTC))
     provider: str | None = None
     model: str | None = None
-    prompt_name: str | None = None
 
     document: Optional["Document"] = Relationship(back_populates="jobs", sa_relationship_kwargs={"lazy": "selectin"})
     job_sources: list["JobSource"] = Relationship(back_populates="job", sa_relationship_kwargs={"lazy": "selectin"})
@@ -173,6 +173,8 @@ class Source(SQLModel, table=True):
     upload_name: str
     filename: str
     file_path: str
+    file_hash: str
+    file_size_bytes: int = Field(sa_column=Column(BigInteger(), nullable=False))
     raw_transcription: str | None = None
     revised_text: str | None = None
     date_uploaded: datetime = Field(default_factory=lambda: datetime.now(UTC))
@@ -221,6 +223,12 @@ class JobSource(SQLModel, table=True):
     source_id: UUID = Field(foreign_key="source.id")
     status: JobSourceStatus = Field(default=JobSourceStatus.PENDING)
     raw_transcription: str | None = None
+    prompt_name: str | None = None
+    prompt_hash: str | None = None
+    system_prompt: str | None = None
+    user_prompt: str | None = None
+    temperature: float | None = None
+    top_p: float | None = None
     ai_metadata: dict[str, Any] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True))
     raw_api_response: dict[str, Any] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True))
     error_detail: str | None = None

src/transcription/providers/base.py

@@ -25,6 +25,11 @@ class TranscriptionResult:
     provider: str
     prompt_name: str
     model: str
+    prompt_hash: str | None = None
+    system_prompt: str | None = None
+    user_prompt: str | None = None
+    temperature: float | None = None
+    top_p: float | None = None
     finish_reason: str | None = None
     usage_input_tokens: int | None = None
     usage_output_tokens: int | None = None
@@ -36,6 +41,14 @@ class TranscriptionResult:
 class TranscriptionProvider(Protocol):
     """Contract every transcription provider adapter must satisfy."""
 
-    async def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
+    async def transcribe(
+        self,
+        *,
+        prompt_text: str,
+        image_bytes: bytes,
+        mime_type: str,
+        temperature: float | None = None,
+        top_p: float | None = None,
+    ) -> TranscriptionResult:
         """Transcribe the provided image according to the prompt text."""
         ...

src/transcription/providers/openrouter.py

@@ -31,6 +31,8 @@ class OpenRouterRequest:
     messages: list[dict[str, Any]]
     http_referer: str | None
     x_open_router_title: str | None
+    temperature: float | None
+    top_p: float | None
 
 
 class OpenRouterTranscriptionProvider:
@@ -46,15 +48,31 @@ class OpenRouterTranscriptionProvider:
         """Return the resolved OpenRouter model slug."""
         return self._model
 
-    async def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
+    async def transcribe(
+        self,
+        *,
+        prompt_text: str,
+        image_bytes: bytes,
+        mime_type: str,
+        temperature: float | None = None,
+        top_p: float | None = None,
+    ) -> TranscriptionResult:
         """Send prompt + image to OpenRouter and return normalized text output."""
-        request = self._build_request(prompt_text=prompt_text, image_bytes=image_bytes, mime_type=mime_type)
+        request = self._build_request(
+            prompt_text=prompt_text,
+            image_bytes=image_bytes,
+            mime_type=mime_type,
+            temperature=temperature,
+            top_p=top_p,
+        )
         try:
             response = await self._client.chat.send_async(
                 messages=cast(list[ChatMessagesTypedDict], request.messages),
                 model=request.model,
                 http_referer=request.http_referer,
                 x_open_router_title=request.x_open_router_title,
+                temperature=request.temperature,
+                top_p=request.top_p,
             )
         except Exception as exc:
             message = str(exc).lower()
@@ -78,6 +96,11 @@ class OpenRouterTranscriptionProvider:
             text=text,
             provider="openrouter",
             prompt_name="",
+            prompt_hash=None,
+            system_prompt=None,
+            user_prompt=prompt_text,
+            temperature=temperature,
+            top_p=top_p,
             model=model,
             finish_reason=finish_reason,
             usage_input_tokens=usage_input_tokens,
@@ -148,7 +171,15 @@ class OpenRouterTranscriptionProvider:
 
         return repr(value)
 
-    def _build_request(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> OpenRouterRequest:
+    def _build_request(
+        self,
+        *,
+        prompt_text: str,
+        image_bytes: bytes,
+        mime_type: str,
+        temperature: float | None,
+        top_p: float | None,
+    ) -> OpenRouterRequest:
         image_b64 = base64.b64encode(image_bytes).decode("ascii")
         data_url = f"data:{mime_type};base64,{image_b64}"
 
@@ -167,6 +198,8 @@ class OpenRouterTranscriptionProvider:
             messages=messages,
             http_referer=self._settings.openrouter_http_referer,
             x_open_router_title=self._settings.openrouter_app_title,
+            temperature=temperature,
+            top_p=top_p,
         )
 
     def _extract_text(self, response: Any) -> str:

src/transcription/services/store.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Sequence
 from dataclasses import dataclass
+import hashlib
 import logging
 from pathlib import Path
 from uuid import UUID
@@ -48,6 +49,8 @@ class PendingStoredUpload:
     source_id: UUID
     original_filename: str
     stored_path: Path
+    file_hash: str
+    file_size_bytes: int
 
 
 async def create_upload_job(
@@ -68,6 +71,7 @@ async def create_upload_job(
         relative_directory=Path("documents") / str(document_id),
         filename_stem=str(source_id),
     )
+    file_hash, file_size_bytes = _compute_file_metadata(file_bytes)
     try:
         document, job = await _create_upload_records(
             session=session,
@@ -75,6 +79,8 @@ async def create_upload_job(
             source_id=source_id,
             original_filename=filename,
             stored_path=stored_path,
+            file_hash=file_hash,
+            file_size_bytes=file_size_bytes,
         )
     except Exception as exc:
         _best_effort_delete(stored_path)
@@ -101,7 +107,6 @@ async def create_job_for_document(
     session: AsyncSession,
     provider: str | None = None,
     model: str | None = None,
-    prompt_name: str | None = None,
     settings: Settings | None = None,
 ) -> JobCreateResult:
     """Create a queued job for an existing document with one or more uploaded sources."""
@@ -128,6 +133,8 @@ async def create_job_for_document(
                     relative_directory=Path("documents") / str(document_id),
                     filename_stem=str(source_id),
                 ),
+                file_hash=_compute_file_hash(file_bytes),
+                file_size_bytes=len(file_bytes),
             )
         )
 
@@ -138,7 +145,6 @@ async def create_job_for_document(
             stored_uploads=stored_uploads,
             provider=provider,
             model=model,
-            prompt_name=prompt_name,
         )
     except Exception as exc:
         for upload in stored_uploads:
@@ -165,6 +171,8 @@ async def _create_upload_records(
     source_id: UUID,
     original_filename: str,
     stored_path: Path,
+    file_hash: str,
+    file_size_bytes: int,
 ) -> tuple[Document, Job]:
     document = Document(
         id=document_id,
@@ -184,6 +192,8 @@ async def _create_upload_records(
         upload_name=Path(original_filename).name,
         filename=stored_path.name,
         file_path=str(stored_path),
+        file_hash=file_hash,
+        file_size_bytes=file_size_bytes,
     )
     session.add(source)
     await session.flush()
@@ -209,7 +219,6 @@ async def _create_job_for_document_records(
     stored_uploads: Sequence[PendingStoredUpload],
     provider: str | None,
     model: str | None,
-    prompt_name: str | None,
 ) -> tuple[Job, list[UUID]]:
     document = await session.get(Document, document_id)
     if document is None:
@@ -228,7 +237,6 @@ async def _create_job_for_document_records(
         document_id=document_id,
         provider=(provider or None),
         model=(model or None),
-        prompt_name=(prompt_name or None),
     )
     session.add(job)
     await session.flush()
@@ -242,6 +250,8 @@ async def _create_job_for_document_records(
             upload_name=Path(upload.original_filename).name,
             filename=upload.stored_path.name,
             file_path=str(upload.stored_path),
+            file_hash=upload.file_hash,
+            file_size_bytes=upload.file_size_bytes,
         )
         session.add(source)
         await session.flush()
@@ -268,6 +278,14 @@ def _best_effort_delete(path: Path) -> None:
         logger.warning("Failed to clean up upload file after DB error: %s", path)
 
 
+def _compute_file_hash(file_bytes: bytes) -> str:
+    return hashlib.sha256(file_bytes).hexdigest()
+
+
+def _compute_file_metadata(file_bytes: bytes) -> tuple[str, int]:
+    return _compute_file_hash(file_bytes), len(file_bytes)
+
+
 def store_file(
     *,
     filename: str,

src/transcription/services/transcription.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
+import hashlib
 import logging
 import mimetypes
 from collections.abc import Sequence
@@ -39,6 +41,18 @@ DEFAULT_PROMPT_FILE = "transcribe_document.md"
 SUPPORTED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"}
 
 
+@dataclass(frozen=True)
+class PromptExecution:
+    """Resolved prompt inputs captured for one page execution."""
+
+    prompt_name: str
+    prompt_hash: str
+    system_prompt: str | None
+    user_prompt: str
+    temperature: float | None
+    top_p: float | None
+
+
 class PromptLoadError(AppError):
     """Raised when prompt artifacts cannot be loaded safely."""
 
@@ -354,6 +368,11 @@ class TranscriptionService(ServiceBase):
             provider=result.provider,
             model=result.model,
             prompt_name=result.prompt_name,
+            prompt_hash=result.prompt_hash,
+            system_prompt=result.system_prompt,
+            user_prompt=result.user_prompt,
+            temperature=result.temperature,
+            top_p=result.top_p,
             session=session,
         )
 
@@ -366,6 +385,11 @@ class TranscriptionService(ServiceBase):
         provider: str | None = None,
         model: str | None = None,
         prompt_name: str = DEFAULT_PROMPT_FILE,
+        prompt_hash: str | None = None,
+        system_prompt: str | None = None,
+        user_prompt: str | None = None,
+        temperature: float | None = None,
+        top_p: float | None = None,
         session: AsyncSession | None = None,
     ) -> Job:
         """Persist transcription output for the first ordered source in a job's document.
@@ -384,7 +408,6 @@ class TranscriptionService(ServiceBase):
 
             job.provider = provider or job.provider or self.settings.provider.value
             job.model = model or job.model or _resolve_transcript_model(provider=self.provider, settings=self.settings)
-            job.prompt_name = prompt_name or job.prompt_name or DEFAULT_PROMPT_FILE
             job.date_updated = datetime.now(UTC)
 
             source = await _session.exec(
@@ -402,6 +425,11 @@ class TranscriptionService(ServiceBase):
                     provider=provider,
                     model=model,
                     prompt_name=prompt_name,
+                    prompt_hash=prompt_hash,
+                    system_prompt=system_prompt,
+                    user_prompt=user_prompt,
+                    temperature=temperature,
+                    top_p=top_p,
                     session=_session,
                 )
 
@@ -420,6 +448,11 @@ class TranscriptionService(ServiceBase):
         provider: str | None = None,
         model: str | None = None,
         prompt_name: str = DEFAULT_PROMPT_FILE,
+        prompt_hash: str | None = None,
+        system_prompt: str | None = None,
+        user_prompt: str | None = None,
+        temperature: float | None = None,
+        top_p: float | None = None,
         session: AsyncSession | None = None,
     ) -> JobSource:
         """Persist transcription fields for one source within a specific job."""
@@ -449,9 +482,9 @@ class TranscriptionService(ServiceBase):
 
             job.provider = provider or job.provider or self.settings.provider.value
             job.model = model or job.model or _resolve_transcript_model(provider=self.provider, settings=self.settings)
-            job.prompt_name = prompt_name or job.prompt_name or DEFAULT_PROMPT_FILE
             job.date_updated = datetime.now(UTC)
 
+            if text is not None:
                 source.raw_transcription = text
 
             existing_job_source = await _session.exec(
@@ -464,6 +497,12 @@ class TranscriptionService(ServiceBase):
                     source_id=source_id,
                     status=JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED,
                     raw_transcription=text,
+                    prompt_name=prompt_name,
+                    prompt_hash=prompt_hash,
+                    system_prompt=system_prompt,
+                    user_prompt=user_prompt,
+                    temperature=temperature,
+                    top_p=top_p,
                     ai_metadata=ai_metadata,
                     raw_api_response=raw_api_response,
                     error_detail=error_detail,
@@ -471,6 +510,12 @@ class TranscriptionService(ServiceBase):
                 _session.add(job_source)
             else:
                 job_source.raw_transcription = text
+                job_source.prompt_name = prompt_name
+                job_source.prompt_hash = prompt_hash
+                job_source.system_prompt = system_prompt
+                job_source.user_prompt = user_prompt
+                job_source.temperature = temperature
+                job_source.top_p = top_p
                 job_source.ai_metadata = ai_metadata
                 job_source.raw_api_response = raw_api_response
                 job_source.error_detail = error_detail
@@ -545,13 +590,13 @@ def _resolve_transcript_model(*, provider: TranscriptionProvider, settings: Sett
 async def transcribe_document_image(
     image_path: str | Path,
     *,
-    prompt_name: str = DEFAULT_PROMPT_FILE,
+    prompt_name: str | None = None,
     settings: Settings | None = None,
     provider: TranscriptionProvider | None = None,
 ) -> TranscriptionResult:
     """Transcribe a local image using the configured prompt and provider."""
     runtime_settings = settings or get_settings()
-    prompt_text = load_prompt_text(prompt_name=prompt_name, settings=runtime_settings)
+    prompt_execution = build_prompt_execution(prompt_name=prompt_name, settings=runtime_settings)
     image_bytes, mime_type = load_image_payload(image_path)
 
     adapter = provider or get_transcription_provider(settings=runtime_settings)
@@ -559,12 +604,45 @@ async def transcribe_document_image(
 
     with handle_transcription_errors():
         result = await adapter.transcribe(
-            prompt_text=prompt_text,
+            prompt_text=prompt_execution.user_prompt,
             image_bytes=image_bytes,
             mime_type=mime_type,
+            temperature=prompt_execution.temperature,
+            top_p=prompt_execution.top_p,
         )
     logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider)
-    return result
+    return TranscriptionResult(
+        text=result.text,
+        provider=result.provider,
+        prompt_name=prompt_execution.prompt_name,
+        prompt_hash=prompt_execution.prompt_hash,
+        system_prompt=prompt_execution.system_prompt,
+        user_prompt=result.user_prompt or prompt_execution.user_prompt,
+        temperature=result.temperature if result.temperature is not None else prompt_execution.temperature,
+        top_p=result.top_p if result.top_p is not None else prompt_execution.top_p,
+        model=result.model,
+        finish_reason=result.finish_reason,
+        usage_input_tokens=result.usage_input_tokens,
+        usage_output_tokens=result.usage_output_tokens,
+        usage_total_tokens=result.usage_total_tokens,
+        ai_metadata=result.ai_metadata,
+        raw_api_response=result.raw_api_response,
+    )
+
+
+def build_prompt_execution(*, prompt_name: str | None = None, settings: Settings | None = None) -> PromptExecution:
+    """Resolve the exact prompt payload and provenance for one execution."""
+    runtime_settings = settings or get_settings()
+    effective_prompt_name = (prompt_name or runtime_settings.default_prompt_name or DEFAULT_PROMPT_FILE).strip()
+    user_prompt = load_prompt_text(prompt_name=effective_prompt_name, settings=runtime_settings)
+    return PromptExecution(
+        prompt_name=effective_prompt_name,
+        prompt_hash=hashlib.sha256(user_prompt.encode("utf-8")).hexdigest(),
+        system_prompt=None,
+        user_prompt=user_prompt,
+        temperature=runtime_settings.transcription_temperature,
+        top_p=runtime_settings.transcription_top_p,
+    )
 
 
 def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settings | None = None) -> str:

src/transcription/services/workflows.py

@@ -16,6 +16,7 @@ from ..errors import format_error_detail
 from ..providers import TranscriptionResult
 from . import ServiceBundle
 from .transcription import DEFAULT_PROMPT_FILE
+from .transcription import build_prompt_execution
 from .transcription import transcribe_document_image
 
 logger = logging.getLogger(__name__)
@@ -84,11 +85,13 @@ async def process_queued_job(
     if not sources:
         return await services.jobs.mark_job_status(job.id, JobStatus.TRANSCRIBED, session=session)
 
-    successful_pages: list[tuple[Source, TranscriptionResult]] = []
+    successful_pages: list[tuple[Source, TranscriptionResult, str, str | None, str, float | None, float | None]] = []
-    failed_pages: list[tuple[Source, AppError]] = []
+    failed_pages: list[tuple[Source, AppError, str, str, str | None, str, float | None, float | None]] = []
     externally_stopped = False
 
     for source in sources:
+        prompt_execution = build_prompt_execution(settings=runtime_settings)
+
         if await _job_no_longer_processing(job_id=job.id, services=services, session=session):
             externally_stopped = True
             break
@@ -96,7 +99,11 @@ async def process_queued_job(
         started_at = asyncio.get_running_loop().time()
         try:
             result = await asyncio.wait_for(
-                transcribe_document_image(source.file_path),
+                transcribe_document_image(
+                    source.file_path,
+                    settings=runtime_settings,
+                    provider=services.transcriptions.provider,
+                ),
                 timeout=runtime_settings.worker_provider_timeout_seconds,
             )
             elapsed_seconds = asyncio.get_running_loop().time() - started_at
@@ -120,7 +127,17 @@ async def process_queued_job(
             )
 
             _validate_transcription_quality(result=result, settings=runtime_settings)
-            successful_pages.append((source, result))
+            successful_pages.append(
+                (
+                    source,
+                    result,
+                    prompt_execution.prompt_hash,
+                    prompt_execution.system_prompt,
+                    prompt_execution.user_prompt,
+                    prompt_execution.temperature,
+                    prompt_execution.top_p,
+                )
+            )
         except TimeoutError:
             error = AppError(
                 f"Provider call timed out after {runtime_settings.worker_provider_timeout_seconds:.1f}s",
@@ -128,7 +145,18 @@ async def process_queued_job(
                 suggestion="Retry the job. If this repeats, verify provider latency and request payload size.",
                 retriable=True,
             )
-            failed_pages.append((source, error))
+            failed_pages.append(
+                (
+                    source,
+                    error,
+                    prompt_execution.prompt_name,
+                    prompt_execution.prompt_hash,
+                    prompt_execution.system_prompt,
+                    prompt_execution.user_prompt,
+                    prompt_execution.temperature,
+                    prompt_execution.top_p,
+                )
+            )
             logger.error(
                 "Source failed operation=worker.process_job job_id=%s document_id=%s source_id=%s error_id=%s category=%s",
                 job.id,
@@ -144,7 +172,18 @@ async def process_queued_job(
                 case _:
                     error = classify_unexpected_error(exc, operation="worker.process_job")
 
-            failed_pages.append((source, error))
+            failed_pages.append(
+                (
+                    source,
+                    error,
+                    prompt_execution.prompt_name,
+                    prompt_execution.prompt_hash,
+                    prompt_execution.system_prompt,
+                    prompt_execution.user_prompt,
+                    prompt_execution.temperature,
+                    prompt_execution.top_p,
+                )
+            )
             logger.error(
                 "Source failed operation=worker.process_job job_id=%s document_id=%s source_id=%s error_id=%s category=%s",
                 job.id,
@@ -368,15 +407,15 @@ async def _finalize_batch_outcome(
     *,
     job: Job,
     services: ServiceBundle,
-    successful_pages: list[tuple[Source, TranscriptionResult]],
+    successful_pages: list[tuple[Source, TranscriptionResult, str, str | None, str, float | None, float | None]],
-    failed_pages: list[tuple[Source, AppError]],
+    failed_pages: list[tuple[Source, AppError, str, str, str | None, str, float | None, float | None]],
     status: JobStatus,
     session: AsyncSession | None = None,
 ) -> Job:
     """Transaction B: write per-source outcomes and terminal job status atomically."""
     if session is None:
         async with services.jobs._session_scope() as local_session:
-            for source, result in successful_pages:
+            for source, result, prompt_hash, system_prompt, user_prompt, temperature, top_p in successful_pages:
                 await services.transcriptions.update_job_source_transcription(
                     job_id=job.id,
                     source_id=source.id,
@@ -387,16 +426,26 @@ async def _finalize_batch_outcome(
                     provider=result.provider,
                     model=result.model,
                     prompt_name=result.prompt_name,
+                    prompt_hash=result.prompt_hash or prompt_hash,
+                    system_prompt=result.system_prompt if result.system_prompt is not None else system_prompt,
+                    user_prompt=result.user_prompt if result.user_prompt is not None else user_prompt,
+                    temperature=result.temperature if result.temperature is not None else temperature,
+                    top_p=result.top_p if result.top_p is not None else top_p,
                     session=local_session,
                 )
 
-            for source, error in failed_pages:
+            for source, error, prompt_name, prompt_hash, system_prompt, user_prompt, temperature, top_p in failed_pages:
                 await services.transcriptions.update_job_source_transcription(
                     job_id=job.id,
                     source_id=source.id,
                     text=None,
                     error_detail=format_error_detail(error),
-                    prompt_name=DEFAULT_PROMPT_FILE,
+                    prompt_name=prompt_name,
+                    prompt_hash=prompt_hash,
+                    system_prompt=system_prompt,
+                    user_prompt=user_prompt,
+                    temperature=temperature,
+                    top_p=top_p,
                     session=local_session,
                 )
 
@@ -404,7 +453,7 @@ async def _finalize_batch_outcome(
             await local_session.commit()
             return updated_job
 
-    for source, result in successful_pages:
+    for source, result, prompt_hash, system_prompt, user_prompt, temperature, top_p in successful_pages:
         await services.transcriptions.update_job_source_transcription(
             job_id=job.id,
             source_id=source.id,
@@ -415,16 +464,26 @@ async def _finalize_batch_outcome(
             provider=result.provider,
             model=result.model,
             prompt_name=result.prompt_name,
+            prompt_hash=result.prompt_hash or prompt_hash,
+            system_prompt=result.system_prompt if result.system_prompt is not None else system_prompt,
+            user_prompt=result.user_prompt if result.user_prompt is not None else user_prompt,
+            temperature=result.temperature if result.temperature is not None else temperature,
+            top_p=result.top_p if result.top_p is not None else top_p,
             session=session,
         )
 
-    for source, error in failed_pages:
+    for source, error, prompt_name, prompt_hash, system_prompt, user_prompt, temperature, top_p in failed_pages:
         await services.transcriptions.update_job_source_transcription(
             job_id=job.id,
             source_id=source.id,
             text=None,
             error_detail=format_error_detail(error),
-            prompt_name=DEFAULT_PROMPT_FILE,
+            prompt_name=prompt_name,
+            prompt_hash=prompt_hash,
+            system_prompt=system_prompt,
+            user_prompt=user_prompt,
+            temperature=temperature,
+            top_p=top_p,
             session=session,
         )
 

src/transcription/ui/components/transcript.py

@@ -29,7 +29,7 @@ def render_original_transcription_card(*, job: Job, classes: str = "w-full") ->
     with card, ui.column().classes("w-full q-gutter-y-sm"):
         ui.label(header).classes("text-subtitle1 text-weight-medium")
         ui.label(caption).classes("text-caption vibe-text-muted")
-        _metadata_row(label="Prompt", value=job.prompt_name or "unknown")
+        _metadata_row(label="Prompt", value=_latest_job_prompt(job) or "unknown")
         _metadata_row(label="Updated", value=_format_created_at(job.date_updated))
 
         latest_transcription = _latest_job_transcription(job)
@@ -114,6 +114,13 @@ def _latest_job_error_detail(job: Job) -> str | None:
     return None
 
 
+def _latest_job_prompt(job: Job) -> str | None:
+    for job_source in sorted(job.job_sources, key=lambda item: item.executed_at, reverse=True):
+        if job_source.prompt_name:
+            return job_source.prompt_name
+    return None
+
+
 def _format_created_at(value: datetime) -> str:
     """Return a compact UTC-like timestamp for row captions."""
     return value.strftime("%Y-%m-%d %H:%M:%S %Z")

src/transcription/ui/pages/jobs_page.py

@@ -91,10 +91,9 @@ def register_page() -> None:  # noqa: PLR0915
                 if requested_document_id in document_options:
                     document_select.value = requested_document_id
 
-                with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-3"):
+                with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"):
                     provider_input = ui.input(label="Provider").props("outlined").classes("ui-form-surface")
                     model_input = ui.input(label="Model").props("outlined").classes("ui-form-surface")
-                    prompt_input = ui.input(label="Prompt").props("outlined").classes("ui-form-surface")
 
             _render_upload_section(uploaded_files)
 
@@ -119,7 +118,6 @@ def register_page() -> None:  # noqa: PLR0915
                             uploads=uploaded_files,
                             provider=(provider_input.value or None),
                             model=(model_input.value or None),
-                            prompt_name=(prompt_input.value or None),
                             session=session,
                         )
                 except Exception as exc:  # noqa: BLE001
@@ -421,7 +419,7 @@ def _render_job_logistics(job: Job) -> None:
     with archival_card(title="Execution Logistics"):
         metadata_row("Provider:", job.provider or "pending")
         metadata_row("Model:", job.model or "pending")
-        metadata_row("Prompt:", job.prompt_name or "pending")
+        metadata_row("Prompt:", _latest_prompt_name(job) or "pending")
         metadata_row("Retry Count:", str(job.retry_count))
         metadata_row("Last Updated:", job.date_updated.isoformat())
 
@@ -449,3 +447,11 @@ def _parse_uuid(value: str | None) -> UUID | None:
         return UUID(value)
     except ValueError:
         return None
+
+
+def _latest_prompt_name(job: Job) -> str | None:
+    ordered = sorted(job.job_sources, key=lambda item: item.executed_at, reverse=True)
+    for job_source in ordered:
+        if job_source.prompt_name:
+            return job_source.prompt_name
+    return None

src/transcription/ui/pages/sources_page.py

@@ -281,7 +281,7 @@ def _render_source_job_metadata_zone(latest_job_source: JobSource | None) -> Non
         )
         metadata_row(
             "Prompt:",
-            latest_job_source.job.prompt_name if latest_job_source.job and latest_job_source.job.prompt_name else "unknown",
+            latest_job_source.prompt_name or "unknown",
         )
 
         if latest_job_source.error_detail:

tests/integration/test_pipeline_flow.py

@@ -46,7 +46,13 @@ class TestPipelineSuccessFlow:
         self, async_session, default_session_factory, tmp_path: Path, monkeypatch
     ):
         """Upload followed by worker processing persists job transcription and transcribed status."""
-        settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
+        settings = Settings(
+            openrouter_api_key="test-key",
+            upload_dir=tmp_path,
+            default_prompt_name="transcribe_document.md",
+            transcription_temperature=0.2,
+            transcription_top_p=0.85,
+        )
         upload_result = await create_upload_job(
             filename="pipeline.jpg",
             file_bytes=b"pipeline-bytes",
@@ -89,13 +95,17 @@ class TestPipelineSuccessFlow:
         queued_job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
         processed = queued_job is not None
         if queued_job is not None:
-            await advance_job(job=queued_job, services=services, session=async_session)
+            await advance_job(job=queued_job, services=services, settings=settings, session=async_session)
         job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
 
         assert processed is True
         assert job is not None
         assert job.status == JobStatus.TRANSCRIBED
         assert any(job_source.raw_transcription == "Pipeline transcript" for job_source in job.job_sources)
+        assert any(job_source.prompt_name == "transcribe_document.md" for job_source in job.job_sources)
+        assert any(job_source.user_prompt is not None for job_source in job.job_sources)
+        assert any(job_source.temperature == 0.2 for job_source in job.job_sources)
+        assert any(job_source.top_p == 0.85 for job_source in job.job_sources)
         assert any(job_source.ai_metadata == {"finish_reason": "stop", "usage": {"total_tokens": 42}} for job_source in job.job_sources)
         assert any(
             job_source.raw_api_response == {"id": "resp_123", "choices": [{"message": {"content": "Pipeline transcript"}}]}
@@ -153,7 +163,7 @@ class TestPipelineSuccessFlow:
         queued_job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
         assert queued_job is not None
 
-        await advance_job(job=queued_job, services=services, session=async_session)
+        await advance_job(job=queued_job, services=services, settings=settings, session=async_session)
         job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
 
         assert job.status == JobStatus.TRANSCRIBED
@@ -216,7 +226,7 @@ class TestPipelineSuccessFlow:
         queued_job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
         assert queued_job is not None
 
-        await advance_job(job=queued_job, services=services, session=async_session)
+        await advance_job(job=queued_job, services=services, settings=settings, session=async_session)
         job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
 
         assert job.status == JobStatus.PARTIAL_SUCCESS
@@ -289,7 +299,7 @@ class TestPipelineSuccessFlow:
 
         queued_job = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
         assert queued_job is not None
-        await advance_job(job=queued_job, services=services, session=async_session)
+        await advance_job(job=queued_job, services=services, settings=settings, session=async_session)
 
         refreshed = await services.jobs.read_job(job_id=create_result.job_id, session=async_session)
         assert call_count == 1
@@ -337,7 +347,7 @@ class TestPipelineFailureFlow:
         queued_job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
         processed = queued_job is not None
         if queued_job is not None:
-            await advance_job(job=queued_job, services=services, session=async_session)
+            await advance_job(job=queued_job, services=services, settings=settings, session=async_session)
         job = await services.jobs.read_job(job_id=upload_result.job_id, session=async_session)
 
         assert processed is True

tests/providers/test_openrouter.py

@@ -73,6 +73,27 @@ class TestOpenRouterProviderTranscribe:
         assert send_call["x_open_router_title"] == "Transcription App"
         assert result.text == "Transcript text"
 
+    @pytest.mark.asyncio
+    async def test_includes_temperature_and_top_p_when_provided(self):
+        """Transcribe passes configured sampling parameters through to OpenRouter."""
+        response = {"model": "vendor/model-a", "choices": [{"message": {"content": "Transcript text"}}]}
+        client = _FakeClient(response=response)
+        provider = OpenRouterTranscriptionProvider(settings=Settings(openrouter_api_key="test-key"), client=client)
+
+        result = await provider.transcribe(
+            prompt_text="Prompt body",
+            image_bytes=b"img-bytes",
+            mime_type="image/png",
+            temperature=0.2,
+            top_p=0.85,
+        )
+
+        send_call = client.chat.calls[0]
+        assert send_call["temperature"] == 0.2
+        assert send_call["top_p"] == 0.85
+        assert result.temperature == 0.2
+        assert result.top_p == 0.85
+
     @pytest.mark.asyncio
     async def test_parses_successful_response_text(self):
         """Transcribe returns normalized text from a valid response payload."""

tests/services/test_document_service.py

@@ -78,6 +78,8 @@ async def test_delete_document_blocks_when_dependencies_exist(default_session_fa
                 upload_name="001_page.png",
                 filename="001_page.png",
                 file_path="uploads/001_page.png",
+                file_hash="a" * 64,
+                file_size_bytes=1,
             )
         )
         session.add(Job(document_id=document.id))

tests/services/test_job_service.py

@@ -80,6 +80,8 @@ class TestJobService:
                 upload_name="letter.jpg",
                 filename="stored-letter.jpg",
                 file_path="/uploads/stored-letter.jpg",
+                file_hash="a" * 64,
+                file_size_bytes=1,
             )
             session.add(source)
             await session.flush()
@@ -120,7 +122,7 @@ class TestJobService:
         assert next_job.id == first.id
 
     @pytest.mark.asyncio
-    async def test_create_job_persists_provider_model_prompt(
+    async def test_create_job_persists_provider_and_model(
         self,
         job_service: JobService,
         document_service: DocumentService,
@@ -132,14 +134,12 @@ class TestJobService:
             document_id=document.id,
             provider="openrouter",
             model="google/gemini-2.5-flash",
-            prompt_name="transcribe_document.md",
         )
         await job_service.create_job(job=job)
 
         fetched = await job_service.read_job(job_id=job.id)
         assert fetched.provider == "openrouter"
         assert fetched.model == "google/gemini-2.5-flash"
-        assert fetched.prompt_name == "transcribe_document.md"
 
     @pytest.mark.asyncio
     async def test_read_job_resolves_filename_from_linked_source(
@@ -160,6 +160,8 @@ class TestJobService:
                 upload_name="page_001.png",
                 filename="stored_page_001.png",
                 file_path="/uploads/stored_page_001.png",
+                file_hash="b" * 64,
+                file_size_bytes=1,
             )
             session.add(source)
             await session.flush()
@@ -210,6 +212,8 @@ class TestJobService:
                 upload_name="delete-job-source.jpg",
                 filename="stored-delete-job-source.jpg",
                 file_path="/uploads/stored-delete-job-source.jpg",
+                file_hash="c" * 64,
+                file_size_bytes=1,
             )
             session.add(source)
             await session.flush()
@@ -246,6 +250,8 @@ class TestJobService:
                 upload_name="cancel-1.jpg",
                 filename="stored-cancel-1.jpg",
                 file_path="/uploads/stored-cancel-1.jpg",
+                file_hash="d" * 64,
+                file_size_bytes=1,
             )
             source_two = Source(
                 document_id=document.id,
@@ -253,6 +259,8 @@ class TestJobService:
                 upload_name="cancel-2.jpg",
                 filename="stored-cancel-2.jpg",
                 file_path="/uploads/stored-cancel-2.jpg",
+                file_hash="e" * 64,
+                file_size_bytes=1,
             )
             session.add(source_one)
             session.add(source_two)
@@ -304,6 +312,8 @@ class TestJobService:
                 upload_name="resubmit-1.jpg",
                 filename="stored-resubmit-1.jpg",
                 file_path="/uploads/stored-resubmit-1.jpg",
+                file_hash="f" * 64,
+                file_size_bytes=1,
                 raw_transcription="existing text",
             )
             source_two = Source(
@@ -312,6 +322,8 @@ class TestJobService:
                 upload_name="resubmit-2.jpg",
                 filename="stored-resubmit-2.jpg",
                 file_path="/uploads/stored-resubmit-2.jpg",
+                file_hash="0" * 64,
+                file_size_bytes=1,
                 raw_transcription="done text",
             )
             session.add(source_one)

tests/services/test_store.py

@@ -48,7 +48,6 @@ async def test_create_job_for_document_sorts_uploads_and_creates_links(async_ses
         ],
         provider="openrouter",
         model="test-model",
-        prompt_name="transcribe_document.md",
         session=async_session,
         settings=settings,
     )
@@ -57,7 +56,6 @@ async def test_create_job_for_document_sorts_uploads_and_creates_links(async_ses
     assert created_job is not None
     assert created_job.provider == "openrouter"
     assert created_job.model == "test-model"
-    assert created_job.prompt_name == "transcribe_document.md"
 
     sources = (
         await async_session.exec(
@@ -71,10 +69,16 @@ async def test_create_job_for_document_sorts_uploads_and_creates_links(async_ses
     assert all("A_page" not in source.filename and "b_page" not in source.filename for source in sources)
     assert all(Path(source.filename).stem == str(source.id) for source in sources)
     assert all(Path(source.file_path).parent == (tmp_path / "documents" / str(document.id)) for source in sources)
+    assert [source.file_hash for source in sources] == [
+        "ca978112ca1bbdcafac231b39a23dc4da786eff8147c4e72b9807785afee48bb",
+        "3e23e8160039594a33894f6564e1b1348bbd7a0088d42c4acb73eeaed59c009d",
+    ]
+    assert [source.file_size_bytes for source in sources] == [1, 1]
 
     job_sources = (await async_session.exec(select(JobSource).where(JobSource.job_id == result.job_id))).all()
     assert len(job_sources) == 2
     assert set(result.source_ids) == {job_source.source_id for job_source in job_sources}
+    assert {job_source.prompt_name for job_source in job_sources} == {None}
 
 
 @pytest.mark.asyncio
@@ -103,6 +107,8 @@ async def test_create_upload_job_stores_source_under_document_id_directory(async
     assert Path(source.filename).stem == str(source.id)
     assert result.stored_path.name == source.filename
     assert Path(source.file_path).parent == expected_parent
+    assert source.file_hash == "2c8648d103e3dd7ad87660da0f126a1443b6d21ac1bd3ec000c5e24e2373a90c"
+    assert source.file_size_bytes == len(b"image-bytes")
 
 
 def test_store_person_portrait_stores_file_under_person_id_directory(tmp_path):

tests/services/test_transcription_service.py

@@ -39,6 +39,8 @@ class TestTranscriptionServiceRevisionUpsert:
             upload_name="source.jpg",
             filename="source.jpg",
             file_path="uploads/source.jpg",
+            file_hash="1" * 64,
+            file_size_bytes=1,
         )
         async with transcriptions._session_scope() as session:
             session.add(source)
@@ -74,6 +76,8 @@ class TestTranscriptionServiceRevisionUpsert:
             upload_name="source.jpg",
             filename="source.jpg",
             file_path="uploads/source.jpg",
+            file_hash="2" * 64,
+            file_size_bytes=1,
         )
         async with transcriptions._session_scope() as session:
             session.add(source)
@@ -115,6 +119,8 @@ class TestTranscriptionServiceRevisionUpsert:
             upload_name="delete.jpg",
             filename="delete.jpg",
             file_path=str(stored_path),
+            file_hash="3" * 64,
+            file_size_bytes=4,
         )
         async with transcriptions._session_scope() as session:
             session.add(source)
@@ -149,6 +155,8 @@ class TestTranscriptionServiceRevisionUpsert:
             upload_name="shared.jpg",
             filename="shared.jpg",
             file_path="uploads/shared.jpg",
+            file_hash="4" * 64,
+            file_size_bytes=1,
         )
         async with transcriptions._session_scope() as session:
             session.add(source)
@@ -180,6 +188,8 @@ class TestTranscriptionServiceRevisionUpsert:
             upload_name="orphan.jpg",
             filename="orphan.jpg",
             file_path=str(stored_path),
+            file_hash="5" * 64,
+            file_size_bytes=4,
         )
         await transcriptions.create_source(source=source)
 
@@ -207,6 +217,8 @@ class TestTranscriptionServiceRevisionUpsert:
             upload_name="linked.jpg",
             filename="linked.jpg",
             file_path="uploads/linked.jpg",
+            file_hash="6" * 64,
+            file_size_bytes=1,
         )
         async with transcriptions._session_scope() as session:
             session.add(source)

tests/services/test_v2_crud.py

@@ -60,6 +60,8 @@ async def test_transcription_service_manages_source_crud(default_session_factory
             upload_name="page-1.jpg",
             filename="page-1.jpg",
             file_path="uploads/page-1.jpg",
+            file_hash="7" * 64,
+            file_size_bytes=1,
         )
     )
 
@@ -101,6 +103,8 @@ async def test_transcription_service_job_source_crud_uses_caller_session(default
             upload_name="job-source.jpg",
             filename="job-source.jpg",
             file_path="uploads/job-source.jpg",
+            file_hash="8" * 64,
+            file_size_bytes=1,
         )
         session.add(source)
         await session.flush()
@@ -165,6 +169,8 @@ async def test_document_delete_is_blocked_with_source_and_job_dependencies(defau
             upload_name="blocked.jpg",
             filename="blocked.jpg",
             file_path="uploads/blocked.jpg",
+            file_hash="9" * 64,
+            file_size_bytes=1,
         )
     )
     await transcriptions.create_job_source(
@@ -200,6 +206,8 @@ async def test_source_delete_blocks_when_linked_to_multiple_jobs(default_session
             upload_name="shared-page.jpg",
             filename="shared-page.jpg",
             file_path="uploads/shared-page.jpg",
+            file_hash="a" * 64,
+            file_size_bytes=1,
         )
     )
     await transcriptions.create_job_source(
@@ -228,6 +236,8 @@ async def test_update_job_source_transcription_persists_provider_json_payloads(d
             upload_name="provider.jpg",
             filename="provider.jpg",
             file_path="uploads/provider.jpg",
+            file_hash="b" * 64,
+            file_size_bytes=1,
         )
     )
     await transcriptions.create_job_source(
@@ -251,5 +261,6 @@ async def test_update_job_source_transcription_persists_provider_json_payloads(d
     stored_rows = await transcriptions.list_job_sources(job_id=job.id)
     assert len(stored_rows) == 1
     assert stored_rows[0].raw_transcription == "provider transcript"
+    assert stored_rows[0].prompt_name == "transcribe_document.md"
     assert stored_rows[0].ai_metadata == metadata
     assert stored_rows[0].raw_api_response == raw_payload

tests/services/test_workflows_reliability.py

@@ -8,6 +8,8 @@ import pytest
 from transcription.config import Settings
 from transcription.db.models import Document
 from transcription.db.models import Job
+from transcription.db.models import JobSource
+from transcription.db.models import JobSourceStatus
 from transcription.db.models import JobStatus
 from transcription.db.models import Source
 from transcription.services import ServiceBundle
@@ -49,12 +51,16 @@ class TestWorkflowReliability:
 
             source = Source(
                 document_id=document.id,
-                job_id=job.id,
+                page_number=1,
                 upload_name="timeout.jpg",
                 filename="timeout.jpg",
                 file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
+                file_hash="c" * 64,
+                file_size_bytes=1,
             )
             session.add(source)
+            await session.flush()
+            session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
             await session.commit()
 
             loaded = await services.jobs.read_job(job_id=job.id, session=session)

tests/test_models.py

@@ -64,6 +64,8 @@ def _persist_source(session, document: Document, *, page_number: int = 1, **over
         "raw_transcription": "Original machine text",
     }
     defaults.update(overrides)
+    defaults["file_hash"] = "a" * 64
+    defaults["file_size_bytes"] = 123
     source = Source(**defaults)
     session.add(source)
     session.commit()

tests/ui/conftest.py

@@ -95,16 +95,18 @@ async def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., Awai
                 retry_count=0,
                 provider="openrouter",
                 model="google/gemini-2.5-flash",
-                prompt_name="transcribe_document.md",
             )
             session.add(job)
             await session.flush()
 
             source = Source(
                 document_id=document.id,
+                page_number=1,
                 upload_name=filename,
                 filename=filename,
                 file_path=str(stored_path),
+                file_hash="b" * 64,
+                file_size_bytes=len(stored_path.read_bytes()),
             )
             session.add(source)
             await session.flush()
@@ -119,6 +121,7 @@ async def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., Awai
                             if transcription_text is not None
                             else JobSourceStatus.FAILED
                         ),
+                        prompt_name="transcribe_document.md",
                         raw_transcription=transcription_text,
                         error_detail=error_detail,
                     )

tests/ui/test_documents_page.py

@@ -146,6 +146,8 @@ class TestDocumentsPageRendering:
                 upload_name="page_1.png",
                 filename="page_1.png",
                 file_path="/tmp/page_1.png",
+                file_hash="0" * 64,
+                file_size_bytes=1,
             )
             session.add(source)
             await session.commit()

tests/ui/test_sources_page.py

@@ -23,6 +23,8 @@ class TestSourceModelProperties:
             upload_name="page_one.png",
             filename="stored_page_one.png",
             file_path="/tmp/stored_page_one.png",
+            file_hash="b" * 64,
+            file_size_bytes=1,
         )
 
         assert source.latest_job_source is None
@@ -86,6 +88,8 @@ class TestSourcesPageRendering:
                     upload_name="page_one.png",
                     filename="stored_page_one.png",
                     file_path="/tmp/stored_page_one.png",
+                    file_hash="c" * 64,
+                    file_size_bytes=1,
                 )
             )
             await session.commit()
@@ -114,6 +118,8 @@ class TestSourcesPageRendering:
                         upload_name="target_page.png",
                         filename="target_stored.png",
                         file_path="/tmp/target_stored.png",
+                        file_hash="d" * 64,
+                        file_size_bytes=1,
                     ),
                     Source(
                         document_id=other.id,
@@ -121,6 +127,8 @@ class TestSourcesPageRendering:
                         upload_name="other_page.png",
                         filename="other_stored.png",
                         file_path="/tmp/other_stored.png",
+                        file_hash="e" * 64,
+                        file_size_bytes=1,
                     ),
                 ]
             )
@@ -240,6 +248,8 @@ class TestSourcesPageRendering:
                 upload_name="orphan-source.png",
                 filename="orphan-source.png",
                 file_path="/tmp/orphan-source.png",
+                file_hash="f" * 64,
+                file_size_bytes=1,
             )
             session.add(source)
             await session.commit()