/** * Per-account background loop for one BlueSky handle: owns the notification * poll timer, the action-processing timer, and the randomized automated-post * scheduler. Polls `BlueSkyClient.getNotifications`, dedupes against a cached * `indexedAt` cursor, and emits `bluesky.*` runtime events (mention/reply, * follow, like, repost, quote, should_respond, create_post) that the * application layer responds to — the manager itself never generates replies or * posts, it only fires events. Automated posts run inside a standalone * trajectory so background generation is traced. Instantiated and started by * `BlueSkyService`. */ import { type IAgentRuntime, logger, setTrajectoryPurpose, withStandaloneTrajectory, } from "@elizaos/core"; import type { BlueSkyClient } from "../client"; import type { BlueSkyConfig, BlueSkyCreatePostEventPayload, BlueSkyNotification, BlueSkyNotificationEventPayload, NotificationReason, } from "../types"; import { DEFAULT_BLUESKY_ACCOUNT_ID, getActionInterval, getMaxActionsProcessing, getPollInterval, getPostIntervalRange, isPostingEnabled, normalizeBlueSkyAccountId, shouldPostImmediately, } from "../utils/config"; function cursorCacheKey(agentId: string, accountId: string): string { return `bluesky:cursor:${agentId}:${accountId}`; } export class BlueSkyAgentManager { private pollTimer: ReturnType | null = null; private actionTimer: ReturnType | null = null; private postTimer: ReturnType | null = null; private running = false; private lastSeenAt: string | null = null; public readonly accountId: string; constructor( public readonly runtime: IAgentRuntime, public readonly config: BlueSkyConfig & { accountId?: string }, public readonly client: BlueSkyClient, ) { this.accountId = normalizeBlueSkyAccountId( config.accountId ?? DEFAULT_BLUESKY_ACCOUNT_ID, ); } getAccountId(): string { return this.accountId; } async start(): Promise { if (this.running) return; await this.client.authenticate(); this.running = true; const cached = await this.runtime.getCache( cursorCacheKey(this.runtime.agentId, this.accountId), ); if (typeof cached === "string" && cached) { this.lastSeenAt = cached; } this.startNotificationPolling(); if (this.config.enableActionProcessing) { this.startActionProcessing(); } if (isPostingEnabled(this.runtime, this.accountId)) { this.startAutomatedPosting(); } logger.success( { agentId: this.runtime.agentId, accountId: this.accountId }, "BlueSky agent manager started", ); } async stop(): Promise { this.running = false; if (this.pollTimer) clearInterval(this.pollTimer); if (this.actionTimer) clearInterval(this.actionTimer); if (this.postTimer) clearTimeout(this.postTimer); this.pollTimer = null; this.actionTimer = null; this.postTimer = null; await this.client.cleanup(); logger.info( { agentId: this.runtime.agentId, accountId: this.accountId }, "BlueSky agent manager stopped", ); } private startNotificationPolling(): void { const interval = getPollInterval(this.runtime, this.accountId); this.pollNotifications(); this.pollTimer = setInterval(() => this.pollNotifications(), interval); } private async pollNotifications(): Promise { if (!this.running) return; const { notifications } = await this.client.getNotifications(50); if (notifications.length === 0) return; const newNotifications = this.lastSeenAt ? notifications.filter((n) => { const lastSeen = this.lastSeenAt; return lastSeen !== null && n.indexedAt > lastSeen; }) : notifications; if (newNotifications.length > 0) { this.lastSeenAt = notifications[0].indexedAt; await this.runtime.setCache( cursorCacheKey(this.runtime.agentId, this.accountId), this.lastSeenAt, ); for (const notification of newNotifications) { this.emitNotificationEvent(notification); } await this.client.updateSeenNotifications(); } } private emitNotificationEvent(notification: BlueSkyNotification): void { const eventMap: Record = { mention: "bluesky.mention_received", reply: "bluesky.mention_received", follow: "bluesky.follow_received", like: "bluesky.like_received", repost: "bluesky.repost_received", quote: "bluesky.quote_received", }; const event = eventMap[notification.reason]; if (event) { const payload: BlueSkyNotificationEventPayload = { runtime: this.runtime, source: "bluesky", accountId: this.accountId, notification, }; void this.runtime.emitEvent(event, payload); } } private startActionProcessing(): void { const interval = getActionInterval(this.runtime, this.accountId); this.processQueuedActions(); this.actionTimer = setInterval(() => this.processQueuedActions(), interval); } private async processQueuedActions(): Promise { if (!this.running) return; const max = getMaxActionsProcessing(this.runtime, this.accountId); const { notifications } = await this.client.getNotifications(max); for (const notification of notifications) { if ( notification.reason === "mention" || notification.reason === "reply" ) { const payload: BlueSkyNotificationEventPayload = { runtime: this.runtime, source: "bluesky", accountId: this.accountId, notification, }; void this.runtime.emitEvent("bluesky.should_respond", payload); } } } private startAutomatedPosting(): void { if (shouldPostImmediately(this.runtime, this.accountId)) { void this.createAutomatedPost(); } this.scheduleNextPost(); } private scheduleNextPost(): void { const { min, max } = getPostIntervalRange(this.runtime, this.accountId); const interval = Math.random() * (max - min) + min; this.postTimer = setTimeout(() => { if (this.running) { void this.createAutomatedPost().finally(() => this.scheduleNextPost()); } }, interval); } private async createAutomatedPost(): Promise { await withStandaloneTrajectory( this.runtime, { source: "plugin-bluesky:auto-post", metadata: { platform: "bluesky", kind: "public_post_generation", automated: true, accountId: this.accountId, }, }, async () => { setTrajectoryPurpose("background"); const payload: BlueSkyCreatePostEventPayload = { runtime: this.runtime, source: "bluesky", automated: true, accountId: this.accountId, }; await this.runtime.emitEvent("bluesky.create_post", payload); }, ); } }