amadan

oficina / memoria Public

🤖 memoria: start bounded catch-up after stale rulings

Draft under existing custody rules; model review and release approval remain pending.
Co-Authored-By: Codex <noreply@openai.com>
Paul Campbell pushed by paul@keymail.dev 1365bcbe5032915f46e64e1f6c6abc42f706bcb1 parent 80f813b
5 files changed, +370 −23
  • docs/superpowers/plans/2026-09-07-stale-refetch.md +11 −0
  • internal/memoria/custody.go +9 −20
  • internal/memoria/stale_refresh.go +53 −0
  • internal/memoria/stale_refresh_test.go +290 −0
  • internal/memoria/store.go +7 −3
diff --git a/docs/superpowers/plans/2026-09-07-stale-refetch.md b/docs/superpowers/plans/2026-09-07-stale-refetch.md
new file mode 100644
index 0000000..acd727f
--- /dev/null
+++ b/docs/superpowers/plans/2026-09-07-stale-refetch.md
@@ -0,0 +1,11 @@
+# Draft implementation note: STALE catch-up
+
+For independent model review; not an approved design or release.
+
+The existing §4.1 ruling requires withholding and refetching. §13 records that S2 supplies no replacement revision. The pinned wire response has only the asked item/revision and its ruling, so no revision can be inferred or incremented. The draft starts the existing S1 cursor poll for the affected source, obtains replacement coordinates from S1, then uses existing S3 ingestion. If the feed lacks the required announcement or reset protocol, the old copy stays withheld; this does not invent either protocol.
+
+The request returns its original withholding result. Refetch runs asynchronously with no asker on S1/S3, under the existing finite Poll budget from the stacked ingest-deadline draft. Caller cancellation is detached only at the request-to-background handoff. There is no permanent timer, authorization cache, or same-response reranking. Every later read needs its own S2 ruling.
+
+One in-flight poll per Store/source is shared by wake warm-up and reader-triggered catch-up. Repeated triggers join the existing attempt. Each attempt stops when Poll completes, fails, or exhausts its bounds; cleanup releases the slot for later attempts. Poll's durable cursor survives process loss; normal wake polling resumes from that cursor. No claim is made that a background goroutine prevents hibernation or guarantees completion.
+
+No policy notice strings change in this draft. Human copy review remains separate. The deployed baseline remains 9b55ab8; neither this note nor green tests approve that full release. Original custody, S1 budget, empty-S3 retirement, rollback and inherited-copy findings must be reviewed together before promotion.
diff --git a/internal/memoria/custody.go b/internal/memoria/custody.go
index b259c6b..938de0e 100644
--- a/internal/memoria/custody.go
+++ b/internal/memoria/custody.go
@@ -6,7 +6,6 @@ import (
"fmt"
"log/slog"
"sort"
- "sync"
"time"
"amadan.net/oficina/suite"
@@ -110,27 +109,17 @@ func (s *Siblings) Rule(ctx context.Context, asker suite.Subject, cands []siblin
func (s *Siblings) Warm(ctx context.Context, store *Store, logger *slog.Logger) {
ctx, cancel := context.WithTimeout(ctx, siblingPollDeadline)
defer cancel()
- var wg sync.WaitGroup
+ done := make([]<-chan struct{}, 0, len(s.byName))
for _, app := range s.byName {
- wg.Add(1)
- go func(sib sibling.Sibling) {
- defer wg.Done()
- res, err := store.Poll(ctx, s.client, sib)
- if err != nil {
- // The app's name and nothing else. A poll error can carry
- // a sibling's response body, and a body is content
- // (AGENTS.md) — so the error is not logged, only that
- // there was one.
- logger.Warn("sibling poll failed", "app", sib.Name)
- return
- }
- // Counts only: no item id, no title, no body.
- logger.Info("sibling polled", "app", sib.Name,
- "stored", res.Stored, "deleted", res.Deleted,
- "missing", res.Missing, "more", res.More)
- }(app)
+ done = append(done, s.startPoll(ctx, store, app, logger))
+ }
+ for _, job := range done {
+ select {
+ case <-job:
+ case <-ctx.Done():
+ return
+ }
}
- wg.Wait()
}
// WireSiblings reads the roster and attaches it to the app, warm-up and
diff --git a/internal/memoria/stale_refresh.go b/internal/memoria/stale_refresh.go
new file mode 100644
index 0000000..ed1bbb9
--- /dev/null
+++ b/internal/memoria/stale_refresh.go
@@ -0,0 +1,53 @@
+package memoria
+
+import (
+ "context"
+ "log/slog"
+
+ "amadan.net/oficina/memoria/internal/memoria/sibling"
+)
+
+func (s *Store) refetchStale(ctx context.Context, candidates []sibling.Candidate, at []int, dispositions []sibling.Disposition) {
+ owner, ok := s.custody.(*Siblings)
+ if !ok || owner == nil {
+ return
+ }
+ seen := make(map[string]bool)
+ for i, candidate := range candidates {
+ if dispositions[at[i]] == sibling.WithholdAndRefetch && !seen[candidate.Sibling.Name] {
+ seen[candidate.Sibling.Name] = true
+ owner.startPoll(context.WithoutCancel(ctx), s, candidate.Sibling, s.logger)
+ }
+ }
+}
+
+func (s *Siblings) startPoll(ctx context.Context, store *Store, sib sibling.Sibling, logger *slog.Logger) <-chan struct{} {
+ store.pollMu.Lock()
+ defer store.pollMu.Unlock()
+ if done := store.polling[sib.Name]; done != nil {
+ return done
+ }
+ if store.polling == nil {
+ store.polling = make(map[string]chan struct{})
+ }
+ done := make(chan struct{})
+ store.polling[sib.Name] = done
+ go func() {
+ defer func() {
+ store.pollMu.Lock()
+ delete(store.polling, sib.Name)
+ close(done)
+ store.pollMu.Unlock()
+ }()
+ result, err := store.Poll(ctx, s.client, sib)
+ if logger == nil {
+ return
+ }
+ if err != nil {
+ logger.Warn("sibling poll failed", "app", sib.Name)
+ return
+ }
+ logger.Info("sibling polled", "app", sib.Name, "stored", result.Stored, "deleted", result.Deleted, "missing", result.Missing, "more", result.More)
+ }()
+ return done
+}
diff --git a/internal/memoria/stale_refresh_test.go b/internal/memoria/stale_refresh_test.go
new file mode 100644
index 0000000..d8d3ac8
--- /dev/null
+++ b/internal/memoria/stale_refresh_test.go
@@ -0,0 +1,290 @@
+package memoria
+
+import (
+ "context"
+ "encoding/json"
+ "io"
+ "log/slog"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "amadan.net/oficina/memoria/internal/memoria/sibling"
+ "amadan.net/oficina/suite"
+)
+
+type staleFixture struct {
+ failFeed atomic.Bool
+ fixedRuling atomic.Value
+ store *Store
+ app *app
+ entered chan struct{}
+ release chan struct{}
+ calls atomic.Int64
+ askerCalls atomic.Int64
+ bodyCalls atomic.Int64
+}
+
+func newStaleFixture(t *testing.T) *staleFixture {
+ t.Helper()
+ f := &staleFixture{store: storeForTest(t), entered: make(chan struct{}, 20), release: make(chan struct{})}
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Header.Get("Authorization") != "Bearer consumer-token" {
+ t.Error("missing consumer token")
+ w.WriteHeader(401)
+ return
+ }
+ switch r.URL.Path {
+ case suite.RouteS2:
+ f.askerCalls.Add(1)
+ var req suite.S2Request
+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+ t.Error(err)
+ return
+ }
+ if req.Asker != (suite.Subject{Kind: suite.SubjectAddress, Value: "reader@example.test"}) {
+ t.Errorf("wrong asker: %#v", req.Asker)
+ }
+ rows := make([]map[string]any, 0, len(req.Pairs))
+ for _, pair := range req.Pairs {
+ ruling := suite.UNKNOWN
+ if pair.ItemID == "item" && pair.Revision == 19 {
+ ruling = suite.STALE
+ }
+ if pair.ItemID == "item" && pair.Revision == 73 {
+ ruling = suite.VISIBLE
+ }
+ if override := f.fixedRuling.Load(); override != nil {
+ ruling = override.(suite.Authorization)
+ }
+ rows = append(rows, map[string]any{"item_id": pair.ItemID, "revision": pair.Revision, "result": ruling})
+ }
+ json.NewEncoder(w).Encode(map[string]any{"results": rows})
+ case suite.RouteS1:
+ f.calls.Add(1)
+ f.entered <- struct{}{}
+ select {
+ case <-f.release:
+ case <-r.Context().Done():
+ return
+ }
+ if f.failFeed.Load() {
+ w.WriteHeader(http.StatusServiceUnavailable)
+ return
+ }
+ switch r.URL.Query().Get("cursor") {
+ case "c19":
+ json.NewEncoder(w).Encode(suite.Feed{Cursor: "c73", Rows: []suite.FeedRow{row("item", 73, false)}})
+ case "c73":
+ json.NewEncoder(w).Encode(suite.Feed{Cursor: "c73"})
+ default:
+ t.Errorf("unexpected cursor %q", r.URL.Query().Get("cursor"))
+ w.WriteHeader(500)
+ }
+ case suite.RouteS3 + "/item":
+ f.bodyCalls.Add(1)
+ if r.URL.Query().Get("revision") != "73" {
+ t.Errorf("guessed revision %q", r.URL.Query().Get("revision"))
+ w.WriteHeader(409)
+ return
+ }
+ io.WriteString(w, "NewQuarterlyCanary quarterly planning.")
+ default:
+ t.Errorf("unexpected route %s", r.URL.Path)
+ w.WriteHeader(404)
+ }
+ }))
+ t.Cleanup(func() {
+ select {
+ case <-f.release:
+ default:
+ close(f.release)
+ }
+ f.wait(t)
+ srv.Close()
+ })
+ f.store.custody = &Siblings{client: sibling.NewForTest(), byName: map[string]sibling.Sibling{"Docs": {Name: "Docs", BaseURL: srv.URL, Token: "consumer-token"}}}
+ if _, err := f.store.Ingest(context.Background(), IngestSpec{Source: "Docs", Key: "item", Body: "OldQuarterlyCanary quarterly planning.", ItemID: "item", SiblingRevision: 19}); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.store.setCursor(context.Background(), "Docs", "c19"); err != nil {
+ t.Fatal(err)
+ }
+ f.app = &app{store: f.store, logger: slog.New(slog.NewTextHandler(io.Discard, nil)), testIdentity: "reader@example.test"}
+ return f
+}
+
+func (f *staleFixture) search(ctx context.Context) *httptest.ResponseRecorder {
+ rec := httptest.NewRecorder()
+ req := httptest.NewRequest(http.MethodGet, "/search?q=quarterly", nil).WithContext(ctx)
+ f.app.searchResults(rec, req)
+ return rec
+}
+
+func TestStaleSearchStartsBoundedCatchupWithoutWaiting(t *testing.T) {
+ f := newStaleFixture(t)
+ started := time.Now()
+ result := f.search(context.Background())
+ if time.Since(started) > sibling.Deadline {
+ t.Error("page waited for refetch")
+ }
+ if strings.Contains(result.Body.String(), "OldQuarterlyCanary") || !strings.Contains(result.Body.String(), "refetching") {
+ t.Fatal("stale page did not withhold with existing notice")
+ }
+ select {
+ case <-f.entered:
+ case <-time.After(time.Second):
+ t.Fatal("STALE said refetching but no S1 catchup started")
+ }
+ close(f.release)
+ limit := time.Now().Add(2 * time.Second)
+ for {
+ cursor, err := f.store.cursor(context.Background(), "Docs")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if cursor == "c73" {
+ break
+ }
+ if time.Now().After(limit) {
+ t.Fatal("catchup did not apply new revision")
+ }
+ time.Sleep(time.Millisecond)
+ }
+ before := f.askerCalls.Load()
+ next := f.search(context.Background())
+ if !strings.Contains(next.Body.String(), "NewQuarterlyCanary") || strings.Contains(next.Body.String(), "OldQuarterlyCanary") {
+ t.Error("next search did not use refreshed content")
+ }
+ if f.askerCalls.Load() <= before {
+ t.Error("refreshed body rendered without fresh S2")
+ }
+ if f.bodyCalls.Load() != 1 {
+ t.Errorf("S3 calls%d", f.bodyCalls.Load())
+ }
+}
+
+func (f *staleFixture) wait(t *testing.T) {
+ t.Helper()
+ f.store.pollMu.Lock()
+ done := f.store.polling["Docs"]
+ f.store.pollMu.Unlock()
+ if done != nil {
+ select {
+ case <-done:
+ case <-time.After(3 * time.Second):
+ t.Fatal("refresh did not finish")
+ }
+ }
+}
+
+func TestStaleRefreshCoalescesWithWakeAndRepeatedRequests(t *testing.T) {
+ for _, warmFirst := range []bool{false, true} {
+ t.Run(map[bool]string{false: "search first", true: "warm first"}[warmFirst], func(t *testing.T) {
+ f := newStaleFixture(t)
+ owner := f.store.custody.(*Siblings)
+ warmDone := make(chan struct{})
+ if warmFirst {
+ go func() { owner.Warm(context.Background(), f.store, f.app.logger); close(warmDone) }()
+ } else {
+ f.search(context.Background())
+ }
+ select {
+ case <-f.entered:
+ case <-time.After(time.Second):
+ t.Fatal("first poll not started")
+ }
+ if !warmFirst {
+ go func() { owner.Warm(context.Background(), f.store, f.app.logger); close(warmDone) }()
+ }
+ for n := 0; n < 5; n++ {
+ f.search(context.Background())
+ }
+ select {
+ case <-f.entered:
+ t.Error("duplicate same-source poll started while first blocked")
+ case <-time.After(30 * time.Millisecond):
+ }
+ if f.calls.Load() != 1 {
+ t.Errorf("S1 calls while blocked%d", f.calls.Load())
+ }
+ close(f.release)
+ select {
+ case <-warmDone:
+ case <-time.After(3 * time.Second):
+ t.Fatal("warm did not join refresh")
+ }
+ f.wait(t)
+ })
+ }
+}
+
+func TestStaleRefreshOutlivesCancelledReader(t *testing.T) {
+ f := newStaleFixture(t)
+ ctx, cancel := context.WithCancel(context.Background())
+ f.search(ctx)
+ select {
+ case <-f.entered:
+ case <-time.After(time.Second):
+ t.Fatal("refresh not started")
+ }
+ cancel()
+ close(f.release)
+ f.wait(t)
+ cursor, err := f.store.cursor(context.Background(), "Docs")
+ if err != nil || cursor != "c73" {
+ t.Errorf("reader cancellation killed refresh: cursor%q err%v", cursor, err)
+ }
+}
+
+func TestStaleRefreshFailurePreservesCursorAndCanRetry(t *testing.T) {
+ f := newStaleFixture(t)
+ f.failFeed.Store(true)
+ f.search(context.Background())
+ select {
+ case <-f.entered:
+ case <-time.After(time.Second):
+ t.Fatal("refresh not started")
+ }
+ close(f.release)
+ f.wait(t)
+ cursor, err := f.store.cursor(context.Background(), "Docs")
+ if err != nil || cursor != "c19" {
+ t.Errorf("failed refresh advanced cursor: %q %v", cursor, err)
+ }
+ docs, err := f.store.Summaries(context.Background())
+ if err != nil || len(docs) != 1 {
+ t.Fatal("failed refresh deleted old corpus")
+ }
+ if f.bodyCalls.Load() != 0 {
+ t.Error("failed S1 guessed a revision for S3")
+ }
+ f.failFeed.Store(false)
+ f.search(context.Background())
+ f.wait(t)
+ cursor, err = f.store.cursor(context.Background(), "Docs")
+ if err != nil || cursor != "c73" {
+ t.Errorf("failed refresh blocked retry: %q %v", cursor, err)
+ }
+}
+
+func TestOnlyStaleTriggersRefresh(t *testing.T) {
+ for _, ruling := range []suite.Authorization{suite.VISIBLE, suite.NOT_VISIBLE, suite.UNKNOWN} {
+ t.Run(string(ruling), func(t *testing.T) {
+ f := newStaleFixture(t)
+ f.fixedRuling.Store(ruling)
+ f.search(context.Background())
+ select {
+ case <-f.entered:
+ t.Error("non-stale ruling triggered ingest")
+ case <-time.After(30 * time.Millisecond):
+ }
+ if f.calls.Load() != 0 {
+ t.Error("non-stale ruling fetched feed")
+ }
+ })
+ }
+}
diff --git a/internal/memoria/store.go b/internal/memoria/store.go
index 5860c0a..d4d19e5 100644
--- a/internal/memoria/store.go
+++ b/internal/memoria/store.go
@@ -6,6 +6,7 @@ import (
"fmt"
"log/slog"
"strings"
+ "sync"
"github.com/carlosframework/rastrillo/db"
@@ -21,9 +22,11 @@ import (
// works in plaintext that never reaches disk; everything below it works
// in ciphertext.
type Store struct {
- db *db.DB
- sealer *seal.Sealer
- ix *index.Index
+ pollMu sync.Mutex
+ polling map[string]chan struct{}
+ db *db.DB
+ sealer *seal.Sealer
+ ix *index.Index
// logger is optional. A dropped breaker counter is not allowed to
// fail a search, but it is not allowed to be silent either: a
// breaker that quietly stops counting reads as "no anomaly" forever,
@@ -751,6 +754,7 @@ func (s *Store) rule(ctx context.Context, asker suite.Subject, all []Result) Pag
// failing the page. If that ever stops being true the fix is a
// gap marker, not a louder comment.
s.recordForBreaker(context.WithoutCancel(ctx), asker, cands, at, disp)
+ s.refetchStale(ctx, cands, at, disp)
}
// **Notices are aggregated once, here, in result order.** The first