feat(chat): realtime session-activity signals (Phase 3)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cleanslice/runtime",
-  "version": "0.26.0",
+  "version": "0.27.0",
   "license": "MIT",
   "type": "module",
   "bin": {

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
+}

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>): 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()
+  })
+})

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<string, Session> = 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<void> {
-    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<Event[]> {

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"

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<void> {
     await this.service.append(sessionId, event)
   }

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)

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<void> {
     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()
   }
 

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.

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

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>): void {
     this.handler = handler
   }