fix: follow source claude active version

Author: SeemSeam

docs/claude-binary-cache-dedup-plan.md

@@ -101,6 +101,14 @@ Default managed Claude launches should still use an agent-scoped private
 `HOME`, but the Claude executable/version cache should resolve outside that
 private home.
 
+When the source user home has an active Claude Code binary under
+`~/.local/bin/claude` pointing into `~/.local/share/claude/versions/`, CCB should
+treat that active source-home version as the preferred managed Claude binary
+version. Managed startup may copy that version into the CCB provider cache, but
+the managed `.local/bin/claude` link should follow the source-home active
+version instead of selecting a different version only because it is already
+present or numerically newer in the shared cache.
+
 Preferred target:
 
 ```text

lib/cli/services/provider_hooks.py

@@ -263,7 +263,7 @@ def _route_claude_binary_cache_if_possible(*, layout, home_root: Path) -> None:
         cache_root = layout.ensure_provider_external_cache_dir('claude')
     except Exception:
         return
-    route_claude_binary_cache(home_root, cache_root)
+    route_claude_binary_cache(home_root, cache_root, source_home=current_provider_source_home())
 
 
 def _claude_versions_cache_signature(versions_dir: Path) -> dict[str, object] | None:

lib/provider_backends/claude/launcher_runtime/binary_cache.py

@@ -9,10 +9,17 @@
 _IGNORED_VERSION_ENTRIES = {'.DS_Store'}
 
 
-def route_claude_binary_cache(home_root: Path, shared_cache_root: Path) -> dict[str, object]:
+def route_claude_binary_cache(
+    home_root: Path,
+    shared_cache_root: Path,
+    *,
+    source_home: Path | None = None,
+) -> dict[str, object]:
     home = Path(home_root).expanduser().resolve(strict=False)
     shared_versions_dir = Path(shared_cache_root).expanduser().resolve(strict=False) / 'versions'
     versions_dir = home / '.local' / 'share' / 'claude' / 'versions'
+    source_active_version = _source_active_version(source_home, managed_home=home)
+    source_active_version_name = source_active_version.name if source_active_version is not None else ''
 
     try:
         shared_versions_dir.mkdir(parents=True, exist_ok=True)
@@ -27,7 +34,18 @@ def route_claude_binary_cache(home_root: Path, shared_cache_root: Path) -> dict[
 
     if versions_dir.is_symlink():
         if _same_path(versions_dir, shared_versions_dir):
-            active_version_name = _ensure_latest_claude_link(home, shared_versions_dir)
+            failure = _copy_source_active_version_to_shared(
+                source_active_version,
+                shared_versions_dir=shared_versions_dir,
+                versions_dir=versions_dir,
+            )
+            if failure is not None:
+                return failure
+            active_version_name = _ensure_claude_link(
+                home,
+                shared_versions_dir,
+                preferred_version_name=source_active_version_name,
+            )
             write_projected_marker(
                 versions_dir,
                 label=_PROJECTION_LABEL,
@@ -75,16 +93,27 @@ def route_claude_binary_cache(home_root: Path, shared_cache_root: Path) -> dict[
             versions_dir=versions_dir,
             version_names=scan['version_names'],
         )
+        if failure is not None:
+            return failure
+        failure = _copy_source_active_version_to_shared(
+            source_active_version,
+            shared_versions_dir=shared_versions_dir,
+            versions_dir=versions_dir,
+        )
         if failure is not None:
             return failure
         linked = _link_versions_dir(
             versions_dir,
             shared_versions_dir,
             reason='migrated_symlink' if scan['version_paths'] else 'linked_empty',
-            version_names=scan['version_names'],
+            version_names=_version_names(shared_versions_dir),
         )
         if linked.get('status') == 'ok':
-            linked['active_version_name'] = _ensure_latest_claude_link(home, shared_versions_dir) or ''
+            linked['active_version_name'] = _ensure_claude_link(
+                home,
+                shared_versions_dir,
+                preferred_version_name=source_active_version_name,
+            ) or ''
         if scan['ignored_entries'] and linked.get('status') == 'ok':
             linked['warnings'] = tuple(scan['ignored_entries'])
         return linked
@@ -98,11 +127,26 @@ def route_claude_binary_cache(home_root: Path, shared_cache_root: Path) -> dict[
         )
 
     if not versions_dir.exists():
-        return _link_versions_dir(
+        failure = _copy_source_active_version_to_shared(
+            source_active_version,
+            shared_versions_dir=shared_versions_dir,
+            versions_dir=versions_dir,
+        )
+        if failure is not None:
+            return failure
+        linked = _link_versions_dir(
             versions_dir,
             shared_versions_dir,
             reason='linked_empty',
+            version_names=_version_names(shared_versions_dir),
         )
+        if linked.get('status') == 'ok':
+            linked['active_version_name'] = _ensure_claude_link(
+                home,
+                shared_versions_dir,
+                preferred_version_name=source_active_version_name,
+            ) or ''
+        return linked
 
     scan = _scan_versions_dir(versions_dir)
     if scan['unknown_entries']:
@@ -121,17 +165,28 @@ def route_claude_binary_cache(home_root: Path, shared_cache_root: Path) -> dict[
         versions_dir=versions_dir,
         version_names=scan['version_names'],
     )
+    if failure is not None:
+        return failure
+    failure = _copy_source_active_version_to_shared(
+        source_active_version,
+        shared_versions_dir=shared_versions_dir,
+        versions_dir=versions_dir,
+    )
     if failure is not None:
         return failure
 
     linked = _link_versions_dir(
         versions_dir,
         shared_versions_dir,
         reason='migrated' if scan['version_paths'] else 'linked_empty',
-        version_names=scan['version_names'],
+        version_names=_version_names(shared_versions_dir),
     )
     if linked.get('status') == 'ok':
-        linked['active_version_name'] = _ensure_latest_claude_link(home, shared_versions_dir) or ''
+        linked['active_version_name'] = _ensure_claude_link(
+            home,
+            shared_versions_dir,
+            preferred_version_name=source_active_version_name,
+        ) or ''
     if scan['ignored_entries'] and linked.get('status') == 'ok':
         linked['warnings'] = tuple(scan['ignored_entries'])
     return linked
@@ -171,6 +226,22 @@ def _copy_versions_to_shared(
     return None
 
 
+def _copy_source_active_version_to_shared(
+    source_active_version: Path | None,
+    *,
+    shared_versions_dir: Path,
+    versions_dir: Path,
+) -> dict[str, object] | None:
+    if source_active_version is None:
+        return None
+    return _copy_versions_to_shared(
+        version_paths=(source_active_version,),
+        shared_versions_dir=shared_versions_dir,
+        versions_dir=versions_dir,
+        version_names=(source_active_version.name,),
+    )
+
+
 def _scan_versions_dir(versions_dir: Path) -> dict[str, object]:
     version_paths: list[Path] = []
     unknown_entries: list[str] = []
@@ -237,25 +308,34 @@ def _link_versions_dir(
     )
 
 
-def _ensure_latest_claude_link(home: Path, shared_versions_dir: Path) -> str:
-    latest = _newest_version_path(shared_versions_dir)
-    if latest is None:
+def _ensure_claude_link(home: Path, shared_versions_dir: Path, *, preferred_version_name: str = '') -> str:
+    target_version = _preferred_or_newest_version_path(shared_versions_dir, preferred_version_name=preferred_version_name)
+    if target_version is None:
         return ''
-    executable = _version_executable_path(latest)
+    executable = _version_executable_path(target_version)
     if executable is None:
         return ''
     link = home / '.local' / 'bin' / 'claude'
     try:
         if link.is_symlink() and _same_path(link, executable):
-            return latest.name
+            return target_version.name
         if link.exists() and not link.is_symlink():
             return ''
         link.parent.mkdir(parents=True, exist_ok=True)
         link.unlink(missing_ok=True)
         link.symlink_to(executable)
     except Exception:
         return ''
-    return latest.name
+    return target_version.name
+
+
+def _preferred_or_newest_version_path(versions_dir: Path, *, preferred_version_name: str) -> Path | None:
+    preferred = str(preferred_version_name or '').strip()
+    if preferred:
+        candidate = versions_dir / preferred
+        if _looks_like_claude_version_name(candidate.name) and _version_executable_path(candidate) is not None:
+            return candidate
+    return _newest_version_path(versions_dir)
 
 
 def _newest_version_path(versions_dir: Path) -> Path | None:
@@ -300,6 +380,33 @@ def _version_names(versions_dir: Path) -> tuple[str, ...]:
         return ()
 
 
+def _source_active_version(source_home: Path | None, *, managed_home: Path) -> Path | None:
+    if source_home is None:
+        return None
+    try:
+        home = Path(source_home).expanduser().resolve(strict=False)
+    except Exception:
+        return None
+    if _same_path(home, managed_home):
+        return None
+    link = home / '.local' / 'bin' / 'claude'
+    if not link.is_symlink():
+        return None
+    try:
+        target = link.resolve(strict=True)
+        versions_dir = (home / '.local' / 'share' / 'claude' / 'versions').resolve(strict=True)
+        relative = target.relative_to(versions_dir)
+    except Exception:
+        return None
+    if not relative.parts:
+        return None
+    version_name = relative.parts[0]
+    if not _looks_like_claude_version_name(version_name):
+        return None
+    candidate = versions_dir / version_name
+    return candidate if _version_executable_path(candidate) is not None else None
+
+
 def _same_path(left: Path, right: Path) -> bool:
     try:
         return left.resolve() == right.resolve()

test/test_provider_hook_settings.py

@@ -520,6 +520,41 @@ def test_prepare_provider_workspace_routes_claude_binary_cache_to_external_cache
         assert all(event.get('event_type') != 'claude_binary_cache_drift' for event in events)
 
 
+def test_prepare_provider_workspace_routes_claude_binary_cache_from_home_active_version(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    project_root = tmp_path / 'repo'
+    workspace = project_root / 'workspace'
+    system_home = tmp_path / 'system-home'
+    active_binary = system_home / '.local' / 'share' / 'claude' / 'versions' / '2.1.141' / 'claude'
+    active_binary.parent.mkdir(parents=True, exist_ok=True)
+    active_binary.write_text('home active binary\n', encoding='utf-8')
+    active_binary.chmod(0o755)
+    (system_home / '.local' / 'bin').mkdir(parents=True, exist_ok=True)
+    (system_home / '.local' / 'bin' / 'claude').symlink_to(active_binary)
+    monkeypatch.setenv('HOME', str(system_home))
+    monkeypatch.setenv('XDG_CACHE_HOME', str(tmp_path / 'xdg-cache'))
+    layout = PathLayout(project_root)
+
+    prepare_provider_workspace(
+        layout=layout,
+        spec=_spec('agent1'),
+        workspace_path=workspace,
+        completion_dir=layout.agent_provider_runtime_dir('agent1', 'claude') / 'completion',
+        agent_name='agent1',
+        refresh_profile=True,
+    )
+
+    managed_home = layout.agent_provider_state_dir('agent1', 'claude') / 'home'
+    shared_active = layout.provider_external_cache_dir('claude') / 'versions' / '2.1.141'
+    assert (managed_home / '.local' / 'share' / 'claude' / 'versions').resolve() == (
+        layout.provider_external_cache_dir('claude') / 'versions'
+    ).resolve()
+    assert (shared_active / 'claude').read_text(encoding='utf-8') == 'home active binary\n'
+    assert (managed_home / '.local' / 'bin' / 'claude').resolve() == (shared_active / 'claude').resolve()
+
+
 def test_prepare_provider_workspace_keeps_unknown_claude_versions_dir_unmodified(
     tmp_path: Path,
     monkeypatch,

test/test_provider_profiles.py

@@ -1077,6 +1077,39 @@ def test_route_claude_binary_cache_points_existing_shared_home_to_latest_version
     assert (home / '.local' / 'bin' / 'claude').resolve() == new_binary.resolve()
 
 
+def test_route_claude_binary_cache_prefers_source_home_active_version(tmp_path: Path) -> None:
+    home = tmp_path / 'home'
+    source_home = tmp_path / 'source-home'
+    shared_cache = tmp_path / 'shared-cache' / 'claude'
+    shared_versions = shared_cache / 'versions'
+    old_shared = shared_versions / '2.1.139'
+    newer_shared = shared_versions / '2.1.140'
+    source_active = source_home / '.local' / 'share' / 'claude' / 'versions' / '2.1.138'
+    old_shared.parent.mkdir(parents=True, exist_ok=True)
+    old_shared.write_text('old shared executable\n', encoding='utf-8')
+    newer_shared.write_text('newer shared executable\n', encoding='utf-8')
+    source_active.parent.mkdir(parents=True, exist_ok=True)
+    source_active.write_text('source active executable\n', encoding='utf-8')
+    source_active.chmod(0o755)
+    (source_home / '.local' / 'bin').mkdir(parents=True, exist_ok=True)
+    (source_home / '.local' / 'bin' / 'claude').symlink_to(source_active)
+    versions = home / '.local' / 'share' / 'claude' / 'versions'
+    versions.parent.mkdir(parents=True, exist_ok=True)
+    versions.symlink_to(shared_versions, target_is_directory=True)
+    (home / '.local' / 'bin').mkdir(parents=True, exist_ok=True)
+    (home / '.local' / 'bin' / 'claude').symlink_to(newer_shared)
+
+    result = route_claude_binary_cache(home, shared_cache, source_home=source_home)
+
+    copied_active = shared_versions / '2.1.138'
+    assert result['status'] == 'ok'
+    assert result['reason'] == 'already_shared'
+    assert result['active_version_name'] == '2.1.138'
+    assert copied_active.read_text(encoding='utf-8') == 'source active executable\n'
+    assert copied_active.stat().st_mode & 0o111
+    assert (home / '.local' / 'bin' / 'claude').resolve() == copied_active.resolve()
+
+
 def test_materialize_claude_home_config_projects_system_settings_into_managed_home(tmp_path: Path) -> None:
     source_home = tmp_path / 'system-home'
     target_home = tmp_path / 'managed-home'