test: 缩小规则夹具并用阶段信号替代固定等待

ShadowsocksX-NG2/App/SystemProxyObserver.swift

@@ -30,13 +30,16 @@ final class SystemProxyObserver: ObservableObject {
   private(set) var systemProxyInitialApplyPending = false
   /// 读面：健康循环的收敛条件。
   private(set) var systemProxyWasUnavailable = false
-  private var systemProxyHealthTask: Task<Void, Never>?
+  /// Read-only task handle lets observers await cancellation completion.
+  private(set) var systemProxyHealthTask: Task<Void, Never>?
   private var systemProxyReadGeneration = 0
   /// 读面：观察机测试。
   private(set) var systemProxyObservationMode = SystemProxyObservationMode.stopped
   private var systemProxyCleanupRescanRequested = false
   private var systemProxyCleanupTask: Task<SystemProxyApplicationFacts, Never>?
   private var systemProxyInspectionScheduled = false
+  /// Last passive inspection; awaiting it observes event convergence without a timing guess.
+  private(set) var systemProxyInspectionTask: Task<Void, Never>?
 
   let systemProxy: SystemProxyControlling
   let systemProxyHelper: SystemProxyHelperServicing
@@ -568,7 +571,7 @@ extension SystemProxyObserver {
       guard isIntentEnabled() else { return }
       guard !systemProxyInspectionScheduled else { return }
       systemProxyInspectionScheduled = true
-      Task { @MainActor [weak self] in
+      systemProxyInspectionTask = Task { @MainActor [weak self] in
         await Task.yield()
         guard let self else { return }
         self.systemProxyInspectionScheduled = false

ShadowsocksX-NG2/Tests/AgentLifecycleTests+ChangeProtocol.swift

@@ -112,9 +112,8 @@ final class AgentLifecycleStaleListenerTests: AgentLifecycleTests {
     let wrapper = try launchWrapper(behavior: "run")
 
     XCTAssertTrue(
-      try waitUntil(timeout: 8) {
-        self.agentLog().contains("listen not established within deadline")
-      }, "初次监听等待结束后，wrapper 应继续监管未就绪的 ACL 子进程")
+      try waitUntil { self.stateLog().contains("invoked:") },
+      "等待候选子进程启动后验证监听归属")
 
     let server = try XCTUnwrap(ProxyRuntimeFixture.makeDocument().servers.first)
     let serverReload = SslocalRuntimeDocument(
@@ -125,6 +124,7 @@ final class AgentLifecycleStaleListenerTests: AgentLifecycleTests {
     XCTAssertTrue(
       try waitUntil { self.agentLog().contains("reload deferred until listeners are ready") },
       "未就绪的直连实例应推迟 server-only reload")
+    XCTAssertFalse(stateLog().contains("SIGUSR1"), "旧监听不能让候选通过就绪门并转发重载")
     XCTAssertTrue(
       pidAlive(wrapper.processIdentifier),
       "候选 wrapper 应继续监管；agent log: " + agentLog() + "; child log: " + stateLog())

ShadowsocksX-NG2/Tests/ProxyControlWorkflowIntegrationTests+SystemProxy.swift

@@ -11,8 +11,8 @@ extension ProxyControlWorkflowIntegrationTests {
     _ = await composition.control.setSystemProxyEnabled(true)
     let writes = systemProxy.applied.count
     XCTAssertEqual(writes, 1)
-    networkMonitor.emit([.networkConfiguration, .proxyConfiguration, .networkPath])
-    try await Task.sleep(nanoseconds: 200_000_000)
+    await emitAndWaitForInspection(
+      [.networkConfiguration, .proxyConfiguration, .networkPath], in: composition)
     XCTAssertEqual(systemProxy.applied.count, writes)
     XCTAssertEqual(systemProxy.clearCount, 0)
   }

ShadowsocksX-NG2/Tests/ProxyControlWorkflowIntegrationTests+SystemProxyInspection.swift

@@ -70,8 +70,7 @@ extension ProxyControlWorkflowIntegrationTests {
     _ = await composition.control.repairSystemProxy()
     XCTAssertEqual(systemProxy.repairedServices, [.init(locationID: "work", serviceID: "vpn")])
     XCTAssertEqual(composition.control.snapshot.systemProxyApplication, .applied)
-    networkMonitor.emit(.networkPath)
-    try await Task.sleep(nanoseconds: 60_000_000)
+    await emitAndWaitForInspection(.networkPath, in: composition)
     XCTAssertEqual(systemProxy.applied.count, 2, "被动事件不重写")
   }
 
@@ -94,13 +93,17 @@ extension ProxyControlWorkflowIntegrationTests {
     systemProxy.services[0] = SystemProxyServiceState(
       identifier: original.identifier, configuration: nil, name: original.name)
     _ = await composition.control.recheckSystemProxy()
-    systemProxy.beforeRepair = { try? await Task.sleep(nanoseconds: 100_000_000) }
+    let gate = ProxyControlOperationGate(started: expectation(description: "repair paused"))
+    defer { gate.resume() }
+    systemProxy.beforeRepair = { await gate.wait() }
     let first = Task { await composition.control.repairSystemProxy() }
-    try await waitForSnapshot(composition) { $0.systemProxyApplication == .repairing }
+    await fulfillment(of: [gate.started], timeout: 2)
+    XCTAssertEqual(composition.control.snapshot.systemProxyApplication, .repairing)
     XCTAssertEqual(
       composition.control.snapshot.systemProxyInspection.differences.map(\.name), ["Wi-Fi"])
     XCTAssertFalse(composition.control.snapshot.systemProxyInspection.canRepair)
     _ = await composition.control.repairSystemProxy()
+    gate.resume()
     _ = await first.value
     XCTAssertEqual(systemProxy.applied.count, 2)
   }
@@ -125,8 +128,7 @@ extension ProxyControlWorkflowIntegrationTests {
     try await waitForSnapshot(composition) { $0.systemProxyApplication == .paused }
     XCTAssertTrue(composition.control.snapshot.systemProxyIntentEnabled)
     XCTAssertEqual(systemProxy.clearCount, 1)
-    networkMonitor.emit(.networkPath)
-    try await Task.sleep(nanoseconds: 60_000_000)
+    await emitAndWaitForInspection(.networkPath, in: composition)
     XCTAssertEqual(systemProxy.clearCount, 1, "失效期间不循环重试")
     probe.setOutcomes([.reachable])
     try await waitForSnapshot(composition) { $0.systemProxyApplication == .applied }
@@ -138,9 +140,12 @@ extension ProxyControlWorkflowIntegrationTests {
     let composition = try await enabledSystemProxyComposition(probe: probe)
     probe.setOutcomes([.refused(detail: "stopped")])
     try await waitForSnapshot(composition) { $0.systemProxyApplication == .paused }
+    let healthObservation = composition.controller.systemProxyObserver.systemProxyHealthTask
+    XCTAssertNotNil(healthObservation)
     _ = await composition.control.setSystemProxyEnabled(false)
     probe.setOutcomes([.reachable])
-    try await Task.sleep(nanoseconds: 80_000_000)
+    await healthObservation?.value
+    XCTAssertFalse(networkMonitor.isObserving, "关闭意图后应停止观察与自动恢复")
     XCTAssertEqual(systemProxy.applied.count, 1)
     XCTAssertFalse(composition.control.snapshot.systemProxyIntentEnabled)
   }
@@ -251,8 +256,7 @@ extension ProxyControlWorkflowIntegrationTests {
     XCTAssertTrue(composition.control.snapshot.systemProxyIntentEnabled)
     XCTAssertTrue(composition.control.snapshot.systemProxyInspection.canRetryClear)
     XCTAssertFalse(composition.control.snapshot.systemProxyInspection.canRepair)
-    networkMonitor.emit(.networkConfiguration)
-    try await Task.sleep(nanoseconds: 80_000_000)
+    await emitAndWaitForInspection(.networkConfiguration, in: composition)
     XCTAssertEqual(systemProxy.clearCount, 1)
     systemProxy.clearError = nil
     _ = await composition.control.retrySystemProxyClear()
@@ -307,11 +311,14 @@ extension ProxyControlWorkflowIntegrationTests {
   func testPassiveReadCannotOverwriteHealthSuspension() async throws {
     let probe = ProxyRuntimeFixture.FakeProbe.reachable()
     let composition = try await enabledSystemProxyComposition(probe: probe)
-    systemProxy.beforeRead = { try? await Task.sleep(nanoseconds: 120_000_000) }
+    let gate = ProxyControlOperationGate(started: expectation(description: "read paused"))
+    defer { gate.resume() }
+    systemProxy.beforeRead = { await gate.wait() }
     let read = Task { await composition.control.recheckSystemProxy() }
-    try await Task.sleep(nanoseconds: 10_000_000)
+    await fulfillment(of: [gate.started], timeout: 2)
     probe.setOutcomes([.refused(detail: "stopped")])
     try await waitForSnapshot(composition) { $0.systemProxyApplication == .paused }
+    gate.resume()
     _ = await read.value
     XCTAssertEqual(composition.control.snapshot.systemProxyApplication, .paused)
     XCTAssertEqual(systemProxy.clearCount, 1)
@@ -321,11 +328,14 @@ extension ProxyControlWorkflowIntegrationTests {
     let probe = ProxyRuntimeFixture.FakeProbe.reachable()
     let composition = try await enabledSystemProxyComposition(probe: probe)
     systemProxy.clearError = SystemProxyError.commitFailed("busy")
-    systemProxy.beforeRead = { try? await Task.sleep(nanoseconds: 120_000_000) }
+    let gate = ProxyControlOperationGate(started: expectation(description: "read paused"))
+    defer { gate.resume() }
+    systemProxy.beforeRead = { await gate.wait() }
     let read = Task { await composition.control.recheckSystemProxy() }
-    try await Task.sleep(nanoseconds: 10_000_000)
+    await fulfillment(of: [gate.started], timeout: 2)
     probe.setOutcomes([.refused(detail: "stopped")])
     try await waitForSnapshot(composition) { $0.systemProxyApplication.isClearFailure }
+    gate.resume()
     _ = await read.value
     XCTAssertEqual(
       composition.control.snapshot.systemProxyApplication, .clearFailed(.operation(.commitFailed)))
@@ -338,10 +348,15 @@ extension ProxyControlWorkflowIntegrationTests {
     systemProxy.services[0] = SystemProxyServiceState(
       identifier: original.identifier, configuration: nil, name: original.name)
     _ = await composition.control.recheckSystemProxy()
-    systemProxy.beforeRead = { try? await Task.sleep(nanoseconds: 120_000_000) }
+    let gate = ProxyControlOperationGate(started: expectation(description: "read paused"))
+    defer { gate.resume() }
+    systemProxy.beforeRead = { await gate.wait() }
     let repair = Task { await composition.control.repairSystemProxy() }
-    try await waitForSnapshot(composition) { $0.systemProxyInspection.isBusy }
+    await fulfillment(of: [gate.started], timeout: 2)
+    XCTAssertTrue(composition.control.snapshot.systemProxyInspection.isBusy)
     let off = Task { await composition.control.setSystemProxyEnabled(false) }
+    try await waitForSnapshot(composition) { !$0.systemProxyIntentEnabled }
+    gate.resume()
     _ = await repair.value
     _ = await off.value
     XCTAssertEqual(systemProxy.applied.count, 1)
@@ -356,10 +371,15 @@ extension ProxyControlWorkflowIntegrationTests {
     systemProxy.services[0] = SystemProxyServiceState(
       identifier: original.identifier, configuration: nil, name: original.name)
     _ = await composition.control.recheckSystemProxy()
-    systemProxy.beforeRead = { try? await Task.sleep(nanoseconds: 120_000_000) }
+    let gate = ProxyControlOperationGate(started: expectation(description: "read paused"))
+    defer { gate.resume() }
+    systemProxy.beforeRead = { await gate.wait() }
     let repair = Task { await composition.control.repairSystemProxy() }
-    try await waitForSnapshot(composition) { $0.systemProxyInspection.isBusy }
+    await fulfillment(of: [gate.started], timeout: 2)
+    XCTAssertTrue(composition.control.snapshot.systemProxyInspection.isBusy)
     probe.setOutcomes([.refused(detail: "stopped")])
+    try await waitForSnapshot(composition) { $0.runtime.status != .running }
+    gate.resume()
     _ = await repair.value
     try await waitForSnapshot(composition) { $0.systemProxyApplication == .paused }
     XCTAssertEqual(systemProxy.applied.count, 1)

ShadowsocksX-NG2/Tests/ProxyControlWorkflowIntegrationTests.swift

@@ -78,8 +78,7 @@ final class ProxyControlWorkflowIntegrationTests: XCTestCase {
       settingsRestore: RestoredProxySettings(
         settings: settings ?? ProxySettings(listen: ActivationFixture.listen),
         unreadableError: nil),
-      agent: agent,
-      probe: probe,
+      agent: agent, probe: probe,
       systemProxy: systemProxy,
       systemProxyHelper: systemProxyHelper,
       systemProxyNetworkChangeMonitor: networkMonitor,
@@ -89,6 +88,7 @@ final class ProxyControlWorkflowIntegrationTests: XCTestCase {
       firewallPollIntervalNanoseconds: 1_000_000,
       // 生产默认 15 秒健康窗；FakeProbe 结果确定，短窗口走同一超时呈现路径。
       launchHealthTimeoutSeconds: 0.05,
+      launchHealthRetryDelay: { try await Task.sleep(for: .milliseconds(1)) },
       systemProxyHealthPollIntervalNanoseconds: 20_000_000,
       helperRefreshDelayNanoseconds: 0,
       sendSignal: { _, _ in 0 },
@@ -306,3 +306,39 @@ final class ProxyControlWorkflowIntegrationTests: XCTestCase {
   }
 
 }
+
+/// Pause the first operation at its I/O seam; subsequent reads can finish normally.
+@MainActor
+final class ProxyControlOperationGate {
+  let started: XCTestExpectation
+  private var didPause = false
+  private var continuation: CheckedContinuation<Void, Never>?
+
+  init(started: XCTestExpectation) { self.started = started }
+
+  func wait() async {
+    guard !didPause else { return }
+    didPause = true
+    await withCheckedContinuation {
+      continuation = $0
+      started.fulfill()
+    }
+  }
+
+  func resume() {
+    continuation?.resume()
+    continuation = nil
+  }
+}
+
+extension ProxyControlWorkflowIntegrationTests {
+  func emitAndWaitForInspection(
+    _ change: SystemProxyNetworkChange, in composition: Composition
+  ) async {
+    XCTAssertTrue(networkMonitor.isObserving)
+    networkMonitor.emit(change)
+    let inspection = composition.controller.systemProxyObserver.systemProxyInspectionTask
+    XCTAssertNotNil(inspection, "被动事件应调度检查")
+    await inspection?.value
+  }
+}

ShadowsocksX-NG2/Tests/RuleEnablementWorkflowTests.swift

@@ -50,13 +50,12 @@ final class RuleEnablementWorkflowTests: XCTestCase {
       workflow.snapshot.rows.first { $0.identity == narrow.identity }!.relationships.isEmpty)
   }
 
-  func testPackagedCollectionSaveKeepsRowOrderAndFixedFactsWithoutWaitingForAnalysis() async throws
-  {
+  func testCollectionSaveKeepsRowOrderAndFixedFactsWithoutWaitingForAnalysis() async throws {
     let directory = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString)
     defer { try? FileManager.default.removeItem(at: directory) }
     let store = CustomRuleStore(fileURL: directory.appendingPathComponent("rules.json"))
     let session = RuleDocumentSession(store: store)
-    let started = expectation(description: "packaged analysis paused")
+    let started = expectation(description: "analysis paused")
     let gate = RulesAnalysisGate(started: started)
     let workflow = RulesWorkflow(
       loadDocument: { try session.load() },
@@ -70,17 +69,25 @@ final class RuleEnablementWorkflowTests: XCTestCase {
         }
       },
       analysisDelay: { await gate.wait() },
-      builtinSnapshots: BuiltinRuleSnapshots(bundle: AppArtifact.bundle))
+      loadBuiltin: { source in
+        if source == .geolocationCN {
+          return rulesFixture(
+            source,
+            rules: [
+              ProxyRule(action: .direct, match: .domainSuffix("cn")),
+              ProxyRule(action: .direct, match: .domainExact("fixture.example")),
+            ])
+        }
+        return rulesFixture(source)
+      })
     await workflow.refresh()
     XCTAssertTrue(workflow.snapshot.isComplete)
-    XCTAssertGreaterThan(workflow.snapshot.rows.count, 10_000)
+    XCTAssertTrue(workflow.snapshot.sources.contains { $0.id == .geolocationCN })
     let rowIDs = workflow.snapshot.rows.map(\.id)
     let fixed = workflow.snapshot.rows.filter(\.isFixed)
+    XCTAssertFalse(fixed.isEmpty)
     let nationalSuffix = RuleIdentity(action: .direct, match: .domainSuffix("cn"))
-    let start = ContinuousClock.now
     await workflow.setEnabled(false, identities: [nationalSuffix])
-    let elapsed = start.duration(to: .now)
-    print("Saved rule publication for \(rowIDs.count) packaged rows: \(elapsed)")
     XCTAssertTrue(try store.loadDocument().disabledIdentities.contains(nationalSuffix))
     XCTAssertEqual(workflow.snapshot.rows.map(\.id), rowIDs)
     XCTAssertEqual(workflow.snapshot.rows.filter(\.isFixed), fixed)

ShadowsocksX-NG2/Tests/RulesWorkflowTests.swift

@@ -172,8 +172,6 @@ final class RulesWorkflowTests: XCTestCase {
     XCTAssertNotNil(geo.conversionReport)
     let rows = workflow.snapshot.rows
     XCTAssertEqual(Set(rows.map(\.id)).count, rows.count)
-    await workflow.refresh()
-    XCTAssertEqual(workflow.snapshot.rows.map(\.id), rows.map(\.id))
     workflow.query(RulesQuery(source: .geolocationCN))
     XCTAssertTrue(workflow.snapshot.rows.contains { $0.content == "cn" })
     XCTAssertEqual(workflow.snapshot.rows.count, 4241)