feat(pour-completed): add route for pour completed events and Redis publishing

Author: TienYuWu

apps/server/src/app.ts

@@ -50,6 +50,7 @@ import { createV1CompletionsRoutes } from './routes/openai/v1'
 import { createProviderRoutes } from './routes/providers'
 import { createStripeRoutes } from './routes/stripe'
 import { createSystemMessageEventsRoute } from './routes/events/system-messages'
+import { createPourCompletedRoutes } from './routes/pour-completed'
 import { createSystemMessageRoutes } from './routes/system-message'
 import { createBillingMq } from './services/billing/billing-events'
 import { createBillingService } from './services/billing/billing-service'
@@ -233,6 +234,13 @@ export async function buildApp(deps: AppDeps) {
      */
     .route('/api/events/system-messages', createSystemMessageEventsRoute(deps.redis))
 
+    /**
+     * Pour-completed signal from main_task_node. Republished to the
+     * `robot:pour-completed` Redis channel so bt_supervisor's PostPour branch
+     * can drive the follow-up question + go_home timeout.
+     */
+    .route('/api/robot/pour-completed', createPourCompletedRoutes(deps.redis))
+
     /**
      * V1 routes for official provider.
      */

apps/server/src/routes/chat-ws/index.ts

@@ -45,6 +45,22 @@ export function createChatWsHandlers(
           const result = await chatService.pushMessages(userId, req!.chatId, req!.messages)
           const payload = await deliverPushedMessages(chatService, redis, userId, req!.chatId, result, ctx)
           metrics?.wsMessagesSent.add(payload.messages.length)
+
+          // Fan out user-role messages to ROS-side consumers (bt_supervisor) over
+          // Redis. Single-purpose channel: any subscriber treats this as "the
+          // human just said something." Used by the post-pour question wait.
+          // Best-effort; never blocks the RPC reply.
+          for (const m of req!.messages) {
+            if (m.role !== 'user')
+              continue
+            redis.publish('robot:user-response', JSON.stringify({
+              text: m.content,
+              userId,
+              chatId: req!.chatId,
+              at: new Date().toISOString(),
+            })).catch(err => log.withFields({ err }).warn('robot:user-response publish failed'))
+          }
+
           return { seq: result.seq }
         })
 

apps/server/src/routes/pour-completed.ts

@@ -0,0 +1,65 @@
+import type Redis from 'ioredis'
+
+import type { HonoEnv } from '../types/hono'
+
+import { useLogger } from '@guiiai/logg'
+import { Hono } from 'hono'
+import { object, optional, safeParse, string } from 'valibot'
+
+import { createBadRequestError } from '../utils/error'
+
+const logger = useLogger('pour-completed')
+
+const POUR_COMPLETED_CHANNEL = 'robot:pour-completed'
+
+/**
+ * Inbound payload for `POST /api/robot/pour-completed`.
+ *
+ * Posted by main_task_node when an `ExecuteDrinkTask` action returns success
+ * (excluding the `__go_home__` cleanup goal). Republished verbatim to the
+ * `robot:pour-completed` Redis channel so the BT supervisor can drive the
+ * post-pour follow-up branch.
+ */
+const PourCompletedSchema = object({
+  drink_type: string(),
+  table_id: optional(string()),
+})
+
+/**
+ * Build the `/api/robot/pour-completed` Hono sub-app.
+ *
+ * Use when:
+ * - Wiring routes in `app.ts`. Mount under `/api/robot/pour-completed`.
+ *
+ * Expects:
+ * - `redis` is the primary connected client; reused for the publish.
+ *
+ * Returns:
+ * - A Hono sub-app exposing `POST /`.
+ */
+export function createPourCompletedRoutes(redis: Redis) {
+  return new Hono<HonoEnv>()
+    .post('/', async (c) => {
+      const raw = await c.req.json().catch(() => null)
+      if (raw === null)
+        throw createBadRequestError('Body must be JSON', 'INVALID_REQUEST')
+
+      const parsed = safeParse(PourCompletedSchema, raw)
+      if (!parsed.success)
+        throw createBadRequestError('Invalid Request', 'INVALID_REQUEST', parsed.issues)
+
+      const payload = JSON.stringify({
+        drink_type: parsed.output.drink_type,
+        table_id: parsed.output.table_id ?? '',
+        at: new Date().toISOString(),
+      })
+
+      redis.publish(POUR_COMPLETED_CHANNEL, payload).catch((err) => {
+        logger.withError(err as Error).warn('Failed to publish robot:pour-completed')
+      })
+
+      logger.withFields({ ...parsed.output }).log('Pour completed event published')
+
+      return c.json({ success: true }, 201)
+    })
+}