From 481408951e0f172d719794296d5633e5e7af8988 Mon Sep 17 00:00:00 2001 From: Gurney Buchanan Date: Wed, 21 Oct 2026 12:14:11 -0400 Subject: [PATCH] Run skills in the background, watchable and detachable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Runs used to live inside the launching terminal: closing it (or an incautious ctrl+c) killed a 10-minute audit, and one attach session owned the only live view. Every run now executes in a detached worker process (setsid, survives the terminal): `run` starts it in the background and attaches by default, `-d` fires and forgets. The worker owns the run's lifetime, turns stop signals into the same graceful context-driven teardown an interactive abort used, and writes the existing journal (spec, stream, result, pid) that everything else reads. The journal is the event bus: `attach` replays it from the start and follows it live — ctrl+c detaches without cancelling (the common case is now instant and non-destructive), x twice cancels, and attaching to a finished run replays the whole transcript with its result. `runs` lists run states (running first, then newest), `stop` cancels through a SIGINT/SIGTERM/SIGKILL ladder, `logs` renders the transcript, and `result` prints the stored outcome exiting with the run's own code. Non-TTY attach degrades to a plain live tail; `--json` and `--no-monitor` keep the synchronous CI semantics. The monitor also shows tool results now — crush tool_result parts, truncated to keep the transcript readable — which are the bulk of a run's output and were previously invisible. Journal raw lines no longer duplicate parsed events, so replays render once. Verified end to end against the fake harness: detached start, live listing, ctrl+c detach with the run surviving, graceful stop writing an aborted result, replay of finished runs, and no orphaned workers. Fifty-five tests (race detector on), vet and staticcheck clean. 💘 Generated with Crush Assisted-by: Crush:rits/zai-org/glm-5-3 --- CHANGELOG.md | 11 ++ README.md | 7 +- cmd/attach.go | 250 ++++++++++++++++++++++++ cmd/background.go | 144 ++++++++++++++ cmd/detach_unix.go | 11 ++ cmd/detach_windows.go | 12 ++ cmd/run.go | 99 +++++++--- cmd/runs.go | 241 +++++++++++++++++++++++ cmd/stop.go | 166 ++++++++++++++++ cmd/worker.go | 120 ++++++++++++ cmd/worker_test.go | 244 +++++++++++++++++++++++ internal/harness/crush_serve.go | 36 ++++ internal/harness/crush_serve_test.go | 23 ++- internal/harness/harness.go | 2 + internal/runner/journal.go | 85 ++++++++ internal/runner/journal_test.go | 95 +++++++++ internal/runner/runner.go | 15 +- internal/tui/monitor.go | 259 +++++++++++-------------- testdata/fake-harnesses/fake-claude.sh | 5 + 19 files changed, 1646 insertions(+), 179 deletions(-) create mode 100644 cmd/attach.go create mode 100644 cmd/background.go create mode 100644 cmd/detach_unix.go create mode 100644 cmd/detach_windows.go create mode 100644 cmd/runs.go create mode 100644 cmd/stop.go create mode 100644 cmd/worker.go create mode 100644 cmd/worker_test.go create mode 100644 internal/runner/journal_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index b9eac61..b8017be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -76,3 +76,14 @@ Initial implementation. the command list (still invocable), over-long command summaries were shortened, and the init --forge default URL moved out of the flag line (an unbreakable token that could not fit pflag's description column). +- Runs execute in the background by default: `traust-cli run` spawns a + detached worker process (setsid, survives the terminal) and attaches to + it; `--detach` fires and forgets. The run journal is the event bus: + `attach` replays it and follows it live (Ctrl-C detaches without + cancelling — x twice cancels), `runs` lists run states, `stop` cancels + through a graceful SIGINT/SIGTERM/SIGKILL ladder, `logs` renders the + transcript and `result` prints the stored outcome with the run's own + exit code. Attaching to a finished run replays the full transcript. +- The monitor now shows tool results (crush tool_result parts, truncated), + which are the bulk of a run's output; journal raw lines no longer + duplicate parsed events, so attach replays render once. diff --git a/README.md b/README.md index 1821caf..97b9cbc 100644 --- a/README.md +++ b/README.md @@ -77,7 +77,12 @@ need is an agent harness you are logged into. | `doctor` | CLI checks + `install_traust --doctor`; `--json` for CI. | | `update` | Move the pin to a newer tag and re-sync. | | `skills` | Catalog parsed from the same SKILL.md frontmatter the harness reads; TUI picker on a TTY, table otherwise; `--json`. | -| `run [args…]` | Compose the SKILL.md into a prompt, pre-flight the model floor, drive the harness headlessly, journal everything, discover and validate artifacts, declare spend. Live TUI monitor on a TTY. | +| `run [args…]` | Compose the SKILL.md into a prompt, pre-flight the model floor, and launch the run **in the background** (a detached worker process that survives the terminal), then attach. `-d/--detach` starts without attaching; `--json`/`--no-monitor` keep synchronous CI semantics. The journal records everything; artifacts are validated and spend declared. | +| `attach ` | Live monitor over a background run: full transcript replay + live tail with thinking, tool calls and tool results. **Ctrl-C detaches without cancelling** (the run keeps going); `x` twice cancels. Works on finished runs too (full replay + result). Non-TTY degrades to a plain tail. | +| `runs` (`ls`/`ps`) | List runs — running first, then newest — with state (running/done/failed/needs-input/budget-stop/aborted/orphaned), age and tokens; `--json`. | +| `stop ` | Graceful cancel: SIGINT (context-driven teardown: crush cancel endpoint, server shutdown) → SIGTERM → SIGKILL. | +| `logs ` | The run's rendered journal transcript. | +| `result ` | The stored result: summary, artifacts, usage, error — exits with the run's own code. | | `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. | diff --git a/cmd/attach.go b/cmd/attach.go new file mode 100644 index 0000000..bfe0b60 --- /dev/null +++ b/cmd/attach.go @@ -0,0 +1,250 @@ +package cmd + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/spf13/cobra" + + "github.com/traust-security/traust-cli/internal/harness" + "github.com/traust-security/traust-cli/internal/runner" + "github.com/traust-security/traust-cli/internal/theme" + "github.com/traust-security/traust-cli/internal/tui" +) + +var attachCmd = &cobra.Command{ + Use: "attach ", + Short: "Watch a run live (Ctrl-C detaches without cancelling)", + Long: `Attach to a background run: replay its journal from the start and follow +it live. Ctrl-C, q or d detaches — the run keeps going in its own process. +x twice cancels the run. Attaching to a finished run replays the whole +transcript and shows the result. Without a terminal, attach degrades to a +plain live tail.`, + Args: cobra.ExactArgs(1), + Run: runAttach, +} + +func init() { + rootCmd.AddCommand(attachCmd) +} + +func runAttach(cmd *cobra.Command, args []string) { + state, err := loadState() + if err != nil { + fatalf(1, "%v", err) + } + info, err := resolveRun(state.RunsDir(), args[0]) + if err != nil { + fatalf(1, "%v", err) + } + attachByID(info) +} + +// attachByID runs the attach flow over a resolved run. +func attachByID(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 { + var spec runner.Spec + if json.Unmarshal(data, &spec) == nil { + header.Harness = spec.Harness + header.Model = spec.Model + header.BudgetCap = spec.BudgetCap + if t, err := time.Parse(time.RFC3339, spec.StartedAt); err == nil { + header.StartedAt = t + } + } + } + + if !isTTY() { + plainAttach(info) + return + } + + follow := make(chan tui.MonitorItem, 128) + go func() { + tailJournal(info.Dir, follow) + }() + + _, err := tui.Attach(header, follow, func() { stopRun(info) }) + if err != nil { + fatalf(1, "attach: %v", err) + } + printAttachExit(info) +} + +// plainAttach is the non-TTY tail: replay + follow until the run finishes. +func plainAttach(info *RunInfo) { + styles := theme.Default() + fmt.Println(styles.Accent.Render(fmt.Sprintf("▸ attaching to %s (%s) — Ctrl-C detaches", info.ID, info.State))) + items := make(chan tui.MonitorItem, 128) + go func() { + tailJournal(info.Dir, items) + }() + for it := range items { + switch it.Kind { + case tui.ItemEvent: + renderPlainEvent(it.Event) + case tui.ItemRaw: + if it.Text != "" { + fmt.Println(tui.Wrap(styles.Muted.Render(" "+it.Text), tui.TerminalWidth())) + } + case tui.ItemProgress: + fmt.Println(tui.Wrap(styles.Accent.Render("› "+it.Text), tui.TerminalWidth())) + case tui.ItemDone: + fmt.Println(styles.Muted.Render("— run complete —")) + } + } + printResultFromDir(info.Dir) +} + +// tailJournal replays the run's stream.jsonl from the start and follows it +// until result.json appears (or the run is already finished). Header usage +// is refreshed in passing (budget cap arrives with spec.json). +func tailJournal(dir string, out chan<- tui.MonitorItem) { + defer close(out) + stream := filepath.Join(dir, "stream.jsonl") + + // Wait briefly for the journal to exist (the worker may still be in + // pre-flight when attach races the start). + deadline := time.Now().Add(20 * time.Second) + for { + if _, err := os.Stat(stream); err == nil { + break + } + if resultExists(dir) || time.Now().After(deadline) { + emitResultLines(dir, out) + return + } + time.Sleep(150 * time.Millisecond) + } + + tailer, err := runner.NewTailer(stream, 0) + if err != nil { + out <- tui.MonitorItem{Kind: tui.ItemProgress, Text: "warning: cannot open journal: " + err.Error()} + emitResultLines(dir, out) + return + } + defer tailer.Close() + + for { + lines, err := tailer.Read() + if err != nil { + out <- tui.MonitorItem{Kind: tui.ItemProgress, Text: "warning: journal read error: " + err.Error()} + break + } + for _, line := range lines { + if it, ok := journalLineToItem(line); ok { + out <- it + } + } + if resultExists(dir) { + // Drain any final lines, then finish. + for { + lines, err := tailer.Read() + if err != nil || len(lines) == 0 { + break + } + for _, line := range lines { + if it, ok := journalLineToItem(line); ok { + out <- it + } + } + } + break + } + time.Sleep(200 * time.Millisecond) + } + emitResultLines(dir, out) +} + +func resultExists(dir string) bool { + _, err := os.Stat(filepath.Join(dir, "result.json")) + return err == nil +} + +// journalLineToItem converts one stream.jsonl line into a monitor item. +func journalLineToItem(line string) (tui.MonitorItem, bool) { + entry, err := runner.ParseEntry([]byte(line)) + if err != nil { + return tui.MonitorItem{}, false + } + switch entry.Kind { + case "event": + if entry.Event == nil { + return tui.MonitorItem{}, false + } + return tui.MonitorItem{Kind: tui.ItemEvent, Event: harnessEvent(*entry.Event)}, true + case "raw": + return tui.MonitorItem{Kind: tui.ItemRaw, Text: entry.Raw}, true + } + return tui.MonitorItem{}, false +} + +// harnessEvent widens a journal event record into the harness event shape +// (identical fields, different types for the kind). +func harnessEvent(e runner.Event) harness.Event { + return harness.Event{ + Kind: harness.EventKind(e.Kind), + Text: e.Text, + Tool: e.Tool, + TokensIn: e.TokensIn, + TokensOut: e.TokensOut, + CostUSD: e.CostUSD, + IsError: e.IsError, + } +} + +// emitResultLines appends terminal progress lines from result.json. +func emitResultLines(dir string, out chan<- tui.MonitorItem) { + data, err := os.ReadFile(filepath.Join(dir, "result.json")) + if err != nil { + out <- tui.MonitorItem{Kind: tui.ItemProgress, Text: "warning: run ended without a result (orphaned worker?)"} + return + } + var res runner.RunResult + if json.Unmarshal(data, &res) != nil { + return + } + for _, a := range res.Artifacts { + mark := "artifact" + if !a.Valid { + mark = "artifact failed schema validation" + } + out <- tui.MonitorItem{Kind: tui.ItemProgress, Text: fmt.Sprintf("%s: %s", mark, a.Path)} + } + if res.HarnessError != "" { + out <- tui.MonitorItem{Kind: tui.ItemProgress, Text: "harness error: " + res.HarnessError} + } +} + +func printAttachExit(info *RunInfo) { + styles := theme.Default() + short := shortRunID(info.ID) + refreshed, err := loadRunInfo(info.Dir) + if err == nil && refreshed != nil && refreshed.State != stateRunning { + // The run finished while attached: show the summary. + printResultFromDir(info.Dir) + return + } + fmt.Println() + fmt.Println(styles.OK.Render("▸ detached — the run continues in the background")) + kv("re-attach", "traust-cli attach "+short) + kv("result", "traust-cli result "+short) +} + +// printResultFromDir renders the stored result.json as the run summary. +func printResultFromDir(dir string) { + data, err := os.ReadFile(filepath.Join(dir, "result.json")) + if err != nil { + fatalf(1, "no result for this run (orphaned?) — journal: %s", dir) + } + var res runner.RunResult + if err := json.Unmarshal(data, &res); err != nil { + fatalf(1, "parse result.json: %v", err) + } + printRunSummary(&res) +} diff --git a/cmd/background.go b/cmd/background.go new file mode 100644 index 0000000..31a4cc9 --- /dev/null +++ b/cmd/background.go @@ -0,0 +1,144 @@ +package cmd + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + "github.com/traust-security/traust-cli/internal/bootstrap" + "github.com/traust-security/traust-cli/internal/theme" +) + +// runRequest is the parent's serialized instruction for the worker: it is +// everything needed to reconstruct the run after the launching terminal is +// gone. Stored as options.json in the run directory. +type runRequest struct { + RunID string `json:"run_id"` + Workspace string `json:"workspace"` + Skill string `json:"skill"` + Args []string `json:"args"` + Harness string `json:"harness"` + Model string `json:"model,omitempty"` + MaxTurns int `json:"max_turns,omitempty"` + UnsafeFullPermission bool `json:"unsafe_full_permission,omitempty"` + BudgetOverride string `json:"budget_override,omitempty"` + StrictValidation bool `json:"strict_validation,omitempty"` + DeclareSpend bool `json:"declare_spend"` +} + +// newRunID mints a collision-safe run id: timestamp + skill + random suffix. +func newRunID(skill string) (string, error) { + buf := make([]byte, 2) + if _, err := rand.Read(buf); err != nil { + return "", err + } + return fmt.Sprintf("%s-%s-%s", time.Now().Format("20060102-150405"), skill, hex.EncodeToString(buf)), nil +} + +// startBackgroundRun launches a detached worker for the run and returns once +// the worker has taken ownership (pid file present). The worker survives +// this process and the terminal it was launched from. +func startBackgroundRun(state *bootstrap.State, req runRequest) (string, error) { + runID := req.RunID + dir := filepath.Join(state.RunsDir(), runID) + if err := os.MkdirAll(dir, 0o755); err != nil { + return "", fmt.Errorf("create run dir: %w", err) + } + req.Workspace = state.Workspace + req.DeclareSpend = true + data, err := json.MarshalIndent(req, "", " ") + if err != nil { + return "", err + } + if err := os.WriteFile(filepath.Join(dir, "options.json"), data, 0o644); err != nil { + return "", err + } + + exe, err := os.Executable() + if err != nil { + return "", fmt.Errorf("resolve executable: %w", err) + } + workerLog, err := os.Create(filepath.Join(dir, "worker.log")) + if err != nil { + return "", err + } + defer workerLog.Close() + + cmd := exec.Command(exe, "_worker", dir) + cmd.SysProcAttr = detachedSysProcAttr() + cmd.Env = os.Environ() + cmd.Stdout = workerLog + cmd.Stderr = workerLog + if err := cmd.Start(); err != nil { + return "", fmt.Errorf("start worker: %w", err) + } + // Reap the worker when it eventually exits; the parent never blocks on + // it, and an unreaped child would read as alive to the pid probe below. + go func() { _ = cmd.Wait() }() + + // Handshake: wait for the worker's pid file — or, for runs that finish + // almost instantly, its result.json. The worker writes the pid before + // loading anything and removes it at exit, so a completed run can beat + // the parent's first poll. + pidPath := filepath.Join(dir, "pid") + deadline := time.Now().Add(15 * time.Second) + for time.Now().Before(deadline) { + if _, err := os.Stat(pidPath); err == nil { + return runID, nil + } + if _, err := os.Stat(filepath.Join(dir, "result.json")); err == nil { + return runID, nil + } + if !processAlive(cmd.Process.Pid) { + return "", fmt.Errorf("worker exited during startup — see %s", filepath.Join(dir, "worker.log")) + } + time.Sleep(100 * time.Millisecond) + } + return "", fmt.Errorf("worker did not start within 15s — see %s", filepath.Join(dir, "worker.log")) +} + +// printStarted shows the background-run affordances. +func printStarted(runID string) { + styles := theme.Default() + short := shortRunID(runID) + fmt.Println() + fmt.Println(styles.OK.Render("▸ run " + runID + " started in the background")) + kv("watch", "traust-cli attach "+short) + kv("status", "traust-cli runs") + kv("stop", "traust-cli stop "+short) + kv("result", "traust-cli result "+short) +} + +// shortRunID returns the unambiguous short form of a run id: the random +// 4-hex suffix for background runs. Legacy ids (no suffix) come back +// unchanged — they are historical, and full-id prefix matching still works. +func shortRunID(runID string) string { + parts := strings.Split(runID, "-") + if len(parts) >= 4 { + last := parts[len(parts)-1] + if isHexSuffix(last) { + return last + } + } + return runID +} + +// isHexSuffix reports whether s is a 4-character hex string (the random +// suffix minted by newRunID). +func isHexSuffix(s string) bool { + if len(s) != 4 { + return false + } + for _, r := range s { + if !((r >= '0' && r <= '9') || (r >= 'a' && r <= 'f')) { + return false + } + } + return true +} diff --git a/cmd/detach_unix.go b/cmd/detach_unix.go new file mode 100644 index 0000000..1691967 --- /dev/null +++ b/cmd/detach_unix.go @@ -0,0 +1,11 @@ +//go:build !windows + +package cmd + +import "syscall" + +// detachedSysProcAttr puts the worker in its own session so it survives the +// launching terminal. +func detachedSysProcAttr() *syscall.SysProcAttr { + return &syscall.SysProcAttr{Setsid: true} +} diff --git a/cmd/detach_windows.go b/cmd/detach_windows.go new file mode 100644 index 0000000..8051937 --- /dev/null +++ b/cmd/detach_windows.go @@ -0,0 +1,12 @@ +//go:build windows + +package cmd + +import "syscall" + +// detachedSysProcAttr creates a new process group on Windows; there is no +// session concept equivalent to setsid, but detaching from the console +// process group keeps the worker alive past the launcher. +func detachedSysProcAttr() *syscall.SysProcAttr { + return &syscall.SysProcAttr{CreationFlags: syscall.CREATE_NEW_PROCESS_GROUP} +} diff --git a/cmd/run.go b/cmd/run.go index 6e96ae3..b31a279 100644 --- a/cmd/run.go +++ b/cmd/run.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "os" + "path/filepath" "strings" "github.com/charmbracelet/huh" @@ -28,6 +29,7 @@ var runFlags struct { json bool strictValidation bool noMonitor bool + detach bool } var runCmd = &cobra.Command{ @@ -54,7 +56,8 @@ func init() { runCmd.Flags().StringVar(&runFlags.budgetOverride, "budget-override", "", "reason to proceed past the budget refuse band (recorded)") runCmd.Flags().BoolVar(&runFlags.json, "json", false, "machine-readable result summary on stdout") runCmd.Flags().BoolVar(&runFlags.strictValidation, "strict", false, "strict schema validation of artifacts") - runCmd.Flags().BoolVar(&runFlags.noMonitor, "no-monitor", false, "plain output even on a TTY") + runCmd.Flags().BoolVar(&runFlags.noMonitor, "no-monitor", false, "plain output even on a TTY (synchronous, in-process)") + runCmd.Flags().BoolVarP(&runFlags.detach, "detach", "d", false, "start in the background without attaching (interactive runs attach by default)") } func runRun(cmd *cobra.Command, args []string) { @@ -126,8 +129,6 @@ func executeRun(state *bootstrap.State, skill *catalog.Skill, args []string) { fatalf(1, "%s binary %q not found — install it or pick another harness with --harness", h.DisplayName(), h.Binary()) } - client := api.NewPythonCLI(state.VenvPython(), state.ChildEnv(), state.TraustRepo()) - // Full-permission mode: explicit per-run flag, or the init-time consent // recorded in state. The harness's skills require unattended tool use. unsafeFullPermission := runFlags.unsafeFullPermission || state.FullPermission @@ -139,9 +140,47 @@ func executeRun(state *bootstrap.State, skill *catalog.Skill, args []string) { warnNote(fmt.Sprintf("%q is not in the standalone profile — it may expect campaign infrastructure (siblings, ledger, tracker)", skill.Name)) } + // Interactive terminals launch the run in the BACKGROUND and attach: + // closing the terminal (or detaching with ctrl+c) leaves the run alive. + // --detach skips the attach; non-TTY and --json/--no-monitor keep the + // synchronous in-process semantics (CI needs the exit code and output). + interactive := isTTY() && !runFlags.json && !runFlags.noMonitor + if runFlags.detach || interactive { + runID, err := newRunID(skill.Name) + if err != nil { + fatalf(1, "mint run id: %v", err) + } + runID, err = startBackgroundRun(state, runRequest{ + RunID: runID, + Skill: skill.Name, + Args: args, + Harness: h.ID(), + Model: runFlags.model, + MaxTurns: runFlags.maxTurns, + UnsafeFullPermission: unsafeFullPermission, + BudgetOverride: runFlags.budgetOverride, + StrictValidation: runFlags.strictValidation, + }) + if err != nil { + fatalf(1, "start background run: %v", err) + } + if !interactive || runFlags.detach { + if runFlags.json { + out, _ := json.MarshalIndent(map[string]string{"run_id": runID}, "", " ") + fmt.Println(string(out)) + } else { + printStarted(runID) + } + return + } + attachRunID(state, runID) + return + } + + // Synchronous path. opts := runner.Options{ State: state, - API: client, + API: api.NewPythonCLI(state.VenvPython(), state.ChildEnv(), state.TraustRepo()), Harness: h, Skill: skill, Args: args, @@ -152,30 +191,10 @@ func executeRun(state *bootstrap.State, skill *catalog.Skill, args []string) { StrictValidation: runFlags.strictValidation, DeclareSpend: true, } - - var res *runner.RunResult - interactive := isTTY() && !runFlags.json && !runFlags.noMonitor - if interactive { - runCtx, cancel := context.WithCancel(ctx) - defer cancel() - err = tui.RunMonitor(tui.MonitorHeader{ - Skill: skill.Name, Harness: h.ID(), Model: runFlags.model, - }, cancel, func(sink runner.Sink) error { - r, rerr := runner.Run(runCtx, opts, sink) - if r != nil { - res = r - } - return rerr - }) - } else { - res, err = runner.Run(ctx, opts, plainSink{}) - } + res, err := runner.Run(ctx, opts, plainSink{}) if err != nil { fatalf(1, "run: %v", err) } - if res == nil { - fatalf(1, "run produced no result") - } if runFlags.json { out, _ := json.MarshalIndent(res, "", " ") @@ -188,6 +207,16 @@ func executeRun(state *bootstrap.State, skill *catalog.Skill, args []string) { os.Exit(res.ExitCode) } +// attachRunID attaches the monitor to a freshly started background run and +// exits with the run's code when it completes while attached. +func attachRunID(state *bootstrap.State, runID string) { + info, err := loadRunInfo(filepath.Join(state.RunsDir(), runID)) + if err != nil || info == nil { + fatalf(1, "run %s vanished", runID) + } + runAttach(&cobra.Command{}, []string{runID}) +} + func printRunSummary(res *runner.RunResult) { styles := theme.Default() section("Run result") @@ -269,7 +298,11 @@ func (plainSink) Progress(msg string) { fmt.Println(tui.Wrap(style.Render("› "+msg), tui.TerminalWidth())) } -func (plainSink) Event(ev harness.Event) { +func (plainSink) Event(ev harness.Event) { renderPlainEvent(ev) } + +// renderPlainEvent prints one harness event as styled plain text — shared +// by the synchronous sink, plain attach and `logs`. +func renderPlainEvent(ev harness.Event) { styles := theme.Default() switch ev.Kind { case harness.EventText: @@ -277,7 +310,19 @@ func (plainSink) Event(ev harness.Event) { case harness.EventThinking: fmt.Println(tui.Wrap(styles.Muted.Render("· "+ev.Text), tui.TerminalWidth())) case harness.EventToolCall: - fmt.Println(styles.Accent.Render("→ " + ev.Tool)) + line := "→ " + ev.Tool + if ev.Text != "" { + line += " " + ev.Text + } + fmt.Println(tui.Wrap(styles.Accent.Render(line), tui.TerminalWidth())) + case harness.EventToolResult: + style := styles.Muted + if ev.IsError { + style = styles.Warn + } + for _, l := range strings.Split(ev.Text, "\n") { + fmt.Println(tui.Wrap(style.Render(" │ "+l), tui.TerminalWidth())) + } case harness.EventResult: // Result text is printed in the summary. case harness.EventError: diff --git a/cmd/runs.go b/cmd/runs.go new file mode 100644 index 0000000..8e97d29 --- /dev/null +++ b/cmd/runs.go @@ -0,0 +1,241 @@ +package cmd + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "time" + + "github.com/spf13/cobra" + + "github.com/traust-security/traust-cli/internal/runner" + "github.com/traust-security/traust-cli/internal/theme" + "github.com/traust-security/traust-cli/internal/tui" +) + +// Run states. +const ( + stateRunning = "running" + stateDone = "done" + stateFailed = "failed" + stateNeedsIn = "needs-input" + stateBudget = "budget-stop" + stateAborted = "aborted" + stateOrphaned = "orphaned" +) + +// RunInfo is the resolved state of one run directory. +type RunInfo struct { + ID string `json:"id"` + Dir string `json:"dir"` + Skill string `json:"skill"` + State string `json:"state"` + Exit *int `json:"exit_code,omitempty"` + PID int `json:"pid,omitempty"` + Tokens struct { + In int64 `json:"in"` + Out int64 `json:"out"` + } `json:"tokens"` + StartedAt time.Time `json:"started_at"` +} + +// listRuns resolves every run directory under the workspace, newest first. +func listRuns(runsDir string) ([]RunInfo, error) { + entries, err := os.ReadDir(runsDir) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + var infos []RunInfo + for _, e := range entries { + if !e.IsDir() || strings.HasPrefix(e.Name(), ".") { + continue + } + info, err := loadRunInfo(filepath.Join(runsDir, e.Name())) + if err != nil || info == nil { + continue + } + infos = append(infos, *info) + } + // Running first, then by start time descending. + sort.SliceStable(infos, func(i, j int) bool { + ri, rj := infos[i].State == stateRunning, infos[j].State == stateRunning + if ri != rj { + return ri + } + return infos[i].StartedAt.After(infos[j].StartedAt) + }) + return infos, nil +} + +// loadRunInfo resolves one run directory's state from its files: result.json +// (terminal states), the pid file plus liveness (running), neither (orphaned +// — the worker died without writing a result, e.g. SIGKILL). +func loadRunInfo(dir string) (*RunInfo, error) { + id := filepath.Base(dir) + info := &RunInfo{ID: id, Dir: dir, Skill: unknownSkill(dir)} + + // spec.json carries the skill and start time. + if data, err := os.ReadFile(filepath.Join(dir, "spec.json")); err == nil { + var spec runner.Spec + if json.Unmarshal(data, &spec) == nil { + info.Skill = spec.Skill + if t, err := time.Parse(time.RFC3339, spec.StartedAt); err == nil { + info.StartedAt = t + } + } + } + + // result.json decides terminal states. + if data, err := os.ReadFile(filepath.Join(dir, "result.json")); err == nil { + var res runner.RunResult + if json.Unmarshal(data, &res) == nil { + info.Tokens.In = res.Usage.TokensIn + info.Tokens.Out = res.Usage.TokensOut + exit := res.ExitCode + info.Exit = &exit + switch { + case res.BudgetRefused: + info.State = stateBudget + case res.NeedsInput: + info.State = stateNeedsIn + case res.Aborted: + info.State = stateAborted + case res.ExitCode == 0: + info.State = stateDone + default: + info.State = stateFailed + } + return info, nil + } + } + + // No result: running or orphaned. + if pidBytes, err := os.ReadFile(filepath.Join(dir, "pid")); err == nil { + if pid, err := strconv.Atoi(strings.TrimSpace(string(pidBytes))); err == nil { + info.PID = pid + if processAlive(pid) { + info.State = stateRunning + return info, nil + } + } + } + info.State = stateOrphaned + if info.StartedAt.IsZero() { + // options.json's mtime anchors orphaned runs that never wrote spec. + if fi, err := os.Stat(filepath.Join(dir, "options.json")); err == nil { + info.StartedAt = fi.ModTime() + } + } + return info, nil +} + +func unknownSkill(dir string) string { + if data, err := os.ReadFile(filepath.Join(dir, "options.json")); err == nil { + var req runRequest + if json.Unmarshal(data, &req) == nil { + return req.Skill + } + } + return "?" +} + +// resolveRun finds the unique run matching an id prefix (or the full id). +func resolveRun(runsDir, prefix string) (*RunInfo, error) { + infos, err := listRuns(runsDir) + if err != nil { + return nil, err + } + var matches []*RunInfo + for i := range infos { + if infos[i].ID == prefix || strings.HasPrefix(infos[i].ID, prefix) || shortRunID(infos[i].ID) == prefix { + matches = append(matches, &infos[i]) + } + } + switch len(matches) { + case 1: + return matches[0], nil + case 0: + return nil, fmt.Errorf("no run matches %q — see `traust-cli runs`", prefix) + default: + var ids []string + for _, m := range matches { + ids = append(ids, m.ID) + } + return nil, fmt.Errorf("ambiguous run prefix %q matches: %s", prefix, strings.Join(ids, ", ")) + } +} + +var runsJSON bool + +var runsCmd = &cobra.Command{ + Use: "runs", + Aliases: []string{"ls", "ps"}, + Short: "List runs (running first, then newest)", + Run: func(cmd *cobra.Command, args []string) { + state, err := loadState() + if err != nil { + fatalf(1, "%v", err) + } + infos, err := listRuns(state.RunsDir()) + if err != nil { + fatalf(1, "%v", err) + } + if runsJSON { + out, _ := json.MarshalIndent(infos, "", " ") + fmt.Println(string(out)) + return + } + if len(infos) == 0 { + fmt.Println(theme.Default().Muted.Render("no runs yet — start one: traust-cli run [args…]")) + return + } + styles := theme.Default() + fmt.Printf(" %-14s %-24s %-12s %-10s %s\n", styles.Label.Render("RUN"), styles.Label.Render("SKILL"), styles.Label.Render("STATE"), styles.Label.Render("AGE"), styles.Label.Render("TOKENS")) + for _, r := range infos { + stateStyle := styles.OK + switch r.State { + case stateFailed, stateOrphaned: + stateStyle = styles.Err + case stateRunning: + stateStyle = styles.Accent + case stateAborted, stateBudget, stateNeedsIn: + stateStyle = styles.Warn + } + age := "—" + if !r.StartedAt.IsZero() { + age = humanAge(time.Since(r.StartedAt)) + } + tokens := "—" + if r.Tokens.In+r.Tokens.Out > 0 { + tokens = fmt.Sprintf("in %dk out %dk", r.Tokens.In/1000, r.Tokens.Out/1000) + } + fmt.Printf(" %-14s %-24s %-12s %-10s %s\n", + tui.Truncate(shortRunID(r.ID), 14, "…"), + tui.Truncate(r.Skill, 24, "…"), + stateStyle.Render(r.State), age, tokens) + } + }, +} + +func humanAge(d time.Duration) string { + switch { + case d < time.Minute: + return fmt.Sprintf("%ds", int(d.Seconds())) + case d < time.Hour: + return fmt.Sprintf("%dm", int(d.Minutes())) + default: + return fmt.Sprintf("%dh", int(d.Hours())) + } +} + +func init() { + rootCmd.AddCommand(runsCmd) + runsCmd.Flags().BoolVar(&runsJSON, "json", false, "machine-readable listing") +} diff --git a/cmd/stop.go b/cmd/stop.go new file mode 100644 index 0000000..547ac87 --- /dev/null +++ b/cmd/stop.go @@ -0,0 +1,166 @@ +package cmd + +import ( + "fmt" + "os" + "path/filepath" + "syscall" + "time" + + "github.com/spf13/cobra" + + "github.com/traust-security/traust-cli/internal/theme" + "github.com/traust-security/traust-cli/internal/tui" +) + +var stopCmd = &cobra.Command{ + Use: "stop ", + Short: "Cancel a background run (graceful: SIGINT, then SIGTERM, then SIGKILL)", + Args: cobra.ExactArgs(1), + Run: runStop, +} + +var logsCmd = &cobra.Command{ + Use: "logs ", + Short: "Print a run's journal (rendered transcript)", + Args: cobra.ExactArgs(1), + Run: runLogs, +} + +var resultCmd = &cobra.Command{ + Use: "result ", + Short: "Print a run's stored result (summary, artifacts, usage)", + Args: cobra.ExactArgs(1), + Run: runResult, +} + +func init() { + rootCmd.AddCommand(stopCmd, logsCmd, resultCmd) +} + +func runStop(cmd *cobra.Command, args []string) { + state, err := loadState() + if err != nil { + fatalf(1, "%v", err) + } + info, err := resolveRun(state.RunsDir(), args[0]) + if err != nil { + fatalf(1, "%v", err) + } + stopRun(info) +} + +// stopRun cancels a background run through the signal ladder, verbosely +// (for the CLI command). The worker turns the first signal into a context +// cancellation, which tears the harness down through the graceful paths +// (crush cancel endpoint, server shutdown) and writes result.json with +// aborted=true. +func stopRun(info *RunInfo) { + if !signalRunStop(info) { + return + } + if waitForResult(info.Dir, 15*time.Second) { + fmt.Println(theme.Default().OK.Render("▸ run cancelled — result written")) + return + } + fmt.Println(theme.Default().Warn.Render("› graceful cancel did not finish — sending SIGTERM")) + syscall.Kill(info.PID, syscall.SIGTERM) + if waitForResult(info.Dir, 10*time.Second) { + fmt.Println(theme.Default().OK.Render("▸ run cancelled — result written")) + return + } + fmt.Println(theme.Default().Err.Render("✗ worker did not stop — sending SIGKILL (no result will be written; the run will show as orphaned)")) + syscall.Kill(info.PID, syscall.SIGKILL) +} + +// signalRunStop sends the first (graceful) stop signal; false when there is +// nothing to stop. It prints nothing — the attach monitor uses it as its +// cancel callback, where stdout belongs to the TUI. +func signalRunStop(info *RunInfo) bool { + if info.State != stateRunning { + return false + } + if !processAlive(info.PID) { + return false + } + return syscall.Kill(info.PID, syscall.SIGINT) == nil +} + +func waitForResult(dir string, timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if resultExists(dir) { + return true + } + time.Sleep(200 * time.Millisecond) + } + return false +} + +func runLogs(cmd *cobra.Command, args []string) { + state, err := loadState() + if err != nil { + fatalf(1, "%v", err) + } + info, err := resolveRun(state.RunsDir(), args[0]) + if err != nil { + fatalf(1, "%v", err) + } + styles := theme.Default() + stream := filepath.Join(info.Dir, "stream.jsonl") + data, err := os.ReadFile(stream) + if err != nil { + fatalf(1, "no journal at %s: %v", stream, err) + } + fmt.Println(styles.Muted.Render("journal: " + stream)) + for _, line := range splitLines(string(data)) { + if it, ok := journalLineToItem(line); ok { + switch it.Kind { + case tui.ItemEvent: + renderPlainEvent(it.Event) + case tui.ItemRaw: + if it.Text != "" { + fmt.Println(tui.Wrap(styles.Muted.Render(" "+it.Text), tui.TerminalWidth())) + } + } + } + } + if resultExists(info.Dir) { + printResultFromDir(info.Dir) + } +} + +func splitLines(s string) []string { + var out []string + start := 0 + for i := 0; i < len(s); i++ { + if s[i] == '\n' { + if i > start { + out = append(out, s[start:i]) + } + start = i + 1 + } + } + if start < len(s) { + out = append(out, s[start:]) + } + return out +} + +func runResult(cmd *cobra.Command, args []string) { + state, err := loadState() + if err != nil { + fatalf(1, "%v", err) + } + info, err := resolveRun(state.RunsDir(), args[0]) + if err != nil { + fatalf(1, "%v", err) + } + if info.State == stateRunning { + styles := theme.Default() + fmt.Println(styles.Accent.Render(fmt.Sprintf("run %s is still running — `traust-cli attach %s` to watch it", info.ID, shortRunID(info.ID)))) + return + } + printResultFromDir(info.Dir) + os.Exit(*info.Exit) +} diff --git a/cmd/worker.go b/cmd/worker.go new file mode 100644 index 0000000..f41ee60 --- /dev/null +++ b/cmd/worker.go @@ -0,0 +1,120 @@ +package cmd + +import ( + "context" + "encoding/json" + "fmt" + "os" + "os/signal" + "path/filepath" + "strconv" + "syscall" + + "github.com/spf13/cobra" + + "github.com/traust-security/traust-cli/internal/api" + "github.com/traust-security/traust-cli/internal/bootstrap" + "github.com/traust-security/traust-cli/internal/catalog" + "github.com/traust-security/traust-cli/internal/harness" + "github.com/traust-security/traust-cli/internal/runner" +) + +// processAlive reports whether a pid is alive (signal 0 probe). +func processAlive(pid int) bool { + if pid <= 0 { + return false + } + return syscall.Kill(pid, 0) == nil +} + +func init() { + rootCmd.AddCommand(workerCmd) +} + +var workerCmd = &cobra.Command{ + Use: "_worker ", + Short: "internal: execute one background run", + Hidden: true, + Args: cobra.ExactArgs(1), + Run: func(cmd *cobra.Command, args []string) { + os.Exit(runWorker(args[0])) + }, +} + +// runWorker executes one background run from its run directory: options.json +// in, journal and result.json out. It owns the run's lifetime; stop signals +// (SIGINT/SIGTERM) cancel the run context, which tears the harness down +// through the same graceful paths an interactive abort uses. +func runWorker(dir string) int { + pidPath := filepath.Join(dir, "pid") + if err := os.WriteFile(pidPath, []byte(strconv.Itoa(os.Getpid())), 0o644); err != nil { + fmt.Fprintf(os.Stderr, "worker: write pid: %v\n", err) + return 1 + } + defer os.Remove(pidPath) + + var req runRequest + data, err := os.ReadFile(filepath.Join(dir, "options.json")) + if err != nil { + fmt.Fprintf(os.Stderr, "worker: read options: %v\n", err) + return 1 + } + if err := json.Unmarshal(data, &req); err != nil { + fmt.Fprintf(os.Stderr, "worker: parse options: %v\n", err) + return 1 + } + + state, err := bootstrap.LoadState(req.Workspace) + if err != nil { + fmt.Fprintf(os.Stderr, "worker: load state: %v\n", err) + return 1 + } + cat, err := catalog.Load(state.TraustRepo()) + if err != nil { + fmt.Fprintf(os.Stderr, "worker: load catalog: %v\n", err) + return 1 + } + skill, ok := cat.ByName(req.Skill) + if !ok { + fmt.Fprintf(os.Stderr, "worker: unknown skill %q\n", req.Skill) + return 1 + } + h, err := harness.Get(req.Harness) + if err != nil { + fmt.Fprintf(os.Stderr, "worker: %v\n", err) + return 1 + } + + client := api.NewPythonCLI(state.VenvPython(), state.ChildEnv(), state.TraustRepo()) + + // Stop signals cancel the run; the runner's teardown (crush cancel + // endpoint, server shutdown, process kill) is entirely context-driven. + ctx, stop := context.WithCancel(context.Background()) + defer stop() + sigs := make(chan os.Signal, 1) + signal.Notify(sigs, os.Interrupt, syscall.SIGTERM) + go func() { + <-sigs + stop() + }() + + res, err := runner.Run(ctx, runner.Options{ + State: state, + API: client, + Harness: h, + Skill: skill, + Args: req.Args, + Model: req.Model, + MaxTurns: req.MaxTurns, + UnsafeFullPermission: req.UnsafeFullPermission, + BudgetOverrideReason: req.BudgetOverride, + StrictValidation: req.StrictValidation, + DeclareSpend: req.DeclareSpend, + RunID: req.RunID, + }, nil) + if err != nil { + fmt.Fprintf(os.Stderr, "worker: %v\n", err) + return 1 + } + return res.ExitCode +} diff --git a/cmd/worker_test.go b/cmd/worker_test.go new file mode 100644 index 0000000..be43c1c --- /dev/null +++ b/cmd/worker_test.go @@ -0,0 +1,244 @@ +package cmd + +import ( + "encoding/json" + "os" + "path/filepath" + "strconv" + "testing" + "time" + + "github.com/traust-security/traust-cli/internal/bootstrap" +) + +// setupWorkerWorkspace builds a minimal workspace: state.json, a one-skill +// harnessing fixture tree, config dir. Returns the loaded bootstrap state. +func setupWorkerWorkspace(t *testing.T, ws string) *bootstrap.State { + t.Helper() + for _, d := range []string{ + filepath.Join(ws, "cli"), + filepath.Join(ws, "config"), + filepath.Join(ws, "traust", "harnessing", "3-audit", "secure-code-audit"), + } { + if err := os.MkdirAll(d, 0o755); err != nil { + t.Fatal(err) + } + } + skill := "---\nname: secure-code-audit\ndescription: test skill\nmetadata:\n harness:\n tier: primary\n---\nAudit $ARGUMENTS.\n" + if err := os.WriteFile(filepath.Join(ws, "traust", "harnessing", "3-audit", "secure-code-audit", "SKILL.md"), []byte(skill), 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 { + t.Fatal(err) + } + fakePy := "#!/bin/sh\nexit 0\n" + if err := os.WriteFile(filepath.Join(ws, "traust", ".venv", "bin", "python3"), []byte(fakePy), 0o755); err != nil { + t.Fatal(err) + } + stateJSON := map[string]any{ + "version": 1, + "workspace": ws, + "config_home": filepath.Join(ws, "config"), + "traust_ref": "test", + "traust_sha": "0000000000000000000000000000000000000000", + "harness": "claude", + "created_at": time.Now().UTC().Format(time.RFC3339), + "updated_at": time.Now().UTC().Format(time.RFC3339), + } + data, _ := json.Marshal(stateJSON) + if err := os.WriteFile(filepath.Join(ws, "cli", "state.json"), data, 0o644); err != nil { + t.Fatal(err) + } + workspaceFlag = ws + state, err := loadState() + if err != nil { + t.Fatal(err) + } + return state +} + +func fakeClaudePath(t *testing.T) string { + t.Helper() + p, err := filepath.Abs(filepath.Join("..", "testdata", "fake-harnesses", "fake-claude.sh")) + if err != nil { + t.Fatal(err) + } + return p +} + +// TestRunWorkerRoundTrip executes the worker in-process (exactly the code +// path the spawned process runs) against a temp workspace and the fake +// claude harness: options.json in, journal and result.json out. +func TestRunWorkerRoundTrip(t *testing.T) { + t.Setenv("TRAUST_CLI_CLAUDE_BIN", fakeClaudePath(t)) + ws := t.TempDir() + state := setupWorkerWorkspace(t, ws) + + runID := "20261021-120000-test-abcd" + dir := filepath.Join(state.RunsDir(), runID) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + req := runRequest{ + RunID: runID, + Workspace: ws, + Skill: "secure-code-audit", + Args: []string{"https://github.com/org/repo"}, + Harness: "claude", + DeclareSpend: true, + } + data, _ := json.MarshalIndent(req, "", " ") + if err := os.WriteFile(filepath.Join(dir, "options.json"), data, 0o644); err != nil { + t.Fatal(err) + } + + code := runWorker(dir) + if code != 0 { + t.Fatalf("worker exit = %d", code) + } + + for _, f := range []string{"result.json", "stream.jsonl", "spec.json"} { + if _, err := os.Stat(filepath.Join(dir, f)); err != nil { + t.Fatalf("missing %s: %v", f, err) + } + } + if _, err := os.Stat(filepath.Join(dir, "pid")); !os.IsNotExist(err) { + t.Errorf("pid file should be removed after completion: %v", err) + } + + var res map[string]any + rd, _ := os.ReadFile(filepath.Join(dir, "result.json")) + if err := json.Unmarshal(rd, &res); err != nil { + t.Fatal(err) + } + if res["run_id"] != runID { + t.Errorf("run_id = %v (the parent-minted id must flow through)", res["run_id"]) + } + if res["exit_code"] != float64(0) { + t.Errorf("exit_code = %v", res["exit_code"]) + } + + info, err := loadRunInfo(dir) + if err != nil || info == nil { + t.Fatal(err) + } + if info.State != stateDone { + t.Errorf("state = %s, want done", info.State) + } +} + +// TestRunWorkerFailurePath verifies a failing harness records a result and +// exits non-zero. +func TestRunWorkerFailurePath(t *testing.T) { + t.Setenv("TRAUST_CLI_CLAUDE_BIN", fakeClaudePath(t)) + t.Setenv("FAKE_CLAUDE_MODE", "fail") + ws := t.TempDir() + state := setupWorkerWorkspace(t, ws) + + runID := "20261021-120100-test-ffff" + dir := filepath.Join(state.RunsDir(), runID) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + req := runRequest{RunID: runID, Workspace: ws, Skill: "secure-code-audit", Harness: "claude", DeclareSpend: true} + data, _ := json.Marshal(req) + os.WriteFile(filepath.Join(dir, "options.json"), data, 0o644) + + if code := runWorker(dir); code != 1 { + t.Fatalf("worker exit = %d, want 1", code) + } + info, err := loadRunInfo(dir) + if err != nil || info == nil { + t.Fatal(err) + } + if info.State != stateFailed { + t.Errorf("state = %s, want failed", info.State) + } +} + +// TestResolveRunPrefix covers id-prefix resolution and ambiguity. +func TestResolveRunPrefix(t *testing.T) { + ws := t.TempDir() + state := setupWorkerWorkspace(t, ws) + + a := filepath.Join(state.RunsDir(), "20261021-120000-test-aaaa") + b := filepath.Join(state.RunsDir(), "20261021-120500-test-bbbb") + for _, d := range []string{a, b} { + os.MkdirAll(d, 0o755) + os.WriteFile(filepath.Join(d, "options.json"), []byte(`{"skill":"vuln-scan"}`), 0o644) + } + + if info, err := resolveRun(state.RunsDir(), "aaaa"); err != nil || info.ID != filepath.Base(a) { + t.Errorf("suffix resolve = %+v (%v)", info, err) + } + if info, err := resolveRun(state.RunsDir(), filepath.Base(a)); err != nil || info.ID != filepath.Base(a) { + t.Errorf("full-id resolve = %+v (%v)", info, err) + } + if _, err := resolveRun(state.RunsDir(), "20261021-12"); err == nil { + t.Error("ambiguous prefix should fail") + } + if _, err := resolveRun(state.RunsDir(), "zzzz"); err == nil { + t.Error("unknown prefix should fail") + } +} + +// TestRunInfoStates covers the state machine over journal artifacts. +func TestRunInfoStates(t *testing.T) { + ws := t.TempDir() + state := setupWorkerWorkspace(t, ws) + + // Running: pid file pointing at this (live) test process. + running := filepath.Join(state.RunsDir(), "20261021-130000-test-1111") + os.MkdirAll(running, 0o755) + os.WriteFile(filepath.Join(running, "pid"), []byte(strconv.Itoa(os.Getpid())), 0o644) + os.WriteFile(filepath.Join(running, "options.json"), []byte(`{"skill":"triage"}`), 0o644) + if info, _ := loadRunInfo(running); info.State != stateRunning { + t.Errorf("state = %s, want running", info.State) + } + + // Orphaned: pid file pointing at a dead process, no result. + orphan := filepath.Join(state.RunsDir(), "20261021-130100-test-2222") + os.MkdirAll(orphan, 0o755) + os.WriteFile(filepath.Join(orphan, "pid"), []byte("999999"), 0o644) + if info, _ := loadRunInfo(orphan); info.State != stateOrphaned { + t.Errorf("state = %s, want orphaned", info.State) + } + + // Budget-stop: result.json with budget_refused. + budget := filepath.Join(state.RunsDir(), "20261021-130200-test-3333") + os.MkdirAll(budget, 0o755) + os.WriteFile(filepath.Join(budget, "result.json"), + []byte(`{"run_id":"x","exit_code":3,"budget_refused":true}`), 0o644) + if info, _ := loadRunInfo(budget); info.State != stateBudget { + t.Errorf("state = %s, want budget-stop", info.State) + } + + // Aborted: result.json with aborted=true. + aborted := filepath.Join(state.RunsDir(), "20261021-130300-test-4444") + os.MkdirAll(aborted, 0o755) + os.WriteFile(filepath.Join(aborted, "result.json"), + []byte(`{"run_id":"x","exit_code":1,"aborted":true}`), 0o644) + if info, _ := loadRunInfo(aborted); info.State != stateAborted { + t.Errorf("state = %s, want aborted", info.State) + } +} + +// TestNewRunIDUniqueness guards the collision-safety of minted ids. +func TestNewRunIDUniqueness(t *testing.T) { + seen := map[string]bool{} + for i := 0; i < 100; i++ { + id, err := newRunID("vuln-scan") + if err != nil { + t.Fatal(err) + } + if seen[id] { + t.Fatalf("duplicate run id %s", id) + } + seen[id] = true + if short := shortRunID(id); short == id || len(short) != 4 { + t.Errorf("short form of %s = %q", id, short) + } + } +} diff --git a/internal/harness/crush_serve.go b/internal/harness/crush_serve.go index 2b4a9df..62ac805 100644 --- a/internal/harness/crush_serve.go +++ b/internal/harness/crush_serve.go @@ -478,12 +478,27 @@ func (s *crushServer) handleSSE(data, sessID, runID string, seenParts map[string } case "tool_call": var p struct { + ID string `json:"id"` Name string `json:"name"` Input string `json:"input"` } if json.Unmarshal(part.Data, &p) == nil && p.Name != "" { emit(Event{Kind: EventToolCall, Tool: p.Name, Text: firstLine(p.Input, 160)}) } + case "tool_result": + // The actual command/scanner output — the bulk of a run's + // output. Truncated to keep the transcript readable; the + // journal keeps what the harness sent. + var p struct { + ToolCallID string `json:"tool_call_id"` + Name string `json:"name"` + Content string `json:"content"` + IsError bool `json:"is_error"` + } + if json.Unmarshal(part.Data, &p) != nil || p.Content == "" { + break + } + emit(Event{Kind: EventToolResult, Tool: p.Name, Text: truncateBlock(p.Content, 6, 400), IsError: p.IsError}) } } seenParts[msg.ID] = len(msg.Parts) @@ -561,6 +576,27 @@ func firstLine(s string, n int) string { return s } +// truncateBlock clips a multi-line tool output to at most maxLines lines +// and maxChars characters, marking the cut. +func truncateBlock(s string, maxLines, maxChars int) string { + s = strings.TrimRight(s, "\n") + lines := strings.Split(s, "\n") + cut := false + if len(lines) > maxLines { + lines = lines[:maxLines] + cut = true + } + out := strings.Join(lines, "\n") + if len(out) > maxChars { + out = out[:maxChars-1] + cut = true + } + if cut { + out += "\n…" + } + return out +} + func randomID() string { buf := make([]byte, 16) if _, err := rand.Read(buf); err != nil { diff --git a/internal/harness/crush_serve_test.go b/internal/harness/crush_serve_test.go index de2103e..c3cb650 100644 --- a/internal/harness/crush_serve_test.go +++ b/internal/harness/crush_serve_test.go @@ -166,11 +166,16 @@ func happyScript(w io.Writer, r *http.Request, sid, runID string) { `{"type":"reasoning","data":{"thinking":"pondering the target","finished_at":1792525268}}`) sse(w, `{"type":"message","payload":{"type":"updated","payload":{"id":"m1","role":"assistant","parts":[%s,%s]}}}`, `{"type":"reasoning","data":{"thinking":"pondering the target","finished_at":1792525268}}`, - `{"type":"tool_call","data":{"name":"bash","input":"{\"command\":\"git clone\"}"}}`) - sse(w, `{"type":"session","payload":{"type":"updated","payload":{"id":"%s","prompt_tokens":100,"completion_tokens":50,"cost":0.42}}}`, sid) + `{"type":"tool_call","data":{"id":"c1","name":"bash","input":"{\"command\":\"git clone\"}"}}`) sse(w, `{"type":"message","payload":{"type":"updated","payload":{"id":"m1","role":"assistant","parts":[%s,%s,%s]}}}`, `{"type":"reasoning","data":{"thinking":"pondering the target","finished_at":1792525268}}`, - `{"type":"tool_call","data":{"name":"bash","input":"{\"command\":\"git clone\"}"}}`, + `{"type":"tool_call","data":{"id":"c1","name":"bash","input":"{\"command\":\"git clone\"}"}}`, + `{"type":"tool_result","data":{"tool_call_id":"c1","name":"bash","content":"Cloning into 'repo'...\ndone\nline3\nline4\nline5\nline6\nline7","is_error":false}}`) + sse(w, `{"type":"session","payload":{"type":"updated","payload":{"id":"%s","prompt_tokens":100,"completion_tokens":50,"cost":0.42}}}`, sid) + sse(w, `{"type":"message","payload":{"type":"updated","payload":{"id":"m1","role":"assistant","parts":[%s,%s,%s,%s]}}}`, + `{"type":"reasoning","data":{"thinking":"pondering the target","finished_at":1792525268}}`, + `{"type":"tool_call","data":{"id":"c1","name":"bash","input":"{\"command\":\"git clone\"}"}}`, + `{"type":"tool_result","data":{"tool_call_id":"c1","name":"bash","content":"Cloning into 'repo'...\ndone\nline3\nline4\nline5\nline6\nline7","is_error":false}}`, `{"type":"text","data":{"text":"audit complete"}}`) sse(w, `{"type":"run_complete","payload":{"type":"updated","payload":{"session_id":"%s","run_id":"%s","message_id":"m1","text":"audit complete"}}}`, sid, runID) } @@ -189,7 +194,7 @@ func TestCrushServeSessionRun(t *testing.T) { t.Fatalf("result = %+v", res) } - var sawText, sawThinking, sawTool, sawUsage bool + var sawText, sawThinking, sawTool, sawUsage, sawToolResult bool for _, ev := range events { switch ev.Kind { case EventText: @@ -198,12 +203,18 @@ func TestCrushServeSessionRun(t *testing.T) { sawThinking = ev.Text == "pondering the target" case EventToolCall: sawTool = ev.Tool == "bash" && strings.Contains(ev.Text, "git clone") + case EventToolResult: + sawToolResult = ev.Tool == "bash" && + strings.Contains(ev.Text, "Cloning into 'repo'...") && + strings.Count(ev.Text, "\n") == 6 && // 6 lines + cut marker + strings.HasSuffix(ev.Text, "…") && + !strings.Contains(ev.Text, "line7") // truncation at 6 lines case EventUsage: sawUsage = ev.TokensIn == 100 && ev.TokensOut == 50 && ev.CostUSD == 0.42 } } - if !sawText || !sawThinking || !sawTool || !sawUsage { - t.Errorf("events incomplete: text=%v thinking=%v tool=%v usage=%v (%d events)", sawText, sawThinking, sawTool, sawUsage, len(events)) + if !sawText || !sawThinking || !sawTool || !sawUsage || !sawToolResult { + t.Errorf("events incomplete: text=%v thinking=%v tool=%v usage=%v result=%v (%d events)", sawText, sawThinking, sawTool, sawUsage, sawToolResult, len(events)) } if len(f.cancelCalls) != 0 { t.Errorf("unexpected cancel calls: %v", f.cancelCalls) diff --git a/internal/harness/harness.go b/internal/harness/harness.go index 76942c7..947d2d9 100644 --- a/internal/harness/harness.go +++ b/internal/harness/harness.go @@ -22,6 +22,8 @@ const ( EventText EventKind = "text" // EventToolCall is a tool invocation by the agent. EventToolCall EventKind = "tool_call" + // EventToolResult is a completed tool invocation's (truncated) output. + EventToolResult EventKind = "tool_result" // EventThinking is a completed reasoning/thinking block. EventThinking EventKind = "thinking" // EventUsage is a cumulative token-usage update. diff --git a/internal/runner/journal.go b/internal/runner/journal.go index 237ca15..9b5b9a1 100644 --- a/internal/runner/journal.go +++ b/internal/runner/journal.go @@ -1,8 +1,11 @@ package runner import ( + "bytes" "encoding/json" + "errors" "fmt" + "io" "os" "path/filepath" "time" @@ -119,4 +122,86 @@ func (j *Journal) write(e journalEntry) { j.stream.Write(append(data, '\n')) } +// Entry is one parsed journal stream record. +type Entry struct { + TS string `json:"ts"` + Kind string `json:"kind"` + Raw string `json:"raw,omitempty"` + Event *Event `json:"event,omitempty"` +} + +// ParseEntry decodes one stream.jsonl line. +func ParseEntry(line []byte) (Entry, error) { + var e Entry + err := json.Unmarshal(line, &e) + return e, err +} + +// Tailer incrementally reads new complete lines from a journal stream file. +// It is the attach mechanism's read side: the worker writes, tailers replay +// and follow. Multiple tailers on the same file are safe (read-only). +type Tailer struct { + path string + offset int64 + file *os.File +} + +// NewTailer opens a stream file for incremental reading, starting at the +// given byte offset (0 to replay from the beginning). +func NewTailer(path string, offset int64) (*Tailer, error) { + f, err := os.Open(path) + if err != nil { + return nil, err + } + if _, err := f.Seek(offset, io.SeekStart); err != nil { + f.Close() + return nil, err + } + return &Tailer{path: path, offset: offset, file: f}, nil +} + +// Offset returns the current read position. +func (t *Tailer) Offset() int64 { return t.offset } + +// Close releases the underlying file. +func (t *Tailer) Close() error { return t.file.Close() } + +// Read returns all complete lines appended since the last call. It never +// blocks: an empty slice means nothing new yet. +func (t *Tailer) Read() ([]string, error) { + var lines []string + buf := make([]byte, 32*1024) + pending := []byte{} + for { + n, err := t.file.Read(buf) + if n > 0 { + pending = append(pending, buf[:n]...) + for { + idx := bytes.IndexByte(pending, '\n') + if idx < 0 { + break + } + lines = append(lines, string(pending[:idx])) + pending = pending[idx+1:] + } + t.offset += int64(n) + } + if err != nil { + if errors.Is(err, io.EOF) { + // Carry partial data into the next call: seek back to the + // last complete line so the fragment is re-read. + if len(pending) > 0 { + t.offset -= int64(len(pending)) + t.file.Seek(t.offset, io.SeekStart) + } + return lines, nil + } + return lines, err + } + if n == 0 { + return lines, nil + } + } +} + func nowRFC() string { return time.Now().UTC().Format(time.RFC3339Nano) } diff --git a/internal/runner/journal_test.go b/internal/runner/journal_test.go new file mode 100644 index 0000000..071b1fe --- /dev/null +++ b/internal/runner/journal_test.go @@ -0,0 +1,95 @@ +package runner + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func TestTailerReplayAndFollow(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "stream.jsonl") + + if err := os.WriteFile(path, []byte("{\"kind\":\"raw\",\"raw\":\"one\"}\n{\"kind\":\"raw\",\"raw\":\"two\"}\n"), 0o644); err != nil { + t.Fatal(err) + } + + tailer, err := NewTailer(path, 0) + if err != nil { + t.Fatal(err) + } + defer tailer.Close() + + lines, err := tailer.Read() + if err != nil || len(lines) != 2 { + t.Fatalf("replay = %v (%v), want 2 lines", lines, err) + } + + // Nothing new. + lines, err = tailer.Read() + if err != nil || len(lines) != 0 { + t.Fatalf("second read = %v (%v), want none", lines, err) + } + + // Appends appear, including a trailing partial line. + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + t.Fatal(err) + } + f.WriteString("{\"kind\":\"raw\",\"raw\":\"three\"}\n{\"kind\":\"raw\",\"raw\":\"par") + f.Close() + + lines, err = tailer.Read() + if err != nil || len(lines) != 1 || lines[0] != `{"kind":"raw","raw":"three"}` { + t.Fatalf("follow = %v (%v)", lines, err) + } + + // Completing the partial line makes it readable. + f, _ = os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644) + f.WriteString("tial\"}\n") + f.Close() + lines, err = tailer.Read() + if err != nil || len(lines) != 1 || lines[0] != `{"kind":"raw","raw":"partial"}` { + t.Fatalf("partial follow = %v (%v)", lines, err) + } +} + +func TestParseEntry(t *testing.T) { + e, err := ParseEntry([]byte(`{"ts":"2026","kind":"event","event":{"kind":"tool_result","tool":"bash","text":"out"}}`)) + if err != nil { + t.Fatal(err) + } + if e.Kind != "event" || e.Event == nil || e.Event.Kind != "tool_result" || e.Event.Tool != "bash" { + t.Errorf("entry = %+v", e) + } + e, err = ParseEntry([]byte(`{"ts":"2026","kind":"raw","raw":"hello"}`)) + if err != nil || e.Raw != "hello" { + t.Errorf("entry = %+v (%v)", e, err) + } + if _, err := ParseEntry([]byte(`not json`)); err == nil { + t.Error("garbage should fail") + } +} + +// TestTailerSpin guards against a busy loop when the writer stalls: Read +// must return promptly with no data. +func TestTailerSpin(t *testing.T) { + path := filepath.Join(t.TempDir(), "s.jsonl") + os.WriteFile(path, nil, 0o644) + tailer, err := NewTailer(path, 0) + if err != nil { + t.Fatal(err) + } + defer tailer.Close() + start := time.Now() + for i := 0; i < 5; i++ { + lines, err := tailer.Read() + if err != nil || len(lines) != 0 { + t.Fatalf("read = %v (%v)", lines, err) + } + } + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("empty reads took %s", elapsed) + } +} diff --git a/internal/runner/runner.go b/internal/runner/runner.go index 3c1a0d2..0886490 100644 --- a/internal/runner/runner.go +++ b/internal/runner/runner.go @@ -71,6 +71,10 @@ type Options struct { BudgetOverrideReason string StrictValidation bool DeclareSpend bool + // RunID, when set, overrides the generated run id — the background + // launcher creates the run directory before spawning the worker, so it + // owns the id. + RunID string } // Usage is the accumulated token usage of a run. @@ -115,7 +119,10 @@ func Run(ctx context.Context, opts Options, sink Sink) (*RunResult, error) { return nil, errors.New("runner: state, api, harness and skill are required") } - runID := time.Now().Format("20060102-150405") + "-" + skill.Name + runID := opts.RunID + if runID == "" { + runID = time.Now().Format("20060102-150405") + "-" + skill.Name + } res := &RunResult{RunID: runID, Skill: skill.Name, Harness: h.ID(), ExitCode: ExitOK} // Full-permission consent: the per-run flag OR the init-time consent @@ -281,14 +288,18 @@ func Run(ctx context.Context, opts Options, sink Sink) (*RunResult, error) { scanner.Buffer(make([]byte, 0, 64*1024), 16*1024*1024) for scanner.Scan() { line := scanner.Text() - journal.Raw(line) if !streaming { + journal.Raw(line) plainOut.WriteString(line) plainOut.WriteString("\n") sink.Raw(line) } else if ev, ok := h.ParseEvent(scanner.Bytes()); ok { handleEvent(ev) } else { + // Unparsed stream lines are journaled raw so attach replays + // show exactly what the harness emitted; parsed lines are + // recorded once, as events. + journal.Raw(line) sink.Raw(line) } } diff --git a/internal/tui/monitor.go b/internal/tui/monitor.go index 1e7bdb8..ebf5a31 100644 --- a/internal/tui/monitor.go +++ b/internal/tui/monitor.go @@ -5,7 +5,6 @@ package tui import ( "fmt" "strings" - "sync" "time" "github.com/charmbracelet/bubbles/viewport" @@ -13,7 +12,6 @@ import ( "github.com/charmbracelet/lipgloss" "github.com/traust-security/traust-cli/internal/harness" - "github.com/traust-security/traust-cli/internal/runner" "github.com/traust-security/traust-cli/internal/theme" ) @@ -23,73 +21,40 @@ type MonitorHeader struct { Harness string Model string BudgetCap int64 + // RunID is shown short in the header (attach context). + RunID string + // StartedAt, when non-zero, anchors the elapsed clock (attach replays + // runs that may have started long before the monitor opened). + StartedAt time.Time } -type monitorItemKind int +// ItemKind classifies a monitor feed item. +type ItemKind int const ( - itemEvent monitorItemKind = iota - itemRaw - itemProgress - itemDone + // ItemEvent is one normalized harness event. + ItemEvent ItemKind = iota + // ItemRaw is one unparsed harness output line. + ItemRaw + // ItemProgress is a run-stage message. + ItemProgress + // ItemDone marks the run as complete. + ItemDone ) -type monitorItem struct { - kind monitorItemKind - ev harness.Event - text string -} - -// monitorSink forwards run updates into the monitor's channel. Sends are -// non-blocking (the journal keeps everything; the UI may drop under -// backpressure) and guarded against a close, so the runner goroutine can -// never panic or deadlock when the program exits before the run does. -type monitorSink struct { - ch chan monitorItem - mu sync.Mutex - closed bool - once sync.Once -} - -func (s *monitorSink) send(it monitorItem) { - s.mu.Lock() - defer s.mu.Unlock() - if s.closed { - return - } - select { - case s.ch <- it: - default: // UI backpressure: drop; the journal keeps everything. - } -} - -func (s *monitorSink) Progress(msg string) { - s.send(monitorItem{kind: itemProgress, text: msg}) -} -func (s *monitorSink) Event(ev harness.Event) { - s.send(monitorItem{kind: itemEvent, ev: ev, text: ev.Text}) -} -func (s *monitorSink) Raw(line string) { - s.send(monitorItem{kind: itemRaw, text: line}) -} -func (s *monitorSink) Done() { s.Close() } - -// Close tears the channel down exactly once, whether the runner finished -// cleanly or the program exited first. -func (s *monitorSink) Close() { - s.once.Do(func() { - s.mu.Lock() - s.closed = true - close(s.ch) - s.mu.Unlock() - }) +// MonitorItem is one unit of monitor input. Live runs and journal replays +// both produce them. +type MonitorItem struct { + Kind ItemKind + Event harness.Event + Text string } type tickMsg time.Time type monitorModel struct { header MonitorHeader - cancel func() + stop func() styles theme.Styles viewport viewport.Model lines []string @@ -98,12 +63,12 @@ type monitorModel struct { band string done bool failed bool - aborting bool - // abortArmed is the ctrl+c confirmation state: the first press asks, - // the second (within armWindow) actually aborts. A single stray ctrl+c - // must not kill a 10-minute audit that merely looks quiet. - abortArmed bool - abortArmedAt time.Time + cancelling bool + // cancelArmed is the x-key confirmation state: the first press asks, + // the second (within armWindow) cancels the run. Cancelling a scan is + // destructive and must never happen on a stray keystroke. + cancelArmed bool + cancelArmedAt time.Time lastOutput time.Time started time.Time width, height int @@ -111,11 +76,15 @@ type monitorModel struct { const armWindow = 6 * time.Second -func newMonitor(header MonitorHeader, cancel func(), width, height int) monitorModel { +func newMonitor(header MonitorHeader, stop func(), width, height int) monitorModel { vp := viewport.New(width-2, height-8) + started := header.StartedAt + if started.IsZero() { + started = time.Now() + } return monitorModel{ - header: header, cancel: cancel, styles: theme.Default(), - viewport: vp, band: "within", started: time.Now(), + header: header, stop: stop, styles: theme.Default(), + viewport: vp, band: "within", started: started, width: width, height: height, } } @@ -127,38 +96,38 @@ func (m monitorModel) Init() tea.Cmd { func (m monitorModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { switch msg := msg.(type) { case tickMsg: - // Disarm the abort confirmation when the window passes. - if m.abortArmed && time.Since(m.abortArmedAt) > armWindow { - m.abortArmed = false + // Disarm the cancel confirmation when the window passes. + if m.cancelArmed && time.Since(m.cancelArmedAt) > armWindow { + m.cancelArmed = false } return m, tea.Tick(time.Second, func(t time.Time) tea.Msg { return tickMsg(t) }) - case monitorItem: + case MonitorItem: m.lastOutput = time.Now() - switch msg.kind { - case itemEvent: - m.applyEvent(msg.ev) - case itemRaw: - m.pushLine(m.styles.Muted.Render(" " + msg.text)) - case itemProgress: + switch msg.Kind { + case ItemEvent: + m.applyEvent(msg.Event) + case ItemRaw: + m.pushLine(m.styles.Muted.Render(" " + msg.Text)) + case ItemProgress: style := m.styles.Accent - if strings.HasPrefix(msg.text, "warning:") || strings.HasPrefix(msg.text, "note:") { + if strings.HasPrefix(msg.Text, "warning:") || strings.HasPrefix(msg.Text, "note:") { style = m.styles.Warn } - if strings.Contains(msg.text, "failed") || strings.Contains(msg.text, "refused") { + if strings.Contains(msg.Text, "failed") || strings.Contains(msg.Text, "refused") { style = m.styles.Err m.failed = true } - if strings.Contains(msg.text, "is not runnable") { + if strings.Contains(msg.Text, "is not runnable") { style = m.styles.Err m.failed = true } - if rest, ok := strings.CutPrefix(msg.text, "budget band "); ok { + if rest, ok := strings.CutPrefix(msg.Text, "budget band "); ok { if fields := strings.Fields(rest); len(fields) > 0 { m.band = fields[0] } } - m.pushLine(style.Render("› " + msg.text)) - case itemDone: + m.pushLine(style.Render("› " + msg.Text)) + case ItemDone: m.done = true m.pushLine(m.styles.Muted.Render("— run complete —")) } @@ -171,31 +140,30 @@ func (m monitorModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { return m, nil case tea.KeyMsg: switch msg.String() { - case "ctrl+c": + case "ctrl+c", "q", "d", "esc": + // Detach: the run keeps going in its own process. This is the + // common case and must be instant and non-destructive. + return m, tea.Quit + case "enter": if m.done { return m, tea.Quit } - if m.aborting { - // Second press while already aborting: leave the view; the - // runner keeps tearing down in the background. + case "x": + if m.done { return m, tea.Quit } - if m.abortArmed { - m.aborting = true - m.pushLine(m.styles.Warn.Render("› aborting — terminating the harness…")) - if m.cancel != nil { - m.cancel() + if m.cancelArmed { + m.cancelling = true + m.pushLine(m.styles.Warn.Render("› cancelling — stopping the run…")) + if m.stop != nil { + m.stop() } return m, nil } - m.abortArmed = true - m.abortArmedAt = time.Now() - m.pushLine(m.styles.Warn.Render("› press ctrl+c again to abort the run")) + m.cancelArmed = true + m.cancelArmedAt = time.Now() + m.pushLine(m.styles.Warn.Render("› press x again to cancel the run (ctrl+c detaches without cancelling)")) return m, nil - case "q", "enter", "esc": - if m.done { - return m, tea.Quit - } } } var cmd tea.Cmd @@ -215,6 +183,19 @@ func (m *monitorModel) applyEvent(ev harness.Event) { line += " " + ev.Text } m.pushLine(m.styles.Accent.Render(line)) + case harness.EventToolResult: + // The tool's actual output, already truncated by the adapter. + if ev.Text == "" { + return + } + style := m.styles.Muted + prefix := " │ " + if ev.IsError { + style = m.styles.Warn + } + for _, l := range strings.Split(ev.Text, "\n") { + m.pushLine(style.Render(prefix + l)) + } case harness.EventUsage: m.tokensIn, m.tokensOut = ev.TokensIn, ev.TokensOut case harness.EventResult: @@ -276,11 +257,19 @@ func (m monitorModel) View() string { } else { status = m.styles.OK.Render("finished") } - } else if m.aborting { - status = m.styles.Warn.Render("aborting…") + } else if m.cancelling { + status = m.styles.Warn.Render("cancelling…") } title := m.styles.App.Render(" traust ") - head := m.styles.Header.Render(fmt.Sprintf(" %s · %s · %s · %s · %s", m.header.Skill, m.header.Harness, ellipsize(m.header.Model, 40), elapsed, status)) + shortID := "" + if m.header.RunID != "" { + parts := strings.Split(m.header.RunID, "-") + last := parts[len(parts)-1] + if len(parts) >= 4 && len(last) == 4 { + shortID = " " + last + " " + } + } + head := m.styles.Header.Render(fmt.Sprintf("%s %s · %s · %s · %s · %s", shortID, m.header.Skill, m.header.Harness, ellipsize(m.header.Model, 40), elapsed, status)) // The header must fit one line at any width: clip after assembly rather // than letting the terminal hard-wrap it into the transcript frame. head = Truncate(head, maxInt(m.width-lipgloss.Width(title), 10), "…") @@ -294,8 +283,7 @@ func (m monitorModel) View() string { b.WriteString("\n") // Status bar: usage + budget bar + output age. The age indicator is the - // answer to "is it stuck?": quiet harnesses (crush run emits one large - // chunk every few minutes) look frozen without it. + // answer to "is it stuck?": quiet harnesses look frozen without it. usage := fmt.Sprintf("in %s · out %s", human(m.tokensIn), human(m.tokensOut)) if m.cost > 0 { usage += fmt.Sprintf(" · $%.2f", m.cost) @@ -316,12 +304,15 @@ func (m monitorModel) View() string { b.WriteString(m.styles.StatusBar.Render(Truncate(usage, maxInt(m.width-2, 20), ""))) b.WriteString("\n") - footer := "ctrl+c×2 abort" - if m.abortArmed && !m.aborting { - footer = "press ctrl+c again to abort" + footer := "ctrl+c detach (run continues) · x×2 cancel" + if m.width < 46 { + footer = "ctrl+c detach · x cancel" + } + if m.cancelArmed && !m.cancelling { + footer = "press x again to cancel · ctrl+c detaches" } - if m.aborting { - footer = "aborting… (ctrl+c again to leave)" + if m.cancelling { + footer = "cancelling…" } if m.done { footer = "q quit · journal written" @@ -358,51 +349,33 @@ func (m monitorModel) budgetBar(out, cap int64, width int) string { return fmt.Sprintf("budget %s/%s [%s]", human(out), human(cap), bar) } -// RunMonitor runs the tea program while runFn executes the pipeline with a -// sink wired to the monitor. cancel is invoked when the user confirms an -// abort. It only returns after runFn has finished (or timed out), so the -// caller can rely on the run result being complete — force-quitting the -// view tears the harness down first. -func RunMonitor(header MonitorHeader, cancel func(), runFn func(sink runner.Sink) error) error { - items := make(chan monitorItem, 128) - sink := &monitorSink{ch: items} +// Attach runs the monitor over a feed of monitor items. It is used both for +// live runs (a producer tailing the worker's journal) and finished runs +// (replay followed by an immediately closed feed). Detaching (ctrl+c, q, d) +// quits the monitor without touching the run; cancelling (x twice) invokes +// stop. The returned bool reports whether the run completed while attached. +func Attach(header MonitorHeader, follow <-chan MonitorItem, stop func()) (bool, error) { + m := newMonitor(header, stop, 80, 24) + prog := tea.NewProgram(m, tea.WithAltScreen(), tea.WithMouseCellMotion()) - prog := tea.NewProgram(newMonitor(header, cancel, 80, 24), - tea.WithAltScreen(), tea.WithMouseCellMotion()) - - runDone := make(chan error, 1) - go func() { - err := runFn(sink) - sink.Close() // terminate the relay even on early error paths - runDone <- err - }() relayDone := make(chan struct{}) go func() { defer close(relayDone) - for it := range items { + for it := range follow { prog.Send(it) } - prog.Send(monitorItem{kind: itemDone}) }() - _, err := prog.Run() - // The program is gone: make sure a still-running harness is torn down, - // then join the runner so the result is complete before returning. - if cancel != nil { - cancel() + out, err := prog.Run() + <-relayDone + if err != nil { + return false, err } - select { - case runErr := <-runDone: - if err != nil { - return err - } - return runErr - case <-time.After(30 * time.Second): - if err != nil { - return err - } - return fmt.Errorf("run did not shut down within 30s of the monitor exiting") + finished := false + if fm, ok := out.(monitorModel); ok { + finished = fm.done } + return finished, nil } func ellipsize(s string, n int) string { diff --git a/testdata/fake-harnesses/fake-claude.sh b/testdata/fake-harnesses/fake-claude.sh index 905f02d..9bb0f6d 100755 --- a/testdata/fake-harnesses/fake-claude.sh +++ b/testdata/fake-harnesses/fake-claude.sh @@ -20,6 +20,11 @@ if [ "${FAKE_CLAUDE_MODE:-}" = "fail" ]; then exit 3 fi +if [ "${FAKE_CLAUDE_MODE:-}" = "slow" ]; then + printf '%s\n' '{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"working"}]}}' + sleep 30 +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}'