diff --git a/solstone/observe/export.py b/solstone/observe/export.py
index 7e52541c9..830ea6ec6 100644
--- a/solstone/observe/export.py
+++ b/solstone/observe/export.py
@@ -4,7 +4,8 @@
 """Export journal data to a remote solstone instance.
 
 Usage:
-    sol export --to HOST --key KEY [--only TYPE] [--dry-run] [--day YYYYMMDD]
+    sol export --to URL --key KEY [--only TYPE] [--dry-run] [--day YYYYMMDD]
+    sol export --to LABEL [--only TYPE] [--dry-run] [--day YYYYMMDD]
 """
 
 from __future__ import annotations
@@ -22,6 +23,9 @@ from typing import Any
 
 import requests
 
+from solstone.observe.peer_lookup import PeerInfo, PeerLookupError, resolve_peer
+from solstone.observe.peer_unpair import maybe_prompt_unpair
+from solstone.observe.pl_http import PlHttpSession
 from solstone.observe.transfer import (
     RETRY_BACKOFF,
     _build_segment_manifest,
@@ -30,6 +34,9 @@ from solstone.observe.transfer import (
 )
 from solstone.think.entities.journal import load_all_journal_entities
 from solstone.think.importers.sync import SYNCABLE_REGISTRY
+from solstone.think.link.bundle import load_client_identity
+from solstone.think.link.dialer import TunnelClient
+from solstone.think.link.paths import relay_url
 from solstone.think.utils import (
     day_path,
     get_config,
@@ -47,6 +54,8 @@ _DAY_JSONL_RE = re.compile(r"^\d{8}\.jsonl$")
 _DAY_MD_RE = re.compile(r"^\d{8}\.md$")
 _IMPORT_ID_RE = re.compile(r"^\d{8}_\d{6}$")
 _NEVER_TRANSFER_PATHS = frozenset({"convey.password_hash", "convey.secret"})
+EXPORT_AREAS = ("segments", "imports", "entities", "facets", "config")
+FULL_EXPORT_SET = frozenset(EXPORT_AREAS)
 
 
 @dataclass
@@ -62,6 +71,45 @@ class ExportResult:
     error: str | None = None
 
 
+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"
+        )
+    try:
+        return ("pl", None, resolve_peer(to))
+    except PeerLookupError as exc:
+        parser.error(str(exc))
+
+
+def _parse_only_set(
+    parser: argparse.ArgumentParser,
+    raw: str | None,
+) -> frozenset[str]:
+    if raw is None:
+        return FULL_EXPORT_SET
+    areas = frozenset(part.strip() for part in raw.split(",") if part.strip())
+    invalid = sorted(areas - FULL_EXPORT_SET)
+    if invalid or not areas:
+        parser.error(
+            "--only must contain one or more of: " + ", ".join(sorted(FULL_EXPORT_SET))
+        )
+    return areas
+
+
 def _query_manifest(
     session: requests.Session, base_url: str, key: str, area: str = "segments"
 ) -> dict[str, Any]:
@@ -1007,118 +1055,71 @@ def export_config(
             session.close()
 
 
-def main() -> None:
-    parser = argparse.ArgumentParser(
-        description="Export journal data to a remote solstone instance"
-    )
-    parser.add_argument(
-        "--to",
-        required=True,
-        help="Remote instance URL (e.g., host:port or https://host)",
-    )
-    parser.add_argument(
-        "--key",
-        required=True,
-        help="API key for the remote journal source",
-    )
-    parser.add_argument(
-        "--only",
-        default=None,
-        help="Export only specific area (segments, entities, facets, imports, config)",
-    )
-    parser.add_argument(
-        "--dry-run",
-        action="store_true",
-        help="Show what would be exported without sending",
-    )
-    parser.add_argument(
-        "--day",
-        default=None,
-        help="Day or range (YYYYMMDD or YYYYMMDD-YYYYMMDD)",
-    )
-    args = setup_cli(parser)
-
-    base_url = _normalize_url(args.to)
-
-    if args.only == "entities":
-        export_entities(base_url, args.key, args.dry_run)
-        return
-    if args.only == "facets":
-        export_facets(base_url, args.key, args.dry_run)
-        return
-    if args.only == "imports":
-        export_imports(base_url, args.key, args.dry_run)
-        return
-    if args.only == "config":
-        export_config(base_url, args.key, args.dry_run)
-        return
-    if args.only == "segments":
-        days = _parse_day_spec(args.day, Path(get_journal()))
-        export_segments(base_url, args.key, days, args.dry_run)
-        return
-    if args.only is not None:
-        print(f"Export of '{args.only}' is not yet implemented")
-        sys.exit(0)
-
-    # Full export: all areas in dependency order
-    days = _parse_day_spec(args.day, Path(get_journal()))
-    session = requests.Session()
-    session.headers["Authorization"] = f"Bearer {args.key}"
-
-    try:
-        _query_manifest(session, base_url, args.key)
-    except requests.ConnectionError:
-        print(f"Connection failed: could not reach {base_url}")
-        session.close()
-        sys.exit(1)
-    except ValueError as e:
-        print(str(e))
-        session.close()
-        sys.exit(1)
-
-    areas = [
-        (
-            "segments",
-            lambda: export_segments(
-                base_url, args.key, days, args.dry_run, session=session
-            ),
+def _run_export_areas(
+    base_url: str,
+    key: str,
+    days: list[str],
+    dry_run: bool,
+    session: requests.Session | PlHttpSession,
+    areas: frozenset[str],
+) -> list[ExportResult]:
+    exporters = {
+        "segments": lambda: export_segments(
+            base_url,
+            key,
+            days,
+            dry_run,
+            session=session,
         ),
-        (
-            "imports",
-            lambda: export_imports(base_url, args.key, args.dry_run, session=session),
+        "imports": lambda: export_imports(
+            base_url,
+            key,
+            dry_run,
+            session=session,
         ),
-        (
-            "entities",
-            lambda: export_entities(base_url, args.key, args.dry_run, session=session),
+        "entities": lambda: export_entities(
+            base_url,
+            key,
+            dry_run,
+            session=session,
         ),
-        (
-            "facets",
-            lambda: export_facets(base_url, args.key, args.dry_run, session=session),
+        "facets": lambda: export_facets(
+            base_url,
+            key,
+            dry_run,
+            session=session,
         ),
-        (
-            "config",
-            lambda: export_config(base_url, args.key, args.dry_run, session=session),
+        "config": lambda: export_config(
+            base_url,
+            key,
+            dry_run,
+            session=session,
         ),
-    ]
+    }
 
     results: list[ExportResult] = []
-    for area_name, export_fn in areas:
+    for area_name in EXPORT_AREAS:
+        if area_name not in areas:
+            continue
         try:
-            area_result = export_fn()
-            results.append(area_result)
+            results.append(exporters[area_name]())
         except Exception:
             logger.exception("Export failed for %s", area_name)
-            error_result = ExportResult(
-                area=area_name, error=f"Exception during {area_name} export"
+            results.append(
+                ExportResult(
+                    area=area_name,
+                    error=f"Exception during {area_name} export",
+                )
             )
-            results.append(error_result)
             if area_name == "entities":
                 print(
-                    "  Warning: entity export failed — facet entity mapping may be incomplete"
+                    "  Warning: entity export failed - "
+                    "facet entity mapping may be incomplete"
                 )
+    return results
 
-    session.close()
 
+def _print_export_summary(results: list[ExportResult]) -> bool:
     print("\n--- Export Summary ---")
     any_failed = False
     for area_result in results:
@@ -1142,6 +1143,85 @@ def main() -> None:
         if not parts:
             parts.append("nothing to send")
         print(f"  {area_result.area}: {', '.join(parts)}")
+    return any_failed
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Export journal data to a remote solstone instance"
+    )
+    parser.add_argument(
+        "--to",
+        required=True,
+        help="Remote URL (http:// or https://) or paired peer label",
+    )
+    parser.add_argument(
+        "--key",
+        required=False,
+        default=None,
+        help="API key for URL mode",
+    )
+    parser.add_argument(
+        "--only",
+        default=None,
+        help="Export only specific area (segments, entities, facets, imports, config)",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Show what would be exported without sending",
+    )
+    parser.add_argument(
+        "--day",
+        default=None,
+        help="Day or range (YYYYMMDD or YYYYMMDD-YYYYMMDD)",
+    )
+    args = setup_cli(parser)
+
+    areas = _parse_only_set(parser, args.only)
+    mode, base_url, peer = _resolve_destination(parser, args.to, args.key)
+    try:
+        days = _parse_day_spec(args.day, Path(get_journal()))
+    except ValueError as exc:
+        parser.error(str(exc))
+
+    if mode == "dl":
+        assert base_url is not None and args.key is not None
+        session = requests.Session()
+        session.headers["Authorization"] = f"Bearer {args.key}"
+        try:
+            results = _run_export_areas(
+                base_url,
+                args.key,
+                days,
+                args.dry_run,
+                session,
+                areas,
+            )
+        finally:
+            session.close()
+        any_failed = _print_export_summary(results)
+        if any_failed:
+            sys.exit(1)
+        return
+
+    assert peer is not None
+    identity = load_client_identity(peer.dir)
+    with TunnelClient(identity, relay_url=relay_url()) as tunnel:
+        session = PlHttpSession(tunnel)
+        results = _run_export_areas(
+            "https://pl.peer",
+            peer.instance_id,
+            days,
+            args.dry_run,
+            session,
+            areas,
+        )
+        any_failed = _print_export_summary(results)
+        # Prompt fires after any successful full migration: explicit --only
+        # and the no-flag default both execute the full area set.
+        if not args.dry_run and areas == FULL_EXPORT_SET and not any_failed:
+            maybe_prompt_unpair(peer, session)
 
     if any_failed:
         sys.exit(1)
diff --git a/solstone/observe/peer_lookup.py b/solstone/observe/peer_lookup.py
new file mode 100644
index 000000000..1f3879fdd
--- /dev/null
+++ b/solstone/observe/peer_lookup.py
@@ -0,0 +1,87 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+
+from solstone.think.link.ca import cert_fingerprint
+from solstone.think.utils import get_journal
+
+
+class PeerLookupError(Exception):
+    pass
+
+
+@dataclass(frozen=True)
+class PeerInfo:
+    dir: Path
+    instance_id: str
+    label: str
+    local_endpoints: list[dict[str, object]]
+    cert_fingerprint: str
+
+
+_cache: dict[tuple[int, str], PeerInfo] = {}
+
+
+def resolve_peer(label: str) -> PeerInfo:
+    peers_dir = Path(get_journal()) / "peers"
+    if not peers_dir.is_dir():
+        raise PeerLookupError('no peers paired (run "sol link join --as peer" first)')
+
+    mtime_ns = peers_dir.stat().st_mtime_ns
+    if any(key[0] != mtime_ns for key in _cache):
+        _cache.clear()
+
+    cache_key = (mtime_ns, label)
+    cached = _cache.get(cache_key)
+    if cached is not None:
+        return cached
+
+    matches: list[tuple[Path, dict[str, object]]] = []
+    labels: set[str] = set()
+    for peer_dir in sorted(peers_dir.iterdir()):
+        if not peer_dir.is_dir():
+            continue
+        peer_json = peer_dir / "peer.json"
+        if not peer_json.is_file():
+            continue
+        try:
+            peer = json.loads(peer_json.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, OSError) as exc:
+            raise PeerLookupError(f"invalid peer.json in {peer_dir}: {exc}") from exc
+        peer_label = peer.get("label")
+        if isinstance(peer_label, str) and peer_label:
+            labels.add(peer_label)
+        if peer_label == label:
+            matches.append((peer_dir, peer))
+
+    if not matches:
+        labels_str = ", ".join(sorted(labels)) if labels else "none"
+        raise PeerLookupError(f'no peer with label "{label}"; available: {labels_str}')
+    if len(matches) > 1:
+        ids = ", ".join(
+            str(peer.get("instance_id") or peer_dir.name) for peer_dir, peer in matches
+        )
+        raise PeerLookupError(
+            f'multiple peers with label "{label}": {ids}; '
+            "use <journal_root>/peers/<instance_id> directly"
+        )
+
+    peer_dir, peer = matches[0]
+    cert_pem = (peer_dir / "cert.pem").read_text(encoding="utf-8")
+    endpoints = peer.get("local_endpoints") or []
+    if not isinstance(endpoints, list):
+        endpoints = []
+    info = PeerInfo(
+        dir=peer_dir,
+        instance_id=str(peer.get("instance_id") or peer_dir.name),
+        label=label,
+        local_endpoints=endpoints,
+        cert_fingerprint=cert_fingerprint(cert_pem),
+    )
+    _cache[cache_key] = info
+    return info
diff --git a/solstone/observe/peer_unpair.py b/solstone/observe/peer_unpair.py
new file mode 100644
index 000000000..5e18db6aa
--- /dev/null
+++ b/solstone/observe/peer_unpair.py
@@ -0,0 +1,62 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+from __future__ import annotations
+
+import select
+import shutil
+import sys
+from typing import TextIO
+
+from solstone.observe.peer_lookup import PeerInfo
+from solstone.observe.pl_http import PlHttpSession
+
+
+def maybe_prompt_unpair(
+    peer: PeerInfo,
+    session: PlHttpSession,
+    *,
+    stdin: TextIO = sys.stdin,
+    stdout: TextIO = sys.stdout,
+) -> None:
+    prompt = f'Unpair "{peer.label}" now? (y/N) '
+    stdout.write(prompt)
+    stdout.flush()
+
+    if stdin.isatty():
+        answer = stdin.readline().strip()
+    else:
+        readable, _, _ = select.select([stdin], [], [], 5.0)
+        if not readable:
+            stdout.write(f'Keeping peer "{peer.label}" (non-interactive default).\n')
+            stdout.flush()
+            return
+        answer = stdin.readline().strip()
+
+    if answer.lower() == "y":
+        _do_unpair(peer, session, stdout=stdout)
+        return
+    stdout.write(f'Keeping peer "{peer.label}".\n')
+    stdout.flush()
+
+
+def _do_unpair(
+    peer: PeerInfo,
+    session: PlHttpSession,
+    *,
+    stdout: TextIO = sys.stdout,
+) -> None:
+    response = session.post(
+        "/app/link/unpair",
+        json={"fingerprint": peer.cert_fingerprint},
+    )
+    if response.status_code == 200:
+        shutil.rmtree(peer.dir)
+        stdout.write(f'Unpaired "{peer.label}".\n')
+        stdout.flush()
+        return
+    stdout.write(
+        f'Failed to unpair "{peer.label}": HTTP {response.status_code}: '
+        f"{response.content[:200].decode('utf-8', errors='replace')}\n"
+    )
+    stdout.flush()
diff --git a/solstone/observe/pl_http.py b/solstone/observe/pl_http.py
new file mode 100644
index 000000000..9c7ee49a3
--- /dev/null
+++ b/solstone/observe/pl_http.py
@@ -0,0 +1,150 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+from __future__ import annotations
+
+import json as json_lib
+from typing import Any
+from urllib.parse import urlsplit
+
+from urllib3.filepost import encode_multipart_formdata
+
+from solstone.think.link.dialer import TunnelClient
+
+
+class PlHttpResponse:
+    def __init__(
+        self,
+        status_code: int,
+        headers: dict[str, str],
+        content: bytes,
+    ) -> None:
+        self.status_code = status_code
+        self.headers = headers
+        self.content = content
+
+    @property
+    def text(self) -> str:
+        return self.content.decode("utf-8", errors="replace")
+
+    def json(self) -> Any:
+        return json_lib.loads(self.content.decode("utf-8") or "{}")
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            raise RuntimeError(f"HTTP {self.status_code}: {self.text[:200]}")
+
+
+class PlHttpSession:
+    def __init__(self, tunnel: TunnelClient) -> None:
+        self._tunnel = tunnel
+        self.headers: dict[str, str] = {}
+
+    def post(
+        self,
+        url: str,
+        *,
+        data: Any = None,
+        json: Any = None,
+        files: Any = None,
+        headers: dict[str, str] | None = None,
+        timeout: float | tuple[float, float | None] | None = None,
+    ) -> PlHttpResponse:
+        request_headers = self._headers(headers, url)
+        body = b""
+
+        if files is not None:
+            body, content_type = _encode_multipart(data=data, files=files)
+            request_headers["Content-Type"] = content_type
+        elif json is not None:
+            body = json_lib.dumps(json).encode("utf-8")
+            request_headers.setdefault("Content-Type", "application/json")
+        elif data is not None:
+            if isinstance(data, bytes):
+                body = data
+            elif isinstance(data, str):
+                body = data.encode("utf-8")
+            elif isinstance(data, dict):
+                body, content_type = encode_multipart_formdata(list(data.items()))
+                request_headers["Content-Type"] = content_type
+            else:
+                body = bytes(data)
+
+        return self._request("POST", url, headers=request_headers, body=body)
+
+    def get(
+        self,
+        url: str,
+        *,
+        headers: dict[str, str] | None = None,
+        timeout: float | tuple[float, float | None] | None = None,
+    ) -> PlHttpResponse:
+        return self._request("GET", url, headers=self._headers(headers, url), body=b"")
+
+    def close(self) -> None:
+        self._tunnel.close()
+
+    def _request(
+        self,
+        method: str,
+        url: str,
+        *,
+        headers: dict[str, str],
+        body: bytes,
+    ) -> PlHttpResponse:
+        path = _path_from_url(url)
+        status, response_headers, response_body = self._tunnel.request(
+            method,
+            path,
+            headers=headers,
+            body=body,
+        )
+        return PlHttpResponse(status, dict(response_headers), response_body)
+
+    def _headers(self, headers: dict[str, str] | None, url: str) -> dict[str, str]:
+        merged = dict(self.headers)
+        if headers:
+            merged.update(headers)
+        # PL is already mutually authenticated; never forward bearer credentials.
+        for name in list(merged):
+            if name.lower() == "authorization":
+                del merged[name]
+        netloc = urlsplit(url).netloc
+        if netloc:
+            merged["Host"] = netloc
+        return merged
+
+
+def _path_from_url(url: str) -> str:
+    parsed = urlsplit(url)
+    if not parsed.scheme:
+        return url
+    path = parsed.path or "/"
+    if parsed.query:
+        return f"{path}?{parsed.query}"
+    return path
+
+
+def _encode_multipart(data: Any, files: Any) -> tuple[bytes, str]:
+    fields: list[tuple[str, Any]] = []
+    if isinstance(data, dict):
+        fields.extend(data.items())
+    elif data:
+        fields.extend(data)
+
+    for field_name, file_value in files:
+        if isinstance(file_value, tuple):
+            filename = file_value[0]
+            payload = file_value[1]
+            content_type = (
+                file_value[2] if len(file_value) > 2 else "application/octet-stream"
+            )
+        else:
+            filename = getattr(file_value, "name", "file")
+            payload = file_value
+            content_type = "application/octet-stream"
+        if hasattr(payload, "read"):
+            payload = payload.read()
+        fields.append((field_name, (filename, payload, content_type)))
+
+    return encode_multipart_formdata(fields)
diff --git a/solstone/observe/transfer.py b/solstone/observe/transfer.py
index 334ddf3ee..464a0f2b5 100644
--- a/solstone/observe/transfer.py
+++ b/solstone/observe/transfer.py
@@ -9,7 +9,8 @@ observation segments between solstone instances.
 Usage:
     sol transfer export --day YYYYMMDD [--output PATH]
     sol transfer import --archive PATH [--dry-run]
-    sol transfer send --to HOST --key KEY [--day YYYYMMDD] [--dry-run]
+    sol transfer send --to URL --key KEY [--day YYYYMMDD] [--dry-run]
+    sol transfer send --to LABEL [--day YYYYMMDD] [--dry-run]
 
 On the RECEIVING host (the machine you are sending TO), run
 `sol observer create <name>` to generate an observer API key, then pass it
@@ -32,7 +33,12 @@ from typing import Any
 
 import requests
 
+from solstone.observe.peer_lookup import PeerInfo, PeerLookupError, resolve_peer
+from solstone.observe.pl_http import PlHttpSession
 from solstone.think.callosum import callosum_send
+from solstone.think.link.bundle import load_client_identity
+from solstone.think.link.dialer import TunnelClient, TunnelRequestError
+from solstone.think.link.paths import relay_url
 from solstone.think.utils import (
     CHRONICLE_DIR,
     day_path,
@@ -395,6 +401,30 @@ 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"
+        )
+    try:
+        return ("pl", None, resolve_peer(to))
+    except PeerLookupError as exc:
+        parser.error(str(exc))
+
+
 def _parse_day_spec(day_spec: str | None, journal_root: Path) -> list[str]:
     """Parse a single day, day range, or default to all journal days."""
     if day_spec is None:
@@ -546,6 +576,121 @@ def _upload_segment(
     return ("error", 0)
 
 
+def _query_journal_segments(
+    session: PlHttpSession,
+    base_url: str,
+    key_prefix: str,
+) -> dict[str, Any]:
+    url = f"{base_url}/app/import/journal/{key_prefix}/manifest/segments"
+    try:
+        response = session.get(url, timeout=UPLOAD_TIMEOUT)
+        if response.status_code == 200:
+            return response.json()
+        logger.warning(
+            "Remote segment manifest query failed: %s %s",
+            response.status_code,
+            response.text,
+        )
+    except (TunnelRequestError, OSError) as exc:
+        logger.warning("Remote segment manifest query failed: %s", exc)
+    return {}
+
+
+def _upload_segment_journal(
+    session: PlHttpSession,
+    base_url: str,
+    key_prefix: str,
+    day: str,
+    segment_key: str,
+    stream_name: str,
+    segment_path: Path,
+) -> tuple[str, int]:
+    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)
+
+    bytes_sent = sum(file_path.stat().st_size for file_path in files)
+    metadata = {
+        "segments": [
+            {
+                "day": day,
+                "stream": stream_name,
+                "segment_key": segment_key,
+                "files": [file_path.name for file_path in files],
+            }
+        ]
+    }
+    url = f"{base_url}/app/import/journal/{key_prefix}/ingest/segments/{day}"
+
+    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_0", (file_path.name, fh, "application/octet-stream"))
+                )
+
+            response = session.post(
+                url,
+                data={"metadata": json.dumps(metadata)},
+                files=files_data,
+                timeout=UPLOAD_TIMEOUT,
+            )
+            if response.status_code == 200:
+                return ("sent", bytes_sent)
+            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(
+                    "PL upload attempt %s failed for %s/%s/%s: %s %s",
+                    attempt + 1,
+                    day,
+                    stream_name,
+                    segment_key,
+                    response.status_code,
+                    response.text,
+                )
+            else:
+                logger.warning(
+                    "PL upload rejected for %s/%s/%s: %s %s",
+                    day,
+                    stream_name,
+                    segment_key,
+                    response.status_code,
+                    response.text,
+                )
+                return ("error", 0)
+        except (TunnelRequestError, OSError) as exc:
+            logger.warning(
+                "PL upload attempt %s failed for %s/%s/%s: %s",
+                attempt + 1,
+                day,
+                stream_name,
+                segment_key,
+                exc,
+            )
+        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 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()
@@ -647,6 +792,104 @@ def send_segments(base_url: str, key: str, days: list[str], dry_run: bool) -> No
         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)
+    key_prefix = peer.instance_id[:8]
+    base_url = "https://pl.peer"
+
+    sent = 0
+    skipped = 0
+    failed = 0
+    bytes_total = 0
+
+    with TunnelClient(identity, relay_url=relay_url()) as tunnel:
+        session = PlHttpSession(tunnel)
+        remote_manifest = _query_journal_segments(session, base_url, key_prefix)
+
+        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
+
+            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_entry = remote_manifest.get(day, {}).get(
+                    f"{stream_name}/{seg_key}", {}
+                )
+                remote_files = {
+                    file_info["name"]: file_info["sha256"]
+                    for file_info in remote_entry.get("files", [])
+                }
+                if local_files == remote_files:
+                    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_journal(
+                    session,
+                    base_url,
+                    key_prefix,
+                    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 == "skip":
+                    logger.info(f"  [skip] {day}/{stream_name}/{seg_key}")
+                    skipped += 1
+                elif status == "auth_invalid":
+                    print(
+                        "Authentication failed: invalid or missing paired-link identity"
+                    )
+                    return
+                elif status == "auth_revoked":
+                    print(
+                        "Authentication failed: paired-link identity revoked or disabled"
+                    )
+                    return
+                else:
+                    logger.info(f"  [FAILED] {day}/{stream_name}/{seg_key}")
+                    failed += 1
+
+    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 sent == 0 and skipped > 0 and failed == 0:
+        print("Nothing to send - remote is up to date")
+
+
 def main() -> None:
     """CLI entry point."""
     parser = argparse.ArgumentParser(
@@ -694,13 +937,14 @@ def main() -> None:
     send_parser.add_argument(
         "--to",
         required=True,
-        help="Remote observer URL (host:port or https://...)",
+        help="Remote observer URL (http:// or https://) or paired peer label",
     )
     send_parser.add_argument(
         "--key",
-        required=True,
+        required=False,
+        default=None,
         help=(
-            "Observer API key (generate on the RECEIVING host with "
+            "Observer API key for URL mode (generate on the RECEIVING host with "
             "`sol observer create <name>`)"
         ),
     )
@@ -749,10 +993,15 @@ def main() -> None:
             parser.error(str(e))
 
     elif args.command == "send":
-        base_url = _normalize_url(args.to)
+        mode, base_url, peer = _resolve_destination(send_parser, args.to, args.key)
         journal = get_journal()
         try:
             days = _parse_day_spec(args.day, Path(journal))
         except ValueError as e:
             parser.error(str(e))
-        send_segments(base_url, args.key, days, args.dry_run)
+        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)
diff --git a/tests/integration/test_transfer_export_over_pl_e2e.py b/tests/integration/test_transfer_export_over_pl_e2e.py
new file mode 100644
index 000000000..b7b917d91
--- /dev/null
+++ b/tests/integration/test_transfer_export_over_pl_e2e.py
@@ -0,0 +1,12 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+from __future__ import annotations
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+
+def test_transfer_export_over_pl_e2e() -> None:
+    pytest.skip("requires Lode F receiver journal-source PL ingest routes")
diff --git a/tests/test_export_integration.py b/tests/test_export_integration.py
index 29bc36b77..523f297af 100644
--- a/tests/test_export_integration.py
+++ b/tests/test_export_integration.py
@@ -551,7 +551,7 @@ def test_error_resilience(monkeypatch, capsys):
     monkeypatch.setattr(
         "solstone.observe.export.setup_cli",
         lambda parser: Namespace(
-            to="localhost:5000",
+            to="https://localhost:5000",
             key="test-key-123456",
             only=None,
             dry_run=False,
diff --git a/tests/test_export_pl.py b/tests/test_export_pl.py
new file mode 100644
index 000000000..33cf38d93
--- /dev/null
+++ b/tests/test_export_pl.py
@@ -0,0 +1,398 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+from __future__ import annotations
+
+import io
+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
+from solstone.observe.peer_lookup import PeerInfo
+from solstone.observe.peer_unpair import maybe_prompt_unpair
+from solstone.observe.pl_http import PlHttpResponse
+from solstone.think.link.ca import cert_fingerprint, generate_ca
+
+
+def _set_journal(monkeypatch: pytest.MonkeyPatch, journal: Path) -> None:
+    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
+    think_utils._journal_path_cache = None
+
+
+def _write_peer(journal: Path) -> PeerInfo:
+    peer_dir = journal / "peers" / "12345678-1234-1234-1234-123456789abc"
+    peer_dir.mkdir(parents=True)
+    ca = generate_ca(peer_dir / "ca")
+    cert_pem = ca.cert.public_bytes(serialization.Encoding.PEM).decode("ascii")
+    peer = {
+        "label": "host-a",
+        "instance_id": "12345678-1234-1234-1234-123456789abc",
+        "home_label": "solstone",
+        "local_endpoints": [],
+    }
+    (peer_dir / "private.pem").write_text("private", encoding="utf-8")
+    (peer_dir / "cert.pem").write_text(cert_pem, encoding="utf-8")
+    (peer_dir / "chain.pem").write_text(cert_pem, encoding="utf-8")
+    (peer_dir / "home_attestation.jwt").write_text("jwt", encoding="utf-8")
+    (peer_dir / "peer.json").write_text(json.dumps(peer), encoding="utf-8")
+    return PeerInfo(
+        dir=peer_dir,
+        instance_id=peer["instance_id"],
+        label=peer["label"],
+        local_endpoints=[],
+        cert_fingerprint=cert_fingerprint(cert_pem),
+    )
+
+
+def _patch_exporters(monkeypatch: pytest.MonkeyPatch, calls: list[tuple]) -> None:
+    from solstone.observe import export
+
+    def make_exporter(area: str):
+        def fake_export(base_url, key, *args, session=None):
+            session.post(f"{base_url}/app/import/journal/{key[:8]}/ingest/{area}")
+            calls.append(("area", area, session))
+            return export.ExportResult(area=area, sent=1)
+
+        return fake_export
+
+    monkeypatch.setattr(export, "export_segments", make_exporter("segments"))
+    monkeypatch.setattr(export, "export_imports", make_exporter("imports"))
+    monkeypatch.setattr(export, "export_entities", make_exporter("entities"))
+    monkeypatch.setattr(export, "export_facets", make_exporter("facets"))
+    monkeypatch.setattr(export, "export_config", make_exporter("config"))
+
+
+def _patch_tunnel(
+    monkeypatch: pytest.MonkeyPatch,
+    requests_seen: list[tuple[str, str, dict[str, str], bytes]],
+) -> None:
+    class FakeTunnelSession:
+        async def request(self, method, path, *, headers=None, body=b""):
+            requests_seen.append((method, path, headers or {}, body))
+            return (200, {}, b'{"ok": true}')
+
+        async def close(self) -> None:
+            return None
+
+    async def fake_open_tunnel(_identity, _relay_url):
+        return FakeTunnelSession()
+
+    monkeypatch.setattr("solstone.think.link.dialer.open_tunnel", fake_open_tunnel)
+    monkeypatch.setenv("SOL_LINK_RELAY_URL", "https://relay.test")
+
+
+def test_export_pl_single_area_uses_pl_url_and_no_authorization(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from solstone.observe import export
+
+    journal = tmp_path / "journal"
+    journal.mkdir()
+    _set_journal(monkeypatch, journal)
+    _write_peer(journal)
+    area_calls: list[tuple] = []
+    requests_seen: list[tuple[str, str, dict[str, str], bytes]] = []
+    _patch_exporters(monkeypatch, area_calls)
+    _patch_tunnel(monkeypatch, requests_seen)
+    prompt_called = False
+
+    def prompt(*_args, **_kwargs):
+        nonlocal prompt_called
+        prompt_called = True
+
+    monkeypatch.setattr(export, "maybe_prompt_unpair", prompt)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["sol", "--to", "host-a", "--only", "entities", "--day", "20260520"],
+    )
+
+    export.main()
+
+    assert [call[1] for call in area_calls] == ["entities"]
+    assert requests_seen[0][1] == "/app/import/journal/12345678/ingest/entities"
+    assert "Authorization" not in requests_seen[0][2]
+    assert prompt_called is False
+
+
+@pytest.mark.parametrize(
+    "only_arg",
+    [
+        None,
+        "segments,imports,entities,facets,config",
+        "entities,facets,segments,imports,config",
+    ],
+)
+def test_export_pl_full_pipeline_reuses_single_tunnel_and_prompts(
+    only_arg: str | None,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from solstone.observe import export
+
+    journal = tmp_path / "journal"
+    journal.mkdir()
+    _set_journal(monkeypatch, journal)
+    _write_peer(journal)
+    area_calls: list[tuple] = []
+    requests_seen: list[tuple[str, str, dict[str, str], bytes]] = []
+    _patch_exporters(monkeypatch, area_calls)
+    _patch_tunnel(monkeypatch, requests_seen)
+    prompt_calls = 0
+
+    def prompt(*_args, **_kwargs):
+        nonlocal prompt_calls
+        prompt_calls += 1
+
+    monkeypatch.setattr(export, "maybe_prompt_unpair", prompt)
+    argv = ["sol", "--to", "host-a", "--day", "20260520"]
+    if only_arg is not None:
+        argv.extend(["--only", only_arg])
+    monkeypatch.setattr(sys, "argv", argv)
+
+    export.main()
+
+    assert [call[1] for call in area_calls] == [
+        "segments",
+        "imports",
+        "entities",
+        "facets",
+        "config",
+    ]
+    assert [call[1] for call in requests_seen] == [
+        "/app/import/journal/12345678/ingest/segments",
+        "/app/import/journal/12345678/ingest/imports",
+        "/app/import/journal/12345678/ingest/entities",
+        "/app/import/journal/12345678/ingest/facets",
+        "/app/import/journal/12345678/ingest/config",
+    ]
+    assert prompt_calls == 1
+
+
+def test_export_pl_partial_and_dry_run_do_not_prompt(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from solstone.observe import export
+
+    journal = tmp_path / "journal"
+    journal.mkdir()
+    _set_journal(monkeypatch, journal)
+    _write_peer(journal)
+    area_calls: list[tuple] = []
+    requests_seen: list[tuple[str, str, dict[str, str], bytes]] = []
+    _patch_exporters(monkeypatch, area_calls)
+    _patch_tunnel(monkeypatch, requests_seen)
+    prompt_calls = 0
+
+    def prompt(*_args, **_kwargs):
+        nonlocal prompt_calls
+        prompt_calls += 1
+
+    monkeypatch.setattr(export, "maybe_prompt_unpair", prompt)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "sol",
+            "--to",
+            "host-a",
+            "--only",
+            "segments,entities",
+            "--day",
+            "20260520",
+        ],
+    )
+
+    export.main()
+
+    assert [call[1] for call in area_calls] == ["segments", "entities"]
+    assert prompt_calls == 0
+
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["sol", "--to", "host-a", "--dry-run", "--day", "20260520"],
+    )
+    export.main()
+
+    assert prompt_calls == 0
+
+
+def test_export_dl_never_prompts(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from solstone.observe import export
+
+    journal = tmp_path / "journal"
+    journal.mkdir()
+    _set_journal(monkeypatch, journal)
+    area_calls: list[tuple] = []
+    _patch_exporters(monkeypatch, area_calls)
+    prompt_calls = 0
+
+    class FakeSession:
+        def __init__(self) -> None:
+            self.headers = {}
+
+        def post(self, *_args, **_kwargs):
+            return PlHttpResponse(200, {}, b'{"ok": true}')
+
+        def close(self) -> None:
+            return None
+
+    monkeypatch.setattr(export.requests, "Session", FakeSession)
+
+    def prompt(*_args, **_kwargs):
+        nonlocal prompt_calls
+        prompt_calls += 1
+
+    monkeypatch.setattr(export, "maybe_prompt_unpair", prompt)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "sol",
+            "--to",
+            "https://receiver.test",
+            "--key",
+            "api-key",
+            "--day",
+            "20260520",
+        ],
+    )
+
+    export.main()
+
+    assert prompt_calls == 0
+    assert [call[1] for call in area_calls] == [
+        "segments",
+        "imports",
+        "entities",
+        "facets",
+        "config",
+    ]
+
+
+def test_export_dl_regression_config_url_headers_and_body(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from solstone.observe.export import export_config
+
+    journal = tmp_path / "journal"
+    _set_journal(monkeypatch, journal)
+    config_dir = journal / "config"
+    config_dir.mkdir(parents=True)
+    config = {
+        "identity": {"name": "Test"},
+        "convey": {
+            "password_hash": "secret",
+            "secret": "secret",
+            "trust_localhost": True,
+        },
+    }
+    (config_dir / "journal.json").write_text(json.dumps(config), encoding="utf-8")
+
+    mock_session = MagicMock(spec=requests.Session)
+    mock_session.headers = {}
+    get_response = MagicMock()
+    get_response.status_code = 200
+    get_response.json.return_value = {"last_hash": "different"}
+    mock_session.get.return_value = get_response
+    post_response = MagicMock()
+    post_response.status_code = 200
+    post_response.json.return_value = {"staged": True, "diff_fields": 1}
+    mock_session.post.return_value = post_response
+
+    with patch("solstone.observe.export.requests.Session", return_value=mock_session):
+        export_config("https://receiver.test", "abcdef123456", False)
+
+    assert mock_session.headers["Authorization"] == "Bearer abcdef123456"
+    assert mock_session.get.call_args.args[0] == (
+        "https://receiver.test/app/import/journal/abcdef12/manifest/config"
+    )
+    assert mock_session.post.call_args.args[0] == (
+        "https://receiver.test/app/import/journal/abcdef12/ingest/config"
+    )
+    payload = mock_session.post.call_args.kwargs["json"]["config"]
+    assert payload["identity"] == {"name": "Test"}
+    assert payload["convey"] == {"trust_localhost": True}
+
+
+class _FakeInput(io.StringIO):
+    def __init__(self, value: str, *, tty: bool) -> None:
+        super().__init__(value)
+        self._tty = tty
+
+    def isatty(self) -> bool:
+        return self._tty
+
+
+def test_unpair_prompt_yes_deletes_peer(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    peer_dir = tmp_path / "peer"
+    peer_dir.mkdir()
+    peer = PeerInfo(peer_dir, "iid", "host-a", [], "sha256:" + ("a" * 64))
+    posts: list[dict] = []
+
+    class FakeSession:
+        def post(self, _url, *, json=None):
+            posts.append(json)
+            return PlHttpResponse(200, {}, b"{}")
+
+    stdout = io.StringIO()
+    maybe_prompt_unpair(
+        peer,
+        FakeSession(),
+        stdin=_FakeInput("y\n", tty=True),
+        stdout=stdout,
+    )
+
+    assert posts == [{"fingerprint": peer.cert_fingerprint}]
+    assert not peer_dir.exists()
+    assert 'Unpaired "host-a".' in stdout.getvalue()
+
+
+def test_unpair_prompt_decline_and_non_tty_timeout(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    peer = PeerInfo(tmp_path / "peer", "iid", "host-a", [], "sha256:" + ("a" * 64))
+    peer.dir.mkdir()
+
+    class FakeSession:
+        def post(self, *_args, **_kwargs):
+            raise AssertionError("unpair should not post")
+
+    stdout = io.StringIO()
+    maybe_prompt_unpair(
+        peer,
+        FakeSession(),
+        stdin=_FakeInput("N\n", tty=True),
+        stdout=stdout,
+    )
+    assert 'Keeping peer "host-a".' in stdout.getvalue()
+    assert peer.dir.exists()
+
+    stdout = io.StringIO()
+    monkeypatch.setattr(
+        "solstone.observe.peer_unpair.select.select", lambda *args: ([], [], [])
+    )
+    maybe_prompt_unpair(
+        peer,
+        FakeSession(),
+        stdin=_FakeInput("", tty=False),
+        stdout=stdout,
+    )
+    assert 'Keeping peer "host-a" (non-interactive default).' in stdout.getvalue()
diff --git a/tests/test_peer_lookup.py b/tests/test_peer_lookup.py
new file mode 100644
index 000000000..a1f720dc9
--- /dev/null
+++ b/tests/test_peer_lookup.py
@@ -0,0 +1,131 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+
+import solstone.observe.peer_lookup as peer_lookup
+import solstone.think.utils as think_utils
+from solstone.think.link.ca import cert_fingerprint, generate_ca
+
+
+def _set_journal(monkeypatch: pytest.MonkeyPatch, journal: Path) -> None:
+    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
+    think_utils._journal_path_cache = None
+    peer_lookup._cache.clear()
+
+
+def _write_peer(
+    journal: Path,
+    instance_id: str,
+    label: str,
+    *,
+    endpoints: list[dict[str, object]] | None = None,
+) -> Path:
+    peer_dir = journal / "peers" / instance_id
+    peer_dir.mkdir(parents=True)
+    ca = generate_ca(peer_dir / "ca")
+    cert_pem = ca.cert.public_bytes(serialization.Encoding.PEM).decode("ascii")
+    (peer_dir / "cert.pem").write_text(cert_pem, encoding="utf-8")
+    (peer_dir / "private.pem").write_text("private", encoding="utf-8")
+    (peer_dir / "chain.pem").write_text(cert_pem, encoding="utf-8")
+    (peer_dir / "home_attestation.jwt").write_text("jwt", encoding="utf-8")
+    (peer_dir / "peer.json").write_text(
+        json.dumps(
+            {
+                "label": label,
+                "instance_id": instance_id,
+                "home_label": "solstone",
+                "local_endpoints": endpoints or [{"ip": "127.0.0.1", "port": 7657}],
+            }
+        ),
+        encoding="utf-8",
+    )
+    return peer_dir
+
+
+def test_resolve_peer_happy_path(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    journal = tmp_path / "journal"
+    _set_journal(monkeypatch, journal)
+    peer_dir = _write_peer(
+        journal,
+        "12345678-1234-1234-1234-123456789abc",
+        "host-a",
+    )
+
+    info = peer_lookup.resolve_peer("host-a")
+
+    assert info.dir == peer_dir
+    assert info.instance_id == "12345678-1234-1234-1234-123456789abc"
+    assert info.label == "host-a"
+    assert info.local_endpoints == [{"ip": "127.0.0.1", "port": 7657}]
+    expected = cert_fingerprint((peer_dir / "cert.pem").read_text(encoding="utf-8"))
+    assert info.cert_fingerprint == expected
+    assert info.cert_fingerprint.startswith("sha256:")
+
+
+def test_resolve_peer_no_match_lists_available(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    journal = tmp_path / "journal"
+    _set_journal(monkeypatch, journal)
+    (journal / "peers").mkdir(parents=True)
+
+    with pytest.raises(peer_lookup.PeerLookupError, match="available: none"):
+        peer_lookup.resolve_peer("missing")
+
+    _write_peer(journal, "iid-a", "host-a")
+    _write_peer(journal, "iid-b", "host-b")
+
+    with pytest.raises(peer_lookup.PeerLookupError) as exc_info:
+        peer_lookup.resolve_peer("missing")
+
+    message = str(exc_info.value)
+    assert 'no peer with label "missing"' in message
+    assert "available: host-a, host-b" in message
+
+
+def test_resolve_peer_multiple_matches(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    journal = tmp_path / "journal"
+    _set_journal(monkeypatch, journal)
+    _write_peer(journal, "iid-a", "host-a")
+    _write_peer(journal, "iid-b", "host-a")
+
+    with pytest.raises(peer_lookup.PeerLookupError) as exc_info:
+        peer_lookup.resolve_peer("host-a")
+
+    message = str(exc_info.value)
+    assert 'multiple peers with label "host-a"' in message
+    assert "iid-a" in message
+    assert "iid-b" in message
+    assert "use <journal_root>/peers/<instance_id>" in message
+
+
+def test_resolve_peer_cache_invalidates_on_peers_dir_mtime(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    journal = tmp_path / "journal"
+    _set_journal(monkeypatch, journal)
+    _write_peer(journal, "iid-a", "host-a")
+
+    assert peer_lookup.resolve_peer("host-a").instance_id == "iid-a"
+
+    peers_dir = journal / "peers"
+    _write_peer(journal, "iid-b", "host-b")
+    stat = peers_dir.stat()
+    os.utime(peers_dir, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000_000))
+
+    assert peer_lookup.resolve_peer("host-b").instance_id == "iid-b"
diff --git a/tests/test_transfer_pl.py b/tests/test_transfer_pl.py
new file mode 100644
index 000000000..caaff824e
--- /dev/null
+++ b/tests/test_transfer_pl.py
@@ -0,0 +1,167 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+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
+from solstone.observe.peer_lookup import PeerInfo
+from solstone.observe.transfer import send_segments_pl
+from solstone.think.link.ca import cert_fingerprint, generate_ca
+
+
+def _set_journal(monkeypatch: pytest.MonkeyPatch, journal: Path) -> None:
+    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
+    monkeypatch.setenv("SOL_SKIP_SUPERVISOR_CHECK", "1")
+    think_utils._journal_path_cache = None
+
+
+def _write_peer(journal: Path) -> PeerInfo:
+    peer_dir = journal / "peers" / "12345678-1234-1234-1234-123456789abc"
+    peer_dir.mkdir(parents=True)
+    ca = generate_ca(peer_dir / "ca")
+    cert_pem = ca.cert.public_bytes(serialization.Encoding.PEM).decode("ascii")
+    peer = {
+        "label": "host-a",
+        "instance_id": "12345678-1234-1234-1234-123456789abc",
+        "home_label": "solstone",
+        "local_endpoints": [],
+    }
+    (peer_dir / "private.pem").write_text("private", encoding="utf-8")
+    (peer_dir / "cert.pem").write_text(cert_pem, encoding="utf-8")
+    (peer_dir / "chain.pem").write_text(cert_pem, encoding="utf-8")
+    (peer_dir / "home_attestation.jwt").write_text("jwt", encoding="utf-8")
+    (peer_dir / "peer.json").write_text(json.dumps(peer), encoding="utf-8")
+    return PeerInfo(
+        dir=peer_dir,
+        instance_id=peer["instance_id"],
+        label=peer["label"],
+        local_endpoints=[],
+        cert_fingerprint=cert_fingerprint(cert_pem),
+    )
+
+
+def _write_segment(journal: Path) -> None:
+    segment = journal / "chronicle" / "20260520" / "laptop" / "143022_300"
+    segment.mkdir(parents=True)
+    (segment / "audio.flac").write_bytes(b"audio")
+    (segment / "transcript.jsonl").write_bytes(b"transcript")
+    (segment / "stream.json").write_bytes(b'{"name": "laptop"}')
+
+
+def test_transfer_send_pl_posts_journal_segment_day_path(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    journal = tmp_path / "journal"
+    _set_journal(monkeypatch, journal)
+    peer = _write_peer(journal)
+    _write_segment(journal)
+    calls: list[tuple[str, str, dict[str, str], bytes]] = []
+
+    class FakeTunnelSession:
+        async def request(self, method, path, *, headers=None, body=b""):
+            calls.append((method, path, headers or {}, body))
+            if method == "GET":
+                return (200, {}, b"{}")
+            return (200, {}, b'{"ok": true}')
+
+        async def close(self) -> None:
+            return None
+
+    async def fake_open_tunnel(_identity, _relay_url):
+        return FakeTunnelSession()
+
+    monkeypatch.setattr("solstone.think.link.dialer.open_tunnel", fake_open_tunnel)
+    monkeypatch.setenv("SOL_LINK_RELAY_URL", "https://relay.test")
+
+    send_segments_pl(peer, ["20260520"], dry_run=False)
+
+    post = next(call for call in calls if call[0] == "POST")
+    assert post[1] == "/app/import/journal/12345678/ingest/segments/20260520"
+    assert "Authorization" not in post[2]
+    assert b'"day": "20260520"' in post[3]
+    assert b'"stream": "laptop"' in post[3]
+    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)
+
+    with pytest.raises(SystemExit):
+        transfer.main()
+
+    assert message in capsys.readouterr().err