From ed2587cfaa73ae732f76afee42e8905c42988870 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 18:14:34 -0400 Subject: [PATCH 1/6] Recover transient background memory failures with reserved fallback time --- internal/e2e/memory_recovery_e2e_test.go | 220 +++++++++++++++++++++++ internal/manual/chat/what-i-remember.md | 15 +- internal/reflex/recovery_test.go | 117 ++++++++++++ internal/reflex/reflex.go | 51 ++++-- internal/session/memory.go | 41 ++++- internal/session/memory_recovery_test.go | 185 +++++++++++++++++++ internal/session/prompts/system.md | 2 +- 7 files changed, 612 insertions(+), 19 deletions(-) create mode 100644 internal/e2e/memory_recovery_e2e_test.go create mode 100644 internal/reflex/recovery_test.go create mode 100644 internal/session/memory_recovery_test.go diff --git a/internal/e2e/memory_recovery_e2e_test.go b/internal/e2e/memory_recovery_e2e_test.go new file mode 100644 index 0000000000..c80f15410d --- /dev/null +++ b/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()) +} diff --git a/internal/manual/chat/what-i-remember.md b/internal/manual/chat/what-i-remember.md index bfee66bf23..5de978b16a 100644 --- a/internal/manual/chat/what-i-remember.md +++ b/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 · `. 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 · `. 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 ` 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 diff --git a/internal/reflex/recovery_test.go b/internal/reflex/recovery_test.go new file mode 100644 index 0000000000..c5d8da87ce --- /dev/null +++ b/internal/reflex/recovery_test.go @@ -0,0 +1,117 @@ +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) { + 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) { + 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) + } +} diff --git a/internal/reflex/reflex.go b/internal/reflex/reflex.go index 24cc1a2423..f92e4dfd6d 100644 --- a/internal/reflex/reflex.go +++ b/internal/reflex/reflex.go @@ -25,8 +25,9 @@ // - REFLEX NEVER BREAKS A TURN. A model that answers with prose, with a code // fence, or with a word this package has never heard of costs ONE repair // retry. A model that answers nothing gets the session's low-tier fallback. -// After that every failure is a zero result and [ErrReflexFailed], and every -// caller treats that as "nothing noticed". +// A transient request failure also gets one bounded fallback attempt. +// Remaining failures return [ErrReflexFailed]; the owner may retry the +// original exchange in the background without interrupting its turn. // // The package is the client only. It makes no decision about WHEN a call // happens and holds no store. @@ -38,6 +39,7 @@ import ( "fmt" "strings" "sync" + "time" "github.com/Agent-Field/agentfield/sdk/go/ai" "github.com/Agent-Field/codeaf/internal/lane" @@ -62,10 +64,9 @@ type Completer interface { // usable: the provider refused, the model went quiet, or two attempts in a row // were not a JSON object this package could read. // -// It is ONE error on purpose. A caller has exactly one thing to do with any of -// those — nothing — and a taxonomy would invite a turn loop to handle a -// provider timeout differently from a bad enum, which is the beginning of a -// reflex that can break a turn. +// The error remains recognizable without losing its wrapped provider cause. +// The owner can recover a transient failure off the turn's path, while malformed +// output and permanent refusals do not trigger another background pass. var ErrReflexFailed = errors.New("reflex: no usable answer") // messageLimit is how much of one side of an exchange the model is shown, in @@ -400,15 +401,44 @@ func ask(ctx context.Context, c Completer, system, user string, read func(reply // WHAT SURVIVES IS THE PART THAT WAS NEVER ABOUT GENERATION: an answer this // package cannot use must not break a turn. A model that says nothing at all // abandons this session's reflex model and asks the configured low tier once. -// There is no loop. +// A transient request failure also tries that fallback, without declaring the +// primary unusable for the rest of the session. Both attempts share one bound. func call(ctx context.Context, c Completer, messages []ai.Message) (string, error) { // Every reflex request in the process passes through here, so this is the // one place its rows in the model-call log get their word. ctx = provider.WithCallTag(ctx, "reflex") ctx = provider.WithoutStream(ctx) - response, err := c.CompleteWithMessages(ctx, messages) + // Reserve half of the remaining patience for recovery. Only a bound client + // knows where to send that recovery; unbound callers keep their deadline. + attemptCtx := ctx + cancel := func() {} + pinned, canRecover := c.(bound) + if canRecover && pinned.fallback != "" { + if deadline, ok := ctx.Deadline(); ok { + attemptCtx, cancel = context.WithTimeout(ctx, time.Until(deadline)/2) + } + } + response, err := c.CompleteWithMessages(attemptCtx, messages) + cancel() if err != nil { - return "", err + // A cancelled owner never starts another paid request. A transient + // failure tries the small-work model once, with the RESERVED parent + // deadline rather than the primary attempt's expired child context. + kind, _ := provider.RouteFailureOf(err) + if ctx.Err() != nil || !canRecover || pinned.fallback == "" || kind != provider.RouteTransient { + return "", err + } + fallback := pinned + fallback.model = pinned.fallback + fallback.session = nil + response, err = fallback.CompleteWithMessages(ctx, messages) + if err != nil { + return "", err + } + if response == nil || AnsweredNothing(response) { + return "", errors.New("the fallback model answered nothing") + } + return response.Text(), nil } if response == nil { return "", errors.New("the model answered nothing") @@ -417,8 +447,7 @@ func call(ctx context.Context, c Completer, messages []ai.Message) (string, erro return response.Text(), nil } - pinned, ok := c.(bound) - if !ok || !pinned.abandon(boundModel(c)) { + if !canRecover || !pinned.abandon(boundModel(c)) { return response.Text(), nil } response, err = c.CompleteWithMessages(ctx, messages) diff --git a/internal/session/memory.go b/internal/session/memory.go index b76cdda1c9..cb16db45c4 100644 --- a/internal/session/memory.go +++ b/internal/session/memory.go @@ -37,7 +37,9 @@ package session // // A REFLEX FAILURE IS INVISIBLE TO THE TURN. The noticing pass runs after the // turn is sealed; a provider that does not answer leaves the turn exactly what -// it was, and the failure is journaled and said once per window. +// it was. A transient failure gets one delayed retry while the session remains +// open; only an exhausted failure is journaled and said once per window. The +// exchange is not queued across application restarts. import ( "bufio" @@ -898,9 +900,13 @@ func (a *Agent) noticeTurn(words, answer string, askMemory bool, kept []store.Me for index, memory := range kept { texts[index] = memory.Text } - found, err := reflex.Notice(a.memoryCtx, client, words, answer, texts, askMemory) + found, err := a.noticeWithRecovery(client, words, answer, texts, askMemory) if err != nil { - a.journalMemoryFailure("notice", err) + // Closing the chat cancels its background jobs deliberately; that is + // not a provider outage and does not need a failure notice. + if a.memoryCtx.Err() == nil { + a.journalMemoryFailure("notice", err) + } return } if found.State != nil { @@ -927,6 +933,29 @@ func (a *Agent) noticeTurn(words, answer string, askMemory bool, kept []store.Me a.keepNoticedDurable(found.Keep, found.Scope, replaced, found.Durability) } +// noticeWithRecovery gives a transient outage one later chance without putting +// another wait on the person's turn. The original exchange and numbered working +// set stay together, so a correction never applies to a newer turn's memories. +// The job remains owned by memoryCtx and memoryJobs, including its backoff. +func (a *Agent) noticeWithRecovery(client reflex.Completer, words, answer string, texts []string, askMemory bool) (reflex.NoticeResult, error) { + found, err := reflex.Notice(a.memoryCtx, client, words, answer, texts, askMemory) + kind, _ := provider.RouteFailureOf(err) + if err == nil || kind != provider.RouteTransient || a.memoryCtx.Err() != nil { + return found, err + } + timer := time.NewTimer(2 * time.Second) + defer timer.Stop() + select { + case <-a.memoryCtx.Done(): + return reflex.NoticeResult{}, a.memoryCtx.Err() + case <-timer.C: + } + if err := a.memoryCtx.Err(); err != nil { + return reflex.NoticeResult{}, err + } + return reflex.Notice(a.memoryCtx, client, words, answer, texts, askMemory) +} + // keepNoticed writes what the reading found through the one door and SAYS SO: // a line kept on the person's behalf is never kept in silence. func (a *Agent) keepNoticed(text, scope string, replaced *store.Memory) { @@ -1203,7 +1232,11 @@ func (a *Agent) journalMemoryFailure(via string, err error) { } a.memory.mu.Unlock() if window { - a.sayMemory("couldn't update memory just now · " + clipWords(reason, 120)) + if via == "notice" { + a.sayMemory("couldn't save this exchange to memory · use /remember to save an important detail") + } else { + a.sayMemory("couldn't update memory just now · " + clipWords(reason, 120)) + } } } diff --git a/internal/session/memory_recovery_test.go b/internal/session/memory_recovery_test.go new file mode 100644 index 0000000000..97e81154fd --- /dev/null +++ b/internal/session/memory_recovery_test.go @@ -0,0 +1,185 @@ +package session + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "testing/synctest" + "time" + + "github.com/Agent-Field/agentfield/sdk/go/ai" + "github.com/Agent-Field/codeaf/internal/store" +) + +type recoveringMemoryClient struct { + mu sync.Mutex + recoverAt time.Time + calls int + shown []string + alwaysFail bool + failure error +} + +func (c *recoveringMemoryClient) CompleteWithMessages(ctx context.Context, messages []ai.Message, _ ...ai.Option) (*ai.Response, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls++ + if c.failure != nil { + return nil, c.failure + } + if len(messages) > 1 { + c.shown = append(c.shown, messageText(messages[1])) + } + if c.alwaysFail || time.Now().Before(c.recoverAt) { + return nil, context.DeadlineExceeded + } + return textResponse(`{"keep":"I prefer concise answers","scope":"user","durability":"standing_rule"}`), nil +} + +func TestMemoryRecoveryKeepsOriginalExchangeOnce(t *testing.T) { + usageWriterFor(UsageLedgerPath()) + synctest.Test(t, func(t *testing.T) { + client := &recoveringMemoryClient{recoverAt: time.Now().Add(2 * time.Second)} + agent, brain := brainAgent(t, client, nil) + agent.stopLaneBeat() // The process-wide lane queue is outside the fake-clock bubble. + agent.learnFromTurn(userMessage{message: textMessage("user", "I prefer concise answers")}, "understood") + settleMemory(agent) + agent.memory.mu.Lock() + failureSaid := agent.memory.failureSaid + agent.memory.mu.Unlock() + if !failureSaid.IsZero() { + t.Fatal("recoverable outage was shown as a failure") + } + got := activeTexts(t, brain, store.OwnerUser) + if len(got) != 1 || got[0] != "I prefer concise answers" { + t.Fatalf("saved=%v", got) + } + client.mu.Lock() + defer client.mu.Unlock() + if client.calls < 2 { + t.Fatalf("calls=%d; recovery never ran", client.calls) + } + for _, shown := range client.shown { + if !strings.Contains(shown, "I prefer concise answers") { + t.Fatalf("original exchange lost: %q", shown) + } + } + }) +} + +func TestMemoryRecoveryCancellationStopsBackoff(t *testing.T) { + usageWriterFor(UsageLedgerPath()) + synctest.Test(t, func(t *testing.T) { + client := &recoveringMemoryClient{alwaysFail: true} + agent, brain := brainAgent(t, client, nil) + agent.stopLaneBeat() // The process-wide lane queue is outside the fake-clock bubble. + agent.learnFromTurn(userMessage{message: textMessage("user", "I prefer concise answers")}, "understood") + synctest.Wait() + client.mu.Lock() + before := client.calls + client.mu.Unlock() + if before == 0 { + t.Fatal("first request never ran") + } + agent.memoryStop() + settleMemory(agent) + time.Sleep(3 * time.Second) + client.mu.Lock() + after := client.calls + client.mu.Unlock() + agent.memory.mu.Lock() + failureSaid := agent.memory.failureSaid + agent.memory.mu.Unlock() + if !failureSaid.IsZero() { + t.Fatal("shutdown cancellation was shown as a failure") + } + if after != before { + t.Fatalf("retried after cancellation: %d -> %d", before, after) + } + if got := activeTexts(t, brain, store.OwnerUser); len(got) != 0 { + t.Fatalf("cancelled pass saved=%v", got) + } + }) +} + +func TestMemoryRecoveryExhaustionOffersManualSave(t *testing.T) { + usageWriterFor(UsageLedgerPath()) + synctest.Test(t, func(t *testing.T) { + client := &recoveringMemoryClient{alwaysFail: true} + agent, brain := brainAgent(t, client, nil) + agent.stopLaneBeat() + agent.learnFromTurn(userMessage{message: textMessage("user", "I prefer concise answers")}, "understood") + settleMemory(agent) + client.mu.Lock() + calls := client.calls + client.mu.Unlock() + if calls < 2 || calls > 4 { + t.Fatalf("unbounded or missing recovery: %d requests", calls) + } + agent.memory.mu.Lock() + said := strings.Join(agent.memory.said, "\n") + agent.memory.mu.Unlock() + if !strings.Contains(said, "use /remember") || strings.Contains(said, "context deadline exceeded") { + t.Fatalf("no actionable final notice: %q", said) + } + if got := activeTexts(t, brain, store.OwnerUser); len(got) != 0 { + t.Fatalf("failed pass saved=%v", got) + } + assertMemoryFailureJournal(t, brain, "context deadline exceeded") + if _, err := agent.Remember("I prefer concise answers"); err != nil { + t.Fatalf("manual save failed: %v", err) + } + if got := activeTexts(t, brain, store.OwnerUser); len(got) != 1 { + t.Fatalf("manual save not retained: %v", got) + } + }) +} + +func assertMemoryFailureJournal(t *testing.T, brain *store.Store, reason string) { + t.Helper() + events, err := brain.Events(0, 100) + if err != nil { + t.Fatal(err) + } + failures := 0 + for _, event := range events { + if event.Kind == store.EventMemoryWriteFailed { + failures++ + if !strings.Contains(string(event.Payload), reason) { + t.Fatalf("technical reason lost: %s", event.Payload) + } + } + } + if failures != 1 { + t.Fatalf("failure journal rows=%d", failures) + } +} + +func TestMemoryRecoveryPermanentFailureDoesNotRetry(t *testing.T) { + usageWriterFor(UsageLedgerPath()) + synctest.Test(t, func(t *testing.T) { + client := &recoveringMemoryClient{failure: errors.New("invalid credentials")} + agent, brain := brainAgent(t, client, nil) + agent.stopLaneBeat() + agent.learnFromTurn(userMessage{message: textMessage("user", "I prefer concise answers")}, "understood") + settleMemory(agent) + client.mu.Lock() + calls := client.calls + client.mu.Unlock() + if calls != 1 { + t.Fatalf("permanent failure retried %d times", calls) + } + if got := activeTexts(t, brain, store.OwnerUser); len(got) != 0 { + t.Fatalf("failed pass saved=%v", got) + } + agent.memory.mu.Lock() + said := strings.Join(agent.memory.said, "\n") + agent.memory.mu.Unlock() + if !strings.Contains(said, "use /remember") { + t.Fatalf("no actionable failure notice: %q", said) + } + assertMemoryFailureJournal(t, brain, "invalid credentials") + }) +} diff --git a/internal/session/prompts/system.md b/internal/session/prompts/system.md index a132a8d776..04d85c1bdd 100644 --- a/internal/session/prompts/system.md +++ b/internal/session/prompts/system.md @@ -178,7 +178,7 @@ deliverable it names, by its full path, and answer out of that. # Session facts - ATTACHED PICTURES TRAVEL IN THE MESSAGE WITH YOU, numbered per message: answer from the attached pixels and cite `[image #1]`, `[image #2]`, etc. Earlier pictureless tokens refer to the vision model's following answer. -- CROSS-CONVERSATION MEMORY is the `` block and lookups. `remember` keeps a lasting preference, rule or decision (timed things are automations). Without `remember`, say you cannot keep it; they can /remember it. +- CROSS-CONVERSATION MEMORY is the `` block and lookups. `remember` keeps a lasting preference, rule or decision (timed things are automations). When enabled, background memory can also notice durable details after a turn without a command or tool call; a later memory notice confirms what was saved. Do not claim a detail was saved before confirmation, or that /remember is required merely because you did not call `remember`. /remember saves directly if background saving fails. - DELIVERABLES ARE FILES. EVERY file you name carries its FULL ABSOLUTE PATH in `Project`'s working directory: `/research/notes.md`. Relative paths break references and can name the wrong task copy. - `bash` WAITS until a foreground call finishes or its armed bound keeps it running as a job. Never re-run running work, and never kill a job for being quiet. - THE PERSON'S OWN MESSAGE IS ATTACHED FOR YOU, verbatim, above whatever you write, on a task and every sub-task under it: never copy, summarise or contradict it, since the worker follows theirs where you disagree. From 281bae115d1ecc465e1f8d5d761039b87cd34421 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 18:15:11 -0400 Subject: [PATCH 2/6] Document bounded background memory recovery --- .../1901-memory-timeout-recovery.md | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) create mode 100644 docs/changes/unreleased/1901-memory-timeout-recovery.md diff --git a/docs/changes/unreleased/1901-memory-timeout-recovery.md b/docs/changes/unreleased/1901-memory-timeout-recovery.md new file mode 100644 index 0000000000..4fbedcb0e8 --- /dev/null +++ b/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. From a15d4519d4418f9ebffc33e0314a78ae10cd9889 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 18:15:34 -0400 Subject: [PATCH 3/6] Pin fallback deadline reservation and clarify recovery bounds --- internal/reflex/recovery_test.go | 10 +++++++++- internal/reflex/reflex.go | 14 ++++++-------- 2 files changed, 15 insertions(+), 9 deletions(-) diff --git a/internal/reflex/recovery_test.go b/internal/reflex/recovery_test.go index c5d8da87ce..5094558f4c 100644 --- a/internal/reflex/recovery_test.go +++ b/internal/reflex/recovery_test.go @@ -33,6 +33,14 @@ func TestStalledMemoryRequestLeavesTimeForFallback(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() @@ -51,7 +59,7 @@ func TestStalledMemoryRequestLeavesTimeForFallback(t *testing.T) { if !reflect.DeepEqual(c.models, []string{"primary", "fallback"}) { t.Fatal(c.models) } - if elapsed := time.Since(start); elapsed >= roles.PatienceFor(roles.RoleReflex) { + 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. diff --git a/internal/reflex/reflex.go b/internal/reflex/reflex.go index f92e4dfd6d..4d47de8f13 100644 --- a/internal/reflex/reflex.go +++ b/internal/reflex/reflex.go @@ -337,15 +337,13 @@ func validUsedMemoryIndices(indices []int, count int) []int { return used } -// ask makes one reflex call and, if the answer was not readable, exactly one -// repair call. The answer-budget recovery lives inside [call], where the -// response still carries its finish reason and usage. +// ask allows one JSON repair per pass. A transport failure can separately use +// the configured fallback inside [call], but all requests and the repair share +// the same patience. The owner may retry a transiently failed pass off the turn's +// path; malformed replies alone never start another pass. // -// ONE retry, never a loop. The repair asks for the same thing in four words -// with the model's own bad answer in front of it, which is what fixes a fenced -// or prefaced reply; a model that misses twice is a model that is going to miss -// again, and a third attempt on a per-turn call is a cost multiplier on a -// feature whose whole claim is that it is nearly free. +// The repair asks for the same thing with the unreadable answer in front of it. +// A second malformed reply is returned to the owner rather than entering a loop. func ask(ctx context.Context, c Completer, system, user string, read func(reply string) error) error { // Reflex bypasses the session's ordinary auxiliary adapter. It must still // share the same tier patience across retries rather than wait on HTTP caps. From da950b582b32af5c4468b65ba22efb37349756a8 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 18:17:19 -0400 Subject: [PATCH 4/6] Separate request recovery and failure reporting phases --- internal/reflex/reflex.go | 55 +++++++++++++++++++++++--------------- internal/session/memory.go | 13 +++++---- 2 files changed, 41 insertions(+), 27 deletions(-) diff --git a/internal/reflex/reflex.go b/internal/reflex/reflex.go index 4d47de8f13..403b77e11d 100644 --- a/internal/reflex/reflex.go +++ b/internal/reflex/reflex.go @@ -406,6 +406,34 @@ func call(ctx context.Context, c Completer, messages []ai.Message) (string, erro // one place its rows in the model-call log get their word. ctx = provider.WithCallTag(ctx, "reflex") ctx = provider.WithoutStream(ctx) + response, err := completeWithRecovery(ctx, c, messages) + if err != nil { + return "", err + } + if response == nil { + return "", errors.New("the model answered nothing") + } + if !AnsweredNothing(response) { + return response.Text(), nil + } + + pinned, ok := c.(bound) + if !ok || !pinned.abandon(boundModel(c)) { + return response.Text(), nil + } + response, err = c.CompleteWithMessages(ctx, messages) + if err != nil { + return "", err + } + if response == nil { + return "", errors.New("the fallback model answered nothing") + } + return response.Text(), nil +} + +// completeWithRecovery reserves time for a transient failure's fallback. +// It returns a response without decoding or applying session failover policy. +func completeWithRecovery(ctx context.Context, c Completer, messages []ai.Message) (*ai.Response, error) { // Reserve half of the remaining patience for recovery. Only a bound client // knows where to send that recovery; unbound callers keep their deadline. attemptCtx := ctx @@ -424,38 +452,21 @@ func call(ctx context.Context, c Completer, messages []ai.Message) (string, erro // deadline rather than the primary attempt's expired child context. kind, _ := provider.RouteFailureOf(err) if ctx.Err() != nil || !canRecover || pinned.fallback == "" || kind != provider.RouteTransient { - return "", err + return nil, err } fallback := pinned fallback.model = pinned.fallback fallback.session = nil response, err = fallback.CompleteWithMessages(ctx, messages) if err != nil { - return "", err + return nil, err } if response == nil || AnsweredNothing(response) { - return "", errors.New("the fallback model answered nothing") + return nil, errors.New("the fallback model answered nothing") } - return response.Text(), nil - } - if response == nil { - return "", errors.New("the model answered nothing") - } - if !AnsweredNothing(response) { - return response.Text(), nil - } - - if !canRecover || !pinned.abandon(boundModel(c)) { - return response.Text(), nil - } - response, err = c.CompleteWithMessages(ctx, messages) - if err != nil { - return "", err + return response, nil } - if response == nil { - return "", errors.New("the fallback model answered nothing") - } - return response.Text(), nil + return response, nil } // AnsweredNothing is the one failure this package can do anything about: a call diff --git a/internal/session/memory.go b/internal/session/memory.go index cb16db45c4..7e75fef899 100644 --- a/internal/session/memory.go +++ b/internal/session/memory.go @@ -902,11 +902,7 @@ func (a *Agent) noticeTurn(words, answer string, askMemory bool, kept []store.Me } found, err := a.noticeWithRecovery(client, words, answer, texts, askMemory) if err != nil { - // Closing the chat cancels its background jobs deliberately; that is - // not a provider outage and does not need a failure notice. - if a.memoryCtx.Err() == nil { - a.journalMemoryFailure("notice", err) - } + a.reportNoticeFailure(err) return } if found.State != nil { @@ -933,6 +929,13 @@ func (a *Agent) noticeTurn(words, answer string, askMemory bool, kept []store.Me a.keepNoticedDurable(found.Keep, found.Scope, replaced, found.Durability) } +// reportNoticeFailure excludes deliberate chat shutdown from provider failures. +func (a *Agent) reportNoticeFailure(err error) { + if a.memoryCtx.Err() == nil { + a.journalMemoryFailure("notice", err) + } +} + // noticeWithRecovery gives a transient outage one later chance without putting // another wait on the person's turn. The original exchange and numbered working // set stay together, so a correction never applies to a newer turn's memories. From 7a7c69a976ab5ddbdfb787f8fe5dc088513876b2 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 18:21:31 -0400 Subject: [PATCH 5/6] Keep automatic memory guidance within prompt capability and size limits --- internal/session/beltfacts.go | 8 ++++---- internal/session/prompts/system.md | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/internal/session/beltfacts.go b/internal/session/beltfacts.go index 8f5e5b7927..58884a41b7 100644 --- a/internal/session/beltfacts.go +++ b/internal/session/beltfacts.go @@ -29,7 +29,7 @@ import ( // composed at render time, each from THE PREDICATE THAT PUTS ITS TOOL ON THE // BELT. Where a tool is absent the fragment says what to do instead rather than // saying nothing, in the voice the page's own `remember` sentence already uses -// ("Without `remember`, say you cannot keep it"): a worker told the +// ("Without either, say saving is unavailable"): a worker told the // truth spends its turn on the work, and a worker told nothing spends it // guessing. // @@ -789,8 +789,8 @@ func promptWithBeltFacts(config Config) string { // honest: // // - WRITTEN FOR BOTH CASES ALREADY. The page names the tool and, in the same -// breath, says what to do without it — "Without `remember`, say you cannot -// keep it". prompt_belt_test.go holds these to that: where the belt +// breath, says what to do without it — "Without either, say saving is +// unavailable". prompt_belt_test.go holds these to that: where the belt // lacks the tool, the sentence stating the absence must be in the rendered // prompt. // @@ -801,7 +801,7 @@ func promptWithBeltFacts(config Config) string { // // The value is the marker the test looks for in the absent case. var promptNamesBeyondTheBelt = map[string]string{ - "remember": "Without `remember`, say you cannot keep it", + "remember": "Without either, say saving is unavailable", // A hand inherits the caller's page but intentionally carries a smaller, // closed belt; its appended tail is the truthful capability account. "ask": "Use `ask` only as the last rung", diff --git a/internal/session/prompts/system.md b/internal/session/prompts/system.md index 04d85c1bdd..210fc0926b 100644 --- a/internal/session/prompts/system.md +++ b/internal/session/prompts/system.md @@ -178,7 +178,7 @@ deliverable it names, by its full path, and answer out of that. # Session facts - ATTACHED PICTURES TRAVEL IN THE MESSAGE WITH YOU, numbered per message: answer from the attached pixels and cite `[image #1]`, `[image #2]`, etc. Earlier pictureless tokens refer to the vision model's following answer. -- CROSS-CONVERSATION MEMORY is the `` block and lookups. `remember` keeps a lasting preference, rule or decision (timed things are automations). When enabled, background memory can also notice durable details after a turn without a command or tool call; a later memory notice confirms what was saved. Do not claim a detail was saved before confirmation, or that /remember is required merely because you did not call `remember`. /remember saves directly if background saving fails. +- CROSS-CONVERSATION MEMORY: `` and lookups. `remember` saves details; times need automations. Enabled background saving confirms after turns. Without either, say saving is unavailable; use /remember in chat. - DELIVERABLES ARE FILES. EVERY file you name carries its FULL ABSOLUTE PATH in `Project`'s working directory: `/research/notes.md`. Relative paths break references and can name the wrong task copy. - `bash` WAITS until a foreground call finishes or its armed bound keeps it running as a job. Never re-run running work, and never kill a job for being quiet. - THE PERSON'S OWN MESSAGE IS ATTACHED FOR YOU, verbatim, above whatever you write, on a task and every sub-task under it: never copy, summarise or contradict it, since the worker follows theirs where you disagree. From 7a6ed127d3e5e70a53f73ad524ad323cb8c6fd9c Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 18:24:00 -0400 Subject: [PATCH 6/6] Preserve direct memory-save guidance for lean sessions --- internal/session/prompts/system.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/session/prompts/system.md b/internal/session/prompts/system.md index 210fc0926b..8c400767e9 100644 --- a/internal/session/prompts/system.md +++ b/internal/session/prompts/system.md @@ -178,7 +178,7 @@ deliverable it names, by its full path, and answer out of that. # Session facts - ATTACHED PICTURES TRAVEL IN THE MESSAGE WITH YOU, numbered per message: answer from the attached pixels and cite `[image #1]`, `[image #2]`, etc. Earlier pictureless tokens refer to the vision model's following answer. -- CROSS-CONVERSATION MEMORY: `` and lookups. `remember` saves details; times need automations. Enabled background saving confirms after turns. Without either, say saving is unavailable; use /remember in chat. +- CROSS-CONVERSATION MEMORY: `` and lookups. `remember` saves details; times need automations. Enabled background saving confirms after turns. Without either, say saving is unavailable; they can /remember it. - DELIVERABLES ARE FILES. EVERY file you name carries its FULL ABSOLUTE PATH in `Project`'s working directory: `/research/notes.md`. Relative paths break references and can name the wrong task copy. - `bash` WAITS until a foreground call finishes or its armed bound keeps it running as a job. Never re-run running work, and never kill a job for being quiet. - THE PERSON'S OWN MESSAGE IS ATTACHED FOR YOU, verbatim, above whatever you write, on a task and every sub-task under it: never copy, summarise or contradict it, since the worker follows theirs where you disagree.