package runtime

import (
	"bufio"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

const (
	managedProcessJournalVersion     = 2
	managedProcessOutputPollInterval = 50 * time.Millisecond
	managedProcessOutputDrainDelay   = 750 * time.Millisecond
	managedProcessOutputRetryDelay   = 500 * time.Millisecond
)

type ProcessIdentity struct {
	Scope              string    `json:"scope"`
	ServerInstanceID   string    `json:"serverInstanceId"`
	RunEndpointID      string    `json:"runEndpointId,omitempty"`
	JobID              string    `json:"jobId,omitempty"`
	Capability         string    `json:"capability,omitempty"`
	ProfileKey         string    `json:"profileKey"`
	LogSessionID       string    `json:"logSessionId,omitempty"`
	PID                int       `json:"pid"`
	SupervisorPID      int       `json:"supervisorPid,omitempty"`
	StartedAt          time.Time `json:"startedAt"`
	CommandFingerprint string    `json:"commandFingerprint"`
	State              string    `json:"state"`
	ExitCode           int       `json:"exitCode,omitempty"`
	ExitClassification string    `json:"exitClassification,omitempty"`
	ObservationSeq     uint64    `json:"observationSeq,omitempty"`
	Attempt            int       `json:"attempt"`
	LeaseTokenHash     string    `json:"leaseTokenHash,omitempty"`
	StdoutLogRef       string    `json:"stdoutLogRef,omitempty"`
	StderrLogRef       string    `json:"stderrLogRef,omitempty"`
	StdoutStreamKey    string    `json:"stdoutStreamKey,omitempty"`
	StderrStreamKey    string    `json:"stderrStreamKey,omitempty"`
	StopEventName      string    `json:"stopEventName,omitempty"`
	StdoutOffset       int64     `json:"stdoutOffset,omitempty"`
	StderrOffset       int64     `json:"stderrOffset,omitempty"`
	UpdatedAt          time.Time `json:"updatedAt"`
}

type processJournal struct {
	Version int                        `json:"version"`
	Items   map[string]ProcessIdentity `json:"items"`
	Retired map[string]ProcessIdentity `json:"retired,omitempty"`
}

type ManagedProcessSupervisor interface {
	Start(context.Context, ProcessCommand, ProcessIdentity, ManagedProcessOutput) (ProcessIdentity, error)
	Stop(context.Context, ProcessIdentity) (ProcessIdentity, error)
	Status(ProcessIdentity) ProcessIdentity
	ResumeOutput(ManagedProcessOutput)
}

// ManagedProcessObservationSource exposes only generic supervised-process
// facts. Worker uses it to report persisted transitions after registration.
type ManagedProcessObservationSource interface {
	ManagedProcessObservations() []ProcessIdentity
}

type ManagedProcessLine struct {
	Text        string
	StartOffset int64
	EndOffset   int64
}

type ManagedProcessLineSink func(ProcessIdentity, ManagedProcessLine) error

type ManagedProcessOutput struct {
	Stdout ManagedProcessLineSink
	Stderr ManagedProcessLineSink
}

type managedProcessFiles struct {
	stdout *os.File
	stderr *os.File
}

type managedProcess interface {
	PID() int
	TargetPID() int
	Wait() (int, error)
	Kill() error
}

type managedProcessTailer struct {
	cancel context.CancelFunc
	drain  chan struct{}
}

type OSManagedProcessSupervisor struct {
	root       string
	path       string
	outputRoot string
	mu         sync.Mutex
	items      map[string]ProcessIdentity
	retired    map[string]ProcessIdentity
	tailers    map[string]*managedProcessTailer
}

func NewOSManagedProcessSupervisor(root string) (*OSManagedProcessSupervisor, error) {
	return NewOSManagedProcessSupervisorWithOutputRoot(root, root)
}

func NewOSManagedProcessSupervisorWithOutputRoot(root string, outputRoot string) (*OSManagedProcessSupervisor, error) {
	rootAbs, err := filepath.Abs(root)
	if err != nil {
		return nil, err
	}
	outputRootAbs, err := filepath.Abs(outputRoot)
	if err != nil {
		return nil, err
	}
	stateDir := filepath.Join(rootAbs, "state")
	if err := ensureDirectory(stateDir); err != nil {
		return nil, err
	}
	supervisor := &OSManagedProcessSupervisor{root: rootAbs, path: filepath.Join(stateDir, "processes.json"), outputRoot: outputRootAbs, items: map[string]ProcessIdentity{}, retired: map[string]ProcessIdentity{}, tailers: map[string]*managedProcessTailer{}}
	if err := supervisor.load(); err != nil {
		return nil, err
	}
	if err := supervisor.migrateLegacySessions(); err != nil {
		return nil, err
	}
	supervisor.Reconcile()
	return supervisor, nil
}

func (supervisor *OSManagedProcessSupervisor) Start(ctx context.Context, command ProcessCommand, identity ProcessIdentity, output ManagedProcessOutput) (ProcessIdentity, error) {
	supervisor.mu.Lock()
	defer supervisor.mu.Unlock()
	key := identity.Scope
	log.Printf("RUN phase=process.managed status=start_requested job=%s server=%s scope=%s command=%s workdir=%s", safeOptional(identity.JobID), identity.ServerInstanceID, safeOptional(identity.Scope), redactedCommandLine(command.Args), safeOptional(command.WorkDir))
	if existing, ok := supervisor.items[key]; ok && existing.State == "running" && supervisor.isAlive(existing) {
		if existing.LogSessionID == "" {
			logSessionID, err := newManagedProcessLogSessionID()
			if err != nil {
				return ProcessIdentity{}, fmt.Errorf("generate managed process log session: %w", err)
			}
			existing.LogSessionID = logSessionID
		}
		if existing.StopEventName == "" {
			existing.StopEventName = managedProcessStopEventName(existing)
		}
		existing.State = "running"
		existing.ObservationSeq++
		existing.UpdatedAt = time.Now().UTC()
		supervisor.items[key] = existing
		if err := supervisor.persistLocked(); err != nil {
			return ProcessIdentity{}, fmt.Errorf("persist managed process session: %w", err)
		}
		supervisor.startTailersLocked(existing, output)
		log.Printf("RUN phase=process.managed status=reusing_existing job=%s pid=%d state=%s", safeOptional(identity.JobID), existing.PID, existing.State)
		return existing, nil
	}
	if err := ctx.Err(); err != nil {
		log.Printf("RUN phase=process.managed status=context_done job=%s error=%s", safeOptional(identity.JobID), RedactText(err.Error()))
		return ProcessIdentity{}, err
	}
	if len(command.Args) == 0 {
		log.Printf("RUN phase=process.managed status=missing_executable job=%s", safeOptional(identity.JobID))
		return ProcessIdentity{}, fmt.Errorf("process executable is required")
	}
	startedAt := time.Now().UTC()
	if identity.LogSessionID == "" {
		logSessionID, err := newManagedProcessLogSessionID()
		if err != nil {
			return ProcessIdentity{}, fmt.Errorf("generate managed process log session: %w", err)
		}
		identity.LogSessionID = logSessionID
	}
	if identity.StopEventName == "" {
		identity.StopEventName = managedProcessStopEventName(identity)
	}
	files, identity, err := supervisor.prepareOutputFilesLocked(identity, startedAt)
	if err != nil {
		log.Printf("RUN phase=process.managed status=prepare_output_failed job=%s error=%s", safeOptional(identity.JobID), RedactText(err.Error()))
		return ProcessIdentity{}, err
	}
	log.Printf("RUN phase=process.managed status=output_ready job=%s stdoutRef=%s stderrRef=%s", safeOptional(identity.JobID), safeOptional(identity.StdoutLogRef), safeOptional(identity.StderrLogRef))
	process, err := startManagedProcess(command, files, identity.StopEventName)
	if err != nil {
		files.close()
		log.Printf("RUN phase=process.managed status=start_failed job=%s error=%s", safeOptional(identity.JobID), RedactText(err.Error()))
		return ProcessIdentity{}, err
	}
	identity.SupervisorPID = process.PID()
	identity.PID = process.TargetPID()
	identity.StartedAt = startedAt
	identity.State = "running"
	identity.ObservationSeq = 1
	identity.UpdatedAt = identity.StartedAt
	identity.CommandFingerprint = fingerprintArgs(command.Args)
	previous, hadPrevious := supervisor.items[key]
	supervisor.items[key] = identity
	if err := supervisor.persistLocked(); err != nil {
		_ = forceManagedProcessStop(identity)
		files.close()
		if hadPrevious {
			supervisor.items[key] = previous
		} else {
			delete(supervisor.items, key)
		}
		log.Printf("RUN phase=process.managed status=persist_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), identity.PID, RedactText(err.Error()))
		return ProcessIdentity{}, err
	}
	supervisor.startTailersLocked(identity, output)
	go supervisor.wait(key, process, identity.PID, files)
	log.Printf("RUN phase=process.managed status=started job=%s pid=%d fingerprint=%s", safeOptional(identity.JobID), identity.PID, safeOptional(identity.CommandFingerprint))
	return identity, nil
}

func (supervisor *OSManagedProcessSupervisor) Stop(ctx context.Context, identity ProcessIdentity) (ProcessIdentity, error) {
	supervisor.mu.Lock()
	current, ok := supervisor.items[identity.Scope]
	if !ok || current.State != "running" || !supervisor.isAlive(current) {
		if ok {
			current.State = "stopped"
			current.ExitClassification = "already-stopped"
			current.ObservationSeq++
			current.UpdatedAt = time.Now().UTC()
			supervisor.items[identity.Scope] = current
			_ = supervisor.persistLocked()
		}
		supervisor.mu.Unlock()
		log.Printf("RUN phase=process.managed status=already_stopped job=%s scope=%s", safeOptional(identity.JobID), safeOptional(identity.Scope))
		return current, nil
	}
	log.Printf("RUN phase=process.managed status=stop_requested job=%s pid=%d scope=%s", safeOptional(identity.JobID), current.PID, safeOptional(identity.Scope))
	if err := requestManagedProcessStop(current); err != nil {
		log.Printf("RUN phase=process.managed status=stop_signal_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), current.PID, RedactText(err.Error()))
	}
	supervisor.mu.Unlock()
	deadline := time.NewTimer(2 * time.Second)
	ticker := time.NewTicker(20 * time.Millisecond)
	defer deadline.Stop()
	defer ticker.Stop()
	for {
		if !supervisor.isAlive(current) {
			current.State = "stopped"
			current.ExitClassification = "requested-stop"
			current.ObservationSeq++
			current.UpdatedAt = time.Now().UTC()
			supervisor.mu.Lock()
			supervisor.items[current.Scope] = current
			_ = supervisor.persistLocked()
			supervisor.mu.Unlock()
			supervisor.drainTailersAfter(current, managedProcessOutputDrainDelay)
			log.Printf("RUN phase=process.managed status=stopped job=%s pid=%d classification=%s", safeOptional(identity.JobID), current.PID, current.ExitClassification)
			return current, nil
		}
		select {
		case <-ctx.Done():
			log.Printf("RUN phase=process.managed status=stop_context_done job=%s pid=%d error=%s", safeOptional(identity.JobID), current.PID, RedactText(ctx.Err().Error()))
			return ProcessIdentity{}, ctx.Err()
		case <-deadline.C:
			if err := forceManagedProcessStop(current); err != nil {
				log.Printf("RUN phase=process.managed status=forced_stop_signal_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), current.PID, RedactText(err.Error()))
			}
			current.State = "stopped"
			current.ExitClassification = "forced-stop"
			current.ObservationSeq++
			current.UpdatedAt = time.Now().UTC()
			supervisor.mu.Lock()
			supervisor.items[current.Scope] = current
			_ = supervisor.persistLocked()
			supervisor.mu.Unlock()
			supervisor.drainTailersAfter(current, managedProcessOutputDrainDelay)
			log.Printf("RUN phase=process.managed status=forced_stop job=%s pid=%d", safeOptional(identity.JobID), current.PID)
			return current, nil
		case <-ticker.C:
		}
	}
}

func (supervisor *OSManagedProcessSupervisor) Status(identity ProcessIdentity) ProcessIdentity {
	supervisor.mu.Lock()
	defer supervisor.mu.Unlock()
	current, ok := supervisor.items[identity.Scope]
	if !ok {
		log.Printf("RUN phase=process.managed status=not_started job=%s scope=%s", safeOptional(identity.JobID), safeOptional(identity.Scope))
		return ProcessIdentity{Scope: identity.Scope, State: "stopped", ExitClassification: "not-started"}
	}
	if current.State == "running" && !supervisor.isAlive(current) {
		current.State = "exited"
		if current.ExitClassification == "" {
			current.ExitClassification = "unexpected-exit"
		}
		current.UpdatedAt = time.Now().UTC()
		current.ObservationSeq++
		supervisor.items[current.Scope] = current
		_ = supervisor.persistLocked()
		supervisor.drainTailersLocked(current)
	}
	log.Printf("RUN phase=process.managed status=current job=%s pid=%d state=%s classification=%s", safeOptional(identity.JobID), current.PID, current.State, safeOptional(current.ExitClassification))
	return current
}

func (supervisor *OSManagedProcessSupervisor) ResumeOutput(output ManagedProcessOutput) {
	supervisor.mu.Lock()
	defer supervisor.mu.Unlock()
	now := time.Now().UTC()
	changed := false
	for _, item := range supervisor.items {
		if item.State == "running" && supervisor.isAlive(item) {
			item.StdoutOffset = supervisor.outputFileSize(item.StdoutLogRef, item.StdoutOffset)
			item.StderrOffset = supervisor.outputFileSize(item.StderrLogRef, item.StderrOffset)
			item.State = "running"
			item.ObservationSeq++
			item.UpdatedAt = now
			supervisor.items[item.Scope] = item
			changed = true
			supervisor.startTailersLocked(item, output)
			log.Printf("RUN phase=process.managed status=resume_output pid=%d server=%s scope=%s", item.PID, item.ServerInstanceID, safeOptional(item.Scope))
		}
	}
	for key := range supervisor.retired {
		delete(supervisor.retired, key)
		changed = true
	}
	if changed {
		_ = supervisor.persistLocked()
	}
}

func (supervisor *OSManagedProcessSupervisor) outputFileSize(ref string, fallback int64) int64 {
	if ref == "" {
		return fallback
	}
	info, err := os.Stat(filepath.Join(supervisor.outputRoot, "state", "process-output", filepath.Base(ref)))
	if err != nil || info.Size() < fallback {
		return fallback
	}
	return info.Size()
}

func (supervisor *OSManagedProcessSupervisor) Reconcile() {
	supervisor.mu.Lock()
	defer supervisor.mu.Unlock()
	changed := false
	for key, item := range supervisor.items {
		if item.State == "running" && !supervisor.isAlive(item) {
			item.State = "exited"
			item.ExitClassification = "unexpected-exit"
			item.UpdatedAt = time.Now().UTC()
			item.ObservationSeq++
			supervisor.items[key] = item
			// A Run restart must never replay output produced before this
			// process became observable again.
			supervisor.stopTailersLocked(item)
			changed = true
		}
	}
	if changed {
		_ = supervisor.persistLocked()
	}
}

func (supervisor *OSManagedProcessSupervisor) wait(key string, process managedProcess, pid int, files managedProcessFiles) {
	exitCode, err := process.Wait()
	files.close()
	supervisor.mu.Lock()
	defer supervisor.mu.Unlock()
	item, ok := supervisor.items[key]
	if !ok || item.PID != pid {
		return
	}
	// A durable helper may exit independently of the target process (legacy
	// shell wrappers can detach their child). Preserve a live target so the
	// next Run can continue monitoring it by PID.
	if processAlivePID(item.PID) {
		item.State = "running"
		item.SupervisorPID = 0
		item.ExitClassification = "supervisor-exited-target-alive"
	} else {
		item.State = "exited"
	}
	item.UpdatedAt = time.Now().UTC()
	item.ObservationSeq++
	item.ExitCode = exitCode
	if item.State == "running" {
		// Keep the classification assigned above.
	} else if err == nil {
		item.ExitClassification = "clean-exit"
	} else {
		item.ExitClassification = "unexpected-exit"
	}
	supervisor.items[key] = item
	_ = supervisor.persistLocked()
	if item.State != "running" {
		supervisor.drainTailersAfter(item, managedProcessOutputDrainDelay)
	}
	if err != nil {
		log.Printf("RUN phase=process.managed status=%s job=%s pid=%d state=%s exitCode=%d classification=%s error=%s", processTerminalStatus(item.State), safeOptional(item.JobID), pid, item.State, item.ExitCode, item.ExitClassification, RedactText(err.Error()))
		return
	}
	log.Printf("RUN phase=process.managed status=%s job=%s pid=%d state=%s exitCode=%d classification=%s", processTerminalStatus(item.State), safeOptional(item.JobID), pid, item.State, item.ExitCode, item.ExitClassification)
}

func processTerminalStatus(state string) string {
	if state == "running" {
		return "supervisor_exited_target_alive"
	}
	return "exited"
}

func (supervisor *OSManagedProcessSupervisor) ManagedProcessObservations() []ProcessIdentity {
	supervisor.mu.Lock()
	defer supervisor.mu.Unlock()
	items := make([]ProcessIdentity, 0, len(supervisor.items))
	for _, item := range supervisor.items {
		items = append(items, item)
	}
	return items
}

func (files managedProcessFiles) close() {
	if files.stdout != nil {
		_ = files.stdout.Close()
	}
	if files.stderr != nil {
		_ = files.stderr.Close()
	}
}

func (supervisor *OSManagedProcessSupervisor) prepareOutputFilesLocked(identity ProcessIdentity, startedAt time.Time) (managedProcessFiles, ProcessIdentity, error) {
	outputDir := filepath.Join(supervisor.outputRoot, "state", "process-output")
	if err := ensureDirectory(outputDir); err != nil {
		return managedProcessFiles{}, ProcessIdentity{}, err
	}
	base := processOutputBase(identity.Scope+"\x00"+identity.LogSessionID, startedAt)
	identity.StdoutLogRef = base + ".stdout.log"
	identity.StderrLogRef = base + ".stderr.log"
	stdout, stdoutOffset, err := openManagedOutputFile(filepath.Join(outputDir, identity.StdoutLogRef))
	if err != nil {
		return managedProcessFiles{}, ProcessIdentity{}, err
	}
	stderr, stderrOffset, err := openManagedOutputFile(filepath.Join(outputDir, identity.StderrLogRef))
	if err != nil {
		_ = stdout.Close()
		return managedProcessFiles{}, ProcessIdentity{}, err
	}
	identity.StdoutOffset = stdoutOffset
	identity.StderrOffset = stderrOffset
	return managedProcessFiles{stdout: stdout, stderr: stderr}, identity, nil
}

func openManagedOutputFile(path string) (*os.File, int64, error) {
	file, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
	if err != nil {
		return nil, 0, err
	}
	info, err := file.Stat()
	if err != nil {
		_ = file.Close()
		return nil, 0, err
	}
	return file, info.Size(), nil
}

func (supervisor *OSManagedProcessSupervisor) startTailersLocked(identity ProcessIdentity, output ManagedProcessOutput) {
	if output.Stdout != nil && identity.StdoutLogRef != "" {
		supervisor.startTailerLocked(identity, "stdout", identity.StdoutLogRef, identity.StdoutOffset, output.Stdout, true)
	}
	if output.Stderr != nil && identity.StderrLogRef != "" {
		supervisor.startTailerLocked(identity, "stderr", identity.StderrLogRef, identity.StderrOffset, output.Stderr, true)
	}
}

func (supervisor *OSManagedProcessSupervisor) startDrainTailersLocked(identity ProcessIdentity, output ManagedProcessOutput) {
	if output.Stdout != nil && identity.StdoutLogRef != "" {
		supervisor.startTailerLocked(identity, "stdout", identity.StdoutLogRef, identity.StdoutOffset, output.Stdout, false)
	}
	if output.Stderr != nil && identity.StderrLogRef != "" {
		supervisor.startTailerLocked(identity, "stderr", identity.StderrLogRef, identity.StderrOffset, output.Stderr, false)
	}
}

func (supervisor *OSManagedProcessSupervisor) startTailerLocked(identity ProcessIdentity, stream string, ref string, offset int64, sink ManagedProcessLineSink, follow bool) {
	if sink == nil {
		return
	}
	tailerID := managedProcessTailerID(identity, stream)
	if tailer, exists := supervisor.tailers[tailerID]; exists {
		if !follow {
			beginManagedProcessTailerDrain(tailer)
		}
		log.Printf("RUN phase=process.managed.output status=tail_reuse job=%s pid=%d stream=%s ref=%s offset=%d", safeOptional(identity.JobID), identity.PID, stream, safeOptional(ref), offset)
		return
	}
	path := filepath.Join(supervisor.outputRoot, "state", "process-output", filepath.Base(ref))
	ctx, cancel := context.WithCancel(context.Background())
	tailer := &managedProcessTailer{cancel: cancel, drain: make(chan struct{})}
	if !follow {
		beginManagedProcessTailerDrain(tailer)
	}
	supervisor.tailers[tailerID] = tailer
	log.Printf("RUN phase=process.managed.output status=tail_start job=%s pid=%d stream=%s ref=%s offset=%d", safeOptional(identity.JobID), identity.PID, stream, safeOptional(ref), offset)
	go supervisor.tailOutput(ctx, tailerID, tailer, identity, stream, path, offset, sink)
}

func (supervisor *OSManagedProcessSupervisor) tailOutput(ctx context.Context, tailerID string, tailer *managedProcessTailer, identity ProcessIdentity, stream string, path string, offset int64, sink ManagedProcessLineSink) {
	defer supervisor.removeTailer(tailerID, tailer, identity)
	file, err := os.Open(path)
	if err != nil {
		log.Printf("RUN phase=process.managed.output status=tail_open_failed job=%s pid=%d stream=%s path=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, safeOptional(path), RedactText(err.Error()))
		return
	}
	defer file.Close()
	if offset > 0 {
		if _, err := file.Seek(offset, io.SeekStart); err != nil {
			log.Printf("RUN phase=process.managed.output status=tail_seek_failed job=%s pid=%d stream=%s path=%s offset=%d error=%s", safeOptional(identity.JobID), identity.PID, stream, safeOptional(path), offset, RedactText(err.Error()))
			return
		}
	}
	reader := bufio.NewReader(file)
	defer func() {
		log.Printf("RUN phase=process.managed.output status=tail_stop job=%s pid=%d stream=%s offset=%d", safeOptional(identity.JobID), identity.PID, stream, offset)
	}()
	for {
		line, err := reader.ReadString('\n')
		if len(line) > 0 {
			startOffset := offset
			endOffset := offset + int64(len(line))
			text := strings.TrimSpace(line)
			if text != "" {
				for {
					if sinkErr := sink(identity, ManagedProcessLine{Text: text, StartOffset: startOffset, EndOffset: endOffset}); sinkErr == nil {
						log.Printf("RUN phase=process.managed.output status=line job=%s pid=%d stream=%s line=%q", safeOptional(identity.JobID), identity.PID, stream, RedactText(text))
						break
					} else {
						log.Printf("RUN phase=process.managed.output status=sink_retry job=%s pid=%d stream=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, RedactText(sinkErr.Error()))
					}
					select {
					case <-ctx.Done():
						return
					case <-time.After(managedProcessOutputRetryDelay):
					}
				}
			}
			for {
				if offsetErr := supervisor.updateOutputOffset(identity, stream, endOffset); offsetErr == nil {
					break
				} else {
					log.Printf("RUN phase=process.managed.output status=offset_retry job=%s pid=%d stream=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, RedactText(offsetErr.Error()))
				}
				select {
				case <-ctx.Done():
					return
				case <-time.After(managedProcessOutputRetryDelay):
				}
			}
			offset = endOffset
		}
		if err == nil {
			continue
		}
		if err != io.EOF {
			return
		}
		select {
		case <-ctx.Done():
			return
		case <-tailer.drain:
			return
		case <-time.After(managedProcessOutputPollInterval):
		}
	}
}

func (supervisor *OSManagedProcessSupervisor) removeTailer(tailerID string, tailer *managedProcessTailer, identity ProcessIdentity) {
	supervisor.mu.Lock()
	defer supervisor.mu.Unlock()
	if current, ok := supervisor.tailers[tailerID]; ok && current == tailer {
		delete(supervisor.tailers, tailerID)
	}
	key := managedProcessGenerationKey(identity)
	if retired, ok := supervisor.retired[key]; ok && !supervisor.hasPendingOutput(retired) {
		delete(supervisor.retired, key)
		if err := supervisor.persistLocked(); err != nil {
			supervisor.retired[key] = retired
			log.Printf("RUN phase=process.managed.output status=retired_prune_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), identity.PID, RedactText(err.Error()))
		}
	}
}

func (supervisor *OSManagedProcessSupervisor) updateOutputOffset(identity ProcessIdentity, stream string, offset int64) error {
	supervisor.mu.Lock()
	defer supervisor.mu.Unlock()
	item, ok := supervisor.items[identity.Scope]
	retiredKey := ""
	if !ok || !sameManagedProcessGeneration(item, identity) {
		retiredKey = managedProcessGenerationKey(identity)
		item, ok = supervisor.retired[retiredKey]
		if !ok || !sameManagedProcessGeneration(item, identity) {
			return nil
		}
	}
	if stream == "stdout" {
		item.StdoutOffset = offset
	} else {
		item.StderrOffset = offset
	}
	item.UpdatedAt = time.Now().UTC()
	if retiredKey == "" {
		supervisor.items[identity.Scope] = item
	} else {
		supervisor.retired[retiredKey] = item
	}
	return supervisor.persistLocked()
}

func sameManagedProcessGeneration(current ProcessIdentity, expected ProcessIdentity) bool {
	if current.LogSessionID != "" || expected.LogSessionID != "" {
		if current.LogSessionID == "" || current.LogSessionID != expected.LogSessionID {
			return false
		}
		return sameManagedProcessOutputFiles(current, expected)
	}
	return current.PID == expected.PID && current.StdoutLogRef == expected.StdoutLogRef && current.StderrLogRef == expected.StderrLogRef
}

func sameManagedProcessOutputFiles(current ProcessIdentity, expected ProcessIdentity) bool {
	if current.StdoutLogRef != "" && expected.StdoutLogRef != "" && current.StdoutLogRef != expected.StdoutLogRef {
		return false
	}
	if current.StderrLogRef != "" && expected.StderrLogRef != "" && current.StderrLogRef != expected.StderrLogRef {
		return false
	}
	return true
}

func (supervisor *OSManagedProcessSupervisor) drainTailersAfter(identity ProcessIdentity, delay time.Duration) {
	go func() {
		time.Sleep(delay)
		supervisor.mu.Lock()
		defer supervisor.mu.Unlock()
		supervisor.drainTailersLocked(identity)
	}()
}

func (supervisor *OSManagedProcessSupervisor) drainTailersLocked(identity ProcessIdentity) {
	for _, stream := range []string{"stdout", "stderr"} {
		if tailer, ok := supervisor.tailers[managedProcessTailerID(identity, stream)]; ok {
			beginManagedProcessTailerDrain(tailer)
		}
	}
}

func beginManagedProcessTailerDrain(tailer *managedProcessTailer) {
	select {
	case <-tailer.drain:
	default:
		close(tailer.drain)
	}
}

func (supervisor *OSManagedProcessSupervisor) stopTailersLocked(identity ProcessIdentity) {
	for _, stream := range []string{"stdout", "stderr"} {
		tailerID := managedProcessTailerID(identity, stream)
		if tailer, ok := supervisor.tailers[tailerID]; ok {
			tailer.cancel()
			delete(supervisor.tailers, tailerID)
		}
	}
}

func managedProcessTailerID(identity ProcessIdentity, stream string) string {
	generation := identity.LogSessionID
	if generation == "" {
		generation = fmt.Sprintf("legacy:%d:%s:%s", identity.PID, identity.StdoutLogRef, identity.StderrLogRef)
	} else {
		generation += "\x00" + identity.StdoutLogRef + "\x00" + identity.StderrLogRef
	}
	return identity.Scope + "\x00" + generation + "\x00" + stream
}

func managedProcessGenerationKey(identity ProcessIdentity) string {
	generation := identity.LogSessionID
	if generation == "" {
		generation = fmt.Sprintf("legacy:%d:%s:%s", identity.PID, identity.StdoutLogRef, identity.StderrLogRef)
	}
	return identity.Scope + "\x00" + generation
}

func (supervisor *OSManagedProcessSupervisor) hasPendingOutput(identity ProcessIdentity) bool {
	for _, item := range []struct {
		ref    string
		offset int64
	}{{identity.StdoutLogRef, identity.StdoutOffset}, {identity.StderrLogRef, identity.StderrOffset}} {
		if item.ref == "" {
			continue
		}
		info, err := os.Stat(filepath.Join(supervisor.outputRoot, "state", "process-output", filepath.Base(item.ref)))
		if err == nil && info.Size() > item.offset {
			return true
		}
	}
	return false
}

func (supervisor *OSManagedProcessSupervisor) isAlive(item ProcessIdentity) bool {
	return processAlivePID(item.PID)
}

func (supervisor *OSManagedProcessSupervisor) load() error {
	body, err := os.ReadFile(supervisor.path)
	if os.IsNotExist(err) {
		return nil
	}
	if err != nil {
		return err
	}
	var file processJournal
	if err := json.Unmarshal(body, &file); err != nil {
		return err
	}
	for key, item := range file.Items {
		supervisor.items[key] = item
	}
	for key, item := range file.Retired {
		supervisor.retired[key] = item
	}
	return nil
}

func (supervisor *OSManagedProcessSupervisor) migrateLegacySessions() error {
	supervisor.mu.Lock()
	defer supervisor.mu.Unlock()
	changed := false
	for key, item := range supervisor.items {
		if item.State != "running" || !supervisor.isAlive(item) {
			continue
		}
		if item.LogSessionID == "" {
			logSessionID, err := newManagedProcessLogSessionID()
			if err != nil {
				return fmt.Errorf("generate legacy managed process log session: %w", err)
			}
			item.LogSessionID = logSessionID
			changed = true
		}
		if item.StartedAt.IsZero() {
			item.StartedAt = time.Now().UTC()
			changed = true
		}
		item.UpdatedAt = time.Now().UTC()
		supervisor.items[key] = item
	}
	if !changed {
		return nil
	}
	if err := supervisor.persistLocked(); err != nil {
		return fmt.Errorf("persist legacy managed process log session: %w", err)
	}
	return nil
}

func (supervisor *OSManagedProcessSupervisor) persistLocked() error {
	body, err := json.Marshal(processJournal{Version: managedProcessJournalVersion, Items: supervisor.items, Retired: supervisor.retired})
	if err != nil {
		return err
	}
	temporary := supervisor.path + ".tmp"
	if err := os.WriteFile(temporary, body, 0o600); err != nil {
		return err
	}
	if err := os.Rename(temporary, supervisor.path); err != nil {
		_ = os.Remove(temporary)
		return err
	}
	return nil
}

func processOutputBase(scope string, startedAt time.Time) string {
	sum := sha256.Sum256([]byte(scope + "\x00" + startedAt.Format(time.RFC3339Nano)))
	return hex.EncodeToString(sum[:])
}

func managedProcessStopEventName(identity ProcessIdentity) string {
	seed := strings.Join([]string{
		"run-managed-stop-v1",
		identity.RunEndpointID,
		identity.ServerInstanceID,
		identity.Scope,
		identity.LogSessionID,
	}, "\x00")
	sum := sha256.Sum256([]byte(seed))
	return "Local\\run-managed-stop-" + hex.EncodeToString(sum[:])
}

func newManagedProcessLogSessionID() (string, error) {
	var value [16]byte
	if _, err := rand.Read(value[:]); err != nil {
		return "", err
	}
	return hex.EncodeToString(value[:]), nil
}

func fingerprintArgs(args []string) string {
	sum := sha256.Sum256([]byte(strings.Join(args, "\x00")))
	return "sha256:" + hex.EncodeToString(sum[:])
}