feat(jobs): filter job listing by namespace and add job deletion

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")

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,

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,

apps/api/app/repositories/job_repository.py

@@ -4,14 +4,29 @@
 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
 
 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,

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",
 ]