diff --git a/api/pluggableharness/kernel/v1/events.proto b/api/pluggableharness/kernel/v1/events.proto
index 3dd1d89..5e8ce4d 100644
--- a/api/pluggableharness/kernel/v1/events.proto
+++ b/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;
 }
diff --git a/api/pluggableharness/session/v1/types.proto b/api/pluggableharness/session/v1/types.proto
index a81f2ac..4446fc6 100644
--- a/api/pluggableharness/session/v1/types.proto
+++ b/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;
 }
diff --git a/docs/specifications/event-bus.md b/docs/specifications/event-bus.md
index d91135e..d7b5e13 100644
--- a/docs/specifications/event-bus.md
+++ b/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
diff --git a/docs/specifications/kernel-callbacks.md b/docs/specifications/kernel-callbacks.md
index d31f4b9..f6bc3eb 100644
--- a/docs/specifications/kernel-callbacks.md
+++ b/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.
diff --git a/internal/kernel/turnstack.go b/internal/kernel/turnstack.go
index 57107ae..930b5e7 100644
--- a/internal/kernel/turnstack.go
+++ b/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
diff --git a/internal/modelcall/complete.go b/internal/modelcall/complete.go
index dea5098..4107601 100644
--- a/internal/modelcall/complete.go
+++ b/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)
 			}
 		}
 	}
diff --git a/internal/modelcall/modelcall.go b/internal/modelcall/modelcall.go
index 73e3365..d4da960 100644
--- a/internal/modelcall/modelcall.go
+++ b/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
diff --git a/internal/session/handle.go b/internal/session/handle.go
index 2cd4666..71375fa 100644
--- a/internal/session/handle.go
+++ b/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 {
diff --git a/internal/session/run.go b/internal/session/run.go
index 14882eb..2afe2bb 100644
--- a/internal/session/run.go
+++ b/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.
diff --git a/internal/sessionstate/emit.go b/internal/sessionstate/emit.go
index 998f42d..6efdb54 100644
--- a/internal/sessionstate/emit.go
+++ b/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 {
diff --git a/internal/sessionstate/emit_test.go b/internal/sessionstate/emit_test.go
index 040838e..2a79773 100644
--- a/internal/sessionstate/emit_test.go
+++ b/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
diff --git a/internal/streamaccum/streamaccum.go b/internal/streamaccum/streamaccum.go
index c7f1974..95d1500 100644
--- a/internal/streamaccum/streamaccum.go
+++ b/internal/streamaccum/streamaccum.go
@@ -87,6 +87,12 @@ type Accumulator struct {
 	// block belongs to, so a summary run and a raw-reasoning run stay
 	// separate blocks even when adjacent.
 	openThinkingChannel modelv1.StreamEvent_ThinkingChannel
+	// openThinkingHasPart / openThinkingPart are the vendor part_index for
+	// the open thinking block. HasPart is false when the provider left
+	// part_index unset (single-part stream); a different part within the
+	// same channel opens a new block per events.proto.
+	openThinkingHasPart bool
+	openThinkingPart    int32
 	pendingSig          []byte
 	tools               map[string]*toolCallState
 
@@ -167,7 +173,7 @@ func (a *Accumulator) Observe(ev *modelv1.StreamEvent) error {
 		a.observeTextDelta(e.TextDelta.GetText())
 		return nil
 	case *modelv1.StreamEvent_ThinkingDelta_:
-		a.observeThinkingDelta(e.ThinkingDelta.GetText(), e.ThinkingDelta.GetChannel())
+		a.observeThinkingDelta(e.ThinkingDelta.GetText(), e.ThinkingDelta.GetChannel(), e.ThinkingDelta.PartIndex)
 		return nil
 	case *modelv1.StreamEvent_ThinkingSignature_:
 		return a.observeThinkingSignature(e.ThinkingSignature.GetSignature())
@@ -270,6 +276,9 @@ func (a *Accumulator) observeMetadata(ev *modelv1.StreamEvent_StreamMetadata) {
 	if v := ev.CatalogEtag; v != nil {
 		m.CatalogEtag = proto.String(*v)
 	}
+	if v := ev.StickyTurnToken; v != nil {
+		m.StickyTurnToken = proto.String(*v)
+	}
 	if len(ev.GetRateLimits()) > 0 {
 		m.RateLimits = ev.GetRateLimits()
 	}
@@ -342,6 +351,8 @@ func (a *Accumulator) closeOpenBlock() {
 	a.openKind = blockKindNone
 	a.openText = nil
 	a.openThinking = nil
+	a.openThinkingHasPart = false
+	a.openThinkingPart = 0
 	a.pendingSig = nil
 }
 
@@ -363,24 +374,52 @@ func (a *Accumulator) observeTextDelta(text string) {
 // thinking block, opening a new one first if the previous event wasn't
 // itself a ThinkingDelta (or a ThinkingSignature belonging to the same
 // block) continuing it.
-func (a *Accumulator) observeThinkingDelta(text string, channel modelv1.StreamEvent_ThinkingChannel) {
-	// A channel switch closes the open block even though both sides are
-	// thinking. A vendor-written summary and the raw reasoning it
-	// summarizes are different text; concatenating them because they are
-	// adjacent and both "thinking" would produce one block that reads as
-	// neither. Providers that set no channel leave this UNSPECIFIED
-	// throughout, so their runs coalesce exactly as before.
-	if a.openKind != blockKindThinking || a.openThinkingChannel != channel {
+//
+// partIndex is the optional vendor part number. Fragments that share a
+// (channel, partIndex) pair are one block; a different partIndex (or a
+// switch between set and unset) starts a new block within the same
+// channel. Absent partIndex means a single-part stream, which is the
+// pre-part_index behaviour and remains the common case.
+func (a *Accumulator) observeThinkingDelta(text string, channel modelv1.StreamEvent_ThinkingChannel, partIndex *int32) {
+	// A channel or part switch closes the open block even though both
+	// sides are thinking. A vendor-written summary and the raw reasoning
+	// it summarizes are different text; concatenating them because they
+	// are adjacent and both "thinking" would produce one block that reads
+	// as neither. Numbered parts within one channel are likewise separate
+	// blocks (events.proto ThinkingDelta.part_index). Providers that set
+	// no channel and no part leave both UNSPECIFIED/absent throughout, so
+	// their runs coalesce exactly as before.
+	same := a.openKind == blockKindThinking &&
+		a.openThinkingChannel == channel &&
+		thinkingPartEqual(a.openThinkingHasPart, a.openThinkingPart, partIndex)
+	if !same {
 		a.closeOpenBlock()
 		th := &contentv1.ThinkingBlock{}
 		a.blocks = append(a.blocks, &contentv1.ContentBlock{Block: &contentv1.ContentBlock_Thinking{Thinking: th}})
 		a.openThinking = th
 		a.openKind = blockKindThinking
 		a.openThinkingChannel = channel
+		if partIndex != nil {
+			a.openThinkingHasPart = true
+			a.openThinkingPart = *partIndex
+		} else {
+			a.openThinkingHasPart = false
+			a.openThinkingPart = 0
+		}
 	}
 	a.openThinking.Text += text
 }
 
+// thinkingPartEqual reports whether the open block's part key matches the
+// incoming optional part_index. Both absent is a match (single-part stream);
+// one absent and one set is not; both set match only when equal.
+func thinkingPartEqual(openHas bool, openPart int32, incoming *int32) bool {
+	if incoming == nil {
+		return !openHas
+	}
+	return openHas && openPart == *incoming
+}
+
 // observeThinkingSignature accumulates signature bytes onto the currently
 // open thinking block's pending signature buffer — stored and returned
 // verbatim, never decoded or reinterpreted, per this package's doc comment
diff --git a/internal/streamaccum/streamaccum_test.go b/internal/streamaccum/streamaccum_test.go
index ef13474..0fc61ba 100644
--- a/internal/streamaccum/streamaccum_test.go
+++ b/internal/streamaccum/streamaccum_test.go
@@ -793,6 +793,42 @@ func TestStreamMetadata_mergesFieldByField(t *testing.T) {
 	}
 }
 
+// TestStreamMetadata_stickyTurnTokenMerges is the edge case the sticky
+// field exists for: a provider reports it once (usually with response.id
+// or a product turn-state header) and later metadata without it must not
+// clear the handle the kernel will echo as sticky_turn_token.
+func TestStreamMetadata_stickyTurnTokenMerges(t *testing.T) {
+	t.Parallel()
+
+	a := observeAll(t, []*modelv1.StreamEvent{
+		metaEvent(&modelv1.StreamEvent_StreamMetadata{
+			StickyTurnToken: proto.String("resp_1"),
+			ActualModel:     proto.String("gpt-x"),
+		}),
+		metaEvent(&modelv1.StreamEvent_StreamMetadata{
+			ServiceTier: proto.String("default"),
+		}),
+		stopEvent(modelv1.StopReason_STOP_REASON_END_TURN, ""),
+	})
+
+	m := a.Metadata()
+	if got := m.GetStickyTurnToken(); got != "resp_1" {
+		t.Errorf("StickyTurnToken = %q, want resp_1 carried forward", got)
+	}
+	if got := m.GetServiceTier(); got != "default" {
+		t.Errorf("ServiceTier = %q, want default", got)
+	}
+	// A later event that does set sticky supersedes.
+	a2 := observeAll(t, []*modelv1.StreamEvent{
+		metaEvent(&modelv1.StreamEvent_StreamMetadata{StickyTurnToken: proto.String("resp_1")}),
+		metaEvent(&modelv1.StreamEvent_StreamMetadata{StickyTurnToken: proto.String("resp_2")}),
+		stopEvent(modelv1.StopReason_STOP_REASON_END_TURN, ""),
+	})
+	if got := a2.Metadata().GetStickyTurnToken(); got != "resp_2" {
+		t.Errorf("StickyTurnToken = %q, want resp_2 from later event", got)
+	}
+}
+
 // TestStreamMetadata_rateLimitsReplaceWholesale asserts rate_limits is
 // snapshot-replaced rather than merged entry by entry. Each event
 // carries the vendor's complete budget picture, so merging would
@@ -883,6 +919,14 @@ func thinkingOn(text string, ch modelv1.StreamEvent_ThinkingChannel) *modelv1.St
 	}}
 }
 
+// thinkingOnPart builds a channel+part_index thinking delta.
+func thinkingOnPart(text string, ch modelv1.StreamEvent_ThinkingChannel, part int32) *modelv1.StreamEvent {
+	idx := part
+	return &modelv1.StreamEvent{Event: &modelv1.StreamEvent_ThinkingDelta_{
+		ThinkingDelta: &modelv1.StreamEvent_ThinkingDelta{Text: text, Channel: ch, PartIndex: &idx},
+	}}
+}
+
 // TestThinkingChannel_switchClosesTheBlock asserts a summary run and a
 // raw-reasoning run stay separate blocks even when adjacent. Merging them
 // on adjacency alone would produce one block that reads as neither.
@@ -932,6 +976,56 @@ func TestThinkingChannel_unspecifiedStillCoalesces(t *testing.T) {
 	}
 }
 
+// TestThinkingPartIndex_switchClosesTheBlock asserts vendor-numbered
+// reasoning parts within one channel stay separate blocks. Concatenating
+// on channel alone would invent a single block the vendor never produced.
+func TestThinkingPartIndex_switchClosesTheBlock(t *testing.T) {
+	t.Parallel()
+
+	ch := modelv1.StreamEvent_THINKING_CHANNEL_CONTENT
+	a := observeAll(t, []*modelv1.StreamEvent{
+		thinkingOnPart("part0a ", ch, 0),
+		thinkingOnPart("part0b", ch, 0),
+		thinkingOnPart("part1", ch, 1),
+		stopEvent(modelv1.StopReason_STOP_REASON_END_TURN, ""),
+	})
+
+	msg, _, _, ok := a.Result()
+	if !ok {
+		t.Fatal("Result reported ok = false after a Stop event")
+	}
+	if len(msg.GetContent()) != 2 {
+		t.Fatalf("Content has %d blocks, want 2 (one per part_index)", len(msg.GetContent()))
+	}
+	if got := msg.GetContent()[0].GetThinking().GetText(); got != "part0a part0b" {
+		t.Errorf("block 0 = %q, want coalesced part 0", got)
+	}
+	if got := msg.GetContent()[1].GetThinking().GetText(); got != "part1" {
+		t.Errorf("block 1 = %q, want part 1", got)
+	}
+}
+
+// TestThinkingPartIndex_absentStillCoalesces is the compatibility
+// guarantee for ThinkingDeltaOn (no part): runs stay one block.
+func TestThinkingPartIndex_absentStillCoalesces(t *testing.T) {
+	t.Parallel()
+
+	ch := modelv1.StreamEvent_THINKING_CHANNEL_SUMMARY
+	a := observeAll(t, []*modelv1.StreamEvent{
+		thinkingOn("sum ", ch),
+		thinkingOn("mary", ch),
+		stopEvent(modelv1.StopReason_STOP_REASON_END_TURN, ""),
+	})
+
+	msg, _, _, _ := a.Result()
+	if len(msg.GetContent()) != 1 {
+		t.Fatalf("Content has %d blocks, want 1", len(msg.GetContent()))
+	}
+	if got := msg.GetContent()[0].GetThinking().GetText(); got != "sum mary" {
+		t.Errorf("text = %q, want %q", got, "sum mary")
+	}
+}
+
 // TestSafetyNotice_recordedInOrderAndNotABlockBoundary covers both
 // properties at once: the sequence is the explanation an operator needs,
 // and interposing on a text run must not split it.
diff --git a/pkg/kernel/proto/v1/events.pb.go b/pkg/kernel/proto/v1/events.pb.go
index cecabb1..648d425 100644
--- a/pkg/kernel/proto/v1/events.pb.go
+++ b/pkg/kernel/proto/v1/events.pb.go
@@ -23,6 +23,62 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
+// 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.
+type DeltaKind int32
+
+const (
+	// 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.
+	DeltaKind_DELTA_KIND_UNSPECIFIED DeltaKind = 0
+	// The model's reasoning output, of any channel.
+	DeltaKind_DELTA_KIND_THINKING DeltaKind = 1
+)
+
+// Enum value maps for DeltaKind.
+var (
+	DeltaKind_name = map[int32]string{
+		0: "DELTA_KIND_UNSPECIFIED",
+		1: "DELTA_KIND_THINKING",
+	}
+	DeltaKind_value = map[string]int32{
+		"DELTA_KIND_UNSPECIFIED": 0,
+		"DELTA_KIND_THINKING":    1,
+	}
+)
+
+func (x DeltaKind) Enum() *DeltaKind {
+	p := new(DeltaKind)
+	*p = x
+	return p
+}
+
+func (x DeltaKind) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (DeltaKind) Descriptor() protoreflect.EnumDescriptor {
+	return file_pluggableharness_kernel_v1_events_proto_enumTypes[0].Descriptor()
+}
+
+func (DeltaKind) Type() protoreflect.EnumType {
+	return &file_pluggableharness_kernel_v1_events_proto_enumTypes[0]
+}
+
+func (x DeltaKind) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use DeltaKind.Descriptor instead.
+func (DeltaKind) EnumDescriptor() ([]byte, []int) {
+	return file_pluggableharness_kernel_v1_events_proto_rawDescGZIP(), []int{0}
+}
+
 // BusEvent is one event delivered to a Subscribe stream. See
 // kernel-callbacks.md's Subscribe and event-bus.md.
 type BusEvent struct {
@@ -37,10 +93,26 @@ type BusEvent struct {
 	// Versions payload_type. See PublishRequest.schema_version.
 	SchemaVersion string `protobuf:"bytes,4,opt,name=schema_version,json=schemaVersion,proto3" json:"schema_version,omitempty"`
 	// 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).
-	Time          *timestamppb.Timestamp `protobuf:"bytes,5,opt,name=time,proto3" json:"time,omitempty"`
+	// 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.
+	Time *timestamppb.Timestamp `protobuf:"bytes,5,opt,name=time,proto3" json:"time,omitempty"`
+	// 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.
+	Sequence      int64 `protobuf:"varint,6,opt,name=sequence,proto3" json:"sequence,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -110,6 +182,13 @@ func (x *BusEvent) GetTime() *timestamppb.Timestamp {
 	return nil
 }
 
+func (x *BusEvent) GetSequence() int64 {
+	if x != nil {
+		return x.Sequence
+	}
+	return 0
+}
+
 // StoredEvent is one persisted event, read back by ReadEvents. Mirrors
 // state-backend.md's events table row. See kernel-callbacks.md's
 // ReadEvents.
@@ -236,7 +315,15 @@ type TokenDelta struct {
 	// deltas into one growing piece of text.
 	TargetId string `protobuf:"bytes,2,opt,name=target_id,json=targetId,proto3" json:"target_id,omitempty"`
 	// The incremental text to append.
-	Text          string `protobuf:"bytes,3,opt,name=text,proto3" json:"text,omitempty"`
+	Text string `protobuf:"bytes,3,opt,name=text,proto3" json:"text,omitempty"`
+	// 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.
+	Kind          DeltaKind `protobuf:"varint,4,opt,name=kind,proto3,enum=pluggableharness.kernel.v1.DeltaKind" json:"kind,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -292,17 +379,25 @@ func (x *TokenDelta) GetText() string {
 	return ""
 }
 
+func (x *TokenDelta) GetKind() DeltaKind {
+	if x != nil {
+		return x.Kind
+	}
+	return DeltaKind_DELTA_KIND_UNSPECIFIED
+}
+
 var File_pluggableharness_kernel_v1_events_proto protoreflect.FileDescriptor
 
 const file_pluggableharness_kernel_v1_events_proto_rawDesc = "" +
 	"\n" +
-	"'pluggableharness/kernel/v1/events.proto\x12\x1apluggableharness.kernel.v1\x1a\x1fgoogle/protobuf/timestamp.proto\x1a&pluggableharness/common/v1/types.proto\x1a&pluggableharness/kernel/v1/types.proto\"\xb4\x01\n" +
+	"'pluggableharness/kernel/v1/events.proto\x12\x1apluggableharness.kernel.v1\x1a\x1fgoogle/protobuf/timestamp.proto\x1a&pluggableharness/common/v1/types.proto\x1a&pluggableharness/kernel/v1/types.proto\"\xd0\x01\n" +
 	"\bBusEvent\x12\x14\n" +
 	"\x05topic\x18\x01 \x01(\tR\x05topic\x12\x18\n" +
 	"\apayload\x18\x02 \x01(\fR\apayload\x12!\n" +
 	"\fpayload_type\x18\x03 \x01(\tR\vpayloadType\x12%\n" +
 	"\x0eschema_version\x18\x04 \x01(\tR\rschemaVersion\x12.\n" +
-	"\x04time\x18\x05 \x01(\v2\x1a.google.protobuf.TimestampR\x04time\"\xaa\x02\n" +
+	"\x04time\x18\x05 \x01(\v2\x1a.google.protobuf.TimestampR\x04time\x12\x1a\n" +
+	"\bsequence\x18\x06 \x01(\x03R\bsequence\"\xaa\x02\n" +
 	"\vStoredEvent\x12\x1a\n" +
 	"\bsequence\x18\x01 \x01(\x03R\bsequence\x12\x0e\n" +
 	"\x02id\x18\x02 \x01(\tR\x02id\x12.\n" +
@@ -310,13 +405,17 @@ const file_pluggableharness_kernel_v1_events_proto_rawDesc = "" +
 	"\x04kind\x18\x04 \x01(\x0e2%.pluggableharness.kernel.v1.EventKindR\x04kind\x12C\n" +
 	"\bproducer\x18\x05 \x01(\v2'.pluggableharness.common.v1.ProducerRefR\bproducer\x12%\n" +
 	"\x0eschema_version\x18\x06 \x01(\tR\rschemaVersion\x12\x18\n" +
-	"\apayload\x18\a \x01(\fR\apayload\"\\\n" +
+	"\apayload\x18\a \x01(\fR\apayload\"\x97\x01\n" +
 	"\n" +
 	"TokenDelta\x12\x1d\n" +
 	"\n" +
 	"session_id\x18\x01 \x01(\tR\tsessionId\x12\x1b\n" +
 	"\ttarget_id\x18\x02 \x01(\tR\btargetId\x12\x12\n" +
-	"\x04text\x18\x03 \x01(\tR\x04textB@Z>github.com/pluggableharness/agent/pkg/kernel/proto/v1;kernelv1b\x06proto3"
+	"\x04text\x18\x03 \x01(\tR\x04text\x129\n" +
+	"\x04kind\x18\x04 \x01(\x0e2%.pluggableharness.kernel.v1.DeltaKindR\x04kind*@\n" +
+	"\tDeltaKind\x12\x1a\n" +
+	"\x16DELTA_KIND_UNSPECIFIED\x10\x00\x12\x17\n" +
+	"\x13DELTA_KIND_THINKING\x10\x01B@Z>github.com/pluggableharness/agent/pkg/kernel/proto/v1;kernelv1b\x06proto3"
 
 var (
 	file_pluggableharness_kernel_v1_events_proto_rawDescOnce sync.Once
@@ -330,25 +429,28 @@ func file_pluggableharness_kernel_v1_events_proto_rawDescGZIP() []byte {
 	return file_pluggableharness_kernel_v1_events_proto_rawDescData
 }
 
+var file_pluggableharness_kernel_v1_events_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
 var file_pluggableharness_kernel_v1_events_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
 var file_pluggableharness_kernel_v1_events_proto_goTypes = []any{
-	(*BusEvent)(nil),              // 0: pluggableharness.kernel.v1.BusEvent
-	(*StoredEvent)(nil),           // 1: pluggableharness.kernel.v1.StoredEvent
-	(*TokenDelta)(nil),            // 2: pluggableharness.kernel.v1.TokenDelta
-	(*timestamppb.Timestamp)(nil), // 3: google.protobuf.Timestamp
-	(EventKind)(0),                // 4: pluggableharness.kernel.v1.EventKind
-	(*v1.ProducerRef)(nil),        // 5: pluggableharness.common.v1.ProducerRef
+	(DeltaKind)(0),                // 0: pluggableharness.kernel.v1.DeltaKind
+	(*BusEvent)(nil),              // 1: pluggableharness.kernel.v1.BusEvent
+	(*StoredEvent)(nil),           // 2: pluggableharness.kernel.v1.StoredEvent
+	(*TokenDelta)(nil),            // 3: pluggableharness.kernel.v1.TokenDelta
+	(*timestamppb.Timestamp)(nil), // 4: google.protobuf.Timestamp
+	(EventKind)(0),                // 5: pluggableharness.kernel.v1.EventKind
+	(*v1.ProducerRef)(nil),        // 6: pluggableharness.common.v1.ProducerRef
 }
 var file_pluggableharness_kernel_v1_events_proto_depIdxs = []int32{
-	3, // 0: pluggableharness.kernel.v1.BusEvent.time:type_name -> google.protobuf.Timestamp
-	3, // 1: pluggableharness.kernel.v1.StoredEvent.time:type_name -> google.protobuf.Timestamp
-	4, // 2: pluggableharness.kernel.v1.StoredEvent.kind:type_name -> pluggableharness.kernel.v1.EventKind
-	5, // 3: pluggableharness.kernel.v1.StoredEvent.producer:type_name -> pluggableharness.common.v1.ProducerRef
-	4, // [4:4] is the sub-list for method output_type
-	4, // [4:4] is the sub-list for method input_type
-	4, // [4:4] is the sub-list for extension type_name
-	4, // [4:4] is the sub-list for extension extendee
-	0, // [0:4] is the sub-list for field type_name
+	4, // 0: pluggableharness.kernel.v1.BusEvent.time:type_name -> google.protobuf.Timestamp
+	4, // 1: pluggableharness.kernel.v1.StoredEvent.time:type_name -> google.protobuf.Timestamp
+	5, // 2: pluggableharness.kernel.v1.StoredEvent.kind:type_name -> pluggableharness.kernel.v1.EventKind
+	6, // 3: pluggableharness.kernel.v1.StoredEvent.producer:type_name -> pluggableharness.common.v1.ProducerRef
+	0, // 4: pluggableharness.kernel.v1.TokenDelta.kind:type_name -> pluggableharness.kernel.v1.DeltaKind
+	5, // [5:5] is the sub-list for method output_type
+	5, // [5:5] is the sub-list for method input_type
+	5, // [5:5] is the sub-list for extension type_name
+	5, // [5:5] is the sub-list for extension extendee
+	0, // [0:5] is the sub-list for field type_name
 }
 
 func init() { file_pluggableharness_kernel_v1_events_proto_init() }
@@ -362,13 +464,14 @@ func file_pluggableharness_kernel_v1_events_proto_init() {
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_pluggableharness_kernel_v1_events_proto_rawDesc), len(file_pluggableharness_kernel_v1_events_proto_rawDesc)),
-			NumEnums:      0,
+			NumEnums:      1,
 			NumMessages:   3,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
 		GoTypes:           file_pluggableharness_kernel_v1_events_proto_goTypes,
 		DependencyIndexes: file_pluggableharness_kernel_v1_events_proto_depIdxs,
+		EnumInfos:         file_pluggableharness_kernel_v1_events_proto_enumTypes,
 		MessageInfos:      file_pluggableharness_kernel_v1_events_proto_msgTypes,
 	}.Build()
 	File_pluggableharness_kernel_v1_events_proto = out.File
diff --git a/pkg/model/doc.go b/pkg/model/doc.go
index 8e1de9b..cb37de9 100644
--- a/pkg/model/doc.go
+++ b/pkg/model/doc.go
@@ -10,9 +10,11 @@
 // A plugin author implements Provider — Capabilities, Configure, and
 // StreamCompletion, the three MUST RPCs
 // (docs/specifications/model/conformance.md's summary matrix) — and
-// optionally TokenCounter and Renderer for the SHOULD/MAY RPCs CountTokens
-// and Render (docs/specifications/model/protocol.md#counttokens,
-// docs/specifications/model/protocol.md#render). NewService adapts a
+// optionally TokenCounter, Renderer, and Accounter for the SHOULD/MAY RPCs
+// CountTokens, Render, and GetAccount
+// (docs/specifications/model/protocol.md#counttokens,
+// docs/specifications/model/protocol.md#render,
+// docs/specifications/model/protocol.md#getaccount). NewService adapts a
 // Provider into the generated modelv1.ModelServiceServer, implementing
 // Describe itself from a plugin.Identity (docs/specifications/model/protocol.md#describe)
 // so no author code is needed for that RPC.
@@ -46,10 +48,14 @@
 // StreamEvent is likewise not mirrored as a struct an author constructs
 // and returns; stream.go's Sink is the domain-friendly StreamEvent
 // surface instead — one method per variant (TextDelta, ThinkingDelta,
-// ToolCallStart, ...), so an author never touches modelv1.StreamEvent's
-// oneof directly, and Sink enforces the "exactly one terminal event"
-// invariant (docs/specifications/model/data-types.md#streamevent)
-// mechanically rather than by convention.
+// ToolCallStart, Metadata, SafetyNotice, ...), so an author never
+// touches modelv1.StreamEvent's oneof directly, and Sink enforces the
+// "exactly one terminal event" invariant
+// (docs/specifications/model/data-types.md#streamevent) mechanically
+// rather than by convention. Optional wire fields that most vendors omit
+// have an explicit helper rather than overloading the simple form:
+// StreamStartWith for correlation_ids, ThinkingDeltaOnPart for
+// part_index, ThinkingDeltaOn for thinking channels.
 //
 // # Errors
 //
diff --git a/pkg/model/stream.go b/pkg/model/stream.go
index ad65ee3..4efa178 100644
--- a/pkg/model/stream.go
+++ b/pkg/model/stream.go
@@ -72,11 +72,42 @@ func (s *Sink) send(ev *modelv1.StreamEvent, terminal bool) error {
 // it is needed, which is why this is a separate early event rather than a
 // field on Stop. A Provider whose vendor publishes no such id simply never
 // calls this.
+//
+// When the vendor also publishes secondary handles (x-request-id, cf-ray,
+// a response id distinct from the stream id), use StreamStartWith so
+// support tooling can look the request up under any of them.
 func (s *Sink) StreamStart(providerRequestID string) error {
+	return s.StreamStartWith(providerRequestID, nil)
+}
+
+// StreamStartWith is StreamStart plus every other handle the vendor uses
+// for the same request, keyed by the vendor's own name for it
+// ("x-request-id", "response_id", "cf-ray").
+//
+// providerRequestID remains the single canonical handle on the wire;
+// correlationIDs carries the rest rather than forcing the adapter to
+// discard them. A nil or empty map is equivalent to StreamStart.
+//
+// The kernel sorts map keys before persisting
+// (docs/specifications/model/data-types.md; .claude/rules/determinism.md);
+// adapters need not pre-sort.
+func (s *Sink) StreamStartWith(providerRequestID string, correlationIDs map[string]string) error {
+	start := &modelv1.StreamEvent_StreamStart{ProviderRequestId: providerRequestID}
+	if len(correlationIDs) > 0 {
+		// Copy so a caller reusing the map cannot mutate an in-flight send.
+		cp := make(map[string]string, len(correlationIDs))
+		for k, v := range correlationIDs {
+			if k == "" {
+				continue
+			}
+			cp[k] = v
+		}
+		if len(cp) > 0 {
+			start.CorrelationIds = cp
+		}
+	}
 	return s.send(&modelv1.StreamEvent{
-		Event: &modelv1.StreamEvent_StreamStart_{
-			StreamStart: &modelv1.StreamEvent_StreamStart{ProviderRequestId: providerRequestID},
-		},
+		Event: &modelv1.StreamEvent_StreamStart_{StreamStart: start},
 	}, false)
 }
 
@@ -263,11 +294,32 @@ func (s *Sink) SafetyNotice(kind modelv1.StreamEvent_SafetyKind, message string,
 // two: the kernel keeps a summary run and a content run as separate
 // blocks, and sending both through the untagged ThinkingDelta would
 // concatenate them into one block that reads as neither.
+//
+// When the vendor also numbers parallel reasoning parts within a channel,
+// use ThinkingDeltaOnPart so fragments of different parts are not
+// concatenated into one block.
 func (s *Sink) ThinkingDeltaOn(text string, channel modelv1.StreamEvent_ThinkingChannel) error {
+	return s.thinkingDelta(text, channel, nil)
+}
+
+// ThinkingDeltaOnPart is ThinkingDeltaOn with a vendor-supplied part index.
+//
+// Fragments that share a (channel, partIndex) pair are one reasoning block;
+// a different partIndex starts a new block within the same channel. Use
+// this only when the vendor actually numbers parts — inventing indices
+// would invent block boundaries the kernel cannot later re-split.
+func (s *Sink) ThinkingDeltaOnPart(text string, channel modelv1.StreamEvent_ThinkingChannel, partIndex int32) error {
+	idx := partIndex
+	return s.thinkingDelta(text, channel, &idx)
+}
+
+func (s *Sink) thinkingDelta(text string, channel modelv1.StreamEvent_ThinkingChannel, partIndex *int32) error {
+	delta := &modelv1.StreamEvent_ThinkingDelta{Text: text, Channel: channel}
+	if partIndex != nil {
+		delta.PartIndex = partIndex
+	}
 	return s.send(&modelv1.StreamEvent{
-		Event: &modelv1.StreamEvent_ThinkingDelta_{
-			ThinkingDelta: &modelv1.StreamEvent_ThinkingDelta{Text: text, Channel: channel},
-		},
+		Event: &modelv1.StreamEvent_ThinkingDelta_{ThinkingDelta: delta},
 	}, false)
 }
 
diff --git a/pkg/model/stream_internal_test.go b/pkg/model/stream_internal_test.go
index 8260636..f274ca6 100644
--- a/pkg/model/stream_internal_test.go
+++ b/pkg/model/stream_internal_test.go
@@ -58,12 +58,24 @@ func TestSink_EventVariants(t *testing.T) {
 	stream := newFakeServerStream(t.Context())
 	sink := newSink(stream)
 
+	if err := sink.StreamStartWith("resp_1", map[string]string{
+		"x-request-id": "req_abc",
+		"cf-ray":       "ray-1",
+	}); err != nil {
+		t.Fatalf("StreamStartWith() = %v, want nil", err)
+	}
 	if err := sink.TextDelta("hello"); err != nil {
 		t.Fatalf("TextDelta() = %v, want nil", err)
 	}
 	if err := sink.ThinkingDelta("reasoning"); err != nil {
 		t.Fatalf("ThinkingDelta() = %v, want nil", err)
 	}
+	if err := sink.ThinkingDeltaOn("summary", modelv1.StreamEvent_THINKING_CHANNEL_SUMMARY); err != nil {
+		t.Fatalf("ThinkingDeltaOn() = %v, want nil", err)
+	}
+	if err := sink.ThinkingDeltaOnPart("part-raw", modelv1.StreamEvent_THINKING_CHANNEL_CONTENT, 2); err != nil {
+		t.Fatalf("ThinkingDeltaOnPart() = %v, want nil", err)
+	}
 	if err := sink.ThinkingSignature([]byte("sig")); err != nil {
 		t.Fatalf("ThinkingSignature() = %v, want nil", err)
 	}
@@ -79,25 +91,66 @@ func TestSink_EventVariants(t *testing.T) {
 	if err := sink.ToolCallDone("call-1"); err != nil {
 		t.Fatalf("ToolCallDone() = %v, want nil", err)
 	}
+	actual := "gpt-served"
+	if err := sink.Metadata(StreamMetadata{ActualModel: &actual}); err != nil {
+		t.Fatalf("Metadata() = %v, want nil", err)
+	}
+	if err := sink.SafetyNotice(modelv1.StreamEvent_SAFETY_KIND_BUFFERING, "reviewing", map[string]string{"k": "v"}); err != nil {
+		t.Fatalf("SafetyNotice() = %v, want nil", err)
+	}
 	reasoning := int64(5)
 	if err := sink.Usage(Usage{InputTokens: 10, OutputTokens: 20, ReasoningTokens: &reasoning}); err != nil {
 		t.Fatalf("Usage() = %v, want nil", err)
 	}
 
 	events := stream.events()
-	if len(events) != 8 {
-		t.Fatalf("len(events) = %d, want 8", len(events))
+	// start + text + think + summary + part + sig + redacted + tool×3 + meta + safety + usage = 13
+	if len(events) != 13 {
+		t.Fatalf("len(events) = %d, want 13", len(events))
 	}
-	if events[0].GetTextDelta().GetText() != "hello" {
-		t.Errorf("events[0].TextDelta.Text = %q, want %q", events[0].GetTextDelta().GetText(), "hello")
+	start := events[0].GetStreamStart()
+	if start.GetProviderRequestId() != "resp_1" {
+		t.Errorf("StreamStart.id = %q, want resp_1", start.GetProviderRequestId())
+	}
+	if start.GetCorrelationIds()["x-request-id"] != "req_abc" {
+		t.Errorf("correlation_ids = %v", start.GetCorrelationIds())
+	}
+	if events[1].GetTextDelta().GetText() != "hello" {
+		t.Errorf("events[1].TextDelta.Text = %q, want %q", events[1].GetTextDelta().GetText(), "hello")
+	}
+	part := events[4].GetThinkingDelta()
+	if part.GetChannel() != modelv1.StreamEvent_THINKING_CHANNEL_CONTENT || part.GetPartIndex() != 2 {
+		t.Errorf("ThinkingDelta part = channel=%v part=%d", part.GetChannel(), part.GetPartIndex())
 	}
 	// RedactedThinking carries the vendor's bytes through untouched — the
 	// kernel round-trips them verbatim or the vendor rejects the next turn.
-	if got := string(events[3].GetRedactedThinking().GetData()); got != "encrypted" {
-		t.Errorf("events[3].RedactedThinking.Data = %q, want %q", got, "encrypted")
+	if got := string(events[6].GetRedactedThinking().GetData()); got != "encrypted" {
+		t.Errorf("RedactedThinking.Data = %q, want %q", got, "encrypted")
+	}
+	if events[10].GetMetadata().GetActualModel() != "gpt-served" {
+		t.Errorf("Metadata.ActualModel = %q", events[10].GetMetadata().GetActualModel())
+	}
+	if events[11].GetSafetyNotice().GetKind() != modelv1.StreamEvent_SAFETY_KIND_BUFFERING {
+		t.Errorf("SafetyNotice.Kind = %v", events[11].GetSafetyNotice().GetKind())
+	}
+	if events[12].GetUsage().GetReasoningTokens() != 5 {
+		t.Errorf("Usage.ReasoningTokens = %d, want 5", events[12].GetUsage().GetReasoningTokens())
+	}
+}
+
+func TestSink_StreamStartWithEmptyMap(t *testing.T) {
+	t.Parallel()
+	stream := newFakeServerStream(t.Context())
+	sink := newSink(stream)
+	if err := sink.StreamStartWith("id", map[string]string{"": "skip", "ok": "v"}); err != nil {
+		t.Fatal(err)
+	}
+	start := stream.events()[0].GetStreamStart()
+	if _, has := start.GetCorrelationIds()[""]; has {
+		t.Fatal("empty key must be dropped")
 	}
-	if events[7].GetUsage().GetReasoningTokens() != 5 {
-		t.Errorf("events[7].Usage.ReasoningTokens = %d, want 5", events[7].GetUsage().GetReasoningTokens())
+	if start.GetCorrelationIds()["ok"] != "v" {
+		t.Fatalf("%v", start.GetCorrelationIds())
 	}
 }
 
diff --git a/pkg/session/proto/v1/types.pb.go b/pkg/session/proto/v1/types.pb.go
index f71cb18..f86e880 100644
--- a/pkg/session/proto/v1/types.pb.go
+++ b/pkg/session/proto/v1/types.pb.go
@@ -111,6 +111,67 @@ func (SessionStatus) EnumDescriptor() ([]byte, []int) {
 	return file_pluggableharness_session_v1_types_proto_rawDescGZIP(), []int{0}
 }
 
+// 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.
+type SessionPhase int32
+
+const (
+	// 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.
+	SessionPhase_SESSION_PHASE_UNSPECIFIED SessionPhase = 0
+	// Nothing is in flight; the session is waiting on the operator.
+	SessionPhase_SESSION_PHASE_IDLE SessionPhase = 1
+	// A turn is in flight — the model is being called, or its tool calls
+	// are running.
+	SessionPhase_SESSION_PHASE_GENERATING SessionPhase = 2
+)
+
+// Enum value maps for SessionPhase.
+var (
+	SessionPhase_name = map[int32]string{
+		0: "SESSION_PHASE_UNSPECIFIED",
+		1: "SESSION_PHASE_IDLE",
+		2: "SESSION_PHASE_GENERATING",
+	}
+	SessionPhase_value = map[string]int32{
+		"SESSION_PHASE_UNSPECIFIED": 0,
+		"SESSION_PHASE_IDLE":        1,
+		"SESSION_PHASE_GENERATING":  2,
+	}
+)
+
+func (x SessionPhase) Enum() *SessionPhase {
+	p := new(SessionPhase)
+	*p = x
+	return p
+}
+
+func (x SessionPhase) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (SessionPhase) Descriptor() protoreflect.EnumDescriptor {
+	return file_pluggableharness_session_v1_types_proto_enumTypes[1].Descriptor()
+}
+
+func (SessionPhase) Type() protoreflect.EnumType {
+	return &file_pluggableharness_session_v1_types_proto_enumTypes[1]
+}
+
+func (x SessionPhase) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use SessionPhase.Descriptor instead.
+func (SessionPhase) EnumDescriptor() ([]byte, []int) {
+	return file_pluggableharness_session_v1_types_proto_rawDescGZIP(), []int{1}
+}
+
 // SessionInfo is a session's shareable summary, mirroring
 // state-backend.md's session_meta row plus a cheap cost_ledger SUM. Used
 // by session lifecycle RPCs (CreateSession/AttachSession/ListSessions)
@@ -474,7 +535,16 @@ type SessionState struct {
 	// otherwise invisible: an operator sees only that answers got worse,
 	// with nothing in the UI to attribute it to. Absent means the vendor
 	// served what was asked for, or said nothing.
-	ActualModel   *string `protobuf:"bytes,13,opt,name=actual_model,json=actualModel,proto3,oneof" json:"actual_model,omitempty"`
+	ActualModel *string `protobuf:"bytes,13,opt,name=actual_model,json=actualModel,proto3,oneof" json:"actual_model,omitempty"`
+	// 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.
+	Phase         SessionPhase `protobuf:"varint,14,opt,name=phase,proto3,enum=pluggableharness.session.v1.SessionPhase" json:"phase,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -600,6 +670,13 @@ func (x *SessionState) GetActualModel() string {
 	return ""
 }
 
+func (x *SessionState) GetPhase() SessionPhase {
+	if x != nil {
+		return x.Phase
+	}
+	return SessionPhase_SESSION_PHASE_UNSPECIFIED
+}
+
 var File_pluggableharness_session_v1_types_proto protoreflect.FileDescriptor
 
 const file_pluggableharness_session_v1_types_proto_rawDesc = "" +
@@ -633,7 +710,7 @@ const file_pluggableharness_session_v1_types_proto_rawDesc = "" +
 	"\fContextState\x12\x1f\n" +
 	"\vused_tokens\x18\x01 \x01(\x03R\n" +
 	"usedTokens\x12#\n" +
-	"\rwindow_tokens\x18\x02 \x01(\x03R\fwindowTokens\"\xcf\x06\n" +
+	"\rwindow_tokens\x18\x02 \x01(\x03R\fwindowTokens\"\x90\a\n" +
 	"\fSessionState\x12<\n" +
 	"\x04info\x18\x01 \x01(\v2(.pluggableharness.session.v1.SessionInfoR\x04info\x12+\n" +
 	"\x11working_directory\x18\x02 \x01(\tR\x10workingDirectory\x12<\n" +
@@ -650,7 +727,8 @@ const file_pluggableharness_session_v1_types_proto_rawDesc = "" +
 	"\aaccount\x18\v \x01(\v2*.pluggableharness.model.v1.AccountSnapshotH\x04R\aaccount\x88\x01\x01\x12K\n" +
 	"\vvendor_cost\x18\f \x01(\v2%.pluggableharness.model.v1.VendorCostH\x05R\n" +
 	"vendorCost\x88\x01\x01\x12&\n" +
-	"\factual_model\x18\r \x01(\tH\x06R\vactualModel\x88\x01\x01B\x06\n" +
+	"\factual_model\x18\r \x01(\tH\x06R\vactualModel\x88\x01\x01\x12?\n" +
+	"\x05phase\x18\x0e \x01(\x0e2).pluggableharness.session.v1.SessionPhaseR\x05phaseB\x06\n" +
 	"\x04_vcsB\b\n" +
 	"\x06_modelB\x12\n" +
 	"\x10_thinking_effortB\n" +
@@ -668,7 +746,11 @@ const file_pluggableharness_session_v1_types_proto_rawDesc = "" +
 	"#SESSION_STATUS_ERROR_MAX_BUDGET_USD\x10\x04\x12'\n" +
 	"#SESSION_STATUS_ERROR_MAX_WALL_CLOCK\x10\x05\x12\x1c\n" +
 	"\x18SESSION_STATUS_CANCELLED\x10\x06\x12\x19\n" +
-	"\x15SESSION_STATUS_FAILED\x10\aBBZ@github.com/pluggableharness/agent/pkg/session/proto/v1;sessionv1b\x06proto3"
+	"\x15SESSION_STATUS_FAILED\x10\a*c\n" +
+	"\fSessionPhase\x12\x1d\n" +
+	"\x19SESSION_PHASE_UNSPECIFIED\x10\x00\x12\x16\n" +
+	"\x12SESSION_PHASE_IDLE\x10\x01\x12\x1c\n" +
+	"\x18SESSION_PHASE_GENERATING\x10\x02BBZ@github.com/pluggableharness/agent/pkg/session/proto/v1;sessionv1b\x06proto3"
 
 var (
 	file_pluggableharness_session_v1_types_proto_rawDescOnce sync.Once
@@ -682,38 +764,40 @@ func file_pluggableharness_session_v1_types_proto_rawDescGZIP() []byte {
 	return file_pluggableharness_session_v1_types_proto_rawDescData
 }
 
-var file_pluggableharness_session_v1_types_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_pluggableharness_session_v1_types_proto_enumTypes = make([]protoimpl.EnumInfo, 2)
 var file_pluggableharness_session_v1_types_proto_msgTypes = make([]protoimpl.MessageInfo, 5)
 var file_pluggableharness_session_v1_types_proto_goTypes = []any{
 	(SessionStatus)(0),            // 0: pluggableharness.session.v1.SessionStatus
-	(*SessionInfo)(nil),           // 1: pluggableharness.session.v1.SessionInfo
-	(*VcsState)(nil),              // 2: pluggableharness.session.v1.VcsState
-	(*ModelState)(nil),            // 3: pluggableharness.session.v1.ModelState
-	(*ContextState)(nil),          // 4: pluggableharness.session.v1.ContextState
-	(*SessionState)(nil),          // 5: pluggableharness.session.v1.SessionState
-	(*timestamppb.Timestamp)(nil), // 6: google.protobuf.Timestamp
-	(*durationpb.Duration)(nil),   // 7: google.protobuf.Duration
-	(*v1.RateLimitSnapshot)(nil),  // 8: pluggableharness.model.v1.RateLimitSnapshot
-	(*v1.AccountSnapshot)(nil),    // 9: pluggableharness.model.v1.AccountSnapshot
-	(*v1.VendorCost)(nil),         // 10: pluggableharness.model.v1.VendorCost
+	(SessionPhase)(0),             // 1: pluggableharness.session.v1.SessionPhase
+	(*SessionInfo)(nil),           // 2: pluggableharness.session.v1.SessionInfo
+	(*VcsState)(nil),              // 3: pluggableharness.session.v1.VcsState
+	(*ModelState)(nil),            // 4: pluggableharness.session.v1.ModelState
+	(*ContextState)(nil),          // 5: pluggableharness.session.v1.ContextState
+	(*SessionState)(nil),          // 6: pluggableharness.session.v1.SessionState
+	(*timestamppb.Timestamp)(nil), // 7: google.protobuf.Timestamp
+	(*durationpb.Duration)(nil),   // 8: google.protobuf.Duration
+	(*v1.RateLimitSnapshot)(nil),  // 9: pluggableharness.model.v1.RateLimitSnapshot
+	(*v1.AccountSnapshot)(nil),    // 10: pluggableharness.model.v1.AccountSnapshot
+	(*v1.VendorCost)(nil),         // 11: pluggableharness.model.v1.VendorCost
 }
 var file_pluggableharness_session_v1_types_proto_depIdxs = []int32{
 	0,  // 0: pluggableharness.session.v1.SessionInfo.status:type_name -> pluggableharness.session.v1.SessionStatus
-	6,  // 1: pluggableharness.session.v1.SessionInfo.started_at:type_name -> google.protobuf.Timestamp
-	6,  // 2: pluggableharness.session.v1.SessionInfo.ended_at:type_name -> google.protobuf.Timestamp
-	1,  // 3: pluggableharness.session.v1.SessionState.info:type_name -> pluggableharness.session.v1.SessionInfo
-	2,  // 4: pluggableharness.session.v1.SessionState.vcs:type_name -> pluggableharness.session.v1.VcsState
-	3,  // 5: pluggableharness.session.v1.SessionState.model:type_name -> pluggableharness.session.v1.ModelState
-	4,  // 6: pluggableharness.session.v1.SessionState.context:type_name -> pluggableharness.session.v1.ContextState
-	7,  // 7: pluggableharness.session.v1.SessionState.elapsed:type_name -> google.protobuf.Duration
-	8,  // 8: pluggableharness.session.v1.SessionState.quotas:type_name -> pluggableharness.model.v1.RateLimitSnapshot
-	9,  // 9: pluggableharness.session.v1.SessionState.account:type_name -> pluggableharness.model.v1.AccountSnapshot
-	10, // 10: pluggableharness.session.v1.SessionState.vendor_cost:type_name -> pluggableharness.model.v1.VendorCost
-	11, // [11:11] is the sub-list for method output_type
-	11, // [11:11] is the sub-list for method input_type
-	11, // [11:11] is the sub-list for extension type_name
-	11, // [11:11] is the sub-list for extension extendee
-	0,  // [0:11] is the sub-list for field type_name
+	7,  // 1: pluggableharness.session.v1.SessionInfo.started_at:type_name -> google.protobuf.Timestamp
+	7,  // 2: pluggableharness.session.v1.SessionInfo.ended_at:type_name -> google.protobuf.Timestamp
+	2,  // 3: pluggableharness.session.v1.SessionState.info:type_name -> pluggableharness.session.v1.SessionInfo
+	3,  // 4: pluggableharness.session.v1.SessionState.vcs:type_name -> pluggableharness.session.v1.VcsState
+	4,  // 5: pluggableharness.session.v1.SessionState.model:type_name -> pluggableharness.session.v1.ModelState
+	5,  // 6: pluggableharness.session.v1.SessionState.context:type_name -> pluggableharness.session.v1.ContextState
+	8,  // 7: pluggableharness.session.v1.SessionState.elapsed:type_name -> google.protobuf.Duration
+	9,  // 8: pluggableharness.session.v1.SessionState.quotas:type_name -> pluggableharness.model.v1.RateLimitSnapshot
+	10, // 9: pluggableharness.session.v1.SessionState.account:type_name -> pluggableharness.model.v1.AccountSnapshot
+	11, // 10: pluggableharness.session.v1.SessionState.vendor_cost:type_name -> pluggableharness.model.v1.VendorCost
+	1,  // 11: pluggableharness.session.v1.SessionState.phase:type_name -> pluggableharness.session.v1.SessionPhase
+	12, // [12:12] is the sub-list for method output_type
+	12, // [12:12] is the sub-list for method input_type
+	12, // [12:12] is the sub-list for extension type_name
+	12, // [12:12] is the sub-list for extension extendee
+	0,  // [0:12] is the sub-list for field type_name
 }
 
 func init() { file_pluggableharness_session_v1_types_proto_init() }
@@ -729,7 +813,7 @@ func file_pluggableharness_session_v1_types_proto_init() {
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_pluggableharness_session_v1_types_proto_rawDesc), len(file_pluggableharness_session_v1_types_proto_rawDesc)),
-			NumEnums:      1,
+			NumEnums:      2,
 			NumMessages:   5,
 			NumExtensions: 0,
 			NumServices:   0,