byre/internal/agents/manager.go

package agents

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

	"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/mcp"
	"g.fc.run/ai/byre/internal/store"
	"g.fc.run/ai/byre/internal/terminal"
	"g.fc.run/ai/byre/internal/workflow"
	"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."

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
	leadAnswered  map[string]bool
	paneFocus     map[string]api.PaneFocus
	paneReported  map[string]time.Time
	paneMoved     map[string]time.Time
	humanMoved    map[string]time.Time
	moveRefused   map[string]bool
	startedTodo   map[string]time.Time
	startRefused  map[string]bool
	finishing     map[string]bool
	handoffModel  map[string]string
	mergeTurns    map[string]mergeTurn
	landing       map[string]landing

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

	snapMu sync.Mutex
	snaps  map[string]*snapState

	mergeMu sync.Mutex
	merges  map[string]*projectMerge

	reviewMu sync.Mutex

	usageMu map[api.AgentSessionVendor]*sync.Mutex

	workflowMu sync.Mutex

	auditMu 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
	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

	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: {}},
		leadAnswered:     map[string]bool{},
		paneFocus:        map[string]api.PaneFocus{},
		paneReported:     map[string]time.Time{},
		paneMoved:        map[string]time.Time{},
		humanMoved:       map[string]time.Time{},
		moveRefused:      map[string]bool{},
		startedTodo:      map[string]time.Time{},
		startRefused:     map[string]bool{},
		finishing:        map[string]bool{},
		handoffModel:     map[string]string{},
		mergeTurns:       map[string]mergeTurn{},
		landing:          map[string]landing{},
		signals:          make(chan []api.Event, 1024),
		snaps:            map[string]*snapState{},
		merges:           map[string]*projectMerge{},
		previewListeners: map[int]*previewListener{},
		bad:              newBadLines(),
		watchStop:        make(chan struct{}),
	}
	go m.signalLoop()
	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)
}

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

	ctx = context.WithoutCancel(ctx)
	prior, sent, err := m.priorSend(ctx, 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, prompt, model, sendID, images...)
	}
	return m.create(ctx, task, creator, "", prompt, model, creator.Login, sendID, images)
}

func (m *Manager) create(ctx context.Context, task api.Task, creator api.User, member, prompt, model, actor, sendID string, images []string) (api.AgentSession, error) {
	if creator.GitName == "" || creator.GitEmail == "" {
		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, model)
	if err != nil {
		return api.AgentSession{}, err
	}
	return m.turn(ctx, sess, task, actor, prompt, sendID, images)
}

func (m *Manager) newSession(ctx context.Context, task api.Task, creator api.User, member, model string) (api.AgentSession, error) {
	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)
		}
	}

	id := newUUID()
	vendor := workflow.VendorOf(model)
	sess := api.AgentSession{
		Id:              id,
		TaskId:          task.Id,
		Vendor:          vendor,
		Model:           model,
		VendorSessionId: initialVendorSessionID(vendor, id),
		State:           api.AgentSessionStateIdle,
		Role:            api.Implementer,
		Member:          member,
		CreatedBy:       creator.Login,
		CreatedAt:       time.Now().UTC(),
	}

	err = m.store.CreateSession(ctx, sess, task.ProjectId)
	return sess, err
}

func (m *Manager) Send(ctx context.Context, sessionID, actor, text, model, sendID string, images ...string) (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)
	}

	if model != "" && model != sess.Model {
		err = checkModel(sess, model)
		if err != nil {
			return api.AgentSession{}, err
		}
		err = m.store.SetSessionModel(ctx, sess.Id, model, actor)
		if err != nil {
			return api.AgentSession{}, err
		}
		sess.Model = model
	}

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

	return m.turn(ctx, sess, task, actor, text, sendID, images)
}

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 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
		}
		err = m.retire(ctx, sess, actor)
		if err != nil {
			return err
		}
	}

	return nil
}

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

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) Recover(ctx context.Context) error {
	err := m.recoverMerges(ctx)
	if err != nil {
		return err
	}

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

	staged, err := m.recoverStages(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
	}
	for _, sess := range stale {
		err = m.store.SetSubagents(ctx, sess.Id, 0, "byre")
		if err != nil {
			return err
		}
	}

	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 {
		if staged[sess.Id] {
			continue
		}
		task, err := m.taskFor(ctx, sess)
		if err != nil {
			slog.Error("resume session: load task", "session", sess.Id, "task", sess.TaskId, "err", err)
			continue
		}

		pending, err := m.store.SessionPending(ctx, sess.Id)
		if err != nil {
			return err
		}
		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 {
				continue
			}
		}

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

		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 {
			ferr := m.failResume(ctx, sess, task, err)
			if ferr != nil {
				return ferr
			}
			continue
		}
		slog.Info("resumed session", "session", sess.Id, "task", task.Id, "owner", sess.Owner)
		m.logUndelivered(ctx, sess.Id, "resumed with restart notice")
	}

	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.recoverReviews(ctx)
	if err != nil {
		return err
	}

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

	go m.deliverAllLeads()
	go m.deliverAllQueued(context.Background())
	go m.startHeldMerges(context.Background())
	go m.recoverMergeAfterReviews(context.Background())
	go m.auditAll(context.Background())
	return nil
}

func (m *Manager) failResume(ctx context.Context, sess api.AgentSession, task api.Task, err error) error {
	slog.Error("resume session", "session", sess.Id, "err", err)
	m.endMergeTurn(sess.Id, true)
	m.logUndelivered(ctx, sess.Id, "resume failed")
	derr := m.store.DropStageOutput(ctx, sess.Id)
	if derr != nil {
		return derr
	}
	next, _, serr := m.store.EndTurn(ctx, sess.Id, "byre", map[string]any{"error": err.Error()}, nil, false, false)
	if serr != nil {
		return serr
	}
	switch {
	case next == store.DiscardClosing:
		go m.finishDiscard(task.Id)
	case next == store.DiscardHeld:
		go m.startDiscard(task.Id)
	case sess.Role == api.Implementer && m.leadsTask(ctx, task.Id, sess.Id):
		m.abandonMerge(task.Id, task.ProjectId)
	}
	return nil
}

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, text, sendID string, images []string) (api.AgentSession, error) {
	err := m.startRunning(ctx, sess, task.Id, actor)
	if err != nil {
		return api.AgentSession{}, err
	}
	return m.launch(ctx, sess, task, actor, text, sendID, images)
}

func (m *Manager) startRunning(ctx context.Context, sess api.AgentSession, taskID, actor string) error {
	held := fmt.Errorf("%w: held while this task merges and continues", ErrConflict)
	if m.heldForLanding(taskID, sess.Id) {
		return held
	}
	err := transition(ctx, m.store, sess.Id, api.AgentSessionStateIdle, api.AgentSessionStateRunning, actor)
	if err != nil {
		return err
	}
	if m.heldForLanding(taskID, sess.Id) {
		err = transition(ctx, m.store, sess.Id, api.AgentSessionStateRunning, api.AgentSessionStateIdle, actor)
		if err != nil {
			return err
		}
		return held
	}
	return nil
}

func (m *Manager) launch(ctx context.Context, sess api.AgentSession, task api.Task, actor, text, sendID string, images []string) (api.AgentSession, error) {
	imgs, compacting, err := m.recordMessage(ctx, sess, task, actor, text, sendID, images)
	if err != nil {
		m.failStart(ctx, sess.Id, actor, err)
		return api.AgentSession{}, err
	}
	err = m.deliver(ctx, sess, task, actor, 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, text string, imgs []image, compacting bool) error {
	l, err := m.ensureLive(ctx, sess, task)
	if err == nil {
		err = m.syncModel(ctx, l, sess.Model)
	}
	if err != nil {
		m.failStart(ctx, sess.Id, actor, err)
		return err
	}

	m.beginTurn(l, compacting)
	err = m.send(ctx, l, agentText(actor, 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)
	ids := []string{}
	raw, _ := ev.Payload["images"].([]any)
	for _, id := range raw {
		s, _ := id.(string)
		ids = append(ids, s)
	}
	imgs, err := m.readImages(ctx, ids)
	if err != nil {
		m.failStart(ctx, sess.Id, ev.Actor, err)
		return true, err
	}
	return true, m.deliver(ctx, sess, task, ev.Actor, text, imgs, isHandoffPrompt(ev.Actor, text))
}

func (m *Manager) readImages(ctx context.Context, ids []string) ([]image, error) {
	imgs := []image{}
	for _, id := range ids {
		b, data, err := m.store.ReadBlob(ctx, id)
		if err != nil {
			return nil, err
		}
		imgs = append(imgs, 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()

	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 string) error {
	if model == l.model {
		return nil
	}

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

	err := l.sess.setModel(cctx, model)
	if err != nil {
		return fmt.Errorf("set model %s: %w", model, err)
	}
	l.model = model
	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, text, "", images)
	if err != nil {
		return err
	}
	m.beginTurn(l, compacting)
	return m.send(ctx, l, agentText(actor, text), imgs)
}

func (m *Manager) send(ctx context.Context, l *live, text string, imgs []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, text, sendID string, images []string) ([]image, bool, error) {
	implementer := sess.Role == api.Implementer
	if implementer && isMergeInPrompt(actor, text) {
		m.ownMerge(sess.Id, task.Id, task.WorktreePath)
	}
	imgs, err := m.readImages(ctx, images)
	if err != nil {
		return nil, false, err
	}

	payload := map[string]any{"text": text}
	if implementer {
		head, err := git.Head(task.WorktreePath)
		if err != nil {
			return nil, false, err
		}
		payload["head"] = head
	}
	if len(images) > 0 {
		payload["images"] = images
	}
	compacting := isHandoffPrompt(actor, text)
	if compacting {
		payload["handoff"] = true
	}
	if isContinuation(actor, text) {
		payload["continuation"] = true
	}
	if isFindingsMessage(actor, text) {
		payload["findings"] = true
	}
	if sendID != "" {
		payload["send_id"] = sendID
	}
	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, compacting, nil
}

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,
		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.setTaskTitleTool(l), m.setMergeTitleTool(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()}, 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))
	}
	env := portEnv(task)
	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
		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)}
		env = nil
		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(m.leadTools(l), m.readKnowledgeTool())
		env = nil
	}
	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
	quiet := closing || l.stopping
	m.mu.Unlock()

	if len(l.subagents) > 0 && !closing {
		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 quiet {
		m.logUndelivered(ctx, l.id, "agent process ended while closing or 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.AgentSessionStateRunning {
		return
	}

	m.logUndelivered(ctx, l.id, "agent process exited mid-turn")
	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)
	next, _, err := m.store.EndTurn(ctx, l.id, agentActor, map[string]any{"error": "agent process exited mid-turn", "stopped": true}, nil, false, false)
	if err != nil {
		slog.Error("record session exit", "session", l.id, "err", err)
	}
	if l.reviewID != "" {
		m.finishReviewer(l, "the reviewer's process exited mid-turn")
		return
	}
	switch next {
	case store.DiscardClosing:
		m.finishDiscard(l.taskID)
		return
	case store.DiscardHeld:
		m.startDiscard(l.taskID)
		return
	}
	if l.owner == "" {
		if m.leadsTask(ctx, l.taskID, l.id) {
			m.abandonMerge(l.taskID, l.projectID)
		}
		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,
			payload: store.ToPayload(res),
			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" {
			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
	payload map[string]any
	window  int
}

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)
	}

	defer m.endLanding(l.taskID, l.id)

	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
	}
	owner := m.ownsTask(l.id)
	keep := false
	if owner {
		defer func() {
			if !keep {
				m.endMergeTurn(l.id, false)
			}
		}()
	}
	if (owner || m.landingBy(l.taskID, l.id) != "") && !finished {
		err := git.AbortMerge(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(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(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
	}
	mergingBy := ""
	if finished && !handoff {
		mergingBy, err = m.store.MergingBy(ctx, l.taskID)
		if err != nil {
			return err
		}
	}
	snapshot := discard.State == store.DiscardNone
	review := snapshot && mergingBy == ""
	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 && mergingBy == "" {
			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)
	next, finalizeBy, err := m.store.EndTurn(ctx, l.id, agentActor, state, &turn, review && !held, finished && !handoff)
	if err != nil {
		return err
	}
	if snapshot && !held {
		m.snapshotSoon(liveTarget(l), api.SnapshotTriggerTurnEnd, 0, review, false)
	}

	switch next {
	case store.DiscardClosing:
		m.logUndelivered(ctx, l.id, "turn ended in discard")
		go m.finishDiscard(l.taskID)
		return nil
	case store.DiscardHeld:
		go m.nextMerge(l.projectID)
		if stageFinished {
			go m.advanceStage(l)
			return nil
		}
		go m.startDiscard(l.taskID)
		return nil
	}
	if handoff {
		m.logUndelivered(ctx, l.id, "turn ended in handoff")
		go m.continueFromHandoff(l.id, res.result)
		return nil
	}
	if finalizeBy != "" {
		m.logUndelivered(ctx, l.id, "turn ended in merge")
		keep = owner
		go m.finishMerge(l, finalizeBy)
		return nil
	}
	mergeBy := ""
	if m.leadsTask(ctx, l.taskID, l.id) {
		mergeBy = m.takeMerge(l.taskID)
	}
	switch {
	case mergeBy != "":
		go m.releaseMerge(l.projectID, l.taskID)
		if !res.isError || interrupted {
			m.recordError(l, "The Merge turn ended unfinished (interrupted, or with no final message), so byre didn't squash or push the task. Merge again to retry.")
		}
	default:
		go m.nextMerge(l.projectID)
	}
	if stageFinished {
		go m.advanceStage(l)
		return nil
	}
	go m.afterTurn(context.WithoutCancel(ctx), l)
	return nil
}

func (m *Manager) afterTurn(ctx context.Context, l *live) {
	discard, err := m.store.GetDiscard(ctx, l.taskID)
	if err != nil {
		slog.Error("read discard state", "task", l.taskID, "err", err)
		return
	}
	if discard.State == store.DiscardHeld {
		m.startDiscard(l.taskID)
		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
}

func newUUID() string {
	b := make([]byte, 16)
	rand.Read(b)
	b[6] = b[6]&0x0f | 0x40
	b[8] = b[8]&0x3f | 0x80
	return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:])
}