diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 98d2ada..67f3c9c 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -205,6 +205,10 @@ async def process_queued_job( # noqa: PLR0915 externally_stopped = False prompt_execution = _resolve_job_prompt_execution(source_job=source_job, settings=runtime_settings) + # Resolve the lazy provider property once, outside the timed region. First access + # constructs the HTTP client (~0.5s), which would otherwise be booked as provider + # latency on the first attempt of every worker process (review log [55]). + provider = services.sources.provider for source in sources: if await _job_no_longer_processing(job_id=job.id, services=services, session=session): @@ -212,6 +216,8 @@ async def process_queued_job( # noqa: PLR0915 break started_at = datetime.now(UTC) + # Fallback start for failures raised before the provider call; reset to the + # true call boundary immediately before the wait_for below. monotonic_started_at = asyncio.get_running_loop().time() result: TranscriptionResult | None = None provider_input = None @@ -225,12 +231,16 @@ async def process_queued_job( # noqa: PLR0915 media_type=provider_input.media_type, page_number=source.page_number, ) + # Restart the clock so duration_ms covers only what the wait_for below + # governs. The pre-loop assignment stays as the fallback for failures + # raised before this point, which would otherwise leave it unbound. + monotonic_started_at = asyncio.get_running_loop().time() result = await asyncio.wait_for( _call_transcriber( input_path=provider_input.path, prompt_execution=prompt_execution, settings=runtime_settings, - provider=services.sources.provider, + provider=provider, source_reference=source_reference, requested_model=source_job.model, ), @@ -283,8 +293,8 @@ async def process_queued_job( # noqa: PLR0915 0, int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000), ), - request_manifest=services.sources.provider.current_request_manifest, - transport_evidence=services.sources.provider.current_transport_evidence, + request_manifest=provider.current_request_manifest, + transport_evidence=provider.current_transport_evidence, failure_phase="local_timeout", ) failed_pages.append(page_outcome) diff --git a/src/transcription/ui/pages/sources_page.py b/src/transcription/ui/pages/sources_page.py index 271a2fc..b8e8576 100644 --- a/src/transcription/ui/pages/sources_page.py +++ b/src/transcription/ui/pages/sources_page.py @@ -366,6 +366,13 @@ def _render_source_job_metadata_zone( _render_provider_evidence(latest_attempt=latest_attempt) +def _format_duration(duration_ms: int) -> str: + """Render an attempt duration with a unit that suits its magnitude.""" + if duration_ms >= 1000: + return f"{duration_ms / 1000:.1f} s" + return f"{duration_ms} ms" + + def _render_provider_evidence(*, latest_attempt: LatestExecutionAttempt | None) -> None: ui.label("Provider Evidence").classes("text-xs font-semibold ui-text-primary mt-3") if latest_attempt is None: @@ -374,7 +381,7 @@ def _render_provider_evidence(*, latest_attempt: LatestExecutionAttempt | None) attempt = latest_attempt.attempt metadata_row("Attempt:", str(attempt.attempt_number)) - metadata_row("Duration:", f"{attempt.duration_ms} ms") + metadata_row("Duration:", _format_duration(attempt.duration_ms)) _render_json_evidence("Request Manifest", attempt.request_manifest) _render_json_evidence("Transport Response", _transport_display(latest_attempt)) _render_json_evidence("OpenRouter SDK Response Snapshot", attempt.sdk_response_snapshot) diff --git a/tests/services/test_workflows_reliability.py b/tests/services/test_workflows_reliability.py index e88cb84..5533338 100644 --- a/tests/services/test_workflows_reliability.py +++ b/tests/services/test_workflows_reliability.py @@ -1,6 +1,7 @@ """Reliability tests for worker workflow timeout behavior.""" import asyncio +import time from pathlib import Path from uuid import uuid4 @@ -18,6 +19,7 @@ from transcription.db.models import JobStatus from transcription.db.models import Source from transcription.providers.base import TranscriptionResult from transcription.services import ServiceBundle +from transcription.services import workflows as workflows_module from transcription.services.workflows import process_queued_job @@ -103,6 +105,79 @@ class TestWorkflowReliability: assert "timed out" in error_detail.lower() assert "20.0s" in error_detail + @pytest.mark.asyncio + async def test_timeout_duration_excludes_pre_call_setup(self, default_session_factory, monkeypatch): + """duration_ms covers only the provider call, not the setup preceding it. + + Regression guard for review log [55]: three historical ``local_timeout`` rows + recorded 0.4-2.0 s more than the configured budget because the measurement + window opened before payload resolution. Blocking setup is simulated here so + the assertion fails if that window ever reopens. + """ + services = ServiceBundle.from_session_factory(default_session_factory) + async with services.jobs._session_scope() as session: + document = Document(id=uuid4(), name="window-doc") + session.add(document) + await session.flush() + job = Job(document_id=document.id, status=JobStatus.QUEUED) + session.add(job) + await session.flush() + source = Source( + document_id=document.id, + page_number=1, + upload_name="window.jpg", + filename="window.jpg", + file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")), + file_hash="d" * 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) + + setup_seconds = 0.40 + budget_seconds = 0.20 + real_build = workflows_module.build_provider_input + + def _slow_build(source_arg): + time.sleep(setup_seconds) + return real_build(source_arg) + + async def _never_returns(*args, **kwargs): + _ = (args, kwargs) + await asyncio.sleep(budget_seconds * 20) + + monkeypatch.setattr("transcription.services.workflows.build_provider_input", _slow_build) + monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _never_returns) + + result = await process_queued_job( + job=loaded, + services=services, + settings=Settings(openrouter_api_key="test-key", worker_provider_timeout_seconds=budget_seconds), + ) + assert result is not None + assert result.status == JobStatus.FAILED + + async with services.jobs._session_scope() as session: + attempts = ( + ( + await session.exec( + select(ExecutionAttempt).where( + col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources]) + ) + ) + ) + .all() + ) + assert len(attempts) == 1 + duration_ms = attempts[0].duration_ms + + # At or just above the budget, and well clear of budget + setup. + assert duration_ms >= int(budget_seconds * 1000 * 0.9) + assert duration_ms < int((budget_seconds + setup_seconds) * 1000 * 0.9) + @pytest.mark.asyncio async def test_completed_page_is_committed_before_next_provider_call_finishes( self,