diff --git a/CHANGELOG.md b/CHANGELOG.md
index d841fb9..1aa84e9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,109 @@ All notable changes to Code Context Control (C3) are documented here.
 The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.128.0] - 2026-09-08
+
+### Fixed — starting an IDE session now brings up that project's UI
+
+Opening a session in a project left you with no UI "in some cases", and the
+cases were not random: **nothing in C3 connected a session starting to a UI
+server running.** `ProjectManager.launch_session` had five callers — two hub
+endpoints, the per-project server, the TUI, and a thread in `run_hub` that
+fired ONCE, two seconds after hub startup, over projects with
+`autostart_ui=true` (true on 0 of the 60 projects registered on the machine
+this was found on). So when a UI *was* there it was a detached `pythonw`
+server a previous session had left behind — survival, not startup. Reboot,
+`taskkill`, or the hub's Stop button, and the next session had nothing.
+
+- **`cli/hook_session_open.ensure_ui(project, host_session_id)`** — the
+  SessionStart hook now launches the project's UI when one is not already
+  running. Guards, in order: `C3_NO_UI_AUTOSTART` / `C3_BENCHMARK_MODE`,
+  `.c3/config.json → {"ui": {"autostart_on_session": false}}`, the project
+  being registered with the hub at all (a scratch clone or a test fixture
+  gets nothing), and a live registry port for this project — which is what
+  keeps `resume` and `clear`, both real SessionStart events, from stacking a
+  second server. `source: compact` returns before any of it. Failures go to
+  `.c3/hook_errors.log`; the hook still prints nothing, because SessionStart
+  stdout becomes model context.
+- **`cli/hub_server.run_hub`** — the one-shot autostart thread is now a 60 s
+  sweep (`_UI_SWEEP_INTERVAL_S`) that reaps orphans and chases autostart
+  projects with a live session, so a hub started *after* the IDE, or a UI that
+  died mid-session, is no longer a dead end. The first pass keeps the old
+  contract ("autostart launches the UI when the hub starts").
+- **Ownership** — `cli/server._register_session` records `pid`, `owner`
+  (`session` when a session launched it, else `user`, from `C3_UI_OWNER`) and
+  `owner_session`. `ProjectManager.sweep_registry` stops a **session**-owned
+  server once its project has no heartbeat left and its activity log has been
+  untouched for `ui_reap_minutes` (new `~/.c3/hub_config.json` key, default
+  30, `0` disables) — and returns `{"dropped", "reaped"}`. A server a human
+  started with `c3 ui` or Open UI is never touched. This is what stops the
+  orphan collection: the registry on the machine above held a UI server from
+  a session that had ended 24 hours earlier.
+
+### Fixed — session liveness is a fact now, not an inference
+
+`ProjectManager._get_live_session_info` read the newest `session_start` row
+and ended it only on a `session_save` — written solely when a human clicks end
+in the hub — or after 20 minutes of tool-call silence. Both halves were wrong
+in the ordinary case: a closed IDE read "live" for up to 20 minutes (the
+`session_end` row added in 2.126.0 was read by NOTHING), an open-but-quiet
+session read dead while its MCP process sat there serving, and two sessions on
+one project collapsed into one.
+
+- **`services/session_live.py`** (new) — one heartbeat file per session,
+  `.c3/live/<session_id>.json` = `{session_id, host_session_id, ide, pid, ts}`,
+  written through `services/atomic_json.write_json_atomic` (the Windows-safe
+  publish; a pid-only temp suffix races across threads). `beat()`, `clear()`,
+  `live_sessions(ttl=180)`, `is_live()`. Liveness is decided by the timestamp,
+  never by probing the pid: pids are reused across reboots and there is no
+  cheap portable "is this pid alive" (C3 has no psutil dependency).
+  `HEARTBEAT_INTERVAL_S = 60`, `HEARTBEAT_TTL_S = 180` — three missed beats.
+  A session id is sanitised before it becomes a filename.
+- **`cli/mcp_server.lifespan`** — beats once right after `session_start`, then
+  on a `c3-session-heartbeat` daemon thread; the `finally` block stops the
+  thread and removes the file *before* the slow teardown steps, so a reader
+  polling during shutdown already sees the session as gone.
+- **`ProjectManager`** — `_get_live_sessions(path)` returns EVERY live session,
+  newest first: heartbeats first (authoritative), then the old activity-log
+  inference for a session with no heartbeat (a pre-2.128 C3 in another
+  checkout, or a process killed without cleanup). The inference now treats a
+  `session_end` row as terminal as well, matching on `session_id` **or**
+  `host_session_id` (the hook writes `session_id: ""` when no link file
+  exists yet). `_get_live_session_info` is kept as "the newest one".
+  `list_projects` gains `sessions` and `session_count`; every singular field
+  (`session_active`, `live_session_id`, `started_at`, …) still describes the
+  newest session, so existing consumers are unaffected.
+- **Hub card** — `MCP×N` when a project has more than one live session (two
+  IDEs, or a repo and its worktrees).
+
+### Fixed — hooks added to the installer now reach installed projects
+
+The 2.126.0 `SessionStart` / `SessionEnd` hooks were registered by
+`c3 install-mcp` only, so no project installed before that release ever wrote
+a `session_open` or `session_end` row — 60 projects here, C3's own repo among
+them, which is why the end of a session had no signal to read.
+`cli/hub_server._migrate_project_hooks` already rewrote every registered
+project's settings at hub startup, but for exactly one hardcoded entry
+(`hook_c3read` on `PostToolUse`).
+
+- **`c3_hook_migrations()` + `migrate_hooks_for_project(path)`** — the
+  migration is now a table of `(settings file, event, "is C3's entry already
+  here", entry)` rows, with the lifecycle hooks added for Claude (Gemini has
+  no session-lifecycle event). Additive and idempotent: a user's own
+  `SessionStart` hook survives, a second run writes nothing, an unparseable
+  settings file is left alone, and a settings file that does not exist is
+  never created — the hub is not the place to decide a project should be
+  installed for an IDE. Future hooks cost one row.
+- **The migration's commands were unrunnable on Windows** — it quoted paths
+  with `shlex.quote`, which on Windows produces a SINGLE-quoted backslash
+  path. cmd.exe does not treat single quotes as quoting, so every entry it
+  wrote for a path containing a space (every path under `1. Projects`, say)
+  was dead on arrival — the `hook_c3read` entry it has been writing since
+  2.21 included. The installer learned this on 2026-07-26 and has used a
+  double-quoted forward-slash path ever since; that logic now lives in
+  `cli/_hook_utils.hook_command_arg` and both writers call it. A test pins
+  the migration's command string against the installer's, byte for byte.
+
 ## [2.127.0] - 2026-09-06
 
 ### Added — override cost data: "this rule is costing you" (C3 Desk, D3a)
diff --git a/README.md b/README.md
index d9d5370..c7455e3 100644
--- a/README.md
+++ b/README.md
@@ -102,6 +102,8 @@ c3 ui                             # binds the first free port from 3333
 
 Twelve tabs. The dashboard above shows token savings, indexed files, the live session, and a stream of recent tool calls and file changes.
 
+Since v2.128.0 you rarely start it by hand: opening an IDE session in a registered project brings its UI up, and a session proves it is alive with a heartbeat rather than with tool-call traffic — so a quiet session stays live and a closed one disappears at once. Servers a session started are reaped once the session is gone; the ones you started are not. → [Session liveness](https://github.com/drknowhow/code-context-control/blob/main/docs/session-liveness.md)
+
 | Tab | What it's for |
 |---|---|
 | **Dashboard** | Token savings, codebase breakdown, live session counters, recent activity |
diff --git a/cli/_hook_utils.py b/cli/_hook_utils.py
index 979679d..9b575a7 100644
--- a/cli/_hook_utils.py
+++ b/cli/_hook_utils.py
@@ -8,6 +8,7 @@
 import json
 import os
 import re
+import shlex
 import sys
 import traceback
 from datetime import datetime, timezone
@@ -34,6 +35,27 @@
 STATE_WARNINGS: list = []
 
 
+def hook_command_arg(raw) -> str:
+    """Quote one argv element for a hook ``command`` string in settings.
+
+    Windows: a DOUBLE-quoted forward-slash path. Claude Code runs hooks through
+    Git Bash, where ``shlex.quote`` produces SINGLE quotes around a backslash
+    path — cmd.exe does not treat single quotes as quoting, so any path with a
+    space or parentheses (``Claude Code Companion (C3)``) breaks. A
+    double-quoted forward-slash path is parsed correctly by bash AND cmd.
+    Verified 2026-07-26 under Git Bash on Windows 11; the installer in
+    cli/c3.py carries the full investigation, including why no ``cmd.exe /c``
+    wrapper belongs here.
+
+    Anything that writes a hook command — the installer and the hub's
+    migration — must go through this, or the entry it writes is dead on
+    Windows.
+    """
+    if sys.platform == "win32":
+        return '"' + str(raw).replace("\\", "/") + '"'
+    return shlex.quote(str(raw))
+
+
 def ensure_utf8_stdio() -> None:
     """Force UTF-8 on the hook process's stdio pipes.
 
diff --git a/cli/c3.py b/cli/c3.py
index f23f875..d74ef75 100644
--- a/cli/c3.py
+++ b/cli/c3.py
@@ -33,7 +33,6 @@
 import logging
 import os
 import re
-import shlex
 import subprocess
 import sys
 import tempfile
@@ -92,7 +91,7 @@
 # Config
 CONFIG_DIR = ".c3"
 CONFIG_FILE = ".c3/config.json"
-__version__ = "2.127.0"
+__version__ = "2.128.0"
 
 
 def _compress_file_cli(compressor, path, mode="smart", **kw):
@@ -5542,10 +5541,10 @@ def cmd_install_mcp(args):
         # the wrapper was never needed. A double-quoted forward-slash path is parsed
         # correctly by bash AND cmd, including paths containing spaces or parentheses
         # (e.g. "Claude Code Companion (C3)"), which is why the wrapper was added.
-        def _hook_arg(raw: str) -> str:
-            if sys.platform == "win32":
-                return '"' + str(raw).replace("\\", "/") + '"'
-            return shlex.quote(str(raw))
+        # Lives in cli/_hook_utils so the hub's hook migration quotes commands
+        # the same way — it used shlex.quote and wrote entries that Windows
+        # could not run.
+        from cli._hook_utils import hook_command_arg as _hook_arg
 
         # v2.42: single dispatcher script per hook event instead of N separate
         # per-hook commands. One interpreter spawn per event; the dispatcher
diff --git a/cli/hook_session_open.py b/cli/hook_session_open.py
index 24af980..5683578 100644
--- a/cli/hook_session_open.py
+++ b/cli/hook_session_open.py
@@ -22,6 +22,14 @@
 - a ``session`` notification (``kind="session"``, ``ref_id=host id``,
   ``info``, title ``Session started <short id>``).
 
+It also brings up the project's UI server when one is not already running
+(``ensure_ui`` below, v2.128.0) — the only place in C3 where "a session
+started" causes "the UI is up". It needs the project to be registered with
+the hub, and it skips a project that already has a live UI port, so a
+``resume`` or ``clear`` never stacks a second server. Opt out with
+``C3_NO_UI_AUTOSTART=1`` or
+``.c3/config.json → {"ui": {"autostart_on_session": false}}``.
+
 ``source: "compact"`` writes NOTHING: a compaction is the same session and
 the same MCP process, and a "Session started" toast on every compaction
 would teach the user to ignore the real ones.
@@ -37,6 +45,8 @@
 """
 from __future__ import annotations
 
+import json
+import os
 import sys
 from pathlib import Path
 
@@ -48,6 +58,98 @@
 _SILENT_SOURCES = {"compact"}
 
 
+def _ui_autostart_enabled(project: Path) -> bool:
+    """Opt-outs, in the order a user reaches for them."""
+    if os.environ.get("C3_NO_UI_AUTOSTART") or os.environ.get("C3_BENCHMARK_MODE"):
+        return False
+    try:
+        cfg = json.loads((project / ".c3" / "config.json").read_text(encoding="utf-8"))
+    except Exception:
+        return True  # no readable config is not a reason to withhold the UI
+    ui = cfg.get("ui")
+    if isinstance(ui, dict) and "autostart_on_session" in ui:
+        return bool(ui.get("autostart_on_session"))
+    return True
+
+
+def _is_registered(pm, project: Path) -> bool:
+    """True when the hub knows this project.
+
+    The UI server is a hub-level surface: a directory the user never
+    registered gets nothing automatic (``c3 ui`` still works). This is also
+    what keeps a test fixture or a scratch clone from growing a web server.
+    """
+    target = os.path.normcase(str(project))
+    for entry in pm._read_projects():
+        raw = str(entry.get("path") or "")
+        if raw and os.path.normcase(raw) == target:
+            return True
+        try:
+            if raw and os.path.normcase(str(Path(raw).resolve())) == target:
+                return True
+        except Exception:
+            continue
+    return False
+
+
+def _ui_running(pm, project: Path) -> bool:
+    """True when a live UI server is already registered for this project.
+
+    This is what keeps `resume` / `clear` — both real SessionStart events —
+    from stacking a second server on every reconnect.
+    """
+    target = os.path.normcase(str(project))
+    for entry in pm._read_registry():
+        raw = str(entry.get("project_path") or "")
+        if not raw:
+            continue
+        try:
+            same = os.path.normcase(str(Path(raw).resolve())) == target
+        except Exception:
+            same = os.path.normcase(raw) == target
+        if same and entry.get("port") and pm._port_alive(entry["port"]):
+            return True
+    return False
+
+
+def ensure_ui(project: Path, host_session_id: str = "") -> str:
+    """Bring up this project's UI server for the session that just opened.
+
+    Nothing else in C3 connected "a session started" to "the UI is running":
+    the hub's sweep only chases projects flagged `autostart_ui`, and before
+    2.128.0 it ran once at hub startup — so a UI was only ever there because
+    an earlier session had left one behind.
+
+    Returns a short reason string (tests and diagnostics read it). Never
+    raises and never prints — SessionStart stdout becomes model context.
+    """
+    try:
+        project = Path(project).resolve()
+    except Exception:
+        project = Path(project)
+    if not _ui_autostart_enabled(project):
+        return "disabled"
+    try:
+        from services.project_manager import ProjectManager
+        pm = ProjectManager()
+        if not _is_registered(pm, project):
+            return "unregistered"
+        if _ui_running(pm, project):
+            return "already-running"
+        # launch_session copies os.environ into the detached child, which is
+        # where cli/server._register_session reads the ownership from.
+        os.environ["C3_UI_OWNER"] = "session"
+        if host_session_id:
+            os.environ["C3_UI_OWNER_SESSION"] = host_session_id
+        result = pm.launch_session(str(project))
+        if result.get("launched"):
+            return "launched"
+        return f"failed: {result.get('error', '')}".strip()
+    except Exception as exc:
+        log_hook_error("hook_session_open", exc)
+        return "error"
+
+
 def run(payload: dict, project_path: Path | None = None):
     """Core logic — importable by the dispatcher and tests. Returns None."""
     start_source = str(payload.get("source") or "").strip().lower()
@@ -76,11 +178,11 @@ def run(payload: dict, project_path: Path | None = None):
                kind="session", ref_id=host_sid)
     except Exception:
         pass
+    ensure_ui(project, host_sid)
     return None
 
 
 def main() -> None:
-    import json
     try:
         data = json.load(sys.stdin)
     except Exception as exc:
diff --git a/cli/hub_server.py b/cli/hub_server.py
index 45f536e..dbf618e 100644
--- a/cli/hub_server.py
+++ b/cli/hub_server.py
@@ -29,6 +29,7 @@
 
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
+from cli._hook_utils import hook_command_arg
 from core.ide import PROFILES, detect_ide, get_profile, load_ide_config, normalize_ide_name
 from services import access_guard
 from services.activity_log import ActivityLog
@@ -71,8 +72,14 @@ def _hub_allowed_hosts():
     "projects_view": "list",
     "main_view": "projects",
     "oracle_url": "",
+    # Minutes of project silence, with no session heartbeat left, before a UI
+    # server that an IDE session launched is stopped. 0 = never reap.
+    "ui_reap_minutes": 30,
 }
 
+# How often the hub reaps orphaned UI servers and chases autostart projects.
+_UI_SWEEP_INTERVAL_S = 60
+
 
 def _read_hub_config() -> dict:
     cfg = dict(_HUB_CONFIG_DEFAULTS)
@@ -4373,64 +4380,137 @@ def method_not_allowed(e):
 
 # ─── Hook migration ──────────────────────────────────────────────────────────
 
-def _migrate_project_hooks():
-    """Idempotently add new C3 hooks to all registered projects' Claude and Gemini settings.
+def _has_matcher(value: str):
+    return lambda entries: any(e.get("matcher") == value for e in entries)
+
+
+def _has_command(fragment: str):
+    return lambda entries: any(
+        fragment in (hk.get("command") or "")
+        for e in entries
+        for hk in e.get("hooks", [])
+    )
+
 
-    Runs at hub startup so that existing projects pick up hook changes
-    without requiring a manual 'c3 mcp install' on each project.
+def c3_hook_migrations(cli_dir: Path | None = None) -> list:
+    """C3's own hook entries, per settings file and hook event.
+
+    Adding a hook to the installer (``cli/c3.py``) does NOT reach a project
+    that is already installed. The v2.126.0 SessionStart / SessionEnd entries
+    landed in the installer only, so every existing project kept writing no
+    ``session_open`` / ``session_end`` rows at all — 60 projects on the
+    machine this was found on, including C3's own. This table is how they
+    catch up: the hub applies it at startup, additively and idempotently.
+
+    ``present`` decides "C3's entry for this event is already here", so a
+    re-run is a no-op and a user's own entries for the same event survive.
     """
-    cli_dir = Path(__file__).parent
-    hook_c3read_cmd = (
-        f"{shlex.quote(sys.executable)} "
-        f"{shlex.quote(str(cli_dir / 'hook_c3read.py'))}"
+    cli_dir = cli_dir or Path(__file__).parent
+    # hook_command_arg, not shlex.quote: on Windows the latter writes a
+    # single-quoted backslash path that cmd.exe does not unquote, so every
+    # command this table produced for a path with a space — every path under
+    # "1. Projects", say — was dead on arrival. The installer has used the
+    # double-quoted forward-slash form since 2026-07-26; this now matches it.
+    dispatch = (
+        f"{hook_command_arg(sys.executable)} "
+        f"{hook_command_arg(str(cli_dir / 'hook_dispatch.py'))}"
     )
-    new_hook = {
+    c3read_entry = {
         "matcher": "mcp__c3__c3_read",
-        "hooks": [{"type": "command", "command": hook_c3read_cmd}],
+        "hooks": [{
+            "type": "command",
+            "command": (f"{hook_command_arg(sys.executable)} "
+                        f"{hook_command_arg(str(cli_dir / 'hook_c3read.py'))}"),
+        }],
     }
+    c3read_present = _has_matcher("mcp__c3__c3_read")
+    dispatch_present = _has_command("hook_dispatch.py")
+
+    def _lifecycle(event: str, arg: str) -> dict:
+        return {
+            "settings": ".claude/settings.local.json",
+            "event": event,
+            "present": dispatch_present,
+            "entry": {"matcher": "",
+                      "hooks": [{"type": "command",
+                                 "command": f"{dispatch} {arg}"}]},
+        }
 
-    # (settings_path, hook_event) pairs to check per project
-    _HOOK_TARGETS = [
-        (".claude/settings.local.json", "PostToolUse"),
-        (".gemini/settings.json", "AfterTool"),
+    return [
+        {"settings": ".claude/settings.local.json", "event": "PostToolUse",
+         "present": c3read_present, "entry": c3read_entry},
+        {"settings": ".gemini/settings.json", "event": "AfterTool",
+         "present": c3read_present, "entry": c3read_entry},
+        # Claude only — Gemini has no session-lifecycle event to hang these on.
+        _lifecycle("SessionStart", "start"),
+        _lifecycle("SessionEnd", "end"),
     ]
 
+
+def migrate_hooks_for_project(path: str, migrations: list | None = None) -> int:
+    """Apply the hook table to one project. Returns settings files changed.
+
+    Never creates a settings file that is not already there: a project with no
+    ``.claude/settings.local.json`` has not been installed for Claude, and the
+    hub is not the place to decide that it should be.
+    """
+    migrations = migrations if migrations is not None else c3_hook_migrations()
+    by_file: dict = {}
+    for migration in migrations:
+        by_file.setdefault(migration["settings"], []).append(migration)
+
+    changed_files = 0
+    for rel_settings, items in by_file.items():
+        settings_path = Path(path) / rel_settings
+        if not settings_path.exists():
+            continue
+        try:
+            with open(settings_path, encoding="utf-8") as f:
+                settings = json.load(f)
+        except Exception:
+            continue
+        changed = False
+        for migration in items:
+            event = migration["event"]
+            existing = list(settings.get("hooks", {}).get(event, []))
+            if migration["present"](existing):
+                continue
+            existing.append(migration["entry"])
+            settings.setdefault("hooks", {})[event] = existing
+            changed = True
+        if not changed:
+            continue
+        try:
+            with open(settings_path, "w", encoding="utf-8") as f:
+                json.dump(settings, f, indent=2)
+            changed_files += 1
+        except Exception:
+            pass
+    return changed_files
+
+
+def _migrate_project_hooks():
+    """Idempotently add missing C3 hooks to every registered project.
+
+    Runs at hub startup so existing projects pick up hook changes without a
+    manual 'c3 install-mcp' per project.
+    """
     try:
         projects = _pm().list_projects()
     except Exception:
         return
 
+    migrations = c3_hook_migrations()
     updated = 0
     for p in projects:
         path = p.get("path", "")
         if not path:
             continue
-        for rel_settings, hook_event in _HOOK_TARGETS:
-            settings_path = Path(path) / rel_settings
-            if not settings_path.exists():
-                continue
-            try:
-                with open(settings_path, encoding="utf-8") as f:
-                    settings = json.load(f)
-            except Exception:
-                continue
-
-            existing = settings.get("hooks", {}).get(hook_event, [])
-            if any(h.get("matcher") == "mcp__c3__c3_read" for h in existing):
-                continue  # already present — skip
-
-            existing.append(new_hook)
-            settings.setdefault("hooks", {})[hook_event] = existing
-            try:
-                with open(settings_path, "w", encoding="utf-8") as f:
-                    json.dump(settings, f, indent=2)
-                updated += 1
-            except Exception:
-                pass
+        updated += migrate_hooks_for_project(path, migrations)
 
     if updated:
         logging.getLogger(__name__).info(
-            "[c3] Migrated hook_c3read to %d project settings file(s)", updated
+            "[c3] Migrated C3 hooks into %d project settings file(s)", updated
         )
 
 
@@ -4478,21 +4558,39 @@ def run_hub(
         pass
     _migrate_project_hooks()
 
-    # Auto-launch UI servers for projects with autostart_ui=True
-    def _autostart_ui_servers():
+    # Periodic UI sweep. Before 2.128 this ran ONCE, two seconds after
+    # startup, over autostart_ui projects — so a project whose session started
+    # after the hub (the normal case), or whose UI died mid-session, never got
+    # one. Now it also reaps the session-owned servers whose session is gone.
+    def _ui_sweep_loop():
         import time
         time.sleep(2)  # Give Flask a moment to bind
-        try:
-            pm = _pm()
-            for proj_path in pm.get_autostart_projects():
-                try:
-                    pm.launch_session(proj_path)
-                except Exception:
-                    pass
-        except Exception:
-            pass
+        first_pass = True
+        while True:
+            try:
+                pm = _pm()
+                pm.sweep_registry()
+                wanted = set(pm.get_autostart_projects())
+                if wanted:
+                    for proj in pm.list_projects():
+                        if proj.get("path") not in wanted or proj.get("ui_active"):
+                            continue
+                        # First pass keeps the old contract ("autostart launches
+                        # the UI when the hub starts"); later passes only chase
+                        # projects that actually have a live session.
+                        if not (first_pass or proj.get("session_active")):
+                            continue
+                        try:
+                            pm.launch_session(proj["path"])
+                        except Exception:
+                            pass
+            except Exception:
+                pass
+            first_pass = False
+            time.sleep(_UI_SWEEP_INTERVAL_S)
 
-    threading.Thread(target=_autostart_ui_servers, daemon=True).start()
+    threading.Thread(target=_ui_sweep_loop, daemon=True,
+                     name="c3-hub-ui-sweep").start()
 
     bind_host = str(cfg.get("host", "127.0.0.1") or "127.0.0.1").strip()
     if bind_host not in ("127.0.0.1", "localhost", "::1") and not quiet:
diff --git a/cli/hub_ui/components/project_card.js b/cli/hub_ui/components/project_card.js
index c7365b8..a0d2ba2 100644
--- a/cli/hub_ui/components/project_card.js
+++ b/cli/hub_ui/components/project_card.js
@@ -493,7 +493,12 @@ function ProjectCard({ p, isChild, rollup, expanded, onToggleExpand, onChanged,
     } catch (err) { notify('IDE: ' + err.message, 'err'); }
   };
 
-  const portLabel = p.port ? `:${p.port}` : (p.session_active ? 'MCP' : '—');
+  // session_count > 1 = two IDEs / a repo and its worktrees on one project.
+  const sessionCount = p.session_count || (p.session_active ? 1 : 0);
+  const mcpLabel = sessionCount > 1 ? `MCP×${sessionCount}` : 'MCP';
+  const portLabel = p.port
+    ? (sessionCount > 1 ? `:${p.port} · ${mcpLabel}` : `:${p.port}`)
+    : (p.session_active ? mcpLabel : '—');
   const metaText = `${ideLabel(p.ide)} · ${portLabel} · ${p.last_session ? timeAgo(p.last_session) : 'never'}`;
 
   const identity = (
diff --git a/cli/mcp_server.py b/cli/mcp_server.py
index 050922d..66a758d 100644
--- a/cli/mcp_server.py
+++ b/cli/mcp_server.py
@@ -321,6 +321,33 @@ def _bg_convo_sync():
             **({"host_session_id": _host_sid} if _host_sid else {}),
         })
 
+    # Liveness heartbeat: THIS process is the session, so say so in a file a
+    # reader can trust. Without it the hub has to infer liveness from tool
+    # activity, which calls a quiet-but-open session dead after 20 min and a
+    # closed one live for 20 min. Refreshed on a 60 s daemon thread (same
+    # shape as _bg_convo_sync) and removed in the finally block below.
+    _heartbeat_stop = threading.Event()
+    _live_session_id = str((services.session_mgr.current_session or {}).get("id") or "")
+    if _live_session_id:
+        from services import session_live
+
+        def _do_beat():
+            session_live.beat(services.project_path, _live_session_id,
+                              host_session_id=_host_session_id(services) or "",
+                              ide=_IDE_NAME)
+
+        _do_beat()
+
+        def _bg_heartbeat():
+            while not _heartbeat_stop.wait(timeout=session_live.HEARTBEAT_INTERVAL_S):
+                try:
+                    _do_beat()
+                except Exception:
+                    pass
+
+        threading.Thread(target=_bg_heartbeat, daemon=True,
+                         name="c3-session-heartbeat").start()
+
     # Auto-restore latest snapshot if recent (< 30 min).
     # Deferred to background thread so first tool call isn't blocked.
     # Skipped in benchmark mode to prevent snapshot budget from carrying over between tasks.
@@ -391,6 +418,15 @@ def _bg_validation_sweep():
         yield services
     finally:
         _convo_sync_stop.set()
+        # Stop claiming to be alive before anything slow runs: a reader that
+        # polls during teardown should already see this session as gone.
+        _heartbeat_stop.set()
+        if _live_session_id:
+            try:
+                from services import session_live as _sl
+                _sl.clear(services.project_path, _live_session_id)
+            except Exception:
+                pass
         # Auto-memory: extract remaining learnings and generate session summary.
         if hasattr(services, "auto_memory"):
             try:
diff --git a/cli/server.py b/cli/server.py
index ea9d39c..bfb4646 100644
--- a/cli/server.py
+++ b/cli/server.py
@@ -108,11 +108,18 @@ def _register_session(port: int, project_path: str, project_name: str):
         entries = _registry_read()
         # Remove our own port and any stale (dead) entries
         entries = [e for e in entries if e.get("port") != port and _port_alive(e.get("port", 0))]
+        # owner says who asked for this server: "session" when an IDE session
+        # start launched it (cli/hook_session_open), "user" for `c3 ui`, the
+        # hub's Open UI button and the TUI. Only session-owned servers are
+        # ever reaped automatically (ProjectManager.sweep_registry).
         entries.append({
             "port": port,
             "project_path": project_path,
             "project_name": project_name,
             "started_at": time.time(),
+            "pid": os.getpid(),
+            "owner": (os.environ.get("C3_UI_OWNER") or "user").strip().lower(),
+            "owner_session": (os.environ.get("C3_UI_OWNER_SESSION") or "").strip(),
         })
         _registry_write(entries)
 
diff --git a/docs/session-liveness.md b/docs/session-liveness.md
new file mode 100644
index 0000000..380ed0a
--- /dev/null
+++ b/docs/session-liveness.md
@@ -0,0 +1,107 @@
+# Session liveness and the project UI
+
+*Since 2.128.0.*
+
+Two different things are called a "session" in C3, and before 2.128.0 they
+never met:
+
+| | what it is | who starts it |
+| --- | --- | --- |
+| **agent session** | the MCP server serving your IDE | the IDE, when it loads the `c3` MCP server |
+| **UI session** | the per-project web UI (`c3 ui`, its own port) | you, or the hub |
+
+Starting an IDE session did not start a UI, and the hub's "live" light was
+inferred from activity rows rather than from the process. This document is the
+contract that replaced both.
+
+## A session says it is alive
+
+The MCP server writes one heartbeat file per session:
+
+```
+.c3/live/<c3_session_id>.json   {session_id, host_session_id, ide, pid, ts}
+```
+
+It is written once right after the `session_start` row and refreshed every
+60 s (`services/session_live.HEARTBEAT_INTERVAL_S`). A reader treats a file
+whose `ts` is within 180 s (`HEARTBEAT_TTL_S` — three missed beats) as a live
+session. Clean shutdown removes the file before the slow teardown steps run,
+so a reader polling during shutdown already sees the session as gone; a crash
+or a `taskkill` leaves a stale file, which the next reader ignores and prunes.
+
+`pid` is recorded for diagnostics only. Liveness is decided by the timestamp:
+pids are reused across reboots, and there is no cheap portable "is this pid
+alive" call (C3 does not depend on psutil).
+
+One file per session id, not per project — so two IDEs, or a repo and its
+worktrees, each report themselves. `ProjectManager.list_projects` exposes
+`sessions` and `session_count`; the hub card shows `MCP×N`.
+
+### Why not just read the activity log
+
+That was the old behaviour, and it is still the fallback for a session with no
+heartbeat (an older C3 in another checkout, or a process killed without
+cleanup). Its two failure modes are the reason the heartbeat exists:
+
+- **a closed IDE read "live" for up to 20 minutes** — a session ended only on
+  a `session_save` row, which is written when a human clicks "end session" in
+  the hub. The `session_end` row the SessionEnd hook has written since 2.126.0
+  was read by nothing. It is honoured now (matched on `session_id` *or*
+  `host_session_id`, because the hook writes an empty C3 id when no link file
+  exists yet);
+- **a quiet session read dead** — no tool call for 20 minutes meant "gone",
+  even with the MCP process right there. A fresh heartbeat now outranks any
+  amount of silence, and it also outranks a `session_save`: the process is
+  still serving, and a bookkeeping row does not kill it.
+
+## The UI follows the session
+
+The SessionStart hook (`cli/hook_session_open.ensure_ui`) launches the
+project's UI server when one is not already running. In order, it declines to
+do anything when:
+
+1. `C3_NO_UI_AUTOSTART=1` or `C3_BENCHMARK_MODE` is set;
+2. `.c3/config.json` says `{"ui": {"autostart_on_session": false}}`;
+3. the project is not registered with the hub (a scratch clone gets nothing —
+   `c3 ui` still works);
+4. the registry already has a live port for this project. This is what keeps
+   `resume` and `clear` — both real SessionStart events — from stacking a
+   second server.
+
+`source: compact` is not a new session and never reaches any of this.
+
+## Ownership and reaping
+
+Registry entries (`~/.c3/registry.json`) carry `pid`, `owner` and
+`owner_session`. `owner` is `session` when a session start launched the server
+and `user` for `c3 ui`, the hub's **Open UI** button and the TUI.
+
+The hub sweeps every 60 s (`ProjectManager.sweep_registry`): dead ports are
+dropped, and a **session**-owned server is stopped once its project has no
+heartbeat left *and* its activity log has not been touched for
+`ui_reap_minutes`.
+
+```jsonc
+// ~/.c3/hub_config.json
+{ "ui_reap_minutes": 30 }   // 0 = never reap
+```
+
+A `user`-owned server is never stopped automatically. Without this, every
+session that ever ran left a web server behind until the next reboot.
+
+The same sweep launches the UI for projects with `autostart_ui` that have a
+live session but no UI — so a hub started *after* the IDE, or a UI that died
+mid-session, recovers on its own. Before 2.128.0 that autostart ran once, two
+seconds after the hub started, and never again.
+
+## Checking it by hand
+
+```bash
+ls .c3/live                      # one file per live session in this project
+python -c "from services import session_live as s; print(s.live_sessions('.'))"
+python -c "import json,pathlib; print(json.loads((pathlib.Path.home()/'.c3'/'registry.json').read_text()))"
+```
+
+The hooks and the MCP server load their code at process start, so a change
+here is not live in the session that made it: a new IDE session is required,
+and the hub must be restarted.
diff --git a/pyproject.toml b/pyproject.toml
index 818ba87..d5b6941 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "code-context-control"
-version = "2.127.0"
+version = "2.128.0"
 description = "Local MCP code-intelligence for AI coding tools: surgical search/read/edit, agent-config version history, path-level access + masking guards, and a multi-project hub."
 readme = "README.md"
 requires-python = ">=3.10"
diff --git a/services/project_manager.py b/services/project_manager.py
index 5de4976..b28ba38 100644
--- a/services/project_manager.py
+++ b/services/project_manager.py
@@ -2,6 +2,7 @@
 
 import json
 import os
+import signal
 import socket
 import subprocess
 import sys
@@ -18,6 +19,10 @@
 _REGISTRY_FILE = _GLOBAL_C3_DIR / "registry.json"
 _SESSION_ACTIVITY_GRACE_SECONDS = 20 * 60
 _REGISTRY_STARTUP_GRACE = 30  # seconds: keep registry entry while server is still starting
+# Minutes of project-wide silence, with no session heartbeat left, before a
+# UI server that a session launched is stopped. Overridable per machine via
+# ~/.c3/hub_config.json -> ui_reap_minutes (0 = never reap).
+_DEFAULT_UI_REAP_MINUTES = 30
 
 
 def _pythonw() -> str:
@@ -139,18 +144,140 @@ def _get_budget_info(self, path: str) -> dict | None:
                 pass
         return None
 
-    def _get_live_session_info(self, path: str) -> dict | None:
-        """Return live session info inferred from the project's activity log."""
+    def _ended_session_ids(self, activity, limit: int = 8) -> set:
+        """Ids (C3 and host) of sessions with a terminal row in the log.
+
+        ``session_save`` is the hub's "end session" button. ``session_end`` is
+        what the SessionEnd hook writes when the IDE actually closes (v2.126.0)
+        and was read by nothing until now — which is why a closed IDE kept
+        reading "live" until the 20-minute idle grace expired.
+        """
+        ended = set()
+        for event_type in ("session_save", "session_end"):
+            try:
+                rows = activity.get_recent(limit=limit, event_type=event_type) or []
+            except Exception:
+                rows = []
+            for row in rows:
+                for key in ("session_id", "host_session_id"):
+                    value = str(row.get(key) or "").strip()
+                    if value:
+                        ended.add(value)
+        return ended
+
+    def _last_activity_since(self, activity, started: str) -> str:
+        try:
+            recent = activity.get_recent(limit=1, since=started)
+        except Exception:
+            recent = []
+        return (recent[0].get("timestamp") if recent else started) or started
+
+    def _session_timing(self, started: str, last_activity: str) -> dict | None:
+        started_dt = self._parse_timestamp(started)
+        last_activity_dt = self._parse_timestamp(last_activity)
+        if not started_dt or not last_activity_dt:
+            return None
+        now = datetime.now(timezone.utc)
+        return {
+            "duration_seconds": max(0, int((now - started_dt).total_seconds())),
+            "idle_seconds": max(0, int((now - last_activity_dt).total_seconds())),
+        }
+
+    def _get_live_sessions(self, path: str) -> list:
+        """Every session live for this project right now, newest start first.
+
+        Two sources, in order of trust:
+
+        1. heartbeat files under ``.c3/live`` — the MCP process saying it is
+           alive. Authoritative, and unlike activity rows it does not call a
+           quiet-but-open session dead.
+        2. the activity log, for a session with no heartbeat: a pre-2.128 C3
+           serving another checkout, or a process that died without cleanup.
+        """
         try:
             activity = ActivityLog(path)
+        except Exception:
+            return []
+        try:
+            from services import session_live
+            beats = session_live.live_sessions(path)
+        except Exception:
+            beats = []
+
+        sessions = []
+        seen = set()
+        # Typed get_recent scans 100x its limit, so these few rows reach well
+        # back past the tool_call traffic between them.
+        starts = []
+        if beats:
+            try:
+                starts = activity.get_recent(limit=8, event_type="session_start") or []
+            except Exception:
+                starts = []
+        for beat in beats:
+            session_id = str(beat.get("session_id") or "")
+            host_sid = str(beat.get("host_session_id") or "")
+            if not session_id or session_id in seen:
+                continue
+            row = next(
+                (
+                    r for r in starts
+                    if str(r.get("session_id") or "") == session_id
+                    or (host_sid and str(r.get("host_session_id") or "") == host_sid)
+                ),
+                {},
+            )
+            started = row.get("timestamp") or self._epoch_to_iso(beat.get("ts"))
+            if not started:
+                continue
+            last_activity = self._last_activity_since(activity, started)
+            timing = self._session_timing(started, last_activity)
+            if timing is None:
+                continue
+            seen.add(session_id)
+            if host_sid:
+                seen.add(host_sid)
+            sessions.append({
+                "session_id": session_id,
+                "host_session_id": host_sid,
+                "started_at": started,
+                "last_activity": last_activity,
+                "description": row.get("description", ""),
+                "source": "heartbeat",
+                "ide": str(beat.get("ide") or ""),
+                "pid": beat.get("pid"),
+                **timing,
+            })
+
+        inferred = self._infer_live_session(path, activity)
+        if inferred and inferred["session_id"] not in seen:
+            sessions.append(inferred)
+        sessions.sort(key=lambda s: str(s.get("started_at") or ""), reverse=True)
+        return sessions
+
+    def _epoch_to_iso(self, value) -> str | None:
+        epoch = _coerce_epoch(value)
+        if epoch is None:
+            return None
+        return datetime.fromtimestamp(epoch, timezone.utc).isoformat()
+
+    def _infer_live_session(self, path: str, activity=None) -> dict | None:
+        """Liveness guessed from activity rows — the pre-heartbeat path.
+
+        Kept as a fallback, so a project served by an older C3 (or one whose
+        MCP process was killed) still reports something. Its weakness is the
+        reason the heartbeat exists: a session is "live" only while tool calls
+        keep arriving, and stays live for the whole grace period after the
+        last one.
+        """
+        try:
+            activity = activity if activity is not None else ActivityLog(path)
 
             # Use the public API — get_recent's scan_factor=100 for typed queries
             # handles sparse rare events (session_start can sit far behind many
             # tool_call entries). Going through the API also lets tests stub it.
             starts = activity.get_recent(limit=1, event_type="session_start")
-            saves = activity.get_recent(limit=1, event_type="session_save")
             last_start = starts[0] if starts else None
-            last_save = saves[0] if saves else None
 
             if last_start is None:
                 return None
@@ -159,32 +286,36 @@ def _get_live_session_info(self, path: str) -> dict | None:
             if not session_id or not started:
                 return None
 
-            # If the most recent save belongs to this session, it has ended.
-            if last_save and last_save.get("session_id") == session_id:
+            # Ended if either id carries a terminal row (hub save, or the
+            # SessionEnd hook's session_end).
+            ended = self._ended_session_ids(activity)
+            host_sid = str(last_start.get("host_session_id") or "").strip()
+            if session_id in ended or (host_sid and host_sid in ended):
                 return None
 
-            recent = activity.get_recent(limit=1, since=started)
-            last_activity = recent[0].get("timestamp") if recent else started
-            started_dt = self._parse_timestamp(started)
-            last_activity_dt = self._parse_timestamp(last_activity)
-            if not started_dt or not last_activity_dt:
+            last_activity = self._last_activity_since(activity, started)
+            timing = self._session_timing(started, last_activity)
+            if timing is None:
                 return None
-            now = datetime.now(timezone.utc)
-            idle_seconds = max(0, int((now - last_activity_dt).total_seconds()))
-            if idle_seconds > _SESSION_ACTIVITY_GRACE_SECONDS:
+            if timing["idle_seconds"] > _SESSION_ACTIVITY_GRACE_SECONDS:
                 return None
-            duration_seconds = max(0, int((now - started_dt).total_seconds()))
             return {
                 "session_id": session_id,
+                "host_session_id": host_sid,
                 "started_at": started,
                 "last_activity": last_activity,
                 "description": last_start.get("description", ""),
-                "duration_seconds": duration_seconds,
-                "idle_seconds": idle_seconds,
+                "source": "activity_log",
+                **timing,
             }
         except Exception:
             return None
 
+    def _get_live_session_info(self, path: str) -> dict | None:
+        """The newest live session, or None. Kept for callers that want one."""
+        sessions = self._get_live_sessions(path)
+        return sessions[0] if sessions else None
+
     def _get_last_session_timestamp(
         self, path: str, stored_value: str | None = None, live_session: dict | None = None
     ) -> str | None:
@@ -287,18 +418,106 @@ def remove_project(self, path: str) -> bool:
             return True
         return False
 
-    def sweep_registry(self):
-        """Remove stale registry entries (dead ports). Call on hub startup."""
+    def _reap_minutes(self) -> int:
+        """Idle grace before a session-owned UI server is stopped.
+
+        ``~/.c3/hub_config.json -> ui_reap_minutes``; 0 disables reaping.
+        """
+        value = _DEFAULT_UI_REAP_MINUTES
+        try:
+            with open(_GLOBAL_C3_DIR / "hub_config.json", encoding="utf-8") as f:
+                value = json.load(f).get("ui_reap_minutes", _DEFAULT_UI_REAP_MINUTES)
+        except Exception:
+            pass
+        try:
+            return max(0, int(value))
+        except (TypeError, ValueError):
+            return _DEFAULT_UI_REAP_MINUTES
+
+    def _is_orphan_session_ui(self, entry: dict, now: float, reap_minutes: int) -> bool:
+        """True for a UI server whose launching session is gone for good.
+
+        Only ``owner == "session"`` entries qualify: a server a human started
+        with ``c3 ui`` or the hub's Open UI button is never reaped. The session
+        is judged gone when no heartbeat is left in the project AND the
+        activity log has not been touched for the whole grace period — the log
+        file's mtime is the cheapest honest answer to "when did work stop".
+        """
+        if str(entry.get("owner") or "user").strip().lower() != "session":
+            return False
+        project = str(entry.get("project_path") or "")
+        if not project or not Path(project).is_dir():
+            # An unreachable path (offline drive, unmounted share) makes both
+            # signals below read "nothing happening here". Fail closed.
+            return False
+        try:
+            from services import session_live
+            if session_live.live_sessions(project, prune=False):
+                return False  # a session is live here; it needs its UI
+        except Exception:
+            return False  # can't tell → leave it alone
+        cutoff = reap_minutes * 60
+        last_seen = None
+        try:
+            log_file = Path(project) / ".c3" / "activity_log.jsonl"
+            if log_file.exists():
+                last_seen = log_file.stat().st_mtime
+        except OSError:
+            last_seen = None
+        if last_seen is None:
+            last_seen = _coerce_epoch(entry.get("started_at"))
+        if last_seen is None:
+            return False
+        return (now - last_seen) > cutoff
+
+    def _stop_entry(self, entry: dict) -> bool:
+        """Stop one registered UI server — by pid when the entry has one."""
+        pid = entry.get("pid")
+        if pid:
+            try:
+                kwargs = {}
+                if sys.platform == "win32":
+                    kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
+                    subprocess.run(f"taskkill /PID {int(pid)} /F", shell=True,
+                                   capture_output=True, **kwargs)
+                else:
+                    os.kill(int(pid), signal.SIGTERM)
+                return True
+            except Exception:
+                pass
+        port = entry.get("port")
+        return self.stop_session(port) if port else False
+
+    def sweep_registry(self) -> dict:
+        """Drop dead registry entries and reap orphaned session-owned servers.
+
+        Called on hub startup and from the hub's periodic sweep. Returns
+        ``{"dropped": n, "reaped": [ports]}``.
+        """
         registry = self._read_registry()
         if not registry:
-            return
-        valid = [e for e in registry if e.get("port") and self._port_alive(e["port"])]
+            return {"dropped": 0, "reaped": []}
+        reap_minutes = self._reap_minutes()
+        now = time.time()
+        reaped: list = []
+        valid: list = []
+        for entry in registry:
+            port = entry.get("port")
+            if not (port and self._port_alive(port)):
+                continue
+            if reap_minutes and self._is_orphan_session_ui(entry, now, reap_minutes):
+                if self._stop_entry(entry):
+                    reaped.append(port)
+                    continue
+            valid.append(entry)
         if len(valid) != len(registry):
             try:
                 with open(_REGISTRY_FILE, "w", encoding="utf-8") as f:
                     json.dump(valid, f, indent=2)
             except Exception:
                 pass
+        return {"dropped": len(registry) - len(valid) - len(reaped),
+                "reaped": reaped}
 
     def list_projects(self) -> list:
         projects = self._read_projects()
@@ -340,9 +559,14 @@ def list_projects(self) -> list:
             path_accessible = Path(p["path"]).is_dir()
             enriched["accessible"] = path_accessible
             ui_active = ui_active_by_path.get(p["path"])
-            live_session = self._get_live_session_info(p["path"]) if path_accessible else None
+            live_sessions = self._get_live_sessions(p["path"]) if path_accessible else []
+            live_session = live_sessions[0] if live_sessions else None
             enriched["ui_active"] = ui_active is not None
             enriched["session_active"] = live_session is not None
+            # Newest session keeps the singular fields every existing consumer
+            # reads; the list is there for two IDEs / a repo and its worktrees.
+            enriched["sessions"] = live_sessions
+            enriched["session_count"] = len(live_sessions)
             # A project is active if either the web UI is live or the activity log shows a
             # currently running C3 session. The hub card should not fall back to "idle"
             # just because the session has no live UI port.
diff --git a/services/session_live.py b/services/session_live.py
new file mode 100644
index 0000000..53ace8b
--- /dev/null
+++ b/services/session_live.py
@@ -0,0 +1,139 @@
+"""Heartbeat files that prove an agent session is alive right now.
+
+Why this module exists
+---------------------
+``ProjectManager._get_live_session_info`` used to INFER liveness from the
+activity log: newest ``session_start`` row, ended by a ``session_save`` row
+(written only when a human clicks "end session" in the hub) or by 20 minutes
+of tool-call silence. Both halves are wrong in practice:
+
+- a closed IDE kept reading "live" for up to 20 minutes, because nothing
+  consumed the ``session_end`` row the SessionEnd hook writes;
+- an open session that was simply quiet — a long model turn, a human reading
+  — dropped off the hub after 20 minutes while its MCP process was right
+  there, serving.
+
+The MCP server process IS the session. So it says so: one small file per live
+session under ``.c3/live/``, refreshed every ``HEARTBEAT_INTERVAL_S`` and
+removed on shutdown. A reader treats a file whose ``ts`` is within
+``HEARTBEAT_TTL_S`` as a live session, and one stale file as a session whose
+process died without cleanup (crash, taskkill, reboot).
+
+A pid is recorded for diagnostics only. Liveness is decided by the timestamp,
+never by probing the pid: pids are reused across reboots, and there is no
+portable, cheap "is this pid alive" call (``os.kill(pid, 0)`` is not reliable
+on Windows and C3 does not depend on psutil).
+
+One file per session id — not one file per project — so two IDEs, or a repo
+and its worktrees, each report themselves instead of collapsing into one.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import time
+from pathlib import Path
+
+# Written by the MCP server's heartbeat thread; read by the hub and the CLI.
+HEARTBEAT_INTERVAL_S = 60
+# Three missed beats. Generous on purpose: a machine under load can stall a
+# daemon thread for longer than one interval, and a false "dead" is worse
+# than a stale entry lingering a minute or two.
+HEARTBEAT_TTL_S = 180
+
+_SAFE_NAME = re.compile(r"[^A-Za-z0-9_.-]")
+
+
+def live_dir(project) -> Path:
+    """``<project>/.c3/live`` — the directory holding one file per session."""
+    return Path(project) / ".c3" / "live"
+
+
+def _file_name(session_id: str) -> str:
+    """Filesystem-safe name for a session id (ids are C3-generated, but a
+    host id can arrive from anywhere, so never trust it as a path)."""
+    cleaned = _SAFE_NAME.sub("_", str(session_id or "").strip())[:120]
+    return f"{cleaned or 'unknown'}.json"
+
+
+def heartbeat_path(project, session_id: str) -> Path:
+    return live_dir(project) / _file_name(session_id)
+
+
+def beat(project, session_id: str, *, host_session_id: str = "",
+         ide: str = "", pid: int | None = None) -> bool:
+    """Record/refresh the heartbeat for one session. Never raises."""
+    if not str(session_id or "").strip():
+        return False
+    try:
+        from services.atomic_json import write_json_atomic
+        target = heartbeat_path(project, session_id)
+        target.parent.mkdir(parents=True, exist_ok=True)
+        write_json_atomic(target, {
+            "session_id": str(session_id),
+            "host_session_id": str(host_session_id or ""),
+            "ide": str(ide or ""),
+            "pid": int(pid if pid is not None else os.getpid()),
+            "ts": time.time(),
+        })
+        return True
+    except Exception:
+        return False
+
+
+def clear(project, session_id: str) -> bool:
+    """Remove one session's heartbeat (clean shutdown). Never raises."""
+    try:
+        heartbeat_path(project, session_id).unlink()
+        return True
+    except Exception:
+        return False
+
+
+def live_sessions(project, ttl: float = HEARTBEAT_TTL_S,
+                  prune: bool = True) -> list[dict]:
+    """Sessions whose heartbeat is fresher than ``ttl``, newest first.
+
+    Stale files are deleted when ``prune`` is set — the next reader would
+    reach the same verdict, and leaving them means ``.c3/live`` grows one
+    file per crashed session forever.
+    """
+    out: list[dict] = []
+    try:
+        entries = list(live_dir(project).iterdir())
+    except Exception:
+        return out
+    now = time.time()
+    for path in entries:
+        if path.suffix != ".json":
+            continue
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+            ts = float(data.get("ts") or 0)
+        except Exception:
+            # Unreadable or half-written: treat as absent, leave it alone.
+            continue
+        age = now - ts
+        if 0 <= age <= ttl or age < 0:
+            data["age_seconds"] = max(0.0, age)
+            out.append(data)
+        elif prune:
+            try:
+                path.unlink()
+            except OSError:
+                pass
+    out.sort(key=lambda d: float(d.get("ts") or 0), reverse=True)
+    return out
+
+
+def is_live(project, session_id: str, ttl: float = HEARTBEAT_TTL_S) -> bool:
+    """True when THIS session id has a fresh heartbeat."""
+    sid = str(session_id or "").strip()
+    if not sid:
+        return False
+    return any(
+        str(s.get("session_id") or "") == sid
+        or str(s.get("host_session_id") or "") == sid
+        for s in live_sessions(project, ttl=ttl, prune=False)
+    )
diff --git a/tests/test_hook_session_end.py b/tests/test_hook_session_end.py
index 7ce156f..7438203 100644
--- a/tests/test_hook_session_end.py
+++ b/tests/test_hook_session_end.py
@@ -59,6 +59,12 @@ def setUp(self):
         (self.project / ".c3" / "config.json").write_text(
             json.dumps({"hybrid": {"ollama_base_url": "http://127.0.0.1:9"}}),
             encoding="utf-8")
+        # SessionStart also brings up the project UI (v2.128.0). These cases
+        # are about the rows and notifications, and none of them wants a real
+        # web server — TestEnsureUI covers that path with the launch patched.
+        env = mock.patch.dict(os.environ, {"C3_NO_UI_AUTOSTART": "1"})
+        env.start()
+        self.addCleanup(env.stop)
 
     def tearDown(self):
         self.tmp.cleanup()
@@ -125,6 +131,125 @@ def test_notification_failure_does_not_lose_the_row(self):
         self.assertEqual(self.notifications(), [])
 
 
+class TestEnsureUI(unittest.TestCase):
+    """v2.128.0: the session that just opened gets its project UI.
+
+    Nothing used to connect the two — the hub's autostart fired once at hub
+    startup for flagged projects (flagged on 0 of 60 registered projects on
+    the machine this was found on), so a UI was only ever present because an
+    earlier session had left one running.
+    """
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.project = Path(self.tmp.name) / "proj"
+        (self.project / ".c3").mkdir(parents=True)
+        self.addCleanup(self.tmp.cleanup)
+        for var in ("C3_NO_UI_AUTOSTART", "C3_BENCHMARK_MODE",
+                    "C3_UI_OWNER", "C3_UI_OWNER_SESSION"):
+            os.environ.pop(var, None)
+        env = mock.patch.dict(os.environ, {})
+        env.start()
+        self.addCleanup(env.stop)
+
+    def _config(self, data: dict):
+        (self.project / ".c3" / "config.json").write_text(
+            json.dumps(data), encoding="utf-8")
+
+    def _patched(self, *, registered=True, registry=None, port_alive=True,
+                 launched=True, error=""):
+        from services.project_manager import ProjectManager
+        projects = [{"path": str(self.project.resolve())}] if registered else []
+        result = {"launched": launched}
+        if not launched:
+            result["error"] = error
+        return (
+            mock.patch.object(ProjectManager, "_read_projects", return_value=projects),
+            mock.patch.object(ProjectManager, "_read_registry",
+                              return_value=registry if registry is not None else []),
+            mock.patch.object(ProjectManager, "_port_alive", return_value=port_alive),
+            mock.patch.object(ProjectManager, "launch_session", return_value=result),
+        )
+
+    def _run(self, **kwargs):
+        patches = self._patched(**kwargs)
+        started = [p.start() for p in patches]
+        try:
+            return hook_session_open.ensure_ui(self.project, HOST_SID), started[-1]
+        finally:
+            for p in reversed(patches):
+                p.stop()
+
+    def test_launches_when_nothing_is_running(self):
+        verdict, launch = self._run()
+        self.assertEqual(verdict, "launched")
+        launch.assert_called_once_with(str(self.project.resolve()))
+        self.assertEqual(os.environ.get("C3_UI_OWNER"), "session")
+        self.assertEqual(os.environ.get("C3_UI_OWNER_SESSION"), HOST_SID)
+
+    def test_live_ui_for_this_project_is_left_alone(self):
+        registry = [{"port": 4321, "project_path": str(self.project.resolve())}]
+        verdict, launch = self._run(registry=registry)
+        self.assertEqual(verdict, "already-running")
+        launch.assert_not_called()
+
+    def test_dead_port_does_not_count_as_running(self):
+        registry = [{"port": 4321, "project_path": str(self.project.resolve())}]
+        verdict, launch = self._run(registry=registry, port_alive=False)
+        self.assertEqual(verdict, "launched")
+        launch.assert_called_once()
+
+    def test_another_projects_ui_does_not_count(self):
+        registry = [{"port": 4321, "project_path": str(Path(self.tmp.name) / "other")}]
+        verdict, launch = self._run(registry=registry)
+        self.assertEqual(verdict, "launched")
+        launch.assert_called_once()
+
+    def test_unregistered_project_gets_nothing(self):
+        verdict, launch = self._run(registered=False)
+        self.assertEqual(verdict, "unregistered")
+        launch.assert_not_called()
+
+    def test_config_opt_out(self):
+        self._config({"ui": {"autostart_on_session": False}})
+        verdict, launch = self._run()
+        self.assertEqual(verdict, "disabled")
+        launch.assert_not_called()
+
+    def test_config_opt_in_is_the_default(self):
+        self._config({"ui": {}})
+        verdict, _ = self._run()
+        self.assertEqual(verdict, "launched")
+
+    def test_env_opt_outs(self):
+        for var in ("C3_NO_UI_AUTOSTART", "C3_BENCHMARK_MODE"):
+            with mock.patch.dict(os.environ, {var: "1"}):
+                verdict, launch = self._run()
+            self.assertEqual(verdict, "disabled", var)
+            launch.assert_not_called()
+
+    def test_launch_failure_is_reported_not_raised(self):
+        verdict, _ = self._run(launched=False, error="boom")
+        self.assertTrue(verdict.startswith("failed"), verdict)
+        self.assertIn("boom", verdict)
+
+    def test_session_open_calls_ensure_ui_with_the_host_id(self):
+        with mock.patch.object(hook_session_open, "ensure_ui",
+                              return_value="launched") as ensure:
+            hook_session_open.run(
+                {"session_id": HOST_SID, "cwd": str(self.project),
+                 "hook_event_name": "SessionStart", "source": "startup"},
+                self.project)
+        ensure.assert_called_once_with(self.project, HOST_SID)
+
+    def test_compact_never_touches_the_ui(self):
+        with mock.patch.object(hook_session_open, "ensure_ui") as ensure:
+            hook_session_open.run(
+                {"session_id": HOST_SID, "cwd": str(self.project),
+                 "source": "compact"}, self.project)
+        ensure.assert_not_called()
+
+
 class TestSessionEnd(_Base):
     def _payload(self, reason="prompt_input_exit", **extra):
         return {"session_id": HOST_SID, "cwd": str(self.project),
diff --git a/tests/test_hub_hook_migration.py b/tests/test_hub_hook_migration.py
new file mode 100644
index 0000000..870510c
--- /dev/null
+++ b/tests/test_hub_hook_migration.py
@@ -0,0 +1,196 @@
+"""The hub's hook migration (v2.128.0) — how an installed project catches up.
+
+Adding a hook to `c3 install-mcp` does nothing for a project that is already
+installed. The v2.126.0 SessionStart / SessionEnd hooks landed in the
+installer only, so every existing project — 60 of them on the machine this was
+found on, C3's own repo among them — kept writing no `session_open` /
+`session_end` rows at all, and liveness had no end-of-session signal to read.
+
+`_migrate_project_hooks` already rewrote every registered project's settings
+at hub startup, but for exactly one hardcoded entry. These pin the
+data-driven version: additive, idempotent, never creates a settings file, and
+never drops a user's own hooks.
+"""
+from __future__ import annotations
+
+import json
+import shlex
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+
+from cli import hub_server  # noqa: E402
+
+CLAUDE_SETTINGS = ".claude/settings.local.json"
+GEMINI_SETTINGS = ".gemini/settings.json"
+
+
+class TestHookMigration(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.project = Path(self.tmp.name)
+
+    def _write(self, rel: str, settings: dict):
+        path = self.project / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(settings, indent=2), encoding="utf-8")
+        return path
+
+    def _read(self, rel: str) -> dict:
+        return json.loads((self.project / rel).read_text(encoding="utf-8"))
+
+    def _commands(self, settings: dict, event: str) -> list:
+        return [hk.get("command", "")
+                for entry in settings.get("hooks", {}).get(event, [])
+                for hk in entry.get("hooks", [])]
+
+    def _migrate(self) -> int:
+        return hub_server.migrate_hooks_for_project(str(self.project))
+
+    # ── the gap this closes ───────────────────────────────────────────────
+
+    def test_lifecycle_hooks_are_added_to_an_installed_project(self):
+        self._write(CLAUDE_SETTINGS, {"hooks": {"PostToolUse": []}})
+        self.assertEqual(self._migrate(), 1)
+        settings = self._read(CLAUDE_SETTINGS)
+        for event, arg in (("SessionStart", "start"), ("SessionEnd", "end")):
+            commands = self._commands(settings, event)
+            self.assertEqual(len(commands), 1, event)
+            self.assertIn("hook_dispatch.py", commands[0])
+            self.assertTrue(commands[0].rstrip().endswith(arg), commands[0])
+
+    def test_c3read_hook_still_lands(self):
+        self._write(CLAUDE_SETTINGS, {"hooks": {}})
+        self._migrate()
+        matchers = [e.get("matcher") for e
+                    in self._read(CLAUDE_SETTINGS)["hooks"]["PostToolUse"]]
+        self.assertIn("mcp__c3__c3_read", matchers)
+
+    def test_settings_with_no_hooks_key_at_all(self):
+        self._write(CLAUDE_SETTINGS, {"permissions": {"allow": []}})
+        self.assertEqual(self._migrate(), 1)
+        settings = self._read(CLAUDE_SETTINGS)
+        self.assertEqual(settings["permissions"], {"allow": []})
+        self.assertIn("SessionStart", settings["hooks"])
+
+    # ── idempotence and user content ──────────────────────────────────────
+
+    def test_second_run_changes_nothing(self):
+        self._write(CLAUDE_SETTINGS, {"hooks": {}})
+        self.assertEqual(self._migrate(), 1)
+        before = self._read(CLAUDE_SETTINGS)
+        self.assertEqual(self._migrate(), 0)
+        self.assertEqual(self._read(CLAUDE_SETTINGS), before)
+
+    def test_user_hooks_for_the_same_event_survive(self):
+        user_entry = {"matcher": "", "hooks": [
+            {"type": "command", "command": "powershell notify.ps1 start"}]}
+        self._write(CLAUDE_SETTINGS, {"hooks": {"SessionStart": [user_entry]}})
+        self._migrate()
+        commands = self._commands(self._read(CLAUDE_SETTINGS), "SessionStart")
+        self.assertEqual(len(commands), 2)
+        self.assertIn("powershell notify.ps1 start", commands)
+
+    def test_an_existing_c3_entry_is_not_duplicated(self):
+        existing = {"matcher": "", "hooks": [
+            {"type": "command", "command": "python /elsewhere/cli/hook_dispatch.py start"}]}
+        self._write(CLAUDE_SETTINGS, {"hooks": {"SessionStart": [existing]}})
+        self.assertEqual(
+            len(self._commands(self._read(CLAUDE_SETTINGS), "SessionStart")), 1)
+        self._migrate()
+        self.assertEqual(
+            len(self._commands(self._read(CLAUDE_SETTINGS), "SessionStart")), 1)
+
+    # ── the command string has to actually run ────────────────────────────
+
+    def test_commands_are_quoted_for_this_platform(self):
+        """shlex.quote is WRONG on Windows: single quotes are not quoting to
+        cmd.exe, so an entry for any path with a space (every path under
+        "1. Projects") never ran. The migration used it until 2.128.0."""
+        self._write(CLAUDE_SETTINGS, {"hooks": {}})
+        self._migrate()
+        settings = self._read(CLAUDE_SETTINGS)
+        commands = (self._commands(settings, "SessionStart")
+                    + self._commands(settings, "SessionEnd")
+                    + self._commands(settings, "PostToolUse"))
+        self.assertTrue(commands)
+        for command in commands:
+            tokens = shlex.split(command, posix=(sys.platform != "win32"))
+            script = Path(tokens[1].strip('"'))
+            self.assertTrue(script.is_file(), command)
+            if sys.platform == "win32":
+                self.assertNotIn("'", command, command)
+                self.assertNotIn("\\", command, command)
+
+    def test_dispatcher_command_matches_the_installer_exactly(self):
+        from cli._hook_utils import hook_command_arg
+        cli_dir = Path(hub_server.__file__).resolve().parent
+        expected = (f"{hook_command_arg(sys.executable)} "
+                    f"{hook_command_arg(str(cli_dir / 'hook_dispatch.py'))} start")
+        self._write(CLAUDE_SETTINGS, {"hooks": {}})
+        self._migrate()
+        self.assertIn(expected,
+                      self._commands(self._read(CLAUDE_SETTINGS), "SessionStart"))
+
+    # ── what it refuses to do ─────────────────────────────────────────────
+
+    def test_missing_settings_file_is_not_created(self):
+        self.assertEqual(self._migrate(), 0)
+        self.assertFalse((self.project / CLAUDE_SETTINGS).exists())
+
+    def test_unparseable_settings_are_left_untouched(self):
+        path = self.project / CLAUDE_SETTINGS
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text("{not json", encoding="utf-8")
+        self.assertEqual(self._migrate(), 0)
+        self.assertEqual(path.read_text(encoding="utf-8"), "{not json")
+
+    def test_gemini_gets_its_tool_hook_and_no_lifecycle_events(self):
+        self._write(GEMINI_SETTINGS, {"hooks": {}})
+        self.assertEqual(self._migrate(), 1)
+        hooks = self._read(GEMINI_SETTINGS)["hooks"]
+        self.assertIn("AfterTool", hooks)
+        self.assertNotIn("SessionStart", hooks)
+        self.assertNotIn("SessionEnd", hooks)
+
+    def test_both_ides_in_one_project_are_one_pass_each(self):
+        self._write(CLAUDE_SETTINGS, {"hooks": {}})
+        self._write(GEMINI_SETTINGS, {"hooks": {}})
+        self.assertEqual(self._migrate(), 2)
+
+
+class TestMigrateAllProjects(unittest.TestCase):
+    def test_every_registered_project_is_visited(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            projects = []
+            for name in ("a", "b"):
+                path = Path(tmp) / name
+                (path / ".claude").mkdir(parents=True)
+                (path / ".claude" / "settings.local.json").write_text(
+                    "{}", encoding="utf-8")
+                projects.append({"path": str(path)})
+            pm = mock.Mock()
+            pm.list_projects.return_value = projects + [{"path": ""}]
+            with mock.patch.object(hub_server, "_pm", return_value=pm):
+                hub_server._migrate_project_hooks()
+            for project in projects:
+                settings = json.loads(
+                    (Path(project["path"]) / ".claude" / "settings.local.json")
+                    .read_text(encoding="utf-8"))
+                self.assertIn("SessionStart", settings["hooks"])
+
+    def test_a_broken_project_list_is_survivable(self):
+        pm = mock.Mock()
+        pm.list_projects.side_effect = OSError("registry gone")
+        with mock.patch.object(hub_server, "_pm", return_value=pm):
+            hub_server._migrate_project_hooks()  # must not raise
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_project_manager.py b/tests/test_project_manager.py
index cd709d1..81699ec 100644
--- a/tests/test_project_manager.py
+++ b/tests/test_project_manager.py
@@ -1,8 +1,13 @@
+import json
+import os
 import tempfile
+import time
 import unittest
 from datetime import datetime, timedelta, timezone
+from pathlib import Path
 from unittest.mock import patch
 
+from services import session_live
 from services.project_manager import ProjectManager
 
 
@@ -99,7 +104,7 @@ def test_list_projects_marks_live_session_active_without_ui_port(self):
         }
         with patch.object(self.pm, "_read_projects", return_value=[{"name": "Proj", "path": self.proj_path, "added_at": now}]), \
              patch.object(self.pm, "_read_registry", return_value=[]), \
-             patch.object(self.pm, "_get_live_session_info", return_value=live_session), \
+             patch.object(self.pm, "_get_live_sessions", return_value=[live_session]), \
              patch.object(self.pm, "_read_project_config", return_value={}), \
              patch.object(self.pm, "_get_budget_info", return_value={"response_tokens": 123}):
             projects = self.pm.list_projects()
@@ -107,6 +112,8 @@ def test_list_projects_marks_live_session_active_without_ui_port(self):
         self.assertTrue(projects[0]["active"])
         self.assertIsNone(projects[0]["port"])
         self.assertEqual(projects[0]["live_session_id"], "sess-1")
+        self.assertEqual(projects[0]["session_count"], 1)
+        self.assertEqual(projects[0]["sessions"], [live_session])
         self.assertEqual(projects[0]["budget"], {"response_tokens": 123})
 
     def test_get_active_sessions_includes_live_session_without_ui_port(self):
@@ -130,5 +137,209 @@ def test_get_active_sessions_includes_live_session_without_ui_port(self):
         }])
 
 
+class TestLivenessSources(unittest.TestCase):
+    """v2.128.0: heartbeat files are the truth, and session_end finally counts.
+
+    Before this, `_get_live_session_info` ended a session only on a
+    `session_save` row (written by the hub's end button) or after 20 minutes of
+    tool-call silence — so a closed IDE read "live" for 20 minutes and a
+    quiet-but-open session read dead.
+    """
+
+    def setUp(self):
+        self.pm = ProjectManager()
+        self._tmp = tempfile.TemporaryDirectory()
+        self.proj_path = self._tmp.name
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _ago(self, **kwargs) -> str:
+        return (datetime.now(timezone.utc) - timedelta(**kwargs)).isoformat()
+
+    def _beat(self, session_id, *, host_session_id="", age_seconds=0):
+        session_live.beat(self.proj_path, session_id,
+                          host_session_id=host_session_id, ide="claude-code")
+        if age_seconds:
+            path = session_live.heartbeat_path(self.proj_path, session_id)
+            data = json.loads(path.read_text(encoding="utf-8"))
+            data["ts"] = time.time() - age_seconds
+            path.write_text(json.dumps(data), encoding="utf-8")
+
+    def _sessions(self, events):
+        with patch("services.project_manager.ActivityLog",
+                   return_value=_StubActivityLog(events)):
+            return self.pm._get_live_sessions(self.proj_path)
+
+    # ── session_end is terminal ───────────────────────────────────────────
+
+    def test_session_end_row_ends_the_session(self):
+        start, recent = self._ago(minutes=5), self._ago(minutes=1)
+        self.assertEqual(self._sessions([
+            {"type": "session_start", "session_id": "s1", "timestamp": start},
+            {"type": "tool_call", "timestamp": recent},
+            {"type": "session_end", "session_id": "s1", "timestamp": recent},
+        ]), [])
+
+    def test_session_end_for_another_session_is_ignored(self):
+        sessions = self._sessions([
+            {"type": "session_end", "session_id": "s0", "timestamp": self._ago(minutes=6)},
+            {"type": "session_start", "session_id": "s1", "timestamp": self._ago(minutes=5)},
+            {"type": "tool_call", "timestamp": self._ago(minutes=1)},
+        ])
+        self.assertEqual([s["session_id"] for s in sessions], ["s1"])
+
+    def test_session_end_matches_on_host_id_alone(self):
+        """The hook writes session_id="" when no link file exists yet."""
+        recent = self._ago(minutes=1)
+        self.assertEqual(self._sessions([
+            {"type": "session_start", "session_id": "s1", "host_session_id": "h1",
+             "timestamp": self._ago(minutes=5)},
+            {"type": "tool_call", "timestamp": recent},
+            {"type": "session_end", "session_id": "", "host_session_id": "h1",
+             "timestamp": recent},
+        ]), [])
+
+    # ── heartbeats outrank inference ──────────────────────────────────────
+
+    def test_heartbeat_keeps_a_long_idle_session_live(self):
+        old = self._ago(hours=2)
+        self._beat("s1", host_session_id="h1")
+        sessions = self._sessions([
+            {"type": "session_start", "session_id": "s1", "host_session_id": "h1",
+             "timestamp": old, "description": "quiet but open"},
+            {"type": "tool_call", "timestamp": old},
+        ])
+        self.assertEqual(len(sessions), 1)
+        self.assertEqual(sessions[0]["source"], "heartbeat")
+        self.assertEqual(sessions[0]["started_at"], old)
+        self.assertEqual(sessions[0]["description"], "quiet but open")
+        self.assertGreater(sessions[0]["idle_seconds"], 20 * 60)
+
+    def test_stale_heartbeat_does_not_resurrect_a_dead_session(self):
+        old = self._ago(hours=2)
+        self._beat("s1", age_seconds=session_live.HEARTBEAT_TTL_S + 60)
+        self.assertEqual(self._sessions([
+            {"type": "session_start", "session_id": "s1", "timestamp": old},
+            {"type": "tool_call", "timestamp": old},
+        ]), [])
+
+    def test_heartbeat_outranks_a_hub_session_save(self):
+        """The process is still serving; a bookkeeping row does not kill it."""
+        self._beat("s1")
+        sessions = self._sessions([
+            {"type": "session_start", "session_id": "s1", "timestamp": self._ago(minutes=10)},
+            {"type": "session_save", "session_id": "s1", "timestamp": self._ago(minutes=1)},
+        ])
+        self.assertEqual([s["session_id"] for s in sessions], ["s1"])
+
+    def test_heartbeat_without_a_start_row_still_counts(self):
+        self._beat("s1")
+        sessions = self._sessions([])
+        self.assertEqual(len(sessions), 1)
+        self.assertTrue(sessions[0]["started_at"])
+
+    def test_heartbeat_session_is_not_duplicated_by_the_inference(self):
+        self._beat("s1")
+        self.assertEqual(len(self._sessions([
+            {"type": "session_start", "session_id": "s1", "timestamp": self._ago(minutes=5)},
+            {"type": "tool_call", "timestamp": self._ago(minutes=1)},
+        ])), 1)
+
+    def test_two_heartbeats_are_two_sessions(self):
+        older, newer = self._ago(hours=3), self._ago(minutes=4)
+        self._beat("s-old")
+        self._beat("s-new")
+        events = [
+            {"type": "session_start", "session_id": "s-old", "timestamp": older},
+            {"type": "session_start", "session_id": "s-new", "timestamp": newer},
+        ]
+        self.assertEqual([s["session_id"] for s in self._sessions(events)],
+                         ["s-new", "s-old"])
+        with patch.object(self.pm, "_read_projects", return_value=[
+                {"name": "Proj", "path": self.proj_path, "added_at": older}]), \
+             patch.object(self.pm, "_read_registry", return_value=[]), \
+             patch.object(self.pm, "_read_project_config", return_value={}), \
+             patch("services.project_manager.ActivityLog",
+                   return_value=_StubActivityLog(events)):
+            projects = self.pm.list_projects()
+        self.assertEqual(projects[0]["session_count"], 2)
+        self.assertEqual(projects[0]["live_session_id"], "s-new")
+
+
+class TestRegistryReaping(unittest.TestCase):
+    """Only session-launched UI servers are ever stopped automatically."""
+
+    def setUp(self):
+        self.pm = ProjectManager()
+        self._tmp = tempfile.TemporaryDirectory()
+        self.proj_path = self._tmp.name
+        self.log = Path(self.proj_path) / ".c3" / "activity_log.jsonl"
+        self.log.parent.mkdir(parents=True, exist_ok=True)
+        self.log.write_text("", encoding="utf-8")
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _entry(self, **extra) -> dict:
+        entry = {"port": 4321, "project_path": self.proj_path,
+                 "project_name": "Proj", "started_at": time.time() - 7200,
+                 "owner": "session", "owner_session": "s1", "pid": 999999}
+        entry.update(extra)
+        return entry
+
+    def _silent_for(self, seconds: float):
+        when = time.time() - seconds
+        os.utime(self.log, (when, when))
+
+    def _sweep(self, entry):
+        registry_file = Path(self.proj_path) / "registry.json"
+        with patch.object(self.pm, "_read_registry", return_value=[entry]), \
+             patch.object(self.pm, "_port_alive", return_value=True), \
+             patch.object(self.pm, "_stop_entry", return_value=True) as stop, \
+             patch("services.project_manager._REGISTRY_FILE", registry_file):
+            return self.pm.sweep_registry(), stop
+
+    def test_orphaned_session_ui_is_reaped(self):
+        self._silent_for(3600)
+        result, stop = self._sweep(self._entry())
+        self.assertEqual(result["reaped"], [4321])
+        stop.assert_called_once()
+
+    def test_user_owned_ui_is_never_reaped(self):
+        self._silent_for(3600)
+        result, stop = self._sweep(self._entry(owner="user"))
+        self.assertEqual(result["reaped"], [])
+        stop.assert_not_called()
+
+    def test_recent_activity_protects_the_ui(self):
+        self._silent_for(60)
+        result, _ = self._sweep(self._entry())
+        self.assertEqual(result["reaped"], [])
+
+    def test_a_live_session_protects_the_ui(self):
+        self._silent_for(3600)
+        session_live.beat(self.proj_path, "s1")
+        result, _ = self._sweep(self._entry())
+        self.assertEqual(result["reaped"], [])
+
+    def test_reaping_can_be_disabled(self):
+        self._silent_for(3600)
+        with patch.object(self.pm, "_reap_minutes", return_value=0):
+            result, stop = self._sweep(self._entry())
+        self.assertEqual(result["reaped"], [])
+        stop.assert_not_called()
+
+    def test_dead_port_is_dropped_not_reaped(self):
+        with patch.object(self.pm, "_read_registry", return_value=[self._entry()]), \
+             patch.object(self.pm, "_port_alive", return_value=False), \
+             patch.object(self.pm, "_stop_entry", return_value=True) as stop, \
+             patch("services.project_manager._REGISTRY_FILE",
+                   Path(self.proj_path) / "registry.json"):
+            result = self.pm.sweep_registry()
+        self.assertEqual(result, {"dropped": 1, "reaped": []})
+        stop.assert_not_called()
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/test_session_live.py b/tests/test_session_live.py
new file mode 100644
index 0000000..3932857
--- /dev/null
+++ b/tests/test_session_live.py
@@ -0,0 +1,142 @@
+"""Session heartbeat files (v2.128.0) — .c3/live/<session>.json.
+
+Liveness used to be inferred from activity rows: a closed IDE read "live" for
+up to 20 minutes (nothing consumed the session_end row), and an open-but-quiet
+session read dead after 20 minutes of no tool calls. These pin the file
+contract the MCP server writes and the hub reads.
+"""
+from __future__ import annotations
+
+import json
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+
+from services import session_live  # noqa: E402
+
+
+class TestSessionLive(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.project = Path(self.tmp.name)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _age(self, session_id: str, seconds: float):
+        """Backdate one heartbeat without waiting for the clock."""
+        path = session_live.heartbeat_path(self.project, session_id)
+        data = json.loads(path.read_text(encoding="utf-8"))
+        data["ts"] = time.time() - seconds
+        path.write_text(json.dumps(data), encoding="utf-8")
+
+    def test_beat_then_read_back(self):
+        self.assertTrue(session_live.beat(
+            self.project, "s1", host_session_id="host-1", ide="claude-code"))
+        live = session_live.live_sessions(self.project)
+        self.assertEqual(len(live), 1)
+        self.assertEqual(live[0]["session_id"], "s1")
+        self.assertEqual(live[0]["host_session_id"], "host-1")
+        self.assertEqual(live[0]["ide"], "claude-code")
+        self.assertIsInstance(live[0]["pid"], int)
+        self.assertLess(live[0]["age_seconds"], 5)
+
+    def test_empty_session_id_is_refused(self):
+        self.assertFalse(session_live.beat(self.project, ""))
+        self.assertEqual(session_live.live_sessions(self.project), [])
+
+    def test_no_live_dir_is_not_an_error(self):
+        self.assertEqual(session_live.live_sessions(self.project), [])
+        self.assertFalse(session_live.is_live(self.project, "s1"))
+
+    def test_clear_removes_the_file(self):
+        session_live.beat(self.project, "s1")
+        self.assertTrue(session_live.clear(self.project, "s1"))
+        self.assertEqual(session_live.live_sessions(self.project), [])
+        # Second clear is a no-op, not a crash (teardown can run twice).
+        self.assertFalse(session_live.clear(self.project, "s1"))
+
+    def test_stale_heartbeat_is_not_live_and_is_pruned(self):
+        session_live.beat(self.project, "s1")
+        self._age("s1", session_live.HEARTBEAT_TTL_S + 60)
+        self.assertEqual(session_live.live_sessions(self.project), [])
+        self.assertFalse(session_live.heartbeat_path(self.project, "s1").exists())
+
+    def test_stale_heartbeat_survives_prune_false(self):
+        session_live.beat(self.project, "s1")
+        self._age("s1", session_live.HEARTBEAT_TTL_S + 60)
+        self.assertEqual(session_live.live_sessions(self.project, prune=False), [])
+        self.assertTrue(session_live.heartbeat_path(self.project, "s1").exists())
+
+    def test_just_inside_the_ttl_is_live(self):
+        session_live.beat(self.project, "s1")
+        self._age("s1", session_live.HEARTBEAT_TTL_S - 5)
+        self.assertEqual(len(session_live.live_sessions(self.project)), 1)
+
+    def test_idle_session_stays_live(self):
+        """The whole point: no tool call for hours, still live."""
+        session_live.beat(self.project, "s1")
+        self._age("s1", 30)  # last beat 30 s ago, last tool call irrelevant
+        self.assertTrue(session_live.is_live(self.project, "s1"))
+
+    def test_two_sessions_are_two_entries_newest_first(self):
+        session_live.beat(self.project, "older")
+        session_live.beat(self.project, "newer")
+        self._age("older", 100)
+        live = session_live.live_sessions(self.project)
+        self.assertEqual([s["session_id"] for s in live], ["newer", "older"])
+
+    def test_is_live_matches_either_id(self):
+        session_live.beat(self.project, "c3-id", host_session_id="host-id")
+        self.assertTrue(session_live.is_live(self.project, "c3-id"))
+        self.assertTrue(session_live.is_live(self.project, "host-id"))
+        self.assertFalse(session_live.is_live(self.project, "someone-else"))
+        self.assertFalse(session_live.is_live(self.project, ""))
+
+    def test_unsafe_session_id_stays_inside_live_dir(self):
+        session_live.beat(self.project, "../../escape")
+        files = list(session_live.live_dir(self.project).iterdir())
+        self.assertEqual(len(files), 1)
+        self.assertEqual(files[0].parent, session_live.live_dir(self.project))
+
+    def test_unreadable_file_is_ignored_not_fatal(self):
+        session_live.beat(self.project, "s1")
+        session_live.heartbeat_path(self.project, "s1").write_text(
+            "{half-written", encoding="utf-8")
+        self.assertEqual(session_live.live_sessions(self.project), [])
+
+    def test_concurrent_beats_never_corrupt(self):
+        """Six threads on one path: the Windows failure mode atomic_json exists for."""
+        errors: list = []
+
+        def hammer():
+            try:
+                for _ in range(15):
+                    session_live.beat(self.project, "shared")
+                    session_live.live_sessions(self.project, prune=False)
+            except Exception as exc:  # pragma: no cover - failure detail
+                errors.append(exc)
+
+        threads = [threading.Thread(target=hammer) for _ in range(6)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        self.assertEqual(errors, [])
+        live = session_live.live_sessions(self.project)
+        self.assertEqual(len(live), 1)
+        self.assertEqual(live[0]["session_id"], "shared")
+        # No abandoned temp files left beside it.
+        self.assertEqual(
+            [p.name for p in session_live.live_dir(self.project).iterdir()],
+            ["shared.json"])
+
+
+if __name__ == "__main__":
+    unittest.main()