fix: make desktop plugin dependency loading safer on Windows#7446
Merged
zouyonghe merged 7 commits intoAstrBotDevs:masterfrom Apr 10, 2026
Merged
fix: make desktop plugin dependency loading safer on Windows#7446zouyonghe merged 7 commits intoAstrBotDevs:masterfrom
zouyonghe merged 7 commits intoAstrBotDevs:masterfrom
Conversation
dosubot
bot
added
the
size:L
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
- In
_import_plugin_with_dependency_recovery, the guardif can_prefer_installed_dependencies:is now used both for preloading and for the recovery branch, which means we no longer attempt any dependency-based recovery when the precheck is unavailable or only reports version mismatches; consider decoupling the conditions so recovery still runs in those cases if that was previously supported behavior. - The new
allow_target_upgradeflag inPipInstaller.installonly affects the desktop--targetbranch; if the intention is to make this behavior explicit and future-proof, it may be worth asserting or commenting that the flag is ignored for non-desktop environments to avoid confusion for callers.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_import_plugin_with_dependency_recovery`, the guard `if can_prefer_installed_dependencies:` is now used both for preloading and for the recovery branch, which means we no longer attempt any dependency-based recovery when the precheck is unavailable or only reports version mismatches; consider decoupling the conditions so recovery still runs in those cases if that was previously supported behavior.
- The new `allow_target_upgrade` flag in `PipInstaller.install` only affects the desktop `--target` branch; if the intention is to make this behavior explicit and future-proof, it may be worth asserting or commenting that the flag is ignored for non-desktop environments to avoid confusion for callers.
## Individual Comments
### Comment 1
<location path="tests/test_main.py" line_range="60-63" />
<code_context>
+def test_check_env_appends_user_site_packages_after_runtime_paths(monkeypatch):
</code_context>
<issue_to_address>
**suggestion (testing):** Consider also testing that `check_env` does not append a duplicate `site_packages_path` when it is already present
Since the implementation checks `if site_packages_path not in sys.path` before appending, a test that pre-populates `sys.path` with `site_packages_path` would verify that no duplicate entry is added and that this guard continues to work as intended.
```suggestion
check_env()
def test_check_env_does_not_append_duplicate_user_site_packages(monkeypatch):
astrbot_root = "/tmp/astrbot-root"
site_packages_path = "/tmp/astrbot-site-packages"
original_sys_path = list(sys.path)
monkeypatch.setattr(sys, "version_info", _version_info(3, 12))
monkeypatch.setattr("main.get_astrbot_root", lambda: astrbot_root)
monkeypatch.setattr("main.get_astrbot_site_packages_path", lambda: site_packages_path)
monkeypatch.setattr("main.get_astrbot_config_path", lambda: "/tmp/config")
monkeypatch.setattr("main.get_astrbot_plugin_path", lambda: "/tmp/plugins")
monkeypatch.setattr("main.get_astrbot_temp_path", lambda: "/tmp/temp")
# Pre-populate sys.path with the user site-packages path to ensure it is not duplicated
sys.path = original_sys_path + [site_packages_path]
try:
check_env()
assert sys.path.count(site_packages_path) == 1
finally:
sys.path = original_sys_path
def test_check_env_appends_user_site_packages_after_runtime_paths(monkeypatch):
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
60
to
+63
| check_env() | ||
|
|
||
|
|
||
| def test_check_env_appends_user_site_packages_after_runtime_paths(monkeypatch): |
Contributor
There was a problem hiding this comment.
suggestion (testing): Consider also testing that check_env does not append a duplicate site_packages_path when it is already present
Since the implementation checks if site_packages_path not in sys.path before appending, a test that pre-populates sys.path with site_packages_path would verify that no duplicate entry is added and that this guard continues to work as intended.
Suggested change
| check_env() | |
| def test_check_env_appends_user_site_packages_after_runtime_paths(monkeypatch): | |
| check_env() | |
| def test_check_env_does_not_append_duplicate_user_site_packages(monkeypatch): | |
| astrbot_root = "/tmp/astrbot-root" | |
| site_packages_path = "/tmp/astrbot-site-packages" | |
| original_sys_path = list(sys.path) | |
| monkeypatch.setattr(sys, "version_info", _version_info(3, 12)) | |
| monkeypatch.setattr("main.get_astrbot_root", lambda: astrbot_root) | |
| monkeypatch.setattr("main.get_astrbot_site_packages_path", lambda: site_packages_path) | |
| monkeypatch.setattr("main.get_astrbot_config_path", lambda: "/tmp/config") | |
| monkeypatch.setattr("main.get_astrbot_plugin_path", lambda: "/tmp/plugins") | |
| monkeypatch.setattr("main.get_astrbot_temp_path", lambda: "/tmp/temp") | |
| # Pre-populate sys.path with the user site-packages path to ensure it is not duplicated | |
| sys.path = original_sys_path + [site_packages_path] | |
| try: | |
| check_env() | |
| assert sys.path.count(site_packages_path) == 1 | |
| finally: | |
| sys.path = original_sys_path | |
| def test_check_env_appends_user_site_packages_after_runtime_paths(monkeypatch): |
dosubot
bot
added
area:core
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces logic to handle dependency version mismatches and optimizes how user site-packages are added to the system path. Key changes include updating the pip_installer to conditionally allow target upgrades, tracking version mismatches in the requirements planning phase, and modifying the plugin import process to prefer installed dependencies when appropriate. Feedback focuses on potential issues with dependency loading order due to sys.path changes and identifies opportunities to reduce redundant operations during plugin initialization and requirement parsing.
astrbot/core/star/star_manager.py
Outdated
Comment on lines
+350
to
+355
| can_prefer_installed_dependencies = False | ||
| if os.path.exists(requirements_path) and not reserved: | ||
| install_plan = plan_missing_requirements_install(requirements_path) | ||
| can_prefer_installed_dependencies = ( | ||
| install_plan is not None and not install_plan.version_mismatch_names | ||
| ) |
Contributor
There was a problem hiding this comment.
If install_plan.version_mismatch_names is not empty, can_prefer_installed_dependencies becomes False. The code then proceeds to __import__(path, ...) without prepending the user site-packages path (since main.py now appends it to the end of sys.path).
If a bundled version of the dependency exists but is incompatible (which is the cause of the mismatch), __import__ will find and load the bundled version first. If this import succeeds (e.g., the bundled version is present but just the wrong version), the plugin will run with an incompatible dependency, and the recovery logic in the except block will never be triggered because no ImportError was raised.
You should consider triggering an update or ensuring the correct path is prepended if a version mismatch is detected before the first import attempt.
astrbot/core/star/star_manager.py
Outdated
| ) -> ModuleType: | ||
| can_prefer_installed_dependencies = False | ||
| if os.path.exists(requirements_path) and not reserved: | ||
| install_plan = plan_missing_requirements_install(requirements_path) |
Contributor
There was a problem hiding this comment.
The plan_missing_requirements_install function is called here during every plugin import. However, for plugins being installed or reloaded, this check has already been performed in _ensure_plugin_requirements (called via install_plugin or reload_failed_plugin). Since this function involves parsing requirements and scanning installed distributions, calling it twice for the same plugin is redundant and impacts performance, especially during startup or bulk plugin operations.
Comment on lines
+359
to
+361
| pip_installer.prefer_installed_dependencies( | ||
| requirements_path=requirements_path | ||
| ) |
Contributor
There was a problem hiding this comment.
Calling pip_installer.prefer_installed_dependencies immediately after plan_missing_requirements_install results in redundant work. prefer_installed_dependencies re-parses the requirements.txt file and re-scans the site-packages to identify candidate modules, both of which were just completed by plan_missing_requirements_install.
Consider refactoring prefer_installed_dependencies to accept the already parsed requirement names or the install_plan to avoid these duplicate operations.
dosubot
bot
added
size:XL
Member
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - I've found 1 issue
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="tests/test_plugin_manager.py" line_range="445-454" />
<code_context>
assert ensure_preferred_calls == [(str(site_packages_path), {"demo-package"})]
+@pytest.mark.asyncio
+async def test_install_keeps_target_upgrade_enabled_by_default_for_desktop_client(
+ monkeypatch, tmp_path
</code_context>
<issue_to_address>
**suggestion (testing):** Add a regression test for the case where a version-mismatched requirement causes import failure to go straight to reinstall, skipping installed-dependency recovery.
The new tests exercise most `_import_plugin_with_dependency_recovery` paths, but they miss the branch where `install_plan.version_mismatch_names` is non-empty and we intentionally skip installed-dependency recovery and go straight to reinstall. Please add a test that:
- uses an `install_plan` with a non-empty `version_mismatch_names`,
- makes the first import raise `ModuleNotFoundError`,
- asserts `pip_installer.prefer_installed_dependencies` is not called,
- asserts `_check_plugin_dept_update` is called (e.g., via monkeypatch), and
- verifies a second import attempt occurs and succeeds.
This will exercise the version-mismatch path end-to-end and guard against regressions where we accidentally try to recover from user site-packages in this case.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+445
to
+454
| @pytest.mark.asyncio | ||
| @pytest.mark.parametrize( | ||
| ("version_mismatch_names", "expected_allow_target_upgrade"), | ||
| [ | ||
| (set(), False), | ||
| ({"networkx"}, True), | ||
| ], | ||
| ) | ||
| async def test_ensure_plugin_requirements_sets_target_upgrade_based_on_version_mismatch( | ||
| plugin_manager_pm: PluginManager, |
Contributor
There was a problem hiding this comment.
suggestion (testing): Add a regression test for the case where a version-mismatched requirement causes import failure to go straight to reinstall, skipping installed-dependency recovery.
The new tests exercise most _import_plugin_with_dependency_recovery paths, but they miss the branch where install_plan.version_mismatch_names is non-empty and we intentionally skip installed-dependency recovery and go straight to reinstall. Please add a test that:
- uses an
install_planwith a non-emptyversion_mismatch_names, - makes the first import raise
ModuleNotFoundError, - asserts
pip_installer.prefer_installed_dependenciesis not called, - asserts
_check_plugin_dept_updateis called (e.g., via monkeypatch), and - verifies a second import attempt occurs and succeeds.
This will exercise the version-mismatch path end-to-end and guard against regressions where we accidentally try to recover from user site-packages in this case.
Member
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - I've left some high level feedback:
- The control flow in
_import_plugin_with_dependency_recoveryis getting hard to follow with multiple boolean flags (can_attempt_installed_dependency_recovery,can_preload_installed_dependencies) andinstall_planchecks; consider refactoring into smaller helpers (e.g., a precheck function returning an enum/state object) so the import/recovery decisions are easier to reason about. - Now that
_find_missing_requirements_from_linesis the core implementation andfind_missing_requirements_from_linesis a thin wrapper, it may be clearer to either rename the private helper to reflect that it returns both missing names and version mismatches (or make it public) to avoid future callers accidentally re-implementing or misusing this logic.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The control flow in `_import_plugin_with_dependency_recovery` is getting hard to follow with multiple boolean flags (`can_attempt_installed_dependency_recovery`, `can_preload_installed_dependencies`) and `install_plan` checks; consider refactoring into smaller helpers (e.g., a precheck function returning an enum/state object) so the import/recovery decisions are easier to reason about.
- Now that `_find_missing_requirements_from_lines` is the core implementation and `find_missing_requirements_from_lines` is a thin wrapper, it may be clearer to either rename the private helper to reflect that it returns both missing names and version mismatches (or make it public) to avoid future callers accidentally re-implementing or misusing this logic.Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Member
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- The new
_resolve_import_dependency_recovery_staterunsplan_missing_requirements_install(and thus a full requirements precheck) on every plugin import; consider caching theImportDependencyRecoveryStateper plugin/requirements path to avoid repeated file I/O and environment scans on hot import paths. - In
_import_plugin_with_dependency_recovery,ImportErrorfrom the plugin module itself (unrelated to dependencies) will trigger reinstall or recovery logic; if that's unintended, you could narrow the recovery handling toModuleNotFoundErrorof missing dependency names (e.g. by inspectingimport_exc.name) and let otherImportErrors surface directly.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new `_resolve_import_dependency_recovery_state` runs `plan_missing_requirements_install` (and thus a full requirements precheck) on every plugin import; consider caching the `ImportDependencyRecoveryState` per plugin/requirements path to avoid repeated file I/O and environment scans on hot import paths.
- In `_import_plugin_with_dependency_recovery`, `ImportError` from the plugin module itself (unrelated to dependencies) will trigger reinstall or recovery logic; if that's unintended, you could narrow the recovery handling to `ModuleNotFoundError` of missing dependency names (e.g. by inspecting `import_exc.name`) and let other `ImportError`s surface directly.
## Individual Comments
### Comment 1
<location path="astrbot/core/star/star_manager.py" line_range="408" />
<code_context>
+ )
+ return None
+
async def _import_plugin_with_dependency_recovery(
self,
path: str,
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying `_import_plugin_with_dependency_recovery` by inlining the recovery decision logic and removing the enum/dataclass-based mini state machine and helpers.
You can get the same behavior with much less abstraction by inlining the decision logic into `_import_plugin_with_dependency_recovery` and dropping the enum/dataclass + static helpers.
### 1. Remove the mini state machine
`ImportDependencyRecoveryMode`, `ImportDependencyRecoveryState`, `_resolve_import_dependency_recovery_state`, and `_try_import_from_installed_dependencies` can be removed and replaced by simple flags derived from `plan_missing_requirements_install`.
You can rewrite `_import_plugin_with_dependency_recovery` like this:
```python
from astrbot.core.utils.requirements_utils import plan_missing_requirements_install
async def _import_plugin_with_dependency_recovery(
self,
path: str,
module_str: str,
root_dir_name: str,
requirements_path: str,
*,
reserved: bool = False,
) -> ModuleType:
# Pre‑compute plan and flags
has_requirements = (not reserved) and os.path.exists(requirements_path)
install_plan = plan_missing_requirements_install(requirements_path) if has_requirements else None
has_version_mismatch = bool(install_plan and install_plan.version_mismatch_names)
can_use_installed_recovery = has_requirements and not has_version_mismatch
# Optional preload before first import
if can_use_installed_recovery and install_plan is not None:
try:
pip_installer.prefer_installed_dependencies(requirements_path=requirements_path)
except Exception as preload_exc:
logger.info(
"插件 %s 预加载已安装依赖失败，将继续常规导入: %s",
root_dir_name,
preload_exc,
)
try:
return __import__(path, fromlist=[module_str])
except (ModuleNotFoundError, ImportError) as import_exc:
# One recovery attempt using already-installed deps
if can_use_installed_recovery:
try:
logger.info(
"插件 %s 导入失败，尝试从已安装依赖恢复: %s",
root_dir_name,
import_exc,
)
pip_installer.prefer_installed_dependencies(requirements_path=requirements_path)
module = __import__(path, fromlist=[module_str])
logger.info("插件 %s 已从 site-packages 恢复依赖，跳过重新安装。", root_dir_name)
return module
except Exception as recover_exc:
logger.info(
"插件 %s 已安装依赖恢复失败，将重新安装依赖: %s",
root_dir_name,
recover_exc,
)
elif has_version_mismatch:
logger.info(
"插件 %s 预检查检测到版本不匹配，跳过已安装依赖恢复: %s",
root_dir_name,
sorted(install_plan.version_mismatch_names), # type: ignore[arg-type]
)
await self._check_plugin_dept_update(target_plugin=root_dir_name)
return __import__(path, fromlist=[module_str])
```
This preserves:
- `reserved` short‑circuiting.
- Preload before first import when there is a clean plan.
- Recovery via `prefer_installed_dependencies` on failure when there is no version mismatch.
- Skipping recovery and logging when `version_mismatch_names` is non‑empty, then forcing reinstall.
After this, you can safely delete:
```python
class ImportDependencyRecoveryMode(Enum): ...
@dataclass(frozen=True)
class ImportDependencyRecoveryState: ...
@staticmethod
def _resolve_import_dependency_recovery_state(...): ...
@staticmethod
def _try_import_from_installed_dependencies(...): ...
```
The behavior remains identical, but the control flow is linear and localized in one method, reducing the cognitive load and the number of moving parts you need to update when policies change.
</issue_to_address>
### Comment 2
<location path="astrbot/core/utils/requirements_utils.py" line_range="32" />
<code_context>
+ REINSTALL_ON_FAILURE = auto()
+
+
+@dataclass(frozen=True)
+class ImportDependencyRecoveryState:
+ mode: ImportDependencyRecoveryMode
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the design by removing the separate MissingRequirementsAnalysis type and using a single, defaulted MissingRequirementsPlan for both analysis and planning stages.
You can collapse one layer without losing any functionality by reusing `MissingRequirementsPlan` for both the “analysis” and “plan” stages.
### 1. Remove `MissingRequirementsAnalysis` and reuse `MissingRequirementsPlan`
`MissingRequirementsAnalysis` and `MissingRequirementsPlan` carry overlapping data and you immediately rewrap one into the other. You can instead make `MissingRequirementsPlan` usable as an “analysis-only” result by giving `install_lines` a default:
```python
@dataclass(frozen=True)
class MissingRequirementsPlan:
missing_names: frozenset[str]
install_lines: tuple[str, ...] = ()
version_mismatch_names: frozenset[str] = frozenset()
fallback_reason: str | None = None
```
Then change `classify_missing_requirements_from_lines` to return `MissingRequirementsPlan` directly:
```python
def classify_missing_requirements_from_lines(
requirement_lines: Sequence[str],
) -> MissingRequirementsPlan | None:
required = list(iter_requirements(lines=requirement_lines))
if not required:
return MissingRequirementsPlan(missing_names=frozenset())
installed = collect_installed_distribution_versions(get_requirement_check_paths())
if installed is None:
return None
missing: set[str] = set()
version_mismatch_names: set[str] = set()
for name, specifier in required:
installed_version = installed.get(name)
if not installed_version:
missing.add(name)
continue
if specifier and not _specifier_contains_version(specifier, installed_version):
missing.add(name)
version_mismatch_names.add(name)
return MissingRequirementsPlan(
missing_names=frozenset(missing),
version_mismatch_names=frozenset(version_mismatch_names),
)
```
Now `MissingRequirementsAnalysis` can be removed entirely.
### 2. Keep `find_missing_requirements_from_lines` as a thin compatibility wrapper
If you still need the old API, keep it but delegate to the enriched plan, avoiding an extra public “analysis” type:
```python
def find_missing_requirements_from_lines(
requirement_lines: Sequence[str],
) -> set[str] | None:
plan = classify_missing_requirements_from_lines(requirement_lines)
if plan is None:
return None
return set(plan.missing_names)
```
Callers that care only about names can continue using this; callers that need more detail can use `classify_missing_requirements_from_lines` and work with the single `MissingRequirementsPlan` type.
### 3. Simplify `plan_missing_requirements_install` using the unified type
With the unified plan/analysis type, you avoid unpacking/repacking:
```python
def plan_missing_requirements_install(
requirements_path: str,
) -> MissingRequirementsPlan | None:
can_precheck, requirement_lines = _load_requirement_lines_for_precheck(
requirements_path
)
if not can_precheck or requirement_lines is None:
return None
analysis = classify_missing_requirements_from_lines(requirement_lines)
if analysis is None:
return None
missing = analysis.missing_names
install_lines = build_missing_requirements_install_lines(
requirements_path,
requirement_lines,
missing,
)
if install_lines is None:
return None
if missing and not install_lines:
logger.warning(
"预检查缺失依赖成功，但无法映射到可安装 requirement 行，将回退到完整安装: %s -> %s",
requirements_path,
sorted(missing),
)
return MissingRequirementsPlan(
missing_names=analysis.missing_names,
version_mismatch_names=analysis.version_mismatch_names,
install_lines=(),
fallback_reason="unmapped missing requirement names",
)
return MissingRequirementsPlan(
missing_names=analysis.missing_names,
version_mismatch_names=analysis.version_mismatch_names,
install_lines=install_lines,
)
```
This keeps all current behavior (`version_mismatch_names` included), but reduces:
- Number of public types (`MissingRequirementsAnalysis` removed).
- Number of “stages” to understand (analysis and plan are the same structure).
- API surface: you now have a simple progression from `classify_missing_requirements_from_lines` → `plan_missing_requirements_install`, with an optional compatibility helper `find_missing_requirements_from_lines`.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| ) | ||
| return None | ||
|
|
||
| async def _import_plugin_with_dependency_recovery( |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider simplifying _import_plugin_with_dependency_recovery by inlining the recovery decision logic and removing the enum/dataclass-based mini state machine and helpers.
You can get the same behavior with much less abstraction by inlining the decision logic into _import_plugin_with_dependency_recovery and dropping the enum/dataclass + static helpers.
1. Remove the mini state machine
ImportDependencyRecoveryMode, ImportDependencyRecoveryState, _resolve_import_dependency_recovery_state, and _try_import_from_installed_dependencies can be removed and replaced by simple flags derived from plan_missing_requirements_install.
You can rewrite _import_plugin_with_dependency_recovery like this:
from astrbot.core.utils.requirements_utils import plan_missing_requirements_install
async def _import_plugin_with_dependency_recovery(
self,
path: str,
module_str: str,
root_dir_name: str,
requirements_path: str,
*,
reserved: bool = False,
) -> ModuleType:
# Pre‑compute plan and flags
has_requirements = (not reserved) and os.path.exists(requirements_path)
install_plan = plan_missing_requirements_install(requirements_path) if has_requirements else None
has_version_mismatch = bool(install_plan and install_plan.version_mismatch_names)
can_use_installed_recovery = has_requirements and not has_version_mismatch
# Optional preload before first import
if can_use_installed_recovery and install_plan is not None:
try:
pip_installer.prefer_installed_dependencies(requirements_path=requirements_path)
except Exception as preload_exc:
logger.info(
"插件 %s 预加载已安装依赖失败，将继续常规导入: %s",
root_dir_name,
preload_exc,
)
try:
return __import__(path, fromlist=[module_str])
except (ModuleNotFoundError, ImportError) as import_exc:
# One recovery attempt using already-installed deps
if can_use_installed_recovery:
try:
logger.info(
"插件 %s 导入失败，尝试从已安装依赖恢复: %s",
root_dir_name,
import_exc,
)
pip_installer.prefer_installed_dependencies(requirements_path=requirements_path)
module = __import__(path, fromlist=[module_str])
logger.info("插件 %s 已从 site-packages 恢复依赖，跳过重新安装。", root_dir_name)
return module
except Exception as recover_exc:
logger.info(
"插件 %s 已安装依赖恢复失败，将重新安装依赖: %s",
root_dir_name,
recover_exc,
)
elif has_version_mismatch:
logger.info(
"插件 %s 预检查检测到版本不匹配，跳过已安装依赖恢复: %s",
root_dir_name,
sorted(install_plan.version_mismatch_names), # type: ignore[arg-type]
)
await self._check_plugin_dept_update(target_plugin=root_dir_name)
return __import__(path, fromlist=[module_str])This preserves:
reservedshort‑circuiting.- Preload before first import when there is a clean plan.
- Recovery via
prefer_installed_dependencieson failure when there is no version mismatch. - Skipping recovery and logging when
version_mismatch_namesis non‑empty, then forcing reinstall.
After this, you can safely delete:
class ImportDependencyRecoveryMode(Enum): ...
@dataclass(frozen=True)
class ImportDependencyRecoveryState: ...
@staticmethod
def _resolve_import_dependency_recovery_state(...): ...
@staticmethod
def _try_import_from_installed_dependencies(...): ...The behavior remains identical, but the control flow is linear and localized in one method, reducing the cognitive load and the number of moving parts you need to update when policies change.
| requirement_names: frozenset[str] | ||
|
|
||
|
|
||
| @dataclass(frozen=True) |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider simplifying the design by removing the separate MissingRequirementsAnalysis type and using a single, defaulted MissingRequirementsPlan for both analysis and planning stages.
You can collapse one layer without losing any functionality by reusing MissingRequirementsPlan for both the “analysis” and “plan” stages.
1. Remove MissingRequirementsAnalysis and reuse MissingRequirementsPlan
MissingRequirementsAnalysis and MissingRequirementsPlan carry overlapping data and you immediately rewrap one into the other. You can instead make MissingRequirementsPlan usable as an “analysis-only” result by giving install_lines a default:
@dataclass(frozen=True)
class MissingRequirementsPlan:
missing_names: frozenset[str]
install_lines: tuple[str, ...] = ()
version_mismatch_names: frozenset[str] = frozenset()
fallback_reason: str | None = NoneThen change classify_missing_requirements_from_lines to return MissingRequirementsPlan directly:
def classify_missing_requirements_from_lines(
requirement_lines: Sequence[str],
) -> MissingRequirementsPlan | None:
required = list(iter_requirements(lines=requirement_lines))
if not required:
return MissingRequirementsPlan(missing_names=frozenset())
installed = collect_installed_distribution_versions(get_requirement_check_paths())
if installed is None:
return None
missing: set[str] = set()
version_mismatch_names: set[str] = set()
for name, specifier in required:
installed_version = installed.get(name)
if not installed_version:
missing.add(name)
continue
if specifier and not _specifier_contains_version(specifier, installed_version):
missing.add(name)
version_mismatch_names.add(name)
return MissingRequirementsPlan(
missing_names=frozenset(missing),
version_mismatch_names=frozenset(version_mismatch_names),
)Now MissingRequirementsAnalysis can be removed entirely.
2. Keep find_missing_requirements_from_lines as a thin compatibility wrapper
If you still need the old API, keep it but delegate to the enriched plan, avoiding an extra public “analysis” type:
def find_missing_requirements_from_lines(
requirement_lines: Sequence[str],
) -> set[str] | None:
plan = classify_missing_requirements_from_lines(requirement_lines)
if plan is None:
return None
return set(plan.missing_names)Callers that care only about names can continue using this; callers that need more detail can use classify_missing_requirements_from_lines and work with the single MissingRequirementsPlan type.
3. Simplify plan_missing_requirements_install using the unified type
With the unified plan/analysis type, you avoid unpacking/repacking:
def plan_missing_requirements_install(
requirements_path: str,
) -> MissingRequirementsPlan | None:
can_precheck, requirement_lines = _load_requirement_lines_for_precheck(
requirements_path
)
if not can_precheck or requirement_lines is None:
return None
analysis = classify_missing_requirements_from_lines(requirement_lines)
if analysis is None:
return None
missing = analysis.missing_names
install_lines = build_missing_requirements_install_lines(
requirements_path,
requirement_lines,
missing,
)
if install_lines is None:
return None
if missing and not install_lines:
logger.warning(
"预检查缺失依赖成功，但无法映射到可安装 requirement 行，将回退到完整安装: %s -> %s",
requirements_path,
sorted(missing),
)
return MissingRequirementsPlan(
missing_names=analysis.missing_names,
version_mismatch_names=analysis.version_mismatch_names,
install_lines=(),
fallback_reason="unmapped missing requirement names",
)
return MissingRequirementsPlan(
missing_names=analysis.missing_names,
version_mismatch_names=analysis.version_mismatch_names,
install_lines=install_lines,
)This keeps all current behavior (version_mismatch_names included), but reduces:
- Number of public types (
MissingRequirementsAnalysisremoved). - Number of “stages” to understand (analysis and plan are the same structure).
- API surface: you now have a simple progression from
classify_missing_requirements_from_lines→plan_missing_requirements_install, with an optional compatibility helperfind_missing_requirements_from_lines.
Member
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- In
_try_import_from_installed_dependencies, consider narrowing theexcept ExceptiontoModuleNotFoundError/ImportError(or a small set of expected failures) so that unexpected programming errors in the recovery path are not silently converted into reinstall attempts. - The new import-recovery tests in
test_plugin_manager.pyhave a lot of repeated boilerplate for setting uprequirements_path,events, and the fake__import__; you could factor a small helper or fixture to reduce duplication and make the individual test intentions clearer.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_try_import_from_installed_dependencies`, consider narrowing the `except Exception` to `ModuleNotFoundError`/`ImportError` (or a small set of expected failures) so that unexpected programming errors in the recovery path are not silently converted into reinstall attempts.
- The new import-recovery tests in `test_plugin_manager.py` have a lot of repeated boilerplate for setting up `requirements_path`, `events`, and the fake `__import__`; you could factor a small helper or fixture to reduce duplication and make the individual test intentions clearer.
## Individual Comments
### Comment 1
<location path="astrbot/core/star/star_manager.py" line_range="417-419" />
<code_context>
+ *,
+ reserved: bool = False,
) -> ModuleType:
+ recovery_state = self._resolve_import_dependency_recovery_state(
+ requirements_path,
+ reserved=reserved,
+ )
+
</code_context>
<issue_to_address>
**question (bug_risk):** Reserved plugins (or missing requirements files) still trigger dependency reinstall on import failure, which might not be desired.
`_resolve_import_dependency_recovery_state` returns `DISABLED` when `reserved=True` or the requirements file is missing, but the import-failure path still calls `await self._check_plugin_dept_update(...)` and re-imports. If `reserved` is intended to block dependency changes, consider short‑circuiting before `_check_plugin_dept_update` when `recovery_state.mode` is `DISABLED`.
</issue_to_address>
### Comment 2
<location path="astrbot/core/star/star_manager.py" line_range="83" />
<code_context>
self.error = error
+class ImportDependencyRecoveryMode(Enum):
+ DISABLED = auto()
+ PRELOAD_AND_RECOVER = auto()
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the new import-dependency recovery logic by replacing the enum/dataclass and helper with straightforward local flags and a single, linear control flow inside `_import_plugin_with_dependency_recovery`.
The new `ImportDependencyRecoveryMode` / `ImportDependencyRecoveryState` plus `_try_import_from_installed_dependencies` do add a mini state machine and extra indirection without buying much reuse. You can keep all the new behaviors (preload, version-mismatch skip, reserved handling) while simplifying control flow by:
* Dropping the enum/dataclass in favor of local flags.
* Inlining `_try_import_from_installed_dependencies` into `_import_plugin_with_dependency_recovery`.
* Keeping all decisions localized in one function so the happy path and failure paths are easy to read.
A possible refactor (abridged for focus) could look like:
```python
async def _import_plugin_with_dependency_recovery(
self,
path: str,
module_str: str,
root_dir_name: str,
requirements_path: str,
*,
reserved: bool = False,
) -> ModuleType:
has_requirements = os.path.exists(requirements_path)
install_plan = None
should_preload = False
should_attempt_recover = False
should_skip_recover_due_to_version_mismatch = False
if not reserved and has_requirements:
install_plan = plan_missing_requirements_install(requirements_path)
if install_plan is None:
# No precomputed plan; we can still try recovery on failure
should_attempt_recover = True
elif install_plan.version_mismatch_names:
# Known-bad versions → skip recovery, force reinstall
should_skip_recover_due_to_version_mismatch = True
else:
# We have a clean plan, so we can safely preload and later recover
should_preload = True
should_attempt_recover = True
if should_preload:
try:
pip_installer.prefer_installed_dependencies(
requirements_path=requirements_path,
)
except Exception as preload_exc:
logger.info(
"插件 %s 预加载已安装依赖失败，将继续常规导入: %s",
root_dir_name,
preload_exc,
)
try:
return __import__(path, fromlist=[module_str])
except ModuleNotFoundError as import_exc:
recovered_module: ModuleType | None = None
if should_attempt_recover and not should_skip_recover_due_to_version_mismatch:
try:
logger.info(
"插件 %s 导入失败，尝试从已安装依赖恢复: %s",
root_dir_name,
import_exc,
)
pip_installer.prefer_installed_dependencies(
requirements_path=requirements_path,
)
recovered_module = __import__(path, fromlist=[module_str])
logger.info(
"插件 %s 已从 site-packages 恢复依赖，跳过重新安装。",
root_dir_name,
)
except Exception as recover_exc:
logger.info(
"插件 %s 已安装依赖恢复失败，将重新安装依赖: %s",
root_dir_name,
recover_exc,
)
if recovered_module is None:
if should_skip_recover_due_to_version_mismatch and install_plan:
logger.info(
"插件 %s 预检查检测到版本不匹配，"
"跳过已安装依赖恢复: %s",
root_dir_name,
sorted(install_plan.version_mismatch_names),
)
await self._check_plugin_dept_update(target_plugin=root_dir_name)
return __import__(path, fromlist=[module_str])
return recovered_module
```
This preserves:
* Preload behavior when we have a safe `install_plan`.
* Recovery attempt when possible.
* Skipping recovery when version mismatches are pre-detected.
* `reserved`/missing requirements disabling recovery entirely.
And it removes:
* `ImportDependencyRecoveryMode`
* `ImportDependencyRecoveryState`
* `_resolve_import_dependency_recovery_state`
* `_try_import_from_installed_dependencies`
making the import-recovery flow read as a single, straightforward sequence.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+417
to
+419
| recovery_state = self._resolve_import_dependency_recovery_state( | ||
| requirements_path, | ||
| reserved=reserved, |
Contributor
There was a problem hiding this comment.
question (bug_risk): Reserved plugins (or missing requirements files) still trigger dependency reinstall on import failure, which might not be desired.
_resolve_import_dependency_recovery_state returns DISABLED when reserved=True or the requirements file is missing, but the import-failure path still calls await self._check_plugin_dept_update(...) and re-imports. If reserved is intended to block dependency changes, consider short‑circuiting before _check_plugin_dept_update when recovery_state.mode is DISABLED.
| self.error = error | ||
|
|
||
|
|
||
| class ImportDependencyRecoveryMode(Enum): |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider simplifying the new import-dependency recovery logic by replacing the enum/dataclass and helper with straightforward local flags and a single, linear control flow inside _import_plugin_with_dependency_recovery.
The new ImportDependencyRecoveryMode / ImportDependencyRecoveryState plus _try_import_from_installed_dependencies do add a mini state machine and extra indirection without buying much reuse. You can keep all the new behaviors (preload, version-mismatch skip, reserved handling) while simplifying control flow by:
- Dropping the enum/dataclass in favor of local flags.
- Inlining
_try_import_from_installed_dependenciesinto_import_plugin_with_dependency_recovery. - Keeping all decisions localized in one function so the happy path and failure paths are easy to read.
A possible refactor (abridged for focus) could look like:
async def _import_plugin_with_dependency_recovery(
self,
path: str,
module_str: str,
root_dir_name: str,
requirements_path: str,
*,
reserved: bool = False,
) -> ModuleType:
has_requirements = os.path.exists(requirements_path)
install_plan = None
should_preload = False
should_attempt_recover = False
should_skip_recover_due_to_version_mismatch = False
if not reserved and has_requirements:
install_plan = plan_missing_requirements_install(requirements_path)
if install_plan is None:
# No precomputed plan; we can still try recovery on failure
should_attempt_recover = True
elif install_plan.version_mismatch_names:
# Known-bad versions → skip recovery, force reinstall
should_skip_recover_due_to_version_mismatch = True
else:
# We have a clean plan, so we can safely preload and later recover
should_preload = True
should_attempt_recover = True
if should_preload:
try:
pip_installer.prefer_installed_dependencies(
requirements_path=requirements_path,
)
except Exception as preload_exc:
logger.info(
"插件 %s 预加载已安装依赖失败，将继续常规导入: %s",
root_dir_name,
preload_exc,
)
try:
return __import__(path, fromlist=[module_str])
except ModuleNotFoundError as import_exc:
recovered_module: ModuleType | None = None
if should_attempt_recover and not should_skip_recover_due_to_version_mismatch:
try:
logger.info(
"插件 %s 导入失败，尝试从已安装依赖恢复: %s",
root_dir_name,
import_exc,
)
pip_installer.prefer_installed_dependencies(
requirements_path=requirements_path,
)
recovered_module = __import__(path, fromlist=[module_str])
logger.info(
"插件 %s 已从 site-packages 恢复依赖，跳过重新安装。",
root_dir_name,
)
except Exception as recover_exc:
logger.info(
"插件 %s 已安装依赖恢复失败，将重新安装依赖: %s",
root_dir_name,
recover_exc,
)
if recovered_module is None:
if should_skip_recover_due_to_version_mismatch and install_plan:
logger.info(
"插件 %s 预检查检测到版本不匹配，"
"跳过已安装依赖恢复: %s",
root_dir_name,
sorted(install_plan.version_mismatch_names),
)
await self._check_plugin_dept_update(target_plugin=root_dir_name)
return __import__(path, fromlist=[module_str])
return recovered_moduleThis preserves:
- Preload behavior when we have a safe
install_plan. - Recovery attempt when possible.
- Skipping recovery when version mismatches are pre-detected.
reserved/missing requirements disabling recovery entirely.
And it removes:
ImportDependencyRecoveryModeImportDependencyRecoveryState_resolve_import_dependency_recovery_state_try_import_from_installed_dependencies
making the import-recovery flow read as a single, straightforward sequence.
Member
Author
|
@sourcery-ai review |
foreveruand
added a commit
to foreveruand/AstrBot
that referenced
this pull request
Apr 11, 2026
…ng origin patches Upstream changes integrated: - feat(discord): add configurable bot message filtering (AstrBotDevs#6505) - docs: fix path concatenation error in storage.md (AstrBotDevs#7448) - chore: remove lxml and bs4 deps (AstrBotDevs#7449) - fix: make desktop plugin dependency loading safer on Windows (AstrBotDevs#7446) - fix: split long telegram final segments (AstrBotDevs#7432) - adopted _send_text_chunks method - perf: merge 3 cron tools into 1 cron manage tool with edit capability (AstrBotDevs#7445) - chore: update logo in README.md Origin patches preserved: - Telegram enhancements: multi-image/video, spoiler, caption, reply_markup support - Telegram inline query and callback query event handlers - Telegram streaming improvements with draft API - Gemini provider fixes and enhancements - WeChat session renewal fix - Shell timeout parameter support Conflicts resolved in tg_event.py: adopted upstream's _send_text_chunks method while preserving origin's advanced message handling features. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
data/site-packagesbehind bundled runtime imports by default so stale user-installed packages no longer override core dependencies on startupTesting
Closes #6958
Summary by Sourcery
Improve desktop plugin dependency handling and import recovery to avoid stale user-installed packages overriding bundled dependencies, especially on Windows.
Bug Fixes:
Enhancements:
Tests: