feat(bench): add CLI startup latency benchmark

Measures wall-clock time from omnigent claude --server invocation to the
Claude terminal being ready (signalled by 'Claude terminal ready.' spinner
message, emitted just before tmux attach).

Unlike the HTTP/API benchmarks in run.py, this drives the real CLI binary
end-to-end against a remote server — auth, daemon tunnel, session create,
runner launch, terminal boot — via pexpect.

Usage:
  uv run --no-sync dev/benchmarks/omnigent/cli_startup.py
  uv run --no-sync dev/benchmarks/omnigent/cli_startup.py --also-isaac-omni --runs 10
  uv run --no-sync dev/benchmarks/omnigent/cli_startup.py --output startup.json
  uv run --no-sync dev/benchmarks/omnigent/cli_startup.py --max-p50-ms 12000

JSON output is compatible with the existing benchmark schema.

Signed-off-by: Tomu Hirata <tomu.hirata@gmail.com>

dev/benchmarks/omnigent/cli_startup.py

@@ -0,0 +1,430 @@
+"""CLI startup latency benchmark.
+
+Measures the wall-clock time from ``omnigent claude --server <url>`` invocation
+to the Claude Code TUI being ready for input (signalled by "This session cost"
+appearing in the output — the last line rendered before the prompt is active).
+
+Unlike the HTTP/API benchmarks in ``run.py``, this drives the real CLI binary
+end-to-end against a remote server: it exercises the full startup sequence
+including auth, daemon tunnel, session create, runner launch, and terminal boot.
+
+Usage::
+
+    # Measure against the ai-devtools workspace (default), 5 runs:
+    uv run --no-sync dev/benchmarks/omnigent/cli_startup.py
+
+    # Custom server, more runs:
+    uv run --no-sync dev/benchmarks/omnigent/cli_startup.py \\
+        --server https://dbc-xxxx.cloud.databricks.com/api/2.0/omnigent \\
+        --runs 10
+
+    # Also run ``isaac omni`` for comparison:
+    uv run --no-sync dev/benchmarks/omnigent/cli_startup.py --also-isaac-omni
+
+    # Write JSON output:
+    uv run --no-sync dev/benchmarks/omnigent/cli_startup.py --output startup.json
+
+    # CI threshold gate (exit 1 if p50 > N ms):
+    uv run --no-sync dev/benchmarks/omnigent/cli_startup.py --max-p50-ms 12000
+
+The JSON schema is compatible with the existing benchmark report so the same
+Databricks ETL notebook can ingest it. The journey name is ``cli_startup``
+(or ``isaac_omni`` for the isaac variant).
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime
+import json
+import shutil
+import statistics
+import sys
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
+
+from dev.benchmarks.omnigent.schema import SCHEMA_VERSION, git_branch, git_sha, host_info
+
+try:
+    import pexpect
+
+    _PEXPECT_AVAILABLE = True
+except ImportError:
+    _PEXPECT_AVAILABLE = False
+
+try:
+    from rich.console import Console
+    from rich.table import Table
+
+    console = Console()
+    _RICH_AVAILABLE = True
+except ImportError:
+    _RICH_AVAILABLE = False
+    console = None  # type: ignore[assignment]
+
+# Signal that the Claude terminal is ready — emitted by the startup spinner
+# just before the tmux attach hands off to the Claude Code TUI. Using this
+# rather than a signal from inside the TUI avoids needing a live tmux attach
+# to complete successfully in the benchmark's PTY environment.
+_READY_SIGNAL = "Claude terminal ready"
+
+# Default remote server used by the ai-devtools workspace.
+_DEFAULT_SERVER = "https://dbc-a5d4177a-49dc.cloud.databricks.com/api/2.0/omnigent"
+
+# Timeout per startup attempt (seconds). Generous to handle slow terminal boots.
+_TIMEOUT_S = 90
+
+
+@dataclass
+class StartupResult:
+    """Results for one benchmark command over all runs.
+
+    :param command: The command label, e.g. ``"omnigent claude --server"``.
+    :param latencies_ms: Per-run wall-clock latency in milliseconds.
+    :param failures: Failure reason mapped to count.
+    """
+
+    command: str
+    latencies_ms: list[float] = field(default_factory=list)
+    failures: dict[str, int] = field(default_factory=dict)
+
+
+def _percentile(data: list[float], p: float) -> float:
+    """Return the *p*-th percentile of *data* (0–100)."""
+    if not data:
+        return float("nan")
+    sorted_data = sorted(data)
+    idx = (p / 100) * (len(sorted_data) - 1)
+    lo, hi = int(idx), min(int(idx) + 1, len(sorted_data) - 1)
+    return sorted_data[lo] + (sorted_data[hi] - sorted_data[lo]) * (idx - lo)
+
+
+def _measure_startup(cmd: list[str], *, timeout_s: float = _TIMEOUT_S) -> float:
+    """Spawn *cmd*, wait for the TUI ready signal, return elapsed ms.
+
+    Sends ``/exit`` once the signal is seen so the session is cleaned up
+    (doesn't count toward the measurement).
+
+    :param cmd: Command + args to spawn, e.g. ``["omnigent", "claude", "--server", ...]``.
+    :param timeout_s: Max seconds to wait for the ready signal.
+    :returns: Wall-clock milliseconds from spawn to ready signal.
+    :raises RuntimeError: On timeout, spawn failure, or known server error.
+    """
+    if not _PEXPECT_AVAILABLE:
+        raise RuntimeError(
+            "pexpect is required for the CLI startup benchmark. "
+            "Install it with: pip install pexpect"
+        )
+
+    env = dict(__import__("os").environ)
+    start = time.perf_counter()
+    child = pexpect.spawn(
+        cmd[0],
+        args=cmd[1:],
+        timeout=timeout_s,
+        encoding="utf-8",
+        codec_errors="ignore",
+        env=env,
+    )
+    try:
+        idx = child.expect([pexpect.TIMEOUT, pexpect.EOF, _READY_SIGNAL])
+        elapsed_ms = (time.perf_counter() - start) * 1000
+        if idx == 0:
+            raise RuntimeError(f"Timed out after {timeout_s}s waiting for TUI ready signal")
+        if idx == 1:
+            # Capture the output to provide an actionable error message.
+            output = (child.before or "").strip()
+            if "another replica" in output:
+                raise RuntimeError(
+                    "host is on another replica — stale daemon from a previous run. "
+                    "Run `omnigent stop` to clear it and retry."
+                )
+            raise RuntimeError(
+                f"Process exited before TUI ready signal appeared. Last output: {output[-200:]!r}"
+            )
+        # idx == 2: matched the ready signal
+        child.sendline("/exit")
+        child.expect([pexpect.EOF, pexpect.TIMEOUT], timeout=10)
+    finally:
+        if child.isalive():
+            child.terminate(force=True)
+    return elapsed_ms
+
+
+def _run_benchmark(
+    cmd: list[str],
+    *,
+    label: str,
+    runs: int,
+    verbose: bool = False,
+) -> StartupResult:
+    """Run the startup benchmark *runs* times and return collected results."""
+    result = StartupResult(command=label)
+    for i in range(runs):
+        run_num = i + 1
+        if verbose:
+            print(f"  {label} run {run_num}/{runs}...", flush=True)
+        try:
+            elapsed_ms = _measure_startup(cmd)
+            result.latencies_ms.append(elapsed_ms)
+            if verbose:
+                print(f"    → {elapsed_ms:.0f}ms", flush=True)
+        except BaseException as exc:  # noqa: BLE001
+            reason = type(exc).__name__
+            result.failures[reason] = result.failures.get(reason, 0) + 1
+            if verbose:
+                print(f"    → FAILED: {exc}", flush=True)
+    return result
+
+
+def _print_table(results: list[StartupResult]) -> None:
+    """Print a Rich summary table."""
+    if not _RICH_AVAILABLE or console is None:
+        _print_plain(results)
+        return
+
+    table = Table(title="CLI Startup Latency", show_header=True, header_style="bold")
+    table.add_column("Command", style="cyan", no_wrap=True)
+    table.add_column("Runs", justify="right")
+    table.add_column("Failures", justify="right")
+    table.add_column("Min (ms)", justify="right")
+    table.add_column("Avg (ms)", justify="right")
+    table.add_column("p50 (ms)", justify="right")
+    table.add_column("p95 (ms)", justify="right")
+    table.add_column("p99 (ms)", justify="right")
+    table.add_column("Max (ms)", justify="right")
+
+    for r in results:
+        n_ok = len(r.latencies_ms)
+        n_fail = sum(r.failures.values())
+        if not r.latencies_ms:
+            table.add_row(r.command, str(n_fail), str(n_fail), *["—"] * 7)
+            continue
+        table.add_row(
+            r.command,
+            str(n_ok),
+            str(n_fail),
+            f"{min(r.latencies_ms):.0f}",
+            f"{statistics.mean(r.latencies_ms):.0f}",
+            f"{_percentile(r.latencies_ms, 50):.0f}",
+            f"{_percentile(r.latencies_ms, 95):.0f}",
+            f"{_percentile(r.latencies_ms, 99):.0f}",
+            f"{max(r.latencies_ms):.0f}",
+        )
+
+    console.print(table)
+
+
+def _print_plain(results: list[StartupResult]) -> None:
+    """Fallback plain-text summary."""
+    for r in results:
+        n_ok = len(r.latencies_ms)
+        n_fail = sum(r.failures.values())
+        if not r.latencies_ms:
+            print(f"{r.command}: all {n_fail} run(s) failed")
+            continue
+        print(
+            f"{r.command}: n={n_ok} failures={n_fail} "
+            f"avg={statistics.mean(r.latencies_ms):.0f}ms "
+            f"p50={_percentile(r.latencies_ms, 50):.0f}ms "
+            f"p95={_percentile(r.latencies_ms, 95):.0f}ms "
+            f"max={max(r.latencies_ms):.0f}ms"
+        )
+
+
+def _build_journey_entry(result: StartupResult) -> dict[str, Any]:
+    """Build a journey entry matching the existing benchmark JSON schema."""
+    runs_data = []
+    for ms in result.latencies_ms:
+        runs_data.append(
+            {
+                "n_success": 1,
+                "n_failures": 0,
+                "failures": {},
+                "wall_time_s": ms / 1000,
+                "mean_ms": ms,
+                "p50_ms": ms,
+                "p95_ms": ms,
+                "p99_ms": ms,
+                "max_ms": ms,
+                "rps": None,
+                "http_requests": None,
+                "http_requests_per_op": None,
+                "route_requests": {},
+            }
+        )
+    n_ok = len(result.latencies_ms)
+    summary: dict[str, Any] = {"runs_total": len(result.latencies_ms), "runs_ok": n_ok}
+    if result.latencies_ms:
+        summary.update(
+            {
+                "avg_mean_ms": statistics.mean(result.latencies_ms),
+                "avg_p50_ms": _percentile(result.latencies_ms, 50),
+                "avg_p95_ms": _percentile(result.latencies_ms, 95),
+                "avg_p99_ms": _percentile(result.latencies_ms, 99),
+                "avg_rps": None,
+            }
+        )
+    return {
+        "kind": "latency",
+        "backend": "remote",
+        "needs_runner": True,
+        "runs": runs_data,
+        "summary": summary,
+    }
+
+
+def _check_thresholds(
+    results: list[StartupResult],
+    *,
+    max_p50_ms: float | None,
+    max_p99_ms: float | None,
+) -> bool:
+    """Return True if all thresholds pass, False if any breach."""
+    passed = True
+    for r in results:
+        if not r.latencies_ms:
+            continue
+        p50 = _percentile(r.latencies_ms, 50)
+        p99 = _percentile(r.latencies_ms, 99)
+        if max_p50_ms is not None and p50 > max_p50_ms:
+            print(
+                f"THRESHOLD BREACH: {r.command} p50={p50:.0f}ms > {max_p50_ms:.0f}ms",
+                file=sys.stderr,
+            )
+            passed = False
+        if max_p99_ms is not None and p99 > max_p99_ms:
+            print(
+                f"THRESHOLD BREACH: {r.command} p99={p99:.0f}ms > {max_p99_ms:.0f}ms",
+                file=sys.stderr,
+            )
+            passed = False
+    return passed
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Benchmark CLI startup latency (time-to-prompt) for omnigent claude.",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+    parser.add_argument(
+        "--server",
+        default=_DEFAULT_SERVER,
+        help="Omnigent server URL passed to omnigent claude --server.",
+    )
+    parser.add_argument(
+        "--runs",
+        type=int,
+        default=5,
+        help="Number of startup attempts to time.",
+    )
+    parser.add_argument(
+        "--also-isaac-omni",
+        action="store_true",
+        help="Also benchmark `isaac omni` end-to-end (includes isaac pre-config overhead).",
+    )
+    parser.add_argument(
+        "--omnigent-bin",
+        default=None,
+        help="Path to the omnigent binary. Defaults to the one on PATH.",
+    )
+    parser.add_argument(
+        "--isaac-bin",
+        default=None,
+        help="Path to the isaac binary. Defaults to the one on PATH.",
+    )
+    parser.add_argument(
+        "--output",
+        default=None,
+        help="Write JSON results to this file (compatible with the benchmark schema).",
+    )
+    parser.add_argument(
+        "--max-p50-ms",
+        type=float,
+        default=None,
+        help="Fail (exit 1) if any command's p50 latency exceeds this value (ms).",
+    )
+    parser.add_argument(
+        "--max-p99-ms",
+        type=float,
+        default=None,
+        help="Fail (exit 1) if any command's p99 latency exceeds this value (ms).",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="Print per-run timing as it runs.",
+    )
+    args = parser.parse_args()
+
+    if not _PEXPECT_AVAILABLE:
+        print(
+            "ERROR: pexpect is required. Install with: pip install pexpect",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    omnigent_bin = args.omnigent_bin or shutil.which("omnigent")
+    if omnigent_bin is None:
+        print("ERROR: omnigent binary not found on PATH. Use --omnigent-bin.", file=sys.stderr)
+        sys.exit(1)
+
+    commands: list[tuple[str, list[str]]] = [
+        (
+            "omnigent claude --server",
+            [omnigent_bin, "claude", "--server", args.server],
+        ),
+    ]
+
+    if args.also_isaac_omni:
+        isaac_bin = args.isaac_bin or shutil.which("isaac")
+        if isaac_bin is None:
+            print(
+                "WARNING: isaac binary not found on PATH, skipping --also-isaac-omni.",
+                file=sys.stderr,
+            )
+        else:
+            commands.append(("isaac omni", [isaac_bin, "omni"]))
+
+    all_results: list[StartupResult] = []
+    for label, cmd in commands:
+        print(f"\nBenchmarking: {label} ({args.runs} run(s))")
+        result = _run_benchmark(cmd, label=label, runs=args.runs, verbose=args.verbose)
+        all_results.append(result)
+
+    print()
+    _print_table(all_results)
+
+    if args.output:
+        journeys = {
+            r.command.replace(" ", "_").replace("--", ""): _build_journey_entry(r)
+            for r in all_results
+        }
+        report: dict[str, Any] = {
+            "schema_version": SCHEMA_VERSION,
+            "generated_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
+            "git_sha": git_sha(),
+            "git_branch": git_branch(),
+            "host": host_info(),
+            "harness": "cli-startup",
+            "config": {
+                "runs": args.runs,
+                "server": args.server,
+                "also_isaac_omni": args.also_isaac_omni,
+            },
+            "journeys": journeys,
+        }
+        output_path = Path(args.output)
+        output_path.write_text(json.dumps(report, indent=2))
+        print(f"\nResults written to {output_path}")
+
+    if not _check_thresholds(all_results, max_p50_ms=args.max_p50_ms, max_p99_ms=args.max_p99_ms):
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

omnigent/antigravity_native.py

@@ -796,6 +796,7 @@ async def _prepare_antigravity_terminal_via_daemon(
         bridge_id: str
         conversation_id: str
         resume = False
+        fresh_session = session_id is None
         if session_id is None:
             if session_bundle is None:
                 raise click.ClickException(
@@ -858,6 +859,7 @@ async def _prepare_antigravity_terminal_via_daemon(
             host_id=host_id,
             session_id=session_id,
             workspace=workspace,
+            fresh=fresh_session,
         )
         _update_progress(startup_progress, "Waiting for runner...")
         await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)

omnigent/chat.py

@@ -662,6 +662,14 @@ def _remote_headers(
     return headers
 
 
+# Cache the resolved _DatabricksBearerAuth object per server URL so that
+# repeated calls to _remote_headers for the same URL reuse the same SDK
+# Config instance. The SDK's Config.authenticate() caches the OAuth token
+# in memory and only re-runs the CLI shell-out when it nears expiry, so
+# reusing the object is both fast and correct for long-running callers.
+_databricks_auth_cache: dict[str, object] = {}
+
+
 def _stored_databricks_record_token(server_url: str) -> str | None:
     """Mint a workspace token from a stored Databricks Apps record.
 
@@ -671,6 +679,12 @@ def _stored_databricks_record_token(server_url: str) -> str | None:
     that issue many requests should use :class:`_DatabricksTokenAuth`,
     which reuses the SDK config across requests.
 
+    The resolved ``_DatabricksBearerAuth`` object is cached per
+    ``server_url`` so repeated calls reuse the same SDK ``Config``
+    instance. The SDK serves the cached OAuth token from memory and only
+    re-runs the Databricks CLI when the token nears expiry, so this is
+    both fast on repeat calls and safe for long-running callers.
+
     :param server_url: The remote server URL, e.g.
         ``"https://myapp-123.aws.databricksapps.com"``.
     :returns: A bearer token, or ``None`` when no pointer record is
@@ -686,8 +700,11 @@ def _stored_databricks_record_token(server_url: str) -> str | None:
     if workspace_host is None:
         return None
     try:
-        auth, _host = _resolve_databricks_auth(host=workspace_host)
-        return auth.current_token()
+        auth = _databricks_auth_cache.get(server_url)
+        if auth is None:
+            auth, _host = _resolve_databricks_auth(host=workspace_host)
+            _databricks_auth_cache[server_url] = auth
+        return auth.current_token()  # type: ignore[union-attr]
     except (DatabricksAuthError, ImportError, ValueError):
         return None
 
@@ -1488,13 +1505,16 @@ async def _prepare_chat_session_via_daemon(
                 if fork_session_id is not None:
                     fork_result = await sdk.sessions.fork(fork_session_id)
                     session_id = fork_result["id"]
+                    fresh_session = False
                 elif resume_conversation_id is not None:
                     session_id = resume_conversation_id
+                    fresh_session = False
                 else:
                     created = await sdk.sessions.create(
                         bundle, filename="agent.tar.gz", workspace=workspace
                     )
                     session_id = created.id
+                    fresh_session = True
             except ClientOmnigentError as exc:
                 # Any create/fork/resume rejection here is a server-side answer, not
                 # a client bug worth a traceback: a wrong base URL that answers
@@ -1523,7 +1543,11 @@ async def _prepare_chat_session_via_daemon(
             if progress is not None:
                 progress.update(STARTUP_PHASE_LAUNCHING_AGENT)
             runner_id = await launch_or_reuse_daemon_runner(
-                client, host_id=host_id, session_id=session_id, workspace=workspace
+                client,
+                host_id=host_id,
+                session_id=session_id,
+                workspace=workspace,
+                fresh=fresh_session,
             )
             await wait_for_runner_online(
                 client, runner_id, timeout_s=_DAEMON_CHAT_RUNNER_ONLINE_TIMEOUT_S

omnigent/claude_native.py

@@ -3290,24 +3290,32 @@ async def _prepare_claude_terminal_via_daemon(
         # Resuming an existing session must not re-close its terminal on
         # exit; a fresh launch owns teardown.
         reattached = session_id is not None
+        fresh_session = session_id is None
         if session_id is None:
             if session_bundle is None:
                 raise click.ClickException("Creating a Claude session requires a session bundle.")
+            # Session creation (POST /v1/sessions, ~2s), daemon tunnel
+            # start (~2s), and host-online polling (~0.2s) are mutually
+            # independent — run all three concurrently so they collapse to
+            # max(session_create, daemon_start) instead of their sum.
             _mark_startup_step(
                 startup_profiler,
-                "creating daemon claude session",
+                "creating daemon claude session and waiting for host online",
                 startup_progress=startup_progress,
                 progress_message="Creating Claude session...",
             )
-            session_id = await _create_claude_session(
-                client,
-                session_bundle,
-                bridge_id=None,
-                terminal_launch_args=persist_args or None,
+            session_id, _ = await asyncio.gather(
+                _create_claude_session(
+                    client,
+                    session_bundle,
+                    bridge_id=None,
+                    terminal_launch_args=persist_args or None,
+                ),
+                wait_for_host_online(client, host_id, timeout_s=_DAEMON_HOST_ONLINE_TIMEOUT_S),
             )
             _mark_startup_step(
                 startup_profiler,
-                "daemon claude session created",
+                "daemon claude session created and host online",
                 startup_progress=startup_progress,
             )
         elif persist_args:
@@ -3329,17 +3337,30 @@ async def _prepare_claude_terminal_via_daemon(
                 "resume launch args persisted",
                 startup_progress=startup_progress,
             )
-        _mark_startup_step(
-            startup_profiler,
-            "waiting for host online",
-            startup_progress=startup_progress,
-        )
-        await wait_for_host_online(client, host_id, timeout_s=_DAEMON_HOST_ONLINE_TIMEOUT_S)
-        _mark_startup_step(
-            startup_profiler,
-            "host online",
-            startup_progress=startup_progress,
-        )
+            _mark_startup_step(
+                startup_profiler,
+                "waiting for host online",
+                startup_progress=startup_progress,
+            )
+            await wait_for_host_online(client, host_id, timeout_s=_DAEMON_HOST_ONLINE_TIMEOUT_S)
+            _mark_startup_step(
+                startup_profiler,
+                "host online",
+                startup_progress=startup_progress,
+            )
+        else:
+            # Resume with no new flags: just wait for the host.
+            _mark_startup_step(
+                startup_profiler,
+                "waiting for host online",
+                startup_progress=startup_progress,
+            )
+            await wait_for_host_online(client, host_id, timeout_s=_DAEMON_HOST_ONLINE_TIMEOUT_S)
+            _mark_startup_step(
+                startup_profiler,
+                "host online",
+                startup_progress=startup_progress,
+            )
         _mark_startup_step(
             startup_profiler,
             "launching or reusing daemon runner",
@@ -3351,6 +3372,7 @@ async def _prepare_claude_terminal_via_daemon(
             host_id=host_id,
             session_id=session_id,
             workspace=workspace,
+            fresh=fresh_session,
         )
         _mark_startup_step(
             startup_profiler,
@@ -3358,27 +3380,30 @@ async def _prepare_claude_terminal_via_daemon(
             startup_progress=startup_progress,
             detail=f"runner={runner_id}",
         )
-        _mark_startup_step(
-            startup_profiler,
-            "waiting for runner online",
-            startup_progress=startup_progress,
-            progress_message="Waiting for runner...",
-        )
-        await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)
-        _mark_startup_step(
-            startup_profiler,
-            "daemon runner online",
-            startup_progress=startup_progress,
-        )
         if reattached:
+            # Resume: runner must be online before we ask it to ensure the
+            # terminal (the POST goes to the runner via the server relay).
+            _mark_startup_step(
+                startup_profiler,
+                "waiting for runner online",
+                startup_progress=startup_progress,
+                progress_message="Waiting for runner...",
+            )
+            await wait_for_runner_online(
+                client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S
+            )
+            _mark_startup_step(
+                startup_profiler,
+                "daemon runner online",
+                startup_progress=startup_progress,
+            )
             # Resume onto an already-online daemon runner reuses it without
             # re-running the session-start auto-create, so a runner whose
             # terminal was torn down (e.g. after a ``-p`` one-shot) comes
             # back terminal-less and the wait below would time out. Ask the
             # runner to ensure the claude terminal: idempotent (returns the
             # live one if present) and otherwise auto-creates it with cold
-            # resume so history is restored. A fresh launch already creates
-            # it on session-start, so this is only needed when reattaching.
+            # resume so history is restored.
             _mark_startup_step(
                 startup_profiler,
                 "ensuring resumed terminal on runner",
@@ -3391,15 +3416,37 @@ async def _prepare_claude_terminal_via_daemon(
                 "resumed terminal ensure requested",
                 startup_progress=startup_progress,
             )
-        _mark_startup_step(
-            startup_profiler,
-            "waiting for claude terminal ready",
-            startup_progress=startup_progress,
-            progress_message="Starting Claude terminal...",
-        )
-        terminal_id = await _wait_for_claude_terminal_ready(
-            client, session_id, timeout_s=_DAEMON_TERMINAL_READY_TIMEOUT_S
-        )
+            _mark_startup_step(
+                startup_profiler,
+                "waiting for claude terminal ready",
+                startup_progress=startup_progress,
+                progress_message="Starting Claude terminal...",
+            )
+            terminal_id = await _wait_for_claude_terminal_ready(
+                client, session_id, timeout_s=_DAEMON_TERMINAL_READY_TIMEOUT_S
+            )
+        else:
+            # Fresh launch: the runner auto-creates the terminal on session-start,
+            # so runner-online and terminal-ready are sequential from the runner's
+            # side but independent from the CLI's perspective — the terminal poll
+            # returns None (404) until the runner creates it. Run both concurrently:
+            # wait_for_runner_online provides the fail-fast dead-runner signal;
+            # _wait_for_claude_terminal_ready drives to completion. The gather
+            # propagates any runner failure immediately, cancelling the terminal wait.
+            _mark_startup_step(
+                startup_profiler,
+                "waiting for runner online and claude terminal ready",
+                startup_progress=startup_progress,
+                progress_message="Starting Claude terminal...",
+            )
+            _, terminal_id = await asyncio.gather(
+                wait_for_runner_online(
+                    client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S
+                ),
+                _wait_for_claude_terminal_ready(
+                    client, session_id, timeout_s=_DAEMON_TERMINAL_READY_TIMEOUT_S
+                ),
+            )
         _mark_startup_step(
             startup_profiler,
             "claude terminal ready",
@@ -3462,7 +3509,6 @@ def _run_with_remote_server(
     :returns: None.
     """
     from omnigent.chat import _bundle_agent, _remote_headers, _server_auth
-    from omnigent.cli import _ensure_host_daemon
     from omnigent.host.identity import load_or_create_host_identity
 
     startup_profiler = startup_profiler or StartupProfiler(name="omnigent claude", enabled=False)
@@ -3514,22 +3560,6 @@ def _run_with_remote_server(
                     startup_progress=progress,
                 )
 
-            # Ensure the connect daemon is up for this server, then route the
-            # runner launch through it. The runner the daemon spawns brings
-            # up the Claude terminal itself, so the CLI just waits and
-            # attaches.
-            _mark_startup_step(
-                startup_profiler,
-                "ensuring host daemon",
-                startup_progress=progress,
-                progress_message="Connecting to local daemon...",
-            )
-            _ensure_host_daemon(base_url)
-            _mark_startup_step(
-                startup_profiler,
-                "host daemon ready",
-                startup_progress=progress,
-            )
             host_id = load_or_create_host_identity().host_id
             _mark_startup_step(
                 startup_profiler,

omnigent/cli.py

@@ -3199,10 +3199,24 @@ def _ensure_backend(server: str | None) -> str:
         # otherwise the session-create call deep in the REPL bring-up
         # surfaces the edge redirect as an opaque non-JSON-response
         # traceback.
+        #
+        # The auth probe (GET /v1/me, ~0.65s) and the daemon tunnel start
+        # (~2s) are independent — run them concurrently so the auth check
+        # is hidden under the longer daemon wait.
+        import concurrent.futures
+
         server = _resolve_server_url(server)
-        _ensure_databricks_server_auth(server)
-        with runner_startup_progress(initial_message=STARTUP_PHASE_CONNECTING_REMOTE):
-            _ensure_host_daemon(server)
+        with (
+            runner_startup_progress(initial_message=STARTUP_PHASE_CONNECTING_REMOTE),
+            concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool,
+        ):
+            auth_future = pool.submit(_ensure_databricks_server_auth, server)
+            daemon_future = pool.submit(_ensure_host_daemon, server)
+            # Raise auth errors before daemon errors: a login failure is
+            # more actionable than a daemon-connect failure that would
+            # have been caused by the same missing credentials.
+            auth_future.result()
+            daemon_future.result()
         return server
     # Local mode: the daemon spawns (or reuses) a persistent local Omnigent server.
     # On a cold start this is the longest silent gap between the user pressing

omnigent/codex_native.py

@@ -850,6 +850,7 @@ async def _prepare_codex_terminal_via_daemon(
         trust_env=not is_loopback_url(base_url),
     ) as client:
         reattached = session_id is not None
+        fresh_session = session_id is None
         if session_id is None:
             if session_bundle is None:
                 raise click.ClickException("Creating a Codex session requires a session bundle.")
@@ -918,6 +919,7 @@ async def _prepare_codex_terminal_via_daemon(
             host_id=host_id,
             session_id=session_id,
             workspace=workspace,
+            fresh=fresh_session,
         )
         _update_startup_progress(startup_progress, "Waiting for runner...")
         await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)

omnigent/cursor_native.py

@@ -510,6 +510,7 @@ async def _prepare_cursor_terminal_via_daemon(
         # running terminal below, so default both flags off here.
         reattached = False
         cold_resumed = False
+        fresh_session = session_id is None
         resume_chat_id: str | None = None
         if session_id is None:
             if session_bundle is None:
@@ -586,6 +587,7 @@ async def _prepare_cursor_terminal_via_daemon(
             host_id=host_id,
             session_id=session_id,
             workspace=workspace,
+            fresh=fresh_session,
         )
         _update_startup_progress(startup_progress, "Waiting for runner...")
         await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)

omnigent/goose_native.py

@@ -326,6 +326,7 @@ async def _prepare_goose_terminal_via_daemon(
     ) as client:
         reattached = False
         cold_resumed = False
+        fresh_session = session_id is None
         if session_id is None:
             if session_bundle is None:
                 raise click.ClickException("Creating a Goose session requires a session bundle.")
@@ -383,6 +384,7 @@ async def _prepare_goose_terminal_via_daemon(
             host_id=host_id,
             session_id=session_id,
             workspace=workspace,
+            fresh=fresh_session,
         )
         _update_startup_progress(startup_progress, "Waiting for runner...")
         await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)

omnigent/hermes_native.py

@@ -324,6 +324,7 @@ async def _prepare_hermes_terminal_via_daemon(
     ) as client:
         reattached = False
         cold_resumed = False
+        fresh_session = session_id is None
         if session_id is None:
             if session_bundle is None:
                 raise click.ClickException("Creating a Hermes session requires a session bundle.")
@@ -381,6 +382,7 @@ async def _prepare_hermes_terminal_via_daemon(
             host_id=host_id,
             session_id=session_id,
             workspace=workspace,
+            fresh=fresh_session,
         )
         _update_startup_progress(startup_progress, "Waiting for runner...")
         await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)

omnigent/host/daemon_launch.py

@@ -170,6 +170,7 @@ async def launch_or_reuse_daemon_runner(
     host_id: str,
     session_id: str,
     workspace: str,
+    fresh: bool = False,
 ) -> str:
     """
     Ensure the session is bound to a daemon-spawned runner; return its id.
@@ -184,11 +185,19 @@ async def launch_or_reuse_daemon_runner(
     :param session_id: Session to bind, e.g. ``"conv_abc123"``.
     :param workspace: Absolute host path for the runner cwd, e.g.
         ``"/Users/me/proj"``.
+    :param fresh: When ``True``, skip the ``GET /v1/sessions/{id}``
+        runner-binding check and go straight to launching a new runner.
+        Safe to set when the session was just created in this same startup
+        sequence — a brand-new session can't have a runner bound yet, so
+        the read would always return empty and only add latency (~2-3s).
     :returns: The bound runner id, e.g. ``"runner_abc123"``.
     :raises click.ClickException: If the launch request fails.
     """
-    snap = await client.get(f"/v1/sessions/{url_component(session_id)}")
-    existing = _json_body(snap).get("runner_id") if snap.status_code == 200 else None
+    if fresh:
+        existing = None
+    else:
+        snap = await client.get(f"/v1/sessions/{url_component(session_id)}")
+        existing = _json_body(snap).get("runner_id") if snap.status_code == 200 else None
     if isinstance(existing, str) and existing:
         if await runner_is_online(client, existing):
             return existing

omnigent/kimi_native.py

@@ -333,6 +333,7 @@ async def _prepare_kimi_terminal_via_daemon(
         # running terminal below, so default both flags off here.
         reattached = False
         cold_resumed = False
+        fresh_session = session_id is None
         if session_id is None:
             if session_bundle is None:
                 raise click.ClickException("Creating a Kimi session requires a session bundle.")
@@ -396,6 +397,7 @@ async def _prepare_kimi_terminal_via_daemon(
             host_id=host_id,
             session_id=session_id,
             workspace=workspace,
+            fresh=fresh_session,
         )
         _update_startup_progress(startup_progress, "Waiting for runner...")
         await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)

omnigent/kiro_native.py

@@ -364,6 +364,7 @@ async def _prepare_kiro_terminal_via_daemon(
     ) as client:
         reattached = False
         cold_resumed = False
+        fresh_session = session_id is None
         if session_id is None:
             if session_bundle is None:
                 raise click.ClickException("Creating a Kiro session requires a session bundle.")
@@ -420,6 +421,7 @@ async def _prepare_kiro_terminal_via_daemon(
             host_id=host_id,
             session_id=session_id,
             workspace=workspace,
+            fresh=fresh_session,
         )
         _update_startup_progress(startup_progress, "Waiting for runner...")
         await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)

omnigent/opencode_native.py

@@ -293,6 +293,7 @@ async def _prepare_opencode_terminal_via_daemon(  # pragma: no cover
         trust_env=not is_loopback_url(base_url),
     ) as client:
         reattached = session_id is not None
+        fresh_session = session_id is None
         if session_id is None:
             if session_bundle is None:
                 raise click.ClickException(
@@ -344,7 +345,11 @@ async def _prepare_opencode_terminal_via_daemon(  # pragma: no cover
         await wait_for_host_online(client, host_id, timeout_s=_DAEMON_HOST_ONLINE_TIMEOUT_S)
         _update_startup_progress(startup_progress, "Starting runner...")
         runner_id = await launch_or_reuse_daemon_runner(
-            client, host_id=host_id, session_id=session_id, workspace=workspace
+            client,
+            host_id=host_id,
+            session_id=session_id,
+            workspace=workspace,
+            fresh=fresh_session,
         )
         _update_startup_progress(startup_progress, "Waiting for runner...")
         await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)

omnigent/pi_native.py

@@ -377,6 +377,7 @@ async def _prepare_pi_terminal_via_daemon(
         trust_env=not is_loopback_url(base_url),
     ) as client:
         reattached = session_id is not None
+        fresh_session = session_id is None
         if session_id is None:
             if session_bundle is None:
                 raise click.ClickException("Creating a Pi session requires a session bundle.")
@@ -432,6 +433,7 @@ async def _prepare_pi_terminal_via_daemon(
             host_id=host_id,
             session_id=session_id,
             workspace=workspace,
+            fresh=fresh_session,
         )
         _update_startup_progress(startup_progress, "Waiting for runner...")
         await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)

omnigent/qwen_native.py

@@ -324,6 +324,7 @@ async def _prepare_qwen_terminal_via_daemon(
     ) as client:
         reattached = False
         cold_resumed = False
+        fresh_session = session_id is None
         if session_id is None:
             if session_bundle is None:
                 raise click.ClickException("Creating a qwen session requires a session bundle.")
@@ -381,6 +382,7 @@ async def _prepare_qwen_terminal_via_daemon(
             host_id=host_id,
             session_id=session_id,
             workspace=workspace,
+            fresh=fresh_session,
         )
         _update_startup_progress(startup_progress, "Waiting for runner...")
         await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)

tests/cli/test_chat.py

@@ -1296,7 +1296,7 @@ def _patch_daemon_launch(monkeypatch: pytest.MonkeyPatch, captured: dict[str, ob
         return None
 
     async def _fake_launch(
-        client: object, *, host_id: str, session_id: str, workspace: str
+        client: object, *, host_id: str, session_id: str, workspace: str, fresh: bool = False
     ) -> str:
         captured["launch"] = {"host_id": host_id, "session_id": session_id, "workspace": workspace}
         return "runner_daemon"

tests/test_claude_native.py

@@ -1682,6 +1682,7 @@ async def test_prepare_daemon_terminal_reports_progress_steps(
         host_id: str,
         session_id: str,
         workspace: str,
+        fresh: bool = False,
     ) -> str:
         """
         Return the runner id that production should wait on.
@@ -1784,7 +1785,6 @@ async def test_prepare_daemon_terminal_reports_progress_steps(
     assert updates == [
         "Creating Claude session...",
         "Starting runner...",
-        "Waiting for runner...",
         "Starting Claude terminal...",
         "Claude terminal ready.",
     ]

tests/test_claude_native_daemon.py

@@ -148,6 +148,37 @@ async def test_launch_or_reuse_daemon_runner_clears_stale_binding() -> None:
     assert events.index(("patch", {"runner_id": ""})) < events.index(("launch", None))
 
 
+async def test_launch_or_reuse_daemon_runner_fresh_skips_session_get() -> None:
+    """
+    ``fresh=True`` skips the ``GET /v1/sessions/{id}`` check entirely.
+
+    A session created in the same startup can't have a runner bound yet,
+    so the read is always empty and only adds latency (~2-3s). The fresh
+    path goes straight to ``POST /v1/hosts/{id}/runners``.
+    """
+    gets: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        """Record any GET to sessions and serve the launch endpoint."""
+        if request.method == "GET" and "/sessions/" in request.url.path:
+            gets.append(request.url.path)
+            return httpx.Response(200, json={})
+        if request.method == "POST" and request.url.path == "/v1/hosts/host_1/runners":
+            return httpx.Response(200, json={"runner_id": "runner_new"})
+        return httpx.Response(404, json={})
+
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(handler), base_url="https://e.com"
+    ) as client:
+        runner_id = await daemon_launch.launch_or_reuse_daemon_runner(
+            client, host_id="host_1", session_id="conv_a", workspace="/w", fresh=True
+        )
+
+    assert runner_id == "runner_new"
+    # No GET to /v1/sessions — the binding check was skipped.
+    assert gets == []
+
+
 async def test_create_claude_session_persists_terminal_launch_args() -> None:
     """
     The daemon-flow create persists pass-through args and omits the
@@ -254,13 +285,14 @@ def test_run_with_remote_server_routes_through_daemon(
     tmp_path: Path,
 ) -> None:
     """
-    A fresh remote launch ensures the daemon and hands the launch to it.
+    A fresh remote launch routes through _prepare_claude_terminal_via_daemon.
 
-    Proves the CLI no longer spawns the runner itself: it calls
-    ``_ensure_host_daemon`` for the server, then routes the launch
+    Proves the CLI no longer spawns the runner itself: it routes the launch
     through ``_prepare_claude_terminal_via_daemon`` with this host's id,
     the cwd as workspace, and the user's ``claude_args`` (so the runner
-    can apply them).
+    can apply them). Daemon start is handled by ``_ensure_backend`` in
+    ``cli_native.py`` before ``_run_with_remote_server`` is called — not
+    inside ``_run_with_remote_server`` itself.
     """
     spec_path = tmp_path / "claude.yaml"
     spec_path.write_text("name: claude-native-ui\nprompt: hi\n")
@@ -289,8 +321,8 @@ def test_run_with_remote_server_routes_through_daemon(
         claude_args=("--dangerously-skip-permissions",),
     )
 
-    # Daemon ensured for exactly this server URL.
-    assert ensured == ["https://example.com"]
+    # Daemon start is now _ensure_backend's responsibility (called from
+    # cli_native before _run_with_remote_server); not asserted here.
     # The launch was routed through the daemon prepare with this host,
     # the cwd workspace, and the user's args.
     assert captured["host_id"] == "host_1"