From 7aff773b722b07728c9944b30f109d78b0ea8af2 Mon Sep 17 00:00:00 2001 From: Ian Gulliver Date: Fri, 23 Oct 2026 16:21:56 -0700 Subject: [PATCH] Simplify codebase per audit of 461aac7 --- api/openapi.yaml | 2 +- cmd/byre/serve.go | 48 +++++++-------- cmd/byre/tool.go | 6 +- internal/agents/compact.go | 15 +---- internal/agents/critical.go | 4 -- internal/agents/inbox.go | 2 +- internal/agents/manager.go | 25 ++++---- internal/agents/mergein.go | 17 ++---- internal/agents/queue.go | 13 ++--- internal/agents/ready.go | 15 ++--- internal/agents/review.go | 10 +--- internal/agents/subtasks.go | 12 ++-- internal/agents/workflows.go | 2 +- internal/api/api.gen.go | 9 +-- internal/store/coordinate.go | 2 +- .../migrations/112-add-task-input-kind.sql | 3 + internal/store/schedule.go | 22 ++++--- internal/store/store.go | 28 +++++---- internal/store/taskinputs.go | 22 +++++-- internal/store/workflow.go | 58 +++++++++++++------ mac/Sources/Byre/Forwards.swift | 16 +++-- mac/Sources/Byre/Panes.swift | 48 +-------------- mac/Sources/Byre/Workflows.swift | 2 +- 23 files changed, 180 insertions(+), 201 deletions(-) create mode 100644 internal/store/migrations/112-add-task-input-kind.sql diff --git a/api/openapi.yaml b/api/openapi.yaml index 430f418..2b24482 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -1486,7 +1486,7 @@ paths: operationId: deleteWorkflow responses: "204": - description: Deleted. Tasks that froze it keep their copy. + description: Deleted, with the schedules that ran it. Tasks that froze it keep their copy. "404": { $ref: "#/components/responses/Error" } components: diff --git a/cmd/byre/serve.go b/cmd/byre/serve.go index c3d2104..6535e3d 100644 --- a/cmd/byre/serve.go +++ b/cmd/byre/serve.go @@ -5,7 +5,6 @@ import ( "crypto/rand" "crypto/tls" "errors" - "flag" "fmt" "log/slog" "net" @@ -25,54 +24,49 @@ import ( ) const ( - defaultListen = "127.0.0.1:8270" - loopbackURL = "http://" + defaultListen + loopbackAddr = "127.0.0.1:8270" + loopbackURL = "http://" + loopbackAddr ) -func serveCmd(args []string) { +func dataDir() string { configDir, err := os.UserConfigDir() if err != nil { fatal("user config dir", "err", err) } + return filepath.Join(configDir, "byre") +} - flags := flag.NewFlagSet("serve", flag.ExitOnError) - dataDir := flags.String("data", filepath.Join(configDir, "byre"), "directory holding byre's database, worktrees, leads and blobs") - listenAddr := flags.String("listen", defaultListen, "loopback address for server status and client commands") - hostname := flags.String("ts-hostname", "byre-"+macSlug(), "hostname of byre's tailnet node") - flags.Parse(args) - - host, _, err := net.SplitHostPort(*listenAddr) - if err != nil { - fatal("parse --listen", "err", err) - } - ip := net.ParseIP(host) - if ip == nil || !ip.IsLoopback() { - fatal("--listen must be a loopback address", "addr", *listenAddr) +func serveCmd(args []string) { + if len(args) != 0 { + fmt.Fprintln(os.Stderr, "usage: byre serve") + os.Exit(2) } + dataDir := dataDir() + hostname := "byre-" + macSlug() // Protect local credentials before tsnet can write state or a login URL. - err = os.MkdirAll(*dataDir, 0o700) + err := os.MkdirAll(dataDir, 0o700) if err != nil { fatal("create data directory", "err", err) } - err = os.Chmod(*dataDir, 0o700) + err = os.Chmod(dataDir, 0o700) if err != nil { fatal("protect data directory", "err", err) } // Keep ownership until shutdown; never unlink the lock file, since a new // inode would let another process lock the same data directory concurrently. - lock, err := os.OpenFile(filepath.Join(*dataDir, "serve.lock"), os.O_CREATE|os.O_RDWR, 0o600) + lock, err := os.OpenFile(filepath.Join(dataDir, "serve.lock"), os.O_CREATE|os.O_RDWR, 0o600) if err != nil { fatal("open service lock", "err", err) } defer lock.Close() err = syscall.Flock(int(lock.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) if err != nil { - fatal("acquire service lock", "data", *dataDir, "err", err) + fatal("acquire service lock", "data", dataDir, "err", err) } token := rand.Text() - err = os.WriteFile(filepath.Join(*dataDir, "loopback-token"), []byte(token), 0o600) + err = os.WriteFile(filepath.Join(dataDir, "loopback-token"), []byte(token), 0o600) if err != nil { fatal("write loopback credential", "err", err) } @@ -100,8 +94,8 @@ func serveCmd(args []string) { logFile.Close() ts := &tsnet.Server{ - Dir: filepath.Join(*dataDir, "tsnet"), - Hostname: *hostname, + Dir: filepath.Join(dataDir, "tsnet"), + Hostname: hostname, UserLogf: tsnetLogf(), } err = ts.Start() @@ -110,7 +104,7 @@ func serveCmd(args []string) { } defer ts.Close() - srv, err := server.New(*dataDir, ts, token) + srv, err := server.New(dataDir, ts, token) if err != nil { fatal("start server", "err", err) } @@ -124,7 +118,7 @@ func serveCmd(args []string) { go updateDaily(ctx) base := func(net.Listener) context.Context { return ctx } - loopback := &http.Server{Addr: *listenAddr, Handler: srv.LoopbackHandler(), BaseContext: base} + loopback := &http.Server{Addr: loopbackAddr, Handler: srv.LoopbackHandler(), BaseContext: base} tailnet := &http.Server{Handler: srv.Handler(), BaseContext: base, ConnContext: peer.ConnContext} go func() { @@ -137,7 +131,7 @@ func serveCmd(args []string) { go serveTailnet(ctx, ts, tailnet) - slog.Info("listening", "addr", *listenAddr, "hostname", *hostname, "data", *dataDir) + slog.Info("listening", "addr", loopbackAddr, "hostname", hostname, "data", dataDir) err = loopback.ListenAndServe() if !errors.Is(err, http.ErrServerClosed) { fatal("serve", "err", err) diff --git a/cmd/byre/tool.go b/cmd/byre/tool.go index 406c656..5ca0d62 100644 --- a/cmd/byre/tool.go +++ b/cmd/byre/tool.go @@ -81,11 +81,7 @@ func installCommand(name string) string { } func codexHome() string { - configDir, err := os.UserConfigDir() - if err != nil { - fatal("user config dir", "err", err) - } - return filepath.Join(configDir, "byre", "codex") + return filepath.Join(dataDir(), "codex") } func useCodexHome() { diff --git a/internal/agents/compact.go b/internal/agents/compact.go index b6f4d5a..9bba748 100644 --- a/internal/agents/compact.go +++ b/internal/agents/compact.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "log/slog" - "strings" "g.fc.run/ai/byre/internal/api" "g.fc.run/ai/byre/internal/store" @@ -59,7 +58,7 @@ func (m *Manager) launchCompaction(ctx context.Context, sess api.AgentSession, t m.mu.Lock() m.handoffModel[sess.Id] = modelChoice{model, effort} m.mu.Unlock() - return m.launch(ctx, sess, task, "byre", Prompt{Text: prompt}) + return m.launch(ctx, sess, task, "byre", Prompt{Text: prompt, Kind: store.MessageHandoff}) } func (m *Manager) continuationModel(sess api.AgentSession) modelChoice { @@ -73,14 +72,6 @@ func (m *Manager) continuationModel(sess api.AgentSession) modelChoice { return c } -func isHandoffPrompt(actor, text string) bool { - return actor == "byre" && (text == handoffPrompt || text == leadHandoffPrompt) -} - -func isContinuation(actor, text string) bool { - return actor == "byre" && (strings.HasPrefix(text, handoffPreamble) || strings.HasPrefix(text, leadHandoffPreamble)) -} - func (m *Manager) continueFromHandoff(sessionID, handoff string) { ctx := context.Background() m.waitResumed(sessionID) @@ -111,7 +102,7 @@ func (m *Manager) handoff(ctx context.Context, sessionID, handoff string) error if err != nil { return err } - _, err = m.createLead(ctx, creator, "byre", Prompt{Text: leadHandoffPreamble + handoff, Model: model.model, Effort: model.effort}) + _, err = m.createLead(ctx, creator, "byre", Prompt{Text: leadHandoffPreamble + handoff, Kind: store.MessageContinuation, Model: model.model, Effort: model.effort}) return err } @@ -132,6 +123,6 @@ func (m *Manager) handoff(ctx context.Context, sessionID, handoff string) error return err } m.stopLive(sess.Id) - _, err = m.launch(ctx, next, task, "byre", Prompt{Text: handoffPreamble + handoff}) + _, err = m.launch(ctx, next, task, "byre", Prompt{Text: handoffPreamble + handoff, Kind: store.MessageContinuation}) return err } diff --git a/internal/agents/critical.go b/internal/agents/critical.go index 97819e7..3aa055a 100644 --- a/internal/agents/critical.go +++ b/internal/agents/critical.go @@ -24,10 +24,6 @@ func (m *Manager) ownsTask(sessionID string) bool { return found } -func (m *Manager) takeMergeTurn(l *live) { - m.ownMerge(l.id, l.taskID, l.dir) -} - func (m *Manager) ownMerge(sessionID, taskID, dir string) { if m.ownsTask(sessionID) { return diff --git a/internal/agents/inbox.go b/internal/agents/inbox.go index a497322..58f6ff0 100644 --- a/internal/agents/inbox.go +++ b/internal/agents/inbox.go @@ -410,7 +410,7 @@ func (m *Manager) deliverMidTurn(ctx context.Context, l *live) midTurn { inputs, held := m.takeQueuedMidTurn(ctx, l) msgs := []queued{} for _, in := range inputs { - msgs = append(msgs, queued{actor: in.Actor, text: in.Text, sendID: in.SendID, client: in.Client}) + msgs = append(msgs, queued{actor: in.Actor, text: in.Text, kind: in.Kind, sendID: in.SendID, client: in.Client}) } mt := midTurn{taskInputs: inputs} parts, hinted := m.queuedParts(ctx, l, msgs, held) diff --git a/internal/agents/manager.go b/internal/agents/manager.go index 93897f8..dd6e689 100644 --- a/internal/agents/manager.go +++ b/internal/agents/manager.go @@ -194,6 +194,7 @@ func (m *Manager) taskFor(ctx context.Context, sess api.AgentSession) (api.Task, type Prompt struct { Text string + Kind store.MessageKind Model string Effort string SendID string @@ -683,12 +684,13 @@ func (m *Manager) deliverRecorded(ctx context.Context, sess api.AgentSession, ta } text, _ := ev.Payload["text"].(string) client, _ := ev.Payload["client"].(string) + handoff, _ := ev.Payload[string(store.MessageHandoff)].(bool) imgs, err := m.readImages(ctx, anyStrings(ev.Payload["images"])) if err != nil { m.failStart(ctx, sess.Id, ev.Actor, err) return true, err } - return true, m.deliver(ctx, sess, task, ev.Actor, client, text, imgs, isHandoffPrompt(ev.Actor, text)) + return true, m.deliver(ctx, sess, task, ev.Actor, client, text, imgs, handoff) } func (m *Manager) readImages(ctx context.Context, ids []string) ([]agentinput.Image, error) { @@ -762,7 +764,7 @@ func (m *Manager) send(ctx context.Context, l *live, text string, imgs []agentin func (m *Manager) recordMessage(ctx context.Context, sess api.AgentSession, task api.Task, actor string, p Prompt) ([]agentinput.Image, bool, error) { implementer := sess.Role == api.Implementer - if implementer && isMergeInPrompt(actor, p.Text) { + if implementer && p.Kind == store.MessageMergeIn { m.ownMerge(sess.Id, task.Id, task.WorktreePath) } imgs, err := m.readImages(ctx, p.Images) @@ -781,16 +783,7 @@ func (m *Manager) recordMessage(ctx context.Context, sess api.AgentSession, task if len(p.Images) > 0 { payload["images"] = p.Images } - compacting := isHandoffPrompt(actor, p.Text) - if compacting { - payload["handoff"] = true - } - if isContinuation(actor, p.Text) { - payload["continuation"] = true - } - if isFindingsMessage(actor, p.Text) { - payload["findings"] = true - } + markKind(payload, p.Kind) if p.SendID != "" { payload["send_id"] = p.SendID } @@ -801,7 +794,13 @@ func (m *Manager) recordMessage(ctx context.Context, sess api.AgentSession, task if err != nil { return nil, false, err } - return imgs, compacting, nil + return imgs, p.Kind == store.MessageHandoff, nil +} + +func markKind(payload map[string]any, kind store.MessageKind) { + if kind != "" { + payload[string(kind)] = true + } } func (m *Manager) beginTurn(l *live, compacting bool) { diff --git a/internal/agents/mergein.go b/internal/agents/mergein.go index 74ca202..260d8c3 100644 --- a/internal/agents/mergein.go +++ b/internal/agents/mergein.go @@ -18,14 +18,8 @@ import ( const maxMergeTitle = 72 -const mergeInMarker = " into this branch: run `git merge --no-ff --no-commit " - -func isMergeInPrompt(actor, text string) bool { - return actor == "byre" && strings.Contains(text, mergeInMarker) -} - func mergeInPrompt(target string) string { - return fmt.Sprintf("Merge %[1]s"+mergeInMarker+"%[1]s`. "+ + return fmt.Sprintf("Merge %[1]s into this branch: run `git merge --no-ff --no-commit %[1]s`. "+ "Resolve any conflicts keeping both sides' intent; regenerate generated files instead of hand-merging them. "+ "Check that the build and tests pass. Leave the merge uncommitted: byre concludes it when the turn ends. "+ "If resolving the conflicts took substantive changes rather than combining both sides as written, stop there: "+ @@ -49,7 +43,7 @@ func (m *Manager) headStartMerge(ctx context.Context, task api.Task, user api.Us } prompt := mergeInPrompt(target) + fmt.Sprintf(" This task's Merge is held while another task merges into %s; this turn gets a head start, and byre merges %s in again when the held merge starts. If you stop for substantive changes, call stop_merge first, so byre drops the held merge and nothing is pushed before the review.", target, target) go func() { - _, err := m.launch(context.Background(), sess, task, "byre", Prompt{Text: prompt}) + _, err := m.launch(context.Background(), sess, task, "byre", Prompt{Text: prompt, Kind: store.MessageMergeIn}) if err != nil { slog.Error("start held merge head start", "task", task.Id, "err", err) } @@ -61,10 +55,11 @@ func (m *Manager) sendMergeIn(ctx context.Context, task api.Task, user api.User, if err != nil { return api.AgentSession{}, err } + p := Prompt{Text: prompt, Kind: store.MessageMergeIn} if found { - return m.Send(ctx, sess.Id, "byre", Prompt{Text: prompt}) + return m.Send(ctx, sess.Id, "byre", p) } - return m.create(ctx, task, user, sess.Member, "byre", Prompt{Text: prompt}) + return m.create(ctx, task, user, sess.Member, "byre", p) } func (m *Manager) Merge(ctx context.Context, task api.Task, project api.Project, user api.User, title string) (api.Task, error) { @@ -184,7 +179,7 @@ func (m *Manager) launchMerge(sess api.AgentSession, task api.Task, project api. } func (m *Manager) deliverMerge(ctx context.Context, sess api.AgentSession, task api.Task, project api.Project, prompt string) error { - _, compacting, err := m.recordMessage(ctx, sess, task, "byre", Prompt{Text: prompt}) + _, compacting, err := m.recordMessage(ctx, sess, task, "byre", Prompt{Text: prompt, Kind: store.MessageMergeIn}) if err != nil { return err } diff --git a/internal/agents/queue.go b/internal/agents/queue.go index 8e2b276..f914b5c 100644 --- a/internal/agents/queue.go +++ b/internal/agents/queue.go @@ -77,7 +77,7 @@ func (m *Manager) Message(ctx context.Context, sessionID, actor string, p Prompt if err != nil { return api.AgentSession{}, false, err } - err = m.store.AddTaskInput(ctx, store.TaskInput{TaskID: sess.TaskId, Member: sess.Member, Actor: actor, Text: p.Text, Images: p.Images, SendID: p.SendID, Client: p.Client}, task.ProjectId, sess.Id) + err = m.store.AddTaskInput(ctx, store.TaskInput{TaskID: sess.TaskId, Member: sess.Member, Actor: actor, Text: p.Text, Kind: p.Kind, Images: p.Images, SendID: p.SendID, Client: p.Client}, task.ProjectId, sess.Id) if err != nil { return api.AgentSession{}, false, err } @@ -157,7 +157,7 @@ func (m *Manager) deliverQueued(ctx context.Context, sess api.AgentSession) (boo } in := inputs[0] - _, err = m.Send(ctx, sess.Id, in.Actor, Prompt{Text: in.Text, SendID: in.SendID, Client: in.Client, Images: in.Images}) + _, err = m.Send(ctx, sess.Id, in.Actor, Prompt{Text: in.Text, Kind: in.Kind, SendID: in.SendID, Client: in.Client, Images: in.Images}) if errors.Is(err, ErrConflict) { slog.Info("queued message held: session changed state", "session", sess.Id, "task", task.Id, "err", err) return true, nil @@ -172,6 +172,7 @@ func (m *Manager) deliverQueued(ctx context.Context, sess api.AgentSession) (boo type queued struct { actor string text string + kind store.MessageKind sendID string client string } @@ -289,13 +290,11 @@ func (m *Manager) imagesHint(l *live, held heldImages) string { func (m *Manager) recordQueued(ctx context.Context, l *live, msgs []queued) string { b := &strings.Builder{} for i, q := range msgs { - if l.implementer() && isMergeInPrompt(q.actor, q.text) { - m.takeMergeTurn(l) + if l.implementer() && q.kind == store.MessageMergeIn { + m.ownMerge(l.id, l.taskID, l.dir) } payload := map[string]any{"text": q.text, "mid_turn": true} - if isFindingsMessage(q.actor, q.text) { - payload["findings"] = true - } + markKind(payload, q.kind) if q.sendID != "" { payload["send_id"] = q.sendID } diff --git a/internal/agents/ready.go b/internal/agents/ready.go index 6601fc1..0ccc1ab 100644 --- a/internal/agents/ready.go +++ b/internal/agents/ready.go @@ -8,6 +8,7 @@ import ( "log/slog" "strings" + "g.fc.run/ai/byre/internal/api" "g.fc.run/ai/byre/internal/mcp" "g.fc.run/ai/byre/internal/store" "g.fc.run/ai/byre/internal/workflow" @@ -38,8 +39,8 @@ func finishStagePrompt(frozen store.FrozenWorkflow, stage string) string { } type readyArgs struct { - Overview string `json:"overview"` - Merge string `json:"merge"` + Overview string `json:"overview"` + Merge api.TaskReadyMerge `json:"merge"` } func (m *Manager) readyForMergeTool(l *live) mcp.Tool { @@ -63,9 +64,9 @@ func (m *Manager) readyForMergeTool(l *live) mcp.Tool { return "", fmt.Errorf("%w: overview must not be empty", ErrInvalid) } if args.Merge == "" { - args.Merge = "complete" + args.Merge = api.ReadyMergeComplete } - if args.Merge != "complete" && args.Merge != "continue" { + if !args.Merge.Valid() { return "", fmt.Errorf("%w: merge must be complete or continue", ErrInvalid) } err := m.store.MarkReady(ctx, l.taskID, overview, args.Merge, agentActor) @@ -138,7 +139,7 @@ func (m *Manager) selfClose(c store.SelfClose) { err := m.approveSelfClose(ctx, c) if err != nil { action := "merge" - if c.Ready == "discard" { + if c.Ready == store.ReadyDiscard { action = "discard" } slog.Error("workflow self close", "task", c.TaskID, "ready", c.Ready, "err", err) @@ -157,7 +158,7 @@ func (m *Manager) approveSelfClose(ctx context.Context, c store.SelfClose) error if err != nil || !taken { return err } - if c.Ready == "discard" { + if c.Ready == store.ReadyDiscard { _, err = m.Discard(ctx, task, "byre") return err } @@ -165,7 +166,7 @@ func (m *Manager) approveSelfClose(ctx context.Context, c store.SelfClose) error if err != nil { return err } - if c.Ready == "continue" { + if c.Ready == store.ReadyContinue { _, err = m.MergeToMain(ctx, task, project, user, task.MergeTitle) return err } diff --git a/internal/agents/review.go b/internal/agents/review.go index a91fa72..8ff9ef3 100644 --- a/internal/agents/review.go +++ b/internal/agents/review.go @@ -397,7 +397,7 @@ func (m *Manager) deliverFindings(ctx context.Context, ref store.SnapshotRef, re slog.Info("review findings not delivered, task has no live agent", "task", ref.TaskID, "snapshot", ref.N) return } - _, _, err = m.Message(ctx, s.Id, "byre", Prompt{Text: text}) + _, _, err = m.Message(ctx, s.Id, "byre", Prompt{Text: text, Kind: store.MessageFindings}) if err != nil { slog.Error("deliver review findings", "task", ref.TaskID, "snapshot", ref.N, "session", s.Id, "err", err) return @@ -405,12 +405,6 @@ func (m *Manager) deliverFindings(ctx context.Context, ref store.SnapshotRef, re slog.Info("review findings delivered", "task", ref.TaskID, "snapshot", ref.N, "session", s.Id) } -const findingsPreamble = "The review of snapshot #" - -func isFindingsMessage(actor, text string) bool { - return actor == "byre" && strings.HasPrefix(text, findingsPreamble) -} - func findingsMessage(n int, reviews []api.Review) string { b := &strings.Builder{} for _, r := range reviews { @@ -434,7 +428,7 @@ func findingsMessage(n int, reviews []api.Review) string { if b.Len() == 0 { return "" } - return fmt.Sprintf(findingsPreamble+"%d of your work has finished, and its reviewers raised the findings below. Reviewers can be wrong or miss context: check each one against the code and your brief. "+ + return fmt.Sprintf("The review of snapshot #%d of your work has finished, and its reviewers raised the findings below. Reviewers can be wrong or miss context: check each one against the code and your brief. "+ "Before you change anything, answer each finding with respond_to_finding (snapshot #%d, the reviewer's name and the finding's number): fixing if it holds up and you'll act on it, ignoring with your reason if you're leaving it. "+ "The human sees your answers on the findings and can overrule an ignore.", n, n) + b.String() } diff --git a/internal/agents/subtasks.go b/internal/agents/subtasks.go index 4083f1a..8c63fee 100644 --- a/internal/agents/subtasks.go +++ b/internal/agents/subtasks.go @@ -147,7 +147,7 @@ func (m *Manager) subtaskTools(l *live) []mcp.Tool { if err != nil { return "", err } - err = m.messageTask(ctx, task, creator, agentActor, args.Text) + err = m.messageTask(ctx, task, creator, agentActor, Prompt{Text: args.Text}) if err != nil { return "", err } @@ -261,16 +261,16 @@ func (m *Manager) checkNoSubtasks(ctx context.Context, task api.Task) error { return fmt.Errorf("%w: task %s has %d active subtasks; merge or discard them first: %s", ErrConflict, task.Id, len(subs), strings.Join(names, ", ")) } -func (m *Manager) messageTask(ctx context.Context, task api.Task, user api.User, actor, text string) error { +func (m *Manager) messageTask(ctx context.Context, task api.Task, user api.User, actor string, p Prompt) error { sess, found, err := m.liveSession(ctx, task.Id) if err != nil { return err } if !found { - _, err = m.create(ctx, task, user, sess.Member, actor, Prompt{Text: text}) + _, err = m.create(ctx, task, user, sess.Member, actor, p) return err } - _, _, err = m.Message(ctx, sess.Id, actor, Prompt{Text: text}) + _, _, err = m.Message(ctx, sess.Id, actor, p) return err } @@ -308,7 +308,7 @@ func (m *Manager) MergeIntoParent(ctx context.Context, task api.Task, user api.U } prompt := mergeInPrompt(task.Branch) + fmt.Sprintf(" This merges in your subtask %q (%s); byre closes the subtask once its branch is in this branch's history at the end of a turn.", task.Title, task.Id) - err = m.messageTask(ctx, parent, user, "byre", prompt) + err = m.messageTask(ctx, parent, user, "byre", Prompt{Text: prompt, Kind: store.MessageMergeIn}) if err != nil { cerr := m.store.ClearParentMerge(ctx, task.Id) if cerr != nil { @@ -401,7 +401,7 @@ func (m *Manager) reportToParent(taskID, result string) { return } text := fmt.Sprintf("Your subtask %q (%s, branch %s) finished a turn:\n\n%s", task.Title, task.Id, task.Branch, quote(result)) - err = m.messageTask(ctx, parent, user, "byre", text) + err = m.messageTask(ctx, parent, user, "byre", Prompt{Text: text}) if err != nil { slog.Error("report subtask turn to parent", "task", taskID, "parent", parent.Id, "err", err) } diff --git a/internal/agents/workflows.go b/internal/agents/workflows.go index 543d6e5..b9e5b92 100644 --- a/internal/agents/workflows.go +++ b/internal/agents/workflows.go @@ -914,7 +914,7 @@ func (m *Manager) workflowTools(l *live) []mcp.Tool { }), leadTool(m, l, "delete_workflow", - "Delete a workflow from the server. Tasks that froze it keep their copy. Only when the human asks.", + "Delete a workflow from the server, with the schedules that run it. Tasks that froze it keep their copy. Only when the human asks.", `{"type": "object", "properties": {"name": {"type": "string"}}, "required": ["name"]}`, func(ctx context.Context, user api.User, args struct { Name string `json:"name"` diff --git a/internal/api/api.gen.go b/internal/api/api.gen.go index 30bb17a..cfade1f 100644 --- a/internal/api/api.gen.go +++ b/internal/api/api.gen.go @@ -5271,10 +5271,11 @@ var swaggerSpec = []string{ "lJH+rjopAZ2MqcyUDpwYLIaBurvYm0Tqo9abANdwlKagc37seXSiDJ66tUalOF0QzY7Fd0+/Y9mk6zim", "KXbtRPrxH1SqdPmufdTMefkwUy4KMMu0D9+i6i5nAqdTN6UI4Bja3F0fjV1qQ/QUvXgBLyL/VtHO8sd9", "T2ogxYA13b57YR1HMnecoT9ml0XW0Q6d6zVyfBfOynohfYjxtBt5BzaVlFuJW0dqDBrIycjMBJLHVPNm", - "E9nFPPle6r14iLerV39Jrb6EfjWOxmLRPipWQqu4GGa0kZRk7TqfgCIETj7AuwEbVyp/0iuwu2/5E/Fe", - "kr+MZDII9DAL8l3Yzd2nsG6TZOtL0KyXicpPYV+2Nd2V8EgUDP/7HJqMqAVEUyMXhundfMlxIFsT6dgA", - "/bRP/qVwZz4TwYpnT5/2/EszrIgaDcKO3jhDPCFBluLUMQUsjOcH8woWpxELgq5bj87t4kKhaGroHo61", - "5nrO7Tex+94DWhQ55RaTJLY50u3Bn8IIu7O2fH5Bdwptv1yJnBGJK58aHzB4K+7rt6Z8/Pjx/x8A", + "E9nFPPle6r14iLerV39Jrb6EfjWOxmLRPipWQqu4GGa0kZRk7TqfgCIETj7AuwEbVyp/0iuwu2f5k4xq", + "pVIuhEtOGlSuvJfkTyOZTAK9zIKAF3Zz9yms3SRZ+xI07WW6BaawM9u67sp4JAqH/30OTUfUEqIpkgvH", + "9G7G5FiQrYl0cICe2if/U7hTn4lgxbOnT3v+pxlWRI0HYUdvnCGekKBLceyYIhbG84N5BYvTiAVD161H", + "53dxoVB0NXRPx1p0Pef3m9h97wEtipx2i0kU2xzp9uBfYYTdWV0+vyA8hbZfroTOiASWT40PGNwV9/Vb", + "U05+/Pj/DwA=", } // decodeSpec returns the embedded OpenAPI spec as raw JSON bytes, diff --git a/internal/store/coordinate.go b/internal/store/coordinate.go index 779ca16..992532a 100644 --- a/internal/store/coordinate.go +++ b/internal/store/coordinate.go @@ -109,7 +109,7 @@ func (s *Store) Coordinate(ctx context.Context) (Work, error) { type SelfClose struct { TaskID string - Ready string + Ready Ready By string } diff --git a/internal/store/migrations/112-add-task-input-kind.sql b/internal/store/migrations/112-add-task-input-kind.sql new file mode 100644 index 0000000..2b71481 --- /dev/null +++ b/internal/store/migrations/112-add-task-input-kind.sql @@ -0,0 +1,3 @@ +ALTER TABLE task_inputs ADD COLUMN kind TEXT NOT NULL DEFAULT ''; +UPDATE task_inputs SET kind = 'findings' WHERE actor = 'byre' AND substr(text, 1, 24) = 'The review of snapshot #'; +UPDATE task_inputs SET kind = 'merge_in' WHERE actor = 'byre' AND instr(text, ' into this branch: run `git merge --no-ff --no-commit ') > 0; diff --git a/internal/store/schedule.go b/internal/store/schedule.go index 8af587b..bc7dce9 100644 --- a/internal/store/schedule.go +++ b/internal/store/schedule.go @@ -126,18 +126,22 @@ func (s *Store) DeleteSchedule(ctx context.Context, id, actor string) error { if err != nil { return err } - _, err = tx.ExecContext(ctx, `DELETE FROM schedule_weekdays WHERE schedule_id = ?`, id) - if err != nil { - return err - } - _, err = tx.ExecContext(ctx, `DELETE FROM schedules WHERE id = ?`, id) - if err != nil { - return err - } - return emit(api.Event{Kind: "schedule.deleted", ProjectId: &projectID, Actor: actor, Payload: map[string]any{"id": id}}) + return deleteSchedule(ctx, tx, emit, id, projectID, actor) }) } +func deleteSchedule(ctx context.Context, tx *sql.Tx, emit emitFunc, id, projectID, actor string) error { + _, err := tx.ExecContext(ctx, `DELETE FROM schedule_weekdays WHERE schedule_id = ?`, id) + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `DELETE FROM schedules WHERE id = ?`, id) + if err != nil { + return err + } + return emit(api.Event{Kind: "schedule.deleted", ProjectId: &projectID, Actor: actor, Payload: map[string]any{"id": id}}) +} + func (s *Store) AdvanceSchedule(ctx context.Context, sc api.Schedule) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { list, err := querySchedules(ctx, tx, `WHERE id = ? AND next_run_at = ?`, sc.Id, formatTime(sc.NextRun)) diff --git a/internal/store/store.go b/internal/store/store.go index 3011434..7973e6c 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -723,12 +723,20 @@ func (s *Store) SetTaskSummary(ctx context.Context, id, summary, actor string) e }) } -func (s *Store) MarkReady(ctx context.Context, id, overview, merge, actor string) error { - return s.setReady(ctx, id, actor, merge, "task.ready", map[string]any{"id": id, "overview": overview, "merge": merge}) +type Ready string + +const ( + ReadyComplete = Ready(api.ReadyMergeComplete) + ReadyContinue = Ready(api.ReadyMergeContinue) + ReadyDiscard = Ready("discard") +) + +func (s *Store) MarkReady(ctx context.Context, id, overview string, merge api.TaskReadyMerge, actor string) error { + return s.setReady(ctx, id, actor, Ready(merge), "task.ready", map[string]any{"id": id, "overview": overview, "merge": string(merge)}) } func (s *Store) SuggestDiscard(ctx context.Context, id, reason, actor string) error { - return s.setReady(ctx, id, actor, "discard", "task.discard_suggested", map[string]any{"id": id, "reason": reason}) + return s.setReady(ctx, id, actor, ReadyDiscard, "task.discard_suggested", map[string]any{"id": id, "reason": reason}) } func (s *Store) ClearReady(ctx context.Context, id, actor, reason string) error { @@ -745,10 +753,10 @@ func (s *Store) ClearReady(ctx context.Context, id, actor, reason string) error }) } -func (s *Store) TakeReady(ctx context.Context, id, ready string) (bool, error) { +func (s *Store) TakeReady(ctx context.Context, id string, ready Ready) (bool, error) { taken := false err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET ready = '' WHERE id = ? AND status = ? AND ready = ? RETURNING project_id`, id, string(api.TaskStatusActive), ready) + projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET ready = '' WHERE id = ? AND status = ? AND ready = ? RETURNING project_id`, id, string(api.TaskStatusActive), string(ready)) if errors.Is(err, ErrNotFound) { return nil } @@ -757,7 +765,7 @@ func (s *Store) TakeReady(ctx context.Context, id, ready string) (bool, error) { } taken = true reason := "merge" - if ready == "discard" { + if ready == ReadyDiscard { reason = "discard" } return emit(api.Event{Kind: "task.ready_cleared", ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id, "reason": reason}}) @@ -765,9 +773,9 @@ func (s *Store) TakeReady(ctx context.Context, id, ready string) (bool, error) { return taken, err } -func (s *Store) setReady(ctx context.Context, id, actor, ready, kind string, payload map[string]any) error { +func (s *Store) setReady(ctx context.Context, id, actor string, ready Ready, kind string, payload map[string]any) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET ready = ? WHERE id = ? AND status = ? RETURNING project_id`, ready, id, string(api.TaskStatusActive)) + projectID, err := updatedProject(ctx, tx, `UPDATE tasks SET ready = ? WHERE id = ? AND status = ? RETURNING project_id`, string(ready), id, string(api.TaskStatusActive)) if err != nil { return err } @@ -1717,7 +1725,7 @@ func scanTask(row scanner) (api.Task, error) { parentMergeBy := "" taskState := "" audit := "" - ready := "" + ready := Ready("") participants := "" f := &t.MergeFacts err := row.Scan(&t.Id, &t.ProjectId, &t.Title, &t.MergeTitle, &t.Branch, &t.WorktreePath, &t.BaseRef, &t.Color, &status, &t.CreatedBy, &createdAt, &closedAt, &mergeCommit, &todoItem, &forkedFrom, &forkedAt, &mergeQueuedAt, &workflowName, &workflowStage, &parentTaskID, &parentMergeBy, &t.MergeAfterReview, &taskState, &t.Merging, &t.Discarding, &t.SummarySeq, &audit, &ready, &t.Background, @@ -1730,7 +1738,7 @@ func scanTask(row scanner) (api.Task, error) { return t, err } t.State = api.TaskState(taskState) - if ready == string(api.ReadyMergeComplete) || ready == string(api.ReadyMergeContinue) { + if ready == ReadyComplete || ready == ReadyContinue { m := api.TaskReadyMerge(ready) t.ReadyMerge = &m } diff --git a/internal/store/taskinputs.go b/internal/store/taskinputs.go index 223d21b..8857ae2 100644 --- a/internal/store/taskinputs.go +++ b/internal/store/taskinputs.go @@ -10,12 +10,22 @@ import ( "g.fc.run/ai/byre/internal/ids" ) +type MessageKind string + +const ( + MessageHandoff MessageKind = "handoff" + MessageContinuation MessageKind = "continuation" + MessageFindings MessageKind = "findings" + MessageMergeIn MessageKind = "merge_in" +) + type TaskInput struct { ID int64 TaskID string Member string Actor string Text string + Kind MessageKind Images []string SendID string Client string @@ -35,8 +45,8 @@ func (s *Store) AddTaskInput(ctx context.Context, in TaskInput, projectID, sessi payload["images"] = in.Images } return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { - res, err := tx.ExecContext(ctx, `INSERT INTO task_inputs (task_id, member, actor, text, images, send_id, client_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, - in.TaskID, in.Member, in.Actor, in.Text, string(images), in.SendID, in.Client, now()) + res, err := tx.ExecContext(ctx, `INSERT INTO task_inputs (task_id, member, actor, text, kind, images, send_id, client_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + in.TaskID, in.Member, in.Actor, in.Text, string(in.Kind), string(images), in.SendID, in.Client, now()) if err != nil { return err } @@ -85,7 +95,7 @@ func (s *Store) WithdrawTaskInput(ctx context.Context, taskID string, id int64, } func (s *Store) TaskInputs(ctx context.Context, taskID, member string) ([]TaskInput, error) { - rows, err := s.db.QueryContext(ctx, `SELECT id, task_id, member, actor, text, images, send_id, client_id, created_at FROM task_inputs WHERE task_id = ? AND member = ? ORDER BY id`, taskID, member) + rows, err := s.db.QueryContext(ctx, `SELECT id, task_id, member, actor, text, kind, images, send_id, client_id, created_at FROM task_inputs WHERE task_id = ? AND member = ? ORDER BY id`, taskID, member) if err != nil { return nil, err } @@ -96,7 +106,7 @@ func (s *Store) TaskInputs(ctx context.Context, taskID, member string) ([]TaskIn in := TaskInput{} images := "" createdAt := "" - err = rows.Scan(&in.ID, &in.TaskID, &in.Member, &in.Actor, &in.Text, &images, &in.SendID, &in.Client, &createdAt) + err = rows.Scan(&in.ID, &in.TaskID, &in.Member, &in.Actor, &in.Text, &in.Kind, &images, &in.SendID, &in.Client, &createdAt) if err != nil { return nil, err } @@ -120,8 +130,8 @@ func (s *Store) RestoreTaskInputs(ctx context.Context, inputs []TaskInput) error if err != nil { return err } - _, err = tx.ExecContext(ctx, `INSERT INTO task_inputs (id, task_id, member, actor, text, images, send_id, client_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, - in.ID, in.TaskID, in.Member, in.Actor, in.Text, string(images), in.SendID, in.Client, formatTime(in.CreatedAt)) + _, err = tx.ExecContext(ctx, `INSERT INTO task_inputs (id, task_id, member, actor, text, kind, images, send_id, client_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + in.ID, in.TaskID, in.Member, in.Actor, in.Text, string(in.Kind), string(images), in.SendID, in.Client, formatTime(in.CreatedAt)) if err != nil { return err } diff --git a/internal/store/workflow.go b/internal/store/workflow.go index 358bedd..8b5f4c0 100644 --- a/internal/store/workflow.go +++ b/internal/store/workflow.go @@ -70,24 +70,10 @@ func (s *Store) PutWorkflow(ctx context.Context, name, from string, def api.Work if err != nil { return err } - err = repointProjects(ctx, tx, emit, from, name, actor) + err = repointWorkflow(ctx, tx, emit, from, name, actor) if err != nil { return err } - _, err = tx.ExecContext(ctx, `UPDATE schedules SET workflow = ? WHERE workflow = ?`, name, from) - if err != nil { - return err - } - repointed, err := querySchedules(ctx, tx, `WHERE workflow = ?`, name) - if err != nil { - return err - } - for _, sc := range repointed { - err = emit(api.Event{Kind: "schedule.saved", ProjectId: &sc.ProjectId, Actor: actor, Payload: ToPayload(sc)}) - if err != nil { - return err - } - } err = emit(api.Event{Kind: "workflow.deleted", Actor: actor, Payload: map[string]any{"name": from}}) if err != nil { return err @@ -106,6 +92,44 @@ func (s *Store) PutWorkflow(ctx context.Context, name, from string, def api.Work }) } +func repointWorkflow(ctx context.Context, tx *sql.Tx, emit emitFunc, from, to, actor string) error { + err := repointProjects(ctx, tx, emit, from, to, actor) + if err != nil { + return err + } + err = repointUsers(ctx, tx, emit, from, to, actor) + if err != nil { + return err + } + return repointSchedules(ctx, tx, emit, from, to, actor) +} + +func repointSchedules(ctx context.Context, tx *sql.Tx, emit emitFunc, from, to, actor string) error { + affected, err := querySchedules(ctx, tx, `WHERE workflow = ?`, from) + if err != nil { + return err + } + for _, sc := range affected { + if to == "" { + err = deleteSchedule(ctx, tx, emit, sc.Id, sc.ProjectId, actor) + if err != nil { + return err + } + continue + } + _, err = tx.ExecContext(ctx, `UPDATE schedules SET workflow = ? WHERE id = ?`, to, sc.Id) + if err != nil { + return err + } + sc.Workflow = to + err = emit(api.Event{Kind: "schedule.saved", ProjectId: &sc.ProjectId, Actor: actor, Payload: ToPayload(sc)}) + if err != nil { + return err + } + } + return nil +} + func repointProjects(ctx context.Context, tx *sql.Tx, emit emitFunc, from, to, actor string) error { rows, err := tx.QueryContext(ctx, ` UPDATE projects SET @@ -135,7 +159,7 @@ func repointProjects(ctx context.Context, tx *sql.Tx, emit emitFunc, from, to, a return err } } - return repointUsers(ctx, tx, emit, from, to, actor) + return nil } func repointUsers(ctx context.Context, tx *sql.Tx, emit emitFunc, from, to, actor string) error { @@ -184,7 +208,7 @@ func (s *Store) DeleteWorkflow(ctx context.Context, name, actor string) error { if n == 0 { return fmt.Errorf("%w: no workflow %q", ErrNotFound, name) } - err = repointProjects(ctx, tx, emit, name, "", actor) + err = repointWorkflow(ctx, tx, emit, name, "", actor) if err != nil { return err } diff --git a/mac/Sources/Byre/Forwards.swift b/mac/Sources/Byre/Forwards.swift index 1d8fedf..aa04f9e 100644 --- a/mac/Sources/Byre/Forwards.swift +++ b/mac/Sources/Byre/Forwards.swift @@ -5,14 +5,16 @@ import Observation typealias Preview = Components.Schemas.Preview +struct ForwardedPreview: Hashable { + let taskId: String + let name: String + let port: Int +} + @MainActor @Observable final class Forwards { - private struct Key: Hashable { - let taskId: String - let name: String - let port: Int - } + private typealias Key = ForwardedPreview private let host: String private var localPorts: [Key: Int] = [:] @@ -22,6 +24,10 @@ final class Forwards { self.host = host } + func previews(_ taskId: String) -> [ForwardedPreview] { + localPorts.keys.filter { $0.taskId == taskId }.sorted { ($0.name, $0.port) < ($1.name, $1.port) } + } + func url(_ taskId: String, _ port: Int) -> URL? { guard let (key, local) = localPorts.first(where: { $0.key.taskId == taskId && $0.key.port == port }) else { return nil diff --git a/mac/Sources/Byre/Panes.swift b/mac/Sources/Byre/Panes.swift index d82af01..574221d 100644 --- a/mac/Sources/Byre/Panes.swift +++ b/mac/Sources/Byre/Panes.swift @@ -252,9 +252,6 @@ struct TaskBar: View { @Environment(Store.self) private var store @Environment(Overlays.self) private var overlays @Environment(\.paneAccent) private var accent - @ViewState private var previews: [Components.Schemas.Preview] = [] - @ViewState private var target = 0 - @ViewState private var off: (() -> Void)? var body: some View { let task = store.selectedTaskId.flatMap { store.findTask($0) } @@ -284,10 +281,8 @@ struct TaskBar: View { .layoutPriority(1) HStack(spacing: 12) { Spacer(minLength: 0) - ForEach(previews, id: \.port) { p in - if store.forwards.url(p.taskId, p.port) != nil { - PreviewLink(preview: p) - } + ForEach(store.selectedTaskId.map { store.forwards.previews($0) } ?? [], id: \.self) { p in + PreviewLink(preview: p) } #if os(macOS) Button { @@ -328,48 +323,11 @@ struct TaskBar: View { .overlay(alignment: .bottom) { Rectangle().fill(Palette.border).frame(height: 1) } - .task(id: store.selectedTaskId) { - previews = [] - await load() - } - .onAppear { - off = store.events.add { ev in - if (ev.kind == "preview.exposed" || ev.kind == "preview.taken") && ev.taskId == store.selectedTaskId { - Task { - await load() - } - } - } - } - .onDisappear { - off?() - off = nil - } - } - - private func load() async { - target += 1 - let started = target - guard let taskId = store.selectedTaskId else { - previews = [] - return - } - do { - let list = try await store.api.client.listPreviews(path: .init(taskId: taskId)).ok.body.json - if started == target { - previews = list - } - } catch { - if started == target { - previews = [] - } - log.error("list previews: \(errorText(error), privacy: .public)") - } } } private struct PreviewLink: View { - let preview: Components.Schemas.Preview + let preview: ForwardedPreview @Environment(Store.self) private var store @Environment(\.paneAccent) private var accent @ViewState private var hover = false diff --git a/mac/Sources/Byre/Workflows.swift b/mac/Sources/Byre/Workflows.swift index 73b1447..54c1480 100644 --- a/mac/Sources/Byre/Workflows.swift +++ b/mac/Sources/Byre/Workflows.swift @@ -141,7 +141,7 @@ private struct WorkflowRow: View { } } .buttonStyle(SmallLinkButtonStyle()) - .help("Delete this workflow; tasks already running it keep their copy") + .help("Delete this workflow and the schedules that run it; tasks already running it keep their copy") } .padding(.vertical, 6) .overlay(alignment: .top) {