From fb65e070c51a05a61ba5787224b292c667f86a58 Mon Sep 17 00:00:00 2001 From: Susanna Shu Date: Mon, 27 Jul 2026 10:48:41 -0400 Subject: [PATCH 1/2] feat(jobs): filter job listing by namespace and add job deletion Adds two capabilities the multi-tenant proxy in keyboard-backend needs to own its own view of a user's uploads. Namespace filtering - `GET /v1/jobs` and `GET /v2/jobs` accept `namespace`, matched against `job_metadata ->> 'namespace'`. - The filter is normalized the same way job creation normalizes it, so a blank or whitespace value means the `default` namespace instead of silently matching nothing. Omitting the parameter still lists every namespace. - Jobs written before the namespace was recorded have no key at all; those count as `default` so the filter does not hide them. - `JobReadModel.list_jobs_for_user` now accepts and forwards `namespace`. Without this the route raised TypeError on every list call. Job deletion - `DELETE /v1/jobs/{job_id}` and `DELETE /v2/jobs/{job_id}` soft-delete a job via a new `jobs.deleted_at` column. The row is retained so in-flight workers, billing records, and audit history are unaffected; the job just stops appearing in the read APIs. - The linked document is archived by default (`archive_document=false` to opt out), but only when no other live job still targets it, so re-parsing a document does not lose its content. - Deleting twice returns 404 the second time, matching the behavior of an unknown job. Soft-deleted jobs are also invisible to GET and to confirm-upload. Adds partial indexes for the two new query shapes: a user's active jobs by creation time, and active jobs by namespace. Co-Authored-By: Claude Opus 5 --- ...add_job_soft_delete_and_namespace_index.py | 50 ++++++ apps/api/app/api/v1/routes/jobs.py | 35 +++++ apps/api/app/api/v2/routes/jobs.py | 34 +++++ apps/api/app/repositories/job_repository.py | 82 +++++++++- apps/api/app/services/jobs/__init__.py | 2 + apps/api/app/services/jobs/job_read_model.py | 105 ++++++++++++- apps/api/app/services/jobs/read_service.py | 20 ++- .../contract/test_job_delete_contract.py | 144 ++++++++++++++++++ .../tests/contract/test_job_read_contract.py | 75 ++++++++- .../tests/migrations/test_schema_contract.py | 36 +++++ .../shared/models/database/job.py | 19 +++ .../shared/models/schemas/job.py | 20 +++ .../models/schemas/retrieval_namespace.py | 7 +- 13 files changed, 620 insertions(+), 9 deletions(-) create mode 100644 apps/api/alembic/versions/fce1d2e3f4a5_add_job_soft_delete_and_namespace_index.py create mode 100644 apps/api/tests/contract/test_job_delete_contract.py diff --git a/apps/api/alembic/versions/fce1d2e3f4a5_add_job_soft_delete_and_namespace_index.py b/apps/api/alembic/versions/fce1d2e3f4a5_add_job_soft_delete_and_namespace_index.py new file mode 100644 index 000000000..48a75af54 --- /dev/null +++ b/apps/api/alembic/versions/fce1d2e3f4a5_add_job_soft_delete_and_namespace_index.py @@ -0,0 +1,50 @@ +"""Add job soft deletion and namespace listing index. + +Revision ID: fce1d2e3f4a5 +Revises: fbe1c2d3e4f5 +Create Date: 2026-07-27 10:30:00.000000 +""" + +from __future__ import annotations + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + + +# revision identifiers, used by Alembic. +revision: str = "fce1d2e3f4a5" +down_revision: Union[str, Sequence[str], None] = "fbe1c2d3e4f5" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "jobs", + sa.Column( + "deleted_at", + sa.DateTime(), + nullable=True, + comment="Soft-deletion time; NULL when active", + ), + ) + op.create_index( + "idx_job_user_active_created_at", + "jobs", + ["user_id", "created_at"], + postgresql_where=sa.text("deleted_at IS NULL"), + ) + op.create_index( + "idx_job_user_namespace", + "jobs", + ["user_id", sa.text("(job_metadata ->> 'namespace')")], + postgresql_where=sa.text("deleted_at IS NULL"), + ) + + +def downgrade() -> None: + op.drop_index("idx_job_user_namespace", table_name="jobs") + op.drop_index("idx_job_user_active_created_at", table_name="jobs") + op.drop_column("jobs", "deleted_at") diff --git a/apps/api/app/api/v1/routes/jobs.py b/apps/api/app/api/v1/routes/jobs.py index 25562ecbd..2736d5105 100644 --- a/apps/api/app/api/v1/routes/jobs.py +++ b/apps/api/app/api/v1/routes/jobs.py @@ -12,6 +12,7 @@ from app.api.dependencies.job_admission import require_billing_limits from app.services.document_ingestion import DocumentIngestionService from app.services.jobs import ( + delete_job_for_user, get_job_result_for_user, list_jobs_for_user, ) @@ -23,6 +24,7 @@ from shared.models.schemas.job import ( ConfirmUploadRequest, JobCreate, + JobDeleteResponse, JobList, JobResponse, JobResultResponse, @@ -69,6 +71,7 @@ async def list_jobs( None, description="Start time in ISO format" ), end_time: Optional[datetime] = Query(None, description="End time in ISO format"), + namespace: Optional[str] = Query(None, description="Namespace filter"), current_user: CurrentUser = Depends(with_current_user), db: AsyncSession = Depends(get_db), ): @@ -85,6 +88,7 @@ async def list_jobs( recent_days=recent_days, start_time=start_time, end_time=end_time, + namespace=namespace, ) @@ -104,6 +108,37 @@ async def get_job_result( ) +@router.delete( + "/{job_id}", response_model=JobDeleteResponse, summary="Delete a job" +) +async def delete_job( + job_id: str, + archive_document: bool = Query( + True, + description=( + "Archive the document this job produced, removing it from " + "retrieval. Skipped when another live job still targets it." + ), + ), + current_user: CurrentUser = Depends(with_current_user), + _write_permission: None = Depends(require_write_permission), + db: AsyncSession = Depends(get_db), +): + """ + Delete a job. + + The job is soft-deleted: its row is retained so in-flight workers, billing + records, and audit history stay intact, but it no longer appears in the + job read APIs. + """ + return await delete_job_for_user( + db, + job_id=job_id, + user_id=current_user.user_id, + archive_document=archive_document, + ) + + @router.post( "/{job_id}/confirm-upload", response_model=dict, diff --git a/apps/api/app/api/v2/routes/jobs.py b/apps/api/app/api/v2/routes/jobs.py index c9116bc05..44ce55110 100644 --- a/apps/api/app/api/v2/routes/jobs.py +++ b/apps/api/app/api/v2/routes/jobs.py @@ -10,6 +10,7 @@ from app.api.dependencies.job_admission import require_billing_limits from app.services.document_ingestion import DocumentIngestionService from app.services.jobs import ( + delete_job_for_user, get_job_result_for_user, list_jobs_for_user, ) @@ -21,6 +22,7 @@ from shared.models.schemas.job import ( ConfirmUploadRequest, JobCreateV2, + JobDeleteResponse, JobList, JobResponse, JobResultResponse, @@ -62,6 +64,7 @@ async def list_jobs( None, description="Start time in ISO format" ), end_time: Optional[datetime] = Query(None, description="End time in ISO format"), + namespace: Optional[str] = Query(None, description="Namespace filter"), current_user: CurrentUser = Depends(with_current_user), db: AsyncSession = Depends(get_db), ): @@ -76,6 +79,7 @@ async def list_jobs( recent_days=recent_days, start_time=start_time, end_time=end_time, + namespace=namespace, ) @@ -93,6 +97,36 @@ async def get_job_result( ) +@router.delete( + "/{job_id}", response_model=JobDeleteResponse, summary="Delete a job" +) +async def delete_job( + job_id: str, + archive_document: bool = Query( + True, + description=( + "Archive the document this job produced, removing it from " + "retrieval. Skipped when another live job still targets it." + ), + ), + current_user: CurrentUser = Depends(with_current_user), + _write_permission: None = Depends(require_write_permission), + db: AsyncSession = Depends(get_db), +): + """Delete a job. + + The job is soft-deleted: its row is retained so in-flight workers, billing + records, and audit history stay intact, but it no longer appears in the + job read APIs. + """ + return await delete_job_for_user( + db, + job_id=job_id, + user_id=current_user.user_id, + archive_document=archive_document, + ) + + @router.post( "/{job_id}/confirm-upload", response_model=dict, diff --git a/apps/api/app/repositories/job_repository.py b/apps/api/app/repositories/job_repository.py index c3750352b..13c6cec13 100644 --- a/apps/api/app/repositories/job_repository.py +++ b/apps/api/app/repositories/job_repository.py @@ -4,7 +4,7 @@ from typing import Any, Dict, Optional, Sequence from loguru import logger -from sqlalchemy import and_, desc, select, update +from sqlalchemy import and_, desc, or_, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload @@ -12,6 +12,21 @@ from shared.core.state_machine.service import AsyncStateMachineService from shared.models.database.job import Job from shared.models.database.job_state_history import JobStateHistory +from shared.models.schemas.retrieval_namespace import DEFAULT_RETRIEVAL_NAMESPACE +from shared.utils.utc_now import utc_now_naive + + +def _namespace_matches(namespace: str): + """Build the namespace predicate for job listing. + + ``namespace`` is expected to already be normalized. Jobs written before + the namespace was recorded in ``job_metadata`` have no key at all; those + belong to the default namespace, so the default filter must match them too. + """ + stored_namespace = Job.job_metadata.op("->>")("namespace") + if namespace == DEFAULT_RETRIEVAL_NAMESPACE: + return or_(stored_namespace == namespace, stored_namespace.is_(None)) + return stored_namespace == namespace class JobRepository: @@ -92,13 +107,14 @@ async def get_jobs_by_user( created_before: Optional[datetime] = None, job_type: Optional[str] = None, job_status: Optional[str] = None, + namespace: Optional[str] = None, ) -> Sequence[Job]: """Get jobs for a user.""" try: stmt = ( select(Job) .options(selectinload(Job.job_result)) - .where(Job.user_id == user_id) + .where(Job.user_id == user_id, Job.deleted_at.is_(None)) .order_by(desc(Job.created_at)) .limit(limit) .offset(offset) @@ -111,6 +127,8 @@ async def get_jobs_by_user( stmt = stmt.where(Job.job_type == job_type) if job_status: stmt = stmt.where(Job.status == job_status) + if namespace: + stmt = stmt.where(_namespace_matches(namespace)) result = await db.execute(stmt) return result.scalars().all() except Exception as e: @@ -125,12 +143,17 @@ async def count_jobs_by_user( created_before: Optional[datetime] = None, job_type: Optional[str] = None, job_status: Optional[str] = None, + namespace: Optional[str] = None, ) -> int: """Count jobs for a user.""" try: from sqlalchemy import func - stmt = select(func.count()).select_from(Job).where(Job.user_id == user_id) + stmt = ( + select(func.count()) + .select_from(Job) + .where(Job.user_id == user_id, Job.deleted_at.is_(None)) + ) if created_after: stmt = stmt.where(Job.created_at >= created_after) if created_before: @@ -139,12 +162,65 @@ async def count_jobs_by_user( stmt = stmt.where(Job.job_type == job_type) if job_status: stmt = stmt.where(Job.status == job_status) + if namespace: + stmt = stmt.where(_namespace_matches(namespace)) result = await db.execute(stmt) return result.scalar() or 0 except Exception as e: logger.error(f"Failed to count jobs for user {user_id}: {e}") return 0 + async def soft_delete_job(self, db: AsyncSession, job_id: str) -> bool: + """Mark a Job as deleted. + + Returns True when this call performed the deletion, False when the job + was already deleted (or no longer exists), which makes the operation + idempotent for retrying clients. + """ + try: + stmt = ( + update(Job) + .where(Job.job_id == job_id, Job.deleted_at.is_(None)) + .values(deleted_at=utc_now_naive()) + ) + result = await db.execute(stmt) + await db.commit() + return (result.rowcount or 0) > 0 + except Exception as e: + logger.error(f"Failed to soft-delete job {job_id}: {e}") + await db.rollback() + raise + + async def count_active_jobs_for_document( + self, + db: AsyncSession, + user_id: str, + document_id: str, + exclude_job_id: Optional[str] = None, + ) -> int: + """Count a user's non-deleted jobs that target the given document.""" + try: + from sqlalchemy import func + + stmt = ( + select(func.count()) + .select_from(Job) + .where( + Job.user_id == user_id, + Job.deleted_at.is_(None), + Job.job_metadata.op("->>")("document_id") == document_id, + ) + ) + if exclude_job_id: + stmt = stmt.where(Job.job_id != exclude_job_id) + result = await db.execute(stmt) + return result.scalar() or 0 + except Exception as e: + logger.error( + f"Failed to count active jobs for document {document_id}: {e}" + ) + return 0 + async def update_job_state( self, db: AsyncSession, diff --git a/apps/api/app/services/jobs/__init__.py b/apps/api/app/services/jobs/__init__.py index fcdc2ccdc..0f8a040e4 100644 --- a/apps/api/app/services/jobs/__init__.py +++ b/apps/api/app/services/jobs/__init__.py @@ -1,11 +1,13 @@ from app.services.jobs.read_service import ( check_job_permission, + delete_job_for_user, get_job_result_for_user, list_jobs_for_user, ) __all__ = [ "check_job_permission", + "delete_job_for_user", "get_job_result_for_user", "list_jobs_for_user", ] diff --git a/apps/api/app/services/jobs/job_read_model.py b/apps/api/app/services/jobs/job_read_model.py index 1d0a4ab39..1866bf782 100644 --- a/apps/api/app/services/jobs/job_read_model.py +++ b/apps/api/app/services/jobs/job_read_model.py @@ -18,7 +18,9 @@ PermissionDeniedException, ValidationException, ) -from shared.models.schemas.job import JobList, JobResultResponse +from shared.models.schemas.job import JobDeleteResponse, JobList, JobResultResponse +from shared.models.schemas.job_metadata import JobMetadataHelper +from shared.models.schemas.retrieval_namespace import normalize_retrieval_namespace from shared.services.redis import RedisServiceFactory from shared.utils.utc_now import utc_now_naive @@ -36,6 +38,7 @@ async def list_jobs_for_user( recent_days: Optional[int], start_time: Optional[datetime], end_time: Optional[datetime], + namespace: Optional[str] = None, ) -> JobList: return await list_jobs_for_user( db, @@ -47,6 +50,7 @@ async def list_jobs_for_user( recent_days=recent_days, start_time=start_time, end_time=end_time, + namespace=namespace, ) async def get_job_result_for_user( @@ -58,6 +62,21 @@ async def get_job_result_for_user( ) -> JobResultResponse: return await get_job_result_for_user(db, job_id=job_id, user_id=user_id) + async def delete_job_for_user( + self, + db: AsyncSession, + *, + job_id: str, + user_id: str, + archive_document: bool = True, + ) -> JobDeleteResponse: + return await delete_job_for_user( + db, + job_id=job_id, + user_id=user_id, + archive_document=archive_document, + ) + def check_job_permission(job, user_id: str, job_id: str) -> None: if not job: @@ -70,6 +89,13 @@ def check_job_permission(job, user_id: str, job_id: str) -> None: user_message="You don't have permission to access this job", ) + # A soft-deleted job is invisible to the owner, exactly like a missing one. + # The permission check runs first so deletion never leaks job existence. + if getattr(job, "deleted_at", None) is not None: + raise NotFoundException( + resource="Job", resource_id=job_id, internal_message="Job is deleted" + ) + def normalize_naive_utc_filter_datetime(dt: Optional[datetime]) -> Optional[datetime]: if dt is None: @@ -90,6 +116,7 @@ async def list_jobs_for_user( recent_days: Optional[int], start_time: Optional[datetime], end_time: Optional[datetime], + namespace: Optional[str] = None, ) -> JobList: try: job_repo = JobRepository() @@ -123,6 +150,13 @@ async def list_jobs_for_user( created_after = normalized_start_time created_before = normalized_end_time + # Jobs persist the normalized namespace, so the filter has to be + # normalized the same way or an unnormalized query silently matches + # nothing. An omitted namespace means "no filter", not "default". + normalized_namespace = ( + normalize_retrieval_namespace(namespace) if namespace is not None else None + ) + total_count = await job_repo.count_jobs_by_user( db=db, user_id=user_id, @@ -130,6 +164,7 @@ async def list_jobs_for_user( created_before=created_before, job_type=job_type, job_status=job_status, + namespace=normalized_namespace, ) jobs = await job_repo.get_jobs_by_user( db=db, @@ -140,6 +175,7 @@ async def list_jobs_for_user( created_before=created_before, job_type=job_type, job_status=job_status, + namespace=normalized_namespace, ) redis_service = RedisServiceFactory.get_service() @@ -207,3 +243,70 @@ async def get_job_result_for_user( raise JobOperationException( internal_message=f"Failed to get job result: {str(exc)}" ) + + +async def delete_job_for_user( + db: AsyncSession, + *, + job_id: str, + user_id: str, + archive_document: bool = True, +) -> JobDeleteResponse: + """Soft-delete a job and, by default, archive the document it produced. + + The job row is retained so in-flight workers, billing records, and audit + history are unaffected; it simply stops appearing in the read APIs. The + linked document is archived only when no other live job still targets it, + which keeps a re-parse of the same document from losing its content. + """ + from app.services.documents.lifecycle_service import DocumentService + + try: + job_repo = JobRepository() + job = await job_repo.get_job_by_id(db, job_id) + check_job_permission(job, user_id, job_id) + assert job is not None + + redis_service = RedisServiceFactory.get_service() + job_metadata = await job_repo.get_job_metadata(db, job_id, redis_service) + document_id = JobMetadataHelper.get_document_id(job_metadata) + + await job_repo.soft_delete_job(db, job_id) + + document_archived = False + if archive_document and document_id: + remaining_jobs = await job_repo.count_active_jobs_for_document( + db, + user_id=user_id, + document_id=document_id, + exclude_job_id=job_id, + ) + if remaining_jobs == 0: + document = await DocumentService().archive_document( + db, + user_id=user_id, + document_id=document_id, + ) + document_archived = document is not None + else: + logger.info( + f"Skipped archiving document {document_id}: " + f"{remaining_jobs} other job(s) still reference it" + ) + + return JobDeleteResponse( + job_id=job_id, + deleted=True, + document_id=document_id, + document_archived=document_archived, + ) + + except NotFoundException: + raise + except PermissionDeniedException: + raise + except Exception as exc: + logger.error(f"Failed to delete job {job_id}: {exc}") + raise JobOperationException( + internal_message=f"Failed to delete job: {str(exc)}" + ) diff --git a/apps/api/app/services/jobs/read_service.py b/apps/api/app/services/jobs/read_service.py index c055dd2e5..d2a7ff9aa 100644 --- a/apps/api/app/services/jobs/read_service.py +++ b/apps/api/app/services/jobs/read_service.py @@ -7,10 +7,11 @@ from app.services.jobs.job_read_model import check_job_permission from sqlalchemy.ext.asyncio import AsyncSession -from shared.models.schemas.job import JobList, JobResultResponse +from shared.models.schemas.job import JobDeleteResponse, JobList, JobResultResponse __all__ = [ "check_job_permission", + "delete_job_for_user", "get_job_result_for_user", "list_jobs_for_user", ] @@ -27,6 +28,7 @@ async def list_jobs_for_user( recent_days: Optional[int], start_time: Optional[datetime], end_time: Optional[datetime], + namespace: Optional[str] = None, ) -> JobList: return await JobReadModel().list_jobs_for_user( db, @@ -38,6 +40,7 @@ async def list_jobs_for_user( recent_days=recent_days, start_time=start_time, end_time=end_time, + namespace=namespace, ) @@ -52,3 +55,18 @@ async def get_job_result_for_user( job_id=job_id, user_id=user_id, ) + + +async def delete_job_for_user( + db: AsyncSession, + *, + job_id: str, + user_id: str, + archive_document: bool = True, +) -> JobDeleteResponse: + return await JobReadModel().delete_job_for_user( + db, + job_id=job_id, + user_id=user_id, + archive_document=archive_document, + ) diff --git a/apps/api/tests/contract/test_job_delete_contract.py b/apps/api/tests/contract/test_job_delete_contract.py new file mode 100644 index 000000000..3b741a22e --- /dev/null +++ b/apps/api/tests/contract/test_job_delete_contract.py @@ -0,0 +1,144 @@ +from collections.abc import Callable +from contextlib import AbstractAsyncContextManager +from typing import cast +from uuid import uuid4 + +import pytest +from httpx import AsyncClient + + +async def _create_waiting_file_job( + api_client: AsyncClient, + *, + namespace: str = "contract-jobs-delete", +) -> dict[str, object]: + payload: dict[str, str] = { + "namespace": namespace, + "source_type": "file", + "file_name": "contract-delete.pdf", + "data_id": f"contract-job-delete-{uuid4().hex[:12]}", + } + + response = await api_client.post("/api/v1/jobs", json=payload) + + assert response.status_code == 200 + return cast(dict[str, object], response.json()) + + +@pytest.mark.asyncio +async def test_should_delete_a_job_and_hide_it_from_the_job_apis( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + async with developer_api_client_factory() as api_client: + created_job = await _create_waiting_file_job(api_client) + job_id = cast(str, created_job["job_id"]) + + delete_response = await api_client.delete(f"/api/v1/jobs/{job_id}") + get_response = await api_client.get(f"/api/v1/jobs/{job_id}") + list_response = await api_client.get("/api/v1/jobs") + + assert delete_response.status_code == 200 + + delete_json = cast(dict[str, object], delete_response.json()) + assert delete_json["job_id"] == job_id + assert delete_json["deleted"] is True + assert delete_json["document_id"] == created_job["document_id"] + + assert get_response.status_code == 404 + + assert list_response.status_code == 200 + list_json = cast(dict[str, object], list_response.json()) + assert list_json["total"] == 0 + assert list_json["jobs"] == [] + + +@pytest.mark.asyncio +async def test_should_keep_other_jobs_listed_after_deleting_one( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + async with developer_api_client_factory() as api_client: + deleted_job = await _create_waiting_file_job(api_client) + surviving_job = await _create_waiting_file_job(api_client) + + await api_client.delete(f"/api/v1/jobs/{deleted_job['job_id']}") + list_response = await api_client.get("/api/v1/jobs") + + assert list_response.status_code == 200 + + list_json = cast(dict[str, object], list_response.json()) + jobs = cast(list[dict[str, object]], list_json["jobs"]) + + assert list_json["total"] == 1 + assert len(jobs) == 1 + assert jobs[0]["job_id"] == surviving_job["job_id"] + + +@pytest.mark.asyncio +async def test_should_return_not_found_when_deleting_an_already_deleted_job( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + async with developer_api_client_factory() as api_client: + created_job = await _create_waiting_file_job(api_client) + job_id = cast(str, created_job["job_id"]) + + first_response = await api_client.delete(f"/api/v1/jobs/{job_id}") + second_response = await api_client.delete(f"/api/v1/jobs/{job_id}") + + assert first_response.status_code == 200 + assert second_response.status_code == 404 + + response_json = cast(dict[str, object], second_response.json()) + error = cast(dict[str, object], response_json["error"]) + + assert response_json["success"] is False + assert error["code"] == "NOT_FOUND" + + +@pytest.mark.asyncio +async def test_should_return_not_found_when_deleting_an_unknown_job( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + missing_job_id = f"job_missing_{uuid4().hex[:12]}" + + async with developer_api_client_factory() as api_client: + response = await api_client.delete(f"/api/v1/jobs/{missing_job_id}") + + assert response.status_code == 404 + + response_json = cast(dict[str, object], response.json()) + error = cast(dict[str, object], response_json["error"]) + + assert error["code"] == "NOT_FOUND" + assert error["details"] == { + "resource": "Job", + "id": missing_job_id, + } + + +@pytest.mark.asyncio +async def test_should_not_archive_the_document_when_archiving_is_disabled( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + async with developer_api_client_factory() as api_client: + created_job = await _create_waiting_file_job(api_client) + job_id = cast(str, created_job["job_id"]) + + response = await api_client.delete( + f"/api/v1/jobs/{job_id}", params={"archive_document": "false"} + ) + + assert response.status_code == 200 + + response_json = cast(dict[str, object], response.json()) + assert response_json["deleted"] is True + assert response_json["document_archived"] is False diff --git a/apps/api/tests/contract/test_job_read_contract.py b/apps/api/tests/contract/test_job_read_contract.py index ef6c68617..adc09ce9e 100644 --- a/apps/api/tests/contract/test_job_read_contract.py +++ b/apps/api/tests/contract/test_job_read_contract.py @@ -10,13 +10,18 @@ from tests.support.contract_database import ContractDatabase -async def _create_waiting_file_job(api_client: AsyncClient) -> dict[str, object]: +async def _create_waiting_file_job( + api_client: AsyncClient, + *, + namespace: str | None = "contract-jobs", +) -> dict[str, object]: payload: dict[str, str] = { - "namespace": "contract-jobs", "source_type": "file", "file_name": "contract-read.pdf", "data_id": f"contract-job-read-{uuid4().hex[:12]}", } + if namespace is not None: + payload["namespace"] = namespace response = await api_client.post("/api/v1/jobs", json=payload) @@ -227,3 +232,69 @@ async def test_should_forbid_access_to_a_job_owned_by_another_user( assert error["code"] == "PERMISSION_DENIED" assert error["message"] == "You don't have permission to access this job" assert "details" not in error + + +@pytest.mark.asyncio +async def test_should_list_only_jobs_in_the_requested_namespace( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + async with developer_api_client_factory() as api_client: + scoped_job = await _create_waiting_file_job(api_client, namespace="tenant-a") + await _create_waiting_file_job(api_client, namespace="tenant-b") + + response = await api_client.get( + "/api/v1/jobs", params={"namespace": "tenant-a"} + ) + + assert response.status_code == 200 + + response_json = cast(dict[str, object], response.json()) + jobs = cast(list[dict[str, object]], response_json["jobs"]) + + assert response_json["total"] == 1 + assert len(jobs) == 1 + assert jobs[0]["job_id"] == scoped_job["job_id"] + assert jobs[0]["namespace"] == "tenant-a" + + +@pytest.mark.asyncio +async def test_should_list_every_namespace_when_no_namespace_filter_is_given( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + async with developer_api_client_factory() as api_client: + await _create_waiting_file_job(api_client, namespace="tenant-a") + await _create_waiting_file_job(api_client, namespace="tenant-b") + + response = await api_client.get("/api/v1/jobs") + + assert response.status_code == 200 + + response_json = cast(dict[str, object], response.json()) + + assert response_json["total"] == 2 + + +@pytest.mark.asyncio +async def test_should_treat_a_blank_namespace_filter_as_the_default_namespace( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + async with developer_api_client_factory() as api_client: + default_job = await _create_waiting_file_job(api_client, namespace=None) + await _create_waiting_file_job(api_client, namespace="tenant-a") + + response = await api_client.get("/api/v1/jobs", params={"namespace": " "}) + + assert response.status_code == 200 + + response_json = cast(dict[str, object], response.json()) + jobs = cast(list[dict[str, object]], response_json["jobs"]) + + assert response_json["total"] == 1 + assert jobs[0]["job_id"] == default_job["job_id"] + assert jobs[0]["namespace"] == "default" diff --git a/apps/api/tests/migrations/test_schema_contract.py b/apps/api/tests/migrations/test_schema_contract.py index 830a2cbd1..e98515f36 100644 --- a/apps/api/tests/migrations/test_schema_contract.py +++ b/apps/api/tests/migrations/test_schema_contract.py @@ -277,3 +277,39 @@ def test_agentic_retrieval_trace_schema_matches_orm(migrated_head_engine: Engine "workflow_step_id", "workflow_plan", }.issubset(run_columns) + + +def test_jobs_schema_should_expose_soft_deletion(migrated_head_engine: Engine) -> None: + with migrated_head_engine.begin() as connection: + job_columns = set( + connection.execute( + text( + """ + SELECT column_name + FROM information_schema.columns + WHERE table_name = 'jobs' + """ + ) + ) + .scalars() + .all() + ) + job_indexes = set( + connection.execute( + text( + """ + SELECT indexname + FROM pg_indexes + WHERE tablename = 'jobs' + """ + ) + ) + .scalars() + .all() + ) + + assert "deleted_at" in job_columns + assert { + "idx_job_user_active_created_at", + "idx_job_user_namespace", + }.issubset(job_indexes) diff --git a/packages/shared-python/shared/models/database/job.py b/packages/shared-python/shared/models/database/job.py index 8a4fb6bf3..f8954f9ec 100644 --- a/packages/shared-python/shared/models/database/job.py +++ b/packages/shared-python/shared/models/database/job.py @@ -85,6 +85,11 @@ class Job(Base): onupdate=utc_now_naive, nullable=False, ) + # Soft deletion. A deleted job is hidden from the read APIs but keeps its + # row so in-flight workers, billing records, and audit history stay intact. + deleted_at: Mapped[Optional[datetime]] = mapped_column( + DateTime, nullable=True, comment="Soft-deletion time; NULL when active" + ) # Billing Information (Per-Page Billing) page_count: Mapped[Optional[int]] = mapped_column( @@ -130,6 +135,20 @@ class Job(Base): Index("idx_job_type", "job_type"), Index("idx_job_created_at", "created_at"), Index("idx_job_user_status", "user_id", "status"), + # Supports the default list query: a user's active jobs, newest first. + Index( + "idx_job_user_active_created_at", + "user_id", + "created_at", + postgresql_where=text("deleted_at IS NULL"), + ), + # Supports namespace-scoped listing for multi-tenant proxies. + Index( + "idx_job_user_namespace", + "user_id", + text("(job_metadata ->> 'namespace')"), + postgresql_where=text("deleted_at IS NULL"), + ), Index( "idx_job_user_active_states", "user_id", diff --git a/packages/shared-python/shared/models/schemas/job.py b/packages/shared-python/shared/models/schemas/job.py index 1ea775fde..65ab6cbf2 100644 --- a/packages/shared-python/shared/models/schemas/job.py +++ b/packages/shared-python/shared/models/schemas/job.py @@ -200,6 +200,26 @@ class JobList(BaseModel): total_pages: int = Field(..., description="Total page count") +class JobDeleteResponse(BaseModel): + """Job-deletion response.""" + + job_id: str = Field(..., description="Deleted job ID") + deleted: bool = Field( + ..., description="Always true; deletion is idempotent for the owner" + ) + document_id: Optional[str] = Field( + None, description="Document the deleted job targeted, when it had one" + ) + document_archived: bool = Field( + False, + description=( + "Whether the linked document was archived as part of this call. " + "False when archiving was not requested or another live job still " + "targets the document." + ), + ) + + class ConfirmUploadRequest(BaseModel): """Request payload for confirming a file upload.""" diff --git a/packages/shared-python/shared/models/schemas/retrieval_namespace.py b/packages/shared-python/shared/models/schemas/retrieval_namespace.py index ef2be2349..bfa54627f 100644 --- a/packages/shared-python/shared/models/schemas/retrieval_namespace.py +++ b/packages/shared-python/shared/models/schemas/retrieval_namespace.py @@ -1,9 +1,12 @@ from __future__ import annotations -_DEFAULT_RETRIEVAL_NAMESPACE = "default" +DEFAULT_RETRIEVAL_NAMESPACE = "default" + +# Backwards-compatible private alias. +_DEFAULT_RETRIEVAL_NAMESPACE = DEFAULT_RETRIEVAL_NAMESPACE def normalize_retrieval_namespace(namespace: str | None) -> str: """Return the canonical namespace value used by jobs, retrieval, and caches.""" normalized = str(namespace or "").strip() - return normalized or _DEFAULT_RETRIEVAL_NAMESPACE + return normalized or DEFAULT_RETRIEVAL_NAMESPACE From de47aaa2cf5c80ca728dcc6d0bd300d7ff4231ab Mon Sep 17 00:00:00 2001 From: Susanna Shu Date: Mon, 27 Jul 2026 11:06:39 -0400 Subject: [PATCH 2/2] test(jobs): seed namespace-filter fixtures instead of creating jobs The namespace-filter tests created two jobs over HTTP, which tripped the job-creation rate limit in CI and returned 429 on the second create. They now insert job rows directly via ContractDatabase, so each test makes a single request and the namespace fixture is explicit. Adds coverage for a job whose job_metadata has no namespace key at all, which the default filter must still match. Also drops the unused _DEFAULT_RETRIEVAL_NAMESPACE alias flagged by CodeQL; nothing referenced the private name. Co-Authored-By: Claude Opus 5 --- .../tests/contract/test_job_read_contract.py | 78 +++++++++++++++---- .../models/schemas/retrieval_namespace.py | 3 - 2 files changed, 62 insertions(+), 19 deletions(-) diff --git a/apps/api/tests/contract/test_job_read_contract.py b/apps/api/tests/contract/test_job_read_contract.py index adc09ce9e..d7363806a 100644 --- a/apps/api/tests/contract/test_job_read_contract.py +++ b/apps/api/tests/contract/test_job_read_contract.py @@ -10,18 +10,13 @@ from tests.support.contract_database import ContractDatabase -async def _create_waiting_file_job( - api_client: AsyncClient, - *, - namespace: str | None = "contract-jobs", -) -> dict[str, object]: +async def _create_waiting_file_job(api_client: AsyncClient) -> dict[str, object]: payload: dict[str, str] = { + "namespace": "contract-jobs", "source_type": "file", "file_name": "contract-read.pdf", "data_id": f"contract-job-read-{uuid4().hex[:12]}", } - if namespace is not None: - payload["namespace"] = namespace response = await api_client.post("/api/v1/jobs", json=payload) @@ -234,6 +229,35 @@ async def test_should_forbid_access_to_a_job_owned_by_another_user( assert "details" not in error +async def _seed_job_in_namespace( + *, + namespace: str | None, + file_name: str = "seeded.pdf", +) -> str: + """Insert a job row directly. + + Job creation is rate limited, so namespace-filter coverage seeds rows + instead of issuing extra POSTs. Passing ``namespace=None`` reproduces a + job written before the namespace key existed in ``job_metadata``. + """ + job_id = f"job_seeded_{uuid4().hex[:12]}" + job_metadata: dict[str, object] = { + "document_id": f"doc_seeded_{uuid4().hex[:12]}", + "source_type": "file", + "original_request": {"file_name": file_name}, + } + if namespace is not None: + job_metadata["namespace"] = namespace + + await ContractDatabase.insert_job( + job_id=job_id, + user_id="local-dev-user", + status="done", + job_metadata=job_metadata, + ) + return job_id + + @pytest.mark.asyncio async def test_should_list_only_jobs_in_the_requested_namespace( developer_api_client_factory: Callable[ @@ -241,8 +265,8 @@ async def test_should_list_only_jobs_in_the_requested_namespace( ], ) -> None: async with developer_api_client_factory() as api_client: - scoped_job = await _create_waiting_file_job(api_client, namespace="tenant-a") - await _create_waiting_file_job(api_client, namespace="tenant-b") + scoped_job_id = await _seed_job_in_namespace(namespace="tenant-a") + await _seed_job_in_namespace(namespace="tenant-b") response = await api_client.get( "/api/v1/jobs", params={"namespace": "tenant-a"} @@ -255,7 +279,7 @@ async def test_should_list_only_jobs_in_the_requested_namespace( assert response_json["total"] == 1 assert len(jobs) == 1 - assert jobs[0]["job_id"] == scoped_job["job_id"] + assert jobs[0]["job_id"] == scoped_job_id assert jobs[0]["namespace"] == "tenant-a" @@ -266,8 +290,8 @@ async def test_should_list_every_namespace_when_no_namespace_filter_is_given( ], ) -> None: async with developer_api_client_factory() as api_client: - await _create_waiting_file_job(api_client, namespace="tenant-a") - await _create_waiting_file_job(api_client, namespace="tenant-b") + await _seed_job_in_namespace(namespace="tenant-a") + await _seed_job_in_namespace(namespace="tenant-b") response = await api_client.get("/api/v1/jobs") @@ -285,8 +309,8 @@ async def test_should_treat_a_blank_namespace_filter_as_the_default_namespace( ], ) -> None: async with developer_api_client_factory() as api_client: - default_job = await _create_waiting_file_job(api_client, namespace=None) - await _create_waiting_file_job(api_client, namespace="tenant-a") + default_job_id = await _seed_job_in_namespace(namespace="default") + await _seed_job_in_namespace(namespace="tenant-a") response = await api_client.get("/api/v1/jobs", params={"namespace": " "}) @@ -296,5 +320,27 @@ async def test_should_treat_a_blank_namespace_filter_as_the_default_namespace( jobs = cast(list[dict[str, object]], response_json["jobs"]) assert response_json["total"] == 1 - assert jobs[0]["job_id"] == default_job["job_id"] - assert jobs[0]["namespace"] == "default" + assert jobs[0]["job_id"] == default_job_id + + +@pytest.mark.asyncio +async def test_should_treat_a_job_without_a_namespace_key_as_the_default_namespace( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + async with developer_api_client_factory() as api_client: + legacy_job_id = await _seed_job_in_namespace(namespace=None) + await _seed_job_in_namespace(namespace="tenant-a") + + response = await api_client.get( + "/api/v1/jobs", params={"namespace": "default"} + ) + + assert response.status_code == 200 + + response_json = cast(dict[str, object], response.json()) + jobs = cast(list[dict[str, object]], response_json["jobs"]) + + assert response_json["total"] == 1 + assert jobs[0]["job_id"] == legacy_job_id diff --git a/packages/shared-python/shared/models/schemas/retrieval_namespace.py b/packages/shared-python/shared/models/schemas/retrieval_namespace.py index bfa54627f..c4ea89467 100644 --- a/packages/shared-python/shared/models/schemas/retrieval_namespace.py +++ b/packages/shared-python/shared/models/schemas/retrieval_namespace.py @@ -2,9 +2,6 @@ DEFAULT_RETRIEVAL_NAMESPACE = "default" -# Backwards-compatible private alias. -_DEFAULT_RETRIEVAL_NAMESPACE = DEFAULT_RETRIEVAL_NAMESPACE - def normalize_retrieval_namespace(namespace: str | None) -> str: """Return the canonical namespace value used by jobs, retrieval, and caches."""