amadan

carlos / platform Public

moves: quiet the evidence's expected refusals; page an anomaly once (Plan 5 Task 9a, step 6)

The claim's regional worker asks Task 5's provider about every host at
rest, every minute, and ordinary states refuse. movesRefusals keeps them
quiet:

- report, the regional worker's report with both switches on: a provider
  refusal (MovesSweepRefusal, NotMovesEvidence) or a named dependency
  (ClaimDependencyError: a free host's lanes, claim Plan 12's guard) is
  logged at most once an hour per host and reason; any other error is
  logged every time, as before. With either switch off the report is the
  plain log line it was.
- evidence wraps the provider the agent registers (movesQuietEvidence): a
  refusal marked Anomaly pages once per host and reason through the box
  pager (PageOperator), as a warning; the worker gets every error unchanged.

Copy: MV-N336 and MV-N337 (pending), the anomaly page's subject and body;
movesrefusals.go joins the copy scan, movesAnomalyPage its returned
sentences.

Tests: TestExpectedMovesRefusalsAreRateLimited, TestMovesAnomalyPagesOnce,
TestRegionalWorkerRefusalsAreQuiet (through the real RegionalWorker).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 34b856d2a3a1ff6c77fb9305ccf7f9f69d1d9cb1 parent a927e15
7 files changed, +346 −8
  • approved_moves_copy_test.go +4 −1
  • docs/superpowers/plans/2026-10-06-moves-phase1-copy.md +7 −0
  • internal/host/agent.go +10 −2
  • internal/host/movesrefusals.go +173 −0
  • internal/host/movesrefusals_test.go +144 −0
  • internal/host/movessources.go +6 −3
  • internal/host/movessources_test.go +2 −2
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index 72bd4af23..746e2bd93 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -76,7 +76,7 @@ var movesCopyIDs = []string{
"MV-N313", "MV-N314",
"MV-N315", "MV-N316", "MV-N317", "MV-N318",
"MV-N325", "MV-N326", "MV-N327", "MV-N328", "MV-N329", "MV-N330", "MV-N331", "MV-N332", "MV-N333", "MV-N334",
- "MV-N335",
+ "MV-N335", "MV-N336", "MV-N337",
}
// copyRow is one row of the record. The File cell is the file the sentence is
@@ -546,6 +546,7 @@ func TestMovesTextAppliedPlaceholders(t *testing.T) {
// fails when the list falls behind.
var movesScanFiles = []string{
"internal/host/movessources.go",
+ "internal/host/movesrefusals.go",
"internal/console/movesrecovery.go",
"internal/console/claim_runtime.go",
"internal/alerting/alerting.go",
@@ -752,6 +753,8 @@ var movesReturnedSentences = map[string]map[string]bool{
"internal/host/movesreport.go": {"t1Below": true, "t3Differs": true, "t3Shadowed": true, "t4Failed": true,
"t6NoRoom": true, "t8Count": true, "t12NotQuiet": true, "replicaReason": true},
"internal/host/movesrollback.go": {"stuckPage": true},
+ // The moves evidence's anomaly page (Plan 5 Task 9a, MV-N336 and MV-N337).
+ "internal/host/movesrefusals.go": {"movesAnomalyPage": true},
// The console's move pass pages (Plan 5 Task 7, MV-N325 to MV-N334).
"internal/console/movesrecovery.go": {"moveWaitingPage": true, "movePreparingPageText": true, "moveOrphanPage": true, "moveCloseOverduePage": true, "moveAbortOverduePage": true},
}
diff --git a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
index 17e095ec5..d70407767 100644
--- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
+++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
@@ -292,6 +292,13 @@ Task 9a's rows (the production park wiring). MV-N335 is the stuck reason a rollb
|---|---|---|---|
| MV-N335 | internal/provision/provision.go | the target's files cannot be archived now that it has stopped | pending |
+The anomaly page (Task 9a's quiet refusals, `internal/host/movesrefusals.go`): a box pages once when the moves evidence for a host's at-rest cleanup contradicts itself. A warning; the reason stays in the agent's log.
+
+| id | file | text | status |
+|---|---|---|---|
+| MV-N336 | internal/host/movesrefusals.go | carlos move cleanup refused: <host> | pending |
+| MV-N337 | internal/host/movesrefusals.go | The cleanup of <host> on <box> stopped: its move history does not add up, or something wrote where no move could. Nothing of <host> is cleaned up in this region until an operator checks the agent's log on <box>. | pending |
+
## Floor gate fix (#267 review)
(root's floor gate for the system units, `carlos system floor-allows-prev`; its stderr reaches the unit's journal)
diff --git a/internal/host/agent.go b/internal/host/agent.go
index f994918e1..a03febbea 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -187,6 +187,7 @@ import (
"github.com/carlosframework/platform/internal/instancefs"
"github.com/carlosframework/platform/internal/layout"
"github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
@@ -279,7 +280,10 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// The moves park providers (Plan 5 Task 9a) register before the claim
// assembly reads them; only with both switches on, and behind the claim
// Plan 12 guard, so the parks still refuse by name until it lands.
- if err := registerProductionMovesParks(claimCtx, claimEntryEnv(claimCtx), claimStore(claimCtx), *cfg); err != nil {
+ // Task 5's expected refusals are quiet in the regional worker's log and
+ // its anomalies page once (movesrefusals.go), with both switches on.
+ movesRefusalLog := newMovesRefusals(metricsBoxLabel(os.Getenv, readMachineID()), PageOperator, log.Printf, time.Now)
+ if err := registerProductionMovesParks(claimCtx, claimEntryEnv(claimCtx), claimStore(claimCtx), *cfg, movesRefusalLog); err != nil {
log.Printf("agent: moves: the park providers are not registered: %v", err)
}
claims, err := newClaimWiring(claimCtx, claimEntryEnv(claimCtx), buildClaims)
@@ -584,7 +588,11 @@ func runAgent(args []string, buildClaims ClaimFactory) {
sd.runner.Start(context.Background()) // runs until the process ends, like every loop below
}
serialClaims := make(chan struct{}, 1)
- if err := claims.startRegional(agentCtx, serialClaims, func(err error) { log.Printf("agent: regional claims: %v", err) }); err != nil {
+ regionalReport := func(err error) { log.Printf("agent: regional claims: %v", err) }
+ if hostwrite.Enabled(claimEntryEnv(claimCtx)) && moves.Enabled(os.Getenv) {
+ regionalReport = movesRefusalLog.report
+ }
+ if err := claims.startRegional(agentCtx, serialClaims, regionalReport); err != nil {
log.Fatalf("agent: regional claims: %v", err)
}
defer func() {
diff --git a/internal/host/movesrefusals.go b/internal/host/movesrefusals.go
new file mode 100644
index 000000000..d8d74eb9a
--- /dev/null
+++ b/internal/host/movesrefusals.go
@@ -0,0 +1,173 @@
+package host
+
+// Quiet refusals (moves Plan 5 Task 9a; the Task 5 rulings). The claim's
+// regional worker asks Task 5's provider about every host at rest, every
+// minute, and an ordinary state refuses: a live move, residue the move's
+// close does not vouch for, a free host whose lanes are claim Plan 12's. The
+// worker reports each refusal, so the agent's log would repeat them every
+// minute. movesRefusals keeps them quiet:
+//
+// - report is the worker's report: a provider refusal (MovesSweepRefusal,
+// NotMovesEvidence) or a named dependency (ClaimDependencyError) is
+// logged at most once an hour per host and reason; anything else is
+// logged every time, as before;
+// - evidence wraps the provider: a refusal marked Anomaly (the history
+// contradicts itself, or something wrote where no move could) pages once
+// per host and reason through the box pager, as a warning. The worker
+// gets every error unchanged.
+//
+// Both are wired only with CARLOS_MOVES_PHASE1 and CARLOS_HOST_CLAIM on;
+// otherwise the worker's report is the plain log line it always was.
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/alerting"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/park"
+)
+
+// movesRefusalLogEvery is how often one expected refusal is logged.
+const movesRefusalLogEvery = time.Hour
+
+// movesRefusals is the agent's refusal log and anomaly pager for the moves
+// evidence. Its memory is the process's: a restarted agent logs and pages
+// each standing condition once more.
+type movesRefusals struct {
+ box string
+ page func(ctx context.Context, level, subject, body string) error
+ logf func(format string, args ...any)
+ now func() time.Time
+
+ mu sync.Mutex
+ logged map[string]time.Time
+ paged map[string]bool
+}
+
+func newMovesRefusals(box string, page func(context.Context, string, string, string) error, logf func(string, ...any), now func() time.Time) *movesRefusals {
+ return &movesRefusals{box: box, page: page, logf: logf, now: now, logged: map[string]time.Time{}, paged: map[string]bool{}}
+}
+
+// movesQuietEvidence is Task 5's provider with its anomalies paged. It is
+// the regional evidence and the sweep verifier the claim registers.
+type movesQuietEvidence struct {
+ movesRetirementEvidence
+ r *movesRefusals
+}
+
+var (
+ _ hostbox.RegionalEvidence = movesQuietEvidence{}
+ _ park.MoveSweepVerifier = movesQuietEvidence{}
+)
+
+func (r *movesRefusals) evidence(e movesRetirementEvidence) movesQuietEvidence {
+ return movesQuietEvidence{movesRetirementEvidence: e, r: r}
+}
+
+func (e movesQuietEvidence) VerifyRetirement(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (string, time.Time, error) {
+ inc, at, err := e.movesRetirementEvidence.VerifyRetirement(ctx, o, versions)
+ if err != nil {
+ e.r.observe(ctx, err)
+ }
+ return inc, at, err
+}
+
+func (e movesQuietEvidence) VerifyMovesSweep(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (park.MoveSweepProof, error) {
+ proof, err := e.movesRetirementEvidence.VerifyMovesSweep(ctx, o, versions)
+ if err != nil {
+ e.r.observe(ctx, err)
+ }
+ return proof, err
+}
+
+// observe pages an anomaly once per host and reason.
+func (r *movesRefusals) observe(ctx context.Context, err error) {
+ var refusal *MovesSweepRefusal
+ if r == nil || !errors.As(err, &refusal) || !refusal.Anomaly {
+ return
+ }
+ key := refusal.Host + "\x00" + refusal.Reason
+ r.mu.Lock()
+ seen := r.paged[key]
+ r.paged[key] = true
+ r.mu.Unlock()
+ if seen || r.page == nil {
+ return
+ }
+ subject, body := movesAnomalyPage(refusal.Host, r.box)
+ if perr := r.page(ctx, alerting.LevelWarning, subject, body); perr != nil && r.logf != nil {
+ r.logf("moves: %s: the page for the refused cleanup: %v", refusal.Host, perr)
+ }
+}
+
+// movesAnomalyPage is the anomaly's page: the subject (MV-N336) and the
+// body (MV-N337). The reason stays in the agent's log, never in the page.
+func movesAnomalyPage(host, box string) (subject, body string) {
+ subject = fmt.Sprintf("carlos move cleanup refused: %s", host)
+ body = fmt.Sprintf("The cleanup of %s on %s stopped: its move history does not add up, or something wrote where no move could. Nothing of %s is cleaned up in this region until an operator checks the agent's log on %s.",
+ host, box, host, box)
+ return subject, body
+}
+
+// report is the regional worker's report: each error of the pass, the
+// expected refusals at most once an hour per host and reason.
+func (r *movesRefusals) report(err error) {
+ for _, leaf := range joinedErrors(err) {
+ key, expected := movesExpectedRefusal(leaf)
+ if !expected {
+ r.logf("agent: regional claims: %v", leaf)
+ continue
+ }
+ now := r.now()
+ r.mu.Lock()
+ last, seen := r.logged[key]
+ if !seen || !now.Before(last.Add(movesRefusalLogEvery)) {
+ r.logged[key] = now
+ seen = false
+ }
+ r.mu.Unlock()
+ if !seen {
+ r.logf("agent: regional claims: %v (logged at most once an hour while it stands)", leaf)
+ }
+ }
+}
+
+// joinedErrors flattens errors.Join's tree into its leaves.
+func joinedErrors(err error) []error {
+ if err == nil {
+ return nil
+ }
+ j, ok := err.(interface{ Unwrap() []error })
+ if !ok {
+ return []error{err}
+ }
+ var out []error
+ for _, e := range j.Unwrap() {
+ out = append(out, joinedErrors(e)...)
+ }
+ return out
+}
+
+// movesExpectedRefusal names a refusal that stands until its condition
+// changes: the provider's typed refusals and a named dependency.
+func movesExpectedRefusal(err error) (string, bool) {
+ var refusal *MovesSweepRefusal
+ var notMoves *NotMovesEvidence
+ var dep *ClaimDependencyError
+ switch {
+ case errors.As(err, &refusal):
+ return "sweep\x00" + refusal.Host + "\x00" + refusal.Reason, true
+ case errors.As(err, &notMoves):
+ return "not-moves\x00" + notMoves.Host + "\x00" + notMoves.Generation + "\x00" + notMoves.Reason, true
+ case errors.As(err, &dep):
+ return "dependency\x00" + strings.Join(dep.Contracts, "\x00"), true
+ }
+ return "", false
+}
diff --git a/internal/host/movesrefusals_test.go b/internal/host/movesrefusals_test.go
new file mode 100644
index 000000000..ab24f46c0
--- /dev/null
+++ b/internal/host/movesrefusals_test.go
@@ -0,0 +1,144 @@
+package host
+
+import (
+ "errors"
+ "fmt"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/alerting"
+ "github.com/carlosframework/platform/internal/park"
+)
+
+// Quiet refusals (Plan 5 Task 9a): the Task 5 provider's expected refusals
+// are rate-limited in the regional worker's log, and an anomaly pages once
+// through the box pager.
+
+type refusalLog struct {
+ mu sync.Mutex
+ lines []string
+ now time.Time
+}
+
+func (l *refusalLog) logf(format string, args ...any) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ l.lines = append(l.lines, fmt.Sprintf(format, args...))
+}
+
+func (l *refusalLog) clock() time.Time {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return l.now
+}
+
+func (l *refusalLog) take() []string {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ out := l.lines
+ l.lines = nil
+ return out
+}
+
+// TestExpectedMovesRefusalsAreRateLimited: a live move, residue the close
+// does not vouch for and a free host (the provider's dependency refusal) are
+// each logged once, then not again within the hour; an unexpected error is
+// logged every time.
+func TestExpectedMovesRefusalsAreRateLimited(t *testing.T) {
+ l := &refusalLog{now: evNow}
+ var pages pagesSeen
+ r := newMovesRefusals("ie-box", pages.hook(), l.logf, l.clock)
+ boom := errors.New("listing failed")
+ pass := errors.Join(
+ &MovesSweepRefusal{Host: "a.example", Reason: "the record carries the live move mv_x"},
+ &NotMovesEvidence{Host: "b.example", Generation: "g", Reason: "written after the close"},
+ &ClaimDependencyError{Contracts: []string{movesLegacyCensus}},
+ boom,
+ )
+ r.report(pass)
+ if got := l.take(); len(got) != 4 {
+ t.Fatalf("the first pass logged %q", got)
+ }
+ l.now = evNow.Add(30 * time.Minute)
+ r.report(pass)
+ if got := l.take(); len(got) != 1 || !strings.Contains(got[0], boom.Error()) {
+ t.Fatalf("within the hour the pass logged %q; want only the unexpected error", got)
+ }
+ l.now = evNow.Add(61 * time.Minute)
+ r.report(pass)
+ if got := l.take(); len(got) != 4 {
+ t.Fatalf("after the hour the pass logged %q", got)
+ }
+ if len(pages) != 0 {
+ t.Fatalf("an expected refusal paged: %+v", pages)
+ }
+}
+
+// TestMovesAnomalyPagesOnce: an anomaly from either method pages once per
+// host and reason, as a warning naming the host and the box and carrying no
+// path; an expected refusal pages nothing, and the error reaches the worker
+// unchanged.
+func TestMovesAnomalyPagesOnce(t *testing.T) {
+ l := &refusalLog{now: evNow}
+ var pages pagesSeen
+ r := newMovesRefusals("ie-box", pages.hook(), l.logf, l.clock)
+ f := newEvFixture(t)
+ f.rec.CreatedAt = evCreated.Add(time.Hour) // a record of another incarnation: an anomaly
+ e := r.evidence(f.provider())
+ obs, census := f.observe(t), f.census(t)
+ for i := 0; i < 3; i++ {
+ _, _, err := e.VerifyRetirement(t.Context(), obs, census)
+ var refusal *MovesSweepRefusal
+ if !errors.As(err, &refusal) || !refusal.Anomaly {
+ t.Fatalf("the wrapper changed the refusal: %v", err)
+ }
+ if _, err := e.VerifyMovesSweep(t.Context(), obs, census); !errors.As(err, &refusal) {
+ t.Fatalf("the wrapper changed the sweep refusal: %v", err)
+ }
+ }
+ if len(pages) != 1 {
+ t.Fatalf("pages %+v, want one", pages)
+ }
+ subject, body := movesAnomalyPage(evHost, "ie-box")
+ if p := pages[0]; p.level != alerting.LevelWarning || p.subject != subject || p.body != body || strings.Contains(p.subject+p.body, "/") {
+ t.Fatalf("page %+v", p)
+ }
+ // An expected refusal of another host pages nothing.
+ g := newEvFixture(t)
+ g.rec.LastMove = nil
+ if _, _, err := r.evidence(g.provider()).VerifyRetirement(t.Context(), g.observe(t), g.census(t)); err == nil {
+ t.Fatal("no refusal")
+ }
+ if len(pages) != 1 {
+ t.Fatalf("an expected refusal paged: %+v", pages)
+ }
+}
+
+// TestRegionalWorkerRefusalsAreQuiet: through the claim's real regional
+// worker, an anomalous host refuses the sweep every pass, pages once and is
+// logged once within the hour.
+func TestRegionalWorkerRefusalsAreQuiet(t *testing.T) {
+ l := &refusalLog{now: evNow}
+ var pages pagesSeen
+ r := newMovesRefusals("ie-box", pages.hook(), l.logf, l.clock)
+ f := newEvFixture(t)
+ f.rec.CreatedAt = evCreated.Add(time.Hour)
+ f.lane(evMoveID, "move.json") // the census finds the host by its objects
+ w := f.worker(t, evNow)
+ e := r.evidence(f.provider())
+ w.Evidence = e
+ var verifier park.MoveSweepVerifier = e
+ w.Regional.Engine.VerifyMovesSweep = verifier.VerifyMovesSweep
+ for i := 0; i < 3; i++ {
+ err := w.Run(t.Context())
+ if err == nil {
+ t.Fatal("the sweep ran on an anomaly")
+ }
+ r.report(err)
+ }
+ if lines := l.take(); len(pages) != 1 || len(lines) != 1 {
+ t.Fatalf("pages %+v, log lines %q over three passes", pages, lines)
+ }
+}
diff --git a/internal/host/movessources.go b/internal/host/movessources.go
index 6743106f0..cbb19e6fd 100644
--- a/internal/host/movessources.go
+++ b/internal/host/movessources.go
@@ -546,12 +546,14 @@ func armMovesParks(ctx context.Context, cfg *moveCfg, providers ClaimProviders,
// bucket read by exact version (lease.NewVersionedS3, through rootaws) and
// its placement region, behind the claim Plan 12 guard. Off, it registers
// nothing and the claim keeps its named refusals.
-func registerProductionMovesParks(ctx context.Context, getenv func(string) string, store release.Store, hib activator.Config) error {
+// Task 5's evidence goes through refusals, which pages its anomalies once
+// (movesrefusals.go).
+func registerProductionMovesParks(ctx context.Context, getenv func(string) string, store release.Store, hib activator.Config, refusals *movesRefusals) error {
if !hostwrite.Enabled(getenv) || !moves.Enabled(getenv) {
return nil
}
region := strings.TrimSpace(getenv("CARLOS_REGION"))
- if ctx == nil || store == nil || hib.Bucket == "" || hib.Region == "" || region == "" {
+ if ctx == nil || store == nil || hib.Bucket == "" || hib.Region == "" || region == "" || refusals == nil {
return errors.Join(hostwrite.ErrUnavailable, errors.New("the moves park view needs the deployment store, the data bucket and CARLOS_REGION"))
}
data, err := lease.NewVersionedS3(ctx, map[string]string{hib.Bucket: hib.Region})
@@ -559,7 +561,8 @@ func registerProductionMovesParks(ctx context.Context, getenv func(string) strin
return err
}
records := MovesRecordsFrom(store)
- src, err := newMovesParkSources(records, data, hib.Bucket, region, movesRetirementEvidence{Records: records, Now: time.Now}, movesPlan12Parks)
+ evidence := refusals.evidence(movesRetirementEvidence{Records: records, Now: time.Now})
+ src, err := newMovesParkSources(records, data, hib.Bucket, region, evidence, movesPlan12Parks)
if err != nil {
return err
}
diff --git a/internal/host/movessources_test.go b/internal/host/movessources_test.go
index a012b4246..3e314e326 100644
--- a/internal/host/movessources_test.go
+++ b/internal/host/movessources_test.go
@@ -535,14 +535,14 @@ func TestRegisterProductionMovesParksNeedsBothSwitches(t *testing.T) {
"claims off": {moves.SwitchEnv: "on", "CARLOS_REGION": "ie"},
"moves off": {"CARLOS_HOST_CLAIM": "1", moves.SwitchEnv: "off", "CARLOS_REGION": "ie"},
} {
- if err := registerProductionMovesParks(t.Context(), env(vals), store, hib); err != nil {
+ if err := registerProductionMovesParks(t.Context(), env(vals), store, hib, newMovesRefusals("ie-box", nil, t.Logf, time.Now)); err != nil {
t.Fatalf("%s: %v", name, err)
}
if claimMoveSources.Parks != nil {
t.Fatalf("%s: registered", name)
}
}
- err := registerProductionMovesParks(t.Context(), env(map[string]string{"CARLOS_HOST_CLAIM": "1", moves.SwitchEnv: "on"}), store, hib)
+ err := registerProductionMovesParks(t.Context(), env(map[string]string{"CARLOS_HOST_CLAIM": "1", moves.SwitchEnv: "on"}), store, hib, newMovesRefusals("ie-box", nil, t.Logf, time.Now))
if !errors.Is(err, hostwrite.ErrUnavailable) || claimMoveSources.Parks != nil {
t.Fatalf("no placement region: %v", err)
}