fix(core): publish schedule completion events to connectors

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,

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({

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 } : {}),
+				});
+			},
 		});
 	}
 

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<string, unknown>;
+		}> = [];
+		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();
+	});
 });

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;

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<string, unknown>)
+						: undefined;
+				const sessionId =
+					typeof eventPayload?.sessionId === "string"
+						? eventPayload.sessionId.trim()
+						: undefined;
 				this.publish(
-					buildHubEvent(
-						mapped,
-						payload && typeof payload === "object"
-							? (payload as Record<string, unknown>)
-							: undefined,
-					),
+					buildHubEvent(mapped, eventPayload, sessionId || undefined),
 				);
 			},
 		});