fix: write config files atomically

[zouyonghe]

This was generated by AI during triage.

Summary

• write AstrBot config files through a same-directory temporary file followed by atomic replace
• fsync the temporary file before replacement so interrupted writes keep either the old complete config or the new complete config
• use the same atomic path when creating the initial config file
• add a regression test proving a failed write leaves the existing config intact and cleans temporary files

Fixes #8298.

Verification

• uv run pytest tests/unit/test_config.py -q
• uv run ruff check astrbot/core/config/astrbot_config.py tests/unit/test_config.py
• uv run ruff format --check astrbot/core/config/astrbot_config.py tests/unit/test_config.py
• git diff --check

Summary by Sourcery

Write AstrBot configuration files atomically to avoid corrupting existing configs on write failures.

Bug Fixes:

• Ensure saving configuration writes through a temporary file and atomic replace so interrupted writes do not corrupt the existing config file.

Enhancements:

• Initialize new config files using the same atomic save path and allow configuring JSON indentation when saving configs.

Tests:

• Add a regression test verifying that failed config writes leave the original file intact and do not leave temporary files behind.

[sourcery-ai]

Hey - I've found 3 issues, and left some high level feedback:

• The new atomic write logic always unlinks the temp file on any exception, which also removes partially written files used by the regression test; consider narrowing the exception handling or adjusting the test so it doesn’t depend on the temp file being present after a failed write.
• The tempfile prefix format f".{os.path.basename(self.config_path)}." produces hidden files that may conflict with other tools; consider using a more specific prefix (e.g. including astrbot_config or a UUID) to reduce the chance of name clashes in shared directories.
• The directory scan assertion in test_save_config_preserves_existing_file_when_write_fails assumes the config directory contains only the config file, which may be brittle in real or parallel test environments; narrowing the scan to only match the temp-file prefix or using a dedicated subdirectory would make this more robust.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The new atomic write logic always unlinks the temp file on any exception, which also removes partially written files used by the regression test; consider narrowing the exception handling or adjusting the test so it doesn’t depend on the temp file being present after a failed write.
- The tempfile prefix format `f".{os.path.basename(self.config_path)}."` produces hidden files that may conflict with other tools; consider using a more specific prefix (e.g. including `astrbot_config` or a UUID) to reduce the chance of name clashes in shared directories.
- The directory scan assertion in `test_save_config_preserves_existing_file_when_write_fails` assumes the config directory contains only the config file, which may be brittle in real or parallel test environments; narrowing the scan to only match the temp-file prefix or using a dedicated subdirectory would make this more robust.

## Individual Comments

### Comment 1
<location path="astrbot/core/config/astrbot_config.py" line_range="222-223" />
<code_context>

         如果传入 replace_config，则将配置替换为 replace_config
         """
         if replace_config:
             self.update(replace_config)
-        with open(self.config_path, "w", encoding="utf-8-sig") as f:
-            json.dump(self, f, indent=2, ensure_ascii=False)
</code_context>
<issue_to_address>
**issue:** Using `if replace_config` will ignore valid but falsy configs such as an empty dict.

This means `self.update` is skipped when `replace_config` is an empty dict (or other falsy mapping), so you can’t clear the config via `save_config`. Use `if replace_config is not None:` to allow empty configs while still treating `None` as “no replacement provided.”
</issue_to_address>

### Comment 2
<location path="tests/unit/test_config.py" line_range="556-562" />
<code_context>
+
+        with open(temp_config_path, encoding="utf-8-sig") as f:
+            assert f.read() == original_content
+        assert [
+            entry.name
+            for entry in os.scandir(os.path.dirname(temp_config_path))
+            if entry.name != os.path.basename(temp_config_path)
+        ] == []
+
     def test_modification_persists_after_reload(
</code_context>
<issue_to_address>
**suggestion (testing):** Temp-file cleanup assertion may be too strict and brittle for directories with other legitimate files

This assertion currently requires the directory to contain only the config file, which makes the test brittle if other legitimate files (e.g., from fixtures or future tests) are present. Instead, assert specifically that no temp files created by `save_config` remain, e.g. by filtering on the temp-file naming pattern:

```python
assert not [
    e.name
    for e in os.scandir(os.path.dirname(temp_config_path))
    if e.name.startswith(f".{os.path.basename(temp_config_path)}.")
    and e.name.endswith(".tmp")
]
```

```suggestion
        with open(temp_config_path, encoding="utf-8-sig") as f:
            assert f.read() == original_content
        assert not [
            entry.name
            for entry in os.scandir(os.path.dirname(temp_config_path))
            if entry.name.startswith(f".{os.path.basename(temp_config_path)}.")
            and entry.name.endswith(".tmp")
        ]
```
</issue_to_address>

### Comment 3
<location path="astrbot/core/config/astrbot_config.py" line_range="215" />
<code_context>
         return has_new

-    def save_config(self, replace_config: dict | None = None) -> None:
+    def save_config(
+        self, replace_config: dict | None = None, *, indent: int = 2
+    ) -> None:
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the atomic JSON write logic into a small helper so `save_config` stays a simple wrapper that focuses only on updating and delegating the write.

You can keep the atomic write + indentation behavior while pushing the low-level details out of `save_config` into a small helper. That keeps `save_config` at its old “orchestrator” level and makes the IO/cleanup logic easier to reason about and reuse.

For example:

```python
import json
import os
import tempfile

def _atomic_write_json(path: str, data: dict, *, indent: int) -> None:
    directory = os.path.dirname(os.path.abspath(path)) or "."
    fd, temp_path = tempfile.mkstemp(
        dir=directory,
        prefix=f".{os.path.basename(path)}.",
        suffix=".tmp",
        text=True,
    )
    try:
        with os.fdopen(fd, "w", encoding="utf-8-sig") as f:
            json.dump(data, f, indent=indent, ensure_ascii=False)
            f.flush()
            os.fsync(f.fileno())
        os.replace(temp_path, path)
    except Exception:
        try:
            os.unlink(temp_path)
        except FileNotFoundError:
            pass
        raise
```

Then `save_config` becomes a thin wrapper again:

```python
def save_config(
    self, replace_config: dict | None = None, *, indent: int = 2
) -> None:
    if replace_config:
        self.update(replace_config)
    _atomic_write_json(self.config_path, self, indent=indent)
```

This keeps all current behavior (atomic-ish write, custom indent, fsync, cleanup on error) but:
- Removes nested IO/cleanup details from `save_config`
- Makes the “save config” responsibility clearer and easier to scan
- Gives you a reusable atomic JSON writer if you need it elsewhere.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

issue: Using if replace_config will ignore valid but falsy configs such as an empty dict.

This means self.update is skipped when replace_config is an empty dict (or other falsy mapping), so you can’t clear the config via save_config. Use if replace_config is not None: to allow empty configs while still treating None as “no replacement provided.”

[sourcery-ai]

suggestion (testing): Temp-file cleanup assertion may be too strict and brittle for directories with other legitimate files

This assertion currently requires the directory to contain only the config file, which makes the test brittle if other legitimate files (e.g., from fixtures or future tests) are present. Instead, assert specifically that no temp files created by save_config remain, e.g. by filtering on the temp-file naming pattern:

assert not [
    e.name
    for e in os.scandir(os.path.dirname(temp_config_path))
    if e.name.startswith(f".{os.path.basename(temp_config_path)}.")
    and e.name.endswith(".tmp")
]

Suggested change

	with open(temp_config_path, encoding="utf-8-sig") as f:
	assert f.read() == original_content
	assert [
	entry.name
	for entry in os.scandir(os.path.dirname(temp_config_path))
	if entry.name != os.path.basename(temp_config_path)
	] == []
	with open(temp_config_path, encoding="utf-8-sig") as f:
	assert f.read() == original_content
	assert not [
	entry.name
	for entry in os.scandir(os.path.dirname(temp_config_path))
	if entry.name.startswith(f".{os.path.basename(temp_config_path)}.")
	and entry.name.endswith(".tmp")
	]

[sourcery-ai]

issue (complexity): Consider extracting the atomic JSON write logic into a small helper so save_config stays a simple wrapper that focuses only on updating and delegating the write.

You can keep the atomic write + indentation behavior while pushing the low-level details out of save_config into a small helper. That keeps save_config at its old “orchestrator” level and makes the IO/cleanup logic easier to reason about and reuse.

For example:

import json
import os
import tempfile

def _atomic_write_json(path: str, data: dict, *, indent: int) -> None:
    directory = os.path.dirname(os.path.abspath(path)) or "."
    fd, temp_path = tempfile.mkstemp(
        dir=directory,
        prefix=f".{os.path.basename(path)}.",
        suffix=".tmp",
        text=True,
    )
    try:
        with os.fdopen(fd, "w", encoding="utf-8-sig") as f:
            json.dump(data, f, indent=indent, ensure_ascii=False)
            f.flush()
            os.fsync(f.fileno())
        os.replace(temp_path, path)
    except Exception:
        try:
            os.unlink(temp_path)
        except FileNotFoundError:
            pass
        raise

Then save_config becomes a thin wrapper again:

def save_config(
    self, replace_config: dict | None = None, *, indent: int = 2
) -> None:
    if replace_config:
        self.update(replace_config)
    _atomic_write_json(self.config_path, self, indent=indent)

This keeps all current behavior (atomic-ish write, custom indent, fsync, cleanup on error) but:

• Removes nested IO/cleanup details from save_config
• Makes the “save config” responsibility clearer and easier to scan
• Gives you a reusable atomic JSON writer if you need it elsewhere.

[gemini-code-assist]

Code Review

This pull request implements atomic configuration saving in AstrBotConfig by writing to a temporary file and replacing the original file, preventing corruption during write failures. A corresponding unit test was also added. The review feedback correctly identifies that using text=True in tempfile.mkstemp and then wrapping it with os.fdopen using a specific encoding can cause issues on Windows, and suggests removing text=True to open the file descriptor in binary mode.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Using text=True in tempfile.mkstemp opens the file descriptor in text mode (using O_TEXT on Windows). When you subsequently wrap this file descriptor with os.fdopen(fd, "w", encoding="utf-8-sig"), Python's io.TextIOWrapper expects a binary file descriptor to correctly handle the specified encoding and newline translation. Wrapping a text-mode file descriptor can lead to encoding conflicts, double-translation of line endings, or write corruption on Windows. Removing text=True (or setting it to False) ensures the file descriptor is opened in binary mode, allowing os.fdopen to portably and safely handle the UTF-8-SIG encoding.

        fd, temp_path = tempfile.mkstemp(
            dir=directory,
            prefix=f".{os.path.basename(self.config_path)}.",
            suffix=".tmp",
        )