Compare commits

...
Author SHA1 Message Date
Nicolò Boschi cfbcc4e5a1 fix(embed): replace requests with httpx in profile_manager
- Use httpx.Client() instead of requests.get() for daemon health check
- Update test mock to use httpx.Client instead of requests.get
- Fixes ModuleNotFoundError in CI (requests not in dependencies)
2026-02-02 14:30:13 +01:00
Nicolò Boschi c7bb693d28 feat(embed): use 'default' profile name consistently
- Configure command now shows "Profile 'default' configured successfully!"
- Profile list shows "default" instead of empty string
- Profile show displays "default" consistently
- All output now uses "default" label for backward-compatible config
- Added port display for default profile in all commands
2026-02-02 14:22:40 +01:00
Nicolò Boschi 2a81395432 ci: run pytest tests for hindsight-embed in CI
- Add pytest test run step to test-embed job
- This ensures profile tests (37 tests) are run in CI
- Smoke test still runs after pytest tests
2026-02-02 14:13:34 +01:00
Nicolò Boschi df188109f3 feat(embed): add hindisght-embed profiles 2026-02-02 14:11:26 +01:00
9 changed files with 1992 additions and 137 deletions
+4
View File
@@ -782,6 +782,10 @@ jobs:
${{ runner.os }}-huggingface-embed-
${{ runner.os }}-huggingface-
- name: Run unit and integration tests
working-directory: ./hindsight-embed
run: uv run pytest tests/ -v
- name: Run smoke test
working-directory: ./hindsight-embed
run: ./test.sh
+14 -3
View File
@@ -131,17 +131,28 @@ def main():
default=DEFAULT_IDLE_TIMEOUT,
help=f"Idle timeout in seconds before auto-exit in daemon mode (default: {DEFAULT_IDLE_TIMEOUT})",
)
parser.add_argument(
"--lockfile",
type=str,
default=None,
help="Custom lockfile path for daemon mode (default: ~/.hindsight/daemon.lock)",
)
args = parser.parse_args()
# Daemon mode handling
if args.daemon:
# Use fixed daemon port
args.port = DEFAULT_DAEMON_PORT
# Use port from args (may be custom for profiles)
if args.port == config.port: # No custom port specified
args.port = DEFAULT_DAEMON_PORT
args.host = "127.0.0.1" # Only bind to localhost for security
# Check if another daemon is already running
daemon_lock = DaemonLock()
# Use custom lockfile if provided (for profile support)
from pathlib import Path
lockfile_path = Path(args.lockfile) if args.lockfile else None
daemon_lock = DaemonLock(lockfile_path) if lockfile_path else DaemonLock()
if not daemon_lock.acquire():
print(f"Daemon already running (PID: {daemon_lock.get_pid()})", file=sys.stderr)
sys.exit(1)
+107 -6
View File
@@ -25,7 +25,7 @@ uvx hindsight-embed --help
## Quick Start
```bash
# Interactive setup (recommended)
# Interactive setup (configures default profile)
hindsight-embed configure
# Or set your LLM API key manually
@@ -38,14 +38,25 @@ hindsight-embed memory retain default "User prefers dark mode"
hindsight-embed memory recall default "What are user preferences?"
```
All commands use the "default" profile unless you specify a different one with `--profile` or `HINDSIGHT_EMBED_PROFILE`.
## Commands
### configure
Interactive setup wizard:
Configure the default profile or create/update named profiles:
```bash
# Interactive setup for default profile
hindsight-embed configure
# Create/update named profile with single command
hindsight-embed configure --profile my-app \
--env HINDSIGHT_EMBED_LLM_PROVIDER=openai \
--env HINDSIGHT_EMBED_LLM_API_KEY=sk-xxx
# Create/update named profile interactively
hindsight-embed configure --profile staging
```
This will:
@@ -94,6 +105,27 @@ List all memory banks:
hindsight-embed bank list
```
### profile
Manage configuration profiles:
```bash
# List all profiles with status
hindsight-embed profile list
# Show current active profile
hindsight-embed profile show
# Set active profile (persists across commands)
hindsight-embed profile set-active my-app
# Clear active profile (revert to default)
hindsight-embed profile set-active --none
# Delete a profile
hindsight-embed profile delete my-app
```
### daemon
Manage the background daemon:
@@ -117,6 +149,7 @@ Run `hindsight-embed configure` for a guided setup that saves to `~/.hindsight/e
| Variable | Description | Default |
|----------|-------------|---------|
| `HINDSIGHT_EMBED_PROFILE` | Profile name to use (overrides active profile) | None (uses default profile) |
| `HINDSIGHT_EMBED_LLM_API_KEY` | LLM API key (or use `OPENAI_API_KEY`) | Required |
| `HINDSIGHT_EMBED_LLM_PROVIDER` | LLM provider (`openai`, `groq`, `google`, `ollama`) | `openai` |
| `HINDSIGHT_EMBED_LLM_MODEL` | LLM model | `gpt-4o-mini` |
@@ -147,14 +180,82 @@ hindsight-embed daemon start
**Note:** All banks share a single database. Bank isolation happens within the database via the `bank_id` parameter passed to CLI commands.
### Configuration Profiles
Profiles let you maintain multiple independent configurations (e.g., different API endpoints, LLM providers, or projects). Each profile runs its own daemon on a unique port (8889-9888).
**The Default Profile:**
When you run `hindsight-embed configure` without specifying a profile, it configures the "default" profile. This uses the backward-compatible configuration at `~/.hindsight/embed` and runs on port 8888.
**Creating Named Profiles:**
```bash
# Create a profile with single command
hindsight-embed configure --profile my-app \
--env HINDSIGHT_EMBED_LLM_PROVIDER=openai \
--env HINDSIGHT_EMBED_LLM_API_KEY=sk-xxx \
--env HINDSIGHT_EMBED_LLM_MODEL=gpt-4o-mini
# Create a profile interactively
hindsight-embed configure --profile staging
```
**Using Profiles:**
```bash
# Option 1: Environment variable (recommended for apps)
HINDSIGHT_EMBED_PROFILE=my-app hindsight-embed memory retain default "text"
# Option 2: CLI flag
hindsight-embed --profile my-app memory recall default "query"
# Option 3: Set as active (persists across commands)
hindsight-embed profile set-active my-app
hindsight-embed memory recall default "query" # Uses my-app profile
# Clear active profile (revert to default)
hindsight-embed profile set-active --none
```
**Profile Management:**
```bash
# List all profiles with status
hindsight-embed profile list
# Show active profile
hindsight-embed profile show
# Delete a profile
hindsight-embed profile delete my-app
```
**Profile Resolution Priority:**
1. `HINDSIGHT_EMBED_PROFILE` environment variable (highest)
2. `--profile` CLI flag
3. Active profile from `~/.hindsight/active_profile` file
4. Default profile (lowest)
**Note:** If a profile is specified but doesn't exist, the command will fail with an error. Profiles must be explicitly created using `hindsight-embed configure --profile <name>`.
### Files
**Default Profile:**
| Path | Description |
|------|-------------|
| `~/.hindsight/embed` | Configuration file |
| `~/.hindsight/config.env` | Alternative config file location |
| `~/.hindsight/daemon.log` | Daemon logs |
| `~/.hindsight/daemon.lock` | Daemon lock file (PID) |
| `~/.hindsight/embed` | Configuration file for default profile |
| `~/.hindsight/daemon.log` | Daemon logs for default profile |
| `~/.hindsight/daemon.lock` | Daemon lock file (PID) for default profile |
**Named Profiles:**
| Path | Description |
|------|-------------|
| `~/.hindsight/profiles/<name>.env` | Configuration file for profile |
| `~/.hindsight/profiles/<name>.log` | Daemon logs for profile |
| `~/.hindsight/profiles/<name>.lock` | Daemon lock file (PID) for profile |
| `~/.hindsight/profiles/metadata.json` | Profile metadata (ports, timestamps) |
| `~/.hindsight/active_profile` | Active profile name (when set with `profile set-active`) |
## Use with AI Coding Assistants
+509 -98
View File
@@ -30,10 +30,30 @@ import os
import sys
from pathlib import Path
from .profile_manager import (
ProfileManager,
resolve_active_profile,
validate_profile_exists,
)
CONFIG_DIR = Path.home() / ".hindsight"
CONFIG_FILE = CONFIG_DIR / "embed"
CONFIG_FILE_ALT = CONFIG_DIR / "config.env" # Alternative config file location
# Global profile context (set by --profile flag)
_cli_profile_override: str | None = None
def set_cli_profile_override(profile: str | None):
"""Set the CLI profile override (from --profile flag)."""
global _cli_profile_override
_cli_profile_override = profile
def get_cli_profile_override() -> str | None:
"""Get the CLI profile override."""
return _cli_profile_override
def setup_logging(verbose: bool = False):
"""Configure logging."""
@@ -57,13 +77,32 @@ def setup_logging(verbose: bool = False):
return logging.getLogger(__name__)
def load_config_file():
"""Load configuration from file if it exists."""
# Check both config file locations
config_files = [CONFIG_FILE, CONFIG_FILE_ALT]
for config_path in config_files:
if config_path.exists():
with open(config_path) as f:
def load_config_file(profile: str | None = None):
"""Load configuration from file if it exists.
Args:
profile: Profile name to load (None = resolve from priority).
"""
# Resolve profile if not specified
if profile is None:
profile = resolve_active_profile()
# Validate profile exists
validate_profile_exists(profile)
# Get config file path for profile
pm = ProfileManager()
paths = pm.resolve_profile_paths(profile)
config_path = paths.config
# For default profile, also check alternative location
config_files = [config_path]
if not profile: # Default profile
config_files.append(CONFIG_FILE_ALT)
for config_file in config_files:
if config_file.exists():
with open(config_file) as f:
for line in f:
line = line.strip()
if line and not line.startswith("#") and "=" in line:
@@ -73,11 +112,19 @@ def load_config_file():
key, value = line.split("=", 1)
if key not in os.environ: # Don't override env vars
os.environ[key] = value
break # Only load first existing config file
def get_config():
"""Get configuration from environment variables."""
load_config_file()
def get_config(profile: str | None = None):
"""Get configuration from environment variables.
Args:
profile: Profile name to load (None = resolve from priority).
Returns:
Config dict with LLM settings.
"""
load_config_file(profile)
return {
"llm_api_key": os.environ.get("HINDSIGHT_EMBED_LLM_API_KEY")
or os.environ.get("HINDSIGHT_API_LLM_API_KEY")
@@ -87,6 +134,7 @@ def get_config():
"llm_model": os.environ.get("HINDSIGHT_EMBED_LLM_MODEL")
or os.environ.get("HINDSIGHT_API_LLM_MODEL", "gpt-4o-mini"),
"bank_id": os.environ.get("HINDSIGHT_EMBED_BANK_ID", "default"),
"profile": profile or resolve_active_profile(),
}
@@ -100,7 +148,24 @@ PROVIDER_DEFAULTS = {
def do_configure(args):
"""Interactive configuration setup."""
"""Configuration setup with optional profile and env vars support.
Args:
args: Parsed arguments with optional --profile and --env flags.
"""
# Get profile and env vars from args
profile = getattr(args, "profile", None)
env_vars = getattr(args, "env", None)
# Check if we're creating a named profile with --env flags
if profile and env_vars:
return _do_configure_profile_with_env(profile, env_vars)
# Check if we're creating a named profile interactively
if profile:
return _do_configure_profile_interactive(profile)
# Default behavior: interactive configuration for default profile
# If stdin is not a terminal (e.g., running via curl | bash),
# redirect stdin from /dev/tty for interactive prompts
original_stdin = None
@@ -110,18 +175,95 @@ def do_configure(args):
sys.stdin = open("/dev/tty", "r")
except OSError:
# No terminal available - try non-interactive mode with env vars
return _do_configure_from_env()
return _do_configure_from_env(None)
try:
return _do_configure_interactive()
return _do_configure_interactive(None)
finally:
if original_stdin is not None:
sys.stdin.close()
sys.stdin = original_stdin
def _do_configure_from_env():
"""Non-interactive configuration from environment variables (for CI)."""
def _do_configure_profile_with_env(profile_name: str, env_vars: list[str]) -> int:
"""Configure a named profile with environment variables (non-interactive).
Args:
profile_name: Name of the profile to create/update.
env_vars: List of KEY=VALUE strings.
Returns:
Exit code (0 = success, 1 = error).
"""
# Parse env vars
config = {}
for env_str in env_vars:
if "=" not in env_str:
print(f"Error: Invalid --env format '{env_str}'. Expected KEY=VALUE", file=sys.stderr)
return 1
key, value = env_str.split("=", 1)
key = key.strip()
value = value.strip()
# Validate key format (should start with HINDSIGHT_EMBED_)
if not key.startswith("HINDSIGHT_EMBED_") and not key.startswith("HINDSIGHT_API_"):
print(
f"Warning: Key '{key}' doesn't start with HINDSIGHT_EMBED_ or HINDSIGHT_API_",
file=sys.stderr,
)
config[key] = value
# Create profile
pm = ProfileManager()
try:
pm.create_profile(profile_name, config)
except ValueError as e:
print(f"Error creating profile: {e}", file=sys.stderr)
return 1
print()
print(f"\033[32m✓ Profile '{profile_name}' configured successfully!\033[0m")
print()
profile_path = CONFIG_DIR / "profiles" / f"{profile_name}.env"
print(f" \033[2mConfig:\033[0m {profile_path}")
print(f" \033[2mPort:\033[0m {pm.resolve_profile_paths(profile_name).port}")
print()
print(" \033[2mUse with:\033[0m")
print(f' \033[36mHINDSIGHT_EMBED_PROFILE={profile_name} hindsight-embed memory retain default "text"\033[0m')
print(f' \033[36mhindsight-embed --profile {profile_name} memory recall default "query"\033[0m')
print()
return 0
def _do_configure_profile_interactive(profile_name: str) -> int:
"""Configure a named profile interactively.
Args:
profile_name: Name of the profile to create/update.
Returns:
Exit code (0 = success, 1 = error).
"""
print()
print(f"\033[1m\033[36m Configuring profile '{profile_name}'\033[0m")
print()
# Use the same interactive flow as default profile
return _do_configure_interactive(profile_name)
def _do_configure_from_env(profile_name: str | None = None):
"""Non-interactive configuration from environment variables (for CI).
Args:
profile_name: Optional profile name. If None, configures default profile.
Returns:
Exit code (0 = success, 1 = error).
"""
# Check for required environment variables
api_key = os.environ.get("HINDSIGHT_EMBED_LLM_API_KEY") or os.environ.get("OPENAI_API_KEY")
provider = os.environ.get("HINDSIGHT_EMBED_LLM_PROVIDER", "openai")
@@ -148,37 +290,59 @@ def _do_configure_from_env():
bank_id = os.environ.get("HINDSIGHT_EMBED_BANK_ID", "default")
print()
print("\033[1m\033[36m Hindsight Embed - Non-interactive Configuration\033[0m")
profile_label = f"profile '{profile_name}'" if profile_name else "default profile"
print(f"\033[1m\033[36m Hindsight Embed - Non-interactive Configuration ({profile_label})\033[0m")
print()
print(f" \033[2mProvider:\033[0m {provider}")
print(f" \033[2mModel:\033[0m {model}")
print(f" \033[2mBank ID:\033[0m {bank_id}")
# Build configuration
config = {
"HINDSIGHT_EMBED_LLM_PROVIDER": provider,
"HINDSIGHT_EMBED_LLM_MODEL": model,
"HINDSIGHT_EMBED_BANK_ID": bank_id,
}
if api_key:
config["HINDSIGHT_EMBED_LLM_API_KEY"] = api_key
# Force CPU mode for embeddings/reranker on macOS to avoid MPS/XPC crashes in daemon mode
import platform
if platform.system() == "Darwin": # macOS
config["HINDSIGHT_API_EMBEDDINGS_LOCAL_FORCE_CPU"] = "1"
config["HINDSIGHT_API_RERANKER_LOCAL_FORCE_CPU"] = "1"
# Save configuration
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
if profile_name:
# Named profile
pm = ProfileManager()
try:
pm.create_profile(profile_name, config)
except ValueError as e:
print(f"Error creating profile: {e}", file=sys.stderr)
return 1
config_path = CONFIG_DIR / "profiles" / f"{profile_name}.env"
else:
# Default profile
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
config_path = CONFIG_FILE
with open(CONFIG_FILE, "w") as f:
f.write("# Hindsight Embed Configuration\n")
f.write("# Generated by hindsight-embed configure (non-interactive)\n\n")
f.write(f"HINDSIGHT_EMBED_LLM_PROVIDER={provider}\n")
f.write(f"HINDSIGHT_EMBED_LLM_MODEL={model}\n")
f.write(f"HINDSIGHT_EMBED_BANK_ID={bank_id}\n")
if api_key:
f.write(f"HINDSIGHT_EMBED_LLM_API_KEY={api_key}\n")
with open(config_path, "w") as f:
f.write("# Hindsight Embed Configuration\n")
f.write("# Generated by hindsight-embed configure (non-interactive)\n\n")
for key, value in config.items():
f.write(f"{key}={value}\n")
# Force CPU mode for embeddings/reranker on macOS to avoid MPS/XPC crashes in daemon mode
# On Linux, users can set these to 0 to use CUDA if available
import platform
if platform.system() == "Darwin": # macOS
f.write("\n# Daemon settings (macOS: force CPU to avoid MPS/XPC issues)\n")
f.write("HINDSIGHT_API_EMBEDDINGS_LOCAL_FORCE_CPU=1\n")
f.write("HINDSIGHT_API_RERANKER_LOCAL_FORCE_CPU=1\n")
CONFIG_FILE.chmod(0o600)
config_path.chmod(0o600)
print()
print("\033[32m ✓ Configuration saved!\033[0m")
display_name = profile_name if profile_name else "default"
print(f"\033[32m ✓ Profile '{display_name}' configured successfully!\033[0m")
print(f" \033[2mConfig:\033[0m {config_path}")
pm = ProfileManager()
port = pm.resolve_profile_paths(profile_name or "").port
print(f" \033[2mPort:\033[0m {port}")
print()
return 0
@@ -268,16 +432,30 @@ def _prompt_confirm(prompt: str, default: bool = True) -> bool | None:
return None
def _do_configure_interactive():
"""Internal interactive configuration."""
def _do_configure_interactive(profile_name: str | None = None):
"""Internal interactive configuration.
Args:
profile_name: Optional profile name. If None, configures default profile.
Returns:
Exit code (0 = success, 1 = error).
"""
print()
display_name = profile_name if profile_name else "default"
profile_label = f" - Profile '{display_name}'"
print("\033[1m\033[36m ╭─────────────────────────────────────╮\033[0m")
print("\033[1m\033[36m │ Hindsight Embed Configuration \033[0m")
print(f"\033[1m\033[36m │ Hindsight Embed Configuration{profile_label:4s}\033[0m")
print("\033[1m\033[36m ╰─────────────────────────────────────╯\033[0m")
print()
# Check existing config
if CONFIG_FILE.exists():
if profile_name:
config_path = CONFIG_DIR / "profiles" / f"{profile_name}.env"
else:
config_path = CONFIG_FILE
if config_path.exists():
if not _prompt_confirm("Existing configuration found. Reconfigure?", default=False):
print("\n\033[32m✓\033[0m Keeping existing configuration.")
return 0
@@ -328,58 +506,87 @@ def _do_configure_interactive():
if bank_id is None:
return 1
# Save configuration
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
with open(CONFIG_FILE, "w") as f:
f.write("# Hindsight Embed Configuration\n")
f.write("# Generated by hindsight-embed configure\n\n")
f.write(f"HINDSIGHT_EMBED_LLM_PROVIDER={provider}\n")
f.write(f"HINDSIGHT_EMBED_LLM_MODEL={model}\n")
f.write(f"HINDSIGHT_EMBED_BANK_ID={bank_id}\n")
if api_key:
f.write(f"HINDSIGHT_EMBED_LLM_API_KEY={api_key}\n")
# Force CPU mode for embeddings/reranker on macOS to avoid MPS/XPC crashes in daemon mode
# On Linux, users can set these to 0 to use CUDA if available
import platform
if platform.system() == "Darwin": # macOS
f.write("\n# Daemon settings (macOS: force CPU to avoid MPS/XPC issues)\n")
f.write("HINDSIGHT_API_EMBEDDINGS_LOCAL_FORCE_CPU=1\n")
f.write("HINDSIGHT_API_RERANKER_LOCAL_FORCE_CPU=1\n")
CONFIG_FILE.chmod(0o600)
# Stop existing daemon if running (it needs to pick up new config)
from . import daemon_client
if daemon_client._is_daemon_running():
print("\n \033[2mRestarting daemon with new configuration...\033[0m")
daemon_client.stop_daemon()
# Start daemon with new config
new_config = {
"llm_api_key": api_key,
"llm_provider": provider,
"llm_model": model,
"bank_id": bank_id,
# Build configuration
config = {
"HINDSIGHT_EMBED_LLM_PROVIDER": provider,
"HINDSIGHT_EMBED_LLM_MODEL": model,
"HINDSIGHT_EMBED_BANK_ID": bank_id,
}
if daemon_client.ensure_daemon_running(new_config):
print(" \033[32m✓ Daemon started\033[0m")
if api_key:
config["HINDSIGHT_EMBED_LLM_API_KEY"] = api_key
# Force CPU mode for embeddings/reranker on macOS to avoid MPS/XPC crashes in daemon mode
import platform
if platform.system() == "Darwin": # macOS
config["HINDSIGHT_API_EMBEDDINGS_LOCAL_FORCE_CPU"] = "1"
config["HINDSIGHT_API_RERANKER_LOCAL_FORCE_CPU"] = "1"
# Save configuration
if profile_name:
# Named profile
pm = ProfileManager()
try:
pm.create_profile(profile_name, config)
except ValueError as e:
print(f"\n\033[31m✗\033[0m Error creating profile: {e}", file=sys.stderr)
return 1
config_path = CONFIG_DIR / "profiles" / f"{profile_name}.env"
else:
print(" \033[33m⚠ Failed to start daemon (will start on first command)\033[0m")
# Default profile
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
config_path = CONFIG_FILE
with open(config_path, "w") as f:
f.write("# Hindsight Embed Configuration\n")
f.write("# Generated by hindsight-embed configure\n\n")
for key, value in config.items():
f.write(f"{key}={value}\n")
config_path.chmod(0o600)
# For default profile only: stop existing daemon if running (it needs to pick up new config)
if not profile_name:
from . import daemon_client
profile = "" # Default profile
if daemon_client._is_daemon_running(profile):
print("\n \033[2mRestarting daemon with new configuration...\033[0m")
daemon_client.stop_daemon(profile)
# Start daemon with new config
new_config = {
"llm_api_key": api_key,
"llm_provider": provider,
"llm_model": model,
"bank_id": bank_id,
"profile": profile,
}
if daemon_client.ensure_daemon_running(new_config, profile):
print(" \033[32m✓ Daemon started\033[0m")
else:
print(" \033[33m⚠ Failed to start daemon (will start on first command)\033[0m")
print()
display_name = profile_name if profile_name else "default"
print("\033[32m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\033[0m")
print("\033[32m ✓ Configuration saved!\033[0m")
print(f"\033[32m ✓ Profile '{display_name}' configured successfully!\033[0m")
print("\033[32m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\033[0m")
print()
print(f" \033[2mConfig:\033[0m {CONFIG_FILE}")
print(f" \033[2mConfig:\033[0m {config_path}")
pm = ProfileManager()
port = pm.resolve_profile_paths(profile_name or "").port
print(f" \033[2mPort:\033[0m {port}")
print()
print(" \033[2mTest with:\033[0m")
print(' \033[36mhindsight-embed retain "Alice works at Google as a software engineer"\033[0m')
print(' \033[36mhindsight-embed recall "Alice"\033[0m')
if profile_name:
print(" \033[2mUse with:\033[0m")
print(f' \033[36mHINDSIGHT_EMBED_PROFILE={profile_name} hindsight-embed memory retain default "text"\033[0m')
print(f' \033[36mhindsight-embed --profile {profile_name} memory recall default "query"\033[0m')
else:
print(" \033[2mTest with:\033[0m")
print(' \033[36mhindsight-embed memory retain default "Alice works at Google as a software engineer"\033[0m')
print(' \033[36mhindsight-embed memory recall default "Alice"\033[0m')
print()
return 0
@@ -391,17 +598,23 @@ def do_daemon(args, config: dict, logger):
from . import daemon_client
daemon_log_path = Path.home() / ".hindsight" / "daemon.log"
# Get profile from config
profile = config.get("profile", "")
# Get profile-specific paths
pm = ProfileManager()
paths = pm.resolve_profile_paths(profile)
daemon_log_path = paths.log
if args.daemon_command == "start":
if daemon_client._is_daemon_running():
if daemon_client._is_daemon_running(profile):
print("Daemon is already running")
return 0
print("Starting daemon...")
if daemon_client.ensure_daemon_running(config):
if daemon_client.ensure_daemon_running(config, profile):
print("Daemon started successfully")
print(f" Port: {daemon_client.DAEMON_PORT}")
print(f" Port: {paths.port}")
print(f" Logs: {daemon_log_path}")
return 0
else:
@@ -409,12 +622,12 @@ def do_daemon(args, config: dict, logger):
return 1
elif args.daemon_command == "stop":
if not daemon_client._is_daemon_running():
if not daemon_client._is_daemon_running(profile):
print("Daemon is not running")
return 0
print("Stopping daemon...")
if daemon_client.stop_daemon():
if daemon_client.stop_daemon(profile):
print("Daemon stopped")
return 0
else:
@@ -422,9 +635,9 @@ def do_daemon(args, config: dict, logger):
return 1
elif args.daemon_command == "status":
if daemon_client._is_daemon_running():
if daemon_client._is_daemon_running(profile):
# Get PID from lockfile
lockfile = Path.home() / ".hindsight" / "daemon.lock"
lockfile = paths.lock
pid = "unknown"
if lockfile.exists():
try:
@@ -432,7 +645,7 @@ def do_daemon(args, config: dict, logger):
except Exception:
pass
print(f"Daemon is running (PID: {pid})")
print(f" URL: http://127.0.0.1:{daemon_client.DAEMON_PORT}")
print(f" URL: {daemon_client.get_daemon_url(profile)}")
print(f" Logs: {daemon_log_path}")
return 0
else:
@@ -471,17 +684,209 @@ def do_daemon(args, config: dict, logger):
return 1
def do_profile_command(args: list[str]) -> int:
"""Handle profile subcommands.
Args:
args: Command arguments (after 'profile').
Returns:
Exit code (0 = success, 1 = error).
"""
parser = argparse.ArgumentParser(prog="hindsight-embed profile")
subparsers = parser.add_subparsers(dest="profile_command", required=True)
# List command
subparsers.add_parser("list", help="List all profiles")
# Delete command
delete_parser = subparsers.add_parser("delete", help="Delete a profile")
delete_parser.add_argument("name", help="Profile name to delete")
# Set-active command
set_active_parser = subparsers.add_parser("set-active", help="Set active profile")
set_active_parser.add_argument("name", nargs="?", help="Profile name (omit to clear)")
set_active_parser.add_argument("--none", action="store_true", help="Clear active profile")
# Show command
subparsers.add_parser("show", help="Show current active profile")
try:
parsed_args = parser.parse_args(args)
except SystemExit as e:
return e.code or 1
pm = ProfileManager()
if parsed_args.profile_command == "list":
# List all profiles
profiles = pm.list_profiles()
if not profiles:
print("No profiles configured.")
print()
print("Create one with:")
print(" hindsight-embed configure --profile my-app --env HINDSIGHT_EMBED_LLM_API_KEY=...")
return 0
print()
print("\033[1mProfiles:\033[0m")
print()
for profile in profiles:
name = profile.name or "default"
active_marker = " \033[32m✓ active\033[0m" if profile.is_active else ""
daemon_marker = " \033[36m● running\033[0m" if profile.daemon_running else ""
print(f" \033[1m{name}\033[0m{active_marker}{daemon_marker}")
print(f" Port: {profile.port}")
if profile.name: # Named profile
config_path = CONFIG_DIR / "profiles" / f"{profile.name}.env"
print(f" Config: {config_path}")
else: # Default profile
config_path = CONFIG_FILE
print(f" Config: {config_path}")
print()
return 0
elif parsed_args.profile_command == "delete":
# Delete profile
profile_name = parsed_args.name
if not pm.profile_exists(profile_name):
print(f"Error: Profile '{profile_name}' does not exist.", file=sys.stderr)
return 1
# Check if daemon is running
profile_info = pm.get_profile(profile_name)
if profile_info and profile_info.daemon_running:
print(f"Warning: Daemon is running for profile '{profile_name}'")
try:
confirm = input("Stop daemon and delete profile? [y/N]: ").strip().lower()
if confirm not in ("y", "yes"):
print("Cancelled.")
return 0
except (EOFError, KeyboardInterrupt):
print("\nCancelled.")
return 0
# Stop daemon
from . import daemon_client
daemon_client.stop_daemon(profile_name)
# Delete profile
try:
pm.delete_profile(profile_name)
print(f"\033[32m✓\033[0m Profile '{profile_name}' deleted.")
return 0
except ValueError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
elif parsed_args.profile_command == "set-active":
# Set active profile
if parsed_args.none:
pm.set_active_profile(None)
print("\033[32m✓\033[0m Active profile cleared.")
return 0
if not parsed_args.name:
print("Error: Specify profile name or use --none to clear.", file=sys.stderr)
return 1
profile_name = parsed_args.name
try:
pm.set_active_profile(profile_name)
print(f"\033[32m✓\033[0m Active profile set to '{profile_name}'.")
return 0
except ValueError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
elif parsed_args.profile_command == "show":
# Show current active profile
# Resolve using full priority chain
active_profile = resolve_active_profile()
# Validate profile exists
validate_profile_exists(active_profile)
print()
display_name = active_profile if active_profile else "default"
print(f"\033[1mActive profile:\033[0m {display_name}")
print()
# Determine source
if not active_profile:
print(" \033[2mSource:\033[0m Default (no profile specified)")
elif os.getenv("HINDSIGHT_EMBED_PROFILE"):
print(" \033[2mSource:\033[0m HINDSIGHT_EMBED_PROFILE environment variable")
elif get_cli_profile_override():
print(" \033[2mSource:\033[0m --profile flag")
elif pm.get_active_profile():
print(" \033[2mSource:\033[0m Active profile file")
else:
print(" \033[2mSource:\033[0m Default")
# Show config path
paths = pm.resolve_profile_paths(active_profile)
print(f" \033[2mConfig:\033[0m {paths.config}")
print(f" \033[2mPort:\033[0m {paths.port}")
print()
return 0
return 1
def main():
"""Main entry point."""
# Parse global --profile flag first
# NOTE: For 'configure' command, we DON'T consume the --profile flag here
# because it has special meaning for that command (which profile to create)
profile_from_flag = None
remaining_args = []
i = 1
# Peek at the command to determine if we should consume --profile
command = sys.argv[1] if len(sys.argv) > 1 else None
should_consume_profile = command != "configure"
while i < len(sys.argv):
arg = sys.argv[i]
if should_consume_profile and arg == "--profile" and i + 1 < len(sys.argv):
profile_from_flag = sys.argv[i + 1]
i += 2
continue
elif should_consume_profile and arg.startswith("--profile="):
profile_from_flag = arg.split("=", 1)[1]
i += 1
continue
remaining_args.append(arg)
i += 1
# Set global profile override if --profile was provided
if profile_from_flag:
set_cli_profile_override(profile_from_flag)
# Check for built-in commands first (before argparse)
# This allows us to forward unknown commands to hindsight-cli
if len(sys.argv) > 1:
command = sys.argv[1]
if len(remaining_args) > 0:
command = remaining_args[0]
# Handle configure
if command == "configure":
# Parse configure arguments
parser = argparse.ArgumentParser(prog="hindsight-embed configure")
parser.add_argument("--profile", help="Profile name to create/update")
parser.add_argument(
"--env",
action="append",
help="Environment variable (KEY=VALUE, can be repeated)",
)
# Parse only the configure arguments (skip 'configure' command)
args = parser.parse_args(remaining_args[1:])
logger = setup_logging(False)
exit_code = do_configure(None)
exit_code = do_configure(args)
sys.exit(exit_code)
# Handle daemon subcommands
@@ -496,12 +901,17 @@ def main():
logs_parser.add_argument("--follow", "-f", action="store_true")
logs_parser.add_argument("--lines", "-n", type=int, default=50)
args = parser.parse_args(sys.argv[2:])
args = parser.parse_args(remaining_args[1:])
logger = setup_logging(False)
config = get_config()
exit_code = do_daemon(args, config, logger)
sys.exit(exit_code)
# Handle profile subcommands
if command == "profile":
exit_code = do_profile_command(remaining_args[1:])
sys.exit(exit_code)
# Handle --help / -h
if command in ("--help", "-h"):
print_help()
@@ -519,7 +929,8 @@ def main():
from . import daemon_client
# Forward to hindsight-cli (handles daemon startup and CLI installation)
exit_code = daemon_client.run_cli(sys.argv[1:], config)
profile = config.get("profile", "")
exit_code = daemon_client.run_cli(remaining_args, config, profile)
sys.exit(exit_code)
# No command - show help
+118 -29
View File
@@ -13,14 +13,48 @@ from pathlib import Path
import httpx # Used only for health check
from .profile_manager import ProfileManager, resolve_active_profile
logger = logging.getLogger(__name__)
DAEMON_PORT = 8888
DAEMON_URL = f"http://127.0.0.1:{DAEMON_PORT}"
# Default port for default profile
DEFAULT_DAEMON_PORT = 8888
DAEMON_PORT = DEFAULT_DAEMON_PORT # Backward compatibility
DAEMON_STARTUP_TIMEOUT = 180 # seconds - needs to be long for first run (downloads dependencies)
# Default idle timeout: 5 minutes - users can override with HINDSIGHT_EMBED_DAEMON_IDLE_TIMEOUT env var
DEFAULT_DAEMON_IDLE_TIMEOUT = 300
def get_daemon_port(profile: str | None = None) -> int:
"""Get daemon port for a profile.
Args:
profile: Profile name (None = resolve from priority).
Returns:
Port number for daemon.
"""
if profile is None:
profile = resolve_active_profile()
pm = ProfileManager()
paths = pm.resolve_profile_paths(profile)
return paths.port
def get_daemon_url(profile: str | None = None) -> str:
"""Get daemon URL for a profile.
Args:
profile: Profile name (None = resolve from priority).
Returns:
URL for daemon.
"""
port = get_daemon_port(profile)
return f"http://127.0.0.1:{port}"
# CLI paths - check multiple locations
CLI_INSTALL_DIRS = [
Path.home() / ".local" / "bin", # Standard location from get-cli installer
@@ -46,25 +80,46 @@ def _find_hindsight_api_command() -> list[str]:
return ["uvx", f"hindsight-api@{api_version}"]
def _is_daemon_running() -> bool:
"""Check if daemon is running and responsive."""
def _is_daemon_running(profile: str | None = None) -> bool:
"""Check if daemon is running and responsive.
Args:
profile: Profile name (None = resolve from priority).
Returns:
True if daemon is running and responsive.
"""
daemon_url = get_daemon_url(profile)
try:
with httpx.Client(timeout=2) as client:
response = client.get(f"{DAEMON_URL}/health")
response = client.get(f"{daemon_url}/health")
return response.status_code == 200
except Exception:
return False
def _start_daemon(config: dict) -> bool:
def _start_daemon(config: dict, profile: str | None = None) -> bool:
"""
Start the daemon in background.
Returns True if daemon started successfully.
Args:
config: Configuration dict with LLM settings.
profile: Profile name (None = resolve from priority).
Returns:
True if daemon started successfully.
"""
import sys
logger.info("Starting daemon...")
if profile is None:
profile = resolve_active_profile()
# Get profile-specific paths
pm = ProfileManager()
paths = pm.resolve_profile_paths(profile)
profile_label = f"profile '{profile}'" if profile else "default profile"
logger.info(f"Starting daemon for {profile_label}...")
# Build environment with LLM config
env = os.environ.copy()
@@ -98,12 +153,20 @@ def _start_daemon(config: dict) -> bool:
# Get idle timeout from environment or use default
idle_timeout = int(os.getenv("HINDSIGHT_EMBED_DAEMON_IDLE_TIMEOUT", str(DEFAULT_DAEMON_IDLE_TIMEOUT)))
cmd = _find_hindsight_api_command() + ["--daemon", "--idle-timeout", str(idle_timeout)]
# Pass profile-specific port and lockfile
cmd = _find_hindsight_api_command() + [
"--daemon",
"--idle-timeout",
str(idle_timeout),
"--port",
str(paths.port),
"--lockfile",
str(paths.lock),
]
# Create log directory
log_dir = Path.home() / ".hindsight"
log_dir.mkdir(parents=True, exist_ok=True)
daemon_log = log_dir / "daemon.log"
# Use profile-specific log file
daemon_log = paths.log
daemon_log.parent.mkdir(parents=True, exist_ok=True)
print(f"Starting daemon with command: {' '.join(cmd)}", file=sys.stderr)
print(f" Log file: {daemon_log}", file=sys.stderr)
@@ -126,13 +189,13 @@ def _start_daemon(config: dict) -> bool:
start_time = time.time()
last_check_time = start_time
while time.time() - start_time < DAEMON_STARTUP_TIMEOUT:
if _is_daemon_running():
if _is_daemon_running(profile):
# Health check passed - but daemon might crash during initialization
# Wait a moment and verify it's still healthy (stability check)
print(" Daemon responding, verifying stability...", file=sys.stderr)
time.sleep(2)
if _is_daemon_running():
logger.info("Daemon started successfully")
if _is_daemon_running(profile):
logger.info(f"Daemon started successfully for {profile_label}")
return True
else:
# Daemon crashed after initial health check
@@ -168,23 +231,45 @@ def _start_daemon(config: dict) -> bool:
return False
def ensure_daemon_running(config: dict) -> bool:
def ensure_daemon_running(config: dict, profile: str | None = None) -> bool:
"""
Ensure daemon is running, starting it if needed.
Returns True if daemon is running.
Args:
config: Configuration dict with LLM settings.
profile: Profile name (None = resolve from priority).
Returns:
True if daemon is running.
"""
if _is_daemon_running():
logger.debug("Daemon already running")
if profile is None:
profile = resolve_active_profile()
if _is_daemon_running(profile):
logger.debug(f"Daemon already running for profile '{profile or 'default'}'")
return True
return _start_daemon(config)
return _start_daemon(config, profile)
def stop_daemon() -> bool:
"""Stop the running daemon and wait for it to fully stop."""
def stop_daemon(profile: str | None = None) -> bool:
"""Stop the running daemon and wait for it to fully stop.
Args:
profile: Profile name (None = resolve from priority).
Returns:
True if daemon stopped successfully.
"""
if profile is None:
profile = resolve_active_profile()
# Get profile-specific lockfile
pm = ProfileManager()
paths = pm.resolve_profile_paths(profile)
lockfile = paths.lock
# Try to kill by PID from lockfile
lockfile = Path.home() / ".hindsight" / "daemon.lock"
if lockfile.exists():
try:
pid = int(lockfile.read_text().strip())
@@ -201,11 +286,11 @@ def stop_daemon() -> bool:
# Wait for health check to fail (daemon fully stopped)
for _ in range(30): # Wait up to 3 seconds
if not _is_daemon_running():
if not _is_daemon_running(profile):
return True
time.sleep(0.1)
return not _is_daemon_running()
return not _is_daemon_running(profile)
def find_cli_binary() -> Path | None:
@@ -294,7 +379,7 @@ def ensure_cli_installed() -> bool:
return install_cli()
def run_cli(args: list[str], config: dict) -> int:
def run_cli(args: list[str], config: dict, profile: str | None = None) -> int:
"""
Run the hindsight CLI with the given arguments.
@@ -303,6 +388,7 @@ def run_cli(args: list[str], config: dict) -> int:
Args:
args: CLI arguments (e.g., ["memory", "retain", "bank", "content"])
config: Configuration dict with llm settings
profile: Profile name (None = resolve from priority)
Returns:
Exit code from CLI
@@ -310,6 +396,9 @@ def run_cli(args: list[str], config: dict) -> int:
import subprocess
import sys
if profile is None:
profile = resolve_active_profile()
# Ensure CLI is installed
if not ensure_cli_installed():
return 1
@@ -327,10 +416,10 @@ def run_cli(args: list[str], config: dict) -> int:
if not api_url:
# No external API specified - ensure our daemon is running
if not ensure_daemon_running(config):
if not ensure_daemon_running(config, profile):
print("Error: Failed to start daemon", file=sys.stderr)
return 1
api_url = DAEMON_URL
api_url = get_daemon_url(profile)
else:
# Using external API - skip daemon startup
logger.debug(f"Using external API at {api_url}")
@@ -0,0 +1,446 @@
"""Profile management for hindsight-embed.
Handles creation, deletion, and management of configuration profiles.
Each profile has its own config, daemon lock, log file, and port.
"""
import fcntl
import hashlib
import json
import os
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import httpx
# Configuration paths
CONFIG_DIR = Path.home() / ".hindsight"
PROFILES_DIR = CONFIG_DIR / "profiles"
METADATA_FILE = PROFILES_DIR / "metadata.json"
ACTIVE_PROFILE_FILE = CONFIG_DIR / "active_profile"
# Port allocation
DEFAULT_PORT = 8888
PROFILE_PORT_BASE = 8889
PROFILE_PORT_RANGE = 1000 # 8889-9888
@dataclass
class ProfilePaths:
"""Paths and port for a profile."""
config: Path
lock: Path
log: Path
port: int
@dataclass
class ProfileInfo:
"""Profile information including metadata."""
name: str
port: int
created_at: str
last_used: Optional[str] = None
is_active: bool = False
daemon_running: bool = False
@dataclass
class ProfileMetadata:
"""Metadata for all profiles."""
version: int = 1
profiles: dict[str, dict] = field(default_factory=dict)
class ProfileManager:
"""Manages configuration profiles for hindsight-embed."""
def __init__(self):
"""Initialize the profile manager."""
self._ensure_directories()
def _ensure_directories(self):
"""Ensure profile directories exist."""
PROFILES_DIR.mkdir(parents=True, exist_ok=True)
def list_profiles(self) -> list[ProfileInfo]:
"""List all profiles with their status.
Returns:
List of ProfileInfo objects with daemon status.
"""
metadata = self._load_metadata()
active_profile = self.get_active_profile()
profiles = []
# Add default profile if config exists
default_config = CONFIG_DIR / "embed"
if default_config.exists():
profiles.append(
ProfileInfo(
name="", # Empty name = default
port=DEFAULT_PORT,
created_at="", # Don't track for default
last_used=None,
is_active=active_profile == "",
daemon_running=self._check_daemon_running(DEFAULT_PORT),
)
)
# Add named profiles
for name, info in metadata.profiles.items():
profiles.append(
ProfileInfo(
name=name,
port=info["port"],
created_at=info.get("created_at", ""),
last_used=info.get("last_used"),
is_active=active_profile == name,
daemon_running=self._check_daemon_running(info["port"]),
)
)
return sorted(profiles, key=lambda p: (p.name != "", p.name))
def profile_exists(self, name: str) -> bool:
"""Check if a profile exists.
Args:
name: Profile name (empty string for default).
Returns:
True if profile exists.
"""
if not name:
# Default profile exists if config file exists
return (CONFIG_DIR / "embed").exists()
# Named profile exists if config file exists
config_path = PROFILES_DIR / f"{name}.env"
return config_path.exists()
def get_profile(self, name: str) -> Optional[ProfileInfo]:
"""Get profile information.
Args:
name: Profile name (empty string for default).
Returns:
ProfileInfo if profile exists, None otherwise.
"""
profiles = self.list_profiles()
for profile in profiles:
if profile.name == name:
return profile
return None
def create_profile(self, name: str, config: dict[str, str]):
"""Create or update a profile.
Args:
name: Profile name.
config: Configuration dict (KEY=VALUE pairs).
Raises:
ValueError: If profile name is invalid.
"""
if not name:
raise ValueError("Profile name cannot be empty")
if not name.replace("-", "").replace("_", "").isalnum():
raise ValueError(f"Invalid profile name '{name}'. Use alphanumeric chars, hyphens, and underscores.")
# Ensure profile directory exists
self._ensure_directories()
# Load metadata to check if profile already exists
metadata = self._load_metadata()
# Allocate port for this profile (preserve existing port if updating)
if name in metadata.profiles and "port" in metadata.profiles[name]:
port = metadata.profiles[name]["port"]
else:
port = self._allocate_port(name)
# Write config file
config_path = PROFILES_DIR / f"{name}.env"
config_lines = [f"{key}={value}" for key, value in config.items()]
config_path.write_text("\n".join(config_lines) + "\n")
# Update metadata (reuse metadata loaded earlier to avoid race conditions)
now_iso = datetime.now(timezone.utc).isoformat()
if name in metadata.profiles:
# Update existing profile
metadata.profiles[name]["last_used"] = now_iso
metadata.profiles[name]["port"] = port
else:
# Create new profile
metadata.profiles[name] = {
"port": port,
"created_at": now_iso,
"last_used": now_iso,
}
self._save_metadata(metadata)
def delete_profile(self, name: str):
"""Delete a profile.
Args:
name: Profile name.
Raises:
ValueError: If profile name is invalid or doesn't exist.
"""
if not name:
raise ValueError("Cannot delete default profile")
if not self.profile_exists(name):
raise ValueError(f"Profile '{name}' does not exist")
# Remove config file
config_path = PROFILES_DIR / f"{name}.env"
if config_path.exists():
config_path.unlink()
# Remove lock file
lock_path = PROFILES_DIR / f"{name}.lock"
if lock_path.exists():
lock_path.unlink()
# Remove log file
log_path = PROFILES_DIR / f"{name}.log"
if log_path.exists():
log_path.unlink()
# Update metadata
metadata = self._load_metadata()
if name in metadata.profiles:
del metadata.profiles[name]
self._save_metadata(metadata)
# Clear active profile if it was deleted
if self.get_active_profile() == name:
self.set_active_profile(None)
def set_active_profile(self, name: Optional[str]):
"""Set the active profile.
Args:
name: Profile name to activate, or None to clear.
Raises:
ValueError: If profile doesn't exist.
"""
if name and not self.profile_exists(name):
raise ValueError(f"Profile '{name}' does not exist")
if name:
ACTIVE_PROFILE_FILE.write_text(name)
else:
# Clear active profile
if ACTIVE_PROFILE_FILE.exists():
ACTIVE_PROFILE_FILE.unlink()
def get_active_profile(self) -> str:
"""Get the currently active profile name.
Returns:
Profile name, or empty string if no active profile.
"""
if ACTIVE_PROFILE_FILE.exists():
return ACTIVE_PROFILE_FILE.read_text().strip()
return ""
def resolve_profile_paths(self, name: str) -> ProfilePaths:
"""Resolve paths for a profile.
Args:
name: Profile name (empty string for default).
Returns:
ProfilePaths with config, lock, log, and port.
"""
if not name:
# Default profile
return ProfilePaths(
config=CONFIG_DIR / "embed",
lock=CONFIG_DIR / "daemon.lock",
log=CONFIG_DIR / "daemon.log",
port=DEFAULT_PORT,
)
# Named profile
metadata = self._load_metadata()
port = metadata.profiles.get(name, {}).get("port", self._allocate_port(name))
return ProfilePaths(
config=PROFILES_DIR / f"{name}.env",
lock=PROFILES_DIR / f"{name}.lock",
log=PROFILES_DIR / f"{name}.log",
port=port,
)
def _allocate_port(self, name: str) -> int:
"""Allocate a port for a profile using hash-based strategy.
Args:
name: Profile name.
Returns:
Port number (8889-9888).
"""
# Hash profile name to get consistent port
hash_val = int(hashlib.sha256(name.encode()).hexdigest(), 16)
port = PROFILE_PORT_BASE + (hash_val % PROFILE_PORT_RANGE)
# Check if port is already allocated in metadata
metadata = self._load_metadata()
allocated_ports = {info["port"] for info in metadata.profiles.values() if info.get("port")}
# If collision, find next available port
attempt = 0
while port in allocated_ports and attempt < PROFILE_PORT_RANGE:
port = PROFILE_PORT_BASE + ((hash_val + attempt) % PROFILE_PORT_RANGE)
attempt += 1
if attempt >= PROFILE_PORT_RANGE:
# Fallback: find first available port
for p in range(PROFILE_PORT_BASE, PROFILE_PORT_BASE + PROFILE_PORT_RANGE):
if p not in allocated_ports:
return p
raise RuntimeError("No available ports for profile")
return port
def _check_daemon_running(self, port: int) -> bool:
"""Check if daemon is running on a port.
Args:
port: Port number to check.
Returns:
True if daemon is responding.
"""
try:
with httpx.Client() as client:
response = client.get(f"http://127.0.0.1:{port}/health", timeout=1)
return response.status_code == 200
except Exception:
return False
def _load_metadata(self) -> ProfileMetadata:
"""Load profile metadata from disk.
Returns:
ProfileMetadata object.
"""
if not METADATA_FILE.exists():
return ProfileMetadata()
try:
with open(METADATA_FILE) as f:
data = json.load(f)
return ProfileMetadata(version=data.get("version", 1), profiles=data.get("profiles", {}))
except (json.JSONDecodeError, IOError) as e:
print(
f"Warning: Failed to load metadata: {e}. Using empty metadata.",
file=sys.stderr,
)
# Backup corrupted metadata
backup_path = METADATA_FILE.with_suffix(".json.bak")
if METADATA_FILE.exists():
METADATA_FILE.rename(backup_path)
return ProfileMetadata()
def _save_metadata(self, metadata: ProfileMetadata):
"""Save profile metadata to disk with file locking.
Args:
metadata: ProfileMetadata to save.
"""
self._ensure_directories()
# Use atomic write with temp file
temp_file = METADATA_FILE.with_suffix(".json.tmp")
with open(temp_file, "w") as f:
# Acquire exclusive lock
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
try:
json.dump(
{"version": metadata.version, "profiles": metadata.profiles},
f,
indent=2,
)
f.flush()
os.fsync(f.fileno())
finally:
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
# Atomic rename
temp_file.rename(METADATA_FILE)
def resolve_active_profile() -> str:
"""Resolve which profile to use based on priority.
Priority (highest to lowest):
1. HINDSIGHT_EMBED_PROFILE environment variable
2. CLI --profile flag (from global context)
3. Active profile from file
4. Default (empty string)
Returns:
Profile name to use (empty string for default).
"""
# 1. Environment variable
if env_profile := os.getenv("HINDSIGHT_EMBED_PROFILE"):
return env_profile
# 2. CLI flag (set by caller before invoking commands)
from . import cli
if cli_profile := cli.get_cli_profile_override():
return cli_profile
# 3. Active profile file
pm = ProfileManager()
if active_profile := pm.get_active_profile():
return active_profile
# 4. Default
return ""
def validate_profile_exists(profile: str):
"""Validate that a profile exists, exit if not.
Args:
profile: Profile name to validate.
Exits:
If profile doesn't exist, prints error and exits.
"""
if not profile:
# Default profile - always valid
return
pm = ProfileManager()
if not pm.profile_exists(profile):
print(
f"Error: Profile '{profile}' not found.",
file=sys.stderr,
)
print(
f"Create it with: hindsight-embed configure --profile {profile}",
file=sys.stderr,
)
sys.exit(1)
+1 -1
View File
@@ -96,7 +96,7 @@ class TestRunCli:
call_args = mock_subprocess_run.call_args
# Verify environment contains the local daemon URL
assert call_args.kwargs["env"]["HINDSIGHT_API_URL"] == daemon_client.DAEMON_URL
assert call_args.kwargs["env"]["HINDSIGHT_API_URL"] == daemon_client.get_daemon_url()
# Verify exit code
assert exit_code == 0
@@ -0,0 +1,388 @@
"""Integration tests for profile functionality."""
import json
import os
import re
import subprocess
from pathlib import Path
import pytest
def strip_ansi(text):
"""Remove ANSI escape sequences from text."""
ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
return ansi_escape.sub("", text)
@pytest.fixture
def temp_home(tmp_path, monkeypatch):
"""Create a temporary home directory for integration tests."""
temp_home = tmp_path / "home"
temp_home.mkdir()
monkeypatch.setenv("HOME", str(temp_home))
return temp_home
@pytest.fixture
def hindsight_embed_cmd():
"""Get the hindsight-embed command."""
return ["uv", "run", "hindsight-embed"]
class TestProfileIntegration:
"""Integration tests for profile workflows."""
def test_create_and_list_profile(self, temp_home, hindsight_embed_cmd):
"""Test creating a profile and listing it."""
# Create profile
result = subprocess.run(
hindsight_embed_cmd
+ [
"configure",
"--profile",
"test-app",
"--env",
"HINDSIGHT_EMBED_LLM_PROVIDER=openai",
"--env",
"HINDSIGHT_EMBED_LLM_API_KEY=sk-test",
"--env",
"HINDSIGHT_EMBED_LLM_MODEL=gpt-4o-mini",
],
capture_output=True,
text=True,
env={**os.environ, "HOME": str(temp_home)},
)
assert result.returncode == 0
assert "Profile 'test-app' configured successfully" in result.stdout
# Verify profile file was created
profile_path = temp_home / ".hindsight" / "profiles" / "test-app.env"
assert profile_path.exists()
config_content = profile_path.read_text()
assert "HINDSIGHT_EMBED_LLM_PROVIDER=openai" in config_content
assert "HINDSIGHT_EMBED_LLM_API_KEY=sk-test" in config_content
# List profiles
result = subprocess.run(
hindsight_embed_cmd + ["profile", "list"],
capture_output=True,
text=True,
env={**os.environ, "HOME": str(temp_home)},
)
assert result.returncode == 0
assert "test-app" in result.stdout
assert "Port:" in result.stdout
def test_profile_show_default(self, temp_home, hindsight_embed_cmd):
"""Test showing the default profile."""
# Create default config
config_dir = temp_home / ".hindsight"
config_dir.mkdir(parents=True, exist_ok=True)
(config_dir / "embed").write_text("KEY=value")
result = subprocess.run(
hindsight_embed_cmd + ["profile", "show"],
capture_output=True,
text=True,
env={**os.environ, "HOME": str(temp_home)},
)
assert result.returncode == 0
output = strip_ansi(result.stdout)
assert "Active profile: default" in output
assert "Source: Default" in output
assert "Port: 8888" in output
def test_profile_show_with_env_var(self, temp_home, hindsight_embed_cmd):
"""Test profile resolution with HINDSIGHT_EMBED_PROFILE env var."""
# Create profile
subprocess.run(
hindsight_embed_cmd
+ [
"configure",
"--profile",
"test-app",
"--env",
"HINDSIGHT_EMBED_LLM_PROVIDER=openai",
"--env",
"HINDSIGHT_EMBED_LLM_API_KEY=sk-test",
],
capture_output=True,
env={**os.environ, "HOME": str(temp_home)},
)
# Show profile with env var
env = {**os.environ, "HOME": str(temp_home), "HINDSIGHT_EMBED_PROFILE": "test-app"}
result = subprocess.run(
hindsight_embed_cmd + ["profile", "show"], capture_output=True, text=True, env=env
)
assert result.returncode == 0
output = strip_ansi(result.stdout)
assert "Active profile: test-app" in output
assert "Source: HINDSIGHT_EMBED_PROFILE environment variable" in output
def test_set_active_profile(self, temp_home, hindsight_embed_cmd):
"""Test setting and using active profile."""
# Create profile
subprocess.run(
hindsight_embed_cmd
+ [
"configure",
"--profile",
"test-app",
"--env",
"HINDSIGHT_EMBED_LLM_PROVIDER=openai",
"--env",
"HINDSIGHT_EMBED_LLM_API_KEY=sk-test",
],
capture_output=True,
env={**os.environ, "HOME": str(temp_home)},
)
# Set as active
result = subprocess.run(
hindsight_embed_cmd + ["profile", "set-active", "test-app"],
capture_output=True,
text=True,
env={**os.environ, "HOME": str(temp_home)},
)
assert result.returncode == 0
output = strip_ansi(result.stdout)
assert "Active profile set to 'test-app'" in output
# Verify active_profile file
active_file = temp_home / ".hindsight" / "active_profile"
assert active_file.exists()
assert active_file.read_text() == "test-app"
# Show profile (should show as active)
result = subprocess.run(
hindsight_embed_cmd + ["profile", "show"],
capture_output=True,
text=True,
env={**os.environ, "HOME": str(temp_home)},
)
assert result.returncode == 0
output = strip_ansi(result.stdout)
assert "Active profile: test-app" in output
def test_delete_profile(self, temp_home, hindsight_embed_cmd):
"""Test deleting a profile."""
# Create profile
subprocess.run(
hindsight_embed_cmd
+ [
"configure",
"--profile",
"test-app",
"--env",
"HINDSIGHT_EMBED_LLM_PROVIDER=openai",
"--env",
"HINDSIGHT_EMBED_LLM_API_KEY=sk-test",
],
capture_output=True,
)
# Delete profile (with 'y' confirmation)
result = subprocess.run(
hindsight_embed_cmd + ["profile", "delete", "test-app"],
input="y\n",
capture_output=True,
text=True,
)
assert result.returncode == 0 or "deleted" in result.stdout.lower()
# Verify profile file is gone
profile_path = temp_home / ".hindsight" / "profiles" / "test-app.env"
# Note: Test may still have file if daemon was running and user cancelled
def test_multiple_profiles_different_ports(self, temp_home, hindsight_embed_cmd):
"""Test that multiple profiles get different ports."""
# Create first profile
subprocess.run(
hindsight_embed_cmd
+ [
"configure",
"--profile",
"app1",
"--env",
"HINDSIGHT_EMBED_LLM_PROVIDER=openai",
"--env",
"HINDSIGHT_EMBED_LLM_API_KEY=sk-test",
],
capture_output=True,
)
# Create second profile
subprocess.run(
hindsight_embed_cmd
+ [
"configure",
"--profile",
"app2",
"--env",
"HINDSIGHT_EMBED_LLM_PROVIDER=openai",
"--env",
"HINDSIGHT_EMBED_LLM_API_KEY=sk-test",
],
capture_output=True,
)
# Read metadata
metadata_path = temp_home / ".hindsight" / "profiles" / "metadata.json"
assert metadata_path.exists()
metadata = json.loads(metadata_path.read_text())
app1_port = metadata["profiles"]["app1"]["port"]
app2_port = metadata["profiles"]["app2"]["port"]
# Ports should be different
assert app1_port != app2_port
assert 8889 <= app1_port <= 9888
assert 8889 <= app2_port <= 9888
def test_profile_validation_fails_for_nonexistent(self, temp_home, hindsight_embed_cmd):
"""Test that using non-existent profile fails."""
# Try to use non-existent profile
env = {**os.environ, "HOME": str(temp_home), "HINDSIGHT_EMBED_PROFILE": "nonexistent"}
result = subprocess.run(
hindsight_embed_cmd + ["profile", "show"], capture_output=True, text=True, env=env
)
assert result.returncode == 1
assert "Profile 'nonexistent' not found" in result.stderr
def test_backward_compatibility_default_profile(self, temp_home, hindsight_embed_cmd):
"""Test that default profile works without any profile commands."""
# Create default config manually (simulating old behavior)
config_dir = temp_home / ".hindsight"
config_dir.mkdir(parents=True, exist_ok=True)
(config_dir / "embed").write_text(
"HINDSIGHT_EMBED_LLM_PROVIDER=openai\n"
"HINDSIGHT_EMBED_LLM_API_KEY=sk-test\n"
"HINDSIGHT_EMBED_LLM_MODEL=gpt-4o-mini\n"
)
# Show profile should work
result = subprocess.run(
hindsight_embed_cmd + ["profile", "show"],
capture_output=True,
text=True,
env={**os.environ, "HOME": str(temp_home)},
)
assert result.returncode == 0
output = strip_ansi(result.stdout)
assert "Active profile: default" in output
assert "Port: 8888" in output
def test_configure_without_profile_flag(self, temp_home, hindsight_embed_cmd):
"""Test that configure without --profile still works (backward compatibility)."""
# Configure without profile flag (should configure default)
# Use non-interactive mode by providing env vars
env = {
**os.environ,
"HOME": str(temp_home),
"HINDSIGHT_EMBED_LLM_PROVIDER": "openai",
"HINDSIGHT_EMBED_LLM_API_KEY": "sk-test",
"HINDSIGHT_EMBED_LLM_MODEL": "gpt-4o-mini",
}
result = subprocess.run(
hindsight_embed_cmd + ["configure"], capture_output=True, text=True, env=env
)
# Should succeed
assert result.returncode == 0
# Verify default config was created
config_path = temp_home / ".hindsight" / "embed"
assert config_path.exists()
def test_clear_active_profile(self, temp_home, hindsight_embed_cmd):
"""Test clearing the active profile."""
# Create and set active profile
subprocess.run(
hindsight_embed_cmd
+ [
"configure",
"--profile",
"test-app",
"--env",
"HINDSIGHT_EMBED_LLM_PROVIDER=openai",
"--env",
"HINDSIGHT_EMBED_LLM_API_KEY=sk-test",
],
capture_output=True,
)
subprocess.run(
hindsight_embed_cmd + ["profile", "set-active", "test-app"],
capture_output=True,
)
# Clear active profile
result = subprocess.run(
hindsight_embed_cmd + ["profile", "set-active", "--none"],
capture_output=True,
text=True,
)
assert result.returncode == 0
assert "Active profile cleared" in result.stdout
# Verify active_profile file is gone
active_file = temp_home / ".hindsight" / "active_profile"
assert not active_file.exists()
def test_profile_port_persistence(self, temp_home, hindsight_embed_cmd):
"""Test that profile port is persistent across recreations."""
# Create profile
subprocess.run(
hindsight_embed_cmd
+ [
"configure",
"--profile",
"test-app",
"--env",
"HINDSIGHT_EMBED_LLM_PROVIDER=openai",
"--env",
"HINDSIGHT_EMBED_LLM_API_KEY=sk-test",
],
capture_output=True,
)
# Get port
metadata_path = temp_home / ".hindsight" / "profiles" / "metadata.json"
metadata1 = json.loads(metadata_path.read_text())
port1 = metadata1["profiles"]["test-app"]["port"]
# Update profile (recreate with different config)
subprocess.run(
hindsight_embed_cmd
+ [
"configure",
"--profile",
"test-app",
"--env",
"HINDSIGHT_EMBED_LLM_PROVIDER=groq",
"--env",
"HINDSIGHT_EMBED_LLM_API_KEY=gsk-test",
],
capture_output=True,
)
# Get port again
metadata2 = json.loads(metadata_path.read_text())
port2 = metadata2["profiles"]["test-app"]["port"]
# Port should be the same
assert port1 == port2
@@ -0,0 +1,405 @@
"""Tests for profile_manager module."""
import json
import os
from pathlib import Path
from unittest.mock import Mock, patch
import pytest
from hindsight_embed.profile_manager import (
CONFIG_DIR,
PROFILES_DIR,
ProfileInfo,
ProfileManager,
ProfilePaths,
resolve_active_profile,
validate_profile_exists,
)
@pytest.fixture
def temp_hindsight_dir(tmp_path, monkeypatch):
"""Create a temporary hindsight directory for tests."""
temp_config = tmp_path / ".hindsight"
temp_config.mkdir()
monkeypatch.setattr("hindsight_embed.profile_manager.CONFIG_DIR", temp_config)
monkeypatch.setattr("hindsight_embed.profile_manager.PROFILES_DIR", temp_config / "profiles")
monkeypatch.setattr(
"hindsight_embed.profile_manager.METADATA_FILE", temp_config / "profiles" / "metadata.json"
)
monkeypatch.setattr("hindsight_embed.profile_manager.ACTIVE_PROFILE_FILE", temp_config / "active_profile")
return temp_config
@pytest.fixture
def profile_manager(temp_hindsight_dir):
"""Create a ProfileManager with temp directory."""
return ProfileManager()
class TestProfileManager:
"""Tests for ProfileManager class."""
def test_create_profile_success(self, profile_manager, temp_hindsight_dir):
"""Test creating a new profile."""
config = {
"HINDSIGHT_EMBED_LLM_PROVIDER": "openai",
"HINDSIGHT_EMBED_LLM_API_KEY": "sk-test",
"HINDSIGHT_EMBED_LLM_MODEL": "gpt-4o-mini",
}
profile_manager.create_profile("test-profile", config)
# Verify config file was created
config_path = temp_hindsight_dir / "profiles" / "test-profile.env"
assert config_path.exists()
# Verify config contents
config_content = config_path.read_text()
assert "HINDSIGHT_EMBED_LLM_PROVIDER=openai" in config_content
assert "HINDSIGHT_EMBED_LLM_API_KEY=sk-test" in config_content
assert "HINDSIGHT_EMBED_LLM_MODEL=gpt-4o-mini" in config_content
# Verify metadata was created
metadata_path = temp_hindsight_dir / "profiles" / "metadata.json"
assert metadata_path.exists()
metadata = json.loads(metadata_path.read_text())
assert "test-profile" in metadata["profiles"]
assert "port" in metadata["profiles"]["test-profile"]
assert "created_at" in metadata["profiles"]["test-profile"]
def test_create_profile_invalid_name(self, profile_manager):
"""Test creating profile with invalid name fails."""
config = {"KEY": "value"}
# Empty name
with pytest.raises(ValueError, match="Profile name cannot be empty"):
profile_manager.create_profile("", config)
# Invalid characters
with pytest.raises(ValueError, match="Invalid profile name"):
profile_manager.create_profile("test profile", config)
with pytest.raises(ValueError, match="Invalid profile name"):
profile_manager.create_profile("test@profile", config)
def test_profile_exists(self, profile_manager, temp_hindsight_dir):
"""Test checking if a profile exists."""
# Default profile doesn't exist initially
assert not profile_manager.profile_exists("")
# Create default config
(temp_hindsight_dir / "embed").write_text("KEY=value")
assert profile_manager.profile_exists("")
# Named profile doesn't exist
assert not profile_manager.profile_exists("test")
# Create named profile
profile_manager.create_profile("test", {"KEY": "value"})
assert profile_manager.profile_exists("test")
def test_delete_profile(self, profile_manager, temp_hindsight_dir):
"""Test deleting a profile."""
# Create profile
profile_manager.create_profile("test-profile", {"KEY": "value"})
assert profile_manager.profile_exists("test-profile")
# Delete profile
profile_manager.delete_profile("test-profile")
assert not profile_manager.profile_exists("test-profile")
# Verify all files are removed
profiles_dir = temp_hindsight_dir / "profiles"
assert not (profiles_dir / "test-profile.env").exists()
assert not (profiles_dir / "test-profile.lock").exists()
assert not (profiles_dir / "test-profile.log").exists()
# Verify metadata no longer contains profile
metadata_path = profiles_dir / "metadata.json"
if metadata_path.exists():
metadata = json.loads(metadata_path.read_text())
assert "test-profile" not in metadata.get("profiles", {})
def test_delete_nonexistent_profile(self, profile_manager):
"""Test deleting a non-existent profile fails."""
with pytest.raises(ValueError, match="does not exist"):
profile_manager.delete_profile("nonexistent")
def test_delete_default_profile_fails(self, profile_manager):
"""Test deleting default profile fails."""
with pytest.raises(ValueError, match="Cannot delete default profile"):
profile_manager.delete_profile("")
def test_set_active_profile(self, profile_manager, temp_hindsight_dir):
"""Test setting active profile."""
# Create a profile
profile_manager.create_profile("test-profile", {"KEY": "value"})
# Set as active
profile_manager.set_active_profile("test-profile")
# Verify active profile file was created
active_file = temp_hindsight_dir / "active_profile"
assert active_file.exists()
assert active_file.read_text() == "test-profile"
# Get active profile
assert profile_manager.get_active_profile() == "test-profile"
def test_clear_active_profile(self, profile_manager, temp_hindsight_dir):
"""Test clearing active profile."""
# Create and set active profile
profile_manager.create_profile("test-profile", {"KEY": "value"})
profile_manager.set_active_profile("test-profile")
assert profile_manager.get_active_profile() == "test-profile"
# Clear active profile
profile_manager.set_active_profile(None)
assert profile_manager.get_active_profile() == ""
# Verify file was removed
active_file = temp_hindsight_dir / "active_profile"
assert not active_file.exists()
def test_set_active_nonexistent_profile_fails(self, profile_manager):
"""Test setting non-existent profile as active fails."""
with pytest.raises(ValueError, match="does not exist"):
profile_manager.set_active_profile("nonexistent")
def test_list_profiles(self, profile_manager, temp_hindsight_dir):
"""Test listing profiles."""
# Initially no profiles
profiles = profile_manager.list_profiles()
assert len(profiles) == 0
# Create default config
(temp_hindsight_dir / "embed").write_text("KEY=value")
# List profiles - should show default
profiles = profile_manager.list_profiles()
assert len(profiles) == 1
assert profiles[0].name == ""
assert profiles[0].port == 8888
# Create named profiles
profile_manager.create_profile("profile1", {"KEY": "value"})
profile_manager.create_profile("profile2", {"KEY": "value"})
# List all profiles
profiles = profile_manager.list_profiles()
assert len(profiles) == 3
# Verify sorting (default first, then alphabetical)
assert profiles[0].name == ""
assert profiles[1].name == "profile1"
assert profiles[2].name == "profile2"
def test_list_profiles_with_active(self, profile_manager, temp_hindsight_dir):
"""Test listing profiles shows active status."""
profile_manager.create_profile("profile1", {"KEY": "value"})
profile_manager.create_profile("profile2", {"KEY": "value"})
# Set profile1 as active
profile_manager.set_active_profile("profile1")
# List profiles
profiles = profile_manager.list_profiles()
# Find profile1
profile1 = next(p for p in profiles if p.name == "profile1")
profile2 = next(p for p in profiles if p.name == "profile2")
assert profile1.is_active is True
assert profile2.is_active is False
def test_resolve_profile_paths_default(self, profile_manager, temp_hindsight_dir):
"""Test resolving paths for default profile."""
paths = profile_manager.resolve_profile_paths("")
assert paths.config == temp_hindsight_dir / "embed"
assert paths.lock == temp_hindsight_dir / "daemon.lock"
assert paths.log == temp_hindsight_dir / "daemon.log"
assert paths.port == 8888
def test_resolve_profile_paths_named(self, profile_manager, temp_hindsight_dir):
"""Test resolving paths for named profile."""
# Create profile first
profile_manager.create_profile("test-profile", {"KEY": "value"})
paths = profile_manager.resolve_profile_paths("test-profile")
assert paths.config == temp_hindsight_dir / "profiles" / "test-profile.env"
assert paths.lock == temp_hindsight_dir / "profiles" / "test-profile.lock"
assert paths.log == temp_hindsight_dir / "profiles" / "test-profile.log"
assert 8889 <= paths.port <= 9888 # Port in valid range
def test_port_allocation_deterministic(self, profile_manager):
"""Test that port allocation is deterministic for same profile name."""
profile_manager.create_profile("test1", {"KEY": "value"})
paths1 = profile_manager.resolve_profile_paths("test1")
# Delete and recreate
profile_manager.delete_profile("test1")
profile_manager.create_profile("test1", {"KEY": "value"})
paths2 = profile_manager.resolve_profile_paths("test1")
# Port should be the same
assert paths1.port == paths2.port
def test_port_allocation_unique(self, profile_manager):
"""Test that different profiles get different ports."""
profile_manager.create_profile("profile1", {"KEY": "value"})
profile_manager.create_profile("profile2", {"KEY": "value"})
paths1 = profile_manager.resolve_profile_paths("profile1")
paths2 = profile_manager.resolve_profile_paths("profile2")
# Ports should be different
assert paths1.port != paths2.port
def test_daemon_running_status(self, profile_manager):
"""Test that daemon running status is checked correctly."""
with patch("httpx.Client") as mock_client:
# Mock successful health check
mock_response = Mock()
mock_response.status_code = 200
mock_client.return_value.__enter__.return_value.get.return_value = mock_response
profile_manager.create_profile("test", {"KEY": "value"})
profiles = profile_manager.list_profiles()
test_profile = next(p for p in profiles if p.name == "test")
assert test_profile.daemon_running is True
def test_update_existing_profile(self, profile_manager, temp_hindsight_dir):
"""Test updating an existing profile."""
# Create profile
profile_manager.create_profile("test", {"KEY": "old_value"})
# Update profile
profile_manager.create_profile("test", {"KEY": "new_value"})
# Verify config was updated
config_path = temp_hindsight_dir / "profiles" / "test.env"
config_content = config_path.read_text()
assert "KEY=new_value" in config_content
assert "KEY=old_value" not in config_content
def test_metadata_persistence(self, profile_manager, temp_hindsight_dir):
"""Test that metadata persists across ProfileManager instances."""
# Create profile with first manager
profile_manager.create_profile("test", {"KEY": "value"})
# Create new manager and verify it sees the profile
new_manager = ProfileManager()
assert new_manager.profile_exists("test")
profiles = new_manager.list_profiles()
test_profile = next(p for p in profiles if p.name == "test")
assert test_profile.port > 0
def test_delete_active_profile_clears_active_file(self, profile_manager, temp_hindsight_dir):
"""Test that deleting active profile clears the active_profile file."""
profile_manager.create_profile("test", {"KEY": "value"})
profile_manager.set_active_profile("test")
# Verify active
assert profile_manager.get_active_profile() == "test"
# Delete profile
profile_manager.delete_profile("test")
# Active profile should be cleared
assert profile_manager.get_active_profile() == ""
class TestResolveActiveProfile:
"""Tests for resolve_active_profile function."""
def test_priority_env_var(self, monkeypatch, profile_manager):
"""Test that HINDSIGHT_EMBED_PROFILE env var has highest priority."""
# Set up all possible sources
monkeypatch.setenv("HINDSIGHT_EMBED_PROFILE", "from-env")
profile_manager.create_profile("from-file", {"KEY": "value"})
profile_manager.set_active_profile("from-file")
# Import cli to set override
from hindsight_embed import cli
cli.set_cli_profile_override("from-flag")
# Env var should win
assert resolve_active_profile() == "from-env"
def test_priority_cli_flag(self, monkeypatch, profile_manager):
"""Test that CLI flag has second highest priority."""
# No env var
monkeypatch.delenv("HINDSIGHT_EMBED_PROFILE", raising=False)
profile_manager.create_profile("from-file", {"KEY": "value"})
profile_manager.set_active_profile("from-file")
# Import cli to set override
from hindsight_embed import cli
cli.set_cli_profile_override("from-flag")
# CLI flag should win
assert resolve_active_profile() == "from-flag"
def test_priority_active_file(self, monkeypatch, profile_manager):
"""Test that active file has third priority."""
# No env var or CLI flag
monkeypatch.delenv("HINDSIGHT_EMBED_PROFILE", raising=False)
from hindsight_embed import cli
cli.set_cli_profile_override(None)
profile_manager.create_profile("from-file", {"KEY": "value"})
profile_manager.set_active_profile("from-file")
# Active file should be used
assert resolve_active_profile() == "from-file"
def test_priority_default(self, monkeypatch, profile_manager):
"""Test that default is used when no sources are set."""
# No env var, CLI flag, or active file
monkeypatch.delenv("HINDSIGHT_EMBED_PROFILE", raising=False)
from hindsight_embed import cli
cli.set_cli_profile_override(None)
# Default should be used
assert resolve_active_profile() == ""
class TestValidateProfileExists:
"""Tests for validate_profile_exists function."""
def test_validate_default_profile_always_passes(self):
"""Test that default profile always passes validation."""
# Default profile (empty string) should never fail
validate_profile_exists("") # Should not raise
def test_validate_existing_profile_passes(self, profile_manager):
"""Test that existing profile passes validation."""
profile_manager.create_profile("test", {"KEY": "value"})
validate_profile_exists("test") # Should not raise
def test_validate_nonexistent_profile_fails(self, profile_manager, capsys):
"""Test that non-existent profile fails validation."""
with pytest.raises(SystemExit) as exc_info:
validate_profile_exists("nonexistent")
assert exc_info.value.code == 1
# Check error message
captured = capsys.readouterr()
assert "Profile 'nonexistent' not found" in captured.err
assert "hindsight-embed configure --profile nonexistent" in captured.err