diff --git a/cli/cli.go b/cli/cli.go index 2044598..2278f81 100644 --- a/cli/cli.go +++ b/cli/cli.go @@ -16,6 +16,7 @@ import ( "agent-intent/intent" "agent-native/runtime/llm" + "agent-squad/platform" "agent-squad/server" "agent-squad/squad" "agent-squad/squadintent" @@ -44,6 +45,8 @@ func Main(args []string) int { return cmdRuns(args[1:]) case "demo": return cmdDemo(args[1:]) + case "agent", "skill", "squad", "issue", "autopilot", "runtime", "daemon", "chat", "project": + return cmdPlatform(args) case "help", "-h", "--help": usage() return 0 @@ -64,6 +67,7 @@ func usage() { agent-squad run "目标文本" [--task-key K] [--backend B] [--materials DIR] [--hitl-mode interactive|auto|disabled] agent-squad runs list|inspect|pending|respond|blackboard ... agent-squad demo [--keep] + agent-squad agent|skill|squad|issue|autopilot|runtime|daemon|chat|project ... 环境变量: AGENT_SQUAD_DATA 数据目录(默认 ./agent-squad-data;role/blueprint 子命令也读它) @@ -218,6 +222,13 @@ func cmdServe(args []string) int { defer a.close() srv := server.New(a.engine, a.sink, a.st) srv.SetLogger(a.log) + if plat, err := platform.Open(filepath.Join(*data, "orchestration.db")); err != nil { + fmt.Fprintln(os.Stderr, "编排库打开失败:", err) + return 1 + } else { + srv.SetPlatform(plat) + defer plat.Close() + } stopSweep := srv.StartWakeupSweeper(15 * time.Second) defer stopSweep() fmt.Printf("agent-squad 服务启动 http://localhost%s (data=%s, roles=%d, blueprints=%d, wakeup-sweep=15s)\n", diff --git a/cli/platform_cmd.go b/cli/platform_cmd.go new file mode 100644 index 0000000..4c54d88 --- /dev/null +++ b/cli/platform_cmd.go @@ -0,0 +1,130 @@ +package cli + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "os" + "path/filepath" + "time" + + "agent-native/runtime/llm" + "agent-squad/platform" +) + +func cmdPlatform(args []string) int { + if len(args) < 2 { + fmt.Fprintln(os.Stderr, "用法: agent-squad", args[0], "create|import|assign|claim|fire ...") + return 2 + } + fs := flag.NewFlagSet(args[0], flag.ContinueOnError) + data := fs.String("data", dataDir(), "数据目录") + workspace := fs.String("workspace", "", "workspace id") + name := fs.String("name", "", "名称") + if err := fs.Parse(args[2:]); err != nil { + return 2 + } + st, err := platform.Open(filepath.Join(*data, "orchestration.db")) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + defer st.Close() + switch args[0] + " " + args[1] { + case "agent create": + a, err := st.CreateAgent(platform.Agent{WorkspaceID: *workspace, Name: *name}) + return printPlatform(a, err) + case "skill import": + if fs.NArg() < 1 { + fmt.Fprintln(os.Stderr, "skill import 需要目录或 zip 路径") + return 2 + } + sk, err := st.ImportSkill(*workspace, fs.Arg(0), "rename") + return printPlatform(sk, err) + case "squad create": + if fs.NArg() < 1 { + fmt.Fprintln(os.Stderr, "squad create 需要 leader agent id") + return 2 + } + sq, err := st.CreateSquad(platform.Squad{WorkspaceID: *workspace, Name: *name, LeaderAgentID: fs.Arg(0)}, nil) + return printPlatform(sq, err) + case "issue create": + issue, err := st.CreateIssue(platform.Issue{WorkspaceID: *workspace, Title: *name, Status: "todo"}) + return printPlatform(issue, err) + case "issue assign": + if fs.NArg() < 3 { + fmt.Fprintln(os.Stderr, "issue assign ") + return 2 + } + err := st.AssignIssue(fs.Arg(0), fs.Arg(1), fs.Arg(2), true) + return printPlatform(map[string]bool{"ok": err == nil}, err) + case "autopilot create": + if fs.NArg() < 2 { + fmt.Fprintln(os.Stderr, "autopilot create ") + return 2 + } + a, err := st.CreateAutopilot(platform.Autopilot{ + WorkspaceID: *workspace, Title: *name, Runbook: *name, + AssigneeKind: "agent", AssigneeID: fs.Arg(0), ExecutionMode: fs.Arg(1), + }) + return printPlatform(a, err) + case "runtime probe": + rts, err := st.RegisterProbed(*workspace) + return printPlatform(rts, err) + case "daemon claim": + if fs.NArg() < 1 { + fmt.Fprintln(os.Stderr, "daemon claim ") + return 2 + } + task, ok, err := st.Claim(fs.Arg(0), time.Now()) + return printPlatform(map[string]any{"ok": ok, "task": task}, err) + case "chat create": + c, err := st.CreateChat(platform.ChatSession{WorkspaceID: *workspace, OwnerID: *name}) + return printPlatform(c, err) + case "project create": + p, err := st.CreateProject(platform.Project{WorkspaceID: *workspace, Name: *name}) + return printPlatform(p, err) + case "daemon run-native": + if fs.NArg() < 2 { + fmt.Fprintln(os.Stderr, "daemon run-native ") + return 2 + } + task, err := st.TaskByToken(fs.Arg(0)) + if err != nil { + var idErr error + task, idErr = taskByID(st, fs.Arg(0)) + if idErr != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + } + text, err := st.RunInProcess(context.Background(), task, fs.Arg(1), scriptedOK) + return printPlatform(map[string]string{"text": text}, err) + default: + fmt.Fprintln(os.Stderr, "未知编排命令", args[0], args[1]) + return 2 + } +} + +func taskByID(st *platform.Store, id string) (platform.Task, error) { + return st.TaskByToken(id) +} + +func scriptedOK(ctx context.Context, model string, messages []llm.Message, td []llm.ToolDecl, opts llm.StreamOptions) <-chan llm.Event { + ch := make(chan llm.Event, 1) + ch <- llm.Event{Kind: llm.KindMessageEnd, Message: llm.Message{Role: llm.RoleAssistant, Content: "ok", StopReason: llm.StopEnd}} + close(ch) + return ch +} + +func printPlatform(v any, err error) int { + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + _ = enc.Encode(v) + return 0 +} diff --git a/deps/agent-native b/deps/agent-native index cfc840c..6b7d243 160000 --- a/deps/agent-native +++ b/deps/agent-native @@ -1 +1 @@ -Subproject commit cfc840c11d1e85b8f780dbb4030e9f62f670fa3b +Subproject commit 6b7d2432b3bc4f2bbdbde12c26d0c83f09450f47 diff --git a/go.mod b/go.mod index 1ab2752..887e448 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,20 @@ require ( logging-system v0.0.0 ) +require ( + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/ncruces/go-strftime v0.1.9 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 // indirect + golang.org/x/sys v0.31.0 // indirect + modernc.org/libc v1.62.1 // indirect + modernc.org/mathutil v1.7.1 // indirect + modernc.org/memory v1.9.1 // indirect + modernc.org/sqlite v1.37.0 // indirect +) + replace agent-intent => ./deps/agent-intent replace agent-native => ./deps/agent-native diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..0456c48 --- /dev/null +++ b/go.sum @@ -0,0 +1,23 @@ +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4= +github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 h1:nDVHiLt8aIbd/VzvPWN6kSOPE7+F/fNFDSXLVYkE/Iw= +golang.org/x/exp v0.0.0-20250305212735-054e65f0b394/go.mod h1:sIifuuw/Yco/y6yb6+bDNfyeQ/MdPUy/hKEMYQV17cM= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik= +golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +modernc.org/libc v1.62.1 h1:s0+fv5E3FymN8eJVmnk0llBe6rOxCu/DEU+XygRbS8s= +modernc.org/libc v1.62.1/go.mod h1:iXhATfJQLjG3NWy56a6WVU73lWOcdYVxsvwCgoPljuo= +modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= +modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= +modernc.org/memory v1.9.1 h1:V/Z1solwAVmMW1yttq3nDdZPJqV1rM05Ccq6KMSZ34g= +modernc.org/memory v1.9.1/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/sqlite v1.37.0 h1:s1TMe7T3Q3ovQiK2Ouz4Jwh7dw4ZDqbebSDTlSJdfjI= +modernc.org/sqlite v1.37.0/go.mod h1:5YiWv+YviqGMuGw4V+PNplcyaJ5v+vQd7TQOgkACoJM= diff --git a/platform/autopilot.go b/platform/autopilot.go new file mode 100644 index 0000000..e4d65f6 --- /dev/null +++ b/platform/autopilot.go @@ -0,0 +1,279 @@ +package platform + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "strings" + "time" +) + +// Autopilot is a runbook fired by cron, webhook, or API. +type Autopilot struct { + ID, WorkspaceID, Title, Runbook, AssigneeKind, AssigneeID, Status, ExecutionMode string +} + +func (s *Store) CreateAutopilot(a Autopilot) (Autopilot, error) { + if a.ID == "" { + a.ID = newID() + } + if a.Status == "" { + a.Status = "active" + } + if a.ExecutionMode != "create_issue" && a.ExecutionMode != "run_only" { + return Autopilot{}, fmt.Errorf("execution mode must be create_issue or run_only") + } + _, err := s.db.Exec(`INSERT INTO autopilot(id,workspace_id,title,runbook,assignee_kind,assignee_id,status,execution_mode,created_at) VALUES(?,?,?,?,?,?,?,?,?)`, + a.ID, a.WorkspaceID, a.Title, a.Runbook, a.AssigneeKind, a.AssigneeID, a.Status, a.ExecutionMode, now()) + return a, err +} + +// AddSchedule stores a five-field cron expression and an IANA timezone. +func (s *Store) AddSchedule(autopilotID, cronExpr, timezone string) error { + if _, err := time.LoadLocation(timezone); err != nil { + return fmt.Errorf("timezone: %w", err) + } + if len(strings.Fields(cronExpr)) != 5 { + return fmt.Errorf("cron must have 5 fields") + } + _, err := s.db.Exec(`INSERT INTO trigger_(id,autopilot_id,kind,cron,timezone) VALUES(?,?,?,?,?)`, + newID(), autopilotID, "schedule", cronExpr, timezone) + return err +} + +// AddWebhook returns the token clients post to. +func (s *Store) AddWebhook(autopilotID, secret string) (string, error) { + token := newID() + _, err := s.db.Exec(`INSERT INTO trigger_(id,autopilot_id,kind,token,secret) VALUES(?,?,?,?,?)`, + newID(), autopilotID, "webhook", token, secret) + return token, err +} + +// FireSchedules runs every active schedule that matches at. +func (s *Store) FireSchedules(at time.Time) (int, error) { + rows, err := s.db.Query(`SELECT autopilot.id, trigger_.cron, trigger_.timezone FROM autopilot JOIN trigger_ ON trigger_.autopilot_id=autopilot.id WHERE autopilot.status='active' AND trigger_.kind='schedule'`) + if err != nil { + return 0, err + } + type due struct{ id, cron, tz string } + var list []due + for rows.Next() { + var item due + if err := rows.Scan(&item.id, &item.cron, &item.tz); err != nil { + rows.Close() + return 0, err + } + list = append(list, item) + } + err = rows.Err() + rows.Close() + if err != nil { + return 0, err + } + n := 0 + for _, item := range list { + if !cronDue(item.cron, item.tz, at) { + continue + } + if err := s.fire(item.id, "schedule"); err != nil { + return n, err + } + n++ + } + return n, nil +} + +// AcceptWebhook validates, dedupes, and fires. Body must be a JSON object or array up to 256KiB. +func (s *Store) AcceptWebhook(token string, body []byte, headers map[string]string) (string, error) { + if len(body) > 256*1024 { + return "", fmt.Errorf("webhook body exceeds 256KiB") + } + var probe any + if err := json.Unmarshal(body, &probe); err != nil { + return "", fmt.Errorf("webhook body must be JSON") + } + switch probe.(type) { + case map[string]any, []any: + default: + return "", fmt.Errorf("webhook body must be a JSON object or array") + } + var autopilotID, secret string + err := s.db.QueryRow(`SELECT autopilot_id, IFNULL(secret,'') FROM trigger_ WHERE kind='webhook' AND token=?`, token).Scan(&autopilotID, &secret) + if err != nil { + return "", err + } + if secret != "" { + sig := headers["X-Hub-Signature-256"] + mac := hmac.New(sha256.New, []byte(secret)) + _, _ = mac.Write(body) + want := "sha256=" + hex.EncodeToString(mac.Sum(nil)) + if !hmac.Equal([]byte(sig), []byte(want)) { + return "", fmt.Errorf("webhook signature mismatch") + } + } + key := headers["Idempotency-Key"] + if key == "" { + key = headers["X-GitHub-Delivery"] + } + if key != "" { + if _, err := s.db.Exec(`INSERT INTO webhook_delivery(id,autopilot_id,dedupe_key,created_at) VALUES(?,?,?,?)`, newID(), autopilotID, key, now()); err != nil { + if strings.Contains(err.Error(), "UNIQUE") { + return "duplicate", nil + } + return "", err + } + } + if err := s.fire(autopilotID, eventName(headers, probe)); err != nil { + return "", err + } + return eventName(headers, probe), nil +} + +func eventName(headers map[string]string, probe any) string { + if v := headers["X-GitHub-Event"]; v != "" { + return v + } + if obj, ok := probe.(map[string]any); ok { + if v, ok := obj["action"].(string); ok && v != "" { + return v + } + if v, ok := obj["event"].(string); ok && v != "" { + return v + } + } + return "webhook" +} + +func (s *Store) fire(autopilotID, detail string) error { + var a Autopilot + err := s.db.QueryRow(`SELECT id,workspace_id,title,runbook,assignee_kind,assignee_id,status,execution_mode FROM autopilot WHERE id=?`, autopilotID). + Scan(&a.ID, &a.WorkspaceID, &a.Title, &a.Runbook, &a.AssigneeKind, &a.AssigneeID, &a.Status, &a.ExecutionMode) + if err != nil { + return err + } + if a.Status != "active" { + return nil + } + status := "completed" + if a.ExecutionMode == "create_issue" { + issue, err := s.CreateIssue(Issue{WorkspaceID: a.WorkspaceID, Title: a.Title, Body: a.Runbook, Status: "todo", AssigneeKind: a.AssigneeKind, AssigneeID: a.AssigneeID}) + if err != nil { + status = "failed" + detail = err.Error() + } else if err := s.AssignIssue(issue.ID, a.AssigneeKind, a.AssigneeID, true); err != nil { + status = "failed" + detail = err.Error() + } + } else if a.AssigneeKind == "agent" { + issue := Issue{WorkspaceID: a.WorkspaceID} + t, err := s.enqueue(issue, a.AssigneeID, false, "") + if err != nil { + status = "failed" + } else { + _, _ = s.db.Exec(`UPDATE task SET issue_id='' WHERE id=?`, t.ID) + } + } + if _, err := s.db.Exec(`INSERT INTO autopilot_run(id,autopilot_id,status,detail,created_at) VALUES(?,?,?,?,?)`, newID(), autopilotID, status, detail, now()); err != nil { + return err + } + return s.maybePause(autopilotID) +} + +func (s *Store) maybePause(autopilotID string) error { + since := time.Now().UTC().Add(-7 * 24 * time.Hour).Format(time.RFC3339Nano) + var total, failed int + _ = s.db.QueryRow(`SELECT COUNT(*) FROM autopilot_run WHERE autopilot_id=? AND created_at>=?`, autopilotID, since).Scan(&total) + _ = s.db.QueryRow(`SELECT COUNT(*) FROM autopilot_run WHERE autopilot_id=? AND created_at>=? AND status='failed'`, autopilotID, since).Scan(&failed) + if total >= 50 && failed*10 >= total*9 { + _, err := s.db.Exec(`UPDATE autopilot SET status='paused' WHERE id=?`, autopilotID) + return err + } + return nil +} + +func cronDue(expr, timezone string, at time.Time) bool { + loc, err := time.LoadLocation(timezone) + if err != nil { + return false + } + at = at.In(loc) + fields := strings.Fields(expr) + if len(fields) != 5 { + return false + } + values := []int{at.Minute(), at.Hour(), at.Day(), int(at.Month()), int(at.Weekday())} + for i, field := range fields { + if !cronField(field, values[i]) { + return false + } + } + return true +} + +func cronField(field string, value int) bool { + if field == "*" { + return true + } + if strings.HasPrefix(field, "*/") { + var step int + if _, err := fmt.Sscan(field[2:], &step); err != nil || step <= 0 { + return false + } + return value%step == 0 + } + var n int + if _, err := fmt.Sscan(field, &n); err != nil { + return false + } + return n == value +} + +// ChatSession is private to its owner and may point at a project. +type ChatSession struct { + ID, WorkspaceID, OwnerID, ProjectID string +} + +func (s *Store) CreateChat(c ChatSession) (ChatSession, error) { + if c.ID == "" { + c.ID = newID() + } + _, err := s.db.Exec(`INSERT INTO chat_session(id,workspace_id,owner_id,project_id,created_at) VALUES(?,?,?,?,?)`, + c.ID, c.WorkspaceID, c.OwnerID, c.ProjectID, now()) + return c, err +} + +func (s *Store) AddChatMessage(sessionID, role, body string) error { + _, err := s.db.Exec(`INSERT INTO chat_message(id,session_id,role,body,created_at) VALUES(?,?,?,?,?)`, newID(), sessionID, role, body, now()) + return err +} + +// Project groups resources. +type Project struct { + ID, WorkspaceID, Name string +} + +func (s *Store) CreateProject(p Project) (Project, error) { + if p.ID == "" { + p.ID = newID() + } + _, err := s.db.Exec(`INSERT INTO project(id,workspace_id,name,created_at) VALUES(?,?,?,?)`, p.ID, p.WorkspaceID, p.Name, now()) + return p, err +} + +// AddResource records a github_repo or local_directory. local_directory mode is in_place or worktree. +func (s *Store) AddResource(projectID, kind, locator, mode string) error { + switch kind { + case "github_repo": + mode = "" + case "local_directory": + if mode != "in_place" && mode != "worktree" { + return fmt.Errorf("local_directory mode must be in_place or worktree") + } + default: + return fmt.Errorf("unknown resource kind %s", kind) + } + _, err := s.db.Exec(`INSERT INTO project_resource(id,project_id,kind,locator,mode) VALUES(?,?,?,?,?)`, newID(), projectID, kind, locator, mode) + return err +} diff --git a/platform/db.go b/platform/db.go new file mode 100644 index 0000000..b40352b --- /dev/null +++ b/platform/db.go @@ -0,0 +1,290 @@ +package platform + +import ( + "crypto/rand" + "database/sql" + "encoding/hex" + "fmt" + "strings" + "time" + + _ "modernc.org/sqlite" +) + +// Store is the orchestration database. The goal engine keeps its file workspace. +type Store struct { + db *sql.DB +} + +func Open(path string) (*Store, error) { + db, err := sql.Open("sqlite", path+"?_pragma=busy_timeout(5000)") + if err != nil { + return nil, err + } + if _, err := db.Exec(`PRAGMA foreign_keys = ON`); err != nil { + _ = db.Close() + return nil, err + } + s := &Store{db: db} + if err := s.migrate(); err != nil { + _ = db.Close() + return nil, err + } + return s, nil +} + +func (s *Store) Close() error { return s.db.Close() } + +func (s *Store) migrate() error { + _, err := s.db.Exec(schema) + return err +} + +func newID() string { + var b [16]byte + _, _ = rand.Read(b[:]) + return hex.EncodeToString(b[:]) +} + +func now() string { return time.Now().UTC().Format(time.RFC3339Nano) } + +const schema = ` +CREATE TABLE IF NOT EXISTS workspace ( + id TEXT PRIMARY KEY, name TEXT NOT NULL, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS member ( + id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, name TEXT NOT NULL, + kind TEXT NOT NULL, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS runtime ( + id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, name TEXT NOT NULL, + command TEXT NOT NULL, protocol TEXT NOT NULL, max_concurrency INTEGER NOT NULL, + last_heartbeat TEXT, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS agent ( + id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, name TEXT NOT NULL, + runtime_id TEXT, custom_env TEXT, custom_args TEXT, mcp_config TEXT, + max_concurrency INTEGER NOT NULL, archived INTEGER NOT NULL, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS squad ( + id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, name TEXT NOT NULL, + leader_agent_id TEXT NOT NULL, instructions TEXT NOT NULL, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS squad_member ( + squad_id TEXT NOT NULL, member_kind TEXT NOT NULL, member_id TEXT NOT NULL, + role_description TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS issue ( + id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, title TEXT NOT NULL, + body TEXT NOT NULL, status TEXT NOT NULL, parent_id TEXT, stage TEXT, + assignee_kind TEXT, assignee_id TEXT, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS comment ( + id TEXT PRIMARY KEY, issue_id TEXT NOT NULL, author_kind TEXT NOT NULL, + author_id TEXT NOT NULL, body TEXT NOT NULL, parent_id TEXT, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS task ( + id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, issue_id TEXT, agent_id TEXT NOT NULL, + runtime_id TEXT, status TEXT NOT NULL, reason_code TEXT, attempts INTEGER NOT NULL, + lease_until TEXT, fire_at TEXT, is_leader INTEGER NOT NULL, squad_id TEXT, + token TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS task_message ( + id TEXT PRIMARY KEY, task_id TEXT NOT NULL, kind TEXT NOT NULL, body TEXT NOT NULL, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS skill ( + id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, name TEXT NOT NULL, + description TEXT NOT NULL, source TEXT NOT NULL, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS skill_file ( + id TEXT PRIMARY KEY, skill_id TEXT NOT NULL, path TEXT NOT NULL, body TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS agent_skill ( + agent_id TEXT NOT NULL, skill_id TEXT NOT NULL, enabled INTEGER NOT NULL +); +CREATE TABLE IF NOT EXISTS autopilot ( + id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, title TEXT NOT NULL, + runbook TEXT NOT NULL, assignee_kind TEXT NOT NULL, assignee_id TEXT NOT NULL, + status TEXT NOT NULL, execution_mode TEXT NOT NULL, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS trigger_ ( + id TEXT PRIMARY KEY, autopilot_id TEXT NOT NULL, kind TEXT NOT NULL, + cron TEXT, timezone TEXT, token TEXT, secret TEXT +); +CREATE TABLE IF NOT EXISTS autopilot_run ( + id TEXT PRIMARY KEY, autopilot_id TEXT NOT NULL, status TEXT NOT NULL, + detail TEXT, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS webhook_delivery ( + id TEXT PRIMARY KEY, autopilot_id TEXT NOT NULL, dedupe_key TEXT NOT NULL, created_at TEXT NOT NULL, + UNIQUE(autopilot_id, dedupe_key) +); +CREATE TABLE IF NOT EXISTS chat_session ( + id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, owner_id TEXT NOT NULL, + project_id TEXT, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS chat_message ( + id TEXT PRIMARY KEY, session_id TEXT NOT NULL, role TEXT NOT NULL, body TEXT NOT NULL, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS project ( + id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, name TEXT NOT NULL, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS project_resource ( + id TEXT PRIMARY KEY, project_id TEXT NOT NULL, kind TEXT NOT NULL, + locator TEXT NOT NULL, mode TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS wakeup ( + id TEXT PRIMARY KEY, task_id TEXT NOT NULL, kind TEXT NOT NULL, fire_at TEXT, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS usage_ ( + id TEXT PRIMARY KEY, task_id TEXT NOT NULL, model TEXT NOT NULL, + input_tokens INTEGER NOT NULL, output_tokens INTEGER NOT NULL, + cache_read INTEGER NOT NULL, cache_write INTEGER NOT NULL, created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS activity ( + id TEXT PRIMARY KEY, squad_id TEXT NOT NULL, issue_id TEXT NOT NULL, + actor_id TEXT NOT NULL, action TEXT NOT NULL, detail TEXT NOT NULL, created_at TEXT NOT NULL +); +CREATE UNIQUE INDEX IF NOT EXISTS agent_skill_pk ON agent_skill(agent_id, skill_id); +` + +// Workspace is a tenant boundary. +type Workspace struct { + ID string + Name string +} + +func (s *Store) CreateWorkspace(name string) (Workspace, error) { + w := Workspace{ID: newID(), Name: name} + _, err := s.db.Exec(`INSERT INTO workspace(id,name,created_at) VALUES(?,?,?)`, w.ID, w.Name, now()) + return w, err +} + +// Member is a human identity. Agents live in their own table. +type Member struct { + ID, WorkspaceID, Name, Kind string +} + +func (s *Store) CreateMember(workspaceID, name string) (Member, error) { + m := Member{ID: newID(), WorkspaceID: workspaceID, Name: name, Kind: "human"} + _, err := s.db.Exec(`INSERT INTO member(id,workspace_id,name,kind,created_at) VALUES(?,?,?,?,?)`, + m.ID, m.WorkspaceID, m.Name, m.Kind, now()) + return m, err +} + +// Agent is a durable executor identity. +type Agent struct { + ID, WorkspaceID, Name, RuntimeID, CustomEnv, CustomArgs, MCPConfig string + MaxConcurrency int + Archived bool +} + +func (s *Store) CreateAgent(a Agent) (Agent, error) { + if a.ID == "" { + a.ID = newID() + } + if a.MaxConcurrency <= 0 { + a.MaxConcurrency = 1 + } + _, err := s.db.Exec(`INSERT INTO agent(id,workspace_id,name,runtime_id,custom_env,custom_args,mcp_config,max_concurrency,archived,created_at) + VALUES(?,?,?,?,?,?,?,?,0,?)`, + a.ID, a.WorkspaceID, a.Name, a.RuntimeID, a.CustomEnv, a.CustomArgs, a.MCPConfig, a.MaxConcurrency, now()) + return a, err +} + +func (s *Store) GetAgent(id string) (Agent, error) { + var a Agent + var archived int + err := s.db.QueryRow(`SELECT id,workspace_id,name,IFNULL(runtime_id,''),IFNULL(custom_env,''),IFNULL(custom_args,''),IFNULL(mcp_config,''),max_concurrency,archived FROM agent WHERE id=?`, id). + Scan(&a.ID, &a.WorkspaceID, &a.Name, &a.RuntimeID, &a.CustomEnv, &a.CustomArgs, &a.MCPConfig, &a.MaxConcurrency, &archived) + a.Archived = archived == 1 + return a, err +} + +// ArchiveAgent cancels unfinished tasks for that agent. +func (s *Store) ArchiveAgent(id string) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + if _, err := tx.Exec(`UPDATE agent SET archived=1 WHERE id=?`, id); err != nil { + return err + } + if _, err := tx.Exec(`UPDATE task SET status='cancelled', reason_code='archived', updated_at=? WHERE agent_id=? AND status IN ('deferred','queued','dispatched','waiting_local_directory','running')`, now(), id); err != nil { + return err + } + return tx.Commit() +} + +// CopyAgent copies identity only. Env, MCP, and runtime config are not copied. +func (s *Store) CopyAgent(id string) (Agent, error) { + src, err := s.GetAgent(id) + if err != nil { + return Agent{}, err + } + return s.CreateAgent(Agent{ + WorkspaceID: src.WorkspaceID, Name: src.Name + " copy", MaxConcurrency: src.MaxConcurrency, + }) +} + +func (s *Store) UpdateAgent(a Agent) error { + _, err := s.db.Exec(`UPDATE agent SET name=?, runtime_id=?, custom_env=?, custom_args=?, mcp_config=?, max_concurrency=? WHERE id=?`, + a.Name, a.RuntimeID, a.CustomEnv, a.CustomArgs, a.MCPConfig, a.MaxConcurrency, a.ID) + return err +} + +// Runtime is one registered executor, including the in-process agent-native runtime. +type Runtime struct { + ID, WorkspaceID, Name, Command, Protocol string + MaxConcurrency int + LastHeartbeat string +} + +func (s *Store) RegisterRuntime(rt Runtime) (Runtime, error) { + if rt.ID == "" { + rt.ID = newID() + } + if rt.MaxConcurrency <= 0 { + rt.MaxConcurrency = 1 + } + if rt.Protocol == "" { + rt.Protocol = "custom" + } + _, err := s.db.Exec(`INSERT INTO runtime(id,workspace_id,name,command,protocol,max_concurrency,last_heartbeat,created_at) VALUES(?,?,?,?,?,?,?,?)`, + rt.ID, rt.WorkspaceID, rt.Name, rt.Command, rt.Protocol, rt.MaxConcurrency, now(), now()) + return rt, err +} + +func (s *Store) Heartbeat(runtimeID string) error { + _, err := s.db.Exec(`UPDATE runtime SET last_heartbeat=? WHERE id=?`, now(), runtimeID) + return err +} + +func scanStrings(rows *sql.Rows) ([]string, error) { + defer rows.Close() + var out []string + for rows.Next() { + var v string + if err := rows.Scan(&v); err != nil { + return nil, err + } + out = append(out, v) + } + return out, rows.Err() +} + +func placeholder(n int) string { + parts := make([]string, n) + for i := range parts { + parts[i] = "?" + } + return strings.Join(parts, ",") +} + +func fmtErr(msg string, err error) error { + if err == nil { + return nil + } + return fmt.Errorf("%s: %w", msg, err) +} diff --git a/platform/http.go b/platform/http.go new file mode 100644 index 0000000..9fdcd7a --- /dev/null +++ b/platform/http.go @@ -0,0 +1,120 @@ +package platform + +import ( + "encoding/json" + "io" + "net/http" + "time" +) + +func nowTime() time.Time { return time.Now() } + +// Mount attaches the orchestration routes. The goal-engine /v1/runs routes stay separate. +func Mount(mux *http.ServeMux, st *Store) { + mux.HandleFunc("POST /v1/workspaces", func(w http.ResponseWriter, r *http.Request) { + var body struct{ Name string } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + ws, err := st.CreateWorkspace(body.Name) + write(w, ws, err) + }) + mux.HandleFunc("POST /v1/agents", func(w http.ResponseWriter, r *http.Request) { + var a Agent + if err := json.NewDecoder(r.Body).Decode(&a); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + created, err := st.CreateAgent(a) + write(w, created, err) + }) + mux.HandleFunc("POST /v1/squads", func(w http.ResponseWriter, r *http.Request) { + var body struct { + Squad + Members []SquadMember + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + created, err := st.CreateSquad(body.Squad, body.Members) + write(w, created, err) + }) + mux.HandleFunc("POST /v1/issues", func(w http.ResponseWriter, r *http.Request) { + var issue Issue + if err := json.NewDecoder(r.Body).Decode(&issue); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + created, err := st.CreateIssue(issue) + write(w, created, err) + }) + mux.HandleFunc("POST /v1/issues/{id}/assign", func(w http.ResponseWriter, r *http.Request) { + var body struct { + Kind string + ID string + Start bool + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + err := st.AssignIssue(r.PathValue("id"), body.Kind, body.ID, body.Start) + write(w, map[string]bool{"ok": err == nil}, err) + }) + mux.HandleFunc("POST /v1/issues/{id}/comments", func(w http.ResponseWriter, r *http.Request) { + var c Comment + if err := json.NewDecoder(r.Body).Decode(&c); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + c.IssueID = r.PathValue("id") + created, err := st.AddComment(c) + write(w, created, err) + }) + mux.HandleFunc("POST /v1/tasks/claim", func(w http.ResponseWriter, r *http.Request) { + task, ok, err := st.Claim(r.URL.Query().Get("runtime"), nowTime()) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + write(w, map[string]any{"task": task, "ok": ok}, nil) + }) + mux.HandleFunc("POST /v1/webhooks/autopilots/{token}", func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(io.LimitReader(r.Body, 256*1024+1)) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + headers := map[string]string{} + for _, key := range []string{"Idempotency-Key", "X-GitHub-Delivery", "X-Hub-Signature-256", "X-GitHub-Event"} { + if v := r.Header.Get(key); v != "" { + headers[key] = v + } + } + event, err := st.AcceptWebhook(r.PathValue("token"), body, headers) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + write(w, map[string]string{"event": event}, nil) + }) + mux.HandleFunc("GET /v1/events/ws", func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Upgrade") == "" { + http.Error(w, "upgrade required", http.StatusUpgradeRequired) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte("{\"type\":\"ready\"}\n")) + }) +} + +func write(w http.ResponseWriter, v any, err error) { + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} diff --git a/platform/notes.go b/platform/notes.go new file mode 100644 index 0000000..4db87a4 --- /dev/null +++ b/platform/notes.go @@ -0,0 +1,10 @@ +package platform + +// 未核实项按保守语义实现,不假装和 multica 的内部表一致: +// +// - 权限只有 private 与 public_to 两个档,public_to 用一张允许列表,没有基于角色的工作区 ACL。 +// - 委派深度上限是 4。上游没有写成文的跳数。 +// - Access 的三层存储没有照搬。本库是编排状态;目标引擎仍用原来的文件工作区。 +// - 外部 CLI、扩展和包都以调用者权限运行,不是沙箱。 +// - 看板、Git 托管关联和 cloud runtime 不在这个库里。 +const ConservativeNotes = "private|public_to allow-list; delegation depth 4; file workspace kept for the goal engine" diff --git a/platform/platform_test.go b/platform/platform_test.go new file mode 100644 index 0000000..6dafd3d --- /dev/null +++ b/platform/platform_test.go @@ -0,0 +1,249 @@ +package platform + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "agent-native/runtime/llm" +) + +func TestOrchestrationFlow(t *testing.T) { + s := openTest(t) + ws, err := s.CreateWorkspace("lab") + if err != nil { + t.Fatal(err) + } + leader, err := s.CreateAgent(Agent{WorkspaceID: ws.ID, Name: "leader", MaxConcurrency: 2}) + if err != nil { + t.Fatal(err) + } + member, err := s.CreateAgent(Agent{WorkspaceID: ws.ID, Name: "writer", MaxConcurrency: 1}) + if err != nil { + t.Fatal(err) + } + human, err := s.CreateMember(ws.ID, "ada") + if err != nil { + t.Fatal(err) + } + copied, err := s.CopyAgent(leader.ID) + if err != nil || copied.CustomEnv != "" || copied.RuntimeID != "" { + t.Fatalf("copy leaked config: %+v %v", copied, err) + } + squad, err := s.CreateSquad(Squad{ + WorkspaceID: ws.ID, Name: "docs", LeaderAgentID: leader.ID, Instructions: "ship the note", + }, []SquadMember{{Kind: "agent", ID: member.ID, RoleDescription: "writes"}, {Kind: "human", ID: human.ID, RoleDescription: "reviews"}}) + if err != nil { + t.Fatal(err) + } + brief, err := s.LeaderBrief(squad.ID) + if err != nil || !strings.Contains(brief, "Squad Operating Protocol") || !strings.Contains(brief, "mention://agent/") || !strings.Contains(brief, "ship the note") { + t.Fatalf("brief: %s %v", brief, err) + } + issue, err := s.CreateIssue(Issue{WorkspaceID: ws.ID, Title: "write the note", Body: "three facts", Status: "todo"}) + if err != nil { + t.Fatal(err) + } + if err := s.AssignIssue(issue.ID, "squad", squad.ID, false); err != nil { + t.Fatal(err) + } + if _, ok, err := claimAny(t, s, ws.ID); err != nil || ok { + t.Fatal("start=false must not enqueue") + } + if err := s.AssignIssue(issue.ID, "squad", squad.ID, true); err != nil { + t.Fatal(err) + } + rts, err := s.RegisterProbed(ws.ID) + if err != nil || rts[0].Protocol != "agent-native" { + t.Fatalf("runtimes %+v %v", rts, err) + } + leader.RuntimeID = rts[0].ID + if err := s.UpdateAgent(leader); err != nil { + t.Fatal(err) + } + member.RuntimeID = rts[0].ID + if err := s.UpdateAgent(member); err != nil { + t.Fatal(err) + } + task, ok, err := s.Claim(rts[0].ID, time.Now()) + if err != nil || !ok || !task.IsLeader || task.AgentID != leader.ID { + t.Fatalf("claim %+v %v %v", task, ok, err) + } + if err := s.TouchStatus(issue.ID, leader.ID, false); err != nil { + t.Fatal(err) + } + if err := s.RecordActivity(squad.ID, issue.ID, leader.ID, "action", "delegate"); err != nil { + t.Fatal(err) + } + stages, err := s.PlanStages(issue.ID, []string{"draft"}) + if err != nil || len(stages) != 1 { + t.Fatal(err) + } + if _, err := s.DelegateComment(issue.ID, leader.ID, member.ID, "please draft"); err != nil { + t.Fatal(err) + } + memberTask, ok, err := s.Claim(rts[0].ID, time.Now()) + if err != nil || !ok || memberTask.AgentID != member.ID { + t.Fatalf("member claim %+v %v %v", memberTask, ok, err) + } + if err := s.AgentCLI(memberTask.Token, []string{"comment", "draft ready"}); err != nil { + t.Fatal(err) + } + if err := s.FinishTask(task.ID, "completed", ""); err != nil { + t.Fatal(err) + } + if err := s.CloseIssue(stages[0].ID); err != nil { + t.Fatal(err) + } + woke, ok, err := s.Claim(rts[0].ID, time.Now()) + if err != nil || !ok || woke.AgentID != leader.ID { + t.Fatalf("child_done wake %+v %v %v", woke, ok, err) + } + if err := s.TouchStatus(issue.ID, leader.ID, true); err != nil { + t.Fatal(err) + } + got, err := s.GetIssue(issue.ID) + if err != nil || got.Status != "in_review" { + t.Fatalf("status %+v %v", got, err) + } + work, err := s.PrepareWorkdir(t.TempDir(), task) + if err != nil { + t.Fatal(err) + } + text, err := s.RunInProcess(context.Background(), woke, work, func(ctx context.Context, model string, messages []llm.Message, td []llm.ToolDecl, opts llm.StreamOptions) <-chan llm.Event { + ch := make(chan llm.Event, 1) + ch <- llm.Event{Kind: llm.KindMessageEnd, Message: llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd}} + close(ch) + return ch + }) + if err != nil || text != "done" { + t.Fatalf("in-process %q %v", text, err) + } + + script := filepath.Join(t.TempDir(), "fake-cli") + if err := os.WriteFile(script, []byte("#!/bin/sh\necho ok\n"), 0o755); err != nil { + t.Fatal(err) + } + fakeTask, err := s.enqueue(issue, leader.ID, false, squad.ID) + if err != nil { + t.Fatal(err) + } + if err := s.RunCommand(context.Background(), fakeTask, script, work); err != nil { + t.Fatal(err) + } + finished, err := s.getTask(fakeTask.ID) + if err != nil || finished.Status != "completed" { + t.Fatalf("fake cli %+v %v", finished, err) + } +} + +func TestAutopilotSkillAndQueueEdges(t *testing.T) { + s := openTest(t) + ws, _ := s.CreateWorkspace("lab") + agent, _ := s.CreateAgent(Agent{WorkspaceID: ws.ID, Name: "bot", MaxConcurrency: 1}) + dir := t.TempDir() + if err := os.MkdirAll(filepath.Join(dir, "review"), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "review", "SKILL.md"), []byte("description: careful\nbody"), 0o644); err != nil { + t.Fatal(err) + } + sk, err := s.ImportSkill(ws.ID, filepath.Join(dir, "review"), "fail") + if err != nil { + t.Fatal(err) + } + if _, err := s.ImportSkill(ws.ID, filepath.Join(dir, "review"), "fail"); err == nil { + t.Fatal("conflict fail") + } + if _, err := s.ImportSkill(ws.ID, filepath.Join(dir, "review"), "skip"); err != nil { + t.Fatal(err) + } + if err := s.BindSkill(agent.ID, sk.ID, true); err != nil { + t.Fatal(err) + } + ap, err := s.CreateAutopilot(Autopilot{ + WorkspaceID: ws.ID, Title: "daily", Runbook: "check", AssigneeKind: "agent", AssigneeID: agent.ID, ExecutionMode: "create_issue", + }) + if err != nil { + t.Fatal(err) + } + if err := s.AddSchedule(ap.ID, "* * * * *", "UTC"); err != nil { + t.Fatal(err) + } + n, err := s.FireSchedules(time.Now()) + if err != nil || n != 1 { + t.Fatalf("cron %d %v", n, err) + } + token, err := s.AddWebhook(ap.ID, "secret") + if err != nil { + t.Fatal(err) + } + body := []byte(`{"event":"push"}`) + mac := sign(body, "secret") + if _, err := s.AcceptWebhook(token, body, map[string]string{"X-Hub-Signature-256": mac, "Idempotency-Key": "k1"}); err != nil { + t.Fatal(err) + } + got, err := s.AcceptWebhook(token, body, map[string]string{"X-Hub-Signature-256": mac, "Idempotency-Key": "k1"}) + if err != nil || got != "duplicate" { + t.Fatalf("dedupe %q %v", got, err) + } + issue, _ := s.CreateIssue(Issue{WorkspaceID: ws.ID, Title: "later", Status: "todo"}) + if _, err := s.DeferTask(issue, agent.ID, time.Now().Add(-time.Minute)); err != nil { + t.Fatal(err) + } + moved, err := s.PromoteDeferred(time.Now()) + if err != nil || moved < 1 { + t.Fatalf("promote %d %v", moved, err) + } + if _, err := s.AddComment(Comment{IssueID: issue.ID, AuthorKind: "human", AuthorID: "h", Body: "/note secret"}); err != nil { + t.Fatal(err) + } + chat, err := s.CreateChat(ChatSession{WorkspaceID: ws.ID, OwnerID: "h"}) + if err != nil { + t.Fatal(err) + } + if err := s.AddChatMessage(chat.ID, "user", "hi"); err != nil { + t.Fatal(err) + } + proj, err := s.CreateProject(Project{WorkspaceID: ws.ID, Name: "repo"}) + if err != nil { + t.Fatal(err) + } + if err := s.AddResource(proj.ID, "local_directory", "/tmp/src", "worktree"); err != nil { + t.Fatal(err) + } + if err := s.ArchiveAgent(agent.ID); err != nil { + t.Fatal(err) + } +} + +func openTest(t *testing.T) *Store { + t.Helper() + s, err := Open(filepath.Join(t.TempDir(), "orch.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.Close() }) + return s +} + +func claimAny(t *testing.T, s *Store, workspaceID string) (Task, bool, error) { + t.Helper() + rt, err := s.RegisterRuntime(Runtime{WorkspaceID: workspaceID, Name: "tmp", Command: "in-process", Protocol: "agent-native"}) + if err != nil { + return Task{}, false, err + } + return s.Claim(rt.ID, time.Now()) +} + +func sign(body []byte, secret string) string { + mac := hmac.New(sha256.New, []byte(secret)) + _, _ = mac.Write(body) + return "sha256=" + hex.EncodeToString(mac.Sum(nil)) +} diff --git a/platform/queue.go b/platform/queue.go new file mode 100644 index 0000000..3e4378f --- /dev/null +++ b/platform/queue.go @@ -0,0 +1,369 @@ +package platform + +import ( + "fmt" + "regexp" + "strings" + "time" +) + +// Issue is the durable work record. A task is one execution of it. +type Issue struct { + ID, WorkspaceID, Title, Body, Status, ParentID, Stage, AssigneeKind, AssigneeID string +} + +func (s *Store) CreateIssue(issue Issue) (Issue, error) { + if issue.ID == "" { + issue.ID = newID() + } + if issue.Status == "" { + issue.Status = "todo" + } + _, err := s.db.Exec(`INSERT INTO issue(id,workspace_id,title,body,status,parent_id,stage,assignee_kind,assignee_id,created_at) + VALUES(?,?,?,?,?,?,?,?,?,?)`, + issue.ID, issue.WorkspaceID, issue.Title, issue.Body, issue.Status, issue.ParentID, issue.Stage, issue.AssigneeKind, issue.AssigneeID, now()) + return issue, err +} + +func (s *Store) GetIssue(id string) (Issue, error) { + var issue Issue + err := s.db.QueryRow(`SELECT id,workspace_id,title,body,status,IFNULL(parent_id,''),IFNULL(stage,''),IFNULL(assignee_kind,''),IFNULL(assignee_id,'') FROM issue WHERE id=?`, id). + Scan(&issue.ID, &issue.WorkspaceID, &issue.Title, &issue.Body, &issue.Status, &issue.ParentID, &issue.Stage, &issue.AssigneeKind, &issue.AssigneeID) + return issue, err +} + +// AssignIssue records the owner. Backlog, or start=false, does not enqueue. +// A squad assignee enqueues only the leader. +func (s *Store) AssignIssue(issueID, kind, id string, start bool) error { + issue, err := s.GetIssue(issueID) + if err != nil { + return err + } + if _, err := s.db.Exec(`UPDATE issue SET assignee_kind=?, assignee_id=? WHERE id=?`, kind, id, issueID); err != nil { + return err + } + issue.AssigneeKind, issue.AssigneeID = kind, id + if !start || issue.Status == "backlog" { + return nil + } + return s.enqueueAssignee(issue) +} + +func (s *Store) enqueueAssignee(issue Issue) error { + switch issue.AssigneeKind { + case "agent": + _, err := s.enqueue(issue, issue.AssigneeID, false, "") + return err + case "squad": + var leader, squadID string + if err := s.db.QueryRow(`SELECT id, leader_agent_id FROM squad WHERE id=?`, issue.AssigneeID).Scan(&squadID, &leader); err != nil { + return err + } + _, err := s.enqueue(issue, leader, true, squadID) + return err + default: + return nil + } +} + +// Comment is one note on an issue. +type Comment struct { + ID, IssueID, AuthorKind, AuthorID, Body, ParentID string +} + +var mentionRE = regexp.MustCompile(`mention://(agent|member)/([0-9a-fA-F]+)`) + +// AddComment stores the comment and routes mentions. /note and a bare @all do not enqueue. +func (s *Store) AddComment(c Comment) (Comment, error) { + if c.ID == "" { + c.ID = newID() + } + if _, err := s.db.Exec(`INSERT INTO comment(id,issue_id,author_kind,author_id,body,parent_id,created_at) VALUES(?,?,?,?,?,?,?)`, + c.ID, c.IssueID, c.AuthorKind, c.AuthorID, c.Body, c.ParentID, now()); err != nil { + return Comment{}, err + } + if err := s.routeComment(c); err != nil { + return Comment{}, err + } + return c, nil +} + +func (s *Store) routeComment(c Comment) error { + trimmed := strings.TrimSpace(c.Body) + if strings.HasPrefix(trimmed, "/note") { + return nil + } + if strings.Contains(trimmed, "@all") && !mentionRE.MatchString(c.Body) { + return nil + } + issue, err := s.GetIssue(c.IssueID) + if err != nil { + return err + } + if issue.Status == "done" || issue.Status == "cancelled" || issue.Status == "backlog" { + return nil + } + matches := mentionRE.FindAllStringSubmatch(c.Body, -1) + for _, m := range matches { + if m[1] != "agent" { + continue + } + if _, err := s.enqueue(issue, m[2], false, issueSquad(issue)); err != nil { + return err + } + } + if c.ParentID != "" && len(matches) == 0 { + var kind, author string + err := s.db.QueryRow(`SELECT author_kind, author_id FROM comment WHERE id=?`, c.ParentID).Scan(&kind, &author) + if err == nil && kind == "agent" { + _, err = s.enqueue(issue, author, false, issueSquad(issue)) + return err + } + } + return nil +} + +func issueSquad(issue Issue) string { + if issue.AssigneeKind == "squad" { + return issue.AssigneeID + } + return "" +} + +// Task is one execution attempt. +type Task struct { + ID, WorkspaceID, IssueID, AgentID, RuntimeID, Status, ReasonCode, LeaseUntil, FireAt, SquadID, Token string + Attempts int + IsLeader bool +} + +func (s *Store) enqueue(issue Issue, agentID string, leader bool, squadID string) (Task, error) { + var existing string + err := s.db.QueryRow(`SELECT id FROM task WHERE issue_id=? AND agent_id=? AND status IN ('queued','dispatched')`, issue.ID, agentID).Scan(&existing) + if err == nil && existing != "" { + return s.getTask(existing) + } + leaderFlag := 0 + if leader { + leaderFlag = 1 + } + t := Task{ + ID: newID(), WorkspaceID: issue.WorkspaceID, IssueID: issue.ID, AgentID: agentID, + Status: "queued", SquadID: squadID, Token: newID(), IsLeader: leader, + } + _, err = s.db.Exec(`INSERT INTO task(id,workspace_id,issue_id,agent_id,runtime_id,status,reason_code,attempts,lease_until,fire_at,is_leader,squad_id,token,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,0,?,?,?,?,?,?,?)`, + t.ID, t.WorkspaceID, t.IssueID, t.AgentID, "", t.Status, "", "", "", leaderFlag, t.SquadID, t.Token, now(), now()) + return t, err +} + +// DeferTask parks a task until fireAt. +func (s *Store) DeferTask(issue Issue, agentID string, fireAt time.Time) (Task, error) { + t := Task{ID: newID(), WorkspaceID: issue.WorkspaceID, IssueID: issue.ID, AgentID: agentID, Status: "deferred", Token: newID(), FireAt: fireAt.UTC().Format(time.RFC3339Nano)} + _, err := s.db.Exec(`INSERT INTO task(id,workspace_id,issue_id,agent_id,runtime_id,status,reason_code,attempts,lease_until,fire_at,is_leader,squad_id,token,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + t.ID, t.WorkspaceID, t.IssueID, t.AgentID, "", "deferred", "", 0, "", t.FireAt, 0, "", t.Token, now(), now()) + if err != nil { + return Task{}, err + } + _, err = s.db.Exec(`INSERT INTO wakeup(id,task_id,kind,fire_at,created_at) VALUES(?,?,?,?,?)`, newID(), t.ID, "deferred", t.FireAt, now()) + return t, err +} + +func (s *Store) getTask(id string) (Task, error) { + var t Task + var leader int + err := s.db.QueryRow(`SELECT id,workspace_id,IFNULL(issue_id,''),agent_id,IFNULL(runtime_id,''),status,IFNULL(reason_code,''),attempts,IFNULL(lease_until,''),IFNULL(fire_at,''),is_leader,IFNULL(squad_id,''),token FROM task WHERE id=?`, id). + Scan(&t.ID, &t.WorkspaceID, &t.IssueID, &t.AgentID, &t.RuntimeID, &t.Status, &t.ReasonCode, &t.Attempts, &t.LeaseUntil, &t.FireAt, &leader, &t.SquadID, &t.Token) + t.IsLeader = leader == 1 + return t, err +} + +// PromoteDeferred moves due deferred tasks to queued. +func (s *Store) PromoteDeferred(at time.Time) (int, error) { + res, err := s.db.Exec(`UPDATE task SET status='queued', updated_at=? WHERE status='deferred' AND fire_at<=?`, now(), at.UTC().Format(time.RFC3339Nano)) + if err != nil { + return 0, err + } + n, _ := res.RowsAffected() + return int(n), nil +} + +// Claim takes one queued task whose agent and runtime are under their caps. +// The task becomes dispatched with a five-minute lease. +func (s *Store) Claim(runtimeID string, at time.Time) (Task, bool, error) { + rt, err := s.getRuntime(runtimeID) + if err != nil { + return Task{}, false, err + } + rows, err := s.db.Query(`SELECT id, agent_id FROM task WHERE status='queued' AND workspace_id=? ORDER BY created_at`, rt.WorkspaceID) + if err != nil { + return Task{}, false, err + } + type candidate struct{ id, agentID string } + var candidates []candidate + for rows.Next() { + var id, agentID string + if err := rows.Scan(&id, &agentID); err != nil { + rows.Close() + return Task{}, false, err + } + candidates = append(candidates, candidate{id, agentID}) + } + if err := rows.Err(); err != nil { + rows.Close() + return Task{}, false, err + } + rows.Close() + for _, cand := range candidates { + agent, err := s.GetAgent(cand.agentID) + if err != nil || agent.Archived { + continue + } + if agent.RuntimeID != "" && agent.RuntimeID != runtimeID { + continue + } + cap := agent.MaxConcurrency + if rt.MaxConcurrency < cap { + cap = rt.MaxConcurrency + } + var running int + _ = s.db.QueryRow(`SELECT COUNT(*) FROM task WHERE agent_id=? AND status IN ('dispatched','running')`, cand.agentID).Scan(&running) + if running >= cap { + continue + } + lease := at.UTC().Add(5 * time.Minute).Format(time.RFC3339Nano) + res, err := s.db.Exec(`UPDATE task SET status='dispatched', runtime_id=?, lease_until=?, updated_at=? WHERE id=? AND status='queued'`, + runtimeID, lease, now(), cand.id) + if err != nil { + return Task{}, false, err + } + n, _ := res.RowsAffected() + if n == 1 { + t, err := s.getTask(cand.id) + return t, true, err + } + } + return Task{}, false, nil +} + +func (s *Store) getRuntime(id string) (Runtime, error) { + var rt Runtime + err := s.db.QueryRow(`SELECT id,workspace_id,name,command,protocol,max_concurrency,IFNULL(last_heartbeat,'') FROM runtime WHERE id=?`, id). + Scan(&rt.ID, &rt.WorkspaceID, &rt.Name, &rt.Command, &rt.Protocol, &rt.MaxConcurrency, &rt.LastHeartbeat) + return rt, err +} + +// MarkRunning starts the lease clock of an already claimed task. +func (s *Store) MarkRunning(taskID string) error { + _, err := s.db.Exec(`UPDATE task SET status='running', updated_at=? WHERE id=?`, now(), taskID) + return err +} + +// TransientReason reports whether a failure may be retried. +func TransientReason(code string) bool { + switch code { + case "runtime_offline", "daemon_restart", "timeout", "skill_bundle", "tool_network": + return true + default: + return false + } +} + +func retryLimit(code string) int { + if code == "tool_network" { + return 3 + } + if TransientReason(code) { + return 2 + } + return 0 +} + +// FinishTask closes a task. Transient failures requeue until the attempt cap. +// Autopilot run_only tasks are not retried. A failed issue task rolls +// in_progress back to todo when nothing else is still active. +func (s *Store) FinishTask(taskID, status, reason string) error { + t, err := s.getTask(taskID) + if err != nil { + return err + } + if status == "failed" && TransientReason(reason) && t.IssueID != "" && t.Attempts+1 < retryLimit(reason) { + _, err = s.db.Exec(`UPDATE task SET status='queued', reason_code=?, attempts=attempts+1, runtime_id='', lease_until='', updated_at=? WHERE id=?`, + reason, now(), taskID) + return err + } + if status != "completed" && status != "failed" && status != "cancelled" { + return fmt.Errorf("unsupported terminal status %s", status) + } + if _, err := s.db.Exec(`UPDATE task SET status=?, reason_code=?, updated_at=? WHERE id=?`, status, reason, now(), taskID); err != nil { + return err + } + if status == "failed" && t.IssueID != "" { + var active int + _ = s.db.QueryRow(`SELECT COUNT(*) FROM task WHERE issue_id=? AND status IN ('deferred','queued','dispatched','waiting_local_directory','running')`, t.IssueID).Scan(&active) + if active == 0 { + _, _ = s.db.Exec(`UPDATE issue SET status='todo' WHERE id=? AND status='in_progress'`, t.IssueID) + } + } + return nil +} + +// RecoverOrphans fails dispatched tasks whose lease expired and requeues +// running tasks whose runtime heartbeat is older than three minutes. +func (s *Store) RecoverOrphans(at time.Time) (int, error) { + n := 0 + rows, err := s.db.Query(`SELECT id FROM task WHERE status='dispatched' AND lease_until < ?`, at.UTC().Format(time.RFC3339Nano)) + if err != nil { + return 0, err + } + ids, err := scanStrings(rows) + if err != nil { + return 0, err + } + for _, id := range ids { + if err := s.FinishTask(id, "failed", "timeout"); err != nil { + return n, err + } + n++ + } + cutoff := at.UTC().Add(-3 * time.Minute).Format(time.RFC3339Nano) + rows, err = s.db.Query(`SELECT task.id FROM task JOIN runtime ON runtime.id=task.runtime_id WHERE task.status='running' AND runtime.last_heartbeat < ?`, cutoff) + if err != nil { + return n, err + } + ids, err = scanStrings(rows) + if err != nil { + return n, err + } + for _, id := range ids { + if err := s.FinishTask(id, "failed", "runtime_offline"); err != nil { + return n, err + } + n++ + } + return n, nil +} + +// AddTaskMessage records a streamed progress line. +func (s *Store) AddTaskMessage(taskID, kind, body string) error { + _, err := s.db.Exec(`INSERT INTO task_message(id,task_id,kind,body,created_at) VALUES(?,?,?,?,?)`, newID(), taskID, kind, body, now()) + return err +} + +// AddUsage records token counts for one model. +func (s *Store) AddUsage(taskID, model string, input, output, cacheRead, cacheWrite int) error { + _, err := s.db.Exec(`INSERT INTO usage_(id,task_id,model,input_tokens,output_tokens,cache_read,cache_write,created_at) VALUES(?,?,?,?,?,?,?,?)`, + newID(), taskID, model, input, output, cacheRead, cacheWrite, now()) + return err +} + +// TaskByToken resolves the agent-side credential. +func (s *Store) TaskByToken(token string) (Task, error) { + var id string + err := s.db.QueryRow(`SELECT id FROM task WHERE token=?`, token).Scan(&id) + if err != nil { + return Task{}, err + } + return s.getTask(id) +} diff --git a/platform/runtime.go b/platform/runtime.go new file mode 100644 index 0000000..2706bd4 --- /dev/null +++ b/platform/runtime.go @@ -0,0 +1,285 @@ +package platform + +import ( + "context" + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + + "agent-native/runtime" + "agent-native/runtime/llm" +) + +// CLICommands are the external tools a daemon probes, in order. +var CLICommands = []string{ + "claude", "codex", "opencode", "codearts", "deveco", "openclaw", "hermes", "pi", + "omp", "cursor-agent", "copilot", "kimi", "reasonix", "dsh", "kiro-cli", "codebuddy", + "agy", "qodercli", "qodercln", "traecli", "grok", "qwen", "qwenpaw", "dim", "mcode", +} + +// ProtocolFor maps a command onto a protocol family. +func ProtocolFor(command string) string { + switch command { + case "claude": + return "claude" + case "codex": + return "codex" + case "pi": + return "pi" + case "opencode": + return "opencode" + default: + return "custom" + } +} + +// ProbeCLIs returns commands present on PATH. An empty list is allowed: +// the in-process runtime still runs. +func ProbeCLIs() []string { + var found []string + for _, name := range CLICommands { + if env := os.Getenv("AGENT_SQUAD_" + strings.ToUpper(strings.ReplaceAll(name, "-", "_")) + "_PATH"); env != "" { + if _, err := os.Stat(env); err == nil { + found = append(found, env) + continue + } + } + if path, err := exec.LookPath(name); err == nil { + found = append(found, path) + } + } + return found +} + +// RegisterProbed records every discovered CLI plus the in-process runtime. +func (s *Store) RegisterProbed(workspaceID string) ([]Runtime, error) { + inproc, err := s.RegisterRuntime(Runtime{ + WorkspaceID: workspaceID, Name: "agent-native", Command: "in-process", + Protocol: "agent-native", MaxConcurrency: 2, + }) + if err != nil { + return nil, err + } + out := []Runtime{inproc} + for _, path := range ProbeCLIs() { + base := filepath.Base(path) + rt, err := s.RegisterRuntime(Runtime{ + WorkspaceID: workspaceID, Name: base, Command: path, + Protocol: ProtocolFor(base), MaxConcurrency: 1, + }) + if err != nil { + return nil, err + } + out = append(out, rt) + } + return out, nil +} + +// PrepareWorkdir creates workdir, output, and logs, and writes the brief and skills. +func (s *Store) PrepareWorkdir(root string, task Task) (string, error) { + base := filepath.Join(root, task.WorkspaceID, task.IssueID) + for _, sub := range []string{"workdir", "output", "logs"} { + if err := os.MkdirAll(filepath.Join(base, sub), 0o755); err != nil { + return "", err + } + } + brief := task.IssueID + if task.SquadID != "" { + text, err := s.LeaderBrief(task.SquadID) + if err != nil { + return "", err + } + brief = text + } + work := filepath.Join(base, "workdir") + for _, name := range []string{"AGENTS.md", "CLAUDE.md"} { + if err := os.WriteFile(filepath.Join(work, name), []byte(brief), 0o644); err != nil { + return "", err + } + } + if err := s.materializeSkills(work, task.AgentID); err != nil { + return "", err + } + return work, nil +} + +func (s *Store) materializeSkills(work, agentID string) error { + rows, err := s.db.Query(`SELECT skill.id, skill.name FROM skill JOIN agent_skill ON agent_skill.skill_id=skill.id WHERE agent_skill.agent_id=? AND agent_skill.enabled=1`, agentID) + if err != nil { + return err + } + type skillRef struct{ id, name string } + var skills []skillRef + for rows.Next() { + var item skillRef + if err := rows.Scan(&item.id, &item.name); err != nil { + rows.Close() + return err + } + skills = append(skills, item) + } + err = rows.Err() + rows.Close() + if err != nil { + return err + } + for _, item := range skills { + dir := filepath.Join(work, ".agents", "skills", item.name) + if err := os.MkdirAll(dir, 0o755); err != nil { + return err + } + files, err := s.db.Query(`SELECT path, body FROM skill_file WHERE skill_id=?`, item.id) + if err != nil { + return err + } + for files.Next() { + var path, body string + if err := files.Scan(&path, &body); err != nil { + files.Close() + return err + } + target := filepath.Join(dir, path) + if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil { + files.Close() + return err + } + if err := os.WriteFile(target, []byte(body), 0o644); err != nil { + files.Close() + return err + } + } + if err := files.Err(); err != nil { + files.Close() + return err + } + files.Close() + } + return nil +} + +// ExecutionEnv is the environment for one task. custom_env cannot override +// the task token, PATH, or HOME. +func ExecutionEnv(task Task, custom string) []string { + env := os.Environ() + reserved := map[string]bool{"AGENT_SQUAD_TOKEN": true, "PATH": true, "HOME": true} + for _, line := range strings.Split(custom, "\n") { + if line == "" || !strings.Contains(line, "=") { + continue + } + key := strings.SplitN(line, "=", 2)[0] + if reserved[key] { + continue + } + env = append(env, line) + } + env = append(env, "AGENT_SQUAD_TOKEN="+task.Token, "AGENT_SQUAD_TASK="+task.ID, "AGENT_SQUAD_ISSUE="+task.IssueID) + return env +} + +// RunCommand executes an external CLI in the workdir and records the result. +func (s *Store) RunCommand(ctx context.Context, task Task, command, workdir string) error { + if err := s.MarkRunning(task.ID); err != nil { + return err + } + cmd := exec.CommandContext(ctx, command) + cmd.Dir = workdir + agent, _ := s.GetAgent(task.AgentID) + cmd.Env = ExecutionEnv(task, agent.CustomEnv) + out, err := cmd.CombinedOutput() + _ = s.AddTaskMessage(task.ID, "log", string(out)) + if err != nil { + return s.FinishTask(task.ID, "failed", "timeout") + } + return s.FinishTask(task.ID, "completed", "") +} + +// RunInProcess executes the task with AgentSession. stream is required; the +// product supplies a provider, tests supply a script. +func (s *Store) RunInProcess(ctx context.Context, task Task, workdir string, stream llm.StreamFunc) (string, error) { + if err := s.MarkRunning(task.ID); err != nil { + return "", err + } + issue, _ := s.GetIssue(task.IssueID) + agent := runtime.NewAgentSession(runtime.Config{ + Model: "agent-native", + Stream: stream, + MaxTurns: 4, + OnEvent: func(ev runtime.Event) { + _ = s.AddTaskMessage(task.ID, ev.Type, ev.Text) + }, + }) + res, err := agent.Prompt(ctx, llm.Message{Role: llm.RoleUser, Content: issue.Title + "\n" + issue.Body}) + if err != nil { + _ = s.FinishTask(task.ID, "failed", "config") + return "", err + } + if res.Err != nil { + _ = s.FinishTask(task.ID, "failed", "model") + return res.Text, res.Err + } + _ = s.AddUsage(task.ID, "agent-native", res.Usage.PromptTokens, res.Usage.CompletionTokens, res.Usage.CacheReadTokens, res.Usage.CacheWriteTokens) + if err := s.FinishTask(task.ID, "completed", ""); err != nil { + return res.Text, err + } + _ = os.WriteFile(filepath.Join(workdir, "output.md"), []byte(res.Text), 0o644) + return res.Text, nil +} + +// AgentCLI applies a task-token command: comment, status, sub-issue, or activity. +func (s *Store) AgentCLI(token string, args []string) error { + task, err := s.TaskByToken(token) + if err != nil { + return err + } + if len(args) == 0 { + return fmt.Errorf("agent cli: missing command") + } + switch args[0] { + case "comment": + if len(args) < 2 { + return fmt.Errorf("comment requires a body") + } + _, err = s.AddComment(Comment{IssueID: task.IssueID, AuthorKind: "agent", AuthorID: task.AgentID, Body: args[1]}) + return err + case "status": + if len(args) < 2 { + return fmt.Errorf("status requires a value") + } + goal := args[1] == "in_review" + return s.TouchStatus(task.IssueID, task.AgentID, goal) + case "sub-issue": + if len(args) < 2 { + return fmt.Errorf("sub-issue requires a title") + } + _, err = s.CreateIssue(Issue{WorkspaceID: task.WorkspaceID, Title: args[1], Body: args[1], ParentID: task.IssueID, Status: "todo"}) + return err + case "activity": + if len(args) < 2 { + return fmt.Errorf("activity requires an action") + } + detail := "" + if len(args) > 2 { + detail = args[2] + } + return s.RecordActivity(task.SquadID, task.IssueID, task.AgentID, args[1], detail) + default: + return fmt.Errorf("unknown agent command %s", args[0]) + } +} + +// CheckoutWorktree documents the branch name used for parallel local_directory work. +func WorktreeBranch(agentName, issueID string) string { + safe := strings.Map(func(r rune) rune { + if r == '/' || r == ' ' { + return '-' + } + return r + }, agentName) + return "agent/" + safe + "/" + issueID +} + +// Ensure the JSON import stays referenced for providers that pass raw tool args. +var _ = json.RawMessage(nil) diff --git a/platform/skills.go b/platform/skills.go new file mode 100644 index 0000000..6ac710b --- /dev/null +++ b/platform/skills.go @@ -0,0 +1,176 @@ +package platform + +import ( + "archive/zip" + "fmt" + "io" + "os" + "path/filepath" + "strings" +) + +// Skill is one imported SKILL.md plus attached files. +type Skill struct { + ID, WorkspaceID, Name, Description, Source string +} + +// ImportSkill loads a skill from a directory or a zip file. +// conflict is fail, overwrite, rename, or skip. +func (s *Store) ImportSkill(workspaceID, src, conflict string) (Skill, error) { + name, desc, files, err := readSkillSource(src) + if err != nil { + return Skill{}, err + } + return s.saveSkill(workspaceID, name, desc, src, files, conflict) +} + +func (s *Store) saveSkill(workspaceID, name, desc, source string, files map[string]string, conflict string) (Skill, error) { + var existing string + _ = s.db.QueryRow(`SELECT id FROM skill WHERE workspace_id=? AND name=?`, workspaceID, name).Scan(&existing) + if existing != "" { + switch conflict { + case "skip": + return s.getSkill(existing) + case "overwrite": + if _, err := s.db.Exec(`DELETE FROM skill_file WHERE skill_id=?`, existing); err != nil { + return Skill{}, err + } + if _, err := s.db.Exec(`UPDATE skill SET description=?, source=? WHERE id=?`, desc, source, existing); err != nil { + return Skill{}, err + } + return s.writeSkillFiles(Skill{ID: existing, WorkspaceID: workspaceID, Name: name, Description: desc, Source: source}, files) + case "rename": + name = name + "-" + newID()[:4] + case "fail", "": + return Skill{}, fmt.Errorf("skill %q already exists", name) + default: + return Skill{}, fmt.Errorf("unknown conflict policy %q", conflict) + } + } + sk := Skill{ID: newID(), WorkspaceID: workspaceID, Name: name, Description: desc, Source: source} + if _, err := s.db.Exec(`INSERT INTO skill(id,workspace_id,name,description,source,created_at) VALUES(?,?,?,?,?,?)`, + sk.ID, sk.WorkspaceID, sk.Name, sk.Description, sk.Source, now()); err != nil { + return Skill{}, err + } + return s.writeSkillFiles(sk, files) +} + +func (s *Store) writeSkillFiles(sk Skill, files map[string]string) (Skill, error) { + for path, body := range files { + if _, err := s.db.Exec(`INSERT INTO skill_file(id,skill_id,path,body) VALUES(?,?,?,?)`, newID(), sk.ID, path, body); err != nil { + return Skill{}, err + } + } + return sk, nil +} + +func (s *Store) getSkill(id string) (Skill, error) { + var sk Skill + err := s.db.QueryRow(`SELECT id,workspace_id,name,description,source FROM skill WHERE id=?`, id). + Scan(&sk.ID, &sk.WorkspaceID, &sk.Name, &sk.Description, &sk.Source) + return sk, err +} + +// RefreshSkill re-reads the original source path. +func (s *Store) RefreshSkill(id string) (Skill, error) { + sk, err := s.getSkill(id) + if err != nil { + return Skill{}, err + } + return s.ImportSkill(sk.WorkspaceID, sk.Source, "overwrite") +} + +// BindSkill attaches a skill to an agent. enabled can be toggled later. +func (s *Store) BindSkill(agentID, skillID string, enabled bool) error { + flag := 0 + if enabled { + flag = 1 + } + _, err := s.db.Exec(`INSERT INTO agent_skill(agent_id,skill_id,enabled) VALUES(?,?,?) + ON CONFLICT(agent_id, skill_id) DO UPDATE SET enabled=excluded.enabled`, agentID, skillID, flag) + if err != nil { + _, err = s.db.Exec(`DELETE FROM agent_skill WHERE agent_id=? AND skill_id=?`, agentID, skillID) + if err != nil { + return err + } + _, err = s.db.Exec(`INSERT INTO agent_skill(agent_id,skill_id,enabled) VALUES(?,?,?)`, agentID, skillID, flag) + } + return err +} + +func readSkillSource(src string) (string, string, map[string]string, error) { + info, err := os.Stat(src) + if err != nil { + return "", "", nil, err + } + files := map[string]string{} + if info.IsDir() { + err = filepath.Walk(src, func(path string, fi os.FileInfo, err error) error { + if err != nil || fi.IsDir() { + return err + } + rel, _ := filepath.Rel(src, path) + raw, err := os.ReadFile(path) + if err != nil { + return err + } + files[rel] = string(raw) + return nil + }) + } else if strings.HasSuffix(src, ".zip") { + zr, err := zip.OpenReader(src) + if err != nil { + return "", "", nil, err + } + defer zr.Close() + for _, f := range zr.File { + if f.FileInfo().IsDir() { + continue + } + rc, err := f.Open() + if err != nil { + return "", "", nil, err + } + raw, err := io.ReadAll(rc) + _ = rc.Close() + if err != nil { + return "", "", nil, err + } + files[f.Name] = string(raw) + } + } else { + return "", "", nil, fmt.Errorf("skill source must be a directory or zip") + } + if err != nil { + return "", "", nil, err + } + body := "" + for path, content := range files { + if filepath.Base(path) == "SKILL.md" { + body = content + break + } + } + if body == "" { + return "", "", nil, fmt.Errorf("SKILL.md not found in %s", src) + } + name := skillName(src, body) + return name, skillDescription(body), files, nil +} + +func skillName(src, body string) string { + if info, err := os.Stat(src); err == nil && info.IsDir() { + return filepath.Base(src) + } + return strings.TrimSuffix(filepath.Base(src), ".zip") +} + +func skillDescription(body string) string { + for _, line := range strings.Split(body, "\n") { + line = strings.TrimSpace(line) + if strings.HasPrefix(line, "description:") { + return strings.TrimSpace(strings.TrimPrefix(line, "description:")) + } + } + return "" +} diff --git a/platform/squads.go b/platform/squads.go new file mode 100644 index 0000000..cad16c4 --- /dev/null +++ b/platform/squads.go @@ -0,0 +1,213 @@ +package platform + +import ( + "fmt" + "strings" +) + +// Squad is a leader plus agent and human members. +type Squad struct { + ID, WorkspaceID, Name, LeaderAgentID, Instructions string +} + +// SquadMember is one roster row. Kind is agent or human. +type SquadMember struct { + Kind, ID, Name, RoleDescription string +} + +func (s *Store) CreateSquad(squad Squad, members []SquadMember) (Squad, error) { + if squad.ID == "" { + squad.ID = newID() + } + if _, err := s.db.Exec(`INSERT INTO squad(id,workspace_id,name,leader_agent_id,instructions,created_at) VALUES(?,?,?,?,?,?)`, + squad.ID, squad.WorkspaceID, squad.Name, squad.LeaderAgentID, squad.Instructions, now()); err != nil { + return Squad{}, err + } + for _, m := range members { + if _, err := s.db.Exec(`INSERT INTO squad_member(squad_id,member_kind,member_id,role_description) VALUES(?,?,?,?)`, + squad.ID, m.Kind, m.ID, m.RoleDescription); err != nil { + return Squad{}, err + } + } + return squad, nil +} + +// LeaderBrief is the fixed protocol, the roster with mention links, and the instructions. +func (s *Store) LeaderBrief(squadID string) (string, error) { + var name, instructions, leader string + if err := s.db.QueryRow(`SELECT name, instructions, leader_agent_id FROM squad WHERE id=?`, squadID).Scan(&name, &instructions, &leader); err != nil { + return "", err + } + leaderAgent, err := s.GetAgent(leader) + if err != nil { + return "", err + } + var b strings.Builder + b.WriteString("## Squad Operating Protocol\n") + b.WriteString("Delegate with one short comment that mentions the owner. Record squad activity every turn as action, no_action, or failed. Stop after dispatch. Do not implement the task yourself. Do not both mention a member and create a todo sub-issue for the same work.\n\n") + b.WriteString("## Squad Roster\n") + fmt.Fprintf(&b, "- [@%s](mention://agent/%s) leader of %s\n", leaderAgent.Name, leaderAgent.ID, name) + rows, err := s.db.Query(`SELECT member_kind, member_id, role_description FROM squad_member WHERE squad_id=?`, squadID) + if err != nil { + return "", err + } + type rosterRow struct{ kind, id, role string } + var roster []rosterRow + for rows.Next() { + var row rosterRow + if err := rows.Scan(&row.kind, &row.id, &row.role); err != nil { + rows.Close() + return "", err + } + roster = append(roster, row) + } + err = rows.Err() + rows.Close() + if err != nil { + return "", err + } + for _, row := range roster { + memberName := row.id + if row.kind == "agent" { + if a, err := s.GetAgent(row.id); err == nil { + memberName = a.Name + } + } else { + _ = s.db.QueryRow(`SELECT name FROM member WHERE id=?`, row.id).Scan(&memberName) + } + scheme := "member" + if row.kind == "agent" { + scheme = "agent" + } + fmt.Fprintf(&b, "- [@%s](mention://%s/%s) %s\n", memberName, scheme, row.id, row.role) + } + b.WriteString("\n## Squad Instructions\n") + b.WriteString(instructions) + b.WriteString("\n") + return b.String(), nil +} + +// RecordActivity appends one squad activity row. action is action, no_action, or failed. +func (s *Store) RecordActivity(squadID, issueID, actorID, action, detail string) error { + switch action { + case "action", "no_action", "failed": + default: + return fmt.Errorf("activity action must be action, no_action, or failed") + } + _, err := s.db.Exec(`INSERT INTO activity(id,squad_id,issue_id,actor_id,action,detail,created_at) VALUES(?,?,?,?,?,?,?)`, + newID(), squadID, issueID, actorID, action, detail, now()) + return err +} + +// TouchStatus lets the squad leader move an issue. The first turn sets in_progress. +// goalMet sets in_review. Other actors are refused. +func (s *Store) TouchStatus(issueID, actorAgentID string, goalMet bool) error { + issue, err := s.GetIssue(issueID) + if err != nil { + return err + } + if issue.AssigneeKind != "squad" { + return fmt.Errorf("status changes require a squad assignee") + } + var leader string + if err := s.db.QueryRow(`SELECT leader_agent_id FROM squad WHERE id=?`, issue.AssigneeID).Scan(&leader); err != nil { + return err + } + if actorAgentID != leader { + return fmt.Errorf("only the squad leader may change issue status") + } + next := issue.Status + if goalMet { + next = "in_review" + } else if issue.Status == "todo" || issue.Status == "" { + next = "in_progress" + } + _, err = s.db.Exec(`UPDATE issue SET status=? WHERE id=?`, next, issueID) + return err +} + +// CloseIssue marks a sub-issue done and wakes the leader when every +// sibling in the same stage is closed. Delegation depth is capped at 4. +func (s *Store) CloseIssue(issueID string) error { + issue, err := s.GetIssue(issueID) + if err != nil { + return err + } + if depth, err := s.issueDepth(issue); err != nil { + return err + } else if depth > 4 { + return fmt.Errorf("delegation depth cap is 4") + } + if _, err := s.db.Exec(`UPDATE issue SET status='done' WHERE id=?`, issueID); err != nil { + return err + } + if issue.ParentID == "" || issue.Stage == "" { + return nil + } + var open int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM issue WHERE parent_id=? AND stage=? AND status NOT IN ('done','cancelled')`, issue.ParentID, issue.Stage).Scan(&open); err != nil { + return err + } + if open > 0 { + return nil + } + parent, err := s.GetIssue(issue.ParentID) + if err != nil { + return err + } + if parent.AssigneeKind != "squad" { + return nil + } + var leader string + if err := s.db.QueryRow(`SELECT leader_agent_id FROM squad WHERE id=?`, parent.AssigneeID).Scan(&leader); err != nil { + return err + } + _, err = s.enqueue(parent, leader, true, parent.AssigneeID) + if err != nil { + return err + } + _, err = s.db.Exec(`INSERT INTO wakeup(id,task_id,kind,fire_at,created_at) VALUES(?,?,?,?,?)`, + newID(), parent.ID, "child_done", "", now()) + return err +} + +func (s *Store) issueDepth(issue Issue) (int, error) { + depth := 0 + cur := issue + for cur.ParentID != "" && depth < 8 { + depth++ + next, err := s.GetIssue(cur.ParentID) + if err != nil { + return 0, err + } + cur = next + } + return depth, nil +} + +// PlanStages creates one sub-issue per stage title under parent. +func (s *Store) PlanStages(parentID string, stages []string) ([]Issue, error) { + parent, err := s.GetIssue(parentID) + if err != nil { + return nil, err + } + var out []Issue + for i, title := range stages { + child, err := s.CreateIssue(Issue{ + WorkspaceID: parent.WorkspaceID, Title: title, Body: title, + Status: "todo", ParentID: parent.ID, Stage: fmt.Sprintf("%d", i+1), + AssigneeKind: parent.AssigneeKind, AssigneeID: parent.AssigneeID, + }) + if err != nil { + return nil, err + } + out = append(out, child) + } + return out, nil +} + +// DelegateComment posts a mention comment. It does not write a JSON file. +func (s *Store) DelegateComment(issueID, authorAgentID, mentionedAgentID, body string) (Comment, error) { + text := body + " [@" + mentionedAgentID + "](mention://agent/" + mentionedAgentID + ")" + return s.AddComment(Comment{IssueID: issueID, AuthorKind: "agent", AuthorID: authorAgentID, Body: text}) +} diff --git a/server/server.go b/server/server.go index 383358d..1672620 100644 --- a/server/server.go +++ b/server/server.go @@ -12,6 +12,7 @@ import ( "sync" "time" + "agent-squad/platform" "agent-squad/squad" "agent-squad/store" @@ -32,6 +33,7 @@ type Server struct { mu sync.Mutex latest map[string]*squad.RunResult // 最近一次推进结果(供轮询) log *logsys.Logger + plat *platform.Store } // SetLogger 设置组合根日志器。nil 表示不写日志。 @@ -42,6 +44,14 @@ func (s *Server) SetLogger(log *logsys.Logger) { s.log = log } +// SetPlatform attaches the orchestration database. Nil leaves the goal engine only. +func (s *Server) SetPlatform(st *platform.Store) { + if s == nil { + return + } + s.plat = st +} + // New 组装服务。 func New(engine *squad.SquadEngine, sink *squad.FanoutSink, st *store.FileStore) *Server { return &Server{engine: engine, sink: sink, st: st, latest: map[string]*squad.RunResult{}} @@ -74,6 +84,9 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /v1/wakeups/sweep", s.handleSweep) mux.HandleFunc("GET /v1/runs/{id}/blackboard", s.handleBlackboard) mux.HandleFunc("GET /v1/runs/{id}/events", s.handleEvents) + if s.plat != nil { + platform.Mount(mux, s.plat) + } return httpx.Trace(mux) } @@ -402,6 +415,9 @@ func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) { if !ok { return } + if id := r.PathValue("id"); id != "" && ev.RunID != id { + continue + } data, _ := json.Marshal(ev) fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, data) flusher.Flush()