C1c-2a: the step runner says a run's answer in step-ended, inline or as an object, and serves it
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
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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 `<seat>.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
|
||||
}}
|
||||
}
|
||||
|
||||
@@ -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: `<seat>.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")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user