From b9b589a02c746c54a48f1c05c446a0475474c4ca Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 14:47:28 +0100 Subject: [PATCH] Teller: the hand acts through the seat's client, its findings through the conditions client's writer, the assigner's asks followed from the shared events stream, the warrants handed over by the views reader's one consumer; contract import paths follow the domain folders (D-12..D-22 teller part; issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-teller-tools/main.go | 4 +-- .../cmd/mesh-delivery-teller/assignerasks.go | 4 +-- .../mesh-delivery-teller/assignerasks_test.go | 4 +-- .../cmd/mesh-delivery-teller/decided.go | 31 ++++++++++++------- .../mesh-delivery-teller/integration_test.go | 12 +++---- .../cmd/mesh-delivery-teller/main.go | 15 ++++----- .../cmd/mesh-delivery-teller/room_test.go | 4 +-- .../cmd/mesh-delivery-teller/teller.go | 4 +-- .../cmd/mesh-delivery-teller/teller_test.go | 4 +-- .../cmd/mesh-delivery-teller/viewanswers.go | 2 +- .../mesh-delivery-teller/viewanswers_test.go | 4 +-- modules/mesh-delivery-teller/go.mod | 6 ++-- 12 files changed, 52 insertions(+), 42 deletions(-) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go index dd9bfa22..ba794d32 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go @@ -12,8 +12,8 @@ import ( "sync" "time" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-teller/internal/ends" @@ -107,7 +107,7 @@ func main() { // teller_first_views and teller_first_view_again: where the teller's first view of the assigner's asks stands, and // asking it again by hand (ADR 0615, design 25 ยง 12.2). tools := append(Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now), - views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, time.Now)...) + views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-teller-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go index 950cc248..60f55915 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go @@ -5,8 +5,8 @@ import ( "fmt" "slices" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go index e1017c9c..1e8a143d 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go @@ -2,8 +2,8 @@ package main import ( "encoding/json" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "strings" "testing" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go index c46c60b5..6f850e5c 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go @@ -4,15 +4,14 @@ import ( "context" "encoding/json" "errors" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/process" ) // The teller's three bus adapters for mesh-sdk go/asker, which in a bundle reaches the bus through the runtime: here the @@ -84,16 +83,24 @@ func decidedVia(hear func(func(jetstream.Msg)) error, logf func(string, ...any)) } } -// pullDecided starts the pull of the module's durable consumer (process.HearEvents) and stops it when ctx ends. -func pullDecided(ctx context.Context, cons jetstream.Consumer, logf func(string, ...any)) func(func(jetstream.Msg)) error { - return func(handle func(jetstream.Msg)) error { - cc, err := process.HearEvents(cons, handle, logf) - if err != nil { - return err - } - context.AfterFunc(ctx, cc.Stop) - return nil +// decidedHandler is where the router's warrants are handed: the module has one durable consumer of the shared events stream, +// pulled by the views reader (views.ProcessReader.Start), which hands every message that is not a views event to its Other. +type decidedHandler struct{ handle func(jetstream.Msg) } + +// hear registers the handler of the warrants; it pulls nothing. +func (d *decidedHandler) hear(handle func(jetstream.Msg)) error { + d.handle = handle + return nil +} + +// other is the views reader's Other: a message of the consumer that is no views event is a warrant (or nothing the teller +// hears: it is left for the bus to deliver again). +func (d *decidedHandler) other(m jetstream.Msg) { + if d.handle == nil { + _ = m.Nak() + return } + d.handle(m) } // sayVia publishes a condition on the seat's one subject for conditions, under the module's own name. diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index cda389ea..0e8c6027 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go @@ -6,8 +6,8 @@ import ( "context" "encoding/json" "fmt" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/views/viewstest" "os" "path/filepath" @@ -189,7 +189,7 @@ func fullConf(dir string) string { teller = append(teller, readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads // The assigner's asks by its read tool and its views family (ADR 0615, DLV-M3), and the teller's own first-views and // findings; the bucket itself is no longer the teller's to read. - teller = append(append(append(teller, viewstest.Grants(delivery.AssignerModule, delivery.AssignerModule+".assigner_read_asks", "assigner-views", tellerConsumer)...), keepsOf(firstViewsBucket)...), keepsOf(findingsBucket)...) + teller = append(append(append(teller, viewstest.Grants(delivery.AssignerModule, delivery.AssignerModule+".assigner_read_asks", tellerConsumer)...), keepsOf(firstViewsBucket)...), keepsOf(findingsBucket)...) view := append(append(append(readOf(tellerAnswers), readOf(gateAsksKV)...), readOf(assignerAsks)...), keepsOf(viewAnswers)...) assigner := keepsOf(assignerAsks) tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...) @@ -265,15 +265,15 @@ func startFull(t *testing.T) *fullBed { Retention: jetstream.WorkQueuePolicy}); err != nil { t.Fatal(err) } - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.seat.operator-channel.event.>"}}); err != nil { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.seat.operator-channel.event.>", "mesh.mod.*.event.>"}}); err != nil { t.Fatal(err) } - if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: tellerConsumer, FilterSubject: seatDecided, + if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: tellerConsumer, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } // The assigner as the owner of its asks: its read tool and its views family (ADR 0615). - viewstest.Serve(t, ctx, js, ctl, delivery.AssignerModule, "asks", delivery.AssignerModule+".assigner_read_asks", "assigner-views", tellerConsumer) + viewstest.Serve(t, ctx, js, ctl, delivery.AssignerModule, "asks", delivery.AssignerModule+".assigner_read_asks", tellerConsumer) return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx} } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index 623f3724..b306cbc0 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -11,7 +11,8 @@ import ( "context" "errors" "fmt" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings" "io" "log" "os" @@ -69,13 +70,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } state := kvStore{ctx: ctx, kv: kvs[1]} publish := publishVia(ctx, js) - cons, err := cred.EventConsumer(ctx, js, process.EventsStream) - if err != nil { + if _, err := cred.EventConsumer(ctx, js, process.EventsStream); err != nil { // Essential: the router's warrants are how an answer reaches the teller. return fmt.Errorf("the consumer %s on %s cannot be read (%v): the teller hears no answer without it, so it stops; where the bus refused it or did not answer, this module's consumer (made with its grant for the seat it uses) has not reached the bus yet", cred.EventConsumerName(), process.EventsStream, err) } + warrants := &decidedHandler{} client := &occ.Client{Module: delivery.TellerModule, State: state, Publish: publish, - Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(pullDecided(ctx, cons, logger.Printf), logger.Printf)} + Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(warrants.hear, logger.Printf)} t := &Teller{Asker: client, Store: state, Answers: kvAnswers{ctx: ctx, kv: kvs[0]}, Gate: kvGate{ctx: ctx, kv: gate}, Say: sayVia(ctx, js, delivery.TellerModule), Logf: logger.Printf, Publish: publish} // Hear the warrants first, so none that comes while the asks are read is missed. @@ -92,10 +93,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("the teller's connection for its tool calls: %w", err) } defer calls.Close() - reader := &views.ProcessReader{Module: delivery.TellerModule, Call: calls.Call, Logf: logger.Printf} + reader := &views.ProcessReader{Module: delivery.TellerModule, Call: calls.Call, Logf: logger.Printf, Other: warrants.other} findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf) - reader.Findings = findings.KV - asks, err := reader.Follow(delivery.AssignerModule, assignerAsksState, delivery.AssignerModule+".assigner_read_asks", "assigner-views") + reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) } + asks, err := reader.Follow(delivery.AssignerModule, assignerAsksState, delivery.AssignerModule+".assigner_read_asks") if err != nil { return err } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go index 2884b08c..cdf1275c 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go @@ -2,8 +2,8 @@ package main import ( "encoding/json" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "strings" "testing" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go index 380f28ab..1ee44804 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go @@ -10,8 +10,8 @@ import ( "sync" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go index 2dea32e7..65b87038 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go @@ -5,8 +5,8 @@ import ( "encoding/json" "errors" "fmt" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "strings" "testing" "time" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go index 42a865dd..293055e7 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go @@ -1,7 +1,7 @@ package main import ( - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" "strings" "time" diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go index da35692f..726ff215 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go @@ -1,8 +1,8 @@ package main import ( - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "strings" "testing" "time" diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index d2606693..3df9f4fc 100644 --- a/modules/mesh-delivery-teller/go.mod +++ b/modules/mesh-delivery-teller/go.mod @@ -3,6 +3,8 @@ module mesh-delivery-teller go 1.26.0 require ( + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 @@ -23,8 +25,8 @@ require ( ) require ( - git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639 - git.novox.be/novox/mesh-contracts/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639 ) replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk