Keep the hand-act log as mesh-hand-acts' module log, the stream ADR 0565 §10 names, so every act is held in its order (issue 1953)

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.
This commit is contained in:
2026-10-29 21:56:42 +01:00
parent e1cdeebcdf
commit 5b2b70cd6c
7 changed files with 309 additions and 102 deletions
+6 -3
View File
@@ -6,9 +6,12 @@ alone writes the log.
## What it keeps
- `mesh-hand-acts_acts`: every act, one key `act-<ns>-<n>` (`warrant-<asker>-<ask>` 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 `<asker>.<ask>` 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-<ns>-<n>` (`warrant-<asker>-<ask>` 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 `<asker>.<ask>` 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
@@ -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 `<asker>.<ask>` 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 `<asker>.<ask>` 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
}
@@ -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")
@@ -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) }}
@@ -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 {
@@ -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_<module>_<name> on mesh.log.<module>.<name>.<key>.
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) {
+4 -8
View File
@@ -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"