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); }