feat(functions-compiler): carry the invocation-logs response channel

packages/functions-compiler/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@base44/functions-compiler",
-  "version": "0.1.0",
+  "version": "0.1.1",
   "description": "Production compiler for Base44 backend functions — turns function sources into a single Cloudflare Workers module.",
   "license": "MIT",
   "publishConfig": {

packages/functions-compiler/src/invocation-logs.ts

@@ -0,0 +1,55 @@
+export const CAPTURE_LOGS_HEADER = "X-B44-Capture-Logs";
+export const INVOCATION_LOGS_HEADER = "X-B44-Invocation-Logs";
+
+// Budgets are in escaped bytes and stay under the 8 KB the backend accepts for
+// this header (INVOCATION_LOGS_MAX_HEADER_BYTES), with room for the function's
+// own headers.
+export const MAX_CAPTURED_LINES = 40;
+const MAX_CAPTURED_BYTES = 6144;
+const LINE_ENVELOPE_BYTES = 30;
+const RESPONSE_HEADER_BUDGET = 7168;
+const HEADER_LINE_OVERHEAD = 4;
+
+// workerd's headers.set throws on any code unit above 0xff, which would silently
+// cost the request its logs; escaped text also makes the byte budget exact.
+export const ASCII_ESCAPE =
+  "const _b44Ascii = (s) => s.replace(/[\\u007f-\\uffff]/g, (c) => '\\\\u' + c.charCodeAt(0).toString(16).padStart(4, '0'));";
+
+export const CAPTURE_PATCH = [
+  "const _b44Capture = (c, lvl, msg) => {",
+  "  if (!c.invocationLogs) return;",
+  `  const cost = _b44Ascii(JSON.stringify(msg)).length + ${LINE_ENVELOPE_BYTES};`,
+  `  if (c.invocationLogs.length >= ${MAX_CAPTURED_LINES} || c.invocationLogsBytes + cost > ${MAX_CAPTURED_BYTES}) { c.invocationLogsDropped += 1; return; }`,
+  "  c.invocationLogsBytes += cost;",
+  "  c.invocationLogs.push({ level: lvl, message: msg });",
+  "};",
+].join("\n");
+
+export const INVOCATION_LOGS_PATCH = [
+  `const _b44CaptureStore = (request) => request.headers.get(${JSON.stringify(CAPTURE_LOGS_HEADER)}) === '1' ? { invocationLogs: [], invocationLogsBytes: 0, invocationLogsDropped: 0 } : {};`,
+  "const _b44AttachInvocationLogs = (res) => {",
+  "  const c = _b44Context();",
+  "  if (!c || !c.invocationLogs) return res;",
+  "  if (!(res instanceof Response) || res.status === 101 || res.webSocket) return res;",
+  `  let existingBytes = ${INVOCATION_LOGS_HEADER.length + HEADER_LINE_OVERHEAD};`,
+  `  res.headers.forEach((v, k) => { existingBytes += k.length + v.length + ${HEADER_LINE_OVERHEAD}; });`,
+  "  let payload = _b44Ascii(JSON.stringify({ lines: c.invocationLogs, dropped: c.invocationLogsDropped }));",
+  `  if (existingBytes + payload.length > ${RESPONSE_HEADER_BUDGET}) payload = JSON.stringify({ lines: [], dropped: c.invocationLogs.length + c.invocationLogsDropped });`,
+  `  if (existingBytes + payload.length > ${RESPONSE_HEADER_BUDGET}) return res;`,
+  "  let wrapped;",
+  "  try { wrapped = new Response(res.body, res); } catch (e) { return res; }",
+  `  try { wrapped.headers.set(${JSON.stringify(INVOCATION_LOGS_HEADER)}, payload); } catch (e) {}`,
+  "  return wrapped;",
+  "};",
+  // Same body as the dispatcher's user-exception answer (apps-dispatcher errors.ts:55);
+  // the dispatcher strips X-Base44-Cf-Error from user responses (run.ts:73), so this
+  // streams as a plain 500 without the dispatch_error metric. Errors the dispatcher
+  // classifies by message (errors.ts:41-45) rethrow so their 429 stands.
+  "const _b44CrashResponse = (e) => {",
+  "  const c = _b44Context();",
+  "  if (!c || !c.invocationLogs) return null;",
+  "  const m = String(e instanceof Error ? e.message : e).toLowerCase();",
+  "  if (m.includes('cpu time limit') || m.includes('too many subrequests')) return null;",
+  "  return _b44AttachInvocationLogs(new Response(JSON.stringify({ error: 'user-exception', detail: 'user worker threw an exception' }), { status: 500, headers: { 'Content-Type': 'application/json' } }));",
+  "};",
+].join("\n");

packages/functions-compiler/src/worker-entry.ts

@@ -9,6 +9,11 @@ import { readFileSync } from "node:fs";
 import type { AppFunctionInput } from "./contracts.js";
 import { DenoCompatError } from "./errors.js";
 import { RUNTIME_CONTEXT_SPECIFIER } from "./esbuild/runtime-context-virtual.js";
+import {
+  ASCII_ESCAPE,
+  CAPTURE_PATCH,
+  INVOCATION_LOGS_PATCH,
+} from "./invocation-logs.js";
 import { TELEMETRY_PATCH, TELEMETRY_STORE_FIELDS } from "./telemetry.js";
 
 // Pre-built by scripts/build-shim.ts; regenerate it after changing the shim.
@@ -75,7 +80,9 @@ export const CONSOLE_PATCH = [
   // by per-app bundles, where one script serves every function and log queries
   // need per-function attribution. `secrets` and `workerEnv` reference the
   // request's authoritative Worker env binding.
-  "const _b44Wrap = (lvl, a) => { const _c = _b44Context() ?? {}; _b44Orig({ _b44_env: _c.env ?? 'preview', ...(_c.fn ? { _b44_function: _c.fn } : {}), level: lvl, message: _b44Fmt(a) }); };",
+  ASCII_ESCAPE,
+  CAPTURE_PATCH,
+  "const _b44Wrap = (lvl, a) => { const _c = _b44Context() ?? {}; const _m = _b44Fmt(a); _b44Capture(_c, lvl, _m); _b44Orig({ _b44_env: _c.env ?? 'preview', ...(_c.fn ? { _b44_function: _c.fn } : {}), level: lvl, message: _m }); };",
   "console.log = (...a) => _b44Wrap('info', a);",
   "console.info = (...a) => _b44Wrap('info', a);",
   "console.warn = (...a) => _b44Wrap('warn', a);",
@@ -268,9 +275,11 @@ function buildAppEntrySource(
   const moduleImports = functionModules
     .map((file) => `import "./${file}";`)
     .join("\n");
-  const handlerExpr = telemetry
-    ? "_b44AttachTelemetry(await handler(request, info))"
-    : "await handler(request, info)";
+  const handlerExpr = `_b44AttachInvocationLogs(${
+    telemetry
+      ? "_b44AttachTelemetry(await handler(request, info))"
+      : "await handler(request, info)"
+  })`;
   const returnExpr = runtimeSecrets
     ? `withoutActivationSignal(${handlerExpr})`
     : handlerExpr;
@@ -281,6 +290,7 @@ import { installStaticEgressFetch, resolveHandler } from "./${SHIM_FILENAME}";${
 ${moduleImports}
 
 ${CONSOLE_PATCH}
+${INVOCATION_LOGS_PATCH}
 // Static egress reads workerEnv from the active request store. Install it
 // before telemetry so telemetry remains the outermost fetch wrapper.
 installStaticEgressFetch();
@@ -290,30 +300,30 @@ export default {
   async fetch(request, env, ctx) {
     const _b44Env = (request.headers.get('base44-functions-version') ?? '') === 'prod' ? 'prod' : 'preview';
     const functionName = request.headers.get("Base44-Function-Name");
-    return _b44Run({ env: _b44Env, fn: functionName ?? '', secrets: ${runtimeSecrets ? "process.env" : "env"}, workerEnv: env, waitUntil: (p) => ctx.waitUntil(p)${telemetry ? `, ${TELEMETRY_STORE_FIELDS}` : ""} }, async () => {
+    return _b44Run({ env: _b44Env, fn: functionName ?? '', secrets: ${runtimeSecrets ? "process.env" : "env"}, workerEnv: env, waitUntil: (p) => ctx.waitUntil(p), ..._b44CaptureStore(request)${telemetry ? `, ${TELEMETRY_STORE_FIELDS}` : ""} }, async () => {
 ${runtimeSecrets ? ACTIVATION_GATE : ""}      // Each early return below logs through the patch first: per-function log
       // queries on per-app scripts keep only stamped lines, so a bare return
       // would leave the failing invocation with no trace in its own logs.
       const pendingHandler = functionName ? resolveHandler(functionName) : undefined;
       if (pendingHandler === undefined) {
         const message = \`No function registered for "\${functionName ?? ""}"\`;
         console.error(message);
-        return new Response(message, { status: 404 });
+        return _b44AttachInvocationLogs(new Response(message, { status: 404 }));
       }
       let handler;
       try {
         handler = await pendingHandler;
       } catch (e) {
         console.error(\`Function "\${functionName}" failed to initialize:\`, e);
-        return new Response(
+        return _b44AttachInvocationLogs(new Response(
           \`Function "\${functionName}" failed to initialize: \${e instanceof Error ? e.message : String(e)}\`,
           { status: 500 },
-        );
+        ));
       }
       if (handler === null) {
         const message = \`Function "\${functionName}" must export default a request handler or call Deno.serve()\`;
         console.error(message);
-        return new Response(message, { status: 503 });
+        return _b44AttachInvocationLogs(new Response(message, { status: 503 }));
       }
       // Real client IP is in the "cf-connecting-ip" header, not this placeholder.
       const info = {
@@ -326,11 +336,14 @@ ${runtimeSecrets ? ACTIVATION_GATE : ""}      // Each early return below logs th
       // stamped lines (see log_query.event_matches_function). Known gap:
       // exceptions thrown while a response body streams happen after this
       // frame returns and cannot be stamped — those crash events are
-      // dropped from per-function views.
+      // dropped from per-function views, and lines logged after the response
+      // starts streaming are past the invocation-logs header too.
       try {
         return ${returnExpr};
       } catch (e) {
         console.error(e);
+        const crash = _b44CrashResponse(e);
+        if (crash) return crash;
         throw e;
       }
     });

packages/functions-compiler/test/invocation-logs.test.ts

@@ -0,0 +1,227 @@
+/**
+ * The invoke response carries this invocation's own console lines — when asked.
+ *
+ * Runs the real CONSOLE_PATCH + INVOCATION_LOGS_PATCH in workerd (Miniflare),
+ * assembled into a minimal module the same way the console-patch tests do —
+ * no npm resolution, so these stay fast and offline.
+ */
+
+import { describe, expect, it } from "vitest";
+import { Miniflare } from "miniflare";
+
+import {
+  CAPTURE_LOGS_HEADER,
+  INVOCATION_LOGS_HEADER,
+  INVOCATION_LOGS_PATCH,
+  MAX_CAPTURED_LINES,
+} from "../src/invocation-logs";
+import { CONSOLE_PATCH } from "../src/worker-entry";
+import { WFP_COMPAT_DATE } from "./workerd";
+
+const PROXY_HEADER_BUDGET = 8192;
+
+/** A stand-in entry with production's prelude, store seeding and crash frame
+ *  (`buildAppEntrySource`, src/worker-entry.ts). `handlerBody` runs inside the
+ *  request store, exactly where a user handler runs. The outer catch stands in
+ *  for the dispatcher, which answers a throw with a Response of its own built
+ *  outside the request store (`mapDispatchError`). */
+const entry = (handlerBody: string) => `
+import { AsyncLocalStorage } from 'node:async_hooks';
+const _b44Store = new AsyncLocalStorage();
+const _b44Context = () => _b44Store.getStore();
+${CONSOLE_PATCH}
+${INVOCATION_LOGS_PATCH}
+export default {
+  async fetch(request, env, ctx) {
+    try {
+      return await _b44Store.run({ env: 'preview', secrets: env, workerEnv: env, waitUntil: (p) => ctx.waitUntil(p), ..._b44CaptureStore(request) }, async () => {
+        const handler = async () => { ${handlerBody} };
+        try {
+          return _b44AttachInvocationLogs(await handler());
+        } catch (e) {
+          console.error(e);
+          const crash = _b44CrashResponse(e);
+          if (crash) return crash;
+          throw e;
+        }
+      });
+    } catch {
+      return new Response(JSON.stringify({ error: 'dispatcher' }), { status: 500 });
+    }
+  },
+};
+`;
+
+async function invoke(handlerBody: string, { capture = true } = {}) {
+  const mf = new Miniflare({
+    modules: [{ type: "ESModule", path: "_bundled.mjs", contents: entry(handlerBody) }],
+    compatibilityDate: WFP_COMPAT_DATE,
+    compatibilityFlags: ["nodejs_compat"],
+  });
+  try {
+    const res = await mf.dispatchFetch("http://localhost/", {
+      headers: capture ? { [CAPTURE_LOGS_HEADER]: "1" } : {},
+    });
+    const text = await res.text();
+    const header = res.headers.get(INVOCATION_LOGS_HEADER);
+    return {
+      status: res.status,
+      headers: res.headers,
+      text,
+      json: () => JSON.parse(text),
+      header,
+      payload: header === null ? null : JSON.parse(header),
+    };
+  } finally {
+    await mf.dispose();
+  }
+}
+
+describe("a request that did not ask for its logs", () => {
+  it("gets no header and pays no capture cost", async () => {
+    const { status, headers, text, json, header } = await invoke(
+      `console.log("hi"); return new Response("ok");`,
+      { capture: false },
+    );
+
+    expect(status).toBe(200);
+    expect(header).toBeNull();
+  });
+
+  it("keeps rethrow semantics: the crash escapes to the dispatcher", async () => {
+    const { status, headers, text, json, header } = await invoke(`throw new Error("boom");`, { capture: false });
+
+    expect(status).toBe(500);
+    expect(json()).toEqual({ error: "dispatcher" });
+    expect(header).toBeNull();
+  });
+});
+
+describe("invocation logs on the response", () => {
+  it("returns the lines this invocation logged, with their levels", async () => {
+    const { status, headers, text, json, payload } = await invoke(`
+      console.log("starting");
+      console.error("boom", new Error("bad").message);
+      console.warn("%s items", 3);
+      return new Response("ok");
+    `);
+
+    expect(status).toBe(200);
+    expect(text).toBe("ok");
+    expect(payload.lines).toEqual([
+      { level: "info", message: "starting" },
+      { level: "error", message: "boom bad" },
+      { level: "warn", message: "3 items" },
+    ]);
+    expect(payload.dropped).toBe(0);
+  });
+
+  it("still answers for a silent invocation — an empty list, not a missing header", async () => {
+    // Absence of the header is the "app has not rebundled" signal, so a quiet
+    // request must not look like one.
+    const { header, payload } = await invoke(`return new Response("ok");`);
+
+    expect(header).not.toBeNull();
+    expect(payload).toEqual({ lines: [], dropped: 0 });
+  });
+
+  it("caps the line count and reports the overflow instead of hiding it", async () => {
+    const { payload } = await invoke(`
+      for (let i = 0; i < ${MAX_CAPTURED_LINES + 20}; i++) console.log("line " + i);
+      return new Response("ok");
+    `);
+
+    expect(payload.lines).toHaveLength(MAX_CAPTURED_LINES);
+    expect(payload.lines[0].message).toBe("line 0");
+    expect(payload.dropped).toBe(20);
+  });
+
+  it("keeps the header inside the proxy's budget however much is logged", async () => {
+    // The backend drops an oversized header outright, so the budget has to hold
+    // for adversarial content too — quotes and newlines are what JSON escaping
+    // expands, and the cap is charged in encoded bytes for that reason.
+    const { header, payload } = await invoke(`
+      for (let i = 0; i < ${MAX_CAPTURED_LINES}; i++) console.log('"'.repeat(400) + "\\n");
+      return new Response("ok");
+    `);
+
+    expect(header!.length).toBeLessThan(PROXY_HEADER_BUDGET);
+    expect(payload.dropped).toBeGreaterThan(0);
+  });
+
+  it("sends only the drop count when the function's own headers leave no room, keeping the response intact", async () => {
+    // Capture-on must never break a response that works capture-off, and the
+    // tool must not mistake the squeeze for a missing capability (store fallback).
+    const { status, headers, text, json, header, payload } = await invoke(`
+      for (let i = 0; i < ${MAX_CAPTURED_LINES}; i++) console.log("a line of ordinary length " + i);
+      return new Response("ok", { headers: { "Set-Cookie": "session=" + "x".repeat(6000) } });
+    `);
+
+    expect(status).toBe(200);
+    expect(text).toBe("ok");
+    expect(headers.get("Set-Cookie")).toHaveLength("session=".length + 6000);
+    expect(header!.length).toBeLessThan(40);
+    expect(payload).toEqual({ lines: [], dropped: MAX_CAPTURED_LINES });
+  });
+
+  it("omits even the drop count when the function's headers already fill the budget", async () => {
+    const { status, headers, text, json, header } = await invoke(`
+      console.log("hi");
+      return new Response("ok", { headers: { "Set-Cookie": "session=" + "x".repeat(7200) } });
+    `);
+
+    expect(status).toBe(200);
+    expect(headers.get("Set-Cookie")).toHaveLength("session=".length + 7200);
+    expect(header).toBeNull();
+  });
+
+  it("escapes non-ASCII text instead of losing the whole header to it", async () => {
+    // headers.set throws on any code unit above 0xff, and the attach swallows
+    // that — so without escaping, one emoji costs the request every line it
+    // logged and the tool silently falls back to the stale store.
+    const message = "שלום 😀 café";
+    const { header, payload } = await invoke(`
+      console.log(${JSON.stringify(message)});
+      return new Response("ok");
+    `);
+
+    expect(header).not.toBeNull();
+    expect(/[^\x00-\x7f]/.test(header!)).toBe(false);
+    expect(payload.lines[0].message).toBe(message);
+  });
+
+  it("passes the handler's own status, body and headers through the wrap", async () => {
+    const { status, headers, text, json } = await invoke(`
+      console.log("hi");
+      return new Response("nope", { status: 418, headers: { "X-Fn": "kept" } });
+    `);
+
+    expect(status).toBe(418);
+    expect(text).toBe("nope");
+    expect(headers.get("X-Fn")).toBe("kept");
+  });
+
+  it("answers a crash with the dispatcher's body shape and the crash's own lines attached", async () => {
+    const { status, headers, text, json, payload } = await invoke(`
+      console.log("before the crash");
+      throw new Error("boom");
+    `);
+
+    expect(status).toBe(500);
+    expect(json()).toEqual({
+      error: "user-exception",
+      detail: "user worker threw an exception",
+    });
+    expect(payload.lines[0]).toEqual({ level: "info", message: "before the crash" });
+    expect(payload.lines[1].level).toBe("error");
+    expect(payload.lines[1].message).toContain("Error: boom");
+  });
+
+  it("lets a quota error escape so the dispatcher's own classification stands", async () => {
+    const { status, headers, text, json, header } = await invoke(`throw new Error("Too many subrequests.");`);
+
+    expect(status).toBe(500);
+    expect(json()).toEqual({ error: "dispatcher" });
+    expect(header).toBeNull();
+  });
+});