Continue past unreadable encrypted relay bundles

SECURITY.md

@@ -150,8 +150,8 @@ repository or its release artifacts.
   use reverse proxy for multi-process/distributed
 - Encryption at rest is opt-in for the local memory DB (SQLCipher via
   `ENGRAPHIS_DB_KEY`). Protect customer credential state and backups separately.
-- Managed relay bundles are HTTPS-protected in transit but remain plaintext at rest until
-  client-side end-to-end encryption ships. `secret` memories are never uploaded.
+- Cloud Sync requires an authorized device to hold its workspace encryption key. The relay cannot
+  recover a lost key or decrypt its ciphertext; `secret` memories are never uploaded.
 - Per-token scope/tenant authorization is partial: isolate distinct tenants by running
   one instance each
 - Legacy v1 REST server/dashboard is a compatibility surface; prefer v2/MCP path

docs/SYNC.md

@@ -128,9 +128,10 @@ outside the authorized workspace merely by changing bundle fields.
 
 - Local-only installations send no memory content to Engraphis. **Cloud Sync encrypts eligible
   shared-workspace changes end-to-end before they leave this device. Engraphis Cloud cannot read
-  their contents; secret and session-scoped memories stay local.** Managed compute is a separate,
-  opt-in service: it sends a readable, bounded snapshot over TLS because Engraphis Cloud must
-  process that snapshot to produce results.
+  their contents; secret and session-scoped memories stay local.** Managed compute is separate:
+  connecting an installation to Engraphis Cloud accepts its terms and enables it by default;
+  operators may opt out with `ENGRAPHIS_MANAGED_COMPUTE_CONSENT=0`. It sends a readable, bounded
+  snapshot over TLS because Engraphis Cloud must process that snapshot to produce results.
 - Treat cloud session and refresh files as credentials; keep their directory owner-only.
 - `secret` memories are excluded from managed uploads. Managed compute also rejects secret rows
   server-side.

engraphis/backends/sync_relay.py

@@ -415,22 +415,42 @@ def push(self, name: str, data: bytes) -> None:
         self.relay.push(stored_name, SYNC_E2EE_MAGIC + nonce + ciphertext)
 
     def pull(self) -> Iterable[Tuple[str, bytes]]:
+        """Yield every authentic bundle and flag an incomplete encrypted round.
+
+        A Cloud Sync workspace may contain bundles written before E2EE existed, or a
+        relay object may have been damaged.  Neither is eligible for plaintext
+        fallback, but neither may prevent a later authenticated peer bundle from
+        being applied.  Match ``RelayTransport.pull``: fail closed for each bad
+        object, yield every valid one, then raise one sanitized error so
+        ``SyncEngine`` reports an incomplete round rather than a false success.
+        """
+        skipped = 0
         for name, data in self.relay.pull():
-            safe = _safe_bundle_name(name)
-            if not safe or not isinstance(data, (bytes, bytearray)):
-                raise RelayError("relay returned an invalid encrypted bundle")
-            raw = bytes(data)
-            if not raw.startswith(SYNC_E2EE_MAGIC):
-                raise RelayError("relay bundle requires end-to-end encryption")
-            payload = raw[len(SYNC_E2EE_MAGIC):]
-            if len(payload) < SYNC_E2EE_NONCE_BYTES + SYNC_E2EE_TAG_BYTES:
-                raise RelayError("bundle could not be authenticated")
-            nonce, ciphertext = payload[:SYNC_E2EE_NONCE_BYTES], payload[SYNC_E2EE_NONCE_BYTES:]
             try:
+                safe = _safe_bundle_name(name)
+                if not safe or not isinstance(data, (bytes, bytearray)):
+                    raise RelayError("relay returned an invalid encrypted bundle")
+                raw = bytes(data)
+                if not raw.startswith(SYNC_E2EE_MAGIC):
+                    raise RelayError("relay bundle requires end-to-end encryption")
+                payload = raw[len(SYNC_E2EE_MAGIC):]
+                if len(payload) < SYNC_E2EE_NONCE_BYTES + SYNC_E2EE_TAG_BYTES:
+                    raise RelayError("bundle could not be authenticated")
+                nonce = payload[:SYNC_E2EE_NONCE_BYTES]
+                ciphertext = payload[SYNC_E2EE_NONCE_BYTES:]
                 plaintext = self._cipher.decrypt(nonce, ciphertext, self._aad(safe))
             except self._invalid_tag:
-                raise RelayError("bundle could not be authenticated") from None
+                skipped += 1
+                continue
+            except RelayError:
+                skipped += 1
+                continue
             yield safe, plaintext
+        if skipped:
+            raise RelayError(
+                "encrypted relay skipped %d unreadable bundle%s this round"
+                % (skipped, "" if skipped == 1 else "s")
+            )
 
     def list_names(self) -> List[str]:
         return self.relay.list_names()

tests/test_sync_e2ee.py

@@ -62,11 +62,11 @@ def test_cloud_sync_rejects_tampered_or_plaintext_bundle():
     name, stored = next(iter(relay.bundles.items()))
     relay.bundles[name] = stored[:-1] + bytes([stored[-1] ^ 1])
 
-    with pytest.raises(RelayError, match="could not be authenticated"):
+    with pytest.raises(RelayError, match="unreadable bundle"):
         list(receiver.pull())
 
     relay.bundles[name] = b'{"legacy":"plaintext"}'
-    with pytest.raises(RelayError, match="requires end-to-end encryption"):
+    with pytest.raises(RelayError, match="unreadable bundle"):
         list(receiver.pull())
 
 
@@ -76,16 +76,59 @@ def test_cloud_sync_rejects_a_bundle_from_another_key_or_workspace():
     wrong_key = _transport(relay, 4)
     sender.push("bundle-dev_a.json", b"private content")
 
-    with pytest.raises(RelayError, match="could not be authenticated"):
+    with pytest.raises(RelayError, match="unreadable bundle"):
         list(wrong_key.pull())
 
     wrong_workspace = _transport(_MemoryRelay("other"), 3)
     name, stored = next(iter(relay.bundles.items()))
     wrong_workspace.relay.bundles[name] = stored
-    with pytest.raises(RelayError, match="could not be authenticated"):
+    with pytest.raises(RelayError, match="unreadable bundle"):
         list(wrong_workspace.pull())
 
 
+@pytest.mark.parametrize("bad_kind", ["legacy", "tampered"], ids=["legacy", "tampered"])
+def test_sync_engine_applies_later_encrypted_bundle_after_unreadable_relay_object(bad_kind):
+    """Legacy/corrupt relay objects cannot starve later authenticated peers."""
+    relay = _MemoryRelay()
+    key = bytes(range(32))
+    sender = MemoryEngine.create(":memory:")
+    receiver = MemoryEngine.create(":memory:")
+    sender_workspace = sender.store.get_or_create_workspace("acme")
+    receiver_workspace = receiver.store.get_or_create_workspace("acme")
+    sender.remember("peer fact survives a bad relay object", workspace_id=sender_workspace,
+                    scope=Scope.WORKSPACE)
+    sender_sync = SyncEngine(sender.store, embedder=sender.embedder, vector_index=sender.index)
+    receiver_sync = SyncEngine(
+        receiver.store, embedder=receiver.embedder, vector_index=receiver.index
+    )
+
+    # The bad object is deliberately inserted before the sender's encrypted bundle.
+    if bad_kind == "legacy":
+        relay.bundles["bundle-legacy.json"] = b'{"legacy":"plaintext"}'
+    else:
+        corrupt_writer = EncryptedRelayTransport(relay, key)
+        corrupt_writer.push("bundle-corrupt.json", b"original authenticated ciphertext")
+        corrupt_name, ciphertext = next(iter(relay.bundles.items()))
+        relay.bundles[corrupt_name] = ciphertext[:-1] + bytes([ciphertext[-1] ^ 1])
+    sender_sync.sync(EncryptedRelayTransport(relay, key), sender_workspace)
+
+    report = receiver_sync.sync(
+        EncryptedRelayTransport(relay, key), receiver_workspace, push=False
+    )
+
+    contents = {
+        memory.content
+        for memory in receiver.store.list_memories(SearchFilter(workspace_id=receiver_workspace))
+    }
+    assert contents == {"peer fact survives a bad relay object"}
+    assert report["totals"]["added"] == 1
+    assert report["peers_applied"] == 1
+    assert report["complete"] is False
+    assert report["errors"] == [
+        {"bundle": "?", "error": "transport failure", "error_type": "RelayError"}
+    ]
+
+
 def test_sync_engine_converges_through_encrypted_relay_without_plaintext_storage():
     relay = _MemoryRelay()
     key = bytes(range(32))