diff --git a/README.md b/README.md
index 5c64ae43..09f76517 100644
--- a/README.md
+++ b/README.md
@@ -69,6 +69,7 @@
 - [Community Hub](#community-hub)
 - [Real-time Event Stream](#real-time-event-stream)
 - [Run Time Machine](#run-time-machine)
+- [Model Time Machine](#model-time-machine)
 - [Budget Guardrails](#budget-guardrails)
 - [Security](#security)
 - [A2A & AG-UI Protocols](#a2a--ag-ui-protocols)
@@ -1547,6 +1548,40 @@ curl -X POST http://localhost:8080/api/runs/{run_id}/fork \
   }'
 ```
 
+### Model Time Machine
+
+Test any model against your real workload, not benchmarks. The Time Machine takes the runs you actually executed - your prompts, your data, your costs - and answers the question every team asks: *"what would last month have looked like on a different model?"*
+
+```bash
+# Free, instant: price last month's recorded token volume on a local Llama 70B
+sandcastle timemachine --model nim/llama-3.1-70b --since 30d
+
+# Live replay: re-execute the recorded steps for real, score quality with an
+# LLM judge, and measure actual cost + latency - capped by an explicit budget
+sandcastle timemachine --model mistral/small --since 30d --live --budget 5.0
+```
+
+Two modes:
+
+- **Dry run (default)** - zero API calls. Recorded token volumes are priced against the target model's pricing table and extrapolated to a monthly figure. Free and instant.
+- **Live replay (`--live`)** - every recorded LLM step is re-executed on the target model. Old and new outputs are scored 0-10 by a configurable LLM judge (`TIMEMACHINE_JUDGE_MODEL`, default `haiku`), and real cost and latency are measured. A live replay **requires** `--budget`: the replay cost is estimated up front from the recorded token counts and the job refuses to start if the estimate exceeds the cap.
+
+The report breaks down quality, cost, and latency deltas per workflow and ends with a one-line verdict: `Switching to nim/llama-3.1-70b saves $312/mo at -2.1% quality.`
+
+Also available in the dashboard (**Operations → Time Machine**) and over the API:
+
+```bash
+# Start a job (dry-run by default; add "live": true + "budget_usd" for live)
+curl -X POST http://localhost:8080/api/timemachine \
+  -H "Content-Type: application/json" \
+  -d '{ "target_model": "nim/llama-3.1-70b", "since": "30d", "max_cassettes": 20 }'
+
+# Fetch the report
+curl http://localhost:8080/api/timemachine/{job_id}
+```
+
+Reports persist as JSON artifacts under `{data_dir}/timemachine/`.
+
 ---
 
 ## Budget Guardrails
diff --git a/dashboard/src/App.tsx b/dashboard/src/App.tsx
index 0a69fd7a..d831e506 100644
--- a/dashboard/src/App.tsx
+++ b/dashboard/src/App.tsx
@@ -36,6 +36,7 @@ const EvolutionPage = lazyWithRetry(() => import("@/pages/EvolutionPage"), "evol
 const SystemHealthPage = lazyWithRetry(() => import("@/pages/SystemHealthPage"), "system-health");
 const CompliancePage = lazyWithRetry(() => import("@/pages/CompliancePage"), "compliance");
 const MemoryPage = lazyWithRetry(() => import("@/pages/MemoryPage"), "memory");
+const TimeMachinePage = lazyWithRetry(() => import("@/pages/TimeMachinePage"), "time-machine");
 const Onboarding = lazyWithRetry(() => import("@/pages/Onboarding"), "onboarding");
 
 /** Wrap a lazy-loaded page in a per-route error boundary so a crash in one
@@ -152,6 +153,7 @@ export default function App() {
               <Route path="/system-health" element={<PageBoundary name="system-health"><SystemHealthPage /></PageBoundary>} />
               <Route path="/compliance" element={<LiteGuard><PageBoundary name="compliance"><CompliancePage /></PageBoundary></LiteGuard>} />
               <Route path="/memory" element={<LiteGuard><PageBoundary name="memory"><MemoryPage /></PageBoundary></LiteGuard>} />
+              <Route path="/time-machine" element={<LiteGuard><PageBoundary name="time-machine"><TimeMachinePage /></PageBoundary></LiteGuard>} />
               <Route path="*" element={<NotFound />} />
             </Route>
           </Routes>
diff --git a/dashboard/src/__tests__/TimeMachinePage.test.tsx b/dashboard/src/__tests__/TimeMachinePage.test.tsx
new file mode 100644
index 00000000..80a90309
--- /dev/null
+++ b/dashboard/src/__tests__/TimeMachinePage.test.tsx
@@ -0,0 +1,168 @@
+/**
+ * TimeMachinePage tests - config form, dry-run job flow with polling,
+ * live-mode budget input, error surfacing, and the empty-selection state.
+ */
+
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { render, screen, fireEvent, waitFor, cleanup } from "@testing-library/react";
+
+const apiGet = vi.fn();
+const apiPost = vi.fn();
+
+vi.mock("@/api/client", () => ({
+  api: {
+    get: (...args: unknown[]) => apiGet(...args),
+    post: (...args: unknown[]) => apiPost(...args),
+    isMockMode: false,
+    onMockChange: vi.fn(() => () => {}),
+  },
+}));
+
+vi.mock("react-router-dom", () => ({
+  useNavigate: () => vi.fn(),
+  useLocation: () => ({ pathname: "/time-machine" }),
+  Link: ({ children, to, ...rest }: { children: React.ReactNode; to: string }) => (
+    <a href={to} {...rest}>{children}</a>
+  ),
+}));
+
+import TimeMachinePage from "@/pages/TimeMachinePage";
+
+const REPORT = {
+  mode: "dry_run",
+  target_model: "nim/llama-3.1-70b",
+  judge_model: null,
+  selection: {
+    runs: 12,
+    steps: 40,
+    workflows: ["lead-enrichment"],
+    original_cost_usd: 20.0,
+    window_days: 28,
+  },
+  cost: { original_usd: 20.0, new_usd: 0.0, delta_usd: -20.0, delta_pct: -100 },
+  quality: null,
+  latency: null,
+  live: null,
+  extrapolation: {
+    window_days: 28,
+    monthly_original_usd: 21.4,
+    monthly_projected_usd: 0,
+    monthly_savings_usd: 21.4,
+  },
+  per_workflow: [
+    {
+      workflow: "lead-enrichment",
+      runs: 12,
+      steps: 40,
+      original_cost_usd: 20.0,
+      new_cost_usd: 0.0,
+      cost_delta_usd: -20.0,
+      cost_delta_pct: -100,
+      quality_old: null,
+      quality_new: null,
+      quality_delta_pct: null,
+      latency_old_seconds: null,
+      latency_new_seconds: null,
+      latency_delta_pct: null,
+    },
+  ],
+  verdict: "Switching to nim/llama-3.1-70b saves $21.40/mo (projected from recorded token volume). Run a live replay to measure quality.",
+};
+
+function mockDefaults(jobReport: unknown = REPORT) {
+  apiGet.mockImplementation((path: string) => {
+    if (path === "/timemachine") return Promise.resolve({ data: [], error: null });
+    if (path === "/workflows") {
+      return Promise.resolve({ data: [{ name: "lead-enrichment" }], error: null });
+    }
+    if (path.startsWith("/timemachine/")) {
+      return Promise.resolve({
+        data: { job_id: "job-1", status: "completed", report: jobReport, error: null },
+        error: null,
+      });
+    }
+    return Promise.resolve({ data: null, error: null });
+  });
+  apiPost.mockResolvedValue({ data: { job_id: "job-1", status: "running" }, error: null });
+}
+
+describe("TimeMachinePage", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    mockDefaults();
+  });
+
+  afterEach(() => {
+    cleanup();
+  });
+
+  it("renders the config form and the confident empty state", async () => {
+    render(<TimeMachinePage />);
+    expect(screen.getByText("Model Time Machine")).toBeInTheDocument();
+    expect(screen.getByLabelText("Target model")).toBeInTheDocument();
+    expect(
+      await screen.findByText("Test any model against your real workload")
+    ).toBeInTheDocument();
+    // dry run is the default - the run button says so
+    expect(screen.getByRole("button", { name: /run dry-run estimate/i })).toBeInTheDocument();
+  });
+
+  it("runs a dry-run job and renders the report after polling", async () => {
+    render(<TimeMachinePage />);
+    fireEvent.click(screen.getByRole("button", { name: /run dry-run estimate/i }));
+
+    await waitFor(() => {
+      expect(apiPost).toHaveBeenCalledWith(
+        "/timemachine",
+        expect.objectContaining({
+          target_model: "nim/llama-3.1-70b",
+          since: "30d",
+          live: false,
+        })
+      );
+    });
+
+    // poll interval is 1.5s - wait for the completed report to land
+    expect(await screen.findByTestId("tm-verdict", {}, { timeout: 4000 })).toBeInTheDocument();
+    expect(screen.getAllByText(/saves \$21\.40\/mo/).length).toBeGreaterThan(0);
+    expect(screen.getAllByText("lead-enrichment").length).toBeGreaterThan(1);
+    expect(screen.getByText(/Per-workflow deltas/)).toBeInTheDocument();
+  }, 10000);
+
+  it("reveals the budget input only in live mode and sends budget_usd", async () => {
+    render(<TimeMachinePage />);
+    expect(screen.queryByLabelText(/budget cap/i)).not.toBeInTheDocument();
+
+    fireEvent.click(screen.getByRole("checkbox"));
+    expect(screen.getByLabelText(/budget cap/i)).toBeInTheDocument();
+
+    fireEvent.click(screen.getByRole("button", { name: /run live replay/i }));
+    await waitFor(() => {
+      expect(apiPost).toHaveBeenCalledWith(
+        "/timemachine",
+        expect.objectContaining({ live: true, budget_usd: 5.0 })
+      );
+    });
+  });
+
+  it("surfaces a budget refusal error from the API", async () => {
+    apiPost.mockResolvedValue({
+      data: null,
+      error: { code: "BUDGET_EXCEEDED", message: "Estimated cost $4.20 exceeds budget $1.00" },
+    });
+    render(<TimeMachinePage />);
+    fireEvent.click(screen.getByRole("button", { name: /run dry-run estimate/i }));
+    expect(
+      await screen.findByText(/exceeds budget \$1\.00/)
+    ).toBeInTheDocument();
+  });
+
+  it("shows the empty-selection state when no recorded workload matched", async () => {
+    mockDefaults({ ...REPORT, selection: { ...REPORT.selection, runs: 0, steps: 0 } });
+    render(<TimeMachinePage />);
+    fireEvent.click(screen.getByRole("button", { name: /run dry-run estimate/i }));
+    expect(
+      await screen.findByText("No recorded workload in this window", {}, { timeout: 4000 })
+    ).toBeInTheDocument();
+  }, 10000);
+});
diff --git a/dashboard/src/api/mock.ts b/dashboard/src/api/mock.ts
index 7355e29a..8628b0cf 100644
--- a/dashboard/src/api/mock.ts
+++ b/dashboard/src/api/mock.ts
@@ -8224,8 +8224,138 @@ const routes: MockRoute[] = [
       };
     },
   },
+  // --- Model Time Machine (counterfactual replay) ---
+  {
+    match: /^\/timemachine$/,
+    method: "POST",
+    handler: (_params, body) => {
+      const b = body as { target_model?: string; live?: boolean } | undefined;
+      mockTimeMachineLast = {
+        target_model: b?.target_model || "nim/llama-3.1-70b",
+        live: Boolean(b?.live),
+      };
+      return {
+        job_id: "tm-mock-0001",
+        status: "running",
+        mode: mockTimeMachineLast.live ? "live" : "dry_run",
+        target_model: mockTimeMachineLast.target_model,
+      };
+    },
+  },
+  {
+    match: /^\/timemachine$/,
+    handler: () => [
+      {
+        job_id: "tm-mock-0001",
+        status: "completed",
+        created_at: h(2),
+        completed_at: h(2),
+        mode: "dry_run",
+        target_model: "nim/llama-3.1-70b",
+        verdict:
+          "Switching to nim/llama-3.1-70b saves $312.40/mo (projected from recorded token volume). Run a live replay to measure quality.",
+        error: null,
+      },
+      {
+        job_id: "tm-mock-0002",
+        status: "completed",
+        created_at: d(3),
+        completed_at: d(3),
+        mode: "live",
+        target_model: "mistral/small",
+        verdict: "Switching to mistral/small saves $268.10/mo at -3.2% quality.",
+        error: null,
+      },
+    ],
+  },
+  {
+    match: /^\/timemachine\/([^/]+)$/,
+    handler: (params) => ({
+      job_id: params._1,
+      status: "completed",
+      params: {},
+      created_at: h(0.01),
+      completed_at: h(0),
+      error: null,
+      report: buildMockTimeMachineReport(
+        mockTimeMachineLast.target_model,
+        params._1 === "tm-mock-0002" ? true : mockTimeMachineLast.live
+      ),
+    }),
+  },
 ];
 
+let mockTimeMachineLast = { target_model: "nim/llama-3.1-70b", live: false };
+
+function buildMockTimeMachineReport(targetModel: string, live: boolean) {
+  const local = targetModel.startsWith("nim/") || targetModel.startsWith("omlx/") || targetModel === "ollama";
+  const ratio = local ? 0 : 0.22; // new cost as a fraction of original
+  const rows = [
+    { workflow: "lead-enrichment", runs: 9, steps: 34, original: 14.92, lat: 11.2 },
+    { workflow: "competitor-monitor", runs: 6, steps: 18, original: 7.41, lat: 8.6 },
+    { workflow: "seo-audit", runs: 5, steps: 15, original: 5.08, lat: 9.9 },
+  ];
+  const originalTotal = rows.reduce((acc, r) => acc + r.original, 0);
+  const newTotal = originalTotal * ratio;
+  const monthlyOriginal = originalTotal * (30 / 28);
+  const monthlySavings = monthlyOriginal * (1 - ratio);
+  return {
+    mode: live ? "live" : "dry_run",
+    target_model: targetModel,
+    judge_model: live ? "haiku" : null,
+    selection: {
+      runs: rows.reduce((acc, r) => acc + r.runs, 0),
+      steps: rows.reduce((acc, r) => acc + r.steps, 0),
+      workflows: rows.map((r) => r.workflow),
+      original_cost_usd: originalTotal,
+      window_days: 28,
+    },
+    cost: {
+      original_usd: originalTotal,
+      new_usd: newTotal,
+      delta_usd: newTotal - originalTotal,
+      delta_pct: (ratio - 1) * 100,
+    },
+    quality: live ? { old_avg: 8.4, new_avg: 8.1, delta_pct: -3.6 } : null,
+    latency: live
+      ? { old_avg_seconds: 10.4, new_avg_seconds: 6.1, delta_pct: -41.3 }
+      : null,
+    live: live
+      ? {
+          measured_cost_usd: newTotal + 0.31,
+          budget_usd: 5.0,
+          steps_replayed: 67,
+          steps_failed: 0,
+          truncated: false,
+        }
+      : null,
+    extrapolation: {
+      window_days: 28,
+      monthly_original_usd: monthlyOriginal,
+      monthly_projected_usd: monthlyOriginal * ratio,
+      monthly_savings_usd: monthlySavings,
+    },
+    per_workflow: rows.map((r) => ({
+      workflow: r.workflow,
+      runs: r.runs,
+      steps: r.steps,
+      original_cost_usd: r.original,
+      new_cost_usd: r.original * ratio,
+      cost_delta_usd: r.original * ratio - r.original,
+      cost_delta_pct: (ratio - 1) * 100,
+      quality_old: live ? 8.4 : null,
+      quality_new: live ? 8.1 : null,
+      quality_delta_pct: live ? -3.6 : null,
+      latency_old_seconds: live ? r.lat : null,
+      latency_new_seconds: live ? r.lat * 0.59 : null,
+      latency_delta_pct: live ? -41.0 : null,
+    })),
+    verdict: live
+      ? `Switching to ${targetModel} saves $${monthlySavings.toFixed(2)}/mo at -3.6% quality.`
+      : `Switching to ${targetModel} saves $${monthlySavings.toFixed(2)}/mo (projected from recorded token volume). Run a live replay to measure quality.`,
+  };
+}
+
 export function mockFetch(
   path: string,
   params?: Record<string, string>,
diff --git a/dashboard/src/components/layout/Header.tsx b/dashboard/src/components/layout/Header.tsx
index c0a04c13..04b62454 100644
--- a/dashboard/src/components/layout/Header.tsx
+++ b/dashboard/src/components/layout/Header.tsx
@@ -32,6 +32,7 @@ const PAGE_TITLES: Record<string, string> = {
   "/autopilot": "AutoPilot",
   "/violations": "Violations",
   "/optimizer": "Optimizer",
+  "/time-machine": "Time Machine",
   "/schedules": "Schedules",
   "/dead-letter": "Dead Letter",
   "/api-keys": "API Keys",
diff --git a/dashboard/src/components/layout/Layout.tsx b/dashboard/src/components/layout/Layout.tsx
index 95e3d43c..8d983195 100644
--- a/dashboard/src/components/layout/Layout.tsx
+++ b/dashboard/src/components/layout/Layout.tsx
@@ -98,6 +98,7 @@ const ROUTE_LABELS: Record<string, string> = {
   "/autopilot": "AutoPilot",
   "/violations": "Violations",
   "/optimizer": "Optimizer",
+  "/time-machine": "Time Machine",
   "/schedules": "Schedules",
   "/dead-letter": "Dead Letter",
   "/api-keys": "API Keys",
diff --git a/dashboard/src/components/layout/Sidebar.tsx b/dashboard/src/components/layout/Sidebar.tsx
index 9b3c9c4d..a172d31b 100644
--- a/dashboard/src/components/layout/Sidebar.tsx
+++ b/dashboard/src/components/layout/Sidebar.tsx
@@ -11,6 +11,7 @@ import {
   Gauge,
   GitBranch,
   HeartPulse,
+  History,
   Inbox,
   Key,
   Layers,
@@ -82,6 +83,7 @@ const navSections: NavSection[] = [
       { to: "/violations", icon: ShieldAlert, label: "Violations" },
       { to: "/compliance", icon: Shield, label: "Compliance" },
       { to: "/optimizer", icon: Gauge, label: "Optimizer" },
+      { to: "/time-machine", icon: History, label: "Time Machine" },
       { to: "/schedules", icon: Calendar, label: "Schedules" },
       { to: "/schedule-monitor", icon: CalendarClock, label: "Schedule Monitor" },
       { to: "/dead-letter", icon: Inbox, label: "Dead Letter", badge: "dlq" },
diff --git a/dashboard/src/pages/TimeMachinePage.tsx b/dashboard/src/pages/TimeMachinePage.tsx
new file mode 100644
index 00000000..833ac5c2
--- /dev/null
+++ b/dashboard/src/pages/TimeMachinePage.tsx
@@ -0,0 +1,645 @@
+import { useCallback, useEffect, useRef, useState } from "react";
+import {
+  ArrowDown,
+  ArrowUp,
+  Clock,
+  DollarSign,
+  FlaskConical,
+  History,
+  Layers,
+  Minus,
+  Play,
+  Sparkles,
+} from "lucide-react";
+import { api } from "@/api/client";
+import { Breadcrumb } from "@/components/shared/Breadcrumb";
+import { EmptyState } from "@/components/shared/EmptyState";
+import { Skeleton } from "@/components/ui/Skeleton";
+import { cn, formatCost } from "@/lib/utils";
+
+interface WorkflowRow {
+  workflow: string;
+  runs: number;
+  steps: number;
+  original_cost_usd: number;
+  new_cost_usd: number;
+  cost_delta_usd: number;
+  cost_delta_pct: number | null;
+  quality_old: number | null;
+  quality_new: number | null;
+  quality_delta_pct: number | null;
+  latency_old_seconds: number | null;
+  latency_new_seconds: number | null;
+  latency_delta_pct: number | null;
+}
+
+interface TimeMachineReport {
+  mode: "dry_run" | "live";
+  target_model: string;
+  judge_model: string | null;
+  selection: {
+    runs: number;
+    steps: number;
+    workflows: string[];
+    original_cost_usd: number;
+    window_days: number;
+  };
+  cost: { original_usd: number; new_usd: number; delta_usd: number; delta_pct: number | null };
+  quality: { old_avg: number | null; new_avg: number | null; delta_pct: number | null } | null;
+  latency: {
+    old_avg_seconds: number | null;
+    new_avg_seconds: number | null;
+    delta_pct: number | null;
+  } | null;
+  live: {
+    measured_cost_usd: number;
+    budget_usd: number | null;
+    steps_replayed: number;
+    steps_failed: number;
+    truncated: boolean;
+  } | null;
+  extrapolation: {
+    window_days: number;
+    monthly_original_usd: number;
+    monthly_projected_usd: number;
+    monthly_savings_usd: number;
+  };
+  per_workflow: WorkflowRow[];
+  verdict: string;
+}
+
+interface TimeMachineJob {
+  job_id: string;
+  status: string;
+  created_at: string | null;
+  completed_at: string | null;
+  report: TimeMachineReport | null;
+  error: string | null;
+}
+
+interface JobSummary {
+  job_id: string;
+  status: string;
+  created_at: string | null;
+  mode: string;
+  target_model: string | null;
+  verdict: string | null;
+  error: string | null;
+}
+
+const MODEL_GROUPS: Array<{ label: string; models: string[] }> = [
+  { label: "Claude", models: ["sonnet", "haiku", "opus"] },
+  { label: "OpenAI", models: ["openai/codex", "openai/codex-mini"] },
+  { label: "Mistral (EU)", models: ["mistral/large", "mistral/small", "mistral/codestral"] },
+  { label: "Google", models: ["google/gemini-2.5-pro"] },
+  { label: "MiniMax", models: ["minimax/m2.5"] },
+  {
+    label: "Local (free)",
+    models: [
+      "nim/llama-3.1-70b",
+      "nim/llama-3.1-8b",
+      "nim/qwen2.5-coder-32b",
+      "ollama",
+      "omlx/llama-4-scout",
+    ],
+  },
+];
+
+const RANGE_OPTIONS = [
+  { value: "7d", label: "Last 7 days" },
+  { value: "30d", label: "Last 30 days" },
+  { value: "90d", label: "Last 90 days" },
+];
+
+function PctChip({ pct, betterWhenNegative = true, suffix }: {
+  pct: number | null | undefined;
+  betterWhenNegative?: boolean;
+  suffix?: string;
+}) {
+  if (pct === null || pct === undefined) {
+    return (
+      <span className="inline-flex items-center gap-0.5 text-xs text-muted">
+        <Minus className="h-3 w-3" />n/a
+      </span>
+    );
+  }
+  if (Math.abs(pct) < 0.05) {
+    return (
+      <span className="inline-flex items-center gap-0.5 text-xs text-muted">
+        <Minus className="h-3 w-3" />same
+      </span>
+    );
+  }
+  const better = betterWhenNegative ? pct < 0 : pct > 0;
+  return (
+    <span
+      className={cn(
+        "inline-flex items-center gap-0.5 text-xs font-medium",
+        better ? "text-success" : "text-error"
+      )}
+    >
+      {pct < 0 ? <ArrowDown className="h-3 w-3" /> : <ArrowUp className="h-3 w-3" />}
+      {pct > 0 ? "+" : ""}{pct.toFixed(1)}%{suffix ? ` ${suffix}` : ""}
+    </span>
+  );
+}
+
+function SummaryCard({ icon: Icon, label, children }: {
+  icon: typeof Clock;
+  label: string;
+  children: React.ReactNode;
+}) {
+  return (
+    <div className="rounded-xl border border-border bg-surface p-3 shadow-sm">
+      <div className="mb-2 flex items-center gap-1.5">
+        <Icon className="h-3.5 w-3.5 text-muted-foreground" />
+        <span className="text-xs font-medium text-muted-foreground">{label}</span>
+      </div>
+      {children}
+    </div>
+  );
+}
+
+function ReportView({ report }: { report: TimeMachineReport }) {
+  const savings = report.extrapolation.monthly_savings_usd;
+
+  if (report.selection.runs === 0) {
+    return (
+      <div className="rounded-xl border border-border bg-surface py-12 shadow-sm">
+        <EmptyState
+          icon={History}
+          title="No recorded workload in this window"
+          description="The Time Machine replays your real runs, not synthetic benchmarks. Run a few workflows (or widen the date range), then come back to test any model against them."
+        />
+      </div>
+    );
+  }
+
+  return (
+    <div className="space-y-4">
+      {/* Headline verdict */}
+      <div
+        className={cn(
+          "rounded-xl border p-4 sm:p-5 shadow-sm",
+          savings >= 0 ? "border-success/40 bg-success/5" : "border-warning/40 bg-warning/5"
+        )}
+        data-testid="tm-verdict"
+      >
+        <div className="flex items-start gap-3">
+          <Sparkles className={cn("mt-0.5 h-5 w-5 shrink-0", savings >= 0 ? "text-success" : "text-warning")} />
+          <div>
+            <div className="text-sm font-semibold text-foreground">{report.verdict}</div>
+            <div className="mt-1 text-xs text-muted-foreground">
+              Based on {report.selection.runs} recorded run{report.selection.runs !== 1 ? "s" : ""} /{" "}
+              {report.selection.steps} LLM steps over ~{Math.round(report.selection.window_days)} day
+              {Math.round(report.selection.window_days) !== 1 ? "s" : ""}
+              {report.mode === "dry_run"
+                ? " - projected from recorded token volume, no API calls made."
+                : ` - live replay judged by ${report.judge_model ?? "LLM judge"}.`}
+            </div>
+          </div>
+        </div>
+      </div>
+
+      {/* Summary cards */}
+      <div className="grid grid-cols-2 gap-3 sm:grid-cols-4">
+        <SummaryCard icon={DollarSign} label="Cost (selection)">
+          <div className="mb-1 flex items-baseline gap-1.5">
+            <span className="text-sm font-semibold text-foreground">
+              {formatCost(report.cost.original_usd)}
+            </span>
+            <span className="text-xs text-muted">to</span>
+            <span className="text-sm font-semibold text-foreground">
+              {formatCost(report.cost.new_usd)}
+            </span>
+          </div>
+          <PctChip pct={report.cost.delta_pct} />
+        </SummaryCard>
+
+        <SummaryCard icon={FlaskConical} label="Quality (judge 0-10)">
+          {report.quality && report.quality.old_avg !== null ? (
+            <>
+              <div className="mb-1 flex items-baseline gap-1.5">
+                <span className="text-sm font-semibold text-foreground">
+                  {report.quality.old_avg?.toFixed(1)}
+                </span>
+                <span className="text-xs text-muted">to</span>
+                <span className="text-sm font-semibold text-foreground">
+                  {report.quality.new_avg?.toFixed(1)}
+                </span>
+              </div>
+              <PctChip pct={report.quality.delta_pct} betterWhenNegative={false} />
+            </>
+          ) : (
+            <div className="text-xs text-muted">
+              Dry run - run a live replay to score quality
+            </div>
+          )}
+        </SummaryCard>
+
+        <SummaryCard icon={Clock} label="Latency (avg / step)">
+          {report.latency && report.latency.old_avg_seconds !== null ? (
+            <>
+              <div className="mb-1 flex items-baseline gap-1.5">
+                <span className="text-sm font-semibold text-foreground">
+                  {report.latency.old_avg_seconds?.toFixed(1)}s
+                </span>
+                <span className="text-xs text-muted">to</span>
+                <span className="text-sm font-semibold text-foreground">
+                  {report.latency.new_avg_seconds?.toFixed(1)}s
+                </span>
+              </div>
+              <PctChip pct={report.latency.delta_pct} />
+            </>
+          ) : (
+            <div className="text-xs text-muted">Measured during live replay</div>
+          )}
+        </SummaryCard>
+
+        <SummaryCard icon={Layers} label="Monthly projection">
+          <div className="mb-1 text-sm font-semibold text-foreground">
+            {formatCost(report.extrapolation.monthly_original_usd)} to{" "}
+            {formatCost(report.extrapolation.monthly_projected_usd)}
+          </div>
+          <span className={cn("text-xs font-medium", savings >= 0 ? "text-success" : "text-error")}>
+            {savings >= 0 ? "saves" : "adds"} {formatCost(Math.abs(savings))}/mo
+          </span>
+        </SummaryCard>
+      </div>
+
+      {/* Live replay stats */}
+      {report.live && (
+        <div className="rounded-xl border border-border bg-surface p-3 text-xs text-muted-foreground shadow-sm">
+          Live replay: {report.live.steps_replayed} steps re-executed
+          {report.live.steps_failed > 0 && (
+            <span className="text-error"> ({report.live.steps_failed} failed)</span>
+          )}
+          , measured {formatCost(report.live.measured_cost_usd)}
+          {report.live.budget_usd !== null && <> of {formatCost(report.live.budget_usd)} budget</>}
+          {report.live.truncated && <span className="text-warning"> - truncated at budget cap</span>}
+        </div>
+      )}
+
+      {/* Per-workflow table */}
+      <div className="overflow-hidden rounded-xl border border-border bg-surface shadow-sm">
+        <div className="border-b border-border px-4 py-3">
+          <h2 className="text-sm font-semibold text-foreground">
+            Per-workflow deltas ({report.per_workflow.length})
+          </h2>
+        </div>
+        <div className="overflow-x-auto">
+          <table className="w-full text-sm">
+            <thead>
+              <tr className="border-b border-border text-left text-xs text-muted-foreground">
+                <th className="px-4 py-2 font-medium">Workflow</th>
+                <th className="px-4 py-2 text-right font-medium">Runs</th>
+                <th className="px-4 py-2 text-right font-medium">Cost</th>
+                <th className="px-4 py-2 text-right font-medium">Quality</th>
+                <th className="px-4 py-2 text-right font-medium">Latency</th>
+              </tr>
+            </thead>
+            <tbody>
+              {report.per_workflow.map((wf) => (
+                <tr key={wf.workflow} className="border-b border-border/50 last:border-0">
+                  <td className="px-4 py-2.5 font-mono text-xs font-medium text-foreground">
+                    {wf.workflow}
+                  </td>
+                  <td className="px-4 py-2.5 text-right text-xs text-muted-foreground">
+                    {wf.runs}
+                  </td>
+                  <td className="px-4 py-2.5 text-right">
+                    <div className="text-xs text-muted-foreground">
+                      {formatCost(wf.original_cost_usd)} to {formatCost(wf.new_cost_usd)}
+                    </div>
+                    <PctChip pct={wf.cost_delta_pct} />
+                  </td>
+                  <td className="px-4 py-2.5 text-right">
+                    {wf.quality_old !== null && wf.quality_new !== null ? (
+                      <>
+                        <div className="text-xs text-muted-foreground">
+                          {wf.quality_old.toFixed(1)} to {wf.quality_new.toFixed(1)}
+                        </div>
+                        <PctChip pct={wf.quality_delta_pct} betterWhenNegative={false} />
+                      </>
+                    ) : (
+                      <span className="text-xs text-muted">-</span>
+                    )}
+                  </td>
+                  <td className="px-4 py-2.5 text-right">
+                    {wf.latency_old_seconds !== null && wf.latency_new_seconds !== null ? (
+                      <>
+                        <div className="text-xs text-muted-foreground">
+                          {wf.latency_old_seconds.toFixed(1)}s to {wf.latency_new_seconds.toFixed(1)}s
+                        </div>
+                        <PctChip pct={wf.latency_delta_pct} />
+                      </>
+                    ) : (
+                      <span className="text-xs text-muted">-</span>
+                    )}
+                  </td>
+                </tr>
+              ))}
+            </tbody>
+          </table>
+        </div>
+      </div>
+    </div>
+  );
+}
+
+export default function TimeMachinePage() {
+  const [targetModel, setTargetModel] = useState("nim/llama-3.1-70b");
+  const [workflow, setWorkflow] = useState("");
+  const [range, setRange] = useState("30d");
+  const [maxCassettes, setMaxCassettes] = useState(20);
+  const [live, setLive] = useState(false);
+  const [budget, setBudget] = useState("5.00");
+  const [workflows, setWorkflows] = useState<string[]>([]);
+  const [history, setHistory] = useState<JobSummary[]>([]);
+  const [running, setRunning] = useState(false);
+  const [report, setReport] = useState<TimeMachineReport | null>(null);
+  const [error, setError] = useState<string | null>(null);
+  const pollRef = useRef<ReturnType<typeof setInterval> | null>(null);
+
+  const stopPolling = useCallback(() => {
+    if (pollRef.current) {
+      clearInterval(pollRef.current);
+      pollRef.current = null;
+    }
+  }, []);
+
+  const fetchHistory = useCallback(async () => {
+    const res = await api.get<JobSummary[]>("/timemachine");
+    if (res.data) setHistory(res.data);
+  }, []);
+
+  useEffect(() => {
+    void fetchHistory();
+    void (async () => {
+      const res = await api.get<Array<{ name: string }>>("/workflows");
+      if (res.data) setWorkflows(res.data.map((w) => w.name).filter(Boolean));
+    })();
+    return stopPolling;
+  }, [fetchHistory, stopPolling]);
+
+  const pollJob = useCallback(
+    (jobId: string) => {
+      stopPolling();
+      pollRef.current = setInterval(async () => {
+        const res = await api.get<TimeMachineJob>(`/timemachine/${jobId}`);
+        const job = res.data;
+        if (!job) return;
+        if (job.status === "running") return;
+        stopPolling();
+        setRunning(false);
+        if (job.status === "completed" && job.report) {
+          setReport(job.report);
+        } else {
+          setError(job.error || "Time Machine job failed");
+        }
+        void fetchHistory();
+      }, 1500);
+    },
+    [fetchHistory, stopPolling]
+  );
+
+  const handleRun = useCallback(async () => {
+    setError(null);
+    setReport(null);
+    setRunning(true);
+    const body: Record<string, unknown> = {
+      target_model: targetModel,
+      since: range,
+      max_cassettes: maxCassettes,
+      live,
+    };
+    if (workflow) body.workflow = workflow;
+    if (live) body.budget_usd = parseFloat(budget) || 0;
+    const res = await api.post<{ job_id: string }>("/timemachine", body);
+    if (res.error || !res.data) {
+      setRunning(false);
+      setError(res.error?.message || "Failed to start Time Machine job");
+      return;
+    }
+    pollJob(res.data.job_id);
+  }, [budget, live, maxCassettes, pollJob, range, targetModel, workflow]);
+
+  const loadJob = useCallback(async (jobId: string) => {
+    setError(null);
+    const res = await api.get<TimeMachineJob>(`/timemachine/${jobId}`);
+    if (res.data?.report) setReport(res.data.report);
+    else if (res.data?.error) setError(res.data.error);
+  }, []);
+
+  return (
+    <div className="space-y-4 sm:space-y-6">
+      <Breadcrumb items={[{ label: "Overview", href: "/" }, { label: "Time Machine" }]} />
+
+      {/* Configuration */}
+      <div className="rounded-xl border border-border bg-surface p-4 shadow-sm sm:p-5">
+        <div className="mb-1 flex items-center gap-2">
+          <History className="h-5 w-5 text-accent" />
+          <h1 className="text-xl font-semibold tracking-tight text-foreground">Model Time Machine</h1>
+        </div>
+        <p className="mb-4 text-sm text-muted-foreground">
+          Re-run your real recorded workload against a different model and get the cost, quality and
+          latency delta - no synthetic benchmarks.
+        </p>
+
+        <div className="grid grid-cols-1 gap-3 sm:grid-cols-2 lg:grid-cols-4">
+          <div>
+            <label htmlFor="tm-model" className="mb-1 block text-xs font-medium text-muted-foreground">
+              Target model
+            </label>
+            <select
+              id="tm-model"
+              value={targetModel}
+              onChange={(e) => setTargetModel(e.target.value)}
+              className="h-9 w-full appearance-none rounded-lg border border-border bg-background px-3 text-sm text-foreground transition-colors focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring/30"
+            >
+              {MODEL_GROUPS.map((g) => (
+                <optgroup key={g.label} label={g.label}>
+                  {g.models.map((m) => (
+                    <option key={m} value={m}>{m}</option>
+                  ))}
+                </optgroup>
+              ))}
+            </select>
+          </div>
+
+          <div>
+            <label htmlFor="tm-workflow" className="mb-1 block text-xs font-medium text-muted-foreground">
+              Workflow
+            </label>
+            <select
+              id="tm-workflow"
+              value={workflow}
+              onChange={(e) => setWorkflow(e.target.value)}
+              className="h-9 w-full appearance-none rounded-lg border border-border bg-background px-3 text-sm text-foreground transition-colors focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring/30"
+            >
+              <option value="">All workflows</option>
+              {workflows.map((w) => (
+                <option key={w} value={w}>{w}</option>
+              ))}
+            </select>
+          </div>
+
+          <div>
+            <label htmlFor="tm-range" className="mb-1 block text-xs font-medium text-muted-foreground">
+              Time range
+            </label>
+            <select
+              id="tm-range"
+              value={range}
+              onChange={(e) => setRange(e.target.value)}
+              className="h-9 w-full appearance-none rounded-lg border border-border bg-background px-3 text-sm text-foreground transition-colors focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring/30"
+            >
+              {RANGE_OPTIONS.map((r) => (
+                <option key={r.value} value={r.value}>{r.label}</option>
+              ))}
+            </select>
+          </div>
+
+          <div>
+            <label htmlFor="tm-max" className="mb-1 block text-xs font-medium text-muted-foreground">
+              Max runs to replay
+            </label>
+            <input
+              id="tm-max"
+              type="number"
+              min={1}
+              max={500}
+              value={maxCassettes}
+              onChange={(e) => setMaxCassettes(Math.max(1, parseInt(e.target.value, 10) || 1))}
+              className="h-9 w-full rounded-lg border border-border bg-background px-3 text-sm text-foreground transition-colors focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring/30"
+            />
+          </div>
+        </div>
+
+        <div className="mt-4 flex flex-col gap-3 sm:flex-row sm:items-center">
+          <label className="flex cursor-pointer items-center gap-2 text-sm text-foreground">
+            <input
+              type="checkbox"
+              checked={live}
+              onChange={(e) => setLive(e.target.checked)}
+              className="h-4 w-4 rounded border-border accent-accent"
+            />
+            Live replay
+            <span className="text-xs text-muted-foreground">
+              (re-executes steps with real API calls + LLM-judge quality scoring)
+            </span>
+          </label>
+
+          {live && (
+            <div className="flex items-center gap-2">
+              <label htmlFor="tm-budget" className="text-xs font-medium text-muted-foreground">
+                Budget cap (USD)
+              </label>
+              <input
+                id="tm-budget"
+                type="number"
+                min={0.01}
+                step={0.5}
+                value={budget}
+                onChange={(e) => setBudget(e.target.value)}
+                className="h-8 w-24 rounded-lg border border-border bg-background px-2 text-sm text-foreground focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring/30"
+              />
+            </div>
+          )}
+
+          <button
+            onClick={handleRun}
+            disabled={running}
+            className={cn(
+              "inline-flex h-9 items-center gap-2 rounded-lg bg-accent px-4 text-sm font-medium text-white",
+              "transition-colors hover:bg-accent/90 disabled:cursor-not-allowed disabled:opacity-50",
+              "sm:ml-auto"
+            )}
+          >
+            <Play className="h-4 w-4" />
+            {running ? "Replaying..." : live ? "Run live replay" : "Run dry-run estimate"}
+          </button>
+        </div>
+
+        {!live && (
+          <p className="mt-2 text-xs text-muted">
+            Dry run is free and instant: recorded token volumes are priced against the target
+            model's pricing table. No API calls are made.
+          </p>
+        )}
+      </div>
+
+      {error && (
+        <div className="rounded-xl border border-error/30 bg-error/5 px-4 py-3 text-sm text-error">
+          {error}
+        </div>
+      )}
+
+      {running && (
+        <div className="space-y-3" data-testid="tm-loading">
+          <Skeleton className="h-20 rounded-xl" />
+          <div className="grid grid-cols-2 gap-3 sm:grid-cols-4">
+            {Array.from({ length: 4 }).map((_, i) => (
+              <Skeleton key={i} className="h-24 rounded-xl" />
+            ))}
+          </div>
+          <Skeleton className="h-40 rounded-xl" />
+        </div>
+      )}
+
+      {!running && report && <ReportView report={report} />}
+
+      {!running && !report && !error && (
+        <div className="rounded-xl border border-border bg-surface py-12 shadow-sm">
+          <EmptyState
+            icon={History}
+            title="Test any model against your real workload"
+            description="Pick a target model and a time range, then run a free dry-run estimate. The Time Machine replays last month's actual runs - your prompts, your data - and reports what would change."
+          />
+        </div>
+      )}
+
+      {/* Recent reports */}
+      {history.length > 0 && (
+        <div className="overflow-hidden rounded-xl border border-border bg-surface shadow-sm">
+          <div className="border-b border-border px-4 py-3">
+            <h2 className="text-sm font-semibold text-foreground">Recent reports</h2>
+          </div>
+          <ul className="divide-y divide-border/50">
+            {history.map((j) => (
+              <li key={j.job_id}>
+                <button
+                  onClick={() => void loadJob(j.job_id)}
+                  className="flex w-full items-center gap-3 px-4 py-2.5 text-left transition-colors hover:bg-border/10"
+                >
+                  <span
+                    className={cn(
+                      "h-1.5 w-1.5 shrink-0 rounded-full",
+                      j.status === "completed" && "bg-success",
+                      j.status === "failed" && "bg-error",
+                      j.status === "running" && "bg-running"
+                    )}
+                  />
+                  <span className="font-mono text-xs font-medium text-foreground">
+                    {j.target_model || "?"}
+                  </span>
+                  <span className="rounded bg-accent/10 px-1.5 py-0.5 text-[10px] uppercase text-accent">
+                    {j.mode === "live" ? "live" : "dry run"}
+                  </span>
+                  <span className="hidden truncate text-xs text-muted-foreground sm:inline">
+                    {j.verdict || j.error || j.status}
+                  </span>
+                  {j.created_at && (
+                    <span className="ml-auto shrink-0 text-xs text-muted">
+                      {new Date(j.created_at).toLocaleString()}
+                    </span>
+                  )}
+                </button>
+              </li>
+            ))}
+          </ul>
+        </div>
+      )}
+    </div>
+  );
+}
diff --git a/src/sandcastle/__main__.py b/src/sandcastle/__main__.py
index 23a73442..4cc8b8d2 100644
--- a/src/sandcastle/__main__.py
+++ b/src/sandcastle/__main__.py
@@ -4752,6 +4752,28 @@ def _build_parser() -> argparse.ArgumentParser:
         "--verbose", "-v", action="store_true", help="Show assertion details for failed cases"
     )
 
+    # --- timemachine ---
+    p_tm = subparsers.add_parser(
+        "timemachine",
+        help="Replay recorded workload against a different model (cost/quality/latency delta)",
+    )
+    p_tm.add_argument("--model", "-m", required=True, help="Target model (e.g. nim/llama-3.1-70b)")
+    p_tm.add_argument("--workflow", "-w", default=None, help="Only replay this workflow")
+    p_tm.add_argument("--since", default="30d", help="Window like 30d / 12h or ISO date (default: 30d)")
+    p_tm.add_argument(
+        "--max-cassettes", type=int, default=20, help="Max recorded runs to replay (default: 20)"
+    )
+    p_tm.add_argument(
+        "--live", action="store_true",
+        help="Re-execute steps with real API calls and judge quality (default: dry-run estimate)",
+    )
+    p_tm.add_argument(
+        "--budget", type=float, default=None,
+        help="Hard cost cap in USD for --live (required with --live)",
+    )
+    p_tm.add_argument("--judge", default=None, help="Judge model for quality scoring (default: haiku)")
+    p_tm.add_argument("--json", action="store_true", help="Print the full report as JSON")
+
     # --- generate ---
     p_gen = subparsers.add_parser("generate", help="Generate workflow from natural language")
     p_gen.add_argument("--description", "-d", help="What the workflow should do")
@@ -5000,6 +5022,119 @@ def _build_parser() -> argparse.ArgumentParser:
     return parser
 
 
+def _cmd_timemachine(args: argparse.Namespace) -> None:
+    """Replay recorded workload against a different model and print the delta report."""
+    import asyncio
+
+    from sandcastle.engine import timemachine as tm
+    from sandcastle.engine.providers import is_known_model
+
+    model = args.model
+    if not is_known_model(model):
+        print(f"Error: unknown model '{model}'.", file=sys.stderr)
+        sys.exit(1)
+    if args.live and args.budget is None:
+        print(
+            "Error: --live makes real API calls - an explicit --budget cap is required.",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    since = None
+    if args.since:
+        try:
+            since = tm.parse_since(args.since)
+        except ValueError as exc:
+            print(f"Error: invalid --since value: {exc}", file=sys.stderr)
+            sys.exit(1)
+
+    mode = "live replay" if args.live else "dry run (pricing only, no API calls)"
+    print()
+    print(_color("  Model Time Machine", _C.BOLD))
+    print(_color("  Target model: ", _C.BOLD) + model)
+    print(_color("  Mode:         ", _C.BOLD) + mode)
+    print()
+
+    async def _go() -> dict:
+        from sandcastle.models.db import init_db
+
+        await init_db()
+        return await tm.run_time_machine(
+            target_model=model,
+            workflow=args.workflow,
+            since=since,
+            max_cassettes=args.max_cassettes,
+            live=args.live,
+            budget_usd=args.budget,
+            judge_model=args.judge,
+        )
+
+    try:
+        report = asyncio.run(_go())
+    except tm.TimeMachineError as exc:
+        print(_color(f"  Refused: {exc}", _C.RED), file=sys.stderr)
+        sys.exit(1)
+    except Exception as exc:
+        print(_color(f"  Time Machine failed: {exc}", _C.RED), file=sys.stderr)
+        sys.exit(1)
+
+    if args.json:
+        print(json.dumps(report, indent=2, default=str))
+        return
+
+    sel = report["selection"]
+    if sel["runs"] == 0:
+        print(_color("  No recorded runs matched the selection.", _C.YELLOW))
+        print("  Run some workflows first, then come back - the Time Machine replays")
+        print("  your real recorded workload, not synthetic benchmarks.")
+        return
+
+    cost = report["cost"]
+    print(
+        f"  Selection: {sel['runs']} runs / {sel['steps']} LLM steps "
+        f"across {len(sel['workflows'])} workflow(s), "
+        f"original spend ${cost['original_usd']:.4f}"
+    )
+    print()
+
+    # Per-workflow table
+    header = (
+        f"  {'WORKFLOW':<28} {'COST':>10} {'->':^4} {'NEW':>10} "
+        f"{'QUALITY':>9} {'LATENCY':>9}"
+    )
+    print(_color(header, _C.BOLD))
+    for wf in report["per_workflow"]:
+        q = wf["quality_delta_pct"]
+        lat = wf["latency_delta_pct"]
+        q_str = f"{q:+.1f}%" if q is not None else "-"
+        l_str = f"{lat:+.1f}%" if lat is not None else "-"
+        print(
+            f"  {wf['workflow'][:28]:<28} ${wf['original_cost_usd']:>9.4f} {'->':^4} "
+            f"${wf['new_cost_usd']:>9.4f} {q_str:>9} {l_str:>9}"
+        )
+    print()
+
+    delta_pct = cost["delta_pct"]
+    delta_str = f" ({delta_pct:+.1f}%)" if delta_pct is not None else ""
+    color = _C.GREEN if cost["delta_usd"] <= 0 else _C.RED
+    print(_color(f"  Cost delta: ${cost['delta_usd']:+.4f}{delta_str}", color))
+    if report.get("live"):
+        live = report["live"]
+        print(
+            f"  Live replay: {live['steps_replayed']} steps replayed, "
+            f"{live['steps_failed']} failed, measured ${live['measured_cost_usd']:.4f} "
+            f"of ${live['budget_usd']:.2f} budget"
+            + (" (truncated at budget)" if live["truncated"] else "")
+        )
+    if report.get("quality"):
+        q = report["quality"]
+        if q["old_avg"] is not None:
+            print(f"  Quality (judge {report['judge_model']}): "
+                  f"{q['old_avg']:.1f} -> {q['new_avg']:.1f} ({q['delta_pct']:+.1f}%)")
+    print()
+    print(_color(f"  {report['verdict']}", _C.BOLD))
+
+
 # ---------------------------------------------------------------------------
 # Main entry point
 # ---------------------------------------------------------------------------
@@ -5071,6 +5206,7 @@ def main() -> None:
         "doctor": _cmd_doctor,
         "generate": _cmd_generate,
         "eval": _cmd_eval,
+        "timemachine": _cmd_timemachine,
         "templates": _cmd_templates,
         "replay": _cmd_replay,
         "fork": _cmd_fork,
diff --git a/src/sandcastle/api/routes.py b/src/sandcastle/api/routes.py
index a5f34eec..15b10110 100644
--- a/src/sandcastle/api/routes.py
+++ b/src/sandcastle/api/routes.py
@@ -99,6 +99,9 @@
     StatsResponse,
     StepDiff,
     StepStatusResponse,
+    TimeMachineJobResponse,
+    TimeMachineStartRequest,
+    TimeMachineStartResponse,
     ToolConnectionCreateRequest,
     ToolConnectionResponse,
     ToolConnectionUpdateRequest,
@@ -12638,3 +12641,155 @@ async def get_evolution_stats(req: Request) -> ApiResponse:
             top_workflows=top_workflows,
         )
     )
+
+
+# --- Model Time Machine (counterfactual replay) ---
+
+
+@router.post("/timemachine", status_code=202)
+async def start_timemachine(req: Request) -> ApiResponse:
+    """Start a Model Time Machine job: replay recorded workload against another model.
+
+    Dry-run (default) prices recorded token volumes against the target model's
+    pricing table - free, no API calls. ``live: true`` re-executes the recorded
+    LLM steps for real and judges old vs new output; it requires an explicit
+    ``budget_usd`` and is refused when the pre-flight estimate exceeds it.
+    Admin-only when auth is enabled.
+    """
+    _require_admin(req)
+    body = await req.json()
+    try:
+        parsed = TimeMachineStartRequest(**body)
+    except Exception as exc:
+        raise HTTPException(status_code=422, detail=str(exc))
+
+    from sandcastle.engine import timemachine as tm
+    from sandcastle.engine.providers import is_known_model
+
+    if not is_known_model(parsed.target_model):
+        raise HTTPException(
+            status_code=400,
+            detail=ApiResponse(
+                error=ErrorResponse(
+                    code="UNKNOWN_MODEL",
+                    message=f"Unknown target model '{parsed.target_model}'",
+                )
+            ).model_dump(),
+        )
+    if parsed.judge_model and not is_known_model(parsed.judge_model):
+        raise HTTPException(
+            status_code=400,
+            detail=ApiResponse(
+                error=ErrorResponse(
+                    code="UNKNOWN_MODEL",
+                    message=f"Unknown judge model '{parsed.judge_model}'",
+                )
+            ).model_dump(),
+        )
+
+    since_dt = until_dt = None
+    try:
+        if parsed.since:
+            since_dt = tm.parse_since(parsed.since)
+        if parsed.until:
+            until_dt = tm.parse_since(parsed.until)
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=422,
+            detail=ApiResponse(
+                error=ErrorResponse(code="INVALID_DATE", message=str(exc))
+            ).model_dump(),
+        )
+
+    if parsed.live:
+        # Pre-flight: a live replay needs an explicit budget, and the projected
+        # cost (replay + judge calls, from recorded token volumes) must fit it.
+        if parsed.budget_usd is None:
+            raise HTTPException(
+                status_code=400,
+                detail=ApiResponse(
+                    error=ErrorResponse(
+                        code="BUDGET_REQUIRED",
+                        message="Live replay makes real API calls - budget_usd is required",
+                    )
+                ).model_dump(),
+            )
+        judge = parsed.judge_model or settings.timemachine_judge_model
+        cassettes = await tm.select_cassettes(
+            workflow=parsed.workflow,
+            since=since_dt,
+            until=until_dt,
+            max_cassettes=parsed.max_cassettes,
+        )
+        estimate = tm.estimate_replay_cost(cassettes, parsed.target_model, judge_model=judge)
+        projected = estimate["projected_total_live_cost_usd"]
+        if projected > parsed.budget_usd:
+            raise HTTPException(
+                status_code=400,
+                detail=ApiResponse(
+                    error=ErrorResponse(
+                        code="BUDGET_EXCEEDED",
+                        message=(
+                            f"Estimated live replay cost ${projected:.4f} exceeds budget "
+                            f"${parsed.budget_usd:.4f} - raise the budget or narrow the selection"
+                        ),
+                    )
+                ).model_dump(),
+            )
+
+    job = tm.start_job(
+        target_model=parsed.target_model,
+        workflow=parsed.workflow,
+        since=since_dt,
+        until=until_dt,
+        max_cassettes=parsed.max_cassettes,
+        live=parsed.live,
+        budget_usd=parsed.budget_usd,
+        judge_model=parsed.judge_model,
+    )
+    return ApiResponse(
+        data=TimeMachineStartResponse(
+            job_id=job.job_id,
+            status=job.status,
+            mode="live" if parsed.live else "dry_run",
+            target_model=parsed.target_model,
+        )
+    )
+
+
+@router.get("/timemachine")
+async def list_timemachine_jobs(req: Request) -> ApiResponse:
+    """List recent Time Machine jobs (newest first). Admin-only when auth is enabled."""
+    _require_admin(req)
+    from sandcastle.engine import timemachine as tm
+
+    return ApiResponse(data=tm.list_jobs(limit=20))
+
+
+@router.get("/timemachine/{job_id}")
+async def get_timemachine_job(job_id: str, req: Request) -> ApiResponse:
+    """Get the status and report of a Time Machine job. Admin-only when auth is enabled."""
+    _require_admin(req)
+    from sandcastle.engine import timemachine as tm
+
+    job = tm.get_job(job_id)
+    if job is None:
+        raise HTTPException(
+            status_code=404,
+            detail=ApiResponse(
+                error=ErrorResponse(
+                    code="NOT_FOUND", message=f"Time Machine job '{job_id}' not found"
+                )
+            ).model_dump(),
+        )
+    return ApiResponse(
+        data=TimeMachineJobResponse(
+            job_id=job["job_id"],
+            status=job["status"],
+            params=job.get("params") or {},
+            created_at=job.get("created_at"),
+            completed_at=job.get("completed_at"),
+            report=job.get("report"),
+            error=job.get("error"),
+        )
+    )
diff --git a/src/sandcastle/api/schemas.py b/src/sandcastle/api/schemas.py
index e9931686..26c7f687 100644
--- a/src/sandcastle/api/schemas.py
+++ b/src/sandcastle/api/schemas.py
@@ -1615,5 +1615,41 @@ class BatchStartedResponse(BaseModel):
     status: str = "running"
 
 
+class TimeMachineStartRequest(BaseModel):
+    """Request to start a Model Time Machine job (counterfactual replay)."""
+
+    target_model: str = Field(..., min_length=1, max_length=200)
+    workflow: str | None = Field(None, min_length=1, max_length=200)
+    since: str | None = Field(
+        None, max_length=64, description="Relative window like '30d'/'12h' or an ISO date"
+    )
+    until: str | None = Field(None, max_length=64, description="ISO date upper bound")
+    max_cassettes: int = Field(20, ge=1, le=500)
+    live: bool = Field(False, description="Re-execute steps with real API calls (needs budget)")
+    budget_usd: float | None = Field(None, gt=0, description="Hard cost cap for live replay")
+    judge_model: str | None = Field(None, min_length=1, max_length=200)
+
+
+class TimeMachineStartResponse(BaseModel):
+    """Response returned immediately when a Time Machine job is accepted."""
+
+    job_id: str
+    status: str = "running"
+    mode: str = "dry_run"  # "dry_run" | "live"
+    target_model: str
+
+
+class TimeMachineJobResponse(BaseModel):
+    """Status + report of a Time Machine job."""
+
+    job_id: str
+    status: str  # "running" | "completed" | "failed"
+    params: dict[str, Any] = {}
+    created_at: str | None = None
+    completed_at: str | None = None
+    report: dict[str, Any] | None = None
+    error: str | None = None
+
+
 # Fix forward reference for ApiResponse.meta
 ApiResponse.model_rebuild()
diff --git a/src/sandcastle/config.py b/src/sandcastle/config.py
index 2b2813ac..acb05568 100644
--- a/src/sandcastle/config.py
+++ b/src/sandcastle/config.py
@@ -125,6 +125,10 @@ class Settings(BaseSettings):
     # Model failover
     failover_cooldown_seconds: float = 60.0
 
+    # Model Time Machine: judge model used to score old vs new outputs on live
+    # replays (any provider-registry model string; default = cheap Claude tier)
+    timemachine_judge_model: str = "haiku"
+
     # Tool connector credentials
     tool_slack_bot_token: str = ""
     tool_jira_api_token: str = ""
diff --git a/src/sandcastle/engine/timemachine.py b/src/sandcastle/engine/timemachine.py
new file mode 100644
index 00000000..b2cd86d3
--- /dev/null
+++ b/src/sandcastle/engine/timemachine.py
@@ -0,0 +1,833 @@
+"""Model Time Machine - counterfactual replay of your real recorded workload.
+
+"What would last month have looked like on a different model?" The Time Machine
+answers that with data instead of benchmarks. It selects recorded run cassettes
+(completed runs whose model steps persisted their resolved prompt, output, cost
+and latency - the same substrate the deterministic cassette/replay system uses),
+then either:
+
+- **dry-run (default, free, instant)**: prices the recorded token volume against
+  the target model's pricing table and projects the cost delta - no API calls; or
+- **live replay (explicit opt-in, budget-capped)**: re-executes every recorded
+  LLM step against the target model, measures real cost and latency, and scores
+  old vs new outputs with an LLM judge (configurable, defaults to a cheap model).
+
+A live replay REQUIRES an explicit ``budget_usd``: the projected replay cost
+(target-model calls + judge calls) is estimated up front from recorded token
+volumes and the job refuses to start if the estimate exceeds the budget. During
+execution the measured spend is re-checked per step and the replay truncates
+rather than overshooting.
+
+Reports are persisted as JSON artifacts under ``{data_dir}/timemachine/`` -
+consistent with the cassette/adapter-registry file substrate, no migration
+needed - and exposed via an in-process async job registry (POST /api/timemachine
+-> job_id, GET /api/timemachine/{job_id} -> status + report).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import re
+import time
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+REPORT_VERSION = 1
+
+# Chars-per-token heuristic used to derive token volumes from recorded text.
+# Token counts are not persisted per step, so ~4 chars/token (the standard
+# rule of thumb for English/code) prices the recorded volume.
+_CHARS_PER_TOKEN = 4
+
+# Map bare Claude aliases to real Anthropic model IDs for direct API calls
+# (same mapping the executor's explicit-llm step uses).
+_CLAUDE_MODEL_ALIASES = {
+    "sonnet": "claude-sonnet-4-6",
+    "haiku": "claude-haiku-4-5",
+    "opus": "claude-opus-4-7",
+}
+
+# Truncation caps for judge inputs - keeps judge calls cheap and bounded.
+_JUDGE_TEXT_CAP = 4000
+_JUDGE_MAX_TOKENS = 64
+_REPLAY_MAX_TOKENS = 4096
+# Cap of per-step detail rows embedded in a report artifact.
+_REPORT_STEP_CAP = 200
+
+_JUDGE_SYSTEM = (
+    "You are an impartial evaluator comparing two AI responses to the same task. "
+    "Score each response 0-10 for correctness, completeness, and instruction-following. "
+    'Respond with ONLY a JSON object: {"a": <score>, "b": <score>}'
+)
+
+
+class TimeMachineError(Exception):
+    """Base error for Time Machine operations."""
+
+
+class BudgetRequiredError(TimeMachineError):
+    """A live replay was requested without an explicit budget_usd cap."""
+
+
+class BudgetExceededError(TimeMachineError):
+    """The pre-flight cost estimate exceeds the explicit budget cap."""
+
+
+# ---------------------------------------------------------------------------
+# Cassette selection (recorded workload from the runs/run_steps substrate)
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class RecordedStep:
+    """One recorded LLM interaction: resolved prompt, output, cost, latency."""
+
+    step_id: str
+    model: str
+    prompt: str
+    output_text: str
+    cost_usd: float
+    duration_seconds: float
+
+
+@dataclass
+class RunCassette:
+    """The recorded LLM steps of one completed run."""
+
+    run_id: str
+    workflow_name: str
+    created_at: datetime | None
+    steps: list[RecordedStep] = field(default_factory=list)
+
+    @property
+    def cost_usd(self) -> float:
+        return sum(s.cost_usd for s in self.steps)
+
+
+def parse_since(spec: str) -> datetime:
+    """Parse a relative window like ``30d`` / ``12h`` or an ISO date into a UTC datetime."""
+    spec = (spec or "").strip()
+    m = re.fullmatch(r"(\d+)\s*([dhw])", spec, re.IGNORECASE)
+    if m:
+        n, unit = int(m.group(1)), m.group(2).lower()
+        delta = {"h": timedelta(hours=n), "d": timedelta(days=n), "w": timedelta(weeks=n)}[unit]
+        return datetime.now(timezone.utc) - delta
+    dt = datetime.fromisoformat(spec)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt
+
+
+def estimate_tokens(text: str) -> int:
+    """Estimate the token count of *text* (~4 chars/token, minimum 1 for non-empty)."""
+    if not text:
+        return 0
+    return max(1, len(text) // _CHARS_PER_TOKEN)
+
+
+def _output_as_text(output: Any) -> str:
+    """Render a recorded step output (str or JSON structure) as comparable text."""
+    if output is None:
+        return ""
+    if isinstance(output, str):
+        return output
+    try:
+        return json.dumps(output, ensure_ascii=False, default=str)
+    except Exception:
+        return str(output)
+
+
+async def select_cassettes(
+    workflow: str | None = None,
+    since: datetime | None = None,
+    until: datetime | None = None,
+    max_cassettes: int = 20,
+    tenant_id: str | None = None,
+) -> list[RunCassette]:
+    """Select recorded run cassettes from the database, newest first.
+
+    A cassette is a completed run with at least one model-bearing step that
+    persisted its resolved prompt (``run_steps.input_prompt``) - exactly the
+    interactions the deterministic cassette recorder captures.
+    """
+    from sqlalchemy import select
+
+    from sandcastle.models.db import Run, RunStatus, RunStep, StepStatus, async_session
+
+    max_cassettes = max(1, min(int(max_cassettes), 500))
+    stmt = select(Run).where(Run.status == RunStatus.COMPLETED)
+    if workflow:
+        stmt = stmt.where(Run.workflow_name == workflow)
+    if since is not None:
+        stmt = stmt.where(Run.created_at >= since)
+    if until is not None:
+        stmt = stmt.where(Run.created_at <= until)
+    if tenant_id is not None:
+        stmt = stmt.where(Run.tenant_id == tenant_id)
+    stmt = stmt.order_by(Run.created_at.desc()).limit(max_cassettes * 3)
+
+    cassettes: list[RunCassette] = []
+    async with async_session() as session:
+        runs = (await session.execute(stmt)).scalars().all()
+        for run in runs:
+            if len(cassettes) >= max_cassettes:
+                break
+            step_stmt = (
+                select(RunStep)
+                .where(RunStep.run_id == run.id, RunStep.status == StepStatus.COMPLETED)
+                .order_by(RunStep.started_at)
+            )
+            rows = (await session.execute(step_stmt)).scalars().all()
+            steps = [
+                RecordedStep(
+                    step_id=s.step_id,
+                    model=s.model or "",
+                    prompt=s.input_prompt or "",
+                    output_text=_output_as_text(s.output_data),
+                    cost_usd=float(s.cost_usd or 0.0),
+                    duration_seconds=float(s.duration_seconds or 0.0),
+                )
+                for s in rows
+                if s.model and s.input_prompt
+            ]
+            if steps:
+                cassettes.append(
+                    RunCassette(
+                        run_id=str(run.id),
+                        workflow_name=run.workflow_name,
+                        created_at=run.created_at,
+                        steps=steps,
+                    )
+                )
+    return cassettes
+
+
+# ---------------------------------------------------------------------------
+# Cost estimation (dry-run pricing - zero API calls)
+# ---------------------------------------------------------------------------
+
+
+def estimate_replay_cost(
+    cassettes: list[RunCassette],
+    target_model: str,
+    judge_model: str | None = None,
+) -> dict[str, Any]:
+    """Price the recorded token volume against the target (and judge) model pricing.
+
+    Returns projected target-model cost for the selection, the projected judge
+    cost for a live replay, and the recorded original spend - all derived from
+    recorded prompt/output text via the chars-per-token heuristic.
+    """
+    from sandcastle.engine.providers import resolve_model
+
+    target = resolve_model(target_model)
+    judge = resolve_model(judge_model) if judge_model else None
+
+    input_tokens = 0
+    output_tokens = 0
+    judge_tokens_in = 0
+    original_cost = 0.0
+    step_count = 0
+    for cas in cassettes:
+        for st in cas.steps:
+            in_tok = estimate_tokens(st.prompt)
+            out_tok = estimate_tokens(st.output_text)
+            input_tokens += in_tok
+            output_tokens += out_tok
+            original_cost += st.cost_usd
+            step_count += 1
+            # Judge sees prompt + both outputs (old + assumed-similar new), capped.
+            judge_tokens_in += min(in_tok, _JUDGE_TEXT_CAP // _CHARS_PER_TOKEN)
+            judge_tokens_in += 2 * min(out_tok, _JUDGE_TEXT_CAP // _CHARS_PER_TOKEN)
+
+    target_cost = (
+        input_tokens * target.input_price_per_m + output_tokens * target.output_price_per_m
+    ) / 1_000_000
+    judge_cost = 0.0
+    if judge is not None:
+        judge_out = step_count * _JUDGE_MAX_TOKENS
+        judge_cost = (
+            judge_tokens_in * judge.input_price_per_m + judge_out * judge.output_price_per_m
+        ) / 1_000_000
+
+    return {
+        "input_tokens": input_tokens,
+        "output_tokens": output_tokens,
+        "step_count": step_count,
+        "original_cost_usd": round(original_cost, 6),
+        "projected_cost_usd": round(target_cost, 6),
+        "projected_judge_cost_usd": round(judge_cost, 6),
+        "projected_total_live_cost_usd": round(target_cost + judge_cost, 6),
+    }
+
+
+# ---------------------------------------------------------------------------
+# Live model + judge calls
+# ---------------------------------------------------------------------------
+
+
+async def call_model(
+    model_str: str,
+    prompt: str,
+    max_tokens: int = _REPLAY_MAX_TOKENS,
+    system: str | None = None,
+    timeout: float = 300.0,
+) -> dict[str, Any]:
+    """Call *model_str* once with *prompt*; return text, token usage, cost and latency.
+
+    Mirrors the executor's explicit-llm step: Anthropic Messages API for Claude
+    models, OpenAI-compatible chat/completions for everything else (including
+    local NIM/Ollama/oMLX, which price at $0).
+    """
+    import httpx
+
+    from sandcastle.engine.providers import get_api_key, resolve_base_url, resolve_model
+
+    info = resolve_model(model_str)
+    api_key = get_api_key(info)
+    started = time.monotonic()
+
+    if info.provider == "claude":
+        api_model = _CLAUDE_MODEL_ALIASES.get(info.api_model_id, info.api_model_id)
+        body: dict[str, Any] = {
+            "model": api_model,
+            "max_tokens": max_tokens,
+            "messages": [{"role": "user", "content": prompt}],
+        }
+        if system:
+            body["system"] = system
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            resp = await client.post(
+                "https://api.anthropic.com/v1/messages",
+                headers={
+                    "x-api-key": api_key,
+                    "anthropic-version": "2023-06-01",
+                    "content-type": "application/json",
+                },
+                json=body,
+            )
+            resp.raise_for_status()
+            data = resp.json()
+        text = data["content"][0]["text"]
+        usage = data.get("usage", {})
+        in_tok = int(usage.get("input_tokens", 0) or 0)
+        out_tok = int(usage.get("output_tokens", 0) or 0)
+    else:
+        base_url = resolve_base_url(info)
+        messages: list[dict[str, str]] = []
+        if system:
+            messages.append({"role": "system", "content": system})
+        messages.append({"role": "user", "content": prompt})
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            resp = await client.post(
+                f"{base_url}/chat/completions",
+                headers={
+                    "Authorization": f"Bearer {api_key}",
+                    "content-type": "application/json",
+                },
+                json={"model": info.api_model_id, "max_tokens": max_tokens, "messages": messages},
+            )
+            resp.raise_for_status()
+            data = resp.json()
+        text = data["choices"][0]["message"]["content"]
+        usage = data.get("usage", {})
+        in_tok = int(usage.get("prompt_tokens", 0) or 0)
+        out_tok = int(usage.get("completion_tokens", 0) or 0)
+
+    latency = time.monotonic() - started
+    cost = (in_tok * info.input_price_per_m + out_tok * info.output_price_per_m) / 1_000_000
+    return {
+        "text": text,
+        "input_tokens": in_tok,
+        "output_tokens": out_tok,
+        "cost_usd": cost,
+        "latency_seconds": latency,
+    }
+
+
+def _parse_judge_scores(raw: str) -> tuple[float | None, float | None]:
+    """Extract {"a": x, "b": y} scores from a judge response, tolerating fences/prose."""
+    text = (raw or "").strip()
+    m = re.search(r"\{[^{}]*\}", text, re.DOTALL)
+    if not m:
+        return None, None
+    try:
+        data = json.loads(m.group(0))
+        a, b = float(data.get("a")), float(data.get("b"))
+    except (TypeError, ValueError, json.JSONDecodeError):
+        return None, None
+    clamp = lambda v: max(0.0, min(10.0, v))  # noqa: E731 - tiny local helper
+    return clamp(a), clamp(b)
+
+
+async def judge_outputs(
+    judge_model: str,
+    prompt: str,
+    old_output: str,
+    new_output: str,
+) -> dict[str, Any]:
+    """Score the original (a) vs replayed (b) output with the judge model, 0-10 each."""
+    user = (
+        f"TASK:\n{prompt[:_JUDGE_TEXT_CAP]}\n\n"
+        f"RESPONSE A:\n{old_output[:_JUDGE_TEXT_CAP]}\n\n"
+        f"RESPONSE B:\n{new_output[:_JUDGE_TEXT_CAP]}\n\n"
+        'JSON scores only: {"a": <0-10>, "b": <0-10>}'
+    )
+    result = await call_model(
+        judge_model, user, max_tokens=_JUDGE_MAX_TOKENS, system=_JUDGE_SYSTEM
+    )
+    score_old, score_new = _parse_judge_scores(result["text"])
+    return {"score_old": score_old, "score_new": score_new, "cost_usd": result["cost_usd"]}
+
+
+# ---------------------------------------------------------------------------
+# The Time Machine itself
+# ---------------------------------------------------------------------------
+
+
+def _pct_delta(old: float, new: float) -> float | None:
+    """Signed percent change from old to new, None when old is 0."""
+    if not old:
+        return None
+    return round((new - old) / old * 100.0, 2)
+
+
+def _avg(values: list[float]) -> float | None:
+    vals = [v for v in values if v is not None]
+    if not vals:
+        return None
+    return sum(vals) / len(vals)
+
+
+def _build_verdict(
+    target_model: str,
+    mode: str,
+    monthly_savings: float,
+    quality_delta_pct: float | None,
+) -> str:
+    """One-line headline: 'Switching to X saves $Y/mo at -Z% quality'."""
+    if monthly_savings >= 0:
+        money = f"saves ${monthly_savings:,.2f}/mo"
+    else:
+        money = f"costs ${abs(monthly_savings):,.2f}/mo more"
+    if mode == "dry_run":
+        return (
+            f"Switching to {target_model} {money} (projected from recorded token volume). "
+            "Run a live replay to measure quality."
+        )
+    if quality_delta_pct is None:
+        return f"Switching to {target_model} {money} (quality not scored)."
+    sign = "+" if quality_delta_pct >= 0 else ""
+    return f"Switching to {target_model} {money} at {sign}{quality_delta_pct:.1f}% quality."
+
+
+async def run_time_machine(
+    target_model: str,
+    workflow: str | None = None,
+    since: datetime | None = None,
+    until: datetime | None = None,
+    max_cassettes: int = 20,
+    live: bool = False,
+    budget_usd: float | None = None,
+    judge_model: str | None = None,
+    tenant_id: str | None = None,
+) -> dict[str, Any]:
+    """Replay the selected recorded workload against *target_model* and report deltas.
+
+    Dry-run (default): pure pricing-table math over recorded token volumes, free.
+    Live: re-executes each recorded LLM step on the target model and judges old vs
+    new output; requires ``budget_usd`` and refuses to start when the pre-flight
+    estimate exceeds it.
+    """
+    from sandcastle.config import settings
+    from sandcastle.engine.providers import resolve_model
+
+    resolve_model(target_model)  # raises KeyError early for unknown models
+    judge = judge_model or getattr(settings, "timemachine_judge_model", "haiku")
+    mode = "live" if live else "dry_run"
+
+    cassettes = await select_cassettes(
+        workflow=workflow,
+        since=since,
+        until=until,
+        max_cassettes=max_cassettes,
+        tenant_id=tenant_id,
+    )
+    estimate = estimate_replay_cost(cassettes, target_model, judge_model=judge if live else None)
+
+    if live:
+        if budget_usd is None:
+            raise BudgetRequiredError(
+                "Live replay makes real API calls - an explicit budget_usd cap is required."
+            )
+        projected = estimate["projected_total_live_cost_usd"]
+        if projected > budget_usd:
+            raise BudgetExceededError(
+                f"Estimated live replay cost ${projected:.4f} exceeds budget "
+                f"${budget_usd:.4f}. Raise the budget or narrow the selection."
+            )
+
+    # -- selection summary ---------------------------------------------------
+    created = [c.created_at for c in cassettes if c.created_at is not None]
+    window_days = 30.0
+    if len(created) >= 2:
+        span = max(created) - min(created)
+        window_days = max(span.total_seconds() / 86400.0, 1.0)
+    elif since is not None:
+        span = (until or datetime.now(timezone.utc)) - since
+        window_days = max(span.total_seconds() / 86400.0, 1.0)
+
+    original_cost = estimate["original_cost_usd"]
+    selection = {
+        "runs": len(cassettes),
+        "steps": estimate["step_count"],
+        "workflows": sorted({c.workflow_name for c in cassettes}),
+        "original_cost_usd": original_cost,
+        "window_days": round(window_days, 2),
+        "first_run_at": min(created).isoformat() if created else None,
+        "last_run_at": max(created).isoformat() if created else None,
+    }
+
+    # -- per-step replay/judging (live) or projection (dry-run) ---------------
+    per_wf: dict[str, dict[str, Any]] = {}
+    step_details: list[dict[str, Any]] = []
+    measured_cost = 0.0
+    steps_replayed = 0
+    steps_failed = 0
+    truncated = False
+
+    for cas in cassettes:
+        agg = per_wf.setdefault(
+            cas.workflow_name,
+            {
+                "workflow": cas.workflow_name,
+                "runs": 0,
+                "steps": 0,
+                "original_cost_usd": 0.0,
+                "new_cost_usd": 0.0,
+                "quality_old": [],
+                "quality_new": [],
+                "latency_old": [],
+                "latency_new": [],
+            },
+        )
+        agg["runs"] += 1
+        for st in cas.steps:
+            agg["steps"] += 1
+            agg["original_cost_usd"] += st.cost_usd
+            agg["latency_old"].append(st.duration_seconds)
+
+            if not live:
+                in_tok = estimate_tokens(st.prompt)
+                out_tok = estimate_tokens(st.output_text)
+                target = resolve_model(target_model)
+                projected_step = (
+                    in_tok * target.input_price_per_m + out_tok * target.output_price_per_m
+                ) / 1_000_000
+                agg["new_cost_usd"] += projected_step
+                continue
+
+            if budget_usd is not None and measured_cost >= budget_usd:
+                truncated = True
+                break
+            detail: dict[str, Any] = {
+                "run_id": cas.run_id,
+                "workflow": cas.workflow_name,
+                "step_id": st.step_id,
+                "old_model": st.model,
+                "old_cost_usd": round(st.cost_usd, 6),
+                "old_latency_seconds": round(st.duration_seconds, 3),
+            }
+            try:
+                replayed = await call_model(target_model, st.prompt)
+                measured_cost += replayed["cost_usd"]
+                steps_replayed += 1
+                agg["new_cost_usd"] += replayed["cost_usd"]
+                agg["latency_new"].append(replayed["latency_seconds"])
+                detail.update(
+                    new_cost_usd=round(replayed["cost_usd"], 6),
+                    new_latency_seconds=round(replayed["latency_seconds"], 3),
+                    new_output_preview=replayed["text"][:500],
+                )
+                try:
+                    scores = await judge_outputs(
+                        judge, st.prompt, st.output_text, replayed["text"]
+                    )
+                    measured_cost += scores["cost_usd"]
+                    if scores["score_old"] is not None:
+                        agg["quality_old"].append(scores["score_old"])
+                        agg["quality_new"].append(scores["score_new"])
+                    detail.update(
+                        score_old=scores["score_old"], score_new=scores["score_new"]
+                    )
+                except Exception as exc:  # noqa: BLE001 - judge failure is non-fatal
+                    logger.warning("Time Machine judge failed for %s: %s", st.step_id, exc)
+                    detail["judge_error"] = str(exc)[:200]
+            except Exception as exc:  # noqa: BLE001 - a step failure must not kill the job
+                logger.warning("Time Machine replay failed for %s: %s", st.step_id, exc)
+                steps_failed += 1
+                detail["error"] = str(exc)[:300]
+            if len(step_details) < _REPORT_STEP_CAP:
+                step_details.append(detail)
+        if truncated:
+            break
+
+    # -- aggregation -----------------------------------------------------------
+    per_workflow: list[dict[str, Any]] = []
+    for wf_name in sorted(per_wf):
+        a = per_wf[wf_name]
+        q_old, q_new = _avg(a["quality_old"]), _avg(a["quality_new"])
+        l_old, l_new = _avg(a["latency_old"]), _avg(a["latency_new"])
+        per_workflow.append(
+            {
+                "workflow": wf_name,
+                "runs": a["runs"],
+                "steps": a["steps"],
+                "original_cost_usd": round(a["original_cost_usd"], 6),
+                "new_cost_usd": round(a["new_cost_usd"], 6),
+                "cost_delta_usd": round(a["new_cost_usd"] - a["original_cost_usd"], 6),
+                "cost_delta_pct": _pct_delta(a["original_cost_usd"], a["new_cost_usd"]),
+                "quality_old": round(q_old, 2) if q_old is not None else None,
+                "quality_new": round(q_new, 2) if q_new is not None else None,
+                "quality_delta_pct": (
+                    _pct_delta(q_old, q_new) if q_old is not None and q_new is not None else None
+                ),
+                "latency_old_seconds": round(l_old, 3) if l_old is not None else None,
+                "latency_new_seconds": round(l_new, 3) if l_new is not None else None,
+                "latency_delta_pct": (
+                    _pct_delta(l_old, l_new) if l_old is not None and l_new is not None else None
+                ),
+            }
+        )
+
+    new_cost_total = sum(w["new_cost_usd"] for w in per_workflow)
+    all_q_old = [s for a in per_wf.values() for s in a["quality_old"]]
+    all_q_new = [s for a in per_wf.values() for s in a["quality_new"]]
+    all_l_old = [s for a in per_wf.values() for s in a["latency_old"]]
+    all_l_new = [s for a in per_wf.values() for s in a["latency_new"]]
+    q_old_avg, q_new_avg = _avg(all_q_old), _avg(all_q_new)
+    l_old_avg, l_new_avg = _avg(all_l_old), _avg(all_l_new)
+    quality_delta_pct = (
+        _pct_delta(q_old_avg, q_new_avg)
+        if q_old_avg is not None and q_new_avg is not None
+        else None
+    )
+
+    # Extrapolate the selection's window to a monthly figure. The cost ratio is
+    # measured (live) or projected (dry-run) on the selection, then applied to
+    # the selection's original spend normalized to 30 days.
+    monthly_original = original_cost * (30.0 / window_days)
+    ratio = (new_cost_total / original_cost) if original_cost else 0.0
+    monthly_projected = monthly_original * ratio
+    monthly_savings = monthly_original - monthly_projected
+
+    report: dict[str, Any] = {
+        "version": REPORT_VERSION,
+        "mode": mode,
+        "target_model": target_model,
+        "judge_model": judge if live else None,
+        "params": {
+            "workflow": workflow,
+            "since": since.isoformat() if since else None,
+            "until": until.isoformat() if until else None,
+            "max_cassettes": max_cassettes,
+            "budget_usd": budget_usd,
+        },
+        "selection": selection,
+        "estimate": estimate,
+        "live": (
+            {
+                "measured_cost_usd": round(measured_cost, 6),
+                "budget_usd": budget_usd,
+                "steps_replayed": steps_replayed,
+                "steps_failed": steps_failed,
+                "truncated": truncated,
+            }
+            if live
+            else None
+        ),
+        "cost": {
+            "original_usd": round(original_cost, 6),
+            "new_usd": round(new_cost_total, 6),
+            "delta_usd": round(new_cost_total - original_cost, 6),
+            "delta_pct": _pct_delta(original_cost, new_cost_total),
+        },
+        "quality": (
+            {
+                "old_avg": round(q_old_avg, 2) if q_old_avg is not None else None,
+                "new_avg": round(q_new_avg, 2) if q_new_avg is not None else None,
+                "delta_pct": quality_delta_pct,
+            }
+            if live
+            else None
+        ),
+        "latency": (
+            {
+                "old_avg_seconds": round(l_old_avg, 3) if l_old_avg is not None else None,
+                "new_avg_seconds": round(l_new_avg, 3) if l_new_avg is not None else None,
+                "delta_pct": (
+                    _pct_delta(l_old_avg, l_new_avg)
+                    if l_old_avg is not None and l_new_avg is not None
+                    else None
+                ),
+            }
+            if live
+            else None
+        ),
+        "extrapolation": {
+            "window_days": round(window_days, 2),
+            "monthly_original_usd": round(monthly_original, 4),
+            "monthly_projected_usd": round(monthly_projected, 4),
+            "monthly_savings_usd": round(monthly_savings, 4),
+        },
+        "per_workflow": per_workflow,
+        "steps": step_details,
+        "verdict": _build_verdict(target_model, mode, monthly_savings, quality_delta_pct),
+    }
+    return report
+
+
+# ---------------------------------------------------------------------------
+# Async job registry + JSON artifact persistence
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class TimeMachineJob:
+    """In-process record of one Time Machine job."""
+
+    job_id: str
+    status: str  # "running" | "completed" | "failed"
+    params: dict[str, Any]
+    created_at: str
+    completed_at: str | None = None
+    report: dict[str, Any] | None = None
+    error: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "job_id": self.job_id,
+            "status": self.status,
+            "params": self.params,
+            "created_at": self.created_at,
+            "completed_at": self.completed_at,
+            "report": self.report,
+            "error": self.error,
+        }
+
+
+_JOBS: dict[str, TimeMachineJob] = {}
+_MAX_JOBS_IN_MEMORY = 50
+
+
+def _artifact_dir() -> Path:
+    from sandcastle.config import settings
+
+    return Path(settings.data_dir) / "timemachine"
+
+
+def _persist_job(job: TimeMachineJob) -> None:
+    try:
+        d = _artifact_dir()
+        d.mkdir(parents=True, exist_ok=True)
+        (d / f"{job.job_id}.json").write_text(json.dumps(job.to_dict(), indent=2, default=str))
+    except Exception as exc:  # noqa: BLE001 - persistence is best-effort
+        logger.warning("Failed to persist Time Machine report %s: %s", job.job_id, exc)
+
+
+def _prune_jobs() -> None:
+    if len(_JOBS) <= _MAX_JOBS_IN_MEMORY:
+        return
+    for job_id in sorted(_JOBS, key=lambda j: _JOBS[j].created_at)[: -_MAX_JOBS_IN_MEMORY]:
+        _JOBS.pop(job_id, None)
+
+
+async def _execute_job(job: TimeMachineJob, kwargs: dict[str, Any]) -> None:
+    try:
+        report = await run_time_machine(**kwargs)
+        job.report = report
+        job.status = "completed"
+    except Exception as exc:  # noqa: BLE001 - job errors surface via status
+        logger.exception("Time Machine job %s failed", job.job_id)
+        job.status = "failed"
+        job.error = str(exc)[:500]
+    job.completed_at = datetime.now(timezone.utc).isoformat()
+    _persist_job(job)
+
+
+def start_job(**kwargs: Any) -> TimeMachineJob:
+    """Create a Time Machine job and run it as a background asyncio task."""
+    job = TimeMachineJob(
+        job_id=str(uuid.uuid4()),
+        status="running",
+        params={
+            k: (v.isoformat() if isinstance(v, datetime) else v)
+            for k, v in kwargs.items()
+            if k != "tenant_id"
+        },
+        created_at=datetime.now(timezone.utc).isoformat(),
+    )
+    _JOBS[job.job_id] = job
+    _prune_jobs()
+    task = asyncio.create_task(_execute_job(job, kwargs))
+    # Keep a reference so the task is not garbage-collected mid-flight.
+    task.add_done_callback(lambda _t: None)
+    return job
+
+
+def get_job(job_id: str) -> dict[str, Any] | None:
+    """Look up a job in memory first, then fall back to its persisted artifact."""
+    job = _JOBS.get(job_id)
+    if job is not None:
+        return job.to_dict()
+    if not re.fullmatch(r"[0-9a-fA-F-]{36}", job_id or ""):
+        return None
+    path = _artifact_dir() / f"{job_id}.json"
+    if path.exists():
+        try:
+            return json.loads(path.read_text())
+        except Exception:  # noqa: BLE001 - corrupt artifact reads as missing
+            return None
+    return None
+
+
+def list_jobs(limit: int = 20) -> list[dict[str, Any]]:
+    """Summaries of recent jobs (memory + disk artifacts), newest first."""
+    seen: dict[str, dict[str, Any]] = {}
+    for job in _JOBS.values():
+        seen[job.job_id] = job.to_dict()
+    try:
+        for path in _artifact_dir().glob("*.json"):
+            if path.stem in seen:
+                continue
+            try:
+                seen[path.stem] = json.loads(path.read_text())
+            except Exception:  # noqa: BLE001
+                continue
+    except OSError:
+        pass
+    jobs = sorted(seen.values(), key=lambda j: j.get("created_at") or "", reverse=True)[:limit]
+    summaries = []
+    for j in jobs:
+        report = j.get("report") or {}
+        summaries.append(
+            {
+                "job_id": j.get("job_id"),
+                "status": j.get("status"),
+                "created_at": j.get("created_at"),
+                "completed_at": j.get("completed_at"),
+                "mode": report.get("mode") or ("live" if (j.get("params") or {}).get("live") else "dry_run"),
+                "target_model": report.get("target_model")
+                or (j.get("params") or {}).get("target_model"),
+                "verdict": report.get("verdict"),
+                "error": j.get("error"),
+            }
+        )
+    return summaries
diff --git a/tests/test_timemachine.py b/tests/test_timemachine.py
new file mode 100644
index 00000000..fe2ae1b6
--- /dev/null
+++ b/tests/test_timemachine.py
@@ -0,0 +1,446 @@
+"""Tests for the Model Time Machine - counterfactual replay against a different model.
+
+Covers: dry-run pricing math against hand-computed fixtures, cassette selection
+filters, the explicit-budget contract for live replays (required + pre-flight
+refusal), live-path aggregation with mocked model/judge calls, and the async
+job lifecycle over the API.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+import uuid
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi.testclient import TestClient
+
+from sandcastle.engine import timemachine as tm
+from sandcastle.main import app
+
+client = TestClient(app)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _wf_name(prefix: str) -> str:
+    return f"{prefix}-{uuid.uuid4().hex[:8]}"
+
+
+async def _seed_run(
+    workflow_name: str,
+    steps: list[dict],
+    created_at: datetime | None = None,
+) -> str:
+    """Insert a completed Run with completed RunSteps and return the run id."""
+    from sandcastle.models.db import Run, RunStatus, RunStep, StepStatus, async_session
+
+    run_id = uuid.uuid4()
+    async with async_session() as session:
+        session.add(
+            Run(
+                id=run_id,
+                workflow_name=workflow_name,
+                status=RunStatus.COMPLETED,
+                total_cost_usd=sum(s.get("cost_usd", 0.0) for s in steps),
+                created_at=created_at or datetime.now(timezone.utc),
+            )
+        )
+        for i, s in enumerate(steps):
+            session.add(
+                RunStep(
+                    run_id=run_id,
+                    step_id=s.get("step_id", f"step{i}"),
+                    status=StepStatus.COMPLETED,
+                    input_prompt=s.get("prompt"),
+                    output_data={"text": s.get("output", "")},
+                    cost_usd=s.get("cost_usd", 0.0),
+                    duration_seconds=s.get("duration", 1.0),
+                    model=s.get("model"),
+                    started_at=datetime.now(timezone.utc) + timedelta(seconds=i),
+                )
+            )
+        await session.commit()
+    return str(run_id)
+
+
+# ---------------------------------------------------------------------------
+# Dry-run pricing math (hand-computed fixture)
+# ---------------------------------------------------------------------------
+
+
+class TestEstimateMath:
+    def test_estimate_tokens_chars_per_token(self):
+        assert tm.estimate_tokens("") == 0
+        assert tm.estimate_tokens("ab") == 1  # minimum 1 for non-empty
+        assert tm.estimate_tokens("x" * 400) == 100
+
+    def test_projected_cost_hand_computed(self):
+        # 4000 chars prompt -> 1000 input tokens; 8000 chars output -> 2000 output
+        # tokens. haiku pricing: $0.80/M input, $4.00/M output.
+        cas = [
+            tm.RunCassette(
+                run_id="r1",
+                workflow_name="wf",
+                created_at=datetime.now(timezone.utc),
+                steps=[tm.RecordedStep("s1", "sonnet", "x" * 4000, "y" * 8000, 0.05, 2.0)],
+            )
+        ]
+        est = tm.estimate_replay_cost(cas, "haiku")
+        assert est["input_tokens"] == 1000
+        assert est["output_tokens"] == 2000
+        expected = (1000 * 0.80 + 2000 * 4.0) / 1_000_000
+        assert est["projected_cost_usd"] == pytest.approx(expected)
+        assert est["original_cost_usd"] == pytest.approx(0.05)
+        # No judge model -> no judge cost, total == target cost
+        assert est["projected_judge_cost_usd"] == 0.0
+        assert est["projected_total_live_cost_usd"] == pytest.approx(expected)
+
+    def test_judge_cost_included_for_live_estimates(self):
+        cas = [
+            tm.RunCassette(
+                run_id="r1",
+                workflow_name="wf",
+                created_at=None,
+                steps=[tm.RecordedStep("s1", "sonnet", "x" * 400, "y" * 400, 0.01, 1.0)],
+            )
+        ]
+        est = tm.estimate_replay_cost(cas, "haiku", judge_model="haiku")
+        assert est["projected_judge_cost_usd"] > 0
+        assert est["projected_total_live_cost_usd"] == pytest.approx(
+            est["projected_cost_usd"] + est["projected_judge_cost_usd"]
+        )
+
+    def test_local_model_projects_to_zero(self):
+        # nim/* resolves with region=local and $0 pricing.
+        cas = [
+            tm.RunCassette(
+                run_id="r1",
+                workflow_name="wf",
+                created_at=None,
+                steps=[tm.RecordedStep("s1", "opus", "p" * 4000, "o" * 4000, 1.25, 3.0)],
+            )
+        ]
+        est = tm.estimate_replay_cost(cas, "nim/llama-3.1-70b")
+        assert est["projected_cost_usd"] == 0.0
+        assert est["original_cost_usd"] == pytest.approx(1.25)
+
+
+class TestParseSince:
+    def test_relative_windows(self):
+        now = datetime.now(timezone.utc)
+        assert (now - tm.parse_since("30d")).days in (29, 30)
+        assert abs((now - tm.parse_since("12h")).total_seconds() - 43200) < 5
+        assert (now - tm.parse_since("2w")).days in (13, 14)
+
+    def test_iso_date(self):
+        dt = tm.parse_since("2026-01-15")
+        assert dt == datetime(2026, 1, 15, tzinfo=timezone.utc)
+
+    def test_invalid_raises(self):
+        with pytest.raises(ValueError):
+            tm.parse_since("next tuesday")
+
+
+# ---------------------------------------------------------------------------
+# Cassette selection
+# ---------------------------------------------------------------------------
+
+
+class TestSelectCassettes:
+    async def test_selects_only_model_steps_of_completed_runs(self):
+        wf = _wf_name("tm-select")
+        await _seed_run(
+            wf,
+            [
+                {"step_id": "llm", "model": "sonnet", "prompt": "p", "output": "o",
+                 "cost_usd": 0.01},
+                {"step_id": "code", "model": None, "prompt": None, "output": "x"},
+            ],
+        )
+        cassettes = await tm.select_cassettes(workflow=wf)
+        assert len(cassettes) == 1
+        assert [s.step_id for s in cassettes[0].steps] == ["llm"]
+        assert cassettes[0].steps[0].output_text == '{"text": "o"}'
+
+    async def test_workflow_filter_and_cap(self):
+        wf_a, wf_b = _wf_name("tm-a"), _wf_name("tm-b")
+        for _ in range(3):
+            await _seed_run(wf_a, [{"model": "haiku", "prompt": "p", "output": "o"}])
+        await _seed_run(wf_b, [{"model": "haiku", "prompt": "p", "output": "o"}])
+        assert len(await tm.select_cassettes(workflow=wf_a)) == 3
+        assert len(await tm.select_cassettes(workflow=wf_a, max_cassettes=2)) == 2
+        assert len(await tm.select_cassettes(workflow=wf_b)) == 1
+
+    async def test_date_range_filter(self):
+        wf = _wf_name("tm-dates")
+        old = datetime.now(timezone.utc) - timedelta(days=60)
+        await _seed_run(wf, [{"model": "haiku", "prompt": "p", "output": "o"}], created_at=old)
+        await _seed_run(wf, [{"model": "haiku", "prompt": "p", "output": "o"}])
+        since = datetime.now(timezone.utc) - timedelta(days=30)
+        assert len(await tm.select_cassettes(workflow=wf)) == 2
+        assert len(await tm.select_cassettes(workflow=wf, since=since)) == 1
+
+
+# ---------------------------------------------------------------------------
+# Dry-run report (default mode - no API calls)
+# ---------------------------------------------------------------------------
+
+
+class TestDryRun:
+    async def test_dry_run_report_math_and_shape(self):
+        wf = _wf_name("tm-dry")
+        await _seed_run(
+            wf,
+            [{"model": "sonnet", "prompt": "x" * 4000, "output": "y" * 8000,
+              "cost_usd": 0.10, "duration": 2.0}],
+        )
+        report = await tm.run_time_machine("haiku", workflow=wf)
+        assert report["mode"] == "dry_run"
+        assert report["quality"] is None and report["live"] is None
+        assert report["selection"]["runs"] == 1
+        # Output JSON-serialized adds {"text": ...} wrapper: 8000 + 12 chars -> 2003 tokens
+        out_tokens = (8000 + len('{"text": ""}')) // 4
+        expected = (1000 * 0.80 + out_tokens * 4.0) / 1_000_000
+        assert report["cost"]["new_usd"] == pytest.approx(expected, rel=1e-6)
+        assert report["cost"]["original_usd"] == pytest.approx(0.10)
+        assert report["cost"]["delta_usd"] < 0
+        assert len(report["per_workflow"]) == 1
+        row = report["per_workflow"][0]
+        assert row["workflow"] == wf and row["steps"] == 1
+        assert "projected" in report["verdict"]
+        assert report["extrapolation"]["monthly_savings_usd"] > 0
+
+    async def test_dry_run_empty_selection(self):
+        report = await tm.run_time_machine("haiku", workflow=_wf_name("tm-none"))
+        assert report["selection"]["runs"] == 0
+        assert report["cost"]["original_usd"] == 0.0
+        assert report["per_workflow"] == []
+
+    async def test_unknown_model_raises(self):
+        with pytest.raises(KeyError):
+            await tm.run_time_machine("gpt-42-ultra", workflow="whatever")
+
+
+# ---------------------------------------------------------------------------
+# Budget contract for live replays
+# ---------------------------------------------------------------------------
+
+
+class TestBudget:
+    async def test_live_requires_budget(self):
+        with pytest.raises(tm.BudgetRequiredError):
+            await tm.run_time_machine("haiku", workflow=_wf_name("x"), live=True)
+
+    async def test_live_refuses_when_estimate_exceeds_budget(self):
+        wf = _wf_name("tm-budget")
+        await _seed_run(
+            wf,
+            [{"model": "sonnet", "prompt": "x" * 400000, "output": "y" * 400000,
+              "cost_usd": 2.0}],
+        )
+        # opus replay of 100k in + 100k out tokens >> $0.000001
+        with pytest.raises(tm.BudgetExceededError):
+            await tm.run_time_machine("opus", workflow=wf, live=True, budget_usd=0.000001)
+
+
+# ---------------------------------------------------------------------------
+# Live replay aggregation (mocked model + judge)
+# ---------------------------------------------------------------------------
+
+
+class TestLiveAggregation:
+    async def test_live_aggregates_cost_quality_latency(self, monkeypatch):
+        wf = _wf_name("tm-live")
+        await _seed_run(
+            wf,
+            [
+                {"step_id": "a", "model": "sonnet", "prompt": "p1", "output": "o1",
+                 "cost_usd": 0.10, "duration": 4.0},
+                {"step_id": "b", "model": "sonnet", "prompt": "p2", "output": "o2",
+                 "cost_usd": 0.30, "duration": 6.0},
+            ],
+        )
+
+        async def fake_call_model(model_str, prompt, max_tokens=4096, system=None, timeout=300.0):
+            return {
+                "text": f"new output for {prompt}",
+                "input_tokens": 100,
+                "output_tokens": 50,
+                "cost_usd": 0.02,
+                "latency_seconds": 1.0,
+            }
+
+        async def fake_judge(judge_model, prompt, old_output, new_output):
+            return {"score_old": 8.0, "score_new": 6.0, "cost_usd": 0.001}
+
+        monkeypatch.setattr(tm, "call_model", fake_call_model)
+        monkeypatch.setattr(tm, "judge_outputs", fake_judge)
+
+        report = await tm.run_time_machine("haiku", workflow=wf, live=True, budget_usd=10.0)
+
+        assert report["mode"] == "live"
+        live = report["live"]
+        assert live["steps_replayed"] == 2 and live["steps_failed"] == 0
+        assert live["truncated"] is False
+        # measured = 2 replay calls + 2 judge calls
+        assert live["measured_cost_usd"] == pytest.approx(2 * 0.02 + 2 * 0.001)
+        # new cost excludes judge overhead; original 0.40 -> 0.04
+        assert report["cost"]["new_usd"] == pytest.approx(0.04)
+        assert report["cost"]["original_usd"] == pytest.approx(0.40)
+        # quality: 8.0 -> 6.0 = -25%
+        assert report["quality"]["old_avg"] == pytest.approx(8.0)
+        assert report["quality"]["new_avg"] == pytest.approx(6.0)
+        assert report["quality"]["delta_pct"] == pytest.approx(-25.0)
+        # latency: avg 5.0s -> 1.0s = -80%
+        assert report["latency"]["old_avg_seconds"] == pytest.approx(5.0)
+        assert report["latency"]["new_avg_seconds"] == pytest.approx(1.0)
+        assert report["latency"]["delta_pct"] == pytest.approx(-80.0)
+        row = report["per_workflow"][0]
+        assert row["quality_delta_pct"] == pytest.approx(-25.0)
+        assert "haiku" in report["verdict"]
+
+    async def test_live_step_failure_is_non_fatal(self, monkeypatch):
+        wf = _wf_name("tm-fail")
+        await _seed_run(
+            wf,
+            [
+                {"step_id": "ok", "model": "sonnet", "prompt": "p1", "output": "o1",
+                 "cost_usd": 0.1},
+                {"step_id": "boom", "model": "sonnet", "prompt": "p2", "output": "o2",
+                 "cost_usd": 0.1},
+            ],
+        )
+        calls = {"n": 0}
+
+        async def flaky_call(model_str, prompt, max_tokens=4096, system=None, timeout=300.0):
+            calls["n"] += 1
+            if prompt == "p2":
+                raise RuntimeError("provider 500")
+            return {"text": "ok", "input_tokens": 10, "output_tokens": 10,
+                    "cost_usd": 0.01, "latency_seconds": 0.5}
+
+        async def fake_judge(judge_model, prompt, old_output, new_output):
+            return {"score_old": 7.0, "score_new": 7.0, "cost_usd": 0.0}
+
+        monkeypatch.setattr(tm, "call_model", flaky_call)
+        monkeypatch.setattr(tm, "judge_outputs", fake_judge)
+
+        report = await tm.run_time_machine("haiku", workflow=wf, live=True, budget_usd=10.0)
+        assert report["live"]["steps_replayed"] == 1
+        assert report["live"]["steps_failed"] == 1
+        failed = [s for s in report["steps"] if s.get("error")]
+        assert len(failed) == 1 and "provider 500" in failed[0]["error"]
+
+
+# ---------------------------------------------------------------------------
+# Judge response parsing
+# ---------------------------------------------------------------------------
+
+
+class TestJudgeParsing:
+    def test_parses_plain_json(self):
+        assert tm._parse_judge_scores('{"a": 7, "b": 9.5}') == (7.0, 9.5)
+
+    def test_parses_fenced_and_prose(self):
+        raw = 'Here are my scores:\n```json\n{"a": 3, "b": 11}\n```'
+        a, b = tm._parse_judge_scores(raw)
+        assert a == 3.0 and b == 10.0  # clamped to 0-10
+
+    def test_garbage_returns_none(self):
+        assert tm._parse_judge_scores("I refuse to answer") == (None, None)
+        assert tm._parse_judge_scores("") == (None, None)
+
+
+# ---------------------------------------------------------------------------
+# API job lifecycle
+# ---------------------------------------------------------------------------
+
+
+class TestTimeMachineApi:
+    def test_dry_run_job_lifecycle(self, monkeypatch, tmp_path):
+        from sandcastle.config import settings
+
+        monkeypatch.setattr(settings, "data_dir", str(tmp_path))
+
+        canned = {"mode": "dry_run", "target_model": "haiku", "verdict": "ok",
+                  "selection": {"runs": 0}}
+
+        async def fake_run(**kwargs):
+            return canned
+
+        monkeypatch.setattr(tm, "run_time_machine", fake_run)
+
+        with TestClient(app) as c:
+            resp = c.post("/api/timemachine", json={"target_model": "haiku"})
+            assert resp.status_code == 202, resp.text
+            data = resp.json()["data"]
+            job_id = data["job_id"]
+            assert data["mode"] == "dry_run"
+
+            job = None
+            for _ in range(100):
+                job = c.get(f"/api/timemachine/{job_id}").json()["data"]
+                if job["status"] != "running":
+                    break
+                time.sleep(0.05)
+            assert job is not None and job["status"] == "completed", job
+            assert job["report"]["verdict"] == "ok"
+
+            # job appears in the list endpoint
+            listing = c.get("/api/timemachine").json()["data"]
+            assert any(j["job_id"] == job_id for j in listing)
+
+        # artifact persisted to {data_dir}/timemachine/{job_id}.json
+        assert (tmp_path / "timemachine" / f"{job_id}.json").exists()
+        # and readable via get_job after the in-memory registry is cleared
+        tm._JOBS.pop(job_id, None)
+        from_disk = tm.get_job(job_id)
+        assert from_disk is not None and from_disk["status"] == "completed"
+
+    def test_live_without_budget_is_rejected(self):
+        resp = client.post(
+            "/api/timemachine", json={"target_model": "haiku", "live": True}
+        )
+        assert resp.status_code == 400
+        assert resp.json()["detail"]["error"]["code"] == "BUDGET_REQUIRED"
+
+    def test_live_over_budget_is_refused_preflight(self):
+        wf = _wf_name("tm-api-budget")
+        asyncio.run(
+            _seed_run(
+                wf,
+                [{"model": "sonnet", "prompt": "x" * 400000, "output": "y" * 400000,
+                  "cost_usd": 2.0}],
+            )
+        )
+        resp = client.post(
+            "/api/timemachine",
+            json={"target_model": "opus", "workflow": wf, "live": True,
+                  "budget_usd": 0.0001},
+        )
+        assert resp.status_code == 400, resp.text
+        assert resp.json()["detail"]["error"]["code"] == "BUDGET_EXCEEDED"
+
+    def test_unknown_model_rejected(self):
+        resp = client.post("/api/timemachine", json={"target_model": "gpt-42-ultra"})
+        assert resp.status_code == 400
+        assert resp.json()["detail"]["error"]["code"] == "UNKNOWN_MODEL"
+
+    def test_invalid_since_rejected(self):
+        resp = client.post(
+            "/api/timemachine", json={"target_model": "haiku", "since": "soonish"}
+        )
+        assert resp.status_code == 422
+
+    def test_missing_job_404(self):
+        resp = client.get(f"/api/timemachine/{uuid.uuid4()}")
+        assert resp.status_code == 404
+
+    def test_job_id_path_traversal_safe(self):
+        resp = client.get("/api/timemachine/..%2F..%2Fetc%2Fpasswd")
+        assert resp.status_code in (404, 422)