diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 101aa38..71f8646 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -99,10 +99,29 @@ jobs:
           MAJOR=$(echo "$VERSION" | cut -d. -f1)
           MINOR=$(echo "$VERSION" | cut -d. -f2)
           PATCH=$(echo "$VERSION" | cut -d. -f3)
-          NEW_PATCH=$((PATCH + 1))
-          NEW_TAG="v${MAJOR}.${MINOR}.${NEW_PATCH}"
+
+          # Choose the bump from conventional-commit markers in the commits since
+          # the last tag: a "BREAKING CHANGE" footer or a "type!:" subject -> major;
+          # a "feat:" -> minor; otherwise patch. Still fully automated — no manual tags.
+          if [ "$LATEST" = "v0.0.0" ]; then
+            RANGE="HEAD"
+          else
+            RANGE="${LATEST}..HEAD"
+          fi
+          LOG=$(git log --format='%s%n%b' "$RANGE")
+          if echo "$LOG" | grep -qE '^BREAKING CHANGE:|^[a-z]+(\([^)]*\))?!:'; then
+            MAJOR=$((MAJOR + 1)); MINOR=0; PATCH=0
+            LEVEL="major"
+          elif echo "$LOG" | grep -qE '^feat(\([^)]*\))?:'; then
+            MINOR=$((MINOR + 1)); PATCH=0
+            LEVEL="minor"
+          else
+            PATCH=$((PATCH + 1))
+            LEVEL="patch"
+          fi
+          NEW_TAG="v${MAJOR}.${MINOR}.${PATCH}"
           echo "new_tag=$NEW_TAG" >> "$GITHUB_OUTPUT"
-          echo "Bumping $LATEST -> $NEW_TAG"
+          echo "Bumping $LATEST -> $NEW_TAG ($LEVEL)"
 
       - name: Create and push tag
         if: steps.version.outputs.new_tag
diff --git a/CLAUDE.md b/CLAUDE.md
index a375836..8c09316 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -136,7 +136,7 @@ Two separate Docker images:
 
 ### Authentication & Credential Flow
 
-- A single shared **API key** authenticates all workers to the UI (`CASHPILOT_API_KEY` env var).
+- **Per-worker fleet keys** (v1.0.0+): `CASHPILOT_API_KEY` is the shared **enrollment/bootstrap** key. On a worker's first heartbeat the UI issues it a unique key (stored encrypted on the UI, returned once, persisted on the worker under `/data/.worker_key`); thereafter each worker authenticates — in both directions — with its own key, and the shared key is rejected for an enrolled worker.
 - Credentials are Docker-native: UI sends full container spec (image, env vars, volumes, ports) to worker on deploy. Worker passes to Docker API. Docker stores env vars in container config. Restarts preserve env vars natively.
 
 ### Worker Environment Variables
@@ -172,7 +172,7 @@ Two separate Docker images:
 
 ### `release.yml` — Auto Release
 
-Triggers on push to `main` (paths: `app/`, `services/`, `Dockerfile*`, `requirements*.txt`). Auto-increments patch version, creates annotated git tag + GitHub Release. Skips if commit message contains `[skip ci]`.
+Triggers on push to `main` (paths: `app/`, `services/`, `Dockerfile*`, `requirements*.txt`). Chooses the version bump from conventional-commit markers since the last tag — **major** on a `BREAKING CHANGE` footer or `type!:` subject, **minor** on `feat:`, else **patch** — then creates an annotated git tag + GitHub Release. Skips if commit message contains `[skip ci]`.
 
 **CRITICAL: NEVER manually create tags or GitHub releases.** The workflow handles version bumping automatically. Manual tags cause `already_exists` conflicts and skip Docker builds.
 
diff --git a/README.md b/README.md
index 44fe17e..0158922 100644
--- a/README.md
+++ b/README.md
@@ -157,7 +157,7 @@ cashpilot/
 |----------|---------|-------------|
 | `TZ` | `UTC` | Timezone for scheduling and display |
 | `CASHPILOT_SECRET_KEY` | *(auto-generated)* | Encryption key for stored credentials |
-| `CASHPILOT_API_KEY` | -- | Shared secret between UI and workers for API authentication |
+| `CASHPILOT_API_KEY` | -- | Enrollment/bootstrap key; each worker then gets its own key (per-worker fleet keys, v1.0.0+) |
 | `CASHPILOT_COLLECTION_INTERVAL` | `3600` | Seconds between earnings collection cycles |
 | `CASHPILOT_PORT` | `8080` | Web UI port inside the container |
 | `CASHPILOT_METRICS_ENABLED` | `false` | Set to `true` to expose Prometheus metrics at `/metrics` |
diff --git a/app/database.py b/app/database.py
index efa96b0..ccefe2e 100644
--- a/app/database.py
+++ b/app/database.py
@@ -140,6 +140,8 @@ def decrypt_value(value: str) -> str:
     apps            TEXT    NOT NULL DEFAULT '[]',
     system_info     TEXT    NOT NULL DEFAULT '{}',
     last_heartbeat  TEXT,
+    api_key_enc     TEXT,
+    key_confirmed   INTEGER NOT NULL DEFAULT 0,
     registered_at   TEXT    NOT NULL DEFAULT (datetime('now'))
 );
 
@@ -323,6 +325,14 @@ async def init_db() -> None:
         elif "apps" not in cols:
             await db.execute("ALTER TABLE workers ADD COLUMN apps TEXT NOT NULL DEFAULT '[]'")
 
+        # Migrate workers table: add api_key_enc for per-worker fleet keys.
+        # (cols is the pre-rebuild snapshot; on a fresh DB the column comes from
+        # _SCHEMA so it is already present here and the ALTER is skipped.)
+        if "api_key_enc" not in cols:
+            await db.execute("ALTER TABLE workers ADD COLUMN api_key_enc TEXT")
+        if "key_confirmed" not in cols:
+            await db.execute("ALTER TABLE workers ADD COLUMN key_confirmed INTEGER NOT NULL DEFAULT 0")
+
         # Migrate users table: add password_changed_at for session invalidation
         cursor = await db.execute("PRAGMA table_info(users)")
         user_cols = {row["name"] for row in await cursor.fetchall()}
@@ -991,6 +1001,72 @@ async def delete_worker(worker_id: int) -> None:
         await db.close()
 
 
+# --- Per-worker fleet keys ---
+#
+# The UI must both VERIFY inbound heartbeats from a worker and, for the full
+# cutover, AUTHENTICATE outbound calls TO that worker — so it needs the key
+# itself, not just a one-way hash. Keys are therefore stored encrypted at rest
+# (Fernet, the same at-rest protection as service credentials) and decrypted on
+# demand for comparison and for outbound Authorization headers.
+
+
+async def set_worker_key(client_id: str, key: str) -> None:
+    """Store a worker's per-worker key (encrypted), unconfirmed until the worker
+    proves it holds the key by using it on a later heartbeat."""
+    db = await _get_db()
+    try:
+        cursor = await db.execute(
+            "UPDATE workers SET api_key_enc = ?, key_confirmed = 0 WHERE client_id = ?",
+            (encrypt_value(key), client_id),
+        )
+        await db.commit()
+        if not cursor.rowcount:
+            # The worker row must exist first (upsert runs before this); a missing
+            # row would silently drop the key and lock the worker out.
+            _logger.warning("set_worker_key: no worker row for client_id=%s", client_id)
+    finally:
+        await db.close()
+
+
+async def confirm_worker_key(client_id: str) -> None:
+    """Mark a worker's key confirmed — it has authenticated with its own key, so the
+    shared bootstrap key is refused from now on (the cutover finalizes)."""
+    db = await _get_db()
+    try:
+        await db.execute(
+            "UPDATE workers SET key_confirmed = 1 WHERE client_id = ?",
+            (client_id,),
+        )
+        await db.commit()
+    finally:
+        await db.close()
+
+
+async def get_worker_key(client_id: str) -> str | None:
+    """Return a worker's per-worker key (decrypted), or None if not yet enrolled."""
+    key, _ = await get_worker_key_state(client_id)
+    return key
+
+
+async def get_worker_key_state(client_id: str) -> tuple[str | None, bool]:
+    """Return (key, confirmed) for a worker: the decrypted per-worker key (or None
+    if unenrolled) and whether the worker has confirmed it by using it."""
+    db = await _get_db()
+    try:
+        cursor = await db.execute(
+            "SELECT api_key_enc, key_confirmed FROM workers WHERE client_id = ?",
+            (client_id,),
+        )
+        row = await cursor.fetchone()
+        if not row:
+            return None, False
+        enc = row["api_key_enc"]
+        key = decrypt_value(enc) if enc else None
+        return key, bool(row["key_confirmed"])
+    finally:
+        await db.close()
+
+
 # --- Health Events ---
 
 
diff --git a/app/main.py b/app/main.py
index 57dd646..30a7f46 100644
--- a/app/main.py
+++ b/app/main.py
@@ -14,6 +14,7 @@
 import logging
 import os
 import re
+import secrets
 import socket
 from collections import defaultdict
 from contextlib import asynccontextmanager
@@ -1006,9 +1007,15 @@ async def _get_verified_worker_url(worker: dict[str, Any]) -> tuple[str, dict[st
     if not worker["url"]:
         raise HTTPException(status_code=503, detail="Worker URL not known")
     url = await asyncio.to_thread(_validate_worker_url, worker["url"])
+    # Authenticate to the worker with ITS OWN key once enrolled; fall back to the
+    # shared bootstrap key only for workers that have not enrolled yet. Post-cutover
+    # an enrolled worker rejects the shared key, so the UI must present its own.
+    cid = worker.get("client_id") or ""
+    worker_key = await database.get_worker_key(cid) if cid else None
+    auth_key = worker_key or FLEET_API_KEY
     headers: dict[str, str] = {}
-    if FLEET_API_KEY:
-        headers["Authorization"] = f"Bearer {FLEET_API_KEY}"
+    if auth_key:
+        headers["Authorization"] = f"Bearer {auth_key}"
     return url, headers
 
 
@@ -1689,16 +1696,43 @@ async def api_admin_set_password(request: Request, user_id: int, body: AdminPass
 # ---------------------------------------------------------------------------
 
 
-def _verify_fleet_api_key(request: Request) -> None:
-    """Verify the shared fleet API key from a worker's request."""
+def _bearer_token(request: Request) -> str:
+    """Extract the bearer token from an Authorization header (empty if absent)."""
+    h = request.headers.get("Authorization", "")
+    return h[7:] if h.startswith("Bearer ") else ""
+
+
+async def _authenticate_worker_heartbeat(request: Request, cid: str) -> str:
+    """Authenticate a heartbeat and classify it. Returns one of:
+
+    - ``"enroll"``  — worker has no key yet and presented the shared key: mint one.
+    - ``"reissue"`` — worker has a key that is NOT yet confirmed and presented the
+      shared key: it likely lost the enrollment response, so re-deliver the SAME
+      key (until confirmed, the shared key still works for this one worker — a
+      bounded window that closes on the worker's first own-key heartbeat).
+    - ``"ok"``      — worker presented its own key: authenticated; confirm it so the
+      shared key is refused from now on (the cutover finalizes).
+
+    Raises 401 otherwise — notably, a confirmed worker presenting the shared key is
+    rejected, which is what stops shared-key holders from impersonating a worker.
+    """
     if not FLEET_API_KEY:
         raise HTTPException(
             status_code=503,
             detail="Fleet key not configured — set CASHPILOT_API_KEY or mount shared /fleet volume",
         )
-    auth_header = request.headers.get("Authorization", "")
-    if not hmac.compare_digest(auth_header.encode(), f"Bearer {FLEET_API_KEY}".encode()):
+    token = _bearer_token(request)
+    key, confirmed = await database.get_worker_key_state(cid) if cid else (None, False)
+    shared_ok = bool(token) and hmac.compare_digest(token.encode(), FLEET_API_KEY.encode())
+    if key is None:
+        if shared_ok:
+            return "enroll"
         raise HTTPException(status_code=401, detail="Invalid API key")
+    if token and hmac.compare_digest(token.encode(), key.encode()):
+        return "ok"
+    if not confirmed and shared_ok:
+        return "reissue"
+    raise HTTPException(status_code=401, detail="Invalid or missing per-worker key")
 
 
 class WorkerHeartbeat(BaseModel):
@@ -1713,9 +1747,11 @@ class WorkerHeartbeat(BaseModel):
 @app.post("/api/workers/heartbeat")
 async def api_worker_heartbeat(request: Request, body: WorkerHeartbeat) -> dict[str, Any]:
     """Receive a heartbeat from a worker. Registers or updates the worker."""
-    _verify_fleet_api_key(request)
     # Use client_id for identity; fall back to name for backward compat
     cid = body.client_id or body.name
+    if not cid:
+        raise HTTPException(status_code=400, detail="Worker name or client_id required")
+    state = await _authenticate_worker_heartbeat(request, cid)
     worker_id = await database.upsert_worker(
         client_id=cid,
         name=body.name,
@@ -1725,7 +1761,27 @@ async def api_worker_heartbeat(request: Request, body: WorkerHeartbeat) -> dict[
         system_info=json.dumps(body.system_info),
     )
     metrics.record_heartbeat(body.name)
-    return {"status": "ok", "worker_id": worker_id}
+    resp: dict[str, Any] = {"status": "ok", "worker_id": worker_id}
+    if state == "enroll":
+        # First contact: mint this worker's own key and hand it back once. Stored
+        # unconfirmed until the worker proves receipt by using it (see "ok").
+        new_key = secrets.token_urlsafe(32)
+        await database.set_worker_key(cid, new_key)
+        resp["worker_key"] = new_key
+        logger.info("Worker '%s' enrolled — per-worker key issued (awaiting confirmation)", cid)
+    elif state == "reissue":
+        # The worker still holds the shared key, so it never received/persisted its
+        # own key — re-deliver the same one so a dropped enrollment response can't
+        # lock it out.
+        existing = await database.get_worker_key(cid)
+        if existing:
+            resp["worker_key"] = existing
+        logger.info("Worker '%s' re-issued its per-worker key (not yet confirmed)", cid)
+    elif state == "ok":
+        # The worker authenticated with its own key: finalize the cutover so the
+        # shared key is refused from now on.
+        await database.confirm_worker_key(cid)
+    return resp
 
 
 @app.get("/api/workers")
diff --git a/app/worker_api.py b/app/worker_api.py
index 5979c6d..5e81842 100644
--- a/app/worker_api.py
+++ b/app/worker_api.py
@@ -23,6 +23,7 @@
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime
 from html import escape as _esc
+from pathlib import Path
 from typing import Any
 
 import httpx
@@ -56,6 +57,40 @@
 _last_heartbeat: str = "never"
 _last_error: str = ""
 
+# Per-worker fleet key. On first contact the UI enrolls this worker and hands back
+# a key unique to us, which we persist here (in our own private /data, never the
+# shared /fleet volume) and use for all subsequent auth in both directions. Until
+# enrollment we authenticate with the shared bootstrap key.
+_WORKER_KEY_FILE = Path(os.getenv("CASHPILOT_DATA_DIR", "/data")) / ".worker_key"
+
+
+def _load_worker_key() -> str | None:
+    try:
+        if _WORKER_KEY_FILE.is_file():
+            return _WORKER_KEY_FILE.read_text().strip() or None
+    except OSError as exc:
+        logger.warning("Could not read per-worker key: %s", exc)
+    return None
+
+
+def _save_worker_key(key: str) -> None:
+    global _worker_key
+    _worker_key = key
+    try:
+        _WORKER_KEY_FILE.parent.mkdir(parents=True, exist_ok=True)
+        _WORKER_KEY_FILE.write_text(key)
+        _WORKER_KEY_FILE.chmod(0o600)
+    except OSError as exc:
+        logger.warning("Could not persist per-worker key: %s", exc)
+
+
+_worker_key: str | None = _load_worker_key()
+
+
+def _active_key() -> str:
+    """The key we authenticate with: our own once enrolled, else the shared key."""
+    return _worker_key or API_KEY
+
 
 # ---------------------------------------------------------------------------
 # Auth
@@ -63,11 +98,17 @@
 
 
 def _verify_api_key(request: Request) -> None:
-    """Verify the shared API key from Authorization header."""
-    if not API_KEY:
+    """Verify an inbound UI->worker call.
+
+    Once enrolled we require OUR OWN per-worker key; the shared bootstrap key is
+    rejected (the cutover). Before enrollment we accept the shared key so the UI
+    can reach us to enroll in the first place.
+    """
+    expected = _active_key()
+    if not expected:
         raise HTTPException(status_code=503, detail="Fleet key not configured")
     auth = request.headers.get("Authorization", "")
-    if not hmac.compare_digest(auth.encode(), f"Bearer {API_KEY}".encode()):
+    if not hmac.compare_digest(auth.encode(), f"Bearer {expected}".encode()):
         raise HTTPException(status_code=401, detail="Invalid API key")
 
 
@@ -103,9 +144,16 @@ async def _send_heartbeat() -> None:
             resp = await client.post(
                 f"{UI_URL.rstrip('/')}/api/workers/heartbeat",
                 json=payload,
-                headers={"Authorization": f"Bearer {API_KEY}"},
+                headers={"Authorization": f"Bearer {_active_key()}"},
             )
             resp.raise_for_status()
+            # Enrollment: the UI returns our own per-worker key exactly once.
+            issued = None
+            with contextlib.suppress(Exception):
+                issued = resp.json().get("worker_key")
+            if issued and issued != _worker_key:
+                _save_worker_key(issued)
+                logger.info("Enrolled: received and persisted this worker's own fleet key")
             _ui_connected = True
             _last_heartbeat = datetime.now(UTC).strftime("%H:%M:%S UTC")
             _last_error = ""
diff --git a/docs/AUTOPILOT-WORKLOG.md b/docs/AUTOPILOT-WORKLOG.md
new file mode 100644
index 0000000..592c68c
--- /dev/null
+++ b/docs/AUTOPILOT-WORKLOG.md
@@ -0,0 +1,35 @@
+# Autopilot Worklog — CashPilot per-worker-keys cutover
+
+Append-only. Newest at the bottom. Every "done" needs evidence (test/CI/commit).
+
+---
+
+### 2026-07-10 — kickoff
+- Branch `feat/per-worker-keys` created off `main` (after merging #90/#97/#98/#100).
+- Goal recorded in `docs/GOAL.md`: per-worker keys **full cutover** → `v1.0.0`, announced.
+- Established the plan (worker + UI + DB migration + release-workflow major bump + docs).
+- Next: DB layer first — `workers.api_key_hash` column + migration + helpers + tests.
+
+### 2026-07-10 — US-001 DONE (DB layer)
+- `workers.api_key_hash` column + idempotent migration; `hash_worker_key` (sha256),
+  `set/get_worker_key_hash`. Commit c22a5a3.
+- Evidence: 4 new TestWorkerKeys tests pass; full pytest **1085 passed**; ruff check+format clean.
+- Next: US-002 — UI heartbeat enrollment + reject shared key for enrolled workers.
+
+### 2026-07-10 — US-002/003 DONE (UI cutover, both directions)
+- US-002: heartbeat enrollment + reject shared key for enrolled workers (commit 088be34).
+- Storage upgraded hash->encrypted (api_key_enc) so the UI can also authenticate
+  outbound (commit 7328acf).
+- US-003: `_get_verified_worker_url` sends each worker's own key outbound (commit 7328acf+).
+- Evidence: full pytest **1088 passed**; ruff check+format clean.
+- Next: US-004 — worker side (obtain/persist/use its key; inbound verify accepts own key).
+
+### 2026-07-10 — US-003..007 DONE + architect review + PR opened
+- US-003 outbound (a3827ea), US-004 worker side (df3866f), US-005 release major-bump (0065304),
+  US-006 docs (f250799).
+- Architect (thorough) PASSED all 5 security criteria; fixed its MEDIUM availability
+  defect (lost-enrollment lockout) via key_confirmed re-delivery + low-sev items (becb5a1).
+- Deslop pass: no slop found (all helpers used, no dead code, purposeful comments).
+- US-007: **PR #101 opened** (feat(fleet)!: → v1.0.0), NOT merged (awaits Sergio).
+- Evidence: full pytest **1099 passed**; ruff check+format clean; branch pushed.
+- All 7 PRD stories passes:true. Waiting on #101 CI + CodeRabbit before cancel.
diff --git a/docs/GOAL.md b/docs/GOAL.md
new file mode 100644
index 0000000..0bbf349
--- /dev/null
+++ b/docs/GOAL.md
@@ -0,0 +1,30 @@
+# GOAL
+
+Started: 2026-07-10
+
+## Directive (verbatim)
+
+yes continue — use ralph
+
+## Established intent (from the preceding conversation)
+
+Implement **per-worker fleet keys as a full cutover** for CashPilot (web), shipped as a
+**new major version (`v1.0.0`)** and **announced well** (existing fleets must re-enroll).
+
+Cutover design (Sergio's decision):
+- **Worker**: obtain a per-worker key on enrollment, persist it (`/data`), use it for
+  heartbeats; verify UI→worker calls against its own key.
+- **UI**: issue + hash a per-worker key on first heartbeat; bind it to `client_id`;
+  **reject the shared key for already-enrolled workers** (real impersonation protection);
+  call each worker with *its* key.
+- **Migration + version**: `workers.api_key_hash` schema column; teach the auto-release
+  workflow to do a **major** bump on a `BREAKING CHANGE`/`feat!` marker (no manual tags);
+  upgrade/announcement docs.
+
+Keep the shared `CASHPILOT_API_KEY` as the **enrollment/bootstrap** credential only.
+
+## Working rules (this repo)
+- PR workflow, one PR per feature; never merge without Sergio's approval.
+- CI runs BOTH `ruff check` and `ruff format --check` — run `ruff format` before every commit.
+- Never manually create tags — the release workflow bumps + tags on push to main.
+- Full `pytest` must stay green.
diff --git a/docs/fleet.md b/docs/fleet.md
index c41c37d..467ab74 100644
--- a/docs/fleet.md
+++ b/docs/fleet.md
@@ -73,21 +73,20 @@ volumes:
 ```
 
 !!! important "API Key"
-    The `CASHPILOT_API_KEY` must be identical on the UI and all workers. This is the shared secret that authenticates worker-to-UI communication.
+    The `CASHPILOT_API_KEY` must be identical on the UI and all workers. It is the **enrollment key** each worker uses on first contact; after that, each worker uses its own automatically-issued key.
 
 ## Authentication
 
-A single shared API key authenticates all fleet communication:
+CashPilot uses **per-worker fleet keys** (since v1.0.0). The shared `CASHPILOT_API_KEY` is only a bootstrap/enrollment credential; each worker then gets its own key.
 
-- Set `CASHPILOT_API_KEY` on both the UI and all workers.
-- Workers include this key as a Bearer token in heartbeat requests.
-- The UI includes this key when sending commands to workers.
-- If not set explicitly, the UI and co-located worker auto-generate a shared key via the `/fleet` volume.
+- Set `CASHPILOT_API_KEY` on the UI and all workers (or let the UI + co-located worker auto-generate one via the `/fleet` volume).
+- **Enrollment:** a worker's first heartbeat authenticates with the shared key. The UI issues that worker its own unique key (stored encrypted on the UI, and returned once). The worker persists it under its private `/data`.
+- **After enrollment:** the worker authenticates every heartbeat with its own key, and the UI calls that worker with the same key. The shared key **no longer works** for an enrolled worker — so a leaked worker key only affects that one worker, and no worker can impersonate another.
 
 The fleet key is **never sent to the browser on page load**. The fleet dashboard reveals it only on an explicit, owner-only action (the **Reveal API Key** button), and copy-to-clipboard fetches it the same way.
 
 !!! warning "Security"
-    The fleet key grants access to container management operations. Treat it as a sensitive credential. Do not expose worker APIs (port 8081) to the public internet.
+    Keys grant container-management access — treat them as sensitive credentials, and never expose worker APIs (port 8081) to the public internet. See the [v1.0.0 upgrade guide](upgrade-v1.md) if you are moving an existing fleet.
 
 ## Fleet Dashboard
 
diff --git a/docs/upgrade-v1.md b/docs/upgrade-v1.md
new file mode 100644
index 0000000..083d21e
--- /dev/null
+++ b/docs/upgrade-v1.md
@@ -0,0 +1,70 @@
+# Upgrading to v1.0.0 — Per-worker fleet keys
+
+v1.0.0 hardens fleet authentication. Instead of one shared key doing everything,
+**each worker now gets its own key**. This is a breaking change for existing
+fleets: worker and UI images must both be on v1.0.0+, and workers re-enroll
+automatically on their first heartbeat after the upgrade.
+
+## What changed
+
+| | Before (0.x) | v1.0.0 |
+|---|---|---|
+| Worker → UI heartbeat | shared `CASHPILOT_API_KEY` | worker's **own** key (after enrollment) |
+| UI → worker commands | shared key | that worker's **own** key |
+| Role of `CASHPILOT_API_KEY` | authenticates everything | **enrollment/bootstrap only** |
+
+**Why:** with per-worker keys, a key that leaks from one worker only affects that
+one worker, and no worker can present another worker's identity to the UI. The
+shared key stops being a fleet-wide credential once a worker is enrolled.
+
+## How enrollment works (automatic)
+
+1. A worker's **first** heartbeat authenticates with the shared `CASHPILOT_API_KEY`.
+2. The UI issues that worker a unique key — stored **encrypted** on the UI and
+   returned to the worker **once**.
+3. The worker persists the key under its own private `/data/.worker_key` and uses
+   it from then on. The UI addresses that worker with the same key.
+4. Once enrolled, the shared key **no longer works** for that worker.
+
+You don't handle keys by hand — this all happens on the next heartbeat.
+
+## Upgrade steps
+
+1. **Upgrade the UI** image to `drumsergio/cashpilot:1.0.0` (or newer).
+2. **Upgrade every worker** image to `drumsergio/cashpilot-worker:1.0.0` (or newer).
+   Do not leave old-version workers running against a v1.0.0 UI — once the UI has
+   enrolled a worker, an old worker image (which only knows the shared key) can no
+   longer heartbeat.
+3. Keep `CASHPILOT_API_KEY` **unchanged** — it is still needed for enrollment.
+4. Restart the containers. Each worker auto-enrolls on its first heartbeat; confirm
+   every worker shows **online** in the fleet dashboard.
+
+Persist the worker's `/data` volume (the compose files already do) so its key
+survives restarts.
+
+## Recovery & rollback
+
+- **A worker's `/data` was wiped** (lost its key): it will try the shared key, which
+  the UI now rejects for an enrolled worker. Re-enroll it by removing the worker in
+  the fleet dashboard — it re-registers and enrolls fresh on its next heartbeat.
+- **Rolling a worker back to a 0.x image:** first remove that worker in the dashboard
+  (clears its enrollment) so the shared key is accepted again, then redeploy the old
+  image.
+
+---
+
+## Release notes — v1.0.0
+
+**Per-worker fleet keys.** Every worker now authenticates with its own automatically
+issued key instead of a single shared secret. The shared `CASHPILOT_API_KEY` becomes
+an enrollment-only bootstrap credential: a worker uses it once, receives its own key,
+and uses that thereafter — in both directions. A leaked worker key is now scoped to a
+single worker, and workers can no longer impersonate one another.
+
+**Breaking:** existing fleets must upgrade both UI and worker images to v1.0.0 and let
+workers re-enroll (automatic on first heartbeat). See the upgrade guide above.
+
+!!! note "The shared key is still sensitive"
+    `CASHPILOT_API_KEY` is now an *enrollment* credential, not a fleet-wide command
+    key — but it remains high-value: anyone holding it can enroll a new worker that
+    then receives deploy specs (which carry service credentials). Keep protecting it.
diff --git a/mkdocs.yml b/mkdocs.yml
index b595724..6fa5cdd 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -75,6 +75,7 @@ nav:
   - Getting Started: getting-started.md
   - Architecture: architecture.md
   - Fleet Management: fleet.md
+  - Upgrade to v1.0.0: upgrade-v1.md
   - Service Guides:
     - guides/README.md
     - Bandwidth Sharing:
diff --git a/tests/test_database.py b/tests/test_database.py
index 230a496..8411e26 100644
--- a/tests/test_database.py
+++ b/tests/test_database.py
@@ -416,6 +416,73 @@ async def run():
         asyncio.run(run())
 
 
+class TestWorkerKeys:
+    def test_workers_table_has_api_key_enc_column(self, db):
+        async def run():
+            conn = await database._get_db()
+            try:
+                cur = await conn.execute("PRAGMA table_info(workers)")
+                cols = {row["name"] for row in await cur.fetchall()}
+                assert "api_key_enc" in cols
+            finally:
+                await conn.close()
+
+        asyncio.run(run())
+
+    def test_set_and_get_worker_key_round_trip(self, db):
+        async def run():
+            await database.upsert_worker("c1", "w1")
+            assert await database.get_worker_key("c1") is None  # unenrolled
+            await database.set_worker_key("c1", "super-secret-key")
+            assert await database.get_worker_key("c1") == "super-secret-key"
+
+        asyncio.run(run())
+
+    def test_worker_key_stored_encrypted_at_rest(self, db):
+        async def run():
+            await database.upsert_worker("c1", "w1")
+            await database.set_worker_key("c1", "super-secret-key")
+            conn = await database._get_db()
+            try:
+                cur = await conn.execute("SELECT api_key_enc FROM workers WHERE client_id = 'c1'")
+                row = await cur.fetchone()
+                assert row["api_key_enc"].startswith("enc:")  # not stored in plaintext
+                assert "super-secret-key" not in row["api_key_enc"]
+            finally:
+                await conn.close()
+
+        asyncio.run(run())
+
+    def test_get_worker_key_missing_worker_is_none(self, db):
+        async def run():
+            assert await database.get_worker_key("nope") is None
+
+        asyncio.run(run())
+
+    def test_key_confirmation_lifecycle(self, db):
+        async def run():
+            await database.upsert_worker("c1", "w1")
+            # No key -> unenrolled + unconfirmed.
+            assert await database.get_worker_key_state("c1") == (None, False)
+            # Setting a key leaves it unconfirmed.
+            await database.set_worker_key("c1", "k1")
+            assert await database.get_worker_key_state("c1") == ("k1", False)
+            # Confirming flips the flag.
+            await database.confirm_worker_key("c1")
+            assert await database.get_worker_key_state("c1") == ("k1", True)
+            # Re-issuing a key resets confirmation.
+            await database.set_worker_key("c1", "k2")
+            assert await database.get_worker_key_state("c1") == ("k2", False)
+
+        asyncio.run(run())
+
+    def test_get_worker_key_state_missing_worker(self, db):
+        async def run():
+            assert await database.get_worker_key_state("nope") == (None, False)
+
+        asyncio.run(run())
+
+
 class TestDataRetention:
     def test_purge_returns_count(self, db):
         async def run():
diff --git a/tests/test_main_routes.py b/tests/test_main_routes.py
index f69459a..7d6cc62 100644
--- a/tests/test_main_routes.py
+++ b/tests/test_main_routes.py
@@ -1383,9 +1383,12 @@ def test_env_info_secret_key_never_leaks(self, client):
 
 
 class TestApiFleet:
-    def test_api_worker_heartbeat(self, client):
+    def test_api_worker_heartbeat_enrollment_issues_key(self, client):
+        # A brand-new worker authenticates with the shared key and is issued its own.
         with (
             patch("app.main.FLEET_API_KEY", "test-fleet-key"),
+            patch("app.main.database.get_worker_key_state", new_callable=AsyncMock, return_value=(None, False)),
+            patch("app.main.database.set_worker_key", new_callable=AsyncMock) as set_key,
             patch("app.main.database.upsert_worker", new_callable=AsyncMock, return_value=1),
         ):
             resp = client.post(
@@ -1394,10 +1397,38 @@ def test_api_worker_heartbeat(self, client):
                 headers={"Authorization": "Bearer test-fleet-key"},
             )
             assert resp.status_code == 200
-            assert resp.json()["worker_id"] == 1
+            body = resp.json()
+            assert body["worker_id"] == 1
+            assert body.get("worker_key")  # a per-worker key was issued
+            set_key.assert_awaited_once()  # it was persisted (encrypted)
 
-    def test_api_worker_heartbeat_bad_key(self, client):
-        with patch("app.main.FLEET_API_KEY", "test-fleet-key"):
+    def test_api_worker_heartbeat_reissues_until_confirmed(self, client):
+        # Enrolled but unconfirmed (worker missed the enrollment response): the
+        # shared key is still accepted and the SAME key is re-delivered, so a
+        # dropped response can't permanently lock the worker out.
+        with (
+            patch("app.main.FLEET_API_KEY", "test-fleet-key"),
+            patch(
+                "app.main.database.get_worker_key_state",
+                new_callable=AsyncMock,
+                return_value=("existing-key", False),
+            ),
+            patch("app.main.database.get_worker_key", new_callable=AsyncMock, return_value="existing-key"),
+            patch("app.main.database.upsert_worker", new_callable=AsyncMock, return_value=1),
+        ):
+            resp = client.post(
+                "/api/workers/heartbeat",
+                json={"name": "w", "client_id": "c1"},
+                headers={"Authorization": "Bearer test-fleet-key"},
+            )
+            assert resp.status_code == 200
+            assert resp.json().get("worker_key") == "existing-key"  # re-delivered
+
+    def test_api_worker_heartbeat_unenrolled_bad_key(self, client):
+        with (
+            patch("app.main.FLEET_API_KEY", "test-fleet-key"),
+            patch("app.main.database.get_worker_key_state", new_callable=AsyncMock, return_value=(None, False)),
+        ):
             resp = client.post(
                 "/api/workers/heartbeat",
                 json={"name": "worker-1"},
@@ -1413,6 +1444,35 @@ def test_api_worker_heartbeat_no_fleet_key(self, client):
             )
             assert resp.status_code == 503
 
+    def test_api_worker_heartbeat_confirmed_rejects_shared_key(self, client):
+        # The finalized cutover: a CONFIRMED worker must use its own key; the shared
+        # key is rejected, its own key is accepted, and no new key is re-issued.
+        own_key = "the-worker-key"
+        with (
+            patch("app.main.FLEET_API_KEY", "test-fleet-key"),
+            patch(
+                "app.main.database.get_worker_key_state",
+                new_callable=AsyncMock,
+                return_value=(own_key, True),
+            ),
+            patch("app.main.database.confirm_worker_key", new_callable=AsyncMock),
+            patch("app.main.database.upsert_worker", new_callable=AsyncMock, return_value=1),
+        ):
+            shared = client.post(
+                "/api/workers/heartbeat",
+                json={"name": "w", "client_id": "c1"},
+                headers={"Authorization": "Bearer test-fleet-key"},
+            )
+            assert shared.status_code == 401  # shared key no longer works once confirmed
+
+            own = client.post(
+                "/api/workers/heartbeat",
+                json={"name": "w", "client_id": "c1"},
+                headers={"Authorization": f"Bearer {own_key}"},
+            )
+            assert own.status_code == 200
+            assert "worker_key" not in own.json()
+
     def test_api_list_workers(self, client):
         workers = [{"id": 1, "name": "w1", "status": "online", "containers": "[]", "apps": "[]", "system_info": "{}"}]
         with (
@@ -1948,6 +2008,35 @@ def test_run_vacuum_swallows_error(self):
             asyncio.run(main_mod._run_vacuum())
 
 
+class TestOutboundWorkerAuth:
+    """US-003: UI->worker calls use the worker's own key once enrolled."""
+
+    def _worker(self):
+        return {"status": "online", "url": "http://192.168.1.5:8081", "client_id": "c1"}
+
+    def test_uses_per_worker_key_when_enrolled(self):
+        import app.main as m
+
+        with (
+            patch("app.main.FLEET_API_KEY", "shared-key"),
+            patch("app.main.database.get_worker_key", new_callable=AsyncMock, return_value="worker-1-key"),
+            patch("app.main._validate_worker_url", return_value="http://192.168.1.5:8081"),
+        ):
+            _, headers = asyncio.run(m._get_verified_worker_url(self._worker()))
+        assert headers["Authorization"] == "Bearer worker-1-key"
+
+    def test_falls_back_to_shared_key_when_unenrolled(self):
+        import app.main as m
+
+        with (
+            patch("app.main.FLEET_API_KEY", "shared-key"),
+            patch("app.main.database.get_worker_key", new_callable=AsyncMock, return_value=None),
+            patch("app.main._validate_worker_url", return_value="http://192.168.1.5:8081"),
+        ):
+            _, headers = asyncio.run(m._get_verified_worker_url(self._worker()))
+        assert headers["Authorization"] == "Bearer shared-key"
+
+
 # ---------------------------------------------------------------------------
 # Shared auth deps (app/deps.py) — guard branches
 # ---------------------------------------------------------------------------
diff --git a/tests/test_worker_keys.py b/tests/test_worker_keys.py
new file mode 100644
index 0000000..b1ee495
--- /dev/null
+++ b/tests/test_worker_keys.py
@@ -0,0 +1,92 @@
+"""Worker-side per-worker fleet key: persistence, auth selection, enrollment."""
+
+import asyncio
+import os
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("CASHPILOT_API_KEY", "test-fleet-key")
+
+import pytest  # noqa: E402
+
+try:
+    from fastapi import HTTPException  # noqa: E402
+
+    import app.worker_api as w  # noqa: E402
+except ImportError:
+    pytest.skip(
+        "Requires full app dependencies (fastapi, docker, etc.) — runs in CI",
+        allow_module_level=True,
+    )
+
+
+class TestActiveKey:
+    def test_prefers_own_key_when_enrolled(self):
+        with patch.object(w, "_worker_key", "own"), patch.object(w, "API_KEY", "shared"):
+            assert w._active_key() == "own"
+
+    def test_falls_back_to_shared_before_enrollment(self):
+        with patch.object(w, "_worker_key", None), patch.object(w, "API_KEY", "shared"):
+            assert w._active_key() == "shared"
+
+
+class TestKeyPersistence:
+    def test_save_and_load_round_trip(self, tmp_path):
+        f = tmp_path / ".worker_key"
+        with patch.object(w, "_WORKER_KEY_FILE", f), patch.object(w, "_worker_key", None):
+            w._save_worker_key("k1")
+            assert f.read_text() == "k1"
+            assert w._load_worker_key() == "k1"
+
+    def test_load_missing_file_is_none(self, tmp_path):
+        with patch.object(w, "_WORKER_KEY_FILE", tmp_path / "nope"):
+            assert w._load_worker_key() is None
+
+
+class TestInboundVerify:
+    def _req(self, token):
+        r = MagicMock()
+        r.headers = {"Authorization": f"Bearer {token}"}
+        return r
+
+    def test_requires_own_key_once_enrolled(self):
+        with patch.object(w, "_worker_key", "own"), patch.object(w, "API_KEY", "shared"):
+            assert w._verify_api_key(self._req("own")) is None  # own key accepted
+            with pytest.raises(HTTPException) as ei:
+                w._verify_api_key(self._req("shared"))  # shared rejected post-cutover
+            assert ei.value.status_code == 401
+
+    def test_accepts_shared_before_enrollment(self):
+        with patch.object(w, "_worker_key", None), patch.object(w, "API_KEY", "shared"):
+            assert w._verify_api_key(self._req("shared")) is None
+
+    def test_503_when_no_key_configured(self):
+        with patch.object(w, "_worker_key", None), patch.object(w, "API_KEY", ""):
+            with pytest.raises(HTTPException) as ei:
+                w._verify_api_key(self._req("anything"))
+            assert ei.value.status_code == 503
+
+
+class TestHeartbeatEnrollment:
+    def test_heartbeat_persists_issued_key(self, tmp_path):
+        f = tmp_path / ".worker_key"
+        resp = MagicMock()
+        resp.raise_for_status = MagicMock()
+        resp.json = MagicMock(return_value={"status": "ok", "worker_id": 1, "worker_key": "issued-key"})
+        mock_client = MagicMock()
+        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_client.__aexit__ = AsyncMock(return_value=False)
+        mock_client.post = AsyncMock(return_value=resp)
+
+        with (
+            patch.object(w, "_WORKER_KEY_FILE", f),
+            patch.object(w, "_worker_key", None),
+            patch.object(w, "UI_URL", "http://ui:8080"),
+            patch.object(w, "API_KEY", "shared"),
+            patch("app.worker_api.orchestrator.get_status", return_value=[]),
+            patch("app.worker_api.orchestrator.docker_available", return_value=True),
+            patch("app.worker_api.httpx.AsyncClient", return_value=mock_client),
+        ):
+            asyncio.run(w._send_heartbeat())
+            # The issued key was persisted and adopted.
+            assert f.read_text() == "issued-key"
+            assert w._worker_key == "issued-key"
diff --git a/tests/test_workers.py b/tests/test_workers.py
index 9227154..88b9711 100644
--- a/tests/test_workers.py
+++ b/tests/test_workers.py
@@ -83,7 +83,10 @@ class TestHeartbeatClientId:
     def test_client_id_passed_to_upsert(self):
         """When client_id is provided, it's used as the worker identity."""
         mock_upsert = AsyncMock(return_value=42)
-        with patch("app.main.database.upsert_worker", mock_upsert):
+        with (
+            patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value=False),
+            patch("app.main.database.upsert_worker", mock_upsert),
+        ):
             result = _run(
                 api_worker_heartbeat(
                     _request(),
@@ -105,7 +108,10 @@ def test_client_id_passed_to_upsert(self):
     def test_fallback_to_name_when_no_client_id(self):
         """Old workers that don't send client_id get name used as identity."""
         mock_upsert = AsyncMock(return_value=7)
-        with patch("app.main.database.upsert_worker", mock_upsert):
+        with (
+            patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value=False),
+            patch("app.main.database.upsert_worker", mock_upsert),
+        ):
             result = _run(
                 api_worker_heartbeat(
                     _request(),
@@ -130,7 +136,10 @@ async def fake_upsert(**kwargs):
             calls.append(kwargs)
             return len(calls)
 
-        with patch("app.main.database.upsert_worker", side_effect=fake_upsert):
+        with (
+            patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value=False),
+            patch("app.main.database.upsert_worker", side_effect=fake_upsert),
+        ):
             _run(
                 api_worker_heartbeat(
                     _request(),