kunchenguid--no-mistakes/internal/daemon/daemon.go

package daemon

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"path/filepath"
	"sync"
	"time"

	"github.com/kunchenguid/no-mistakes/internal/agent"
	"github.com/kunchenguid/no-mistakes/internal/config"
	"github.com/kunchenguid/no-mistakes/internal/db"
	"github.com/kunchenguid/no-mistakes/internal/git"
	"github.com/kunchenguid/no-mistakes/internal/ipc"
	"github.com/kunchenguid/no-mistakes/internal/paths"
	"github.com/kunchenguid/no-mistakes/internal/shellenv"
	"github.com/kunchenguid/no-mistakes/internal/telemetry"
	"github.com/kunchenguid/no-mistakes/internal/types"
)

var applyShellEnvToProcess = shellenv.ApplyToProcess
var createDaemonPIDTempFile = os.CreateTemp
var renameDaemonPIDFile = os.Rename

// Run starts the daemon process. It blocks until a shutdown signal is received
// or the shutdown IPC method is called. This is called via the hidden
// `no-mistakes daemon run` entrypoint used by managed and detached services.
func Run() error {
	p, err := paths.New()
	if err != nil {
		return fmt.Errorf("resolve paths: %w", err)
	}
	if err := p.EnsureDirs(); err != nil {
		return fmt.Errorf("create directories: %w", err)
	}
	if err := prepareDaemonEnvironment(); err != nil {
		return err
	}

	// Ensure default config exists, then load it.
	config.EnsureDefaultGlobalConfig(p.ConfigFile())
	globalCfg, err := config.LoadGlobal(p.ConfigFile())
	if err != nil {
		return fmt.Errorf("load config: %w", err)
	}
	initLogger(globalCfg.LogLevel)

	d, err := db.Open(p.DB())
	if err != nil {
		return fmt.Errorf("open database: %w", err)
	}
	defer d.Close()

	return RunWithResources(p, d)
}

func prepareDaemonEnvironment() error {
	nmHome := os.Getenv("NM_HOME")
	for _, key := range []string{
		"CLAUDECODE",
		"CLAUDE_CODE_ENTRYPOINT",
		"CLAUDE_CODE_ENTRY_POINT",
		"CLAUDE_CODE_SESSION_ID",
		"CLAUDE_CODE_SESSION_ACCESS_TOKEN",
	} {
		if err := os.Unsetenv(key); err != nil {
			return fmt.Errorf("unset %s: %w", key, err)
		}
	}
	if err := applyShellEnvToProcess(); err != nil {
		return fmt.Errorf("apply login shell environment: %w", err)
	}
	if nmHome != "" {
		if err := os.Setenv("NM_HOME", nmHome); err != nil {
			return fmt.Errorf("restore NM_HOME: %w", err)
		}
	}
	logDaemonPathSummary()
	return nil
}

// logDaemonPathSummary records the effective PATH at daemon startup so that
// "agent binary not in PATH" failures (see #143) can be diagnosed from the
// daemon log alone. We emit it via slog.Default because this runs before
// initLogger; the default handler still writes to stderr, which launchd and
// systemd redirect into the daemon log file.
func logDaemonPathSummary() {
	path := os.Getenv("PATH")
	entries := 0
	if path != "" {
		entries = len(filepath.SplitList(path))
	}
	slog.Info("daemon environment ready",
		"path_entries", entries,
		"path", path,
	)
}

// initLogger sets up the global slog handler with the configured log level.
func initLogger(level string) {
	lvl := config.ParseLogLevel(level)
	handler := slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: lvl})
	slog.SetDefault(slog.New(handler))
}

// RunWithResources starts the daemon with pre-initialized paths and DB.
// Useful for testing where the caller controls resource setup.
func RunWithResources(p *paths.Paths, d *db.DB) error {
	return RunWithOptions(p, d, nil)
}

// RunWithOptions starts the daemon with optional overrides.
// stepFactory overrides the default pipeline steps (for testing).
func RunWithOptions(p *paths.Paths, d *db.DB, stepFactory StepFactory) error {
	// Singleton guard: only one live daemon may own this NM_HOME at a time.
	// This must be acquired before recoverOnStartup (global stale-run
	// recovery and orphan-worktree cleanup) and before the IPC socket is
	// bound, and held for the rest of the process lifetime - otherwise a
	// second daemon racing to start against the same root can mark another
	// live daemon's active runs as crashed and delete worktrees out from
	// under it (see AGENTS.md "Daemon Singleton Lock"). Covers both the
	// `daemon start` -> detached child path and a direct `daemon run --root`
	// invocation, since both funnel through here.
	lock, err := acquireSingletonLock(p)
	if err != nil {
		return err
	}
	defer lock.Release()

	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), 750*time.Millisecond)
		defer cancel()
		_ = telemetry.Close(ctx)
	}()

	// Point the agent package at our PID tracking dir so any managed
	// servers we spawn from here on leave crash-recovery breadcrumbs.
	agent.SetServerPIDsDir(p.ServerPIDsDir())
	defer agent.SetServerPIDsDir("")

	mgr := NewRunManager(d, p, stepFactory)

	// Recover stale runs from a previous daemon crash.
	recoverOnStartup(d, p, mgr)

	srv := ipc.NewServer()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var shutdownOnce sync.Once
	doShutdown := func(reason string) {
		shutdownOnce.Do(func() {
			slog.Info("shutting down", "reason", reason)
			mgr.Shutdown()
			cancel()
			srv.Close()
		})
	}

	registerHandlers(srv, mgr, d, func() { doShutdown("ipc request") })

	// Write PID file
	pidPath := p.PIDFile()
	pidRecord, err := currentDaemonPIDRecord(processStartTime, func() time.Time { return time.Now().UTC() })
	if err != nil {
		return fmt.Errorf("build pid file: %w", err)
	}
	if err := writeDaemonPIDFile(pidPath, pidRecord); err != nil {
		return fmt.Errorf("write pid file: %w", err)
	}
	defer func() {
		if pidData, err := os.ReadFile(pidPath); err == nil {
			if current, readErr := readDaemonPIDFileData(pidData); readErr == nil && current.PID == pidRecord.PID && current.StartedAt.Equal(pidRecord.StartedAt) {
				os.Remove(pidPath)
			}
		}
	}()

	// Handle OS signals
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, daemonSignals()...)
	defer signal.Stop(sigCh)
	go func() {
		select {
		case sig := <-sigCh:
			doShutdown(sig.String())
		case <-ctx.Done():
		}
	}()

	socketPath := p.Socket()
	slog.Info("daemon starting", "socket", socketPath, "pid", os.Getpid())

	if err := srv.Serve(socketPath); err != nil {
		return fmt.Errorf("serve: %w", err)
	}
	doShutdown("listener closed")

	// Clean up socket file only if we still own the PID file.
	// A new daemon may have already replaced the socket.
	if pidData, err := os.ReadFile(pidPath); err == nil {
		if current, readErr := readDaemonPIDFileData(pidData); readErr == nil && current.PID == pidRecord.PID && current.StartedAt.Equal(pidRecord.StartedAt) {
			os.Remove(pidPath)
			os.Remove(socketPath)
		}
	}
	slog.Info("daemon stopped")
	return nil
}

func currentDaemonPIDRecord(startTime func(int) (time.Time, error), now func() time.Time) (daemonPIDFile, error) {
	pid := os.Getpid()
	startedAt, err := startTime(pid)
	if err != nil {
		startedAt = agent.CurrentProcessStartedAt()
		if startedAt.IsZero() {
			startedAt = now()
		}
	}
	return daemonPIDFile{PID: pid, StartedAt: startedAt.UTC()}, nil
}

func writeDaemonPIDFile(path string, record daemonPIDFile) error {
	data, err := json.Marshal(record)
	if err != nil {
		return fmt.Errorf("marshal pid file: %w", err)
	}
	tmp, err := createDaemonPIDTempFile(filepath.Dir(path), filepath.Base(path)+".tmp-*")
	if err != nil {
		return fmt.Errorf("create pid temp file: %w", err)
	}
	tmpPath := tmp.Name()
	defer func() {
		if tmpPath != "" {
			_ = os.Remove(tmpPath)
		}
	}()
	if err := tmp.Chmod(0o644); err != nil {
		_ = tmp.Close()
		return fmt.Errorf("chmod pid temp file: %w", err)
	}
	if _, err := tmp.Write(data); err != nil {
		_ = tmp.Close()
		return fmt.Errorf("write pid temp file: %w", err)
	}
	if err := tmp.Close(); err != nil {
		return fmt.Errorf("close pid temp file: %w", err)
	}
	if err := renameDaemonPIDFile(tmpPath, path); err != nil {
		return fmt.Errorf("rename pid file: %w", err)
	}
	tmpPath = ""
	return nil
}

// recoverOnStartup cleans up after a previous daemon crash by marking stale
// runs/steps as failed, killing orphaned managed-server subprocesses
// (opencode, rovodev), and removing orphaned worktree directories. It also
// best-effort migrates gate bare repos in place so older installs pick up
// the per-worktree hookspath isolation introduced for issue #122 when Git
// supports config --worktree.
func recoverOnStartup(d *db.DB, p *paths.Paths, mgr *RunManager) {
	reapOrphanedServers(p)
	migrateGateConfigs(context.Background(), p)

	plans := mgr.recoverableParkedRuns(context.Background())
	preserved := make(map[string]struct{}, len(plans))
	for _, plan := range plans {
		preserved[plan.run.ID] = struct{}{}
	}
	count, err := d.RecoverStaleRunsExcept("daemon crashed during execution", preserved)
	if err != nil {
		slog.Error("failed to recover stale runs", "error", err)
		for _, plan := range plans {
			_ = plan.agent.Close()
		}
		return
	}
	if count > 0 {
		slog.Info("recovered stale runs from previous crash", "count", count)
	}

	cleanupOrphanWorktrees(d, p)
	mgr.resumeRecoveredRuns(plans)
}

// cleanupOrphanWorktrees removes worktree directories left behind by runs
// that are no longer active. It is DB-aware: a worktree is only removed when
// its run row is terminal, or when there is no matching run row at all.
// This is what keeps cleanup from deleting the checkout out from under a
// pipeline that is still actually running (see skipWorktreeCleanup).
// Called from recoverOnStartup after
// RecoverStaleRuns, so in the normal single-daemon path every run this loop
// sees has already been resolved to a terminal status; it is factored out
// separately so it can also be exercised - and its DB-aware skip behavior
// verified - independent of stale-run recovery's side effects.
func cleanupOrphanWorktrees(d *db.DB, p *paths.Paths) {
	wtRoot := p.WorktreesDir()
	entries, err := os.ReadDir(wtRoot)
	if err != nil {
		return // directory may not exist yet
	}
	ctx := context.Background()
	for _, repoEntry := range entries {
		if !repoEntry.IsDir() {
			continue
		}
		repoPath := filepath.Join(wtRoot, repoEntry.Name())
		gateDir := p.RepoDir(repoEntry.Name())
		runEntries, err := os.ReadDir(repoPath)
		if err != nil {
			continue
		}
		for _, runEntry := range runEntries {
			if !runEntry.IsDir() {
				continue
			}
			runID := runEntry.Name()
			wtPath := filepath.Join(repoPath, runID)
			if skip, reason := skipWorktreeCleanup(d, runID); skip {
				slog.Info("skipping worktree cleanup", "path", wtPath, "reason", reason)
				continue
			}
			if err := git.WorktreeRemove(ctx, gateDir, wtPath); err != nil {
				slog.Warn("git worktree remove failed, falling back to os.RemoveAll", "path", wtPath, "error", err)
				if err := os.RemoveAll(wtPath); err != nil {
					slog.Warn("failed to remove orphaned worktree", "path", wtPath, "error", err)
				}
			} else {
				slog.Info("removed orphaned worktree", "path", wtPath)
			}
		}
		// Remove empty repo dir.
		os.Remove(repoPath)
	}
}

// skipWorktreeCleanup reports whether the worktree directory for runID must
// be left alone during startup cleanup. It is the active-run guard that
// makes cleanup safe even if the singleton lock were ever bypassed: a
// worktree is never removed while its run is still pending or running -
// only terminal-run leftovers or directories with no matching run row at
// all (e.g. a directory left behind after its run row was independently
// pruned) are eligible for removal. RunManager.startRun always inserts the
// run row before creating the worktree directory, so on a single daemon a
// "no matching run" directory is never one whose insert simply hasn't landed
// yet - it is safe to remove immediately.
func skipWorktreeCleanup(d *db.DB, runID string) (bool, string) {
	run, err := d.GetRun(runID)
	if err != nil {
		return true, fmt.Sprintf("failed to look up run %s: %v", runID, err)
	}
	if run != nil && (run.Status == types.RunPending || run.Status == types.RunRunning) {
		return true, fmt.Sprintf("run %s is %s", runID, run.Status)
	}
	return false, ""
}

// migrateGateConfigs walks every bare repo under p.ReposDir() and refreshes
// no-mistakes-managed post-receive hooks, enables push options, and applies
// git.IsolateHooksPath. The operation is idempotent: bare repos already
// configured by gate.Init are left effectively unchanged, and custom hooks are
// preserved. Older bare repos (from before issue #122 was fixed) best-effort
// get their managed hook updated and per-worktree hookspath pinned when Git
// supports config --worktree, so subsequent husky-style writes to shared local
// config can no longer disable the post-receive hook.
func migrateGateConfigs(ctx context.Context, p *paths.Paths) {
	entries, err := os.ReadDir(p.ReposDir())
	if err != nil {
		// Repos dir may not exist yet on a fresh install.
		return
	}
	for _, entry := range entries {
		if !entry.IsDir() {
			continue
		}
		bareDir := filepath.Join(p.ReposDir(), entry.Name())
		if _, err := git.RefreshManagedPostReceiveHook(bareDir); err != nil {
			slog.Warn("refresh gate post-receive hook failed", "bare", bareDir, "error", err)
		}
		if _, err := git.Run(ctx, bareDir, "config", "receive.advertisePushOptions", "true"); err != nil {
			slog.Warn("enable gate push options failed", "bare", bareDir, "error", err)
		}
		if err := git.IsolateHooksPath(ctx, bareDir); err != nil {
			slog.Warn("isolate gate hooks path failed", "bare", bareDir, "error", err)
		}
	}
}

func registerHandlers(srv *ipc.Server, mgr *RunManager, d *db.DB, shutdown func()) {
	srv.Handle(ipc.MethodHealth, func(_ context.Context, _ json.RawMessage) (interface{}, error) {
		return &ipc.HealthResult{Status: "ok"}, nil
	})

	srv.Handle(ipc.MethodShutdown, func(_ context.Context, _ json.RawMessage) (interface{}, error) {
		go shutdown()
		return &ipc.ShutdownResult{OK: true}, nil
	})

	srv.Handle(ipc.MethodGetRun, func(_ context.Context, params json.RawMessage) (interface{}, error) {
		var p ipc.GetRunParams
		if err := json.Unmarshal(params, &p); err != nil {
			return nil, fmt.Errorf("invalid params: %w", err)
		}
		run, err := d.GetRun(p.RunID)
		if err != nil {
			return nil, fmt.Errorf("get run: %w", err)
		}
		if run == nil {
			return nil, fmt.Errorf("run not found: %s", p.RunID)
		}
		steps, err := d.GetStepsByRun(p.RunID)
		if err != nil {
			return nil, fmt.Errorf("get steps: %w", err)
		}
		return &ipc.GetRunResult{Run: runToInfo(d, run, steps)}, nil
	})

	srv.Handle(ipc.MethodGetRuns, func(_ context.Context, params json.RawMessage) (interface{}, error) {
		var p ipc.GetRunsParams
		if err := json.Unmarshal(params, &p); err != nil {
			return nil, fmt.Errorf("invalid params: %w", err)
		}
		runs, err := d.GetRunsByRepo(p.RepoID)
		if err != nil {
			return nil, fmt.Errorf("get runs: %w", err)
		}
		infos := make([]ipc.RunInfo, 0, len(runs))
		for _, r := range runs {
			steps, err := d.GetStepsByRun(r.ID)
			if err != nil {
				return nil, fmt.Errorf("get steps for run %s: %w", r.ID, err)
			}
			infos = append(infos, *runToInfo(d, r, steps))
		}
		return &ipc.GetRunsResult{Runs: infos}, nil
	})

	srv.Handle(ipc.MethodGetRunsForHead, func(_ context.Context, params json.RawMessage) (interface{}, error) {
		var p ipc.GetRunsForHeadParams
		if err := json.Unmarshal(params, &p); err != nil {
			return nil, fmt.Errorf("invalid params: %w", err)
		}
		runs, err := d.GetRunsByRepoHead(p.RepoID, p.Branch, p.HeadSHA)
		if err != nil {
			return nil, fmt.Errorf("get runs for head: %w", err)
		}
		infos := make([]ipc.RunInfo, 0, len(runs))
		for _, r := range runs {
			steps, err := d.GetStepsByRun(r.ID)
			if err != nil {
				return nil, fmt.Errorf("get steps for run %s: %w", r.ID, err)
			}
			infos = append(infos, *runToInfo(d, r, steps))
		}
		return &ipc.GetRunsResult{Runs: infos}, nil
	})

	srv.Handle(ipc.MethodGetActiveRun, func(_ context.Context, params json.RawMessage) (interface{}, error) {
		var p ipc.GetActiveRunParams
		if err := json.Unmarshal(params, &p); err != nil {
			return nil, fmt.Errorf("invalid params: %w", err)
		}
		run, err := d.GetActiveRun(p.RepoID, p.Branch)
		if err != nil {
			return nil, fmt.Errorf("get active run: %w", err)
		}
		if run == nil {
			return &ipc.GetActiveRunResult{}, nil
		}
		steps, err := d.GetStepsByRun(run.ID)
		if err != nil {
			return nil, fmt.Errorf("get steps: %w", err)
		}
		return &ipc.GetActiveRunResult{Run: runToInfo(d, run, steps)}, nil
	})

	srv.Handle(ipc.MethodRerun, func(ctx context.Context, params json.RawMessage) (interface{}, error) {
		var p ipc.RerunParams
		if err := json.Unmarshal(params, &p); err != nil {
			return nil, fmt.Errorf("invalid params: %w", err)
		}
		runID, err := mgr.HandleRerun(ctx, p.RepoID, p.Branch, p.SkipSteps, p.Intent)
		if err != nil {
			return nil, err
		}
		return &ipc.RerunResult{RunID: runID}, nil
	})

	srv.Handle(ipc.MethodPushReceived, func(ctx context.Context, params json.RawMessage) (interface{}, error) {
		var p ipc.PushReceivedParams
		if err := json.Unmarshal(params, &p); err != nil {
			return nil, fmt.Errorf("invalid params: %w", err)
		}
		slog.Info("push received", "ref", p.Ref, "old", p.Old, "new", p.New, "gate", p.Gate)
		runID, err := mgr.HandlePushReceived(ctx, &p)
		if err != nil {
			return nil, err
		}
		return &ipc.PushReceivedResult{RunID: runID}, nil
	})

	srv.Handle(ipc.MethodRespond, func(_ context.Context, params json.RawMessage) (interface{}, error) {
		var p ipc.RespondParams
		if err := json.Unmarshal(params, &p); err != nil {
			return nil, fmt.Errorf("invalid params: %w", err)
		}
		if err := mgr.HandleRespondWithOverrides(p.RunID, p.Step, p.Action, p.FindingIDs, p.Instructions, p.AddedFindings); err != nil {
			return nil, err
		}
		return &ipc.RespondResult{OK: true}, nil
	})

	srv.Handle(ipc.MethodCancelRun, func(_ context.Context, params json.RawMessage) (interface{}, error) {
		var p ipc.CancelRunParams
		if err := json.Unmarshal(params, &p); err != nil {
			return nil, fmt.Errorf("invalid params: %w", err)
		}
		if err := mgr.HandleCancel(p.RunID); err != nil {
			return nil, err
		}
		return &ipc.CancelRunResult{OK: true}, nil
	})

	srv.HandleStream(ipc.MethodSubscribe, func(ctx context.Context, params json.RawMessage, send func(interface{}) error) error {
		var p ipc.SubscribeParams
		if err := json.Unmarshal(params, &p); err != nil {
			return fmt.Errorf("invalid params: %w", err)
		}

		ch, unsub := mgr.Subscribe(p.RunID)
		defer unsub()

		for {
			select {
			case event, ok := <-ch:
				if !ok {
					return nil // channel closed (run completed)
				}
				if err := send(event); err != nil {
					return err // client disconnected
				}
			case <-ctx.Done():
				return ctx.Err()
			}
		}
	})
}

func runToInfo(d *db.DB, r *db.Run, steps []*db.StepResult) *ipc.RunInfo {
	info := &ipc.RunInfo{
		ID:                 r.ID,
		RepoID:             r.RepoID,
		Branch:             r.Branch,
		HeadSHA:            r.HeadSHA,
		BaseSHA:            r.BaseSHA,
		Status:             r.Status,
		PRURL:              r.PRURL,
		Error:              r.Error,
		AwaitingAgent:      r.AwaitingAgentSince != nil,
		AwaitingAgentSince: r.AwaitingAgentSince,
		CreatedAt:          r.CreatedAt,
		UpdatedAt:          r.UpdatedAt,
	}
	if len(steps) > 0 {
		info.Steps = make([]ipc.StepResultInfo, 0, len(steps))
		for _, s := range steps {
			info.Steps = append(info.Steps, stepToInfo(d, s))
		}
	}
	return info
}

func stepToInfo(d *db.DB, s *db.StepResult) ipc.StepResultInfo {
	info := ipc.StepResultInfo{
		ID:             s.ID,
		RunID:          s.RunID,
		StepName:       s.StepName,
		StepOrder:      s.StepOrder,
		Status:         s.Status,
		ExitCode:       s.ExitCode,
		DurationMS:     s.DurationMS,
		FindingsJSON:   s.FindingsJSON,
		Error:          s.Error,
		StartedAt:      s.StartedAt,
		CompletedAt:    s.CompletedAt,
		LastActivityAt: s.LastActivityAt,
		LastActivity:   s.LastActivity,
		AgentPID:       s.AgentPID,
	}
	if s.AutoFixLimit != nil {
		info.AutoFixLimit = *s.AutoFixLimit
	}
	if stats, err := d.StepFindingStats(s); err == nil {
		info.ReportedFindings = stats.ReportedFindings
		info.FixedFindings = stats.FixedFindings
	}
	if summaries, err := d.StepFixSummaries(s.ID); err == nil {
		info.FixSummaries = summaries
	}
	if rounds, err := d.StepRoundStats(s.ID); err == nil {
		info.RoundCount = rounds.TotalRounds
		info.FixRoundCount = rounds.FixRounds
		info.PendingFixSource = rounds.PendingFixSource
	}
	return info
}