flutterApp/test/v2_work_task_tracker_test.dart

import 'dart:async';
import 'dart:convert';

import 'package:flutter_test/flutter_test.dart';
import 'package:fpdart/fpdart.dart';
import 'package:shared_preferences/shared_preferences.dart';
import 'package:latlong2/latlong.dart';
import 'package:maibu_satabot_v2/core/error/failure.dart';
import 'package:maibu_satabot_v2/core/logging/work_trace_diagnostics.dart';
import 'package:maibu_satabot_v2/core/network/mqtt/domain/interfaces/mqtt_client.dart';
import 'package:maibu_satabot_v2/core/network/mqtt/domain/models/mqtt_config.dart';
import 'package:maibu_satabot_v2/core/network/mqtt/domain/models/mqtt_message.dart';
import 'package:maibu_satabot_v2/features/autonomous_work/data/v2_work_store.dart';
import 'package:maibu_satabot_v2/features/autonomous_work/data/v2_work_task_tracker.dart';
import 'package:maibu_satabot_v2/features/autonomous_work/domain/entities/work_session_models.dart';
import 'package:maibu_satabot_v2/features/devices/domain/entities/device_schedule_plan_entity.dart';
import 'package:maibu_satabot_v2/features/devices/domain/entities/device_task_entity.dart';
import 'package:maibu_satabot_v2/features/devices/domain/repositories/device_task_repository.dart';
import 'package:maibu_satabot_v2/features/devices/domain/repositories/path_repository.dart';
import 'package:maibu_satabot_v2/features/devices/domain/usecases/create_device_task_usecase.dart';
import 'package:maibu_satabot_v2/features/pv_control/presentation/models/work_trace_recorder.dart';

Future<void> flush() async {
  for (var i = 0; i < 15; i++) {
    await Future<void>.delayed(Duration.zero);
  }
}

class FakeMqtt implements MqttClient {
  final messages = StreamController<MqttMessage>.broadcast(sync: true);
  final topics = <String>{};
  final configs = <MqttConfig>[];
  @override
  bool isConnected = false;
  @override
  Stream<MqttMessage> get messageStream => messages.stream;
  @override
  Future<void> connect(MqttConfig config) async {
    configs.add(config);
    isConnected = true;
  }

  @override
  Future<void> disconnect() async {
    isConnected = false;
  }

  @override
  Future<void> subscribe(String topic) async {
    topics.add(topic);
  }

  @override
  Future<void> unsubscribe(String topic) async {
    topics.remove(topic);
  }

  @override
  Future<void> publish(String topic, String message) async {}
  void status(
    String status, {
    int taskId = 42,
    String deviceId = 'MC700-test',
    int? topicId,
  }) {
    messages.add(
      MqttMessage(
        topic: 'task/${topicId ?? taskId}/status',
        payload: jsonEncode({
          'taskId': taskId,
          'deviceId': deviceId,
          'status': status,
        }),
      ),
    );
  }

  void arrive(dynamic entity, {int taskId = 42}) => messages.add(
    MqttMessage(
      topic: 'task/$taskId/arrive',
      payload: jsonEncode({
        'type': 'device_task_arrive_point',
        'deviceId': 'MC700-test',
        'taskId': taskId,
        'entity': entity,
      }),
    ),
  );
}

class FakeTasks implements DeviceTaskRepository {
  List<DeviceTaskEntity> tasks = [entity()];
  Completer<Either<Failure, List<DeviceTaskEntity>>>? pending;
  String? error;
  int? operatedTask;
  bool? force;
  static DeviceTaskEntity entity({
    int id = 42,
    String state = 'EXECUTING',
    int routeId = 8,
  }) => DeviceTaskEntity(
    id: id,
    deviceId: 'MC700-test',
    taskStatus: state,
    taskStatusTranslate: '',
    routeId: routeId,
    siteId: 2,
    orgId: 3,
  );
  @override
  Future<Either<Failure, List<DeviceTaskEntity>>> getDeviceTaskPool({
    required String userId,
    required int siteId,
    required int orgId,
    int pageNum = 1,
    int pageSize = 99999999,
  }) async =>
      pending?.future ?? (error == null ? Right(tasks) : Left(Failure(error!)));
  @override
  Future<Either<Failure, bool>> cancelTask({
    required String deviceId,
    required int taskId,
    required int orgId,
    required int siteId,
    bool forceCancel = true,
  }) async {
    operatedTask = taskId;
    force = forceCancel;
    return error == null ? const Right(true) : Left(Failure(error!));
  }

  @override
  Future<Either<Failure, bool>> pauseTask({
    required String deviceId,
    required int taskId,
    required int orgId,
    required int siteId,
  }) async {
    operatedTask = taskId;
    return error == null ? const Right(true) : Left(Failure(error!));
  }

  @override
  Future<Either<Failure, Map<String, dynamic>>> recoveryTask({
    required String deviceId,
    required int taskId,
    required int orgId,
    required int siteId,
  }) async {
    operatedTask = taskId;
    return error == null ? const Right({}) : Left(Failure(error!));
  }

  @override
  Future<Either<Failure, bool>> saveDeviceSchedulePlan({
    required DeviceSchedulePlanEntity plan,
  }) async {
    return error == null ? const Right(true) : Left(Failure(error!));
  }
}

class FakePath implements PathRepository {
  int creations = 0;
  @override
  Future<int> createDeviceTask({
    required String deviceId,
    required int routeId,
    required int siteId,
    required int orgId,
  }) async {
    creations++;
    return 100;
  }

  @override
  dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
}

void main() {
  TestWidgetsFlutterBinding.ensureInitialized();
  late FakeTasks tasks;
  late FakePath path;
  late FakeMqtt mqtt;
  late V2WorkTaskTracker tracker;
  late StreamSubscription<WorkTaskSnapshot?> sub;
  late List<WorkTaskSnapshot?> updates;
  late List<String> traceLogs;
  late bool failTraceLogs;
  late int failedTraceWrites;
  setUp(() {
    SharedPreferences.setMockInitialValues({
      'saved_device_task_ids': '{"legacy":77}',
    });
    tasks = FakeTasks();
    path = FakePath();
    mqtt = FakeMqtt();
    updates = [];
    traceLogs = [];
    failTraceLogs = false;
    failedTraceWrites = 0;
    tracker = V2WorkTaskTracker(
      repository: tasks,
      createTask: CreateDeviceTaskUseCase(path),
      resolveScope: (_) async => const WorkScope(siteId: 2, orgId: 3),
      userId: () => 'test-user',
      store: V2WorkStore('test-user'),
      mqtt: mqtt,
      diagnostics: WorkTraceDiagnostics(
        now: () => DateTime(2026, 11, 5),
        write: (line) {
          if (failTraceLogs) {
            failedTraceWrites++;
            throw StateError('诊断输出失败');
          }
          traceLogs.add(line);
        },
      ),
    );
    sub = tracker.watch('MC700-test').listen(updates.add);
  });
  tearDown(() async {
    await sub.cancel();
    await flush();
    await mqtt.messages.close();
  });

  test(
    'restore task subscribes with task ID, filters other devices and tasks',
    () async {
      final result = await tracker.load('MC700-test');
      await flush();
      expect(result.isOk, isTrue);
      expect(mqtt.topics, {'task/42/status', 'task/42/arrive'});
      expect(mqtt.configs.single.clientId, 'v2_work_task');
      mqtt.status('FINISH', taskId: 8); // routeId cannot be used as taskId
      mqtt.status('FINISH', deviceId: 'other-device');
      mqtt.status('FINISH', taskId: 99, topicId: 42);
      await flush();
      expect(tracker.current('MC700-test')!.state, WorkRunState.running);
      mqtt.status('PAUSE');
      await flush();
      expect(updates.last!.state, WorkRunState.paused);
    },
  );

  test(
    'FINISH leaves terminal snapshot, removes active ID and preserves legacy cache',
    () async {
      await tracker.load('MC700-test');
      await flush();
      mqtt.status('EXECUTING');
      mqtt.status('FINISH');
      await flush();
      expect(updates.last!.state, WorkRunState.finished);
      expect(mqtt.topics, isEmpty);
      expect(await tracker.history('MC700-test'), hasLength(1));
      expect(
        traceLogs.any((s) => s.contains('[MQTT收到]') && s.contains('FINISH')),
        isTrue,
      );
      expect(
        traceLogs.any(
          (s) =>
              s.contains('[结束消息收到]') &&
              s.contains('"source":"mqtt_status"') &&
              s.contains('task/42/status'),
        ),
        isTrue,
      );
      expect(
        traceLogs.any(
          (s) =>
              s.contains('[结束状态落地]') &&
              s.contains('"removedFromActive":true') &&
              s.contains('"inHistory":true'),
        ),
        isTrue,
      );
      final prefs = await SharedPreferences.getInstance();
      expect(prefs.getString('saved_device_task_ids'), '{"legacy":77}');
      final stored = await V2WorkStore('test-user').read('tasks');
      expect(stored!['current'], isEmpty);
      final reloaded = await tracker.load('MC700-test');
      expect(
        reloaded.data,
        isEmpty,
        reason: 'stale pool cannot revive a finished task',
      );
      expect(updates.last!.state, WorkRunState.finished);
    },
  );

  test('MQTT finish wins over an earlier pending HTTP response', () async {
    await tracker.load('MC700-test');
    await flush();
    tasks.pending = Completer<Either<Failure, List<DeviceTaskEntity>>>();
    final loading = tracker.load('MC700-test');
    await flush();
    mqtt.status('FINISH');
    tasks.pending!.complete(Right([FakeTasks.entity()]));
    expect((await loading).data, isEmpty);
    expect(tracker.current('MC700-test')!.state, WorkRunState.finished);
  });

  test(
    'arrival points accumulate only during execution and empty arrival completes',
    () async {
      await tracker.load('MC700-test');
      await flush();
      mqtt.arrive({'lat': 31.1, 'lng': 120.1});
      await flush();
      expect(tracker.current('MC700-test')!.completedPoints, hasLength(1));
      mqtt.status('PAUSE');
      mqtt.arrive({'lat': 31.2, 'lng': 120.2});
      await flush();
      expect(tracker.current('MC700-test')!.completedPoints, hasLength(1));
      mqtt.arrive(null);
      await flush();
      expect(tracker.current('MC700-test')!.state, WorkRunState.finished);
      expect(
        traceLogs.any(
          (s) =>
              s.contains('[结束消息收到]') &&
              s.contains('"source":"mqtt_arrive_empty"') &&
              s.contains('"entityPresent":true'),
        ),
        isTrue,
      );
      expect(
        traceLogs.any(
          (s) => s.contains('[结束状态落地]') && s.contains('task/42/arrive'),
        ),
        isTrue,
      );
    },
  );

  test('NEW状态延迟时接受有效到达点，保留原始状态与诊断链路', () async {
    tasks.tasks = [FakeTasks.entity(state: 'NEW')];
    await tracker.load('MC700-test');
    await flush();
    expect(
      traceLogs.any(
        (s) => s.contains('[订阅登记]') && s.contains('task/42/arrive'),
      ),
      isTrue,
    );
    mqtt.arrive({'lat': 31.1, 'lng': 120.1});
    await flush();
    expect(tracker.current('MC700-test')!.completedPoints, hasLength(1));
    expect(tracker.current('MC700-test')!.rawStatus, 'NEW');
    expect(tracker.current('MC700-test')!.executionStarted, isTrue);
    expect(tracker.current('MC700-test')!.canRecordTrace, isTrue);
    expect(
      traceLogs.any(
        (s) =>
            s.contains('[MQTT收到]') &&
            s.contains('task/42/arrive') &&
            s.contains('31.1'),
      ),
      isTrue,
    );
    mqtt.status('EXECUTING');
    mqtt.arrive({'lat': 31.1, 'lng': 120.1});
    mqtt.arrive({'lat': 31.2, 'lng': 120.2});
    await flush();
    expect(tracker.current('MC700-test')!.completedPoints, hasLength(2));
    expect(
      traceLogs.any((s) => s.contains('[状态收到]') && s.contains('EXECUTING')),
      isTrue,
    );
    expect(
      traceLogs.any((s) => s.contains('[轨迹追加]') && s.contains('"points":2')),
      isTrue,
    );
    expect(
      traceLogs.any(
        (s) => s.contains('[任务快照]') && s.contains('"selected":true'),
      ),
      isTrue,
    );
  });

  test('过滤日志说明 topic、ID、type、坐标和解析失败，拒绝消息不修改轨迹', () async {
    tasks.tasks = [FakeTasks.entity(state: 'NEW')];
    await tracker.load('MC700-test');
    await flush();
    final valid = {
      'type': 'device_task_arrive_point',
      'taskId': 42,
      'deviceId': 'MC700-test',
      'entity': {'lat': 31.1, 'lng': 120.1},
    };
    void send(Map<String, dynamic> data, {String topic = 'task/42/arrive'}) =>
        mqtt.messages.add(MqttMessage(topic: topic, payload: jsonEncode(data)));
    send(valid, topic: 'task/99/arrive');
    send({...valid, 'taskId': 8});
    send({...valid, 'deviceId': 'other'});
    send({...valid, 'type': 'unexpected'});
    send({
      ...valid,
      'entity': {'lat': 0, 'lng': 120.1},
    });
    send({...valid, 'entity': 'not a point'});
    mqtt.messages.add(
      MqttMessage(topic: 'task/42/arrive', payload: 'invalid json'),
    );
    await flush();
    expect(tracker.current('MC700-test')!.completedPoints, isEmpty);
    expect(tracker.current('MC700-test')!.executionStarted, isFalse);
    expect(tracker.current('MC700-test')!.canRecordTrace, isFalse);
    for (final reason in [
      'topic格式不符或不在当前订阅列表',
      '消息taskId与topic任务不一致',
      'deviceId不匹配',
      '到达点type不匹配',
      '坐标为空、非数值、零或超出范围',
      'entity不是坐标对象',
      '[解析失败]',
    ]) {
      expect(traceLogs.any((s) => s.contains(reason)), isTrue, reason: reason);
    }
    mqtt.arrive({'lat': 31.1, 'lng': 120.1});
    mqtt.arrive({'lat': 31.1, 'lng': 120.1});
    await flush();
    expect(tracker.current('MC700-test')!.completedPoints, hasLength(1));
    expect(traceLogs.any((s) => s.contains('与上一轨迹点重复')), isTrue);
  });

  test('诊断采样只减少日志，连续到达点完整保留；日志写入异常也不影响轨迹', () async {
    await tracker.load('MC700-test');
    await flush();
    for (var i = 0; i < 20; i++) {
      mqtt.arrive({'lat': 31.1 + i * 0.00001, 'lng': 120.1});
    }
    await flush();
    expect(tracker.current('MC700-test')!.completedPoints, hasLength(20));
    expect(traceLogs.where((s) => s.contains('[MQTT收到]')), hasLength(1));
    failTraceLogs = true;
    mqtt.status('PAUSE');
    expect(tracker.current('MC700-test')!.state, WorkRunState.paused);
    mqtt.status('EXECUTING');
    mqtt.arrive({'lat': 31.2, 'lng': 120.2});
    await flush();
    expect(tracker.current('MC700-test')!.completedPoints, hasLength(21));
    expect(failedTraceWrites, greaterThan(0));
  });

  test(
    'new task survives empty pool delay and a new watcher receives its state',
    () async {
      tasks.tasks = [];
      final created = await tracker.start(
        const WorkStartRequest(
          deviceId: 'MC700-test',
          route: WorkRouteOption(id: '8', name: 'route'),
        ),
      );
      await flush();
      expect(created.data!.taskId, '100');
      expect(created.data!.rawStatus, 'NEW');
      expect(created.data!.executionStarted, isTrue);
      expect(created.data!.canRecordTrace, isTrue);
      expect(mqtt.topics, {'task/100/status', 'task/100/arrive'});
      final secondUpdates = <WorkTaskSnapshot?>[];
      final second = tracker.watch('MC700-test').listen(secondUpdates.add);
      await flush();
      expect(secondUpdates.single!.taskId, '100');
      await second.cancel();
      await flush();
      expect(mqtt.topics, isNotEmpty);
      expect(path.creations, 1);
      final refreshed = await tracker.load('MC700-test');
      expect(refreshed.data!.single.taskId, '100');
      expect(tracker.current('MC700-test')!.taskId, '100');
    },
  );

  test(
    'pause resume stop use current task ID and stop cleans subscriptions',
    () async {
      await tracker.load('MC700-test');
      await flush();
      var current = tracker.current('MC700-test')!;
      current = (await tracker.command('MC700-test', current, 'pause')).data!;
      expect(current.state, WorkRunState.paused);
      current = (await tracker.command('MC700-test', current, 'resume')).data!;
      expect(current.state, WorkRunState.running);
      current = (await tracker.command('MC700-test', current, 'cancel')).data!;
      await flush();
      expect(tasks.operatedTask, 42);
      expect(tasks.force, isTrue);
      expect(current.rawStatus, 'CANCELED');
      expect(
        traceLogs.any(
          (s) => s.contains('[结束状态落地]') && s.contains('"source":"http_cancel"'),
        ),
        isTrue,
      );
      expect(traceLogs.where((s) => s.contains('[结束消息收到]')), isEmpty);
      expect(mqtt.topics, isEmpty);
      expect(
        (await tracker.command('MC700-test', current, 'resume')).isOk,
        isFalse,
      );
    },
  );

  test('只选中NEW任务不记录GPS，确认开始复用任务后才记录', () async {
    tasks.tasks = [FakeTasks.entity(state: 'NEW')];
    await tracker.load('MC700-test');
    final selected = await tracker.select(tracker.current('MC700-test')!);
    expect(selected.data!.canRecordTrace, isFalse);
    final started = await tracker.start(
      const WorkStartRequest(
        deviceId: 'MC700-test',
        route: WorkRouteOption(id: '8', name: 'route'),
        reuseTaskId: '42',
      ),
    );
    expect(started.data!.rawStatus, 'NEW');
    expect(started.data!.executionStarted, isTrue);
    expect(started.data!.canRecordTrace, isTrue);
    expect(path.creations, 0);
    await tracker.load('MC700-test');
    expect(tracker.current('MC700-test')!.canRecordTrace, isTrue);
  });

  test('开始标记持久化，重进页面遇到NEW也不会丢掉轨迹接收权限', () async {
    tasks.tasks = [FakeTasks.entity(state: 'NEW')];
    await tracker.load('MC700-test');
    await tracker.start(
      const WorkStartRequest(
        deviceId: 'MC700-test',
        route: WorkRouteOption(id: '8', name: 'route'),
        reuseTaskId: '42',
      ),
    );
    final stored = await V2WorkStore('test-user').read('tasks');
    expect((stored!['current'] as List).single['executionStarted'], isTrue);
    await sub.cancel();
    await flush();
    tracker = V2WorkTaskTracker(
      repository: tasks,
      createTask: CreateDeviceTaskUseCase(path),
      resolveScope: (_) async => const WorkScope(siteId: 2, orgId: 3),
      userId: () => 'test-user',
      store: V2WorkStore('test-user'),
      mqtt: mqtt,
      diagnostics: WorkTraceDiagnostics(write: traceLogs.add),
    );
    sub = tracker.watch('MC700-test').listen(updates.add);
    await tracker.load('MC700-test');
    await flush();
    expect(tracker.current('MC700-test')!.rawStatus, 'NEW');
    expect(tracker.current('MC700-test')!.canRecordTrace, isTrue);
    mqtt.arrive({'lat': 31.1, 'lng': 120.1});
    expect(tracker.current('MC700-test')!.completedPoints, hasLength(1));
  });

  for (final status in ['EXECUTING', 'PAUSE', 'FAILED']) {
    test('$status之后的延迟NEW不能回退执行状态或解除暂停', () async {
      await tracker.load('MC700-test');
      await flush();
      mqtt.status(status);
      mqtt.status('NEW');
      expect(tracker.current('MC700-test')!.rawStatus, status);
      tasks.tasks = [FakeTasks.entity(state: 'NEW')];
      await tracker.load('MC700-test');
      expect(tracker.current('MC700-test')!.rawStatus, status);
      mqtt.arrive({'lat': 31.1, 'lng': 120.1});
      expect(
        tracker.current('MC700-test')!.completedPoints,
        hasLength(status == 'EXECUTING' ? 1 : 0),
      );
      mqtt.status('EXECUTING');
      mqtt.arrive({'lat': 31.2, 'lng': 120.2});
      expect(tracker.current('MC700-test')!.canRecordTrace, isTrue);
      mqtt.status('FINISH');
      mqtt.status('NEW');
      expect(tracker.current('MC700-test')!.state, WorkRunState.finished);
    });
  }

  test('开始、NEW到达、实时车位、暂停与结束连通，GPS不写入任务到达点', () async {
    tasks.tasks = [];
    await tracker.start(
      const WorkStartRequest(
        deviceId: 'MC700-test',
        route: WorkRouteOption(id: '8', name: 'route'),
      ),
    );
    await flush();
    final trace = WorkTraceRecorder();
    void sync() {
      final current = tracker.current('MC700-test')!;
      trace.synchronize(
        deviceId: current.deviceId,
        input: WorkTraceInput(
          taskId: current.taskId,
          recording: current.canRecordTrace,
          visible: current.state != WorkRunState.finished,
          completedPoints: current.completedPoints,
        ),
      );
    }

    const a = LatLng(31.1, 120.1);
    const b = LatLng(31.1001, 120.1001);
    const c = LatLng(31.1002, 120.1002);
    sync();
    trace.addPosition(a);
    trace.addPosition(b);
    expect(trace.points, [a, b]);
    mqtt.arrive({'lat': b.latitude, 'lng': b.longitude}, taskId: 100);
    sync();
    trace.addPosition(c);
    expect(trace.points, [b, c]);
    expect(tracker.current('MC700-test')!.completedPoints, [b]);
    await tracker.command(
      'MC700-test',
      tracker.current('MC700-test')!,
      'pause',
    );
    mqtt.status('NEW', taskId: 100);
    sync();
    expect(trace.addPosition(a), isFalse);
    expect(trace.points, [b, c]);
    mqtt.status('FINISH', taskId: 100);
    sync();
    expect(trace.points, isEmpty);
    expect((await tracker.history('MC700-test')).single.completedPoints, [b]);
  });

  test('failed query cannot create or erase a running task', () async {
    await tracker.load('MC700-test');
    await flush();
    tasks.error = 'offline';
    expect((await tracker.load('MC700-test')).isOk, isFalse);
    expect(tracker.current('MC700-test')!.taskId, '42');
    expect(path.creations, 0);
  });

  test(
    'V2 draft store serializes writes and isolates users from legacy keys',
    () async {
      final store = V2WorkStore('first');
      final points = <int>[1, 2, 3];
      final first = store.write('editor/2/device', {'points': points});
      points.clear();
      await first;
      expect((await store.read('editor/2/device'))!['points'], [1, 2, 3]);
      expect(await V2WorkStore('second').read('editor/2/device'), isNull);
      await store.write('editor/2/device', null);
      expect(await store.read('editor/2/device'), isNull);
    },
  );
  test(
    'missed FINISH push is recovered from task pool with its arrival points',
    () async {
      await tracker.load('MC700-test');
      await flush();
      mqtt.arrive({'lat': 31.1, 'lng': 120.1});
      await flush();
      tasks.tasks = [FakeTasks.entity(state: 'FINISH')];
      expect((await tracker.load('MC700-test')).data, isEmpty);
      await flush();
      expect(updates.last!.rawStatus, 'FINISH');
      expect(updates.last!.completedPoints, hasLength(1));
      expect(
        traceLogs.any(
          (s) => s.contains('[结束任务池确认]') && s.contains('"source":"task_pool"'),
        ),
        isTrue,
      );
      expect(
        traceLogs.any(
          (s) => s.contains('[结束状态落地]') && s.contains('"source":"task_pool"'),
        ),
        isTrue,
      );
      expect(traceLogs.where((s) => s.contains('[结束消息收到]')), isEmpty);
    },
  );

  test(
    'queue receives other task completion without replacing current task',
    () async {
      tasks.tasks = [
        FakeTasks.entity(),
        FakeTasks.entity(id: 43, state: 'NEW'),
      ];
      final changes = <WorkTaskSnapshot?>[];
      final queue = tracker
          .watch('MC700-test', allTasks: true)
          .listen(changes.add);
      await tracker.load('MC700-test');
      await flush();
      mqtt.status('FINISH', taskId: 43);
      await flush();
      expect(changes.last!.taskId, '43');
      expect(changes.last!.state, WorkRunState.finished);
      expect(updates.last!.taskId, '42');
      expect(updates.last!.state, WorkRunState.running);
      expect(
        traceLogs.any(
          (s) =>
              s.contains('[结束状态落地]') &&
              s.contains('"taskId":"43"') &&
              s.contains('"selected":false'),
        ),
        isTrue,
      );
      expect(mqtt.topics, {'task/42/status', 'task/42/arrive'});
      await queue.cancel();
    },
  );

  for (final status in ['CANCELED', 'CANCELLED']) {
    test('$status结束信号独立记录并保留原有结束处理', () async {
      await tracker.load('MC700-test');
      await flush();
      mqtt.status(status);
      await flush();
      expect(tracker.current('MC700-test')!.state, WorkRunState.finished);
      expect(
        traceLogs.any((s) => s.contains('[结束消息收到]') && s.contains(status)),
        isTrue,
      );
      expect(
        traceLogs.any(
          (s) => s.contains('[结束状态落地]') && s.contains('"source":"mqtt_status"'),
        ),
        isTrue,
      );
      expect(mqtt.topics, isEmpty);
    });
  }

  test('结束消息过滤记录ID不匹配及重复原因，不重复完成任务', () async {
    await tracker.load('MC700-test');
    await flush();
    mqtt.status('FINISH', deviceId: 'other-device');
    mqtt.status('FINISH', taskId: 99, topicId: 42);
    expect(tracker.current('MC700-test')!.state, WorkRunState.running);
    expect(
      traceLogs.any((s) => s.contains('[结束消息过滤]') && s.contains('deviceId不匹配')),
      isTrue,
    );
    expect(
      traceLogs.any(
        (s) => s.contains('[结束消息过滤]') && s.contains('消息taskId与topic任务不一致'),
      ),
      isTrue,
    );
    mqtt.status('FINISH');
    mqtt.status('FINISH');
    await flush();
    expect(
      traceLogs.any(
        (s) => s.contains('[结束消息过滤]') && s.contains('任务已结束，忽略重复或迟到消息'),
      ),
      isTrue,
    );
    expect(traceLogs.where((s) => s.contains('[结束状态落地]')), hasLength(1));
    expect(
      updates.where((t) => t?.state == WorkRunState.finished),
      hasLength(1),
    );
    expect(await tracker.history('MC700-test'), hasLength(1));
  });

  test('空到达点日志记录entity是否缺失，保持原有结束语义', () async {
    await tracker.load('MC700-test');
    await flush();
    mqtt.messages.add(
      MqttMessage(
        topic: 'task/42/arrive',
        payload: jsonEncode({
          'type': 'device_task_arrive_point',
          'deviceId': 'MC700-test',
          'taskId': 42,
        }),
      ),
    );
    await flush();
    expect(tracker.current('MC700-test')!.state, WorkRunState.finished);
    expect(
      traceLogs.any(
        (s) => s.contains('[结束消息收到]') && s.contains('"entityPresent":false'),
      ),
      isTrue,
    );
  });

  test('未收到结束消息时定期说明等待状态，不能把消息静默当成已结束', () async {
    await tracker.load('MC700-test');
    await flush();
    expect(mqtt.isConnected, isTrue);
    // 真实异步订阅与计时器共用同一时钟，验证生产中的10秒诊断周期。
    await Future<void>.delayed(const Duration(seconds: 11));
    expect(tracker.current('MC700-test')!.state, WorkRunState.running);
    expect(
      traceLogs.any(
        (s) =>
            s.contains('[结束确认等待]') &&
            s.contains('task/42/status') &&
            s.contains('"statusReceived":0'),
      ),
      isTrue,
    );
    expect(traceLogs.where((s) => s.contains('[结束状态落地]')), isEmpty);
    await sub.cancel();
    await flush();
    expect(mqtt.topics, isEmpty);
  });

  test(
    'last page exit unsubscribes and reentry resubscribes the active task',
    () async {
      await tracker.load('MC700-test');
      await flush();
      await sub.cancel();
      await flush();
      expect(mqtt.topics, isEmpty);
      expect(mqtt.isConnected, isFalse);
      sub = tracker.watch('MC700-test').listen(updates.add);
      await tracker.load('MC700-test');
      await flush();
      expect(mqtt.topics, {'task/42/status', 'task/42/arrive'});
      mqtt.status('FINISH');
      await flush();
      expect(updates.last!.state, WorkRunState.finished);
    },
  );
}