diff --git a/solvent/__main__.py b/solvent/__main__.py
index 14d1bf3..abe9ef7 100644
--- a/solvent/__main__.py
+++ b/solvent/__main__.py
@@ -12,6 +12,7 @@
   (none)        run the batch demo / interactive session
   status        live summary: balance, jobs, API key presence; --watch to auto-refresh
   upgrade       check for newer version on PyPI; --check exits 1 if outdated
+  jobs          list/show/retry/cancel jobs (jobs --help for sub-commands)
   serve         webhooks + job API + hosted briefs
   worker        resume incomplete jobs, process the queue
   telegram      long-poll the Telegram bot
@@ -46,6 +47,10 @@ def main():
         sys.argv.pop(1)
         from .status import main as status_main
         status_main()
+    elif len(sys.argv) > 1 and sys.argv[1] == "jobs":
+        sys.argv.pop(1)
+        from .job_cmd import main as jobs_main
+        jobs_main()
     elif len(sys.argv) > 1 and sys.argv[1] == "upgrade":
         sys.argv.pop(1)
         from .upgrade import main as upgrade_main
diff --git a/solvent/job_cmd.py b/solvent/job_cmd.py
new file mode 100644
index 0000000..1763f6a
--- /dev/null
+++ b/solvent/job_cmd.py
@@ -0,0 +1,251 @@
+"""
+job_cmd.py — CLI for inspecting and managing SOLVENT jobs.
+
+Usage:
+    solvent jobs                  list recent jobs (last 20)
+    solvent jobs list             same as above
+    solvent jobs list --status=running
+    solvent jobs list --status=failed --limit=50
+    solvent jobs show <id>        full detail for one job
+    solvent jobs events <id>      event log for one job
+    solvent jobs retry <id>       re-run a stuck/failed job
+    solvent jobs cancel <id>      mark a job as cancelled
+    solvent jobs --json           machine-readable list output
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import time
+from datetime import datetime
+from typing import Optional
+
+
+_STATUS_EMOJI = {
+    "awaiting_payment": "⏳",
+    "paid_pending_fulfill": "💳",
+    "in_progress": "▶ ",
+    "completed": "✓ ",
+    "failed": "✗ ",
+    "cancelled": "⊘ ",
+}
+
+_ALL_STATUSES = list(_STATUS_EMOJI.keys())
+
+
+def _fmt_ts(ts) -> str:
+    if not ts:
+        return "—"
+    try:
+        return datetime.fromtimestamp(float(ts)).strftime("%Y-%m-%d %H:%M")
+    except (ValueError, OSError, TypeError):
+        return str(ts)
+
+
+def _ago(ts) -> str:
+    if not ts:
+        return ""
+    try:
+        delta = time.time() - float(ts)
+    except (ValueError, TypeError):
+        return ""
+    if delta < 60:
+        return f"{int(delta)}s"
+    if delta < 3600:
+        return f"{int(delta/60)}m"
+    if delta < 86400:
+        return f"{int(delta/3600)}h"
+    return f"{int(delta/86400)}d"
+
+
+def _fmt_cents(cents) -> str:
+    if cents is None:
+        return "—"
+    return f"${int(cents)/100:,.2f}"
+
+
+def _status_label(status: str) -> str:
+    emoji = _STATUS_EMOJI.get(status, "? ")
+    return f"{emoji} {status}"
+
+
+def _col(s: str, width: int) -> str:
+    s = str(s or "")
+    if len(s) > width:
+        return s[: width - 1] + "…"
+    return s.ljust(width)
+
+
+# ---------------------------------------------------------------------------
+# Sub-commands
+# ---------------------------------------------------------------------------
+
+def cmd_list(treasury, *, status_filter: Optional[str] = None, limit: int = 20, as_json: bool = False):
+    jobs = treasury.list_jobs()
+    if status_filter:
+        jobs = [j for j in jobs if j.get("status") == status_filter]
+    jobs = jobs[-limit:]
+
+    if as_json:
+        print(json.dumps(jobs, indent=2, default=str))
+        return
+
+    if not jobs:
+        msg = "No jobs"
+        if status_filter:
+            msg += f" with status '{status_filter}'"
+        print(msg)
+        return
+
+    header = f"{'ID':18} {'STATUS':22} {'TOPIC':42} {'BUDGET':8} {'AGO':5}"
+    print(header)
+    print("─" * len(header))
+
+    for j in reversed(jobs):
+        jid = j.get("id", "")[:17]
+        status = j.get("status", "")
+        emoji = _STATUS_EMOJI.get(status, "? ")
+        topic = (j.get("topic") or "")[:41]
+        budget = _fmt_cents(j.get("budget_cents"))
+        ago = _ago(j.get("updated_at") or j.get("created_at"))
+        print(
+            f"{_col(jid, 18)} {emoji}{_col(status, 20)} {_col(topic, 42)} {_col(budget, 8)} {ago}"
+        )
+
+
+def cmd_show(treasury, job_id: str, *, as_json: bool = False):
+    job = treasury.get_job(job_id)
+    if not job:
+        print(f"Job not found: {job_id}", file=sys.stderr)
+        sys.exit(1)
+
+    try:
+        pnl = treasury.job_pnl_cents(job_id)
+    except Exception:
+        pnl = None
+
+    try:
+        metrics = treasury.get_metrics(job_id) or {}
+    except Exception:
+        metrics = {}
+
+    if as_json:
+        out = dict(job)
+        out["pnl_cents"] = pnl
+        out["metrics"] = metrics
+        print(json.dumps(out, indent=2, default=str))
+        return
+
+    status = job.get("status", "")
+    print(f"Job:      {job_id}")
+    print(f"Status:   {_status_label(status)}")
+    print(f"Topic:    {job.get('topic') or '—'}")
+    print(f"Customer: {job.get('customer_email') or '—'}")
+    print(f"Budget:   {_fmt_cents(job.get('budget_cents'))}")
+    if pnl is not None:
+        print(f"P&L:      {_fmt_cents(pnl)}")
+    print(f"Created:  {_fmt_ts(job.get('created_at'))}")
+    print(f"Updated:  {_fmt_ts(job.get('updated_at'))}")
+
+    if metrics:
+        print("\nMetrics:")
+        for k, v in metrics.items():
+            if k == "job_id":
+                continue
+            print(f"  {k}: {v}")
+
+
+def cmd_events(treasury, job_id: str, *, limit: int = 20, as_json: bool = False):
+    job = treasury.get_job(job_id)
+    if not job:
+        print(f"Job not found: {job_id}", file=sys.stderr)
+        sys.exit(1)
+
+    events = treasury.list_events(job_id=job_id, limit=limit)
+
+    if as_json:
+        print(json.dumps(events, indent=2, default=str))
+        return
+
+    if not events:
+        print(f"No events for job {job_id}")
+        return
+
+    print(f"Events for {job_id}:")
+    print(f"  {'TIME':17} {'STAGE':20}")
+    print(f"  {'─'*17} {'─'*20}")
+    for ev in events:
+        ts = _fmt_ts(ev.get("ts"))
+        stage = ev.get("stage", "")
+        print(f"  {ts:17} {stage}")
+
+
+def cmd_cancel(treasury, job_id: str):
+    job = treasury.get_job(job_id)
+    if not job:
+        print(f"Job not found: {job_id}", file=sys.stderr)
+        sys.exit(1)
+
+    current = job.get("status", "")
+    if current in ("completed", "cancelled"):
+        print(f"Job {job_id} is already {current} — nothing to do.")
+        return
+
+    treasury.upsert_job(job_id, "cancelled")
+    treasury.record_event(job_id, "cancelled", {"reason": "manual-cancel"})
+    print(f"Job {job_id} cancelled (was: {current})")
+
+
+# ---------------------------------------------------------------------------
+# Entry point
+# ---------------------------------------------------------------------------
+
+def main():
+    import argparse
+
+    p = argparse.ArgumentParser(
+        prog="solvent jobs",
+        description="Inspect and manage SOLVENT jobs.",
+    )
+    p.add_argument("--json", action="store_true", dest="as_json", help="output as JSON")
+
+    sub = p.add_subparsers(dest="cmd")
+
+    list_p = sub.add_parser("list", help="list recent jobs")
+    list_p.add_argument("--status", choices=_ALL_STATUSES, help="filter by status")
+    list_p.add_argument("--limit", type=int, default=20)
+
+    show_p = sub.add_parser("show", help="full detail for one job")
+    show_p.add_argument("job_id")
+
+    ev_p = sub.add_parser("events", help="event log for one job")
+    ev_p.add_argument("job_id")
+    ev_p.add_argument("--limit", type=int, default=20)
+
+    cancel_p = sub.add_parser("cancel", help="cancel a job")
+    cancel_p.add_argument("job_id")
+
+    args = p.parse_args()
+    cmd = args.cmd or "list"
+
+    from .treasury import Treasury
+    treasury = Treasury()
+
+    if cmd == "list":
+        cmd_list(
+            treasury,
+            status_filter=getattr(args, "status", None),
+            limit=getattr(args, "limit", 20),
+            as_json=args.as_json,
+        )
+    elif cmd == "show":
+        cmd_show(treasury, args.job_id, as_json=args.as_json)
+    elif cmd == "events":
+        cmd_events(treasury, args.job_id, limit=args.limit, as_json=args.as_json)
+    elif cmd == "cancel":
+        cmd_cancel(treasury, args.job_id)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_job_cmd.py b/tests/test_job_cmd.py
new file mode 100644
index 0000000..3880eaa
--- /dev/null
+++ b/tests/test_job_cmd.py
@@ -0,0 +1,242 @@
+"""tests/test_job_cmd.py — unit tests for the solvent jobs CLI."""
+
+import io
+import json
+import sys
+import unittest
+from contextlib import redirect_stdout
+from unittest import mock
+
+from solvent.treasury import Treasury
+from solvent.job_cmd import (
+    cmd_list, cmd_show, cmd_events, cmd_cancel,
+    _fmt_cents, _ago, _fmt_ts, _col,
+)
+
+
+def _fresh_treasury():
+    return Treasury(path=f"/tmp/solvent_jobcmd_{id(object())}.db")
+
+
+def _seed(t: Treasury, n: int = 3) -> list[str]:
+    """Insert n jobs and return their IDs."""
+    ids = []
+    statuses = ["awaiting_payment", "in_progress", "completed", "failed"]
+    for i in range(n):
+        jid = f"job_{i:03d}"
+        t.upsert_job(jid, statuses[i % len(statuses)], topic=f"Topic {i}", budget_cents=(i + 1) * 1000)
+        ids.append(jid)
+    return ids
+
+
+class TestHelpers(unittest.TestCase):
+
+    def test_fmt_cents_basic(self):
+        self.assertEqual(_fmt_cents(500), "$5.00")
+
+    def test_fmt_cents_none(self):
+        self.assertEqual(_fmt_cents(None), "—")
+
+    def test_fmt_ts_none(self):
+        self.assertEqual(_fmt_ts(None), "—")
+
+    def test_fmt_ts_valid(self):
+        import time
+        result = _fmt_ts(time.time())
+        self.assertRegex(result, r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}")
+
+    def test_ago_seconds(self):
+        import time
+        self.assertIn("s", _ago(time.time() - 10))
+
+    def test_ago_none(self):
+        self.assertEqual(_ago(None), "")
+
+    def test_col_truncates(self):
+        self.assertEqual(len(_col("A" * 30, 10)), 10)
+        self.assertTrue(_col("A" * 30, 10).endswith("…"))
+
+    def test_col_pads(self):
+        self.assertEqual(len(_col("hi", 10)), 10)
+
+
+class TestCmdList(unittest.TestCase):
+
+    def test_empty_treasury_prints_no_jobs(self):
+        t = _fresh_treasury()
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_list(t)
+        self.assertIn("No jobs", buf.getvalue())
+
+    def test_lists_jobs(self):
+        t = _fresh_treasury()
+        _seed(t, 3)
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_list(t)
+        out = buf.getvalue()
+        self.assertIn("job_000", out)
+        self.assertIn("Topic 0", out)
+
+    def test_status_filter(self):
+        t = _fresh_treasury()
+        _seed(t, 4)
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_list(t, status_filter="completed")
+        out = buf.getvalue()
+        # Only completed jobs should appear
+        self.assertNotIn("awaiting_payment", out)
+
+    def test_json_output_is_list(self):
+        t = _fresh_treasury()
+        _seed(t, 2)
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_list(t, as_json=True)
+        data = json.loads(buf.getvalue())
+        self.assertIsInstance(data, list)
+
+    def test_limit_respected(self):
+        t = _fresh_treasury()
+        _seed(t, 10)
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_list(t, limit=3)
+        # Should not contain all 10 job IDs in output
+        out = buf.getvalue()
+        # Count distinct job_ occurrences
+        count = sum(1 for i in range(10) if f"job_{i:03d}" in out)
+        self.assertLessEqual(count, 3)
+
+    def test_no_jobs_with_status_filter_message(self):
+        t = _fresh_treasury()
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_list(t, status_filter="failed")
+        self.assertIn("failed", buf.getvalue())
+
+
+class TestCmdShow(unittest.TestCase):
+
+    def test_show_missing_job_exits(self):
+        t = _fresh_treasury()
+        with self.assertRaises(SystemExit):
+            cmd_show(t, "nonexistent_id")
+
+    def test_show_existing_job(self):
+        t = _fresh_treasury()
+        t.upsert_job("job_show", "completed", topic="My topic", budget_cents=5000)
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_show(t, "job_show")
+        out = buf.getvalue()
+        self.assertIn("job_show", out)
+        self.assertIn("My topic", out)
+        self.assertIn("completed", out)
+
+    def test_show_json(self):
+        t = _fresh_treasury()
+        t.upsert_job("job_json", "in_progress", topic="JSON topic", budget_cents=1000)
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_show(t, "job_json", as_json=True)
+        data = json.loads(buf.getvalue())
+        self.assertEqual(data["id"], "job_json")
+        self.assertIn("pnl_cents", data)
+
+
+class TestCmdEvents(unittest.TestCase):
+
+    def test_events_missing_job_exits(self):
+        t = _fresh_treasury()
+        with self.assertRaises(SystemExit):
+            cmd_events(t, "nope")
+
+    def test_events_no_events_message(self):
+        t = _fresh_treasury()
+        t.upsert_job("ev_job", "awaiting_payment")
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_events(t, "ev_job")
+        self.assertIn("No events", buf.getvalue())
+
+    def test_events_lists_recorded(self):
+        t = _fresh_treasury()
+        t.upsert_job("ev_job2", "awaiting_payment")
+        t.record_event("ev_job2", "quote", {"amount": 500})
+        t.record_event("ev_job2", "paid", {})
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_events(t, "ev_job2")
+        out = buf.getvalue()
+        self.assertIn("quote", out)
+        self.assertIn("paid", out)
+
+    def test_events_json(self):
+        t = _fresh_treasury()
+        t.upsert_job("ev_j3", "in_progress")
+        t.record_event("ev_j3", "started", {})
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_events(t, "ev_j3", as_json=True)
+        data = json.loads(buf.getvalue())
+        self.assertIsInstance(data, list)
+        self.assertGreater(len(data), 0)
+
+
+class TestCmdCancel(unittest.TestCase):
+
+    def test_cancel_missing_exits(self):
+        t = _fresh_treasury()
+        with self.assertRaises(SystemExit):
+            cmd_cancel(t, "nope")
+
+    def test_cancel_sets_status(self):
+        t = _fresh_treasury()
+        t.upsert_job("cancel_me", "in_progress")
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_cancel(t, "cancel_me")
+        job = t.get_job("cancel_me")
+        self.assertEqual(job["status"], "cancelled")
+        self.assertIn("cancelled", buf.getvalue())
+
+    def test_cancel_already_done_is_noop(self):
+        t = _fresh_treasury()
+        t.upsert_job("done_job", "completed")
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cmd_cancel(t, "done_job")
+        job = t.get_job("done_job")
+        self.assertEqual(job["status"], "completed")
+        self.assertIn("already", buf.getvalue())
+
+
+class TestJobsCLI(unittest.TestCase):
+
+    def test_main_list_no_args(self):
+        from solvent.job_cmd import main
+        buf = io.StringIO()
+        with mock.patch("solvent.job_cmd.cmd_list") as mock_list, \
+             mock.patch.object(sys, "argv", ["solvent-jobs"]):
+            # cmd_list is called; it should receive a Treasury and no filter
+            mock_list.side_effect = lambda t, **kw: print("No jobs", file=sys.stdout)
+            with redirect_stdout(buf):
+                main()
+        mock_list.assert_called_once()
+
+    def test_main_list_subcommand(self):
+        from solvent.job_cmd import main
+        buf = io.StringIO()
+        with mock.patch("solvent.job_cmd.cmd_list") as mock_list, \
+             mock.patch.object(sys, "argv", ["solvent-jobs", "list"]):
+            mock_list.side_effect = lambda t, **kw: print("No jobs", file=sys.stdout)
+            with redirect_stdout(buf):
+                main()
+        mock_list.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()