feat(extension): emit session turn lifecycle events#23
Merged
Conversation
📝 WalkthroughSummary by CodeRabbit
WalkthroughAdds bounded assistant lifecycle events emitted during prompt/session/turn/entry flows, payload builders, dispatch helpers, runtime integration (Runtime.Prompt), reactive event-bus APIs, tests validating event ordering/streams, and docs/roadmap updates describing the event model. ChangesAssistant Lifecycle Events
Sequence Diagram(s)sequenceDiagram
participant Client
participant Runtime
participant EventBus
participant Extensions
Client->>Runtime: Prompt(request)
Runtime->>EventBus: dispatchLifecycle(input)
Runtime->>EventBus: dispatchLifecycle(prompt_prepare)
Runtime->>Runtime: resolveSession -> (session, eventName)
Runtime->>EventBus: dispatchLifecycle(session_start/load)
Runtime->>EventBus: dispatchMessageAppend (user)
Runtime->>Runtime: create promptTurnLifecycle
Runtime->>EventBus: dispatchTurnStartLifecycle
Runtime->>EventBus: dispatchMessageAppend (thinking/toolResult)
Runtime->>EventBus: dispatchMessageAppend (assistant)
Runtime->>Runtime: turnLifecycle.dispatchEnd -> dispatchTurnEndLifecycle
Runtime->>Extensions: optional DispatchLifecycle forwarding
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #23 +/- ##
==========================================
+ Coverage 56.93% 57.29% +0.36%
==========================================
Files 159 160 +1
Lines 15572 15758 +186
==========================================
+ Hits 8866 9029 +163
- Misses 5752 5767 +15
- Partials 954 962 +8
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
internal/assistant/runtime_lifecycle_test.go (2)
46-57: ⚡ Quick winAssert lifecycle ordering explicitly instead of only checking substrings.
Line 48-Line 57 validate presence but not sequence, so ordering regressions can slip through even if all event names exist.
Suggested update
import ( "context" "errors" + "strings" "testing" @@ output, err := manager.ExecuteCommand(context.Background(), "events", "") require.NoError(t, err) - assert.Contains(t, output, "input::") - assert.Contains(t, output, "prompt_prepare::") - assert.Contains(t, output, "session_start:"+response.SessionID+":") - assert.Contains(t, output, "before_agent_start:"+response.SessionID+":") - assert.Contains(t, output, "agent_start:"+response.SessionID+":") - assert.Contains(t, output, "turn_start:"+response.SessionID+":") - assert.Contains(t, output, "message_append:"+response.SessionID+":user") - assert.Contains(t, output, "message_append:"+response.SessionID+":assistant") - assert.Contains(t, output, "turn_end:"+response.SessionID+":") - assert.Contains(t, output, "agent_end:"+response.SessionID+":") + lines := strings.Split(strings.TrimSpace(output), "\n") + require.GreaterOrEqual(t, len(lines), 10) + assert.Equal(t, []string{ + "input::", + "prompt_prepare::", + "session_start:" + response.SessionID + ":", + "before_agent_start:" + response.SessionID + ":", + "agent_start:" + response.SessionID + ":", + "turn_start:" + response.SessionID + ":", + "message_append:" + response.SessionID + ":user", + "message_append:" + response.SessionID + ":assistant", + "turn_end:" + response.SessionID + ":", + "agent_end:" + response.SessionID + ":", + }, lines[:10])🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/assistant/runtime_lifecycle_test.go` around lines 46 - 57, The test currently only asserts presence of lifecycle event substrings from manager.ExecuteCommand output but not their order; modify the test in internal/assistant/runtime_lifecycle_test.go to assert ordering by locating each event occurrence (use response.SessionID with the exact event tokens like "input::", "prompt_prepare::", "session_start:"+response.SessionID+":", "before_agent_start:"+response.SessionID+":", "agent_start:"+response.SessionID+":", "turn_start:"+response.SessionID+":", "message_append:"+response.SessionID+":user", "message_append:"+response.SessionID+":assistant", "turn_end:"+response.SessionID+":", "agent_end:"+response.SessionID+":"), record their positions (e.g. with strings.Index or by splitting into lines and finding indices) and assert that each successive position is greater than the previous one so the sequence is enforced rather than only presence.
15-173: 🏗️ Heavy liftConsider consolidating these lifecycle behavior tests into table-driven cases.
These are core behavior tests with repeated setup/assert flow; table-driven structure would reduce duplication and make adding lifecycle scenarios safer.
As per coding guidelines,
**/*_test.go: Prefer table-driven tests for core behavior and regression tests for terminal rendering bugs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/assistant/runtime_lifecycle_test.go` around lines 15 - 173, Consolidate the five similar tests (TestRuntime_PromptEmitsSessionTurnLifecycleEvents, TestRuntime_PromptEmitsSessionLoadForExistingSession, TestRuntime_PromptEmitsTurnEndOnPromptError, TestRuntime_PromptEmitsSideEffectMessageAppendEvents, TestRuntime_PromptLifecycleIgnoresHandlerErrors) into a single table-driven test: create a slice of cases where each case holds a name, setup function (to produce the client and extension Lua string using newTestRuntimeWithManager and loadRuntimeExtension), the Prompt request (or modified SessionID), and the expected post-conditions to verify via manager.ExecuteCommand output or response (e.g., contains strings or equals); then iterate cases with t.Run, create runtime/manager inside the loop, call runtime.Prompt and manager.ExecuteCommand as needed, and perform the same assertions per-case. Keep existing helpers (newTestRuntimeWithManager, loadRuntimeExtension, manager.ExecuteCommand, runtime.Prompt) and preserve each test’s exact assertions by encoding them in the case definitions so behavior remains identical.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/assistant/runtime.go`:
- Around line 302-305: In resolveSession, wrap all repository errors returned
from runtime.sessions calls (e.g., runtime.sessions.GetSession, any
Create/Update/Ack calls referenced near variables loadedSession/found/err) using
the package's oops pattern — e.g., oops.In("assistant").Code(...).Wrapf(err,
"context message") — instead of returning raw err; ensure each returned error
includes an appropriate oops.Code and a descriptive Wrapf message that mentions
the operation (GetSession/CreateSession/UpdateSession/AckSession) so the
assistant-domain metadata is preserved.
---
Nitpick comments:
In `@internal/assistant/runtime_lifecycle_test.go`:
- Around line 46-57: The test currently only asserts presence of lifecycle event
substrings from manager.ExecuteCommand output but not their order; modify the
test in internal/assistant/runtime_lifecycle_test.go to assert ordering by
locating each event occurrence (use response.SessionID with the exact event
tokens like "input::", "prompt_prepare::",
"session_start:"+response.SessionID+":",
"before_agent_start:"+response.SessionID+":",
"agent_start:"+response.SessionID+":", "turn_start:"+response.SessionID+":",
"message_append:"+response.SessionID+":user",
"message_append:"+response.SessionID+":assistant",
"turn_end:"+response.SessionID+":", "agent_end:"+response.SessionID+":"), record
their positions (e.g. with strings.Index or by splitting into lines and finding
indices) and assert that each successive position is greater than the previous
one so the sequence is enforced rather than only presence.
- Around line 15-173: Consolidate the five similar tests
(TestRuntime_PromptEmitsSessionTurnLifecycleEvents,
TestRuntime_PromptEmitsSessionLoadForExistingSession,
TestRuntime_PromptEmitsTurnEndOnPromptError,
TestRuntime_PromptEmitsSideEffectMessageAppendEvents,
TestRuntime_PromptLifecycleIgnoresHandlerErrors) into a single table-driven
test: create a slice of cases where each case holds a name, setup function (to
produce the client and extension Lua string using newTestRuntimeWithManager and
loadRuntimeExtension), the Prompt request (or modified SessionID), and the
expected post-conditions to verify via manager.ExecuteCommand output or response
(e.g., contains strings or equals); then iterate cases with t.Run, create
runtime/manager inside the loop, call runtime.Prompt and manager.ExecuteCommand
as needed, and perform the same assertions per-case. Keep existing helpers
(newTestRuntimeWithManager, loadRuntimeExtension, manager.ExecuteCommand,
runtime.Prompt) and preserve each test’s exact assertions by encoding them in
the case definitions so behavior remains identical.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 89724e58-3547-4bba-b3d3-9f1138b892ec
📒 Files selected for processing (8)
docs/extension-api.mddocs/extension-roadmap.mdinternal/assistant/client.gointernal/assistant/lifecycle.gointernal/assistant/runtime.gointernal/assistant/runtime_lifecycle_test.gointernal/assistant/runtime_test.gointernal/assistant/usage_events.go
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
internal/event/bus_test.go (1)
77-133: ⚡ Quick winCollapse the new bus behavior cases into a table-driven test.
These three tests exercise the same subscription/emit/assert shape, so a single table over subscription strategy and expected envelopes would be easier to extend and aligns with the repo’s test style. As per coding guidelines,
**/*_test.go: Prefer table-driven tests for core behavior and regression tests for terminal rendering bugs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/event/bus_test.go` around lines 77 - 133, Collapse the three tests into a single table-driven test that iterates cases describing the subscription strategy and expected results: for each case include a name, a subscribe function (using event.NewBus().Stream().Subscribe, .Channel("agent").Subscribe, or bus.OnEnvelope) that sets up the collector (envelopes or calls), a sequence of bus.Emit calls, and the expected slice to assert; inside the loop create a fresh bus via event.NewBus(testLogger()), call the case's subscribe function to collect outputs (and defer Unsubscribe when applicable), perform the emits with bus.Emit, then assert the collected results match the case.expected (use require/ assert as in existing tests). Reference TestBus_StreamExposesReactiveEnvelopeStream, TestBus_ChannelReturnsFilteredReactiveStream and TestBus_OnEnvelopeReceivesAllChannels and the methods bus.Stream, bus.Channel, bus.OnEnvelope, and bus.Emit to locate the code to refactor.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/event/bus.go`:
- Around line 94-95: On and OnEnvelope build the observable outside of bus.lock
so Clear() can rotate bus.subject and cause subscriptions to attach to an
already-completed subject; fix by acquiring bus.lock before building the
observable and before calling subscribe so the observable is created against the
current bus.subject, or alternatively change subscribe to re-check/obtain the
current bus.subject under lock before registering; update the code paths in On,
OnEnvelope and the subscribe call (referencing functions/methods: On,
OnEnvelope, subscribe, Clear, bus.subject and Channel) to ensure the lock is
held while creating the observable and attaching the observer.
---
Nitpick comments:
In `@internal/event/bus_test.go`:
- Around line 77-133: Collapse the three tests into a single table-driven test
that iterates cases describing the subscription strategy and expected results:
for each case include a name, a subscribe function (using
event.NewBus().Stream().Subscribe, .Channel("agent").Subscribe, or
bus.OnEnvelope) that sets up the collector (envelopes or calls), a sequence of
bus.Emit calls, and the expected slice to assert; inside the loop create a fresh
bus via event.NewBus(testLogger()), call the case's subscribe function to
collect outputs (and defer Unsubscribe when applicable), perform the emits with
bus.Emit, then assert the collected results match the case.expected (use
require/ assert as in existing tests). Reference
TestBus_StreamExposesReactiveEnvelopeStream,
TestBus_ChannelReturnsFilteredReactiveStream and
TestBus_OnEnvelopeReceivesAllChannels and the methods bus.Stream, bus.Channel,
bus.OnEnvelope, and bus.Emit to locate the code to refactor.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 0b4e20b3-8e38-4c51-896d-c6ce32fe29d5
📒 Files selected for processing (8)
docs/extension-api.mddocs/extension-roadmap.mdinternal/assistant/runtime.gointernal/assistant/runtime_lifecycle_test.gointernal/assistant/runtime_test.gointernal/event/bus.gointernal/event/bus_test.gointernal/terminal/session_commands.go
✅ Files skipped from review due to trivial changes (2)
- docs/extension-api.md
- docs/extension-roadmap.md
🚧 Files skipped from review as they are similar to previous changes (3)
- internal/assistant/runtime_test.go
- internal/assistant/runtime_lifecycle_test.go
- internal/assistant/runtime.go
|
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/event/bus_test.go`:
- Line 172: Replace the unordered assertion with an ordered assertion: in the
test where assert.ElementsMatch(t, expected, got) is used (the emitted stream
sequence check), change it to assert.Equal(t, expected, got) so the test
verifies the exact ordering of the emitted events; locate the assertion call by
the symbol assert.ElementsMatch and update it to use assert.Equal with the same
t, expected, got parameters.
In `@internal/event/bus.go`:
- Around line 63-78: Stream() and Channel() currently return observables that
can be subscribed to after the bus lock is released, allowing Clear() to
replace/complete the subject and causing subscribers to bind to a completed
subject; fix by making Stream and Channel create observables whose Subscribe
action acquires bus.lock and attaches the subscription to the current
bus.subject while the lock is held (i.e., reuse the existing subscribe helper or
implement the same acquire-lock + subject.Subscribe semantics inside
Stream/Channel), and for Channel apply the filter after obtaining the
subscription to ensure the subscribe is atomic with respect to Clear()/subject
replacement.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: a742b4da-6b04-4989-9ff6-8a49d80f391d
📒 Files selected for processing (5)
internal/assistant/lifecycle.gointernal/assistant/runtime.gointernal/assistant/runtime_lifecycle_test.gointernal/event/bus.gointernal/event/bus_test.go
🚧 Files skipped from review as they are similar to previous changes (2)
- internal/assistant/lifecycle.go
- internal/assistant/runtime.go
| bus.Emit(context.Background(), envelope.Channel, envelope.Data) | ||
| } | ||
|
|
||
| assert.ElementsMatch(t, expected, got) |
There was a problem hiding this comment.
Use ordered assertion for emitted stream sequence.
At Line 172, assert.ElementsMatch can mask ordering regressions. These cases emit in a deterministic sequence, so assert.Equal is a stronger guard.
Suggested diff
- assert.ElementsMatch(t, expected, got)
+ assert.Equal(t, expected, got)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| assert.ElementsMatch(t, expected, got) | |
| assert.Equal(t, expected, got) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/event/bus_test.go` at line 172, Replace the unordered assertion with
an ordered assertion: in the test where assert.ElementsMatch(t, expected, got)
is used (the emitted stream sequence check), change it to assert.Equal(t,
expected, got) so the test verifies the exact ordering of the emitted events;
locate the assertion call by the symbol assert.ElementsMatch and update it to
use assert.Equal with the same t, expected, got parameters.
Comment on lines
+63
to
+78
| func (bus *Bus) Stream() ro.Observable[Envelope] { | ||
| bus.lock.Lock() | ||
| bus.nextID++ | ||
| subscriptionID := bus.nextID | ||
| subject := bus.subject | ||
| subscription := subject.Subscribe(bus.observer(channel, handler)) | ||
| bus.subscriptions[subscriptionID] = subscription | ||
| bus.lock.Unlock() | ||
| defer bus.lock.Unlock() | ||
|
|
||
| return func() { | ||
| bus.unsubscribe(subscriptionID) | ||
| } | ||
| return bus.subject.AsObservable() | ||
| } | ||
|
|
||
| // Channel returns a filtered view of Stream for one channel. | ||
| func (bus *Bus) Channel(channel string) ro.Observable[Envelope] { | ||
| return ro.Pipe1( | ||
| bus.Stream(), | ||
| ro.Filter(func(envelope Envelope) bool { | ||
| return envelope.Channel == channel | ||
| }), | ||
| ) | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Direct reactive subscriptions (potentially affected):"
rg -nP --type=go '\.(Stream|Channel)\([^)]*\)\.Subscribe\s*\(' -C2
echo
echo "Clear() call sites to evaluate concurrent rotation risk:"
rg -nP --type=go '\.Clear\s*\(' -C2Repository: omarluq/librecode
Length of output: 2257
🏁 Script executed:
cat -n internal/event/bus.go | head -100Repository: omarluq/librecode
Length of output: 3589
🏁 Script executed:
tail -100 internal/event/bus.goRepository: omarluq/librecode
Length of output: 2701
🏁 Script executed:
rg -A10 "ro\.NewPublishSubject" internal/event/bus.goRepository: omarluq/librecode
Length of output: 648
🏁 Script executed:
# Check test cases that might show the race condition or how Clear() is used with subscriptions
rg -B5 -A10 "Clear.*Stream|Stream.*Clear|Channel.*Clear" internal/event/bus_test.goRepository: omarluq/librecode
Length of output: 43
🏁 Script executed:
cat -n internal/event/bus_test.go | head -200Repository: omarluq/librecode
Length of output: 6060
🏁 Script executed:
# Search for any documentation or tests about PublishSubject behavior with completed subjects
web_search "samber/ro PublishSubject Complete behavior subscription"Repository: omarluq/librecode
Length of output: 112
🏁 Script executed:
# Check if there are any other files in the codebase using ro.Subject or similar patterns
rg "\.Complete\(\)" --type=goRepository: omarluq/librecode
Length of output: 107
🏁 Script executed:
# Check if ro package is a vendored dependency or external
fd -e go -path "*/ro/*" | head -20Repository: omarluq/librecode
Length of output: 230
🏁 Script executed:
cat go.mod | grep samberRepository: omarluq/librecode
Length of output: 449
🏁 Script executed:
# Check how Stream() and Channel() are actually used in the broader codebase
rg "\.Stream\(\)" --type=go -A2 | head -50Repository: omarluq/librecode
Length of output: 478
🏁 Script executed:
# Let's trace the exact control flow for Stream().Subscribe() pattern
rg -B3 -A8 "\.Stream\(\)\.Subscribe" internal/event/bus_test.goRepository: omarluq/librecode
Length of output: 441
🏁 Script executed:
# Let's verify the exact sequence by looking at ro.Pipe1 and understanding if Filter is lazy
# Check if there's any documentation in the codebase about reactive stream semantics
rg "ro\.Pipe" internal/event/bus.go -B2 -A5Repository: omarluq/librecode
Length of output: 540
🏁 Script executed:
# Let's verify if this is actually a real issue by checking the overall design
# Are there any concurrent tests that would catch this?
rg -B5 -A10 "t\.Parallel\(\)" internal/event/bus_test.go | head -40Repository: omarluq/librecode
Length of output: 935
Stream()/Channel() expose a race condition where subscriptions can bind to a completed subject.
When Stream() returns an observable (while holding the lock) and then Channel() applies filtering, the lock is released before .Subscribe() is called. If Clear() runs in that window, it completes the original subject and creates a new one. The subsequent subscription binds to the now-completed subject and silently misses all future events emitted to the new subject.
The On() and OnEnvelope() methods are safe because they use the subscribe() helper which holds the lock for the entire atomic operation of creating the observable and attaching the subscription. Direct calls to .Stream().Subscribe() and .Channel().Subscribe() (as shown in TestBus_ReactiveStreams) have this vulnerability.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/event/bus.go` around lines 63 - 78, Stream() and Channel() currently
return observables that can be subscribed to after the bus lock is released,
allowing Clear() to replace/complete the subject and causing subscribers to bind
to a completed subject; fix by making Stream and Channel create observables
whose Subscribe action acquires bus.lock and attaches the subscription to the
current bus.subject while the lock is held (i.e., reuse the existing subscribe
helper or implement the same acquire-lock + subject.Subscribe semantics inside
Stream/Channel), and for Channel apply the filter after obtaining the
subscription to ensure the subscribe is atomic with respect to Clear()/subject
replacement.
This was referenced May 19, 2026
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
2 participants
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.
Summary
Validation