feat(eve): stamp schedule-dispatched sessions with $eve.schedule

.changeset/schedule-attribution-attribute.md

@@ -0,0 +1,7 @@
+---
+"eve": minor
+---
+
+feat(eve): stamp schedule-dispatched sessions with a `$eve.schedule` workflow attribute
+
+Sessions started by a schedule dispatch now carry `$eve.schedule` (the authored schedule's name) in their workflow run attributes — including sessions a handler starts through `args.receive(...)`, which keep the target channel's kind in `$eve.trigger`. Observability tooling can attribute runs to a specific schedule instead of relying on trigger heuristics.

docs/guides/instrumentation.md

@@ -129,6 +129,7 @@ Structural tags describe each run's place in the tree:
 - `$eve.subagent`: compiled graph node id (subagent runs only)
 - `$eve.trigger`: the channel kind that started the run
 - `$eve.title`: truncated title derived from the first user message
+- `$eve.schedule`: name of the authored schedule whose dispatch started the session, present only on schedule-dispatched sessions — including sessions a schedule handler starts through `args.receive(...)`, which keep the target channel's kind in `$eve.trigger`
 
 Per-turn usage tags are written on each step of a turn, accumulating cumulative totals (last write wins):
 

packages/eve/src/channel/schedule.test.ts

@@ -9,6 +9,8 @@ import {
   SCHEDULE_APP_AUTH,
   ScheduleDispatcher,
 } from "#channel/schedule.js";
+import { contextStorage } from "#context/container.js";
+import { ScheduleIdKey } from "#context/keys.js";
 import type { RunHandle, Runtime } from "#channel/types.js";
 import { slackChannel } from "#public/channels/slack/slackChannel.js";
 import type { ResolvedChannelDefinition } from "#runtime/types.js";
@@ -169,6 +171,56 @@ describe("ScheduleDispatcher", () => {
     });
   });
 
+  describe("schedule dispatch scope", () => {
+    const activeScheduleId = () => contextStorage.getStore()?.get(ScheduleIdKey);
+
+    it("is active when the markdown run starts, and cleared after the dispatch", async () => {
+      const runtime = createMockRuntime();
+      const observedScheduleIds: Array<string | undefined> = [];
+      runtime.run = vi.fn(async () => {
+        observedScheduleIds.push(activeScheduleId());
+        return createMockRunHandle();
+      });
+      const dispatcher = new ScheduleDispatcher({ runtime, channels: [] });
+
+      await dispatcher.trigger({ scheduleId: "heartbeat", markdown: "Run heartbeat task." });
+
+      expect(observedScheduleIds).toEqual(["heartbeat"]);
+      expect(activeScheduleId()).toBeUndefined();
+    });
+
+    it("is active when a handler starts a session through args.receive()", async () => {
+      const runtime = createMockRuntime();
+      const observedScheduleIds: Array<string | undefined> = [];
+      runtime.run = vi.fn(async () => {
+        observedScheduleIds.push(activeScheduleId());
+        return createMockRunHandle();
+      });
+      vi.stubEnv("SLACK_BOT_TOKEN", "xoxb-test");
+      vi.stubEnv("SLACK_SIGNING_SECRET", "test-secret");
+      try {
+        const { definition, resolved } = makeSlackChannelEntry();
+        const dispatcher = new ScheduleDispatcher({ runtime, channels: [resolved] });
+
+        await dispatcher.trigger({
+          scheduleId: "daily-digest",
+          async run({ receive, appAuth }) {
+            await receive(definition, {
+              message: "post the digest",
+              target: { channelId: "C0123ABC" },
+              auth: appAuth,
+            });
+          },
+        });
+
+        expect(observedScheduleIds).toEqual(["daily-digest"]);
+        expect(activeScheduleId()).toBeUndefined();
+      } finally {
+        vi.unstubAllEnvs();
+      }
+    });
+  });
+
   it("throws when neither run nor markdown is provided", async () => {
     const runtime = createMockRuntime();
     const dispatcher = new ScheduleDispatcher({ runtime, channels: [] });

packages/eve/src/channel/schedule.ts

@@ -1,5 +1,7 @@
 import type { ChannelAdapter } from "#channel/adapter.js";
 import { SCHEDULE_APP_AUTH } from "#channel/schedule-auth.js";
+import { ContextContainer, contextStorage } from "#context/container.js";
+import { ScheduleIdKey } from "#context/keys.js";
 import {
   createCrossChannelReceiveFn,
   toCrossChannelTargets,
@@ -75,34 +77,46 @@ export class ScheduleDispatcher {
   }
 
   async trigger(input: ScheduleDispatchInput): Promise<ScheduleDispatchResult> {
-    const sessions: Session[] = [];
-    const waitUntilTasks: Promise<unknown>[] = [];
-    const receive = createCrossChannelReceiveFn(this.runtime, toCrossChannelTargets(this.channels));
+    // The dispatch runs inside an eve context scope carrying the schedule
+    // name, so every run started within it — the synthesized markdown run
+    // and sessions the handler starts through `args.receive(...)` (whose
+    // `RunInput` is built by the target channel's authored `receive` hook,
+    // out of the dispatcher's hands) — is attributable to this schedule.
+    const scope = new ContextContainer();
+    scope.set(ScheduleIdKey, input.scheduleId);
+    return await contextStorage.run(scope, async () => {
+      const sessions: Session[] = [];
+      const waitUntilTasks: Promise<unknown>[] = [];
+      const receive = createCrossChannelReceiveFn(
+        this.runtime,
+        toCrossChannelTargets(this.channels),
+      );
 
-    const args: ScheduleHandlerArgs = {
-      appAuth: SCHEDULE_APP_AUTH,
-      receive: async (channel, options) => {
-        const session = await receive(channel, options);
-        sessions.push(session);
-        return session;
-      },
-      waitUntil(task) {
-        waitUntilTasks.push(task);
-      },
-    };
+      const args: ScheduleHandlerArgs = {
+        appAuth: SCHEDULE_APP_AUTH,
+        receive: async (channel, options) => {
+          const session = await receive(channel, options);
+          sessions.push(session);
+          return session;
+        },
+        waitUntil(task) {
+          waitUntilTasks.push(task);
+        },
+      };
 
-    if (input.run) {
-      await input.run(args);
-    } else if (input.markdown !== undefined) {
-      const session = await this.runMarkdown(input.markdown);
-      sessions.push(session);
-    } else {
-      throw new Error(
-        `Schedule "${input.scheduleId}" has neither "run" nor "markdown" — at least one must be set.`,
-      );
-    }
+      if (input.run) {
+        await input.run(args);
+      } else if (input.markdown !== undefined) {
+        const session = await this.runMarkdown(input.markdown);
+        sessions.push(session);
+      } else {
+        throw new Error(
+          `Schedule "${input.scheduleId}" has neither "run" nor "markdown" — at least one must be set.`,
+        );
+      }
 
-    return { sessions, waitUntilTasks };
+      return { sessions, waitUntilTasks };
+    });
   }
 
   private async runMarkdown(markdown: string): Promise<Session> {

packages/eve/src/context/keys.ts

@@ -61,6 +61,13 @@ export const InitiatorAuthKey = new ContextKey<SessionAuthContext | null>("eve.i
 export const SessionIdKey = new ContextKey<string>("eve.sessionId");
 export const ContinuationTokenKey = new ContextKey<string>("eve.continuationToken");
 export const ChannelRequestIdKey = new ContextKey<string>("eve.channelRequestId");
+/**
+ * Name of the authored schedule whose dispatch started this run. The
+ * `ScheduleDispatcher` sets it on an ambient context scope around the
+ * dispatch so sessions a handler starts through `args.receive(...)`
+ * carry it too.
+ */
+export const ScheduleIdKey = new ContextKey<string>("eve.scheduleId");
 export const ChannelInstrumentationKey = new ContextKey<ChannelInstrumentationProjection>(
   "eve.channelInstrumentation",
 );

packages/eve/src/execution/eve-workflow-attributes.test.ts

@@ -11,8 +11,9 @@ import {
   readParentLineage,
   readParentSessionId,
   readRootSessionId,
+  readScheduleId,
 } from "#execution/eve-workflow-attributes.js";
-import { ChannelRequestIdKey } from "#context/keys.js";
+import { ChannelRequestIdKey, ScheduleIdKey } from "#context/keys.js";
 
 const slackChannelCtx = {
   "eve.channel": { kind: "slack", state: { team: "T1" } },
@@ -102,6 +103,22 @@ describe("readChannelRequestId", () => {
   });
 });
 
+describe("readScheduleId", () => {
+  it("returns the schedule name when the context slot is well-formed", () => {
+    expect(
+      readScheduleId({
+        [ScheduleIdKey.name]: "daily-digest",
+      }),
+    ).toBe("daily-digest");
+  });
+
+  it("returns undefined when the slot is missing or malformed", () => {
+    expect(readScheduleId({})).toBeUndefined();
+    expect(readScheduleId({ [ScheduleIdKey.name]: "" })).toBeUndefined();
+    expect(readScheduleId({ [ScheduleIdKey.name]: 42 })).toBeUndefined();
+  });
+});
+
 describe("deriveSessionTitle", () => {
   it("collapses whitespace and trims plain string messages", () => {
     expect(deriveSessionTitle("  hello\n\nworld   ")).toBe("hello world");
@@ -173,6 +190,21 @@ describe("buildSessionAttributes", () => {
 
     expect(attrs["$eve.channel_request_id"]).toBe("req_session");
   });
+
+  it("emits the schedule name for schedule-dispatched sessions", () => {
+    const attrs = buildSessionAttributes({
+      inputMessage: "post the digest",
+      serializedContext: {
+        ...slackChannelCtx,
+        [ScheduleIdKey.name]: "daily-digest",
+      },
+    });
+
+    // Channel-dispatched sessions keep their channel kind in the trigger;
+    // the schedule attribute carries provenance alongside it.
+    expect(attrs["$eve.trigger"]).toBe("slack");
+    expect(attrs["$eve.schedule"]).toBe("daily-digest");
+  });
 });
 
 describe("buildSubagentRootAttributes", () => {

packages/eve/src/execution/eve-workflow-attributes.ts

@@ -23,9 +23,11 @@
  * - `$eve.trigger`      — channel adapter kind (session/subagent rows)
  * - `$eve.title`        — truncated session title from the first user message
  * - `$eve.channel_request_id` — inbound channel request id
+ * - `$eve.schedule`     — name of the authored schedule whose dispatch
+ *                         started the session (schedule-dispatched rows only)
  */
 
-import { ChannelRequestIdKey } from "#context/keys.js";
+import { ChannelRequestIdKey, ScheduleIdKey } from "#context/keys.js";
 import type { EveAttributeValue } from "#runtime/attributes/normalize.js";
 import { isNonEmptyString } from "#shared/guards.js";
 
@@ -132,6 +134,17 @@ export function readChannelRequestId(
   return isNonEmptyString(channelRequestId) ? channelRequestId : undefined;
 }
 
+/**
+ * Reads the schedule name seeded by the schedule dispatch scope.
+ * Returns `undefined` when the run was not started by a schedule
+ * dispatch — including channel-dispatched sessions, which keep their
+ * channel kind in `$eve.trigger` and carry provenance here instead.
+ */
+export function readScheduleId(serializedContext: Record<string, unknown>): string | undefined {
+  const scheduleId = serializedContext[ScheduleIdKey.name];
+  return isNonEmptyString(scheduleId) ? scheduleId : undefined;
+}
+
 /**
  * Maximum visible length (in code points) of a derived `$eve.title`.
  *
@@ -212,6 +225,7 @@ export function buildSessionAttributes(input: {
     "$eve.type": "session",
     "$eve.trigger": readChannelKind(input.serializedContext),
     "$eve.title": deriveSessionTitle(input.inputMessage),
+    "$eve.schedule": readScheduleId(input.serializedContext),
   };
 }
 

packages/eve/src/execution/runtime-context.ts

@@ -1,5 +1,5 @@
 import type { RunInput, SessionAuthContext } from "#channel/types.js";
-import { ContextContainer } from "#context/container.js";
+import { ContextContainer, contextStorage } from "#context/container.js";
 import { setChannelContext } from "#execution/channel-context.js";
 import {
   AuthKey,
@@ -10,6 +10,7 @@ import {
   InitiatorAuthKey,
   ModeKey,
   ParentSessionKey,
+  ScheduleIdKey,
   SessionCallbackKey,
   SubagentDepthKey,
   SubagentMaxDepthKey,
@@ -51,6 +52,15 @@ export function buildRunContext(input: {
     ctx.set(ChannelRequestIdKey, run.requestId);
   }
 
+  // Inherited from the ambient scope rather than `RunInput`: sessions a
+  // schedule handler starts through `args.receive(...)` get their RunInput
+  // from the target channel's authored `receive` hook, which cannot thread
+  // the schedule identity through.
+  const scheduleId = contextStorage.getStore()?.get(ScheduleIdKey);
+  if (scheduleId !== undefined) {
+    ctx.set(ScheduleIdKey, scheduleId);
+  }
+
   if (run.callback !== undefined) {
     ctx.set(SessionCallbackKey, run.callback);
   }