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

feat(x): turn event spine — one bus for every turn's events

Browse source 
TurnRuntime now publishes every turn's events (durable ones tagged with
their 1-based file offset, deltas without) to an injected TurnEventHub,
regardless of who started the turn. Main forwards durable events to all
windows on one turns:events channel; the renderer's new useTurn(turnId)
hook joins live turns gap/duplicate-free via the offset protocol. The
sub-agent card drops its 1s polling for push updates, and the nine
per-channel window fan-out loops collapse into one broadcastToWindows.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Ramnique Singh 2026-07-09 11:42:22 +05:30
parent 61344effea
commit 12a7118445
19 changed files with 790 additions and 132 deletions
Download patch file Download diff file Expand all files Collapse all files

10
apps/x/packages/shared/src/ipc.ts
View file

@ -14,7 +14,7 @@ import {
TriggersSchema,
} from './background-task.js';
import { UserMessage, UserMessageContent } from './message.js';
import { RequestedAgent, type TurnEvent } from './turns.js';
import { RequestedAgent, type TurnBusEvent, type TurnEvent } from './turns.js';
import type { SessionBusEvent, SessionIndexEntry, SessionState } from './sessions.js';
import { RowboatApiConfig } from './rowboat-account.js';
import { ZListToolkitsResponse } from './composio.js';
@ -543,6 +543,14 @@ const ipcSchemas = {
req: z.custom<SessionBusEvent>(),
res: z.null(),
},
// Process-wide turn event spine: every turn's durable events (with file
// offsets), regardless of who started the turn — session chat, headless
// background/knowledge runners, spawned sub-agents. Deltas are not
// broadcast here in v1; session chat streams them via sessions:events.
'turns:events': {
req: z.custom<TurnBusEvent>(),
res: z.null(),
},
'services:events': {
req: ServiceEvent,
res: z.null(),

20
apps/x/packages/shared/src/turns.ts
View file

@ -470,6 +470,26 @@ export type TurnStreamEvent =
| TextDelta
| ReasoningDelta;
// One entry on the process-wide turn event bus: every event of every turn the
// runtime executes, tagged with its origin so consumers can subscribe without
// knowing who started the turn. `offset` is the 1-based line index of a
// durable event in the turn's JSONL file — a late subscriber can fetch the
// turn snapshot and discard bus events with offset <= snapshot length to join
// a live turn without gaps or duplicates. Deltas are not durable and carry no
// offset.
export interface TurnBusEvent {
turnId: string;
sessionId: string | null;
event: TurnStreamEvent;
offset?: number;
}
export function isDurableTurnEvent(
event: TurnStreamEvent,
): event is z.infer<typeof TurnEvent> {
return event.type !== "text_delta" && event.type !== "reasoning_delta";
}
// ---------------------------------------------------------------------------
// Derived turn state
// ---------------------------------------------------------------------------