feat: sequential investigate mode and bulk workflow terminate

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/<id>` 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/<id>` 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.

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"`

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(&parallel, "parallel", 3, "Parallel limit (daily only)")
+	cmd.Flags().IntVar(&parallel, "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")