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/misc/byre/internal/api"
	"g.fc.run/misc/byre/internal/claude"
	"g.fc.run/misc/byre/internal/git"
	"g.fc.run/misc/byre/internal/mcp"
	"g.fc.run/misc/byre/internal/store"
	"g.fc.run/misc/byre/internal/workflow"
)

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
	codexHome    string
	status       func(context.Context) (api.Status, error)
	addProject   func(ctx context.Context, name, repoPath, createdBy string) (api.Project, error)

	mu           sync.Mutex
	live         map[string]*live
	starting     map[string]*start
	closing      bool
	rateLimits   map[api.AgentSessionVendor]map[string]any
	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

	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

	workflowMu sync.Mutex

	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
	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, codexHome string, status func(context.Context) (api.Status, error), addProject func(ctx context.Context, name, repoPath, createdBy string) (api.Project, error)) *Manager {
	m := &Manager{
		store:        st,
		worktreeRoot: worktreeRoot,
		leadRoot:     leadRoot,
		codexHome:    codexHome,
		status:       status,
		addProject:   addProject,
		live:         map[string]*live{},
		starting:     map[string]*start{},
		deliverLocks: map[string]*sessionLock{},
		rateLimits:   map[api.AgentSessionVendor]map[string]any{},
		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{},
		signals:      make(chan []api.Event, 1024),
		snaps:        map[string]*snapState{},
		merges:       map[string]*projectMerge{},
		bad:          newBadLines(),
		watchStop:    make(chan struct{}),
	}
	go m.signalLoop()
	go m.readCodexRateLimit()
	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 string, images ...string) (api.AgentSession, error) {
	return m.create(ctx, task, creator, prompt, model, creator.Login, images)
}

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

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

	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
	}

	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
		}

		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, 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())
	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.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.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]map[string]any
}

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

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

func (m *Manager) launch(ctx context.Context, sess api.AgentSession, task api.Task, actor, text string, images []string) (api.AgentSession, 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 api.AgentSession{}, err
	}

	imgs, err := m.prepareMessage(ctx, l, actor, text, images)
	if err != nil {
		m.failStart(ctx, sess.Id, actor, err)
		return api.AgentSession{}, err
	}

	err = l.sess.send(ctx, text, imgs)
	if err != nil {
		return api.AgentSession{}, err
	}

	return m.store.GetSession(ctx, sess.Id)
}

func (m *Manager) failStart(ctx context.Context, sessionID, actor string, err error) {
	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, l *live, actor, text string, images []string) error {
	imgs, err := m.prepareMessage(ctx, l, actor, text, images)
	if err != nil {
		return err
	}
	return l.sess.send(ctx, text, imgs)
}

func (m *Manager) prepareMessage(ctx context.Context, l *live, actor, text string, images []string) ([]image, error) {
	imgs := []image{}
	for _, id := range images {
		b, data, err := m.store.ReadBlob(ctx, id)
		if err != nil {
			return nil, err
		}
		imgs = append(imgs, image{mediaType: b.ContentType, data: data})
	}

	payload := map[string]any{"text": text}
	if l.implementer() {
		head, err := git.Head(l.dir)
		if err != nil {
			return nil, 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
	}
	err := m.emit(ctx, l, actor, "user.message", payload)
	if err != nil {
		return nil, err
	}

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

	return imgs, nil
}

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,
		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.setMergeTitleTool(l), m.stopMergeTool(l), m.setTaskSummaryTool(l), m.searchClosedTool(l), m.exposePortTool(l), m.runOnClientTool(l), m.takeSnapshotTool(l)}, m.subtaskTools(l)...)
	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))
	}
	if sess.Role == api.Implementer && task.WorkflowStage != nil {
		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.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 = m.leadTools(l)
		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)
	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)

	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)
		}
	}
	next, _, err := m.store.EndTurn(ctx, l.id, agentActor, map[string]any{"error": "agent process exited mid-turn"}, 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 == "" {
		m.abandonMerge(l.taskID, l.projectID)
		m.fileError(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
		}

		addClaudeWindowMins(info.RateLimitInfo)

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

	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 {
	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
	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.fileError(l, fmt.Sprintf("byre couldn't commit the turn's work, so the turn counts as unfinished: %v", err))
		finished = false
	}
	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.fileRateLimited(l, rateLimited, failure)
	default:
		m.fileError(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 {
			return err
		}
	}

	held := len(l.subagents) > 0
	turn := m.event(l, agentActor, "turn.ended", payload)
	next, finalizeBy, err := m.store.EndTurn(ctx, l.id, agentActor, nil, &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")
		go m.finishMerge(l, finalizeBy)
		return nil
	}
	mergeBy := m.takeMerge(l.taskID)
	switch {
	case mergeBy != "":
		go m.releaseMerge(l.projectID, l.taskID)
		if !res.isError || interrupted {
			m.fileError(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
	}
	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 {
	ev := api.Event{
		Kind:           kind,
		AgentSessionId: &l.id,
		Actor:          actor,
		Payload:        payload,
	}
	if l.owner != "" {
		ev.Owner = &l.owner
		return ev
	}
	ev.ProjectId = &l.projectID
	ev.TaskId = &l.taskID
	return ev
}

func addClaudeWindowMins(info map[string]any) {
	windows, _ := info["unifiedWindows"].(map[string]any)
	for name, w := range windows {
		window, ok := w.(map[string]any)
		if !ok {
			continue
		}
		switch {
		case strings.HasPrefix(name, "five_hour"):
			window["windowMins"] = 5 * 60
		case strings.HasPrefix(name, "seven_day"):
			window["windowMins"] = 7 * 24 * 60
		}
	}
}

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