lilith-daemon-registry

A lightweight Python library for managing multiple daemon instances with persistent registry, port allocation, and process tracking.

Features

• Multi-instance tracking: Register and track multiple daemon processes
• Port allocation: Automatic assignment of available ports in configurable ranges
• Process existence checking: Verify daemon processes are still running
• Stale daemon cleanup: Automatically remove dead process entries
• Directory-based lookup: Find daemons by monitored directory (exact or parent match)
• JSON persistence: Registry stored in human-readable JSON format
• Metadata storage: Store arbitrary daemon-specific configuration
• Type-safe: Full type hints and mypy strict mode compliance

Installation

pip install lilith-daemon-registry

Quick Start

from lilith_daemon_registry import DaemonRegistry, DaemonInfo, assign_port
from datetime import datetime
import uuid
import os

# Load or create registry (default: ~/.config/daemons/registry.json)
registry = DaemonRegistry.load()

# Assign an available port (default range: 8200-8299)
port = assign_port(registry)

# Create daemon info
daemon = DaemonInfo(
    id=str(uuid.uuid4())[:8],
    directory="/home/user/projects/myapp",
    pid=os.getpid(),
    port=port,
    started_at=datetime.now().isoformat(),
    last_seen=datetime.now().isoformat(),
    metadata={
        "interval_seconds": 60,
        "recursive": True,
        "config_version": "1.0",
    }
)

# Register the daemon
registry.register(daemon)

# Find daemon by directory
found = registry.find_by_directory("/home/user/projects/myapp")
if found:
    print(f"Daemon running on port {found.port}")

# Cleanup stale daemons
removed = registry.cleanup_stale()
print(f"Removed {removed} stale daemon(s)")

# Unregister when done
registry.unregister("/home/user/projects/myapp")

API Reference

DaemonInfo

Data model representing a daemon instance.

@dataclass
class DaemonInfo:
    id: str                      # Unique identifier
    directory: str               # Monitored directory
    pid: int                     # Process ID
    port: int                    # HTTP port
    started_at: str              # ISO timestamp
    last_seen: str               # ISO timestamp
    metadata: dict[str, Any]     # Arbitrary metadata

Methods:

• to_dict() -> dict[str, Any]: Serialize to dictionary
• from_dict(data: dict[str, Any]) -> DaemonInfo: Deserialize from dictionary

DaemonRegistry

Registry for managing daemon instances.

@dataclass
class DaemonRegistry:
    daemons: list[DaemonInfo]
    registry_path: Path

Methods:

Class Methods

• load(registry_path: Path | None = None) -> DaemonRegistry
  • Load registry from disk (default: ~/.config/daemons/registry.json)

Instance Methods

• save() -> None

  • Persist registry to disk

• register(daemon: DaemonInfo) -> None

  • Register a new daemon (auto-cleans stale entries)

• unregister(directory: str) -> bool

  • Unregister daemon by directory

• unregister_by_id(daemon_id: str) -> bool

  • Unregister daemon by ID

• unregister_by_pid(pid: int) -> bool

  • Unregister daemon by process ID

• cleanup_stale() -> int

  • Remove daemons with dead processes, returns count removed

• find_by_id(daemon_id: str) -> DaemonInfo | None

  • Find daemon by ID

• find_by_directory(directory: str) -> DaemonInfo | None

  • Find daemon by exact directory match

• find_by_directory_or_parent(directory: str) -> DaemonInfo | None

  • Find daemon monitoring directory or parent directory

• find_by_port(port: int) -> DaemonInfo | None

  • Find daemon by port

• find_by_pid(pid: int) -> DaemonInfo | None

  • Find daemon by process ID

• get_used_ports() -> set[int]

  • Get set of ports currently in use

• get_all_directories() -> list[str]

  • Get list of all monitored directories

Port Utilities

def assign_port(
    registry: DaemonRegistry,
    port_range_start: int = 8200,
    port_range_end: int = 8300,
    host: str = "localhost",
) -> int

Find an available port in the specified range.

def port_in_use(port: int, host: str = "localhost") -> bool

Check if a port is currently in use.

def process_exists(pid: int) -> bool

Check if a process with given PID exists.

Usage Examples

Custom Registry Path

from pathlib import Path
from lilith_daemon_registry import DaemonRegistry

# Use custom registry location
registry_path = Path("/var/lib/myapp/daemons.json")
registry = DaemonRegistry.load(registry_path)

Custom Port Range

from lilith_daemon_registry import assign_port

# Use custom port range (9000-9099)
port = assign_port(
    registry,
    port_range_start=9000,
    port_range_end=9100
)

Directory Hierarchy Monitoring

# Register daemon monitoring parent directory
parent_daemon = DaemonInfo(
    id="parent",
    directory="/home/user/projects",
    pid=12345,
    port=8200,
    started_at=datetime.now().isoformat(),
    last_seen=datetime.now().isoformat(),
)
registry.register(parent_daemon)

# Check if subdirectory is already monitored
found = registry.find_by_directory_or_parent("/home/user/projects/subdir")
if found:
    print(f"Subdirectory already monitored by daemon {found.id}")

Daemon Metadata

# Store daemon-specific configuration in metadata
daemon = DaemonInfo(
    id="worker1",
    directory="/var/www/app",
    pid=os.getpid(),
    port=8200,
    started_at=datetime.now().isoformat(),
    last_seen=datetime.now().isoformat(),
    metadata={
        "interval_seconds": 30,
        "max_retries": 3,
        "log_level": "INFO",
        "features": ["auto-commit", "cache-update"],
    }
)
registry.register(daemon)

Periodic Cleanup

import time

# Periodically cleanup stale daemons
while True:
    removed = registry.cleanup_stale()
    if removed > 0:
        print(f"Cleaned up {removed} stale daemon(s)")
    time.sleep(60)

Listing All Daemons

# Get all registered daemons
for daemon in registry.daemons:
    print(f"Daemon {daemon.id}:")
    print(f"  Directory: {daemon.directory}")
    print(f"  Port: {daemon.port}")
    print(f"  PID: {daemon.pid}")
    print(f"  Started: {daemon.started_at}")

Registry File Format

The registry is stored as JSON with the following structure:

{
  "daemons": [
    {
      "id": "abc123",
      "directory": "/home/user/projects/myapp",
      "pid": 12345,
      "port": 8200,
      "started_at": "2026-01-20T10:00:00",
      "last_seen": "2026-01-20T10:05:00",
      "metadata": {
        "interval_seconds": 60,
        "recursive": true
      }
    }
  ]
}

Development

Setup

# Install dev dependencies
pip install -e ".[dev]"

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=lilith_daemon_registry --cov-report=html

# Run specific test file
pytest tests/test_registry.py

Type Checking

mypy src/lilith_daemon_registry

Linting

ruff check src/ tests/
ruff format src/ tests/

License

MIT License

Author

Lilith noreply@lilith.dev