diff --git a/shard_core/service/backup.py b/shard_core/service/backup.py
index e623fa5..aa05d69 100644
--- a/shard_core/service/backup.py
+++ b/shard_core/service/backup.py
@@ -2,7 +2,6 @@
 import datetime
 import json
 import logging
-import subprocess
 import traceback
 from pathlib import Path
 from typing import List
@@ -30,23 +29,54 @@
 STORE_KEY_BACKUP_PASSPHRASE_LAST_ACCESS = "backup_passphrase_last_access"
 BACKUP_IN_PROGESS_LOCK = asyncio.Lock()
 
-COMMAND_TEMPLATE = """
-rclone
---azureblob-sas-url {sas_token}
---crypt-password {obscured_password}
---crypt-remote :azureblob:{container_name}
-sync {directory} :crypt:{container_name}/{directory}
---create-empty-src-dirs --stats-log-level NOTICE --stats 1000m --use-json-log
---fast-list
-"""
-
-CLEARTEXT_COMMAND_TEMPLATE = """
-rclone
---azureblob-sas-url {sas_token}
-sync {directory} :azureblob:{container_name}/{directory}
---create-empty-src-dirs --stats-log-level NOTICE --stats 1000m --use-json-log
---fast-list
-"""
+# Strong references to fire-and-forget tasks. asyncio only holds tasks weakly, so
+# without this a long-running backup could be garbage-collected mid-run. Mirrors
+# app_lifecycle.background_tasks.
+background_tasks = set()
+
+
+def _build_backup_command(
+    sas_token: str, obscured_password: str, container_name: str, directory: Path
+) -> List[str]:
+    return [
+        "rclone",
+        "--azureblob-sas-url",
+        sas_token,
+        "--crypt-password",
+        obscured_password,
+        "--crypt-remote",
+        f":azureblob:{container_name}",
+        "sync",
+        str(directory),
+        f":crypt:{container_name}/{directory}",
+        "--create-empty-src-dirs",
+        "--stats-log-level",
+        "NOTICE",
+        "--stats",
+        "1000m",
+        "--use-json-log",
+        "--fast-list",
+    ]
+
+
+def _build_cleartext_backup_command(
+    sas_token: str, container_name: str, directory: Path
+) -> List[str]:
+    return [
+        "rclone",
+        "--azureblob-sas-url",
+        sas_token,
+        "sync",
+        str(directory),
+        f":azureblob:{container_name}/{directory}",
+        "--create-empty-src-dirs",
+        "--stats-log-level",
+        "NOTICE",
+        "--stats",
+        "1000m",
+        "--use-json-log",
+        "--fast-list",
+    ]
 
 
 async def start_backup():
@@ -67,6 +97,7 @@ async def start_backup():
             directories, sas_url_response.container_name, sas_url_response.sas_url
         )
     )
+    _register_background_task(task)
 
     async def on_task_done(task: asyncio.Task):
         if task.exception():
@@ -78,7 +109,16 @@ async def on_task_done(task: asyncio.Task):
         else:
             signals.on_backup_update.send()
 
-    task.add_done_callback(lambda task: asyncio.create_task(on_task_done(task)))
+    task.add_done_callback(
+        lambda completed: _register_background_task(
+            asyncio.create_task(on_task_done(completed))
+        )
+    )
+
+
+def _register_background_task(task: asyncio.Task):
+    background_tasks.add(task)
+    task.add_done_callback(background_tasks.discard)
 
 
 async def backup_directories(
@@ -130,11 +170,11 @@ async def backup_directories(
         )
         async with db_conn() as conn:
             await db_backups.insert(conn, report)
-        _write_marker_blob(container_name, sas_token)
+        await _write_marker_blob(container_name, sas_token)
         log.info("Backup done")
 
 
-def _write_marker_blob(container_name: str, sas_token: str):
+async def _write_marker_blob(container_name: str, sas_token: str):
     """Write a marker blob to the backup container to record the time of the last backup.
 
     The blob's Last-Modified timestamp on Azure is updated on every call, giving the
@@ -149,7 +189,7 @@ def _write_marker_blob(container_name: str, sas_token: str):
         blob_url = urlunparse(parsed._replace(path=f"/{container_name}/_last_backup"))
         blob_client = BlobClient.from_blob_url(blob_url)
         timestamp = datetime.datetime.now(datetime.timezone.utc).isoformat()
-        blob_client.upload_blob(timestamp, overwrite=True)
+        await asyncio.to_thread(blob_client.upload_blob, timestamp, overwrite=True)
         log.debug("Wrote _last_backup marker blob")
     except Exception:
         log.warning("Failed to write _last_backup marker blob", exc_info=True)
@@ -162,7 +202,7 @@ def is_backup_in_progress():
 async def _backup_directory(
     container_name, rel_directory, obscured_passphrase, sas_token
 ):
-    command = COMMAND_TEMPLATE.format(
+    command = _build_backup_command(
         sas_token=sas_token,
         obscured_password=obscured_passphrase,
         container_name=container_name,
@@ -170,7 +210,7 @@ async def _backup_directory(
     )
     path_root = Path(settings().path_root)
     process = await asyncio.create_subprocess_exec(
-        *command.split(),
+        *command,
         cwd=path_root,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
@@ -191,10 +231,19 @@ async def _backup_directory(
 
 async def _get_obscured_passphrase():
     passphrase = await database.get_value(STORE_KEY_BACKUP_PASSPHRASE)
-    obscured_passphrase = subprocess.run(
-        ["rclone", "obscure", passphrase], capture_output=True, text=True
-    ).stdout.strip()
-    return obscured_passphrase
+    process = await asyncio.create_subprocess_exec(
+        "rclone",
+        "obscure",
+        passphrase,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await process.communicate()
+    if process.returncode != 0:
+        message = stderr.decode().strip()
+        log.error(f"rclone obscure exited with {process.returncode}: {message}")
+        raise BackupFailedError(message)
+    return stdout.decode().strip()
 
 
 def _get_relative_directory(directory: Path) -> Path:
diff --git a/tests/test_backup.py b/tests/test_backup.py
index d6cc880..214ec02 100644
--- a/tests/test_backup.py
+++ b/tests/test_backup.py
@@ -1,31 +1,68 @@
+import asyncio
 import json
-from unittest.mock import AsyncMock, patch
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
 from shard_core.service import backup
 from shard_core.service.backup import (
-    COMMAND_TEMPLATE,
-    CLEARTEXT_COMMAND_TEMPLATE,
     BackupFailedError,
     _backup_directory,
+    _build_backup_command,
+    _build_cleartext_backup_command,
+    _get_obscured_passphrase,
+    _write_marker_blob,
+    start_backup,
 )
 
 
-def test_command_templates_only_use_supported_flags():
+def test_backup_commands_only_use_supported_flags():
     """Guard against reintroducing rclone flags missing from the bundled binary.
 
     The core image ships rclone 1.60.1 (Debian bookworm), which lacks
     --azureblob-no-check-container (added in 1.61.0). See issue #117.
     """
-    for template in (COMMAND_TEMPLATE, CLEARTEXT_COMMAND_TEMPLATE):
-        assert "--azureblob-no-check-container" not in template
-        assert "--fast-list" in template
+    argvs = (
+        _build_backup_command("sas", "obscured", "container", "dir"),
+        _build_cleartext_backup_command("sas", "container", "dir"),
+    )
+    for argv in argvs:
+        assert "--azureblob-no-check-container" not in argv
+        assert "--fast-list" in argv
 
 
-def _mock_process(returncode, stderr):
+def test_build_backup_command_keeps_whitespace_values_as_single_argv():
+    """A SAS URL or path containing whitespace must stay one argv element,
+    not be split into several tokens (the bug behind command.split())."""
+    sas_with_space = "https://acct.blob.core.windows.net/c?sv=a b&sig=x y"
+    argv = _build_backup_command(sas_with_space, "obscured", "container", "my dir")
+
+    assert sas_with_space in argv
+    assert "my dir" in argv
+    assert argv[0] == "rclone"
+    # the crypt destination for a spaced directory stays intact
+    assert ":crypt:container/my dir" in argv
+
+
+@pytest.mark.parametrize(
+    "builder,args",
+    [
+        (_build_backup_command, ("sas", "obscured", "container", Path("my dir"))),
+        (_build_cleartext_backup_command, ("sas", "container", Path("my dir"))),
+    ],
+)
+def test_backup_command_argv_is_flat_string_list(builder, args):
+    # A Path directory must be stringified into a single str token, never leak
+    # a Path object into argv (which create_subprocess_exec would reject).
+    argv = builder(*args)
+    assert all(isinstance(token, str) for token in argv)
+
+
+def _mock_process(returncode, stdout=b"", stderr=b""):
     process = AsyncMock()
-    process.communicate.return_value = (b"", stderr)
+    process.communicate.return_value = (stdout, stderr)
     process.returncode = returncode
     return process
 
@@ -34,17 +71,198 @@ def _mock_process(returncode, stderr):
 async def test_backup_directory_returns_stats_on_success():
     stats = {"bytes": 42, "errors": 0}
     stderr = (json.dumps({"stats": stats}) + "\n").encode()
-    process = _mock_process(0, stderr)
+    process = _mock_process(0, stderr=stderr)
     with patch.object(backup.asyncio, "create_subprocess_exec", return_value=process):
         result = await _backup_directory("container", "dir", "obscured", "sas")
     assert result == stats
 
 
+@pytest.mark.asyncio
+async def test_backup_directory_passes_unsplit_argv_to_subprocess():
+    """The whitespace fix lives at the call site: the built list must reach
+    create_subprocess_exec verbatim, not be re-split."""
+    stats = {"bytes": 1, "errors": 0}
+    stderr = (json.dumps({"stats": stats}) + "\n").encode()
+    create = AsyncMock(return_value=_mock_process(0, stderr=stderr))
+    with patch.object(backup.asyncio, "create_subprocess_exec", create):
+        await _backup_directory("container", Path("my dir"), "obscured", "sas url")
+
+    expected = _build_backup_command(
+        sas_token="sas url",
+        obscured_password="obscured",
+        container_name="container",
+        directory=Path("my dir"),
+    )
+    assert create.await_args.args == tuple(expected)
+
+
 @pytest.mark.asyncio
 async def test_backup_directory_raises_on_nonzero_exit():
     stderr = b"Error: unknown flag: --azureblob-no-check-container\n"
-    process = _mock_process(1, stderr)
+    process = _mock_process(1, stderr=stderr)
     with patch.object(backup.asyncio, "create_subprocess_exec", return_value=process):
         with pytest.raises(BackupFailedError) as exc_info:
             await _backup_directory("container", "dir", "obscured", "sas")
     assert "unknown flag: --azureblob-no-check-container" in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_get_obscured_passphrase_uses_async_subprocess():
+    process = _mock_process(0, stdout=b"OBSCURED\n")
+    create = AsyncMock(return_value=process)
+    with (
+        patch.object(backup.asyncio, "create_subprocess_exec", create),
+        patch.object(backup.database, "get_value", AsyncMock(return_value="secret")),
+    ):
+        result = await _get_obscured_passphrase()
+
+    assert result == "OBSCURED"
+    create.assert_awaited_once()
+    assert create.await_args.args == ("rclone", "obscure", "secret")
+
+
+@pytest.mark.asyncio
+async def test_get_obscured_passphrase_raises_on_nonzero_exit():
+    process = _mock_process(1, stderr=b"boom\n")
+    with (
+        patch.object(
+            backup.asyncio, "create_subprocess_exec", AsyncMock(return_value=process)
+        ),
+        patch.object(backup.database, "get_value", AsyncMock(return_value="secret")),
+    ):
+        with pytest.raises(BackupFailedError) as exc_info:
+            await _get_obscured_passphrase()
+    assert "boom" in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_write_marker_blob_offloads_upload_to_thread():
+    """The blocking Azure upload must run in a worker thread, never directly on
+    the event loop, and the timestamp payload + overwrite flag must be forwarded."""
+    blob_client = MagicMock()
+    ctor = MagicMock(return_value=blob_client)
+    to_thread = AsyncMock()
+    with (
+        patch.object(backup.BlobClient, "from_blob_url", ctor),
+        patch.object(backup.asyncio, "to_thread", to_thread),
+    ):
+        await _write_marker_blob(
+            "container", "https://acct.blob.core.windows.net/c?sv=x&sig=y"
+        )
+
+    # the blob path is inserted before the query string, replacing the container
+    ctor.assert_called_once_with(
+        "https://acct.blob.core.windows.net/container/_last_backup?sv=x&sig=y"
+    )
+    # upload happened via to_thread, not a direct (loop-blocking) call
+    blob_client.upload_blob.assert_not_called()
+    to_thread.assert_awaited_once()
+    func, *rest = to_thread.await_args.args
+    assert func == blob_client.upload_blob
+    assert rest and isinstance(rest[0], str)  # ISO timestamp payload
+    assert to_thread.await_args.kwargs == {"overwrite": True}
+
+
+@pytest.mark.asyncio
+async def test_write_marker_blob_swallows_setup_errors():
+    with patch.object(
+        backup.BlobClient, "from_blob_url", side_effect=RuntimeError("nope")
+    ):
+        # must not raise - the marker is best-effort
+        await _write_marker_blob("container", "https://acct/c?sig=y")
+
+
+@pytest.mark.asyncio
+async def test_write_marker_blob_swallows_upload_errors():
+    with (
+        patch.object(backup.BlobClient, "from_blob_url", return_value=MagicMock()),
+        patch.object(
+            backup.asyncio,
+            "to_thread",
+            AsyncMock(side_effect=RuntimeError("upload failed")),
+        ),
+    ):
+        # an error raised inside the worker thread is also best-effort
+        await _write_marker_blob("container", "https://acct/c?sig=y")
+
+
+async def _drain_background_tasks():
+    # Chained done-callbacks (backup task -> spawned on_task_done task -> discard)
+    # need several loop turns to settle; yield until the set drains.
+    for _ in range(1000):
+        await asyncio.sleep(0)
+        if not backup.background_tasks:
+            return
+    raise AssertionError("background tasks did not drain")
+
+
+async def _start_backup_and_capture_signal(fake_backup):
+    sas = SimpleNamespace(container_name="c", sas_url="s")
+    signal = MagicMock()
+    with (
+        patch.object(backup, "get_backup_sas_url", AsyncMock(return_value=sas)),
+        patch.object(backup, "backup_directories", fake_backup),
+        patch.object(backup.signals, "on_backup_update", signal),
+    ):
+        await start_backup()
+        await _drain_background_tasks()
+    return signal
+
+
+@pytest.mark.asyncio
+async def test_start_backup_registers_strong_refs_for_both_tasks():
+    """Both the backup task and the spawned on_task_done wrapper must be held by
+    the module-level set (asyncio holds tasks only weakly), then drain to empty."""
+    release = asyncio.Event()
+    started = asyncio.Event()
+
+    async def fake_backup(*_args, **_kwargs):
+        started.set()
+        await release.wait()
+
+    sas = SimpleNamespace(container_name="c", sas_url="s")
+    register_spy = MagicMock(wraps=backup._register_background_task)
+    with (
+        patch.object(backup, "get_backup_sas_url", AsyncMock(return_value=sas)),
+        patch.object(backup, "backup_directories", fake_backup),
+        patch.object(backup, "_register_background_task", register_spy),
+        patch.object(backup.signals, "on_backup_update"),
+    ):
+        assert backup.background_tasks == set()
+        await start_backup()
+        await asyncio.wait_for(started.wait(), timeout=5)
+
+        # the still-running backup task is held strongly
+        assert len(backup.background_tasks) == 1
+        assert register_spy.call_count == 1
+
+        release.set()
+        for _ in range(1000):
+            await asyncio.sleep(0)
+            if register_spy.call_count >= 2 and not backup.background_tasks:
+                break
+
+        # the on_task_done wrapper was also registered (not left unreferenced)
+        assert register_spy.call_count == 2
+        assert backup.background_tasks == set()
+
+
+@pytest.mark.asyncio
+async def test_start_backup_signals_success_on_completion():
+    async def fake_backup(*_args, **_kwargs):
+        return None
+
+    signal = await _start_backup_and_capture_signal(fake_backup)
+    signal.send.assert_called_once_with()
+
+
+@pytest.mark.asyncio
+async def test_start_backup_signals_failure_with_exception():
+    boom = RuntimeError("backup blew up")
+
+    async def fake_backup(*_args, **_kwargs):
+        raise boom
+
+    signal = await _start_backup_and_capture_signal(fake_backup)
+    signal.send.assert_called_once()
+    assert signal.send.call_args.args[0] is boom