generated from john/python-template
zoltan57andCopilot App
7dd0d2c9bf
Decompose SourceService along the aggregate boundary and then correct the
instruction file that caused it to grow, in that order. The refactor is the
empirical test of the rule.
services/evidence.py (new)
EvidenceService owns ExecutionAttempt: read_latest_execution_attempt,
list_execution_attempts, promote_machine_attempt, build_evidence_export,
plus the LatestExecutionAttempt projection. Moved verbatim from sources.py.
services/errors.py (new)
The five-class error hierarchy (PromptLoadError, TranscriptionError,
TranscriptionNotFoundError, SourceDeleteBlockedError,
CandidatePromotionError) moved out of sources.py. evidence.py needs
TranscriptionNotFoundError, and test_service_boundaries.py correctly
rejected the sibling import. errors.py defines no *Service class, so it is
a legal shared home. This was the boundary test doing its job, not an
obstacle to route around.
sources.py 1,389 -> 885 lines (1,063 after Phase 2).
services/__init__.py
ServiceBundle and from_session_factory register evidence. Note that
field-by-field ServiceBundle construction silently binds services to the
process-global session factory via default_factory; from_session_factory is
the only safe constructor. Two test bundles were fixed for this.
.github/instructions/services.instructions.md
Rewritten to describe the boundaries the decomposition actually produced,
per plan Phase 3 task 7 and review log [59].
- "1 service class per data model" -> one service class per aggregate.
The table-shaped rule is the measured cause of sources.py reaching
1,389 lines; DocumentType has no lifecycle without Document.
- New Model Ownership section. Junctions are owned by their lifecycle
owner, the service that creates and deletes the rows: document_person
to PeopleService (sole writer, measured), job_source to SourceService.
Two carve-outs are stated rather than left as silent violations:
cascade deletion when a service deletes its own aggregate root, and
status transitions that create and delete nothing (cancel_job,
resubmit_failed_sources), which are Job lifecycle events on the work
queue. EvidenceService.promote_machine_attempt's two-field write to
Source is named and scoped.
- Mandatory CRUD softened to intent. It was already false: five modules
define no service class, EvidenceService has no create/delete because
ExecutionAttempt is append-only, RegistryService uses <op>_entry.
- Separated reading across models via eager loads from the owning root,
which is allowed, from importing another service, which is not. The old
line 13 and lines 75-77 read as contradictory.
- Typo: picutre.
No code was moved to satisfy the rule.
tests/test_service_boundaries.py
Docstring no longer cites the instruction file by line number; that anchor
would desynchronise silently. errors.py added to the neutral-module list.
Verified: 292 passed, 4 skipped, 0 ruff, 0 ty. All 25 /ui/* routes walked
against the live app; 24x 200. /ui/documents/{id}/sources 404s via a 307 that
drops the /ui prefix, confirmed pre-existing (last touched in 6a3ee26) and
left alone as out of scope.
Co-authored-by: Copilot App <[email protected]>
561 lines
20 KiB
Python
561 lines
20 KiB
Python
from datetime import UTC
|
|
from datetime import datetime
|
|
from datetime import timedelta
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import event
|
|
|
|
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.documents import DocumentService
|
|
from transcription.services.evidence import EvidenceService
|
|
from transcription.services.jobs import JobCancelBlockedError
|
|
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:
|
|
@pytest.mark.asyncio
|
|
async def test_create_and_read_job(self, job_service: JobService, document_service: DocumentService):
|
|
document = Document(id=uuid4(), name="test-bundle")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id)
|
|
await job_service.create_job(job=job)
|
|
|
|
fetched = await job_service.read_job(job_id=job.id)
|
|
assert fetched.id == job.id
|
|
assert fetched.document is not None
|
|
assert fetched.document.id == document.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_job_state_updates_status_and_retry(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="test-bundle")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id)
|
|
await job_service.create_job(job=job)
|
|
|
|
updated = await job_service.update_job_state(
|
|
job_id=job.id,
|
|
status=JobStatus.PROCESSING,
|
|
retry_count_increment=1,
|
|
)
|
|
|
|
assert updated.status == JobStatus.PROCESSING
|
|
assert updated.retry_count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_jobs_by_status(self, job_service: JobService, document_service: DocumentService):
|
|
document = Document(id=uuid4(), name="query-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
await job_service.create_job(job=Job(document_id=document.id, status=JobStatus.PROCESSING))
|
|
await job_service.create_job(job=Job(document_id=document.id, status=JobStatus.QUEUED))
|
|
|
|
result = await job_service.query_jobs(status=JobStatus.PROCESSING)
|
|
assert len(result) == 1
|
|
assert result[0].status == JobStatus.PROCESSING
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_jobs_by_source_filename(self, job_service: JobService, document_service: DocumentService):
|
|
document = Document(id=uuid4(), name="source-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
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()
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
result = await job_service.query_jobs(filename="stored-letter.jpg")
|
|
assert len(result) == 1
|
|
assert result[0].id == job.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_next_queued_job_claims_oldest_and_marks_processing(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="ordered-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
created_at = datetime.now(UTC)
|
|
first = Job(document_id=document.id, status=JobStatus.QUEUED, date_created=created_at)
|
|
second = Job(
|
|
document_id=document.id,
|
|
status=JobStatus.QUEUED,
|
|
date_created=created_at + timedelta(microseconds=1),
|
|
)
|
|
await job_service.create_job(job=first)
|
|
await job_service.create_job(job=second)
|
|
|
|
claimed = await job_service.claim_next_queued_job()
|
|
assert claimed is not None
|
|
assert claimed.id == first.id
|
|
assert claimed.status == JobStatus.PROCESSING
|
|
|
|
# The claim is exclusive: the same job is never handed out twice.
|
|
next_claim = await job_service.claim_next_queued_job()
|
|
assert next_claim is not None
|
|
assert next_claim.id == second.id
|
|
|
|
assert await job_service.claim_next_queued_job() is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_next_queued_job_emits_a_bounded_unadorned_query(
|
|
self,
|
|
job_service: JobService,
|
|
):
|
|
"""CRIT-01: the hot poll must not select a subgraph or scan the queue."""
|
|
statements: list[str] = []
|
|
|
|
async with job_service._session_scope() as session:
|
|
bind = session.get_bind()
|
|
|
|
def capture(_conn, _cursor, statement, *_rest):
|
|
statements.append(statement)
|
|
|
|
event.listen(bind, "before_cursor_execute", capture)
|
|
try:
|
|
await job_service.claim_next_queued_job(session=session)
|
|
finally:
|
|
event.remove(bind, "before_cursor_execute", capture)
|
|
|
|
selects = [item for item in statements if item.lstrip().upper().startswith("SELECT")]
|
|
assert len(selects) == 1, selects
|
|
assert "LIMIT" in selects[0].upper()
|
|
assert "JOIN" not in selects[0].upper()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_job_persists_provider_and_model(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="provider-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(
|
|
document_id=document.id,
|
|
provider="openrouter",
|
|
model="google/gemini-2.5-flash",
|
|
)
|
|
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"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_job_resolves_filename_from_linked_source(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="filename-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
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()
|
|
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
fetched = await job_service.read_job(job_id=job.id)
|
|
assert fetched.filename == "stored_page_001.png"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_job_with_guardrails_blocks_processing_jobs(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="processing-delete-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.PROCESSING)
|
|
await job_service.create_job(job=job)
|
|
|
|
with pytest.raises(JobDeleteBlockedError):
|
|
await job_service.delete_job_with_guardrails(job_id=job.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_job_with_guardrails_removes_jobsource_links(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="delete-job-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.QUEUED)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
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()
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
await job_service.delete_job_with_guardrails(job_id=job.id)
|
|
|
|
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)
|
|
evidence_service = EvidenceService(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 evidence_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,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="cancel-job-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.QUEUED)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source_one = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
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,
|
|
page_number=2,
|
|
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)
|
|
await session.flush()
|
|
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_one.id,
|
|
status=JobSourceStatus.TRANSCRIBED,
|
|
)
|
|
)
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_two.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
cancelled = await job_service.cancel_job(job_id=job.id)
|
|
assert cancelled.status == JobStatus.FAILED
|
|
|
|
refreshed = await job_service.read_job(job_id=job.id)
|
|
statuses = {item.status for item in refreshed.job_sources}
|
|
assert JobSourceStatus.TRANSCRIBED in statuses
|
|
assert JobSourceStatus.CANCELLED in statuses
|
|
assert JobSourceStatus.FAILED not in statuses
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resubmit_failed_sources_resets_only_failed(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="resubmit-job-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.FAILED)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source_one = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
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(
|
|
document_id=document.id,
|
|
page_number=2,
|
|
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)
|
|
session.add(source_two)
|
|
await session.flush()
|
|
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_one.id,
|
|
status=JobSourceStatus.FAILED,
|
|
)
|
|
)
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_two.id,
|
|
status=JobSourceStatus.TRANSCRIBED,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
count = await job_service.resubmit_failed_sources(job_id=job.id)
|
|
assert count == 1
|
|
|
|
refreshed = await job_service.read_job(job_id=job.id)
|
|
assert refreshed.status == JobStatus.QUEUED
|
|
|
|
failed_entry = next(
|
|
item for item in refreshed.job_sources if item.source is not None and item.source.page_number == 1
|
|
)
|
|
transcribed_entry = next(
|
|
item for item in refreshed.job_sources if item.source is not None and item.source.page_number == 2
|
|
)
|
|
assert failed_entry.status == JobSourceStatus.PENDING
|
|
assert failed_entry.source is not None
|
|
assert failed_entry.source.raw_transcription == "existing text"
|
|
assert transcribed_entry.status == JobSourceStatus.TRANSCRIBED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resubmit_failed_sources_blocks_when_only_pending_or_transcribed(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="resubmit-no-failed-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.FAILED)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source_one = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="resubmit-pending.jpg",
|
|
filename="stored-resubmit-pending.jpg",
|
|
file_path="/uploads/stored-resubmit-pending.jpg",
|
|
file_hash="1" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
source_two = Source(
|
|
document_id=document.id,
|
|
page_number=2,
|
|
upload_name="resubmit-done.jpg",
|
|
filename="stored-resubmit-done.jpg",
|
|
file_path="/uploads/stored-resubmit-done.jpg",
|
|
file_hash="2" * 64,
|
|
file_size_bytes=1,
|
|
raw_transcription="done text",
|
|
)
|
|
session.add(source_one)
|
|
session.add(source_two)
|
|
await session.flush()
|
|
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_one.id,
|
|
status=JobSourceStatus.PENDING,
|
|
)
|
|
)
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source_two.id,
|
|
status=JobSourceStatus.TRANSCRIBED,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
with pytest.raises(JobResubmitBlockedError):
|
|
await job_service.resubmit_failed_sources(job_id=job.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resubmit_failed_sources_includes_cancelled(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
"""Cancel is recoverable: cancelled pages are re-attempted on resubmit."""
|
|
document = Document(id=uuid4(), name="resubmit-cancelled-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.FAILED)
|
|
await job_service.create_job(job=job)
|
|
|
|
async with job_service._session_scope() as session:
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="resubmit-cancelled.jpg",
|
|
filename="stored-resubmit-cancelled.jpg",
|
|
file_path="/uploads/stored-resubmit-cancelled.jpg",
|
|
file_hash="3" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
session.add(
|
|
JobSource(
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
status=JobSourceStatus.CANCELLED,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
assert await job_service.resubmit_failed_sources(job_id=job.id) == 1
|
|
|
|
refreshed = await job_service.read_job(job_id=job.id)
|
|
assert refreshed.status == JobStatus.QUEUED
|
|
assert refreshed.job_sources[0].status == JobSourceStatus.PENDING
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resubmit_failed_sources_blocks_when_processing(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="resubmit-blocked-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.PROCESSING)
|
|
await job_service.create_job(job=job)
|
|
|
|
with pytest.raises(JobResubmitBlockedError):
|
|
await job_service.resubmit_failed_sources(job_id=job.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_job_blocks_transcribed_terminal_jobs(
|
|
self,
|
|
job_service: JobService,
|
|
document_service: DocumentService,
|
|
):
|
|
document = Document(id=uuid4(), name="cancel-blocked-doc")
|
|
await document_service.create_document(document=document)
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.TRANSCRIBED)
|
|
await job_service.create_job(job=job)
|
|
|
|
with pytest.raises(JobCancelBlockedError):
|
|
await job_service.cancel_job(job_id=job.id)
|