mesh-sdk/go/buildstep/steplog.go

package buildstep

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// A build step's log on its build module's own stream (the operator, 2026-10-30: a verdict cut a failure to its last four
// kilobytes and named no place that held the rest, so a check's cause was guessed; novox/hq ADR 0157 for the shape: a
// build says what it does on the bus, under its id, in sequenced lines, read back from the stream). The step's result
// keeps each failure's Output clipped to OutputBytes, the summary; the failure's whole output, up to FullBytes, is said
// as delivery.StepLog events on delivery.StepLogEvent(run, step), at most LogEventBytes each, on the stream the module
// declares for `log.**` (max age and size in its manifest, dropping its oldest). ReadStepLog reads it back in pages.

const (
	// FullBytes is the most of one failure's output that is said whole (its last bytes).
	FullBytes = 1 << 20
	// LogEventBytes is the most lines one log event holds, and LogPageBytes the most one page of a read answers.
	LogEventBytes = 64 << 10
	LogPageBytes  = 64 << 10
)

// stepLog says one step's log: its node, id and run, the build module, how to say an event, and the next line's number.
type stepLog struct {
	say          func(event string, body any) error
	module, node string
	id, run      string
	seq          int
	// dead says a publish failed: the step says nothing more on its log, so a bus that does not take its lines costs the
	// step at most one publish's timeout, and never fails it (the 2026-10-30 review).
	dead bool
}

// stepLog is the log of the step whose ask runs under ctx, or nil when no ask runs (a tool called by hand) or the module
// says no log.
func (s AskSteps) stepLog(ctx context.Context) *stepLog {
	r, ok := ctx.Value(phaseKey{}).(*phases)
	if !ok || r == nil || s.Narrate == nil {
		return nil
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.log == nil {
		r.log = &stepLog{say: s.Narrate, module: s.Module, node: r.step.Node, id: r.step.ID, run: r.run}
	}
	return r.log
}

// narrate says every failure's Full of a step answer's "failed" on the step's log and empties it, setting the failure's
// Step, Module, Machine and Run; a failure whose lines could not all be said keeps no address, that is said once through
// logf, and the step says nothing more on its log (dead). With l nil, Full is only emptied: the step's result never holds it.
func (l *stepLog) narrate(raw json.RawMessage, logf func(string, ...any)) json.RawMessage {
	if !strings.Contains(string(raw), `"full_output"`) {
		return raw
	}
	var answer map[string]json.RawMessage
	if err := json.Unmarshal(raw, &answer); err != nil || answer["failed"] == nil {
		return raw
	}
	var failed []map[string]json.RawMessage
	if err := json.Unmarshal(answer["failed"], &failed); err != nil {
		return raw
	}
	for i, f := range failed {
		var full string
		_ = json.Unmarshal(f["full_output"], &full)
		delete(failed[i], "full_output")
		if full == "" || l == nil || l.dead {
			continue
		}
		var name struct {
			Package string `json:"package"`
			Test    string `json:"test"`
		}
		b, _ := json.Marshal(f)
		_ = json.Unmarshal(b, &name)
		if err := l.say1(name.Package, name.Test, strings.Split(full, "\n")); err != nil {
			l.dead = true
			if logf != nil {
				logf("the step %s says no more on its log, from the whole output of %s %s on: %v", l.id, name.Package, name.Test, err)
			}
			continue
		}
		for field, v := range map[string]string{"step": l.id, "module": l.module, "machine": l.node, "run": l.run} {
			failed[i][field], _ = json.Marshal(v)
		}
	}
	answer["failed"], _ = json.Marshal(failed)
	out, err := json.Marshal(answer)
	if err != nil {
		return raw
	}
	return out
}

// say1 says one failure's lines, in events of at most LogEventBytes, numbered on from the step's last line.
func (l *stepLog) say1(pkg, test string, lines []string) error {
	event := delivery.StepLogEvent(l.run, l.id)
	for len(lines) > 0 {
		n, size := 0, 0
		for n < len(lines) && (n == 0 || size+len(lines[n])+1 <= LogEventBytes) {
			if len(lines[n]) > LogEventBytes {
				lines[n] = strings.ToValidUTF8(lines[n][:LogEventBytes], "") // one line longer than an event is cut
			}
			size += len(lines[n]) + 1
			n++
		}
		if err := l.say(event, delivery.StepLog{Seq: l.seq + 1, Package: pkg, Test: test, Lines: lines[:n]}); err != nil {
			return err
		}
		l.seq += n
		lines = lines[n:]
	}
	return nil
}

// StreamRead reads a module's own stream: stdio.Stream(name).Read's shape.
type StreamRead func(events string, from uint64, limit int) (*stdio.StreamPage, error)

// StepLogPage is one page of a step's log: the lines from the stream sequence From on, headed "=== <package> <test>"
// where a failure's lines start on the page, Next the sequence to read on from (0 when nothing follows yet; a step still running may
// say more), Last the stream's last sequence, and Gap when a line number was skipped (the stream dropped it with age).
type StepLogPage struct {
	Run   string   `json:"run"`
	Step  string   `json:"step"`
	Event string   `json:"event"`
	From  uint64   `json:"from"`
	Lines []string `json:"lines"`
	Bytes int      `json:"bytes"`
	Next  uint64   `json:"next"`
	Last  uint64   `json:"last"`
	Gap   bool     `json:"gap,omitempty"`
}

// ReadStepLog reads one page of about LogPageBytes of the log of the step id under run, from the stream sequence from on.
func ReadStepLog(read StreamRead, run, id string, from uint64) (StepLogPage, error) {
	if !stepToken(run) || !stepToken(id) {
		return StepLogPage{}, fmt.Errorf("a step's log is named by its run and its step id, each one token of letters, digits and -: %q, %q", run, id)
	}
	event := delivery.StepLogEvent(run, id)
	page := StepLogPage{Run: run, Step: id, Event: event, From: from, Lines: []string{}}
	seq, test := 0, "\x00"
	for {
		got, err := read(event, from, 16)
		if err != nil {
			return page, fmt.Errorf("the log %s cannot be read: %w", event, err)
		}
		page.Last = got.Last
		for _, e := range got.Entries {
			var l delivery.StepLog
			if err := json.Unmarshal(e.Body, &l); err != nil {
				return page, fmt.Errorf("the log %s holds at %d what is no log event: %v", event, e.Sequence, err)
			}
			size := 0
			for _, line := range l.Lines {
				size += len(line) + 1
			}
			if page.Bytes > 0 && page.Bytes+size > LogPageBytes {
				page.Next = e.Sequence
				return page, nil
			}
			if seq > 0 && l.Seq != seq {
				page.Gap = true
			}
			if name := l.Package + " " + l.Test; name != test {
				page.Lines = append(page.Lines, "=== "+strings.TrimSpace(name))
				test = name
			}
			page.Lines = append(page.Lines, l.Lines...)
			page.Bytes += size
			seq = l.Seq + len(l.Lines)
		}
		if got.Next == 0 {
			return page, nil
		}
		from = got.Next
	}
}

// stepToken is one subject token of a run or a step id.
func stepToken(s string) bool {
	if s == "" {
		return false
	}
	for _, r := range s {
		if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_') {
			return false
		}
	}
	return true
}

// LogStreamOf is the local name of a build module's own log stream, <language>-build-logs (a stream's bus name is
// mesh-wide): LogStreamOf names it.
func LogStreamOf(language string) string { return language + "-build-logs" }

// LogTool is a build module's tool that reads one step's log from its own stream, {run, step, from}: a failed test's
// whole output, which the step's result and the checker's verdict keep cut. The stream is the mesh's, so any machine's
// instance answers for a step of any machine.
func LogTool(name string, read StreamRead) stdio.Tool {
	return stdio.Tool{Name: name,
		Description: "The full build log of one build step: the whole output of every test that failed in it, which a check's " +
			"verdict and the step's result keep cut to their last lines (\"... (N earlier lines)\"). One page of about 64 KB; " +
			"read on with from = the answer's next until it is 0 (a step still running may say more: read again from next). " +
			"run is the check (its heads hash) or the delivery plan the step ran for, \"none\" when the ask named none; step " +
			"is the step id; both are in a failed verdict (checker_failure, checker_pull). Kept 14 days. Read-only.",
		Input: map[string]any{"type": "object", "required": []string{"run", "step"}, "properties": map[string]any{
			"run":  map[string]any{"type": "string", "description": "the check or delivery plan the step ran for, or none"},
			"step": map[string]any{"type": "string", "description": "the step id (the ask's id)"},
			"from": map[string]any{"type": "integer", "description": "the stream sequence to read on from: the last answer's next; 0 is the start"}}},
		Run: func(args map[string]any) (any, error) {
			run, _ := args["run"].(string)
			step, _ := args["step"].(string)
			from, _ := args["from"].(float64)
			if from < 0 {
				return nil, fmt.Errorf("from is a stream sequence, 0 or more")
			}
			return ReadStepLog(read, run, step, uint64(from))
		}}
}