feat(w-12): JobBrief model, endpoints, migration + brief→job linkage

- JobBrief model (DRAFT→SUBMITTED→APPROVED→FULFILLED) with 6 CRUD
  endpoints: list, create, get, patch (DRAFT only), submit, approve
- All endpoints use MembershipContext; read=VIEWER, mutate=MANAGER,
  approve=ADMIN for org-scoped access
- create_job accepts brief_id Form field; validates APPROVED brief,
  copies organization_id/project_id/deadline from brief, marks brief
  FULFILLED after job insert
- organization_id now populated from project client_id on job create
  (fixes missing multi-tenant field on new jobs)
- migration_2026-04-29-000001: job_briefs collection + 4 indexes
- Wired briefs router into main.py

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

backend/app/api/v1/routes_briefs.py

@@ -0,0 +1,201 @@
+"""Job Brief CRUD endpoints."""
+from datetime import datetime
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from motor.motor_asyncio import AsyncIOMotorDatabase
+
+from ...core.authz import MembershipContext, assert_user_in_org, get_membership_context
+from ...core.database import get_database
+from ...core.logging import get_logger
+from ...models.job_brief import (
+    BriefStatus,
+    JobBriefCreate,
+    JobBriefResponse,
+    JobBriefUpdate,
+)
+from ...models.organization import OrgRole
+
+logger = get_logger(__name__)
+router = APIRouter(prefix="/briefs", tags=["briefs"])
+
+
+def _doc_to_response(doc: dict) -> JobBriefResponse:
+    return JobBriefResponse(
+        id=str(doc["_id"]),
+        organization_id=doc["organization_id"],
+        project_id=doc.get("project_id"),
+        title=doc["title"],
+        description=doc.get("description"),
+        requested_outputs=doc["requested_outputs"],
+        languages=doc.get("languages", []),
+        deadline=doc.get("deadline"),
+        status=doc["status"],
+        created_by=doc["created_by"],
+        job_id=doc.get("job_id"),
+        created_at=doc["created_at"].isoformat(),
+        updated_at=doc["updated_at"].isoformat(),
+        submitted_at=doc["submitted_at"].isoformat() if doc.get("submitted_at") else None,
+        approved_by=doc.get("approved_by"),
+    )
+
+
+@router.get("", response_model=list[JobBriefResponse])
+async def list_briefs(
+    ctx: MembershipContext = Depends(get_membership_context),
+    db: AsyncIOMotorDatabase = Depends(get_database),
+):
+    org_ids = [m.organization_id for m in ctx.memberships] if hasattr(ctx, "memberships") else []
+    if ctx.is_platform_admin:
+        query: dict = {}
+    elif org_ids:
+        query = {"organization_id": {"$in": org_ids}}
+    else:
+        raise HTTPException(status_code=403, detail="No org memberships")
+
+    cursor = db.job_briefs.find(query).sort("created_at", -1).limit(100)
+    docs = await cursor.to_list(length=100)
+    return [_doc_to_response(d) for d in docs]
+
+
+@router.post("", response_model=JobBriefResponse, status_code=status.HTTP_201_CREATED)
+async def create_brief(
+    payload: JobBriefCreate,
+    ctx: MembershipContext = Depends(get_membership_context),
+    db: AsyncIOMotorDatabase = Depends(get_database),
+):
+    # Resolve org from project if not directly identifiable
+    org_id: str | None = None
+    if payload.project_id:
+        project = await db.projects.find_one({"_id": payload.project_id}, {"client_id": 1})
+        if project:
+            org_id = project.get("client_id")
+    if not org_id:
+        # Use first membership org if user has only one (or admin)
+        if ctx.is_platform_admin:
+            raise HTTPException(status_code=400, detail="Admin must supply project_id or org_id cannot be inferred")
+        memberships = [m for m in (ctx.memberships if hasattr(ctx, "memberships") else [])
+                       if ctx.can_access_org(m.organization_id, OrgRole.MANAGER)]
+        if len(memberships) == 1:
+            org_id = memberships[0].organization_id
+        else:
+            raise HTTPException(status_code=400, detail="Cannot infer organization; supply project_id")
+
+    assert_user_in_org(ctx, org_id, OrgRole.MANAGER)
+
+    now = datetime.utcnow()
+    doc = {
+        "_id": f"brief_{now.strftime('%Y%m%d%H%M%S%f')}_{str(ctx.user.id)[-6:]}",
+        "organization_id": org_id,
+        "project_id": payload.project_id,
+        "title": payload.title,
+        "description": payload.description,
+        "requested_outputs": payload.requested_outputs.model_dump(),
+        "languages": payload.languages,
+        "deadline": payload.deadline,
+        "status": BriefStatus.DRAFT.value,
+        "created_by": str(ctx.user.id),
+        "job_id": None,
+        "created_at": now,
+        "updated_at": now,
+        "submitted_at": None,
+        "approved_by": None,
+    }
+    await db.job_briefs.insert_one(doc)
+    return _doc_to_response(doc)
+
+
+@router.get("/{brief_id}", response_model=JobBriefResponse)
+async def get_brief(
+    brief_id: str,
+    ctx: MembershipContext = Depends(get_membership_context),
+    db: AsyncIOMotorDatabase = Depends(get_database),
+):
+    doc = await db.job_briefs.find_one({"_id": brief_id})
+    if not doc:
+        raise HTTPException(status_code=404, detail="Brief not found")
+    assert_user_in_org(ctx, doc["organization_id"], OrgRole.VIEWER)
+    return _doc_to_response(doc)
+
+
+@router.patch("/{brief_id}", response_model=JobBriefResponse)
+async def update_brief(
+    brief_id: str,
+    payload: JobBriefUpdate,
+    ctx: MembershipContext = Depends(get_membership_context),
+    db: AsyncIOMotorDatabase = Depends(get_database),
+):
+    doc = await db.job_briefs.find_one({"_id": brief_id})
+    if not doc:
+        raise HTTPException(status_code=404, detail="Brief not found")
+    assert_user_in_org(ctx, doc["organization_id"], OrgRole.MANAGER)
+    if doc["status"] != BriefStatus.DRAFT.value:
+        raise HTTPException(status_code=400, detail="Only DRAFT briefs can be updated")
+
+    updates: dict = {"updated_at": datetime.utcnow()}
+    if payload.title is not None:
+        updates["title"] = payload.title
+    if payload.description is not None:
+        updates["description"] = payload.description
+    if payload.requested_outputs is not None:
+        updates["requested_outputs"] = payload.requested_outputs.model_dump()
+    if payload.languages is not None:
+        updates["languages"] = payload.languages
+    if payload.deadline is not None:
+        updates["deadline"] = payload.deadline
+
+    result = await db.job_briefs.find_one_and_update(
+        {"_id": brief_id},
+        {"$set": updates},
+        return_document=True,
+    )
+    return _doc_to_response(result)
+
+
+@router.post("/{brief_id}/submit", response_model=JobBriefResponse)
+async def submit_brief(
+    brief_id: str,
+    ctx: MembershipContext = Depends(get_membership_context),
+    db: AsyncIOMotorDatabase = Depends(get_database),
+):
+    doc = await db.job_briefs.find_one({"_id": brief_id})
+    if not doc:
+        raise HTTPException(status_code=404, detail="Brief not found")
+    assert_user_in_org(ctx, doc["organization_id"], OrgRole.MANAGER)
+    if doc["status"] != BriefStatus.DRAFT.value:
+        raise HTTPException(status_code=400, detail="Only DRAFT briefs can be submitted")
+
+    now = datetime.utcnow()
+    result = await db.job_briefs.find_one_and_update(
+        {"_id": brief_id},
+        {"$set": {"status": BriefStatus.SUBMITTED.value, "submitted_at": now, "updated_at": now}},
+        return_document=True,
+    )
+    return _doc_to_response(result)
+
+
+@router.post("/{brief_id}/approve", response_model=JobBriefResponse)
+async def approve_brief(
+    brief_id: str,
+    ctx: MembershipContext = Depends(get_membership_context),
+    db: AsyncIOMotorDatabase = Depends(get_database),
+):
+    doc = await db.job_briefs.find_one({"_id": brief_id})
+    if not doc:
+        raise HTTPException(status_code=404, detail="Brief not found")
+    assert_user_in_org(ctx, doc["organization_id"], OrgRole.ADMIN)
+    if doc["status"] != BriefStatus.SUBMITTED.value:
+        raise HTTPException(status_code=400, detail="Only SUBMITTED briefs can be approved")
+
+    now = datetime.utcnow()
+    result = await db.job_briefs.find_one_and_update(
+        {"_id": brief_id},
+        {
+            "$set": {
+                "status": BriefStatus.APPROVED.value,
+                "approved_by": str(ctx.user.id),
+                "updated_at": now,
+            }
+        },
+        return_document=True,
+    )
+    return _doc_to_response(result)

backend/app/api/v1/routes_jobs.py

@@ -87,6 +87,7 @@ async def create_job(
     file: UploadFile = File(...),
     brand_context: str | None = Form(None),
     project_id: str | None = Form(None),
+    brief_id: str | None = Form(None),
     deadline: str | None = Form(None),  # ISO date string e.g. "2026-05-15"
     initial_linguist_id: str | None = Form(None),
     initial_reviewer_id: str | None = Form(None),
@@ -112,6 +113,29 @@ async def create_job(
             detail="Invalid requested_outputs format"
         )
 
+    # Resolve brief if provided — overrides some fields and sets organization_id
+    brief_doc = None
+    organization_id: str | None = None
+    if brief_id:
+        from ...models.job_brief import BriefStatus as _BriefStatus
+        brief_doc = await db.job_briefs.find_one({"_id": brief_id})
+        if not brief_doc:
+            raise HTTPException(status_code=404, detail="Brief not found")
+        if brief_doc["status"] != _BriefStatus.APPROVED.value:
+            raise HTTPException(status_code=400, detail="Brief must be APPROVED before creating a job from it")
+        organization_id = brief_doc["organization_id"]
+        # Brief fields take precedence for pre-fill values
+        if not project_id:
+            project_id = brief_doc.get("project_id")
+        if not deadline:
+            deadline = brief_doc["deadline"].isoformat() if brief_doc.get("deadline") else None
+
+    # Resolve organization_id from project if not yet set
+    if not organization_id and project_id:
+        project = await db.projects.find_one({"_id": project_id}, {"client_id": 1})
+        if project:
+            organization_id = project.get("client_id")
+
     # Generate job ID and upload file
     job_id = str(ObjectId())
     gcs_uri = await upload_file_to_gcs(
@@ -123,6 +147,8 @@ async def create_job(
     job_data = {
         "_id": job_id,
         "client_id": str(current_user.id),
+        "organization_id": organization_id,
+        "brief_id": brief_id or None,
         "title": title,
         "source": {
             "filename": f"{job_id}/source.mp4",
@@ -151,6 +177,14 @@ async def create_job(
 
     await db.jobs.insert_one(job_data)
 
+    # Mark brief as fulfilled if job was created from one
+    if brief_doc:
+        from ...models.job_brief import BriefStatus as _BriefStatus2
+        await db.job_briefs.update_one(
+            {"_id": brief_id},
+            {"$set": {"job_id": job_id, "status": _BriefStatus2.FULFILLED.value, "updated_at": datetime.utcnow()}},
+        )
+
     # Enqueue processing task
     logger.info(f"Dispatching ingest_and_ai_task for job {job_id}")
     logger.info(f"Using Celery app: {celery_app}")

backend/app/main.py

@@ -12,6 +12,7 @@ from sentry_sdk.integrations.redis import RedisIntegration
 
 from .api.v1.routes_admin import router as admin_router
 from .api.v1.routes_admin_production import router as admin_production_router
+from .api.v1.routes_briefs import router as briefs_router
 from .api.v1.routes_auth import router as auth_router
 from .api.v1.routes_clients import router as clients_router
 from .api.v1.routes_files import router as files_router
@@ -268,6 +269,7 @@ app.include_router(glossaries_router, prefix="/api/v1")
 app.include_router(tts_router, prefix="/api/v1")
 app.include_router(admin_router, prefix="/api/v1")
 app.include_router(admin_production_router, prefix="/api/v1")
+app.include_router(briefs_router, prefix="/api/v1")
 app.include_router(share_router, prefix="/api/v1")
 app.include_router(websockets_router, prefix="/api/v1")
 

backend/app/migrations/scripts/migration_2026-04-29-000001_create_job_briefs_collection.py

@@ -0,0 +1,46 @@
+"""Create job_briefs collection with indexes."""
+from ..migrator import Migration
+
+
+class Migration(Migration):
+    version = "2026-04-29-000001"
+    description = "Create job_briefs collection and indexes"
+
+    async def up(self) -> None:
+        db = self.db
+
+        # Ensure collection exists (insert + delete a dummy doc)
+        try:
+            await db.create_collection("job_briefs")
+        except Exception:
+            pass  # already exists
+
+        await db.job_briefs.create_index(
+            [("organization_id", 1), ("status", 1), ("created_at", -1)],
+            name="idx_briefs_org_status_created",
+            background=True,
+        )
+        await db.job_briefs.create_index(
+            [("created_by", 1)],
+            name="idx_briefs_created_by",
+            background=True,
+        )
+        await db.job_briefs.create_index(
+            [("project_id", 1)],
+            name="idx_briefs_project_id",
+            background=True,
+            sparse=True,
+        )
+        await db.job_briefs.create_index(
+            [("job_id", 1)],
+            name="idx_briefs_job_id",
+            background=True,
+            sparse=True,
+        )
+
+    async def down(self) -> None:
+        db = self.db
+        await db.job_briefs.drop_index("idx_briefs_org_status_created")
+        await db.job_briefs.drop_index("idx_briefs_created_by")
+        await db.job_briefs.drop_index("idx_briefs_project_id")
+        await db.job_briefs.drop_index("idx_briefs_job_id")

backend/app/models/job_brief.py

@@ -0,0 +1,73 @@
+"""Job Brief model — pre-approved work order submitted before job creation."""
+from datetime import datetime
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from .job import RequestedOutputs
+
+
+class BriefStatus(str, Enum):
+    DRAFT = "draft"
+    SUBMITTED = "submitted"
+    APPROVED = "approved"
+    REJECTED = "rejected"
+    FULFILLED = "fulfilled"
+
+
+class JobBrief(BaseModel):
+    id: Optional[str] = Field(None, alias="_id")
+    organization_id: str
+    project_id: Optional[str] = None
+    title: str
+    description: Optional[str] = None
+    requested_outputs: RequestedOutputs
+    languages: list[str] = []
+    deadline: Optional[datetime] = None
+    status: BriefStatus = BriefStatus.DRAFT
+    created_by: str
+    job_id: Optional[str] = None
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+    submitted_at: Optional[datetime] = None
+    approved_by: Optional[str] = None
+    reject_reason: Optional[str] = None
+
+    class Config:
+        populate_by_name = True
+
+
+class JobBriefCreate(BaseModel):
+    title: str
+    description: Optional[str] = None
+    requested_outputs: RequestedOutputs
+    languages: list[str] = []
+    deadline: Optional[datetime] = None
+    project_id: Optional[str] = None
+
+
+class JobBriefUpdate(BaseModel):
+    title: Optional[str] = None
+    description: Optional[str] = None
+    requested_outputs: Optional[RequestedOutputs] = None
+    languages: Optional[list[str]] = None
+    deadline: Optional[datetime] = None
+
+
+class JobBriefResponse(BaseModel):
+    id: str
+    organization_id: str
+    project_id: Optional[str] = None
+    title: str
+    description: Optional[str] = None
+    requested_outputs: RequestedOutputs
+    languages: list[str]
+    deadline: Optional[datetime] = None
+    status: BriefStatus
+    created_by: str
+    job_id: Optional[str] = None
+    created_at: str
+    updated_at: str
+    submitted_at: Optional[str] = None
+    approved_by: Optional[str] = None