diff --git a/go.mod b/go.mod index 3ee31cb..384cf0b 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.25.0 require ( connectrpc.com/connect v1.19.2 - github.com/chaitin/agent-compose/proto v0.1.8 + github.com/chaitin/agent-compose/proto v0.1.12 github.com/go-openapi/errors v0.22.7 github.com/go-openapi/runtime v0.29.3 github.com/go-openapi/strfmt v0.26.1 diff --git a/go.sum b/go.sum index 6426c03..716aeb5 100644 --- a/go.sum +++ b/go.sum @@ -2,8 +2,8 @@ connectrpc.com/connect v1.19.2 h1:McQ83FGdzL+t60peksi0gXC7MQ/iLKgLduAnThbM0mo= connectrpc.com/connect v1.19.2/go.mod h1:tN20fjdGlewnSFeZxLKb0xwIZ6ozc3OQs2hTXy4du9w= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= -github.com/chaitin/agent-compose/proto v0.1.8 h1:OeUGngog8NRpZjkHL9tXz7IbuOrNeTfhOK3bOAPCwGI= -github.com/chaitin/agent-compose/proto v0.1.8/go.mod h1:ctLhXiWw/iKsBeanJ6J0bllf+WCFFpY5rrcN7tm6bgU= +github.com/chaitin/agent-compose/proto v0.1.12 h1:pdj67C41+PAoBHd8UVBFk5Nd3a8oBoU9e1OqPejNZTU= +github.com/chaitin/agent-compose/proto v0.1.12/go.mod h1:ctLhXiWw/iKsBeanJ6J0bllf+WCFFpY5rrcN7tm6bgU= github.com/cpuguy83/go-md2man/v2 v2.0.6 h1:XJtiaUW6dEEqVuZiMTn1ldk455QWwEIsMIJlo5vtkx0= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= diff --git a/products/agentcompose/auth.go b/products/agentcompose/auth.go index cd14b1c..9cf70d9 100644 --- a/products/agentcompose/auth.go +++ b/products/agentcompose/auth.go @@ -1,16 +1,17 @@ package agentcompose import ( + "context" "fmt" "io" "os" "strings" "connectrpc.com/connect" + agentcomposev2 "github.com/chaitin/agent-compose/proto/agentcompose/v2" "github.com/chaitin/chaitin-cli/config" "github.com/spf13/cobra" "golang.org/x/term" - "google.golang.org/protobuf/types/known/emptypb" ) func newAuthCommand(state *commandState) *cobra.Command { @@ -28,7 +29,7 @@ func newAuthCommand(state *commandState) *cobra.Command { state.options.Token = token ctx, cancel := requestContext(cmd, state) defer cancel() - _, err = state.clients().health.Status(ctx, connect.NewRequest(&emptypb.Empty{})) + err = verifyToken(ctx, state) if err != nil { state.options.Token = oldToken return mapConnectError(err, state.options.URL, state.options.JSON) @@ -137,8 +138,7 @@ func runAuthStatus(cmd *cobra.Command, state *commandState) error { } ctx, cancel := requestContext(cmd, state) defer cancel() - _, err := state.clients().health.Status(ctx, connect.NewRequest(&emptypb.Empty{})) - if err != nil { + if err := verifyToken(ctx, state); err != nil { return mapConnectError(err, state.options.URL, state.options.JSON) } valid = true @@ -148,3 +148,10 @@ func runAuthStatus(cmd *cobra.Command, state *commandState) error { fmt.Fprintf(cmd.OutOrStdout(), "URL: %s\nToken source: %s\nToken configured: yes\nToken valid: yes\n", state.options.URL, state.options.TokenSource) return nil } + +// verifyToken calls a protected RPC. HealthService is exempt from daemon +// authentication, so it accepts any Token and cannot validate one. +func verifyToken(ctx context.Context, state *commandState) error { + _, err := state.clients().project.ListProjects(ctx, connect.NewRequest(&agentcomposev2.ListProjectsRequest{Limit: 1})) + return err +} diff --git a/products/agentcompose/client.go b/products/agentcompose/client.go index 8f973ce..cb8dfbf 100644 --- a/products/agentcompose/client.go +++ b/products/agentcompose/client.go @@ -11,6 +11,7 @@ type clients struct { exec agentcomposev2connect.ExecServiceClient sandbox agentcomposev2connect.SandboxServiceClient resource agentcomposev2connect.ResourceServiceClient + llm agentcomposev2connect.LLMServiceClient health healthv1connect.HealthServiceClient } @@ -23,6 +24,7 @@ func (s *commandState) clients() clients { exec: agentcomposev2connect.NewExecServiceClient(httpClient, baseURL), sandbox: agentcomposev2connect.NewSandboxServiceClient(httpClient, baseURL), resource: agentcomposev2connect.NewResourceServiceClient(httpClient, baseURL), + llm: agentcomposev2connect.NewLLMServiceClient(httpClient, baseURL), health: healthv1connect.NewHealthServiceClient(httpClient, baseURL), } } diff --git a/products/agentcompose/command.go b/products/agentcompose/command.go index cf53391..4a515ef 100644 --- a/products/agentcompose/command.go +++ b/products/agentcompose/command.go @@ -20,7 +20,7 @@ func NewCommand() *cobra.Command { } cmd.SetContext(context.WithValue(context.Background(), stateKey{}, state)) cmd.PersistentFlags().StringVar(&state.options.URL, "url", "", "Token/RBAC API base URL, including an optional path prefix") - cmd.PersistentFlags().StringVar(&state.options.Project, "project", "", "Default project name or ID") + cmd.PersistentFlags().StringVarP(&state.options.Project, "project", "p", "", "Default project name or ID") cmd.PersistentFlags().StringVar(&state.options.timeoutText, "timeout", "30s", "Unary request timeout") cmd.PersistentFlags().BoolVar(&state.options.Insecure, "insecure", false, "Skip TLS certificate verification") cmd.PersistentFlags().BoolVar(&state.options.JSON, "json", false, "Output JSON or NDJSON") @@ -42,6 +42,7 @@ func NewCommand() *cobra.Command { newSchedulerCommand(state), newSandboxCommand(state), newExecCommand(state), + newLLMCommand(state), ) return cmd } diff --git a/products/agentcompose/diagnostics_test.go b/products/agentcompose/diagnostics_test.go new file mode 100644 index 0000000..1a6a304 --- /dev/null +++ b/products/agentcompose/diagnostics_test.go @@ -0,0 +1,276 @@ +package agentcompose + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "testing" + "time" + + "connectrpc.com/connect" + agentcomposev2 "github.com/chaitin/agent-compose/proto/agentcompose/v2" + "github.com/chaitin/agent-compose/proto/agentcompose/v2/agentcomposev2connect" +) + +func (s *runStub) GetRun(_ context.Context, req *connect.Request[agentcomposev2.GetRunRequest]) (*connect.Response[agentcomposev2.GetRunResponse], error) { + detail, ok := s.details[req.Msg.GetRunId()] + if !ok { + return nil, connect.NewError(connect.CodeNotFound, fmt.Errorf("Run not found")) + } + return connect.NewResponse(&agentcomposev2.GetRunResponse{Run: detail}), nil +} + +type llmStub struct { + agentcomposev2connect.UnimplementedLLMServiceHandler + providers []*agentcomposev2.LLMProvider + defaultModel *agentcomposev2.LLMModelReference +} + +func (s *llmStub) ListProviders(_ context.Context, req *connect.Request[agentcomposev2.ListProvidersRequest]) (*connect.Response[agentcomposev2.ListProvidersResponse], error) { + start := min(int(req.Msg.GetOffset()), len(s.providers)) + end := min(start+int(req.Msg.GetLimit()), len(s.providers)) + return connect.NewResponse(&agentcomposev2.ListProvidersResponse{Providers: s.providers[start:end], Total: uint32(len(s.providers))}), nil +} + +func (s *llmStub) GetDefaultModel(context.Context, *connect.Request[agentcomposev2.GetDefaultModelRequest]) (*connect.Response[agentcomposev2.GetDefaultModelResponse], error) { + return connect.NewResponse(&agentcomposev2.GetDefaultModelResponse{Model: s.defaultModel}), nil +} + +func TestParseTimeBound(t *testing.T) { + now := time.Date(2026, 10, 15, 12, 0, 0, 123456789, time.UTC) + tests := []struct { + name, input string + want time.Time + wantNil bool + wantErr bool + }{ + {name: "empty", input: " ", wantNil: true}, + {name: "rfc3339", input: "2026-10-14T08:00:00Z", want: time.Date(2026, 10, 14, 8, 0, 0, 0, time.UTC)}, + {name: "rfc3339 sub-millisecond is truncated", input: "2026-10-14T08:00:00.123456Z", want: time.Date(2026, 10, 14, 8, 0, 0, 123000000, time.UTC)}, + {name: "minutes", input: "30m", want: time.Date(2026, 10, 15, 11, 30, 0, 123000000, time.UTC)}, + {name: "hours", input: "24h", want: time.Date(2026, 10, 14, 12, 0, 0, 123000000, time.UTC)}, + {name: "days", input: "7d", want: time.Date(2026, 10, 8, 12, 0, 0, 123000000, time.UTC)}, + {name: "zero duration", input: "0s", wantErr: true}, + {name: "negative duration", input: "-1h", wantErr: true}, + {name: "fractional days", input: "1.5d", wantErr: true}, + {name: "garbage", input: "yesterday", wantErr: true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got, err := parseTimeBound(test.input, now) + if (err != nil) != test.wantErr { + t.Fatalf("error = %v, wantErr %t", err, test.wantErr) + } + if test.wantErr { + return + } + if test.wantNil { + if got != nil { + t.Fatalf("got %v, want nil", got) + } + return + } + if !got.AsTime().Equal(test.want) { + t.Fatalf("got %s, want %s", got.AsTime(), test.want) + } + }) + } +} + +func TestParseLabels(t *testing.T) { + tests := []struct { + name string + input []string + want map[string]string + wantErr bool + }{ + {name: "none", input: nil, want: nil}, + {name: "pairs", input: []string{"env=prod", "team=sec"}, want: map[string]string{"env": "prod", "team": "sec"}}, + {name: "empty value", input: []string{"flag="}, want: map[string]string{"flag": ""}}, + {name: "value with equals", input: []string{"expr=a=b"}, want: map[string]string{"expr": "a=b"}}, + {name: "missing equals", input: []string{"env"}, wantErr: true}, + {name: "empty key", input: []string{"=prod"}, wantErr: true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got, err := parseLabels(test.input) + if (err != nil) != test.wantErr { + t.Fatalf("error = %v, wantErr %t", err, test.wantErr) + } + if len(got) != len(test.want) { + t.Fatalf("got %v, want %v", got, test.want) + } + for key, value := range test.want { + if got[key] != value { + t.Fatalf("got %v, want %v", got, test.want) + } + } + }) + } +} + +func TestRunListForwardsDiagnosticFilters(t *testing.T) { + project := &projectStub{project: fixtureProject()} + run := &runStub{} + server, _ := newTestServer(t, project, run) + before := time.Now() + _, _, err := executeCommand(t, server.URL, false, "--json", "run", "ls", "--started-from", "24h", "--scheduler-run", "srun-1", "--label", "env=prod", "--label", "team=sec") + if err != nil { + t.Fatal(err) + } + if len(run.listRequests) != 1 { + t.Fatalf("requests = %+v", run.listRequests) + } + req := run.listRequests[0] + if req.GetProjectId() != "project-aaaaaaaaaaaaaaaa" || req.GetSchedulerRunId() != "srun-1" || req.GetLabels()["env"] != "prod" || req.GetLabels()["team"] != "sec" { + t.Fatalf("request = %+v", req) + } + from := req.GetStartedFrom().AsTime() + if from.Nanosecond()%int(time.Millisecond) != 0 { + t.Fatalf("started_from has sub-millisecond precision: %s", from) + } + if lower, upper := before.Add(-24*time.Hour-time.Second), time.Now().Add(-24*time.Hour); from.Before(lower) || from.After(upper) { + t.Fatalf("started_from = %s, want about 24h ago", from) + } +} + +func TestRunListRejectsInvalidDiagnosticFilters(t *testing.T) { + for _, args := range [][]string{ + {"run", "ls", "--started-from", "yesterday"}, + {"run", "ls", "--started-to", "-1h"}, + {"run", "ls", "--label", "env"}, + {"run", "ls", "--all-projects", "--project", "fixture"}, + } { + cmd := NewCommand() + cmd.SetArgs(args) + state := stateFromCommand(cmd) + state.options.URL = "https://example.com" + state.options.Token = "token" + err := cmd.Execute() + coder, ok := err.(interface{ ExitCode() int }) + if !ok || coder.ExitCode() != exitUsage { + t.Fatalf("%v error = %v", args, err) + } + } +} + +func TestRunListAllProjectsSkipsProjectResolution(t *testing.T) { + project := &projectStub{project: fixtureProject()} + run := &runStub{runs: []*agentcomposev2.RunSummary{ + {RunId: "run-aaaaaaaaaaaaaaaa", ProjectId: "project-a", ProjectName: "alpha", AgentName: "triage", Status: agentcomposev2.RunStatus_RUN_STATUS_FAILED}, + {RunId: "run-bbbbbbbbbbbbbbbb", ProjectId: "project-b", ProjectName: "beta", AgentName: "review", Status: agentcomposev2.RunStatus_RUN_STATUS_FAILED}, + }} + server, _ := newTestServer(t, project, run) + out, _, err := executeCommand(t, server.URL, false, "run", "ls", "--all-projects", "--status", "failed") + if err != nil { + t.Fatal(err) + } + if len(project.requests) != 0 { + t.Fatalf("ListProjects was called %d times", len(project.requests)) + } + if len(run.listRequests) != 1 || run.listRequests[0].GetProjectId() != "" || run.listRequests[0].GetStatus() != agentcomposev2.RunStatus_RUN_STATUS_FAILED { + t.Fatalf("requests = %+v", run.listRequests) + } + if !strings.Contains(out, "PROJECT") || !strings.Contains(out, "alpha") || !strings.Contains(out, "beta") { + t.Fatalf("output = %s", out) + } + + out, _, err = executeCommand(t, server.URL, false, "--json", "run", "ls", "--all-projects") + if err != nil { + t.Fatal(err) + } + var payload map[string]any + if err := json.Unmarshal([]byte(out), &payload); err != nil { + t.Fatal(err) + } + if _, present := payload["project"]; present { + t.Fatalf("all-projects JSON has a project field: %s", out) + } + if runs, _ := payload["runs"].([]any); len(runs) != 2 { + t.Fatalf("output = %s", out) + } +} + +func TestInspectRunIncludesDiagnosticFields(t *testing.T) { + project := &projectStub{project: fixtureProject()} + summary := &agentcomposev2.RunSummary{RunId: "run-aaaaaaaaaaaaaaaa", ProjectId: "project-aaaaaaaaaaaaaaaa", AgentName: "agent", Status: agentcomposev2.RunStatus_RUN_STATUS_FAILED, Error: "agent exited", SchedulerId: "scheduler-1", TriggerId: "trigger-1", SchedulerRunId: "srun-1"} + run := &runStub{ + runs: []*agentcomposev2.RunSummary{summary}, + details: map[string]*agentcomposev2.RunDetail{"run-aaaaaaaaaaaaaaaa": { + Summary: summary, + ErrorStack: "panic: boom\ngoroutine 1", + CleanupError: "sandbox stop timed out", + Warnings: []string{"image pull slow"}, + Labels: map[string]string{"env": "prod"}, + LogsPath: "/data/runs/run-aaaa/logs", + ArtifactsDir: "/data/runs/run-aaaa/artifacts", + }}, + } + server, _ := newTestServer(t, project, run, &resourceStub{}) + out, _, err := executeCommand(t, server.URL, false, "--json", "inspect", "run", "run-aaaaaaaaaaaaaaaa") + if err != nil { + t.Fatal(err) + } + var got runDTO + if err := json.Unmarshal([]byte(out), &got); err != nil { + t.Fatal(err) + } + if got.ErrorStack != "panic: boom\ngoroutine 1" || got.CleanupError != "sandbox stop timed out" || len(got.Warnings) != 1 || got.Labels["env"] != "prod" || got.LogsPath == "" || got.ArtifactsDir == "" { + t.Fatalf("detail fields missing: %s", out) + } + if got.SchedulerID != "scheduler-1" || got.TriggerID != "trigger-1" || got.SchedulerRunID != "srun-1" { + t.Fatalf("summary fields missing: %s", out) + } +} + +func TestLLMListMarksDefaultModelAndOmitsHeaders(t *testing.T) { + maxTokens := int32(4096) + llm := &llmStub{ + providers: []*agentcomposev2.LLMProvider{{ + Id: "provider-1", Name: "openai", Protocol: "responses", Enabled: true, ApiKeySet: true, + Models: []*agentcomposev2.LLMModelSpec{ + {Id: "gpt-a", Headers: map[string]string{"X-Secret": "header-secret"}, MaxOutputTokens: &maxTokens}, + {Id: "gpt-b"}, + }, + Capabilities: &agentcomposev2.LLMProviderCapabilities{Probes: []*agentcomposev2.LLMProtocolProbe{{Protocol: "responses", Outcome: "auth_failed", Detail: "401"}}}, + }}, + defaultModel: &agentcomposev2.LLMModelReference{ProviderId: "provider-1", ModelId: "gpt-b"}, + } + server, _ := newTestServer(t, nil, nil, llm) + out, _, err := executeCommand(t, server.URL, false, "llm", "ls") + if err != nil { + t.Fatal(err) + } + if !strings.Contains(out, "gpt-b (default)") || strings.Contains(out, "gpt-a (default)") || !strings.Contains(out, "responses=auth_failed") { + t.Fatalf("output = %s", out) + } + out, _, err = executeCommand(t, server.URL, false, "--json", "llm", "ls") + if err != nil { + t.Fatal(err) + } + if strings.Contains(out, "header-secret") || strings.Contains(out, "X-Secret") { + t.Fatalf("JSON leaks model headers: %s", out) + } + if !strings.Contains(out, `"default_model":{"provider_id":"provider-1","model_id":"gpt-b"}`) || !strings.Contains(out, `"api_key_set":true`) { + t.Fatalf("output = %s", out) + } +} + +func TestLLMListWithoutDefaultModel(t *testing.T) { + server, _ := newTestServer(t, nil, nil, &llmStub{}) + out, errOut, err := executeCommand(t, server.URL, false, "--json", "llm", "ls") + if err != nil { + t.Fatal(err) + } + if strings.Contains(out, "default_model") || errOut != "" { + t.Fatalf("stdout = %s, stderr = %s", out, errOut) + } + _, errOut, err = executeCommand(t, server.URL, false, "llm", "ls") + if err != nil { + t.Fatal(err) + } + if !strings.Contains(errOut, "No default model is configured") { + t.Fatalf("stderr = %s", errOut) + } +} diff --git a/products/agentcompose/integration_test.go b/products/agentcompose/integration_test.go index 4e97ab9..dcb13fe 100644 --- a/products/agentcompose/integration_test.go +++ b/products/agentcompose/integration_test.go @@ -37,15 +37,24 @@ type projectStub struct { func (s *projectStub) ListSchedulerRuns(_ context.Context, req *connect.Request[agentcomposev2.ListSchedulerRunsRequest]) (*connect.Response[agentcomposev2.ListSchedulerRunsResponse], error) { s.schedulerRunRequests = append(s.schedulerRunRequests, proto.Clone(req.Msg).(*agentcomposev2.ListSchedulerRunsRequest)) + runs := s.schedulerRuns + if agent := req.Msg.GetAgentName(); agent != "" { + runs = make([]*agentcomposev2.SchedulerRun, 0) + for _, run := range s.schedulerRuns { + if run.GetAgentName() == agent { + runs = append(runs, run) + } + } + } start := int(req.Msg.GetOffset()) - if start >= len(s.schedulerRuns) { - return connect.NewResponse(&agentcomposev2.ListSchedulerRunsResponse{Total: uint32(len(s.schedulerRuns))}), nil + if start >= len(runs) { + return connect.NewResponse(&agentcomposev2.ListSchedulerRunsResponse{Total: uint32(len(runs))}), nil } end := start + int(req.Msg.GetLimit()) - if end > len(s.schedulerRuns) { - end = len(s.schedulerRuns) + if end > len(runs) { + end = len(runs) } - return connect.NewResponse(&agentcomposev2.ListSchedulerRunsResponse{Runs: s.schedulerRuns[start:end], Total: uint32(len(s.schedulerRuns))}), nil + return connect.NewResponse(&agentcomposev2.ListSchedulerRunsResponse{Runs: runs[start:end], Total: uint32(len(runs))}), nil } func (s *projectStub) ListProjectSchedulerEvents(_ context.Context, req *connect.Request[agentcomposev2.ListProjectSchedulerEventsRequest]) (*connect.Response[agentcomposev2.ListProjectSchedulerEventsResponse], error) { @@ -104,6 +113,7 @@ type runStub struct { agentcomposev2connect.UnimplementedRunServiceHandler mu sync.Mutex runs []*agentcomposev2.RunSummary + details map[string]*agentcomposev2.RunDetail listRequests []*agentcomposev2.ListRunsRequest followCalls int chunks int @@ -228,16 +238,30 @@ type healthStub struct { healthv1connect.UnimplementedHealthServiceHandler } -type validatingHealthStub struct { - healthv1connect.UnimplementedHealthServiceHandler +// validatingProjectStub mirrors the daemon: control-plane services check the +// Bearer Token while HealthService stays exempt. +type validatingProjectStub struct { + agentcomposev2connect.UnimplementedProjectServiceHandler want string } -func (s validatingHealthStub) Status(_ context.Context, req *connect.Request[emptypb.Empty]) (*connect.Response[healthv1.HealthStatusResponse], error) { +func (s validatingProjectStub) ListProjects(_ context.Context, req *connect.Request[agentcomposev2.ListProjectsRequest]) (*connect.Response[agentcomposev2.ListProjectsResponse], error) { if req.Header().Get("Authorization") != "Bearer "+s.want { return nil, connect.NewError(connect.CodeUnauthenticated, fmt.Errorf("invalid Token")) } - return connect.NewResponse(&healthv1.HealthStatusResponse{BuildVersion: "test"}), nil + return connect.NewResponse(&agentcomposev2.ListProjectsResponse{}), nil +} + +func newAuthTestServer(t *testing.T, want string) *httptest.Server { + t.Helper() + mux := http.NewServeMux() + path, handler := healthv1connect.NewHealthServiceHandler(healthStub{}) + mux.Handle(path, handler) + path, handler = agentcomposev2connect.NewProjectServiceHandler(validatingProjectStub{want: want}) + mux.Handle(path, handler) + server := httptest.NewServer(mux) + t.Cleanup(server.Close) + return server } func (healthStub) Status(context.Context, *connect.Request[emptypb.Empty]) (*connect.Response[healthv1.HealthStatusResponse], error) { @@ -263,6 +287,9 @@ func newTestServer(t *testing.T, project *projectStub, run *runStub, extras ...a case *resourceStub: path, connectHandler := agentcomposev2connect.NewResourceServiceHandler(handler) mux.Handle(path, connectHandler) + case *llmStub: + path, connectHandler := agentcomposev2connect.NewLLMServiceHandler(handler) + mux.Handle(path, connectHandler) } } path, handler := healthv1connect.NewHealthServiceHandler(healthStub{}) @@ -716,11 +743,7 @@ func TestMissingTokenFailsBeforeNetwork(t *testing.T) { } func TestAuthLoginValidatesBeforeSavingAndHidesToken(t *testing.T) { - mux := http.NewServeMux() - path, handler := healthv1connect.NewHealthServiceHandler(validatingHealthStub{want: "new-secret-token"}) - mux.Handle(path, handler) - server := httptest.NewServer(mux) - defer server.Close() + server := newAuthTestServer(t, "new-secret-token") configPath := t.TempDir() + "/nested/config.yaml" cmd := NewCommand() var out, errOut bytes.Buffer @@ -745,11 +768,7 @@ func TestAuthLoginValidatesBeforeSavingAndHidesToken(t *testing.T) { } func TestAuthLoginInvalidTokenDoesNotWrite(t *testing.T) { - mux := http.NewServeMux() - path, handler := healthv1connect.NewHealthServiceHandler(validatingHealthStub{want: "valid"}) - mux.Handle(path, handler) - server := httptest.NewServer(mux) - defer server.Close() + server := newAuthTestServer(t, "valid") configPath := t.TempDir() + "/config.yaml" cmd := NewCommand() cmd.SetIn(strings.NewReader("invalid\n")) @@ -797,7 +816,7 @@ func TestAuthLogoutOnlyClearsStoredTokenAndWarnsForEnvironment(t *testing.T) { func TestAuthStatusReportsEnvironmentTokenSource(t *testing.T) { t.Setenv("AGENT_COMPOSE_API_TOKEN", "test-token") - server, _ := newTestServer(t, nil, nil) + server, _ := newTestServer(t, &projectStub{project: fixtureProject()}, nil) out, _, err := executeCommand(t, server.URL, false, "--json", "auth", "status") if err != nil { t.Fatal(err) @@ -806,3 +825,33 @@ func TestAuthStatusReportsEnvironmentTokenSource(t *testing.T) { t.Fatalf("output = %s", out) } } + +func TestAuthStatusRejectsInvalidTokenDespiteHealthExemption(t *testing.T) { + server := newAuthTestServer(t, "valid") + cmd := NewCommand() + cmd.SetArgs([]string{"--url", server.URL, "auth", "status"}) + var node yaml.Node + if err := node.Encode(productConfig{APIToken: "invalid"}); err != nil { + t.Fatal(err) + } + ApplyRuntimeConfig(cmd, config.Raw{productName: node}, t.TempDir()+"/config.yaml", false) + err := cmd.Execute() + coder, ok := err.(interface{ ExitCode() int }) + if !ok || coder.ExitCode() != exitAuth { + t.Fatalf("error = %v", err) + } +} + +func TestProjectShorthandFlag(t *testing.T) { + project := &projectStub{project: fixtureProject()} + run := &runStub{} + server, _ := newTestServer(t, project, run, &resourceStub{}) + // The configured default project exists, so only an honored -p can make this fail. + _, _, err := executeCommand(t, server.URL, false, "-p", "missing-project", "run", "ls") + if !isNotFoundError(err) { + t.Fatalf("error = %v", err) + } + if len(run.listRequests) != 0 { + t.Fatalf("requests = %+v", run.listRequests) + } +} diff --git a/products/agentcompose/llm.go b/products/agentcompose/llm.go new file mode 100644 index 0000000..46907c0 --- /dev/null +++ b/products/agentcompose/llm.go @@ -0,0 +1,121 @@ +package agentcompose + +import ( + "context" + "fmt" + "strings" + + "connectrpc.com/connect" + agentcomposev2 "github.com/chaitin/agent-compose/proto/agentcompose/v2" + "github.com/chaitin/agent-compose/proto/agentcompose/v2/agentcomposev2connect" + "github.com/spf13/cobra" +) + +type llmProbeDTO struct { + Protocol string `json:"protocol"` + Outcome string `json:"outcome"` + Detail string `json:"detail,omitempty"` +} + +// llmProviderDTO deliberately omits per-model headers: they are free-form and +// may carry credentials. +type llmProviderDTO struct { + ID string `json:"id"` + Name string `json:"name"` + BaseURL string `json:"base_url"` + Protocol string `json:"protocol"` + Auth string `json:"auth"` + Enabled bool `json:"enabled"` + APIKeySet bool `json:"api_key_set"` + Models []string `json:"models"` + Probes []llmProbeDTO `json:"probes,omitempty"` + ProbedAt string `json:"probed_at,omitempty"` + UpdatedAt string `json:"updated_at,omitempty"` +} + +type llmModelRefDTO struct { + ProviderID string `json:"provider_id"` + ModelID string `json:"model_id"` +} + +func llmProviderFromProto(provider *agentcomposev2.LLMProvider) llmProviderDTO { + models := make([]string, 0, len(provider.GetModels())) + for _, model := range provider.GetModels() { + models = append(models, model.GetId()) + } + probes := make([]llmProbeDTO, 0, len(provider.GetCapabilities().GetProbes())) + for _, probe := range provider.GetCapabilities().GetProbes() { + probes = append(probes, llmProbeDTO{Protocol: probe.GetProtocol(), Outcome: probe.GetOutcome(), Detail: probe.GetDetail()}) + } + return llmProviderDTO{ID: provider.GetId(), Name: provider.GetName(), BaseURL: provider.GetBaseUrl(), Protocol: provider.GetProtocol(), Auth: enumText(provider.GetAuth(), "LLM_PROVIDER_AUTH_"), Enabled: provider.GetEnabled(), APIKeySet: provider.GetApiKeySet(), Models: models, Probes: probes, ProbedAt: timestampText(provider.GetCapabilities().GetProbedAt()), UpdatedAt: timestampText(provider.GetUpdatedAt())} +} + +func newLLMCommand(state *commandState) *cobra.Command { + cmd := &cobra.Command{Use: "llm", Short: "Query LLM provider configuration", Args: noArgs(state), RunE: func(cmd *cobra.Command, _ []string) error { return cmd.Help() }} + list := &cobra.Command{Use: "ls", Short: "List LLM providers, their models, and the default model", Args: noArgs(state), RunE: func(cmd *cobra.Command, _ []string) error { + ctx, cancel := requestContext(cmd, state) + defer cancel() + client := state.clients().llm + providers, err := listLLMProviders(ctx, client) + if err != nil { + return mapConnectError(err, state.options.URL, state.options.JSON) + } + resp, err := client.GetDefaultModel(ctx, connect.NewRequest(&agentcomposev2.GetDefaultModelRequest{})) + if err != nil { + return mapConnectError(err, state.options.URL, state.options.JSON) + } + var defaultModel *llmModelRefDTO + if model := resp.Msg.GetModel(); model != nil { + defaultModel = &llmModelRefDTO{ProviderID: model.GetProviderId(), ModelID: model.GetModelId()} + } + output := make([]llmProviderDTO, 0, len(providers)) + for _, provider := range providers { + output = append(output, llmProviderFromProto(provider)) + } + if state.options.JSON { + return writeJSON(cmd.OutOrStdout(), struct { + Providers []llmProviderDTO `json:"providers"` + DefaultModel *llmModelRefDTO `json:"default_model,omitempty"` + }{output, defaultModel}) + } + table := newTable(cmd.OutOrStdout(), "NAME\tPROTOCOL\tENABLED\tAPI KEY\tMODELS\tPROBES") + for _, provider := range output { + models := make([]string, 0, len(provider.Models)) + for _, model := range provider.Models { + if defaultModel != nil && defaultModel.ProviderID == provider.ID && defaultModel.ModelID == model { + model += " (default)" + } + models = append(models, model) + } + probes := make([]string, 0, len(provider.Probes)) + for _, probe := range provider.Probes { + probes = append(probes, probe.Protocol+"="+probe.Outcome) + } + fmt.Fprintf(table, "%s\t%s\t%t\t%t\t%s\t%s\n", provider.Name, provider.Protocol, provider.Enabled, provider.APIKeySet, strings.Join(models, ","), strings.Join(probes, ",")) + } + if err := table.Flush(); err != nil { + return err + } + if defaultModel == nil { + fmt.Fprintln(cmd.ErrOrStderr(), "No default model is configured") + } + return nil + }} + cmd.AddCommand(list) + return cmd +} + +func listLLMProviders(ctx context.Context, client agentcomposev2connect.LLMServiceClient) ([]*agentcomposev2.LLMProvider, error) { + providers := make([]*agentcomposev2.LLMProvider, 0) + for { + resp, err := client.ListProviders(ctx, connect.NewRequest(&agentcomposev2.ListProvidersRequest{Offset: uint32(len(providers)), Limit: 100})) + if err != nil { + return nil, err + } + page := resp.Msg.GetProviders() + providers = append(providers, page...) + if len(page) == 0 || uint32(len(providers)) >= resp.Msg.GetTotal() { + return providers, nil + } + } +} diff --git a/products/agentcompose/output.go b/products/agentcompose/output.go index 7620f68..d3aa5ec 100644 --- a/products/agentcompose/output.go +++ b/products/agentcompose/output.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "io" + "strconv" "strings" "text/tabwriter" "time" @@ -20,19 +21,59 @@ func timestampText(value *timestamppb.Timestamp) string { return value.AsTime().Format("2006-01-02T15:04:05.999999999Z07:00") } -func parseOptionalTimestamp(raw string) (*timestamppb.Timestamp, error) { +// parseTimeBound accepts an RFC3339 timestamp or a duration relative to now +// ("30m", "24h", "7d"). The result is truncated to milliseconds because the +// daemon rejects sub-millisecond time filters. +func parseTimeBound(raw string, now time.Time) (*timestamppb.Timestamp, error) { raw = strings.TrimSpace(raw) if raw == "" { return nil, nil } parsed, err := time.Parse(time.RFC3339Nano, raw) if err != nil { - parsed, err = time.Parse(time.RFC3339, raw) + ago, durationErr := parseRelativeDuration(raw) + if durationErr != nil { + return nil, fmt.Errorf("%q is neither an RFC3339 timestamp nor a duration such as 30m, 24h, or 7d", raw) + } + parsed = now.Add(-ago) } - if err != nil { - return nil, err + return timestamppb.New(parsed.Truncate(time.Millisecond)), nil +} + +func parseRelativeDuration(raw string) (time.Duration, error) { + var ago time.Duration + if days, ok := strings.CutSuffix(raw, "d"); ok { + count, err := strconv.ParseUint(days, 10, 16) + if err != nil { + return 0, err + } + ago = time.Duration(count) * 24 * time.Hour + } else { + var err error + if ago, err = time.ParseDuration(raw); err != nil { + return 0, err + } + } + if ago <= 0 { + return 0, fmt.Errorf("duration must be positive") + } + return ago, nil +} + +func parseLabels(raw []string) (map[string]string, error) { + if len(raw) == 0 { + return nil, nil + } + labels := make(map[string]string, len(raw)) + for _, item := range raw { + key, value, ok := strings.Cut(item, "=") + key = strings.TrimSpace(key) + if !ok || key == "" { + return nil, fmt.Errorf("invalid --label %q; expected key=value", item) + } + labels[key] = value } - return timestamppb.New(parsed), nil + return labels, nil } type runEventDTO struct { diff --git a/products/agentcompose/ps_scan_test.go b/products/agentcompose/ps_scan_test.go new file mode 100644 index 0000000..0f53218 --- /dev/null +++ b/products/agentcompose/ps_scan_test.go @@ -0,0 +1,185 @@ +package agentcompose + +import ( + "encoding/json" + "fmt" + "testing" + "time" + + agentcomposev2 "github.com/chaitin/agent-compose/proto/agentcompose/v2" + "google.golang.org/protobuf/types/known/timestamppb" +) + +// hourlySchedulerRuns returns count Scheduler Runs one hour apart, newest first, +// matching the daemon's started_at DESC order. +func hourlySchedulerRuns(newest time.Time, count int) []*agentcomposev2.SchedulerRun { + runs := make([]*agentcomposev2.SchedulerRun, count) + for i := range runs { + runs[i] = &agentcomposev2.SchedulerRun{ + RunId: fmt.Sprintf("scheduler-run-%04d", i), + StartedAt: timestamppb.New(newest.Add(-time.Duration(i) * time.Hour)), + SandboxIds: []string{fmt.Sprintf("other-sandbox-%04d", i)}, + } + } + return runs +} + +func psSandbox(id string, createdAt *timestamppb.Timestamp) *agentcomposev2.Sandbox { + return &agentcomposev2.Sandbox{SandboxId: id, ProjectId: "project-aaaaaaaaaaaaaaaa", Status: agentcomposev2.SandboxStatus_SANDBOX_STATUS_RUNNING, CreatedAt: createdAt} +} + +func psRunIDs(t *testing.T, out string) map[string]string { + t.Helper() + var payload struct { + Sandboxes []sandboxDTO `json:"sandboxes"` + } + if err := json.Unmarshal([]byte(out), &payload); err != nil { + t.Fatalf("output = %s: %v", out, err) + } + runIDs := make(map[string]string, len(payload.Sandboxes)) + for _, sandbox := range payload.Sandboxes { + runIDs[sandbox.SandboxID] = sandbox.RunID + } + return runIDs +} + +func TestPSStopsSchedulerRunScanOnceEverySandboxIsMatched(t *testing.T) { + newest := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + runs := hourlySchedulerRuns(newest, 1000) + runs[150].SandboxIds = []string{"target"} + project := &projectStub{project: fixtureProject(), schedulerRuns: runs} + // No creation time means no cutoff, so only the matched-everything stop can + // end the scan on the second page. + sandbox := &sandboxStub{sandboxes: []*agentcomposev2.Sandbox{psSandbox("target", nil)}} + server, _ := newTestServer(t, project, &runStub{}, sandbox) + out, _, err := executeCommand(t, server.URL, false, "--json", "ps") + if err != nil { + t.Fatal(err) + } + if got := psRunIDs(t, out)["target"]; got != "scheduler-run-0150" { + t.Fatalf("target run_id = %q", got) + } + if len(project.schedulerRunRequests) != 2 { + t.Fatalf("ListSchedulerRuns calls = %d, want 2", len(project.schedulerRunRequests)) + } +} + +func TestPSStopsSchedulerRunScanAtCutoff(t *testing.T) { + newest := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + project := &projectStub{project: fixtureProject(), schedulerRuns: hourlySchedulerRuns(newest, 1000)} + // Created 10h before the newest run, so the cutoff is 34h back: runs 0..34 + // are in range and the scan ends with the page that crosses it. + sandbox := &sandboxStub{sandboxes: []*agentcomposev2.Sandbox{psSandbox("orphan", timestamppb.New(newest.Add(-10*time.Hour)))}} + server, _ := newTestServer(t, project, &runStub{}, sandbox) + out, _, err := executeCommand(t, server.URL, false, "--json", "ps") + if err != nil { + t.Fatal(err) + } + if got := psRunIDs(t, out)["orphan"]; got != "" { + t.Fatalf("orphan run_id = %q, want none", got) + } + if len(project.schedulerRunRequests) != 1 { + t.Fatalf("ListSchedulerRuns calls = %d, want 1 of 10 pages", len(project.schedulerRunRequests)) + } +} + +func TestPSFindsSchedulerRunStartedBeforeSandboxWithinMargin(t *testing.T) { + newest := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + runs := hourlySchedulerRuns(newest, 1000) + // The run started 20h before it created the Sandbox, inside the 24h margin, + // and sits on the second page. + runs[120].SandboxIds = []string{"late"} + project := &projectStub{project: fixtureProject(), schedulerRuns: runs} + sandbox := &sandboxStub{sandboxes: []*agentcomposev2.Sandbox{psSandbox("late", timestamppb.New(newest.Add(-100*time.Hour)))}} + server, _ := newTestServer(t, project, &runStub{}, sandbox) + out, _, err := executeCommand(t, server.URL, false, "--json", "ps") + if err != nil { + t.Fatal(err) + } + if got := psRunIDs(t, out)["late"]; got != "scheduler-run-0120" { + t.Fatalf("late run_id = %q", got) + } +} + +func TestPSScansAllSchedulerRunsWhenSandboxCreationTimeIsUnknown(t *testing.T) { + newest := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + project := &projectStub{project: fixtureProject(), schedulerRuns: hourlySchedulerRuns(newest, 1000)} + sandbox := &sandboxStub{sandboxes: []*agentcomposev2.Sandbox{ + psSandbox("recent", timestamppb.New(newest)), + psSandbox("unknown", nil), + }} + server, _ := newTestServer(t, project, &runStub{}, sandbox) + if _, _, err := executeCommand(t, server.URL, false, "--json", "ps"); err != nil { + t.Fatal(err) + } + if len(project.schedulerRunRequests) != 10 { + t.Fatalf("ListSchedulerRuns calls = %d, want all 10 pages", len(project.schedulerRunRequests)) + } +} + +func TestPSSkipsSchedulerRunsWhenRunsCoverEverySandbox(t *testing.T) { + newest := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + project := &projectStub{project: fixtureProject(), schedulerRuns: hourlySchedulerRuns(newest, 1000)} + run := &runStub{runs: []*agentcomposev2.RunSummary{{RunId: "run-1", ProjectId: "project-aaaaaaaaaaaaaaaa", SandboxId: "covered"}}} + sandbox := &sandboxStub{sandboxes: []*agentcomposev2.Sandbox{psSandbox("covered", timestamppb.New(newest))}} + server, _ := newTestServer(t, project, run, sandbox) + out, _, err := executeCommand(t, server.URL, false, "--json", "ps") + if err != nil { + t.Fatal(err) + } + if got := psRunIDs(t, out)["covered"]; got != "run-1" { + t.Fatalf("covered run_id = %q", got) + } + if len(project.schedulerRunRequests) != 0 { + t.Fatalf("ListSchedulerRuns calls = %d, want 0", len(project.schedulerRunRequests)) + } +} + +func TestPSLooksUpSchedulerRunsByAgentFirst(t *testing.T) { + newest := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + // A noisy agent owns the 1000 newest runs; the quiet agent's run that + // created the Sandbox is older than all of them. + runs := hourlySchedulerRuns(newest, 1000) + for _, run := range runs { + run.AgentName = "noisy" + } + created := newest.Add(-1000 * time.Hour) + runs = append(runs, &agentcomposev2.SchedulerRun{RunId: "quiet-run", AgentName: "quiet", StartedAt: timestamppb.New(created), SandboxIds: []string{"quiet-sandbox"}}) + project := &projectStub{project: fixtureProject(), schedulerRuns: runs} + target := psSandbox("quiet-sandbox", timestamppb.New(created)) + target.AgentName = "quiet" + server, _ := newTestServer(t, project, &runStub{}, &sandboxStub{sandboxes: []*agentcomposev2.Sandbox{target}}) + out, _, err := executeCommand(t, server.URL, false, "--json", "ps") + if err != nil { + t.Fatal(err) + } + if got := psRunIDs(t, out)["quiet-sandbox"]; got != "quiet-run" { + t.Fatalf("quiet-sandbox run_id = %q", got) + } + if len(project.schedulerRunRequests) != 1 || project.schedulerRunRequests[0].GetAgentName() != "quiet" { + t.Fatalf("requests = %+v, want one filtered by agent", project.schedulerRunRequests) + } +} + +func TestPSFallsBackToAllSchedulerRunsWhenAgentDiffers(t *testing.T) { + newest := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + runs := hourlySchedulerRuns(newest, 300) + for _, run := range runs { + run.AgentName = "other" + } + runs[5].SandboxIds = []string{"cross"} + project := &projectStub{project: fixtureProject(), schedulerRuns: runs} + sandbox := psSandbox("cross", timestamppb.New(newest.Add(-5*time.Hour))) + sandbox.AgentName = "mismatched" + server, _ := newTestServer(t, project, &runStub{}, &sandboxStub{sandboxes: []*agentcomposev2.Sandbox{sandbox}}) + out, _, err := executeCommand(t, server.URL, false, "--json", "ps") + if err != nil { + t.Fatal(err) + } + if got := psRunIDs(t, out)["cross"]; got != "scheduler-run-0005" { + t.Fatalf("cross run_id = %q", got) + } + if len(project.schedulerRunRequests) != 2 || project.schedulerRunRequests[0].GetAgentName() != "mismatched" || project.schedulerRunRequests[1].GetAgentName() != "" { + t.Fatalf("requests = %+v, want an agent scan then an unfiltered fallback", project.schedulerRunRequests) + } +} diff --git a/products/agentcompose/run.go b/products/agentcompose/run.go index 919b071..79c7ced 100644 --- a/products/agentcompose/run.go +++ b/products/agentcompose/run.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "strings" + "time" "connectrpc.com/connect" agentcomposev2 "github.com/chaitin/agent-compose/proto/agentcompose/v2" @@ -13,29 +14,38 @@ import ( ) type runDTO struct { - ID string `json:"id" yaml:"id"` - ShortID string `json:"short_id" yaml:"short_id"` - ProjectID string `json:"project_id" yaml:"project_id"` - ProjectName string `json:"project_name" yaml:"project_name"` - AgentName string `json:"agent_name" yaml:"agent_name"` - Source string `json:"source" yaml:"source"` - Status string `json:"status" yaml:"status"` - SandboxID string `json:"sandbox_id,omitempty" yaml:"sandbox_id,omitempty"` - SandboxShortID string `json:"sandbox_short_id,omitempty" yaml:"sandbox_short_id,omitempty"` - ExitCode int32 `json:"exit_code" yaml:"exit_code"` - Error string `json:"error,omitempty" yaml:"error,omitempty"` - StartedAt string `json:"started_at,omitempty" yaml:"started_at,omitempty"` - CompletedAt string `json:"completed_at,omitempty" yaml:"completed_at,omitempty"` - DurationMS int64 `json:"duration_ms,omitempty" yaml:"duration_ms,omitempty"` - Prompt string `json:"prompt,omitempty" yaml:"prompt,omitempty"` - Output string `json:"output,omitempty" yaml:"output,omitempty"` - ResultJSON string `json:"result_json,omitempty" yaml:"result_json,omitempty"` - Driver string `json:"driver,omitempty" yaml:"driver,omitempty"` - Image string `json:"image_ref,omitempty" yaml:"image_ref,omitempty"` + ID string `json:"id" yaml:"id"` + ShortID string `json:"short_id" yaml:"short_id"` + ProjectID string `json:"project_id" yaml:"project_id"` + ProjectName string `json:"project_name" yaml:"project_name"` + AgentName string `json:"agent_name" yaml:"agent_name"` + Source string `json:"source" yaml:"source"` + Status string `json:"status" yaml:"status"` + SandboxID string `json:"sandbox_id,omitempty" yaml:"sandbox_id,omitempty"` + SandboxShortID string `json:"sandbox_short_id,omitempty" yaml:"sandbox_short_id,omitempty"` + ExitCode int32 `json:"exit_code" yaml:"exit_code"` + Error string `json:"error,omitempty" yaml:"error,omitempty"` + StartedAt string `json:"started_at,omitempty" yaml:"started_at,omitempty"` + CompletedAt string `json:"completed_at,omitempty" yaml:"completed_at,omitempty"` + DurationMS int64 `json:"duration_ms,omitempty" yaml:"duration_ms,omitempty"` + Prompt string `json:"prompt,omitempty" yaml:"prompt,omitempty"` + Output string `json:"output,omitempty" yaml:"output,omitempty"` + ResultJSON string `json:"result_json,omitempty" yaml:"result_json,omitempty"` + Driver string `json:"driver,omitempty" yaml:"driver,omitempty"` + Image string `json:"image_ref,omitempty" yaml:"image_ref,omitempty"` + SchedulerID string `json:"scheduler_id,omitempty" yaml:"scheduler_id,omitempty"` + TriggerID string `json:"trigger_id,omitempty" yaml:"trigger_id,omitempty"` + SchedulerRunID string `json:"scheduler_run_id,omitempty" yaml:"scheduler_run_id,omitempty"` + Warnings []string `json:"warnings,omitempty" yaml:"warnings,omitempty"` + Labels map[string]string `json:"labels,omitempty" yaml:"labels,omitempty"` + ErrorStack string `json:"error_stack,omitempty" yaml:"error_stack,omitempty"` + CleanupError string `json:"cleanup_error,omitempty" yaml:"cleanup_error,omitempty"` + LogsPath string `json:"logs_path,omitempty" yaml:"logs_path,omitempty"` + ArtifactsDir string `json:"artifacts_dir,omitempty" yaml:"artifacts_dir,omitempty"` } func runFromSummary(run *agentcomposev2.RunSummary) runDTO { - return runDTO{ID: run.GetRunId(), ShortID: firstNonEmpty(run.GetRunShortId(), shortID(run.GetRunId())), ProjectID: run.GetProjectId(), ProjectName: run.GetProjectName(), AgentName: run.GetAgentName(), Source: enumText(run.GetSource(), "RUN_SOURCE_"), Status: enumText(run.GetStatus(), "RUN_STATUS_"), SandboxID: run.GetSandboxId(), SandboxShortID: firstNonEmpty(run.GetSandboxShortId(), shortID(run.GetSandboxId())), ExitCode: run.GetExitCode(), Error: run.GetError(), StartedAt: timestampText(run.GetStartedAt()), CompletedAt: timestampText(run.GetCompletedAt()), DurationMS: run.GetDurationMs()} + return runDTO{ID: run.GetRunId(), ShortID: firstNonEmpty(run.GetRunShortId(), shortID(run.GetRunId())), ProjectID: run.GetProjectId(), ProjectName: run.GetProjectName(), AgentName: run.GetAgentName(), Source: enumText(run.GetSource(), "RUN_SOURCE_"), Status: enumText(run.GetStatus(), "RUN_STATUS_"), SandboxID: run.GetSandboxId(), SandboxShortID: firstNonEmpty(run.GetSandboxShortId(), shortID(run.GetSandboxId())), ExitCode: run.GetExitCode(), Error: run.GetError(), StartedAt: timestampText(run.GetStartedAt()), CompletedAt: timestampText(run.GetCompletedAt()), DurationMS: run.GetDurationMs(), SchedulerID: run.GetSchedulerId(), TriggerID: run.GetTriggerId(), SchedulerRunID: run.GetSchedulerRunId(), Warnings: run.GetWarnings()} } func runFromDetail(detail *agentcomposev2.RunDetail) runDTO { result := runFromSummary(detail.GetSummary()) @@ -44,6 +54,14 @@ func runFromDetail(detail *agentcomposev2.RunDetail) runDTO { result.ResultJSON = detail.GetResultJson() result.Driver = detail.GetDriver() result.Image = detail.GetImageRef() + if len(detail.GetWarnings()) > 0 { + result.Warnings = detail.GetWarnings() + } + result.Labels = detail.GetLabels() + result.ErrorStack = detail.GetErrorStack() + result.CleanupError = detail.GetCleanupError() + result.LogsPath = detail.GetLogsPath() + result.ArtifactsDir = detail.GetArtifactsDir() return result } @@ -223,7 +241,9 @@ func resolveAgent(ctx context.Context, project *agentcomposev2.Project, ref stri type runListOptions struct { offsetOptions - Agent, Scheduler, Status, Source, StartedFrom, StartedTo, Sandbox string + Agent, Scheduler, SchedulerRun, Status, Source, StartedFrom, StartedTo, Sandbox string + Labels []string + AllProjects bool } func newRunListCommand(state *commandState) *cobra.Command { @@ -238,19 +258,31 @@ func newRunListCommand(state *commandState) *cobra.Command { if _, err := parseRunSource(options.Source); err != nil { return usageError(err.Error(), state.options.JSON) } - if _, err := parseOptionalTimestamp(options.StartedFrom); err != nil { - return usageError("invalid --started-from timestamp", state.options.JSON) + if _, err := parseTimeBound(options.StartedFrom, time.Now()); err != nil { + return usageError("invalid --started-from: "+err.Error(), state.options.JSON) } - if _, err := parseOptionalTimestamp(options.StartedTo); err != nil { - return usageError("invalid --started-to timestamp", state.options.JSON) + if _, err := parseTimeBound(options.StartedTo, time.Now()); err != nil { + return usageError("invalid --started-to: "+err.Error(), state.options.JSON) + } + if _, err := parseLabels(options.Labels); err != nil { + return usageError(err.Error(), state.options.JSON) + } + if options.AllProjects && cmd.Flags().Changed("project") { + return usageError("--all-projects and --project are mutually exclusive", state.options.JSON) } ctx, cancel := requestContext(cmd, state) defer cancel() - project, err := resolveProject(ctx, state, state.clients().project) - if err != nil { - return err + var project *projectDTO + projectID := "" + if !options.AllProjects { + resolved, err := resolveProject(ctx, state, state.clients().project) + if err != nil { + return err + } + summary := projectFromProto(resolved.GetSummary()) + project, projectID = &summary, summary.ID } - runs, hasMore, next, err := listRuns(ctx, state.clients().run, project.GetSummary().GetProjectId(), options) + runs, hasMore, next, err := listRuns(ctx, state.clients().run, projectID, options) if err != nil { return mapConnectError(err, state.options.URL, state.options.JSON) } @@ -260,15 +292,24 @@ func newRunListCommand(state *commandState) *cobra.Command { } if state.options.JSON { return writeJSON(cmd.OutOrStdout(), struct { - Project projectDTO `json:"project"` - Runs []runDTO `json:"runs"` - HasMore bool `json:"has_more"` - Next uint32 `json:"next_offset,omitempty"` - }{projectFromProto(project.GetSummary()), output, hasMore, next}) + Project *projectDTO `json:"project,omitempty"` + Runs []runDTO `json:"runs"` + HasMore bool `json:"has_more"` + Next uint32 `json:"next_offset,omitempty"` + }{project, output, hasMore, next}) + } + header := "RUN\tAGENT\tSTATUS\tSOURCE\tSANDBOX\tSTARTED" + if options.AllProjects { + header = "RUN\tPROJECT\tAGENT\tSTATUS\tSOURCE\tSANDBOX\tSTARTED" } - table := newTable(cmd.OutOrStdout(), "RUN\tAGENT\tSTATUS\tSOURCE\tSANDBOX\tSTARTED") + table := newTable(cmd.OutOrStdout(), header) for _, run := range output { - fmt.Fprintf(table, "%s\t%s\t%s\t%s\t%s\t%s\n", run.ShortID, run.AgentName, run.Status, run.Source, run.SandboxShortID, run.StartedAt) + if options.AllProjects { + fmt.Fprintf(table, "%s\t%s\t", run.ShortID, run.ProjectName) + } else { + fmt.Fprintf(table, "%s\t", run.ShortID) + } + fmt.Fprintf(table, "%s\t%s\t%s\t%s\t%s\n", run.AgentName, run.Status, run.Source, run.SandboxShortID, run.StartedAt) } if err := table.Flush(); err != nil { return err @@ -283,9 +324,12 @@ func newRunListCommand(state *commandState) *cobra.Command { cmd.Flags().StringVar(&options.Scheduler, "scheduler", "", "Filter by Scheduler ID") cmd.Flags().StringVar(&options.Status, "status", "", "Filter by status") cmd.Flags().StringVar(&options.Source, "source", "", "Filter by source") - cmd.Flags().StringVar(&options.StartedFrom, "started-from", "", "Filter by start time lower bound") - cmd.Flags().StringVar(&options.StartedTo, "started-to", "", "Filter by start time upper bound") + cmd.Flags().StringVar(&options.SchedulerRun, "scheduler-run", "", "Filter by Scheduler Run ID") + cmd.Flags().StringVar(&options.StartedFrom, "started-from", "", "Filter by start time lower bound (RFC3339, or relative such as 30m, 24h, 7d)") + cmd.Flags().StringVar(&options.StartedTo, "started-to", "", "Filter by start time upper bound (RFC3339, or relative such as 30m, 24h, 7d)") cmd.Flags().StringVar(&options.Sandbox, "sandbox", "", "Filter by Sandbox ID") + cmd.Flags().StringArrayVar(&options.Labels, "label", nil, "Filter by label key=value; repeat to require several") + cmd.Flags().BoolVar(&options.AllProjects, "all-projects", false, "List Runs across every project the token can access") return cmd } @@ -298,11 +342,16 @@ func listRuns(ctx context.Context, client agentcomposev2connect.RunServiceClient if err != nil { return nil, false, 0, err } - startedFrom, err := parseOptionalTimestamp(options.StartedFrom) + now := time.Now() + startedFrom, err := parseTimeBound(options.StartedFrom, now) + if err != nil { + return nil, false, 0, err + } + startedTo, err := parseTimeBound(options.StartedTo, now) if err != nil { return nil, false, 0, err } - startedTo, err := parseOptionalTimestamp(options.StartedTo) + labels, err := parseLabels(options.Labels) if err != nil { return nil, false, 0, err } @@ -310,7 +359,7 @@ func listRuns(ctx context.Context, client agentcomposev2connect.RunServiceClient runs := make([]*agentcomposev2.RunSummary, 0) for { pageSize := offsetPageSize(options.offsetOptions, len(runs)) - resp, err := client.ListRuns(ctx, connect.NewRequest(&agentcomposev2.ListRunsRequest{ProjectId: projectID, AgentName: options.Agent, SchedulerId: options.Scheduler, Status: status, Source: source, StartedFrom: startedFrom, StartedTo: startedTo, SandboxId: options.Sandbox, Offset: offset, Limit: pageSize})) + resp, err := client.ListRuns(ctx, connect.NewRequest(&agentcomposev2.ListRunsRequest{ProjectId: projectID, AgentName: options.Agent, SchedulerId: options.Scheduler, SchedulerRunId: options.SchedulerRun, Labels: labels, Status: status, Source: source, StartedFrom: startedFrom, StartedTo: startedTo, SandboxId: options.Sandbox, Offset: offset, Limit: pageSize})) if err != nil { return nil, false, 0, err } diff --git a/products/agentcompose/sandbox.go b/products/agentcompose/sandbox.go index 0d6ba42..080c4e6 100644 --- a/products/agentcompose/sandbox.go +++ b/products/agentcompose/sandbox.go @@ -102,24 +102,19 @@ func executeSandboxList(cmd *cobra.Command, state *commandState, options sandbox } } } - needsSchedulerRuns := false + wanted := make([]*agentcomposev2.Sandbox, 0) for _, sandbox := range sandboxes { if _, ok := runBySandbox[sandbox.GetSandboxId()]; !ok { - needsSchedulerRuns = true - break + wanted = append(wanted, sandbox) } } - if needsSchedulerRuns { - schedulerRuns, _, _, schedulerErr := listSchedulerRuns(ctx, state.clients().project, project.GetSummary().GetProjectId(), "", schedulerRunsOptions{offsetOptions: offsetOptions{AllPages: true, Limit: 50}}) + if len(wanted) > 0 { + schedulerRuns, schedulerErr := associateSchedulerRuns(ctx, state.clients().project, project.GetSummary().GetProjectId(), wanted) if schedulerErr != nil { return mapConnectError(schedulerErr, state.options.URL, state.options.JSON) } - for _, run := range schedulerRuns { - for _, sandboxID := range run.GetSandboxIds() { - if _, ok := runBySandbox[sandboxID]; !ok { - runBySandbox[sandboxID] = &agentcomposev2.RunSummary{RunId: run.GetRunId(), RunShortId: shortID(run.GetRunId()), SandboxId: sandboxID} - } - } + for sandboxID, run := range schedulerRuns { + runBySandbox[sandboxID] = &agentcomposev2.RunSummary{RunId: run.GetRunId(), RunShortId: shortID(run.GetRunId()), SandboxId: sandboxID} } } output := make([]sandboxDTO, 0, len(sandboxes)) @@ -202,6 +197,90 @@ func parseSandboxStatus(raw string) (agentcomposev2.SandboxStatus, error) { } } +// schedulerRunScanMargin is how far before the oldest wanted Sandbox's +// creation the scan still looks for its Scheduler Run. A Scheduler Run starts +// before the Sandboxes it creates, but a long run may create one hours later; +// 24h covers common SCHEDULER_RUN_TIMEOUT values. +const schedulerRunScanMargin = 24 * time.Hour + +// associateSchedulerRuns finds the Scheduler Run that created each wanted +// Sandbox. The API cannot filter Scheduler Runs by Sandbox, and a busy project +// keeps tens of thousands of them, so reading the whole history is too slow. +// A scheduler creates Sandboxes for its own agent, so each agent's Sandboxes +// are first looked up among that agent's runs only; any still unmatched are +// then looked up across all runs, so no association is lost if that ever +// does not hold. +func associateSchedulerRuns(ctx context.Context, client agentcomposev2connect.ProjectServiceClient, projectID string, wanted []*agentcomposev2.Sandbox) (map[string]*agentcomposev2.SchedulerRun, error) { + byAgent := map[string][]*agentcomposev2.Sandbox{} + for _, sandbox := range wanted { + if agent := sandbox.GetAgentName(); agent != "" { + byAgent[agent] = append(byAgent[agent], sandbox) + } + } + found := make(map[string]*agentcomposev2.SchedulerRun, len(wanted)) + for agent, sandboxes := range byAgent { + if err := scanSchedulerRuns(ctx, client, projectID, agent, sandboxes, found); err != nil { + return nil, err + } + } + unmatched := make([]*agentcomposev2.Sandbox, 0) + for _, sandbox := range wanted { + if found[sandbox.GetSandboxId()] == nil { + unmatched = append(unmatched, sandbox) + } + } + if len(unmatched) > 0 { + if err := scanSchedulerRuns(ctx, client, projectID, "", unmatched, found); err != nil { + return nil, err + } + } + return found, nil +} + +// scanSchedulerRuns records in found the newest Scheduler Run that lists each +// of the given Sandboxes. ListSchedulerRuns returns newest first, so the scan +// stops once every Sandbox is matched, or once a page reaches runs that +// started before the cutoff. A Sandbox without a creation time disables the +// cutoff rather than risk missing its run. +func scanSchedulerRuns(ctx context.Context, client agentcomposev2connect.ProjectServiceClient, projectID, agentName string, sandboxes []*agentcomposev2.Sandbox, found map[string]*agentcomposev2.SchedulerRun) error { + pending := make(map[string]struct{}, len(sandboxes)) + cutoff, bounded := time.Time{}, true + for _, sandbox := range sandboxes { + pending[sandbox.GetSandboxId()] = struct{}{} + createdAt := sandbox.GetCreatedAt() + if createdAt == nil { + bounded = false + continue + } + if earliest := createdAt.AsTime().Add(-schedulerRunScanMargin); cutoff.IsZero() || earliest.Before(cutoff) { + cutoff = earliest + } + } + for offset := uint32(0); len(pending) > 0; { + resp, err := client.ListSchedulerRuns(ctx, connect.NewRequest(&agentcomposev2.ListSchedulerRunsRequest{Project: projectRefID(projectID), AgentName: agentName, Offset: offset, Limit: 100})) + if err != nil { + return err + } + page := resp.Msg.GetRuns() + for _, run := range page { + for _, sandboxID := range run.GetSandboxIds() { + if _, ok := pending[sandboxID]; ok { + found[sandboxID] = run + delete(pending, sandboxID) + } + } + } + offset += uint32(len(page)) + if len(page) == 0 || offset >= resp.Msg.GetTotal() { + return nil + } + if oldest := page[len(page)-1].GetStartedAt(); bounded && oldest != nil && oldest.AsTime().Before(cutoff) { + return nil + } + } + return nil +} + func listSandboxes(ctx context.Context, client agentcomposev2connect.SandboxServiceClient, projectID string, status []agentcomposev2.SandboxStatus, options offsetOptions) ([]*agentcomposev2.Sandbox, bool, uint32, error) { offset := options.Offset sandboxes := make([]*agentcomposev2.Sandbox, 0)