[SleepySuke]

Description

This PR adds openGauss backend support to MemOS as part of internal feature development, targeting dev-v2.0.36.

It integrates AGE graph storage, native vector retrieval, and BM25 full-text search with the existing memory components, API, and Scheduler.

Related Issue / Task: N/A — internally planned feature development with no associated issue or task number.

Main changes

• Add openGauss configuration validation, factory registration, environment settings, and connection pooling.
• Implement scoped node and relationship operations, graph traversal, metadata filtering, vector search, BM25 search, and SQL LIKE lookup.
• Support memory counts, capacity management, snapshot export, atomic import, and scoped clearing.
• Add optional Dashboard range selection through all_cubes and exclude_cube_id, with documentation and OpenAPI schema updates.
• Improve resource ownership and shutdown ordering across API, Cube, Tree memory, and Scheduler components.
• Preserve Cube identity during search and distinct identities for working and long-term memory pairs.
• Propagate required storage failures through search and Feedback instead of treating failures as successful empty results.

Dependencies

• openGauss Enterprise with AGE and native DataVec/BM25 capabilities.
• psycopg2-binary >=2.9.13,<3.0.0, included in the optional tree-mem and all extras.

Enable the backend with GRAPH_DB_BACKEND=opengauss and configure the corresponding OPENGAUSS_* connection settings. The database and required extensions must be provisioned before use.

Behavior changes and limitations

• Empty Dashboard categories return empty lists instead of synthetic empty Cube buckets.
• Non-openGauss backends reject explicit non-null values for the new Dashboard range fields.
• Required storage errors propagate to callers. Multi-step Feedback updates are not a single transaction.
• Database-level semantic deduplication, conflict detection, and node merging remain unsupported.
• Full service E2E, Java client integration, real-model validation, and production-scale performance testing remain follow-up work.

Type of change

• New feature
• Bug fix
• Refactor
• Documentation update

How Has This Been Tested?

Latest recorded local validation:

• Full repository suite with real openGauss integration enabled: 2771 passed, 9 skipped, 45 warnings.
• Unit and integration tests cover scoped storage, retrieval, rollback, backup/restore, Dashboard responses, memory identities, Feedback failures, and concurrent shutdown.
• A separate local HTTP smoke test passed with real openGauss and controlled model assembly, covering Dashboard results, graceful shutdown, and database session cleanup.
• Ruff checks and formatting passed.

Integration tests use openGauss 7.0.0 with AGE 1.0.0. Test fixtures provision disposable databases and roles; application operations run with an ordinary database role.

To reproduce, configure MEMOS_OPENGAUSS_TEST_CONFIG with a JSON OpenGaussGraphDBConfig for a test administrator, then run:

poetry install --extras all --with dev --with test
make format
make test

OpenAPI was exported using the repository's component fixtures. Standard make openapi remains blocked in the local environment by missing InternetRetriever model configuration.

These results describe local validation. Hosted CI and full service E2E are not claimed as complete.

• Unit Test
• Test Script Or Test Steps
• Pipeline Automated API Test

Checklist

• I have performed a self-review of the final PR diff
• I have commented code in hard-to-understand areas
• I have added tests covering the new functionality and fixes
• I have included backend setup and behavior documentation in this PR
• I have assigned or mentioned the reviewer

Issue/task linkage: N/A — no associated issue or internal task number.

Reviewer Checklist

• Confirmed the implementation scope and behavior changes
• Made sure required checks passed
• Reviewed the provided tests and validation results

[Memtensor-AI]

🤖 Open Code Review

Target: PR #2564
Task: ff99fb0b2d1fab95
Base: dev-v2.0.36
Head: feat/opengauss-backend
Head SHA: ad10805219524785580074010aac852d8249b76e

🔍 OpenCodeReview found 15 issue(s) in this PR.

⚠️ Review warnings (2)

• bounded analysis finalized with explicit task_done after reviewing all group files
• bounded analysis finalized with explicit task_done after reviewing all group files

---

1. docs/openapi.json (L4176)

The documented error contract does not match the app's actual responses. src/memos/api/exceptions.py overrides the handlers so a 422 body is {"code", "message", "detail", "data"} (not just HTTPValidationError.detail), ValueError → 400 {"code","message","data"}, unhandled → 500, and the chat routes raise 503 when the handler is not initialized. None of 400/500/503 are declared for any path, so SDKs generated from this spec will type 422/error payloads incorrectly and will not model the real failure modes. Since this file is produced by make openapi, add shared responses/error models at the FastAPI app level and regenerate rather than editing the JSON.

---

2. docs/openapi.json (L1339)

Fields documented as "(Deprecated)" are not marked with the machine-readable "deprecated": true keyword (absent everywhere in the file: mem_cube_id, memory_content, doc_path, source, moscube). Generated clients therefore get no deprecation signal and will happily keep using them. Because this artifact is regenerated by make openapi, do not hand-edit it — set Field(deprecated=True) on these fields in src/memos/api/product_models.py so the keyword is emitted.

---

3. src/memos/mem_cube/general.py (L59)

The docstring promises that subsequently replaced memories are closed too, but nothing keeps _owned_memories in sync with the text_mem / act_mem / para_mem / pref_mem property setters: a memory assigned through a setter is never appended to _owned_memories, so it is never closed by the Cube (leak), while the previous instance stays tracked. Either drop the "including subsequently replaced ones" claim or register the new value in the setters (and decide whether the replaced memory should be closed there).

---

4. src/memos/mem_feedback/feedback.py (L123)

GraphDBError is not only used for unrecoverable errors in the openGauss backend: it is also raised for transient conditions such as "openGauss connection wait timed out" and "Could not open the openGauss connection pool". Setting retry_if_not_exception_type((GraphDBError, ValueError)) therefore disables the retry (previously 3 attempts with backoff) for retryable connection/pool-timeout failures on the write paths that use _retry_db_operation (_pure_add, _feedback_memory, ...). If the intent is only to avoid retrying deterministic/validation failures, consider excluding a narrower, dedicated exception (or a non-retryable subclass) instead of all GraphDBError.

---

5. src/memos/mem_feedback/feedback.py (L1070-L1071)

ValueError here is very broad for a try that wraps answer_future.result() / core_future.result(). Pydantic ValidationError and json.JSONDecodeError are ValueError subclasses, and process_feedback_core / _generate_answer do LLM response parsing and make_mem_item model construction, so a malformed LLM response now escapes as a hard failure instead of returning the degraded {"answer": "", "record": {"add": [], "update": []}} result as before. The same pattern was added in process_feedback_core, where the guarded block also covers all LLM/embedding/model work. Consider propagating only the specific error type(s) you actually want to surface (e.g. a dedicated config/DB exception) rather than the whole ValueError hierarchy.

---

6. src/memos/mem_scheduler/base_mixins/queue_ops.py (L309-L314)

_process_consumer runs in a forked child, which inherited self.dispatcher together with its live ThreadPoolExecutor (created in BaseScheduler.__init__). dispatcher.close() -> _close_executors() calls executor.shutdown(wait=True), which joins the inherited Thread objects. Those threads only exist in the parent process, so Thread.join() blocks on _tstate_lock that will never be released and the child never exits on its own. The child cannot drain anything either: tasks submitted by the child land in the forked executor whose workers do not exist in the child.

Effect: stop_consumer()/_join_consumers() always hit the 5s timeout, log "did not stop gracefully" and fall back to terminate(), instead of the graceful exit the docstring promises. Consider not touching the dispatcher in the child (just let the loop exit), or bounding the drain explicitly.

---

7. src/memos/mem_scheduler/base_scheduler.py (L480-L482)

Setting _borrowed_components = True here permanently disables the lazy fallback for both mem_cube and feedback_server, not just for the cube being assigned. init_mem_cube() sets self.feedback_server = feedback_server (frequently None, e.g. the mem_cubes setter -> init_mem_cube(mem_cube=selected_cube) path used by MOSCore.mem_scheduler), so after the first borrowed cube the feedback_server property silently returns None forever. Callers dereference it unconditionally, e.g. feedback_handler.process_single_feedback -> self.scheduler_context.get_feedback_server().process_feedback(...), which now raises AttributeError: 'NoneType' instead of lazily creating the server as before.

Suggest limiting the "borrowed" flag to the cube, and/or raising an explicit error (or keeping lazy init) when feedback_server is still None:

💡 Suggested Change

Before:

        with self._lifecycle_entry():
            self._borrowed_components = True
            self.current_mem_cube = value

After:

        with self._lifecycle_entry():
            self._borrowed_mem_cube = True
            self.current_mem_cube = value

---

8. src/memos/mem_scheduler/base_scheduler.py (L387)

_join_consumers joins the consumer and monitor threads without any timeout, while the process branch right below uses timeouts (join(timeout=5) + terminate()). If a handler dispatched by the consumer thread is long-running or the consumer is blocked inside memos_message_queue.get_messages()/qsize(), close() blocks indefinitely and the whole shutdown barrier never completes. Please bound these joins (e.g. thread.join(timeout=...) plus a warning, mirroring the process branch) so a stuck consumer cannot hang shutdown.

💡 Suggested Change

Before:

                thread.join()

After:

                thread.join(timeout=5)
                if thread.is_alive():
                    logger.warning("Scheduler worker %s did not stop in time", thread.name)

---

9. src/memos/memories/textual/tree.py (L544-L545)

Behavior change: a missing memory file used to log a warning and return (no-op), and the sibling implementations (memos/memories/textual/general.py, preference.py) still do that. GeneralMemCube.load() with default memory_types always calls text_mem.load(dir), so a cube that was dumped with a subset of memory types, or any directory without memory.json, will now fail the whole cube load with MemoryError instead of skipping text memory. Suggest keeping FileNotFoundError tolerant (warn + return) and only raising for genuinely unreadable/corrupt files.

💡 Suggested Change

Before:

        except (OSError, UnicodeError, json.JSONDecodeError) as error:
            raise MemoryError(f"Unable to read memory file: {memory_file}") from error

After:

        except FileNotFoundError:
            logger.warning("Memory file not found: %s", memory_file)
            return
        except (OSError, UnicodeError, json.JSONDecodeError) as error:
            raise MemoryError(f"Unable to read memory file: {memory_file}") from error

---

10. src/memos/memories/textual/tree.py (L567-L568)

The atomic rewrite changes the file mode of the dumped memory file: tempfile.NamedTemporaryFile creates the temp file with mode 0600, and os.replace moves that inode into place. Previously the file was created via open(memory_file, "w"), so it inherited the process umask (typically 0644). After this change a dump that is later shared/loaded by another user (sidecar process, different uid in the same volume) can fail with PermissionError, and load() will surface it as MemoryError. If owner-only dumps are not intentional, restore the previous permissions before publishing.

💡 Suggested Change

Before:

            os.replace(temporary_path, memory_file)
            temporary_path = None

After:

            os.chmod(temporary_path, 0o666 & ~os.umask(0) if False else 0o644)
            os.replace(temporary_path, memory_file)
            temporary_path = None

---

11. src/memos/memories/textual/tree_text_memory/organize/manager.py (L133)

This drops both the as_completed(..., timeout=500) guard and the per-future try/except. Consequences: (1) a hung graph write now blocks add() indefinitely with no timeout; (2) the first failing future aborts the whole call while futures after it in the list are never awaited, so their exceptions are silently lost (previously every failure was logged and processing continued). Consider restoring a timeout and per-future isolation, re-raising an aggregated error at the end if failures must be surfaced.

💡 Suggested Change

Before:

            added_ids = [node_id for future in futures for node_id in future.result()]

After:

            added_ids: list[str] = []
            errors: list[BaseException] = []
            for future in futures:
                try:
                    added_ids.extend(future.result())
                except Exception as error:
                    logger.exception("Memory processing error", exc_info=error)
                    errors.append(error)
            if errors:
                raise errors[0]

---

12. src/memos/memories/textual/tree_text_memory/organize/manager.py (L159-L160)

Per-batch failures are no longer logged/isolated (the old _submit_batches caught and logged each batch error). Now a single failing batch raises out of the loop after earlier batches have already been written, without any diagnostic of which batch failed or that partial state was left behind. Consider catching per future and logging the offset/batch index before re-raising.

---

13. src/memos/graph_dbs/opengauss.py (L1955-L1956)

The digit-string conversion has no length bound. On Python 3.10 (still covered by requires-python >= 3.10) int() has no 4300-digit safety limit, so a long all-zero string such as "0"*5000 parses to 0 and remove_oldest_memory silently deletes every memory of the type. On 3.11+ the rejection happens only as an accidental side effect of the interpreter's int-conversion limit, not from an explicit check. Bound the number of digits before converting so malformed input fails validation on every supported interpreter.

💡 Suggested Change

Before:

        if isinstance(keep_latest, str) and re.fullmatch(r"[0-9]+", keep_latest):
            keep_latest = int(keep_latest)

After:

        if isinstance(keep_latest, str) and re.fullmatch(r"[0-9]{1,9}", keep_latest):
            keep_latest = int(keep_latest)

---

14. src/memos/graph_dbs/opengauss.py (L2076)

datetime.fromisoformat does not accept the common trailing Z UTC designator on Python 3.10 (only from 3.11), which is a supported version per requires-python >= 3.10. A filter such as {"created_at": {"gte": "2024-01-01T00:00:00Z"}} will therefore raise ValueError: Invalid ISO date/time filter on 3.10 even though the value is valid ISO-8601. Normalize a trailing Z/z to +00:00 before parsing.

💡 Suggested Change

Before:

                    instant = datetime.fromisoformat(value)

After:

                    normalized = value[:-1] + "+00:00" if value.endswith(("Z", "z")) else value
                    instant = datetime.fromisoformat(normalized)

---

15. src/memos/graph_dbs/opengauss.py (L1416-L1429)

delete_node_by_prams issues one Cypher round-trip sequence per selected node inside the loop. Each _cypher call performs 4 statements (DEALLOCATE ALL, PREPARE, EXECUTE, DEALLOCATE), so deleting N memories costs ~4N extra round trips while holding the graph advisory lock (lock=2) and the semaphore slot — the whole delete becomes very slow and blocks other writers on the same graph. This is also inconsistent with _delete_locked_nodes, which already batches the same edge-deletion Cypher in chunks of 500. Batch the incident-edge deletion the same way (ids are already locked FOR UPDATE, so a single batched Cypher + one SQL DELETE ... business_id = ANY(...) keeps the same semantics):

selected = cursor.fetchall()
if not selected:
    return 0
by_scope: dict[str, list[str]] = {}
for owner, node_id in selected:
    by_scope.setdefault(owner, []).append(node_id)
return sum(
    self._delete_locked_nodes(cursor, owner, ids) for owner, ids in sorted(by_scope.items())
)

Generated by cloud-assistant via Open Code Review.

[Memtensor-AI]

⚠️ Automated Test Results: ENV ISSUE

The test environment encountered an issue that requires manual attention.

Details: Executor error: unsupported_lock/profile: changed poetry.lock requires its lock installer adapter
Branch: feat/opengauss-backend

Tested source: cdd17dc54c53f0f42977d9ae4bab54b1f9b1efb1
Tested base: 0e58d0e9fc3124adfc1a8cf6965eaf6b72c01309
Tested commit: cb681c2dced92a7871b7117e605664a4be8464da
These results apply only to this immutable source/base snapshot.

[Memtensor-AI]

⚠️ Automated Test Results: ENV ISSUE

The test environment encountered an issue that requires manual attention.

Details: Executor error: unsupported_lock/profile: changed poetry.lock requires its lock installer adapter
Branch: feat/opengauss-backend

Tested source: ad10805219524785580074010aac852d8249b76e
Tested base: 0e58d0e9fc3124adfc1a8cf6965eaf6b72c01309
Tested commit: ad10805219524785580074010aac852d8249b76e
These results apply only to this immutable source/base snapshot.