diff --git a/package.json b/package.json index bf3be56..7b8af85 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@cleanslice/runtime", - "version": "0.26.0", + "version": "0.27.0", "license": "MIT", "type": "module", "bin": { diff --git a/src/slices/agent/session/domain/activity.ts b/src/slices/agent/session/domain/activity.ts new file mode 100644 index 0000000..74f8784 --- /dev/null +++ b/src/slices/agent/session/domain/activity.ts @@ -0,0 +1,23 @@ +// Lightweight per-message signal emitted from the single persistence funnel +// (SessionService.append) so the ranch chat index can update live, without +// waiting for the S3 reconcile interval. Carries metadata only — the message +// content still lives in the JSONL/S3. +export interface SessionActivity { + sessionKey: string // "{channel}:{externalUserId}" — the JSONL basename + channel: string + externalUserId: string + eventId: string // Event.id — dedup watermark on the ranch side + role: "user" | "assistant" + ts: number + preview: string // truncated message text for the list row +} + +/** + * Port for reporting session activity. No-op by default; the BridleRepository + * registers a transport that emits over the always-on agent↔hub socket when + * connected. Agents without a hub connection simply drop these — the S3 + * reconcile is the safety net. + */ +export interface IActivityReporter { + report(activity: SessionActivity): void +} diff --git a/src/slices/agent/session/domain/session.service.spec.ts b/src/slices/agent/session/domain/session.service.spec.ts new file mode 100644 index 0000000..fff2135 --- /dev/null +++ b/src/slices/agent/session/domain/session.service.spec.ts @@ -0,0 +1,72 @@ +import { describe, expect, test } from "bun:test" +import { SessionService } from "./session.service" +import type { ISessionGateway } from "./session.gateway" +import type { SessionActivity } from "./activity" +import type { Event } from "../../../setup/event" + +function noopGateway(): ISessionGateway { + return { + append: async () => {}, + read: async () => [], + rewrite: async () => {}, + } +} + +function evt(over: Partial): Event { + return { id: "e1", type: "user", ts: 1000, data: { text: "hi" }, ...over } +} + +function harness() { + const reports: SessionActivity[] = [] + const svc = new SessionService(noopGateway()) + svc.setActivityReporter({ report: (a) => reports.push(a) }) + return { svc, reports } +} + +describe("SessionService activity emit", () => { + test("emits for a real user turn", async () => { + const { svc, reports } = harness() + await svc.append("bridle:admin", evt({ type: "user", data: { text: "hello there" } })) + expect(reports).toHaveLength(1) + expect(reports[0]).toMatchObject({ + sessionKey: "bridle:admin", + channel: "bridle", + externalUserId: "admin", + role: "user", + preview: "hello there", + }) + }) + + test("emits for an assistant turn", async () => { + const { svc, reports } = harness() + await svc.append("telegram:12345", evt({ type: "assistant", data: { text: "hi" } })) + expect(reports).toHaveLength(1) + expect(reports[0]).toMatchObject({ channel: "telegram", externalUserId: "12345", role: "assistant" }) + }) + + test("skips tool_call / tool_result / summary events", async () => { + const { svc, reports } = harness() + await svc.append("bridle:admin", evt({ type: "tool_call", data: { name: "x" } })) + await svc.append("bridle:admin", evt({ type: "tool_result", data: { result: "y" } })) + await svc.append("bridle:admin", evt({ type: "summary", data: { text: "z" } })) + expect(reports).toHaveLength(0) + }) + + test("skips internal channel (cron/heartbeat)", async () => { + const { svc, reports } = harness() + await svc.append("internal:heartbeat", evt({ type: "assistant", data: { text: "HEARTBEAT_OK" } })) + expect(reports).toHaveLength(0) + }) + + test("skips synthetic transient events (continuation prompts / partials)", async () => { + const { svc, reports } = harness() + await svc.append("bridle:admin", evt({ type: "user", data: { text: "continue", transient: true } })) + await svc.append("bridle:admin", evt({ type: "assistant", data: { text: "part", transient: true } })) + expect(reports).toHaveLength(0) + }) + + test("does nothing when no reporter is wired", async () => { + const svc = new SessionService(noopGateway()) + await expect(svc.append("bridle:admin", evt({}))).resolves.toBeUndefined() + }) +}) diff --git a/src/slices/agent/session/domain/session.service.ts b/src/slices/agent/session/domain/session.service.ts index 62060e2..7b59daa 100644 --- a/src/slices/agent/session/domain/session.service.ts +++ b/src/slices/agent/session/domain/session.service.ts @@ -1,12 +1,21 @@ import type { ISessionGateway } from "./session.gateway" import type { Session } from "./session.types" import type { Event } from "../../../setup/event" +import type { IActivityReporter } from "./activity" + +const PREVIEW_MAX = 200 export class SessionService { private sessions: Map = new Map() + private reporter?: IActivityReporter constructor(private gateway: ISessionGateway) {} + /** Wire a transport for live session-activity signals (see IActivityReporter). */ + setActivityReporter(reporter: IActivityReporter): void { + this.reporter = reporter + } + getOrCreate(channelId: string, userId: string): Session { const id = `${channelId}:${userId}` if (this.sessions.has(id)) return this.sessions.get(id)! @@ -36,7 +45,35 @@ export class SessionService { } async append(sessionId: string, event: Event): Promise { - return this.gateway.append(sessionId, event) + await this.gateway.append(sessionId, event) + this.emitActivity(sessionId, event) + } + + // Fire a live activity signal for real user/assistant turns only. Skips + // internal sessions (cron/heartbeat) and synthetic loop-control events + // (continuation prompts + partial chunks tagged `data.transient`). + private emitActivity(sessionId: string, event: Event): void { + if (!this.reporter) return + if (event.type !== "user" && event.type !== "assistant") return + const colon = sessionId.indexOf(":") + const channel = colon >= 0 ? sessionId.slice(0, colon) : sessionId + if (channel === "internal") return + const data = event.data as { text?: string; transient?: boolean } | undefined + if (data?.transient === true) return + const text = typeof data?.text === "string" ? data.text : "" + try { + this.reporter.report({ + sessionKey: sessionId, + channel, + externalUserId: colon >= 0 ? sessionId.slice(colon + 1) : "", + eventId: event.id, + role: event.type, + ts: event.ts, + preview: text.slice(0, PREVIEW_MAX), + }) + } catch { + // Activity is best-effort telemetry — never let it break persistence. + } } async read(sessionId: string): Promise { diff --git a/src/slices/agent/session/index.ts b/src/slices/agent/session/index.ts index 6a89fc9..017fbaf 100644 --- a/src/slices/agent/session/index.ts +++ b/src/slices/agent/session/index.ts @@ -1,5 +1,6 @@ export * from "./domain/session.types" export * from "./domain/session.gateway" export * from "./domain/session.service" +export * from "./domain/activity" export * from "./data/session.gateway" export * from "./session.module" diff --git a/src/slices/agent/session/session.module.ts b/src/slices/agent/session/session.module.ts index d6d5f88..6812979 100644 --- a/src/slices/agent/session/session.module.ts +++ b/src/slices/agent/session/session.module.ts @@ -3,6 +3,7 @@ import type { ILlmGateway } from "../../setup/llm/domain/llm.gateway" import { SessionGateway } from "./data/session.gateway" import { SessionService } from "./domain/session.service" import { CompactionService } from "./domain/compaction.service" +import type { IActivityReporter } from "./domain/activity" export interface SessionConfig { compactionThreshold?: number @@ -32,6 +33,10 @@ export class SessionModule { this.service.touch(sessionId) } + setActivityReporter(reporter: IActivityReporter): void { + this.service.setActivityReporter(reporter) + } + async append(sessionId: string, event: Event): Promise { await this.service.append(sessionId, event) } diff --git a/src/slices/runtime/loop/domain/loop.service.ts b/src/slices/runtime/loop/domain/loop.service.ts index 3bfb19b..816e04e 100644 --- a/src/slices/runtime/loop/domain/loop.service.ts +++ b/src/slices/runtime/loop/domain/loop.service.ts @@ -155,7 +155,7 @@ export class LoopService { id: randomUUID(), type: "user", ts: Date.now(), - data: { text: CONTINUATION_PROMPT, from: ctx.from }, + data: { text: CONTINUATION_PROMPT, from: ctx.from, transient: true }, } await this.deps.session.append(sessionId, continueEvent) history.push(continueEvent) @@ -171,7 +171,7 @@ export class LoopService { id: randomUUID(), type: "assistant", ts: Date.now(), - data: { text: response.text }, + data: { text: response.text, transient: true }, } await this.deps.session.append(sessionId, partialEvent) history.push(partialEvent) @@ -181,7 +181,7 @@ export class LoopService { id: randomUUID(), type: "user", ts: Date.now(), - data: { text: buildAnchoredContinuationPrompt(lastSnippet), from: ctx.from }, + data: { text: buildAnchoredContinuationPrompt(lastSnippet), from: ctx.from, transient: true }, } await this.deps.session.append(sessionId, continueEvent) history.push(continueEvent) diff --git a/src/slices/runtime/runtime/runtime.module.ts b/src/slices/runtime/runtime/runtime.module.ts index 5c99339..3660ff8 100644 --- a/src/slices/runtime/runtime/runtime.module.ts +++ b/src/slices/runtime/runtime/runtime.module.ts @@ -300,6 +300,11 @@ export class AgentRuntime { /** Connect to messaging channels and start listening. */ private async connectChannels(): Promise { this.channel.onMessage(msg => this.handleMessage(msg)) + // Live chat-index signals: every persisted user/assistant turn is reported + // to the hub over the agent socket (no-op when bridle isn't connected). + this.session.setActivityReporter({ + report: activity => this.channel.reportSessionActivity(activity), + }) await this.channel.start() } diff --git a/src/slices/setup/channel/channel.module.ts b/src/slices/setup/channel/channel.module.ts index cf3e13c..c1c13af 100644 --- a/src/slices/setup/channel/channel.module.ts +++ b/src/slices/setup/channel/channel.module.ts @@ -3,6 +3,7 @@ import { ChannelService } from "./domain/channel.service" import { ChannelGateway } from "./data/channel.gateway" import type { ChannelConfig } from "./domain/channel.types" import type { BridleSyncHandler, IBridleDebugPayload } from "./data/repositories/bridle/bridle.repository" +import type { SessionActivity } from "../../agent/session/domain/activity" import { migrateLegacyChannelFiles, type ChannelFileType } from "./data/channelFiles" import { type ITelegramFile, @@ -122,6 +123,17 @@ export class ChannelModule { } } + /** + * Emit a live session-activity signal to the bridle hub (Phase 3 realtime + * chat index). No-op when bridle isn't configured / connected. + */ + reportSessionActivity(activity: SessionActivity): void { + const bridle = this.service.get("bridle") + if (bridle instanceof ChannelGateway) { + bridle.reportActivity(activity) + } + } + /** * Whether the hub has pushed debug=true to this agent. Drives the loop's * emission gate alongside the BRIDLE_DEBUG env override. diff --git a/src/slices/setup/channel/data/channel.gateway.ts b/src/slices/setup/channel/data/channel.gateway.ts index 759f87e..ea62e49 100644 --- a/src/slices/setup/channel/data/channel.gateway.ts +++ b/src/slices/setup/channel/data/channel.gateway.ts @@ -1,6 +1,7 @@ import type { IChannelGateway } from "../domain/channel.gateway" import type { ChannelConfig, IChannelGroup, Message, MessagePart } from "../domain/channel.types" import { isSilentReply } from "../../../agent/agent/domain/silentReply" +import type { SessionActivity } from "../../../agent/session/domain/activity" import { TelegramRepository } from "./repositories/telegram/telegram.repository" import { SlackRepository } from "./repositories/slack/slack.repository" import { BridleRepository, type BridleSyncHandler, type IBridleDebugPayload } from "./repositories/bridle/bridle.repository" @@ -77,6 +78,13 @@ export class ChannelGateway implements IChannelGateway { } } + /** Emit a live session-activity signal — only effective on a bridle channel. */ + reportActivity(activity: SessionActivity): void { + if (this.repository instanceof BridleRepository) { + this.repository.reportActivity(activity) + } + } + /** Hub-pushed debug flag — false unless this is a bridle channel and hub set it. */ isDebugEnabled(): boolean { return this.repository instanceof BridleRepository diff --git a/src/slices/setup/channel/data/repositories/bridle/bridle.repository.ts b/src/slices/setup/channel/data/repositories/bridle/bridle.repository.ts index 3532385..f338fb6 100644 --- a/src/slices/setup/channel/data/repositories/bridle/bridle.repository.ts +++ b/src/slices/setup/channel/data/repositories/bridle/bridle.repository.ts @@ -8,6 +8,7 @@ import { type IMessageUiSubmitPart, } from "../../../domain/channel.types" import { isSilentReply, isSilentReplyPrefix } from "../../../../../agent/agent/domain/silentReply" +import type { SessionActivity } from "../../../../../agent/session/domain/activity" import { randomUUID } from "crypto" import { io, type Socket } from "socket.io-client" import { createLogger } from "../../../../logger" @@ -238,6 +239,16 @@ export class BridleRepository implements IChannelGateway { return this.debugEnabled } + /** + * Emit a live session-activity signal to the hub (Phase 3 realtime chat + * index). Best-effort: silent no-op when the socket is offline — the ranch + * S3 reconcile is the safety net. + */ + reportActivity(activity: SessionActivity): void { + if (!this.socket?.connected) return + this.socket.emit("session_activity", activity) + } + onMessage(handler: (msg: Message) => Promise): void { this.handler = handler }