From 3c5170600744bfe01344f65d510a8c4e07679f78 Mon Sep 17 00:00:00 2001 From: Gurney Buchanan Date: Wed, 21 Oct 2026 14:05:27 -0400 Subject: [PATCH] Add multi-stage flows: audit-triage, remediate-verify, full MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The harness's core loop โ€” audit, triage, patch, verify โ€” was four manual invocations whose wiring lived in the skills' docs: triage needs the audit report plus --auto to run headless, patch consumes the triaged findings, and verify needs the original report and a patched checkout that nothing produced automatically. flow commands chain the stages through their artifacts. Each stage runs as an ordinary background run (its own journal, result and attach session), and the pipeline worker derives the next stage's inputs from the previous stage's result: the audit report feeds triage (--auto), the triaged findings feed patch, and verify audits a fresh clone of the target with the generated PATCHES applied (conflicting patches are skipped, not fatal). flow audit-triage audit + triage; output = triaged findings, the hand-off point flow remediate-verify patch + verify from triaged findings (or an audit report) flow full the whole loop flow status [id] list flows / stage detail Flows are detached workers like runs: Ctrl-C detaches without cancelling, interactive flows follow stage-by-stage, and every stage is inspectable with attach/result/logs. Verified end to end against the fake harness and a real git target: all four stages chained, the patch applied to the verify checkout, failure stopping the flow with later stages pending. Tests cover the artifact chaining (phase-1 output wired into phase-2 input), failure propagation, and patch application against real git repositories. ๐Ÿ’˜ Generated with Crush Assisted-by: Crush:rits/zai-org/glm-5-3 --- CHANGELOG.md | 9 + README.md | 16 ++ cmd/attach.go | 11 +- cmd/flow.go | 322 +++++++++++++++++++++++++ cmd/pipeline.go | 254 +++++++++++++++++++ cmd/pipeline_test.go | 159 ++++++++++++ cmd/pipeline_worker.go | 177 ++++++++++++++ cmd/worker_test.go | 15 ++ internal/runner/runner_test.go | 2 +- testdata/fake-harnesses/fake-claude.sh | 22 +- 10 files changed, 980 insertions(+), 7 deletions(-) create mode 100644 cmd/flow.go create mode 100644 cmd/pipeline.go create mode 100644 cmd/pipeline_test.go create mode 100644 cmd/pipeline_worker.go 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