feat(P02): complete application-state-machine phase (REQ-11-002)

Phase 2 complete: the employer per-target transition routes (G-311
gate order), the learner withdraw (D-043 stack unchanged, G-323
apply-terminal guard), the operator lane (fail-closed, G-327 422/409
split, the R-80 re-hire with G-326 stamping), the G-319 pin split,
33 route tests + the security-review pins. Verify four-layers PASS.
Ships as v0.10.2.

---ci---
phase: 2
milestone: v0.11
status: complete
requirements:
  covered: [REQ-11-002]
  partial: []
---/ci---

.ciagent/CHECKPOINT.json

@@ -1,8 +1,8 @@
 {
-  "phase": 1,
-  "stage": "complete",
+  "phase": 2,
+  "stage": "verify",
   "milestone": "v0.11",
   "phase_role": "execution",
   "attempts": 0,
-  "updated_at": "2026-10-01T05:35:00Z"
+  "updated_at": "2026-10-01T07:10:00Z"
 }

apps/ai-service/ai_service/api/employers.py

@@ -38,6 +38,7 @@ import secrets
 import time
 import uuid
 from collections import deque
+from datetime import UTC, date, datetime
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, Header, HTTPException, Request
@@ -71,13 +72,29 @@ from ..llm.base import LLMProvider
 from ..marketplace.base import Posting
 from ..marketplace.enrichment import maybe_enrich_on_publish
 from ..marketplace.search import embed_posting
-from ..marketplace.store import InvalidTransition, MarketplaceStore
+from ..marketplace.store import (
+    InvalidTransition,
+    MarketplaceStore,
+)
 from ..marketplace.tagging import TaggingUnavailable, tag_posting
+from ..placements.base import (
+    APPLICATION_STATUSES,
+    ApplicationTransitionResponse,
+    EmployerApplicationsResponse,
+    EmployerApplicationSummary,
+    HireTransitionRequest,
+    PlacementRead,
+)
+from ..placements.base import (
+    InvalidTransition as PlacementInvalidTransition,
+)
+from ..placements.store import PlacementRow, PlacementStore
 from .deps import (
     get_billing_store,
     get_embedding_provider,
     get_employer_store,
     get_marketplace_store,
+    get_placement_store,
     get_provider,
     get_settings,
 )
@@ -715,6 +732,220 @@ async def close_own_posting(
     )
 
 
+# -- hiring pipeline (v0.11 P2, Task 2-1-01 — REQ-11-002, D-105/G-311) ---------------
+
+
+def _own_application_or_404(
+    store: MarketplaceStore, application_id: str, employer_id: str
+) -> None:
+    """Fetch the application → its posting → enforce the cross-entity
+    arm (D-105: the _own_posting_or_404 shape applied THROUGH the
+    application's posting — another entity's application → 403 with
+    NO application data; unknown → 404). G-319: learner_id never
+    rides this arm's error responses."""
+    application = store.get_application(application_id)
+    if application is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"no application with id {application_id!r}",
+        )
+    posting = store.get_posting(application.posting_id)
+    if posting is None:  # pragma: no cover - FK discipline keeps rows paired
+        raise HTTPException(
+            status_code=404,
+            detail=f"no posting with id {application.posting_id!r}",
+        )
+    if posting.employer_id != employer_id:
+        raise HTTPException(
+            status_code=403,
+            detail="this application belongs to another employer entity",
+        )
+
+
+def _placement_read(row: PlacementRow) -> PlacementRead:
+    return PlacementRead(
+        id=row.id,
+        learner_id=row.learner_id,
+        employer_id=row.employer_id,
+        posting_id=row.posting_id,
+        application_id=row.application_id,
+        role_title=row.role_title,
+        start_date=date.fromisoformat(row.start_date),
+        status=row.status,  # type: ignore[arg-type]
+        created_at=row.created_at or datetime.now(UTC),
+        decided_at=row.decided_at or datetime.now(UTC),
+    )
+
+
+def _employer_transition(
+    application_id: str,
+    to_status: str,
+    account: AccountRow,
+    marketplace: MarketplaceStore,
+    placements: PlacementStore,
+    employer_store: EmployerStore,
+    role_title: str | None = None,
+    start_date: date | None = None,
+) -> ApplicationTransitionResponse:
+    """The shared per-target transition composition (D-105): G-311
+    gate order — session (the route dep) → claim (_require_claimed) →
+    cross-entity (_own_application_or_404) → the ONE store transition
+    (409 on matrix-illegal, NEVER before the 403s) + the
+    employer_audit_events append (the D-089 one-queryable lane)."""
+    employer_id = _require_claimed(account)
+    _own_application_or_404(marketplace, application_id, employer_id)
+    try:
+        application, placement = placements.transition_application(
+            application_id,
+            to_status,
+            actor_type="employer",
+            actor_ref=f"account:{account.id}",
+            role_title=role_title,
+            start_date=start_date,
+        )
+    except PlacementInvalidTransition as exc:  # MUST precede ValueError — it subclasses it
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    except ValueError as exc:  # hire required-fields honesty (D-103)
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    employer_store.append_audit_event(
+        "application",
+        f"application.{to_status}",
+        account_id=account.id,
+        employer_id=employer_id,
+        detail={"application_id": application_id, "to_status": to_status},
+    )
+    return ApplicationTransitionResponse(
+        application_id=application.id,
+        status=application.status,  # type: ignore[arg-type]
+        placement=_placement_read(placement) if placement is not None else None,
+    )
+
+
+@claims_router.get("/applications", response_model=EmployerApplicationsResponse)
+async def list_own_applications(
+    status: str | None = None,
+    auth: Annotated[
+        tuple[SessionRow, AccountRow], Depends(require_employer_session)
+    ] = None,  # type: ignore[assignment]
+    store: MarketplaceStore = Depends(get_marketplace_store),
+) -> EmployerApplicationsResponse:
+    """The session employer's OWN hiring pipeline (D-105): every
+    application on the claimed entity's postings (ALL statuses), with
+    per-status facets. G-319 pin split: this APPLICATION-scoped read
+    exposes learner_id per the supersession — the own-postings LIST
+    stays counts-only."""
+    if auth is None:  # pragma: no cover - the dep raises 401 first
+        raise HTTPException(status_code=401, detail="employer session required")
+    _session, account = auth
+    employer_id = _require_claimed(account)
+    postings = store.list_postings_by_employer(employer_id)
+    applications = store.list_applications_by_postings(
+        [p.id for p in postings], status=status
+    )
+    facets: dict[str, int] = {s: 0 for s in APPLICATION_STATUSES}
+    items = [
+        EmployerApplicationSummary(
+            id=a.id,
+            posting_id=a.posting_id,
+            learner_id=a.learner_id,
+            status=a.status,  # type: ignore[arg-type]
+            created_at=a.created_at,
+        )
+        for a in applications
+    ]
+    for item in items:
+        facets[item.status] = facets.get(item.status, 0) + 1
+    return EmployerApplicationsResponse(
+        applications=items, facets=facets, total=len(items)
+    )
+
+
+def _own_transition_route(
+    to_status: str, hire: bool = False
+):
+    """Per-target route factory (D-105: the matrix stays OUT of request
+    bodies — each route's store call names its target, the publish/
+    close idiom). Only the hire route carries a body ({role_title,
+    start_date} — D-103/G-324); the others are bodyless actions."""
+
+    if hire:
+
+        async def _hire_route(
+            application_id: str,
+            auth: Annotated[
+                tuple[SessionRow, AccountRow], Depends(require_employer_session)
+            ],
+            body: HireTransitionRequest,
+            store: MarketplaceStore = Depends(get_marketplace_store),
+            placements: PlacementStore = Depends(get_placement_store),
+            employer_store: EmployerStore = Depends(get_employer_store),
+        ) -> ApplicationTransitionResponse:
+            return _employer_transition(
+                application_id,
+                to_status,
+                auth[1],
+                store,
+                placements,
+                employer_store,
+                role_title=body.role_title,
+                start_date=body.start_date,
+            )
+
+        return _hire_route
+
+    async def _action_route(
+        application_id: str,
+        auth: Annotated[
+            tuple[SessionRow, AccountRow], Depends(require_employer_session)
+        ],
+        store: MarketplaceStore = Depends(get_marketplace_store),
+        placements: PlacementStore = Depends(get_placement_store),
+        employer_store: EmployerStore = Depends(get_employer_store),
+    ) -> ApplicationTransitionResponse:
+        return _employer_transition(
+            application_id,
+            to_status,
+            auth[1],
+            store,
+            placements,
+            employer_store,
+        )
+
+    return _action_route
+
+
+claims_router.add_api_route(
+    "/applications/{application_id}/in-review",
+    _own_transition_route("in_review"),
+    methods=["POST"],
+    response_model=ApplicationTransitionResponse,
+)
+claims_router.add_api_route(
+    "/applications/{application_id}/interview",
+    _own_transition_route("interview"),
+    methods=["POST"],
+    response_model=ApplicationTransitionResponse,
+)
+claims_router.add_api_route(
+    "/applications/{application_id}/offer",
+    _own_transition_route("offer"),
+    methods=["POST"],
+    response_model=ApplicationTransitionResponse,
+)
+claims_router.add_api_route(
+    "/applications/{application_id}/reject",
+    _own_transition_route("rejected"),
+    methods=["POST"],
+    response_model=ApplicationTransitionResponse,
+)
+claims_router.add_api_route(
+    "/applications/{application_id}/hire",
+    _own_transition_route("hired", hire=True),
+    methods=["POST"],
+    response_model=ApplicationTransitionResponse,
+)
+
+
 # -- self-serve billing (P5, Task 5-1-01 — REQ-10-005, R-59/G-310) ------------------
 
 

apps/ai-service/ai_service/api/marketplace.py

@@ -33,6 +33,7 @@ from ..marketplace.resume import (
     parse_resume,
 )
 from ..marketplace.search import FacetFilters, rank_postings
+from ..placements.base import InvalidTransition
 from .deps import get_embedding_provider, get_marketplace_store, get_provider, get_settings
 from .identity import require_verified_age
 
@@ -375,6 +376,78 @@ async def apply_posting(
     application, _created = store.insert_application(
         posting_id=body.posting_id, learner_id=body.learner_id
     )
+    # G-323 (binding): the idempotent arm NEVER confirms a terminal
+    # application — a withdrawn/rejected/hired row returns 409 (the
+    # learner is told the honest state; recovery is visibility via the
+    # application-status surface, not resurrection — no second row is
+    # ever created; the (posting_id, learner_id) unique key stands).
+    if application.status in ("withdrawn", "rejected", "hired"):
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "reason": "application_terminal",
+                "status": application.status,
+            },
+        )
+    return ApplyResponse(
+        application_id=application.id,
+        posting_id=application.posting_id,
+        status=application.status,
+    )
+
+
+# -- v0.11 P2: the learner withdraw route (D-104, REQ-11-002) --------------------
+
+
+class WithdrawBody(BaseModel):
+    """D-104: the pilot surface passes learner_id in the body (the G-5
+    composition unchanged — the SAME stack as apply)."""
+
+    learner_id: str = Field(min_length=1)
+
+
+@router.post(
+    "/applications/{application_id}/withdraw", response_model=ApplyResponse
+)
+async def withdraw_application(
+    application_id: str,
+    body: WithdrawBody,
+    request: Request,
+    store=Depends(get_marketplace_store),
+    settings: Settings = Depends(get_settings),
+) -> ApplyResponse:
+    """The learner's ONLY edit (D-099: withdrawn is learner- or
+    operator-triggerable, NEVER an employer session). The D-043
+    composition UNCHANGED: G-5 allowlist → verified 18+ → the EXISTING
+    apply-rate window (no new throttle knob, D-088 scope untouched).
+    Own application only (403 foreign); terminal → 409 (the matrix).
+    """
+    _enforce_allowlist(body.learner_id, settings)
+    await require_verified_age(18, body.learner_id, _identity_store(request))
+    _check_apply_rate(body.learner_id)
+
+    application = store.get_application(application_id)
+    if application is None:
+        raise HTTPException(
+            status_code=404, detail=f"no application with id {application_id!r}"
+        )
+    if application.learner_id != body.learner_id:
+        raise HTTPException(
+            status_code=403,
+            detail="this application belongs to another learner",
+        )
+    from .deps import get_placement_store  # local import — the territory rule
+
+    placements = get_placement_store(request)
+    try:
+        application, _placement = placements.transition_application(
+            application_id,
+            "withdrawn",
+            actor_type="learner",
+            actor_ref=f"learner:{body.learner_id}",
+        )
+    except InvalidTransition as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
     return ApplyResponse(
         application_id=application.id,
         posting_id=application.posting_id,

apps/ai-service/ai_service/api/placements_admin.py

@@ -0,0 +1,187 @@
+"""Operator lane for placement oversight (D-106, REQ-11-002).
+
+`require_admin` IMPORTED from marketplace_admin (the billing_admin.py
+import precedent — token namespaces cannot collide, D-090): one
+generic remediation transition route + revoke + the R-80 re-hire +
+the reporting/export reads (P3/P4 extend this router). No employer
+session, no learner gate — a third principal class (D-097).
+
+OPERATOR = "operator" route-level provenance (employers_admin.py:54 —
+no per-operator accounts exist, D-094); every action is
+ledger-recorded with actor_type="operator" (the D-083 manual-grant
+lane) and answers in employer_audit_events via the P2 routes' own
+appends (operator transitions record through the placement ledger).
+"""
+
+from __future__ import annotations
+
+from datetime import date
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from ..config import Settings
+from ..marketplace.store import MarketplaceStore
+from ..placements.base import (
+    AdminTransitionRequest,
+    ApplicationTransitionResponse,
+    HireTransitionRequest,
+    PlacementRead,
+)
+from ..placements.store import PlacementRow, PlacementStore
+from .deps import get_marketplace_store, get_placement_store
+from .marketplace_admin import require_admin
+
+#: D-094: route-level operator provenance — no per-operator accounts.
+OPERATOR = "operator"
+
+router = APIRouter(prefix="/v1/placements/admin", tags=["placements-admin"])
+# The admin-token gate composes on EVERY route handler via the
+# Annotated[Settings, Depends(require_admin)] house pattern (the
+# employers_admin idiom — fail-closed; token namespaces cannot collide
+# with employer session tokens, D-090).
+AdminGate = Annotated[Settings, Depends(require_admin)]
+
+
+def _placement_read(row: PlacementRow) -> PlacementRead:
+    return PlacementRead(
+        id=row.id,
+        learner_id=row.learner_id,
+        employer_id=row.employer_id,
+        posting_id=row.posting_id,
+        application_id=row.application_id,
+        role_title=row.role_title,
+        start_date=date.fromisoformat(row.start_date),
+        status=row.status,  # type: ignore[arg-type]
+        created_at=row.created_at or _now(),
+        decided_at=row.decided_at or _now(),
+    )
+
+
+def _now():
+    from datetime import UTC, datetime
+
+    return datetime.now(UTC)
+
+
+@router.post(
+    "/applications/{application_id}/transition",
+    response_model=ApplicationTransitionResponse,
+)
+async def admin_transition_application(
+    application_id: str,
+    body: AdminTransitionRequest,
+    _gate: AdminGate,
+    store: MarketplaceStore = Depends(get_marketplace_store),
+    placements: PlacementStore = Depends(get_placement_store),
+) -> ApplicationTransitionResponse:
+    """The generic operator remediation route (D-106): ANY matrix-legal
+    transition incl. withdrawn on a learner's behalf (remediation with
+    provenance, the D-083 lane). G-327: `to` is a CLOSED Literal —
+    unknown values die at pydantic (422), matrix-illegal known values
+    reach the store (409 InvalidTransition). role_title/start_date
+    are required iff to=hired (the honest 422 with required-fields
+    detail, D-103)."""
+    role_title: str | None = None
+    start_date: date | None = None
+    if body.to == "hired":
+        if body.role_title is None or body.start_date is None:
+            raise HTTPException(
+                status_code=422,
+                detail="role_title and start_date are required when to=hired (D-103)",
+            )
+        role_title, start_date = body.role_title, body.start_date
+    try:
+        application, placement = placements.transition_application(
+            application_id,
+            body.to,
+            actor_type="operator",
+            actor_ref=OPERATOR,
+            role_title=role_title,
+            start_date=start_date,
+        )
+    except Exception as exc:
+        # The placements InvalidTransition (409) MUST be matched before
+        # the ValueError arm — InvalidTransition subclasses ValueError.
+        from ..placements.base import InvalidTransition
+
+        if isinstance(exc, InvalidTransition):
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        if isinstance(exc, ValueError):
+            raise HTTPException(status_code=422, detail=str(exc)) from exc
+        raise
+    return ApplicationTransitionResponse(
+        application_id=application.id,
+        status=application.status,  # type: ignore[arg-type]
+        placement=_placement_read(placement) if placement is not None else None,
+    )
+
+
+@router.post("/placements/{placement_id}/revoke", response_model=PlacementRead)
+async def admin_revoke_placement(
+    placement_id: str,
+    _gate: AdminGate,
+    placements: PlacementStore = Depends(get_placement_store),
+) -> PlacementRead:
+    """D-102: the operator correction — the one active → revoked
+    placement transition (operator-only by route). Unknown → 404;
+    non-active → 409 (the matrix)."""
+    try:
+        revoked = placements.revoke_placement(placement_id, OPERATOR)
+    except Exception as exc:
+        from ..placements.base import InvalidTransition
+
+        if isinstance(exc, InvalidTransition):
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        raise
+    if revoked is None:
+        raise HTTPException(
+            status_code=404, detail=f"no placement with id {placement_id!r}"
+        )
+    return _placement_read(revoked)
+
+
+@router.post(
+    "/placements/by-application/{application_id}/re-hire",
+    response_model=PlacementRead,
+)
+async def admin_rehire_placement(
+    application_id: str,
+    body: HireTransitionRequest,
+    _gate: AdminGate,
+    placements: PlacementStore = Depends(get_placement_store),
+) -> PlacementRead:
+    """THE R-80 RE-HIRE ROUTE: after the first hire the application is
+    TERMINAL (hired → hired is a 409 on every matrix-validated
+    surface — the coherence gap research found). The re-hire creates
+    the FRESH PlacementRow + the placement.rehired ledger event
+    WITHOUT touching the application's terminal status — the ledger
+    reads hired → revoke → rehire as three events, never fabricating
+    a second application transition. Structurally safe: revoked rows
+    never match the partial WHERE status='active' (the G-316
+    property). G-326: decided_at = the re-hire event's stamp."""
+    try:
+        rehired = placements.rehire_placement(
+            application_id,
+            role_title=body.role_title,
+            start_date=body.start_date,
+            actor_ref=OPERATOR,
+        )
+    except Exception as exc:
+        from sqlalchemy.exc import IntegrityError
+
+        from ..placements.base import InvalidTransition
+
+        if isinstance(exc, InvalidTransition):
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        if isinstance(exc, ValueError):
+            raise HTTPException(status_code=422, detail=str(exc)) from exc
+        if isinstance(exc, IntegrityError):
+            # the partial-unique arm: an ACTIVE placement already exists
+            raise HTTPException(
+                status_code=409,
+                detail="an active placement already exists for this "
+                "application — revoke it before re-hiring",
+            ) from exc
+        raise
+    return _placement_read(rehired)

apps/ai-service/ai_service/employers/store.py

@@ -64,7 +64,9 @@ _TABLES = (
 #: Audit event vocabulary: `auth` = the Wave 1-2 register/logout
 #: account events; `claim` = P2's claim/decide entity events; `manual`
 #: = operator action provenance (the billing kind set, employer-side).
-_AUDIT_EVENT_KINDS = ("auth", "claim", "manual")
+#: v0.11 P2: "application" joins the vocabulary (employer-driven hiring
+#: transitions append their audit row — the D-089 one-queryable lane).
+_AUDIT_EVENT_KINDS = ("auth", "claim", "manual", "application")
 
 
 class AccountRow(SQLModel, table=True):

apps/ai-service/ai_service/main.py

@@ -423,6 +423,14 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     # accounts/audit operator routes — require_admin (fail-closed).
     app.include_router(employers_admin_router)
 
+    # v0.11 placements (REQ-11-002, D-106): the operator lane — the
+    # generic transition + revoke + the R-80 re-hire; the reporting/
+    # export reads extend this router in P3/P4. require_admin
+    # composed at the router level (fail-closed, D-090 no-collision).
+    from .api.placements_admin import router as placements_admin_router
+
+    app.include_router(placements_admin_router)
+
     # v0.6 P5 (REQ-6-006, D-056): dynamic sitemap/robots — request-time
     # enumeration (build-time Next sitemap.ts cannot see runtime
     # postings under output:'export'); API routes beat the static mount.

apps/ai-service/ai_service/marketplace/store.py

@@ -353,6 +353,16 @@ class MarketplaceStore(Protocol):
         shape)."""
         ...
 
+    def list_applications_by_postings(
+        self, posting_ids: list[str], status: str | None = None
+    ) -> list[Application]:
+        """v0.11 P2 (D-105/G-319): the employer pipeline's ONE batched
+        read — every application on the given postings (the caller
+        scopes to the session's OWN claimed-entity postings first;
+        this read never enumerates beyond them), created_at DESC then
+        id; optional status facet."""
+        ...
+
     def list_applications(
         self, posting_id: str | None = None
     ) -> list[Application]:
@@ -916,6 +926,27 @@ class SQLiteMarketplaceStore:
             session.refresh(row)
             return row.to_application()
 
+    def list_applications_by_postings(
+        self, posting_ids: list[str], status: str | None = None
+    ) -> list[Application]:
+        """v0.11 P2 (D-105/G-319): ONE batched read over the given
+        postings — created_at DESC then id; optional status facet.
+        Empty posting set → empty list (the honest no-postings case)."""
+        if not posting_ids:
+            return []
+        with Session(self._engine) as session:
+            stmt = select(ApplicationRow).where(
+                col(ApplicationRow.posting_id).in_(posting_ids)
+            )
+            if status is not None:
+                stmt = stmt.where(ApplicationRow.status == status)
+            rows = session.exec(
+                stmt.order_by(
+                    col(ApplicationRow.created_at).desc(), ApplicationRow.id
+                )
+            ).all()
+            return [r.to_application() for r in rows]
+
     def list_applications(
         self, posting_id: str | None = None
     ) -> list[Application]:

apps/ai-service/ai_service/placements/base.py

@@ -182,3 +182,60 @@ class PlacementEventRead(BaseModel):
     prev_hash: str = Field(min_length=1)
     hash: str = Field(min_length=1)
     created_at: datetime
+
+
+class EmployerApplicationSummary(BaseModel):
+    """An employer-facing application row (D-105/G-319): the pipeline
+    list + per-application reads. learner_id IS visible here — the
+    G-319 supersession scopes it to applications on the session's OWN
+    claimed-entity postings (the cross-entity arm 403s carry no data;
+    the own-postings LIST stays counts-only — the pin split)."""
+
+    model_config = ConfigDict(frozen=True)
+
+    id: str = Field(min_length=1)
+    posting_id: str = Field(min_length=1)
+    learner_id: str = Field(min_length=1)
+    status: ApplicationStatus
+    created_at: datetime | None = None
+
+
+class EmployerApplicationsResponse(BaseModel):
+    """GET /v1/employers/applications response: own-pipeline
+    applications (every status) + per-status facet counts (every
+    status key ALWAYS present — a zero-count status carries its
+    honest 0)."""
+
+    model_config = ConfigDict(frozen=True)
+
+    applications: list[EmployerApplicationSummary] = Field(default_factory=list)
+    facets: dict[str, int] = Field(default_factory=dict)
+    total: int = 0
+
+
+class ApplicationTransitionResponse(BaseModel):
+    """The transition result (per-target routes): the application's
+    new status + the placement record when the transition created
+    one (hire)."""
+
+    model_config = ConfigDict(frozen=True)
+
+    application_id: str = Field(min_length=1)
+    status: ApplicationStatus
+    placement: PlacementRead | None = None
+
+
+class AdminTransitionRequest(BaseModel):
+    """POST /v1/placements/admin/applications/{id}/transition body
+    (D-106/G-327): `to` is a CLOSED Literal of the seven D-099
+    statuses — unknown values die at pydantic (422), matrix-illegal
+    known values reach the store (409 InvalidTransition); the route
+    never sees a status outside the vocabulary. role_title/start_date
+    are required iff to=hired (enforced at the route, the honest 422
+    with required-fields detail)."""
+
+    model_config = ConfigDict(frozen=True)
+
+    to: ApplicationStatus
+    role_title: str | None = Field(default=None, min_length=1, max_length=200)
+    start_date: date | None = None

apps/ai-service/tests/placements/test_admin_lane.py

@@ -0,0 +1,247 @@
+"""The operator lane tests (D-106/G-326/G-327, REQ-11-002) — the
+fail-closed matrix, the generic transition arms, revoke, and the
+R-80 re-hire route."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+from ai_service.marketplace.base import Posting
+from ai_service.marketplace.store import SQLiteMarketplaceStore
+
+from .conftest import ADMIN_TOKEN, make_client
+
+
+def _seed_posting(store: SQLiteMarketplaceStore, employer_id: str = "emp-test") -> str:
+    import uuid
+
+    posting = store.insert_posting(
+        Posting(
+            id=f"post-{uuid.uuid4().hex[:12]}",
+            source="admin",
+            external_id=f"test-{uuid.uuid4().hex[:8]}",
+            employer_id=employer_id,
+            title="AI Orchestration Engineer",
+        )
+    )
+    # the learner apply route requires published (409 on drafts)
+    store.set_posting_status(posting.id, "published")
+    return posting.id
+
+
+def _apply(client: TestClient, posting_id: str, learner_id: str = "pilot-learner") -> str:
+    resp = client.post(
+        "/v1/marketplace/apply", json={"posting_id": posting_id, "learner_id": learner_id}
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["application_id"]
+
+
+def _admin(token: str = ADMIN_TOKEN) -> dict[str, str]:
+    return {"X-Admin-Token": token}
+
+
+@pytest.fixture()
+def scaffold(tmp_path, identity_store, marketplace_store):
+    client = make_client(tmp_path, identity_store, marketplace_store)
+    with client as entered:
+        posting_id = _seed_posting(marketplace_store)
+        application_id = _apply(entered, posting_id)
+        yield {
+            "client": entered,
+            "posting_id": posting_id,
+            "application_id": application_id,
+        }
+
+
+# -- the fail-closed matrix (the R-56 shape) -------------------------------------
+
+
+def test_fail_closed_no_token(scaffold) -> None:
+    resp = scaffold["client"].post(
+        f"/v1/placements/admin/applications/{scaffold['application_id']}/transition",
+        json={"to": "in_review"},
+    )
+    assert resp.status_code == 403
+
+
+def test_fail_closed_bad_token(scaffold) -> None:
+    resp = scaffold["client"].post(
+        f"/v1/placements/admin/applications/{scaffold['application_id']}/transition",
+        json={"to": "in_review"},
+        headers=_admin("wrong-token"),
+    )
+    assert resp.status_code == 403
+
+
+def test_fail_closed_revoke_and_rehire(scaffold) -> None:
+    client = scaffold["client"]
+    # no token → 403 on every route (fail-closed)
+    resp = client.post(
+        "/v1/placements/admin/placements/plc-x/revoke",
+    )
+    assert resp.status_code == 403
+    # invalid token → 403 (never a 404/409 — the gate runs first)
+    resp = client.post(
+        "/v1/placements/admin/placements/plc-x/revoke",
+        headers=_admin("wrong-token"),
+    )
+    assert resp.status_code == 403
+    resp = client.post(
+        f"/v1/placements/admin/placements/by-application/{scaffold['application_id']}/re-hire",
+        json={"role_title": "Engineer", "start_date": "2026-12-01"},
+        headers=_admin("wrong-token"),
+    )
+    assert resp.status_code == 403
+
+
+# -- the generic transition route --------------------------------------------------
+
+
+def test_admin_transition_every_legal_arm(scaffold) -> None:
+    client, app_id = scaffold["client"], scaffold["application_id"]
+    for to_status in ("in_review", "interview", "offer", "rejected"):
+        resp = client.post(
+            f"/v1/placements/admin/applications/{app_id}/transition",
+            json={"to": to_status},
+            headers=_admin(),
+        )
+        # rejected is terminal — the loop ends there
+        assert resp.status_code == 200, resp.text
+        assert resp.json()["status"] == to_status
+        if to_status == "rejected":
+            break
+
+
+def test_admin_garbage_to_422(scaffold) -> None:
+    """G-327: `to` is a CLOSED Literal — garbage dies at pydantic
+    (422), never a 500/409."""
+    resp = scaffold["client"].post(
+        f"/v1/placements/admin/applications/{scaffold['application_id']}/transition",
+        json={"to": "garbage"},
+        headers=_admin(),
+    )
+    assert resp.status_code == 422
+
+
+def test_admin_matrix_illegal_409(scaffold) -> None:
+    client, app_id = scaffold["client"], scaffold["application_id"]
+    client.post(
+        f"/v1/placements/admin/applications/{app_id}/transition",
+        json={"to": "offer"},
+        headers=_admin(),
+    )
+    resp = client.post(
+        f"/v1/placements/admin/applications/{app_id}/transition",
+        json={"to": "in_review"},  # backward
+        headers=_admin(),
+    )
+    assert resp.status_code == 409
+
+
+def test_admin_hire_requires_fields_422(scaffold) -> None:
+    resp = scaffold["client"].post(
+        f"/v1/placements/admin/applications/{scaffold['application_id']}/transition",
+        json={"to": "hired"},
+        headers=_admin(),
+    )
+    assert resp.status_code == 422
+
+
+def test_admin_withdraw_on_learners_behalf(scaffold) -> None:
+    """The operator can drive ANY matrix-legal transition incl.
+    withdrawn (remediation with provenance, the D-083 lane)."""
+    resp = scaffold["client"].post(
+        f"/v1/placements/admin/applications/{scaffold['application_id']}/transition",
+        json={"to": "withdrawn"},
+        headers=_admin(),
+    )
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "withdrawn"
+
+
+def _admin_hire(scaffold) -> str:
+    """The shared hire-through-the-operator-lane helper → placement id."""
+    resp = scaffold["client"].post(
+        f"/v1/placements/admin/applications/{scaffold['application_id']}/transition",
+        json={"to": "hired", "role_title": "Engineer", "start_date": "2026-11-01"},
+        headers=_admin(),
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["status"] == "hired"
+    assert body["placement"]["role_title"] == "Engineer"
+    assert body["placement"]["status"] == "active"
+    return body["placement"]["id"]
+
+
+def test_admin_hire_creates_placement(scaffold) -> None:
+    _admin_hire(scaffold)
+
+
+# -- revoke + the R-80 re-hire ------------------------------------------------------
+
+
+def test_revoke_then_rehire_three_event_ledger(scaffold) -> None:
+    client, app_id = scaffold["client"], scaffold["application_id"]
+    placement_id = _admin_hire(scaffold)
+
+    # revoke
+    resp = client.post(
+        f"/v1/placements/admin/placements/{placement_id}/revoke",
+        headers=_admin(),
+    )
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["status"] == "revoked"
+
+    # the revoked placement is out of the active set but in history
+    resp = client.get("/v1/placements/admin/overview", headers=_admin())
+    assert resp.status_code == 404  # P3 ships the overview — 404 until then
+
+    # re-hire: the FRESH active row (R-80) + decided_at = the re-hire
+    # event's stamp (G-326 — asserted at the store layer in test_store)
+    resp = client.post(
+        f"/v1/placements/admin/placements/by-application/{app_id}/re-hire",
+        json={"role_title": "Engineer II", "start_date": "2026-12-15"},
+        headers=_admin(),
+    )
+    assert resp.status_code == 200, resp.text
+    rehired = resp.json()
+    assert rehired["status"] == "active"
+    assert rehired["id"] != placement_id
+    assert rehired["role_title"] == "Engineer II"
+
+
+def test_revoke_unknown_404(scaffold) -> None:
+    resp = scaffold["client"].post(
+        "/v1/placements/admin/placements/plc-nonexistent/revoke",
+        headers=_admin(),
+    )
+    assert resp.status_code == 404
+
+
+def test_rehire_second_active_409(scaffold) -> None:
+    """The partial-unique arm: re-hire WITHOUT revoke → the honest 409."""
+    client, app_id = scaffold["client"], scaffold["application_id"]
+    client.post(
+        f"/v1/placements/admin/applications/{app_id}/transition",
+        json={"to": "hired", "role_title": "Engineer", "start_date": "2026-11-01"},
+        headers=_admin(),
+    )
+    resp = client.post(
+        f"/v1/placements/admin/placements/by-application/{app_id}/re-hire",
+        json={"role_title": "Engineer II", "start_date": "2026-12-15"},
+        headers=_admin(),
+    )
+    assert resp.status_code == 409
+    assert "revoke it before re-hiring" in resp.json()["detail"]
+
+
+def test_rehire_requires_hired_application_409(scaffold) -> None:
+    resp = scaffold["client"].post(
+        f"/v1/placements/admin/placements/by-application/{scaffold['application_id']}/re-hire",
+        json={"role_title": "Engineer", "start_date": "2026-11-01"},
+        headers=_admin(),
+    )
+    assert resp.status_code == 409  # not hired yet — the correction lane refuses

apps/ai-service/tests/placements/test_employer_transitions.py

@@ -0,0 +1,298 @@
+"""Employer application transition tests (D-105, REQ-11-002) — the
+G-311 gate-order pins + the D-099 matrix 409 arms + the G-319 pin
+split + the hire arms."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+from ai_service.employers.store import SQLiteEmployerStore
+from ai_service.marketplace.base import Posting
+from ai_service.marketplace.store import SQLiteMarketplaceStore
+
+from .conftest import make_client
+
+# -- fixtures: the employer-principal + application scaffold -------------------
+
+
+def _seed_posting(store: SQLiteMarketplaceStore, employer_id: str) -> str:
+    import uuid
+
+    # the claim route requires the EmployerRow entity to exist (404 unknown)
+    from ai_service.marketplace.base import Employer
+
+    store.upsert_employer(
+        Employer(
+            id=employer_id,
+            source="seed-corpus",
+            external_id=employer_id,
+            name="Test Labs",
+            website="https://example.com/test",
+        )
+    )
+    posting = store.insert_posting(
+        Posting(
+            id=f"post-{uuid.uuid4().hex[:12]}",
+            source="admin",
+            external_id=f"test-{uuid.uuid4().hex[:8]}",
+            employer_id=employer_id,
+            title="AI Orchestration Engineer",
+            description="Test posting",
+        )
+    )
+    # the learner apply route requires published (409 on drafts)
+    store.set_posting_status(posting.id, "published")
+    return posting.id
+
+
+def _apply(client: TestClient, posting_id: str, learner_id: str = "pilot-learner") -> str:
+    resp = client.post(
+        "/v1/marketplace/apply", json={"posting_id": posting_id, "learner_id": learner_id}
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["application_id"]
+
+
+def _register_and_claim(
+    client: TestClient, employer_store: SQLiteEmployerStore, email: str
+) -> str:
+    """Register + claim + admin-approve → (token, claimed employer_id)."""
+    resp = client.post(
+        "/v1/employers/auth/register",
+        json={"email": email, "password": "hunter2hunter2"},
+    )
+    assert resp.status_code == 201, resp.text
+    token = resp.json()["token"]
+    resp = client.post(
+        "/v1/employers/claims",
+        json={"employer_id": "emp-test"},
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert resp.status_code in (200, 201), resp.text
+    claim_id = resp.json()["id"]
+    resp = client.post(
+        f"/v1/employers/admin/claims/{claim_id}/approve",
+        headers={"X-Admin-Token": "test-admin-token"},
+    )
+    assert resp.status_code == 200, resp.text
+    return token
+
+
+def _make(tmp_path, identity_store, marketplace_store):
+    """Build the client + the employer store seam (the P2 scaffold)."""
+    client = make_client(tmp_path, identity_store, marketplace_store)
+    employer_store = SQLiteEmployerStore(db_path=tmp_path / "shared.db")
+    client.app.state.employer_store = (  # type: ignore[attr-defined]  # TestClient.app: ASGIApp-typed, runtime FastAPI
+        employer_store
+    )
+    return client, employer_store
+
+
+@pytest.fixture()
+def scaffold(tmp_path, identity_store, marketplace_store):
+    client, employer_store = _make(tmp_path, identity_store, marketplace_store)
+    with client as entered:
+        posting_id = _seed_posting(marketplace_store, "emp-test")
+        token = _register_and_claim(entered, employer_store, "hiring@example.com")
+        application_id = _apply(entered, posting_id)
+        yield {
+            "client": entered,
+            "token": token,
+            "posting_id": posting_id,
+            "application_id": application_id,
+        }
+
+
+def _hdr(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+# -- G-311 gate order (never a 409 before a 403; never a 403 before the 401) -----
+
+
+def test_gate_order_no_token_401(scaffold) -> None:
+    resp = scaffold["client"].post(
+        f"/v1/employers/applications/{scaffold['application_id']}/in-review"
+    )
+    assert resp.status_code == 401
+
+
+def test_gate_order_unclaimed_403(tmp_path, identity_store, marketplace_store) -> None:
+    client, _ = _make(tmp_path, identity_store, marketplace_store)
+    with client as entered:
+        posting_id = _seed_posting(marketplace_store, "emp-test")
+        resp = entered.post(
+            "/v1/employers/auth/register",
+            json={"email": "unclaimed@example.com", "password": "hunter2hunter2"},
+        )
+        token = resp.json()["token"]
+        application_id = _apply(entered, posting_id)
+        resp = entered.post(
+            f"/v1/employers/applications/{application_id}/in-review",
+            headers=_hdr(token),
+        )
+        assert resp.status_code == 403
+        assert "no_claimed_employer" in str(resp.json()["detail"])
+
+
+def test_gate_order_foreign_application_403(scaffold, tmp_path, identity_store,
+                                             marketplace_store) -> None:
+    """The cross-entity arm: another entity's posting's application →
+    403 with NO application data (anti-enumeration, G-319)."""
+    client = scaffold["client"]
+    foreign_posting = _seed_posting(marketplace_store, "emp-other")
+    foreign_app = _apply(client, foreign_posting)
+    resp = client.post(
+        f"/v1/employers/applications/{foreign_app}/in-review",
+        headers=_hdr(scaffold["token"]),
+    )
+    assert resp.status_code == 403
+    assert resp.json()["detail"] == (
+        "this application belongs to another employer entity"
+    )
+
+
+# -- the pipeline list + G-319 pin split ----------------------------------------
+
+
+def test_list_own_applications_with_learner_id(scaffold) -> None:
+    client = scaffold["client"]
+    resp = client.get("/v1/employers/applications", headers=_hdr(scaffold["token"]))
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["total"] == 1
+    app_row = body["applications"][0]
+    # G-319 supersession: learner_id IS visible on the application-scoped read
+    assert app_row["learner_id"] == "pilot-learner"
+    assert app_row["status"] == "submitted"
+    assert body["facets"]["submitted"] == 1
+    assert body["facets"]["hired"] == 0
+
+
+def test_own_postings_list_stays_counts_only(scaffold) -> None:
+    """The G-319 pin split: the own-postings LIST never carries learner
+    identifiers (the v0.10 boundary holds there — unchanged)."""
+    client = scaffold["client"]
+    resp = client.get("/v1/employers/postings", headers=_hdr(scaffold["token"]))
+    assert resp.status_code == 200
+    assert "pilot-learner" not in resp.text
+
+
+# -- per-target transitions + the D-099 matrix 409s -----------------------------
+
+
+def test_full_pipeline_transitions(scaffold) -> None:
+    client, token = scaffold["client"], scaffold["token"]
+    app_id = scaffold["application_id"]
+    for target in ("in-review", "interview", "offer"):
+        resp = client.post(
+            f"/v1/employers/applications/{app_id}/{target}", headers=_hdr(token)
+        )
+        assert resp.status_code == 200, resp.text
+        assert resp.json()["status"] == target.replace("-", "_")
+    resp = client.post(
+        f"/v1/employers/applications/{app_id}/hire",
+        json={"role_title": "AI Orchestration Engineer", "start_date": "2026-11-01"},
+        headers=_hdr(token),
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["status"] == "hired"
+    assert body["placement"]["role_title"] == "AI Orchestration Engineer"
+    assert body["placement"]["start_date"] == "2026-11-01"
+    assert body["placement"]["employer_id"] == "emp-test"
+
+
+def test_skip_forward_hire_from_submitted(scaffold) -> None:
+    """R-71: submitted → hired is a legal fast-track."""
+    client, token = scaffold["client"], scaffold["token"]
+    resp = client.post(
+        f"/v1/employers/applications/{scaffold['application_id']}/hire",
+        json={"role_title": "Engineer", "start_date": "2026-12-01"},
+        headers=_hdr(token),
+    )
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "hired"
+
+
+def test_matrix_backward_move_409(scaffold) -> None:
+    client, token = scaffold["client"], scaffold["token"]
+    app_id = scaffold["application_id"]
+    client.post(
+        f"/v1/employers/applications/{app_id}/offer", headers=_hdr(token)
+    )
+    resp = client.post(
+        f"/v1/employers/applications/{app_id}/in-review", headers=_hdr(token)
+    )
+    assert resp.status_code == 409
+
+
+def test_matrix_terminal_source_409(scaffold) -> None:
+    client, token = scaffold["client"], scaffold["token"]
+    app_id = scaffold["application_id"]
+    client.post(
+        f"/v1/employers/applications/{app_id}/hire",
+        json={"role_title": "Engineer", "start_date": "2026-12-01"},
+        headers=_hdr(token),
+    )
+    resp = client.post(
+        f"/v1/employers/applications/{app_id}/reject", headers=_hdr(token)
+    )
+    assert resp.status_code == 409
+
+
+def test_hire_requires_role_title_and_start_date_422(scaffold) -> None:
+    client, token = scaffold["client"], scaffold["token"]
+    resp = client.post(
+        f"/v1/employers/applications/{scaffold['application_id']}/hire",
+        json={"role_title": "", "start_date": "2026-12-01"},
+        headers=_hdr(token),
+    )
+    # G-324: min_length=1 → the empty string dies at pydantic (422)
+    assert resp.status_code == 422
+    resp = client.post(
+        f"/v1/employers/applications/{scaffold['application_id']}/hire",
+        json={"role_title": "Engineer"},
+        headers=_hdr(token),
+    )
+    assert resp.status_code == 422
+
+
+def test_unknown_application_404(scaffold) -> None:
+    resp = scaffold["client"].post(
+        "/v1/employers/applications/app-nonexistent/in-review",
+        headers=_hdr(scaffold["token"]),
+    )
+    assert resp.status_code == 404
+
+
+def test_withdraw_never_offered_to_employers(scaffold) -> None:
+    """D-099: employers reject — there is NO employer withdraw route."""
+    client = scaffold["client"]
+    resp = client.post(
+        f"/v1/employers/applications/{scaffold['application_id']}/withdraw",
+        headers=_hdr(scaffold["token"]),
+    )
+    assert resp.status_code in (404, 405)
+
+
+# -- the P2-review pin (P2-2 disposition): the D-089 audit lane -------------------
+
+
+def test_transition_appends_employer_audit_row(scaffold) -> None:
+    """P2-2 pin: every employer transition appends its
+    employer_audit_events row (kind='application' — the vocabulary
+    extended in this phase; the one-queryable lane)."""
+
+    scaffold["client"].post(
+        f"/v1/employers/applications/{scaffold['application_id']}/in-review",
+        headers=_hdr(scaffold["token"]),
+    )
+    store = scaffold["client"].app.state.employer_store
+    events = store.list_audit_events(kind="application")
+    assert len(events) == 1
+    assert events[0].name == "application.in_review"
+    assert events[0].application_id if False else True  # noqa: the id rides detail
+    assert events[0].detail["application_id"] == scaffold["application_id"]
+    assert events[0].detail["to_status"] == "in_review"

apps/ai-service/tests/placements/test_learner_withdraw.py

@@ -0,0 +1,226 @@
+"""Learner withdraw tests (D-104/G-323, REQ-11-002) — the D-043 gate
+stack + the terminal arms + the apply-route terminal guard."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+from ai_service.marketplace.base import Posting
+from ai_service.marketplace.store import SQLiteMarketplaceStore
+
+from .conftest import PLACEMENT_LEARNERS, make_client
+
+
+def _seed_posting(store: SQLiteMarketplaceStore, employer_id: str = "emp-test") -> str:
+    import uuid
+
+    posting = store.insert_posting(
+        Posting(
+            id=f"post-{uuid.uuid4().hex[:12]}",
+            source="admin",
+            external_id=f"test-{uuid.uuid4().hex[:8]}",
+            employer_id=employer_id,
+            title="AI Orchestration Engineer",
+        )
+    )
+    # the learner apply route requires published (409 on drafts)
+    store.set_posting_status(posting.id, "published")
+    return posting.id
+
+
+def _apply(client: TestClient, posting_id: str, learner_id: str = "pilot-learner") -> str:
+    resp = client.post(
+        "/v1/marketplace/apply", json={"posting_id": posting_id, "learner_id": learner_id}
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["application_id"]
+
+
+@pytest.fixture()
+def scaffold(tmp_path, identity_store, marketplace_store):
+    client = make_client(tmp_path, identity_store, marketplace_store)
+    with client as entered:
+        posting_id = _seed_posting(marketplace_store)
+        application_id = _apply(entered, posting_id)
+        yield {
+            "client": entered,
+            "posting_id": posting_id,
+            "application_id": application_id,
+        }
+
+
+# -- the D-043 gate stack (G-5 → verified 18+ → the apply-rate window) ------------
+
+
+def test_withdraw_own_submitted(scaffold) -> None:
+    resp = scaffold["client"].post(
+        f"/v1/marketplace/applications/{scaffold['application_id']}/withdraw",
+        json={"learner_id": "pilot-learner"},
+    )
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["status"] == "withdrawn"
+
+
+def test_withdraw_unallowlisted_403(scaffold) -> None:
+    resp = scaffold["client"].post(
+        f"/v1/marketplace/applications/{scaffold['application_id']}/withdraw",
+        json={"learner_id": "intruder"},
+    )
+    assert resp.status_code == 403
+
+
+def test_withdraw_unverified_403(tmp_path, identity_store, marketplace_store) -> None:
+    """A learner on the roster but WITHOUT a verified identity record →
+    the verified-18+ arm refuses (the D-043 composition)."""
+    client = make_client(tmp_path, identity_store, marketplace_store)
+    with client as entered:
+        posting_id = _seed_posting(marketplace_store)
+        # an allowlisted-but-unverified learner: the suite roster carries
+        # learners beyond PLACEMENT_LEARNERS (the tests/conftest SUITE
+        # list); the identity store only seeds PLACEMENT_LEARNERS.
+        from ..conftest import SUITE_LEARNERS
+
+        unverified = next(name for name in SUITE_LEARNERS if name not in PLACEMENT_LEARNERS)
+        # even the APPLY is refused for an unverified learner (the same
+        # D-043 stack) — the withdraw arm pins identically: build the
+        # application row directly through the store, then attempt the
+        # withdraw through the route.
+        marketplace_store.insert_application(posting_id, unverified)
+        application = marketplace_store.list_applications(posting_id)[0]
+        resp = entered.post(
+            f"/v1/marketplace/applications/{application.id}/withdraw",
+            json={"learner_id": unverified},
+        )
+        assert resp.status_code == 403
+
+
+def test_withdraw_foreign_learner_403(scaffold) -> None:
+    resp = scaffold["client"].post(
+        f"/v1/marketplace/applications/{scaffold['application_id']}/withdraw",
+        json={"learner_id": "pilot-learner-2"},
+    )
+    assert resp.status_code == 403
+
+
+def test_withdraw_terminal_409(scaffold) -> None:
+    client = scaffold["client"]
+    client.post(
+        f"/v1/marketplace/applications/{scaffold['application_id']}/withdraw",
+        json={"learner_id": "pilot-learner"},
+    )
+    resp = client.post(
+        f"/v1/marketplace/applications/{scaffold['application_id']}/withdraw",
+        json={"learner_id": "pilot-learner"},
+    )
+    assert resp.status_code == 409
+
+
+def test_withdraw_unknown_404(scaffold) -> None:
+    resp = scaffold["client"].post(
+        "/v1/marketplace/applications/app-nonexistent/withdraw",
+        json={"learner_id": "pilot-learner"},
+    )
+    assert resp.status_code == 404
+
+
+# -- G-323: apply on a terminal application → 409, never a 200 lie --------------
+
+
+def test_reapply_after_withdraw_409(scaffold) -> None:
+    client = scaffold["client"]
+    client.post(
+        f"/v1/marketplace/applications/{scaffold['application_id']}/withdraw",
+        json={"learner_id": "pilot-learner"},
+    )
+    resp = client.post(
+        "/v1/marketplace/apply",
+        json={"posting_id": scaffold["posting_id"], "learner_id": "pilot-learner"},
+    )
+    assert resp.status_code == 409
+    assert resp.json()["detail"]["reason"] == "application_terminal"
+    assert resp.json()["detail"]["status"] == "withdrawn"
+
+
+def test_apply_idempotent_still_200_on_submitted(scaffold) -> None:
+    """A-408 unchanged: a duplicate apply on a LIVE application still
+    returns the 200 with the existing row (G-323 guards only terminals)."""
+    resp = scaffold["client"].post(
+        "/v1/marketplace/apply",
+        json={"posting_id": scaffold["posting_id"], "learner_id": "pilot-learner"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["application_id"] == scaffold["application_id"]
+
+
+# -- the P2-review pins (P2-1/P2-2 disposition): provenance + budget -------------
+
+
+def test_withdraw_ledger_actor_is_learner(scaffold) -> None:
+    """D-104 pin: the withdraw transition's ledger event records
+    actor_type='learner' with the learner's actor_ref."""
+
+    scaffold["client"].post(
+        f"/v1/marketplace/applications/{scaffold['application_id']}/withdraw",
+        json={"learner_id": "pilot-learner"},
+    )
+    placements = scaffold["client"].app.state.placement_store
+    events = placements.get_application_events(scaffold["application_id"])
+    assert len(events) == 1
+    assert events[0].event_type == "application.withdrawn"
+    assert events[0].actor_type == "learner"
+    assert events[0].actor_ref == "learner:pilot-learner"
+
+
+def test_withdraw_rides_the_apply_rate_window(scaffold) -> None:
+    """D-104/G-313 pin: withdraw shares the apply-rate window. Budget:
+    the scaffold's apply consumed 1; two withdraws (2,3) + one apply
+    (4) exhaust the 5/60s window; the NEXT gated action 429s honestly
+    (the conftest reset proves the next test starts clean)."""
+    client = scaffold["client"]
+    import uuid
+
+    from ai_service.marketplace.base import Posting
+
+    store = client.app.state.marketplace_store
+    # withdraw the scaffold application (action 2)
+    resp = client.post(
+        f"/v1/marketplace/applications/{scaffold['application_id']}/withdraw",
+        json={"learner_id": "pilot-learner"},
+    )
+    assert resp.status_code == 200, resp.text
+    # a second application: apply (action 3) + withdraw (action 4)
+    posting = store.insert_posting(
+        Posting(
+            id=f"post-w-{uuid.uuid4().hex[:6]}",
+            source="admin",
+            external_id=f"w-{uuid.uuid4().hex[:8]}",
+            employer_id="emp-test",
+            title="Posting",
+        )
+    )
+    store.set_posting_status(posting.id, "published")
+    application_id = _apply(client, posting.id)
+    resp = client.post(
+        f"/v1/marketplace/applications/{application_id}/withdraw",
+        json={"learner_id": "pilot-learner"},
+    )
+    assert resp.status_code == 200, resp.text
+    # a third application: the apply is the 5th gated action — green;
+    # its withdraw would be the 6th → 429
+    posting2 = store.insert_posting(
+        Posting(
+            id=f"post-w2-{uuid.uuid4().hex[:6]}",
+            source="admin",
+            external_id=f"w2-{uuid.uuid4().hex[:8]}",
+            employer_id="emp-test",
+            title="Posting",
+        )
+    )
+    store.set_posting_status(posting2.id, "published")
+    application2 = _apply(client, posting2.id)
+    resp = client.post(
+        f"/v1/marketplace/applications/{application2}/withdraw",
+        json={"learner_id": "pilot-learner"},
+    )
+    assert resp.status_code == 429

apps/ai-service/tests/placements/test_store.py

@@ -41,7 +41,7 @@ def _seed_posting(store: SQLiteMarketplaceStore, employer_id: str = "emp-test")
         Posting(
             id=f"post-{uuid.uuid4().hex[:12]}",
             source="admin",
-            external_id="test",
+            external_id=f"test-{uuid.uuid4().hex[:8]}",
             employer_id=employer_id,
             title="AI Orchestration Engineer",
             description="Test posting",