fix(remote-control): harden self-hosted session flows (#278)

Co-authored-by: chengzifeng <chengzifeng@meituan.com>

packages/remote-control-server/src/routes/v2/worker-events-stream.ts

@@ -1,6 +1,6 @@
 import { Hono } from "hono";
 import { sessionIngressAuth, acceptCliHeaders } from "../../auth/middleware";
-import { createSSEStream } from "../../transport/sse-writer";
+import { createWorkerEventStream } from "../../transport/sse-writer";
 import { getSession } from "../../services/session";
 
 const app = new Hono();
@@ -18,7 +18,7 @@ app.get("/:id/worker/events/stream", acceptCliHeaders, sessionIngressAuth, async
   const fromSeq = c.req.query("from_sequence_num");
   const fromSeqNum = fromSeq ? parseInt(fromSeq) : lastEventId ? parseInt(lastEventId) : 0;
 
-  return createSSEStream(c, sessionId, fromSeqNum);
+  return createWorkerEventStream(c, sessionId, fromSeqNum);
 });
 
 export default app;

packages/remote-control-server/src/routes/v2/worker-events.ts

@@ -1,32 +1,66 @@
 import { Hono } from "hono";
 import { sessionIngressAuth, acceptCliHeaders } from "../../auth/middleware";
 import { publishSessionEvent } from "../../services/transport";
-import { getSession, updateSessionStatus } from "../../services/session";
+import { getSession, touchSession, updateSessionStatus } from "../../services/session";
 
 const app = new Hono();
 
+function extractWorkerEvents(body: unknown): Array<Record<string, unknown>> {
+  if (!body || typeof body !== "object") {
+    return [];
+  }
+
+  const payload = body as Record<string, unknown>;
+  const rawEvents = Array.isArray(payload.events)
+    ? payload.events
+    : Array.isArray(body)
+      ? body
+      : [body];
+
+  return rawEvents
+    .filter((evt): evt is Record<string, unknown> => !!evt && typeof evt === "object")
+    .map((evt) => {
+      const wrappedPayload = evt.payload;
+      if (wrappedPayload && typeof wrappedPayload === "object" && !Array.isArray(wrappedPayload)) {
+        return wrappedPayload as Record<string, unknown>;
+      }
+      return evt;
+    });
+}
+
 /** POST /v1/code/sessions/:id/worker/events — Write events */
 app.post("/:id/worker/events", acceptCliHeaders, sessionIngressAuth, async (c) => {
   const sessionId = c.req.param("id")!;
+  if (!getSession(sessionId)) {
+    return c.json({ error: { type: "not_found", message: "Session not found" } }, 404);
+  }
   const body = await c.req.json();
 
-  const events = Array.isArray(body) ? body : [body];
+  const events = extractWorkerEvents(body);
   const published = [];
   for (const evt of events) {
-    const result = publishSessionEvent(sessionId, evt.type || "message", evt, "inbound");
+    const eventType = typeof evt.type === "string" ? evt.type : "message";
+    const result = publishSessionEvent(sessionId, eventType, evt, "inbound");
     published.push(result);
   }
 
+  touchSession(sessionId);
+
   return c.json({ status: "ok", count: published.length }, 200);
 });
 
 /** PUT /v1/code/sessions/:id/worker/state — Report worker state */
 app.put("/:id/worker/state", acceptCliHeaders, sessionIngressAuth, async (c) => {
   const sessionId = c.req.param("id")!;
+  if (!getSession(sessionId)) {
+    return c.json({ error: { type: "not_found", message: "Session not found" } }, 404);
+  }
   const body = await c.req.json();
 
   if (body.status) {
     updateSessionStatus(sessionId, body.status);
+  } else {
+    touchSession(sessionId);
   }
 
   return c.json({ status: "ok" }, 200);
@@ -34,12 +68,29 @@ app.put("/:id/worker/state", acceptCliHeaders, sessionIngressAuth, async (c) =>
 
 /** PUT /v1/code/sessions/:id/worker/external_metadata — Report worker metadata (no-op) */
 app.put("/:id/worker/external_metadata", acceptCliHeaders, sessionIngressAuth, async (c) => {
+  const sessionId = c.req.param("id")!;
+  if (!getSession(sessionId)) {
+    return c.json({ error: { type: "not_found", message: "Session not found" } }, 404);
+  }
   // TUI's CCRClient calls this for metadata reporting. Accept and discard.
   return c.json({ status: "ok" }, 200);
 });
 
+/** POST /v1/code/sessions/:id/worker/events/delivery — Batch delivery tracking (no-op) */
+app.post("/:id/worker/events/delivery", acceptCliHeaders, sessionIngressAuth, async (c) => {
+  const sessionId = c.req.param("id")!;
+  if (!getSession(sessionId)) {
+    return c.json({ error: { type: "not_found", message: "Session not found" } }, 404);
+  }
+  return c.json({ status: "ok" }, 200);
+});
+
 /** POST /v1/code/sessions/:id/worker/events/:eventId/delivery — Delivery tracking (no-op) */
 app.post("/:id/worker/events/:eventId/delivery", acceptCliHeaders, sessionIngressAuth, async (c) => {
+  const sessionId = c.req.param("id")!;
+  if (!getSession(sessionId)) {
+    return c.json({ error: { type: "not_found", message: "Session not found" } }, 404);
+  }
   // TUI's CCRClient reports event delivery status (received/processing/processed).
   // Accept and discard — event bus doesn't track per-event delivery.
   return c.json({ status: "ok" }, 200);

packages/remote-control-server/src/routes/v2/worker.ts

@@ -1,9 +1,75 @@
 import { Hono } from "hono";
-import { getSession, incrementEpoch } from "../../services/session";
-import { apiKeyAuth, acceptCliHeaders } from "../../auth/middleware";
+import { getSession, incrementEpoch, touchSession, updateSessionStatus } from "../../services/session";
+import { apiKeyAuth, acceptCliHeaders, sessionIngressAuth } from "../../auth/middleware";
+import { storeGetSessionWorker, storeUpsertSessionWorker } from "../../store";
 
 const app = new Hono();
 
+/** GET /v1/code/sessions/:id/worker — Read worker state */
+app.get("/:id/worker", acceptCliHeaders, sessionIngressAuth, async (c) => {
+  const sessionId = c.req.param("id")!;
+  const session = getSession(sessionId);
+  if (!session) {
+    return c.json({ error: { type: "not_found", message: "Session not found" } }, 404);
+  }
+
+  const worker = storeGetSessionWorker(sessionId);
+  return c.json({
+    worker: {
+      worker_status: worker?.workerStatus ?? session.status,
+      external_metadata: worker?.externalMetadata ?? null,
+      requires_action_details: worker?.requiresActionDetails ?? null,
+      last_heartbeat_at: worker?.lastHeartbeatAt?.toISOString() ?? null,
+    },
+  }, 200);
+});
+
+/** PUT /v1/code/sessions/:id/worker — Update worker state */
+app.put("/:id/worker", acceptCliHeaders, sessionIngressAuth, async (c) => {
+  const sessionId = c.req.param("id")!;
+  const session = getSession(sessionId);
+  if (!session) {
+    return c.json({ error: { type: "not_found", message: "Session not found" } }, 404);
+  }
+
+  const body = await c.req.json();
+  if (body.worker_status) {
+    updateSessionStatus(sessionId, body.worker_status);
+  } else {
+    touchSession(sessionId);
+  }
+
+  const worker = storeUpsertSessionWorker(sessionId, {
+    workerStatus: body.worker_status,
+    externalMetadata: body.external_metadata,
+    requiresActionDetails: body.requires_action_details,
+  });
+
+  return c.json({
+    status: "ok",
+    worker: {
+      worker_status: worker.workerStatus ?? session.status,
+      external_metadata: worker.externalMetadata,
+      requires_action_details: worker.requiresActionDetails,
+      last_heartbeat_at: worker.lastHeartbeatAt?.toISOString() ?? null,
+    },
+  }, 200);
+});
+
+/** POST /v1/code/sessions/:id/worker/heartbeat — Keep worker alive */
+app.post("/:id/worker/heartbeat", acceptCliHeaders, sessionIngressAuth, async (c) => {
+  const sessionId = c.req.param("id")!;
+  const session = getSession(sessionId);
+  if (!session) {
+    return c.json({ error: { type: "not_found", message: "Session not found" } }, 404);
+  }
+
+  const now = new Date();
+  storeUpsertSessionWorker(sessionId, { lastHeartbeatAt: now });
+  touchSession(sessionId);
+  return c.json({ status: "ok", last_heartbeat_at: now.toISOString() }, 200);
+});
+
 /** POST /v1/code/sessions/:id/worker/register — Register worker */
 app.post("/:id/worker/register", acceptCliHeaders, apiKeyAuth, async (c) => {
   const sessionId = c.req.param("id")!;