diff --git a/apps/speakers/owner.py b/apps/speakers/owner.py
new file mode 100644
index 000000000..6b937cd25
--- /dev/null
+++ b/apps/speakers/owner.py
@@ -0,0 +1,335 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Owner voice identification helpers for the speakers app."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+from sklearn.cluster import HDBSCAN
+
+from think.awareness import update_state
+from think.entities.journal import get_journal_principal, journal_entity_memory_path
+from think.utils import day_dirs, get_journal, segment_path
+
+logger = logging.getLogger(__name__)
+
+MIN_SEGMENTS = 50
+MAX_EMBEDDINGS = 30000
+OWNER_THRESHOLD = 0.82
+
+
+def _mark_no_cluster(segment_count: int) -> None:
+    """Record that detection ran but did not produce a usable cluster."""
+    update_state(
+        "voiceprint",
+        {
+            "status": "no_cluster",
+            "segments_checked": segment_count,
+            "attempted_at": _iso_now(),
+        },
+    )
+
+
+def _routes_helpers():
+    """Load speakers route helpers lazily to avoid import cycles."""
+    from apps.speakers.routes import (
+        _load_embeddings_file,
+        _normalize_embedding,
+        _scan_segment_embeddings,
+    )
+
+    return _load_embeddings_file, _normalize_embedding, _scan_segment_embeddings
+
+
+def _owner_candidate_path() -> Path:
+    """Return the temporary owner candidate NPZ path."""
+    awareness_dir = Path(get_journal()) / "awareness"
+    awareness_dir.mkdir(parents=True, exist_ok=True)
+    return awareness_dir / "owner_candidate.npz"
+
+
+def _iso_now() -> str:
+    """Return a timestamp string for persisted metadata."""
+    return datetime.now().isoformat()
+
+
+def _audio_url(day: str, stream: str, segment_key: str, source: str) -> str:
+    """Build the existing speakers audio-serving URL for a sample."""
+    return f"/app/speakers/api/serve_audio/{day}/{stream}__{segment_key}__{source}.flac"
+
+
+def count_segments_with_embeddings() -> int:
+    """Count all journal segments that contain audio embedding files."""
+    _, _, scan_segment_embeddings = _routes_helpers()
+
+    total = 0
+    for day in day_dirs().keys():
+        total += len(scan_segment_embeddings(day))
+    return total
+
+
+def _subsample_embeddings(
+    embeddings: np.ndarray, provenance: list[dict[str, Any]]
+) -> tuple[np.ndarray, list[dict[str, Any]]]:
+    """Subsample embeddings proportionally across streams when over the limit."""
+    total = len(embeddings)
+    if total <= MAX_EMBEDDINGS:
+        return embeddings, provenance
+
+    rng = np.random.default_rng(42)
+    stream_indices: dict[str, list[int]] = {}
+    for idx, record in enumerate(provenance):
+        stream_indices.setdefault(record["stream"], []).append(idx)
+
+    allocations: dict[str, int] = {}
+    remainders: list[tuple[float, str]] = []
+    allocated = 0
+
+    for stream, indices in stream_indices.items():
+        count = len(indices)
+        proportional = MAX_EMBEDDINGS * count / total
+        allocation = min(count, int(proportional))
+        allocations[stream] = allocation
+        allocated += allocation
+        remainders.append((proportional - allocation, stream))
+
+    remaining = MAX_EMBEDDINGS - allocated
+    for _, stream in sorted(remainders, reverse=True):
+        if remaining <= 0:
+            break
+        available = len(stream_indices[stream]) - allocations[stream]
+        if available <= 0:
+            continue
+        allocations[stream] += 1
+        remaining -= 1
+
+    selected_indices: list[int] = []
+    for stream, indices in stream_indices.items():
+        take = allocations[stream]
+        if take <= 0:
+            continue
+        if take >= len(indices):
+            selected_indices.extend(indices)
+            continue
+        sampled = rng.choice(indices, size=take, replace=False)
+        selected_indices.extend(int(idx) for idx in sampled)
+
+    selected_indices.sort()
+    sampled_embeddings = embeddings[selected_indices]
+    sampled_provenance = [provenance[idx] for idx in selected_indices]
+    return sampled_embeddings, sampled_provenance
+
+
+def detect_owner_candidate() -> dict[str, Any] | None:
+    """Detect a likely owner voice centroid from journal embeddings."""
+    load_embeddings_file, normalize_embedding, scan_segment_embeddings = (
+        _routes_helpers()
+    )
+
+    segment_count = count_segments_with_embeddings()
+    if segment_count < MIN_SEGMENTS:
+        return None
+
+    embedding_chunks: list[np.ndarray] = []
+    provenance: list[dict[str, Any]] = []
+
+    for day in day_dirs().keys():
+        for segment in scan_segment_embeddings(day):
+            stream = segment["stream"]
+            segment_key = segment["key"]
+            segment_dir = segment_path(day, segment_key, stream)
+
+            for source in segment["sources"]:
+                emb_data = load_embeddings_file(segment_dir / f"{source}.npz")
+                if emb_data is None:
+                    continue
+
+                embeddings, statement_ids = emb_data
+                if len(embeddings) == 0:
+                    continue
+
+                embedding_chunks.append(embeddings.astype(np.float32))
+                provenance.extend(
+                    {
+                        "day": day,
+                        "stream": stream,
+                        "segment_key": segment_key,
+                        "source": source,
+                        "sentence_id": int(sid),
+                    }
+                    for sid in statement_ids
+                )
+
+    if not embedding_chunks:
+        _mark_no_cluster(segment_count)
+        return None
+
+    embeddings_matrix = np.vstack(embedding_chunks)
+    embeddings_matrix, provenance = _subsample_embeddings(embeddings_matrix, provenance)
+
+    if len(embeddings_matrix) < 50:
+        _mark_no_cluster(segment_count)
+        return None
+
+    clusterer = HDBSCAN(
+        min_cluster_size=50,
+        min_samples=10,
+        metric="euclidean",
+    )
+    clusterer.fit(embeddings_matrix)
+    labels = clusterer.labels_
+
+    valid_labels = labels[labels != -1]
+    if len(valid_labels) == 0:
+        _mark_no_cluster(segment_count)
+        return None
+
+    largest_label = int(np.bincount(valid_labels).argmax())
+    cluster_indices = np.flatnonzero(labels == largest_label)
+    if len(cluster_indices) == 0:
+        _mark_no_cluster(segment_count)
+        return None
+
+    cluster_embeddings = embeddings_matrix[cluster_indices]
+    centroid = normalize_embedding(np.mean(cluster_embeddings, axis=0))
+    if centroid is None:
+        _mark_no_cluster(segment_count)
+        return None
+
+    cluster_size = int(len(cluster_indices))
+    similarities = np.dot(cluster_embeddings, centroid)
+    sorted_cluster_positions = np.argsort(similarities)[::-1]
+
+    samples: list[dict[str, Any]] = []
+    seen_segments: set[tuple[str, str, str]] = set()
+
+    for position in sorted_cluster_positions:
+        record = provenance[int(cluster_indices[position])]
+        segment_triplet = (record["day"], record["stream"], record["segment_key"])
+        if segment_triplet in seen_segments:
+            continue
+        seen_segments.add(segment_triplet)
+        samples.append(
+            {
+                **record,
+                "audio_url": _audio_url(
+                    record["day"],
+                    record["stream"],
+                    record["segment_key"],
+                    record["source"],
+                ),
+            }
+        )
+        if len(samples) == 3:
+            break
+
+    if len(samples) < 3:
+        for position in sorted_cluster_positions:
+            record = provenance[int(cluster_indices[position])]
+            sample = {
+                **record,
+                "audio_url": _audio_url(
+                    record["day"],
+                    record["stream"],
+                    record["segment_key"],
+                    record["source"],
+                ),
+            }
+            if sample in samples:
+                continue
+            samples.append(sample)
+            if len(samples) == 3:
+                break
+
+    version = _iso_now()
+    np.savez_compressed(
+        _owner_candidate_path(),
+        centroid=centroid.astype(np.float32),
+        cluster_size=np.array(cluster_size, dtype=np.int32),
+        threshold=np.array(OWNER_THRESHOLD, dtype=np.float32),
+        version=np.array(version),
+    )
+
+    update_state(
+        "voiceprint",
+        {
+            "status": "candidate",
+            "cluster_size": cluster_size,
+            "samples": samples,
+            "detected_at": version,
+        },
+    )
+
+    return {
+        "status": "candidate",
+        "cluster_size": cluster_size,
+        "samples": samples,
+    }
+
+
+def load_owner_centroid() -> tuple[np.ndarray, float] | None:
+    """Load the confirmed owner centroid and threshold for the principal entity."""
+    principal = get_journal_principal()
+    if not principal:
+        return None
+
+    centroid_path = journal_entity_memory_path(principal["id"]) / "owner_centroid.npz"
+    if not centroid_path.exists():
+        return None
+
+    try:
+        data = np.load(centroid_path, allow_pickle=False)
+        centroid = data.get("centroid")
+        threshold = data.get("threshold")
+        if centroid is None or threshold is None:
+            return None
+
+        normalized = centroid.astype(np.float32).reshape(-1)
+        norm = np.linalg.norm(normalized)
+        if norm == 0:
+            return None
+        normalized = normalized / norm
+        return normalized, float(np.asarray(threshold).item())
+    except Exception as exc:
+        logger.warning("Failed to load owner centroid %s: %s", centroid_path, exc)
+        return None
+
+
+def classify_sentences(
+    day: str, stream: str, segment_key: str, source: str
+) -> list[dict[str, Any]]:
+    """Classify segment sentences against the confirmed owner centroid."""
+    load_embeddings_file, normalize_embedding, _ = _routes_helpers()
+
+    centroid_data = load_owner_centroid()
+    if centroid_data is None:
+        return []
+
+    centroid, threshold = centroid_data
+    emb_data = load_embeddings_file(
+        segment_path(day, segment_key, stream) / f"{source}.npz"
+    )
+    if emb_data is None:
+        return []
+
+    embeddings, statement_ids = emb_data
+    results = []
+    for embedding, statement_id in zip(embeddings, statement_ids):
+        normalized = normalize_embedding(embedding)
+        if normalized is None:
+            continue
+        score = float(np.dot(normalized, centroid))
+        results.append(
+            {
+                "sentence_id": int(statement_id),
+                "is_owner": score >= threshold,
+                "score": round(score, 4),
+            }
+        )
+    return results
diff --git a/apps/speakers/routes.py b/apps/speakers/routes.py
index 10763b31f..c40706d74 100644
--- a/apps/speakers/routes.py
+++ b/apps/speakers/routes.py
@@ -13,7 +13,7 @@ import json
 import logging
 import os
 import re
-from datetime import date
+from datetime import date, datetime
 from pathlib import Path
 from typing import Any
 
@@ -28,21 +28,36 @@ from flask import (
     url_for,
 )
 
+from apps.speakers.owner import (
+    classify_sentences,
+    count_segments_with_embeddings,
+    detect_owner_candidate,
+)
 from apps.utils import log_app_action
 from convey import state
 from convey.utils import DATE_RE, error_response, format_date, success_response
+from think.awareness import get_current, update_state
 from think.entities import (
     entity_slug,
     find_matching_entity,
 )
+from think.entities.core import get_identity_names
 from think.entities.journal import (
     ensure_journal_entity_memory,
+    get_journal_principal,
     get_or_create_journal_entity,
     journal_entity_memory_path,
     load_all_journal_entities,
     load_journal_entity,
 )
-from think.utils import day_dirs, day_path, iter_segments, now_ms, segment_parse
+from think.utils import (
+    day_dirs,
+    day_path,
+    get_journal,
+    iter_segments,
+    now_ms,
+    segment_parse,
+)
 from think.utils import segment_key as validate_segment_key
 from think.utils import segment_path as get_segment_path
 
@@ -627,6 +642,156 @@ def api_sentences(day: str, stream: str, segment_key: str, source: str) -> Any:
     )
 
 
+@speakers_bp.route("/api/owner/status")
+def api_owner_status() -> Any:
+    """Return the current owner voiceprint confirmation state."""
+    voiceprint = get_current().get("voiceprint", {})
+    status = voiceprint.get("status", "none")
+
+    if status == "confirmed":
+        return jsonify({"status": "confirmed"})
+
+    if status == "candidate":
+        return jsonify(
+            {
+                "status": "candidate",
+                "cluster_size": voiceprint.get("cluster_size"),
+                "samples": voiceprint.get("samples", []),
+            }
+        )
+
+    if status == "no_cluster":
+        return jsonify({"status": "no_cluster"})
+
+    if status in {"none", "rejected"}:
+        seg_count = count_segments_with_embeddings()
+        if seg_count >= 50:
+            return jsonify(
+                {
+                    "status": "needs_detection",
+                    "segments_with_embeddings": seg_count,
+                }
+            )
+        return jsonify({"status": "none", "segments_with_embeddings": seg_count})
+
+    return jsonify({"status": "none"})
+
+
+@speakers_bp.route("/api/owner/detect", methods=["POST"])
+def api_owner_detect() -> Any:
+    """Run owner voice candidate detection."""
+    result = detect_owner_candidate()
+    if result is None:
+        return jsonify({"status": "none", "reason": "No valid cluster found"})
+    return jsonify(result)
+
+
+@speakers_bp.route("/api/owner/confirm", methods=["POST"])
+def api_owner_confirm() -> Any:
+    """Confirm the current owner voice candidate and persist the centroid."""
+    candidate_path = Path(get_journal()) / "awareness" / "owner_candidate.npz"
+    if not candidate_path.exists():
+        return error_response("No candidate available", 404)
+
+    try:
+        data = np.load(candidate_path, allow_pickle=False)
+        centroid = data["centroid"]
+        cluster_size = int(np.asarray(data["cluster_size"]).item())
+        threshold = float(np.asarray(data["threshold"]).item())
+        version = str(np.asarray(data["version"]).item())
+    except Exception as e:
+        logger.warning("Failed to load owner candidate %s: %s", candidate_path, e)
+        return error_response("No candidate available", 404)
+
+    principal = get_journal_principal()
+    if principal is None:
+        identity_names = get_identity_names()
+        if not identity_names:
+            return error_response(
+                "No principal entity found. Set up your identity first.",
+                400,
+            )
+        principal_name = identity_names[0]
+        principal = get_or_create_journal_entity(
+            entity_id=entity_slug(principal_name),
+            name=principal_name,
+            entity_type="Person",
+        )
+
+    owner_path = ensure_journal_entity_memory(principal["id"]) / "owner_centroid.npz"
+    np.savez_compressed(
+        owner_path,
+        centroid=np.asarray(centroid, dtype=np.float32).reshape(-1),
+        cluster_size=np.array(cluster_size, dtype=np.int32),
+        threshold=np.array(threshold, dtype=np.float32),
+        version=np.array(version),
+    )
+    candidate_path.unlink(missing_ok=True)
+
+    update_state(
+        "voiceprint",
+        {
+            "status": "confirmed",
+            "cluster_size": cluster_size,
+            "confirmed_at": datetime.now().isoformat(),
+        },
+    )
+
+    log_app_action(
+        app="speakers",
+        facet=None,
+        action="owner_voiceprint_confirm",
+        params={
+            "principal_id": principal["id"],
+            "cluster_size": cluster_size,
+            "version": version,
+        },
+    )
+
+    return jsonify({"status": "confirmed", "principal_id": principal["id"]})
+
+
+@speakers_bp.route("/api/owner/reject", methods=["POST"])
+def api_owner_reject() -> Any:
+    """Reject the current owner voice candidate."""
+    candidate_path = Path(get_journal()) / "awareness" / "owner_candidate.npz"
+    candidate_path.unlink(missing_ok=True)
+    update_state(
+        "voiceprint",
+        {
+            "status": "rejected",
+            "rejected_at": datetime.now().isoformat(),
+        },
+    )
+    return jsonify({"status": "needs_detection"})
+
+
+@speakers_bp.route("/api/owner/classify", methods=["POST"])
+def api_owner_classify() -> Any:
+    """Classify segment sentences against the confirmed owner centroid."""
+    data = request.get_json()
+    if not data:
+        return error_response("No data provided", 400)
+
+    day = data.get("day")
+    stream = data.get("stream")
+    segment_key = data.get("segment_key")
+    source = data.get("source")
+
+    if not all([day, stream, segment_key, source]):
+        return error_response("Missing required fields", 400)
+    if not DATE_RE.fullmatch(day):
+        return error_response("Invalid day format", 400)
+    if not validate_segment_key(segment_key):
+        return error_response("Invalid segment key", 400)
+
+    return jsonify(
+        {
+            "sentences": classify_sentences(day, stream, segment_key, source),
+        }
+    )
+
+
 @speakers_bp.route("/api/serve_audio/<day>/<path:encoded_path>")
 def serve_audio(day: str, encoded_path: str) -> Any:
     """Serve audio files for playback."""
diff --git a/apps/speakers/tests/test_owner.py b/apps/speakers/tests/test_owner.py
new file mode 100644
index 000000000..7a6f3218c
--- /dev/null
+++ b/apps/speakers/tests/test_owner.py
@@ -0,0 +1,448 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Tests for owner voice identification."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import numpy as np
+from flask import Flask
+
+from think.awareness import get_current, update_state
+
+
+def _normalized(vector: np.ndarray) -> np.ndarray:
+    return vector / np.linalg.norm(vector)
+
+
+def _write_segment(
+    journal: Path,
+    day: str,
+    stream: str,
+    segment_key: str,
+    source: str,
+    embeddings: np.ndarray,
+) -> Path:
+    segment_dir = journal / day / stream / segment_key
+    segment_dir.mkdir(parents=True, exist_ok=True)
+
+    statement_ids = np.arange(1, len(embeddings) + 1, dtype=np.int32)
+    np.savez_compressed(
+        segment_dir / f"{source}.npz",
+        embeddings=np.asarray(embeddings, dtype=np.float32),
+        statement_ids=statement_ids,
+    )
+
+    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
+
+    lines = [json.dumps({"raw": f"{source}.flac", "model": "medium.en"})]
+    for idx in range(len(embeddings)):
+        abs_seconds = base_seconds + idx * 5
+        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}",
+                    "text": f"Sentence {idx + 1}",
+                }
+            )
+        )
+
+    (segment_dir / f"{source}.jsonl").write_text("\n".join(lines) + "\n")
+    (segment_dir / f"{source}.flac").write_bytes(b"")
+    return segment_dir
+
+
+def _owner_embeddings(count: int) -> np.ndarray:
+    base = np.zeros(256, dtype=np.float32)
+    base[0] = 1.0
+    return np.repeat(base.reshape(1, -1), count, axis=0)
+
+
+def _other_cluster_embeddings(count: int) -> np.ndarray:
+    base = np.zeros(256, dtype=np.float32)
+    base[1] = 1.0
+    return np.repeat(base.reshape(1, -1), count, axis=0)
+
+
+def _candidate_path(journal: Path) -> Path:
+    return journal / "awareness" / "owner_candidate.npz"
+
+
+def test_count_segments_with_embeddings(speakers_env):
+    from apps.speakers.owner import count_segments_with_embeddings
+
+    env = speakers_env()
+    env.create_segment("20240101", "090000_300", ["mic_audio"])
+    env.create_segment("20240101", "091000_300", ["sys_audio"])
+    env.create_segment("20240102", "090000_300", ["audio"])
+
+    assert count_segments_with_embeddings() == 3
+
+
+def test_detect_owner_insufficient_segments(speakers_env):
+    from apps.speakers.owner import detect_owner_candidate
+
+    env = speakers_env()
+    for idx in range(10):
+        _write_segment(
+            env.journal,
+            "20240101",
+            "mic",
+            f"{9 + idx:02d}0000_300",
+            "audio",
+            _owner_embeddings(1),
+        )
+
+    assert detect_owner_candidate() is None
+
+
+def test_detect_owner_no_cluster(speakers_env):
+    from apps.speakers.owner import detect_owner_candidate
+
+    env = speakers_env()
+    for idx in range(50):
+        embedding = np.zeros((1, 256), dtype=np.float32)
+        embedding[0, idx] = 1.0
+        _write_segment(
+            env.journal,
+            "20240101",
+            "mic",
+            f"{9 + idx // 12:02d}{(idx % 12) * 5:02d}00_300",
+            "audio",
+            embedding,
+        )
+
+    assert detect_owner_candidate() is None
+    assert get_current()["voiceprint"]["status"] == "no_cluster"
+
+
+def test_detect_owner_basic(speakers_env):
+    from apps.speakers.owner import detect_owner_candidate
+
+    env = speakers_env()
+
+    for idx in range(55):
+        hour = 9 + (idx // 12)
+        minute = (idx % 12) * 5
+        stream = "mic" if idx % 2 == 0 else "sys"
+        _write_segment(
+            env.journal,
+            "20240101",
+            stream,
+            f"{hour:02d}{minute:02d}00_300",
+            "audio",
+            _owner_embeddings(2),
+        )
+
+    for idx in range(50):
+        hour = 9 + (idx // 12)
+        minute = (idx % 12) * 5
+        stream = "other" if idx % 2 == 0 else "other_sys"
+        _write_segment(
+            env.journal,
+            "20240102",
+            stream,
+            f"{hour:02d}{minute:02d}00_300",
+            "audio",
+            _other_cluster_embeddings(2),
+        )
+
+    result = detect_owner_candidate()
+
+    assert result is not None
+    assert result["status"] == "candidate"
+    assert result["cluster_size"] >= 50
+    assert len(result["samples"]) == 3
+    assert _candidate_path(env.journal).exists()
+    assert get_current()["voiceprint"]["status"] == "candidate"
+
+
+def test_load_owner_centroid_no_principal(speakers_env):
+    from apps.speakers.owner import load_owner_centroid
+
+    speakers_env()
+    assert load_owner_centroid() is None
+
+
+def test_load_owner_centroid_no_file(speakers_env):
+    from apps.speakers.owner import load_owner_centroid
+
+    env = speakers_env()
+    env.create_entity("Self Person", is_principal=True)
+
+    assert load_owner_centroid() is None
+
+
+def test_load_owner_centroid_success(speakers_env):
+    from apps.speakers.owner import OWNER_THRESHOLD, load_owner_centroid
+
+    env = speakers_env()
+    principal_dir = env.create_entity("Self Person", is_principal=True)
+    centroid = _normalized(np.array([1.0] + [0.0] * 255, dtype=np.float32))
+    np.savez_compressed(
+        principal_dir / "owner_centroid.npz",
+        centroid=centroid,
+        cluster_size=np.array(60, dtype=np.int32),
+        threshold=np.array(OWNER_THRESHOLD, dtype=np.float32),
+        version=np.array("2026-03-15T12:00:00"),
+    )
+
+    loaded = load_owner_centroid()
+
+    assert loaded is not None
+    loaded_centroid, threshold = loaded
+    assert np.allclose(loaded_centroid, centroid)
+    assert np.isclose(threshold, OWNER_THRESHOLD)
+
+
+def test_classify_sentences_no_centroid(speakers_env):
+    from apps.speakers.owner import classify_sentences
+
+    env = speakers_env()
+    env.create_segment("20240101", "090000_300", ["audio"], num_sentences=2)
+
+    assert classify_sentences("20240101", "test", "090000_300", "audio") == []
+
+
+def test_classify_sentences_with_centroid(speakers_env):
+    from apps.speakers.owner import OWNER_THRESHOLD, classify_sentences
+
+    env = speakers_env()
+    principal_dir = env.create_entity("Self Person", is_principal=True)
+    centroid = _normalized(np.array([1.0] + [0.0] * 255, dtype=np.float32))
+    np.savez_compressed(
+        principal_dir / "owner_centroid.npz",
+        centroid=centroid,
+        cluster_size=np.array(70, dtype=np.int32),
+        threshold=np.array(OWNER_THRESHOLD, dtype=np.float32),
+        version=np.array("2026-03-15T12:00:00"),
+    )
+
+    close = _normalized(np.array([0.95, 0.05] + [0.0] * 254, dtype=np.float32))
+    far = _normalized(np.array([0.1, 0.99] + [0.0] * 254, dtype=np.float32))
+    _write_segment(
+        env.journal,
+        "20240101",
+        "mic",
+        "090000_300",
+        "audio",
+        np.vstack([close, far]),
+    )
+
+    results = classify_sentences("20240101", "mic", "090000_300", "audio")
+
+    assert len(results) == 2
+    assert results[0]["sentence_id"] == 1
+    assert results[0]["is_owner"] is True
+    assert results[1]["sentence_id"] == 2
+    assert results[1]["is_owner"] is False
+
+
+def test_api_owner_status_none(speakers_env):
+    from apps.speakers.routes import speakers_bp
+
+    speakers_env()
+    app = Flask(__name__)
+    app.register_blueprint(speakers_bp)
+
+    with app.test_client() as client:
+        response = client.get("/app/speakers/api/owner/status")
+
+    assert response.status_code == 200
+    assert response.get_json() == {"status": "none", "segments_with_embeddings": 0}
+
+
+def test_api_owner_status_needs_detection(speakers_env):
+    from apps.speakers.routes import speakers_bp
+
+    env = speakers_env()
+    for idx in range(50):
+        env.create_segment(
+            "20240101", f"{idx // 12 + 9:02d}{(idx % 12) * 5:02d}00_300", ["audio"]
+        )
+
+    app = Flask(__name__)
+    app.register_blueprint(speakers_bp)
+
+    with app.test_client() as client:
+        response = client.get("/app/speakers/api/owner/status")
+
+    data = response.get_json()
+    assert response.status_code == 200
+    assert data["status"] == "needs_detection"
+    assert data["segments_with_embeddings"] == 50
+
+
+def test_api_owner_status_candidate(speakers_env):
+    from apps.speakers.routes import speakers_bp
+
+    speakers_env()
+    update_state(
+        "voiceprint",
+        {
+            "status": "candidate",
+            "cluster_size": 55,
+            "samples": [{"day": "20240101"}],
+        },
+    )
+    app = Flask(__name__)
+    app.register_blueprint(speakers_bp)
+
+    with app.test_client() as client:
+        response = client.get("/app/speakers/api/owner/status")
+
+    assert response.status_code == 200
+    assert response.get_json()["status"] == "candidate"
+
+
+def test_api_owner_status_no_cluster(speakers_env):
+    from apps.speakers.routes import speakers_bp
+
+    speakers_env()
+    update_state("voiceprint", {"status": "no_cluster"})
+    app = Flask(__name__)
+    app.register_blueprint(speakers_bp)
+
+    with app.test_client() as client:
+        response = client.get("/app/speakers/api/owner/status")
+
+    assert response.status_code == 200
+    assert response.get_json()["status"] == "no_cluster"
+
+
+def test_api_owner_status_confirmed(speakers_env):
+    from apps.speakers.routes import speakers_bp
+
+    speakers_env()
+    update_state("voiceprint", {"status": "confirmed"})
+    app = Flask(__name__)
+    app.register_blueprint(speakers_bp)
+
+    with app.test_client() as client:
+        response = client.get("/app/speakers/api/owner/status")
+
+    assert response.status_code == 200
+    assert response.get_json() == {"status": "confirmed"}
+
+
+def test_api_owner_classify_no_centroid(speakers_env):
+    from apps.speakers.routes import speakers_bp
+
+    env = speakers_env()
+    env.create_segment("20240101", "090000_300", ["audio"], num_sentences=2)
+    app = Flask(__name__)
+    app.register_blueprint(speakers_bp)
+
+    with app.test_client() as client:
+        response = client.post(
+            "/app/speakers/api/owner/classify",
+            json={
+                "day": "20240101",
+                "stream": "test",
+                "segment_key": "090000_300",
+                "source": "audio",
+            },
+        )
+
+    assert response.status_code == 200
+    assert response.get_json() == {"sentences": []}
+
+
+def test_api_owner_confirm(speakers_env):
+    from apps.speakers.routes import speakers_bp
+
+    env = speakers_env()
+    principal_dir = env.create_entity("Self Person", is_principal=True)
+    candidate_path = _candidate_path(env.journal)
+    candidate_path.parent.mkdir(parents=True, exist_ok=True)
+    centroid = _normalized(np.array([1.0] + [0.0] * 255, dtype=np.float32))
+    np.savez_compressed(
+        candidate_path,
+        centroid=centroid,
+        cluster_size=np.array(88, dtype=np.int32),
+        threshold=np.array(0.82, dtype=np.float32),
+        version=np.array("2026-03-15T12:00:00"),
+    )
+
+    app = Flask(__name__)
+    app.register_blueprint(speakers_bp)
+
+    with app.test_client() as client:
+        response = client.post("/app/speakers/api/owner/confirm")
+
+    assert response.status_code == 200
+    assert response.get_json()["status"] == "confirmed"
+    assert not candidate_path.exists()
+    assert (principal_dir / "owner_centroid.npz").exists()
+    assert get_current()["voiceprint"]["status"] == "confirmed"
+
+
+def test_api_owner_reject(speakers_env):
+    from apps.speakers.routes import speakers_bp
+
+    env = speakers_env()
+    candidate_path = _candidate_path(env.journal)
+    candidate_path.parent.mkdir(parents=True, exist_ok=True)
+    candidate_path.write_bytes(b"test")
+
+    app = Flask(__name__)
+    app.register_blueprint(speakers_bp)
+
+    with app.test_client() as client:
+        response = client.post("/app/speakers/api/owner/reject")
+
+    assert response.status_code == 200
+    assert response.get_json() == {"status": "needs_detection"}
+    assert not candidate_path.exists()
+    assert get_current()["voiceprint"]["status"] == "rejected"
+
+
+def test_api_owner_detect(speakers_env):
+    from apps.speakers.routes import speakers_bp
+
+    env = speakers_env()
+    for idx in range(55):
+        hour = 9 + (idx // 12)
+        minute = (idx % 12) * 5
+        stream = "mic" if idx % 2 == 0 else "sys"
+        _write_segment(
+            env.journal,
+            "20240101",
+            stream,
+            f"{hour:02d}{minute:02d}00_300",
+            "audio",
+            _owner_embeddings(2),
+        )
+    for idx in range(50):
+        hour = 9 + (idx // 12)
+        minute = (idx % 12) * 5
+        stream = "other" if idx % 2 == 0 else "other_sys"
+        _write_segment(
+            env.journal,
+            "20240102",
+            stream,
+            f"{hour:02d}{minute:02d}00_300",
+            "audio",
+            _other_cluster_embeddings(2),
+        )
+
+    app = Flask(__name__)
+    app.register_blueprint(speakers_bp)
+
+    with app.test_client() as client:
+        response = client.post("/app/speakers/api/owner/detect")
+
+    data = response.get_json()
+    assert response.status_code == 200
+    assert data["status"] == "candidate"
+    assert data["cluster_size"] >= 50
diff --git a/apps/speakers/workspace.html b/apps/speakers/workspace.html
index 945836504..5047d1fbd 100644
--- a/apps/speakers/workspace.html
+++ b/apps/speakers/workspace.html
@@ -475,9 +475,98 @@
   opacity: 0.8;
 }
 
+.spk-owner-banner {
+  display: none;
+  margin-bottom: 20px;
+  padding: 16px 18px;
+  border-radius: 12px;
+  border: 1px solid #bfdbfe;
+  background: linear-gradient(135deg, #eff6ff 0%, #e0f2fe 100%);
+}
+
+.spk-owner-panel {
+  display: flex;
+  flex-direction: column;
+  gap: 14px;
+}
+
+.spk-owner-title {
+  font-size: 18px;
+  font-weight: 600;
+  color: #1e3a8a;
+}
+
+.spk-owner-copy {
+  font-size: 14px;
+  color: #1f2937;
+  line-height: 1.5;
+}
+
+.spk-owner-samples {
+  display: grid;
+  grid-template-columns: repeat(auto-fit, minmax(220px, 1fr));
+  gap: 12px;
+}
+
+.spk-owner-sample {
+  padding: 12px;
+  border-radius: 10px;
+  border: 1px solid #dbeafe;
+  background: rgba(255, 255, 255, 0.8);
+}
+
+.spk-owner-sample-title {
+  font-size: 12px;
+  font-weight: 600;
+  color: #1d4ed8;
+  margin-bottom: 8px;
+}
+
+.spk-owner-sample-meta {
+  font-size: 12px;
+  color: #475569;
+  margin-bottom: 8px;
+}
+
+.spk-owner-sample audio {
+  width: 100%;
+}
+
+.spk-owner-actions {
+  display: flex;
+  gap: 10px;
+  flex-wrap: wrap;
+}
+
+.spk-owner-btn {
+  padding: 8px 14px;
+  border-radius: 8px;
+  border: 1px solid transparent;
+  font-size: 13px;
+  font-weight: 600;
+  cursor: pointer;
+}
+
+.spk-owner-btn:disabled {
+  opacity: 0.65;
+  cursor: wait;
+}
+
+.spk-owner-btn-confirm {
+  background: #16a34a;
+  color: #fff;
+}
+
+.spk-owner-btn-reject {
+  background: #fff;
+  border-color: #93c5fd;
+  color: #1d4ed8;
+}
+
 </style>
 
 <div class="spk-wrap">
+  <div id="spkOwnerBanner" class="spk-owner-banner"></div>
   <div class="spk-layout">
     <!-- Segment list -->
     <div class="spk-segments">
@@ -503,6 +592,7 @@
   const day = '{{ day }}';
   const segmentList = document.getElementById('spkSegmentList');
   const detailPanel = document.getElementById('spkDetail');
+  const ownerBanner = document.getElementById('spkOwnerBanner');
 
   let segments = [];
   let selectedSegment = null;
@@ -511,8 +601,10 @@
   let currentSentences = [];  // Sentence data with offsets for time tracking
   let playingSentenceId = null;  // Currently playing/highlighted sentence
   let filterMatched = false;  // Toggle: hide high-confidence matches
+  let ownerDetectionInFlight = false;
 
   // Load segments on page load
+  checkOwnerStatus();
   loadSegments();
 
   // Listen for hash changes
@@ -521,6 +613,107 @@
     if (hash) selectSegmentByKey(hash);
   });
 
+  function checkOwnerStatus() {
+    fetch('/app/speakers/api/owner/status')
+      .then(r => r.json())
+      .then(data => {
+        if (data.status === 'needs_detection') {
+          renderOwnerBanner('Analyzing voice patterns...');
+          if (!ownerDetectionInFlight) {
+            ownerDetectionInFlight = true;
+            fetch('/app/speakers/api/owner/detect', { method: 'POST' })
+              .then(r => r.json())
+              .then(() => {
+                ownerDetectionInFlight = false;
+                checkOwnerStatus();
+              })
+              .catch(() => {
+                ownerDetectionInFlight = false;
+                hideOwnerBanner();
+              });
+          }
+          return;
+        }
+
+        ownerDetectionInFlight = false;
+        if (data.status === 'no_cluster') {
+          hideOwnerBanner();
+          return;
+        }
+        if (data.status === 'candidate') {
+          renderOwnerCandidate(data);
+          return;
+        }
+
+        hideOwnerBanner();
+      })
+      .catch(() => {
+        hideOwnerBanner();
+      });
+  }
+
+  function renderOwnerBanner(message) {
+    ownerBanner.style.display = 'block';
+    ownerBanner.innerHTML = `
+      <div class="spk-owner-panel">
+        <div class="spk-owner-title">Owner voiceprint</div>
+        <div class="spk-owner-copy">${escapeHtml(message)}</div>
+      </div>
+    `;
+  }
+
+  function hideOwnerBanner() {
+    ownerBanner.style.display = 'none';
+    ownerBanner.innerHTML = '';
+  }
+
+  function renderOwnerCandidate(data) {
+    const samples = data.samples || [];
+    ownerBanner.style.display = 'block';
+    ownerBanner.innerHTML = `
+      <div class="spk-owner-panel">
+        <div class="spk-owner-title">Is this your voice?</div>
+        <div class="spk-owner-copy">
+          We found a likely owner voice pattern from ${escapeHtml(data.cluster_size || 0)} voice samples. Review a few examples before confirming.
+        </div>
+        <div class="spk-owner-samples">
+          ${samples.map((sample, index) => `
+            <div class="spk-owner-sample">
+              <div class="spk-owner-sample-title">Sample ${index + 1}</div>
+              <div class="spk-owner-sample-meta">
+                ${escapeHtml(sample.day)} · ${escapeHtml(sample.stream)} · ${escapeHtml(sample.segment_key)}
+              </div>
+              <audio controls preload="metadata" src="${sample.audio_url}"></audio>
+            </div>
+          `).join('')}
+        </div>
+        <div class="spk-owner-actions">
+          <button class="spk-owner-btn spk-owner-btn-confirm" id="spkOwnerConfirm">Yes, that's me</button>
+          <button class="spk-owner-btn spk-owner-btn-reject" id="spkOwnerReject">No, try again</button>
+        </div>
+      </div>
+    `;
+
+    document.getElementById('spkOwnerConfirm')?.addEventListener('click', () => {
+      submitOwnerChoice('/app/speakers/api/owner/confirm');
+    });
+    document.getElementById('spkOwnerReject')?.addEventListener('click', () => {
+      submitOwnerChoice('/app/speakers/api/owner/reject');
+    });
+  }
+
+  function submitOwnerChoice(path) {
+    ownerBanner.querySelectorAll('button').forEach(btn => { btn.disabled = true; });
+    fetch(path, { method: 'POST' })
+      .then(r => r.json())
+      .then(() => {
+        checkOwnerStatus();
+      })
+      .catch(() => {
+        ownerBanner.querySelectorAll('button').forEach(btn => { btn.disabled = false; });
+      });
+  }
+
   function loadSegments() {
     fetch(`/app/speakers/api/segments/${day}`)
       .then(r => r.json())
diff --git a/convey/apps.py b/convey/apps.py
index 26ee53b9a..1d0f127c9 100644
--- a/convey/apps.py
+++ b/convey/apps.py
@@ -79,7 +79,7 @@ class AttentionItem:
 
 
 def _resolve_attention(awareness_current: dict) -> AttentionItem | None:
-    """Check attention sources P0-P3, return highest priority or None."""
+    """Check attention sources P0-P4, return highest priority or None."""
     # P0: Cortex errors
     try:
         import json
@@ -200,6 +200,17 @@ def _resolve_attention(awareness_current: dict) -> AttentionItem | None:
         except Exception:
             pass
 
+    # P4: Owner voiceprint candidate ready for confirmation
+    voiceprint = awareness_current.get("voiceprint", {})
+    if voiceprint.get("status") == "candidate":
+        cluster_size = voiceprint.get("cluster_size", 0)
+        placeholder = "Voice pattern detected — confirm in Speakers"
+        context = [
+            f"System detected owner voice pattern from {cluster_size} voice samples. "
+            "Direct user to the Speakers app (/app/speakers) to confirm their voiceprint."
+        ]
+        return AttentionItem(placeholder_text=placeholder, context_lines=context)
+
     return None