diff --git a/solstone/apps/health/workspace.html b/solstone/apps/health/workspace.html
index 26ae912f6..58819222c 100644
--- a/solstone/apps/health/workspace.html
+++ b/solstone/apps/health/workspace.html
@@ -3175,7 +3175,7 @@
 
       const nameEl = document.createElement('span');
       nameEl.className = 'registered-observer-name';
-      nameEl.textContent = observer.name || observer.key_prefix || 'observer';
+      nameEl.textContent = observer.name || observer.prefix || 'observer';
       row.appendChild(nameEl);
 
       const labelEl = document.createElement('span');
diff --git a/solstone/apps/import/ingest.py b/solstone/apps/import/ingest.py
index fb205185f..5e1b56a7d 100644
--- a/solstone/apps/import/ingest.py
+++ b/solstone/apps/import/ingest.py
@@ -103,6 +103,9 @@ from .facet_ingest import process_facet
 
 
 def register_ingest_routes(bp) -> None:
+    # Import ingest is the bulk producer stream for peer/export segments plus
+    # entities, facets, imports, and config; it is deliberately separate from
+    # live observer ingest.
     @bp.route("/journal/<key_prefix>/ingest/segments", methods=["POST"])
     @require_journal_source
     def ingest_segments(key_prefix: str):
diff --git a/solstone/apps/observer/routes.py b/solstone/apps/observer/routes.py
index 00cd6c349..b4c2567f4 100644
--- a/solstone/apps/observer/routes.py
+++ b/solstone/apps/observer/routes.py
@@ -6,7 +6,6 @@
 Provides endpoints for:
 - Managing observer registrations (UI)
 - Receiving file uploads from observers (ingest)
-- Receiving transferred segments from other instances (transfer ingest)
 - Serving segment manifests for transfer diffing
 - Relaying events from observers to local Callosum
 - Retrieving segment upload history for sync verification
@@ -186,7 +185,7 @@ def _serialize_observer(observer: dict[str, Any], current_now: int) -> dict[str,
     )
     key_prefix = observer_filename_prefix(observer)
     return {
-        "key_prefix": key_prefix,
+        "prefix": key_prefix,
         "name": observer.get("name", ""),
         "created_at": observer.get("created_at", 0),
         "last_seen": observer.get("last_seen"),
@@ -219,7 +218,7 @@ def api_list() -> Any:
             group_order[observer.get("group", "inactive")],
             1 if observer.get("last_seen") is None else 0,
             -(observer.get("last_seen") or 0),
-            observer.get("key_prefix", ""),
+            observer.get("prefix", ""),
         )
     )
 
@@ -253,10 +252,14 @@ def api_list() -> Any:
 # The feed does NOT filter events.
 # The feed does NOT redact fields (v1 trust call; same trust boundary as the existing
 # Convey SSE bridge — observers are inside it).
+# Keyless `/app/observer/callosum` is deferred until solstone-linux and
+# solstone-macos ship clients that no longer hardcode this keyed URL. Shipped
+# 0.3.0 / 1.3.x clients use `/app/observer/<key>/callosum`; the `<key>` segment
+# is retained transitionally and is no longer used for auth.
 @observer_bp.route(_OBSERVER_CALLOSUM_SSE_RULE, methods=["GET"])
 def callosum_sse(key: str) -> Any:
     """Stream Callosum events to an authenticated observer process."""
-    observer, key_prefix, error = resolve_observer_identity(key)
+    observer, key_prefix, error = resolve_observer_identity()
     if error is not None:
         return error
     auth_key = observer.get("key")
@@ -322,8 +325,7 @@ def api_create() -> Any:
     (apps/observer/workspace.html). Auto-registering observer clients use
     POST /app/observer/register instead, which takes a self-descriptor and
     locks a stream identity onto the record. This route is kept for the
-    human-facing management flow; it always mints and returns ``key_prefix``
-    (vs /register's ``prefix``).
+    human-facing management flow; it always mints and returns ``prefix``.
     """
     data = request.get_json(force=True) if request.is_json else {}
     name = data.get("name", "").strip()
@@ -367,7 +369,7 @@ def api_create() -> Any:
     return jsonify(
         {
             "key": key,
-            "key_prefix": key[:8],
+            "prefix": key[:8],
             "name": name,
             "ingest_url": ingest_url,
             "protocol_version": protocol.OBSERVER_PROTOCOL_VERSION,
@@ -613,10 +615,9 @@ def _save_to_failed(
 
 
 @observer_bp.route("/source/<stream>", methods=["DELETE"])
-@observer_bp.route("/source/<stream>/<key>", methods=["DELETE"])
-def delete_source(stream: str, key: str | None = None) -> Any:
+def delete_source(stream: str) -> Any:
     """Delete an allowed source stream for an authenticated observer."""
-    observer, key_prefix, error = resolve_observer_identity(key)
+    observer, key_prefix, error = resolve_observer_identity()
     if error is not None:
         return error
 
@@ -858,10 +859,11 @@ def _process_ingest_files(
 
 
 @observer_bp.route("/ingest", methods=["POST"])
-@observer_bp.route("/ingest/<key>", methods=["POST"])
-def ingest_upload(key: str | None = None) -> Any:
+def ingest_upload() -> Any:
     """Receive file uploads from observer.
 
+    Observer ingest is the live, single capture-segment stream from one observer.
+
     Expects multipart form with:
     - segment: Segment key (HHMMSS_LEN)
     - day: Day string (YYYYMMDD)
@@ -878,7 +880,7 @@ def ingest_upload(key: str | None = None) -> Any:
     - "duplicate": All files already received (no processing triggered)
     - "collision": New segment saved with adjusted key (directory conflict)
     """
-    observer, key_prefix, error = resolve_observer_identity(key)
+    observer, key_prefix, error = resolve_observer_identity()
     if error is not None:
         return error
 
@@ -993,86 +995,10 @@ def ingest_upload(key: str | None = None) -> Any:
     return jsonify(body), status
 
 
-@observer_bp.route("/ingest/<key>/transfer", methods=["POST"])
-def ingest_transfer(key: str) -> Any:
-    """Receive transferred file uploads from another solstone instance."""
-    observer, key_prefix, error = resolve_observer_identity(key)
-    if error is not None:
-        return error
-
-    segment = request.form.get("segment", "").strip()
-    day = request.form.get("day", "").strip()
-    stream = request.form.get("stream", "").strip()
-    host = request.form.get("host", "").strip()
-    platform_name = request.form.get("platform", "").strip()
-    meta_str = request.form.get("meta", "").strip()
-
-    meta: dict = {}
-    if meta_str:
-        try:
-            meta = json.loads(meta_str)
-        except json.JSONDecodeError:
-            logger.warning(f"Invalid meta JSON from observer: {meta_str[:100]}")
-    if host and "host" not in meta:
-        meta["host"] = host
-    if platform_name and "platform" not in meta:
-        meta["platform"] = platform_name
-
-    if not segment:
-        return error_response(MISSING_REQUIRED_FIELD, detail="Missing segment")
-    if not day:
-        return error_response(MISSING_REQUIRED_FIELD, detail="Missing day")
-    if not stream:
-        return error_response(MISSING_REQUIRED_FIELD, detail="Missing stream")
-    if not re.match(r"^\d{6}_\d+$", segment):
-        return error_response(
-            INVALID_SEGMENT_OR_STREAM,
-            detail="Invalid segment format",
-        )
-    if not re.match(r"^\d{8}$", day):
-        return error_response(INVALID_DAY, detail="Invalid day format")
-    if not re.match(r"^[a-z0-9][a-z0-9._-]*$", stream):
-        return error_response(
-            INVALID_SEGMENT_OR_STREAM,
-            detail="Invalid stream format",
-        )
-
-    files = request.files.getlist("files")
-    if not files:
-        return error_response(INGEST_NO_FILES, detail="No files uploaded")
-
-    body, status = _process_ingest_files(
-        observer,
-        key_prefix,
-        segment,
-        day,
-        stream,
-        files,
-        source="transfer",
-    )
-    if status != 200 or body.get("status") == "duplicate":
-        return jsonify(body), status
-
-    observer_name = observer.get("name", "")
-    event_fields: dict[str, Any] = {
-        "segment": body["segment"],
-        "day": day,
-        "files": body["files"],
-        "observer": observer_name,
-        "stream": stream,
-    }
-    if meta:
-        event_fields["meta"] = meta
-    emit("observe", "transferred", **event_fields)
-
-    return jsonify(body), status
-
-
 @observer_bp.route("/ingest/manifest", methods=["GET"])
-@observer_bp.route("/ingest/<key>/manifest", methods=["GET"])
-def ingest_manifest(key: str | None = None) -> Any:
+def ingest_manifest() -> Any:
     """List available manifest days for an observer."""
-    _observer, key_prefix, error = resolve_observer_identity(key)
+    _observer, key_prefix, error = resolve_observer_identity()
     if error is not None:
         return error
 
@@ -1094,10 +1020,9 @@ def ingest_manifest(key: str | None = None) -> Any:
 
 
 @observer_bp.route("/ingest/manifest/<day>", methods=["GET"])
-@observer_bp.route("/ingest/<key>/manifest/<day>", methods=["GET"])
-def ingest_manifest_day(day: str, key: str | None = None) -> Any:
+def ingest_manifest_day(day: str) -> Any:
     """Return a transfer manifest for all segments on a given day."""
-    _observer, _key_prefix, error = resolve_observer_identity(key)
+    _observer, _key_prefix, error = resolve_observer_identity()
     if error is not None:
         return error
 
@@ -1130,8 +1055,7 @@ def ingest_manifest_day(day: str, key: str | None = None) -> Any:
 
 
 @observer_bp.route("/ingest/event", methods=["POST"])
-@observer_bp.route("/ingest/<key>/event", methods=["POST"])
-def ingest_event(key: str | None = None) -> Any:
+def ingest_event() -> Any:
     """Receive events from observer and relay to local Callosum.
 
     Expects JSON body with:
@@ -1139,7 +1063,7 @@ def ingest_event(key: str | None = None) -> Any:
     - event: Event name
     - ...additional fields
     """
-    observer, _key_prefix, error = resolve_observer_identity(key)
+    observer, _key_prefix, error = resolve_observer_identity()
     if error is not None:
         return error
 
@@ -1199,8 +1123,7 @@ def _respond_observer_segments(items: list[dict], *, client_pv: int) -> Any:
 
 
 @observer_bp.route("/ingest/segments/<day>")
-@observer_bp.route("/ingest/<key>/segments/<day>")
-def ingest_segments(day: str, key: str | None = None) -> Any:
+def ingest_segments(day: str) -> Any:
     """List uploaded segments for a day with file verification.
 
     Returns JSON array of segments with file status:
@@ -1210,9 +1133,8 @@ def ingest_segments(day: str, key: str | None = None) -> Any:
 
     Args:
         day: Day string (YYYYMMDD)
-        key: Observer authentication key (from URL path, legacy)
     """
-    observer, key_prefix, error = resolve_observer_identity(key)
+    observer, key_prefix, error = resolve_observer_identity()
     if error is not None:
         return error
 
diff --git a/solstone/apps/observer/tests/test_callosum_sse.py b/solstone/apps/observer/tests/test_callosum_sse.py
index 47d805b19..4cb1390c8 100644
--- a/solstone/apps/observer/tests/test_callosum_sse.py
+++ b/solstone/apps/observer/tests/test_callosum_sse.py
@@ -56,7 +56,7 @@ def _create_observer(env, name: str = "sse-test") -> tuple[str, str]:
     )
     assert resp.status_code == 200
     data = resp.get_json()
-    return data["key"], data["key_prefix"]
+    return data["key"], data["prefix"]
 
 
 def _route_for(key: str) -> str:
@@ -103,11 +103,16 @@ def test_callosum_sse_missing_key_returns_401(observer_env):
     )
 
 
-def test_callosum_sse_unknown_key_returns_401(observer_env):
+def test_callosum_sse_path_key_without_bearer_returns_401(observer_env):
     env = observer_env()
-    resp = env.client.get(_route_for("unknown-key"), buffered=False)
+    key, _ = _create_observer(env)
+    resp = env.client.get(_route_for(key), buffered=False)
     assert resp.status_code == 401
-    _assert_reason(resp, reason_code="auth_key_invalid", detail="Invalid key")
+    _assert_reason(
+        resp,
+        reason_code="auth_required",
+        detail="Authorization required",
+    )
 
 
 def test_callosum_sse_revoked_key_returns_403(observer_env):
@@ -116,7 +121,11 @@ def test_callosum_sse_revoked_key_returns_403(observer_env):
     revoke = env.client.delete(f"/app/observer/api/{key_prefix}")
     assert revoke.status_code == 200
 
-    resp = env.client.get(_route_for(key), buffered=False)
+    resp = env.client.get(
+        _route_for("vestigial-path-key"),
+        headers={"Authorization": f"Bearer {key}"},
+        buffered=False,
+    )
     assert resp.status_code == 403
     _assert_reason(
         resp,
@@ -133,7 +142,11 @@ def test_callosum_sse_disabled_key_returns_403(observer_env):
     observer["enabled"] = False
     assert save_observer(observer)
 
-    resp = env.client.get(_route_for(key), buffered=False)
+    resp = env.client.get(
+        _route_for("vestigial-path-key"),
+        headers={"Authorization": f"Bearer {key}"},
+        buffered=False,
+    )
     assert resp.status_code == 403
     _assert_reason(
         resp,
@@ -142,7 +155,7 @@ def test_callosum_sse_disabled_key_returns_403(observer_env):
     )
 
 
-def test_callosum_sse_bearer_header_overrides_path_key(observer_env):
+def test_callosum_sse_bearer_header_authenticates(observer_env):
     env = observer_env()
     valid_key, _ = _create_observer(env, "valid-sse")
     bogus_key = "bogus-key"
@@ -171,7 +184,11 @@ def test_callosum_sse_success_content_type(observer_env):
     env = observer_env()
     key, _ = _create_observer(env)
 
-    resp = env.client.get(_route_for(key), buffered=False)
+    resp = env.client.get(
+        _route_for("vestigial-path-key"),
+        headers={"Authorization": f"Bearer {key}"},
+        buffered=False,
+    )
     try:
         assert resp.status_code == 200
         assert resp.content_type.startswith("text/event-stream")
@@ -182,7 +199,11 @@ def test_callosum_sse_success_content_type(observer_env):
 def test_callosum_sse_round_trip_payload(observer_env):
     env = observer_env()
     key, key_prefix = _create_observer(env)
-    resp = env.client.get(_route_for(key), buffered=False)
+    resp = env.client.get(
+        _route_for("vestigial-path-key"),
+        headers={"Authorization": f"Bearer {key}"},
+        buffered=False,
+    )
     try:
         assert resp.status_code == 200
         assert convey_bridge.subscription_count(key_prefix) == 1
@@ -268,7 +289,11 @@ def test_callosum_sse_heartbeat(observer_env, monkeypatch):
     key, _ = _create_observer(env)
     monkeypatch.setattr(routes_module, "_SSE_HEARTBEAT_SECONDS", 0.01)
 
-    resp = env.client.get(_route_for(key), buffered=False)
+    resp = env.client.get(
+        _route_for("vestigial-path-key"),
+        headers={"Authorization": f"Bearer {key}"},
+        buffered=False,
+    )
     try:
         assert resp.status_code == 200
         assert _next_chunk(resp) == ": heartbeat\n\n"
diff --git a/solstone/apps/observer/tests/test_resolve_identity.py b/solstone/apps/observer/tests/test_resolve_identity.py
index 6299cd89b..0285397dd 100644
--- a/solstone/apps/observer/tests/test_resolve_identity.py
+++ b/solstone/apps/observer/tests/test_resolve_identity.py
@@ -57,16 +57,14 @@ def test_resolve_dl_success_from_bearer(app_env):
     assert prefix == DL_KEY[:8]
 
 
-def test_resolve_dl_bearer_key_wins_over_url_key(app_env):
+def test_resolve_dl_uses_bearer_key(app_env):
     header_key = "headerkey123456789"
-    url_key = "urlkey123456789"
     save_observer({"key": header_key, "name": "header", "enabled": True, "stats": {}})
-    save_observer({"key": url_key, "name": "url", "enabled": True, "stats": {}})
 
     with app_env.test_request_context(
         headers={"Authorization": f"Bearer {header_key}"}
     ):
-        observer, prefix, error = resolve_observer_identity(url_key=url_key)
+        observer, prefix, error = resolve_observer_identity()
 
     assert error is None
     assert observer["name"] == "header"
@@ -125,7 +123,7 @@ def test_resolve_pl_success(app_env):
 
     with app_env.test_request_context():
         g.identity = _pl_identity(FINGERPRINT)
-        observer, prefix, error = resolve_observer_identity("ignored-route-key")
+        observer, prefix, error = resolve_observer_identity()
 
     assert error is None
     assert observer["name"] == "observer"
@@ -182,12 +180,12 @@ def test_resolve_pl_disabled(app_env):
     assert payload["reason_code"] == "feature_unavailable"
 
 
-def test_resolve_pl_does_not_require_url_key(app_env):
+def test_resolve_pl_does_not_require_bearer(app_env):
     mint_pl_observer_record(FINGERPRINT, "observer", "2026-04-20T00:00:00Z")
 
     with app_env.test_request_context("/app/observer/ingest/not-the-fingerprint/event"):
         g.identity = _pl_identity(FINGERPRINT)
-        observer, prefix, error = resolve_observer_identity("not-the-fingerprint")
+        observer, prefix, error = resolve_observer_identity()
 
     assert error is None
     assert prefix == "c" * 16
diff --git a/solstone/apps/observer/tests/test_routes.py b/solstone/apps/observer/tests/test_routes.py
index c76bff576..272a2eccc 100644
--- a/solstone/apps/observer/tests/test_routes.py
+++ b/solstone/apps/observer/tests/test_routes.py
@@ -9,8 +9,6 @@ import io
 import json
 from unittest.mock import MagicMock
 
-import requests
-
 import solstone.apps.observer.routes as routes_module
 import solstone.convey.bridge as convey_bridge
 from solstone.apps.observer.routes import (
@@ -259,7 +257,8 @@ def test_api_create_observer(observer_env):
 
     assert "key" in data
     assert len(data["key"]) > 32  # 256 bits = 43 base64 chars
-    assert data["key_prefix"] == data["key"][:8]
+    assert data["prefix"] == data["key"][:8]
+    assert "key_prefix" not in data
     assert data["name"] == "test-laptop"
     assert data["ingest_url"] == "/app/observer/ingest"
     assert data["key"] not in data["ingest_url"]
@@ -299,7 +298,7 @@ def test_api_list_shows_created_observer(observer_env):
         content_type="application/json",
     )
     assert resp.status_code == 200
-    key_prefix = resp.get_json()["key_prefix"]
+    key_prefix = resp.get_json()["prefix"]
 
     # List should show it
     payload = _api_list_payload(env)
@@ -307,7 +306,8 @@ def test_api_list_shows_created_observer(observer_env):
 
     assert len(observers) == 1
     assert payload["thresholds"] == {"active_ms": 30000, "stale_ms": 120000}
-    assert observers[0]["key_prefix"] == key_prefix
+    assert observers[0]["prefix"] == key_prefix
+    assert "key_prefix" not in observers[0]
     assert observers[0]["name"] == "my-observer"
     assert observers[0]["enabled"] is True
     assert observers[0]["stats"]["segments_received"] == 0
@@ -327,7 +327,7 @@ def test_api_list_includes_last_chat_request_at(observer_env):
         content_type="application/json",
     )
     assert resp.status_code == 200
-    key_prefix = resp.get_json()["key_prefix"]
+    key_prefix = resp.get_json()["prefix"]
     handle = convey_bridge.register_sse_subscriber(key_prefix)
     try:
         convey_bridge._broadcast_to_sse_clients(
@@ -339,7 +339,7 @@ def test_api_list_includes_last_chat_request_at(observer_env):
         with convey_bridge._SSE_LOCK:
             convey_bridge._SSE_LAST_CHAT_REQUEST_AT_BY_KEY.pop(key_prefix, None)
 
-    assert observers[0]["key_prefix"] == key_prefix
+    assert observers[0]["prefix"] == key_prefix
     assert observers[0]["last_chat_request_at"] == 9876
 
 
@@ -353,7 +353,7 @@ def test_api_delete_observer(observer_env):
         json={"name": "to-revoke"},
         content_type="application/json",
     )
-    key_prefix = resp.get_json()["key_prefix"]
+    key_prefix = resp.get_json()["prefix"]
 
     # Revoke it
     resp = env.client.delete(f"/app/observer/api/{key_prefix}")
@@ -363,7 +363,7 @@ def test_api_delete_observer(observer_env):
     # List should still show it, but marked as revoked
     observers = _api_list_observers(env)
     assert len(observers) == 1
-    assert observers[0]["key_prefix"] == key_prefix
+    assert observers[0]["prefix"] == key_prefix
     assert observers[0]["revoked"] is True
     assert observers[0]["revoked_at"] is not None
     assert observers[0]["state"] == "revoked"
@@ -409,7 +409,7 @@ def test_api_delete_dl_observer_does_not_touch_authorized_clients(observer_env):
         json={"name": "dl-delete"},
         content_type="application/json",
     )
-    key_prefix = resp.get_json()["key_prefix"]
+    key_prefix = resp.get_json()["prefix"]
     fingerprint = "sha256:" + ("e" * 64)
     AuthorizedClients(authorized_clients_path()).add(
         fingerprint,
@@ -495,8 +495,8 @@ def test_api_list_sorts_by_group_and_last_seen(observer_env, monkeypatch):
     ]
 
 
-def test_api_list_tie_breaks_by_key_prefix(observer_env, monkeypatch):
-    """Observers with the same last_seen sort by key_prefix ascending."""
+def test_api_list_tie_breaks_by_prefix(observer_env, monkeypatch):
+    """Observers with the same last_seen sort by prefix ascending."""
     env = observer_env()
     fixed_now = 3_000_000
     monkeypatch.setattr(routes_module, "now_ms", lambda: fixed_now)
@@ -515,10 +515,11 @@ def test_api_list_tie_breaks_by_key_prefix(observer_env, monkeypatch):
     )
 
     observers = _api_list_observers(env)
-    assert [observer["key_prefix"] for observer in observers] == [
+    assert [observer["prefix"] for observer in observers] == [
         "aaaa0000",
         "bbbb0000",
     ]
+    assert all("key_prefix" not in observer for observer in observers)
     assert all(observer["state"] == "connected" for observer in observers)
     assert all(observer["group"] == "active" for observer in observers)
     assert all(
@@ -636,13 +637,14 @@ def test_delete_source_hard_pin_rejects_other_stream(observer_env):
         content_type="application/json",
     )
     key = create_resp.get_json()["key"]
+    headers = {"Authorization": f"Bearer {key}"}
 
     other_seg = _plant_source_segment(
         env,
         stream="import.audio",
         segment="130000_300",
     )
-    resp = env.client.delete(f"/app/observer/source/import.audio/{key}")
+    resp = env.client.delete("/app/observer/source/import.audio", headers=headers)
     assert resp.status_code == 400
     assert resp.get_json()["detail"] == "Only known source streams can be deleted"
     assert other_seg.exists()
@@ -653,7 +655,8 @@ def test_delete_source_hard_pin_rejects_other_stream(observer_env):
         segment="140000_300",
     )
     resp = env.client.delete(
-        f"/app/observer/source/import.share/{key}",
+        "/app/observer/source/import.share",
+        headers=headers,
         data={"stream": "import.audio"},
     )
     assert resp.status_code == 400
@@ -661,7 +664,8 @@ def test_delete_source_hard_pin_rejects_other_stream(observer_env):
     assert share_seg.exists()
 
     resp = env.client.delete(
-        f"/app/observer/source/import.share/{key}",
+        "/app/observer/source/import.share",
+        headers=headers,
         data={"meta": json.dumps({"stream": "import.audio"})},
     )
     assert resp.status_code == 400
@@ -680,7 +684,10 @@ def test_delete_source_happy_path(observer_env):
     key = create_resp.get_json()["key"]
     seg_dir = _plant_source_segment(env)
 
-    resp = env.client.delete(f"/app/observer/source/import.share/{key}")
+    resp = env.client.delete(
+        "/app/observer/source/import.share",
+        headers={"Authorization": f"Bearer {key}"},
+    )
 
     assert resp.status_code == 200
     receipt = resp.get_json()
@@ -718,7 +725,10 @@ def test_delete_source_location_happy_path(observer_env):
     key = create_resp.get_json()["key"]
     seg_dir = _plant_location_segment(env)
 
-    resp = env.client.delete(f"/app/observer/source/location/{key}")
+    resp = env.client.delete(
+        "/app/observer/source/location",
+        headers={"Authorization": f"Bearer {key}"},
+    )
 
     assert resp.status_code == 200
     receipt = resp.get_json()
@@ -737,6 +747,7 @@ def test_delete_source_path_wins_over_candidate(observer_env):
         content_type="application/json",
     )
     key = create_resp.get_json()["key"]
+    headers = {"Authorization": f"Bearer {key}"}
     form_location_seg = _plant_location_segment(
         env,
         day="20250105",
@@ -749,7 +760,8 @@ def test_delete_source_path_wins_over_candidate(observer_env):
     )
 
     form_resp = env.client.delete(
-        f"/app/observer/source/location/{key}",
+        "/app/observer/source/location",
+        headers=headers,
         data={"stream": "import.share"},
     )
 
@@ -769,7 +781,8 @@ def test_delete_source_path_wins_over_candidate(observer_env):
     )
 
     meta_resp = env.client.delete(
-        f"/app/observer/source/location/{key}",
+        "/app/observer/source/location",
+        headers=headers,
         data={"meta": json.dumps({"stream": "import.share"})},
     )
 
@@ -1011,44 +1024,6 @@ def test_ingest_event_relay(observer_env):
     assert resp.get_json()["status"] == "ok"
 
 
-def test_ingest_event_pl_ignores_url_key(observer_env, monkeypatch):
-    env = observer_env()
-    other_key = _save_test_observer(
-        "deadbeef",
-        "other-dl",
-        created_at=100,
-        last_seen=None,
-    )
-    mint_pl_observer_record(
-        fingerprint=PL_FINGERPRINT,
-        device_label="pl-event",
-        paired_at="2026-05-20T00:00:00Z",
-    )
-    emitted: list[tuple[str, str, dict]] = []
-    monkeypatch.setattr(
-        routes_module,
-        "emit",
-        lambda tract, event, **kwargs: emitted.append((tract, event, kwargs)),
-    )
-
-    resp = env.client.post(
-        f"/app/observer/ingest/{other_key[:8]}/event",
-        environ_overrides={"pl.identity": _pl_identity()},
-        json={"tract": "observe", "event": "status", "mode": "screencast"},
-        content_type="application/json",
-    )
-
-    assert resp.status_code == 200
-    assert resp.get_json()["status"] == "ok"
-    assert emitted == [
-        (
-            "observe",
-            "status",
-            {"mode": "screencast", "observer": "pl-event"},
-        )
-    ]
-
-
 def test_dl_and_pl_observers_coexist_and_ingest(observer_env):
     env = observer_env()
     dl_resp = env.client.post(
@@ -1164,7 +1139,7 @@ def test_ingest_revoked_key(observer_env):
     )
     data = resp.get_json()
     key = data["key"]
-    key_prefix = data["key_prefix"]
+    key_prefix = data["prefix"]
 
     resp = env.client.delete(f"/app/observer/api/{key_prefix}")
     assert resp.status_code == 200
@@ -1195,7 +1170,7 @@ def test_keyless_ingest_bearer_rejects_revoked_and_disabled_keys(observer_env):
     revoked_data = resp.get_json()
     revoked_key = revoked_data["key"]
 
-    resp = env.client.delete(f"/app/observer/api/{revoked_data['key_prefix']}")
+    resp = env.client.delete(f"/app/observer/api/{revoked_data['prefix']}")
     assert resp.status_code == 200
 
     resp = env.client.post(
@@ -1263,7 +1238,7 @@ def test_ingest_event_revoked_key(observer_env):
     )
     data = resp.get_json()
     key = data["key"]
-    key_prefix = data["key_prefix"]
+    key_prefix = data["prefix"]
 
     resp = env.client.delete(f"/app/observer/api/{key_prefix}")
     assert resp.status_code == 200
@@ -1291,7 +1266,7 @@ def test_api_get_key(observer_env):
     )
     create_data = resp.get_json()
     key = create_data["key"]
-    key_prefix = create_data["key_prefix"]
+    key_prefix = create_data["prefix"]
 
     # Get the key
     resp = env.client.get(f"/app/observer/api/{key_prefix}/key")
@@ -1321,7 +1296,7 @@ def test_mint_responses_protocol_version_single_source_and_keyless_unconditional
     assert create_data["protocol_version"] == 99
     assert create_data["ingest_url"] == "/app/observer/ingest"
 
-    resp = env.client.get(f"/app/observer/api/{create_data['key_prefix']}/key")
+    resp = env.client.get(f"/app/observer/api/{create_data['prefix']}/key")
     assert resp.status_code == 200
     key_data = resp.get_json()
     assert key_data["protocol_version"] == 99
@@ -1347,7 +1322,7 @@ def test_api_get_key_revoked(observer_env):
         content_type="application/json",
     )
     create_data = resp.get_json()
-    key_prefix = create_data["key_prefix"]
+    key_prefix = create_data["prefix"]
 
     env.client.delete(f"/app/observer/api/{key_prefix}")
 
@@ -1369,7 +1344,7 @@ def test_api_get_key_audit_log(observer_env):
         content_type="application/json",
     )
     create_data = resp.get_json()
-    key_prefix = create_data["key_prefix"]
+    key_prefix = create_data["prefix"]
 
     with patch("solstone.apps.observer.routes.log_app_action") as mock_log:
         resp = env.client.get(f"/app/observer/api/{key_prefix}/key")
@@ -1607,7 +1582,7 @@ def test_ingest_creates_sync_history(observer_env):
     )
     data = resp.get_json()
     key = data["key"]
-    key_prefix = data["key_prefix"]
+    key_prefix = data["prefix"]
 
     # Upload a file
     test_data = b"test audio content for history"
@@ -1663,7 +1638,7 @@ def test_ingest_history_with_collision(observer_env):
     )
     data = resp.get_json()
     key = data["key"]
-    key_prefix = data["key_prefix"]
+    key_prefix = data["prefix"]
 
     # Create conflicting segment directory under the stream
     day_dir = _day_dir(env)
@@ -1812,41 +1787,6 @@ def test_segments_endpoint_lists_uploads(observer_env):
     )  # Original name preserved
 
 
-def test_legacy_url_key_ingest_and_segments_still_sync(observer_env):
-    env = observer_env()
-
-    resp = env.client.post(
-        "/app/observer/api/create",
-        json={"name": "legacy-url-key-test"},
-        content_type="application/json",
-    )
-    key = resp.get_json()["key"]
-
-    test_data = b"legacy url key content"
-    resp = env.client.post(
-        f"/app/observer/ingest/{key}",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "files": (io.BytesIO(test_data), "120000_300_audio.flac"),
-        },
-    )
-    assert resp.status_code == 200
-
-    resp = env.client.get(f"/app/observer/ingest/{key}/segments/20250103")
-    assert resp.status_code == 200
-    data = resp.get_json()
-
-    assert isinstance(data, list)
-    assert len(data) == 1
-    segment = data[0]
-    assert segment["key"] == "120000_300"
-    assert len(segment["files"]) == 1
-
-    file_info = segment["files"][0]
-    assert file_info["status"] == "present"
-
-
 def test_segments_endpoint_v2_empty(observer_env):
     """Test v2 segments endpoint returns collection envelope for no uploads."""
     env = observer_env()
@@ -1949,20 +1889,6 @@ def test_protocol_version_single_source(observer_env, monkeypatch):
     assert resp.status_code == 200
     assert isinstance(resp.get_json(), list)
 
-    from solstone.observe import transfer
-
-    session = MagicMock(spec=requests.Session)
-    response = MagicMock()
-    response.status_code = 200
-    response.json.return_value = []
-    session.get.return_value = response
-
-    transfer._query_remote_segments(session, "http://x", "20250103")
-
-    assert (
-        session.get.call_args.kwargs["headers"]["X-Solstone-Protocol-Version"] == "99"
-    )
-
 
 def test_segments_endpoint_shows_collision(observer_env):
     """Test segments endpoint shows collision info."""
@@ -2145,7 +2071,7 @@ def test_segments_endpoint_revoked_key(observer_env):
     )
     data = resp.get_json()
     key = data["key"]
-    key_prefix = data["key_prefix"]
+    key_prefix = data["prefix"]
 
     env.client.delete(f"/app/observer/api/{key_prefix}")
 
@@ -2228,7 +2154,7 @@ def test_segments_endpoint_shows_observed_status(observer_env):
     )
     data = resp.get_json()
     key = data["key"]
-    key_prefix = data["key_prefix"]
+    key_prefix = data["prefix"]
 
     # Upload a file
     test_data = b"test audio content"
@@ -2280,7 +2206,7 @@ def test_api_list_includes_segments_observed_stat(observer_env):
         content_type="application/json",
     )
     data = resp.get_json()
-    key_prefix = data["key_prefix"]
+    key_prefix = data["prefix"]
 
     # Initially no segments_observed
     data = _api_list_observers(env)
@@ -2523,7 +2449,7 @@ def test_ingest_partial_match_logged_in_history(observer_env):
     )
     data = resp.get_json()
     key = data["key"]
-    key_prefix = data["key_prefix"]
+    key_prefix = data["prefix"]
 
     audio_data = b"test audio for partial log"
 
@@ -2716,286 +2642,6 @@ def test_ingest_stream_qualifier_preserved(observer_env):
     assert not (_day_dir(env) / "fedora" / "120000_300" / "tmux.jsonl").exists()
 
 
-def test_transfer_success(observer_env):
-    """Test successful transfer upload."""
-    env = observer_env()
-
-    resp = env.client.post(
-        "/app/observer/api/create",
-        json={"name": "transfer-test"},
-        content_type="application/json",
-    )
-    key = resp.get_json()["key"]
-
-    test_data = b"transferred audio content"
-    resp = env.client.post(
-        f"/app/observer/ingest/{key}/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "stream": "remote.host",
-            "files": (io.BytesIO(test_data), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 200
-    data = resp.get_json()
-    assert data["status"] == "ok"
-    assert data["segment"] == "120000_300"
-    assert data["files"] == ["audio.flac"]
-    assert data["bytes"] == len(test_data)
-
-    expected_file = _day_dir(env) / "remote.host" / "120000_300" / "audio.flac"
-    assert expected_file.exists()
-    assert expected_file.read_bytes() == test_data
-
-
-def test_transfer_requires_stream(observer_env):
-    """Test that transfer requires stream."""
-    env = observer_env()
-
-    resp = env.client.post(
-        "/app/observer/api/create",
-        json={"name": "transfer-stream-test"},
-        content_type="application/json",
-    )
-    key = resp.get_json()["key"]
-
-    resp = env.client.post(
-        f"/app/observer/ingest/{key}/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "files": (io.BytesIO(b"content"), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 400
-    assert resp.get_json()["detail"] == "Missing stream"
-
-
-def test_transfer_invalid_stream(observer_env):
-    """Test that transfer validates stream format."""
-    env = observer_env()
-
-    resp = env.client.post(
-        "/app/observer/api/create",
-        json={"name": "transfer-invalid-stream"},
-        content_type="application/json",
-    )
-    key = resp.get_json()["key"]
-
-    resp = env.client.post(
-        f"/app/observer/ingest/{key}/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "stream": "INVALID!",
-            "files": (io.BytesIO(b"content"), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 400
-    assert resp.get_json()["detail"] == "Invalid stream format"
-
-
-def test_transfer_duplicate_detection(observer_env):
-    """Test transfer duplicate detection."""
-    env = observer_env()
-
-    resp = env.client.post(
-        "/app/observer/api/create",
-        json={"name": "transfer-duplicate-test"},
-        content_type="application/json",
-    )
-    key = resp.get_json()["key"]
-
-    test_data = b"duplicate transfer content"
-    resp = env.client.post(
-        f"/app/observer/ingest/{key}/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "stream": "remote.host",
-            "files": (io.BytesIO(test_data), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 200
-    assert resp.get_json()["status"] == "ok"
-
-    resp = env.client.post(
-        f"/app/observer/ingest/{key}/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "stream": "remote.host",
-            "files": (io.BytesIO(test_data), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 200
-    data = resp.get_json()
-    assert data["status"] == "duplicate"
-    assert data["existing_segment"] == "120000_300"
-
-
-def test_transfer_deconfliction(observer_env):
-    """Test transfer deconflicts existing segment directories."""
-    env = observer_env()
-
-    resp = env.client.post(
-        "/app/observer/api/create",
-        json={"name": "transfer-collision-test"},
-        content_type="application/json",
-    )
-    key = resp.get_json()["key"]
-
-    stream_dir = _day_dir(env) / "remote.host"
-    stream_dir.mkdir(parents=True)
-    (stream_dir / "120000_300").mkdir()
-
-    resp = env.client.post(
-        f"/app/observer/ingest/{key}/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "stream": "remote.host",
-            "files": (io.BytesIO(b"collision content"), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 200
-    data = resp.get_json()
-    assert data["status"] == "collision"
-    assert data["segment"] != "120000_300"
-    assert (stream_dir / data["segment"] / "audio.flac").exists()
-
-
-def test_transfer_emits_transferred_event(observer_env, monkeypatch):
-    """Test transfer emits observe.transferred."""
-    env = observer_env()
-
-    import solstone.apps.observer.routes as routes_module
-
-    calls = []
-
-    def mock_emit(*args, **kwargs):
-        calls.append((args, kwargs))
-
-    monkeypatch.setattr(routes_module, "emit", mock_emit)
-
-    resp = env.client.post(
-        "/app/observer/api/create",
-        json={"name": "transfer-event-test"},
-        content_type="application/json",
-    )
-    key = resp.get_json()["key"]
-
-    resp = env.client.post(
-        f"/app/observer/ingest/{key}/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "stream": "remote.host",
-            "files": (io.BytesIO(b"event content"), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 200
-    assert len(calls) == 1
-    assert calls[0][0] == ("observe", "transferred")
-
-
-def test_transfer_does_not_emit_observing(observer_env, monkeypatch):
-    """Test transfer does not emit observe.observing."""
-    env = observer_env()
-
-    import solstone.apps.observer.routes as routes_module
-
-    calls = []
-
-    def mock_emit(*args, **kwargs):
-        calls.append((args, kwargs))
-
-    monkeypatch.setattr(routes_module, "emit", mock_emit)
-
-    resp = env.client.post(
-        "/app/observer/api/create",
-        json={"name": "transfer-no-observing-test"},
-        content_type="application/json",
-    )
-    key = resp.get_json()["key"]
-
-    resp = env.client.post(
-        f"/app/observer/ingest/{key}/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "stream": "remote.host",
-            "files": (io.BytesIO(b"event content"), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 200
-    assert all(args[1] != "observing" for args, _kwargs in calls)
-
-
-def test_transfer_history_record(observer_env):
-    """Test transfer upload history records source='transfer'."""
-    from solstone.apps.observer.utils import load_history
-
-    env = observer_env()
-
-    resp = env.client.post(
-        "/app/observer/api/create",
-        json={"name": "transfer-history-test"},
-        content_type="application/json",
-    )
-    data = resp.get_json()
-    key = data["key"]
-    key_prefix = data["key_prefix"]
-
-    resp = env.client.post(
-        f"/app/observer/ingest/{key}/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "stream": "remote.host",
-            "files": (io.BytesIO(b"history content"), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 200
-
-    records = load_history(key_prefix, "20250103")
-    upload_record = next(record for record in records if not record.get("type"))
-    assert upload_record["source"] == "transfer"
-
-
-def test_transfer_auth_required(observer_env):
-    """Test transfer rejects invalid path key without auth header."""
-    env = observer_env()
-
-    resp = env.client.post(
-        "/app/observer/ingest/badkey/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "stream": "remote.host",
-            "files": (io.BytesIO(b"content"), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 401
-
-
-def test_transfer_invalid_key(observer_env):
-    """Test transfer rejects invalid key."""
-    env = observer_env()
-
-    resp = env.client.post(
-        "/app/observer/ingest/not-a-real-key/transfer",
-        data={
-            "day": "20250103",
-            "segment": "120000_300",
-            "stream": "remote.host",
-            "files": (io.BytesIO(b"content"), "audio.flac"),
-        },
-    )
-    assert resp.status_code == 401
-
-
 def test_manifest_day_listing(observer_env):
     """Test manifest day listing from observer history."""
     env = observer_env()
@@ -3018,7 +2664,10 @@ def test_manifest_day_listing(observer_env):
     )
     assert resp.status_code == 200
 
-    resp = env.client.get(f"/app/observer/ingest/{key}/manifest")
+    resp = env.client.get(
+        "/app/observer/ingest/manifest",
+        headers={"Authorization": f"Bearer {key}"},
+    )
     assert resp.status_code == 200
     assert resp.get_json() == {"days": {"20250103": {"segments": 1}}}
 
@@ -3035,11 +2684,12 @@ def test_manifest_per_day(observer_env):
     key = resp.get_json()["key"]
 
     resp = env.client.post(
-        f"/app/observer/ingest/{key}/transfer",
+        "/app/observer/ingest",
+        headers={"Authorization": f"Bearer {key}"},
         data={
             "day": "20250103",
             "segment": "120000_300",
-            "stream": "remote.host",
+            "meta": json.dumps({"stream": "remote.host"}),
             "files": [
                 (io.BytesIO(b"audio bytes"), "audio.flac"),
                 (io.BytesIO(b"screen bytes"), "screen.webm"),
@@ -3048,7 +2698,10 @@ def test_manifest_per_day(observer_env):
     )
     assert resp.status_code == 200
 
-    resp = env.client.get(f"/app/observer/ingest/{key}/manifest/20250103")
+    resp = env.client.get(
+        "/app/observer/ingest/manifest/20250103",
+        headers={"Authorization": f"Bearer {key}"},
+    )
     assert resp.status_code == 200
     data = resp.get_json()
 
@@ -3059,7 +2712,8 @@ def test_manifest_per_day(observer_env):
     assert "remote.host/120000_300" in data["segments"]
 
     files = data["segments"]["remote.host/120000_300"]["files"]
-    assert len(files) == 2
+    names = {file_info["name"] for file_info in files}
+    assert {"audio.flac", "screen.webm"}.issubset(names)
     for file_info in files:
         assert set(file_info) == {"name", "sha256", "size"}
         assert len(file_info["sha256"]) == 64
@@ -3069,5 +2723,5 @@ def test_manifest_auth_required(observer_env):
     """Test manifest endpoint rejects invalid key."""
     env = observer_env()
 
-    resp = env.client.get("/app/observer/ingest/badkey/manifest")
+    resp = env.client.get("/app/observer/ingest/manifest")
     assert resp.status_code == 401
diff --git a/solstone/apps/observer/utils.py b/solstone/apps/observer/utils.py
index c5030d26d..30a33fae2 100644
--- a/solstone/apps/observer/utils.py
+++ b/solstone/apps/observer/utils.py
@@ -361,7 +361,7 @@ def find_observer_by_name(name: str) -> dict | None:
     return ObserverRegistry.singleton().by_name(name)
 
 
-def _get_auth_key(url_key: str | None = None) -> str | None:
+def _get_auth_key() -> str | None:
     from flask import request
 
     auth = request.headers.get("Authorization", "")
@@ -369,7 +369,7 @@ def _get_auth_key(url_key: str | None = None) -> str | None:
         bearer = auth[7:].strip()
         if bearer:
             return bearer
-    return url_key or None
+    return None
 
 
 def _identity_fingerprint() -> str | None:
@@ -394,8 +394,8 @@ def _check_observer_enabled(observer: dict):
     return None
 
 
-# Observer still resolves through ObserverRegistry; minted pairings use header-Bearer auth, with legacy key-in-URL fallback deferring require_ingest_identity until URL-key auth is retired.
-def resolve_observer_identity(url_key: str | None = None):
+def resolve_observer_identity():
+    """Resolve an observer from PL identity first, then Authorization Bearer."""
     fingerprint = _identity_fingerprint()
     if fingerprint is not None:
         observer = load_observer_by_fingerprint(fingerprint)
@@ -406,7 +406,7 @@ def resolve_observer_identity(url_key: str | None = None):
             return None, None, error
         return observer, observer["filename_prefix"], None
 
-    auth_key = _get_auth_key(url_key)
+    auth_key = _get_auth_key()
     if not auth_key:
         return None, None, _auth_failure()
     observer = load_observer(auth_key)
diff --git a/solstone/apps/observer/workspace.html b/solstone/apps/observer/workspace.html
index 85a6a542a..432e679da 100644
--- a/solstone/apps/observer/workspace.html
+++ b/solstone/apps/observer/workspace.html
@@ -707,7 +707,7 @@ function observerCardHTML(observer, labels = {}) {
   const ariaLabel = [observer.name, label, liveLabel].filter(Boolean).join(', ');
 
   return `
-    <div class="observer-card ${cardClass}" data-key="${observer.key_prefix}" role="listitem" aria-label="${escapeHtml(ariaLabel)}">
+    <div class="observer-card ${cardClass}" data-key="${observer.prefix}" role="listitem" aria-label="${escapeHtml(ariaLabel)}">
       <div class="observer-header">
         <span class="observer-name">${escapeHtml(observer.name)}</span>
         <span class="observer-status ${statusClass}">${safeLabel}</span>
@@ -715,9 +715,9 @@ function observerCardHTML(observer, labels = {}) {
       </div>
       <div class="observer-stats">${statsHTML(observer, statusClass)}</div>
       <div class="observer-actions">
-        ${statusClass === 'stale' && !observer.revoked ? `<button onclick="restartObserver('${observer.key_prefix}', this)">${window.CONVEY_COPY?.ACTION_RECONNECT || 'Reconnect'}</button><span class="observer-reconnect-result" data-reconnect-result="${observer.key_prefix}"></span>` : ''}
-        ${observer.revoked ? '' : `<button onclick="viewObserverKey('${observer.key_prefix}', '${escapeHtml(observer.name)}')">view key</button>`}
-        ${observer.revoked ? '' : `<button class="danger" onclick="revokeObserver('${observer.key_prefix}', '${escapeHtml(observer.name)}')">revoke</button>`}
+        ${statusClass === 'stale' && !observer.revoked ? `<button onclick="restartObserver('${observer.prefix}', this)">${window.CONVEY_COPY?.ACTION_RECONNECT || 'Reconnect'}</button><span class="observer-reconnect-result" data-reconnect-result="${observer.prefix}"></span>` : ''}
+        ${observer.revoked ? '' : `<button onclick="viewObserverKey('${observer.prefix}', '${escapeHtml(observer.name)}')">view key</button>`}
+        ${observer.revoked ? '' : `<button class="danger" onclick="revokeObserver('${observer.prefix}', '${escapeHtml(observer.name)}')">revoke</button>`}
       </div>
     </div>
   `;
diff --git a/solstone/convey/root.py b/solstone/convey/root.py
index e594fbbd2..334ebafa1 100644
--- a/solstone/convey/root.py
+++ b/solstone/convey/root.py
@@ -110,7 +110,6 @@ def require_login() -> Any:
         "app:observer.ingest_upload",
         "app:observer.ingest_event",
         "app:observer.ingest_segments",
-        "app:observer.ingest_transfer",
         "app:observer.ingest_manifest",
         "app:observer.ingest_manifest_day",
         "app:observer.register",
diff --git a/solstone/observe/transfer.py b/solstone/observe/transfer.py
index 5c301b997..e4ea517b9 100644
--- a/solstone/observe/transfer.py
+++ b/solstone/observe/transfer.py
@@ -9,12 +9,9 @@ observation segments between solstone instances.
 Usage:
     journal transfer export --day YYYYMMDD [--output PATH]
     journal transfer import --archive PATH [--dry-run]
-    journal transfer send --to URL --key KEY [--day YYYYMMDD] [--dry-run]
     journal transfer send --to LABEL [--day YYYYMMDD] [--dry-run]
 
-On the RECEIVING host (the machine you are sending TO), run
-`journal observer create <name>` to generate an observer API key, then pass it
-as `--key`.
+Use `journal transfer send --to LABEL` to send segments to a paired peer.
 """
 
 from __future__ import annotations
@@ -33,9 +30,6 @@ from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any
 
-import requests
-
-from solstone.observe import protocol
 from solstone.observe.peer_lookup import PeerInfo, PeerLookupError, resolve_peer
 from solstone.observe.pl_http import PlHttpSession
 from solstone.think.callosum import callosum_send
@@ -56,16 +50,6 @@ from solstone.think.utils import (
 
 from .utils import compute_file_sha256, find_available_segment
 
-OBSERVER_KEY_HINT = (
-    "On the RECEIVING host (the machine you are sending TO), run "
-    "`journal observer create <name>` to generate an observer API key, then "
-    "pass it as `--key`."
-)
-
-AUTH_INVALID_OBSERVER_KEY = (
-    "Authentication failed: invalid or missing observer API key. " + OBSERVER_KEY_HINT
-)
-
 logger = logging.getLogger(__name__)
 
 # Archive manifest version
@@ -448,26 +432,12 @@ def _normalize_url(to: str) -> str:
     return f"https://{to}"
 
 
-def _is_url_destination(to: str) -> bool:
-    return to.startswith(("http://", "https://"))
-
-
 def _resolve_destination(
     parser: argparse.ArgumentParser,
     to: str,
-    key: str | None,
-) -> tuple[str, str | None, PeerInfo | None]:
-    if _is_url_destination(to):
-        if key is None:
-            parser.error("'--to <URL>' requires '--key <KEY>'")
-        return ("dl", _normalize_url(to), None)
-    if key is not None:
-        parser.error(
-            "'--key' is only valid with '--to <URL>'; "
-            "use '--to <label>' for pl peer transfers"
-        )
+) -> PeerInfo:
     try:
-        return ("pl", None, resolve_peer(to))
+        return resolve_peer(to)
     except PeerLookupError as exc:
         parser.error(str(exc))
 
@@ -510,160 +480,6 @@ def _parse_day_spec(day_spec: str | None, journal_root: Path) -> list[str]:
     raise ValueError("Invalid day format: use YYYYMMDD or YYYYMMDD-YYYYMMDD")
 
 
-def _parse_remote_segments(data: Any, day: str) -> dict[str, dict[str, str]]:
-    """Parse a segments 200 body (envelope or bare array) to {key: {name: sha}}.
-
-    Accepts both the v2 ``{"items": [...]}`` envelope and the legacy bare array,
-    yielding an identical map. Any unrecognized body degrades to ``{}`` with a
-    logged warning; never raises.
-    """
-    if isinstance(data, dict):
-        items = data.get("items")
-    elif isinstance(data, list):
-        items = data
-    else:
-        items = None
-    if not isinstance(items, list):
-        logger.warning(
-            f"Remote segment query for {day}: unrecognized response body, ignoring"
-        )
-        return {}
-    try:
-        return {
-            entry["key"]: {
-                file_info["name"]: file_info["sha256"]
-                for file_info in entry.get("files", [])
-            }
-            for entry in items
-            if isinstance(entry, dict) and entry.get("key")
-        }
-    except (KeyError, TypeError, AttributeError) as e:
-        logger.warning(
-            f"Remote segment query for {day}: malformed segment entry, ignoring: {e}"
-        )
-        return {}
-
-
-def _query_remote_segments(
-    session: requests.Session,
-    base_url: str,
-    day: str,
-) -> dict[str, dict[str, str]]:
-    """Query remote observer for existing segments on a day."""
-    url = f"{base_url}/app/observer/ingest/segments/{day}"
-    try:
-        response = session.get(
-            url,
-            headers={
-                protocol.OBSERVER_PROTOCOL_VERSION_HEADER: str(
-                    protocol.OBSERVER_PROTOCOL_VERSION
-                )
-            },
-            timeout=UPLOAD_TIMEOUT,
-        )
-        if response.status_code == 200:
-            try:
-                data = response.json()
-            except ValueError:
-                logger.warning(
-                    f"Remote segment query for {day}: non-JSON 200 body, ignoring"
-                )
-                return {}
-            return _parse_remote_segments(data, day)
-        if response.status_code == 401:
-            raise ValueError(AUTH_INVALID_OBSERVER_KEY)
-        if response.status_code == 403:
-            raise ValueError("Authentication failed: observer revoked or disabled")
-        logger.warning(
-            f"Remote segment query failed for {day}: "
-            f"{response.status_code} {response.text}"
-        )
-    except requests.RequestException as e:
-        logger.warning(f"Remote segment query failed for {day}: {e}")
-
-    return {}
-
-
-def _upload_segment(
-    session: requests.Session,
-    base_url: str,
-    day: str,
-    segment_key: str,
-    stream_name: str,
-    segment_path: Path,
-) -> tuple[str, int]:
-    """Upload a single segment to the remote observer."""
-    files = [
-        file_path
-        for file_path in sorted(segment_path.iterdir())
-        if file_path.is_file() and file_path.name != "stream.json"
-    ]
-    if not files:
-        return ("skip", 0)
-
-    url = f"{base_url}/app/observer/ingest"
-    data = {
-        "day": day,
-        "segment": segment_key,
-        "meta": json.dumps({"stream": stream_name}),
-    }
-
-    for attempt, delay in enumerate(RETRY_BACKOFF):
-        file_handles = []
-        files_data = []
-        try:
-            for file_path in files:
-                fh = open(file_path, "rb")
-                file_handles.append(fh)
-                files_data.append(
-                    ("files", (file_path.name, fh, "application/octet-stream"))
-                )
-
-            response = session.post(
-                url,
-                data=data,
-                files=files_data,
-                timeout=UPLOAD_TIMEOUT,
-            )
-            if response.status_code == 200:
-                status = response.json().get("status")
-                if status == "duplicate":
-                    return ("duplicate", 0)
-                return ("sent", response.json().get("bytes", 0))
-            if response.status_code == 401:
-                return ("auth_invalid", 0)
-            if response.status_code == 403:
-                return ("auth_revoked", 0)
-            if 500 <= response.status_code <= 599:
-                logger.warning(
-                    f"Upload attempt {attempt + 1} failed for "
-                    f"{day}/{stream_name}/{segment_key}: "
-                    f"{response.status_code} {response.text}"
-                )
-            else:
-                logger.warning(
-                    f"Upload rejected for {day}/{stream_name}/{segment_key}: "
-                    f"{response.status_code} {response.text}"
-                )
-                return ("error", 0)
-        except (requests.RequestException, OSError) as e:
-            logger.warning(
-                f"Upload attempt {attempt + 1} failed for "
-                f"{day}/{stream_name}/{segment_key}: {e}"
-            )
-        finally:
-            for fh in file_handles:
-                try:
-                    fh.close()
-                except Exception:
-                    pass
-
-        if attempt < len(RETRY_BACKOFF) - 1:
-            time.sleep(delay)
-
-    return ("error", 0)
-
-
 def _query_journal_segments(
     session: PlHttpSession,
     base_url: str,
@@ -779,107 +595,6 @@ def _upload_segment_journal(
     return ("error", 0)
 
 
-def send_segments(base_url: str, key: str, days: list[str], dry_run: bool) -> None:
-    """Send local journal segments to a remote observer."""
-    session = requests.Session()
-    session.headers["Authorization"] = f"Bearer {key}"
-
-    sent = 0
-    skipped = 0
-    failed = 0
-    bytes_total = 0
-    duplicates = 0
-
-    try:
-        for day in days:
-            day_dir = day_path(day, create=False)
-            if not day_dir.exists():
-                logger.debug(f"Day directory not found: {day}")
-                continue
-
-            segment_entries = iter_segments(day_dir)
-            if not segment_entries:
-                continue
-
-            try:
-                remote_manifest = _query_remote_segments(session, base_url, day)
-            except ValueError as e:
-                print(str(e))
-                return
-
-            for stream_name, seg_key, seg_path in segment_entries:
-                manifest = _build_segment_manifest(seg_path)
-                local_files = {
-                    file_info["name"]: file_info["sha256"]
-                    for file_info in manifest["files"]
-                    if file_info["name"] != "stream.json"
-                }
-                remote_files = remote_manifest.get(seg_key, {})
-                if all(
-                    remote_files.get(name) == sha256
-                    for name, sha256 in local_files.items()
-                ):
-                    logger.info(f"  [skip] {day}/{stream_name}/{seg_key}")
-                    skipped += 1
-                    continue
-
-                if dry_run:
-                    logger.info(f"  [would send] {day}/{stream_name}/{seg_key}")
-                    sent += 1
-                    continue
-
-                status, bytes_sent = _upload_segment(
-                    session,
-                    base_url,
-                    day,
-                    seg_key,
-                    stream_name,
-                    seg_path,
-                )
-                if status == "sent":
-                    logger.info(
-                        f"  [sent] {day}/{stream_name}/{seg_key} ({bytes_sent} bytes)"
-                    )
-                    sent += 1
-                    bytes_total += bytes_sent
-                elif status == "duplicate":
-                    logger.info(f"  [skip] {day}/{stream_name}/{seg_key}")
-                    skipped += 1
-                    duplicates += 1
-                elif status == "skip":
-                    logger.info(f"  [skip] {day}/{stream_name}/{seg_key}")
-                    skipped += 1
-                elif status == "auth_invalid":
-                    print(AUTH_INVALID_OBSERVER_KEY)
-                    return
-                elif status == "auth_revoked":
-                    print("Authentication failed: observer revoked or disabled")
-                    return
-                else:
-                    logger.info(f"  [FAILED] {day}/{stream_name}/{seg_key}")
-                    failed += 1
-    finally:
-        session.close()
-
-    total = sent + skipped + failed
-    if total == 0:
-        print("No segments found to transfer")
-        return
-
-    if dry_run:
-        print(f"\nDry run: would send {sent}, skip {skipped}")
-        return
-
-    print(
-        f"\nTransfer complete: {sent} sent, {skipped} skipped, "
-        f"{failed} failed, {bytes_total} bytes transferred"
-    )
-    if duplicates > 0:
-        print(f"  ({duplicates} duplicate segments already on remote)")
-    if sent == 0 and skipped > 0 and failed == 0:
-        print("Nothing to send - remote is up to date")
-
-
 def send_segments_pl(peer: PeerInfo, days: list[str], dry_run: bool) -> None:
     """Send local journal segments to a paired peer over PL."""
     identity = load_client_identity(peer.dir)
@@ -1019,22 +734,12 @@ def main() -> None:
     # Send subcommand
     send_parser = subparsers.add_parser(
         "send",
-        help="Send segments to remote observer",
-        description=OBSERVER_KEY_HINT,
+        help="Send segments to paired peer",
     )
     send_parser.add_argument(
         "--to",
         required=True,
-        help="Remote observer URL (http:// or https://) or paired peer label",
-    )
-    send_parser.add_argument(
-        "--key",
-        required=False,
-        default=None,
-        help=(
-            "Observer API key for URL mode (generate on the RECEIVING host with "
-            "`journal observer create <name>`)"
-        ),
+        help="Paired peer label",
     )
     send_parser.add_argument(
         "--day",
@@ -1081,15 +786,10 @@ def main() -> None:
             parser.error(str(e))
 
     elif args.command == "send":
-        mode, base_url, peer = _resolve_destination(send_parser, args.to, args.key)
+        peer = _resolve_destination(send_parser, args.to)
         journal = get_journal()
         try:
             days = _parse_day_spec(args.day, Path(journal))
         except ValueError as e:
             parser.error(str(e))
-        if mode == "dl":
-            assert base_url is not None and args.key is not None
-            send_segments(base_url, args.key, days, args.dry_run)
-        else:
-            assert peer is not None
-            send_segments_pl(peer, days, args.dry_run)
+        send_segments_pl(peer, days, args.dry_run)
diff --git a/solstone/think/link/auth.py b/solstone/think/link/auth.py
index 999191aed..5c10f6511 100644
--- a/solstone/think/link/auth.py
+++ b/solstone/think/link/auth.py
@@ -20,10 +20,15 @@ renameable label for the paired client. Solstone also stores local-only
             }
 
 Role-less linked systems are stored with `role: ""`; peers are stored with
-`role: "peer"`. Readers reload the file on mtime change so an unpair action
-takes effect within ~500 ms of the file write. Convey's pair and unpair routes
-own the pairing writer surface; the secure listener updates `last_seen_at` and
-uses this ledger for TLS verification and per-request authorization.
+`role: "peer"`. The peer role is provenance, not a behavioral authorization
+role: it denotes a linked system whose pairing provisioned a journal-content
+source, minted a journal-source record, and records the sender `instance_id`.
+That provenance is durable in data via per-segment `sender_instance_id` /
+`sender_fingerprint` and identity-derived source directories. Readers reload
+the file on mtime change so an unpair action takes effect within ~500 ms of the
+file write. Convey's pair and unpair routes own the pairing writer surface; the
+secure listener updates `last_seen_at` and uses this ledger for TLS verification
+and per-request authorization.
 
 `last_seen_at`, `network`, and `client_label` are local-only — never transmitted
 externally.
diff --git a/solstone/think/link/join_cli.py b/solstone/think/link/join_cli.py
index 574fd2bf8..b4c683a0e 100644
--- a/solstone/think/link/join_cli.py
+++ b/solstone/think/link/join_cli.py
@@ -20,7 +20,10 @@ Both layouts contain `private.pem`, `cert.pem`, `chain.pem`,
 `home_attestation.jwt`, and `peer.json`. `peer.json` fields are deterministic:
 `label`, `paired_at`, `instance_id`, `home_label`, `fingerprint`,
 `local_endpoints`, and `role`; role is `peer` or `""` for role-less linked
-systems.
+systems. `peer` is provenance, not a behavioral authorization role: pairing a
+peer provisions a journal-content source, records the sender `instance_id`, and
+leaves durable in-data provenance through per-segment `sender_instance_id` /
+`sender_fingerprint` and identity-derived source directories.
 """
 
 from __future__ import annotations
diff --git a/tests/baselines/api/observer/ingest-day.json b/tests/baselines/api/observer/ingest-day.json
index 66ba6f705..21712cb1f 100644
--- a/tests/baselines/api/observer/ingest-day.json
+++ b/tests/baselines/api/observer/ingest-day.json
@@ -1,5 +1,5 @@
 {
-  "detail": "Invalid key",
-  "error": "I couldn't verify that key.",
-  "reason_code": "auth_key_invalid"
+  "detail": "Authorization required",
+  "error": "I couldn't verify this request.",
+  "reason_code": "auth_required"
 }
diff --git a/tests/link/test_certless_structural_gate.py b/tests/link/test_certless_structural_gate.py
index 178352400..3c63cd1d0 100644
--- a/tests/link/test_certless_structural_gate.py
+++ b/tests/link/test_certless_structural_gate.py
@@ -53,7 +53,6 @@ def test_named_non_pair_endpoints_are_refused(
         "app:observer.ingest_upload": "POST",
         "app:observer.ingest_event": "POST",
         "app:observer.ingest_segments": "GET",
-        "app:observer.ingest_transfer": "POST",
         "app:observer.ingest_manifest": "GET",
         "app:observer.ingest_manifest_day": "GET",
         "app:import.journal_source_manifest": "GET",
diff --git a/tests/test_init.py b/tests/test_init.py
index 76353a7d6..55ecf9904 100644
--- a/tests/test_init.py
+++ b/tests/test_init.py
@@ -472,7 +472,7 @@ class TestInitObservers:
         observers = data["observers"]
         assert len(observers) == 1
         assert observers[0]["name"] == "my-phone"
-        assert observers[0]["key_prefix"] == "abcd1234"
+        assert observers[0]["prefix"] == "abcd1234"
         assert observers[0]["state"] == "disconnected"
         assert observers[0]["group"] == "inactive"
         assert observers[0]["label"] == "Disconnected"
@@ -509,12 +509,12 @@ class TestInitObservers:
         assert init_resp.status_code == 200
 
         api_by_key = {
-            observer["key_prefix"]: observer
+            observer["prefix"]: observer
             for observer in api_resp.get_json()["observers"]
             if not observer["revoked"]
         }
         init_by_key = {
-            observer["key_prefix"]: observer
+            observer["prefix"]: observer
             for observer in init_resp.get_json()["observers"]
         }
 
diff --git a/tests/test_transfer.py b/tests/test_transfer.py
index 9a52a822c..84a0231eb 100644
--- a/tests/test_transfer.py
+++ b/tests/test_transfer.py
@@ -5,14 +5,11 @@
 
 import inspect
 import json
-import logging
 import tarfile
 from pathlib import Path
-from typing import Any
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 import pytest
-import requests
 
 
 class TestSegmentDeconfliction:
@@ -993,90 +990,6 @@ class TestTransferSend:
 
         think_utils._journal_path_cache = None
 
-    def _make_session(
-        self,
-        *,
-        get_status: int = 200,
-        get_json: Any | None = None,
-        post_status: int = 200,
-        post_json: dict | None = None,
-    ) -> MagicMock:
-        mock_session = MagicMock(spec=requests.Session)
-        mock_session.headers = {}
-
-        get_response = MagicMock()
-        get_response.status_code = get_status
-        get_response.json.return_value = get_json if get_json is not None else []
-        get_response.text = "GET error"
-        mock_session.get.return_value = get_response
-
-        post_response = MagicMock()
-        post_response.status_code = post_status
-        post_response.json.return_value = (
-            post_json if post_json is not None else {"status": "ok", "bytes": 100}
-        )
-        post_response.text = "POST error"
-        mock_session.post.return_value = post_response
-
-        return mock_session
-
-    def test_query_remote_segments_parses_envelope_and_array_identically(self):
-        from solstone.observe.transfer import _query_remote_segments
-
-        segment_data = [{"key": "s1", "files": [{"name": "a.flac", "sha256": "abc"}]}]
-        envelope_session = self._make_session(
-            get_json={
-                "items": segment_data,
-                "total": 1,
-                "protocol_version": 2,
-            }
-        )
-        array_session = self._make_session(get_json=segment_data)
-
-        envelope_result = _query_remote_segments(
-            envelope_session,
-            "https://example.com",
-            "20250103",
-        )
-        array_result = _query_remote_segments(
-            array_session,
-            "https://example.com",
-            "20250103",
-        )
-
-        assert envelope_result == {"s1": {"a.flac": "abc"}}
-        assert array_result == envelope_result
-        assert envelope_session.get.call_args.kwargs["headers"] == {
-            "X-Solstone-Protocol-Version": "2"
-        }
-
-    @pytest.mark.parametrize(
-        "body",
-        [
-            {"error": "nope"},
-            {"items": 5},
-            "oops",
-        ],
-    )
-    def test_query_remote_segments_degrades_unrecognized_bodies(
-        self,
-        body,
-        caplog,
-    ):
-        from solstone.observe.transfer import _query_remote_segments
-
-        mock_session = self._make_session(get_json=body)
-
-        with caplog.at_level(logging.WARNING):
-            result = _query_remote_segments(
-                mock_session,
-                "https://example.com",
-                "20250103",
-            )
-
-        assert result == {}
-        assert caplog.records
-
     def test_parse_day_spec_single(self, tmp_path):
         from solstone.observe.transfer import _parse_day_spec
 
@@ -1125,206 +1038,3 @@ class TestTransferSend:
         assert _normalize_url("example.com/") == "https://example.com"
         assert _normalize_url("https://example.com/") == "https://example.com"
         assert _normalize_url("http://example.com/api/") == "http://example.com/api"
-
-    def test_send_dry_run(self, tmp_path, monkeypatch, capsys):
-        from solstone.observe.transfer import send_segments
-
-        journal = self._setup_journal(tmp_path)
-        self._set_journal_override(monkeypatch, journal)
-
-        mock_session = self._make_session(get_json=[])
-
-        with patch(
-            "solstone.observe.transfer.requests.Session", return_value=mock_session
-        ):
-            send_segments("https://example.com", "test-key", ["20250103"], dry_run=True)
-
-        assert mock_session.get.call_count == 1
-        assert mock_session.get.call_args.args[0].endswith(
-            "/app/observer/ingest/segments/20250103"
-        )
-        assert mock_session.post.call_count == 0
-        assert "Dry run: would send 1, skip 0" in capsys.readouterr().out
-
-    def test_send_skips_matching(self, tmp_path, monkeypatch, capsys):
-        from solstone.observe.transfer import send_segments
-        from solstone.observe.utils import compute_file_sha256
-
-        journal = self._setup_journal(tmp_path)
-        self._set_journal_override(monkeypatch, journal)
-
-        segment_dir = journal / "chronicle" / "20250103" / "default" / "120000_300"
-        remote_files = [
-            {
-                "name": "audio.flac",
-                "sha256": compute_file_sha256(segment_dir / "audio.flac"),
-            },
-            {
-                "name": "transcript.jsonl",
-                "sha256": compute_file_sha256(segment_dir / "transcript.jsonl"),
-            },
-        ]
-        mock_session = self._make_session(
-            get_json=[{"key": "120000_300", "observed": False, "files": remote_files}]
-        )
-
-        with patch(
-            "solstone.observe.transfer.requests.Session", return_value=mock_session
-        ):
-            send_segments(
-                "https://example.com", "test-key", ["20250103"], dry_run=False
-            )
-
-        assert mock_session.post.call_count == 0
-        output = capsys.readouterr().out
-        assert "Transfer complete: 0 sent, 1 skipped, 0 failed" in output
-        assert "Nothing to send - remote is up to date" in output
-
-    def test_send_uploads_new(self, tmp_path, monkeypatch, capsys):
-        from solstone.observe.transfer import send_segments
-
-        journal = self._setup_journal(tmp_path)
-        self._set_journal_override(monkeypatch, journal)
-
-        mock_session = self._make_session(
-            get_json=[],
-            post_json={"status": "ok", "bytes": 100},
-        )
-
-        with patch(
-            "solstone.observe.transfer.requests.Session", return_value=mock_session
-        ):
-            send_segments(
-                "https://example.com", "test-key", ["20250103"], dry_run=False
-            )
-
-        assert mock_session.post.call_count == 1
-        post_kwargs = mock_session.post.call_args.kwargs
-        assert post_kwargs["data"]["day"] == "20250103"
-        assert post_kwargs["data"]["segment"] == "120000_300"
-        assert json.loads(post_kwargs["data"]["meta"]) == {"stream": "default"}
-        # Auth is set on the session, not per-request
-        assert mock_session.headers["Authorization"] == "Bearer test-key"
-        assert (
-            "Transfer complete: 1 sent, 0 skipped, 0 failed, 100 bytes transferred"
-            in (capsys.readouterr().out)
-        )
-
-    def test_send_retry_on_5xx(self, tmp_path, monkeypatch, capsys):
-        from solstone.observe.transfer import send_segments
-
-        journal = self._setup_journal(tmp_path)
-        self._set_journal_override(monkeypatch, journal)
-
-        mock_session = self._make_session(get_json=[])
-        first = MagicMock(status_code=500, text="server error")
-        second = MagicMock(status_code=500, text="server error")
-        success = MagicMock(status_code=200)
-        success.json.return_value = {"status": "ok", "bytes": 100}
-        mock_session.post.side_effect = [first, second, success]
-
-        with (
-            patch(
-                "solstone.observe.transfer.requests.Session", return_value=mock_session
-            ),
-            patch("solstone.observe.transfer.time.sleep"),
-        ):
-            send_segments(
-                "https://example.com", "test-key", ["20250103"], dry_run=False
-            )
-
-        assert mock_session.post.call_count == 3
-        assert (
-            "Transfer complete: 1 sent, 0 skipped, 0 failed, 100 bytes transferred"
-            in (capsys.readouterr().out)
-        )
-
-    @pytest.mark.parametrize(
-        ("status", "message"),
-        [
-            (401, "Authentication failed"),
-            (403, "revoked or disabled"),
-        ],
-    )
-    def test_send_auth_error(self, status, message):
-        from solstone.observe.transfer import _query_remote_segments
-
-        mock_session = self._make_session(get_status=status)
-
-        with pytest.raises(ValueError, match=message):
-            _query_remote_segments(
-                mock_session,
-                "https://example.com",
-                "20250103",
-            )
-
-    def test_send_idempotent(self, tmp_path, monkeypatch, capsys):
-        from solstone.observe.transfer import send_segments
-        from solstone.observe.utils import compute_file_sha256
-
-        journal = self._setup_journal(tmp_path)
-        self._set_journal_override(monkeypatch, journal)
-
-        segment_dir = journal / "chronicle" / "20250103" / "default" / "120000_300"
-        remote_files = [
-            {
-                "name": "audio.flac",
-                "sha256": compute_file_sha256(segment_dir / "audio.flac"),
-            },
-            {
-                "name": "transcript.jsonl",
-                "sha256": compute_file_sha256(segment_dir / "transcript.jsonl"),
-            },
-        ]
-        first_get = MagicMock(status_code=200)
-        first_get.json.return_value = []
-        first_get.text = "GET error"
-        second_get = MagicMock(status_code=200)
-        second_get.json.return_value = [
-            {"key": "120000_300", "observed": False, "files": remote_files}
-        ]
-        second_get.text = "GET error"
-
-        mock_session = self._make_session()
-        mock_session.get.side_effect = [first_get, second_get]
-
-        with patch(
-            "solstone.observe.transfer.requests.Session", return_value=mock_session
-        ):
-            send_segments(
-                "https://example.com", "test-key", ["20250103"], dry_run=False
-            )
-            send_segments(
-                "https://example.com", "test-key", ["20250103"], dry_run=False
-            )
-
-        assert mock_session.post.call_count == 1
-        output = capsys.readouterr().out
-        assert (
-            "Transfer complete: 1 sent, 0 skipped, 0 failed, 100 bytes transferred"
-            in (output)
-        )
-        assert (
-            "Transfer complete: 0 sent, 1 skipped, 0 failed, 0 bytes transferred"
-            in output
-        )
-
-    def test_send_excludes_stream_json(self, tmp_path, monkeypatch):
-        from solstone.observe.transfer import send_segments
-
-        journal = self._setup_journal(tmp_path, include_stream_json=True)
-        self._set_journal_override(monkeypatch, journal)
-
-        mock_session = self._make_session(get_json=[])
-
-        with patch(
-            "solstone.observe.transfer.requests.Session", return_value=mock_session
-        ):
-            send_segments(
-                "https://example.com", "test-key", ["20250103"], dry_run=False
-            )
-
-        files_arg = mock_session.post.call_args.kwargs["files"]
-        uploaded_names = [entry[1][0] for entry in files_arg]
-        assert "stream.json" not in uploaded_names
-        assert uploaded_names == ["audio.flac", "transcript.jsonl"]
diff --git a/tests/test_transfer_pl.py b/tests/test_transfer_pl.py
index caaff824e..bb9a7024e 100644
--- a/tests/test_transfer_pl.py
+++ b/tests/test_transfer_pl.py
@@ -6,10 +6,8 @@ from __future__ import annotations
 import json
 import sys
 from pathlib import Path
-from unittest.mock import MagicMock, patch
 
 import pytest
-import requests
 from cryptography.hazmat.primitives import serialization
 
 import solstone.think.utils as think_utils
@@ -93,75 +91,19 @@ def test_transfer_send_pl_posts_journal_segment_day_path(
     assert b"audio" in post[3]
 
 
-def test_transfer_send_dl_regression_url_headers_and_body(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    from solstone.observe.transfer import send_segments
-
-    journal = tmp_path / "journal"
-    _set_journal(monkeypatch, journal)
-    _write_segment(journal)
-
-    mock_session = MagicMock(spec=requests.Session)
-    mock_session.headers = {}
-    get_response = MagicMock()
-    get_response.status_code = 200
-    get_response.json.return_value = {}
-    mock_session.get.return_value = get_response
-    post_response = MagicMock()
-    post_response.status_code = 200
-    post_response.json.return_value = {"bytes": 15}
-    mock_session.post.return_value = post_response
-
-    with patch("solstone.observe.transfer.requests.Session", return_value=mock_session):
-        send_segments("https://receiver.test", "test-key", ["20260520"], False)
-
-    assert mock_session.headers["Authorization"] == "Bearer test-key"
-    assert mock_session.get.call_args.args[0] == (
-        "https://receiver.test/app/observer/ingest/segments/20260520"
-    )
-    assert mock_session.post.call_args.args[0] == (
-        "https://receiver.test/app/observer/ingest"
-    )
-    post_kwargs = mock_session.post.call_args.kwargs
-    assert post_kwargs["data"] == {
-        "day": "20260520",
-        "segment": "143022_300",
-        "meta": json.dumps({"stream": "laptop"}),
-    }
-    assert [entry[1][0] for entry in post_kwargs["files"]] == [
-        "audio.flac",
-        "transcript.jsonl",
-    ]
-
-
-@pytest.mark.parametrize(
-    ("argv", "message"),
-    [
-        (
-            ["sol", "send", "--to", "host-a", "--key", "x"],
-            "'--key' is only valid with '--to <URL>'",
-        ),
-        (
-            ["sol", "send", "--to", "https://receiver.test"],
-            "'--to <URL>' requires '--key <KEY>'",
-        ),
-    ],
-)
 def test_transfer_send_destination_validation(
-    argv: list[str],
-    message: str,
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
     capsys: pytest.CaptureFixture[str],
 ) -> None:
     from solstone.observe import transfer
 
-    _set_journal(monkeypatch, tmp_path / "journal")
-    monkeypatch.setattr(sys, "argv", argv)
+    journal = tmp_path / "journal"
+    (journal / "peers").mkdir(parents=True)
+    _set_journal(monkeypatch, journal)
+    monkeypatch.setattr(sys, "argv", ["sol", "send", "--to", "https://receiver.test"])
 
     with pytest.raises(SystemExit):
         transfer.main()
 
-    assert message in capsys.readouterr().err
+    assert 'no peer with label "https://receiver.test"' in capsys.readouterr().err
diff --git a/tests/verify_api.py b/tests/verify_api.py
index 11ff8082b..0c6e0e690 100644
--- a/tests/verify_api.py
+++ b/tests/verify_api.py
@@ -187,7 +187,7 @@ ENDPOINTS = [
     {
         "app": "observer",
         "name": "ingest-day",
-        "path": "/app/observer/ingest/example-key/segments/20260304",
+        "path": "/app/observer/ingest/segments/20260304",
         "params": {},
         "status": 401,
     },