feat: forward rawRequest and disconnect AbortSignal into onCallGenkit…

[IzaakGough]

Summary

Fixes #1888

Expose the underlying callable raw request and a client-disconnect AbortSignal to Genkit actions invoked via https.onCallGenkit, so Genkit flows can read request-level details and reliably stop work when the client disconnects (especially for SSE streaming).

Changes

• Added two symbol keys in src/common/providers/https.ts:

CALLABLE_RAW_REQUEST = Symbol.for("firebase.callable.rawRequest")

CALLABLE_RESPONSE_SIGNAL = Symbol.for("firebase.callable.responseSignal")

• Exported those symbols from src/v2/providers/https.ts so callers can reference the same symbol instances.
• Updated onCallGenkit to populate the Genkit action context for both JSON and streaming callables:

context[CALLABLE_RAW_REQUEST] = req.rawRequest

context[CALLABLE_RESPONSE_SIGNAL] = res.signal

Why

• Some Genkit actions need access to request-level data not represented in the typed callable payload (headers, IP, etc.).
• Streaming callables need a robust way to cancel work on disconnect; res.signal provides a standard AbortSignal that aborts when the client disconnects.

Tests

Updated spec/v2/providers/https.spec.ts to verify:

• rawRequest and AbortSignal are forwarded into Genkit context for JSON requests.
• The same forwarding happens for SSE/streaming requests.
• The forwarded AbortSignal is aborted when the client disconnects (req.emit("close")), and the handler stops accordingly.

Usage

Inside a Genkit action invoked via https.onCallGenkit, read:

• opts.context[https.CALLABLE_RAW_REQUEST] (Express Request)
• opts.context[https.CALLABLE_RESPONSE_SIGNAL] (AbortSignal)

Verification

• Deployed the following function

import {
  onCallGenkit,
  CALLABLE_RAW_REQUEST,
  CALLABLE_RESPONSE_SIGNAL,
} from "firebase-functions/v2/https";
import { genkit, z } from "genkit";

const ai = genkit({});

function sleep(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

const flow = ai.defineFlow(
  {
    name: "demo",
    inputSchema: z.object({
      prompt: z.string(),
      totalSteps: z.number().optional(),
      stepDelayMs: z.number().optional(),
    }),
    streamSchema: z.object({
      step: z.number(),
      message: z.string(),
    }),
  },
  async (input, { sendChunk, context }) => {
    const rawRequest = context[CALLABLE_RAW_REQUEST];
    const signal = context[CALLABLE_RESPONSE_SIGNAL];

    console.log("hasRawRequest", rawRequest !== undefined);
    console.log("hasSignal", signal !== undefined);

    signal?.addEventListener("abort", () => {
      console.log("abort signal fired inside flow");
    });

    for (let step = 1; step <= (input.totalSteps ?? 30); step++) {
      if (signal?.aborted) {
        console.log(`stopping early at step=${step}`);
        return { aborted: true, step };
      }

      await sleep(input.stepDelayMs ?? 2000);

      if (signal?.aborted) {
        console.log(`stopping after silent work at step=${step}`);
        return { aborted: true, step };
      }

      sendChunk({ step, message: `completed step ${step}` });
    }

    return { done: true };
  }
);

export const issue1888 = onCallGenkit(flow);

• Ran the test script as in MRE
• Observed the following logs:

Before the fix

Before the fix running the MRE produces the following logs:

These continue until:

Known Limitation / Follow-Up

There is still an underlying timing/race issue around disconnect vs. in-flight work completion
It appears to be governed by upstream behavior (request/stream lifecycle and/or Genkit action execution), and isn’t something we can reliably fix within firebase-functions alone; addressing it likely requires changes in the upstream caller/runtime (or Genkit).

[gemini-code-assist]

Code Review

This pull request introduces CALLABLE_RAW_REQUEST and CALLABLE_RESPONSE_SIGNAL symbols to expose the underlying raw request and abort signal within the Genkit action context in onCallGenkit. Corresponding tests were added to ensure these values are correctly forwarded and that the abort signal functions as expected upon client disconnection. Feedback was provided to use optional chaining when accessing the response signal to avoid potential runtime errors in environments where the response object might be missing.

[gemini-code-assist]

The res parameter in the onCall handler is marked as optional in its type definition. Accessing res.signal directly could lead to a runtime error if the handler is invoked without a second argument (e.g., in certain unit testing scenarios). Using optional chaining res?.signal is safer and more idiomatic here.

      context[CALLABLE_RESPONSE_SIGNAL] = res?.signal;