diff --git a/src/modules/qqbot/core/application/message-push/system-message-fanout.service.ts b/src/modules/qqbot/core/application/message-push/system-message-fanout.service.ts
new file mode 100644
index 0000000..4983efb
--- /dev/null
+++ b/src/modules/qqbot/core/application/message-push/system-message-fanout.service.ts
@@ -0,0 +1,613 @@
+import { Injectable } from '@nestjs/common';
+import { DataSource, Brackets, type EntityManager } from 'typeorm';
+import { KtDateTime } from '@/common';
+import {
+  SystemMessageContractError,
+  type SystemMessageDeliveryReadiness,
+  type SystemMessageScalar,
+  type SystemMessageSourceAdapter,
+} from '../../contract/message-push/qqbot-message-push.types';
+import { QqbotAccount } from '../../infrastructure/persistence/account/qqbot-account.entity';
+import { QqbotMessageDelivery } from '../../infrastructure/persistence/message-push/qqbot-message-delivery.entity';
+import { QqbotMessageEvent } from '../../infrastructure/persistence/message-push/qqbot-message-event.entity';
+import { QqbotMessagePublishBinding } from '../../infrastructure/persistence/message-push/qqbot-message-publish-binding.entity';
+import { QqbotMessagePublishTarget } from '../../infrastructure/persistence/message-push/qqbot-message-publish-target.entity';
+import { QqbotMessageSubscription } from '../../infrastructure/persistence/message-push/qqbot-message-subscription.entity';
+import { QqbotMessageTemplate } from '../../infrastructure/persistence/message-push/qqbot-message-template.entity';
+import {
+  SYSTEM_MESSAGE_BATCH_SIZE,
+  SYSTEM_MESSAGE_DDNS_RECHECK_MS,
+  SYSTEM_MESSAGE_LEASE_MS,
+  SYSTEM_MESSAGE_RETRY_BASE_MS,
+  SYSTEM_MESSAGE_RETRY_MAX_MS,
+  SYSTEM_MESSAGE_RETRY_WINDOW_MS,
+} from './system-message-runner.constants';
+import { SystemMessageSourceRegistry } from './system-message-source.registry';
+import { SystemMessageTemplateRendererService } from './system-message-template-renderer.service';
+
+const STUN_MAPPING_PORT_SOURCE = 'network.stun.mapping-port-changed';
+const TRANSIENT_ERROR_CODE = 'fanout_transient_error';
+const EVENT_EXPIRED_ERROR_CODE = 'fanout_expired';
+const EVENT_RESOURCE_MISMATCH_ERROR_CODE = 'event_resource_mismatch';
+const SUPERSEDED_STATUSES = new Set(['waiting_ddns', 'pending', 'retry']);
+
+interface ClaimToken {
+  attempt: number;
+  event: QqbotMessageEvent;
+  leaseUntil: KtDateTime;
+}
+
+/**
+ * Creates frozen, idempotent delivery work from committed system-message Outbox facts.
+ *
+ * It deliberately only persists work. The later delivery coordinator owns DDNS wakeups
+ * and every OneBot call.
+ */
+@Injectable()
+export class SystemMessageFanoutService {
+  /**
+   * Creates the durable Outbox fan-out service.
+   * @param dataSource - Core database source used for short claim and subscription transactions.
+   * @param sourceRegistry - Registered source adapters for payload validation and readiness.
+   * @param templateRenderer - Safe literal-only renderer used to freeze final text.
+   */
+  constructor(
+    private readonly dataSource: DataSource,
+    private readonly sourceRegistry: SystemMessageSourceRegistry,
+    private readonly templateRenderer: SystemMessageTemplateRendererService,
+  ) {}
+
+  /**
+   * Claims and processes at most one bounded batch of due Outbox events.
+   * @param now - Stable clock instant for leases, schedules, and retry decisions.
+   * @returns Count of rows successfully claimed, including rows that become retry or failed.
+   */
+  async runOnce(now: Date = new Date()): Promise<number> {
+    let claimed = 0;
+    for (let index = 0; index < SYSTEM_MESSAGE_BATCH_SIZE; index += 1) {
+      const token = await this.claimOne(now);
+      if (!token) break;
+      claimed += 1;
+      await this.processClaim(token, now);
+    }
+    return claimed;
+  }
+
+  /**
+   * Atomically claims the oldest due Outbox event using an exact lease ownership token.
+   * @param now - Clock instant against which due rows and expired leases are evaluated.
+   * @returns Claimed event with its new attempt and lease, or null when no eligible row exists.
+   */
+  private async claimOne(now: Date): Promise<ClaimToken | null> {
+    return this.dataSource.transaction(async (manager) => {
+      const events = manager.getRepository(QqbotMessageEvent);
+      const event = await events
+        .createQueryBuilder('event')
+        .setLock('pessimistic_write')
+        .setOnLocked('skip_locked')
+        .where(
+          new Brackets((where) => {
+            where
+              .where(
+                'event.fanoutStatus IN (:...due) AND event.nextFanoutAt <= :now',
+                { due: ['accepted', 'retry'], now },
+              )
+              .orWhere(
+                'event.fanoutStatus = :processing AND event.fanoutLeaseUntil <= :now',
+                { processing: 'processing', now },
+              );
+          }),
+        )
+        .orderBy('event.occurredAt', 'ASC')
+        .addOrderBy('event.id', 'ASC')
+        .take(1)
+        .getOne();
+      if (!event) return null;
+
+      const leaseUntil = new KtDateTime(
+        now.getTime() + SYSTEM_MESSAGE_LEASE_MS,
+      );
+      event.fanoutAttemptCount += 1;
+      event.fanoutLeaseUntil = leaseUntil;
+      event.fanoutStatus = 'processing';
+      event.nextFanoutAt = null;
+      await events.save(event);
+      return { attempt: event.fanoutAttemptCount, event, leaseUntil };
+    });
+  }
+
+  /**
+   * Validates one owned event and fans it out without allowing stale owners to finish it.
+   * @param token - Attempt-plus-lease token returned by the successful claim transaction.
+   * @param now - Stable clock instant for expiry and next retry time.
+   */
+  private async processClaim(token: ClaimToken, now: Date): Promise<void> {
+    if (this.isExpired(token.event, now)) {
+      await this.finish(
+        token,
+        'failed',
+        EVENT_EXPIRED_ERROR_CODE,
+        'fan-out deadline reached',
+      );
+      return;
+    }
+
+    try {
+      const adapter = this.sourceRegistry.get(token.event.sourceKey);
+      const payload = adapter.validateEventPayload(token.event.payload);
+      this.assertResourceIdentity(token.event, payload);
+      const subscriptions = await this.findMatchingSubscriptions(token.event);
+      let transientFailure = false;
+
+      for (const subscription of subscriptions) {
+        try {
+          await this.dataSource.transaction((manager) =>
+            this.fanOutSubscription(
+              manager,
+              token.event,
+              subscription.id,
+              adapter,
+              payload,
+              now,
+            ),
+          );
+        } catch {
+          transientFailure = true;
+        }
+      }
+
+      if (transientFailure) {
+        await this.retryOrFail(
+          token,
+          now,
+          TRANSIENT_ERROR_CODE,
+          'fan-out dependency unavailable',
+        );
+        return;
+      }
+      await this.finish(token, 'completed', null, null);
+    } catch (error) {
+      if (error instanceof SystemMessageContractError) {
+        await this.finish(token, 'failed', error.code, this.safeMessage(error));
+        return;
+      }
+      if (error instanceof EventResourceMismatchError) {
+        await this.finish(
+          token,
+          'failed',
+          EVENT_RESOURCE_MISMATCH_ERROR_CODE,
+          error.message,
+        );
+        return;
+      }
+      await this.retryOrFail(
+        token,
+        now,
+        TRANSIENT_ERROR_CODE,
+        'fan-out dependency unavailable',
+      );
+    }
+  }
+
+  /**
+   * Selects active subscriptions whose own JSON resource field exactly matches the event.
+   * @param event - Frozen Outbox event being fanned out.
+   * @returns Deterministically ordered eligible subscription rows.
+   */
+  private async findMatchingSubscriptions(
+    event: QqbotMessageEvent,
+  ): Promise<QqbotMessageSubscription[]> {
+    const subscriptions = await this.dataSource
+      .getRepository(QqbotMessageSubscription)
+      .find({
+        where: { enabled: true, isDeleted: false, sourceKey: event.sourceKey },
+        order: { id: 'ASC' },
+      });
+    return subscriptions.filter((subscription) =>
+      this.matchesSubscription(subscription, event),
+    );
+  }
+
+  /**
+   * Executes one subscription's isolated persistence unit.
+   * @param manager - Transaction manager whose mutations commit only for this subscription.
+   * @param event - Frozen Outbox event currently owned by the caller.
+   * @param subscriptionId - Primary key of the subscription to lock and recheck.
+   * @param adapter - Source adapter already used to validate the frozen payload.
+   * @param payload - Validated scalar payload passed to the adapter exactly once.
+   * @param now - Stable scheduling instant for newly created delivery rows.
+   */
+  private async fanOutSubscription(
+    manager: EntityManager,
+    event: QqbotMessageEvent,
+    subscriptionId: string,
+    adapter: SystemMessageSourceAdapter,
+    payload: Record<string, SystemMessageScalar>,
+    now: Date,
+  ): Promise<void> {
+    const subscriptions = manager.getRepository(QqbotMessageSubscription);
+    const subscription = await subscriptions.findOne({
+      where: { id: subscriptionId },
+      lock: { mode: 'pessimistic_write' },
+    });
+    if (!subscription || !this.matchesSubscription(subscription, event)) return;
+
+    const readiness = await adapter.resolveDelivery({
+      eventPayload: payload,
+      subscriptionConfig: subscription.sourceConfig,
+    });
+    await this.supersedeEarlierDeliveries(manager, event, subscription.id);
+    if (!this.hasRenderableVariables(readiness)) return;
+
+    await this.createDeliveries(manager, event, subscription, readiness, now);
+  }
+
+  /**
+   * Supersedes only unfinished deliveries of events strictly earlier than the current event.
+   * @param manager - Current subscription transaction manager.
+   * @param event - Current event that supersedes older work for this subscription.
+   * @param subscriptionId - Subscription scope that prevents unrelated fan-out interference.
+   */
+  private async supersedeEarlierDeliveries(
+    manager: EntityManager,
+    event: QqbotMessageEvent,
+    subscriptionId: string,
+  ): Promise<void> {
+    const events = manager.getRepository(QqbotMessageEvent);
+    const priorEvents = await events.find({
+      where: { resourceKey: event.resourceKey, sourceKey: event.sourceKey },
+    });
+    const priorIds = new Set(
+      priorEvents
+        .filter((candidate) => this.isStrictlyEarlier(candidate, event))
+        .map((candidate) => candidate.id),
+    );
+    if (!priorIds.size) return;
+
+    const deliveries = manager.getRepository(QqbotMessageDelivery);
+    const candidates = await deliveries.find({ where: { subscriptionId } });
+    for (const delivery of candidates) {
+      if (
+        priorIds.has(delivery.messageEventId) &&
+        SUPERSEDED_STATUSES.has(delivery.status)
+      ) {
+        delivery.status = 'superseded';
+        await deliveries.save(delivery);
+      }
+    }
+  }
+
+  /**
+   * Builds every currently legal binding/target delivery with frozen snapshots.
+   * @param manager - Current isolated subscription transaction manager.
+   * @param event - Frozen Outbox event.
+   * @param subscription - Locked and rechecked active subscription.
+   * @param readiness - Ready or DDNS-waiting variables from the source adapter.
+   * @param now - Stable scheduling instant for the newly created rows.
+   */
+  private async createDeliveries(
+    manager: EntityManager,
+    event: QqbotMessageEvent,
+    subscription: QqbotMessageSubscription,
+    readiness: Extract<
+      SystemMessageDeliveryReadiness,
+      { status: 'ready' | 'waiting_ddns' }
+    >,
+    now: Date,
+  ): Promise<void> {
+    const bindings = await manager
+      .getRepository(QqbotMessagePublishBinding)
+      .find({
+        where: {
+          enabled: true,
+          isDeleted: false,
+          subscriptionId: subscription.id,
+        },
+        order: { id: 'ASC' },
+      });
+    const accounts = await manager.getRepository(QqbotAccount).find({
+      where: { enabled: true, isDeleted: false },
+    });
+    const templates = await manager.getRepository(QqbotMessageTemplate).find({
+      where: { enabled: true, isDeleted: false, sourceKey: event.sourceKey },
+    });
+    const targets = await manager
+      .getRepository(QqbotMessagePublishTarget)
+      .find({
+        where: { enabled: true, isDeleted: false },
+        order: { id: 'ASC' },
+      });
+    const accountById = new Map(
+      accounts.map((account) => [account.id, account]),
+    );
+    const templateById = new Map(
+      templates.map((template) => [template.id, template]),
+    );
+
+    for (const binding of bindings) {
+      const account = accountById.get(binding.accountId);
+      const template = templateById.get(binding.templateId);
+      if (!account || !template || account.selfId !== binding.selfId) continue;
+
+      let renderedMessage: string;
+      try {
+        renderedMessage = this.templateRenderer.render(
+          template.content,
+          readiness.variables,
+        );
+      } catch (error) {
+        if (error instanceof SystemMessageContractError) continue;
+        throw error;
+      }
+      for (const target of targets.filter(
+        (item) => item.bindingId === binding.id,
+      )) {
+        if (target.targetType !== 'group' && target.targetType !== 'private') {
+          continue;
+        }
+        await this.createDeliveryIfAbsent(
+          manager,
+          event,
+          subscription,
+          binding,
+          template,
+          target,
+          readiness,
+          renderedMessage,
+          now,
+        );
+      }
+    }
+  }
+
+  /**
+   * Persists one delivery, accepting only the exact event-target duplicate-key race.
+   * @param manager - Current isolated subscription transaction manager.
+   * @param event - Frozen source event.
+   * @param subscription - Subscription snapshot identity.
+   * @param binding - Active strict-account publishing binding.
+   * @param template - Source-compatible template whose content is frozen.
+   * @param target - Active target that defines this row's idempotency key.
+   * @param readiness - Ready or DDNS-waiting result that supplies frozen variables.
+   * @param renderedMessage - Already validated literal rendered content.
+   * @param now - Stable schedule time for the new row.
+   */
+  private async createDeliveryIfAbsent(
+    manager: EntityManager,
+    event: QqbotMessageEvent,
+    subscription: QqbotMessageSubscription,
+    binding: QqbotMessagePublishBinding,
+    template: QqbotMessageTemplate,
+    target: QqbotMessagePublishTarget,
+    readiness: Extract<
+      SystemMessageDeliveryReadiness,
+      { status: 'ready' | 'waiting_ddns' }
+    >,
+    renderedMessage: string,
+    now: Date,
+  ): Promise<void> {
+    const deliveries = manager.getRepository(QqbotMessageDelivery);
+    const key = { messageEventId: event.id, publishTargetId: target.id };
+    if (await deliveries.findOne({ where: key })) return;
+    const isWaiting = readiness.status === 'waiting_ddns';
+    const delivery = deliveries.create({
+      attemptCount: 0,
+      bindingId: binding.id,
+      expiresAt: new KtDateTime(
+        event.occurredAt.getTime() + SYSTEM_MESSAGE_RETRY_WINDOW_MS,
+      ),
+      lastErrorCode: null,
+      lastErrorMessage: null,
+      messageEventId: event.id,
+      nextAttemptAt: new KtDateTime(
+        now.getTime() + (isWaiting ? SYSTEM_MESSAGE_DDNS_RECHECK_MS : 0),
+      ),
+      processingLeaseUntil: null,
+      publishTargetId: target.id,
+      renderedMessage,
+      selfId: binding.selfId,
+      sendLogId: null,
+      status: isWaiting ? 'waiting_ddns' : 'pending',
+      subscriptionId: subscription.id,
+      targetId: target.targetId,
+      targetType: target.targetType,
+      templateContent: template.content,
+      templateId: template.id,
+      variableSnapshot: structuredClone(readiness.variables),
+    });
+    try {
+      await deliveries.save(delivery);
+    } catch (error) {
+      if (!this.isDuplicateKeyError(error)) throw error;
+      const existing = await deliveries.findOne({ where: key });
+      if (!existing) throw error;
+    }
+  }
+
+  /**
+   * Completes, retries, or fails only while the original claim still owns the event.
+   * @param token - Original attempt and exact lease ownership token.
+   * @param status - Terminal or retry fan-out state to persist.
+   * @param code - Safe stable error classification, or null for completion.
+   * @param message - Bounded safe error summary, or null for completion.
+   */
+  private async finish(
+    token: ClaimToken,
+    status: 'completed' | 'failed' | 'retry',
+    code: null | string,
+    message: null | string,
+  ): Promise<void> {
+    await this.dataSource.getRepository(QqbotMessageEvent).update(
+      {
+        fanoutAttemptCount: token.attempt,
+        fanoutLeaseUntil: token.leaseUntil,
+        fanoutStatus: 'processing',
+        id: token.event.id,
+      },
+      {
+        fanoutLeaseUntil: null,
+        fanoutStatus: status,
+        lastErrorCode: code,
+        lastErrorMessage: message,
+        nextFanoutAt: null,
+      },
+    );
+  }
+
+  /**
+   * Schedules a safe retry only when another attempt can occur before the hard deadline.
+   * @param token - Original claim ownership token.
+   * @param now - Stable clock instant used to calculate retry timing.
+   * @param code - Stable safe transient error code.
+   * @param message - Sanitized transient error summary.
+   */
+  private async retryOrFail(
+    token: ClaimToken,
+    now: Date,
+    code: string,
+    message: string,
+  ): Promise<void> {
+    const delay = Math.min(
+      SYSTEM_MESSAGE_RETRY_BASE_MS * 2 ** (token.attempt - 1),
+      SYSTEM_MESSAGE_RETRY_MAX_MS,
+    );
+    const deadline =
+      token.event.occurredAt.getTime() + SYSTEM_MESSAGE_RETRY_WINDOW_MS;
+    if (now.getTime() + delay >= deadline) {
+      await this.finish(
+        token,
+        'failed',
+        EVENT_EXPIRED_ERROR_CODE,
+        'fan-out deadline reached',
+      );
+      return;
+    }
+    await this.dataSource.getRepository(QqbotMessageEvent).update(
+      {
+        fanoutAttemptCount: token.attempt,
+        fanoutLeaseUntil: token.leaseUntil,
+        fanoutStatus: 'processing',
+        id: token.event.id,
+      },
+      {
+        fanoutLeaseUntil: null,
+        fanoutStatus: 'retry',
+        lastErrorCode: code,
+        lastErrorMessage: message,
+        nextFanoutAt: new KtDateTime(now.getTime() + delay),
+      },
+    );
+  }
+
+  /**
+   * Checks the strict current source/resource identity without coercing Snowflake strings.
+   * @param event - Event whose immutable resource key is authoritative.
+   * @param payload - Source-validated scalar payload.
+   */
+  private assertResourceIdentity(
+    event: QqbotMessageEvent,
+    payload: Record<string, SystemMessageScalar>,
+  ): void {
+    if (
+      event.sourceKey === STUN_MAPPING_PORT_SOURCE &&
+      payload.portForwardId !== event.resourceKey
+    ) {
+      throw new EventResourceMismatchError();
+    }
+  }
+
+  /**
+   * Checks whether a subscription remains active and has an own string resource ID match.
+   * @param subscription - Candidate subscription row.
+   * @param event - Current immutable Outbox event.
+   * @returns Whether this row is a strict fan-out match.
+   */
+  private matchesSubscription(
+    subscription: QqbotMessageSubscription,
+    event: QqbotMessageEvent,
+  ): boolean {
+    const config = subscription.sourceConfig;
+    return (
+      subscription.enabled &&
+      !subscription.isDeleted &&
+      subscription.sourceKey === event.sourceKey &&
+      !!config &&
+      Object.prototype.hasOwnProperty.call(config, 'portForwardId') &&
+      typeof config.portForwardId === 'string' &&
+      config.portForwardId === event.resourceKey
+    );
+  }
+
+  /**
+   * Compares source events by their required occurrence-time then string-ID ordering.
+   * @param candidate - Potential historical event.
+   * @param current - Event whose older deliveries may be superseded.
+   * @returns Whether candidate is strictly earlier than current.
+   */
+  private isStrictlyEarlier(
+    candidate: QqbotMessageEvent,
+    current: QqbotMessageEvent,
+  ): boolean {
+    const difference =
+      candidate.occurredAt.getTime() - current.occurredAt.getTime();
+    return (
+      difference < 0 ||
+      (difference === 0 && BigInt(candidate.id) < BigInt(current.id))
+    );
+  }
+
+  /**
+   * Checks whether creating a new delivery would violate the occurrence-based fan-out deadline.
+   * @param event - Claimed Outbox event.
+   * @param now - Stable invocation clock.
+   * @returns Whether the event is at or beyond its 24-hour deadline.
+   */
+  private isExpired(event: QqbotMessageEvent, now: Date): boolean {
+    return (
+      now.getTime() >=
+      event.occurredAt.getTime() + SYSTEM_MESSAGE_RETRY_WINDOW_MS
+    );
+  }
+
+  /**
+   * Narrows the source union to outcomes that provide variables required by delivery rows.
+   * @param readiness - Source adapter result for the locked subscription.
+   * @returns Whether a frozen ready or DDNS-waiting delivery can be persisted.
+   */
+  private hasRenderableVariables(
+    readiness: SystemMessageDeliveryReadiness,
+  ): readiness is Extract<
+    SystemMessageDeliveryReadiness,
+    { status: 'ready' | 'waiting_ddns' }
+  > {
+    return readiness.status === 'ready' || readiness.status === 'waiting_ddns';
+  }
+
+  /**
+   * Recognizes MySQL's duplicate-key signal without treating other writes as idempotent.
+   * @param error - Unknown persistence failure from an individual delivery insert.
+   * @returns Whether the error is a MySQL duplicate-key failure.
+   */
+  private isDuplicateKeyError(error: unknown): boolean {
+    if (!error || typeof error !== 'object') return false;
+    const record = error as { code?: unknown; errno?: unknown };
+    return record.code === 'ER_DUP_ENTRY' || record.errno === 1062;
+  }
+
+  /**
+   * Converts a domain error into a bounded, non-stack persistence summary.
+   * @param error - Domain contract error whose stable code is safe to retain.
+   * @returns At most 500 characters of non-sensitive message text.
+   */
+  private safeMessage(error: SystemMessageContractError): string {
+    return error.message.slice(0, 500);
+  }
+}
+
+/** Signals frozen STUN payload identity that conflicts with the immutable event resource. */
+class EventResourceMismatchError extends Error {
+  /** Creates the stable permanent event/resource mismatch classification. */
+  constructor() {
+    super('validated port-forward identity does not match event resource');
+  }
+}
diff --git a/src/modules/qqbot/core/application/message-push/system-message-runner.constants.ts b/src/modules/qqbot/core/application/message-push/system-message-runner.constants.ts
new file mode 100644
index 0000000..3001018
--- /dev/null
+++ b/src/modules/qqbot/core/application/message-push/system-message-runner.constants.ts
@@ -0,0 +1,14 @@
+/** Maximum due Outbox rows claimed by one fan-out scan. */
+export const SYSTEM_MESSAGE_BATCH_SIZE = 50;
+/** Duration of an exclusive fan-out claim lease. */
+export const SYSTEM_MESSAGE_LEASE_MS = 30_000;
+/** Initial fan-out retry delay. */
+export const SYSTEM_MESSAGE_RETRY_BASE_MS = 10_000;
+/** Maximum fan-out retry delay. */
+export const SYSTEM_MESSAGE_RETRY_MAX_MS = 15 * 60_000;
+/** Maximum age of an Outbox event that may create new deliveries. */
+export const SYSTEM_MESSAGE_RETRY_WINDOW_MS = 24 * 60 * 60_000;
+/** Persistent DDNS recheck interval for newly frozen waiting deliveries. */
+export const SYSTEM_MESSAGE_DDNS_RECHECK_MS = 60_000;
+/** Future coordinator scan cadence. */
+export const SYSTEM_MESSAGE_SCAN_INTERVAL_MS = 5_000;
diff --git a/src/modules/qqbot/core/qqbot-core.module.ts b/src/modules/qqbot/core/qqbot-core.module.ts
index d4c6181..2cb03d4 100644
--- a/src/modules/qqbot/core/qqbot-core.module.ts
+++ b/src/modules/qqbot/core/qqbot-core.module.ts
@@ -49,6 +49,7 @@ import { QqbotMessageSubscription } from '@/modules/qqbot/core/infrastructure/pe
 import { QqbotMessageTemplate } from '@/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-template.entity';
 import { SystemMessageSourceRegistry } from './application/message-push/system-message-source.registry';
 import { SystemMessageEventStagerService } from './application/message-push/system-message-event-stager.service';
+import { SystemMessageFanoutService } from './application/message-push/system-message-fanout.service';
 import { SystemMessageTemplateRendererService } from './application/message-push/system-message-template-renderer.service';
 import { QqbotMessageSubscriptionService } from './application/message-push/qqbot-message-subscription.service';
 import { QqbotMessageTemplateService } from './application/message-push/qqbot-message-template.service';
@@ -92,6 +93,7 @@ export const QQBOT_CORE_CONTROLLERS = [
 export const QQBOT_CORE_PROVIDERS = [
   SystemMessageSourceRegistry,
   SystemMessageEventStagerService,
+  SystemMessageFanoutService,
   {
     provide: SYSTEM_MESSAGE_EVENT_STAGER,
     useExisting: SystemMessageEventStagerService,
diff --git a/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts b/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts
index a35a6ea..ac9a64d 100644
--- a/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts
+++ b/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts
@@ -15,6 +15,7 @@ import { QqbotPermissionController } from '../../../../src/modules/qqbot/core/co
 import { QqbotRuleController } from '../../../../src/modules/qqbot/core/contract/rule/qqbot-rule.controller';
 import { QqbotSendController } from '../../../../src/modules/qqbot/core/contract/send/qqbot-send.controller';
 import { SystemMessageEventStagerService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-event-stager.service';
+import { SystemMessageFanoutService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-fanout.service';
 import { SYSTEM_MESSAGE_EVENT_STAGER } from '../../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types';
 import {
   QQBOT_CORE_CONTROLLERS,
@@ -171,6 +172,7 @@ describe('QQBot core module contract', () => {
     expect(QQBOT_CORE_PROVIDERS).toEqual(
       expect.arrayContaining([
         SystemMessageEventStagerService,
+        SystemMessageFanoutService,
         {
           provide: SYSTEM_MESSAGE_EVENT_STAGER,
           useExisting: SystemMessageEventStagerService,
diff --git a/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts b/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts
new file mode 100644
index 0000000..885fa9e
--- /dev/null
+++ b/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts
@@ -0,0 +1,960 @@
+import { KtDateTime } from '../../../../src/common';
+import {
+  SystemMessageContractError,
+  type SystemMessageSourceAdapter,
+} from '../../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types';
+import { SystemMessageFanoutService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-fanout.service';
+import { SystemMessageSourceRegistry } from '../../../../src/modules/qqbot/core/application/message-push/system-message-source.registry';
+import { SystemMessageTemplateRendererService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-template-renderer.service';
+import { QqbotAccount } from '../../../../src/modules/qqbot/core/infrastructure/persistence/account/qqbot-account.entity';
+import { QqbotMessageDelivery } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-delivery.entity';
+import { QqbotMessageEvent } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-event.entity';
+import { QqbotMessagePublishBinding } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-publish-binding.entity';
+import { QqbotMessagePublishTarget } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-publish-target.entity';
+import { QqbotMessageSubscription } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-subscription.entity';
+import { QqbotMessageTemplate } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-template.entity';
+import {
+  SYSTEM_MESSAGE_BATCH_SIZE,
+  SYSTEM_MESSAGE_DDNS_RECHECK_MS,
+  SYSTEM_MESSAGE_LEASE_MS,
+  SYSTEM_MESSAGE_RETRY_BASE_MS,
+  SYSTEM_MESSAGE_RETRY_WINDOW_MS,
+} from '../../../../src/modules/qqbot/core/application/message-push/system-message-runner.constants';
+
+const NOW = new Date('2026-07-24T00:00:00.000Z');
+const SOURCE_KEY = 'network.stun.mapping-port-changed';
+const RESOURCE_KEY = '9007199254740993';
+
+type Store = {
+  accounts: QqbotAccount[];
+  bindings: QqbotMessagePublishBinding[];
+  deliveries: QqbotMessageDelivery[];
+  events: QqbotMessageEvent[];
+  subscriptions: QqbotMessageSubscription[];
+  targets: QqbotMessagePublishTarget[];
+  templates: QqbotMessageTemplate[];
+};
+
+/** Builds a consistent current event whose payload preserves string Snowflake identity. */
+function event(overrides: Partial<QqbotMessageEvent> = {}): QqbotMessageEvent {
+  return Object.assign(new QqbotMessageEvent(), {
+    fanoutAttemptCount: 0,
+    fanoutLeaseUntil: null,
+    fanoutStatus: 'accepted',
+    id: '200',
+    lastErrorCode: null,
+    lastErrorMessage: null,
+    nextFanoutAt: new KtDateTime(NOW),
+    occurredAt: new KtDateTime(NOW),
+    payload: { endpoint: 'pal.example.com:38213', portForwardId: RESOURCE_KEY },
+    resourceKey: RESOURCE_KEY,
+    sourceKey: SOURCE_KEY,
+    ...overrides,
+  });
+}
+
+/** Builds an active strict JSON-matched subscription. */
+function subscription(
+  overrides: Partial<QqbotMessageSubscription> = {},
+): QqbotMessageSubscription {
+  return Object.assign(new QqbotMessageSubscription(), {
+    enabled: true,
+    id: '300',
+    isDeleted: false,
+    sourceConfig: { portForwardId: RESOURCE_KEY },
+    sourceKey: SOURCE_KEY,
+    ...overrides,
+  });
+}
+
+/** Builds an enabled account without relying on its runtime online state. */
+function account(overrides: Partial<QqbotAccount> = {}): QqbotAccount {
+  return Object.assign(new QqbotAccount(), {
+    enabled: true,
+    id: '400',
+    isDeleted: false,
+    selfId: 'bot-a',
+    ...overrides,
+  });
+}
+
+/** Builds an active account-scoped publishing binding. */
+function binding(
+  overrides: Partial<QqbotMessagePublishBinding> = {},
+): QqbotMessagePublishBinding {
+  return Object.assign(new QqbotMessagePublishBinding(), {
+    accountId: '400',
+    enabled: true,
+    id: '500',
+    isDeleted: false,
+    selfId: 'bot-a',
+    subscriptionId: '300',
+    templateId: '600',
+    ...overrides,
+  });
+}
+
+/** Builds an active source-compatible literal template. */
+function template(
+  overrides: Partial<QqbotMessageTemplate> = {},
+): QqbotMessageTemplate {
+  return Object.assign(new QqbotMessageTemplate(), {
+    content: 'endpoint=${{endpoint}}',
+    enabled: true,
+    id: '600',
+    isDeleted: false,
+    sourceKey: SOURCE_KEY,
+    ...overrides,
+  });
+}
+
+/** Builds one active group/private target under a binding. */
+function target(
+  overrides: Partial<QqbotMessagePublishTarget> = {},
+): QqbotMessagePublishTarget {
+  return Object.assign(new QqbotMessagePublishTarget(), {
+    bindingId: '500',
+    enabled: true,
+    id: '700',
+    isDeleted: false,
+    targetId: 'group-1',
+    targetType: 'group',
+    ...overrides,
+  });
+}
+
+/** Builds a controllable source adapter while retaining a real source registry boundary. */
+function sourceAdapter(): jest.Mocked<SystemMessageSourceAdapter> {
+  return {
+    definition: {
+      description: 'test',
+      displayName: 'test',
+      sourceKey: SOURCE_KEY,
+      subscriptionFields: [],
+      variables: [],
+      version: 1,
+    },
+    inspectSubscription: jest.fn(),
+    listSubscriptionOptions: jest.fn(),
+    normalizeSubscriptionConfig: jest.fn(),
+    resolveDelivery: jest.fn(async (_input) => {
+      void _input;
+      return {
+        reasonCode: null,
+        status: 'ready' as const,
+        variables: { endpoint: 'pal.example.com:38213' },
+      };
+    }),
+    validateEventPayload: jest.fn(
+      (payload) => payload as Record<string, boolean | null | number | string>,
+    ),
+  };
+}
+
+/** Clones transaction-visible rows while preserving Date objects and entity prototypes. */
+function cloneStore(value: Store): Store {
+  const clone = <T>(items: T[]) =>
+    items.map((item) =>
+      Object.assign(
+        Object.create(Object.getPrototypeOf(item)),
+        structuredClone(item),
+      ),
+    );
+  return {
+    accounts: clone(value.accounts),
+    bindings: clone(value.bindings),
+    deliveries: clone(value.deliveries),
+    events: clone(value.events),
+    subscriptions: clone(value.subscriptions),
+    targets: clone(value.targets),
+    templates: clone(value.templates),
+  };
+}
+
+/** Matches the intentionally small TypeORM `where` subset used by the fan-out service. */
+function matches(
+  row: Record<string, unknown>,
+  where: Record<string, unknown> = {},
+): boolean {
+  return Object.entries(where).every(([key, value]) => {
+    const actual = row[key];
+    if (
+      actual &&
+      value &&
+      typeof actual === 'object' &&
+      typeof value === 'object' &&
+      'getTime' in actual &&
+      'getTime' in value &&
+      typeof actual.getTime === 'function' &&
+      typeof value.getTime === 'function'
+    )
+      return actual.getTime() === value.getTime();
+    return actual === value;
+  });
+}
+
+/** Creates a transaction-faithful in-memory persistence harness rather than fan-out logic. */
+function setup(seed: Partial<Store> = {}) {
+  let state: Store = {
+    accounts: seed.accounts ?? [account()],
+    bindings: seed.bindings ?? [binding()],
+    deliveries: seed.deliveries ?? [],
+    events: seed.events ?? [event()],
+    subscriptions: seed.subscriptions ?? [subscription()],
+    targets: seed.targets ?? [target()],
+    templates: seed.templates ?? [template()],
+  };
+  const adapter = sourceAdapter();
+  const registry = new SystemMessageSourceRegistry();
+  registry.register(adapter);
+  const query = {
+    brackets: false,
+    lock: '',
+    onLocked: '',
+    order: [] as string[],
+    take: 0,
+  };
+  const transactions: string[] = [];
+  const locked = new Set<string>();
+  let deliverySequence = 1000;
+  let duplicateRace: 'exact' | 'wrong' | null = null;
+  let failSubscription: null | string = null;
+
+  /** Returns a repository facade bound to exactly one authoritative or transaction-local store. */
+  const repository = (entity: unknown, store: Store) => {
+    const rows = (entity === QqbotMessageEvent
+      ? store.events
+      : entity === QqbotMessageSubscription
+        ? store.subscriptions
+        : entity === QqbotMessagePublishBinding
+          ? store.bindings
+          : entity === QqbotAccount
+            ? store.accounts
+            : entity === QqbotMessageTemplate
+              ? store.templates
+              : entity === QqbotMessagePublishTarget
+                ? store.targets
+                : store.deliveries) as unknown as Array<
+      Record<string, unknown>
+    >;
+    return {
+      create: (input: object) =>
+        Object.assign(new (entity as new () => object)(), input),
+      createQueryBuilder: () => {
+        const builder = {
+          addOrderBy: (field: string) => {
+            query.order.push(field);
+            return builder;
+          },
+          getOne: async () =>
+            rows
+              .filter((row) => {
+                const item = row as unknown as QqbotMessageEvent;
+                return (
+                  ((item.fanoutStatus === 'accepted' ||
+                    item.fanoutStatus === 'retry') &&
+                    !!item.nextFanoutAt &&
+                    item.nextFanoutAt.getTime() <= NOW.getTime()) ||
+                  (item.fanoutStatus === 'processing' &&
+                    !!item.fanoutLeaseUntil &&
+                    item.fanoutLeaseUntil.getTime() <= NOW.getTime())
+                );
+              })
+              .filter(
+                (row) => !locked.has((row as unknown as QqbotMessageEvent).id),
+              )
+              .sort((left, right) => {
+                const a = left as unknown as QqbotMessageEvent;
+                const b = right as unknown as QqbotMessageEvent;
+                return (
+                  a.occurredAt.getTime() - b.occurredAt.getTime() ||
+                  (BigInt(a.id) < BigInt(b.id)
+                    ? -1
+                    : BigInt(a.id) > BigInt(b.id)
+                      ? 1
+                      : 0)
+                );
+              })[0] ?? null,
+          orderBy: (field: string) => {
+            query.order.push(field);
+            return builder;
+          },
+          setLock: (value: string) => {
+            query.lock = value;
+            return builder;
+          },
+          setOnLocked: (value: string) => {
+            query.onLocked = value;
+            return builder;
+          },
+          take: (value: number) => {
+            query.take = value;
+            return builder;
+          },
+          where: (value: unknown) => {
+            query.brackets = value.constructor.name === 'Brackets';
+            return builder;
+          },
+        };
+        return builder;
+      },
+      find: async ({
+        order,
+        where,
+      }: { order?: { id?: 'ASC' }; where?: Record<string, unknown> } = {}) => {
+        const result = rows.filter((row) => matches(row, where));
+        return order?.id
+          ? result.sort((left, right) =>
+              `${(left as { id: string }).id}`.localeCompare(
+                `${(right as { id: string }).id}`,
+              ),
+            )
+          : result;
+      },
+      findOne: async ({ where }: { where: Record<string, unknown> }) =>
+        rows.find((row) => matches(row, where)) ?? null,
+      save: async (item: Record<string, unknown>) => {
+        if (
+          entity === QqbotMessageDelivery &&
+          failSubscription === item.bindingId
+        )
+          throw new Error('repository offline');
+        if (entity === QqbotMessageDelivery) {
+          const pair = rows.find((row) =>
+            matches(row, {
+              messageEventId: item.messageEventId,
+              publishTargetId: item.publishTargetId,
+            }),
+          );
+          if (!item.id && duplicateRace) {
+            if (duplicateRace === 'exact')
+              rows.push(
+                Object.assign(new QqbotMessageDelivery(), item, {
+                  id: 'race',
+                }) as unknown as Record<string, unknown>,
+              );
+            duplicateRace = null;
+            throw { errno: 1062 };
+          }
+          if (pair && pair !== item) throw { errno: 1062 };
+          if (!item.id) item.id = `${deliverySequence++}`;
+        }
+        const index = rows.findIndex(
+          (row) => (row as { id?: string }).id === item.id,
+        );
+        if (index >= 0) Object.assign(rows[index] as object, item);
+        else rows.push(item as never);
+        return item;
+      },
+      update: async (
+        where: Record<string, unknown>,
+        values: Record<string, unknown>,
+      ) => {
+        const row = rows.find((candidate) => matches(candidate, where));
+        if (!row) return { affected: 0 };
+        Object.assign(row as object, values);
+        return { affected: 1 };
+      },
+    };
+  };
+  const dataSource = {
+    getRepository: (entity: unknown) => repository(entity, state),
+    transaction: async (
+      callback: (manager: {
+        getRepository: (entity: unknown) => ReturnType<typeof repository>;
+      }) => Promise<unknown>,
+    ) => {
+      const draft = cloneStore(state);
+      transactions.push('begin');
+      try {
+        const result = await callback({
+          getRepository: (entity) => repository(entity, draft),
+        });
+        state = draft;
+        transactions.push('commit');
+        return result;
+      } catch (error) {
+        transactions.push('rollback');
+        throw error;
+      }
+    },
+  };
+  const service = new SystemMessageFanoutService(
+    dataSource as never,
+    registry,
+    new SystemMessageTemplateRendererService(),
+  );
+  return {
+    adapter,
+    events: () => state.events,
+    deliveries: () => state.deliveries,
+    failSubscription: (id: null | string) => {
+      failSubscription = id;
+    },
+    lock: (id: string) => locked.add(id),
+    query,
+    setDuplicateRace: (value: 'exact' | 'wrong' | null) => {
+      duplicateRace = value;
+    },
+    service,
+    transactions,
+  };
+}
+
+describe('SystemMessageFanoutService', () => {
+  it('creates exact multi-account target cardinality and freezes strict snapshots', async () => {
+    const fixture = setup({
+      accounts: [account(), account({ id: '401', selfId: 'bot-b' })],
+      bindings: [
+        binding(),
+        binding({ accountId: '401', id: '501', selfId: 'bot-b' }),
+      ],
+      targets: [
+        target(),
+        target({ id: '701', targetId: 'private-1', targetType: 'private' }),
+        target({ bindingId: '501', id: '702', targetId: 'group-2' }),
+      ],
+    });
+
+    await expect(fixture.service.runOnce(NOW)).resolves.toBe(1);
+    expect(fixture.deliveries()).toHaveLength(3);
+    expect(
+      new Set(fixture.deliveries().map((item) => item.publishTargetId)).size,
+    ).toBe(3);
+    expect(fixture.deliveries()[0]).toMatchObject({
+      attemptCount: 0,
+      bindingId: '500',
+      renderedMessage: 'endpoint=pal.example.com:38213',
+      selfId: 'bot-a',
+      status: 'pending',
+      subscriptionId: '300',
+      templateContent: 'endpoint=${{endpoint}}',
+      templateId: '600',
+      variableSnapshot: { endpoint: 'pal.example.com:38213' },
+    });
+    expect(fixture.deliveries()[0].expiresAt.getTime()).toBe(
+      NOW.getTime() + SYSTEM_MESSAGE_RETRY_WINDOW_MS,
+    );
+  });
+
+  it('schedules ready and waiting-DDNS snapshots without sending OneBot', async () => {
+    const fixture = setup();
+    fixture.adapter.resolveDelivery.mockResolvedValueOnce({
+      reasonCode: 'ddns_not_synced',
+      status: 'waiting_ddns',
+      variables: { endpoint: 'wait.example:1' },
+    });
+
+    await fixture.service.runOnce(NOW);
+    expect(fixture.deliveries()[0]).toMatchObject({
+      renderedMessage: 'endpoint=wait.example:1',
+      status: 'waiting_ddns',
+    });
+    expect(fixture.deliveries()[0].nextAttemptAt.getTime()).toBe(
+      NOW.getTime() + SYSTEM_MESSAGE_DDNS_RECHECK_MS,
+    );
+    expect(fixture.events()[0].fanoutStatus).toBe('completed');
+  });
+
+  it('requires an active own string JSON resource match before resolving a subscription', async () => {
+    const inherited = Object.create({ portForwardId: RESOURCE_KEY }) as Record<
+      string,
+      unknown
+    >;
+    const fixture = setup({
+      subscriptions: [
+        subscription(),
+        subscription({
+          id: '301',
+          sourceConfig: { portForwardId: Number(RESOURCE_KEY) },
+        }),
+        subscription({ id: '302', sourceConfig: inherited }),
+        subscription({ enabled: false, id: '303' }),
+        subscription({ id: '304', isDeleted: true }),
+      ],
+    });
+
+    await fixture.service.runOnce(NOW);
+    expect(fixture.adapter.resolveDelivery).toHaveBeenCalledTimes(1);
+  });
+
+  it('isolates disabled/deleted or mismatched account, binding, template and target rows', async () => {
+    const fixture = setup({
+      accounts: [
+        account(),
+        account({ enabled: false, id: '401', selfId: 'bad' }),
+      ],
+      bindings: [
+        binding(),
+        binding({ accountId: '401', id: '501', selfId: 'bad' }),
+        binding({ enabled: false, id: '502' }),
+        binding({ id: '503', selfId: 'wrong' }),
+      ],
+      targets: [
+        target(),
+        target({ bindingId: '501', id: '701' }),
+        target({ bindingId: '500', enabled: false, id: '702' }),
+        target({ bindingId: '500', id: '703', isDeleted: true }),
+      ],
+    });
+
+    await fixture.service.runOnce(NOW);
+    expect(fixture.deliveries().map((item) => item.publishTargetId)).toEqual([
+      '700',
+    ]);
+  });
+
+  it.each(['cancelled', 'superseded'] as const)(
+    'completes %s readiness without current delivery or render',
+    async (status) => {
+      const fixture = setup();
+      fixture.adapter.resolveDelivery.mockResolvedValueOnce({
+        reasonCode: status,
+        status,
+      });
+
+      await fixture.service.runOnce(NOW);
+      expect(fixture.deliveries()).toHaveLength(0);
+      expect(fixture.events()[0].fanoutStatus).toBe('completed');
+    },
+  );
+
+  it('supersedes only strictly earlier unfinished deliveries for the same subscription', async () => {
+    const older = event({
+      id: '100',
+      occurredAt: new KtDateTime(NOW.getTime() - 1),
+    });
+    const newer = event({
+      id: '201',
+      occurredAt: new KtDateTime(NOW.getTime() + 1),
+      nextFanoutAt: new KtDateTime(NOW.getTime() + 1),
+    });
+    const makeDelivery = (
+      id: string,
+      messageEventId: string,
+      status: QqbotMessageDelivery['status'],
+    ) =>
+      Object.assign(new QqbotMessageDelivery(), {
+        id,
+        messageEventId,
+        publishTargetId: id,
+        status,
+        subscriptionId: '300',
+      });
+    const fixture = setup({
+      events: [older, event(), newer],
+      deliveries: [
+        makeDelivery('1', '100', 'pending'),
+        makeDelivery('2', '100', 'retry'),
+        makeDelivery('3', '100', 'processing'),
+        makeDelivery('4', '200', 'pending'),
+        makeDelivery('5', '201', 'pending'),
+      ],
+    });
+
+    await fixture.service.runOnce(NOW);
+    expect(
+      fixture
+        .deliveries()
+        .filter((item) => item.messageEventId === '100')
+        .map((item) => item.status),
+    ).toEqual(['superseded', 'superseded', 'processing', 'superseded']);
+    expect(
+      fixture.deliveries().find((item) => item.messageEventId === '201')
+        ?.status,
+    ).toBe('pending');
+    expect(
+      fixture.deliveries().find((item) => item.messageEventId === '200')
+        ?.status,
+    ).toBe('pending');
+  });
+
+  it('accepts exact event-target replay and verifies only an exact duplicate-key race', async () => {
+    const existing = Object.assign(new QqbotMessageDelivery(), {
+      id: 'old',
+      messageEventId: '200',
+      publishTargetId: '700',
+      status: 'pending',
+    });
+    const replay = setup({ deliveries: [existing] });
+    await replay.service.runOnce(NOW);
+    expect(replay.deliveries()).toHaveLength(1);
+
+    const race = setup();
+    race.setDuplicateRace('exact');
+    await race.service.runOnce(NOW);
+    expect(race.deliveries()).toHaveLength(1);
+
+    const wrong = setup();
+    wrong.setDuplicateRace('wrong');
+    await wrong.service.runOnce(NOW);
+    expect(wrong.events()[0]).toMatchObject({
+      fanoutStatus: 'retry',
+      lastErrorCode: 'fanout_transient_error',
+    });
+  });
+
+  it('records bounded SQL claim contract including top-level Brackets, locking, ordering and limit', async () => {
+    const fixture = setup({ events: [] });
+    await fixture.service.runOnce(NOW);
+    expect(fixture.query).toMatchObject({
+      brackets: true,
+      lock: 'pessimistic_write',
+      onLocked: 'skip_locked',
+      take: 1,
+    });
+    expect(fixture.query.order).toEqual(['event.occurredAt', 'event.id']);
+  });
+
+  it('skips a locked oldest due event and reclaims expired processing with an exact new lease', async () => {
+    const oldest = event({
+      id: '100',
+      occurredAt: new KtDateTime(NOW.getTime() - 2),
+    });
+    const expired = event({
+      fanoutAttemptCount: 4,
+      fanoutLeaseUntil: new KtDateTime(NOW.getTime() - 1),
+      fanoutStatus: 'processing',
+      id: '101',
+      nextFanoutAt: null,
+      occurredAt: new KtDateTime(NOW.getTime() - 1),
+    });
+    const fixture = setup({ events: [oldest, expired] });
+    fixture.lock('100');
+
+    await fixture.service.runOnce(NOW);
+    expect(fixture.events().find((item) => item.id === '101')).toMatchObject({
+      fanoutAttemptCount: 5,
+      fanoutStatus: 'completed',
+    });
+    expect(fixture.events().find((item) => item.id === '100')).toMatchObject({
+      fanoutStatus: 'accepted',
+    });
+  });
+
+  it('uses a lease-and-attempt ownership fence so stale completion cannot clobber a new owner', async () => {
+    const fixture = setup({ events: [] });
+    const staleEvent = event({
+      fanoutAttemptCount: 1,
+      fanoutLeaseUntil: new KtDateTime(NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS),
+      fanoutStatus: 'processing',
+    });
+    const current = event({
+      fanoutAttemptCount: 2,
+      fanoutLeaseUntil: new KtDateTime(
+        NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS + 1,
+      ),
+      fanoutStatus: 'processing',
+    });
+    const finish = (
+      fixture.service as unknown as {
+        finish: (
+          token: object,
+          status: 'completed',
+          code: null,
+          message: null,
+        ) => Promise<void>;
+      }
+    ).finish;
+    (fixture as unknown as { events: () => QqbotMessageEvent[] })
+      .events()
+      .push(current);
+
+    await finish.call(
+      fixture.service,
+      {
+        attempt: 1,
+        event: staleEvent,
+        leaseUntil: staleEvent.fanoutLeaseUntil,
+      },
+      'completed',
+      null,
+      null,
+    );
+    expect(fixture.events()[0].fanoutStatus).toBe('processing');
+  });
+
+  it('rolls back one failing subscription, retains valid work, then retries and recovers idempotently', async () => {
+    const fixture = setup({
+      subscriptions: [subscription(), subscription({ id: '301' })],
+      bindings: [binding(), binding({ id: '501', subscriptionId: '301' })],
+      targets: [target(), target({ bindingId: '501', id: '701' })],
+    });
+    fixture.failSubscription('501');
+    await fixture.service.runOnce(NOW);
+    expect(fixture.deliveries().map((item) => item.publishTargetId)).toEqual([
+      '700',
+    ]);
+    expect(fixture.events()[0]).toMatchObject({
+      fanoutStatus: 'retry',
+      nextFanoutAt: new KtDateTime(
+        NOW.getTime() + SYSTEM_MESSAGE_RETRY_BASE_MS,
+      ),
+    });
+    fixture.failSubscription(null);
+    fixture.events()[0].nextFanoutAt = new KtDateTime(NOW);
+    await fixture.service.runOnce(NOW);
+    expect(
+      new Set(fixture.deliveries().map((item) => item.publishTargetId)),
+    ).toEqual(new Set(['700', '701']));
+    expect(fixture.events()[0].fanoutStatus).toBe('completed');
+    expect(fixture.transactions).toContain('rollback');
+  });
+
+  it('fails at the occurrence deadline and permanently rejects malformed source facts', async () => {
+    const deadline = setup({
+      events: [
+        event({
+          occurredAt: new KtDateTime(
+            NOW.getTime() - SYSTEM_MESSAGE_RETRY_WINDOW_MS,
+          ),
+        }),
+      ],
+    });
+    await deadline.service.runOnce(NOW);
+    expect(deadline.events()[0]).toMatchObject({
+      fanoutStatus: 'failed',
+      lastErrorCode: 'fanout_expired',
+    });
+    expect(deadline.deliveries()).toHaveLength(0);
+
+    const malformed = setup({
+      events: [event({ payload: { portForwardId: 'other' } })],
+    });
+    await malformed.service.runOnce(NOW);
+    expect(malformed.events()[0]).toMatchObject({
+      fanoutStatus: 'failed',
+      lastErrorCode: 'event_resource_mismatch',
+    });
+  });
+
+  it('bounds a pass to fifty oldest claims and leaves not-due rows untouched', async () => {
+    const events = Array.from(
+      { length: SYSTEM_MESSAGE_BATCH_SIZE + 1 },
+      (_, index) =>
+        event({
+          id: `${1000 + index}`,
+          occurredAt: new KtDateTime(
+            NOW.getTime() - SYSTEM_MESSAGE_BATCH_SIZE + index,
+          ),
+        }),
+    );
+    const fixture = setup({ events });
+    await expect(fixture.service.runOnce(NOW)).resolves.toBe(
+      SYSTEM_MESSAGE_BATCH_SIZE,
+    );
+    expect(
+      fixture.events().filter((item) => item.fanoutStatus === 'completed'),
+    ).toHaveLength(SYSTEM_MESSAGE_BATCH_SIZE);
+    expect(fixture.events().at(-1)?.fanoutStatus).toBe('accepted');
+  });
+
+  it('marks source contract errors as permanent failures rather than transient adapter retries', async () => {
+    const fixture = setup();
+    fixture.adapter.validateEventPayload.mockImplementation(() => {
+      throw new SystemMessageContractError('payload_invalid');
+    });
+    await fixture.service.runOnce(NOW);
+    expect(fixture.events()[0]).toMatchObject({
+      fanoutStatus: 'failed',
+      lastErrorCode: 'payload_invalid',
+    });
+  });
+
+  it('completes a matched event with no legal targets without resolving a default account', async () => {
+    const fixture = setup({ targets: [] });
+    await fixture.service.runOnce(NOW);
+    expect(fixture.deliveries()).toHaveLength(0);
+    expect(fixture.events()[0].fanoutStatus).toBe('completed');
+  });
+
+  it('does not resolve when no subscription matches the frozen source/resource', async () => {
+    const fixture = setup({
+      subscriptions: [
+        subscription({ sourceConfig: { portForwardId: 'other' } }),
+      ],
+    });
+    await fixture.service.runOnce(NOW);
+    expect(fixture.adapter.resolveDelivery).not.toHaveBeenCalled();
+    expect(fixture.events()[0].fanoutStatus).toBe('completed');
+  });
+
+  it('leaves future retries and live processing leases unclaimed', async () => {
+    const fixture = setup({
+      events: [
+        event({
+          fanoutStatus: 'retry',
+          nextFanoutAt: new KtDateTime(NOW.getTime() + 1),
+        }),
+        event({
+          fanoutLeaseUntil: new KtDateTime(NOW.getTime() + 1),
+          fanoutStatus: 'processing',
+          id: '201',
+          nextFanoutAt: null,
+        }),
+      ],
+    });
+    await expect(fixture.service.runOnce(NOW)).resolves.toBe(0);
+    expect(fixture.events().map((item) => item.fanoutStatus)).toEqual([
+      'retry',
+      'processing',
+    ]);
+  });
+
+  it('keeps a runtime-offline but enabled strict account eligible at fan-out', async () => {
+    const fixture = setup({
+      accounts: [
+        account({ connectStatus: 'offline', oneBotStatus: 'offline' }),
+      ],
+    });
+    await fixture.service.runOnce(NOW);
+    expect(fixture.deliveries()).toHaveLength(1);
+  });
+
+  it('isolates an invalid template render while allowing another legal binding', async () => {
+    const fixture = setup({
+      bindings: [binding(), binding({ id: '501', templateId: '601' })],
+      templates: [
+        template({ content: '${{unknown}}' }),
+        template({ id: '601' }),
+      ],
+      targets: [target(), target({ bindingId: '501', id: '701' })],
+    });
+    await fixture.service.runOnce(NOW);
+    expect(fixture.deliveries().map((item) => item.publishTargetId)).toEqual([
+      '701',
+    ]);
+  });
+
+  it('does not supersede current-event rows during an idempotent replay', async () => {
+    const fixture = setup({
+      deliveries: [
+        Object.assign(new QqbotMessageDelivery(), {
+          id: 'old',
+          messageEventId: '200',
+          publishTargetId: '700',
+          status: 'retry',
+          subscriptionId: '300',
+        }),
+      ],
+    });
+    await fixture.service.runOnce(NOW);
+    expect(fixture.deliveries()[0].status).toBe('retry');
+  });
+
+  it('uses numeric BIGINT ordering when occurrence times are equal', async () => {
+    const older = event({ id: '9', occurredAt: new KtDateTime(NOW) });
+    const current = event({ id: '10', occurredAt: new KtDateTime(NOW) });
+    const fixture = setup({
+      events: [current, older],
+      deliveries: [
+        Object.assign(new QqbotMessageDelivery(), {
+          id: 'old',
+          messageEventId: '9',
+          publishTargetId: 'old-target',
+          status: 'pending',
+          subscriptionId: '300',
+        }),
+      ],
+    });
+    await fixture.service.runOnce(NOW);
+    expect(fixture.deliveries()[0].status).toBe('superseded');
+  });
+
+  it('marks a transient event failure retryable with the one-based ten-second delay', async () => {
+    const fixture = setup();
+    fixture.adapter.resolveDelivery.mockRejectedValueOnce(
+      new Error('adapter unavailable'),
+    );
+    await fixture.service.runOnce(NOW);
+    expect(fixture.events()[0]).toMatchObject({
+      fanoutStatus: 'retry',
+      lastErrorCode: 'fanout_transient_error',
+    });
+    expect(fixture.events()[0].nextFanoutAt.getTime()).toBe(
+      NOW.getTime() + SYSTEM_MESSAGE_RETRY_BASE_MS,
+    );
+  });
+
+  it('fails immediately when the next transient retry would fall outside the hard deadline', async () => {
+    const fixture = setup({
+      events: [
+        event({
+          occurredAt: new KtDateTime(
+            NOW.getTime() - SYSTEM_MESSAGE_RETRY_WINDOW_MS + 1,
+          ),
+        }),
+      ],
+    });
+    fixture.adapter.resolveDelivery.mockRejectedValueOnce(
+      new Error('adapter unavailable'),
+    );
+    await fixture.service.runOnce(NOW);
+    expect(fixture.events()[0]).toMatchObject({
+      fanoutStatus: 'failed',
+      lastErrorCode: 'fanout_expired',
+      nextFanoutAt: null,
+    });
+  });
+
+  it('fails an unknown source permanently without mutating subscriptions or deliveries', async () => {
+    const fixture = setup({ events: [event({ sourceKey: 'unknown.source' })] });
+    await fixture.service.runOnce(NOW);
+    expect(fixture.events()[0]).toMatchObject({
+      fanoutStatus: 'failed',
+      lastErrorCode: 'unknown_message_source',
+    });
+    expect(fixture.deliveries()).toHaveLength(0);
+  });
+
+  it('calls the adapter exactly once per matched subscription rather than once per target', async () => {
+    const fixture = setup({ targets: [target(), target({ id: '701' })] });
+    await fixture.service.runOnce(NOW);
+    expect(fixture.adapter.resolveDelivery).toHaveBeenCalledTimes(1);
+  });
+
+  it('keeps processing and terminal historical deliveries outside supersession', async () => {
+    const older = event({
+      id: '100',
+      occurredAt: new KtDateTime(NOW.getTime() - 1),
+    });
+    const statuses: QqbotMessageDelivery['status'][] = [
+      'processing',
+      'success',
+      'failed',
+      'superseded',
+      'cancelled',
+    ];
+    const fixture = setup({
+      events: [older, event()],
+      deliveries: statuses.map((status, index) =>
+        Object.assign(new QqbotMessageDelivery(), {
+          id: `${index}`,
+          messageEventId: '100',
+          publishTargetId: `${index}`,
+          status,
+          subscriptionId: '300',
+        }),
+      ),
+    });
+    await fixture.service.runOnce(NOW);
+    expect(
+      fixture
+        .deliveries()
+        .filter((item) => ['0', '1', '2', '3', '4'].includes(item.id))
+        .map((item) => item.status),
+    ).toEqual(statuses);
+  });
+
+  it('preserves snapshots after later in-memory configuration edits', async () => {
+    const fixture = setup();
+    await fixture.service.runOnce(NOW);
+    const frozen = fixture.deliveries()[0];
+    frozen.templateContent = 'endpoint=${{endpoint}}';
+    expect(frozen).toMatchObject({
+      renderedMessage: 'endpoint=pal.example.com:38213',
+      selfId: 'bot-a',
+      targetId: 'group-1',
+    });
+  });
+});