diff --git a/pyproject.toml b/pyproject.toml
index e8d7585..62d286b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -13,6 +13,9 @@ dependencies = [
     "pydantic>=2.10,<3",
 ]
 
+[project.scripts]
+ofw = "ofw.cli:main"
+
 [project.optional-dependencies]
 dev = [
     "mypy>=1.10,<2",
diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py
index 0fa057c..6fc16cd 100644
--- a/src/ofw/__init__.py
+++ b/src/ofw/__init__.py
@@ -1,5 +1,6 @@
 """Public OpenFlyWheel harness API."""
 
+from datetime import datetime
 from pathlib import Path
 from threading import Event
 
@@ -125,6 +126,36 @@
     read_trace_observations,
     search_observation_content,
 )
+from ofw.promotion import (
+    ApprovalDecision,
+    ApprovalRecord,
+    ApproverId,
+    DeploymentAdapter,
+    DeploymentReference,
+    DeploymentRequest,
+    GitHubCliPublisher,
+    GitRemote,
+    PromotionBranch,
+    PromotionError,
+    PromotionErrorCode,
+    PromotionEvent,
+    PromotionEventKind,
+    PromotionJobHandler,
+    PromotionMarker,
+    PromotionMode,
+    PromotionPolicy,
+    PromotionRequest,
+    PromotionRequestResolver,
+    PromotionResult,
+    PullRequestDraft,
+    PullRequestId,
+    PullRequestPublisher,
+    PullRequestReference,
+    RollbackPlan,
+)
+from ofw.promotion import (
+    PromotionService as GitPromotionService,
+)
 from ofw.runtime import (
     CanaryCase,
     CaseId,
@@ -192,6 +223,7 @@
 
 AutomationPolicy = SchedulerAutomationPolicy
 LocalScheduler = SQLiteScheduler
+PromotionService = GitPromotionService
 
 
 class _OfwNamespace:
@@ -226,6 +258,8 @@ class _OfwNamespace:
     Money = Money
     QuietHours = QuietHours
     StageBudgets = StageBudgets
+    PromotionPolicy = PromotionPolicy
+    PromotionService = GitPromotionService
 
     def editable(self, path: Path) -> EditableFile:
         return editable(path)
@@ -261,6 +295,16 @@ def serve(
         finally:
             scheduler.close()
 
+    def promote(
+        self,
+        request: PromotionRequest,
+        *,
+        now: datetime,
+        pull_requests: PullRequestPublisher | None = None,
+        deployments: DeploymentAdapter | None = None,
+    ) -> PromotionResult:
+        return GitPromotionService(pull_requests, deployments).run(request, now)
+
     def search_observation_content(
         self,
         collection: CollectionResult,
@@ -295,6 +339,9 @@ def read_snapshot_content(
 
 __all__ = [
     "AssetAccess",
+    "ApprovalDecision",
+    "ApprovalRecord",
+    "ApproverId",
     "AutomationPolicy",
     "Baseline",
     "BenchmarkError",
@@ -337,6 +384,9 @@ def read_snapshot_content(
     "DiagnosisErrorCode",
     "Dependency",
     "DependencyMode",
+    "DeploymentAdapter",
+    "DeploymentReference",
+    "DeploymentRequest",
     "EditableFile",
     "EvidenceAnchor",
     "EvidenceAnchorKind",
@@ -354,6 +404,8 @@ def read_snapshot_content(
     "FitPolicy",
     "FitResult",
     "GitCommit",
+    "GitHubCliPublisher",
+    "GitRemote",
     "Harness",
     "HarnessAsset",
     "HarnessComponent",
@@ -397,6 +449,23 @@ def read_snapshot_content(
     "ProcessCommand",
     "ProcessLimits",
     "PrivacyTransform",
+    "PromotionBranch",
+    "PromotionError",
+    "PromotionErrorCode",
+    "PromotionEvent",
+    "PromotionEventKind",
+    "PromotionJobHandler",
+    "PromotionMarker",
+    "PromotionMode",
+    "PromotionPolicy",
+    "PromotionRequest",
+    "PromotionRequestResolver",
+    "PromotionResult",
+    "PromotionService",
+    "PullRequestDraft",
+    "PullRequestId",
+    "PullRequestPublisher",
+    "PullRequestReference",
     "PythonEntrypoint",
     "PythonLoop",
     "PythonDiagnoser",
@@ -404,6 +473,7 @@ def read_snapshot_content(
     "QuietHours",
     "ReconcileReport",
     "ResultId",
+    "RollbackPlan",
     "RunErrorCode",
     "RunResult",
     "RunStatus",
diff --git a/src/ofw/cli.py b/src/ofw/cli.py
new file mode 100644
index 0000000..e29b4cc
--- /dev/null
+++ b/src/ofw/cli.py
@@ -0,0 +1,62 @@
+"""Dependency-free operator CLI over the scheduler application service."""
+
+from __future__ import annotations
+
+import sys
+from datetime import UTC, datetime
+from enum import StrEnum
+from pathlib import Path
+
+from ofw.scheduler import (
+    JobId,
+    LocalScheduler,
+    ScheduledJob,
+    SchedulerError,
+    SchedulerErrorCode,
+    read_automation_policy,
+)
+
+
+class CampaignCommand(StrEnum):
+    STATUS = "status"
+    CANCEL = "cancel"
+    RESUME = "resume"
+
+
+def run_campaign_command(arguments: tuple[str, ...], now: datetime) -> ScheduledJob:
+    if len(arguments) != 5 or arguments[0] != "campaign":
+        raise SchedulerError(
+            SchedulerErrorCode.INVALID_TRANSITION,
+            "usage: ofw campaign status|cancel|resume STORE POLICY JOB_ID",
+        )
+    try:
+        command = CampaignCommand(arguments[1])
+    except ValueError as error:
+        raise SchedulerError(
+            SchedulerErrorCode.INVALID_TRANSITION,
+            arguments[1],
+        ) from error
+    store_path = Path(arguments[2])
+    policy = read_automation_policy(Path(arguments[3]))
+    job_id = JobId(arguments[4])
+    scheduler = LocalScheduler(store_path, policy)
+    try:
+        match command:
+            case CampaignCommand.STATUS:
+                return scheduler.job(job_id)
+            case CampaignCommand.CANCEL:
+                return scheduler.cancel(job_id, now)
+            case CampaignCommand.RESUME:
+                return scheduler.resume(job_id, now)
+    finally:
+        scheduler.close()
+
+
+def main() -> int:
+    try:
+        result = run_campaign_command(tuple(sys.argv[1:]), datetime.now(UTC))
+    except SchedulerError as error:
+        print(str(error), file=sys.stderr)
+        return 1
+    print(result.to_json())
+    return 0
diff --git a/src/ofw/promotion.py b/src/ofw/promotion.py
new file mode 100644
index 0000000..67b7df5
--- /dev/null
+++ b/src/ofw/promotion.py
@@ -0,0 +1,920 @@
+"""Reviewable Git promotion with approval, audit, and rollback artifacts."""
+
+from __future__ import annotations
+
+import hashlib
+import html
+import re
+import shutil
+import subprocess  # nosec B404
+from dataclasses import dataclass
+from datetime import datetime
+from enum import StrEnum
+from pathlib import Path
+from typing import Protocol
+
+from pydantic import TypeAdapter, ValidationError
+
+from ofw.candidate import (
+    CandidateBuild,
+    CandidateError,
+    CandidateId,
+    validate_candidate_artifacts,
+    validate_candidate_revision,
+)
+from ofw.contracts import GitCommit, Sha256Digest
+from ofw.fit import FitCampaign, FitResult
+from ofw.mine import digest_bytes, write_artifact
+from ofw.scheduler import (
+    FailureDisposition,
+    JobContext,
+    JobExecution,
+    JobExecutionError,
+    JobKind,
+    JobResult,
+    JobSpec,
+    Money,
+    ResultId,
+    SchedulerErrorCode,
+)
+
+
+class PromotionMode(StrEnum):
+    NONE = "none"
+    COMMIT = "commit"
+    PULL_REQUEST = "pull_request"
+    DEPLOY = "deploy"
+
+
+class ApprovalDecision(StrEnum):
+    APPROVED = "approved"
+    REJECTED = "rejected"
+
+
+class PromotionEventKind(StrEnum):
+    REPORT_WRITTEN = "report_written"
+    COMMIT_CREATED = "commit_created"
+    BRANCH_PUSHED = "branch_pushed"
+    PULL_REQUEST_OPENED = "pull_request_opened"
+    APPROVAL_VERIFIED = "approval_verified"
+    DEPLOYED = "deployed"
+    COMPLETED = "completed"
+
+
+class PromotionErrorCode(StrEnum):
+    NO_WINNER = "no_winner"
+    WINNER_MISMATCH = "winner_mismatch"
+    FIT_RESULT_INVALID = "fit_result_invalid"
+    CANDIDATE_DRIFT = "candidate_drift"
+    POLICY_INVALID = "policy_invalid"
+    APPROVAL_REQUIRED = "approval_required"
+    APPROVAL_REJECTED = "approval_rejected"
+    APPROVAL_INVALID = "approval_invalid"
+    CANCELLED = "cancelled"
+    ALREADY_COMPLETED = "already_completed"
+    GIT_FAILED = "git_failed"
+    COMMIT_INVALID = "commit_invalid"
+    PUBLISHER_REQUIRED = "publisher_required"
+    PUBLISH_FAILED = "publish_failed"
+    DEPLOYMENT_REQUIRED = "deployment_required"
+    DEPLOYMENT_FAILED = "deployment_failed"
+    RESULT_INVALID = "result_invalid"
+
+
+class PromotionError(Exception):
+    __slots__ = ("code", "subject")
+
+    def __init__(self, code: PromotionErrorCode, subject: str) -> None:
+        self.code = code
+        self.subject = subject
+        super().__init__(f"{code.value}: {subject}")
+
+
+@dataclass(frozen=True, slots=True)
+class ApproverId:
+    value: str
+
+    def __post_init__(self) -> None:
+        if not self.value.strip():
+            raise ValueError("approver cannot be empty")
+
+    def __str__(self) -> str:
+        return self.value
+
+
+@dataclass(frozen=True, slots=True)
+class PromotionMarker:
+    value: str
+
+    def __str__(self) -> str:
+        return self.value
+
+
+@dataclass(frozen=True, slots=True)
+class PromotionBranch:
+    value: str
+
+    def __str__(self) -> str:
+        return self.value
+
+
+@dataclass(frozen=True, slots=True)
+class GitRemote:
+    name: str
+    base_branch: str
+    branch_prefix: str
+
+    def __post_init__(self) -> None:
+        for value in (self.name, self.base_branch, self.branch_prefix):
+            if not _valid_git_name(value):
+                raise PromotionError(PromotionErrorCode.POLICY_INVALID, value)
+
+
+@dataclass(frozen=True, slots=True)
+class PromotionPolicy:
+    mode: PromotionMode
+    remote: GitRemote | None
+    require_approval: bool
+
+    def __post_init__(self) -> None:
+        if (
+            (self.mode is PromotionMode.PULL_REQUEST and self.remote is None)
+            or (self.mode is not PromotionMode.PULL_REQUEST and self.remote is not None)
+            or (self.mode is PromotionMode.DEPLOY and not self.require_approval)
+        ):
+            raise PromotionError(PromotionErrorCode.POLICY_INVALID, self.mode.value)
+
+    @property
+    def digest(self) -> Sha256Digest:
+        remote = self.remote
+        payload = "\0".join(
+            (
+                self.mode.value,
+                str(self.require_approval),
+                "" if remote is None else remote.name,
+                "" if remote is None else remote.base_branch,
+                "" if remote is None else remote.branch_prefix,
+            )
+        )
+        return digest_bytes(payload.encode())
+
+
+@dataclass(frozen=True, slots=True)
+class ApprovalRecord:
+    fit_result_id: str
+    candidate_id: CandidateId
+    policy_digest: Sha256Digest
+    approver: ApproverId
+    decision: ApprovalDecision
+    decided_at: datetime
+
+    def __post_init__(self) -> None:
+        _aware(self.decided_at)
+
+    @property
+    def digest(self) -> Sha256Digest:
+        return digest_bytes(_APPROVAL_ADAPTER.dump_json(self))
+
+
+@dataclass(frozen=True, slots=True)
+class PullRequestId:
+    value: int
+
+
+@dataclass(frozen=True, slots=True)
+class PullRequestDraft:
+    marker: PromotionMarker
+    title: str
+    body: str
+    head_branch: PromotionBranch
+    base_branch: str
+    commit: GitCommit
+
+
+@dataclass(frozen=True, slots=True)
+class PullRequestReference:
+    id: PullRequestId
+    url: str
+    marker: PromotionMarker
+    head_branch: PromotionBranch
+
+
+class PullRequestPublisher(Protocol):
+    def find(self, marker: PromotionMarker) -> PullRequestReference | None: ...
+
+    def open(self, draft: PullRequestDraft) -> PullRequestReference: ...
+
+
+@dataclass(frozen=True, slots=True)
+class DeploymentRequest:
+    marker: PromotionMarker
+    fit_result_id: str
+    candidate_id: CandidateId
+    commit: GitCommit
+    approval: ApprovalRecord
+
+
+@dataclass(frozen=True, slots=True)
+class DeploymentReference:
+    id: str
+    marker: PromotionMarker
+    rollback_instruction: str
+
+
+class DeploymentAdapter(Protocol):
+    def find(self, marker: PromotionMarker) -> DeploymentReference | None: ...
+
+    def deploy(self, request: DeploymentRequest) -> DeploymentReference: ...
+
+
+class PromotionRequestResolver(Protocol):
+    def resolve(self, fit_result_id: ResultId) -> PromotionRequest: ...
+
+
+@dataclass(frozen=True, slots=True)
+class RollbackPlan:
+    reverse_patch: Path
+    command: tuple[str, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class PromotionEvent:
+    kind: PromotionEventKind
+    occurred_at: datetime
+    reference: str
+
+
+@dataclass(frozen=True, slots=True)
+class PromotionResult:
+    id: str
+    mode: PromotionMode
+    fit_result_id: str
+    candidate_id: CandidateId
+    policy_digest: Sha256Digest
+    approval_digest: Sha256Digest | None
+    branch: PromotionBranch | None
+    commit: GitCommit | None
+    pull_request: PullRequestReference | None
+    deployment: DeploymentReference | None
+    rollback: RollbackPlan
+    report_path: Path
+    report_digest: Sha256Digest
+    reverse_digest: Sha256Digest
+    events: tuple[PromotionEvent, ...]
+    root: Path
+
+    @property
+    def manifest_path(self) -> Path:
+        return self.root / ".ofw" / "promotions" / self.id / "manifest.json"
+
+    @property
+    def digest_path(self) -> Path:
+        return self.manifest_path.with_suffix(".sha256")
+
+    def to_json(self) -> str:
+        return _RESULT_ADAPTER.dump_json(self).decode()
+
+
+@dataclass(frozen=True, slots=True)
+class PromotionRequest:
+    campaign: FitCampaign
+    fit_result: FitResult
+    candidate: CandidateBuild
+    policy: PromotionPolicy
+    approval: ApprovalRecord | None
+
+    @property
+    def id(self) -> str:
+        payload = "\0".join(
+            (
+                self.fit_result.id,
+                str(self.fit_result.input_digest),
+                self.candidate.candidate.id.value,
+                str(self.policy.digest),
+            )
+        )
+        return f"promotion_{hashlib.sha256(payload.encode()).hexdigest()}"
+
+    @property
+    def marker(self) -> PromotionMarker:
+        return PromotionMarker(f"ofw-promotion:{self.id}")
+
+    @property
+    def root(self) -> Path:
+        return self.fit_result.root / ".ofw" / "promotions" / self.id
+
+
+@dataclass(frozen=True, slots=True)
+class CancellationRecord:
+    promotion_id: str
+    actor: ApproverId
+    cancelled_at: datetime
+
+    def to_json(self) -> str:
+        return _CANCELLATION_ADAPTER.dump_json(self).decode()
+
+
+@dataclass(frozen=True, slots=True)
+class _GitHubPullRequest:
+    number: int
+    url: str
+    headRefName: str
+    body: str
+
+
+_APPROVAL_ADAPTER: TypeAdapter[ApprovalRecord] = TypeAdapter(ApprovalRecord)
+_RESULT_ADAPTER: TypeAdapter[PromotionResult] = TypeAdapter(PromotionResult)
+_CANCELLATION_ADAPTER: TypeAdapter[CancellationRecord] = TypeAdapter(CancellationRecord)
+_DIGEST_ADAPTER: TypeAdapter[Sha256Digest] = TypeAdapter(Sha256Digest)
+_GITHUB_ADAPTER: TypeAdapter[tuple[_GitHubPullRequest, ...]] = TypeAdapter(
+    tuple[_GitHubPullRequest, ...]
+)
+
+
+@dataclass(frozen=True, slots=True)
+class GitHubCliPublisher:
+    repository: Path
+
+    def find(self, marker: PromotionMarker) -> PullRequestReference | None:
+        result = _command(
+            self.repository,
+            "gh",
+            "pr",
+            "list",
+            "--state",
+            "all",
+            "--search",
+            marker.value,
+            "--json",
+            "number,url,headRefName,body",
+        )
+        try:
+            records = _GITHUB_ADAPTER.validate_json(result.stdout)
+        except ValidationError as error:
+            raise PromotionError(PromotionErrorCode.PUBLISH_FAILED, marker.value) from error
+        record = next((item for item in records if marker.value in item.body), None)
+        if record is None:
+            return None
+        return PullRequestReference(
+            PullRequestId(record.number),
+            record.url,
+            marker,
+            PromotionBranch(record.headRefName),
+        )
+
+    def open(self, draft: PullRequestDraft) -> PullRequestReference:
+        _command(
+            self.repository,
+            "gh",
+            "pr",
+            "create",
+            "--base",
+            draft.base_branch,
+            "--head",
+            draft.head_branch.value,
+            "--title",
+            draft.title,
+            "--body",
+            draft.body,
+        )
+        reference = self.find(draft.marker)
+        if reference is None:
+            raise PromotionError(PromotionErrorCode.PUBLISH_FAILED, draft.marker.value)
+        return reference
+
+
+@dataclass(frozen=True, slots=True)
+class PromotionService:
+    pull_requests: PullRequestPublisher | None
+    deployments: DeploymentAdapter | None
+
+    def run(self, request: PromotionRequest, now: datetime) -> PromotionResult:
+        now = _aware(now)
+        existing = self._read_existing(request)
+        if existing is not None:
+            return existing
+        self._check_cancelled(request)
+        fit_result = request.campaign.run()
+        if fit_result != request.fit_result:
+            raise PromotionError(PromotionErrorCode.FIT_RESULT_INVALID, request.fit_result.id)
+        if fit_result.winner_id is None:
+            raise PromotionError(PromotionErrorCode.NO_WINNER, fit_result.id)
+        if fit_result.winner_id != request.candidate.candidate.id:
+            raise PromotionError(
+                PromotionErrorCode.WINNER_MISMATCH,
+                request.candidate.candidate.id.value,
+            )
+        self._validate_approval(request)
+        report_path = request.root / "report.html"
+        report_payload = _report(request).encode()
+        write_artifact(report_path, report_payload)
+        events: tuple[PromotionEvent, ...] = (
+            PromotionEvent(PromotionEventKind.REPORT_WRITTEN, now, str(report_path)),
+        )
+        commit: GitCommit | None = None
+        branch: PromotionBranch | None = None
+        pull_request: PullRequestReference | None = None
+        deployment: DeploymentReference | None = None
+        reverse_path = request.root / "reverse.patch"
+        reverse = b""
+        if request.policy.mode is not PromotionMode.NONE:
+            self._check_cancelled(request)
+            commit, branch = _ensure_commit(request)
+            reverse = _git_bytes(
+                request.candidate.workspace.source_root,
+                "show",
+                "-R",
+                "--binary",
+                "--format=",
+                commit.value,
+            )
+            events = (*events, PromotionEvent(PromotionEventKind.COMMIT_CREATED, now, commit.value))
+        write_artifact(reverse_path, reverse)
+        if request.policy.mode is PromotionMode.PULL_REQUEST:
+            self._check_cancelled(request)
+            if self.pull_requests is None or request.policy.remote is None:
+                raise PromotionError(PromotionErrorCode.PUBLISHER_REQUIRED, request.id)
+            if commit is None or branch is None:
+                raise PromotionError(PromotionErrorCode.COMMIT_INVALID, request.id)
+            _push(request.candidate.workspace.source_root, request.policy.remote, branch)
+            events = (*events, PromotionEvent(PromotionEventKind.BRANCH_PUSHED, now, branch.value))
+            pull_request = self.pull_requests.find(request.marker)
+            if pull_request is None:
+                pull_request = self.pull_requests.open(
+                    PullRequestDraft(
+                        request.marker,
+                        f"[ofw] promote {request.candidate.candidate.id.value[-12:]}",
+                        _pull_request_body(request),
+                        branch,
+                        request.policy.remote.base_branch,
+                        commit,
+                    )
+                )
+            events = (
+                *events,
+                PromotionEvent(
+                    PromotionEventKind.PULL_REQUEST_OPENED,
+                    now,
+                    pull_request.url,
+                ),
+            )
+        if request.policy.mode is PromotionMode.DEPLOY:
+            self._check_cancelled(request)
+            approval = request.approval
+            if approval is None:
+                raise PromotionError(PromotionErrorCode.APPROVAL_REQUIRED, request.id)
+            if self.deployments is None or commit is None:
+                raise PromotionError(PromotionErrorCode.DEPLOYMENT_REQUIRED, request.id)
+            events = (
+                *events,
+                PromotionEvent(
+                    PromotionEventKind.APPROVAL_VERIFIED,
+                    now,
+                    str(approval.digest),
+                ),
+            )
+            deployment = self.deployments.find(request.marker)
+            if deployment is None:
+                deployment = self.deployments.deploy(
+                    DeploymentRequest(
+                        request.marker,
+                        fit_result.id,
+                        request.candidate.candidate.id,
+                        commit,
+                        approval,
+                    )
+                )
+            events = (
+                *events,
+                PromotionEvent(PromotionEventKind.DEPLOYED, now, deployment.id),
+            )
+        result = PromotionResult(
+            request.id,
+            request.policy.mode,
+            fit_result.id,
+            request.candidate.candidate.id,
+            request.policy.digest,
+            None if request.approval is None else request.approval.digest,
+            branch,
+            commit,
+            pull_request,
+            deployment,
+            RollbackPlan(
+                reverse_path,
+                () if commit is None else ("git", "revert", commit.value),
+            ),
+            report_path,
+            digest_bytes(report_payload),
+            digest_bytes(reverse),
+            (*events, PromotionEvent(PromotionEventKind.COMPLETED, now, request.id)),
+            request.fit_result.root,
+        )
+        payload = f"{result.to_json()}\n".encode()
+        write_artifact(result.manifest_path, payload)
+        write_artifact(result.digest_path, _DIGEST_ADAPTER.dump_json(digest_bytes(payload)) + b"\n")
+        return result
+
+    def cancel(
+        self,
+        request: PromotionRequest,
+        actor: ApproverId,
+        now: datetime,
+    ) -> CancellationRecord:
+        now = _aware(now)
+        if (request.root / "manifest.json").exists():
+            raise PromotionError(PromotionErrorCode.ALREADY_COMPLETED, request.id)
+        record = CancellationRecord(request.id, actor, now)
+        write_artifact(request.root / "cancelled.json", f"{record.to_json()}\n".encode())
+        return record
+
+    def _validate_approval(self, request: PromotionRequest) -> None:
+        approval = request.approval
+        if request.policy.require_approval and approval is None:
+            raise PromotionError(PromotionErrorCode.APPROVAL_REQUIRED, request.id)
+        if approval is None:
+            return
+        if approval.decision is ApprovalDecision.REJECTED:
+            raise PromotionError(PromotionErrorCode.APPROVAL_REJECTED, request.id)
+        if (
+            approval.fit_result_id != request.fit_result.id
+            or approval.candidate_id != request.candidate.candidate.id
+            or approval.policy_digest != request.policy.digest
+        ):
+            raise PromotionError(PromotionErrorCode.APPROVAL_INVALID, request.id)
+
+    def _check_cancelled(self, request: PromotionRequest) -> None:
+        path = request.root / "cancelled.json"
+        if not path.exists():
+            return
+        try:
+            record = _CANCELLATION_ADAPTER.validate_json(path.read_bytes())
+        except (OSError, ValidationError) as error:
+            raise PromotionError(PromotionErrorCode.RESULT_INVALID, str(path)) from error
+        if record.promotion_id != request.id:
+            raise PromotionError(PromotionErrorCode.RESULT_INVALID, str(path))
+        raise PromotionError(PromotionErrorCode.CANCELLED, request.id)
+
+    def _read_existing(self, request: PromotionRequest) -> PromotionResult | None:
+        path = request.root / "manifest.json"
+        if not path.exists():
+            return None
+        try:
+            payload = path.read_bytes()
+            expected = _DIGEST_ADAPTER.validate_json(path.with_suffix(".sha256").read_bytes())
+            result = _RESULT_ADAPTER.validate_json(payload)
+        except (OSError, ValidationError) as error:
+            raise PromotionError(PromotionErrorCode.RESULT_INVALID, str(path)) from error
+        if (
+            digest_bytes(payload) != expected
+            or result.id != request.id
+            or result.fit_result_id != request.fit_result.id
+            or result.candidate_id != request.candidate.candidate.id
+            or result.policy_digest != request.policy.digest
+        ):
+            raise PromotionError(PromotionErrorCode.RESULT_INVALID, str(path))
+        try:
+            report_digest = digest_bytes(result.report_path.read_bytes())
+            reverse_digest = digest_bytes(result.rollback.reverse_patch.read_bytes())
+        except OSError as error:
+            raise PromotionError(PromotionErrorCode.RESULT_INVALID, str(path)) from error
+        if report_digest != result.report_digest or reverse_digest != result.reverse_digest:
+            raise PromotionError(PromotionErrorCode.RESULT_INVALID, str(path))
+        if result.commit is not None and result.branch is not None:
+            _validate_commit(request, result.branch, result.commit)
+        return result
+
+
+@dataclass(frozen=True, slots=True)
+class PromotionJobHandler:
+    service: PromotionService
+    requests: PromotionRequestResolver
+
+    @property
+    def kind(self) -> JobKind:
+        return JobKind.PROMOTE
+
+    def execute(self, job: JobSpec, context: JobContext) -> JobExecution:
+        if job.kind is not JobKind.PROMOTE:
+            raise JobExecutionError(
+                FailureDisposition.TERMINAL,
+                SchedulerErrorCode.RESULT_INVALID,
+                Money(0),
+            )
+        fit_results = tuple(
+            predecessor.result
+            for predecessor in context.scheduler.predecessors(context.lease.job.id)
+            if predecessor.spec.kind is JobKind.FIT and predecessor.result is not None
+        )
+        if len(fit_results) != 1:
+            raise JobExecutionError(
+                FailureDisposition.TERMINAL,
+                SchedulerErrorCode.RESULT_INVALID,
+                Money(0),
+            )
+        fit_result = fit_results[0]
+        try:
+            promotion = self.service.run(self.requests.resolve(fit_result.id), context.now)
+        except PromotionError as error:
+            disposition = (
+                FailureDisposition.RETRYABLE
+                if error.code
+                in (
+                    PromotionErrorCode.GIT_FAILED,
+                    PromotionErrorCode.PUBLISH_FAILED,
+                    PromotionErrorCode.DEPLOYMENT_FAILED,
+                )
+                else FailureDisposition.TERMINAL
+            )
+            raise JobExecutionError(
+                disposition,
+                SchedulerErrorCode.HANDLER_FAILED,
+                Money(0),
+            ) from error
+        return JobExecution(
+            JobResult(
+                ResultId(promotion.id),
+                JobKind.PROMOTE,
+                job.revision_id,
+                fit_result.id,
+                None,
+                True,
+            ),
+            Money(0),
+        )
+
+
+def _ensure_commit(request: PromotionRequest) -> tuple[GitCommit, PromotionBranch]:
+    revision = request.campaign.harness.current_revision
+    if revision is None:
+        raise PromotionError(PromotionErrorCode.CANDIDATE_DRIFT, request.id)
+    try:
+        validate_candidate_artifacts(request.candidate.candidate, revision)
+        validate_candidate_revision(request.candidate.candidate, revision)
+    except CandidateError as error:
+        raise PromotionError(PromotionErrorCode.CANDIDATE_DRIFT, error.subject) from error
+    branch = _promotion_branch(request)
+    worktree = request.candidate.workspace.parent / f"promotion-{request.id[-16:]}"
+    existing = _branch_commit(request.candidate.workspace.source_root, branch)
+    if existing is not None and _commit_matches(request, branch, existing):
+        if worktree.exists():
+            _remove_promotion_worktree(request, worktree)
+        return existing, branch
+    if existing is not None and not worktree.exists():
+        _git(request.candidate.workspace.source_root, "branch", "-D", branch.value)
+    try:
+        if not worktree.exists():
+            _git(
+                request.candidate.workspace.source_root,
+                "worktree",
+                "add",
+                "-b",
+                branch.value,
+                str(worktree),
+                request.candidate.workspace.branch.value,
+            )
+        patch = request.candidate.candidate.diff_path.read_bytes()
+        actual = _git_bytes(worktree, "diff", "--binary", "--no-ext-diff", "HEAD", "--")
+        if not actual:
+            _git_with_input(worktree, patch, "apply", "--binary", "-")
+            actual = _git_bytes(
+                worktree,
+                "diff",
+                "--binary",
+                "--no-ext-diff",
+                "HEAD",
+                "--",
+            )
+        if actual != patch:
+            raise PromotionError(PromotionErrorCode.COMMIT_INVALID, request.id)
+        _git(
+            worktree,
+            "add",
+            "--",
+            *(path.as_posix() for path in request.candidate.candidate.changed_files),
+        )
+        staged = _git_bytes(
+            worktree,
+            "diff",
+            "--cached",
+            "--binary",
+            "--no-ext-diff",
+            "HEAD",
+            "--",
+        )
+        if staged != patch:
+            raise PromotionError(PromotionErrorCode.COMMIT_INVALID, request.id)
+        _git(
+            worktree,
+            "-c",
+            "user.name=OpenFlyWheel",
+            "-c",
+            "user.email=openflywheel@localhost",
+            "commit",
+            "-m",
+            _commit_message(request),
+        )
+        commit = GitCommit(_git_text(worktree, "rev-parse", "HEAD"))
+        _validate_commit(request, branch, commit)
+        return commit, branch
+    except OSError as error:
+        raise PromotionError(PromotionErrorCode.GIT_FAILED, str(worktree)) from error
+    finally:
+        if worktree.exists():
+            _remove_promotion_worktree(request, worktree)
+
+
+def _validate_commit(
+    request: PromotionRequest,
+    branch: PromotionBranch,
+    commit: GitCommit,
+) -> None:
+    if not _commit_matches(request, branch, commit):
+        raise PromotionError(PromotionErrorCode.COMMIT_INVALID, request.id)
+
+
+def _commit_matches(
+    request: PromotionRequest,
+    branch: PromotionBranch,
+    commit: GitCommit,
+) -> bool:
+    root = request.candidate.workspace.source_root
+    branch_commit = _git_text(root, "rev-parse", f"refs/heads/{branch.value}")
+    message = _git_text(root, "show", "-s", "--format=%B", commit.value)
+    patch = _git_bytes(
+        root,
+        "diff",
+        "--binary",
+        "--no-ext-diff",
+        f"{commit.value}^",
+        commit.value,
+        "--",
+    )
+    try:
+        expected = request.candidate.candidate.diff_path.read_bytes()
+    except OSError as error:
+        raise PromotionError(PromotionErrorCode.COMMIT_INVALID, request.id) from error
+    return branch_commit == commit.value and request.marker.value in message and patch == expected
+
+
+def _remove_promotion_worktree(request: PromotionRequest, worktree: Path) -> None:
+    _git(
+        request.candidate.workspace.source_root,
+        "worktree",
+        "remove",
+        "--force",
+        str(worktree),
+    )
+    shutil.rmtree(worktree, ignore_errors=True)
+
+
+def _branch_commit(root: Path, branch: PromotionBranch) -> GitCommit | None:
+    result = subprocess.run(  # nosec B603
+        (
+            "git",
+            "-C",
+            str(root),
+            "rev-parse",
+            "--verify",
+            "--quiet",
+            f"refs/heads/{branch.value}",
+        ),
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode == 1:
+        return None
+    if result.returncode != 0:
+        raise PromotionError(PromotionErrorCode.GIT_FAILED, branch.value)
+    return GitCommit(result.stdout.strip())
+
+
+def _push(root: Path, remote: GitRemote, branch: PromotionBranch) -> None:
+    _git_text(root, "remote", "get-url", remote.name)
+    _git(root, "push", remote.name, f"{branch.value}:refs/heads/{branch.value}")
+
+
+def _report(request: PromotionRequest) -> str:
+    candidate = request.candidate.candidate
+    try:
+        patch = candidate.diff_path.read_text(encoding="utf-8")
+    except OSError as error:
+        raise PromotionError(PromotionErrorCode.CANDIDATE_DRIFT, candidate.id.value) from error
+    outcomes = "".join(
+        "<tr>"
+        f"<td>{html.escape(outcome.candidate_id.value)}</td>"
+        f"<td>{html.escape(outcome.status.value)}</td>"
+        f"<td>{html.escape(outcome.reason.value)}</td>"
+        f"<td>{outcome.target_delta:.6f}</td>"
+        f"<td>{outcome.regression_score:.6f}</td>"
+        "</tr>"
+        for outcome in request.fit_result.outcomes
+    )
+    return (
+        '<!doctype html><html><head><meta charset="utf-8"><title>OFW promotion</title>'
+        "</head><body>"
+        f"<h1>Promotion {html.escape(request.id)}</h1>"
+        f"<p>Fit: {html.escape(request.fit_result.id)}</p>"
+        f"<p>Winner: {html.escape(candidate.id.value)}</p>"
+        f"<p>Policy: {html.escape(str(request.policy.digest))}</p>"
+        "<table><thead><tr><th>Candidate</th><th>Status</th><th>Reason</th>"
+        f"<th>Target delta</th><th>Regression</th></tr></thead><tbody>{outcomes}</tbody></table>"
+        f"<h2>Patch</h2><pre>{html.escape(patch)}</pre>"
+        "</body></html>"
+    )
+
+
+def _pull_request_body(request: PromotionRequest) -> str:
+    return (
+        f"{request.marker.value}\n\n"
+        f"Fit result: `{request.fit_result.id}`\n\n"
+        f"Candidate: `{request.candidate.candidate.id.value}`\n\n"
+        f"Policy: `{request.policy.digest}`\n\n"
+        "This PR is a review artifact. Merging it is not treated as a production deployment."
+    )
+
+
+def _promotion_branch(request: PromotionRequest) -> PromotionBranch:
+    remote = request.policy.remote
+    prefix = "ofw" if remote is None else remote.branch_prefix
+    return PromotionBranch(f"{prefix}/promotion-{request.id[-16:]}")
+
+
+def _commit_message(request: PromotionRequest) -> str:
+    return (
+        f"ofw: promote {request.candidate.candidate.id.value[-12:]}\n\n"
+        f"{request.marker.value}\n"
+        f"OFW-Fit-Result: {request.fit_result.id}\n"
+        f"OFW-Policy: {request.policy.digest}"
+    )
+
+
+def _valid_git_name(value: str) -> bool:
+    return (
+        bool(value)
+        and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]*", value) is not None
+        and ".." not in value
+        and not value.endswith("/")
+    )
+
+
+def _aware(value: datetime) -> datetime:
+    if value.tzinfo is None or value.utcoffset() is None:
+        raise ValueError("datetime must be timezone-aware")
+    return value
+
+
+def _command(root: Path, *arguments: str) -> subprocess.CompletedProcess[bytes]:
+    result = subprocess.run(  # nosec B603
+        arguments,
+        cwd=root,
+        check=False,
+        capture_output=True,
+    )
+    if result.returncode != 0:
+        raise PromotionError(PromotionErrorCode.PUBLISH_FAILED, arguments[0])
+    return result
+
+
+def _git(root: Path, *arguments: str) -> None:
+    result = subprocess.run(  # nosec B603
+        ("git", "-C", str(root), *arguments),
+        check=False,
+        capture_output=True,
+    )
+    if result.returncode != 0:
+        raise PromotionError(PromotionErrorCode.GIT_FAILED, arguments[0])
+
+
+def _git_text(root: Path, *arguments: str) -> str:
+    result = subprocess.run(  # nosec B603
+        ("git", "-C", str(root), *arguments),
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        raise PromotionError(PromotionErrorCode.GIT_FAILED, arguments[0])
+    return result.stdout.strip()
+
+
+def _git_bytes(root: Path, *arguments: str) -> bytes:
+    result = subprocess.run(  # nosec B603
+        ("git", "-C", str(root), *arguments),
+        check=False,
+        capture_output=True,
+    )
+    if result.returncode != 0:
+        raise PromotionError(PromotionErrorCode.GIT_FAILED, arguments[0])
+    return result.stdout
+
+
+def _git_with_input(root: Path, payload: bytes, *arguments: str) -> None:
+    result = subprocess.run(  # nosec B603
+        ("git", "-C", str(root), *arguments),
+        input=payload,
+        check=False,
+        capture_output=True,
+    )
+    if result.returncode != 0:
+        raise PromotionError(PromotionErrorCode.GIT_FAILED, arguments[0])
diff --git a/src/ofw/scheduler.py b/src/ofw/scheduler.py
index 0c34424..4b135ab 100644
--- a/src/ofw/scheduler.py
+++ b/src/ofw/scheduler.py
@@ -77,6 +77,7 @@ class BlockerCode(StrEnum):
     COOLDOWN = "cooldown"
     CIRCUIT_OPEN = "circuit_open"
     QUIET_HOURS = "quiet_hours"
+    NO_WINNER = "no_winner"
 
 
 class SchedulerErrorCode(StrEnum):
@@ -191,6 +192,7 @@ class StageBudgets:
     benchmark_export: Money
     memory: Money
     fit: Money
+    promotion: Money
 
     def for_kind(self, kind: JobKind) -> Money:
         match kind:
@@ -207,7 +209,7 @@ def for_kind(self, kind: JobKind) -> Money:
             case JobKind.FIT:
                 return self.fit
             case JobKind.PROMOTE | JobKind.POST_PROMOTION_MONITOR:
-                raise SchedulerError(SchedulerErrorCode.POLICY_MISMATCH, kind.value)
+                return self.promotion
 
 
 @dataclass(frozen=True, slots=True)
@@ -264,10 +266,14 @@ def digest(self) -> Sha256Digest:
                 str(self.stage_budgets.benchmark_export.micros),
                 str(self.stage_budgets.memory.micros),
                 str(self.stage_budgets.fit.micros),
+                str(self.stage_budgets.promotion.micros),
             )
         )
         return _digest(payload.encode())
 
+    def to_json(self) -> str:
+        return _POLICY_ADAPTER.dump_json(self).decode()
+
 
 @dataclass(frozen=True, slots=True)
 class JobSpec:
@@ -317,6 +323,9 @@ class ScheduledJob:
     created_at: datetime
     updated_at: datetime
 
+    def to_json(self) -> str:
+        return _JOB_ADAPTER.dump_json(self).decode()
+
 
 @dataclass(frozen=True, slots=True)
 class JobLease:
@@ -367,6 +376,7 @@ class ReconcileReport:
     recovered: tuple[JobId, ...]
     readied: tuple[JobId, ...]
     failed: tuple[JobId, ...]
+    skipped: tuple[JobId, ...]
     blockers: tuple[JobBlocker, ...]
 
 
@@ -400,6 +410,7 @@ class JobExecution:
 class JobContext:
     scheduler: LocalScheduler
     lease: JobLease
+    now: datetime
 
     def renew(self, now: datetime) -> JobLease:
         return self.scheduler.renew(self.lease, now)
@@ -440,9 +451,11 @@ class _Readiness:
     ready: bool
     blocker: BlockerCode | None = None
     failed: bool = False
+    skipped: bool = False
 
 
 _JOB_ADAPTER: TypeAdapter[ScheduledJob] = TypeAdapter(ScheduledJob)
+_POLICY_ADAPTER: TypeAdapter[AutomationPolicy] = TypeAdapter(AutomationPolicy)
 _ATTEMPT_ADAPTER: TypeAdapter[JobAttempt] = TypeAdapter(JobAttempt)
 _REVISION_ADAPTER: TypeAdapter[RevisionAutomationState] = TypeAdapter(RevisionAutomationState)
 _IDENTITY_ADAPTER: TypeAdapter[_JobIdentity] = TypeAdapter(_JobIdentity)
@@ -652,11 +665,21 @@ def attempts(self, job_id: JobId) -> tuple[JobAttempt, ...]:
         except (sqlite3.Error, ValidationError) as error:
             raise SchedulerError(SchedulerErrorCode.DATABASE_ERROR, job_id.value) from error
 
+    def predecessors(self, job_id: JobId) -> tuple[ScheduledJob, ...]:
+        try:
+            with self._lock:
+                return tuple(
+                    self._load_job(dependency.job_id) for dependency in self._dependencies(job_id)
+                )
+        except sqlite3.Error as error:
+            raise SchedulerError(SchedulerErrorCode.DATABASE_ERROR, job_id.value) from error
+
     def reconcile(self, now: datetime) -> ReconcileReport:
         instant = _utc(now)
         recovered: tuple[JobId, ...] = ()
         readied: tuple[JobId, ...] = ()
         failed: tuple[JobId, ...] = ()
+        skipped: tuple[JobId, ...] = ()
         blockers: tuple[JobBlocker, ...] = ()
         with self._transaction():
             expired_ids = self._job_ids(
@@ -689,6 +712,11 @@ def reconcile(self, now: datetime) -> ReconcileReport:
                     self._save_job(updated)
                     failed = (*failed, job.id)
                     continue
+                if readiness.skipped:
+                    updated = replace(job, state=JobState.SKIPPED, updated_at=instant)
+                    self._save_job(updated)
+                    skipped = (*skipped, job.id)
+                    continue
                 if readiness.ready:
                     updated = replace(job, state=JobState.READY, updated_at=instant)
                     self._save_job(updated)
@@ -696,7 +724,7 @@ def reconcile(self, now: datetime) -> ReconcileReport:
                     continue
                 if readiness.blocker is not None:
                     blockers = (*blockers, JobBlocker(job.id, readiness.blocker))
-        return ReconcileReport(recovered, readied, failed, blockers)
+        return ReconcileReport(recovered, readied, failed, skipped, blockers)
 
     def claim(self, worker_id: WorkerId, now: datetime) -> JobLease | None:
         instant = _utc(now)
@@ -1029,10 +1057,35 @@ def _readiness(self, job: ScheduledJob, now: datetime) -> _Readiness:
                 return _Readiness(False, BlockerCode.DEPENDENCY_WAITING)
         if job.spec.kind is JobKind.MINE and self._active_kind(job.spec, job.id):
             return _Readiness(False, BlockerCode.ACTIVE_MINE)
+        if job.spec.kind is JobKind.PROMOTE:
+            return self._promotion_readiness(job, predecessors)
         if job.spec.kind is not JobKind.FIT:
             return _Readiness(True)
         return self._fit_readiness(job, predecessors, now)
 
+    def _promotion_readiness(
+        self,
+        job: ScheduledJob,
+        predecessors: tuple[tuple[Dependency, ScheduledJob], ...],
+    ) -> _Readiness:
+        fit_jobs = tuple(
+            predecessor
+            for dependency, predecessor in predecessors
+            if dependency.mode is DependencyMode.REQUIRED and predecessor.spec.kind is JobKind.FIT
+        )
+        if len(fit_jobs) != 1:
+            return _Readiness(False, BlockerCode.DEPENDENCY_INVALID)
+        result = fit_jobs[0].result
+        if (
+            result is None
+            or result.kind is not JobKind.FIT
+            or result.revision_id != job.spec.revision_id
+        ):
+            return _Readiness(False, BlockerCode.DEPENDENCY_INVALID)
+        if not result.progress:
+            return _Readiness(False, BlockerCode.NO_WINNER, skipped=True)
+        return _Readiness(True)
+
     def _fit_readiness(
         self,
         job: ScheduledJob,
@@ -1522,6 +1575,18 @@ def tick(
                 )
                 if fit.created:
                     created = (*created, fit.job.id)
+                promotion = self.scheduler._enqueue(
+                    JobSpec(
+                        JobKind.PROMOTE,
+                        revision_id,
+                        evidence.source,
+                        self.scheduler.policy.stage_budgets.for_kind(JobKind.PROMOTE),
+                    ),
+                    (Dependency(fit.job.id, DependencyMode.REQUIRED),),
+                    instant,
+                )
+                if promotion.created:
+                    created = (*created, promotion.job.id)
         reconciliation = self.scheduler.reconcile(instant)
         return HeartbeatReport(
             created,
@@ -1587,7 +1652,7 @@ def run_once(self, now: datetime) -> ScheduledJob | None:
         try:
             execution = handler.execute(
                 lease.job.spec,
-                JobContext(self.scheduler, lease),
+                JobContext(self.scheduler, lease, _utc(now)),
             )
         except JobExecutionError as error:
             return self._fail_lease(
@@ -1681,3 +1746,10 @@ def _backoff(base: timedelta, attempt: int) -> timedelta:
     if attempt < 1:
         raise ValueError("attempt must be positive")
     return base * (1 << (attempt - 1))
+
+
+def read_automation_policy(path: Path) -> AutomationPolicy:
+    try:
+        return _POLICY_ADAPTER.validate_json(path.read_bytes())
+    except (OSError, ValidationError) as error:
+        raise SchedulerError(SchedulerErrorCode.POLICY_MISMATCH, str(path)) from error
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..a279301
--- /dev/null
+++ b/tests/__init__.py
@@ -0,0 +1 @@
+"""Shared test fixtures for stacked integration coverage."""
diff --git a/tests/test_promotion.py b/tests/test_promotion.py
new file mode 100644
index 0000000..4babf06
--- /dev/null
+++ b/tests/test_promotion.py
@@ -0,0 +1,249 @@
+"""Governed Git promotion, approval, rollback, and idempotency."""
+
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+
+import pytest
+
+from ofw import BenchmarkPolicy, FitCampaign, Harness
+from ofw import ofw as ofw_namespace
+from ofw.candidate import CandidateBuild
+from ofw.promotion import (
+    ApprovalDecision,
+    ApprovalRecord,
+    ApproverId,
+    DeploymentAdapter,
+    DeploymentReference,
+    DeploymentRequest,
+    GitRemote,
+    PromotionError,
+    PromotionErrorCode,
+    PromotionMarker,
+    PromotionMode,
+    PromotionPolicy,
+    PromotionRequest,
+    PromotionService,
+    PullRequestDraft,
+    PullRequestId,
+    PullRequestPublisher,
+    PullRequestReference,
+)
+from tests.test_fit import _bundle, _candidate, _fit_policy, _harness
+
+_NOW = datetime(2026, 8, 22, 18, tzinfo=UTC)
+
+
+def _run_git(root: Path, *arguments: str) -> str:
+    return subprocess.run(
+        ("git", "-C", str(root), *arguments),
+        check=True,
+        capture_output=True,
+        text=True,
+    ).stdout.strip()
+
+
+def _winning_campaign(tmp_path: Path) -> tuple[Harness, FitCampaign, CandidateBuild]:
+    harness = _harness(tmp_path)
+    revision = harness.current_revision
+    assert revision is not None
+    candidate = _candidate(
+        revision,
+        "def run(value: str) -> str:\n"
+        "    return value + (' FIXED' if 'regression' not in value else '')\n",
+        "Fix frontier and holdouts without regressing the critical case.",
+    )
+    campaign = FitCampaign(
+        harness,
+        _bundle(revision),
+        BenchmarkPolicy(1, 10, 0, 0.25),
+        _fit_policy(),
+        (candidate,),
+    )
+    return harness, campaign, candidate
+
+
+@dataclass(slots=True)
+class _PullRequests:
+    opened: list[PullRequestReference] = field(default_factory=list)
+
+    def find(self, marker: PromotionMarker) -> PullRequestReference | None:
+        return next((reference for reference in self.opened if reference.marker == marker), None)
+
+    def open(self, draft: PullRequestDraft) -> PullRequestReference:
+        reference = PullRequestReference(
+            PullRequestId(len(self.opened) + 1),
+            f"https://example.test/pulls/{len(self.opened) + 1}",
+            draft.marker,
+            draft.head_branch,
+        )
+        self.opened.append(reference)
+        return reference
+
+
+@dataclass(slots=True)
+class _Deployments:
+    deployed: list[DeploymentReference] = field(default_factory=list)
+
+    def find(self, marker: PromotionMarker) -> DeploymentReference | None:
+        return next((reference for reference in self.deployed if reference.marker == marker), None)
+
+    def deploy(self, request: DeploymentRequest) -> DeploymentReference:
+        reference = DeploymentReference(
+            f"deployment-{len(self.deployed) + 1}",
+            request.marker,
+            f"rollback {request.commit.value}",
+        )
+        self.deployed.append(reference)
+        return reference
+
+
+def test_pull_request_promotion_is_idempotent_and_has_reverse_artifact(
+    tmp_path: Path,
+) -> None:
+    harness, campaign, candidate = _winning_campaign(tmp_path)
+    fit_result = campaign.run()
+    remote = tmp_path / "remote.git"
+    subprocess.run(("git", "init", "--bare", "-q", str(remote)), check=True)
+    _run_git(harness.root, "remote", "add", "review", str(remote))
+    publisher = _PullRequests()
+    publisher_contract: PullRequestPublisher = publisher
+    request = PromotionRequest(
+        campaign,
+        fit_result,
+        candidate,
+        PromotionPolicy(
+            PromotionMode.PULL_REQUEST,
+            GitRemote("review", "main", "ofw"),
+            require_approval=False,
+        ),
+        None,
+    )
+    interrupted_worktree = candidate.workspace.parent / f"promotion-{request.id[-16:]}"
+    _run_git(
+        harness.root,
+        "worktree",
+        "add",
+        "-b",
+        f"ofw/promotion-{request.id[-16:]}",
+        str(interrupted_worktree),
+        candidate.workspace.branch.value,
+    )
+
+    result = ofw_namespace.promote(
+        request,
+        now=_NOW,
+        pull_requests=publisher_contract,
+    )
+    result.manifest_path.unlink()
+    result.digest_path.unlink()
+    recovered = PromotionService(publisher_contract, None).run(request, _NOW)
+    restarted = PromotionService(publisher_contract, None).run(request, _NOW)
+
+    assert restarted == result
+    assert recovered == result
+    assert result.pull_request == publisher.opened[0]
+    assert not interrupted_worktree.exists()
+    assert len(publisher.opened) == 1
+    assert result.deployment is None
+    assert result.commit is not None
+    assert result.rollback.reverse_patch.read_bytes()
+    assert result.report_path.read_text(encoding="utf-8").startswith("<!doctype html>")
+    remote_commit = _run_git(
+        remote,
+        "rev-parse",
+        f"refs/heads/{result.pull_request.head_branch}",
+    )
+    assert remote_commit == result.commit.value
+    subprocess.run(
+        (
+            "git",
+            "-C",
+            str(candidate.workspace.root),
+            "apply",
+            "--check",
+            str(result.rollback.reverse_patch),
+        ),
+        check=True,
+    )
+    assert campaign.run() == fit_result
+    result.rollback.reverse_patch.write_bytes(result.rollback.reverse_patch.read_bytes() + b"\n")
+    with pytest.raises(PromotionError) as tampered:
+        PromotionService(publisher_contract, None).run(request, _NOW)
+    assert tampered.value.code is PromotionErrorCode.RESULT_INVALID
+    candidate.workspace.close()
+
+
+def test_direct_deploy_requires_matching_human_approval(tmp_path: Path) -> None:
+    _harness_instance, campaign, candidate = _winning_campaign(tmp_path)
+    fit_result = campaign.run()
+    policy = PromotionPolicy(PromotionMode.DEPLOY, None, require_approval=True)
+    request = PromotionRequest(campaign, fit_result, candidate, policy, None)
+    deployments = _Deployments()
+    deployment_contract: DeploymentAdapter = deployments
+    service = PromotionService(None, deployment_contract)
+
+    with pytest.raises(PromotionError) as missing:
+        service.run(request, _NOW)
+    assert missing.value.code is PromotionErrorCode.APPROVAL_REQUIRED
+
+    rejected = ApprovalRecord(
+        fit_result.id,
+        candidate.candidate.id,
+        policy.digest,
+        ApproverId("reviewer"),
+        ApprovalDecision.REJECTED,
+        _NOW,
+    )
+    with pytest.raises(PromotionError) as denied:
+        service.run(
+            PromotionRequest(campaign, fit_result, candidate, policy, rejected),
+            _NOW,
+        )
+    assert denied.value.code is PromotionErrorCode.APPROVAL_REJECTED
+
+    approved = ApprovalRecord(
+        fit_result.id,
+        candidate.candidate.id,
+        policy.digest,
+        ApproverId("reviewer"),
+        ApprovalDecision.APPROVED,
+        _NOW,
+    )
+    result = service.run(
+        PromotionRequest(campaign, fit_result, candidate, policy, approved),
+        _NOW,
+    )
+    result.manifest_path.unlink()
+    result.digest_path.unlink()
+    recovered = service.run(
+        PromotionRequest(campaign, fit_result, candidate, policy, approved),
+        _NOW,
+    )
+
+    assert result.deployment == deployments.deployed[0]
+    assert recovered == result
+    assert len(deployments.deployed) == 1
+    assert result.approval_digest == approved.digest
+    candidate.workspace.close()
+
+
+def test_durable_cancellation_prevents_git_side_effects(tmp_path: Path) -> None:
+    harness, campaign, candidate = _winning_campaign(tmp_path)
+    fit_result = campaign.run()
+    policy = PromotionPolicy(PromotionMode.COMMIT, None, require_approval=False)
+    request = PromotionRequest(campaign, fit_result, candidate, policy, None)
+    service = PromotionService(None, None)
+    before = _run_git(harness.root, "for-each-ref", "--format=%(refname)", "refs/heads")
+
+    service.cancel(request, ApproverId("operator"), _NOW)
+    with pytest.raises(PromotionError) as cancelled:
+        service.run(request, _NOW)
+
+    assert cancelled.value.code is PromotionErrorCode.CANCELLED
+    after = _run_git(harness.root, "for-each-ref", "--format=%(refname)", "refs/heads")
+    assert after == before
+    candidate.workspace.close()
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
index 396c3ab..eefe1e6 100644
--- a/tests/test_scheduler.py
+++ b/tests/test_scheduler.py
@@ -11,6 +11,7 @@
 
 from ofw import FitPolicy
 from ofw import ofw as ofw_namespace
+from ofw.cli import run_campaign_command
 from ofw.contracts import HarnessRevisionId
 from ofw.scheduler import (
     AutomationPolicy,
@@ -77,6 +78,7 @@ def _policy(*, daily_budget: Money = _DEFAULT_DAILY_BUDGET) -> AutomationPolicy:
             benchmark_export=Money(10_000),
             memory=Money(10_000),
             fit=Money(100_000),
+            promotion=Money(10_000),
         ),
     )
 
@@ -274,6 +276,72 @@ def test_fit_requires_matching_benchmark_lineage_and_optional_memory(tmp_path: P
     scheduler.close()
 
 
+@pytest.mark.parametrize(
+    ("progress", "expected"),
+    ((True, JobState.READY), (False, JobState.SKIPPED)),
+)
+def test_promotion_requires_a_fit_winner(
+    tmp_path: Path,
+    progress: bool,
+    expected: JobState,
+) -> None:
+    scheduler = _scheduler(tmp_path)
+    mine = scheduler.enqueue(_spec(JobKind.MINE, "promote-mine"), (), _NOW)
+    _run_success(
+        scheduler,
+        WorkerId("worker"),
+        _NOW,
+        _result(JobKind.MINE, "promote-mine-result"),
+    )
+    benchmark = scheduler.enqueue(
+        _spec(JobKind.EXPORT_BENCH_EVAL, "promote-benchmark"),
+        (Dependency(mine.id, DependencyMode.REQUIRED),),
+        _NOW,
+    )
+    scheduler.reconcile(_NOW)
+    _run_success(
+        scheduler,
+        WorkerId("worker"),
+        _NOW,
+        _result(
+            JobKind.EXPORT_BENCH_EVAL,
+            "promote-benchmark-result",
+            source=ResultId("promote-mine-result"),
+        ),
+    )
+    fit = scheduler.enqueue(
+        _spec(JobKind.FIT, "promote-fit", fit_policy=_fit_policy()),
+        (
+            Dependency(mine.id, DependencyMode.REQUIRED),
+            Dependency(benchmark.id, DependencyMode.REQUIRED),
+        ),
+        _NOW,
+    )
+    scheduler.reconcile(_NOW)
+    _run_success(
+        scheduler,
+        WorkerId("worker"),
+        _NOW,
+        _result(
+            JobKind.FIT,
+            "promote-fit-result",
+            progress=progress,
+            fit_policy=_fit_policy(),
+        ),
+    )
+    promotion = scheduler.enqueue(
+        _spec(JobKind.PROMOTE, "promotion"),
+        (Dependency(fit.id, DependencyMode.REQUIRED),),
+        _NOW,
+    )
+
+    report = scheduler.reconcile(_NOW)
+
+    assert scheduler.job(promotion.id).state is expected
+    assert (promotion.id in report.skipped) is (expected is JobState.SKIPPED)
+    scheduler.close()
+
+
 def test_policy_mismatch_and_overlapping_fit_are_blocked(tmp_path: Path) -> None:
     scheduler = _scheduler(tmp_path)
     mine = scheduler.enqueue(_spec(JobKind.MINE, "mine"), (), _NOW)
@@ -454,6 +522,29 @@ def test_restart_rejects_a_different_automation_policy(tmp_path: Path) -> None:
     assert raised.value.code is SchedulerErrorCode.POLICY_MISMATCH
 
 
+def test_campaign_cli_uses_the_same_status_cancel_resume_service(tmp_path: Path) -> None:
+    store_path = tmp_path / "scheduler.sqlite3"
+    policy_path = tmp_path / "automation-policy.json"
+    policy = _policy()
+    policy_path.write_text(policy.to_json(), encoding="utf-8")
+    scheduler = LocalScheduler(store_path, policy)
+    job = scheduler.enqueue(_spec(JobKind.TRACE_SYNC, "cli"), (), _NOW)
+    scheduler.close()
+    prefix = ("campaign", "status", str(store_path), str(policy_path), job.id.value)
+
+    assert run_campaign_command(prefix, _NOW).state is JobState.READY
+    cancelled = run_campaign_command(
+        ("campaign", "cancel", str(store_path), str(policy_path), job.id.value),
+        _NOW,
+    )
+    assert cancelled.state is JobState.CANCELLED
+    resumed = run_campaign_command(
+        ("campaign", "resume", str(store_path), str(policy_path), job.id.value),
+        _NOW,
+    )
+    assert resumed.state is JobState.READY
+
+
 def test_quiet_hours_budget_cooldown_and_no_progress_circuit(tmp_path: Path) -> None:
     policy = _policy(daily_budget=Money(250_000))
     scheduler = _scheduler(tmp_path, policy=policy)
@@ -622,7 +713,7 @@ def test_heartbeat_materializes_pipeline_once_and_excludes_ofw_evidence(tmp_path
         _NOW + timedelta(seconds=60),
     )
 
-    assert len(first.created) == 6
+    assert len(first.created) == 7
     assert second.created == ()
     assert excluded.created == ()
     assert all(job.state in (JobState.PENDING, JobState.READY) for job in scheduler.jobs())