From 7125a9f92f1f3ce3e0ebcd4b4eff00fa83fe085a Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:37:51 +0100 Subject: [PATCH] C1c-2a: the step runner says a run's answer in step-ended, inline or as an object, and serves it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The puller's end of a run carries the run's program's last line of standard output as `output` up to StepEndedInlineMost, else keeps it with KeepOutput and names the object in `output-object`; a program that exits other than 0 ends failed. OutputTool serves the seat's verb `output` from the kept object, as the whole file the tool runner sends in chunks; OutputsStore names the store. Builds: design 47 G A-254, A-257, §602 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../buildstep/pipelinestep.go | 11 +++ modules/mesh-delivery-sdk/steprun/askers.go | 2 +- modules/mesh-delivery-sdk/steprun/outputs.go | 42 ++++++++++ .../mesh-delivery-sdk/steprun/outputs_test.go | 42 ++++++++++ modules/mesh-delivery-sdk/steprun/pull.go | 49 +++++++++-- .../steprun/run_output_test.go | 83 +++++++++++++++++++ .../mesh-delivery-sdk/steprun/seat_tools.go | 30 +++++++ .../steprun/seat_tools_test.go | 24 ++++++ 8 files changed, 276 insertions(+), 7 deletions(-) create mode 100644 modules/mesh-delivery-sdk/steprun/outputs.go create mode 100644 modules/mesh-delivery-sdk/steprun/outputs_test.go create mode 100644 modules/mesh-delivery-sdk/steprun/run_output_test.go diff --git a/modules/mesh-delivery-sdk/buildstep/pipelinestep.go b/modules/mesh-delivery-sdk/buildstep/pipelinestep.go index 40e51f74..cee86f4a 100644 --- a/modules/mesh-delivery-sdk/buildstep/pipelinestep.go +++ b/modules/mesh-delivery-sdk/buildstep/pipelinestep.go @@ -39,6 +39,14 @@ var StepOutcomes = []string{OutcomePassed, OutcomeFailed, OutcomeNothingToRun, O // StepEndedVersion is the layout version of a step's end. const StepEndedVersion = 1 +// StepEndedInlineMost is the most bytes of a run's answer a step-ended carries inline as `output`: the bus's message limit +// less the room its headers take and 16 KiB for the end's other fields. An answer over it is kept by the holder as an object, +// which `output-object` names and the seat's verb `output` answers (design 47 G A-254). +const StepEndedInlineMost = (1 << 20) - (8 << 10) - (16 << 10) + +// OutputObjectName is the name of the object a build seat's holder keeps a run's answer in, by the ask's id. +func OutputObjectName(ask string) string { return "output." + ask } + // KeptResult is a package's passing result at the digest of what its tests read (ADR 0666 §8). type KeptResult struct { Package string `json:"package"` @@ -154,6 +162,9 @@ func (e StepEnded) check() error { if err := checkNode(e.Node); err != nil { return err } + if e.Output != "" && e.OutputObject != "" { + return fmt.Errorf("buildstep: the end of %s carries its answer inline and as an object", e.ID) + } if e.Pipeline == "" { // G §548, §602: the end of an ask outside a pipeline names its asker, the ask and when, and no commit or // fingerprint. diff --git a/modules/mesh-delivery-sdk/steprun/askers.go b/modules/mesh-delivery-sdk/steprun/askers.go index 3023ad38..eb2c77c6 100644 --- a/modules/mesh-delivery-sdk/steprun/askers.go +++ b/modules/mesh-delivery-sdk/steprun/askers.go @@ -106,7 +106,7 @@ func (p *Puller) refuseAsker(msg *Pulled, ask sdkstep.StepAsk, why string) { } p.logf("the ask %s is refused on %s: %s", ask.ID, p.Node, why) if ask.Step == nil { - p.sayEnded(msg, ask, sdkstep.OutcomeFailed, why) + p.sayEnded(msg, ask, sdkstep.OutcomeFailed, why, nil) } else if end, err := sdkstep.EndOf(ask, p.Node, sdkstep.OutcomeFailed); err == nil { // A pipeline step's end names its pipeline, step, commit and fingerprint (its own form, §602). end.Said, end.At, end.Asker = why, now().UTC(), askerOf(msg, ask) diff --git a/modules/mesh-delivery-sdk/steprun/outputs.go b/modules/mesh-delivery-sdk/steprun/outputs.go new file mode 100644 index 00000000..b228cf89 --- /dev/null +++ b/modules/mesh-delivery-sdk/steprun/outputs.go @@ -0,0 +1,42 @@ +package steprun + +import ( + "fmt" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// OutputsStore is the local name of the object store a build seat's holder keeps a run's answer in when it is over what +// step-ended carries inline (design 47 G A-254), as the holder's manifest `requires` asks for it with the template objects. +const OutputsStore = "outputs" + +// OutputObjects is the object store of OutputsStore as a holder uses it (mesh-sdk busstate.ObjectsOf(OutputsStore)). +type OutputObjects interface { + Put(name string, data []byte, description string) (buscalls.ObjectInfo, error) + Get(name string) (buscalls.ObjectData, error) +} + +// KeepOutputIn is a holder's Puller.KeepOutput over its store of outputs. +func KeepOutputIn(o OutputObjects) func(object string, body []byte) error { + return func(object string, body []byte) error { + if o == nil { + return fmt.Errorf("this module keeps no %s store", OutputsStore) + } + _, err := o.Put(object, body, "a run's answer, named by its step-ended") + return err + } +} + +// ReadOutputFrom is a holder's read of a kept answer for OutputTool over its store of outputs. +func ReadOutputFrom(o OutputObjects) func(object string) ([]byte, error) { + return func(object string) ([]byte, error) { + if o == nil { + return nil, fmt.Errorf("this module keeps no %s store", OutputsStore) + } + got, err := o.Get(object) + if err != nil { + return nil, fmt.Errorf("the answer %s in the %s store: %w", object, OutputsStore, err) + } + return got.Data, nil + } +} diff --git a/modules/mesh-delivery-sdk/steprun/outputs_test.go b/modules/mesh-delivery-sdk/steprun/outputs_test.go new file mode 100644 index 00000000..5013b04d --- /dev/null +++ b/modules/mesh-delivery-sdk/steprun/outputs_test.go @@ -0,0 +1,42 @@ +package steprun + +import ( + "errors" + "testing" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +type fakeOutputs struct{ m map[string][]byte } + +func (f fakeOutputs) Put(name string, data []byte, _ string) (buscalls.ObjectInfo, error) { + f.m[name] = data + return buscalls.ObjectInfo{}, nil +} +func (f fakeOutputs) Get(name string) (buscalls.ObjectData, error) { + d, ok := f.m[name] + if !ok { + return buscalls.ObjectData{}, errors.New("object not found") + } + return buscalls.ObjectData{Data: d}, nil +} + +// Design 47 G A-254: what a holder keeps through KeepOutputIn it reads again through ReadOutputFrom; with no store both refuse by name. +func TestOutputsKeptAreRead(t *testing.T) { + st := fakeOutputs{m: map[string][]byte{}} + if err := KeepOutputIn(st)("output.a1", []byte("x")); err != nil { + t.Fatal(err) + } + if got, err := ReadOutputFrom(st)("output.a1"); err != nil || string(got) != "x" { + t.Fatalf("read %q, %v", got, err) + } + if _, err := ReadOutputFrom(st)("output.none"); err == nil { + t.Fatal("an absent answer was read") + } + if KeepOutputIn(nil)("o", nil) == nil { + t.Fatal("no store kept an answer") + } + if _, err := ReadOutputFrom(nil)("o"); err == nil { + t.Fatal("no store read an answer") + } +} diff --git a/modules/mesh-delivery-sdk/steprun/pull.go b/modules/mesh-delivery-sdk/steprun/pull.go index 77a772d6..f95a8718 100644 --- a/modules/mesh-delivery-sdk/steprun/pull.go +++ b/modules/mesh-delivery-sdk/steprun/pull.go @@ -81,6 +81,10 @@ type Puller struct { // every ProgressEvery while it runs, step-ended when an ask outside a pipeline ends (design 47 G §496, §602). Nil says // none, a test's. ProgressEvery defaults to DefaultProgressEvery. Say func(event string, body any, id string) error + // KeepOutput keeps a run's answer over what step-ended carries inline (sdkstep.StepEndedInlineMost) as the object named + // object, which the seat's verb `output` answers (design 47 G A-254); a holder keeps it in the object store its manifest + // requires. Nil keeps none: such an answer ends the ask failed, saying so. + KeepOutput func(object string, body []byte) error // Callers says which seats the build seat's definition lists as callers of an accepted kind (CallersFromDefinitions); // found is false while the seat list holds no such kind. Nil checks no named asker (design 47 G §543, A-163). Callers func(kind string) (callers []string, found bool) @@ -753,7 +757,7 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key var dec *Declining if errors.As(err, &dec) && ctx.Err() == nil { p.declineRun(msg, ask, key, step, dec) - p.sayEnded(msg, ask, sdkstep.OutcomeCouldNotRun, dec.Text) + p.sayEnded(msg, ask, sdkstep.OutcomeCouldNotRun, dec.Text, nil) return } if ctx.Err() != nil { @@ -791,9 +795,9 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key if len(outcome) > 300 { outcome = outcome[:300] + "..." } - p.sayEnded(msg, ask, sdkstep.OutcomeFailed, err.Error()) + p.sayEnded(msg, ask, sdkstep.OutcomeFailed, err.Error(), nil) } else { - p.sayEnded(msg, ask, sdkstep.OutcomePassed, "") + p.sayEnded(msg, ask, sdkstep.OutcomePassed, "", result) } p.logf("the %s ask %s ended on %s after %s, in its phase %q; %s", ask.Kind, ask.ID, p.Node, now().Sub(step.Started).Round(time.Second), step.Phase, outcome) @@ -976,18 +980,35 @@ func (p *Puller) say(event string, body any, id string) { } // sayEnded says step-ended for an ask outside a pipeline, whatever its asker (design 47 G §602): a pipeline step's end is -// its run's own (AskSteps.SayStep). -func (p *Puller) sayEnded(msg *Pulled, ask sdkstep.StepAsk, outcome, said string) { +// its run's own (AskSteps.SayStep). The end of a run carries the run's answer (A-254): the last line of its program's +// standard output, inline as `output` up to sdkstep.StepEndedInlineMost, else as `output-object`, kept by KeepOutput; a +// program that exits other than 0 ends failed with its answer carried. +func (p *Puller) sayEnded(msg *Pulled, ask sdkstep.StepAsk, outcome, said string, result any) { if p.Say == nil || ask.Step != nil { return } + end := sdkstep.StepEnded{ID: ask.ID, Asker: askerOf(msg, ask), Ask: ask.ID, Node: p.Node, Outcome: outcome, At: now().UTC()} + if ans, ok := runAnswerOf(ask, result); ok { + if ans.Exit != 0 && outcome == sdkstep.OutcomePassed { + outcome, said = sdkstep.OutcomeFailed, fmt.Sprintf("the program exited %d: %s", ans.Exit, ans.Said) + } + if len(ans.Output) <= sdkstep.StepEndedInlineMost { + end.Output = string(ans.Output) + } else if object := sdkstep.OutputObjectName(ask.ID); p.KeepOutput == nil { + outcome, said = sdkstep.OutcomeFailed, fmt.Sprintf("the run's answer is %d bytes, over what the end carries, and this holder keeps no object for it", len(ans.Output)) + } else if err := p.KeepOutput(object, ans.Output); err != nil { + outcome, said = sdkstep.OutcomeFailed, fmt.Sprintf("the run's answer of %d bytes could not be kept: %v", len(ans.Output), err) + } else { + end.OutputObject = object + } + } if len(said) > 300 { said = said[:300] + "..." } if said == "" && outcome != sdkstep.OutcomePassed { said = "the ask did not pass" } - end := sdkstep.StepEnded{ID: ask.ID, Asker: askerOf(msg, ask), Ask: ask.ID, Node: p.Node, Outcome: outcome, Said: said, At: now().UTC()} + end.Outcome, end.Said = outcome, said body, err := sdkstep.EncodeStepEnded(end) if err != nil { p.logf("the end of the ask %s cannot be said on %s: %v", ask.ID, p.Node, err) @@ -995,3 +1016,19 @@ func (p *Puller) sayEnded(msg *Pulled, ask sdkstep.StepAsk, outcome, said string } p.say("step-ended", json.RawMessage(body), "step-ended."+ask.ID) } + +// runAnswerOf is the run answer a run ask's step handed back, whatever its form (the value or a pointer). +func runAnswerOf(ask sdkstep.StepAsk, result any) (sdkstep.RunAnswer, bool) { + if ask.Run == nil { + return sdkstep.RunAnswer{}, false + } + switch a := result.(type) { + case sdkstep.RunAnswer: + return a, true + case *sdkstep.RunAnswer: + if a != nil { + return *a, true + } + } + return sdkstep.RunAnswer{}, false +} diff --git a/modules/mesh-delivery-sdk/steprun/run_output_test.go b/modules/mesh-delivery-sdk/steprun/run_output_test.go new file mode 100644 index 00000000..240555e6 --- /dev/null +++ b/modules/mesh-delivery-sdk/steprun/run_output_test.go @@ -0,0 +1,83 @@ +package steprun + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" +) + +// Design 47 G A-254: a build seat's step-ended of a run carries the run's answer, the last line of its program's standard output: +// inline as `output` up to the event bound, else as `output-object`, the object the holder keeps, which the verb `output` answers. +func TestStepEndedOfARunCarriesItsAnswer(t *testing.T) { + big := `"` + strings.Repeat("x", sdkstep.StepEndedInlineMost) + `"` + for _, c := range []struct { + name string + answer sdkstep.RunAnswer + keepErr error + outcome string + inline string + objectKept bool + said string + }{ + {name: "a small answer is inline", answer: sdkstep.RunAnswer{Output: json.RawMessage(`{"verdict":true}`)}, outcome: sdkstep.OutcomePassed, inline: `{"verdict":true}`}, + {name: "an answer over the event bound is an object", answer: sdkstep.RunAnswer{Output: json.RawMessage(big)}, outcome: sdkstep.OutcomePassed, objectKept: true}, + {name: "a program that exits other than 0 failed, its answer kept", answer: sdkstep.RunAnswer{Exit: 3, Output: json.RawMessage(`"no"`), Said: "bad input"}, + outcome: sdkstep.OutcomeFailed, inline: `"no"`, said: "the program exited 3: bad input"}, + {name: "an answer the holder cannot keep is said", answer: sdkstep.RunAnswer{Output: json.RawMessage(big)}, keepErr: errors.New("store full"), + outcome: sdkstep.OutcomeFailed, said: "store full"}, + } { + t.Run(c.name, func(t *testing.T) { + src, st, sy := NewFakeAsks(), newSteps(), &sayer{} + a := testAsk(sdkstep.AskCheck) + a.Check = nil + in, _ := sdkstep.RunInputOf("seat:mesh-catalog.input", []byte(`{"v":1}`)) + commit := "0123456789abcdef0123456789abcdef01234567" + a.Run = &sdkstep.RunJob{Program: sdkstep.RunProgram{Repository: "novox/mesh-delivery", Commit: commit, Dir: "modules/x", Package: "./cmd/x"}, Input: in} + _ = src.PublishAsk(a) + var mu sync.Mutex + kept := map[string][]byte{} + p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 }, + PullWait: 10 * time.Millisecond, Heartbeat: time.Hour, Say: sy.say, + KeepOutput: func(object string, body []byte) error { + mu.Lock() + defer mu.Unlock() + if c.keepErr != nil { + return c.keepErr + } + kept[object] = body + return nil + }, + Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) { return c.answer, nil }} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go p.Pull(ctx) + waitFor(t, "step-ended", func() bool { return len(sy.of("step-ended")) == 1 }) + end := sy.of("step-ended")[0].body + if end["outcome"] != c.outcome { + t.Fatalf("outcome %v, want %s (%v)", end["outcome"], c.outcome, end) + } + if c.inline != "" && end["output"] != c.inline { + t.Fatalf("output %v, want %s inline", end["output"], c.inline) + } + if c.said != "" && !strings.Contains(end["said"].(string), c.said) { + t.Fatalf("said %v, want it to hold %q", end["said"], c.said) + } + mu.Lock() + defer mu.Unlock() + if c.objectKept { + name := sdkstep.OutputObjectName(a.ID) + if end["output-object"] != name || end["output"] != nil || string(kept[name]) != big { + t.Fatalf("end %v and kept %d bytes, want the object %s holding the whole answer and no inline output", end["output-object"], len(kept[name]), name) + } + } else if end["output-object"] != nil || len(kept) != 0 { + t.Fatalf("an object %v was named or kept (%d), want none", end["output-object"], len(kept)) + } + }) + } +} diff --git a/modules/mesh-delivery-sdk/steprun/seat_tools.go b/modules/mesh-delivery-sdk/steprun/seat_tools.go index 408245c8..1c423783 100644 --- a/modules/mesh-delivery-sdk/steprun/seat_tools.go +++ b/modules/mesh-delivery-sdk/steprun/seat_tools.go @@ -1,6 +1,10 @@ package steprun import ( + "encoding/base64" + "errors" + + sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" stdio "git.novox.be/novox/mesh-sdk/go" ) @@ -50,3 +54,29 @@ func SeatTools(seat, what string, p *Puller) []stdio.Tool { }), } } + +// OutputTool serves `.output` (design 47 G A-254, A-257): a run's answer by its ask, as the object the holder keeps it in +// (sdkstep.OutputObjectName), read with read, which answers nil for none. The verb's answer is marked chunks: the bundle answers +// the whole answer as the member `file` (base64) and the tool runner sends it in chunks and answers their count; an ask +// whose answer this node does not keep answers no chunks. +func OutputTool(seat string, read func(object string) ([]byte, error)) stdio.Tool { + return stdio.Tool{Name: seat + ".output", + Description: "A run's answer by its ask, when its step-ended named it as an output-object: the answer's bytes in chunks, of any size.", + Input: map[string]any{"type": "object", "required": []string{"ask", "call"}, "properties": map[string]any{ + "ask": map[string]any{"type": "string", "description": "the ask's id"}, + "call": map[string]any{"type": "string", "description": "the subject token of this send's chunks"}}}, + Run: func(args map[string]any) (any, error) { + ask, _ := args["ask"].(string) + if ask == "" { + return nil, errors.New("the verb output is asked for no ask") + } + body, err := read(sdkstep.OutputObjectName(ask)) + if err != nil { + return nil, err + } + if body == nil { + return map[string]any{"ask": ask, "chunks": 0}, nil + } + return map[string]any{"ask": ask, "file": base64.StdEncoding.EncodeToString(body)}, nil + }} +} diff --git a/modules/mesh-delivery-sdk/steprun/seat_tools_test.go b/modules/mesh-delivery-sdk/steprun/seat_tools_test.go index 74affb11..ddd615da 100644 --- a/modules/mesh-delivery-sdk/steprun/seat_tools_test.go +++ b/modules/mesh-delivery-sdk/steprun/seat_tools_test.go @@ -50,3 +50,27 @@ func TestSeatToolsCurrentPauseResume(t *testing.T) { } waitFor(t, "the ask taken after the resume", func() bool { return src.Waiting(sdkstep.AskDelivery) == 0 }) } + +// Design 47 G A-254: `.output` answers a run's answer by its ask as the whole file (base64), which the tool runner sends in +// chunks (A-257); an ask this node keeps no answer of answers no chunks. +func TestOutputToolAnswersTheKeptAnswer(t *testing.T) { + kept := map[string][]byte{sdkstep.OutputObjectName("a1"): []byte(`{"verdict":"x"}`)} + tool := OutputTool("node-build-go", func(object string) ([]byte, error) { return kept[object], nil }) + if tool.Name != "node-build-go.output" { + t.Fatalf("name %q", tool.Name) + } + got, err := tool.Run(map[string]any{"ask": "a1", "call": "c"}) + if err != nil { + t.Fatal(err) + } + if m := got.(map[string]any); m["ask"] != "a1" || m["file"] != "eyJ2ZXJkaWN0IjoieCJ9" { + t.Fatalf("answer %v, want the ask and its answer's bytes as the file", got) + } + got, _ = tool.Run(map[string]any{"ask": "none", "call": "c"}) + if m := got.(map[string]any); m["ask"] != "none" || m["chunks"] != 0 || m["file"] != nil { + t.Fatalf("answer %v, want no chunks and no file", got) + } + if _, err := tool.Run(map[string]any{"call": "c"}); err == nil { + t.Fatal("a call with no ask was answered") + } +}