feat(host-service): nudge clients with chat:sessions-changed on /events

packages/chat-runtime/src/commands/commands/commands.test.ts

@@ -43,6 +43,25 @@ function createSession(runtime: ChatRuntime, scopeId = "workspace-1") {
 }
 
 describe("chat commands", () => {
+	test("closeSession reports the session's workspace once its agent stops", async () => {
+		const { harnesses } = fakeHarnessRegistry(SCRIPT);
+		const scopes: string[] = [];
+		const runtime = createTestRuntime({
+			harnesses,
+			onSessionChanged: ({ scopeId }) => {
+				scopes.push(scopeId);
+			},
+		});
+		const created = createSession(runtime, "workspace-2");
+		scopes.length = 0;
+
+		await runtime.commands.closeSession({ sessionId: created.sessionId });
+		await runtime.commands.closeSession({ sessionId: created.sessionId });
+
+		expect(scopes).toEqual(["workspace-2"]);
+		await runtime.dispose();
+	});
+
 	test("createSession opens a journal and starts the harness", async () => {
 		const { runtime, adapterCount } = newRuntime();
 		const created = createSession(runtime);

packages/chat-runtime/src/commands/commands/commands.ts

@@ -253,9 +253,14 @@ export function createCommands(options: CommandsOptions): ChatCommands {
 			});
 		},
 
-		closeSession(input) {
+		async closeSession(input) {
 			const parsed: CloseSessionInput = closeSessionInputSchema.parse(input);
-			return options.live.dispose(parsed.sessionId);
+			const wasLive = options.live.get(parsed.sessionId) !== null;
+			try {
+				await options.live.dispose(parsed.sessionId);
+			} finally {
+				if (wasLive) options.journal.announce(parsed.sessionId);
+			}
 		},
 
 		getSession(input) {

packages/chat-runtime/src/db/schema/schema.ts

@@ -1,3 +1,4 @@
+import type { SessionStatus } from "@superset/chat/protocol";
 import {
 	index,
 	integer,
@@ -30,7 +31,7 @@ export const chatSessionsLocal = sqliteTable(
 		harness: text().notNull(),
 		harnessSessionId: text("harness_session_id"),
 		epoch: text().notNull(),
-		status: text().notNull(),
+		status: text().$type<SessionStatus>().notNull(),
 		title: text(),
 		queuedCount: integer("queued_count").notNull().default(0),
 		updatedAt: integer("updated_at").notNull(),

packages/chat-runtime/src/index.ts

@@ -2,6 +2,7 @@ import type { ChatCommands } from "./commands";
 import { CommandDedupe, createCommands } from "./commands";
 import type { ChatDb, OpenChatDb } from "./db";
 import { createChatDb } from "./db";
+import type { ChatJournalOptions } from "./journal";
 import { ChatJournal } from "./journal";
 import { ChatSessionStore } from "./projection";
 import type { HarnessRegistry, LiveSessionObserver } from "./sessions";
@@ -28,6 +29,7 @@ export type ChatRuntimeOptions = {
 	schedule?: Schedule;
 	bootstrapLimit?: number;
 	dedupeCapacity?: number;
+	onSessionChanged?: ChatJournalOptions["onSessionChanged"];
 };
 
 export type ChatRuntime = {
@@ -50,7 +52,9 @@ export function createChatRuntime(options: ChatRuntimeOptions): ChatRuntime {
 		dataDir: options.dataDir,
 		migrationsFolder: options.migrationsFolder,
 	});
-	const journal = new ChatJournal(db);
+	const journal = new ChatJournal(db, {
+		onSessionChanged: options.onSessionChanged,
+	});
 	const sessions = new ChatSessionStore(db);
 	const subscriptions = new SubscriptionHub(db, {
 		schedule: options.schedule,

packages/chat-runtime/src/journal/epoch/epoch.ts

@@ -1,4 +1,5 @@
 import { randomUUID } from "node:crypto";
+import type { SessionStatus } from "@superset/chat/protocol";
 import type { ChatDb } from "../../db";
 import {
 	insertSessionRow,
@@ -12,7 +13,7 @@ export type ChatSessionInit = {
 	scopeId: string;
 	harness: string;
 	harnessSessionId?: string | null;
-	status?: string;
+	status?: SessionStatus;
 	title?: string | null;
 };
 

packages/chat-runtime/src/journal/index.ts

@@ -1,4 +1,8 @@
 export type { ChatSessionInit, OpenedEpoch } from "./epoch";
 export { mintEpoch, openEpoch } from "./epoch";
-export type { OpenedSession } from "./journal";
+export type {
+	ChatJournalOptions,
+	ChatSessionChange,
+	OpenedSession,
+} from "./journal";
 export { ChatJournal } from "./journal";

packages/chat-runtime/src/journal/journal/index.ts

@@ -1,2 +1,6 @@
-export type { OpenedSession } from "./journal";
+export type {
+	ChatJournalOptions,
+	ChatSessionChange,
+	OpenedSession,
+} from "./journal";
 export { ChatJournal } from "./journal";

packages/chat-runtime/src/journal/journal/journal.test.ts

@@ -12,6 +12,7 @@ import {
 	userMessage,
 } from "../../testing/fixtures";
 import { createBunChatDb, createTestRuntime } from "../../testing/testRuntime";
+import type { ChatSessionChange } from "./journal";
 
 const SESSION = "session-1";
 
@@ -106,6 +107,63 @@ describe("ChatJournal", () => {
 		});
 	});
 
+	test("reports changes to the session listing, not every append", () => {
+		const changes: ChatSessionChange[] = [];
+		const watched = createTestRuntime({
+			onSessionChanged: (change) => {
+				changes.push(change);
+			},
+		});
+		watched.journal.open(init());
+		watched.journal.append(SESSION, {
+			type: "session",
+			session: sessionState({ status: "starting" }),
+		});
+		watched.journal.append(SESSION, {
+			type: "session",
+			session: sessionState({ status: "running" }),
+		});
+		watched.journal.append(SESSION, {
+			type: "item",
+			item: agentMessage("a", "a"),
+			turnId: "turn-1",
+		});
+		watched.journal.append(SESSION, {
+			type: "session",
+			session: sessionState({ status: "running", title: "Renamed" }),
+		});
+		watched.journal.append(SESSION, {
+			type: "session",
+			session: sessionState({ status: "idle" }),
+		});
+		watched.journal.discard(SESSION);
+
+		expect(changes).toHaveLength(5);
+		for (const change of changes) {
+			expect(change).toEqual({
+				sessionId: SESSION,
+				scopeId: "workspace-1",
+				occurredAt: expect.any(Number),
+			});
+		}
+	});
+
+	test("a throwing session listener does not fail the append", () => {
+		const watched = createTestRuntime({
+			onSessionChanged: () => {
+				throw new Error("listener failed");
+			},
+		});
+		watched.journal.open(init());
+		const cursor = watched.journal.append(SESSION, {
+			type: "session",
+			session: sessionState({ status: "running" }),
+		});
+
+		expect(cursor.seq).toBe(1);
+		expect(watched.sessions.get(SESSION)?.status).toBe("running");
+	});
+
 	test("projects queued_count from queued user messages", () => {
 		runtime.journal.open(init());
 		runtime.journal.append(SESSION, {

packages/chat-runtime/src/journal/journal/journal.ts

@@ -2,6 +2,7 @@ import type {
 	Cursor,
 	DurableEnvelope,
 	DurableEvent,
+	SessionStatus,
 } from "@superset/chat/protocol";
 import { durableEventSchema } from "@superset/chat/protocol";
 import { eq } from "drizzle-orm";
@@ -23,29 +24,48 @@ export type OpenedSession = {
 	queuedCount: number;
 };
 
+export type ChatSessionChange = {
+	sessionId: string;
+	scopeId: string;
+	occurredAt: number;
+};
+
+export type ChatJournalOptions = {
+	onSessionChanged?: (change: ChatSessionChange) => void | Promise<void>;
+};
+
+function reportListenerFailure(error: unknown): void {
+	console.error("[chat-journal] onSessionChanged listener failed", error);
+}
+
 type SessionCache = {
+	scopeId: string;
 	epoch: string;
 	lastSeq: number;
 	queuedItemIds: Set<string>;
-	status: string;
+	status: SessionStatus;
 	title: string | null;
 };
 
 type NextProjection = {
-	status: string;
+	status: SessionStatus;
 	title: string | null;
 	queuedItemIds: Set<string>;
 };
 
 export class ChatJournal {
 	private readonly sessions = new Map<string, SessionCache>();
 
-	constructor(private readonly db: ChatDb) {}
+	constructor(
+		private readonly db: ChatDb,
+		private readonly options: ChatJournalOptions = {},
+	) {}
 
 	open(init: ChatSessionInit): OpenedSession {
-		const { epoch } = openEpoch(this.db, init);
+		const { epoch, minted } = openEpoch(this.db, init);
 		this.sessions.delete(init.sessionId);
 		const cache = this.cacheFor(init.sessionId, epoch);
+		if (minted) this.notify(init.sessionId, cache.scopeId, Date.now());
 		return {
 			sessionId: init.sessionId,
 			epoch: cache.epoch,
@@ -64,6 +84,10 @@ export class ChatJournal {
 		const seq = cache.lastSeq + 1;
 		const ts = Date.now();
 		const next = this.projectionFor(cache, parsed);
+		const listingChanged =
+			next.status !== cache.status ||
+			next.title !== cache.title ||
+			next.queuedItemIds.size !== cache.queuedItemIds.size;
 
 		this.db.transaction(() => {
 			this.db
@@ -88,6 +112,7 @@ export class ChatJournal {
 		cache.status = next.status;
 		cache.title = next.title;
 		cache.queuedItemIds = next.queuedItemIds;
+		if (listingChanged) this.notify(sessionId, cache.scopeId, ts);
 
 		return {
 			v: 1,
@@ -108,6 +133,7 @@ export class ChatJournal {
 	}
 
 	discard(sessionId: string): void {
+		const row = readSessionRow(this.db, sessionId);
 		this.sessions.delete(sessionId);
 		this.db.transaction(() => {
 			this.db
@@ -116,6 +142,22 @@ export class ChatJournal {
 				.run();
 			removeSessionRow(this.db, sessionId);
 		});
+		if (row) this.notify(sessionId, row.scopeId, Date.now());
+	}
+
+	announce(sessionId: string): void {
+		const row = readSessionRow(this.db, sessionId);
+		if (row) this.notify(sessionId, row.scopeId, Date.now());
+	}
+
+	private notify(sessionId: string, scopeId: string, occurredAt: number): void {
+		try {
+			void Promise.resolve(
+				this.options.onSessionChanged?.({ sessionId, scopeId, occurredAt }),
+			).catch(reportListenerFailure);
+		} catch (error) {
+			reportListenerFailure(error);
+		}
 	}
 
 	private projectionFor(
@@ -183,6 +225,7 @@ export class ChatJournal {
 		}
 
 		const cache: SessionCache = {
+			scopeId: row.scopeId,
 			epoch,
 			lastSeq,
 			queuedItemIds,

packages/chat-runtime/src/projection/projection.ts

@@ -1,11 +1,12 @@
+import type { SessionStatus } from "@superset/chat/protocol";
 import { desc, eq } from "drizzle-orm";
 import type { ChatDb, ChatSessionRow } from "../db";
 import { chatSessionsLocal } from "../db";
 
 export type SessionRowInsert = typeof chatSessionsLocal.$inferInsert;
 
 export type SessionProjection = {
-	status: string;
+	status: SessionStatus;
 	title: string | null;
 	queuedCount: number;
 	updatedAt: number;
@@ -44,7 +45,7 @@ export function resetSessionForEpoch(
 	db: ChatDb,
 	sessionId: string,
 	epoch: string,
-	status: string,
+	status: SessionStatus,
 	updatedAt: number,
 ): void {
 	db.update(chatSessionsLocal)

packages/host-service/src/app.ts

@@ -334,6 +334,11 @@ export function createApp(options: CreateAppOptions): CreateAppResult {
 		db,
 		dbPath: config.dbPath,
 		agents: createChatAgentBridge(chatAgentContext),
+		onSessionChanged: ({ scopeId, occurredAt }) =>
+			eventBus.broadcastChatSessionsChanged({
+				workspaceId: scopeId,
+				occurredAt,
+			}),
 	});
 
 	// Startup sweeps run in the background so they don't block server

packages/host-service/src/chat-v3/mount.ts

@@ -7,6 +7,7 @@ import type { DeltaChannel } from "@superset/chat/protocol";
 import { parseCursor } from "@superset/chat/protocol";
 import type {
 	ChatRuntime,
+	ChatSessionChange,
 	HarnessFactory,
 	HarnessRegistry,
 	WsSinkSocket,
@@ -113,6 +114,7 @@ export function createChatV3Mount(options: {
 	db: HostDb;
 	dbPath: string;
 	agents?: ChatAgentBridge;
+	onSessionChanged?: (change: ChatSessionChange) => void;
 }): ChatV3Mount {
 	let built: ChatRuntime | null = null;
 
@@ -123,6 +125,7 @@ export function createChatV3Mount(options: {
 			migrationsFolder: migrationsFolder(),
 			harnesses: harnessRegistry(options.db, options.agents),
 			observer: options.agents,
+			onSessionChanged: options.onSessionChanged,
 		});
 		return built;
 	};