diff --git a/CHANGELOG.md b/CHANGELOG.md index ae0500c..2c649ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -92,3 +92,12 @@ Initial implementation. process (and the terminal) stayed open until a second ctrl+c killed it. The relay now abandons the feed the moment the monitor exits; one ctrl+c detaches cleanly, printing the re-attach hint and exiting 0. +- Multi-stage flows: `flow audit-triage` (audit then triage, outputting + triaged findings), `flow remediate-verify` (patch then verify against a + patched checkout with the generated patches applied) and `flow full` + (the whole loop). Stages chain through their artifacts — the audit + report feeds triage (--auto for headless), triaged findings feed patch, + verify audits a fresh clone with the patches applied — and every stage + is an ordinary background run. `flow status` lists flows and stage + detail; interactive flows follow stage-by-stage with Ctrl-C detaching + without cancelling. diff --git a/README.md b/README.md index 97b9cbc..293b7cc 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,22 @@ need is an agent harness you are logged into. | `report ` | Validate (`reporting validate`), lint threat models (`reporting lint`), render (`reporting render`), and browse findings (severity-colored TUI). | | `ledger whoami\|findings\|layers\|verify` | Read-only views over a ledger service (`--url` or `TRAUST_LEDGER_URL`; the token comes from the env var named by `--token-env`). No submissions, no countersign. | +### Flows (multi-stage pipelines) + +```bash +traust-cli flow audit-triage # audit, then triage (--auto); output = triaged findings +traust-cli flow remediate-verify # patch the findings, then verify the patched code +traust-cli flow full # the whole loop end to end +traust-cli flow status [id] # list flows / show stage detail +``` + +Every stage is an ordinary background run (own journal, result, attach +session); each stage's inputs are derived from the previous stage's +artifacts — the audit report feeds triage, the triaged findings feed patch, +and verify runs against a fresh checkout with the generated patches +applied. Ctrl-C detaches without cancelling; `flow status` shows progress +and `attach ` follows the current stage. + ### Exit codes (a contract) | Code | Meaning | diff --git a/cmd/attach.go b/cmd/attach.go index bfe0b60..f334d4d 100644 --- a/cmd/attach.go +++ b/cmd/attach.go @@ -43,8 +43,16 @@ func runAttach(cmd *cobra.Command, args []string) { attachByID(info) } -// attachByID runs the attach flow over a resolved run. +// attachByID runs the attach flow over a resolved run, then exits with the +// run's own code when it finished while attached. func attachByID(info *RunInfo) { + attachOnce(info) + printAttachExit(info) +} + +// attachOnce attaches without any exit handling — the pipeline follower +// reuses it stage after stage. +func attachOnce(info *RunInfo) { // Header from the run's spec. header := tui.MonitorHeader{Skill: info.Skill, RunID: info.ID} if data, err := os.ReadFile(filepath.Join(info.Dir, "spec.json")); err == nil { @@ -73,7 +81,6 @@ func attachByID(info *RunInfo) { if err != nil { fatalf(1, "attach: %v", err) } - printAttachExit(info) } // plainAttach is the non-TTY tail: replay + follow until the run finishes. diff --git a/cmd/flow.go b/cmd/flow.go new file mode 100644 index 0000000..6415226 --- /dev/null +++ b/cmd/flow.go @@ -0,0 +1,322 @@ +package cmd + +import ( + "context" + "fmt" + "os" + "path/filepath" + "sort" + "time" + + "github.com/spf13/cobra" + + "github.com/traust-security/traust-cli/internal/harness" + "github.com/traust-security/traust-cli/internal/theme" +) + +var flowFlags struct { + harnessID string + model string + unsafeFullPermission bool + budgetOverride string + detach bool +} + +var flowCmd = &cobra.Command{ + Use: "flow", + Short: "Multi-stage pipelines: audit → triage → patch → verify", + Long: `Chained skill pipelines. Every stage runs as an ordinary background +run (its own journal, result and attach session), with each stage's inputs +derived from the previous stage's artifacts: + + audit-triage secure-code-audit, then triage (--auto) of + its report. Output: the triaged findings + (-triage.json) — the hand-off point. + + remediate-verify + patch (fix generation) from triaged + findings or an audit report, then + verify-remediation against a patched + checkout with the patches applied. + + full all four stages end to end. + +Ctrl-C detaches without cancelling; the flow keeps going in its own +process. ` + "`flow status`" + ` lists flows, ` + "`flow status `" + ` shows +stage detail, and ` + "`attach `" + ` attaches to the current stage.`, +} + +var flowAuditTriageCmd = &cobra.Command{ + Use: "audit-triage ", + Short: "Audit a repository, then triage the findings", + Args: cobra.ExactArgs(1), + Run: func(cmd *cobra.Command, args []string) { startFlowCmd("audit-triage", args[0], "") }, +} + +var flowRemediateVerifyCmd = &cobra.Command{ + Use: "remediate-verify ", + Short: "Patch triaged findings, then verify the remediations", + Long: `Consumes the output of ` + "`flow audit-triage`" + ` (a -triage.json, +or a -security-audit.json) and the target repository: generates +fixes (PATCHES/), applies them to a fresh checkout, and verifies each +original finding against the patched code.`, + Args: cobra.ExactArgs(2), + Run: func(cmd *cobra.Command, args []string) { startFlowCmd("remediate-verify", args[1], args[0]) }, +} + +var flowFullCmd = &cobra.Command{ + Use: "full ", + Short: "The whole loop: audit, triage, patch, verify", + Args: cobra.ExactArgs(1), + Run: func(cmd *cobra.Command, args []string) { startFlowCmd("full", args[0], "") }, +} + +var flowStatusCmd = &cobra.Command{ + Use: "status [flow]", + Short: "List flows, or show one flow's stages", + Args: cobra.MaximumNArgs(1), + Run: runFlowStatus, +} + +func init() { + rootCmd.AddCommand(flowCmd) + flowCmd.AddCommand(flowAuditTriageCmd, flowRemediateVerifyCmd, flowFullCmd, flowStatusCmd) + for _, c := range []*cobra.Command{flowAuditTriageCmd, flowRemediateVerifyCmd, flowFullCmd} { + c.Flags().StringVar(&flowFlags.harnessID, "harness", "", "agent harness for every stage (default: state default)") + c.Flags().StringVar(&flowFlags.model, "model", "", "model override for every stage (default: per-stage registry floor)") + c.Flags().BoolVar(&flowFlags.unsafeFullPermission, "unsafe-full-permission", false, "grant the harness full permission for every stage (also settable at init)") + c.Flags().StringVar(&flowFlags.budgetOverride, "budget-override", "", "reason to proceed past budget refuse bands (recorded)") + c.Flags().BoolVarP(&flowFlags.detach, "detach", "d", false, "start without attaching (interactive flows attach by default)") + } +} + +func startFlowCmd(flow, target, findingsPath string) { + state, err := loadState() + if err != nil { + fatalf(1, "%v", err) + } + + harnessID := flowFlags.harnessID + if harnessID == "" { + harnessID = state.Harness + } + h, err := harness.Get(harnessID) + if err != nil { + fatalf(1, "%v", err) + } + if d := h.Detect(context.Background()); !d.Available { + fatalf(1, "%s binary %q not found — install it or pick another harness with --harness", h.DisplayName(), h.Binary()) + } + + if findingsPath != "" { + if _, err := os.Stat(findingsPath); err != nil { + fatalf(1, "findings path %q: %v", findingsPath, err) + } + } + + unsafe := flowFlags.unsafeFullPermission || state.FullPermission + id, err := startFlow(state, flowRequest{ + Flow: flow, + Target: target, + FindingsPath: findingsPath, + Harness: h.ID(), + Model: flowFlags.model, + UnsafeFullPermission: unsafe, + BudgetOverride: flowFlags.budgetOverride, + }) + if err != nil { + fatalf(1, "start flow: %v", err) + } + + short := shortFlowID(id) + if flowFlags.detach || !isTTY() { + printFlowStarted(id, short) + return + } + // Interactive: follow the flow stage by stage (attaching to whichever + // stage is current) until it ends or the user detaches. + attachFlow(id) +} + +// attachFlow follows a pipeline: attach to the current stage's run; when +// that stage completes, roll on to the next. Detaching (ctrl+c) during a +// stage leaves the flow running and stops following. +func attachFlow(flowID string) { + state, err := loadState() + if err != nil { + fatalf(1, "%v", err) + } + dir := filepath.Join(flowsDir(state.Workspace), flowID) + runsDir := state.RunsDir() + styles := theme.Default() + + for { + m, err := loadFlowManifest(dir) + if err != nil { + fatalf(1, "flow vanished: %v", err) + } + if m.State != stateRunning { + printFlowManifest(m) + fmt.Println() + if m.State == stateFailed { + os.Exit(1) + } + return + } + stage := m.currentStage() + if stage == nil || stage.RunID == "" || stage.State != stateRunning { + time.Sleep(300 * time.Millisecond) + continue + } + info, err := loadRunInfo(filepath.Join(runsDir, stage.RunID)) + if err != nil || info == nil || info.State != stateRunning { + time.Sleep(300 * time.Millisecond) + continue + } + fmt.Println(styles.Accent.Render(fmt.Sprintf("▸ flow stage: %s (run %s)", stage.Name, shortRunID(stage.RunID)))) + attachOnce(info) + if !resultExists(info.Dir) { + // The stage is still going: the user detached. + fmt.Println() + fmt.Println(styles.OK.Render("▸ detached — the flow continues in the background")) + kv("status", "traust-cli flow status "+shortFlowID(flowID)) + kv("watch", "traust-cli attach "+shortFlowID(flowID)) + return + } + // Stage finished; roll on to the next. + } +} + +func printFlowStarted(id, short string) { + styles := theme.Default() + fmt.Println() + fmt.Println(styles.OK.Render("▸ flow " + id + " started in the background")) + kv("status", "traust-cli flow status "+short) + kv("watch", "traust-cli attach "+short) + kv("stages", "traust-cli runs") +} + +func shortFlowID(id string) string { return shortRunID(id) } + +func runFlowStatus(cmd *cobra.Command, args []string) { + state, err := loadState() + if err != nil { + fatalf(1, "%v", err) + } + dir := flowsDir(state.Workspace) + if len(args) == 1 { + m := resolveFlow(dir, args[0]) + printFlowManifest(m) + return + } + entries, err := os.ReadDir(dir) + if err != nil { + if os.IsNotExist(err) { + fmt.Println(theme.Default().Muted.Render("no flows yet — start one: traust-cli flow full ")) + return + } + fatalf(1, "%v", err) + } + var manifests []*flowManifest + for _, e := range entries { + if !e.IsDir() { + continue + } + if m, err := loadFlowManifest(filepath.Join(dir, e.Name())); err == nil { + manifests = append(manifests, m) + } + } + sort.Slice(manifests, func(i, j int) bool { + ri, rj := manifests[i].State == stateRunning, manifests[j].State == stateRunning + if ri != rj { + return ri + } + return manifests[i].CreatedAt > manifests[j].CreatedAt + }) + styles := theme.Default() + fmt.Printf(" %-14s %-16s %-10s %s\n", styles.Label.Render("FLOW"), styles.Label.Render("KIND"), styles.Label.Render("STATE"), styles.Label.Render("TARGET")) + for _, m := range manifests { + stateStyle := styles.OK + switch m.State { + case stateFailed: + stateStyle = styles.Err + case stateRunning: + stateStyle = styles.Accent + } + fmt.Printf(" %-14s %-16s %-10s %s\n", shortFlowID(m.ID), m.Flow, stateStyle.Render(m.State), m.Target) + } +} + +func resolveFlow(dir, prefix string) *flowManifest { + entries, err := os.ReadDir(dir) + if err != nil { + fatalf(1, "no flows: %v", err) + } + var matches []*flowManifest + for _, e := range entries { + if !e.IsDir() { + continue + } + if e.Name() == prefix || (len(prefix) >= 1 && (e.Name() == prefix || hasSuffixID(e.Name(), prefix))) { + if m, err := loadFlowManifest(filepath.Join(dir, e.Name())); err == nil { + matches = append(matches, m) + } + } + } + switch len(matches) { + case 1: + return matches[0] + case 0: + fatalf(1, "no flow matches %q", prefix) + default: + fatalf(1, "ambiguous flow prefix %q", prefix) + } + return nil +} + +func hasSuffixID(name, prefix string) bool { + return len(name) > len(prefix) && shortRunID(name) == prefix +} + +func printFlowManifest(m *flowManifest) { + styles := theme.Default() + fmt.Println() + fmt.Println(styles.Header.Render("flow " + m.ID)) + kv("kind", m.Flow) + kv("target", m.Target) + kv("state", m.State) + fmt.Println() + for _, s := range m.Stages { + mark := styles.OK.Render("✓") + stateStyle := styles.OK + switch s.State { + case stateFailed: + mark = styles.Err.Render("✗") + stateStyle = styles.Err + case stateRunning: + mark = styles.Accent.Render("▸") + stateStyle = styles.Accent + case "pending": + mark = styles.Muted.Render("·") + stateStyle = styles.Muted + case "skipped": + mark = styles.Muted.Render("-") + stateStyle = styles.Muted + } + line := fmt.Sprintf(" %s %-10s %-16s %s", mark, s.Name, stateStyle.Render(s.State), s.Skill) + if s.RunID != "" { + line += styles.Muted.Render(" (run " + shortRunID(s.RunID) + ")") + } + fmt.Println(line) + if s.Detail != "" { + fmt.Println(styles.Muted.Render(" " + s.Detail)) + } + if s.Artifact != "" && s.State == stateDone { + fmt.Println(styles.Muted.Render(" artifact: " + s.Artifact)) + } + } + if cur := m.currentStage(); cur != nil && cur.RunID != "" && cur.State == stateRunning { + fmt.Println() + fmt.Println(styles.Muted.Render("watch: traust-cli attach " + shortRunID(cur.RunID))) + } +} diff --git a/cmd/pipeline.go b/cmd/pipeline.go new file mode 100644 index 0000000..64aab8e --- /dev/null +++ b/cmd/pipeline.go @@ -0,0 +1,254 @@ +package cmd + +import ( + "context" + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "time" + + "github.com/traust-security/traust-cli/internal/bootstrap" +) + +// flowRequest is the serialized instruction for a pipeline worker: which +// flow, the target repo, and the run options every stage inherits. +type flowRequest struct { + Flow string `json:"flow"` // audit-triage | remediate-verify | full + Target string `json:"target"` + FindingsPath string `json:"findings_path,omitempty"` // remediate-verify entry point + Harness string `json:"harness"` + Model string `json:"model,omitempty"` + UnsafeFullPermission bool `json:"unsafe_full_permission,omitempty"` + BudgetOverride string `json:"budget_override,omitempty"` + Workspace string `json:"workspace"` +} + +// flowStage is one stage's record in the pipeline manifest. +type flowStage struct { + Name string `json:"name"` + Skill string `json:"skill"` + RunID string `json:"run_id,omitempty"` + State string `json:"state"` // pending | running | done | failed | skipped + Detail string `json:"detail,omitempty"` + Artifact string `json:"artifact,omitempty"` +} + +// flowManifest is the pipeline's persisted state, at +// /cli/pipelines//pipeline.json. Stage runs are ordinary run +// directories, so `runs`, `attach`, `result` and `logs` all work on them. +type flowManifest struct { + ID string `json:"id"` + Flow string `json:"flow"` + Target string `json:"target"` + State string `json:"state"` // running | done | failed + CreatedAt string `json:"created_at"` + Stages []flowStage `json:"stages"` +} + +func flowsDir(workspace string) string { return filepath.Join(workspace, "cli", "pipelines") } + +func newFlowID() (string, error) { return newRunID("flow") } + +// loadFlowManifest reads a pipeline manifest by directory. +func loadFlowManifest(dir string) (*flowManifest, error) { + data, err := os.ReadFile(filepath.Join(dir, "pipeline.json")) + if err != nil { + return nil, err + } + var m flowManifest + if err := json.Unmarshal(data, &m); err != nil { + return nil, err + } + return &m, nil +} + +func (m *flowManifest) save(dir string) error { + data, err := json.MarshalIndent(m, "", " ") + if err != nil { + return err + } + tmp := filepath.Join(dir, "pipeline.json.tmp") + if err := os.WriteFile(tmp, data, 0o644); err != nil { + return err + } + return os.Rename(tmp, filepath.Join(dir, "pipeline.json")) +} + +// currentStage returns the running stage, else the last non-pending one. +func (m *flowManifest) currentStage() *flowStage { + var last *flowStage + for i := range m.Stages { + s := &m.Stages[i] + if s.State == stateRunning { + return s + } + if s.State != "pending" { + last = s + } + } + return last +} + +// startFlow spawns a detached pipeline worker and returns the flow id. +func startFlow(state *bootstrap.State, req flowRequest) (string, error) { + id, err := newFlowID() + if err != nil { + return "", err + } + dir := filepath.Join(flowsDir(state.Workspace), id) + if err := os.MkdirAll(dir, 0o755); err != nil { + return "", err + } + req.Workspace = state.Workspace + data, err := json.MarshalIndent(req, "", " ") + if err != nil { + return "", err + } + if err := os.WriteFile(filepath.Join(dir, "request.json"), data, 0o644); err != nil { + return "", err + } + + // Initial manifest so `flow status` works from the first second. + manifest := &flowManifest{ID: id, Flow: req.Flow, Target: req.Target, State: stateRunning, CreatedAt: time.Now().UTC().Format(time.RFC3339)} + manifest.Stages = flowStagePlan(req.Flow) + if err := manifest.save(dir); err != nil { + return "", err + } + + // Write the pid file up front: the handshake below accepts it. + if err := os.WriteFile(filepath.Join(dir, "pid"), []byte(fmt.Sprintf("%d", os.Getpid())), 0o644); err != nil { + return "", err + } + + exe, err := os.Executable() + if err != nil { + return "", err + } + log, err := os.Create(filepath.Join(dir, "worker.log")) + if err != nil { + return "", err + } + defer log.Close() + cmd := exec.Command(exe, "_pipeline", dir) + cmd.SysProcAttr = detachedSysProcAttr() + cmd.Env = os.Environ() + cmd.Stdout = log + cmd.Stderr = log + if err := cmd.Start(); err != nil { + return "", err + } + go func() { _ = cmd.Wait() }() + return id, nil +} + +// flowStagePlan is the static stage list per flow; the worker fills in run +// ids and states as it goes. +func flowStagePlan(flow string) []flowStage { + switch flow { + case "audit-triage": + return []flowStage{ + {Name: "audit", Skill: "secure-code-audit", State: "pending"}, + {Name: "triage", Skill: "triage", State: "pending"}, + } + case "remediate-verify": + return []flowStage{ + {Name: "patch", Skill: "patch", State: "pending"}, + {Name: "verify", Skill: "verify-remediation", State: "pending"}, + } + case "full": + return []flowStage{ + {Name: "audit", Skill: "secure-code-audit", State: "pending"}, + {Name: "triage", Skill: "triage", State: "pending"}, + {Name: "patch", Skill: "patch", State: "pending"}, + {Name: "verify", Skill: "verify-remediation", State: "pending"}, + } + } + return nil +} + +// runStage executes one pipeline stage as an ordinary background-style run +// (in-process, exactly like the run worker) and returns its exit code. +func runStage(dir string, req flowRequest, skill, runID string, args []string) (int, error) { + stageDir := filepath.Join(filepath.Dir(filepath.Dir(dir)), "runs", runID) + if err := os.MkdirAll(stageDir, 0o755); err != nil { + return 0, err + } + stageReq := runRequest{ + RunID: runID, + Workspace: req.Workspace, + Skill: skill, + Args: args, + Harness: req.Harness, + Model: req.Model, + UnsafeFullPermission: req.UnsafeFullPermission, + BudgetOverride: req.BudgetOverride, + DeclareSpend: true, + } + data, err := json.MarshalIndent(stageReq, "", " ") + if err != nil { + return 0, err + } + if err := os.WriteFile(filepath.Join(stageDir, "options.json"), data, 0o644); err != nil { + return 0, err + } + return runWorker(stageDir), nil +} + +// findRunArtifact returns the path of a run's first valid artifact matching +// suffix, from its result.json. +func findRunArtifact(runDir, suffix string) (string, error) { + data, err := os.ReadFile(filepath.Join(runDir, "result.json")) + if err != nil { + return "", err + } + var res struct { + Artifacts []struct { + Path string `json:"path"` + Valid bool `json:"valid"` + } `json:"artifacts"` + } + if err := json.Unmarshal(data, &res); err != nil { + return "", err + } + for _, a := range res.Artifacts { + if (a.Valid || suffix == ".diff") && len(a.Path) >= len(suffix) && a.Path[len(a.Path)-len(suffix):] == suffix { + return a.Path, nil + } + } + return "", fmt.Errorf("no %s artifact in %s", suffix, runDir) +} + +// buildPatchedRepo clones the target and applies every generated patch, +// producing the patched checkout the verify stage audits. +func buildPatchedRepo(ctx context.Context, target, flowDir, dst string) (applied int, err error) { + if out, err := exec.CommandContext(ctx, "git", "clone", "--quiet", target, dst).CombinedOutput(); err != nil { + return 0, fmt.Errorf("clone %s: %v: %s", target, err, string(out)) + } + patchesRoot := filepath.Join(harnessRootOf(flowDir), "PATCHES") + entries, err := filepath.Glob(filepath.Join(patchesRoot, "bug_*", "patch.diff")) + if err != nil { + return 0, err + } + for _, diff := range entries { + check := exec.CommandContext(ctx, "git", "-C", dst, "apply", "--check", diff) + if err := check.Run(); err != nil { + continue // a conflicting patch is skipped, not fatal + } + if err := exec.CommandContext(ctx, "git", "-C", dst, "apply", diff).Run(); err != nil { + continue + } + applied++ + } + return applied, nil +} + +// harnessRootOf resolves the harness root (where stage artifacts like +// PATCHES/ land) from a run or flow directory inside the workspace: +// /cli/runs/ or /cli/pipelines/. +func harnessRootOf(dir string) string { + cli := filepath.Dir(filepath.Dir(dir)) + ws := filepath.Dir(cli) + return filepath.Join(ws, "traust") +} diff --git a/cmd/pipeline_test.go b/cmd/pipeline_test.go new file mode 100644 index 0000000..278ac56 --- /dev/null +++ b/cmd/pipeline_test.go @@ -0,0 +1,159 @@ +package cmd + +import ( + "context" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" +) + +// TestPipelineAuditTriageChain runs a two-stage flow in-process with the +// fake claude harness: the audit stage's artifact must become the triage +// stage's input (plus --auto), and the manifest must record both stages. +func TestPipelineAuditTriageChain(t *testing.T) { + t.Setenv("TRAUST_CLI_CLAUDE_BIN", fakeClaudePath(t)) + ws := t.TempDir() + state := setupWorkerWorkspace(t, ws) + + // The fake harness writes fixture-security-audit.json into its cwd (the + // harness root) for every skill, so both stages "succeed" and produce + // the report artifact; that is enough to verify the chaining. + flowID := "20261021-170000-flow-aa01" + flowDir := filepath.Join(flowsDir(ws), flowID) + if err := os.MkdirAll(flowDir, 0o755); err != nil { + t.Fatal(err) + } + req := flowRequest{Flow: "audit-triage", Target: "https://github.com/example/demo-shop", Harness: "claude", Workspace: ws} + data, _ := json.Marshal(req) + os.WriteFile(filepath.Join(flowDir, "request.json"), data, 0o644) + manifest := &flowManifest{ID: flowID, Flow: req.Flow, Target: req.Target, State: stateRunning, Stages: flowStagePlan("audit-triage")} + if err := manifest.save(flowDir); err != nil { + t.Fatal(err) + } + + if code := runPipelineWorker(flowDir); code != 0 { + t.Fatalf("pipeline exit = %d", code) + } + + m, err := loadFlowManifest(flowDir) + if err != nil { + t.Fatal(err) + } + if m.State != stateDone { + t.Fatalf("flow state = %s, want done", m.State) + } + if len(m.Stages) != 2 { + t.Fatalf("stages = %+v", m.Stages) + } + for _, s := range m.Stages { + if s.State != stateDone || s.RunID == "" { + t.Errorf("stage %s = %+v", s.Name, s) + } + } + + // The triage stage's options.json must carry the audit artifact and + // --auto: phase-1 output wired into phase-2 input. + triageStage := m.Stages[1] + optsData, err := os.ReadFile(filepath.Join(state.RunsDir(), triageStage.RunID, "options.json")) + if err != nil { + t.Fatal(err) + } + var triageReq runRequest + if err := json.Unmarshal(optsData, &triageReq); err != nil { + t.Fatal(err) + } + if len(triageReq.Args) != 2 || !strings.HasSuffix(triageReq.Args[0], "-security-audit.json") || triageReq.Args[1] != "--auto" { + t.Errorf("triage args = %v (want [, --auto])", triageReq.Args) + } + if triageReq.Args[0] != m.Stages[0].Artifact { + t.Errorf("triage input %q != audit artifact %q", triageReq.Args[0], m.Stages[0].Artifact) + } + if m.Stages[1].Artifact != m.Stages[0].Artifact { + // The fake writes the same fixture file for both skills, so the + // triage artifact equals the audit artifact; what matters is that + // chaining used it. + t.Logf("note: triage artifact = %q (fake harness writes the same file)", m.Stages[1].Artifact) + } +} + +// TestPipelineStopsOnStageFailure verifies a failing stage fails the flow +// and later stages stay pending. +func TestPipelineStopsOnStageFailure(t *testing.T) { + t.Setenv("TRAUST_CLI_CLAUDE_BIN", fakeClaudePath(t)) + t.Setenv("FAKE_CLAUDE_MODE", "fail") + ws := t.TempDir() + setupWorkerWorkspace(t, ws) + + flowID := "20261021-170100-flow-bb02" + flowDir := filepath.Join(flowsDir(ws), flowID) + os.MkdirAll(flowDir, 0o755) + req := flowRequest{Flow: "full", Target: "https://github.com/example/demo-shop", Harness: "claude", Workspace: ws} + data, _ := json.Marshal(req) + os.WriteFile(filepath.Join(flowDir, "request.json"), data, 0o644) + manifest := &flowManifest{ID: flowID, Flow: req.Flow, Target: req.Target, State: stateRunning, Stages: flowStagePlan("full")} + manifest.save(flowDir) + + if code := runPipelineWorker(flowDir); code == 0 { + t.Fatal("pipeline should fail when the audit stage fails") + } + m, _ := loadFlowManifest(flowDir) + if m.State != stateFailed || m.Stages[0].State != stateFailed { + t.Fatalf("manifest = %+v", m) + } + for _, s := range m.Stages[1:] { + if s.State != "pending" { + t.Errorf("stage %s should be pending, got %s", s.Name, s.State) + } + } +} + +// TestBuildPatchedRepo covers the patch-application step against a real git +// repository: applicable patches apply, conflicting ones are skipped. +func TestBuildPatchedRepo(t *testing.T) { + ctx := context.Background() + + // Source repo with one committed file. + src := t.TempDir() + run := func(args ...string) string { + cmd := exec.Command("git", args...) + cmd.Dir = src + cmd.Env = append(os.Environ(), "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@t", "GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@t") + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("git %v: %v: %s", args, err, out) + } + return string(out) + } + run("init", "-q") + os.WriteFile(filepath.Join(src, "a.txt"), []byte("one\n"), 0o644) + os.WriteFile(filepath.Join(src, "b.txt"), []byte("bee\n"), 0o644) + run("add", ".") + run("commit", "-qm", "init") + + // PATCHES layout under a fake workspace harness root. + ws := t.TempDir() + patches := filepath.Join(ws, "traust", "PATCHES", "bug_01") + os.MkdirAll(patches, 0o755) + os.WriteFile(filepath.Join(patches, "patch.diff"), []byte("diff --git a/a.txt b/a.txt\n--- a/a.txt\n+++ b/a.txt\n@@ -1 +1 @@\n-one\n+two\n"), 0o644) + conflict := filepath.Join(ws, "traust", "PATCHES", "bug_02") + os.MkdirAll(conflict, 0o755) + os.WriteFile(filepath.Join(conflict, "patch.diff"), []byte("diff --git a/missing.txt b/missing.txt\n--- a/missing.txt\n+++ b/missing.txt\n@@ -1 +1 @@\n-absent\n+present\n"), 0o644) + + dst := filepath.Join(t.TempDir(), "patched") + flowDir := filepath.Join(ws, "cli", "pipelines", "x") + os.MkdirAll(flowDir, 0o755) + applied, err := buildPatchedRepo(ctx, src, flowDir, dst) + if err != nil { + t.Fatal(err) + } + if applied != 1 { + t.Fatalf("applied = %d, want 1 (conflicting patch skipped)", applied) + } + data, _ := os.ReadFile(filepath.Join(dst, "a.txt")) + if string(data) != "two\n" { + t.Errorf("a.txt = %q, patch not applied", data) + } +} diff --git a/cmd/pipeline_worker.go b/cmd/pipeline_worker.go new file mode 100644 index 0000000..feb848f --- /dev/null +++ b/cmd/pipeline_worker.go @@ -0,0 +1,177 @@ +package cmd + +import ( + "context" + "encoding/json" + "fmt" + "os" + "os/signal" + "path/filepath" + "syscall" + + "github.com/spf13/cobra" +) + +var pipelineWorkerCmd = &cobra.Command{ + Use: "_pipeline ", + Short: "internal: execute one background pipeline", + Hidden: true, + Args: cobra.ExactArgs(1), + Run: func(cmd *cobra.Command, args []string) { + os.Exit(runPipelineWorker(args[0])) + }, +} + +func init() { + rootCmd.AddCommand(pipelineWorkerCmd) +} + +// runPipelineWorker executes a flow: stages run sequentially as ordinary +// runs, each stage's inputs derived from the previous stage's artifacts, +// with the manifest updated after every transition. Stop signals cancel the +// current stage's run (the worker tears it down gracefully) and fail the +// flow. +func runPipelineWorker(dir string) int { + manifest, err := loadFlowManifest(dir) + if err != nil { + fmt.Fprintf(os.Stderr, "pipeline: load manifest: %v\n", err) + return 1 + } + data, err := os.ReadFile(filepath.Join(dir, "request.json")) + if err != nil { + fmt.Fprintf(os.Stderr, "pipeline: read request: %v\n", err) + return 1 + } + var req flowRequest + if err := json.Unmarshal(data, &req); err != nil { + fmt.Fprintf(os.Stderr, "pipeline: parse request: %v\n", err) + return 1 + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + // A cancelled context fails the flow after the current stage unwinds. + cancelled := func() bool { return ctx.Err() != nil } + + runsDir := filepath.Join(filepath.Dir(filepath.Dir(dir)), "runs") + auditReport := "" + triageJSON := req.FindingsPath + + fail := func(stage *flowStage, code int, detail string) int { + stage.State = stateFailed + stage.Detail = detail + manifest.State = stateFailed + manifest.save(dir) + return code + } + + for i := range manifest.Stages { + stage := &manifest.Stages[i] + if stage.State == stateDone || stage.State == "skipped" { + continue + } + + // Derive this stage's arguments from what previous stages produced. + var args []string + switch stage.Name { + case "audit": + args = []string{req.Target} + case "triage": + if auditReport == "" { + // Entry via remediate-verify or a completed audit-triage + // flow: the findings path may itself be a report. + src := triageJSON + if src == "" { + return fail(stage, 1, "no audit report to triage") + } + args = []string{src, "--auto"} + } else { + args = []string{auditReport, "--auto"} + } + case "patch": + src := triageJSON + if src == "" { + return fail(stage, 1, "no triaged findings to patch") + } + args = []string{src} + if req.Target != "" { + args = append(args, "--repo", req.Target) + } + case "verify": + if auditReport == "" { + // remediate-verify entry: the findings path carries the + // report when the user passes a security-audit json. + if len(filepath.Ext(triageJSON)) == 4 { // .json + auditReport = triageJSON + } + } + if auditReport == "" { + stage.State = "skipped" + stage.Detail = "no original audit report available for verification" + manifest.save(dir) + continue + } + applied, err := buildPatchedRepo(ctx, req.Target, dir, filepath.Join(dir, "patched-repo")) + if err != nil { + return fail(stage, 1, fmt.Sprintf("build patched repo: %v", err)) + } + if applied == 0 { + stage.State = "skipped" + stage.Detail = "no patches were applicable — nothing to verify" + manifest.save(dir) + continue + } + stage.Detail = fmt.Sprintf("%d patch(es) applied", applied) + args = []string{auditReport, filepath.Join(dir, "patched-repo")} + default: + return fail(stage, 1, "unknown stage "+stage.Name) + } + + stage.State = stateRunning + manifest.save(dir) + + runID, err := newRunID(stage.Skill) + if err != nil { + return fail(stage, 1, err.Error()) + } + stage.RunID = runID + manifest.save(dir) + + code, err := runStage(dir, req, stage.Skill, runID, args) + if err != nil { + return fail(stage, 1, err.Error()) + } + if cancelled() { + return fail(stage, code, "pipeline cancelled") + } + if code != 0 { + return fail(stage, code, fmt.Sprintf("stage exited %d — see `traust-cli result %s`", code, shortRunID(runID))) + } + + stageRunDir := filepath.Join(runsDir, runID) + switch stage.Name { + case "audit": + artifact, err := findRunArtifact(stageRunDir, "-security-audit.json") + if err != nil { + return fail(stage, 1, err.Error()) + } + auditReport = artifact + stage.Artifact = artifact + case "triage": + artifact, err := findRunArtifact(stageRunDir, "-triage.json") + if err != nil { + return fail(stage, 1, err.Error()) + } + triageJSON = artifact + stage.Artifact = artifact + case "patch": + stage.Artifact = filepath.Join(harnessRootOf(dir), "PATCHES.json") + } + stage.State = stateDone + manifest.save(dir) + } + + manifest.State = stateDone + manifest.save(dir) + return 0 +} diff --git a/cmd/worker_test.go b/cmd/worker_test.go index be43c1c..8c92ddc 100644 --- a/cmd/worker_test.go +++ b/cmd/worker_test.go @@ -28,6 +28,21 @@ func setupWorkerWorkspace(t *testing.T, ws string) *bootstrap.State { if err := os.WriteFile(filepath.Join(ws, "traust", "harnessing", "3-audit", "secure-code-audit", "SKILL.md"), []byte(skill), 0o644); err != nil { t.Fatal(err) } + // The pipeline skills, for flow chaining tests. + for _, s := range []struct{ dir, name string }{ + {"4-triage/triage", "triage"}, + {"7-remediate/patch", "patch"}, + {"8-verify/verify-remediation", "verify-remediation"}, + } { + d := filepath.Join(ws, "traust", "harnessing", s.dir) + if err := os.MkdirAll(d, 0o755); err != nil { + t.Fatal(err) + } + body := "---\nname: " + s.name + "\ndescription: test skill\nmetadata:\n harness:\n tier: primary\n---\nRun on $ARGUMENTS.\n" + if err := os.WriteFile(filepath.Join(d, "SKILL.md"), []byte(body), 0o644); err != nil { + t.Fatal(err) + } + } // A fake venv python: the worker delegates validation and spend to it, // and this double reports success for every engine call. if err := os.MkdirAll(filepath.Join(ws, "traust", ".venv", "bin"), 0o755); err != nil { diff --git a/internal/runner/runner_test.go b/internal/runner/runner_test.go index 3ba4617..99481f0 100644 --- a/internal/runner/runner_test.go +++ b/internal/runner/runner_test.go @@ -137,7 +137,7 @@ func TestRunPipelineClaudeFake(t *testing.T) { if res.Usage.TokensIn != 100 || res.Usage.TokensOut != 50 || res.Usage.CostUSD != 0.42 { t.Errorf("usage = %+v", res.Usage) } - if res.ResultText != "audit complete" { + if res.ResultText != "stage complete" { t.Errorf("result text = %q", res.ResultText) } if len(res.Artifacts) != 1 { diff --git a/testdata/fake-harnesses/fake-claude.sh b/testdata/fake-harnesses/fake-claude.sh index 9bb0f6d..f153ede 100755 --- a/testdata/fake-harnesses/fake-claude.sh +++ b/testdata/fake-harnesses/fake-claude.sh @@ -2,13 +2,18 @@ # Fake Claude Code for traust-cli integration tests. # # Consumes the prompt on stdin, emits a canned stream-json transcript, and -# writes a fixture report into the working directory (the fake harness root). +# writes a fixture artifact into the working directory (the fake harness +# root) matching the skill named in the prompt: +# triage → fixture-triage.json +# patch → PATCHES.json + PATCHES/bug_01/patch.diff +# otherwise → fixture-security-audit.json # FAKE_CLAUDE_MODE selects a scenario: # needs-input emits an AskUserQuestion tool_use # fail writes an error to stderr and exits 3 # partial writes a report but exits 1 (failed run, partial output) +# slow sleeps 30s first (background/stop testing), then completes set -eu -cat > /dev/null +PROMPT="$(cat)" if [ "${FAKE_CLAUDE_MODE:-}" = "needs-input" ]; then printf '%s\n' '{"type":"assistant","message":{"content":[{"type":"tool_use","name":"AskUserQuestion","input":{}}]}}' @@ -27,8 +32,17 @@ fi printf '%s\n' '{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"starting audit"}]}}' printf '%s\n' '{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Bash","input":{}}]}}' -printf '%s\n' '{"type":"result","subtype":"success","result":"audit complete","usage":{"input_tokens":100,"output_tokens":50},"total_cost_usd":0.42}' -printf '{"findings": [], "metadata": {"harness_version": "test"}}' > fixture-security-audit.json +printf '%s\n' '{"type":"result","subtype":"success","result":"stage complete","usage":{"input_tokens":100,"output_tokens":50},"total_cost_usd":0.42}' + +if printf '%s' "$PROMPT" | grep -q 'harness skill "triage"'; then + printf '{"findings": [], "triage_completed": "2026-10-21"}' > fixture-triage.json +elif printf '%s' "$PROMPT" | grep -q 'harness skill "patch"'; then + mkdir -p PATCHES/bug_01 + printf '{"patches": [{"bug": "bug_01", "status": "generated"}]}' > PATCHES.json + printf 'diff --git a/a.txt b/a.txt\n--- a/a.txt\n+++ b/a.txt\n@@ -1 +1 @@\n-one\n+two\n' > PATCHES/bug_01/patch.diff +else + printf '{"findings": [], "metadata": {"harness_version": "test"}}' > fixture-security-audit.json +fi if [ "${FAKE_CLAUDE_MODE:-}" = "partial" ]; then exit 1