fix(ci): prune marks, sweep approvals, replace usage period, fold sweep triggers (D-1319, D-1342, D-1317, D-1320)

.github/wiki/guides/ci/Scheduling-and-Cost.md

@@ -60,4 +60,4 @@ An allowance is the declared ceiling (`economics.ceilings.<provider>`) minus the
 
 - Not wired yet (debt D-1280): the dispatcher must bind the economics environment with `economics.Bind` (store, spend policy, clock, coordinator id) and copy the spend keys into the scheduler's policy with `economics.ApplyPolicy`. Until then the hooks are inert and nothing paid is placed.
 - A job the scheduler holds with `approval.pending` never reaches the admission step. The dispatcher records the whole pipeline's worst case with `economics.NotePending`, so `ApproveSpend` has a figure to bind to and one approval releases the run. Without it, the figure is the pipeline total at the moment a job is held, and each job that grows the pipeline past it is held again.
-- Burst (P7-SCHED-08) calls `economics.ReserveResource` and `economics.ReleaseResource` inside its own transactions. The GitHub billing importer (P7-HOST-10) writes through `economics.RecordUsage`.
+- Burst (P7-SCHED-08) calls `economics.ReserveResource` and `economics.ReleaseResource` inside its own transactions. The GitHub billing importer (P7-HOST-10) writes through `economics.ReplaceUsage`.

free/ci/cmd/providers_usage.go

@@ -8,7 +8,7 @@ package main
 // Outputs: `disabled` when providers.github.billing_import is not true (no API call, exit 0); else a
 //          one-line summary or one v1 JSON envelope (an error too, with --json); exit 0, 1
 //          (usage, E700, E709 and other provider refusals) or 2 (rate limit, network, anything else).
-// Constraints: V15 only; writes the store only through economics.RecordUsage; requests only the
+// Constraints: V15 only; writes the store only through economics.ReplaceUsage; requests only the
 //          non-summary billing usage endpoint (see internal/providers/github/usage); never asks for
 //          or refreshes a token scope. Live use waits for the owner to name the token scope.
 

free/ci/cmd/serve_rereport.go

@@ -0,0 +1,64 @@
+package main
+
+// Purpose: the coordinator's Rereport hook (D-1320): a sweep found a commit without a terminal nself-ci
+//          status although a run of it finished within the redelivery horizon; the stored result is sent
+//          again instead of building the commit a second time.
+// Inputs:  the finished intake row and the evidence the run recorded.
+// Outputs: one status through the outbox (a new seq, so it replaces any parked older one).
+// Constraints: only a bound, whole-gate GitHub run is re-sent (a partial run, a cancelled one or an
+//          unbound one never posted a status on purpose); the state is the one the run was authorized to
+//          post (the outbox payload: a failed release publication vetoes a passing result there, not in
+//          the evidence), and a run without that record is not re-sent; the description names the run,
+//          never evidence text the publication profile might filter.
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/nself-org/plugins/free/ci/internal/model"
+	"github.com/nself-org/plugins/free/ci/internal/report"
+	"github.com/nself-org/plugins/free/ci/internal/store"
+)
+
+func rereportDone(st *store.Store) func(context.Context, store.IntakeRow) error {
+	return func(ctx context.Context, row store.IntakeRow) error {
+		owner, repo, ok := strings.Cut(row.Project, "/")
+		if !ok || row.Source != "github" || row.Pipeline != "" || row.PipelineID == "" {
+			return nil
+		}
+		body, found, err := st.RunEvidence(ctx, row.PipelineID)
+		if err != nil || !found {
+			return err
+		}
+		var ev model.Evidence
+		if err = json.Unmarshal(body, &ev); err != nil || ev.Binding.State != "bound" || ev.Revision != row.Revision {
+			return nil
+		}
+		if ev.Result != "pass" && ev.Result != "fail" && ev.Result != "error" {
+			return nil
+		}
+		key := report.KeyFor(owner, repo, row.Revision)
+		payload, found, err := st.OutboxDecision(ctx, key.Revision, key.Context)
+		if err != nil || !found {
+			return err
+		}
+		var posted report.Status
+		if err = json.Unmarshal(payload, &posted); err != nil {
+			return nil
+		}
+		state := posted.State
+		if state != report.Success && state != report.Failure && state != report.Error {
+			return nil
+		}
+		if state == report.Success && ev.Result != "pass" { // evidence never allows a success the run could not post
+			return nil
+		}
+		sender := report.Default
+		if sender == nil || sender.Store != st {
+			sender = &report.Sender{Store: st}
+		}
+		return sender.Repost(ctx, []report.Key{key}, state, fmt.Sprintf("result of run %s sent again", row.PipelineID))
+	}
+}

free/ci/cmd/serve_rereport_test.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"path/filepath"
+	"testing"
+
+	"github.com/nself-org/plugins/free/ci/internal/model"
+	"github.com/nself-org/plugins/free/ci/internal/report"
+	"github.com/nself-org/plugins/free/ci/internal/store"
+)
+
+// D-1320: the status a finished, bound run was authorized to post is queued again, not the evidence result
+// (a failed release publication vetoes a passing run in the outbox only); anything the run would not have
+// posted is not.
+func TestRereportDoneQueuesTheStoredResult(t *testing.T) {
+	ctx := context.Background()
+	rev := "0123456789abcdef0123456789abcdef01234567"
+	for _, c := range []struct {
+		name     string
+		bind     model.BindingState
+		result   model.Result
+		pipeline string
+		posted   report.State // the status the run queued; "" = none was recorded
+		want     report.State
+	}{
+		{"pass", "bound", "pass", "", report.Success, report.Success},
+		{"fail", "bound", "fail", "", report.Failure, report.Failure},
+		{"error", "bound", "error", "", report.Error, report.Error},
+		{"publication failed after a passing gate", "bound", "pass", "", report.Failure, report.Failure},
+		{"pass without a recorded decision", "bound", "pass", "", "", ""},
+		{"success recorded for a failed run", "bound", "fail", "", report.Success, ""},
+		{"unbound run", "unbound", "pass", "", report.Success, ""},
+		{"partial run", "bound", "pass", "release", report.Success, ""},
+		{"other result", "bound", "cancelled", "", report.Failure, ""},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			st, err := store.Open(filepath.Join(t.TempDir(), "state.db"), store.Options{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() { _ = st.Close() })
+			if err = st.CreatePipeline(ctx, store.PipelineRow{ID: "run-1", Revision: rev, Project: "/proj", Trigger: "push", SourceTrust: "trusted", PrivacyZone: "local-only",
+				PolicyDigest: "p", InputDigest: "cfg", SelectionMode: "full", Status: "passed"}); err != nil {
+				t.Fatal(err)
+			}
+			ev := model.Evidence{Revision: rev, Binding: model.Binding{State: c.bind}, Result: c.result}
+			body, _ := json.Marshal(ev)
+			if err = st.RecordRunEvidence(ctx, "ev1", "run-1", rev, "cfg", body); err != nil {
+				t.Fatal(err)
+			}
+			if c.posted != "" { // what finishing the run queued through the outbox, after the release veto
+				seq, err := st.OutboxReserve(ctx, rev, "nself-ci@o/r")
+				if err != nil {
+					t.Fatal(err)
+				}
+				payload, _ := json.Marshal(report.Status{Owner: "o", Repo: "r", Revision: rev, State: c.posted, Description: "run"})
+				if err = st.OutboxReady(ctx, rev, "nself-ci@o/r", seq, payload); err != nil {
+					t.Fatal(err)
+				}
+			}
+			row := store.IntakeRow{Source: "github", Project: "o/r", Revision: rev, PipelineID: "run-1", Pipeline: c.pipeline}
+			if err = rereportDone(st)(ctx, row); err != nil {
+				t.Logf("delivery error (no remote in the test): %v", err)
+			}
+			all, err := st.OutboxList(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+			var items []store.OutboxItem // without the entry the run itself recorded
+			for _, it := range all {
+				if it.Seq > 1 || c.posted == "" {
+					items = append(items, it)
+				}
+			}
+			if c.want == "" {
+				if len(items) != 0 {
+					t.Fatalf("queued %d status(es), want none", len(items))
+				}
+				return
+			}
+			if len(items) != 1 || items[0].Revision != rev || items[0].Context != "nself-ci@o/r" {
+				t.Fatalf("outbox = %+v", items)
+			}
+			var s report.Status
+			if err = json.Unmarshal(items[0].Payload, &s); err != nil || s.State != c.want || s.Owner != "o" || s.Repo != "r" {
+				t.Fatalf("payload %s (%v), want state %s for o/r", items[0].Payload, err, c.want)
+			}
+		})
+	}
+}

free/ci/cmd/serve_wire.go

@@ -86,6 +86,7 @@ func serveWire(ctx context.Context, st *store.Store, home, coordinatorID string,
 	}
 	undoHosted := serveHosted(st, home, cfg.AllowedRepos)
 	undoNetscope := serveNetscope(st, home)
+	cfg.Rereport = rereportDone(st)
 	cfg.Drain = func(dctx context.Context) error {
 		d, err := drain.New(drain.Config{
 			Intake: agentIntake{}, Canceller: liveCanceller{}, Deadline: drain.FlagDeadline(), Logf: cfg.Logf,

free/ci/internal/coordinator/clock.go

@@ -0,0 +1,18 @@
+package coordinator
+
+// Purpose: the time source of the coordinator's schedules and maintenance (moved out of server.go to keep it
+// inside the file-size budget).
+// Inputs:  none. Outputs: Clock and its real implementation. Constraints: tests inject a fake Clock.
+
+import "time"
+
+// Clock is the time source of the schedules; tests inject a fake.
+type Clock interface {
+	Now() time.Time
+	After(d time.Duration) <-chan time.Time
+}
+
+type realClock struct{}
+
+func (realClock) Now() time.Time                         { return time.Now() }
+func (realClock) After(d time.Duration) <-chan time.Time { return time.After(d) }

free/ci/internal/coordinator/dedup.go

@@ -1,6 +1,6 @@
 package coordinator
 
-// Purpose: the semantic key, the maintenance loop (retention, claim heartbeat, recovery) of the V15
+// Purpose: the semantic key, the maintenance loop (retention, claim heartbeat, recovery, approval voids) of the V15
 // coordinator (P7-NODE-10, Epic D7, X7).
 // Inputs:  project, revision, trigger class and policy digest of a trigger; the injected clock.
 // Outputs: a stable semantic key; a store that forgets what is older than the 72 h redelivery horizon.
@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/nself-org/plugins/free/ci/internal/store"
+	"github.com/nself-org/plugins/free/ci/internal/trust/source"
 )
 
 // semanticKey is the key that coalesces triggers for the same code under the same policy. A trigger that
@@ -45,10 +46,9 @@ func (c *Coordinator) maintain(ctx context.Context) {
 		if err := st.IntakeBeat(ctx, c.cfg.ID); err != nil && ctx.Err() == nil {
 			c.cfg.Logf("intake heartbeat failed: %v", err)
 		}
+		c.sweepApprovals(ctx)
 		if now := c.cfg.Clock.Now(); now.Sub(pruned) >= time.Hour {
-			if _, err := st.DedupPrune(ctx, 0); err != nil && ctx.Err() == nil {
-				c.cfg.Logf("intake prune failed: %v", err)
-			}
+			c.prune(ctx, now)
 			pruned = now
 		}
 		c.nudge()
@@ -60,6 +60,53 @@ func (c *Coordinator) maintain(ctx context.Context) {
 	}
 }
 
+// sweepApprovals writes the void (and its audit row) of every standing approval that no longer matches its
+// run, so the void does not wait for `ci runs pending`, a facts write or a new approval (D-1342).
+func (c *Coordinator) sweepApprovals(ctx context.Context) {
+	if _, err := c.cfg.Store.SweepApprovals(ctx, "", approvalVoidCheck); err != nil && ctx.Err() == nil {
+		c.cfg.Logf("approval sweep failed: %v", err)
+	}
+}
+
+// approvalVoidCheck adapts the pure void rule of trust/source to the store's binding type.
+func approvalVoidCheck(approved, cur store.ApprovalBinding) []string {
+	return source.VoidReasons(source.Binding(approved), source.Binding(cur))
+}
+
+// MarkRetention is how long the marks of one provider_mark name stay. KeyPrefix narrows the rule to the
+// keys that start with it ("" is every key of the name); a mark older than MaxAge is deleted.
+type MarkRetention struct {
+	Name      string
+	KeyPrefix string
+	MaxAge    time.Duration
+}
+
+// DefaultMarkRetention is the horizon of the GitHub reconcile marks (keys are "<kind>|<repo>|<id>"): a run
+// update and a demand job are read again only while GitHub still lists them, so a week is generous; the
+// head (term, enq) and redelivery marks keep 30 days, far beyond the 72 h redelivery horizon.
+var DefaultMarkRetention = []MarkRetention{
+	{Name: "github-reconcile", KeyPrefix: "run|", MaxAge: 7 * 24 * time.Hour},
+	{Name: "github-reconcile", KeyPrefix: "demand|", MaxAge: 7 * 24 * time.Hour},
+	{Name: "github-reconcile", MaxAge: 30 * 24 * time.Hour},
+}
+
+// prune forgets what is past its retention: finished intake rows and keys, and the provider marks.
+func (c *Coordinator) prune(ctx context.Context, now time.Time) {
+	st := c.cfg.Store
+	if _, err := st.DedupPrune(ctx, 0); err != nil && ctx.Err() == nil {
+		c.cfg.Logf("intake prune failed: %v", err)
+	}
+	rules := c.cfg.MarkRetention
+	if rules == nil {
+		rules = DefaultMarkRetention
+	}
+	for _, r := range rules {
+		if _, err := st.ProviderMarkPrune(ctx, r.Name, r.KeyPrefix, now.Add(-r.MaxAge)); err != nil && ctx.Err() == nil {
+			c.cfg.Logf("provider mark prune failed (%s): %v", r.Name, err)
+		}
+	}
+}
+
 // dispatchLoop hands queued rows to the Dispatcher, at most Concurrency at a time, until ctx ends.
 func (c *Coordinator) dispatchLoop(ctx context.Context) {
 	if c.cfg.Dispatch == nil {

free/ci/internal/coordinator/ingress.go

@@ -29,6 +29,10 @@ type Delivery struct {
 	Facts        json.RawMessage // source facts (event, ref, clone_url, ...); opaque to the intake
 	Pipeline     string          // the requested pipeline, "" for the default
 	Synthetic    *model.Pipeline // a pipeline the source built itself (runner demand), if any
+	// Reconcile marks a delivery found by a sweep (a head without a terminal status). It coalesces into a
+	// run of the same source and semantic key that finished within the redelivery horizon, and the
+	// coordinator then asks Config.Rereport to send that run's stored result again instead of building twice.
+	Reconcile bool
 }
 
 // Ingress turns the requests of one provider into Deliveries.

free/ci/internal/coordinator/intake.go

@@ -81,7 +81,7 @@ func (c *Coordinator) enqueue(ctx context.Context, d Delivery, webhook bool, o e
 	out, err := c.cfg.Store.IntakeEnqueue(ctx, store.IntakeRequest{
 		Source: d.Source, Event: event, DeliveryID: d.DeliveryID, Project: d.Project, Revision: d.Revision, TriggerClass: d.TriggerClass,
 		PolicyDigest: c.cfg.PolicyDigest, SemanticKey: semanticKey(strings.ToLower(d.Project), d.Revision, d.TriggerClass, c.cfg.PolicyDigest, keyPipeline),
-		Pipeline: d.Pipeline, Facts: blob, Webhook: webhook, Force: c.cfg.Force || o.force, Actor: o.actor,
+		Pipeline: d.Pipeline, Facts: blob, Webhook: webhook, Reconcile: d.Reconcile, Force: c.cfg.Force || o.force, Actor: o.actor,
 		IdemDigest: o.idem, ReqDigest: o.reqDigest,
 	})
 	if err != nil {
@@ -94,9 +94,33 @@ func (c *Coordinator) enqueue(ctx context.Context, d Delivery, webhook bool, o e
 	if out.Kind == store.IntakeEnqueued {
 		c.nudge()
 	}
+	if d.Reconcile && (out.Kind == store.IntakeCoalesced || out.Kind == store.IntakeDuplicate) { // a duplicate: this sweep delivery id was folded or built before
+
+		c.rereport(ctx, out.ID)
+	}
 	return out, nil
 }
 
+// rereport sends the stored result of a finished intake row again, for a sweep that found its commit without
+// a terminal status (D-1320). A row that is still queued or running reports for itself; a failure is logged,
+// never returned: the sweep offers the head again.
+func (c *Coordinator) rereport(ctx context.Context, id string) {
+	if c.cfg.Rereport == nil {
+		return
+	}
+	row, err := c.cfg.Store.IntakeGet(ctx, id)
+	if err != nil {
+		c.cfg.Logf("intake row %s not read for a re-report: %v", id, err)
+		return
+	}
+	if row.State != store.IntakeDone {
+		return
+	}
+	if err = c.cfg.Rereport(ctx, row); err != nil {
+		c.cfg.Logf("re-report of intake row %s failed: %v", id, err)
+	}
+}
+
 func writeJSON(w http.ResponseWriter, code int, v any) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(code)