package agents

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"g.fc.run/ai/byre/internal/agentinput"
	"g.fc.run/ai/byre/internal/api"
	"g.fc.run/ai/byre/internal/claude"
	"g.fc.run/ai/byre/internal/clients"
	"g.fc.run/ai/byre/internal/git"
	"g.fc.run/ai/byre/internal/ids"
	"g.fc.run/ai/byre/internal/mcp"
	"g.fc.run/ai/byre/internal/store"
	"g.fc.run/ai/byre/internal/terminal"
	"tailscale.com/tsnet"
)

var (
	ErrConflict      = errors.New("conflict")
	ErrInvalid       = errors.New("invalid")
	ErrNoGitIdentity = errors.New("set your git name and email (PUT /api/me) before starting agents")
)

const agentActor = "agent"

const restartNotice = "byre restarted while you were mid-turn, so your last turn was cut off. Re-check the working tree (git status and the files you were editing) before continuing where you left off."

const subagentsLostNotice = "Your agent process ended while background subagents you started were still running, so they were stopped and will never report back. Anything they found is lost; redo that work (run them again or do it yourself) and carry on where you left off."

var progressSubtypes = map[string]bool{"status": true, "thinking_tokens": true}

type Manager struct {
	store           *store.Store
	worktreeRoot    string
	leadRoot        string
	handsRoot       string
	codexHome       string
	antigravityRoot string
	status          func(context.Context) (api.Status, error)
	addProject      func(ctx context.Context, name, repoPath, createdBy string) (api.Project, error)
	clients         *clients.Registry
	terminals       *terminal.Terminals
	tailnet         *tsnet.Server

	mu            sync.Mutex
	live          map[string]*live
	starting      map[string]*start
	closing       bool
	rateLimits    map[api.AgentSessionVendor]api.Usage
	usageCommands map[api.AgentSessionVendor]string
	paneReports   map[string]paneReport
	paneMoved     map[string]time.Time
	moveHolds     map[string]showHold
	startHolds    map[string]showHold
	handoffModel  map[string]modelChoice
	mergeTurns    map[string]mergeTurn
	paused        map[string]chan struct{}

	deliverMu    sync.Mutex
	deliverLocks map[string]*sessionLock
	leadMu       sync.Mutex
	signals      chan []api.Event

	snapMu sync.Mutex
	snaps  map[string]*snapState

	pokes chan struct{}

	factsMu    sync.Mutex
	factsRun   sync.Mutex
	factsDirty map[string]bool
	factsPoke  chan struct{}

	reviewMu sync.Mutex

	usageMu map[api.AgentSessionVendor]*sync.Mutex

	workflowMu sync.Mutex

	auditMu sync.Mutex

	exposeMu         sync.Mutex
	previewMu        sync.Mutex
	previewListeners map[int]*previewListener

	bad       *badLines
	watchStop chan struct{}
}

type live struct {
	sess        agentProcess
	id          string
	dir         string
	model       string
	effort      string
	gitName     string
	gitEmail    string
	vendor      api.AgentSessionVendor
	taskID      string
	projectID   string
	owner       string
	member      string
	notes       string
	reviewID    string
	stage       string
	stopping    bool
	interrupted bool
	fetching    bool
	asked       bool
	bounced     bool
	compacting  bool
	wantCompact bool
	rateLimited map[string]any
	lastText    string
	subagents   map[string]bool

	hintedImages int64

	contextModel  string
	contextUsed   int
	contextWindow int
	contextWarned bool
}

type start struct {
	done     chan struct{}
	cancel   context.CancelFunc
	l        *live
	err      error
	stopping bool
}

func (l *live) implementer() bool {
	return l.owner == "" && l.reviewID == ""
}

func New(st *store.Store, worktreeRoot, leadRoot, handsRoot, codexHome, antigravityRoot string, status func(context.Context) (api.Status, error), addProject func(ctx context.Context, name, repoPath, createdBy string) (api.Project, error), clientRegistry *clients.Registry, terminals *terminal.Terminals, tailnet *tsnet.Server) *Manager {
	m := &Manager{
		store:            st,
		worktreeRoot:     worktreeRoot,
		leadRoot:         leadRoot,
		handsRoot:        handsRoot,
		codexHome:        codexHome,
		antigravityRoot:  antigravityRoot,
		status:           status,
		addProject:       addProject,
		clients:          clientRegistry,
		terminals:        terminals,
		tailnet:          tailnet,
		live:             map[string]*live{},
		starting:         map[string]*start{},
		deliverLocks:     map[string]*sessionLock{},
		rateLimits:       map[api.AgentSessionVendor]api.Usage{},
		usageCommands:    map[api.AgentSessionVendor]string{},
		usageMu:          map[api.AgentSessionVendor]*sync.Mutex{api.Claude: {}, api.Codex: {}, api.Antigravity: {}},
		paneReports:      map[string]paneReport{},
		paneMoved:        map[string]time.Time{},
		moveHolds:        map[string]showHold{},
		startHolds:       map[string]showHold{},
		handoffModel:     map[string]modelChoice{},
		mergeTurns:       map[string]mergeTurn{},
		paused:           map[string]chan struct{}{},
		signals:          make(chan []api.Event, 1024),
		snaps:            map[string]*snapState{},
		pokes:            make(chan struct{}, 1),
		factsDirty:       map[string]bool{},
		factsPoke:        make(chan struct{}, 1),
		previewListeners: map[int]*previewListener{},
		bad:              newBadLines(),
		watchStop:        make(chan struct{}),
	}
	go m.signalLoop()
	go m.factsLoop()
	go m.loadUsage()
	return m
}

func (m *Manager) taskFor(ctx context.Context, sess api.AgentSession) (api.Task, error) {
	if sess.Owner != "" {
		return api.Task{}, nil
	}
	return m.store.GetTask(ctx, sess.TaskId)
}

type Prompt struct {
	Text   string
	Kind   store.MessageKind
	Model  string
	Effort string
	SendID string
	Client string
	Images []string
}

func (m *Manager) Create(ctx context.Context, task api.Task, creator api.User, p Prompt) (api.AgentSession, error) {
	unlock := m.lockDelivery("create:" + task.Id)
	defer unlock()

	ctx = context.WithoutCancel(ctx)
	prior, sent, err := m.priorSend(ctx, p.SendID)
	if err != nil || sent {
		return prior, err
	}
	unused, found, err := m.store.UnusedSession(ctx, task.Id, "")
	if err != nil {
		return api.AgentSession{}, err
	}
	if found {
		slog.Info("first message to an unused session", "session", unused, "task", task.Id)
		return m.Send(ctx, unused, creator.Login, p)
	}
	return m.create(ctx, task, creator, "", creator.Login, p)
}

func (m *Manager) create(ctx context.Context, task api.Task, creator api.User, member, actor string, p Prompt) (api.AgentSession, error) {
	if !creator.HasGitIdentity() {
		return api.AgentSession{}, ErrNoGitIdentity
	}

	if task.Status != api.TaskStatusActive {
		return api.AgentSession{}, fmt.Errorf("%w: task %s is %s", ErrConflict, task.Id, task.Status)
	}

	sess, err := m.newSession(ctx, task, creator, member, p.Model, p.Effort)
	if err != nil {
		return api.AgentSession{}, err
	}
	return m.turn(ctx, sess, task, actor, p)
}

func (m *Manager) newSession(ctx context.Context, task api.Task, creator api.User, member, model, effort string) (api.AgentSession, error) {
	if task.Workflow != nil {
		return api.AgentSession{}, fmt.Errorf("%w: task %s is a workflow run, which has no agent; its stage tasks have theirs", ErrInvalid, task.Id)
	}
	existing, err := m.store.ListSessions(ctx, task.Id)
	if err != nil {
		return api.AgentSession{}, err
	}
	for _, s := range existing {
		if s.State != api.AgentSessionStateCompleted && s.Member == member {
			return api.AgentSession{}, fmt.Errorf("%w: task %s already has session %s", ErrConflict, task.Id, s.Id)
		}
	}

	sess := sessionFor(task, creator, member, model, effort)
	err = m.store.CreateSession(ctx, sess, task.ProjectId)
	return sess, err
}

func sessionFor(task api.Task, creator api.User, member, model, effort string) api.AgentSession {
	id := ids.New(ids.Session)
	vendor := vendorOf(model)
	return api.AgentSession{
		Id:              id,
		TaskId:          task.Id,
		Vendor:          vendor,
		Model:           model,
		Effort:          effort,
		VendorSessionId: initialVendorSessionID(vendor),
		State:           api.AgentSessionStateIdle,
		Role:            api.Implementer,
		Member:          member,
		CreatedBy:       creator.Login,
		CreatedAt:       time.Now().UTC(),
	}
}

func (m *Manager) Send(ctx context.Context, sessionID, actor string, p Prompt) (api.AgentSession, error) {
	sess, err := m.store.GetSession(ctx, sessionID)
	if err != nil {
		return api.AgentSession{}, err
	}

	if sess.State != api.AgentSessionStateIdle {
		return api.AgentSession{}, fmt.Errorf("%w: session %s is %s", ErrConflict, sess.Id, sess.State)
	}

	sess, err = m.changeModel(ctx, sess, p.Model, p.Effort, actor)
	if err != nil {
		return api.AgentSession{}, err
	}

	task, err := m.taskFor(ctx, sess)
	if err != nil {
		return api.AgentSession{}, err
	}

	return m.turn(ctx, sess, task, actor, p)
}

func (m *Manager) Interrupt(ctx context.Context, sessionID string) (api.AgentSession, error) {
	sess, err := m.store.GetSession(ctx, sessionID)
	if err != nil {
		return api.AgentSession{}, err
	}

	m.mu.Lock()
	l := m.live[sessionID]
	m.mu.Unlock()

	if sess.State != api.AgentSessionStateRunning || l == nil {
		return api.AgentSession{}, fmt.Errorf("%w: session %s has no turn running", ErrConflict, sess.Id)
	}

	m.mu.Lock()
	l.interrupted = true
	fetching := l.fetching
	m.mu.Unlock()

	if fetching {
		return sess, nil
	}

	ictx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	err = l.sess.interrupt(ictx)
	if errors.Is(err, context.DeadlineExceeded) {
		slog.Warn("agent didn't answer interrupt, ending its process", "session", sess.Id)
		l.sess.Kill()
		return sess, nil
	}
	if err != nil {
		return api.AgentSession{}, err
	}

	return sess, nil
}

func (m *Manager) StopTask(ctx context.Context, taskID, actor string) error {
	sessions, err := m.store.ListSessions(ctx, taskID)
	if err != nil {
		return err
	}

	for _, sess := range sessions {
		if sess.State == api.AgentSessionStateCompleted {
			continue
		}
		unlock := m.lockDelivery(sess.Id)
		err = m.retire(ctx, sess, actor)
		unlock()
		if err != nil {
			return err
		}
	}

	return nil
}

func (m *Manager) retire(ctx context.Context, sess api.AgentSession, actor string) error {
	m.stopLive(sess.Id)
	err := m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateCompleted, actor, nil)
	m.dropPause(sess.Id)
	return err
}

func (m *Manager) stopLive(sessionID string) {
	m.mu.Lock()
	l := m.live[sessionID]
	if l != nil {
		l.stopping = true
	}
	st := m.starting[sessionID]
	if st != nil {
		st.stopping = true
	}
	m.mu.Unlock()

	if l != nil {
		l.sess.Kill()
	}
}

func (m *Manager) recoverSession(ctx context.Context, sess api.AgentSession) error {
	task, err := m.taskFor(ctx, sess)
	if err != nil {
		slog.Error("resume session: load task", "session", sess.Id, "task", sess.TaskId, "err", err)
		return nil
	}

	pending, err := m.store.SessionPending(ctx, sess.Id)
	if err != nil {
		return err
	}
	started, err := m.store.SessionStarted(ctx, sess.Id)
	if err != nil {
		return err
	}
	finished, err := m.store.StageFinished(ctx, sess.Id)
	if err != nil {
		return err
	}
	if finished || !pending && !started {
		m.endMergeTurn(sess.Id, true)
		err = m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateIdle, "byre", nil)
		if err != nil {
			return err
		}
		slog.Info("session left idle at startup: its stage finished or it never got its first message", "session", sess.Id, "task", task.Id)
		return nil
	}
	if pending {
		found, err := m.deliverRecorded(ctx, sess, task)
		if err != nil {
			slog.Error("resume session: deliver its recorded message", "session", sess.Id, "task", task.Id, "err", err)
		}
		if found {
			return nil
		}
	}

	l, err := m.ensureLive(ctx, sess, task)
	if err != nil {
		return m.failResume(ctx, sess, err)
	}

	notice := restartNotice
	if sess.Role == api.Implementer {
		discard, err := m.store.GetDiscard(ctx, task.Id)
		if err != nil {
			return err
		}
		if discard.State == store.DiscardRunning && discard.Session == sess.Id {
			notice = discardRestartPrompt
		}
	}
	err = m.message(ctx, sess, task, l, "byre", notice, nil)
	if err != nil {
		return m.failResume(ctx, sess, err)
	}
	slog.Info("resumed session", "session", sess.Id, "task", task.Id, "owner", sess.Owner)
	m.logUndelivered(ctx, sess.Id, "resumed with restart notice")
	return nil
}

func (m *Manager) Recover(ctx context.Context) error {
	err := m.store.ClearMainPushes(ctx)
	if err != nil {
		return err
	}

	err = m.loadPauses(ctx)
	if err != nil {
		return err
	}

	err = m.recoverPreviews(ctx)
	if err != nil {
		return err
	}

	stale, err := m.store.SessionsWithSubagents(ctx)
	if err != nil {
		return err
	}
	orphaned := []string{}
	for _, sess := range stale {
		err = m.store.SetSubagents(ctx, sess.Id, 0, "byre")
		if err != nil {
			return err
		}
		if sess.State == api.AgentSessionStateIdle {
			orphaned = append(orphaned, sess.Id)
		}
	}

	sessions, err := m.store.SessionsInState(ctx, api.AgentSessionStateRunning)
	if err != nil {
		return err
	}

	err = m.recoverMergeTurns(ctx, sessions)
	if err != nil {
		return err
	}

	for _, sess := range sessions {
		m.mu.Lock()
		resumed := m.paused[sess.Id]
		m.mu.Unlock()
		if resumed != nil {
			slog.Info("session resume held: paused", "session", sess.Id, "task", sess.TaskId)
			go m.recoverAfterPause(sess.Id)
			continue
		}
		err = m.recoverSession(ctx, sess)
		if err != nil {
			return err
		}
	}

	for _, id := range orphaned {
		m.wakeOrphaned(ctx, id)
	}

	idle, err := m.store.SessionsInState(ctx, api.AgentSessionStateIdle)
	if err != nil {
		return err
	}
	for _, sess := range idle {
		m.logUndelivered(ctx, sess.Id, "idle at startup")
	}

	err = m.store.ClearReviewRequests(ctx)
	if err != nil {
		return err
	}

	go m.coordinate()
	m.poke()
	m.MarkFacts(allTasks)
	go m.auditAll(context.Background())
	return nil
}

func (m *Manager) wakeOrphaned(ctx context.Context, sessionID string) {
	_, _, err := m.Message(ctx, sessionID, "byre", Prompt{Text: subagentsLostNotice})
	if err != nil {
		slog.Error("wake session whose subagents were stopped", "session", sessionID, "err", err)
		return
	}
	slog.Info("woke session whose subagents were stopped", "session", sessionID)
}

func (m *Manager) failResume(ctx context.Context, sess api.AgentSession, err error) error {
	slog.Error("resume session", "session", sess.Id, "err", err)
	if sess.Role == api.Reviewer {
		r, rerr := m.store.ReviewBySession(ctx, sess.Id)
		if rerr == nil {
			m.failReview(ctx, r.Id, "byre restarted and couldn't resume the reviewer: "+err.Error())
		}
	}
	m.endMergeTurn(sess.Id, true)
	m.logUndelivered(ctx, sess.Id, "resume failed")
	derr := m.store.DropStageOutput(ctx, sess.Id)
	if derr != nil {
		return derr
	}
	_, err = m.store.EndTurn(ctx, sess.Id, "byre", map[string]any{"error": err.Error()}, nil, false, false, false)
	return err
}

func (m *Manager) logUndelivered(ctx context.Context, sessionID, reason string) {
	pending, err := m.store.Undelivered(ctx, sessionID)
	if err != nil {
		slog.Error("list undelivered responses", "session", sessionID, "err", err)
		return
	}
	if len(pending) == 0 {
		return
	}
	slog.Info("undelivered responses", "session", sessionID, "reason", reason, "responses", responseIDs(pending))
}

func (m *Manager) Close() {
	close(m.watchStop)

	m.mu.Lock()
	m.closing = true
	all := []*live{}
	for _, l := range m.live {
		all = append(all, l)
	}
	pending := []*start{}
	for _, st := range m.starting {
		pending = append(pending, st)
	}
	m.mu.Unlock()

	for _, st := range pending {
		st.cancel()
	}
	for _, l := range all {
		l.sess.Kill()
	}
	for _, st := range pending {
		<-st.done
	}

	err := m.parkLeads(context.Background())
	if err != nil {
		slog.Error("park lead sessions", "err", err)
	}
}

func (m *Manager) LiveCount() int {
	m.mu.Lock()
	defer m.mu.Unlock()
	return len(m.live)
}

type Usage struct {
	Running    []api.TurnRef
	RateLimits map[string]api.Usage
}

func (m *Manager) Usage(ctx context.Context) (Usage, error) {
	u := Usage{
		Running:    []api.TurnRef{},
		RateLimits: map[string]api.Usage{},
	}

	running, err := m.store.SessionsInState(ctx, api.AgentSessionStateRunning)
	if err != nil {
		return u, err
	}
	for _, sess := range running {
		if sess.Owner != "" {
			continue
		}
		u.Running = append(u.Running, api.TurnRef{SessionId: sess.Id, TaskId: sess.TaskId, Vendor: string(sess.Vendor)})
	}

	m.mu.Lock()
	defer m.mu.Unlock()
	for vendor, info := range m.rateLimits {
		u.RateLimits[string(vendor)] = info
	}
	return u, nil
}

func (m *Manager) turn(ctx context.Context, sess api.AgentSession, task api.Task, actor string, p Prompt) (api.AgentSession, error) {
	if m.isPaused(sess.Id) {
		return api.AgentSession{}, fmt.Errorf("%w: session %s is paused; resume it first", ErrConflict, sess.Id)
	}
	err := m.startRunning(ctx, sess, actor)
	if err != nil {
		return api.AgentSession{}, err
	}
	return m.launch(ctx, sess, task, actor, p)
}

func (m *Manager) startRunning(ctx context.Context, sess api.AgentSession, actor string) error {
	return transition(ctx, m.store, sess.Id, api.AgentSessionStateIdle, api.AgentSessionStateRunning, actor)
}

func (m *Manager) launch(ctx context.Context, sess api.AgentSession, task api.Task, actor string, p Prompt) (api.AgentSession, error) {
	imgs, compacting, err := m.recordMessage(ctx, sess, task, actor, p)
	if err != nil {
		m.failStart(ctx, sess.Id, actor, err)
		return api.AgentSession{}, err
	}
	err = m.deliver(ctx, sess, task, actor, p.Client, p.Text, imgs, compacting)
	if err != nil {
		return api.AgentSession{}, err
	}
	return m.store.GetSession(ctx, sess.Id)
}

func (m *Manager) deliver(ctx context.Context, sess api.AgentSession, task api.Task, actor, client, text string, imgs []agentinput.Image, compacting bool) error {
	l, err := m.ensureLive(ctx, sess, task)
	if err == nil {
		err = m.syncModel(ctx, l, sess.Model, sess.Effort)
	}
	if err != nil {
		m.failStart(ctx, sess.Id, actor, err)
		return err
	}

	m.beginTurn(l, compacting)
	err = m.send(ctx, l, m.agentText(actor, client, text), imgs)
	if err != nil {
		m.endMergeTurn(sess.Id, false)
		return err
	}
	return nil
}

func (m *Manager) deliverRecorded(ctx context.Context, sess api.AgentSession, task api.Task) (bool, error) {
	ev, found, err := m.store.LastMessage(ctx, sess.Id)
	if err != nil || !found {
		return false, err
	}
	text, _ := ev.Payload["text"].(string)
	client, _ := ev.Payload["client"].(string)
	handoff, _ := ev.Payload[string(store.MessageHandoff)].(bool)
	imgs, err := m.readImages(ctx, anyStrings(ev.Payload["images"]))
	if err != nil {
		m.failStart(ctx, sess.Id, ev.Actor, err)
		return true, err
	}
	return true, m.deliver(ctx, sess, task, ev.Actor, client, text, imgs, handoff)
}

func (m *Manager) readImages(ctx context.Context, ids []string) ([]agentinput.Image, error) {
	imgs := []agentinput.Image{}
	for _, id := range ids {
		b, data, err := m.store.ReadBlob(ctx, id)
		if err != nil {
			return nil, err
		}
		imgs = append(imgs, agentinput.Image{MediaType: b.ContentType, Data: data})
	}
	return imgs, nil
}

func (m *Manager) failStart(ctx context.Context, sessionID, actor string, err error) {
	m.endMergeTurn(sessionID, false)
	cctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
	defer cancel()

	r, rerr := m.store.ReviewBySession(cctx, sessionID)
	if rerr == nil {
		m.failReview(cctx, r.Id, "the reviewer didn't start: "+err.Error())
	}
	serr := m.store.SetSessionState(cctx, sessionID, api.AgentSessionStateIdle, actor, map[string]any{"error": err.Error()})
	if serr != nil {
		slog.Error("record failed start", "session", sessionID, "err", serr)
	}
}

func (m *Manager) syncModel(ctx context.Context, l *live, model, effort string) error {
	if model == l.model && effort == l.effort {
		return nil
	}

	cctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	err := l.sess.setModel(cctx, model, effort)
	if err != nil {
		return fmt.Errorf("set model %s, effort %q: %w", model, effort, err)
	}
	l.model = model
	l.effort = effort
	return nil
}

func transition(ctx context.Context, st *store.Store, id string, from, to api.AgentSessionState, actor string) error {
	err := st.TransitionSession(ctx, id, from, to, actor)
	if errors.Is(err, store.ErrStateChanged) {
		return fmt.Errorf("%w: %w", ErrConflict, err)
	}
	return err
}

func (m *Manager) message(ctx context.Context, sess api.AgentSession, task api.Task, l *live, actor, text string, images []string) error {
	imgs, compacting, err := m.recordMessage(ctx, sess, task, actor, Prompt{Text: text, Images: images})
	if err != nil {
		return err
	}
	m.beginTurn(l, compacting)
	return m.send(ctx, l, m.agentText(actor, "", text), imgs)
}

func (m *Manager) send(ctx context.Context, l *live, text string, imgs []agentinput.Image) error {
	err := l.sess.send(ctx, text, imgs)
	if err != nil {
		return err
	}
	return m.store.MarkSessionDelivered(ctx, l.id)
}

func (m *Manager) recordMessage(ctx context.Context, sess api.AgentSession, task api.Task, actor string, p Prompt) ([]agentinput.Image, bool, error) {
	implementer := sess.Role == api.Implementer
	if implementer && p.Kind == store.MessageMergeIn {
		m.ownMerge(sess.Id, task.Id, task.WorktreePath)
	}
	imgs, err := m.readImages(ctx, p.Images)
	if err != nil {
		return nil, false, err
	}

	payload := map[string]any{"text": p.Text}
	if implementer {
		head, err := git.Head(ctx, task.WorktreePath)
		if err != nil {
			return nil, false, err
		}
		payload["head"] = head
	}
	if len(p.Images) > 0 {
		payload["images"] = p.Images
	}
	markKind(payload, p.Kind)
	if p.SendID != "" {
		payload["send_id"] = p.SendID
	}
	if p.Client != "" {
		payload["client"] = p.Client
	}
	err = m.store.RecordTurnMessage(ctx, sess.Id, newEvent(sess.Id, sess.Owner, task.ProjectId, task.Id, actor, "user.message", payload))
	if err != nil {
		return nil, false, err
	}
	return imgs, p.Kind == store.MessageHandoff, nil
}

func markKind(payload map[string]any, kind store.MessageKind) {
	if kind != "" {
		payload[string(kind)] = true
	}
}

func (m *Manager) beginTurn(l *live, compacting bool) {
	m.mu.Lock()
	l.interrupted = false
	l.asked = false
	l.bounced = false
	l.wantCompact = false
	l.compacting = compacting
	l.rateLimited = nil
	l.lastText = ""
	m.mu.Unlock()
}

func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task api.Task) (*live, error) {
	m.mu.Lock()
	if m.closing {
		m.mu.Unlock()
		return nil, errors.New("shutting down")
	}
	l := m.live[sess.Id]
	if l != nil {
		m.mu.Unlock()
		return l, nil
	}
	st := m.starting[sess.Id]
	if st != nil {
		m.mu.Unlock()
		select {
		case <-st.done:
			return st.l, st.err
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
	sctx, cancel := context.WithCancel(ctx)
	st = &start{done: make(chan struct{}), cancel: cancel}
	m.starting[sess.Id] = st
	m.mu.Unlock()

	l, err := m.startLive(sctx, sess, task)
	cancel()

	m.mu.Lock()
	delete(m.starting, sess.Id)
	closing := m.closing
	stopping := st.stopping
	if err == nil && !closing {
		l.stopping = stopping
		m.live[sess.Id] = l
	}
	m.mu.Unlock()

	if err == nil && closing {
		go l.sess.discard()
		l.sess.Kill()
		l, err = nil, errors.New("shutting down")
	}
	st.l, st.err = l, err
	close(st.done)
	if err != nil {
		return nil, err
	}

	go m.pump(l)
	if stopping {
		l.sess.Kill()
	}
	return l, nil
}

func (m *Manager) startLive(ctx context.Context, sess api.AgentSession, task api.Task) (*live, error) {
	creator, err := m.store.GetUser(ctx, sess.CreatedBy)
	if err != nil {
		return nil, err
	}

	l := &live{
		id:        sess.Id,
		dir:       task.WorktreePath,
		model:     sess.Model,
		effort:    sess.Effort,
		gitName:   creator.GitName,
		gitEmail:  creator.GitEmail,
		vendor:    sess.Vendor,
		taskID:    task.Id,
		projectID: task.ProjectId,
		owner:     sess.Owner,
		member:    sess.Member,
		subagents: map[string]bool{},
	}

	g, err := m.taskGuidance(ctx, task)
	if err != nil {
		return nil, err
	}
	tools := append([]mcp.Tool{m.askTool(l), m.cancelQuestionTool(l), m.setTitlesTool(l), m.stopMergeTool(l), m.mergeToMainTool(l), m.setTaskSummaryTool(l), m.searchClosedTool(l), m.exposePortTool(l), m.takeSnapshotTool(l), m.terminalReadTool(l), m.terminalTypeTool(l), m.bashTool(l), m.readKnowledgeTool(), m.renderDiagramTool(), m.exportDiagramTool(l), m.randomChoiceTool()}, append(append(m.fileTools(l), m.subtaskTools(l)...), m.clientTools(l)...)...)
	lead := true
	if sess.Role == api.Implementer {
		g += longTitleGuidance(task)
		notes, err := m.notesText(ctx, task.CreatedBy, task.ProjectId)
		if err != nil {
			return nil, err
		}
		l.notes = "\n\n" + taskNotesGuidance + "\n\n" + notes
		tools = append(tools, m.suggestNoteTool(l), m.readyForMergeTool(l), m.suggestDiscardTool(l), m.cancelReadyTool(l), m.respondToFindingTool(l))
		if sess.Vendor == api.Claude {
			tools = append(tools, m.forkTool(l))
		}
	}
	if sess.Role == api.Implementer && sess.Member != "" {
		sessions, err := m.store.ListSessions(ctx, task.Id)
		if err != nil {
			return nil, err
		}
		lead = sess.Member == teamLead(sessions)
		g += teamGuidance(sessions, sess.Member, task.WorkflowStage != nil)
		tools = append(tools, m.boardTools(l)...)
	}
	if sess.Role == api.Implementer && task.WorkflowStage != nil && lead {
		frozen, err := m.store.TaskWorkflow(ctx, task.Id)
		if err != nil {
			return nil, err
		}
		l.stage = *task.WorkflowStage
		g += stageGuidance(frozen, l.stage)
		tools = append(tools, m.finishStageTool(l))
	}
	stdio := []mcp.Stdio{}
	if sess.Role == api.Implementer {
		media := MediaDir(task.WorktreePath)
		err = os.MkdirAll(media, 0o755)
		if err != nil {
			return nil, err
		}
		g += screenshotGuidance
		tools = append(tools, m.generateImageTool(l))
		hands, err := m.handsMCP(sess.Id, media)
		if err != nil {
			return nil, err
		}
		stdio = []mcp.Stdio{hands, playwrightMCP(media)}
	}
	if sess.Role == api.Reviewer {
		r, err := m.store.ReviewBySession(ctx, sess.Id)
		if err != nil {
			return nil, err
		}
		l.reviewID = r.Id
		l.dir = m.reviewDir(task.ProjectId, r.Id)
		g = reviewGuidance
		tools = []mcp.Tool{m.submitReviewTool(l)}
		stdio = []mcp.Stdio{playwrightMCP(m.reviewMediaDir(task.ProjectId, r.Id))}
	}
	if l.owner != "" {
		l.dir = filepath.Join(m.leadRoot, url.PathEscape(l.owner))
		err = os.MkdirAll(l.dir, 0o755)
		if err != nil {
			return nil, err
		}
		g = leadGuidance
		notes, err := m.notesText(ctx, l.owner, "")
		if err != nil {
			return nil, err
		}
		l.notes = "\n\n" + notes
		tools = append(append(m.leadTools(l), m.readKnowledgeTool(), m.renderDiagramTool(), m.exportDiagramTool(l), m.randomChoiceTool()), m.clientShellTools()...)
	}
	env := git.IdentityEnv(l.gitName, l.gitEmail)
	servers := []*mcp.Server{{Name: "byre", Version: "0.0.0", Tools: tools}}

	sctx, cancel := context.WithTimeout(ctx, 60*time.Second)
	defer cancel()

	switch l.vendor {
	case api.Codex:
		l.sess, err = m.startCodex(sctx, sess, l, g, servers, stdio, env)
	case api.Antigravity:
		l.sess, err = m.startAntigravity(sctx, sess, l, g, servers, stdio, env)
	default:
		l.sess, err = m.startClaude(sctx, sess, l, g, servers, stdio, env)
	}
	if err != nil {
		return nil, err
	}
	return l, nil
}

func (m *Manager) pump(l *live) {
	ctx := context.Background()

	l.sess.drive(m, l)

	err := os.RemoveAll(m.handsDir(l.id))
	if err != nil {
		slog.Error("remove session's byre-hands", "session", l.id, "err", err)
	}

	m.mu.Lock()
	delete(m.live, l.id)
	closing := m.closing
	stopping := l.stopping
	interrupted := l.interrupted
	m.mu.Unlock()

	if closing {
		return
	}

	orphaned := len(l.subagents) > 0
	if orphaned {
		err := m.store.SetSubagents(ctx, l.id, 0, "byre")
		if err != nil {
			slog.Error("clear exited session's subagents", "session", l.id, "err", err)
		}
	}

	if stopping {
		m.logUndelivered(ctx, l.id, "agent process ended while stopping")
		return
	}

	sess, err := m.store.GetSession(ctx, l.id)
	if err != nil {
		slog.Error("load exited session", "session", l.id, "err", err)
		return
	}

	if sess.State == api.AgentSessionStateIdle && orphaned {
		m.wakeOrphaned(ctx, l.id)
		return
	}
	if sess.State != api.AgentSessionStateRunning {
		return
	}

	m.logUndelivered(ctx, l.id, "agent process exited mid-turn")
	if l.reviewID != "" {
		m.finishReviewer(l, "the reviewer's process exited mid-turn")
		return
	}
	if l.stage != "" {
		err = m.store.DropStageOutput(ctx, l.id)
		if err != nil {
			slog.Error("drop stage output", "session", l.id, "err", err)
		}
	}
	m.endMergeTurn(l.id, true)
	reason := "agent process exited mid-turn"
	if interrupted {
		reason = "interrupted; the agent didn't answer, so byre ended its process"
	}
	end, err := m.store.EndTurn(ctx, l.id, agentActor, map[string]any{"error": reason, "stopped": true}, nil, false, false, false)
	if err != nil {
		slog.Error("record session exit", "session", l.id, "err", err)
	}
	if end.Discard != store.DiscardNone {
		return
	}
	if l.owner == "" && !interrupted {
		m.recordError(l, "The agent process exited mid-turn. Send a message to resume it.")
	}
}

func (m *Manager) handle(ctx context.Context, l *live, ev claude.Event) error {
	switch ev.Type {
	case "assistant":
		blocks, err := ev.Blocks()
		if err != nil {
			return err
		}
		parent, err := ev.ParentToolUseID()
		if err != nil {
			return err
		}
		uuid, err := ev.UUID()
		if err != nil {
			return err
		}
		for _, b := range blocks {
			err = m.handleAssistantBlock(ctx, l, b, parent, uuid)
			if err != nil {
				return err
			}
		}
		if parent != "" {
			return nil
		}
		usage, err := ev.ContextUsage()
		if err != nil {
			return err
		}
		if usage.Tokens == 0 {
			return nil
		}
		l.contextModel = usage.Model
		return m.updateContext(ctx, l, usage.Tokens, l.contextWindow)

	case "user":
		blocks, err := ev.Blocks()
		if err != nil {
			return err
		}
		parent, err := ev.ParentToolUseID()
		if err != nil {
			return err
		}
		uuid, err := ev.UUID()
		if err != nil {
			return err
		}
		for _, b := range blocks {
			if b.Type != "tool_result" {
				continue
			}
			content, err := decode(b.Content)
			if err != nil {
				return err
			}
			err = m.emit(ctx, l, agentActor, "agent.tool_result", withParent(map[string]any{
				"tool_use_id": b.ToolUseID,
				"is_error":    b.IsError,
				"content":     content,
			}, parent, uuid))
			if err != nil {
				return err
			}
		}
		return nil

	case "result":
		res, err := ev.Result()
		if err != nil {
			return err
		}
		m.mu.Lock()
		interrupted := l.interrupted
		m.mu.Unlock()
		if res.NumTurns == 0 && !res.IsError && res.Result == "" && !interrupted {
			slog.Info("ignored result with no model turns", "session", l.id, "subtype", res.Subtype, "duration_ms", res.DurationMS)
			break
		}
		return m.turnEnded(ctx, l, endResult{
			subtype: res.Subtype,
			isError: res.IsError,
			result:  res.Result,
			metadata: map[string]any{
				"duration_ms":    res.DurationMS,
				"num_turns":      res.NumTurns,
				"total_cost_usd": res.TotalCostUSD,
				"modelUsage":     store.ToPayload(res.ModelUsage),
			},
			window: res.ModelUsage[l.contextModel].ContextWindow,
		})

	case "stream_event":
		text, err := ev.TextDelta()
		if err != nil || text == "" {
			return err
		}
		m.store.Broadcast(m.event(l, agentActor, "agent.delta", map[string]any{"text": text}))
		return nil

	case "rate_limit_event":
		info := struct {
			RateLimitInfo map[string]any `json:"rate_limit_info"`
		}{}
		err := json.Unmarshal(ev.Raw, &info)
		if err != nil {
			return err
		}

		m.mu.Lock()
		if info.RateLimitInfo["status"] == "rejected" {
			l.rateLimited = info.RateLimitInfo
		}
		m.mu.Unlock()
		m.setVendorUsage(l.vendor, claudeUsage(info.RateLimitInfo))

	case "system":
		if ev.Subtype == "init" || progressSubtypes[ev.Subtype] {
			return nil
		}

	case "tool_progress":
		return nil
	}

	raw := map[string]any{}
	err := json.Unmarshal(ev.Raw, &raw)
	if err != nil {
		return err
	}
	err = m.emit(ctx, l, agentActor, "agent.vendor", raw)
	if err != nil {
		return err
	}
	return m.trackSubagents(ctx, l, raw)
}

func (m *Manager) trackSubagents(ctx context.Context, l *live, raw map[string]any) error {
	if raw["type"] != "system" {
		return nil
	}
	id, _ := raw["task_id"].(string)
	n := len(l.subagents)
	switch raw["subtype"] {
	case "task_started":
		if raw["task_type"] == "local_agent" {
			l.subagents[id] = true
		}
	case "task_notification":
		delete(l.subagents, id)
	}
	if len(l.subagents) == n {
		return nil
	}
	return m.store.SetSubagents(ctx, l.id, len(l.subagents), agentActor)
}

type endResult struct {
	subtype  string
	isError  bool
	result   string
	metadata map[string]any
	window   int
}

func (r endResult) payload() map[string]any {
	payload := map[string]any{}
	for key, value := range r.metadata {
		payload[key] = value
	}
	payload["subtype"] = r.subtype
	payload["is_error"] = r.isError
	payload["result"] = r.result
	return payload
}

func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
	go m.readUsage(l.vendor)
	if l.owner != "" {
		return m.leadTurnEnded(ctx, l, res)
	}
	if l.reviewID != "" {
		return m.reviewTurnEnded(ctx, l, res)
	}

	m.mu.Lock()
	interrupted := l.interrupted
	rateLimited := l.rateLimited
	compacting := l.compacting
	asked := l.asked
	m.mu.Unlock()

	finished := !res.isError && !interrupted && strings.TrimSpace(res.result) != ""

	payload := res.payload()
	if l.member != "" {
		payload["member"] = l.member
	}
	if interrupted {
		payload["interrupted"] = true
	}
	owner := m.ownsTask(l.id)
	if owner {
		defer m.endMergeTurn(l.id, false)
	}
	landingBy, err := m.store.LandingBy(ctx, l.taskID, l.id)
	if err != nil {
		return err
	}
	if (owner || landingBy != "") && !finished {
		err := git.AbortMerge(ctx, l.dir)
		if err != nil {
			slog.Error("abort unfinished merge", "session", l.id, "err", err)
		}
	}
	unlock := func() {}
	if !owner {
		locked := m.lockTask(l.taskID)
		once := sync.OnceFunc(locked)
		unlock = once
		defer once()
		task, err := m.store.GetTask(ctx, l.taskID)
		if err != nil {
			return err
		}
		if task.Status != api.TaskStatusActive {
			return nil
		}
	}
	sha, err := git.CommitAll(ctx, l.dir, commitMessage(res), l.gitName, l.gitEmail)
	if err != nil {
		slog.Error("commit turn", "session", l.id, "err", err)
		m.recordError(l, fmt.Sprintf("byre couldn't commit the turn's work, so the turn counts as unfinished: %v", err))
		finished = false
		if owner {
			err = git.AbortMerge(ctx, l.dir)
			if err != nil {
				slog.Error("abort unfinished merge", "session", l.id, "err", err)
			}
		}
	}
	handoff := compacting && finished
	if sha != "" {
		payload["commit"] = sha
	}
	discard, err := m.store.GetDiscard(ctx, l.taskID)
	if err != nil {
		return err
	}
	merging := false
	if finished && !handoff {
		merging, err = m.store.MergeTurnPending(ctx, l.taskID)
		if err != nil {
			return err
		}
	}
	snapshot := discard.State == store.DiscardNone
	review := snapshot && !merging
	switch {
	case handoff:
		payload["handoff"] = true
	case finished && !compacting && !asked && review:
		payload["summary"] = true
	}
	go m.closeMergedSubtasks(l.taskID, l.dir)
	if payload["summary"] == true && snapshot {
		go m.reportToParent(l.taskID, res.result)
	}

	if res.window > 0 {
		err = m.updateContext(ctx, l, l.contextUsed, res.window)
		if err != nil {
			return err
		}
	}

	failure := fmt.Sprintf("The turn ended with an error (%s): %s", res.subtype, res.result)
	switch {
	case !res.isError || interrupted:
	case rateLimited != nil:
		m.recordRateLimited(l, rateLimited, failure)
	default:
		m.recordError(l, failure)
	}

	stageFinished := false
	if l.stage != "" {
		if finished && !handoff && !merging {
			stageFinished, err = m.store.FinishStage(ctx, l.id)
		}
		if err == nil && !stageFinished {
			err = m.store.DropStageOutput(ctx, l.id)
		}
		if err == nil && stageFinished {
			err = m.recordStage(ctx, l)
		}
		if err != nil {
			return err
		}
	}
	unlock()

	held := len(l.subagents) > 0 || interrupted
	var state map[string]any
	if interrupted || res.isError {
		state = map[string]any{"stopped": true}
	}
	turn := m.event(l, agentActor, "turn.ended", payload)
	end, err := m.store.EndTurn(ctx, l.id, agentActor, state, &turn, review && !held, finished && !handoff, handoff)
	if err != nil {
		return err
	}
	if snapshot && !held {
		m.snapshotSoon(liveTarget(l), api.SnapshotTriggerTurnEnd, 0, review, false)
	}

	if end.Discard != store.DiscardNone {
		m.logUndelivered(ctx, l.id, "turn ended in discard")
		return nil
	}
	if handoff {
		m.logUndelivered(ctx, l.id, "turn ended in handoff")
		go m.continueFromHandoff(l.id, res.result)
		return nil
	}
	if end.FinalizeBy != "" {
		m.logUndelivered(ctx, l.id, "turn ended in merge")
		return nil
	}
	if stageFinished {
		return nil
	}
	go m.afterTurn(context.WithoutCancel(ctx), l)
	return nil
}

func (m *Manager) afterTurn(ctx context.Context, l *live) {
	task, err := m.store.GetTask(ctx, l.taskID)
	if err != nil {
		slog.Error("read task after turn", "task", l.taskID, "err", err)
		return
	}
	if task.Merging || task.Discarding {
		m.logUndelivered(ctx, l.id, "task is merging or being discarded")
		return
	}
	m.deliverTeam(ctx, l)
	if m.interrupted(l.id) {
		m.logUndelivered(ctx, l.id, "turn was interrupted")
		return
	}
	err = m.Deliver(ctx, l.id)
	if err != nil {
		slog.Error("deliver inbox responses", "session", l.id, "err", err)
	}
}

func commitMessage(res endResult) string {
	line, _, _ := strings.Cut(strings.TrimSpace(res.result), "\n")
	sentence, _, found := strings.Cut(line, ". ")
	if found {
		sentence += "."
	}
	sentence = strings.TrimSpace(strings.ReplaceAll(sentence, "**", ""))
	if sentence == "" {
		return fmt.Sprintf("Work from a turn that ended without a summary (%s)", res.subtype)
	}
	runes := []rune(sentence)
	if len(runes) > 100 {
		return string(runes[:99]) + "…"
	}
	return sentence
}

func (m *Manager) updateContext(ctx context.Context, l *live, used, window int) error {
	if used == l.contextUsed && window == l.contextWindow {
		return nil
	}
	l.contextUsed = used
	l.contextWindow = window
	return m.emit(ctx, l, agentActor, "session.context", map[string]any{"model": l.contextModel, "used": used, "window": window})
}

func (m *Manager) handleAssistantBlock(ctx context.Context, l *live, b claude.Block, parent, uuid string) error {
	switch b.Type {
	case "text":
		return m.emit(ctx, l, agentActor, "agent.message", withParent(map[string]any{"text": b.Text}, parent, uuid))

	case "thinking":
		if b.Thinking == "" {
			return nil
		}
		return m.emit(ctx, l, agentActor, "agent.thinking", withParent(map[string]any{"text": b.Thinking}, parent, uuid))

	case "tool_use":
		input, err := decode(b.Input)
		if err != nil {
			return err
		}
		return m.emit(ctx, l, agentActor, "agent.tool_call", withParent(map[string]any{
			"id":    b.ID,
			"name":  b.Name,
			"input": input,
		}, parent, uuid))
	}

	return m.emit(ctx, l, agentActor, "agent.vendor", withParent(store.ToPayload(b), parent, uuid))
}

func withParent(payload map[string]any, parent, uuid string) map[string]any {
	if parent != "" {
		payload["parent_tool_use_id"] = parent
		return payload
	}
	if uuid != "" {
		payload["vendor_uuid"] = uuid
	}
	return payload
}

func (m *Manager) emit(ctx context.Context, l *live, actor, kind string, payload map[string]any) error {
	return m.store.AppendEvent(ctx, m.event(l, actor, kind, payload))
}

func (m *Manager) event(l *live, actor, kind string, payload map[string]any) api.Event {
	return newEvent(l.id, l.owner, l.projectID, l.taskID, actor, kind, payload)
}

func newEvent(sessionID, owner, projectID, taskID, actor, kind string, payload map[string]any) api.Event {
	ev := api.Event{
		Kind:           kind,
		AgentSessionId: &sessionID,
		Actor:          actor,
		Payload:        payload,
	}
	if owner != "" {
		ev.Owner = &owner
		return ev
	}
	ev.ProjectId = &projectID
	ev.TaskId = &taskID
	return ev
}

func decode(raw json.RawMessage) (any, error) {
	if len(raw) == 0 {
		return nil, nil
	}
	var v any
	err := json.Unmarshal(raw, &v)
	return v, err
}