Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions src/context8/cli/commands/lifecycle.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ def start(detach: bool):
console.print(f" gRPC endpoint: [cyan]{DB_URL}[/]\n")
else:
console.print(f"\n[red]X Failed:[/] {msg}")
console.print(" Is Docker Desktop running?\n")
console.print(" Is your container runtime (Docker or Podman) running?\n")
raise SystemExit(1)


Expand All @@ -58,16 +58,16 @@ def init(seed: bool, github: bool, force: bool):
"""Initialize Context8 — start DB, create collection, download models, seed.

This is the one-stop setup command. It:
1. Starts Docker container if not running
1. Starts the DB container if not running (Docker or Podman)
2. Creates the Actian VectorAI DB collection
3. Downloads the embedding model (~80MB, cached after first run)
4. Seeds with starter data (if --seed)
5. Imports from GitHub repos (if --github)
"""
console.print("\n[bold blue]Context8[/] Initializing...\n")

# Step 1: Ensure Docker is running
console.print(" [dim]1/4[/] Docker container...", end="")
# Step 1: Ensure container runtime is running
console.print(" [dim]1/4[/] DB container...", end="")
try:
from ...docker import ensure_running

Expand All @@ -76,7 +76,7 @@ def init(seed: bool, github: bool, force: bool):
console.print(f" [green]OK[/] {msg}")
else:
console.print(f" [red]X[/] {msg}")
console.print(" Is Docker Desktop running?\n")
console.print(" Is your container runtime (Docker or Podman) running?\n")
raise SystemExit(1)
except ImportError:
console.print(" [yellow]skipped[/] (check Docker manually)")
Expand Down
30 changes: 24 additions & 6 deletions src/context8/cli/commands/ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,12 +60,30 @@ def doctor():

checks: list[tuple[str, bool, str]] = []

try:
result = subprocess.run(["docker", "info"], capture_output=True, text=True, timeout=5)
docker_ok = result.returncode == 0
checks.append(("Docker", docker_ok, "running" if docker_ok else "not running"))
except (FileNotFoundError, subprocess.TimeoutExpired):
checks.append(("Docker", False, "not found — install Docker Desktop"))
from ...docker import detect_runtime

runtime = detect_runtime()
if runtime is None:
checks.append(
(
"Container runtime",
False,
"not found — install Docker or Podman",
)
)
else:
try:
result = subprocess.run([runtime, "info"], capture_output=True, text=True, timeout=5)
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.

Source: opengrep

running = result.returncode == 0
checks.append(
(
f"Container runtime ({runtime})",
running,
"running" if running else "installed but daemon not reachable",
)
)
except (FileNotFoundError, subprocess.TimeoutExpired):
checks.append((f"Container runtime ({runtime})", False, "probe failed"))

try:
from ...docker import is_container_running
Expand Down
54 changes: 52 additions & 2 deletions src/context8/cli/commands/serve.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,63 @@
from __future__ import annotations

import asyncio
import sys

import click


def _log(msg: str) -> None:
"""Bootstrap logging — stderr only, never stdout (stdio MCP uses stdout)."""
print(f"[context8] {msg}", file=sys.stderr, flush=True)


def _bootstrap() -> None:
"""Idempotent bootstrap: container up, collection ready, models cached.

Safe to run on every `serve` invocation — each step is a no-op when already
satisfied. All output goes to stderr so the MCP stdio protocol stays clean.
"""
from ...docker import ensure_running, is_container_running

if not is_container_running():
_log("starting DB container...")
ok, msg = ensure_running(timeout_secs=30)
if not ok:
_log(f"FATAL: container failed to start: {msg}")
raise SystemExit(1)
_log(f"container ready ({msg})")

Comment on lines +22 to +29
Copy link

Copilot AI Apr 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_bootstrap() only calls ensure_running() when is_container_running() is false. If the container exists but the DB is still starting (or unhealthy), the bootstrap will skip the health-check/wait and may fail immediately when StorageService() connects. Consider calling ensure_running() unconditionally here (and letting it be idempotent) or otherwise explicitly waiting for VectorAIClient.health_check() even when the container is already "Up".

Suggested change
if not is_container_running():
_log("starting DB container...")
ok, msg = ensure_running(timeout_secs=30)
if not ok:
_log(f"FATAL: container failed to start: {msg}")
raise SystemExit(1)
_log(f"container ready ({msg})")
was_running = is_container_running()
if was_running:
_log("verifying DB container readiness...")
else:
_log("starting DB container...")
ok, msg = ensure_running(timeout_secs=30)
if not ok:
_log(f"FATAL: container failed to start: {msg}")
raise SystemExit(1)
_log(f"container ready ({msg})")

Copilot uses AI. Check for mistakes.
from ...storage import StorageService

storage = StorageService()
created = storage.initialize()
if created:
_log("collection created")
storage.close()
Comment on lines +30 to +36
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: Consider handling storage initialization failures to produce a clearer fatal message.

If StorageService() or storage.initialize() fails (e.g. DB unreachable, bad config), the exception will currently bubble up and crash without the clear stderr message you use for the container bootstrap. Consider wrapping storage initialization in a try/except, logging a FATAL via _log, and exiting with SystemExit(1) so failures are reported consistently and predictably.

Suggested change
from ...storage import StorageService
storage = StorageService()
created = storage.initialize()
if created:
_log("collection created")
storage.close()
from ...storage import StorageService
storage = None
try:
storage = StorageService()
created = storage.initialize()
if created:
_log("collection created")
except Exception as e:
_log(f"FATAL: storage initialization failed: {e}")
raise SystemExit(1)
finally:
if storage is not None:
try:
storage.close()
except Exception as e:
_log(f"warning: failed to close storage cleanly ({e})")

Comment on lines +33 to +36
Copy link

Copilot AI Apr 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

storage.close() won’t run if storage.initialize() raises (e.g., DB connection issue). Wrap the StorageService lifecycle in a try/finally (or context manager) so the client connection is always closed during bootstrap failures.

Suggested change
created = storage.initialize()
if created:
_log("collection created")
storage.close()
try:
created = storage.initialize()
if created:
_log("collection created")
finally:
storage.close()

Copilot uses AI. Check for mistakes.

try:
from ...embeddings import EmbeddingService

EmbeddingService.ensure_models_downloaded()
except Exception as e:
_log(f"warning: model pre-download failed ({e}) — will lazy-load")


@click.command()
def serve():
"""Start the Context8 MCP server (stdio transport)."""
@click.option(
"--no-bootstrap",
is_flag=True,
help="Skip auto-start of container/collection (assume already initialized)",
)
def serve(no_bootstrap: bool):
"""Start the Context8 MCP server (stdio transport).

By default, ensures the DB container is up and the collection exists before
starting the server, so a single `context8 serve` works from a cold machine.
"""
if not no_bootstrap:
_bootstrap()

from ...mcp import run_server

asyncio.run(run_server())
15 changes: 14 additions & 1 deletion src/context8/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,20 @@ def continue_config_path() -> Path:


def get_server_command() -> list[str]:
return ["python", "-m", "context8.mcp.server"]
"""Absolute command Claude Code (or any agent) should launch for the MCP server.

Routes through the `serve` CLI so the auto-bootstrap (container, collection,
models) runs before the stdio loop starts. Prefers the installed `context8`
script on PATH; falls back to the current interpreter so the plugin works
even when the entry-point script isn't globally available.
"""
import shutil
import sys

script = shutil.which("context8")
if script:
return [script, "serve"]
return [sys.executable, "-m", "context8", "serve"]


SUPPORTED_AGENTS = {
Expand Down
118 changes: 96 additions & 22 deletions src/context8/docker.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
"""Docker management for Context8 — start/stop/check the Actian VectorAI DB container.
"""Container runtime management for Context8 — start/stop/check the Actian
VectorAI DB container.

Generates docker-compose.yml on demand into ~/.context8/ so it works
whether installed via pip, uv, or from source. Never requires the user
to have a compose file in their project.
Generates a compose file on demand into ~/.context8/ so it works whether
installed via pip, uv, or from source. Supports both Docker and Podman —
the runtime is detected automatically.
"""

from __future__ import annotations
Expand All @@ -19,7 +20,6 @@
CONTEXT8_DIR = _home() / ".context8"

COMPOSE_TEMPLATE = """\
version: "3.8"
services:
vectoraidb:
image: docker.io/williamimoh/actian-vectorai-db:latest
Expand All @@ -34,6 +34,9 @@

CONTAINER_NAME = "context8_db"

_runtime_cache: str | None = None
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (complexity): Consider simplifying runtime and compose detection by replacing the multiple helper functions and global caches with a single linear resolver that returns both the compose command and runtime.

You can drop most of the indirection by collapsing runtime/compose detection into a single linear resolver and removing the global caches/sentinels. That keeps Docker/Podman support but makes the control flow much easier to follow.

For example, you can replace _runtime_cache, _compose_cache, _probe, detect_runtime and _compose_cmd with something like:

def _iter_compose_candidates() -> list[tuple[list[str], str]]:
    # command, inferred runtime
    return [
        (["docker", "compose"], "docker"),
        (["docker-compose"], "docker"),
        (["podman", "compose"], "podman"),
        (["podman-compose"], "podman"),
    ]


def _resolve_compose() -> tuple[list[str], str] | None:
    """Return (compose_cmd, runtime) or None if nothing is usable."""
    for cmd, runtime in _iter_compose_candidates():
        try:
            subprocess.run(
                cmd + ["version"],
                capture_output=True,
                check=True,
                timeout=5,
            )
            return cmd, runtime
        except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired):
            continue
    return None

Then run_compose, is_container_running, and ensure_running become simpler and don’t need to know about probing or caches:

def run_compose(args: list[str]) -> subprocess.CompletedProcess:
    """Run a compose command against the Context8 compose file."""
    _ensure_compose_file()
    resolved = _resolve_compose()
    if resolved is None:
        return subprocess.CompletedProcess(
            args=args,
            returncode=1,
            stdout="",
            stderr=(
                "no compose tool found — install one of: "
                "`docker compose`, `podman compose`, `podman-compose`, `docker-compose`"
            ),
        )

    cmd, _runtime = resolved
    return subprocess.run(
        cmd + args,
        cwd=str(_compose_dir()),
        capture_output=True,
        text=True,
    )
def is_container_running() -> bool:
    """Check if the context8_db container is running under docker or podman."""
    resolved = _resolve_compose()
    if resolved is None:
        return False

    _cmd, runtime = resolved
    try:
        result = subprocess.run(
            [runtime, "ps", "--filter", f"name={CONTAINER_NAME}", "--format", "{{.Status}}"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        return bool(result.stdout.strip()) and "Up" in result.stdout
    except Exception:
        return False
def ensure_running(timeout_secs: int = 30) -> tuple[bool, str]:
    """Start the container if not running, wait for it to be healthy."""
    if is_container_running():
        return True, "already running"

    resolved = _resolve_compose()
    runtime = resolved[1] if resolved is not None else "container runtime"
    logger.info(f"Starting Actian VectorAI DB container via {runtime}...")

    result = run_compose(["up", "-d"])
    if result.returncode != 0:
        return False, f"compose up failed: {result.stderr.strip()}"
    ...

This keeps all existing functionality (Docker + Podman, the specific error message and CompletedProcess return from run_compose) but removes:

  • global _runtime_cache / _compose_cache and the ""/[] sentinels
  • the two-phase info/--version probing split across detect_runtime and _compose_cmd
  • the generic _probe() helper used in subtly different ways

The behavior becomes: “try a small ordered set of compose commands, pick the first working one, infer runtime from that”, which matches the reviewer’s suggested mental model and is easier to reason about and test.

_compose_cache: list[str] | None = None


def _compose_dir() -> Path:
"""Return ~/.context8/ — where we keep the generated compose file."""
Expand All @@ -55,36 +58,106 @@ def _ensure_compose_file() -> Path:
return compose_path


def _docker_compose_cmd() -> list[str]:
"""Return the docker compose command (v2 first, v1 fallback)."""
def _probe(cmd: list[str]) -> bool:
try:
subprocess.run(
["docker", "compose", "version"],
capture_output=True,
check=True,
)
return ["docker", "compose"]
except (subprocess.CalledProcessError, FileNotFoundError):
return ["docker-compose"]
subprocess.run(cmd, capture_output=True, check=True, timeout=5)
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.

Source: opengrep

return True
except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired):
return False


def detect_runtime() -> str | None:
"""Return 'docker' or 'podman' — the first runtime whose daemon is reachable.

Prefers a *working* runtime over a merely installed one: if docker is
installed but its daemon is down, falls through to podman. Returns None
if neither is usable.
"""
global _runtime_cache
if _runtime_cache is not None:
return _runtime_cache or None

# First pass: pick a runtime whose daemon is actually reachable.
for candidate in ("docker", "podman"):
if _probe([candidate, "info"]):
_runtime_cache = candidate
return candidate

# Second pass: any installed runtime, even if the daemon is down — so
# callers can still surface a useful error message.
for candidate in ("docker", "podman"):
if _probe([candidate, "--version"]):
_runtime_cache = candidate
return candidate

_runtime_cache = ""
return None


def _compose_cmd() -> list[str] | None:
"""Return the compose command for the detected runtime, or None.

Tries (in order): `docker compose`, `docker-compose`, `podman compose`,
`podman-compose`.
"""
global _compose_cache
if _compose_cache is not None:
return _compose_cache or None

candidates: list[list[str]] = []
runtime = detect_runtime()
if runtime == "docker":
candidates = [["docker", "compose"], ["docker-compose"]]
elif runtime == "podman":
candidates = [["podman", "compose"], ["podman-compose"]]
else:
# Try everything anyway in case the runtime probe missed something
candidates = [
["docker", "compose"],
["podman", "compose"],
["podman-compose"],
["docker-compose"],
]

for cmd in candidates:
if _probe(cmd + ["version"]):
_compose_cache = cmd
return cmd

_compose_cache = []
return None


def run_compose(args: list[str]) -> subprocess.CompletedProcess:
"""Run a docker compose command against the Context8 compose file."""
"""Run a compose command against the Context8 compose file."""
_ensure_compose_file()
cmd = _docker_compose_cmd() + args
cmd = _compose_cmd()
if cmd is None:
return subprocess.CompletedProcess(
args=args,
returncode=1,
stdout="",
stderr=(
"no compose tool found — install one of: "
"`docker compose`, `podman compose`, `podman-compose`, `docker-compose`"
),
)
Comment on lines +136 to +144
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'CompletedProcess' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.

Source: opengrep

return subprocess.run(
cmd,
cmd + args,
cwd=str(_compose_dir()),
capture_output=True,
text=True,
Comment on lines 145 to 149
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.

Source: opengrep

)


def is_container_running() -> bool:
"""Check if the context8_db container is running."""
"""Check if the context8_db container is running under docker or podman."""
runtime = detect_runtime()
if runtime is None:
return False
try:
result = subprocess.run(
["docker", "ps", "--filter", f"name={CONTAINER_NAME}", "--format", "{{{{.Status}}}}"],
[runtime, "ps", "--filter", f"name={CONTAINER_NAME}", "--format", "{{.Status}}"],
capture_output=True,
text=True,
timeout=5,
Comment on lines 159 to 163
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.

Source: opengrep

Expand All @@ -102,11 +175,12 @@ def ensure_running(timeout_secs: int = 30) -> tuple[bool, str]:
if is_container_running():
return True, "already running"
Comment on lines 175 to 176
Copy link

Copilot AI Apr 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ensure_running() returns (True, "already running") purely based on docker/podman ps output, without verifying the DB is actually accepting connections. Since callers use this as a readiness gate, consider still running the VectorAIClient.health_check() loop when the container is already running (and only return success once the DB responds).

Copilot uses AI. Check for mistakes.

logger.info("Starting Actian VectorAI DB container...")
runtime = detect_runtime() or "container runtime"
logger.info(f"Starting Actian VectorAI DB container via {runtime}...")
result = run_compose(["up", "-d"])

if result.returncode != 0:
return False, f"docker compose up failed: {result.stderr.strip()}"
return False, f"compose up failed: {result.stderr.strip()}"

# Wait for the DB to accept connections
for _ in range(timeout_secs):
Expand Down
Loading
Loading