diff --git a/cmd/gitmote/main.go b/cmd/gitmote/main.go
index e617478..42817ad 100644
--- a/cmd/gitmote/main.go
+++ b/cmd/gitmote/main.go
@@ -450,7 +450,10 @@ func buildGitHandler(ctx context.Context, logger *slog.Logger) (http.Handler, *w
// The runner's authenticated claim/complete API. Only the leader may write
// completions (a follower returns a retryable 503), and it authenticates with
// the same WORKER_SECRET injected into the runner env.
- reportAPI := ci.NewReportAPI(md, objs, md.IsLeader, workerSecret, logger)
+ // The in-memory live-log store the runner streams into and the UI tails —
+ // leader-side, ephemeral, shared by the report API (writer) and the UI (reader).
+ liveLogs := ci.NewLiveLogs()
+ reportAPI := ci.NewReportAPI(md, objs, liveLogs, md.IsLeader, workerSecret, logger)
writer.AfterCommit = func(ctx context.Context, pusherID int64, commits []githttp.CommitInfo) {
for _, c := range commits {
dispatcher.Dispatch(ctx, ci.Event{
@@ -489,7 +492,7 @@ func buildGitHandler(ctx context.Context, logger *slog.Logger) (http.Handler, *w
return nil, nil, nil, nil, noop, err
}
- ui, err := buildUI(ctx, md, materializer, objs, guard, secretsSvc, logger)
+ ui, err := buildUI(ctx, md, materializer, objs, liveLogs, guard, secretsSvc, logger)
if err != nil {
_ = cleanup()
return nil, nil, nil, nil, noop, err
@@ -501,12 +504,12 @@ func buildGitHandler(ctx context.Context, logger *slog.Logger) (http.Handler, *w
// GITMOTE_COOKIE_KEY when set, else an auto-generated key persisted in meta and
// restored across restart / scale-to-zero — so a session stays valid through an
// idle→wake and the UI needs no secret config.
-func buildUI(ctx context.Context, md *meta.Metadata, mz *repo.Materializer, objs store.Store, guard *auth.Guard, secretsSvc *secrets.Service, logger *slog.Logger) (*webui.Handler, error) {
+func buildUI(ctx context.Context, md *meta.Metadata, mz *repo.Materializer, objs store.Store, live webui.LiveLog, guard *auth.Guard, secretsSvc *secrets.Service, logger *slog.Logger) (*webui.Handler, error) {
cookieKey, err := resolveCookieKey(ctx, md)
if err != nil {
return nil, fmt.Errorf("cookie key: %w", err)
}
- return webui.New(md, mz, objs, guard, secretsSvc, cookieKey, logger)
+ return webui.New(md, mz, objs, live, guard, secretsSvc, cookieKey, logger)
}
// resolveCookieKey returns the session cookie key: GITMOTE_COOKIE_KEY when set,
diff --git a/cmd/gitmote/main_test.go b/cmd/gitmote/main_test.go
index 8947dd5..b595958 100644
--- a/cmd/gitmote/main_test.go
+++ b/cmd/gitmote/main_test.go
@@ -110,7 +110,7 @@ func TestUIRoutesDoNotShadowGitEndpoints(t *testing.T) {
}
t.Cleanup(func() { _ = m.Close() })
mz := repo.New(m, store.NewMem(), t.TempDir())
- ui, err := webui.New(m, mz, store.NewMem(), auth.NewGuard(m), nil, []byte("k"), slog.New(slog.DiscardHandler))
+ ui, err := webui.New(m, mz, store.NewMem(), nil, auth.NewGuard(m), nil, []byte("k"), slog.New(slog.DiscardHandler))
if err != nil {
t.Fatalf("webui.New: %v", err)
}
@@ -486,7 +486,7 @@ func TestResolvedWorkerSecretReachesDispatcherAndReportAPI(t *testing.T) {
// The report API validates that same secret — closing the loop the runner
// rides: injected at trigger, checked on report-back.
- api := ci.NewReportAPI(md, s, nil, secret, nil)
+ api := ci.NewReportAPI(md, s, nil, nil, secret, nil)
mux := http.NewServeMux()
api.Register(mux)
srv := httptest.NewServer(mux)
diff --git a/docs/architecture/ci.md b/docs/architecture/ci.md
index 2955709..bab2919 100644
--- a/docs/architecture/ci.md
+++ b/docs/architecture/ci.md
@@ -80,10 +80,28 @@ recorded, with a size cap (explicit truncation marker, never silent).
The runner-facing report API lives under `/internal`, apart from the browser UI
([`internal/ci/report.go`](../../internal/ci/report.go)):
`GET /internal/ci/jobs/{id}` claims a job (`queued→running`),
-`POST …/complete` uploads the log and sets a terminal status (idempotent). Both
-do a constant-time `WORKER_SECRET` compare; only the **leader** writes
-completions (a follower returns a retryable 503). A leader-only reconcile ticker
-sweeps jobs stuck in `running` to `error`.
+`POST …/log` appends a live-log chunk (below), `POST …/complete` uploads the log
+and sets a terminal status (idempotent). All do a constant-time `WORKER_SECRET`
+compare; only the **leader** writes completions (a follower returns a retryable
+503). A leader-only reconcile ticker sweeps jobs stuck in `running` to `error`.
+
+## Live logs
+
+The durable log above is shipped once, at completion — so while a job runs there
+is otherwise nothing to watch. A **best-effort live tail** fills that gap without
+touching the durable path. The runner tees `act`'s combined output and, on a ~1 s
+ticker, POSTs the delta to `…/log`; the leader appends it to an **in-memory**,
+per-job buffer ([`LiveLogs`](../../internal/ci/livelogs.go)) capped like the
+durable blob. The UI's job-log page polls
+`GET /{repo}/runs/{id}/job/{jid}/log/live?offset=N` (`{bytes, next, done}`) and,
+on `done`, reloads to the rendered durable log.
+
+It is deliberately **never replicated or persisted**: a lost chunk or a leader
+restart only blips the live view — the runner still ships the whole log at
+completion, so the durable record is unaffected. This is the fire-and-forget
+discipline again: the tail is a UX nicety, never a correctness path. The buffer is
+leader-only (the browser polls the same instance the runner streams to) and swept
+after a short idle TTL on the reconcile tick.
## Secrets
diff --git a/internal/ci/livelogs.go b/internal/ci/livelogs.go
new file mode 100644
index 0000000..d55722e
--- /dev/null
+++ b/internal/ci/livelogs.go
@@ -0,0 +1,106 @@
+package ci
+
+import (
+ "sync"
+ "time"
+)
+
+// LiveLogs holds in-memory, per-job live CI log buffers on the leader — the
+// ephemeral tail the UI polls while a job runs. It is deliberately NOT replicated
+// or persisted: the durable log is the blob the runner ships on completion
+// (content-before-pointer, see report.go). If this is lost (a leader restart), the
+// live tail blips but the durable record is unaffected — live streaming is a
+// best-effort UX layer, never a correctness path. Safe for concurrent use: the
+// runner's chunk POSTs append while the browser polls Read.
+type LiveLogs struct {
+ mu sync.Mutex
+ jobs map[int64]*liveLog
+ cap int
+}
+
+type liveLog struct {
+ buf []byte
+ done bool
+ truncated bool
+ updated time.Time
+}
+
+// NewLiveLogs returns an empty store capped at the same size as the durable log.
+func NewLiveLogs() *LiveLogs {
+ return &LiveLogs{jobs: map[int64]*liveLog{}, cap: logCap}
+}
+
+// Append adds chunk to jobID's live buffer, creating it on first write and
+// enforcing the shared size cap with a one-time truncation marker (never a silent
+// cut). Best-effort: appends after the cap are dropped.
+func (l *LiveLogs) Append(jobID int64, chunk []byte, now time.Time) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ lg := l.jobs[jobID]
+ if lg == nil {
+ lg = &liveLog{}
+ l.jobs[jobID] = lg
+ }
+ lg.updated = now
+ if lg.truncated {
+ return
+ }
+ if room := l.cap - len(lg.buf); len(chunk) > room {
+ if room > 0 {
+ lg.buf = append(lg.buf, chunk[:room]...)
+ }
+ lg.buf = append(lg.buf, logTruncationMarker...)
+ lg.truncated = true
+ return
+ }
+ lg.buf = append(lg.buf, chunk...)
+}
+
+// Finish marks jobID's live tail complete so a reader knows to switch to the
+// durable log. A no-op for an unknown job (its buffer may already be swept — the
+// reader then falls back to the durable log by job status).
+func (l *LiveLogs) Finish(jobID int64, now time.Time) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ if lg := l.jobs[jobID]; lg != nil {
+ lg.done = true
+ lg.updated = now
+ }
+}
+
+// Read returns jobID's buffered bytes from offset onward, the next offset to poll
+// from, and whether the job has finished. ok is false when no live buffer exists
+// (never started, or swept) — the caller should fall back to the durable log. A
+// negative or over-long offset is clamped, so a stale client never panics or
+// re-reads the whole buffer.
+func (l *LiveLogs) Read(jobID int64, offset int) (data []byte, next int, done, ok bool) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ lg := l.jobs[jobID]
+ if lg == nil {
+ return nil, offset, false, false
+ }
+ switch {
+ case offset < 0:
+ offset = 0
+ case offset > len(lg.buf):
+ offset = len(lg.buf)
+ }
+ out := make([]byte, len(lg.buf)-offset)
+ copy(out, lg.buf[offset:])
+ return out, len(lg.buf), lg.done, true
+}
+
+// Sweep drops buffers idle (no append/finish) for longer than ttl, reclaiming
+// memory once a run is done or its runner has died. Active jobs flush every second
+// so their buffers stay fresh; a swept buffer just makes Read fall back to the
+// durable log.
+func (l *LiveLogs) Sweep(now time.Time, ttl time.Duration) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ for id, lg := range l.jobs {
+ if now.Sub(lg.updated) > ttl {
+ delete(l.jobs, id)
+ }
+ }
+}
diff --git a/internal/ci/livelogs_test.go b/internal/ci/livelogs_test.go
new file mode 100644
index 0000000..653eacb
--- /dev/null
+++ b/internal/ci/livelogs_test.go
@@ -0,0 +1,99 @@
+package ci
+
+import (
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestLiveLogsAppendAndRead(t *testing.T) {
+ l := NewLiveLogs()
+ now := time.Unix(0, 0)
+ l.Append(7, []byte("hello "), now)
+ l.Append(7, []byte("world"), now)
+
+ data, next, done, ok := l.Read(7, 0)
+ if !ok || done {
+ t.Fatalf("Read = ok %v done %v, want ok=true done=false", ok, done)
+ }
+ if string(data) != "hello world" || next != 11 {
+ t.Fatalf("Read = %q next %d, want %q next 11", data, next, "hello world")
+ }
+ // A follow-up read from the last offset returns only the new bytes.
+ l.Append(7, []byte("!"), now)
+ data, next, _, _ = l.Read(7, next)
+ if string(data) != "!" || next != 12 {
+ t.Fatalf("incremental Read = %q next %d, want %q next 12", data, next, "!")
+ }
+}
+
+func TestLiveLogsReadUnknownJob(t *testing.T) {
+ // The failure path: no buffer for the job → ok=false so the caller falls back
+ // to the durable log rather than showing an empty live tail forever.
+ data, next, done, ok := NewLiveLogs().Read(99, 3)
+ if ok || done || data != nil || next != 3 {
+ t.Fatalf("Read unknown = data %q next %d done %v ok %v, want nil/3/false/false", data, next, done, ok)
+ }
+}
+
+func TestLiveLogsReadClampsOffset(t *testing.T) {
+ l := NewLiveLogs()
+ l.Append(1, []byte("abc"), time.Unix(0, 0))
+ // An over-long offset (stale client) clamps to the end, never panics.
+ if data, next, _, ok := l.Read(1, 999); !ok || len(data) != 0 || next != 3 {
+ t.Fatalf("clamped Read = %q next %d ok %v, want empty/3/true", data, next, ok)
+ }
+ // A negative offset clamps to 0.
+ if data, _, _, _ := l.Read(1, -5); string(data) != "abc" {
+ t.Fatalf("negative-offset Read = %q, want abc", data)
+ }
+}
+
+func TestLiveLogsFinish(t *testing.T) {
+ l := NewLiveLogs()
+ now := time.Unix(0, 0)
+ l.Append(2, []byte("x"), now)
+ if _, _, done, _ := l.Read(2, 0); done {
+ t.Fatal("job reported done before Finish")
+ }
+ l.Finish(2, now)
+ if _, _, done, ok := l.Read(2, 0); !ok || !done {
+ t.Fatalf("after Finish: ok %v done %v, want both true", ok, done)
+ }
+ l.Finish(404, now) // unknown job is a no-op, not a panic
+}
+
+func TestLiveLogsCapTruncatesOnce(t *testing.T) {
+ l := &LiveLogs{jobs: map[int64]*liveLog{}, cap: 8}
+ now := time.Unix(0, 0)
+ l.Append(1, []byte("0123456"), now) // 7 bytes, under cap 8
+ l.Append(1, []byte("789abc"), now) // crosses the cap
+ l.Append(1, []byte("more"), now) // dropped (already truncated)
+
+ data, _, _, _ := l.Read(1, 0)
+ if !strings.HasPrefix(string(data), "01234567") {
+ t.Fatalf("data = %q, want the first 8 bytes kept", data)
+ }
+ if !strings.Contains(string(data), logTruncationMarker) {
+ t.Errorf("data = %q, want an explicit truncation marker", data)
+ }
+ if strings.Contains(string(data), "more") {
+ t.Errorf("data = %q, want post-cap appends dropped", data)
+ }
+}
+
+func TestLiveLogsSweep(t *testing.T) {
+ l := NewLiveLogs()
+ start := time.Unix(1000, 0)
+ l.Append(1, []byte("old"), start)
+ l.Append(2, []byte("fresh"), start.Add(9*time.Minute))
+
+ // Sweep at start+10m with a 5m TTL: job 1 (idle 10m) goes, job 2 (idle 1m) stays.
+ l.Sweep(start.Add(10*time.Minute), 5*time.Minute)
+ if _, _, _, ok := l.Read(1, 0); ok {
+ t.Error("job 1 should have been swept (idle past TTL)")
+ }
+ if _, _, _, ok := l.Read(2, 0); !ok {
+ t.Error("job 2 should survive (idle within TTL)")
+ }
+}
diff --git a/internal/ci/report.go b/internal/ci/report.go
index 161c2a3..2c6062e 100644
--- a/internal/ci/report.go
+++ b/internal/ci/report.go
@@ -24,6 +24,10 @@ const (
logCap = 10 << 20 // 10 MiB
// logTruncationMarker is appended when a log is truncated at the cap.
logTruncationMarker = "\n\n[log truncated: exceeded 10 MiB cap]\n"
+ // liveLogTTL bounds how long a job's in-memory live buffer survives without an
+ // update before Sweep reclaims it — long enough for a browser to catch the
+ // terminal status and switch to the durable log.
+ liveLogTTL = 10 * time.Minute
)
// ReportMeta is the slice of meta the report API reads and writes. *meta.Metadata
@@ -47,27 +51,33 @@ type ReportMeta interface {
type ReportAPI struct {
meta ReportMeta
store store.Store
+ live *LiveLogs
isLeader func() bool
secret string
logger *slog.Logger
}
-// NewReportAPI returns a report API. isLeader gates writes (nil means always
+// NewReportAPI returns a report API. live is the in-memory live-log store the
+// browser tails (nil makes a fresh one). isLeader gates writes (nil means always
// leader, for tests/local). An empty secret rejects every request.
-func NewReportAPI(m ReportMeta, s store.Store, isLeader func() bool, secret string, logger *slog.Logger) *ReportAPI {
+func NewReportAPI(m ReportMeta, s store.Store, live *LiveLogs, isLeader func() bool, secret string, logger *slog.Logger) *ReportAPI {
+ if live == nil {
+ live = NewLiveLogs()
+ }
if isLeader == nil {
isLeader = func() bool { return true }
}
if logger == nil {
logger = slog.New(slog.DiscardHandler)
}
- return &ReportAPI{meta: m, store: s, isLeader: isLeader, secret: secret, logger: logger}
+ return &ReportAPI{meta: m, store: s, live: live, isLeader: isLeader, secret: secret, logger: logger}
}
// Register mounts the internal CI routes on mux (not under /ui — these are
// machine-to-machine, WORKER_SECRET-authenticated).
func (a *ReportAPI) Register(mux *http.ServeMux) {
mux.Handle("GET /internal/ci/jobs/{id}", http.HandlerFunc(a.handleClaim))
+ mux.Handle("POST /internal/ci/jobs/{id}/log", http.HandlerFunc(a.handleLogChunk))
mux.Handle("POST /internal/ci/jobs/{id}/complete", http.HandlerFunc(a.handleComplete))
}
@@ -179,10 +189,34 @@ func (a *ReportAPI) handleComplete(w http.ResponseWriter, r *http.Request) {
return
}
a.rollupRun(r.Context(), run.ID)
+ // Flip the live tail to done so a watching browser switches to the durable log.
+ a.live.Finish(id, time.Now())
writeJSON(w, map[string]bool{"ok": true})
}
+// handleLogChunk appends a runner's incremental log bytes to the job's in-memory
+// live buffer — the best-effort tail the UI polls while the job runs. It never
+// touches s3lite or S3 (the durable log is written once at completion); a lost
+// chunk is a cosmetic gap, not data loss. Leader-only (the browser reads the same
+// leader's buffer), so a follower's 503 just makes the runner drop the chunk.
+func (a *ReportAPI) handleLogChunk(w http.ResponseWriter, r *http.Request) {
+ if !a.authed(w, r) {
+ return
+ }
+ id, ok := jobID(w, r)
+ if !ok {
+ return
+ }
+ chunk, err := io.ReadAll(io.LimitReader(r.Body, logCap))
+ if err != nil {
+ http.Error(w, "read chunk", http.StatusBadRequest)
+ return
+ }
+ a.live.Append(id, chunk, time.Now())
+ w.WriteHeader(http.StatusNoContent)
+}
+
// storeLog writes the body under key, enforcing the size cap with an explicit
// truncation marker rather than a silent cut.
func (a *ReportAPI) storeLog(ctx context.Context, key string, body io.Reader) error {
@@ -219,6 +253,8 @@ func (a *ReportAPI) ReconcileStuck(ctx context.Context, maxAge time.Duration, no
if !a.isLeader() {
return nil
}
+ // Reclaim in-memory live buffers for finished or dead jobs on the same tick.
+ a.live.Sweep(now, liveLogTTL)
stuck, err := a.meta.SweepStuckJobs(ctx, now.Add(-maxAge))
if err != nil {
return err
diff --git a/internal/ci/report_test.go b/internal/ci/report_test.go
index 5a77b3d..7eb699e 100644
--- a/internal/ci/report_test.go
+++ b/internal/ci/report_test.go
@@ -53,7 +53,7 @@ func newReportFixture(t *testing.T, isLeader func() bool) reportFixture {
t.Fatalf("CreateJob: %v", err)
}
- api := NewReportAPI(md, s, isLeader, workerSecret, nil)
+ api := NewReportAPI(md, s, nil, isLeader, workerSecret, nil)
mux := http.NewServeMux()
api.Register(mux)
srv := httptest.NewServer(mux)
@@ -257,4 +257,49 @@ func TestReconcileStuckSweepsRunningJobs(t *testing.T) {
}
}
+func TestLogChunkAppendsToLiveBuffer(t *testing.T) {
+ f := newReportFixture(t, nil)
+
+ resp := f.do(t, http.MethodPost, "/internal/ci/jobs/"+id(f.jobID)+"/log", workerSecret, "step 1...\n")
+ resp.Body.Close()
+ if resp.StatusCode != http.StatusNoContent {
+ t.Fatalf("log chunk status = %d, want 204", resp.StatusCode)
+ }
+ f.do(t, http.MethodPost, "/internal/ci/jobs/"+id(f.jobID)+"/log", workerSecret, "step 2...\n").Body.Close()
+
+ data, _, done, ok := f.api.live.Read(f.jobID, 0)
+ if !ok || done {
+ t.Fatalf("live Read = ok %v done %v, want ok=true done=false", ok, done)
+ }
+ if string(data) != "step 1...\nstep 2...\n" {
+ t.Errorf("live buffer = %q, want the two appended chunks", data)
+ }
+}
+
+func TestLogChunkRequiresAuth(t *testing.T) {
+ // The failure path: an unauthenticated chunk is rejected and never buffered.
+ f := newReportFixture(t, nil)
+ resp := f.do(t, http.MethodPost, "/internal/ci/jobs/"+id(f.jobID)+"/log", "", "sneaky\n")
+ resp.Body.Close()
+ if resp.StatusCode != http.StatusUnauthorized {
+ t.Fatalf("unauthenticated chunk status = %d, want 401", resp.StatusCode)
+ }
+ if _, _, _, ok := f.api.live.Read(f.jobID, 0); ok {
+ t.Error("rejected chunk must not create a live buffer")
+ }
+}
+
+func TestCompleteFinishesLiveTail(t *testing.T) {
+ f := newReportFixture(t, nil)
+ f.do(t, http.MethodGet, "/internal/ci/jobs/"+id(f.jobID), workerSecret, "").Body.Close()
+ f.do(t, http.MethodPost, "/internal/ci/jobs/"+id(f.jobID)+"/log", workerSecret, "running...\n").Body.Close()
+ f.do(t, http.MethodPost, "/internal/ci/jobs/"+id(f.jobID)+"/complete?status=passed", workerSecret, "running...\ndone\n").Body.Close()
+
+ // Completion flips the live tail to done so a watching browser reloads to the
+ // durable log.
+ if _, _, done, ok := f.api.live.Read(f.jobID, 0); !ok || !done {
+ t.Fatalf("live tail after complete: ok %v done %v, want both true", ok, done)
+ }
+}
+
func id(n int64) string { return strconv.FormatInt(n, 10) }
diff --git a/internal/runner/act.go b/internal/runner/act.go
index addab0b..b1b681c 100644
--- a/internal/runner/act.go
+++ b/internal/runner/act.go
@@ -5,6 +5,7 @@ import (
"context"
"errors"
"fmt"
+ "io"
"os"
"os/exec"
"strconv"
@@ -61,7 +62,7 @@ type ActEngine struct{}
// as `-s NAME`, which act reads from its environment (so values never touch the
// argv, and multiline values survive) and exposes to the workflow as
// `${{ secrets.NAME }}` — exactly as on GitHub.
-func (ActEngine) Run(ctx context.Context, repoDir, workflowDir string) ([]byte, bool, error) {
+func (ActEngine) Run(ctx context.Context, repoDir, workflowDir string, sink io.Writer) ([]byte, bool, error) {
if _, err := exec.LookPath("act"); err != nil {
return nil, false, fmt.Errorf("act not found on PATH — install nektos/act (https://github.com/nektos/act): %w", err)
}
@@ -72,9 +73,15 @@ func (ActEngine) Run(ctx context.Context, repoDir, workflowDir string) ([]byte,
cmd := exec.CommandContext(ctx, "act", args...)
cmd.Dir = repoDir
cmd.Env = actEnv
+ // Capture the full combined output for the durable log, and tee it to sink for
+ // the live tail (nil sink → capture only).
var buf bytes.Buffer
- cmd.Stdout = &buf
- cmd.Stderr = &buf
+ var out io.Writer = &buf
+ if sink != nil {
+ out = io.MultiWriter(&buf, sink)
+ }
+ cmd.Stdout = out
+ cmd.Stderr = out
err := cmd.Run()
if err == nil {
diff --git a/internal/runner/runner.go b/internal/runner/runner.go
index c7fd54e..95ccedb 100644
--- a/internal/runner/runner.go
+++ b/internal/runner/runner.go
@@ -38,9 +38,11 @@ type Cloner interface {
// Engine runs the workflows under workflowDir within repoDir. It returns the
// combined log and whether the run passed. A non-nil error means the engine
// could not execute at all (e.g. act missing) — distinct from a workflow that
-// ran and failed (passed=false, err=nil).
+// ran and failed (passed=false, err=nil). It also tees the combined output to
+// sink as it is produced (nil to disable), which the runner uses to stream a live
+// tail; sink is best-effort and never affects the returned log or status.
type Engine interface {
- Run(ctx context.Context, repoDir, workflowDir string) (log []byte, passed bool, err error)
+ Run(ctx context.Context, repoDir, workflowDir string, sink io.Writer) (log []byte, passed bool, err error)
}
// Config wires one runner invocation.
@@ -107,7 +109,14 @@ func Run(ctx context.Context, cfg Config) error {
return complete(ctx, cfg, sp.JobID, "error", []byte("clone failed: "+err.Error()+"\n"))
}
- log, passed, err := cfg.Engine.Run(ctx, dest, sp.WorkflowDir)
+ // Stream the combined output to the leader's live buffer as it is produced, so
+ // the UI can tail the job. Best-effort: the durable log is still shipped whole
+ // by complete below, so a dropped chunk (or an unreachable leader) is only a
+ // cosmetic gap in the live view.
+ streamer := newLogStreamer(cfg, sp.JobID)
+ streamer.start()
+ log, passed, err := cfg.Engine.Run(ctx, dest, sp.WorkflowDir, streamer)
+ streamer.stop() // final flush before the durable completion below
if err != nil {
cfg.Logger.Error("ci runner: engine error", "job", sp.JobID, "error", err)
body := append(append([]byte{}, log...), []byte("\nengine error: "+err.Error()+"\n")...)
diff --git a/internal/runner/runner_test.go b/internal/runner/runner_test.go
index 06d71b7..a9b6454 100644
--- a/internal/runner/runner_test.go
+++ b/internal/runner/runner_test.go
@@ -17,11 +17,12 @@ var testSpec = spec{
Ref: "refs/heads/main", WorkflowDir: ".gitmote/workflows",
}
-// capture records the completion the runner posts.
+// capture records the completion the runner posts, plus any streamed log chunks.
type capture struct {
calls int
status string
body string
+ stream string // concatenated live-log chunks posted to /log
}
// newTestServer serves the two report endpoints. claimStatus controls the claim
@@ -41,6 +42,15 @@ func newTestServer(t *testing.T, claimStatus int) (*httptest.Server, *capture) {
}
_ = json.NewEncoder(w).Encode(testSpec)
})
+ mux.HandleFunc("POST /internal/ci/jobs/{id}/log", func(w http.ResponseWriter, r *http.Request) {
+ if r.Header.Get("X-Worker-Secret") != testSecret {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ b, _ := io.ReadAll(r.Body)
+ cap.stream += string(b)
+ w.WriteHeader(http.StatusNoContent)
+ })
mux.HandleFunc("POST /internal/ci/jobs/{id}/complete", func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("X-Worker-Secret") != testSecret {
http.Error(w, "unauthorized", http.StatusUnauthorized)
@@ -76,8 +86,11 @@ type fakeEngine struct {
called bool
}
-func (e *fakeEngine) Run(_ context.Context, _, _ string) ([]byte, bool, error) {
+func (e *fakeEngine) Run(_ context.Context, _, _ string, sink io.Writer) ([]byte, bool, error) {
e.called = true
+ if sink != nil && len(e.log) > 0 {
+ _, _ = sink.Write(e.log) // mimic act teeing its output to the live sink
+ }
return e.log, e.passed, e.err
}
@@ -174,6 +187,50 @@ func TestRunClaimUnauthorizedFails(t *testing.T) {
}
}
+func TestRunStreamsLiveLog(t *testing.T) {
+ srv, cap := newTestServer(t, http.StatusOK)
+ en := &fakeEngine{log: []byte("live line\n"), passed: true}
+ if err := Run(context.Background(), baseConfig(t, srv.URL, &fakeCloner{}, en)); err != nil {
+ t.Fatalf("Run: %v", err)
+ }
+ // The engine's output was streamed live (final flush on stop, before complete)...
+ if cap.stream != "live line\n" {
+ t.Errorf("streamed = %q, want the engine output streamed to /log", cap.stream)
+ }
+ // ...and the durable completion still carries the full log.
+ if cap.body != "live line\n" {
+ t.Errorf("completion body = %q, want the full durable log", cap.body)
+ }
+}
+
+func TestRunStreamFailureDoesNotBreakRun(t *testing.T) {
+ // The failure path: a leader that rejects live chunks must not fail the run —
+ // the durable completion still lands. /log 500s, /complete works.
+ cap := &capture{}
+ mux := http.NewServeMux()
+ mux.HandleFunc("GET /internal/ci/jobs/{id}", func(w http.ResponseWriter, _ *http.Request) {
+ _ = json.NewEncoder(w).Encode(testSpec)
+ })
+ mux.HandleFunc("POST /internal/ci/jobs/{id}/log", func(w http.ResponseWriter, _ *http.Request) {
+ http.Error(w, "nope", http.StatusInternalServerError)
+ })
+ mux.HandleFunc("POST /internal/ci/jobs/{id}/complete", func(w http.ResponseWriter, r *http.Request) {
+ cap.calls++
+ cap.status = r.URL.Query().Get("status")
+ w.WriteHeader(http.StatusOK)
+ })
+ srv := httptest.NewServer(mux)
+ t.Cleanup(srv.Close)
+
+ en := &fakeEngine{log: []byte("output\n"), passed: true}
+ if err := Run(context.Background(), baseConfig(t, srv.URL, &fakeCloner{}, en)); err != nil {
+ t.Fatalf("Run must succeed despite live-stream failures: %v", err)
+ }
+ if cap.calls != 1 || cap.status != "passed" {
+ t.Errorf("completion calls %d status %q, want 1/passed", cap.calls, cap.status)
+ }
+}
+
type errStub string
func (e errStub) Error() string { return string(e) }
diff --git a/internal/runner/stream.go b/internal/runner/stream.go
new file mode 100644
index 0000000..21bee20
--- /dev/null
+++ b/internal/runner/stream.go
@@ -0,0 +1,115 @@
+package runner
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "log/slog"
+ "net/http"
+ "sync"
+ "time"
+)
+
+// liveLogFlushInterval is how often the streamer ships accumulated output to the
+// leader's live buffer. Short enough to feel live, long enough to keep the HTTP
+// overhead negligible against a multi-minute build.
+const liveLogFlushInterval = time.Second
+
+// logStreamer is the io.Writer the engine tees its combined output to. A
+// background goroutine periodically POSTs the accumulated bytes to the leader's
+// live-log endpoint, so the UI can tail a running job. It is deliberately
+// best-effort: a failed POST drops that chunk (the durable log is shipped whole at
+// completion), so streaming never blocks or fails the run.
+type logStreamer struct {
+ url string
+ secret string
+ client *http.Client
+ logger *slog.Logger
+
+ mu sync.Mutex
+ pending []byte
+
+ stopCh chan struct{}
+ doneCh chan struct{}
+}
+
+// newLogStreamer builds a streamer targeting cfg's leader for jobID.
+func newLogStreamer(cfg Config, jobID int64) *logStreamer {
+ return &logStreamer{
+ url: fmt.Sprintf("%s/internal/ci/jobs/%d/log", cfg.BaseURL, jobID),
+ secret: cfg.WorkerSecret,
+ client: cfg.HTTPClient,
+ logger: cfg.Logger,
+ stopCh: make(chan struct{}),
+ doneCh: make(chan struct{}),
+ }
+}
+
+// Write buffers p for the next flush. It never errors, so teeing it into the
+// engine's output can't disturb the real capture.
+func (s *logStreamer) Write(p []byte) (int, error) {
+ s.mu.Lock()
+ s.pending = append(s.pending, p...)
+ s.mu.Unlock()
+ return len(p), nil
+}
+
+// start launches the flush loop. Call stop exactly once to end it.
+func (s *logStreamer) start() { go s.loop() }
+
+// stop ends the flush loop after a final flush, and blocks until it returns.
+func (s *logStreamer) stop() {
+ close(s.stopCh)
+ <-s.doneCh
+}
+
+func (s *logStreamer) loop() {
+ defer close(s.doneCh)
+ t := time.NewTicker(liveLogFlushInterval)
+ defer t.Stop()
+ for {
+ select {
+ case <-t.C:
+ s.flush()
+ case <-s.stopCh:
+ s.flush() // ship whatever's left before the durable completion
+ return
+ }
+ }
+}
+
+// flush POSTs the pending bytes and clears them. On failure the chunk is dropped
+// (best-effort): the durable log still carries everything, so a live gap is
+// cosmetic.
+func (s *logStreamer) flush() {
+ s.mu.Lock()
+ chunk := s.pending
+ s.pending = nil
+ s.mu.Unlock()
+ if len(chunk) == 0 {
+ return
+ }
+ if err := s.post(chunk); err != nil {
+ s.logger.Debug("ci runner: live log flush dropped", "error", err)
+ }
+}
+
+func (s *logStreamer) post(chunk []byte) error {
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.url, bytes.NewReader(chunk))
+ if err != nil {
+ return err
+ }
+ req.Header.Set("X-Worker-Secret", s.secret)
+ req.Header.Set("Content-Type", "text/plain; charset=utf-8")
+ resp, err := s.client.Do(req)
+ if err != nil {
+ return err
+ }
+ defer func() { _ = resp.Body.Close() }()
+ if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK {
+ return fmt.Errorf("live log chunk: status %d", resp.StatusCode)
+ }
+ return nil
+}
diff --git a/internal/webui/ci.go b/internal/webui/ci.go
index c73d8a0..1e2f154 100644
--- a/internal/webui/ci.go
+++ b/internal/webui/ci.go
@@ -1,6 +1,7 @@
package webui
import (
+ "encoding/json"
"errors"
"io"
"net/http"
@@ -66,9 +67,11 @@ func (h *Handler) ciRun(w http.ResponseWriter, r *http.Request, repoName, arg st
h.ciRunDetail(w, r, rp, run)
return
}
- // The only sub-resource is a job's log: "job/{jid}/log".
+ // Sub-resources of a run: "job/{jid}/log" (the log page) and
+ // "job/{jid}/log/live" (the live-tail JSON the page polls while it runs).
parts := strings.Split(rest, "/")
- if len(parts) != 3 || parts[0] != "job" || parts[2] != "log" {
+ live := len(parts) == 4 && parts[3] == "live"
+ if (len(parts) != 3 && !live) || parts[0] != "job" || parts[2] != "log" {
http.NotFound(w, r)
return
}
@@ -77,6 +80,10 @@ func (h *Handler) ciRun(w http.ResponseWriter, r *http.Request, repoName, arg st
http.NotFound(w, r)
return
}
+ if live {
+ h.ciJobLogLive(w, r, run, jobID)
+ return
+ }
h.ciJobLog(w, r, rp, run, jobID)
}
@@ -111,14 +118,69 @@ func (h *Handler) ciJobLog(w http.ResponseWriter, r *http.Request, rp *meta.Repo
Run: *run,
Job: *job,
}
- if job.LogKey == "" {
+ switch {
+ case !isTerminalStatus(job.Status):
+ // Still queued/running: the page tails the live buffer and reloads on done.
+ data.Live = true
+ case job.LogKey == "":
data.Note = "No log yet — the job hasn't reported."
- } else {
+ default:
h.loadLog(r, job.LogKey, &data)
}
h.render(w, "ci_log.html", data)
}
+// ciJobLogLive answers the log page's poll for a running job: the live buffer's
+// bytes from offset onward, the next offset, and whether the job has finished (so
+// the page reloads to the durable log). Falls back to done-by-status when there is
+// no live buffer (never started, swept, or the live store is disabled).
+func (h *Handler) ciJobLogLive(w http.ResponseWriter, r *http.Request, run *meta.Run, jobID int64) {
+ job, err := h.md.GetJob(r.Context(), jobID)
+ if errors.Is(err, meta.ErrNotFound) || (err == nil && job.RunID != run.ID) {
+ http.NotFound(w, r)
+ return
+ }
+ if err != nil {
+ h.serverError(w, "get job", err)
+ return
+ }
+ offset, _ := strconv.Atoi(r.URL.Query().Get("offset"))
+ resp := liveLogResp{Next: offset}
+ if data, next, done, ok := h.readLive(jobID, offset); ok {
+ resp.Bytes, resp.Next, resp.Done = string(data), next, done
+ } else {
+ resp.Done = isTerminalStatus(job.Status)
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _ = json.NewEncoder(w).Encode(resp)
+}
+
+// readLive reads the live buffer, returning ok=false when the store is disabled.
+func (h *Handler) readLive(jobID int64, offset int) (data []byte, next int, done, ok bool) {
+ if h.live == nil {
+ return nil, offset, false, false
+ }
+ return h.live.Read(jobID, offset)
+}
+
+// liveLogResp is the JSON the log page polls: new bytes, the next offset to poll
+// from, and whether the job has finished.
+type liveLogResp struct {
+ Bytes string `json:"bytes"`
+ Next int `json:"next"`
+ Done bool `json:"done"`
+}
+
+// isTerminalStatus reports whether a run/job status is a finished state.
+func isTerminalStatus(s meta.RunStatus) bool {
+ switch s {
+ case meta.RunPassed, meta.RunFailed, meta.RunError, meta.RunSuperseded:
+ return true
+ default:
+ return false
+ }
+}
+
// loadLog fetches a job's log blob and fills data with either the ANSI-rendered
// HTML or, over the render cap, a plain-text prefix with a note. Read failures
// degrade to a note — the run detail already showed the status.
diff --git a/internal/webui/ci_test.go b/internal/webui/ci_test.go
index cb68975..48a2650 100644
--- a/internal/webui/ci_test.go
+++ b/internal/webui/ci_test.go
@@ -3,6 +3,7 @@ package webui
import (
"bytes"
"context"
+ "encoding/json"
"net/http"
"strings"
"testing"
@@ -10,6 +11,49 @@ import (
"github.com/atmin/gitmote/internal/meta"
)
+// fakeLive is a fixed live-log buffer for exercising the live-tail endpoint.
+type fakeLive struct {
+ data []byte
+ done bool
+ ok bool
+}
+
+func (f fakeLive) Read(_ int64, offset int) ([]byte, int, bool, bool) {
+ if !f.ok {
+ return nil, offset, false, false
+ }
+ if offset < 0 {
+ offset = 0
+ }
+ if offset > len(f.data) {
+ offset = len(f.data)
+ }
+ return f.data[offset:], len(f.data), f.done, true
+}
+
+// seedQueuedJob creates an ACL-granted repo, a run, and a queued (non-terminal)
+// job with no log yet — the running-job state the live tail serves.
+func (x *harness) seedQueuedJob(repoName, sha string) (runID, jobID int64) {
+ x.t.Helper()
+ ctx := context.Background()
+ r, err := x.md.CreateRepo(ctx, repoName, "main")
+ if err != nil {
+ x.t.Fatalf("CreateRepo: %v", err)
+ }
+ if err := x.md.SetACL(ctx, r.ID, x.admin.ID, meta.PermAdmin); err != nil {
+ x.t.Fatalf("SetACL: %v", err)
+ }
+ run, err := x.md.CreateRun(ctx, r.ID, "refs/heads/main", sha)
+ if err != nil {
+ x.t.Fatalf("CreateRun: %v", err)
+ }
+ job, err := x.md.CreateJob(ctx, run.ID, "ci.yml")
+ if err != nil {
+ x.t.Fatalf("CreateJob: %v", err)
+ }
+ return run.ID, job.ID
+}
+
// seedRun creates a repo (if repoName is new), a run, and one job with a stored
// log blob, returning the ids. The log carries ANSI + HTML to exercise rendering.
func (x *harness) seedRun(repoName, sha, log string) (repoID, runID, jobID int64) {
@@ -135,6 +179,64 @@ func TestCIRunsRequireAdmin(t *testing.T) {
}
}
+func TestCILiveLogStreamsBytes(t *testing.T) {
+ x := newHarness(t)
+ x.h.live = fakeLive{data: []byte("building...\n"), ok: true}
+ session := x.login(x.mintTokenFor(x.admin.ID))
+ runID, jobID := x.seedQueuedJob("app", "sha1")
+
+ rec := x.do(http.MethodGet, "/app/runs/"+itoa(runID)+"/job/"+itoa(jobID)+"/log/live?offset=0", nil, session)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("live poll: %d (%s)", rec.Code, rec.Body)
+ }
+ var resp struct {
+ Bytes string `json:"bytes"`
+ Next int `json:"next"`
+ Done bool `json:"done"`
+ }
+ if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
+ t.Fatalf("decode: %v", err)
+ }
+ if resp.Bytes != "building...\n" || resp.Next != 12 || resp.Done {
+ t.Errorf("live resp = %+v, want the buffered bytes, next 12, not done", resp)
+ }
+}
+
+func TestCILiveLogDoneWhenNoBufferAndTerminal(t *testing.T) {
+ // The fallback path: no live buffer + a finished job → done=true, so the page
+ // stops polling and reloads to the durable log.
+ x := newHarness(t)
+ x.h.live = fakeLive{ok: false}
+ session := x.login(x.mintTokenFor(x.admin.ID))
+ _, runID, jobID := x.seedRun("app", "sha1", "done\n") // seedRun marks the job passed
+
+ rec := x.do(http.MethodGet, "/app/runs/"+itoa(runID)+"/job/"+itoa(jobID)+"/log/live", nil, session)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("live poll: %d", rec.Code)
+ }
+ if !strings.Contains(rec.Body.String(), `"done":true`) {
+ t.Errorf("live resp = %s, want done=true for a finished job with no buffer", rec.Body)
+ }
+}
+
+func TestCIJobLogPageTailsRunningJob(t *testing.T) {
+ x := newHarness(t)
+ x.h.live = fakeLive{data: []byte("x"), ok: true}
+ session := x.login(x.mintTokenFor(x.admin.ID))
+ runID, jobID := x.seedQueuedJob("app", "sha1")
+
+ // A running job's log page renders the live-tail element + poll script instead
+ // of a durable log.
+ rec := x.do(http.MethodGet, "/app/runs/"+itoa(runID)+"/job/"+itoa(jobID)+"/log", nil, session)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("log page: %d", rec.Code)
+ }
+ body := rec.Body.String()
+ if !strings.Contains(body, `id="livelog"`) || !strings.Contains(body, "/log/live") {
+ t.Errorf("running-job log page missing live tail: %s", body)
+ }
+}
+
func TestCICommitBadge(t *testing.T) {
x := newHarness(t)
ctx := context.Background()
diff --git a/internal/webui/data.go b/internal/webui/data.go
index 9b422a1..efd007b 100644
--- a/internal/webui/data.go
+++ b/internal/webui/data.go
@@ -155,4 +155,5 @@ type ciLogData struct {
Log template.HTML // ANSI-rendered log (safe HTML)
Plain string // plain-text fallback for an over-cap log
Note string // an explanation when there is no rendered log
+ Live bool // job still running: the page tails the live buffer
}
diff --git a/internal/webui/templates/ci_log.html b/internal/webui/templates/ci_log.html
index 96d9a02..f6b0b2e 100644
--- a/internal/webui/templates/ci_log.html
+++ b/internal/webui/templates/ci_log.html
@@ -7,7 +7,34 @@
</p>
{{if .Note}}<div class="flash err">{{.Note}}</div>{{end}}
-{{if .Plain}}
+{{if .Live}}
+<p><em>Live — streaming while the job runs; the page reloads when it finishes.</em></p>
+<pre id="livelog"></pre>
+<script>
+(function () {
+ var pre = document.getElementById('livelog');
+ var url = "/{{.Repo}}/runs/{{.Run.ID}}/job/{{.Job.ID}}/log/live";
+ var offset = 0, raw = "";
+ // Strip ANSI CSI escapes (colors, cursor moves) for a clean live view; the
+ // durable log shown after reload renders them properly.
+ var ansi = /\x1b\[[0-9;?]*[ -\/]*[@-~]/g;
+ function poll() {
+ fetch(url + "?offset=" + offset).then(function (r) { return r.json(); }).then(function (d) {
+ if (d.bytes) {
+ raw += d.bytes;
+ offset = d.next;
+ var atBottom = (window.innerHeight + window.scrollY) >= document.body.offsetHeight - 40;
+ pre.textContent = raw.replace(ansi, '');
+ if (atBottom) { window.scrollTo(0, document.body.scrollHeight); }
+ }
+ if (d.done) { location.reload(); return; }
+ setTimeout(poll, 1000);
+ }).catch(function () { setTimeout(poll, 2000); });
+ }
+ poll();
+})();
+</script>
+{{else if .Plain}}
<pre>{{.Plain}}</pre>
{{else if .Log}}
<pre>{{.Log}}</pre>
diff --git a/internal/webui/webui.go b/internal/webui/webui.go
index cdc550f..b02a2e2 100644
--- a/internal/webui/webui.go
+++ b/internal/webui/webui.go
@@ -60,11 +60,19 @@ type SecretsAdmin interface {
// nameSegment matches a valid owner or repo path segment.
var nameSegment = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]*$`)
+// LiveLog reads a job's in-memory live CI log tail (the leader's ephemeral buffer
+// the browser polls while a job runs). *ci.LiveLogs satisfies it. nil disables the
+// live tail — the UI then shows only the durable log once the job completes.
+type LiveLog interface {
+ Read(jobID int64, offset int) (data []byte, next int, done, ok bool)
+}
+
// Handler serves the management UI.
type Handler struct {
md *meta.Metadata
mz *repo.Materializer
store store.Store
+ live LiveLog
auth Authenticator
secrets SecretsAdmin
sess *sessions
@@ -75,9 +83,10 @@ type Handler struct {
// New builds the UI handler. cookieKey signs session cookies and must be
// non-empty; the auth verifier backs the login form; mz materializes repos for
-// the read-only browse pages; objs reads CI log blobs for the run views; secrets
-// (may be nil) backs the CI secrets panel.
-func New(md *meta.Metadata, mz *repo.Materializer, objs store.Store, a Authenticator, secrets SecretsAdmin, cookieKey []byte, logger *slog.Logger) (*Handler, error) {
+// the read-only browse pages; objs reads CI log blobs for the run views; live
+// (may be nil) backs the live log tail of a running job; secrets (may be nil)
+// backs the CI secrets panel.
+func New(md *meta.Metadata, mz *repo.Materializer, objs store.Store, live LiveLog, a Authenticator, secrets SecretsAdmin, cookieKey []byte, logger *slog.Logger) (*Handler, error) {
if len(cookieKey) == 0 {
return nil, errors.New("webui: empty cookie key")
}
@@ -125,6 +134,7 @@ func New(md *meta.Metadata, mz *repo.Materializer, objs store.Store, a Authentic
md: md,
mz: mz,
store: objs,
+ live: live,
auth: a,
secrets: secrets,
sess: &sessions{key: cookieKey},
diff --git a/internal/webui/webui_test.go b/internal/webui/webui_test.go
index f004b65..5ca946e 100644
--- a/internal/webui/webui_test.go
+++ b/internal/webui/webui_test.go
@@ -51,7 +51,7 @@ func newHarness(t *testing.T) *harness {
}
svc := secrets.NewService(kr, md)
- h, err := New(md, mz, objs, auth.NewGuard(md), svc, []byte("test-cookie-key"), nil)
+ h, err := New(md, mz, objs, nil, auth.NewGuard(md), svc, []byte("test-cookie-key"), nil)
if err != nil {
t.Fatalf("New: %v", err)
}