Backend observability and hygiene for Durable Object resets

packages/backend-utils/__tests__/tracing.test.ts

@@ -0,0 +1,49 @@
+// Runs against the real workerd tracing API (vitest-pool-workers), pinning the platform
+// contract `traced` depends on: enterSpan ends its span when the returned promise settles, not
+// when the synchronous callback returns (docs: workers/observability/traces/custom-spans/).
+//
+// Caveats:
+// - Invocations are only traced when a tail consumer is attached, so vitest.config.ts wires a
+//   no-op streaming tail sink behind the experimental `streaming_tail_worker` flag. If workerd
+//   renames or removes that flag, setup fails loudly — a mechanical config fix, not a tracer bug.
+// - Attributes are write-only, so "span still open" is inferred from the documented guarantee
+//   that `span.isTraced` flips to false once the span ends. The entry assertion in each test
+//   guards the signal itself: if invocations stop being traced, isTraced starts false and the
+//   test fails rather than silently passing.
+
+import { describe, expect, it } from "vitest";
+import { createTracer } from "../src/tracing";
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+const traced = createTracer(() => ({}));
+
+describe("traced span lifetime", () => {
+  it("keeps the span open across awaits in an async callback", async () => {
+    const observed: boolean[] = [];
+    await traced("probe", async (span) => {
+      observed.push(span.isTraced); // must start true — guards the isTraced signal
+      await sleep(50);
+      observed.push(span.isTraced); // still true ⇒ span did not end at the sync return
+    });
+    expect(observed).toEqual([true, true]);
+  });
+
+  it("records the error attribute before the span can close on rejection", async () => {
+    let openWhenErrorSet: boolean | undefined;
+    await expect(
+      traced("probe-reject", (span) => {
+        // Shadow setAttribute to observe the instant traced()'s catch marks the failure.
+        const original = span.setAttribute.bind(span);
+        span.setAttribute = (key, value) => {
+          if (key === "error") openWhenErrorSet = span.isTraced;
+          original(key, value);
+        };
+        return (async () => {
+          await sleep(50);
+          throw new Error("boom");
+        })();
+      }),
+    ).rejects.toThrow("boom");
+    expect(openWhenErrorSet).toBe(true);
+  });
+});

packages/backend-utils/package.json

@@ -12,6 +12,10 @@
       "types": "./src/observability-context.ts",
       "import": "./src/observability-context.ts"
     },
+    "./tracing": {
+      "types": "./src/tracing.ts",
+      "import": "./src/tracing.ts"
+    },
     "./error-reporting": {
       "types": "./src/error-reporting.ts",
       "import": "./src/error-reporting.ts"

packages/backend-utils/src/tracing.ts

@@ -0,0 +1,46 @@
+import { tracing } from "cloudflare:workers";
+
+type Attribute = boolean | number | string;
+
+// The span surface exposed to callbacks. Lifetime is managed by `traced`, so no `end()`.
+export interface TraceSpan {
+  readonly isTraced: boolean;
+  setAttribute(key: string, value?: Attribute): void;
+}
+
+/**
+ * Creates a span helper that stamps the ambient observability context onto each span as
+ * attributes. Tracing only: never logs, never modifies context. Exceptions propagate
+ * unchanged, marked on the span via an `error` attribute (the beta API has no outcome).
+ * Sync and async callbacks both get correct spans: enterSpan ends the span only when the
+ * returned promise settles, not at the synchronous return (pinned by __tests__/tracing.test.ts).
+ */
+export function createTracer(getContext: () => Readonly<Record<string, unknown>>) {
+  return function traced<Result>(name: string, callback: (span: TraceSpan) => Result): Result {
+    return tracing.enterSpan(name, (span) => {
+      if (span.isTraced) {
+        for (const [key, value] of Object.entries(getContext())) {
+          if (typeof value === "boolean" || typeof value === "number" || typeof value === "string") {
+            span.setAttribute(key, value);
+          }
+        }
+      }
+      // Boolean marker only: error text is unbounded and possibly sensitive, so it belongs to
+      // logs/reporting, not trace attributes.
+      const fail = () => span.setAttribute("error", true);
+      try {
+        const result = callback(span);
+        // enterSpan keeps the span open until the returned promise settles, so async work gets
+        // its real duration and fail() runs before the span can close (the runtime watches the
+        // .catch-wrapped promise returned here). That wrapper is a new promise, not `result` —
+        // fine for data results; don't wrap pipelined RPC stubs in `traced`.
+        return result instanceof Promise
+            ? result.catch((err) => { fail(); throw err; }) as Result
+            : result;
+      } catch (err) {
+        fail();
+        throw err;
+      }
+    });
+  };
+}

packages/backend-utils/vitest.config.ts

@@ -6,12 +6,24 @@ export default defineConfig({
     cloudflareTest({
       miniflare: {
         compatibilityDate: "2026-02-02",
-        // nodejs_als enables observability context; experimental enables the Reporter stub below.
-        compatibilityFlags: ["experimental", "nodejs_als"],
+        // nodejs_als enables observability context; experimental enables the Reporter stub below
+        // and the streaming_tail_worker flag (which workerd refuses without experimental mode).
+        compatibilityFlags: ["experimental", "nodejs_als", "streaming_tail_worker"],
         serviceBindings: {
           ERROR_REPORTER: { name: "reporter", entrypoint: "ErrorReporter" },
         },
+        // Invocations are only traced (span.isTraced === true) when a tail consumer is attached;
+        // the no-op "span-sink" below exists solely so tracing.test.ts can observe span lifetime.
+        tails: ["span-sink"],
         workers: [{
+          name: "span-sink",
+          modules: true,
+          compatibilityDate: "2026-02-02",
+          compatibilityFlags: ["streaming_tail_worker"],
+          // The no-op tail() silences workerd's legacy tail delivery, which it attempts alongside
+          // the streaming path.
+          script: `export default { tail: () => {}, tailStream: () => () => {} }`,
+        }, {
           name: "reporter",
           modules: true,
           script: `

packages/workshop-backend/src/observability.ts

@@ -1,4 +1,5 @@
 import { createObservabilityContext } from "@gadgets/backend-utils/observability-context";
+import { createTracer } from "@gadgets/backend-utils/tracing";
 
 /** Observability fields emitted by the Workshop backend. */
 export type WorkshopObservabilityFields = {
@@ -14,6 +15,7 @@ export type WorkshopObservabilityFields = {
   failureCount: number;
   gadgetId: string;
   gatekeeperId: number | string;
+  logBytes: number;
   modelId: string;
   observerId: string;
   operation: string;
@@ -37,3 +39,6 @@ export const obsContext = createObservabilityContext<WorkshopObservabilityFields
 export function createWorkshopLogger(component: string) {
   return obsContext.createLogger({ component });
 }
+
+/** Runs `callback` in a trace span carrying the ambient observability fields as attributes. */
+export const traced = createTracer(obsContext.get);

packages/workshop-backend/src/overseer.ts

@@ -37,7 +37,7 @@ import { refreshCachedBalance } from "./ai-gateway-billing/cloudflare/connection
 import { SharingManager, SharingCaller, CollaboratorRecord, ShareKeyRecord } from "./sharing";
 import { AutoApprovalDrainer } from "./auto-approval";
 import { collectSlashCommands, invokeSlashCommand } from "./slash-commands";
-import { createWorkshopLogger, obsContext } from "./observability";
+import { createWorkshopLogger, obsContext, traced } from "./observability";
 import type { ChatGatewayRpcTarget, SubmitExternalMessageResult } from "@gadgets/workshop-shared/external-message-gateway";
 import {
   assertChatAttachmentSupportedByProvider,
@@ -934,7 +934,7 @@ function makeOverseerStorage(storage: DurableObjectStorage) {
 type OverseerStorage = ReturnType<typeof makeOverseerStorage>;
 
 // Don't build a snapshot until we have at least 64k of logs since the last one.
-const MIN_SNAPSHOT_THRESHOLD: number = 256; //65536;
+const MIN_SNAPSHOT_THRESHOLD: number = 65536;
 
 // Common internals that several interfaces implemented by the Overseer need to use. Can't just
 // declare private methods because some of the methods are needed by multiple classes.
@@ -1376,6 +1376,7 @@ class OverseerImpl implements AgentHooks {
 
     // Run the whole migration in one transaction so that a mid-migration error can't leave the
     // workspace half-migrated.
+    let startedAt = Date.now();
     this.ctx.storage.transactionSync(() => {
       // Version 0 -> 1: the workspace predates multi-gadget support. If it has any gadget content
       // (code beyond the initial empty snapshot, or named bindings), register that content as the
@@ -1455,6 +1456,10 @@ class OverseerImpl implements AgentHooks {
 
       this.storage.version.put(1);
     });
+
+    this.logger.info("migrated workspace storage", {
+      event: "storage.migration.completed", durationMs: Date.now() - startedAt,
+    });
   }
 
   // Allocate a workpiece ID from the shared counter. (The counter is named `nextGatekeeperId`
@@ -2003,18 +2008,24 @@ class OverseerImpl implements AgentHooks {
       this.#snapshotMetrics.logSize += update.length;
       if (this.#snapshotMetrics.logSize >
           Math.max(this.#snapshotMetrics.snapshotSize, MIN_SNAPSHOT_THRESHOLD)) {
-        let {ydoc} = this.buildYDoc("current");
-        let snapshotUpdate = Y.encodeStateAsUpdateV2(ydoc);
-        this.storage.snapshots.put({
-          version,
-          timestamp,
-          update: snapshotUpdate
+        let logBytes = this.#snapshotMetrics.logSize;
+        let startedAt = Date.now();
+        traced("code.snapshot.rebuild", (span) => {
+          let {ydoc} = this.buildYDoc("current");
+          let snapshotUpdate = Y.encodeStateAsUpdateV2(ydoc);
+          this.storage.snapshots.put({version, timestamp, update: snapshotUpdate});
+          span.setAttribute("gadgetId", this.ctx.id.toString());
+          span.setAttribute("size", snapshotUpdate.length);
+          span.setAttribute("logBytes", logBytes);
+          this.#snapshotMetrics = {
+            snapshotSize: snapshotUpdate.length,
+            logSize: 0,
+          };
+          this.logger.info("rebuilt code snapshot", {
+            event: "code.snapshot.rebuilt", durationMs: Date.now() - startedAt,
+            size: snapshotUpdate.length, logBytes, sequence: version,
+          });
         });
-
-        this.#snapshotMetrics = {
-          snapshotSize: snapshotUpdate.length,
-          logSize: 0,
-        };
       }
     }
 
@@ -3857,8 +3868,8 @@ class OverseerImpl implements AgentHooks {
       gadgetId: this.ctx.id.toString(),
       chatId,
       modelId: aiModel.profile.id,
-    }, () => this.#runAgentTurnWithContext(
-        chatId, aiModel, initiator, callbackInitiated, liveChat));
+    }, () => traced("agent.run", () => this.#runAgentTurnWithContext(
+        chatId, aiModel, initiator, callbackInitiated, liveChat)));
   }
 
   async #runAgentTurnWithContext(chatId: number, aiModel: UserAiModelRecord,
@@ -7311,6 +7322,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer {
     if (!this.isOwner) {
       throw new Error("Only the workspace owner can delete it.");
     }
+    let startedAt = Date.now();
 
     this.impl.recordGadgetAnalytics({
       event_name: "gadget_deleted",
@@ -7337,6 +7349,10 @@ class OverseerClientInterface extends RpcTarget implements Overseer {
       this.impl.scheduleRevocationRestart();
       this.impl.ownerId = undefined;
     });
+
+    this.impl.logger.info("deleted workspace", {
+      event: "workspace.delete.completed", durationMs: Date.now() - startedAt,
+    });
   }
 
   async subscribeToCode(subscriber: RpcStub<CodeSubscriber>, fromVersion: number = 0)
@@ -8390,6 +8406,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer {
   }
 
   async deleteChat(chatId: number): Promise<void> {
+    let startedAt = Date.now();
     let response = this.impl.storage.gadgetResponseDeliveries.undeliveredByChatId.get(chatId);
     if (response?.status === "waiting") {
       this.impl.deliverExternalMessageResponse(response, "The chat was deleted before the agent responded.");
@@ -8459,6 +8476,10 @@ class OverseerClientInterface extends RpcTarget implements Overseer {
 
     // Clean up all in-memory live state for this chat.
     this.impl.destroyLiveChat(chatId);
+
+    this.impl.logger.info("deleted chat", {
+      event: "chat.delete.completed", chatId, durationMs: Date.now() - startedAt,
+    });
   }
 
   async stopAgent(chatId: number): Promise<void> {

packages/workshop-backend/src/server.ts

@@ -235,7 +235,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi {
       if (started && !closed) {
         // this.ctx.abort() would be nicer here, but it is still marked experimental in the
         // workers runtime.
-        this.abortSession(new Error("lost connection to workspace DO"));
+        this.abortSession(new Error(`lost connection to workspace DO (gadget ${id})`));
       }
     }
 
@@ -846,6 +846,8 @@ export default {
       let resp: Response | undefined;
       let aborted = false;
       let abortSession = (reason: Error) => {
+        // Closing the socket fails no invocation, so nothing else logs this.
+        logger.warn("aborting api session", { event: "session.abort", error: reason });
         aborted = true;
         resp?.webSocket?.close();
       };

packages/workshop-backend/wrangler.jsonc

@@ -94,7 +94,10 @@
   "observability": {
     "enabled": true,
     "head_sampling_rate": 1,
-    "logs": { "invocation_logs": false }
+    "logs": { "invocation_logs": false },
+    // Traces are free during the beta; spans bill against the Logs quota from 2026-10-01,
+    // so revisit the sampling rate before then.
+    "traces": { "enabled": true, "head_sampling_rate": 0.5 }
   }
 
 }

scripts/testdata/golden-manifest.json

@@ -1111,6 +1111,10 @@
         "head_sampling_rate": 1,
         "logs": {
           "invocation_logs": false
+        },
+        "traces": {
+          "enabled": true,
+          "head_sampling_rate": 0.5
         }
       },
       "vars": {