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.
func TestTheResyncMaterializesThePhaseSplitAndNeverLowersACounter(t *testing.T) {
	s, sink, runID, _ := sinkFixture(t)
	if err := apply(t, sink, 2, ingest.TypeProgress, ingest.Progress{
		Draft: ingest.Counter{Done: 9, Total: 20}, Edit: ingest.Counter{Done: 7, Total: 20}}); err != nil {
		t.Fatal(err)
	}
	// A report taken BEFORE the engine's own figures caught up, folded over a projection the stream
	// has already moved: nothing may walk backwards.
	stale := ingest.StatusReport{TotalUnits: 20, Done: 3,
		Progress: ingest.Progress{Draft: ingest.Counter{Done: 5, Total: 20}, Edit: ingest.Counter{Done: 3, Total: 20}}}
	if err := s.ApplyStatus(t.Context(), runID, "bk1", stale, time.Now().UTC()); err != nil {
		t.Fatal(err)
	}
	var draft, edit int
	if err := s.pool.QueryRow(t.Context(),
		`select draft_done, edit_done from runs where id=$1`, runID).Scan(&draft, &edit); err != nil {
		t.Fatal(err)
	}
	if draft != 9 || edit != 7 {
		t.Errorf("a stale resync walked the projection back to draft=%d edit=%d, from draft=9 edit=7", draft, edit)
	}
	// A fresher one moves BOTH waves, which is the half no aggregate could do.
	fresh := ingest.StatusReport{TotalUnits: 20, Done: 11,
		Progress: ingest.Progress{Draft: ingest.Counter{Done: 20, Total: 20}, Edit: ingest.Counter{Done: 11, Total: 20}}}
	if err := s.ApplyStatus(t.Context(), runID, "bk1", fresh, time.Now().UTC()); err != nil {
		t.Fatal(err)
	}
	var dtotal, etotal int
	if err := s.pool.QueryRow(t.Context(),
		`select draft_done, draft_total, edit_done, edit_total from runs where id=$1`, runID).
		Scan(&draft, &dtotal, &edit, &etotal); err != nil {
		t.Fatal(err)
	}
	if draft != 20 || dtotal != 20 || edit != 11 || etotal != 20 {
		t.Errorf("after a fresher resync: draft %d/%d, edit %d/%d", draft, dtotal, edit, etotal)
	}
}

// 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)
	}
}