From aadae572f896c8d057890e94029c086f7cdb2b45 Mon Sep 17 00:00:00 2001 From: Oscar Jiang Date: Sat, 11 Jul 2026 18:09:26 +0800 Subject: [PATCH] feat(eve): stamp schedule-dispatched sessions with $eve.schedule MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sessions started by a schedule dispatch carry no record of which schedule fired: the markdown form only gets $eve.trigger="schedule", and sessions a handler starts through args.receive(...) carry the target channel's kind, so trigger-based filtering cannot attribute runs to a schedule at all. The ScheduleDispatcher now wraps the dispatch in an eve context scope carrying the schedule name under the new ScheduleIdKey. buildRunContext inherits the key from the ambient scope into each run started inside the dispatch — covering RunInputs built out of the dispatcher's hands by authored channel receive hooks — and buildSessionAttributes emits it as $eve.schedule. $eve.trigger semantics are unchanged. Closes #708 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_0164WUaSv9mN58wgnnoN2vKG Signed-off-by: Oscar Jiang --- .changeset/schedule-attribution-attribute.md | 7 ++ docs/guides/instrumentation.md | 1 + packages/eve/src/channel/schedule.test.ts | 52 +++++++++++++++ packages/eve/src/channel/schedule.ts | 64 +++++++++++-------- packages/eve/src/context/keys.ts | 7 ++ .../execution/eve-workflow-attributes.test.ts | 34 +++++++++- .../src/execution/eve-workflow-attributes.ts | 16 ++++- packages/eve/src/execution/runtime-context.ts | 12 +++- 8 files changed, 165 insertions(+), 28 deletions(-) create mode 100644 .changeset/schedule-attribution-attribute.md diff --git a/.changeset/schedule-attribution-attribute.md b/.changeset/schedule-attribution-attribute.md new file mode 100644 index 000000000..67c8696fc --- /dev/null +++ b/.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. diff --git a/docs/guides/instrumentation.md b/docs/guides/instrumentation.md index 7357cb328..dd9070f3e 100644 --- a/docs/guides/instrumentation.md +++ b/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): diff --git a/packages/eve/src/channel/schedule.test.ts b/packages/eve/src/channel/schedule.test.ts index 374dabb34..c24d649d4 100644 --- a/packages/eve/src/channel/schedule.test.ts +++ b/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 = []; + 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 = []; + 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: [] }); diff --git a/packages/eve/src/channel/schedule.ts b/packages/eve/src/channel/schedule.ts index 7fed44100..04b2945d2 100644 --- a/packages/eve/src/channel/schedule.ts +++ b/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 { - const sessions: Session[] = []; - const waitUntilTasks: Promise[] = []; - 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[] = []; + 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 { diff --git a/packages/eve/src/context/keys.ts b/packages/eve/src/context/keys.ts index 8c46b9bbb..95a092909 100644 --- a/packages/eve/src/context/keys.ts +++ b/packages/eve/src/context/keys.ts @@ -61,6 +61,13 @@ export const InitiatorAuthKey = new ContextKey("eve.i export const SessionIdKey = new ContextKey("eve.sessionId"); export const ContinuationTokenKey = new ContextKey("eve.continuationToken"); export const ChannelRequestIdKey = new ContextKey("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("eve.scheduleId"); export const ChannelInstrumentationKey = new ContextKey( "eve.channelInstrumentation", ); diff --git a/packages/eve/src/execution/eve-workflow-attributes.test.ts b/packages/eve/src/execution/eve-workflow-attributes.test.ts index 4a91b4d49..37920c801 100644 --- a/packages/eve/src/execution/eve-workflow-attributes.test.ts +++ b/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", () => { diff --git a/packages/eve/src/execution/eve-workflow-attributes.ts b/packages/eve/src/execution/eve-workflow-attributes.ts index 4e76b0e13..654c5a632 100644 --- a/packages/eve/src/execution/eve-workflow-attributes.ts +++ b/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 | 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), }; } diff --git a/packages/eve/src/execution/runtime-context.ts b/packages/eve/src/execution/runtime-context.ts index f63c9dce3..16fc448a1 100644 --- a/packages/eve/src/execution/runtime-context.ts +++ b/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); }