Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414package 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 // A ready daemon's owner observations may queue behind other work. discoveryProbeTimeout = 5 * time.Second)
type DiscoveryKind int
const ( Absent DiscoveryKind = iota Stale Running)
type Discovery struct { Kind DiscoveryKind Snapshot ConnectionSnapshot Server ServerInfo}
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" IncompatibleDaemon LocalErrorKind = "incompatible_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, discoveryProbeTimeout) connection := NewConnection(snapshot, nil) server, err := ProbeServer(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 } if api, ok := errors.AsType[*APIError](err); ok { if api.StatusCode == http.StatusUnauthorized || api.StatusCode == http.StatusForbidden { return fail(AuthenticationFailed, err) } if api.StatusCode == http.StatusNotFound && snapshot.Version != ProtocolVersion { return fail(IncompatibleDaemon, &CompatibilityError{Version: snapshot.Version}) } return fail(UnhealthyDaemon, err) } if _, ok := errors.AsType[*ProtocolError](err); ok { return fail(UnhealthyDaemon, err) } return fail(UnreachableDaemon, err) } if server.Protocol != snapshot.Version || (server.Build != "" && server.Build != snapshot.Build) || (server.Digest != "" && snapshot.Digest != "" && server.Digest != snapshot.Digest) { return fail(InvalidDiscovery, errors.New("server identity does not match daemon.json")) } if snapshot.InstanceID != "" && snapshot.InstanceID != server.InstanceID { return fail(InvalidDiscovery, errors.New("server identity does not match daemon.json")) } snapshot.InstanceID, snapshot.Digest = server.InstanceID, server.Digest discovery := Discovery{Kind: Running, Snapshot: snapshot, Server: server} return discovery, CheckCompatible(server)}
// Rediscover waits for a verified replacement endpoint without starting one.func Rediscover(ctx context.Context, homeDir string) (ConnectionSnapshot, error) { for range 20 { 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 } if _, ok := errors.AsType[*ProtocolError](err); ok { 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) { return LocalCommand(context.Background(), options)}
// LocalCommand selects the local daemon executable without launching its server.func LocalCommand(ctx context.Context, options LocalOptions, args ...string) (*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.CommandContext(ctx, executable, args...) 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 3 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.Server.Protocol != ProtocolVersion { return nil, &CompatibilityError{Version: discovery.Server.Protocol} } 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 local *LocalError var api *APIError unreachable := errors.As(err, &local) && local.Kind == UnreachableDaemon starting := errors.As(err, &api) && (api.Code == "daemon_stopping" || api.Code == "daemon_unavailable") if !unreachable && !starting { 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)}