From 5b2b70cd6c73eb4cfee195e6cf03e4d6ad0486b8 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 21:56:42 +0100 Subject: [PATCH] =?UTF-8?q?Keep=20the=20hand-act=20log=20as=20mesh-hand-ac?= =?UTF-8?q?ts'=20module=20log,=20the=20stream=20ADR=200565=20=C2=A710=20na?= =?UTF-8?q?mes,=20so=20every=20act=20is=20held=20in=20its=20order=20(issue?= =?UTF-8?q?=201953)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The warrants stay a bucket and no longer age, so a warrant is recorded once; one whose act was lost is written again from the router's record. hand-acts says the log's fill against its cap. --- modules/mesh-hand-acts/README.md | 9 +- .../mesh-hand-acts/cmd/mesh-hand-acts/acts.go | 116 +++++++++++------- .../cmd/mesh-hand-acts/acts_test.go | 109 +++++++++++++--- .../mesh-hand-acts/cmd/mesh-hand-acts/main.go | 47 +++++-- .../cmd/mesh-hand-acts/manifest_test.go | 31 ++--- .../mesh-hand-acts/whole_integration_test.go | 87 +++++++++++-- modules/mesh-hand-acts/module.json | 12 +- 7 files changed, 309 insertions(+), 102 deletions(-) diff --git a/modules/mesh-hand-acts/README.md b/modules/mesh-hand-acts/README.md index c7085dfd..d01b8b49 100644 --- a/modules/mesh-hand-acts/README.md +++ b/modules/mesh-hand-acts/README.md @@ -6,9 +6,12 @@ alone writes the log. ## What it keeps -- `mesh-hand-acts_acts`: every act, one key `act--` (`warrant--` for a warrant) in the layout of the - controller's log, kept 400 days by the bucket's age. Each act is said as `act-recorded`. -- `mesh-hand-acts_warrants`: one key `.` per warrant recorded, made once by create, naming its act. +- the log `acts` (the stream `LOG_mesh-hand-acts_acts`, a module log of ADR 0297, as ADR 0565 §10 has it a stream): + every act appended in its order under its key `act--` (`warrant--` for a warrant), in the layout + of the controller's log. A log ages nothing and refuses a new act when full, so every act is held 400 days and longer; + `hand-acts` says how full it is against its cap. Each act is said as `act-recorded`. +- `mesh-hand-acts_warrants`: one key `.` per warrant, made once by create before its act is appended, so of + several instances one records it; a warrant whose act was lost is written again from the router's record. ## Its verbs diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go index 26969cb1..01e09440 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go @@ -2,8 +2,10 @@ package main // The hand-act log (novox/hq ADR 0565 §10, §11): every act done by hand, with who, why and its cause, so a repair a // person keeps making is found (S15) and a person's decision is read back. mesh-hand-acts holds the seat -// mesh-hand-acts, once for the mesh, and is the one writer of its state `acts` (each act a key that the bucket keeps -// 400 days) and `warrants` (one key `.` made once by create). Each act is said as `act-recorded`. +// mesh-hand-acts, once for the mesh, and is the one writer of its log `acts` — a stream, as ADR 0565 §10 has it: +// the mesh's module log (ADR 0297), append-only, every act kept in its order under its key, refusing a new act when +// full rather than dropping an old one, so every act is held 400 days and longer — and of its bucket `warrants` (one +// key `.` made once by create). Each act is said as `act-recorded`. // // **Who did it is the runtime's word, never the caller's**: `by` is the bus user the runtime proved. A warrant is the // router's word, never the caller's: `warranted` reads the router's own record of the asker's ask and records who @@ -33,7 +35,7 @@ const ( // The states it keeps and reads, and the event it says. const ( - ActsState = "acts" + ActsLog = "acts" WarrantsState = "warrants" // ControllerActs is the controller's log, read until its lease ends. ControllerActs = "mesh-controller.hand-acts" @@ -80,13 +82,25 @@ type Act struct { Log string `json:"log,omitempty"` } -// Store is the bus's buckets as this module reaches them, through its runtime. Create writes a key only when it has -// none, and says ErrExists otherwise. +// Store is the bus's buckets and this module's log as it reaches them, through its runtime. Create writes a key only +// when it has none, and says ErrExists otherwise. type Store interface { Get(state, key string) ([]byte, bool, error) - Put(state, key string, value any) error Create(state, key string, value any) error Keys(state string) ([]string, error) + // Append adds an entry to the end of a log; Last is a key's newest entry; Entries every entry in order; Fill how + // many entries and bytes the log holds against its cap. + Append(log, key string, value any) error + Last(log, key string) ([]byte, bool, error) + Entries(log string) ([][]byte, error) + Fill(log string) (Fill, error) +} + +// Fill is how full a log is: its entries, its bytes and its cap in bytes. +type Fill struct { + Entries uint64 `json:"entries"` + Bytes uint64 `json:"bytes"` + MaxBytes uint64 `json:"max-bytes"` } // ErrExists is a create of a key another writer made first. @@ -153,7 +167,7 @@ func (l *Log) HandAct(by, verb, what, why, cause, condition string) (Recorded, e // record writes one act and says it. An act not said is still kept, and the log says so. func (l *Log) record(act Act) error { - if err := l.Store.Create(ActsState, act.ID, act); err != nil { + if err := l.Store.Append(ActsLog, act.ID, act); err != nil { return fmt.Errorf("the act could not be recorded: %w", err) } if l.Emit != nil { @@ -192,10 +206,14 @@ func (l *Log) Warranted(caller, asker, ask string) (Recorded, error) { return Recorded{}, fmt.Errorf("%q is not an ask's id", ask) } key := asker + "." + ask - if raw, found, err := l.Store.Get(WarrantsState, key); err != nil { + claimed, found, err := l.Store.Get(WarrantsState, key) + if err != nil { return Recorded{}, err - } else if found { - return l.already(key, raw) + } + if found { + if r, ok, err := l.already(key, claimed); err != nil || ok { + return r, err + } } raw, found, err := l.Store.Get(RouterRecords, key) if err != nil { @@ -229,38 +247,34 @@ func (l *Log) Warranted(caller, asker, ask string) (Recorded, error) { Cause: CauseOperatorAnswer, Via: via(*w), Ask: ask, Proofs: w.Proofs, RequestedBy: asker + ", recorded at the word of " + caller, Outcome: "chosen; what " + asker + " did with it is in its own record"} - if err := l.Store.Create(WarrantsState, key, Warrant{Act: act.ID, At: at, By: caller}); errors.Is(err, ErrExists) { - raw, _, gerr := l.Store.Get(WarrantsState, key) - if gerr != nil { - return Recorded{}, gerr + if claimed == nil { + // The claim first: of any number of instances asking at once, the one whose create lands records the act. + if err := l.Store.Create(WarrantsState, key, Warrant{Act: act.ID, At: at, By: caller}); errors.Is(err, ErrExists) { + return Recorded{Said: "another instance records " + act.ID + " now"}, nil + } else if err != nil { + return Recorded{}, fmt.Errorf("the warrant could not be recorded: %w", err) } - return l.already(key, raw) - } else if err != nil { - return Recorded{}, fmt.Errorf("the warrant could not be recorded: %w", err) } - if err := l.record(act); err != nil && !errors.Is(err, ErrExists) { + if err := l.record(act); err != nil { return Recorded{}, err } return Recorded{Act: act, Said: fmt.Sprintf("recorded as %s: %s, through %s", act.ID, act.Why, act.Via)}, nil } -// already answers a warrant recorded before; an act lost between the warrant and its act is written now. -func (l *Log) already(key string, raw []byte) (Recorded, error) { +// already answers a warrant recorded before, when its act is in the log; false when the act was lost between the +// claim and the log, so it is read from the router again and written. +func (l *Log) already(key string, raw []byte) (Recorded, bool, error) { var w Warrant if err := json.Unmarshal(raw, &w); err != nil { - return Recorded{}, fmt.Errorf("the warrant %s is not readable: %w", key, err) + return Recorded{}, false, fmt.Errorf("the warrant %s is not readable: %w", key, err) } - actRaw, found, err := l.Store.Get(ActsState, w.Act) - if err != nil { - return Recorded{}, err - } - if !found { - return Recorded{}, fmt.Errorf("the warrant %s names the act %s, which is not kept: asked again, it is read from "+ - "the router again", key, w.Act) + actRaw, found, err := l.Store.Last(ActsLog, w.Act) + if err != nil || !found { + return Recorded{}, false, err } var act Act _ = json.Unmarshal(actRaw, &act) - return Recorded{Act: act, Said: "already recorded as " + act.ID + ": " + act.Why}, nil + return Recorded{Act: act, Said: "already recorded as " + act.ID + ": " + act.Why}, true, nil } func via(w asks.Warrant) string { @@ -289,6 +303,8 @@ type Listing struct { // Unread says which log could not be read, so what is missing is said and never read as nothing. Unread map[string]string `json:"unread,omitempty"` Kept map[string]int `json:"kept"` + // Fill is how full this seat's log is against its cap: full, it refuses the next act. + Fill *Fill `json:"fill,omitempty"` } // HandActs reads both logs. @@ -297,7 +313,7 @@ func (l *Log) HandActs(days int, cause, who string) Listing { since := now.Add(-time.Duration(days) * 24 * time.Hour) out := Listing{Acts: []Act{}, Repeated: map[string]int{}, Days: days, Kept: map[string]int{}} var all []Act - for _, log := range []string{ActsState, ControllerActs} { + for _, log := range []string{ActsLog, ControllerActs} { acts, kept, err := l.read(log, now.Add(-max(time.Duration(days)*24*time.Hour, RepeatedWithin))) if err != nil { if out.Unread == nil { @@ -324,17 +340,38 @@ func (l *Log) HandActs(days int, cause, who string) Listing { } } sort.SliceStable(out.Acts, func(i, j int) bool { return out.Acts[i].At.After(out.Acts[j].At) }) + if f, err := l.Store.Fill(ActsLog); err == nil { + out.Fill = &f + } return out } -// read is every act of one log since a moment, and how many the log keeps. A key names its time, so only the keys -// within are read whole. +// read is every act of one log since a moment, and how many it keeps: this seat's log entry by entry, the +// controller's bucket by its keys, each of which names its time, so only those within are read whole. func (l *Log) read(log string, since time.Time) ([]Act, int, error) { + var out []Act + keep := func(raw []byte, from string) { + var a Act + if json.Unmarshal(raw, &a) != nil || a.At.Before(since) { + return + } + a.Log = from + out = append(out, a) + } + if log == ActsLog { + entries, err := l.Store.Entries(ActsLog) + if err != nil { + return nil, 0, err + } + for _, raw := range entries { + keep(raw, Seat) + } + return out, len(entries), nil + } keys, err := l.Store.Keys(log) if err != nil { return nil, 0, err } - var out []Act for _, k := range keys { if at, ok := keyTime(k); ok && at.Before(since) { continue @@ -343,18 +380,9 @@ func (l *Log) read(log string, since time.Time) ([]Act, int, error) { if err != nil { return nil, len(keys), err } - if !found { - continue + if found { + keep(raw, log) } - var a Act - if json.Unmarshal(raw, &a) != nil || a.At.Before(since) { - continue - } - a.Log = log - if log == ActsState { - a.Log = Seat - } - out = append(out, a) } return out, len(keys), nil } diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go index 9c0ea2d4..b2762666 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go @@ -10,14 +10,75 @@ import ( "time" ) +type entry struct { + key string + raw []byte +} + type memStore struct { mu sync.Mutex kv map[string]map[string][]byte + logs map[string][]entry keysErr map[string]error } func newStore() *memStore { - return &memStore{kv: map[string]map[string][]byte{}, keysErr: map[string]error{}} + return &memStore{kv: map[string]map[string][]byte{}, logs: map[string][]entry{}, keysErr: map[string]error{}} +} + +func (s *memStore) Append(log, key string, value any) error { + raw, err := json.Marshal(value) + if err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + if err := s.keysErr[log]; err != nil { + return err + } + s.logs[log] = append(s.logs[log], entry{key, raw}) + return nil +} + +func (s *memStore) Last(log, key string) ([]byte, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + for i := len(s.logs[log]) - 1; i >= 0; i-- { + if s.logs[log][i].key == key { + return s.logs[log][i].raw, true, nil + } + } + return nil, false, nil +} + +func (s *memStore) Entries(log string) ([][]byte, error) { + s.mu.Lock() + defer s.mu.Unlock() + if err := s.keysErr[log]; err != nil { + return nil, err + } + var out [][]byte + for _, e := range s.logs[log] { + out = append(out, e.raw) + } + return out, nil +} + +func (s *memStore) Fill(log string) (Fill, error) { + s.mu.Lock() + defer s.mu.Unlock() + return Fill{Entries: uint64(len(s.logs[log])), MaxBytes: 1 << 30}, nil +} + +// logKeys are a log's keys in order. +func (s *memStore) logKeys(log string) []string { + s.mu.Lock() + defer s.mu.Unlock() + var out []string + for _, e := range s.logs[log] { + out = append(out, e.key) + } + return out } func (s *memStore) Get(state, key string) ([]byte, bool, error) { @@ -96,7 +157,7 @@ func TestHandActRecordsWhatTheProvenCallerDid(t *testing.T) { for _, tc := range cases { l, store, said, _ := newLog() r, err := l.HandAct(tc.by, tc.verb, tc.what, tc.why, tc.cause, "bus.x.missing") - keys, _ := store.Keys(ActsState) + keys := store.logKeys(ActsLog) if tc.err != "" { if err == nil || !strings.Contains(err.Error(), tc.err) || len(keys) != 0 || len(*said) != 0 { t.Errorf("%s: %v, kept %v", tc.name, err, keys) @@ -107,7 +168,7 @@ func TestHandActRecordsWhatTheProvenCallerDid(t *testing.T) { t.Fatalf("%s: %v, kept %v, said %d", tc.name, err, keys, len(*said)) } var kept Act - raw, _, _ := store.Get(ActsState, keys[0]) + raw, _, _ := store.Last(ActsLog, keys[0]) _ = json.Unmarshal(raw, &kept) if kept.By != tc.by || kept.Verb != tc.wantVerb || kept.Cause != tc.wantCause || kept.Condition != "bus.x.missing" || kept.Args[0] != tc.what || kept.Log != "" { @@ -124,25 +185,31 @@ func TestWarrantedRecordsTheRoutersWordOnce(t *testing.T) { "at":"2026-10-29T21:00:00Z","ask-digest":"sha256:x"}}` cases := []struct { name, record, asker, ask, err string + // claimed is a warrant claimed by an instance that stopped before its act reached the log. + claimed bool }{ - {"chosen", chosen, "claude-code", "a1", ""}, - {"still open", `{"state":"open"}`, "claude-code", "a1", "still open"}, - {"no record", "", "claude-code", "a1", "no closed record"}, + {"chosen", chosen, "claude-code", "a1", "", false}, + {"claimed, its act lost: written now", chosen, "claude-code", "a1", "", true}, + {"still open", `{"state":"open"}`, "claude-code", "a1", "still open", false}, + {"no record", "", "claude-code", "a1", "no closed record", false}, {"expired", `{"state":"closed","warrant":{"ask":"a1","asker":"claude-code","outcome":"expired","at":"2026-10-29T21:00:00Z"}}`, - "claude-code", "a1", "no person chose"}, + "claude-code", "a1", "no person chose", false}, {"another asker's", strings.Replace(chosen, `"asker":"claude-code"`, `"asker":"mesh-config"`, 1), "claude-code", "a1", - "record is for mesh-config"}, - {"no digest", strings.Replace(chosen, `,"ask-digest":"sha256:x"`, "", 1), "claude-code", "a1", "no ask digest"}, - {"not a module", chosen, "Claude Code", "a1", "not a module"}, + "record is for mesh-config", false}, + {"no digest", strings.Replace(chosen, `,"ask-digest":"sha256:x"`, "", 1), "claude-code", "a1", "no ask digest", false}, + {"not a module", chosen, "Claude Code", "a1", "not a module", false}, } for _, tc := range cases { l, store, said, _ := newLog() if tc.record != "" { _ = store.Put(RouterRecords, "claude-code.a1", json.RawMessage(tc.record)) } + if tc.claimed { + _ = store.Create(WarrantsState, "claude-code.a1", Warrant{Act: "warrant-claude-code-a1"}) + } r, err := l.Warranted("ace.claude-code", tc.asker, tc.ask) if tc.err != "" { - if keys, _ := store.Keys(ActsState); err == nil || !strings.Contains(err.Error(), tc.err) || len(keys) != 0 { + if keys := store.logKeys(ActsLog); err == nil || !strings.Contains(err.Error(), tc.err) || len(keys) != 0 { t.Errorf("%s: %v, kept %v", tc.name, err, keys) } continue @@ -154,7 +221,7 @@ func TestWarrantedRecordsTheRoutersWordOnce(t *testing.T) { } // Asked again, by another instance: nothing more is recorded. again, err := l.Warranted("g14.claude-code", "claude-code", "a1") - keys, _ := store.Keys(ActsState) + keys := store.logKeys(ActsLog) if err != nil || !strings.HasPrefix(again.Said, "already recorded") || len(keys) != 1 || len(*said) != 1 { t.Errorf("%s: a second warranted: %q, %v, kept %v, said %d", tc.name, again.Said, err, keys, len(*said)) } @@ -167,14 +234,19 @@ func TestHandActsReadsBothLogs(t *testing.T) { l, store, _, now := newLog() put := func(log, verb, cause, by string, ago time.Duration) { at := now.Add(-ago) - _ = store.Put(log, actKey(at), Act{At: at, Verb: verb, Cause: cause, By: by, Why: "x"}) + a := Act{At: at, Verb: verb, Cause: cause, By: by, Why: "x"} + if log == ActsLog { + _ = store.Append(log, actKey(at), a) + return + } + _ = store.Put(log, actKey(at), a) } - put(ActsState, VerbRecord, "relogin", "person.jochen", time.Hour) + put(ActsLog, VerbRecord, "relogin", "person.jochen", time.Hour) put(ControllerActs, VerbRecord, "relogin", "shanks.claude-code", 3*24*time.Hour) put(ControllerActs, VerbDrill, CauseDrill, "person.jochen", time.Hour) - put(ActsState, VerbDrill, CauseDrill, "person.jochen", 2*time.Hour) - put(ActsState, VerbWarrant, CauseOperatorAnswer, "the operator", time.Hour) - put(ActsState, VerbWarrant, CauseOperatorAnswer, "the operator", 2*time.Hour) + put(ActsLog, VerbDrill, CauseDrill, "person.jochen", 2*time.Hour) + put(ActsLog, VerbWarrant, CauseOperatorAnswer, "the operator", time.Hour) + put(ActsLog, VerbWarrant, CauseOperatorAnswer, "the operator", 2*time.Hour) put(ControllerActs, VerbRecord, "config", "ace.mesh-config", 30*24*time.Hour) cases := []struct { name string @@ -199,7 +271,8 @@ func TestHandActsReadsBothLogs(t *testing.T) { } } } - if got := l.HandActs(14, "", ""); got.Acts[0].Log != Seat || got.Kept[ControllerActs] != 3 { + if got := l.HandActs(14, "", ""); got.Acts[0].Log != Seat || got.Kept[ControllerActs] != 3 || got.Fill == nil || + got.Fill.Entries != 4 { t.Errorf("the log of an act or the kept counts are not said: %+v", got.Kept) } store.keysErr[ControllerActs] = errors.New("the bus is away") diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go index 497875ea..0023c53c 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go @@ -1,6 +1,6 @@ // mesh-hand-acts: the holder of the seat mesh-hand-acts (novox/hq ADR 0565 §10). A Go bundle served by a runtime of -// its own, on its own bus account (`runs-as`), so that its account alone writes the hand-act log (`acts`, each act kept -// 400 days) and `warrants`. It reads the router's record of each ask for a warrant and the controller's log for +// its own, on its own bus account (`runs-as`), so that its account alone writes the hand-act log (`acts`, its module +// log: a stream that keeps every act in order) and `warrants`. It reads the router's record of each ask for a warrant and the controller's log for // `hand-acts`. stdout is the MCP channel; what this module says, it says on stderr. package main @@ -26,11 +26,6 @@ func (runtimeStore) Get(state, key string) ([]byte, bool, error) { return e.Value, true, nil } -func (runtimeStore) Put(state, key string, value any) error { - _, err := stdio.State(state).Put(key, value) - return err -} - func (runtimeStore) Create(state, key string, value any) error { _, err := stdio.State(state).Create(key, value) if errors.Is(err, stdio.ErrChanged) { @@ -41,6 +36,44 @@ func (runtimeStore) Create(state, key string, value any) error { func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } +func (runtimeStore) Append(log, key string, value any) error { + _, err := stdio.Log(log).Append(key, value) + return err +} + +func (runtimeStore) Last(log, key string) ([]byte, bool, error) { + e, err := stdio.Log(log).Last(key) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil +} + +func (runtimeStore) Entries(log string) ([][]byte, error) { + var out [][]byte + for from := uint64(0); ; { + page, err := stdio.Log(log).Read(from, "", 1000) + if err != nil { + return nil, err + } + for _, e := range page.Entries { + out = append(out, e.Value) + } + if page.Next == 0 { + return out, nil + } + from = page.Next + } +} + +func (runtimeStore) Fill(log string) (Fill, error) { + i, err := stdio.Log(log).Info() + if err != nil { + return Fill{}, err + } + return Fill{Entries: i.Entries, Bytes: i.Bytes, MaxBytes: i.MaxBytes}, nil +} + func main() { l := &Log{Store: runtimeStore{}, Now: time.Now, Logf: logf, Emit: func(event string, body any) error { return stdio.Emit(event, body) }} diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go index 79bfe248..90669933 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go @@ -9,8 +9,9 @@ import ( ) // The manifest says what the code does: it claims the seat mesh-hand-acts once for the mesh with its verbs (ADR 0565 -// §1, §10), runs on its own account, says act-recorded, reads the controller's log and the router's records, and keeps -// acts and warrants 400 days. Every state it keeps and reads has a tool that reads it (ADR 0565 §15). +// §1, §10), runs on its own account, says act-recorded, reads the controller's log and the router's records, keeps its +// acts as its log (a stream, §10) and warrants as a bucket that never ages, so a warrant is recorded once. Every state +// and log it keeps and reads has a tool that reads it (ADR 0565 §15). func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { raw, err := os.ReadFile(filepath.Join("..", "..", "module.json")) if err != nil { @@ -24,12 +25,10 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { Scope string `json:"scope"` Serves []string `json:"serves"` } `json:"claims"` - Emits []string `json:"emits"` - Reads []string `json:"reads"` - State []struct { - Name string `json:"name"` - TTL int `json:"ttl-seconds"` - } `json:"state"` + Emits []string `json:"emits"` + Reads []string `json:"reads"` + Logs []string `json:"logs"` + State []json.RawMessage `json:"state"` } if err := json.Unmarshal(raw, &m); err != nil { t.Fatal(err) @@ -43,18 +42,22 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { if !slices.Equal(m.Emits, []string{RecordedEvent}) || !slices.Equal(m.Reads, []string{ControllerActs, RouterRecords}) { t.Errorf("emits %v, reads %v", m.Emits, m.Reads) } - readBy := map[string]string{ActsState: HandActsTool, WarrantsState: HandActsTool, ControllerActs: HandActsTool, + readBy := map[string]string{ActsLog: HandActsTool, WarrantsState: HandActsTool, ControllerActs: HandActsTool, RouterRecords: WarrantedTool} tools := map[string]bool{} for _, tool := range Tools(&Log{}) { tools[tool.Name] = true } - for _, s := range m.State { - if s.TTL != 400*24*60*60 { - t.Errorf("%s is kept %d s, want 400 days", s.Name, s.TTL) + if !slices.Equal(m.Logs, []string{ActsLog}) { + t.Errorf("logs %v, want the acts as a log", m.Logs) + } + for _, raw := range m.State { + var name string + if json.Unmarshal(raw, &name) != nil || name != WarrantsState { + t.Errorf("the state %s is not the warrants, or ages", raw) } - if !tools[readBy[s.Name]] { - t.Errorf("the state %s has no tool that reads it", s.Name) + if !tools[readBy[name]] { + t.Errorf("the state %s has no tool that reads it", name) } } for _, r := range m.Reads { diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go index 62c4600f..3de340f2 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go @@ -4,7 +4,7 @@ package main // mesh-hand-acts whole against a real bus (novox/hq ADR 0565 §10; issue 1953 L5-C2): the built binary, played over its // real stdin and stdout as the node's tool runtime plays it (tools/call with the caller in _meta; mesh/state.* answered -// from a real nats-server's key-value buckets, mesh/publish onto its subjects). A recorder's hand act is kept in the +// from a real nats-server's key-value buckets, mesh/log.* from the log's real stream, mesh/publish onto its subjects). A recorder's hand act is kept in the // log on the bus under the proven caller and heard as act-recorded; a warrant the router recorded is recorded once. import ( @@ -185,6 +185,8 @@ func (p *player) handle(m wire) { } case "mesh/subscribe": p.answer(m.ID, map[string]any{}, nil) + case "mesh/log.append", "mesh/log.read", "mesh/log.last", "mesh/log.info": + p.log(m) case "mesh/state.keys": kv, err := p.kv(a.State) if err != nil { @@ -214,6 +216,69 @@ func (p *player) handle(m wire) { } } +// log answers the runtime's log verbs from the log's real stream, LOG__ on mesh.log.... +func (p *player) log(m wire) { + var a struct { + Log string `json:"log"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + From uint64 `json:"from"` + } + _ = json.Unmarshal(m.Params, &a) + stream := "LOG_" + Module + "_" + a.Log + subject := "mesh.log." + Module + "." + a.Log + s, err := p.js.Stream(p.ctx, stream) + if err != nil { + p.answer(m.ID, nil, err) + return + } + entry := func(r *jetstream.RawStreamMsg) map[string]any { + return map[string]any{"sequence": r.Sequence, "key": strings.TrimPrefix(r.Subject, subject+"."), + "value": json.RawMessage(r.Data), "at": r.Time} + } + switch m.Method { + case "mesh/log.append": + ack, err := p.js.Publish(p.ctx, subject+"."+a.Key, a.Value) + if err != nil { + p.answer(m.ID, nil, err) + return + } + p.answer(m.ID, map[string]any{"sequence": ack.Sequence}, nil) + case "mesh/log.last": + r, err := s.GetLastMsgForSubject(p.ctx, subject+"."+a.Key) + if errors.Is(err, jetstream.ErrMsgNotFound) { + p.answer(m.ID, nil, nil) + return + } + if err != nil { + p.answer(m.ID, nil, err) + return + } + p.answer(m.ID, entry(r), nil) + case "mesh/log.read": + info, err := s.Info(p.ctx) + if err != nil { + p.answer(m.ID, nil, err) + return + } + entries := []map[string]any{} + for seq := max(a.From, 1); seq <= info.State.LastSeq; seq++ { + if r, err := s.GetMsg(p.ctx, seq); err == nil { + entries = append(entries, entry(r)) + } + } + p.answer(m.ID, map[string]any{"entries": entries, "next": 0, "last": info.State.LastSeq}, nil) + case "mesh/log.info": + info, err := s.Info(p.ctx) + if err != nil { + p.answer(m.ID, nil, err) + return + } + p.answer(m.ID, map[string]any{"entries": info.State.Msgs, "bytes": info.State.Bytes, + "max_bytes": info.Config.MaxBytes, "first": info.State.FirstSeq, "last": info.State.LastSeq}, nil) + } +} + func (p *player) call(name, caller string, args map[string]any) (json.RawMessage, error) { p.mu.Lock() p.next++ @@ -266,11 +331,18 @@ func TestTheHandActLogWholeRecordsOnTheBus(t *testing.T) { if err != nil { t.Fatal(err) } - for _, b := range []string{Module + "_" + ActsState, Module + "_" + WarrantsState, "mesh-controller_hand-acts", "messenger_asks"} { + for _, b := range []string{Module + "_" + WarrantsState, "mesh-controller_hand-acts", "messenger_asks"} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil { t.Fatal(err) } } + // The log as the mesh makes it (ADR 0297): a stream kept by limits, refusing when full, never deleting. + logStream, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "LOG_" + Module + "_" + ActsLog, + Subjects: []string{"mesh.log." + Module + "." + ActsLog + ".>"}, MaxBytes: 64 << 20, Discard: jetstream.DiscardNew, + DenyDelete: true, DenyPurge: true, AllowDirect: true}) + if err != nil { + t.Fatal(err) + } recorded, err := nc.SubscribeSync("mesh.mod." + Module + ".event." + RecordedEvent) if err != nil { t.Fatal(err) @@ -296,10 +368,9 @@ func TestTheHandActLogWholeRecordsOnTheBus(t *testing.T) { if err := json.Unmarshal(raw, &r); err != nil || r.Act.By != "ace.mesh-config" { t.Fatalf("hand-act answered %s", raw) } - acts, _ := js.KeyValue(ctx, Module+"_"+ActsState) - e, err := acts.Get(ctx, r.Act.ID) - if err != nil || !strings.Contains(string(e.Value()), `"by":"ace.mesh-config"`) { - t.Fatalf("the act is not kept on the bus: %v", err) + e, err := logStream.GetLastMsgForSubject(ctx, "mesh.log."+Module+"."+ActsLog+"."+r.Act.ID) + if err != nil || !strings.Contains(string(e.Data), `"by":"ace.mesh-config"`) { + t.Fatalf("the act is not kept in the log on the bus: %v", err) } msg, err := recorded.NextMsg(10 * time.Second) if err != nil || !strings.Contains(string(msg.Data), r.Act.ID) { @@ -317,8 +388,8 @@ func TestTheHandActLogWholeRecordsOnTheBus(t *testing.T) { t.Fatal(err) } } - if keys, _ := acts.Keys(ctx); len(keys) != 2 { - t.Errorf("the log keeps %v, want the act and one warrant", keys) + if info, err := logStream.Info(ctx); err != nil || info.State.Msgs != 2 { + t.Errorf("the log keeps %v, want the act and one warrant (%v)", info, err) } listed, err := p.call(HandActsTool, "person.jochen", map[string]any{}) if err != nil || !strings.Contains(string(listed), "warrant-claude-code-a1") || !strings.Contains(string(listed), r.Act.ID) { diff --git a/modules/mesh-hand-acts/module.json b/modules/mesh-hand-acts/module.json index 3e562b9a..768f3e23 100644 --- a/modules/mesh-hand-acts/module.json +++ b/modules/mesh-hand-acts/module.json @@ -20,15 +20,11 @@ "mesh-controller.hand-acts", "messenger.asks" ], + "logs": [ + "acts" + ], "state": [ - { - "name": "acts", - "ttl-seconds": 34560000 - }, - { - "name": "warrants", - "ttl-seconds": 34560000 - } + "warrants" ], "own-secrets": { "broker": "${dir:state}/broker"