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

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 Limits struct {
	Global    int
	PerVendor map[api.AgentSessionVendor]int
}

var DefaultLimits = Limits{
	Global:    8,
	PerVendor: map[api.AgentSessionVendor]int{api.Claude: 6},
}

type Manager struct {
	store        *store.Store
	limits       Limits
	worktreeRoot string
	leadRoot     string

	mu           sync.Mutex
	live         map[string]*live
	closing      bool
	rateLimits   map[api.AgentSessionVendor]map[string]any
	leadAnswered map[string]bool
	paneFocused  map[string]bool

	deliverMu sync.Mutex
	capMu     sync.Mutex
	leadMu    sync.Mutex
	signals   chan api.Event

	snapMu sync.Mutex
	snaps  map[string]*snapState

	bad       *badLines
	watchStop chan struct{}
}

type live struct {
	sess        *claude.Session
	id          string
	dir         string
	model       string
	gitName     string
	gitEmail    string
	vendor      api.AgentSessionVendor
	taskID      string
	projectID   string
	owner       string
	stopping    bool
	interrupted bool
	asked       bool
	askedBlock  bool
	bounced     bool
	compacting  bool
	wantCompact bool
	rateLimited map[string]any

	contextModel  string
	contextUsed   int
	contextWindow int
}

func New(st *store.Store, limits Limits, worktreeRoot, leadRoot string) *Manager {
	m := &Manager{
		store:        st,
		limits:       limits,
		worktreeRoot: worktreeRoot,
		leadRoot:     leadRoot,
		live:         map[string]*live{},
		rateLimits:   map[api.AgentSessionVendor]map[string]any{},
		leadAnswered: map[string]bool{},
		paneFocused:  map[string]bool{},
		signals:      make(chan api.Event, 1024),
		snaps:        map[string]*snapState{},
		bad:          newBadLines(),
		watchStop:    make(chan struct{}),
	}
	go m.signalLoop()
	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)
	}

	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()
	sess := api.AgentSession{
		Id:              id,
		TaskId:          task.Id,
		Vendor:          api.Claude,
		Model:           model,
		VendorSessionId: id,
		State:           api.AgentSessionStateIdle,
		Role:            api.Implementer,
		CreatedBy:       creator.Login,
		CreatedAt:       time.Now().UTC(),
	}

	err = m.store.CreateSession(ctx, sess, task.ProjectId)
	if err != nil {
		return api.AgentSession{}, err
	}

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

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

	if sess.State == api.AgentSessionStateQueued {
		return m.cancelQueued(ctx, sess)
	}

	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
	m.mu.Unlock()

	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 m.drain(ctx)
}

func (m *Manager) retire(ctx context.Context, sess api.AgentSession, actor string) error {
	m.mu.Lock()
	l := m.live[sess.Id]
	if l != nil {
		l.stopping = true
	}
	m.mu.Unlock()

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

	if sess.State == api.AgentSessionStateQueued {
		err := m.store.DeleteQueuedTurn(ctx, sess.Id)
		if err != nil {
			return err
		}
	}

	return m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateCompleted, actor, nil)
}

func (m *Manager) Recover(ctx context.Context) error {
	sessions, err := m.store.SessionsInState(ctx, api.AgentSessionStateRunning)
	if err != nil {
		return err
	}

	for _, sess := range sessions {
		task, err := m.taskFor(ctx, sess)
		if err != nil {
			return err
		}

		l, err := m.ensureLive(ctx, sess, task)
		if err != nil {
			slog.Error("resume session", "session", sess.Id, "err", err)
			m.logUndelivered(ctx, sess.Id, "resume failed")
			err = m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateIdle, "byre", map[string]any{"error": err.Error()})
			if err != nil {
				return err
			}
			if sess.Owner == "" {
				m.takeMerge(task.Id)
			}
			continue
		}

		err = m.message(ctx, l, "byre", restartNotice, nil)
		if err != nil {
			return err
		}
		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")
	}

	go m.deliverAllLeads()
	return m.drain(ctx)
}

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)
	}
	m.mu.Unlock()

	for _, l := range all {
		l.sess.Kill()
	}

	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
	Queued     []api.TurnRef
	Caps       api.Caps
	RateLimits map[string]map[string]any
}

func (m *Manager) Usage(ctx context.Context) (Usage, error) {
	u := Usage{
		Running:    []api.TurnRef{},
		Queued:     []api.TurnRef{},
		Caps:       api.Caps{Global: m.limits.Global, PerVendor: map[string]int{}},
		RateLimits: map[string]map[string]any{},
	}
	for vendor, n := range m.limits.PerVendor {
		u.Caps.PerVendor[string(vendor)] = n
	}

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

	queued, err := m.store.QueuedTurns(ctx)
	if err != nil {
		return u, err
	}
	for _, q := range queued {
		u.Queued = append(u.Queued, api.TurnRef{SessionId: q.SessionID, TaskId: q.TaskID, Vendor: string(q.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) {
	if sess.Owner != "" {
		return m.start(ctx, sess, task, actor, text, images, api.AgentSessionStateIdle)
	}

	m.capMu.Lock()
	defer m.capMu.Unlock()

	room, err := m.hasRoom(ctx, sess.Vendor)
	if err != nil {
		return api.AgentSession{}, err
	}

	if room {
		return m.start(ctx, sess, task, actor, text, images, api.AgentSessionStateIdle)
	}

	err = m.store.EnqueueTurn(ctx, store.QueuedTurn{
		SessionID: sess.Id,
		Vendor:    sess.Vendor,
		Actor:     actor,
		Text:      text,
		Images:    images,
		CreatedAt: time.Now().UTC(),
	})
	if err != nil {
		return api.AgentSession{}, err
	}

	err = transition(ctx, m.store, sess.Id, api.AgentSessionStateIdle, api.AgentSessionStateQueued, actor)
	if err != nil {
		derr := m.store.DeleteQueuedTurn(ctx, sess.Id)
		if derr != nil {
			slog.Error("remove queued turn", "session", sess.Id, "err", derr)
		}
		return api.AgentSession{}, err
	}

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

func (m *Manager) start(ctx context.Context, sess api.AgentSession, task api.Task, actor, text string, images []string, from api.AgentSessionState) (api.AgentSession, error) {
	err := transition(ctx, m.store, sess.Id, from, api.AgentSessionStateRunning, actor)
	if err != nil {
		return api.AgentSession{}, err
	}

	l, err := m.ensureLive(ctx, sess, task)
	if err == nil {
		err = m.syncModel(ctx, l, sess.Model)
	}
	if err != nil {
		serr := m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateIdle, actor, map[string]any{"error": err.Error()})
		if serr != nil {
			slog.Error("record failed start", "session", sess.Id, "err", serr)
		}
		return api.AgentSession{}, err
	}

	err = m.message(ctx, l, actor, text, images)
	if err != nil {
		return api.AgentSession{}, err
	}

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

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.Control(cctx, map[string]any{"subtype": "set_model", "model": model})
	if err != nil {
		return fmt.Errorf("set model %s: %w", model, err)
	}
	l.model = model
	return nil
}

func (m *Manager) hasRoom(ctx context.Context, vendor api.AgentSessionVendor) (bool, error) {
	counts, err := m.store.CountRunning(ctx)
	if err != nil {
		return false, err
	}

	total := 0
	for _, n := range counts {
		total += n
	}

	return total < m.limits.Global && counts[vendor] < m.limits.PerVendor[vendor], nil
}

func (m *Manager) drain(ctx context.Context) error {
	m.capMu.Lock()
	defer m.capMu.Unlock()

	queued, err := m.store.QueuedTurns(ctx)
	if err != nil {
		return err
	}

	for _, q := range queued {
		room, err := m.hasRoom(ctx, q.Vendor)
		if err != nil {
			return err
		}
		if !room {
			continue
		}

		sess, err := m.store.GetSession(ctx, q.SessionID)
		if err != nil {
			return err
		}
		task, err := m.store.GetTask(ctx, sess.TaskId)
		if err != nil {
			return err
		}

		err = m.store.DeleteQueuedTurn(ctx, q.SessionID)
		if err != nil {
			return err
		}

		_, err = m.start(ctx, sess, task, q.Actor, q.Text, q.Images, api.AgentSessionStateQueued)
		if err != nil {
			slog.Error("start queued turn", "session", q.SessionID, "err", err)
		}
	}

	return nil
}

func (m *Manager) cancelQueued(ctx context.Context, sess api.AgentSession) (api.AgentSession, error) {
	m.capMu.Lock()
	defer m.capMu.Unlock()

	queued, err := m.store.QueuedTurns(ctx)
	if err != nil {
		return api.AgentSession{}, err
	}
	for _, q := range queued {
		if q.SessionID == sess.Id && strings.HasPrefix(q.Text, "Responses from your inbox:") {
			slog.Info("cancelled queued turn carried inbox responses", "session", sess.Id, "text", q.Text)
		}
	}

	err = m.store.DeleteQueuedTurn(ctx, sess.Id)
	if errors.Is(err, store.ErrNotFound) {
		return api.AgentSession{}, fmt.Errorf("%w: session %s has no queued turn", ErrConflict, sess.Id)
	}
	if err != nil {
		return api.AgentSession{}, err
	}

	err = transition(ctx, m.store, sess.Id, api.AgentSessionStateQueued, api.AgentSessionStateIdle, "byre")
	if err != nil {
		return api.AgentSession{}, err
	}

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

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 := []claude.Image{}
	for _, id := range images {
		b, data, err := m.store.ReadBlob(ctx, id)
		if err != nil {
			return err
		}
		imgs = append(imgs, claude.Image{MediaType: b.ContentType, Data: data})
	}

	payload := map[string]any{"text": text}
	if l.owner == "" {
		head, err := git.Head(l.dir)
		if err != nil {
			return err
		}
		payload["head"] = head
	}
	if len(images) > 0 {
		payload["images"] = images
	}
	compacting := isHandoffPrompt(actor, text)
	if compacting {
		payload["handoff"] = true
	}
	err := m.emit(ctx, l, actor, "user.message", payload)
	if err != nil {
		return err
	}

	m.mu.Lock()
	l.interrupted = false
	l.asked = false
	l.askedBlock = false
	l.bounced = false
	l.wantCompact = false
	l.compacting = compacting
	l.rateLimited = nil
	m.mu.Unlock()

	return l.sess.Send(text, imgs...)
}

func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task api.Task) (*live, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.closing {
		return nil, errors.New("shutting down")
	}

	l := m.live[sess.Id]
	if l != nil {
		return l, nil
	}

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

	g := taskGuidance(task)
	tools := []mcp.Tool{m.askTool(l), m.cancelQuestionTool(l), m.createTaskTool(l), m.setMergeTitleTool(l)}
	hooks := map[string][]claude.HookMatcher{
		"UserPromptSubmit": {{Func: remindGuidance(g)}},
		"Stop":             {{Func: m.bounceProseQuestion(l)}},
		"PostToolUse":      {{Func: m.afterTool(l)}},
	}
	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
		tools = m.leadTools(l)
		hooks = map[string][]claude.HookMatcher{"UserPromptSubmit": {{Func: remindGuidance(g)}}}
	}

	opts := claude.Options{
		Cwd:        l.dir,
		Model:      sess.Model,
		GitName:    creator.GitName,
		GitEmail:   creator.GitEmail,
		Guidance:   g,
		MCPServers: []*mcp.Server{{Name: "byre", Version: "0.0.0", Tools: tools}},
		Hooks:      hooks,
		OnBadLine:  func(line []byte, reason string) { m.recordBadLine(l, line, reason) },
	}
	started, err := m.store.SessionStarted(ctx, sess.Id)
	if err != nil {
		return nil, err
	}
	if started {
		opts.Resume = sess.VendorSessionId
	} else {
		opts.SessionID = sess.VendorSessionId
	}

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

	l.sess, err = claude.Start(sctx, opts)
	if err != nil {
		return nil, fmt.Errorf("start claude: %w", err)
	}

	m.live[sess.Id] = l
	go m.pump(l)
	if l.owner == "" {
		go m.snapshotTimer(l)
	}
	return l, nil
}

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

	for ev := range l.sess.Events() {
		start := time.Now()
		if wait := start.Sub(ev.Read); wait > 200*time.Millisecond {
			slog.Info("agent event waited", "session", l.id, "type", ev.Type, "wait", wait)
		}
		err := m.handle(ctx, l, ev)
		if err != nil {
			slog.Error("record agent event", "session", l.id, "type", ev.Type, "err", err)
		}
		if isDecodeError(err) {
			m.recordBadLine(l, ev.Raw, fmt.Sprintf("decode %s: %s", ev.Type, err))
		}
		if took := time.Since(start); took > 200*time.Millisecond {
			slog.Info("agent event slow", "session", l.id, "type", ev.Type, "took", took)
		}
	}

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

	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")
	err = m.store.SetSessionState(ctx, l.id, api.AgentSessionStateIdle, agentActor, map[string]any{"error": "agent process exited mid-turn"})
	if err != nil {
		slog.Error("record session exit", "session", l.id, "err", err)
	}
	if l.owner == "" {
		m.takeMerge(l.taskID)
		m.fileError(l, "The agent process exited mid-turn. Send a message to resume it.")
	}

	err = m.drain(ctx)
	if err != nil {
		slog.Error("start queued turns", "err", err)
	}
}

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
		}
		for _, b := range blocks {
			err = m.handleAssistantBlock(ctx, l, b)
			if err != nil {
				return err
			}
		}
		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
		}
		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", map[string]any{
				"tool_use_id": b.ToolUseID,
				"is_error":    b.IsError,
				"content":     content,
			})
			if err != nil {
				return err
			}
		}
		return nil

	case "result":
		res, err := ev.Result()
		if err != nil {
			return err
		}
		if l.owner != "" {
			return m.leadTurnEnded(ctx, l, res)
		}

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

		finished := !res.IsError && !interrupted && strings.TrimSpace(res.Result) != ""
		handoff := compacting && finished

		payload := store.ToPayload(res)
		sha, err := git.CommitAll(l.dir, commitMessage(res), l.gitName, l.gitEmail)
		if err != nil {
			slog.Error("commit turn", "session", l.id, "err", err)
		}
		if sha != "" {
			payload["commit"] = sha
		}
		m.snapshotSoon(l, api.SnapshotTriggerTurnEnd)
		switch {
		case handoff:
			payload["handoff"] = true
		case finished && !compacting && !askedBlock:
			payload["inbox_item_id"] = m.fileSummary(l, res.Result)
		}
		err = m.emit(ctx, l, agentActor, "turn.ended", payload)
		if err != nil {
			return err
		}

		window := res.ModelUsage[l.contextModel].ContextWindow
		if window > 0 {
			err = m.updateContext(ctx, l, l.contextUsed, 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)
		}

		err = m.store.SetSessionState(ctx, l.id, api.AgentSessionStateIdle, agentActor, nil)
		if err != nil {
			return err
		}

		err = m.drain(ctx)
		if err != nil {
			slog.Error("start queued turns", "err", err)
		}
		if handoff {
			m.logUndelivered(ctx, l.id, "turn ended in handoff")
			go m.continueFromHandoff(l.id, res.Result)
			return nil
		}
		mergeBy := m.takeMerge(l.taskID)
		if mergeBy != "" && finished {
			m.logUndelivered(ctx, l.id, "turn ended in merge")
			go m.finishMerge(l, mergeBy)
			return nil
		}
		err = m.Deliver(ctx, l.id)
		if err != nil {
			slog.Error("deliver inbox responses", "session", l.id, "err", err)
		}
		return nil

	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()
		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
	}
	return m.emit(ctx, l, agentActor, "agent.vendor", raw)
}

func commitMessage(res claude.Result) 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) error {
	switch b.Type {
	case "text":
		return m.emit(ctx, l, agentActor, "agent.message", map[string]any{"text": b.Text})

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

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

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

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