mesh-issues: an issue knows whether its fix is delivered, says it ready to verify, and a fix stuck after its merge is said (hq ADR 0281)
mesh/merge-gate pass: builds mesh-issues → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery delivered

The operator approved measuring from the delivery's own events whether each fix an
issue names reached the machines; delivered is not resolved, and a merged fix that
does not arrive is said on the issue and the operator's channel instead of found.
This commit is contained in:
jochen
2026-10-10 18:12:52 +02:00
parent 0b4ae8097f
commit 01bde68f96
10 changed files with 792 additions and 167 deletions
+20 -1
View File
@@ -81,6 +81,24 @@ to two minutes, which the tool runner allows a handler while keeping its event a
The new values are optional keys of an issue's one JSON value: an issue kept before reads as having none; the
key-value state has no schema to migrate.
## Whether the fix is delivered (hq ADR 0281)
For each pull request an issue's `fixed-by` names, the module keeps what `mesh-delivery.transition` last said —
state, why, merged and when — or asks `seat:mesh-delivery.checks` once (at its own `moved`) for one it has not
heard of. Nothing sets it by hand. `get` and every `list` row carry `delivery`, one line derived from it, worst
first: "fix not delivered: …", "held: …", "fix merged, not delivered: …", "fix not merged: …", "nothing heard from
the delivery of …", "delivered to every machine it should reach" (covered by a later merge says so).
Delivered is not resolved. When every fix of an issue not yet resolved is delivered, the issue gains a note and the
event `ready-to-verify`, once per set of fixes; it is resolved once its acceptance is proven live. A fix is named
while the issue stays `located`: `move` to the status it has changes what it holds and is refused when it changes
nothing.
Every five minutes the module looks for a fix merged and not delivered after `stuck-after` (default `2h`): it is
said once — a note on the issue and a warning on the operator's channel (`seat:operator-channel.notify`, key
`issue-tracker.fix-stuck.<number>`), cleared there when every fix of the issue is delivered. A module that runs only
its bundle cannot raise a controller condition (hq ADR 0240 rule 8, ADR 0256), so the warning is the channel's, not the controller's.
## The import of hq's issues (slice 3)
`issue-tracker.import` reads every `04-ISSUES/NNN-slug/00-report.md` of hq **through the records module
@@ -174,7 +192,8 @@ verb takes an argument naming a caller, and none can be given one.
## Settings
`duplicate-check` (`on`; `off` stops the triage of new issues), `judge-model` (`sonnet`; `haiku` or `opus`) and
`forge-owner` (the forge's owner of the repositories `fixed-by` names, for asking a delivery) steer the triage.
`forge-owner` (the forge's owner of the repositories `fixed-by` names, for asking a delivery) steer the triage;
`stuck-after` (`2h`) is how long a merged fix may take to be delivered before it is said.
```
{
@@ -0,0 +1,460 @@
package main
// Whether an issue's fix is delivered (novox/hq ADR 0281). The tracker hears mesh-delivery's `transition` event
// for every pull request an issue's fixed-by names and keeps, per pull request, the state the delivery last said
// — merged or not, held and why, covered by a later merge, delivered. From that it derives one line for `get` and
// `list`: a measured fact, never set by a verb. Delivered is not resolved: when every fix of an issue not yet
// resolved is delivered, the issue gains a note and the event `ready-to-verify`, and an agent proves its
// acceptance live before resolving it. A fix merged and not delivered past a bound is said on the issue and to
// the operator's channel.
import (
"encoding/json"
"errors"
"fmt"
"sort"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// EventReadyToVerify is said once when every fix of an issue not yet resolved is delivered.
const EventReadyToVerify = "ready-to-verify"
// DefaultStuckAfter is how long a fix may be merged and not delivered before it is said stuck: a walk takes
// minutes, a gate ten at most, and a hold waits for a person; two hours is past every ordinary delivery and
// still the same working day.
const DefaultStuckAfter = 2 * time.Hour
// Fix is one pull request of an issue's fixed-by, as the delivery last said it.
type Fix struct {
PullRequest string `json:"pull-request"`
State string `json:"state"`
Why string `json:"why,omitempty"`
Merged bool `json:"merged,omitempty"`
MergedAt *time.Time `json:"merged-at,omitempty"`
Since time.Time `json:"since"`
// StuckSaid says it was said stuck, on the issue and to the operator's channel.
StuckSaid bool `json:"stuck-said,omitempty"`
}
// reached says whether a delivery's state says the change reached the machines (mesh-delivery's Reached).
func reached(state string) bool {
return state == "delivered" || state == "covered" || state == "retired"
}
// afterMerge says whether a state is one a delivery reaches only once merged.
func afterMerge(state string) bool {
switch state {
case "published", "delivering", "held", "delivered", "failed", "stopped", "covered", "retired":
return true
}
return false
}
func fixOf(is *Issue, pr string) *Fix {
for i := range is.Fixes {
if is.Fixes[i].PullRequest == pr {
return &is.Fixes[i]
}
}
return nil
}
// allDelivered says whether an issue names a pull request and every one is delivered.
func allDelivered(is *Issue) bool {
prs := pullRequests(is)
if len(prs) == 0 {
return false
}
for _, pr := range prs {
if !contains(is.Delivered, pr) {
return false
}
}
return true
}
// DeliveryLine is the derived delivery state of an issue's fix, in words; empty when fixed-by names no pull
// request.
func DeliveryLine(is *Issue) string {
prs := pullRequests(is)
if len(prs) == 0 {
return ""
}
var held, failed, merged, open, unheard []string
covered := false
for _, pr := range prs {
f := fixOf(is, pr)
switch {
case contains(is.Delivered, pr):
covered = covered || (f != nil && f.State == "covered")
case f == nil:
unheard = append(unheard, pr)
case f.State == "held":
held = append(held, strings.TrimSpace(pr+": "+f.Why))
case f.State == "failed" || f.State == "stopped":
failed = append(failed, strings.TrimSpace(pr+" "+f.State+": "+f.Why))
case f.Merged:
merged = append(merged, pr)
default:
open = append(open, pr)
}
}
switch {
case len(failed) > 0:
return "fix not delivered: " + strings.Join(failed, "; ")
case len(held) > 0:
return "held: " + strings.Join(held, "; ")
case len(merged) > 0:
return "fix merged, not delivered: " + strings.Join(merged, ", ")
case len(open) > 0:
return "fix not merged: " + strings.Join(open, ", ")
case len(unheard) > 0:
return "nothing heard from the delivery of " + strings.Join(unheard, ", ")
case covered:
return "delivered to every machine it should reach (by a later merge that contains it)"
}
return "delivered to every machine it should reach"
}
// recordFix keeps what the delivery said of a pull request on every issue whose fixed-by names it (only, when
// given, on that one issue), and answers the issues it changed.
func (t *Tracker) recordFix(pr, state, why string, merged bool, at time.Time, only uint64) ([]uint64, error) {
all, err := t.allIssues()
if err != nil {
return nil, err
}
var changed []uint64
for n, is := range all {
if (only != 0 && n != only) || !contains(pullRequests(is), pr) {
continue
}
kept := cut(why, MostReasonKept)
if f := fixOf(is, pr); f != nil && (at.Before(f.Since) || (f.State == state && f.Why == kept && f.Merged == (merged || f.Merged))) {
// Older than what is kept, or nothing new: transitions can arrive out of order.
continue
}
if contains(is.Delivered, pr) && !reached(state) {
// Delivered stays delivered: a later state that has not reached the machines is a newer delivery of
// the same pull request at most, and an older one is history.
continue
}
if _, err := t.change(n, func(i *Issue) error {
f := fixOf(i, pr)
if f == nil {
i.Fixes = append(i.Fixes, Fix{PullRequest: pr})
f = &i.Fixes[len(i.Fixes)-1]
}
f.State, f.Why, f.Since = state, kept, at
if merged && !f.Merged {
f.Merged = true
m := at
f.MergedAt = &m
}
if reached(state) && !contains(i.Delivered, pr) {
i.Delivered = append(i.Delivered, pr)
}
return nil
}); err != nil {
return changed, err
}
changed = append(changed, n)
}
sort.Slice(changed, func(i, j int) bool { return changed[i] < changed[j] })
return changed, nil
}
// afterFix is what follows a change of an issue's fixes: ready to verify, once, when every fix is delivered and
// the issue is not yet resolved; a stuck fix delivered at last cleared on the operator's channel; and what
// waited on it let go.
func (t *Tracker) afterFix(n uint64) error {
is, _, err := read(t.Store, n)
if err != nil {
return err
}
if err := t.clearStuck(is); err != nil {
t.Logf("[mesh-issues] the stuck fix of %d could not be cleared on the operator's channel (tried again at the next sweep): %v", n, err)
}
if allDelivered(is) {
if unresolved(is) {
prs := pullRequests(is)
sort.Strings(prs)
if strings.Join(is.VerifyFor, ",") != strings.Join(prs, ",") {
// The note first, kept once by its words; then the claim; then the event. A failure before the claim
// is tried again at the next offer, and the note is not written twice.
prefix := fmt.Sprintf("Ready to verify: every fix of this issue is delivered (%s).", strings.Join(prs, ", "))
if !noted(is, prefix) {
if _, err := t.note(GroomBy, n, prefix+" Delivered is not resolved: prove the acceptance live, then "+
"resolve it. (novox/hq ADR 0281)"); err != nil {
return err
}
}
claimed := false
if _, err := t.change(n, func(i *Issue) error {
claimed = false
if strings.Join(i.VerifyFor, ",") == strings.Join(prs, ",") {
return nil
}
i.VerifyFor, claimed = prs, true
return nil
}); err != nil {
return err
}
if claimed {
t.emit(EventReadyToVerify, map[string]any{"number": n, "fixed-by": prs, "at": t.now()})
}
}
}
}
return t.letGo(n)
}
// stillStuck says whether an issue has a fix said stuck that is still named, not delivered, and the issue not
// closed as wontfix.
func stillStuck(is *Issue) bool {
if is.Status == StatusWontfix {
return false
}
named := pullRequests(is)
for _, f := range is.Fixes {
if f.StuckSaid && contains(named, f.PullRequest) && !contains(is.Delivered, f.PullRequest) {
return true
}
}
return false
}
// clearStuck clears the warning said for an issue on the operator's channel once nothing it was said for is
// stuck — delivered, no longer named, or the issue closed as wontfix; one that fails is tried again at the next
// sweep.
func (t *Tracker) clearStuck(is *Issue) error {
if !is.StuckOpen || stillStuck(is) || t.Notify == nil {
return nil
}
if err := t.Notify(stuckKey(is.Number), "", "", "", "", "the fix of issue "+keyOf(is.Number)+" is delivered or no longer waited for", true); err != nil {
return err
}
_, err := t.change(is.Number, func(i *Issue) error { i.StuckOpen = false; return nil })
return err
}
func stuckKey(n uint64) string { return "issue-tracker.fix-stuck." + keyOf(n) }
func prKey(repository string, number int) string {
_, repo, ok := strings.Cut(repository, "/")
if !ok {
repo = repository
}
return fmt.Sprintf("%s#%d", strings.ToLower(repo), number)
}
// OnTransition is the handler of mesh-delivery's `transition` event: every state a pull request's delivery moves
// to is kept on each issue whose fixed-by names it.
func (t *Tracker) OnTransition(e stdio.Envelope) error {
var body struct {
Repository string `json:"repository"`
Number int `json:"number"`
To string `json:"to"`
Why string `json:"why"`
MergedAs string `json:"merged_as"`
At time.Time `json:"at"`
}
if err := json.Unmarshal(e.Body, &body); err != nil || body.To == "" || body.Number <= 0 {
return nil
}
at := body.At.UTC()
if at.IsZero() {
at = t.now()
}
merged := body.MergedAs != "" || afterMerge(body.To)
changed, err := t.recordFix(prKey(body.Repository, body.Number), body.To, body.Why, merged, at, 0)
if err != nil {
return err
}
for _, n := range changed {
if err := t.afterFix(n); err != nil {
return err
}
}
return nil
}
// Delivery is a pull request's delivery as mesh-delivery's `checks` answers it; a test stubs the asking.
type Delivery struct {
State string
Why string
// Since is when the delivery entered its state, as it says; zero when it does not.
Since time.Time
}
// DeliveryState asks mesh-delivery for one pull request's delivery.
type DeliveryState func(repository string, number string) (Delivery, error)
// busDeliveryState asks mesh-delivery's checks over the bus.
func busDeliveryState(owner string) DeliveryState {
return func(repository, number string) (Delivery, error) {
raw, err := stdio.Ask("seat:mesh-delivery.checks", map[string]any{"repository": owner + "/" + repository, "number": number})
if err != nil {
return Delivery{}, err
}
var out struct {
Delivery struct {
State string `json:"state"`
Waits string `json:"waits"`
Since time.Time `json:"since"`
} `json:"delivery"`
}
if err := json.Unmarshal(raw, &out); err != nil {
return Delivery{}, fmt.Errorf("mesh-delivery's checks answered no delivery: %.200s", raw)
}
return Delivery{State: out.Delivery.State, Why: out.Delivery.Waits, Since: out.Delivery.Since.UTC()}, nil
}
}
// OnMoved is the handler of the tracker's own `moved` event: each pull request of the issue's fix not heard of
// yet is asked of mesh-delivery, at most a few per handling, for a delivery that moved before the issue named
// it; then what follows its fixes, and an issue closed lets go what waited on it.
func (t *Tracker) OnMoved(e stdio.Envelope) error {
var body struct {
Number uint64 `json:"number"`
}
if err := json.Unmarshal(e.Body, &body); err != nil || body.Number == 0 {
return nil
}
is, _, err := read(t.Store, body.Number)
if errors.Is(err, ErrNoIssue) {
return nil
}
if err != nil {
return err
}
if t.Deliveries != nil {
// Each ask up to the bus's 30 seconds, under the runtime's two minutes; one that fails, or more left to
// ask, has the event offered again, so a delivery never asked ends in max-deliveries, not in silence.
asked := 0
for _, pr := range pullRequests(is) {
if fixOf(is, pr) != nil {
continue
}
if asked == MostDeliveryAsks {
return fmt.Errorf("issue %d names more pull requests than are asked at once; the rest at the next offer", body.Number)
}
asked++
repo, number, _ := strings.Cut(pr, "#")
d, err := t.Deliveries(repo, number)
if err != nil {
return fmt.Errorf("the delivery of %s (issue %d) could not be asked: %v", pr, body.Number, err)
}
state := d.State
if state == "" {
// The delivery knows nothing of it yet: not kept, so it is asked again at the next move, and its own
// events say it when they come.
continue
}
// The delivery's own time, so a transition stamped the same moment is not taken as older.
since := d.Since
if since.IsZero() {
since = t.now()
}
if _, err := t.recordFix(pr, state, d.Why, afterMerge(state), since, body.Number); err != nil {
return err
}
}
}
return t.afterFix(body.Number)
}
// Notifier tells the operator's channel something (operator-channel.notify), or clears it.
type Notifier func(key, severity, summary, headline, explanation, resolved string, clear bool) error
// busNotifier asks the operator's channel over the bus.
func busNotifier() Notifier {
return func(key, severity, summary, headline, explanation, resolved string, clear bool) error {
body := map[string]any{"key": key, "subject": "the issue tracker"}
if clear {
body["clear"] = true
body["resolved"] = resolved
} else {
body["severity"], body["summary"], body["headline"], body["explanation"] = severity, summary, headline, explanation
}
_, err := stdio.Ask("seat:operator-channel.notify", body)
return err
}
}
// stuckAfter is the setting's bound, or the default.
func (t *Tracker) stuckAfter() time.Duration {
if s, err := t.Settings(); err == nil && strings.TrimSpace(s.StuckAfter) != "" {
if d, err := time.ParseDuration(strings.TrimSpace(s.StuckAfter)); err == nil && d > 0 {
return d
}
}
return DefaultStuckAfter
}
// Sweep says every fix merged and not delivered past the bound: a note on its issue and a warning on the
// operator's channel, once; it answers how many it said. Run every few minutes.
func (t *Tracker) Sweep() (int, error) {
all, err := t.allIssues()
if err != nil {
return 0, err
}
bound, now := t.stuckAfter(), t.now()
said := 0
for n, is := range all {
if is.Status == StatusWontfix {
if err := t.clearStuck(is); err != nil {
t.Logf("[mesh-issues] the stuck fix of %d could not be cleared on the operator's channel: %v", n, err)
}
continue
}
if err := t.clearStuck(is); err != nil {
t.Logf("[mesh-issues] the stuck fix of %d could not be cleared on the operator's channel: %v", n, err)
}
named := pullRequests(is)
for _, f := range is.Fixes {
if !contains(named, f.PullRequest) || contains(is.Delivered, f.PullRequest) || !f.Merged || f.MergedAt == nil ||
reached(f.State) || f.StuckSaid || now.Sub(*f.MergedAt) < bound {
continue
}
state := f.State
if f.Why != "" {
state += ": " + f.Why
}
prefix := fmt.Sprintf("Stuck: the fix %s", f.PullRequest)
if !noted(is, prefix) {
if _, err := t.note(GroomBy, n, fmt.Sprintf("%s was merged %s ago and is not delivered (%s). A merged fix "+
"that does not reach the machines is said on the operator's channel. (novox/hq ADR 0281)", prefix,
now.Sub(*f.MergedAt).Round(time.Minute), state)); err != nil {
return said, err
}
}
if t.Notify != nil {
if err := t.Notify(stuckKey(n), "warning",
fmt.Sprintf("issue %d: fix %s merged %s ago, not delivered", n, f.PullRequest, now.Sub(*f.MergedAt).Round(time.Minute)),
fmt.Sprintf("A fix is merged and not delivered (issue %d)", n),
fmt.Sprintf("The fix of issue %d was merged %s ago and has not reached the machines: %s. Look at its "+
"delivery; nothing else is needed from you unless it waits for a person.", n,
now.Sub(*f.MergedAt).Round(time.Minute), state), "", false); err != nil {
// Tried again at the next sweep: the note is not written twice.
t.Logf("[mesh-issues] the stuck fix of %d could not be said on the operator's channel: %v", n, err)
continue
}
}
pr := f.PullRequest
if _, err := t.change(n, func(i *Issue) error {
if g := fixOf(i, pr); g != nil {
g.StuckSaid = true
}
i.StuckOpen = t.Notify != nil
return nil
}); err != nil {
return said, err
}
said++
}
}
return said, nil
}
@@ -0,0 +1,246 @@
package main
import (
"encoding/json"
"strings"
"sync"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
type notified struct {
mu sync.Mutex
said []string
clear []string
err error
}
func (n *notified) notify(key, severity, summary, headline, explanation, resolved string, clear bool) error {
n.mu.Lock()
defer n.mu.Unlock()
if n.err != nil {
return n.err
}
if clear {
n.clear = append(n.clear, key)
} else {
n.said = append(n.said, key+" "+severity+" "+summary)
}
return nil
}
func transition(t *testing.T, h *harness, number int, to, why, mergedAs string, at time.Time) {
t.Helper()
raw, _ := json.Marshal(map[string]any{"repository": "novox/mesh-catalog", "number": number, "to": to, "why": why,
"merged_as": mergedAs, "at": at})
if err := h.OnTransition(stdio.Envelope{Key: "mesh-delivery.transition", Body: raw}); err != nil {
t.Fatal(err)
}
}
func countEvents(h *harness, event string) int {
n := 0
for _, e := range h.said() {
if e.Event == event {
n++
}
}
return n
}
// The delivery line follows what the delivery says, as a measured fact: not merged, merged, held and why,
// delivered; an issue not yet resolved is said ready to verify once, and stays unresolved.
func TestAnIssueKnowsWhetherItsFixIsDelivered(t *testing.T) {
h := newHarness(t)
n := h.open(t, "the counter skips a number")
if _, err := h.Move(MoveAsk{Number: n, Status: StatusLocated, LocatedIn: []string{"mesh-catalog modules/records"},
FixedBy: []string{"mesh-catalog#9"}}); err != nil {
t.Fatal(err)
}
line := func() string {
v, err := h.View(n)
if err != nil {
t.Fatal(err)
}
return v.Delivery
}
if got := line(); got != "nothing heard from the delivery of mesh-catalog#9" {
t.Fatal(got)
}
at := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC)
transition(t, h, 9, "ready", "", "", at)
if got := line(); got != "fix not merged: mesh-catalog#9" {
t.Fatal(got)
}
transition(t, h, 9, "published", "", "abc", at)
if got := line(); got != "fix merged, not delivered: mesh-catalog#9" {
t.Fatal(got)
}
transition(t, h, 9, "held", "dnsmasq waits for a person's push on two machines", "abc", at)
if got := line(); got != "held: mesh-catalog#9: dnsmasq waits for a person's push on two machines" {
t.Fatal(got)
}
rows, _ := h.List(ListAsk{})
if rows.Rows[0].Delivery != line() {
t.Fatalf("%+v", rows.Rows[0])
}
for i := 0; i < 2; i++ {
transition(t, h, 9, "delivered", "", "abc", at)
}
if got := line(); got != "delivered to every machine it should reach" {
t.Fatal(got)
}
is := get(t, h, n)
if is.Status != StatusLocated || countEvents(h, EventReadyToVerify) != 1 {
t.Fatalf("%s, %d ready-to-verify", is.Status, countEvents(h, EventReadyToVerify))
}
var verify []string
for _, nt := range is.Notes {
if strings.HasPrefix(nt.Text, "Ready to verify:") {
verify = append(verify, nt.Text)
}
}
if len(verify) != 1 || !strings.Contains(verify[0], "Delivered is not resolved") {
t.Fatalf("%q", verify)
}
// Covered by a later merge counts as delivered, and says so; a resolved issue is not said ready to verify.
h2 := newHarness(t)
m := h2.open(t, "x")
resolve(t, h2, m, "mesh-catalog#9")
transition(t, h2, 9, "covered", "", "abc", at)
if v, _ := h2.View(m); !strings.HasPrefix(v.Delivery, "delivered to every machine it should reach (by a later merge") ||
countEvents(h2, EventReadyToVerify) != 0 {
t.Fatalf("%q", v.Delivery)
}
}
// A fix merged and not delivered past the bound is said once on the issue and the operator's channel, and
// cleared there when it is delivered; one that fails to be said is said at the next sweep.
func TestAStuckFixIsSaidOnceAndClearedWhenDelivered(t *testing.T) {
h := newHarness(t)
n := h.open(t, "the counter skips a number")
resolve(t, h, n, "mesh-catalog#9")
note := &notified{}
h.Notify = note.notify
merged := time.Date(2026, 10, 10, 9, 0, 0, 0, time.UTC) // three hours before the harness's clock
transition(t, h, 9, "held", "waits for a person", "abc", merged)
note.err = errStuck
if said, err := h.Sweep(); err != nil || said != 0 {
t.Fatalf("%d %v", said, err)
}
note.err = nil
for i := 0; i < 2; i++ {
if _, err := h.Sweep(); err != nil {
t.Fatal(err)
}
}
stuck := 0
for _, nt := range get(t, h, n).Notes {
if strings.HasPrefix(nt.Text, "Stuck: the fix mesh-catalog#9") {
stuck++
}
}
if stuck != 1 || len(note.said) != 1 || !strings.HasPrefix(note.said[0], "issue-tracker.fix-stuck.365 warning") {
t.Fatalf("%d notes, said %q", stuck, note.said)
}
transition(t, h, 9, "delivered", "", "abc", merged)
if len(note.clear) != 1 {
t.Fatalf("not cleared: %q", note.clear)
}
// Under the bound nothing is said.
h2 := newHarness(t)
m := h2.open(t, "y")
resolve(t, h2, m, "mesh-catalog#9")
n2 := &notified{}
h2.Notify = n2.notify
transition(t, h2, 9, "delivering", "", "abc", time.Date(2026, 10, 10, 11, 30, 0, 0, time.UTC))
if said, _ := h2.Sweep(); said != 0 || len(n2.said) != 0 {
t.Fatal("said stuck under the bound")
}
h2.settings.StuckAfter = "10m"
if said, _ := h2.Sweep(); said != 1 {
t.Fatal("the setting's bound was not read")
}
}
var errStuck = stdioErr("the operator's channel did not answer")
type stdioErr string
func (e stdioErr) Error() string { return string(e) }
// A fix is named while the issue stays located: a move to the same status changes only what it holds.
func TestAFixIsNamedWhileTheIssueStaysLocated(t *testing.T) {
h := newHarness(t)
n := h.open(t, "x")
if _, err := h.Move(MoveAsk{Number: n, Status: StatusLocated, LocatedIn: []string{"mesh-catalog modules/records"}}); err != nil {
t.Fatal(err)
}
if _, err := h.Move(MoveAsk{Number: n, Status: StatusLocated}); err == nil {
t.Fatal("a move that changes nothing was taken")
}
if _, err := h.Move(MoveAsk{Number: n, Status: StatusLocated, FixedBy: []string{"mesh-catalog#9"}}); err != nil {
t.Fatal(err)
}
if is := get(t, h, n); is.Status != StatusLocated || is.Closed != nil || len(is.FixedBy) != 1 {
t.Fatalf("%+v", is)
}
}
// Transitions out of order never take a delivered fix back, and a delivered fix is never said stuck; a clear
// that fails is tried again; a fix no longer named is not swept; an empty answer from the delivery is asked again.
func TestDeliveredStaysDeliveredAndClearsAreRetried(t *testing.T) {
h := newHarness(t)
n := h.open(t, "x")
resolve(t, h, n, "mesh-catalog#9")
note := &notified{}
h.Notify = note.notify
merged := time.Date(2026, 10, 10, 9, 0, 0, 0, time.UTC)
transition(t, h, 9, "held", "waits", "abc", merged)
if said, _ := h.Sweep(); said != 1 {
t.Fatal("not said stuck")
}
note.err = errStuck
transition(t, h, 9, "delivered", "", "abc", merged.Add(time.Hour))
transition(t, h, 9, "held", "an older word", "abc", merged.Add(30*time.Minute)) // out of order
if v, _ := h.View(n); v.Delivery != "delivered to every machine it should reach" {
t.Fatal(v.Delivery)
}
if len(note.clear) != 0 || !get(t, h, n).StuckOpen {
t.Fatal("cleared while the channel failed")
}
note.err = nil
if said, _ := h.Sweep(); said != 0 || len(note.clear) != 1 || get(t, h, n).StuckOpen {
t.Fatalf("said %d, cleared %q", said, note.clear)
}
// A fix no longer named is not swept, and its warning is cleared.
h2 := newHarness(t)
m := h2.open(t, "y")
if _, err := h2.Move(MoveAsk{Number: m, Status: StatusLocated, LocatedIn: []string{"mesh-catalog modules/records"}, FixedBy: []string{"mesh-catalog#9"}}); err != nil {
t.Fatal(err)
}
n2 := &notified{}
h2.Notify = n2.notify
transition(t, h2, 9, "held", "waits", "abc", merged)
_, _ = h2.Sweep()
if _, err := h2.Move(MoveAsk{Number: m, Status: StatusLocated, FixedBy: []string{"mesh-catalog#10"}}); err != nil {
t.Fatal(err)
}
if said, _ := h2.Sweep(); said != 0 || len(n2.said) != 1 || len(n2.clear) != 1 {
t.Fatalf("said %d %q, cleared %q", said, n2.said, n2.clear)
}
if _, err := h2.Move(MoveAsk{Number: m, Status: StatusLocated, FixedBy: []string{"mesh-catalog#10"}}); err == nil {
t.Fatal("a move that changes nothing was taken")
}
// The delivery knows nothing yet: not kept, asked again at the next move.
asks := 0
h2.Deliveries = func(string, string) (Delivery, error) { asks++; return Delivery{}, nil }
for i := 0; i < 2; i++ {
_ = h2.OnMoved(stdio.Envelope{Body: json.RawMessage(`{"number":365}`)})
}
if v, _ := h2.View(m); asks != 2 || v.Delivery != "nothing heard from the delivery of mesh-catalog#10" {
t.Fatalf("%d asks, %q", asks, v.Delivery)
}
}
+7 -3
View File
@@ -91,9 +91,13 @@ type Issue struct {
SuggestedOwner string `json:"suggested-owner,omitempty"`
// Delivered is the pull requests of FixedBy the delivery said are delivered.
Delivered []string `json:"delivered,omitempty"`
// Undelivered is the pull requests of FixedBy asked of the delivery at the move to resolved and not yet
// delivered then; their own delivery event marks them.
Undelivered []string `json:"undelivered,omitempty"`
// Fixes is each pull request of FixedBy as the delivery last said it (ADR 0281): measured, never set by a
// verb.
Fixes []Fix `json:"fixes,omitempty"`
// VerifyFor is the pull requests it was last said ready to verify for, so it is said once.
VerifyFor []string `json:"verify-for,omitempty"`
// StuckOpen is the warning on the operator's channel said for a stuck fix and not yet cleared there.
StuckOpen bool `json:"stuck-open,omitempty"`
// ReadyFor is the issues it depended on when it was last said ready (ADR 0279 §3), so one spell of
// readiness is said once however many dependencies let it go at one delivery.
ReadyFor []uint64 `json:"ready-for,omitempty"`
+21 -3
View File
@@ -41,6 +41,9 @@ type Settings struct {
// ForgeOwner is the owner on the forge of the repositories fixed-by names, for asking a pull request's
// delivery (ADR 0279 §3).
ForgeOwner string `json:"forge-owner,omitempty"`
// StuckAfter is how long a fix may be merged and not delivered before it is said stuck (ADR 0281), as a
// duration such as 2h; DefaultStuckAfter when empty.
StuckAfter string `json:"stuck-after,omitempty"`
}
func readSettings(path string) (Settings, error) {
@@ -76,7 +79,8 @@ func main() {
Sleep: time.Sleep,
Judge: busJudgement(os.Getenv("MESH_NODE")),
}
t.Deliveries = func(repository, number string) (string, error) {
t.Notify = busNotifier()
t.Deliveries = func(repository, number string) (Delivery, error) {
owner := "novox"
if s, err := t.Settings(); err == nil && strings.TrimSpace(s.ForgeOwner) != "" {
owner = strings.TrimSpace(s.ForgeOwner)
@@ -90,6 +94,7 @@ func main() {
logf("[mesh-issues] the settings cannot be read (%v): every open and move will say so", err)
}
go hearOpened(t)
go sweep(t)
if err := stdio.Serve("", Tools(t)); err != nil {
logf("%v", err)
os.Exit(1)
@@ -117,6 +122,17 @@ func hearOpened(t *Tracker) {
}
}
// sweep says, every five minutes, every fix merged and not delivered past the bound (ADR 0281).
func sweep(t *Tracker) {
for range time.Tick(5 * time.Minute) {
if n, err := t.Sweep(); err != nil {
logf("[mesh-issues] the sweep for stuck fixes failed: %v", err)
} else if n > 0 {
logf("[mesh-issues] said %d stuck fix(es)", n)
}
}
}
// dryRunAtTheTerminal is `mesh-issues dry-run <hq checkout> <settings.json>`: the import's dry run over a
// checkout on disk, against no tracker, before the module is rolled out. It reads, writes nothing, and
// prints what the import would create from that checkout, as the verb's dry run would answer against an
@@ -320,7 +336,8 @@ func Tools(t *Tracker) []stdio.Tool {
{Name: Seat + ".get",
Description: "One issue whole, by its number: title, status, severity, what it needs, report, who opened it " +
"and when, the owners it is located in, what fixed it and whether that is delivered, its notes and its " +
"links; and the issues it waits for, those that depend on it (needed-by), and whether it is ready.",
"links; the issues it waits for, those that depend on it (needed-by), and whether it is ready; and " +
"whether its fix is delivered, as the delivery last said it (merged, held and why, delivered).",
Input: schema([]string{"number"}, map[string]any{"number": number("the issue's number")}),
Run: func(m map[string]any) (any, error) {
a, err := readArgs("get", m, "number")
@@ -338,7 +355,7 @@ func Tools(t *Tracker) []stdio.Tool {
"those at one status (" + statuses + "), located in one owner, at one severity, needing one thing, " +
"ready or not, or whose title or report holds a phrase. An agent looking for work asks severity minor, " +
"needs none, ready. An issue is ready when everything it depends on is closed, or resolved with its fix " +
"delivered. Answers what is open in one call.",
"delivered. Each row says whether its fix is delivered. Answers what is open in one call.",
Input: schema(nil, map[string]any{
"status": map[string]any{"type": "string", "enum": Statuses, "description": "only issues at this status"},
"owner": text("only issues located in this owner, such as mesh-controller"),
@@ -390,6 +407,7 @@ func Tools(t *Tracker) []stdio.Tool {
"development cycle's rules: located needs located-in, owners on the registered list; resolved needs " +
"located-in and fixed-by (the pull requests or commits), and an issue in the mesh's core also replay " +
"(the replay's id, such as R361) or replay-none (why no replay is possible); wontfix needs a reason. " +
"A move to the status it has changes only what it holds, such as naming the fix while it stays located. " +
"A refused move changes nothing and says which rule refused it.",
Input: schema([]string{"number", "status"}, map[string]any{
"number": number("the issue's number"),
@@ -84,7 +84,7 @@ func TestItDeclaresAndHoldsTheIssueTrackerSeatOnceForTheMesh(t *testing.T) {
t.Fatalf("module %q, seats %+v, claims %+v", m.Module, m.Seats, m.Claims)
}
if !reflect.DeepEqual(m.Emits, []string{EventOpened, EventMoved, EventNoted, EventLinked, EventImported, EventSet,
EventTriaged, EventReady}) {
EventTriaged, EventReady, EventReadyToVerify}) {
t.Fatalf("emits %v", m.Emits)
}
// Three states, none expiring: issues keep a short history per key, the counter one value, the backlog's
@@ -97,7 +97,7 @@ func TestItDeclaresAndHoldsTheIssueTrackerSeatOnceForTheMesh(t *testing.T) {
}
// The duplicate check hears the tracker's own opens and asks the agent module on its node (ADR 0278 §1, §5).
if !reflect.DeepEqual(m.Consumes, []string{"mesh-issues." + EventOpened, "mesh-issues." + EventMoved, "mesh-delivery.transition"}) ||
!reflect.DeepEqual(m.Invokes, []string{"claude-code.claude_code_judge", "seat:mesh-delivery.checks"}) {
!reflect.DeepEqual(m.Invokes, []string{"claude-code.claude_code_judge", "seat:mesh-delivery.checks", "seat:operator-channel.notify"}) {
t.Fatalf("consumes %v, invokes %v", m.Consumes, m.Invokes)
}
if len(m.Build.Artifacts) != 1 || m.Build.Artifacts[0].From != "cmd/mesh-issues" ||
+4 -139
View File
@@ -7,14 +7,11 @@ package main
// the event `ready`.
import (
"encoding/json"
"errors"
"fmt"
"regexp"
"sort"
"strings"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// The events of ADR 0279.
@@ -271,6 +268,8 @@ type View struct {
NeededBy []uint64 `json:"needed-by,omitempty"`
WaitsFor []uint64 `json:"waits-for,omitempty"`
Ready bool `json:"ready"`
// Delivery is whether its fix is delivered, in words (ADR 0281).
Delivery string `json:"delivery,omitempty"`
}
// View is `get`.
@@ -283,33 +282,8 @@ func (t *Tracker) View(n uint64) (*View, error) {
if is == nil {
return nil, fmt.Errorf("there is no issue %d: %w", n, ErrNoIssue)
}
return &View{Issue: is, NeededBy: neededBy(all, n), WaitsFor: waitsFor(all, is), Ready: isReady(all, is)}, nil
}
// markDelivered records that a pull request is delivered on every issue whose fixed-by names it, and answers
// those issues' numbers.
func (t *Tracker) markDelivered(pr string, only uint64) ([]uint64, error) {
all, err := t.allIssues()
if err != nil {
return nil, err
}
var marked []uint64
for n, is := range all {
if (only != 0 && n != only) || !contains(pullRequests(is), pr) || contains(is.Delivered, pr) {
continue
}
if _, err := t.change(n, func(i *Issue) error {
if !contains(i.Delivered, pr) {
i.Delivered = append(i.Delivered, pr)
}
return nil
}); err != nil {
return marked, err
}
marked = append(marked, n)
}
sort.Slice(marked, func(i, j int) bool { return marked[i] < marked[j] })
return marked, nil
return &View{Issue: is, NeededBy: neededBy(all, n), WaitsFor: waitsFor(all, is), Ready: isReady(all, is),
Delivery: DeliveryLine(is)}, nil
}
// letGo says, for an issue that may have just let others go (its fix delivered, or wontfix), every issue it
@@ -369,112 +343,3 @@ func (t *Tracker) letGo(dep uint64) error {
}
return nil
}
// DeliveryState answers a pull request's delivery state, as mesh-delivery's `checks` says it; a test stubs it.
type DeliveryState func(repository string, number string) (string, error)
// busDeliveryState asks mesh-delivery's checks over the bus.
func busDeliveryState(owner string) DeliveryState {
return func(repository, number string) (string, error) {
raw, err := stdio.Ask("seat:mesh-delivery.checks", map[string]any{"repository": owner + "/" + repository, "number": number})
if err != nil {
return "", err
}
var out struct {
Delivery struct {
State string `json:"state"`
} `json:"delivery"`
}
if err := json.Unmarshal(raw, &out); err != nil {
return "", fmt.Errorf("mesh-delivery's checks answered no delivery: %.200s", raw)
}
return out.Delivery.State, nil
}
}
// OnTransition is the handler of mesh-delivery's `transition` event: a pull request delivered marks every issue
// whose fixed-by names it, and lets go what waited on them.
func (t *Tracker) OnTransition(e stdio.Envelope) error {
var body struct {
Repository string `json:"repository"`
Number int `json:"number"`
To string `json:"to"`
}
if err := json.Unmarshal(e.Body, &body); err != nil || body.To != "delivered" || body.Number <= 0 {
return nil
}
_, repo, ok := strings.Cut(body.Repository, "/")
if !ok {
repo = body.Repository
}
pr := fmt.Sprintf("%s#%d", strings.ToLower(repo), body.Number)
marked, err := t.markDelivered(pr, 0)
if err != nil {
return err
}
for _, n := range marked {
if err := t.letGo(n); err != nil {
return err
}
}
return nil
}
// OnMoved is the handler of the tracker's own `moved` event: an issue resolved has each pull request of its fix
// asked of mesh-delivery once, for a delivery that happened before it was resolved; an issue resolved or
// closed lets go what waited on it.
func (t *Tracker) OnMoved(e stdio.Envelope) error {
var body struct {
Number uint64 `json:"number"`
To string `json:"to"`
}
if err := json.Unmarshal(e.Body, &body); err != nil || body.Number == 0 {
return nil
}
if body.To != StatusResolved && body.To != StatusWontfix {
return nil
}
is, _, err := read(t.Store, body.Number)
if errors.Is(err, ErrNoIssue) {
return nil
}
if err != nil {
return err
}
if body.To == StatusResolved && t.Deliveries != nil {
// At most a few asks per handling, each up to the bus's 30 seconds, under the runtime's two minutes; one
// that fails, or more left to ask, has the event offered again, so a delivery not asked is never silent:
// past the last offer the controller raises max-deliveries.
asked := 0
for _, pr := range pullRequests(is) {
if contains(is.Delivered, pr) || contains(is.Undelivered, pr) {
continue
}
if asked == MostDeliveryAsks {
return fmt.Errorf("issue %d names more pull requests than are asked at once; the rest at the next offer", body.Number)
}
asked++
repo, number, _ := strings.Cut(pr, "#")
state, err := t.Deliveries(repo, number)
if err != nil {
return fmt.Errorf("the delivery of %s (issue %d) could not be asked: %v", pr, body.Number, err)
}
if state == "delivered" {
if _, err := t.markDelivered(pr, body.Number); err != nil {
return err
}
continue
}
// Asked and not delivered yet: the delivery's own event marks it when it comes.
if _, err := t.change(body.Number, func(i *Issue) error {
if !contains(i.Undelivered, pr) {
i.Undelivered = append(i.Undelivered, pr)
}
return nil
}); err != nil {
return err
}
}
}
return t.letGo(body.Number)
}
+21 -10
View File
@@ -30,9 +30,11 @@ type Tracker struct {
Records Records
// Judge is the triage's way to a model (ADR 0278, ADR 0279); nil judges nothing.
Judge Judgement
// Deliveries answers a pull request's delivery state (ADR 0279 §3); nil asks nothing.
// Deliveries answers a pull request's delivery (ADR 0279 §3, ADR 0281); nil asks nothing.
Deliveries DeliveryState
groom grooming
// Notify tells the operator's channel of a stuck fix (ADR 0281); nil says nothing there.
Notify Notifier
groom grooming
}
// The events this module says, one per kind of change.
@@ -395,12 +397,14 @@ func (t *Tracker) move(by string, a MoveAsk) (*Moved, error) {
is, err := t.change(a.Number, func(is *Issue) error {
changed = nil
from = is.Status
if from == to {
return fmt.Errorf("issue %d is already %s", is.Number, to)
}
// A field counts as changed only when its value does, so a move that changes nothing is refused.
set := func(name string, cond bool, apply func()) {
if cond {
apply()
if !cond {
return
}
before, _ := json.Marshal(is)
apply()
if after, _ := json.Marshal(is); string(before) != string(after) {
changed = append(changed, name)
}
}
@@ -410,14 +414,19 @@ func (t *Tracker) move(by string, a MoveAsk) (*Moved, error) {
set("replay", replay != "", func() { is.Replay, is.ReplayNone = replay, "" })
set("replay-none", replayNone != "", func() { is.ReplayNone, is.Replay = replayNone, "" })
set("reason", reason != "", func() { is.Reason = reason })
// The same status again only to change what it holds: a fix named while the issue stays located, to be
// resolved once it is delivered and proven (ADR 0281 §3).
if from == to && len(changed) == 0 {
return fmt.Errorf("issue %d is already %s", is.Number, to)
}
if err := cycleRules(is, to); err != nil {
return err
}
is.Status = to
if to == StatusResolved || to == StatusWontfix {
if (to == StatusResolved || to == StatusWontfix) && from != to {
at := t.now()
is.Closed = &at
} else {
} else if to != StatusResolved && to != StatusWontfix {
is.Closed = nil
}
return nil
@@ -622,6 +631,8 @@ type Row struct {
Ready bool `json:"ready"`
// SuggestedOwner is the triage's or a person's suggestion, for an issue located nowhere yet.
SuggestedOwner string `json:"suggested-owner,omitempty"`
// Delivery is whether its fix is delivered, in words (ADR 0281); empty when fixed-by names no pull request.
Delivery string `json:"delivery,omitempty"`
}
// Listed is the answer to a list.
@@ -700,7 +711,7 @@ func (t *Tracker) List(a ListAsk) (*Listed, error) {
out.Matched++
if len(out.Rows) < limit {
out.Rows = append(out.Rows, Row{Number: n, Status: is.Status, Owner: strings.Join(is.LocatedIn, "; "), Title: is.Title,
Severity: is.Severity, Needs: is.Needs, Ready: ready, SuggestedOwner: is.SuggestedOwner})
Severity: is.Severity, Needs: is.Needs, Ready: ready, SuggestedOwner: is.SuggestedOwner, Delivery: DeliveryLine(is)})
}
}
switch {
@@ -457,12 +457,12 @@ func TestAnIssueIsReadyOnlyWhenWhatItDependsOnIsDelivered(t *testing.T) {
t.Fatal("ready wrong with an open dependency")
}
asked := 0
h.Deliveries = func(repo, number string) (string, error) {
h.Deliveries = func(repo, number string) (Delivery, error) {
asked++
if repo != "mesh-catalog" || number != "9" {
t.Errorf("asked %s#%s", repo, number)
}
return "ready", nil
return Delivery{State: "ready"}, nil
}
resolve(t, h, dep, "novox/mesh-catalog#9", "a commit abc123")
if err := h.OnMoved(stdio.Envelope{Body: json.RawMessage(`{"number":365,"to":"resolved"}`)}); err != nil {
@@ -505,7 +505,7 @@ func TestAnIssueIsReadyOnlyWhenWhatItDependsOnIsDelivered(t *testing.T) {
w3 := h2.open(t, "waits on the abandoned one")
_, _ = h2.Link(LinkAsk{Number: w2, Kind: LinkDependsOn, Target: keyOf(d2)})
_, _ = h2.Link(LinkAsk{Number: w3, Kind: LinkDependsOn, Target: keyOf(d3)})
h2.Deliveries = func(string, string) (string, error) { return "delivered", nil }
h2.Deliveries = func(string, string) (Delivery, error) { return Delivery{State: "delivered"}, nil }
resolve(t, h2, d2, "mesh-controller#5")
_ = h2.OnMoved(stdio.Envelope{Body: json.RawMessage(`{"number":365,"to":"resolved"}`)})
_, _ = h2.Move(MoveAsk{Number: d3, Status: StatusWontfix, Reason: "abandoned"})
@@ -761,7 +761,7 @@ func TestReadyIsSaidOnceFollowsDuplicatesAndAskingFailsLoudly(t *testing.T) {
w := h.open(t, "waits on both")
_, _ = h.Link(LinkAsk{Number: w, Kind: LinkDependsOn, Target: keyOf(a)})
_, _ = h.Link(LinkAsk{Number: w, Kind: LinkDependsOn, Target: keyOf(b)})
h.Deliveries = func(string, string) (string, error) { return "merged", nil }
h.Deliveries = func(string, string) (Delivery, error) { return Delivery{State: "published"}, nil }
resolve(t, h, a, "Mesh-Catalog#9")
resolve(t, h, b, "https://forge.example/novox/mesh-catalog/pulls/9")
if got := pullRequests(get(t, h, a)); !reflect.DeepEqual(got, []string{"mesh-catalog#9"}) {
@@ -795,12 +795,12 @@ func TestReadyIsSaidOnceFollowsDuplicatesAndAskingFailsLoudly(t *testing.T) {
if v, _ := h2.View(w2); v.Ready {
t.Fatal("ready while the original is open")
}
h2.Deliveries = func(string, string) (string, error) { return "", errors.New("mesh-delivery did not answer") }
h2.Deliveries = func(string, string) (Delivery, error) { return Delivery{}, errors.New("mesh-delivery did not answer") }
resolve(t, h2, orig, "mesh-catalog#3")
if err := h2.OnMoved(stdio.Envelope{Body: json.RawMessage(`{"number":365,"to":"resolved"}`)}); err == nil {
t.Fatal("a delivery that could not be asked was taken")
}
h2.Deliveries = func(string, string) (string, error) { return "delivered", nil }
h2.Deliveries = func(string, string) (Delivery, error) { return Delivery{State: "delivered"}, nil }
if err := h2.OnMoved(stdio.Envelope{Body: json.RawMessage(`{"number":365,"to":"resolved"}`)}); err != nil {
t.Fatal(err)
}
File diff suppressed because one or more lines are too long