diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..534108f --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,52 @@ +# Changelog + +## 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/commands/investigate.go b/internal/commands/investigate.go index 4e55e41..36ef691 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() @@ -209,51 +212,210 @@ 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 { + 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 { + 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 { + 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 +430,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..e95198d 100644 --- a/internal/commands/investigate_helpers.go +++ b/internal/commands/investigate_helpers.go @@ -2,9 +2,16 @@ package commands import ( "fmt" + osexec "os/exec" + "path/filepath" + "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 +81,94 @@ 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 +} + // 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 {