apunkt/rowboat
1
0
Fork 0
mirror of https://github.com/rowboatlabs/rowboat.git synced 2026-07-15 21:11:08 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

refactor(x): one delivery path for turn events — chat, channels, deltas

Browse source 
Completes the turn-spine unification. Chat's SessionChatStore now
consumes turns:events like every other surface, joining live turns by
file offset (drop covered, append contiguous, refetch on gap) instead
of blind-appending session-bus events. Text/reasoning deltas cross IPC
only for turns a window subscribed to (turns:subscribe/unsubscribe, a
per-webContents registry that also survives window teardown), so
headless pipeline chatter never reaches windows that aren't watching.
The channels bridge settles turns off the turn event bus instead of
filtering the session broadcast. With no turn-event consumers left,
SessionsImpl stops forwarding entirely (it drains its execution streams
and keeps outcome/index handling) and sessions:events shrinks to
index-changed entries — one channel for turn events, one for session
metadata.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Ramnique Singh 2026-07-09 15:47:21 +05:30
parent f916cb1d70
commit 876bc35e9e
15 changed files with 284 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

10
apps/x/packages/core/src/channels/bridge.test.ts
View file

@ -1,7 +1,7 @@
import { describe, expect, it, vi } from "vitest";
import type { SessionIndexEntry } from "@x/shared/dist/sessions.js";
import type { TurnStreamEvent } from "@x/shared/dist/turns.js";
import { EmitterSessionBus } from "../sessions/bus.js";
import { TurnEventHub } from "../turns/event-hub.js";
import { TurnInputError } from "../turns/api.js";
import type { ISessions } from "../sessions/api.js";
import { ChannelBridge, type ModelChoice } from "./bridge.js";
@ -50,7 +50,7 @@ function askEvent(turnId: string, question: string, options?: string[]): TurnStr
interface Harness {
bridge: ChannelBridge;
bus: EmitterSessionBus;
bus: TurnEventHub;
replies: string[];
reply: (text: string) => Promise<void>;
sessions: {
@ -72,9 +72,9 @@ const MODELS: ModelChoice[] = [
];
function harness(entries: SessionIndexEntry[] = []): Harness {
const bus = new EmitterSessionBus();
const bus = new TurnEventHub();
const publish = (turnId: string, event: TurnStreamEvent) =>
bus.publish({ kind: "turn-event", sessionId: "s1", turnId, event });
bus.publish({ turnId, sessionId: "s1", event });
const sessions = {
createSession: vi.fn(async () => "s1"),
sendMessage: vi.fn(async () => ({ turnId: "t1" })),
@ -86,7 +86,7 @@ function harness(entries: SessionIndexEntry[] = []): Harness {
const listModels = vi.fn(async () => MODELS);
const bridge = new ChannelBridge({
sessions: sessions as unknown as ISessions,
sessionBus: bus,
turnEventBus: bus,
listModels,
});
const replies: string[] = [];

10
apps/x/packages/core/src/channels/bridge.ts
View file

@ -4,7 +4,7 @@ import { assistantText, lastAssistantText } from "../agents/headless.js";
import { TurnInputError } from "../turns/api.js";
import { ASK_HUMAN_TOOL } from "../turns/bridges/real-agent-resolver.js";
import { TurnNotSettledError, type ISessions } from "../sessions/api.js";
import type { EmitterSessionBus } from "../sessions/bus.js";
import type { TurnEventHub } from "../turns/event-hub.js";
// Transport-agnostic command layer: inbound texts from a messaging channel
// are parsed into commands (list / resume / new / stop / status) or forwarded
@ -136,7 +136,7 @@ export class ChannelBridge {
constructor(
private readonly deps: {
sessions: ISessions;
sessionBus: EmitterSessionBus;
turnEventBus: TurnEventHub;
listModels: () => Promise<ModelChoice[]>;
},
) {}
@ -502,8 +502,10 @@ export class ChannelBridge {
const buffered: Array<{ turnId: string; settled: Settled }> = [];
let waiter: { turnId: string; resolve: (settled: Settled) => void } | null = null;
let cancelTimer: (() => void) | null = null;
const unsubscribe = this.deps.sessionBus.subscribe((event) => {
if (event.kind !== "turn-event") return;
// The turn event spine carries every turn's events (deltas included
// for subscribed windows, but settleOf ignores those); subscribeAll
// because the turnId is unknown until sendMessage returns.
const unsubscribe = this.deps.turnEventBus.subscribeAll((event) => {
const settled = settleOf(event.event);
if (!settled) return;
if (waiter) {

4
apps/x/packages/core/src/channels/service.ts
View file

@ -5,7 +5,7 @@ import type { ChannelsConfig, ChannelsStatus } from "@x/shared/dist/channels.js"
import container from "../di/container.js";
import { WorkDir } from "../config/config.js";
import type { ISessions } from "../sessions/api.js";
import type { EmitterSessionBus } from "../sessions/bus.js";
import type { TurnEventHub } from "../turns/event-hub.js";
import { isSignedIn } from "../account/account.js";
import { listGatewayModels } from "../models/gateway.js";
import { listOnboardingModels } from "../models/models-dev.js";
@ -98,7 +98,7 @@ function ensureBridge(): ChannelBridge {
if (!bridge) {
bridge = new ChannelBridge({
sessions: container.resolve<ISessions>("sessions"),
sessionBus: container.resolve<EmitterSessionBus>("sessionBus"),
turnEventBus: container.resolve<TurnEventHub>("turnEventBus"),
listModels: listBridgeModels,
});
}

18
apps/x/packages/core/src/sessions/sessions.test.ts
View file

@ -644,7 +644,10 @@ describe("stopTurn and resumeTurn", () => {
});
describe("event forwarding and index maintenance (13.6, 13.7)", () => {
it("forwards stream events to the bus tagged with sessionId in order", async () => {
it("drains the execution stream and publishes only index updates", async () => {
// Live turn delivery is the turn event bus's job (published by the
// runtime itself); the session bus carries index changes only. The
// stream must still be consumed so it never buffers until settle.
const { sessions, fake, bus } = makeSessions();
const streamed: TurnStreamEvent[] = [
{ type: "text_delta", turnId: "x", modelCallIndex: 0, delta: "he" },
@ -652,19 +655,12 @@ describe("event forwarding and index maintenance (13.6, 13.7)", () => {
];
fake.script = () => ({ events: streamed, outcome: completedOutcome() });
const sessionId = await sessions.createSession();
const { turnId } = await sessions.sendMessage(sessionId, user("go"), {
await sessions.sendMessage(sessionId, user("go"), {
agent: { agentId: "copilot" },
});
await flush();
const turnEvents = bus.events.filter((e) => e.kind === "turn-event");
expect(turnEvents).toEqual(
streamed.map((event) => ({
kind: "turn-event",
sessionId,
turnId,
event,
})),
);
expect(bus.events.length).toBeGreaterThan(0);
expect(bus.events.every((e) => e.kind === "index-changed")).toBe(true);
});
it("outcome settlement updates the index entry's latest turn status", async () => {

17
apps/x/packages/core/src/sessions/sessions.ts
View file

@ -343,9 +343,11 @@ export class SessionsImpl implements ISessions {
await execution.outcome;
}
// Every advance this layer initiates: forward its stream to the bus
// tagged with sessionId, keep the abort controller for stopTurn, and
// update the index entry when the outcome settles.
// Every advance this layer initiates: keep the abort controller for
// stopTurn and update the index entry when the outcome settles. Live
// event delivery is not this layer's job — the runtime publishes every
// event to the turn event bus; the execution stream is drained so an
// unconsumed HotStream never buffers events until settle.
private startTrackedAdvance(
sessionId: string | null,
turnId: string,
@ -368,14 +370,7 @@ export class SessionsImpl implements ISessions {
void (async () => {
try {
for await (const event of execution.events) {
if (sessionId !== null) {
this.sessionBus.publish({
kind: "turn-event",
sessionId,
turnId,
event,
});
}
void event;
}
} catch {
// Infrastructure failures surface through the outcome.