270 lines
8.2 KiB
Go
270 lines
8.2 KiB
Go
package agents
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
"g.fc.run/ai/byre/internal/api"
|
|
"g.fc.run/ai/byre/internal/mcp"
|
|
"g.fc.run/ai/byre/internal/store"
|
|
)
|
|
|
|
func (m *Manager) Fork(ctx context.Context, src api.Task, project api.Project, n int, creator api.User) (api.Task, error) {
|
|
t, _, err := m.fork(ctx, src, project, n, creator, forkTitle(src, n))
|
|
return t, err
|
|
}
|
|
|
|
func forkTitle(src api.Task, n int) string {
|
|
return fmt.Sprintf("%s (fork of #%d)", src.Title, n)
|
|
}
|
|
|
|
func (m *Manager) forkTool(l *live) mcp.Tool {
|
|
return mcp.Tool{
|
|
Name: "fork",
|
|
Description: "Fork this task here, for when two ways forward are both worth finishing and the human should compare the results rather than you picking one. " +
|
|
"byre snapshots your worktree now and starts a new task from that snapshot, with its own branch and worktree, whose agent continues this conversation up to just before this call " +
|
|
"and then gets prompt as its next message. You carry on here with the other way. " +
|
|
"Each fork doubles the work and gives the human another result to compare, so fork only when you strongly believe both paths are worth following.",
|
|
InputSchema: json.RawMessage(`{"type": "object", "properties": {
|
|
"title": {"type": "string", "description": "The new task's title, naming the path it takes."},
|
|
"prompt": {"type": "string", "description": "The forked agent's next message: tell it that it is the other side of a fork and which path to take."}
|
|
}, "required": ["title", "prompt"]}`),
|
|
Handler: func(ctx context.Context, raw json.RawMessage) (string, error) {
|
|
args := struct {
|
|
Title string `json:"title"`
|
|
Prompt string `json:"prompt"`
|
|
}{}
|
|
err := json.Unmarshal(raw, &args)
|
|
if err != nil {
|
|
return "", fmt.Errorf("malformed fork call: %w", err)
|
|
}
|
|
if strings.TrimSpace(args.Title) == "" || strings.TrimSpace(args.Prompt) == "" {
|
|
return "", errors.New("title and prompt are required")
|
|
}
|
|
sess, err := m.store.GetSession(ctx, l.id)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
creator, err := m.store.GetUser(ctx, sess.CreatedBy)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
src, err := m.store.GetTask(ctx, l.taskID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
project, err := m.store.GetProject(ctx, src.ProjectId)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
snap, err := m.cutSnapshot(ctx, liveTarget(l), m.ownsTask(l.id), api.SnapshotTriggerFork)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if snap == nil {
|
|
return "", fmt.Errorf("%w: task %s is closing", ErrConflict, src.Id)
|
|
}
|
|
t, sessionID, err := m.fork(ctx, src, project, snap.N, creator, args.Title)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
_, _, err = m.Message(ctx, sessionID, agentActor, args.Prompt, "")
|
|
if err != nil {
|
|
return "", fmt.Errorf("task %s forked, but its agent did not start: %w", t.Id, err)
|
|
}
|
|
slog.Info("task forked by its agent", "task", src.Id, "fork", t.Id, "snapshot", snap.N)
|
|
return fmt.Sprintf("Forked at snapshot #%d as task %s %q, branch %s, worktree %s; its agent continues from just before this call with your prompt. Carry on here with your own path.",
|
|
snap.N, t.Id, t.Title, t.Branch, t.WorktreePath), nil
|
|
},
|
|
}
|
|
}
|
|
|
|
func (m *Manager) AnswerAgain(ctx context.Context, creator api.User, create api.InboxResponseCreate) (api.Task, error) {
|
|
item, err := m.store.GetInboxItem(ctx, create.ItemId)
|
|
if err != nil {
|
|
return api.Task{}, err
|
|
}
|
|
if item.State != api.Answered {
|
|
return api.Task{}, fmt.Errorf("%w: question %s is %s, not answered", ErrConflict, item.Id, item.State)
|
|
}
|
|
n := 0
|
|
for _, r := range item.Responses {
|
|
if r.SnapshotN != nil {
|
|
n = *r.SnapshotN
|
|
}
|
|
}
|
|
if n == 0 {
|
|
return api.Task{}, fmt.Errorf("%w: question %s's answer reached the agent without a snapshot to fork from", ErrConflict, item.Id)
|
|
}
|
|
open := item
|
|
open.State = api.Open
|
|
_, err = CheckResponse(open, create)
|
|
if err != nil {
|
|
return api.Task{}, err
|
|
}
|
|
|
|
src, err := m.store.GetTask(ctx, item.TaskId)
|
|
if err != nil {
|
|
return api.Task{}, err
|
|
}
|
|
project, err := m.store.GetProject(ctx, src.ProjectId)
|
|
if err != nil {
|
|
return api.Task{}, err
|
|
}
|
|
t, sessionID, err := m.fork(ctx, src, project, n, creator, forkTitle(src, n))
|
|
if err != nil {
|
|
return api.Task{}, err
|
|
}
|
|
|
|
id := store.NewID()
|
|
err = m.store.CreateAnsweredItem(ctx, api.InboxItem{
|
|
Id: id,
|
|
ProjectId: t.ProjectId,
|
|
TaskId: t.Id,
|
|
AgentSessionId: sessionID,
|
|
ThreadId: id,
|
|
Payload: item.Payload,
|
|
CreatedAt: time.Now().UTC(),
|
|
}, newResponse(id, creator.Login, create))
|
|
if err != nil {
|
|
return api.Task{}, fmt.Errorf("task %s forked, but its answer was not recorded: %w", t.Id, err)
|
|
}
|
|
slog.Info("question answered again", "item", item.Id, "fork", t.Id, "snapshot", n, "user", creator.Login)
|
|
|
|
go func() {
|
|
err := m.Deliver(context.Background(), sessionID)
|
|
if err != nil {
|
|
slog.Error("deliver inbox responses", "session", sessionID, "err", err)
|
|
}
|
|
}()
|
|
return t, nil
|
|
}
|
|
|
|
func (m *Manager) fork(ctx context.Context, src api.Task, project api.Project, n int, creator api.User, title string) (api.Task, string, error) {
|
|
if creator.GitName == "" || creator.GitEmail == "" {
|
|
return api.Task{}, "", ErrNoGitIdentity
|
|
}
|
|
if src.Status != api.TaskStatusActive {
|
|
return api.Task{}, "", fmt.Errorf("%w: task %s is %s", ErrConflict, src.Id, src.Status)
|
|
}
|
|
|
|
snap, err := m.store.GetSnapshot(ctx, src.Id, n)
|
|
if err != nil {
|
|
return api.Task{}, "", err
|
|
}
|
|
srcSess, err := m.store.GetSession(ctx, snap.AgentSessionId)
|
|
if err != nil {
|
|
return api.Task{}, "", err
|
|
}
|
|
if srcSess.Vendor == api.Antigravity {
|
|
return api.Task{}, "", fmt.Errorf("%w: snapshot %d of task %s is from an Antigravity session, and agy can't fork a conversation", ErrConflict, n, src.Id)
|
|
}
|
|
events, err := m.store.TaskEvents(ctx, src.Id, 0, -1)
|
|
if err != nil {
|
|
return api.Task{}, "", err
|
|
}
|
|
at := forkPoint(events, snap.AgentSessionId, snap.EventSeq)
|
|
if at == "" {
|
|
return api.Task{}, "", fmt.Errorf("%w: snapshot %d of task %s has no recorded agent message at or before it to fork the conversation from", ErrConflict, n, src.Id)
|
|
}
|
|
|
|
var frozen *store.FrozenWorkflow
|
|
if src.Workflow != nil {
|
|
f, err := m.store.TaskWorkflow(ctx, src.Id)
|
|
if err != nil {
|
|
return api.Task{}, "", err
|
|
}
|
|
frozen = &f
|
|
}
|
|
|
|
t, err := m.createTask(ctx, project, api.Task{
|
|
Title: title,
|
|
MergeTitle: src.MergeTitle,
|
|
CreatedBy: creator.Login,
|
|
TodoItem: src.TodoItem,
|
|
ForkedFrom: &src.Id,
|
|
ForkedAt: &n,
|
|
WorkflowRun: src.WorkflowRun,
|
|
WorkflowStage: src.WorkflowStage,
|
|
ParentTaskId: src.ParentTaskId,
|
|
}, frozen, snap.CommitSha, creator.Login)
|
|
if err != nil {
|
|
return api.Task{}, "", err
|
|
}
|
|
|
|
id := newUUID()
|
|
err = m.store.CreateSession(ctx, api.AgentSession{
|
|
Id: id,
|
|
TaskId: t.Id,
|
|
Vendor: srcSess.Vendor,
|
|
Model: srcSess.Model,
|
|
VendorSessionId: initialVendorSessionID(srcSess.Vendor, id),
|
|
State: api.AgentSessionStateIdle,
|
|
Role: api.Implementer,
|
|
CreatedBy: creator.Login,
|
|
CreatedAt: time.Now().UTC(),
|
|
ForkedFrom: &srcSess.VendorSessionId,
|
|
ForkedAt: &at,
|
|
}, t.ProjectId)
|
|
if err != nil {
|
|
return api.Task{}, "", fmt.Errorf("task %s created, but its agent session was not: %w", t.Id, err)
|
|
}
|
|
return t, id, nil
|
|
}
|
|
|
|
func forkPoint(events []api.Event, sessionID string, seq int64) string {
|
|
at := ""
|
|
prev := ""
|
|
cut := ""
|
|
open := map[string]bool{}
|
|
for _, ev := range events {
|
|
if ev.AgentSessionId == nil || *ev.AgentSessionId != sessionID {
|
|
continue
|
|
}
|
|
uuid, _ := ev.Payload["vendor_uuid"].(string)
|
|
if uuid == "" {
|
|
continue
|
|
}
|
|
id := ""
|
|
switch ev.Kind {
|
|
case "agent.tool_call":
|
|
id, _ = ev.Payload["id"].(string)
|
|
case "agent.tool_result":
|
|
id, _ = ev.Payload["tool_use_id"].(string)
|
|
}
|
|
after := ev.Seq > seq
|
|
if after && (len(open) == 0 || id == "") {
|
|
return at
|
|
}
|
|
switch ev.Kind {
|
|
case "agent.tool_call":
|
|
if len(open) == 0 {
|
|
cut = at
|
|
if at == uuid {
|
|
cut = prev
|
|
}
|
|
}
|
|
open[id] = true
|
|
case "agent.tool_result":
|
|
delete(open, id)
|
|
}
|
|
if uuid != at {
|
|
prev = at
|
|
}
|
|
at = uuid
|
|
if after && len(open) == 0 {
|
|
return at
|
|
}
|
|
}
|
|
// Calls still open have no results yet, as when the agent forks from inside a tool call: cut before them.
|
|
if len(open) > 0 {
|
|
return cut
|
|
}
|
|
return at
|
|
}
|