From b48467fb6e4a91a1b969323db4f3c477cd0901a4 Mon Sep 17 00:00:00 2001 From: mrhid6 Date: Mon, 20 Jul 2026 12:32:57 +0100 Subject: [PATCH] docs: add workflow log streaming plan --- .../2026-07-20-workflow-log-streaming.md | 887 ++++++++++++++++++ 1 file changed, 887 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-20-workflow-log-streaming.md diff --git a/docs/superpowers/plans/2026-07-20-workflow-log-streaming.md b/docs/superpowers/plans/2026-07-20-workflow-log-streaming.md new file mode 100644 index 0000000..1e00043 --- /dev/null +++ b/docs/superpowers/plans/2026-07-20-workflow-log-streaming.md @@ -0,0 +1,887 @@ +# Workflow Log Streaming Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Stream workflow step output live from agents to per-server-run log files on the server, tail them live in the UI over SSE, and auto-expire them on a configurable retention period. + +**Architecture:** Agent streams interleaved stdout/stderr chunks over the existing `CommandStream` (`AgentMessage.StepOutput`). Server appends secret-masked chunks to `//.log` via a per-command log-writer registry, records a per-step byte offset, and stops persisting log bodies in Mongo. UI tails via an SSE endpoint while running and fetches the whole file after. An hourly sweeper deletes run-log dirs older than the retention setting. + +**Tech Stack:** Go (gin, mongo-driver v2), hand-written JSON-codec gRPC structs (no protoc), Next.js 16 app-router + react-query + EventSource, MongoDB, local filesystem for logs. + +## Global Constraints + +- **No tests this iteration** — do not write `*_test.go` or frontend tests. Verify each task with `go build ./...`, `go vet ./...`, and (frontend) `npm run build`. +- gRPC uses a **JSON codec** — proto messages are hand-written Go structs in **two** files that must stay identical: `server/internal/grpc/pb/vantage.pb.go` and `agent/internal/grpc/pb/vantage.pb.go`. No codegen. Also update `proto/vantage/v1/vantage.proto` as documentation. +- Mongo access pattern: `db.Col("collection_name")` with `context.WithTimeout`. Follow `server/internal/services/workflows.go`. +- Secret values must never be written into log files unmasked — mask by literal `***` replacement at write time, boundary-safe via a carry buffer. +- Interpreter values are the literals `"bash"` and `"powershell"`. +- Go module path: `github.com/mrhid6/vantage`. +- Log dir from env `VANTAGE_WORKFLOW_LOG_DIR`, default `/workflow-logs`; files `0600`, dirs `0700`. +- Retention default **30** days, stored `settings.workflow_log_retention_days`; `0`/negative = keep forever. +- The agent's stream `Send` is only safe through the existing per-connection mutex-guarded `send()` closure in `connectAndHandleStream` — all `StepOutput`/`StepResult` sends MUST go through it. + +--- + +## Task 1: Proto/pb — StepOutputChunk + +**Files:** +- Modify: `proto/vantage/v1/vantage.proto` +- Modify: `server/internal/grpc/pb/vantage.pb.go` +- Modify: `agent/internal/grpc/pb/vantage.pb.go` + +**Interfaces:** +- Produces: `pb.StepOutputChunk{CommandId string, Seq uint64, Data []byte, Eof bool}`; `pb.AgentMessage` gains `StepOutput *StepOutputChunk`. + +- [ ] **Step 1: Document in the proto file** + +In `proto/vantage/v1/vantage.proto`, add to the `AgentMessage` oneof: `StepOutputChunk step_output = 6;` and add the message: + +```protobuf +message StepOutputChunk { + string command_id = 1; + uint64 seq = 2; + bytes data = 3; + bool eof = 4; +} +``` + +- [ ] **Step 2: Add struct + field to server pb file** + +In `server/internal/grpc/pb/vantage.pb.go`, add to `type AgentMessage struct { ... }`: + +```go + StepOutput *StepOutputChunk `json:"step_output,omitempty"` +``` + +and add the new struct: + +```go +type StepOutputChunk struct { + CommandId string `json:"command_id"` + Seq uint64 `json:"seq"` + Data []byte `json:"data,omitempty"` + Eof bool `json:"eof,omitempty"` +} +``` + +- [ ] **Step 3: Mirror identical additions into the agent pb file** + +Apply the identical `AgentMessage.StepOutput` field and `StepOutputChunk` struct to `agent/internal/grpc/pb/vantage.pb.go`. + +- [ ] **Step 4: Verify build** + +Run: `cd server && go build ./... && cd ../agent && go build ./...` +Expected: both succeed. + +- [ ] **Step 5: Commit** + +```bash +git add proto/vantage/v1/vantage.proto server/internal/grpc/pb/vantage.pb.go agent/internal/grpc/pb/vantage.pb.go +git commit -m "feat(proto): add StepOutputChunk streaming message" +``` + +--- + +## Task 2: Agent — stream step output + +**Files:** +- Modify: `agent/internal/exec/exec.go` +- Modify: `agent/internal/sync/sync.go` (the `cmd.RunStep != nil` goroutine) + +**Interfaces:** +- Consumes: `pb.RunStepCmd`, `pb.StepResult`, `pb.StepOutputChunk` (Task 1). +- Produces: `exec.RunStep(cmd *pb.RunStepCmd, emit func(seq uint64, data []byte)) *pb.StepResult` — streams output via `emit`, returns terminal result with empty stdout/stderr but populated exit_code/output_env. + +- [ ] **Step 1: Rework `exec.RunStep` to stream** + +In `agent/internal/exec/exec.go`, change the signature and replace the two `bytes.Buffer`s with a single mutex-guarded streaming writer. Full new body of the run/capture section (keep the existing temp-dir, env-file, interpreter-selection, timeout, and `parseEnvFile` logic exactly as-is): + +Add this type at package scope: + +```go +// streamWriter forwards every write to emit() as an ordered chunk. Used as both +// Stdout and Stderr so output interleaves in real execution order. The mutex +// ensures a single stdout/stderr write is not interleaved mid-slice with another. +type streamWriter struct { + mu sync.Mutex + seq uint64 + emit func(seq uint64, data []byte) +} + +func (w *streamWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + if w.emit != nil { + buf := make([]byte, len(p)) + copy(buf, p) + w.emit(w.seq, buf) + w.seq++ + } + return len(p), nil +} +``` + +Add `"sync"` to the imports. Change the signature to: + +```go +func RunStep(cmd *pb.RunStepCmd, emit func(seq uint64, data []byte)) *pb.StepResult { +``` + +Replace the block that currently declares `var stdout, stderr bytes.Buffer`, assigns `c.Stdout`/`c.Stderr`, and sets `res.Stdout`/`res.Stderr` from them, with: + +```go + sw := &streamWriter{emit: emit} + c.Stdout = sw + c.Stderr = sw + runErr := c.Run() + + // stdout/stderr are streamed via emit, not returned in the result. + if ctx.Err() == context.DeadlineExceeded { + res.ExitCode = 124 + res.Stderr = "[vantage] step timed out" + } else if ee, ok := runErr.(*exec.ExitError); ok { + res.ExitCode = ee.ExitCode() + } else if runErr != nil { + res.ExitCode = 1 + res.Stderr = "[vantage] " + runErr.Error() + } + + res.OutputEnv = parseEnvFile(envFile) + return res +``` + +Remove the now-unused `"bytes"` and `"bufio"` imports **only if** they are no longer referenced (`parseEnvFile` uses `bufio` + `os` — keep `bufio`; `bytes` is likely now unused — remove it if so). Verify with `go build`. + +- [ ] **Step 2: Wire streaming into the agent loop** + +In `agent/internal/sync/sync.go`, the `cmd.RunStep != nil` goroutine currently calls `agentexec.RunStep(rc)` and sends one `StepResult` via `send()`. Change it to pass an `emit` closure that streams chunks, then send an eof chunk, then the terminal result — all through the existing mutex-guarded `send()`: + +```go + if cmd.RunStep != nil { + go func(rc *pb.RunStepCmd, cid string) { + emit := func(seq uint64, data []byte) { + _ = send(&pb.AgentMessage{ + ServerId: cfg.ServerID, + AgentToken: cfg.AgentToken, + StepOutput: &pb.StepOutputChunk{CommandId: cid, Seq: seq, Data: data}, + }) + } + res := agentexec.RunStep(rc, emit) + res.CommandId = cid + // Final eof marker so the server closes the log file. + _ = send(&pb.AgentMessage{ + ServerId: cfg.ServerID, + AgentToken: cfg.AgentToken, + StepOutput: &pb.StepOutputChunk{CommandId: cid, Eof: true}, + }) + _ = send(&pb.AgentMessage{ + ServerId: cfg.ServerID, + AgentToken: cfg.AgentToken, + StepResult: res, + }) + }(cmd.RunStep, cmd.CommandId) + continue + } +``` + +(Match the exact field names already used by the existing `send()` calls in this function — `cfg.ServerID`, `cfg.AgentToken`, and the `send` closure. If the existing RunStep branch used different local names, keep those.) + +- [ ] **Step 3: Verify build** + +Run: `cd agent && go build ./... && go vet ./...` +Expected: success. Resolve any leftover unused-import error from Step 1. + +- [ ] **Step 4: Commit** + +```bash +git add agent/internal/exec/exec.go agent/internal/sync/sync.go +git commit -m "feat(agent): stream step output chunks over CommandStream" +``` + +--- + +## Task 3: Server log-writer registry + retention sweeper + +**Files:** +- Create: `server/internal/services/steplogs.go` + +**Interfaces:** +- Consumes: `settings` service (retention), `db.Col("workflow_runs")` (sweeper), env `VANTAGE_WORKFLOW_LOG_DIR`. +- Produces: + - `WorkflowLogDir() string` — resolved base dir (env or default), created on first call. + - `ServerRunLogPath(runID, serverID string) string` — `//.log`. + - `AppendMarker(runID, serverID, line string) (int64, error)` — appends a marker line, returns the byte offset **before** the write (the step's `log_offset`). + - `var StepLogs *stepLogRegistry` with `Open(commandID, path string, secrets []string) error`, `Append(commandID string, data []byte)`, `Close(commandID string)`. + - `StartLogSweeper()` — launches the hourly retention goroutine; also sweeps once immediately. + +- [ ] **Step 1: Write the registry, paths, and sweeper** + +```go +package services + +import ( + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/mrhid6/vantage/server/internal/db" + "go.mongodb.org/mongo-driver/v2/bson" +) + +// WorkflowLogDir returns the base directory for workflow step logs, creating it. +func WorkflowLogDir() string { + dir := os.Getenv("VANTAGE_WORKFLOW_LOG_DIR") + if dir == "" { + dir = filepath.Join("data", "workflow-logs") + } + _ = os.MkdirAll(dir, 0700) + return dir +} + +// ServerRunLogPath is the per-server-run log file path. +func ServerRunLogPath(runID, serverID string) string { + return filepath.Join(WorkflowLogDir(), runID, serverID+".log") +} + +// AppendMarker appends a line to the server-run log and returns the byte offset +// at which the write began (used as a step's log_offset). +func AppendMarker(runID, serverID, line string) (int64, error) { + path := ServerRunLogPath(runID, serverID) + if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { + return 0, err + } + f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600) + if err != nil { + return 0, err + } + defer f.Close() + off, _ := f.Seek(0, 2) // current end = offset before write + if _, err := f.WriteString(line); err != nil { + return off, err + } + return off, nil +} + +// ---- streamed chunk writer, boundary-safe secret masking ---- + +type stepLogWriter struct { + mu sync.Mutex + f *os.File + carry []byte + secrets []string + maxSecret int +} + +type stepLogRegistry struct { + mu sync.Mutex + writers map[string]*stepLogWriter +} + +var StepLogs = &stepLogRegistry{writers: make(map[string]*stepLogWriter)} + +// Open opens (append) the server-run file for a step's streamed chunks. +func (r *stepLogRegistry) Open(commandID, path string, secrets []string) error { + if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { + return err + } + f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600) + if err != nil { + return err + } + max := 0 + for _, s := range secrets { + if len(s) > max { + max = len(s) + } + } + w := &stepLogWriter{f: f, secrets: secrets, maxSecret: max} + r.mu.Lock() + r.writers[commandID] = w + r.mu.Unlock() + return nil +} + +func (r *stepLogRegistry) get(commandID string) *stepLogWriter { + r.mu.Lock() + defer r.mu.Unlock() + return r.writers[commandID] +} + +// Append masks and writes a chunk, holding back the last maxSecret-1 bytes so a +// secret split across a chunk boundary is still masked on the next append/close. +func (r *stepLogRegistry) Append(commandID string, data []byte) { + w := r.get(commandID) + if w == nil { + return + } + w.mu.Lock() + defer w.mu.Unlock() + if len(w.secrets) == 0 || w.maxSecret <= 1 { + _, _ = w.f.Write(data) + return + } + buf := append(w.carry, data...) + hold := w.maxSecret - 1 + if len(buf) <= hold { + w.carry = buf + return + } + flush := buf[:len(buf)-hold] + w.carry = append([]byte{}, buf[len(buf)-hold:]...) + _, _ = w.f.Write(maskBytes(flush, w.secrets)) +} + +// Close flushes the carry (masked) and closes the file. +func (r *stepLogRegistry) Close(commandID string) { + r.mu.Lock() + w := r.writers[commandID] + delete(r.writers, commandID) + r.mu.Unlock() + if w == nil { + return + } + w.mu.Lock() + defer w.mu.Unlock() + if len(w.carry) > 0 { + _, _ = w.f.Write(maskBytes(w.carry, w.secrets)) + w.carry = nil + } + _ = w.f.Close() +} + +func maskBytes(b []byte, secrets []string) []byte { + s := string(b) + for _, v := range secrets { + if v == "" { + continue + } + s = strings.ReplaceAll(s, v, "***") + } + return []byte(s) +} + +// ---- retention sweeper ---- + +// StartLogSweeper sweeps expired run-log dirs hourly (and once now). +func StartLogSweeper() { + go func() { + sweepLogs() + t := time.NewTicker(time.Hour) + defer t.Stop() + for range t.C { + sweepLogs() + } + }() +} + +func sweepLogs() { + days := retentionDays() + if days <= 0 { + return + } + cutoff := time.Now().AddDate(0, 0, -days) + base := WorkflowLogDir() + entries, err := os.ReadDir(base) + if err != nil { + return + } + for _, e := range entries { + if !e.IsDir() { + continue + } + runID := e.Name() + dir := filepath.Join(base, runID) + if runExpired(runID, dir, cutoff) { + _ = os.RemoveAll(dir) + } + } +} + +// runExpired is true when the run finished before cutoff (falling back to dir +// mtime when the run doc is gone). +func runExpired(runID, dir string, cutoff time.Time) bool { + ctx, cancel := wfCtx() + defer cancel() + var run struct { + FinishedAt *time.Time `bson:"finished_at"` + } + err := db.Col("workflow_runs").FindOne(ctx, bson.M{"run_id": runID}).Decode(&run) + if err == nil { + if run.FinishedAt == nil { + return false // still running / never finished — keep + } + return run.FinishedAt.Before(cutoff) + } + // run doc gone: use dir mtime + if fi, e := os.Stat(dir); e == nil { + return fi.ModTime().Before(cutoff) + } + return false +} + +func retentionDays() int { + if v, err := GetWorkflowLogRetentionDays(); err == nil { + return v + } + return 30 +} +``` + +Note: `wfCtx` is defined in `workflows.go` (same package) — reuse it. `GetWorkflowLogRetentionDays` is added in Task 4; this file references it (same package, compiles together). + +- [ ] **Step 2: Verify build** + +Run: `cd server && go build ./... && go vet ./...` +Expected: FAIL — `GetWorkflowLogRetentionDays` undefined until Task 4. This is expected; proceed to commit the file so Task 4 completes it. (If you prefer a green build, do Task 4's settings accessor first, then return — but committing here is fine since Task 4 immediately follows.) + +Actually to keep every commit buildable: **temporarily** add a local stub at the bottom of this file and remove it in Task 4: + +```go +// TEMP stub, replaced in Task 4. +func GetWorkflowLogRetentionDays() (int, error) { return 30, nil } +``` + +Then `cd server && go build ./... && go vet ./...` must succeed. + +- [ ] **Step 3: Commit** + +```bash +git add server/internal/services/steplogs.go +git commit -m "feat(server): workflow log-writer registry, paths, retention sweeper" +``` + +--- + +## Task 4: Settings — retention accessor + startup wiring + +**Files:** +- Modify: `server/internal/services/settings.go` (or wherever settings get/set lives — search `settings` collection usage) +- Modify: `server/internal/services/steplogs.go` (remove the temp stub) +- Modify: `server/cmd/main.go` (start the sweeper) + +**Interfaces:** +- Produces: `GetWorkflowLogRetentionDays() (int, error)` (default 30 when unset), `SetWorkflowLogRetentionDays(int) error`. If settings are exposed as a single document/struct, add the field there and derive these accessors. + +- [ ] **Step 1: Inspect the settings service** + +Read the existing settings service (search for the `settings` collection: `grep -rn "\"settings\"" server/internal/services`). Determine whether settings are a typed struct document or key/value. Match that pattern. + +- [ ] **Step 2: Add the retention accessor** + +If settings are a **typed document** (e.g. a `GetSettings()/UpdateSettings()`), add a field `WorkflowLogRetentionDays int `bson:"workflow_log_retention_days" json:"workflow_log_retention_days"`` to the settings struct and implement: + +```go +func GetWorkflowLogRetentionDays() (int, error) { + s, err := GetSettings() // use the real accessor name + if err != nil { + return 30, err + } + if s.WorkflowLogRetentionDays == 0 && /* unset sentinel */ !s.WorkflowLogRetentionSet { + return 30, nil + } + return s.WorkflowLogRetentionDays, nil +} +``` + +Simplify to match reality: if the settings doc uses zero-value-means-unset and you cannot distinguish "0 = keep forever" from "unset", store the retention as a pointer `*int` or default at read: **treat a missing field as 30, an explicit 0 as keep-forever.** Prefer `*int` in the struct so the three states (unset→30, 0→forever, N→N) are representable. Implement `GetWorkflowLogRetentionDays` to return 30 when the pointer is nil, else its value. `SetWorkflowLogRetentionDays(n int)` sets the pointer. + +If settings are **key/value**, implement both accessors against that store with the same nil→30 / 0→forever semantics (store empty/absent = 30). + +- [ ] **Step 3: Remove the temp stub from `steplogs.go`** + +Delete the `// TEMP stub` `GetWorkflowLogRetentionDays` added in Task 3 so the real one is used. + +- [ ] **Step 4: Start the sweeper at boot** + +In `server/cmd/main.go`, next to `EnsureWorkflowIndexes()`, add `services.StartLogSweeper()`. + +- [ ] **Step 5: Verify build** + +Run: `cd server && go build ./... && go vet ./...` +Expected: success (real accessor now resolves the reference from Task 3). + +- [ ] **Step 6: Commit** + +```bash +git add server/internal/services/settings.go server/internal/services/steplogs.go server/cmd/main.go +git commit -m "feat(server): workflow log retention setting + sweeper startup" +``` + +--- + +## Task 5: Runner + model — write to files, drop log bodies from Mongo + +**Files:** +- Modify: `server/internal/models/workflow.go` (`StepRun`) +- Modify: `server/internal/services/workflow_runner.go` +- Modify: `server/internal/grpc/server.go` (stream delivery of `StepOutput`) + +**Interfaces:** +- Consumes: `StepLogs`, `AppendMarker`, `ServerRunLogPath` (Task 3), `pb.StepOutputChunk` (Task 1). +- Produces: runner writes markers + streams chunks to files; `StepRun.LogOffset` persisted; `StepRun.Stdout/Stderr` removed. + +- [ ] **Step 1: Update the `StepRun` model** + +In `server/internal/models/workflow.go`, in `type StepRun struct`: +- Remove the `Stdout` and `Stderr` fields. +- Add: `LogOffset int64 `bson:"log_offset" json:"log_offset"`` + +- [ ] **Step 2: Deliver StepOutput chunks in the gRPC receive loop** + +In `server/internal/grpc/server.go`, after the existing `if m.StepResult != nil { services.StepResults.Deliver(m.StepResult) }` block, add: + +```go + if m.StepOutput != nil { + if m.StepOutput.Eof { + services.StepLogs.Close(m.StepOutput.CommandId) + } else { + services.StepLogs.Append(m.StepOutput.CommandId, m.StepOutput.Data) + } + } +``` + +- [ ] **Step 3: Rework `runServer` to open logs + write markers, drop persisted bodies** + +In `server/internal/services/workflow_runner.go`, `runServer`: + +Inside the per-step loop, **before** `dispatchAndWait`, add marker + open (compute `secretVals` first, which already exists in the loop): + +```go + // Write the step marker and remember the offset for later slicing. + marker := fmt.Sprintf("\n===== step %d: %s =====\n", step.Order, step.Name) + offset, _ := AppendMarker(runID, serverID, marker) + logPath := ServerRunLogPath(runID, serverID) + _ = StepLogs.Open(commandID_placeholder, logPath, secretsSlice(secretVals)) +``` + +There is a chicken-and-egg with `commandID`: today `dispatchAndWait` generates the `commandID` internally. Refactor so the runner owns the `commandID`: + +1. Change `dispatchAndWait(serverID string, cmd *pb.RunStepCmd)` to `dispatchAndWait(serverID, commandID string, cmd *pb.RunStepCmd)` and remove its internal `commandID := uuid.New().String()` (use the passed one). +2. In `runServer`, generate `commandID := uuid.New().String()` at the top of each attempt-group (before the marker/open), open the log with it, then call `dispatchAndWait(serverID, commandID, cmd)`. +3. After the step completes (result received), call `StepLogs.Close(commandID)` defensively (idempotent — the agent's eof usually closed it already; Close on a missing key is a no-op). + +Add a helper to convert the `secretVals map[string]string` to a `[]string` of values: + +```go +func secretsSlice(m map[string]string) []string { + out := make([]string, 0, len(m)) + for _, v := range m { + out = append(out, v) + } + return out +} +``` + +Update `finishStep(...)` call + signature: **remove** the `stdout, stderr string` params and the `output_env` masking stays. Persist `log_offset` instead. New `finishStep`: + +```go +func finishStep(runID, serverID string, order int, status string, attempts, exit int, logOffset int64, outEnv map[string]string) { + now := time.Now() + updateStep(runID, serverID, order, bson.M{ + "server_runs.$[s].steps.$[t].status": status, + "server_runs.$[s].steps.$[t].attempts": attempts, + "server_runs.$[s].steps.$[t].exit_code": exit, + "server_runs.$[s].steps.$[t].log_offset": logOffset, + "server_runs.$[s].steps.$[t].output_env": outEnv, + "server_runs.$[s].steps.$[t].finished_at": now, + }) +} +``` + +In the loop, after receiving `res`, drop the `stdout, stderr := ...` masking of `res.Stdout/res.Stderr` (those are now streamed to file). Keep the `outEnv` build **with existing masking** (`maskSecrets(v, allSecrets)` per the merged secret-leak fix) — `output_env`/`run_env` masking is unchanged. Call: + +```go + finishStep(runID, serverID, i, status, attempts, exit, offset, outEnv) +``` + +where `offset` is the marker offset captured before dispatch. If `res == nil`, still write a short note to the file so failures are visible: + +```go + if res == nil { + _, _ = AppendMarker(runID, serverID, "[vantage] agent did not return a result\n") + } +``` + +Remove the initial `StepRun{... Status:"queued"}` `Stdout/Stderr` references if any (the model no longer has them — the queued StepRun in `TriggerWorkflow` set only `Order/Name/Status/OutputEnv`, so no change needed there; verify). + +Ensure `fmt` is imported (it already is). + +- [ ] **Step 4: Verify build** + +Run: `cd server && go build ./... && go vet ./...` +Expected: success. Fix any remaining references to the removed `Stdout`/`Stderr` fields or the old `finishStep`/`dispatchAndWait` signatures. + +- [ ] **Step 5: Commit** + +```bash +git add server/internal/models/workflow.go server/internal/services/workflow_runner.go server/internal/grpc/server.go +git commit -m "feat(server): stream step logs to files, drop log bodies from run docs" +``` + +--- + +## Task 6: REST — log fetch + SSE stream endpoints + +**Files:** +- Modify: `server/internal/api/workflows.go` + +**Interfaces:** +- Consumes: `ServerRunLogPath`, `GetRun` (existing). +- Produces: `GET /api/runs/:runId/servers/:serverId/logs` and `GET /api/runs/:runId/servers/:serverId/logs/stream` (SSE). + +- [ ] **Step 1: Add the two handlers + routes** + +In `registerWorkflowRoutes`, add: + +```go + g.GET("/runs/:runId/servers/:serverId/logs", getServerRunLog) + g.GET("/runs/:runId/servers/:serverId/logs/stream", streamServerRunLog) +``` + +Add a UUID-ish validator and the handlers: + +```go +var uuidLike = regexp.MustCompile(`^[a-zA-Z0-9-]{1,64}$`) + +func getServerRunLog(c *gin.Context) { + runID, serverID := c.Param("runId"), c.Param("serverId") + if !uuidLike.MatchString(runID) || !uuidLike.MatchString(serverID) { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid id"}) + return + } + path := services.ServerRunLogPath(runID, serverID) + b, err := os.ReadFile(path) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "no logs"}) + return + } + c.Data(http.StatusOK, "text/plain; charset=utf-8", b) +} + +func streamServerRunLog(c *gin.Context) { + runID, serverID := c.Param("runId"), c.Param("serverId") + if !uuidLike.MatchString(runID) || !uuidLike.MatchString(serverID) { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid id"}) + return + } + path := services.ServerRunLogPath(runID, serverID) + + c.Writer.Header().Set("Content-Type", "text/event-stream") + c.Writer.Header().Set("Cache-Control", "no-cache") + c.Writer.Header().Set("Connection", "keep-alive") + c.Writer.Header().Set("X-Accel-Buffering", "no") + + flusher, ok := c.Writer.(http.Flusher) + if !ok { + c.JSON(http.StatusInternalServerError, gin.H{"error": "stream unsupported"}) + return + } + + var offset int64 + sendNew := func() bool { + f, err := os.Open(path) + if err != nil { + return true // file may not exist yet; keep waiting + } + defer f.Close() + if _, err := f.Seek(offset, 0); err != nil { + return true + } + buf := make([]byte, 32*1024) + for { + n, _ := f.Read(buf) + if n <= 0 { + break + } + offset += int64(n) + // SSE data frame; split on newlines to keep frames well-formed. + for _, line := range splitSSE(buf[:n]) { + _, _ = c.Writer.WriteString("data: " + line + "\n") + } + _, _ = c.Writer.WriteString("\n") + flusher.Flush() + } + return true + } + + ctx := c.Request.Context() + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + for { + sendNew() + if serverRunTerminal(runID, serverID) { + sendNew() // final drain + _, _ = c.Writer.WriteString("event: done\ndata: end\n\n") + flusher.Flush() + return + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +// serverRunTerminal reports whether the given server-run has reached a terminal status. +func serverRunTerminal(runID, serverID string) bool { + r, err := services.GetRun(runID) + if err != nil { + return true + } + for _, sr := range r.ServerRuns { + if sr.ServerID == serverID { + switch sr.Status { + case "success", "failed", "skipped", "cancelled": + return true + } + return false + } + } + return true +} + +// splitSSE turns a raw byte slice into SSE-safe payload lines (newlines become +// separate data lines; carriage returns stripped). +func splitSSE(b []byte) []string { + s := strings.ReplaceAll(string(b), "\r", "") + return strings.Split(s, "\n") +} +``` + +Add imports: `"os"`, `"regexp"`, `"strings"`, `"time"`, `"net/http"` (already present). Confirm `services.GetRun` and `ServerRun.Status`/`ServerID` fields exist (they do from the Workflows feature). + +- [ ] **Step 2: Verify build** + +Run: `cd server && go build ./... && go vet ./...` +Expected: success. + +- [ ] **Step 3: Commit** + +```bash +git add server/internal/api/workflows.go +git commit -m "feat(api): server-run log fetch and SSE stream endpoints" +``` + +--- + +## Task 7: Frontend — live SSE tail + retention setting + +**Files:** +- Modify: `web/lib/api.ts` +- Modify: `web/app/workflows/[id]/runs/[runId]/page.tsx` +- Modify: `web/app/settings/page.tsx` + +**Interfaces:** +- Consumes: SSE endpoint, logs endpoint, settings mutation. + +- [ ] **Step 1: Update API types + helpers** + +In `web/lib/api.ts`: +- In `StepRun`, remove `stdout` and `stderr`; add `log_offset: number`. +- Add: `getServerRunLog: (runId: string, serverId: string) => request(...)` — but the logs endpoint returns `text/plain`, so add a dedicated fetch that reads text. If `request` assumes JSON, add a sibling: + +```ts + async getServerRunLog(runId: string, serverId: string): Promise { + const res = await fetch(`${API_BASE}/api/runs/${runId}/servers/${serverId}/logs`, { credentials: "include" }); + if (!res.ok) throw new Error("no logs"); + return res.text(); + }, +``` + +(Use the file's real base-URL constant / credentials pattern — inspect how `request` builds URLs and mirror it. If the app is same-origin with a rewrite, a relative `/api/...` fetch is fine.) +- Export a helper to build the SSE URL: `serverRunLogStreamUrl(runId, serverId)` returning the `/api/runs/:runId/servers/:serverId/logs/stream` URL against the same base. +- In the Settings type, add `workflow_log_retention_days?: number | null`. + +- [ ] **Step 2: Live tail in the run detail page** + +In `web/app/workflows/[id]/runs/[runId]/page.tsx`: +- Remove all use of `st.stdout` / `st.stderr` (fields gone). Step `
` now show status/exit/attempts pills only. +- Add a per-server live terminal. For each `server_run`, render a `
` and, while `sr.status === "running"`, subscribe via `EventSource`:
+
+```tsx
+function ServerLog({ runId, serverId, status }: { runId: string; serverId: string; status: string }) {
+  const [text, setText] = useState("");
+  const preRef = useRef(null);
+  const running = status === "running";
+
+  useEffect(() => {
+    if (running) {
+      const es = new EventSource(api.serverRunLogStreamUrl(runId, serverId), { withCredentials: true });
+      es.onmessage = (e) => setText((t) => t + e.data + "\n");
+      es.addEventListener("done", () => es.close());
+      es.onerror = () => es.close();
+      return () => es.close();
+    }
+    // terminal: fetch the whole file once
+    api.getServerRunLog(runId, serverId).then(setText).catch(() => setText(""));
+  }, [running, runId, serverId]);
+
+  useEffect(() => { preRef.current?.scrollTo(0, preRef.current.scrollHeight); }, [text]);
+
+  return (
+    
+      {text || (running ? "Waiting for output…" : "No output.")}
+    

+  );
+}
+```
+
+Render `` inside each server card, below the step pills. Keep the existing react-query `refetchInterval` on the run (drives status pills); the SSE handles live text.
+
+- [ ] **Step 3: Retention field in Settings**
+
+In `web/app/settings/page.tsx`, add a "Workflow log retention (days)" number input bound to `workflow_log_retention_days`, saved through the existing settings save mutation. Add helper text: "0 = keep forever." Match the page's existing input styling.
+
+- [ ] **Step 4: Verify build**
+
+Run: `cd web && npm run build`
+Expected: type-checks and builds. Fix any lingering `st.stdout`/`st.stderr` references.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add web/lib/api.ts web/app/workflows/[id]/runs/[runId]/page.tsx web/app/settings/page.tsx
+git commit -m "feat(web): live SSE log tail and log retention setting"
+```
+
+---
+
+## Task 8: End-to-end verification
+
+**Files:** none (verification only).
+
+- [ ] **Step 1: Build everything**
+
+Run: `cd server && go build ./... && go vet ./... && cd ../agent && go build ./... && go vet ./... && cd ../web && npm run build`
+Expected: all succeed.
+
+- [ ] **Step 2: Manual smoke (documented, run if an environment is available)**
+
+With server + MongoDB + a connected agent:
+1. Run a workflow with a step that emits output slowly (e.g. `for i in $(seq 1 10); do echo "line $i"; sleep 1; done`). Open the run detail page while running; confirm lines appear live (SSE), not only at the end.
+2. Confirm `//.log` exists on the server with step markers and the output.
+3. Confirm `workflow_runs` doc no longer stores stdout/stderr bodies; `steps[].log_offset` is set.
+4. Add a secret ref and echo it; confirm the file shows `***`, including when the secret would straddle a chunk boundary.
+5. Set retention to 0 in Settings → confirm sweeper keeps files; set to a small value and backdate a run's `finished_at` → confirm the dir is removed within the hour (or call `sweepLogs` path manually).
+
+- [ ] **Step 3: Commit any fixes found**
+
+```bash
+git add -A
+git commit -m "fix: workflow log streaming e2e fixes"
+```
+
+---
+
+## Self-Review Notes
+
+- **Spec coverage:** §3 proto → T1; §4 agent streaming → T2; §5.1 registry + §7 sweeper → T3; §7.1 setting + startup → T4; §5.3/§5.4 runner+model → T5; §6 REST/SSE → T6; §8 frontend → T7. Tests omitted per Global Constraints.
+- **Masking** boundary-safe carry buffer in `StepLogs.Append`, flushed in `Close` (T3); `output_env`/`run_env` masking unchanged (T5 keeps the merged fix).
+- **commandID ownership** moved to the runner so the log file can be opened before dispatch (T5) — mirrors the `StepResults.Await`-before-dispatch ordering.
+- **Buildable commits:** T3 adds a temp stub for `GetWorkflowLogRetentionDays`, removed in T4.
+- **Removed fields** `StepRun.Stdout/Stderr` — every reader updated in T5 (runner) and T7 (frontend).
+- **Open follow-ups (out of scope):** per-step SSE channels, log download/zip, compression, pre-existing runs have no files.
+```