textmachine/platform/internal/pgstore/sink.go

package pgstore

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"

	"textmachine/platform/internal/ingest"
)

// RunSink materializes one attempt's event stream into the reporting database.
//
// It is bound to an ATTEMPT rather than to a run, because the ratified idempotency key is
// (engine_run_id, seq) and the engine mints a new run id — with seq restarting at 1 — on every
// invocation. Keying on the platform's run would drop the whole stream of attempt two.
type RunSink struct {
	store     *Store
	attemptID int64
	runID     string
	bookID    string
}

// NewRunSink builds the sink for one attempt.
func (s *Store) NewRunSink(attemptID int64, runID, bookID string) *RunSink {
	return &RunSink{store: s, attemptID: attemptID, runID: runID, bookID: bookID}
}

// Begin binds the engine's run id to this attempt.
//
// ⚠ It is a LEGACY path now: since the platform names an attempt's stream when the attempt row is
// created (pgstore.EngineStreamID), the tailer is always given the id it wants and never adopts a
// handshake, so this is reached only for attempts written by an older build. What it used to do
// besides the binding — recording the chunker version — moved into `effect`, where the handshake
// arrives as an ordinary event.
//
// The binding is refused if the attempt already carries a DIFFERENT id: that means two engine
// processes wrote into one journal under one attempt, and materializing either of them would mix
// two runs' counters into one projection.
//
// It is refused for an ENDED attempt too, and that half is newer than the rule above. A stop that
// landed before its engine was spawned leaves an attempt that is closed and carries no engine run id
// — a shape this zone could not produce before P5 — and an unbound, ended attempt would otherwise
// ADOPT the handshake of the attempt that replaced it: the same events would then be materialized
// twice, once per attempt, and every counter they increment would be counted twice. Found by
// cross-family review of the acceptance dofix.
func (r *RunSink) Begin(ctx context.Context, h ingest.Hello) error {
	tag, err := r.store.pool.Exec(ctx, `
		update run_attempts set engine_run_id = $2
		 where id = $1 and ended_at is null
		   and (engine_run_id is null or engine_run_id = $2)`, r.attemptID, h.EngineRunID)
	if err != nil {
		return fmt.Errorf("pgstore: bind engine run: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return fmt.Errorf("pgstore: attempt %d is over, or already bound to another engine run", r.attemptID)
	}
	if h.ChunkerVersion != "" {
		if _, err := r.store.pool.Exec(ctx,
			`update books set chunker_version = $2 where id = $1`, r.bookID, h.ChunkerVersion); err != nil {
			return fmt.Errorf("pgstore: record chunker version: %w", err)
		}
	}
	return nil
}

// Apply materializes one event AND the cursor it moves, in ONE transaction.
//
// That is the whole reason this method exists rather than two: an implementation that applies the
// effect and then records the position has a window in which a crash re-applies the effect, and an
// implementation that records first has a window in which the effect is lost. The high-water mark is
// re-checked inside the transaction, so a duplicate line that raced a live writer is dropped here
// too and not only in the reader (PD-105: at-least-once delivery, a duplicate is not an error).
//
// The BOOK is locked first — see lockBook. This path used to take the attempt first and every other
// path that touches both takes the book first, which is a deadlock the moment a materializer and a
// reconciler work on one run: measured on the real API at 258 of 300 concurrent pairs.
func (r *RunSink) Apply(ctx context.Context, ev ingest.Envelope, c ingest.Cursor) error {
	return r.store.inTx(ctx, func(tx pgx.Tx) error {
		if err := lockBook(ctx, tx, r.bookID); err != nil {
			return err
		}
		var last int64
		if err := tx.QueryRow(ctx,
			`select last_seq from run_attempts where id = $1 for update`, r.attemptID).Scan(&last); err != nil {
			return fmt.Errorf("pgstore: lock attempt: %w", err)
		}
		if ev.Seq <= last {
			return nil // already applied; the cursor cannot move backwards either
		}
		if err := r.effect(ctx, tx, ev); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `
			update run_attempts
			   set last_seq = $2, last_offset = greatest(last_offset, $3), last_line_sha256 = $4
			 where id = $1`, r.attemptID, ev.Seq, c.Offset, c.SHA256); err != nil {
			return fmt.Errorf("pgstore: move cursor: %w", err)
		}
		return nil
	})
}

// effect is what one event changes. An unknown type changes nothing and is not an error: tolerating
// it is the minor-version rule of the stream (D39.85).
func (r *RunSink) effect(ctx context.Context, tx pgx.Tx, ev ingest.Envelope) error {
	switch ev.Type {
	case ingest.TypeHello:
		// The handshake reaches here as an ordinary event — it is seq 1 and it moves the cursor — and
		// what it changes is one fact: WHICH cut of the source the engine is working from. A reader
		// that persisted a chapter manifest cut by a different chunker would be joining on ordinals
		// that were silently re-numbered.
		//
		// ⚠ It used to be recorded in Begin, and Begin is now unreachable for any attempt this build
		// created: the platform names the stream when the attempt row is written, so the tailer never
		// has to adopt a handshake. Moving the effect here is what kept it from being silently lost —
		// found by the adversarial review of this pack.
		var h ingest.Hello
		if err := decode(ev, &h); err != nil {
			return err
		}
		if h.ChunkerVersion == "" {
			return nil
		}
		if _, err := tx.Exec(ctx,
			`update books set chunker_version = $2 where id = $1`, r.bookID, h.ChunkerVersion); err != nil {
			return fmt.Errorf("pgstore: record chunker version: %w", err)
		}
		return nil
	case ingest.TypeProgress:
		var p ingest.Progress
		if err := decode(ev, &p); err != nil {
			return err
		}
		return r.bump(ctx, tx, `
			update runs set draft_done = $2, draft_total = $3, edit_done = $4, edit_total = $5,
			                eta_seconds = $6, revision = $7 where id = $1`,
			r.runID, p.Draft.Done, p.Draft.Total, p.Edit.Done, p.Edit.Total, etaOrNil(p.ETASeconds))
	case ingest.TypeUnitDone:
		var u ingest.UnitDone
		if err := decode(ev, &u); err != nil {
			return err
		}
		return r.unitDone(ctx, tx, ev, u)
	case ingest.TypeBankStop:
		return r.bump(ctx, tx, `update runs set status = 'awaiting_bank', revision = $2 where id = $1`, r.runID)
	case ingest.TypeCeiling:
		var c ingest.Ceiling
		if err := decode(ev, &c); err != nil {
			return err
		}
		if !c.Halted {
			return nil
		}
		// A ceiling stop is `paused`, never `failed`: it is resumable, and mapping it to a failure
		// would lie about that (contract §BookStatus). WHICH ceiling decides the reason — and, one
		// step later, whether a resume can do anything about it (see CeilingPause).
		return r.bump(ctx, tx, `
			update runs set status = 'paused', paused_reason = $2, revision = $3
			 where id = $1`, r.runID, CeilingPause(c.Scope))
	case ingest.TypeSpend:
		var s ingest.Spend
		if err := decode(ev, &s); err != nil {
			return err
		}
		// The MAXIMUM seen, not a sum: the counter is cumulative, so a redelivered line is harmless
		// only as long as nothing adds it up. Freshness only — no balance moves here.
		_, err := tx.Exec(ctx, `
			update run_attempts set spend_micro_usd = greatest(spend_micro_usd, $2) where id = $1`,
			r.attemptID, s.CommittedMicroUSD)
		if err != nil {
			return fmt.Errorf("pgstore: record spend: %w", err)
		}
		return nil
	case ingest.TypeFinished:
		// The stream says the engine believes it is done. The RUN is not closed here: closing it
		// settles money, and money is settled from what the process actually did — which is known
		// once the unit is gone, not once a line was written.
		return nil
	default:
		return nil
	}
}

// unitDone records one resolved output unit and re-derives the counters of its chapter.
//
// ⚠ The fold is an ASSIGNMENT and not an increment, and that is ratified rather than stylistic
// (D39.131 п.2г). Delivery on this seam is at-least-once, so absorbing a duplicate is the
// CONSUMER's duty: `+ 1` per line counted a re-read line twice, and re-reading is the ordinary
// consequence of a cursor that was written before its effect — or of the emitter's own named
// residual, a line committed to the outbox that never reached the file and is therefore announced
// again by the next process. Recording the (chapter, unit, wave) identity the event carries and
// counting the set makes the answer the same however many times the line arrives, and it self-heals
// the undrained tail rather than merely tolerating it.
//
// Unknown waves are dropped rather than stored: `wave` is a closed vocabulary on both sides of the
// seam, and a value from outside it would land in a column whose constraint refuses it and take the
// whole materialization down with it. Under the minor-version rule an unknown value is ignored.
//
// ⚠ Bounded by what exists, unchanged: without the engine's persisted chapter manifest (unified
// backlog row 100) there is no `chapters` row to carry the derived counters, so the fold records the
// unit and stops there. The counters the screen reads today come from the progress event.
func (r *RunSink) unitDone(ctx context.Context, tx pgx.Tx, ev ingest.Envelope, u ingest.UnitDone) error {
	if u.Wave != ingest.WaveDraft && u.Wave != ingest.WaveEdit {
		return nil
	}
	// The LATEST resolution wins, and "latest" is by the event's own timestamp rather than by arrival.
	// A unit can legitimately be resolved twice with different dispositions — a redrive re-attacks a
	// flagged one — and the read model wants the second.
	//
	// ⚠ The `where` guards against ARRIVAL ORDER, and it is worth being exact about what it does not
	// do, because the acceptance found this comment claiming the opposite: the engine does not emit
	// an inversion. Every announcement carries the clock of the process making it (pipeline/events.go,
	// `at := e.now()`), the re-announcement of a line a dead process never got onto the file included
	// — such a row carries no key, ForgetEvents drops it, and the next process announces it afresh
	// with its own stamp (store/outbox.go). So this is not the repair of a known regression.
	//
	// What it buys is that the fold's result stops depending on WHO WROTE LAST. The seam is
	// at-least-once (D39.119 п.3) and this consumer assigns rather than increments, so a disposition
	// is otherwise decided by delivery order — which is a property of the tailer, not of the data. One
	// predicate makes it a property of the data, and a redrive that moved a unit from flagged to
	// shipped cannot be walked backwards by a re-read.
	if _, err := tx.Exec(ctx, `
		insert into unit_resolutions (book_id, chapter, unit, wave, shipped, flagged, reason, at)
		values ($1, $2, $3, $4, $5, $6, $7, $8)
		on conflict (book_id, chapter, unit, wave)
		do update set shipped = excluded.shipped, flagged = excluded.flagged,
		              reason = excluded.reason, at = excluded.at
		 where excluded.at >= unit_resolutions.at`,
		r.bookID, u.Chapter, u.Unit, u.Wave, u.Shipped, u.Flagged, u.Reason, ev.Time); err != nil {
		return fmt.Errorf("pgstore: record unit: %w", err)
	}
	// units_done stays what it has always been — the count of the EDIT wave, the aggregate the
	// resync path can restore (00002, K-10 is open). Derived from the same set as the two phase
	// counters so the three can never disagree.
	tag, err := tx.Exec(ctx, `
		update chapters c
		   set units_draft_done = (select count(*) from unit_resolutions
		                            where book_id = c.book_id and chapter = c.number and wave = 'draft'),
		       units_edit_done  = (select count(*) from unit_resolutions
		                            where book_id = c.book_id and chapter = c.number and wave = 'edit'),
		       units_done       = (select count(*) from unit_resolutions
		                            where book_id = c.book_id and chapter = c.number and wave = 'edit'),
		       revision = (select revision + 1 from books where id = $1)
		 where c.book_id = $1 and c.number = $2`, r.bookID, u.Chapter)
	if err != nil {
		return fmt.Errorf("pgstore: fold unit: %w", err)
	}
	if tag.RowsAffected() == 0 {
		// No manifest yet — see the note above. The unit itself IS recorded, so the counters come out
		// exact the moment a chapter row exists; the book's revision is deliberately NOT bumped,
		// because nothing a client can read has changed and every bump costs one refetch.
		return nil
	}
	return r.bumpBook(ctx, tx)
}

// bump runs a statement whose LAST argument is the book's next revision, and stamps the book with
// it. One transaction is one revision but possibly several rows, which is why catch-up reads use
// `>=` and not `>` (contract §Revision).
func (r *RunSink) bump(ctx context.Context, tx pgx.Tx, q string, args ...any) error {
	rev, err := r.nextRevision(ctx, tx)
	if err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, q, append(args, rev)...); err != nil {
		return fmt.Errorf("pgstore: materialize event: %w", err)
	}
	return r.bumpBook(ctx, tx)
}

func (r *RunSink) nextRevision(ctx context.Context, tx pgx.Tx) (int64, error) {
	var rev int64
	if err := tx.QueryRow(ctx, `select revision + 1 from books where id = $1 for update`, r.bookID).Scan(&rev); err != nil {
		return 0, fmt.Errorf("pgstore: read revision: %w", err)
	}
	return rev, nil
}

func (r *RunSink) bumpBook(ctx context.Context, tx pgx.Tx) error {
	if _, err := tx.Exec(ctx, `update books set revision = revision + 1 where id = $1`, r.bookID); err != nil {
		return fmt.Errorf("pgstore: bump book revision: %w", err)
	}
	return nil
}

func decode(ev ingest.Envelope, into any) error {
	if len(ev.Data) == 0 {
		return fmt.Errorf("pgstore: event %s seq %d carries no data", ev.Type, ev.Seq)
	}
	if err := json.Unmarshal(ev.Data, into); err != nil {
		return fmt.Errorf("pgstore: event %s seq %d: %w", ev.Type, ev.Seq, err)
	}
	return nil
}

func etaOrNil(s int) *int {
	if s <= 0 {
		return nil // absent, not zero: the screen renders without it rather than showing "0 s left"
	}
	return &s
}

// ApplyStatus folds a `tmctl status --json` report into the read model.
//
// This is the RESYNC channel: a snapshot of a run that keeps moving, taken every few minutes. Since
// D39.122 the report carries the per-wave split, so it materializes through the same four counters
// as the stream and no longer flattens them — what stays true of it is that it is STALE by up to
// the poll interval, which is why it stamps last_resync_at rather than leaving a reader to guess
// how old the figures on a quarantined run are.
func (s *Store) ApplyStatus(ctx context.Context, runID, bookID string, rep ingest.StatusReport, now time.Time) error {
	return s.inTx(ctx, func(tx pgx.Tx) error {
		var rev int64
		if err := tx.QueryRow(ctx, `select revision + 1 from books where id = $1 for update`, bookID).Scan(&rev); err != nil {
			return fmt.Errorf("pgstore: read revision: %w", err)
		}
		// greatest(), so a resync can never move a counter BACKWARDS. A report taken before the
		// engine's own figures caught up would otherwise walk a visible progress bar back down — the
		// one thing the contract asks a client never to do and which the server must not do either.
		// It is also what makes this safe to run over a projection the stream has already moved.
		if _, err := tx.Exec(ctx, `
			update runs set draft_done = greatest(draft_done, $2), draft_total = greatest(draft_total, $3),
			                edit_done = greatest(edit_done, $4), edit_total = greatest(edit_total, $5),
			                eta_seconds = $6, last_resync_at = $7, revision = $8
			 where id = $1`, runID, rep.Progress.Draft.Done, rep.Progress.Draft.Total,
			rep.Progress.Edit.Done, rep.Progress.Edit.Total,
			etaOrNil(int(rep.ETASeconds)), now, rev); err != nil {
			return fmt.Errorf("pgstore: apply status: %w", err)
		}
		if _, err := tx.Exec(ctx,
			`update books set revision = revision + 1 where id = $1`, bookID); err != nil {
			return fmt.Errorf("pgstore: bump book revision: %w", err)
		}
		return nil
	})
}

// RunEnding is one attempt's end, as the reconciler decided it.
type RunEnding struct {
	RunID     string
	AttemptID int64
	// Status is the run's product status. PausedReason accompanies `paused` and must be empty for
	// every other status: a run that is not paused carrying a reason it is paused would be read by
	// the resume path, which asks that column what to do next.
	Status       string
	PausedReason string
	// ExitResult is systemd's $SERVICE_RESULT and ExitCode the engine's own code, when it exited.
	ExitResult string
	ExitCode   *int
	Now        time.Time
}

// FinishRun closes a run and its attempt: the read-model status, the end of the attempt and what
// systemd said about it. Money is NOT touched here — see Settlement.
//
// ⚠ The paused reason is written HERE and not only by PauseRun, and that is the ceiling half of
// PD-113. A ceiling halt reaches this platform twice — as a stream event and as exit code 4 — and
// the two fail independently: a journal that could not be written still leaves the code. On that
// path there is no event to have set the reason, and a `paused` run with a null reason gives the
// screen nothing to say and the resume path nothing to judge.
//
// closed is false when the write did not apply, and the caller must then do nothing else: the run was
// already finished, or — the case that made this a bool — the attempt it was asked to close is no
// longer the run's live one.
//
// ⚠ That second guard is money. A sweep decides from a snapshot and writes seconds later; between the
// two, the user can stop the run and RESUME it, and the resumed run is live again with a second
// attempt holding a second reservation. The old attempt's exit marker is still on disk — nothing
// deletes markers of attempts that ended — so the stale pass would close the run from it, leaving
// attempt 2's hold in NO worklist: `ListLiveRuns` selects runs with `finished_at is null` and
// `UnsettledRuns` attempts with `ended_at is not null`, and the resumed run matches neither once it
// has been re-finished. Reachable only since a run can come back to life at all, which is this pack.
func (s *Store) FinishRun(ctx context.Context, in RunEnding) (closed bool, err error) {
	if !validRunStatus(in.Status) {
		return false, fmt.Errorf("pgstore: %q is not a run status", in.Status)
	}
	switch {
	case in.PausedReason != "" && in.Status != "paused":
		return false, fmt.Errorf("pgstore: a %s run cannot carry a paused reason", in.Status)
	case in.PausedReason != "" && !validPauseReason(in.PausedReason):
		return false, fmt.Errorf("pgstore: %q is not a pause reason", in.PausedReason)
	}
	runID, attemptID, now := in.RunID, in.AttemptID, in.Now
	err = s.inTx(ctx, func(tx pgx.Tx) error {
		// The BOOK's row is locked first, here and in the materializer. The two used to take them in
		// opposite orders — the materializer books-then-runs, this one runs-then-books — and two
		// reconcilers on one run (overlapping deploy generations) then deadlocked in both directions;
		// Postgres aborts one side, so it cost a failed sweep rather than corruption. Measured.
		var bookID string
		if err := tx.QueryRow(ctx,
			`select id from books where id = (select book_id from runs where id = $1) for update`,
			runID).Scan(&bookID); err != nil {
			if errors.Is(err, pgx.ErrNoRows) {
				return nil // the run is gone
			}
			return fmt.Errorf("pgstore: lock book: %w", err)
		}
		// nullif: a run that ends in any other status must not keep the reason a PREVIOUS pause of it
		// left behind — the resume path reads that column to decide what it may do.
		if err := tx.QueryRow(ctx, `
			update runs set status = $2, paused_reason = nullif($5, ''), finished_at = $3,
			                revision = (select revision + 1 from books where id = runs.book_id)
			 where id = $1 and finished_at is null
			   and exists (select 1 from run_attempts a
			                where a.id = $4 and a.run_id = runs.id and a.ended_at is null)
			returning book_id`, runID, in.Status, now, attemptID, in.PausedReason).Scan(&bookID); err != nil {
			if errors.Is(err, pgx.ErrNoRows) {
				// Already finished, or asked about an attempt that is no longer the live one. Both are
				// "someone else got here first"; finishing is idempotent by refusal, not by repetition.
				return nil
			}
			return fmt.Errorf("pgstore: finish run: %w", err)
		}
		if _, err := tx.Exec(ctx, `
			update run_attempts set ended_at = $2, exit_code = $3, exit_result = $4
			 where id = $1 and ended_at is null`, attemptID, now, in.ExitCode, in.ExitResult); err != nil {
			return fmt.Errorf("pgstore: finish attempt: %w", err)
		}
		if _, err := tx.Exec(ctx, `
			update books set status = $2,
			                 revision = `+nextRevisionOfThisBooksLibrary+` where id = $1`, bookID, in.Status); err != nil {
			return fmt.Errorf("pgstore: finish book: %w", err)
		}
		closed = true
		return nil
	})
	return closed, err
}

func validRunStatus(s string) bool {
	switch s {
	case "translating", "awaiting_bank", "finalizing", "ready", "stopped", "failed", "paused":
		return true
	}
	return false
}

// FinishUnspawnedStop closes a run that was stopped before its unit ever existed, and only if that is
// still true when the write happens.
//
// The re-check is the whole method. The reconciler decides from a snapshot in which the attempt had
// no unit; between that read and this write the queue worker can claim the attempt and create one,
// and closing the run then leaves an engine spending against a book with no open reservation and no
// list that looks at it — the reconciler lists by `finished_at is null` and settlement by an open
// reservation, so it would be in neither. `ReleaseUnspawned` already guards the MONEY of exactly this
// window under a lock (PD-159); this is the same guard for the lifecycle the money follows.
//
// finished is false when the attempt was spawned after all: the caller does nothing, and the next
// sweep meets an ordinary live run — with a unit to signal and an intent that says to.
func (s *Store) FinishUnspawnedStop(ctx context.Context, runID string, attemptID int64, now time.Time) (finished bool, err error) {
	err = s.inTx(ctx, func(tx pgx.Tx) error {
		// Book first, then the attempt: the order every transaction in this package takes (lockBook).
		var bookID string
		if err := tx.QueryRow(ctx,
			`select id from books where id = (select book_id from runs where id = $1) for update`,
			runID).Scan(&bookID); err != nil {
			if errors.Is(err, pgx.ErrNoRows) {
				return nil // the run is gone
			}
			return fmt.Errorf("pgstore: lock book: %w", err)
		}
		var unit *string
		var ended *time.Time
		if err := tx.QueryRow(ctx,
			`select unit_name, ended_at from run_attempts where id = $1 and run_id = $2 for update`,
			attemptID, runID).Scan(&unit, &ended); err != nil {
			if errors.Is(err, pgx.ErrNoRows) {
				return nil
			}
			return fmt.Errorf("pgstore: lock attempt: %w", err)
		}
		if unit != nil && *unit != "" {
			return nil // spawned inside the window; not this path's business any more
		}
		if ended != nil {
			// The attempt this pass is holding is over, which means the run moved on without it — a stop
			// and a RESUME can both have happened since the snapshot was taken, and the run is live
			// again on a second attempt holding a second reservation. Closing it from here would put
			// that hold in no worklist at all. The same guard FinishRun carries (PD-181); acceptance
			// found this path missing it.
			return nil
		}
		tag, err := tx.Exec(ctx, `
			update runs set status = 'stopped', finished_at = $2,
			                revision = (select revision + 1 from books where id = runs.book_id)
			 where id = $1 and finished_at is null`, runID, now)
		if err != nil {
			return fmt.Errorf("pgstore: finish stopped run: %w", err)
		}
		if tag.RowsAffected() == 0 {
			return nil // already finished by an earlier pass
		}
		if _, err := tx.Exec(ctx, `
			update run_attempts set ended_at = $2, exit_result = $3
			 where id = $1 and ended_at is null`, attemptID, now, StopRequestedResult); err != nil {
			return fmt.Errorf("pgstore: finish attempt: %w", err)
		}
		if _, err := tx.Exec(ctx, `
			update books set status = 'stopped',
			                 revision = `+nextRevisionOfThisBooksLibrary+` where id = $1`, bookID); err != nil {
			return fmt.Errorf("pgstore: finish book: %w", err)
		}
		finished = true
		return nil
	})
	return finished, err
}

// StopRequestedResult is the exit_result of an attempt that ended by this platform's request without
// systemd getting to write a marker. Deliberately a value systemd cannot produce: an operator reading
// the column must be able to tell what the machine saw from what the platform concluded.
const StopRequestedResult = "stop-requested"

// MarkSettled records that the money of a run has been resolved.
func (s *Store) MarkSettled(ctx context.Context, runID string, now time.Time) error {
	_, err := s.pool.Exec(ctx, `update runs set settled_at = $2 where id = $1 and settled_at is null`, runID, now)
	if err != nil {
		return fmt.Errorf("pgstore: mark settled: %w", err)
	}
	return nil
}

// UnsettledRuns lists finished runs whose money is still open. A run can finish and fail to settle
// — the engine's committed figure is read from a process that has to be asked, and asking can fail —
// and without this list that hold would stay reserved forever.
func (s *Store) UnsettledRuns(ctx context.Context) ([]LiveRun, error) {
	// Keyed on the ATTEMPT being over, not on the RUN being over. An attempt that was interrupted and
	// replaced leaves its reservation open while its run goes on, and a list that filtered on the run
	// never looked at it again — the hold stayed reserved for the life of the account.
	return s.queryRuns(ctx, `
		  join run_attempts a on a.run_id = r.id and a.ended_at is not null
		  join reservations res on res.engine_run_id = r.id || '#' || a.attempt_no and res.state = 'open'
		 order by a.ended_at`)
}

// ReservationKey is the attempt's reservation id, exported so the reconciler can settle without
// re-deriving a format that lives in this package.
func ReservationKey(runID string, attempt int) string { return engineRunKey(runID, attempt) }