package daemon
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"os"
"os/exec"
"path/filepath"
"strings"
"time"
)
const (
pollInterval = 10 * time.Millisecond
startupTimeout = 30 * time.Second
)
type DiscoveryKind int
const (
Absent DiscoveryKind = iota
Stale
Running
)
type Discovery struct {
Kind DiscoveryKind
Snapshot ConnectionSnapshot
Health Health
}
type LocalOptions struct {
HomeDir string
ProjectRoot string
}
type LocalErrorKind string
const (
InvalidDiscovery LocalErrorKind = "invalid_discovery"
DiscoveryUnavailable LocalErrorKind = "discovery_unavailable"
UnreachableDaemon LocalErrorKind = "unreachable_daemon"
UnhealthyDaemon LocalErrorKind = "unhealthy_daemon"
AuthenticationFailed LocalErrorKind = "authentication_failed"
TargetChanged LocalErrorKind = "target_changed"
LauncherLockFailure LocalErrorKind = "launcher_lock_failure"
StartupFailed LocalErrorKind = "startup_failed"
HomeInUse LocalErrorKind = "home_in_use"
)
type LocalError struct {
Kind LocalErrorKind
HomeDir string
Cause error
}
func (e *LocalError) Error() string {
return fmt.Sprintf("Albedo %s in %s: %v", strings.ReplaceAll(string(e.Kind), "_", " "), e.HomeDir, e.Cause)
}
func (e *LocalError) Unwrap() error { return e.Cause }
// Discover distinguishes missing and proven stale records from unsafe failures.
// A valid live endpoint is returned even when its API contract needs upgrading.
func Discover(ctx context.Context, homeDir string) (Discovery, error) {
if err := ctx.Err(); err != nil {
return Discovery{}, err
}
fail := func(kind LocalErrorKind, cause error) (Discovery, error) {
return Discovery{}, &LocalError{Kind: kind, HomeDir: homeDir, Cause: cause}
}
file, err := os.Open(filepath.Join(homeDir, "daemon.json"))
if errors.Is(err, os.ErrNotExist) {
return Discovery{Kind: Absent}, nil
}
if err != nil {
return fail(DiscoveryUnavailable, err)
}
defer file.Close()
data, err := io.ReadAll(io.LimitReader(file, 64*1024+1))
if err != nil {
return fail(DiscoveryUnavailable, err)
}
if len(data) > 64*1024 {
return fail(InvalidDiscovery, errors.New("daemon.json exceeds 64 KiB"))
}
var snapshot ConnectionSnapshot
if err := json.Unmarshal(data, &snapshot); err != nil {
return fail(InvalidDiscovery, err)
}
if snapshot.Pid <= 0 || snapshot.Port < 1 || snapshot.Port > 65535 || snapshot.Token == "" || snapshot.Version <= 0 {
return fail(InvalidDiscovery, errors.New("daemon.json requires a positive PID, valid port, token and protocol version"))
}
probeCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
connection := NewConnection(snapshot, nil)
health, err := ProbeHealth(probeCtx, connection)
connection.HTTPClient().CloseIdleConnections()
cancel()
if err != nil {
if ctx.Err() != nil {
return Discovery{}, ctx.Err()
}
if !processAlive(snapshot.Pid) && endpointRefused(err) {
return Discovery{Kind: Stale, Snapshot: snapshot}, nil
}
var api *APIError
if errors.As(err, &api) {
if api.StatusCode == http.StatusUnauthorized || api.StatusCode == http.StatusForbidden {
return fail(AuthenticationFailed, err)
}
return fail(UnhealthyDaemon, err)
}
var protocol *ProtocolError
if errors.As(err, &protocol) {
return fail(UnhealthyDaemon, err)
}
return fail(UnreachableDaemon, err)
}
if health.Version != snapshot.Version || (health.Build != "" && health.Build != snapshot.Build) {
return fail(InvalidDiscovery, errors.New("health identity does not match daemon.json"))
}
discovery := Discovery{Kind: Running, Snapshot: snapshot, Health: health}
return discovery, CheckCompatible(health)
}
// Rediscover waits for a verified replacement endpoint without starting one.
func Rediscover(ctx context.Context, homeDir string) (ConnectionSnapshot, error) {
for attempt := 0; attempt < 20; attempt++ {
discovery, err := Discover(ctx, homeDir)
if err == nil && discovery.Kind == Running {
return discovery.Snapshot, nil
}
if err != nil {
var local *LocalError
if !errors.As(err, &local) || (local.Kind != UnreachableDaemon && local.Kind != UnhealthyDaemon) {
return ConnectionSnapshot{}, err
}
var api *APIError
if errors.As(err, &api) && api.Code != "daemon_stopping" && api.Code != "daemon_unavailable" {
return ConnectionSnapshot{}, err
}
var protocol *ProtocolError
if errors.As(err, &protocol) {
return ConnectionSnapshot{}, err
}
}
timer := time.NewTimer(100 * time.Millisecond)
select {
case <-ctx.Done():
timer.Stop()
return ConnectionSnapshot{}, ctx.Err()
case <-timer.C:
}
}
return ConnectionSnapshot{}, &LocalError{Kind: UnreachableDaemon, HomeDir: homeDir, Cause: errors.New("could not rediscover a ready daemon; check whether it is running")}
}
func resolveDaemonExecutable(path string) (string, error) {
if !filepath.IsAbs(path) {
return "", fmt.Errorf("ALBEDO_DAEMON must point to an executable using its full path: %q", path)
}
resolved, err := exec.LookPath(path)
if err != nil {
return "", fmt.Errorf("cannot run ALBEDO_DAEMON (%q): %w", path, err)
}
info, err := os.Stat(resolved)
if err != nil {
return "", err
}
if info.IsDir() {
return "", fmt.Errorf("ALBEDO_DAEMON points to a directory (%q)", path)
}
return resolved, nil
}
// Local launch filters provider overrides; daemon bootstrap owns all defaults.
func daemonCommand(options LocalOptions) (*exec.Cmd, error) {
executable := os.Getenv("ALBEDO_DAEMON")
if executable == "" {
if options.ProjectRoot == "" {
return nil, errors.New("set ALBEDO_DAEMON to its full executable path or ALBEDO_ROOT to a source checkout")
}
root, err := filepath.Abs(options.ProjectRoot)
if err != nil {
return nil, err
}
if _, err := exec.LookPath("gleam"); err != nil {
return nil, fmt.Errorf("source daemon requires gleam: %w", err)
}
executable = filepath.Join(root, "priv", "bin", "albedo-daemon")
}
executable, err := resolveDaemonExecutable(executable)
if err != nil {
return nil, err
}
cmd := exec.Command(executable)
for _, entry := range os.Environ() {
key, _, _ := strings.Cut(entry, "=")
switch key {
case "ALBEDO_API_KEY", "ALBEDO_MODEL", "ALBEDO_BASE_URL", "ALBEDO_PROTOCOL", "ALBEDO_HOME":
default:
cmd.Env = append(cmd.Env, entry)
}
}
cmd.Env = append(cmd.Env, "ALBEDO_HOME="+options.HomeDir)
return cmd, nil
}
func waitForPoll(ctx context.Context) error {
timer := time.NewTimer(pollInterval)
defer timer.Stop()
select {
case <-ctx.Done():
return ctx.Err()
case <-timer.C:
return nil
}
}
func acquireLauncher(ctx context.Context, homeDir string) (*os.File, error) {
// The launcher must create its lock directory before the daemon can start.
if err := os.MkdirAll(homeDir, 0700); err != nil {
return nil, err
}
file, err := os.OpenFile(filepath.Join(homeDir, "launcher.lock"), os.O_CREATE|os.O_RDWR, 0600)
if err != nil {
return nil, err
}
for {
if err := ctx.Err(); err != nil {
file.Close()
return nil, err
}
acquired, err := tryLauncherLock(file)
if err != nil {
file.Close()
return nil, err
}
if acquired {
return file, nil
}
if err := waitForPoll(ctx); err != nil {
file.Close()
return nil, err
}
}
}
// Launch serializes local startup without weakening the daemon's home lock.
func Launch(parent context.Context, options LocalOptions) (*Connection, error) {
return launchLocal(parent, options, nil)
}
// Upgrade stops only the exact authenticated protocol-v2 daemon approved by the caller.
func Upgrade(parent context.Context, options LocalOptions, approved ConnectionSnapshot) (*Connection, error) {
return launchLocal(parent, options, &approved)
}
func launchLocal(parent context.Context, options LocalOptions, approved *ConnectionSnapshot) (*Connection, error) {
ctx, cancel := context.WithTimeout(parent, startupTimeout)
defer cancel()
if err := ctx.Err(); err != nil {
return nil, err
}
if options.HomeDir == "" {
return nil, errors.New("local launch requires HomeDir")
}
home, err := filepath.Abs(options.HomeDir)
if err != nil {
return nil, err
}
options.HomeDir = home
// Validate the replacement executable before taking any destructive action.
cmd, err := daemonCommand(options)
if err != nil {
return nil, err
}
lock, err := acquireLauncher(ctx, home)
if err != nil {
return nil, &LocalError{Kind: LauncherLockFailure, HomeDir: home, Cause: err}
}
defer lock.Close() // Closing releases the advisory lock; its inode persists.
discovery, discoveryErr := Discover(ctx, home)
if approved == nil {
if discoveryErr != nil {
return nil, discoveryErr
}
if discovery.Kind == Running {
return Attach(ctx, discovery.Snapshot, func(ctx context.Context) (ConnectionSnapshot, error) { return Rediscover(ctx, home) })
}
} else {
var compatible *CompatibilityError
if discoveryErr != nil && !errors.As(discoveryErr, &compatible) {
return nil, discoveryErr
}
if discovery.Kind != Running || discovery.Snapshot != *approved {
return nil, &LocalError{Kind: TargetChanged, HomeDir: home, Cause: errors.New("the daemon changed after approval; review the current daemon before restarting")}
}
if discovery.Health.Version != ProtocolVersion {
return nil, &CompatibilityError{Version: discovery.Health.Version}
}
connection := NewConnection(discovery.Snapshot, nil)
stopErr := StopDaemon(ctx, connection)
connection.HTTPClient().CloseIdleConnections()
if stopErr != nil {
return nil, stopErr
}
for processAlive(discovery.Snapshot.Pid) {
if err := waitForPoll(ctx); err != nil {
return nil, err
}
}
current, err := Discover(ctx, home)
if err != nil {
return nil, err
}
if current.Kind == Running {
return nil, &LocalError{Kind: TargetChanged, HomeDir: home, Cause: errors.New("another daemon started during shutdown")}
}
}
return startLocal(ctx, cmd, options)
}
func startLocal(ctx context.Context, cmd *exec.Cmd, options LocalOptions) (*Connection, error) {
logPath := filepath.Join(options.HomeDir, "daemon.log")
logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600)
if err != nil {
return nil, err
}
logStart, _ := logFile.Seek(0, io.SeekEnd)
cmd.Stdout, cmd.Stderr = logFile, logFile
detach(cmd)
passFileLimit()
if err := ctx.Err(); err != nil {
logFile.Close()
return nil, err
}
if err := cmd.Start(); err != nil {
logFile.Close()
return nil, err
}
logFile.Close()
exited := make(chan error, 1)
go func() { exited <- cmd.Wait() }()
contended := false
for {
discovery, err := Discover(ctx, options.HomeDir)
if err == nil && discovery.Kind == Running {
return Attach(ctx, discovery.Snapshot, func(ctx context.Context) (ConnectionSnapshot, error) { return Rediscover(ctx, options.HomeDir) })
}
if err != nil && ctx.Err() == nil {
var api *APIError
if !errors.As(err, &api) || (api.Code != "daemon_stopping" && api.Code != "daemon_unavailable") {
return nil, err
}
}
select {
case waitErr := <-exited:
exited = nil
var exit *exec.ExitError
if errors.As(waitErr, &exit) && exit.ExitCode() == 75 {
contended = true
} else {
return nil, &LocalError{Kind: StartupFailed, HomeDir: options.HomeDir, Cause: startupExitError(waitErr, logPath, logStart, options.ProjectRoot)}
}
case <-ctx.Done():
if contended && errors.Is(ctx.Err(), context.DeadlineExceeded) {
return nil, &LocalError{Kind: HomeInUse, HomeDir: options.HomeDir, Cause: fmt.Errorf("home is held by another daemon or maintenance command; wait for it to finish and retry: %w", ctx.Err())}
}
return nil, ctx.Err()
case <-time.After(pollInterval):
}
}
}
func startupExitError(waitErr error, logPath string, logStart int64, projectRoot string) error {
message := "daemon stopped before it finished starting"
if file, err := os.Open(logPath); err == nil {
defer file.Close()
if _, err := file.Seek(logStart, io.SeekStart); err == nil {
output, _ := io.ReadAll(io.LimitReader(file, 4096))
if tail := strings.TrimSpace(string(output)); tail != "" {
message += ":\n" + tail
}
}
}
if projectRoot != "" {
message += fmt.Sprintf("\nsource checkout: %s; check ALBEDO_ROOT", projectRoot)
}
if waitErr != nil {
return fmt.Errorf("%s: %w", message, waitErr)
}
return errors.New(message)
}