ai/byre
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

Bold questions

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-14 19:35:34 -07:00
parent ab4c3d2c6f
commit 1ea055682b
54 changed files with 3696 additions and 762 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/agents/manager.go
+157 -50
View File
@@ -7,13 +7,14 @@ import (
"errors"
"fmt"
"log/slog"
"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/report"
"g.fc.run/misc/byre/internal/store"
)
@@ -37,8 +38,9 @@ var DefaultLimits = Limits{
}
type Manager struct {
store *store.Store
limits Limits
store *store.Store
limits Limits
worktreeRoot string
mu sync.Mutex
live map[string]*live
@@ -51,13 +53,19 @@ type Manager struct {
type live struct {
sess *claude.Session
gate *report.Gate
id string
dir string
model string
gitName string
gitEmail string
vendor api.AgentSessionVendor
taskID string
projectID string
stopping bool
interrupted bool
asked bool
bounced bool
compacting bool
rateLimited map[string]any
contextModel string
@@ -65,16 +73,21 @@ type live struct {
contextWindow int
}
func New(st *store.Store, limits Limits) *Manager {
func New(st *store.Store, limits Limits, worktreeRoot string) *Manager {
return &Manager{
store: st,
limits: limits,
live: map[string]*live{},
rateLimits: map[api.AgentSessionVendor]map[string]any{},
store: st,
limits: limits,
worktreeRoot: worktreeRoot,
live: map[string]*live{},
rateLimits: map[api.AgentSessionVendor]map[string]any{},
}
}
func (m *Manager) Create(ctx context.Context, task api.Task, creator api.User, prompt, model string) (api.AgentSession, error) {
return m.create(ctx, task, creator, prompt, model, creator.Login)
}
func (m *Manager) create(ctx context.Context, task api.Task, creator api.User, prompt, model, actor string) (api.AgentSession, error) {
if creator.GitName == "" || creator.GitEmail == "" {
return api.AgentSession{}, ErrNoGitIdentity
}
@@ -111,10 +124,10 @@ func (m *Manager) Create(ctx context.Context, task api.Task, creator api.User, p
return api.AgentSession{}, err
}
return m.turn(ctx, sess, task, creator.Login, prompt)
return m.turn(ctx, sess, task, actor, prompt)
}
func (m *Manager) Send(ctx context.Context, sessionID, actor, text string) (api.AgentSession, error) {
func (m *Manager) Send(ctx context.Context, sessionID, actor, text, model string) (api.AgentSession, error) {
sess, err := m.store.GetSession(ctx, sessionID)
if err != nil {
return api.AgentSession{}, err
@@ -124,6 +137,14 @@ func (m *Manager) Send(ctx context.Context, sessionID, actor, text string) (api.
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.store.GetTask(ctx, sess.TaskId)
if err != nil {
return api.AgentSession{}, err
@@ -175,26 +196,7 @@ func (m *Manager) StopTask(ctx context.Context, taskID, actor string) error {
if sess.State == api.AgentSessionStateCompleted {
continue
}
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
}
}
err = m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateCompleted, actor, nil)
err = m.retire(ctx, sess, actor)
if err != nil {
return err
}
@@ -203,6 +205,28 @@ func (m *Manager) StopTask(ctx context.Context, taskID, actor string) error {
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 {
@@ -325,6 +349,9 @@ func (m *Manager) start(ctx context.Context, sess api.AgentSession, task api.Tas
}
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 {
@@ -341,6 +368,22 @@ func (m *Manager) start(ctx context.Context, sess api.AgentSession, task api.Tas
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 {
@@ -425,17 +468,23 @@ func transition(ctx context.Context, st *store.Store, id string, from, to api.Ag
}
func (m *Manager) message(ctx context.Context, l *live, actor, text string) error {
err := m.emit(ctx, l, actor, "user.message", map[string]any{"text": text})
head, err := git.Head(l.dir)
if err != nil {
return err
}
err = m.emit(ctx, l, actor, "user.message", map[string]any{"text": text, "head": head})
if err != nil {
return err
}
m.mu.Lock()
l.interrupted = false
l.asked = false
l.bounced = false
l.compacting = actor == "byre" && text == handoffPrompt
l.rateLimited = nil
m.mu.Unlock()
l.gate.Reset()
return l.sess.Send(text)
}
@@ -457,24 +506,27 @@ func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task ap
return nil, err
}
l = &live{id: sess.Id, vendor: sess.Vendor, taskID: task.Id, projectID: task.ProjectId}
l.gate = &report.Gate{OnReport: func(r report.Report) {
payload := store.ToPayload(r)
payload["inbox_item_id"] = m.fileReport(l, r)
err := m.emit(context.Background(), l, agentActor, "agent.report", payload)
if err != nil {
slog.Error("record report", "session", l.id, "err", 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,
}
opts := claude.Options{
Cwd: task.WorktreePath,
Model: sess.Model,
GitName: creator.GitName,
GitEmail: creator.GitEmail,
MCPServers: []*mcp.Server{{Name: "byre", Version: "0.0.0", Tools: []mcp.Tool{l.gate.Tool(), m.askTool(l)}}},
Guidance: guidance,
MCPServers: []*mcp.Server{{Name: "byre", Version: "0.0.0", Tools: []mcp.Tool{m.askTool(l), m.createTaskTool(l)}}},
Hooks: map[string][]claude.HookMatcher{
"Stop": {{Func: l.gate.StopHook}},
"UserPromptSubmit": {{Func: remindGuidance}},
"Stop": {{Func: m.bounceProseQuestion(l)}},
},
}
started, err := m.store.SessionStarted(ctx, sess.Id)
@@ -593,7 +645,31 @@ func (m *Manager) handle(ctx context.Context, l *live, ev claude.Event) error {
if err != nil {
return err
}
err = m.emit(ctx, l, agentActor, "turn.ended", store.ToPayload(res))
m.mu.Lock()
interrupted := l.interrupted
rateLimited := l.rateLimited
compacting := l.compacting
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
}
switch {
case handoff:
payload["handoff"] = true
case finished && !compacting:
payload["inbox_item_id"] = m.fileSummary(l, res.Result)
}
err = m.emit(ctx, l, agentActor, "turn.ended", payload)
if err != nil {
return err
}
@@ -606,11 +682,6 @@ func (m *Manager) handle(ctx context.Context, l *live, ev claude.Event) error {
}
}
m.mu.Lock()
interrupted := l.interrupted
rateLimited := l.rateLimited
m.mu.Unlock()
failure := fmt.Sprintf("The turn ended with an error (%s): %s", res.Subtype, res.Result)
switch {
case !res.IsError || interrupted:
@@ -629,12 +700,31 @@ func (m *Manager) handle(ctx context.Context, l *live, ev claude.Event) error {
if err != nil {
slog.Error("start queued turns", "err", err)
}
if handoff {
go m.continueFromHandoff(l.id, res.Result)
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(api.Event{
Kind: "agent.delta",
ProjectId: &l.projectID,
TaskId: &l.taskID,
AgentSessionId: &l.id,
Actor: agentActor,
Payload: map[string]any{"text": text},
})
return nil
case "rate_limit_event":
info := struct {
RateLimitInfo map[string]any `json:"rate_limit_info"`
@@ -665,6 +755,23 @@ func (m *Manager) handle(ctx context.Context, l *live, ev claude.Event) error {
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