Fodesu · Oct 3, 2026 · Oct 3, 2026
diff --git a/agentcore/observe/bus.go b/agentcore/observe/bus.go
@@ -0,0 +1,244 @@
+// Package observe carries a Session's observations (OBS-1) in two streams.
+// The Bus is the committed stream: a writer.CommitObserver that decodes
+// every applied commit through the Registry and fans it out, in commit
+// order, to the subscribers of each Session. Its subscription is a catch-up
+// subscription: it may start from a CommitSeq, reading the ledger up to its
+// head and continuing live, so a consumer that keeps its own checkpoint
+// resumes after a crash without a gap. The Progresses are the transient
+// stream: what the executor reported of an effect while it ran, and the
+// failures of background work -- none of it a fact, all of it expendable. A
+// host that serves one audience from both merges the two.
+package observe
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"github.com/felinics/twilight/agentcore/ledger"
+	"github.com/felinics/twilight/agentcore/module"
+	"github.com/felinics/twilight/agentcore/run"
+	"github.com/felinics/twilight/agentcore/session"
+	"sync"
+)
+
+// Event is one item of a Session's observation stream, in either of two
+// shapes. A committed Event carries one event decoded from the applied
+// commits through the Registry: Module, Version and Value are the decoded
+// payload, Unknown reports a type or version this process has no codec for
+// (the event is still delivered) and Err reports the decode or history-read
+// failure. A transient Event has a zero Row: Progress carries what the
+// executor reported of an effect while it ran, or Err alone carries a
+// failure of background work; neither is a fact, either may be lost, and
+// the committed result that follows replaces it.
+type Event struct {
+	Session session.SessionID
+	// Position is the event's place in the Session's ledger: the commit's
+	// Seq and the event's index within the commit. It is what a consumer
+	// checkpoints; zero on a transient Event.
+	Position ledger.Position
+	Row      ledger.Event
+	Module   module.ModuleKey
+	Version  module.PayloadVersion
+	Value    any
+	Unknown  bool
+	Err      error
+	// Progress is set for a transient observation of an effect in flight
+	// (RUN-EXE-12, OBS-1): a model or tool delta relayed from the executor,
+	// or a reset that voids the deltas received so far. Such an Event has a
+	// zero Row: it is not a fact, may be lost, and is replaced by the
+	// committed result that follows.
+	Progress *Progress
+}
+
+// Progress is the transient part of an Event: what the executor reported of
+// an effect while it ran.
+type Progress struct {
+	RunID      run.RunID
+	Effect     run.EffectID
+	Generation int
+	Sequence   uint64
+	Kind       string
+	Payload    json.RawMessage
+}
+
+// History reads a Session's committed history: the Store, or anything that
+// serves its ReadCommits. It is what a catch-up subscription reads before
+// it goes live.
+type History interface {
+	ReadCommits(context.Context, session.CommitReadRequest) (session.CommitPage, error)
+}
+
+// ErrNoHistory reports SubscribeFrom on a Bus built without a History.
+var ErrNoHistory = errors.New("observe: the bus has no history to catch up from")
+
+// Bus decodes applied commits and fans them out per Session. It carries
+// committed facts only; the transient observations of the running effects
+// are the Progresses' stream.
+type Bus struct {
+	registry *module.Registry
+	history  History
+	mu       sync.Mutex
+	subs     map[session.SessionID]map[*subscriber]struct{}
+}
+
+// NewBus returns a Bus decoding through registry. history, when not nil,
+// lets SubscribeFrom catch up from the ledger; a Bus without it serves live
+// subscriptions only.
+func NewBus(registry *module.Registry, history History) *Bus {
+	return &Bus{registry: registry, history: history, subs: make(map[session.SessionID]map[*subscriber]struct{})}
+}
+
+// Committed is writer.CommitObserver: one Event per committed event, in
+// commit order.
+func (b *Bus) Committed(_ context.Context, sid session.SessionID, commit ledger.Commit) {
+	b.publish(sid, b.decode(sid, &commit)...)
+}
+
+// decode renders one commit as its Events, each at its Position.
+func (b *Bus) decode(sid session.SessionID, commit *ledger.Commit) []Event {
+	var events []Event
+	index := uint32(0)
+	for _, batch := range commit.Batches {
+		for _, row := range batch.Events {
+			e := Event{Session: sid, Position: ledger.Position{Commit: commit.Seq, Index: index}, Row: row}
+			index++
+			decoded, err := b.registry.Decode(row)
+			if err != nil {
+				e.Unknown, e.Err = true, err
+			} else {
+				e.Module, e.Version, e.Value, e.Unknown = decoded.Module, decoded.Version, decoded.Value, decoded.Unknown
+			}
+			events = append(events, e)
+		}
+	}
+	return events
+}
+
+func (b *Bus) publish(sid session.SessionID, events ...Event) {
+	b.mu.Lock()
+	subs := make([]*subscriber, 0, len(b.subs[sid]))
+	for s := range b.subs[sid] {
+		subs = append(subs, s)
+	}
+	b.mu.Unlock()
+	for _, s := range subs {
+		s.push(events)
+	}
+}
+
+// Subscribe registers a subscriber to one Session's stream from this moment
+// on; the channel closes when ctx is done.
+func (b *Bus) Subscribe(ctx context.Context, sid session.SessionID) <-chan Event {
+	s := b.attach(sid)
+	go s.drain(ctx, func() { b.detach(sid, s) })
+	return s.out
+}
+
+// SubscribeFrom is the catch-up form of Subscribe: every committed event of
+// the Session from CommitSeq from, in order, then the live stream. The
+// subscriber is attached before the history is read, so no commit made
+// meanwhile is lost, and a live commit the history already covered is
+// dropped, so none is delivered twice. The channel closes when ctx is done;
+// a read failure closes it after an Event with Err.
+func (b *Bus) SubscribeFrom(ctx context.Context, sid session.SessionID, from ledger.CommitSeq) (<-chan Event, error) {
+	if b.history == nil {
+		return nil, ErrNoHistory
+	}
+	s := b.attach(sid)
+	page, err := b.history.ReadCommits(ctx, session.CommitReadRequest{SessionID: sid, From: from})
+	if err != nil {
+		b.detach(sid, s)
+		return nil, err
+	}
+	var history []Event
+	for i := range page.Commits {
+		history = append(history, b.decode(sid, &page.Commits[i])...)
+	}
+	s.mu.Lock()
+	s.history, s.skipBelow = history, page.Head.Next
+	s.mu.Unlock()
+	go s.drain(ctx, func() { b.detach(sid, s) })
+	return s.out, nil
+}
+
+func (b *Bus) attach(sid session.SessionID) *subscriber {
+	s := &subscriber{out: make(chan Event, 64), wake: make(chan struct{}, 1)}
+	b.mu.Lock()
+	if b.subs[sid] == nil {
+		b.subs[sid] = make(map[*subscriber]struct{})
+	}
+	b.subs[sid][s] = struct{}{}
+	b.mu.Unlock()
+	return s
+}
+
+func (b *Bus) detach(sid session.SessionID, s *subscriber) {
+	b.mu.Lock()
+	delete(b.subs[sid], s)
+	if len(b.subs[sid]) == 0 {
+		delete(b.subs, sid)
+	}
+	b.mu.Unlock()
+}
+
+// subscriber decouples the committer from the consumer: push appends to an
+// unbounded queue and never blocks, so a slow reader delays only its own
+// delivery, never a Commit; drain moves the history, then the queue, into
+// the channel in order and closes it when the subscription ends. skipBelow
+// is the ledger head the history covered: a live committed Event below it
+// was delivered from the history already.
+type subscriber struct {
+	mu        sync.Mutex
+	history   []Event
+	skipBelow ledger.CommitSeq
+	queue     []Event
+	out       chan Event
+	wake      chan struct{}
+}
+
+func (s *subscriber) push(events []Event) {
+	s.mu.Lock()
+	s.queue = append(s.queue, events...)
+	s.mu.Unlock()
+	select {
+	case s.wake <- struct{}{}:
+	default:
+	}
+}
+
+func (s *subscriber) drain(ctx context.Context, unsubscribe func()) {
+	defer func() {
+		unsubscribe()
+		close(s.out)
+	}()
+	for {
+		s.mu.Lock()
+		history, live, skipBelow := s.history, s.queue, s.skipBelow
+		s.history, s.queue = nil, nil
+		s.mu.Unlock()
+		for i := range history {
+			select {
+			case s.out <- history[i]:
+			case <-ctx.Done():
+				return
+			}
+		}
+		for i := range live {
+			e := &live[i]
+			if e.Position.Commit < skipBelow {
+				// Delivered from the history already.
+				continue
+			}
+			select {
+			case s.out <- *e:
+			case <-ctx.Done():
+				return
+			}
+		}
+		select {
+		case <-s.wake:
+		case <-ctx.Done():
+			return
+		}
+	}
+}
diff --git a/agentcore/observe/bus_test.go b/agentcore/observe/bus_test.go
@@ -0,0 +1,145 @@
+package observe_test
+
+import (
+	"context"
+	"errors"
+	"github.com/felinics/twilight/agentcore/ledger"
+	"github.com/felinics/twilight/agentcore/module"
+	"github.com/felinics/twilight/agentcore/observe"
+	"github.com/felinics/twilight/agentcore/session"
+	"github.com/felinics/twilight/agentcore/session/filestore/filestoretest"
+	"github.com/felinics/twilight/agentcore/session/writer"
+	"testing"
+	"time"
+)
+
+type rowPayload struct {
+	Text string `json:"text"`
+}
+
+const rowType ledger.EventType = "twilight/z/row"
+
+func registry(t *testing.T) *module.Registry {
+	t.Helper()
+	r, err := module.BuildRegistry(module.ModuleDescriptor{Source: module.SourceTwilight, ID: "z",
+		Streams: []module.StreamDefinition{{Domain: "z", Inheritance: module.Inherited}},
+		Events: []module.EventDefinition{{Type: rowType, Domain: "z",
+			Codecs: map[module.PayloadVersion]module.PayloadCodec{1: module.JSONCodec[rowPayload]{}}}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return r
+}
+
+func commitRow(t *testing.T, w writer.Writer, id, text string) {
+	t.Helper()
+	res, err := w.Commit(context.Background(), func(writer.View) (*writer.SemanticGroup, error) {
+		return &writer.SemanticGroup{CommitID: ledger.CommitID(id), Batches: []writer.TypedBatch{{Domain: ledger.Domain{Name: "z"},
+			Events: []writer.TypedEvent{{Type: rowType, Value: rowPayload{Text: text}}}}}}, nil
+	})
+	if err != nil || res.Outcome != writer.CommitApplied {
+		t.Fatalf("commit %s: outcome=%s err=%v", id, res.Outcome, err)
+	}
+}
+
+func texts(t *testing.T, ch <-chan observe.Event, n int) ([]string, []ledger.Position) {
+	t.Helper()
+	var got []string
+	var at []ledger.Position
+	deadline := time.After(5 * time.Second)
+	for len(got) < n {
+		select {
+		case e, ok := <-ch:
+			if !ok {
+				t.Fatalf("stream closed after %d of %d events", len(got), n)
+			}
+			if e.Err != nil {
+				t.Fatal(e.Err)
+			}
+			got = append(got, e.Value.(rowPayload).Text)
+			at = append(at, e.Position)
+		case <-deadline:
+			t.Fatalf("got %d of %d events: %v", len(got), n, got)
+    		}
+	}
+	return got, at
+}
+
+// A catch-up subscription delivers the ledger from the requested position,
+// then the live commits, each event once and in order, with the Position a
+// consumer checkpoints; a live-only subscription sees only what follows.
+func TestSubscribeFromCatchesUpThenGoesLive(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	store := filestoretest.Store(t)
+	reg := registry(t)
+	bus := observe.NewBus(reg, store)
+	const sid session.SessionID = "s1"
+	if _, err := store.Create(ctx, session.CreateRequest{SessionID: sid}); err != nil {
+		t.Fatal(err)
+	}
+	w, err := writer.NewWriters(store, reg, writer.Admission{}, session.OpenOptions{}, writer.WritersConfig{Observers: []writer.CommitObserver{bus}}).Writer(ctx, sid)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer w.Close(ctx)
+	commitRow(t, w, "c1", "a")
+	commitRow(t, w, "c2", "b")
+	commitRow(t, w, "c3", "c")
+
+	// Each case commits one more row after subscribing, so the ledger grows
+	// a, b, c | d | e | f and every case sees its history plus its own live
+	// commit, nothing of the cases before or after.
+	cases := []struct {
+		name string
+		from ledger.CommitSeq
+		live string
+		want []string
+	}{
+		{"from the start", 0, "d", []string{"a", "b", "c", "d"}},
+		{"from a checkpoint", 2, "e", []string{"c", "d", "e"}},
+		{"from the head", 5, "f", []string{"f"}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			sctx, scancel := context.WithCancel(ctx)
+			defer scancel()
+			ch, err := bus.SubscribeFrom(sctx, sid, tc.from)
+			if err != nil {
+				t.Fatal(err)
+			}
+			commitRow(t, w, "live-"+tc.live, tc.live)
+			got, at := texts(t, ch, len(tc.want))
+			for i := range got {
+				if got[i] != tc.want[i] {
+					t.Fatalf("events = %v, want %v", got, tc.want)
+				}
+			}
+			for i := 1; i < len(at); i++ {
+				if !at[i-1].Less(at[i]) {
+					t.Fatalf("positions not increasing: %v", at)
+				}
+			}
+			if at[0].Commit < tc.from {
+				t.Fatalf("first position %v is below from %d", at[0], tc.from)
+			}
+			// Nothing else is pending: the history and the live commit were
+			// delivered exactly once.
+			select {
+			case e := <-ch:
+				t.Fatalf("unexpected extra event %v", e)
+			case <-time.After(50 * time.Millisecond):
+			}
+		})
+	}
+
+	live := bus.Subscribe(ctx, sid)
+	commitRow(t, w, "g", "g")
+	got, _ := texts(t, live, 1)
+	if got[0] != "g" {
+		t.Fatalf("live = %v, want [g]", got)
+	}
+	if _, err := observe.NewBus(reg, nil).SubscribeFrom(ctx, sid, 0); !errors.Is(err, observe.ErrNoHistory) {
+		t.Fatalf("SubscribeFrom without history = %v, want ErrNoHistory", err)
+	}
+}