fix: roll back failed resource imports completely

docs/REWORK_EXECUTION.md

@@ -81,6 +81,37 @@ changes have not been performed by these local changes. Ordinary local engineeri
 and verification are already authorized; missing hardware and independent evidence
 are execution constraints, not reasons to claim completion or invent results.
 
+## Resource-import transaction follow-up
+
+Legacy folder/upload imports now isolate each file, including its chunks, FTS,
+canonical vectors, transactional index rows and receipts, before returning a
+recoverable per-file error. An optional fact-derivation failure rolls back its
+complete derived prefix while retaining the successful source import. Unexpected
+failures and final commit failures still abort the service-owned batch; a caller's
+preceding transaction remains caller-owned.
+
+The additional counterexamples at `dc4382d1` included an FTS failure leaving three
+canonical records despite a two-import/one-error report, and a second-chunk
+embedding failure leaving an unreported first chunk. Fourteen new regression
+cases reproduced these integrity failures against that unchanged dependency
+checkout. Review also reproduced failed savepoint rollback/release being treated
+as a recoverable file error. Savepoint settlement now raises `SavepointError` and
+aborts the enclosing operation, including optional conflict repair.
+
+This follow-up changes no schema, public signature, response field, ranking or
+approval rule. There is no data migration. Reverting the patch restores the prior
+partial-write risk; it does not reconcile fragments left by earlier imports.
+Do not delete suspected fragments automatically: retain provenance and inspect
+the applicable import report before governed correction or erasure.
+
+Preparation remains the next dependency. Folder enumeration, resource parsing,
+chunking, embedding and explicitly enabled derivation still occur inside the
+legacy batch writer. Move them through immutable prepared commands, with current
+workspace/embedding validation and explicit post-commit index publication, in a
+separate change. Preserve the existing caller-owned separate-index rejection until
+that publication contract exists. These integrity tests establish no throughput,
+100k capacity or production recovery claim.
+
 ## Schema 17 to 18 and recovery
 
 Schema 18 adds memory-command receipts/source claims, portable browsing revisions

engraphis/core/engine.py

@@ -74,6 +74,7 @@
 )
 from engraphis.core.secrets import redact_secrets as _redact_secrets, reject_secrets
 from engraphis.core.store import (
+    SavepointError,
     Store,
     _is_memory_database_path,
     memory_matches_filter,
@@ -1758,6 +1759,10 @@ def _resolve_and_store(self, content: str, *, text: str, vec: Optional[np.ndarra
                         "UPDATE memories SET confidence=MIN(confidence, ?) WHERE id=?",
                         (round(CONFLICT_CONFIDENCE_FACTOR, 4), conflicted_with),
                     )
+            except SavepointError:
+                # A failed rollback/release cannot be treated as an optional repair
+                # failure: it may leave partial writes in the enclosing transaction.
+                raise
             except Exception as exc:  # noqa: BLE001 - derived repair must not discard the memory
                 self._warn_redacted_failure("conflict repair", exc)
         out: dict[str, object]

engraphis/core/store.py

@@ -79,6 +79,10 @@
 )
 
 
+class SavepointError(RuntimeError):
+    """A sub-operation could not settle; its enclosing transaction must abort."""
+
+
 # Rows materialized per locked batch when streaming the vector table (see iter_vectors).
 VECTOR_SCAN_BATCH = 2000
 _STARTUP_GRAPH_TRANSFORMS = {"edge_supports": 1, "live_edge_deduplication": 1}
@@ -3791,17 +3795,23 @@ def opener(*, timeout):
 
     @contextmanager
     def write_savepoint(self):
-        """Isolate a best-effort sub-operation inside an authoritative transaction."""
+        """Isolate a sub-operation; settlement failures must abort its outer owner."""
         name = f"engraphis_optional_{threading.get_ident()}_{time.monotonic_ns()}"
         self.conn.execute(f"SAVEPOINT {name}")
         try:
             yield
         except BaseException:
-            self.conn.execute(f"ROLLBACK TO SAVEPOINT {name}")
-            self.conn.execute(f"RELEASE SAVEPOINT {name}")
+            try:
+                self.conn.execute(f"ROLLBACK TO SAVEPOINT {name}")
+                self.conn.execute(f"RELEASE SAVEPOINT {name}")
+            except Exception as exc:
+                raise SavepointError("could not roll back the write savepoint") from exc
             raise
         else:
-            self.conn.execute(f"RELEASE SAVEPOINT {name}")
+            try:
+                self.conn.execute(f"RELEASE SAVEPOINT {name}")
+            except Exception as exc:
+                raise SavepointError("could not release the write savepoint") from exc
 
     # ── local source-import manifest ─────────────────────────────────────────
     def _authorize_source_workspace_id(self, workspace_id: str) -> str:

engraphis/service.py

@@ -2300,43 +2300,53 @@ def _import_one(self, name: str, content: str, *, ws: str, mt: MemoryType,
         )
         chunks = chunker.extract(content) if chunker is not None else None
         try:
-            if chunks:
-                total = len(chunks)
-                first: Optional[dict] = None
-                for i, fact in enumerate(chunks):
-                    title = (
-                        fact.title or resource_title
-                        or _title_from_content(fact.content, fallback)
-                    )
-                    r = self.remember(
-                        fact.content, workspace=ws,
-                        mtype=(fact.mtype.value if fact.mtype else mt.value),
-                        scope="workspace", title=title[:MAX_TITLE_CHARS],
-                        source="import", trusted=False, kind=kind,
-                        keywords=fact.keywords,
-                        metadata={**(extra_provenance or {}), "import_file": name,
-                                  "chunk": {"index": i, "of": total,
-                                            "heading": (fact.title or "")[:200]}},
-                        resolve_conflicts=False,
-                    )
-                    first = first or r
-                return {"file": name, "id": first["id"], "op": first["op"], "chunks": total}
-            title = resource_title or _title_from_content(content, fallback=fallback)
-            r = self.remember(
-                content, workspace=ws, mtype=mt.value, scope="workspace",
-                title=title[:MAX_TITLE_CHARS], source="import", trusted=False, kind=kind,
-                metadata={**(extra_provenance or {}), "import_file": name},
-            )
-            return {"file": name, "id": r["id"], "op": r["op"]}
+            # Expected per-file errors are caught below the batch boundary. Give the
+            # complete file (including all chunks and receipts) its own rollback scope
+            # before converting a write failure into a successful batch response.
+            with self.store.write_savepoint():
+                return self._store_import_chunks(
+                    name, content, ws=ws, mt=mt, kind=kind, chunks=chunks,
+                    fallback=fallback, extra_provenance=extra_provenance,
+                    resource_title=resource_title,
+                )
         except (ValidationError, ValueError, sqlite3.Error, RecursionError,
                 MemoryError) as exc:
-            # One bad file must degrade to a per-file error, not void the whole batch
-            # (e.g. sqlite3.OperationalError "database is locked" from a concurrent
-            # CLI/MCP writer, embedder ValueError, or a crafted deep-nested JSON upload
-            # blowing json.loads recursion).
             logger.info("uploaded resource import rejected (%s)", type(exc).__name__)
             return {"file": name, "error": "resource could not be imported"}
 
+    def _store_import_chunks(self, name: str, content: str, *, ws: str, mt: MemoryType,
+                             kind: str, chunks, fallback: str,
+                             extra_provenance: Optional[dict], resource_title: str) -> dict:
+        """Apply a resource inside its caller's per-file savepoint."""
+        if chunks:
+            total = len(chunks)
+            first: Optional[dict] = None
+            for i, fact in enumerate(chunks):
+                title = (
+                    fact.title or resource_title
+                    or _title_from_content(fact.content, fallback)
+                )
+                r = self.remember(
+                    fact.content, workspace=ws,
+                    mtype=(fact.mtype.value if fact.mtype else mt.value),
+                    scope="workspace", title=title[:MAX_TITLE_CHARS],
+                    source="import", trusted=False, kind=kind,
+                    keywords=fact.keywords,
+                    metadata={**(extra_provenance or {}), "import_file": name,
+                              "chunk": {"index": i, "of": total,
+                                        "heading": (fact.title or "")[:200]}},
+                    resolve_conflicts=False,
+                )
+                first = first or r
+            return {"file": name, "id": first["id"], "op": first["op"], "chunks": total}
+        title = resource_title or _title_from_content(content, fallback=fallback)
+        r = self.remember(
+            content, workspace=ws, mtype=mt.value, scope="workspace",
+            title=title[:MAX_TITLE_CHARS], source="import", trusted=False, kind=kind,
+            metadata={**(extra_provenance or {}), "import_file": name},
+        )
+        return {"file": name, "id": r["id"], "op": r["op"]}
+
     def _derive_import_facts(self, content: str, *, ws: str, mt: MemoryType,
                              resource_name: str, resource_kind: str,
                              resource_meta: dict) -> tuple[int, str]:
@@ -2357,17 +2367,20 @@ def _derive_import_facts(self, content: str, *, ws: str, mt: MemoryType,
 
         created = 0
         extracted = False
-        for chunk in inputs:
-            derived = self.ingest(
-                chunk, workspace=ws, mtype=mt.value, scope="workspace",
-                metadata={"derived_from_resource": resource_name, **resource_meta},
-                source="resource_extractor", trusted=False,
-                kind=f"{resource_kind}_facts",
-            )
-            extracted = extracted or bool(derived["extracted"])
-            created += sum(
-                1 for fact in derived["facts"] if fact.get("op") != "noop"
-            )
+        # This optional pass may fail without failing the imported source. Its count
+        # must describe committed facts, so discard the entire derived prefix first.
+        with self.store.write_savepoint():
+            for chunk in inputs:
+                derived = self.ingest(
+                    chunk, workspace=ws, mtype=mt.value, scope="workspace",
+                    metadata={"derived_from_resource": resource_name, **resource_meta},
+                    source="resource_extractor", trusted=False,
+                    kind=f"{resource_kind}_facts",
+                )
+                extracted = extracted or bool(derived["extracted"])
+                created += sum(
+                    1 for fact in derived["facts"] if fact.get("op") != "noop"
+                )
         if not extracted or created == 0:
             return created, "configured extractor produced no new discrete facts"
         return created, ""