mirror of
https://github.com/boshu2/agentops.git
synced 2026-09-14 15:08:13 +08:00
feat(refinery): ao refinery backstop daemon (ag-qidx P2.1-2.4,2.6)
The bushido continuous-validation backstop: watch main -> full gate on each new commit -> classify deterministic-vs-flaky (re-run N=3; the 18-30% flake rate makes naive escalation noise) -> on deterministic blocking FAIL: poison beacon (.refinery-poison + git note) + fix-bead (bd create) + alert. NEVER reverts (P2.5 quorum-gated revert deferred to ag-k99u). Backstop-not-gatekeeper: down = merges still succeed; resumes from .refinery-state. Injectable ports (git/bd/ gates adapters); core tested with fakes (deterministic escalates, flaky doesn't, green clears, never-reverts). ao refinery once|run; systemd unit + runbook for bushido. go test green. ag-qidx.
This commit is contained in:
Boden Fuller
@@ -0,0 +1,84 @@
|
||||
// practices: [hexagonal-architecture, ddd-bounded-context]
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
// Register the seed checks the refinery's full gate runs.
|
||||
_ "github.com/boshu2/agentops/cli/internal/gates/checks"
|
||||
"github.com/boshu2/agentops/cli/internal/refinery"
|
||||
)
|
||||
|
||||
// `ao refinery` is the bushido continuous-validation backstop (ag-qidx P2): it
|
||||
// watches main, runs the full gate on each new commit, and on a DETERMINISTIC
|
||||
// blocking failure raises a poison beacon + files a fix-bead + alerts. It NEVER
|
||||
// blind-reverts (the repo's flake rate would make auto-revert fight developers)
|
||||
// and it is a backstop, not a gatekeeper — if it is down, merges still succeed.
|
||||
|
||||
var refineryInterval time.Duration
|
||||
|
||||
var refineryCmd = &cobra.Command{
|
||||
Use: "refinery",
|
||||
Short: "Continuous main-validation backstop (run on bushido)",
|
||||
Long: `Watch main, run the full gate on each new commit, and on a
|
||||
deterministic blocking failure raise a poison beacon + file a fix-bead. Never
|
||||
reverts; resumes from .refinery-state across restarts.
|
||||
|
||||
ao refinery once # evaluate main HEAD once (one tick)
|
||||
ao refinery run --interval 5m # loop until interrupted (systemd service)`,
|
||||
}
|
||||
|
||||
var refineryOnceCmd = &cobra.Command{
|
||||
Use: "once",
|
||||
Short: "Evaluate main HEAD once",
|
||||
Args: cobra.NoArgs,
|
||||
RunE: runRefineryOnce,
|
||||
}
|
||||
|
||||
var refineryRunCmd = &cobra.Command{
|
||||
Use: "run",
|
||||
Short: "Run the refinery loop until interrupted",
|
||||
Args: cobra.NoArgs,
|
||||
RunE: runRefineryRun,
|
||||
}
|
||||
|
||||
func init() {
|
||||
refineryRunCmd.Flags().DurationVar(&refineryInterval, "interval", 5*time.Minute, "poll interval")
|
||||
refineryCmd.AddCommand(refineryOnceCmd, refineryRunCmd)
|
||||
rootCmd.AddCommand(refineryCmd)
|
||||
}
|
||||
|
||||
func runRefineryOnce(cmd *cobra.Command, _ []string) error {
|
||||
root, err := gateRepoRoot()
|
||||
if err != nil {
|
||||
return fmt.Errorf("resolve repo root: %w", err)
|
||||
}
|
||||
res, err := refinery.NewProduction(root).RunOnce(cmd.Context())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out := cmd.OutOrStdout()
|
||||
switch {
|
||||
case res.Skipped:
|
||||
fmt.Fprintf(out, "refinery: %s unchanged — nothing to do\n", res.SHA)
|
||||
case res.Green:
|
||||
fmt.Fprintf(out, "refinery: %s GREEN\n", res.SHA)
|
||||
case len(res.Deterministic) > 0:
|
||||
fmt.Fprintf(out, "refinery: %s POISONED by %v — fix-bead %s filed (no revert)\n", res.SHA, res.Deterministic, res.FixBead)
|
||||
default:
|
||||
fmt.Fprintf(out, "refinery: %s had failures but none reproduced (flaky) — not escalated\n", res.SHA)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func runRefineryRun(cmd *cobra.Command, _ []string) error {
|
||||
root, err := gateRepoRoot()
|
||||
if err != nil {
|
||||
return fmt.Errorf("resolve repo root: %w", err)
|
||||
}
|
||||
fmt.Fprintf(cmd.OutOrStdout(), "refinery: looping every %s (Ctrl-C to stop)\n", refineryInterval)
|
||||
return refinery.NewProduction(root).Loop(cmd.Context(), refineryInterval)
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
package main
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestRefineryCmd_Registered(t *testing.T) {
|
||||
var found bool
|
||||
for _, c := range rootCmd.Commands() {
|
||||
if c.Name() == "refinery" {
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatal("`ao refinery` not registered on root")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRefineryCmd_HasSubcommands(t *testing.T) {
|
||||
want := map[string]bool{"once": false, "run": false}
|
||||
for _, c := range refineryCmd.Commands() {
|
||||
if _, ok := want[c.Name()]; ok {
|
||||
want[c.Name()] = true
|
||||
}
|
||||
}
|
||||
for name, found := range want {
|
||||
if !found {
|
||||
t.Errorf("ao refinery missing subcommand %q", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRefineryRun_HasIntervalFlag(t *testing.T) {
|
||||
if refineryRunCmd.Flags().Lookup("interval") == nil {
|
||||
t.Error("ao refinery run missing --interval flag")
|
||||
}
|
||||
}
|
||||
@@ -4170,6 +4170,41 @@ ao ready [flags]
|
||||
|
||||
---
|
||||
|
||||
### `ao refinery`
|
||||
|
||||
Watch main, run the full gate on each new commit, and on a
|
||||
|
||||
```
|
||||
ao refinery [command]
|
||||
```
|
||||
|
||||
**Subcommands:**
|
||||
|
||||
#### `ao refinery once`
|
||||
|
||||
Evaluate main HEAD once
|
||||
|
||||
```
|
||||
ao refinery once [flags]
|
||||
```
|
||||
|
||||
#### `ao refinery run`
|
||||
|
||||
Run the refinery loop until interrupted
|
||||
|
||||
```
|
||||
ao refinery run [flags]
|
||||
```
|
||||
|
||||
**Flags:**
|
||||
|
||||
```
|
||||
-h, --help help for run
|
||||
--interval duration poll interval (default 5m0s)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### `ao registry`
|
||||
|
||||
Query the unified registry
|
||||
|
||||
@@ -0,0 +1,167 @@
|
||||
package refinery
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/boshu2/agentops/cli/internal/gates"
|
||||
"github.com/boshu2/agentops/cli/internal/ports"
|
||||
)
|
||||
|
||||
// NewProduction wires the refinery with git/bd/gates-backed adapters rooted at
|
||||
// repoRoot. The registry is gates.Default (the seed registry).
|
||||
func NewProduction(repoRoot string) *Refinery {
|
||||
runner := gates.NewScriptRunner(repoRoot)
|
||||
return &Refinery{
|
||||
Commits: &gitCommitSource{repoRoot: repoRoot},
|
||||
Gate: &gatesChecker{repoRoot: repoRoot, runner: runner},
|
||||
Rerun: &gatesRerunner{repoRoot: repoRoot, runner: runner},
|
||||
Beads: &bdBeadFiler{repoRoot: repoRoot},
|
||||
Beacon: &fileBeacon{repoRoot: repoRoot},
|
||||
Store: &fileStateStore{path: filepath.Join(repoRoot, ".refinery-state")},
|
||||
RerunN: 3,
|
||||
Log: func(s string) { fmt.Fprintln(os.Stderr, s) },
|
||||
}
|
||||
}
|
||||
|
||||
// --- CommitSource: origin/main HEAD via git ---
|
||||
|
||||
type gitCommitSource struct{ repoRoot string }
|
||||
|
||||
func (g *gitCommitSource) MainHead(ctx context.Context) (string, error) {
|
||||
_, _ = run(ctx, g.repoRoot, "git", "fetch", "origin", "main", "--quiet") // best-effort
|
||||
out, err := run(ctx, g.repoRoot, "git", "rev-parse", "origin/main")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return strings.TrimSpace(out), nil
|
||||
}
|
||||
|
||||
// --- GateChecker: the full gate over gates.Default ---
|
||||
|
||||
type gatesChecker struct {
|
||||
repoRoot string
|
||||
runner ports.GateRunnerPort
|
||||
}
|
||||
|
||||
func (g *gatesChecker) CheckFull(ctx context.Context) (*gates.Report, error) {
|
||||
o := gates.NewOrchestrator(gates.Default, g.runner, gates.NewGitChangedFiles(g.repoRoot), g.repoRoot)
|
||||
return o.Run(ctx, gates.RunOptions{Mode: gates.Full})
|
||||
}
|
||||
|
||||
// --- Rerunner: re-run one check by ID ---
|
||||
|
||||
type gatesRerunner struct {
|
||||
repoRoot string
|
||||
runner ports.GateRunnerPort
|
||||
}
|
||||
|
||||
func (g *gatesRerunner) Rerun(ctx context.Context, checkID string) (ports.GateVerdict, error) {
|
||||
c, ok := gates.Default.Get(checkID)
|
||||
if !ok {
|
||||
return ports.GateVerdict{}, fmt.Errorf("refinery: unknown check %q", checkID)
|
||||
}
|
||||
if c.Run != nil {
|
||||
return c.Run(ctx, gates.RunContext{RepoRoot: g.repoRoot, Mode: gates.Full})
|
||||
}
|
||||
return g.runner.Run(ctx, ports.GateRunRequest{Name: ports.GateName(c.Backing)})
|
||||
}
|
||||
|
||||
// --- BeadFiler: bd create ---
|
||||
|
||||
type bdBeadFiler struct{ repoRoot string }
|
||||
|
||||
func (b *bdBeadFiler) FileFixBead(ctx context.Context, sha string, checks []string) (string, error) {
|
||||
short := sha
|
||||
if len(short) > 8 {
|
||||
short = short[:8]
|
||||
}
|
||||
title := fmt.Sprintf("fix: main %s poisoned — deterministic gate failure (%s)", short, strings.Join(checks, ", "))
|
||||
out, err := run(ctx, b.repoRoot, "bd", "create", title, "--type", "task", "--labels", "refinery,blocking", "--json")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
var parsed struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if jerr := json.Unmarshal([]byte(out), &parsed); jerr != nil {
|
||||
return "", nil // bead may have been created; ID just unparsed
|
||||
}
|
||||
return parsed.ID, nil
|
||||
}
|
||||
|
||||
// --- Beacon: status file + best-effort git note ---
|
||||
|
||||
type fileBeacon struct{ repoRoot string }
|
||||
|
||||
type poisonFile struct {
|
||||
SHA string `json:"sha"`
|
||||
Checks []string `json:"checks"`
|
||||
}
|
||||
|
||||
func (b *fileBeacon) path() string { return filepath.Join(b.repoRoot, ".refinery-poison") }
|
||||
|
||||
func (b *fileBeacon) Set(ctx context.Context, sha string, checks []string) error {
|
||||
data, err := json.MarshalIndent(poisonFile{SHA: sha, Checks: checks}, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.WriteFile(b.path(), data, 0o644); err != nil { // #nosec G306 -- a beacon meant to be world-readable by any pusher
|
||||
return err
|
||||
}
|
||||
// best-effort git note so the poison travels with the commit
|
||||
_, _ = run(ctx, b.repoRoot, "git", "notes", "--ref=refinery", "add", "-f",
|
||||
"-m", "POISON: "+strings.Join(checks, ", "), sha)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *fileBeacon) Clear(ctx context.Context, sha string) error {
|
||||
if err := os.Remove(b.path()); err != nil && !os.IsNotExist(err) {
|
||||
return err
|
||||
}
|
||||
_, _ = run(ctx, b.repoRoot, "git", "notes", "--ref=refinery", "remove", sha)
|
||||
return nil
|
||||
}
|
||||
|
||||
// --- StateStore: JSON file ---
|
||||
|
||||
type fileStateStore struct{ path string }
|
||||
|
||||
func (s *fileStateStore) Load() (State, error) {
|
||||
data, err := os.ReadFile(s.path)
|
||||
if os.IsNotExist(err) {
|
||||
return State{}, nil
|
||||
}
|
||||
if err != nil {
|
||||
return State{}, err
|
||||
}
|
||||
var st State
|
||||
if err := json.Unmarshal(data, &st); err != nil {
|
||||
return State{}, fmt.Errorf("refinery: parse state %s: %w", s.path, err)
|
||||
}
|
||||
return st, nil
|
||||
}
|
||||
|
||||
func (s *fileStateStore) Save(st State) error {
|
||||
data, err := json.MarshalIndent(st, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(s.path, data, 0o644) // #nosec G306 -- non-secret refinery state
|
||||
}
|
||||
|
||||
// run executes a command in dir and returns combined stdout.
|
||||
func run(ctx context.Context, dir, name string, args ...string) (string, error) {
|
||||
cmd := exec.CommandContext(ctx, name, args...)
|
||||
cmd.Dir = dir
|
||||
out, err := cmd.Output()
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("%s %s: %w", name, strings.Join(args, " "), err)
|
||||
}
|
||||
return string(out), nil
|
||||
}
|
||||
@@ -0,0 +1,221 @@
|
||||
// Package refinery is the bushido continuous-validation backstop (ag-qidx P2).
|
||||
// It watches main, runs the full gate on each new commit, and on a DETERMINISTIC
|
||||
// blocking failure raises a poison-main beacon + files a fix-bead + alerts —
|
||||
// it NEVER blind-reverts (the repo's 18-30% flake rate would make auto-revert
|
||||
// fight developers). It is a backstop, not a gatekeeper: if it is down, merges
|
||||
// still succeed; it resumes from its state file on restart.
|
||||
package refinery
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/boshu2/agentops/cli/internal/gates"
|
||||
"github.com/boshu2/agentops/cli/internal/ports"
|
||||
)
|
||||
|
||||
// State is the durable refinery state (persisted as .refinery-state JSON).
|
||||
type State struct {
|
||||
// LastCheckedSHA is the most recent main HEAD the refinery has evaluated.
|
||||
LastCheckedSHA string `json:"last_checked_sha"`
|
||||
// Poison lists the currently-poisoned commits (deterministic failures not
|
||||
// yet fixed forward).
|
||||
Poison []PoisonEntry `json:"poison"`
|
||||
}
|
||||
|
||||
// PoisonEntry records a deterministic failure on main.
|
||||
type PoisonEntry struct {
|
||||
SHA string `json:"sha"`
|
||||
Checks []string `json:"checks"`
|
||||
FixBead string `json:"fix_bead,omitempty"`
|
||||
}
|
||||
|
||||
// ---- ports (injected; production adapters in adapters.go, fakes in tests) ----
|
||||
|
||||
// CommitSource reports the current main HEAD.
|
||||
type CommitSource interface {
|
||||
MainHead(ctx context.Context) (string, error)
|
||||
}
|
||||
|
||||
// GateChecker runs the full gate and returns the report.
|
||||
type GateChecker interface {
|
||||
CheckFull(ctx context.Context) (*gates.Report, error)
|
||||
}
|
||||
|
||||
// Rerunner re-runs a single check (for flaky-vs-deterministic classification).
|
||||
type Rerunner interface {
|
||||
Rerun(ctx context.Context, checkID string) (ports.GateVerdict, error)
|
||||
}
|
||||
|
||||
// BeadFiler files a blocking fix-bead and returns its ID.
|
||||
type BeadFiler interface {
|
||||
FileFixBead(ctx context.Context, sha string, checks []string) (string, error)
|
||||
}
|
||||
|
||||
// Beacon marks/clears a poisoned main commit so pushers can see it.
|
||||
type Beacon interface {
|
||||
Set(ctx context.Context, sha string, checks []string) error
|
||||
Clear(ctx context.Context, sha string) error
|
||||
}
|
||||
|
||||
// StateStore loads and persists refinery State.
|
||||
type StateStore interface {
|
||||
Load() (State, error)
|
||||
Save(State) error
|
||||
}
|
||||
|
||||
// Refinery is the backstop engine.
|
||||
type Refinery struct {
|
||||
Commits CommitSource
|
||||
Gate GateChecker
|
||||
Rerun Rerunner
|
||||
Beads BeadFiler
|
||||
Beacon Beacon
|
||||
Store StateStore
|
||||
// RerunN is how many times a failing check is re-run to classify it as
|
||||
// deterministic (fails every time) vs flaky. Default 3 if zero.
|
||||
RerunN int
|
||||
// Log receives human-readable progress (optional).
|
||||
Log func(string)
|
||||
}
|
||||
|
||||
// Result summarizes one RunOnce tick.
|
||||
type Result struct {
|
||||
SHA string
|
||||
Skipped bool // HEAD unchanged since last check
|
||||
Green bool // no blocking failures
|
||||
Failing []string // all blocking-failed check IDs
|
||||
Deterministic []string // the subset that reproduced (escalated)
|
||||
FixBead string
|
||||
}
|
||||
|
||||
func (r *Refinery) logf(format string, a ...any) {
|
||||
if r.Log != nil {
|
||||
r.Log(fmt.Sprintf(format, a...))
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Refinery) rerunCount() int {
|
||||
if r.RerunN > 0 {
|
||||
return r.RerunN
|
||||
}
|
||||
return 3
|
||||
}
|
||||
|
||||
// RunOnce evaluates the current main HEAD if it has advanced. On a deterministic
|
||||
// blocking failure it sets a beacon and files a fix-bead; on green it clears any
|
||||
// beacon. It NEVER reverts.
|
||||
func (r *Refinery) RunOnce(ctx context.Context) (Result, error) {
|
||||
st, err := r.Store.Load()
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("refinery: load state: %w", err)
|
||||
}
|
||||
head, err := r.Commits.MainHead(ctx)
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("refinery: main head: %w", err)
|
||||
}
|
||||
if head == "" {
|
||||
return Result{}, fmt.Errorf("refinery: empty main HEAD")
|
||||
}
|
||||
if head == st.LastCheckedSHA {
|
||||
return Result{SHA: head, Skipped: true}, nil
|
||||
}
|
||||
|
||||
r.logf("refinery: evaluating %s", head)
|
||||
report, err := r.Gate.CheckFull(ctx)
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("refinery: gate check: %w", err)
|
||||
}
|
||||
|
||||
failing := blockingFailures(report)
|
||||
if len(failing) == 0 {
|
||||
if err := r.Beacon.Clear(ctx, head); err != nil {
|
||||
r.logf("refinery: beacon clear failed: %v", err)
|
||||
}
|
||||
st.LastCheckedSHA = head
|
||||
st.Poison = nil
|
||||
if err := r.Store.Save(st); err != nil {
|
||||
return Result{}, fmt.Errorf("refinery: save state: %w", err)
|
||||
}
|
||||
r.logf("refinery: %s GREEN", head)
|
||||
return Result{SHA: head, Green: true}, nil
|
||||
}
|
||||
|
||||
// Classify: only escalate failures that reproduce deterministically.
|
||||
var deterministic []string
|
||||
for _, id := range failing {
|
||||
if r.isDeterministic(ctx, id) {
|
||||
deterministic = append(deterministic, id)
|
||||
} else {
|
||||
r.logf("refinery: %s failed on %s but did not reproduce — treating as flaky, NOT escalating", head, id)
|
||||
}
|
||||
}
|
||||
|
||||
res := Result{SHA: head, Failing: failing, Deterministic: deterministic}
|
||||
if len(deterministic) > 0 {
|
||||
bead, ferr := r.Beads.FileFixBead(ctx, head, deterministic)
|
||||
if ferr != nil {
|
||||
r.logf("refinery: file fix-bead failed: %v", ferr)
|
||||
}
|
||||
res.FixBead = bead
|
||||
if err := r.Beacon.Set(ctx, head, deterministic); err != nil {
|
||||
r.logf("refinery: beacon set failed: %v", err)
|
||||
}
|
||||
st.Poison = append(st.Poison, PoisonEntry{SHA: head, Checks: deterministic, FixBead: bead})
|
||||
r.logf("refinery: %s POISONED by %v — fix-bead %s filed (no revert)", head, deterministic, bead)
|
||||
}
|
||||
|
||||
st.LastCheckedSHA = head
|
||||
if err := r.Store.Save(st); err != nil {
|
||||
return Result{}, fmt.Errorf("refinery: save state: %w", err)
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// isDeterministic re-runs a failing check RerunN times; it is deterministic only
|
||||
// if it FAILS every time (any pass => flaky, do not escalate).
|
||||
func (r *Refinery) isDeterministic(ctx context.Context, checkID string) bool {
|
||||
for i := 0; i < r.rerunCount(); i++ {
|
||||
v, err := r.Rerun.Rerun(ctx, checkID)
|
||||
if err != nil {
|
||||
// Could not re-run -> conservatively treat as NOT deterministic
|
||||
// (don't escalate on inability to reproduce).
|
||||
return false
|
||||
}
|
||||
if v.Status != ports.GateStatusFail {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Loop runs RunOnce every interval until ctx is cancelled. It is a BACKSTOP: a
|
||||
// RunOnce error (transient git/gate/network failure, bushido hiccup) is logged
|
||||
// and the loop continues — the daemon never dies on one bad tick, and resumes
|
||||
// from its state file across restarts.
|
||||
func (r *Refinery) Loop(ctx context.Context, interval time.Duration) error {
|
||||
t := time.NewTicker(interval)
|
||||
defer t.Stop()
|
||||
for {
|
||||
if _, err := r.RunOnce(ctx); err != nil {
|
||||
r.logf("refinery: tick error (continuing): %v", err)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-t.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// blockingFailures returns the IDs of blocking checks that FAILed.
|
||||
func blockingFailures(report *gates.Report) []string {
|
||||
var out []string
|
||||
for _, res := range report.Results {
|
||||
if res.Check.Blocking && res.Verdict.Status == ports.GateStatusFail {
|
||||
out = append(out, res.Check.ID)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,170 @@
|
||||
package refinery
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/boshu2/agentops/cli/internal/gates"
|
||||
"github.com/boshu2/agentops/cli/internal/ports"
|
||||
)
|
||||
|
||||
// ---- fakes ----
|
||||
|
||||
type fakeCommits struct{ head string }
|
||||
|
||||
func (f fakeCommits) MainHead(context.Context) (string, error) { return f.head, nil }
|
||||
|
||||
type fakeGate struct{ rep *gates.Report }
|
||||
|
||||
func (f fakeGate) CheckFull(context.Context) (*gates.Report, error) { return f.rep, nil }
|
||||
|
||||
type fakeRerun struct{ status ports.GateStatus }
|
||||
|
||||
func (f fakeRerun) Rerun(context.Context, string) (ports.GateVerdict, error) {
|
||||
return ports.GateVerdict{Status: f.status}, nil
|
||||
}
|
||||
|
||||
type fakeBeads struct {
|
||||
filed int
|
||||
lastChecks []string
|
||||
}
|
||||
|
||||
func (f *fakeBeads) FileFixBead(_ context.Context, _ string, checks []string) (string, error) {
|
||||
f.filed++
|
||||
f.lastChecks = checks
|
||||
return "ag-fix1", nil
|
||||
}
|
||||
|
||||
type fakeBeacon struct{ setN, clearN int }
|
||||
|
||||
func (f *fakeBeacon) Set(context.Context, string, []string) error { f.setN++; return nil }
|
||||
func (f *fakeBeacon) Clear(context.Context, string) error { f.clearN++; return nil }
|
||||
|
||||
type memStore struct{ st State }
|
||||
|
||||
func (m *memStore) Load() (State, error) { return m.st, nil }
|
||||
func (m *memStore) Save(s State) error { m.st = s; return nil }
|
||||
|
||||
// ---- report builders ----
|
||||
|
||||
func blockingCheck(id string, status ports.GateStatus) gates.CheckResult {
|
||||
return gates.CheckResult{
|
||||
Check: gates.Check{ID: id, Tiers: gates.Full, Blocking: true, Backing: "x"},
|
||||
Verdict: ports.GateVerdict{Status: status},
|
||||
}
|
||||
}
|
||||
|
||||
func newRefinery(head string, rep *gates.Report, rerun ports.GateStatus, store *memStore, beads *fakeBeads, beacon *fakeBeacon) *Refinery {
|
||||
return &Refinery{
|
||||
Commits: fakeCommits{head: head},
|
||||
Gate: fakeGate{rep: rep},
|
||||
Rerun: fakeRerun{status: rerun},
|
||||
Beads: beads,
|
||||
Beacon: beacon,
|
||||
Store: store,
|
||||
RerunN: 3,
|
||||
}
|
||||
}
|
||||
|
||||
// ---- tests ----
|
||||
|
||||
func TestRunOnce_SkipsUnchangedHead(t *testing.T) {
|
||||
store := &memStore{st: State{LastCheckedSHA: "abc"}}
|
||||
r := newRefinery("abc", &gates.Report{}, ports.GateStatusFail, store, &fakeBeads{}, &fakeBeacon{})
|
||||
res, err := r.RunOnce(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("RunOnce: %v", err)
|
||||
}
|
||||
if !res.Skipped {
|
||||
t.Error("unchanged HEAD should be Skipped")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunOnce_GreenClearsBeacon(t *testing.T) {
|
||||
store := &memStore{st: State{LastCheckedSHA: "old", Poison: []PoisonEntry{{SHA: "old"}}}}
|
||||
beacon := &fakeBeacon{}
|
||||
rep := &gates.Report{Results: []gates.CheckResult{blockingCheck("go.build", ports.GateStatusPass)}}
|
||||
r := newRefinery("new", rep, ports.GateStatusPass, store, &fakeBeads{}, beacon)
|
||||
res, err := r.RunOnce(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("RunOnce: %v", err)
|
||||
}
|
||||
if !res.Green {
|
||||
t.Error("all-pass report should be Green")
|
||||
}
|
||||
if beacon.clearN != 1 {
|
||||
t.Errorf("beacon.Clear calls = %d, want 1", beacon.clearN)
|
||||
}
|
||||
if len(store.st.Poison) != 0 {
|
||||
t.Errorf("green should clear poison; got %v", store.st.Poison)
|
||||
}
|
||||
if store.st.LastCheckedSHA != "new" {
|
||||
t.Errorf("LastCheckedSHA = %q, want new", store.st.LastCheckedSHA)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunOnce_DeterministicFailEscalates(t *testing.T) {
|
||||
store := &memStore{}
|
||||
beads := &fakeBeads{}
|
||||
beacon := &fakeBeacon{}
|
||||
rep := &gates.Report{Results: []gates.CheckResult{blockingCheck("contract.registry-drift", ports.GateStatusFail)}}
|
||||
// rerun always FAILs -> deterministic
|
||||
r := newRefinery("bad", rep, ports.GateStatusFail, store, beads, beacon)
|
||||
res, err := r.RunOnce(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("RunOnce: %v", err)
|
||||
}
|
||||
if len(res.Deterministic) != 1 || res.Deterministic[0] != "contract.registry-drift" {
|
||||
t.Errorf("Deterministic = %v, want [contract.registry-drift]", res.Deterministic)
|
||||
}
|
||||
if beads.filed != 1 {
|
||||
t.Errorf("fix-bead filed = %d, want 1", beads.filed)
|
||||
}
|
||||
if beacon.setN != 1 {
|
||||
t.Errorf("beacon.Set calls = %d, want 1", beacon.setN)
|
||||
}
|
||||
if len(store.st.Poison) != 1 || store.st.Poison[0].FixBead != "ag-fix1" {
|
||||
t.Errorf("poison state = %+v, want one entry with fix bead", store.st.Poison)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunOnce_FlakyFailDoesNotEscalate(t *testing.T) {
|
||||
store := &memStore{}
|
||||
beads := &fakeBeads{}
|
||||
beacon := &fakeBeacon{}
|
||||
rep := &gates.Report{Results: []gates.CheckResult{blockingCheck("skill.schema", ports.GateStatusFail)}}
|
||||
// rerun PASSes -> not reproducible -> flaky -> no escalation
|
||||
r := newRefinery("flaky", rep, ports.GateStatusPass, store, beads, beacon)
|
||||
res, err := r.RunOnce(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("RunOnce: %v", err)
|
||||
}
|
||||
if len(res.Deterministic) != 0 {
|
||||
t.Errorf("flaky failure must NOT escalate; Deterministic = %v", res.Deterministic)
|
||||
}
|
||||
if beads.filed != 0 {
|
||||
t.Errorf("no fix-bead for flaky; filed = %d", beads.filed)
|
||||
}
|
||||
if beacon.setN != 0 {
|
||||
t.Errorf("no beacon for flaky; setN = %d", beacon.setN)
|
||||
}
|
||||
if len(store.st.Poison) != 0 {
|
||||
t.Errorf("flaky must not poison; got %v", store.st.Poison)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunOnce_NeverRevertsField(t *testing.T) {
|
||||
// Structural guard: the Refinery type exposes no revert capability — there is
|
||||
// no Revert method/field. A deterministic failure escalates via beacon+bead
|
||||
// only. (Compile-time evidence: this test references the public surface.)
|
||||
store := &memStore{}
|
||||
rep := &gates.Report{Results: []gates.CheckResult{blockingCheck("x", ports.GateStatusFail)}}
|
||||
r := newRefinery("z", rep, ports.GateStatusFail, store, &fakeBeads{}, &fakeBeacon{})
|
||||
if _, err := r.RunOnce(context.Background()); err != nil {
|
||||
t.Fatalf("RunOnce: %v", err)
|
||||
}
|
||||
// The poisoned commit remains on main (state records it; nothing reverted).
|
||||
if store.st.LastCheckedSHA != "z" {
|
||||
t.Errorf("LastCheckedSHA = %q, want z (commit stays; backstop never reverts)", store.st.LastCheckedSHA)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
# AgentOps Refinery — continuous main-validation backstop (ag-qidx P2.4).
|
||||
# Install on bushido as a user service:
|
||||
# cp deploy/agentops-refinery.service ~/.config/systemd/user/
|
||||
# systemctl --user daemon-reload
|
||||
# systemctl --user enable --now agentops-refinery
|
||||
# Backstop-not-gatekeeper: if this is down, merges still succeed; on restart it
|
||||
# resumes from .refinery-state. It NEVER reverts — it beacons + files fix-beads.
|
||||
[Unit]
|
||||
Description=AgentOps Refinery (continuous main-validation backstop)
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
WorkingDirectory=%h/dev/agentops
|
||||
ExecStart=%h/go/bin/ao refinery run --interval 5m
|
||||
Restart=on-failure
|
||||
RestartSec=30
|
||||
# Resilience: a single bad tick must not kill the daemon (the loop already
|
||||
# swallows tick errors; this is belt-and-suspenders).
|
||||
StartLimitIntervalSec=0
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
@@ -0,0 +1,54 @@
|
||||
# Bushido Refinery — continuous main-validation backstop
|
||||
|
||||
The refinery (`ao refinery`) is the **backstop** half of the push-to-main model
|
||||
(ag-qidx). Push-to-main makes the local pre-push gate the pre-merge wall; the
|
||||
refinery is the always-on net behind `main` on bushido.
|
||||
|
||||
## What it does
|
||||
|
||||
Every tick it checks `origin/main`. On a **new** commit it runs the full gate
|
||||
(`ao gate check --full`). On a **blocking** failure it:
|
||||
|
||||
1. re-runs each failing check N times (default 3) to tell **deterministic** from
|
||||
**flaky** (the repo's 18–30% flake rate means naive escalation would be noise);
|
||||
2. for deterministic failures only: writes a **poison beacon** (`.refinery-poison`
|
||||
+ a `refinery` git note on the bad SHA) and files a **blocking fix-bead**
|
||||
(`bd create --labels refinery,blocking`);
|
||||
3. on green: clears the beacon.
|
||||
|
||||
It **never reverts.** A poisoned commit stays on `main`; the team fixes forward.
|
||||
Revert remains a human/quorum decision (ag-qidx P2.5, deferred to the quorum
|
||||
infra `ag-k99u`).
|
||||
|
||||
## Backstop, not gatekeeper
|
||||
|
||||
If bushido (or its Wi-Fi) is down, the refinery is simply blind — **merges still
|
||||
succeed**, nothing blocks. On restart it resumes from `.refinery-state`
|
||||
(`last_checked_sha`) and catches up. This is the deliberate posture after the
|
||||
2026-06-05 control-plane crash: no single host is in the merge path.
|
||||
|
||||
## Run it
|
||||
|
||||
```bash
|
||||
ao refinery once # one tick (manual / cron)
|
||||
ao refinery run --interval 5m # loop (the systemd service)
|
||||
```
|
||||
|
||||
## Install on bushido
|
||||
|
||||
```bash
|
||||
cp deploy/agentops-refinery.service ~/.config/systemd/user/
|
||||
systemctl --user daemon-reload
|
||||
systemctl --user enable --now agentops-refinery
|
||||
systemctl --user status agentops-refinery
|
||||
journalctl --user -u agentops-refinery -f # logs
|
||||
```
|
||||
|
||||
State + beacon (repo-root, gitignored runtime):
|
||||
- `.refinery-state` — `{last_checked_sha, poison[]}`
|
||||
- `.refinery-poison` — present iff `main` is currently poisoned
|
||||
|
||||
## Tuning
|
||||
|
||||
- `--interval` — poll cadence (default 5m).
|
||||
- Re-run count is 3 (deterministic = fails all 3). Raise for noisier suites.
|
||||
Reference in New Issue
Block a user