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": 0,
-  "stage": "mvp_ux_check",
+  "phase": 2,
+  "stage": "verify",
   "milestone": "v0.11",
-  "phase_role": "pre_execution",
+  "phase_role": "execution",
   "attempts": 0,
-  "updated_at": "2026-10-01T03:15:00Z"
+  "updated_at": "2026-10-01T07:10:00Z"
 }

.ciagent/REQUIREMENTS.md

@@ -8,7 +8,7 @@
 
 | ID | Description | Priority | Phase | Status |
 |----|-------------|----------|-------|--------|
-| REQ-11-001 | Placement store + evidence model: `ai_service/placements/` module (D-031 extend pattern — 9th D-027 store family, shared DB file, `create_all` idempotent): `PlacementRow` (id, learner_id, employer_id, posting_id, application_id, role_title, start_date, status, created_at, decided_at) with partial UNIQUE on `application_id` WHERE status='active' — one ACTIVE placement per application, revoked rows stay in history (D-102, the G-316 conditional-uniqueness shape); `PlacementEventRow` (append-only ledger: application_id nullable+indexed, placement_id nullable, event_type, actor_type, actor_ref, payload, prev_hash, hash, created_at) — one global hash chain, sha256 over canonical JSON + prev_hash, `UNIQUE(prev_hash)` fork backstop (D-101), tamper-evident by append-only discipline + hash chain (the chain makes silent rewrites detectable, G-301-class integrity); evidence links resolved from existing rows only (posting ref from postings store, application ref from applications store, employer ref from the claimed entity — G-302 doctrine: never trust denormalized copies); SQLModel typed access, Protocol + SQLite impl pair (the employer-store pattern), tz-normalization, fail-closed on unknown inputs | critical | 1 | planned |
+| REQ-11-001 | Placement store + evidence model: `ai_service/placements/` module (D-031 extend pattern — 9th D-027 store family, shared DB file, `create_all` idempotent): `PlacementRow` (id, learner_id, employer_id, posting_id, application_id, role_title, start_date, status, created_at, decided_at) with partial UNIQUE on `application_id` WHERE status='active' — one ACTIVE placement per application, revoked rows stay in history (D-102, the G-316 conditional-uniqueness shape); `PlacementEventRow` (append-only ledger: application_id nullable+indexed, placement_id nullable, event_type, actor_type, actor_ref, payload, prev_hash, hash, created_at) — one global hash chain, sha256 over canonical JSON + prev_hash, `UNIQUE(prev_hash)` fork backstop (D-101), tamper-evident by append-only discipline + hash chain (the chain makes silent rewrites detectable, G-301-class integrity); evidence links resolved from existing rows only (posting ref from postings store, application ref from applications store, employer ref from the claimed entity — G-302 doctrine: never trust denormalized copies); SQLModel typed access, Protocol + SQLite impl pair (the employer-store pattern), tz-normalization, fail-closed on unknown inputs | critical | 1 | complete (v0.10.1) |
 
 ### Application State Machine
 

.ciagent/ROADMAP.md

@@ -43,7 +43,7 @@ Payment processing and subscription management (REQ-F-018 activated per D-074, a
 | # | Name | Status | Depends On | Requirements | Success Criteria |
 |---|------|--------|------------|--------------|------------------|
 | 0 | Pre-execution | active | — | — | Specify (D-098), clarify, research, plan, grill, MVP/UX check |
-| 1 | Placement store + evidence ledger | planned | 0 | REQ-11-001 | `ai_service/placements/` module: 9th D-027 store (PlacementRow + PlacementEventRow, hash-chained append-only ledger), evidence resolved from source rows only (G-302 doctrine), `create_all` idempotent on the shared DB file, Protocol + SQLite pair |
+| 1 | Placement store + evidence ledger | complete (v0.10.1) | 0 | REQ-11-001 | `ai_service/placements/` module: 9th D-027 store (PlacementRow + PlacementEventRow, hash-chained append-only ledger), evidence resolved from source rows only (G-302 doctrine), `create_all` idempotent on the shared DB file, Protocol + SQLite pair |
 | 2 | Application state machine | planned | 1 | REQ-11-002 | Employer routes extend the v0.6 application status from submitted-only to the full hiring lifecycle; claimed-entity authorization (403 cross-entity, G-311 gate order); every transition ledger-logged; hired creates the placement; admin oversight retained |
 | 3 | Placement reporting | planned | 2 | REQ-11-003 | Aggregate counts from the ledger (per employer, per cohort month, cumulative toward ≥1,000); employer dashboard placements section; learner outcome states; admin network overview counts |
 | 4 | Audit export | planned | 3 | REQ-11-004 | Admin-gated export: cohort snapshot + per-placement evidence bundles + methodology doc + chain head hash; JSON + human-readable summary; re-export reproducibility test |

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

@@ -13,6 +13,7 @@ from ..grading.store import GradeStore
 from ..llm.base import LLMProvider
 from ..marketplace.embedding.base import EmbeddingProvider
 from ..marketplace.store import MarketplaceStore
+from ..placements.store import PlacementStore
 from ..sandbox.manager import SandboxManager
 from ..sandbox.workdir import SandboxDir
 from ..telemetry.ingest import TraceIntegrityMap
@@ -101,6 +102,13 @@ def get_employer_store(request: Request) -> EmployerStore:
     return request.app.state.employer_store
 
 
+def get_placement_store(request: Request) -> PlacementStore:
+    """v0.11 placements (D-100/D-112): the same app.state DI territory
+    rule — placement routes compose the store via app.state, never
+    import it."""
+    return request.app.state.placement_store
+
+
 def get_embedding_provider(request: Request) -> EmbeddingProvider:
     """v0.6 P3 (D-051): the embedding port via app.state DI — the
     search route + publish hooks compose it, never import it."""

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

@@ -199,6 +199,18 @@ def create_app(settings: Settings | None = None) -> FastAPI:
                 db_path=settings.db_path
             )
 
+        # Placement tracking (v0.11, REQ-11-001): 9th D-027 store
+        # family on the same SQLite file (D-100: the transactional
+        # transition writes the marketplace application row through the
+        # shared metadata — one file, one commit). State-injection
+        # override preserved (the same getattr-guard template).
+        if getattr(app.state, "placement_store", None) is None:
+            from .placements.store import SQLitePlacementStore
+
+            app.state.placement_store = SQLitePlacementStore(
+                db_path=settings.db_path
+            )
+
         # Payment provider (v0.9 P3, REQ-9-003, D-075): mock-first
         # default; stripe over the shared httpx pool. G-11 (boot
         # survival): a misconfigured real provider must never crash the
@@ -334,6 +346,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
             identity_store.close()
             app.state.billing_store.close()
             app.state.employer_store.close()
+            app.state.placement_store.close()
             await app.state.http_client.aclose()
 
     app = FastAPI(title="Nextcraft AI Service", version=__version__, lifespan=lifespan)
@@ -410,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

@@ -346,6 +346,23 @@ class MarketplaceStore(Protocol):
         — never a second application."""
         ...
 
+    def get_application(self, application_id: str) -> Application | None:
+        """Point lookup by application id (D-100: the placements
+        store's transition path reads the row through the shared
+        metadata; this read serves routes + tests — the get_posting
+        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]:
@@ -900,6 +917,36 @@ class SQLiteMarketplaceStore:
             session.refresh(row)
             return row.to_application(), True
 
+    def get_application(self, application_id: str) -> Application | None:
+        """Point lookup by application id (D-100, the get_posting shape)."""
+        with Session(self._engine) as session:
+            row = session.get(ApplicationRow, application_id)
+            if row is None:
+                return None
+            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

@@ -0,0 +1,241 @@
+"""Placement contracts: the D-099 hiring-state matrix, the D-101
+event vocabulary + GENESIS, and the placement wire shapes (REQ-11-001).
+
+Boundary (D-031 family): `placements/` never imports `api/`; stdlib +
+pydantic + the marketplace WIRE contracts only. The store imports the
+marketplace ROW (the sanctioned D-100 cross-family seam — one
+transaction covers both tables on the shared DB file).
+
+Id honesty (A-305, the family rule): every id in this module is an
+opaque handle minted UNIQUELY per creation call — never derived from
+user input or a deterministic hash of it.
+"""
+
+from __future__ import annotations
+
+from datetime import date, datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+#: D-099 application status vocabulary (snake_case, the billing
+#: `past_due` precedent): the hiring lifecycle. `submitted` is the
+#: v0.6 status (zero migration); hired/rejected/withdrawn are TERMINAL.
+ApplicationStatus = Literal[
+    "submitted",
+    "in_review",
+    "interview",
+    "offer",
+    "hired",
+    "rejected",
+    "withdrawn",
+]
+
+#: Tuple mirror of the Literal (docs + tests import this).
+APPLICATION_STATUSES: tuple[ApplicationStatus, ...] = (
+    "submitted",
+    "in_review",
+    "interview",
+    "offer",
+    "hired",
+    "rejected",
+    "withdrawn",
+)
+
+#: Terminal statuses — no outgoing edges (D-099).
+TERMINAL_APPLICATION_STATUSES: tuple[ApplicationStatus, ...] = (
+    "hired",
+    "rejected",
+    "withdrawn",
+)
+
+#: D-099 transition matrix (the billing/store.py `_VALID_TRANSITIONS`
+#: shape): forward-only WITH skip-forward — from any non-terminal
+#: status, any LATER stage and any terminal are legal (a fast-track
+#: `submitted → hired` is a legal hire — CIRR counts the hire, not the
+#: ceremony); NO backward moves; terminals have empty tuples.
+TRANSITION_MATRIX: dict[str, tuple[str, ...]] = {
+    "submitted": (
+        "in_review",
+        "interview",
+        "offer",
+        "hired",
+        "rejected",
+        "withdrawn",
+    ),
+    "in_review": ("interview", "offer", "hired", "rejected", "withdrawn"),
+    "interview": ("offer", "hired", "rejected", "withdrawn"),
+    "offer": ("hired", "rejected", "withdrawn"),
+    "hired": (),
+    "rejected": (),
+    "withdrawn": (),
+}
+
+#: Placement lifecycle (D-102): created `active`; the ONLY transition
+#: is the operator correction `active → revoked`. Revoked rows stay in
+#: history (the partial unique index never touches them — re-hire
+#: creates a fresh row, R-80).
+PlacementStatus = Literal["active", "revoked"]
+PLACEMENT_STATUSES: tuple[PlacementStatus, ...] = ("active", "revoked")
+
+#: D-101 actor vocabulary (the billing `_BILLING_EVENT_KINDS` shape).
+ACTOR_TYPES: tuple[str, ...] = ("employer", "learner", "operator", "system")
+
+#: D-101 event-name vocabulary: application-stage events (one per
+#: transition — NO backfill: pre-v0.11 `submitted` rows carry only
+#: their created_at, never a fabricated event, A-304) + the placement
+#: lane (created on hired, revoked on the D-102 correction, rehired
+#: on the R-80 operator re-hire).
+EVENT_APPLICATION_TRANSITION: dict[str, str] = {
+    "in_review": "application.in_review",
+    "interview": "application.interview",
+    "offer": "application.offer",
+    "hired": "application.hired",
+    "rejected": "application.rejected",
+    "withdrawn": "application.withdrawn",
+}
+EVENT_PLACEMENT_CREATED = "placement.created"
+EVENT_PLACEMENT_REVOKED = "placement.revoked"
+EVENT_PLACEMENT_REHIRED = "placement.rehired"
+EVENT_NAMES: tuple[str, ...] = (
+    *EVENT_APPLICATION_TRANSITION.values(),
+    EVENT_PLACEMENT_CREATED,
+    EVENT_PLACEMENT_REVOKED,
+    EVENT_PLACEMENT_REHIRED,
+)
+
+#: D-101: the chain's genesis — the first event's prev_hash. A module
+#: constant, test-pinned (never a row — the walk starts here).
+GENESIS_HASH = "0" * 64
+
+
+class InvalidTransition(ValueError):
+    """A matrix-illegal application status move (D-099) or placement
+    status move (D-102) — the route's honest 409."""
+
+
+def is_terminal(status: str) -> bool:
+    """True when the application status has no outgoing edges."""
+    return status in TERMINAL_APPLICATION_STATUSES
+
+
+def validate_transition(from_status: str, to_status: str) -> None:
+    """D-099 matrix validation — raises InvalidTransition on any
+    illegal move (backward, terminal-source, unknown status)."""
+    if from_status not in TRANSITION_MATRIX:
+        raise InvalidTransition(f"unknown application status {from_status!r}")
+    if to_status not in APPLICATION_STATUSES:
+        raise InvalidTransition(f"unknown application status {to_status!r}")
+    if to_status not in TRANSITION_MATRIX[from_status]:
+        legal = TRANSITION_MATRIX[from_status] or "(terminal)"
+        raise InvalidTransition(
+            f"illegal transition {from_status!r} → {to_status!r} "
+            f"(legal: {legal})"
+        )
+
+
+class HireTransitionRequest(BaseModel):
+    """The hire body (D-103/G-324): role_title is employer-attested
+    free text — never silently the posting title — with the house
+    Field bounds (empty-string → 422); start_date is the employment
+    start DATE (first day of work), never a datetime: timezone-free by
+    construction, future + past allowed (D-103)."""
+
+    role_title: str = Field(min_length=1, max_length=200)
+    start_date: date
+
+
+class PlacementRead(BaseModel):
+    """A placement record over the wire (the PlacementRow read shape).
+
+    employer/learner/posting are ID REFERENCES (G-302) — resolved from
+    source rows at read/export, never trusted denormalized copies."""
+
+    model_config = ConfigDict(frozen=True)
+
+    id: str = Field(min_length=1)
+    learner_id: str = Field(min_length=1)
+    employer_id: str = Field(min_length=1)
+    posting_id: str = Field(min_length=1)
+    application_id: str = Field(min_length=1)
+    role_title: str = Field(min_length=1)
+    start_date: date
+    status: PlacementStatus
+    created_at: datetime
+    decided_at: datetime
+
+
+class PlacementEventRead(BaseModel):
+    """One ledger event over the wire (the PlacementEventRow read
+    shape) — application-stage events carry application_id with
+    placement_id None; the placement lane carries placement_id set."""
+
+    model_config = ConfigDict(frozen=True)
+
+    id: int
+    application_id: str | None = None
+    placement_id: str | None = None
+    event_type: str = Field(min_length=1)
+    actor_type: str = Field(min_length=1)
+    actor_ref: str = Field(min_length=1)
+    payload: dict[str, Any] = Field(default_factory=dict)
+    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/ai_service/placements/store.py

@@ -0,0 +1,676 @@
+"""Placement persistence — the 9th D-027 store (REQ-11-001, D-112).
+
+TWO tables on the shared DB file (R-72): `placement` (immutable at
+creation except status — D-102) + `placement_event` (the append-only
+GLOBAL hash chain — D-101). Mechanics mirror employers/store.py
+exactly (pragmas, _validate/_normalize, Protocol + SQLite pair,
+surfaced IntegrityError, _constraint_error helper).
+
+NO DELETE PATH (R-70, binding): this module contains NO delete method
+and NO DELETE SQL — deleting the top ledger row would make SQLite's
+INTEGER PRIMARY KEY reuse max+1 and break the chain. Pinned by test.
+
+STATUS-COLUMN OWNERSHIP (R-78, binding): the marketplace store writes
+`application.status` EXACTLY ONCE at creation (insert_application);
+THIS store is the ONLY post-creation writer — via the ONE
+`transition_application` method, never a second append path (D-100:
+the status write + the ledger event + the placement-on-hired land in
+ONE transaction; a two-commit alternative leaves a crash window that
+voids the tamper-evidence this milestone sells).
+
+Canonical serialization (D-101/R-69/R-79, binding): the hash covers
+the canonical JSON of {id, application_id, placement_id, event_type,
+actor_type, actor_ref, payload, created_at} + prev_hash — sorted keys,
+compact separators, created_at ALWAYS `_as_utc(dt).isoformat()` (ONE
+pinned normalization: naive-in-SQLite vs aware-on-read must never
+drift), payload values str/int/bool/None only (floats REJECTED —
+R-69).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sqlite3
+import uuid
+from datetime import UTC, date, datetime
+from pathlib import Path
+from typing import Any, Protocol
+
+from sqlalchemy import Index, event, func, text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.types import JSON, String
+from sqlmodel import Field, Session, SQLModel, col, create_engine, select
+
+from ..config import Settings
+from ..marketplace.store import ApplicationRow
+from .base import (
+    ACTOR_TYPES,
+    EVENT_APPLICATION_TRANSITION,
+    EVENT_PLACEMENT_CREATED,
+    EVENT_PLACEMENT_REHIRED,
+    EVENT_PLACEMENT_REVOKED,
+    GENESIS_HASH,
+    InvalidTransition,
+    validate_transition,
+)
+
+_PLACEMENT_ID_PREFIX = "plc-"
+
+
+def _as_utc(ts: datetime) -> datetime:
+    """SQLite stores naive; read paths re-label tz-aware UTC (D-027
+    pattern). Aware NON-UTC inputs CONVERT to the true instant (the
+    security review's P0-1: re-labeling preserves the offset — an
+    append-time canonical form of 12:00-04:00 would re-read as
+    12:00+00:00 and permanently break the chain, with no delete path
+    to recover)."""
+    if ts.tzinfo is None:
+        return ts.replace(tzinfo=UTC)
+    return ts.astimezone(UTC)
+
+
+def _constraint_error(message: str, params: dict[str, Any]) -> IntegrityError:
+    """Hand-raised IntegrityError (the guard arms) — the caller's
+    honest 409, never a silent overwrite."""
+    return IntegrityError(message, params, ValueError(message))
+
+
+def _sqlite_connect(dbapi_connection: sqlite3.Connection, _: object) -> None:
+    """Per-connection pragmas — mirrors the other D-027 stores."""
+    cursor = dbapi_connection.cursor()
+    cursor.execute("PRAGMA journal_mode=WAL")
+    cursor.execute("PRAGMA synchronous=NORMAL")
+    cursor.execute("PRAGMA busy_timeout=5000")
+    cursor.execute("PRAGMA foreign_keys=ON")
+    cursor.close()
+
+
+def _check_payload(value: Any) -> None:
+    """R-69: payload values are str/int/bool/None ONLY — floats (and
+    any nested container) are rejected at the store boundary: float
+    representation is a canonical-serialization drift the chain must
+    never depend on."""
+    if isinstance(value, bool) or value is None:
+        return
+    if isinstance(value, str):
+        return
+    if isinstance(value, int):
+        return
+    raise ValueError(
+        f"payload values must be str/int/bool/None, got {type(value).__name__}"
+    )
+
+
+class PlacementRow(SQLModel, table=True):
+    """A placement record (D-102): created by the hired transition,
+    IMMUTABLE except `status` (exactly one transition active → revoked,
+    operator-only by route). role_title/start_date/decided_at and the
+    four ID references are set once and NEVER rewritten — a silent
+    rewrite is exactly the tamper the ledger exists to detect.
+
+    G-316 shape: the partial UNIQUE index uq_active_placement_per_
+    application makes one-ACTIVE-placement-per-application a
+    STRUCTURAL guard — revoked rows stay in history and never match
+    the partial WHERE, so a re-hire after a mistaken revoke creates a
+    fresh row (R-80) and an operator fat-finger never permanently
+    erases a learner from the ≥1,000 count.
+    """
+
+    __tablename__ = "placement"
+    __table_args__ = (
+        Index(
+            "uq_active_placement_per_application",
+            "application_id",
+            unique=True,
+            sqlite_where=text("status='active'"),
+        ),
+    )
+
+    id: str = Field(primary_key=True, sa_type=String)
+    learner_id: str = Field(index=True, sa_type=String)
+    employer_id: str = Field(sa_type=String)
+    #: G-302: ID references resolved from source rows at read/export.
+    posting_id: str = Field(sa_type=String)
+    application_id: str = Field(index=True, sa_type=String)
+    role_title: str = Field(sa_type=String)
+    #: D-103: ISO YYYY-MM-DD — a calendar date, no datetime coercion.
+    start_date: str = Field(sa_type=String)
+    status: str = Field(default="active", sa_type=String)
+    created_at: datetime | None = Field(default=None)
+    #: D-102: the hired transition's store-stamped created_at (G-326:
+    #: the re-hire arm stamps the RE-HIRE event's created_at).
+    decided_at: datetime | None = Field(default=None)
+
+    def _validate(self) -> None:
+        if not self.id:
+            raise ValueError("id must be non-empty")
+        for field_name in (
+            "learner_id",
+            "employer_id",
+            "posting_id",
+            "application_id",
+            "role_title",
+            "start_date",
+        ):
+            if not getattr(self, field_name):
+                raise ValueError(f"{field_name} must be non-empty")
+        if self.status not in ("active", "revoked"):
+            raise ValueError(f"invalid placement status {self.status!r}")
+
+    def _normalize(self) -> None:
+        self.created_at = _as_utc(self.created_at) if self.created_at else None
+        self.decided_at = _as_utc(self.decided_at) if self.decided_at else None
+
+
+class PlacementEventRow(SQLModel, table=True):
+    """One append-only ledger event (D-101): ONE GLOBAL chain ordered
+    by autoincrement id. Each row stores prev_hash (the prior event's
+    hash; GENESIS_HASH for the first) + its own hash = sha256 over the
+    canonical JSON + prev_hash. UNIQUE(prev_hash) makes a chain fork
+    structurally impossible — two events claiming the same
+    predecessor surface IntegrityError (the G-316 backstop class).
+
+    Application-stage events carry application_id (placement_id
+    None); the placement lane (created/revoked/rehired) carries
+    placement_id set from hired onward.
+    """
+
+    __tablename__ = "placement_event"
+
+    id: int | None = Field(default=None, primary_key=True)
+    application_id: str | None = Field(default=None, index=True, sa_type=String)
+    placement_id: str | None = Field(default=None, sa_type=String)
+    event_type: str = Field(sa_type=String)
+    actor_type: str = Field(sa_type=String)
+    actor_ref: str = Field(sa_type=String)
+    payload: dict[str, Any] = Field(default_factory=dict, sa_type=JSON)
+    prev_hash: str = Field(unique=True, sa_type=String)
+    hash: str = Field(default="", sa_type=String)
+    created_at: datetime | None = Field(default=None)
+
+    def _validate(self) -> None:
+        if self.id is not None and self.id < 1:
+            raise ValueError("id must be a positive int when set")
+        if not self.event_type:
+            raise ValueError("event_type must be non-empty")
+        if self.actor_type not in ACTOR_TYPES:
+            raise ValueError(f"unknown actor_type {self.actor_type!r}")
+        if not self.actor_ref:
+            raise ValueError("actor_ref must be non-empty")
+        if not self.prev_hash:
+            raise ValueError("prev_hash must be non-empty")
+        for value in self.payload.values():
+            _check_payload(value)
+
+    def _normalize(self) -> None:
+        self.created_at = _as_utc(self.created_at) if self.created_at else None
+
+    def canonical_json(self) -> str:
+        """D-101/R-79: the canonical serialization — sorted keys,
+        compact separators, created_at ALWAYS `_as_utc(dt).isoformat()`
+        (ONE pinned normalization; the hashing and reading paths share
+        this function so they can never drift)."""
+        return json.dumps(
+            {
+                "id": self.id,
+                "application_id": self.application_id,
+                "placement_id": self.placement_id,
+                "event_type": self.event_type,
+                "actor_type": self.actor_type,
+                "actor_ref": self.actor_ref,
+                "payload": self.payload,
+                "created_at": (
+                    _as_utc(self.created_at).isoformat()
+                    if self.created_at is not None
+                    else None
+                ),
+            },
+            sort_keys=True,
+            separators=(",", ":"),
+        )
+
+
+def _compute_hash(canonical_json: str, prev_hash: str) -> str:
+    """sha256 over the canonical JSON + prev_hash (D-101 — the D-087
+    stdlib sha256 precedent; hexdigest)."""
+    return hashlib.sha256((canonical_json + prev_hash).encode("utf-8")).hexdigest()
+
+
+class PlacementStore(Protocol):
+    """Persistence contract for placements (D-101/D-102/D-112)."""
+
+    def create_schema(self) -> None:
+        """Ensure both tables exist (idempotent — shared DB file)."""
+        ...
+
+    def transition_application(
+        self,
+        application_id: str,
+        to_status: str,
+        actor_type: str,
+        actor_ref: str,
+        role_title: str | None = None,
+        start_date: date | None = None,
+        now: datetime | None = None,
+    ) -> tuple[ApplicationRow, PlacementRow | None]:
+        """D-100: the ONE transactional transition — update the
+        marketplace application row + append the ledger event + create
+        the PlacementRow on hired, ALL IN ONE COMMIT. One append per
+        transaction per lane (transition/revoke/rehire each append
+        exactly once); R-70: the appends race nothing — the chain's
+        prev_hash read-then-write is single-threaded-safe through
+        SQLite's writer serialization + the UNIQUE backstop."""
+        ...
+
+    def verify_chain(self) -> tuple[bool, str]:
+        """Recompute the walk from GENESIS_HASH → head; (ok, head_hash)."""
+        ...
+
+    def get_placement(self, placement_id: str) -> PlacementRow | None:
+        ...
+
+    def list_placements(self, status: str | None = None) -> list[PlacementRow]:
+        ...
+
+    def get_application_events(self, application_id: str) -> list[PlacementEventRow]:
+        ...
+
+    def list_events(self) -> list[PlacementEventRow]:
+        ...
+
+    def revoke_placement(
+        self, placement_id: str, actor_ref: str, now: datetime | None = None
+    ) -> PlacementRow | None:
+        """D-102: the one status-only mutation active → revoked + the
+        placement.revoked event, one transaction (operator-only by
+        route)."""
+        ...
+
+    def rehire_placement(
+        self,
+        application_id: str,
+        role_title: str,
+        start_date: date,
+        actor_ref: str,
+        now: datetime | None = None,
+    ) -> PlacementRow:
+        """R-80/G-326: a FRESH PlacementRow + placement.rehired event
+        WITHOUT touching the application's terminal status — the
+        correction lane. decided_at = the re-hire event's created_at.
+        An existing ACTIVE placement on the application surfaces its
+        IntegrityError (the caller's honest 409)."""
+        ...
+
+    def close(self) -> None:
+        ...
+
+
+class SQLitePlacementStore:
+    """SQLite implementation of PlacementStore (D-027/D-112)."""
+
+    def __init__(self, db_path: Path | None = None) -> None:
+        self._db_path: Path = db_path if db_path is not None else Settings().db_path
+        self._engine = create_engine(
+            f"sqlite:///{self._db_path}",
+            connect_args={"check_same_thread": False},
+        )
+        event.listen(self._engine, "connect", _sqlite_connect)
+        self.create_schema()
+
+    def create_schema(self) -> None:
+        """Ensure both tables exist (idempotent — shared DB file)."""
+        SQLModel.metadata.create_all(self._engine)
+
+    # -- chain core (D-101/R-70) -------------------------------------------
+
+    @staticmethod
+    def _chain_head(session: Session) -> tuple[int | None, str]:
+        """(last event id, its hash) or (None, GENESIS_HASH) — the
+        next append's prev_hash source."""
+        row = session.exec(
+            select(PlacementEventRow).order_by(col(PlacementEventRow.id).desc())
+        ).first()
+        if row is None or row.hash == "":
+            return None, GENESIS_HASH
+        return row.id, row.hash
+
+    @staticmethod
+    def _append_event(
+        session: Session,
+        *,
+        application_id: str | None,
+        placement_id: str | None,
+        event_type: str,
+        actor_type: str,
+        actor_ref: str,
+        payload: dict[str, Any] | None,
+        prev_hash: str,
+        created_at: datetime,
+    ) -> PlacementEventRow:
+        """R-70: flush-then-hash-then-commit — the row's own
+        autoincrement id participates in its own hash, so the append
+        flushes (id assigned), computes the hash, sets the column, and
+        flushes again. The CALLER owns the transaction (D-100: the
+        status write + the events + the placement land in one commit);
+        this helper never commits."""
+        row = PlacementEventRow(
+            application_id=application_id,
+            placement_id=placement_id,
+            event_type=event_type,
+            actor_type=actor_type,
+            actor_ref=actor_ref,
+            payload=dict(payload or {}),
+            prev_hash=prev_hash,
+            hash="",
+            created_at=created_at,
+        )
+        row._validate()
+        row._normalize()
+        session.add(row)
+        # flush #1: the autoincrement id is assigned
+        session.flush()
+        assert row.id is not None  # flush assigns it
+        # the hash covers the canonical JSON + prev_hash
+        row.hash = _compute_hash(row.canonical_json(), row.prev_hash)
+        # flush #2: the hash column lands
+        session.flush()
+        return row
+
+    def verify_chain(self) -> tuple[bool, str]:
+        """Recompute the walk GENESIS_HASH → head (R-70: O(n) sha256
+        over tiny rows — milliseconds at pilot scale)."""
+        with Session(self._engine) as session:
+            prev = GENESIS_HASH
+            last_hash = GENESIS_HASH
+            rows = session.exec(select(PlacementEventRow).order_by(col(PlacementEventRow.id))).all()
+            for row in rows:
+                row._normalize()
+                expected = _compute_hash(row.canonical_json(), prev)
+                if row.prev_hash != prev or row.hash != expected:
+                    return False, last_hash
+                prev = row.hash
+                last_hash = row.hash
+            return True, last_hash
+
+    # -- the ONE transition (D-100) ------------------------------------------
+
+    def transition_application(
+        self,
+        application_id: str,
+        to_status: str,
+        actor_type: str,
+        actor_ref: str,
+        role_title: str | None = None,
+        start_date: date | None = None,
+        now: datetime | None = None,
+    ) -> tuple[ApplicationRow, PlacementRow | None]:
+        """D-100: ONE Session on the placements engine covering BOTH
+        tables + the marketplace application row (the sanctioned
+        cross-family write — R-72: global SQLModel metadata means the
+        placements engine sees ApplicationRow; both stores share the
+        one SQLite file, so one commit covers everything atomically,
+        the decide_claim all-or-nothing precedent).
+
+        R-78: this is the ONLY post-creation writer of
+        application.status — never a second caller.
+        """
+        moment = _as_utc(now) if now is not None else datetime.now(UTC)
+        with Session(self._engine) as session:
+            application = session.get(ApplicationRow, application_id)
+            if application is None:
+                raise InvalidTransition(f"unknown application {application_id!r}")
+            validate_transition(application.status, to_status)
+            if to_status == "hired":
+                if not role_title:
+                    raise ValueError("role_title is required on hire (D-103)")
+                if start_date is None:
+                    raise ValueError("start_date is required on hire (D-103)")
+
+            # the status write (the only post-creation writer — R-78);
+            # the pre-transition status is captured BEFORE the mutation
+            from_status = application.status
+            application.status = to_status
+            application._normalize()
+
+            prev_id, prev_hash = self._chain_head(session)
+            payload: dict[str, Any] = {"from_status": from_status, "to_status": to_status}
+            self._append_event(
+                session,
+                application_id=application_id,
+                placement_id=None,
+                event_type=EVENT_APPLICATION_TRANSITION[to_status],
+                actor_type=actor_type,
+                actor_ref=actor_ref,
+                payload=payload,
+                prev_hash=prev_hash,
+                created_at=moment,
+            )
+
+            placement: PlacementRow | None = None
+            if to_status == "hired":
+                assert role_title is not None and start_date is not None
+                placement = PlacementRow(
+                    id=f"{_PLACEMENT_ID_PREFIX}{uuid.uuid4().hex[:16]}",
+                    learner_id=application.learner_id,
+                    employer_id=self._posting_employer_id(session, application.posting_id),
+                    posting_id=application.posting_id,
+                    application_id=application.id,
+                    role_title=role_title,
+                    start_date=start_date.isoformat(),
+                    status="active",
+                    created_at=moment,
+                    decided_at=moment,
+                )
+                placement._validate()
+                placement._normalize()
+                session.add(placement)
+                # the placement row must exist before its event references it
+                session.flush()
+                _, prev_hash_2 = self._chain_head(session)
+                self._append_event(
+                    session,
+                    application_id=application_id,
+                    placement_id=placement.id,
+                    event_type=EVENT_PLACEMENT_CREATED,
+                    actor_type=actor_type,
+                    actor_ref=actor_ref,
+                    payload={
+                        "role_title": role_title,
+                        "start_date": start_date.isoformat(),
+                    },
+                    prev_hash=prev_hash_2,
+                    created_at=moment,
+                )
+
+            session.commit()
+            session.refresh(application)
+            application._normalize()
+            if placement is not None:
+                session.refresh(placement)
+                placement._normalize()
+            return application, placement
+
+    @staticmethod
+    def _posting_employer_id(session: Session, posting_id: str) -> str:
+        """G-302: resolve the posting's employer_id from the SOURCE row
+        at placement-creation time — never trusted from the caller."""
+        from ..marketplace.store import PostingRow
+
+        posting = session.get(PostingRow, posting_id)
+        if posting is None:
+            raise _constraint_error(
+                f"guard: application references unknown posting "
+                f"{posting_id!r} — refusing to create placement",
+                {"posting_id": posting_id},
+            )
+        return posting.employer_id
+
+    # -- reads (consume-free, G-302 ID references) ---------------------------
+
+    def get_placement(self, placement_id: str) -> PlacementRow | None:
+        with Session(self._engine) as session:
+            row = session.get(PlacementRow, placement_id)
+            if row is None:
+                return None
+            row._normalize()
+            return row
+
+    def list_placements(self, status: str | None = None) -> list[PlacementRow]:
+        """created_at then id (the list_postings ordering discipline,
+        R-73 — deterministic export ordering)."""
+        with Session(self._engine) as session:
+            stmt = select(PlacementRow)
+            if status is not None:
+                stmt = stmt.where(PlacementRow.status == status)
+            rows = session.exec(
+                stmt.order_by(col(PlacementRow.created_at), PlacementRow.id)
+            ).all()
+            for row in rows:
+                row._normalize()
+            return list(rows)
+
+    def get_application_events(self, application_id: str) -> list[PlacementEventRow]:
+        """id-ordered (the chain order) — scoped to ONE application."""
+        with Session(self._engine) as session:
+            rows = session.exec(
+                select(PlacementEventRow)
+                .where(PlacementEventRow.application_id == application_id)
+                .order_by(col(PlacementEventRow.id))
+            ).all()
+            for row in rows:
+                row._normalize()
+            return list(rows)
+
+    def list_events(self) -> list[PlacementEventRow]:
+        """The whole ledger, id-ordered (the export embeds every event
+        so the auditor's walk is offline — R-70)."""
+        with Session(self._engine) as session:
+            rows = session.exec(
+                select(PlacementEventRow).order_by(col(PlacementEventRow.id))
+            ).all()
+            for row in rows:
+                row._normalize()
+            return list(rows)
+
+    def count_events(self, event_type: str | None = None) -> int:
+        with Session(self._engine) as session:
+            stmt = select(func.count()).select_from(PlacementEventRow)
+            if event_type is not None:
+                stmt = stmt.where(PlacementEventRow.event_type == event_type)
+            return session.exec(stmt).one()
+
+    # -- the D-102 correction lane ------------------------------------------
+
+    def revoke_placement(
+        self, placement_id: str, actor_ref: str, now: datetime | None = None
+    ) -> PlacementRow | None:
+        """D-102: the one status-only mutation active → revoked + the
+        placement.revoked event, one transaction (operator-only by
+        route). None = unknown placement (route 404s)."""
+        moment = _as_utc(now) if now is not None else datetime.now(UTC)
+        with Session(self._engine) as session:
+            row = session.get(PlacementRow, placement_id)
+            if row is None:
+                return None
+            if row.status != "active":
+                raise InvalidTransition(
+                    f"placement {placement_id!r} is {row.status!r} — "
+                    "only an active placement can be revoked"
+                )
+            row.status = "revoked"
+            row._validate()
+            row._normalize()
+            _, prev_hash = self._chain_head(session)
+            self._append_event(
+                session,
+                application_id=row.application_id,
+                placement_id=row.id,
+                event_type=EVENT_PLACEMENT_REVOKED,
+                actor_type="operator",
+                actor_ref=actor_ref,
+                payload={"placement_id": placement_id},
+                prev_hash=prev_hash,
+                created_at=moment,
+            )
+            session.commit()
+            session.refresh(row)
+            row._normalize()
+            return row
+
+    def rehire_placement(
+        self,
+        application_id: str,
+        role_title: str,
+        start_date: date,
+        actor_ref: str,
+        now: datetime | None = None,
+    ) -> PlacementRow:
+        """R-80/G-326: the operator re-hire — a FRESH PlacementRow +
+        placement.rehired event WITHOUT touching the application's
+        terminal status (hired → hired is a 409 on every
+        matrix-validated surface; placements, not applications, are
+        the correction lane). decided_at = the re-hire event's
+        store-stamped created_at (G-326 — D-102's stamp rule extended).
+
+        The partial unique index is the guard: an existing ACTIVE
+        placement on the application refuses the insert and its
+        IntegrityError SURFACES (the caller's honest 409 — revoked
+        rows never match the partial WHERE, the exact G-316 property).
+        """
+        if not role_title:
+            raise ValueError("role_title is required on re-hire (D-103)")
+        if start_date is None:
+            raise ValueError("start_date is required on re-hire (D-103)")
+        moment = _as_utc(now) if now is not None else datetime.now(UTC)
+        with Session(self._engine) as session:
+            application = session.get(ApplicationRow, application_id)
+            if application is None:
+                raise InvalidTransition(f"unknown application {application_id!r}")
+            if application.status != "hired":
+                raise InvalidTransition(
+                    f"application {application_id!r} is {application.status!r} — "
+                    "re-hire is the correction lane for a hired application"
+                )
+            placement = PlacementRow(
+                id=f"{_PLACEMENT_ID_PREFIX}{uuid.uuid4().hex[:16]}",
+                learner_id=application.learner_id,
+                employer_id=self._posting_employer_id(session, application.posting_id),
+                posting_id=application.posting_id,
+                application_id=application.id,
+                role_title=role_title,
+                start_date=start_date.isoformat(),
+                status="active",
+                created_at=moment,
+                # G-326: the re-hire event's stamp, not the original hire's
+                decided_at=moment,
+            )
+            placement._validate()
+            placement._normalize()
+            session.add(placement)
+            # the partial unique index refuses a second ACTIVE row —
+            # its IntegrityError SURFACES (never swallowed, the
+            # insert_claim discipline)
+            session.flush()
+            _, prev_hash = self._chain_head(session)
+            self._append_event(
+                session,
+                application_id=application.id,
+                placement_id=placement.id,
+                event_type=EVENT_PLACEMENT_REHIRED,
+                actor_type="operator",
+                actor_ref=actor_ref,
+                payload={
+                    "role_title": role_title,
+                    "start_date": start_date.isoformat(),
+                },
+                prev_hash=prev_hash,
+                created_at=moment,
+            )
+            session.commit()
+            session.refresh(placement)
+            placement._normalize()
+            return placement
+
+    def close(self) -> None:
+        self._engine.dispose()

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

@@ -0,0 +1,116 @@
+"""Placement test fixtures (hermeticity rule: db_path pins tmp_path —
+never ~/.nextcraft, never the cloud).
+
+Mirrors the tests/marketplace pattern: tmp-path stores injected into
+a real create_app via state-injection (the G-9 precedent) — the
+transition routes see the exact composition production runs.
+
+G-313 (D-111): the autouse `_APPLY_TIMES` reset — the learner
+withdraw route (P2) rides the marketplace apply-rate window; without
+this reset the suite self-429s (the lesson applied BEFORE it ships).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+from pydantic import SecretStr
+
+from ai_service.config import Settings
+from ai_service.identity.store import SQLiteIdentityStore
+from ai_service.main import create_app
+from ai_service.marketplace.store import SQLiteMarketplaceStore
+from ai_service.placements.store import SQLitePlacementStore
+
+from ..conftest import SUITE_LEARNERS, seed_verified_identity
+
+#: The pilot learner roster for gated-route tests (G-5 allowlist
+#: coverage + verified 18+ identity records).
+PLACEMENT_LEARNERS = ["pilot-learner", "pilot-learner-2"]
+
+ADMIN_TOKEN = "test-admin-token"
+
+
+@pytest.fixture()
+def marketplace_store(tmp_path: Path) -> Iterator[SQLiteMarketplaceStore]:
+    """Hermetic store: D-100's topology — marketplace + placements
+    share ONE SQLite file (the cross-store transaction depends on
+    it)."""
+    store = SQLiteMarketplaceStore(db_path=tmp_path / "shared.db")
+    yield store
+    store.close()
+
+
+@pytest.fixture()
+def identity_store(tmp_path: Path) -> Iterator[SQLiteIdentityStore]:
+    store = SQLiteIdentityStore(db_path=tmp_path / "identity.db")
+    seed_verified_identity(store, PLACEMENT_LEARNERS)
+    yield store
+    store.close()
+
+
+@pytest.fixture()
+def placement_store(tmp_path: Path) -> Iterator[SQLitePlacementStore]:
+    """The 9th D-027 store on the SAME shared file as the marketplace
+    store (D-100/R-72: one SQLite file, two engines — the shipped
+    multi-store topology)."""
+    store = SQLitePlacementStore(db_path=tmp_path / "shared.db")
+    yield store
+    store.close()
+
+
+def make_client(
+    tmp_path: Path,
+    identity_store: SQLiteIdentityStore,
+    marketplace_store: SQLiteMarketplaceStore,
+    admin_token: str | None = ADMIN_TOKEN,
+) -> TestClient:
+    """Full app with state-injected stores; admin_token=None exercises
+    the D-057 fail-closed matrix. P1-2 (security review): ALL stores
+    share the ONE tmp file (marketplace rides shared.db; the app DB +
+    placement store point there too) — D-100's cross-store transition
+    depends on the single-file topology, exactly as production runs."""
+    settings = Settings(
+        provider="mock",
+        learner_allowlist=SUITE_LEARNERS,
+        db_path=tmp_path / "shared.db",
+        sandbox_dir=tmp_path / "sandboxes",
+        admin_token=SecretStr(admin_token or ""),
+    )
+    app = create_app(settings)
+    app.state.identity_store = identity_store  # state-injection (G-9)
+    app.state.marketplace_store = marketplace_store
+    app.state.placement_store = SQLitePlacementStore(db_path=tmp_path / "shared.db")
+    return TestClient(app)
+
+
+@pytest.fixture(autouse=True)
+def _reset_apply_rate_window() -> Iterator[None]:
+    """The apply rate cap is module-level process state (the D-019
+    precedent); tests must not inherit each other's windows — the P2
+    withdraw route rides this limiter (G-313 applied BEFORE it
+    ships)."""
+    from ai_service.api.marketplace import _APPLY_TIMES
+
+    _APPLY_TIMES.clear()
+    yield
+    _APPLY_TIMES.clear()
+
+
+@pytest.fixture()
+def client(
+    tmp_path: Path,
+    identity_store: SQLiteIdentityStore,
+    marketplace_store: SQLiteMarketplaceStore,
+) -> Iterator[TestClient]:
+    with make_client(
+        tmp_path, identity_store, marketplace_store, admin_token=ADMIN_TOKEN
+    ) as c:
+        yield c
+
+
+def admin_headers(token: str = ADMIN_TOKEN) -> dict[str, str]:
+    return {"X-Admin-Token": token}

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

@@ -0,0 +1,585 @@
+"""Placement store tests (D-111, REQ-11-001) — the ledger's adversarial
+suite: chain verification, tamper, fork, concurrency, no-delete-path,
+the D-099 matrix, the D-102 revoke/re-hire lane.
+
+Real stores on tmp_path DBs, zero network by construction (no
+MockTransport needed — no external calls exist), real scrypt-free
+paths, honest assertions (behavior over bytes).
+"""
+
+from __future__ import annotations
+
+import sqlite3
+import uuid
+from datetime import UTC, date, datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+from sqlmodel import Session
+
+from ai_service.marketplace.base import Posting
+from ai_service.marketplace.store import (
+    ApplicationRow,
+    SQLiteMarketplaceStore,
+)
+from ai_service.placements.base import (
+    APPLICATION_STATUSES,
+    GENESIS_HASH,
+    TRANSITION_MATRIX,
+    InvalidTransition,
+    validate_transition,
+)
+from ai_service.placements.store import PlacementEventRow, SQLitePlacementStore
+
+# -- helpers -----------------------------------------------------------------
+
+
+def _seed_posting(store: SQLiteMarketplaceStore, employer_id: str = "emp-test") -> str:
+    """Insert a draft posting via the store's own path → its id."""
+    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",
+            required_competencies=["orchestration"],
+            skills=["python"],
+        )
+    )
+    return posting.id
+
+
+def _apply(store: SQLiteMarketplaceStore, posting_id: str, learner_id: str) -> ApplicationRow:
+    application, _created = store.insert_application(posting_id, learner_id)
+    with Session(store._engine) as session:
+        row = session.get(ApplicationRow, application.id)
+        assert row is not None
+        return row
+
+
+def _hire(
+    store: SQLitePlacementStore,
+    application_id: str,
+    actor: str = "operator",
+    now: datetime | None = None,
+) -> object:
+    return store.transition_application(
+        application_id,
+        "hired",
+        actor_type="operator",
+        actor_ref=actor,
+        role_title="AI Orchestration Engineer",
+        start_date=date(2026, 11, 1),
+        now=now,
+    )
+
+
+# -- D-099 matrix (base contracts) ---------------------------------------------
+
+
+def test_genesis_hash_pinned() -> None:
+    assert GENESIS_HASH == "0" * 64
+
+
+def test_transition_matrix_forward_only_with_skip_forward() -> None:
+    # forward legal
+    assert "in_review" in TRANSITION_MATRIX["submitted"]
+    # skip-forward legal (the fast-track hire — R-71)
+    assert "hired" in TRANSITION_MATRIX["submitted"]
+    assert "interview" in TRANSITION_MATRIX["submitted"]
+    # terminals empty
+    assert TRANSITION_MATRIX["hired"] == ()
+    assert TRANSITION_MATRIX["rejected"] == ()
+    assert TRANSITION_MATRIX["withdrawn"] == ()
+    # every status is a matrix key
+    for status in APPLICATION_STATUSES:
+        assert status in TRANSITION_MATRIX
+
+
+def test_validate_transition_arms() -> None:
+    validate_transition("submitted", "hired")  # skip-forward
+    validate_transition("in_review", "rejected")
+    with pytest.raises(InvalidTransition):
+        validate_transition("offer", "in_review")  # backward — illegal
+    with pytest.raises(InvalidTransition):
+        validate_transition("hired", "rejected")  # terminal-source
+    with pytest.raises(InvalidTransition):
+        validate_transition("submitted", "garbage")  # unknown target
+    with pytest.raises(InvalidTransition):
+        validate_transition("garbage", "hired")  # unknown source
+
+
+# -- schema pins ---------------------------------------------------------------
+
+
+def test_tables_exist_in_sqlite_master(placement_store: SQLitePlacementStore) -> None:
+    with sqlite3.connect(placement_store._db_path) as conn:
+        rows = {
+            name
+            for (name,) in conn.execute(
+                "SELECT name FROM sqlite_master WHERE type='table'"
+            )
+        }
+    assert "placement" in rows
+    assert "placement_event" in rows
+
+
+def test_partial_unique_index_pinned(placement_store: SQLitePlacementStore) -> None:
+    """G-316 shape: the partial UNIQUE index exists in sqlite_master
+    (the test_claims_store template)."""
+    with sqlite3.connect(placement_store._db_path) as conn:
+        rows = {
+            sql
+            for (sql,) in conn.execute(
+                "SELECT sql FROM sqlite_master WHERE type='index' AND name=?",
+                ("uq_active_placement_per_application",),
+            )
+        }
+    assert rows, "partial index missing"
+    sql = rows.pop()
+    assert "CREATE UNIQUE INDEX" in sql
+    assert "status='active'" in sql
+
+
+def test_no_delete_path_in_source() -> None:
+    """R-70: the module carries NO delete method and NO DELETE SQL —
+    deleting the top ledger row would make INTEGER PRIMARY KEY reuse
+    max+1 (chain-integrity hole)."""
+    from ai_service.placements import store as placements_store_module
+
+    source = Path(placements_store_module.__file__).read_text()
+    assert "DELETE FROM" not in source
+    assert "def delete" not in source
+    assert "session.delete" not in source
+
+
+# -- the transition + chain -----------------------------------------------------
+
+
+def test_transition_flow_creates_chained_events(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+
+    app1, placement_none = placement_store.transition_application(
+        application.id, "in_review", actor_type="employer", actor_ref="account:a1"
+    )
+    assert app1.status == "in_review"
+    assert placement_none is None
+
+    app2, _ = placement_store.transition_application(
+        application.id, "interview", actor_type="employer", actor_ref="account:a1"
+    )
+    assert app2.status == "interview"
+
+    app3, placement = placement_store.transition_application(
+        application.id,
+        "hired",
+        actor_type="employer",
+        actor_ref="account:a1",
+        role_title="AI Orchestration Engineer",
+        start_date=date(2026, 11, 1),
+    )
+    assert app3.status == "hired"
+    assert placement is not None
+    assert placement.status == "active"
+    assert placement.role_title == "AI Orchestration Engineer"
+    assert placement.start_date == "2026-11-01"
+    assert placement.learner_id == "pilot-learner"
+    assert placement.employer_id == "emp-test"
+
+    events = placement_store.get_application_events(application.id)
+    assert [e.event_type for e in events] == [
+        "application.in_review",
+        "application.interview",
+        "application.hired",
+        "placement.created",
+    ]
+    # the chain walks green end-to-end
+    ok, head = placement_store.verify_chain()
+    assert ok is True
+    assert head == events[-1].hash
+
+
+def test_hire_requires_role_title_and_start_date(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    with pytest.raises(ValueError, match="role_title"):
+        placement_store.transition_application(
+            application.id, "hired", actor_type="employer", actor_ref="account:a1"
+        )
+    with pytest.raises(ValueError, match="start_date"):
+        placement_store.transition_application(
+            application.id,
+            "hired",
+            actor_type="employer",
+            actor_ref="account:a1",
+            role_title="Engineer",
+        )
+
+
+def test_matrix_illegal_moves_raise(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    placement_store.transition_application(
+        application.id, "offer", actor_type="employer", actor_ref="account:a1"
+    )
+    # backward — 409 class
+    with pytest.raises(InvalidTransition):
+        placement_store.transition_application(
+            application.id, "in_review", actor_type="employer", actor_ref="account:a1"
+        )
+    # terminal-source after hire
+    placement_store.transition_application(
+        application.id,
+        "hired",
+        actor_type="employer",
+        actor_ref="account:a1",
+        role_title="Engineer",
+        start_date=date(2026, 12, 1),
+    )
+    with pytest.raises(InvalidTransition):
+        placement_store.transition_application(
+            application.id, "rejected", actor_type="employer", actor_ref="account:a1"
+        )
+    # unknown application
+    with pytest.raises(InvalidTransition):
+        placement_store.transition_application(
+            "app-nonexistent", "in_review", actor_type="operator", actor_ref="operator"
+        )
+
+
+def test_unknown_status_rejected(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    with pytest.raises(InvalidTransition):
+        placement_store.transition_application(
+            application.id, "garbage", actor_type="operator", actor_ref="operator"
+        )
+
+
+def test_tz_round_trip_canonical_stable(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    """R-79: naive stored → _as_utc read → canonical isoformat stable —
+    the hashing path and the reading path share canonical_json()."""
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    moment = datetime(2026, 10, 1, 12, 0, 0, tzinfo=UTC)
+    placement_store.transition_application(
+        application.id,
+        "in_review",
+        actor_type="employer",
+        actor_ref="account:a1",
+        now=moment,
+    )
+    events = placement_store.list_events()
+    assert len(events) == 1
+    raw = events[0]
+    # the canonical form is a pure function of the row — recompute twice
+    assert raw.canonical_json() == raw.canonical_json()
+    ok, _ = placement_store.verify_chain()
+    assert ok is True
+
+
+def test_aware_non_utc_now_converted_chain_stays_green(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    """P0-1 (security review): an aware NON-UTC `now` CONVERTS to the
+    true instant — the append-time and read-back canonical forms are
+    identical, so the chain walks green (a re-labeled offset would
+    hash 12:00-04:00 and re-read 12:00+00:00 — permanently broken)."""
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    edt = timezone(timedelta(hours=-4))
+    placement_store.transition_application(
+        application.id,
+        "in_review",
+        actor_type="employer",
+        actor_ref="account:a1",
+        now=datetime(2026, 10, 1, 12, 0, 0, tzinfo=edt),
+    )
+    ok, _ = placement_store.verify_chain()
+    assert ok is True
+
+
+def test_actor_type_vocabulary_enforced(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    """P1-1 (security review): the store boundary rejects an actor_type
+    outside the closed vocabulary — a bad label would be frozen in
+    immutable history."""
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    with pytest.raises(ValueError, match="actor_type"):
+        placement_store.transition_application(
+            application.id,
+            "in_review",
+            actor_type="ghost-actor",
+            actor_ref="account:a1",
+        )
+
+
+def test_payload_floats_rejected(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    """R-69: payload values str/int/bool/None only — floats REJECTED."""
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    session = Session(placement_store._engine)
+    try:
+        row = PlacementEventRow(
+            application_id=application.id,
+            placement_id=None,
+            event_type="application.in_review",
+            actor_type="employer",
+            actor_ref="account:a1",
+            payload={"ratio": 0.5},
+            prev_hash=GENESIS_HASH,
+            hash="",
+            created_at=datetime.now(UTC),
+        )
+        with pytest.raises(ValueError, match="str/int/bool/None"):
+            row._validate()
+    finally:
+        session.close()
+
+
+# -- tamper / fork / concurrency (the adversarial arms, D-111) ------------------
+
+
+def test_tamper_detected_via_second_engine(
+    tmp_path: Path,
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    """Direct row mutation through a SECOND engine on the same file →
+    the verification walk FAILS (R-78: the mutation commits before the
+    walk reads — WAL snapshot isolation)."""
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    placement_store.transition_application(
+        application.id, "in_review", actor_type="employer", actor_ref="account:a1"
+    )
+    ok_before, _ = placement_store.verify_chain()
+    assert ok_before is True
+
+    with sqlite3.connect(placement_store._db_path) as conn:
+        conn.execute(
+            "UPDATE placement_event SET actor_ref = 'account:evil' WHERE id = 1"
+        )
+        conn.commit()
+
+    ok_after, _ = placement_store.verify_chain()
+    assert ok_after is False
+
+
+def test_fork_prev_hash_surfaces_integrity_error(
+    placement_store: SQLitePlacementStore,
+) -> None:
+    """G-316 backstop class: a second row claiming the same prev_hash
+    → UNIQUE refuses it, IntegrityError SURFACES."""
+    with pytest.raises(IntegrityError):
+        with Session(placement_store._engine) as session:
+            for _i in range(2):
+                row = PlacementEventRow(
+                    application_id="app-x",
+                    placement_id=None,
+                    event_type="application.in_review",
+                    actor_type="employer",
+                    actor_ref="account:a1",
+                    payload={},
+                    prev_hash=GENESIS_HASH,
+                    hash="",
+                    created_at=datetime.now(UTC),
+                )
+                session.add(row)
+                session.flush()
+                row.hash = "computed-later"
+                session.flush()
+            session.commit()
+
+
+def test_concurrent_transition_one_winner(
+    marketplace_store: SQLiteMarketplaceStore,
+    tmp_path: Path,
+) -> None:
+    """R-78: two stores (separate engines) race the same application —
+    one winner commits; the loser surfaces an honest error (never two
+    events on one prev_hash)."""
+    db_path = tmp_path / "shared.db"
+    store_a = SQLitePlacementStore(db_path=db_path)
+    store_b = SQLitePlacementStore(db_path=db_path)
+    marketplace = SQLiteMarketplaceStore(db_path=db_path)
+    try:
+        posting_id = _seed_posting(marketplace)
+        application = _apply(marketplace, posting_id, "pilot-learner")
+        # both read the same submitted status; A commits first
+        store_a.transition_application(
+            application.id, "in_review", actor_type="employer", actor_ref="account:a"
+        )
+        with pytest.raises((InvalidTransition, IntegrityError)):
+            store_b.transition_application(
+                application.id, "in_review", actor_type="employer", actor_ref="account:b"
+            )
+        ok, _ = store_a.verify_chain()
+        assert ok is True
+        # exactly ONE in_review event exists
+        events = store_a.get_application_events(application.id)
+        assert [e.event_type for e in events] == ["application.in_review"]
+    finally:
+        store_a.close()
+        store_b.close()
+        marketplace.close()
+
+
+# -- the D-102 correction lane ---------------------------------------------------
+
+
+def test_revoke_and_rehire(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    _, placement = placement_store.transition_application(
+        application.id,
+        "hired",
+        actor_type="employer",
+        actor_ref="account:a1",
+        role_title="Engineer",
+        start_date=date(2026, 11, 1),
+    )
+    assert placement is not None
+
+    # revoke: active → revoked + event
+    revoked = placement_store.revoke_placement(placement.id, "operator")
+    assert revoked is not None and revoked.status == "revoked"
+
+    # rehire after revoke: a FRESH active row (R-80) — decided_at =
+    # the re-hire event's stamp (G-326)
+    moment = datetime(2026, 10, 2, 9, 30, 0, tzinfo=UTC)
+    rehired = placement_store.rehire_placement(
+        application.id,
+        role_title="Engineer II",
+        start_date=date(2026, 12, 15),
+        actor_ref="operator",
+        now=moment,
+    )
+    assert rehired.status == "active"
+    assert rehired.id != placement.id
+    assert rehired.decided_at == moment
+
+    # the three-event ledger sequence: hired → revoked → rehired
+    events = placement_store.get_application_events(application.id)
+    assert [e.event_type for e in events] == [
+        "application.hired",
+        "placement.created",
+        "placement.revoked",
+        "placement.rehired",
+    ]
+
+    # the revoked placement stays in history (D-102)
+    statuses = {p.status for p in placement_store.list_placements()}
+    assert statuses == {"active", "revoked"}
+
+    ok, _ = placement_store.verify_chain()
+    assert ok is True
+
+
+def test_revoke_only_active(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    _, placement = placement_store.transition_application(
+        application.id,
+        "hired",
+        actor_type="employer",
+        actor_ref="account:a1",
+        role_title="Engineer",
+        start_date=date(2026, 11, 1),
+    )
+    assert placement is not None
+    placement_store.revoke_placement(placement.id, "operator")
+    with pytest.raises(InvalidTransition):
+        placement_store.revoke_placement(placement.id, "operator")
+    assert placement_store.revoke_placement("plc-nonexistent", "operator") is None
+
+
+def test_rehire_requires_hired_application(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    # not hired yet → re-hire is the correction lane for a hired app only
+    with pytest.raises(InvalidTransition):
+        placement_store.rehire_placement(
+            application.id,
+            role_title="Engineer",
+            start_date=date(2026, 11, 1),
+            actor_ref="operator",
+        )
+
+
+def test_rehire_second_active_surfaces_integrity_error(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    """The partial-unique arm: re-hire WITHOUT revoke → the partial
+    UNIQUE index refuses the second ACTIVE row (the caller's honest
+    409)."""
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    placement_store.transition_application(
+        application.id,
+        "hired",
+        actor_type="employer",
+        actor_ref="account:a1",
+        role_title="Engineer",
+        start_date=date(2026, 11, 1),
+    )
+    with pytest.raises(IntegrityError):
+        placement_store.rehire_placement(
+            application.id,
+            role_title="Engineer II",
+            start_date=date(2026, 12, 1),
+            actor_ref="operator",
+        )
+
+
+def test_chain_walk_after_n_appends(
+    marketplace_store: SQLiteMarketplaceStore,
+    placement_store: SQLitePlacementStore,
+) -> None:
+    posting_id = _seed_posting(marketplace_store)
+    application = _apply(marketplace_store, posting_id, "pilot-learner")
+    for to_status in ("in_review", "interview", "offer", "rejected"):
+        placement_store.transition_application(
+            application.id, to_status, actor_type="employer", actor_ref="account:a1"
+        )
+    events = placement_store.list_events()
+    assert len(events) == 4
+    ok, head = placement_store.verify_chain()
+    assert ok is True
+    assert head == events[-1].hash