diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..42583ab --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,84 @@ +# Changelog + +## 2026-03-08 — Fix dashboard TIME column showing elapsed-since-creation + +### Bug Fixes + +- **dashboard: show actual workflow duration instead of time-since-creation** — The dashboard TIME column was calling `elapsed(w.StartTime)` which computes `now - startTime`, causing completed workflows to show ever-increasing times (e.g. 8m11s for a repo that only took 15s). Changed to use the existing `duration()` function which correctly computes `closeTime - startTime` for completed workflows and `now - startTime` for still-running ones. +- **dashboard: don't fake Completed status from wiki sections** — The dashboard was overriding the Temporal workflow status to "Completed" when all wiki sections existed (17/17), even if the workflow was still Running (e.g. during save_to_arch_hub or cleanup). This caused the dashboard to show repos as Completed while the CLI was still waiting for actual completion. Now uses the real Temporal status and shows "Saving..." when all analysis steps are done but the workflow is still running. + +--- + +## 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/dashboard.go b/internal/commands/dashboard.go index 927817e..8789044 100644 --- a/internal/commands/dashboard.go +++ b/internal/commands/dashboard.go @@ -152,14 +152,15 @@ func renderDashboard(client *api.Client, focusRepo string) error { } } - // If all investigation steps are done, show as Completed - // (Temporal may still show Running during cleanup/arch-hub save) + // Use the actual Temporal workflow status — don't override based on wiki sections, + // since wiki may have results from previous runs and the workflow may still be running. effectiveStatus := w.Status - if completed >= total && total > 0 { - effectiveStatus = "Completed" - } - if effectiveStatus != "Running" && currentStep == "" { + if effectiveStatus == "Running" { + if completed >= total && total > 0 { + currentStep = "Saving..." + } + } else if currentStep == "" { if effectiveStatus == "Completed" { currentStep = "Done" } else { @@ -173,7 +174,7 @@ func renderDashboard(client *api.Client, focusRepo string) error { Completed: completed, Total: total, Current: currentStep, - Elapsed: elapsed(w.StartTime), + Elapsed: duration(w), WorkflowID: w.WorkflowID, }) } 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 {