oficina / memoria Public

🤖 feat: poll a sibling's feed, advancing the cursor only on applied pages

S1 to stored documents. The order is the design: read a page, apply every
row in it, *then* store the cursor. §5.1 and gate 7 — a cursor moved past
a page nobody applied skips changes the sibling will never offer again,
silently, because from its side the consumer read that page.

Re-applying is chosen as the cheap failure. A row that fails leaves the
cursor put, so the whole page comes back and its earlier rows are applied
again; Ingest is keyed on (source, external id) and deleting an absent
row is a no-op. The expensive failure is skipping a row nobody will
mention again.

**A listed item the sibling then refuses is counted, never deleted.**
§4.2 makes S3's 404 cover both "gone" and "not team-visible" on purpose,
so memoria cannot tell which — and deleting on it would let a visibility
change erase history the sibling still has. That count is what §8.1's S3
breaker watches.

**The cursor is sealed, and an unreadable one is an error.** An empty
cursor is not a degraded value: it is the instruction "start from the
beginning", so a corrupted row read as empty would re-announce the
sibling's whole corpus. `PolledAt` moves only when a poll completes,
because it is what the sources page reports the corpus current as of.

`More` exists because "there is no more" and "I stopped looking" are the
same value to a caller that only checks the error. A feed whose cursor
stops advancing is bounded rather than walked to the page limit.

Eleven mutations. **Three survived the first pass and all three were
worth having:**

- **The cursor stored before the rows were applied** — the file's central
  claim — passed everything. Every existing failure case failed at the S1
  fetch, where no row has been applied yet and the two orders are
  indistinguishable. The case that catches it is a row failing *inside* a
  page that fetched fine.
- **Hard-coding `SiblingRevision: 1`** passed, because every fixture used
  revision 1. Now 7 and 9.
- **A row with no item id** could be skipped freely, and the reason is
  the better finding: memoria's guard against it **could not fail.**
  `suite.DecodeS1` refuses the whole page first ("S1 row %d has no
  item_id"), so the check never ran. The guard is deleted and the test
  now asserts the module's refusal — the thing memoria actually depends
  on — so it goes red the day the module softens.

`sibling.NewForTest` is a named seam under the same `cmd/` gate as
`AppForTest`: in production it would be §6.4's cross-team refusal
removed, and it would compile and serve perfectly well.

`make ci` green.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01H3VKtXnq6fSmM2F11Bjcsy

Paul Campbell pushed by paul@keymail.dev 1 month ago 1fc0f93a838552bba67fe768336c875efbfd7859 parent 3fdfb9e

9 files changed, +855 −0

internal/memoria/gorm.go +7 −0
internal/memoria/migrations/0006_create_sibling_feeds.sql +4 −0
internal/memoria/migrations/schema.sql +4 −0
internal/memoria/models.go +41 −0
internal/memoria/seal/seal.go +15 −0
internal/memoria/sibling/sibling.go +17 −0
internal/memoria/siblingsync.go +280 −0
internal/memoria/siblingsync_test.go +482 −0
internal/memoriatest/gates_test.go +5 −0

diff --git a/internal/memoria/gorm.go b/internal/memoria/gorm.go

index 3702232..c59b287 100644

--- a/internal/memoria/gorm.go

+++ b/internal/memoria/gorm.go

@@ -5,3 +5,10 @@ import "gorm.io/gorm"

// gormTx is the transaction handle type. Aliased so store.go's closures

// read as prose rather than as generics.

type gormTx = gorm.DB

+// errRecordNotFound is GORM's "no such row". Named here so a caller reads

+// as a sentence and, more usefully, so the one place that decides what a

+// missing row MEANS is a place somebody can find: for a feed position it

+// means "never polled", which is an empty cursor and a legitimate

+// instruction, and everywhere else it is a 404.

+var errRecordNotFound = gorm.ErrRecordNotFound

diff --git a/internal/memoria/migrations/0006_create_sibling_feeds.sql b/internal/memoria/migrations/0006_create_sibling_feeds.sql

new file mode 100644

index 0000000..d696031

--- /dev/null

+++ b/internal/memoria/migrations/0006_create_sibling_feeds.sql

@@ -0,0 +1,4 @@

+CREATE TABLE `sibling_feeds` (`id` integer PRIMARY KEY AUTOINCREMENT,`app` text NOT NULL,`sealed_cursor` blob,`polled_at` datetime,`reconciling` numeric NOT NULL DEFAULT false,`created_at` datetime,`updated_at` datetime);

+CREATE UNIQUE INDEX `idx_sibling_feeds_app` ON `sibling_feeds`(`app`);

diff --git a/internal/memoria/migrations/schema.sql b/internal/memoria/migrations/schema.sql

index 5e670b1..0f3a6ac 100644

--- a/internal/memoria/migrations/schema.sql

+++ b/internal/memoria/migrations/schema.sql

@@ -7,6 +7,8 @@ CREATE TABLE `connectors` (`id` integer PRIMARY KEY AUTOINCREMENT,`kind` text NO

CREATE TABLE `documents` (`id` integer PRIMARY KEY AUTOINCREMENT,`source` text NOT NULL,`external_id` text NOT NULL,`revision` integer NOT NULL DEFAULT 1,`sealed_title` blob NOT NULL,`sealed_body` blob NOT NULL,`bytes` integer,`created_at` datetime,`updated_at` datetime, `body_print` text NOT NULL DEFAULT "", `sealed_path` blob, `sibling_item_id` blob, `sibling_revision` integer NOT NULL DEFAULT 0);

CREATE TABLE `source_pages` (`id` integer PRIMARY KEY AUTOINCREMENT,`connector_id` integer NOT NULL,`page_print` text NOT NULL,`sealed_page_id` blob NOT NULL,`sealed_path` blob,`path_print` text,`seen_edited` text,`ingested_edited` text,`ingested_path` text,`state` text NOT NULL DEFAULT "pending",`error_code` text,`last_seen` datetime,`document_id` integer,`created_at` datetime,`updated_at` datetime);

CREATE TABLE `sync_runs` (`id` integer PRIMARY KEY AUTOINCREMENT,`connector_id` integer NOT NULL,`state` text NOT NULL,`roots` integer,`total` integer NOT NULL DEFAULT -1,`fetched` integer,`skipped` integer,`empty` integer,`failed_pages` integer,`unread_blocks` integer,`error_code` text,`heartbeat` datetime,`started_at` datetime,`finished_at` datetime);

@@ -19,6 +21,8 @@ CREATE INDEX `idx_documents_external_id` ON `documents`(`external_id`);

CREATE INDEX `idx_documents_source` ON `documents`(`source`);

+CREATE UNIQUE INDEX `idx_sibling_feeds_app` ON `sibling_feeds`(`app`);

CREATE INDEX `idx_source_pages_connector_id` ON `source_pages`(`connector_id`);

CREATE INDEX `idx_source_pages_page_print` ON `source_pages`(`page_print`);

diff --git a/internal/memoria/models.go b/internal/memoria/models.go

index a8fc2ec..83eb137 100644

--- a/internal/memoria/models.go

+++ b/internal/memoria/models.go

@@ -12,6 +12,7 @@ var Models = []any{

&Connector{},

&SyncRun{},

&SourcePage{},

+ &SiblingFeed{},

}

// Document is one thing memoria holds.

@@ -125,3 +126,43 @@ type Chunk struct {

SealedText []byte `gorm:"not null"`

CreatedAt time.Time

}

+// SiblingFeed is memoria's position in one sibling's change feed.

+//

+// One row per configured app of record, created on first poll.

+type SiblingFeed struct {

+ ID int64

+ // App is the sibling's configured name, which is also the Source on

+ // every document it owns. The unique index is what makes "one

+ // position per app" a property of the schema rather than of the code

+ // that writes it.

+ App string `gorm:"uniqueIndex;not null"`

+ // SealedCursor is the sibling's own opaque cursor, sealed

+ // (seal.PurposeCursor) because memoria does not know what is in it.

+ //

+ // **Written only after a page has been fully applied.** §5.1 and gate

+ // 7: a cursor moved past a page nobody applied skips changes the

+ // sibling will never offer again, and the sibling has no way to know

+ // it happened. Every failure path in Poll returns before this column

+ // is touched.

+ SealedCursor []byte

+ // PolledAt is when the last poll COMPLETED, not when it started.

+ //

+ // It is what the sources page says the corpus is current as of, so a

+ // poll that failed halfway must not move it — §8.1: "a failed or

+ // stale S1 → the page states the corpus is current as of a time,

+ // rather than implying freshness".

+ PolledAt time.Time

+ // Reconciling marks a source whose sibling has been restored behind

+ // memoria (§4.1's post-restore rule 1): its chunks leave the ranking

+ // entirely until a re-announce completes, so the other sources fill

+ // the page to its full depth instead of it rendering nearly empty.

+ //

+ // Not set by anything yet — AHEAD is not on the wire (§13) — and here

+ // rather than added later because the column is what makes the

+ // re-announce state durable across the hibernation that is certain to

+ // happen in the middle of one.

+ Reconciling bool `gorm:"not null;default:false"`

+ CreatedAt time.Time

+ UpdatedAt time.Time

diff --git a/internal/memoria/seal/seal.go b/internal/memoria/seal/seal.go

index 32dd479..6749bd2 100644

--- a/internal/memoria/seal/seal.go

+++ b/internal/memoria/seal/seal.go

@@ -63,6 +63,21 @@ const (

// at all. "One purpose per kind of thing sealed" earns its keep here:

// the kinds differ by who receives them, not by what they look like.

PurposeItemID = "sibling.item_id"

+ // PurposeCursor seals a sibling's S1 cursor.

+ //

+ // Reversible, because the whole use of a cursor is handing it back on

+ // the next poll. Its own purpose rather than PurposeItemID's for the

+ // reason that one is separate from PurposeCoordinate: these are the

+ // values memoria gives back to somebody else, and a cursor sent where

+ // an item id belongs would be a well-formed request about the wrong

+ // thing rather than a request that fails.

+ //

+ // Sealed at all because a cursor is the sibling's own encoding and

+ // memoria does not know what is in it. §4.1 requires it to be treated

+ // as opaque — never parsed, never ordered — and storing plaintext

+ // something whose contents are unknown is how a coordinate turns out

+ // to have been a timestamp, a row id, or a count.

+ PurposeCursor = "sibling.cursor"

)

func sha256New() hash.Hash { return sha256.New() }

diff --git a/internal/memoria/sibling/sibling.go b/internal/memoria/sibling/sibling.go

index cbf6451..a7d2b1c 100644

--- a/internal/memoria/sibling/sibling.go

+++ b/internal/memoria/sibling/sibling.go

@@ -68,6 +68,23 @@ func New(team string, dial suite.DialFunc) *Client {

return &Client{http: suite.NewSiblingClient(team, dial)}

}

+// NewForTest returns a Client with a PLAIN HTTP CLIENT and no team

+// binding, so a test can point it at a loopback server.

+//

+// **Named so it cannot be reached by accident**, exactly as `AppForTest`

+// is, and held by the same gate: the production-seam scan over `cmd/`

+// lists it, so wiring it into main fails the build's tests rather than

+// quietly removing §6.4's cross-team refusal from a running instance.

+//

+// It exists because the real constructor's transport refuses any host

+// outside the team's own label — correctly, and before it dials — and a

+// httptest server is on 127.0.0.1. That refusal is the module's to test

+// and has its own fixture; what the tests using this seam are about is

+// what memoria does with an answer, or the absence of one.

+func NewForTest() *Client {

+ return &Client{http: &http.Client{}}

// Authorize asks one sibling whether this person may read these items, at

// these revisions, right now.

diff --git a/internal/memoria/siblingsync.go b/internal/memoria/siblingsync.go

new file mode 100644

index 0000000..636dda9

--- /dev/null

+++ b/internal/memoria/siblingsync.go

@@ -0,0 +1,280 @@

+package memoria

+import (

+ "context"

+ "errors"

+ "fmt"

+ "time"

+ "amadan.net/oficina/suite"

+ "amadan.net/oficina/memoria/internal/memoria/seal"

+ "amadan.net/oficina/memoria/internal/memoria/sibling"

+// maxFeedPages bounds one poll.

+//

+// A sibling whose cursor never advances, or whose feed loops, must not

+// hold a wake open forever. A ceiling on damage rather than a

+// measurement: at a few hundred rows a page this is far past any ordinary

+// catch-up, and reaching it is reported as an INCOMPLETE poll rather than

+// as a finished one — the difference between "there is no more" and "I

+// stopped looking" is the difference this whole file is about.

+const maxFeedPages = 50

+// PollResult is what one poll did. Counts only: no item id, no title, no

+// body, so it can go in a log line or on the sources page (AGENTS.md).

+type PollResult struct {

+ Stored int

+ Deleted int

+ // Missing counts rows the sibling listed and then would not serve.

+ //

+ // **Not an error, and emphatically not a deletion.** §4.2 makes S3's

+ // 404 cover both "gone" and "not team-visible", deliberately, so

+ // memoria cannot tell which — and deleting on it would let a

+ // visibility change erase history that the sibling still has. What it

+ // feeds is §8.1's S3 breaker, which is the thing that notices when

+ // this happens wholesale.

+ Missing int

+ // More means the poll stopped before the sibling ran out. What

+ // memoria holds is consistent; it is not current, and the sources

+ // page must not say it is.

+ More bool

+// Poll brings memoria's copy of one sibling's content up to date.

+//

+// **The cursor advances only after a page has been fully applied**, and

+// that ordering is the whole design. §5.1 and gate 7: a timed-out,

+// rate-limited or half-applied enumeration returns no information rather

+// than no documents, and a cursor moved past a page nobody applied skips

+// changes the sibling will never offer again — silently, because the

+// sibling has no way to know. Every failure below returns before the

+// cursor is stored.

+//

+// **A failed poll deletes nothing**, which is the same rule from the other

+// end. Deletions arrive as `gone` rows on a page that was fetched

+// successfully; they are never inferred from an item's absence, because to

+// a consumer that infers, a short page and a complete one look the same.

+//

+// Re-applying is the cheap failure and is chosen deliberately. When a row

+// fails mid-page the cursor stays put, so the whole page comes back next

+// time and its earlier rows are applied again — which is safe, since

+// Ingest is keyed on (source, external id) and deleting an absent row is a

+// no-op. The expensive failure is skipping a row nobody will mention

+// again.

+func (s *Store) Poll(ctx context.Context, c *sibling.Client, sib sibling.Sibling) (PollResult, error) {

+ var res PollResult

+ cursor, err := s.cursor(ctx, sib.Name)

+ if err != nil {

+ return res, err

+ }

+ for page := 0; ; page++ {

+ if page >= maxFeedPages {

+ // Neither an error nor a completed poll: what memoria has is

+ // consistent, and there is more. Returning nil with More unset

+ // would let the caller call the corpus current.

+ res.More = true

+ return res, nil

+ }

+ feed, err := c.Feed(ctx, sib, cursor)

+ if err != nil {

+ return res, fmt.Errorf("poll %s: %w", sib.Name, err)

+ }

+ for _, row := range feed.Rows {

+ if err := s.applyRow(ctx, c, sib, row, &res); err != nil {

+ return res, fmt.Errorf("poll %s: %w", sib.Name, err)

+ }

+ // The page is applied. Only now may the position move.

+ if feed.Cursor != "" && feed.Cursor != cursor {

+ if err := s.setCursor(ctx, sib.Name, feed.Cursor); err != nil {

+ return res, err

+ }

+ cursor = feed.Cursor

+ continue

+ }

+ // The cursor did not move. If the page was empty, the sibling

+ // answered and has nothing further — **the one place an empty

+ // result may be read as "nothing", and only because Feed returns

+ // an error for every other reason it could be empty.** If the page

+ // had rows, a cursor that does not advance would loop forever, so

+ // stop and say the poll did not conclude.

+ res.More = len(feed.Rows) > 0

+ if !res.More {

+ if err := s.markPolled(ctx, sib.Name); err != nil {

+ return res, err

+ }

+ return res, nil

+ }

+func (s *Store) applyRow(ctx context.Context, c *sibling.Client, sib sibling.Sibling, row suite.FeedRow, res *PollResult) error {

+ // **There is no check for an empty item id here, and that is

+ // deliberate.** One was written; it could not fail. Every row reaching

+ // this function came out of `suite.DecodeS1`, which refuses the whole

+ // page — "S1 row %d has no item_id" (s1.go) — so the guard never ran,

+ // and a mutation deleting it left the suite green because the module

+ // was doing the work. A defence nothing can reach is a guard written

+ // to pass, whatever the intent.

+ //

+ // What memoria depends on is the module's refusal, and what makes that

+ // dependency visible is a test asserting the MODULE behaves that way

+ // rather than a duplicate check here pretending to.

+ if row.Gone {

+ n, err := s.deleteBySiblingItem(ctx, sib.Name, row.ItemID)

+ if err != nil {

+ return err

+ }

+ res.Deleted += n

+ return nil

+ }

+ body, err := c.Content(ctx, sib, row.ItemID, row.Revision)

+ switch {

+ case errors.Is(err, suite.ErrNotAvailable):

+ // Listed, then not served. See PollResult.Missing: counted, not

+ // deleted, not failed.

+ res.Missing++

+ return nil

+ case err != nil:

+ return err

+ }

+ // The item id is the stable key, so ExternalID's fingerprint is

+ // computed over it. A title would not be: a rename would make a second

+ // document and leave the first indexed under the old name.

+ _, err = s.Ingest(ctx, IngestSpec{

+ Source: sib.Name,

+ Key: row.ItemID,

+ Body: string(body),

+ ItemID: row.ItemID,

+ SiblingRevision: row.Revision,

+ // A sibling's item can legitimately be empty — a stub page, a

+ // heading with nothing under it yet. Failing the row would let one

+ // empty document stop a whole catch-up; skipping it would make the

+ // page an absence with nothing to explain it. Same reasoning as

+ // the connectors'.

+ AllowTitleOnly: true,

+ })

+ switch {

+ case errors.Is(err, ErrIndexLag):

+ // Stored, and only the index is behind. The document is in the

+ // database, so the cursor may pass this row: re-fetching it would

+ // not fix the index either, and the app says the lag out loud.

+ case err != nil:

+ return err

+ }

+ res.Stored++

+ return nil

+// cursor reads memoria's position in one sibling's feed.

+//

+// **A cursor that will not open is an error, never an empty cursor.** An

+// empty cursor means "start from the beginning", which is a legitimate and

+// very different instruction: it would re-announce the sibling's entire

+// corpus, and — worse — it is exactly what a corrupted row would silently

+// turn into.

+func (s *Store) cursor(ctx context.Context, app string) (string, error) {

+ var row SiblingFeed

+ err := s.db.G.WithContext(ctx).Where("app = ?", app).First(&row).Error

+ switch {

+ case errors.Is(err, errRecordNotFound):

+ // No row yet: the first poll of a newly configured sibling starts

+ // from the beginning, which is what an empty cursor asks for.

+ return "", nil

+ case err != nil:

+ return "", err

+ }

+ if len(row.SealedCursor) == 0 {

+ return "", nil

+ }

+ raw, err := s.sealer.Open(seal.PurposeCursor, row.SealedCursor)

+ if err != nil {

+ return "", fmt.Errorf("open cursor for %s: %w", app, err)

+ }

+ return string(raw), nil

+// setCursor stores a position and marks the poll current, in one write.

+//

+// The two move together on purpose: PolledAt is what the sources page

+// reports the corpus as current as of, and a position that advanced

+// without the timestamp — or the reverse — would make one of the two lie.

+func (s *Store) setCursor(ctx context.Context, app, cursor string) error {

+ sealed, err := s.sealer.Seal(seal.PurposeCursor, []byte(cursor))

+ if err != nil {

+ return err

+ }

+ return s.upsertFeed(ctx, app, map[string]any{

+ "sealed_cursor": sealed,

+ "polled_at": time.Now().UTC(),

+ })

+// markPolled records that a poll reached the end of the feed without

+// moving the cursor — the sibling answered and had nothing new.

+func (s *Store) markPolled(ctx context.Context, app string) error {

+ return s.upsertFeed(ctx, app, map[string]any{"polled_at": time.Now().UTC()})

+func (s *Store) upsertFeed(ctx context.Context, app string, fields map[string]any) error {

+ return s.db.G.WithContext(ctx).Transaction(func(tx *gormTx) error {

+ var row SiblingFeed

+ err := tx.Where("app = ?", app).First(&row).Error

+ switch {

+ case errors.Is(err, errRecordNotFound):

+ row = SiblingFeed{App: app}

+ if v, ok := fields["sealed_cursor"].([]byte); ok {

+ row.SealedCursor = v

+ }

+ if v, ok := fields["polled_at"].(time.Time); ok {

+ row.PolledAt = v

+ }

+ return tx.Create(&row).Error

+ case err != nil:

+ return err

+ }

+ // Scoped to tx, never the outer handle (AGENTS.md).

+ return tx.Model(&SiblingFeed{}).Where("id = ?", row.ID).Updates(fields).Error

+ })

+// deleteBySiblingItem applies one `gone` row.

+//

+// It resolves the sibling's item id the same way Ingest stores it — the

+// keyed fingerprint over (source, key) — because that is the only handle

+// memoria has on a row from the outside. Returns how many documents went,

+// which is 0 for a tombstone memoria has already applied or never held:

+// **not an error.** A feed can legitimately re-announce a deletion, and a

+// poll that failed mid-page re-applies the rows before the failure by

+// design.

+func (s *Store) deleteBySiblingItem(ctx context.Context, app, itemID string) (int, error) {

+ external, err := s.externalID(itemID)

+ if err != nil {

+ return 0, err

+ }

+ var docs []Document

+ if err := s.db.G.WithContext(ctx).

+ Where("source = ? AND external_id = ?", app, external).

+ Find(&docs).Error; err != nil {

+ return 0, err

+ }

+ for _, d := range docs {

+ // Store.Delete removes the index entries BEFORE the rows and

+ // refuses the whole delete if that fails, so a tombstone cannot

+ // leave passages indexed under a document that is gone. That

+ // ordering is AGENTS.md's rule and the reason this does not do

+ // its own deleting.

+ if err := s.Delete(ctx, d.ID); err != nil {

+ return 0, err

+ }

+ return len(docs), nil

diff --git a/internal/memoria/siblingsync_test.go b/internal/memoria/siblingsync_test.go

new file mode 100644

index 0000000..7cb5f17

--- /dev/null

+++ b/internal/memoria/siblingsync_test.go

@@ -0,0 +1,482 @@

+package memoria

+import (

+ "context"

+ "encoding/json"

+ "fmt"

+ "net/http"

+ "net/http/httptest"

+ "strings"

+ "testing"

+ "time"

+ "amadan.net/oficina/suite"

+ "amadan.net/oficina/memoria/internal/memoria/sibling"

+// feedServer is a sibling that serves a scripted sequence of S1 pages and

+// an S3 body per item.

+type feedServer struct {

+ t *testing.T

+ // pages is consulted by cursor: the empty string is the first page.

+ pages map[string]suite.Feed

+ bodies map[string]string

+ failS1 map[string]bool

+ failS3 map[string]bool

+ // brokeS3 is a 5xx rather than a 404: an ERROR, where failS3 is a

+ // refusal. The two must not be confused — a 404 is counted and moves

+ // on, a 500 stops the page — and no fixture distinguished them until

+ // a mutation showed the cursor could advance mid-page unnoticed.

+ brokeS3 map[string]bool

+ s1Calls int

+ s3Calls int

+func (f *feedServer) start(t *testing.T) (sibling.Sibling, *sibling.Client) {

+ t.Helper()

+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

+ switch {

+ case r.URL.Path == suite.RouteS1:

+ f.s1Calls++

+ cur := r.URL.Query().Get("cursor")

+ if f.failS1[cur] {

+ w.WriteHeader(http.StatusInternalServerError)

+ // A well-formed empty feed with a bad status: the shape

+ // that would authorise a purge if the status were ignored.

+ _, _ = w.Write([]byte(`{"cursor":"","rows":[]}`))

+ return

+ }

+ page, ok := f.pages[cur]

+ if !ok {

+ f.t.Errorf("sibling asked for an unscripted cursor %q", cur)

+ w.WriteHeader(http.StatusInternalServerError)

+ return

+ }

+ _ = json.NewEncoder(w).Encode(page)

+ case strings.HasPrefix(r.URL.Path, suite.RouteS3+"/"):

+ f.s3Calls++

+ id := strings.TrimPrefix(r.URL.Path, suite.RouteS3+"/")

+ if f.brokeS3[id] {

+ w.WriteHeader(http.StatusInternalServerError)

+ return

+ }

+ if f.failS3[id] {

+ w.WriteHeader(http.StatusNotFound)

+ return

+ }

+ body, ok := f.bodies[id]

+ if !ok {

+ w.WriteHeader(http.StatusNotFound)

+ return

+ }

+ _, _ = w.Write([]byte(body))

+ default:

+ f.t.Errorf("sibling asked for %s %s", r.Method, r.URL.Path)

+ w.WriteHeader(http.StatusNotFound)

+ }

+ }))

+ t.Cleanup(srv.Close)

+ s := sibling.Sibling{Name: "Docs", BaseURL: srv.URL, Token: "consumer-token"}

+ return s, sibling.NewForTest()

+func row(id string, rev int64, gone bool) suite.FeedRow {

+ return suite.FeedRow{ItemID: id, Revision: rev, Gone: gone, TouchedAt: time.Unix(0, 0).UTC()}

+// Gate 7 where it actually bites: **a failed poll advances nothing and

+// deletes nothing.**

+//

+// The clients already refuse to turn a failure into an empty feed. This is

+// the layer above, where the consequences live: a cursor moved past a page

+// nobody applied skips changes the sibling will never offer again, and it

+// does so silently, because the sibling has no way to know its consumer

+// stopped listening.

+//

+// TWO-SIDED, and the first subtest is what makes the rest mean anything: a

+// Poll that never stores would satisfy every "nothing was deleted"

+// assertion below.

+//

+// MUTATION-TESTED 2026-09-07, each verified red then reverted:

+//

+// setCursor called before the rows are applied

+// -> the mid-page-failure case red (the cursor moved)

+// the S1 error swallowed and the loop continued

+// -> the failed-poll case red

+// a `gone` row applied by deleting on a failed S3 instead

+// -> the missing case red (the document went)

+// Missing counted as Deleted

+// -> the missing case red

+// the empty final page treated as More

+// -> the completion case red

+func TestAFailedPollAdvancesNothing(t *testing.T) {

+ ctx := context.Background()

+ t.Run("a good poll stores the sibling's documents", func(t *testing.T) {

+ f := &feedServer{

+ t: t,

+ pages: map[string]suite.Feed{

+ // Revisions 7 and 9, not 1: the sibling issues these and

+ // memoria stores what it was given. With a fixture at

+ // revision 1, hard-coding `SiblingRevision: 1` passes —

+ // a mutation did.

+ "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 7, false), row("b", 9, false)}},

+ "c1": {Cursor: "c1", Rows: []suite.FeedRow{}},

+ },

+ bodies: map[string]string{

+ "a": "The quarterly planning process runs in March.",

+ "b": "Expenses are filed monthly.",

+ },

+ }

+ sib, client := f.start(t)

+ s := storeForTest(t)

+ res, err := s.Poll(ctx, client, sib)

+ if err != nil {

+ t.Fatalf("poll: %v", err)

+ }

+ if res.Stored != 2 || res.Deleted != 0 || res.Missing != 0 || res.More {

+ t.Errorf("result = %+v, want 2 stored and nothing else", res)

+ }

+ // Stored WITH their coordinates, or the ruling has nothing to ask

+ // about and every one of them withholds forever.

+ assertHeld(t, s, "Docs", "a", 7)

+ assertHeld(t, s, "Docs", "b", 9)

+ })

+ t.Run("a row failing mid-page leaves the cursor before that page", func(t *testing.T) {

+ // **This is the file's central claim, and nothing reached it.**

+ // The other failure cases fail at the S1 fetch, where no row has

+ // been applied yet and storing the cursor first is indisitinguishable

+ // from storing it last. A mutation that moved setCursor above the

+ // apply loop passed every one of them.

+ //

+ // Here the page is fetched fine and a row inside it fails. With

+ // the cursor stored first, page one is marked done while document

+ // "b" was never stored — and the sibling will not mention it

+ // again, because from its side the consumer read that page.

+ f := &feedServer{

+ t: t,

+ pages: map[string]suite.Feed{

+ "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false), row("b", 1, false)}},

+ "c1": {Cursor: "c1", Rows: []suite.FeedRow{}},

+ },

+ bodies: map[string]string{"a": "Applied.", "b": "Never fetched."},

+ brokeS3: map[string]bool{"b": true},

+ }

+ sib, client := f.start(t)

+ s := storeForTest(t)

+ if _, err := s.Poll(ctx, client, sib); err == nil {

+ t.Fatal("a 500 from S3 mid-page came back as a completed poll")

+ }

+ if got := cursorOf(t, s, "Docs"); got != "" {

+ t.Errorf("cursor = %q, want empty: a page with an unapplied row in "+

+ "it must not be marked read, or the sibling never offers it "+

+ "again and the document is lost silently", got)

+ }

+ // Row "a" was applied before the failure and stands — re-applying

+ // it next time is the cheap failure, and the point of the

+ // assertion is that the page will be re-offered at all.

+ assertHeld(t, s, "Docs", "a", 1)

+ if heldCount(t, s, "Docs") != 1 {

+ t.Errorf("%d documents held; the row that failed must not be "+

+ "stored", heldCount(t, s, "Docs"))

+ }

+ // Let it work, and the whole page arrives.

+ f.brokeS3 = nil

+ if _, err := s.Poll(ctx, client, sib); err != nil {

+ t.Fatalf("second poll: %v", err)

+ }

+ assertHeld(t, s, "Docs", "b", 1)

+ })

+ t.Run("a page carrying a row that names nothing is refused whole", func(t *testing.T) {

+ // A row with no item id can neither be stored nor deleted, and

+ // passing over it would advance the cursor past a change memoria

+ // never applied.

+ //

+ // **The refusal is the MODULE's**, and this test says so on

+ // purpose. memoria had its own check; it could not fail, because

+ // DecodeS1 rejects the page before applyRow sees a row — a

+ // mutation deleting memoria's guard left the suite green. So the

+ // guard is gone and this asserts the thing memoria actually

+ // depends on. It goes red if the module ever softens, which is

+ // the day memoria owes a check of its own.

+ f := &feedServer{

+ t: t,

+ pages: map[string]suite.Feed{

+ "": {Cursor: "c1", Rows: []suite.FeedRow{row("", 1, false)}},

+ // A clean second page, so SKIPPING the malformed row would

+ // produce a completely successful poll. Without it the

+ // poll fails anyway — on an unscripted cursor — and the

+ // assertion passes for the wrong reason.

+ "c1": {Cursor: "c1", Rows: []suite.FeedRow{}},

+ },

+ bodies: map[string]string{},

+ }

+ sib, client := f.start(t)

+ s := storeForTest(t)

+ if _, err := s.Poll(ctx, client, sib); err == nil {

+ t.Fatal("a feed row with no item id was passed over silently")

+ }

+ if got := cursorOf(t, s, "Docs"); got != "" {

+ t.Errorf("cursor = %q, want empty: a page memoria could not apply "+

+ "whole must not be marked read", got)

+ }

+ })

+ t.Run("a failed page leaves the cursor where it was", func(t *testing.T) {

+ f := &feedServer{

+ t: t,

+ pages: map[string]suite.Feed{

+ "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false)}},

+ "c1": {Cursor: "c2", Rows: []suite.FeedRow{row("b", 1, false)}},

+ },

+ bodies: map[string]string{"a": "First document.", "b": "Second document."},

+ failS1: map[string]bool{"c1": true},

+ }

+ sib, client := f.start(t)

+ s := storeForTest(t)

+ if _, err := s.Poll(ctx, client, sib); err == nil {

+ t.Fatal("a 500 from S1 came back as a completed poll")

+ }

+ // Page one was applied and stands.

+ assertHeld(t, s, "Docs", "a", 1)

+ // And the position is page one's, so page two is offered again

+ // rather than skipped forever.

+ if got := cursorOf(t, s, "Docs"); got != "c1" {

+ t.Errorf("cursor = %q, want c1: a poll that failed on the second "+

+ "page must leave the position at the first", got)

+ }

+ // Now let it succeed, and the page it could not read arrives.

+ f.failS1 = nil

+ f.pages["c2"] = suite.Feed{Cursor: "c2", Rows: []suite.FeedRow{}}

+ if _, err := s.Poll(ctx, client, sib); err != nil {

+ t.Fatalf("second poll: %v", err)

+ }

+ assertHeld(t, s, "Docs", "b", 1)

+ })

+ t.Run("a listed item the sibling will not serve is counted, not deleted", func(t *testing.T) {

+ f := &feedServer{

+ t: t,

+ pages: map[string]suite.Feed{

+ "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false)}},

+ "c1": {Cursor: "c1", Rows: []suite.FeedRow{}},

+ },

+ bodies: map[string]string{"a": "Held already."},

+ }

+ sib, client := f.start(t)

+ s := storeForTest(t)

+ if _, err := s.Poll(ctx, client, sib); err != nil {

+ t.Fatalf("first poll: %v", err)

+ }

+ assertHeld(t, s, "Docs", "a", 1)

+ // The sibling now lists it at a new revision and refuses to serve

+ // it. §4.2 makes that 404 cover both "gone" and "not

+ // team-visible", so memoria cannot tell which — and deleting on it

+ // would let a visibility change erase history the sibling still

+ // has.

+ f.pages[""] = suite.Feed{Cursor: "c2", Rows: []suite.FeedRow{row("a", 2, false)}}

+ f.pages["c2"] = suite.Feed{Cursor: "c2", Rows: []suite.FeedRow{}}

+ f.failS3 = map[string]bool{"a": true}

+ s2 := storeForTest(t)

+ if _, err := s2.Poll(ctx, client, sib); err != nil {

+ t.Fatalf("reset poll: %v", err)

+ }

+ f.pages["c1"] = suite.Feed{Cursor: "c2", Rows: []suite.FeedRow{row("a", 2, false)}}

+ res, err := s.Poll(ctx, client, sib)

+ if err != nil {

+ t.Fatalf("second poll: %v", err)

+ }

+ if res.Missing != 1 {

+ t.Errorf("result = %+v, want 1 missing", res)

+ }

+ if res.Deleted != 0 {

+ t.Errorf("a 404 on S3 deleted %d documents: an item that is not "+

+ "team-visible is indistinguishable from one that is gone, and "+

+ "acting on that erases history the sibling still has", res.Deleted)

+ }

+ // Still held, at the revision memoria could actually fetch.

+ assertHeld(t, s, "Docs", "a", 1)

+ })

+ t.Run("a gone row deletes, so the case above is not vacuous", func(t *testing.T) {

+ f := &feedServer{

+ t: t,

+ pages: map[string]suite.Feed{

+ "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false)}},

+ "c1": {Cursor: "c2", Rows: []suite.FeedRow{row("a", 2, true)}},

+ "c2": {Cursor: "c2", Rows: []suite.FeedRow{}},

+ },

+ bodies: map[string]string{"a": "Here for now."},

+ }

+ sib, client := f.start(t)

+ s := storeForTest(t)

+ if _, err := s.Poll(ctx, client, sib); err != nil {

+ t.Fatalf("poll: %v", err)

+ }

+ // One poll walks both pages: stored, then tombstoned.

+ if held := heldCount(t, s, "Docs"); held != 0 {

+ t.Errorf("%d documents survived a tombstone", held)

+ }

+ })

+// A sibling whose cursor never advances must not loop forever, and the

+// poll must say it did not finish.

+//

+// "Stopped looking" and "there is no more" are the same value to a caller

+// that only checks the error, which is why More exists.

+//

+// MUTATION-TESTED 2026-09-07: drop the maxFeedPages bound and the test

+// hangs rather than failing — which is itself the finding, so the bound is

+// asserted by the call count as well. Reverted.

+func TestAFeedThatNeverAdvancesStopsAndSaysSo(t *testing.T) {

+ f := &feedServer{

+ t: t,

+ // A page that hands back the cursor it was asked with. The module

+ // will not serve an empty cursor at all, so this — not a blank —

+ // is what "the feed did not advance" looks like on the wire.

+ pages: map[string]suite.Feed{

+ "": {Cursor: "same", Rows: []suite.FeedRow{row("a", 1, false)}},

+ "same": {Cursor: "same", Rows: []suite.FeedRow{row("a", 1, false)}},

+ },

+ bodies: map[string]string{"a": "Going nowhere."},

+ }

+ sib, client := f.start(t)

+ s := storeForTest(t)

+ res, err := s.Poll(context.Background(), client, sib)

+ if err != nil {

+ t.Fatalf("poll: %v", err)

+ }

+ if !res.More {

+ t.Error("a feed that never advanced its cursor reported a complete " +

+ "poll; the corpus would be called current")

+ }

+ // Two: the first page advanced the cursor, the second handed back the

+ // same one and the poll stopped. Bounded, not walked to maxFeedPages.

+ if f.s1Calls != 2 {

+ t.Errorf("%d S1 calls for a feed that stops advancing, want 2: a feed "+

+ "repeating itself must not be walked to the page bound", f.s1Calls)

+ }

+func assertHeld(t *testing.T, s *Store, app, itemID string, revision int64) {

+ t.Helper()

+ external, err := s.externalID(itemID)

+ if err != nil {

+ t.Fatalf("externalID: %v", err)

+ }

+ var d Document

+ if err := s.db.G.Where("source = ? AND external_id = ?", app, external).First(&d).Error; err != nil {

+ t.Fatalf("%s/%s not held: %v", app, itemID, err)

+ }

+ if d.SiblingRevision != revision {

+ t.Errorf("%s/%s held at revision %d, want %d", app, itemID, d.SiblingRevision, revision)

+ }

+ if len(d.SiblingItemID) == 0 {

+ t.Errorf("%s/%s stored without its coordinate, so nothing can be "+

+ "asked about it and it withholds forever", app, itemID)

+ }

+func heldCount(t *testing.T, s *Store, app string) int {

+ t.Helper()

+ var n int64

+ if err := s.db.G.Model(&Document{}).Where("source = ?", app).Count(&n).Error; err != nil {

+ t.Fatalf("count: %v", err)

+ }

+ return int(n)

+func cursorOf(t *testing.T, s *Store, app string) string {

+ t.Helper()

+ got, err := s.cursor(context.Background(), app)

+ if err != nil {

+ t.Fatalf("cursor: %v", err)

+ }

+ return got

+// The cursor is sealed at rest, and a cursor that will not open is an

+// error rather than an empty one.

+//

+// An empty cursor is not a degraded value: it is the instruction "start

+// from the beginning", which would re-announce the sibling's whole corpus.

+// Reading a corrupted row as that instruction is the plausible-empty-list

+// failure wearing its other face — a confident restart rather than a

+// confident absence.

+//

+// MUTATION-TESTED 2026-09-07: return ("", nil) when Open fails, and the

+// second half goes red. Reverted.

+func TestACursorIsSealedAndACorruptOneIsAnError(t *testing.T) {

+ ctx := context.Background()

+ s := storeForTest(t)

+ if err := s.setCursor(ctx, "Docs", "opaque-cursor-value"); err != nil {

+ t.Fatalf("setCursor: %v", err)

+ }

+ var stored SiblingFeed

+ if err := s.db.G.Where("app = ?", "Docs").First(&stored).Error; err != nil {

+ t.Fatalf("read back: %v", err)

+ }

+ // The presence assertion: it round-trips, so the absence check below

+ // is about sealing rather than about an empty column.

+ if got := cursorOf(t, s, "Docs"); got != "opaque-cursor-value" {

+ t.Fatalf("cursor round-tripped as %q", got)

+ }

+ if strings.Contains(string(stored.SealedCursor), "opaque-cursor-value") {

+ t.Error("the cursor is at rest in plaintext")

+ }

+ // Corrupt it. The row is present and unreadable, which must not read

+ // as "never polled".

+ if err := s.db.G.Model(&SiblingFeed{}).Where("id = ?", stored.ID).

+ Update("sealed_cursor", []byte("not a sealed value at all")).Error; err != nil {

+ t.Fatalf("corrupt: %v", err)

+ }

+ if got, err := s.cursor(ctx, "Docs"); err == nil {

+ t.Errorf("an unreadable cursor came back as %q with no error: that is "+

+ "the instruction to re-announce the sibling's entire corpus", got)

+ }

+// PolledAt says when the corpus was last known current, so a poll that

+// failed must not move it. §8.1: a failed or stale S1 makes the page state

+// what it is current as of, rather than implying freshness.

+//

+// MUTATION-TESTED 2026-09-07: set polled_at before walking the pages and

+// this goes red. Reverted.

+func TestAFailedPollDoesNotClaimFreshness(t *testing.T) {

+ f := &feedServer{

+ t: t,

+ pages: map[string]suite.Feed{"": {Cursor: "c1", Rows: []suite.FeedRow{}}},

+ failS1: map[string]bool{"": true},

+ }

+ sib, client := f.start(t)

+ s := storeForTest(t)

+ if _, err := s.Poll(context.Background(), client, sib); err == nil {

+ t.Fatal("a failing S1 reported a completed poll")

+ }

+ var row SiblingFeed

+ err := s.db.G.Where("app = ?", "Docs").First(&row).Error

+ if err == nil && !row.PolledAt.IsZero() {

+ t.Errorf("a poll that never read a page recorded PolledAt=%v; the "+

+ "sources page would report the corpus current as of a moment "+

+ "nothing was read", row.PolledAt)

+ }

+ if err != nil && !strings.Contains(fmt.Sprint(err), "record not found") {

+ t.Fatalf("read back: %v", err)

+ }

diff --git a/internal/memoriatest/gates_test.go b/internal/memoriatest/gates_test.go

index 0d64b0b..b10e73c 100644

--- a/internal/memoriatest/gates_test.go

+++ b/internal/memoriatest/gates_test.go

@@ -443,6 +443,11 @@ func TestProductionDoesNotUseTheTestConstructor(t *testing.T) {

// gate that greps for AppForTest alone. Found by a code review.

for _, seam := range []string{

"AppForTest", "AsPerson", "UseFakeSource", "WaitForSync",

+ // sibling.NewForTest builds a plain http.Client with no team

+ // binding. In production that is §6.4's cross-team refusal

+ // removed — the guard that runs before the transport dials —

+ // and it would compile and serve perfectly well.

+ "NewForTest",

} {

if strings.Contains(string(raw), seam) {

t.Errorf("%s calls the test seam %s; production must not reach it", p, seam)