acyclic-fs 0.1.5

Rust SDK for versioned workspaces and embedded native mounts
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
//! Bounded native change notifications for materialized checkouts.
//!
//! Watch events are acceleration hints, never filesystem truth. Consumers read
//! the affected paths from the host and produce ordinary authenticated checkout
//! mutations. Queue overflow, backend uncertainty, ambiguous rename, root
//! replacement, restart, or an unrepresentable path invalidates the hint stream
//! and requires a new bounded baseline scan.

use crate::NativeRootIdentity;
use crate::cancellation::{CancellationError, CancellationToken};
use crate::kernel::{LogicalName, NamespacePath, NamespacePathError};
use crate::model::{FilesystemProfile, VolumeLimits};
use crate::performance::{
    MeasuredResult, OperationFailure, OperationReceipt, WorkBudget, WorkCounters, WorkError,
};
use notify::event::{ModifyKind, RenameMode};
use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use std::collections::BTreeSet;
use std::mem::size_of;
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::mpsc::{Receiver, SyncSender, TryRecvError, TrySendError, sync_channel};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
use thiserror::Error;

/// Native notification mechanism compiled for this target.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum NativeWatchBackend {
    /// Linux inotify through the reviewed `notify` adapter.
    LinuxInotify,
    /// macOS `FSEvents` through the reviewed `notify` adapter.
    MacosFsevents,
    /// Windows `ReadDirectoryChangesW` through the reviewed `notify` adapter.
    WindowsReadDirectoryChanges,
    /// No target-specific native notification mechanism is available.
    Unsupported,
}

impl NativeWatchBackend {
    /// Stable SDK spelling used by generated-language capability surfaces.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::LinuxInotify => "linux-inotify",
            Self::MacosFsevents => "macos-fsevents",
            Self::WindowsReadDirectoryChanges => "windows-read-directory-changes",
            Self::Unsupported => "unsupported",
        }
    }
}

/// Exact guarantees supplied by the compiled native watcher.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct NativeWatchCapabilities {
    /// Host notification mechanism selected at compile time.
    pub backend: NativeWatchBackend,
    /// Whether descendant changes can be requested recursively.
    pub recursive: bool,
    /// Whether a persisted cursor can prove continuity across process restart.
    ///
    /// When false, reopening always requires the authenticated baseline
    /// handshake. A process-local [`WatchSequence`] is never a restart token.
    pub persistent_restart: bool,
    /// Whether every poll fences replacement of the admitted root identity.
    pub root_identity_fencing: bool,
}

/// Returns compile-time native watcher guarantees without touching a user path.
#[must_use]
pub const fn native_watch_capabilities() -> NativeWatchCapabilities {
    #[cfg(target_os = "linux")]
    let backend = NativeWatchBackend::LinuxInotify;
    #[cfg(target_os = "macos")]
    let backend = NativeWatchBackend::MacosFsevents;
    #[cfg(target_os = "windows")]
    let backend = NativeWatchBackend::WindowsReadDirectoryChanges;
    #[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
    let backend = NativeWatchBackend::Unsupported;

    NativeWatchCapabilities {
        backend,
        recursive: !matches!(backend, NativeWatchBackend::Unsupported),
        persistent_restart: false,
        root_identity_fencing: cfg!(any(unix, windows)),
    }
}

/// Process-local watcher epoch. It changes after every completed baseline scan.
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct WatchEpoch(u64);

impl WatchEpoch {
    /// Constructs an epoch from a persisted process-local adapter value.
    #[must_use]
    pub const fn from_u64(value: u64) -> Self {
        Self(value)
    }

    /// Returns the monotonic process-local value.
    #[must_use]
    pub const fn get(self) -> u64 {
        self.0
    }
}

/// Contiguous sequence within one [`WatchEpoch`].
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct WatchSequence(u64);

impl WatchSequence {
    /// Constructs a contiguous sequence from an adapter cursor.
    #[must_use]
    pub const fn from_u64(value: u64) -> Self {
        Self(value)
    }

    /// Returns the contiguous process-local value.
    #[must_use]
    pub const fn get(self) -> u64 {
        self.0
    }
}

/// One bounded host-side change hint.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum WatchChange {
    /// A namespace entry appeared.
    Created(NamespacePath),
    /// File content or an unspecified property may have changed.
    Modified(NamespacePath),
    /// Metadata or named attributes may have changed.
    MetadataChanged(NamespacePath),
    /// A namespace entry disappeared.
    Removed(NamespacePath),
    /// One paired rename with exact old and new relative paths.
    Renamed {
        /// Old relative path.
        from: NamespacePath,
        /// New relative path.
        to: NamespacePath,
    },
}

/// Why watcher hints can no longer prove a contiguous change interval.
#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
pub enum WatchInvalidationReason {
    /// Every newly opened watcher starts without an authenticated baseline.
    #[error("an initial authenticated baseline is required")]
    InitialSnapshotRequired,
    /// The bounded notification queue overflowed.
    #[error("the native notification queue overflowed")]
    QueueOverflow,
    /// The native backend reported lost or unreliable events.
    #[error("the native backend requires a rescan")]
    NativeRescanRequired,
    /// The native backend returned an asynchronous error.
    #[error("the native backend reported an asynchronous error")]
    BackendError,
    /// A native path escaped or could not be represented by volume semantics.
    #[error("a native path cannot be represented by volume semantics")]
    UnrepresentablePath,
    /// A rename could not be paired without guessing.
    #[error("a native rename could not be paired exactly")]
    AmbiguousRename,
    /// The watched root itself was removed, renamed, or replaced.
    #[error("the watched root changed identity")]
    RootChanged,
}

/// One bounded watcher poll result.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum WatchBatch {
    /// A contiguous process-local interval of change hints.
    Changes {
        /// Baseline epoch against which paths are interpreted.
        epoch: WatchEpoch,
        /// Sequence assigned to the first returned change.
        first_sequence: WatchSequence,
        /// Sequence to request after applying every returned change.
        next_sequence: WatchSequence,
        /// Bounded hints moved from the native queue without path copies.
        changes: Vec<WatchChange>,
    },
    /// All hints from this epoch must be discarded and a baseline rescanned.
    RescanRequired {
        /// Invalidated watcher epoch.
        epoch: WatchEpoch,
        /// Exact invalidation class.
        reason: WatchInvalidationReason,
    },
}

/// Bounded native watcher configuration.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct NativeWatchOptions {
    /// Volume path/name bounds used while converting host paths.
    pub limits: VolumeLimits,
    /// Maximum queued native changes before typed invalidation.
    pub maximum_queued_changes: u32,
    /// Whether descendants are observed recursively.
    pub recursive: bool,
}

#[cfg(unix)]
fn open_native_root(path: &Path) -> std::io::Result<std::fs::File> {
    use std::os::unix::fs::OpenOptionsExt;
    let mut options = std::fs::OpenOptions::new();
    options
        .read(true)
        .custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW);
    options.open(path)
}

#[cfg(windows)]
fn open_native_root(path: &Path) -> std::io::Result<std::fs::File> {
    use std::os::windows::fs::MetadataExt;
    use std::os::windows::fs::OpenOptionsExt;
    use windows::Win32::Storage::FileSystem::{
        FILE_ATTRIBUTE_REPARSE_POINT, FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT,
        FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE,
    };
    let mut options = std::fs::OpenOptions::new();
    options
        .read(true)
        .share_mode(FILE_SHARE_READ.0 | FILE_SHARE_WRITE.0 | FILE_SHARE_DELETE.0)
        .custom_flags(FILE_FLAG_BACKUP_SEMANTICS.0 | FILE_FLAG_OPEN_REPARSE_POINT.0);
    let file = options.open(path)?;
    if file.metadata()?.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT.0 != 0 {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "watch root is a reparse point",
        ));
    }
    Ok(file)
}

#[cfg(not(any(unix, windows)))]
fn open_native_root(path: &Path) -> std::io::Result<std::fs::File> {
    std::fs::File::open(path)
}

const fn native_filesystem_profile() -> FilesystemProfile {
    #[cfg(unix)]
    {
        FilesystemProfile::Posix
    }
    #[cfg(windows)]
    {
        FilesystemProfile::Windows
    }
    #[cfg(not(any(unix, windows)))]
    {
        FilesystemProfile::Portable
    }
}

impl NativeWatchOptions {
    /// Conservative recursive defaults.
    #[must_use]
    pub const fn new(limits: VolumeLimits) -> Self {
        Self {
            limits,
            maximum_queued_changes: 16_384,
            recursive: true,
        }
    }

    fn validate(self) -> Result<Self, NativeWatchError> {
        if self.maximum_queued_changes == 0 {
            return Err(NativeWatchError::InvalidOptions);
        }
        Ok(self)
    }
}

#[derive(Debug)]
struct SharedState {
    invalidation: Option<WatchInvalidationReason>,
    #[cfg(target_os = "linux")]
    pending_rename: Option<PendingRename>,
    /// Host path of the outstanding fence cookie, cleared on delivery.
    fence: Option<PathBuf>,
    /// Host subtrees whose events are never hints, in fence preference order.
    excluded: Vec<PathBuf>,
}

#[cfg(target_os = "linux")]
#[derive(Debug)]
struct PendingRename {
    tracker: usize,
    from: PathBuf,
    to: Option<PathBuf>,
}

impl SharedState {
    const fn new(invalidation: Option<WatchInvalidationReason>) -> Self {
        Self {
            invalidation,
            #[cfg(target_os = "linux")]
            pending_rename: None,
            fence: None,
            excluded: Vec::new(),
        }
    }

    fn seal_invalidation(&mut self) -> Option<WatchInvalidationReason> {
        #[cfg(target_os = "linux")]
        if self.invalidation.is_none() && self.pending_rename.is_some() {
            self.invalidation = Some(WatchInvalidationReason::AmbiguousRename);
        }
        self.invalidation
    }
}

struct NativeEventContext {
    root: PathBuf,
    root_identity: NativeRootIdentity,
    profile: FilesystemProfile,
    limits: VolumeLimits,
    #[cfg(target_os = "linux")]
    maximum_queued_changes: u32,
}

/// One live native watcher over a materialized checkout root.
///
/// Opening starts invalidated. Call [`Self::begin_rescan`], build an exact
/// baseline while notifications continue to queue, and call
/// [`Self::finish_rescan`] before polling. The same handshake repairs overflow
/// without losing changes concurrent with the baseline scan.
pub struct NativeWatch {
    watcher: RecommendedWatcher,
    root: PathBuf,
    recursive: bool,
    watched_directories: BTreeSet<PathBuf>,
    receiver: Receiver<WatchChange>,
    queued: Arc<AtomicU32>,
    shared: Arc<Mutex<SharedState>>,
    fenced: Arc<Condvar>,
    fences: u64,
    excluded: Vec<(NamespacePath, PathBuf)>,
    epoch: WatchEpoch,
    next_sequence: WatchSequence,
    rescan_in_progress: bool,
    root_identity: NativeRootIdentity,
}

impl NativeWatch {
    /// Returns exact guarantees for this watcher's compiled backend.
    #[must_use]
    pub const fn capabilities(&self) -> NativeWatchCapabilities {
        native_watch_capabilities()
    }

    /// Opens the platform-recommended event source.
    ///
    /// Linux uses inotify, macOS uses `FSEvents`, and Windows uses
    /// `ReadDirectoryChangesW` through the reviewed `notify` adapter. Native
    /// journals can later implement the same invalidation protocol.
    ///
    /// # Errors
    ///
    /// Rejects invalid bounds, an unavailable/non-directory root, allocation
    /// conversion, poisoned synchronization, or native watcher startup failure.
    pub fn open(
        root: impl AsRef<Path>,
        options: NativeWatchOptions,
    ) -> Result<Self, NativeWatchError> {
        Self::open_with_profile(root, native_filesystem_profile(), options)
    }

    /// Opens a watcher whose emitted names use the supplied volume profile.
    ///
    /// Use this when watcher paths feed a volume whose profile can differ from
    /// the host-native profile selected by [`Self::open`].
    ///
    /// # Errors
    ///
    /// Rejects invalid bounds, an unavailable/non-directory root, allocation
    /// conversion, poisoned synchronization, or native watcher startup failure.
    pub fn open_with_profile(
        root: impl AsRef<Path>,
        profile: FilesystemProfile,
        options: NativeWatchOptions,
    ) -> Result<Self, NativeWatchError> {
        let options = options.validate()?;
        let requested_root = root.as_ref();
        let admission = requested_root
            .symlink_metadata()
            .map_err(|error| NativeWatchError::Io(error.to_string()))?;
        if admission.file_type().is_symlink() {
            return Err(NativeWatchError::RootIsNotDirectory);
        }
        let root = requested_root
            .canonicalize()
            .map_err(|error| NativeWatchError::Io(error.to_string()))?;
        let root_file =
            open_native_root(&root).map_err(|error| NativeWatchError::Io(error.to_string()))?;
        let metadata = root_file
            .metadata()
            .map_err(|error| NativeWatchError::Io(error.to_string()))?;
        if !metadata.is_dir() {
            return Err(NativeWatchError::RootIsNotDirectory);
        }
        let root_identity = NativeRootIdentity::from_file(&root_file)
            .map_err(|error| NativeWatchError::Io(error.to_string()))?;
        let capacity = usize::try_from(options.maximum_queued_changes)
            .map_err(|_| NativeWatchError::InvalidOptions)?;
        let (sender, receiver) = sync_channel(capacity);
        let queued = Arc::new(AtomicU32::new(0));
        let shared = Arc::new(Mutex::new(SharedState::new(Some(
            WatchInvalidationReason::InitialSnapshotRequired,
        ))));
        let fenced = Arc::new(Condvar::new());
        let callback_shared = Arc::clone(&shared);
        let callback_fenced = Arc::clone(&fenced);
        let callback_queued = Arc::clone(&queued);
        let callback_context = NativeEventContext {
            root: root.clone(),
            root_identity,
            profile,
            limits: options.limits,
            #[cfg(target_os = "linux")]
            maximum_queued_changes: options.maximum_queued_changes,
        };
        let mut watcher = notify::recommended_watcher(move |event: notify::Result<Event>| {
            accept_native_event(
                &event,
                &callback_context,
                &sender,
                &callback_shared,
                &callback_queued,
            );
            // Signal only after the cookie's own hint is handled: delivery is
            // ordered, so every earlier host change is already queued.
            if let (Ok(event), Ok(mut state)) = (&event, callback_shared.lock())
                && state
                    .fence
                    .as_ref()
                    .is_some_and(|fence| event.paths.contains(fence))
            {
                state.fence = None;
                callback_fenced.notify_all();
            }
        })
        .map_err(|error| NativeWatchError::Backend(error.to_string()))?;
        watcher
            .watch(
                &root,
                if options.recursive {
                    RecursiveMode::Recursive
                } else {
                    RecursiveMode::NonRecursive
                },
            )
            .map_err(|error| NativeWatchError::Backend(error.to_string()))?;
        Ok(Self {
            watcher,
            watched_directories: BTreeSet::from([root.clone()]),
            root,
            recursive: options.recursive,
            receiver,
            queued,
            shared,
            fenced,
            fences: 0,
            excluded: Vec::new(),
            epoch: WatchEpoch(0),
            next_sequence: WatchSequence(0),
            rescan_in_progress: false,
            root_identity,
        })
    }

    /// Adds one exact directory to a non-recursive native watcher.
    ///
    /// Recursive backends already cover the path and return immediately. A
    /// non-recursive watcher validates every path prefix without following a
    /// symbolic link, then installs exactly one additional subscription. No
    /// descendant enumeration occurs.
    ///
    /// # Errors
    ///
    /// Rejects an absent/non-directory path, a symbolic-link prefix, a path
    /// that cannot be represented on this host, or a backend registration
    /// failure.
    pub fn watch_directory(&mut self, directory: &NamespacePath) -> Result<(), NativeWatchError> {
        if self.recursive || directory.is_root() {
            return Ok(());
        }
        let relative = crate::native_capture::namespace_to_host_path(directory)
            .map_err(|_| NativeWatchError::UnrepresentablePath)?;
        let mut candidate = self.root.clone();
        for component in relative.components() {
            let Component::Normal(component) = component else {
                return Err(NativeWatchError::UnrepresentablePath);
            };
            candidate.push(component);
            let metadata = match candidate.symlink_metadata() {
                Ok(metadata) => metadata,
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                    candidate.pop();
                    break;
                }
                Err(error) => return Err(NativeWatchError::Io(error.to_string())),
            };
            if metadata.file_type().is_symlink() {
                return Err(NativeWatchError::SymbolicLinkDirectory);
            }
            if !metadata.is_dir() {
                return Err(NativeWatchError::ObservedPathNotDirectory);
            }
        }
        if self.watched_directories.contains(&candidate) {
            return Ok(());
        }
        self.watcher
            .watch(&candidate, RecursiveMode::NonRecursive)
            .map_err(|error| NativeWatchError::Backend(error.to_string()))?;
        self.watched_directories.insert(candidate);
        Ok(())
    }

    fn forget_removed_directories(&mut self, changes: &[WatchChange]) {
        if self.recursive {
            return;
        }
        let mut removed = Vec::new();
        for change in changes {
            match change {
                WatchChange::Removed(path) => removed.push(path),
                WatchChange::Renamed { from, to } => {
                    removed.push(from);
                    removed.push(to);
                }
                WatchChange::Created(_)
                | WatchChange::Modified(_)
                | WatchChange::MetadataChanged(_) => {}
            }
        }
        for path in removed {
            let Ok(relative) = crate::native_capture::namespace_to_host_path(path) else {
                continue;
            };
            let removed = self.root.join(relative);
            let stale = self
                .watched_directories
                .iter()
                .filter(|watched| **watched != self.root && watched.starts_with(&removed))
                .cloned()
                .collect::<Vec<_>>();
            for watched in stale {
                let _ = self.watcher.unwatch(&watched);
                self.watched_directories.remove(&watched);
            }
        }
    }

    fn reset_demand_watches(&mut self) {
        if self.recursive {
            return;
        }
        let stale = self
            .watched_directories
            .iter()
            .filter(|path| **path != self.root)
            .cloned()
            .collect::<Vec<_>>();
        for path in stale {
            let _ = self.watcher.unwatch(&path);
            self.watched_directories.remove(&path);
        }
    }

    /// Admits a demand-backed lazy baseline without enumerating the root.
    ///
    /// The paired demand source must hold the same root capability and resolve
    /// first observations from the live root. Hints arriving after this
    /// boundary remain queued and must be captured before the corresponding
    /// operation barrier closes.
    ///
    /// # Errors
    ///
    /// Returns the same admission and root-identity failures as the ordinary
    /// rescan handshake.
    pub fn accept_lazy_baseline(&mut self) -> Result<WatchBatch, NativeWatchError> {
        self.begin_rescan()?;
        self.finish_rescan()
    }

    /// Returns the exact root identity admitted when the watcher opened.
    #[must_use]
    pub const fn root_identity(&self) -> NativeRootIdentity {
        self.root_identity
    }

    /// Verifies that a separately held capture root is the directory admitted
    /// by this watcher. A mismatch permanently invalidates the current epoch.
    ///
    /// # Errors
    ///
    /// Returns a typed root-change or synchronization failure.
    pub fn verify_root_identity(
        &self,
        observed: NativeRootIdentity,
    ) -> Result<(), NativeWatchError> {
        if observed == self.root_identity {
            return Ok(());
        }
        self.shared
            .lock()
            .map_err(|_| NativeWatchError::StatePoisoned)?
            .invalidation = Some(WatchInvalidationReason::RootChanged);
        Err(NativeWatchError::RootChanged)
    }

    /// Starts a baseline scan and atomically discards only pre-scan hints.
    ///
    /// Native events arriving after this method returns remain queued. The
    /// caller must read the exact host tree before calling [`Self::finish_rescan`].
    ///
    /// # Errors
    ///
    /// Rejects a duplicate scan, epoch exhaustion, or poisoned watcher state.
    pub fn begin_rescan(&mut self) -> Result<WatchEpoch, NativeWatchError> {
        if self.rescan_in_progress {
            return Err(NativeWatchError::RescanAlreadyInProgress);
        }
        let mut shared = self
            .shared
            .lock()
            .map_err(|_| NativeWatchError::StatePoisoned)?;
        while self.receiver.try_recv().is_ok() {
            self.queued.fetch_sub(1, Ordering::AcqRel);
        }
        shared.invalidation = None;
        #[cfg(target_os = "linux")]
        {
            shared.pending_rename = None;
        }
        self.epoch = WatchEpoch(
            self.epoch
                .0
                .checked_add(1)
                .ok_or(NativeWatchError::SequenceExhausted)?,
        );
        self.next_sequence = WatchSequence(0);
        drop(shared);
        self.reset_demand_watches();
        self.rescan_in_progress = true;
        Ok(self.epoch)
    }

    #[cfg(target_os = "windows")]
    fn accept_verified_baseline<T, E>(
        &mut self,
        verify: impl FnOnce() -> Result<Option<T>, E>,
    ) -> Result<Result<Option<T>, E>, NativeWatchError> {
        if self.rescan_in_progress {
            return Err(NativeWatchError::RescanAlreadyInProgress);
        }
        let mut shared = self
            .shared
            .lock()
            .map_err(|_| NativeWatchError::StatePoisoned)?;
        if shared.invalidation != Some(WatchInvalidationReason::InitialSnapshotRequired)
            || self.epoch.0 != 0
        {
            return Err(NativeWatchError::RescanAlreadyInProgress);
        }
        let verified = match verify() {
            Ok(Some(verified)) => verified,
            Ok(None) => return Ok(Ok(None)),
            Err(error) => return Ok(Err(error)),
        };
        // A callback that ran before admission could only observe the initial
        // invalidation and would have dropped its hint. The durable verifier
        // must cover that interval. Holding the lock now fences later callbacks.
        self.epoch = WatchEpoch(
            self.epoch
                .0
                .checked_add(1)
                .ok_or(NativeWatchError::SequenceExhausted)?,
        );
        self.next_sequence = WatchSequence(0);
        shared.invalidation = None;
        Ok(Ok(Some(verified)))
    }

    /// Admits an existing Windows baseline only when the volume USN journal
    /// proves that neither the watched root nor any entry on its volume
    /// changed since the checkpoint.
    ///
    /// Validation runs while callback delivery is fenced by the watcher state
    /// lock. Any discontinuity leaves the watcher invalidated and returns the
    /// exact baseline requirement.
    #[cfg(target_os = "windows")]
    pub fn accept_windows_usn_baseline(
        &mut self,
        checkpoint: crate::WindowsUsnCheckpoint,
    ) -> Result<Result<crate::WindowsUsnContinuity, crate::WindowsUsnError>, NativeWatchError> {
        let root = self.root.clone();
        self.accept_verified_baseline(|| {
            crate::validate_windows_usn_checkpoint(&root, checkpoint).map(|continuity| {
                (continuity == crate::WindowsUsnContinuity::Unchanged).then_some(continuity)
            })
        })
        .map(|result| {
            result.map(|admitted| {
                admitted.unwrap_or(crate::WindowsUsnContinuity::BaselineRequired(
                    crate::WindowsUsnDiscontinuity::VolumeAdvanced,
                ))
            })
        })
    }

    /// Seals the caller's baseline scan.
    ///
    /// If events overflowed during scanning, the returned batch remains
    /// invalidated and the scan must be repeated. Otherwise all concurrently
    /// queued hints form the first contiguous interval after the baseline.
    ///
    /// # Errors
    ///
    /// Rejects completion without a matching scan or poisoned state.
    pub fn finish_rescan(&mut self) -> Result<WatchBatch, NativeWatchError> {
        if !self.rescan_in_progress {
            return Err(NativeWatchError::NoRescanInProgress);
        }
        let observed = open_native_root(&self.root)
            .and_then(|root| NativeRootIdentity::from_file(&root))
            .ok();
        if observed != Some(self.root_identity) {
            self.shared
                .lock()
                .map_err(|_| NativeWatchError::StatePoisoned)?
                .invalidation = Some(WatchInvalidationReason::RootChanged);
        }
        self.rescan_in_progress = false;
        let mut shared = self
            .shared
            .lock()
            .map_err(|_| NativeWatchError::StatePoisoned)?;
        Ok(shared.seal_invalidation().map_or(
            WatchBatch::Changes {
                epoch: self.epoch,
                first_sequence: self.next_sequence,
                next_sequence: self.next_sequence,
                changes: Vec::new(),
            },
            |reason| WatchBatch::RescanRequired {
                epoch: self.epoch,
                reason,
            },
        ))
    }

    /// Aborts an in-progress baseline and returns the watcher to an explicitly
    /// invalidated, retryable state.
    ///
    /// Events that arrived during the failed scan remain bounded in the queue;
    /// the next [`Self::begin_rescan`] discards them before establishing a new
    /// baseline interval.
    ///
    /// # Errors
    ///
    /// Rejects an abort without a matching scan or poisoned watcher state.
    pub fn abort_rescan(
        &mut self,
        reason: WatchInvalidationReason,
    ) -> Result<(), NativeWatchError> {
        if !self.rescan_in_progress {
            return Err(NativeWatchError::NoRescanInProgress);
        }
        self.rescan_in_progress = false;
        self.shared
            .lock()
            .map_err(|_| NativeWatchError::StatePoisoned)?
            .invalidation = Some(reason);
        Ok(())
    }

    /// Stops reporting hints for events wholly inside `subtrees`.
    ///
    /// Excluded subtrees are invisible to the consumer (for example a VCS
    /// directory or reserved SDK state), so their churn never invalidates it.
    /// Their order is the [`Self::fence`] placement preference.
    ///
    /// # Errors
    ///
    /// Rejects the root, unrepresentable paths, or poisoned synchronization.
    pub fn exclude_subtrees(&mut self, subtrees: &[NamespacePath]) -> Result<(), NativeWatchError> {
        let excluded = subtrees
            .iter()
            .map(|subtree| {
                if subtree.is_root() {
                    return Err(NativeWatchError::UnrepresentablePath);
                }
                crate::native_capture::namespace_to_host_path(subtree)
                    .map(|relative| self.root.join(relative))
                    .map_err(|_| NativeWatchError::UnrepresentablePath)
            })
            .collect::<Result<Vec<_>, _>>()?;
        self.shared
            .lock()
            .map_err(|_| NativeWatchError::StatePoisoned)?
            .excluded
            .clone_from(&excluded);
        self.excluded = subtrees.iter().cloned().zip(excluded).collect();
        Ok(())
    }

    /// Waits until every host change completed before this call is queued.
    ///
    /// Native delivery is asynchronous (`FSEvents` most visibly), so a poll
    /// alone can miss a write that finished before it. The fence creates and
    /// removes a uniquely named cookie inside the first excluded subtree that
    /// is a directory (creating the first one when none is), then waits for
    /// the cookie's ordered notification. Excluded cookies are never hints.
    ///
    /// A fence either proves delivery or invalidates the watcher: when the
    /// cookie cannot be placed, written, or observed within `timeout`, the
    /// next poll demands an authenticated rescan instead.
    ///
    /// # Errors
    ///
    /// Returns only poisoned synchronization.
    pub fn fence(&mut self, timeout: Duration) -> Result<(), NativeWatchError> {
        let delivered = self.fence_cookie().is_ok_and(|cookie| {
            self.await_cookie(&cookie, timeout)
                .is_ok_and(|delivered| delivered)
        });
        let mut state = self
            .shared
            .lock()
            .map_err(|_| NativeWatchError::StatePoisoned)?;
        state.fence = None;
        if !delivered && state.invalidation.is_none() {
            state.invalidation = Some(WatchInvalidationReason::NativeRescanRequired);
        }
        Ok(())
    }

    fn fence_cookie(&mut self) -> Result<PathBuf, NativeWatchError> {
        let (directory, host_directory) = self.fence_directory()?;
        self.watch_directory(&directory)?;
        self.fences = self.fences.wrapping_add(1);
        Ok(host_directory.join(format!(
            ".acyclic-fence-{}-{}",
            std::process::id(),
            self.fences
        )))
    }

    /// Writes `cookie` and reports whether its notification arrived in time.
    fn await_cookie(&self, cookie: &Path, timeout: Duration) -> Result<bool, NativeWatchError> {
        let mut state = self
            .shared
            .lock()
            .map_err(|_| NativeWatchError::StatePoisoned)?;
        if state.invalidation.is_some() {
            return Ok(false);
        }
        state.fence = Some(cookie.to_path_buf());
        drop(state);
        std::fs::File::create_new(cookie)
            .and_then(|_| std::fs::remove_file(cookie))
            .map_err(|error| NativeWatchError::Io(error.to_string()))?;
        let deadline = Instant::now() + timeout;
        let mut state = self
            .shared
            .lock()
            .map_err(|_| NativeWatchError::StatePoisoned)?;
        while state.fence.is_some() && state.invalidation.is_none() {
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                return Ok(false);
            }
            state = self
                .fenced
                .wait_timeout(state, remaining)
                .map_err(|_| NativeWatchError::StatePoisoned)?
                .0;
        }
        Ok(state.invalidation.is_none())
    }

    fn fence_directory(&self) -> Result<(NamespacePath, PathBuf), NativeWatchError> {
        for (subtree, host) in &self.excluded {
            match host.symlink_metadata() {
                Ok(metadata) if metadata.is_dir() => return Ok((subtree.clone(), host.clone())),
                Ok(_) => {}
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(error) => return Err(NativeWatchError::Io(error.to_string())),
            }
        }
        let (subtree, host) = self
            .excluded
            .first()
            .cloned()
            .ok_or(NativeWatchError::InvalidOptions)?;
        std::fs::create_dir(&host).map_err(|error| NativeWatchError::Io(error.to_string()))?;
        Ok((subtree, host))
    }

    /// Moves at most `maximum_changes` contiguous hints from the native queue.
    ///
    /// # Errors
    ///
    /// Fails before output allocation for zero bounds, cancellation, an active
    /// rescan, poisoned state, sequence overflow, allocation failure, or an
    /// exceeded exact work budget.
    pub fn poll(
        &mut self,
        maximum_changes: u32,
        budget: WorkBudget,
        cancellation: &CancellationToken,
    ) -> MeasuredResult<OperationReceipt<WatchBatch>, NativeWatchError> {
        cancellation
            .check()
            .map_err(|error| OperationFailure::before_work(error.into()))?;
        if maximum_changes == 0 {
            return Err(OperationFailure::before_work(
                NativeWatchError::ZeroPollLimit,
            ));
        }
        if self.rescan_in_progress {
            return Err(OperationFailure::before_work(
                NativeWatchError::RescanInProgress,
            ));
        }
        if let Some(reason) = self.invalidation()? {
            return Ok(OperationReceipt {
                value: WatchBatch::RescanRequired {
                    epoch: self.epoch,
                    reason,
                },
                work: WorkCounters::default(),
            });
        }
        let root_work = WorkCounters {
            backend_read_operations: 1,
            ..WorkCounters::default()
        };
        root_work
            .verify(budget)
            .map_err(|error| OperationFailure::before_work(error.into()))?;
        self.verify_current_root()?;
        if let Some(reason) = self.invalidation()? {
            return Ok(OperationReceipt {
                value: WatchBatch::RescanRequired {
                    epoch: self.epoch,
                    reason,
                },
                work: root_work,
            });
        }
        self.poll_contiguous(maximum_changes, budget, root_work)
    }

    fn poll_contiguous(
        &mut self,
        maximum_changes: u32,
        budget: WorkBudget,
        initial_work: WorkCounters,
    ) -> MeasuredResult<OperationReceipt<WatchBatch>, NativeWatchError> {
        let queued = self.queued.load(Ordering::Acquire).min(maximum_changes);
        if queued == 0 {
            return Ok(OperationReceipt {
                value: WatchBatch::Changes {
                    epoch: self.epoch,
                    first_sequence: self.next_sequence,
                    next_sequence: self.next_sequence,
                    changes: Vec::new(),
                },
                work: initial_work,
            });
        }
        let capacity = usize::try_from(queued)
            .map_err(|_| OperationFailure::before_work(NativeWatchError::InvalidOptions))?;
        let allocation_bytes = capacity
            .checked_mul(size_of::<WatchChange>())
            .map(crate::foundation::usize_to_u64)
            .ok_or_else(|| {
                OperationFailure::before_work(NativeWatchError::Work(WorkError::Overflow))
            })?;
        let mut work = initial_work
            .checked_add(WorkCounters {
                allocation_operations: 1,
                peak_allocation_bytes: allocation_bytes,
                ..WorkCounters::default()
            })
            .map_err(|error| OperationFailure::before_work(error.into()))?;
        work.verify(budget)
            .map_err(|error| OperationFailure::before_work(error.into()))?;
        let mut changes = Vec::new();
        changes
            .try_reserve_exact(capacity)
            .map_err(|_| OperationFailure::new(NativeWatchError::AllocationFailed, work))?;
        while changes.len() < capacity {
            match self.receiver.try_recv() {
                Ok(change) => {
                    self.queued.fetch_sub(1, Ordering::AcqRel);
                    changes.push(change);
                }
                Err(TryRecvError::Empty) => break,
                Err(TryRecvError::Disconnected) => {
                    self.invalidate(WatchInvalidationReason::BackendError)?;
                    break;
                }
            }
        }
        if let Some(reason) = self.invalidation()? {
            return Ok(OperationReceipt {
                value: WatchBatch::RescanRequired {
                    epoch: self.epoch,
                    reason,
                },
                work,
            });
        }
        let count = crate::foundation::usize_to_u64(changes.len());
        let next = self
            .next_sequence
            .0
            .checked_add(count)
            .ok_or_else(|| OperationFailure::new(NativeWatchError::SequenceExhausted, work))?;
        work = work
            .checked_add(WorkCounters {
                items_examined: count,
                items_returned: count,
                ..WorkCounters::default()
            })
            .map_err(|error| OperationFailure::new(error.into(), work))?;
        work.verify(budget)
            .map_err(|error| OperationFailure::new(error.into(), work))?;
        let first_sequence = self.next_sequence;
        self.next_sequence = WatchSequence(next);
        self.forget_removed_directories(&changes);
        Ok(OperationReceipt {
            value: WatchBatch::Changes {
                epoch: self.epoch,
                first_sequence,
                next_sequence: self.next_sequence,
                changes,
            },
            work,
        })
    }

    fn verify_current_root(&self) -> Result<(), OperationFailure<NativeWatchError>> {
        let observed = open_native_root(&self.root)
            .and_then(|root| NativeRootIdentity::from_file(&root))
            .ok();
        if observed == Some(self.root_identity) {
            return Ok(());
        }
        self.shared
            .lock()
            .map_err(|_| OperationFailure::before_work(NativeWatchError::StatePoisoned))?
            .invalidation = Some(WatchInvalidationReason::RootChanged);
        Ok(())
    }

    fn invalidation(
        &self,
    ) -> Result<Option<WatchInvalidationReason>, OperationFailure<NativeWatchError>> {
        self.shared
            .lock()
            .map(|mut state| state.seal_invalidation())
            .map_err(|_| OperationFailure::before_work(NativeWatchError::StatePoisoned))
    }

    fn invalidate(
        &self,
        reason: WatchInvalidationReason,
    ) -> Result<(), OperationFailure<NativeWatchError>> {
        let mut shared = self
            .shared
            .lock()
            .map_err(|_| OperationFailure::before_work(NativeWatchError::StatePoisoned))?;
        shared.invalidation.get_or_insert(reason);
        Ok(())
    }
}

impl crate::demand::DemandDirectoryObserver for Mutex<NativeWatch> {
    fn observe_directory(
        &self,
        directory: &NamespacePath,
    ) -> Result<(), crate::demand::DemandError> {
        self.lock()
            .map_err(|_| NativeWatchError::StatePoisoned)?
            .watch_directory(directory)?;
        Ok(())
    }
}

fn accept_native_event(
    event: &notify::Result<Event>,
    context: &NativeEventContext,
    sender: &SyncSender<WatchChange>,
    shared: &Arc<Mutex<SharedState>>,
    queued: &Arc<AtomicU32>,
) {
    let Ok(mut state) = shared.lock() else {
        return;
    };
    if state.invalidation.is_some() {
        return;
    }
    if let Ok(event) = event
        && !event.paths.is_empty()
        && event.paths.iter().all(|path| {
            state
                .excluded
                .iter()
                .any(|excluded| path.starts_with(excluded))
        })
    {
        return;
    }
    let mapped = event
        .as_ref()
        .map_err(|_| WatchInvalidationReason::BackendError)
        .and_then(|event| map_native_event(event, context));
    let changes = match mapped {
        Ok(MappedNativeEvent::Changes(changes)) => changes,
        #[cfg(target_os = "linux")]
        Ok(MappedNativeEvent::RenameHalf {
            tracker,
            mode,
            path,
        }) => {
            match mode {
                RenameMode::From if state.pending_rename.is_none() => {
                    if queued.load(Ordering::Acquire) >= context.maximum_queued_changes {
                        state.invalidation = Some(WatchInvalidationReason::QueueOverflow);
                        return;
                    }
                    state.pending_rename = Some(PendingRename {
                        tracker,
                        from: path,
                        to: None,
                    });
                }
                RenameMode::To => {
                    let Some(pending) = state.pending_rename.as_mut() else {
                        state.invalidation = Some(WatchInvalidationReason::AmbiguousRename);
                        return;
                    };
                    if pending.tracker != tracker || pending.to.is_some() {
                        state.invalidation = Some(WatchInvalidationReason::AmbiguousRename);
                    } else {
                        pending.to = Some(path);
                    }
                }
                _ => state.invalidation = Some(WatchInvalidationReason::AmbiguousRename),
            }
            return;
        }
        #[cfg(target_os = "linux")]
        Ok(MappedNativeEvent::RenameBoth {
            tracker,
            from,
            to,
            changes,
        }) => {
            let matched = state.pending_rename.as_ref().is_some_and(|pending| {
                pending.tracker == tracker
                    && pending.from == from
                    && pending.to.as_ref() == Some(&to)
            });
            if !matched {
                state.invalidation = Some(WatchInvalidationReason::AmbiguousRename);
                return;
            }
            state.pending_rename = None;
            changes
        }
        Err(reason) => {
            state.invalidation = Some(reason);
            return;
        }
    };
    for change in changes {
        #[cfg(target_os = "linux")]
        if queued
            .load(Ordering::Acquire)
            .saturating_add(u32::from(state.pending_rename.is_some()))
            >= context.maximum_queued_changes
        {
            state.invalidation = Some(WatchInvalidationReason::QueueOverflow);
            break;
        }
        match sender.try_send(change) {
            Ok(()) => {
                queued.fetch_add(1, Ordering::Release);
            }
            Err(TrySendError::Full(_) | TrySendError::Disconnected(_)) => {
                state.invalidation = Some(WatchInvalidationReason::QueueOverflow);
                break;
            }
        }
    }
}

enum MappedNativeEvent {
    Changes(Vec<WatchChange>),
    #[cfg(target_os = "linux")]
    RenameHalf {
        tracker: usize,
        mode: RenameMode,
        path: PathBuf,
    },
    #[cfg(target_os = "linux")]
    RenameBoth {
        tracker: usize,
        from: PathBuf,
        to: PathBuf,
        changes: Vec<WatchChange>,
    },
}

fn map_native_event(
    event: &Event,
    context: &NativeEventContext,
) -> Result<MappedNativeEvent, WatchInvalidationReason> {
    validate_replayed_root_creation(event, &context.root, context.root_identity)?;
    #[cfg(target_os = "linux")]
    if let EventKind::Modify(ModifyKind::Name(mode)) = event.kind {
        if event.need_rescan() {
            return Err(WatchInvalidationReason::NativeRescanRequired);
        }
        let tracker = event
            .attrs
            .tracker()
            .ok_or(WatchInvalidationReason::AmbiguousRename)?;
        if matches!(mode, RenameMode::From | RenameMode::To) {
            let [path] = event.paths.as_slice() else {
                return Err(WatchInvalidationReason::AmbiguousRename);
            };
            let relative =
                relative_namespace_path(&context.root, path, context.profile, context.limits)?;
            if relative.is_root() {
                return Err(WatchInvalidationReason::RootChanged);
            }
            return Ok(MappedNativeEvent::RenameHalf {
                tracker,
                mode,
                path: path.clone(),
            });
        }
        if mode == RenameMode::Both {
            let changes = map_event(event, &context.root, context.profile, context.limits)?;
            let [from, to] = event.paths.as_slice() else {
                return Err(WatchInvalidationReason::AmbiguousRename);
            };
            return Ok(MappedNativeEvent::RenameBoth {
                tracker,
                from: from.clone(),
                to: to.clone(),
                changes,
            });
        }
    }
    map_event(event, &context.root, context.profile, context.limits).map(MappedNativeEvent::Changes)
}

fn validate_replayed_root_creation(
    event: &Event,
    root: &Path,
    admitted_root_identity: NativeRootIdentity,
) -> Result<(), WatchInvalidationReason> {
    #[cfg(target_os = "macos")]
    if event.paths.iter().any(|path| path == root) && matches!(event.kind, EventKind::Create(_)) {
        let observed = open_native_root(root)
            .and_then(|file| NativeRootIdentity::from_file(&file))
            .ok();
        if observed != Some(admitted_root_identity) {
            return Err(WatchInvalidationReason::RootChanged);
        }
    }
    #[cfg(not(target_os = "macos"))]
    let _ = (event, root, admitted_root_identity);
    Ok(())
}

fn map_event(
    event: &Event,
    root: &Path,
    profile: FilesystemProfile,
    limits: VolumeLimits,
) -> Result<Vec<WatchChange>, WatchInvalidationReason> {
    if event.need_rescan() {
        return Err(WatchInvalidationReason::NativeRescanRequired);
    }
    if matches!(event.kind, EventKind::Access(_)) {
        return Ok(Vec::new());
    }
    if event.paths.is_empty() || event.paths.len() > 2 {
        return Err(WatchInvalidationReason::UnrepresentablePath);
    }
    let mut paths = Vec::new();
    paths
        .try_reserve_exact(event.paths.len())
        .map_err(|_| WatchInvalidationReason::UnrepresentablePath)?;
    for path in &event.paths {
        // The watched root necessarily predates watcher admission. FSEvents
        // may replay a creation hint for it after the baseline handshake;
        // that hint cannot describe a new namespace entry and has no capture
        // target. Root removal and rename remain fail-closed below.
        #[cfg(target_os = "macos")]
        if path == root && matches!(event.kind, EventKind::Create(_)) {
            continue;
        }
        paths.push(relative_namespace_path(root, path, profile, limits)?);
    }
    if paths.iter().any(NamespacePath::is_root)
        && matches!(
            event.kind,
            EventKind::Remove(_) | EventKind::Modify(ModifyKind::Name(_))
        )
    {
        return Err(WatchInvalidationReason::RootChanged);
    }
    // FSEvents may report a coalesced change against the watched directory
    // itself without identifying the affected descendant. The volume root is
    // not a mutable namespace entry, so require a bounded baseline instead of
    // inventing an invalid root mutation or silently losing the change.
    #[cfg(target_os = "macos")]
    if paths.iter().any(NamespacePath::is_root) {
        return Err(WatchInvalidationReason::NativeRescanRequired);
    }
    if let EventKind::Modify(ModifyKind::Name(mode)) = event.kind {
        return match (mode, paths.as_slice()) {
            (RenameMode::Both, [from, to]) => Ok(vec![WatchChange::Renamed {
                from: from.clone(),
                to: to.clone(),
            }]),
            // inotify pairs both halves by cookie before they reach here
            // (see `map_native_event`), so a lone half on Linux means the
            // other half was lost.
            #[cfg(target_os = "linux")]
            _ => Err(WatchInvalidationReason::AmbiguousRename),
            // FSEvents never pairs renames, and `ReadDirectoryChangesW`
            // delivers each half as its own event, so a lone half is the
            // ordinary case there. It proves the same thing a creation hint
            // does: this path changed, look at it. The capture resolves a
            // modified hint by stat, treating a vanished path as removed
            // (subtree and all), so both ends of a rename land exactly
            // without guessing which was which. Invalidating instead made
            // every atomic save, `mv`, and `git checkout` cost a full rescan.
            #[cfg(not(target_os = "linux"))]
            (_, paths) => Ok(paths.iter().cloned().map(WatchChange::Modified).collect()),
        };
    }
    let mut changes = Vec::new();
    changes
        .try_reserve_exact(paths.len())
        .map_err(|_| WatchInvalidationReason::UnrepresentablePath)?;
    for path in paths {
        changes.push(match event.kind {
            // FSEvents coalesces flags and can replay creation hints for
            // paths that predate the caller's baseline, so a Darwin creation
            // event only proves the path changed. The created hint's
            // identity-replacing capture semantics require the exact
            // notifications inotify and `ReadDirectoryChangesW` deliver.
            #[cfg(target_os = "macos")]
            EventKind::Create(_) => WatchChange::Modified(path),
            #[cfg(not(target_os = "macos"))]
            EventKind::Create(_) => WatchChange::Created(path),
            EventKind::Remove(_) => WatchChange::Removed(path),
            // The same coalescing applies to metadata hints: a Darwin
            // metadata event may stand for a coalesced data write, so it only
            // proves the path changed. Treating it as metadata-only lets a
            // capture skip restaging content that did change.
            #[cfg(target_os = "macos")]
            EventKind::Modify(ModifyKind::Metadata(_)) => WatchChange::Modified(path),
            #[cfg(not(target_os = "macos"))]
            EventKind::Modify(ModifyKind::Metadata(_)) => WatchChange::MetadataChanged(path),
            EventKind::Modify(_) | EventKind::Any | EventKind::Other => WatchChange::Modified(path),
            EventKind::Access(_) => continue,
        });
    }
    Ok(changes)
}

fn relative_namespace_path(
    root: &Path,
    path: &Path,
    profile: FilesystemProfile,
    limits: VolumeLimits,
) -> Result<NamespacePath, WatchInvalidationReason> {
    let relative = path
        .strip_prefix(root)
        .map_err(|_| WatchInvalidationReason::UnrepresentablePath)?;
    let mut components = Vec::new();
    for component in relative.components() {
        let Component::Normal(name) = component else {
            return Err(WatchInvalidationReason::UnrepresentablePath);
        };
        let (encoding, bytes) =
            crate::native_name::host_name_bytes(name, profile, limits.maximum_component_bytes)
                .map_err(|_| WatchInvalidationReason::UnrepresentablePath)?;
        let logical = LogicalName::new(encoding, bytes, limits.maximum_component_bytes)
            .map_err(|_| WatchInvalidationReason::UnrepresentablePath)?;
        components.push(logical);
    }
    NamespacePath::new(components, limits)
        .map_err(|_error: NamespacePathError| WatchInvalidationReason::UnrepresentablePath)
}

/// Native watcher failures that are not recoverable through a baseline rescan.
#[derive(Debug, Error)]
pub enum NativeWatchError {
    /// Queue or volume bounds are invalid.
    #[error("native watcher options are invalid")]
    InvalidOptions,
    /// Watch root is not a directory.
    #[error("native watcher root is not a directory")]
    RootIsNotDirectory,
    /// A demanded directory traverses a symbolic link and cannot be mapped
    /// back to one exact namespace path.
    #[error("native watcher directory traverses a symbolic link")]
    SymbolicLinkDirectory,
    /// A demanded observer path is not a directory.
    #[error("native watcher observed path is not a directory")]
    ObservedPathNotDirectory,
    /// A demanded namespace path cannot be represented exactly on this host.
    #[error("native watcher directory is not representable on this host")]
    UnrepresentablePath,
    /// A separately opened capture root does not match the watched directory.
    #[error("native watcher root changed identity")]
    RootChanged,
    /// Root path I/O failed.
    #[error("native watcher I/O failed: {0}")]
    Io(String),
    /// Native backend setup failed.
    #[error("native watcher backend failed: {0}")]
    Backend(String),
    /// Watcher state synchronization was poisoned.
    #[error("native watcher state is poisoned")]
    StatePoisoned,
    /// A baseline scan is already active.
    #[error("native watcher baseline scan is already active")]
    RescanAlreadyInProgress,
    /// No baseline scan is active.
    #[error("native watcher baseline scan is not active")]
    NoRescanInProgress,
    /// Polling while a baseline is being built is ambiguous.
    #[error("native watcher baseline scan is in progress")]
    RescanInProgress,
    /// Poll output must have a positive bound.
    #[error("native watcher poll limit must be non-zero")]
    ZeroPollLimit,
    /// Process-local cursor arithmetic overflowed.
    #[error("native watcher sequence is exhausted")]
    SequenceExhausted,
    /// Bounded output allocation failed.
    #[error("native watcher output allocation failed")]
    AllocationFailed,
    /// Polling was cooperatively cancelled.
    #[error(transparent)]
    Cancelled(#[from] CancellationError),
    /// Exact work overflowed or exceeded the admitted budget.
    #[error(transparent)]
    Work(#[from] WorkError),
}

#[cfg(test)]
#[path = "tests/watch.rs"]
mod tests;