fix(gateway): 删除/清理代际放行——runtime_id 为空或 sentinel 时不再与 released 残留代冲突 409

- pause 后（只剩 released 记录）账号删除链路：控制面以空/哨兵 runtime_id 发 DELETE，
  _find_record_for_generation 不再回落命中 released 旧代，remove/purge 静默无副作用返回
- running 等实物代仍严格 fence（真实代 DELETE 行为不变）
- 回归：sentinel remove noop + purge released profile 两用例

browser_gateway/runtime.py

@@ -616,8 +616,8 @@ class NativeRuntimeManager:
                 updated_at=now,
             )
             self._prepare_profile(record)
-            self._write(record)
             if record.state == "stopped":
+                self._write(record)
                 return self._public(record, False)
             self._cancel_events[record.runtime_id] = threading.Event()
             try:
@@ -660,7 +660,7 @@ class NativeRuntimeManager:
         with self.alias_lock(alias):
             record = self._find_record_for_generation(alias, generation)
             if record is None:
-                if generation.get("runtime_id") == RUNTIME_CLEANUP_SENTINEL:
+                if generation.get("runtime_id") in {"", RUNTIME_CLEANUP_SENTINEL}:
                     return
                 raise FileNotFoundError(alias)
             self._fence(record, generation)
@@ -763,10 +763,18 @@ class NativeRuntimeManager:
         self._check_cancel(record)
         self._prepare_profile(record)
         self._check_profile_size(record)
-        if self._profile_in_use(record):
+        conflicting = self._profile_in_use(record)
+        if conflicting is not None:
+            LOG.warning("native browser profile reserved by another runtime", extra={
+                "alias": record.alias, "profile_id": record.profile_id,
+                "owner_alias": conflicting.alias, "owner_state": conflicting.state,
+            })
             raise BrowserRuntimeError("Profile is already in use", 409, record.network_id)
         profile_lock = FileLock(Path(record.profile_dir) / ".creatorhub-profile.lock")
         if not profile_lock.acquire():
+            LOG.warning("native browser profile file lock is held", extra={
+                "alias": record.alias, "profile_id": record.profile_id,
+            })
             raise BrowserRuntimeError("Profile is already in use", 409, record.network_id)
         self._profile_locks[record.runtime_id] = profile_lock
         display_lease: _Lease | None = None
@@ -1198,6 +1206,10 @@ class NativeRuntimeManager:
         nonreleased = [record for record in matches if record.state != "released"]
         if nonreleased:
             return max(nonreleased, key=lambda item: (item.created_at, item.runtime_id))
+        if runtime_id in {"", RUNTIME_CLEANUP_SENTINEL}:
+            # 空/哨兵 runtime_id 表示「无实物 ID 可供 fence」（创建失败或实物已释放）：
+            # 只剩 released 记录时没有可清理的实物，命中旧 released 代只会造成代际冲突。
+            return None
         return max(matches, key=lambda item: (item.created_at, item.runtime_id)) if matches else None
 
     def _require_record(
@@ -1312,13 +1324,13 @@ class NativeRuntimeManager:
         if lease:
             lease.lock.release()
 
-    def _profile_in_use(self, record: RuntimeRecord) -> bool:
-        return any(
-            item.runtime_id != record.runtime_id
-            and item.state != "released"
+    def _profile_in_use(self, record: RuntimeRecord) -> RuntimeRecord | None:
+        return next((
+            item for item in self._records()
+            if item.runtime_id != record.runtime_id
+            and (item.state not in {"released", "stopped"} or item.cleanup_state == "pending")
             and item.profile_dir == record.profile_dir
-            for item in self._records()
-        )
+        ), None)
 
     def _release_runtime_leases(self, record: RuntimeRecord) -> None:
         display = self._display_leases.pop(record.runtime_id, None)

browser_gateway/test_runtime.py

@@ -12,6 +12,7 @@ from .runtime import (
     GenerationConflict,
     NativeRuntimeManager,
     RuntimeCleanupPending,
+    RUNTIME_CLEANUP_SENTINEL,
     UnitStatus,
 )
 
@@ -181,6 +182,8 @@ class NativeRuntimeManagerTests(unittest.TestCase):
             browser_path="/bin/true",
             unit_manager=self.units,
             min_free_bytes=0,
+            display_start=1000,
+            display_end=1010,
         )
         self.manager._wait_for_display = lambda record: None
         self.manager._wait_for_cdp = lambda record: None
@@ -308,6 +311,24 @@ class NativeRuntimeManagerTests(unittest.TestCase):
         self.assertEqual(len(profile_dirs), 1)
         self.assertTrue(profile_dirs[0].is_dir())
 
+    def test_stopped_runtime_does_not_reserve_profile_for_another_alias(self) -> None:
+        stopped = self.manager.create(self.payload(stopped=True))
+        running = self.manager.create(self.payload(alias="account-b"))
+        self.assertEqual(running["state"], "running")
+        self.assertEqual(self.manager.list_public()[0]["state"], "stopped")
+        with self.assertRaisesRegex(BrowserRuntimeError, "Profile is already in use"):
+            self.manager.change_state("account-a", "start", {
+                "binding_version": 1,
+                "runtime_id": stopped["runtime_id"],
+                "network_id": stopped["network_id"],
+            })
+
+    def test_profile_conflict_does_not_leave_an_orphaned_runtime(self) -> None:
+        self.manager.create(self.payload())
+        with self.assertRaisesRegex(BrowserRuntimeError, "Profile is already in use"):
+            self.manager.create(self.payload(alias="account-b"))
+        self.assertEqual([item["alias"] for item in self.manager.list_public()], ["account-a"])
+
     def test_same_profile_is_exclusive_across_manager_instances(self) -> None:
         first = self.manager.create(self.payload())
         del first
@@ -436,6 +457,31 @@ class NativeRuntimeManagerTests(unittest.TestCase):
         self.assertEqual(self.manager.list_public()[-1]["runtime_id"], new["runtime_id"])
         self.assertEqual(self.manager.list_public()[-1]["state"], "stopped")
 
+    def test_sentinel_remove_with_only_released_records_is_noop(self) -> None:
+        """pause 后（只剩 released 记录）再删除：sentinel 代不代表实物，不应 409。"""
+        self.manager.create(self.payload())
+        record = next(r for r in self.manager._records() if r.alias == "account-a")
+        self.manager._stop_locked(record, released=True)
+        self.assertEqual(record.state, "released")
+        # 控制面在 DB runtime 已清空后发 sentinel 代删除。
+        self.manager.remove("account-a", {
+            "binding_version": 1,
+            "runtime_id": RUNTIME_CLEANUP_SENTINEL,
+            "network_id": "",
+        })
+
+    def test_empty_or_sentinel_runtime_id_purges_released_profile(self) -> None:
+        """账号删除（purge_profile）在 runtime 已释放后到达：空/哨兵代放行且清掉档案。"""
+        self.manager.create(self.payload())
+        record = next(r for r in self.manager._records() if r.alias == "account-a")
+        self.manager._stop_locked(record, released=True)
+        self.manager.remove("account-a", {
+            "binding_version": 1,
+            "runtime_id": "",
+            "network_id": "",
+        }, purge_profile=True)
+        self.assertFalse((self.profiles / "account-a").exists())
+
 
 if __name__ == "__main__":
     unittest.main()