fix(stream-plugin): trust server-side avatar go-live readiness

rndrntwrk · rndrntwrk · commit 1faefeb86105 · 2026-03-20T07:54:44.000-05:00
diff --git a/src/actions/legacyCompat.test.ts b/src/actions/legacyCompat.test.ts
@@ -193,6 +193,160 @@ describe('legacyCompatibilityActions', () => {
     },
   );
 
+  it(
+    'trusts server-reported avatar ready phases from stream/start without a second hard failure gate',
+    { timeout: 10000 },
+    async () => {
+      setEnv('STREAM555_BASE_URL', 'https://stream.example');
+      setEnv('STREAM555_AGENT_TOKEN', 'static-token');
+
+      const stopCalls: string[] = [];
+      const service = {
+        getCurrentSessionId: () => 'session-1',
+        getBoundSessionId: () => 'session-1',
+        getConfig: () => ({
+          baseUrl: 'https://stream.example',
+          agentToken: 'static-token',
+          defaultSessionId: 'session-1',
+        }),
+        stopStream: async (sessionId?: string) => {
+          stopCalls.push(sessionId ?? '');
+          return { stopped: true, wasActive: true };
+        },
+      } as unknown as StreamControlService;
+      const runtime = {
+        getService: (name: string) => (name === 'stream555' ? service : undefined),
+      } as IAgentRuntime;
+
+      const requestedUrls: string[] = [];
+      globalThis.fetch = (async (input: unknown) => {
+        requestedUrls.push(String(input));
+        return buildJsonResponse(201, {
+          status: 'started',
+          phase: 'outputs_pending',
+          sessionId: 'session-1',
+          active: true,
+          cfSessionId: 'cf-session-1',
+          publisher: 'capture_service_rtmps',
+          statusReason: 'ingest connected, waiting for outputs: twitch(pending), kick(pending)',
+          requiredOutputsReady: false,
+          cloudflare: { isConnected: false, state: 'unknown' },
+          platforms: {
+            twitch: { enabled: true, status: 'idle' },
+            kick: { enabled: true, status: 'idle' },
+          },
+        });
+      }) as typeof fetch;
+
+      const callbackPayloads: Array<Record<string, unknown>> = [];
+      const action = findAction('STREAM555_GO_LIVE');
+      const ok = await action.handler(
+        runtime,
+        {},
+        undefined,
+        {
+          sessionId: 'session-1',
+          inputType: 'avatar',
+          layoutMode: 'camera-full',
+        },
+        (payload) => {
+          callbackPayloads.push(payload as Record<string, unknown>);
+        },
+      );
+
+      assert.equal(ok, true);
+      assert.equal(requestedUrls.length, 1);
+      assert.match(requestedUrls[0] ?? '', /\/stream\/start$/);
+      assert.equal(stopCalls.length, 0);
+
+      const lastPayload = callbackPayloads.at(-1) as {
+        content?: { success?: boolean; data?: Record<string, unknown> };
+      };
+      assert.equal(lastPayload.content?.success, true);
+      assert.equal(lastPayload.content?.data?.phase, 'outputs_pending');
+      assert.equal(lastPayload.content?.data?.cfSessionId, 'cf-session-1');
+      assert.equal(lastPayload.content?.data?.requiredOutputsReady, false);
+      assert.equal(lastPayload.content?.data?.statusReason, 'ingest connected, waiting for outputs: twitch(pending), kick(pending)');
+    },
+  );
+
+  it(
+    'accepts ready avatar phases observed during status polling even if Cloudflare lagged the boolean flag',
+    { timeout: 10000 },
+    async () => {
+      setEnv('STREAM555_BASE_URL', 'https://stream.example');
+      setEnv('STREAM555_AGENT_TOKEN', 'static-token');
+
+      const stopCalls: string[] = [];
+      const service = {
+        getCurrentSessionId: () => 'session-1',
+        getBoundSessionId: () => 'session-1',
+        getConfig: () => ({
+          baseUrl: 'https://stream.example',
+          agentToken: 'static-token',
+          defaultSessionId: 'session-1',
+        }),
+        stopStream: async (sessionId?: string) => {
+          stopCalls.push(sessionId ?? '');
+          return { stopped: true, wasActive: true };
+        },
+      } as unknown as StreamControlService;
+      const runtime = {
+        getService: (name: string) => (name === 'stream555' ? service : undefined),
+      } as IAgentRuntime;
+
+      const requestedUrls: string[] = [];
+      const responses = [
+        buildJsonResponse(201, { status: 'started', cfSessionId: 'cf-session-1' }),
+        buildJsonResponse(200, {
+          sessionId: 'session-1',
+          active: true,
+          phase: 'outputs_pending',
+          cfSessionId: 'cf-session-1',
+          publisher: 'capture_service_rtmps',
+          statusReason: 'ingest connected, waiting for outputs: twitch(pending)',
+          requiredOutputsReady: false,
+          cloudflare: { isConnected: false, state: 'unknown' },
+          platforms: { twitch: { enabled: true, status: 'idle' } },
+        }),
+      ];
+      globalThis.fetch = (async (input: unknown) => {
+        requestedUrls.push(String(input));
+        const next = responses.shift();
+        assert.ok(next, 'unexpected fetch');
+        return next;
+      }) as typeof fetch;
+
+      const callbackPayloads: Array<Record<string, unknown>> = [];
+      const action = findAction('STREAM555_GO_LIVE');
+      const ok = await action.handler(
+        runtime,
+        {},
+        undefined,
+        {
+          sessionId: 'session-1',
+          inputType: 'avatar',
+          layoutMode: 'camera-full',
+        },
+        (payload) => {
+          callbackPayloads.push(payload as Record<string, unknown>);
+        },
+      );
+
+      assert.equal(ok, true);
+      assert.equal(stopCalls.length, 0);
+      assert.match(requestedUrls[0] ?? '', /\/stream\/start$/);
+      assert.match(requestedUrls[1] ?? '', /\/stream\/status$/);
+
+      const lastPayload = callbackPayloads.at(-1) as {
+        content?: { success?: boolean; data?: Record<string, unknown> };
+      };
+      assert.equal(lastPayload.content?.success, true);
+      assert.equal(lastPayload.content?.data?.phase, 'outputs_pending');
+      assert.equal(lastPayload.content?.data?.requiredOutputsReady, false);
+    },
+  );
+
   it(
     'prefers fresh API-key exchange over a stale configured bearer for go-live',
     { timeout: 10000 },
diff --git a/src/actions/legacyCompat.ts b/src/actions/legacyCompat.ts
@@ -67,8 +67,12 @@ interface StreamReadinessSnapshot {
   active: boolean;
   jobId?: string;
   cfSessionId?: string;
+  phase?: string;
+  publisher?: string;
   cloudflareConnected: boolean;
   cloudflareState?: string;
+  requiredOutputsReady?: boolean;
+  statusReason?: string;
   platforms: Record<string, unknown>;
   raw?: JsonObject;
 }
@@ -588,13 +592,65 @@ async function fetchStreamReadinessSnapshot(
     active: Boolean(response.data?.active),
     jobId: getStringField(response.data, 'jobId'),
     cfSessionId: getStringField(response.data, 'cfSessionId'),
+    phase: getStringField(response.data, 'phase'),
+    publisher: getStringField(response.data, 'publisher'),
     cloudflareConnected: Boolean(cloudflare?.isConnected),
     cloudflareState: getStringField(cloudflare, 'state'),
+    requiredOutputsReady:
+      typeof response.data?.requiredOutputsReady === 'boolean'
+        ? response.data.requiredOutputsReady
+        : undefined,
+    statusReason: getStringField(response.data, 'statusReason'),
     platforms,
     raw: response.data,
   };
 }
 
+function extractReadinessSnapshot(
+  sessionId: string,
+  data: JsonObject | undefined,
+): StreamReadinessSnapshot | undefined {
+  if (!data) return undefined;
+  const cloudflare = getObject(data.cloudflare);
+  const platforms = getObject(data.platforms) ?? {};
+  return {
+    sessionId,
+    active: Boolean(data.active),
+    jobId: getStringField(data, 'jobId'),
+    cfSessionId: getStringField(data, 'cfSessionId'),
+    phase: getStringField(data, 'phase'),
+    publisher: getStringField(data, 'publisher'),
+    cloudflareConnected: Boolean(cloudflare?.isConnected),
+    cloudflareState: getStringField(cloudflare, 'state'),
+    requiredOutputsReady:
+      typeof data.requiredOutputsReady === 'boolean' ? data.requiredOutputsReady : undefined,
+    statusReason: getStringField(data, 'statusReason'),
+    platforms,
+    raw: data,
+  };
+}
+
+function isReadyPhase(phase: string | undefined): boolean {
+  const normalized = typeof phase === 'string' ? phase.trim().toLowerCase() : '';
+  return (
+    normalized === 'live' ||
+    normalized === 'ingest_connected' ||
+    normalized === 'outputs_pending'
+  );
+}
+
+function isStreamReadinessSatisfied(
+  snapshot: StreamReadinessSnapshot | undefined,
+): boolean {
+  if (!snapshot?.active || !snapshot.cfSessionId) {
+    return false;
+  }
+  if (snapshot.cloudflareConnected) {
+    return true;
+  }
+  return isReadyPhase(snapshot.phase);
+}
+
 async function waitForStreamReadiness(
   baseUrl: string,
   headers: Record<string, string>,
@@ -607,11 +663,7 @@ async function waitForStreamReadiness(
 
   while (Date.now() <= deadline) {
     lastSnapshot = await fetchStreamReadinessSnapshot(baseUrl, headers, sessionId);
-    if (
-      lastSnapshot.active &&
-      lastSnapshot.cfSessionId &&
-      lastSnapshot.cloudflareConnected
-    ) {
+    if (isStreamReadinessSatisfied(lastSnapshot)) {
       return { ready: true, lastSnapshot };
     }
     if (Date.now() >= deadline) break;
@@ -786,10 +838,15 @@ const goLiveAction: Action = {
         },
       );
       let initialSnapshot: StreamReadinessSnapshot | undefined;
+      let immediateReadySnapshot: StreamReadinessSnapshot | undefined;
       const responseCfSessionId = getStringField(startResponse.data, 'cfSessionId');
+      if (startResponse.ok) {
+        immediateReadySnapshot = extractReadinessSnapshot(sessionId, startResponse.data);
+      }
       if (!startResponse.ok) {
         if (startResponse.status === 409) {
           initialSnapshot = await fetchStreamReadinessSnapshot(baseUrl, headers, sessionId);
+          immediateReadySnapshot = initialSnapshot;
         }
         if (!(startResponse.status === 409 && (responseCfSessionId || initialSnapshot?.cfSessionId))) {
           throw new Error(
@@ -798,7 +855,9 @@ const goLiveAction: Action = {
         }
       }
 
-      const readiness = await waitForStreamReadiness(baseUrl, headers, sessionId);
+      const readiness = isStreamReadinessSatisfied(immediateReadySnapshot)
+        ? { ready: true, lastSnapshot: immediateReadySnapshot }
+        : await waitForStreamReadiness(baseUrl, headers, sessionId);
       if (!readiness.ready) {
         try {
           await service.stopStream(sessionId);
@@ -832,8 +891,17 @@ const goLiveAction: Action = {
           readiness.lastSnapshot?.cfSessionId ??
           responseCfSessionId ??
           initialSnapshot?.cfSessionId,
+        phase: readiness.lastSnapshot?.phase ?? getStringField(startResponse.data, 'phase'),
+        publisher: readiness.lastSnapshot?.publisher ?? getStringField(startResponse.data, 'publisher'),
         cloudflareConnected: readiness.lastSnapshot?.cloudflareConnected ?? false,
         cloudflareState: readiness.lastSnapshot?.cloudflareState,
+        requiredOutputsReady:
+          readiness.lastSnapshot?.requiredOutputsReady ??
+          (typeof startResponse.data?.requiredOutputsReady === 'boolean'
+            ? startResponse.data.requiredOutputsReady
+            : undefined),
+        statusReason:
+          readiness.lastSnapshot?.statusReason ?? getStringField(startResponse.data, 'statusReason'),
         destinationSync,
         platformStatuses: readiness.lastSnapshot?.platforms,
         streamStatus: readiness.lastSnapshot?.raw,
