Carry the persisted sequence on bus events, and SDK stream helpers - #18
Merged
Conversation
correlation_ids and thinking part_index landed on the wire without an SDK path to set them, so an author had to reach past Sink into modelv1.StreamEvent's oneof — exactly what Sink exists to prevent. Rather than widening StreamStart and ThinkingDeltaOn with parameters most vendors pass nil for, each gets a named sibling: StreamStartWith for secondary request handles, ThinkingDeltaOnPart for vendor-numbered reasoning parts. The simple forms delegate, so the common call site is unchanged. StreamStartWith copies the caller's map and drops empty keys, so a caller reusing its map cannot mutate an in-flight send.
A subscriber that also reads history has to merge two streams into one view, and only ReadEvents carried a sequence. Live bus events arrived with nothing to order by, so a frontend attaching to a session already in progress had no basis for placing a live message relative to its own backfill and sorted new content above replayed content. BusEvent gains sequence, set for the reserved kernel.event.* topics from the value sessionstate already had in hand at republish time. It stays zero for a plugin's own Publish, which has no persisted event behind it — a real distinction rather than a missing value. This does not make the bus ordered. It makes bus events orderable against the log, which is the property a merging subscriber actually needs. determinism.md already makes sequence the sole ordering authority; a transport that dropped it forced every subscriber to invent one, and the one that did invented arrival order. BusEvent.time is now documented as display-only on both the proto and the spec. It was already unusable for ordering and nothing said so.
Dependency Review✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.Scanned FilesNone |
There was a problem hiding this comment.
Pull request overview
This PR extends the kernel/frontend protocol surfaces to improve “honest” live rendering (orderable live bus events, richer state/delta signals) and fills SDK gaps in pkg/model so providers can set newer wire fields without touching protobuf oneofs directly.
Changes:
- Carry persisted
sequenceon republishedkernel.event.*bus events (and documentBusEvent.timeas display-only). - Add
pkg/modelstream helpers forcorrelation_idsandThinkingDelta.part_index/ thinking channels; update accumulator + tests to preserve these semantics. - Improve frontend fast-path/state fidelity (e.g.,
SessionState.phase,TokenDelta.kind, context window occupancy) and fixkernel.staterepublish to send aBusEventpayload shape.
Reviewed changes
Copilot reviewed 16 out of 18 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/session/proto/v1/types.pb.go | Regenerated session protobuf Go for SessionPhase + SessionState.phase. |
| pkg/model/stream.go | Adds StreamStartWith, ThinkingDeltaOn, ThinkingDeltaOnPart; keeps simple helpers delegating. |
| pkg/model/stream_internal_test.go | Extends sink variant test coverage for new helper methods and correlation-id behavior. |
| pkg/model/doc.go | Updates SDK documentation to reflect new optional-field helpers and interfaces. |
| pkg/kernel/proto/v1/events.pb.go | Regenerated kernel protobuf Go for BusEvent.sequence and TokenDelta.kind/DeltaKind. |
| internal/streamaccum/streamaccum.go | Treats ThinkingDelta.part_index as part of the coalescing key; merges StickyTurnToken. |
| internal/streamaccum/streamaccum_test.go | Adds tests for sticky turn token merging and part-index block splitting/coalescing. |
| internal/sessionstate/emit.go | Populates BusEvent.sequence when republishing persisted kernel.event.* events. |
| internal/sessionstate/emit_test.go | Asserts republished bus events carry the persisted sequence. |
| internal/session/run.go | Tracks “context window occupancy” from usage (input + cache_read) and stores session phase. |
| internal/session/handle.go | Publishes SessionState.phase, fixes kernel.state republish payload shape to *kernelv1.BusEvent, adjusts context used token selection. |
| internal/modelcall/modelcall.go | Replaces OnTextDelta with OnDelta (tagged with kernelv1.DeltaKind). |
| internal/modelcall/complete.go | Forwards both text and thinking deltas on the fast path, tagged via DeltaKind. |
| internal/kernel/turnstack.go | Publishes TokenDelta with the new kind field via the OnDelta callback. |
| docs/specifications/kernel-callbacks.md | Updates BusEvent spec to include sequence and clarifies time is display-only. |
| docs/specifications/event-bus.md | Documents BusEvent.sequence semantics for kernel.event.* republish and merge-ordering guidance. |
| api/pluggableharness/session/v1/types.proto | Adds SessionPhase enum and SessionState.phase. |
| api/pluggableharness/kernel/v1/events.proto | Adds BusEvent.sequence and TokenDelta.kind + DeltaKind enum. |
Files not reviewed (2)
- pkg/kernel/proto/v1/events.pb.go: Generated file
- pkg/session/proto/v1/types.pb.go: Generated file
Comment on lines
+106
to
+114
| // 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; | ||
| } |
Comment on lines
+190
to
+199
| // 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; | ||
| } |
Comment on lines
+134
to
149
| // 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() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What & why
Two changes that a frontend needs in order to render a session honestly, both additive.
BusEvent.sequenceA subscriber that also reads history has to merge two streams into one view, and only
ReadEventscarried a sequence. Live bus events arrived with nothing to order by, so a frontend attaching to a session already in progress had no basis for placing a live message relative to its own backfill and sorted new content above replayed content.BusEventgainssequence, set for the reservedkernel.event.*topics from the valuesessionstate.Livealready had in hand at republish time (EmitOutcome.Sequence). It stays zero for a plugin's ownPublish, which has no persisted event behind it — a real distinction rather than a missing value.This does not make the bus ordered. It makes bus events orderable against the log, which is the property a merging subscriber actually needs.
.claude/rules/determinism.mdalready makessequencethe sole ordering authority; a transport that dropped it forced every subscriber to invent one, and the one that did invented arrival order.BusEvent.timeis now documented as display-only on both the proto and the spec. It was already unusable for ordering and nothing said so.pkg/modelstream helperscorrelation_idsandThinkingDelta.part_indexlanded on the wire in #17 without an SDK path to set them, so an author had to reach pastSinkintomodelv1.StreamEvent's oneof — exactly whatSinkexists to prevent. Rather than widening the existing methods with parameters most vendors pass nil for, each gets a named sibling:StreamStartWithandThinkingDeltaOnPart. The simple forms delegate, so the common call site is unchanged.StreamStartWithcopies the caller's map and drops empty keys, so a caller reusing its map cannot mutate an in-flight send.Checklist
go mod tidyis a no-op,go build ./...,go vet ./...,gofmt -l -s .prints nothing,go test -race -covermode=atomic ./...,golangci-lint rundocs/specifications/document in this same PR (event-bus.md,kernel-callbacks.md#subscribe)pkg/*/proto/v1/—.protochanged inapi/, regenerated withbuf generatemkdocs build --strictpassesinternal/packages includeREADME.md+CLAUDE.md— none addedbin/Also verified:
gosec -exclude-generated0 issues,buf lint+buf formatclean.Notes for reviewers
Consumed by a real frontend, not hypothetical.
plugin-frontend-tuinow merges itsReadEventsbackfill with live bus traffic through one identity-keyed, sequence-ordered transcript, which is what surfaced the missing field. That work lives in its own repo and is not part of this PR.One follow-up is known and not done here: the TUI's bridge does not yet read the new field —
MapBusEventTopicstill passes0. That is a two-line change in the frontend repo, tracked there. The kernel side is complete and the field is populated; nothing in this PR depends on the consumer landing.Not fixed by this PR: a multi-line transcript rendering defect in the TUI. I verified the kernel persists correct content — message id set, thinking and text as separate blocks, line breaks intact — against a real session, so it is not a kernel-side fault. Root cause is still open in the frontend repo.