CUST-WP-0061-T01: intake work-record entity (stage 3)

Fresh hub entity per the founder-reviewed decision (not a suggestions
rename-bridge): kind: intake per canon/standards/work-record-types_v0.1.md,
lifecycle open -> vetted -> routed -> closed(promoted|declined|absorbed).

- api/models/base.py::new_uuid7 -- dependency-free RFC 9562 UUIDv7
  generator (48-bit ms timestamp, version/variant bits, random remainder);
  existing tables keep new_uuid (UUIDv4) unchanged, this is opt-in for new
  work-record entities per the identity-layering canon
- api/models/intake.py: Intake + IntakeNote ORM models, mirroring
  Decision's shape (topic/workplan/repo scope, lane, status, outcome,
  promoted_to back-link); CHECK constraints enforce scope-required,
  closed-requires-outcome, promoted-requires-promoted_to at the DB level
- migrations/a7c3e9f1b4d2: intakes + intake_notes tables, 3 enum types
- api/routers/intake.py: list/create/get/patch + /route + /close + /notes
  actions, mirroring decisions.py's pattern (409 on invalid transitions,
  progress event on close)
- api/schemas/intake.py: Pydantic create/update/route/close/note schemas
- mcp_server/server.py: create_intake, list_intakes, route_intake,
  close_intake tool wrappers
- tests/test_intake.py: 12 tests against the real Postgres test DB
  (create/list/scope-validation, full lifecycle incl. 409s and the
  promoted-requires-promoted_to constraint, notes, UUIDv7 verification)

Verified live against the running dev API + DB (not just pytest): applied
the migration, restarted the MCP server, and ran a full create -> route ->
close cycle over the real REST endpoints. No regressions: full existing
suite (test_routers_core, test_suggestions, test_mcp_smoke,
test_mcp_write_tools, test_mcp_registration, test_consistency_check,
test_consistency_sweep) all green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

api/main.py

@@ -12,7 +12,7 @@ from starlette.responses import Response as StarletteResponse
 from api.database import engine
 from api.events import shutdown_publisher
 from api.services.write_idempotency import WriteIdempotencyMiddleware
-from api.routers import decisions, extension_points, progress, state, suggestions, tasks, technical_debt, topics, workstreams, workstream_dependencies
+from api.routers import decisions, extension_points, intake, progress, state, suggestions, tasks, technical_debt, topics, workstreams, workstream_dependencies
 from api.routers import domains, repos, contributions, sbom, policy, domain_goals, repo_goals, messages, capability_requests, tpsc, services
 from api.routers import token_events
 from api.routers import interface_changes
@@ -114,6 +114,7 @@ app.include_router(workstream_dependencies.router)
 app.include_router(workstream_dependencies.workplan_router)
 app.include_router(tasks.router)
 app.include_router(decisions.router)
+app.include_router(intake.router)
 app.include_router(extension_points.router)
 app.include_router(technical_debt.router)
 app.include_router(progress.router)

api/models/__init__.py

@@ -10,6 +10,13 @@ from api.models.workstream import Workstream
 from api.models.workstream_dependency import WorkstreamDependency
 from api.models.task import Task, TaskStatus, TaskPriority
 from api.models.decision import Decision, DecisionType, DecisionStatus
+from api.models.intake import (
+    Intake,
+    IntakeNote,
+    IntakeLane,
+    IntakeStatus,
+    IntakeOutcome,
+)
 from api.models.progress_event import ProgressEvent
 from api.models.extension_point import ExtensionPoint, EPStatus
 from api.models.technical_debt import TechnicalDebt, TDStatus
@@ -54,6 +61,7 @@ __all__ = [
     "WorkstreamDependency",
     "Task", "TaskStatus", "TaskPriority",
     "Decision", "DecisionType", "DecisionStatus",
+    "Intake", "IntakeNote", "IntakeLane", "IntakeStatus", "IntakeOutcome",
     "ProgressEvent",
     "ExtensionPoint", "EPStatus",
     "TechnicalDebt", "TDStatus",

api/models/base.py

@@ -1,3 +1,5 @@
+import os
+import time
 import uuid
 from datetime import datetime
 
@@ -24,3 +26,28 @@ class TimestampMixin:
 
 def new_uuid() -> uuid.UUID:
     return uuid.uuid4()
+
+
+def new_uuid7() -> uuid.UUID:
+    """Generate a UUIDv7 (RFC 9562): 48-bit big-endian ms timestamp, version
+    and variant bits, remaining bits random. Time-sortable, so primary keys
+    generated with this helper order chronologically without a separate
+    created_at index lookup — the identity layering canon
+    (work-record-types_v0.1.md) calls this out as the primary internal key
+    for new work-record entities.
+
+    Dependency-free (no uuid7 in stdlib before Python 3.14, no third-party
+    lib added for a ~15-line, non-cryptographic layout).
+    """
+    unix_ts_ms = int(time.time() * 1000)
+    rand = int.from_bytes(os.urandom(10), "big")
+    rand_a = (rand >> 62) & 0x0FFF          # top 12 bits of the 80 random bits
+    rand_b = rand & 0x3FFFFFFFFFFFFFFF       # bottom 62 bits
+    value = (
+        (unix_ts_ms << 80)
+        | (0x7 << 76)       # version 7
+        | (rand_a << 64)
+        | (0x2 << 62)        # variant 10
+        | rand_b
+    )
+    return uuid.UUID(int=value)

api/models/intake.py

@@ -0,0 +1,126 @@
+import enum
+import uuid
+from datetime import datetime
+
+from sqlalchemy import CheckConstraint, DateTime, Enum, ForeignKey, String, Text
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.sql import func
+
+from api.models.base import Base, TimestampMixin, new_uuid7
+
+
+class IntakeLane(str, enum.Enum):
+    green = "green"
+    blue = "blue"
+    yellow = "yellow"
+    orange = "orange"
+    red = "red"
+
+
+class IntakeStatus(str, enum.Enum):
+    open = "open"
+    vetted = "vetted"
+    routed = "routed"
+    closed = "closed"
+
+
+class IntakeOutcome(str, enum.Enum):
+    promoted = "promoted"
+    declined = "declined"
+    absorbed = "absorbed"
+
+
+OPEN_INTAKE_STATUSES = (IntakeStatus.open, IntakeStatus.vetted, IntakeStatus.routed)
+
+
+class Intake(Base, TimestampMixin):
+    """A `kind: intake` work record — a spark: idea, finding, directive, or
+    request, per canon/standards/work-record-types_v0.1.md. Lifecycle:
+    open -> vetted -> routed -> closed(promoted|declined|absorbed).
+
+    Fresh entity per the founder-reviewed architecture draft (2026-07-20,
+    WorkOrchestrationArchitectureDraft.md §8 item 6): not a rename/reuse of
+    the legacy `suggestions` table.
+    """
+
+    __tablename__ = "intakes"
+    __table_args__ = (
+        CheckConstraint(
+            "topic_id IS NOT NULL OR workplan_id IS NOT NULL OR repo_id IS NOT NULL",
+            name="ck_intakes_topic_or_workplan_or_repo",
+        ),
+        CheckConstraint(
+            "(status != 'closed') OR (outcome IS NOT NULL)",
+            name="ck_intakes_closed_requires_outcome",
+        ),
+        CheckConstraint(
+            "(outcome != 'promoted') OR (promoted_to IS NOT NULL)",
+            name="ck_intakes_promoted_requires_promoted_to",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=new_uuid7
+    )
+    topic_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("topics.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+    workplan_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("workplans.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+    repo_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("managed_repos.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+    title: Mapped[str] = mapped_column(String(500), nullable=False)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    lane: Mapped[IntakeLane] = mapped_column(
+        Enum(IntakeLane, name="intakelane"), nullable=False, default=IntakeLane.green
+    )
+    status: Mapped[IntakeStatus] = mapped_column(
+        Enum(IntakeStatus, name="intakestatus"),
+        nullable=False,
+        default=IntakeStatus.open,
+        index=True,
+    )
+    outcome: Mapped[IntakeOutcome | None] = mapped_column(
+        Enum(IntakeOutcome, name="intakeoutcome"), nullable=True
+    )
+    origin: Mapped[str | None] = mapped_column(String(200), nullable=True)
+    origin_ref: Mapped[str | None] = mapped_column(String(200), nullable=True, index=True)
+    promoted_to: Mapped[str | None] = mapped_column(String(200), nullable=True)
+    source_repo_path: Mapped[str | None] = mapped_column(
+        Text, nullable=True, doc="Repo-relative path of the source file this record was authored in."
+    )
+    routed_note: Mapped[str | None] = mapped_column(Text, nullable=True)
+    closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    topic: Mapped["Topic | None"] = relationship("Topic", lazy="selectin")  # noqa: F821
+    workplan: Mapped["Workplan | None"] = relationship("Workplan", lazy="selectin")  # noqa: F821
+    repo: Mapped["ManagedRepo | None"] = relationship("ManagedRepo", lazy="selectin")  # noqa: F821
+    notes: Mapped[list["IntakeNote"]] = relationship(
+        "IntakeNote",
+        back_populates="intake",
+        lazy="selectin",
+        order_by="IntakeNote.created_at",
+        cascade="all, delete-orphan",
+    )
+
+
+class IntakeNote(Base):
+    __tablename__ = "intake_notes"
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=new_uuid7)
+    intake_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("intakes.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    author: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    content: Mapped[str] = mapped_column(Text, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+
+    intake: Mapped["Intake"] = relationship("Intake", back_populates="notes")

api/routers/intake.py

@@ -0,0 +1,173 @@
+import uuid
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session
+from api.models.intake import Intake, IntakeNote, IntakeOutcome, IntakeStatus
+from api.models.progress_event import ProgressEvent
+from api.schemas.intake import (
+    IntakeClose,
+    IntakeCreate,
+    IntakeNoteCreate,
+    IntakeRead,
+    IntakeRoute,
+    IntakeUpdate,
+)
+
+router = APIRouter(prefix="/intakes", tags=["intakes"])
+
+_ALLOWED_ROUTE_FROM = {IntakeStatus.open, IntakeStatus.vetted}
+_ALLOWED_CLOSE_FROM = {IntakeStatus.open, IntakeStatus.vetted, IntakeStatus.routed}
+
+
+def _reject_status(intake: Intake, allowed: set[IntakeStatus], action: str) -> None:
+    if intake.status not in allowed:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=(
+                f"Cannot {action} intake in status '{intake.status.value}'; "
+                f"allowed from: {sorted(s.value for s in allowed)}"
+            ),
+        )
+
+
+@router.get("/", response_model=list[IntakeRead])
+async def list_intakes(
+    topic_id: uuid.UUID | None = None,
+    workplan_id: uuid.UUID | None = None,
+    repo_id: uuid.UUID | None = None,
+    status_: IntakeStatus | None = None,
+    session: AsyncSession = Depends(get_session),
+) -> list[Intake]:
+    q = select(Intake)
+    if topic_id:
+        q = q.where(Intake.topic_id == topic_id)
+    if workplan_id:
+        q = q.where(Intake.workplan_id == workplan_id)
+    if repo_id:
+        q = q.where(Intake.repo_id == repo_id)
+    if status_:
+        q = q.where(Intake.status == status_)
+    q = q.order_by(Intake.created_at)
+    result = await session.execute(q)
+    return list(result.scalars().all())
+
+
+@router.post("/", response_model=IntakeRead, status_code=status.HTTP_201_CREATED)
+async def create_intake(
+    body: IntakeCreate,
+    session: AsyncSession = Depends(get_session),
+) -> Intake:
+    intake = Intake(**body.model_dump())
+    session.add(intake)
+    await session.commit()
+    await session.refresh(intake)
+    return intake
+
+
+@router.get("/{intake_id}", response_model=IntakeRead)
+async def get_intake(
+    intake_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> Intake:
+    intake = await session.get(Intake, intake_id)
+    if intake is None:
+        raise HTTPException(status_code=404, detail="Intake not found")
+    return intake
+
+
+@router.patch("/{intake_id}", response_model=IntakeRead)
+async def update_intake(
+    intake_id: uuid.UUID,
+    body: IntakeUpdate,
+    session: AsyncSession = Depends(get_session),
+) -> Intake:
+    intake = await session.get(Intake, intake_id)
+    if intake is None:
+        raise HTTPException(status_code=404, detail="Intake not found")
+    for field, value in body.model_dump(exclude_unset=True).items():
+        setattr(intake, field, value)
+    await session.commit()
+    await session.refresh(intake)
+    return intake
+
+
+@router.post("/{intake_id}/route", response_model=IntakeRead)
+async def route_intake(
+    intake_id: uuid.UUID,
+    body: IntakeRoute,
+    session: AsyncSession = Depends(get_session),
+) -> Intake:
+    """Move an intake into `routed` — eligible for the promotion transition."""
+    intake = await session.get(Intake, intake_id)
+    if intake is None:
+        raise HTTPException(status_code=404, detail="Intake not found")
+    _reject_status(intake, _ALLOWED_ROUTE_FROM, "route")
+
+    intake.status = IntakeStatus.routed
+    if body.routed_note:
+        intake.routed_note = body.routed_note
+    await session.commit()
+    await session.refresh(intake)
+    return intake
+
+
+@router.post("/{intake_id}/close", response_model=IntakeRead)
+async def close_intake(
+    intake_id: uuid.UUID,
+    body: IntakeClose,
+    session: AsyncSession = Depends(get_session),
+) -> Intake:
+    """Close an intake with an outcome. `outcome=promoted` requires
+    `promoted_to` (the canonical id of the record it became) — this is
+    normally called by the promotion transition (CUST-WP-0061-T03), not by
+    hand, but a manual close (declined/absorbed, or a promotion recorded
+    after the fact) is supported directly."""
+    intake = await session.get(Intake, intake_id)
+    if intake is None:
+        raise HTTPException(status_code=404, detail="Intake not found")
+    _reject_status(intake, _ALLOWED_CLOSE_FROM, "close")
+
+    intake.status = IntakeStatus.closed
+    intake.outcome = body.outcome
+    intake.closed_at = datetime.now(tz=timezone.utc)
+    if body.promoted_to:
+        intake.promoted_to = body.promoted_to
+    await session.commit()
+    await session.refresh(intake)
+
+    event = ProgressEvent(
+        topic_id=intake.topic_id,
+        workplan_id=intake.workplan_id,
+        event_type="intake_closed",
+        summary=f"Intake closed ({body.outcome.value}): {intake.title}",
+        detail={
+            "intake_id": str(intake.id),
+            "outcome": body.outcome.value,
+            "promoted_to": body.promoted_to,
+            "note": body.note,
+        },
+    )
+    session.add(event)
+    await session.commit()
+
+    return intake
+
+
+@router.post("/{intake_id}/notes", response_model=IntakeRead, status_code=status.HTTP_201_CREATED)
+async def add_intake_note(
+    intake_id: uuid.UUID,
+    body: IntakeNoteCreate,
+    session: AsyncSession = Depends(get_session),
+) -> Intake:
+    intake = await session.get(Intake, intake_id)
+    if intake is None:
+        raise HTTPException(status_code=404, detail="Intake not found")
+    note = IntakeNote(intake_id=intake.id, author=body.author, content=body.content)
+    session.add(note)
+    await session.commit()
+    await session.refresh(intake)
+    return intake

api/schemas/intake.py

@@ -0,0 +1,88 @@
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, model_validator
+
+from api.models.intake import IntakeLane, IntakeOutcome, IntakeStatus
+
+
+class IntakeCreate(BaseModel):
+    topic_id: uuid.UUID | None = None
+    workplan_id: uuid.UUID | None = None
+    repo_id: uuid.UUID | None = None
+    title: str
+    description: str | None = None
+    lane: IntakeLane = IntakeLane.green
+    origin: str | None = None
+    origin_ref: str | None = None
+    source_repo_path: str | None = None
+
+    @model_validator(mode="after")
+    def scope_required(self) -> "IntakeCreate":
+        if self.topic_id is None and self.workplan_id is None and self.repo_id is None:
+            raise ValueError("At least one of topic_id, workplan_id, or repo_id must be set")
+        return self
+
+
+class IntakeUpdate(BaseModel):
+    title: str | None = None
+    description: str | None = None
+    lane: IntakeLane | None = None
+    status: IntakeStatus | None = None
+    origin: str | None = None
+    origin_ref: str | None = None
+    routed_note: str | None = None
+
+
+class IntakeRoute(BaseModel):
+    """Move an intake from open/vetted into routed — the state that makes
+    it eligible for the promotion transition (CUST-WP-0061-T03)."""
+
+    routed_note: str | None = None
+
+
+class IntakeClose(BaseModel):
+    outcome: IntakeOutcome
+    promoted_to: str | None = None
+    note: str | None = None
+
+    @model_validator(mode="after")
+    def promoted_requires_target(self) -> "IntakeClose":
+        if self.outcome == IntakeOutcome.promoted and not self.promoted_to:
+            raise ValueError("outcome=promoted requires promoted_to")
+        return self
+
+
+class IntakeNoteCreate(BaseModel):
+    content: str
+    author: str | None = None
+
+
+class IntakeNoteRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    author: str | None = None
+    content: str
+    created_at: datetime
+
+
+class IntakeRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    topic_id: uuid.UUID | None = None
+    workplan_id: uuid.UUID | None = None
+    repo_id: uuid.UUID | None = None
+    title: str
+    description: str | None = None
+    lane: IntakeLane
+    status: IntakeStatus
+    outcome: IntakeOutcome | None = None
+    origin: str | None = None
+    origin_ref: str | None = None
+    promoted_to: str | None = None
+    source_repo_path: str | None = None
+    routed_note: str | None = None
+    closed_at: datetime | None = None
+    created_at: datetime
+    updated_at: datetime
+    notes: list[IntakeNoteRead] = []

mcp_server/server.py

@@ -1013,6 +1013,114 @@ def resolve_decision(
     return _json_result(decision)
 
 
+@mcp.tool()
+def create_intake(
+    title: str,
+    topic_id: str | None = None,
+    workplan_id: str | None = None,
+    repo_id: str | None = None,
+    description: str | None = None,
+    lane: str = "green",
+    origin: str | None = None,
+    origin_ref: str | None = None,
+    source_repo_path: str | None = None,
+) -> str:
+    """Record an intake item — a spark: idea, finding, directive, or
+    request (work-record kind `intake`, canon/standards/
+    work-record-types_v0.1.md). Lifecycle: open -> vetted -> routed ->
+    closed(promoted|declined|absorbed).
+
+    Args:
+        title: short description of the intake item
+        topic_id: optional topic UUID (at least one of topic_id/workplan_id/repo_id required)
+        workplan_id: optional workplan UUID to scope this to
+        repo_id: optional managed-repo UUID to scope this to
+        description: optional longer context
+        lane: autonomy lane — green | blue | yellow | orange | red
+        origin: free-text source of the finding (e.g. "mail-triage", "founder-directive")
+        origin_ref: stable reference into the origin (e.g. a mail-log row id)
+        source_repo_path: repo-relative path of the file this was authored in, if known
+    """
+    intake = _post("/intakes", {
+        "title": title,
+        "topic_id": topic_id,
+        "workplan_id": workplan_id,
+        "repo_id": repo_id,
+        "description": description,
+        "lane": lane,
+        "origin": origin,
+        "origin_ref": origin_ref,
+        "source_repo_path": source_repo_path,
+    })
+    if error := _response_error("create_intake", intake, ("id",)):
+        return _json_result(error)
+    return _json_result(intake)
+
+
+@mcp.tool()
+def list_intakes(
+    topic_id: str | None = None,
+    workplan_id: str | None = None,
+    repo_id: str | None = None,
+    status: str | None = None,
+) -> str:
+    """List intake items, optionally filtered by scope and/or status.
+
+    Args:
+        topic_id: optional topic UUID
+        workplan_id: optional workplan UUID
+        repo_id: optional managed-repo UUID
+        status: open | vetted | routed | closed
+    """
+    return _json_result(_get("/intakes", {
+        "topic_id": topic_id,
+        "workplan_id": workplan_id,
+        "repo_id": repo_id,
+        "status_": status,
+    }))
+
+
+@mcp.tool()
+def route_intake(intake_id: str, routed_note: str | None = None) -> str:
+    """Move an intake from open/vetted into routed — the state that makes
+    it eligible for promotion into a workplan, task, decision, or
+    engagement.
+
+    Args:
+        intake_id: UUID of the intake item
+        routed_note: optional note on where/how it should be routed
+    """
+    result = _post(f"/intakes/{intake_id}/route", {"routed_note": routed_note})
+    if error := _response_error("route_intake", result, ("id",)):
+        return _json_result(error)
+    return _json_result(result)
+
+
+@mcp.tool()
+def close_intake(
+    intake_id: str,
+    outcome: str,
+    promoted_to: str | None = None,
+    note: str | None = None,
+) -> str:
+    """Close an intake item with an outcome.
+
+    Args:
+        intake_id: UUID of the intake item
+        outcome: promoted | declined | absorbed
+        promoted_to: canonical id of the record it became (required if outcome=promoted)
+        note: optional closing note
+    """
+    result = _post(f"/intakes/{intake_id}/close", {
+        "outcome": outcome,
+        "promoted_to": promoted_to,
+        "note": note,
+    })
+    if error := _response_error("close_intake", result, ("id",)):
+        return _json_result(error)
+    return _json_result(result)
+
+
 @mcp.tool()
 def add_progress_event(
     summary: str,

migrations/versions/a7c3e9f1b4d2_intake_work_record_entity.py

@@ -0,0 +1,100 @@
+"""add intakes + intake_notes tables (CUST-WP-0061-T01, work-record stage 3)
+
+Revision ID: a7c3e9f1b4d2
+Revises: f0a1b2c3d4e5
+Create Date: 2026-07-21
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.dialects.postgresql import UUID
+
+revision = "a7c3e9f1b4d2"
+down_revision = "f0a1b2c3d4e5"
+branch_labels = None
+depends_on = None
+
+intakelane = postgresql.ENUM(
+    "green", "blue", "yellow", "orange", "red",
+    name="intakelane",
+    create_type=False,
+)
+intakestatus = postgresql.ENUM(
+    "open", "vetted", "routed", "closed",
+    name="intakestatus",
+    create_type=False,
+)
+intakeoutcome = postgresql.ENUM(
+    "promoted", "declined", "absorbed",
+    name="intakeoutcome",
+    create_type=False,
+)
+
+
+def upgrade() -> None:
+    intakelane.create(op.get_bind(), checkfirst=True)
+    intakestatus.create(op.get_bind(), checkfirst=True)
+    intakeoutcome.create(op.get_bind(), checkfirst=True)
+
+    op.create_table(
+        "intakes",
+        sa.Column("id", UUID(as_uuid=True), primary_key=True),
+        sa.Column("topic_id", UUID(as_uuid=True), sa.ForeignKey("topics.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("workplan_id", UUID(as_uuid=True), sa.ForeignKey("workplans.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("repo_id", UUID(as_uuid=True), sa.ForeignKey("managed_repos.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("title", sa.String(length=500), nullable=False),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("lane", intakelane, nullable=False, server_default="green"),
+        sa.Column("status", intakestatus, nullable=False, server_default="open"),
+        sa.Column("outcome", intakeoutcome, nullable=True),
+        sa.Column("origin", sa.String(length=200), nullable=True),
+        sa.Column("origin_ref", sa.String(length=200), nullable=True),
+        sa.Column("promoted_to", sa.String(length=200), nullable=True),
+        sa.Column("source_repo_path", sa.Text(), nullable=True),
+        sa.Column("routed_note", sa.Text(), nullable=True),
+        sa.Column("closed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.CheckConstraint(
+            "topic_id IS NOT NULL OR workplan_id IS NOT NULL OR repo_id IS NOT NULL",
+            name="ck_intakes_topic_or_workplan_or_repo",
+        ),
+        sa.CheckConstraint(
+            "(status != 'closed') OR (outcome IS NOT NULL)",
+            name="ck_intakes_closed_requires_outcome",
+        ),
+        sa.CheckConstraint(
+            "(outcome != 'promoted') OR (promoted_to IS NOT NULL)",
+            name="ck_intakes_promoted_requires_promoted_to",
+        ),
+    )
+    op.create_index("ix_intakes_topic_id", "intakes", ["topic_id"])
+    op.create_index("ix_intakes_workplan_id", "intakes", ["workplan_id"])
+    op.create_index("ix_intakes_repo_id", "intakes", ["repo_id"])
+    op.create_index("ix_intakes_status", "intakes", ["status"])
+    op.create_index("ix_intakes_origin_ref", "intakes", ["origin_ref"])
+
+    op.create_table(
+        "intake_notes",
+        sa.Column("id", UUID(as_uuid=True), primary_key=True),
+        sa.Column("intake_id", UUID(as_uuid=True), sa.ForeignKey("intakes.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("author", sa.String(length=100), nullable=True),
+        sa.Column("content", sa.Text(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+    )
+    op.create_index("ix_intake_notes_intake_id", "intake_notes", ["intake_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_intake_notes_intake_id", table_name="intake_notes")
+    op.drop_table("intake_notes")
+    op.drop_index("ix_intakes_origin_ref", table_name="intakes")
+    op.drop_index("ix_intakes_status", table_name="intakes")
+    op.drop_index("ix_intakes_repo_id", table_name="intakes")
+    op.drop_index("ix_intakes_workplan_id", table_name="intakes")
+    op.drop_index("ix_intakes_topic_id", table_name="intakes")
+    op.drop_table("intakes")
+    intakeoutcome.drop(op.get_bind(), checkfirst=True)
+    intakestatus.drop(op.get_bind(), checkfirst=True)
+    intakelane.drop(op.get_bind(), checkfirst=True)

tests/test_intake.py

@@ -0,0 +1,157 @@
+"""Tests for the `intake` work-record entity (CUST-WP-0061-T01, work-record
+stage 3). Real PostgreSQL test database, no mocking — matches
+tests/test_routers_core.py conventions.
+"""
+from __future__ import annotations
+
+import pytest
+
+
+async def _create_domain(client, slug="testdomain", name="Test Domain"):
+    r = await client.post("/domains/", json={"slug": slug, "name": name})
+    assert r.status_code == 201, r.text
+    return r.json()
+
+
+async def _create_topic(client, domain_slug="testdomain", slug="testtopic", title="Test Topic"):
+    r = await client.post("/topics/", json={
+        "slug": slug, "title": title, "domain": domain_slug,
+    })
+    assert r.status_code == 201, r.text
+    return r.json()
+
+
+async def _create_intake(client, topic_id=None, workplan_id=None, repo_id=None,
+                          title="Qonto MCP mailing", lane="green", **extra):
+    payload = {"title": title, "lane": lane, **extra}
+    if topic_id is not None:
+        payload["topic_id"] = topic_id
+    if workplan_id is not None:
+        payload["workplan_id"] = workplan_id
+    if repo_id is not None:
+        payload["repo_id"] = repo_id
+    r = await client.post("/intakes/", json=payload)
+    assert r.status_code == 201, r.text
+    return r.json()
+
+
+class TestIntakeCreateAndRead:
+    async def test_create_requires_a_scope(self, client):
+        r = await client.post("/intakes/", json={"title": "orphan intake", "lane": "green"})
+        assert r.status_code == 422
+
+    async def test_create_with_topic_scope(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        body = await _create_intake(client, topic_id=topic["id"])
+        assert body["status"] == "open"
+        assert body["lane"] == "green"
+        assert body["outcome"] is None
+
+    async def test_get_unknown_intake_404s(self, client):
+        r = await client.get("/intakes/00000000-0000-0000-0000-000000000000")
+        assert r.status_code == 404
+
+    async def test_list_filters_by_topic(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        other_topic = await _create_topic(client, slug="other-topic", title="Other")
+        await _create_intake(client, topic_id=topic["id"], title="in scope")
+        await _create_intake(client, topic_id=other_topic["id"], title="out of scope")
+
+        r = await client.get("/intakes/", params={"topic_id": topic["id"]})
+        assert r.status_code == 200
+        titles = [i["title"] for i in r.json()]
+        assert titles == ["in scope"]
+
+
+class TestIntakeLifecycle:
+    async def test_route_moves_open_to_routed(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        intake = await _create_intake(client, topic_id=topic["id"])
+
+        r = await client.post(f"/intakes/{intake['id']}/route", json={"routed_note": "green lane, ready"})
+        assert r.status_code == 200
+        assert r.json()["status"] == "routed"
+        assert r.json()["routed_note"] == "green lane, ready"
+
+    async def test_route_from_closed_is_rejected(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        intake = await _create_intake(client, topic_id=topic["id"])
+        await client.post(f"/intakes/{intake['id']}/close", json={"outcome": "declined"})
+
+        r = await client.post(f"/intakes/{intake['id']}/route", json={})
+        assert r.status_code == 409
+
+    async def test_close_declined_does_not_require_promoted_to(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        intake = await _create_intake(client, topic_id=topic["id"])
+
+        r = await client.post(f"/intakes/{intake['id']}/close", json={"outcome": "declined"})
+        assert r.status_code == 200
+        body = r.json()
+        assert body["status"] == "closed"
+        assert body["outcome"] == "declined"
+        assert body["closed_at"] is not None
+
+    async def test_close_promoted_without_promoted_to_is_rejected(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        intake = await _create_intake(client, topic_id=topic["id"])
+
+        r = await client.post(f"/intakes/{intake['id']}/close", json={"outcome": "promoted"})
+        assert r.status_code == 422
+
+    async def test_close_promoted_with_promoted_to_succeeds(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        intake = await _create_intake(client, topic_id=topic["id"], title="AWQ-010")
+
+        r = await client.post(
+            f"/intakes/{intake['id']}/close",
+            json={"outcome": "promoted", "promoted_to": "BINKY-WP-0005"},
+        )
+        assert r.status_code == 200
+        body = r.json()
+        assert body["outcome"] == "promoted"
+        assert body["promoted_to"] == "BINKY-WP-0005"
+
+    async def test_close_already_closed_is_rejected(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        intake = await _create_intake(client, topic_id=topic["id"])
+        await client.post(f"/intakes/{intake['id']}/close", json={"outcome": "declined"})
+
+        r = await client.post(f"/intakes/{intake['id']}/close", json={"outcome": "absorbed"})
+        assert r.status_code == 409
+
+
+class TestIntakeNotes:
+    async def test_add_note_appears_on_intake(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        intake = await _create_intake(client, topic_id=topic["id"])
+
+        r = await client.post(
+            f"/intakes/{intake['id']}/notes",
+            json={"content": "vetted, looks real", "author": "agt-rhythm-bridge"},
+        )
+        assert r.status_code == 201
+        notes = r.json()["notes"]
+        assert len(notes) == 1
+        assert notes[0]["content"] == "vetted, looks real"
+        assert notes[0]["author"] == "agt-rhythm-bridge"
+
+
+class TestIntakeUUIDv7:
+    async def test_id_is_uuidv7(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        intake = await _create_intake(client, topic_id=topic["id"])
+
+        import uuid
+        u = uuid.UUID(intake["id"])
+        assert u.version == 7