diff --git a/node-tools/cmd/node-tools/approvals.go b/node-tools/cmd/node-tools/approvals.go index d048b7c..48dfe63 100644 --- a/node-tools/cmd/node-tools/approvals.go +++ b/node-tools/cmd/node-tools/approvals.go @@ -3,6 +3,7 @@ package main import ( "encoding/json" "errors" + "fmt" "log" "time" @@ -147,3 +148,17 @@ func warranted(conn *bus.Conn) func(asker, ask string, open func() (bool, error) return err } } + +// gateTools is the gate as mesh_approvals reads it, with its state's fill read from the bucket (§13, §18). +type gateTools struct { + *approval.Gate + state approvalsState +} + +func (t gateTools) Fill() (uint64, int64, error) { + info, err := t.state.conn.StateInfo(t.state.module, approval.StateName) + if err != nil || info == nil { + return 0, 0, fmt.Errorf("the state %s's fill cannot be read: %v", approval.StateName, err) + } + return info.Bytes, info.MaxBytes, nil +} diff --git a/node-tools/cmd/node-tools/main.go b/node-tools/cmd/node-tools/main.go index 5796757..8a81f04 100644 --- a/node-tools/cmd/node-tools/main.go +++ b/node-tools/cmd/node-tools/main.go @@ -118,7 +118,9 @@ func main() { node = "?" } who := node + "." + cred.Module - up, err = console.Serve(console.NewSurface(conn, source, who), listen) + surface := console.NewSurface(conn, source, who) + surface.Approvals = gateTools{Gate: gate, state: approvalsState{conn: conn, module: cred.Module}} + up, err = console.Serve(surface, listen) if err != nil { log.Fatalf("node-tools: %v", err) } diff --git a/node-tools/internal/approval/approval_test.go b/node-tools/internal/approval/approval_test.go index 9c42034..230cd1e 100644 --- a/node-tools/internal/approval/approval_test.go +++ b/node-tools/internal/approval/approval_test.go @@ -563,6 +563,35 @@ func TestTheRefusalsAndAnswers(t *testing.T) { t.Fatalf("findings %v, bounds %v", w.findings.raised, w.bounds.keys) } }}, + {"§18: the asks of the last day, one read whole, the marks applied with their sources, a record again and a cancel", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{"dir":"scratch"}`)) + list, err := w.g.List() + if err != nil || len(list) != 1 || list[0].State != "asked" || list[0].Address != "shell.wipe" || list[0].Fingerprint == "" { + t.Fatalf("%+v %v", list, err) + } + whole, err := w.g.Read(list[0].Ask) + if err != nil || string(whole.Arguments) != `{"dir":"scratch"}` || !strings.HasPrefix(whole.Digest, "sha256:") { + t.Fatalf("%+v %v", whole, err) + } + applied := w.g.Applied() + if len(applied) != 1 || applied[0].Tools["wipe"] != "entry shell src2:1" || applied[0].Verbs["node-login-shell.execute"] != "seat node-login-shell" || applied[0].Revision != "7" { + t.Fatalf("applied %+v", applied) + } + if ok, src := w.g.Describe("shell", "", "wipe"); !ok || src != "entry shell src2:1" { + t.Errorf("describe %v %q", ok, src) + } + if ok, src := w.g.Describe("", "node-login-shell", "execute"); !ok || src != "seat node-login-shell" { + t.Errorf("describe a seat verb %v %q", ok, src) + } + if err := w.g.Again(list[0].Ask); err == nil { + t.Error("an ask not run was asked its hand-act again") + } + cancelled := "" + w.g.d.Publish = func(module, subject string, body any, id string) error { cancelled = subject + " " + id; return nil } + if err := w.g.Cancel(list[0].Ask); err != nil || cancelled != "mesh.seat.operator-channel.accept.cancel.shell cancel."+list[0].Ask { + t.Fatalf("cancel %v %q", err, cancelled) + } + }}, } { t.Run(tc.name, func(t *testing.T) { tc.run(t, newWorld()) }) } diff --git a/node-tools/internal/approval/gate.go b/node-tools/internal/approval/gate.go index 419197c..787014a 100644 --- a/node-tools/internal/approval/gate.go +++ b/node-tools/internal/approval/gate.go @@ -73,6 +73,8 @@ type Gate struct { inFlight map[string]bool // failingSince is when the state approvals began to fail; zero while it is read. failingSince time.Time + // served is each module this runner serves, with the build its bundle serves, as its last load said (§18). + served map[string]string } // New is a gate over d. @@ -102,6 +104,10 @@ func (g *Gate) Unserved(module, fingerprint string, registered []string) []strin } g.mu.Lock() g.registered[module] = set + if g.served == nil { + g.served = map[string]string{} + } + g.served[module] = fingerprint g.mu.Unlock() if g.d.Catalog == nil || g.d.Catalog.Ready() != nil { return nil diff --git a/node-tools/internal/approval/tools.go b/node-tools/internal/approval/tools.go new file mode 100644 index 0000000..13b4cbf --- /dev/null +++ b/node-tools/internal/approval/tools.go @@ -0,0 +1,168 @@ +package approval + +import ( + "encoding/json" + "fmt" + "sort" + "strings" + "time" +) + +// The mesh MCP server's mesh_approvals reads and steers what the gate asked on this machine (§18). + +// Listed is one ask of the last day as mesh_approvals lists it: its state and its act's fingerprint. +type Listed struct { + Ask string `json:"ask"` + Address string `json:"address"` + Module string `json:"module"` + Asked time.Time `json:"asked"` + State string `json:"state"` + Fingerprint string `json:"fingerprint"` +} + +// Whole is one ask read whole: its arguments and the act's digest. +type Whole struct { + Listed + Digest string `json:"digest"` + Act map[string]string `json:"act"` + Arguments json.RawMessage `json:"arguments"` + Recorded bool `json:"recorded"` +} + +// stateOf is a kept ask's state as the runner knows it. +func stateOf(acted, recorded bool, ended string) string { + switch { + case acted && recorded: + return "run, recorded" + case acted: + return "run, not yet recorded" + case ended != "": + return ended + } + return "asked" +} + +// List is the calls asked on this machine in the last day, newest first (§18). +func (g *Gate) List() ([]Listed, error) { + if g.d.State == nil { + return nil, fmt.Errorf("the tool runner's membership binds no state %s", StateName) + } + keys, err := g.d.State.Keys() + if err != nil { + return nil, err + } + prefix := g.d.Node + ".ask." + out := []Listed{} + for _, key := range keys { + if !strings.HasPrefix(key, prefix) { + continue + } + w, err := g.Read(strings.TrimPrefix(key, prefix)) + if err != nil || g.d.Now().Sub(w.Asked) > 24*time.Hour { + continue + } + out = append(out, w.Listed) + } + sort.Slice(out, func(i, j int) bool { return out[i].Asked.After(out[j].Asked) }) + return out, nil +} + +// Read is one ask whole: its arguments and the act's digest (§18). +func (g *Gate) Read(id string) (Whole, error) { + if g.d.State == nil { + return Whole{}, fmt.Errorf("the tool runner's membership binds no state %s", StateName) + } + k, _, err := g.kept(id) + if err != nil { + return Whole{}, err + } + if k == nil { + return Whole{}, fmt.Errorf("no ask %s is kept on %s", id, g.d.Node) + } + module := k.Act["module"] + if module == "" { + module = moduleOf(k) + } + return Whole{Listed: Listed{Ask: id, Address: k.Act["address"], Module: module, Asked: k.Asked, + State: stateOf(k.Acted, k.Recorded, k.Ended), Fingerprint: Fingerprint(k.Ask.About)}, + Digest: k.Ask.About, Act: k.Act, Arguments: k.Arguments, Recorded: k.Recorded}, nil +} + +// Applied are the marks the gate applies on this machine, each module's marked tools and verbs with their source and the +// catalog revision read (§18). +func (g *Gate) Applied() []Marks { + g.mu.Lock() + modules := make([]string, 0, len(g.served)) + for m := range g.served { + modules = append(modules, m) + } + g.mu.Unlock() + sort.Strings(modules) + var out []Marks + if g.d.Catalog == nil { + return out + } + for _, m := range modules { + if marks := Derive(g.d.Catalog, m, g.servedFingerprint(m)); marks.Any() { + out = append(out, marks) + } + } + return out +} + +func (g *Gate) servedFingerprint(module string) string { + g.mu.Lock() + defer g.mu.Unlock() + return g.served[module] +} + +// Describe says whether a tool or a seat verb of a module is marked, and its source, for mesh_describe (§18): the served +// build's marks where this machine serves the module, else the newest entry's and its seat's. +func (g *Gate) Describe(module, seat, name string) (bool, string) { + if g.d.Catalog == nil || g.d.Catalog.Ready() != nil { + return false, "" + } + if seat != "" && module == "" { + for s, d := range g.d.Catalog.Definitions() { + if s != seat { + continue + } + for _, v := range d.Verbs { + if v.Name == name && v.NeedsApproval { + return true, "seat " + seat + } + } + } + return false, "" + } + return Derive(g.d.Catalog, module, g.servedFingerprint(module)).Marked(seat, name) +} + +// Again asks `warranted` again for one acted, unrecorded ask, by hand, in the kept-call form of §16 (§18); the caller +// records the hand act first. +func (g *Gate) Again(id string) error { + w, err := g.Read(id) + if err != nil { + return err + } + if !strings.HasPrefix(w.State, "run, not yet") { + return fmt.Errorf("the ask %s is %s: only a run not yet recorded is asked again", id, w.State) + } + g.recordRun(w.Module, id) + if again, _ := g.Read(id); !again.Recorded { + return fmt.Errorf("mesh-hand-acts.warranted has not answered for ask %s yet", id) + } + return nil +} + +// Cancel takes an open ask back through operator-channel's client (§18); the caller records the hand act first. +func (g *Gate) Cancel(id string) error { + w, err := g.Read(id) + if err != nil { + return err + } + if w.State != "asked" { + return fmt.Errorf("the ask %s is %s, not open", id, w.State) + } + return g.client(w.Module).Cancel(id) +} diff --git a/node-tools/internal/console/address.go b/node-tools/internal/console/address.go index b8860a7..84de6fd 100644 --- a/node-tools/internal/console/address.go +++ b/node-tools/internal/console/address.go @@ -24,6 +24,7 @@ import ( "context" "encoding/json" + "errors" "fmt" "sort" "strings" @@ -103,12 +104,13 @@ func discovery() []map[string]any { }), "description": "Ask the tool runner's four first views again, now, and say what each answers. A hand act, recorded in " + "the runner's log with its reason; for when a picture is doubted, or after an owner's state was made again."}, waitTool(), + approvalsTool(), } } func isDiscovery(name string) bool { switch name { - case verbOverview, verbMachine, verbSearch, verbDescribe, verbCall, verbRuntimes, verbFirstViews, verbFirstAgain, verbWait: + case verbOverview, verbMachine, verbSearch, verbDescribe, verbCall, verbRuntimes, verbFirstViews, verbFirstAgain, verbWait, verbApprovals: return true } return false @@ -949,6 +951,9 @@ func (s *Surface) discover(ctx context.Context, name string, args map[string]any if name == verbWait { return s.waitAnswer(ctx, args) } + if name == verbApprovals { + return s.approvalsAnswer(args) + } x, err := s.index() if err != nil { return failure("the mesh's discovery failed: " + err.Error()) @@ -1109,6 +1114,10 @@ func (s *Surface) discover(ctx context.Context, name string, args map[string]any if t.Stale != nil { described["found"] = t.Stale.say() } + if ok, source := s.markedTool(t); ok { + described["needs-approval"] = map[string]any{"source": source, + "says": "the first call asks the operator and runs nothing; the same call again within 30 minutes of the Approve runs once"} + } return answerText(described) case verbCall: @@ -1129,6 +1138,11 @@ func (s *Surface) discover(ctx context.Context, name string, args map[string]any } if err != nil { why := whyItFailed(t.Address, err) + if ok, _ := s.markedTool(t); ok && errors.Is(err, bus.ErrNoResponders) { + // A marked subject withheld from a machine whose tool runner does not honour the mark (§6). + why += "; it needs the operator's approval, so it may be withheld where the machine's tool runner does not state " + + "approval-mark/1: mesh-grants.grants_user_list names it under withheld" + } if t.Stale != nil { why += " (" + t.Stale.say() + ")" } diff --git a/node-tools/internal/console/approvals.go b/node-tools/internal/console/approvals.go new file mode 100644 index 0000000..f189e8b --- /dev/null +++ b/node-tools/internal/console/approvals.go @@ -0,0 +1,99 @@ +package console + +import ( + "log" + "strings" + + "github.com/novox/mesh-tools/node-tools/internal/approval" +) + +// verbApprovals is the mesh MCP server's tool that reads and steers what the gate asked on this machine (design 46 +// 7-approval-mark §18). +const verbApprovals = "mesh_approvals" + +// Approvals is the gate of this machine's tool runner, as mesh_approvals and mesh_describe read it. +type Approvals interface { + List() ([]approval.Listed, error) + Read(id string) (approval.Whole, error) + Applied() []approval.Marks + Describe(module, seat, name string) (bool, string) + Again(id string) error + Cancel(id string) error + // Fill is the state approvals' bytes against its cap, every machine's keys together. + Fill() (bytes uint64, max int64, err error) +} + +func approvalsTool() map[string]any { + str := func(desc string) map[string]any { return map[string]any{"type": "string", "description": desc} } + return map[string]any{"name": verbApprovals, "inputSchema": map[string]any{"type": "object", "properties": map[string]any{ + "ask": str("an ask's id: its whole arguments and the act's digest"), + "record": str("an ask's id: ask mesh-hand-acts.warranted again for its run, recorded in the hand-act log first"), + "cancel": str("an ask's id: take that open ask back, recorded in the hand-act log first"), + "why": str("why, for record and cancel, kept in the hand-act log"), + }}, "description": "The calls that needed the operator's approval on this machine: with no argument, those asked in " + + "the last day, newest first, with their state and fingerprint, the marks the tool runner applies (each module's " + + "marked tools and verbs with their source and the catalog revision read) and its state's fill; `ask` reads one " + + "whole, `record` asks its hand-act again, `cancel` takes an open ask back."} +} + +// approvalsAnswer is mesh_approvals. +func (s *Surface) approvalsAnswer(args map[string]any) map[string]any { + if s.Approvals == nil { + return failure("this tool runner has no approval gate") + } + str := func(k string) string { v, _ := args[k].(string); return strings.TrimSpace(v) } + why := str("why") + if why == "" { + why = "asked by hand with no reason given" + } + act := func(id, what, cause string, do func(string) error) map[string]any { + // A hand act, recorded through the seat mesh-hand-acts before it is done (§18); one that cannot be recorded is refused. + log.Printf("[mesh-tools] hand act: %s %s ask %s: %s", s.who, what, id, why) + if err := s.recordAct(map[string]any{"what": what + " ask " + id, "why": why, "cause": cause, "verb": verbApprovals}); err != nil { + return failure("the hand act was not recorded through mesh-hand-acts, so nothing was done: " + err.Error()) + } + if err := do(id); err != nil { + return failure(err.Error()) + } + return answerText(map[string]any{"done": what + " ask " + id, "recorded": "mesh-hand-acts.hand-act"}) + } + switch { + case str("record") != "": + return act(str("record"), "asked the hand-act of", "approval-record", s.Approvals.Again) + case str("cancel") != "": + return act(str("cancel"), "cancelled", "approval-cancel", s.Approvals.Cancel) + case str("ask") != "": + w, err := s.Approvals.Read(str("ask")) + if err != nil { + return failure(err.Error()) + } + return answerText(w) + } + asks, err := s.Approvals.List() + out := map[string]any{"asks": asks, "marks": s.Approvals.Applied()} + if err != nil { + out["asks"], out["asks_why"] = []approval.Listed{}, "the state approvals cannot be read: "+err.Error() + } + if b, max, err := s.Approvals.Fill(); err != nil { + out["fill_why"] = err.Error() + } else { + out["fill"] = map[string]any{"bytes": b, "max_bytes": max, "of": "every machine's keys together"} + } + return answerText(out) +} + +// marked is what mesh_describe and a call that reached no one say of a marked tool or verb (§6, §18). +func (s *Surface) marked(module, seat, name string) (bool, string) { + if s.Approvals == nil { + return false, "" + } + return s.Approvals.Describe(module, seat, name) +} + +// markedTool is marked for a resolved address. +func (s *Surface) markedTool(t target) (bool, string) { + if t.Seat { + return s.marked("", t.Tool.Module, t.Tool.Name) + } + return s.marked(t.Tool.Module, "", t.Tool.Name) +} diff --git a/node-tools/internal/console/mcp.go b/node-tools/internal/console/mcp.go index 1874356..57e4e8b 100644 --- a/node-tools/internal/console/mcp.go +++ b/node-tools/internal/console/mcp.go @@ -63,6 +63,8 @@ type Surface struct { Flat bool // Record, when set, records a hand act in place of the seat mesh-hand-acts' verb (a test's). Record func(args map[string]any) error + // Approvals is this machine's approval gate, read by mesh_approvals and mesh_describe; nil when the runner has none. + Approvals Approvals } // MethodSeats is the console's JSON-RPC method beside MCP's that answers every seat's verbs and holders, in the shape diff --git a/node-tools/internal/console/wait_test.go b/node-tools/internal/console/wait_test.go index 862517f..3cfc486 100644 --- a/node-tools/internal/console/wait_test.go +++ b/node-tools/internal/console/wait_test.go @@ -3,6 +3,7 @@ package console import ( "context" "encoding/json" + "errors" "net/http" "os" "reflect" @@ -11,6 +12,7 @@ import ( "testing" "time" + "github.com/novox/mesh-tools/node-tools/internal/approval" "github.com/novox/mesh-tools/node-tools/internal/bus" mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) @@ -194,7 +196,7 @@ func TestMeshWaitIsAnnouncedAmongSeven(t *testing.T) { } } sort.Strings(names) - want := []string{"mesh_call", "mesh_describe", "mesh_first_view_again", "mesh_first_views", "mesh_machine", "mesh_overview", "mesh_runtimes", "mesh_search", "mesh_wait"} + want := []string{"mesh_approvals", "mesh_call", "mesh_describe", "mesh_first_view_again", "mesh_first_views", "mesh_machine", "mesh_overview", "mesh_runtimes", "mesh_search", "mesh_wait"} if !reflect.DeepEqual(names, want) { t.Errorf("announced %v, want %v", names, want) } @@ -207,8 +209,43 @@ func TestMeshWaitIsAnnouncedAmongSeven(t *testing.T) { t.Errorf("mesh_wait's schema has no %s: %s", word, schema) } } + // Design 46 7-approval-mark §18: mesh_approvals lists, and records a hand act before it cancels; refused without a record. + gate := &fakeApprovals{} + var recorded []string + s := &Surface{who: "n1.node-tools", Approvals: gate, Record: func(a map[string]any) error { + if a["why"] == "unrecordable" { + return errors.New("mesh-hand-acts does not answer") + } + recorded = append(recorded, a["what"].(string)) + return nil + }} + if out := s.discover(context.Background(), verbApprovals, map[string]any{}); out["isError"] == true || !strings.Contains(fmtAnswer(out), "every machine") { + t.Errorf("the list %v", out) + } + if out := s.discover(context.Background(), verbApprovals, map[string]any{"cancel": "a1", "why": "no longer needed"}); out["isError"] == true || + gate.cancelled != "a1" || len(recorded) != 1 || recorded[0] != "cancelled ask a1" { + t.Errorf("a cancel %v, recorded %v", out, recorded) + } + if out := s.discover(context.Background(), verbApprovals, map[string]any{"cancel": "a2", "why": "unrecordable"}); out["isError"] != true || gate.cancelled != "a1" { + t.Errorf("a cancel without a record %v", out) + } } +func fmtAnswer(out map[string]any) string { raw, _ := json.Marshal(out); return string(raw) } + +// fakeApprovals is a gate as mesh_approvals reads it. +type fakeApprovals struct{ cancelled string } + +func (f *fakeApprovals) List() ([]approval.Listed, error) { + return []approval.Listed{{Ask: "a1", State: "asked"}}, nil +} +func (f *fakeApprovals) Read(id string) (approval.Whole, error) { return approval.Whole{}, nil } +func (f *fakeApprovals) Applied() []approval.Marks { return nil } +func (f *fakeApprovals) Describe(_, _, _ string) (bool, string) { return false, "" } +func (f *fakeApprovals) Again(string) error { return nil } +func (f *fakeApprovals) Cancel(id string) error { f.cancelled = id; return nil } +func (f *fakeApprovals) Fill() (uint64, int64, error) { return 10, 64 << 20, nil } + // A client that hangs up or cancels ends its wait at once: the subscription is closed and nothing is left // waiting for the timeout (the review of issue 745). Over the console's own HTTP, as an agent host calls it. func TestMeshWaitEndsWhenItsClientHangsUp(t *testing.T) {