diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 2dfcf64..00fcfe5 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -20,7 +20,7 @@ jobs:
 
     strategy:
       matrix:
-        os: [ubuntu-latest, windows-latest, macOS-latest]
+        os: [ubuntu-24.04, windows-latest, macOS-latest]
         python_version: [3.9]
 
     steps:
diff --git a/aw_client/client.py b/aw_client/client.py
index 1cacba4..59867ae 100644
--- a/aw_client/client.py
+++ b/aw_client/client.py
@@ -21,7 +21,7 @@
     Tuple,
     Union,
 )
-from urllib.parse import urlsplit
+from urllib.parse import quote, urlsplit
 
 import persistqueue
 import requests as req
@@ -262,7 +262,7 @@ def get_event(
         bucket_id: str,
         event_id: int,
     ) -> Optional[Event]:
-        endpoint = f"buckets/{bucket_id}/events/{event_id}"
+        endpoint = f"buckets/{quote(bucket_id, safe='')}/events/{event_id}"
         try:
             event = self._get(endpoint).json()
             return Event(**event)
@@ -279,7 +279,7 @@ def get_events(
         start: Optional[datetime] = None,
         end: Optional[datetime] = None,
     ) -> List[Event]:
-        endpoint = f"buckets/{bucket_id}/events"
+        endpoint = f"buckets/{quote(bucket_id, safe='')}/events"
 
         params = dict()  # type: Dict[str, str]
         if limit is not None:
@@ -293,7 +293,7 @@ def get_events(
         return [Event(**event) for event in events]
 
     def insert_event(self, bucket_id: str, event: Event) -> Optional[Event]:
-        endpoint = f"buckets/{bucket_id}/events"
+        endpoint = f"buckets/{quote(bucket_id, safe='')}/events"
         data = [event.to_json_dict()]
         response = self._post(endpoint, data)
         if response.json():
@@ -303,7 +303,7 @@ def insert_event(self, bucket_id: str, event: Event) -> Optional[Event]:
     def insert_events(
         self, bucket_id: str, events: List[Event]
     ) -> Optional[List[Event]]:
-        endpoint = f"buckets/{bucket_id}/events"
+        endpoint = f"buckets/{quote(bucket_id, safe='')}/events"
         data = [event.to_json_dict() for event in events]
         response = self._post(endpoint, data)
         if response.json():
@@ -311,7 +311,7 @@ def insert_events(
         return None
 
     def delete_event(self, bucket_id: str, event_id: int) -> None:
-        endpoint = f"buckets/{bucket_id}/events/{event_id}"
+        endpoint = f"buckets/{quote(bucket_id, safe='')}/events/{event_id}"
         self._delete(endpoint)
 
     def delete_events(self, bucket_id: str, event_ids: List[int]) -> int:
@@ -321,7 +321,7 @@ def delete_events(self, bucket_id: str, event_ids: List[int]) -> int:
         Unknown and duplicate ids are ignored. Returns the number of events
         that existed and were deleted.
         """
-        endpoint = f"buckets/{bucket_id}/events/delete"
+        endpoint = f"buckets/{quote(bucket_id, safe='')}/events/delete"
         return int(self._post(endpoint, {"ids": list(event_ids)}).json())
 
     def get_eventcount(
@@ -331,7 +331,7 @@ def get_eventcount(
         start: Optional[datetime] = None,
         end: Optional[datetime] = None,
     ) -> int:
-        endpoint = f"buckets/{bucket_id}/events/count"
+        endpoint = f"buckets/{quote(bucket_id, safe='')}/events/count"
 
         params = dict()  # type: Dict[str, str]
         if start is not None:
@@ -363,7 +363,9 @@ def heartbeat(
               the function will in that case always returns None.
         """
 
-        endpoint = f"buckets/{bucket_id}/heartbeat?pulsetime={pulsetime}"
+        endpoint = (
+            f"buckets/{quote(bucket_id, safe='')}/heartbeat?pulsetime={pulsetime}"
+        )
         _commit_interval = commit_interval or self.commit_interval
 
         if queued:
@@ -435,7 +437,7 @@ def create_bucket(self, bucket_id: str, event_type: str, queued=False):
             self._warn_queue_before_connect()
             self.request_queue.register_bucket(bucket_id, event_type)
         else:
-            endpoint = f"buckets/{bucket_id}"
+            endpoint = f"buckets/{quote(bucket_id, safe='')}"
             data = {
                 "client": self.client_name,
                 "hostname": self.client_hostname,
@@ -444,7 +446,9 @@ def create_bucket(self, bucket_id: str, event_type: str, queued=False):
             self._post(endpoint, data)
 
     def delete_bucket(self, bucket_id: str, force: bool = False):
-        self._delete(f"buckets/{bucket_id}" + ("?force=1" if force else ""))
+        self._delete(
+            f"buckets/{quote(bucket_id, safe='')}" + ("?force=1" if force else "")
+        )
 
     # @deprecated
     def setup_bucket(self, bucket_id: str, event_type: str):
@@ -456,7 +460,7 @@ def export_all(self) -> dict:
         return self._get("export").json()
 
     def export_bucket(self, bucket_id) -> dict:
-        return self._get(f"buckets/{bucket_id}/export").json()
+        return self._get(f"buckets/{quote(bucket_id, safe='')}/export").json()
 
     def import_bucket(self, bucket: dict) -> None:
         endpoint = "import"
diff --git a/tests/test_bucket_paths.py b/tests/test_bucket_paths.py
new file mode 100644
index 0000000..c9959bc
--- /dev/null
+++ b/tests/test_bucket_paths.py
@@ -0,0 +1,316 @@
+"""Bucket IDs are literal URL segments, even after requests prepares the URL.
+
+Use a real localhost HTTP server: mocking requests would miss its percent-escape
+normalization and the loss of endpoint suffixes to query strings/fragments.
+"""
+
+import json
+import threading
+from datetime import datetime, timedelta, timezone
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any, Dict, List, Tuple
+from urllib.parse import parse_qs, quote, unquote, urlsplit
+
+import pytest
+from aw_core.models import Event
+
+from aw_client import ActivityWatchClient
+from aw_client import client as client_module
+
+BUCKET_IDS = [
+    "ordinary-bucket",
+    "bucket#fragment",
+    "bucket?force=1&other=query",
+    "bucket%literal",
+    "bucket%23escape%3F%25%41",
+    "bucket-雪-é",
+]
+START = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
+END = START + timedelta(minutes=1)
+
+
+@pytest.fixture
+def wire_client(monkeypatch, tmp_path):
+    records = []
+    responses: List[Any] = []
+
+    class Handler(BaseHTTPRequestHandler):
+        def handle_request(self):
+            raw_body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
+            records.append(
+                (self.command, self.path, json.loads(raw_body) if raw_body else None)
+            )
+            if responses and callable(responses[0]):
+                status, body = responses[0](records[-1])
+            else:
+                status, body = responses.pop(0) if responses else (200, {})
+            payload = json.dumps(body).encode("utf-8")
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+
+        do_GET = do_POST = do_DELETE = handle_request
+
+        def log_message(self, *args):
+            pass
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(
+        target=server.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True
+    )
+    thread.start()
+    # Isolate config, locks, API keys and the real persistent queue, not HTTP.
+    monkeypatch.setattr(
+        client_module,
+        "load_config",
+        lambda: {
+            "server": {"hostname": "127.0.0.1", "port": server.server_port},
+            "client": {"commit_interval": 10},
+        },
+    )
+    monkeypatch.setattr(client_module, "SingleInstance", lambda name: object())
+    monkeypatch.setattr(client_module, "get_data_dir", lambda *args: str(tmp_path))
+    monkeypatch.setattr(
+        client_module, "load_local_server_api_key", lambda *a, **kw: None
+    )
+    client = ActivityWatchClient(
+        "bucket-path-test", host="127.0.0.1", port=server.server_port, profile="default"
+    )
+    try:
+        yield client, records, responses
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+def _assert_bucket_target(record, method, bucket_id, suffix, params):
+    actual_method, target, _body = record
+    parsed = urlsplit(target)
+    assert actual_method == method
+    assert parsed.path == f"/api/0/buckets/{quote(bucket_id, safe='')}{suffix}"
+    assert unquote(parsed.path.split("/")[4]) == bucket_id
+    assert parse_qs(parsed.query) == params
+    assert not parsed.fragment
+
+
+# Every constructor is exercised separately, so a missed site cannot be hidden
+# by a failure in an earlier operation on the same bucket.
+OPERATIONS = [
+    ("get_event", "GET", "/events/7"),
+    ("get_events", "GET", "/events"),
+    ("insert_event", "POST", "/events"),
+    ("insert_events", "POST", "/events"),
+    ("delete_event", "DELETE", "/events/7"),
+    ("delete_events", "POST", "/events/delete"),
+    ("get_eventcount", "GET", "/events/count"),
+    ("create_bucket", "POST", ""),
+    ("delete_bucket", "DELETE", ""),
+    ("export_bucket", "GET", "/export"),
+    ("heartbeat", "POST", "/heartbeat"),
+]
+
+
+@pytest.mark.parametrize("bucket_id", BUCKET_IDS)
+@pytest.mark.parametrize("operation,method,suffix", OPERATIONS)
+def test_immediate_bucket_endpoints(wire_client, bucket_id, operation, method, suffix):
+    client, records, responses = wire_client
+    event = Event(timestamp=START, duration=1, data={"literal_id": bucket_id})
+    event_json = event.to_json_dict()
+    params = {}
+    body: Any = None
+    response: Any = {}
+    args: Tuple[Any, ...] = ()
+    kwargs: Dict[str, Any] = {}
+    if operation in {"get_event", "delete_event"}:
+        args = (7,)
+        response = event_json
+    elif operation in {"get_events", "get_eventcount"}:
+        kwargs = {"start": START, "end": END}
+        params = {"start": [START.isoformat()], "end": [END.isoformat()]}
+        if operation == "get_events":
+            kwargs["limit"] = 3
+            params["limit"] = ["3"]
+            response = [event_json]
+        else:
+            response = 1
+    elif operation in {"insert_event", "insert_events"}:
+        args = (event if operation == "insert_event" else [event],)
+        body = response = [event_json]
+    elif operation == "delete_events":
+        args = ([7, 8],)
+        body = {"ids": [7, 8]}
+        response = 2
+    elif operation == "create_bucket":
+        args = ("test-event",)
+        body = {
+            "client": client.client_name,
+            "hostname": client.client_hostname,
+            "type": "test-event",
+        }
+    elif operation == "delete_bucket":
+        kwargs = {"force": True}
+        params = {"force": ["1"]}
+    elif operation == "heartbeat":
+        args = (event,)
+        kwargs = {"pulsetime": 12.5}
+        params = {"pulsetime": ["12.5"]}
+        body = event_json
+    if method == "DELETE":
+        body = {}
+    responses.append((200, response))
+
+    result = getattr(client, operation)(bucket_id, *args, **kwargs)
+
+    assert len(records) == 1
+    _assert_bucket_target(records[0], method, bucket_id, suffix, params)
+    assert records[0][2] == body
+    if operation == "get_eventcount":
+        assert result == 1
+    elif operation == "delete_events":
+        assert result == 2
+    elif operation == "get_event":
+        assert result.data == {"literal_id": bucket_id}
+    elif operation in {"get_events", "insert_events"}:
+        assert result[0].data == {"literal_id": bucket_id}
+    elif operation == "insert_event":
+        assert result.data == {"literal_id": bucket_id}
+
+
+@pytest.mark.parametrize("bucket_id", BUCKET_IDS)
+def test_queued_heartbeat_retries_literal_bucket(wire_client, bucket_id):
+    client, records, responses = wire_client
+    queue = client.request_queue
+    # Drive the real queue deterministically without a background dispatcher.
+    with pytest.warns(UserWarning, match="connect"):
+        client.create_bucket(bucket_id, "test-event", queued=True)
+    assert queue._registered_buckets[0].id == bucket_id
+    connected = queue._try_connect()
+    assert connected
+    first = Event(timestamp=START, duration=1, data={"literal_id": bucket_id})
+    second = Event(timestamp=END, duration=1, data={"next": bucket_id})
+    client.heartbeat(bucket_id, first, pulsetime=12.5, queued=True)
+    client.heartbeat(bucket_id, second, pulsetime=12.5, queued=True)
+    assert set(client.last_heartbeat) == {bucket_id}
+    assert queue._persistqueue.qsize() == 1
+    responses.extend([(503, {"message": "retry"}), (200, {})])
+
+    queue._dispatch_request()
+    assert queue._current is not None
+    assert not queue.wait_for_queue_empty(timeout=0)
+    queue._dispatch_request()
+    assert queue._current is None
+    assert queue._persistqueue.qsize() == 0
+
+    assert len(records) == 3
+    _assert_bucket_target(records[0], "POST", bucket_id, "", {})
+    for record in records[1:]:
+        _assert_bucket_target(
+            record, "POST", bucket_id, "/heartbeat", {"pulsetime": ["12.5"]}
+        )
+        assert record[2] == first.to_json_dict()
+    assert records[1] == records[2]
+
+
+@pytest.mark.parametrize("bucket_id", BUCKET_IDS)
+def test_delete_bucket_does_not_add_force_by_default(wire_client, bucket_id):
+    client, records, _responses = wire_client
+    client.delete_bucket(bucket_id)
+    _assert_bucket_target(records[0], "DELETE", bucket_id, "", {})
+
+
+@pytest.mark.parametrize("bucket_id", BUCKET_IDS)
+def test_import_keeps_literal_bucket_id_in_body(wire_client, bucket_id):
+    client, records, _responses = wire_client
+    bucket = {"id": bucket_id, "type": "test-event", "events": []}
+    client.import_bucket(bucket)
+    assert records == [("POST", "/api/0/import", {"buckets": {bucket_id: bucket}})]
+
+
+@pytest.fixture
+def memory_server(wire_client, monkeypatch):
+    # Optional integration dependencies must not skip the standalone wire suite.
+    server_module = pytest.importorskip("aw_server.server")
+    storages = pytest.importorskip("aw_datastore.storages")
+    monkeypatch.delenv("AW_PROFILE", raising=False)
+    app = server_module.AWFlask(
+        "127.0.0.1",
+        testing=True,
+        storage_method=storages.MemoryStorage,
+        query_cache=False,
+    )
+    flask_client = app.test_client()
+    client, _records, responses = wire_client
+    # Independently establish server support before exercising SDK routing.
+    for bucket_id in BUCKET_IDS:
+        route = f"/api/0/buckets/{quote(bucket_id, safe='')}"
+        created = flask_client.post(
+            route,
+            json={"type": "test-event", "client": "control", "hostname": "fixture"},
+        )
+        assert created.status_code == 200
+        assert bucket_id in flask_client.get("/api/0/buckets/").get_json()
+        deleted = flask_client.delete(route + "?force=1")
+        assert deleted.status_code == 200
+
+    def forward(record):
+        method, target, body = record
+        response = flask_client.open(target, method=method, json=body)
+        # Existing buckets return 304 with no body. Do not forward attachment
+        # headers: Unicode filenames are unrelated to path-segment routing.
+        payload = response.get_data(as_text=True)
+        return response.status_code, json.loads(payload) if payload else None
+
+    responses.append(forward)
+    return client
+
+
+@pytest.mark.parametrize("bucket_id", BUCKET_IDS)
+def test_memory_server_keeps_buckets_distinct(memory_server, bucket_id):
+    client = memory_server
+    # These sentinels catch fragment/query truncation and percent decoding into
+    # another real bucket, not just the happy-path response of the target.
+    unrelated = {"bucket", "bucket#escape?%A", "sentinel"} - {bucket_id}
+    for sentinel in unrelated:
+        client.create_bucket(sentinel, "test-event")
+        client.insert_event(
+            sentinel, Event(timestamp=START, duration=1, data={"sentinel": sentinel})
+        )
+    before = {sentinel: client.export_bucket(sentinel) for sentinel in unrelated}
+    client.create_bucket(bucket_id, "test-event")
+    assert bucket_id in client.get_buckets()
+    event = Event(timestamp=START, duration=1, data={"literal_id": bucket_id})
+    inserted = client.insert_event(bucket_id, event)
+    assert inserted is not None
+    assert client.get_event(bucket_id, inserted.id).data == event.data
+    assert client.get_events(bucket_id, start=START, end=END)[0].data == event.data
+    assert client.get_eventcount(bucket_id, start=START, end=END) == 1
+    assert bucket_id in client.export_bucket(bucket_id)["buckets"]
+    client.delete_event(bucket_id, inserted.id)
+    inserted_many = client.insert_events(bucket_id, [event])
+    assert inserted_many
+    deleted_count = client.delete_events(bucket_id, [inserted_many[0].id])
+    assert deleted_count == 1
+    client.heartbeat(bucket_id, event, pulsetime=12.5)
+    assert client.get_eventcount(bucket_id) == 1
+    with pytest.warns(UserWarning, match="connect"):
+        client.setup_bucket(bucket_id, "test-event")
+    connected = client.request_queue._try_connect()
+    assert connected
+    client.heartbeat(bucket_id, event, pulsetime=12.5, queued=True)
+    client.heartbeat(
+        bucket_id,
+        Event(timestamp=END, duration=1, data={"next": bucket_id}),
+        pulsetime=12.5,
+        queued=True,
+    )
+    client.request_queue._dispatch_request()
+    assert client.request_queue._current is None
+    assert client.get_eventcount(bucket_id) == 1
+    client.delete_bucket(bucket_id, force=True)
+    assert bucket_id not in client.get_buckets()
+    assert {
+        sentinel: client.export_bucket(sentinel) for sentinel in unrelated
+    } == before