feat(P01): complete placement-store-ledger phase (REQ-11-001)

Phase 1 complete: the 9th D-027 store (PlacementRow partial-unique
active guard + PlacementEventRow global sha256 hash chain with the
UNIQUE(prev_hash) fork backstop), the D-099 matrix + event
vocabulary, the ONE transactional transition_application (D-100,
flush-then-hash-then-commit), main.py/deps wiring, the 22-test
adversarial suite (tamper/fork/concurrency/no-delete-path), the
security review fixes (tz conversion, actor vocabulary, conftest
topology), verify four-layers PASS. Ships as v0.10.1.

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

.ciagent/CHECKPOINT.json

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

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

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

@@ -346,6 +346,13 @@ 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(
         self, posting_id: str | None = None
     ) -> list[Application]:
@@ -900,6 +907,15 @@ 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(
         self, posting_id: str | None = None
     ) -> list[Application]:

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

@@ -0,0 +1,184 @@
+"""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

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_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="test",
+            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