sanderling/cmd/internal-tools/analyze/load.go

package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"slices"
	"strings"
)

const (
	manifestFileName = "campaign.json"
	recordsFileName  = "runs.jsonl"
	maxRecordBytes   = 4 * 1024 * 1024
)

// manifest mirrors the fields analyze reads from campaign.json. The step budget
// lives here rather than in any run, because it is the exposure every run in an
// arm was given and the ceiling a clean run can be censored at.
type manifest struct {
	Arm       string  `json:"arm"`
	Generator string  `json:"generator"`
	Platform  string  `json:"platform"`
	MaxSteps  int     `json:"max_steps"`
	Seeds     []int64 `json:"seeds"`
	Host      string  `json:"host"`
}

// runRecord mirrors the fields analyze reads from one line of runs.jsonl.
type runRecord struct {
	Seed        int64  `json:"seed"`
	ExitCode    int    `json:"exit_code"`
	LaunchError string `json:"launch_error"`
	TimedOut    bool   `json:"timed_out"`
	// MonotonicMillis is how long the run worked, and it is what every
	// per-hour rate here divides by: a host asleep mid-run tested nothing, so
	// charging that time to the arm would report it as slower for a reason
	// that has nothing to do with the arm. The wall clock the campaign also
	// records answers the other question, how much time passed.
	MonotonicMillis int64 `json:"monotonic_millis"`
	// DurationMillis is the name campaigns written before the two clocks were
	// split gave the same monotonic reading, so those files still read.
	DurationMillis           int64  `json:"duration_millis"`
	TraceError               string `json:"trace_error"`
	Steps                    int    `json:"steps"`
	FirstViolationOriginStep *int   `json:"first_violation_origin_step"`
	// FirstViolationDetectedStep is the step the violation was reported on,
	// which is the origin step for a safety property tripping under its own
	// action and the end of the budget for an obligation that never discharged.
	// It is what the survival analysis times the event by; see eventStep.
	FirstViolationDetectedStep *int     `json:"first_violation_detected_step"`
	ViolatedProperties         []string `json:"violated_properties"`
	// Actions is the count of steps on which the action generator dispatched an
	// action, which excludes the steps the spec's setup drove before the
	// generator was consulted wherever the trace names them. It is a pointer so
	// that a runs.jsonl written before the campaign tool counted them is refused
	// rather than read as an arm that acted zero times. The campaign tool always
	// emits the field, so its absence dates the file.
	Actions *int `json:"actions"`
	// UnattributedActions is how many of the run's actions name no producer, so
	// nothing can say whether the spec's setup drove them. It is a pointer
	// because a runs.jsonl written before actions named a producer at all has no
	// field, and reading that silence as none would let a denominator of unknown
	// provenance pass for one that excludes the setup's login.
	UnattributedActions *int `json:"unattributed_actions"`
}

// Exclusion reasons. A run that failed or timed out is missing data, not a
// censored observation: it broke off, so its step count is not exposure the app
// survived and counting it as one would bias the survival estimate downward.
const (
	reasonLaunchError   = "launch error"
	reasonTimedOut      = "timed out"
	reasonNonzeroExit   = "nonzero exit"
	reasonTraceError    = "unreadable trace"
	reasonMalformedStep = "violation step outside the budget"
)

type classifiedRun struct {
	Seed                int64
	Steps               int
	Actions             int
	UnattributedActions int
	MonotonicMillis     int64
	OriginStep          int
	// EventStep is when the run could know, and it is what the survival
	// analysis measures. It is the origin step whenever the two agree.
	EventStep          int
	Violated           bool
	ClampedToBudget    bool
	ViolatedProperties []string
	ExcludedBecause    string
}

type arm struct {
	Name         string
	Budget       int
	Generator    string
	Platform     string
	Directories  []string
	Runs         []classifiedRun
	MissingSeeds []int64
}

func loadCampaign(directory string) (manifest, []runRecord, error) {
	body, err := os.ReadFile(filepath.Join(directory, manifestFileName))
	if err != nil {
		return manifest{}, nil, fmt.Errorf("read %s: %w", manifestFileName, err)
	}
	var declared manifest
	if err := json.Unmarshal(body, &declared); err != nil {
		return manifest{}, nil, fmt.Errorf("parse %s in %s: %w", manifestFileName, directory, err)
	}
	if declared.Arm == "" {
		return manifest{}, nil, fmt.Errorf("%s in %s has no arm", manifestFileName, directory)
	}
	if declared.MaxSteps <= 0 {
		return manifest{}, nil, fmt.Errorf("%s in %s has max_steps %d: clean runs have nothing to be censored at",
			manifestFileName, directory, declared.MaxSteps)
	}

	file, err := os.Open(filepath.Join(directory, recordsFileName))
	if err != nil {
		return manifest{}, nil, fmt.Errorf("read %s: %w", recordsFileName, err)
	}
	defer file.Close()

	var records []runRecord
	scanner := bufio.NewScanner(file)
	scanner.Buffer(make([]byte, 0, 64*1024), maxRecordBytes)
	lineNumber := 0
	for scanner.Scan() {
		lineNumber++
		raw := strings.TrimSpace(scanner.Text())
		if raw == "" {
			continue
		}
		var record runRecord
		if err := json.Unmarshal([]byte(raw), &record); err != nil {
			return manifest{}, nil, fmt.Errorf("%s line %d in %s: %w", recordsFileName, lineNumber, directory, err)
		}
		if record.Actions == nil {
			return manifest{}, nil, fmt.Errorf("%s line %d in %s has no actions count: it was written before "+
				"dispatched actions were counted, and reading the missing count as zero would report every "+
				"per-action rate wrongly; re-run the campaign to produce it",
				recordsFileName, lineNumber, directory)
		}
		records = append(records, record)
	}
	if err := scanner.Err(); err != nil {
		return manifest{}, nil, fmt.Errorf("read %s in %s: %w", recordsFileName, directory, err)
	}
	return declared, records, nil
}

// unattributedActions is how many of the record's actions carry no producer. A
// campaign written before the field existed carries none at all, so its whole
// action count is of unknown provenance rather than none of it.
func unattributedActions(record runRecord) int {
	if record.UnattributedActions != nil {
		return *record.UnattributedActions
	}
	if record.Actions == nil {
		return 0
	}
	return *record.Actions
}

func (r runRecord) workingMillis() int64 {
	if r.MonotonicMillis != 0 {
		return r.MonotonicMillis
	}
	return r.DurationMillis
}

// classify turns one record into the run the analysis works with, deciding
// whether it is usable and, if it is, whether it is an event or censored.
func classify(record runRecord, budget int) classifiedRun {
	item := classifiedRun{
		Seed:               record.Seed,
		Steps:              record.Steps,
		MonotonicMillis:    record.workingMillis(),
		ViolatedProperties: slices.Clone(record.ViolatedProperties),
	}
	if record.Actions != nil {
		item.Actions = *record.Actions
	}
	item.UnattributedActions = unattributedActions(record)
	switch {
	case record.LaunchError != "":
		item.ExcludedBecause = reasonLaunchError
		return item
	case record.TimedOut:
		item.ExcludedBecause = reasonTimedOut
		return item
	case record.ExitCode != 0:
		item.ExcludedBecause = reasonNonzeroExit
		return item
	case record.TraceError != "":
		item.ExcludedBecause = reasonTraceError
		return item
	}
	if record.FirstViolationOriginStep == nil {
		if len(record.ViolatedProperties) > 0 {
			item.ExcludedBecause = reasonMalformedStep
		}
		return item
	}
	origin := *record.FirstViolationOriginStep
	if origin < 1 {
		item.ExcludedBecause = reasonMalformedStep
		return item
	}
	item.Violated = true
	item.OriginStep = origin
	item.EventStep = eventStep(record, origin)
	if item.EventStep > budget {
		// The run-end finalize line reports obligations that never discharged
		// at an index one past the last executed step. That is a real detection
		// but not a real step, so it is held at the budget and counted.
		item.EventStep = budget
		item.ClampedToBudget = true
	}
	return item
}

// eventStep is the step at which the run could know it had violated. A safety
// property tripping under its own action is detected on the step that armed it
// and the two agree. An obligation that never discharges is reported when the
// run ends, and timing that at the step that armed it would record a liveness
// failure flushed at the budget as a violation found on the first step, which
// is a number the run cannot support and which no censored run can be compared
// against: the end of the run is the clock the clean runs are censored on, so
// the events have to be on it too. A campaign written before the field existed carries no
// detected step and keeps the origin.
func eventStep(record runRecord, origin int) int {
	if record.FirstViolationDetectedStep == nil {
		return origin
	}
	if detected := *record.FirstViolationDetectedStep; detected > origin {
		return detected
	}
	return origin
}

// groupArms folds every campaign directory into its arm. Two directories with
// the same arm label are pooled, which is how a campaign split across hosts is
// analysed, but they must agree on the step budget.
func groupArms(directories []string) ([]arm, error) {
	byName := map[string]*arm{}
	var order []string
	for _, directory := range directories {
		declared, records, err := loadCampaign(directory)
		if err != nil {
			return nil, err
		}
		current, seen := byName[declared.Arm]
		if !seen {
			current = &arm{
				Name:      declared.Arm,
				Budget:    declared.MaxSteps,
				Generator: declared.Generator,
				Platform:  declared.Platform,
			}
			byName[declared.Arm] = current
			order = append(order, declared.Arm)
		}
		if current.Budget != declared.MaxSteps {
			return nil, fmt.Errorf("arm %q has step budget %d in an earlier campaign and %d in %s: "+
				"runs censored at different budgets cannot be pooled",
				declared.Arm, current.Budget, declared.MaxSteps, directory)
		}
		current.Directories = append(current.Directories, directory)

		present := map[int64]bool{}
		for _, record := range records {
			present[record.Seed] = true
			current.Runs = append(current.Runs, classify(record, declared.MaxSteps))
		}
		for _, seed := range declared.Seeds {
			if !present[seed] {
				current.MissingSeeds = append(current.MissingSeeds, seed)
			}
		}
	}
	slices.Sort(order)
	arms := make([]arm, 0, len(order))
	for _, name := range order {
		arms = append(arms, *byName[name])
	}
	return arms, nil
}

// unattributedActions is how much of the arm's per-action denominator has no
// producer behind it, counted over the runs the denominator is built from.
func (a arm) unattributedActions() int {
	total := 0
	for _, item := range a.Runs {
		if item.ExcludedBecause != "" {
			continue
		}
		total += item.UnattributedActions
	}
	return total
}

// observations returns the usable runs as survival data: an event at the step
// that armed the first violation, or a censored observation at the last step
// the run reached.
func (a arm) observations() []observation {
	var result []observation
	for _, item := range a.Runs {
		if item.ExcludedBecause != "" {
			continue
		}
		result = append(result, observationOf(item, a.Budget))
	}
	return result
}

func observationOf(item classifiedRun, budget int) observation {
	if item.Violated {
		return observation{Steps: float64(item.EventStep), Event: true}
	}
	// A run that hit the campaign's wall clock stopped short of the budget, and
	// the steps it never ran are not exposure it survived.
	return observation{Steps: float64(min(item.Steps, budget)), Event: false}
}