From 652a60803f3b4b5f2533a99e5dab3e40481da510 Mon Sep 17 00:00:00 2001 From: starboyate <2925776766@qq.com> Date: Mon, 11 May 2026 18:33:55 +0800 Subject: [PATCH] fix: add durable Open API ingestion --- README.md | 10 + .../openmemind/ai/client/MemindClient.java | 32 ++- .../model/response/AddMessageResponse.java | 19 ++ .../model/response/ExtractMemoryResponse.java | 27 ++ .../ai/client/MemindClientTest.java | 158 ++++++++++- memind-clients/python/README.md | 11 +- memind-clients/python/src/memind/__init__.py | 4 + .../src/memind/resources/async_memory.py | 12 +- .../python/src/memind/resources/memory.py | 12 +- .../python/src/memind/types/__init__.py | 4 + .../python/src/memind/types/memory.py | 15 ++ .../python/tests/test_async_client.py | 35 ++- memind-clients/python/tests/test_client.py | 100 ++++++- memind-clients/python/tests/test_models.py | 38 +++ .../core/extraction/rawdata/RawDataLayer.java | 46 +++- .../rawdata/InMemoryRawDataOperations.java | 36 +++ .../core/store/rawdata/RawDataOperations.java | 7 + .../rawdata/RawDataLayerProcessorTest.java | 200 ++++++++++++++ memind-integrations/claude-code/README.md | 61 ++--- .../claude-code/scripts/ingest.py | 80 +++--- .../claude-code/scripts/lib/client.py | 11 + .../claude-code/scripts/lib/config.py | 2 +- .../claude-code/scripts/session_start.py | 16 +- memind-integrations/claude-code/settings.json | 2 +- .../claude-code/tests/test_client.py | 32 +++ .../claude-code/tests/test_config.py | 2 +- .../claude-code/tests/test_hooks.py | 209 +++++++++++++++ .../claude-code/tests/test_manifest.py | 2 +- memind-integrations/codex/README.md | 30 ++- memind-integrations/codex/scripts/ingest.py | 81 ++---- .../codex/scripts/lib/client.py | 6 + .../codex/scripts/lib/config.py | 2 +- .../codex/scripts/session_start.py | 15 ++ memind-integrations/codex/settings.json | 2 +- .../codex/tests/test_client.py | 30 +++ memind-integrations/codex/tests/test_hooks.py | 162 +++++++++++- .../codex/tests/test_manifest.py | 2 +- .../plugin/jdbc/mysql/MysqlMemoryStore.java | 15 ++ .../postgresql/PostgresqlMemoryStore.java | 15 ++ .../plugin/jdbc/sqlite/SqliteMemoryStore.java | 15 ++ .../store/mybatis/MybatisPlusMemoryStore.java | 12 + .../openapi/OpenMemoryController.java | 51 ++++ .../openapi/OpenMemoryControllerTest.java | 247 ++++++++++++++++++ 43 files changed, 1644 insertions(+), 224 deletions(-) create mode 100644 memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/model/response/AddMessageResponse.java create mode 100644 memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/model/response/ExtractMemoryResponse.java diff --git a/README.md b/README.md index 30d32854..2c064345 100644 --- a/README.md +++ b/README.md @@ -239,6 +239,16 @@ For a runnable version with centralized configuration defaults, start with `memind-examples/memind-example-java/README.md` and `memind-examples/memind-example-java/src/main/java/com/openmemind/ai/memory/example/java/support/ExampleSettings.java`. +### Open API ingestion semantics + +The default ingestion endpoints (`/open/v1/memory/extract`, `/add-message`, and `/commit`) are +fire-and-forget: a successful HTTP response means Memind accepted and dispatched the work, not that extraction +completed. Retry-aware clients should use `/open/v1/memory/extract/sync` with a caller-owned raw-content payload +and clear their local retry state only when the returned extraction status is `SUCCESS`. + +`/open/v1/memory/add-message/sync` and `/open/v1/memory/commit/sync` report immediate server-buffer success or +failure, but they are not durable replay boundaries because the server owns the buffered conversation state. + --- ## Examples diff --git a/memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/MemindClient.java b/memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/MemindClient.java index c4a4467d..583a2db0 100644 --- a/memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/MemindClient.java +++ b/memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/MemindClient.java @@ -21,6 +21,8 @@ import com.openmemind.ai.client.model.request.CommitMemoryRequest; import com.openmemind.ai.client.model.request.ExtractMemoryRequest; import com.openmemind.ai.client.model.request.RetrieveMemoryRequest; +import com.openmemind.ai.client.model.response.AddMessageResponse; +import com.openmemind.ai.client.model.response.ExtractMemoryResponse; import com.openmemind.ai.client.model.response.HealthResponse; import com.openmemind.ai.client.model.response.RetrieveMemoryResponse; import java.time.Duration; @@ -50,8 +52,8 @@ public void addMessage(AddMessageRequest request) { joinAndUnwrap(addMessageAsync(request)); } - public void extract(ExtractMemoryRequest request) { - joinAndUnwrap(extractAsync(request)); + public ExtractMemoryResponse extract(ExtractMemoryRequest request) { + return joinAndUnwrap(extractAsync(request)); } public void commit(CommitMemoryRequest request) { @@ -68,26 +70,30 @@ public HealthResponse health() { public CompletableFuture addMessageAsync(AddMessageRequest request) { ensureOpen(); - return httpClient.post( - "/open/v1/memory/add-message", - Objects.requireNonNull(request, "request"), - new TypeReference>() {}); + return httpClient + .post( + "/open/v1/memory/add-message/sync", + Objects.requireNonNull(request, "request"), + new TypeReference>() {}) + .thenApply(ignored -> null); } - public CompletableFuture extractAsync(ExtractMemoryRequest request) { + public CompletableFuture extractAsync(ExtractMemoryRequest request) { ensureOpen(); return httpClient.post( - "/open/v1/memory/extract", + "/open/v1/memory/extract/sync", Objects.requireNonNull(request, "request"), - new TypeReference>() {}); + new TypeReference>() {}); } public CompletableFuture commitAsync(CommitMemoryRequest request) { ensureOpen(); - return httpClient.post( - "/open/v1/memory/commit", - Objects.requireNonNull(request, "request"), - new TypeReference>() {}); + return httpClient + .post( + "/open/v1/memory/commit/sync", + Objects.requireNonNull(request, "request"), + new TypeReference>() {}) + .thenApply(ignored -> null); } public CompletableFuture retrieveAsync(RetrieveMemoryRequest request) { diff --git a/memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/model/response/AddMessageResponse.java b/memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/model/response/AddMessageResponse.java new file mode 100644 index 00000000..e607847a --- /dev/null +++ b/memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/model/response/AddMessageResponse.java @@ -0,0 +1,19 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.openmemind.ai.client.model.response; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; + +@JsonIgnoreProperties(ignoreUnknown = true) +public record AddMessageResponse(boolean triggered, ExtractMemoryResponse result) {} diff --git a/memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/model/response/ExtractMemoryResponse.java b/memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/model/response/ExtractMemoryResponse.java new file mode 100644 index 00000000..2c7d039f --- /dev/null +++ b/memind-clients/java/memind-client/src/main/java/com/openmemind/ai/client/model/response/ExtractMemoryResponse.java @@ -0,0 +1,27 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.openmemind.ai.client.model.response; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import java.util.List; + +@JsonIgnoreProperties(ignoreUnknown = true) +public record ExtractMemoryResponse( + String status, + List rawDataIds, + List itemIds, + List insightIds, + boolean insightPending, + Long durationMillis, + String errorMessage) {} diff --git a/memind-clients/java/memind-client/src/test/java/com/openmemind/ai/client/MemindClientTest.java b/memind-clients/java/memind-client/src/test/java/com/openmemind/ai/client/MemindClientTest.java index b52426b7..a92529bb 100644 --- a/memind-clients/java/memind-client/src/test/java/com/openmemind/ai/client/MemindClientTest.java +++ b/memind-clients/java/memind-client/src/test/java/com/openmemind/ai/client/MemindClientTest.java @@ -33,8 +33,10 @@ import com.openmemind.ai.client.model.common.Message; import com.openmemind.ai.client.model.common.Strategy; import com.openmemind.ai.client.model.request.AddMessageRequest; +import com.openmemind.ai.client.model.request.CommitMemoryRequest; import com.openmemind.ai.client.model.request.ExtractMemoryRequest; import com.openmemind.ai.client.model.request.RetrieveMemoryRequest; +import com.openmemind.ai.client.model.response.ExtractMemoryResponse; import com.openmemind.ai.client.model.response.HealthResponse; import com.openmemind.ai.client.model.response.RetrieveMemoryResponse; import java.util.List; @@ -62,11 +64,11 @@ void health_returnsResponse(WireMockRuntimeInfo wmInfo) { @Test void addMessage_sendsCorrectPayload(WireMockRuntimeInfo wmInfo) { stubFor( - post("/open/v1/memory/add-message") + post("/open/v1/memory/add-message/sync") .willReturn( okJson( """ - {"code":"200"} + {"code":"success","data":{"triggered":false}} """))); try (MemindClient client = MemindClient.builder().baseUrl(wmInfo.getHttpBaseUrl()).build()) { @@ -79,7 +81,7 @@ void addMessage_sendsCorrectPayload(WireMockRuntimeInfo wmInfo) { } verify( - postRequestedFor(urlEqualTo("/open/v1/memory/add-message")) + postRequestedFor(urlEqualTo("/open/v1/memory/add-message/sync")) .withRequestBody(matchingJsonPath("$.userId", equalTo("user-1"))) .withRequestBody(matchingJsonPath("$.message.role", equalTo("USER")))); } @@ -87,28 +89,160 @@ void addMessage_sendsCorrectPayload(WireMockRuntimeInfo wmInfo) { @Test void extract_sendsRawContent(WireMockRuntimeInfo wmInfo) { stubFor( - post("/open/v1/memory/extract") + post("/open/v1/memory/extract/sync") .willReturn( okJson( """ - {"code":"200"} + {"code":"success","data":{ + "status":"SUCCESS", + "rawDataIds":["rd-1"], + "itemIds":[101], + "insightIds":[], + "insightPending":false, + "durationMillis":12 + }} """))); try (MemindClient client = MemindClient.builder().baseUrl(wmInfo.getHttpBaseUrl()).build()) { - client.extract( - ExtractMemoryRequest.builder() - .userId("user-1") - .agentId("agent-1") - .rawContent(ConversationContent.of(List.of(Message.user("test")))) - .build()); + ExtractMemoryResponse response = + client.extract( + ExtractMemoryRequest.builder() + .userId("user-1") + .agentId("agent-1") + .rawContent(ConversationContent.of(List.of(Message.user("test")))) + .build()); + + assertThat(response.status()).isEqualTo("SUCCESS"); + assertThat(response.rawDataIds()).containsExactly("rd-1"); } verify( - postRequestedFor(urlEqualTo("/open/v1/memory/extract")) + postRequestedFor(urlEqualTo("/open/v1/memory/extract/sync")) .withRequestBody( matchingJsonPath("$.rawContent.type", equalTo("conversation")))); } + @Test + void commit_usesSyncEndpoint(WireMockRuntimeInfo wmInfo) { + stubFor( + post("/open/v1/memory/commit/sync") + .willReturn( + okJson( + """ + {"code":"success","data":{ + "status":"SUCCESS", + "rawDataIds":[], + "itemIds":[], + "insightIds":[], + "insightPending":false + }} + """))); + + try (MemindClient client = MemindClient.builder().baseUrl(wmInfo.getHttpBaseUrl()).build()) { + client.commit( + CommitMemoryRequest.builder().userId("user-1").agentId("agent-1").build()); + } + + verify(postRequestedFor(urlEqualTo("/open/v1/memory/commit/sync"))); + } + + @Test + void extract_partialSuccessIsReturnedToCaller(WireMockRuntimeInfo wmInfo) { + stubFor( + post("/open/v1/memory/extract/sync") + .willReturn( + okJson( + """ + {"code":"success","data":{ + "status":"PARTIAL_SUCCESS", + "rawDataIds":["rd-1"], + "itemIds":[], + "insightIds":[], + "insightPending":false, + "errorMessage":"insight failed" + }} + """))); + + try (MemindClient client = MemindClient.builder().baseUrl(wmInfo.getHttpBaseUrl()).build()) { + ExtractMemoryResponse response = + client.extract( + ExtractMemoryRequest.builder() + .userId("u") + .agentId("a") + .rawContent(ConversationContent.of(List.of(Message.user("test")))) + .build()); + + assertThat(response.status()).isEqualTo("PARTIAL_SUCCESS"); + assertThat(response.errorMessage()).isEqualTo("insight failed"); + } + } + + @Test + void extract_failureEnvelopeThrowsApiException(WireMockRuntimeInfo wmInfo) { + stubFor( + post("/open/v1/memory/extract/sync") + .willReturn( + aResponse() + .withStatus(500) + .withHeader("Content-Type", "application/json") + .withBody( + """ + {"code":"extraction_failed","message":"extract failed","traceId":"t1"} + """))); + + try (MemindClient client = MemindClient.builder().baseUrl(wmInfo.getHttpBaseUrl()).build()) { + assertThatThrownBy( + () -> + client.extract( + ExtractMemoryRequest.builder() + .userId("u") + .agentId("a") + .rawContent( + ConversationContent.of( + List.of(Message.user("test")))) + .build())) + .isInstanceOf(MemindApiException.class) + .satisfies( + ex -> { + var apiEx = (MemindApiException) ex; + assertThat(apiEx.getHttpStatus()).isEqualTo(500); + assertThat(apiEx.getErrorCode()).isEqualTo("extraction_failed"); + }); + } + } + + @Test + void extractAsync_returnsExtractionResponse(WireMockRuntimeInfo wmInfo) { + stubFor( + post("/open/v1/memory/extract/sync") + .willReturn( + okJson( + """ + {"code":"success","data":{ + "status":"SUCCESS", + "rawDataIds":["rd-async"], + "itemIds":[], + "insightIds":[], + "insightPending":false + }} + """))); + + try (MemindClient client = MemindClient.builder().baseUrl(wmInfo.getHttpBaseUrl()).build()) { + ExtractMemoryResponse response = + client.extractAsync( + ExtractMemoryRequest.builder() + .userId("user-1") + .agentId("agent-1") + .rawContent(ConversationContent.of(List.of(Message.user("test")))) + .build()) + .join(); + + assertThat(response.rawDataIds()).containsExactly("rd-async"); + } + + verify(postRequestedFor(urlEqualTo("/open/v1/memory/extract/sync"))); + } + @Test void retrieve_returnsMemories(WireMockRuntimeInfo wmInfo) { stubFor( diff --git a/memind-clients/python/README.md b/memind-clients/python/README.md index fd054189..8851dbe3 100644 --- a/memind-clients/python/README.md +++ b/memind-clients/python/README.md @@ -17,13 +17,12 @@ from memind.types import ConversationContent with MemindClient(base_url="http://localhost:8080") as client: health = client.health() - client.memory.extract( + response = client.memory.extract( user_id="user-1", agent_id="agent-1", - raw_content=ConversationContent(messages=[Message.user("I like coffee")]), + raw_content=ConversationContent(messages=[Message.user("Remember that I prefer concise answers.")]), ) - - client.memory.commit(user_id="user-1", agent_id="agent-1") + print(response.status) result = client.memory.retrieve( user_id="user-1", @@ -34,6 +33,10 @@ with MemindClient(base_url="http://localhost:8080") as client: ) ``` +`memory.extract()` uses Memind's synchronous extraction endpoint and returns `ExtractMemoryResponse`. Treat only +`status == "SUCCESS"` as safe to clear caller-owned retry payloads; `PARTIAL_SUCCESS` is surfaced so applications +can keep or re-enqueue the original payload. + ## Asynchronous Usage ```python diff --git a/memind-clients/python/src/memind/__init__.py b/memind-clients/python/src/memind/__init__.py index e3c094b5..ca2c0774 100644 --- a/memind-clients/python/src/memind/__init__.py +++ b/memind-clients/python/src/memind/__init__.py @@ -25,6 +25,7 @@ from memind._version import __version__ from memind.types import ( AddMessageRequest, + AddMessageResponse, ApiResult, AudioBlock, Base64Source, @@ -32,6 +33,7 @@ ContentBlock, ConversationContent, ExtractMemoryRequest, + ExtractMemoryResponse, FinalView, HealthResponse, ImageBlock, @@ -57,6 +59,7 @@ __all__ = [ "AddMessageRequest", + "AddMessageResponse", "ApiResult", "AsyncMemindClient", "AudioBlock", @@ -65,6 +68,7 @@ "ContentBlock", "ConversationContent", "ExtractMemoryRequest", + "ExtractMemoryResponse", "FinalView", "HealthResponse", "ImageBlock", diff --git a/memind-clients/python/src/memind/resources/async_memory.py b/memind-clients/python/src/memind/resources/async_memory.py index b74a7803..e43654c6 100644 --- a/memind-clients/python/src/memind/resources/async_memory.py +++ b/memind-clients/python/src/memind/resources/async_memory.py @@ -19,8 +19,10 @@ from memind.types.common import Strategy from memind.types.memory import ( AddMessageRequest, + AddMessageResponse, CommitMemoryRequest, ExtractMemoryRequest, + ExtractMemoryResponse, RetrieveMemoryRequest, RetrieveMemoryResponse, ) @@ -42,14 +44,16 @@ async def extract( agent_id: str | None = None, raw_content: RawContentValue | None = None, source_client: str | None = None, - ) -> None: + ) -> ExtractMemoryResponse: payload = request or _build_extract_request( user_id=user_id, agent_id=agent_id, raw_content=raw_content, source_client=source_client, ) - await self._client._post("/memory/extract", payload, None) + result = await self._client._post("/memory/extract/sync", payload, ExtractMemoryResponse) + assert result is not None + return result async def add_message( self, @@ -66,7 +70,7 @@ async def add_message( message=_required(message, "message"), source_client=source_client, ) - await self._client._post("/memory/add-message", payload, None) + await self._client._post("/memory/add-message/sync", payload, AddMessageResponse) async def commit( self, @@ -81,7 +85,7 @@ async def commit( agent_id=_required(agent_id, "agent_id"), source_client=source_client, ) - await self._client._post("/memory/commit", payload, None) + await self._client._post("/memory/commit/sync", payload, ExtractMemoryResponse) async def retrieve( self, diff --git a/memind-clients/python/src/memind/resources/memory.py b/memind-clients/python/src/memind/resources/memory.py index a3789080..3e82413e 100644 --- a/memind-clients/python/src/memind/resources/memory.py +++ b/memind-clients/python/src/memind/resources/memory.py @@ -19,8 +19,10 @@ from memind.types.common import Strategy from memind.types.memory import ( AddMessageRequest, + AddMessageResponse, CommitMemoryRequest, ExtractMemoryRequest, + ExtractMemoryResponse, RetrieveMemoryRequest, RetrieveMemoryResponse, ) @@ -42,14 +44,16 @@ def extract( agent_id: str | None = None, raw_content: RawContentValue | None = None, source_client: str | None = None, - ) -> None: + ) -> ExtractMemoryResponse: payload = request or _build_extract_request( user_id=user_id, agent_id=agent_id, raw_content=raw_content, source_client=source_client, ) - self._client._post("/memory/extract", payload, None) + result = self._client._post("/memory/extract/sync", payload, ExtractMemoryResponse) + assert result is not None + return result def add_message( self, @@ -66,7 +70,7 @@ def add_message( message=_required(message, "message"), source_client=source_client, ) - self._client._post("/memory/add-message", payload, None) + self._client._post("/memory/add-message/sync", payload, AddMessageResponse) def commit( self, @@ -81,7 +85,7 @@ def commit( agent_id=_required(agent_id, "agent_id"), source_client=source_client, ) - self._client._post("/memory/commit", payload, None) + self._client._post("/memory/commit/sync", payload, ExtractMemoryResponse) def retrieve( self, diff --git a/memind-clients/python/src/memind/types/__init__.py b/memind-clients/python/src/memind/types/__init__.py index 47d1cb2a..c1883c8c 100644 --- a/memind-clients/python/src/memind/types/__init__.py +++ b/memind-clients/python/src/memind/types/__init__.py @@ -16,8 +16,10 @@ from memind.types.health import HealthResponse from memind.types.memory import ( AddMessageRequest, + AddMessageResponse, CommitMemoryRequest, ExtractMemoryRequest, + ExtractMemoryResponse, FinalView, MergeView, RetrievalTraceView, @@ -46,6 +48,7 @@ __all__ = [ "AddMessageRequest", + "AddMessageResponse", "ApiResult", "AudioBlock", "Base64Source", @@ -53,6 +56,7 @@ "ContentBlock", "ConversationContent", "ExtractMemoryRequest", + "ExtractMemoryResponse", "FinalView", "HealthResponse", "ImageBlock", diff --git a/memind-clients/python/src/memind/types/memory.py b/memind-clients/python/src/memind/types/memory.py index 03a9cfe9..b26c5bf8 100644 --- a/memind-clients/python/src/memind/types/memory.py +++ b/memind-clients/python/src/memind/types/memory.py @@ -43,6 +43,21 @@ class CommitMemoryRequest(MemindModel): source_client: str | None = None +class ExtractMemoryResponse(MemindModel): + status: str + raw_data_ids: list[str] = Field(default_factory=list) + item_ids: list[int] = Field(default_factory=list) + insight_ids: list[int] = Field(default_factory=list) + insight_pending: bool = False + duration_millis: int | None = None + error_message: str | None = None + + +class AddMessageResponse(MemindModel): + triggered: bool + result: ExtractMemoryResponse | None = None + + class RetrieveMemoryRequest(MemindModel): user_id: str agent_id: str diff --git a/memind-clients/python/tests/test_async_client.py b/memind-clients/python/tests/test_async_client.py index 49bfc2e5..f9cefaa4 100644 --- a/memind-clients/python/tests/test_async_client.py +++ b/memind-clients/python/tests/test_async_client.py @@ -39,22 +39,40 @@ async def test_async_health_returns_response(httpx_mock) -> None: async def test_async_memory_methods_send_payloads(httpx_mock) -> None: httpx_mock.add_response( method="POST", - url="https://api.example.test/open/v1/memory/extract", - json={"code": "200"}, + url="https://api.example.test/open/v1/memory/extract/sync", + json={ + "code": "success", + "data": { + "status": "SUCCESS", + "rawDataIds": ["rd-1"], + "itemIds": [], + "insightIds": [], + "insightPending": False, + }, + }, ) httpx_mock.add_response( method="POST", - url="https://api.example.test/open/v1/memory/add-message", - json={"code": "200"}, + url="https://api.example.test/open/v1/memory/add-message/sync", + json={"code": "success", "data": {"triggered": False}}, ) httpx_mock.add_response( method="POST", - url="https://api.example.test/open/v1/memory/commit", - json={"code": "200"}, + url="https://api.example.test/open/v1/memory/commit/sync", + json={ + "code": "success", + "data": { + "status": "SUCCESS", + "rawDataIds": [], + "itemIds": [], + "insightIds": [], + "insightPending": False, + }, + }, ) client = AsyncMemindClient(base_url="https://api.example.test") - await client.memory.extract( + extract_response = await client.memory.extract( user_id="u1", agent_id="a1", raw_content=ConversationContent(messages=[Message.user("hello")]), @@ -65,6 +83,7 @@ async def test_async_memory_methods_send_payloads(httpx_mock) -> None: requests = httpx_mock.get_requests() assert len(requests) == 3 + assert extract_response.status == "SUCCESS" assert b'"rawContent":{"type":"conversation"' in requests[0].content assert b'"message":{"role":"USER"' in requests[1].content assert b'"userId":"u1"' in requests[2].content @@ -128,7 +147,7 @@ async def test_async_close_then_call_raises_memind_error() -> None: async def test_async_mutating_post_methods_do_not_retry_by_default(httpx_mock) -> None: httpx_mock.add_response( method="POST", - url="https://api.example.test/open/v1/memory/add-message", + url="https://api.example.test/open/v1/memory/add-message/sync", status_code=503, json={"code": "unavailable"}, ) diff --git a/memind-clients/python/tests/test_client.py b/memind-clients/python/tests/test_client.py index e000ef91..ebf1f618 100644 --- a/memind-clients/python/tests/test_client.py +++ b/memind-clients/python/tests/test_client.py @@ -18,7 +18,13 @@ from memind._client import MemindClient from memind._exceptions import MemindAPIError, MemindError -from memind.types import ConversationContent, Message, RetrieveMemoryRequest, Strategy +from memind.types import ( + ConversationContent, + ExtractMemoryResponse, + Message, + RetrieveMemoryRequest, + Strategy, +) def test_health_returns_response(httpx_mock) -> None: @@ -39,8 +45,8 @@ def test_health_returns_response(httpx_mock) -> None: def test_add_message_sends_payload_and_auth_header(httpx_mock) -> None: httpx_mock.add_response( method="POST", - url="https://api.example.test/open/v1/memory/add-message", - json={"code": "200"}, + url="https://api.example.test/open/v1/memory/add-message/sync", + json={"code": "success", "data": {"triggered": False}}, ) client = MemindClient(base_url="https://api.example.test", api_token="sk-test") @@ -56,17 +62,30 @@ def test_add_message_sends_payload_and_auth_header(httpx_mock) -> None: def test_extract_sends_raw_content(httpx_mock) -> None: httpx_mock.add_response( method="POST", - url="https://api.example.test/open/v1/memory/extract", - json={"code": "200"}, + url="https://api.example.test/open/v1/memory/extract/sync", + json={ + "code": "success", + "data": { + "status": "SUCCESS", + "rawDataIds": ["rd-1"], + "itemIds": [101], + "insightIds": [], + "insightPending": False, + "durationMillis": 12, + }, + }, ) client = MemindClient(base_url="https://api.example.test") - client.memory.extract( + response = client.memory.extract( user_id="u1", agent_id="a1", raw_content=ConversationContent(messages=[Message.user("hello")]), ) + assert isinstance(response, ExtractMemoryResponse) + assert response.status == "SUCCESS" + assert response.raw_data_ids == ["rd-1"] assert b'"rawContent":{"type":"conversation"' in httpx_mock.get_request().content client.close() @@ -74,8 +93,17 @@ def test_extract_sends_raw_content(httpx_mock) -> None: def test_commit_sends_payload(httpx_mock) -> None: httpx_mock.add_response( method="POST", - url="https://api.example.test/open/v1/memory/commit", - json={"code": "200"}, + url="https://api.example.test/open/v1/memory/commit/sync", + json={ + "code": "success", + "data": { + "status": "SUCCESS", + "rawDataIds": [], + "itemIds": [], + "insightIds": [], + "insightPending": False, + }, + }, ) client = MemindClient(base_url="https://api.example.test") @@ -85,6 +113,60 @@ def test_commit_sends_payload(httpx_mock) -> None: client.close() +def test_extract_partial_success_is_returned(httpx_mock) -> None: + httpx_mock.add_response( + method="POST", + url="https://api.example.test/open/v1/memory/extract/sync", + json={ + "code": "success", + "data": { + "status": "PARTIAL_SUCCESS", + "rawDataIds": ["rd-1"], + "itemIds": [], + "insightIds": [], + "insightPending": False, + "errorMessage": "insight failed", + }, + }, + ) + + client = MemindClient(base_url="https://api.example.test") + response = client.memory.extract( + user_id="u1", + agent_id="a1", + raw_content=ConversationContent(messages=[Message.user("hello")]), + ) + + assert response.status == "PARTIAL_SUCCESS" + assert response.error_message == "insight failed" + client.close() + + +def test_extract_failure_envelope_raises_api_error(httpx_mock) -> None: + httpx_mock.add_response( + method="POST", + url="https://api.example.test/open/v1/memory/extract/sync", + status_code=500, + json={ + "code": "extraction_failed", + "message": "extract failed", + "traceId": "t1", + }, + ) + + client = MemindClient(base_url="https://api.example.test") + with pytest.raises(MemindAPIError) as exc_info: + client.memory.extract( + user_id="u1", + agent_id="a1", + raw_content=ConversationContent(messages=[Message.user("hello")]), + ) + + assert exc_info.value.status_code == 500 + assert exc_info.value.error_code == "extraction_failed" + client.close() + + def test_retrieve_accepts_expanded_parameters(httpx_mock) -> None: httpx_mock.add_response( method="POST", @@ -161,7 +243,7 @@ def test_close_then_call_raises_memind_error() -> None: def test_mutating_post_methods_do_not_retry_by_default(httpx_mock) -> None: httpx_mock.add_response( method="POST", - url="https://api.example.test/open/v1/memory/add-message", + url="https://api.example.test/open/v1/memory/add-message/sync", status_code=503, json={"code": "unavailable"}, ) diff --git a/memind-clients/python/tests/test_models.py b/memind-clients/python/tests/test_models.py index db3c8af0..aa170946 100644 --- a/memind-clients/python/tests/test_models.py +++ b/memind-clients/python/tests/test_models.py @@ -22,8 +22,10 @@ from memind.types.health import HealthResponse from memind.types.memory import ( AddMessageRequest, + AddMessageResponse, CommitMemoryRequest, ExtractMemoryRequest, + ExtractMemoryResponse, RetrieveMemoryRequest, RetrieveMemoryResponse, ) @@ -210,6 +212,23 @@ def test_extract_memory_request(self) -> None: assert dumped["rawContent"]["messages"][0]["content"][0]["text"] == "hi" assert "sourceClient" not in dumped + def test_extract_memory_response_aliases(self) -> None: + response = ExtractMemoryResponse.model_validate( + { + "status": "SUCCESS", + "rawDataIds": ["rd-1"], + "itemIds": [101], + "insightIds": [201], + "insightPending": True, + "durationMillis": 12, + "errorMessage": None, + } + ) + + assert response.raw_data_ids == ["rd-1"] + assert response.item_ids == [101] + assert response.insight_pending is True + def test_add_message_request(self) -> None: req = AddMessageRequest( user_id="u1", @@ -222,6 +241,25 @@ def test_add_message_request(self) -> None: assert dumped["message"]["role"] == "USER" assert dumped["sourceClient"] == "python-sdk" + def test_add_message_response_aliases(self) -> None: + response = AddMessageResponse.model_validate( + { + "triggered": True, + "result": { + "status": "SUCCESS", + "rawDataIds": ["rd-1"], + "itemIds": [], + "insightIds": [], + "insightPending": False, + }, + } + ) + + assert response.triggered is True + assert response.result is not None + assert response.result.status == "SUCCESS" + assert response.result.raw_data_ids == ["rd-1"] + def test_commit_memory_request(self) -> None: req = CommitMemoryRequest(user_id="u1", agent_id="a1") dumped = req.model_dump(by_alias=True, exclude_none=True) diff --git a/memind-core/src/main/java/com/openmemind/ai/memory/core/extraction/rawdata/RawDataLayer.java b/memind-core/src/main/java/com/openmemind/ai/memory/core/extraction/rawdata/RawDataLayer.java index e3f4d116..f2cc68c5 100644 --- a/memind-core/src/main/java/com/openmemind/ai/memory/core/extraction/rawdata/RawDataLayer.java +++ b/memind-core/src/main/java/com/openmemind/ai/memory/core/extraction/rawdata/RawDataLayer.java @@ -149,8 +149,7 @@ private Mono process(RawDataInput input, String language) // Idempotency check Optional existing = memoryStore.rawDataOperations().getRawDataByContentId(memoryId, contentId); - return existing.map( - memoryRawData -> Mono.just(RawDataProcessResult.existing(memoryRawData))) + return existing.map(memoryRawData -> Mono.just(existingRawDataResult(memoryId, contentId))) .orElseGet(() -> doProcess(input, memoryId, contentId, language)); } @@ -177,7 +176,7 @@ public Mono processSegment( Optional existing = memoryStore.rawDataOperations().getRawDataByContentId(memoryId, contentId); if (existing.isPresent()) { - return Mono.just(RawDataResult.existing(existing.get())); + return Mono.just(toRawDataResult(memoryId, contentId)); } String contentType = (type != null && !type.isBlank()) ? type : ConversationContent.TYPE; @@ -446,6 +445,47 @@ private RawDataProcessResult buildAndPersist( return new RawDataProcessResult(rawDataList, parsedSegments, false); } + private RawDataProcessResult existingRawDataResult(MemoryId memoryId, String contentId) { + RawDataResult result = toRawDataResult(memoryId, contentId); + return new RawDataProcessResult(result.rawDataList(), result.segments(), true); + } + + private RawDataResult toRawDataResult(MemoryId memoryId, String contentId) { + List rawDataList = + memoryStore.rawDataOperations().listRawDataByContentId(memoryId, contentId); + List segments = rawDataList.stream().map(this::toParsedSegment).toList(); + return new RawDataResult(rawDataList, segments, true); + } + + private ParsedSegment toParsedSegment(MemoryRawData rawData) { + Segment segment = rawData.segment(); + boolean hasBoundary = segment != null && segment.boundary() != null; + return new ParsedSegment( + segment == null ? null : segment.content(), + rawData.caption(), + hasBoundary ? getBoundaryStart(segment) : 0, + hasBoundary ? getBoundaryEnd(segment) : 0, + rawData.id(), + rawData.metadata(), + replayRuntimeContext(rawData, segment)); + } + + private SegmentRuntimeContext replayRuntimeContext(MemoryRawData rawData, Segment segment) { + if (segment != null && segment.runtimeContext() != null) { + return segment.runtimeContext(); + } + String sourceClient = rawData.sourceClient(); + if (sourceClient == null && rawData.metadata() != null) { + Object metadataSource = rawData.metadata().get("sourceClient"); + sourceClient = metadataSource == null ? null : metadataSource.toString(); + } + if (rawData.startTime() == null && rawData.endTime() == null && sourceClient == null) { + return null; + } + return new SegmentRuntimeContext( + rawData.startTime(), rawData.endTime(), null, sourceClient); + } + private String resolveRawDataContentId(RawDataInput input) { String textContentId = input.content().getContentId(); if (!getTypedProcessor(input.content()).usesSourceIdentity()) { diff --git a/memind-core/src/main/java/com/openmemind/ai/memory/core/store/rawdata/InMemoryRawDataOperations.java b/memind-core/src/main/java/com/openmemind/ai/memory/core/store/rawdata/InMemoryRawDataOperations.java index ba682ec9..c20f4186 100644 --- a/memind-core/src/main/java/com/openmemind/ai/memory/core/store/rawdata/InMemoryRawDataOperations.java +++ b/memind-core/src/main/java/com/openmemind/ai/memory/core/store/rawdata/InMemoryRawDataOperations.java @@ -17,6 +17,7 @@ import com.openmemind.ai.memory.core.data.MemoryRawData; import java.time.Duration; import java.time.Instant; +import java.util.Comparator; import java.util.List; import java.util.Map; import java.util.Objects; @@ -56,6 +57,17 @@ public Optional getRawDataByContentId(MemoryId id, String content .findFirst(); } + @Override + public List listRawDataByContentId(MemoryId id, String contentId) { + return rawDataStore.getOrDefault(key(id), Map.of()).values().stream() + .filter(r -> Objects.equals(r.contentId(), contentId)) + .sorted( + Comparator.comparing(InMemoryRawDataOperations::boundaryStart) + .thenComparing(InMemoryRawDataOperations::boundaryEnd) + .thenComparing(MemoryRawData::id)) + .toList(); + } + @Override public List listRawData(MemoryId id) { return List.copyOf(rawDataStore.getOrDefault(key(id), Map.of()).values()); @@ -93,4 +105,28 @@ public void updateRawDataVectorIds( } }); } + + private static int boundaryStart(MemoryRawData rawData) { + if (rawData.segment() == null || rawData.segment().boundary() == null) { + return Integer.MAX_VALUE; + } + return switch (rawData.segment().boundary()) { + case com.openmemind.ai.memory.core.extraction.rawdata.segment.MessageBoundary mb -> + mb.startMessage(); + case com.openmemind.ai.memory.core.extraction.rawdata.segment.CharBoundary cb -> + cb.startChar(); + }; + } + + private static int boundaryEnd(MemoryRawData rawData) { + if (rawData.segment() == null || rawData.segment().boundary() == null) { + return Integer.MAX_VALUE; + } + return switch (rawData.segment().boundary()) { + case com.openmemind.ai.memory.core.extraction.rawdata.segment.MessageBoundary mb -> + mb.endMessage(); + case com.openmemind.ai.memory.core.extraction.rawdata.segment.CharBoundary cb -> + cb.endChar(); + }; + } } diff --git a/memind-core/src/main/java/com/openmemind/ai/memory/core/store/rawdata/RawDataOperations.java b/memind-core/src/main/java/com/openmemind/ai/memory/core/store/rawdata/RawDataOperations.java index 9e546f91..4ab6aa7c 100644 --- a/memind-core/src/main/java/com/openmemind/ai/memory/core/store/rawdata/RawDataOperations.java +++ b/memind-core/src/main/java/com/openmemind/ai/memory/core/store/rawdata/RawDataOperations.java @@ -18,6 +18,7 @@ import java.time.Duration; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; /** @@ -31,6 +32,12 @@ public interface RawDataOperations { Optional getRawDataByContentId(MemoryId id, String contentId); + default List listRawDataByContentId(MemoryId id, String contentId) { + return listRawData(id).stream() + .filter(rawData -> Objects.equals(rawData.contentId(), contentId)) + .toList(); + } + List listRawData(MemoryId id); List pollRawDataWithoutVector(MemoryId id, int limit, Duration minAge); diff --git a/memind-core/src/test/java/com/openmemind/ai/memory/core/extraction/rawdata/RawDataLayerProcessorTest.java b/memind-core/src/test/java/com/openmemind/ai/memory/core/extraction/rawdata/RawDataLayerProcessorTest.java index 17e03125..cfe4d34e 100644 --- a/memind-core/src/test/java/com/openmemind/ai/memory/core/extraction/rawdata/RawDataLayerProcessorTest.java +++ b/memind-core/src/test/java/com/openmemind/ai/memory/core/extraction/rawdata/RawDataLayerProcessorTest.java @@ -31,6 +31,7 @@ import com.openmemind.ai.memory.core.extraction.rawdata.caption.CaptionGenerator; import com.openmemind.ai.memory.core.extraction.rawdata.content.ConversationContent; import com.openmemind.ai.memory.core.extraction.rawdata.content.RawContent; +import com.openmemind.ai.memory.core.extraction.rawdata.content.conversation.message.Message; import com.openmemind.ai.memory.core.extraction.rawdata.processor.ConversationContentProcessor; import com.openmemind.ai.memory.core.extraction.rawdata.segment.CharBoundary; import com.openmemind.ai.memory.core.extraction.rawdata.segment.MessageBoundary; @@ -361,6 +362,8 @@ void processSegmentReturnsExistingRawDataWithoutVectorizingAgain() { Instant.parse("2026-03-27T02:18:00Z")); when(rawDataOps.getRawDataByContentId(memoryId, "content-1")) .thenReturn(java.util.Optional.of(existing)); + when(rawDataOps.listRawDataByContentId(memoryId, "content-1")) + .thenReturn(List.of(existing)); var result = layer.processSegment( @@ -370,9 +373,184 @@ void processSegmentReturnsExistingRawDataWithoutVectorizingAgain() { assertThat(result).isNotNull(); assertThat(result.existed()).isTrue(); assertThat(result.rawDataList()).containsExactly(existing); + assertThat(result.segments()).hasSize(1); + assertThat(result.segments().get(0).rawDataId()).isEqualTo("raw-1"); + assertThat(result.segments().get(0).text()).isEqualTo("persisted"); + assertThat(result.segments().get(0).caption()).isEqualTo("caption"); verifyNoInteractions(defaultCaption, vector); } + @Test + @DisplayName("idempotency hit should restore parsed segment for item extraction replay") + void idempotencyHitRestoresParsedSegmentForItemExtractionReplay() { + var memoryId = new com.openmemind.ai.memory.core.data.DefaultMemoryId("test", "agent"); + var convProcessor = mock(ConversationContentProcessor.class); + when(convProcessor.contentClass()).thenReturn(ConversationContent.class); + var layer = new RawDataLayer(List.of(convProcessor), defaultCaption, store, vector, 64); + var content = + new ConversationContent(List.of(Message.user("remember that I like espresso"))) { + @Override + public String getContentId() { + return "content-1"; + } + }; + MemoryRawData existing = + rawData( + memoryId, + "rd-existing", + "content-1", + Segment.single("remember that I like espresso"), + "espresso preference"); + when(rawDataOps.getRawDataByContentId(memoryId, "content-1")) + .thenReturn(Optional.of(existing)); + when(rawDataOps.listRawDataByContentId(memoryId, "content-1")) + .thenReturn(List.of(existing)); + + var result = + layer.extract(memoryId, content, ConversationContent.TYPE, Map.of(), null).block(); + + assertThat(result.existed()).isTrue(); + assertThat(result.rawDataList()) + .extracting(MemoryRawData::id) + .containsExactly("rd-existing"); + assertThat(result.segments()).hasSize(1); + assertThat(result.segments().get(0).rawDataId()).isEqualTo("rd-existing"); + assertThat(result.segments().get(0).text()).isEqualTo("remember that I like espresso"); + assertThat(result.segments().get(0).caption()).isEqualTo("espresso preference"); + } + + @Test + @DisplayName("idempotency hit should restore all segments in stable order") + void idempotencyHitRestoresAllSegmentsInStableOrder() { + var memoryId = new com.openmemind.ai.memory.core.data.DefaultMemoryId("test", "agent"); + var convProcessor = mock(ConversationContentProcessor.class); + when(convProcessor.contentClass()).thenReturn(ConversationContent.class); + var layer = new RawDataLayer(List.of(convProcessor), defaultCaption, store, vector, 64); + var content = + new ConversationContent( + List.of(Message.user("first"), Message.assistant("second"))) { + @Override + public String getContentId() { + return "content-1"; + } + }; + MemoryRawData second = + rawData( + memoryId, + "rd-2", + "content-1", + new Segment("second", "second caption", new CharBoundary(10, 20), Map.of()), + "second caption"); + MemoryRawData first = + rawData( + memoryId, + "rd-1", + "content-1", + new Segment("first", "first caption", new CharBoundary(0, 9), Map.of()), + "first caption"); + when(rawDataOps.getRawDataByContentId(memoryId, "content-1")) + .thenReturn(Optional.of(second)); + when(rawDataOps.listRawDataByContentId(memoryId, "content-1")) + .thenReturn(List.of(first, second)); + + var result = + layer.extract(memoryId, content, ConversationContent.TYPE, Map.of(), null).block(); + + assertThat(result.existed()).isTrue(); + assertThat(result.rawDataList()) + .extracting(MemoryRawData::id) + .containsExactly("rd-1", "rd-2"); + assertThat(result.segments()) + .extracting(ParsedSegment::rawDataId) + .containsExactly("rd-1", "rd-2"); + assertThat(result.segments()) + .extracting(ParsedSegment::text) + .containsExactly("first", "second"); + } + + @Test + @DisplayName("idempotency hit should restore boundaryless segment without throwing") + void idempotencyHitRestoresBoundarylessSegmentWithoutThrowing() { + var memoryId = new com.openmemind.ai.memory.core.data.DefaultMemoryId("test", "agent"); + var convProcessor = mock(ConversationContentProcessor.class); + when(convProcessor.contentClass()).thenReturn(ConversationContent.class); + var layer = new RawDataLayer(List.of(convProcessor), defaultCaption, store, vector, 64); + var content = + new ConversationContent(List.of(Message.user("legacy segment"))) { + @Override + public String getContentId() { + return "content-legacy"; + } + }; + MemoryRawData existing = + rawData( + memoryId, + "rd-legacy", + "content-legacy", + new Segment("legacy segment", "legacy caption", null, Map.of()), + "legacy caption"); + when(rawDataOps.getRawDataByContentId(memoryId, "content-legacy")) + .thenReturn(Optional.of(existing)); + when(rawDataOps.listRawDataByContentId(memoryId, "content-legacy")) + .thenReturn(List.of(existing)); + + var result = + layer.extract(memoryId, content, ConversationContent.TYPE, Map.of(), null).block(); + + assertThat(result.existed()).isTrue(); + assertThat(result.segments()).hasSize(1); + assertThat(result.segments().get(0).rawDataId()).isEqualTo("rd-legacy"); + assertThat(result.segments().get(0).startIndex()).isZero(); + assertThat(result.segments().get(0).endIndex()).isZero(); + } + + @Test + @DisplayName("idempotency hit should restore persisted timing and source client context") + void idempotencyHitRestoresPersistedTimingAndSourceClientContext() { + var memoryId = new com.openmemind.ai.memory.core.data.DefaultMemoryId("test", "agent"); + var convProcessor = mock(ConversationContentProcessor.class); + when(convProcessor.contentClass()).thenReturn(ConversationContent.class); + var layer = new RawDataLayer(List.of(convProcessor), defaultCaption, store, vector, 64); + var content = + new ConversationContent(List.of(Message.user("remember espresso"))) { + @Override + public String getContentId() { + return "content-timed"; + } + }; + MemoryRawData existing = + new MemoryRawData( + "rd-timed", + memoryId.toIdentifier(), + ConversationContent.TYPE, + "claude-code", + "content-timed", + Segment.single("remember espresso"), + "espresso caption", + "vec-1", + Map.of("sourceClient", "claude-code"), + null, + null, + Instant.parse("2026-05-11T00:00:00Z"), + Instant.parse("2026-05-10T09:30:00Z"), + Instant.parse("2026-05-10T09:31:00Z")); + when(rawDataOps.getRawDataByContentId(memoryId, "content-timed")) + .thenReturn(Optional.of(existing)); + when(rawDataOps.listRawDataByContentId(memoryId, "content-timed")) + .thenReturn(List.of(existing)); + + var result = + layer.extract(memoryId, content, ConversationContent.TYPE, Map.of(), null).block(); + + assertThat(result.existed()).isTrue(); + assertThat(result.segments()).hasSize(1); + var context = result.segments().get(0).runtimeContext(); + assertThat(context).isNotNull(); + assertThat(context.startTime()).isEqualTo(Instant.parse("2026-05-10T09:30:00Z")); + assertThat(context.observedAt()).isEqualTo(Instant.parse("2026-05-10T09:31:00Z")); + assertThat(context.sourceClient()).isEqualTo("claude-code"); + } + @Test @DisplayName("processSegment keeps runtime context transient while using it for rawdata timing") void processSegmentKeepsRuntimeContextTransientWhileUsingItForRawdataTiming() { @@ -904,6 +1082,28 @@ private static RawContentProcessor mockTestDocumentProcesso return (RawContentProcessor) mock(RawContentProcessor.class); } + private static MemoryRawData rawData( + com.openmemind.ai.memory.core.data.MemoryId memoryId, + String id, + String contentId, + Segment segment, + String caption) { + return new MemoryRawData( + id, + memoryId.toIdentifier(), + ConversationContent.TYPE, + contentId, + segment, + caption, + null, + Map.of(), + null, + null, + Instant.parse("2026-05-11T00:00:00Z"), + null, + null); + } + private static final class TestPluginRawContent extends RawContent { @Override diff --git a/memind-integrations/claude-code/README.md b/memind-integrations/claude-code/README.md index 9a443ee6..b0d83869 100644 --- a/memind-integrations/claude-code/README.md +++ b/memind-integrations/claude-code/README.md @@ -1,8 +1,8 @@ # Memind Claude Code Integration Memind adds persistent project memory to Claude Code. The plugin retrieves relevant Memind context before each -user prompt and appends Claude Code conversation messages to Memind's conversation buffer during session -lifecycle hooks. +user prompt and submits Claude Code conversation messages through Memind's reliable extraction endpoint during +session lifecycle hooks. Use this plugin when you want Claude Code to remember project facts, preferences, implementation decisions, and previous discussions across sessions. The plugin connects Claude Code to an already-running Memind server; it @@ -20,9 +20,8 @@ The integration is intentionally small: - **Retrieval**: `UserPromptSubmit` calls Memind `/open/v1/memory/retrieve` and injects relevant memories into Claude Code as `...` additional context. - **Ingestion**: `Stop`, `PreCompact`, and `SessionEnd` read the Claude Code transcript, filter - user/assistant messages, and send new messages to Memind `/open/v1/memory/add-message`. -- **Commit on boundaries**: `PreCompact` and `SessionEnd` commit by default, so Memind can extract memories at - natural context/session boundaries. + user/assistant messages, and submit a caller-owned conversation payload to Memind + `/open/v1/memory/extract/sync`. - **Retry**: failed ingestion payloads are spooled under `~/.memind/claude-code/retry/` and replayed on later `SessionStart` hooks. - **Source tagging**: all requests use `sourceClient = "claude-code"` by default, so Memind can distinguish @@ -93,9 +92,9 @@ The installed hooks are: | --- | --- | ---: | --- | | `SessionStart` | `scripts/session_start.py` | 5s | Health check, replay at most one failed retry payload, and clean old state. | | `UserPromptSubmit` | `scripts/retrieve.py` | 12s | Retrieve relevant Memind context for the current user prompt. | -| `PreCompact` | `scripts/pre_compact.py` | 30s | Submit recent transcript messages and commit before context compaction. | -| `Stop` | `scripts/ingest.py` | 15s | Append new transcript messages to Memind after a turn. | -| `SessionEnd` | `scripts/session_end.py` | 10s | Submit remaining transcript messages and commit at session end. | +| `PreCompact` | `scripts/pre_compact.py` | 30s | Submit recent transcript messages through reliable extraction before context compaction. | +| `Stop` | `scripts/ingest.py` | 15s | Submit new transcript messages through reliable extraction after a turn. | +| `SessionEnd` | `scripts/session_end.py` | 10s | Submit remaining transcript messages through reliable extraction at session end. | `Stop` is configured as async so regular turn completion stays fast. `PreToolUse` and `PostToolUse` are intentionally unused in v0.1 because tool-call memory needs an explicit privacy and data-model design. @@ -114,6 +113,7 @@ User configuration is optional. Save overrides as `~/.memind/claude-code.json`: "agentId": "claude-code", "agentIdMode": "project", "sourceClient": "claude-code", + "ingestionMode": "extract-sync", "preCompactCommit": true, "commitOnSessionEnd": true, "retrieveContextTurns": 0 @@ -143,12 +143,13 @@ Settings are loaded in this order: | `retrieveMaxEntries` | `8` | Maximum formatted memory entries injected into Claude Code. | | `retrieveMaxChars` | `6000` | Maximum injected context characters. | | `retrieveContextTurns` | `0` | Number of recent transcript turns to include in the retrieval query. | +| `ingestionMode` | `extract-sync` | Default reliable ingestion mode. | | `ingestionRoles` | `["user", "assistant"]` | Transcript roles eligible for ingestion. | | `ingestionMaxMessagesPerHook` | `20` | Maximum new messages sent during one regular ingestion hook. | -| `preCompactCommit` | `true` | Commits after successful `PreCompact` ingestion. | +| `preCompactCommit` | `true` | Compatibility flag for server-buffer ingestion mode; ignored by the default reliable mode. | | `preCompactMaxMessages` | `20` | Maximum messages submitted during one `PreCompact` hook. | -| `commitOnSessionEnd` | `true` | Commits after successful `SessionEnd` ingestion. | -| `ingestRetrySpool` | `true` | Enables file-backed retry for failed ingestion and commit payloads. | +| `commitOnSessionEnd` | `true` | Compatibility flag for server-buffer ingestion mode; ignored by the default reliable mode. | +| `ingestRetrySpool` | `true` | Enables file-backed retry for failed extraction payloads. | | `debug` | `false` | Writes debug logs to `~/.memind/claude-code.log`. | ### Environment Overrides @@ -162,6 +163,7 @@ export MEMIND_USER_ID=local__alice export MEMIND_AGENT_ID=claude-code export MEMIND_AGENT_ID_MODE=project export MEMIND_SOURCE_CLIENT=claude-code +export MEMIND_INGESTION_MODE=extract-sync export MEMIND_PRE_COMPACT_COMMIT=true export MEMIND_COMMIT_ON_SESSION_END=true export MEMIND_RETRIEVE_CONTEXT_TURNS=0 @@ -227,13 +229,14 @@ The ingestion flow: 3. Strips previously injected `` blocks to avoid feedback loops. 4. Skips tool/event payloads, unsupported roles, and Claude Code interruption placeholders. 5. Computes stable fingerprints and sends only messages that have not already been submitted. -6. Retries each failed `add-message` once before writing a retry payload. +6. Builds one caller-owned conversation raw-content payload and submits it to `/open/v1/memory/extract/sync`. -Accepted fingerprints are persisted immediately, so partial progress is preserved even if later messages fail. -Unsubmitted messages are retried by future hooks. +The local retry spool stores the full extraction payload plus the covered message fingerprints. Fingerprints are +marked submitted only after Memind returns `SUCCESS`; `PARTIAL_SUCCESS` and failures keep the payload available +for later replay. -`Stop` only appends messages by default. `PreCompact` and `SessionEnd` append messages and then commit by -default, which gives Memind a natural boundary for extraction without forcing extraction after every turn. +Commit flags apply only to explicit server-buffer mode. In the default reliable mode, hooks do not issue an +additional `/commit` call after successful `/extract/sync`. ## Verify Installation @@ -285,21 +288,8 @@ Then start a new Claude Code session and say something specific: Please remember: the Memind Claude Code smoke test topic is blue-lake-42. ``` -After Claude Code responds, the `Stop` hook should append the turn to Memind's conversation buffer. To make the -message available for retrieval, either end the Claude Code session so `SessionEnd` commits, or manually commit: - -```bash -curl -fsSL -X POST http://127.0.0.1:8366/open/v1/memory/commit \ - -H 'Content-Type: application/json' \ - -d '{ - "userId": "local__memind-smoke", - "agentId": "claude-code-smoke", - "sourceClient": "claude-code" - }' -``` - -Commit requests extraction. Depending on server configuration, extraction may finish asynchronously; wait until -the Memind server logs show memory item or insight extraction completing. +After Claude Code responds, the `Stop` hook should submit the turn to Memind via `/open/v1/memory/extract/sync`. +No manual commit is needed for the default reliable path. Finally, verify retrieval: @@ -411,11 +401,10 @@ curl -fsSL http://127.0.0.1:8366/open/v1/health ### New memories are not immediately retrieved -Raw messages are appended first. They become retrievable after Memind commits and extracts memory items or -insights. By default, Claude Code commits on `PreCompact` and `SessionEnd`, not after every `Stop` hook. - -If you need faster availability during testing, manually commit the same `userId` and `agentId` through Memind's -API or end the Claude Code session to trigger `SessionEnd`. +The default reliable mode submits transcript batches through `/open/v1/memory/extract/sync`, so a `SUCCESS` +response means extraction finished for that batch. If retrieval still does not surface the expected memory, +confirm the same `userId` and `agentId` are used for ingestion and retrieval, then inspect +`~/.memind/claude-code.log` with `MEMIND_DEBUG=true`. ### Duplicate messages appear diff --git a/memind-integrations/claude-code/scripts/ingest.py b/memind-integrations/claude-code/scripts/ingest.py index 4f403b85..ae5d4925 100644 --- a/memind-integrations/claude-code/scripts/ingest.py +++ b/memind-integrations/claude-code/scripts/ingest.py @@ -16,7 +16,6 @@ import json import os import sys -import time from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) @@ -38,6 +37,33 @@ def retry_root(): return Path.home() / ".memind" / "claude-code" / "retry" +def _message_payload(raw_message): + return {key: value for key, value in raw_message.items() if key != "fingerprint"} + + +def _extract_payload(messages): + return { + "type": "conversation", + "messages": [_message_payload(message) for message in messages], + } + + +def _spool_extract(retry_spool, identity, source_client, session_id, messages): + if retry_spool is None or not messages: + return + retry_spool.enqueue( + { + "kind": "extract", + "userId": identity["userId"], + "agentId": identity["agentId"], + "sourceClient": source_client, + "sessionId": session_id, + "fingerprints": [message["fingerprint"] for message in messages], + "rawContent": _extract_payload(messages), + } + ) + + def ingest_messages(config, hook_input, commit=False, max_messages=None): identity = resolve_identity(config, hook_input) client = MemindClient(config["memindApiUrl"], config.get("memindApiToken"), timeout=10) @@ -53,46 +79,26 @@ def ingest_messages(config, hook_input, commit=False, max_messages=None): source_client = config.get("sourceClient") with store.locked(session_id) as state: new_messages = [message for message in messages if not state.is_submitted(message["fingerprint"])] - for raw_message in new_messages[:limit]: - fingerprint = raw_message["fingerprint"] - message = {key: value for key, value in raw_message.items() if key != "fingerprint"} + selected = new_messages[:limit] + if selected: + response = None try: - client.add_message(identity["userId"], identity["agentId"], message, source_client) - submitted.append(fingerprint) + response = client.extract( + identity["userId"], + identity["agentId"], + _extract_payload(selected), + source_client, + ) except Exception: - try: - time.sleep(0.5) - client.add_message(identity["userId"], identity["agentId"], message, source_client) - submitted.append(fingerprint) - except Exception: - if retry_spool is not None: - retry_spool.enqueue( - { - "kind": "add-message", - "userId": identity["userId"], - "agentId": identity["agentId"], - "message": message, - "sourceClient": source_client, - "sessionId": session_id, - "fingerprint": fingerprint, - } - ) + _spool_extract(retry_spool, identity, source_client, session_id, selected) + else: + status = ((response or {}).get("data") or {}).get("status") + if status == "SUCCESS": + submitted = [message["fingerprint"] for message in selected] + else: + _spool_extract(retry_spool, identity, source_client, session_id, selected) state.mark_submitted(submitted) committed = False - if commit: - try: - client.commit(identity["userId"], identity["agentId"], source_client) - committed = True - except Exception: - if retry_spool is not None: - retry_spool.enqueue( - { - "kind": "commit", - "userId": identity["userId"], - "agentId": identity["agentId"], - "sourceClient": source_client, - } - ) return {"submitted": len(submitted), "committed": committed} diff --git a/memind-integrations/claude-code/scripts/lib/client.py b/memind-integrations/claude-code/scripts/lib/client.py index 89536282..a931850f 100644 --- a/memind-integrations/claude-code/scripts/lib/client.py +++ b/memind-integrations/claude-code/scripts/lib/client.py @@ -78,6 +78,17 @@ def add_message(self, user_id, agent_id, message, source_client=None): payload, ) + def extract(self, user_id, agent_id, raw_content, source_client=None): + payload = {"userId": user_id, "agentId": agent_id, "rawContent": raw_content} + if source_client: + payload["sourceClient"] = source_client + return self._request( + "POST", + "/open/v1/memory/extract/sync", + payload, + require_data=True, + ) + def commit(self, user_id, agent_id, source_client=None): payload = {"userId": user_id, "agentId": agent_id} if source_client: diff --git a/memind-integrations/claude-code/scripts/lib/config.py b/memind-integrations/claude-code/scripts/lib/config.py index 6b93e59e..8286ea1f 100644 --- a/memind-integrations/claude-code/scripts/lib/config.py +++ b/memind-integrations/claude-code/scripts/lib/config.py @@ -30,7 +30,7 @@ "retrieveMaxChars": 6000, "retrievePromptPreamble": "Relevant memories from Memind. Use only when directly helpful:", "retrieveContextTurns": 0, - "ingestionMode": "add-message", + "ingestionMode": "extract-sync", "ingestionRoles": ["user", "assistant"], "ingestionMaxMessagesPerHook": 20, "preCompactCommit": True, diff --git a/memind-integrations/claude-code/scripts/session_start.py b/memind-integrations/claude-code/scripts/session_start.py index 57bc8831..b1b5c4a2 100644 --- a/memind-integrations/claude-code/scripts/session_start.py +++ b/memind-integrations/claude-code/scripts/session_start.py @@ -52,7 +52,21 @@ def main(): if claimed: payload = spool.load_claimed(claimed) replay_client = MemindClient(config["memindApiUrl"], config.get("memindApiToken"), timeout=10) - if payload.get("kind") == "add-message": + if payload.get("kind") == "extract": + response = replay_client.extract( + payload["userId"], + payload["agentId"], + payload["rawContent"], + payload.get("sourceClient"), + ) + status = ((response or {}).get("data") or {}).get("status") + if status != "SUCCESS": + raise RuntimeError(f"extract replay did not fully succeed: {status}") + if payload.get("sessionId") and payload.get("fingerprints"): + with SessionStateStore(state_root()).locked(payload["sessionId"]) as state: + state.mark_submitted(payload["fingerprints"]) + spool.complete(claimed) + elif payload.get("kind") == "add-message": replay_client.add_message( payload["userId"], payload["agentId"], diff --git a/memind-integrations/claude-code/settings.json b/memind-integrations/claude-code/settings.json index e3a6efaa..f1db3037 100644 --- a/memind-integrations/claude-code/settings.json +++ b/memind-integrations/claude-code/settings.json @@ -12,7 +12,7 @@ "retrieveMaxChars": 6000, "retrievePromptPreamble": "Relevant memories from Memind. Use only when directly helpful:", "retrieveContextTurns": 0, - "ingestionMode": "add-message", + "ingestionMode": "extract-sync", "ingestionRoles": ["user", "assistant"], "ingestionMaxMessagesPerHook": 20, "preCompactCommit": true, diff --git a/memind-integrations/claude-code/tests/test_client.py b/memind-integrations/claude-code/tests/test_client.py index 30f2a43c..bf01760c 100644 --- a/memind-integrations/claude-code/tests/test_client.py +++ b/memind-integrations/claude-code/tests/test_client.py @@ -83,6 +83,38 @@ def test_add_message_and_commit(self): self.assertEqual(Handler.requests[1][1], "/open/v1/memory/commit") self.assertEqual(Handler.requests[1][2]["sourceClient"], "claude-code") + def test_extract_sync_sends_raw_content_and_returns_data(self): + Handler.responses.append( + ( + 200, + { + "code": "success", + "data": { + "status": "SUCCESS", + "rawDataIds": ["rd-1"], + "itemIds": [], + "insightIds": [], + "insightPending": False, + }, + }, + ) + ) + client = MemindClient(self.base_url, timeout=2) + response = client.extract( + "u", + "a", + { + "type": "conversation", + "messages": [ + {"role": "USER", "content": [{"type": "text", "text": "hello"}]} + ], + }, + "claude-code", + ) + self.assertEqual(response["data"]["status"], "SUCCESS") + self.assertEqual(Handler.requests[0][1], "/open/v1/memory/extract/sync") + self.assertEqual(Handler.requests[0][2]["sourceClient"], "claude-code") + def test_retrieve_requires_data(self): Handler.responses.append((200, {"code": "success", "data": {"items": [], "insights": []}})) client = MemindClient(self.base_url, timeout=2) diff --git a/memind-integrations/claude-code/tests/test_config.py b/memind-integrations/claude-code/tests/test_config.py index 439d0dbe..ba24906a 100644 --- a/memind-integrations/claude-code/tests/test_config.py +++ b/memind-integrations/claude-code/tests/test_config.py @@ -42,7 +42,7 @@ def test_parse_list(self): def test_defaults_match_spec(self): self.assertEqual(DEFAULT_SETTINGS["retrieveContextTurns"], 0) - self.assertEqual(DEFAULT_SETTINGS["ingestionMode"], "add-message") + self.assertEqual(DEFAULT_SETTINGS["ingestionMode"], "extract-sync") self.assertEqual(DEFAULT_SETTINGS["sourceClient"], "claude-code") self.assertEqual(DEFAULT_SETTINGS["ingestionMaxMessagesPerHook"], 20) self.assertEqual(DEFAULT_SETTINGS["stateMaxAgeDays"], 14) diff --git a/memind-integrations/claude-code/tests/test_hooks.py b/memind-integrations/claude-code/tests/test_hooks.py index e3aa99ce..063c9438 100644 --- a/memind-integrations/claude-code/tests/test_hooks.py +++ b/memind-integrations/claude-code/tests/test_hooks.py @@ -18,6 +18,7 @@ import tempfile import unittest from pathlib import Path +from unittest import mock ROOT = Path(__file__).resolve().parents[1] @@ -106,6 +107,214 @@ def test_session_start_fail_open_when_memind_unavailable(self): output = self.run_hook("session_start.py", {"cwd": tmp, "session_id": "s1"}, env=env) self.assertEqual(output, {"continue": True, "suppressOutput": True}) + def test_ingest_uses_extract_sync_payload_and_marks_submitted_only_on_success(self): + sys.path.insert(0, str(ROOT / "scripts")) + import ingest + + with tempfile.NamedTemporaryFile("w", delete=False) as handle: + handle.write( + json.dumps( + { + "type": "user", + "uuid": "msg-1", + "timestamp": "2026-05-11T00:00:00Z", + "message": {"content": "remember espresso"}, + } + ) + + "\n" + ) + transcript = Path(handle.name) + try: + config = { + "memindApiUrl": "http://127.0.0.1:8366", + "memindApiToken": None, + "autoIngest": True, + "ingestionRoles": ["user", "assistant"], + "ingestionMaxMessagesPerHook": 20, + "ingestRetrySpool": True, + "sourceClient": "claude-code", + "agentId": "claude-code", + "agentIdMode": "global", + "userId": "u", + } + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.object(ingest, "state_root", return_value=Path(tmp) / "state"): + with mock.patch.object(ingest, "retry_root", return_value=Path(tmp) / "retry"): + with mock.patch.object(ingest, "MemindClient") as client_cls: + client = client_cls.return_value + client.extract.return_value = {"data": {"status": "SUCCESS"}} + result = ingest.ingest_messages( + config, + { + "session_id": "s1", + "transcript_path": str(transcript), + "cwd": tmp, + }, + commit=True, + ) + self.assertEqual(result["submitted"], 1) + self.assertFalse(result["committed"]) + client.extract.assert_called_once() + client.commit.assert_not_called() + raw_content = client.extract.call_args.args[2] + self.assertEqual(raw_content["type"], "conversation") + self.assertEqual(raw_content["messages"][0]["role"], "USER") + finally: + transcript.unlink() + + def test_ingest_spools_full_extract_payload_on_partial_success(self): + sys.path.insert(0, str(ROOT / "scripts")) + import ingest + + with tempfile.NamedTemporaryFile("w", delete=False) as handle: + handle.write( + json.dumps( + { + "type": "user", + "uuid": "msg-1", + "timestamp": "2026-05-11T00:00:00Z", + "message": {"content": "remember espresso"}, + } + ) + + "\n" + ) + transcript = Path(handle.name) + try: + config = { + "memindApiUrl": "http://127.0.0.1:8366", + "memindApiToken": None, + "autoIngest": True, + "ingestionRoles": ["user", "assistant"], + "ingestionMaxMessagesPerHook": 20, + "ingestRetrySpool": True, + "sourceClient": "claude-code", + "agentId": "claude-code", + "agentIdMode": "global", + "userId": "u", + } + with tempfile.TemporaryDirectory() as tmp: + retry_dir = Path(tmp) / "retry" + with mock.patch.object(ingest, "state_root", return_value=Path(tmp) / "state"): + with mock.patch.object(ingest, "retry_root", return_value=retry_dir): + with mock.patch.object(ingest, "MemindClient") as client_cls: + client = client_cls.return_value + client.extract.return_value = { + "data": { + "status": "PARTIAL_SUCCESS", + "errorMessage": "partial", + } + } + result = ingest.ingest_messages( + config, + { + "session_id": "s1", + "transcript_path": str(transcript), + "cwd": tmp, + }, + commit=False, + ) + payload = json.loads(next(retry_dir.glob("*.json")).read_text()) + self.assertEqual(result["submitted"], 0) + self.assertEqual(payload["kind"], "extract") + self.assertEqual(payload["rawContent"]["type"], "conversation") + self.assertEqual(len(payload["fingerprints"]), 1) + finally: + transcript.unlink() + + def test_session_start_replays_extract_payload_and_marks_fingerprints(self): + sys.path.insert(0, str(ROOT / "scripts")) + import session_start + from scripts.lib.retry import RetrySpool + from scripts.lib.state import SessionStateStore + + with tempfile.TemporaryDirectory() as tmp: + retry_dir = Path(tmp) / "retry" + state_dir = Path(tmp) / "state" + RetrySpool(retry_dir).enqueue( + { + "kind": "extract", + "userId": "u", + "agentId": "a", + "sourceClient": "claude-code", + "sessionId": "s1", + "fingerprints": ["fp1"], + "rawContent": { + "type": "conversation", + "messages": [ + {"role": "USER", "content": [{"type": "text", "text": "hello"}]} + ], + }, + } + ) + config = { + "memindApiUrl": "http://127.0.0.1:8366", + "memindApiToken": None, + "ingestRetryMaxFiles": 20, + "ingestRetryMaxAgeDays": 7, + "stateMaxAgeDays": 14, + "debug": False, + } + with mock.patch.object(session_start, "load_config", return_value=config): + with mock.patch.object(session_start, "retry_root", return_value=retry_dir): + with mock.patch.object(session_start, "state_root", return_value=state_dir): + with mock.patch.object(session_start, "MemindClient") as client_cls: + client = client_cls.return_value + client.health.return_value = {"data": {"status": "UP"}} + client.extract.return_value = {"data": {"status": "SUCCESS"}} + session_start.main() + with SessionStateStore(state_dir).locked("s1") as state: + self.assertTrue(state.is_submitted("fp1")) + self.assertEqual(list(retry_dir.glob("*.json")), []) + client.extract.assert_called_once() + client.add_message.assert_not_called() + client.commit.assert_not_called() + + def test_session_start_keeps_extract_payload_when_replay_is_not_success(self): + sys.path.insert(0, str(ROOT / "scripts")) + import session_start + from scripts.lib.retry import RetrySpool + from scripts.lib.state import SessionStateStore + + with tempfile.TemporaryDirectory() as tmp: + retry_dir = Path(tmp) / "retry" + state_dir = Path(tmp) / "state" + RetrySpool(retry_dir).enqueue( + { + "kind": "extract", + "userId": "u", + "agentId": "a", + "sourceClient": "claude-code", + "sessionId": "s1", + "fingerprints": ["fp1"], + "rawContent": { + "type": "conversation", + "messages": [ + {"role": "USER", "content": [{"type": "text", "text": "hello"}]} + ], + }, + } + ) + config = { + "memindApiUrl": "http://127.0.0.1:8366", + "memindApiToken": None, + "ingestRetryMaxFiles": 20, + "ingestRetryMaxAgeDays": 7, + "stateMaxAgeDays": 14, + "debug": False, + } + with mock.patch.object(session_start, "load_config", return_value=config): + with mock.patch.object(session_start, "retry_root", return_value=retry_dir): + with mock.patch.object(session_start, "state_root", return_value=state_dir): + with mock.patch.object(session_start, "MemindClient") as client_cls: + client = client_cls.return_value + client.health.return_value = {"data": {"status": "UP"}} + client.extract.return_value = {"data": {"status": "PARTIAL_SUCCESS"}} + session_start.main() + with SessionStateStore(state_dir).locked("s1") as state: + self.assertFalse(state.is_submitted("fp1")) + self.assertEqual(len(list(retry_dir.glob("*.json"))), 1) + client.extract.assert_called_once() + if __name__ == "__main__": unittest.main() diff --git a/memind-integrations/claude-code/tests/test_manifest.py b/memind-integrations/claude-code/tests/test_manifest.py index decffd0d..1351a641 100644 --- a/memind-integrations/claude-code/tests/test_manifest.py +++ b/memind-integrations/claude-code/tests/test_manifest.py @@ -39,7 +39,7 @@ def test_hooks_json_shape(self): def test_default_settings(self): settings = json.loads((ROOT / "settings.json").read_text()) self.assertEqual(settings["retrieveContextTurns"], 0) - self.assertEqual(settings["ingestionMode"], "add-message") + self.assertEqual(settings["ingestionMode"], "extract-sync") self.assertEqual(settings["ingestionMaxMessagesPerHook"], 20) self.assertEqual(settings["stateMaxAgeDays"], 14) diff --git a/memind-integrations/codex/README.md b/memind-integrations/codex/README.md index 65d3be1b..efd78e33 100644 --- a/memind-integrations/codex/README.md +++ b/memind-integrations/codex/README.md @@ -1,7 +1,8 @@ # Memind Codex Integration Memind adds persistent project memory to Codex CLI. The integration retrieves relevant Memind context before -each user prompt and appends Codex conversation messages to Memind's conversation buffer after each turn. +each user prompt and submits Codex conversation messages through Memind's reliable extraction endpoint after +each turn. Use this integration when you want Codex to remember project facts, preferences, and previous decisions across sessions. The plugin connects Codex to an already-running Memind server; it does not start the server itself. @@ -17,8 +18,8 @@ The integration is intentionally small: - **Retrieval**: `UserPromptSubmit` calls Memind `/open/v1/memory/retrieve` and injects relevant memories into the Codex prompt as `...`. -- **Ingestion**: `Stop` reads the Codex transcript, filters user/assistant messages, and sends new messages to - Memind `/open/v1/memory/add-message`. +- **Ingestion**: `Stop` reads the Codex transcript, filters user/assistant messages, and submits a caller-owned + conversation payload to Memind `/open/v1/memory/extract/sync`. - **Retry**: failed ingestion batches are spooled under `~/.memind/codex/retry/` and replayed on the next `SessionStart`. - **Source tagging**: all requests use `sourceClient = "codex"` by default, so Memind can distinguish Codex @@ -107,7 +108,7 @@ The installed hooks are: | --- | --- | ---: | --- | | `SessionStart` | `scripts/session_start.py` | 5s | Replay at most one failed ingestion batch and clean old state. | | `UserPromptSubmit` | `scripts/retrieve.py` | 12s | Retrieve relevant Memind context for the current user prompt. | -| `Stop` | `scripts/ingest.py` | 15s | Append new Codex transcript messages to Memind. | +| `Stop` | `scripts/ingest.py` | 15s | Submit new Codex transcript messages through reliable extraction. | `PreToolUse`, `PostToolUse`, and `PermissionRequest` are intentionally unused in v0.1. Tool-call memory can be added later after the data model and privacy behavior are explicitly designed. @@ -126,6 +127,7 @@ User configuration is optional. Save overrides as `~/.memind/codex.json`: "agentId": "codex", "agentIdMode": "project", "sourceClient": "codex", + "ingestionMode": "extract-sync", "commitOnStop": false, "retrieveContextTurns": 0 } @@ -149,11 +151,12 @@ Settings are loaded in this order: | `sourceClient` | `codex` | Source marker stored with Memind data. | | `autoRetrieve` | `true` | Enables prompt-time memory retrieval. | | `autoIngest` | `true` | Enables transcript ingestion after Codex turns. | -| `commitOnStop` | `false` | When true, commits after each successful Stop ingestion. | +| `commitOnStop` | `false` | Compatibility flag for server-buffer ingestion mode; ignored by the default reliable mode. | | `retrieveStrategy` | `SIMPLE` | Memind retrieval strategy. | | `retrieveMaxEntries` | `8` | Maximum formatted memory entries injected into Codex. | | `retrieveMaxChars` | `6000` | Maximum injected context characters. | | `retrieveContextTurns` | `0` | Number of recent transcript turns to include in the retrieval query. | +| `ingestionMode` | `extract-sync` | Default reliable ingestion mode. | | `ingestionRoles` | `["user", "assistant"]` | Transcript roles eligible for ingestion. | | `ingestionMaxMessagesPerHook` | `20` | Maximum new messages sent during one Stop hook. | | `ingestRetrySpool` | `true` | Enables file-backed retry for failed ingestion. | @@ -233,13 +236,14 @@ The Stop hook: 3. Strips previously injected `` blocks to avoid feedback loops. 4. Skips tool/event payloads and Codex control context blocks. 5. Computes stable fingerprints and sends only messages that have not already been submitted. -6. Retries each failed `add-message` once before writing a retry payload. +6. Builds one caller-owned conversation raw-content payload and submits it to `/open/v1/memory/extract/sync`. -Accepted fingerprints are persisted immediately, so partial progress is preserved even if later messages fail. -Unsubmitted messages are retried by future hooks. +The local retry spool stores the full extraction payload plus the covered message fingerprints. Fingerprints are +marked submitted only after Memind returns `SUCCESS`; `PARTIAL_SUCCESS` and failures keep the payload available +for later replay. -`commitOnStop` defaults to `false`. This lets Memind's own boundary detector decide when to extract memory. -Set it to `true` only when you want faster cross-session availability and accept more frequent extraction work. +Commit flags apply only to explicit server-buffer mode. In the default reliable mode, hooks do not issue an +additional `/commit` call after successful `/extract/sync`. ## Verify Installation @@ -357,7 +361,7 @@ curl -fsSL http://127.0.0.1:8366/open/v1/health - Confirm Memind server responds quickly. - Reduce `ingestionMaxMessagesPerHook`. -- Keep `commitOnStop = false` unless immediate extraction is required. +- Keep the default `extract-sync` mode and reduce batch size before increasing hook timeout. ### Duplicate messages appear @@ -371,5 +375,5 @@ The integration uses per-session fingerprints stored under `~/.memind/codex/stat - v0.1 supports conversation memory only; tool calls are not ingested. - Retrieval quality depends on existing extracted Memind items and insights. -- `commitOnStop = false` means newly appended messages may not become retrievable until Memind naturally commits - and extracts them. +- `commitOnStop` applies only to compatibility server-buffer ingestion mode and is ignored by the default + reliable extraction mode. diff --git a/memind-integrations/codex/scripts/ingest.py b/memind-integrations/codex/scripts/ingest.py index 2be9568b..83a86ef1 100644 --- a/memind-integrations/codex/scripts/ingest.py +++ b/memind-integrations/codex/scripts/ingest.py @@ -3,7 +3,6 @@ import json import os import sys -import time from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) @@ -29,46 +28,29 @@ def _message_payload(raw_message): return {key: value for key, value in raw_message.items() if key != "fingerprint"} -def _operation(user_id, agent_id, message, source_client): +def _extract_payload(messages): return { - "kind": "add-message", - "userId": user_id, - "agentId": agent_id, - "sourceClient": source_client, - "message": message, + "type": "conversation", + "messages": [_message_payload(message) for message in messages], } -def _spool_batch(retry_spool, identity, source_client, session_key, messages, commit_on_success): +def _spool_extract(retry_spool, identity, source_client, session_key, messages): if retry_spool is None or not messages: return retry_spool.enqueue( { - "kind": "ingestion-batch", + "kind": "extract", "userId": identity["userId"], "agentId": identity["agentId"], "sourceClient": source_client, - "operations": [ - _operation(identity["userId"], identity["agentId"], _message_payload(message), source_client) - for message in messages - ], "sessionKey": session_key, "fingerprints": [message["fingerprint"] for message in messages], - "commitOnSuccess": bool(commit_on_success), + "rawContent": _extract_payload(messages), } ) -def _add_message_with_retry(client, user_id, agent_id, message, source_client): - try: - client.add_message(user_id, agent_id, message, source_client) - return True - except Exception: - time.sleep(0.5) - client.add_message(user_id, agent_id, message, source_client) - return True - - def ingest_messages(config, hook_input): identity = resolve_identity(config, hook_input) client = MemindClient(config["memindApiUrl"], config.get("memindApiToken"), timeout=10) @@ -82,51 +64,30 @@ def ingest_messages(config, hook_input): store = SessionStateStore(state_root()) session_key = state_key(hook_input) source_client = config.get("sourceClient") - commit_on_stop = bool(config.get("commitOnStop", False)) with store.locked(session_key) as state: selected = [message for message in messages if not state.is_submitted(message["fingerprint"])][:limit] submitted = [] - failed = False - for index, raw_message in enumerate(selected): - fingerprint = raw_message["fingerprint"] - with store.locked(session_key) as state: - if state.is_submitted(fingerprint): - continue - message = _message_payload(raw_message) + if selected: try: - _add_message_with_retry(client, identity["userId"], identity["agentId"], message, source_client) - store.mark_submitted(session_key, [fingerprint]) - submitted.append(fingerprint) - except Exception: - failed = True - _spool_batch( - retry_spool, - identity, + response = client.extract( + identity["userId"], + identity["agentId"], + _extract_payload(selected), source_client, - session_key, - selected[index:], - commit_on_success=commit_on_stop, ) - break - - committed = False - if commit_on_stop and submitted and not failed: - try: - client.commit(identity["userId"], identity["agentId"], source_client) - committed = True except Exception: - if retry_spool is not None: - retry_spool.enqueue( - { - "kind": "commit", - "userId": identity["userId"], - "agentId": identity["agentId"], - "sourceClient": source_client, - } - ) - return {"submitted": len(submitted), "committed": committed} + _spool_extract(retry_spool, identity, source_client, session_key, selected) + else: + status = ((response or {}).get("data") or {}).get("status") + if status == "SUCCESS": + submitted = [message["fingerprint"] for message in selected] + store.mark_submitted(session_key, submitted) + else: + _spool_extract(retry_spool, identity, source_client, session_key, selected) + + return {"submitted": len(submitted), "committed": False} def main(): diff --git a/memind-integrations/codex/scripts/lib/client.py b/memind-integrations/codex/scripts/lib/client.py index 04e390ed..1cdbeb43 100644 --- a/memind-integrations/codex/scripts/lib/client.py +++ b/memind-integrations/codex/scripts/lib/client.py @@ -60,6 +60,12 @@ def add_message(self, user_id, agent_id, message, source_client=None): payload["sourceClient"] = source_client return self._request("POST", "/open/v1/memory/add-message", payload) + def extract(self, user_id, agent_id, raw_content, source_client=None): + payload = {"userId": user_id, "agentId": agent_id, "rawContent": raw_content} + if source_client: + payload["sourceClient"] = source_client + return self._request("POST", "/open/v1/memory/extract/sync", payload, require_data=True) + def commit(self, user_id, agent_id, source_client=None): payload = {"userId": user_id, "agentId": agent_id} if source_client: diff --git a/memind-integrations/codex/scripts/lib/config.py b/memind-integrations/codex/scripts/lib/config.py index b84591bb..24a44e5a 100644 --- a/memind-integrations/codex/scripts/lib/config.py +++ b/memind-integrations/codex/scripts/lib/config.py @@ -17,7 +17,7 @@ "retrieveMaxChars": 6000, "retrievePromptPreamble": "Relevant memories from Memind. Use only when directly helpful:", "retrieveContextTurns": 0, - "ingestionMode": "add-message", + "ingestionMode": "extract-sync", "ingestionRoles": ["user", "assistant"], "ingestionMaxMessagesPerHook": 20, "stateMaxAgeDays": 14, diff --git a/memind-integrations/codex/scripts/session_start.py b/memind-integrations/codex/scripts/session_start.py index 5a82a03e..196256e3 100644 --- a/memind-integrations/codex/scripts/session_start.py +++ b/memind-integrations/codex/scripts/session_start.py @@ -54,6 +54,21 @@ def _replay_ingestion_batch(client, payload): def _replay_payload(client, payload): kind = payload.get("kind") + if kind == "extract": + response = client.extract( + payload["userId"], + payload["agentId"], + payload["rawContent"], + payload.get("sourceClient"), + ) + status = ((response or {}).get("data") or {}).get("status") + if status != "SUCCESS": + raise RuntimeError(f"extract replay did not fully succeed: {status}") + session_key = payload.get("sessionKey") + fingerprints = payload.get("fingerprints") or [] + if session_key and fingerprints: + SessionStateStore(state_root()).mark_submitted(session_key, fingerprints) + return len(fingerprints) if kind == "ingestion-batch": return _replay_ingestion_batch(client, payload) if kind == "commit": diff --git a/memind-integrations/codex/settings.json b/memind-integrations/codex/settings.json index b76b6405..22b59be0 100644 --- a/memind-integrations/codex/settings.json +++ b/memind-integrations/codex/settings.json @@ -13,7 +13,7 @@ "retrieveMaxChars": 6000, "retrievePromptPreamble": "Relevant memories from Memind. Use only when directly helpful:", "retrieveContextTurns": 0, - "ingestionMode": "add-message", + "ingestionMode": "extract-sync", "ingestionRoles": ["user", "assistant"], "ingestionMaxMessagesPerHook": 20, "stateMaxAgeDays": 14, diff --git a/memind-integrations/codex/tests/test_client.py b/memind-integrations/codex/tests/test_client.py index d7e299f4..e82206af 100644 --- a/memind-integrations/codex/tests/test_client.py +++ b/memind-integrations/codex/tests/test_client.py @@ -42,6 +42,36 @@ def test_commit_sends_source_client(self, urlopen): payload = json.loads(urlopen.call_args.args[0].data.decode("utf-8")) self.assertEqual(payload, {"userId": "u", "agentId": "a", "sourceClient": "codex"}) + @mock.patch("urllib.request.urlopen") + def test_extract_sync_sends_source_client_and_raw_content(self, urlopen): + urlopen.return_value = _FakeResponse( + json.dumps( + { + "code": "success", + "data": { + "status": "SUCCESS", + "rawDataIds": ["rd-1"], + "itemIds": [], + "insightIds": [], + "insightPending": False, + }, + } + ) + ) + client = MemindClient("http://127.0.0.1:8366") + response = client.extract( + "u", + "a", + {"type": "conversation", "messages": [{"role": "USER", "content": []}]}, + "codex", + ) + request = urlopen.call_args.args[0] + payload = json.loads(request.data.decode("utf-8")) + self.assertTrue(request.full_url.endswith("/open/v1/memory/extract/sync")) + self.assertEqual(response["data"]["status"], "SUCCESS") + self.assertEqual(payload["sourceClient"], "codex") + self.assertEqual(payload["rawContent"]["type"], "conversation") + @mock.patch("urllib.request.urlopen") def test_retrieve_requires_data(self, urlopen): urlopen.return_value = _FakeResponse(json.dumps({"code": "success"})) diff --git a/memind-integrations/codex/tests/test_hooks.py b/memind-integrations/codex/tests/test_hooks.py index daa153ee..d88142cf 100644 --- a/memind-integrations/codex/tests/test_hooks.py +++ b/memind-integrations/codex/tests/test_hooks.py @@ -73,7 +73,7 @@ def test_ingest_without_transcript_fails_open(self): output = self.run_hook("ingest.py", {"cwd": tmp, "session_id": "s1"}, env=env) self.assertEqual(output, {"continue": True}) - def test_ingest_commits_only_when_enabled_and_all_selected_messages_succeed(self): + def test_ingest_uses_extract_sync_and_ignores_commit_flag_in_reliable_mode(self): sys.path.insert(0, str(ROOT / "scripts")) import ingest @@ -99,15 +99,17 @@ def test_ingest_commits_only_when_enabled_and_all_selected_messages_succeed(self with mock.patch.object(ingest, "retry_root", return_value=Path(tmp) / "retry"): with mock.patch.object(ingest, "MemindClient") as client_cls: client = client_cls.return_value + client.extract.return_value = {"data": {"status": "SUCCESS"}} result = ingest.ingest_messages(config, {"session_id": "s1", "transcript_path": str(transcript), "cwd": tmp}) self.assertEqual(result["submitted"], 1) - self.assertTrue(result["committed"]) - client.add_message.assert_called_once() - client.commit.assert_called_once() + self.assertFalse(result["committed"]) + client.extract.assert_called_once() + client.add_message.assert_not_called() + client.commit.assert_not_called() finally: transcript.unlink() - def test_ingest_does_not_commit_after_partial_append_failure(self): + def test_ingest_does_not_mark_submitted_or_commit_when_extract_fails_without_spool(self): sys.path.insert(0, str(ROOT / "scripts")) import ingest from scripts.lib.content import extract_messages @@ -136,19 +138,21 @@ def test_ingest_does_not_commit_after_partial_append_failure(self): with mock.patch.object(ingest, "retry_root", return_value=Path(tmp) / "retry"): with mock.patch.object(ingest, "MemindClient") as client_cls: client = client_cls.return_value - client.add_message.side_effect = [None, RuntimeError("down"), RuntimeError("down")] + client.extract.side_effect = RuntimeError("down") result = ingest.ingest_messages(config, {"session_id": "s1", "transcript_path": str(transcript), "cwd": tmp}) messages = extract_messages(transcript, ["user", "assistant"]) with SessionStateStore(Path(tmp) / "state").locked("s1") as state: - self.assertTrue(state.is_submitted(messages[0]["fingerprint"])) + self.assertFalse(state.is_submitted(messages[0]["fingerprint"])) self.assertFalse(state.is_submitted(messages[1]["fingerprint"])) - self.assertEqual(result["submitted"], 1) + self.assertEqual(result["submitted"], 0) self.assertFalse(result["committed"]) + client.extract.assert_called_once() + client.add_message.assert_not_called() client.commit.assert_not_called() finally: transcript.unlink() - def test_ingest_spools_failed_and_unattempted_messages(self): + def test_ingest_spools_failed_extract_payload(self): sys.path.insert(0, str(ROOT / "scripts")) import ingest @@ -176,21 +180,61 @@ def test_ingest_spools_failed_and_unattempted_messages(self): with mock.patch.object(ingest, "retry_root", return_value=retry_dir): with mock.patch.object(ingest, "MemindClient") as client_cls: client = client_cls.return_value - client.add_message.side_effect = [RuntimeError("down"), RuntimeError("down")] + client.extract.side_effect = RuntimeError("down") result = ingest.ingest_messages(config, {"session_id": "s1", "transcript_path": str(transcript), "cwd": tmp}) payload_files = list(retry_dir.glob("*.json")) self.assertEqual(len(payload_files), 1) payload = json.loads(payload_files[0].read_text()) self.assertEqual(result["submitted"], 0) self.assertFalse(result["committed"]) - self.assertEqual(payload["kind"], "ingestion-batch") - self.assertEqual(len(payload["operations"]), 2) + self.assertEqual(payload["kind"], "extract") + self.assertEqual(payload["rawContent"]["type"], "conversation") + self.assertEqual(len(payload["rawContent"]["messages"]), 2) self.assertEqual(len(payload["fingerprints"]), 2) - self.assertTrue(payload["commitOnSuccess"]) + self.assertNotIn("commitOnSuccess", payload) + client.extract.assert_called_once() + client.add_message.assert_not_called() client.commit.assert_not_called() finally: transcript.unlink() + def test_ingest_spools_full_extract_payload_on_partial_success(self): + sys.path.insert(0, str(ROOT / "scripts")) + import ingest + + with tempfile.NamedTemporaryFile("w", delete=False) as handle: + handle.write(json.dumps({"role": "user", "content": "first"}) + "\n") + transcript = Path(handle.name) + try: + config = { + "memindApiUrl": "http://127.0.0.1:8366", + "memindApiToken": None, + "autoIngest": True, + "ingestionRoles": ["user", "assistant"], + "ingestionMaxMessagesPerHook": 20, + "ingestRetrySpool": True, + "sourceClient": "codex", + "agentId": "codex", + "agentIdMode": "global", + "userId": "u", + "commitOnStop": False, + } + with tempfile.TemporaryDirectory() as tmp: + retry_dir = Path(tmp) / "retry" + with mock.patch.object(ingest, "state_root", return_value=Path(tmp) / "state"): + with mock.patch.object(ingest, "retry_root", return_value=retry_dir): + with mock.patch.object(ingest, "MemindClient") as client_cls: + client = client_cls.return_value + client.extract.return_value = {"data": {"status": "PARTIAL_SUCCESS"}} + result = ingest.ingest_messages(config, {"session_id": "s1", "transcript_path": str(transcript), "cwd": tmp}) + payload = json.loads(next(retry_dir.glob("*.json")).read_text()) + self.assertEqual(result["submitted"], 0) + self.assertEqual(payload["kind"], "extract") + self.assertEqual(payload["rawContent"]["type"], "conversation") + self.assertEqual(len(payload["fingerprints"]), 1) + finally: + transcript.unlink() + def test_session_start_fail_open_when_memind_unavailable(self): with tempfile.TemporaryDirectory() as tmp: env = { @@ -245,6 +289,98 @@ def test_session_start_replays_ingestion_batch_before_commit(self): method_names = [call[0] for call in client.method_calls] self.assertLess(method_names.index("add_message"), method_names.index("commit")) + def test_session_start_replays_extract_payload_and_marks_fingerprints(self): + sys.path.insert(0, str(ROOT / "scripts")) + import session_start + from scripts.lib.retry import RetrySpool + from scripts.lib.state import SessionStateStore + + with tempfile.TemporaryDirectory() as tmp: + retry_root = Path(tmp) / "retry" + state_root = Path(tmp) / "state" + RetrySpool(retry_root).enqueue( + { + "kind": "extract", + "userId": "u", + "agentId": "a", + "sourceClient": "codex", + "sessionKey": "s1", + "fingerprints": ["fp1"], + "rawContent": { + "type": "conversation", + "messages": [ + {"role": "USER", "content": [{"type": "text", "text": "hello"}]} + ], + }, + } + ) + config = { + "memindApiUrl": "http://127.0.0.1:8366", + "memindApiToken": None, + "ingestRetryMaxFiles": 20, + "ingestRetryMaxAgeDays": 7, + "stateMaxAgeDays": 14, + "debug": False, + } + with mock.patch.object(session_start, "retry_root", return_value=retry_root): + with mock.patch.object(session_start, "state_root", return_value=state_root): + with mock.patch.object(session_start, "MemindClient") as client_cls: + client = client_cls.return_value + client.health.return_value = {"data": {"status": "UP"}} + client.extract.return_value = {"data": {"status": "SUCCESS"}} + session_start.run_session_start(config) + with SessionStateStore(state_root).locked("s1") as state: + self.assertTrue(state.is_submitted("fp1")) + self.assertEqual(list(retry_root.glob("*.json")), []) + client.extract.assert_called_once() + client.add_message.assert_not_called() + client.commit.assert_not_called() + + def test_session_start_keeps_extract_payload_when_replay_is_not_success(self): + sys.path.insert(0, str(ROOT / "scripts")) + import session_start + from scripts.lib.retry import RetrySpool + from scripts.lib.state import SessionStateStore + + with tempfile.TemporaryDirectory() as tmp: + retry_root = Path(tmp) / "retry" + state_root = Path(tmp) / "state" + RetrySpool(retry_root).enqueue( + { + "kind": "extract", + "userId": "u", + "agentId": "a", + "sourceClient": "codex", + "sessionKey": "s1", + "fingerprints": ["fp1"], + "rawContent": { + "type": "conversation", + "messages": [ + {"role": "USER", "content": [{"type": "text", "text": "hello"}]} + ], + }, + } + ) + config = { + "memindApiUrl": "http://127.0.0.1:8366", + "memindApiToken": None, + "ingestRetryMaxFiles": 20, + "ingestRetryMaxAgeDays": 7, + "stateMaxAgeDays": 14, + "debug": False, + } + with mock.patch.object(session_start, "retry_root", return_value=retry_root): + with mock.patch.object(session_start, "state_root", return_value=state_root): + with mock.patch.object(session_start, "MemindClient") as client_cls: + client = client_cls.return_value + client.health.return_value = {"data": {"status": "UP"}} + client.extract.return_value = {"data": {"status": "PARTIAL_SUCCESS"}} + session_start.run_session_start(config) + with SessionStateStore(state_root).locked("s1") as state: + self.assertFalse(state.is_submitted("fp1")) + self.assertEqual(len(list(retry_root.glob("*.json"))), 1) + client.extract.assert_called_once() + def test_session_start_skips_replayed_message_already_submitted_by_later_stop(self): sys.path.insert(0, str(ROOT / "scripts")) import session_start diff --git a/memind-integrations/codex/tests/test_manifest.py b/memind-integrations/codex/tests/test_manifest.py index 9a236ded..69a1674c 100644 --- a/memind-integrations/codex/tests/test_manifest.py +++ b/memind-integrations/codex/tests/test_manifest.py @@ -33,7 +33,7 @@ def test_default_settings_match_spec(self): self.assertEqual(settings["sourceClient"], "codex") self.assertFalse(settings["commitOnStop"]) self.assertEqual(settings["retrieveContextTurns"], 0) - self.assertEqual(settings["ingestionMode"], "add-message") + self.assertEqual(settings["ingestionMode"], "extract-sync") self.assertEqual(settings["ingestionMaxMessagesPerHook"], 20) self.assertEqual(settings["stateMaxAgeDays"], 14) diff --git a/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-mysql/src/main/java/com/openmemind/ai/memory/plugin/jdbc/mysql/MysqlMemoryStore.java b/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-mysql/src/main/java/com/openmemind/ai/memory/plugin/jdbc/mysql/MysqlMemoryStore.java index 026e766a..1b1d0e88 100644 --- a/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-mysql/src/main/java/com/openmemind/ai/memory/plugin/jdbc/mysql/MysqlMemoryStore.java +++ b/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-mysql/src/main/java/com/openmemind/ai/memory/plugin/jdbc/mysql/MysqlMemoryStore.java @@ -286,6 +286,21 @@ public Optional getRawDataByContentId(MemoryId memoryId, String c contentId)); } + @Override + public List listRawDataByContentId(MemoryId memoryId, String contentId) { + ScopeContext scope = scopeOf(memoryId); + return queryList( + """ + SELECT * FROM memory_raw_data + WHERE user_id = ? AND agent_id = ? AND content_id = ? AND deleted = 0 + ORDER BY id ASC + """, + this::mapRawData, + scope.userId(), + scope.agentId(), + contentId); + } + @Override public List listRawData(MemoryId memoryId) { ScopeContext scope = scopeOf(memoryId); diff --git a/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-postgresql/src/main/java/com/openmemind/ai/memory/plugin/jdbc/postgresql/PostgresqlMemoryStore.java b/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-postgresql/src/main/java/com/openmemind/ai/memory/plugin/jdbc/postgresql/PostgresqlMemoryStore.java index ac63cfc9..ee8682a5 100644 --- a/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-postgresql/src/main/java/com/openmemind/ai/memory/plugin/jdbc/postgresql/PostgresqlMemoryStore.java +++ b/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-postgresql/src/main/java/com/openmemind/ai/memory/plugin/jdbc/postgresql/PostgresqlMemoryStore.java @@ -287,6 +287,21 @@ public Optional getRawDataByContentId(MemoryId memoryId, String c contentId)); } + @Override + public List listRawDataByContentId(MemoryId memoryId, String contentId) { + ScopeContext scope = scopeOf(memoryId); + return queryList( + """ + SELECT * FROM memory_raw_data + WHERE user_id = ? AND agent_id = ? AND content_id = ? AND deleted = FALSE + ORDER BY id ASC + """, + this::mapRawData, + scope.userId(), + scope.agentId(), + contentId); + } + @Override public List listRawData(MemoryId memoryId) { ScopeContext scope = scopeOf(memoryId); diff --git a/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-sqlite/src/main/java/com/openmemind/ai/memory/plugin/jdbc/sqlite/SqliteMemoryStore.java b/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-sqlite/src/main/java/com/openmemind/ai/memory/plugin/jdbc/sqlite/SqliteMemoryStore.java index ee646d95..53ff65e4 100644 --- a/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-sqlite/src/main/java/com/openmemind/ai/memory/plugin/jdbc/sqlite/SqliteMemoryStore.java +++ b/memind-plugins/memind-plugin-jdbc/memind-plugin-jdbc-sqlite/src/main/java/com/openmemind/ai/memory/plugin/jdbc/sqlite/SqliteMemoryStore.java @@ -297,6 +297,21 @@ public Optional getRawDataByContentId(MemoryId memoryId, String c contentId)); } + @Override + public List listRawDataByContentId(MemoryId memoryId, String contentId) { + ScopeContext scope = scopeOf(memoryId); + return queryList( + """ + SELECT * FROM memory_raw_data + WHERE user_id = ? AND agent_id = ? AND content_id = ? AND deleted = 0 + ORDER BY id ASC + """, + this::mapRawData, + scope.userId(), + scope.agentId(), + contentId); + } + @Override public List listRawData(MemoryId memoryId) { ScopeContext scope = scopeOf(memoryId); diff --git a/memind-plugins/memind-plugin-spring-boot-starters/memind-plugin-mybatis-plus-starter/src/main/java/com/openmemind/ai/memory/plugin/store/mybatis/MybatisPlusMemoryStore.java b/memind-plugins/memind-plugin-spring-boot-starters/memind-plugin-mybatis-plus-starter/src/main/java/com/openmemind/ai/memory/plugin/store/mybatis/MybatisPlusMemoryStore.java index ebfcdd06..0ac46bfe 100644 --- a/memind-plugins/memind-plugin-spring-boot-starters/memind-plugin-mybatis-plus-starter/src/main/java/com/openmemind/ai/memory/plugin/store/mybatis/MybatisPlusMemoryStore.java +++ b/memind-plugins/memind-plugin-spring-boot-starters/memind-plugin-mybatis-plus-starter/src/main/java/com/openmemind/ai/memory/plugin/store/mybatis/MybatisPlusMemoryStore.java @@ -438,6 +438,18 @@ public Optional getRawDataByContentId(MemoryId id, String content return Optional.ofNullable(dataObject).map(RawDataConverter::toRecord); } + @Override + public List listRawDataByContentId(MemoryId id, String contentId) { + return rawDataMapper + .selectList( + memoryQuery(id, MemoryRawDataDO.class) + .eq("content_id", contentId) + .orderByAsc("id")) + .stream() + .map(RawDataConverter::toRecord) + .toList(); + } + public List pollRawDataWithoutVector(MemoryId id, int limit, Duration minAge) { if (limit <= 0) { return List.of(); diff --git a/memind-server/src/main/java/com/openmemind/ai/memory/server/controller/openapi/OpenMemoryController.java b/memind-server/src/main/java/com/openmemind/ai/memory/server/controller/openapi/OpenMemoryController.java index 6663b717..04da1311 100644 --- a/memind-server/src/main/java/com/openmemind/ai/memory/server/controller/openapi/OpenMemoryController.java +++ b/memind-server/src/main/java/com/openmemind/ai/memory/server/controller/openapi/OpenMemoryController.java @@ -18,9 +18,13 @@ import com.openmemind.ai.memory.server.domain.memory.request.CommitMemoryRequest; import com.openmemind.ai.memory.server.domain.memory.request.ExtractMemoryRequest; import com.openmemind.ai.memory.server.domain.memory.request.RetrieveMemoryRequest; +import com.openmemind.ai.memory.server.domain.memory.response.AddMessageResponse; +import com.openmemind.ai.memory.server.domain.memory.response.ExtractMemoryResponse; import com.openmemind.ai.memory.server.domain.memory.response.RetrieveMemoryResponse; import com.openmemind.ai.memory.server.service.memory.OpenMemoryApplicationService; import jakarta.validation.Valid; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; @@ -55,6 +59,53 @@ public Mono> commit(@Valid @RequestBody CommitMemoryRequest requ return Mono.just(ApiResult.ok()); } + @PostMapping("/extract/sync") + public ResponseEntity> extractSync( + @Valid @RequestBody ExtractMemoryRequest request) { + return extractionResponse(service.extract(request)); + } + + @PostMapping("/add-message/sync") + public ResponseEntity> addMessageSync( + @Valid @RequestBody AddMessageRequest request) { + AddMessageResponse response = service.addMessage(request); + if (!response.triggered() || response.result() == null) { + return ResponseEntity.ok(ApiResult.success(response)); + } + if (isFailed(response.result())) { + return failedExtraction(response.result()); + } + return ResponseEntity.ok(ApiResult.success(response)); + } + + @PostMapping("/commit/sync") + public ResponseEntity> commitSync( + @Valid @RequestBody CommitMemoryRequest request) { + return extractionResponse(service.commit(request)); + } + + private static ResponseEntity> extractionResponse( + ExtractMemoryResponse response) { + if (isFailed(response)) { + return failedExtraction(response); + } + return ResponseEntity.ok(ApiResult.success(response)); + } + + private static ResponseEntity> failedExtraction( + ExtractMemoryResponse response) { + String message = + response.errorMessage() == null || response.errorMessage().isBlank() + ? "Memory extraction failed" + : response.errorMessage(); + return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) + .body(ApiResult.failure("extraction_failed", message, null, null)); + } + + private static boolean isFailed(ExtractMemoryResponse response) { + return response != null && "FAILED".equals(response.status()); + } + @PostMapping("/retrieve") public ApiResult retrieve( @Valid @RequestBody RetrieveMemoryRequest request) { diff --git a/memind-server/src/test/java/com/openmemind/ai/memory/server/controller/openapi/OpenMemoryControllerTest.java b/memind-server/src/test/java/com/openmemind/ai/memory/server/controller/openapi/OpenMemoryControllerTest.java index 1b2d7091..e23930e3 100644 --- a/memind-server/src/test/java/com/openmemind/ai/memory/server/controller/openapi/OpenMemoryControllerTest.java +++ b/memind-server/src/test/java/com/openmemind/ai/memory/server/controller/openapi/OpenMemoryControllerTest.java @@ -25,6 +25,8 @@ import com.openmemind.ai.memory.server.domain.memory.request.CommitMemoryRequest; import com.openmemind.ai.memory.server.domain.memory.request.ExtractMemoryRequest; import com.openmemind.ai.memory.server.domain.memory.request.RetrieveMemoryRequest; +import com.openmemind.ai.memory.server.domain.memory.response.AddMessageResponse; +import com.openmemind.ai.memory.server.domain.memory.response.ExtractMemoryResponse; import com.openmemind.ai.memory.server.domain.memory.response.RetrieveMemoryResponse; import com.openmemind.ai.memory.server.handler.ApiExceptionHandler; import com.openmemind.ai.memory.server.service.memory.OpenMemoryApplicationService; @@ -158,6 +160,226 @@ void commitReturnsOkWithoutPayload() throws Exception { org.assertj.core.api.Assertions.assertThat(service.lastCommitRequest).isNotNull(); } + @Test + void extractSyncReturnsResponseOnSuccess() throws Exception { + service.extractResponse = + new ExtractMemoryResponse( + "SUCCESS", + List.of("rd-1"), + List.of(101L), + List.of(201L), + false, + 123L, + null); + + mockMvc.perform( + post("/open/v1/memory/extract/sync") + .contentType(APPLICATION_JSON) + .content( + """ + { + "userId": "u1", + "agentId": "a1", + "rawContent": { + "type": "conversation", + "messages": [ + { + "role": "USER", + "content": [{"type": "text", "text": "hello"}] + } + ] + } + } + """)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.code").value("success")) + .andExpect(jsonPath("$.data.status").value("SUCCESS")) + .andExpect(jsonPath("$.data.rawDataIds[0]").value("rd-1")) + .andExpect(jsonPath("$.data.itemIds[0]").value(101)) + .andExpect(jsonPath("$.data.insightIds[0]").value(201)) + .andExpect(jsonPath("$.data.insightPending").value(false)); + + org.assertj.core.api.Assertions.assertThat(service.lastExtractRequest).isNotNull(); + } + + @Test + void extractSyncPreservesPartialSuccessAndInsightPending() throws Exception { + service.extractResponse = + new ExtractMemoryResponse( + "PARTIAL_SUCCESS", + List.of("rd-1"), + List.of(101L), + List.of(), + true, + 234L, + "insight scheduling deferred"); + + mockMvc.perform( + post("/open/v1/memory/extract/sync") + .contentType(APPLICATION_JSON) + .content( + """ + { + "userId": "u1", + "agentId": "a1", + "rawContent": { + "type": "conversation", + "messages": [ + { + "role": "USER", + "content": [{"type": "text", "text": "hello"}] + } + ] + } + } + """)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.code").value("success")) + .andExpect(jsonPath("$.data.status").value("PARTIAL_SUCCESS")) + .andExpect(jsonPath("$.data.insightPending").value(true)) + .andExpect(jsonPath("$.data.errorMessage").value("insight scheduling deferred")); + } + + @Test + void extractSyncReturnsFailureEnvelopeOnFailedStatus() throws Exception { + service.extractResponse = + new ExtractMemoryResponse( + "FAILED", List.of(), List.of(), List.of(), false, 50L, "extraction failed"); + + mockMvc.perform( + post("/open/v1/memory/extract/sync") + .contentType(APPLICATION_JSON) + .content( + """ + { + "userId": "u1", + "agentId": "a1", + "rawContent": { + "type": "conversation", + "messages": [ + { + "role": "USER", + "content": [{"type": "text", "text": "hello"}] + } + ] + } + } + """)) + .andExpect(status().is5xxServerError()) + .andExpect(jsonPath("$.code").value("extraction_failed")) + .andExpect(jsonPath("$.message").value("extraction failed")); + } + + @Test + void addMessageSyncReturnsSuccessWhenNoExtractionTriggered() throws Exception { + service.addMessageResponse = new AddMessageResponse(false, null); + + mockMvc.perform( + post("/open/v1/memory/add-message/sync") + .contentType(APPLICATION_JSON) + .content( + """ + { + "userId": "u1", + "agentId": "a1", + "message": { + "role": "USER", + "content": [{"type": "text", "text": "hello"}] + } + } + """)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.code").value("success")) + .andExpect(jsonPath("$.data.triggered").value(false)) + .andExpect(jsonPath("$.data.result").doesNotExist()); + + org.assertj.core.api.Assertions.assertThat(service.lastAddMessageRequest).isNotNull(); + } + + @Test + void addMessageSyncReturnsFailureWhenTriggeredExtractionFailed() throws Exception { + service.addMessageResponse = + new AddMessageResponse( + true, + new ExtractMemoryResponse( + "FAILED", + List.of(), + List.of(), + List.of(), + false, + 50L, + "boundary extraction failed")); + + mockMvc.perform( + post("/open/v1/memory/add-message/sync") + .contentType(APPLICATION_JSON) + .content( + """ + { + "userId": "u1", + "agentId": "a1", + "message": { + "role": "USER", + "content": [{"type": "text", "text": "hello"}] + } + } + """)) + .andExpect(status().is5xxServerError()) + .andExpect(jsonPath("$.code").value("extraction_failed")) + .andExpect(jsonPath("$.message").value("boundary extraction failed")); + } + + @Test + void commitSyncReturnsExtractionResponse() throws Exception { + service.commitResponse = + new ExtractMemoryResponse( + "SUCCESS", List.of("rd-2"), List.of(102L), List.of(), false, 77L, null); + + mockMvc.perform( + post("/open/v1/memory/commit/sync") + .contentType(APPLICATION_JSON) + .content( + """ + { + "userId": "u1", + "agentId": "a1" + } + """)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.code").value("success")) + .andExpect(jsonPath("$.data.status").value("SUCCESS")) + .andExpect(jsonPath("$.data.rawDataIds[0]").value("rd-2")); + + org.assertj.core.api.Assertions.assertThat(service.lastCommitRequest).isNotNull(); + } + + @Test + void commitSyncReturnsFailureEnvelopeOnFailedStatus() throws Exception { + service.commitResponse = + new ExtractMemoryResponse( + "FAILED", + List.of(), + List.of(), + List.of(), + false, + 50L, + "commit extraction failed"); + + mockMvc.perform( + post("/open/v1/memory/commit/sync") + .contentType(APPLICATION_JSON) + .content( + """ + { + "userId": "u1", + "agentId": "a1" + } + """)) + .andExpect(status().is5xxServerError()) + .andExpect(jsonPath("$.code").value("extraction_failed")) + .andExpect(jsonPath("$.message").value("commit extraction failed")); + } + @Test void retrieveReturnsRankedMemoryPayload() throws Exception { mockMvc.perform( @@ -186,6 +408,13 @@ private static final class StubOpenMemoryApplicationService private ExtractMemoryRequest lastExtractRequest; private AddMessageRequest lastAddMessageRequest; private CommitMemoryRequest lastCommitRequest; + private ExtractMemoryResponse extractResponse = + new ExtractMemoryResponse( + "SUCCESS", List.of(), List.of(), List.of(), false, 1L, null); + private AddMessageResponse addMessageResponse = new AddMessageResponse(false, null); + private ExtractMemoryResponse commitResponse = + new ExtractMemoryResponse( + "SUCCESS", List.of(), List.of(), List.of(), false, 1L, null); private StubOpenMemoryApplicationService() { super(null); @@ -206,6 +435,24 @@ public void commitAsync(CommitMemoryRequest request) { this.lastCommitRequest = request; } + @Override + public ExtractMemoryResponse extract(ExtractMemoryRequest request) { + this.lastExtractRequest = request; + return extractResponse; + } + + @Override + public AddMessageResponse addMessage(AddMessageRequest request) { + this.lastAddMessageRequest = request; + return addMessageResponse; + } + + @Override + public ExtractMemoryResponse commit(CommitMemoryRequest request) { + this.lastCommitRequest = request; + return commitResponse; + } + @Override public RetrieveMemoryResponse retrieve(RetrieveMemoryRequest request) { return new RetrieveMemoryResponse(