diff --git a/solstone/think/cluster.py b/solstone/think/cluster.py
index 7e9008f13..6a2f6ed9c 100644
--- a/solstone/think/cluster.py
+++ b/solstone/think/cluster.py
@@ -341,21 +341,25 @@ def _group_entries(
 
 
 def _count_by_source(entries: list[dict[str, Any]]) -> dict[str, int]:
-    """Count entries by source type (prefix).
+    """Count entries by returned source-count key.
 
-    Maps the internal prefix names to source config names:
+    Maps internal entry prefixes to returned source-count keys:
     - "transcript" -> "transcripts"
     - "percept" -> "percepts"
-    - "agent_output" -> "agents"
+    - "agent_output" -> "talents"
+
+    Note: cluster input config still uses the internal "agents" source key for
+    talent-output markdown discovery. Returned counts use the public talent
+    frontmatter vocabulary, so "agent_output" is counted as "talents".
 
     Returns:
-        Dict with counts for each source type, e.g., {"transcripts": 2, "percepts": 1, "agents": 0}
+        Dict with counts for each source type, e.g., {"transcripts": 2, "percepts": 1, "talents": 0}
     """
     # Map internal prefix to source config name
     prefix_to_source = {
         "transcript": "transcripts",
         "percept": "percepts",
-        "agent_output": "agents",
+        "agent_output": "talents",
     }
 
     counts = Counter(prefix_to_source.get(e["prefix"], e["prefix"]) for e in entries)
@@ -364,7 +368,7 @@ def _count_by_source(entries: list[dict[str, Any]]) -> dict[str, int]:
     return {
         "transcripts": counts.get("transcripts", 0),
         "percepts": counts.get("percepts", 0),
-        "agents": counts.get("agents", 0),
+        "talents": counts.get("talents", 0),
     }
 
 
@@ -724,9 +728,9 @@ def cluster(
 
     Returns:
         Tuple of (markdown, source_counts) where source_counts is a dict
-        with keys "transcripts", "percepts", "agents" mapping to entry counts.
+        with keys "transcripts", "percepts", "talents" mapping to entry counts.
     """
-    empty_counts = {"transcripts": 0, "screen": 0, "agents": 0}
+    empty_counts = {"transcripts": 0, "percepts": 0, "talents": 0}
 
     day_dir = str(day_path(day))
     # day_path now ensures dir exists, but check anyway for safety
@@ -767,9 +771,9 @@ def cluster_period(
 
     Returns:
         Tuple of (markdown, source_counts) where source_counts is a dict
-        with keys "transcripts", "percepts", "agents" mapping to entry counts.
+        with keys "transcripts", "percepts", "talents" mapping to entry counts.
     """
-    empty_counts = {"transcripts": 0, "screen": 0, "agents": 0}
+    empty_counts = {"transcripts": 0, "percepts": 0, "talents": 0}
 
     segment_dir = _find_segment_dir(day, segment, stream)
 
@@ -844,12 +848,12 @@ def cluster_span(
 
     Returns:
         Tuple of (markdown, source_counts) where source_counts is a dict
-        with keys "transcripts", "percepts", "agents" mapping to entry counts.
+        with keys "transcripts", "percepts", "talents" mapping to entry counts.
 
     Raises:
         ValueError: If any segment directories are missing
     """
-    empty_counts = {"transcripts": 0, "screen": 0, "agents": 0}
+    empty_counts = {"transcripts": 0, "percepts": 0, "talents": 0}
 
     # Validate all segments in span exist upfront (fail fast)
     missing = []
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
index 48898846d..a4b87a10d 100644
--- a/tests/test_cluster.py
+++ b/tests/test_cluster.py
@@ -13,6 +13,63 @@ import pytest
 from solstone.observe.processing_record import STATE_EMPTY
 from solstone.think.utils import day_path
 
+DAY = "20240101"
+SEGMENT = "120000_300"
+
+
+def _patch_prepare_config_model_deps(monkeypatch) -> None:
+    from solstone.think.models import TIER_LITE
+
+    monkeypatch.setattr(
+        "solstone.think.models.resolve_provider",
+        lambda _context, _type: ("google", "gemini-test"),
+    )
+    monkeypatch.setattr(
+        "solstone.think.models.type_default_is_local", lambda _type: False
+    )
+    monkeypatch.setattr(
+        "solstone.think.models._resolve_tier",
+        lambda _context, _type: TIER_LITE,
+    )
+    monkeypatch.setattr("solstone.think.models.load_health_status", lambda: {})
+    monkeypatch.setattr(
+        "solstone.think.models.should_recheck_health", lambda _health: False
+    )
+    monkeypatch.setattr(
+        "solstone.think.models.is_provider_model_interface_healthy",
+        lambda _provider, _model, _type, _health: True,
+    )
+
+
+def _write_probe_talent(talent_dir: Path, load: dict | str) -> str:
+    name = "required_talent_probe"
+    metadata = {
+        "type": "generate",
+        "schedule": "segment",
+        "priority": 10,
+        "output": "md",
+        "load": {
+            "transcripts": False,
+            "percepts": False,
+            "talents": load,
+        },
+    }
+    (talent_dir / f"{name}.md").write_text(
+        f"{json.dumps(metadata, indent=2)}\n\nProbe prompt\n",
+        encoding="utf-8",
+    )
+    return name
+
+
+def _seed_talent_output(
+    root: Path,
+    name: str,
+    content: str = "Agent output with enough substantive content for input gating.",
+) -> None:
+    segment_dir = root / "chronicle" / DAY / "default" / SEGMENT / "talents"
+    segment_dir.mkdir(parents=True, exist_ok=True)
+    (segment_dir / f"{name}.md").write_text(content, encoding="utf-8")
+
 
 def test_cluster(tmp_path, monkeypatch):
     """Test cluster() uses transcripts and agent output summaries (*.md files)."""
@@ -34,7 +91,7 @@ def test_cluster(tmp_path, monkeypatch):
         "20240101", sources={"transcripts": True, "percepts": False, "agents": True}
     )
     assert counts["transcripts"] == 1
-    assert counts["agents"] == 1
+    assert counts["talents"] == 1
     assert "### Transcript" in result
     # Now uses insight rendering: "### {stem} summary"
     assert "screen summary" in result
@@ -478,7 +535,7 @@ def test_cluster_with_agent_filter_dict(tmp_path, monkeypatch):
     )
 
     assert counts["transcripts"] == 1
-    assert counts["agents"] == 1  # Only entities should be counted
+    assert counts["talents"] == 1  # Only entities should be counted
     assert "Entity extraction results" in result
     assert "Meeting summary results" not in result
     assert "Flow analysis results" not in result
@@ -511,7 +568,7 @@ def test_cluster_with_agent_filter_multiple(tmp_path, monkeypatch):
     )
 
     assert counts["transcripts"] == 1
-    assert counts["agents"] == 2  # entities + meetings
+    assert counts["talents"] == 2  # entities + meetings
     assert "Entity extraction results" in result
     assert "Meeting summary results" in result
     assert "Flow analysis results" not in result
@@ -544,7 +601,7 @@ def test_cluster_with_agent_filter_app_namespaced(tmp_path, monkeypatch):
     )
 
     assert counts["transcripts"] == 1
-    assert counts["agents"] == 1  # Only app:name
+    assert counts["talents"] == 1  # Only app:name
     assert "System entity results" not in result
     assert "App agent results" in result
 
@@ -569,10 +626,93 @@ def test_cluster_with_empty_agent_filter(tmp_path, monkeypatch):
     )
 
     assert counts["transcripts"] == 1
-    assert counts["agents"] == 0
+    assert counts["talents"] == 0
     assert "Entity extraction results" not in result
 
 
+def test_cluster_count_keys_match_for_empty_and_populated_results(
+    tmp_path, monkeypatch
+):
+    monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
+    empty_day = day_path("20240102")
+    empty_day.mkdir(parents=True, exist_ok=True)
+
+    mod = importlib.import_module("solstone.think.cluster")
+    _empty_result, empty_counts = mod.cluster(
+        "20240102",
+        sources={"transcripts": True, "percepts": True, "agents": True},
+    )
+
+    populated_day = day_path("20240103")
+    segment = populated_day / "default" / "120000_300"
+    (segment / "talents").mkdir(parents=True)
+    (segment / "audio.jsonl").write_text(
+        '{"raw": "audio.flac"}\n{"start": "00:00:01", "text": "hello"}\n',
+        encoding="utf-8",
+    )
+    (segment / "screen.jsonl").write_text(
+        '{"raw": "screen.webm"}\n'
+        '{"timestamp": 10, "analysis": {"primary": "code_editor"}}\n',
+        encoding="utf-8",
+    )
+    (segment / "talents" / "sense.md").write_text(
+        "Sense markdown with entity candidates.",
+        encoding="utf-8",
+    )
+    _populated_result, populated_counts = mod.cluster(
+        "20240103",
+        sources={"transcripts": True, "percepts": True, "agents": True},
+    )
+
+    assert (
+        set(empty_counts)
+        == set(populated_counts)
+        == {
+            "transcripts",
+            "percepts",
+            "talents",
+        }
+    )
+
+
+@pytest.mark.parametrize(
+    ("load", "agent_name", "expected_skip"),
+    [
+        ("required", "sense", None),
+        ({"sense": "required"}, "sense", None),
+        ("required", None, "missing_required_talents"),
+    ],
+)
+def test_prepare_config_required_talent_source_uses_talent_count_key(
+    tmp_path,
+    monkeypatch,
+    load,
+    agent_name,
+    expected_skip,
+):
+    from solstone.think import talent as talent_module
+    from solstone.think import talents
+
+    monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
+    monkeypatch.setattr(talent_module, "TALENT_DIR", tmp_path / "talents")
+    talent_module.TALENT_DIR.mkdir()
+    _patch_prepare_config_model_deps(monkeypatch)
+    prompt_name = _write_probe_talent(talent_module.TALENT_DIR, load)
+    if agent_name is not None:
+        _seed_talent_output(tmp_path, agent_name)
+
+    config = talents.prepare_config(
+        {
+            "name": prompt_name,
+            "day": DAY,
+            "segment": SEGMENT,
+            "output": "md",
+        }
+    )
+
+    assert config.get("skip_reason") == expected_skip
+
+
 def test_filename_to_agent_key():
     """Test _filename_to_agent_key conversion."""
     from solstone.think.cluster import _filename_to_agent_key
diff --git a/tests/test_cluster_full.py b/tests/test_cluster_full.py
index a13e82496..0bfdfbb5e 100644
--- a/tests/test_cluster_full.py
+++ b/tests/test_cluster_full.py
@@ -27,7 +27,7 @@ def test_cluster_full(tmp_path, monkeypatch):
     )
     # Transcript entries come from 2 segments on 20240101 (default + import.apple)
     assert counts["transcripts"] == 2
-    assert counts["agents"] == 2  # audio.md + screen.md
+    assert counts["talents"] == 2  # audio.md + screen.md
     assert "### Transcript" in md
     # Now uses insight format: "### {stem} summary"
     assert "### screen summary" in md
diff --git a/tests/test_think_segment_prephase.py b/tests/test_think_segment_prephase.py
index 6bec7de6b..64c2b8c2f 100644
--- a/tests/test_think_segment_prephase.py
+++ b/tests/test_think_segment_prephase.py
@@ -926,6 +926,44 @@ def test_json_cache_reuse_requires_current_schema_validation(
     assert out.read_text(encoding="utf-8") == '{"ok": true}'
 
 
+def test_identity_hash_ignores_source_count_vocabulary():
+    from solstone.think import talents
+
+    config = {
+        "type": "generate",
+        "name": "sense",
+        "provider": "google",
+        "model": "x",
+        "prompt": "think about this segment",
+        "day": DAY,
+        "segment": ACTIVE_SEGMENT,
+        "stream": STREAM,
+        "output": "json",
+        "json_schema": {
+            "type": "object",
+            "required": ["ok"],
+            "properties": {"ok": {"type": "boolean"}},
+        },
+        "schedule": "segment",
+        "sources": {
+            "transcripts": True,
+            "percepts": False,
+            "talents": {"sense": True},
+        },
+        "source_counts": {"transcripts": 1, "percepts": 0, "agents": 1},
+    }
+    runtime_schema = talents.hydrate_runtime_enums(config["json_schema"])
+    renamed_counts = {
+        **config,
+        "source_counts": {"transcripts": 1, "percepts": 0, "talents": 1},
+    }
+
+    assert talents._identity_hash(config, runtime_schema) == talents._identity_hash(
+        renamed_counts,
+        runtime_schema,
+    )
+
+
 def test_activity_replay_dedupes_records_and_preserves_non_refresh(
     tmp_path,
     monkeypatch,