diff --git a/sdk/packages/core/src/cron/runner/cron-runner.ts b/sdk/packages/core/src/cron/runner/cron-runner.ts index c777cb1bfe..860288a46d 100644 --- a/sdk/packages/core/src/cron/runner/cron-runner.ts +++ b/sdk/packages/core/src/cron/runner/cron-runner.ts @@ -129,6 +129,13 @@ export interface CronRunnerOptions { pollIntervalMs?: number; claimLeaseSeconds?: number; globalMaxConcurrency?: number; + onRunComplete?: (event: { + run: CronRunRecord; + spec: CronSpecRecord; + status: CronRunRecord["status"]; + sessionId?: string; + error?: string; + }) => void; } export class CronRunner { @@ -340,6 +347,12 @@ export class CronRunner { claimToken: claim.claimToken, }); this.store.updateSpecLastRunAt(spec.specId, nowIso()); + this.notifyRunComplete({ + run, + spec, + status: "done", + sessionId, + }); } catch (err) { const isTimeout = err instanceof TimeoutError; if (sessionId && isTimeout) { @@ -365,6 +378,13 @@ export class CronRunner { error: message, claimToken: claim.claimToken, }); + this.notifyRunComplete({ + run, + spec, + status: "failed", + sessionId, + error: message, + }); } finally { releaseLeaseHeartbeat?.(); if (sessionId) { @@ -379,6 +399,25 @@ export class CronRunner { } } + private notifyRunComplete(event: { + run: CronRunRecord; + spec: CronSpecRecord; + status: CronRunRecord["status"]; + sessionId?: string; + error?: string; + }): void { + try { + this.options.onRunComplete?.(event); + } catch (error) { + const log = this.options.logger; + if (log) { + if (log.error) + log.error("cron.runner.run_complete_callback.failed", { error }); + else log.log("cron.runner.run_complete_callback.failed", { error }); + } + } + } + private buildPrompt( spec: CronSpecRecord, triggerEvent: CronEventLogRecord | undefined, diff --git a/sdk/packages/core/src/cron/service/schedule-service.test.ts b/sdk/packages/core/src/cron/service/schedule-service.test.ts index a30aedfca1..160e3f0a04 100644 --- a/sdk/packages/core/src/cron/service/schedule-service.test.ts +++ b/sdk/packages/core/src/cron/service/schedule-service.test.ts @@ -37,6 +37,7 @@ describe("HubScheduleService", () => { sqliteIt("creates, triggers, and reports schedule history", async () => { const dbPath = await createTempDbPath(); cleanupPaths.push(dbPath); + const events: Array<{ eventType: string; payload: unknown }> = []; const service = new HubScheduleService({ dbPath, runtimeHandlers: { @@ -53,6 +54,9 @@ describe("HubScheduleService", () => { abortSession: vi.fn(async () => ({ applied: true })), stopSession: vi.fn(async () => ({ applied: true })), }, + eventPublisher: (eventType, payload) => { + events.push({ eventType, payload }); + }, }); try { const created = service.createSchedule({ @@ -73,6 +77,17 @@ describe("HubScheduleService", () => { const execution = await service.triggerScheduleNow(created.scheduleId); expect(execution?.status).toBe("success"); expect(execution?.sessionId).toBe("session-1"); + expect(events).toEqual([ + { + eventType: "schedule.execution.completed", + payload: { + scheduleId: created.scheduleId, + executionId: execution?.executionId, + sessionId: "session-1", + status: "success", + }, + }, + ]); const schedule = service.getSchedule(created.scheduleId); expect(schedule?.metadata).toEqual({ diff --git a/sdk/packages/core/src/cron/service/schedule-service.ts b/sdk/packages/core/src/cron/service/schedule-service.ts index 765bb27ce9..7eeaaa5920 100644 --- a/sdk/packages/core/src/cron/service/schedule-service.ts +++ b/sdk/packages/core/src/cron/service/schedule-service.ts @@ -209,6 +209,18 @@ export class HubScheduleService { pollIntervalMs: options.pollIntervalMs, claimLeaseSeconds: options.claimLeaseSeconds, globalMaxConcurrency: options.globalMaxConcurrency, + onRunComplete: ({ run, spec, status, sessionId, error }) => { + if (spec.source !== "hub-schedule") { + return; + } + options.eventPublisher?.("schedule.execution.completed", { + scheduleId: spec.externalId, + executionId: run.runId, + sessionId, + status: runStatusToScheduleStatus(status), + ...(error ? { errorMessage: error } : {}), + }); + }, }); } diff --git a/sdk/packages/core/src/hub/client/session-client.test.ts b/sdk/packages/core/src/hub/client/session-client.test.ts index f4e3af6ab8..6b38bb5835 100644 --- a/sdk/packages/core/src/hub/client/session-client.test.ts +++ b/sdk/packages/core/src/hub/client/session-client.test.ts @@ -189,4 +189,62 @@ describe("HubSessionClient", () => { unsubscribe(); client.close(); }); + + it("maps schedule completion events for connector delivery streams", async () => { + vi.stubGlobal("WebSocket", MockWebSocket); + const client = new HubSessionClient({ + address: "ws://127.0.0.1:25463/hub", + clientId: "client-1", + }); + await client.connect(); + const socket = MockWebSocket.instances[0]; + if (!socket) { + throw new Error("expected websocket"); + } + const received: Array<{ + sessionId: string; + eventType: string; + payload: Record; + }> = []; + const unsubscribe = client.streamEvents( + {}, + { + onEvent: (event) => { + received.push(event); + }, + }, + ); + + socket.emitFrame({ + kind: "event", + envelope: { + version: "v1", + eventId: "evt-schedule", + event: "schedule.execution_completed", + timestamp: Date.now(), + payload: { + scheduleId: "sched_1", + executionId: "crun_1", + sessionId: "session-1", + status: "success", + }, + }, + }); + + expect(received).toEqual([ + { + sessionId: "session-1", + eventType: "schedule.execution.completed", + payload: { + scheduleId: "sched_1", + executionId: "crun_1", + sessionId: "session-1", + status: "success", + }, + }, + ]); + + unsubscribe(); + client.close(); + }); }); diff --git a/sdk/packages/core/src/hub/client/session-client.ts b/sdk/packages/core/src/hub/client/session-client.ts index dfd8a701cc..1a40e8f12d 100644 --- a/sdk/packages/core/src/hub/client/session-client.ts +++ b/sdk/packages/core/src/hub/client/session-client.ts @@ -158,58 +158,73 @@ function extractCheckpoint( } function mapHubEvent(event: HubEventEnvelope): HubStreamEvent | undefined { - const sessionId = event.sessionId?.trim(); + const payload = cloneRecord(event.payload); + const payloadSessionId = + typeof payload.sessionId === "string" ? payload.sessionId.trim() : ""; + const sessionId = event.sessionId?.trim() || payloadSessionId; if (!sessionId) { return undefined; } switch (event.event) { + case "schedule.execution_completed": + return { + sessionId, + eventType: "schedule.execution.completed", + payload, + }; + case "schedule.execution_failed": + return { + sessionId, + eventType: "schedule.execution.failed", + payload, + }; case "iteration.started": return { sessionId, eventType: "runtime.chat.iteration_start", - payload: cloneRecord(event.payload), + payload, }; case "iteration.finished": return { sessionId, eventType: "runtime.chat.iteration_end", - payload: cloneRecord(event.payload), + payload, }; case "assistant.delta": return { sessionId, eventType: "runtime.chat.text_delta", - payload: cloneRecord(event.payload), + payload, }; case "usage.updated": return { sessionId, eventType: "runtime.chat.usage", - payload: cloneRecord(event.payload), + payload, }; case "tool.started": return { sessionId, eventType: "runtime.chat.tool_call_start", - payload: cloneRecord(event.payload), + payload, }; case "tool.finished": return { sessionId, eventType: "runtime.chat.tool_call_end", - payload: cloneRecord(event.payload), + payload, }; case "approval.requested": return { sessionId, eventType: "approval.requested", - payload: cloneRecord(event.payload), + payload, }; case "run.aborted": return { sessionId, eventType: "runtime.chat.aborted", - payload: cloneRecord(event.payload), + payload, }; case "run.failed": return { @@ -221,7 +236,7 @@ function mapHubEvent(event: HubEventEnvelope): HubStreamEvent | undefined { return { sessionId, eventType: "runtime.chat.completed", - payload: cloneRecord(event.payload), + payload, }; default: return undefined; diff --git a/sdk/packages/core/src/hub/server/hub-server-transport.ts b/sdk/packages/core/src/hub/server/hub-server-transport.ts index 4b539c0d49..3aee15d4f5 100644 --- a/sdk/packages/core/src/hub/server/hub-server-transport.ts +++ b/sdk/packages/core/src/hub/server/hub-server-transport.ts @@ -227,13 +227,16 @@ export class HubServerTransport implements NativeHubTransport { if (!mapped) { return; } + const eventPayload = + payload && typeof payload === "object" + ? (payload as Record) + : undefined; + const sessionId = + typeof eventPayload?.sessionId === "string" + ? eventPayload.sessionId.trim() + : undefined; this.publish( - buildHubEvent( - mapped, - payload && typeof payload === "object" - ? (payload as Record) - : undefined, - ), + buildHubEvent(mapped, eventPayload, sessionId || undefined), ); }, });