Recover background memory updates after transient timeouts

docs/changes/unreleased/1901-memory-timeout-recovery.md

@@ -0,0 +1,20 @@
+---
+kind: fixed
+title: recover transient background memory failures before reporting a failed save
+pr: 1901
+surface: [chat, engine]
+invalidates:
+  - >-
+    A stalled memory request could exhaust its shared deadline without leaving
+    time for recovery. The primary now reserves half the remaining patience for
+    one small-work attempt, and a transiently failed pass retries the original
+    exchange once after two seconds in the background.
+  - >-
+    The chat prompt told models that /remember was required without a remember
+    tool call. It now describes automatic background capture and waits for the
+    save notice before claiming success.
+---
+
+Recovery respects chat cancellation and does not persist pending exchanges across
+restarts. Permanent failures are not retried. Exhausted recovery retains the
+technical journal entry and names /remember as a direct save path.

internal/e2e/memory_recovery_e2e_test.go

@@ -0,0 +1,220 @@
+//go:build e2e
+
+package e2e
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/http/httputil"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Agent-Field/codeaf/internal/config"
+	"github.com/Agent-Field/codeaf/internal/gitidentity"
+	"github.com/Agent-Field/codeaf/internal/store"
+)
+
+// The terminal answers normally while two injected memory stalls exhaust one
+// noticing pass. Its later recovery must save the ORIGINAL exchange, and a new
+// chat must recall it through the real model rather than a synthetic answer.
+func TestMemoryRecoversFromTimeoutOnTheRealTerminal(t *testing.T) {
+	requireTmuxAndKey(t)
+	const model = "deepseek/deepseek-v4.1-flash"
+	if e2eModel != model {
+		t.Fatalf("requires CODEAF_E2E_MODEL=%s", model)
+	}
+	target, _ := url.Parse(config.DefaultBaseURL)
+	proxy := httputil.NewSingleHostReverseProxy(target)
+	director := proxy.Director
+	proxy.Director = func(r *http.Request) { director(r); r.Host = target.Host }
+	proxy.FlushInterval = -1
+	var mu sync.Mutex
+	stalled, forwarded := 0, 0
+	var observed []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/chat/completions") {
+			body, err := io.ReadAll(r.Body)
+			if err != nil {
+				http.Error(w, "read", 400)
+				return
+			}
+			r.Body = io.NopCloser(strings.NewReader(string(body)))
+			var request struct {
+				Model    string `json:"model"`
+				Messages []struct {
+					Content json.RawMessage `json:"content"`
+				} `json:"messages"`
+			}
+			if json.Unmarshal(body, &request) != nil {
+				http.Error(w, "json", 400)
+				return
+			}
+			isMemory := len(request.Messages) > 0 && strings.Contains(string(request.Messages[0].Content), "You read one exchange between a person and an assistant")
+			mu.Lock()
+			observed = append(observed, request.Model)
+			if isMemory && stalled < 2 {
+				stalled++
+				mu.Unlock()
+				// Deliberately withhold all response bytes until the app's own attempt
+				// deadline cancels the request. No fabricated model response is supplied.
+				<-r.Context().Done()
+				return
+			}
+			if isMemory {
+				forwarded++
+			}
+			mu.Unlock()
+		}
+		// Strip the fixture prefix; everything after the injected outage is an
+		// authenticated request to the actual OpenRouter endpoint.
+		r.URL.Path = strings.TrimPrefix(r.URL.Path, "/fixture")
+		proxy.ServeHTTP(w, r)
+	}))
+	defer server.Close()
+	home := newHome(t, map[string]any{"memory.enabled": true})
+	ws := newWorkspace(t, "juniper-recovery", false)
+	project, err := gitidentity.ProjectKey(ws)
+	if err != nil {
+		t.Fatal(err)
+	}
+	st, err := store.Open(filepath.Join(home, "graph.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	r := startWithEnv(t, []string{config.APIKeyEnv + "=" + liveKey(t), "CODEAF_BASE_URL=" + server.URL + "/fixture"}, "memory_recovery", home, ws, 150, 45, "chat", "--one-model", "--debug")
+	r.skipSetup(t)
+	prompt := "For the fictional Juniper demo, our standing packaging convention is violet envelopes for weekend parcels. Please acknowledge in one sentence without calling any tools."
+	r.lit(prompt)
+	r.keys("Enter")
+	answer, tools := privateMemoryAwaitAnswer(home, prompt, 2*time.Minute)
+	if answer == "" {
+		t.Fatal("chat did not finish during background memory outage")
+	}
+	for _, tool := range tools {
+		if strings.Contains(tool, "remember") {
+			t.Fatal("main turn bypassed background recovery")
+		}
+	}
+	deadline := time.Now().Add(2 * time.Minute)
+	for {
+		rows, err := st.MemoriesFor([]string{store.OwnerProject(project), store.OwnerUser})
+		if err != nil {
+			t.Fatal(err)
+		}
+		n := 0
+		for _, m := range rows {
+			if m.Status == store.MemoryActive && strings.Contains(strings.ToLower(m.Text), "violet") && strings.Contains(strings.ToLower(m.Text), "weekend") {
+				n++
+			}
+		}
+		if n > 1 {
+			t.Fatal("recovery duplicated the memory")
+		}
+		if n == 1 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("background recovery never saved original packaging convention")
+		}
+		time.Sleep(pollEvery)
+	}
+	r.lit("/new")
+	r.keys("Enter")
+	time.Sleep(time.Second)
+	recall := "For the fictional Juniper demo, which envelopes do we use for weekend parcels? Answer from saved memory, without tools."
+	r.lit(recall)
+	r.keys("Enter")
+	recalled, _ := privateMemoryAwaitAnswer(home, recall, 2*time.Minute)
+	if !strings.Contains(strings.ToLower(recalled), "violet") {
+		t.Fatalf("fresh chat failed recall: %s", recalled)
+	}
+	if strings.Contains(r.capture(), "couldn't save this exchange") {
+		t.Fatal("recovered memory displayed terminal failure")
+	}
+	correction := "For the fictional Juniper demo, change our standing weekend-parcel packaging convention from violet envelopes to copper envelopes. This replaces the earlier convention. Acknowledge briefly without tools."
+	r.lit(correction)
+	r.keys("Enter")
+	if completed, _ := privateMemoryAwaitAnswer(home, correction, 2*time.Minute); completed == "" {
+		t.Fatal("correction did not finish")
+	}
+	deadline = time.Now().Add(90 * time.Second)
+	for {
+		rows, err := st.MemoriesFor([]string{store.OwnerProject(project), store.OwnerUser})
+		if err != nil {
+			t.Fatal(err)
+		}
+		copper, violet := 0, 0
+		for _, m := range rows {
+			if m.Status != store.MemoryActive {
+				continue
+			}
+			text := strings.ToLower(m.Text)
+			if strings.Contains(text, "weekend") {
+				if strings.Contains(text, "copper") {
+					copper++
+				}
+				if strings.Contains(text, "violet") && !strings.Contains(text, "copper") {
+					violet++
+				}
+			}
+		}
+		if copper == 1 && violet == 0 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("correction not applied: copper=%d old=%d", copper, violet)
+		}
+		time.Sleep(pollEvery)
+	}
+	r.lit("/new")
+	r.keys("Enter")
+	time.Sleep(time.Second)
+	apply := "For the fictional Juniper demo, create packaging.json with a weekend_envelope_color field set from our current convention. Use saved memory and do not ask me to repeat it."
+	r.lit(apply)
+	r.keys("Enter")
+	if completed, _ := privateMemoryAwaitAnswer(home, apply, 2*time.Minute); completed == "" {
+		t.Fatal("application did not finish")
+	}
+	body, err := os.ReadFile(filepath.Join(ws, "packaging.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var packaging struct {
+		Color string `json:"weekend_envelope_color"`
+	}
+	if err := json.Unmarshal(body, &packaging); err != nil || !strings.EqualFold(packaging.Color, "copper") {
+		t.Fatalf("saved correction not applied: %s (%v)", body, err)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if stalled != 2 || forwarded == 0 {
+		t.Fatalf("stalled=%d live memory requests=%d", stalled, forwarded)
+	}
+	if len(observed) == 0 {
+		t.Fatal("no requests observed")
+	}
+	for _, m := range observed {
+		if m != model {
+			t.Fatalf("unexpected actual model %q", m)
+		}
+	}
+	models := callModels(t, home)
+	if len(models) == 0 {
+		t.Fatal("no model receipts")
+	}
+	for _, m := range models {
+		if m != model {
+			t.Fatalf("unexpected receipt model %q", m)
+		}
+	}
+	t.Logf("PASS: two stalled memory requests recovered in background; one saved convention recalled, corrected, and applied to an actual file across fresh chats; %d requests and %d receipts used %s", len(observed), len(models), model)
+	t.Logf("Live terminal after fresh-chat recall:\n%s", r.capture())
+}

internal/manual/chat/what-i-remember.md

@@ -469,7 +469,10 @@ once, keeps the conversation's own running summary of where the work stands, and
 conversation's own model), sends no token ceiling or reasoning switch, and is
 charged to the session total, so `/cost` includes it. An answer it cannot read
 costs one retry; a model that answers nothing is replaced for the session by the
-**small work** model, with one line saying so. Turning noticing off does not
+**small work** model, with one line saying so. A stalled or temporarily failed
+request reserves time for one small-work attempt within the same 45-second
+limit. If that also fails temporarily, the original exchange is tried once more
+after two seconds in the background. Turning noticing off does not
 remove the call, because the running summary still needs it; it only stops lines
 being kept.
 
@@ -487,10 +490,16 @@ and the import tries again on the next turn.
 
 A save, a forget or the noticing pass that cannot reach the store is written to
 the store's journal with its reason, and the conversation says so once every
-five minutes at most: `couldn't update memory just now · <reason>`. When the
+five minutes at most. A failed background reading says
+`couldn't save this exchange to memory · use /remember to save an important detail`;
+a failed direct update says `couldn't update memory just now · <reason>`. When the
 journal itself cannot be written, the reason goes to
 `~/.codeaf/v3/memory-failures.log` instead. The turn you were having is never
-affected.
+affected. Background recovery stays owned by the open chat: closing it cancels
+pending attempts, and failed exchanges are not queued across restarts. A recovered
+reading saves normally without showing a failure notice. If all bounded attempts
+fail, `/remember <detail>` saves the important statement directly without a model
+request; it does not depend on the failed background service.
 
 
 ## What did we decide last week — can you look up an earlier conversation or search old chats

internal/reflex/recovery_test.go

@@ -0,0 +1,125 @@
+package reflex
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/Agent-Field/agentfield/sdk/go/ai"
+	"github.com/Agent-Field/codeaf/internal/roles"
+)
+
+type recoveryCompleter struct {
+	run    func(context.Context, string) (*ai.Response, error)
+	models []string
+}
+
+func (c *recoveryCompleter) CompleteWithMessages(ctx context.Context, messages []ai.Message, options ...ai.Option) (*ai.Response, error) {
+	var request ai.Request
+	for _, option := range options {
+		if err := option(&request); err != nil {
+			return nil, err
+		}
+	}
+	c.models = append(c.models, request.Model)
+	return c.run(ctx, request.Model)
+}
+
+func TestStalledMemoryRequestLeavesTimeForFallback(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		start := time.Now()
+		c := &recoveryCompleter{}
+		c.run = func(ctx context.Context, model string) (*ai.Response, error) {
+			deadline, ok := ctx.Deadline()
+			want := start.Add(roles.PatienceFor(roles.RoleReflex))
+			if model == "primary" {
+				want = start.Add(roles.PatienceFor(roles.RoleReflex) / 2)
+			}
+			if !ok || !deadline.Equal(want) {
+				t.Fatalf("%s deadline=%s want=%s", model, deadline, want)
+			}
+			if model == "primary" {
+				<-ctx.Done()
+				return nil, ctx.Err()
+			}
+			if ctx.Err() != nil {
+				t.Fatal("fallback inherited expired primary context")
+			}
+			time.Sleep(time.Second)
+			return reply(`{"keep":"Use tabs","scope":"user"}`), nil
+		}
+		client := (&Session{}).Bind(c, "primary", "fallback", func(string) { t.Error("transient recovery announced permanent failure") })
+		found, err := Notice(context.Background(), client, "I use tabs", "okay", nil, true)
+		if err != nil || found.Keep != "Use tabs" {
+			t.Fatalf("found=%+v error=%v", found, err)
+		}
+		if !reflect.DeepEqual(c.models, []string{"primary", "fallback"}) {
+			t.Fatal(c.models)
+		}
+		if elapsed := time.Since(start); elapsed != roles.PatienceFor(roles.RoleReflex)/2+time.Second {
+			t.Fatalf("recovery exhausted patience: %s", elapsed)
+		}
+		// A slow request does not declare the primary permanently unusable.
+		c.run = func(context.Context, string) (*ai.Response, error) { return reply(`{"keep":""}`), nil }
+		if _, err := Notice(context.Background(), client, "hello", "hi", nil, true); err != nil {
+			t.Fatal(err)
+		}
+		if c.models[2] != "primary" {
+			t.Fatal(c.models)
+		}
+	})
+}
+
+func TestMemoryFallbackCannotOutliveOwnerOrOverallPatience(t *testing.T) {
+	for _, cancelOwner := range []bool{false, true} {
+		t.Run(map[bool]string{false: "both stalled", true: "owner cancelled"}[cancelOwner], func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				ctx, cancel := context.WithCancel(context.Background())
+				defer cancel()
+				if cancelOwner {
+					go func() { time.Sleep(time.Second); cancel() }()
+				}
+				start := time.Now()
+				c := &recoveryCompleter{run: func(ctx context.Context, _ string) (*ai.Response, error) { <-ctx.Done(); return nil, ctx.Err() }}
+				_, err := Notice(ctx, (&Session{}).Bind(c, "primary", "fallback", nil), "tabs", "okay", nil, true)
+				wantCalls := 2
+				wantErr := context.DeadlineExceeded
+				if cancelOwner {
+					wantCalls = 1
+					wantErr = context.Canceled
+				}
+				if !errors.Is(err, wantErr) || len(c.models) != wantCalls {
+					t.Fatalf("calls=%v err=%v", c.models, err)
+				}
+				if time.Since(start) > roles.PatienceFor(roles.RoleReflex) {
+					t.Fatal("exceeded total bound")
+				}
+			})
+		})
+	}
+}
+
+func TestTransientMemoryFailureCanRetrySameConfiguredModel(t *testing.T) {
+	c := &recoveryCompleter{}
+	c.run = func(context.Context, string) (*ai.Response, error) {
+		if len(c.models) == 1 {
+			return nil, context.DeadlineExceeded
+		}
+		return reply(`{"keep":"Use tabs"}`), nil
+	}
+	found, err := Notice(context.Background(), (&Session{}).Bind(c, "same", "same", nil), "tabs", "okay", nil, true)
+	if err != nil || found.Keep != "Use tabs" || len(c.models) != 2 {
+		t.Fatalf("found=%+v calls=%v err=%v", found, c.models, err)
+	}
+}
+
+func TestPermanentMemoryFailureDoesNotStartFallback(t *testing.T) {
+	c := &recoveryCompleter{run: func(context.Context, string) (*ai.Response, error) { return nil, errors.New("invalid credentials") }}
+	_, err := Notice(context.Background(), (&Session{}).Bind(c, "primary", "fallback", nil), "tabs", "okay", nil, true)
+	if err == nil || len(c.models) != 1 {
+		t.Fatalf("calls=%v err=%v", c.models, err)
+	}
+}