diff --git a/Dockerfile b/Dockerfile index 568d25e..e0be951 100644 --- a/Dockerfile +++ b/Dockerfile @@ -9,7 +9,7 @@ RUN go mod download COPY cmd ./cmd COPY internal ./internal -RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /out/agora ./cmd/agora +RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /out/agora-server ./cmd/agora-server FROM alpine:3 @@ -24,7 +24,7 @@ ENV AGORA_DATA=/data/agora.jsonl EXPOSE 8080 VOLUME ["/data"] -COPY --from=build /out/agora /usr/local/bin/agora +COPY --from=build /out/agora-server /usr/local/bin/agora-server USER agora -ENTRYPOINT ["/usr/local/bin/agora"] +ENTRYPOINT ["/usr/local/bin/agora-server"] diff --git a/Makefile b/Makefile index 782ab7d..1064d0d 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,8 @@ SHELL = /usr/bin/env bash -o pipefail .SHELLFLAGS = -ec -BINARY ?= agora +CLI_BINARY ?= agora +SERVER_BINARY ?= agora-server BINDIR ?= bin REGISTRY ?= ghcr.io/kelos-dev IMAGE_NAME ?= agora @@ -41,9 +42,17 @@ verify: ## Verify formatting, module metadata, tests, and vet checks. ##@ Build .PHONY: build -build: ## Build the Agora binary. +build: build-cli build-server ## Build the Agora CLI and server binaries. + +.PHONY: build-cli +build-cli: ## Build the Agora CLI binary. + mkdir -p $(BINDIR) + CGO_ENABLED=0 go build -o $(BINDIR)/$(CLI_BINARY) ./cmd/agora + +.PHONY: build-server +build-server: ## Build the Agora server binary. mkdir -p $(BINDIR) - CGO_ENABLED=0 go build -o $(BINDIR)/$(BINARY) ./cmd/agora + CGO_ENABLED=0 go build -o $(BINDIR)/$(SERVER_BINARY) ./cmd/agora-server .PHONY: image image: ## Build the Agora server container image. @@ -51,7 +60,7 @@ image: ## Build the Agora server container image. .PHONY: run run: ## Run the Agora server. - go run ./cmd/agora + go run ./cmd/agora-server .PHONY: clean clean: ## Clean build artifacts. diff --git a/README.md b/README.md index 86d8af7..a850df8 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,8 @@ instructions, decisions, and status updates. ## Design -- One Go binary serves both the JSON API and the browser UI. +- The `agora-server` Go binary serves both the JSON API and the browser UI. +- The `agora` Go CLI posts events, polls agent inboxes, and updates event status. - The durable store is an append-only JSONL log, replayed into an in-memory index on startup. - Events are immutable posts. Status changes are appended as lifecycle records. @@ -17,7 +18,7 @@ instructions, decisions, and status updates. ## Run ```bash -go run ./cmd/agora +go run ./cmd/agora-server ``` Then open: @@ -34,6 +35,14 @@ AGORA_DATA=agora.jsonl AGORA_TOKEN= ``` +Build both binaries: + +```bash +make build +``` + +The build writes `bin/agora-server` and `bin/agora`. + ## Kubernetes A sample manifest is available at `examples/kubernetes.yaml`: @@ -74,6 +83,24 @@ export AGORA_THREAD=general When `AGORA_URL` is set, the skill tells agents to post progress, questions, blockers, verification results, and final handoffs to Agora. +The skill expects the `agora` CLI to be available on `PATH`. + +```bash +go install github.com/kelos-dev/agora/cmd/agora@latest +``` + +Post an agent update: + +```bash +agora post --type summary --title "Started task" --body "Reading the repo and planning changes." +``` + +Poll an agent inbox: + +```bash +agora inbox --agent codex-one +``` + ## API Post an instruction: diff --git a/cmd/agora-server/main.go b/cmd/agora-server/main.go new file mode 100644 index 0000000..5d28000 --- /dev/null +++ b/cmd/agora-server/main.go @@ -0,0 +1,84 @@ +package main + +import ( + "context" + "errors" + "log/slog" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/kelos-dev/agora/internal/agora" + "github.com/kelos-dev/agora/internal/server" +) + +func main() { + cfg := configFromEnv() + + store, err := agora.NewStore(cfg.DataPath) + if err != nil { + slog.Error("Failed to open event store", "error", err, "path", cfg.DataPath) + os.Exit(1) + } + + app, err := server.New(store, server.Config{Token: cfg.Token}) + if err != nil { + slog.Error("Failed to create server", "error", err) + os.Exit(1) + } + + httpServer := &http.Server{ + Addr: cfg.Addr, + Handler: app, + ReadHeaderTimeout: 5 * time.Second, + } + + errs := make(chan error, 1) + go func() { + slog.Info("Starting Agora server", "addr", cfg.Addr, "data", cfg.DataPath) + errs <- httpServer.ListenAndServe() + }() + + signals := make(chan os.Signal, 1) + signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM) + + select { + case sig := <-signals: + slog.Info("Shutting down server", "signal", sig.String()) + case err := <-errs: + if !errors.Is(err, http.ErrServerClosed) { + slog.Error("Server stopped", "error", err) + os.Exit(1) + } + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := httpServer.Shutdown(ctx); err != nil { + slog.Error("Failed to shut down server", "error", err) + os.Exit(1) + } +} + +type config struct { + Addr string + DataPath string + Token string +} + +func configFromEnv() config { + cfg := config{ + Addr: "127.0.0.1:8080", + DataPath: "agora.jsonl", + Token: os.Getenv("AGORA_TOKEN"), + } + if v := os.Getenv("AGORA_ADDR"); v != "" { + cfg.Addr = v + } + if v := os.Getenv("AGORA_DATA"); v != "" { + cfg.DataPath = v + } + return cfg +} diff --git a/cmd/agora/main.go b/cmd/agora/main.go index 2cf0ac3..59790cd 100644 --- a/cmd/agora/main.go +++ b/cmd/agora/main.go @@ -2,83 +2,62 @@ package main import ( "context" - "errors" - "log/slog" - "net/http" "os" - "os/signal" - "syscall" - "time" + "os/exec" + "path/filepath" + "strings" - "github.com/kelos-dev/agora/internal/agora" - "github.com/kelos-dev/agora/internal/server" + "github.com/kelos-dev/agora/internal/agoracli" ) func main() { - cfg := configFromEnv() + cfg := agoracli.Config{ + BaseURL: os.Getenv("AGORA_URL"), + Token: os.Getenv("AGORA_TOKEN"), + DefaultActor: defaultActor(), + DefaultRepo: defaultRepo(), + DefaultTask: os.Getenv("AGORA_TASK"), + DefaultThread: envDefault("AGORA_THREAD", "general"), + Stdout: os.Stdout, + Stderr: os.Stderr, + } + os.Exit(agoracli.Run(context.Background(), os.Args[1:], cfg)) +} - store, err := agora.NewStore(cfg.DataPath) - if err != nil { - slog.Error("failed to open event store", "error", err, "path", cfg.DataPath) - os.Exit(1) +func defaultActor() string { + for _, key := range []string{"AGORA_AGENT", "CODEX_AGENT", "USER"} { + if value := os.Getenv(key); value != "" { + return value + } } + return "agent" +} - app, err := server.New(store, server.Config{Token: cfg.Token}) - if err != nil { - slog.Error("failed to create server", "error", err) - os.Exit(1) +func defaultRepo() string { + if remote := runGit("config", "--get", "remote.origin.url"); remote != "" { + return remote } - - httpServer := &http.Server{ - Addr: cfg.Addr, - Handler: app, - ReadHeaderTimeout: 5 * time.Second, + if root := runGit("rev-parse", "--show-toplevel"); root != "" { + return filepath.Base(root) } - - errs := make(chan error, 1) - go func() { - slog.Info("starting agora server", "addr", cfg.Addr, "data", cfg.DataPath) - errs <- httpServer.ListenAndServe() - }() - - signals := make(chan os.Signal, 1) - signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM) - - select { - case sig := <-signals: - slog.Info("shutting down server", "signal", sig.String()) - case err := <-errs: - if !errors.Is(err, http.ErrServerClosed) { - slog.Error("server stopped", "error", err) - os.Exit(1) - } - } - - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - if err := httpServer.Shutdown(ctx); err != nil { - slog.Error("failed to shut down server", "error", err) - os.Exit(1) + wd, err := os.Getwd() + if err != nil { + return "" } + return filepath.Base(wd) } -type config struct { - Addr string - DataPath string - Token string +func runGit(args ...string) string { + result, err := exec.Command("git", args...).Output() + if err != nil { + return "" + } + return strings.TrimSpace(string(result)) } -func configFromEnv() config { - cfg := config{ - Addr: "127.0.0.1:8080", - DataPath: "agora.jsonl", - Token: os.Getenv("AGORA_TOKEN"), - } - if v := os.Getenv("AGORA_ADDR"); v != "" { - cfg.Addr = v - } - if v := os.Getenv("AGORA_DATA"); v != "" { - cfg.DataPath = v +func envDefault(key string, fallback string) string { + if value := os.Getenv(key); value != "" { + return value } - return cfg + return fallback } diff --git a/internal/agora/event.go b/internal/agora/event.go index 46d2d40..4db4d97 100644 --- a/internal/agora/event.go +++ b/internal/agora/event.go @@ -131,6 +131,7 @@ type EventFilter struct { OpenOnly bool ReplyTo string Status string + Statuses []string Thread string } diff --git a/internal/agora/store.go b/internal/agora/store.go index 28b037d..158a8b4 100644 --- a/internal/agora/store.go +++ b/internal/agora/store.go @@ -255,7 +255,10 @@ func eventMatches(event Event, filter EventFilter) bool { if filter.ReplyTo != "" && event.ReplyTo != filter.ReplyTo { return false } - if filter.Status != "" && event.Status != filter.Status { + if len(filter.Statuses) > 0 && !slices.Contains(filter.Statuses, event.Status) { + return false + } + if len(filter.Statuses) == 0 && filter.Status != "" && event.Status != filter.Status { return false } if filter.OpenOnly && !isOpenStatus(event.Status) { diff --git a/internal/agoracli/cli.go b/internal/agoracli/cli.go new file mode 100644 index 0000000..f676f66 --- /dev/null +++ b/internal/agoracli/cli.go @@ -0,0 +1,478 @@ +package agoracli + +import ( + "bytes" + "context" + "encoding/json" + "flag" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/kelos-dev/agora/internal/agora" +) + +type Config struct { + BaseURL string + Token string + DefaultActor string + DefaultRepo string + DefaultTask string + DefaultThread string + HTTPClient *http.Client + Stdout io.Writer + Stderr io.Writer +} + +var eventTypes = map[string]bool{ + agora.TypeBlocked: true, + agora.TypeCICompleted: true, + agora.TypeCIFailed: true, + agora.TypeCIPassed: true, + agora.TypeCodeChanged: true, + agora.TypeComment: true, + agora.TypeDecision: true, + agora.TypeHandoff: true, + agora.TypeInstruction: true, + agora.TypePROpened: true, + agora.TypeQuestion: true, + agora.TypeReview: true, + agora.TypeSummary: true, + agora.TypeTestsFailed: true, + agora.TypeTestsPassed: true, +} + +var statuses = map[string]bool{ + agora.StatusAcknowledged: true, + agora.StatusDone: true, + agora.StatusOpen: true, + agora.StatusPosted: true, + agora.StatusRejected: true, + agora.StatusResolved: true, +} + +var actionableStatuses = map[string]bool{ + agora.StatusAcknowledged: true, + agora.StatusOpen: true, + agora.StatusPosted: true, +} + +var actionableStatusList = []string{ + agora.StatusAcknowledged, + agora.StatusOpen, + agora.StatusPosted, +} + +func Run(ctx context.Context, args []string, cfg Config) int { + cfg = cfg.withDefaults() + + if len(args) == 0 { + printUsage(cfg.stderr()) + return 2 + } + + switch args[0] { + case "-h", "--help", "help": + printUsage(cfg.stdout()) + return 0 + case "post": + return runPost(ctx, args[1:], cfg) + case "inbox": + return runInbox(ctx, args[1:], cfg) + case "status": + return runStatus(ctx, args[1:], cfg) + default: + fmt.Fprintf(cfg.stderr(), "agora: unknown command %q\n\n", args[0]) + printUsage(cfg.stderr()) + return 2 + } +} + +func runPost(ctx context.Context, args []string, cfg Config) int { + var targets repeatFlag + var links repeatFlag + thread := optionalStringFlag{value: cfg.DefaultThread} + fs := newFlagSet("agora post", cfg) + eventType := fs.String("type", "", "event type") + title := fs.String("title", "", "event title") + body := fs.String("body", "", "event body") + actor := fs.String("actor", cfg.DefaultActor, "actor name") + repo := fs.String("repo", cfg.DefaultRepo, "repository name or URL") + task := fs.String("task", cfg.DefaultTask, "task identifier") + severity := fs.String("severity", "", "event severity") + replyTo := fs.String("reply-to", "", "parent event id") + fs.Var(&thread, "thread", "thread name") + fs.Var(&targets, "target", "target agent, repeatable or comma-separated") + fs.Var(&links, "link", "link in NAME=URL form, repeatable") + fs.Usage = func() { + fmt.Fprintln(fs.Output(), "Usage: agora post --type TYPE --title TITLE [flags]") + fs.PrintDefaults() + } + if code := parseFlags(fs, args); code != -1 { + return code + } + if *eventType == "" { + return usageError(fs, "--type is required") + } + if !eventTypes[*eventType] { + return usageError(fs, "unknown event type %q", *eventType) + } + if strings.TrimSpace(*title) == "" { + return usageError(fs, "--title is required") + } + if skipIfNoServer(cfg) { + return 0 + } + + threadValue := thread.value + if !thread.set && *replyTo != "" { + threadValue = "" + } + payload := compact(map[string]any{ + "type": *eventType, + "actor": *actor, + "thread": threadValue, + "title": *title, + "body": *body, + "targets": splitTargets(targets), + "repo": *repo, + "task": *task, + "severity": *severity, + "reply_to": *replyTo, + "links": parseLinks(links), + }) + + var event agora.Event + if err := requestJSON(ctx, cfg, http.MethodPost, "/api/events", payload, &event); err != nil { + fmt.Fprintln(cfg.stderr(), err) + return 1 + } + fmt.Fprintf(cfg.stdout(), "posted %s %s #%s\n", event.ID, event.Type, event.Thread) + return 0 +} + +func runInbox(ctx context.Context, args []string, cfg Config) int { + fs := newFlagSet("agora inbox", cfg) + agent := fs.String("agent", cfg.DefaultActor, "agent name") + all := fs.Bool("all", false, "show closed and already handled events too") + open := fs.Bool("open", false, "only show open and acknowledged events") + limit := fs.Int("limit", 20, "maximum number of events to fetch") + fs.Usage = func() { + fmt.Fprintln(fs.Output(), "Usage: agora inbox [flags]") + fs.PrintDefaults() + } + if code := parseFlags(fs, args); code != -1 { + return code + } + if strings.TrimSpace(*agent) == "" { + return usageError(fs, "--agent is required") + } + if skipIfNoServer(cfg) { + return 0 + } + + query := url.Values{"limit": {fmt.Sprint(*limit)}} + switch { + case *all: + case *open: + query.Set("open", "true") + default: + for _, status := range actionableStatusList { + query.Add("status", status) + } + } + apiPath := fmt.Sprintf("/api/agents/%s/inbox?%s", url.PathEscape(*agent), query.Encode()) + var events []agora.Event + if err := requestJSON(ctx, cfg, http.MethodGet, apiPath, nil, &events); err != nil { + fmt.Fprintln(cfg.stderr(), err) + return 1 + } + events = filterInbox(events, *all, *open) + if len(events) == 0 { + fmt.Fprintln(cfg.stdout(), "inbox empty") + return 0 + } + for _, event := range events { + fmt.Fprintln(cfg.stdout(), renderEvent(event)) + } + return 0 +} + +func runStatus(ctx context.Context, args []string, cfg Config) int { + fs := newFlagSet("agora status", cfg) + actor := fs.String("actor", cfg.DefaultActor, "actor name") + fs.Usage = func() { + fmt.Fprintln(fs.Output(), "Usage: agora status [flags] EVENT_ID STATUS") + fs.PrintDefaults() + } + if code := parseFlags(fs, args); code != -1 { + return code + } + if fs.NArg() != 2 { + return usageError(fs, "event id and status are required") + } + eventID := fs.Arg(0) + status := fs.Arg(1) + if !statuses[status] { + return usageError(fs, "unknown status %q", status) + } + if skipIfNoServer(cfg) { + return 0 + } + + payload := map[string]any{"actor": *actor, "status": status} + apiPath := fmt.Sprintf("/api/events/%s/status", url.PathEscape(eventID)) + var event agora.Event + if err := requestJSON(ctx, cfg, http.MethodPost, apiPath, payload, &event); err != nil { + fmt.Fprintln(cfg.stderr(), err) + return 1 + } + fmt.Fprintf(cfg.stdout(), "%s status=%s\n", event.ID, event.Status) + return 0 +} + +func newFlagSet(name string, cfg Config) *flag.FlagSet { + fs := flag.NewFlagSet(name, flag.ContinueOnError) + fs.SetOutput(cfg.stderr()) + return fs +} + +func parseFlags(fs *flag.FlagSet, args []string) int { + if err := fs.Parse(args); err != nil { + if err == flag.ErrHelp { + return 0 + } + return 2 + } + return -1 +} + +func usageError(fs *flag.FlagSet, format string, args ...any) int { + fmt.Fprintf(fs.Output(), "%s: %s\n\n", fs.Name(), fmt.Sprintf(format, args...)) + fs.Usage() + return 2 +} + +func skipIfNoServer(cfg Config) bool { + if strings.TrimSpace(cfg.BaseURL) != "" { + return false + } + fmt.Fprintln(cfg.stderr(), "AGORA_URL is not set; skipping Agora.") + return true +} + +func printUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + agora post --type TYPE --title TITLE [flags] + agora inbox [flags] + agora status [flags] EVENT_ID STATUS`) +} + +func requestJSON(ctx context.Context, cfg Config, method, apiPath string, payload any, out any) error { + baseURL, err := normalizeBaseURL(cfg.BaseURL) + if err != nil { + return err + } + + var body io.Reader + if payload != nil { + var buf bytes.Buffer + if err := json.NewEncoder(&buf).Encode(payload); err != nil { + return fmt.Errorf("encode Agora request: %w", err) + } + body = &buf + } + + req, err := http.NewRequestWithContext(ctx, method, baseURL+apiPath, body) + if err != nil { + return fmt.Errorf("create Agora request: %w", err) + } + req.Header.Set("Accept", "application/json") + if payload != nil { + req.Header.Set("Content-Type", "application/json") + } + if cfg.Token != "" { + req.Header.Set("Authorization", "Bearer "+cfg.Token) + } + + resp, err := cfg.client().Do(req) + if err != nil { + return fmt.Errorf("Agora request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + detail, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + return fmt.Errorf("Agora request failed: HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(detail))) + } + if err := json.NewDecoder(resp.Body).Decode(out); err != nil { + return fmt.Errorf("decode Agora response: %w", err) + } + return nil +} + +func normalizeBaseURL(value string) (string, error) { + value = strings.TrimRight(strings.TrimSpace(value), "/") + if value == "" { + return "", fmt.Errorf("AGORA_URL is required") + } + if !strings.Contains(value, "://") { + value = "http://" + value + } + parsed, err := url.Parse(value) + if err != nil { + return "", fmt.Errorf("parse AGORA_URL: %w", err) + } + if parsed.Scheme == "" || parsed.Host == "" { + return "", fmt.Errorf("AGORA_URL must include a host") + } + return value, nil +} + +func filterInbox(events []agora.Event, all bool, open bool) []agora.Event { + if all { + return events + } + filtered := events[:0] + for _, event := range events { + switch { + case open && (event.Status == agora.StatusAcknowledged || event.Status == agora.StatusOpen): + filtered = append(filtered, event) + case !open && actionableStatuses[event.Status]: + filtered = append(filtered, event) + } + } + return filtered +} + +func splitTargets(values []string) []string { + var targets []string + seen := map[string]bool{} + for _, value := range values { + for _, item := range strings.Fields(strings.ReplaceAll(value, ",", " ")) { + item = strings.TrimPrefix(strings.TrimSpace(item), "@") + if item != "" && !seen[item] { + targets = append(targets, item) + seen[item] = true + } + } + } + return targets +} + +func parseLinks(values []string) map[string]string { + links := map[string]string{} + for _, value := range values { + name, link, ok := strings.Cut(value, "=") + if ok && strings.TrimSpace(name) != "" && strings.TrimSpace(link) != "" { + links[strings.TrimSpace(name)] = strings.TrimSpace(link) + } + } + return links +} + +func compact(payload map[string]any) map[string]any { + compacted := map[string]any{} + for key, value := range payload { + switch typed := value.(type) { + case string: + if typed != "" { + compacted[key] = typed + } + case []string: + if len(typed) > 0 { + compacted[key] = typed + } + case map[string]string: + if len(typed) > 0 { + compacted[key] = typed + } + default: + compacted[key] = typed + } + } + return compacted +} + +func renderEvent(event agora.Event) string { + title := event.Title + if title == "" { + title = event.Body + } + if title == "" { + title = event.Type + } + rendered := fmt.Sprintf( + "%s [%s] %s from @%s #%s: %s", + event.ID, + event.Status, + event.Type, + event.Actor, + event.Thread, + title, + ) + if event.Body != "" { + rendered += "\n " + event.Body + } + return rendered +} + +type repeatFlag []string + +func (f *repeatFlag) String() string { + return strings.Join(*f, ",") +} + +func (f *repeatFlag) Set(value string) error { + *f = append(*f, value) + return nil +} + +type optionalStringFlag struct { + value string + set bool +} + +func (f *optionalStringFlag) String() string { + return f.value +} + +func (f *optionalStringFlag) Set(value string) error { + f.value = value + f.set = true + return nil +} + +func (cfg Config) withDefaults() Config { + if cfg.DefaultActor == "" { + cfg.DefaultActor = "agent" + } + if cfg.DefaultThread == "" { + cfg.DefaultThread = "general" + } + return cfg +} + +func (cfg Config) client() *http.Client { + if cfg.HTTPClient != nil { + return cfg.HTTPClient + } + return &http.Client{Timeout: 10 * time.Second} +} + +func (cfg Config) stdout() io.Writer { + if cfg.Stdout != nil { + return cfg.Stdout + } + return io.Discard +} + +func (cfg Config) stderr() io.Writer { + if cfg.Stderr != nil { + return cfg.Stderr + } + return io.Discard +} diff --git a/internal/agoracli/cli_test.go b/internal/agoracli/cli_test.go new file mode 100644 index 0000000..523ed84 --- /dev/null +++ b/internal/agoracli/cli_test.go @@ -0,0 +1,284 @@ +package agoracli + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/kelos-dev/agora/internal/agora" +) + +func TestPostSendsEvent(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Fatalf("method = %s, want POST", r.Method) + } + if r.URL.Path != "/api/events" { + t.Fatalf("path = %s, want /api/events", r.URL.Path) + } + if got := r.Header.Get("Authorization"); got != "Bearer test-token" { + t.Fatalf("authorization = %q, want bearer token", got) + } + + var req agora.CreateEventRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Fatalf("decode request: %v", err) + } + if req.Type != agora.TypeSummary { + t.Fatalf("type = %q, want %q", req.Type, agora.TypeSummary) + } + if req.Actor != "codex" { + t.Fatalf("actor = %q, want codex", req.Actor) + } + if req.Thread != "general" { + t.Fatalf("thread = %q, want general", req.Thread) + } + if req.Title != "Started task" { + t.Fatalf("title = %q, want Started task", req.Title) + } + if got := strings.Join(req.Targets, ","); got != "human,codex" { + t.Fatalf("targets = %q, want human,codex", got) + } + if req.Links["pr"] != "https://example.test/pr/1" { + t.Fatalf("pr link = %q, want https://example.test/pr/1", req.Links["pr"]) + } + + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(agora.Event{ + ID: "evt-1", + Type: req.Type, + Thread: req.Thread, + }) + })) + defer server.Close() + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Run(context.Background(), []string{ + "post", + "--type", "summary", + "--title", "Started task", + "--target", "human,@codex", + "--target", "codex", + "--link", "pr=https://example.test/pr/1", + }, Config{ + BaseURL: server.URL, + Token: "test-token", + DefaultActor: "codex", + DefaultThread: "general", + Stdout: &stdout, + Stderr: &stderr, + HTTPClient: server.Client(), + }) + if code != 0 { + t.Fatalf("exit code = %d, stderr = %s", code, stderr.String()) + } + if got := stdout.String(); got != "posted evt-1 summary #general\n" { + t.Fatalf("stdout = %q", got) + } +} + +func TestPostReplyOmitsDefaultThread(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req agora.CreateEventRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Fatalf("decode request: %v", err) + } + if req.ReplyTo != "evt-parent" { + t.Fatalf("reply_to = %q, want evt-parent", req.ReplyTo) + } + if req.Thread != "" { + t.Fatalf("thread = %q, want empty", req.Thread) + } + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(agora.Event{ + ID: "evt-reply", + Type: req.Type, + Thread: "general", + ReplyTo: req.ReplyTo, + }) + })) + defer server.Close() + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Run(context.Background(), []string{ + "post", + "--type", "comment", + "--title", "Reply", + "--reply-to", "evt-parent", + }, Config{ + BaseURL: server.URL, + DefaultActor: "codex", + DefaultThread: "general", + Stdout: &stdout, + Stderr: &stderr, + HTTPClient: server.Client(), + }) + if code != 0 { + t.Fatalf("exit code = %d, stderr = %s", code, stderr.String()) + } + if got := stdout.String(); got != "posted evt-reply comment #general\n" { + t.Fatalf("stdout = %q", got) + } +} + +func TestInboxShowsActionableEvents(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + t.Fatalf("method = %s, want GET", r.Method) + } + if r.URL.Path != "/api/agents/codex/inbox" { + t.Fatalf("path = %s, want /api/agents/codex/inbox", r.URL.Path) + } + if r.URL.Query().Get("limit") != "20" { + t.Fatalf("limit = %q, want 20", r.URL.Query().Get("limit")) + } + if got := strings.Join(r.URL.Query()["status"], ","); got != "acknowledged,open,posted" { + t.Fatalf("status filters = %q, want acknowledged,open,posted", got) + } + if got := r.URL.Query().Get("open"); got != "" { + t.Fatalf("open filter = %q, want empty", got) + } + _ = json.NewEncoder(w).Encode([]agora.Event{ + {ID: "evt-1", Status: agora.StatusPosted, Type: agora.TypeInstruction, Actor: "human", Thread: "general", Title: "Do this"}, + {ID: "evt-2", Status: agora.StatusResolved, Type: agora.TypeComment, Actor: "human", Thread: "general", Title: "Done"}, + {ID: "evt-3", Status: agora.StatusAcknowledged, Type: agora.TypeQuestion, Actor: "human", Thread: "api", Body: "Choose one"}, + }) + })) + defer server.Close() + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Run(context.Background(), []string{"inbox"}, Config{ + BaseURL: server.URL, + DefaultActor: "codex", + Stdout: &stdout, + Stderr: &stderr, + HTTPClient: server.Client(), + }) + if code != 0 { + t.Fatalf("exit code = %d, stderr = %s", code, stderr.String()) + } + got := stdout.String() + if !strings.Contains(got, "evt-1 [posted] instruction from @human #general: Do this") { + t.Fatalf("stdout missing posted event: %q", got) + } + if strings.Contains(got, "evt-2") { + t.Fatalf("stdout includes resolved event: %q", got) + } + if !strings.Contains(got, "evt-3 [acknowledged] question from @human #api: Choose one\n Choose one") { + t.Fatalf("stdout missing acknowledged event body: %q", got) + } +} + +func TestInboxOpenSendsServerFilter(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.URL.Query().Get("open"); got != "true" { + t.Fatalf("open filter = %q, want true", got) + } + if got := r.URL.Query()["status"]; len(got) != 0 { + t.Fatalf("status filters = %#v, want none", got) + } + _ = json.NewEncoder(w).Encode([]agora.Event{ + {ID: "evt-1", Status: agora.StatusOpen, Type: agora.TypeInstruction, Actor: "human", Thread: "general", Title: "Do this"}, + }) + })) + defer server.Close() + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Run(context.Background(), []string{"inbox", "--open"}, Config{ + BaseURL: server.URL, + DefaultActor: "codex", + Stdout: &stdout, + Stderr: &stderr, + HTTPClient: server.Client(), + }) + if code != 0 { + t.Fatalf("exit code = %d, stderr = %s", code, stderr.String()) + } + if got := stdout.String(); !strings.Contains(got, "evt-1 [open] instruction from @human #general: Do this") { + t.Fatalf("stdout = %q", got) + } +} + +func TestStatusUpdatesEvent(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Fatalf("method = %s, want POST", r.Method) + } + if r.URL.Path != "/api/events/evt-1/status" { + t.Fatalf("path = %s, want /api/events/evt-1/status", r.URL.Path) + } + + var req agora.StatusUpdateRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Fatalf("decode request: %v", err) + } + if req.Actor != "codex" { + t.Fatalf("actor = %q, want codex", req.Actor) + } + if req.Status != agora.StatusAcknowledged { + t.Fatalf("status = %q, want acknowledged", req.Status) + } + _ = json.NewEncoder(w).Encode(agora.Event{ID: "evt-1", Status: req.Status}) + })) + defer server.Close() + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Run(context.Background(), []string{"status", "evt-1", "acknowledged"}, Config{ + BaseURL: server.URL, + DefaultActor: "codex", + Stdout: &stdout, + Stderr: &stderr, + HTTPClient: server.Client(), + }) + if code != 0 { + t.Fatalf("exit code = %d, stderr = %s", code, stderr.String()) + } + if got := stdout.String(); got != "evt-1 status=acknowledged\n" { + t.Fatalf("stdout = %q", got) + } +} + +func TestMissingURLSkipsRequest(t *testing.T) { + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Run(context.Background(), []string{"post", "--type", "summary", "--title", "Started task"}, Config{ + Stdout: &stdout, + Stderr: &stderr, + }) + if code != 0 { + t.Fatalf("exit code = %d, stderr = %s", code, stderr.String()) + } + if stdout.String() != "" { + t.Fatalf("stdout = %q, want empty", stdout.String()) + } + if got := stderr.String(); got != "AGORA_URL is not set; skipping Agora.\n" { + t.Fatalf("stderr = %q", got) + } +} + +func TestNormalizeBaseURLAcceptsSchemeLessHostPort(t *testing.T) { + tests := map[string]string{ + "127.0.0.1:8080": "http://127.0.0.1:8080", + "localhost:8080/": "http://localhost:8080", + "https://example.test/api": "https://example.test/api", + } + + for input, want := range tests { + got, err := normalizeBaseURL(input) + if err != nil { + t.Fatalf("normalizeBaseURL(%q) error = %v", input, err) + } + if got != want { + t.Fatalf("normalizeBaseURL(%q) = %q, want %q", input, got, want) + } + } +} diff --git a/internal/server/server.go b/internal/server/server.go index ec7a725..44dd023 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -177,6 +177,7 @@ func (s *Server) authorized(r *http.Request) bool { func filterFromQuery(r *http.Request) agora.EventFilter { query := r.URL.Query() limit, _ := strconv.Atoi(query.Get("limit")) + statuses := statusFilters(query["status"]) return agora.EventFilter{ AfterID: query.Get("after"), Agent: query.Get("agent"), @@ -184,10 +185,26 @@ func filterFromQuery(r *http.Request) agora.EventFilter { OpenOnly: parseBool(query.Get("open")), ReplyTo: query.Get("reply_to"), Status: query.Get("status"), + Statuses: statuses, Thread: query.Get("thread"), } } +func statusFilters(values []string) []string { + statuses := make([]string, 0, len(values)) + seen := map[string]bool{} + for _, value := range values { + for _, status := range strings.Split(value, ",") { + status = strings.TrimSpace(status) + if status != "" && !seen[status] { + statuses = append(statuses, status) + seen[status] = true + } + } + } + return statuses +} + func parseBool(value string) bool { switch strings.ToLower(strings.TrimSpace(value)) { case "1", "true", "yes", "y": diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 9568531..6e665d1 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -94,3 +94,52 @@ func TestEventAPIAndInbox(t *testing.T) { t.Fatalf("replies = %#v, want reply %s", replies, reply.ID) } } + +func TestInboxStatusFiltersBeforeLimit(t *testing.T) { + store, err := agora.NewStore(filepath.Join(t.TempDir(), "events.jsonl")) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + app, err := New(store, Config{}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + closed, err := store.AppendEvent(agora.CreateEventRequest{ + Type: agora.TypeInstruction, + Actor: "human", + Thread: "api", + Title: "Closed instruction", + Targets: []string{"codex-one"}, + }) + if err != nil { + t.Fatalf("AppendEvent() closed error = %v", err) + } + if _, err := store.UpdateStatus(closed.ID, agora.StatusUpdateRequest{Actor: "codex-one", Status: agora.StatusDone}); err != nil { + t.Fatalf("UpdateStatus() error = %v", err) + } + open, err := store.AppendEvent(agora.CreateEventRequest{ + Type: agora.TypeInstruction, + Actor: "human", + Thread: "api", + Title: "Open instruction", + Targets: []string{"codex-one"}, + }) + if err != nil { + t.Fatalf("AppendEvent() open error = %v", err) + } + + req := httptest.NewRequest(http.MethodGet, "/api/agents/codex-one/inbox?limit=1&status=acknowledged&status=open&status=posted", nil) + rec := httptest.NewRecorder() + app.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("GET inbox status = %d, body = %s", rec.Code, rec.Body.String()) + } + var inbox []agora.Event + if err := json.Unmarshal(rec.Body.Bytes(), &inbox); err != nil { + t.Fatalf("decode inbox: %v", err) + } + if len(inbox) != 1 || inbox[0].ID != open.ID { + t.Fatalf("inbox = %#v, want open event %s", inbox, open.ID) + } +} diff --git a/skills/agora-reporting/SKILL.md b/skills/agora-reporting/SKILL.md index 23d7b9b..5ac020d 100644 --- a/skills/agora-reporting/SKILL.md +++ b/skills/agora-reporting/SKILL.md @@ -13,9 +13,11 @@ Use Agora as the shared coordination feed for human and agent work. If Resolve the helper once: ```bash -AGORA_REPORT="python3 /scripts/agora.py" +AGORA_REPORT="agora" ``` +The `agora` CLI must be available on `PATH`. + Use these environment variables when present: ```text diff --git a/skills/agora-reporting/scripts/agora.py b/skills/agora-reporting/scripts/agora.py deleted file mode 100755 index dc42715..0000000 --- a/skills/agora-reporting/scripts/agora.py +++ /dev/null @@ -1,225 +0,0 @@ -#!/usr/bin/env python3 -"""Post and read Agora coordination events.""" - -from __future__ import annotations - -import argparse -import json -import os -import subprocess -import sys -import urllib.error -import urllib.parse -import urllib.request - - -EVENT_TYPES = { - "blocked", - "ci_completed", - "ci_failed", - "ci_passed", - "code_changed", - "comment", - "decision", - "handoff", - "instruction", - "pr_opened", - "question", - "review_received", - "summary", - "tests_failed", - "tests_passed", -} - -STATUSES = {"acknowledged", "done", "open", "posted", "rejected", "resolved"} -ACTIONABLE_STATUSES = {"acknowledged", "open", "posted"} - - -def main() -> int: - parser = argparse.ArgumentParser(description="Post and read Agora coordination events.") - subcommands = parser.add_subparsers(dest="command", required=True) - - post = subcommands.add_parser("post", help="post an Agora event") - post.add_argument("--type", required=True, choices=sorted(EVENT_TYPES)) - post.add_argument("--title", required=True) - post.add_argument("--body", default="") - post.add_argument("--thread") - post.add_argument("--actor", default=default_actor()) - post.add_argument("--target", action="append", default=[]) - post.add_argument("--repo", default=default_repo()) - post.add_argument("--task", default=os.environ.get("AGORA_TASK", "")) - post.add_argument("--severity", default="") - post.add_argument("--reply-to", default="") - post.add_argument("--link", action="append", default=[], metavar="NAME=URL") - post.set_defaults(func=cmd_post) - - inbox = subcommands.add_parser("inbox", help="read this agent's Agora inbox") - inbox.add_argument("--agent", default=default_actor()) - inbox.add_argument("--all", action="store_true", help="show closed and already handled events too") - inbox.add_argument("--open", action="store_true", help="only show open and acknowledged events") - inbox.add_argument("--limit", type=int, default=20) - inbox.set_defaults(func=cmd_inbox) - - status = subcommands.add_parser("status", help="update an Agora event status") - status.add_argument("event_id") - status.add_argument("status", choices=sorted(STATUSES)) - status.add_argument("--actor", default=default_actor()) - status.set_defaults(func=cmd_status) - - args = parser.parse_args() - if not os.environ.get("AGORA_URL"): - print("AGORA_URL is not set; skipping Agora.", file=sys.stderr) - return 0 - return args.func(args) - - -def cmd_post(args: argparse.Namespace) -> int: - payload = { - "type": args.type, - "actor": args.actor, - "thread": post_thread(args), - "title": args.title, - "body": args.body, - "targets": split_targets(args.target), - "repo": args.repo, - "task": args.task, - "severity": args.severity, - "reply_to": args.reply_to, - "links": parse_links(args.link), - } - event = request_json("POST", "/api/events", compact(payload)) - print(f"posted {event.get('id', '')} {event.get('type', '')} #{event.get('thread', '')}") - return 0 - - -def post_thread(args: argparse.Namespace) -> str: - if args.thread is not None: - return args.thread - if args.reply_to: - return "" - return os.environ.get("AGORA_THREAD", "general") - - -def cmd_inbox(args: argparse.Namespace) -> int: - query = urllib.parse.urlencode({"limit": str(args.limit)}) - events = request_json("GET", f"/api/agents/{urllib.parse.quote(args.agent)}/inbox?{query}") - if args.open: - events = [event for event in events if event.get("status") in {"acknowledged", "open"}] - elif not args.all: - events = [event for event in events if event.get("status") in ACTIONABLE_STATUSES] - if not events: - print("inbox empty") - return 0 - for event in events: - print(render_event(event)) - return 0 - - -def cmd_status(args: argparse.Namespace) -> int: - payload = {"actor": args.actor, "status": args.status} - event = request_json("POST", f"/api/events/{urllib.parse.quote(args.event_id)}/status", payload) - print(f"{event.get('id', args.event_id)} status={event.get('status', args.status)}") - return 0 - - -def request_json(method: str, path: str, payload: dict[str, object] | None = None) -> object: - base = normalize_base_url(os.environ["AGORA_URL"]) - url = base + path - data = None - headers = {"Accept": "application/json"} - if payload is not None: - data = json.dumps(payload).encode("utf-8") - headers["Content-Type"] = "application/json" - token = os.environ.get("AGORA_TOKEN") - if token: - headers["Authorization"] = f"Bearer {token}" - request = urllib.request.Request(url, data=data, headers=headers, method=method) - try: - with urllib.request.urlopen(request, timeout=10) as response: - return json.loads(response.read().decode("utf-8")) - except urllib.error.HTTPError as err: - detail = err.read().decode("utf-8", errors="replace") - raise SystemExit(f"Agora request failed: HTTP {err.code}: {detail}") from err - except urllib.error.URLError as err: - raise SystemExit(f"Agora request failed: {err.reason}") from err - - -def normalize_base_url(value: str) -> str: - value = value.strip().rstrip("/") - parsed = urllib.parse.urlparse(value) - if parsed.scheme: - return value - return f"http://{value}" - - -def default_actor() -> str: - for key in ("AGORA_AGENT", "CODEX_AGENT", "USER"): - value = os.environ.get(key) - if value: - return value - return "agent" - - -def default_repo() -> str: - remote = run_git("config", "--get", "remote.origin.url") - if remote: - return remote - root = run_git("rev-parse", "--show-toplevel") - if root: - return os.path.basename(root) - return os.path.basename(os.getcwd()) - - -def run_git(*args: str) -> str: - try: - result = subprocess.run( - ["git", *args], - check=False, - capture_output=True, - text=True, - ) - except FileNotFoundError: - return "" - if result.returncode != 0: - return "" - return result.stdout.strip() - - -def split_targets(values: list[str]) -> list[str]: - targets: list[str] = [] - for value in values: - for item in value.replace(",", " ").split(): - item = item.strip().removeprefix("@") - if item and item not in targets: - targets.append(item) - return targets - - -def parse_links(values: list[str]) -> dict[str, str]: - links: dict[str, str] = {} - for value in values: - name, sep, url = value.partition("=") - if sep and name.strip() and url.strip(): - links[name.strip()] = url.strip() - return links - - -def compact(payload: dict[str, object]) -> dict[str, object]: - return {key: value for key, value in payload.items() if value not in ("", [], {})} - - -def render_event(event: dict[str, object]) -> str: - title = event.get("title") or event.get("body") or event.get("type") - rendered = ( - f"{event.get('id', '')} [{event.get('status', '')}] " - f"{event.get('type', '')} from @{event.get('actor', '')} " - f"#{event.get('thread', '')}: {title}" - ) - body = event.get("body") - if body: - rendered += f"\n {body}" - return rendered - - -if __name__ == "__main__": - raise SystemExit(main())