A-206: the seat verb release and the tool runner_release start the held pipelines by hand

The seat mesh-pipeline-runner gains the act verb `release` (why; answers how many it started) beside start and ask-again, its
hand act written first through mesh-hand-acts; each released wait ends `released` and its bound is dropped. The tool
runner_release asks the verb. Contract regenerated with seatgen.

Builds: design 47 G §276, §534, A-206 (A-195)

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 06:01:22 +01:00
parent e053605e82
commit 6635dda18f
12 changed files with 176 additions and 9 deletions
@@ -27,6 +27,7 @@ const (
VerbAccept = "accept"
VerbCancel = "cancel"
VerbAskAgain = "ask-again"
VerbRelease = "release"
)
const (
@@ -42,7 +43,7 @@ const (
// Verbs is every verb the seat serves, in its definition's order: a new copy each call.
func Verbs() []string {
return []string{VerbRun, VerbStart, VerbShow, VerbList, VerbStands, VerbWhyCarried, VerbKept, VerbReview, VerbRerun, VerbAccept, VerbCancel, VerbAskAgain}
return []string{VerbRun, VerbStart, VerbShow, VerbList, VerbStands, VerbWhyCarried, VerbKept, VerbReview, VerbRerun, VerbAccept, VerbCancel, VerbAskAgain, VerbRelease}
}
// Events is every event published on the seat's subject: a new copy each call.
@@ -959,6 +960,18 @@ type AskAgainAnswer struct {
Result StepResult `json:"result"`
}
// ReleaseInput is what release takes.
type ReleaseInput struct {
// why, written with the caller the runtime proves in the hand-act log
Why string `json:"why"`
}
// ReleaseAnswer is what release answers.
type ReleaseAnswer struct {
// how many held pipelines were started
Released int64 `json:"released"`
}
// PipelineStartedEvent is the body of the event pipeline-started.
type PipelineStartedEvent = PipelineStarted
@@ -501,6 +501,35 @@
]
},
"class": "act"
},
{
"name": "release",
"description": "The pipelines the runner holds for its new version started by hand with the running runner, whatever the new version's state; written in the hand-act log (design 47 G 276, A-206).",
"input": {
"type": "object",
"properties": {
"why": {
"type": "string",
"description": "why, written with the caller the runtime proves in the hand-act log"
}
},
"required": [
"why"
]
},
"answer": {
"type": "object",
"properties": {
"released": {
"type": "integer",
"description": "how many held pipelines were started"
}
},
"required": [
"released"
]
},
"class": "act"
}
],
"events": [
@@ -17,6 +17,7 @@ export declare const verbs: {
readonly accept: "accept";
readonly cancel: "cancel";
readonly askAgain: "ask-again";
readonly release: "release";
};
export declare const events: {
@@ -578,6 +579,18 @@ export interface AskAgainAnswer {
"result": StepResult;
}
/** is what release takes. */
export interface ReleaseInput {
/** why, written with the caller the runtime proves in the hand-act log */
"why": string;
}
/** is what release answers. */
export interface ReleaseAnswer {
/** how many held pipelines were started */
"released": number;
}
/** is the body of the event pipeline-started. */
export type PipelineStartedEvent = PipelineStarted;
@@ -17,6 +17,7 @@ export const verbs = {
accept: "accept",
cancel: "cancel",
askAgain: "ask-again",
release: "release",
};
export const events = {
@@ -118,10 +118,28 @@ func WaitsTool(r pipeline.Reads) stdio.Tool {
Run: func(map[string]any) (any, error) { return r.Waits() }}
}
// ReleaseTool is the runner's tool runner_release: it starts the pipelines held for the runner's new version with the running
// runner, through the seat's verb release, which the process that holds the pipelines answers and records in the hand-act log
// first (design 47 G §276, A-206).
func ReleaseTool(ask func(key string, body any) (json.RawMessage, error)) stdio.Tool {
return stdio.Tool{Name: Prefix + "_release",
Description: "Start now, with the running pipeline runner, the pipelines it holds until its new version runs healthy " +
"(runner_waits lists them), each wait ending released. Written in the hand-act log (mesh-hand-acts).",
Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{
"why": map[string]any{"type": "string", "description": "why, in plain words"}}},
Run: func(args map[string]any) (any, error) {
in, err := as[seat.ReleaseInput](args)
if err != nil {
return nil, err
}
return ask("seat:"+seat.Seat+"."+seat.VerbRelease, in)
}}
}
func main() {
r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}}
tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...)
tools = append(tools, WaitsTool(r))
tools = append(tools, WaitsTool(r), ReleaseTool(stdio.Ask))
tools = append(tools, HealthTool(r.Pipelines, kept{stdio.State(pipeline.StateConsumers)}, time.Now))
if err := stdio.Serve("", tools); err != nil {
fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] %v\n", err)
@@ -2,6 +2,7 @@ package main
import (
"encoding/json"
"fmt"
"strings"
"testing"
@@ -28,7 +29,17 @@ func TestTools(t *testing.T) {
}
// RN25 (§468, §76; A-164): the health check is the bundle's runner_health.
names = append(names, HealthTool(none{}, none{}, nil).Name)
want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again runner_health"
// A-206: the tool runner_release asks the seat's verb release with its why.
var asked string
rel := ReleaseTool(func(key string, body any) (json.RawMessage, error) {
asked = key + fmt.Sprintf(" %+v", body)
return json.RawMessage(`{"released":1}`), nil
})
if _, err := rel.Run(map[string]any{"why": "w"}); err != nil || asked != "seat:mesh-pipeline-runner.release {Why:w}" {
t.Errorf("runner_release asked %q, %v", asked, err)
}
names = append(names, rel.Name)
want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again runner_health runner_release"
if got := strings.Join(names, " "); got != want {
t.Errorf("tools %q, want %q", got, want)
}
@@ -38,8 +38,8 @@ func answerVerbs(conn *process.Conn, v pipeline.Verbs) (func(), error) {
return conn.AnswerSeat(seat.Seat, writeHandlers(v))
}
// writeHandlers are the seat's write verbs the process answers: review, rerun, accept, cancel, start and ask-again
// (§306; RN14).
// writeHandlers are the seat's write verbs the process answers: review, rerun, accept, cancel, start, ask-again and
// release (§306; RN14; A-206).
func writeHandlers(v pipeline.Verbs) map[string]process.SeatHandler {
return map[string]process.SeatHandler{
seat.VerbReview: verbOf(func(c string, in seat.ReviewInput) (any, error) { return v.Review(c, in) }),
@@ -48,6 +48,7 @@ func writeHandlers(v pipeline.Verbs) map[string]process.SeatHandler {
seat.VerbCancel: verbOf(func(c string, in seat.CancelInput) (any, error) { return v.Cancel(c, in) }),
seat.VerbStart: verbOf(func(c string, in seat.StartInput) (any, error) { return v.Start(c, in) }),
seat.VerbAskAgain: verbOf(func(c string, in seat.AskAgainInput) (any, error) { return v.AskAgain(c, in) }),
seat.VerbRelease: verbOf(func(c string, in seat.ReleaseInput) (any, error) { return v.Release(c, in) }),
}
}
@@ -18,7 +18,7 @@ func TestEveryWriteVerbIsServed(t *testing.T) {
got = append(got, v)
}
slices.Sort(got)
want := []string{seat.VerbAccept, seat.VerbAskAgain, seat.VerbCancel, seat.VerbRerun, seat.VerbReview, seat.VerbStart}
want := []string{seat.VerbAccept, seat.VerbAskAgain, seat.VerbCancel, seat.VerbRerun, seat.VerbRelease, seat.VerbReview, seat.VerbStart}
slices.Sort(want)
if !slices.Equal(got, want) {
t.Errorf("the process answers %v, want %v", got, want)
@@ -5,6 +5,7 @@ import (
"encoding/hex"
"errors"
"fmt"
"slices"
"time"
runnerclient "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient"
@@ -161,3 +162,40 @@ func (s *Starter) ReleaseExpired(store Updater, key string, view func(branch str
}
return 0, nil
}
// heldFingerprints are the fingerprints of the runner the held pipelines in store await, each once.
func heldFingerprints(store Updater) []string {
pkeys, err := store.Keys("*.*.*")
if err != nil {
return nil
}
var fps []string
for _, key := range pkeys {
if p, _, err := store.Get(key); err == nil && p.Held != nil && p.Outcome == seat.PipelineOutcomeRunning && !slices.Contains(fps, p.Held.Fingerprint) {
fps = append(fps, p.Held.Fingerprint)
}
}
return fps
}
// Release starts by hand the pipelines held for the runner's new version with the running runner, recorded in the hand-act
// log first (§276, A-206): each says wait-ended `released` and pipeline-started, and the bound of each wait is dropped. It
// says how many it started.
func (v Verbs) Release(_ string, in seat.ReleaseInput) (seat.ReleaseAnswer, error) {
if v.Starter == nil || v.View == nil {
return seat.ReleaseAnswer{}, errors.New("the runner cannot release a pipeline: it holds no starter or no catalog half")
}
if err := v.record(seat.VerbRelease, "started the pipelines held for the new runner by hand", in.Why); err != nil {
return seat.ReleaseAnswer{}, err
}
fps := heldFingerprints(v.Hearer.Store)
n, err := v.Starter.Release(v.Hearer.Store, "", "released", v.View)
if n > 0 && v.Drop != nil {
for _, fp := range fps {
if derr := v.Drop(HeldKey(fp)); derr != nil && err == nil {
err = fmt.Errorf("the bound of the wait for %s is not dropped: %w", fp, derr)
}
}
}
return seat.ReleaseAnswer{Released: int64(n)}, err
}
@@ -84,3 +84,44 @@ func TestPipelinesAreHeldForTheNewRunnerAndReleasedWhenItRuns(t *testing.T) {
t.Error("a pipeline is still held")
}
}
// The verb release starts the held pipelines by hand, recorded in the hand-act log first, saying why and ending each wait
// `released`, and drops the bound of the wait; a release with no why does nothing (design 47 G §276, A-206).
func TestReleaseVerbStartsTheHeldPipelinesByHand(t *testing.T) {
reg, entries := placeFixture()
entries["runner"] = catalog.CatalogEntry{Module: "runner"}
store := memCreated{memStore{}}
asks := &stepAsks{}
var said []string
s := &Starter{Pipelines: store, Steps: asks, Name: "the-runner", Self: "runner", SelfFp: "src2:old", Awaiting: Awaiting(store.memStore),
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:])
return nil
}}
view := func(string) (Half, bool) { return Half{Entries: entries}, true }
if _, err := s.TakenIn(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"}}}, view); err != nil {
t.Fatal(err)
}
acts, dropped := &actLog{}, []string{}
v := Verbs{Hearer: &Hearer{Store: store.memStore}, Starter: s, View: view, Acts: acts,
Drop: func(k string) error { dropped = append(dropped, k); return nil }}
if _, err := v.Release("operator", seat.ReleaseInput{}); err == nil || len(*acts) != 0 {
t.Fatalf("a release saying no why: %v, acts %v", err, *acts)
}
said = nil
a, err := v.Release("operator", seat.ReleaseInput{Why: "the new runner will not come"})
if err != nil || a.Released != 2 {
t.Fatalf("release = %+v, %v", a, err)
}
if p, _, _ := store.Get("main.4.app"); p.Held != nil || len(p.Steps) == 0 || len(asks.asks) == 0 {
t.Errorf("after the release: held %+v, steps %d", p.Held, len(p.Steps))
}
if len(*acts) != 1 || !strings.HasPrefix((*acts)[0], "mesh-pipeline-runner.release: ") || len(dropped) != 1 || dropped[0] != HeldKey("src2:new") {
t.Errorf("acts %v, dropped %v", *acts, dropped)
}
if !strings.Contains(strings.Join(said, "|"), "wait-ended.app.main.c1") {
t.Errorf("said %v", said)
}
}
@@ -91,7 +91,7 @@ func TestManifest(t *testing.T) {
"seat:node-build-image.check.step"})},
{"uses operator-channel", equalSets(m.Uses, []string{"operator-channel"})},
// RN25 (§539): the runner's tools carry the prefix runner_.
{"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health", "runner_waits"})},
{"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health", "runner_waits", "runner_release"})},
// RN25 (§468, §76): the health check, run outside the process, is the tool runner_health.
{"health is runner_health", healthTool(raw) == "tool runner_health"},
// RN1/AC5 (§95, §130): the new assigner is delivered before the runner.
+4 -2
View File
@@ -18,7 +18,8 @@
"rerun",
"accept",
"cancel",
"ask-again"
"ask-again",
"release"
]
}
],
@@ -26,7 +27,8 @@
"runner_first_views",
"runner_first_view_again",
"runner_health",
"runner_waits"
"runner_waits",
"runner_release"
],
"consumes": [
"mesh-delivery-indexer.take-in.taken",