feat: 接入消息投递取消与唤醒

src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts

@@ -12,7 +12,9 @@ import * as mqtt from 'mqtt';
 import type { IClientOptions, MqttClient } from 'mqtt';
 import { KtDateTime } from '@/common';
 import {
+  SYSTEM_MESSAGE_DELIVERY_COORDINATOR,
   SYSTEM_MESSAGE_EVENT_STAGER,
+  type SystemMessageDeliveryCoordinator,
   type SystemMessageEventStager,
 } from '@/modules/qqbot/core/contract/message-push/qqbot-message-push.types';
 import { NetworkAgentState } from './network-agent-state.entity';
@@ -65,6 +67,7 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy {
    * @param dataSource - Transaction boundary for publish acknowledgements and inbound state.
    * @param eventStream - SSE fan-out notified only after accepted inbound commits.
    * @param eventStager - Core-owned Outbox port that shares endpoint-history transactions.
+   * @param deliveryCoordinator - Core-owned durable delivery wake port used only after commit.
    * @param clientFactory - Optional deterministic MQTT client factory used by tests.
    * @param ddnsService - Optional automatic-DDNS reconciler notified after address semantics commit.
    */
@@ -74,6 +77,8 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy {
     private readonly eventStream: NetworkManagementEventStreamService,
     @Inject(SYSTEM_MESSAGE_EVENT_STAGER)
     private readonly eventStager: SystemMessageEventStager,
+    @Inject(SYSTEM_MESSAGE_DELIVERY_COORDINATOR)
+    private readonly deliveryCoordinator: SystemMessageDeliveryCoordinator,
     @Optional()
     @Inject(NETWORK_MQTT_CLIENT_FACTORY)
     private readonly clientFactory?: NetworkMqttClientFactory,
@@ -200,7 +205,17 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy {
       ddnsSourceChanged = result.ddnsSourceChanged;
     } else if (topic === this.topic('events')) {
       source = 'events';
-      changed = await this.appendEndpointEvent(parseEndpointEvent(parsed));
+      const endpointResult = await this.appendEndpointEvent(
+        parseEndpointEvent(parsed),
+      );
+      changed = endpointResult.changed;
+      if (endpointResult.deliveryAccepted) {
+        try {
+          this.deliveryCoordinator.requestDrain();
+        } catch {
+          this.logger.warn('System message delivery wake failed');
+        }
+      }
     } else {
       throw new NetworkMessageValidationError('Unexpected network MQTT topic');
     }
@@ -638,11 +653,11 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy {
   /**
    * Appends one endpoint change event exactly once by event ID.
    * @param event - Strict endpoint transition parsed from the events topic.
-   * @returns True only when a new history row committed.
+   * @returns Committed history visibility plus whether the same transaction accepted Outbox work.
    */
   private async appendEndpointEvent(
     event: NetworkEndpointEvent,
-  ): Promise<boolean> {
+  ): Promise<{ changed: boolean; deliveryAccepted: boolean }> {
     this.assertAgentId(event.agentId);
     return await this.dataSource.transaction(async (manager) => {
       const state = await manager.getRepository(NetworkAgentState).findOne({
@@ -660,7 +675,7 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy {
       }
       const repository = manager.getRepository(NetworkEndpointHistory);
       if (await repository.findOne({ where: { eventId: event.eventId } })) {
-        return false;
+        return { changed: false, deliveryAccepted: false };
       }
       const previousHistory = await repository.findOne({
         lock: { mode: 'pessimistic_read' },
@@ -680,8 +695,10 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy {
       });
       try {
         await repository.save(history);
+        let deliveryAccepted = false;
         if (this.shouldStagePortChange(event, previousHistory)) {
-          await this.eventStager.stage(manager, {
+          deliveryAccepted =
+            (await this.eventStager.stage(manager, {
               eventId: event.eventId,
               occurredAt: event.occurredAt,
               payload: {
@@ -693,12 +710,12 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy {
               },
               resourceKey: event.mappingId,
               sourceKey: 'network.stun.mapping-port-changed',
-          });
+            })) === 'accepted';
         }
-        return true;
+        return { changed: true, deliveryAccepted };
       } catch (error) {
         if (!this.isDuplicateKeyError(error)) throw error;
-        return false;
+        return { changed: false, deliveryAccepted: false };
       }
     });
   }

src/modules/admin/platform-config/network-management/network-ddns.service.ts

@@ -1,7 +1,9 @@
 import { isIP } from 'node:net';
 import {
   HttpStatus,
+  Inject,
   Injectable,
+  Logger,
   type OnModuleDestroy,
   type OnModuleInit,
 } from '@nestjs/common';
@@ -9,6 +11,10 @@ import { ConfigService } from '@nestjs/config';
 import { InjectRepository } from '@nestjs/typeorm';
 import { IsNull, Repository } from 'typeorm';
 import { KtDateTime, throwVbenError } from '@/common';
+import {
+  SYSTEM_MESSAGE_DELIVERY_COORDINATOR,
+  type SystemMessageDeliveryCoordinator,
+} from '@/modules/qqbot/core/contract/message-push/qqbot-message-push.types';
 import { NetworkAgentState } from './network-agent-state.entity';
 import { NetworkDdnsRecord } from './network-ddns.entity';
 import {
@@ -72,6 +78,7 @@ const PROVIDER_ERROR_CODES: Record<string, string> = {
  */
 @Injectable()
 export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(NetworkDdnsService.name);
   private destroyed = false;
   private reconcileInterval?: NodeJS.Timeout;
   private reconcileRequestTimer?: NodeJS.Timeout;
@@ -88,6 +95,7 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy {
    * @param configService - Reconcile cadence and Agent identity configuration.
    * @param dnsPodClient - Redacted Tencent Cloud DNS provider boundary.
    * @param eventStream - Committed semantic-change notification stream.
+   * @param deliveryCoordinator - Core-owned wake port notified only after authoritative DDNS sync.
    */
   constructor(
     @InjectRepository(NetworkDdnsRecord)
@@ -99,6 +107,8 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy {
     private readonly configService: ConfigService,
     private readonly dnsPodClient: NetworkDnsPodClient,
     private readonly eventStream: NetworkManagementEventStreamService,
+    @Inject(SYSTEM_MESSAGE_DELIVERY_COORDINATOR)
+    private readonly deliveryCoordinator: SystemMessageDeliveryCoordinator,
   ) {}
 
   /**
@@ -593,11 +603,19 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy {
     current.retryCount = 0;
     current.sourceAddress = targetAddress;
     current.syncStatus = 'synced';
-    await this.saveReconcileState(
+    const saved = await this.saveReconcileState(
       current,
       beforeSynced,
       expectedProviderRecordId,
     );
+    if (saved && current.syncStatus === 'synced' && current.appliedAddress) {
+      void this.deliveryCoordinator
+        .notifyDdnsSynced({
+          appliedAddress: current.appliedAddress,
+          ddnsRecordId: String(current.id),
+        })
+        .catch(() => this.loggerWarnDeliveryWake());
+    }
   }
 
   /**
@@ -1209,6 +1227,11 @@ export class NetworkDdnsService implements OnModuleInit, OnModuleDestroy {
     }
   }
 
+  /** Logs a non-authoritative delivery wake failure without changing DDNS commit success. */
+  private loggerWarnDeliveryWake(): void {
+    this.logger.warn('System message delivery wake failed after DDNS sync');
+  }
+
   /**
    * Captures user-visible fields while excluding retry counts and timestamp-only churn.
    * @param record - Current persisted or pending entity state.

src/modules/qqbot/core/application/account/qqbot-account.service.ts

@@ -1,7 +1,7 @@
 import { Inject, Injectable, Optional } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { InjectRepository } from '@nestjs/typeorm';
-import { Repository } from 'typeorm';
+import { In, Repository, type EntityManager } from 'typeorm';
 import {
   SYSTEM_NOTICE_PUBLISHER,
   SystemNoticePublisher,
@@ -15,6 +15,8 @@ import {
 } from './qqbot-account-napcat-runtime.port';
 import { QqbotAccountAbility } from '../../infrastructure/persistence/account/qqbot-account-ability.entity';
 import { QqbotAccount } from '../../infrastructure/persistence/account/qqbot-account.entity';
+import { QqbotMessageDelivery } from '../../infrastructure/persistence/message-push/qqbot-message-delivery.entity';
+import { QqbotMessagePublishBinding } from '../../infrastructure/persistence/message-push/qqbot-message-publish-binding.entity';
 import type {
   QqbotAccountBodyDto,
   QqbotAccountQueryDto,
@@ -402,13 +404,27 @@ export class QqbotAccountService {
     if (!body.accessToken) {
       delete payload.accessToken;
     }
-    await this.accountRepository.update({ id: body.id }, payload);
+    await this.accountRepository.manager.transaction(async (manager) => {
-    if (payload.selfId) {
+      const accounts = manager.getRepository(QqbotAccount);
-      await this.accountAbilityRepository.update(
+      const current = await accounts.findOne({
-        { accountId: body.id },
+        lock: { mode: 'pessimistic_write' },
-        { selfId: payload.selfId },
+        where: { id: body.id, isDeleted: false },
-      );
+      });
+      if (!current) {
+        throwVbenError('QQBot 账号不存在或已删除');
       }
+      const selfIdChanged =
+        typeof payload.selfId === 'string' && payload.selfId !== current.selfId;
+      await accounts.update({ id: body.id }, payload);
+      if (selfIdChanged) {
+        await manager
+          .getRepository(QqbotAccountAbility)
+          .update({ accountId: body.id }, { selfId: payload.selfId! });
+      }
+      if (selfIdChanged || payload.enabled === false) {
+        await this.cancelAccountDeliveries(manager, body.id);
+      }
+    });
     return true;
   }
 
@@ -432,7 +448,16 @@ export class QqbotAccountService {
     )) || {
       deletedContainers: 0,
     };
-    await this.accountRepository.update(
+    await this.accountRepository.manager.transaction(async (manager) => {
+      const accounts = manager.getRepository(QqbotAccount);
+      const current = await accounts.findOne({
+        lock: { mode: 'pessimistic_write' },
+        where: { id, isDeleted: false },
+      });
+      if (!current) {
+        throwVbenError('QQBot 账号不存在或已删除');
+      }
+      await accounts.update(
         { id },
         {
           containerStatus: 'unknown',
@@ -444,10 +469,11 @@ export class QqbotAccountService {
           webuiStatus: 'unknown',
         },
       );
-    await this.accountAbilityRepository.update(
+      await manager
-      { accountId: id },
+        .getRepository(QqbotAccountAbility)
-      { isDeleted: true },
+        .update({ accountId: id }, { isDeleted: true });
-    );
+      await this.cancelAccountDeliveries(manager, id);
+    });
     return {
       deletedContainers: containerResult.deletedContainers,
     };
@@ -478,6 +504,27 @@ export class QqbotAccountService {
     await this.accountRepository.update({ selfId }, payload);
   }
 
+  /** Cancels only claimable rows through an account's bindings in the caller transaction. */
+  private async cancelAccountDeliveries(
+    manager: EntityManager,
+    accountId: string,
+  ): Promise<void> {
+    const bindings = await manager
+      .getRepository(QqbotMessagePublishBinding)
+      .find({
+        select: { id: true },
+        where: { accountId: String(accountId) },
+      });
+    if (!bindings.length) return;
+    await manager.getRepository(QqbotMessageDelivery).update(
+      {
+        bindingId: In(bindings.map((binding) => binding.id)),
+        status: In(['waiting_ddns', 'pending', 'retry']),
+      },
+      { nextAttemptAt: null, processingLeaseUntil: null, status: 'cancelled' },
+    );
+  }
+
   /**
    * 执行 QQBot 核心流程。
    * @param selfId - 账号 ID；定位本次读取、更新、删除或关联的账号。
@@ -634,9 +681,7 @@ export class QqbotAccountService {
    * Persists computed NapCat split statuses back to qqbot_account when the entity exposes the v3 status columns.
    * @param account - Enriched account list row; its `napcat` snapshot is the latest runtime evidence produced for Admin.
    */
-  private async syncPersistedNapcatSplitStatus(
+  private async syncPersistedNapcatSplitStatus(account: QqbotAccountListItem) {
-    account: QqbotAccountListItem,
-  ) {
     if (!this.hasPersistedNapcatSplitStatus(account)) return;
 
     const payload = this.buildNapcatSplitStatusPayload(account);

src/modules/qqbot/core/application/message-push/qqbot-account-message-push.service.ts

@@ -1,7 +1,7 @@
 import { HttpStatus, Injectable } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 import { throwVbenError } from '@/common';
-import { Repository, type EntityManager } from 'typeorm';
+import { In, Repository, type EntityManager } from 'typeorm';
 import {
   SystemMessageContractError,
   type QqbotMessagePublishBindingInput,
@@ -13,6 +13,7 @@ import {
 import { QqbotAccountService } from '../account/qqbot-account.service';
 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 { QqbotMessageDelivery } from '../../infrastructure/persistence/message-push/qqbot-message-delivery.entity';
 import { QqbotMessageSubscription } from '../../infrastructure/persistence/message-push/qqbot-message-subscription.entity';
 import { QqbotMessageTemplate } from '../../infrastructure/persistence/message-push/qqbot-message-template.entity';
 import { QqbotMessageSubscriptionService } from './qqbot-message-subscription.service';
@@ -197,6 +198,14 @@ export class QqbotAccountMessagePushService {
           );
           const saved = await bindings.save(current);
           await this.synchronizeTargets(manager, saved, normalizedTargets);
+          if (
+            !saved.enabled ||
+            snapshot!.subscriptionId !== saved.subscriptionId
+          ) {
+            await this.cancelUnfinishedDeliveries(manager, {
+              bindingId: saved.id,
+            });
+          }
           return saved;
         },
       );
@@ -240,7 +249,13 @@ export class QqbotAccountMessagePushService {
         );
         this.assertStableBindingSnapshot(current, snapshot!);
         current.enabled = enabled;
-        return bindings.save(current);
+        const saved = await bindings.save(current);
+        if (!saved.enabled) {
+          await this.cancelUnfinishedDeliveries(manager, {
+            bindingId: saved.id,
+          });
+        }
+        return saved;
       },
     );
     return this.toView(binding);
@@ -267,6 +282,7 @@ export class QqbotAccountMessagePushService {
       binding.enabled = false;
       binding.isDeleted = true;
       await bindings.save(binding);
+      await this.cancelUnfinishedDeliveries(manager, { bindingId: binding.id });
     });
     return true;
   }
@@ -362,6 +378,22 @@ export class QqbotAccountMessagePushService {
       saves.push(target);
     });
     if (saves.length > 0) await targets.save(saves);
+    const removedIds = saves
+      .filter((target) => target.isDeleted)
+      .map((target) => target.id);
+    if (removedIds.length > 0) {
+      await manager.getRepository(QqbotMessageDelivery).update(
+        {
+          publishTargetId: In(removedIds),
+          status: In(['waiting_ddns', 'pending', 'retry']),
+        },
+        {
+          nextAttemptAt: null,
+          processingLeaseUntil: null,
+          status: 'cancelled',
+        },
+      );
+    }
   }
 
   /** Builds a detached binding view using live dependency availability and active targets only. */
@@ -590,4 +622,19 @@ export class QqbotAccountMessagePushService {
     const value = error as { code?: unknown; errno?: unknown };
     return value.code === 'ER_DUP_ENTRY' || value.errno === 1062;
   }
+
+  /** Cancels only not-yet-processing deliveries in the active binding mutation transaction. */
+  private async cancelUnfinishedDeliveries(
+    manager: EntityManager,
+    where: Pick<QqbotMessageDelivery, 'bindingId'>,
+  ): Promise<void> {
+    await manager.getRepository(QqbotMessageDelivery).update(
+      { ...where, status: In(['waiting_ddns', 'pending', 'retry']) },
+      {
+        nextAttemptAt: null,
+        processingLeaseUntil: null,
+        status: 'cancelled',
+      },
+    );
+  }
 }

src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts

@@ -4,6 +4,7 @@ import { InjectRepository } from '@nestjs/typeorm';
 import { throwVbenError } from '@/common';
 import {
   Like,
+  In,
   Repository,
   type EntityManager,
   type FindOptionsWhere,
@@ -16,6 +17,7 @@ import {
 } from '../../contract/message-push/qqbot-message-push.types';
 import { QqbotMessageSubscription } from '../../infrastructure/persistence/message-push/qqbot-message-subscription.entity';
 import { QqbotMessagePublishBinding } from '../../infrastructure/persistence/message-push/qqbot-message-publish-binding.entity';
+import { QqbotMessageDelivery } from '../../infrastructure/persistence/message-push/qqbot-message-delivery.entity';
 import { SystemMessageSourceRegistry } from './system-message-source.registry';
 
 const DEFAULT_PAGE_NO = 1;
@@ -164,8 +166,17 @@ export class QqbotMessageSubscriptionService {
           if (conflict && conflict.id !== current.id) {
             this.throwNaturalKeyConflict();
           }
+          const sourceIdentityChanged =
+            current.sourceKey !== normalized.sourceKey ||
+            current.sourceConfigDigest !== normalized.sourceConfigDigest;
           Object.assign(current, normalized);
-          return repository.save(current);
+          const saved = await repository.save(current);
+          if (!saved.enabled || sourceIdentityChanged) {
+            await this.cancelUnfinishedDeliveries(manager, {
+              subscriptionId: saved.id,
+            });
+          }
+          return saved;
         },
       );
       return this.toView(saved);
@@ -202,7 +213,13 @@ export class QqbotMessageSubscriptionService {
           }
         }
         current.enabled = enabled;
-        return repository.save(current);
+        const saved = await repository.save(current);
+        if (!saved.enabled) {
+          await this.cancelUnfinishedDeliveries(manager, {
+            subscriptionId: saved.id,
+          });
+        }
+        return saved;
       },
     );
     return this.toView(saved);
@@ -224,6 +241,9 @@ export class QqbotMessageSubscriptionService {
         current.enabled = false;
         current.isDeleted = true;
         await repository.save(current);
+        await this.cancelUnfinishedDeliveries(manager, {
+          subscriptionId: current.id,
+        });
         return true;
       },
     );
@@ -345,6 +365,21 @@ export class QqbotMessageSubscriptionService {
     }
   }
 
+  /** Cancels only claimable historical deliveries in the caller's configuration transaction. */
+  private async cancelUnfinishedDeliveries(
+    manager: EntityManager,
+    where: Pick<QqbotMessageDelivery, 'subscriptionId'>,
+  ): Promise<void> {
+    await manager.getRepository(QqbotMessageDelivery).update(
+      { ...where, status: In(['waiting_ddns', 'pending', 'retry']) },
+      {
+        status: 'cancelled',
+        nextAttemptAt: null,
+        processingLeaseUntil: null,
+      },
+    );
+  }
+
   /**
    * Maps duplicate natural keys to the existing Vben-compatible HTTP conflict response.
    * @returns Never returns because it always throws an HTTP 409 exception.

src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service.ts

@@ -91,8 +91,10 @@ export class SystemMessageDeliveryCoordinatorService
   }): Promise<void> {
     if (
       this.destroyed ||
+      typeof input.appliedAddress !== 'string' ||
       isIP(input.appliedAddress) !== 4 ||
-      !input.ddnsRecordId
+      typeof input.ddnsRecordId !== 'string' ||
+      input.ddnsRecordId.length === 0
     )
       return;
     const advanced = await this.dataSource.transaction(async (manager) => {

test/admin/network-management/network-agent-mqtt.service.spec.ts

@@ -23,10 +23,12 @@ import {
 type MqttHarness = {
   client: MqttClient & EventEmitter;
   clientOptions: () => IClientOptions;
+  deliveryCoordinator: { requestDrain: jest.Mock };
   histories: NetworkEndpointHistory[];
   historyFindOne: jest.Mock;
   historySave: jest.Mock;
   mappingSave: jest.Mock;
+  operations: string[];
   publishCommitted: jest.Mock;
   mapping: NetworkPortForward;
   publishCallback: () => (error?: Error) => void;
@@ -64,6 +66,7 @@ function createDeferred<T>(): Deferred<T> {
 
 /** Creates a fake MQTT client plus in-memory TypeORM state for bridge tests. */
 function createHarness(): MqttHarness {
+  const operations: string[] = [];
   const state = Object.assign(new NetworkAgentState(), {
     agentId: 'nas-main',
     appliedRevision: '0',
@@ -150,6 +153,7 @@ function createHarness(): MqttHarness {
   const stagedEvents: SystemMessageEventInput[] = [];
   const stager = {
     stage: jest.fn(async (_manager, input) => {
+      operations.push('stager:accepted');
       stagedEvents.push(input);
       return 'accepted' as const;
     }),
@@ -159,13 +163,17 @@ function createHarness(): MqttHarness {
     getRepository: manager.getRepository.bind(manager),
     transaction: async (work) => {
       transactionCallCount += 1;
+      operations.push('transaction:start');
       const historiesBefore = [...histories];
       const stagedEventsBefore = [...stagedEvents];
       try {
-        return await work(manager);
+        const result = await work(manager);
+        operations.push('transaction:commit');
+        return result;
       } catch (error) {
         histories.splice(0, histories.length, ...historiesBefore);
         stagedEvents.splice(0, stagedEvents.length, ...stagedEventsBefore);
+        operations.push('transaction:rollback');
         throw error;
       }
     },
@@ -209,22 +217,31 @@ function createHarness(): MqttHarness {
     publishCommitted,
   } as unknown as NetworkManagementEventStreamService;
   const requestDdnsReconcile = jest.fn();
+  const deliveryCoordinator = {
+    notifyDdnsSynced: jest.fn().mockResolvedValue(undefined),
+    requestDrain: jest.fn(() => {
+      operations.push('delivery:wake');
+    }),
+  };
   const service = new NetworkAgentMqttService(
     configService,
     dataSource,
     eventStream,
     stager,
+    deliveryCoordinator,
     factory,
     { requestReconcile: requestDdnsReconcile } as never,
   );
   return {
     client,
     clientOptions: () => options,
+    deliveryCoordinator,
     histories,
     historyFindOne,
     historySave,
     mappingSave,
     mapping,
+    operations,
     publishCallback: () => publishAck,
     publishCommitted,
     requestDdnsReconcile,
@@ -389,6 +406,10 @@ function createConcurrentEndpointHarness(): ConcurrentEndpointHarness {
     dataSource,
     { publishCommitted: jest.fn() } as never,
     stager,
+    {
+      notifyDdnsSynced: jest.fn().mockResolvedValue(undefined),
+      requestDrain: jest.fn(),
+    },
   );
   return {
     historyStageEntered: firstStageEntered.promise,
@@ -1064,6 +1085,52 @@ describe('NetworkAgentMqttService', () => {
     );
     expect(harness.histories).toHaveLength(2);
     expect(harness.stagedEvents).toHaveLength(1);
+    expect(harness.deliveryCoordinator.requestDrain).toHaveBeenCalledTimes(1);
+    expect(harness.operations).toEqual([
+      'transaction:start',
+      'stager:accepted',
+      'transaction:commit',
+      'delivery:wake',
+    ]);
+  });
+
+  it('wakes accepted work only after commit and never for duplicate or non-port transitions', async () => {
+    const accepted = createHarness();
+    accepted.histories.push(endpointHistory());
+    await accepted.service.consumeMessage(
+      'kt/network/v1/agents/nas-main/events',
+      endpointEvent(),
+    );
+    expect(accepted.operations.indexOf('transaction:commit')).toBeLessThan(
+      accepted.operations.indexOf('delivery:wake'),
+    );
+
+    const duplicate = createHarness();
+    duplicate.histories.push(endpointHistory());
+    duplicate.stager.stage.mockImplementationOnce(async () => {
+      duplicate.operations.push('stager:duplicate');
+      return 'duplicate';
+    });
+    await duplicate.service.consumeMessage(
+      'kt/network/v1/agents/nas-main/events',
+      endpointEvent(),
+    );
+    expect(duplicate.deliveryCoordinator.requestDrain).not.toHaveBeenCalled();
+
+    const nonPort = createHarness();
+    nonPort.histories.push(endpointHistory());
+    await nonPort.service.consumeMessage(
+      'kt/network/v1/agents/nas-main/events',
+      endpointEvent({
+        endpoint: {
+          observedAt: '2026-07-22T01:02:04.000Z',
+          publicIpv4: '8.8.4.4',
+          publicPort: 8213,
+          validUntil: '2026-07-22T01:04:04.000Z',
+        },
+      }),
+    );
+    expect(nonPort.deliveryCoordinator.requestDrain).not.toHaveBeenCalled();
   });
 
   it.each(['published', 'withdrawn', 'restored'])(
@@ -1252,6 +1319,58 @@ describe('NetworkAgentMqttService', () => {
     ).rejects.toThrow('outbox unavailable');
     expect(harness.histories).toHaveLength(1);
     expect(harness.stagedEvents).toHaveLength(0);
+    expect(harness.deliveryCoordinator.requestDrain).not.toHaveBeenCalled();
+    expect(harness.operations).toContain('transaction:rollback');
+    expect(harness.operations).not.toContain('transaction:commit');
+  });
+
+  it('keeps a committed event ACK-successful when the post-commit wake throws synchronously', async () => {
+    const harness = createHarness();
+    harness.histories.push(endpointHistory());
+    harness.deliveryCoordinator.requestDrain.mockImplementationOnce(() => {
+      harness.operations.push('delivery:wake-throw');
+      throw new Error('wake failed');
+    });
+    const loggerWarn = jest
+      .spyOn((harness.service as any).logger, 'warn')
+      .mockImplementation();
+    const callback = jest.fn();
+
+    await (harness.service as any).acknowledgeIncoming(
+      'kt/network/v1/agents/nas-main/events',
+      endpointEvent(),
+      callback,
+    );
+
+    expect(harness.histories).toHaveLength(2);
+    expect(harness.stagedEvents).toHaveLength(1);
+    expect(harness.operations.indexOf('transaction:commit')).toBeLessThan(
+      harness.operations.indexOf('delivery:wake-throw'),
+    );
+    expect(callback).toHaveBeenCalledWith(0);
+    expect(loggerWarn).toHaveBeenCalledWith(
+      'System message delivery wake failed',
+    );
+  });
+
+  it('keeps staging failures ACK-negative and never wakes after rollback', async () => {
+    const harness = createHarness();
+    harness.histories.push(endpointHistory());
+    harness.stager.stage.mockRejectedValueOnce(new Error('outbox unavailable'));
+    const callback = jest.fn();
+
+    await (harness.service as any).acknowledgeIncoming(
+      'kt/network/v1/agents/nas-main/events',
+      endpointEvent(),
+      callback,
+    );
+
+    expect(callback).toHaveBeenCalledWith(expect.any(Error));
+    expect(callback.mock.calls[0][0]).toMatchObject({
+      message: 'outbox unavailable',
+    });
+    expect(harness.deliveryCoordinator.requestDrain).not.toHaveBeenCalled();
+    expect(harness.operations).toContain('transaction:rollback');
   });
 
   it('leaves no staged event when history persistence fails', async () => {

test/admin/network-management/network-ddns.service.spec.ts

@@ -14,6 +14,7 @@ import { NetworkPortForward } from '../../../src/modules/admin/platform-config/n
 
 type Harness = {
   client: jest.Mocked<Pick<NetworkDnsPodClient, 'getStatus' | 'reconcile'>>;
+  deliveryCoordinator: { notifyDdnsSynced: jest.Mock };
   mapping: NetworkPortForward;
   recordUpdate: jest.Mock;
   records: NetworkDdnsRecord[];
@@ -161,6 +162,10 @@ function createHarness(): Harness {
   const eventStream = {
     publishCommitted: jest.fn(),
   } as unknown as NetworkManagementEventStreamService;
+  const deliveryCoordinator = {
+    notifyDdnsSynced: jest.fn().mockResolvedValue(undefined),
+    requestDrain: jest.fn(),
+  };
   const service = new NetworkDdnsService(
     recordRepository,
     mappingRepository,
@@ -168,8 +173,32 @@ function createHarness(): Harness {
     config,
     client as unknown as NetworkDnsPodClient,
     eventStream,
+    deliveryCoordinator,
   );
-  return { client, mapping, recordUpdate, records, service, state };
+  return {
+    client,
+    deliveryCoordinator,
+    mapping,
+    recordUpdate,
+    records,
+    service,
+    state,
+  };
+}
+
+/** Persists one enabled pending IPv4 DDNS row for reconcile-path assertions. */
+async function prepareEnabledA(harness: Harness): Promise<void> {
+  await harness.service.create({
+    domain: 'kwitsukasa.top',
+    enabled: false,
+    name: 'Pal A',
+    portForwardId: '100',
+    recordType: 'A',
+    sourceType: 'port_forward_ipv4',
+    subDomain: 'pal',
+  });
+  harness.records[0].enabled = true;
+  harness.records[0].syncStatus = 'pending';
 }
 
 /** Creates the fluent list query subset used by the service. */
@@ -347,6 +376,181 @@ describe('NetworkDdnsService', () => {
       'synced',
       'synced',
     ]);
+    expect(harness.deliveryCoordinator.notifyDdnsSynced).toHaveBeenCalledTimes(
+      2,
+    );
+    expect(harness.deliveryCoordinator.notifyDdnsSynced).toHaveBeenCalledWith(
+      expect.objectContaining({
+        appliedAddress: '8.8.8.8',
+        ddnsRecordId: '200',
+      }),
+    );
+  });
+
+  it('notifies exactly once only after the final guarded synced update commits', async () => {
+    const harness = createHarness();
+    let providerStarted!: () => void;
+    let resolveProvider!: (result: {
+      appliedAddress: string;
+      changed: boolean;
+      providerRecordId: string;
+    }) => void;
+    const started = new Promise<void>((resolve) => {
+      providerStarted = resolve;
+    });
+    const providerResult = new Promise<{
+      appliedAddress: string;
+      changed: boolean;
+      providerRecordId: string;
+    }>((resolve) => {
+      resolveProvider = resolve;
+    });
+    harness.client.reconcile.mockImplementation(async () => {
+      providerStarted();
+      return providerResult;
+    });
+    await prepareEnabledA(harness);
+
+    const reconcile = harness.service.reconcileNow('200', true);
+    await started;
+    expect(harness.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled();
+    resolveProvider({
+      appliedAddress: '8.8.8.8',
+      changed: true,
+      providerRecordId: '300',
+    });
+    await reconcile;
+    await Promise.resolve();
+
+    expect(harness.recordUpdate).toHaveBeenCalledTimes(2);
+    expect(harness.records[0]).toMatchObject({
+      appliedAddress: '8.8.8.8',
+      syncStatus: 'synced',
+    });
+    expect(harness.deliveryCoordinator.notifyDdnsSynced).toHaveBeenCalledTimes(
+      1,
+    );
+    expect(harness.deliveryCoordinator.notifyDdnsSynced).toHaveBeenCalledWith({
+      appliedAddress: '8.8.8.8',
+      ddnsRecordId: '200',
+    });
+    expect(harness.recordUpdate.mock.invocationCallOrder[1]).toBeLessThan(
+      harness.deliveryCoordinator.notifyDdnsSynced.mock.invocationCallOrder[0],
+    );
+  });
+
+  it('does not notify when the final guarded synced update affects zero rows', async () => {
+    const harness = createHarness();
+    const update = harness.recordUpdate.getMockImplementation()!;
+    harness.recordUpdate
+      .mockImplementationOnce(update)
+      .mockResolvedValueOnce({ affected: 0, generatedMaps: [], raw: [] });
+    harness.client.reconcile.mockResolvedValue({
+      appliedAddress: '8.8.8.8',
+      changed: true,
+      providerRecordId: '300',
+    });
+    await prepareEnabledA(harness);
+
+    await harness.service.reconcileNow('200', true);
+    await Promise.resolve();
+
+    expect(harness.recordUpdate).toHaveBeenCalledTimes(2);
+    expect(harness.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled();
+    expect(harness.records[0].syncStatus).toBe('syncing');
+  });
+
+  it('does not notify on pre-provider stale CAS, provider failure, disabled reread, or synced no-op', async () => {
+    const stale = createHarness();
+    stale.recordUpdate.mockResolvedValueOnce({
+      affected: 0,
+      generatedMaps: [],
+      raw: [],
+    });
+    await prepareEnabledA(stale);
+    await stale.service.reconcileNow('200', true);
+    expect(stale.client.reconcile).not.toHaveBeenCalled();
+    expect(stale.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled();
+
+    const failed = createHarness();
+    failed.client.reconcile.mockRejectedValueOnce(
+      new NetworkDnsPodClientError('DNSPOD_RATE_LIMITED', 'rate limited', true),
+    );
+    await prepareEnabledA(failed);
+    await failed.service.reconcileNow('200', true);
+    expect(failed.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled();
+
+    const disabled = createHarness();
+    let providerStarted!: () => void;
+    let resolveProvider!: (result: {
+      appliedAddress: string;
+      changed: boolean;
+      providerRecordId: string;
+    }) => void;
+    const started = new Promise<void>((resolve) => {
+      providerStarted = resolve;
+    });
+    disabled.client.reconcile.mockImplementation(
+      () =>
+        new Promise((resolve) => {
+          resolveProvider = resolve;
+          providerStarted();
+        }),
+    );
+    await prepareEnabledA(disabled);
+    const disabledRun = disabled.service.reconcileNow('200', true);
+    await started;
+    disabled.records[0].enabled = false;
+    resolveProvider({
+      appliedAddress: '8.8.8.8',
+      changed: true,
+      providerRecordId: '300',
+    });
+    await disabledRun;
+    expect(
+      disabled.deliveryCoordinator.notifyDdnsSynced,
+    ).not.toHaveBeenCalled();
+
+    const synced = createHarness();
+    await prepareEnabledA(synced);
+    synced.records[0].appliedAddress = '8.8.8.8';
+    synced.records[0].providerRecordId = '300';
+    synced.records[0].sourceAddress = '8.8.8.8';
+    synced.records[0].syncStatus = 'synced';
+    await synced.service.reconcileNow('200');
+    expect(synced.client.reconcile).not.toHaveBeenCalled();
+    expect(synced.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled();
+  });
+
+  it('logs notification rejection without changing authoritative DDNS success', async () => {
+    const harness = createHarness();
+    const wakeFailure = new Error('wake unavailable');
+    harness.deliveryCoordinator.notifyDdnsSynced.mockRejectedValueOnce(
+      wakeFailure,
+    );
+    const loggerWarn = jest
+      .spyOn((harness.service as any).logger, 'warn')
+      .mockImplementation();
+    harness.client.reconcile.mockResolvedValue({
+      appliedAddress: '8.8.8.8',
+      changed: true,
+      providerRecordId: '300',
+    });
+    await prepareEnabledA(harness);
+
+    await expect(
+      harness.service.reconcileNow('200', true),
+    ).resolves.toBeUndefined();
+    await Promise.resolve();
+
+    expect(harness.records[0]).toMatchObject({
+      appliedAddress: '8.8.8.8',
+      providerRecordId: '300',
+      syncStatus: 'synced',
+    });
+    expect(loggerWarn).toHaveBeenCalledWith(
+      'System message delivery wake failed after DDNS sync',
+    );
   });
 
   it('waits without calling DNSPod when the IPv6 source is stale', async () => {
@@ -373,6 +577,7 @@ describe('NetworkDdnsService', () => {
       syncStatus: 'waiting_source',
     });
     expect(harness.client.reconcile).not.toHaveBeenCalled();
+    expect(harness.deliveryCoordinator.notifyDdnsSynced).not.toHaveBeenCalled();
   });
 
   it('waits without calling the provider when an IPv4 source keeps an ineligible residual lease', async () => {

test/modules/qqbot/message-push/qqbot-account-message-push.service.spec.ts

@@ -3,6 +3,7 @@ import { QqbotAccountMessagePushService } from '../../../../src/modules/qqbot/co
 import { SystemMessageContractError } from '../../../../src/modules/qqbot/core/contract/message-push/qqbot-message-push.types';
 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 { QqbotMessageDelivery } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-delivery.entity';
 
 const ACCOUNT = {
   id: '2041700000000000000',
@@ -150,6 +151,9 @@ function setupBindingFixture(options: BindingFixtureOptions = {}) {
     getRepository: jest.fn((entity) => {
       if (entity === QqbotMessagePublishBinding) return bindingStore;
       if (entity === QqbotMessagePublishTarget) return targetStore;
+      if (entity === QqbotMessageDelivery) {
+        return { update: jest.fn(async () => ({ affected: 0 })) };
+      }
       throw new Error('unexpected repository');
     }),
   };
@@ -250,6 +254,333 @@ describe('QqbotAccountMessagePushService', () => {
     ).toThrow(new SystemMessageContractError('invalid_target_type'));
   });
 
+  describe('binding and target delivery cancellation transaction', () => {
+    /** Builds isolated binding/target/delivery drafts and commits them only on transaction success. */
+    function cancellationHarness() {
+      const subscription = {
+        enabled: true,
+        id: '2041700000000000001',
+        isDeleted: false,
+        name: '订阅',
+        sourceConfig: {},
+        sourceKey: SOURCE_KEY,
+      };
+      const template = {
+        content: '正文',
+        enabled: true,
+        id: '2041700000000000002',
+        isDeleted: false,
+        name: '模板',
+        sourceKey: SOURCE_KEY,
+      };
+      const bindings = [bindingRow()];
+      const targets = [
+        targetRow({ id: 'target-retained' }),
+        targetRow({
+          id: 'target-removed',
+          targetId: '30000000000000001',
+          targetType: 'private',
+        }),
+      ];
+      const deliveries = [
+        ...['waiting_ddns', 'pending', 'retry', 'processing', 'success'].map(
+          (status) => ({
+            bindingId: bindings[0].id,
+            frozen: `binding-${status}`,
+            id: `binding-${status}`,
+            nextAttemptAt: `schedule-${status}`,
+            processingLeaseUntil: `lease-${status}`,
+            publishTargetId: 'target-retained',
+            status,
+          }),
+        ),
+        {
+          bindingId: bindings[0].id,
+          frozen: 'removed',
+          id: 'removed-target-delivery',
+          nextAttemptAt: 'schedule-removed',
+          processingLeaseUntil: 'lease-removed',
+          publishTargetId: 'target-removed',
+          status: 'pending',
+        },
+        {
+          bindingId: 'other-binding',
+          frozen: 'other-binding',
+          id: 'other-binding-delivery',
+          nextAttemptAt: 'schedule-other',
+          processingLeaseUntil: 'lease-other',
+          publishTargetId: 'other-target',
+          status: 'pending',
+        },
+      ];
+      const deliveryUpdates: Array<Record<string, unknown>> = [];
+      let failCancellation = false;
+      let nextTargetId = 1;
+      const transaction = jest.fn(
+        async (work: (manager: any) => Promise<unknown>) => {
+          const draftBindings = structuredClone(bindings);
+          const draftTargets = structuredClone(targets);
+          const draftDeliveries = structuredClone(deliveries);
+          const bindingStore = {
+            create: (value: Record<string, unknown>) => bindingRow(value),
+            findOne: jest.fn(
+              async ({ where }: { where: Record<string, unknown> }) =>
+                structuredClone(
+                  draftBindings.find((row) =>
+                    Object.entries(where).every(
+                      ([key, value]) => row[key] === value,
+                    ),
+                  ) ?? null,
+                ),
+            ),
+            save: jest.fn(async (row: Record<string, unknown>) => {
+              const index = draftBindings.findIndex(
+                (candidate) => candidate.id === row.id,
+              );
+              if (index >= 0) draftBindings[index] = structuredClone(row);
+              else draftBindings.push(structuredClone(row));
+              return row;
+            }),
+          };
+          const targetStore = {
+            create: (value: Record<string, unknown>) =>
+              targetRow({ id: `target-new-${nextTargetId++}`, ...value }),
+            find: jest.fn(
+              async ({ where }: { where: Record<string, unknown> }) =>
+                structuredClone(
+                  draftTargets.filter((row) =>
+                    Object.entries(where).every(
+                      ([key, value]) => row[key] === value,
+                    ),
+                  ),
+                ),
+            ),
+            save: jest.fn(async (rows: Array<Record<string, unknown>>) => {
+              rows.forEach((row) => {
+                const index = draftTargets.findIndex(
+                  (candidate) => candidate.id === row.id,
+                );
+                if (index >= 0) draftTargets[index] = structuredClone(row);
+                else draftTargets.push(structuredClone(row));
+              });
+              return rows;
+            }),
+          };
+          const manager = {
+            getRepository: (entity: unknown) => {
+              if (entity === QqbotMessagePublishBinding) return bindingStore;
+              if (entity === QqbotMessagePublishTarget) return targetStore;
+              if (entity === QqbotMessageDelivery) {
+                return {
+                  update: async (
+                    where: Record<string, any>,
+                    patch: Record<string, unknown>,
+                  ) => {
+                    deliveryUpdates.push(where);
+                    const statuses = where.status._value as string[];
+                    const targetIds = where.publishTargetId?._value as
+                      | string[]
+                      | undefined;
+                    let affected = 0;
+                    draftDeliveries.forEach((row) => {
+                      const identityMatches =
+                        (where.bindingId &&
+                          row.bindingId === where.bindingId) ||
+                        (targetIds &&
+                          targetIds.includes(String(row.publishTargetId)));
+                      if (identityMatches && statuses.includes(row.status)) {
+                        Object.assign(row, patch);
+                        affected += 1;
+                      }
+                    });
+                    if (failCancellation) {
+                      throw new Error('delivery cancellation failed');
+                    }
+                    return { affected };
+                  },
+                };
+              }
+              throw new Error('unexpected cancellation repository');
+            },
+          };
+          const result = await work(manager);
+          bindings.splice(0, bindings.length, ...draftBindings);
+          targets.splice(0, targets.length, ...draftTargets);
+          deliveries.splice(0, deliveries.length, ...draftDeliveries);
+          return result;
+        },
+      );
+      const outerBindingRepository = {
+        find: async () => bindings,
+        manager: { transaction },
+      };
+      const service = new QqbotAccountMessagePushService(
+        outerBindingRepository as never,
+        {
+          find: async ({ where }: { where: Record<string, unknown> }) =>
+            targets.filter((row) =>
+              Object.entries(where).every(([key, value]) => row[key] === value),
+            ),
+        } as never,
+        { findOne: async () => subscription } as never,
+        { findOne: async () => template } as never,
+        { findBySelfId: async () => ACCOUNT } as never,
+        {
+          requireAvailableForBinding: async (
+            _manager: unknown,
+            subscriptionId: string,
+          ) => ({ ...subscription, id: subscriptionId }),
+        } as never,
+        {
+          requireAvailableForBinding: async (
+            _manager: unknown,
+            templateId: string,
+          ) => ({ ...template, id: templateId }),
+        } as never,
+        {
+          get: () => ({
+            definition: { displayName: 'STUN 映射', variables: [] },
+            inspectSubscription: async () => ({ valid: true }),
+          }),
+        } as never,
+        { validate: jest.fn() } as never,
+      );
+      return {
+        bindings,
+        deliveries,
+        deliveryUpdates,
+        failCancellation: () => {
+          failCancellation = true;
+        },
+        service,
+        targets,
+      };
+    }
+
+    it.each([
+      [
+        'disable',
+        async (service: QqbotAccountMessagePushService) =>
+          service.setBindingEnabled(ACCOUNT.selfId, bindingRow().id, false),
+      ],
+      [
+        'remove',
+        async (service: QqbotAccountMessagePushService) =>
+          service.removeBinding(ACCOUNT.selfId, bindingRow().id),
+      ],
+      [
+        'subscription change',
+        async (service: QqbotAccountMessagePushService) =>
+          service.updateBinding(ACCOUNT.selfId, bindingRow().id, {
+            enabled: true,
+            subscriptionId: '2041700000000000099',
+            targets: [
+              { targetId: '20000000000000001', targetType: 'group' },
+              { targetId: '30000000000000001', targetType: 'private' },
+            ],
+            templateId: bindingRow().templateId,
+          }),
+      ],
+    ])(
+      '%s cancels only exact binding unfinished rows',
+      async (_name, mutate) => {
+        const harness = cancellationHarness();
+        const before = structuredClone(harness.deliveries);
+
+        await mutate(harness.service);
+
+        expect(harness.deliveries).toEqual(
+          before.map((delivery) =>
+            delivery.bindingId === bindingRow().id &&
+            ['waiting_ddns', 'pending', 'retry'].includes(delivery.status)
+              ? {
+                  ...delivery,
+                  nextAttemptAt: null,
+                  processingLeaseUntil: null,
+                  status: 'cancelled',
+                }
+              : delivery,
+          ),
+        );
+      },
+    );
+
+    it('keeps frozen deliveries on template-only switch', async () => {
+      const harness = cancellationHarness();
+      const before = structuredClone(harness.deliveries);
+
+      await harness.service.updateBinding(ACCOUNT.selfId, bindingRow().id, {
+        enabled: true,
+        subscriptionId: bindingRow().subscriptionId,
+        targets: [
+          { targetId: '20000000000000001', targetType: 'group' },
+          { targetId: '30000000000000001', targetType: 'private' },
+        ],
+        templateId: '2041700000000000098',
+      });
+
+      expect(harness.deliveries).toEqual(before);
+      expect(harness.deliveryUpdates).toHaveLength(0);
+    });
+
+    it('target synchronization cancels only removed target IDs and retains new/sibling scopes', async () => {
+      const harness = cancellationHarness();
+      const before = structuredClone(harness.deliveries);
+
+      await harness.service.updateBinding(ACCOUNT.selfId, bindingRow().id, {
+        enabled: true,
+        subscriptionId: bindingRow().subscriptionId,
+        targets: [
+          { targetId: '20000000000000001', targetType: 'group' },
+          { targetId: '40000000000000001', targetType: 'private' },
+        ],
+        templateId: bindingRow().templateId,
+      });
+
+      expect(harness.deliveries).toEqual(
+        before.map((delivery) =>
+          delivery.id === 'removed-target-delivery'
+            ? {
+                ...delivery,
+                nextAttemptAt: null,
+                processingLeaseUntil: null,
+                status: 'cancelled',
+              }
+            : delivery,
+        ),
+      );
+      expect(
+        harness.targets.find((target) => target.id === 'target-retained'),
+      ).toMatchObject({ enabled: true, isDeleted: false });
+      expect(
+        harness.targets.find((target) => target.id === 'target-new-1'),
+      ).toMatchObject({ enabled: true, isDeleted: false });
+    });
+
+    it('rolls back binding, target, and cancellation drafts on a cancellation write failure', async () => {
+      const harness = cancellationHarness();
+      const before = {
+        bindings: structuredClone(harness.bindings),
+        deliveries: structuredClone(harness.deliveries),
+        targets: structuredClone(harness.targets),
+      };
+      harness.failCancellation();
+
+      await expect(
+        harness.service.updateBinding(ACCOUNT.selfId, bindingRow().id, {
+          enabled: false,
+          subscriptionId: bindingRow().subscriptionId,
+          targets: [{ targetId: '20000000000000001', targetType: 'group' }],
+          templateId: bindingRow().templateId,
+        }),
+      ).rejects.toThrow('delivery cancellation failed');
+
+      expect(harness.bindings).toEqual(before.bindings);
+      expect(harness.targets).toEqual(before.targets);
+      expect(harness.deliveries).toEqual(before.deliveries);
+    });
+  });
+
   it('holds subscription then template gates through multi-target persistence without coercing IDs', async () => {
     const events: string[] = [];
     const targetRows: any[] = [];

test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts

@@ -13,6 +13,7 @@ import { NetworkDdnsRecord } from '../../../../src/modules/admin/platform-config
 import { NetworkPortForward } from '../../../../src/modules/admin/platform-config/network-management/network-management.entity';
 import { QqbotMessageSubscription } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-subscription.entity';
 import { QqbotMessagePublishBinding } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-publish-binding.entity';
+import { QqbotMessageDelivery } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-delivery.entity';
 
 const SOURCE_KEY = 'network.stun.mapping-port-changed';
 const NOW = new KtDateTime('2026-07-24 08:09:10');
@@ -211,6 +212,9 @@ function setup(
     getRepository: jest.fn((entity) => {
       if (entity === QqbotMessageSubscription) return subscriptionRepository;
       if (entity === QqbotMessagePublishBinding) return bindingRepository;
+      if (entity === QqbotMessageDelivery) {
+        return { update: jest.fn(async () => ({ affected: 0 })) };
+      }
       throw new Error('unexpected repository');
     }),
   } as unknown as jest.Mocked<EntityManager>;
@@ -741,6 +745,215 @@ describe('QqbotMessageSubscriptionService', () => {
     }
   });
 
+  describe('delivery cancellation transaction', () => {
+    /**
+     * Builds authoritative subscription/delivery stores whose transaction drafts commit only
+     * when the complete lifecycle callback resolves.
+     */
+    function cancellationHarness() {
+      const subscriptions = [subscription()];
+      const deliveries = [
+        ...[
+          'waiting_ddns',
+          'pending',
+          'retry',
+          'processing',
+          'success',
+          'failed',
+          'superseded',
+          'cancelled',
+        ].map((status, index) => ({
+          frozen: `exact-${status}`,
+          id: `delivery-${index}`,
+          nextAttemptAt: `schedule-${status}`,
+          processingLeaseUntil: `lease-${status}`,
+          status,
+          subscriptionId: '100',
+        })),
+        {
+          frozen: 'other',
+          id: 'delivery-other',
+          nextAttemptAt: 'schedule-other',
+          processingLeaseUntil: 'lease-other',
+          status: 'pending',
+          subscriptionId: '101',
+        },
+      ];
+      let failCancellation = false;
+      const deliveryUpdates: Array<Record<string, unknown>> = [];
+      const transaction = jest.fn(
+        async (work: (manager: EntityManager) => Promise<unknown>) => {
+          const draftSubscriptions = subscriptions.map((row) => ({
+            ...row,
+            sourceConfig: { ...row.sourceConfig },
+          })) as QqbotMessageSubscription[];
+          const draftDeliveries = structuredClone(deliveries);
+          const subscriptionStore = {
+            findOne: jest.fn(
+              async ({ where }: { where: Record<string, unknown> }) =>
+                draftSubscriptions.find((row) =>
+                  Object.entries(where).every(
+                    ([key, value]) => row[key] === value,
+                  ),
+                ) ?? null,
+            ),
+            save: jest.fn(async (row: QqbotMessageSubscription) => row),
+          };
+          const manager = {
+            getRepository: jest.fn((entity) => {
+              if (entity === QqbotMessageSubscription) return subscriptionStore;
+              if (entity === QqbotMessagePublishBinding) {
+                return { count: jest.fn().mockResolvedValue(0) };
+              }
+              if (entity === QqbotMessageDelivery) {
+                return {
+                  update: jest.fn(
+                    async (
+                      where: Record<string, unknown>,
+                      patch: Record<string, unknown>,
+                    ) => {
+                      deliveryUpdates.push(where);
+                      const statuses = (where.status as { _value?: unknown[] })
+                        ._value;
+                      let affected = 0;
+                      draftDeliveries.forEach((row) => {
+                        if (
+                          row.subscriptionId === where.subscriptionId &&
+                          statuses?.includes(row.status)
+                        ) {
+                          Object.assign(row, patch);
+                          affected += 1;
+                        }
+                      });
+                      if (failCancellation) {
+                        throw new Error('cancellation persistence failed');
+                      }
+                      return { affected };
+                    },
+                  ),
+                };
+              }
+              throw new Error('unexpected cancellation repository');
+            }),
+          } as unknown as EntityManager;
+          const result = await work(manager);
+          subscriptions.splice(0, subscriptions.length, ...draftSubscriptions);
+          deliveries.splice(0, deliveries.length, ...draftDeliveries);
+          return result;
+        },
+      );
+      const source = adapter();
+      source.normalizeSubscriptionConfig.mockImplementation(async (input) => {
+        const canonicalConfig = input as Record<string, string>;
+        return {
+          canonicalConfig,
+          resourceKey: canonicalConfig.portForwardId,
+          sourceSummary: 'transaction fixture',
+        };
+      });
+      const service = new QqbotMessageSubscriptionService(
+        {
+          manager: { transaction },
+        } as unknown as Repository<QqbotMessageSubscription>,
+        registry(source),
+      );
+      return {
+        deliveries,
+        deliveryUpdates,
+        failCancellation: () => {
+          failCancellation = true;
+        },
+        service,
+        subscriptions,
+      };
+    }
+
+    it.each([
+      [
+        'disable',
+        async (service: QqbotMessageSubscriptionService) =>
+          service.setEnabled('100', false),
+      ],
+      [
+        'soft delete',
+        async (service: QqbotMessageSubscriptionService) =>
+          service.remove('100'),
+      ],
+      [
+        'canonical config change',
+        async (service: QqbotMessageSubscriptionService) =>
+          service.update('100', {
+            enabled: true,
+            name: '端口提醒',
+            remark: null,
+            sourceConfig: {
+              ddnsRecordId: '2041700000000000099',
+              portForwardId: CONFIG.portForwardId,
+            },
+            sourceKey: SOURCE_KEY,
+          }),
+      ],
+    ])(
+      '%s commits config and cancels only exact claimable rows without rewriting snapshots',
+      async (_name, mutate) => {
+        const harness = cancellationHarness();
+        const before = structuredClone(harness.deliveries);
+
+        await mutate(harness.service);
+
+        expect(harness.deliveries).toEqual(
+          before.map((delivery) =>
+            delivery.subscriptionId === '100' &&
+            ['waiting_ddns', 'pending', 'retry'].includes(delivery.status)
+              ? {
+                  ...delivery,
+                  nextAttemptAt: null,
+                  processingLeaseUntil: null,
+                  status: 'cancelled',
+                }
+              : delivery,
+          ),
+        );
+        expect(
+          (harness.deliveryUpdates[0].status as { _value: string[] })._value,
+        ).toEqual(['waiting_ddns', 'pending', 'retry']);
+      },
+    );
+
+    it('does not cancel for name/remark-only edits or the same canonical identity', async () => {
+      const harness = cancellationHarness();
+      const before = structuredClone(harness.deliveries);
+
+      await harness.service.update('100', {
+        enabled: true,
+        name: '新名称',
+        remark: '新备注',
+        sourceConfig: { ...CONFIG },
+        sourceKey: SOURCE_KEY,
+      });
+
+      expect(harness.deliveries).toEqual(before);
+      expect(harness.deliveryUpdates).toHaveLength(0);
+    });
+
+    it('rolls back both subscription mutation and draft cancellations when cancellation persistence fails', async () => {
+      const harness = cancellationHarness();
+      const subscriptionsBefore = harness.subscriptions.map((row) => ({
+        ...row,
+        sourceConfig: { ...row.sourceConfig },
+      }));
+      const deliveriesBefore = structuredClone(harness.deliveries);
+      harness.failCancellation();
+
+      await expect(harness.service.setEnabled('100', false)).rejects.toThrow(
+        'cancellation persistence failed',
+      );
+
+      expect(harness.subscriptions).toEqual(subscriptionsBefore);
+      expect(harness.deliveries).toEqual(deliveriesBefore);
+    });
+  });
+
   it('pages only undeleted records with real filters and returns detached, real-time source status', async () => {
     const source = adapter({
       invalidReasonCode: 'ddns_mapping_mismatch',
@@ -975,6 +1188,9 @@ describe('QqbotMessageSubscriptionService', () => {
           if (entity === QqbotMessagePublishBinding) {
             return { count: jest.fn(async () => 0) };
           }
+          if (entity === QqbotMessageDelivery) {
+            return { update: jest.fn(async () => ({ affected: 0 })) };
+          }
           throw new Error('unexpected repository');
         }),
       } as unknown as EntityManager;

test/modules/qqbot/message-push/system-message-delivery-coordinator-integration.spec.ts

@@ -0,0 +1,328 @@
+import type { DataSource, EntityManager } from 'typeorm';
+import { KtDateTime } from '../../../../src/common';
+import { SystemMessageDeliveryCoordinatorService } from '../../../../src/modules/qqbot/core/application/message-push/system-message-delivery-coordinator.service';
+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 { QqbotMessageSubscription } from '../../../../src/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-subscription.entity';
+
+const SOURCE_KEY = 'network.stun.mapping-port-changed';
+const NOW = new Date('2026-07-24T03:04:05.000Z');
+
+type CoordinatorStore = {
+  deliveries: Array<Record<string, unknown>>;
+  events: Array<Record<string, unknown>>;
+  subscriptions: Array<Record<string, unknown>>;
+};
+
+type CoordinatorHarness = {
+  coordinator: SystemMessageDeliveryCoordinatorService;
+  failUpdate: (error: Error | null) => void;
+  operations: string[];
+  requestDrain: jest.SpyInstance<void, []>;
+  store: CoordinatorStore;
+  transaction: jest.Mock;
+};
+
+/** Deep-clones the JSON-shaped coordinator fixture while preserving project timestamps. */
+function cloneStore(store: CoordinatorStore): CoordinatorStore {
+  return structuredClone(store);
+}
+
+/** Reads the value carried by TypeORM's in-memory `In` find operator. */
+function inValues(value: unknown): unknown[] {
+  if (!value || typeof value !== 'object') return [];
+  const values = (value as { _value?: unknown })._value;
+  return Array.isArray(values) ? values : [];
+}
+
+/** Builds a transaction-faithful Core harness with commit-on-resolve and discard-on-reject. */
+function createCoordinatorHarness(): CoordinatorHarness {
+  const store: CoordinatorStore = {
+    deliveries: [
+      {
+        id: 'delivery-match',
+        messageEventId: 'event-match',
+        nextAttemptAt: 'future-match',
+        status: 'waiting_ddns',
+        subscriptionId: 'subscription-match',
+      },
+      {
+        id: 'delivery-old-address',
+        messageEventId: 'event-old-address',
+        nextAttemptAt: 'future-old',
+        status: 'waiting_ddns',
+        subscriptionId: 'subscription-match',
+      },
+      {
+        id: 'delivery-pending',
+        messageEventId: 'event-match',
+        nextAttemptAt: 'future-pending',
+        status: 'pending',
+        subscriptionId: 'subscription-match',
+      },
+      {
+        id: 'delivery-inactive-subscription',
+        messageEventId: 'event-match',
+        nextAttemptAt: 'future-inactive',
+        status: 'waiting_ddns',
+        subscriptionId: 'subscription-disabled',
+      },
+      {
+        id: 'delivery-numeric-config',
+        messageEventId: 'event-match',
+        nextAttemptAt: 'future-numeric',
+        status: 'waiting_ddns',
+        subscriptionId: 'subscription-numeric',
+      },
+      {
+        id: 'delivery-other-source',
+        messageEventId: 'event-other-source',
+        nextAttemptAt: 'future-source',
+        status: 'waiting_ddns',
+        subscriptionId: 'subscription-match',
+      },
+    ],
+    events: [
+      {
+        id: 'event-match',
+        payload: { publicIpv4: '8.8.8.8' },
+        sourceKey: SOURCE_KEY,
+      },
+      {
+        id: 'event-old-address',
+        payload: { publicIpv4: '1.1.1.1' },
+        sourceKey: SOURCE_KEY,
+      },
+      {
+        id: 'event-other-source',
+        payload: { publicIpv4: '8.8.8.8' },
+        sourceKey: 'other.source',
+      },
+      {
+        id: 'event-numeric-address',
+        payload: { publicIpv4: 8_888 },
+        sourceKey: SOURCE_KEY,
+      },
+    ],
+    subscriptions: [
+      {
+        enabled: true,
+        id: 'subscription-match',
+        isDeleted: false,
+        sourceConfig: { ddnsRecordId: '9007199254740993' },
+        sourceKey: SOURCE_KEY,
+      },
+      {
+        enabled: false,
+        id: 'subscription-disabled',
+        isDeleted: false,
+        sourceConfig: { ddnsRecordId: '9007199254740993' },
+        sourceKey: SOURCE_KEY,
+      },
+      {
+        enabled: true,
+        id: 'subscription-numeric',
+        isDeleted: false,
+        sourceConfig: { ddnsRecordId: 9_007_199_254_740_993 },
+        sourceKey: SOURCE_KEY,
+      },
+      {
+        enabled: true,
+        id: 'subscription-wrong',
+        isDeleted: false,
+        sourceConfig: { ddnsRecordId: 'other' },
+        sourceKey: SOURCE_KEY,
+      },
+      {
+        enabled: true,
+        id: 'subscription-deleted',
+        isDeleted: true,
+        sourceConfig: { ddnsRecordId: '9007199254740993' },
+        sourceKey: SOURCE_KEY,
+      },
+    ],
+  };
+  const operations: string[] = [];
+  let updateFailure: Error | null = null;
+  const transaction = jest.fn(
+    async (work: (manager: EntityManager) => Promise<unknown>) => {
+      operations.push('transaction:start');
+      const draft = cloneStore(store);
+      const manager = {
+        getRepository: (entity: unknown) => {
+          if (entity === QqbotMessageSubscription) {
+            return {
+              find: async ({ where }: { where: Record<string, unknown> }) =>
+                draft.subscriptions.filter((row) =>
+                  Object.entries(where).every(
+                    ([key, value]) => row[key] === value,
+                  ),
+                ),
+            };
+          }
+          if (entity === QqbotMessageEvent) {
+            return {
+              find: async ({ where }: { where: Record<string, unknown> }) =>
+                draft.events.filter((row) =>
+                  Object.entries(where).every(
+                    ([key, value]) => row[key] === value,
+                  ),
+                ),
+            };
+          }
+          if (entity === QqbotMessageDelivery) {
+            return {
+              update: async (
+                where: Record<string, unknown>,
+                patch: Record<string, unknown>,
+              ) => {
+                operations.push('delivery:update');
+                if (updateFailure) throw updateFailure;
+                const eventIds = inValues(where.messageEventId);
+                const subscriptionIds = inValues(where.subscriptionId);
+                let affected = 0;
+                draft.deliveries.forEach((delivery) => {
+                  if (
+                    eventIds.includes(delivery.messageEventId) &&
+                    subscriptionIds.includes(delivery.subscriptionId) &&
+                    (where.status === undefined ||
+                      delivery.status === where.status)
+                  ) {
+                    Object.assign(delivery, patch);
+                    affected += 1;
+                  }
+                });
+                return { affected };
+              },
+            };
+          }
+          throw new Error('unexpected coordinator repository');
+        },
+      } as unknown as EntityManager;
+      const result = await work(manager);
+      store.deliveries = draft.deliveries;
+      store.events = draft.events;
+      store.subscriptions = draft.subscriptions;
+      operations.push('transaction:commit');
+      return result;
+    },
+  );
+  const coordinator = new SystemMessageDeliveryCoordinatorService(
+    { transaction } as unknown as DataSource,
+    { runOnce: jest.fn().mockResolvedValue(0) } as never,
+    { runOnce: jest.fn().mockResolvedValue(0) } as never,
+  );
+  const requestDrain = jest
+    .spyOn(coordinator, 'requestDrain')
+    .mockImplementation(() => {
+      operations.push('drain');
+    });
+  return {
+    coordinator,
+    failUpdate: (error) => {
+      updateFailure = error;
+    },
+    operations,
+    requestDrain,
+    store,
+    transaction,
+  };
+}
+
+describe('SystemMessageDeliveryCoordinatorService DDNS integration', () => {
+  beforeEach(() => {
+    jest.useFakeTimers().setSystemTime(NOW);
+  });
+
+  afterEach(() => {
+    jest.useRealTimers();
+  });
+
+  it('requires an exact non-empty string DDNS ID and IPv4 address', async () => {
+    const harness = createCoordinatorHarness();
+
+    for (const input of [
+      { appliedAddress: '8.8.8.8', ddnsRecordId: 9_007_199_254_740_993 },
+      { appliedAddress: '8.8.8.8', ddnsRecordId: '' },
+      { appliedAddress: 'not-an-ip', ddnsRecordId: '9007199254740993' },
+      { appliedAddress: '::1', ddnsRecordId: '9007199254740993' },
+      { appliedAddress: 8_888, ddnsRecordId: '9007199254740993' },
+    ]) {
+      await harness.coordinator.notifyDdnsSynced(input as never);
+    }
+
+    expect(harness.transaction).not.toHaveBeenCalled();
+    expect(harness.requestDrain).not.toHaveBeenCalled();
+  });
+
+  it('advances only exact active-subscription and address-relevant waiting rows after commit', async () => {
+    const harness = createCoordinatorHarness();
+    const before = cloneStore(harness.store);
+
+    await harness.coordinator.notifyDdnsSynced({
+      appliedAddress: '8.8.8.8',
+      ddnsRecordId: '9007199254740993',
+    });
+
+    expect(harness.store.deliveries).toEqual(
+      before.deliveries.map((delivery) =>
+        delivery.id === 'delivery-match'
+          ? { ...delivery, nextAttemptAt: new KtDateTime(NOW) }
+          : delivery,
+      ),
+    );
+    expect(harness.operations).toEqual([
+      'transaction:start',
+      'delivery:update',
+      'transaction:commit',
+      'drain',
+    ]);
+    expect(harness.requestDrain).toHaveBeenCalledTimes(1);
+  });
+
+  it('keeps the write status-fenced and does not drain when the update advances zero rows', async () => {
+    const harness = createCoordinatorHarness();
+    harness.store.deliveries.find(
+      (delivery) => delivery.id === 'delivery-match',
+    )!.status = 'processing';
+    const before = cloneStore(harness.store);
+
+    await harness.coordinator.notifyDdnsSynced({
+      appliedAddress: '8.8.8.8',
+      ddnsRecordId: '9007199254740993',
+    });
+
+    expect(harness.store).toEqual(before);
+    expect(harness.requestDrain).not.toHaveBeenCalled();
+  });
+
+  it('discards transaction drafts and never drains after an update failure', async () => {
+    const harness = createCoordinatorHarness();
+    const before = cloneStore(harness.store);
+    harness.failUpdate(new Error('delivery update failed'));
+
+    await expect(
+      harness.coordinator.notifyDdnsSynced({
+        appliedAddress: '8.8.8.8',
+        ddnsRecordId: '9007199254740993',
+      }),
+    ).rejects.toThrow('delivery update failed');
+
+    expect(harness.store).toEqual(before);
+    expect(harness.requestDrain).not.toHaveBeenCalled();
+    expect(harness.operations).not.toContain('transaction:commit');
+  });
+
+  it('does not query or drain after coordinator destruction', async () => {
+    const harness = createCoordinatorHarness();
+    await harness.coordinator.onModuleDestroy();
+
+    await harness.coordinator.notifyDdnsSynced({
+      appliedAddress: '8.8.8.8',
+      ddnsRecordId: '9007199254740993',
+    });
+
+    expect(harness.transaction).not.toHaveBeenCalled();
+    expect(harness.requestDrain).not.toHaveBeenCalled();
+  });
+});

test/qqbot/account/qqbot-account.service.spec.ts

@@ -1,6 +1,10 @@
 import { ToolsService } from '@/common';
 import type { QqbotAccountNapcatRuntimePort } from '@/modules/qqbot/core/application/account/qqbot-account-napcat-runtime.port';
 import { QqbotAccountService } from '@/modules/qqbot/core/application/account/qqbot-account.service';
+import { QqbotAccountAbility } from '@/modules/qqbot/core/infrastructure/persistence/account/qqbot-account-ability.entity';
+import { QqbotAccount } from '@/modules/qqbot/core/infrastructure/persistence/account/qqbot-account.entity';
+import { QqbotMessageDelivery } from '@/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-delivery.entity';
+import { QqbotMessagePublishBinding } from '@/modules/qqbot/core/infrastructure/persistence/message-push/qqbot-message-publish-binding.entity';
 import { QqbotNapcatAccountRuntimeService } from '@/modules/qqbot/napcat/application/account-runtime/qqbot-napcat-account-runtime.service';
 
 /**
@@ -74,6 +78,369 @@ const createAccountServiceWithNapcatRuntime = (input: {
 };
 
 describe('QqbotAccountService', () => {
+  it('fences administrative account cancellation to unfinished delivery statuses', async () => {
+    const deliveryUpdate = jest.fn().mockResolvedValue({ affected: 2 });
+    const manager = {
+      getRepository: (entity: unknown) => {
+        if (entity === QqbotMessagePublishBinding) {
+          return { find: jest.fn().mockResolvedValue([{ id: 'binding-1' }]) };
+        }
+        if (entity === QqbotMessageDelivery) return { update: deliveryUpdate };
+        throw new Error('unexpected repository');
+      },
+    };
+    const service = createAccountService({ accountRepository: {} });
+
+    await (service as any).cancelAccountDeliveries(manager, 'account-1');
+
+    expect(deliveryUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({
+        bindingId: expect.anything(),
+        status: expect.anything(),
+      }),
+      expect.objectContaining({ status: 'cancelled' }),
+    );
+    const [where] = deliveryUpdate.mock.calls[0];
+    expect(where.status.value).toEqual(['waiting_ddns', 'pending', 'retry']);
+  });
+
+  describe('administrative delivery cancellation transaction', () => {
+    /** Builds account/ability/delivery authority with isolated commit-on-resolve drafts. */
+    function cancellationHarness() {
+      const accounts = [
+        {
+          accessToken: null,
+          connectionMode: 'reverse-ws',
+          enabled: true,
+          id: 'account-1',
+          isDeleted: false,
+          name: 'Primary',
+          remark: '',
+          selfId: '1914728559',
+        },
+      ];
+      const abilities = [
+        {
+          accountId: 'account-1',
+          id: 'ability-1',
+          isDeleted: false,
+          selfId: '1914728559',
+        },
+      ];
+      const bindings = [
+        { accountId: 'account-1', id: 'binding-1' },
+        { accountId: 'account-1', id: 'binding-2' },
+        { accountId: 'account-2', id: 'binding-other' },
+      ];
+      const deliveries = [
+        ...['waiting_ddns', 'pending', 'retry', 'processing', 'success'].map(
+          (status) => ({
+            bindingId: 'binding-1',
+            frozen: `binding-1-${status}`,
+            id: `binding-1-${status}`,
+            nextAttemptAt: `schedule-${status}`,
+            processingLeaseUntil: `lease-${status}`,
+            status,
+          }),
+        ),
+        {
+          bindingId: 'binding-2',
+          frozen: 'binding-2',
+          id: 'binding-2-pending',
+          nextAttemptAt: 'schedule-binding-2',
+          processingLeaseUntil: 'lease-binding-2',
+          status: 'pending',
+        },
+        {
+          bindingId: 'binding-other',
+          frozen: 'other',
+          id: 'binding-other-pending',
+          nextAttemptAt: 'schedule-other',
+          processingLeaseUntil: 'lease-other',
+          status: 'pending',
+        },
+      ];
+      const operations: string[] = [];
+      let failCancellation = false;
+      const transaction = jest.fn(
+        async (work: (manager: any) => Promise<unknown>) => {
+          operations.push('transaction:start');
+          const draftAccounts = structuredClone(accounts);
+          const draftAbilities = structuredClone(abilities);
+          const draftDeliveries = structuredClone(deliveries);
+          const accountStore = {
+            findOne: jest.fn(
+              async ({
+                lock,
+                where,
+              }: {
+                lock?: { mode: string };
+                where: Record<string, unknown>;
+              }) => {
+                if (lock) operations.push('account:lock');
+                const row = draftAccounts.find((candidate) =>
+                  Object.entries(where).every(
+                    ([key, value]) => candidate[key] === value,
+                  ),
+                );
+                return row ? structuredClone(row) : null;
+              },
+            ),
+            update: jest.fn(
+              async (
+                where: Record<string, unknown>,
+                patch: Record<string, unknown>,
+              ) => {
+                operations.push('account:update');
+                const row = draftAccounts.find((candidate) =>
+                  Object.entries(where).every(
+                    ([key, value]) => candidate[key] === value,
+                  ),
+                );
+                if (!row) return { affected: 0 };
+                Object.assign(row, patch);
+                return { affected: 1 };
+              },
+            ),
+          };
+          const manager = {
+            getRepository: (entity: unknown) => {
+              if (entity === QqbotAccount) return accountStore;
+              if (entity === QqbotAccountAbility) {
+                return {
+                  update: async (
+                    where: Record<string, unknown>,
+                    patch: Record<string, unknown>,
+                  ) => {
+                    operations.push('ability:update');
+                    draftAbilities
+                      .filter((row) =>
+                        Object.entries(where).every(
+                          ([key, value]) => row[key] === value,
+                        ),
+                      )
+                      .forEach((row) => Object.assign(row, patch));
+                    return { affected: 1 };
+                  },
+                };
+              }
+              if (entity === QqbotMessagePublishBinding) {
+                return {
+                  find: async ({
+                    where,
+                  }: {
+                    where: Record<string, unknown>;
+                  }) => {
+                    operations.push('bindings:read');
+                    return bindings.filter((row) =>
+                      Object.entries(where).every(
+                        ([key, value]) => row[key] === value,
+                      ),
+                    );
+                  },
+                };
+              }
+              if (entity === QqbotMessageDelivery) {
+                return {
+                  update: async (
+                    where: Record<string, any>,
+                    patch: Record<string, unknown>,
+                  ) => {
+                    operations.push('delivery:update');
+                    const bindingIds = where.bindingId._value as string[];
+                    const statuses = where.status._value as string[];
+                    draftDeliveries.forEach((row) => {
+                      if (
+                        bindingIds.includes(row.bindingId) &&
+                        statuses.includes(row.status)
+                      ) {
+                        Object.assign(row, patch);
+                      }
+                    });
+                    if (failCancellation) {
+                      throw new Error('account cancellation failed');
+                    }
+                    return { affected: 4 };
+                  },
+                };
+              }
+              throw new Error('unexpected account repository');
+            },
+          };
+          const result = await work(manager);
+          accounts.splice(0, accounts.length, ...draftAccounts);
+          abilities.splice(0, abilities.length, ...draftAbilities);
+          deliveries.splice(0, deliveries.length, ...draftDeliveries);
+          operations.push('transaction:commit');
+          return result;
+        },
+      );
+      const accountRepository = {
+        findOne: jest.fn(
+          async ({ where }: { where: Record<string, unknown> }) => {
+            const row = accounts.find((candidate) =>
+              Object.entries(where).every(
+                ([key, value]) => candidate[key] === value,
+              ),
+            );
+            return row ? structuredClone(row) : null;
+          },
+        ),
+        manager: { transaction },
+        update: jest.fn(),
+      };
+      const removeAccountContainers = jest.fn(async () => {
+        operations.push('external:remove');
+        return { deletedContainers: 1 };
+      });
+      const service = createAccountService({
+        accountAbilityRepository: { update: jest.fn() },
+        accountRepository,
+        napcatRuntime: { removeAccountContainers } as never,
+      });
+      return {
+        abilities,
+        accounts,
+        deliveries,
+        failCancellation: () => {
+          failCancellation = true;
+        },
+        operations,
+        removeAccountContainers,
+        service,
+        transaction,
+      };
+    }
+
+    /** Asserts only account-owned claimable delivery rows entered cancellation state. */
+    function expectAccountDeliveriesCancelled(
+      before: Array<Record<string, unknown>>,
+      after: Array<Record<string, unknown>>,
+    ): void {
+      expect(after).toEqual(
+        before.map((delivery) =>
+          ['binding-1', 'binding-2'].includes(String(delivery.bindingId)) &&
+          ['waiting_ddns', 'pending', 'retry'].includes(String(delivery.status))
+            ? {
+                ...delivery,
+                nextAttemptAt: null,
+                processingLeaseUntil: null,
+                status: 'cancelled',
+              }
+            : delivery,
+        ),
+      );
+    }
+
+    it.each([
+      [
+        'administrative disable',
+        async (service: QqbotAccountService) =>
+          service.update({ enabled: false, id: 'account-1' }),
+      ],
+      [
+        'actual selfId replacement',
+        async (service: QqbotAccountService) =>
+          service.update({
+            id: 'account-1',
+            selfId: '1914728560',
+          }),
+      ],
+      [
+        'administrative delete',
+        async (service: QqbotAccountService) => service.remove('account-1'),
+      ],
+    ])(
+      '%s commits account state and exact delivery cancellations',
+      async (_name, mutate) => {
+        const harness = cancellationHarness();
+        const before = structuredClone(harness.deliveries);
+
+        await mutate(harness.service);
+
+        expectAccountDeliveriesCancelled(before, harness.deliveries);
+        expect(harness.operations.indexOf('account:lock')).toBeLessThan(
+          harness.operations.indexOf('bindings:read'),
+        );
+        expect(harness.operations.indexOf('bindings:read')).toBeLessThan(
+          harness.operations.indexOf('delivery:update'),
+        );
+        expect(harness.operations.at(-1)).toBe('transaction:commit');
+      },
+    );
+
+    it('same selfId and metadata-only updates do not cancel frozen deliveries', async () => {
+      const sameSelfId = cancellationHarness();
+      const sameBefore = structuredClone(sameSelfId.deliveries);
+      await sameSelfId.service.update({
+        id: 'account-1',
+        name: 'Renamed',
+        selfId: '1914728559',
+      });
+      expect(sameSelfId.deliveries).toEqual(sameBefore);
+      expect(sameSelfId.operations).not.toContain('delivery:update');
+
+      const metadata = cancellationHarness();
+      const metadataBefore = structuredClone(metadata.deliveries);
+      await metadata.service.update({ id: 'account-1', remark: 'metadata' });
+      expect(metadata.deliveries).toEqual(metadataBefore);
+      expect(metadata.operations).not.toContain('delivery:update');
+    });
+
+    it('runtime online/offline changes stay outside administrative cancellation', async () => {
+      const harness = cancellationHarness();
+      const before = structuredClone(harness.deliveries);
+
+      await harness.service.markOnline('1914728559', 'Universal');
+      await harness.service.markOffline('1914728559');
+      await harness.service.markHeartbeat('1914728559');
+
+      expect(harness.transaction).not.toHaveBeenCalled();
+      expect(harness.deliveries).toEqual(before);
+    });
+
+    it('rolls back account, ability, and delivery drafts together when cancellation fails', async () => {
+      const harness = cancellationHarness();
+      const before = {
+        abilities: structuredClone(harness.abilities),
+        accounts: structuredClone(harness.accounts),
+        deliveries: structuredClone(harness.deliveries),
+      };
+      harness.failCancellation();
+
+      await expect(
+        harness.service.update({
+          id: 'account-1',
+          selfId: '1914728560',
+        }),
+      ).rejects.toThrow('account cancellation failed');
+
+      expect(harness.accounts).toEqual(before.accounts);
+      expect(harness.abilities).toEqual(before.abilities);
+      expect(harness.deliveries).toEqual(before.deliveries);
+      expect(harness.operations).not.toContain('transaction:commit');
+    });
+
+    it('keeps external removal outside the DB transaction while rolling back authoritative delete state', async () => {
+      const harness = cancellationHarness();
+      const before = {
+        abilities: structuredClone(harness.abilities),
+        accounts: structuredClone(harness.accounts),
+        deliveries: structuredClone(harness.deliveries),
+      };
+      harness.failCancellation();
+
+      await expect(harness.service.remove('account-1')).rejects.toThrow(
+        'account cancellation failed',
+      );
+
+      expect(harness.operations[0]).toBe('external:remove');
+      expect(harness.accounts).toEqual(before.accounts);
+      expect(harness.abilities).toEqual(before.abilities);
+      expect(harness.deliveries).toEqual(before.deliveries);
+      expect(harness.removeAccountContainers).toHaveBeenCalledTimes(1);
+    });
+  });
   it('stores NapCat login password as encrypted secret and never persists the transport field', async () => {
     const toolsService = new ToolsService();
     const accountRepository = {
@@ -208,8 +575,29 @@ describe('QqbotAccountService', () => {
   });
 
   it('does not update NapCat login password when edit leaves the password blank', async () => {
+    const accountUpdate = jest.fn().mockResolvedValue({ affected: 1 });
+    const account = {
+      enabled: true,
+      id: 'account-1',
+      isDeleted: false,
+      selfId: '1914728559',
+    };
     const accountRepository = {
-      findOne: jest.fn().mockResolvedValue({ id: 'account-1' }),
+      findOne: jest.fn().mockResolvedValue(account),
+      manager: {
+        transaction: async (work: (manager: any) => Promise<unknown>) =>
+          work({
+            getRepository: (entity: unknown) => {
+              if (entity === QqbotAccount) {
+                return {
+                  findOne: jest.fn().mockResolvedValue({ ...account }),
+                  update: accountUpdate,
+                };
+              }
+              throw new Error('unexpected repository');
+            },
+          }),
+      },
       update: jest.fn(),
     };
     const service = createAccountService({
@@ -223,7 +611,7 @@ describe('QqbotAccountService', () => {
       selfId: '1914728559',
     });
 
-    expect(accountRepository.update).toHaveBeenCalledWith(
+    expect(accountUpdate).toHaveBeenCalledWith(
       { id: 'account-1' },
       expect.not.objectContaining({
         encryptedLoginPassword: expect.anything(),