264 lines
9.2 KiB
Go
264 lines
9.2 KiB
Go
package agents
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
"g.fc.run/ai/byre/internal/api"
|
|
"g.fc.run/ai/byre/internal/ids"
|
|
"g.fc.run/ai/byre/internal/mcp"
|
|
"g.fc.run/ai/byre/internal/schedule"
|
|
"g.fc.run/ai/byre/internal/store"
|
|
)
|
|
|
|
func (m *Manager) SaveSchedule(ctx context.Context, projectID, scheduleID string, save api.ScheduleSave, login, actor string) (api.Schedule, error) {
|
|
sc, err := schedule.Timing(save)
|
|
if err != nil {
|
|
return api.Schedule{}, err
|
|
}
|
|
sc.Prompt = strings.TrimSpace(deref(save.Prompt))
|
|
sc.Model = strings.TrimSpace(deref(save.Model))
|
|
sc.Workflow = strings.TrimSpace(deref(save.Workflow))
|
|
switch {
|
|
case sc.Workflow != "" && sc.Model != "":
|
|
return api.Schedule{}, errModelWithWorkflow
|
|
case sc.Workflow == "" && sc.Prompt == "":
|
|
return api.Schedule{}, fmt.Errorf("%w: prompt is required without a workflow", ErrInvalid)
|
|
case sc.Workflow == "" && sc.Model == "":
|
|
return api.Schedule{}, fmt.Errorf("%w: model is required without a workflow", ErrInvalid)
|
|
case sc.Workflow != "":
|
|
_, err = m.store.GetWorkflow(ctx, sc.Workflow)
|
|
if err != nil {
|
|
return api.Schedule{}, err
|
|
}
|
|
}
|
|
|
|
if scheduleID != "" {
|
|
existing, err := m.store.GetSchedule(ctx, scheduleID)
|
|
if err != nil {
|
|
return api.Schedule{}, err
|
|
}
|
|
sc.Id = existing.Id
|
|
sc.ProjectId = existing.ProjectId
|
|
sc.Owner = existing.Owner
|
|
sc, err = m.store.UpdateSchedule(ctx, sc, actor)
|
|
if err != nil {
|
|
return api.Schedule{}, err
|
|
}
|
|
m.poke()
|
|
return sc, nil
|
|
}
|
|
|
|
user, err := m.store.GetUser(ctx, login)
|
|
if err != nil {
|
|
return api.Schedule{}, err
|
|
}
|
|
if user.GitName == "" || user.GitEmail == "" {
|
|
return api.Schedule{}, ErrNoGitIdentity
|
|
}
|
|
project, err := m.store.GetProject(ctx, projectID)
|
|
if err != nil {
|
|
return api.Schedule{}, err
|
|
}
|
|
sc.Id = ids.New(ids.Schedule)
|
|
sc.ProjectId = project.Id
|
|
sc.Owner = user.Login
|
|
sc, err = m.store.CreateSchedule(ctx, sc, actor)
|
|
if err != nil {
|
|
return api.Schedule{}, err
|
|
}
|
|
m.poke()
|
|
return sc, nil
|
|
}
|
|
|
|
func (m *Manager) DeleteSchedule(ctx context.Context, scheduleID, actor string) error {
|
|
err := m.store.DeleteSchedule(ctx, scheduleID, actor)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
m.poke()
|
|
return nil
|
|
}
|
|
|
|
func (m *Manager) runSchedule(sc api.Schedule) {
|
|
ctx := context.Background()
|
|
err := m.startScheduled(ctx, sc)
|
|
if err == nil {
|
|
slog.Info("scheduled task started", "schedule", sc.Id, "project", sc.ProjectId)
|
|
} else {
|
|
slog.Error("start scheduled task", "schedule", sc.Id, "project", sc.ProjectId, "err", err)
|
|
m.signalSchedule(ctx, sc, fmt.Sprintf("Schedule %s %q didn't start its run: %v.", sc.Id, sc.Title, err))
|
|
}
|
|
for i := 0; ; i++ {
|
|
err = m.store.AdvanceSchedule(ctx, sc)
|
|
if err == nil {
|
|
break
|
|
}
|
|
slog.Error("advance schedule", "schedule", sc.Id, "err", err)
|
|
if i == 0 {
|
|
m.signalSchedule(ctx, sc, fmt.Sprintf("Schedule %s %q couldn't move on to its next run: %v. byre retries every minute; the schedule doesn't run again until it does.", sc.Id, sc.Title, err))
|
|
}
|
|
time.Sleep(time.Minute)
|
|
}
|
|
m.poke()
|
|
}
|
|
|
|
func (m *Manager) signalSchedule(ctx context.Context, sc api.Schedule, text string) {
|
|
err := m.addLeadInput(ctx, store.LeadInput{Owner: sc.Owner, Kind: store.LeadSignal, Actor: "byre", Text: text})
|
|
if err != nil {
|
|
slog.Error("signal lead", "schedule", sc.Id, "err", err)
|
|
}
|
|
}
|
|
|
|
func (m *Manager) startScheduled(ctx context.Context, sc api.Schedule) error {
|
|
project, err := m.store.GetProject(ctx, sc.ProjectId)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
user, err := m.store.GetUser(ctx, sc.Owner)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if user.GitName == "" || user.GitEmail == "" {
|
|
return ErrNoGitIdentity
|
|
}
|
|
if sc.Workflow != "" {
|
|
_, err = m.StartWorkflow(ctx, project, user, sc.Title, project.DefaultBranch, "byre", sc.Workflow, sc.Prompt, nil, "")
|
|
return err
|
|
}
|
|
task, err := m.CreateTask(ctx, project, sc.Title, user.Login, project.DefaultBranch, "byre")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = m.create(ctx, task, user, "", sc.Prompt, sc.Model, "byre", "", "", nil)
|
|
if err != nil {
|
|
return fmt.Errorf("task %s created, but its agent did not start: %w", task.Id, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func scheduleText(sc api.Schedule) string {
|
|
b := &strings.Builder{}
|
|
fmt.Fprintf(b, "- %s %q: project %s, owner %s, every %s", sc.Id, sc.Title, sc.ProjectId, sc.Owner, sc.Every)
|
|
switch sc.Every {
|
|
case api.Hour:
|
|
fmt.Fprintf(b, " at minute %d", sc.Minute)
|
|
case api.Week:
|
|
days := []string{}
|
|
for _, d := range sc.Weekdays {
|
|
days = append(days, string(d))
|
|
}
|
|
fmt.Fprintf(b, " on %s at %02d:%02d", strings.Join(days, ", "), sc.Hour, sc.Minute)
|
|
case api.Month:
|
|
fmt.Fprintf(b, " on day %d at %02d:%02d", sc.DayOfMonth, sc.Hour, sc.Minute)
|
|
default:
|
|
fmt.Fprintf(b, " at %02d:%02d", sc.Hour, sc.Minute)
|
|
}
|
|
fmt.Fprintf(b, ", next run %s", sc.NextRun.Local().Format(time.RFC3339))
|
|
if sc.Workflow != "" {
|
|
fmt.Fprintf(b, ", workflow %q", sc.Workflow)
|
|
} else {
|
|
fmt.Fprintf(b, ", model %s", sc.Model)
|
|
}
|
|
if sc.Prompt != "" {
|
|
fmt.Fprintf(b, "\n prompt: %s", clip(sc.Prompt, 500))
|
|
}
|
|
return b.String()
|
|
}
|
|
|
|
func (m *Manager) scheduleTools(l *live, tool func(name, description, schema string, run func(ctx context.Context, user api.User, raw json.RawMessage) (string, error)) mcp.Tool) []mcp.Tool {
|
|
return []mcp.Tool{
|
|
tool("list_schedules",
|
|
"List the schedules: each one's id, project, owner, title, timing, model or workflow, prompt, and next run in the server's local time. Give project_id for one project's.",
|
|
`{"type": "object", "properties": {"project_id": {"type": "string"}}}`,
|
|
func(ctx context.Context, user api.User, raw json.RawMessage) (string, error) {
|
|
args := struct {
|
|
ProjectID string `json:"project_id"`
|
|
}{}
|
|
err := json.Unmarshal(raw, &args)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
list, err := m.store.ListSchedules(ctx, args.ProjectID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if len(list) == 0 {
|
|
return "No schedules.", nil
|
|
}
|
|
lines := []string{}
|
|
for _, sc := range list {
|
|
lines = append(lines, scheduleText(sc))
|
|
}
|
|
return strings.Join(lines, "\n"), nil
|
|
}),
|
|
|
|
tool("save_schedule",
|
|
"Create a schedule, or with schedule_id replace one's settings (project_id is then ignored). On each run byre creates a task in the project from its default branch, as create_task does, "+
|
|
"as the schedule's owner (you, for one you create), and you get its created signal. Times are the server's local time. "+
|
|
"every hour runs at minute past each hour (hour 0); day at hour:minute; week at hour:minute on each of weekdays; month at hour:minute on day_of_month, or the month's last day when shorter. "+
|
|
"weekdays only for week, day_of_month only for month. Each run's agent sees nothing else, so the prompt must be a complete brief. "+
|
|
"With workflow, each run starts that workflow, prompt is optional and is its brief, and model can't be given; the workflow is looked up by name at each run. "+
|
|
"A run missed while byre was down happens once when it starts. The human sees and edits schedules in the Project tab. Only when the human asks.",
|
|
`{"type": "object", "properties": {
|
|
"project_id": {"type": "string"},
|
|
"schedule_id": {"type": "string"},
|
|
"title": {"type": "string", "description": "The title of each task it creates."},
|
|
"every": {"type": "string", "enum": ["hour", "day", "week", "month"]},
|
|
"weekdays": {"type": "array", "items": {"type": "string", "enum": ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]}},
|
|
"hour": {"type": "integer", "minimum": 0, "maximum": 23},
|
|
"minute": {"type": "integer", "minimum": 0, "maximum": 59},
|
|
"day_of_month": {"type": "integer", "minimum": 1, "maximum": 31},
|
|
"prompt": {"type": "string"},
|
|
"model": {"type": "string", "description": "Without workflow, the agent's model; defaults to yours."},
|
|
"workflow": {"type": "string", "description": "The name of a workflow for each run to freeze a copy of and run."}
|
|
}, "required": ["title", "every", "hour", "minute"]}`,
|
|
func(ctx context.Context, user api.User, raw json.RawMessage) (string, error) {
|
|
args := struct {
|
|
ProjectID string `json:"project_id"`
|
|
ScheduleID string `json:"schedule_id"`
|
|
}{}
|
|
err := json.Unmarshal(raw, &args)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
save := api.ScheduleSave{}
|
|
err = json.Unmarshal(raw, &save)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if deref(save.Workflow) == "" {
|
|
model := m.modelOr(deref(save.Model), l)
|
|
save.Model = &model
|
|
}
|
|
sc, err := m.SaveSchedule(ctx, args.ProjectID, args.ScheduleID, save, user.Login, leadActor(user.Login))
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "Saved:\n" + scheduleText(sc), nil
|
|
}),
|
|
|
|
tool("delete_schedule",
|
|
"Delete a schedule. Tasks it already created stay. Only when the human asks.",
|
|
`{"type": "object", "properties": {"schedule_id": {"type": "string"}}, "required": ["schedule_id"]}`,
|
|
func(ctx context.Context, user api.User, raw json.RawMessage) (string, error) {
|
|
args := struct {
|
|
ScheduleID string `json:"schedule_id"`
|
|
}{}
|
|
err := json.Unmarshal(raw, &args)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
err = m.DeleteSchedule(ctx, args.ScheduleID, leadActor(user.Login))
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return fmt.Sprintf("Deleted schedule %s.", args.ScheduleID), nil
|
|
}),
|
|
}
|
|
}
|