Update services/secondary_brain.py

services/secondary_brain.py

@@ -2,8 +2,32 @@
 import os
 import json
 import datetime
+import tempfile
 from pathlib import Path
 
+
+def _safe_write(filepath: str, content: str):
+    """
+    Bucket-safe atomic write. Writes to a temp file in the SAME directory,
+    then renames over the target. Within one directory on a FUSE-mounted
+    bucket, rename is atomic. Direct open('w') is NOT safe — a crash
+    mid-write silently zeroes the file on object storage.
+    """
+    dirpath = os.path.dirname(os.path.abspath(filepath))
+    os.makedirs(dirpath, exist_ok=True)
+    fd, tmp_path = tempfile.mkstemp(prefix=".tmp_sb_", dir=dirpath)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            f.write(content)
+            f.flush()
+        os.replace(tmp_path, filepath)
+    except Exception:
+        try:
+            os.remove(tmp_path)
+        except FileNotFoundError:
+            pass
+        raise
+
 # Tags that indicate procedural/how-to content worth extracting separately
 PROCEDURAL_TAGS = {
     "algorithm", "method", "formula", "process", "technique",
@@ -148,21 +172,23 @@ class DomainLayer:
             condensed = json.loads(cleaned)
 
             if isinstance(condensed, list) and len(condensed) > 0:
-                # Overwrite legend with condensed entries
-                with open(self.legend_path, "w", encoding="utf-8") as f:
-                    for item in condensed:
-                        item["domain"]    = self.domain_name
-                        item["timestamp"] = datetime.datetime.now().isoformat()
-                        f.write(json.dumps(item, ensure_ascii=False) + "\n")
+                # Build condensed legend content
+                now_iso = datetime.datetime.now().isoformat()
+                legend_lines = []
+                for item in condensed:
+                    item["domain"]    = self.domain_name
+                    item["timestamp"] = now_iso
+                    legend_lines.append(json.dumps(item, ensure_ascii=False))
+                # Atomic write — safe on bucket/FUSE storage
+                _safe_write(self.legend_path, "\n".join(legend_lines) + "\n")
 
                 print(f"[SecondaryBrain] '{self.domain_name}' condensed from {count} → {len(condensed)} entries.", flush=True)
 
-                # Also update condensed_ontology.txt with a summary
-                ontology_lines = [f"Domain: {self.domain_name}", f"Condensed at: {datetime.datetime.now().isoformat()}", ""]
+                # Also update condensed_ontology.txt with a summary (atomic)
+                ontology_lines = [f"Domain: {self.domain_name}", f"Condensed at: {now_iso}", ""]
                 for item in condensed:
                     ontology_lines.append(f"  [{item.get('sqt','')}] {item.get('summary','')}")
-                with open(self.ontology_path, "w", encoding="utf-8") as f:
-                    f.write("\n".join(ontology_lines))
+                _safe_write(self.ontology_path, "\n".join(ontology_lines))
 
                 return True
 
@@ -310,8 +336,8 @@ class SecondaryBrain:
                 "concept_count": concept_count,
                 "last_active":   last_active
             }
-        with open(self.index_path, "w", encoding="utf-8") as f:
-            json.dump(index, f, indent=2, ensure_ascii=False)
+        # Atomic write — safe on bucket/FUSE storage
+        _safe_write(self.index_path, json.dumps(index, indent=2, ensure_ascii=False))
 
     def _get_or_create_domain(self, domain_name: str) -> DomainLayer:
         """