From 8015c592e7b6387c6405c4c2dbf05e3d1d2f6334 Mon Sep 17 00:00:00 2001 From: zhishu Date: Fri, 21 Aug 2026 09:13:28 +0800 Subject: [PATCH 1/2] fix(memorylake): extend the conversation head so appends past the first succeed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Appending a message to a MemoryLake conversation has to name the conversation's current head as parent_message_id. engram never sent that field, so only the very first message of a session's conversation was ever accepted — every append after it came back 409: append turn: memorylake CUSTOM_ID_CONFLICT (409): Conversation 'conv-403f4cc2…' already has a head entry; use the current head as `parent_entry_id` (or omit it only for the very first entry). This was not confined to per-turn conversation sync. AppendObservation has two callers that share one conversation per session (keyed by session id), and routing.go's SetSkipPromptAppend picks which one uses it: with conversations=on it is turn sync, breaking from turn 2; with conversations=off it is prompt capture, breaking from prompt 2. Every MemoryLake-enabled project was affected. mem_save is not — it writes facts directly and never touches a conversation. ensureConversation now also reports the head (current_message_id, documented as "use as parent_message_id when appending the next message"), and appends send it. It is omitted for a conversation that has no messages yet, which is the one case the API wants it absent. Because engram writes one conversation from up to three processes (serve for prompts, mcp for observations, turn for turns), the head can move between the read and the append. A 409 therefore re-reads the head and retries once — bounded, so a caller that keeps losing the race gets the error rather than looping. That retry doubles as the recovery path when the head was unknown (empty) but the conversation had one, which keeps this a no-op rather than a regression against a deployment that does not populate current_message_id. Field naming is per-layer and worth stating: the gateway engram talks to uses parent_message_id / current_message_id, while parent_entry_id / current_entry_id are the knowledge service's internal names and only surface in the error text that bubbles up. Both gateway fields come from the vendored v3 OpenAPI spec. Also corrects ensureConversation's doc comment, which claimed the create endpoint is idempotent on custom_id and needs no existence lookup — the opposite of what the code right below it does, and of why the head is available at all. Closes #20 --- DOCS.md | 5 + internal/memorylake/writequeue.go | 134 +++++++++++++++------ internal/memorylake/writequeue_test.go | 154 +++++++++++++++++++++++++ 3 files changed, 260 insertions(+), 33 deletions(-) diff --git a/DOCS.md b/DOCS.md index c91b9c57c..a4912e0ac 100644 --- a/DOCS.md +++ b/DOCS.md @@ -1731,6 +1731,11 @@ timeout killing it from outside. the user presses ESC. - **Failures are dropped, not retried.** A turn lost to a network outage is gone; the failure is recorded in `~/.engram/logs/turn.log` and nowhere else. + The one exception is a rejected append: MemoryLake requires a message to + extend its conversation's current head, so if a concurrent writer moves that + head first, engram re-reads it and retries the append once before giving up. + Note that because only failures are logged, an absent `turn.log` cannot tell + you whether every turn succeeded or whether the hook never ran at all. - **Every message is stored with role `USER`.** MemoryLake derives a message's role from its actor's type and only HUMAN actors can be created through the API, so the speaker is carried in the message text instead. diff --git a/internal/memorylake/writequeue.go b/internal/memorylake/writequeue.go index 706489a5d..f98b1a859 100644 --- a/internal/memorylake/writequeue.go +++ b/internal/memorylake/writequeue.go @@ -5,6 +5,7 @@ import ( "encoding/hex" "fmt" "log" + "net/http" "net/url" "github.com/Gentleman-Programming/engram/internal/store" @@ -30,29 +31,74 @@ import ( // MemoryLake resolves the duplicate custom_id to the message it already has // instead of creating a second one. // +// Appending is not just "POST to the conversation": a message has to extend +// the conversation's current head by naming it as parent_message_id, and +// MemoryLake rejects an append that names the wrong parent (or names none once +// a head exists) with a 409. Ensuring the conversation therefore also reads its +// head, and a 409 is recovered from by re-reading the head and retrying — see +// maxAppendHeadRetries. +// // AppendObservation returns the MemoryLake message id. func (c *Client) AppendObservation(ws, projID, convCustomID, actorID string, p store.AddObservationParams) (string, error) { - convID, err := c.ensureConversation(ws, projID, convCustomID, actorID) + conv, err := c.ensureConversation(ws, projID, convCustomID, actorID) if err != nil { return "", err } - var msg struct { - ID string `json:"id"` - } - body := map[string]any{ - "custom_id": contentHash(p.Content), - "actor_id": actorID, - "content": []map[string]any{ - {"block_type": "TEXT", "text": p.Content}, - }, - } - if err := c.doJSON("POST", "/api/v3/conversations/"+convID+"/messages", body, &msg); err != nil { - return "", err + head := conv.CurrentMessageID + for attempt := 0; ; attempt++ { + var msg struct { + ID string `json:"id"` + } + body := map[string]any{ + "custom_id": contentHash(p.Content), + "actor_id": actorID, + "content": []map[string]any{ + {"block_type": "TEXT", "text": p.Content}, + }, + } + // Appends must extend the conversation's current head, so the head we + // just read goes back as parent_message_id. It is omitted only for the + // very first entry of a conversation, which has no head to extend — + // sending a parent there is itself rejected. + if head != "" { + body["parent_message_id"] = head + } + + err := c.doJSON("POST", "/api/v3/conversations/"+conv.ID+"/messages", body, &msg) + if err == nil { + return msg.ID, nil + } + + apiErr, ok := err.(*APIError) + if !ok || apiErr.HTTP != http.StatusConflict || attempt >= maxAppendHeadRetries { + return "", err + } + + // Head drift: engram writes one session's conversation from up to three + // separate processes (`engram serve` for prompts, `engram mcp` for + // observations, `engram turn` for turns), so another writer can extend + // the conversation between our read and this append. MemoryLake answers + // that with a 409 and the documented recovery is to re-read the head and + // extend that instead. Re-reading also covers the case where the head we + // started from was unknown (empty) but the conversation did have one. + refreshed, gErr := c.getConversationByCustomID(ws, convCustomID) + if gErr != nil || refreshed.CurrentMessageID == "" || refreshed.CurrentMessageID == head { + // Nothing new to extend — report the original append failure rather + // than masking it with a lookup error. + return "", err + } + head = refreshed.CurrentMessageID } - return msg.ID, nil } +// maxAppendHeadRetries bounds how many times AppendObservation re-reads the +// conversation head and retries after a 409. One retry absorbs a single +// concurrent writer, which is the realistic case for a per-session +// conversation; a caller losing the race repeatedly gets the error instead of +// looping against a server that keeps moving the head. +const maxAppendHeadRetries = 1 + // AddFacts writes each string in facts verbatim into project projID via // MemoryLake's direct fact-add endpoint (POST .../memories/facts, shipped // 2026-07-23), bypassing the conversation → mem0-extraction pipeline entirely: @@ -78,14 +124,16 @@ func (c *Client) AddFacts(ws, projID string, facts []string) ([]Fact, error) { // ensureConversation creates the MemoryLake conversation identified by // custom_id within workspace ws, scoped to actorID and granted read-write -// access to project projID. MemoryLake treats POST .../conversations as -// idempotent on custom_id — calling this repeatedly with the same customID -// returns the same conversation id rather than creating duplicates, so no -// separate "does it already exist" lookup is needed here. -func (c *Client) ensureConversation(ws, projID, customID, actorID string) (string, error) { - var conv struct { - ID string `json:"id"` - } +// access to project projID, and returns it along with its current head (see +// conversationRef). +// +// POST .../conversations is NOT idempotent on custom_id — it rejects a +// duplicate rather than returning the existing row — so the conflict is +// recovered from explicitly below. That recovery is also what supplies the +// head: a freshly created conversation has none, while an existing one +// generally does, and an append has to name it. +func (c *Client) ensureConversation(ws, projID, customID, actorID string) (conversationRef, error) { + var conv conversationRef body := map[string]any{ "custom_id": customID, "kind": "DIRECT", @@ -99,22 +147,42 @@ func (c *Client) ensureConversation(ws, projID, customID, actorID string) (strin // mem_save after the first within a session. Recover by fetching the // existing conversation by custom_id. if apiErr, ok := err.(*APIError); !ok || apiErr.Code != "CUSTOM_ID_CONFLICT" { - return "", err - } - var existing struct { - ID string `json:"id"` + return conversationRef{}, err } - getPath := "/api/v3/workspaces/" + ws + "/memories/conversations/" + - url.PathEscape(customID) + "?by_custom_id=true" - if gErr := c.doJSON("GET", getPath, nil, &existing); gErr != nil { - return "", gErr + existing, gErr := c.getConversationByCustomID(ws, customID) + if gErr != nil { + return conversationRef{}, gErr } if existing.ID == "" { - return "", err + return conversationRef{}, err } - return existing.ID, nil + return existing, nil + } + return conv, nil +} + +// conversationRef is the subset of MemoryLake's conversation object this +// package reads. CurrentMessageID is the conversation's head — the id of its +// latest message, which an append has to name as its parent_message_id. It is +// empty for a conversation that has no messages yet, and it is the reason a +// plain "ensure the conversation exists" is not enough to append to one. +type conversationRef struct { + ID string `json:"id"` + CurrentMessageID string `json:"current_message_id"` +} + +// getConversationByCustomID resolves a conversation by its caller-defined +// custom_id (the engram session id), returning its id and current head. +// Used both to recover from ensureConversation's create conflict and to +// re-read a head that moved under a concurrent writer. +func (c *Client) getConversationByCustomID(ws, customID string) (conversationRef, error) { + var conv conversationRef + path := "/api/v3/workspaces/" + ws + "/memories/conversations/" + + url.PathEscape(customID) + "?by_custom_id=true" + if err := c.doJSON("GET", path, nil, &conv); err != nil { + return conversationRef{}, err } - return conv.ID, nil + return conv, nil } // contentHash derives a stable idempotency key from s: the first 16 hex diff --git a/internal/memorylake/writequeue_test.go b/internal/memorylake/writequeue_test.go index b0315e819..461835024 100644 --- a/internal/memorylake/writequeue_test.go +++ b/internal/memorylake/writequeue_test.go @@ -173,3 +173,157 @@ func TestPatchFactMetadata_OverwritesMetadata(t *testing.T) { t.Fatalf("patchFactMetadata must only send metadata, not fact: %v", gotBody) } } + +// TestAppendObservation_SendsHeadAsParentMessageID covers the normal case for +// every append after a session's first: the conversation already exists, so +// ensureConversation recovers via GET, and the head it reports must be sent +// back as parent_message_id. Without it MemoryLake rejects the append with a +// 409 ("already has a head entry"), which is what silently capped every +// conversation at one message. +func TestAppendObservation_SendsHeadAsParentMessageID(t *testing.T) { + var lastMsgBody map[string]any + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == "POST" && r.URL.Path == "/api/v3/workspaces/ws-1/memories/conversations": + w.WriteHeader(http.StatusConflict) + json.NewEncoder(w).Encode(map[string]any{ + "success": false, + "error_code": "CUSTOM_ID_CONFLICT", + "message": "conversation with this custom_id already exists", + }) + case r.Method == "GET" && r.URL.Path == "/api/v3/workspaces/ws-1/memories/conversations/session-abc": + json.NewEncoder(w).Encode(map[string]any{ + "success": true, + "data": map[string]any{"id": "conv-1", "current_message_id": "msg-7"}, + }) + case r.Method == "POST" && r.URL.Path == "/api/v3/conversations/conv-1/messages": + json.NewDecoder(r.Body).Decode(&lastMsgBody) + json.NewEncoder(w).Encode(map[string]any{ + "success": true, + "data": map[string]any{"id": "msg-8"}, + }) + default: + t.Fatalf("unexpected request %s %s", r.Method, r.URL.Path) + } + })) + defer srv.Close() + + c := NewClient(Config{BaseURL: srv.URL, APIKey: "sk-test", TimeoutMS: 5000}) + msgID, err := c.AppendObservation("ws-1", "proj-1", "session-abc", "actor-1", + store.AddObservationParams{Content: "second turn"}) + if err != nil { + t.Fatalf("AppendObservation: %v", err) + } + if msgID != "msg-8" { + t.Fatalf("msgID=%q, want msg-8", msgID) + } + if lastMsgBody["parent_message_id"] != "msg-7" { + t.Fatalf("parent_message_id=%v, want msg-7", lastMsgBody["parent_message_id"]) + } +} + +// TestAppendObservation_OmitsParentMessageIDForFirstMessage pins the other half +// of the contract: parent_message_id must be omitted for the very first entry +// of a conversation. Sending an empty or bogus parent there is itself a 409 +// ("not the current head"), so a fresh conversation must not carry the field +// at all. +func TestAppendObservation_OmitsParentMessageIDForFirstMessage(t *testing.T) { + var lastMsgBody map[string]any + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == "POST" && r.URL.Path == "/api/v3/workspaces/ws-1/memories/conversations": + // Fresh conversation: created here, so it has no head yet. + json.NewEncoder(w).Encode(map[string]any{ + "success": true, + "data": map[string]any{"id": "conv-1"}, + }) + case r.Method == "POST" && r.URL.Path == "/api/v3/conversations/conv-1/messages": + json.NewDecoder(r.Body).Decode(&lastMsgBody) + json.NewEncoder(w).Encode(map[string]any{ + "success": true, + "data": map[string]any{"id": "msg-1"}, + }) + default: + t.Fatalf("unexpected request %s %s", r.Method, r.URL.Path) + } + })) + defer srv.Close() + + c := NewClient(Config{BaseURL: srv.URL, APIKey: "sk-test", TimeoutMS: 5000}) + if _, err := c.AppendObservation("ws-1", "proj-1", "session-abc", "actor-1", + store.AddObservationParams{Content: "first turn"}); err != nil { + t.Fatalf("AppendObservation: %v", err) + } + if _, ok := lastMsgBody["parent_message_id"]; ok { + t.Fatalf("parent_message_id must be absent for the first message, got %v", + lastMsgBody["parent_message_id"]) + } +} + +// TestAppendObservation_RetriesWithRefreshedHeadOn409 covers head drift: a +// concurrent writer (engram serve, engram mcp and engram turn are separate +// processes writing the same session's conversation) can advance the head +// between our read and our append. MemoryLake answers that with a 409, and the +// documented recovery is to re-read the head and extend it instead. +func TestAppendObservation_RetriesWithRefreshedHeadOn409(t *testing.T) { + var gets, msgPosts int32 + var parents []any + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == "POST" && r.URL.Path == "/api/v3/workspaces/ws-1/memories/conversations": + w.WriteHeader(http.StatusConflict) + json.NewEncoder(w).Encode(map[string]any{ + "success": false, "error_code": "CUSTOM_ID_CONFLICT", "message": "exists", + }) + case r.Method == "GET" && r.URL.Path == "/api/v3/workspaces/ws-1/memories/conversations/session-abc": + // First read reports msg-7; by the time we append, another writer + // has moved the head to msg-9. + head := "msg-7" + if atomic.AddInt32(&gets, 1) > 1 { + head = "msg-9" + } + json.NewEncoder(w).Encode(map[string]any{ + "success": true, + "data": map[string]any{"id": "conv-1", "current_message_id": head}, + }) + case r.Method == "POST" && r.URL.Path == "/api/v3/conversations/conv-1/messages": + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + parents = append(parents, body["parent_message_id"]) + if atomic.AddInt32(&msgPosts, 1) == 1 { + w.WriteHeader(http.StatusConflict) + json.NewEncoder(w).Encode(map[string]any{ + "success": false, "error_code": "CUSTOM_ID_CONFLICT", + "message": "`parent_message_id` 'msg-7' is not the current head", + }) + return + } + json.NewEncoder(w).Encode(map[string]any{ + "success": true, + "data": map[string]any{"id": "msg-10"}, + }) + default: + t.Fatalf("unexpected request %s %s", r.Method, r.URL.Path) + } + })) + defer srv.Close() + + c := NewClient(Config{BaseURL: srv.URL, APIKey: "sk-test", TimeoutMS: 5000}) + msgID, err := c.AppendObservation("ws-1", "proj-1", "session-abc", "actor-1", + store.AddObservationParams{Content: "concurrent turn"}) + if err != nil { + t.Fatalf("AppendObservation: %v", err) + } + if msgID != "msg-10" { + t.Fatalf("msgID=%q, want msg-10", msgID) + } + if got := atomic.LoadInt32(&msgPosts); got != 2 { + t.Fatalf("msgPosts=%d, want 2 (one 409, one retry)", got) + } + if len(parents) != 2 || parents[0] != "msg-7" || parents[1] != "msg-9" { + t.Fatalf("parent_message_id sequence=%v, want [msg-7 msg-9]", parents) + } +} From 67d775cf1ccb7bf790ed6dea3d6a8ad25705896e Mon Sep 17 00:00:00 2001 From: zhishu Date: Fri, 21 Aug 2026 09:55:05 +0800 Subject: [PATCH 2/2] chore(release): bump plugin manifests to 0.6.3 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RELEASING.md step 3: the manifests have to carry the version being tagged before the tag is pushed, or release.yml's check fails ahead of GoReleaser. Folded into this PR because the bump is part of shipping this fix — the conversation-append fix lives in the Go binary, so it only reaches anyone through a release. --- .claude-plugin/marketplace.json | 2 +- plugin/claude-code/.claude-plugin/plugin.json | 2 +- plugin/codex/.codex-plugin/plugin.json | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 797ccb70d..3c25480b6 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -9,7 +9,7 @@ { "name": "engram", "description": "Persistent memory with session tracking, compaction recovery, and Memory Protocol. Gives Claude a brain that survives across sessions.", - "version": "0.6.2", + "version": "0.6.3", "author": { "name": "relytcloud" }, diff --git a/plugin/claude-code/.claude-plugin/plugin.json b/plugin/claude-code/.claude-plugin/plugin.json index 52133c064..b7a459b46 100644 --- a/plugin/claude-code/.claude-plugin/plugin.json +++ b/plugin/claude-code/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "engram", "description": "Persistent memory for AI coding agents. Survives across sessions and compactions.", - "version": "0.6.2", + "version": "0.6.3", "author": { "name": "Gentleman Programming" }, diff --git a/plugin/codex/.codex-plugin/plugin.json b/plugin/codex/.codex-plugin/plugin.json index 95ac14690..96bebfbd7 100644 --- a/plugin/codex/.codex-plugin/plugin.json +++ b/plugin/codex/.codex-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "engram", "displayName": "Engram Persistent Memory", - "version": "0.6.2", + "version": "0.6.3", "description": "Persistent memory for AI coding agents. Survives across sessions and compactions.", "shortDescription": "Persistent memory with session tracking, compaction recovery, and Memory Protocol.", "author": {