textmachine/platform/internal/pgstore/sink_test.go

package pgstore

import (
	"context"
	"encoding/json"
	"sync"
	"testing"
	"time"

	"github.com/jackc/pgx/v5"

	"textmachine/platform/internal/ingest"
	"textmachine/platform/internal/money"
)

// sinkFixture is one live run with a chapter to fold units into.
func sinkFixture(t *testing.T) (*Store, *RunSink, string, StartedRun) {
	t.Helper()
	s, ctx := testDB(t)
	now := fundedAccount(t, s, ctx, "u1", "10")
	seedBook(t, s, ctx, "bk1", "u1", 500)
	exec(t, s, ctx, `insert into chapters (id, book_id, number, units_total) values ('c1','bk1',1,50)`)
	run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", CeilingChapters: 10,
		Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
	if err != nil {
		t.Fatal(err)
	}
	sink := s.NewRunSink(run.AttemptID, run.ID, "bk1")
	if err := sink.Begin(t.Context(), ingest.Hello{EngineRunID: EngineStreamID(run.ID, 1)}); err != nil {
		t.Fatal(err)
	}
	return s, sink, run.ID, run
}

func apply(t *testing.T, s *RunSink, seq int64, typ ingest.Type, data any) error {
	t.Helper()
	body, err := json.Marshal(data)
	if err != nil {
		t.Fatal(err)
	}
	return s.Apply(t.Context(), ingest.Envelope{Seq: seq, Type: typ, Data: body}, ingest.Cursor{Offset: seq * 10})
}

// assertBookFirst proves an operation takes the BOOK's row lock BEFORE the row `probe` asks about,
// which is the package's global order (lockBook) and the thing a concurrency stress test can only
// sample.
//
// It holds the book lock, starts the operation, waits until the operation is actually blocked, and
// then asks whether the other row is still free. Free means the operation is waiting for the book
// with nothing else held — book-first. Taken means it grabbed the other row on the way, which is the
// order that deadlocks against every other transaction in this package.
func assertBookFirst(t *testing.T, s *Store, ctx context.Context, bookID string, probe func(context.Context, pgx.Tx) error, op func() error) {
	t.Helper()
	holder, err := s.pool.Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = holder.Rollback(ctx) }()
	var id string
	if err := holder.QueryRow(ctx, `select id from books where id = $1 for update`, bookID).Scan(&id); err != nil {
		t.Fatal(err)
	}
	done := make(chan error, 1)
	go func() { done <- op() }()

	// Wait for the operation to block on something. Without this the probe below could run before the
	// operation had taken any lock at all and pass for the wrong reason.
	deadline := time.Now().Add(20 * time.Second)
	for {
		// pg_stat_activity, not pg_locks: a transaction waiting for a ROW lock waits on the HOLDER's
		// transactionid, and those pg_locks rows carry a null database, so a database-scoped query over
		// pg_locks never sees the wait this test is built on.
		var blocked int
		if err := s.pool.QueryRow(ctx, `
			select count(*) from pg_stat_activity
			 where datname = current_database() and wait_event_type = 'Lock'`).Scan(&blocked); err != nil {
			t.Fatal(err)
		}
		if blocked > 0 {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("the operation never blocked on the book lock")
		}
		time.Sleep(10 * time.Millisecond)
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	err = probe(ctx, tx)
	_ = tx.Rollback(ctx)
	if err != nil {
		t.Errorf("the row was already locked while the book was not: %v — this order deadlocks", err)
	}
	if err := holder.Rollback(ctx); err != nil {
		t.Fatal(err)
	}
	if err := <-done; err != nil {
		t.Errorf("the operation itself failed: %v", err)
	}
}

// lockedAttempt and lockedReservation are the probes assertBookFirst asks with: NOWAIT turns "someone
// else holds this row" into an error instead of a wait.
func lockedAttempt(attemptID int64) func(context.Context, pgx.Tx) error {
	return func(ctx context.Context, tx pgx.Tx) error {
		var id int64
		return tx.QueryRow(ctx, `select id from run_attempts where id = $1 for update nowait`, attemptID).Scan(&id)
	}
}

func lockedReservation(key string) func(context.Context, pgx.Tx) error {
	return func(ctx context.Context, tx pgx.Tx) error {
		var id string
		return tx.QueryRow(ctx,
			`select engine_run_id from reservations where engine_run_id = $1 for update nowait`, key).Scan(&id)
	}
}

// Every path that touches a run's book and one of the rows below takes them in the package's order,
// asserted directly rather than sampled: a stress test can only fail to observe an inversion.
func TestTheMaterializerTakesTheBookBeforeTheAttempt(t *testing.T) {
	s, sink, _, run := sinkFixture(t)
	ctx := t.Context()
	body, err := json.Marshal(ingest.Progress{Draft: ingest.Counter{Done: 1, Total: 50}})
	if err != nil {
		t.Fatal(err)
	}
	assertBookFirst(t, s, ctx, "bk1", lockedAttempt(run.AttemptID), func() error {
		return sink.Apply(ctx, ingest.Envelope{Seq: 2, Type: ingest.TypeProgress, Data: body}, ingest.Cursor{Offset: 20})
	})
}

func TestARestartTakesTheBookBeforeTheAttempt(t *testing.T) {
	s, _, runID, run := sinkFixture(t)
	ctx := t.Context()
	assertBookFirst(t, s, ctx, "bk1", lockedAttempt(run.AttemptID), func() error {
		_, err := s.RestartRun(ctx, RestartInput{RunID: runID, AttemptID: run.AttemptID,
			UserID: "u1", BookID: "bk1", Ceiling: money.MicroUSD(1_000), Now: time.Now().UTC()})
		return err
	})
}

// Deleting a book takes the book before the reservations it clears. No cycle exists for this pair
// today — the delete only touches CLOSED reservations, and nothing else locks those — but an
// invariant with an undeclared exception stops being one, and the next writer of this package reads
// the invariant, not the exception.
func TestDeletingABookTakesItBeforeItsReservations(t *testing.T) {
	s, ctx := testDB(t)
	now := fundedAccount(t, s, ctx, "u1", "10")
	seedBook(t, s, ctx, "bk1", "u1", 500)
	if err := s.Hold(ctx, "u1", "bk1", "run_x#1", money.MicroUSD(300_000), now); err != nil {
		t.Fatal(err)
	}
	if err := s.Release(ctx, "run_x#1", now); err != nil {
		t.Fatal(err)
	}
	assertBookFirst(t, s, ctx, "bk1", lockedReservation("run_x#1"), func() error {
		return s.DeleteBook(ctx, "bk1")
	})
}

// The materializer and the reconciler work on one run at the same time as a matter of course — a
// live stream while a sweep ends an attempt — and they touch the same two rows. Taking them in
// opposite orders is a deadlock Postgres resolves by aborting a side, which costs a sweep or an API
// call and, worse, used to reach the reconciler as "this journal cannot be read" and quarantine the
// projection of a paying run for good.
//
// ⚠ Measured before the order was made global (lockBook): 258 of 300 concurrent pairs aborted.
func TestAMaterializerAndAReconcilerOnOneRunDoNotDeadlock(t *testing.T) {
	s, sink, runID, run := sinkFixture(t)
	const pairs = 60
	var wg sync.WaitGroup
	errs := make(chan error, 2*pairs)
	body, err := json.Marshal(ingest.Progress{Draft: ingest.Counter{Done: 1, Total: 50}})
	if err != nil {
		t.Fatal(err)
	}
	ctx := t.Context()
	now := time.Now().UTC()
	// The restart path takes the same two rows and used to take them attempt-first as well. Its own
	// LOGICAL outcome is not what this test is about — only the first of these closes the attempt and
	// the rest legitimately find nothing — so its non-transient errors are tolerated and only a broken
	// lock is not.
	restartErrs := make(chan error, pairs)
	for i := range pairs {
		wg.Add(3)
		go func() {
			defer wg.Done()
			seq := int64(i) + 2
			errs <- sink.Apply(ctx, ingest.Envelope{Seq: seq, Type: ingest.TypeProgress, Data: body},
				ingest.Cursor{Offset: seq * 10})
		}()
		go func() {
			defer wg.Done()
			_, err := s.FinishRun(ctx, RunEnding{RunID: runID, AttemptID: run.AttemptID,
				Status: "ready", ExitResult: "exit-code", Now: now})
			errs <- err
		}()
		go func() {
			defer wg.Done()
			_, err := s.RestartRun(ctx, RestartInput{RunID: runID, AttemptID: run.AttemptID,
				UserID: "u1", BookID: "bk1", Ceiling: money.MicroUSD(1_000), Now: now})
			restartErrs <- err
		}()
	}
	wg.Wait()
	close(errs)
	close(restartErrs)
	for err := range errs {
		if IsTransient(err) {
			t.Fatalf("the materializer and the reconciler deadlocked: %v", err)
		}
		if err != nil {
			t.Errorf("concurrent write: %v", err)
		}
	}
	for err := range restartErrs {
		if IsTransient(err) {
			t.Fatalf("the materializer and a restart deadlocked: %v", err)
		}
	}
}

// Each wave moves its OWN counter and no other. Which of the two then says a chapter is finished is
// a question asked at read time, of the pipeline the engine announced (pgstore.finishedUnits).
func TestADraftUnitMovesTheDraftCounterAndNotTheEditOne(t *testing.T) {
	s, sink, _, _ := sinkFixture(t)
	if err := apply(t, sink, 2, ingest.TypeUnitDone,
		ingest.UnitDone{Chapter: 1, Unit: 1, Wave: "draft", Shipped: true}); err != nil {
		t.Fatal(err)
	}
	var draft, edit int
	if err := s.pool.QueryRow(t.Context(),
		`select units_draft_done, units_edit_done from chapters where id='c1'`).
		Scan(&draft, &edit); err != nil {
		t.Fatal(err)
	}
	if draft != 1 || edit != 0 {
		t.Errorf("after a draft unit: draft=%d edit=%d", draft, edit)
	}
}

// The fold is an ASSIGNMENT, and this is the property that makes it one: the SAME unit announced
// again — by a re-read of the journal, or by a later process re-announcing what a crash lost before
// it reached the file — leaves the counters where they were.
//
// It is the consumer's duty and not the emitter's: delivery on this seam is at-least-once by
// ratification (D39.119 п.3), so `+ 1` per line was a counter that could only be right by luck. Two
// distinct units on the same chapter still count as two, which is the other half — an idempotent
// fold that also swallows real work is worse than the increment it replaces.
//
// Mutation caught: going back to `units_draft_done + 1`, and dropping the (chapter, unit, wave)
// primary key to (chapter, wave).
func TestAUnitAnnouncedTwiceIsCountedOnce(t *testing.T) {
	s, sink, _, _ := sinkFixture(t)
	counts := func() (int, int) {
		t.Helper()
		var draft, edit int
		if err := s.pool.QueryRow(t.Context(),
			`select units_draft_done, units_edit_done from chapters where id='c1'`).
			Scan(&draft, &edit); err != nil {
			t.Fatal(err)
		}
		return draft, edit
	}
	unit := ingest.UnitDone{Chapter: 1, Unit: 4, Wave: ingest.WaveDraft, Shipped: true}
	if err := apply(t, sink, 2, ingest.TypeUnitDone, unit); err != nil {
		t.Fatal(err)
	}
	// A NEW seq carrying a unit that was already announced. That is not a duplicate the cursor can
	// absorb — the high-water mark has moved past it — which is exactly the case the identity is for.
	if err := apply(t, sink, 3, ingest.TypeUnitDone, unit); err != nil {
		t.Fatal(err)
	}
	if draft, edit := counts(); draft != 1 || edit != 0 {
		t.Errorf("one unit announced twice: draft=%d edit=%d, want 1/0", draft, edit)
	}
	// A different unit of the same chapter is a different fact and does move the counter.
	unit.Unit = 9
	if err := apply(t, sink, 4, ingest.TypeUnitDone, unit); err != nil {
		t.Fatal(err)
	}
	// …and the same unit resolved by the OTHER wave is a different fact too.
	unit.Wave, unit.Flagged, unit.Reason = ingest.WaveEdit, true, "glossary_miss"
	if err := apply(t, sink, 5, ingest.TypeUnitDone, unit); err != nil {
		t.Fatal(err)
	}
	if draft, edit := counts(); draft != 2 || edit != 1 {
		t.Errorf("two draft units and one edit: draft=%d edit=%d, want 2/1", draft, edit)
	}
	// What the engine SAID about the unit is kept and not just the fact that it happened. The stream
	// delivers a disposition once and the cursor does not rewind, so a field dropped here is a field
	// lost — and shipped-and-flagged together is exactly the pair the contract derives unit state
	// from.
	var reason string
	var shipped, flagged bool
	if err := s.pool.QueryRow(t.Context(), `
		select reason, shipped, flagged from unit_resolutions
		 where book_id='bk1' and chapter=1 and unit=9 and wave='edit'`).Scan(&reason, &shipped, &flagged); err != nil {
		t.Fatal(err)
	}
	if reason != "glossary_miss" || !shipped || !flagged {
		t.Errorf("the unit was recorded as reason=%q shipped=%v flagged=%v", reason, shipped, flagged)
	}
}

// A wave the engine invents in a later minor is IGNORED, not fatal. The minor rule says an unknown
// value is ignored; letting it through would put it in front of a column constraint that refuses it
// and take the whole materialization of a paying run down with it.
func TestAWaveThisBuildDoesNotKnowIsIgnoredRatherThanFatal(t *testing.T) {
	s, sink, runID, _ := sinkFixture(t)
	if err := apply(t, sink, 2, ingest.TypeUnitDone,
		ingest.UnitDone{Chapter: 1, Unit: 1, Wave: "polish", Shipped: true}); err != nil {
		t.Fatalf("an unknown wave broke the materializer: %v", err)
	}
	var n int
	if err := s.pool.QueryRow(t.Context(),
		`select count(*) from unit_resolutions where book_id='bk1'`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	if n != 0 {
		t.Errorf("an unknown wave was stored (%d rows)", n)
	}
	// And the cursor still moved: an ignored event is applied, not skipped, or the stream stalls.
	var seq int64
	if err := s.pool.QueryRow(t.Context(),
		`select last_seq from run_attempts where engine_run_id = $1`, EngineStreamID(runID, 1)).Scan(&seq); err != nil {
		t.Fatal(err)
	}
	if seq != 2 {
		t.Errorf("last_seq = %d after an ignored event, want 2", seq)
	}
}

// WHICH ceiling stopped the run decides the reason, and the reason decides whether a resume can do
// anything about it. `day` is the engine's own limit out of a book.yaml the platform never wrote:
// calling it `credit_exhausted` would tell the user their credit ran out and would light the
// account-level halted flag, which keys on exactly that value.
//
// Mutation caught: mapping every scope onto credit_exhausted, and mapping an ABSENT scope onto
// daily_ceiling.
func TestTheCeilingScopeDecidesWhichPauseTheRunGets(t *testing.T) {
	for _, tc := range []struct {
		scope string
		want  string
	}{
		{ingest.ScopeBook, PausedCreditExhausted},
		{ingest.ScopeDay, PausedDailyCeiling},
		// Neither of the two the platform can place. Named as such rather than guessed at: the run is
		// resumable under any of the three reasons, so a guess buys nothing and `credit_exhausted`
		// would light the account's halted flag over somebody else's limit.
		{"", PausedCeilingUnknown},        // a stream written before the field existed
		{"account", PausedCeilingUnknown}, // a scope a later minor invents
	} {
		s, sink, runID, _ := sinkFixture(t)
		if err := apply(t, sink, 2, ingest.TypeCeiling,
			ingest.Ceiling{Halted: true, Scope: tc.scope}); err != nil {
			t.Fatal(err)
		}
		var status, reason string
		if err := s.pool.QueryRow(t.Context(),
			`select status, coalesce(paused_reason,'') from runs where id=$1`, runID).
			Scan(&status, &reason); err != nil {
			t.Fatal(err)
		}
		// `paused`, NEVER `failed`, for every scope: the stop is resumable and the contract forbids it
		// (§BookStatus, PD-113).
		if status != "paused" || reason != tc.want {
			t.Errorf("scope %q gave %q/%q, want paused/%s", tc.scope, status, reason, tc.want)
		}
	}
}

// A ceiling event that did NOT halt the run must not pause it.
func TestACeilingEventThatDidNotHaltChangesNothing(t *testing.T) {
	s, sink, runID, _ := sinkFixture(t)
	if err := apply(t, sink, 2, ingest.TypeCeiling, ingest.Ceiling{Halted: false}); err != nil {
		t.Fatal(err)
	}
	var status, reason string
	if err := s.pool.QueryRow(t.Context(),
		`select status, coalesce(paused_reason,'') from runs where id=$1`, runID).Scan(&status, &reason); err != nil {
		t.Fatal(err)
	}
	if status != "translating" || reason != "" {
		t.Errorf("a ceiling event with halted=false gave status %q reason %q", status, reason)
	}
}

// The bank stop arriving on the STREAM is the same product fact as exit code 3, and the exit path is
// pinned elsewhere; this is the other half.
func TestABankStopOnTheStreamPutsTheRunInAwaitingBank(t *testing.T) {
	s, sink, runID, _ := sinkFixture(t)
	if err := apply(t, sink, 2, ingest.TypeBankStop, ingest.BankStop{TermsProposed: 7}); err != nil {
		t.Fatal(err)
	}
	var status string
	if err := s.pool.QueryRow(t.Context(), `select status from runs where id=$1`, runID).Scan(&status); err != nil {
		t.Fatal(err)
	}
	if status != "awaiting_bank" {
		t.Errorf("after a bank stop the run is %q", status)
	}
}

// An event with no payload is refused rather than applied as a zero struct: a `progress` line with
// empty data would otherwise write draft 0/0 over a live projection.
func TestAnEventWithNoPayloadIsRefused(t *testing.T) {
	_, sink, _, _ := sinkFixture(t)
	err := sink.Apply(t.Context(), ingest.Envelope{Seq: 2, Type: ingest.TypeProgress}, ingest.Cursor{})
	if err == nil {
		t.Fatal("an event with no data was applied")
	}
}

// An ETA of zero is ABSENT, not zero: the screen renders without an estimate rather than showing
// "0 s left".
func TestAZeroEtaIsStoredAsAbsent(t *testing.T) {
	s, sink, runID, _ := sinkFixture(t)
	if err := apply(t, sink, 2, ingest.TypeProgress, ingest.Progress{
		Draft: ingest.Counter{Done: 1, Total: 10}, ETASeconds: 0}); err != nil {
		t.Fatal(err)
	}
	var eta *int
	if err := s.pool.QueryRow(t.Context(), `select eta_seconds from runs where id=$1`, runID).Scan(&eta); err != nil {
		t.Fatal(err)
	}
	if eta != nil {
		t.Errorf("an ETA of zero was stored as %d", *eta)
	}
}

// The resync materializes the engine's PER-WAVE split and never LOWERS a counter the stream has
// already moved further.
//
// ⚠ This replaces a pin whose premise the engine retired: while status carried only one end-to-end
// aggregate (backlog row 99) the resync could put it in the edit counter and nothing else, and the
// old test asserted that the draft counter stayed untouched. The split landed on 09.08 (D39.122), so
// asserting the old shape would now pin a limitation instead of a property.
// ⚠ RE-POINTED with PD-411 and the claim NARROWED to what is true, which is the finding rather than
// a concession. Its predecessor was called «the resync materializes the phase split and never lowers
// a counter» and it asserted over `runs.draft_done/draft_total/edit_done/edit_total` — four columns
// with two writers and NO reader, dropped in migration 00031. So the property it proved was real of
// those columns and meaningless of the screen: the bar is derived from `chapters`, which this channel
// does not write at all. There is nothing left for a resync to walk backwards, because there is
// nothing left for a resync to move.
//
// What that leaves is a GAP, not a passing test: the repair channel for a run whose stream is
// quarantined materialises no progress, so such a run's bar stands still however faithfully `tmctl
// status` answers. It has a register row of its own. This test pins the honest remainder — the ETA,
// the freshness stamp and the announced SHAPE — and the gap is stated here so the next reader does
// not re-derive it from a green test.
func TestTheResyncRecordsFreshnessAndShapeAndNoProgress(t *testing.T) {
	s, sink, runID, _ := sinkFixture(t)
	// ⚠ The stream announces a DRAFT-ONLY shape, and the resync below announces an EDITING one. The
	// two must differ or the shape assertion proves nothing: this pack's adversarial pass caught the
	// first edition, where the stream had already written the same shape and deleting the
	// recordWaveShape call from ApplyStatus entirely left the whole eighteen-package battery green.
	if err := apply(t, sink, 2, ingest.TypeProgress, ingest.Progress{
		Draft: ingest.Counter{Done: 9, Total: 20}, Edit: ingest.Counter{Done: 0, Total: 0},
		ETASeconds: 99}); err != nil {
		t.Fatal(err)
	}
	if editor, _ := bookShape(t, s, "bk1"); editor == nil || *editor {
		t.Fatalf("the stream announced a draft-only shape and the book reads %v: the fixture cannot"+
			" observe what the RESYNC writes", editor)
	}
	before := bookRevision(t, s, "bk1")
	rep := ingest.StatusReport{TotalUnits: 20, Done: 11,
		Progress: ingest.Progress{Draft: ingest.Counter{Done: 20, Total: 20},
			Edit: ingest.Counter{Done: 11, Total: 20}}, ETASeconds: 42}
	at := time.Now().UTC()
	if err := s.ApplyStatus(t.Context(), runID, "bk1", rep, at); err != nil {
		t.Fatal(err)
	}
	var eta *int
	var stamp *time.Time
	if err := s.pool.QueryRow(t.Context(),
		`select eta_seconds, last_resync_at from runs where id=$1`, runID).Scan(&eta, &stamp); err != nil {
		t.Fatal(err)
	}
	if eta == nil || *eta != 42 {
		t.Errorf("the resync's eta folded to %v, want 42", eta)
	}
	if stamp == nil {
		t.Error("the resync left no freshness stamp: how stale its figures are is the one thing only it can say")
	}
	// The SHAPE the report announced reaches the book — this is the resync's other real effect, and
	// the one that decides how the whole library counts this book. The stream said draft-only; only
	// ApplyStatus can have written what is read here.
	editor, epoch := bookShape(t, s, "bk1")
	if editor == nil || !*editor {
		t.Errorf("the report announced an edit total of 20 and the book's shape folded to %v: the"+
			" resync's own announcement never reached the book", editor)
	}
	if epoch != 1 {
		t.Errorf("the resync crossed a shape boundary and the epoch reads %d, want 1", epoch)
	}
	// And a client is told to re-read, because a resync is a materialisation like any other.
	if after := bookRevision(t, s, "bk1"); after <= before {
		t.Errorf("the resync left the book at revision %d: a watching client is never told to re-read", after)
	}
	// ⚠ THE GAP, asserted so it cannot be forgotten: the bar is derived from `chapters`, and the
	// resync does not write there. A quarantined run shows what its stream last delivered.
	var draftDone int
	if err := s.pool.QueryRow(t.Context(),
		`select units_draft_done from chapters where id='c1'`).Scan(&draftDone); err != nil {
		t.Fatal(err)
	}
	if draftDone != 0 {
		t.Errorf("the resync moved a chapter counter to %d: if this channel has learned to materialise"+
			" progress, the register row that says it cannot is stale and the bar's source has changed", draftDone)
	}
}

func bookShape(t *testing.T, s *Store, bookID string) (editor *bool, epoch int) {
	t.Helper()
	if err := s.pool.QueryRow(t.Context(),
		`select epoch_editor, shape_epoch from books where id = $1`, bookID).Scan(&editor, &epoch); err != nil {
		t.Fatal(err)
	}
	return editor, epoch
}

func bookRevision(t *testing.T, s *Store, bookID string) int64 {
	t.Helper()
	var rev int64
	if err := s.pool.QueryRow(t.Context(), `select revision from books where id = $1`, bookID).
		Scan(&rev); err != nil {
		t.Fatal(err)
	}
	return rev
}

// What a resync produces is stale by up to the poll interval, and the ONLY thing that says how stale
// is the stamp it leaves. A run being tailed has none: the stream carries its own freshness.
func TestAResyncRecordsWhenItWasTaken(t *testing.T) {
	s, _, runID, _ := sinkFixture(t)
	var stamp *time.Time
	if err := s.pool.QueryRow(t.Context(), `select last_resync_at from runs where id=$1`, runID).Scan(&stamp); err != nil {
		t.Fatal(err)
	}
	if stamp != nil {
		t.Fatalf("a run that was never resynced is stamped %s", stamp)
	}
	now := time.Now().UTC().Truncate(time.Millisecond)
	if err := s.ApplyStatus(t.Context(), runID, "bk1", ingest.StatusReport{TotalUnits: 20}, now); err != nil {
		t.Fatal(err)
	}
	if err := s.pool.QueryRow(t.Context(), `select last_resync_at from runs where id=$1`, runID).Scan(&stamp); err != nil {
		t.Fatal(err)
	}
	if stamp == nil || !stamp.UTC().Equal(now) {
		t.Fatalf("the resync stamped %v, want %s", stamp, now)
	}
	// And EVERY resync moves it. A stamp that only records the first one answers "when did polling
	// begin", which is not the question a reader of a stale projection is asking.
	later := now.Add(7 * time.Minute)
	if err := s.ApplyStatus(t.Context(), runID, "bk1", ingest.StatusReport{TotalUnits: 20}, later); err != nil {
		t.Fatal(err)
	}
	if err := s.pool.QueryRow(t.Context(), `select last_resync_at from runs where id=$1`, runID).Scan(&stamp); err != nil {
		t.Fatal(err)
	}
	if stamp == nil || !stamp.UTC().Equal(later) {
		t.Errorf("the second resync left the stamp at %v, want %s", stamp, later)
	}
}

// Cross-family review of the acceptance dofix (M2): a stop that lands before its engine was spawned
// leaves an attempt that is ENDED and carries no engine run id — a shape this zone could not produce
// before the stop path existed. An old materializer still holding that attempt would then bind it to
// the handshake of the attempt that REPLACED it and apply the same journal a second time, counting
// every unit and chapter twice. Binding is refused for an attempt that is over.
func TestAMaterializerOfAnEndedAttemptDoesNotAdoptTheNextAttemptsEngine(t *testing.T) {
	s, ctx := testDB(t)
	now := fundedAccount(t, s, ctx, "u1", "10")
	seedBook(t, s, ctx, "bk1", "u1", 500)
	run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", CeilingChapters: 10,
		Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
	if err != nil {
		t.Fatal(err)
	}
	// The stop-before-spawn shape: the attempt is closed, and its name is cleared to model an attempt
	// written by a build from BEFORE the platform named streams at creation — the only shape that can
	// still reach Begin's adoption path at all.
	exec(t, s, ctx, `update run_attempts set ended_at = now(), engine_run_id = null where id = $1`, run.AttemptID)
	stale := s.NewRunSink(run.AttemptID, run.ID, "bk1")
	if err := stale.Begin(ctx, ingest.Hello{EngineRunID: "eng-of-the-next-attempt"}); err == nil {
		t.Fatal("an ended attempt adopted an engine run: its journal would be materialized twice")
	}
	var bound *string
	if err := s.pool.QueryRow(ctx,
		`select engine_run_id from run_attempts where id = $1`, run.AttemptID).Scan(&bound); err != nil {
		t.Fatal(err)
	}
	if bound != nil {
		t.Fatalf("the ended attempt was bound to %q", *bound)
	}
}

// A resolution that arrives LATE must not drag a unit's disposition backwards.
//
// The fold assigns rather than increments, so without a comparison the upsert is "last to arrive
// wins" — and what a unit IS would then be a property of delivery order over an at-least-once seam
// (D39.119 п.3) rather than of the data. The engine itself does not emit such an inversion, and the
// acceptance was right to strike the claim that it does (see unitDone): every announcement, the
// re-announcement of a lost line included, carries the clock of the process making it. This pins the
// consumer's own guarantee, which is what a consumer of an at-least-once stream owes.
//
// The COUNTERS are unaffected either way — they count rows — which is exactly why this needs its own
// assertion: the ratified half of the fold cannot see it.
//
// Mutation caught: dropping `where excluded.at >= unit_resolutions.at` from unitDone.
func TestAReAnnouncedOlderResolutionDoesNotOverwriteANewerOne(t *testing.T) {
	s, sink, _, _ := sinkFixture(t)
	older := time.Date(2026, 8, 14, 10, 0, 0, 0, time.UTC)
	newer := older.Add(time.Hour)
	unit := ingest.UnitDone{Chapter: 1, Unit: 3, Wave: ingest.WaveEdit}

	apply := func(seq int64, at time.Time, u ingest.UnitDone) {
		t.Helper()
		body, err := json.Marshal(u)
		if err != nil {
			t.Fatal(err)
		}
		if err := sink.Apply(t.Context(),
			ingest.Envelope{Seq: seq, Type: ingest.TypeUnitDone, Time: at, Data: body},
			ingest.Cursor{Offset: seq * 10}); err != nil {
			t.Fatal(err)
		}
	}
	// The redrive's verdict: the unit ships.
	unit.Shipped, unit.Flagged, unit.Reason = true, false, ""
	apply(2, newer, unit)
	// …and then the older announcement of the same unit arrives, under a fresh seq.
	unit.Shipped, unit.Flagged, unit.Reason = false, true, "glossary_miss"
	apply(3, older, unit)

	var shipped, flagged bool
	var reason string
	if err := s.pool.QueryRow(t.Context(), `
		select shipped, flagged, reason from unit_resolutions
		 where book_id='bk1' and chapter=1 and unit=3 and wave='edit'`).Scan(&shipped, &flagged, &reason); err != nil {
		t.Fatal(err)
	}
	if !shipped || flagged || reason != "" {
		t.Errorf("the unit regressed to shipped=%v flagged=%v reason=%q on a re-announcement of an older event",
			shipped, flagged, reason)
	}
	// And a genuinely newer disposition still wins, or the guard would have frozen the first answer.
	unit.Shipped, unit.Flagged, unit.Reason = false, true, "sanitizer_stripped"
	apply(4, newer.Add(time.Minute), unit)
	if err := s.pool.QueryRow(t.Context(), `
		select shipped, flagged, reason from unit_resolutions
		 where book_id='bk1' and chapter=1 and unit=3 and wave='edit'`).Scan(&shipped, &flagged, &reason); err != nil {
		t.Fatal(err)
	}
	if shipped || !flagged || reason != "sanitizer_stripped" {
		t.Errorf("a newer disposition did not win: shipped=%v flagged=%v reason=%q", shipped, flagged, reason)
	}
}

// WHICH cut of the source the engine is working from reaches the book from the handshake.
//
// ⚠ It used to be recorded in `Begin`, and `Begin` is unreachable for any attempt this build creates:
// the platform names an attempt's stream when the row is written, so the tailer is never without a
// name and never adopts a handshake. Moving the effect to where the handshake arrives as an ORDINARY
// event is what kept the fact from being silently lost — a reader joining a persisted chapter tree
// cut by a different chunker would be joining on ordinals that were re-numbered.
//
// Mutation caught: removing the TypeHello case from effect.
func TestTheChunkerVersionOfTheStreamReachesTheBook(t *testing.T) {
	s, sink, _, _ := sinkFixture(t)
	if err := apply(t, sink, 2, ingest.TypeHello,
		ingest.Hello{StreamVersion: "1.1", EngineRunID: "x", BookID: "bk1", ChunkerVersion: "chunk-2026.08"}); err != nil {
		t.Fatal(err)
	}
	var got string
	if err := s.pool.QueryRow(t.Context(),
		`select coalesce(chunker_version,'') from books where id='bk1'`).Scan(&got); err != nil {
		t.Fatal(err)
	}
	if got != "chunk-2026.08" {
		t.Errorf("chunker_version = %q after the handshake", got)
	}
	// A handshake without one leaves the recorded value alone rather than blanking it.
	if err := apply(t, sink, 3, ingest.TypeHello,
		ingest.Hello{StreamVersion: "1.1", EngineRunID: "x", BookID: "bk1"}); err != nil {
		t.Fatal(err)
	}
	if err := s.pool.QueryRow(t.Context(),
		`select coalesce(chunker_version,'') from books where id='bk1'`).Scan(&got); err != nil {
		t.Fatal(err)
	}
	if got != "chunk-2026.08" {
		t.Errorf("a handshake with no chunker version overwrote the recorded one: %q", got)
	}
}

// A flagged unit moves the chapter's NOTE counter as well as its wave counter, in the same statement
// and from the same rows. The card sums those counters, so a sink that stopped maintaining this one
// would leave every book on the library page reading zero notes while `GET /notes` listed them.
//
// Mutation caught: dropping note_count from the fold.
func TestAFlaggedUnitMovesTheChaptersNoteCounter(t *testing.T) {
	s, sink, _, _ := sinkFixture(t)
	count := func() int {
		t.Helper()
		var n int
		if err := s.pool.QueryRow(t.Context(),
			`select note_count from chapters where id='c1'`).Scan(&n); err != nil {
			t.Fatal(err)
		}
		return n
	}
	if got := count(); got != 0 {
		t.Fatalf("the chapter starts at %d notes: this fixture cannot observe the move", got)
	}
	if err := apply(t, sink, 2, ingest.TypeUnitDone,
		ingest.UnitDone{Chapter: 1, Unit: 0, Wave: ingest.WaveEdit, Shipped: true, Flagged: true,
			Reason: "glossary_miss"}); err != nil {
		t.Fatal(err)
	}
	if got := count(); got != 1 {
		t.Errorf("after a flagged unit the chapter counts %d notes", got)
	}
	// An unflagged one does not, which is what makes the counter a NOTE counter.
	if err := apply(t, sink, 3, ingest.TypeUnitDone,
		ingest.UnitDone{Chapter: 1, Unit: 4, Wave: ingest.WaveEdit, Shipped: true}); err != nil {
		t.Fatal(err)
	}
	if got := count(); got != 1 {
		t.Errorf("a clean unit moved the note counter to %d", got)
	}
}

// The progress FRAME and the run's card answer ONE bar — same done, same total, same stage — and
// the pin rides the flipped-flag fixture on purpose: the frame is minted inside the transaction
// that flips `edit_wave` (recordWaveShape), which is exactly where a frame computed from different
// fragments than the card's would diverge first. This is the wire half of the through-bar (row
// 200): `stage` had no value pin at all, and a planted done/total swap in the payload outlived
// both batteries before it.
//
// Mutation caught: swapping done/total (or dropping stage) in emitProgress's payload; computing
// the frame from fragments other than the card's runDone/runTotal/runStage.
func TestTheProgressFrameMatchesTheCardOnAFlippedFlag(t *testing.T) {
	s, ctx := testDB(t)
	now := fundedAccount(t, s, ctx, "u1", "10")
	seedBook(t, s, ctx, "bk1", "u1", 500)
	if err := s.SaveStructure(ctx, "bk1", twoChapters("k1")); err != nil {
		t.Fatal(err)
	}
	// The book's draft era: drafted end to end, the flag SAYS no editor, one chapter already edited
	// by the time the engine speaks — so the bar is non-trivial in the very frame that flips.
	exec(t, s, ctx, `update books set edit_wave = false where id = 'bk1'`)
	exec(t, s, ctx, `update chapters set units_draft_done = units_total where book_id = 'bk1'`)
	run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", CeilingChapters: 2,
		Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
	if err != nil {
		t.Fatal(err)
	}
	exec(t, s, ctx, `update chapters set units_edit_done = units_total
	                  where book_id = 'bk1' and number = 1`)
	sink := s.NewRunSink(run.AttemptID, run.ID, "bk1")
	if err := sink.Begin(ctx, ingest.Hello{EngineRunID: EngineStreamID(run.ID, 1)}); err != nil {
		t.Fatal(err)
	}
	// The engine's first progress event announces the editor: the flag flips INSIDE this apply, and
	// the frame is minted right after it.
	if err := apply(t, sink, 2, ingest.TypeProgress, ingest.Progress{
		Draft: ingest.Counter{Done: 3, Total: 3}, Edit: ingest.Counter{Done: 2, Total: 3},
	}); err != nil {
		t.Fatal(err)
	}
	card, err := s.ReadRun(ctx, "u1", run.ID)
	if err != nil {
		t.Fatal(err)
	}
	if card.Progress.Done != 1 || card.Progress.Total != 2 || card.Progress.Stage != "editing" {
		t.Fatalf("the fixture's card reads %d/%d %q: it cannot tell a swap from the truth",
			card.Progress.Done, card.Progress.Total, card.Progress.Stage)
	}
	frames, err := s.ReadFrames(ctx, "bk1", 0, 100)
	if err != nil {
		t.Fatal(err)
	}
	var last json.RawMessage
	for _, f := range frames {
		if f.Event == FrameProgress {
			last = f.Data
		}
	}
	if last == nil {
		t.Fatal("the progress event minted no frame")
	}
	var frame struct {
		Progress struct {
			Done       int    `json:"done"`
			Total      int    `json:"total"`
			Stage      string `json:"stage"`
			ETASeconds *int   `json:"eta_seconds"`
		} `json:"progress"`
	}
	if err := json.Unmarshal(last, &frame); err != nil {
		t.Fatalf("the frame's payload: %v (%s)", err, last)
	}
	if frame.Progress.Done != card.Progress.Done || frame.Progress.Total != card.Progress.Total ||
		frame.Progress.Stage != card.Progress.Stage {
		t.Errorf("the frame reads %d/%d %q and the card %d/%d %q — one bar, two answers",
			frame.Progress.Done, frame.Progress.Total, frame.Progress.Stage,
			card.Progress.Done, card.Progress.Total, card.Progress.Stage)
	}
	if (frame.Progress.ETASeconds == nil) != (card.Progress.ETASeconds == nil) {
		t.Errorf("eta: frame %v, card %v", frame.Progress.ETASeconds, card.Progress.ETASeconds)
	}
}