Add speakers link-import and seed-from-imports commands

Two new CLI commands for the speakers app:
- `link-import`: links an import participant name to a journal entity by adding it as an aka, with alias collision validation
- `seed-from-imports`: scans import-stream segments for speaker-attributed conversation transcripts and seeds voiceprints from corresponding embeddings

Files changed:
- apps/speakers/call.py (new commands + updated docstring)
- apps/speakers/bootstrap.py (new seed_from_imports function + updated docstring)
- apps/speakers/tests/conftest.py (stream kwarg on create_segment, new create_import_segment helper)
- apps/speakers/tests/test_seed_imports.py (new test file, 12 tests)

apps/speakers/bootstrap.py

···
 subtract the owner's sentences, then saves the remaining embeddings as
 that speaker's voiceprint.
 
+Import seeding: Scans import-stream segments for conversation transcripts
+with per-line speaker attribution, maps speakers to journal entities, and
+saves corresponding embeddings as voiceprints.
+
 Name resolution: Compares voiceprint centroids between name variants.
 Pairs with cosine similarity > 0.90 are the same person. Unambiguous
 variants are auto-merged by adding the short name as an aka on the
···
     load_journal_entity,
     save_journal_entity,
 )
-from think.utils import day_dirs, now_ms, segment_path
+from think.utils import day_dirs, iter_segments, now_ms, segment_path
 
 logger = logging.getLogger(__name__)
 
···
                 stats["segments_scanned"],
                 sum(len(v) for v in entity_embeddings.values()),
             )
+
+    # Batch save all collected embeddings
+    if not dry_run:
+        for entity_id, emb_list in entity_embeddings.items():
+            try:
+                saved = _save_voiceprints_batch(entity_id, emb_list)
+                stats["embeddings_saved"] += saved
+            except Exception as e:
+                name = entity_names.get(entity_id, entity_id)
+                stats["errors"].append(f"Failed to save for {name}: {e}")
+                logger.exception("Failed to save voiceprints for %s", entity_id)
+    else:
+        stats["embeddings_saved"] = sum(len(v) for v in entity_embeddings.values())
+
+    return stats
+
+
+# Generic speaker names to skip (AI conversation imports)
+_GENERIC_SPEAKERS = frozenset({"Human", "Assistant", "human", "assistant", ""})
+
+
+def seed_from_imports(dry_run: bool = False) -> dict[str, Any]:
+    """Seed voiceprints from speaker-attributed import transcripts.
+
+    Scans import-stream segments for conversation_transcript.jsonl files
+    with per-line speaker attribution. Maps speaker names to journal
+    entities and saves corresponding embeddings as voiceprints.
+
+    Unlike bootstrap_voiceprints(), this does NOT create new entities —
+    unmatched speaker names are skipped.
+
+    Args:
+        dry_run: If True, report what would be saved without saving
+
+    Returns:
+        Dict with statistics about the seed run
+    """
+    (
+        load_embeddings_file,
+        normalize_embedding,
+        _,
+        _,
+    ) = _routes_helpers()
+
+    # Load owner centroid — required for owner subtraction
+    centroid_data = load_owner_centroid()
+    if centroid_data is None:
+        return {"error": "No confirmed owner centroid. Run owner detection first."}
+
+    owner_centroid, owner_threshold = centroid_data
+
+    # Load all journal entities for speaker name matching
+    journal_entities = load_all_journal_entities()
+    entities_list = [e for e in journal_entities.values() if not e.get("blocked")]
+
+    stats: dict[str, Any] = {
+        "segments_scanned": 0,
+        "segments_with_speakers": 0,
+        "speakers_found": {},
+        "embeddings_saved": 0,
+        "embeddings_skipped_owner": 0,
+        "embeddings_skipped_duplicate": 0,
+        "errors": [],
+    }
+
+    # Collect embeddings per entity for efficient batch saves
+    entity_embeddings: dict[str, list[tuple[np.ndarray, dict]]] = defaultdict(list)
+    entity_existing: dict[str, set] = {}
+    entity_names: dict[str, str] = {}
+
+    days = sorted(day_dirs().keys())
+
+    for day in days:
+        for stream, seg_key, seg_dir in iter_segments(day):
+            # Only process import streams
+            if not stream.startswith("import."):
+                continue
+
+            stats["segments_scanned"] += 1
+
+            # Read conversation_transcript.jsonl
+            jsonl_path = seg_dir / "conversation_transcript.jsonl"
+            if not jsonl_path.exists():
+                continue
+
+            try:
+                lines = jsonl_path.read_text(encoding="utf-8").strip().split("\n")
+            except OSError as e:
+                stats["errors"].append(f"Failed to read {jsonl_path}: {e}")
+                continue
+
+            if len(lines) < 2:
+                continue
+
+            # Build sentence_id -> speaker mapping
+            # Line 0 is metadata header. Lines 1+ are entries.
+            # sentence_id is 1-based, matching statement_ids in NPZ.
+            sid_to_speaker: dict[int, str] = {}
+            has_real_speakers = False
+            for line_idx in range(1, len(lines)):
+                try:
+                    entry = json.loads(lines[line_idx])
+                except (json.JSONDecodeError, IndexError):
+                    continue
+                speaker = entry.get("speaker", "")
+                if speaker and speaker not in _GENERIC_SPEAKERS:
+                    sid_to_speaker[line_idx] = speaker
+                    has_real_speakers = True
+
+            if not has_real_speakers:
+                continue
+
+            stats["segments_with_speakers"] += 1
+
+            # Find audio embedding NPZ files in this segment
+            # Accept both "<source>_audio" pattern and plain "audio"
+            npz_files = list(seg_dir.glob("*.npz"))
+            sources = [
+                f.stem
+                for f in npz_files
+                if f.stem.endswith("_audio") or f.stem == "audio"
+            ]
+            if not sources:
+                continue
+
+            for source in sources:
+                emb_data = load_embeddings_file(seg_dir / f"{source}.npz")
+                if emb_data is None:
+                    continue
+
+                embeddings, statement_ids = emb_data
+
+                for embedding, sid in zip(embeddings, statement_ids):
+                    sentence_id = int(sid)
+                    speaker_name = sid_to_speaker.get(sentence_id)
+                    if not speaker_name:
+                        continue
+
+                    # Match speaker to entity — skip if no match
+                    entity = find_matching_entity(speaker_name, entities_list)
+                    if not entity:
+                        continue
+
+                    entity_id = entity["id"]
+                    entity_name = entity.get("name", speaker_name)
+                    entity_names[entity_id] = entity_name
+                    stats["speakers_found"].setdefault(entity_name, 0)
+
+                    # 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
+                        )
+
+                    existing_keys = entity_existing[entity_id]
+                    vp_key = (day, seg_key, source, sentence_id)
+
+                    # Idempotency: skip if already saved
+                    if vp_key in existing_keys:
+                        stats["embeddings_skipped_duplicate"] += 1
+                        continue
+
+                    normalized = normalize_embedding(embedding)
+                    if normalized is None:
+                        continue
+
+                    # Contamination guard: reject embeddings too similar to owner
+                    owner_score = float(np.dot(normalized, owner_centroid))
+                    if owner_score >= owner_threshold:
+                        stats["embeddings_skipped_owner"] += 1
+                        continue
+
+                    metadata = {
+                        "day": day,
+                        "segment_key": seg_key,
+                        "source": source,
+                        "stream": stream,
+                        "sentence_id": sentence_id,
+                        "added_at": now_ms(),
+                    }
+
+                    entity_embeddings[entity_id].append((normalized, metadata))
+                    existing_keys.add(vp_key)
+                    stats["speakers_found"][entity_name] += 1
 
     # Batch save all collected embeddings
     if not dry_run:

apps/speakers/call.py

···
     sol call speakers identify <cluster-id> <name> [--entity-id ID]
     sol call speakers merge-names <alias> <canonical>
     sol call speakers suggest [--limit N] [--json]
+    sol call speakers link-import <name> --entity-id <ID>
+    sol call speakers seed-from-imports [--dry-run] [--json]
 """
 
 from __future__ import annotations
···
     from apps.speakers.suggest import format_suggestions
 
     typer.echo(format_suggestions(results))
+
+
[e-mail]("link-import")
+def link_import(
+    name: str = typer.Argument(..., help="Import participant name to link."),
+    entity_id: str = typer.Option(
+        ..., "--entity-id", help="Journal entity ID to link to."
+    ),
+) -> None:
+    """Link an import participant name to a journal entity as an aka."""
+    import json as json_mod
+
+    from think.entities.journal import (
+        load_all_journal_entities,
+        load_journal_entity,
+        save_journal_entity,
+    )
+    from think.entities.matching import validate_aka_uniqueness
+    from think.utils import now_ms
+
+    entity = load_journal_entity(entity_id)
+    if entity is None:
+        typer.echo(
+            json_mod.dumps({"error": f"Entity not found: {entity_id}"}, indent=2),
+            err=True,
+        )
+        raise typer.Exit(1)
+
+    existing_aka = set(entity.get("aka", []))
+    already_present = name in existing_aka
+    if not already_present:
+        # Check for alias collision with other entities
+        all_entities = load_all_journal_entities()
+        entities_list = [e for e in all_entities.values() if not e.get("blocked")]
+        collision = validate_aka_uniqueness(
+            name, entities_list, exclude_entity_name=entity.get("name")
+        )
+        if collision:
+            typer.echo(
+                json_mod.dumps(
+                    {
+                        "error": f"Name '{name}' conflicts with entity: {collision}",
+                    },
+                    indent=2,
+                ),
+                err=True,
+            )
+            raise typer.Exit(1)
+
+        existing_aka.add(name)
+        entity["aka"] = sorted(existing_aka)
+        entity["updated_at"] = now_ms()
+        save_journal_entity(entity)
+
+    typer.echo(
+        json_mod.dumps(
+            {
+                "linked": True,
+                "entity_id": entity_id,
+                "name_added": name,
+                "already_present": already_present,
+            },
+            indent=2,
+            default=str,
+        )
+    )
+
+
[e-mail]("seed-from-imports")
+def seed_from_imports_cmd(
+    dry_run: bool = typer.Option(
+        False, "--dry-run", help="Show what would be saved without saving."
+    ),
+    json_output: bool = typer.Option(
+        False, "--json", help="Output full result as JSON."
+    ),
+) -> None:
+    """Seed voiceprints from speaker-attributed import transcripts.
+
+    Scans import-stream segments for conversation_transcript.jsonl files
+    with per-line speaker attribution. Maps speaker names to journal
+    entities and saves corresponding embeddings as voiceprints.
+    """
+    from apps.speakers.bootstrap import seed_from_imports
+
+    if dry_run and not json_output:
+        typer.echo("DRY RUN — no voiceprints will be saved\n")
+
+    if not json_output:
+        typer.echo("Seeding voiceprints from import transcripts...")
+    stats = seed_from_imports(dry_run=dry_run)
+
+    if "error" in stats:
+        typer.echo(f"Error: {stats['error']}", err=True)
+        raise typer.Exit(1)
+    if json_output:
+        import json as json_mod
+
+        typer.echo(json_mod.dumps(stats, indent=2, default=str))
+        return
+
+    typer.echo(f"\nImport segments scanned: {stats['segments_scanned']}")
+    typer.echo(f"Segments with speakers: {stats['segments_with_speakers']}")
+    typer.echo(f"Unique speakers: {len(stats['speakers_found'])}")
+    typer.echo(f"Embeddings saved: {stats['embeddings_saved']}")
+    typer.echo(f"Embeddings skipped (owner): {stats['embeddings_skipped_owner']}")
+    typer.echo(
+        f"Embeddings skipped (duplicate): {stats['embeddings_skipped_duplicate']}"
+    )
+
+    if stats["speakers_found"]:
+        typer.echo("\nTop speakers by embedding count:")
+        sorted_speakers = sorted(
+            stats["speakers_found"].items(), key=lambda x: x[1], reverse=True
+        )
+        for name, count in sorted_speakers[:15]:
+            typer.echo(f"  {name}: {count}")
+        if len(sorted_speakers) > 15:
+            typer.echo(f"  ... and {len(sorted_speakers) - 15} more")
+
+    if stats["errors"]:
+        typer.echo(f"\nErrors ({len(stats['errors'])}):", err=True)
+        for err in stats["errors"]:
+            typer.echo(f"  {err}", err=True)

apps/speakers/tests/conftest.py

···
             sources: list[str],
             num_sentences: int = 5,
             *,
+            stream: str | None = None,
             embeddings: np.ndarray | None = None,
         ) -> Path:
             """Create a segment with sentence embeddings.
···
                 sources: List of audio sources (e.g., ["mic_audio", "sys_audio"])
                 num_sentences: Number of sentences to create
             """
-            segment_dir = self.journal / day / STREAM / segment_key
+            segment_dir = self.journal / day / (stream or STREAM) / segment_key
             segment_dir.mkdir(parents=True, exist_ok=True)
 
             sentence_count = (
···
                 # Create dummy audio file
                 audio_path = segment_dir / f"{source}.flac"
                 audio_path.write_bytes(b"")  # Empty placeholder
+
+            return segment_dir
+
+        def create_import_segment(
+            self,
+            day: str,
+            segment_key: str,
+            speakers_text: list[tuple[str, str]],
+            source: str = "audio",
+            *,
+            stream: str = "import.granola",
+            embeddings: np.ndarray | None = None,
+        ) -> Path:
+            """Create an import segment with conversation_transcript.jsonl and NPZ.
+
+            Args:
+                day: Day string (YYYYMMDD)
+                segment_key: Segment key (HHMMSS_LEN)
+                speakers_text: List of (speaker_name, text) tuples for transcript lines
+                source: Audio source stem for NPZ file (default "audio")
+                stream: Import stream name (default "import.granola")
+                embeddings: Optional custom embeddings array (N×256)
+            """
+            segment_dir = self.journal / day / stream / segment_key
+            segment_dir.mkdir(parents=True, exist_ok=True)
+
+            num_sentences = len(speakers_text)
+
+            # Write conversation_transcript.jsonl
+            jsonl_path = segment_dir / "conversation_transcript.jsonl"
+            lines = [
+                json.dumps(
+                    {
+                        "imported": {"id": "test-import", "source": "test"},
+                        "topics": "test",
+                        "setting": "meeting",
+                    }
+                )
+            ]
+
+            time_part = segment_key.split("_")[0]
+            base_h = int(time_part[0:2])
+            base_m = int(time_part[2:4])
+            base_s = int(time_part[4:6])
+            base_seconds = base_h * 3600 + base_m * 60 + base_s
+
+            for i, (speaker, text) in enumerate(speakers_text):
+                offset = i * 5
+                abs_seconds = base_seconds + offset
+                h = (abs_seconds // 3600) % 24
+                m = (abs_seconds % 3600) // 60
+                s = abs_seconds % 60
+                lines.append(
+                    json.dumps(
+                        {
+                            "start": f"{h:02d}:{m:02d}:{s:02d}",
+                            "speaker": speaker,
+                            "text": text,
+                            "source": "import",
+                        }
+                    )
+                )
+            jsonl_path.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(num_sentences, 256).astype(
+                    np.float32
+                )
+                norms = np.linalg.norm(source_embeddings, axis=1, keepdims=True)
+                source_embeddings = source_embeddings / norms
+            else:
+                source_embeddings = embeddings.astype(np.float32)
+            statement_ids = np.arange(1, num_sentences + 1, dtype=np.int32)
+            np.savez_compressed(
+                npz_path,
+                embeddings=source_embeddings,
+                statement_ids=statement_ids,
+            )
 
             return segment_dir
 

apps/speakers/tests/test_seed_imports.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Tests for link-import and seed-from-imports CLI commands."""
+
+from __future__ import annotations
+
+import json
+
+import numpy as np
+from typer.testing import CliRunner
+
+from apps.speakers.call import app as speakers_app
+
+_runner = CliRunner()
+
+
+# --- link-import tests ---
+
+
+def test_link_import_success(speakers_env):
+    """link-import adds name as aka on entity."""
+    env = speakers_env()
+    env.create_entity("Sarah Chen")
+
+    result = _runner.invoke(
+        speakers_app,
+        ["link-import", "Sarah C", "--entity-id", "sarah_chen"],
+    )
+    assert result.exit_code == 0
+    data = json.loads(result.output)
+    assert data["linked"] is True
+    assert data["entity_id"] == "sarah_chen"
+    assert data["name_added"] == "Sarah C"
+    assert data["already_present"] is False
+
+    # Verify entity was actually updated
+    entity_path = env.journal / "entities" / "sarah_chen" / "entity.json"
+    entity = json.loads(entity_path.read_text())
+    assert "Sarah C" in entity["aka"]
+    assert "updated_at" in entity
+
+
+def test_link_import_already_present(speakers_env):
+    """link-import reports already_present when name is already an aka."""
+    env = speakers_env()
+    entity_dir = env.create_entity("Sarah Chen")
+    # Manually add aka
+    entity_path = entity_dir / "entity.json"
+    entity = json.loads(entity_path.read_text())
+    entity["aka"] = ["Sarah C"]
+    entity_path.write_text(json.dumps(entity))
+
+    result = _runner.invoke(
+        speakers_app,
+        ["link-import", "Sarah C", "--entity-id", "sarah_chen"],
+    )
+    assert result.exit_code == 0
+    data = json.loads(result.output)
+    assert data["already_present"] is True
+
+
+def test_link_import_entity_not_found(speakers_env):
+    """link-import exits 1 with error JSON for missing entity."""
+    speakers_env()
+    result = _runner.invoke(
+        speakers_app,
+        ["link-import", "Nobody", "--entity-id", "nonexistent"],
+    )
+    assert result.exit_code == 1
+
+
+def test_link_import_collision(speakers_env):
+    """link-import exits 1 when aka collides with another entity."""
+    env = speakers_env()
+    env.create_entity("Alice Johnson")
+    env.create_entity("Bob Smith")
+
+    result = _runner.invoke(
+        speakers_app,
+        ["link-import", "Bob Smith", "--entity-id", "alice_johnson"],
+    )
+    assert result.exit_code == 1
+    data = json.loads(result.output)
+    assert "conflicts" in data["error"]
+
+
+# --- seed-from-imports tests ---
+
+
+def _create_owner_centroid(env, *, threshold: float = 0.85):
+    """Helper: create owner centroid file for seed tests."""
+    env.create_entity("Owner", is_principal=True)
+
+    owner_dir = env.journal / "entities" / "owner"
+    owner_dir.mkdir(parents=True, exist_ok=True)
+
+    # Create a distinct owner embedding (pointing in a specific direction)
+    owner_emb = np.zeros(256, dtype=np.float32)
+    owner_emb[0] = 1.0  # Owner points along axis 0
+
+    centroid_path = owner_dir / "owner_centroid.npz"
+    np.savez_compressed(
+        centroid_path,
+        centroid=owner_emb,
+        threshold=np.float32(threshold),
+    )
+    return owner_emb
+
+
+def test_seed_from_imports_happy_path(speakers_env):
+    """seed-from-imports seeds voiceprints for matched speakers."""
+    env = speakers_env()
+    _create_owner_centroid(env)
+    env.create_entity("Alice Johnson")
+
+    # Create embeddings that are NOT owner-like (orthogonal to owner)
+    embs = np.zeros((3, 256), dtype=np.float32)
+    embs[0, 1] = 1.0  # Points along axis 1 (orthogonal to owner axis 0)
+    embs[1, 2] = 1.0
+    embs[2, 3] = 1.0
+
+    env.create_import_segment(
+        "20240101",
+        "100000_300",
+        [
+            ("Alice Johnson", "Hello everyone"),
+            ("Alice Johnson", "Let's discuss the project"),
+            ("Alice Johnson", "Thanks for joining"),
+        ],
+        embeddings=embs,
+    )
+
+    result = _runner.invoke(speakers_app, ["seed-from-imports", "--json"])
+    assert result.exit_code == 0, result.output
+    data = json.loads(result.output)
+    assert data["segments_scanned"] >= 1
+    assert data["segments_with_speakers"] >= 1
+    assert data["embeddings_saved"] == 3
+    assert "Alice Johnson" in data["speakers_found"]
+
+
+def test_seed_from_imports_skips_generic_speakers(speakers_env):
+    """seed-from-imports skips Human/Assistant speakers."""
+    env = speakers_env()
+    _create_owner_centroid(env)
+
+    embs = np.zeros((2, 256), dtype=np.float32)
+    embs[0, 1] = 1.0
+    embs[1, 2] = 1.0
+
+    env.create_import_segment(
+        "20240101",
+        "100000_300",
+        [
+            ("Human", "How do I fix this?"),
+            ("Assistant", "Try restarting the service."),
+        ],
+        stream="import.chatgpt",
+        embeddings=embs,
+    )
+
+    result = _runner.invoke(speakers_app, ["seed-from-imports", "--json"])
+    assert result.exit_code == 0
+    data = json.loads(result.output)
+    assert data["embeddings_saved"] == 0
+    assert data["segments_with_speakers"] == 0
+
+
+def test_seed_from_imports_skips_unmatched_speakers(speakers_env):
+    """seed-from-imports skips speakers with no matching entity."""
+    env = speakers_env()
+    _create_owner_centroid(env)
+    # Don't create an entity for "Unknown Person"
+
+    embs = np.zeros((1, 256), dtype=np.float32)
+    embs[0, 1] = 1.0
+
+    env.create_import_segment(
+        "20240101",
+        "100000_300",
+        [("Unknown Person", "Hello")],
+        embeddings=embs,
+    )
+
+    result = _runner.invoke(speakers_app, ["seed-from-imports", "--json"])
+    assert result.exit_code == 0
+    data = json.loads(result.output)
+    assert data["embeddings_saved"] == 0
+
+
+def test_seed_from_imports_owner_contamination(speakers_env):
+    """seed-from-imports skips embeddings too similar to owner."""
+    env = speakers_env()
+    _create_owner_centroid(env, threshold=0.85)
+    env.create_entity("Alice Johnson")
+
+    # Create an embedding that IS owner-like (same direction as owner centroid)
+    embs = np.zeros((1, 256), dtype=np.float32)
+    embs[0, 0] = 1.0  # Same direction as owner (axis 0)
+
+    env.create_import_segment(
+        "20240101",
+        "100000_300",
+        [("Alice Johnson", "Hello")],
+        embeddings=embs,
+    )
+
+    result = _runner.invoke(speakers_app, ["seed-from-imports", "--json"])
+    assert result.exit_code == 0
+    data = json.loads(result.output)
+    assert data["embeddings_skipped_owner"] == 1
+    assert data["embeddings_saved"] == 0
+
+
+def test_seed_from_imports_dedup(speakers_env):
+    """seed-from-imports is idempotent — second run skips duplicates."""
+    env = speakers_env()
+    _create_owner_centroid(env)
+    env.create_entity("Alice Johnson")
+
+    embs = np.zeros((1, 256), dtype=np.float32)
+    embs[0, 1] = 1.0
+
+    env.create_import_segment(
+        "20240101",
+        "100000_300",
+        [("Alice Johnson", "Hello")],
+        embeddings=embs,
+    )
+
+    # First run
+    result1 = _runner.invoke(speakers_app, ["seed-from-imports", "--json"])
+    assert result1.exit_code == 0
+    data1 = json.loads(result1.output)
+    assert data1["embeddings_saved"] == 1
+
+    # Second run — should be all duplicates
+    result2 = _runner.invoke(speakers_app, ["seed-from-imports", "--json"])
+    assert result2.exit_code == 0
+    data2 = json.loads(result2.output)
+    assert data2["embeddings_saved"] == 0
+    assert data2["embeddings_skipped_duplicate"] == 1
+
+
+def test_seed_from_imports_dry_run(speakers_env):
+    """seed-from-imports --dry-run reports stats but doesn't write."""
+    env = speakers_env()
+    _create_owner_centroid(env)
+    env.create_entity("Alice Johnson")
+
+    embs = np.zeros((1, 256), dtype=np.float32)
+    embs[0, 1] = 1.0
+
+    env.create_import_segment(
+        "20240101",
+        "100000_300",
+        [("Alice Johnson", "Hello")],
+        embeddings=embs,
+    )
+
+    result = _runner.invoke(
+        speakers_app,
+        ["seed-from-imports", "--dry-run", "--json"],
+    )
+    assert result.exit_code == 0
+    data = json.loads(result.output)
+    assert data["embeddings_saved"] == 1  # Would-be saved count
+
+    # Verify nothing was actually written
+    vp_path = env.journal / "entities" / "alice_johnson" / "voiceprints.npz"
+    assert not vp_path.exists()
+
+
+def test_seed_from_imports_no_owner_centroid(speakers_env):
+    """seed-from-imports errors when no owner centroid exists."""
+    speakers_env()
+
+    result = _runner.invoke(speakers_app, ["seed-from-imports", "--json"])
+    assert result.exit_code == 1
+
+
+def test_seed_from_imports_no_import_segments(speakers_env):
+    """seed-from-imports returns zeroed stats when no import segments exist."""
+    env = speakers_env()
+    _create_owner_centroid(env)
+
+    result = _runner.invoke(speakers_app, ["seed-from-imports", "--json"])
+    assert result.exit_code == 0
+    data = json.loads(result.output)
+    assert data["segments_scanned"] == 0
+    assert data["embeddings_saved"] == 0