feat: 支持复用已有进程执行任务 - #234
Merged
Merged
Conversation
Reviewer's GuideAdds end-to-end support for reusing an already-running MFW process to execute a scheduled configuration, including new single-instance IPC commands, CLI flag wiring, scheduler serialization/parsing, and UI controls. Sequence diagram for reusing an existing process to run a scheduled configsequenceDiagram
actor Scheduler
participant CliProcess as mfw_cli_process
participant SingleInstanceGuard
participant ActivationServer
participant MainWindow
participant ServiceCoordinator
Scheduler->>CliProcess: start MFW (--config-id, --reuse-existing)
CliProcess->>SingleInstanceGuard: acquire()
alt existing_instance_not_running
SingleInstanceGuard-->>CliProcess: acquired = True
CliProcess->>CliProcess: normal startup
else existing_instance_running
SingleInstanceGuard-->>CliProcess: acquired = False
CliProcess->>SingleInstanceGuard: request_existing_instance_run(config_id, force_start)
SingleInstanceGuard->>ActivationServer: CMD_RUN_CONFIG + JSON
ActivationServer->>ActivationServer: _handle_reuse_run_request(config_id, force_start)
alt window_not_ready
ActivationServer->>ActivationServer: pending_reuse_request[request] = (config_id, force_start)
ActivationServer-->>SingleInstanceGuard: RESP_ACCEPTED
else window_ready_and_config_valid
ActivationServer->>MainWindow: _schedule_reused_run(window, config_id, force_start)
ActivationServer-->>SingleInstanceGuard: RESP_ACCEPTED
MainWindow->>ServiceCoordinator: stop_task(manual=True) [optional]
MainWindow->>ServiceCoordinator: select_config(config_id)
MainWindow->>ServiceCoordinator: run_tasks_flow()
else task_running_and_not_force_start
ActivationServer-->>SingleInstanceGuard: RESP_BUSY
else invalid_config
ActivationServer-->>SingleInstanceGuard: RESP_INVALID
end
SingleInstanceGuard-->>CliProcess: response (RESP_ACCEPTED | RESP_BUSY | RESP_INVALID)
alt RESP_ACCEPTED or RESP_BUSY
CliProcess->>CliProcess: exit with code 0
else other_response
CliProcess->>CliProcess: exit with code 1
end
end
File-Level Changes
Assessment against linked issues
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 7 issues, and left some high level feedback:
- In
_handle_reuse_run_request, a request with an invalidconfig_idis accepted whenwindow is Noneand only rejected later in_schedule_reused_run, so the IPC caller receivesRESP_ACCEPTEDeven though the config doesn’t exist; consider validatingconfig_idbefore settingpending_reuse_requestto keep responses consistent. - When the main window already exists in
_handle_reuse_run_request, you bypasspending_reuse_requestand schedule_schedule_reused_rundirectly; it may be clearer and less error-prone to route both paths through the same pending/dispatch mechanism so reuse requests are handled uniformly.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_handle_reuse_run_request`, a request with an invalid `config_id` is accepted when `window is None` and only rejected later in `_schedule_reused_run`, so the IPC caller receives `RESP_ACCEPTED` even though the config doesn’t exist; consider validating `config_id` before setting `pending_reuse_request` to keep responses consistent.
- When the main window already exists in `_handle_reuse_run_request`, you bypass `pending_reuse_request` and schedule `_schedule_reused_run` directly; it may be clearer and less error-prone to route both paths through the same pending/dispatch mechanism so reuse requests are handled uniformly.
## Individual Comments
### Comment 1
<location path="main.py" line_range="171-180" />
<code_context>
single_instance = SingleInstanceGuard(instance_key)
if not single_instance.acquire():
+ if options.reuse_existing:
+ from app.utils.single_instance import RESP_ACCEPTED, RESP_BUSY
+ from PySide6.QtCore import QCoreApplication
+
+ _ = QCoreApplication.instance()
+ if ipc_app is None:
+ _ = QCoreApplication([sys.argv[0]])
+ response = single_instance.request_existing_instance_run(
+ options.config_id or "", force_start=options.force_restart
+ )
+ if response == RESP_ACCEPTED:
+ logger.info("已有实例已接收复用执行请求")
+ return 0
+ if response == RESP_BUSY:
+ logger.info("已有实例正在执行任务，已跳过复用执行请求")
+ return 0
+ logger.warning("向已有实例发送复用执行请求失败: %s", response)
+ return 1
+
from app.utils.startup_dialog import run_duplicate_instance_flow
</code_context>
<issue_to_address>
**question:** When reuse IPC fails, the process exits with code 1 without trying the duplicate-instance flow, which may be surprising for users.
In the `not single_instance.acquire()` branch, when `options.reuse_existing` is set and the IPC request fails or returns `RESP_INVALID`/`RESP_FAIL`, you log a warning and return `1`, skipping `run_duplicate_instance_flow`. Previously, duplicate-instance cases would still show the startup dialog. For `--reuse-existing`, it may be preferable to fall back to the usual duplicate-instance UI (or allow starting a new instance) rather than a hard error exit, especially if the existing instance is stuck. Consider routing non-success responses to the duplicate-instance flow instead of always exiting nonzero.
</issue_to_address>
### Comment 2
<location path="app/view/schedule_interface/schedule_interface.py" line_range="468-471" />
<code_context>
self.schedule_table = TableWidget(self)
- self.schedule_table.setColumnCount(8)
+ self.schedule_table.setColumnCount(9)
self.schedule_table.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed
</code_context>
<issue_to_address>
**suggestion:** Column indices for the schedule table are now more fragile; consider centralizing them to avoid off-by-one errors.
The new 9-column layout hard-codes the indices for "Force", "Reuse", "Admin", "Enabled", and "Action" in several places (headers, resize modes, widths, hidden-column logic, row population). This scatters magic numbers (5/6/7/8) and makes future reordering brittle. Defining named constants or an enum for these column indices would localize changes and reduce the risk of mismatched columns as the table evolves.
Suggested implementation:
```python
layout.addWidget(title)
# Centralized column indices for schedule_table to avoid scattered magic numbers.
# NOTE: Verify these indices match the actual header order:
# [ ..., "Pattern", "Next run", "Force", "Reuse", "Admin", "Enabled", "Action" ]
self.COL_FORCE = 5
self.COL_REUSE = 6
self.COL_ADMIN = 7
self.COL_ENABLED = 8
self.COL_ACTION = 9 # Adjust to the correct index based on actual column layout.
self.schedule_table = TableWidget(self)
self.schedule_table.setColumnCount(9)
self.schedule_table.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed
)
```
To fully implement the centralization and remove fragile indices, you should:
1. Replace all hard-coded indices for these columns in the file:
- Any occurrence of `5` that refers to "Force" should become `self.COL_FORCE`.
- Any occurrence of `6` that refers to "Reuse" should become `self.COL_REUSE`.
- Any occurrence of `7` that refers to "Admin" should become `self.COL_ADMIN`.
- Any occurrence of the "Enabled" column index (likely `4` or `8` depending on your layout) should become `self.COL_ENABLED`.
- Any occurrence of `8` or `9` that refers to "Action" should become `self.COL_ACTION`.
2. Update:
- Header setup (if you manually set header labels by index rather than via `enumerate`).
- Column resize modes (e.g. `self.schedule_table.horizontalHeader().setSectionResizeMode(index, ...)`).
- Column widths (e.g. `setColumnWidth(index, ...)`).
- Hidden/visible column logic (e.g. `setColumnHidden(index, ...)`).
- Row population logic (e.g. `setItem(row, index, ...)`, `setCellWidget(row, index, ...)`).
3. After replacing all usages, double-check the actual header order and adjust the values assigned to `self.COL_FORCE`, `self.COL_REUSE`, `self.COL_ADMIN`, `self.COL_ENABLED`, and `self.COL_ACTION` so they match the real indices (0-based) of your table columns. This will ensure future column reordering only requires changes in this one centralized block.
</issue_to_address>
### Comment 3
<location path="tests/test_reuse_existing.py" line_range="13-22" />
<code_context>
+)
+
+
+class ReuseExistingTests(unittest.TestCase):
+ def test_cli_parses_reuse_existing(self) -> None:
+ options, _, _ = parse_startup_cli(
+ ["--config-id=cfg_demo", "--direct-run", FLAG_REUSE_EXISTING]
+ )
+ self.assertTrue(options.reuse_existing)
+ self.assertEqual(options.config_id, "cfg_demo")
+
+ def test_cli_serializes_reuse_existing(self) -> None:
+ argv = build_startup_argv(
+ StartupOptions(config_id="cfg_demo", reuse_existing=True)
+ )
+ self.assertIn(FLAG_REUSE_EXISTING, argv)
+
+ def test_schedule_entry_defaults_reuse_existing_to_false(self) -> None:
</code_context>
<issue_to_address>
**suggestion (testing):** Add tests to cover the negative/default cases for `reuse_existing` in CLI
Right now only the "flag present" path is tested. Please also add tests that assert: (1) `parse_startup_cli` sets `options.reuse_existing` to `False` when the flag is omitted, and (2) `build_startup_argv` omits `FLAG_REUSE_EXISTING` when `reuse_existing` is `False`, so changes to defaults or unintended flag inclusion are caught early.
</issue_to_address>
### Comment 4
<location path="tests/test_unix_system_scheduler.py" line_range="147-151" />
<code_context>
self.assertTrue(job.startswith("sudo "), f"expected sudo prefix, got: {job}")
self.assertIn("--config-id=cfg_demo", job)
+ def test_build_shell_job_includes_reuse_existing(self) -> None:
+ job = build_shell_job(
+ "cfg_demo", force_start=False, reuse_existing=True
+ )
+ self.assertIn("--reuse-existing", job)
+
def test_split_preserves_other_instance_blocks(self) -> None:
</code_context>
<issue_to_address>
**suggestion (testing):** Also assert that `--reuse-existing` is absent when `reuse_existing=False`
Since we now cover the `reuse_existing=True` case, please add a complementary test for `reuse_existing=False` that asserts `"--reuse-existing"` is not present in the job string. This helps ensure the flag is only added when explicitly requested and protects against regressions where it could be appended unconditionally.
</issue_to_address>
### Comment 5
<location path="tests/test_windows_system_scheduler.py" line_range="74-70" />
<code_context>
+ "cfg_demo", force_start=True, reuse_existing=False
+ )
+
+ @patch(
+ "app.core.service.system_scheduler.windows.resolve_schedule_launch_command",
+ return_value=(
+ r"C:\MFW\MFW.exe",
+ "--config-id=cfg_demo --direct-run --reuse-existing",
+ ),
+ )
+ def test_reuse_existing_uses_cli_arguments(self, mock_command: object) -> None:
+ build_task_xml(
+ self._entry(
+ schedule_type=SCHEDULE_SINGLE,
+ params={"run_at": "2025-06-18T09:30:00"},
+ reuse_existing=True,
+ )
+ )
+ mock_command.assert_called_once_with(
+ "cfg_demo", force_start=False, reuse_existing=True
+ )
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding tests for parsing `--reuse-existing` back into `ScheduleEntry` from Windows tasks
The current test verifies `build_task_xml` forwards `reuse_existing=True` into `resolve_schedule_launch_command`, but we’re missing coverage for the reverse direction. Since the Windows scheduler now parses `--reuse-existing` into `ScheduleEntry.reuse_existing`, please add tests that construct minimal task XML with and without `--reuse-existing` in the arguments and assert that the parsed `ScheduleEntry.reuse_existing` is `True` and `False` respectively to validate round-trip behavior.
Suggested implementation:
```python
@patch(
"app.core.service.system_scheduler.windows.resolve_schedule_launch_command",
return_value=(
r"C:\MFW\MFW.exe",
"--config-id=cfg_demo --direct-run --reuse-existing",
),
)
def test_reuse_existing_uses_cli_arguments(self, mock_command: object) -> None:
build_task_xml(
self._entry(
schedule_type=SCHEDULE_SINGLE,
params={"run_at": "2025-06-18T09:30:00"},
reuse_existing=True,
)
)
mock_command.assert_called_once_with(
"cfg_demo", force_start=False, reuse_existing=True
)
def test_parse_task_xml_sets_reuse_existing_true(self) -> None:
"""Task XML with --reuse-existing in the arguments should set ScheduleEntry.reuse_existing=True."""
# Minimal task XML containing the expected launch command arguments including --reuse-existing
task_xml = """
<Task xmlns="http://schemas.microsoft.com/windows/2004/02/mit/task">
<RegistrationInfo>
<Date>2025-06-18T09:30:00</Date>
</RegistrationInfo>
<Triggers>
<TimeTrigger>
<StartBoundary>2025-06-18T09:30:00</StartBoundary>
</TimeTrigger>
</Triggers>
<Actions Context="Author">
<Exec>
<Command>C:\\MFW\\MFW.exe</Command>
<Arguments>--config-id=cfg_demo --direct-run --reuse-existing</Arguments>
</Exec>
</Actions>
</Task>
""".strip()
# This helper/function name should match whatever is currently used in tests
# to parse Windows task XML into ScheduleEntry instances.
entry = parse_task_xml_to_schedule_entry(task_xml)
assert isinstance(entry, ScheduleEntry)
assert entry.reuse_existing is True
def test_parse_task_xml_sets_reuse_existing_false_by_default(self) -> None:
"""Task XML without --reuse-existing should set ScheduleEntry.reuse_existing=False."""
task_xml = """
<Task xmlns="http://schemas.microsoft.com/windows/2004/02/mit/task">
<RegistrationInfo>
<Date>2025-06-18T09:30:00</Date>
</RegistrationInfo>
<Triggers>
<TimeTrigger>
<StartBoundary>2025-06-18T09:30:00</StartBoundary>
</TimeTrigger>
</Triggers>
<Actions Context="Author">
<Exec>
<Command>C:\\MFW\\MFW.exe</Command>
<Arguments>--config-id=cfg_demo --direct-run</Arguments>
</Exec>
</Actions>
</Task>
""".strip()
entry = parse_task_xml_to_schedule_entry(task_xml)
assert isinstance(entry, ScheduleEntry)
assert entry.reuse_existing is False
```
The added tests assume the existence of:
- A function `parse_task_xml_to_schedule_entry(task_xml: str) -> ScheduleEntry` that takes the Windows task XML and returns a `ScheduleEntry`.
- A `ScheduleEntry` type with a `reuse_existing` boolean attribute.
To integrate with your existing codebase, you will likely need to:
1. Replace `parse_task_xml_to_schedule_entry` with the actual helper/function already used in this test module to parse Windows task XML (for example, it might be a method on a scheduler instance or a module-level function).
2. Ensure the minimal XML structure in these tests matches what the parser expects; if your parser requires additional elements or specific namespaces, adjust the XML strings accordingly (follow patterns from other parsing tests in `tests/test_windows_system_scheduler.py`).
3. If you already have tests that parse tasks into `ScheduleEntry`, you may prefer to extend those existing tests by adding `reuse_existing` assertions instead of introducing a new helper; in that case, inline the XML and parsing calls following the existing conventions.
4. If `ScheduleEntry` lives in a different module, add/adjust the import at the top of the test file (e.g., `from app.core.service.system_scheduler.model import ScheduleEntry`) to make the `isinstance(entry, ScheduleEntry)` checks compile.
These adjustments will ensure the new tests correctly validate round-trip behavior of the `reuse_existing` flag between CLI arguments and `ScheduleEntry` parsing for Windows tasks.
</issue_to_address>
### Comment 6
<location path="tests/test_single_instance_restart.py" line_range="29-32" />
<code_context>
def test_shutdown_command_constants(self) -> None:
self.assertEqual(CMD_ACTIVATE, b"activate")
self.assertEqual(CMD_SHUTDOWN, b"shutdown")
+ self.assertEqual(CMD_RUN_CONFIG, b"run-config:")
self.assertEqual(RESP_OK, b"ok")
</code_context>
<issue_to_address>
**suggestion (testing):** Add tests for the new IPC `run-config` server/client behavior, not just the constant
The constant is covered, but the core IPC behavior isn’t. Please add tests that:
- Mock `_on_run_config` and verify that a valid JSON payload (`config_id`, `force_start`) causes `_consume_activation_request` to return the bytes from `_on_run_config`, and that invalid JSON or missing `config_id` yields `RESP_INVALID`.
- Cover `request_existing_instance_run` by mocking `QLocalSocket` to simulate connect/write/read success and different server responses (`RESP_ACCEPTED`, `RESP_BUSY`, `RESP_INVALID`, failures), including connection failures, partial writes, and timeouts, and assert the client always returns lowercased response bytes.
Suggested implementation:
```python
import json
from unittest import TestCase
from unittest.mock import patch
from app.utils.single_instance import (
CMD_ACTIVATE,
CMD_RUN_CONFIG,
CMD_SHUTDOWN,
RESP_OK,
RESP_INVALID,
RESP_ACCEPTED,
RESP_BUSY,
_consume_activation_request,
request_existing_instance_run,
process_matches_install_anchor,
)
class SingleInstanceRestartTests(TestCase):
def test_shutdown_command_constants(self) -> None:
self.assertEqual(CMD_ACTIVATE, b"activate")
self.assertEqual(CMD_SHUTDOWN, b"shutdown")
self.assertEqual(CMD_RUN_CONFIG, b"run-config:")
def test_consume_activation_request_run_config_valid_payload(self) -> None:
payload = {
"config_id": "abc123",
"force_start": True,
}
data = CMD_RUN_CONFIG + json.dumps(payload).encode("utf-8")
with patch("app.utils.single_instance._on_run_config", return_value=b"config-run") as on_run:
resp = _consume_activation_request(data)
on_run.assert_called_once_with(config_id="abc123", force_start=True)
self.assertEqual(resp, b"config-run")
def test_consume_activation_request_run_config_invalid_json(self) -> None:
# invalid JSON after the run-config prefix
data = CMD_RUN_CONFIG + b'{"config_id": '
resp = _consume_activation_request(data)
self.assertEqual(resp, RESP_INVALID)
def test_consume_activation_request_run_config_missing_config_id(self) -> None:
# valid JSON but missing required config_id
payload = {"force_start": False}
data = CMD_RUN_CONFIG + json.dumps(payload).encode("utf-8")
resp = _consume_activation_request(data)
self.assertEqual(resp, RESP_INVALID)
def test_request_existing_instance_run_success_responses(self) -> None:
# Simulate a server that accepts the request
with patch("app.utils.single_instance.QLocalSocket") as socket_cls:
socket = socket_cls.return_value
socket.connectToServer.return_value = None
socket.waitForConnected.return_value = True
socket.write.return_value = len(CMD_RUN_CONFIG)
socket.waitForBytesWritten.return_value = True
socket.readAll.return_value = RESP_ACCEPTED
resp = request_existing_instance_run(config_id="abc123", force_start=False)
socket.connectToServer.assert_called_once()
socket.write.assert_called_once()
socket.waitForBytesWritten.assert_called_once()
socket.readAll.assert_called_once()
self.assertEqual(resp, RESP_ACCEPTED.lower())
# Simulate a busy server
with patch("app.utils.single_instance.QLocalSocket") as socket_cls:
socket = socket_cls.return_value
socket.connectToServer.return_value = None
socket.waitForConnected.return_value = True
socket.write.return_value = len(CMD_RUN_CONFIG)
socket.waitForBytesWritten.return_value = True
socket.readAll.return_value = RESP_BUSY
resp = request_existing_instance_run(config_id="abc123", force_start=False)
self.assertEqual(resp, RESP_BUSY.lower())
# Simulate invalid request response
with patch("app.utils.single_instance.QLocalSocket") as socket_cls:
socket = socket_cls.return_value
socket.connectToServer.return_value = None
socket.waitForConnected.return_value = True
socket.write.return_value = len(CMD_RUN_CONFIG)
socket.waitForBytesWritten.return_value = True
socket.readAll.return_value = RESP_INVALID
resp = request_existing_instance_run(config_id="abc123", force_start=False)
self.assertEqual(resp, RESP_INVALID.lower())
def test_request_existing_instance_run_failures(self) -> None:
# Connection failure
with patch("app.utils.single_instance.QLocalSocket") as socket_cls:
socket = socket_cls.return_value
socket.connectToServer.return_value = None
socket.waitForConnected.return_value = False
resp = request_existing_instance_run(config_id="abc123", force_start=False)
self.assertEqual(resp, RESP_INVALID.lower())
# Partial write
with patch("app.utils.single_instance.QLocalSocket") as socket_cls:
socket = socket_cls.return_value
socket.connectToServer.return_value = None
socket.waitForConnected.return_value = True
socket.write.return_value = 0 # less than expected
socket.waitForBytesWritten.return_value = False
resp = request_existing_instance_run(config_id="abc123", force_start=False)
self.assertEqual(resp, RESP_INVALID.lower())
# Timeout / empty read
with patch("app.utils.single_instance.QLocalSocket") as socket_cls:
socket = socket_cls.return_value
socket.connectToServer.return_value = None
socket.waitForConnected.return_value = True
socket.write.return_value = len(CMD_RUN_CONFIG)
socket.waitForBytesWritten.return_value = True
socket.readAll.return_value = b""
resp = request_existing_instance_run(config_id="abc123", force_start=False)
self.assertEqual(resp, RESP_INVALID.lower())
```
- Ensure `app.utils.single_instance` actually exposes `_consume_activation_request`, `_on_run_config`, `request_existing_instance_run`, `RESP_INVALID`, `RESP_ACCEPTED`, `RESP_BUSY`, and `QLocalSocket` at the import paths used above; if they live in a submodule, adjust the imports and patch targets accordingly.
- The tests assume `_consume_activation_request` treats any JSON parsing error or missing `config_id` as `RESP_INVALID`, and that `request_existing_instance_run` normalizes all server responses to lowercase and returns `RESP_INVALID` on connection/write/read failures; align the implementation with these semantics if it currently differs.
- If the test module already defines a `TestCase` subclass, you may want to merge these tests into that existing class instead of introducing `SingleInstanceRestartTests`, or rename to match your existing naming convention.
</issue_to_address>
### Comment 7
<location path="tests/test_reuse_existing.py" line_range="39-48" />
<code_context>
+ )
+ self.assertFalse(entry.reuse_existing)
+
+ def test_schedule_entry_serializes_reuse_existing(self) -> None:
+ entry = ScheduleEntry(
+ entry_id="sched_demo",
+ config_id="cfg_demo",
+ name="Demo",
+ schedule_type="daily",
+ params={},
+ force_start=False,
+ enabled=True,
+ created_at=datetime.now(),
+ reuse_existing=True,
+ )
+ self.assertTrue(entry.to_dict()["reuse_existing"])
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Consider a round‑trip test for `ScheduleEntry.reuse_existing` through system schedulers
These tests cover `ScheduleEntry`’s serialization/deserialization of `reuse_existing` in isolation. Given the new Windows/Unix scheduler integrations, please also add an end‑to‑end test that builds scheduler artifacts from a `ScheduleEntry` with `reuse_existing=True`, parses them back via the scheduler APIs, and asserts that `reuse_existing` is preserved. This helps detect any mismatch between `ScheduleEntry` and scheduler handling of the new flag.
Suggested implementation:
```python
class ReuseExistingTests(unittest.TestCase):
def test_startup_cli_round_trip_preserves_reuse_existing(self) -> None:
# Build a ScheduleEntry with reuse_existing enabled
entry = ScheduleEntry(
entry_id="sched_demo_cli",
config_id="cfg_demo_cli",
name="Demo CLI",
schedule_type="daily",
params={},
force_start=False,
enabled=True,
created_at=datetime.now(),
reuse_existing=True,
)
# Convert ScheduleEntry into StartupOptions and build argv for the system scheduler
startup_options = StartupOptions(
config_id=entry.config_id,
force_start=entry.force_start,
reuse_existing=entry.reuse_existing,
)
argv = build_startup_argv(startup_options)
# Sanity check that the CLI flag for reuse_existing is present
self.assertIn(FLAG_REUSE_EXISTING, argv)
# Parse back via the scheduler CLI and assert reuse_existing is preserved
parsed_options = parse_startup_cli(argv)
self.assertTrue(parsed_options.reuse_existing)
def test_startup_cli_round_trip_without_reuse_existing(self) -> None:
# Build a ScheduleEntry with reuse_existing disabled
entry = ScheduleEntry(
entry_id="sched_demo_cli_disabled",
config_id="cfg_demo_cli_disabled",
name="Demo CLI Disabled",
schedule_type="daily",
params={},
force_start=False,
enabled=True,
created_at=datetime.now(),
reuse_existing=False,
)
startup_options = StartupOptions(
config_id=entry.config_id,
force_start=entry.force_start,
reuse_existing=entry.reuse_existing,
)
argv = build_startup_argv(startup_options)
# Sanity check that the CLI flag for reuse_existing is not present
self.assertNotIn(FLAG_REUSE_EXISTING, argv)
# Parse back via the scheduler CLI and assert reuse_existing remains False
parsed_options = parse_startup_cli(argv)
self.assertFalse(parsed_options.reuse_existing)
```
Depending on the actual `StartupOptions` signature and scheduler integration, you may need to:
1. Adjust the `StartupOptions` constructor arguments to match the real fields (e.g. if `config_id` or `force_start` are named differently, or additional mandatory parameters exist).
2. Ensure that `build_startup_argv` and `parse_startup_cli` are the correct APIs for round‑tripping through both Windows and Unix schedulers; if there are OS‑specific wrappers, you might want mirrored tests for each wrapper that call into these functions.
3. If there is already a helper that maps `ScheduleEntry` to `StartupOptions`, replace the manual `StartupOptions(...)` construction with that helper to stay consistent with the production pipeline.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
overflow65537
requested changes
Aug 11, 2026
overflow65537
requested changes
Aug 12, 2026
overflow65537
approved these changes
Aug 12, 2026
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
2 participants
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.
fix #233
已在 MAA_bbb 项目中本地测试复用有效。
Summary by Sourcery
Add support for reusing an already-running MFW instance to execute a specified configuration, including CLI, IPC, scheduler, and UI integration.
New Features:
Enhancements:
Tests: