A-206: the held wait's clock is the build.ended of the awaited runner, heard from the builder's stream

The runner hears the builder's stream of build ends (kept last per subject, read whole at its start, A-181: consumes
`mesh-delivery-builder.build.ended.**`) and keeps when each build ended, the stream's own time. wait-held carries it as
`clock-from`, the bound of the wait is armed from it (30 minutes less what passed), and a bound whose build has not ended is armed
when its end is heard, with wait-held said again. A failed end starts no clock.

Builds: design 47 G §266, §269, §703, A-181, A-206

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends.go

@@ -0,0 +1,44 @@
+package main
+
+import (
+	"context"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"mesh-delivery-pipeline/internal/pipeline"
+)
+
+// hearBuildEnds hears the builder's stream of build ends, kept last per subject and read whole at the start, then followed
+// (design 47 G A-181, §556): when the build of the runner's new version ended is the time the wait's bound counts from, and
+// each end heard for a runner held pipelines await starts their clock (§266, A-206). A stream it cannot hear is said, and the
+// bound of such a wait is not armed until the runner starts again.
+func hearBuildEnds(ctx context.Context, cred process.Account, js jetstream.JetStream, ends *pipeline.BuildEnds, st *pipeline.Starter, store pipeline.Updater,
+	logf func(string, ...any)) {
+	stream := catalogue.StreamBusName(builds.BuildsStream)
+	cc, err := cred.HearStream(ctx, js, stream, func(m jetstream.Msg) {
+		defer m.Ack()
+		md, err := m.Metadata()
+		if err != nil {
+			logf("%s has no time the stream stamped: %v", m.Subject(), err)
+			return
+		}
+		module, fp, ok := ends.Heard(m.Data(), md.Timestamp)
+		if !ok {
+			return
+		}
+		if n, err := st.Clocked(store, module, fp); err != nil {
+			logf("the clock of the %d pipelines held for %s: %v", n, fp, err)
+		} else if n > 0 {
+			logf("the build of %s ended: the bound of the %d pipelines held for it counts from it", fp, n)
+		}
+	}, logf)
+	if err != nil {
+		logf("the stream %s is not heard, so no wait for the runner's new version has its bound armed until the runner starts again: %v", stream, err)
+		return
+	}
+	go func() { <-ctx.Done(); cc.Stop() }()
+}

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends_integration_test.go

@@ -0,0 +1,76 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
+	seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/process"
+	"git.novox.be/novox/mesh-sdk/go/testbus"
+
+	"mesh-delivery-pipeline/internal/pipeline"
+)
+
+// The runner hears the builder's stream of build ends, kept last per subject, read whole at its start: the end of its new version's
+// build, published before it started, starts the clock of the pipelines held for it, counted from the time the stream stamped the end
+// (design 47 G A-181, §266, A-206).
+func TestTheRunnerHearsTheBuildEndOfItsNewVersionAtItsStart(t *testing.T) {
+	nc, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	stream, subject := catalogue.StreamBusName(builds.BuildsStream), process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEvent+".>")
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}, MaxMsgsPerSubject: 1}); err != nil {
+		t.Fatal(err)
+	}
+	published := time.Now().UTC()
+	if _, err := js.Publish(ctx, process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEventName("mesh-delivery-pipeline", "src2:new")),
+		[]byte(`{"module":"mesh-delivery-pipeline","fingerprint":"src2:new","outcome":"done"}`)); err != nil {
+		t.Fatal(err)
+	}
+	cred := process.Account{Node: "n1", Module: "mesh-delivery-pipeline"}
+	durable := cred.SeatEventConsumerConfig([]string{subject}, true)
+	durable.DeliverPolicy = jetstream.DeliverNewPolicy
+	if _, err := js.CreateOrUpdateConsumer(ctx, stream, durable); err != nil {
+		t.Fatal(err)
+	}
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "pipelines", History: 1})
+	if err != nil {
+		t.Fatal(err)
+	}
+	store := pipelineStore{ctx: ctx, kv: kv}
+	held, _ := json.Marshal(seat.Pipeline{Pipeline: "main.4.app", Branch: "main", Module: "app", Outcome: seat.PipelineOutcomeRunning,
+		Held: &seat.Held{Since: published, Fingerprint: "src2:new"}})
+	if _, err := kv.Create(ctx, "main.4.app", held); err != nil {
+		t.Fatal(err)
+	}
+	ends := &pipeline.BuildEnds{}
+	armed := make(chan time.Time, 1)
+	st := &pipeline.Starter{Self: "mesh-delivery-pipeline", Ends: ends.At, Say: func(string, any) error { return nil },
+		Hold: func(fp string, from time.Time) error { armed <- from; return nil }}
+	hearBuildEnds(ctx, cred, js, ends, st, store, t.Logf)
+	select {
+	case from := <-armed:
+		if from.Before(published.Add(-time.Minute)) || from.After(time.Now().Add(time.Second)) {
+			t.Errorf("the bound counts from %s, not from the end published at %s", from, published)
+		}
+	case <-ctx.Done():
+		t.Fatal("the end of the new runner's build was not heard at the start")
+	}
+}

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go

@@ -104,12 +104,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	hearer.Where = routerRecordOf(cred.Module, calls.Call, calls.Ask)
 	// Its asking half (§22, §24): a pipeline's steps are placed from the module's registration and each build seat is asked
 	// for its step.
+	ends := &pipeline.BuildEnds{}
 	starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say,
 		Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module,
 		Arm: armBound(ctx, bounds), Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains),
 		// The runner's own module and fingerprint (A-149): a take-in giving it another holds the pipelines it starts (§65, A-195).
 		Self: cred.Module, SelfFp: getenv("MESH_FINGERPRINT"), Awaiting: pipeline.Awaiting(pipelineStore{ctx: ctx, kv: pipelines}),
-		Hold: armHeld(ctx, bounds),
+		Ends: ends.At, Hold: armHeld(ctx, bounds),
 		// The bound's expiry writes the finding pipeline-waits for each pipeline it started (§266, A-206).
 		Late: func(p seat.Pipeline, held seat.Held) {
 			module, node := rv.Findings.ModuleNode()
@@ -153,6 +154,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}, logger.Printf); err != nil {
 		return err
 	}
+	hearBuildEnds(ctx, cred, js, ends, starter, store, logger.Printf)
 	if err := hearRunning(ctx, heard, cred.Module, starter, store, bounds, rv.Catalog.View, logger.Printf); err != nil {
 		return err
 	}

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -174,10 +175,11 @@ func armBound(ctx context.Context, bounds jetstream.KeyValue) pipeline.Arm {
 }
 
 // armHeld is the bound of the pipelines held for the runner's new version: a key of the state bounds named for the awaited
-// fingerprint, with its own time to live of 30 minutes (design 47 G §266, A-195, A-77).
-func armHeld(ctx context.Context, bounds jetstream.KeyValue) func(string) error {
-	return func(fp string) error {
-		_, err := bounds.Create(ctx, pipeline.HeldKey(fp), []byte(`{"fingerprint":"`+fp+`"}`), jetstream.KeyTTL(pipeline.HeldBound))
+// fingerprint, with its own time to live of 30 minutes counted from the end of that runner's build (design 47 G §266, A-195, A-77,
+// A-206).
+func armHeld(ctx context.Context, bounds jetstream.KeyValue) func(string, time.Time) error {
+	return func(fp string, from time.Time) error {
+		_, err := bounds.Create(ctx, pipeline.HeldKey(fp), []byte(`{"fingerprint":"`+fp+`"}`), jetstream.KeyTTL(pipeline.HeldTTL(from, time.Now())))
 		if errors.Is(err, jetstream.ErrKeyExists) {
 			return nil
 		}

modules/mesh-delivery-pipeline/go.mod

@@ -11,6 +11,7 @@ require (
 require (
 	git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
 	git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000
 	git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
 	git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient v0.0.0-00010101000000-000000000000
 	git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000

modules/mesh-delivery-pipeline/internal/pipeline/ends.go

@@ -0,0 +1,46 @@
+package pipeline
+
+import (
+	"encoding/json"
+	"sync"
+	"time"
+)
+
+// BuildEnds are the done builds' ends heard from the builder's stream of build ends (event build.ended, kept last per subject
+// and read whole at the start, design 47 G A-181): when each build, by module and fingerprint, ended, which is the time a held
+// wait's bound counts from (§266, A-206).
+type BuildEnds struct {
+	mu sync.Mutex
+	at map[string]time.Time
+}
+
+// buildEnd is the part of the body of build.ended the runner reads.
+type buildEnd struct {
+	Module      string `json:"module"`
+	Fingerprint string `json:"fingerprint"`
+	Outcome     string `json:"outcome"`
+}
+
+// Heard keeps the end a message of the stream says, at the time the stream stamped it; a failed end, or a body naming no module
+// and fingerprint, is kept for nothing. It says the module and fingerprint it kept.
+func (e *BuildEnds) Heard(data []byte, at time.Time) (module, fp string, ok bool) {
+	var b buildEnd
+	if err := json.Unmarshal(data, &b); err != nil || b.Module == "" || b.Fingerprint == "" || b.Outcome == "failed" {
+		return "", "", false
+	}
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.at == nil {
+		e.at = map[string]time.Time{}
+	}
+	e.at[b.Module+"\x00"+b.Fingerprint] = at.UTC()
+	return b.Module, b.Fingerprint, true
+}
+
+// At is when the build of the module at the fingerprint ended, false when no end was heard.
+func (e *BuildEnds) At(module, fp string) (time.Time, bool) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	t, ok := e.at[module+"\x00"+fp]
+	return t, ok
+}

modules/mesh-delivery-pipeline/internal/pipeline/hold.go

@@ -64,17 +64,68 @@ func (s *Starter) startHeld(b Begin, key string, now time.Time, awaiting string)
 	} else if err != nil {
 		return false, fmt.Errorf("the pipeline %s is not written: %w", key, err)
 	}
-	if s.Hold != nil {
-		if err := s.Hold(awaiting); err != nil {
+	from := s.clockOf(awaiting)
+	if from != nil && s.Hold != nil {
+		if err := s.Hold(awaiting, *from); err != nil {
 			s.logf("the bound of the wait for the runner at %s is not armed: %v", awaiting, err)
 		}
 	}
-	if err := s.Say(runnerclient.Subject(seat.EventWaitHeld, b.Module, b.Branch, commitOrGroup(p)), s.waitOf(p, now, "")); err != nil {
+	w := s.waitOf(p, now, "")
+	w.ClockFrom = from
+	if err := s.Say(runnerclient.Subject(seat.EventWaitHeld, b.Module, b.Branch, commitOrGroup(p)), w); err != nil {
 		s.logf("wait-held of %s is not said: %v", key, err)
 	}
 	return true, nil
 }
 
+// clockOf is when the build of the runner at fp ended, the time the bound counts from, nil while it has not (§266, A-181, A-206).
+func (s *Starter) clockOf(fp string) *time.Time {
+	if s.Ends == nil {
+		return nil
+	}
+	if at, ok := s.Ends(s.Self, fp); ok {
+		return &at
+	}
+	return nil
+}
+
+// Clocked starts the clock of the pipelines held for the runner at fp, the end of whose build was heard: it arms the bound from that
+// end and says wait-held again for each with `clock-from`, so the watchman times it by the bound (§269, A-206). It says how many
+// pipelines it said, and nothing when none is held for fp or the end is not known.
+func (s *Starter) Clocked(store Updater, module, fp string) (int, error) {
+	if module != s.Self {
+		return 0, nil
+	}
+	from := s.clockOf(fp)
+	if from == nil {
+		return 0, nil
+	}
+	pkeys, err := store.Keys("*.*.*")
+	if err != nil {
+		return 0, fmt.Errorf("the pipelines are not listed: %w", err)
+	}
+	said := 0
+	var first error
+	for _, key := range pkeys {
+		p, _, err := store.Get(key)
+		if err != nil || p.Held == nil || p.Outcome != seat.PipelineOutcomeRunning || p.Held.Fingerprint != fp {
+			continue
+		}
+		if said == 0 && s.Hold != nil {
+			if err := s.Hold(fp, *from); err != nil {
+				first = fmt.Errorf("the bound of the wait for the runner at %s is not armed: %w", fp, err)
+			}
+		}
+		w := s.waitOf(p, p.Held.Since, "")
+		w.ClockFrom = from
+		if err := s.Say(runnerclient.Subject(seat.EventWaitHeld, p.Module, p.Branch, commitOrGroup(p)), w); err != nil && first == nil {
+			first = fmt.Errorf("wait-held of %s is not said: %w", key, err)
+		}
+		said++
+	}
+	return said, first
+}
+
 func commitOrGroup(p seat.Pipeline) string {
 	if p.Group != "" {
 		return p.Group
@@ -210,3 +261,9 @@ func (v Verbs) Release(_ string, in seat.ReleaseInput) (seat.ReleaseAnswer, erro
 	}
 	return seat.ReleaseAnswer{Released: int64(n)}, err
 }
+
+// HeldTTL is how long the bound of a wait that counts from the build end at from still has to run at now: the 30 minutes less what passed since
+// the end, at least a second, the least a key's time to live holds (design 47 G §266, A-206).
+func HeldTTL(from, now time.Time) time.Duration {
+	return max(HeldBound-now.Sub(from), time.Second)
+}

modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go

@@ -39,7 +39,8 @@ func TestPipelinesAreHeldForTheNewRunnerAndReleasedWhenItRuns(t *testing.T) {
 	var said []string
 	var bounds []string
 	s := &Starter{Pipelines: store, Steps: asks, Name: "the-runner", Self: "runner", SelfFp: "src2:old",
-		Awaiting: Awaiting(store.memStore), Hold: func(fp string) error { bounds = append(bounds, fp); return nil },
+		Awaiting: Awaiting(store.memStore), Ends: func(string, string) (time.Time, bool) { return time.Unix(9, 0), true },
+		Hold:         func(fp string, _ time.Time) error { bounds = append(bounds, fp); return nil },
 		Registration: func(m string) (catalog.Registration, bool) { return reg, m == "app" },
 		Say: func(subject string, _ any) error {
 			said = append(said, subject[strings.Index(subject, ".event.")+7:])
@@ -162,3 +163,72 @@ func TestHeldWaitIsOnTheRunnerVersion(t *testing.T) {
 		t.Errorf("wait %+v", w)
 	}
 }
+
+// The wait of a held pipeline counts from the build end of the awaited runner: wait-held says it as `clock-from` and the bound is
+// armed from it; while that build has not ended nothing is armed, and hearing its end arms the bound and says wait-held again
+// with `clock-from`; a failed end starts no clock (design 47 G §266, §269, A-181, A-206).
+func TestHeldWaitCountsFromTheBuildEndOfTheAwaitedRunner(t *testing.T) {
+	reg, entries := placeFixture()
+	store := memCreated{memStore{}}
+	ends := &BuildEnds{}
+	var waits []seat.Wait
+	var armed []string
+	s := &Starter{Pipelines: store, Steps: &stepAsks{}, Name: "the-runner", Self: "runner", SelfFp: "src2:old", Awaiting: Awaiting(store.memStore),
+		Ends: ends.At, Hold: func(fp string, from time.Time) error {
+			armed = append(armed, fp+"@"+from.Format(time.RFC3339))
+			return nil
+		},
+		Registration: func(string) (catalog.Registration, bool) { return reg, true },
+		Say: func(subject string, body any) error {
+			if w, ok := body.(seat.Wait); ok && strings.Contains(subject, ".wait-held.") {
+				waits = append(waits, w)
+			}
+			return nil
+		}}
+	view := func(string) (Half, bool) { return Half{Entries: entries}, true }
+	taken := catalog.TakeInTaken{Key: "m.1", Branch: "main", Source: 4, Entries: []catalog.TakeInEntry{
+		{Module: "app", Commit: "c1", Fingerprint: "src2:aa"}, {Module: "runner", Commit: "c2", Fingerprint: "src2:new"}}}
+	end := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)
+	// The runner's build has not ended: held, no clock, no bound.
+	if _, err := s.TakenIn(taken, view); err != nil {
+		t.Fatal(err)
+	}
+	if len(waits) != 2 || waits[0].ClockFrom != nil || len(armed) != 0 {
+		t.Fatalf("waits %+v, armed %v", waits, armed)
+	}
+	// A failed end starts no clock; a done end does.
+	if _, _, ok := ends.Heard([]byte(`{"module":"runner","fingerprint":"src2:new","outcome":"failed"}`), end); ok {
+		t.Error("a failed end was kept")
+	}
+	if m, fp, ok := ends.Heard([]byte(`{"module":"runner","fingerprint":"src2:new","outcome":"done"}`), end); !ok || m != "runner" || fp != "src2:new" {
+		t.Fatalf("heard %q %q %v", m, fp, ok)
+	}
+	waits = nil
+	if n, err := s.Clocked(store.memStore, "runner", "src2:new"); err != nil || n != 2 {
+		t.Fatalf("clocked = %d, %v", n, err)
+	}
+	if len(waits) != 2 || waits[0].ClockFrom == nil || !waits[0].ClockFrom.Equal(end) || len(armed) != 1 || armed[0] != "src2:new@2026-11-11T12:00:00Z" {
+		t.Errorf("waits %+v, armed %v", waits, armed)
+	}
+	// A pipeline held after the end is known says the clock at once.
+	waits = nil
+	taken.Key, taken.Source = "m.2", 5
+	if _, err := s.TakenIn(taken, view); err != nil {
+		t.Fatal(err)
+	}
+	if len(waits) == 0 || waits[0].ClockFrom == nil || !waits[0].ClockFrom.Equal(end) {
+		t.Errorf("waits %+v", waits)
+	}
+}
+
+// The bound counts from the build end: the 30 minutes less what passed since, never under a second.
+func TestHeldTTLCountsFromTheBuildEnd(t *testing.T) {
+	end := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)
+	for _, c := range []struct {
+		after, want time.Duration
+	}{{0, 30 * time.Minute}, {10 * time.Minute, 20 * time.Minute}, {45 * time.Minute, time.Second}} {
+		if got := HeldTTL(end, end.Add(c.after)); got != c.want {
+			t.Errorf("after %s the bound has %s, want %s", c.after, got, c.want)
+		}
+	}
+}

modules/mesh-delivery-pipeline/internal/pipeline/start.go

@@ -74,10 +74,12 @@ type Starter struct {
 	Arm Arm
 	// Self is the runner's own module and SelfFp the fingerprint it runs (A-149); a take-in giving Self another fingerprint
 	// holds its pipelines for the new runner (§65, §266, A-195). Awaiting is the fingerprint the held pipelines await, while
-	// any is held; Hold arms the bound of that wait (HeldBound). All nil: nothing is held.
+	// any is held; Hold arms the bound of that wait (HeldBound) counting from the time the awaited build ended. All nil: nothing is held.
 	Self, SelfFp string
 	Awaiting     func() (string, bool)
-	Hold         func(fp string) error
+	Hold         func(fp string, from time.Time) error
+	// Ends says when the build of a module at a fingerprint ended (BuildEnds.At): the clock of the wait's bound (A-181, A-206).
+	Ends func(module, fp string) (time.Time, bool)
 	// Late says a pipeline the bound's expiry started with the running runner (A-206): the finding pipeline-waits.
 	Late func(p seat.Pipeline, held seat.Held)
 }

modules/mesh-delivery-pipeline/manifest_test.go

@@ -75,7 +75,7 @@ func TestManifest(t *testing.T) {
 		{"sent-objects 256 MiB", objectMiB[catalogue.SentObjectsLocal] == 256},
 		// §333, §54, §44, §53, §376, §24, §37, §4, §77: what the runner hears.
 		{"consumes", equalSets(m.Consumes, []string{
-			"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended",
+			"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended", "mesh-delivery-builder.build.ended.**",
 			"operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended",
 			"node-build-image.step-ended", "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", "node-engine.running", "mesh-wants.step-ended", "mesh-catalog.registration-changed",
 			"mesh-catalog.catalog-revision", "mesh-conditions.findings-asked"})},

modules/mesh-delivery-pipeline/module.json

@@ -35,6 +35,7 @@
     "mesh-builds.package-published",
     "mesh-builds.toolchain-agreed",
     "mesh-builds.step-ended",
+    "mesh-delivery-builder.build.ended.**",
     "operator-channel.decided",
     "node-build-go.step-ended",
     "node-build-node.step-ended",