Carry the persisted sequence on bus events, and SDK stream helpers

api/pluggableharness/kernel/v1/events.proto

@@ -29,10 +29,27 @@ message BusEvent {
   string schema_version = 4;
 
   // When the kernel received the Publish call this event fans out from.
-  // MUST be set. Display-only — this bus assigns no sequence number and
-  // makes no cross-subscriber ordering guarantee
-  // (event-bus.md#delivery-semantics).
+  // MUST be set. Display-only: never order by this. The bus makes no
+  // cross-subscriber ordering guarantee (event-bus.md#delivery-semantics),
+  // and .claude/rules/determinism.md is explicit that wall-clock time is
+  // never an ordering authority — sequence below is.
   google.protobuf.Timestamp time = 5;
+
+  // The state-backend sequence of the persisted event this republishes,
+  // for events on the reserved kernel.event.* topics.
+  //
+  // Zero for everything else: a plugin's own Publish carries no persisted
+  // event and so has no sequence, which is a real distinction rather than
+  // a missing value.
+  //
+  // This exists because a subscriber has to merge two streams — the live
+  // bus and a ReadEvents backfill — into one ordered view, and until now
+  // only the backfill carried a sequence. A frontend attaching to a
+  // session in progress therefore had no way to order a live message
+  // against its own replayed history, and sorted new content above it.
+  // determinism.md makes sequence the sole ordering authority; a
+  // transport that drops it forces every subscriber to invent one.
+  int64 sequence = 6;
 }
 
 // StoredEvent is one persisted event, read back by ReadEvents. Mirrors
@@ -85,4 +102,28 @@ message TokenDelta {
   string target_id = 2;
   // The incremental text to append.
   string text = 3;
+
+  // Which of the model's output streams this fragment belongs to.
+  //
+  // Without it a frontend cannot tell reasoning from the answer on the
+  // fast path, so it must either show both as the reply — which reads as
+  // the model talking to itself — or show neither, which is what left
+  // reasoning invisible while a turn was thinking. Unset means text, so a
+  // producer written before this field behaves exactly as before.
+  DeltaKind kind = 4;
+}
+
+// DeltaKind names which model output stream a TokenDelta carries.
+//
+// Deliberately coarser than model.v1's ThinkingChannel: a frontend needs
+// to know that reasoning is happening, not which of a vendor's two
+// reasoning streams produced a fragment. The finer distinction survives on
+// the durable message, where a reader can act on it.
+enum DeltaKind {
+  // Zero value, and the assistant's ordinary reply text — a producer that
+  // never sets this field emits reply text, which is what every producer
+  // written before this enum did.
+  DELTA_KIND_UNSPECIFIED = 0;
+  // The model's reasoning output, of any channel.
+  DELTA_KIND_THINKING = 1;
 }

api/pluggableharness/session/v1/types.proto

@@ -186,4 +186,32 @@ message SessionState {
   // with nothing in the UI to attribute it to. Absent means the vendor
   // served what was asked for, or said nothing.
   optional string actual_model = 13;
+
+  // What the session is doing right now, as opposed to info.status, which
+  // is where the session sits in its lifecycle.
+  //
+  // The two answer different questions and a status bar needs the second
+  // one: a session is `running` from creation until it ends, so a frontend
+  // driven by status alone reads "running" while idle at the prompt and
+  // "running" while a model call is in flight. Phase is what distinguishes
+  // them.
+  SessionPhase phase = 14;
+}
+
+// SessionPhase is what a session is doing at this instant.
+//
+// Deliberately short. Every value here is one a frontend renders
+// differently; a phase nothing displays differently is a log line, not
+// protocol. Finer breakdowns (which tool, which turn step) are already
+// available as events on the transcript.
+enum SessionPhase {
+  // Zero value: the session has not reported a phase. A frontend SHOULD
+  // treat this as IDLE rather than rendering an unknown state — a session
+  // that never reports is not working.
+  SESSION_PHASE_UNSPECIFIED = 0;
+  // Nothing is in flight; the session is waiting on the operator.
+  SESSION_PHASE_IDLE = 1;
+  // A turn is in flight — the model is being called, or its tool calls
+  // are running.
+  SESSION_PHASE_GENERATING = 2;
 }

docs/specifications/event-bus.md

@@ -42,7 +42,11 @@ No other wildcard form exists in v1 — no mid-string wildcard, no multi-segment
 ## Delivery semantics
 
 - **Best-effort, not guaranteed.** A `Publish` call returns as soon as the kernel has fanned the event out to every currently-subscribed stream's queue; it does not wait for any subscriber to actually receive or process the event, and a subscriber that connects after a `Publish` call already returned never sees that event. There is no backlog and no replay — this is `internal/eventbus`'s own ephemeral contract, inherited unchanged.
-- **Per-subscriber ordering only.** A single `Subscribe` stream sees events matching its filters in the order they were published; there is no ordering guarantee across two different `Subscribe` streams, and no ordering guarantee relative to `ReadEvents` or anything hook-dispatch related. Nothing here carries a `sequence` number — `.claude/rules/determinism.md`'s ordering-authority rule governs persisted, replay-critical ordering, and this bus persists nothing and participates in no replay, so it is deliberately outside that rule's scope, exactly as `internal/eventbus`'s own design notes already state.
+- **Per-subscriber ordering only.** A single `Subscribe` stream sees events matching its filters in the order they were published; there is no ordering guarantee across two different `Subscribe` streams, or relative to anything hook-dispatch related. The bus itself assigns no sequence and participates in no replay, so `.claude/rules/determinism.md`'s ordering-authority rule does not govern the bus as a transport.
+
+- **A republished kernel event carries the sequence it was persisted under.** `BusEvent.sequence` is set for events on the reserved `kernel.event.*` topics and zero for everything else — a plugin's own `Publish` has no persisted event behind it, so it has no sequence, which is a real distinction rather than an absent value.
+
+  This does not make the bus ordered; it makes bus events *orderable against the log*. A subscriber that also reads history has to merge two streams into one view, and only `ReadEvents` carried a sequence, so a frontend attaching to a session already in progress had nothing to order a live message against its own backfill by, and sorted new content above replayed content. A subscriber merging the two MUST order by `sequence` and MUST NOT order by `BusEvent.time`, which is display-only for the reason determinism.md gives: wall-clock is never an ordering authority.
 - **Observe-only, never a veto.** Unlike a hook subscriber in `veto` mode, a bus subscriber cannot block, modify, or reject the event it received — the bus has no response channel for that at all. A slow or broken subscriber can only ever affect itself (see "Backpressure" below), never the publisher or any other subscriber.
 
 ## Backpressure

docs/specifications/kernel-callbacks.md

@@ -294,10 +294,17 @@ BusEvent {
   payload_type    string     // see Publish
   schema_version  string     // see Publish
   time            Timestamp  // when the kernel received the Publish this
-                              // event fans out from
+                              // event fans out from — display only,
+                              // never an ordering key
+  sequence        int64      // the state-backend sequence this event was
+                              // persisted under, for kernel.event.* topics;
+                              // zero for a plugin's own Publish, which has
+                              // no persisted event behind it
 }
 ```
 
+A subscriber that merges this live stream with a [`ReadEvents`](#readevents) backfill — which every frontend attaching to a session in progress does — MUST order the merged view by `sequence` and MUST NOT order it by `time`. Before `sequence` was carried here, only the backfill had it, so a frontend had no basis for ordering a live message against its own replayed history and placed new content above it.
+
 ## `ReadEvents`
 
 `ReadEvents` is server-streaming: it reads back the calling plugin's own session's persisted event log, ordered by `sequence` — never by `time`, per `.claude/rules/determinism.md`'s ordering rule, restated here because this is the one RPC on this service that reads the ordering-authoritative column back out.

internal/kernel/turnstack.go

@@ -226,23 +226,24 @@ func (k *kernel) newTurnDriver(ctx context.Context, sessionID string) (session.T
 		Logger:    k.logger,
 	})
 
-	var onDelta func(sessionID, targetID, text string)
+	var onDelta func(sessionID, targetID, text string, kind kernelv1.DeltaKind)
 	if k.deltas != nil {
-		onDelta = func(sessionID, targetID, text string) {
+		onDelta = func(sessionID, targetID, text string, kind kernelv1.DeltaKind) {
 			k.deltas.Publish(&kernelv1.TokenDelta{
 				SessionId: sessionID,
 				TargetId:  targetID,
 				Text:      text,
+				Kind:      kind,
 			})
 		}
 	}
 
 	caller := modelcall.New(modelcall.Config{
-		Retry:       retrypolicy.FromConfig(k.cfg.Settings.Retry, sessionMaxRetries),
-		Events:      k.sink,
-		Telemetry:   k.telem,
-		Logger:      k.logger,
-		OnTextDelta: onDelta,
+		Retry:     retrypolicy.FromConfig(k.cfg.Settings.Retry, sessionMaxRetries),
+		Events:    k.sink,
+		Telemetry: k.telem,
+		Logger:    k.logger,
+		OnDelta:   onDelta,
 	})
 
 	planRes, err := newPlanResolver(k) //nolint:contextcheck

internal/modelcall/complete.go

@@ -217,9 +217,17 @@ func (c *Caller) doAttempt(ctx context.Context, req Request, attemptNum int) (ou
 			spanErr = err
 			return completion{}, nil, err
 		}
-		if c.cfg.OnTextDelta != nil {
+		if c.cfg.OnDelta != nil {
 			if td := ev.GetTextDelta(); td != nil && td.GetText() != "" {
-				c.cfg.OnTextDelta(req.SessionID, req.MessageID, td.GetText())
+				c.cfg.OnDelta(req.SessionID, req.MessageID, td.GetText(), kernelv1.DeltaKind_DELTA_KIND_UNSPECIFIED)
+			}
+			// Thinking rides the same fast path, tagged. The channel a
+			// vendor put it on is deliberately not forwarded: a status
+			// bar needs to know reasoning is happening, and the finer
+			// summary-vs-content split survives on the durable message
+			// for a reader that can act on it.
+			if td := ev.GetThinkingDelta(); td != nil && td.GetText() != "" {
+				c.cfg.OnDelta(req.SessionID, req.MessageID, td.GetText(), kernelv1.DeltaKind_DELTA_KIND_THINKING)
 			}
 		}
 	}

internal/modelcall/modelcall.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	contentv1 "github.com/pluggableharness/agent/pkg/content/proto/v1"
+	kernelv1 "github.com/pluggableharness/agent/pkg/kernel/proto/v1"
 	modelv1 "github.com/pluggableharness/agent/pkg/model/proto/v1"
 
 	"github.com/pluggableharness/agent/internal/providercatalog"
@@ -67,11 +68,16 @@ type Config struct {
 	Telemetry *telemetry.Provider
 	// Logger is this Caller's structured logger.
 	Logger *slog.Logger
-	// OnTextDelta is called for every text_delta StreamEvent after it is
-	// successfully Observed, for the live token fast path. MAY be nil.
-	// targetID is the completion's MessageID so frontends can correlate
-	// consecutive deltas into one growing block.
-	OnTextDelta func(sessionID, targetID, text string)
+	// OnDelta is called for every text_delta and thinking_delta
+	// StreamEvent after it is successfully Observed, for the live token
+	// fast path. MAY be nil. targetID is the completion's MessageID so
+	// frontends can correlate consecutive deltas into one growing block.
+	//
+	// kind distinguishes the two. Forwarding thinking at all is the point:
+	// reasoning previously reached the durable message and nothing else,
+	// so a frontend had no way to show that a turn was thinking rather
+	// than hung — and a reasoning model spends most of a turn there.
+	OnDelta func(sessionID, targetID, text string, kind kernelv1.DeltaKind)
 }
 
 // Request is one StreamCompletion invocation: which model to call, the

internal/session/handle.go

@@ -13,6 +13,7 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	contentv1 "github.com/pluggableharness/agent/pkg/content/proto/v1"
+	kernelv1 "github.com/pluggableharness/agent/pkg/kernel/proto/v1"
 	sessionv1 "github.com/pluggableharness/agent/pkg/session/proto/v1"
 
 	"github.com/pluggableharness/agent/internal/bounds"
@@ -24,6 +25,11 @@ import (
 // TopicState is the event-bus topic for SessionState republish.
 const TopicState = "kernel.state"
 
+// stateSchemaVersion is the SessionState payload's schema version, per
+// state-backend.md#events' schema_version convention. A breaking change
+// to SessionState's shape ships as session.v2 plus this becoming "2".
+const stateSchemaVersion = "1"
+
 // ErrHandleClosed reports use of a Handle after Close.
 var ErrHandleClosed = errors.New("session: handle closed")
 
@@ -125,11 +131,21 @@ func (h *Handle) Submit(ctx context.Context, content []*contentv1.ContentBlock)
 	h.cancel = cancel
 	h.mu.Unlock()
 
+	// Announce the phase change before any work starts and again once the
+	// turn loop is done, whichever way it exits. Publishing only on
+	// completion would leave a frontend showing "idle" for the whole time
+	// the model is actually working, which is the entire thing the phase
+	// exists to fix.
+	h.setPhase(runCtx, sessionv1.SessionPhase_SESSION_PHASE_GENERATING)
+
 	defer func() {
 		h.mu.Lock()
 		h.busy = false
 		h.cancel = nil
 		h.mu.Unlock()
+		// WithoutCancel: an interrupted or failed turn still has to leave
+		// the status bar honest, and runCtx is already Done on those paths.
+		h.setPhase(context.WithoutCancel(runCtx), sessionv1.SessionPhase_SESSION_PHASE_IDLE)
 		cancel()
 	}()
 
@@ -270,9 +286,18 @@ func (h *Handle) State(ctx context.Context) (*sessionv1.SessionState, error) {
 			Provider: h.st.res.model.Ref.Provider,
 		}
 	}
+	state.Phase = h.st.phase
 	if h.st.res.target != nil && h.st.res.target.GetEffectiveCeiling() > 0 {
+		// The last completion's own occupancy when there is one, falling
+		// back to the assembled-context sum before the first model call —
+		// which is the only point at which that sum is the best available
+		// answer rather than a structural zero.
+		used := h.st.contextTokens
+		if used == 0 {
+			used = h.st.assembled
+		}
 		state.Context = &sessionv1.ContextState{
-			UsedTokens:   h.st.assembled,
+			UsedTokens:   used,
 			WindowTokens: h.st.res.target.GetEffectiveCeiling(),
 		}
 	}
@@ -302,6 +327,19 @@ func (h *Handle) Info(ctx context.Context) (*sessionv1.SessionInfo, error) {
 	return state.GetInfo(), nil
 }
 
+// setPhase records the session's current activity and publishes the new
+// state, so a frontend learns about it without polling.
+//
+// A publish failure is logged and swallowed: the phase is a status-bar
+// hint, and failing a turn because a bus publish did not land would trade
+// a working session for a cosmetic update.
+func (h *Handle) setPhase(ctx context.Context, phase sessionv1.SessionPhase) {
+	h.st.phase = phase
+	if err := h.publishState(ctx); err != nil {
+		h.r.logger.DebugContext(ctx, "session: publish phase", "phase", phase.String(), "err", err)
+	}
+}
+
 func (h *Handle) publishState(ctx context.Context) error {
 	state, err := h.State(ctx)
 	if err != nil {
@@ -311,7 +349,23 @@ func (h *Handle) publishState(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	return h.r.bus.Publish(ctx, eventbus.Event{Topic: TopicState, Payload: payload})
+	// The bus event's Payload must be a *kernelv1.BusEvent, not the raw
+	// marshaled bytes. internal/kernelcallback's Subscribe handler type-
+	// asserts for that shape and drops anything else, so publishing bytes
+	// here meant every SessionState republish was logged as
+	// "received a non-BusEvent payload, dropping" and no frontend ever saw
+	// a state update after its initial snapshot — the status bar simply
+	// froze at whatever was true when it attached.
+	return h.r.bus.Publish(ctx, eventbus.Event{
+		Topic: TopicState,
+		Payload: &kernelv1.BusEvent{
+			Topic:         TopicState,
+			Payload:       payload,
+			PayloadType:   string((&sessionv1.SessionState{}).ProtoReflect().Descriptor().FullName()),
+			SchemaVersion: stateSchemaVersion,
+			Time:          timestamppb.New(h.r.clock()),
+		},
+	})
 }
 
 func (h *Handle) markTerminal(ctx context.Context, status sessionv1.SessionStatus) error {

internal/session/run.go

@@ -38,6 +38,22 @@ type run struct {
 	inTokens  int64
 	outTokens int64
 
+	// contextTokens is how much of the model's window the last completion
+	// actually occupied: that call's input tokens plus the cache reads
+	// that also sat in the window but are billed separately (Usage
+	// documents cache_read_tokens as never also counted in input_tokens).
+	//
+	// This is the honest answer to "how full is the context", and it is
+	// not `assembled` above. assembled sums the context-provider sections
+	// the chain contributed, which is zero for a session with no context
+	// plugins loaded — so a status bar reading it showed 0% forever while
+	// the conversation grew.
+	contextTokens int64
+
+	// phase is what the session is doing right now, distinct from its
+	// lifecycle status. Written by Handle.Submit around a turn.
+	phase sessionv1.SessionPhase
+
 	// Vendor-reported state from the most recent completion, surfaced on
 	// SessionState. Last-write-wins rather than accumulated: each is a
 	// reading of "how is the vendor serving me right now", and an older
@@ -307,6 +323,15 @@ func (st *run) absorb(result turn.Result) {
 	st.inTokens += result.Usage.GetInputTokens()
 	st.outTokens += result.Usage.GetOutputTokens()
 
+	// Window occupancy is the last call's figure, not a running sum:
+	// each completion resends the whole conversation, so adding turns
+	// together would report several multiples of a window that only ever
+	// held one conversation. A turn that reported no input tokens leaves
+	// the previous reading alone rather than zeroing the gauge.
+	if in := result.Usage.GetInputTokens(); in > 0 {
+		st.contextTokens = in + result.Usage.GetCacheReadTokens()
+	}
+
 	// Only overwrite what this turn actually reported. A vendor that
 	// publishes budgets on some responses and not others would otherwise
 	// blank the operator's meter on every quiet turn.

internal/sessionstate/emit.go

@@ -201,6 +201,11 @@ func (l *Live) republish(ctx context.Context, id string, seq int64, kind kernelv
 		PayloadType:   payloadType,
 		SchemaVersion: schemaVersion,
 		Time:          timestamppb.New(at),
+		// The sequence the state backend just assigned. Carrying it is
+		// what lets a subscriber merge this live event with a ReadEvents
+		// backfill into one correctly ordered view; without it a frontend
+		// attaching mid-session sorts new messages above its own history.
+		Sequence: seq,
 	}
 
 	if pubErr := l.bus.Publish(ctx, eventbus.Event{Topic: topic, Payload: busEvent}); pubErr != nil {

internal/sessionstate/emit_test.go

@@ -92,6 +92,14 @@ func TestLive_Emit_writesAndRepublishes(t *testing.T) {
 	if !busEvent.GetTime().AsTime().Equal(now) {
 		t.Errorf("BusEvent.Time = %v, want %v", busEvent.GetTime().AsTime(), now)
 	}
+	// The sequence must ride along, and must match the durable row's.
+	// A subscriber merges this live event with a ReadEvents backfill into
+	// one ordered view; without the sequence it has nothing to order by
+	// but arrival, and a frontend attaching mid-session sorts new
+	// messages above its own replayed history.
+	if busEvent.GetSequence() != outcome.Sequence {
+		t.Errorf("BusEvent.Sequence = %d, want %d (the persisted sequence)", busEvent.GetSequence(), outcome.Sequence)
+	}
 
 	// The persisted row must match what was published.
 	var found *statebackend.Event