fix(queue): guarantee a completion, and make the state machine testable

app/lib/services/job_runner.dart

@@ -0,0 +1,48 @@
+import '../models/progress_info.dart';
+import '../models/video_job.dart';
+import 'worker_manager.dart' show CompletionResult, LogMessage;
+
+/// What [MainViewModel] needs from the thing that runs jobs.
+///
+/// This exists as a seam for testing, and the reason is worth stating: three
+/// separate fixes for the same cancellation hang shipped without ever being
+/// exercised, because `MainViewModel` constructed its `WorkerManager` directly
+/// and so the queue state machine could not be driven from a test at all. Every
+/// fix rested on reading the code, and two of them were "verified" by tests that
+/// only read the source text.
+///
+/// The interface is deliberately the *observed* surface — the streams the
+/// viewmodel listens to and the three calls it makes — not everything
+/// WorkerManager can do. Anything outside this (GPU probing, process groups)
+/// stays on the concrete class.
+abstract interface class JobRunner {
+  /// Progress updates for the running job.
+  Stream<ProgressInfo> get progressStream;
+
+  /// Log lines from the worker.
+  Stream<LogMessage> get logStream;
+
+  /// Exactly one event per started job — see [startJob].
+  Stream<CompletionResult> get completionStream;
+
+  /// Whether a job is currently running.
+  bool get isRunning;
+
+  /// Starts [job].
+  ///
+  /// Implementations must emit exactly one [CompletionResult] for every call
+  /// that returns normally: no more (the UI would act twice) and no fewer (the
+  /// UI latches on `cancelling`/`processing` and never returns to idle, which is
+  /// the hang this interface was introduced to make testable).
+  Future<void> startJob(VideoJob job);
+
+  /// Cancels the running job.
+  ///
+  /// Must still emit a completion when there is nothing to kill — the caller has
+  /// already moved the UI into `cancelling` by the time this is called, so
+  /// emitting nothing strands it there with Start and Cancel both disabled.
+  Future<void> cancel();
+
+  /// Releases resources.
+  void dispose();
+}

app/lib/services/tool_locator.dart

@@ -111,6 +111,16 @@ class ToolLocator {
 
   /// Resolve the vapourbox-worker executable path.
   String? _resolveWorker() {
+    // Explicit override, mirroring VAPOURBOX_DEPS_DIR above. Under `flutter
+    // test` the resolved executable is the test runner, not the app bundle, so
+    // neither the production nor the dev path below can find the worker — which
+    // left WorkerManager, and therefore the whole cancel/restart flow, with no
+    // way to be tested against a real worker at all.
+    final override = Platform.environment['VAPOURBOX_WORKER'];
+    if (override != null && override.isNotEmpty && File(override).existsSync()) {
+      return override;
+    }
+
     final exeDir = path.dirname(Platform.resolvedExecutable);
     final ext = Platform.isWindows ? '.exe' : '';
     final workerExe = 'vapourbox-worker$ext';

app/lib/services/worker_manager.dart

@@ -2,14 +2,17 @@ import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
+import 'package:flutter/foundation.dart';
+
 import '../models/progress_info.dart';
 import '../models/video_job.dart';
+import 'job_runner.dart';
 import 'process_tree.dart';
 import 'temp_directory_service.dart';
 import 'tool_locator.dart';
 
 /// Manages the worker process lifecycle and IPC.
-class WorkerManager {
+class WorkerManager implements JobRunner {
   Process? _process;
   StreamSubscription<String>? _stdoutSubscription;
   StreamSubscription<String>? _stderrSubscription;
@@ -51,31 +54,46 @@ class WorkerManager {
   /// "Worker exited with code 143".
   bool _cancelRequested = false;
 
+  /// Whether a started job still owes a [CompletionResult].
+  ///
+  /// `_completionEmitted` enforces "at most one" completion per job. This
+  /// enforces the other half — "at least one" — which nothing did before, and
+  /// whose absence is the hang: `cancel()` returned silently when there was no
+  /// process, the viewmodel had already latched `_state = cancelling`, and with
+  /// no event to retire that latch the UI sat with Start *and* Cancel disabled.
+  bool _jobInFlight = false;
+
   /// Stream of progress updates from the worker.
   final _progressController = StreamController<ProgressInfo>.broadcast();
+  @override
   Stream<ProgressInfo> get progressStream => _progressController.stream;
 
   /// Stream of log messages from the worker.
   final _logController = StreamController<LogMessage>.broadcast();
+  @override
   Stream<LogMessage> get logStream => _logController.stream;
 
   /// Stream of completion events.
   final _completionController = StreamController<CompletionResult>.broadcast();
+  @override
   Stream<CompletionResult> get completionStream => _completionController.stream;
 
   /// Whether the worker is currently running.
+  @override
   bool get isRunning => _process != null;
 
   /// Starts a deinterlacing job.
   ///
   /// Creates a temporary JSON config file and spawns the worker process.
+  @override
   Future<void> startJob(VideoJob job) async {
     if (_process != null) {
       throw StateError('Worker is already running');
     }
 
     _completionEmitted = false;
     _cancelRequested = false;
+    _jobInFlight = true;
     final generation = ++_generation;
 
     final toolLocator = ToolLocator.instance;
@@ -99,6 +117,7 @@ class WorkerManager {
         workingDirectory: File(workerPath).parent.path,
       );
 
+
       // Listen to stdout for JSON messages
       _stdoutSubscription = _process!.stdout
           .transform(utf8.decoder)
@@ -220,17 +239,40 @@ class WorkerManager {
   /// Waits for the worker to genuinely exit rather than assuming it has, so its
   /// children are torn down by the worker itself. Only escalates to SIGKILL if
   /// it is still alive after [_shutdownGrace].
+  @override
   Future<void> cancel() async {
     // Hold a local reference: `_cleanup()` nulls the field, and the old code
     // tested `_process != null` *before* that ran, so its "force kill if still
     // running" check was never actually false.
     final process = _process;
-    if (process == null) return;
     // Which job this call is cancelling. Everything after the await below is
     // conditional on it still being the current one.
     final generation = _generation;
     _cancelRequested = true;
 
+
+    if (process == null) {
+      // Nothing to signal — the job finished microseconds ago, `startJob` threw,
+      // or we are still inside `preparingJob` and the process does not exist
+      // yet. That last case is reachable from the UI: `canCancel` includes
+      // `preparingJob`, so the Cancel button is live before there is anything to
+      // cancel.
+      //
+      // The caller has already moved the UI into `cancelling`, so returning
+      // silently here is what stranded it: `cancelling` is neither `idle` nor in
+      // `canCancel`, leaving Start and Cancel both disabled with no way back.
+      // Report the cancellation so the latch can be retired.
+      if (_jobInFlight) {
+        _emitCompletion(const CompletionResult(
+          success: false,
+          cancelled: true,
+          errorMessage: 'Job cancelled by user',
+        ));
+        _cleanup();
+      }
+      return;
+    }
+
     if (Platform.isWindows) {
       // No SIGTERM on Windows, and Process.kill maps to TerminateProcess, which
       // does not touch children. taskkill /T walks the tree, so nothing is
@@ -310,7 +352,14 @@ class WorkerManager {
     required String? lastError,
     required int exitCode,
   }) {
-    if (cancelRequested) {
+    // Cancellation is observed two ways, and either is sufficient.
+    //
+    // `cancelRequested` is what the app asked for. Exit 130 is what the worker
+    // actually did (`worker/src/main.rs` returns it once the cancel flag is
+    // seen). Taking both means a cancellation initiated anywhere — including a
+    // SIGINT the app never issued — is still reported as one, rather than being
+    // inferred solely from a flag the app sets for itself.
+    if (cancelRequested || exitCode == 130) {
       return const CompletionResult(
         success: false,
         cancelled: true,
@@ -331,8 +380,11 @@ class WorkerManager {
 
   /// Emit [result] unless this job has already reported one.
   void _emitCompletion(CompletionResult result) {
-    if (_completionEmitted || _completionController.isClosed) return;
+    if (_completionEmitted || _completionController.isClosed) {
+      return;
+    }
     _completionEmitted = true;
+    _jobInFlight = false;
     _completionController.add(result);
   }
 
@@ -396,6 +448,7 @@ class WorkerManager {
   }
 
   /// Disposes of resources.
+  @override
   void dispose() {
     cancel();
     _progressController.close();

app/lib/viewmodels/main_viewmodel.dart

@@ -21,11 +21,14 @@ import '../services/frame_math.dart';
 import '../services/preset_service.dart';
 import '../services/preview_generator.dart';
 import '../services/temp_directory_service.dart';
+import '../services/job_runner.dart';
 import '../services/worker_manager.dart';
 
 /// Main view model managing application state.
 class MainViewModel extends ChangeNotifier {
-  final WorkerManager _workerManager = WorkerManager();
+  /// Runs jobs. Injectable so the queue state machine can be driven from a
+  /// test — see JobRunner for why that matters.
+  final JobRunner _workerManager;
   final FieldOrderDetector _fieldOrderDetector = FieldOrderDetector();
   final PreviewGenerator _previewGenerator = PreviewGenerator();
   final DvdService _dvdService = DvdService();
@@ -337,7 +340,8 @@ class MainViewModel extends ChangeNotifier {
     return _manualFieldOrder;
   }
 
-  MainViewModel() {
+  MainViewModel({JobRunner? runner})
+      : _workerManager = runner ?? WorkerManager() {
     _setupSubscriptions();
     _initializePreviewGenerator();
     _probeGpuCapabilities();
@@ -1088,11 +1092,10 @@ class MainViewModel extends ChangeNotifier {
 
     if (nextIndex == -1) {
       // No more items to process
-      _isQueueProcessing = false;
-      _currentProcessingIndex = -1;
-      // Show failed state if any items failed, otherwise completed
       final hasFailed = _queue.any((q) => q.status == QueueItemStatus.failed);
-      _state = hasFailed ? ProcessingState.failed : ProcessingState.completed;
+      _stopProcessing(
+        state: hasFailed ? ProcessingState.failed : ProcessingState.completed,
+      );
       notifyListeners();
       return;
     }
@@ -1193,16 +1196,26 @@ class MainViewModel extends ChangeNotifier {
   /// Handles completion of a queue item.
   Future<void> _handleQueueItemCompletion(CompletionResult result) async {
     if (_currentProcessingIndex < 0 || _currentProcessingIndex >= _queue.length) {
+      // A completion with no item to attribute it to. Dropping it silently is
+      // what poisons the app: both latches stay set, `canProcess` stays false
+      // and `startQueueProcessing` returns at its guard, so the UI spins with
+      // Start disabled and no way back short of a restart.
+      //
+      // The event is unattributable, not meaningless — the worker has stopped.
+      // Stand down rather than latch.
+      _stopProcessing();
+      notifyListeners();
       return;
     }
 
     final item = _queue[_currentProcessingIndex];
 
     if (result.cancelled) {
       item.status = QueueItemStatus.cancelled;
-      _isQueueProcessing = false;
-      _currentProcessingIndex = -1;
-      _state = ProcessingState.idle;
+      // Cancel stops the whole queue; remaining ready items stay ready for a
+      // later Start. _stopProcessing runs after the status is set, so this item
+      // is not caught by its rescue of stranded `processing` items.
+      _stopProcessing();
     } else if (result.success) {
       item.status = QueueItemStatus.completed;
       // Process next item
@@ -1223,6 +1236,24 @@ class MainViewModel extends ChangeNotifier {
     notifyListeners();
   }
 
+  /// Return the queue to a state the user can act on.
+  ///
+  /// Every path that stops processing goes through here, so recovery cannot be
+  /// forgotten on one of them. In particular it rescues items stranded in
+  /// `processing`: that status is neither `canProcess` nor `canReprocess`, so
+  /// `startQueueProcessing`'s reset loop skips it and the item can never be run
+  /// again — the queue stays permanently unusable for that file.
+  void _stopProcessing({ProcessingState state = ProcessingState.idle}) {
+    for (final item in _queue) {
+      if (item.status == QueueItemStatus.processing) {
+        item.status = QueueItemStatus.ready;
+      }
+    }
+    _isQueueProcessing = false;
+    _currentProcessingIndex = -1;
+    _state = state;
+  }
+
   /// Gets the effective field order for a queue item.
   FieldOrder _getEffectiveFieldOrder(QueueItem item) {
     if (_autoFieldOrder && item.videoInfo?.fieldOrder != null) {
@@ -1255,6 +1286,29 @@ class MainViewModel extends ChangeNotifier {
     } else {
       await _workerManager.cancel();
     }
+
+    // A completion emitted during the cancel is delivered asynchronously
+    // (broadcast stream), so checking immediately races it — and winning that
+    // race is worse than losing it: the queue's own handler never runs, the item
+    // is never marked `cancelled`, and recovery happens by fallback instead of
+    // by design. Give delivery a turn of the event loop first.
+    await Future<void>.delayed(Duration.zero);
+    await Future<void>.delayed(Duration.zero);
+
+    // By now a completion should have retired the `cancelling` latch. If one
+    // never arrived, stand down anyway: `cancelling` is neither `idle` nor in
+    // `canCancel`, so staying there disables Start *and* Cancel with no way back
+    // short of restarting the app. A missed event should cost the user a
+    // mislabelled item, not a dead window.
+    if (_state == ProcessingState.cancelling) {
+      _logMessages.add(LogMessage(
+        level: LogLevel.warning,
+        message: 'Cancellation completed without a result from the worker; '
+            'returning to idle.',
+      ));
+      _stopProcessing();
+      notifyListeners();
+    }
   }
 
   /// Retries all failed items in the queue.

app/test/cancel_shutdown_grace_test.dart

@@ -21,6 +21,12 @@
 // This is the same class of guard as `test_native_formats_match_pipe_source`
 // (Rust `NATIVE_FORMATS` vs Python `_FORMAT_MAP`).
 
+// NOTE: always normalise line endings when reading source here. git checks these
+// files out CRLF on Windows, and every scan below is written against "\n". This
+// has broken CI three separate times — twice with a confusing "expected a value
+// less than N" and once with a RangeError from indexOf returning -1. If a scan
+// can be replaced by a behavioural assertion, prefer that; these read source
+// only because the thing they guard is a code shape, not an observable.
 import 'dart:io';
 
 import 'package:path/path.dart' as p;
@@ -50,10 +56,12 @@ void main() {
     test('the app gives the worker longer than its cancellation poll interval',
         () {
       final rs = File(p.join(root, 'worker', 'src', 'pipeline_executor.rs'))
-          .readAsStringSync();
+          .readAsStringSync()
+          .replaceAll('\r\n', '\n');
       final dart = File(p.join(
               root, 'app', 'lib', 'services', 'worker_manager.dart'))
-          .readAsStringSync();
+          .readAsStringSync()
+          .replaceAll('\r\n', '\n');
 
       final pollMatch = RegExp(
         r'let\s+progress_interval\s*=\s*Duration::from_millis\((\d+)\)',
@@ -86,10 +94,16 @@ void main() {
     test('cancel waits for the process to exit instead of sleeping', () {
       final dart = File(p.join(
               root, 'app', 'lib', 'services', 'worker_manager.dart'))
-          .readAsStringSync();
+          .readAsStringSync()
+          .replaceAll('\r\n', '\n');
       final cancelStart = dart.indexOf('Future<void> cancel()');
       expect(cancelStart, greaterThan(-1));
-      final body = dart.substring(cancelStart, dart.indexOf('_cleanup();', cancelStart));
+      // Read the whole method, not up to the first `_cleanup()`. cancel() now
+      // has an early branch for "nothing to kill" which cleans up before the
+      // await, so anchoring on the first occurrence pointed at that instead —
+      // a false failure with the behaviour perfectly correct.
+      final body =
+          dart.substring(cancelStart, dart.indexOf('\n  }\n', cancelStart));
 
       // It must observe the exit, not assume it after a fixed delay.
       expect(body, contains('exitCode'),