diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..5e7b2aa --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,75 @@ +# Changelog + +## 2026-03-08 — Fix failed investigation counter reporting + +### Bug Fixes + +- **investigate: correctly count and report failed investigations** — In both sequential (`--parallel=1`) and batched (`--parallel=N`) modes, the `failed` counter was only incremented when the API call to start an investigation failed, not when the Temporal workflow itself completed with a `Failed` status. This caused the summary to report `0 failed` even when workflows had failed. +- **investigate: show warning icon on failure summary** — The summary line now displays `⚠` instead of `✓` when any investigations failed. + +--- + +## 2026-03-08 — Doctor and DynamoDB credential fixes, auto-sync repos.json + +### Bug Fixes + +- **doctor: fix Anthropic API key reported as NOT SET for Docker installs** — `checkProviderCredentials()` was fetching environment variables exclusively from the API endpoint (`/workers/worker-1/env?reveal=true`), which returns empty results for Docker-based installations. The fix applies the same Docker-aware pattern already used by `checkWorkerEnv()`: detecting the install type and reading from the local `worker.env` file via `bootstrap.ReadWorkerEnvFile()` for Docker installs, falling back to the API endpoint for source installs. + +- **fix DynamoDB not connecting for Docker installs** — The `TemporalComposeLocal()` docker-compose template was missing `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` in both the API and worker service environments. The local DynamoDB container requires dummy AWS credentials to accept connections. Added `AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID:-local}` and `AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY:-local}` to both services. + +### Features + +- **auto-sync repos from repos.json** — When `investigate --all` finds zero registered repos, it now automatically syncs from `repos.json` (checked in order: `~/.reposwarm/repos.json` local override → Docker container `/app/prompts/repos.json` → source install `/worker/prompts/repos.json`). Zero-touch setup for fresh Docker installs. + +--- + +## 2026-03-08 — Bulk terminate and workflow prune fix + +### Changes + +- `internal/commands/workflows.go` — Added `--all` flag to `workflows terminate`. Running `reposwarm workflows terminate --all --yes` now terminates all running workflows without interactive prompts. Previously each workflow had to be terminated individually. +- `internal/commands/workflows_prune.go` — Fixed `workflows prune` to actually delete workflows from Temporal history using `DELETE /workflows/` instead of just re-terminating them. Previously prune reported success but workflows remained in the list. + +### Usage + +``` +reposwarm workflows terminate --all --yes # Terminate all running workflows +reposwarm workflows prune --yes # Delete old completed/failed/terminated workflows +``` + +--- + +## 2026-03-08 — Sequential investigation mode (`--parallel` flag) + +### Problem + +Running `reposwarm investigate --all` on resource-constrained machines (16GB RAM) caused Temporal deadlock errors (`TMPRL1101`). The CLI fired `POST /investigate/single` for all repos simultaneously, starting 7+ concurrent Temporal workflows. The worker had no concurrency limits, so all repos cloned and analyzed in parallel, saturating I/O/CPU and blocking the Temporal event loop for >2 seconds. + +### Solution + +Single control point: the `--parallel` flag on `investigate --all` now controls both CLI dispatch behaviour and worker concurrency. No separate env vars to remember. + +``` +reposwarm investigate --all # Unchanged (fire-and-forget, parallel) +reposwarm investigate --all --parallel=1 # Sequential: one repo at a time +reposwarm investigate --all --parallel=2 # Batched: two repos at a time +``` + +When `--parallel` is set, the CLI: +1. Writes `REPOSWARM_PARALLEL=N` to `worker.env` (skips if already set) +2. Restarts the worker to apply the new concurrency limit (only if value changed) +3. Checks for running workflows before restarting to avoid killing in-flight work +4. Dispatches repos sequentially (or in batches of N), polling for completion between each + +### Changes + +**lac-reposwarm-cli (Go CLI):** + +- `internal/commands/investigate.go` — Rewrote `--all` loop with three modes: sequential (`--parallel=0` or `1`), batched (`--parallel=N`), and fire-and-forget (no flag, unchanged). Updated help text with examples. +- `internal/commands/investigate_helpers.go` — Added `waitForWorkflow()` (polls `GET /workflows/` until terminal state) and `ensureWorkerParallel()` (writes env var, restarts worker if changed, aborts if workflows are running). +- `internal/api/types.go` — Added `InvestigateResponse` struct to capture `workflowId` from `POST /investigate/single`. + +**lac-repo-swarm (Python worker):** + +- `src/investigate_worker.py` — Reads `REPOSWARM_PARALLEL` env var and passes `max_concurrent_activities` / `max_concurrent_workflow_task_polls` to the Temporal `Worker()` constructor. Default (unset/0) = unlimited, preserving cloud behaviour. +- `.env.example` — Documented `REPOSWARM_PARALLEL` with note that it is managed by the CLI. diff --git a/internal/api/types.go b/internal/api/types.go index 81b859e..7a4f6df 100644 --- a/internal/api/types.go +++ b/internal/api/types.go @@ -68,6 +68,12 @@ type InvestigateRequest struct { ChunkSize int `json:"chunk_size"` } +// InvestigateResponse from POST /investigate/single. +type InvestigateResponse struct { + WorkflowID string `json:"workflowId"` + Message string `json:"message"` +} + // InvestigateDailyRequest for POST /investigate/daily. type InvestigateDailyRequest struct { Model string `json:"model,omitempty"` diff --git a/internal/bootstrap/local.go b/internal/bootstrap/local.go index 5b0acc3..c79b066 100644 --- a/internal/bootstrap/local.go +++ b/internal/bootstrap/local.go @@ -674,6 +674,8 @@ services: - TEMPORAL_HTTP_URL=http://temporal-ui:8080 - TEMPORAL_NAMESPACE=default - AWS_REGION=${AWS_REGION:-us-east-1} + - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID:-local} + - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY:-local} - DYNAMODB_ENDPOINT=http://dynamodb-local:8000 - DYNAMODB_TABLE=${DYNAMODB_TABLE:-reposwarm-cache} - REPOSWARM_INSTALL_DIR=/data @@ -704,6 +706,8 @@ services: - TEMPORAL_NAMESPACE=default - TEMPORAL_TASK_QUEUE=investigate-task-queue - AWS_REGION=${AWS_REGION:-us-east-1} + - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID:-local} + - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY:-local} - DYNAMODB_ENDPOINT=http://localhost:8000 - DYNAMODB_TABLE=${DYNAMODB_TABLE:-reposwarm-cache} - LOCAL_TESTING=true diff --git a/internal/commands/doctor.go b/internal/commands/doctor.go index 2bb836d..0d07392 100644 --- a/internal/commands/doctor.go +++ b/internal/commands/doctor.go @@ -905,20 +905,36 @@ func checkProviderCredentials(priorChecks []checkResult) []checkResult { provider := cfg.EffectiveProvider() pc := cfg.ProviderConfig - // Fetch worker env for validation - var envResp struct { - Entries []struct { - Key string `json:"key"` - Value string `json:"value"` - Set bool `json:"set"` - } `json:"entries"` - } + // Fetch worker env for validation — use worker.env file for Docker installs, API for source installs + isDockerEnv := cfg.IsDockerInstall() || bootstrap.IsDockerInstall(cfg.EffectiveInstallDir()) currentEnv := make(map[string]string) - if err := client.Get(ctx(), "/workers/worker-1/env?reveal=true", &envResp); err == nil { - for _, e := range envResp.Entries { - if e.Set { - currentEnv[e.Key] = e.Value + if isDockerEnv { + workerEnv, _ := bootstrap.ReadWorkerEnvFile(cfg.EffectiveInstallDir()) + containerEnv, _ := bootstrap.DockerServiceEnv(cfg.EffectiveInstallDir(), "worker") + for k, v := range workerEnv { + if v != "" { + currentEnv[k] = v + } + } + for k, v := range containerEnv { + if v != "" { + currentEnv[k] = v + } + } + } else { + var envResp struct { + Entries []struct { + Key string `json:"key"` + Value string `json:"value"` + Set bool `json:"set"` + } `json:"entries"` + } + if err := client.Get(ctx(), "/workers/worker-1/env?reveal=true", &envResp); err == nil { + for _, e := range envResp.Entries { + if e.Set { + currentEnv[e.Key] = e.Value + } } } } diff --git a/internal/commands/investigate.go b/internal/commands/investigate.go index 4e55e41..1272163 100644 --- a/internal/commands/investigate.go +++ b/internal/commands/investigate.go @@ -3,6 +3,7 @@ package commands import ( "fmt" "strings" + "time" "github.com/reposwarm/reposwarm-cli/internal/api" "github.com/reposwarm/reposwarm-cli/internal/config" @@ -23,6 +24,8 @@ func newInvestigateCmd() *cobra.Command { Examples: reposwarm investigate is-odd # Single repo reposwarm investigate --all # All enabled repos + reposwarm investigate --all --parallel=1 # Sequential (one at a time) + reposwarm investigate --all --parallel=2 # Two repos at a time reposwarm investigate is-odd --model us.anthropic.claude-opus-4-6`, RunE: func(cmd *cobra.Command, args []string) error { client, err := getClient() @@ -154,7 +157,22 @@ Examples: } if len(enabledRepos) == 0 { - return fmt.Errorf("no enabled repos found\n Add repos first: reposwarm repos add --url --source GitHub") + // Try auto-sync from worker's repos.json + added, syncErr := syncReposFromJSON(client) + if syncErr != nil || added == 0 { + return fmt.Errorf("no enabled repos found\n Add repos first: reposwarm repos add --url --source GitHub") + } + + // Re-fetch after sync + enabledReposList = nil + if err := client.Get(ctx(), "/repos", &enabledReposList); err != nil { + return fmt.Errorf("fetching repos after sync: %w", err) + } + for _, r := range enabledReposList { + if r.Enabled { + enabledRepos = append(enabledRepos, r.Name) + } + } } // Pre-flight (check once, not per repo) @@ -209,51 +227,216 @@ Examples: recentlyInvestigated = checkRecentInvestigations(client, enabledRepos) } - // Start individual investigations for each enabled repo + // When --parallel is explicitly set, configure worker and use sequential/batched dispatch + parallelSet := cmd.Flags().Changed("parallel") + + if parallelSet { + // Dynamically configure worker concurrency + if err := ensureWorkerParallel(client, parallel); err != nil { + return err + } + } + + // Start investigations for each enabled repo started := 0 skipped := 0 - for _, repoName := range enabledRepos { - // Skip if recently investigated (unless --force) - if timeAgo, wasRecent := recentlyInvestigated[repoName]; wasRecent { - skipped++ + failed := 0 + completed := 0 + + if parallelSet && parallel <= 1 { + // --- Sequential mode: one repo at a time --- + total := len(enabledRepos) + for i, rn := range enabledRepos { + if timeAgo, wasRecent := recentlyInvestigated[rn]; wasRecent { + skipped++ + if !flagJSON { + output.F.Printf(" %s Skipping %s (investigated %s)\n", + output.Dim("⊘"), output.Bold(rn), timeAgo) + } + continue + } + if !flagJSON { - output.F.Printf(" %s Skipping %s (investigated %s)\n", - output.Dim("⊘"), output.Bold(repoName), timeAgo) + output.F.Printf("\n [%d/%d] Starting %s...\n", i+1, total, output.Bold(rn)) + } + + req := api.InvestigateRequest{ + RepoName: rn, + Model: model, + ChunkSize: chunkSize, + } + var resp api.InvestigateResponse + if err := client.Post(ctx(), "/investigate/single", req, &resp); err != nil { + failed++ + if !flagJSON { + output.F.Warning(fmt.Sprintf("Failed to start %s: %v", rn, err)) + } + continue + } + started++ + + // Wait for completion before starting the next repo + if resp.WorkflowID != "" { + if !flagJSON { + output.F.Printf(" %s Waiting for %s to finish...\n", output.Dim("⏳"), output.Bold(rn)) + } + startTime := time.Now() + finalStatus, err := waitForWorkflow(client, resp.WorkflowID, 5) + elapsed := time.Since(startTime).Round(time.Second) + if err != nil { + if !flagJSON { + output.F.Warning(fmt.Sprintf("Error watching %s: %v", rn, err)) + } + } else if strings.EqualFold(finalStatus, "Completed") { + completed++ + if !flagJSON { + output.Successf("%s completed (%s)", output.Bold(rn), elapsed) + } + } else { + failed++ + if !flagJSON { + output.F.Warning(fmt.Sprintf("%s finished with status: %s (%s)", rn, finalStatus, elapsed)) + } + } } - continue } + } else if parallelSet && parallel > 1 { + // --- Batched mode: N repos at a time --- + total := len(enabledRepos) + batchIdx := 0 + for batchIdx < total { + end := batchIdx + parallel + if end > total { + end = total + } + batch := enabledRepos[batchIdx:end] + + // Start all repos in this batch + type inflight struct { + name string + workflowID string + startTime time.Time + } + var active []inflight + + for _, rn := range batch { + if timeAgo, wasRecent := recentlyInvestigated[rn]; wasRecent { + skipped++ + if !flagJSON { + output.F.Printf(" %s Skipping %s (investigated %s)\n", + output.Dim("⊘"), output.Bold(rn), timeAgo) + } + continue + } - req := api.InvestigateRequest{ - RepoName: repoName, - Model: model, - ChunkSize: chunkSize, + if !flagJSON { + output.F.Printf(" [batch %d-%d/%d] Starting %s...\n", batchIdx+1, end, total, output.Bold(rn)) + } + + req := api.InvestigateRequest{ + RepoName: rn, + Model: model, + ChunkSize: chunkSize, + } + var resp api.InvestigateResponse + if err := client.Post(ctx(), "/investigate/single", req, &resp); err != nil { + failed++ + if !flagJSON { + output.F.Warning(fmt.Sprintf("Failed to start %s: %v", rn, err)) + } + continue + } + started++ + if resp.WorkflowID != "" { + active = append(active, inflight{name: rn, workflowID: resp.WorkflowID, startTime: time.Now()}) + } + } + + // Wait for all in this batch to complete + for _, a := range active { + if !flagJSON { + output.F.Printf(" %s Waiting for %s...\n", output.Dim("⏳"), output.Bold(a.name)) + } + finalStatus, err := waitForWorkflow(client, a.workflowID, 5) + elapsed := time.Since(a.startTime).Round(time.Second) + if err != nil { + if !flagJSON { + output.F.Warning(fmt.Sprintf("Error watching %s: %v", a.name, err)) + } + } else if strings.EqualFold(finalStatus, "Completed") { + completed++ + if !flagJSON { + output.Successf("%s completed (%s)", output.Bold(a.name), elapsed) + } + } else { + failed++ + if !flagJSON { + output.F.Warning(fmt.Sprintf("%s finished with status: %s (%s)", a.name, finalStatus, elapsed)) + } + } + } + + batchIdx = end } - var result any - if err := client.Post(ctx(), "/investigate/single", req, &result); err != nil { + } else { + // --- Default mode: fire-and-forget (existing behavior) --- + for _, repoName := range enabledRepos { + if timeAgo, wasRecent := recentlyInvestigated[repoName]; wasRecent { + skipped++ + if !flagJSON { + output.F.Printf(" %s Skipping %s (investigated %s)\n", + output.Dim("⊘"), output.Bold(repoName), timeAgo) + } + continue + } + + req := api.InvestigateRequest{ + RepoName: repoName, + Model: model, + ChunkSize: chunkSize, + } + var result any + if err := client.Post(ctx(), "/investigate/single", req, &result); err != nil { + if !flagJSON { + output.F.Warning(fmt.Sprintf("Failed to start %s: %v", repoName, err)) + } + continue + } + started++ if !flagJSON { - output.F.Warning(fmt.Sprintf("Failed to start %s: %v", repoName, err)) + output.Successf("Investigation started for %s", output.Bold(repoName)) } - continue - } - started++ - if !flagJSON { - output.Successf("Investigation started for %s", output.Bold(repoName)) } } if flagJSON { + mode := "parallel" + if parallelSet && parallel <= 1 { + mode = "sequential" + } else if parallelSet { + mode = fmt.Sprintf("batched(%d)", parallel) + } return output.JSON(map[string]any{ - "started": started, - "skipped": skipped, - "total": len(enabledRepos), - "repos": enabledRepos, + "started": started, + "completed": completed, + "skipped": skipped, + "failed": failed, + "total": len(enabledRepos), + "repos": enabledRepos, + "mode": mode, }) } if started == 0 && skipped == 0 { return fmt.Errorf("failed to start any investigations") } output.F.Println() - if skipped > 0 { + if parallelSet { + if failed > 0 { + output.F.Warning(fmt.Sprintf("Completed %d/%d investigations (%d skipped, %d failed)", completed, len(enabledRepos), skipped, failed)) + } else { + output.Successf("Completed %d/%d investigations (%d skipped, %d failed)", completed, len(enabledRepos), skipped, failed) + } + } else if skipped > 0 { output.Successf("Started %d/%d investigations (%d skipped, use --force to override)", started, len(enabledRepos), skipped) } else { output.Successf("Started %d/%d investigations", started, len(enabledRepos)) @@ -268,7 +451,7 @@ Examples: cmd.Flags().BoolVar(&all, "all", false, "Investigate all enabled repos") cmd.Flags().StringVar(&model, "model", "", "Model ID (default from config)") cmd.Flags().IntVar(&chunkSize, "chunk-size", 0, "Files per chunk (default from config)") - cmd.Flags().IntVar(¶llel, "parallel", 3, "Parallel limit (daily only)") + cmd.Flags().IntVar(¶llel, "parallel", 0, "Max parallel investigations (0=sequential, default=all-at-once when not set)") cmd.Flags().BoolVar(&force, "force", false, "Skip pre-flight checks and re-investigate recently completed repos") cmd.Flags().BoolVar(&replace, "replace", false, "Terminate existing workflow for this repo before starting") cmd.Flags().BoolVar(&dryRun, "dry-run", false, "Run pre-flight only, don't create workflow") diff --git a/internal/commands/investigate_helpers.go b/internal/commands/investigate_helpers.go index 8ad581e..9a3c3d7 100644 --- a/internal/commands/investigate_helpers.go +++ b/internal/commands/investigate_helpers.go @@ -1,10 +1,20 @@ package commands import ( + "encoding/json" "fmt" + "os" + osexec "os/exec" + "path/filepath" + "sort" + "strconv" + "strings" "time" "github.com/reposwarm/reposwarm-cli/internal/api" + "github.com/reposwarm/reposwarm-cli/internal/bootstrap" + "github.com/reposwarm/reposwarm-cli/internal/config" + "github.com/reposwarm/reposwarm-cli/internal/output" ) // checkRecentInvestigations returns a map of repo names that have completed @@ -74,6 +84,192 @@ func checkRecentInvestigations(client *api.Client, repoNames []string) map[strin return recentMap } +// waitForWorkflow polls GET /workflows/ until the workflow reaches a terminal state. +// Returns the final status string ("Completed", "Failed", etc.) and any error. +func waitForWorkflow(client *api.Client, workflowID string, interval int) (string, error) { + for { + var wf api.WorkflowExecution + if err := client.Get(ctx(), "/workflows/"+workflowID, &wf); err != nil { + // Transient error — retry + time.Sleep(time.Duration(interval) * time.Second) + continue + } + + lower := strings.ToLower(wf.Status) + if lower == "completed" || lower == "failed" || lower == "terminated" || lower == "timed_out" || lower == "cancelled" { + return wf.Status, nil + } + + time.Sleep(time.Duration(interval) * time.Second) + } +} + +// ensureWorkerParallel sets REPOSWARM_PARALLEL in worker.env and restarts the +// worker if the value has changed. This is the mechanism by which the CLI +// --parallel flag dynamically controls worker concurrency. +func ensureWorkerParallel(client *api.Client, parallel int) error { + cfg, err := config.Load() + if err != nil { + return nil // non-Docker install or no config — skip silently + } + + installDir := cfg.EffectiveInstallDir() + if !bootstrap.IsDockerInstall(installDir) { + return nil // not a Docker install — nothing to configure + } + + // Read current worker.env + envVars, err := bootstrap.ReadWorkerEnvFile(installDir) + if err != nil { + envVars = make(map[string]string) + } + + desired := strconv.Itoa(parallel) + current := envVars["REPOSWARM_PARALLEL"] + + if current == desired { + return nil // already set — no restart needed + } + + // Check for running workflows before restarting + var wfResult api.WorkflowsResponse + if err := client.Get(ctx(), "/workflows?pageSize=50", &wfResult); err == nil { + for _, w := range wfResult.Executions { + if strings.EqualFold(w.Status, "Running") { + return fmt.Errorf("cannot change parallelism: workflow %s is still running\n Wait for it to finish or terminate it first: reposwarm workflows terminate %s", w.WorkflowID, w.WorkflowID) + } + } + } + + // Write new value + envVars["REPOSWARM_PARALLEL"] = desired + envPath := filepath.Join(installDir, bootstrap.ComposeSubDir, "worker.env") + if err := writeWorkerEnvMap(envPath, envVars); err != nil { + return fmt.Errorf("writing worker.env: %w", err) + } + + // Also sync via API (best-effort) + body := map[string]string{"value": desired} + var resp any + _ = client.Put(ctx(), "/workers/worker-1/env/REPOSWARM_PARALLEL", body, &resp) + + // Restart worker to pick up the new value + output.F.Info(fmt.Sprintf("Setting REPOSWARM_PARALLEL=%s, restarting worker...", desired)) + composeDir := filepath.Join(installDir, bootstrap.ComposeSubDir) + stopCmd := osexec.Command("docker", "compose", "stop", "worker") + stopCmd.Dir = composeDir + stopCmd.CombinedOutput() // ignore error if already stopped + + restartCmd := osexec.Command("docker", "compose", "up", "-d", "--force-recreate", "worker") + restartCmd.Dir = composeDir + if out, err := restartCmd.CombinedOutput(); err != nil { + return fmt.Errorf("could not restart worker: %v (%s)", err, string(out)) + } + + // Wait for worker to be ready + time.Sleep(5 * time.Second) + output.Successf("Worker restarted with REPOSWARM_PARALLEL=%s", desired) + return nil +} + +// syncReposFromJSON reads repos.json and registers each repo with the API. +// Search order: +// 1. /repos.json (local override) +// 2. Docker: docker exec reposwarm-worker cat /app/prompts/repos.json +// Source: /worker/prompts/repos.json +// +// Returns the number of repos successfully added. +func syncReposFromJSON(client *api.Client) (int, error) { + cfg, err := config.Load() + if err != nil { + return 0, fmt.Errorf("loading config: %w", err) + } + + installDir := cfg.EffectiveInstallDir() + + // Read repos.json content — check local override first + var data []byte + localPath := filepath.Join(installDir, "repos.json") + if out, err := os.ReadFile(localPath); err == nil { + data = out + } else if bootstrap.IsDockerInstall(installDir) { + cmd := osexec.Command("docker", "exec", "reposwarm-worker", "cat", "/app/prompts/repos.json") + out, err := cmd.Output() + if err != nil { + return 0, fmt.Errorf("reading repos.json from worker container: %w", err) + } + data = out + } else { + path := filepath.Join(installDir, "worker", "prompts", "repos.json") + out, err := os.ReadFile(path) + if err != nil { + return 0, fmt.Errorf("reading %s: %w", path, err) + } + data = out + } + + // Parse repos.json — repositories values can be objects or comment strings + var raw struct { + Repositories map[string]json.RawMessage `json:"repositories"` + } + if err := json.Unmarshal(data, &raw); err != nil { + return 0, fmt.Errorf("parsing repos.json: %w", err) + } + + type repoEntry struct { + URL string `json:"url"` + Description string `json:"description"` + Type string `json:"type"` + } + + repos := make(map[string]repoEntry) + for name, rawVal := range raw.Repositories { + var entry repoEntry + if err := json.Unmarshal(rawVal, &entry); err != nil { + continue // skip comment strings like "_comment_react": "=== ..." + } + if entry.URL == "" { + continue // skip entries without a URL + } + repos[name] = entry + } + + if len(repos) == 0 { + return 0, nil + } + + // Sort names for deterministic output + var names []string + for name := range repos { + names = append(names, name) + } + sort.Strings(names) + + output.F.Info(fmt.Sprintf("No repos registered — found repos.json with %d repos, syncing...", len(names))) + + added := 0 + for _, name := range names { + repo := repos[name] + body := map[string]any{ + "name": name, + "url": repo.URL, + "source": detectSourceFromURL(repo.URL), + } + var result any + if err := client.Post(ctx(), "/repos", body, &result); err != nil { + output.F.Warning(fmt.Sprintf(" failed to add %s: %v", name, err)) + continue + } + output.F.Printf(" + %s\n", name) + added++ + } + + if added > 0 { + output.Successf("Synced %d repos from repos.json", added) + } + return added, nil +} + // formatTimeAgo formats a duration as a human-readable "time ago" string. func formatTimeAgo(d time.Duration) string { if d < time.Minute { diff --git a/internal/commands/workflows.go b/internal/commands/workflows.go index 80b877d..77d6f3a 100644 --- a/internal/commands/workflows.go +++ b/internal/commands/workflows.go @@ -177,14 +177,92 @@ func newWorkflowsStatusCmd() *cobra.Command { func newWorkflowsTerminateCmd() *cobra.Command { var yes bool + var all bool var reason string cmd := &cobra.Command{ - Use: "terminate ", + Use: "terminate [workflow-id]", Short: "Terminate a running workflow", - Args: friendlyExactArgs(1, "reposwarm workflows terminate \n\nExample:\n reposwarm workflows terminate wf-12345"), + Long: `Terminate one or all running workflows. + +Examples: + reposwarm workflows terminate wf-12345 + reposwarm workflows terminate --all --yes`, + Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error { - if !yes { + if !all && len(args) == 0 { + return fmt.Errorf("specify a workflow ID or use --all\n\nExamples:\n reposwarm workflows terminate \n reposwarm workflows terminate --all --yes") + } + + client, err := getClient() + if err != nil { + return err + } + + if all { + // Fetch running workflows + var wfResult api.WorkflowsResponse + if err := client.Get(ctx(), "/workflows?pageSize=100", &wfResult); err != nil { + return fmt.Errorf("fetching workflows: %w", err) + } + + var running []api.WorkflowExecution + for _, w := range wfResult.Executions { + if strings.EqualFold(w.Status, "Running") { + running = append(running, w) + } + } + + if len(running) == 0 { + if flagJSON { + return output.JSON(map[string]any{"terminated": 0}) + } + output.F.Info("No running workflows to terminate") + return nil + } + + if !yes && !flagJSON { + fmt.Printf(" Terminate %d running workflow(s)? [y/N] ", len(running)) + var confirm string + fmt.Scanln(&confirm) + if strings.ToLower(confirm) != "y" { + output.F.Info("Cancelled") + return nil + } + } + + terminated := 0 + failed := 0 + for _, w := range running { + body := map[string]string{"reason": reason} + var result any + if err := client.Post(ctx(), "/workflows/"+w.WorkflowID+"/terminate", body, &result); err != nil { + failed++ + if !flagJSON { + output.F.Warning(fmt.Sprintf("Failed to terminate %s: %v", w.WorkflowID, err)) + } + continue + } + terminated++ + if !flagJSON { + output.F.Success(fmt.Sprintf("Terminated %s", w.WorkflowID)) + } + } + + if flagJSON { + return output.JSON(map[string]any{ + "terminated": terminated, + "failed": failed, + "total": len(running), + }) + } + output.F.Println() + output.Successf("Terminated %d/%d running workflows", terminated, len(running)) + return nil + } + + // Single workflow terminate + if !yes && !flagJSON { fmt.Printf(" Terminate workflow %s? [y/N] ", args[0]) var confirm string fmt.Scanln(&confirm) @@ -194,11 +272,6 @@ func newWorkflowsTerminateCmd() *cobra.Command { } } - client, err := getClient() - if err != nil { - return err - } - body := map[string]string{"reason": reason} var result any if err := client.Post(ctx(), "/workflows/"+args[0]+"/terminate", body, &result); err != nil { @@ -214,6 +287,7 @@ func newWorkflowsTerminateCmd() *cobra.Command { } cmd.Flags().BoolVarP(&yes, "yes", "y", false, "Skip confirmation") + cmd.Flags().BoolVar(&all, "all", false, "Terminate all running workflows") cmd.Flags().StringVar(&reason, "reason", "Terminated via CLI", "Termination reason") return cmd } diff --git a/internal/commands/workflows_prune.go b/internal/commands/workflows_prune.go index e7e1253..3533953 100644 --- a/internal/commands/workflows_prune.go +++ b/internal/commands/workflows_prune.go @@ -139,13 +139,17 @@ Examples: } } - // Terminate/delete each + // Delete each workflow from Temporal history pruned := 0 for _, c := range candidates { - body := map[string]string{"reason": "Pruned via CLI"} - var termResult any - // Try terminate (for still-visible workflows) - _ = client.Post(ctx(), "/workflows/"+c.WorkflowID+"/terminate", body, &termResult) + var delResult any + if err := client.Delete(ctx(), "/workflows/"+c.WorkflowID, &delResult); err != nil { + // Fallback: try terminate then delete + body := map[string]string{"reason": "Pruned via CLI"} + var termResult any + _ = client.Post(ctx(), "/workflows/"+c.WorkflowID+"/terminate", body, &termResult) + _ = client.Delete(ctx(), "/workflows/"+c.WorkflowID, &delResult) + } pruned++ if !flagJSON {