diff --git a/apps/entities/call.py b/apps/entities/call.py
index c7bceeb91..6e0d995e1 100644
--- a/apps/entities/call.py
+++ b/apps/entities/call.py
@@ -6,6 +6,7 @@
 Auto-discovered by ``think.call`` and mounted as ``sol call entities ...``.
 """
 
+import json
 import re
 import shutil
 from pathlib import Path
@@ -438,6 +439,33 @@ def consolidate(
     typer.echo(f"Wrote {n} new entities.")
 
 
+@app.command("merge")
+def merge(
+    source_slug: str = typer.Argument(help="Source entity slug to merge from."),
+    target_slug: str = typer.Argument(help="Target entity slug to merge into."),
+    commit: bool = typer.Option(False, "--commit/--no-commit"),
+    keep_source_as_aka: bool = typer.Option(
+        True,
+        "--keep-source-as-aka/--no-keep-source-as-aka",
+    ),
+) -> None:
+    """Plan or commit a journal-entity merge."""
+    from think.entities import merge_entity
+
+    result = merge_entity(
+        source_slug,
+        target_slug,
+        keep_source_as_aka=keep_source_as_aka,
+        commit=commit,
+        caller="entities.merge",
+    )
+    output = json.dumps(result, indent=2, default=str)
+    if "error" in result:
+        typer.echo(output, err=True)
+        raise typer.Exit(1)
+    typer.echo(output)
+
+
 @app.command("observations")
 def list_observations(
     entity: str = typer.Argument(help="Entity name or identifier."),
diff --git a/apps/entities/tests/conftest.py b/apps/entities/tests/conftest.py
index 337c8060f..cdd74429b 100644
--- a/apps/entities/tests/conftest.py
+++ b/apps/entities/tests/conftest.py
@@ -6,13 +6,32 @@
 from __future__ import annotations
 
 import json
+import sys
+from pathlib import Path
 
 import pytest
 
-from think.entities.observations import add_observation, save_observations
+ROOT = Path(__file__).resolve().parents[3]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from apps.speakers.tests.conftest import speakers_env as _speakers_env
+from think.entities.journal import clear_journal_entity_cache
+from think.entities.loading import clear_entity_loading_cache
+from think.entities.observations import (
+    add_observation,
+    clear_observation_cache,
+    save_observations,
+)
+from think.entities.relationships import clear_relationship_caches
 from think.entities.saving import save_entities
 
 
+@pytest.fixture
+def speakers_env(tmp_path, monkeypatch):
+    yield from _speakers_env.__wrapped__(tmp_path, monkeypatch)
+
+
 @pytest.fixture
 def entity_env(tmp_path, monkeypatch):
     """Create a temporary journal with entity data.
@@ -25,6 +44,13 @@ def entity_env(tmp_path, monkeypatch):
             # _SOLSTONE_JOURNAL_OVERRIDE is set, entity files exist
     """
     monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path))
+    clear_journal_entity_cache()
+    clear_entity_loading_cache()
+    clear_relationship_caches()
+    clear_observation_cache()
+    import think.utils
+
+    think.utils._journal_path_cache = None
 
     def _create(
         attached: list[dict] | None = None,
@@ -43,13 +69,27 @@ def entity_env(tmp_path, monkeypatch):
                 add_observation(facet, observation_entity, content, i)
         return tmp_path
 
-    return _create
+    yield _create
+    clear_journal_entity_cache()
+    clear_entity_loading_cache()
+    clear_relationship_caches()
+    clear_observation_cache()
+    import think.utils
+
+    think.utils._journal_path_cache = None
 
 
 @pytest.fixture
 def entity_move_env(tmp_path, monkeypatch):
     """Create a two-facet environment for entity move tests."""
     monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path))
+    clear_journal_entity_cache()
+    clear_entity_loading_cache()
+    clear_relationship_caches()
+    clear_observation_cache()
+    import think.utils
+
+    think.utils._journal_path_cache = None
 
     def _create(
         entity_name: str = "Alice Johnson",
@@ -87,4 +127,11 @@ def entity_move_env(tmp_path, monkeypatch):
 
         return tmp_path, src_facet, dst_facet, entity_name
 
-    return _create
+    yield _create
+    clear_journal_entity_cache()
+    clear_entity_loading_cache()
+    clear_relationship_caches()
+    clear_observation_cache()
+    import think.utils
+
+    think.utils._journal_path_cache = None
diff --git a/apps/entities/tests/test_merge.py b/apps/entities/tests/test_merge.py
new file mode 100644
index 000000000..88fd1b374
--- /dev/null
+++ b/apps/entities/tests/test_merge.py
@@ -0,0 +1,435 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Tests for ``sol call entities merge``."""
+
+from __future__ import annotations
+
+import json
+
+import numpy as np
+from typer.testing import CliRunner
+
+from apps.entities.call import app as entities_app
+from think.entities.journal import load_journal_entity
+
+runner = CliRunner()
+STREAM = "test"
+
+
+def _read_json(path):
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _write_json(path, payload) -> None:
+    path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
+
+
+def _entity_path(env, entity_id: str):
+    return env.journal / "entities" / entity_id / "entity.json"
+
+
+def _update_entity(env, entity_id: str, **fields) -> None:
+    path = _entity_path(env, entity_id)
+    payload = _read_json(path)
+    payload.update(fields)
+    _write_json(path, payload)
+
+
+def _labels_path(env, day: str, segment_key: str):
+    return env.journal / day / STREAM / segment_key / "talents" / "speaker_labels.json"
+
+
+def _corrections_path(env, day: str, segment_key: str):
+    return (
+        env.journal
+        / day
+        / STREAM
+        / segment_key
+        / "talents"
+        / "speaker_corrections.json"
+    )
+
+
+def _audit_log_path(env):
+    return env.journal / "logs" / "entity-merges.jsonl"
+
+
+def _voiceprint_count(env, entity_id: str) -> int:
+    path = env.journal / "entities" / entity_id / "voiceprints.npz"
+    with np.load(path, allow_pickle=False) as data:
+        return len(data["embeddings"])
+
+
+def test_merge_dry_run_plans_without_writing(speakers_env):
+    env = speakers_env()
+    env.create_segment("20240101", "143022_300", ["mic_audio"])
+    env.create_entity(
+        "Dry Alias",
+        voiceprints=[
+            ("20240101", "143022_300", "mic_audio", 1),
+            ("20240101", "143022_300", "mic_audio", 2),
+        ],
+    )
+    env.create_entity(
+        "Dry Canon",
+        voiceprints=[("20240101", "143022_300", "mic_audio", 3)],
+    )
+    env.create_facet_relationship(
+        "work",
+        "dry_alias",
+        observations=["Likes coffee"],
+    )
+    env.create_facet_relationship(
+        "work",
+        "dry_canon",
+        observations=["Senior role"],
+    )
+    env.create_facet_relationship("personal", "dry_alias", description="Runner")
+    env.create_speaker_labels(
+        "20240101",
+        "143022_300",
+        [
+            {
+                "sentence_id": 1,
+                "speaker": "dry_alias",
+                "confidence": "high",
+                "method": "acoustic",
+            }
+        ],
+    )
+    env.create_speaker_corrections(
+        "20240101",
+        "143022_300",
+        [
+            {
+                "sentence_id": 1,
+                "original_speaker": "dry_alias",
+                "corrected_speaker": "dry_alias",
+                "timestamp": 1700000000000,
+            }
+        ],
+    )
+    cache_path = env.journal / "awareness" / "discovery_clusters.json"
+    cache_path.parent.mkdir(parents=True, exist_ok=True)
+    cache_path.write_text('{"clusters": []}', encoding="utf-8")
+
+    source_before = _entity_path(env, "dry_alias").read_text(encoding="utf-8")
+    target_before = _entity_path(env, "dry_canon").read_text(encoding="utf-8")
+    labels_before = _labels_path(env, "20240101", "143022_300").read_text(
+        encoding="utf-8"
+    )
+    corrections_before = _corrections_path(env, "20240101", "143022_300").read_text(
+        encoding="utf-8"
+    )
+
+    result = runner.invoke(entities_app, ["merge", "dry_alias", "dry_canon"])
+
+    assert result.exit_code == 0, f"{result.output}\n{result.exception!r}"
+    data = json.loads(result.output)
+    assert data["merged"] is False
+    assert data["identity"]["akas_added"] == []
+    assert data["voiceprints"]["added"] == 0
+    assert data["facets"]["moved"] == []
+    assert data["segments"]["files_scanned"] == 0
+    assert "Dry Alias" in data["would_identity"]["akas_added"]
+    assert data["would_voiceprints"]["added"] == 2
+    assert data["would_facets"]["merged"] == ["work"]
+    assert data["would_facets"]["moved"] == ["personal"]
+    assert data["would_segments"]["labels_rewritten"] == 1
+    assert data["would_segments"]["corrections_rewritten"] == 1
+    assert data["audit_log_path"] is None
+    assert data["caches_cleared"] == []
+
+    assert _entity_path(env, "dry_alias").read_text(encoding="utf-8") == source_before
+    assert _entity_path(env, "dry_canon").read_text(encoding="utf-8") == target_before
+    assert (
+        _labels_path(env, "20240101", "143022_300").read_text(encoding="utf-8")
+        == labels_before
+    )
+    assert (
+        _corrections_path(env, "20240101", "143022_300").read_text(encoding="utf-8")
+        == corrections_before
+    )
+    assert cache_path.exists()
+    assert load_journal_entity("dry_alias") is not None
+    assert not _audit_log_path(env).exists()
+
+
+def test_merge_commit_deep_merges_and_logs(speakers_env):
+    env = speakers_env()
+    env.create_segment("20240101", "143022_300", ["mic_audio"])
+    env.create_entity(
+        "Alice Alias",
+        voiceprints=[
+            ("20240101", "143022_300", "mic_audio", 1),
+            ("20240101", "143022_300", "mic_audio", 2),
+        ],
+    )
+    env.create_entity(
+        "Alice Canonical",
+        voiceprints=[("20240101", "143022_300", "mic_audio", 3)],
+    )
+    env.create_facet_relationship(
+        "work",
+        "alice_alias",
+        description="Works at Acme",
+        attached_at=1600000000000,
+        observations=["Likes coffee", "Morning person"],
+    )
+    env.create_facet_relationship(
+        "work",
+        "alice_canonical",
+        description="Senior engineer",
+        attached_at=1700000000000,
+        observations=["Staff role"],
+    )
+    env.create_facet_relationship("personal", "alice_alias", description="Hiker")
+    env.create_speaker_labels(
+        "20240101",
+        "143022_300",
+        [
+            {
+                "sentence_id": 1,
+                "speaker": "alice_alias",
+                "confidence": "high",
+                "method": "acoustic",
+            },
+            {
+                "sentence_id": 2,
+                "speaker": "alice_canonical",
+                "confidence": "high",
+                "method": "acoustic",
+            },
+            {
+                "sentence_id": 3,
+                "speaker": "alice_alias",
+                "confidence": "medium",
+                "method": "context",
+            },
+        ],
+    )
+    env.create_speaker_corrections(
+        "20240101",
+        "143022_300",
+        [
+            {
+                "sentence_id": 1,
+                "original_speaker": "alice_alias",
+                "corrected_speaker": "alice_alias",
+                "timestamp": 1700000000000,
+            },
+        ],
+    )
+    cache_path = env.journal / "awareness" / "discovery_clusters.json"
+    cache_path.parent.mkdir(parents=True, exist_ok=True)
+    cache_path.write_text('{"clusters": []}', encoding="utf-8")
+
+    result = runner.invoke(
+        entities_app,
+        ["merge", "alice_alias", "alice_canonical", "--commit"],
+    )
+
+    assert result.exit_code == 0, f"{result.output}\n{result.exception!r}"
+    data = json.loads(result.output)
+    assert data["merged"] is True
+    assert "Alice Alias" in data["identity"]["akas_added"]
+    assert data["voiceprints"]["added"] == 2
+    assert data["voiceprints"]["target_total"] == 3
+    assert "work" in data["facets"]["merged"]
+    assert "personal" in data["facets"]["moved"]
+    assert data["facets"]["observations_appended"] == 2
+    assert data["segments"]["labels_rewritten"] == 1
+    assert data["segments"]["corrections_rewritten"] == 1
+    assert data["segments"]["errors"] == []
+    assert data["audit_log_path"] == str(_audit_log_path(env))
+    assert set(data["caches_cleared"]) >= {
+        "journal_entity_cache",
+        "relationship_caches",
+        "observation_cache",
+        "entity_loading_cache",
+        "discovery_clusters",
+    }
+
+    assert load_journal_entity("alice_alias") is None
+    canonical = load_journal_entity("alice_canonical")
+    assert canonical is not None
+    assert "Alice Alias" in canonical["aka"]
+
+    assert _voiceprint_count(env, "alice_canonical") == 3
+
+    labels = _read_json(_labels_path(env, "20240101", "143022_300"))
+    speakers = [label["speaker"] for label in labels["labels"]]
+    assert "alice_alias" not in speakers
+    assert speakers.count("alice_canonical") == 3
+
+    corrections = _read_json(_corrections_path(env, "20240101", "143022_300"))
+    for correction in corrections["corrections"]:
+        assert correction.get("original_speaker") != "alice_alias"
+        assert correction.get("corrected_speaker") != "alice_alias"
+
+    observations_path = (
+        env.journal
+        / "facets"
+        / "work"
+        / "entities"
+        / "alice_canonical"
+        / "observations.jsonl"
+    )
+    contents = [
+        json.loads(line)["content"]
+        for line in observations_path.read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]
+    assert set(contents) == {"Staff role", "Likes coffee", "Morning person"}
+    assert not cache_path.exists()
+
+    audit_entries = [
+        json.loads(line)
+        for line in _audit_log_path(env).read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]
+    assert len(audit_entries) == 1
+    assert isinstance(audit_entries[0]["ts"], int)
+    assert audit_entries[0]["caller"] == "entities.merge"
+    assert audit_entries[0]["source_id"] == "alice_alias"
+    assert audit_entries[0]["source_display_name"] == "Alice Alias"
+    assert audit_entries[0]["target_id"] == "alice_canonical"
+    assert audit_entries[0]["target_display_name"] == "Alice Canonical"
+    assert audit_entries[0]["principal_transferred"] is False
+    assert set(audit_entries[0]["counts"]) == {
+        "identity",
+        "voiceprints",
+        "facets",
+        "segments",
+    }
+
+
+def test_merge_default_keeps_source_as_aka(speakers_env):
+    env = speakers_env()
+    env.create_entity("Keep Alias")
+    env.create_entity("Keep Canon")
+
+    result = runner.invoke(
+        entities_app,
+        ["merge", "keep_alias", "keep_canon", "--commit"],
+    )
+
+    assert result.exit_code == 0, f"{result.output}\n{result.exception!r}"
+    canonical = load_journal_entity("keep_canon")
+    assert canonical is not None
+    assert "Keep Alias" in canonical["aka"]
+
+
+def test_merge_no_keep_source_as_aka_keeps_only_existing_aliases(speakers_env):
+    env = speakers_env()
+    env.create_entity("Skip Alias")
+    env.create_entity("Skip Canon")
+    _update_entity(env, "skip_alias", aka=["SA", "S.A."])
+
+    result = runner.invoke(
+        entities_app,
+        [
+            "merge",
+            "skip_alias",
+            "skip_canon",
+            "--commit",
+            "--no-keep-source-as-aka",
+        ],
+    )
+
+    assert result.exit_code == 0, f"{result.output}\n{result.exception!r}"
+    canonical = load_journal_entity("skip_canon")
+    assert canonical is not None
+    assert "Skip Alias" not in canonical.get("aka", [])
+    assert {"SA", "S.A."} <= set(canonical["aka"])
+
+
+def test_merge_transfers_principal_from_source_to_target(speakers_env):
+    env = speakers_env()
+    env.create_entity("Principal Source", is_principal=True)
+    env.create_entity("Principal Target")
+
+    result = runner.invoke(
+        entities_app,
+        ["merge", "principal_source", "principal_target", "--commit"],
+    )
+
+    assert result.exit_code == 0, f"{result.output}\n{result.exception!r}"
+    data = json.loads(result.output)
+    assert data["identity"]["principal_transferred"] is True
+    assert load_journal_entity("principal_source") is None
+    target = load_journal_entity("principal_target")
+    assert target is not None
+    assert target["is_principal"] is True
+
+
+def test_merge_errors_when_both_entities_are_principal(speakers_env):
+    env = speakers_env()
+    env.create_entity("First Principal", is_principal=True)
+    env.create_entity("Second Principal", is_principal=True)
+
+    result = runner.invoke(
+        entities_app,
+        ["merge", "first_principal", "second_principal", "--commit"],
+    )
+
+    assert result.exit_code == 1, f"{result.output}\n{result.exception!r}"
+    data = json.loads(result.output)
+    assert data["error"] == "Cannot merge two principal entities."
+    assert load_journal_entity("first_principal") is not None
+    assert load_journal_entity("second_principal") is not None
+
+
+def test_merge_errors_on_aka_cross_reference(speakers_env):
+    env = speakers_env()
+    env.create_entity("Cross Source")
+    env.create_entity("Cross Target")
+    env.create_entity("Cross Watcher")
+    _update_entity(env, "cross_watcher", aka=["cross_source", "Watcher Alias"])
+
+    result = runner.invoke(
+        entities_app,
+        ["merge", "cross_source", "cross_target", "--commit"],
+    )
+
+    assert result.exit_code == 1, f"{result.output}\n{result.exception!r}"
+    data = json.loads(result.output)
+    assert (
+        data["error"]
+        == "Cannot merge 'cross_source': referenced in aka lists of entity ids: cross_watcher"
+    )
+    assert load_journal_entity("cross_source") is not None
+    assert load_journal_entity("cross_target") is not None
+
+
+def test_merge_validation_errors(speakers_env):
+    env = speakers_env()
+    env.create_entity("Blocked Source")
+    env.create_entity("Validation Target")
+    _update_entity(env, "blocked_source", blocked=True)
+
+    cases = [
+        (
+            ["merge", "validation_target", "validation_target", "--commit"],
+            "Source and target must be different entities.",
+        ),
+        (
+            ["merge", "missing_source", "validation_target", "--commit"],
+            "Source entity not found: missing_source",
+        ),
+        (
+            ["merge", "validation_target", "missing_target", "--commit"],
+            "Target entity not found: missing_target",
+        ),
+        (
+            ["merge", "blocked_source", "validation_target", "--commit"],
+            "Cannot merge blocked entity: blocked_source",
+        ),
+    ]
+
+    for argv, expected_error in cases:
+        result = runner.invoke(entities_app, argv)
+        assert result.exit_code == 1, f"{result.output}\n{result.exception!r}"
+        data = json.loads(result.output)
+        assert data["error"] == expected_error
diff --git a/apps/speakers/attribution.py b/apps/speakers/attribution.py
index 446da130e..280ca05bd 100644
--- a/apps/speakers/attribution.py
+++ b/apps/speakers/attribution.py
@@ -547,9 +547,9 @@ def accumulate_voiceprints(
 
     Returns dict mapping entity_id -> number of new embeddings saved.
     """
-    from apps.speakers.bootstrap import (
-        _load_existing_voiceprint_keys,
-        _save_voiceprints_batch,
+    from think.entities import (
+        load_existing_voiceprint_keys,
+        save_voiceprints_batch,
     )
 
     (
@@ -612,7 +612,7 @@ def accumulate_voiceprints(
 
         # Idempotency check
         if speaker not in entity_existing:
-            entity_existing[speaker] = _load_existing_voiceprint_keys(speaker)
+            entity_existing[speaker] = load_existing_voiceprint_keys(speaker)
         vp_key = (day, segment_key, source, sid)
         if vp_key in entity_existing[speaker]:
             continue
@@ -630,7 +630,7 @@ def accumulate_voiceprints(
 
     for eid, items in entity_new.items():
         try:
-            count = _save_voiceprints_batch(eid, items)
+            count = save_voiceprints_batch(eid, items)
             saved_counts[eid] = count
         except Exception as exc:
             logger.warning("Failed to accumulate voiceprints for %s: %s", eid, exc)
diff --git a/apps/speakers/bootstrap.py b/apps/speakers/bootstrap.py
index d3533d923..b311da2b2 100644
--- a/apps/speakers/bootstrap.py
+++ b/apps/speakers/bootstrap.py
@@ -24,7 +24,6 @@ from __future__ import annotations
 import bisect
 import json
 import logging
-import shutil
 from collections import defaultdict
 from pathlib import Path
 from typing import Any
@@ -35,12 +34,11 @@ from apps.speakers.owner import load_owner_centroid
 from think.entities import entity_slug, find_matching_entity, is_name_variant_match
 from think.entities.journal import (
     create_journal_entity,
-    ensure_journal_entity_memory,
     load_all_journal_entities,
     load_journal_entity,
     save_journal_entity,
 )
-from think.utils import day_dirs, get_journal, iter_segments, now_ms, segment_path
+from think.utils import day_dirs, now_ms, segment_path
 
 logger = logging.getLogger(__name__)
 
@@ -48,124 +46,6 @@ logger = logging.getLogger(__name__)
 NAME_MERGE_THRESHOLD = 0.90
 
 
-def _routes_helpers():
-    """Load speakers route helpers lazily to avoid import cycles."""
-    from apps.speakers.routes import (
-        _load_embeddings_file,
-        _load_entity_voiceprints_file,
-        _normalize_embedding,
-        _scan_segment_embeddings,
-    )
-
-    return (
-        _load_embeddings_file,
-        _normalize_embedding,
-        _scan_segment_embeddings,
-        _load_entity_voiceprints_file,
-    )
-
-
-def _load_existing_voiceprint_keys(entity_id: str) -> set[tuple]:
-    """Load already-saved voiceprint keys for idempotency.
-
-    Returns set of (day, segment_key, source, sentence_id) tuples.
-    """
-    _, _, _, load_entity_voiceprints_file = _routes_helpers()
-
-    result = load_entity_voiceprints_file(entity_id)
-    if result is None:
-        return set()
-
-    _, metadata_list = result
-    return {
-        (m.get("day"), m.get("segment_key"), m.get("source"), m.get("sentence_id"))
-        for m in metadata_list
-    }
-
-
-def _save_voiceprints_batch(
-    entity_id: str,
-    new_items: list[tuple[np.ndarray, dict]],
-) -> int:
-    """Save multiple voiceprints to an entity's voiceprints.npz in one write.
-
-    Args:
-        entity_id: Entity ID (slug)
-        new_items: List of (normalized_embedding, metadata_dict) tuples
-
-    Returns:
-        Number of embeddings saved
-    """
-    if not new_items:
-        return 0
-
-    folder = ensure_journal_entity_memory(entity_id)
-    npz_path = folder / "voiceprints.npz"
-
-    # Load existing voiceprints
-    if npz_path.exists():
-        try:
-            # Use np.load with allow_pickle=False for safety, adjust if metadata requires it.
-            with np.load(npz_path, allow_pickle=False) as data:
-                existing_emb = data["embeddings"]
-                # Existing metadata was likely saved as JSON strings. Deserialize them.
-                # Assuming np.load returns an array of strings if saved as dtype=str.
-                existing_meta_strings = data["metadata"]
-                existing_meta_dicts = [json.loads(m) for m in existing_meta_strings]
-        except (FileNotFoundError, ValueError, np.lib.npyio.NpzFile) as e:
-            logger.warning(
-                f"Failed to load existing voiceprints for {entity_id} from {npz_path}: {e}. Starting fresh."
-            )
-            existing_emb = np.empty((0, 256), dtype=np.float32)
-            existing_meta_dicts = []
-        except Exception as e:  # Catch other potential errors during loading
-            logger.error(
-                f"Unexpected error loading existing voiceprints for {entity_id} from {npz_path}: {e}"
-            )
-            raise
-    else:
-        existing_emb = np.empty((0, 256), dtype=np.float32)
-        existing_meta_dicts = []
-
-    # Prepare new embeddings and metadata dicts
-    new_emb_list = []
-    new_meta_dicts = []
-    for emb, meta_dict in new_items:
-        new_emb_list.append(emb.reshape(1, -1).astype(np.float32))
-        new_meta_dicts.append(meta_dict)
-
-    # Combine existing and new data
-    if new_emb_list:
-        new_emb_np = np.vstack(new_emb_list)
-        combined_emb = (
-            np.vstack([existing_emb, new_emb_np])
-            if len(existing_emb) > 0
-            else new_emb_np
-        )
-        # Combine the metadata dictionaries
-        combined_meta_dicts = existing_meta_dicts + new_meta_dicts
-    else:  # Should not happen if new_items is not empty, but for safety
-        combined_emb = existing_emb
-        combined_meta_dicts = existing_meta_dicts
-
-    # Use the new safe saving utility
-    try:
-        # Import the utility function
-        from apps.speakers.voiceprint_io import save_voiceprints_safely
-
-        save_voiceprints_safely(
-            npz_path=npz_path,
-            embeddings=combined_emb,
-            metadata=combined_meta_dicts,  # Pass metadata as a list of dicts
-        )
-        return len(new_items)
-    except Exception as e:
-        logger.error(f"Failed to safely save voiceprints for {entity_id}: {e}")
-        # The save_voiceprints_safely function already logs critical errors and re-raises.
-        # We re-raise here to propagate the failure.
-        raise
-
-
 def bootstrap_voiceprints(dry_run: bool = False) -> dict[str, Any]:
     """Bootstrap voiceprints from 1-listed-speaker segments across the full journal.
 
@@ -186,12 +66,15 @@ def bootstrap_voiceprints(dry_run: bool = False) -> dict[str, Any]:
     Returns:
         Dict with statistics about the bootstrap run
     """
-    (
-        load_embeddings_file,
+    from apps.speakers.routes import _load_embeddings_file, _scan_segment_embeddings
+    from think.entities import (
+        load_existing_voiceprint_keys,
         normalize_embedding,
-        scan_segment_embeddings,
-        _,
-    ) = _routes_helpers()
+        save_voiceprints_batch,
+    )
+
+    load_embeddings_file = _load_embeddings_file
+    scan_segment_embeddings = _scan_segment_embeddings
 
     # Load owner centroid — required for owner subtraction
     centroid_data = load_owner_centroid()
@@ -262,7 +145,7 @@ def bootstrap_voiceprints(dry_run: bool = False) -> dict[str, Any]:
 
             # Load existing voiceprint keys for idempotency (once per entity)
             if entity_id not in entity_existing:
-                entity_existing[entity_id] = _load_existing_voiceprint_keys(entity_id)
+                entity_existing[entity_id] = load_existing_voiceprint_keys(entity_id)
 
             existing_keys = entity_existing[entity_id]
             seg_dir = segment_path(day, seg_key, stream)
@@ -320,7 +203,7 @@ def bootstrap_voiceprints(dry_run: bool = False) -> dict[str, Any]:
     if not dry_run:
         for entity_id, emb_list in entity_embeddings.items():
             try:
-                saved = _save_voiceprints_batch(entity_id, emb_list)
+                saved = save_voiceprints_batch(entity_id, emb_list)
                 stats["embeddings_saved"] += saved
             except Exception as e:
                 name = entity_names.get(entity_id, entity_id)
@@ -333,26 +216,12 @@ def bootstrap_voiceprints(dry_run: bool = False) -> dict[str, Any]:
 
 
 def merge_names(alias_name: str, canonical_name: str) -> dict[str, Any]:
-    """Deep merge a speaker entity into a canonical entity.
-
-    Performs a phased deep merge: identity data, voiceprints, facet
-    relationships, speaker references, then deletes the alias entity.
-    Designed for interrupt safety — delete-last ordering ensures the
-    system is never in an unrecoverable state. Every phase is idempotent.
-
-    Args:
-        alias_name: The alias/variant name to merge from
-        canonical_name: The canonical/full name to merge into
-
-    Returns:
-        Dict with merge statistics or error
-    """
-    _, normalize_embedding, _, load_entity_voiceprints_file = _routes_helpers()
+    """Deep merge a speaker entity into a canonical entity."""
+    from think.entities import merge_entity
 
     journal_entities = load_all_journal_entities()
     entities_list = list(journal_entities.values())
 
-    # --- Phase 0: Resolve and validate ---
     alias_entity = find_matching_entity(alias_name, entities_list)
     if not alias_entity:
         return {"error": f"No entity found for alias: {alias_name}"}
@@ -377,254 +246,39 @@ def merge_names(alias_name: str, canonical_name: str) -> dict[str, Any]:
 
     if alias.get("is_principal") or canonical.get("is_principal"):
         return {"error": "Cannot merge the principal entity."}
-    if alias.get("blocked"):
-        return {"error": f"Cannot merge blocked entity: {alias_id}"}
-    if canonical.get("blocked"):
-        return {"error": f"Cannot merge blocked entity: {canonical_id}"}
-
-    alias_display = alias.get("name", alias_name)
-
-    # Set merged_into resume marker on alias
-    alias["merged_into"] = canonical_id
-    alias["updated_at"] = now_ms()
-    save_journal_entity(alias)
-
-    # --- Phase 1: Merge identity data ---
-    akas_added: list[str] = []
-    existing_aka = set(canonical.get("aka", []))
-    canonical_name_val = canonical.get("name", "")
-
-    # Add alias display name as aka
-    if alias_display not in existing_aka and alias_display != canonical_name_val:
-        existing_aka.add(alias_display)
-        akas_added.append(alias_display)
-
-    # Merge alias's akas
-    for aka in alias.get("aka", []):
-        if aka not in existing_aka and aka != canonical_name_val:
-            existing_aka.add(aka)
-            akas_added.append(aka)
-
-    canonical["aka"] = sorted(existing_aka)
-
-    # Merge emails
-    canonical_emails = {e.lower() for e in canonical.get("emails", [])}
-    for email in alias.get("emails", []):
-        canonical_emails.add(email.lower())
-    if canonical_emails:
-        canonical["emails"] = sorted(canonical_emails)
-
-    canonical["updated_at"] = now_ms()
-    save_journal_entity(canonical)
-
-    # --- Phase 2: Merge voiceprints ---
-    alias_vp = load_entity_voiceprints_file(alias_id)
-    voiceprints_merged = 0
-
-    if alias_vp is not None:
-        alias_embeddings, alias_metadata = alias_vp
-        existing_keys = _load_existing_voiceprint_keys(canonical_id)
-
-        new_items: list[tuple[np.ndarray, dict]] = []
-        for emb, meta in zip(alias_embeddings, alias_metadata):
-            key = (
-                meta.get("day"),
-                meta.get("segment_key"),
-                meta.get("source"),
-                meta.get("sentence_id"),
-            )
-            if key in existing_keys:
-                continue
-            normalized = normalize_embedding(emb)
-            if normalized is not None:
-                new_items.append((normalized, meta))
-                existing_keys.add(key)
-
-        if new_items:
-            voiceprints_merged = _save_voiceprints_batch(canonical_id, new_items)
-
-    canonical_vp = load_entity_voiceprints_file(canonical_id)
-    voiceprints_total = len(canonical_vp[0]) if canonical_vp else 0
-
-    # --- Phase 3: Merge facet relationships ---
-    facets_merged: list[str] = []
-    facets_moved: list[str] = []
-    journal = get_journal()
-    facets_dir = Path(journal) / "facets"
-
-    if facets_dir.exists():
-        for facet_entry in sorted(facets_dir.iterdir()):
-            if not facet_entry.is_dir():
-                continue
-            facet_name = facet_entry.name
-            alias_rel_dir = facet_entry / "entities" / alias_id
-            alias_rel_path = alias_rel_dir / "entity.json"
-            if not alias_rel_path.is_file():
-                continue
-
-            canonical_rel_dir = facet_entry / "entities" / canonical_id
-            canonical_rel_path = canonical_rel_dir / "entity.json"
-
-            if not canonical_rel_path.is_file():
-                # Move: rename alias relationship dir to canonical
-                if canonical_rel_dir.exists():
-                    shutil.rmtree(canonical_rel_dir)
-                alias_rel_dir.rename(canonical_rel_dir)
-                # Update entity_id inside the moved entity.json
-                moved_path = canonical_rel_dir / "entity.json"
-                try:
-                    with open(moved_path, encoding="utf-8") as f:
-                        rel_data = json.load(f)
-                    rel_data["entity_id"] = canonical_id
-                    tmp = moved_path.with_suffix(".tmp")
-                    with open(tmp, "w", encoding="utf-8") as f:
-                        json.dump(rel_data, f, ensure_ascii=False, indent=2)
-                        f.write("\n")
-                    tmp.rename(moved_path)
-                except (json.JSONDecodeError, OSError):
-                    pass
-                facets_moved.append(facet_name)
-            else:
-                # Both have relationships: merge timestamps and data
-                try:
-                    with open(alias_rel_path, encoding="utf-8") as f:
-                        alias_rel = json.load(f)
-                    with open(canonical_rel_path, encoding="utf-8") as f:
-                        canonical_rel = json.load(f)
-                except (json.JSONDecodeError, OSError):
-                    continue
-
-                # Merge timestamps: earliest attached_at
-                alias_attached = alias_rel.get("attached_at")
-                canonical_attached = canonical_rel.get("attached_at")
-                if alias_attached and (
-                    not canonical_attached or alias_attached < canonical_attached
-                ):
-                    canonical_rel["attached_at"] = alias_attached
-
-                # Latest updated_at and last_seen
-                for ts_field in ("updated_at", "last_seen"):
-                    alias_ts = alias_rel.get(ts_field)
-                    canonical_ts = canonical_rel.get(ts_field)
-                    if alias_ts and (not canonical_ts or alias_ts > canonical_ts):
-                        canonical_rel[ts_field] = alias_ts
-
-                # Merge description: keep canonical's if non-empty
-                if not canonical_rel.get("description") and alias_rel.get(
-                    "description"
-                ):
-                    canonical_rel["description"] = alias_rel["description"]
-
-                # Save merged relationship (atomic write)
-                canonical_rel["entity_id"] = canonical_id
-                content = json.dumps(canonical_rel, ensure_ascii=False, indent=2) + "\n"
-                tmp = canonical_rel_path.with_suffix(".tmp")
-                with open(tmp, "w", encoding="utf-8") as f:
-                    f.write(content)
-                tmp.rename(canonical_rel_path)
-
-                # Merge observations: append alias's to canonical's
-                alias_obs_path = alias_rel_dir / "observations.jsonl"
-                if alias_obs_path.exists():
-                    alias_obs = alias_obs_path.read_text(encoding="utf-8")
-                    if alias_obs.strip():
-                        canonical_obs_path = canonical_rel_dir / "observations.jsonl"
-                        existing_obs = ""
-                        if canonical_obs_path.exists():
-                            existing_obs = canonical_obs_path.read_text(
-                                encoding="utf-8"
-                            )
-                        with open(canonical_obs_path, "a", encoding="utf-8") as f:
-                            if existing_obs and not existing_obs.endswith("\n"):
-                                f.write("\n")
-                            f.write(alias_obs)
-                            if not alias_obs.endswith("\n"):
-                                f.write("\n")
-
-                # Delete alias relationship directory
-                shutil.rmtree(alias_rel_dir)
-                facets_merged.append(facet_name)
-
-    # --- Phase 4: Rewrite speaker references ---
-    segments_scanned = 0
-    labels_rewritten = 0
-    corrections_rewritten = 0
-    errors: list[str] = []
-    alias_id_bytes = alias_id.encode("utf-8")
-
-    for day in sorted(day_dirs().keys()):
-        for _stream, _seg_key, seg_path in iter_segments(day):
-            segments_scanned += 1
-            agents_dir = seg_path / "talents"
-
-            # Rewrite speaker_labels.json
-            labels_path = agents_dir / "speaker_labels.json"
-            if labels_path.is_file():
-                try:
-                    raw = labels_path.read_bytes()
-                    if alias_id_bytes in raw:
-                        data = json.loads(raw)
-                        changed = False
-                        for label in data.get("labels", []):
-                            if label.get("speaker") == alias_id:
-                                label["speaker"] = canonical_id
-                                changed = True
-                        if changed:
-                            tmp = labels_path.with_suffix(".tmp")
-                            with open(tmp, "w", encoding="utf-8") as f:
-                                json.dump(data, f, indent=2)
-                            tmp.rename(labels_path)
-                            labels_rewritten += 1
-                except Exception as e:
-                    errors.append(f"{labels_path}: {e}")
-
-            # Rewrite speaker_corrections.json
-            corrections_path = agents_dir / "speaker_corrections.json"
-            if corrections_path.is_file():
-                try:
-                    raw = corrections_path.read_bytes()
-                    if alias_id_bytes in raw:
-                        data = json.loads(raw)
-                        changed = False
-                        for correction in data.get("corrections", []):
-                            if correction.get("original_speaker") == alias_id:
-                                correction["original_speaker"] = canonical_id
-                                changed = True
-                            if correction.get("corrected_speaker") == alias_id:
-                                correction["corrected_speaker"] = canonical_id
-                                changed = True
-                        if changed:
-                            tmp = corrections_path.with_suffix(".tmp")
-                            with open(tmp, "w", encoding="utf-8") as f:
-                                json.dump(data, f, indent=2)
-                            tmp.rename(corrections_path)
-                            corrections_rewritten += 1
-                except Exception as e:
-                    errors.append(f"{corrections_path}: {e}")
-
-    # --- Phase 5: Cleanup ---
-    alias_entity_dir = Path(journal) / "entities" / alias_id
-    if alias_entity_dir.exists():
-        shutil.rmtree(alias_entity_dir)
-
-    discovery_cache = Path(journal) / "awareness" / "discovery_clusters.json"
-    if discovery_cache.exists():
-        discovery_cache.unlink()
+    result = merge_entity(
+        alias_id,
+        canonical_id,
+        keep_source_as_aka=True,
+        commit=True,
+        caller="speakers.merge_names",
+    )
+    if "error" in result:
+        return result
+
+    errors = []
+    for error in result["segments"]["errors"]:
+        path = error.get("path")
+        message = error.get("message", "")
+        if path:
+            errors.append(f"{path}: {message}")
+        else:
+            errors.append(str(message))
 
     return {
         "merged": True,
-        "alias": alias_display,
+        "alias": alias.get("name", alias_name),
         "alias_id": alias_id,
         "canonical_name": canonical.get("name", canonical_name),
         "canonical_id": canonical_id,
-        "akas_added": akas_added,
-        "voiceprints_merged": voiceprints_merged,
-        "voiceprints_total": voiceprints_total,
-        "facets_merged": facets_merged,
-        "facets_moved": facets_moved,
-        "segments_scanned": segments_scanned,
-        "labels_rewritten": labels_rewritten,
-        "corrections_rewritten": corrections_rewritten,
+        "akas_added": result["identity"]["akas_added"],
+        "voiceprints_merged": result["voiceprints"]["added"],
+        "voiceprints_total": result["voiceprints"]["target_total"],
+        "facets_merged": result["facets"]["merged"],
+        "facets_moved": result["facets"]["moved"],
+        "segments_scanned": result["segments"]["files_scanned"],
+        "labels_rewritten": result["segments"]["labels_rewritten"],
+        "corrections_rewritten": result["segments"]["corrections_rewritten"],
         "errors": errors,
     }
 
@@ -649,7 +303,10 @@ def resolve_name_variants(dry_run: bool = False) -> dict[str, Any]:
     Returns:
         Dict with merge statistics
     """
-    _, normalize_embedding, _, load_entity_voiceprints_file = _routes_helpers()
+    from think.entities import (
+        load_entity_voiceprints_file,
+        normalize_embedding,
+    )
 
     journal_entities = load_all_journal_entities()
 
@@ -909,12 +566,15 @@ def link_import(name: str, entity_id: str) -> dict[str, Any]:
 
 def seed_from_imports(dry_run: bool = False) -> dict[str, Any]:
     """Seed voiceprints from import segments with speaker-attributed transcripts."""
-    (
-        load_embeddings_file,
+    from apps.speakers.routes import _load_embeddings_file, _scan_segment_embeddings
+    from think.entities import (
+        load_existing_voiceprint_keys,
         normalize_embedding,
-        scan_segment_embeddings,
-        _,
-    ) = _routes_helpers()
+        save_voiceprints_batch,
+    )
+
+    load_embeddings_file = _load_embeddings_file
+    scan_segment_embeddings = _scan_segment_embeddings
 
     centroid_data = load_owner_centroid()
     if centroid_data is None:
@@ -1015,7 +675,7 @@ def seed_from_imports(dry_run: bool = False) -> dict[str, Any]:
                     stats["speakers_found"].setdefault(entity_name, 0)
 
                     if entity_id not in entity_existing:
-                        entity_existing[entity_id] = _load_existing_voiceprint_keys(
+                        entity_existing[entity_id] = load_existing_voiceprint_keys(
                             entity_id
                         )
 
@@ -1060,7 +720,7 @@ def seed_from_imports(dry_run: bool = False) -> dict[str, Any]:
     if not dry_run:
         for entity_id, emb_list in entity_embeddings.items():
             try:
-                saved = _save_voiceprints_batch(entity_id, emb_list)
+                saved = save_voiceprints_batch(entity_id, emb_list)
                 stats["embeddings_saved"] += saved
             except Exception as e:
                 stats["errors"].append(f"Failed to save for {entity_id}: {e}")
diff --git a/apps/speakers/discovery.py b/apps/speakers/discovery.py
index d86c95af2..206a9f343 100644
--- a/apps/speakers/discovery.py
+++ b/apps/speakers/discovery.py
@@ -48,16 +48,6 @@ def _routes_helpers():
     )
 
 
-def _bootstrap_helpers():
-    """Load bootstrap helpers lazily to avoid import cycles."""
-    from apps.speakers.bootstrap import (
-        _load_existing_voiceprint_keys,
-        _save_voiceprints_batch,
-    )
-
-    return _load_existing_voiceprint_keys, _save_voiceprints_batch
-
-
 def _owner_helpers():
     """Load owner helpers lazily to avoid import cycles."""
     from apps.speakers.owner import load_owner_centroid
@@ -315,6 +305,11 @@ def identify_cluster(
     cluster_id: int, name: str, entity_id: str | None = None
 ) -> dict[str, Any]:
     """Identify a discovered unknown speaker cluster."""
+    from think.entities import (
+        load_existing_voiceprint_keys,
+        save_voiceprints_batch,
+    )
+
     (
         load_embeddings_file,
         load_speaker_labels,
@@ -324,7 +319,6 @@ def identify_cluster(
         append_speaker_correction,
         check_owner_contamination,
     ) = _routes_helpers()
-    load_existing_voiceprint_keys, save_voiceprints_batch = _bootstrap_helpers()
 
     cache_path = _discovery_cache_path()
     if not cache_path.exists():
diff --git a/apps/speakers/routes.py b/apps/speakers/routes.py
index 91d3cbae6..cb16a9bd7 100644
--- a/apps/speakers/routes.py
+++ b/apps/speakers/routes.py
@@ -68,12 +68,9 @@ speakers_bp = Blueprint(
 
 
 def _normalize_embedding(emb: np.ndarray) -> np.ndarray | None:
-    """L2-normalize an embedding vector. Returns None if norm is zero."""
-    emb = emb.astype(np.float32)
-    norm = np.linalg.norm(emb)
-    if norm > 0:
-        return emb / norm
-    return None
+    from think.entities import normalize_embedding
+
+    return normalize_embedding(emb)
 
 
 def _parse_time_to_seconds(time_str: str) -> int:
@@ -140,42 +137,9 @@ def _load_segment_speakers(segment_dir: Path) -> list[str]:
 def _load_entity_voiceprints_file(
     entity_id: str,
 ) -> tuple[np.ndarray, list[dict]] | None:
-    """Load voiceprints for an entity from journal-level voiceprints.npz.
-
-    Voiceprints are stored at the journal level (entities/<id>/voiceprints.npz)
-    since a person's voice is the same across all facets.
-
-    Args:
-        entity_id: Entity ID (slug)
-
-    Returns:
-        Tuple of (embeddings, metadata_list) or None if not found.
-        - embeddings: (N, 256) float32 array
-        - metadata_list: List of dicts parsed from JSON metadata strings
-    """
-    try:
-        folder = journal_entity_memory_path(entity_id)
-    except (RuntimeError, ValueError):
-        return None
+    from think.entities import load_entity_voiceprints_file
 
-    npz_path = folder / "voiceprints.npz"
-    if not npz_path.exists():
-        return None
-
-    try:
-        data = np.load(npz_path, allow_pickle=False)
-        embeddings = data.get("embeddings")
-        metadata_arr = data.get("metadata")
-
-        if embeddings is None or metadata_arr is None:
-            return None
-
-        # Parse JSON metadata strings
-        metadata_list = [json.loads(m) for m in metadata_arr]
-        return embeddings, metadata_list
-    except Exception as e:
-        logger.warning("Failed to load voiceprints for entity %s: %s", entity_id, e)
-        return None
+    return load_entity_voiceprints_file(entity_id)
 
 
 def _save_voiceprint(
diff --git a/apps/speakers/tests/conftest.py b/apps/speakers/tests/conftest.py
index 32eaf1557..107dd5cda 100644
--- a/apps/speakers/tests/conftest.py
+++ b/apps/speakers/tests/conftest.py
@@ -6,12 +6,21 @@
 from __future__ import annotations
 
 import json
+import sys
 from pathlib import Path
 
 import numpy as np
 import pytest
 
+ROOT = Path(__file__).resolve().parents[3]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
 from think.entities import entity_slug
+from think.entities.journal import clear_journal_entity_cache
+from think.entities.loading import clear_entity_loading_cache
+from think.entities.observations import clear_observation_cache
+from think.entities.relationships import clear_relationship_caches
 
 # Default stream name for test fixtures
 STREAM = "test"
@@ -43,6 +52,32 @@ def speakers_env(tmp_path, monkeypatch):
         def __init__(self, journal_path: Path):
             self.journal = journal_path
             monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(journal_path))
+            monkeypatch.setenv("SOL_SKIP_SUPERVISOR_CHECK", "1")
+            clear_journal_entity_cache()
+            clear_entity_loading_cache()
+            clear_relationship_caches()
+            clear_observation_cache()
+            import think.utils
+
+            think.utils._journal_path_cache = None
+
+        def _segment_dirs(
+            self,
+            day: str,
+            segment_key: str,
+            *,
+            stream: str | None = None,
+        ) -> tuple[Path, Path]:
+            stream_name = stream or STREAM
+            chronicle_day = self.journal / "chronicle" / day
+            chronicle_day.mkdir(parents=True, exist_ok=True)
+            flat_day = self.journal / day
+            if not flat_day.exists():
+                flat_day.symlink_to(chronicle_day, target_is_directory=True)
+            flat_dir = flat_day / stream_name / segment_key
+            chronicle_dir = chronicle_day / stream_name / segment_key
+            chronicle_dir.mkdir(parents=True, exist_ok=True)
+            return flat_dir, chronicle_dir
 
         def create_segment(
             self,
@@ -64,16 +99,17 @@ def speakers_env(tmp_path, monkeypatch):
                 sources: List of audio sources (e.g., ["mic_audio", "sys_audio"])
                 num_sentences: Number of sentences to create
             """
-            segment_dir = self.journal / day / (stream or STREAM) / segment_key
-            segment_dir.mkdir(parents=True, exist_ok=True)
+            flat_dir, chronicle_dir = self._segment_dirs(
+                day,
+                segment_key,
+                stream=stream,
+            )
 
             sentence_count = (
                 embeddings.shape[0] if embeddings is not None else num_sentences
             )
 
             for source in sources:
-                # Create JSONL transcript
-                jsonl_path = segment_dir / f"{source}.jsonl"
                 lines = [json.dumps({"raw": f"{source}.flac", "model": "medium.en"})]
 
                 # Parse segment_key to get base time (e.g., "143022_300" -> 14:30:22)
@@ -98,10 +134,12 @@ def speakers_env(tmp_path, monkeypatch):
                             }
                         )
                     )
-                jsonl_path.write_text("\n".join(lines) + "\n")
+                for segment_dir in (flat_dir, chronicle_dir):
+                    (segment_dir / f"{source}.jsonl").write_text(
+                        "\n".join(lines) + "\n"
+                    )
 
                 # Create NPZ embeddings
-                npz_path = segment_dir / f"{source}.npz"
                 if embeddings is None:
                     source_embeddings = np.random.randn(sentence_count, 256).astype(
                         np.float32
@@ -111,17 +149,15 @@ def speakers_env(tmp_path, monkeypatch):
                 else:
                     source_embeddings = embeddings.astype(np.float32)
                 statement_ids = np.arange(1, sentence_count + 1, dtype=np.int32)
-                np.savez_compressed(
-                    npz_path,
-                    embeddings=source_embeddings,
-                    statement_ids=statement_ids,
-                )
-
-                # Create dummy audio file
-                audio_path = segment_dir / f"{source}.flac"
-                audio_path.write_bytes(b"")  # Empty placeholder
+                for segment_dir in (flat_dir, chronicle_dir):
+                    np.savez_compressed(
+                        segment_dir / f"{source}.npz",
+                        embeddings=source_embeddings,
+                        statement_ids=statement_ids,
+                    )
+                    (segment_dir / f"{source}.flac").write_bytes(b"")
 
-            return segment_dir
+            return flat_dir
 
         def create_embedding(self, vector: list[float] | None = None) -> np.ndarray:
             """Create a normalized 256-dim embedding."""
@@ -194,14 +230,17 @@ def speakers_env(tmp_path, monkeypatch):
                 segment_key: Segment key (HHMMSS_LEN)
                 speakers: List of speaker names
             """
-            agents_dir = self.journal / day / STREAM / segment_key / "talents"
-            agents_dir.mkdir(parents=True, exist_ok=True)
-
-            speakers_path = agents_dir / "speakers.json"
-            with open(speakers_path, "w", encoding="utf-8") as f:
-                json.dump(speakers, f)
-
-            return speakers_path
+            flat_dir, chronicle_dir = self._segment_dirs(day, segment_key)
+            paths = []
+            for segment_dir in (flat_dir, chronicle_dir):
+                agents_dir = segment_dir / "talents"
+                agents_dir.mkdir(parents=True, exist_ok=True)
+                speakers_path = agents_dir / "speakers.json"
+                with open(speakers_path, "w", encoding="utf-8") as f:
+                    json.dump(speakers, f)
+                paths.append(speakers_path)
+
+            return paths[0]
 
         def create_speaker_labels(
             self,
@@ -220,21 +259,23 @@ def speakers_env(tmp_path, monkeypatch):
                 metadata: Optional extra metadata (owner_centroid_version,
                     voiceprint_versions)
             """
-            agents_dir = self.journal / day / STREAM / segment_key / "talents"
-            agents_dir.mkdir(parents=True, exist_ok=True)
-
             data = {"labels": labels}
             if metadata:
                 data.update(metadata)
             else:
                 data["owner_centroid_version"] = None
                 data["voiceprint_versions"] = {}
-
-            labels_path = agents_dir / "speaker_labels.json"
-            with open(labels_path, "w", encoding="utf-8") as f:
-                json.dump(data, f)
-
-            return labels_path
+            flat_dir, chronicle_dir = self._segment_dirs(day, segment_key)
+            paths = []
+            for segment_dir in (flat_dir, chronicle_dir):
+                agents_dir = segment_dir / "talents"
+                agents_dir.mkdir(parents=True, exist_ok=True)
+                labels_path = agents_dir / "speaker_labels.json"
+                with open(labels_path, "w", encoding="utf-8") as f:
+                    json.dump(data, f)
+                paths.append(labels_path)
+
+            return paths[0]
 
         def create_speaker_corrections(
             self,
@@ -253,17 +294,22 @@ def speakers_env(tmp_path, monkeypatch):
                     original_speaker, corrected_speaker, timestamp
                 stream: Optional stream name (defaults to STREAM)
             """
-            agents_dir = (
-                self.journal / day / (stream or STREAM) / segment_key / "talents"
-            )
-            agents_dir.mkdir(parents=True, exist_ok=True)
-
             data = {"corrections": corrections}
-            corrections_path = agents_dir / "speaker_corrections.json"
-            with open(corrections_path, "w", encoding="utf-8") as f:
-                json.dump(data, f)
+            flat_dir, chronicle_dir = self._segment_dirs(
+                day,
+                segment_key,
+                stream=stream,
+            )
+            paths = []
+            for segment_dir in (flat_dir, chronicle_dir):
+                agents_dir = segment_dir / "talents"
+                agents_dir.mkdir(parents=True, exist_ok=True)
+                corrections_path = agents_dir / "speaker_corrections.json"
+                with open(corrections_path, "w", encoding="utf-8") as f:
+                    json.dump(data, f)
+                paths.append(corrections_path)
 
-            return corrections_path
+            return paths[0]
 
         def create_facet_relationship(
             self,
@@ -336,8 +382,11 @@ def speakers_env(tmp_path, monkeypatch):
                 stream: Import stream name (default: import.granola)
                 embeddings: Optional pre-built embeddings array (num_sentences x 256)
             """
-            segment_dir = self.journal / day / stream / segment_key
-            segment_dir.mkdir(parents=True, exist_ok=True)
+            flat_dir, chronicle_dir = self._segment_dirs(
+                day,
+                segment_key,
+                stream=stream,
+            )
 
             num_sentences = len(speakers)
 
@@ -366,8 +415,10 @@ def speakers_env(tmp_path, monkeypatch):
                         }
                     )
                 )
-            ct_path = segment_dir / "conversation_transcript.jsonl"
-            ct_path.write_text("\n".join(ct_lines) + "\n")
+            for segment_dir in (flat_dir, chronicle_dir):
+                (segment_dir / "conversation_transcript.jsonl").write_text(
+                    "\n".join(ct_lines) + "\n"
+                )
 
             audio_lines = [
                 json.dumps({"raw": "imported_audio.flac", "model": "medium.en"})
@@ -386,8 +437,10 @@ def speakers_env(tmp_path, monkeypatch):
                         }
                     )
                 )
-            audio_jsonl_path = segment_dir / "imported_audio.jsonl"
-            audio_jsonl_path.write_text("\n".join(audio_lines) + "\n")
+            for segment_dir in (flat_dir, chronicle_dir):
+                (segment_dir / "imported_audio.jsonl").write_text(
+                    "\n".join(audio_lines) + "\n"
+                )
 
             if embeddings is None:
                 source_embeddings = np.random.randn(num_sentences, 256).astype(
@@ -398,17 +451,24 @@ def speakers_env(tmp_path, monkeypatch):
             else:
                 source_embeddings = embeddings.astype(np.float32)
             statement_ids = np.arange(1, num_sentences + 1, dtype=np.int32)
-            np.savez_compressed(
-                segment_dir / "imported_audio.npz",
-                embeddings=source_embeddings,
-                statement_ids=statement_ids,
-            )
-
-            (segment_dir / "imported_audio.flac").write_bytes(b"")
+            for segment_dir in (flat_dir, chronicle_dir):
+                np.savez_compressed(
+                    segment_dir / "imported_audio.npz",
+                    embeddings=source_embeddings,
+                    statement_ids=statement_ids,
+                )
+                (segment_dir / "imported_audio.flac").write_bytes(b"")
 
-            return segment_dir
+            return flat_dir
 
     def _create():
         return SpeakersEnv(tmp_path)
 
-    return _create
+    yield _create
+    clear_journal_entity_cache()
+    clear_entity_loading_cache()
+    clear_relationship_caches()
+    clear_observation_cache()
+    import think.utils
+
+    think.utils._journal_path_cache = None
diff --git a/docs/coding-standards.md b/docs/coding-standards.md
index d53b1a956..9a79168bd 100644
--- a/docs/coding-standards.md
+++ b/docs/coding-standards.md
@@ -67,7 +67,7 @@ Each domain has exactly **one** write-owning module. No other module may call `a
 
 | Domain | Write-owning module(s) |
 |--------|------------------------|
-| Entities (`entities/*/entity.json`, `entities/*/*.npz`) | `think/entities/journal.py` + `think/entities/consolidation.py` + `think/entities/saving.py` + `apps/entities/call.py` |
+| Entities (`entities/*/entity.json`, `entities/*/*.npz`) | `think/entities/journal.py` + `think/entities/consolidation.py` + `think/entities/saving.py` + `think/entities/merge.py` + `apps/entities/call.py` |
 | Facets (`facets/*/facet.json`, `facets/*/relationships/`) | `think/facets.py` + `apps/facets/*` (if/when created) |
 | Observations (`observations.jsonl`) | `think/entities/observations.py` |
 | Activities (`facets/*/activities/*.jsonl`) | `think/activities.py` |
diff --git a/think/entities/__init__.py b/think/entities/__init__.py
index 2832a17d9..ee7935226 100644
--- a/think/entities/__init__.py
+++ b/think/entities/__init__.py
@@ -84,6 +84,7 @@ from think.entities.matching import (
     resolve_entity,
     validate_aka_uniqueness,
 )
+from think.entities.merge import merge_entity
 
 # Observations
 from think.entities.observations import (
@@ -110,6 +111,14 @@ from think.entities.saving import (
     save_entities,
     update_detected_entity,
 )
+from think.entities.voiceprints import (
+    load_entity_voiceprints_file,
+    load_existing_voiceprint_keys,
+    normalize_embedding,
+    save_voiceprints_batch,
+    save_voiceprints_safely,
+    voiceprint_file_path,
+)
 
 __all__ = [
     # Core
@@ -150,11 +159,15 @@ __all__ = [
     "load_entities",
     "load_entity_names",
     "load_recent_entity_names",
+    "merge_entity",
     "parse_entity_file",
     # Saving
     "save_detected_entity",
     "save_entities",
+    "save_voiceprints_batch",
+    "save_voiceprints_safely",
     "update_detected_entity",
+    "voiceprint_file_path",
     # Matching
     "MatchResult",
     "MatchTier",
@@ -174,6 +187,9 @@ __all__ = [
     "load_observations",
     "observations_file_path",
     "save_observations",
+    "load_entity_voiceprints_file",
+    "load_existing_voiceprint_keys",
+    "normalize_embedding",
     # Formatting
     "format_entities",
     "format_observations",
diff --git a/think/entities/merge.py b/think/entities/merge.py
new file mode 100644
index 000000000..3c7351a2e
--- /dev/null
+++ b/think/entities/merge.py
@@ -0,0 +1,726 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Journal-entity merge primitive."""
+
+from __future__ import annotations
+
+import json
+import shutil
+from pathlib import Path
+from typing import Any
+
+from think.entities.journal import (
+    clear_journal_entity_cache,
+    load_journal_entity,
+    save_journal_entity,
+    scan_journal_entities,
+)
+from think.entities.loading import clear_entity_loading_cache
+from think.entities.observations import clear_observation_cache, save_observations
+from think.entities.relationships import (
+    clear_relationship_caches,
+    save_facet_relationship,
+)
+from think.entities.voiceprints import (
+    load_entity_voiceprints_file,
+    load_existing_voiceprint_keys,
+    normalize_embedding,
+    save_voiceprints_batch,
+)
+from think.utils import day_dirs, get_journal, iter_segments, now_ms
+
+
+def _dedupe_akas(target_values: list[Any], source_values: list[Any]) -> list[str]:
+    """Case-insensitive aka dedup, preserving first-seen spelling."""
+    aka_by_lower: dict[str, str] = {}
+    for values in (target_values, source_values):
+        if not isinstance(values, list):
+            continue
+        for value in values:
+            if not value:
+                continue
+            key = str(value).lower()
+            if key not in aka_by_lower:
+                aka_by_lower[key] = str(value)
+    return sorted(aka_by_lower.values(), key=str.lower)
+
+
+def _dedupe_emails(target_values: list[Any], source_values: list[Any]) -> list[str]:
+    """Case-insensitive email dedup, preserving first-seen order/spelling."""
+    merged_emails: list[str] = []
+    seen_emails: set[str] = set()
+    for values in (target_values, source_values):
+        if not isinstance(values, list):
+            continue
+        for value in values:
+            if not value:
+                continue
+            email = str(value)
+            key = email.lower()
+            if key in seen_emails:
+                continue
+            seen_emails.add(key)
+            merged_emails.append(email)
+    return merged_emails
+
+
+def _dedupe_observations(
+    source_observations: list[dict[str, Any]],
+    target_observations: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Deduplicate observations on (content, observed_at)."""
+    seen = {
+        (item.get("content", ""), item.get("observed_at"))
+        for item in target_observations
+    }
+    merged_observations = list(target_observations)
+    for item in source_observations:
+        key = (item.get("content", ""), item.get("observed_at"))
+        if key in seen:
+            continue
+        seen.add(key)
+        merged_observations.append(item)
+    return merged_observations
+
+
+def _read_jsonl(path: Path) -> list[dict[str, Any]]:
+    if not path.is_file():
+        return []
+    rows: list[dict[str, Any]] = []
+    try:
+        with open(path, encoding="utf-8") as handle:
+            for line in handle:
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    rows.append(json.loads(line))
+                except json.JSONDecodeError:
+                    continue
+    except OSError:
+        return []
+    return rows
+
+
+def _identity_section(
+    akas_added: list[str],
+    emails_added: list[str],
+    principal_transferred: bool,
+) -> dict[str, Any]:
+    return {
+        "akas_added": akas_added,
+        "akas_added_count": len(akas_added),
+        "emails_added": emails_added,
+        "emails_added_count": len(emails_added),
+        "principal_transferred": principal_transferred,
+    }
+
+
+def _voiceprint_section(
+    added: int, skipped_duplicate: int, target_total: int
+) -> dict[str, Any]:
+    return {
+        "added": added,
+        "skipped_duplicate": skipped_duplicate,
+        "target_total": target_total,
+    }
+
+
+def _facet_section(
+    moved: list[str],
+    merged: list[str],
+    observations_appended: int,
+) -> dict[str, Any]:
+    return {
+        "moved": moved,
+        "moved_count": len(moved),
+        "merged": merged,
+        "merged_count": len(merged),
+        "observations_appended": observations_appended,
+    }
+
+
+def _segment_section(
+    labels_rewritten: int,
+    corrections_rewritten: int,
+    files_scanned: int,
+    errors: list[dict[str, Any]],
+) -> dict[str, Any]:
+    return {
+        "labels_rewritten": labels_rewritten,
+        "corrections_rewritten": corrections_rewritten,
+        "files_scanned": files_scanned,
+        "errors": errors,
+    }
+
+
+def _empty_result_section() -> dict[str, Any]:
+    return {
+        "identity": _identity_section([], [], False),
+        "voiceprints": _voiceprint_section(0, 0, 0),
+        "facets": _facet_section([], [], 0),
+        "segments": _segment_section(0, 0, 0, []),
+    }
+
+
+def _is_missing_value(value: Any) -> bool:
+    return value in (None, "", [], {})
+
+
+def _plan_resume_marker(
+    source_entity: dict[str, Any],
+    target_id: str,
+    *,
+    principal_transferred: bool,
+) -> dict[str, Any]:
+    updated = dict(source_entity)
+    updated["merged_into"] = target_id
+    updated["updated_at"] = now_ms()
+    if principal_transferred:
+        updated.pop("is_principal", None)
+    return updated
+
+
+def _plan_identity_merge(
+    source_entity: dict[str, Any],
+    target_entity: dict[str, Any],
+    *,
+    keep_source_as_aka: bool,
+) -> tuple[dict[str, Any], dict[str, Any]]:
+    target_after = dict(target_entity)
+    source_display = str(source_entity.get("name", source_entity.get("id", "")))
+    target_name = str(target_entity.get("name", ""))
+
+    target_akas = target_entity.get("aka", [])
+    if not isinstance(target_akas, list):
+        target_akas = []
+
+    source_aka_values = source_entity.get("aka", [])
+    if not isinstance(source_aka_values, list):
+        source_aka_values = []
+
+    aka_candidates: list[str] = []
+    if keep_source_as_aka and source_display and source_display != target_name:
+        aka_candidates.append(source_display)
+    aka_candidates.extend(str(value) for value in source_aka_values if value)
+
+    target_aka_keys = {str(value).lower() for value in target_akas if value}
+    added_akas: list[str] = []
+    seen_added_akas: set[str] = set()
+    for value in aka_candidates:
+        key = value.lower()
+        if key in target_aka_keys or key in seen_added_akas:
+            continue
+        seen_added_akas.add(key)
+        added_akas.append(value)
+
+    merged_akas = _dedupe_akas(target_akas, aka_candidates)
+    if merged_akas:
+        target_after["aka"] = merged_akas
+
+    target_emails = target_entity.get("emails", [])
+    if not isinstance(target_emails, list):
+        target_emails = []
+    source_emails = source_entity.get("emails", [])
+    if not isinstance(source_emails, list):
+        source_emails = []
+
+    target_email_keys = {str(value).lower() for value in target_emails if value}
+    added_emails: list[str] = []
+    seen_added_emails: set[str] = set()
+    for value in source_emails:
+        if not value:
+            continue
+        email = str(value)
+        key = email.lower()
+        if key in target_email_keys or key in seen_added_emails:
+            continue
+        seen_added_emails.add(key)
+        added_emails.append(email)
+
+    merged_emails = _dedupe_emails(target_emails, source_emails)
+    if merged_emails:
+        target_after["emails"] = merged_emails
+
+    principal_transferred = bool(
+        source_entity.get("is_principal") and not target_entity.get("is_principal")
+    )
+    if principal_transferred:
+        target_after["is_principal"] = True
+
+    for key, value in source_entity.items():
+        if key in {
+            "id",
+            "name",
+            "aka",
+            "emails",
+            "created_at",
+            "updated_at",
+            "merged_into",
+            "blocked",
+            "is_principal",
+        }:
+            continue
+        if _is_missing_value(target_after.get(key)) and not _is_missing_value(value):
+            target_after[key] = value
+
+    target_after["updated_at"] = now_ms()
+    return target_after, _identity_section(
+        added_akas, added_emails, principal_transferred
+    )
+
+
+def _plan_voiceprint_merge(source_id: str, target_id: str) -> dict[str, Any]:
+    source_vp = load_entity_voiceprints_file(source_id)
+    target_vp = load_entity_voiceprints_file(target_id)
+    existing_keys = load_existing_voiceprint_keys(target_id)
+
+    new_items: list[tuple[Any, dict[str, Any]]] = []
+    skipped_duplicate = 0
+    if source_vp is not None:
+        source_embeddings, source_metadata = source_vp
+        for emb, meta in zip(source_embeddings, source_metadata):
+            key = (
+                meta.get("day"),
+                meta.get("segment_key"),
+                meta.get("source"),
+                meta.get("sentence_id"),
+            )
+            if key in existing_keys:
+                skipped_duplicate += 1
+                continue
+            normalized = normalize_embedding(emb)
+            if normalized is None:
+                continue
+            new_items.append((normalized, meta))
+            existing_keys.add(key)
+
+    target_existing_total = len(target_vp[0]) if target_vp else 0
+    added = len(new_items)
+    return {
+        "items": new_items,
+        "section": _voiceprint_section(
+            added=added,
+            skipped_duplicate=skipped_duplicate,
+            target_total=target_existing_total + added,
+        ),
+    }
+
+
+def _plan_facet_merge(source_id: str, target_id: str) -> dict[str, Any]:
+    journal = Path(get_journal())
+    facets_dir = journal / "facets"
+    operations: list[dict[str, Any]] = []
+    moved: list[str] = []
+    merged: list[str] = []
+    observations_appended = 0
+
+    if not facets_dir.exists():
+        return {
+            "operations": operations,
+            "section": _facet_section(moved, merged, observations_appended),
+        }
+
+    for facet_entry in sorted(facets_dir.iterdir()):
+        if not facet_entry.is_dir():
+            continue
+        facet_name = facet_entry.name
+        source_rel_dir = facet_entry / "entities" / source_id
+        source_rel_path = source_rel_dir / "entity.json"
+        if not source_rel_path.is_file():
+            continue
+
+        target_rel_dir = facet_entry / "entities" / target_id
+        target_rel_path = target_rel_dir / "entity.json"
+
+        if not target_rel_path.is_file():
+            operations.append(
+                {
+                    "kind": "move",
+                    "facet": facet_name,
+                    "source_rel_dir": source_rel_dir,
+                    "target_rel_dir": target_rel_dir,
+                }
+            )
+            moved.append(facet_name)
+            continue
+
+        try:
+            with open(source_rel_path, encoding="utf-8") as handle:
+                source_rel = json.load(handle)
+            with open(target_rel_path, encoding="utf-8") as handle:
+                target_rel = json.load(handle)
+        except (json.JSONDecodeError, OSError):
+            continue
+
+        merged_rel = dict(target_rel)
+        source_attached = source_rel.get("attached_at")
+        target_attached = merged_rel.get("attached_at")
+        if source_attached and (
+            not target_attached or source_attached < target_attached
+        ):
+            merged_rel["attached_at"] = source_attached
+
+        for field in ("updated_at", "last_seen"):
+            source_ts = source_rel.get(field)
+            target_ts = merged_rel.get(field)
+            if source_ts and (not target_ts or source_ts > target_ts):
+                merged_rel[field] = source_ts
+
+        if not merged_rel.get("description") and source_rel.get("description"):
+            merged_rel["description"] = source_rel["description"]
+
+        source_obs = _read_jsonl(source_rel_dir / "observations.jsonl")
+        target_obs = _read_jsonl(target_rel_dir / "observations.jsonl")
+        merged_obs = _dedupe_observations(source_obs, target_obs)
+        observations_added = len(merged_obs) - len(target_obs)
+        observations_appended += observations_added
+
+        operations.append(
+            {
+                "kind": "merge",
+                "facet": facet_name,
+                "source_rel_dir": source_rel_dir,
+                "target_rel_dir": target_rel_dir,
+                "relationship": merged_rel,
+                "observations": merged_obs,
+                "observations_added": observations_added,
+            }
+        )
+        merged.append(facet_name)
+
+    return {
+        "operations": operations,
+        "section": _facet_section(moved, merged, observations_appended),
+    }
+
+
+def _plan_segment_rewrites(source_id: str, target_id: str) -> dict[str, Any]:
+    labels_rewritten = 0
+    corrections_rewritten = 0
+    files_scanned = 0
+    errors: list[dict[str, Any]] = []
+    operations: list[dict[str, Any]] = []
+    source_id_bytes = source_id.encode("utf-8")
+
+    for day_path in _segment_day_dirs():
+        for _stream, _seg_key, seg_path in iter_segments(day_path):
+            files_scanned += 1
+            talents_dir = seg_path / "talents"
+
+            labels_path = talents_dir / "speaker_labels.json"
+            if labels_path.is_file():
+                try:
+                    raw = labels_path.read_bytes()
+                    if source_id_bytes in raw:
+                        data = json.loads(raw)
+                        changed = False
+                        for label in data.get("labels", []):
+                            if label.get("speaker") == source_id:
+                                label["speaker"] = target_id
+                                changed = True
+                        if changed:
+                            labels_rewritten += 1
+                            operations.append(
+                                {
+                                    "kind": "speaker_labels",
+                                    "path": labels_path,
+                                    "data": data,
+                                }
+                            )
+                except Exception as exc:
+                    errors.append(
+                        {
+                            "kind": "speaker_labels",
+                            "path": str(labels_path),
+                            "message": str(exc),
+                        }
+                    )
+
+            corrections_path = talents_dir / "speaker_corrections.json"
+            if corrections_path.is_file():
+                try:
+                    raw = corrections_path.read_bytes()
+                    if source_id_bytes in raw:
+                        data = json.loads(raw)
+                        changed = False
+                        for correction in data.get("corrections", []):
+                            if correction.get("original_speaker") == source_id:
+                                correction["original_speaker"] = target_id
+                                changed = True
+                            if correction.get("corrected_speaker") == source_id:
+                                correction["corrected_speaker"] = target_id
+                                changed = True
+                        if changed:
+                            corrections_rewritten += 1
+                            operations.append(
+                                {
+                                    "kind": "speaker_corrections",
+                                    "path": corrections_path,
+                                    "data": data,
+                                }
+                            )
+                except Exception as exc:
+                    errors.append(
+                        {
+                            "kind": "speaker_corrections",
+                            "path": str(corrections_path),
+                            "message": str(exc),
+                        }
+                    )
+
+    return {
+        "operations": operations,
+        "section": _segment_section(
+            labels_rewritten=labels_rewritten,
+            corrections_rewritten=corrections_rewritten,
+            files_scanned=files_scanned,
+            errors=errors,
+        ),
+    }
+
+
+def _segment_day_dirs() -> list[Path]:
+    chronicle_days = [Path(path) for _, path in sorted(day_dirs().items())]
+    journal = Path(get_journal())
+    flat_days = sorted(
+        entry
+        for entry in journal.iterdir()
+        if entry.is_dir() and entry.name.isdigit() and len(entry.name) == 8
+    )
+    return chronicle_days or flat_days
+
+
+def _check_aka_cross_references(
+    source_id: str, source_display: str, target_id: str
+) -> list[str]:
+    offenders: list[str] = []
+    for entity_id in scan_journal_entities():
+        if entity_id in {source_id, target_id}:
+            continue
+        entity = load_journal_entity(entity_id)
+        if not entity:
+            continue
+        aka_values = entity.get("aka", [])
+        if not isinstance(aka_values, list):
+            continue
+        if source_id in aka_values or source_display in aka_values:
+            offenders.append(entity_id)
+    offenders.sort()
+    return offenders
+
+
+def _apply_facet_plan(operations: list[dict[str, Any]], target_id: str) -> None:
+    for operation in operations:
+        if operation["kind"] == "move":
+            source_rel_dir = operation["source_rel_dir"]
+            target_rel_dir = operation["target_rel_dir"]
+            if target_rel_dir.exists():
+                shutil.rmtree(target_rel_dir)
+            source_rel_dir.rename(target_rel_dir)
+            moved_path = target_rel_dir / "entity.json"
+            try:
+                with open(moved_path, encoding="utf-8") as handle:
+                    rel_data = json.load(handle)
+                rel_data["entity_id"] = target_id
+                save_facet_relationship(operation["facet"], target_id, rel_data)
+            except (json.JSONDecodeError, OSError):
+                pass
+            continue
+
+        save_facet_relationship(
+            operation["facet"], target_id, operation["relationship"]
+        )
+        save_observations(operation["facet"], target_id, operation["observations"])
+        shutil.rmtree(operation["source_rel_dir"])
+
+
+def _apply_segment_plan(operations: list[dict[str, Any]]) -> None:
+    for operation in operations:
+        out_path = operation["path"]
+        tmp_path = out_path.with_suffix(".tmp")
+        with open(tmp_path, "w", encoding="utf-8") as handle:
+            json.dump(operation["data"], handle, indent=2)
+        tmp_path.rename(out_path)
+
+
+def _clear_merge_caches() -> list[str]:
+    clear_journal_entity_cache()
+    clear_relationship_caches()
+    clear_observation_cache()
+    clear_entity_loading_cache()
+    return [
+        "journal_entity_cache",
+        "relationship_caches",
+        "observation_cache",
+        "entity_loading_cache",
+    ]
+
+
+def _audit_counts(result: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "identity": {
+            "akas_added": result["identity"]["akas_added_count"],
+            "emails_added": result["identity"]["emails_added_count"],
+            "principal_transferred": result["identity"]["principal_transferred"],
+        },
+        "voiceprints": {
+            "added": result["voiceprints"]["added"],
+            "skipped_duplicate": result["voiceprints"]["skipped_duplicate"],
+            "target_total": result["voiceprints"]["target_total"],
+        },
+        "facets": {
+            "moved": result["facets"]["moved_count"],
+            "merged": result["facets"]["merged_count"],
+            "observations_appended": result["facets"]["observations_appended"],
+        },
+        "segments": {
+            "labels_rewritten": result["segments"]["labels_rewritten"],
+            "corrections_rewritten": result["segments"]["corrections_rewritten"],
+            "files_scanned": result["segments"]["files_scanned"],
+            "errors": len(result["segments"]["errors"]),
+        },
+    }
+
+
+def _append_audit_log(
+    *,
+    source_id: str,
+    source_display_name: str,
+    target_id: str,
+    target_display_name: str,
+    result: dict[str, Any],
+    caller: str,
+) -> str:
+    logs_dir = Path(get_journal()) / "logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    audit_path = logs_dir / "entity-merges.jsonl"
+    payload = {
+        "ts": now_ms(),
+        "source_id": source_id,
+        "source_display_name": source_display_name,
+        "target_id": target_id,
+        "target_display_name": target_display_name,
+        "principal_transferred": result["identity"]["principal_transferred"],
+        "counts": _audit_counts(result),
+        "caller": caller,
+    }
+    with open(audit_path, "a", encoding="utf-8") as handle:
+        handle.write(json.dumps(payload, ensure_ascii=False) + "\n")
+    return str(audit_path)
+
+
+def merge_entity(
+    source_id: str,
+    target_id: str,
+    *,
+    keep_source_as_aka: bool = True,
+    commit: bool = False,
+    caller: str = "entities.merge",
+) -> dict[str, Any]:
+    if source_id == target_id:
+        return {"error": "Source and target must be different entities."}
+
+    source_entity = load_journal_entity(source_id)
+    if not source_entity:
+        return {"error": f"Source entity not found: {source_id}"}
+
+    target_entity = load_journal_entity(target_id)
+    if not target_entity:
+        return {"error": f"Target entity not found: {target_id}"}
+
+    if source_entity.get("blocked"):
+        return {"error": f"Cannot merge blocked entity: {source_id}"}
+    if target_entity.get("blocked"):
+        return {"error": f"Cannot merge blocked entity: {target_id}"}
+    if source_entity.get("is_principal") and target_entity.get("is_principal"):
+        return {"error": "Cannot merge two principal entities."}
+
+    source_display = str(source_entity.get("name", source_id))
+    target_display = str(target_entity.get("name", target_id))
+
+    offenders = _check_aka_cross_references(source_id, source_display, target_id)
+    if offenders:
+        offender_str = ", ".join(offenders)
+        return {
+            "error": f"Cannot merge '{source_id}': referenced in aka lists of entity ids: {offender_str}"
+        }
+
+    planned_target, identity_plan = _plan_identity_merge(
+        source_entity,
+        target_entity,
+        keep_source_as_aka=keep_source_as_aka,
+    )
+    resume_source = _plan_resume_marker(
+        source_entity,
+        target_id,
+        principal_transferred=identity_plan["principal_transferred"],
+    )
+    voiceprint_plan = _plan_voiceprint_merge(source_id, target_id)
+    facet_plan = _plan_facet_merge(source_id, target_id)
+    segment_plan = _plan_segment_rewrites(source_id, target_id)
+
+    zero = _empty_result_section()
+    result: dict[str, Any] = {
+        "merged": commit,
+        "source_id": source_id,
+        "target_id": target_id,
+        "identity": identity_plan if commit else zero["identity"],
+        "voiceprints": voiceprint_plan["section"] if commit else zero["voiceprints"],
+        "facets": facet_plan["section"] if commit else zero["facets"],
+        "segments": segment_plan["section"] if commit else zero["segments"],
+        "caches_cleared": [],
+        "audit_log_path": None,
+        "would_identity": None if commit else identity_plan,
+        "would_voiceprints": None if commit else voiceprint_plan["section"],
+        "would_facets": None if commit else facet_plan["section"],
+        "would_segments": None if commit else segment_plan["section"],
+    }
+
+    if not commit:
+        return result
+
+    try:
+        save_journal_entity(resume_source)
+        save_journal_entity(planned_target)
+
+        if voiceprint_plan["items"]:
+            save_voiceprints_batch(target_id, voiceprint_plan["items"])
+
+        _apply_facet_plan(facet_plan["operations"], target_id)
+        _apply_segment_plan(segment_plan["operations"])
+
+        discovery_cache = Path(get_journal()) / "awareness" / "discovery_clusters.json"
+        caches_cleared = _clear_merge_caches()
+        if discovery_cache.exists():
+            discovery_cache.unlink()
+            caches_cleared.append("discovery_clusters")
+
+        source_entity_dir = Path(get_journal()) / "entities" / source_id
+        if source_entity_dir.exists():
+            shutil.rmtree(source_entity_dir)
+
+        result["caches_cleared"] = caches_cleared
+
+        try:
+            result["audit_log_path"] = _append_audit_log(
+                source_id=source_id,
+                source_display_name=source_display,
+                target_id=target_id,
+                target_display_name=str(planned_target.get("name", target_display)),
+                result=result,
+                caller=caller,
+            )
+        except OSError as exc:
+            result["segments"]["errors"].append(
+                {
+                    "kind": "audit_log",
+                    "path": str(Path(get_journal()) / "logs" / "entity-merges.jsonl"),
+                    "message": str(exc),
+                }
+            )
+
+        return result
+    except Exception as exc:
+        return {"error": str(exc)}
diff --git a/think/entities/voiceprints.py b/think/entities/voiceprints.py
new file mode 100644
index 000000000..2a6473bc2
--- /dev/null
+++ b/think/entities/voiceprints.py
@@ -0,0 +1,188 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Shared voiceprint helpers for entity-aware speaker workflows."""
+
+from __future__ import annotations
+
+import fcntl
+import json
+import logging
+from pathlib import Path
+
+import numpy as np
+
+from think.entities.journal import (
+    ensure_journal_entity_memory,
+    journal_entity_memory_path,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def normalize_embedding(emb: np.ndarray) -> np.ndarray | None:
+    """L2-normalize an embedding vector. Returns None if norm is zero."""
+    emb = emb.astype(np.float32)
+    norm = np.linalg.norm(emb)
+    if norm > 0:
+        return emb / norm
+    return None
+
+
+def load_entity_voiceprints_file(
+    entity_id: str,
+) -> tuple[np.ndarray, list[dict]] | None:
+    """Load an entity's voiceprints.npz, returning embeddings and parsed metadata."""
+    try:
+        folder = journal_entity_memory_path(entity_id)
+    except (RuntimeError, ValueError):
+        return None
+
+    npz_path = folder / "voiceprints.npz"
+    if not npz_path.exists():
+        return None
+
+    try:
+        with np.load(npz_path, allow_pickle=False) as data:
+            embeddings = data.get("embeddings")
+            metadata_arr = data.get("metadata")
+            if embeddings is None or metadata_arr is None:
+                return None
+            metadata_list = [json.loads(m) for m in metadata_arr]
+        return embeddings, metadata_list
+    except Exception as exc:
+        logger.warning("Failed to load voiceprints for entity %s: %s", entity_id, exc)
+        return None
+
+
+def load_existing_voiceprint_keys(entity_id: str) -> set[tuple]:
+    """Return saved voiceprint identity keys for idempotency checks."""
+    result = load_entity_voiceprints_file(entity_id)
+    if result is None:
+        return set()
+
+    _, metadata_list = result
+    return {
+        (m.get("day"), m.get("segment_key"), m.get("source"), m.get("sentence_id"))
+        for m in metadata_list
+    }
+
+
+def save_voiceprints_batch(
+    entity_id: str,
+    new_items: list[tuple[np.ndarray, dict]],
+) -> int:
+    """Append a batch of normalized voiceprints to an entity in one write."""
+    if not new_items:
+        return 0
+
+    folder = ensure_journal_entity_memory(entity_id)
+    npz_path = folder / "voiceprints.npz"
+
+    if npz_path.exists():
+        try:
+            with np.load(npz_path, allow_pickle=False) as data:
+                existing_emb = data["embeddings"]
+                existing_meta_strings = data["metadata"]
+                existing_meta_dicts = [json.loads(m) for m in existing_meta_strings]
+        except (FileNotFoundError, ValueError, np.lib.npyio.NpzFile) as exc:
+            logger.warning(
+                "Failed to load existing voiceprints for %s from %s: %s. Starting fresh.",
+                entity_id,
+                npz_path,
+                exc,
+            )
+            existing_emb = np.empty((0, 256), dtype=np.float32)
+            existing_meta_dicts = []
+        except Exception:
+            logger.exception(
+                "Unexpected error loading existing voiceprints for %s from %s",
+                entity_id,
+                npz_path,
+            )
+            raise
+    else:
+        existing_emb = np.empty((0, 256), dtype=np.float32)
+        existing_meta_dicts = []
+
+    new_emb_list = [emb.reshape(1, -1).astype(np.float32) for emb, _ in new_items]
+    new_meta_dicts = [meta_dict for _, meta_dict in new_items]
+
+    if new_emb_list:
+        new_emb_np = np.vstack(new_emb_list)
+        combined_emb = (
+            np.vstack([existing_emb, new_emb_np])
+            if len(existing_emb) > 0
+            else new_emb_np
+        )
+        combined_meta_dicts = existing_meta_dicts + new_meta_dicts
+    else:
+        combined_emb = existing_emb
+        combined_meta_dicts = existing_meta_dicts
+
+    save_voiceprints_safely(
+        npz_path=npz_path,
+        embeddings=combined_emb,
+        metadata=combined_meta_dicts,
+    )
+    return len(new_items)
+
+
+def voiceprint_file_path(entity_id: str) -> Path:
+    """Return the canonical voiceprints.npz path for an entity."""
+    return ensure_journal_entity_memory(entity_id) / "voiceprints.npz"
+
+
+def save_voiceprints_safely(
+    npz_path: Path,
+    embeddings: np.ndarray,
+    metadata: list[dict],
+) -> None:
+    """Safely save a voiceprint NPZ with file locking and integrity check."""
+    lock_path = npz_path.with_suffix(".lock")
+    tmp_path = npz_path.with_name(npz_path.stem + ".tmp.npz")
+
+    npz_path.parent.mkdir(parents=True, exist_ok=True)
+
+    try:
+        with open(lock_path, "w", encoding="utf-8") as lock_file:
+            try:
+                fcntl.flock(lock_file, fcntl.LOCK_EX)
+                np.savez_compressed(
+                    tmp_path,
+                    embeddings=embeddings,
+                    metadata=metadata,
+                )
+                if not tmp_path.exists():
+                    raise FileNotFoundError(
+                        f"Temporary voiceprint file not found: {tmp_path}"
+                    )
+                tmp_path.rename(npz_path)
+
+                with np.load(npz_path, allow_pickle=False) as data:
+                    if "embeddings" not in data or "metadata" not in data:
+                        raise ValueError(
+                            "Missing 'embeddings' or 'metadata' keys in loaded NPZ."
+                        )
+                logger.info(
+                    "Successfully wrote and verified voiceprint file: %s",
+                    npz_path,
+                )
+            except Exception:
+                if tmp_path.exists():
+                    try:
+                        tmp_path.unlink()
+                    except OSError:
+                        logger.exception(
+                            "Failed to clean up temporary voiceprint file %s",
+                            tmp_path,
+                        )
+                raise
+            finally:
+                try:
+                    fcntl.flock(lock_file, fcntl.LOCK_UN)
+                except OSError:
+                    logger.exception("Failed to release lock on %s", lock_path)
+    except OSError:
+        logger.exception("Failed to acquire or manage lock file %s", lock_path)
+        raise
diff --git a/think/merge.py b/think/merge.py
index d0d1f9df7..dbf299742 100644
--- a/think/merge.py
+++ b/think/merge.py
@@ -18,6 +18,7 @@ from think.entities.journal import (
     save_journal_entity,
 )
 from think.entities.matching import find_matching_entity
+from think.entities.merge import _dedupe_akas, _dedupe_emails, _dedupe_observations
 from think.entities.observations import save_observations
 from think.entities.relationships import save_facet_relationship
 from think.utils import CHRONICLE_DIR, iter_segments
@@ -268,36 +269,17 @@ def _merge_entities(
             target_entity = dict(target_entities.get(target_id, match))
             pre_merge_snapshot = dict(target_entity)
 
-            aka_by_lower: dict[str, str] = {}
-            for values in (target_entity.get("aka", []), source_entity.get("aka", [])):
-                if not isinstance(values, list):
-                    continue
-                for value in values:
-                    if not value:
-                        continue
-                    key = str(value).lower()
-                    if key not in aka_by_lower:
-                        aka_by_lower[key] = str(value)
-            if aka_by_lower:
-                target_entity["aka"] = sorted(aka_by_lower.values(), key=str.lower)
-
-            merged_emails: list[str] = []
-            seen_emails: set[str] = set()
-            for values in (
+            merged_akas = _dedupe_akas(
+                target_entity.get("aka", []),
+                source_entity.get("aka", []),
+            )
+            if merged_akas:
+                target_entity["aka"] = merged_akas
+
+            merged_emails = _dedupe_emails(
                 target_entity.get("emails", []),
                 source_entity.get("emails", []),
-            ):
-                if not isinstance(values, list):
-                    continue
-                for value in values:
-                    if not value:
-                        continue
-                    email = str(value)
-                    key = email.lower()
-                    if key in seen_emails:
-                        continue
-                    seen_emails.add(key)
-                    merged_emails.append(email)
+            )
             if merged_emails:
                 target_entity["emails"] = merged_emails
 
@@ -424,17 +406,10 @@ def _merge_overlapping_facet(
                     target_observations = _read_jsonl(
                         target_entity_dir / "observations.jsonl"
                     )
-                    seen = {
-                        (item.get("content", ""), item.get("observed_at"))
-                        for item in target_observations
-                    }
-                    merged_observations = list(target_observations)
-                    for item in source_observations:
-                        key = (item.get("content", ""), item.get("observed_at"))
-                        if key in seen:
-                            continue
-                        seen.add(key)
-                        merged_observations.append(item)
+                    merged_observations = _dedupe_observations(
+                        source_observations,
+                        target_observations,
+                    )
 
                     if not dry_run:
                         save_facet_relationship(