feat: 实现 FSEvents 到校准扫描的应用层流水线

Packages/SpaceTraceKit/Package.swift

@@ -51,6 +51,7 @@ let package = Package(
             name: "SpaceTracePersistenceTests",
             dependencies: [
                 "SpaceTraceApplication",
+                "SpaceTraceFileSystem",
                 "SpaceTracePersistence",
             ]
         ),

Packages/SpaceTraceKit/README.md

@@ -7,11 +7,11 @@
 | Module | Responsibility | Platform coupling |
 | --- | --- | --- |
 | `SpaceTraceDomain` | Validated storage quantities, observation identities, coverage, and deltas | None |
-| `SpaceTraceApplication` | Application-owned ports and crash-consistency contracts | None |
-| `SpaceTraceFileSystem` | FSEvents callback bridge and conservative flag interpretation | CoreServices |
-| `SpaceTracePersistence` | Actor-isolated raw SQLite prototype for the event journal | SQLite3 |
+| `SpaceTraceApplication` | Dirty-region planning, calibration orchestration, ports, and crash-consistency contracts | Foundation |
+| `SpaceTraceFileSystem` | FSEvents callback bridge, conservative flag interpretation, and application-semantic mapping | CoreServices |
+| `SpaceTracePersistence` | Actor-isolated raw SQLite prototype for cursor, dirty work, and conditional finalization | SQLite3 |
 
-The filesystem and persistence modules are architecture-spike implementations for proposed ADR-003 and ADR-004. They are not yet accepted production decisions and are not connected to a user-visible workflow.
+The filesystem, application pipeline, and persistence modules are architecture-spike implementations for proposed ADR-003 and ADR-004. The pipeline is validated with an injected scanner; a production metadata enumerator and user-visible workflow are not connected yet.
 
 ## Verify
 

Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/DirtyRegionPlanner.swift

@@ -0,0 +1,141 @@
+import Foundation
+
+public struct DirtyRegionPlanner: Sendable {
+    public init() {}
+
+    /// Converts invalidation hints into the narrowest safe directory work.
+    /// Replay duplicates are ignored unless they carry a calibration signal.
+    public func plan(
+        watchRoot: DirtyRegionPath,
+        invalidations: [FileSystemInvalidation],
+        after storedCheckpoint: EventJournalCursor?
+    ) throws -> DirtyRegionPlan {
+        var journaled: [DirtyRegion] = []
+        var outOfBand: [DirtyRegion] = []
+
+        for invalidation in invalidations {
+            let isCalibrationSignal = invalidation.reasons.contains(.requiresCalibration)
+            let requiresScopeCalibration = invalidation.reasons.isDisjoint(
+                with: [.droppedEvents, .rootChanged, .mountChanged]
+            ) == false
+            let target = targetDirectory(
+                for: invalidation,
+                watchRoot: watchRoot,
+                forceRoot: requiresScopeCalibration
+            )
+            var reasons = invalidation.reasons
+            if target.fellBackToRoot {
+                reasons.insert(.requiresCalibration)
+            }
+
+            if let cursor = invalidation.cursor,
+               storedCheckpoint.map({ cursor > $0 }) ?? true {
+                journaled.append(
+                    try DirtyRegion(path: target.path, reasons: reasons, maximumCursor: cursor)
+                )
+            } else if isCalibrationSignal || invalidation.cursor == nil {
+                outOfBand.append(
+                    try DirtyRegion(path: target.path, reasons: reasons, maximumCursor: nil)
+                )
+            }
+        }
+
+        let coalescedJournaled = try coalesce(journaled)
+        let coalescedOutOfBand = try coalesce(outOfBand)
+        return DirtyRegionPlan(
+            checkpoint: coalescedJournaled.compactMap(\.maximumCursor).max(),
+            journaledRegions: coalescedJournaled,
+            outOfBandRegions: coalescedOutOfBand
+        )
+    }
+
+    private func targetDirectory(
+        for invalidation: FileSystemInvalidation,
+        watchRoot: DirtyRegionPath,
+        forceRoot: Bool
+    ) -> (path: DirtyRegionPath, fellBackToRoot: Bool) {
+        guard forceRoot == false,
+              let rawPath = invalidation.path,
+              let normalizedPath = normalize(rawPath),
+              contains(normalizedPath, in: watchRoot.rawValue) else {
+            return (watchRoot, true)
+        }
+
+        let directory: String
+        switch invalidation.itemKind {
+        case .directory:
+            directory = normalizedPath
+        case .file, .symbolicLink:
+            directory = deletingLastComponent(of: normalizedPath)
+        case .unknown:
+            return (watchRoot, true)
+        }
+
+        guard contains(directory, in: watchRoot.rawValue),
+              let path = try? DirtyRegionPath(directory) else {
+            return (watchRoot, true)
+        }
+        return (path, false)
+    }
+
+    private func normalize(_ rawPath: String) -> String? {
+        guard rawPath.first == "/", rawPath.utf8.contains(0) == false else {
+            return nil
+        }
+        let normalized = URL(fileURLWithPath: rawPath).standardizedFileURL.path
+        return normalized.isEmpty ? nil : normalized
+    }
+
+    private func deletingLastComponent(of path: String) -> String {
+        let result = (path as NSString).deletingLastPathComponent
+        return result.isEmpty ? "/" : result
+    }
+
+    private func contains(_ candidate: String, in root: String) -> Bool {
+        root == "/" || candidate == root || candidate.hasPrefix(root + "/")
+    }
+
+    private func coalesce(_ regions: [DirtyRegion]) throws -> [DirtyRegion] {
+        var result: [DirtyRegion] = []
+        for region in regions.sorted(by: regionOrder) {
+            if let ancestorIndex = result.firstIndex(where: {
+                contains(region.path.rawValue, in: $0.path.rawValue)
+            }) {
+                let ancestor = result[ancestorIndex]
+                result[ancestorIndex] = try merged(ancestor, region, at: ancestor.path)
+                continue
+            }
+
+            let descendants = result.filter {
+                contains($0.path.rawValue, in: region.path.rawValue)
+            }
+            var mergedRegion = region
+            for descendant in descendants {
+                mergedRegion = try merged(mergedRegion, descendant, at: region.path)
+            }
+            result.removeAll { descendants.contains($0) }
+            result.append(mergedRegion)
+        }
+        return result.sorted { $0.path.rawValue < $1.path.rawValue }
+    }
+
+    private func regionOrder(_ lhs: DirtyRegion, _ rhs: DirtyRegion) -> Bool {
+        let lhsDepth = lhs.path.rawValue.split(separator: "/").count
+        let rhsDepth = rhs.path.rawValue.split(separator: "/").count
+        return lhsDepth == rhsDepth
+            ? lhs.path.rawValue < rhs.path.rawValue
+            : lhsDepth < rhsDepth
+    }
+
+    private func merged(
+        _ lhs: DirtyRegion,
+        _ rhs: DirtyRegion,
+        at path: DirtyRegionPath
+    ) throws -> DirtyRegion {
+        try DirtyRegion(
+            path: path,
+            reasons: lhs.reasons.union(rhs.reasons),
+            maximumCursor: [lhs.maximumCursor, rhs.maximumCursor].compactMap { $0 }.max()
+        )
+    }
+}

Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/FileSystemCalibrationPipeline.swift

@@ -0,0 +1,90 @@
+public struct CalibrationRequest: Sendable, Equatable {
+    public let streamID: EventStreamID
+    public let workItem: DirtyRegionWorkItem
+
+    public init(streamID: EventStreamID, workItem: DirtyRegionWorkItem) {
+        self.streamID = streamID
+        self.workItem = workItem
+    }
+}
+
+public enum CalibrationCoverage: Sendable, Equatable {
+    /// Every descendant in the requested region was enumerated successfully.
+    case complete
+    /// Cancellation, permission denial, or an enumeration failure left gaps.
+    case partial
+}
+
+public protocol CalibrationScanner: Sendable {
+    func scan(_ request: CalibrationRequest) async throws -> CalibrationCoverage
+}
+
+public actor FileSystemCalibrationPipeline {
+    private let streamID: EventStreamID
+    private let watchRoot: DirtyRegionPath
+    private let repository: any EventJournalRepository
+    private let scanner: any CalibrationScanner
+    private let planner: DirtyRegionPlanner
+
+    public init(
+        streamID: EventStreamID,
+        watchRoot: DirtyRegionPath,
+        repository: any EventJournalRepository,
+        scanner: any CalibrationScanner,
+        planner: DirtyRegionPlanner = DirtyRegionPlanner()
+    ) {
+        self.streamID = streamID
+        self.watchRoot = watchRoot
+        self.repository = repository
+        self.scanner = scanner
+        self.planner = planner
+    }
+
+    public func ingest(_ invalidations: [FileSystemInvalidation]) async throws {
+        guard invalidations.isEmpty == false else { return }
+
+        let checkpoint = try await repository.checkpoint(for: streamID)
+        let plan = try planner.plan(
+            watchRoot: watchRoot,
+            invalidations: invalidations,
+            after: checkpoint
+        )
+
+        if plan.outOfBandRegions.isEmpty == false {
+            try await repository.markDirty(
+                streamID: streamID,
+                regions: plan.outOfBandRegions
+            )
+        }
+        if let checkpoint = plan.checkpoint, plan.journaledRegions.isEmpty == false {
+            try await repository.commit(
+                EventJournalBatch(
+                    streamID: streamID,
+                    checkpoint: checkpoint,
+                    dirtyRegions: plan.journaledRegions
+                )
+            )
+        }
+    }
+
+    /// Processes a bounded snapshot. New events can safely arrive during a
+    /// scan because `resolve` compares the revision captured before scanning.
+    @discardableResult
+    public func calibratePending(limit: Int) async throws -> Int {
+        guard limit > 0 else { return 0 }
+        let workItems = try await repository.pendingDirtyWork(for: streamID, limit: limit)
+        var resolvedCount = 0
+
+        for workItem in workItems {
+            try Task.checkCancellation()
+            let coverage = try await scanner.scan(
+                CalibrationRequest(streamID: streamID, workItem: workItem)
+            )
+            guard coverage == .complete else { continue }
+            if try await repository.resolve(workItem, for: streamID) {
+                resolvedCount += 1
+            }
+        }
+        return resolvedCount
+    }
+}

Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/FileSystemInvalidation.swift

@@ -0,0 +1,50 @@
+/// Application-level item classification. It intentionally contains no
+/// FSEvents or Foundation types so other observation adapters can reuse it.
+public enum FileSystemItemKind: Sendable, Equatable, Hashable, Codable {
+    case file
+    case directory
+    case symbolicLink
+    case unknown
+}
+
+/// One lossy filesystem invalidation hint after adapter-specific flags have
+/// been translated into application semantics.
+public struct FileSystemInvalidation: Sendable, Equatable, Hashable, Codable {
+    public let path: String?
+    public let cursor: EventJournalCursor?
+    public let reasons: DirtyRegionReason
+    public let itemKind: FileSystemItemKind
+
+    public init(
+        path: String?,
+        cursor: EventJournalCursor?,
+        reasons: DirtyRegionReason,
+        itemKind: FileSystemItemKind = .unknown
+    ) throws(EventJournalModelError) {
+        guard reasons.isEmpty == false else {
+            throw .emptyDirtyRegionReasons
+        }
+
+        self.path = path
+        self.cursor = cursor
+        self.reasons = reasons
+        self.itemKind = itemKind
+    }
+}
+
+/// Durable work split by whether it can safely advance the journal cursor.
+public struct DirtyRegionPlan: Sendable, Equatable {
+    public let checkpoint: EventJournalCursor?
+    public let journaledRegions: [DirtyRegion]
+    public let outOfBandRegions: [DirtyRegion]
+
+    public init(
+        checkpoint: EventJournalCursor?,
+        journaledRegions: [DirtyRegion],
+        outOfBandRegions: [DirtyRegion]
+    ) {
+        self.checkpoint = checkpoint
+        self.journaledRegions = journaledRegions
+        self.outOfBandRegions = outOfBandRegions
+    }
+}