Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269package daemon
import ( "bufio" "context" "encoding/json" "errors" "fmt" "net/http" "net/url" "strings")
// AgentEvent is one event from the daemon's agent bus.type AgentEvent struct { Progress *ToolProgress `json:"progress"` Type string `json:"type"` Session string `json:"session"` Parent string `json:"parent"` Name string `json:"name"` Model string `json:"model"` From string `json:"from"` To string `json:"to"` FromName string `json:"fromName"` Kind string `json:"kind"` Text string `json:"text"` CallID string `json:"callId"` Output string `json:"output"` Source string `json:"source"` Depth int `json:"depth"` Bytes int `json:"bytes"` Running bool `json:"running"`}
type Member struct { Session string `json:"session"` Parent string `json:"parent"` Name string `json:"name"` Depth int `json:"depth"` Closed bool `json:"closed"`}
type ChildResult struct { Session Session `json:"session"` Member Member `json:"member"`}
type AgentNode struct { Parent *string `json:"parent"` Address *string `json:"address"` Session Session `json:"session"` Name string `json:"name"` Depth int `json:"depth"` Running bool `json:"running"` Closed bool `json:"closed"`}
type AgentsSnapshot struct { Root string `json:"root"` Nodes []AgentNode `json:"nodes"`}
type ChildRequest struct { Name string `json:"name"` Task string `json:"task"`}
// StreamAgents delivers the first batch, including an empty readiness batch,// then nonempty batches until the stream ends or onBatch fails.// The caller controls cancellation, including any blocking work in onBatch.func StreamAgents(ctx context.Context, conn *Connection, onBatch func([]AgentEvent) error) error { operation := operation{Name: "stream agents", Method: http.MethodGet, Path: "/agents/stream", Policy: readRecovery} first := true err := scanEventStream(ctx, conn, operation, streamLimits{requireSSE: true, lineBytes: 8 * 1024 * 1024}, func(scanner *bufio.Scanner) error { for scanner.Scan() { line := scanner.Text() if !strings.HasPrefix(line, "data:") { continue } var batch struct { Events []json.RawMessage `json:"events"` } if err := json.Unmarshal([]byte(strings.TrimSpace(line[5:])), &batch); err != nil { return streamFailure(StreamProtocol, fmt.Errorf("invalid agent batch: %w", err)) } if batch.Events == nil { return streamFailure(StreamProtocol, errors.New("invalid agent batch: missing events")) } events := make([]AgentEvent, 0, len(batch.Events)) for _, raw := range batch.Events { event, err := decodeAgentEvent(raw) if err != nil { return streamFailure(StreamProtocol, err) } if event != nil { if event.Type == "overflow" && len(batch.Events) != 1 { return streamFailure(StreamProtocol, errors.New("overflow must be the sole agent event")) } events = append(events, *event) } } if len(events) == 0 && !first { continue } first = false if err := onBatch(events); err != nil { return streamFailure(StreamTerminal, err) } if len(events) == 1 && events[0].Type == "overflow" { return nil } } return nil }) return classifyStreamFailure(err)}
func CreateChild(ctx context.Context, conn *Connection, id string, body ChildRequest) (ChildResult, error) { var result ChildResult err := executeMutation(ctx, conn, operation{Name: "create child", Method: http.MethodPost, Path: sessionPath(id, "/children"), Body: body, Policy: authRecovery}, []int{201}, func(body []byte, _ int) error { var err error result, err = decodeChild(body) return err }) return result, err}
func GetAgents(ctx context.Context, conn *Connection, session string) (AgentsSnapshot, error) { var result AgentsSnapshot err := executeRead(ctx, conn, operation{Name: "get agents", Method: http.MethodGet, Path: "/agents?session=" + url.QueryEscape(session), Policy: readRecovery}, func(data []byte) error { fields, err := object(data) if err != nil { return err } if err = required(fields, "root", &result.Root); err != nil { return err } var rows []json.RawMessage if err = required(fields, "nodes", &rows); err != nil { return err } result.Nodes = make([]AgentNode, 0, len(rows)) for _, row := range rows { fields, err := object(row) if err != nil { return err } var node AgentNode var sessionData json.RawMessage if err = nullable(fields, "parent", &node.Parent); err != nil { return err } if err = nullable(fields, "address", &node.Address); err != nil { return err } if err = required(fields, "session", &sessionData); err != nil { return err } node.Session, err = decodeSession(sessionData) if err != nil { return err } if err = required(fields, "name", &node.Name); err != nil { return err } if err = required(fields, "depth", &node.Depth); err != nil { return err } if node.Depth < 0 { return fieldError("depth") } if err = required(fields, "running", &node.Running); err != nil { return err } if err = required(fields, "closed", &node.Closed); err != nil { return err } result.Nodes = append(result.Nodes, node) } return nil }) return result, err}
func decodeAgentEvent(raw json.RawMessage) (*AgentEvent, error) { var header struct { Type string `json:"type"` } if err := json.Unmarshal(raw, &header); err != nil { return nil, fmt.Errorf("invalid agent event: %w", err) } if header.Type == "" { return nil, errors.New("invalid agent event: missing type") } switch header.Type { case "overflow": return &AgentEvent{Type: "overflow"}, nil case "spawn", "gone", "mail", "running", "text", "thinking", "arguments_delta", "tool_progress", "tool", "user", "message", "error", "interrupted", "progress", "renamed", "closed": default: return nil, nil } var event AgentEvent if err := json.Unmarshal(raw, &event); err != nil { return nil, fmt.Errorf("invalid %s agent event: %w", header.Type, err) } var fields map[string]json.RawMessage if err := json.Unmarshal(raw, &fields); err != nil { return nil, err } required := []string{"session"} switch event.Type { case "mail": required = []string{"to", "bytes", "kind"} case "spawn": required = append(required, "parent", "name", "model", "depth") case "running": required = append(required, "running") case "text", "thinking", "arguments_delta", "user", "message", "error", "progress": required = append(required, "text") if event.Type == "arguments_delta" { required = append(required, "callId", "name") } case "tool_progress": if err := validateToolProgress(fields["progress"]); err != nil { return nil, err } required = append(required, "progress") case "tool": required = append(required, "output") case "renamed": required = append(required, "name") } for _, name := range required { if value, exists := fields[name]; !exists || (string(value) == "null" && name != "progress") { return nil, fmt.Errorf("invalid %s agent event: missing %s", event.Type, name) } } return &event, nil}
func decodeChild(data []byte) (ChildResult, error) { fields, err := object(data) if err != nil { return ChildResult{}, err } var result ChildResult raw, ok := fields["session"] if !ok { return result, fieldError("session") } result.Session, err = decodeSession(raw) if err != nil { return ChildResult{}, err } member, err := object(fields["member"]) if err != nil { return ChildResult{}, err } for _, field := range []struct { target any name string }{{&result.Member.Session, "session"}, {&result.Member.Parent, "parent"}, {&result.Member.Name, "name"}, {&result.Member.Depth, "depth"}, {&result.Member.Closed, "closed"}} { if err := required(member, field.name, field.target); err != nil { return ChildResult{}, err } } return result, nil}